From 16d98707aa04246777bb9d15a2c474330559928d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 12 Sep 2026 12:23:30 +0000 Subject: [PATCH] Keep A2A canceled sticky against message/send persist. tasks/cancel can land while unary message/send is still unwinding. The stream path already refused to overwrite canceled; update_a2a_task now enforces the same so completed/failed cannot clobber cancel, and the send RPC returns request cancelled instead of a completed Task. Co-authored-by: Tyler Reckart --- CHANGELOG.md | 8 ++++++++ include/tenant_store.h | 4 +++- src/api_server.cpp | 31 +++++++++++++++++++++++------- src/tenant_store.cpp | 12 +++++++++++- tests/test_request_log.cpp | 39 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 85 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dae12b43..9173f508 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ loosely while pre-1.0 (breaking changes can land on minor bumps). ## [Unreleased] +### Fixed +- **A2A `tasks/cancel` survives `message/send`.** `update_a2a_task` no + longer overwrites a persisted `canceled` row with `completed`/`failed` + when an in-flight unary send finishes after `tasks/cancel`. The stream + path already refused that overwrite; the send path now matches, and the + RPC returns `request cancelled` instead of a completed Task that + `tasks/get` would contradict. + ## [0.13.4] — 2026-09-10 ### Fixed diff --git a/include/tenant_store.h b/include/tenant_store.h index 8afa1db1..92cdeb93 100644 --- a/include/tenant_store.h +++ b/include/tenant_store.h @@ -986,7 +986,9 @@ class TenantStore { const std::string& state); // Update state + payload columns. No-op if the row is missing for - // this tenant; returns true on actual change. + // this tenant, or if the row is already `canceled` and `state` is + // not `canceled` (tasks/cancel must win over an in-flight + // message/send terminal persist). Returns true on actual change. bool update_a2a_task(int64_t tenant_id, const std::string& task_id, const std::string& state, diff --git a/src/api_server.cpp b/src/api_server.cpp index 019b9b40..fd059ac6 100644 --- a/src/api_server.cpp +++ b/src/api_server.cpp @@ -8033,14 +8033,31 @@ void handle_a2a_message_send(int fd, // Message — history/artifacts can be reconstructed by combining // the user's input (which the client already has) with the // assistant's reply. Smaller column, simpler reads. + // + // tasks/cancel may have persisted canceled while send() was still + // unwinding. update_a2a_task refuses to overwrite canceled; if + // that CAS misses, surface the same cancelled RPC as the + // error_type=="cancelled" path instead of returning a completed + // Task that tasks/get would contradict. std::string final_msg_json; if (task.status.message) { final_msg_json = json_serialize(*a2a::to_json(*task.status.message)); } - tenants.update_a2a_task(tenant.id, task_id, - a2a::task_state_to_string(task.status.state), - final_msg_json, - resp.ok ? "" : sanitised_api_response_error(resp)); + const std::string terminal = + a2a::task_state_to_string(task.status.state); + if (!tenants.update_a2a_task(tenant.id, task_id, + terminal, + final_msg_json, + resp.ok ? "" : sanitised_api_response_error(resp))) { + if (auto rec = tenants.get_a2a_task(tenant.id, task_id); + rec && rec->state == + a2a::task_state_to_string(a2a::TaskState::canceled)) { + write_a2a_rpc(fd, a2a::make_error_response( + rpc_id, a2a::RPC_INVALID_REQUEST, + "request cancelled")); + return; + } + } write_a2a_rpc(fd, a2a::make_result_response(rpc_id, a2a::to_json(task))); } @@ -8503,9 +8520,9 @@ void handle_a2a_tasks_cancel(int fd, if (cancelled_in_flight) { // Persist canceled state so a follow-up tasks/get reflects the - // outcome. The orchestrator's send may still be unwinding — - // its terminal-state update will run, but we want canceled to - // win over completed/failed for clarity. Re-update. + // outcome. The orchestrator's send may still be unwinding; + // update_a2a_task keeps canceled sticky so a later + // completed/failed persist cannot clobber this write. tenants.update_a2a_task(tenant.id, task_id, a2a::task_state_to_string(a2a::TaskState::canceled), "", "canceled by tasks/cancel"); diff --git a/src/tenant_store.cpp b/src/tenant_store.cpp index b48878fe..a68ad986 100644 --- a/src/tenant_store.cpp +++ b/src/tenant_store.cpp @@ -3073,17 +3073,27 @@ bool TenantStore::update_a2a_task(int64_t tenant_id, const std::string& error_message) { if (!db_) return false; const int64_t ts = now_epoch(); + // `canceled` is sticky: message/send and message/stream persist + // completed/failed after the RPC returns, and that write races + // tasks/cancel. The stream path already refused the overwrite + // after a GET; the unary path did not. Enforcing it here so + // every caller (including exception / tenant-disabled paths) + // keeps cancel as the durable outcome. Re-writing canceled + // (tasks/cancel itself) is still allowed so updated_at / error + // stay current. Stmt q(db_, "UPDATE a2a_tasks " " SET state = ?, updated_at = ?, " " final_message_json = ?, error_message = ? " - " WHERE tenant_id = ? AND task_id = ?;"); + " WHERE tenant_id = ? AND task_id = ? " + " AND (state != 'canceled' OR ? = 'canceled');"); q.bind(1, state); q.bind(2, ts); q.bind(3, final_message_json); q.bind(4, error_message); q.bind(5, tenant_id); q.bind(6, task_id); + q.bind(7, state); q.step(); return sqlite3_changes(db_) > 0; } diff --git a/tests/test_request_log.cpp b/tests/test_request_log.cpp index b6ba847f..2b26b04d 100644 --- a/tests/test_request_log.cpp +++ b/tests/test_request_log.cpp @@ -687,3 +687,42 @@ TEST_CASE("should_persist_conversation_turn keeps unfinished-but-useful turns") CHECK_FALSE(should_persist_conversation_turn(provider)); } +TEST_CASE("a2a task: canceled is not overwritten by completed or failed") { + TempDb db; TenantStore s; s.open(db.path.string()); + const int64_t tid = make_tenant(s, "acme"); + + s.create_a2a_task(tid, "task-send", "index", "ctx-1", "working"); + CHECK(s.update_a2a_task(tid, "task-send", "canceled", "", + "canceled by tasks/cancel")); + + // Unary message/send finishing after tasks/cancel must not revive + // the row as completed/failed — same contract the stream path + // already enforced in the handler. + CHECK_FALSE(s.update_a2a_task(tid, "task-send", "completed", + R"({"role":"agent"})", "")); + CHECK_FALSE(s.update_a2a_task(tid, "task-send", "failed", "", + "internal error")); + + auto rec = s.get_a2a_task(tid, "task-send"); + REQUIRE(rec); + CHECK(rec->state == "canceled"); + CHECK(rec->error_message == "canceled by tasks/cancel"); + CHECK(rec->final_message_json.empty()); + + // Re-writing canceled (tasks/cancel itself) still lands so + // updated_at / error stay current. + CHECK(s.update_a2a_task(tid, "task-send", "canceled", "", + "canceled by tasks/cancel")); + rec = s.get_a2a_task(tid, "task-send"); + REQUIRE(rec); + CHECK(rec->state == "canceled"); + + // Working → completed still works when cancel never landed. + s.create_a2a_task(tid, "task-ok", "index", "ctx-2", "working"); + CHECK(s.update_a2a_task(tid, "task-ok", "completed", "{}", "")); + auto ok = s.get_a2a_task(tid, "task-ok"); + REQUIRE(ok); + CHECK(ok->state == "completed"); + CHECK(ok->final_message_json == "{}"); +} +