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    /// The request could not be served right now and the identical request may
194    /// succeed later: the store was momentarily out of reach, a connection was
195    /// refused or dropped, or a concurrency conflict was aborted. Retry with
196    /// backoff rather than treating it as a fault.
197    #[error("temporarily unavailable: {0}")]
198    Unavailable(String),
199    #[error("unsupported agent op version (expected {expected}, got {got})")]
200    Version { expected: u32, got: u32 },
201    /// This plane does not own the mutation partition for the run.
202    #[error("not the partition leader for this run")]
203    NotLeader,
204}
205
206#[cfg(all(test, feature = "cbor"))]
207mod tests {
208    use super::*;
209    use crate::codes::AGENT_WORKFLOW_OP_VERSION;
210    use crate::framing::{decode_named, encode_named};
211
212    #[test]
213    fn given_a_submit_when_round_tripped_then_should_decode_unchanged() {
214        let submit = AgentSubmit {
215            v: AGENT_WORKFLOW_OP_VERSION,
216            agent_id: "diagnoser".to_owned(),
217            run_id: Some("run-7".to_owned()),
218            params: BTreeMap::from([("priority".to_owned(), "high".to_owned())]),
219            input: Some(br#"{"incident":"INC-7"}"#.to_vec()),
220            budget: None,
221        };
222        let bytes = encode_named(&submit).expect("encodes");
223        let back: AgentSubmit = decode_named(&bytes).expect("decodes");
224        assert_eq!(back.agent_id, submit.agent_id);
225        assert_eq!(back.run_id, submit.run_id);
226        assert_eq!(back.input, submit.input);
227    }
228
229    #[test]
230    fn given_a_reply_when_round_tripped_then_should_preserve_the_variant() {
231        let reply = AgentReply::Ok(AgentOutcome::Status(AgentRunInfo {
232            run_id: "run-7".to_owned(),
233            agent_id: "diagnoser".to_owned(),
234            user_id: 42,
235            state: AgentRunState::Running,
236            created_at_micros: 1_717_171_717_000_000,
237            updated_at_micros: 1_717_171_718_000_000,
238            detail: None,
239            cancel_requested: false,
240        }));
241        let bytes = encode_named(&reply).expect("encodes");
242        let back: AgentReply = decode_named(&bytes).expect("decodes");
243        assert!(matches!(
244            back,
245            AgentReply::Ok(AgentOutcome::Status(info)) if info.run_id == "run-7"
246        ));
247    }
248
249    #[test]
250    fn given_a_filtered_list_when_round_tripped_then_should_keep_filters_and_cursor() {
251        let list = AgentList {
252            v: AGENT_WORKFLOW_OP_VERSION,
253            agent_id: Some("diagnoser".to_owned()),
254            state: Some(AgentRunState::Running),
255            limit: Some(25),
256            cursor: Some(vec![0x01, 0x02]),
257        };
258        let bytes = encode_named(&list).expect("encodes");
259        let back: AgentList = decode_named(&bytes).expect("decodes");
260        assert_eq!(back.agent_id, list.agent_id);
261        assert_eq!(back.state, list.state);
262        assert_eq!(back.limit, list.limit);
263        assert_eq!(back.cursor, list.cursor);
264    }
265
266    #[test]
267    fn given_run_state_words_when_parsed_then_should_round_trip_through_display() {
268        for state in [
269            AgentRunState::Submitted,
270            AgentRunState::Running,
271            AgentRunState::Completed,
272            AgentRunState::Cancelled,
273            AgentRunState::Failed,
274        ] {
275            let parsed: AgentRunState = state.as_str().parse().expect("pinned word parses");
276            assert_eq!(parsed, state);
277        }
278        assert!("paused".parse::<AgentRunState>().is_err());
279    }
280
281    #[test]
282    fn given_an_empty_list_request_when_encoded_then_should_skip_absent_fields() {
283        let bare = AgentList {
284            v: AGENT_WORKFLOW_OP_VERSION,
285            ..AgentList::default()
286        };
287        let bytes = encode_named(&bare).expect("encodes");
288        let back: AgentList = decode_named(&bytes).expect("decodes");
289        assert!(back.agent_id.is_none());
290        assert!(back.state.is_none());
291        assert!(back.limit.is_none());
292        assert!(back.cursor.is_none());
293    }
294}
laser_wire/agent_workflow.rs

laser_wire/
agent_workflow.rs
