diff --git a/docs/bulk-endpoints.md b/docs/bulk-endpoints.md index 48c3add7..d3c6a422 100644 --- a/docs/bulk-endpoints.md +++ b/docs/bulk-endpoints.md @@ -12,7 +12,6 @@ Existing bulk endpoints at time of writing: |---------------------------------------------------------|--------------------------|------------------------------------------| | `GET /v1/api/cluster/ws/live?ids=a,b,c` | bulk read | `{results, denied, truncated}` | | model tool `spawn_batch` | bulk create (per-item) | `{results, denied}` | -| `POST /v1/api/workstreams/{ws_id}/stop_cascade` | cascade mutation | `{cancelled, failed, skipped}` | | `POST /v1/api/workstreams/{ws_id}/close_all_children` | cascade mutation | `{closed, failed, skipped}` | --- @@ -146,7 +145,7 @@ consistently-typed across the read and create cases. ``` Where `` is the endpoint-specific name for "succeeded" — -`cancelled` for `stop_cascade`, `closed` for `close_all_children`. +`closed` for `close_all_children`. The three buckets partition the input set exactly once: | Bucket | Meaning | @@ -161,20 +160,6 @@ be partial. `skipped` is pre-resolved — the target is already in the terminal state the cascade was aiming at, so it's neither a win to report nor a fault to fix. -### Example — `stop_cascade` - -```json -{ - "status": "ok", - "cancelled": ["child-1", "child-3"], - "failed": [], - "skipped": ["child-2"] -} -``` - -A subsequent retry would target only `failed` ids, not `skipped` -ones — the latter are already done. - ### Example — `close_all_children` ```json @@ -186,10 +171,12 @@ ones — the latter are already done. } ``` -Same partition, different success-bucket name. When `coord_client` -is unavailable (session loaded but no HTTP client attached — a -construction bug) every id goes to `failed` so the operator notices -rather than getting a silent all-skipped response. +Here the success bucket is `closed`. A subsequent retry would +target only `failed` ids, not `skipped` ones — the latter are +already done. When `coord_client` is unavailable (session loaded +but no HTTP client attached — a construction bug) every id goes to +`failed` so the operator notices rather than getting a silent +all-skipped response. --- @@ -232,12 +219,12 @@ rather than getting a silent all-skipped response. - **Phase 6** shipped `cluster/ws/live` as the first Shape A endpoint (`{results, denied, truncated}`). -- **Phase 7** shipped `stop_cascade` as the first Shape B endpoint - (`{cancelled, failed, skipped}`). +- **Phase 7** introduced the Shape B cascade-mutation envelope + (`{, failed, skipped}`) for the coordinator's + cancel-cascade path. - **Phase 8 PR A** shipped `spawn_batch` (Shape A, keyed by idx) and - `close_all_children` (Shape B, twin of `stop_cascade`), which - crystallised the two-shape-per-semantic-category policy codified - here. + `close_all_children` (Shape B), which crystallised the + two-shape-per-semantic-category policy codified here. Before adding a third shape, read this doc and argue for why the new surface doesn't fit either A or B. Two idioms in the cluster diff --git a/docs/coordinator-api-tour.md b/docs/coordinator-api-tour.md index 18fcd192..991b8190 100644 --- a/docs/coordinator-api-tour.md +++ b/docs/coordinator-api-tour.md @@ -18,7 +18,7 @@ schema changes. > auth and the `admin.coordinator` permission. A session-scoped JWT > is minted per login (see [docs/oidc.md](oidc.md) / [docs/security.md](security.md)); > a service token may call the read paths but destructive governance -> paths (`/restrict`, `/stop_cascade`, `/close_all_children`) require +> paths (`/restrict`, `/close_all_children`) require > the explicit `admin.coordinator` grant — a service-token owner > match isn't enough. @@ -44,7 +44,6 @@ schema changes. | 6 | Wait for fan-out | model-side tool `wait_for_workstream` | | 7 | Govern | `POST /v1/api/workstreams/{ws_id}/trust` | | | | `POST /v1/api/workstreams/{ws_id}/restrict` | -| | | `POST /v1/api/workstreams/{ws_id}/stop_cascade` | | | | `POST /v1/api/workstreams/{ws_id}/close_all_children` | | 8 | Approve / cancel | `POST /v1/api/workstreams/{ws_id}/approve` | | | | `POST /v1/api/workstreams/{ws_id}/cancel` | @@ -53,7 +52,7 @@ schema changes. Refer to `/openapi.json` (Swagger UI at `/docs`) on any `turnstone-console` process for the authoritative operation ids and schemas. Coordinator-only verbs (`/children`, `/trust`, `/restrict`, -`/stop_cascade`, `/close_all_children`) 404 against `kind=interactive` +`/close_all_children`) 404 against `kind=interactive` rows; the shared verbs (`/send`, `/approve`, `/cancel`, `/events`, `/history`, `/open`, `/close`, etc.) work on both kinds. @@ -261,10 +260,10 @@ rounds to a 10× token-efficiency win. --- -## 7. Governance — trust, restrict, stop_cascade, close_all_children +## 7. Governance — trust, restrict, close_all_children -These four endpoints let an operator steer a live coordinator session -mid-flight. All four emit an audit event tagged +These three endpoints let an operator steer a live coordinator session +mid-flight. All three emit an audit event tagged `coordinator.` via the dedicated audit executor so a cascade burst can't starve audit writes. @@ -294,28 +293,6 @@ idempotent — calling twice with overlapping lists converges to the union. Revocations don't survive a session close/reopen; operators opt in per session. Cap 256 tool names per request, 128 chars each. -### `POST /stop_cascade` — cancel the subtree - -```http -POST /v1/api/workstreams/{ws_id}/stop_cascade -{} -``` - -Cancels the coordinator's in-flight generation AND dispatches -`cancel_workstream` through the routing proxy for every direct -child in the in-memory registry. Returns: - -```json -{"status": "ok", "cancelled": ["child-1", "child-3"], "failed": [], "skipped": ["child-2"]} -``` - -Response uses the [cascade-mutation bulk shape](bulk-endpoints.md): -`cancelled` = accepted, `failed` = dispatch error worth retrying, -`skipped` = upstream 404 (already gone — stale registry entry or -the row was deleted between snapshot and dispatch). Grandchildren -aren't touched directly; they sit behind their parent's cancel and -propagate via the child's SSE stream. - ### `POST /close_all_children` — soft-close the direct fan-out ```http @@ -329,16 +306,16 @@ Response: {"status": "ok", "closed": ["c-1", "c-2"], "failed": [], "skipped": []} ``` -Soft-close cascade bounded by the same semaphore as `stop_cascade`. -The `reason` (up to 512 chars) propagates into each closed child's -audit + `workstream_config` for postmortem. Unlike `stop_cascade` -this does NOT recurse into grandchildren — the model-facing tool -that pairs with this endpoint asks for a bounded teardown of the -coordinator's own fan-out. For a full-subtree teardown, use -`stop_cascade`. +Soft-close cascade bounded by a concurrency semaphore. The `reason` +(up to 512 chars) propagates into each closed child's audit + +`workstream_config` for postmortem. The model-facing tool that +pairs with this endpoint asks for a bounded teardown of the +coordinator's own fan-out. This *soft-closes*; to *cancel* the +fan-out instead, cancel the coordinator (§8) — a coordinator cancel +auto-cascades to its direct children. -See [bulk-endpoints.md](bulk-endpoints.md) for why both endpoints -share the cascade-mutation shape and how it differs from the +See [bulk-endpoints.md](bulk-endpoints.md) for why `close_all_children` +uses the cascade-mutation shape and how it differs from the `spawn_batch` / `cluster/ws/live` shape. --- @@ -356,7 +333,10 @@ POST /v1/api/workstreams/{ws_id}/approve {"approved": true, "feedback": null, "always": true} // always-approve this tool name ``` -`cancel` drops the in-flight generation but leaves the coordinator +`cancel` drops the coordinator's in-flight generation and, for a +coordinator, auto-cascades the cancel to its direct children: +`cancel_workstream` is dispatched through the routing proxy for +every direct child in the registry. The coordinator itself is left idle and open for a fresh `send`: ```http @@ -373,9 +353,10 @@ POST /v1/api/workstreams/{ws_id}/close {} ``` -Soft-closes the session — state persists, children keep running (use -`close_all_children` or `stop_cascade` first to wind them down), the -worker thread exits, SSE streams send a final `stream_end` and +Soft-closes the session — state persists, children keep running +(wind them down first with `close_all_children`, or by cancelling +the coordinator, which cascades the cancel to its direct children), +the worker thread exits, SSE streams send a final `stream_end` and disconnect. The row is reopenable via `POST /v1/api/workstreams/{ws_id}/open` so long as it hasn't been deleted. @@ -390,7 +371,7 @@ deleted. - [bulk-endpoints.md](bulk-endpoints.md) — the two bulk-shape idioms (`{results, denied, truncated}` vs `{, failed, skipped}`) used by `cluster/ws/live`, - `spawn_batch`, `stop_cascade`, and `close_all_children`. + `spawn_batch`, and `close_all_children`. - [architecture.md](architecture.md) — cluster-wide architecture including how coordinator sessions fit next to node-hosted interactive workstreams. diff --git a/docs/coordinator-skills.md b/docs/coordinator-skills.md index 9d35a0c5..e66c9ece 100644 --- a/docs/coordinator-skills.md +++ b/docs/coordinator-skills.md @@ -351,7 +351,7 @@ persona drift without a real LLM in the loop. `spawn_batch` and `close_all_children` use, so your skill can parse results / denied arrays correctly. - [governance.md](governance.md) — the broader governance surface - (`/trust`, `/restrict`, `/stop_cascade`, role-based permissions) + (`/trust`, `/restrict`, role-based permissions) that wraps every coord session. - [settings.md](settings.md) — `coordinator.model_alias` and `coordinator.reasoning_effort` settings that gate which LLM runs diff --git a/tests/test_cancel.py b/tests/test_cancel.py index fa9f9261..064c1129 100644 --- a/tests/test_cancel.py +++ b/tests/test_cancel.py @@ -888,8 +888,12 @@ class TestSynthesizeCancelledResults: # All emitted as errors so the live UI renders them as # ``coord-tool-row-result--error``. assert all(tr[3] is True for tr in ui.tool_results) - # Reason text propagates as the synthetic tool output. - assert all(tr[2] == "Cancelled by user." for tr in ui.tool_results) + # Reason text propagates as a prefix, now followed by an explicit + # UNKNOWN-outcome clause (unknown, never none — see HYPOTHESIS.md): + # the call may have begun executing before cancel, so the synthetic + # result must not read as "it didn't happen." + assert all(tr[2].startswith("Cancelled by user.") for tr in ui.tool_results) + assert all("UNKNOWN" in tr[2] for tr in ui.tool_results) # And the message list has the synthesized tool entries # (preserves the prior contract). tool_msgs = [m for m in dicts_from_turns(session.messages) if m.get("role") == "tool"] @@ -952,3 +956,95 @@ class TestSynthesizeCancelledResults: tool_msgs = [m for m in dicts_from_turns(session.messages) if m.get("role") == "tool"] assert len(tool_msgs) == 1 + + +class TestCancelledAgentDisposition: + """A cancelled task_agent folds back an honest ledger, not a bare string. + + Regression guard for the HYPOTHESIS.md cancellation appendix: ρ may + fabricate the acknowledgment but must not fabricate the outcome — + ``unknown``, never ``none``. + """ + + @staticmethod + def _assistant(call_id, name): + return { + "role": "assistant", + "content": "", + "tool_calls": [{"id": call_id, "function": {"name": name}}], + } + + @staticmethod + def _result(call_id, text="ok"): + return {"role": "tool", "tool_call_id": call_id, "content": text} + + def test_no_actions_reports_no_side_effects(self, tmp_db): + session = _make_session() + out = session._cancelled_agent_disposition([], "task") + assert "no side effects" in out + assert "UNKNOWN" not in out + + def test_marks_most_recent_action_unknown(self, tmp_db): + session = _make_session() + # bash completed; web_fetch was in flight (issued, no result yet). + msgs = [ + self._assistant("t1", "bash"), + self._result("t1"), + self._assistant("t2", "web_fetch"), + ] + out = session._cancelled_agent_disposition(msgs, "task") + assert out != "(task interrupted by user)" + assert "Completed before cancel: bash." in out + assert "In flight at cancel: web_fetch" in out + assert "UNKNOWN" in out + + def test_partial_result_boundary_still_unknown(self, tmp_db): + # A SIGKILL'd bash appends a partial result before the next + # checkpoint raises — it is the boundary and must read UNKNOWN, + # not as a clean completion. + session = _make_session() + msgs = [self._assistant("t1", "bash"), self._result("t1", "(killed)")] + out = session._cancelled_agent_disposition(msgs, "task") + assert "In flight at cancel: bash" in out + assert "UNKNOWN" in out + assert "Completed before cancel" not in out + + def test_counts_and_not_started(self, tmp_db): + session = _make_session() + # Two tool_calls in one turn: t1 ran, t2 cancelled before running, + # t3 (a later turn's call) is the in-flight boundary. + msgs = [ + { + "role": "assistant", + "content": "", + "tool_calls": [ + {"id": "t1", "function": {"name": "bash"}}, + {"id": "t2", "function": {"name": "bash"}}, + ], + }, + self._result("t1"), + self._assistant("t3", "search"), + ] + out = session._cancelled_agent_disposition(msgs, "task") + assert "In flight at cancel: search" in out + assert "Not started (cancelled first): bash." in out + + def test_exec_task_routes_cancel_to_disposition(self, tmp_db): + """_exec_task converts a GenerationCancelled from _run_agent into the + honest disposition, reading the in-place-mutated agent_messages.""" + session = _make_session() + + def fake_run_agent(agent_messages, **kwargs): + agent_messages.append(self._assistant("t1", "bash")) + agent_messages.append(self._result("t1")) + agent_messages.append(self._assistant("t2", "web_fetch")) + raise GenerationCancelled() + + with patch.object(session, "_run_agent", side_effect=fake_run_agent): + call_id, result = session._exec_task({"call_id": "c1", "prompt": "do x"}) + + assert call_id == "c1" + assert result != "(task interrupted by user)" + assert "UNKNOWN" in result + assert "web_fetch" in result # in-flight boundary + assert "bash" in result # completed diff --git a/tests/test_coordinator_close_all_children.py b/tests/test_coordinator_close_all_children.py index 1dc63dd5..f548eb17 100644 --- a/tests/test_coordinator_close_all_children.py +++ b/tests/test_coordinator_close_all_children.py @@ -1,8 +1,7 @@ """Tests for the coordinator ``close_all_children`` endpoint. -Near-twin of the ``stop_cascade`` tests in -``test_coordinator_governance.py``. Keeps the close-cascade surface in -its own file so PR A's review surface stays tight. +Keeps the close-cascade surface in its own file so the review surface +stays tight. """ from __future__ import annotations @@ -219,7 +218,7 @@ def test_close_all_children_404_when_session_not_loaded(storage): def test_close_all_children_service_token_cannot_bypass_admin_coordinator(storage): """Destructive endpoint — a service token matching the coord owner still needs the explicit ``admin.coordinator`` grant. Mirrors the - stop_cascade treatment.""" + ``restrict`` treatment.""" mgr = _build_mgr(storage) coord = mgr.create(user_id="user-1", name="coord-a") coord.session = MagicMock() diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index 2fb49ba2..458b5435 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -31,6 +31,7 @@ from tests._coord_test_helpers import ( _build_mgr_with_factory, _fake_registry, _FakeConfigStore, + _seed_children, ) from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( @@ -1685,6 +1686,89 @@ def test_cancel_idle_workstream_does_not_broadcast_approval_resolved(storage): assert "approval_resolved" not in seen_types +def test_coord_cancel_cascades_to_children(storage): + """Cancelling a coordinator auto-propagates the cancel down its + spawned subtree (HYPOTHESIS.md cancellation appendix: cancel flows + down the subtree). The ``post_cancel`` hook fans ``coord_client.cancel`` + over the direct children after the coordinator's own session is + cancelled.""" + from unittest.mock import MagicMock + + from starlette.applications import Starlette + from starlette.routing import Route + from starlette.testclient import TestClient + + from turnstone.console.server import _cascade_cancel_to_children + from turnstone.core.session_routes import SessionEndpointConfig, make_cancel_handler + + mgr = _build_mgr(storage) + coord = mgr.create(user_id="user-1", name="coord-a") + _seed_children(mgr._adapter, coord.id, ["child-1", "child-2"]) + + coord_client = MagicMock() + coord_client.cancel.return_value = {"status": "ok"} + coord.session = MagicMock() + coord.session._coord_client = coord_client + + cfg = SessionEndpointConfig( + permission_gate=_require_admin_coordinator, + manager_lookup=lambda r: (mgr, None), + tenant_check=None, + not_found_label="coordinator not found", + audit_action_prefix="coordinator", + ) + handler = make_cancel_handler(cfg, post_cancel=_cascade_cancel_to_children) + app = Starlette(routes=[Route("/v1/api/workstreams/{ws_id}/cancel", handler, methods=["POST"])]) + app.state.coord_adapter = mgr._adapter + app.add_middleware(_AuthMiddleware) + client = TestClient(app) + + resp = client.post(f"/v1/api/workstreams/{coord.id}/cancel", headers=_COORD_HEADERS) + assert resp.status_code == 200 + # The coordinator's own session was cancelled (owner first)... + coord.session.cancel.assert_called_once() + # ...and every direct child received a cancel (subtree propagation). + cascaded = {c.args[0] for c in coord_client.cancel.call_args_list} + assert cascaded == {"child-1", "child-2"} + + +def test_coord_cancel_cascade_failure_does_not_fail_owner_cancel(storage): + """A cascade error must not strand the owner half-cancelled: the + ``post_cancel`` exception is swallowed and the owner's cancel still + returns 200 (the owner's own session was already cancelled).""" + from unittest.mock import MagicMock + + from starlette.applications import Starlette + from starlette.routing import Route + from starlette.testclient import TestClient + + from turnstone.core.session_routes import SessionEndpointConfig, make_cancel_handler + + mgr = _build_mgr(storage) + coord = mgr.create(user_id="user-1", name="coord-a") + coord.session = MagicMock() + + async def _boom(request, ws_id, ws): # noqa: ARG001 + raise RuntimeError("cascade blew up") + + cfg = SessionEndpointConfig( + permission_gate=_require_admin_coordinator, + manager_lookup=lambda r: (mgr, None), + tenant_check=None, + not_found_label="coordinator not found", + audit_action_prefix="coordinator", + ) + handler = make_cancel_handler(cfg, post_cancel=_boom) + app = Starlette(routes=[Route("/v1/api/workstreams/{ws_id}/cancel", handler, methods=["POST"])]) + app.add_middleware(_AuthMiddleware) + client = TestClient(app) + + resp = client.post(f"/v1/api/workstreams/{coord.id}/cancel", headers=_COORD_HEADERS) + assert resp.status_code == 200 + assert resp.json()["status"] == "ok" + coord.session.cancel.assert_called_once() + + # --------------------------------------------------------------------------- # Events (SSE replay shape) # --------------------------------------------------------------------------- diff --git a/tests/test_coordinator_governance.py b/tests/test_coordinator_governance.py index f2f5109a..f4e92bd3 100644 --- a/tests/test_coordinator_governance.py +++ b/tests/test_coordinator_governance.py @@ -1,10 +1,10 @@ """Tests for the coordinator governance endpoints and session hooks. -Covers the three console endpoints that let an operator steer a live -coordinator session mid-flight (``/trust``, ``/restrict``, -``/stop_cascade``), the two ``ChatSession`` methods the endpoints -toggle (``set_trust_send`` / ``revoke_tools``), the audit rows the -handlers emit, and the ``_prepare_tool`` revocation gate. +Covers the console endpoints that let an operator steer a live +coordinator session mid-flight (``/trust``, ``/restrict``), the two +``ChatSession`` methods the endpoints toggle (``set_trust_send`` / +``revoke_tools``), the audit rows the handlers emit, and the +``_prepare_tool`` revocation gate. """ from __future__ import annotations @@ -29,7 +29,6 @@ from tests._coord_test_helpers import ( ) from turnstone.console.server import ( coordinator_restrict, - coordinator_stop_cascade, coordinator_trust, ) from turnstone.core.auth import AuthResult @@ -42,7 +41,7 @@ def storage(tmp_path): def _make_client(storage, *, coord_mgr, alias="my-model", registry=None) -> TestClient: - """Starlette app exposing only the three governance endpoints.""" + """Starlette app exposing only the governance endpoints.""" app = Starlette( routes=[ Route( @@ -55,11 +54,6 @@ def _make_client(storage, *, coord_mgr, alias="my-model", registry=None) -> Test coordinator_restrict, methods=["POST"], ), - Route( - "/v1/api/workstreams/{ws_id}/stop_cascade", - coordinator_stop_cascade, - methods=["POST"], - ), ], middleware=[Middleware(_AuthMiddleware)], ) @@ -161,9 +155,9 @@ def _service_token_client( """Build a TestClient whose middleware injects a service-scoped token. Used to verify that the capability-escalating endpoints (``/trust``, - ``/restrict``, ``/stop_cascade``) do NOT honor the normal - ``require_permission`` service-scope bypass when the caller lacks - the specific grant they need. + ``/restrict``) do NOT honor the normal ``require_permission`` + service-scope bypass when the caller lacks the specific grant they + need. """ app = Starlette( routes=[ @@ -177,11 +171,6 @@ def _service_token_client( coordinator_restrict, methods=["POST"], ), - Route( - "/v1/api/workstreams/{ws_id}/stop_cascade", - coordinator_stop_cascade, - methods=["POST"], - ), ], ) app.state.coord_mgr = coord_mgr @@ -275,25 +264,6 @@ def test_restrict_service_token_cannot_bypass_admin_coordinator(storage): assert resp.status_code == 403 -def test_stop_cascade_service_token_cannot_bypass_admin_coordinator(storage): - """/stop_cascade mirrors /restrict — same destructive treatment.""" - mgr = _build_mgr(storage) - coord = mgr.create(user_id="svc-user", name="coord-a") - coord.session, _ = _make_session_mock() - - client = _service_token_client( - storage, - mgr, - user_id="svc-user", - permissions=frozenset(), - ) - resp = client.post( - f"/v1/api/workstreams/{coord.id}/stop_cascade", - json={}, - ) - assert resp.status_code == 403 - - def test_trust_toggle_rejects_non_bool(storage): mgr = _build_mgr(storage) coord = mgr.create(user_id="user-1", name="coord-a") @@ -633,139 +603,10 @@ def test_prepare_tool_allows_non_revoked_tool(): # --------------------------------------------------------------------------- -# /stop_cascade endpoint (item 5b) +# children_snapshot (used by the cancel cascade + close_all_children) # --------------------------------------------------------------------------- -def test_stop_cascade_cancels_coord_and_each_child(storage): - mgr = _build_mgr(storage) - coord = mgr.create(user_id="user-1", name="coord-a") - _seed_children(mgr._adapter, coord.id, ["child-1", "child-2", "child-3"]) - - def _cancel(wid: str) -> dict: - if wid == "child-2": - return {"error": "gateway_timeout", "status": 502} - return {"status": "ok"} - - coord_client = MagicMock() - coord_client.cancel.side_effect = _cancel - coord.session = MagicMock() - coord.session._coord_client = coord_client - - client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) - resp = client.post( - f"/v1/api/workstreams/{coord.id}/stop_cascade", - json={}, - headers=_COORD_HEADERS, - ) - assert resp.status_code == 200 - body = resp.json() - assert set(body["cancelled"] + body["failed"] + body["skipped"]) == { - "child-1", - "child-2", - "child-3", - } - assert body["failed"] == ["child-2"] - assert set(body["cancelled"]) == {"child-1", "child-3"} - assert body["skipped"] == [] - assert coord_client.cancel.call_count == 3 - - events = [ - e for e in storage.list_audit_events() if e["action"] == "coordinator.stopped_cascade" - ] - assert len(events) == 1 - detail = json.loads(events[0]["detail"]) - assert set(detail["cancelled"] + detail["failed"] + detail["skipped"]) == { - "child-1", - "child-2", - "child-3", - } - - -def test_stop_cascade_routes_404_to_skipped_bucket(storage): - """A stale registry entry (child row already deleted from storage) - or an upstream-404 on cancel is semantically 'already gone', not a - dispatch failure. Report it in ``skipped`` so operators can tell - them apart.""" - mgr = _build_mgr(storage) - coord = mgr.create(user_id="user-1", name="coord-a") - _seed_children(mgr._adapter, coord.id, ["stale-child"]) - - coord_client = MagicMock() - coord_client.cancel.return_value = { - "error": "workstream not in coordinator subtree: stale-child", - "status": 404, - } - coord.session = MagicMock() - coord.session._coord_client = coord_client - - client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) - resp = client.post( - f"/v1/api/workstreams/{coord.id}/stop_cascade", - json={}, - headers=_COORD_HEADERS, - ) - assert resp.status_code == 200 - body = resp.json() - assert body["cancelled"] == [] - assert body["failed"] == [] - assert body["skipped"] == ["stale-child"] - - -def test_stop_cascade_empty_children_still_audits(storage): - mgr = _build_mgr(storage) - coord = mgr.create(user_id="user-1", name="coord-a") - coord.session = MagicMock() - coord.session._coord_client = MagicMock() - - client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) - resp = client.post( - f"/v1/api/workstreams/{coord.id}/stop_cascade", - json={}, - headers=_COORD_HEADERS, - ) - assert resp.status_code == 200 - body = resp.json() - assert body == {"status": "ok", "cancelled": [], "failed": [], "skipped": []} - assert [e for e in storage.list_audit_events() if e["action"] == "coordinator.stopped_cascade"] - - -def test_stop_cascade_without_coord_client_marks_all_failed(storage): - """If the coord session has no attached coord_client (unexpected - state for a loaded session), every child routes to ``failed`` so - the operator can investigate.""" - mgr = _build_mgr(storage) - coord = mgr.create(user_id="user-1", name="coord-a") - _seed_children(mgr._adapter, coord.id, ["child-a", "child-b"]) - coord.session = MagicMock() - coord.session._coord_client = None - - client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) - resp = client.post( - f"/v1/api/workstreams/{coord.id}/stop_cascade", - json={}, - headers=_COORD_HEADERS, - ) - assert resp.status_code == 200 - body = resp.json() - assert body["cancelled"] == [] - assert body["skipped"] == [] - assert set(body["failed"]) == {"child-a", "child-b"} - - -def test_stop_cascade_404_when_session_not_loaded(storage): - mgr = _build_mgr(storage) - coord = mgr.create(user_id="user-1", name="coord-a") - coord.session = None - client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) - resp = client.post( - f"/v1/api/workstreams/{coord.id}/stop_cascade", - json={}, - headers=_COORD_HEADERS, - ) - assert resp.status_code == 404 - - def test_children_snapshot_returns_copy_not_live_set(storage): mgr = _build_mgr(storage) coord = mgr.create(user_id="user-1", name="coord-a") diff --git a/tests/test_session_routes.py b/tests/test_session_routes.py index c73aacc7..96516ab5 100644 --- a/tests/test_session_routes.py +++ b/tests/test_session_routes.py @@ -188,7 +188,6 @@ def test_register_coord_verbs_mounts_seven_paths() -> None: metrics=_stub, trust=_stub, restrict=_stub, - stop_cascade=_stub, close_all_children=_stub, ), ) @@ -199,7 +198,6 @@ def test_register_coord_verbs_mounts_seven_paths() -> None: ("/api/workstreams/{ws_id}/metrics", frozenset({"GET", "HEAD"})), ("/api/workstreams/{ws_id}/trust", frozenset({"POST"})), ("/api/workstreams/{ws_id}/restrict", frozenset({"POST"})), - ("/api/workstreams/{ws_id}/stop_cascade", frozenset({"POST"})), ("/api/workstreams/{ws_id}/close_all_children", frozenset({"POST"})), } diff --git a/turnstone/api/console_schemas.py b/turnstone/api/console_schemas.py index 1edb9023..741d99c7 100644 --- a/turnstone/api/console_schemas.py +++ b/turnstone/api/console_schemas.py @@ -1349,33 +1349,6 @@ class CoordinatorRestrictResponse(BaseModel): revoked_tools: list[str] = Field(description="Full post-revocation set of revoked tool names.") -class CoordinatorStopCascadeResponse(BaseModel): - """Response body for POST /v1/api/workstreams/{ws_id}/stop_cascade.""" - - status: str = Field(default="ok") - cancelled: list[str] = Field( - default_factory=list, - description="Child ws_ids that accepted the cancel dispatch.", - ) - failed: list[str] = Field( - default_factory=list, - description=( - "Child ws_ids whose cancel dispatch returned an error other " - "than an already-gone 404 — the cascade continues on per-" - "child failure so a single unreachable node doesn't abort " - "the whole batch." - ), - ) - skipped: list[str] = Field( - default_factory=list, - description=( - "Child ws_ids that returned 404 on cancel (already gone). " - "Reported separately from ``failed`` so operators can " - "distinguish already-done from dispatch-broken." - ), - ) - - class CoordinatorCloseAllChildrenRequest(BaseModel): """Body for POST /v1/api/workstreams/{ws_id}/close_all_children.""" diff --git a/turnstone/api/console_spec.py b/turnstone/api/console_spec.py index 5588f956..8e7a817d 100644 --- a/turnstone/api/console_spec.py +++ b/turnstone/api/console_spec.py @@ -35,7 +35,6 @@ from turnstone.api.console_schemas import ( CoordinatorRestrictResponse, CoordinatorSendRequest, CoordinatorSendResponse, - CoordinatorStopCascadeResponse, CoordinatorTaskInfo, CoordinatorTasksResponse, CoordinatorTrustRequest, @@ -1504,24 +1503,6 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [ error_codes=[400, 403, 404, 503], tags=["Coordinator"], ), - EndpointSpec( - "/v1/api/workstreams/{ws_id}/stop_cascade", - "POST", - "Cancel the coordinator and every direct child", - description=( - "Cancels the coordinator's in-flight generation AND dispatches " - "``cancel_workstream`` through the routing proxy for every " - "direct child in the in-memory registry. Grandchildren are " - "not touched directly — they sit behind their parent's cancel, " - "which propagates via the child's SSE stream. Returns the " - "per-child disposition (``cancelled`` / ``failed``) so the UI " - "can show which children responded. Writes " - "``coordinator.stopped_cascade`` with the two lists." - ), - response_model=CoordinatorStopCascadeResponse, - error_codes=[400, 403, 404, 503], - tags=["Coordinator"], - ), EndpointSpec( "/v1/api/workstreams/{ws_id}/close_all_children", "POST", @@ -1529,10 +1510,10 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [ description=( "Reads the in-memory child registry and dispatches " "``close_workstream`` via the routing proxy for every direct " - "child under a bounded (16-concurrency) semaphore. Unlike " - "``stop_cascade`` this does not touch grandchildren — the " - "model-facing tool asks for a bounded teardown of its own " - "fan-out. Returns ``{closed, failed, skipped}`` where " + "child under a bounded (16-concurrency) semaphore. Soft-close " + "only — it does not touch grandchildren (the model-facing tool " + "asks for a bounded teardown of its own direct fan-out). " + "Returns ``{closed, failed, skipped}`` where " "``skipped`` distinguishes already-gone (404) from dispatch-" "broken (``failed``). The optional ``reason`` propagates to " "every closed child's audit + workstream_config. Writes " @@ -1618,7 +1599,6 @@ _ALL_MODELS: list[type[BaseModel]] = [ CoordinatorRestrictResponse, CoordinatorSendRequest, CoordinatorSendResponse, - CoordinatorStopCascadeResponse, CoordinatorTaskInfo, CoordinatorTasksResponse, CoordinatorTrustRequest, diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index f418771a..69760187 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -440,9 +440,9 @@ class CoordinatorAdapter: def children_snapshot(self, coord_ws_id: str) -> list[str]: """Return a snapshot of the coordinator's direct child ws_ids. - Used by ``stop_cascade`` to iterate children without holding - the registry lock during the per-child HTTP dispatch. A - mutation racing with the snapshot (child spawned mid-cascade) + Used by the cancel cascade and ``close_all_children`` to iterate + children without holding the registry lock during the per-child + HTTP dispatch. A mutation racing with the snapshot (child spawned mid-cascade) either lands before (cancelled) or after (out of scope for this batch) — both safe. Returns an empty list for unknown coordinators. diff --git a/turnstone/console/coordinator_client.py b/turnstone/console/coordinator_client.py index a567536e..cd4c1965 100644 --- a/turnstone/console/coordinator_client.py +++ b/turnstone/console/coordinator_client.py @@ -818,7 +818,7 @@ class CoordinatorClient: def close_all_children(self, reason: str = "") -> dict[str, Any]: """Soft-close every direct child of this coordinator (console-side fan-out). - Returns ``{closed, failed, skipped}`` — mirrors ``stop_cascade``. + Returns ``{closed, failed, skipped}``. The console does the Semaphore-bounded gather so the model-side tool call stays a single HTTP round-trip regardless of fan-out size. No tenant guard here: ownership is enforced on the diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 22ee964e..e0578a6c 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -3115,7 +3115,7 @@ def _require_admin_coordinator( ) -> JSONResponse | None: """Gate a coordinator endpoint on the ``admin.coordinator`` permission. - Destructive endpoints (/restrict, /stop_cascade) pass + Destructive endpoints (/restrict, /close_all_children) pass ``allow_service_bypass=False`` so a service-scoped caller whose ``user_id`` matches the coord owner still needs an explicit grant. """ @@ -3869,8 +3869,8 @@ async def coordinator_metrics(request: Request) -> JSONResponse: _RESTRICT_MAX_TOOLS = 256 _RESTRICT_MAX_TOOL_NAME_LEN = 128 -# Bounded concurrency on bulk coordinator fan-out (stop_cascade, -# close_all_children). Upstream coord_client calls have a 30s timeout; +# Bounded concurrency on bulk coordinator fan-out (close_all_children, +# coordinator-cancel cascade). Upstream coord_client calls have a 30s timeout; # a 100-child cascade at this cap finishes in ~200s worst case, # comfortably inside typical 300s proxy limits. _COORD_FANOUT_MAX_CONCURRENCY = 16 @@ -3926,8 +3926,8 @@ async def _fanout_on_children( # silently no-op'd on placeholders; this keeps cascade # behaviour parity with the pre-lift outcome. # NOTE: this branch is reachable from the cancel-cascade - # caller (``stop_cascade``) but unreachable from the - # close-cascade caller (``close_all_children``); the + # caller (``_cascade_cancel_to_children``) but unreachable + # from the close-cascade caller (``close_all_children``); the # close handler at ``session_routes.py:852-854`` 404s # for both missing and already-closed-evicted rows and # never emits a 400 "No session". Kept as shared code @@ -4132,51 +4132,45 @@ async def coordinator_restrict(request: Request) -> JSONResponse: return JSONResponse({"status": "ok", "revoked_tools": sorted(after)}) -async def coordinator_stop_cascade(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/{ws_id}/stop_cascade — cancel the subtree.""" - resolved = await _resolve_coord_session(request, allow_service_bypass=False) - if isinstance(resolved, JSONResponse): - return resolved - session, storage, user_id, ws_id = resolved +async def _cascade_cancel_to_children(request: Request, ws_id: str, ws: Any) -> None: + """Fan a coordinator's cancel out to its direct children. - coord_mgr, err503 = _require_coord_mgr(request) - if err503 is not None: - return err503 # pragma: no cover — _resolve_coord_session already gated this + Wired into the coordinator cancel handler via ``post_cancel`` so that + cancelling a coordinator propagates down its spawned subtree — the + cancellation appendix's "cancel flows down the subtree." The + coordinator's own session is already cancelled by ``make_cancel_handler`` + before this runs; here we only dispatch the per-child cancels via the + same ``children_snapshot`` + ``_fanout_on_children`` path the + ``close_all_children`` bulk verb uses. + + Trigger, not drain: each child cancel is itself cooperative and + fire-and-forget (it sets the child's cancel flag and returns). We do + not block on the subtree reaching a terminal state — that barrier is + deferred. Children are one level deep today (a coordinator spawns only + interactive leaves), so the direct-child fan-out is the whole subtree. + """ coord_adapter = getattr(request.app.state, "coord_adapter", None) - - child_ids = list(coord_adapter.children_snapshot(ws_id)) if coord_adapter is not None else [] - coord_mgr.cancel(ws_id) - - coord_client: Any = getattr(session, "_coord_client", None) - # ``action`` is only called when coord_client is live — the helper - # short-circuits on None before invoking it. - cancelled, failed, skipped = await _fanout_on_children( + if coord_adapter is None: + return + child_ids = list(coord_adapter.children_snapshot(ws_id)) + if not child_ids: + return + session = getattr(ws, "session", None) + coord_client: Any = getattr(session, "_coord_client", None) if session is not None else None + if coord_client is None: + return + ok, failed, skipped = await _fanout_on_children( child_ids, coord_client, lambda cid: coord_client.cancel(cid), - log_tag="coordinator_stop_cascade", + log_tag="coordinator_cancel_cascade", ) - - await _emit_coord_audit( - storage, - user_id, - "coordinator.stopped_cascade", - ws_id, - { - "src": "coordinator", - "cancelled": cancelled, - "failed": failed, - "skipped": skipped, - }, - request.client.host if request.client else "", - ) - return JSONResponse( - { - "status": "ok", - "cancelled": cancelled, - "failed": failed, - "skipped": skipped, - } + log.info( + "coordinator.cancel_cascaded ws=%s cancelled=%d failed=%d skipped=%d", + ws_id[:8], + len(ok), + len(failed), + len(skipped), ) @@ -4186,12 +4180,10 @@ _CLOSE_ALL_CHILDREN_MAX_REASON_LEN = 512 async def coordinator_close_all_children(request: Request) -> JSONResponse: """POST /v1/api/workstreams/{ws_id}/close_all_children — soft-close the direct children. - Near-twin of ``coordinator_stop_cascade`` — both fan out over - ``children_snapshot`` via ``_fanout_on_children``. Returns - ``{closed, failed, skipped}``. Unlike ``stop_cascade``, this does - NOT recurse into grandchildren (the coordinator's model tool asks - for a bounded teardown of its own fan-out; operator-level cascade - stays behind ``stop_cascade``). + Fans out over ``children_snapshot`` via ``_fanout_on_children`` and + returns ``{closed, failed, skipped}``. Soft-close only: this does NOT + recurse into grandchildren (the coordinator's model tool asks for a + bounded teardown of its own direct fan-out). """ resolved = await _resolve_coord_session(request, allow_service_bypass=False) if isinstance(resolved, JSONResponse): @@ -4953,8 +4945,8 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: app.state.dashboard_cache = _NodeDashboardCache() # Dedicated small executor for governance audit writes. Without # this, audit dispatches share the default thread pool with - # ``coord_client.cancel`` calls from ``stop_cascade`` and any - # other ``asyncio.to_thread`` caller — a burst on one path can + # ``coord_client.cancel`` calls from the coordinator-cancel cascade + # and any other ``asyncio.to_thread`` caller — a burst on one path can # starve the other. 4 workers is ample headroom for # admin-driven audit traffic. audit_exec = ThreadPoolExecutor(max_workers=4, thread_name_prefix="coord-audit") @@ -13030,6 +13022,10 @@ def create_app( cancel=make_cancel_handler( # lifted: shared body coord_endpoint_config, audit_emit=_audit_cancel_coordinator, + # Auto-propagate: cancelling a coordinator fans the cancel + # out to its spawned children (see + # ``_cascade_cancel_to_children``). + post_cancel=_cascade_cancel_to_children, ), rewind=make_rewind_handler( # lifted: shared body (#549) coord_endpoint_config, @@ -13059,7 +13055,6 @@ def create_app( metrics=coordinator_metrics, trust=coordinator_trust, restrict=coordinator_restrict, - stop_cascade=coordinator_stop_cascade, close_all_children=coordinator_close_all_children, ), ) diff --git a/turnstone/core/audit.py b/turnstone/core/audit.py index fdd00759..fa41b6e6 100644 --- a/turnstone/core/audit.py +++ b/turnstone/core/audit.py @@ -11,8 +11,7 @@ Action-name conventions (non-exhaustive — grep ``.cancel``) plus governance sub-prefixes (``coordinator.trust.toggled``, ``coordinator.send.auto_approved``, - ``coordinator.restricted``, - ``coordinator.stopped_cascade``). + ``coordinator.restricted``). route.* multi-node routing proxy hops (``route.workstream.create`` / ``.send`` / diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 4727c2ff..030cfc9f 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -4680,17 +4680,30 @@ class ChatSession: if msg.role is Role.TOOL: answered_ids.add(msg.tool_call_id or "") + # An orphaned tool_call had no result when cancel landed. We can't + # tell here whether it was mid-execution (outcome unobserved) or + # never started, so we mark it UNKNOWN rather than let the bare + # reason read as "it didn't happen" — which invites a re-send as + # readily as a dropped record causes an orphan (cancellation + # appendix, HYPOTHESIS.md: unknown, never none). ``is_error`` stays + # True: it is genuinely not a successful result, and both the SSE + # batch completion and the existing UI rendering key on it. + detail = ( + f"{reason} 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." + ) # Synthesize results for unanswered tool_calls for tc in self.messages[assistant_idx].tool_calls: tc_id = tc.id func_name = tc.name if tc_id and tc_id not in answered_ids: - self.messages.append(Turn.tool(tc_id, reason, is_error=True)) + self.messages.append(Turn.tool(tc_id, detail, is_error=True)) self._msg_tokens.append(1) save_message( self._ws_id, "tool", - reason, + detail, func_name, tool_call_id=tc_id, event_id=self._ui_event_id(), @@ -4703,7 +4716,7 @@ class ChatSession: # Defensive: we're already on a cancel/error path, so # a UI hook failure must not compound the problem. try: - self.ui.on_tool_result(tc_id, func_name, reason, is_error=True) + self.ui.on_tool_result(tc_id, func_name, detail, is_error=True) except Exception: log.debug( "session.synthesize_cancelled.ui_emit_failed ws=%s", @@ -11833,12 +11846,86 @@ class ChatSession: auto_tools=TASK_AUTO_TOOLS, agent_alias=item.get("model_override"), ) - except (KeyboardInterrupt, GenerationCancelled): + except GenerationCancelled: + # Fold back an honest disposition built from the agent's own + # ledger. ``agent_messages`` is mutated in place by + # ``_run_agent`` (it appends every assistant turn and tool + # result), so at this catch point it holds the full record of + # what the sub-agent did before cancel — including a partial + # result from a tool that was SIGKILL'd mid-flight. The old + # bare "(task interrupted by user)" string discarded all of + # that and fabricated an *outcome*: downstream it read as + # "nothing happened" and invited a re-dispatch / double-send. + # See the cancellation appendix in HYPOTHESIS.md ("ρ may + # fabricate the acknowledgment but must not fabricate the + # outcome … unknown, never none"). + return call_id, self._cancelled_agent_disposition(agent_messages, "task") + except KeyboardInterrupt: + # CLI Ctrl-C: keep the terse string and let the outer loop own + # propagation (unchanged behavior). return call_id, "(task interrupted by user)" except Exception as e: self.ui.on_info(f"[task error] {e}") return call_id, f"Task error: {e}" + def _cancelled_agent_disposition(self, agent_messages: list[dict[str, Any]], label: str) -> str: + """Build an honest, deterministic disposition for a cancelled sub-agent. + + A cancelled agent must not report a fabricated *outcome*. The bare + "(interrupted)" string read downstream as *the task did not happen* + — which causes a double-send as readily as a dropped record causes + an orphan (cancellation appendix, HYPOTHESIS.md). Instead we fold + back the agent's actual ledger: which tool actions completed, which + never started, and an explicit ``UNKNOWN`` flag on the single most + recent action — the one that was crossing the gate / executing in + the tool when cancel landed, whose side effect may or may not have + landed and whose result was never observed. + + Pure string assembly over the in-memory ``agent_messages`` — no + model call, because we are on the cancel path and the gate is + closed. The owner (parent / coordinator) reads this to decide what, + if anything, to compensate. + """ + answered: set[str] = set() + for m in agent_messages: + if m.get("role") == "tool" and m.get("tool_call_id"): + answered.add(str(m["tool_call_id"])) + # Ordered (name, was-answered) for every tool_call the agent issued. + issued: list[tuple[str, bool]] = [] + for m in agent_messages: + if m.get("role") != "assistant": + continue + for tc in m.get("tool_calls") or []: + name = ((tc.get("function") or {}).get("name") or "tool").strip() + issued.append((name, str(tc.get("id") or "") in answered)) + if not issued: + return f"({label} cancelled by user before any action — no side effects)" + + def _summ(names: list[str]) -> str: + counts: dict[str, int] = {} + for n in names: + counts[n] = counts.get(n, 0) + 1 + return ", ".join(f"{n}×{c}" if c > 1 else n for n, c in counts.items()) + + # The last action issued is the boundary: it was in flight (or about + # to run) when cancel was observed, so its outcome is unobserved. + # Mark it UNKNOWN rather than implying it did or did not happen. + boundary = issued[-1][0] + completed = [n for n, ans in issued[:-1] if ans] + not_run = [n for n, ans in issued[:-1] if not ans] + parts = [f"({label} cancelled by user before completion)"] + if completed: + parts.append("Completed before cancel: " + _summ(completed) + ".") + parts.append( + f"In flight at cancel: {boundary} — outcome UNKNOWN. It may have " + "completed, partially executed, or caused side effects before the " + "agent was stopped; its result was never observed. Do not assume " + "it did or did not happen — reconcile before re-running." + ) + if not_run: + parts.append("Not started (cancelled first): " + _summ(not_run) + ".") + return "\n".join(parts) + def _audit_memory_event( self, action: str, diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 3237d2fe..66502be5 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -16,9 +16,9 @@ Three registrar functions: All handlers in :class:`SharedSessionVerbHandlers` are optional; ``None`` skips the route, so one bundle describes either kind. - :func:`register_coord_verbs` — coord-only verbs (``trust``, - ``restrict``, ``stop_cascade``, ``close_all_children``, - ``children``, ``tasks``, ``metrics``) that read or mutate state - that doesn't exist on interactive workstreams. + ``restrict``, ``close_all_children``, ``children``, ``tasks``, + ``metrics``) that read or mutate state that doesn't exist on + interactive workstreams. Some verbs in :class:`SharedSessionVerbHandlers` ship as factory- returned closures (e.g. :func:`make_approve_handler`, @@ -546,7 +546,6 @@ class CoordOnlyVerbHandlers: metrics: Handler # GET {prefix}/{ws_id}/metrics trust: Handler # POST {prefix}/{ws_id}/trust restrict: Handler # POST {prefix}/{ws_id}/restrict - stop_cascade: Handler # POST {prefix}/{ws_id}/stop_cascade close_all_children: Handler # POST {prefix}/{ws_id}/close_all_children @@ -672,7 +671,6 @@ def register_coord_verbs( routes.append(Route(f"{p}/{{ws_id}}/metrics", handlers.metrics, methods=["GET"])) routes.append(Route(f"{p}/{{ws_id}}/trust", handlers.trust, methods=["POST"])) routes.append(Route(f"{p}/{{ws_id}}/restrict", handlers.restrict, methods=["POST"])) - routes.append(Route(f"{p}/{{ws_id}}/stop_cascade", handlers.stop_cascade, methods=["POST"])) routes.append( Route( f"{p}/{{ws_id}}/close_all_children", @@ -1092,11 +1090,20 @@ CancelAuditEmitter = Callable[ None, ] +# Optional async step run AFTER the workstream's own cancel sequence +# (session.cancel + approval resolution + force/SSE + audit), receiving +# ``(request, ws_id, ws)``. Coordinator wires this to fan the cancel out +# to its spawned children (auto-propagation down the subtree); interactive +# wires ``None`` and the handler is unchanged. Errors are logged, never +# surfaced — a cascade failure must not fail the owner's own cancel. +PostCancelHook = Callable[["Request", str, "Workstream"], Awaitable[None]] + def make_cancel_handler( cfg: SessionEndpointConfig, *, audit_emit: CancelAuditEmitter | None = None, + post_cancel: PostCancelHook | None = None, ) -> Handler: """Lifted body for ``POST {prefix}/{ws_id}/cancel``. @@ -1296,6 +1303,21 @@ def make_cancel_handler( exc_info=True, ) + # Propagate the cancel down the subtree (coordinator only). Runs + # after the owner's own cancel is fully recorded so a cascade error + # can't strand the owner half-cancelled. Cooperative/fire-and-forget + # per child — we dispatch the cancels, we don't block on the subtree + # draining (that barrier is deferred). + if post_cancel is not None: + try: + await post_cancel(request, ws_id, ws) + except Exception: + log.warning( + "ws.cancel.cascade_failed ws=%s", + ws_id[:8] if ws_id else "", + exc_info=True, + ) + return JSONResponse({"status": "ok", "dropped": dropped}) return cancel diff --git a/turnstone/tools/close_all_children.json b/turnstone/tools/close_all_children.json index 9987a984..3ebc3519 100644 --- a/turnstone/tools/close_all_children.json +++ b/turnstone/tools/close_all_children.json @@ -1,6 +1,6 @@ { "name": "close_all_children", - "description": "Soft-close every direct child of this coordinator. Fans out close requests with bounded concurrency. Returns `{status, closed, failed, skipped}`. `skipped` holds 404s — covers both already-hard-deleted children (row purged) AND already-closed children whose in-memory session was evicted (row remains marked `state=closed`); the wire shape doesn't distinguish, so treat `skipped` as \"nothing to close\" rather than \"definitely deleted\". `failed` holds dispatch errors worth retrying. Indirect descendants (children of children) are NOT closed — run `stop_cascade` from the operator UI for a full-subtree teardown. The optional `reason` is propagated to every closed child's audit + workstream_config for postmortem (max 512 chars; over the cap returns a tool error).", + "description": "Soft-close every direct child of this coordinator. Fans out close requests with bounded concurrency. Returns `{status, closed, failed, skipped}`. `skipped` holds 404s — covers both already-hard-deleted children (row purged) AND already-closed children whose in-memory session was evicted (row remains marked `state=closed`); the wire shape doesn't distinguish, so treat `skipped` as \"nothing to close\" rather than \"definitely deleted\". `failed` holds dispatch errors worth retrying. The optional `reason` is propagated to every closed child's audit + workstream_config for postmortem (max 512 chars; over the cap returns a tool error).", "parameters": { "type": "object", "properties": {