lean_ctx/core/a2a/
task_control.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Receiver-side authorization for signed task control operations.
4//!
5//! This is the storage boundary, not a wire response: the transport must sign
6//! the returned status and bind it to the request before publishing it.
7
8use std::path::Path;
9
10use super::task::{
11    TASK_ACTION_CANCEL, TASK_ACTION_GET, Task, TaskAuthorityConfigV1, TaskAuthorityError,
12    TaskAuthorityExpectationV1, TaskControlDescriptorV1, TaskState, TaskStatusV1, TaskStore,
13};
14
15#[derive(Debug)]
16pub enum TaskControlError {
17    Authority(TaskAuthorityError),
18    Storage(std::io::Error),
19    /// Deliberately identical for an absent task and any target-binding mismatch.
20    NotFound,
21    /// An owned task cannot make the requested terminal transition.
22    InvalidTransition,
23}
24
25impl From<std::io::Error> for TaskControlError {
26    fn from(error: std::io::Error) -> Self {
27        Self::Storage(error)
28    }
29}
30
31/// Authenticate the operation before touching storage; then enforce target
32/// ownership under the same lock as the status read. Never return task payloads
33/// or artifact references from this status-only boundary.
34pub fn get_status(
35    path: &Path,
36    request: &TaskControlDescriptorV1,
37    policy: &TaskAuthorityConfigV1,
38    expected: &TaskAuthorityExpectationV1<'_>,
39) -> Result<TaskStatusV1, TaskControlError> {
40    if request.action != TASK_ACTION_GET {
41        return Err(TaskControlError::Authority(
42            TaskAuthorityError::UnsupportedAction,
43        ));
44    }
45    request
46        .verify_authority(policy, expected)
47        .map_err(TaskControlError::Authority)?;
48    TaskStore::read_locked(path, |store| {
49        store
50            .get_task(&request.task_id)
51            .filter(|task| matches_target(task, request))
52            .map(Task::status_v1)
53            .ok_or(TaskControlError::NotFound)
54    })
55    .map_err(TaskControlError::Storage)?
56}
57
58/// Persist an authenticated owner's cancellation before acknowledging it.
59/// Repeated cancellation returns the original terminal status and adds no
60/// history entry; a completed or failed task cannot be rewritten as canceled.
61/// Like get_status, this is a storage operation, not a signed wire response.
62pub fn cancel(
63    path: &Path,
64    request: &TaskControlDescriptorV1,
65    policy: &TaskAuthorityConfigV1,
66    expected: &TaskAuthorityExpectationV1<'_>,
67) -> Result<TaskStatusV1, TaskControlError> {
68    if request.action != TASK_ACTION_CANCEL {
69        return Err(TaskControlError::Authority(
70            TaskAuthorityError::UnsupportedAction,
71        ));
72    }
73    request
74        .verify_authority(policy, expected)
75        .map_err(TaskControlError::Authority)?;
76    TaskStore::mutate_locked(path, |store| {
77        let task = store
78            .get_task_mut(&request.task_id)
79            .filter(|task| matches_target(task, request))
80            .ok_or(TaskControlError::NotFound)?;
81        if task.state != TaskState::Canceled {
82            task.transition(TaskState::Canceled, request.reason.as_deref())
83                .map_err(|_| TaskControlError::InvalidTransition)?;
84        }
85        Ok(task.status_v1())
86    })
87}
88
89fn matches_target(task: &Task, request: &TaskControlDescriptorV1) -> bool {
90    // A rotated authorized key may represent the same durable owner. Check
91    // signed-task provenance, not equality with the original signing key.
92    task.authority_key_id.is_some()
93        && task.id == request.task_id
94        && task.from_agent == request.sender
95        && task.to_agent == request.recipient
96        && task.tenant_id.as_deref() == Some(request.tenant_id.as_str())
97        && task.project_id.as_deref() == Some(request.project_id.as_str())
98        && task.descriptor_digest.as_deref() == Some(request.task_descriptor_digest.as_str())
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use chrono::{Duration, Utc};
105
106    fn fixture() -> (Task, TaskControlDescriptorV1) {
107        let mut task = Task::new("owner", "receiver", "private payload");
108        task.tenant_id = Some("tenant".into());
109        task.project_id = Some("project".into());
110        task.authority_key_id = Some("original-key".into());
111        task.descriptor_digest = Some(format!("sha256:{}", "a".repeat(64)));
112        let now = Utc::now();
113        let request = TaskControlDescriptorV1::new(
114            "owner",
115            "receiver",
116            "tenant",
117            "project",
118            TASK_ACTION_GET,
119            &task.id,
120            task.descriptor_digest.as_deref().expect("fixture digest"),
121            now,
122            now + Duration::minutes(1),
123            "nonce",
124            None,
125            "grant",
126            "rotated-key",
127        );
128        (task, request)
129    }
130
131    #[test]
132    fn target_binding_requires_all_persisted_authority_dimensions() {
133        let (task, request) = fixture();
134        assert!(matches_target(&task, &request));
135        for field in 0..7 {
136            let mut changed = task.clone();
137            match field {
138                0 => changed.id.push('x'),
139                1 => changed.from_agent.push('x'),
140                2 => changed.to_agent.push('x'),
141                3 => changed.tenant_id = None,
142                4 => changed.project_id = None,
143                5 => changed.descriptor_digest = None,
144                6 => changed.authority_key_id = None,
145                _ => unreachable!(),
146            }
147            assert!(!matches_target(&changed, &request), "dimension {field}");
148        }
149    }
150
151    #[test]
152    fn unauthenticated_status_request_never_touches_storage() {
153        let (_, request) = fixture();
154        let dir = tempfile::tempdir().expect("temporary directory");
155        let path = dir.path().join("absent-parent").join("tasks.json");
156        let policy = TaskAuthorityConfigV1::default();
157        let expected = TaskAuthorityExpectationV1 {
158            sender: "owner",
159            recipient: "receiver",
160            tenant_id: "tenant",
161            project_id: "project",
162            now: Utc::now(),
163        };
164        assert!(matches!(
165            get_status(&path, &request, &policy, &expected),
166            Err(TaskControlError::Authority(_))
167        ));
168        assert!(!path.parent().expect("parent").exists());
169    }
170
171    fn authorized_cancel() -> (Task, TaskControlDescriptorV1, TaskAuthorityConfigV1) {
172        use super::super::task::{TaskCapabilityGrantV1, TaskPeerTrustV1, TaskScopeV1};
173        let (task, mut request) = fixture();
174        let key = ed25519_dalek::SigningKey::from_bytes(&[7; 32]);
175        request.action = TASK_ACTION_CANCEL.into();
176        request.reason = Some("owner requested cancellation".into());
177        request.sign(&key);
178        let policy = TaskAuthorityConfigV1 {
179            schema_version: 1,
180            peers: vec![TaskPeerTrustV1 {
181                schema_version: 1,
182                key_id: request.key_id.clone(),
183                agent_id: request.sender.clone(),
184                public_key: crate::core::agent_identity::hex_encode(key.verifying_key().as_bytes()),
185                allowed_actions: vec![TASK_ACTION_CANCEL.into()],
186                allowed_scopes: vec![TaskScopeV1 {
187                    schema_version: 1,
188                    tenant_id: request.tenant_id.clone(),
189                    project_id: request.project_id.clone(),
190                }],
191                not_before: request.issued_at - Duration::minutes(1),
192                expires_at: request.expires_at + Duration::minutes(1),
193                revoked: false,
194            }],
195            grants: vec![TaskCapabilityGrantV1 {
196                schema_version: 1,
197                grant_id: request.grant_ref.grant_id.clone(),
198                key_id: request.key_id.clone(),
199                action: TASK_ACTION_CANCEL.into(),
200                tenant_id: request.tenant_id.clone(),
201                project_id: request.project_id.clone(),
202                not_before: request.issued_at - Duration::minutes(1),
203                expires_at: request.expires_at + Duration::minutes(1),
204                revoked: false,
205            }],
206        };
207        (task, request, policy)
208    }
209
210    fn expected(request: &TaskControlDescriptorV1) -> TaskAuthorityExpectationV1<'_> {
211        TaskAuthorityExpectationV1 {
212            sender: "owner",
213            recipient: "receiver",
214            tenant_id: "tenant",
215            project_id: "project",
216            now: request.issued_at,
217        }
218    }
219
220    fn persist(path: &Path, task: Task) -> Vec<u8> {
221        TaskStore::mutate_locked(path, |store| {
222            store.tasks = vec![task];
223            Ok::<(), std::io::Error>(())
224        })
225        .expect("persist fixture");
226        std::fs::read(path).expect("fixture bytes")
227    }
228
229    #[test]
230    fn signed_cancel_is_durable_concurrent_and_hidden_from_legacy() {
231        let (task, request, policy) = authorized_cancel();
232        let dir = tempfile::tempdir().unwrap();
233        let path = dir.path().join("tasks.json");
234        persist(&path, task);
235        let lock_timeout = format!(
236            "lock {} timed out after {}ms",
237            path.with_extension("lock").display(),
238            crate::core::agents::ACQUIRE_TIMEOUT.as_millis()
239        );
240        let operation = || {
241            // Like concurrent_mutations_do_not_lose_tasks, test durable
242            // idempotence under contention, not a loaded runner's scheduling.
243            // This exact lock timeout occurs before loading or mutating storage.
244            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
245            loop {
246                match cancel(&path, &request, &policy, &expected(&request)) {
247                    Ok(status) => break status,
248                    Err(TaskControlError::Storage(error))
249                        if error.to_string() == lock_timeout
250                            && std::time::Instant::now() < deadline => {}
251                    Err(error) => panic!("signed cancellation failed: {error:?}"),
252                }
253            }
254        };
255        let (left, right) = std::thread::scope(|scope| {
256            let first = scope.spawn(operation);
257            let second = scope.spawn(operation);
258            (first.join().unwrap(), second.join().unwrap())
259        });
260        assert_eq!(left, right);
261        assert_eq!(left.state, TaskState::Canceled);
262        TaskStore::read_locked(&path, |store| {
263            let task = store.get_task(&request.task_id).unwrap();
264            assert_eq!(task.status_v1(), left);
265            assert_eq!(task.history.len(), 2);
266            assert_eq!(task.history[1].reason, request.reason);
267        })
268        .unwrap();
269        let before = std::fs::read(&path).unwrap();
270        for method in [TASK_ACTION_GET, TASK_ACTION_CANCEL] {
271            let legacy = super::super::a2a_compat::JsonRpcRequest {
272                jsonrpc: "2.0".into(),
273                id: 1.into(),
274                method: method.into(),
275                params: serde_json::json!({"id": request.task_id}),
276            };
277            let response = super::super::a2a_compat::handle_a2a_jsonrpc_at_path(&legacy, &path);
278            assert_eq!(response.error.unwrap().message, "task not found");
279            assert_eq!(std::fs::read(&path).unwrap(), before);
280        }
281    }
282
283    #[test]
284    fn signed_cancel_denies_every_target_mismatch_without_writing() {
285        let (task, request, policy) = authorized_cancel();
286        let dir = tempfile::tempdir().unwrap();
287        let path = dir.path().join("tasks.json");
288        for field in 0..7 {
289            let mut changed = task.clone();
290            match field {
291                0 => changed.id.push('x'),
292                1 => changed.from_agent.push('x'),
293                2 => changed.to_agent.push('x'),
294                3 => changed.tenant_id = None,
295                4 => changed.project_id = None,
296                5 => changed.descriptor_digest = None,
297                6 => changed.authority_key_id = None,
298                _ => unreachable!(),
299            }
300            let before = persist(&path, changed);
301            assert!(
302                matches!(
303                    cancel(&path, &request, &policy, &expected(&request)),
304                    Err(TaskControlError::NotFound)
305                ),
306                "dimension {field}"
307            );
308            assert_eq!(std::fs::read(&path).unwrap(), before);
309        }
310    }
311
312    #[test]
313    fn signed_cancel_rejects_invalid_authority_before_storage() {
314        let (_, request, policy) = authorized_cancel();
315        let dir = tempfile::tempdir().unwrap();
316        let path = dir.path().join("absent-parent/tasks.json");
317        for field in 0..5 {
318            let mut request = request.clone();
319            let mut policy = policy.clone();
320            match field {
321                0 => request.action = TASK_ACTION_GET.into(),
322                1 => request.signature = "00".repeat(64),
323                2 => policy.peers[0].revoked = true,
324                3 => policy.grants[0].revoked = true,
325                4 => policy.grants.clear(),
326                _ => unreachable!(),
327            }
328            assert!(matches!(
329                cancel(&path, &request, &policy, &expected(&request)),
330                Err(TaskControlError::Authority(_))
331            ));
332            assert!(!path.parent().unwrap().exists());
333        }
334    }
335
336    #[test]
337    fn signed_cancel_preserves_other_terminal_states_and_corrupt_storage() {
338        let (task, request, policy) = authorized_cancel();
339        let dir = tempfile::tempdir().unwrap();
340        let path = dir.path().join("tasks.json");
341        for state in [TaskState::Completed, TaskState::Failed] {
342            let mut task = task.clone();
343            task.transition(TaskState::Working, None).unwrap();
344            task.transition(state, None).unwrap();
345            let before = persist(&path, task);
346            assert!(matches!(
347                cancel(&path, &request, &policy, &expected(&request)),
348                Err(TaskControlError::InvalidTransition)
349            ));
350            assert_eq!(std::fs::read(&path).unwrap(), before);
351        }
352        std::fs::write(&path, "invalid fixture").unwrap();
353        assert!(matches!(
354            cancel(&path, &request, &policy, &expected(&request)),
355            Err(TaskControlError::Storage(_))
356        ));
357        assert_eq!(std::fs::read(&path).unwrap(), b"invalid fixture");
358    }
359
360    #[test]
361    fn signed_cancel_never_persists_an_unreadable_history_overflow() {
362        let (mut task, request, policy) = authorized_cancel();
363        for index in 0..2046 {
364            task.transition(
365                if index % 2 == 0 {
366                    TaskState::Working
367                } else {
368                    TaskState::InputRequired
369                },
370                None,
371            )
372            .unwrap();
373        }
374        // Reconstruct the last entry of a legacy full history explicitly: new
375        // authoring reserves it for a terminal state. Preserve the denial test.
376        task.history.push(super::super::task::TaskTransition {
377            from: task.state.clone(),
378            to: TaskState::Working,
379            timestamp: task.updated_at,
380            reason: None,
381        });
382        task.state = TaskState::Working;
383        assert_eq!(task.history.len(), 2048);
384        let dir = tempfile::tempdir().unwrap();
385        let path = dir.path().join("tasks.json");
386        let before = persist(&path, task);
387        assert!(matches!(
388            cancel(&path, &request, &policy, &expected(&request)),
389            Err(TaskControlError::InvalidTransition)
390        ));
391        assert_eq!(std::fs::read(&path).unwrap(), before);
392        assert!(TaskStore::read_locked(&path, |_| ()).is_ok());
393    }
394
395    #[test]
396    fn signed_cancel_uses_reserved_capacity_and_retries_without_growth() {
397        let (mut task, request, policy) = authorized_cancel();
398        while task.history.len() < 2047 {
399            let next = if task.state == TaskState::Working {
400                TaskState::InputRequired
401            } else {
402                TaskState::Working
403            };
404            task.transition(next, None).unwrap();
405        }
406        let dir = tempfile::tempdir().unwrap();
407        let path = dir.path().join("tasks.json");
408        persist(&path, task);
409        let first = cancel(&path, &request, &policy, &expected(&request)).unwrap();
410        assert_eq!(first.state, TaskState::Canceled);
411        let stored = TaskStore::read_locked(&path, |store| store.tasks[0].clone()).unwrap();
412        assert_eq!(stored.history.len(), 2048);
413        assert_eq!(stored.history.last().unwrap().timestamp, stored.updated_at);
414        assert_eq!(
415            cancel(&path, &request, &policy, &expected(&request)).unwrap(),
416            first
417        );
418        let retry = TaskStore::read_locked(&path, |store| store.tasks[0].clone()).unwrap();
419        assert_eq!(
420            serde_json::to_vec(&retry).unwrap(),
421            serde_json::to_vec(&stored).unwrap()
422        );
423    }
424
425    #[test]
426    fn signed_cancel_rejects_backward_clock_without_touching_storage() {
427        let (mut task, request, policy) = authorized_cancel();
428        task.updated_at = Utc::now() + Duration::hours(1);
429        let dir = tempfile::tempdir().unwrap();
430        let path = dir.path().join("tasks.json");
431        let before = persist(&path, task);
432        assert!(matches!(
433            cancel(&path, &request, &policy, &expected(&request)),
434            Err(TaskControlError::InvalidTransition)
435        ));
436        assert_eq!(std::fs::read(&path).unwrap(), before);
437        assert!(TaskStore::read_locked(&path, |_| ()).is_ok());
438    }
439}
lean_ctx/core/a2a/task_control.rs

lean_ctx/core/a2a/
task_control.rs
