diff --git a/CHANGELOG.md b/CHANGELOG.md index 4cef2530..1ea0b0fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,16 +40,26 @@ Three release tracks are maintained: Two handler bodies (`approve`, `close`) lifted into the shared registrar with kind branching behind `SessionEndpointConfig` — - both kinds share one implementation per verb. The `close` - failure-status code standardized to 404 across both kinds (was - 500 on coord; "popped between .get() and .close()" is a not- - found semantic, not a server error). Other shared verbs (`send`, - `cancel`, `open`, `events`, `create`, `list`, `saved`, `history`, - `detail`) keep their per-kind handlers — body convergence for - those requires SessionManager-side refactors (e.g. Priority 1's - worker-dispatch unification for `send`) or coordinated frontend - changes (response-shape unification for `list` / `saved`) that - fall outside Priority 0 scope. + both kinds share one implementation per verb. Two related + behavior changes on the interactive close path: + + - `mgr.close()` race-loss returns 404 (was 500 on coord; + "popped between .get() and .close()" is a not-found semantic, + not a server error). + - Audit-write failures (`record_audit` raising on the storage + write) are now caught and logged at `warning` level; the close + still returns 200. Previously the interactive path let the + exception propagate as HTTP 500. Coord previously already + swallowed; convergence is intentional — operators monitor the + `ws.close.audit_failed` log line in both kinds the same way. + + Other shared verbs (`send`, `cancel`, `open`, `events`, `create`, + `list`, `saved`, `history`, `detail`) keep their per-kind + handlers — body convergence for those requires SessionManager- + side refactors (e.g. Priority 1's worker-dispatch unification + for `send`) or coordinated frontend changes (response-shape + unification for `list` / `saved`) that fall outside Priority 0 + scope. - **TypeScript SDK bumped to 0.4.0** to flag the URL change for any 1.5.0aN-era consumer of the experimental coord client. The diff --git a/tests/test_coordinator_end_to_end.py b/tests/test_coordinator_end_to_end.py index 6f30e7bb..196c51c3 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 TYPE_CHECKING, Any +from typing import Any from unittest.mock import MagicMock import httpx @@ -31,14 +31,13 @@ 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 ( - _auth_user_id, + _audit_close_coordinator, _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 ( @@ -47,26 +46,14 @@ from turnstone.core.session_routes import ( ) 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 "", - ) +# Per-kind config the lifted handler factories capture by closure. +_coord_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", +) # --------------------------------------------------------------------------- @@ -155,7 +142,8 @@ def _make_client( Route( "/v1/api/workstreams/{ws_id}/close", make_close_handler( - audit_emit=_audit_close_coordinator_e2e, + _coord_endpoint_config, + audit_emit=_audit_close_coordinator, supports_close_reason=False, ), methods=["POST"], @@ -175,13 +163,6 @@ 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 98972699..3d0f0130 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -3,12 +3,14 @@ Builds a minimal Starlette app wiring only the coordinator routes and an auth-injector middleware. Verifies the permission gate, 503 remediation when coord_mgr / model alias is missing, ownership -enforcement, and lazy rehydration on GET /{ws_id}. +enforcement, and lazy rehydration on GET /{ws_id}. Also exercises +the lifted ``approve`` and ``close`` handlers from +``turnstone.core.session_routes`` wired through the coord +``SessionEndpointConfig`` — same code path the live console uses. """ from __future__ import annotations -from typing import TYPE_CHECKING from unittest.mock import MagicMock import httpx @@ -29,7 +31,7 @@ from tests._coord_test_helpers import ( ) from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( - _auth_user_id, + _audit_close_coordinator, _require_admin_coordinator, _require_coord_mgr, cluster_ws_detail, @@ -44,7 +46,6 @@ 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, @@ -53,33 +54,23 @@ from turnstone.core.session_routes import ( ) 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 # --------------------------------------------------------------------------- +# Per-kind config the lifted handler factories capture by closure. +# Mirrors the production console wiring so tests exercise the same +# code path as the live server. +_coord_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", +) + + @pytest.fixture def storage(tmp_path): return SQLiteBackend(str(tmp_path / "coord.db")) @@ -115,7 +106,7 @@ def _make_client( ), Route( "/v1/api/workstreams/{ws_id}/approve", - make_approve_handler(), + make_approve_handler(_coord_endpoint_config), methods=["POST"], ), Route( @@ -126,7 +117,8 @@ def _make_client( Route( "/v1/api/workstreams/{ws_id}/close", make_close_handler( - audit_emit=_audit_close_coordinator_for_test, + _coord_endpoint_config, + audit_emit=_audit_close_coordinator, supports_close_reason=False, ), methods=["POST"], @@ -171,14 +163,6 @@ 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 - # Wire the per-kind endpoint config the lifted handlers consult. - 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/turnstone/console/server.py b/turnstone/console/server.py index dbfe2f82..bd8d4cf4 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -2388,6 +2388,33 @@ def _auth_scopes(request: Request) -> set[str]: return set(getattr(auth, "scopes", []) or []) +def _audit_close_coordinator( + request: Request, + ws_id: str, + ws_before: Workstream, # noqa: ARG001 — coord audit detail doesn't use it yet + reason: str, # noqa: ARG001 — coord doesn't expose close_reason yet +) -> None: + """Record the ``coordinator.close`` audit event. + + Passed to :func:`make_close_handler` as the ``audit_emit`` + callable. ``storage`` is guaranteed non-``None`` by the lifted + handler's upstream gate; the ``getattr`` fallback is defensive + consistency with the rest of the storage access pattern. + """ + storage = getattr(request.app.state, "auth_storage", None) + if storage is None: + return + 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 "", + ) + + async def coordinator_create(request: Request) -> JSONResponse: """POST /v1/api/workstreams/new — create a new coordinator session.""" from turnstone.core.audit import record_audit @@ -10093,24 +10120,6 @@ 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, @@ -10122,11 +10131,12 @@ def create_app( detail=coordinator_detail, open=coordinator_open, close=make_close_handler( # lifted: shared body + coord_endpoint_config, audit_emit=_audit_close_coordinator, supports_close_reason=False, ), send=coordinator_send, - approve=make_approve_handler(), # lifted: shared body + approve=make_approve_handler(coord_endpoint_config), # lifted: shared body cancel=coordinator_cancel, events=coordinator_events, history=coordinator_history, diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 3520c042..56d032bd 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -30,17 +30,22 @@ from __future__ import annotations from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from starlette.responses import JSONResponse from starlette.routing import Route +from turnstone.core.log import get_logger + if TYPE_CHECKING: from starlette.requests import Request from starlette.responses import Response from starlette.routing import BaseRoute from turnstone.core.session_manager import SessionManager + from turnstone.core.workstream import Workstream + +log = get_logger(__name__) Handler = Callable[["Request"], Awaitable["Response"]] @@ -316,7 +321,7 @@ def register_coord_verbs( # --------------------------------------------------------------------------- -def make_approve_handler() -> Handler: +def make_approve_handler(cfg: SessionEndpointConfig) -> Handler: """Lifted body for ``POST {prefix}/{ws_id}/approve``. Resolves a pending tool approval on the workstream's UI. Both @@ -329,15 +334,18 @@ def make_approve_handler() -> Handler: from turnstone.core.web_helpers import read_json_or_400 async def approve(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) + mgr_opt, err503 = cfg.manager_lookup(request) if err503 is not None: return err503 - assert mgr is not None # narrowed by the err503 None-check + # ``manager_lookup`` returns ``(None, JSONResponse)`` when the + # subsystem is unavailable (returned above) or + # ``(SessionManager, None)`` otherwise; ``cast`` makes the + # type-checker-only narrowing explicit and survives ``python -O``. + mgr = cast("SessionManager", mgr_opt) body = await read_json_or_400(request) if isinstance(body, JSONResponse): return body @@ -423,6 +431,7 @@ CloseAuditEmitter = Callable[ def make_close_handler( + cfg: SessionEndpointConfig, *, audit_emit: CloseAuditEmitter | None = None, supports_close_reason: bool = False, @@ -437,6 +446,9 @@ def make_close_handler( the workstream's config row. Args: + cfg: per-kind policy bundle (auth, manager lookup, tenant + check, error labels). Captured by closure so the request- + time handler doesn't reach into ``app.state``. 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 @@ -450,18 +462,30 @@ def make_close_handler( ``workstream_config`` from unbounded growth on a model- generated dump; the redact protects audit logs from captured-secret leakage under prompt injection. + + Behavior change vs the pre-lift handlers: + + - The interactive handler previously let ``record_audit`` failures + surface as HTTP 500 (no try/except). The lifted body wraps + ``audit_emit`` in try/except and demotes failures to a + ``warning`` log, returning 200 to the caller. Coord previously + already swallowed; convergence is intentional — operators + monitor the audit-fail log line in both kinds the same way. + - The coord ``mgr.close()`` race-loss returned 500; standardized + to 404 ("popped between ``.get()`` and ``.close()``" is a + not-found semantic, not a server error). """ 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) + mgr_opt, err503 = cfg.manager_lookup(request) if err503 is not None: return err503 - assert mgr is not None + # See ``make_approve_handler`` for the cast rationale. + mgr = cast("SessionManager", mgr_opt) ws_id = request.path_params.get("ws_id", "") reason = "" @@ -478,9 +502,7 @@ def make_close_handler( # 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" - ) + capped = raw_reason.strip().encode("utf-8")[:512].decode("utf-8", errors="ignore") reason = redact_credentials(capped) if cfg.tenant_check is not None: @@ -492,10 +514,6 @@ def make_close_handler( 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) @@ -503,9 +521,7 @@ def make_close_handler( try: storage.save_workstream_config(ws_id, {"close_reason": reason}) except Exception: - from turnstone.core.log import get_logger as _gl - - _gl(__name__).debug( + log.warning( "ws.close.reason_persist_failed ws=%s", ws_id[:8] if ws_id else "", exc_info=True, @@ -515,9 +531,11 @@ def make_close_handler( try: audit_emit(request, ws_id, ws_before, reason) except Exception: - from turnstone.core.log import get_logger as _gl - - _gl(__name__).debug( + # Audit-write failure is a compliance signal — + # ``warning`` so it surfaces in ops logs. Behavior change + # vs the original interactive handler (which would have + # 500'd here); see the function docstring. + log.warning( "ws.close.audit_failed ws=%s", ws_id[:8] if ws_id else "", exc_info=True, @@ -526,7 +544,3 @@ def make_close_handler( 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 6a0e2fa3..30757cbe 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -803,6 +803,72 @@ def _audit_context(request: Request) -> tuple[str, str]: return uid, ip +# --------------------------------------------------------------------------- +# Per-kind policies passed to the lifted session_routes handlers +# --------------------------------------------------------------------------- + + +def _interactive_manager_lookup( + request: Request, +) -> tuple[SessionManager | None, JSONResponse | None]: + """Return the interactive ``SessionManager`` from app.state. + + Interactive always has the manager loaded (it's constructed + synchronously at server startup), so the 503 branch is unused + on this side. Match the :data:`SessionRouteHandlers` signature + so the lifted handler bodies can call it uniformly. + """ + return request.app.state.workstreams, None + + +def _interactive_tenant_check( + request: Request, ws_id: str, mgr: SessionManager +) -> JSONResponse | None: + """Cross-tenant gate for the lifted session handlers. + + Forwards to :func:`_require_ws_access`, which returns 404 on + owner mismatch (the interactive trusted-team model). + """ + _owner, err = _require_ws_access(request, ws_id, mgr=mgr) + return err + + +def _audit_close_workstream( + request: Request, + ws_id: str, + ws_before: Workstream, + reason: str, +) -> None: + """Record the ``workstream.closed`` audit event for interactive close. + + Passed to :func:`make_close_handler` as the ``audit_emit`` + callable. ``storage`` is guaranteed non-``None`` by the lifted + handler's upstream gate; the ``getattr`` fallback is defensive + consistency with the rest of the storage access pattern. + """ + from turnstone.core.audit import record_audit + + storage = getattr(request.app.state, "auth_storage", None) + if storage is None: + return + _, 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, + ) + + # --------------------------------------------------------------------------- # Route handlers — all async # --------------------------------------------------------------------------- @@ -4305,51 +4371,17 @@ def create_app( # ``turnstone.core.session_routes`` so the console mounts the same # shape against its coord manager. ``session_endpoint_config`` # carries the kind-specific policy (auth, manager lookup, audit - # prefix) the lifted handler bodies consult at request time. - def _interactive_tenant_check( - request: Request, ws_id: str, mgr: SessionManager - ) -> JSONResponse | None: - # ``_require_ws_access`` returns ``(owner_uid, err)``; only the - # err matters for the gate. - _owner, err = _require_ws_access(request, ws_id, mgr=mgr) - return err - + # prefix) the lifted handler bodies consult. interactive_endpoint_config = SessionEndpointConfig( permission_gate=None, # interactive auth is enforced at the middleware layer - manager_lookup=lambda r: (r.app.state.workstreams, None), + manager_lookup=_interactive_manager_lookup, tenant_check=_interactive_tenant_check, not_found_label="Workstream not found", 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, - ) - + approve_handler = make_approve_handler(interactive_endpoint_config) close_handler = make_close_handler( + interactive_endpoint_config, audit_emit=_audit_close_workstream, supports_close_reason=True, )