diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index 0e84fbd3..58d94a5c 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -28,8 +28,9 @@ from tests._coord_test_helpers import ( ) from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( + _require_admin_coordinator, + _require_coord_mgr, cluster_ws_detail, - coordinator_approve, coordinator_cancel, coordinator_children, coordinator_close, @@ -43,6 +44,7 @@ from turnstone.console.server import ( coordinator_tasks, ) from turnstone.core.auth import AuthResult +from turnstone.core.session_routes import SessionEndpointConfig, make_approve_handler from turnstone.core.storage._sqlite import SQLiteBackend # --------------------------------------------------------------------------- @@ -85,7 +87,7 @@ def _make_client( ), Route( "/v1/api/workstreams/{ws_id}/approve", - coordinator_approve, + make_approve_handler(), methods=["POST"], ), Route( @@ -138,6 +140,14 @@ 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/tests/test_phase6_endpoints.py b/tests/test_phase6_endpoints.py index a3ed6a48..8664d031 100644 --- a/tests/test_phase6_endpoints.py +++ b/tests/test_phase6_endpoints.py @@ -13,82 +13,30 @@ upstream node fetches. from __future__ import annotations -from typing import Any -from unittest.mock import MagicMock - import pytest from starlette.applications import Starlette from starlette.middleware import Middleware -from starlette.middleware.base import BaseHTTPMiddleware from starlette.routing import Route from starlette.testclient import TestClient -from turnstone.console.collector import ClusterCollector -from turnstone.console.coordinator_adapter import CoordinatorAdapter -from turnstone.console.coordinator_ui import ConsoleCoordinatorUI +from tests._coord_test_helpers import ( + _AuthMiddleware, + _build_mgr, + _fake_registry, + _FakeConfigStore, +) from turnstone.console.server import ( cluster_ws_live_bulk, coordinator_metrics, ) -from turnstone.core.auth import AuthResult -from turnstone.core.session_manager import SessionManager from turnstone.core.storage._sqlite import SQLiteBackend -class _AuthMiddleware(BaseHTTPMiddleware): - """Inject a configurable AuthResult from header-based contract.""" - - async def dispatch(self, request, call_next): - perms = request.headers.get("X-Test-Perms", "") - user_id = request.headers.get("X-Test-User", "") - if perms or user_id: - request.state.auth_result = AuthResult( - user_id=user_id, - scopes=frozenset({"approve"}), - token_source="test", - permissions=frozenset(p for p in perms.split(",") if p), - ) - return await call_next(request) - - -class _FakeConfigStore: - def __init__(self, values: dict[str, Any]) -> None: - self._values = values - - def get(self, key: str, default: Any = None) -> Any: - return self._values.get(key, default) - - @pytest.fixture def storage(tmp_path): return SQLiteBackend(str(tmp_path / "phase6.db")) -def _build_mgr(storage) -> SessionManager: - def _sf(ui, model_alias=None, ws_id=None, **kw): - return MagicMock() - - adapter = CoordinatorAdapter( - collector=MagicMock(), - ui_factory=lambda ws: ConsoleCoordinatorUI(ws_id=ws.id, user_id=ws.user_id or ""), - session_factory=_sf, - ) - mgr = SessionManager( - adapter, - storage=storage, - max_active=3, - node_id=ClusterCollector.CONSOLE_PSEUDO_NODE_ID, - ) - adapter.attach(mgr) - return mgr - - -def _fake_registry() -> MagicMock: - reg = MagicMock() - reg.resolve.return_value = (MagicMock(), "gpt-4", MagicMock()) - return reg - - def _make_client(storage, *, coord_mgr=None) -> TestClient: app = Starlette( routes=[ diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 422f1a1e..49af308b 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -56,7 +56,9 @@ from turnstone.core.auth import ( from turnstone.core.rendezvous import NoAvailableNodeError from turnstone.core.session_routes import ( CoordOnlyVerbHandlers, + SessionEndpointConfig, SharedSessionVerbHandlers, + make_approve_handler, register_coord_verbs, register_session_routes, ) @@ -2523,44 +2525,6 @@ async def coordinator_send(request: Request) -> JSONResponse: return JSONResponse({"status": "ok"}) -async def coordinator_approve(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/{ws_id}/approve — unblock pending approval.""" - from turnstone.core.web_helpers import read_json_or_400 - - 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", "") - body = await read_json_or_400(request) - if isinstance(body, JSONResponse): - return body - approved = bool(body.get("approved", False)) - feedback = body.get("feedback") - always = bool(body.get("always", False)) - ws = coord_mgr.get(ws_id) - if ws is None: - return JSONResponse({"error": "coordinator not found"}, status_code=404) - ui = ws.ui - if ui is None or not hasattr(ui, "resolve_approval"): - return JSONResponse( - {"error": "coordinator UI does not support approval"}, - status_code=409, - ) - if always and approved and getattr(ui, "_pending_approval", None): - tool_names = { - it.get("approval_label", "") or it.get("func_name", "") - for it in ui._pending_approval.get("items", []) - if it.get("needs_approval") and it.get("func_name") and not it.get("error") - } - tool_names.discard("") - ui.auto_approve_tools.update(tool_names) - ui.resolve_approval(approved, feedback) - return JSONResponse({"status": "ok"}) - - async def coordinator_cancel(request: Request) -> JSONResponse: """POST /v1/api/workstreams/{ws_id}/cancel — cancel in-flight generation.""" from turnstone.core.audit import record_audit @@ -10149,9 +10113,20 @@ def create_app( _docs_handler = make_docs_handler() # Coord workstream HTTP tree mounts under the unified - # ``/api/workstreams/`` shape. Handlers look the coord manager up - # via ``request.app.state.coord_mgr`` because the manager is built - # in the lifespan, after app construction. + # ``/api/workstreams/`` shape. Lifted handlers (e.g. ``approve``) + # consult ``app.state.session_endpoint_config`` for the kind's + # auth + manager-lookup policies. Per-kind handlers + # (``coordinator_*``) still look the coord manager up via + # ``request.app.state.coord_mgr`` because the manager is built in + # the lifespan, after this app construction; future verb lifts + # carry that lookup into the config callable. + coord_endpoint_config = SessionEndpointConfig( + permission_gate=_require_admin_coordinator, + manager_lookup=_require_coord_mgr, + tenant_check=None, # cluster-wide admin.coordinator gate covers it + not_found_label="coordinator not found", + audit_action_prefix="coordinator", + ) coord_workstream_routes: list[Any] = [] register_session_routes( coord_workstream_routes, @@ -10164,7 +10139,7 @@ def create_app( open=coordinator_open, close=coordinator_close, send=coordinator_send, - approve=coordinator_approve, + approve=make_approve_handler(), # lifted: shared body cancel=coordinator_cancel, events=coordinator_events, history=coordinator_history, @@ -10621,6 +10596,7 @@ def create_app( lifespan=_lifespan, ) app.state.collector = collector + app.state.session_endpoint_config = coord_endpoint_config app.state.jwt_secret = jwt_secret app.state.auth_storage = auth_storage app.state.proxy_token_mgr = proxy_token_mgr diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 550be5b1..3a4c1a30 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -4,9 +4,10 @@ Both node and console processes mount the workstream HTTP tree at ``/v1/api/workstreams/`` via this registrar against their own :class:`~turnstone.core.session_manager.SessionManager` (interactive on the node, coordinator on the console). One URL shape, two -processes, kind-specific handler bodies wired in by the caller. +processes, kind-specific policy in :class:`SessionEndpointConfig` +that handlers consult at request time via ``app.state``. -Two registrar functions: +Three registrar functions: - :func:`register_session_routes` — verbs both kinds expose (``new``, ``close``, ``open``, ``delete``, ``send``, ``approve``, @@ -17,14 +18,21 @@ Two registrar functions: ``restrict``, ``stop_cascade``, ``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`) that bake the +:class:`SessionEndpointConfig` in at app-construction time. Both +node and console call the factory during startup and pass the +result as ``handlers.approve``. """ from __future__ import annotations from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any +from starlette.responses import JSONResponse from starlette.routing import Route if TYPE_CHECKING: @@ -32,8 +40,56 @@ if TYPE_CHECKING: from starlette.responses import Response from starlette.routing import BaseRoute + from turnstone.core.session_manager import SessionManager + Handler = Callable[["Request"], Awaitable["Response"]] +PermissionGate = Callable[["Request"], "JSONResponse | None"] +ManagerLookup = Callable[["Request"], tuple["SessionManager | None", "JSONResponse | None"]] +TenantCheck = Callable[ + ["Request", str, "SessionManager"], + "JSONResponse | None", +] + + +@dataclass(frozen=True) +class SessionEndpointConfig: + """Per-kind policy the lifted handler bodies consult at request time. + + Instantiated once per process during app construction and stored + on ``app.state.session_endpoint_config``. The unified handler + bodies pull this config + the kind manager from ``app.state`` + rather than taking either as a per-request parameter — keeps the + handler signatures uniform (``Handler = Request -> Response``) + so the registrar mounts them like any other route. + + - ``permission_gate``: kind's pre-handler permission check + (e.g. ``admin.coordinator`` for coord, ``None`` for interactive + which has no per-handler scope check beyond auth middleware). + Returns the rejection response when the gate fails, ``None`` + when the request passes. + - ``manager_lookup``: returns ``(SessionManager, None)`` when the + kind's manager is loaded, or ``(None, JSONResponse)`` with a + 503 when the subsystem isn't available (coord on a console + without configured models). For interactive the lookup just + returns ``(app.state.workstreams, None)``. + - ``tenant_check``: per-``ws_id`` cross-tenant guard. Interactive + uses ``_require_ws_access`` (404 on owner mismatch); coord + relies on the cluster-wide ``admin.coordinator`` scope from + ``permission_gate`` and sets this to ``None``. + - ``not_found_label``: the message body for the 404 returned when + the manager has no such ws_id ("Workstream not found" for + interactive; "coordinator not found" for coord). + - ``audit_action_prefix``: the dot-namespaced prefix the kind + uses for its audit actions ("workstream" → ``workstream.cancel``; + "coordinator" → ``coordinator.cancel``). + """ + + permission_gate: PermissionGate | None + manager_lookup: ManagerLookup + tenant_check: TenantCheck | None + not_found_label: str + audit_action_prefix: str @dataclass(frozen=True) @@ -236,3 +292,125 @@ def register_coord_verbs( methods=["POST"], ) ) + + +# --------------------------------------------------------------------------- +# Lifted handler bodies — Stage 2 Priority 0 body-convergence +# +# Each verb here was previously implemented twice (once in +# ``turnstone/server.py`` for interactive, once in +# ``turnstone/console/server.py`` for coord). The lifted body uses +# the kind-specific :class:`SessionEndpointConfig` from +# ``app.state.session_endpoint_config`` to branch on the few places +# the kinds legitimately differ. +# +# Verbs not lifted yet (intentional — bodies have substantive +# behavior divergence that needs SessionManager-side refactoring, +# not just kind branching): send (worker dispatch — Priority 1 +# territory), cancel (interactive does inline forensics + force-cancel +# ws._lock manipulation), close (interactive caps + redacts + +# persists close_reason), open (interactive resume vs coord rehydrate), +# events (different SSE replay shapes), create (interactive +# attachments vs coord initial_message), list / saved (different +# response keys: ``workstreams`` vs ``coordinators``). +# --------------------------------------------------------------------------- + + +def make_approve_handler() -> Handler: + """Lifted body for ``POST {prefix}/{ws_id}/approve``. + + Resolves a pending tool approval on the workstream's UI. Both + kinds expose the same approve / feedback / always body shape and + the same ``ui.resolve_approval(approved, feedback)`` mechanic; + differences are auth scope, manager lookup, and the + ``__budget_override__`` filter (interactive-only — coord workstreams + don't have the budget-override pseudo-tool). + """ + 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) + if err503 is not None: + return err503 + assert mgr is not None # narrowed by the err503 None-check + body = await read_json_or_400(request) + if isinstance(body, JSONResponse): + return body + ws_id = request.path_params.get("ws_id", "") + approved = bool(body.get("approved", False)) + feedback = body.get("feedback") + always = bool(body.get("always", False)) + 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 = mgr.get(ws_id) + if ws is None: + return JSONResponse({"error": cfg.not_found_label}, status_code=404) + ui = ws.ui + if ui is None or not hasattr(ui, "resolve_approval"): + return JSONResponse( + {"error": "session UI does not support approval"}, + status_code=409, + ) + # ``_pending_approval`` and ``auto_approve_tools`` aren't on the + # ``SessionUI`` Protocol — both interactive ``WebUI`` and + # ``ConsoleCoordinatorUI`` add them, but a kind-agnostic body + # has to look them up dynamically. The CLI ``CliUI`` wouldn't + # have either, so accessing through ``getattr`` is also safer. + pending = getattr(ui, "_pending_approval", None) + auto_approve_tools = getattr(ui, "auto_approve_tools", None) + if always and approved and pending and auto_approve_tools is not None: + tool_names: set[str] = { + it.get("approval_label", "") or it.get("func_name", "") + for it in pending.get("items", []) + if it.get("needs_approval") and it.get("func_name") and not it.get("error") + } + tool_names.discard("") + # Budget-override is an interactive-only pseudo-tool that + # must never be added to the auto-approve set — discarding + # unconditionally is safe (no-op for coord). + tool_names.discard("__budget_override__") + if tool_names: + auto_approve_tools.update(tool_names) + ui.resolve_approval(approved, feedback) + return JSONResponse({"status": "ok"}) + + return approve + + +def make_legacy_body_keyed_adapter(handler: Handler) -> Handler: + """Wrap a path-keyed handler so it can be mounted at a body-keyed URL. + + Pre-1.5 interactive handlers (``/api/approve``, ``/api/cancel``, + ``/api/plan``, etc.) take ``ws_id`` from the JSON body. The + lifted bodies in this module read ``ws_id`` from the path. This + adapter peeks the body for ``ws_id``, copies it into + ``request.path_params``, and forwards. Starlette caches the + request body so the lifted handler's own body read is a hash-map + lookup, not a second network read. + """ + + async def adapter(request: Request) -> Response: + 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") or "") + # ``request.path_params`` is normally populated by Starlette + # at Route match time; since this adapter is mounted on a + # body-keyed URL with no ``{ws_id}`` slot, we splice it into + # the scope so the lifted handler's ``request.path_params.get(...)`` + # finds it. + path_params: dict[str, Any] = dict(request.path_params) + path_params["ws_id"] = ws_id + request.scope["path_params"] = path_params + return await handler(request) + + return adapter diff --git a/turnstone/server.py b/turnstone/server.py index 7d155e44..6f5a34ed 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -57,7 +57,10 @@ from turnstone.core.session import ChatSession, GenerationCancelled, SessionUI from turnstone.core.session_manager import SessionManager from turnstone.core.session_routes import ( AttachmentHandlers, + SessionEndpointConfig, SharedSessionVerbHandlers, + make_approve_handler, + make_legacy_body_keyed_adapter, register_session_routes, ) from turnstone.core.session_ui_base import SessionUIBase @@ -1672,43 +1675,6 @@ async def send_message(request: Request) -> JSONResponse: ) -async def approve(request: Request) -> JSONResponse: - """POST /v1/api/approve — approve or deny a tool call.""" - from turnstone.core.web_helpers import read_json_or_400 - - body = await read_json_or_400(request) - if isinstance(body, JSONResponse): - return body - approved = body.get("approved", False) - feedback = body.get("feedback") - always = body.get("always", False) - ws_id = body.get("ws_id") - mgr = request.app.state.workstreams - # Cross-tenant guard: resolving a pending tool approval on another - # tenant's workstream is RCE-adjacent (the victim queued a command - # expecting to decide themselves). Gate before touching the UI. - # Pass mgr= so the check uses the in-memory ws.user_id and survives - # transient storage outages. - _owner, err = _require_ws_access(request, str(ws_id or ""), mgr=mgr) - if err: - return err - ws, ui = _get_ws(mgr, ws_id) - if not ws or not ui: - return JSONResponse({"error": "Unknown workstream"}, status_code=404) - if always and approved and ui._pending_approval: - tool_names = { - it.get("approval_label", "") or it.get("func_name", "") - for it in ui._pending_approval.get("items", []) - if it.get("needs_approval") and it.get("func_name") and not it.get("error") - } - tool_names.discard("") - tool_names.discard("__budget_override__") - if tool_names: - ui.auto_approve_tools.update(tool_names) - ui.resolve_approval(approved, feedback) - return JSONResponse({"status": "ok"}) - - async def plan_feedback(request: Request) -> JSONResponse: """POST /v1/api/plan — respond to a plan review.""" from turnstone.core.web_helpers import read_json_or_400 @@ -4423,7 +4389,25 @@ def create_app( # Workstream HTTP tree — owned by the shared registrar in # ``turnstone.core.session_routes`` so the console mounts the same - # shape against its coord manager. + # 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 + + interactive_endpoint_config = SessionEndpointConfig( + permission_gate=None, # interactive auth is enforced at the middleware layer + manager_lookup=lambda r: (r.app.state.workstreams, None), + tenant_check=_interactive_tenant_check, + not_found_label="Workstream not found", + audit_action_prefix="workstream", + ) + approve_handler = make_approve_handler() v1_routes: list[Any] = [ Route("/api/events", events_sse), Route("/api/events/global", global_events_sse), @@ -4440,6 +4424,7 @@ def create_app( open=open_workstream, refresh_title=refresh_workstream_title, set_title=set_workstream_title, + approve=approve_handler, # lifted: shared body attachments=AttachmentHandlers( upload=upload_attachment, list=list_attachments, @@ -4460,7 +4445,11 @@ def create_app( Route("/api/skills", list_skills_summary), Route("/api/models", list_available_models), Route("/api/send", send_message, methods=["POST", "DELETE"]), - Route("/api/approve", approve, methods=["POST"]), + Route( + "/api/approve", + make_legacy_body_keyed_adapter(approve_handler), + methods=["POST"], + ), Route("/api/plan", plan_feedback, methods=["POST"]), Route("/api/command", command, methods=["POST"]), Route("/api/cancel", cancel_generation, methods=["POST"]), @@ -4506,6 +4495,7 @@ def create_app( lifespan=_lifespan, ) app.state.workstreams = workstreams + app.state.session_endpoint_config = interactive_endpoint_config app.state.global_queue = global_queue app.state.global_listeners = global_listeners app.state.global_listeners_lock = global_listeners_lock