hatchery_observation_api/
lib.rs

1//! Read-only monitoring contracts built from bounded observation facts.
2//!
3//! This crate intentionally contains no provider transport, terminal data,
4//! persistence, or provider-specific identifiers.
5
6use hatchery_observation_protocol::{ObservationKindV1, ObservationV1};
7use serde::{Deserialize, Serialize};
8use thiserror::Error;
9
10pub use gate4agent_node_protocol::{
11    NodeCursor, NodeId, NodeIncarnationId, SessionRecordId, WorkspaceId,
12};
13pub use gate4agent_types::{AgentInstanceId, SessionGeneration};
14
15pub const PROJECTIONS_MAX: usize = 8_192;
16pub const TIMELINE_PER_SESSION_MAX: usize = 512;
17pub const TOOLS_PER_SESSION_MAX: usize = 128;
18pub const SUBAGENTS_PER_SESSION_MAX: usize = 128;
19pub const FILES_PER_SESSION_MAX: usize = 128;
20pub const OWNED_PROCESSES_PER_SESSION_MAX: usize = 128;
21pub const INTERACTIONS_PER_SESSION_MAX: usize = 64;
22pub const INGRESS_BATCH_MAX: usize = 8_192;
23/// Most observations one node event may project to -- a lifecycle event mints
24/// a handful (source declaration plus its facts), never a stream.
25pub const OBSERVATIONS_PER_INGRESS_MAX: usize = 16;
26pub const CURSOR_JOURNAL_MAX: usize = 65_536;
27pub const NODE_ROUTES_MAX: usize = PROJECTIONS_MAX;
28pub const RETIRED_INCARNATIONS_MAX: usize = CURSOR_JOURNAL_MAX;
29pub const RESYNC_RECORDS_MAX: usize = PROJECTIONS_MAX;
30pub const RESYNC_GAPS_MAX: usize = 512;
31pub const RESYNC_EVENTS_MAX: usize = PROJECTIONS_MAX;
32
33#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
34#[serde(deny_unknown_fields)]
35pub struct RuntimeSessionKey {
36    pub node_id: NodeId,
37    pub incarnation_id: NodeIncarnationId,
38    pub workspace_id: WorkspaceId,
39    pub instance_id: AgentInstanceId,
40    pub generation: SessionGeneration,
41}
42
43impl RuntimeSessionKey {
44    pub fn validate(&self) -> Result<(), ObservationApiError> {
45        if self.instance_id.0 == 0 {
46            return Err(ObservationApiError::ZeroAgentInstanceId);
47        }
48        Ok(())
49    }
50}
51
52#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
53#[serde(deny_unknown_fields)]
54pub struct ManagedSessionKey {
55    pub node_id: NodeId,
56    pub incarnation_id: NodeIncarnationId,
57    pub record_id: SessionRecordId,
58}
59
60#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
61#[serde(tag = "target", rename_all = "kebab-case", deny_unknown_fields)]
62pub enum ObservationTarget {
63    Runtime { key: RuntimeSessionKey },
64    Managed { key: ManagedSessionKey },
65}
66
67impl ObservationTarget {
68    pub fn node_id(&self) -> &NodeId {
69        match self {
70            Self::Runtime { key } => &key.node_id,
71            Self::Managed { key } => &key.node_id,
72        }
73    }
74
75    pub fn runtime_key(&self) -> Option<&RuntimeSessionKey> {
76        match self {
77            Self::Runtime { key } => Some(key),
78            Self::Managed { .. } => None,
79        }
80    }
81
82    pub fn incarnation_id(&self) -> NodeIncarnationId {
83        match self {
84            Self::Runtime { key } => key.incarnation_id,
85            Self::Managed { key } => key.incarnation_id,
86        }
87    }
88
89    pub fn validate_route(
90        &self,
91        node_id: &NodeId,
92        incarnation_id: NodeIncarnationId,
93    ) -> Result<(), ObservationApiError> {
94        if self.node_id() != node_id {
95            return Err(ObservationApiError::RouteNodeMismatch);
96        }
97        if self.incarnation_id() != incarnation_id {
98            return Err(ObservationApiError::RouteIncarnationMismatch);
99        }
100        if let Some(runtime) = self.runtime_key() {
101            runtime.validate()?;
102        }
103        Ok(())
104    }
105}
106
107#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
108#[serde(deny_unknown_fields)]
109pub struct ManagedRecordLink {
110    pub managed: ManagedSessionKey,
111    pub runtime: Option<RuntimeSessionKey>,
112}
113
114impl ManagedRecordLink {
115    pub fn validate_route(
116        &self,
117        node_id: &NodeId,
118        incarnation_id: NodeIncarnationId,
119    ) -> Result<(), ObservationApiError> {
120        if &self.managed.node_id != node_id {
121            return Err(ObservationApiError::RouteNodeMismatch);
122        }
123        if self.managed.incarnation_id != incarnation_id {
124            return Err(ObservationApiError::RouteIncarnationMismatch);
125        }
126        if let Some(runtime) = self.runtime.as_ref() {
127            runtime.validate()?;
128            if &runtime.node_id != node_id {
129                return Err(ObservationApiError::RouteNodeMismatch);
130            }
131            if runtime.incarnation_id != incarnation_id {
132                return Err(ObservationApiError::RouteIncarnationMismatch);
133            }
134        }
135        Ok(())
136    }
137}
138
139#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
140#[serde(rename_all = "kebab-case")]
141pub enum ObservationTransport {
142    DirectNode,
143    C2,
144}
145
146/// What one node route delivers to the observation pipeline.
147///
148/// The harness derives every observation from the node's own event stream
149/// (`hatchery_observation_engine::node_projection`), so an online route
150/// supports all of it; the type stays so a route's support is still recorded and
151/// read per route, and a route with nothing recorded stays "unknown".
152#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
153pub struct ObservationSupport {
154    pub events: bool,
155    pub managed_target: bool,
156    pub workflow_detail: bool,
157}
158
159impl ObservationSupport {
160    /// Full support: every projection the harness derives is available.
161    pub const fn full() -> Self {
162        Self { events: true, managed_target: true, workflow_detail: true }
163    }
164}
165
166#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
167#[serde(tag = "kind", rename_all = "kebab-case", deny_unknown_fields)]
168pub enum ObservationIngressPayload {
169    /// The observations one node event projected to, applied atomically at the
170    /// event's cursor. An observation addressed to a runtime session also
171    /// reaches every managed record currently linked to it.
172    Observations {
173        address: ObservationTarget,
174        observations: Vec<ObservationV1>,
175    },
176    ManagedRecordUpserted {
177        link: ManagedRecordLink,
178    },
179    ManagedRecordRemoved {
180        key: ManagedSessionKey,
181    },
182    ResyncRequired {
183        oldest: NodeCursor,
184    },
185    CursorOnly,
186}
187
188impl ObservationIngressPayload {
189    pub fn validate_route(
190        &self,
191        node_id: &NodeId,
192        cursor: NodeCursor,
193    ) -> Result<(), ObservationApiError> {
194        match self {
195            Self::Observations {
196                address,
197                observations,
198            } => {
199                address.validate_route(node_id, cursor.incarnation_id)?;
200                validate_count(
201                    "observations per ingress",
202                    observations.len(),
203                    OBSERVATIONS_PER_INGRESS_MAX,
204                )?;
205                for observation in observations {
206                    observation.validate().map_err(ObservationApiError::Observation)?;
207                    if let ObservationKindV1::Error { detail } = &observation.kind {
208                        validate_error_category(detail)?;
209                    }
210                }
211                Ok(())
212            }
213            Self::ManagedRecordUpserted { link } => {
214                link.validate_route(node_id, cursor.incarnation_id)
215            }
216            Self::ManagedRecordRemoved { key } => {
217                if &key.node_id != node_id {
218                    return Err(ObservationApiError::RouteNodeMismatch);
219                }
220                if key.incarnation_id != cursor.incarnation_id {
221                    return Err(ObservationApiError::RouteIncarnationMismatch);
222                }
223                Ok(())
224            }
225            Self::ResyncRequired { oldest } => {
226                if oldest.incarnation_id != cursor.incarnation_id {
227                    return Err(ObservationApiError::RouteIncarnationMismatch);
228                }
229                if oldest.sequence == 0 || oldest.sequence > cursor.sequence {
230                    return Err(ObservationApiError::InvalidOldestCursor);
231                }
232                Ok(())
233            }
234            Self::CursorOnly => Ok(()),
235        }
236    }
237}
238
239#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
240#[serde(deny_unknown_fields)]
241pub struct ObservationIngressEnvelope {
242    pub node_id: NodeId,
243    pub cursor: NodeCursor,
244    pub received_at_ms: u64,
245    pub transport: ObservationTransport,
246    pub payload: ObservationIngressPayload,
247}
248
249impl ObservationIngressEnvelope {
250    pub fn validate(&self) -> Result<(), ObservationApiError> {
251        if self.cursor.sequence == 0 {
252            return Err(ObservationApiError::ZeroEventSequence);
253        }
254        if self.received_at_ms == 0 {
255            return Err(ObservationApiError::ZeroReceivedAt);
256        }
257        self.payload.validate_route(&self.node_id, self.cursor)
258    }
259
260    /// Equality of the Node-authoritative ingress fact.
261    ///
262    /// Transport and local receive time are delivery metadata. They may differ
263    /// when the same exact Node cursor is observed directly and through C2.
264    pub fn canonical_eq(&self, other: &Self) -> bool {
265        self.node_id == other.node_id
266            && self.cursor == other.cursor
267            && self.payload == other.payload
268    }
269}
270
271#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
272#[serde(deny_unknown_fields)]
273pub struct ObservationGap {
274    pub first_sequence: u64,
275    pub last_sequence: u64,
276}
277
278impl ObservationGap {
279    pub fn validate(&self, requested_after: u64, high_watermark: u64) -> Result<(), ObservationApiError> {
280        if self.first_sequence == 0
281            || self.first_sequence > self.last_sequence
282            || self.first_sequence <= requested_after
283            || self.last_sequence > high_watermark
284        {
285            return Err(ObservationApiError::InvalidGap);
286        }
287        Ok(())
288    }
289}
290
291#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
292#[serde(deny_unknown_fields)]
293pub struct ObservationResyncBatch {
294    pub node_id: NodeId,
295    pub incarnation_id: NodeIncarnationId,
296    pub requested_after: u64,
297    pub high_watermark: NodeCursor,
298    /// Node-authoritative replay floor. `high_watermark.sequence + 1` means
299    /// that no replay event is available through the high watermark.
300    pub oldest_available_sequence: u64,
301    pub records: Vec<ManagedRecordLink>,
302    pub records_complete: bool,
303    /// The exact retention-evicted prefix between `requested_after` and
304    /// `oldest_available_sequence`; sparse non-observation sequences are not gaps.
305    pub gaps: Vec<ObservationGap>,
306    pub events: Vec<ObservationIngressEnvelope>,
307}
308
309/// One exact managed-record inventory operation for a Node incarnation.
310///
311/// This is inventory state only. Applying it must not imply that an
312/// observation transport gap has been repaired.
313#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
314#[serde(deny_unknown_fields)]
315pub struct ObservationRecordInventory {
316    pub node_id: NodeId,
317    pub incarnation_id: NodeIncarnationId,
318    pub records: Vec<ManagedRecordLink>,
319    pub complete: bool,
320}
321
322impl ObservationRecordInventory {
323    pub fn validate(&self) -> Result<(), ObservationApiError> {
324        validate_count("managed record inventory", self.records.len(), RESYNC_RECORDS_MAX)?;
325        for record in &self.records {
326            record.validate_route(&self.node_id, self.incarnation_id)?;
327        }
328        Ok(())
329    }
330}
331
332impl ObservationResyncBatch {
333    pub fn validate(&self) -> Result<(), ObservationApiError> {
334        if self.high_watermark.incarnation_id != self.incarnation_id
335            || self.high_watermark.sequence < self.requested_after
336        {
337            return Err(ObservationApiError::InvalidHighWatermark);
338        }
339        validate_count("resync records", self.records.len(), RESYNC_RECORDS_MAX)?;
340        validate_count("resync gaps", self.gaps.len(), RESYNC_GAPS_MAX)?;
341        validate_count("resync events", self.events.len(), RESYNC_EVENTS_MAX)?;
342
343        let maximum_floor = self
344            .high_watermark
345            .sequence
346            .checked_add(1)
347            .unwrap_or(u64::MAX);
348        let requested_next = self.requested_after.checked_add(1).unwrap_or(u64::MAX);
349        if self.oldest_available_sequence == 0
350            || self.oldest_available_sequence > maximum_floor
351        {
352            return Err(ObservationApiError::InvalidReplayFloor);
353        }
354        let expected_gap = (requested_next < self.oldest_available_sequence).then(|| {
355            ObservationGap {
356                first_sequence: requested_next,
357                last_sequence: self.oldest_available_sequence - 1,
358            }
359        });
360        if self.gaps.as_slice() != expected_gap.as_slice() {
361            return Err(ObservationApiError::InvalidReplayGapProof);
362        }
363
364        for record in &self.records {
365            record.validate_route(&self.node_id, self.incarnation_id)?;
366        }
367        for gap in &self.gaps {
368            gap.validate(self.requested_after, self.high_watermark.sequence)?;
369        }
370        for event in &self.events {
371            event.validate()?;
372            if event.node_id != self.node_id
373                || event.cursor.incarnation_id != self.incarnation_id
374                || event.cursor.sequence <= self.requested_after
375                || event.cursor.sequence < self.oldest_available_sequence
376                || event.cursor.sequence > self.high_watermark.sequence
377            {
378                return Err(ObservationApiError::EventOutsideResyncRoute);
379            }
380        }
381        Ok(())
382    }
383}
384
385#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
386#[serde(rename_all = "kebab-case")]
387pub enum ProjectionAvailability {
388    Unknown,
389    NotObserved,
390    Current,
391    Partial,
392    Frozen,
393}
394
395#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
396#[serde(rename_all = "kebab-case")]
397pub enum ProjectionFreshness {
398    Live,
399    LastKnown,
400    Stale,
401    IncompleteAfterGap,
402    ReplacedIncarnation,
403    Unavailable,
404}
405
406#[derive(Clone, Debug, Eq, Error, PartialEq)]
407pub enum ObservationApiError {
408    #[error("agent instance ID must be non-zero")]
409    ZeroAgentInstanceId,
410    #[error("node event sequence must be non-zero")]
411    ZeroEventSequence,
412    #[error("received_at_ms must be non-zero")]
413    ZeroReceivedAt,
414    #[error("observation route node does not match its envelope")]
415    RouteNodeMismatch,
416    #[error("observation route incarnation does not match its cursor")]
417    RouteIncarnationMismatch,
418    #[error("observation failed protocol validation: {0}")]
419    Observation(#[source] hatchery_observation_protocol::ObservationValidationError),
420    #[error("resync-required oldest cursor is invalid")]
421    InvalidOldestCursor,
422    #[error("resync high watermark is invalid")]
423    InvalidHighWatermark,
424    #[error("resync oldest available sequence is invalid")]
425    InvalidReplayFloor,
426    #[error("resync gaps do not exactly prove the authoritative replay floor")]
427    InvalidReplayGapProof,
428    #[error("resync gap is outside the requested cursor range")]
429    InvalidGap,
430    #[error("resync event is outside the batch route or cursor range")]
431    EventOutsideResyncRoute,
432    #[error("error detail must be a lowercase categorical slug")]
433    InvalidErrorDetail,
434    #[error("{field} count {actual} exceeds maximum {max}")]
435    TooMany {
436        field: &'static str,
437        max: usize,
438        actual: usize,
439    },
440}
441
442fn validate_error_category(detail: &str) -> Result<(), ObservationApiError> {
443    let mut segments = detail.split('-');
444    if segments.any(|segment| {
445        segment.is_empty()
446            || !segment
447                .bytes()
448                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit())
449    })
450    {
451        return Err(ObservationApiError::InvalidErrorDetail);
452    }
453    Ok(())
454}
455
456fn validate_count(
457    field: &'static str,
458    actual: usize,
459    max: usize,
460) -> Result<(), ObservationApiError> {
461    if actual > max {
462        return Err(ObservationApiError::TooMany { field, max, actual });
463    }
464    Ok(())
465}
466
467#[cfg(test)]
468mod tests {
469    use super::*;
470    use hatchery_observation_protocol::{
471        ObservationEvidenceV1, ObservationKindV1,
472    };
473
474    fn node(value: &str) -> NodeId {
475        NodeId::new(value).expect("node ID")
476    }
477
478    fn incarnation(byte: u8) -> NodeIncarnationId {
479        NodeIncarnationId::from_bytes([byte; 16])
480    }
481
482    fn runtime(node_id: NodeId, incarnation_id: NodeIncarnationId) -> RuntimeSessionKey {
483        RuntimeSessionKey {
484            node_id,
485            incarnation_id,
486            workspace_id: WorkspaceId::new("workspace-a").expect("workspace ID"),
487            instance_id: AgentInstanceId(1),
488            generation: SessionGeneration(0),
489        }
490    }
491
492    fn event(node_id: NodeId, incarnation_id: NodeIncarnationId) -> ObservationIngressEnvelope {
493        ObservationIngressEnvelope {
494            node_id: node_id.clone(),
495            cursor: NodeCursor {
496                incarnation_id,
497                sequence: 1,
498            },
499            received_at_ms: 10,
500            transport: ObservationTransport::DirectNode,
501            payload: ObservationIngressPayload::Observations {
502                address: ObservationTarget::Runtime {
503                    key: runtime(node_id, incarnation_id),
504                },
505                observations: vec![ObservationV1 {
506                    source_sequence: 1,
507                    observed_at_unix_ms: Some(9),
508                    evidence: ObservationEvidenceV1::NodeLifecycle,
509                    kind: ObservationKindV1::Ready,
510                    truncated: false,
511                }],
512            },
513        }
514    }
515
516    #[test]
517    fn ingress_validation_and_resync_bounds_are_enforced() {
518        let node_a = node("node-a");
519        let node_b = node("node-b");
520        let incarnation_id = incarnation(1);
521        let valid = event(node_a.clone(), incarnation_id);
522        valid.validate().expect("valid ingress");
523
524        let mut wrong_route = valid.clone();
525        wrong_route.node_id = node_b;
526        assert_eq!(
527            wrong_route.validate(),
528            Err(ObservationApiError::RouteNodeMismatch)
529        );
530
531        let mut zero_received = valid.clone();
532        zero_received.received_at_ms = 0;
533        assert_eq!(
534            zero_received.validate(),
535            Err(ObservationApiError::ZeroReceivedAt)
536        );
537
538        let batch = ObservationResyncBatch {
539            node_id: node_a,
540            incarnation_id,
541            requested_after: 0,
542            high_watermark: NodeCursor {
543                incarnation_id,
544                sequence: 1,
545            },
546            oldest_available_sequence: 1,
547            records: Vec::new(),
548            records_complete: true,
549            gaps: vec![ObservationGap {
550                first_sequence: 1,
551                last_sequence: 1,
552            }; RESYNC_GAPS_MAX + 1],
553            events: vec![valid],
554        };
555        assert!(matches!(
556            batch.validate(),
557            Err(ObservationApiError::TooMany {
558                field: "resync gaps",
559                ..
560            })
561        ));
562    }
563
564    #[test]
565    fn managed_route_is_exactly_incarnation_scoped() {
566        let node_id = node("node-a");
567        let first_incarnation = incarnation(1);
568        let second_incarnation = incarnation(2);
569        let key = ManagedSessionKey {
570            node_id: node_id.clone(),
571            incarnation_id: first_incarnation,
572            record_id: SessionRecordId::new("record-inline-a").expect("record ID"),
573        };
574        let target = ObservationTarget::Managed { key: key.clone() };
575        target
576            .validate_route(&node_id, first_incarnation)
577            .expect("exact managed route");
578        assert_eq!(
579            target.validate_route(&node_id, second_incarnation),
580            Err(ObservationApiError::RouteIncarnationMismatch)
581        );
582
583        let link = ManagedRecordLink {
584            managed: key,
585            runtime: None,
586        };
587        link.validate_route(&node_id, first_incarnation)
588            .expect("exact managed link route");
589        assert_eq!(
590            link.validate_route(&node_id, second_incarnation),
591            Err(ObservationApiError::RouteIncarnationMismatch)
592        );
593    }
594}
hatchery_observation_api/lib.rs

hatchery_observation_api/
lib.rs
