traverse_runtime/events/
types.rs

1//! Core types for the in-process event system.
2//!
3//! Governed by spec 026-event-broker and spec 036-event-subscription-replay.
4
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7use std::sync::Arc;
8
9/// Lifecycle status of an event type in the catalog.
10#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "snake_case")]
12pub enum LifecycleStatus {
13    Draft,
14    Active,
15    Deprecated,
16}
17
18/// A CloudEvents-formatted event with Traverse governance metadata.
19#[derive(Debug, Clone, Serialize, Deserialize)]
20pub struct TraverseEvent {
21    /// UUID for this event instance.
22    pub id: String,
23    /// Originating capability: `"traverse-runtime/<capability_id>"`.
24    pub source: String,
25    /// Reverse-DNS event type, e.g. `"dev.traverse.expedition.planned"`.
26    pub event_type: String,
27    /// Always `"application/json"`.
28    pub datacontenttype: String,
29    /// RFC 3339 timestamp.
30    pub time: String,
31    /// Event payload.
32    pub data: Value,
33    // --- governance metadata ---
34    /// Capability ID that emits this event.
35    pub owner: String,
36    /// Event contract version.
37    pub version: String,
38    /// Lifecycle status at the time the event was created.
39    pub lifecycle_status: LifecycleStatus,
40    /// Authenticated subject that caused this event, when known.
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub subject_id: Option<String>,
43    /// Delegated actor distinct from the subject, when known.
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub actor_id: Option<String>,
46}
47
48/// Errors that can occur during event broker operations.
49#[derive(Debug, PartialEq, Eq)]
50pub enum EventError {
51    /// Attempted to publish an event whose catalog entry is `Deprecated` or `Draft`.
52    LifecycleViolation(String),
53    /// Attempted to publish an event type not registered in the catalog.
54    UnregisteredEventType(String),
55    /// Cursor string could not be parsed.
56    InvalidCursor(String),
57    /// The requested cursor is outside the active retention window.
58    CursorExpired {
59        event_type: String,
60        oldest_available_cursor: String,
61    },
62    /// Subscription id is unknown or was cancelled.
63    SubscriptionNotFound(String),
64    /// Broker was configured with an invalid retention window.
65    InvalidRetentionWindow(String),
66    /// Durable journal write failed; the event was not acknowledged (066 FR-006).
67    JournalWrite(String),
68    /// Durable journal write exceeded the configured timeout; the event was
69    /// rejected, not delivered (067 FR-003/FR-004).
70    JournalWriteTimeout(String),
71    /// Durable journal read failed while serving replay (066 FR-005/FR-009).
72    JournalRead(String),
73}
74
75impl std::fmt::Display for EventError {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        match self {
78            Self::LifecycleViolation(msg) => write!(f, "lifecycle violation: {msg}"),
79            Self::UnregisteredEventType(t) => write!(f, "unregistered event type: {t}"),
80            Self::InvalidCursor(msg) => write!(f, "invalid cursor: {msg}"),
81            Self::CursorExpired {
82                event_type,
83                oldest_available_cursor,
84            } => write!(
85                f,
86                "cursor expired for event type '{event_type}': oldest available cursor is {oldest_available_cursor}"
87            ),
88            Self::SubscriptionNotFound(id) => write!(f, "subscription not found: {id}"),
89            Self::InvalidRetentionWindow(msg) => write!(f, "invalid retention window: {msg}"),
90            Self::JournalWrite(msg) => write!(f, "journal write failed: {msg}"),
91            Self::JournalWriteTimeout(msg) => write!(f, "journal_write_timeout: {msg}"),
92            Self::JournalRead(msg) => write!(f, "journal read failed: {msg}"),
93        }
94    }
95}
96
97impl std::error::Error for EventError {}
98
99/// Pub/sub interface for in-process event delivery.
100pub trait EventBroker: Send + Sync {
101    /// Publish an event. Fails if the event type is not `Active` in the catalog.
102    ///
103    /// # Errors
104    ///
105    /// Returns [`EventError::UnregisteredEventType`] if the event type is not in the catalog,
106    /// or [`EventError::LifecycleViolation`] if the catalog entry is not `Active`.
107    fn publish(&self, event: TraverseEvent) -> Result<(), EventError>;
108
109    /// Publish an event, threading a pre-assigned durable cursor through to
110    /// the broker's own cursor space when the implementation supports cursor
111    /// injection. [`DurableBroker`](super::durable::DurableBroker) uses this
112    /// to keep live-delivery cursors numerically consistent with the durable
113    /// journal's cursor space (spec 066 FR-007), so a cursor issued during
114    /// live polling remains valid when later resolved through durable
115    /// replay. The default implementation ignores `cursor` and behaves like
116    /// [`Self::publish`].
117    ///
118    /// # Errors
119    ///
120    /// Returns the same errors as [`Self::publish`].
121    fn publish_with_cursor(&self, event: TraverseEvent, cursor: &str) -> Result<(), EventError> {
122        let _ = cursor;
123        self.publish(event)
124    }
125
126    /// Establishes a floor below which any cursor, for any event type this
127    /// broker has not itself observed, is treated as potentially predating
128    /// this broker instance's own history (spec 066 FR-007: restart cursor
129    /// continuity). [`DurableBroker::open`](super::durable::DurableBroker::open)
130    /// calls this once at construction, seeded from the durable journal's
131    /// latest cursor, so a freshly constructed in-memory broker correctly
132    /// defers stale-looking cursors to durable replay after a restart
133    /// instead of accepting them outright. The default implementation is a
134    /// no-op.
135    fn seed_restart_floor(&self, floor: u64) {
136        let _ = floor;
137    }
138
139    /// Create a subscription for the given `event_type` starting from `from_cursor`.
140    ///
141    /// `from_cursor` is an opaque cursor string previously returned by [`poll`](Self::poll).
142    /// The special value `"0"` requests replay from the start of the active retention window.
143    ///
144    /// # Errors
145    ///
146    /// Returns [`EventError::UnregisteredEventType`] if the event type is not in the catalog,
147    /// [`EventError::InvalidCursor`] if the cursor string is malformed, or
148    /// [`EventError::CursorExpired`] if the cursor is outside the retention window.
149    fn subscribe(&self, event_type: &str, from_cursor: &str) -> Result<Subscription, EventError>;
150
151    /// Create a subscription optionally limited to one event subject.
152    ///
153    /// # Errors
154    ///
155    /// Returns the same errors as [`Self::subscribe`] when the event type or
156    /// cursor is invalid.
157    fn subscribe_for_subject(
158        &self,
159        event_type: &str,
160        from_cursor: &str,
161        subject_id: Option<&str>,
162    ) -> Result<Subscription, EventError>;
163
164    /// Poll a subscription for up to `max_events`.
165    ///
166    /// # Errors
167    ///
168    /// Returns [`EventError::SubscriptionNotFound`] if the subscription id is unknown or cancelled.
169    fn poll(
170        &self,
171        subscription_id: &str,
172        max_events: usize,
173    ) -> Result<SubscriptionPoll, EventError>;
174
175    /// Cancel a subscription and free all associated queues.
176    ///
177    /// # Errors
178    ///
179    /// Returns [`EventError::SubscriptionNotFound`] if the subscription id is unknown.
180    fn cancel(&self, subscription_id: &str) -> Result<(), EventError>;
181}
182
183/// Narrow runtime boundary for publishing lifecycle envelopes.
184///
185/// The runtime depends on this seam rather than a concrete broker so embedders
186/// can choose an in-memory broker, durable broker, or a no-op delivery policy.
187pub trait RuntimeEventSink: Send + Sync + std::fmt::Debug {
188    /// Deliver one already-materialized runtime lifecycle envelope.
189    ///
190    /// # Errors
191    ///
192    /// Implementations return an error when the configured delivery mechanism
193    /// cannot accept the envelope. Runtime execution remains authoritative and
194    /// reports delivery failures as warnings.
195    fn emit(&self, event: TraverseEvent) -> Result<(), EventError>;
196}
197
198/// Default sink used by existing runtime constructors.
199#[derive(Debug, Default)]
200pub struct NoopRuntimeEventSink;
201
202impl RuntimeEventSink for NoopRuntimeEventSink {
203    fn emit(&self, _event: TraverseEvent) -> Result<(), EventError> {
204        Ok(())
205    }
206}
207
208/// Adapter that routes runtime lifecycle envelopes through an event broker.
209pub struct BrokerEventSink {
210    broker: Arc<dyn EventBroker>,
211}
212
213impl BrokerEventSink {
214    #[must_use]
215    pub fn new(broker: Arc<dyn EventBroker>) -> Self {
216        Self { broker }
217    }
218}
219
220impl std::fmt::Debug for BrokerEventSink {
221    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
222        formatter
223            .debug_struct("BrokerEventSink")
224            .finish_non_exhaustive()
225    }
226}
227
228impl RuntimeEventSink for BrokerEventSink {
229    fn emit(&self, event: TraverseEvent) -> Result<(), EventError> {
230        self.broker.publish(event)
231    }
232}
233
234/// A broker-issued event cursor string.
235pub type EventCursor = String;
236
237/// A broker-assigned subscription identifier.
238pub type SubscriptionId = String;
239
240/// Event delivered by the broker, carrying a cursor for replay.
241#[derive(Debug, Clone, Serialize, Deserialize)]
242pub struct BrokerEvent {
243    pub cursor: EventCursor,
244    pub event: TraverseEvent,
245}
246
247/// A broker subscription handle.
248#[derive(Debug, Clone, Serialize, Deserialize)]
249pub struct Subscription {
250    pub subscription_id: SubscriptionId,
251    pub event_type: String,
252    pub cursor: EventCursor,
253}
254
255/// Result of polling a subscription.
256#[derive(Debug, Clone, Serialize, Deserialize)]
257pub struct SubscriptionPoll {
258    pub subscription_id: SubscriptionId,
259    pub event_type: String,
260    pub cursor: EventCursor,
261    pub events: Vec<BrokerEvent>,
262}
263
264#[cfg(test)]
265mod tests {
266    #![allow(clippy::expect_used)]
267
268    use super::*;
269
270    fn sample_event(event_type: &str) -> TraverseEvent {
271        TraverseEvent {
272            id: "f0f83e66-4d87-4dd6-884d-0128d94f730f".to_string(),
273            source: "traverse-runtime".to_string(),
274            event_type: event_type.to_string(),
275            datacontenttype: "application/json".to_string(),
276            time: "2026-07-14T00:00:00Z".to_string(),
277            data: serde_json::json!({"execution_id": "exec_test"}),
278            owner: "traverse-runtime".to_string(),
279            version: "1.0.0".to_string(),
280            lifecycle_status: LifecycleStatus::Active,
281            subject_id: Some("subject_test".to_string()),
282            actor_id: Some("actor_test".to_string()),
283        }
284    }
285
286    #[test]
287    fn event_error_display_covers_all_variants() {
288        let cases: Vec<EventError> = vec![
289            EventError::LifecycleViolation("x".to_string()),
290            EventError::UnregisteredEventType("t".to_string()),
291            EventError::InvalidCursor("c".to_string()),
292            EventError::CursorExpired {
293                event_type: "evt".to_string(),
294                oldest_available_cursor: "7".to_string(),
295            },
296            EventError::SubscriptionNotFound("sub-1".to_string()),
297            EventError::InvalidRetentionWindow("bad".to_string()),
298            EventError::JournalWrite("disk gone".to_string()),
299            EventError::JournalWriteTimeout("exceeded 2000ms".to_string()),
300            EventError::JournalRead("disk gone".to_string()),
301        ];
302
303        for err in cases {
304            let rendered = err.to_string();
305            assert!(!rendered.is_empty());
306        }
307    }
308
309    #[test]
310    fn noop_runtime_event_sink_accepts_an_envelope() {
311        assert!(
312            NoopRuntimeEventSink
313                .emit(sample_event("dev.traverse.noop"))
314                .is_ok()
315        );
316    }
317
318    #[test]
319    fn broker_event_sink_forwards_the_original_envelope() {
320        let event_type = "dev.traverse.runtime.execution.completed";
321        let catalog = Arc::new(crate::events::EventCatalog::new());
322        catalog
323            .register(crate::events::EventCatalogEntry {
324                event_type: event_type.to_string(),
325                owner: "traverse-runtime".to_string(),
326                version: "1.0.0".to_string(),
327                lifecycle_status: LifecycleStatus::Active,
328                consumer_count: 0,
329            })
330            .expect("catalog registration must succeed");
331        let broker =
332            Arc::new(crate::events::InProcessBroker::new(catalog).expect("broker must be created"));
333        let sink = BrokerEventSink::new(broker.clone());
334        let event = sample_event(event_type);
335
336        sink.emit(event.clone())
337            .expect("sink delivery must succeed");
338        let subscription = broker
339            .subscribe_for_subject(event_type, "0", Some("subject_test"))
340            .expect("subject subscription must succeed");
341        let delivered = broker
342            .poll(&subscription.subscription_id, 1)
343            .expect("poll must succeed");
344
345        assert_eq!(format!("{sink:?}"), "BrokerEventSink { .. }");
346        assert_eq!(delivered.events.len(), 1);
347        assert_eq!(delivered.events[0].event.subject_id, event.subject_id);
348        assert_eq!(delivered.events[0].event.actor_id, event.actor_id);
349        assert_eq!(delivered.events[0].event.data, event.data);
350    }
351}
traverse_runtime/events/types.rs

traverse_runtime/events/
types.rs
