laser_wire/
dispatch.rs

1use crate::agent::{
2    AgentEnvelope, AgentId, AgentKind, METADATA_ROLE, OPERATION_CHAT, OPERATION_PROGRESS,
3    OPERATION_REASONING, OPERATION_SESSION, OPERATION_SESSION_PARKED, OPERATION_SESSION_UNPARKED,
4    OPERATION_STATE_DELTA, OPERATION_STATE_SNAPSHOT, OPERATION_TASK, TaskState,
5};
6use crate::filter::{ConsumerFilter, FilterExpr};
7use crate::headers::TARGET_AGENT_ID;
8use crate::memory::MemoryRecord;
9use crate::query::{CmpOp, Value};
10use crate::schema::TypedValue;
11use crate::topics::{AGENT_CONTROL, AGENT_HEARTBEATS};
12
13pub use crate::headers::BROADCAST;
14
15/// The `command` operation that asks a session to pause.
16pub const OPERATION_SESSION_PAUSE: &str = "session_pause";
17/// The `command` operation that asks a paused session to resume.
18pub const OPERATION_SESSION_RESUME: &str = "session_resume";
19/// The `command` operation that asks a session to cancel.
20pub const OPERATION_SESSION_CANCEL: &str = "session_cancel";
21/// The `command` operation that ends a session as canceled without its agent.
22pub const OPERATION_FORCE_CANCEL: &str = "force_cancel";
23/// The OTel operation for a text completion model call.
24pub const OPERATION_TEXT_COMPLETION: &str = "text_completion";
25/// The OTel operation for a content generation model call.
26pub const OPERATION_GENERATE_CONTENT: &str = "generate_content";
27/// The OTel operation for a tool call.
28pub const OPERATION_EXECUTE_TOOL: &str = "execute_tool";
29/// The OTel operation for a call to another agent.
30pub const OPERATION_INVOKE_AGENT: &str = "invoke_agent";
31/// The `event` operation that records an assembled model context.
32pub const OPERATION_CONTEXT_ASSEMBLED: &str = "context_assembled";
33/// The `event` operation that records a context compaction.
34pub const OPERATION_CONTEXT_COMPACTED: &str = "context_compacted";
35/// The `event` operation that records a context retrieval.
36pub const OPERATION_CONTEXT_RETRIEVED: &str = "context_retrieved";
37/// The `event` operation that records a policy decision.
38pub const OPERATION_POLICY_DECISION: &str = "policy_decision";
39
40/// The session control operations. They are honored only on `agent.control`.
41pub const CONTROL_OPERATIONS: [&str; 4] = [
42    OPERATION_SESSION_PAUSE,
43    OPERATION_SESSION_RESUME,
44    OPERATION_SESSION_CANCEL,
45    OPERATION_FORCE_CANCEL,
46];
47
48/// One record a session timeline can show.
49#[derive(Clone, Copy, Debug)]
50pub enum SessionRecord<'a> {
51    Envelope(&'a AgentEnvelope),
52    Memory(&'a MemoryRecord),
53    DeadLetter,
54    Journal,
55    KvMutation,
56    GraphMutation,
57    /// A record whose body does not decode, with its kind when the header
58    /// names one.
59    Undecodable(Option<AgentKind>),
60}
61
62/// How a session timeline shows one record. The value is derived from the
63/// record and never stored as a separate field.
64#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, strum::IntoStaticStr, strum::EnumString)]
65pub enum DisplayType {
66    #[strum(serialize = "session.submitted")]
67    SessionSubmitted,
68    #[strum(serialize = "session.started")]
69    SessionStarted,
70    #[strum(serialize = "session.paused")]
71    SessionPaused,
72    #[strum(serialize = "session.resumed")]
73    SessionResumed,
74    #[strum(serialize = "session.completed")]
75    SessionCompleted,
76    #[strum(serialize = "session.failed")]
77    SessionFailed,
78    #[strum(serialize = "session.canceled")]
79    SessionCanceled,
80    #[strum(serialize = "session.heartbeat")]
81    SessionHeartbeat,
82    /// A work record a paused session's agent held.
83    #[strum(serialize = "session.parked")]
84    SessionParked,
85    /// A held work record handled after the session resumed.
86    #[strum(serialize = "session.unparked")]
87    SessionUnparked,
88    #[strum(serialize = "session.control")]
89    SessionControl,
90    #[strum(serialize = "user.message")]
91    UserMessage,
92    #[strum(serialize = "model.request")]
93    ModelRequest,
94    #[strum(serialize = "model.response")]
95    ModelResponse,
96    #[strum(serialize = "model.stream")]
97    ModelStream,
98    #[strum(serialize = "tool.call")]
99    ToolCall,
100    #[strum(serialize = "tool.result")]
101    ToolResult,
102    #[strum(serialize = "agent.handoff")]
103    AgentHandoff,
104    /// An agent record that matches no more specific row.
105    #[strum(serialize = "agent.message")]
106    AgentMessage,
107    #[strum(serialize = "state.updated")]
108    StateUpdated,
109    #[strum(serialize = "context.assembled")]
110    ContextAssembled,
111    #[strum(serialize = "context.compacted")]
112    ContextCompacted,
113    #[strum(serialize = "context.retrieved")]
114    ContextRetrieved,
115    #[strum(serialize = "memory.created")]
116    MemoryCreated,
117    #[strum(serialize = "memory.forgotten")]
118    MemoryForgotten,
119    #[strum(serialize = "memory.feedback")]
120    MemoryFeedback,
121    #[strum(serialize = "task.status")]
122    TaskStatus,
123    #[strum(serialize = "workflow.step")]
124    WorkflowStep,
125    #[strum(serialize = "policy.decision")]
126    PolicyDecision,
127    #[strum(serialize = "error")]
128    Error,
129    #[strum(serialize = "dead_letter")]
130    DeadLetter,
131    #[strum(serialize = "undecodable")]
132    Undecodable,
133    /// A record whose header and body name different identities.
134    #[strum(serialize = "invalid")]
135    Invalid,
136    #[strum(serialize = "kv.set")]
137    KvSet,
138    #[strum(serialize = "graph.upsert")]
139    GraphUpsert,
140    #[strum(serialize = "unauthorized_control")]
141    UnauthorizedControl,
142}
143
144impl DisplayType {
145    /// The pinned display word.
146    pub fn as_str(self) -> &'static str {
147        self.into()
148    }
149}
150
151impl std::fmt::Display for DisplayType {
152    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        formatter.write_str(self.as_str())
154    }
155}
156
157/// How a session timeline shows `record`, read from the topic named `topic`.
158pub fn display_type(record: SessionRecord<'_>, topic: &str) -> DisplayType {
159    let envelope = match record {
160        SessionRecord::Envelope(envelope) => envelope,
161        SessionRecord::Memory(MemoryRecord::Item { .. }) => return DisplayType::MemoryCreated,
162        SessionRecord::Memory(MemoryRecord::Forget { .. }) => return DisplayType::MemoryForgotten,
163        SessionRecord::Memory(MemoryRecord::Feedback { .. }) => {
164            return DisplayType::MemoryFeedback;
165        }
166        SessionRecord::DeadLetter => return DisplayType::DeadLetter,
167        SessionRecord::Journal => return DisplayType::WorkflowStep,
168        SessionRecord::KvMutation => return DisplayType::KvSet,
169        SessionRecord::GraphMutation => return DisplayType::GraphUpsert,
170        SessionRecord::Undecodable(_) => return DisplayType::Undecodable,
171    };
172    let operation = envelope.operation.as_deref().unwrap_or_default();
173    match envelope.kind {
174        AgentKind::Status => status_display(envelope, operation, topic),
175        AgentKind::Command if is_control(operation) => {
176            if topic == AGENT_CONTROL {
177                DisplayType::SessionControl
178            } else {
179                DisplayType::UnauthorizedControl
180            }
181        }
182        AgentKind::Command if is_user(envelope) => DisplayType::UserMessage,
183        AgentKind::Command if is_model(operation) => DisplayType::ModelRequest,
184        AgentKind::Command if operation == OPERATION_EXECUTE_TOOL => DisplayType::ToolCall,
185        AgentKind::Command if operation == OPERATION_INVOKE_AGENT => DisplayType::AgentHandoff,
186        AgentKind::Event if is_user(envelope) => DisplayType::UserMessage,
187        AgentKind::Event => match operation {
188            OPERATION_STATE_DELTA | OPERATION_STATE_SNAPSHOT => DisplayType::StateUpdated,
189            OPERATION_CONTEXT_ASSEMBLED => DisplayType::ContextAssembled,
190            OPERATION_CONTEXT_COMPACTED => DisplayType::ContextCompacted,
191            OPERATION_CONTEXT_RETRIEVED => DisplayType::ContextRetrieved,
192            OPERATION_POLICY_DECISION => DisplayType::PolicyDecision,
193            OPERATION_SESSION_PARKED => DisplayType::SessionParked,
194            OPERATION_SESSION_UNPARKED => DisplayType::SessionUnparked,
195            _ => DisplayType::AgentMessage,
196        },
197        AgentKind::Response if is_model(operation) => DisplayType::ModelResponse,
198        AgentKind::Response | AgentKind::Error if operation == OPERATION_EXECUTE_TOOL => {
199            DisplayType::ToolResult
200        }
201        AgentKind::Error => DisplayType::Error,
202        AgentKind::Chunk if operation == OPERATION_CHAT || operation == OPERATION_REASONING => {
203            DisplayType::ModelStream
204        }
205        _ => DisplayType::AgentMessage,
206    }
207}
208
209fn status_display(envelope: &AgentEnvelope, operation: &str, topic: &str) -> DisplayType {
210    if operation == OPERATION_PROGRESS && topic == AGENT_HEARTBEATS {
211        return DisplayType::SessionHeartbeat;
212    }
213    if operation != OPERATION_SESSION {
214        return if operation == OPERATION_TASK {
215            DisplayType::TaskStatus
216        } else {
217            DisplayType::AgentMessage
218        };
219    }
220    match envelope.task_state {
221        Some(TaskState::Submitted) => DisplayType::SessionSubmitted,
222        Some(TaskState::Working) if is_session_start(&envelope.body) => DisplayType::SessionStarted,
223        Some(TaskState::Working) => DisplayType::SessionResumed,
224        Some(TaskState::Paused) => DisplayType::SessionPaused,
225        Some(TaskState::Completed) => DisplayType::SessionCompleted,
226        Some(TaskState::Failed | TaskState::Rejected) => DisplayType::SessionFailed,
227        Some(TaskState::Canceled) => DisplayType::SessionCanceled,
228        _ => DisplayType::TaskStatus,
229    }
230}
231
232// A start body names the agent and SDK. A transition body names neither.
233fn is_session_start(body: &[u8]) -> bool {
234    crate::framing::decode_named::<ciborium::value::Value>(body).is_ok_and(|value| {
235        matches!(value, ciborium::value::Value::Map(entries)
236        if entries.iter().any(|(key, _)| {
237            matches!(key, ciborium::value::Value::Text(name) if name == "agent" || name == "sdk")
238        }))
239    })
240}
241
242fn is_control(operation: &str) -> bool {
243    CONTROL_OPERATIONS.contains(&operation)
244}
245
246fn is_model(operation: &str) -> bool {
247    matches!(
248        operation,
249        OPERATION_CHAT | OPERATION_TEXT_COMPLETION | OPERATION_GENERATE_CONTENT
250    )
251}
252
253fn is_user(envelope: &AgentEnvelope) -> bool {
254    envelope
255        .metadata
256        .as_ref()
257        .and_then(|metadata| metadata.get(METADATA_ROLE))
258        .is_some_and(|role| matches!(role, Value::Str(role) if role == "user"))
259}
260
261/// What a reliable consumer does with one record it read.
262#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, strum::IntoStaticStr, strum::EnumString)]
263#[strum(serialize_all = "snake_case")]
264pub enum Dispatch {
265    /// A command addressed to this agent for an operation its handler serves.
266    Work,
267    /// A record that informs the timeline and is never handled: events, model
268    /// and tool records for operations the handler does not serve, and control
269    /// operations found outside `agent.control`.
270    Observational,
271    /// A response, error, or chunk. Reply waiters read it, handlers never do.
272    Reply,
273    /// A status record.
274    Lifecycle,
275    /// A control operation on `agent.control`, for the control follower.
276    Control,
277    /// A record addressed to another agent.
278    Foreign,
279}
280
281impl Dispatch {
282    /// The pinned snake-case word.
283    pub fn as_str(self) -> &'static str {
284        self.into()
285    }
286}
287
288/// The command operations a handler serves.
289#[derive(Clone, Copy, Debug)]
290pub enum HandledOperations<'a> {
291    /// Every command operation.
292    Any,
293    /// Only the listed operations.
294    Only(&'a [&'a str]),
295}
296
297impl HandledOperations<'_> {
298    fn handles(self, operation: &str) -> bool {
299        match self {
300            Self::Any => true,
301            Self::Only(operations) => operations.contains(&operation),
302        }
303    }
304}
305
306/// What the agent `me` does with `envelope`, read from the topic named
307/// `topic`. The author is never a discriminator, so an agent may send work to
308/// itself.
309pub fn classify(
310    envelope: &AgentEnvelope,
311    topic: &str,
312    me: &AgentId,
313    handled: HandledOperations<'_>,
314) -> Dispatch {
315    let operation = envelope.operation.as_deref().unwrap_or_default();
316    match envelope.kind {
317        AgentKind::Status => Dispatch::Lifecycle,
318        AgentKind::Response | AgentKind::Error | AgentKind::Chunk => Dispatch::Reply,
319        AgentKind::Event => Dispatch::Observational,
320        AgentKind::Command => {
321            if envelope.target.as_ref().is_some_and(|target| target != me) {
322                return Dispatch::Foreign;
323            }
324            match (is_control(operation), topic == AGENT_CONTROL) {
325                (true, true) => Dispatch::Control,
326                (true, false) => Dispatch::Observational,
327                (false, true) => Dispatch::Foreign,
328                (false, false) if handled.handles(operation) => Dispatch::Work,
329                (false, false) => Dispatch::Observational,
330            }
331        }
332    }
333}
334
335/// The headers-only group filter that selects records addressed to `me` or to
336/// every agent. Its digest differs per agent identity.
337pub fn addressee_filter(me: &AgentId) -> ConsumerFilter {
338    ConsumerFilter::headers_only(FilterExpr::header(
339        TARGET_AGENT_ID,
340        CmpOp::In,
341        TypedValue::List(vec![
342            TypedValue::String(me.as_str().to_owned()),
343            TypedValue::String(BROADCAST.to_owned()),
344        ]),
345    ))
346}
347
348/// The headers-only group filter that selects broadcast records only.
349pub fn broadcast_filter() -> ConsumerFilter {
350    ConsumerFilter::headers_only(FilterExpr::header(
351        TARGET_AGENT_ID,
352        CmpOp::Eq,
353        TypedValue::String(BROADCAST.to_owned()),
354    ))
355}
356
357/// What the agent `me` does with a generic record, one without an envelope,
358/// read from the topic named `topic`. Such a record has no kind, so a record
359/// that names both its causal parent and a correlation is a reply, and any
360/// other record is work unless it is addressed to another agent.
361pub fn classify_generic(
362    addressee: Option<&str>,
363    has_cause: bool,
364    has_correlation: bool,
365    topic: &str,
366    me: &AgentId,
367) -> Dispatch {
368    if addressee.is_some_and(|to| to != me.as_str() && to != BROADCAST) || topic == AGENT_CONTROL {
369        return Dispatch::Foreign;
370    }
371    if has_cause && has_correlation {
372        Dispatch::Reply
373    } else {
374        Dispatch::Work
375    }
376}
laser_wire/dispatch.rs

laser_wire/
dispatch.rs
