fix(cancel): close workstream self-cancel gaps from the completeness review

Follow-up to the cancellation review — harden how cancel interacts with a
workstream's OWN turn and tools, not just its children and agents.

- wait_for_workstream: the wait loop holds no cancel handle and blocks on the
  child-event bus, so a cancelled coordinator parked in a wait stayed pinned
  for up to WAIT_MAX_TIMEOUT (600s). Add a cooperative check to the ~2s
  progress heartbeat — it raises GenerationCancelled, which propagates out of
  the otherwise cancel-blind wait (~2s abort).
- spawn_batch: stop creating the rest of the children once cancel is observed;
  already-spawned children stay recorded (they are live, durably parent-linked
  workstreams), the remainder are marked not-spawned.
- session worker: only clear _worker_running if this thread is still the
  current worker, so a late-finishing abandoned worker (force-cancel) can't
  clobber a live successor's flag — which would let a third send spawn a
  duplicate worker on the same session.
- bash silent-cancel: a SIGKILL'd silent command now records outcome-UNKNOWN
  (is_error, partial output kept) instead of a clean "Cancelled by user." that
  read as a successful empty result on replay.
- wire-repair: the last-resort orphan disposition now reads outcome-UNKNOWN,
  matching the cooperative-cancel message (unknown, never none).

Deferred: MCP / web_fetch / web_search remain uninterruptible mid-call,
bounded by tool_timeout; only bash is truly preemptible.

tests/data/wire_payloads/anthropic_hoist__mid_orphan.json

@@ -37,7 +37,7 @@
           "type": "tool_result"
         },
         {
-          "content": "Tool execution was cancelled.",
+          "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
           "is_error": true,
           "tool_use_id": "call_2",
           "type": "tool_result"

tests/data/wire_payloads/anthropic_hoist__native_orphan.json

@@ -33,7 +33,7 @@
     {
       "content": [
         {
-          "content": "Tool execution was cancelled.",
+          "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
           "is_error": true,
           "tool_use_id": "call_1",
           "type": "tool_result"

tests/data/wire_payloads/anthropic_hoist__trailing_orphan.json

@@ -24,7 +24,7 @@
     {
       "content": [
         {
-          "content": "Tool execution was cancelled.",
+          "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
           "is_error": true,
           "tool_use_id": "call_1",
           "type": "tool_result"

tests/data/wire_payloads/anthropic_native__mid_orphan.json

@@ -37,7 +37,7 @@
           "type": "tool_result"
         },
         {
-          "content": "Tool execution was cancelled.",
+          "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
           "is_error": true,
           "tool_use_id": "call_2",
           "type": "tool_result"

tests/data/wire_payloads/anthropic_native__native_orphan.json

@@ -33,7 +33,7 @@
     {
       "content": [
         {
-          "content": "Tool execution was cancelled.",
+          "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
           "is_error": true,
           "tool_use_id": "call_1",
           "type": "tool_result"

tests/data/wire_payloads/anthropic_native__trailing_orphan.json

@@ -24,7 +24,7 @@
     {
       "content": [
         {
-          "content": "Tool execution was cancelled.",
+          "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
           "is_error": true,
           "tool_use_id": "call_1",
           "type": "tool_result"

tests/data/wire_payloads/google__mid_orphan.json

@@ -33,7 +33,7 @@
       "tool_call_id": "call_1"
     },
     {
-      "content": "Tool execution was cancelled.",
+      "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "role": "tool",
       "tool_call_id": "call_2"
     },

tests/data/wire_payloads/google__native_orphan.json

@@ -20,7 +20,7 @@
       ]
     },
     {
-      "content": "Tool execution was cancelled.",
+      "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "role": "tool",
       "tool_call_id": "call_1"
     }

tests/data/wire_payloads/google__trailing_orphan.json

@@ -20,7 +20,7 @@
       ]
     },
     {
-      "content": "Tool execution was cancelled.",
+      "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "role": "tool",
       "tool_call_id": "call_1"
     }

tests/data/wire_payloads/openai_chat__mid_orphan.json

@@ -33,7 +33,7 @@
       "tool_call_id": "call_1"
     },
     {
-      "content": "Tool execution was cancelled.",
+      "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "role": "tool",
       "tool_call_id": "call_2"
     },

tests/data/wire_payloads/openai_chat__native_orphan.json

@@ -20,7 +20,7 @@
       ]
     },
     {
-      "content": "Tool execution was cancelled.",
+      "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "role": "tool",
       "tool_call_id": "call_1"
     }

tests/data/wire_payloads/openai_chat__trailing_orphan.json

@@ -20,7 +20,7 @@
       ]
     },
     {
-      "content": "Tool execution was cancelled.",
+      "content": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "role": "tool",
       "tool_call_id": "call_1"
     }

tests/data/wire_payloads/openai_responses__mid_orphan.json

@@ -27,7 +27,7 @@
     },
     {
       "call_id": "call_2",
-      "output": "Tool execution was cancelled.",
+      "output": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "type": "function_call_output"
     },
     {

tests/data/wire_payloads/openai_responses__native_orphan.json

@@ -21,7 +21,7 @@
     },
     {
       "call_id": "call_1",
-      "output": "Tool execution was cancelled.",
+      "output": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "type": "function_call_output"
     }
   ],

tests/data/wire_payloads/openai_responses__trailing_orphan.json

@@ -16,7 +16,7 @@
     },
     {
       "call_id": "call_1",
-      "output": "Tool execution was cancelled.",
+      "output": "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun executing before the generation was stopped; do not assume it did not run, and reconcile before re-issuing it.",
       "type": "function_call_output"
     }
   ],

tests/test_coordinator_tools.py

@@ -501,6 +501,29 @@ def test_wait_exec_dispatches_raw_args_to_client(coord_session):
     assert parsed["mode"] == "any"
 
 
+def test_wait_exec_progress_callback_observes_cancel(coord_session):
+    """The wait progress heartbeat is the cancel seam.  ``wait_for_workstream``
+    holds no cancel handle, so without this a cancelled coordinator parked in a
+    wait stays pinned for up to WAIT_MAX_TIMEOUT.  A GenerationCancelled raised
+    from the heartbeat callback propagates out of the (otherwise cancel-blind)
+    wait — _exec_wait_for_workstream's ``except Exception`` can't swallow it
+    (GenerationCancelled is a BaseException)."""
+    from turnstone.core.session import GenerationCancelled
+
+    sess, coord, _ui = coord_session
+
+    def _wait(ws_ids, *, timeout, mode, since, progress_callback):
+        # Simulate the wait loop's ~2s heartbeat firing after the owner cancels.
+        sess._cancel_event.set()
+        progress_callback({"a": {"state": "running"}}, 0.1)  # must raise
+        return {"results": {}, "complete": True, "elapsed": 0.1, "mode": mode}
+
+    coord.wait_for_workstream.side_effect = _wait
+    item = sess._prepare_tool(_tc("wait_for_workstream", {"ws_ids": ["a"]}))
+    with pytest.raises(GenerationCancelled):
+        sess._exec_wait_for_workstream(item)
+
+
 def test_wait_exec_default_timeout_when_omitted(coord_session):
     """timeout=None (omitted) becomes 60.0 in exec so the client receives
     a numeric value — explicit ``timeout=0`` is preserved (one-shot
@@ -1385,6 +1408,35 @@ def test_spawn_batch_exec_surfaces_per_item_errors_in_denied(coord_session):
     assert "skill not found" in body["denied"][0]["reason"]
 
 
+def test_spawn_batch_exec_stops_spawning_after_cancel(coord_session):
+    """A cancel mid-batch stops creating the REST of the children.  The
+    already-spawned child stays in ``results`` (it is a live remote
+    workstream); the remainder are marked not-spawned rather than created."""
+    sess, coord, _ui = coord_session
+
+    spawned: list[dict[str, Any]] = []
+
+    def _spawn(**kwargs):
+        n = len(spawned)
+        spawned.append(kwargs)
+        # Owner cancels right after the first child is created.
+        sess._cancel_event.set()
+        return {"ws_id": f"child-{n}", "name": "n", "node_id": "node", "status": 200}
+
+    coord.spawn.side_effect = _spawn
+    item = sess._prepare_tool(_tc("spawn_batch", {"children": _three_children()}))
+    _call_id, output = sess._exec_spawn_batch(item)
+    body = json.loads(output)
+
+    # Only the first child was actually spawned — the cancel halted the rest.
+    assert len(spawned) == 1
+    assert set(body["results"].keys()) == {"0"}
+    assert body["results"]["0"]["child_ws_id"] == "child-0"
+    # The remaining two are reported not-spawned (cancelled), not created.
+    cancelled = [d for d in body["denied"] if "cancelled" in d["reason"].lower()]
+    assert {d["idx"] for d in cancelled} == {1, 2}
+
+
 def test_spawn_batch_exec_continues_past_client_exception(coord_session):
     sess, coord, _ui = coord_session
 

tests/test_session_worker.py

@@ -189,6 +189,39 @@ def test_worker_finally_clears_flag_when_run_swallows() -> None:
 # ---------------------------------------------------------------------------
 
 
+def test_abandoned_worker_does_not_clear_successor_running_flag() -> None:
+    """A force-cancel abandons the worker (``ws.worker_thread`` is cleared /
+    reassigned to a successor).  When the abandoned thread finishes late, its
+    ``finally`` must NOT clear ``_worker_running`` out from under the live
+    successor — otherwise a third send sees ``_worker_running=False`` and
+    spawns a second concurrent worker on the same session."""
+    send_gate = threading.Event()
+    session = _SendSession(send_gate=send_gate)
+    ws = _make_ws(session)
+
+    ok = _send_message(ws, session, "hello")
+    assert ok is True
+    abandoned = ws.worker_thread
+    assert abandoned is not None
+
+    # Simulate force-abandon + a successor send claiming ownership while the
+    # original worker is still pinned inside run().
+    sentinel = threading.Thread(target=lambda: None, name="successor")
+    with ws._lock:
+        ws.worker_thread = sentinel
+        ws._worker_running = True
+
+    # Release the abandoned worker; it runs its finally.
+    send_gate.set()
+    abandoned.join(timeout=3.0)
+    assert not abandoned.is_alive()
+
+    # The successor's ownership is intact — the abandoned worker did not
+    # clobber the flag or the thread handle.
+    assert ws._worker_running is True
+    assert ws.worker_thread is sentinel
+
+
 def test_concurrent_send_produces_exactly_one_worker_thread() -> None:
     """Two simultaneous send() calls must land as exactly one worker
     spawn and one queued message — not two parallel workers on the

turnstone/core/lowering.py

@@ -50,11 +50,18 @@ from typing import Any
 from turnstone.core import fence
 from turnstone.core.trajectory import Turn, dicts_from_turns
 
-# The synthetic result body for a tool call that never produced output.  The
-# neutral turn carries ``is_error=True``; each translator renders that per its
-# format (Anthropic ``tool_result.is_error``; the OpenAI-compatible lanes have
-# no such field and drop it).
-CANCELLED_TOOL_RESULT = "Tool execution was cancelled."
+# The synthetic result body for a tool call that never produced output (the
+# last-resort wire-repair for an orphan the session layer didn't synthesize —
+# e.g. a force-abandoned worker).  The neutral turn carries ``is_error=True``;
+# each translator renders that per its format (Anthropic ``tool_result.is_error``;
+# the OpenAI-compatible lanes have no such field and drop it).  The body reads
+# outcome-UNKNOWN, matching the cooperative-cancel disposition: an unobserved
+# call must not read as "did not run" (unknown, never none).
+CANCELLED_TOOL_RESULT = (
+    "Tool execution was cancelled. Outcome UNKNOWN — this call may have begun "
+    "executing before the generation was stopped; do not assume it did not run, "
+    "and reconcile before re-issuing it."
+)
 
 
 def _find_orphaned_tool_calls(

turnstone/core/session.py

@@ -8484,6 +8484,19 @@ class ChatSession:
         denied: list[dict[str, Any]] = []
         for spec in children:
             idx = spec["idx"]
+            # A cancel mid-batch stops creating the REST of the children —
+            # don't keep spawning workstreams the owner asked to stop.
+            # Cooperative: observed at this safe point between spawns, never
+            # mid-spawn.  Children already spawned this batch stay in
+            # ``results`` and are reported below (their ws_ids must survive —
+            # they are live remote workstreams, also durably parent-linked in
+            # storage); the rest are marked not-spawned.  Checked with the
+            # raw flag, not ``_check_cancelled``, so we fall through to the
+            # normal report path rather than raising and dropping the
+            # already-spawned ws_ids.
+            if self._cancel_event.is_set():
+                denied.append({"idx": idx, "reason": "not spawned: cancelled"})
+                continue
             # Validation failures from _prepare surface here as denied
             # rows — partial-success: don't abort the rest of the batch.
             if "_error" in spec:
@@ -10434,6 +10447,16 @@ class ChatSession:
         progress_heartbeat_s = 5.0
 
         def _progress(snap: dict[str, Any], elapsed: float) -> None:
+            # Cooperative cancel seam.  ``wait_for_workstream`` holds no
+            # cancel handle and its wait loop blocks on the ChildEventBus
+            # (woken only by *child* state changes), so without this a
+            # cancelled coordinator parked in a wait stays pinned for up to
+            # WAIT_MAX_TIMEOUT (600s).  The loop calls this callback every
+            # ~2s heartbeat and wraps it in ``except Exception`` — but
+            # ``GenerationCancelled`` is a ``BaseException``, so raising
+            # here propagates cleanly out of the wait into the send() cancel
+            # handler.  ~2s abort instead of up to 600s.
+            self._check_cancelled()
             now = time.monotonic()
             changed = snap != progress_state["last_snap"]
             heartbeat_due = (now - progress_state["last_emit_mono"]) >= progress_heartbeat_s
@@ -11157,8 +11180,21 @@ class ChatSession:
             # Distinguish user cancel from unexpected SIGKILL.
             # Popen.returncode is negative of the signal number when killed.
             if cancel.is_set() and proc.returncode == -signal.SIGKILL:
-                msg = "Cancelled by user."
-                self._report_tool_result(call_id, "bash", msg)
+                # SIGKILL'd mid-flight (the command was parked on a silent
+                # read, so the in-loop cooperative check never fired).  Its
+                # side effects are unobserved: record outcome UNKNOWN and
+                # mark it an error, not a clean empty success — a destructive
+                # command killed here must not read as "did not run" on
+                # replay.  Keep whatever partial stdout we captured.
+                partial = "".join(stdout_parts).strip()
+                msg = (
+                    "Cancelled by user. Outcome UNKNOWN — the command was "
+                    "stopped mid-execution; it may have run partially or had "
+                    "side effects. Do not assume it did not run."
+                )
+                if partial:
+                    msg += "\n\nPartial output before cancel:\n" + self._truncate_output(partial)
+                self._report_tool_result(call_id, "bash", msg, is_error=True)
                 return call_id, msg
 
             output = "".join(stdout_parts)

turnstone/core/session_worker.py

@@ -86,7 +86,16 @@ def send(
             log.exception("session_worker.uncaught ws=%s", ws.id[:8])
         finally:
             with ws._lock:
-                ws._worker_running = False
+                # Only clear the flag if THIS thread is still the current
+                # worker.  A force-cancel abandons the worker
+                # (``ws.worker_thread = None``) and a follow-up send may
+                # already have spawned a successor (``ws.worker_thread`` =
+                # the new thread); an abandoned thread finishing late must
+                # not clear the flag out from under that live successor —
+                # else a third send sees ``_worker_running=False`` and
+                # spawns a second concurrent worker on the same session.
+                if ws.worker_thread is threading.current_thread():
+                    ws._worker_running = False
 
     with ws._lock:
         if ws._worker_running: