runifold_agent/
stream.rs

1use std::{
2    collections::VecDeque,
3    future::Future,
4    pin::Pin,
5    sync::{Arc, Mutex, MutexGuard},
6    task::{Context, Poll},
7};
8
9use futures_core::Stream;
10use runifold_core::Usage;
11use runifold_model::{ModelStreamEvent, ToolCall};
12use serde::{Deserialize, Serialize};
13
14use crate::{
15    AgentConversationError, AgentError, AgentFuture, AgentOutcome, ConversationVersion,
16    TerminalRequirementFailure, TerminalReviewVerdictKind,
17};
18
19/// The callable boundary represented by an Agent stream event.
20#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
21#[non_exhaustive]
22pub enum CallableKind {
23    /// A locally registered Tool.
24    Tool,
25    /// A child Agent route.
26    Agent,
27}
28
29/// One real-time event from the canonical Agent execution loop.
30#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
31#[non_exhaustive]
32pub enum AgentStreamEvent {
33    /// The Agent execution started.
34    Started {
35        /// Stable local Agent name.
36        agent: String,
37    },
38    /// A model turn started.
39    TurnStarted {
40        /// One-based turn number.
41        turn: u32,
42    },
43    /// One provider-neutral model streaming event.
44    Model {
45        /// One-based owning turn.
46        turn: u32,
47        /// Canonical event accepted by the model stream accumulator.
48        event: ModelStreamEvent,
49    },
50    /// One dynamic context source completed retrieval.
51    ContextRetrieved {
52        /// Operator-visible source name.
53        source: String,
54        /// Number of documents injected into the transcript.
55        documents: usize,
56    },
57    /// A Tool or child Agent call started.
58    CallableStarted {
59        /// One-based owning turn.
60        turn: u32,
61        /// Callable boundary.
62        kind: CallableKind,
63        /// Canonical model-emitted call.
64        call: ToolCall,
65    },
66    /// A Tool or child Agent call reached a recoverable terminal result.
67    CallableCompleted {
68        /// One-based owning turn.
69        turn: u32,
70        /// Callable boundary.
71        kind: CallableKind,
72        /// Model-emitted call identity.
73        call_id: String,
74        /// Model-facing callable name.
75        name: String,
76        /// Whether execution produced a successful output.
77        success: bool,
78    },
79    /// Shared run-tree resource usage changed.
80    UsageUpdated {
81        /// Latest cumulative usage snapshot.
82        usage: Usage,
83    },
84    /// The Agent reached a terminal model response.
85    Completed {
86        /// Complete canonical outcome.
87        outcome: AgentOutcome,
88    },
89    /// A checkpointed summary batch is about to run or resume.
90    ConversationSummaryStarted {
91        /// Stable summary Agent checkpoint identity.
92        checkpoint_id: runifold_core::CheckpointId,
93        /// Last transcript entry included in this batch.
94        through_sequence: crate::ConversationSequence,
95    },
96    /// A summary batch and its session progress have been persisted.
97    ConversationSummaryCommitted {
98        /// Last transcript entry covered by the committed summary.
99        through_sequence: crate::ConversationSequence,
100        /// Shared cumulative resource usage including summary generation.
101        usage: Usage,
102    },
103    /// The terminal checkpoint and conversation transcript committed atomically.
104    ConversationCommitted {
105        /// Complete canonical outcome.
106        outcome: AgentOutcome,
107        /// Committed transcript version.
108        conversation_version: ConversationVersion,
109    },
110    /// An invalid terminal candidate scheduled a bounded repair turn.
111    TerminalRepairScheduled {
112        /// One-based repair attempt number.
113        attempt: u32,
114        /// Safe reason the candidate was rejected.
115        failure: TerminalRequirementFailure,
116    },
117    /// Internal review started before a model response can affect execution.
118    TurnReviewStarted {
119        /// One-based model turn being reviewed.
120        turn: u32,
121    },
122    /// Internal review returned a validated verdict.
123    TurnReviewCompleted {
124        /// One-based model turn that was reviewed.
125        turn: u32,
126        /// Stable semantic verdict category.
127        verdict: TerminalReviewVerdictKind,
128    },
129    /// An internal review verdict scheduled a replacement model turn.
130    TurnReviewRepairScheduled {
131        /// Cumulative one-based internal repair number.
132        attempt: u32,
133        /// Model turn whose response was rejected.
134        turn: u32,
135    },
136    /// Semantic review of a locally valid terminal candidate started.
137    TerminalReviewStarted {
138        /// One-based review attempt.
139        attempt: u32,
140    },
141    /// Semantic review returned a validated verdict.
142    TerminalReviewCompleted {
143        /// One-based review attempt.
144        attempt: u32,
145        /// Stable semantic verdict category.
146        verdict: TerminalReviewVerdictKind,
147    },
148    /// A semantic review verdict scheduled a bounded regeneration turn.
149    TerminalReviewRepairScheduled {
150        /// One-based repair number.
151        attempt: u32,
152    },
153}
154
155pub(crate) trait AgentObserver: Send + Sync {
156    fn emit(&self, event: AgentStreamEvent);
157
158    fn backpressured(&self) -> bool {
159        false
160    }
161}
162
163#[derive(Debug)]
164pub(crate) struct NoopObserver;
165
166impl AgentObserver for NoopObserver {
167    fn emit(&self, _event: AgentStreamEvent) {}
168}
169
170#[derive(Clone, Debug, Default)]
171pub(crate) struct BufferedObserver {
172    suppress_completed: bool,
173    events: Arc<Mutex<VecDeque<AgentStreamEvent>>>,
174}
175
176impl BufferedObserver {
177    pub(crate) fn durable() -> Self {
178        Self {
179            suppress_completed: true,
180            ..Self::default()
181        }
182    }
183
184    pub(crate) fn events(&self) -> Arc<Mutex<VecDeque<AgentStreamEvent>>> {
185        self.events.clone()
186    }
187}
188
189impl AgentObserver for BufferedObserver {
190    fn emit(&self, event: AgentStreamEvent) {
191        if self.suppress_completed && matches!(event, AgentStreamEvent::Completed { .. }) {
192            return;
193        }
194        self.events
195            .lock()
196            .unwrap_or_else(std::sync::PoisonError::into_inner)
197            .push_back(event);
198    }
199
200    fn backpressured(&self) -> bool {
201        true
202    }
203}
204
205pub(crate) async fn emit_agent_event(observer: &dyn AgentObserver, event: AgentStreamEvent) {
206    observer.emit(event);
207    if observer.backpressured() {
208        YieldOnce::new().await;
209    }
210}
211
212struct YieldOnce {
213    yielded: bool,
214}
215
216impl YieldOnce {
217    const fn new() -> Self {
218        Self { yielded: false }
219    }
220}
221
222impl Future for YieldOnce {
223    type Output = ();
224
225    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
226        if self.yielded {
227            Poll::Ready(())
228        } else {
229            self.yielded = true;
230            context.waker().wake_by_ref();
231            Poll::Pending
232        }
233    }
234}
235
236/// A borrow-scoped stream that drives the canonical Agent loop when polled.
237#[must_use = "streams do nothing unless polled"]
238pub struct AgentEventStream<'a, E = AgentError> {
239    execution: Option<AgentFuture<'a, Result<AgentOutcome, E>>>,
240    events: Arc<Mutex<VecDeque<AgentStreamEvent>>>,
241    failure: Option<E>,
242    finished: bool,
243}
244
245impl<'a, E> AgentEventStream<'a, E> {
246    pub(crate) fn new(
247        execution: AgentFuture<'a, Result<AgentOutcome, E>>,
248        events: Arc<Mutex<VecDeque<AgentStreamEvent>>>,
249    ) -> Self {
250        Self {
251            execution: Some(execution),
252            events,
253            failure: None,
254            finished: false,
255        }
256    }
257
258    fn events(&self) -> MutexGuard<'_, VecDeque<AgentStreamEvent>> {
259        self.events
260            .lock()
261            .unwrap_or_else(std::sync::PoisonError::into_inner)
262    }
263
264    fn pop_event(&self) -> Option<AgentStreamEvent> {
265        self.events().pop_front()
266    }
267}
268
269impl<E: Unpin> Stream for AgentEventStream<'_, E> {
270    type Item = Result<AgentStreamEvent, E>;
271
272    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
273        let this = self.get_mut();
274        if let Some(event) = this.pop_event() {
275            return Poll::Ready(Some(Ok(event)));
276        }
277        if let Some(execution) = this.execution.as_mut() {
278            match execution.as_mut().poll(context) {
279                Poll::Pending => {
280                    return this
281                        .pop_event()
282                        .map_or(Poll::Pending, |event| Poll::Ready(Some(Ok(event))));
283                }
284                Poll::Ready(Ok(_outcome)) => {
285                    this.execution = None;
286                }
287                Poll::Ready(Err(error)) => {
288                    this.execution = None;
289                    this.failure = Some(error);
290                }
291            }
292        }
293        if let Some(event) = this.pop_event() {
294            return Poll::Ready(Some(Ok(event)));
295        }
296        if let Some(error) = this.failure.take() {
297            return Poll::Ready(Some(Err(error)));
298        }
299        if this.execution.is_none() {
300            this.finished = true;
301        }
302        if this.finished {
303            Poll::Ready(None)
304        } else {
305            Poll::Pending
306        }
307    }
308}
309
310impl<E> std::fmt::Debug for AgentEventStream<'_, E> {
311    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
312        formatter
313            .debug_struct("AgentEventStream")
314            .field("queued_events", &self.events().len())
315            .field("has_execution", &self.execution.is_some())
316            .field("has_failure", &self.failure.is_some())
317            .field("finished", &self.finished)
318            .finish()
319    }
320}
321
322/// A poll-driven durable conversation stream. Dropping it stops local execution;
323/// remote work already dispatched may remain ambiguous and requires recovery.
324/// Success is reported only by [`AgentStreamEvent::ConversationCommitted`].
325pub type DurableConversationEventStream<'a> = AgentEventStream<'a, AgentConversationError>;
runifold_agent/stream.rs

runifold_agent/
stream.rs
