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

traverse_runtime/events/
types.rs
