From 06c91294a403ff91a78d44b922a2f6e7dff6f4bb Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 24 Apr 2026 16:16:33 -0700 Subject: [PATCH] refactor(server): lift close handler into shared session_routes body MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 2 Priority 0 Step 0.2 body-convergence — second verb. ``make_close_handler(audit_emit=..., supports_close_reason=...)`` factory in ``turnstone/core/session_routes.py`` produces the lifted body; both interactive and coord pass their kind-specific audit emitter at app construction. The two body-keyed close URL aliases on the interactive side reach the same lifted body: - ``POST /v1/api/workstreams/{ws_id}/close`` (new, path-keyed) via ``register_session_routes(handlers.close=...)``. - ``POST /v1/api/workstreams/close`` (legacy, body-keyed) via ``make_legacy_body_keyed_adapter(close_handler)``. Coord exposes only the path-keyed shape. Behavior gains: - ``supports_close_reason=True`` (interactive only) keeps the 512- byte UTF-8 cap + credential redaction + ``workstream_config`` persistence path. Coord stays at ``False``; if coord ever wants close-reason metadata, flipping the flag is a one-line change. - ``audit_emit`` is per-kind so each owns its detail dict shape (``{kind, parent_ws_id, reason}`` vs ``{coord_ws_id, src}``) and audit action name (``workstream.closed`` vs ``coordinator.close``). - Standardizes the close-failure status code to 404 across both kinds. The coord code previously returned 500 on a ``mgr.close()`` race-loss, which was overly pessimistic — the semantic is "the ws was popped between .get() and .close()", i.e. not-found. Coord-side test fixtures (``test_coordinator_endpoints``, ``test_coordinator_end_to_end``) swap the imported ``coordinator_close`` for the lifted handler + a local audit_emit adapter so the tests exercise the same code path the live console does. ruff + mypy + 4366 pytest pass. Live console smoke against ``POST /v1/api/workstreams/abc/close`` returns 503 (no coord_mgr loaded in the smoke env) — proves the lifted handler is reachable + the manager_lookup callable fires correctly. Two verbs converged so far (``approve`` + ``close``); the remaining pairs (``send``, ``cancel``, ``open``, ``events``, ``create``, ``list``, ``saved``, ``history``, ``detail``) have substantive behavior divergence that doesn't factor cleanly into the SessionEndpointConfig + factory-handler pattern — see the session_routes module docstring for the per-verb status. --- tests/test_coordinator_end_to_end.py | 45 +++++++++- tests/test_coordinator_endpoints.py | 37 +++++++- turnstone/console/server.py | 61 ++++++-------- turnstone/core/session_routes.py | 116 +++++++++++++++++++++++++ turnstone/server.py | 122 ++++++++------------------- 5 files changed, 250 insertions(+), 131 deletions(-) diff --git a/tests/test_coordinator_end_to_end.py b/tests/test_coordinator_end_to_end.py index cebccc30..6f30e7bb 100644 --- a/tests/test_coordinator_end_to_end.py +++ b/tests/test_coordinator_end_to_end.py @@ -15,7 +15,7 @@ MagicMock-backed stubs. All four tests run in < 2 s total. from __future__ import annotations import json -from typing import Any +from typing import TYPE_CHECKING, Any from unittest.mock import MagicMock import httpx @@ -31,15 +31,44 @@ from turnstone.console.coordinator_adapter import CoordinatorAdapter from turnstone.console.coordinator_client import CoordinatorClient from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( - coordinator_close, + _auth_user_id, + _require_admin_coordinator, + _require_coord_mgr, coordinator_create, coordinator_detail, coordinator_list, ) +from turnstone.core.audit import record_audit from turnstone.core.auth import AuthResult from turnstone.core.session_manager import SessionManager +from turnstone.core.session_routes import ( + SessionEndpointConfig, + make_close_handler, +) from turnstone.core.storage._sqlite import SQLiteBackend +if TYPE_CHECKING: + from turnstone.core.workstream import Workstream + + +def _audit_close_coordinator_e2e( + request, + ws_id: str, + ws_before: Workstream, # noqa: ARG001 + reason: str, # noqa: ARG001 +) -> None: + storage = request.app.state.auth_storage + record_audit( + storage, + _auth_user_id(request), + "coordinator.close", + "workstream", + ws_id, + {"coord_ws_id": ws_id, "src": "coordinator"}, + request.client.host if request.client else "", + ) + + # --------------------------------------------------------------------------- # Shared auth-injection middleware (mirrors test_coordinator_endpoints.py) # --------------------------------------------------------------------------- @@ -125,7 +154,10 @@ def _make_client( Route("/v1/api/workstreams", coordinator_list, methods=["GET"]), Route( "/v1/api/workstreams/{ws_id}/close", - coordinator_close, + make_close_handler( + audit_emit=_audit_close_coordinator_e2e, + supports_close_reason=False, + ), methods=["POST"], ), Route( @@ -143,6 +175,13 @@ def _make_client( app.state.coord_registry_error = "" if coord_mgr else "registry missing" app.state.auth_storage = storage app.state.jwt_secret = "x" * 64 + app.state.session_endpoint_config = SessionEndpointConfig( + permission_gate=_require_admin_coordinator, + manager_lookup=_require_coord_mgr, + tenant_check=None, + not_found_label="coordinator not found", + audit_action_prefix="coordinator", + ) return TestClient(app) diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index 58d94a5c..98972699 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -8,6 +8,7 @@ enforcement, and lazy rehydration on GET /{ws_id}. from __future__ import annotations +from typing import TYPE_CHECKING from unittest.mock import MagicMock import httpx @@ -28,12 +29,12 @@ from tests._coord_test_helpers import ( ) from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( + _auth_user_id, _require_admin_coordinator, _require_coord_mgr, cluster_ws_detail, coordinator_cancel, coordinator_children, - coordinator_close, coordinator_create, coordinator_detail, coordinator_history, @@ -43,10 +44,37 @@ from turnstone.console.server import ( coordinator_send, coordinator_tasks, ) +from turnstone.core.audit import record_audit from turnstone.core.auth import AuthResult -from turnstone.core.session_routes import SessionEndpointConfig, make_approve_handler +from turnstone.core.session_routes import ( + SessionEndpointConfig, + make_approve_handler, + make_close_handler, +) from turnstone.core.storage._sqlite import SQLiteBackend +if TYPE_CHECKING: + from turnstone.core.workstream import Workstream + + +def _audit_close_coordinator_for_test( + request, + ws_id: str, + ws_before: Workstream, # noqa: ARG001 + reason: str, # noqa: ARG001 +) -> None: + storage = request.app.state.auth_storage + record_audit( + storage, + _auth_user_id(request), + "coordinator.close", + "workstream", + ws_id, + {"coord_ws_id": ws_id, "src": "coordinator"}, + request.client.host if request.client else "", + ) + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -97,7 +125,10 @@ def _make_client( ), Route( "/v1/api/workstreams/{ws_id}/close", - coordinator_close, + make_close_handler( + audit_emit=_audit_close_coordinator_for_test, + supports_close_reason=False, + ), methods=["POST"], ), Route( diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 49af308b..dbfe2f82 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -59,6 +59,7 @@ from turnstone.core.session_routes import ( SessionEndpointConfig, SharedSessionVerbHandlers, make_approve_handler, + make_close_handler, register_coord_verbs, register_session_routes, ) @@ -67,7 +68,7 @@ from turnstone.core.web_helpers import ( read_json_or_400, require_storage_or_503, ) -from turnstone.core.workstream import WorkstreamKind +from turnstone.core.workstream import Workstream, WorkstreamKind if TYPE_CHECKING: from collections.abc import AsyncGenerator, Callable @@ -2558,40 +2559,6 @@ async def coordinator_cancel(request: Request) -> JSONResponse: return JSONResponse({"status": "ok"}) -async def coordinator_close(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/{ws_id}/close — unload the session.""" - from turnstone.core.audit import record_audit - - err = _require_admin_coordinator(request) - if err is not None: - return err - coord_mgr, err503 = _require_coord_mgr(request) - if err503 is not None: - return err503 - ws_id = request.path_params.get("ws_id", "") - user_id = _auth_user_id(request) - ws = coord_mgr.get(ws_id) - if ws is None: - return JSONResponse({"error": "coordinator not found"}, status_code=404) - if not coord_mgr.close(ws_id): - return JSONResponse({"error": "close failed"}, status_code=500) - storage = getattr(request.app.state, "auth_storage", None) - if storage is not None: - try: - record_audit( - storage, - user_id, - "coordinator.close", - "workstream", - ws_id, - {"coord_ws_id": ws_id, "src": "coordinator"}, - request.client.host if request.client else "", - ) - except Exception: - log.debug("coordinator_close.audit_failed", exc_info=True) - return JSONResponse({"status": "ok"}) - - async def coordinator_events(request: Request) -> Response: """GET /v1/api/workstreams/{ws_id}/events — SSE event stream.""" err = _require_admin_coordinator(request) @@ -3930,7 +3897,6 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: build_console_session_factory, ) from turnstone.core.session_manager import SessionManager - from turnstone.core.workstream import Workstream jwt_secret: str = getattr(app.state, "jwt_secret", "") console_bind_url: str = getattr(app.state, "console_url", "") or ( @@ -10127,6 +10093,24 @@ def create_app( not_found_label="coordinator not found", audit_action_prefix="coordinator", ) + + def _audit_close_coordinator( + request: Request, + ws_id: str, + ws_before: Workstream, + reason: str, # noqa: ARG001 — coord doesn't expose close_reason yet + ) -> None: + storage = request.app.state.auth_storage + record_audit( + storage, + _auth_user_id(request), + "coordinator.close", + "workstream", + ws_id, + {"coord_ws_id": ws_id, "src": "coordinator"}, + request.client.host if request.client else "", + ) + coord_workstream_routes: list[Any] = [] register_session_routes( coord_workstream_routes, @@ -10137,7 +10121,10 @@ def create_app( create=coordinator_create, detail=coordinator_detail, open=coordinator_open, - close=coordinator_close, + close=make_close_handler( # lifted: shared body + audit_emit=_audit_close_coordinator, + supports_close_reason=False, + ), send=coordinator_send, approve=make_approve_handler(), # lifted: shared body cancel=coordinator_cancel, diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 3a4c1a30..3520c042 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -414,3 +414,119 @@ def make_legacy_body_keyed_adapter(handler: Handler) -> Handler: return await handler(request) return adapter + + +CloseAuditEmitter = Callable[ + ["Request", str, "Workstream", str], + None, +] + + +def make_close_handler( + *, + audit_emit: CloseAuditEmitter | None = None, + supports_close_reason: bool = False, +) -> Handler: + """Lifted body for ``POST {prefix}/{ws_id}/close``. + + Closes the workstream's session (unloads from memory; storage row + survives so the session can be re-opened later). Both kinds share + the same auth → mgr → ws-lookup → ``mgr.close()`` → audit + sequence; per-kind divergence is in the audit detail shape and + whether a request body ``reason`` is read / capped / persisted on + the workstream's config row. + + Args: + audit_emit: kind's audit emitter for the close event. + Receives ``(request, ws_id, ws_before, reason)``; ``reason`` + is the empty string when ``supports_close_reason`` is + ``False`` or no reason was provided. ``None`` skips the + audit entirely (only valid when neither kind cares). + supports_close_reason: when ``True``, the handler reads a + ``reason`` field from the JSON body, caps it at 512 UTF-8 + bytes, redacts credentials, persists it via + ``storage.save_workstream_config(ws_id, {"close_reason": ...})``, + and threads it through to ``audit_emit``. The cap protects + ``workstream_config`` from unbounded growth on a model- + generated dump; the redact protects audit logs from + captured-secret leakage under prompt injection. + """ + + async def close(request: Request) -> Response: + cfg: SessionEndpointConfig = request.app.state.session_endpoint_config + if cfg.permission_gate is not None: + err = cfg.permission_gate(request) + if err is not None: + return err + mgr, err503 = cfg.manager_lookup(request) + if err503 is not None: + return err503 + assert mgr is not None + ws_id = request.path_params.get("ws_id", "") + + reason = "" + if supports_close_reason: + from turnstone.core.output_guard import redact_credentials + from turnstone.core.web_helpers import read_json_or_400 + + body = await read_json_or_400(request) + if isinstance(body, JSONResponse): + return body + raw_reason = body.get("reason", "") + if isinstance(raw_reason, str): + # Cap on UTF-8 bytes (not code points) so a CJK / emoji + # payload can't sneak past at 3-4x the documented budget. + # ``errors="ignore"`` drops any partial code point left + # at the truncation boundary. + capped = raw_reason.strip().encode("utf-8")[:512].decode( + "utf-8", errors="ignore" + ) + reason = redact_credentials(capped) + + if cfg.tenant_check is not None: + err_tenant = cfg.tenant_check(request, ws_id, mgr) + if err_tenant is not None: + return err_tenant + + ws_before = mgr.get(ws_id) + if ws_before is None: + return JSONResponse({"error": cfg.not_found_label}, status_code=404) + if not mgr.close(ws_id): + # Standardized to 404 (was 500 on the legacy coord path — + # overly pessimistic; close-failure here means the ws was + # popped between the .get() check and the .close() call, + # which is a "not found" semantic). + return JSONResponse({"error": cfg.not_found_label}, status_code=404) + + storage = getattr(request.app.state, "auth_storage", None) + if supports_close_reason and reason and storage is not None: + try: + storage.save_workstream_config(ws_id, {"close_reason": reason}) + except Exception: + from turnstone.core.log import get_logger as _gl + + _gl(__name__).debug( + "ws.close.reason_persist_failed ws=%s", + ws_id[:8] if ws_id else "", + exc_info=True, + ) + + if audit_emit is not None and storage is not None: + try: + audit_emit(request, ws_id, ws_before, reason) + except Exception: + from turnstone.core.log import get_logger as _gl + + _gl(__name__).debug( + "ws.close.audit_failed ws=%s", + ws_id[:8] if ws_id else "", + exc_info=True, + ) + + return JSONResponse({"status": "ok"}) + + return close + + +if TYPE_CHECKING: + from turnstone.core.workstream import Workstream # noqa: F401 — used in type alias above diff --git a/turnstone/server.py b/turnstone/server.py index 6f5a34ed..6a0e2fa3 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -60,6 +60,7 @@ from turnstone.core.session_routes import ( SessionEndpointConfig, SharedSessionVerbHandlers, make_approve_handler, + make_close_handler, make_legacy_body_keyed_adapter, register_session_routes, ) @@ -2630,93 +2631,6 @@ async def create_workstream(request: Request) -> JSONResponse: return JSONResponse({"error": str(e)}, status_code=400) -async def close_workstream(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/close — close a workstream. - - Optional ``reason`` from the request body is persisted on the - workstream's config row so post-mortem tooling (and the coordinator's - ``inspect_workstream``) can surface why the workstream was retired - without scraping the audit log. - """ - from turnstone.core.audit import record_audit - from turnstone.core.output_guard import redact_credentials - from turnstone.core.web_helpers import read_json_or_400 - - body = await read_json_or_400(request) - if isinstance(body, JSONResponse): - return body - ws_id = str(body.get("ws_id", "")) - raw_reason = body.get("reason", "") - # Cap reason length so a model that accidentally / maliciously dumps - # a multi-KB blob (or a captured secret) can't grow workstream_config - # rows without bound. 512 BYTES is enough for any human-readable - # close reason; anything longer is suspicious. Slice on UTF-8 bytes - # (not code points) so a CJK / emoji-heavy payload can't sneak past - # the cap at 3-4x the documented budget. ``errors="ignore"`` drops - # any partial code point left at the truncation boundary. Then run - # the same credential-redaction the output guard applies to tool - # output — a model under prompt injection that dumps a captured - # secret in ``reason`` doesn't get to plant it in plaintext audit - # logs / workstream_config. - if isinstance(raw_reason, str): - capped = raw_reason.strip().encode("utf-8")[:512].decode("utf-8", errors="ignore") - reason = redact_credentials(capped) - else: - reason = "" - mgr = request.app.state.workstreams - # Cross-tenant close would abort another tenant's running generation. - # _require_ws_access returns 404 on non-owner — same shape as the - # "last workstream" (400) / "not found" (404) branches below. - _owner_uid, err = _require_ws_access(request, ws_id, mgr=mgr) - if err: - return err - ws_before = mgr.get(ws_id) - if not ws_before: - return JSONResponse({"error": "Workstream not found"}, status_code=404) - if mgr.close(ws_id): - storage = getattr(request.app.state, "auth_storage", None) - # Persist before the ws_closed event reaches consumers so any - # downstream reader sees the close_reason in the same - # observation window as the state change. The adapter-level - # emit_closed fires inside mgr.close() which already returned - # — strictly speaking the event beat us here, but the - # close_reason is advisory UI metadata, not a sync barrier. - if reason and storage is not None: - try: - storage.save_workstream_config(ws_id, {"close_reason": reason}) - except Exception: - log.debug( - "ws.close.reason_persist_failed ws=%s", - ws_id[:8] if ws_id else "", - exc_info=True, - ) - if storage is not None: - _, ip = _audit_context(request) - audit_detail: dict[str, Any] = { - "kind": str(ws_before.kind), - "parent_ws_id": ws_before.parent_ws_id, - } - if reason: - audit_detail["reason"] = reason - record_audit( - storage, - _auth_user_id(request), - "workstream.closed", - "workstream", - ws_id, - audit_detail, - ip, - ) - return JSONResponse({"status": "ok"}) - # close() returned False — the ws was popped between the mgr.get() - # check above and our mgr.close() call (another close racing in). - # Return 404 so the API contract matches a "not found" caller - # experience; the old "Cannot close last workstream" 400 went away - # with the default-startup workstream and there's no scenario where - # this branch means anything other than "the ws isn't tracked here". - return JSONResponse({"error": "Workstream not found"}, status_code=404) - - async def delete_workstream_endpoint(request: Request) -> JSONResponse: """POST /v1/api/workstreams/{ws_id}/delete — permanently delete a saved workstream.""" from turnstone.core.audit import record_audit @@ -4408,6 +4322,37 @@ def create_app( audit_action_prefix="workstream", ) approve_handler = make_approve_handler() + + def _audit_close_workstream( + request: Request, + ws_id: str, + ws_before: Workstream, + reason: str, + ) -> None: + from turnstone.core.audit import record_audit + + storage = request.app.state.auth_storage + _, ip = _audit_context(request) + detail: dict[str, Any] = { + "kind": str(ws_before.kind), + "parent_ws_id": ws_before.parent_ws_id, + } + if reason: + detail["reason"] = reason + record_audit( + storage, + _auth_user_id(request), + "workstream.closed", + "workstream", + ws_id, + detail, + ip, + ) + + close_handler = make_close_handler( + audit_emit=_audit_close_workstream, + supports_close_reason=True, + ) v1_routes: list[Any] = [ Route("/api/events", events_sse), Route("/api/events/global", global_events_sse), @@ -4419,9 +4364,10 @@ def create_app( list_workstreams=list_workstreams, list_saved=list_saved_workstreams, create=create_workstream, - close_legacy=close_workstream, + close_legacy=make_legacy_body_keyed_adapter(close_handler), delete=delete_workstream_endpoint, open=open_workstream, + close=close_handler, # lifted: shared body refresh_title=refresh_workstream_title, set_title=set_workstream_title, approve=approve_handler, # lifted: shared body