diff --git a/tests/data/wire_payloads/anthropic_hoist__mid_orphan.json b/tests/data/wire_payloads/anthropic_hoist__mid_orphan.json index 2991de1b..4d6c1554 100644 --- a/tests/data/wire_payloads/anthropic_hoist__mid_orphan.json +++ b/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" diff --git a/tests/data/wire_payloads/anthropic_hoist__native_orphan.json b/tests/data/wire_payloads/anthropic_hoist__native_orphan.json index 849d51ea..b8cb4249 100644 --- a/tests/data/wire_payloads/anthropic_hoist__native_orphan.json +++ b/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" diff --git a/tests/data/wire_payloads/anthropic_hoist__trailing_orphan.json b/tests/data/wire_payloads/anthropic_hoist__trailing_orphan.json index 6f1c242b..b362c8d0 100644 --- a/tests/data/wire_payloads/anthropic_hoist__trailing_orphan.json +++ b/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" diff --git a/tests/data/wire_payloads/anthropic_native__mid_orphan.json b/tests/data/wire_payloads/anthropic_native__mid_orphan.json index 1f0fbbc7..5a126b77 100644 --- a/tests/data/wire_payloads/anthropic_native__mid_orphan.json +++ b/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" diff --git a/tests/data/wire_payloads/anthropic_native__native_orphan.json b/tests/data/wire_payloads/anthropic_native__native_orphan.json index bd675879..a3c30f57 100644 --- a/tests/data/wire_payloads/anthropic_native__native_orphan.json +++ b/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" diff --git a/tests/data/wire_payloads/anthropic_native__trailing_orphan.json b/tests/data/wire_payloads/anthropic_native__trailing_orphan.json index b35b649c..73e5eb2b 100644 --- a/tests/data/wire_payloads/anthropic_native__trailing_orphan.json +++ b/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" diff --git a/tests/data/wire_payloads/google__mid_orphan.json b/tests/data/wire_payloads/google__mid_orphan.json index 051efc68..e9e15e7b 100644 --- a/tests/data/wire_payloads/google__mid_orphan.json +++ b/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" }, diff --git a/tests/data/wire_payloads/google__native_orphan.json b/tests/data/wire_payloads/google__native_orphan.json index a3729c19..63b42760 100644 --- a/tests/data/wire_payloads/google__native_orphan.json +++ b/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" } diff --git a/tests/data/wire_payloads/google__trailing_orphan.json b/tests/data/wire_payloads/google__trailing_orphan.json index e44f6a50..a32ac0cd 100644 --- a/tests/data/wire_payloads/google__trailing_orphan.json +++ b/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" } diff --git a/tests/data/wire_payloads/openai_chat__mid_orphan.json b/tests/data/wire_payloads/openai_chat__mid_orphan.json index afc6621b..7eee7ef2 100644 --- a/tests/data/wire_payloads/openai_chat__mid_orphan.json +++ b/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" }, diff --git a/tests/data/wire_payloads/openai_chat__native_orphan.json b/tests/data/wire_payloads/openai_chat__native_orphan.json index ece1e77b..880f52de 100644 --- a/tests/data/wire_payloads/openai_chat__native_orphan.json +++ b/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" } diff --git a/tests/data/wire_payloads/openai_chat__trailing_orphan.json b/tests/data/wire_payloads/openai_chat__trailing_orphan.json index 5a23a92a..dfa8b38f 100644 --- a/tests/data/wire_payloads/openai_chat__trailing_orphan.json +++ b/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" } diff --git a/tests/data/wire_payloads/openai_responses__mid_orphan.json b/tests/data/wire_payloads/openai_responses__mid_orphan.json index 04d2b032..a8c133f2 100644 --- a/tests/data/wire_payloads/openai_responses__mid_orphan.json +++ b/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" }, { diff --git a/tests/data/wire_payloads/openai_responses__native_orphan.json b/tests/data/wire_payloads/openai_responses__native_orphan.json index 8b73d35d..c0d9346a 100644 --- a/tests/data/wire_payloads/openai_responses__native_orphan.json +++ b/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" } ], diff --git a/tests/data/wire_payloads/openai_responses__trailing_orphan.json b/tests/data/wire_payloads/openai_responses__trailing_orphan.json index 3eb13fbf..2afc12ae 100644 --- a/tests/data/wire_payloads/openai_responses__trailing_orphan.json +++ b/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" } ], diff --git a/tests/test_coordinator_tools.py b/tests/test_coordinator_tools.py index 54c00770..60e59630 100644 --- a/tests/test_coordinator_tools.py +++ b/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 diff --git a/tests/test_session_worker.py b/tests/test_session_worker.py index 7decf3ef..e2a4021b 100644 --- a/tests/test_session_worker.py +++ b/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 diff --git a/turnstone/core/lowering.py b/turnstone/core/lowering.py index 2e440304..56abe1ba 100644 --- a/turnstone/core/lowering.py +++ b/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( diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 030cfc9f..7ad590c4 100644 --- a/turnstone/core/session.py +++ b/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) diff --git a/turnstone/core/session_worker.py b/turnstone/core/session_worker.py index 982eb87b..b60345b2 100644 --- a/turnstone/core/session_worker.py +++ b/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: