feat: add workload observation contract (#9)

Co-authored-by: RoyLin <roylin@RoyLindeMacBook-Pro.local>

Author: ZhiXiao-Lin

CHANGELOG.md

@@ -2,6 +2,27 @@
 
 All notable changes to a3s-observer will be documented in this file.
 
+## [Unreleased]
+
+### Added
+
+- A provider-neutral workload attribution contract: `WorkloadIdentity` requires stable workload,
+  deployment, immutable revision, logical replica, provider-unit, and node IDs. Each
+  `WorkloadIdentityValue` is bounded to 128 ASCII bytes and rejects whitespace, control
+  characters, free-form Unicode, and non-canonical boundary characters without echoing rejected
+  input in validation errors.
+- Explicit `ObservationMetadata` for sampled signals, including observation timestamp, optional
+  sample timestamp and collection interval, and `fresh`, `stale`, `unavailable`, or `unknown`
+  state. Unavailable and unknown observations represent missing data without a zero-valued
+  sentinel.
+- Optional `workload` and `observation` fields in NDJSON `EnrichedEvent` output.
+  `IdentityResolver::resolve_workload` defaults to `None`, preserving existing resolver
+  implementations and preventing incomplete attribution.
+
+This contract is the first slice of workload metrics support. Multi-replica Linux collection,
+resource/restart/availability measurements, lifecycle fixtures, and OTLP/Prometheus metric parity
+are not included yet.
+
 ## [0.11.0] — SecurityAction: privesc / injection / open-port
 
 ### Added

README.md

@@ -83,6 +83,49 @@ Filter with `jq`, e.g. every LLM call and its provider:
 Set `A3S_OBSERVER_COLLECTOR_ID` and `A3S_NODE_NAME` in DaemonSets when you want stable collector
 identity in downstream fleet-health views. If unset, the collector falls back to pod/host names.
 
+### Workload attribution and freshness contract
+
+`EnrichedEvent` can also carry provider-neutral `workload` and `observation` objects for
+per-replica signals:
+
+```json
+{
+  "workload": {
+    "workload_id": "workload-01HV7F5N",
+    "deployment_id": "deployment-01HV7F6A",
+    "revision_id": "revision-sha256:8f3a",
+    "replica_id": "replica-0007",
+    "provider_unit_id": "containerd:4f6c2d8a",
+    "node_id": "node-us-east-1a-03"
+  },
+  "observation": {
+    "observed_at_unix_nanos": 1720000015000000000,
+    "sampled_at_unix_nanos": 1720000014000000000,
+    "collection_interval_nanos": 15000000000,
+    "freshness": "fresh"
+  }
+}
+```
+
+`WorkloadIdentity` is complete by construction: a producer must provide stable workload,
+deployment, immutable revision, logical replica, current provider-unit, and node IDs. Each ID is
+an opaque platform identifier limited to 128 ASCII bytes and a label-safe alphabet. Producers
+must normalize provider IDs and must never copy tenant secrets, display names, or raw user labels
+into these fields. A logical `replica_id` survives process restart, adoption, and rescheduling;
+`provider_unit_id` changes when the runtime unit is replaced.
+
+`ObservationMetadata` reports an observation time, an optional collection interval, and one of
+`fresh`, `stale`, `unavailable`, or `unknown`. Fresh and stale data include the actual sample
+timestamp. Unavailable and unknown observations omit it, giving producers an explicit
+missing-data state instead of a zero-usage sentinel.
+
+This is the transport contract, not a claim that per-replica resource collection is complete.
+Existing `IdentityResolver` implementations return no workload identity by default, and the
+node-wide collector intentionally does not apply one static environment identity to every event.
+Multi-replica Linux collection, CPU/memory/network/process/restart/availability samples,
+restart/adoption fixtures, and equivalent OTLP and Prometheus metric exporters remain follow-up
+work.
+
 ## Intervene — egress / file / exec (opt-in)
 
 The same vantage point enforces an **external policy** — a plain file any controller writes; the

a3s-observer-collector/src/main.rs

@@ -260,6 +260,8 @@ async fn main() -> anyhow::Result<()> {
                     if let Some(ev) = read_pod::<ExecEvent>(&item) {
                         emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                             identity: identity_for(&resolver, ev.pid, &ev.comm),
+                            workload: resolver.resolve_workload(ev.pid, 0, 0),
+                            observation: None,
                             process: Some(process_context(ev.pid, &ev.comm)),
                             provider: None,
                             event: AgentEvent::ToolExec {
@@ -276,6 +278,8 @@ async fn main() -> anyhow::Result<()> {
                     if let Some(ev) = read_pod::<ExitEvent>(&item) {
                         emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                             identity: identity_for(&resolver, ev.pid, &ev.comm),
+                            workload: resolver.resolve_workload(ev.pid, 0, 0),
+                            observation: None,
                             process: Some(process_context(ev.pid, &ev.comm)),
                             provider: None,
                             event: AgentEvent::ProcessExit {
@@ -296,6 +300,8 @@ async fn main() -> anyhow::Result<()> {
                         };
                         emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                             identity: identity_for(&resolver, ev.pid, &ev.comm),
+                            workload: resolver.resolve_workload(ev.pid, 0, 0),
+                            observation: None,
                             process: Some(process_context(ev.pid, &ev.comm)),
                             provider: None,
                             event: AgentEvent::SecurityAction {
@@ -316,6 +322,8 @@ async fn main() -> anyhow::Result<()> {
                         peers.insert(sock_key(ev.pid, ev.fd), (peer, ev.port));
                         emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                             identity: identity_for(&resolver, ev.pid, &ev.comm),
+                            workload: resolver.resolve_workload(ev.pid, 0, 0),
+                            observation: None,
                             process: Some(process_context(ev.pid, &ev.comm)),
                             provider: None,
                             event: AgentEvent::Egress {
@@ -347,6 +355,8 @@ async fn main() -> anyhow::Result<()> {
                         llm_meta.insert(sock_key(ev.pid, ev.fd), (sni.clone(), provider.clone(), peer));
                         emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                             identity: identity_for(&resolver, ev.pid, &ev.comm),
+                            workload: resolver.resolve_workload(ev.pid, 0, 0),
+                            observation: None,
                             process: Some(process_context(ev.pid, &ev.comm)),
                             provider,
                             event: AgentEvent::Egress {
@@ -365,6 +375,8 @@ async fn main() -> anyhow::Result<()> {
                         if let Some(query) = parse_dns_qname(&ev.data[..len]) {
                             emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                                 identity: identity_for(&resolver, ev.pid, &ev.comm),
+                                workload: resolver.resolve_workload(ev.pid, 0, 0),
+                                observation: None,
                                 process: Some(process_context(ev.pid, &ev.comm)),
                                 provider: None,
                                 event: AgentEvent::Dns { pid: ev.pid, query },
@@ -388,6 +400,8 @@ async fn main() -> anyhow::Result<()> {
                             };
                             emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                                 identity: identity_for(&resolver, ev.pid, &ev.comm),
+                                workload: resolver.resolve_workload(ev.pid, 0, 0),
+                                observation: None,
                                 process: Some(process_context(ev.pid, &ev.comm)),
                                 provider: None,
                                 event,
@@ -405,6 +419,8 @@ async fn main() -> anyhow::Result<()> {
                             if provider.is_some() {
                                 emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                                     identity: identity_for(&resolver, ev.pid, &ev.comm),
+                                    workload: resolver.resolve_workload(ev.pid, 0, 0),
+                                    observation: None,
                                     process: Some(process_context(ev.pid, &ev.comm)),
                                     provider,
                                     event: AgentEvent::LlmCall {
@@ -428,12 +444,15 @@ async fn main() -> anyhow::Result<()> {
                         let content = String::from_utf8_lossy(&ev.data[..len]).into_owned();
                         if !content.is_empty() {
                             let identity = identity_for(&resolver, ev.pid, &ev.comm);
+                            let workload = resolver.resolve_workload(ev.pid, 0, 0);
                             // Structured LLM telemetry (model/tokens) alongside the raw content.
                             if let Some((model, prompt_tokens, completion_tokens)) =
                                 parse_llm_meta(&content)
                             {
                                 emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                                     identity: identity.clone(),
+                                    workload: workload.clone(),
+                                    observation: None,
                                     process: Some(process_context(ev.pid, &ev.comm)),
                                     provider: None,
                                     event: AgentEvent::LlmApi {
@@ -447,6 +466,8 @@ async fn main() -> anyhow::Result<()> {
                             }
                             emit(exporter.as_ref(), &mut stats, EnrichedEvent {
                                 identity,
+                                workload,
+                                observation: None,
                                 process: Some(process_context(ev.pid, &ev.comm)),
                                 provider: None,
                                 event: AgentEvent::SslContent {
@@ -674,6 +695,8 @@ fn emit_collector_heartbeat(
 ) {
     exporter.export(&EnrichedEvent {
         identity: Identity::default(),
+        workload: None,
+        observation: None,
         process: None,
         provider: None,
         event: AgentEvent::CollectorHeartbeat {

src/lib.rs

@@ -16,10 +16,15 @@
 pub mod model;
 pub mod policy;
 pub mod traits;
+pub mod workload;
 
 pub use model::{AgentEvent, EnrichedEvent, ProcessContext};
 pub use policy::{parse_egress_policy, AllowAll, Policy, ProviderPolicy, Verdict};
 pub use traits::{
     read_ppid, Exporter, Identity, IdentityResolver, JsonExporter, KubeResolver, LogExporter,
     ProcResolver, Provider, ServiceClassifier, SniClassifier,
 };
+pub use workload::{
+    Freshness, ObservationMetadata, ObservationMetadataError, WorkloadIdentity,
+    WorkloadIdentityValue, WorkloadIdentityValueError, MAX_WORKLOAD_IDENTITY_VALUE_LEN,
+};

src/model.rs

@@ -2,6 +2,7 @@
 //! events the [`Exporter`](crate::Exporter) receives.
 
 use crate::traits::{Identity, Provider};
+use crate::workload::{ObservationMetadata, WorkloadIdentity};
 use serde::Serialize;
 use std::net::IpAddr;
 use std::time::Duration;
@@ -137,6 +138,12 @@ pub enum AgentEvent {
 #[derive(Debug, Clone, Serialize)]
 pub struct EnrichedEvent {
     pub identity: Identity,
+    /// Complete workload attribution, when a resolver can prove every stable identity field.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub workload: Option<WorkloadIdentity>,
+    /// Explicit timing and freshness for sampled signals. Consumers must not infer zero when absent.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub observation: Option<ObservationMetadata>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub process: Option<ProcessContext>,
     pub provider: Option<Provider>,

src/traits.rs

@@ -5,6 +5,7 @@
 //! probes.
 
 use crate::model::EnrichedEvent;
+use crate::workload::WorkloadIdentity;
 use serde::Serialize;
 use std::net::IpAddr;
 
@@ -21,6 +22,19 @@ pub struct Identity {
 /// Implementations: k8s (cgroup→pod), docker, a3s-box (pid/netns→box), bare pid-tree.
 pub trait IdentityResolver: Send + Sync {
     fn resolve(&self, pid: u32, cgroup_id: u64, netns: u64) -> Identity;
+
+    /// Resolve a complete, provider-neutral workload identity when one is available.
+    ///
+    /// Existing process-only resolvers remain valid and default to no workload attribution.
+    /// Implementations must return `None` rather than inventing or partially filling identity.
+    fn resolve_workload(
+        &self,
+        _pid: u32,
+        _cgroup_id: u64,
+        _netns: u64,
+    ) -> Option<WorkloadIdentity> {
+        None
+    }
 }
 
 /// Known service providers, identified language-agnostically from TLS SNI / DNS.
@@ -346,6 +360,8 @@ mod tests {
                 task: Some("1".into()),
                 session: None,
             },
+            workload: None,
+            observation: None,
             process: None,
             provider: None,
             event: AgentEvent::ProcessExit {