starweaver_stream/
raw.rs

1//! Typed raw agent stream protocol.
2
3use std::sync::{Arc, Mutex, PoisonError};
4
5use serde::{Deserialize, Serialize};
6use starweaver_core::{
7    AgentEvent, AgentExecutionNode, AgentId, ConversationId, Metadata, RunId,
8    RunLifecycle as RunStatus, TaskId,
9};
10use starweaver_model::{ModelResponse, ModelResponseStreamEvent, ToolCallPart, ToolReturnPart};
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 was cooperatively cancelled after preserving recoverable context state.
256    RunCancelled {
257        /// Run identifier.
258        run_id: RunId,
259        /// Human-readable cancellation reason.
260        reason: String,
261    },
262    /// A run failed after preserving recoverable context state.
263    RunFailed {
264        /// Run identifier.
265        run_id: RunId,
266        /// Failure kind.
267        error_kind: String,
268        /// Human-readable error message.
269        message: String,
270    },
271}
272
273impl AgentStreamEvent {
274    /// Return a typed sideband event view for known context events carried by `Custom`.
275    #[must_use]
276    pub fn sideband_event(&self) -> Option<AgentSidebandEvent> {
277        match self {
278            Self::Custom { event } => AgentSidebandEvent::from_agent_event(event),
279            _ => None,
280        }
281    }
282}
283
284/// Origin type for a stream record emitted by a nested runtime component.
285#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
286#[serde(rename_all = "snake_case")]
287pub enum AgentStreamSourceKind {
288    /// Record emitted by a delegated subagent run.
289    Subagent,
290}
291
292/// Source attribution for records merged from nested agent runs.
293#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
294pub struct AgentStreamSource {
295    /// Source category.
296    pub kind: AgentStreamSourceKind,
297    /// Source agent identifier.
298    pub agent_id: AgentId,
299    /// Human-readable source agent name.
300    pub agent_name: String,
301    /// Delegated task identifier when the source is task scoped.
302    #[serde(default, skip_serializing_if = "Option::is_none")]
303    pub task_id: Option<TaskId>,
304    /// Source run identifier when known.
305    #[serde(default, skip_serializing_if = "Option::is_none")]
306    pub run_id: Option<RunId>,
307    /// Parent run identifier when known.
308    #[serde(default, skip_serializing_if = "Option::is_none")]
309    pub parent_run_id: Option<RunId>,
310    /// Original sequence number in the source run before parent stream rebasing.
311    pub source_sequence: usize,
312}
313
314impl AgentStreamSource {
315    /// Build subagent source attribution.
316    #[must_use]
317    pub fn subagent(
318        agent_id: AgentId,
319        agent_name: impl Into<String>,
320        task_id: TaskId,
321        run_id: Option<RunId>,
322        parent_run_id: Option<RunId>,
323        source_sequence: usize,
324    ) -> Self {
325        Self {
326            kind: AgentStreamSourceKind::Subagent,
327            agent_id,
328            agent_name: agent_name.into(),
329            task_id: Some(task_id),
330            run_id,
331            parent_run_id,
332            source_sequence,
333        }
334    }
335}
336
337/// In-memory sink used by tools that need to merge child stream records into the parent stream.
338#[derive(Clone, Debug, Default)]
339pub struct AgentStreamSink {
340    records: Arc<Mutex<Vec<AgentStreamRecord>>>,
341}
342
343impl AgentStreamSink {
344    /// Add one record to the sink.
345    pub fn push(&self, record: AgentStreamRecord) {
346        self.records
347            .lock()
348            .unwrap_or_else(PoisonError::into_inner)
349            .push(record);
350    }
351
352    /// Add several records to the sink.
353    pub fn extend(&self, records: impl IntoIterator<Item = AgentStreamRecord>) {
354        self.records
355            .lock()
356            .unwrap_or_else(PoisonError::into_inner)
357            .extend(records);
358    }
359
360    /// Drain all currently buffered records.
361    #[must_use]
362    pub fn drain(&self) -> Vec<AgentStreamRecord> {
363        self.records
364            .lock()
365            .unwrap_or_else(PoisonError::into_inner)
366            .drain(..)
367            .collect()
368    }
369
370    /// Return whether no records are buffered.
371    #[must_use]
372    pub fn is_empty(&self) -> bool {
373        self.records
374            .lock()
375            .unwrap_or_else(PoisonError::into_inner)
376            .is_empty()
377    }
378}
379
380/// Sequenced stream event record.
381#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
382pub struct AgentStreamRecord {
383    /// Monotonic event sequence number within one run.
384    pub sequence: usize,
385    /// Source attribution for records merged from nested runs.
386    #[serde(default, skip_serializing_if = "Option::is_none")]
387    pub source: Option<AgentStreamSource>,
388    /// Typed event payload.
389    pub event: AgentStreamEvent,
390}
391
392impl starweaver_core::VersionedRecord for AgentStreamRecord {
393    const SCHEMA: &'static str = "starweaver.runtime.stream_record";
394    const ALLOW_BARE_V0: bool = true;
395}
396
397impl AgentStreamRecord {
398    /// Create a sequenced stream record.
399    #[must_use]
400    pub const fn new(sequence: usize, event: AgentStreamEvent) -> Self {
401        Self {
402            sequence,
403            source: None,
404            event,
405        }
406    }
407
408    /// Attach source attribution.
409    #[must_use]
410    pub fn with_source(mut self, source: AgentStreamSource) -> Self {
411        self.source = Some(source);
412        self
413    }
414
415    /// Replace the parent stream sequence after merging into another stream.
416    #[must_use]
417    pub const fn with_sequence(mut self, sequence: usize) -> Self {
418        self.sequence = sequence;
419        self
420    }
421
422    /// Project this raw runtime stream record into its JSON representation.
423    ///
424    /// # Errors
425    ///
426    /// Returns a serialization error if a nested event payload cannot be encoded.
427    pub fn to_raw_json(&self) -> serde_json::Result<serde_json::Value> {
428        serde_json::to_value(self)
429    }
430}
starweaver_stream/raw.rs

starweaver_stream/
raw.rs
