mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
feat(cancel): honest cancellation dispositions + coordinator subtree propagation
A cancelled agent previously discarded its own ledger and reported a bare "(task interrupted by user)" — fabricating the *outcome* (read downstream as "nothing happened"), which invites a double-send as readily as a dropped record causes an orphan. Make the fold-back honest, and propagate an owner's cancel down the coordinator subtree. - task_agent (single + parallel): on cancel, fold back a deterministic disposition built from the agent's in-memory ledger — actions completed, the in-flight action flagged outcome-UNKNOWN, and not-started calls — instead of the opaque interrupted string. - coordinator cancel now auto-propagates to its direct children via a post_cancel hook on the shared cancel handler (cooperative fan-out; no blocking drain). - synthesized cancelled tool results now read outcome-UNKNOWN rather than implying the call never ran. - remove the now-redundant stop_cascade operator endpoint (handler, route, OpenAPI spec + schema, tests, docs); a coordinator cancel supersedes it.
This commit is contained in:
Patrick Buckley
+12
-25
@@ -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 `<bucket>` 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
|
||||
(`{<bucket>, 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
|
||||
|
||||
@@ -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, close_all_children
|
||||
|
||||
|
||||
These three endpoints let an operator steer a live coordinator session
|
||||
mid-flight. All four emit an audit event tagged
|
||||
`coordinator.<action>` via the dedicated audit executor so a cascade
|
||||
mid-flight. All three emit an audit event tagged
|
||||
`coordinator.<action>` via the dedicated audit executor so a cascade
|
||||
burst can't starve audit writes.
|
||||
|
||||
### `POST /trust` — auto-approve own-subtree sends
|
||||
@@ -294,28 +293,6 @@ idempotent — calling twice with overlapping lists converges to the
|
||||
opt in per session. Cap 256 tool names per request, 128 chars each.
|
||||
|
||||
### `POST /close_all_children` — soft-close the direct fan-out
|
||||
|
||||
```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
|
||||
POST /v1/api/workstreams/{ws_id}/close_all_children
|
||||
@@ -329,16 +306,16 @@ Response:
|
||||
```
|
||||
|
||||
Soft-close cascade bounded by a concurrency semaphore. The `reason`
|
||||
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`.
|
||||
|
||||
(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 `close_all_children`
|
||||
share the cascade-mutation shape and how it differs from the
|
||||
`spawn_batch` / `cluster/ws/live` shape.
|
||||
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
|
||||
```
|
||||
|
||||
`cancel` drops the coordinator's in-flight generation and, for a
|
||||
idle and open for a fresh `send`:
|
||||
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
|
||||
POST /v1/api/workstreams/{ws_id}/cancel
|
||||
@@ -373,9 +353,10 @@ POST /v1/api/workstreams/{ws_id}/close
|
||||
```
|
||||
|
||||
Soft-closes the session — state persists, children keep running
|
||||
`close_all_children` or `stop_cascade` first to wind them down), the
|
||||
worker thread exits, SSE streams send a final `stream_end` and
|
||||
disconnect. The row is reopenable via
|
||||
(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.
|
||||
idioms (`{results, denied, truncated}` vs
|
||||
`{<bucket>, failed, skipped}`) used by `cluster/ws/live`,
|
||||
`spawn_batch`, and `close_all_children`.
|
||||
- [architecture.md](architecture.md) — cluster-wide architecture
|
||||
- [architecture.md](architecture.md) — cluster-wide architecture
|
||||
including how coordinator sessions fit next to node-hosted
|
||||
interactive workstreams.
|
||||
- The live OpenAPI spec (`/openapi.json` on any console process)
|
||||
|
||||
@@ -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
|
||||
|
||||
+98
-2
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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"})),
|
||||
}
|
||||
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
+48
-53
@@ -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,
|
||||
),
|
||||
)
|
||||
|
||||
@@ -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`` /
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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": {
|
||||
|
||||
Reference in New Issue
Block a user