starweaver_runtime/
stream.rs

1//! Typed agent stream event foundations.
2
3use std::sync::{Arc, Mutex, PoisonError};
4
5use serde::{Deserialize, Serialize};
6use starweaver_context::AgentEvent;
7use starweaver_core::{AgentId, ConversationId, Metadata, RunId, TaskId};
8use starweaver_model::{ModelResponse, ModelResponseStreamEvent, ToolCallPart, ToolReturnPart};
9
10use crate::{executor::AgentExecutionNode, run::RunStatus, AgentResult};
11
12/// Stable category for context sideband events that are bridged into the runtime stream.
13#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
14#[serde(rename_all = "snake_case")]
15pub enum AgentSidebandEventCategory {
16    /// Run lifecycle and status events.
17    Run,
18    /// Model retry or model-side diagnostics.
19    Model,
20    /// Tool availability or direct tool lifecycle events.
21    Tool,
22    /// Dynamic tool-search discovery, loading, initialization, or refresh events.
23    ToolSearch,
24    /// Human-in-the-loop approval or deferred-tool decision events.
25    Hitl,
26    /// Skill scan, activation, or reload events.
27    Skill,
28    /// Task list or task state events.
29    Task,
30    /// Note state events.
31    Note,
32    /// File state or file-operation events.
33    File,
34    /// Media state or media-operation events.
35    Media,
36    /// Host adapter or host operation events.
37    HostOperation,
38    /// Subagent lifecycle events.
39    Subagent,
40    /// Message bus events.
41    Message,
42    /// Usage snapshot or usage-limit events.
43    Usage,
44    /// Context compaction lifecycle events.
45    Compact,
46    /// User steering events.
47    Steering,
48    /// Background process or background shell events.
49    Background,
50    /// Capability-specific events that do not fit a narrower category.
51    Capability,
52}
53
54/// Typed view of an application sideband event carried by [`AgentStreamEvent::Custom`].
55#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
56pub struct AgentSidebandEvent {
57    /// Stable event category.
58    pub category: AgentSidebandEventCategory,
59    /// Event type.
60    pub kind: String,
61    /// Event payload.
62    #[serde(default)]
63    pub payload: serde_json::Value,
64    /// Event metadata.
65    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
66    pub metadata: Metadata,
67}
68
69impl AgentSidebandEvent {
70    /// Build a typed sideband event when the context event kind belongs to the stable taxonomy.
71    #[must_use]
72    pub fn from_agent_event(event: &AgentEvent) -> Option<Self> {
73        Self::category_for_kind(&event.kind).map(|category| Self {
74            category,
75            kind: event.kind.clone(),
76            payload: event.payload.clone(),
77            metadata: event.metadata.clone(),
78        })
79    }
80
81    /// Classify a context event kind into the stable sideband taxonomy.
82    #[must_use]
83    pub fn category_for_kind(kind: &str) -> Option<AgentSidebandEventCategory> {
84        match kind {
85            "run_start" | "run_complete" | "run_failed" | "run_waiting" | "run_cancelled" => {
86                Some(AgentSidebandEventCategory::Run)
87            }
88            "model_error_retry" | "model_stream_resume" => Some(AgentSidebandEventCategory::Model),
89            "tools_unavailable"
90            | "toolset_initialized"
91            | "toolset_unavailable"
92            | "toolset_failed"
93            | "toolset_refreshed"
94            | "toolset_closed" => Some(AgentSidebandEventCategory::Tool),
95            "tool_search_loaded" | "tool_search_initialized" | "tool_search_refreshed" => {
96                Some(AgentSidebandEventCategory::ToolSearch)
97            }
98            "hitl_resolved" => Some(AgentSidebandEventCategory::Hitl),
99            "skills_scanned" | "skill_activated" | "skills_reloaded" => {
100                Some(AgentSidebandEventCategory::Skill)
101            }
102            "task_snapshot" => Some(AgentSidebandEventCategory::Task),
103            "usage_snapshot" => Some(AgentSidebandEventCategory::Usage),
104            "compact_start" | "compact_failed" | "compact_complete" => {
105                Some(AgentSidebandEventCategory::Compact)
106            }
107            "steering_received" | "steering_submitted" => {
108                Some(AgentSidebandEventCategory::Steering)
109            }
110            "background_shell_complete" => Some(AgentSidebandEventCategory::Background),
111            "message_received" => Some(AgentSidebandEventCategory::Message),
112            "subagent_started" | "subagent_completed" | "subagent_failed" => {
113                Some(AgentSidebandEventCategory::Subagent)
114            }
115            _ if kind.starts_with("task_") => Some(AgentSidebandEventCategory::Task),
116            _ if kind.starts_with("note_") => Some(AgentSidebandEventCategory::Note),
117            _ if kind.starts_with("file_") => Some(AgentSidebandEventCategory::File),
118            _ if kind.starts_with("media_") => Some(AgentSidebandEventCategory::Media),
119            _ if kind.starts_with("host_") => Some(AgentSidebandEventCategory::HostOperation),
120            _ if kind.starts_with("tool_search_") => Some(AgentSidebandEventCategory::ToolSearch),
121            _ if kind.starts_with("tool_") || kind.starts_with("toolset_") => {
122                Some(AgentSidebandEventCategory::Tool)
123            }
124            _ if kind.starts_with("approval_")
125                || kind.starts_with("deferred_")
126                || kind.starts_with("hitl_") =>
127            {
128                Some(AgentSidebandEventCategory::Hitl)
129            }
130            _ if kind.starts_with("skill_") || kind.starts_with("skills_") => {
131                Some(AgentSidebandEventCategory::Skill)
132            }
133            _ if kind.starts_with("subagent_") => Some(AgentSidebandEventCategory::Subagent),
134            _ if kind.starts_with("message_") => Some(AgentSidebandEventCategory::Message),
135            _ if kind.starts_with("usage_") => Some(AgentSidebandEventCategory::Usage),
136            _ if kind.starts_with("compact_") => Some(AgentSidebandEventCategory::Compact),
137            _ if kind.starts_with("steering_") => Some(AgentSidebandEventCategory::Steering),
138            _ if kind.starts_with("background_") => Some(AgentSidebandEventCategory::Background),
139            _ if kind.starts_with("capability_") => Some(AgentSidebandEventCategory::Capability),
140            _ => None,
141        }
142    }
143}
144
145/// Typed event emitted by the agent runtime while a run progresses.
146#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
147#[serde(tag = "kind", rename_all = "snake_case")]
148pub enum AgentStreamEvent {
149    /// A run started.
150    RunStart {
151        /// Run identifier.
152        run_id: RunId,
153        /// Conversation identifier.
154        conversation_id: ConversationId,
155    },
156    /// Runtime execution entered a durable node boundary.
157    NodeStart {
158        /// Execution boundary being entered.
159        node: AgentExecutionNode,
160        /// Completed run step at this boundary.
161        step: usize,
162        /// Current run status at this boundary.
163        status: RunStatus,
164    },
165    /// Runtime execution completed a durable node boundary.
166    NodeComplete {
167        /// Execution boundary that completed.
168        node: AgentExecutionNode,
169        /// Completed run step at this boundary.
170        step: usize,
171        /// Current run status after this boundary.
172        status: RunStatus,
173    },
174    /// A context sideband event was published during the run.
175    Custom {
176        /// Application or capability event.
177        event: AgentEvent,
178    },
179    /// A model request was prepared for a loop step.
180    ModelRequest {
181        /// Completed run step before sending the request.
182        step: usize,
183    },
184    /// A model response stream event was received.
185    ModelStream {
186        /// Completed run step for the active model request.
187        step: usize,
188        /// Canonical model stream event.
189        event: ModelResponseStreamEvent,
190    },
191    /// A model response was received.
192    ModelResponse {
193        /// Completed run step after receiving the response.
194        step: usize,
195        /// Canonical model response.
196        response: ModelResponse,
197    },
198    /// A durable execution checkpoint was persisted or inspected.
199    Checkpoint {
200        /// Execution boundary.
201        node: AgentExecutionNode,
202        /// Completed run step at this boundary.
203        step: usize,
204    },
205    /// Execution was suspended at a durable checkpoint.
206    Suspended {
207        /// Execution boundary that requested suspension.
208        node: AgentExecutionNode,
209        /// Human-readable suspend reason.
210        reason: String,
211    },
212    /// A model requested a function tool call.
213    ToolCall {
214        /// Current run step.
215        step: usize,
216        /// Tool call part.
217        call: ToolCallPart,
218    },
219    /// A function tool returned a result or structured control-flow error.
220    ToolReturn {
221        /// Current run step.
222        step: usize,
223        /// Tool return part.
224        tool_return: ToolReturnPart,
225    },
226    /// Output validation or output function validation requested another model turn.
227    OutputRetry {
228        /// Retry count after this retry was scheduled.
229        retries: usize,
230        /// Retry prompt sent to the model.
231        prompt: String,
232    },
233    /// Pending user steering requested another model turn before finalization.
234    SteeringGuard {
235        /// Current run step.
236        step: usize,
237        /// Control prompt sent to the model before finalization.
238        prompt: String,
239    },
240    /// A run completed successfully.
241    RunComplete {
242        /// Run identifier.
243        run_id: RunId,
244        /// Final output text.
245        output: String,
246    },
247    /// A run failed after preserving recoverable context state.
248    RunFailed {
249        /// Run identifier.
250        run_id: RunId,
251        /// Failure kind.
252        error_kind: String,
253        /// Human-readable error message.
254        message: String,
255    },
256}
257
258impl AgentStreamEvent {
259    /// Return a typed sideband event view for known context events carried by `Custom`.
260    #[must_use]
261    pub fn sideband_event(&self) -> Option<AgentSidebandEvent> {
262        match self {
263            Self::Custom { event } => AgentSidebandEvent::from_agent_event(event),
264            _ => None,
265        }
266    }
267}
268
269/// Origin type for a stream record emitted by a nested runtime component.
270#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
271#[serde(rename_all = "snake_case")]
272pub enum AgentStreamSourceKind {
273    /// Record emitted by a delegated subagent run.
274    Subagent,
275}
276
277/// Source attribution for records merged from nested agent runs.
278#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
279pub struct AgentStreamSource {
280    /// Source category.
281    pub kind: AgentStreamSourceKind,
282    /// Source agent identifier.
283    pub agent_id: AgentId,
284    /// Human-readable source agent name.
285    pub agent_name: String,
286    /// Delegated task identifier when the source is task scoped.
287    #[serde(default, skip_serializing_if = "Option::is_none")]
288    pub task_id: Option<TaskId>,
289    /// Source run identifier when known.
290    #[serde(default, skip_serializing_if = "Option::is_none")]
291    pub run_id: Option<RunId>,
292    /// Parent run identifier when known.
293    #[serde(default, skip_serializing_if = "Option::is_none")]
294    pub parent_run_id: Option<RunId>,
295    /// Original sequence number in the source run before parent stream rebasing.
296    pub source_sequence: usize,
297}
298
299impl AgentStreamSource {
300    /// Build subagent source attribution.
301    #[must_use]
302    pub fn subagent(
303        agent_id: AgentId,
304        agent_name: impl Into<String>,
305        task_id: TaskId,
306        run_id: Option<RunId>,
307        parent_run_id: Option<RunId>,
308        source_sequence: usize,
309    ) -> Self {
310        Self {
311            kind: AgentStreamSourceKind::Subagent,
312            agent_id,
313            agent_name: agent_name.into(),
314            task_id: Some(task_id),
315            run_id,
316            parent_run_id,
317            source_sequence,
318        }
319    }
320}
321
322/// In-memory sink used by tools that need to merge child stream records into the parent stream.
323#[derive(Clone, Debug, Default)]
324pub struct AgentStreamSink {
325    records: Arc<Mutex<Vec<AgentStreamRecord>>>,
326}
327
328impl AgentStreamSink {
329    /// Add one record to the sink.
330    pub fn push(&self, record: AgentStreamRecord) {
331        self.records
332            .lock()
333            .unwrap_or_else(PoisonError::into_inner)
334            .push(record);
335    }
336
337    /// Add several records to the sink.
338    pub fn extend(&self, records: impl IntoIterator<Item = AgentStreamRecord>) {
339        self.records
340            .lock()
341            .unwrap_or_else(PoisonError::into_inner)
342            .extend(records);
343    }
344
345    /// Drain all currently buffered records.
346    #[must_use]
347    pub fn drain(&self) -> Vec<AgentStreamRecord> {
348        self.records
349            .lock()
350            .unwrap_or_else(PoisonError::into_inner)
351            .drain(..)
352            .collect()
353    }
354
355    /// Return whether no records are buffered.
356    #[must_use]
357    pub fn is_empty(&self) -> bool {
358        self.records
359            .lock()
360            .unwrap_or_else(PoisonError::into_inner)
361            .is_empty()
362    }
363}
364
365/// Sequenced stream event record.
366#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
367pub struct AgentStreamRecord {
368    /// Monotonic event sequence number within one run.
369    pub sequence: usize,
370    /// Source attribution for records merged from nested runs.
371    #[serde(default, skip_serializing_if = "Option::is_none")]
372    pub source: Option<AgentStreamSource>,
373    /// Typed event payload.
374    pub event: AgentStreamEvent,
375}
376
377impl AgentStreamRecord {
378    /// Create a sequenced stream record.
379    #[must_use]
380    pub const fn new(sequence: usize, event: AgentStreamEvent) -> Self {
381        Self {
382            sequence,
383            source: None,
384            event,
385        }
386    }
387
388    /// Attach source attribution.
389    #[must_use]
390    pub fn with_source(mut self, source: AgentStreamSource) -> Self {
391        self.source = Some(source);
392        self
393    }
394
395    /// Replace the parent stream sequence after merging into another stream.
396    #[must_use]
397    pub const fn with_sequence(mut self, sequence: usize) -> Self {
398        self.sequence = sequence;
399        self
400    }
401}
402
403/// Result returned by collection-based stream runs.
404#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
405pub struct AgentStreamResult {
406    /// Final agent result.
407    pub result: AgentResult,
408    /// Events captured while the run progressed.
409    pub events: Vec<AgentStreamRecord>,
410}
411
412impl AgentStreamResult {
413    /// Return captured stream records.
414    #[must_use]
415    pub fn events(&self) -> &[AgentStreamRecord] {
416        &self.events
417    }
418
419    /// Return the final result.
420    #[must_use]
421    pub const fn result(&self) -> &AgentResult {
422        &self.result
423    }
424}
425
426pub(crate) fn push_stream_event(
427    events: &mut Option<&mut Vec<AgentStreamRecord>>,
428    event: AgentStreamEvent,
429) {
430    if let Some(events) = events.as_deref_mut() {
431        events.push(AgentStreamRecord::new(events.len(), event));
432    }
433}
434
435pub(crate) fn push_stream_record(
436    events: &mut Option<&mut Vec<AgentStreamRecord>>,
437    record: AgentStreamRecord,
438) {
439    if let Some(events) = events.as_deref_mut() {
440        events.push(record.with_sequence(events.len()));
441    }
442}
starweaver_runtime/stream.rs

starweaver_runtime/
stream.rs
