laser_wire/
dispatch.rs

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

laser_wire/
dispatch.rs
