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

starweaver_runtime/
stream.rs
