Keep A2A canceled sticky against message/send persist

CHANGELOG.md

@@ -8,12 +8,13 @@ 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.
 - **Reopened todos clear `completed_at`.** `update_todo` (HTTP PATCH and
-  batch PATCH) now zeros `completed_at` when status moves back to
-  `pending` or `in_progress`, unless the caller passes an explicit
-  timestamp. Completing then reopening left the old stamp, so the
-  documented `completed_at = 0` until terminal contract was violated.
-  Subject/description edits on a terminal row still leave the stamp.
 - **MCP string JSON-RPC ids.** `parse_response` now accepts a decimal-string
 - **Reconcile rollback no longer wipes the workspace on a failed restore.**
 - **LaTeX math recursion depth.** `latex_math_to_plain` now stops converting

include/tenant_store.h

@@ -998,7 +998,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,

src/api_server.cpp

@@ -8279,14 +8279,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)));
 }
@@ -8750,9 +8767,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");

src/tenant_store.cpp

@@ -3143,17 +3143,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;
 }

tests/test_request_log.cpp

@@ -711,3 +711,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 == "{}");
+}
+