ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(server): lift approve handler into shared session_routes body

Browse Source 
Stage 2 Priority 0 Step 0.2 body-convergence — first verb. Both
interactive ``approve`` and coord ``coordinator_approve`` handler
bodies collapse into ``make_approve_handler()`` in
``turnstone/core/session_routes.py``. Each kind sets a
``SessionEndpointConfig`` on ``app.state`` carrying the kind-
specific policies (auth gate, manager lookup, tenant check, audit
prefix, not-found label) the lifted body consults at request time.

The two interactive URLs converge:

- ``POST /v1/api/workstreams/{ws_id}/approve`` (new, path-keyed)
  reaches the lifted body directly via ``register_session_routes``.
- ``POST /v1/api/approve`` (legacy, body-keyed) keeps shipping;
  ``make_legacy_body_keyed_adapter`` peeks the body for ``ws_id``,
  splices it into ``request.path_params``, and forwards to the same
  lifted body. Frontend can keep using the legacy URL — no caller
  churn.

Coord exposes only the path-keyed shape (its URLs were experimental
in 1.5.0aN; the URL-shape commit already removed the ``coordinator/``
prefix).

Tenant-check is split out from permission-gate so interactive's
``_require_ws_access`` (404 on cross-owner) and coord's
``_require_admin_coordinator`` (cluster-wide scope) coexist without
either kind triggering the wrong gate.

Coord-side test fixture (``test_coordinator_endpoints._make_client``)
swaps the imported ``coordinator_approve`` for the lifted handler
and seeds ``app.state.session_endpoint_config`` so the tests
exercise the same code path the live console does.

Net delta: ~−25 LOC for this verb on top of the SessionEndpointConfig
+ legacy-adapter scaffolding (~80 LOC paid once). Subsequent verb
lifts amortize against that scaffolding.

ruff + mypy + 4366 pytest pass. Live console smoke against the
unified URL returns 503 (no coord_mgr loaded in the smoke env) —
proves the lifted handler is reachable + the manager_lookup callable
fires correctly.

Verbs still kind-specific (deferred — bodies have substantive
behavior divergence, not just naming): ``send`` (Priority 1
worker dispatch), ``cancel`` (interactive forensics + force flag),
``close`` (interactive close-reason cap+redact+persist), ``open``
(interactive resume vs coord rehydrate), ``events`` (different SSE
replay shapes), ``create`` (interactive attachments vs coord
initial_message), ``list`` / ``saved`` (different response keys).
This commit is contained in:
Patrick Buckley
2026-04-24 16:04:33 -07:00
committed by Patrick Buckley
parent 4a72b2ce19
commit 6415eeb91e
5 changed files with 246 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_coordinator_endpoints.py
+12 -2
View File
@@ -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)
tests/test_phase6_endpoints.py
+6 -58
View File
@@ -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=[
turnstone/console/server.py
+18 -42
View File
@@ -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
turnstone/core/session_routes.py
+181 -3
View File
@@ -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
turnstone/server.py
+29 -39
View File
@@ -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