laser_wire/
agent_workflow.rs

1use serde::{Deserialize, Serialize};
2use std::collections::BTreeMap;
3
4/// Submit a task to an agent or workflow (`AGDX_AGENT_SUBMIT`). `agent_id` names
5/// the target, `run_id` lets the caller assign the run id (else the backend
6/// mints one), `params` is scalar control, and `input` is the opaque task body.
7#[derive(Clone, Debug, Serialize, Deserialize)]
8pub struct AgentSubmit {
9    pub v: u32,
10    pub agent_id: String,
11    #[serde(default, skip_serializing_if = "Option::is_none")]
12    pub run_id: Option<String>,
13    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
14    pub params: BTreeMap<String, String>,
15    #[serde(
16        default,
17        skip_serializing_if = "Option::is_none",
18        with = "crate::encoding::opt_bin_bytes"
19    )]
20    pub input: Option<Vec<u8>>,
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    pub budget: Option<RunBudget>,
23}
24
25/// A per-run resource ceiling carried on submit: caps across independent
26/// dimensions, each optional (absent is unbounded on that dimension). A run that
27/// crosses any cap is failed with a budget reason.
28#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
29pub struct RunBudget {
30    #[serde(default, skip_serializing_if = "Option::is_none")]
31    pub max_events: Option<u64>,
32    #[serde(default, skip_serializing_if = "Option::is_none")]
33    pub max_model_calls: Option<u64>,
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub max_tool_calls: Option<u64>,
36    #[serde(default, skip_serializing_if = "Option::is_none")]
37    pub max_patches: Option<u64>,
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub max_depth: Option<u32>,
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub max_wall_clock_micros: Option<u64>,
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    pub max_cost_usd: Option<f64>,
44}
45
46/// Cancel a run (`AGDX_AGENT_CANCEL`).
47#[derive(Clone, Debug, Serialize, Deserialize)]
48pub struct AgentCancel {
49    pub v: u32,
50    pub run_id: String,
51}
52
53/// Read a run's status (`AGDX_AGENT_STATUS`).
54#[derive(Clone, Debug, Serialize, Deserialize)]
55pub struct AgentStatusReq {
56    pub v: u32,
57    pub run_id: String,
58}
59
60/// List runs (`AGDX_AGENT_LIST`), filtered and paged. Every field but `v` is
61/// optional: an empty request lists the caller's runs from the newest. `limit`
62/// is clamped to [`MAX_PAGE_SIZE`](crate::limits::MAX_PAGE_SIZE) and `cursor`
63/// is the opaque continuation from the previous page, the kv scan pattern.
64#[derive(Clone, Debug, Default, Serialize, Deserialize)]
65pub struct AgentList {
66    pub v: u32,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub agent_id: Option<String>,
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub state: Option<AgentRunState>,
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub limit: Option<u32>,
73    #[serde(
74        default,
75        skip_serializing_if = "Option::is_none",
76        with = "crate::encoding::opt_bin_bytes"
77    )]
78    pub cursor: Option<Vec<u8>>,
79}
80
81/// A run's lifecycle state. A pinned snake-case vocabulary, additive. The
82/// strum derives keep the display, parse, and static-str spellings identical
83/// to the serde one by construction (`ForkKind` and `ContentType` set the
84/// same pattern).
85#[derive(
86    Clone,
87    Copy,
88    Debug,
89    Default,
90    PartialEq,
91    Eq,
92    Serialize,
93    Deserialize,
94    strum::Display,
95    strum::EnumString,
96    strum::IntoStaticStr,
97    strum::VariantArray,
98)]
99#[serde(rename_all = "snake_case")]
100#[strum(serialize_all = "snake_case")]
101#[non_exhaustive]
102pub enum AgentRunState {
103    #[default]
104    Submitted,
105    Running,
106    Completed,
107    Cancelled,
108    Failed,
109}
110
111impl AgentRunState {
112    /// The pinned snake-case word, the same spelling serde uses, for query
113    /// strings and display.
114    pub fn as_str(self) -> &'static str {
115        self.into()
116    }
117
118    /// Whether the run can never leave this state (the fold refuses to move a
119    /// terminal run backward).
120    pub fn is_terminal(self) -> bool {
121        matches!(
122            self,
123            AgentRunState::Completed | AgentRunState::Cancelled | AgentRunState::Failed
124        )
125    }
126}
127
128/// A run's metadata, returned by submit, status, and list. `updated_at_micros`
129/// is the time of the last state mark (equal to `created_at_micros` until one
130/// lands), and `detail` is the terminal summary (an error message on `failed`,
131/// absent when clean), so a console renders a run without a second read.
132/// `cancel_requested` is the recorded cancel intent, not a state: the engine
133/// observes it at its next step boundary and routes it into its own
134/// cancellation path, and the state moves only when the engine reports it.
135#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
136pub struct AgentRunInfo {
137    pub run_id: String,
138    pub agent_id: String,
139    pub user_id: u32,
140    pub state: AgentRunState,
141    pub created_at_micros: u64,
142    pub updated_at_micros: u64,
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    pub detail: Option<String>,
145    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
146    pub cancel_requested: bool,
147}
148
149/// One page of runs: the rows plus the opaque cursor for the next page, absent
150/// on the last one.
151#[derive(Clone, Debug, Default, Serialize, Deserialize)]
152pub struct RunPage {
153    pub runs: Vec<AgentRunInfo>,
154    #[serde(
155        default,
156        skip_serializing_if = "Option::is_none",
157        with = "crate::encoding::opt_bin_bytes"
158    )]
159    pub cursor: Option<Vec<u8>>,
160}
161
162/// The result of an agent or workflow control op: `Ok` with the outcome, or
163/// `Err` with a failure.
164#[derive(Clone, Debug, Serialize, Deserialize)]
165#[non_exhaustive]
166pub enum AgentReply {
167    Ok(AgentOutcome),
168    Err(AgentError),
169}
170
171/// The successful outcome of an agent or workflow command, shaped per op.
172#[derive(Clone, Debug, Serialize, Deserialize)]
173#[non_exhaustive]
174pub enum AgentOutcome {
175    Submitted(AgentRunInfo),
176    Cancelled(AgentRunInfo),
177    Status(AgentRunInfo),
178    List(RunPage),
179}
180
181/// Why an agent or workflow control op failed.
182#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
183#[non_exhaustive]
184pub enum AgentError {
185    #[error("agent ops not supported: {0}")]
186    Unsupported(String),
187    #[error("run not found: {0}")]
188    NotFound(String),
189    #[error("invalid agent request: {0}")]
190    Invalid(String),
191    #[error("agent backend error: {0}")]
192    Backend(String),
193    #[error("unsupported agent op version (expected {expected}, got {got})")]
194    Version { expected: u32, got: u32 },
195    /// This plane does not own the mutation partition for the run.
196    #[error("not the partition leader for this run")]
197    NotLeader,
198}
199
200#[cfg(all(test, feature = "cbor"))]
201mod tests {
202    use super::*;
203    use crate::codes::AGENT_WORKFLOW_OP_VERSION;
204    use crate::framing::{decode_named, encode_named};
205
206    #[test]
207    fn given_a_submit_when_round_tripped_then_should_decode_unchanged() {
208        let submit = AgentSubmit {
209            v: AGENT_WORKFLOW_OP_VERSION,
210            agent_id: "diagnoser".to_owned(),
211            run_id: Some("run-7".to_owned()),
212            params: BTreeMap::from([("priority".to_owned(), "high".to_owned())]),
213            input: Some(br#"{"incident":"INC-7"}"#.to_vec()),
214            budget: None,
215        };
216        let bytes = encode_named(&submit).expect("encodes");
217        let back: AgentSubmit = decode_named(&bytes).expect("decodes");
218        assert_eq!(back.agent_id, submit.agent_id);
219        assert_eq!(back.run_id, submit.run_id);
220        assert_eq!(back.input, submit.input);
221    }
222
223    #[test]
224    fn given_a_reply_when_round_tripped_then_should_preserve_the_variant() {
225        let reply = AgentReply::Ok(AgentOutcome::Status(AgentRunInfo {
226            run_id: "run-7".to_owned(),
227            agent_id: "diagnoser".to_owned(),
228            user_id: 42,
229            state: AgentRunState::Running,
230            created_at_micros: 1_717_171_717_000_000,
231            updated_at_micros: 1_717_171_718_000_000,
232            detail: None,
233            cancel_requested: false,
234        }));
235        let bytes = encode_named(&reply).expect("encodes");
236        let back: AgentReply = decode_named(&bytes).expect("decodes");
237        assert!(matches!(
238            back,
239            AgentReply::Ok(AgentOutcome::Status(info)) if info.run_id == "run-7"
240        ));
241    }
242
243    #[test]
244    fn given_a_filtered_list_when_round_tripped_then_should_keep_filters_and_cursor() {
245        let list = AgentList {
246            v: AGENT_WORKFLOW_OP_VERSION,
247            agent_id: Some("diagnoser".to_owned()),
248            state: Some(AgentRunState::Running),
249            limit: Some(25),
250            cursor: Some(vec![0x01, 0x02]),
251        };
252        let bytes = encode_named(&list).expect("encodes");
253        let back: AgentList = decode_named(&bytes).expect("decodes");
254        assert_eq!(back.agent_id, list.agent_id);
255        assert_eq!(back.state, list.state);
256        assert_eq!(back.limit, list.limit);
257        assert_eq!(back.cursor, list.cursor);
258    }
259
260    #[test]
261    fn given_run_state_words_when_parsed_then_should_round_trip_through_display() {
262        for state in [
263            AgentRunState::Submitted,
264            AgentRunState::Running,
265            AgentRunState::Completed,
266            AgentRunState::Cancelled,
267            AgentRunState::Failed,
268        ] {
269            let parsed: AgentRunState = state.as_str().parse().expect("pinned word parses");
270            assert_eq!(parsed, state);
271        }
272        assert!("paused".parse::<AgentRunState>().is_err());
273    }
274
275    #[test]
276    fn given_an_empty_list_request_when_encoded_then_should_skip_absent_fields() {
277        let bare = AgentList {
278            v: AGENT_WORKFLOW_OP_VERSION,
279            ..AgentList::default()
280        };
281        let bytes = encode_named(&bare).expect("encodes");
282        let back: AgentList = decode_named(&bytes).expect("decodes");
283        assert!(back.agent_id.is_none());
284        assert!(back.state.is_none());
285        assert!(back.limit.is_none());
286        assert!(back.cursor.is_none());
287    }
288}
laser_wire/agent_workflow.rs

laser_wire/
agent_workflow.rs
