agent_sdk_core/ports/
subscription.rs

1//! Host adapter boundaries for the SDK core. Use these traits and registries when
2//! hosts provide providers, journals, sinks, tools, isolation, extensions, telemetry,
3//! or subscriptions. Implementations may perform external side effects and must honor
4//! policy, redaction, idempotency, and replay contracts. This file contains the
5//! subscription portion of that contract.
6//!
7use std::sync::{Arc, Mutex};
8
9use crate::{
10    domain::{
11        AgentError, AgentId, DestinationKind, DestinationRef, EntityRef, EventId, JournalCursor,
12        RunId, SourceKind, SourceRef, SpanId, TraceId,
13    },
14    event::{
15        AgentEvent, CompiledEventFilter, ContentCaptureMode, EVENT_SCHEMA_VERSION, EventCursor,
16        EventDeliverySemantics, EventEnvelope, EventFamily, EventFilter, EventFrame, EventKind,
17        EventStreamScope, cursor_compatible,
18    },
19    event_bus::AgentEventStream,
20};
21
22/// Port or behavior contract for run subscription source. Implementors
23/// should preserve policy, redaction, idempotency, and replay
24/// expectations from the surrounding module. Implementations may
25/// perform side effects only as described by the trait methods.
26pub trait RunSubscriptionSource: Send + Sync {
27    /// Creates a read-only event stream for all visible events.
28    /// This is data-only and does not perform I/O, call host ports, append journals, publish
29    /// events, or start processes.
30    fn subscribe_all(&self, cursor: Option<EventCursor>) -> Result<AgentEventStream, AgentError>;
31
32    /// Creates a read-only event stream scoped to one run.
33    /// This is data-only and does not perform I/O, call host ports, append journals, publish
34    /// events, or start processes.
35    fn subscribe_run(
36        &self,
37        run_id: RunId,
38        cursor: Option<EventCursor>,
39    ) -> Result<AgentEventStream, AgentError>;
40
41    /// Creates a read-only event stream scoped to one agent.
42    /// Implementations create a read-only subscription or replay stream; the call must not
43    /// start runs, publish events, or append journal records.
44    fn subscribe_agent(
45        &self,
46        agent_id: AgentId,
47        cursor: Option<EventCursor>,
48    ) -> Result<AgentEventStream, AgentError>;
49
50    /// Creates a read-only event stream matching a compiled filter.
51    /// Implementations create a read-only subscription or replay stream; the call must not
52    /// start runs, publish events, or append journal records.
53    fn subscribe_events(
54        &self,
55        filter: CompiledEventFilter,
56        cursor: Option<EventCursor>,
57    ) -> Result<AgentEventStream, AgentError>;
58
59    /// Replays one run from durable cursor state.
60    /// Implementations create a read-only subscription or replay stream; the call must not
61    /// start runs, publish events, or append journal records.
62    fn replay_run_from_cursor(
63        &self,
64        run_id: RunId,
65        cursor: JournalCursor,
66    ) -> Result<AgentEventStream, AgentError>;
67
68    /// Reads the latest terminal event for one run, when available.
69    /// Implementations read subscription state for the latest terminal frame; the lookup must
70    /// not publish events or alter run state.
71    fn latest_terminal_event(&self, run_id: &RunId) -> Result<Option<EventFrame>, AgentError>;
72}
73
74#[derive(Clone, Debug, Default)]
75/// Carries in memory subscription hub data across a host-port boundary.
76/// Constructing the value does not call the host; the port method that receives it documents any adapter, network, or storage effect.
77pub struct InMemorySubscriptionHub {
78    frames: Arc<Mutex<Vec<EventFrame>>>,
79    live_floor_seq: Arc<Mutex<u64>>,
80}
81
82impl InMemorySubscriptionHub {
83    /// Mutates the in-memory event/subscription state and may wake local
84    /// subscribers. It does not persist durable journal truth or call network
85    /// sinks.
86    pub fn publish(&self, frame: EventFrame) -> Result<(), AgentError> {
87        self.frames
88            .lock()
89            .map_err(|_| AgentError::contract_violation("subscription hub lock poisoned"))?
90            .push(frame);
91        Ok(())
92    }
93
94    /// Mutates the in-memory event/subscription state and may wake local
95    /// subscribers. It does not persist durable journal truth or call network
96    /// sinks.
97    pub fn publish_all(
98        &self,
99        frames: impl IntoIterator<Item = EventFrame>,
100    ) -> Result<(), AgentError> {
101        let mut locked = self
102            .frames
103            .lock()
104            .map_err(|_| AgentError::contract_violation("subscription hub lock poisoned"))?;
105        locked.extend(frames);
106        Ok(())
107    }
108
109    /// Mutates the in-memory event/subscription state and may wake local
110    /// subscribers. It does not persist durable journal truth or call network
111    /// sinks.
112    pub fn expire_live_before(&self, event_seq: u64) -> Result<(), AgentError> {
113        *self
114            .live_floor_seq
115            .lock()
116            .map_err(|_| AgentError::contract_violation("subscription hub lock poisoned"))? =
117            event_seq;
118        Ok(())
119    }
120
121    /// Builds the frames value.
122    /// This is data construction and performs no I/O, journal append, event publication, or
123    /// process work.
124    pub fn frames(&self) -> Result<Vec<EventFrame>, AgentError> {
125        Ok(self
126            .frames
127            .lock()
128            .map_err(|_| AgentError::contract_violation("subscription hub lock poisoned"))?
129            .clone())
130    }
131
132    fn subscribe_scope(
133        &self,
134        requested_scope: EventStreamScope,
135        filter: CompiledEventFilter,
136        cursor: Option<EventCursor>,
137    ) -> Result<AgentEventStream, AgentError> {
138        cursor_compatible(&requested_scope, cursor.as_ref())?;
139
140        if let Some(cursor) = cursor.as_ref() {
141            if self.cursor_expired(cursor)? {
142                return self.resume_expired_cursor(requested_scope, filter, cursor.clone());
143            }
144        }
145
146        let start_after = cursor.as_ref().map(|cursor| cursor.event_seq);
147        let live_floor = self.live_floor()?;
148        Ok(AgentEventStream::new(
149            self.frames()?
150                .into_iter()
151                .filter(|frame| frame.event.envelope.event_seq >= live_floor)
152                .filter(|frame| {
153                    start_after.is_none_or(|event_seq| frame.event.envelope.event_seq > event_seq)
154                })
155                .filter(|frame| filter.matches_envelope(&frame.event.envelope))
156                .map(|frame| frame_for_scope(frame, requested_scope.clone())),
157        ))
158    }
159
160    fn resume_expired_cursor(
161        &self,
162        requested_scope: EventStreamScope,
163        filter: CompiledEventFilter,
164        cursor: EventCursor,
165    ) -> Result<AgentEventStream, AgentError> {
166        match (&requested_scope, cursor.journal_cursor.as_ref()) {
167            (EventStreamScope::Run(run_id), Some(journal_cursor)) => {
168                let live_floor = self.live_floor()?;
169                let mut frames = self
170                    .journal_backed_run_frames_after(run_id, journal_cursor)?
171                    .into_iter()
172                    .filter(|frame| frame.event.envelope.event_seq < live_floor)
173                    .map(|frame| derived_replay_frame(frame, requested_scope.clone()))
174                    .collect::<Vec<_>>();
175                frames.extend(
176                    self.frames()?
177                        .into_iter()
178                        .filter(|frame| frame.event.envelope.event_seq >= live_floor)
179                        .filter(|frame| frame.event.envelope.event_seq > cursor.event_seq)
180                        .filter(|frame| filter.matches_envelope(&frame.event.envelope))
181                        .map(|frame| frame_for_scope(frame, requested_scope.clone())),
182                );
183                Ok(AgentEventStream::new(frames))
184            }
185            (EventStreamScope::Run(run_id), None) => {
186                Ok(AgentEventStream::new([self.gap_diagnostic_frame(
187                    run_id.clone(),
188                    requested_scope,
189                    cursor,
190                )?]))
191            }
192            (_, Some(_)) => Err(AgentError::host_configuration_needed(
193                "host archive required for expired non-run event cursor replay",
194            )),
195            (_, None) => Err(AgentError::host_configuration_needed(
196                "expired non-run event cursor has no durable archive cursor",
197            )),
198        }
199    }
200
201    fn replay_run(
202        &self,
203        run_id: &RunId,
204        cursor: &JournalCursor,
205    ) -> Result<Vec<EventFrame>, AgentError> {
206        self.journal_backed_run_frames_after(run_id, cursor)
207            .map(|frames| {
208                frames
209                    .into_iter()
210                    .map(|frame| derived_replay_frame(frame, EventStreamScope::Run(run_id.clone())))
211                    .collect()
212            })
213    }
214
215    fn journal_backed_run_frames_after(
216        &self,
217        run_id: &RunId,
218        cursor: &JournalCursor,
219    ) -> Result<Vec<EventFrame>, AgentError> {
220        let cursor_seq = journal_cursor_seq(cursor);
221        Ok(self
222            .frames()?
223            .into_iter()
224            .filter(|frame| &frame.event.envelope.run_id == run_id)
225            .filter(|frame| {
226                frame
227                    .event
228                    .envelope
229                    .journal_cursor
230                    .as_ref()
231                    .is_some_and(|journal_cursor| journal_cursor_seq(journal_cursor) > cursor_seq)
232            })
233            .collect())
234    }
235
236    fn cursor_expired(&self, cursor: &EventCursor) -> Result<bool, AgentError> {
237        Ok(cursor.event_seq < self.live_floor()?)
238    }
239
240    fn live_floor(&self) -> Result<u64, AgentError> {
241        Ok(*self
242            .live_floor_seq
243            .lock()
244            .map_err(|_| AgentError::contract_violation("subscription hub lock poisoned"))?)
245    }
246
247    fn gap_diagnostic_frame(
248        &self,
249        run_id: RunId,
250        requested_scope: EventStreamScope,
251        cursor: EventCursor,
252    ) -> Result<EventFrame, AgentError> {
253        let agent_id = self
254            .frames()?
255            .into_iter()
256            .rev()
257            .find(|frame| frame.event.envelope.run_id == run_id)
258            .map(|frame| frame.event.envelope.agent_id)
259            .unwrap_or_else(|| AgentId::new("agent.replay.unknown"));
260        let next_seq = cursor.event_seq.saturating_add(1);
261        let event = AgentEvent::with_redacted_summary(
262            EventEnvelope {
263                schema_version: EVENT_SCHEMA_VERSION,
264                event_id: EventId::new(format!(
265                    "event.replay_failed.{}.{}",
266                    run_id.as_str(),
267                    next_seq
268                )),
269                event_seq: next_seq,
270                event_family: EventFamily::Recovery,
271                event_kind: EventKind::ReplayFailed,
272                payload_schema_version: 1,
273                timestamp: "1970-01-01T00:00:00Z".to_string(),
274                recorded_at: "1970-01-01T00:00:00Z".to_string(),
275                run_id: run_id.clone(),
276                session_id: None,
277                agent_id,
278                turn_id: None,
279                attempt_id: None,
280                message_id: None,
281                context_item_id: None,
282                trace_id: TraceId::new(format!("trace.replay_failed.{}", run_id.as_str())),
283                span_id: SpanId::new(format!("span.replay_failed.{next_seq}")),
284                parent_event_id: Some(cursor.event_id),
285                caused_by: None,
286                subject_ref: EntityRef::run(run_id),
287                related_refs: Vec::new(),
288                causal_refs: Vec::new(),
289                correlation: Default::default(),
290                tags: Vec::new(),
291                source: SourceRef::with_kind(SourceKind::Replay, "source.replay.subscription"),
292                destination: Some(DestinationRef::with_kind(
293                    DestinationKind::EventStream,
294                    "destination.event_stream.subscription",
295                )),
296                policy_refs: Vec::new(),
297                journal_cursor: None,
298                state_before: None,
299                state_after: None,
300                delivery_semantics: EventDeliverySemantics::DiagnosticOnly,
301                privacy: crate::domain::PrivacyClass::ContentRefsOnly,
302                content_capture: ContentCaptureMode::Off,
303                redaction_policy_id: "policy.redaction.default".to_string(),
304                runtime_package_fingerprint: "runtime.package.fingerprint.subscription".to_string(),
305            },
306            "live cursor expired and no journal cursor was available for run replay",
307        );
308        Ok(frame_for_scope(
309            EventFrame {
310                cursor: event.envelope.cursor(requested_scope.clone()),
311                event,
312                archive_cursor: None,
313                overflow: None,
314            },
315            requested_scope,
316        ))
317    }
318}
319
320impl RunSubscriptionSource for InMemorySubscriptionHub {
321    fn subscribe_all(&self, cursor: Option<EventCursor>) -> Result<AgentEventStream, AgentError> {
322        self.subscribe_scope(
323            EventStreamScope::All,
324            EventFilter::default().compile()?,
325            cursor,
326        )
327    }
328
329    fn subscribe_run(
330        &self,
331        run_id: RunId,
332        cursor: Option<EventCursor>,
333    ) -> Result<AgentEventStream, AgentError> {
334        self.subscribe_scope(
335            EventStreamScope::Run(run_id.clone()),
336            EventFilter::run(run_id).compile()?,
337            cursor,
338        )
339    }
340
341    fn subscribe_agent(
342        &self,
343        agent_id: AgentId,
344        cursor: Option<EventCursor>,
345    ) -> Result<AgentEventStream, AgentError> {
346        self.subscribe_scope(
347            EventStreamScope::Agent(agent_id.clone()),
348            EventFilter::agent(agent_id).compile()?,
349            cursor,
350        )
351    }
352
353    fn subscribe_events(
354        &self,
355        filter: CompiledEventFilter,
356        cursor: Option<EventCursor>,
357    ) -> Result<AgentEventStream, AgentError> {
358        self.subscribe_scope(filter.cursor_scope(), filter, cursor)
359    }
360
361    fn replay_run_from_cursor(
362        &self,
363        run_id: RunId,
364        cursor: JournalCursor,
365    ) -> Result<AgentEventStream, AgentError> {
366        Ok(AgentEventStream::new(self.replay_run(&run_id, &cursor)?))
367    }
368
369    fn latest_terminal_event(&self, run_id: &RunId) -> Result<Option<EventFrame>, AgentError> {
370        Ok(self.frames()?.into_iter().rev().find(|frame| {
371            &frame.event.envelope.run_id == run_id
372                && matches!(
373                    frame.event.envelope.event_kind,
374                    EventKind::RunCompleted | EventKind::RunFailed | EventKind::RunCancelled
375                )
376        }))
377    }
378}
379
380fn frame_for_scope(mut frame: EventFrame, scope: EventStreamScope) -> EventFrame {
381    frame.cursor = frame.event.envelope.cursor(scope);
382    frame
383}
384
385fn derived_replay_frame(mut frame: EventFrame, scope: EventStreamScope) -> EventFrame {
386    frame.event.envelope.delivery_semantics = EventDeliverySemantics::DerivedReplay;
387    frame.cursor = frame.event.envelope.cursor(scope);
388    frame
389}
390
391fn journal_cursor_seq(cursor: &JournalCursor) -> u64 {
392    cursor
393        .as_str()
394        .rsplit_once('.')
395        .and_then(|(_, seq)| seq.parse::<u64>().ok())
396        .unwrap_or(0)
397}
agent_sdk_core/ports/subscription.rs

agent_sdk_core/ports/
subscription.rs
