agentd/a2a/
tasks.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//! **A2A tasks**: a durable unit of work a principal started — a root-turn
3//! answer, a workflow run, or a subagent — projected as an A2A `Task` (spec
4//! shape, `TASK_STATE_*`). Tasks are persisted, so `GetTask` answers across a
5//! restart; they stream status/artifact frames from run and turn events; and
6//! cancelling one cancels the work it links to, which in turn cancels that
7//! work's own children, so no orphan keeps running behind a cancelled task.
8
9use crate::state::now_ms;
10use serde::{Deserialize, Serialize};
11use serde_json::{Value, json};
12
13/// The A2A task state (mirrors `mcp::a2a::TaskState`, kept here so the runtime
14/// does not depend on the `a2a` feature-gated module).
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
16#[serde(rename_all = "snake_case")]
17pub enum State {
18    #[default]
19    Submitted,
20    Working,
21    InputRequired,
22    Completed,
23    Failed,
24    Canceled,
25    Rejected,
26}
27
28impl State {
29    pub fn wire(self) -> &'static str {
30        match self {
31            State::Submitted => "TASK_STATE_SUBMITTED",
32            State::Working => "TASK_STATE_WORKING",
33            State::InputRequired => "TASK_STATE_INPUT_REQUIRED",
34            State::Completed => "TASK_STATE_COMPLETED",
35            State::Failed => "TASK_STATE_FAILED",
36            State::Canceled => "TASK_STATE_CANCELED",
37            State::Rejected => "TASK_STATE_REJECTED",
38        }
39    }
40    pub fn is_terminal(self) -> bool {
41        matches!(
42            self,
43            State::Completed | State::Failed | State::Canceled | State::Rejected
44        )
45    }
46    /// The task state a run status maps to.
47    pub fn from_run(status: &str) -> State {
48        match status {
49            "completed" => State::Completed,
50            "refused" => State::Rejected,
51            "cancelled" => State::Canceled,
52            "running" | "suspended" | "paused" | "pending" => State::Working,
53            _ => State::Failed,
54        }
55    }
56}
57
58/// What a task is attached to.
59#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
60#[serde(rename_all = "snake_case")]
61pub enum Link {
62    Run {
63        id: String,
64    },
65    Subagent {
66        handle: String,
67    },
68    /// A short-lived turn answer for a conversation.
69    Turn {
70        ctx: String,
71    },
72}
73
74/// A webhook a caller registered for this task's updates (A2A push
75/// notifications). `token` is echoed back in `X-A2A-Notification-Token` so the
76/// receiver can tell a real delivery from a stray POST; `bearer` is a
77/// credential agentd presents *to* the receiver.
78#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
79pub struct PushTarget {
80    pub id: String,
81    pub url: String,
82    #[serde(default, skip_serializing_if = "String::is_empty")]
83    pub token: String,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub bearer: Option<String>,
86}
87
88/// The durable task record.
89#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
90pub struct Task {
91    pub id: String,
92    pub context_id: String,
93    #[serde(default)]
94    pub state: State,
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub principal: Option<String>,
97    pub link: Link,
98    /// The status message (for `input-required` and terminal explanations).
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub message: Option<String>,
101    /// The shape a gate's answer must take (`human.schema` / `ask_human`).
102    ///
103    /// Carried on the task because the QUESTION alone does not tell a client
104    /// how to ask it. With the schema, "pick one of these three" renders as
105    /// three options instead of a text box the person has to guess the wording
106    /// for — and the answer is already the right shape when it comes back.
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub ask_schema: Option<Value>,
109    /// Artifact ids delivered on this task.
110    #[serde(default, skip_serializing_if = "Vec::is_empty")]
111    pub artifacts: Vec<String>,
112    /// The terminal result (a distillate / output).
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub result: Option<Value>,
115    #[serde(default)]
116    pub created: u64,
117    #[serde(default)]
118    pub updated: u64,
119    /// The transition history (state, ts).
120    #[serde(default, skip_serializing_if = "Vec::is_empty")]
121    pub history: Vec<Value>,
122    /// Where to POST this task's updates, for a caller that would rather be
123    /// told than hold a stream open. Durable with the task, so a restart keeps
124    /// the promise the caller was given.
125    #[serde(default, skip_serializing_if = "Vec::is_empty")]
126    pub push: Vec<PushTarget>,
127    #[serde(skip)]
128    pub dirty: bool,
129}
130
131impl Task {
132    pub fn new(id: &str, context_id: &str, principal: Option<&str>, link: Link) -> Task {
133        let now = now_ms();
134        Task {
135            ask_schema: None,
136            id: id.to_string(),
137            context_id: context_id.to_string(),
138            state: State::Submitted,
139            principal: principal.map(str::to_string),
140            link,
141            message: None,
142            artifacts: Vec::new(),
143            result: None,
144            created: now,
145            updated: now,
146            history: vec![json!({"state": State::Submitted.wire(), "ts": now})],
147            push: Vec::new(),
148            dirty: true,
149        }
150    }
151
152    pub fn transition(&mut self, state: State, message: Option<String>) {
153        if self.state == state && self.message == message {
154            return;
155        }
156        self.state = state;
157        if message.is_some() {
158            self.message = message;
159        }
160        self.updated = now_ms();
161        self.history
162            .push(json!({"state": state.wire(), "ts": self.updated}));
163        if self.history.len() > 64 {
164            self.history.remove(0);
165        }
166        self.dirty = true;
167    }
168
169    pub fn add_artifact(&mut self, id: &str) {
170        if !self.artifacts.iter().any(|a| a == id) {
171            self.artifacts.push(id.to_string());
172            self.updated = now_ms();
173            self.dirty = true;
174        }
175    }
176
177    pub fn set_result(&mut self, v: Value) {
178        self.result = Some(v);
179        self.updated = now_ms();
180        self.dirty = true;
181    }
182
183    /// The A2A `Task` object — what `GetTask`, `CancelTask` and a `SendMessage`
184    /// reply carry. Built from the specification's own types, so the wire
185    /// spellings are not ours to get wrong; see [`crate::a2a::wire`].
186    #[cfg(feature = "a2a")]
187    pub fn to_a2a(&self) -> Value {
188        serde_json::to_value(crate::a2a::wire::task(self)).unwrap_or(Value::Null)
189    }
190
191    /// The light projection `ListTasks` returns: the same `Task` minus the
192    /// artifacts a listing does not resolve.
193    #[cfg(feature = "a2a")]
194    pub fn summary(&self) -> Value {
195        serde_json::to_value(crate::a2a::wire::task_summary(self)).unwrap_or(Value::Null)
196    }
197}
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    /// The durable record's own behaviour. What it looks like on the wire is
203    /// `a2a::wire`'s job, and is tested there against the spec's types.
204    #[test]
205    fn a_task_records_its_lifecycle_and_survives_a_round_trip() {
206        let mut t = Task::new(
207            "task-1",
208            "ctx-1",
209            Some("user:a"),
210            Link::Run { id: "r1".into() },
211        );
212        assert_eq!(t.state, State::Submitted);
213        t.transition(State::Working, None);
214        t.transition(State::Working, None); // idempotent
215        assert_eq!(t.history.len(), 2);
216        t.add_artifact("art-9");
217        t.add_artifact("art-9"); // idempotent
218        assert_eq!(t.artifacts.len(), 1);
219        t.set_result(json!({"answer": 42}));
220        t.transition(State::Completed, Some("done".into()));
221        assert!(t.state.is_terminal());
222        assert_eq!(t.message.as_deref(), Some("done"));
223        assert_eq!(State::from_run("refused"), State::Rejected);
224        assert_eq!(State::from_run("running"), State::Working);
225
226        let v = serde_json::to_value(&t).unwrap();
227        let back: Task = serde_json::from_value(v).unwrap();
228        assert_eq!(back.state, t.state);
229        assert_eq!(back.history.len(), t.history.len());
230        assert!(!back.dirty);
231    }
232}
agentd/a2a/tasks.rs

agentd/a2a/
tasks.rs
