agentd/a2a/
tasks.rs

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

agentd/a2a/
tasks.rs
