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

starweaver_runtime/
stream.rs
