diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ea2e9eb..7adc8f71 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -400,6 +400,151 @@ Three release tracks are maintained: semantics are preserved by wrapping the iteration in the same try/except. +- **`create` verb body lifted across both kinds** ([Stage 2 Verb + Lift — `create`]). The interactive + ``POST /v1/api/workstreams/new`` and coord + ``POST /v1/api/workstreams/new`` handlers now share one body via + ``make_create_handler(cfg, *, audit_emit=None)``. Per-kind + divergence captured by five new ``SessionEndpointConfig`` fields: + + - ``create_supports_attachments: bool`` — multipart body parsing + + attachment validation+save+rollback. Both kinds wire ``True``. + - ``create_supports_user_id_override: bool`` — trusted-source + body ``user_id`` override (interactive ``True`` for console- + proxied creates; coord ``False``). + - ``create_validate_request: CreateRequestValidator | None`` — + per-kind pre-create gates (interactive: ws_id format, kind, + parent ownership, attachments+resume_ws combo; coord: 401-on- + empty-uid). + - ``create_build_kwargs: CreateKwargsBuilder | None`` — per-kind + kwargs dict for ``mgr.create``. + - ``create_post_install: CreatePostInstall | None`` — per-kind + tail end (interactive: WebUI auto_approve + watch_runner + + ``ws_created`` global broadcast + atomic resume + skill session + config + notify_targets + routing override + initial-message + worker thread; coord: ``coord_adapter.send`` for the optional + initial_message). + + The pure helper ``_validate_and_save_uploaded_files`` lifted from + ``turnstone.server`` to ``turnstone.core.attachments`` as + ``validate_and_save_uploaded_files`` so both processes can call + the same kind-agnostic implementation. + + **§ Post-P3 reckoning item #1 done — coord gains create-time + attachments.** Pre-lift ``coordinator_create`` accepted JSON only + and ignored uploads; the lifted body parses ``multipart/form-data`` + on coord and saves attachments through the kind-agnostic storage + layer. Coord's adapter ``send`` does NOT yet take attachments, + so when a create request carries both ``initial_message`` and + uploads, the rows save as pending and the next ``/send`` picks + them up via the standard send-with-attachments path. Initial- + message + create-time attachments coordination on coord is a + follow-up — out of scope for the verb-shape lift. + + Note on broadcast timing: coord's ``mgr.create`` fires + ``emit_created`` (cluster collector fan-out) BEFORE the lifted + body runs attachment validation. If validation fails on coord and + the rollback (``mgr.close`` → ``emit_closed``) fires, the cluster + events stream sees a phantom create→close pair. Cluster consumers + handle this gracefully (same shape as any quick-create-close); + decoupling ``emit_created`` from ``mgr.create`` would be a bigger + refactor that doesn't belong in the verb lift. Interactive's + broadcast (``gq.put_nowait("ws_created")``) is held until after + attachment validation by the post-install callback, so interactive + never sees the phantom pair. + + Five observable behaviour changes on the create response: + + - **Both kinds converge on 200 OK.** Pre-lift interactive + returned 200 (default JSONResponse status); pre-lift coord + returned 201. Picked 200 over 201 for response-shape parity + with every other shared verb at the cost of REST-strict + correctness — a one-time release note rather than ongoing + client churn (the rest of the v1 SDK already uses + ``response.ok`` per ``feedback_test_frontend_locally.md``). + SDK consumers that branched on ``status == 201`` for coord + must switch to ``response.ok``. + - **Always-include response shape.** Pre-lift interactive + returned ``{ws_id, name, resumed, message_count, attachment_ids}`` + (5 fields); pre-lift coord returned ``{ws_id, name}`` (2). The + lifted body always returns the full shape, with ``resumed=False`` + / ``message_count=0`` / ``attachment_ids=[]`` on kinds whose + post-install doesn't populate them. Coord callers will see the + parity fields appear with default values. + - **Both kinds converge on the manager-at-capacity 429 + semantic.** Pre-lift interactive translated ``mgr.create``'s + ``RuntimeError`` to 400; coord already translated to 429. The + documented contract on ``SessionManager.create`` is "raises + RuntimeError when the manager is at capacity" — 429 (rate- + limit / try-later) is the correct shape. + - **Both kinds converge on the factory-misconfig 503 semantic.** + Pre-lift interactive let ``ValueError`` propagate as 500 with + a stack trace; coord already translated to 503 with the + factory's remediation text. Operators get the actionable + message instead of the trace. + - **Both kinds get a correlation_id'd 500 on unexpected + ``mgr.create`` failure.** Pre-lift interactive let unexpected + exceptions propagate as 500 with a stack trace (potential + information leak via frame names / file paths); coord already + returned a correlation_id'd 500 with the message redacted. The + lifted body adopts coord's safer pattern on both kinds. + + Two coord-specific parity gains: + + - **Coord rejects disabled skills.** Pre-lift + ``coordinator_create`` silently allowed disabled skills to + flow through to ``mgr.create`` — the row would create with a + skill the operator had marked inert, surprising both the + operator and the next user. The lifted body returns 400 + "Skill not found or disabled" matching interactive's + behaviour. + - **Coord audit-emit failures no longer 500.** Pre-lift + ``coordinator_create`` already swallowed; pre-lift interactive + let the failure propagate as 500. The lifted body wraps + ``audit_emit`` in try/except + ``warning`` log, returning the + successful 200 to the caller. Mirrors the close / cancel / + open / events lift contracts. + + No legacy adapter is needed for create — both kinds already + mounted ``POST {prefix}/new`` pre-lift; the lifted handler slots + in at the same path on each kind. + + Three /review fixes folded into the same commit: + + - **Pre-lift's 400 on malformed ``notify_targets`` preserved.** The + initial draft surfaced ``notify_targets`` validation errors from + inside the interactive ``post_install`` callback, which the + factory had no return-the-400 channel for — the only signal was + to ``raise``, which the factory's generic exception handler + turned into a redacted 500. Worse, by the time ``post_install`` + ran the workstream was fully built (audit row written, + ``ws_created`` broadcast emitted), so a malformed-input request + surfaced as "create failed" with the workstream actually live. + Fixed by moving the ``notify_targets`` validation into + :func:`_interactive_create_validate_request` (the pre-create + gate), which returns the 400 before ``mgr.create`` runs and + keeps storage clean. New regression test: + ``test_create_lift_400s_on_malformed_notify_targets``. + - **Skill-lookup storage failure now correlation_id'd.** The + initial draft swallowed ``get_skill_by_name`` exceptions into + ``skill_data = None`` and returned a 400 "Skill not found or + disabled" — masking storage outages as user-input misses and + making operator triage of skill-related reports impossible. The + lifted body now lets the storage exception propagate to the + same correlation_id'd 500 path that ``mgr.create`` failures + use; the skill-lookup + version count + ``mgr.create`` all live + inside one ``try / except`` so storage outages anywhere in the + create-prelude get the redacted-message-with-correlation-id + treatment instead of a stack-traced 500 leak. + - **Whitespace-only ``skill`` field treated as empty.** The + initial draft took ``body.get("skill") or ""`` literally — a + payload with ``"skill": " "`` would have hit + ``get_skill_by_name(" ")`` and 400'd as "Skill not found". + Pre-lift coord stripped via ``(body.get("skill") or "").strip() + or None``; the lifted body now strips for both kinds (interactive + never received whitespace-only skills from the web UI but the + convergence is the safer default). + ### Security - **Coord attachment endpoints are now kind-strict** diff --git a/sdk/typescript/openapi-console.json b/sdk/typescript/openapi-console.json index 1390fc86..9a9f2bd6 100644 --- a/sdk/typescript/openapi-console.json +++ b/sdk/typescript/openapi-console.json @@ -4504,7 +4504,7 @@ "tags": [ "Coordinator" ], - "description": "Allocates a console-hosted ``kind=\"coordinator\"`` ChatSession. 201 on create; 429 when the ``coordinator.max_active`` cap is reached and no idle coordinator can be evicted.", + "description": "Allocates a console-hosted ``kind=\"coordinator\"`` ChatSession. 200 on create; 429 when the ``coordinator.max_active`` cap is reached and no idle coordinator can be evicted. Pre-1.5.0 this returned 201; the lifted ``create`` factory (Stage 2 verb lift) converges on 200 across both kinds.", "requestBody": { "required": true, "content": { @@ -4516,7 +4516,7 @@ } }, "responses": { - "201": { + "200": { "description": "Success", "content": { "application/json": { @@ -7458,7 +7458,7 @@ "type": "object" }, "CoordinatorCreateResponse": { - "description": "Response body for POST /v1/api/workstreams/new (201).", + "description": "Response body for POST /v1/api/workstreams/new (200).", "properties": { "ws_id": { "title": "Ws Id", @@ -7467,6 +7467,23 @@ "name": { "title": "Name", "type": "string" + }, + "resumed": { + "default": false, + "title": "Resumed", + "type": "boolean" + }, + "message_count": { + "default": 0, + "title": "Message Count", + "type": "integer" + }, + "attachment_ids": { + "items": { + "type": "string" + }, + "title": "Attachment Ids", + "type": "array" } }, "required": [ diff --git a/tests/test_coordinator_end_to_end.py b/tests/test_coordinator_end_to_end.py index 4dba98e1..c011232f 100644 --- a/tests/test_coordinator_end_to_end.py +++ b/tests/test_coordinator_end_to_end.py @@ -32,9 +32,12 @@ from turnstone.console.coordinator_client import CoordinatorClient from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( _audit_close_coordinator, + _audit_coordinator_create, + _coord_create_build_kwargs, + _coord_create_post_install, + _coord_create_validate_request, _require_admin_coordinator, _require_coord_mgr, - coordinator_create, coordinator_detail, coordinator_list, ) @@ -43,6 +46,7 @@ from turnstone.core.session_manager import SessionManager from turnstone.core.session_routes import ( SessionEndpointConfig, make_close_handler, + make_create_handler, ) from turnstone.core.storage._sqlite import SQLiteBackend @@ -53,6 +57,11 @@ _coord_endpoint_config = SessionEndpointConfig( tenant_check=None, not_found_label="coordinator not found", audit_action_prefix="coordinator", + create_supports_attachments=True, + create_supports_user_id_override=False, + create_validate_request=_coord_create_validate_request, + create_build_kwargs=_coord_create_build_kwargs, + create_post_install=_coord_create_post_install, ) @@ -136,7 +145,7 @@ def _make_client( routes=[ Route( "/v1/api/workstreams/new", - coordinator_create, + make_create_handler(_coord_endpoint_config, audit_emit=_audit_coordinator_create), methods=["POST"], ), Route("/v1/api/workstreams", coordinator_list, methods=["GET"]), @@ -186,7 +195,7 @@ def test_create_list_detail_lifecycle(tmp_path): json={"name": "e2e-coord"}, headers=_COORD_HEADERS, ) - assert resp.status_code == 201, resp.text + assert resp.status_code == 200, resp.text body = resp.json() ws_id = body["ws_id"] assert ws_id diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index 446692bb..619da14d 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -33,11 +33,14 @@ from turnstone.console.coordinator_ui import ConsoleCoordinatorUI from turnstone.console.server import ( _audit_cancel_coordinator, _audit_close_coordinator, + _audit_coordinator_create, + _coord_create_build_kwargs, + _coord_create_post_install, + _coord_create_validate_request, _require_admin_coordinator, _require_coord_mgr, cluster_ws_detail, coordinator_children, - coordinator_create, coordinator_detail, coordinator_history, coordinator_list, @@ -61,6 +64,7 @@ from turnstone.core.session_routes import ( make_attachment_handlers, make_cancel_handler, make_close_handler, + make_create_handler, make_open_handler, make_send_handler, ) @@ -106,6 +110,11 @@ _coord_endpoint_config = SessionEndpointConfig( ), spawn_metrics=None, emit_message_queued=True, + create_supports_attachments=True, + create_supports_user_id_override=False, + create_validate_request=_coord_create_validate_request, + create_build_kwargs=_coord_create_build_kwargs, + create_post_install=_coord_create_post_install, ) @@ -123,11 +132,14 @@ def _make_client( ) -> TestClient: """Build a TestClient exposing just the coordinator routes.""" coord_attachments = make_attachment_handlers(_coord_endpoint_config) + coord_create_handler = make_create_handler( + _coord_endpoint_config, audit_emit=_audit_coordinator_create + ) app = Starlette( routes=[ Route( "/v1/api/workstreams/new", - coordinator_create, + coord_create_handler, methods=["POST"], ), Route("/v1/api/workstreams", coordinator_list, methods=["GET"]), @@ -334,19 +346,125 @@ def test_create_returns_ws_id_and_records_audit(storage): json={"name": "my-coord"}, headers=_COORD_HEADERS, ) - assert resp.status_code == 201 + assert resp.status_code == 200 body = resp.json() assert body["ws_id"] assert "my-coord" in body["name"] + # Always-include parity fields land on coord post-`create` lift — + # SDK consumers don't have to branch on kind to read them. + assert body["resumed"] is False + assert body["message_count"] == 0 + assert body["attachment_ids"] == [] # Audit row recorded on storage. - from turnstone.core.audit import record_audit # noqa: F401 (verify import works) - - # Query audit_events via storage. events = storage.list_audit_events(user_id="user-1", limit=10) actions = [e["action"] for e in events] assert "coordinator.create" in actions +def test_create_with_multipart_attachments_saves_pending_rows(storage): + """§ Post-P3 reckoning item #1 regression — coord gains create-time + attachments. Multipart create with a magic-byte-valid PNG saves + a pending attachment row scoped to the new coord ws_id.""" + from turnstone.core.memory import list_pending_attachments + + # Magic-byte-valid 1x1 PNG (matches test_server_attachments_endpoints.py). + png_1x1 = ( + b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" + b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\rIDATx\x9cc\xfc\xcf" + b"\xc0\xc0\xc0\x00\x00\x00\x05\x00\x01\xa5\xf6E@\x00\x00\x00\x00IEND" + b"\xaeB`\x82" + ) + mgr = _build_mgr(storage) + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + + # Inject the test storage backend as the global singleton so + # ``save_attachment`` / ``list_pending_attachments`` (which both + # go through ``turnstone.core.memory`` → ``get_storage()``) + # resolve onto our SQLiteBackend instead of the real one. + import turnstone.core.storage._registry as _reg + + _old_storage = _reg._storage + _reg._storage = storage + try: + resp = client.post( + "/v1/api/workstreams/new", + data={"meta": '{"name": "with-image"}'}, + files={"file": ("img.png", png_1x1, "image/png")}, + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + ws_id = body["ws_id"] + assert ws_id + assert len(body["attachment_ids"]) == 1 + # Attachment row is pending (saved but not consumed) — coord's + # adapter ``send`` doesn't reserve attachments at create time + # yet, so the next ``/send`` picks it up via the standard + # send-with-attachments path. + pending = list_pending_attachments(ws_id, "user-1") + assert len(pending) == 1 + assert pending[0]["kind"] == "image" + finally: + _reg._storage = _old_storage + + +def test_create_rejects_disabled_skill(storage): + """Coord parity gain — disabled skills now rejected at the lift, + matching interactive's pre-lift behaviour. Pre-lift coord silently + let disabled skills through.""" + import turnstone.core.storage._registry as _reg + + _old_storage = _reg._storage + _reg._storage = storage + try: + # Save a disabled skill row so ``get_skill_by_name`` resolves + # but the lifted body's enabled check rejects it. Mirrors the + # ``_create_template`` helper in ``tests/test_skills.py``; + # inlined here so this regression test stays self-contained. + storage.create_prompt_template( + template_id="sk-disabled", + name="dormant-skill", + category="general", + content="dormant", + variables="[]", + is_default=False, + org_id="", + created_by="test", + origin="manual", + mcp_server="", + readonly=False, + description="", + tags="[]", + source_url="", + version="1.0.0", + author="", + activation="named", + token_estimate=0, + model="", + auto_approve=False, + temperature=None, + reasoning_effort="", + max_tokens=None, + token_budget=0, + agent_max_turns=None, + notify_on_complete="{}", + enabled=False, # the gate under test + allowed_tools="[]", + priority=0, + ) + mgr = _build_mgr(storage) + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + resp = client.post( + "/v1/api/workstreams/new", + json={"name": "x", "skill": "dormant-skill"}, + headers=_COORD_HEADERS, + ) + assert resp.status_code == 400 + assert "dormant-skill" in resp.json()["error"] + finally: + _reg._storage = _old_storage + + def test_list_returns_cluster_wide(storage): # Trusted-team visibility: any caller with admin.coordinator sees # every active coordinator regardless of owner. ``user_id`` stays diff --git a/tests/test_openapi.py b/tests/test_openapi.py index 03b3ffbe..ec70d990 100644 --- a/tests/test_openapi.py +++ b/tests/test_openapi.py @@ -139,16 +139,20 @@ class TestConsoleSpec: } assert expected.issubset(paths), f"Missing: {expected - paths}" - def test_coordinator_create_has_request_body_and_201(self): - """Coordinator create returns 201 (not 200) and accepts a body.""" + def test_coordinator_create_has_request_body_and_200(self): + """Coordinator create returns 200 and accepts a body. + + Pre-1.5.0 this returned 201 (REST-strict for create); the lifted + ``make_create_handler`` factory converges on 200 across both + kinds for response-shape parity with every other shared verb. + """ from turnstone.api.console_spec import build_console_spec spec = build_console_spec() op = spec["paths"]["/v1/api/workstreams/new"]["post"] assert "requestBody" in op assert "application/json" in op["requestBody"]["content"] - # Pin the 201 success code. - assert "201" in op["responses"] + assert "200" in op["responses"] def test_coordinator_history_has_limit_query_param(self): from turnstone.api.console_spec import build_console_spec diff --git a/tests/test_server_attachments_on_create.py b/tests/test_server_attachments_on_create.py index d4aea6d0..0e8dcaf6 100644 --- a/tests/test_server_attachments_on_create.py +++ b/tests/test_server_attachments_on_create.py @@ -50,9 +50,11 @@ def _auth(user: str) -> dict[str, str]: class TestValidateAndSaveUploadedFiles: def test_saves_image_and_text(self, tmp_path): + from turnstone.core.attachments import ( + validate_and_save_uploaded_files as _validate_and_save_uploaded_files, + ) from turnstone.core.memory import list_pending_attachments from turnstone.core.storage import init_storage, reset_storage - from turnstone.server import _validate_and_save_uploaded_files reset_storage() init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) @@ -73,8 +75,10 @@ class TestValidateAndSaveUploadedFiles: def test_rejects_oversized_image(self, tmp_path): from turnstone.core.attachments import IMAGE_SIZE_CAP + from turnstone.core.attachments import ( + validate_and_save_uploaded_files as _validate_and_save_uploaded_files, + ) from turnstone.core.storage import init_storage, reset_storage - from turnstone.server import _validate_and_save_uploaded_files reset_storage() init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) @@ -90,8 +94,10 @@ class TestValidateAndSaveUploadedFiles: reset_storage() def test_rejects_unsupported_text(self, tmp_path): + from turnstone.core.attachments import ( + validate_and_save_uploaded_files as _validate_and_save_uploaded_files, + ) from turnstone.core.storage import init_storage, reset_storage - from turnstone.server import _validate_and_save_uploaded_files reset_storage() init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) @@ -107,9 +113,11 @@ class TestValidateAndSaveUploadedFiles: def test_pending_cap_returns_409(self, tmp_path): from turnstone.core.attachments import MAX_PENDING_ATTACHMENTS_PER_USER_WS + from turnstone.core.attachments import ( + validate_and_save_uploaded_files as _validate_and_save_uploaded_files, + ) from turnstone.core.memory import save_attachment from turnstone.core.storage import init_storage, reset_storage - from turnstone.server import _validate_and_save_uploaded_files reset_storage() init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) diff --git a/tests/test_skills.py b/tests/test_skills.py index 7bdb603e..98e5645a 100644 --- a/tests/test_skills.py +++ b/tests/test_skills.py @@ -1731,7 +1731,18 @@ class TestSkillConfigAppliedToWorkstream: import turnstone.core.storage._registry as _reg from turnstone.core.adapters.interactive_adapter import InteractiveAdapter from turnstone.core.session_manager import SessionManager - from turnstone.server import WebUI, create_workstream + from turnstone.core.session_routes import ( + SessionEndpointConfig, + make_create_handler, + ) + from turnstone.server import ( + WebUI, + _interactive_create_build_kwargs, + _interactive_create_post_install, + _interactive_create_validate_request, + _interactive_manager_lookup, + _interactive_tenant_check, + ) storage = SQLiteBackend(str(tmp_path / "ws_test.db")) @@ -1769,13 +1780,29 @@ class TestSkillConfigAppliedToWorkstream: ) mgr = SessionManager(adapter, storage=storage, max_active=10, event_emitter=adapter) + # Build the same lifted create handler the production app + # mounts so this fixture exercises the make_create_handler + # factory rather than a parallel pre-lift body. + _test_cfg = SessionEndpointConfig( + permission_gate=None, + manager_lookup=_interactive_manager_lookup, + tenant_check=_interactive_tenant_check, + not_found_label="Workstream not found", + audit_action_prefix="workstream", + create_supports_attachments=True, + create_supports_user_id_override=True, + create_validate_request=_interactive_create_validate_request, + create_build_kwargs=_interactive_create_build_kwargs, + create_post_install=_interactive_create_post_install, + ) + _test_create_handler = make_create_handler(_test_cfg) routes = [ Mount( "/v1", routes=[ Route( "/api/workstreams/new", - create_workstream, + _test_create_handler, methods=["POST"], ), ], @@ -1798,6 +1825,23 @@ class TestSkillConfigAppliedToWorkstream: # Restore original storage singleton. _reg._storage = old_storage + def test_create_lift_400s_on_malformed_notify_targets(self, _ws_app): + """Regression for the lifted create handler — malformed + ``notify_targets`` returns 400 from the validator (pre-create + gate), not 500 from a post_install raise.""" + client, mgr, storage = _ws_app + + resp = client.post( + "/v1/api/workstreams/new", + json={"name": "x", "notify_targets": "{not json"}, + ) + assert resp.status_code == 400, resp.text + body = resp.json() + assert "error" in body + # The workstream must NOT have been created — the validator + # gates BEFORE mgr.create, so storage stays clean. + assert len(list(storage.list_workstreams())) == 0 + def test_session_receives_temperature(self, _ws_app): """Skill temperature overrides the session default.""" client, mgr, storage = _ws_app diff --git a/turnstone/api/console_schemas.py b/turnstone/api/console_schemas.py index f4c3dc08..7502ba96 100644 --- a/turnstone/api/console_schemas.py +++ b/turnstone/api/console_schemas.py @@ -1010,10 +1010,18 @@ class CoordinatorCreateRequest(BaseModel): class CoordinatorCreateResponse(BaseModel): - """Response body for POST /v1/api/workstreams/new (201).""" + """Response body for POST /v1/api/workstreams/new (200).""" ws_id: str name: str + # Always-include parity fields from the Stage 2 ``create`` verb + # lift. Coord doesn't populate ``resumed`` or ``message_count`` + # today (no resume-on-create surface yet), so they default to + # ``False`` / ``0``. ``attachment_ids`` carries the saved-but- + # pending attachment ids when the request was multipart. + resumed: bool = False + message_count: int = 0 + attachment_ids: list[str] = Field(default_factory=list) class CoordinatorInfo(BaseModel): diff --git a/turnstone/api/console_spec.py b/turnstone/api/console_spec.py index 408c1f24..2e52501f 100644 --- a/turnstone/api/console_spec.py +++ b/turnstone/api/console_spec.py @@ -1140,12 +1140,14 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [ "Create a new coordinator workstream", description=( 'Allocates a console-hosted ``kind="coordinator"`` ChatSession. ' - "201 on create; 429 when the ``coordinator.max_active`` cap is " - "reached and no idle coordinator can be evicted." + "200 on create; 429 when the ``coordinator.max_active`` cap is " + "reached and no idle coordinator can be evicted. " + "Pre-1.5.0 this returned 201; the lifted ``create`` factory " + "(Stage 2 verb lift) converges on 200 across both kinds." ), request_model=CoordinatorCreateRequest, response_model=CoordinatorCreateResponse, - response_code=201, + response_code=200, error_codes=[400, 401, 403, 429, 500, 503], tags=["Coordinator"], ), diff --git a/turnstone/console/server.py b/turnstone/console/server.py index eaa3ab3f..e9385c64 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -63,6 +63,7 @@ from turnstone.core.session_routes import ( make_attachment_handlers, make_cancel_handler, make_close_handler, + make_create_handler, make_events_handler, make_open_handler, make_send_handler, @@ -2481,110 +2482,111 @@ def _coord_events_replay( yield pending_plan -async def coordinator_create(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/new — create a new coordinator session.""" - from turnstone.core.audit import record_audit - from turnstone.core.web_helpers import read_json_or_400 +async def _coord_create_validate_request( + request: Request, + body: dict[str, Any], + uid: str, + uploaded_files: list[tuple[str, str, bytes]], +) -> JSONResponse | None: + """Per-kind pre-create gate for coord. - 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 - body = await read_json_or_400(request) - if isinstance(body, JSONResponse): - return body - user_id = _auth_user_id(request) - if not user_id: + Wired onto :attr:`SessionEndpointConfig.create_validate_request` + and called by :func:`make_create_handler` after body parsing + but before skill resolution / ``mgr.create``. The single gate is + a 401 when the auth result resolved to an empty user id — + coord's ``admin.coordinator`` scope check at the + ``permission_gate`` is the primary access boundary, but a token + that passes the scope check with ``sub=""`` would still land + here, and ``mgr.create`` requires a non-empty ``user_id``. + """ + if not uid: return JSONResponse({"error": "authentication required"}, status_code=401) - name = (body.get("name") or "").strip() - skill = (body.get("skill") or "").strip() or None - initial_message = (body.get("initial_message") or "").strip() - # Pre-resolve skill → (template_id, applied_version) so the - # persisted row records what was applied. SessionManager.create - # is kind-agnostic and no longer does this lookup itself. - skill_id_resolved = "" - skill_version_resolved = 0 - if skill: - from turnstone.core.memory import get_skill_by_name - from turnstone.core.storage._registry import get_storage as _get_storage + return None - _st = _get_storage() - skill_data = await asyncio.to_thread(get_skill_by_name, skill) - if skill_data and skill_data.get("template_id") and _st is not None: - skill_id_resolved = str(skill_data["template_id"]) - try: - skill_version_resolved = ( - await asyncio.to_thread(_st.count_skill_versions, skill_id_resolved) + 1 - ) - except Exception: - log.debug("coord_create.skill_version_failed skill=%s", skill, exc_info=True) - skill_version_resolved = 1 - try: - # Offload to a worker thread — SessionManager.create runs - # blocking storage calls (register_workstream) and the - # session-factory invocation. Running inline on the async - # event loop stalled the SSE manager + every other async - # handler for the duration of the create (#perf-4). - ws = await asyncio.to_thread( - coord_mgr.create, - user_id=user_id, - name=name, - skill=skill, - skill_id=skill_id_resolved, - skill_version=skill_version_resolved, - ) - except RuntimeError as exc: - return JSONResponse({"error": str(exc)}, status_code=429) - except ValueError as exc: - # Session factory raises ValueError on misconfigured alias — - # surface as 503 with the factory's remediation text. - return JSONResponse({"error": str(exc)}, status_code=503) - except Exception: - # Don't echo the exception text to the caller — it can leak - # internals (stack frame names, file paths, etc.). Log with a - # short correlation id and return that to the client so support - # can match a user report to the log line. - correlation_id = secrets.token_hex(4) - log.warning( - "coordinator_create.failed correlation_id=%s", - correlation_id, - exc_info=True, - ) - return JSONResponse( - { - "error": ( - "failed to create coordinator (internal error). " - f"correlation_id={correlation_id}" - ) - }, - status_code=500, - ) - # Initial message spawns the first worker after the session has - # been installed on the adapter. Previously this rode inside - # ``coord_mgr.create`` via the old ``initial_message`` kwarg; the - # unified SessionManager.create is kind-agnostic so the worker spawn - # moved to the caller. + +def _coord_create_build_kwargs( + request: Request, + body: dict[str, Any], + uid: str, + skill_data: dict[str, Any] | None, + skill_id: str, + applied_skill_version: int, +) -> dict[str, Any]: + """Build kwargs for ``coord_mgr.create`` from a parsed coord create body. + + Coord's create takes a smaller set than interactive's (no + ``model`` / ``judge_model`` / ``client_type`` / ``parent_ws_id`` / + ``ws_id``) — those concepts either don't apply to coordinators + (no parent on coord; coord ws_id is always server-generated) + or live on a separate ConfigStore knob (the dashboard-managed + plan/task model + reasoning_effort settings). + """ + body_skill = (body.get("skill") or "").strip() or None + name = (body.get("name") or "").strip() + return { + "user_id": uid, + "name": name, + "skill": body_skill, + "skill_id": skill_id, + "skill_version": applied_skill_version, + } + + +async def _coord_create_post_install( + request: Request, + ws: Workstream, + body: dict[str, Any], + uid: str, + skill_data: dict[str, Any] | None, + applied_skill_version: int, + attachment_ids: list[str], +) -> dict[str, Any]: + """Tail end of coord create: dispatch the initial message. + + Wired onto :attr:`SessionEndpointConfig.create_post_install`. The + factory has already saved any uploaded attachments through the + kind-agnostic storage layer; coord's adapter ``send`` does not + yet take attachments, so the rows save as pending and the next + ``/send`` picks them up via the standard send-with-attachments + path. Initial-message + create-time attachments coordination is + a follow-up. + + Returns ``{}`` — coord's response carries only the always-include + parity fields populated by the factory. + """ + initial_message = (body.get("initial_message") or "").strip() if initial_message: coord_adapter = getattr(request.app.state, "coord_adapter", None) if coord_adapter is not None: coord_adapter.send(ws.id, initial_message) + return {} + + +def _audit_coordinator_create( + request: Request, + ws: Workstream, + body: dict[str, Any], + uid: str, +) -> None: + """Audit emitter for the coord ``coordinator.create`` event. + + Wired onto :func:`make_create_handler` as ``audit_emit``. Failures + are caught + logged at ``warning`` by the factory. + """ + from turnstone.core.audit import record_audit + storage = getattr(request.app.state, "auth_storage", None) - if storage is not None: - try: - record_audit( - storage, - user_id, - "coordinator.create", - "workstream", - ws.id, - {"coord_ws_id": ws.id, "src": "coordinator", "name": ws.name}, - request.client.host if request.client else "", - ) - except Exception: - log.debug("coordinator_create.audit_failed", exc_info=True) - return JSONResponse({"ws_id": ws.id, "name": ws.name}, status_code=201) + if storage is None: + return + record_audit( + storage, + uid, + "coordinator.create", + "workstream", + ws.id, + {"coord_ws_id": ws.id, "src": "coordinator", "name": ws.name}, + request.client.host if request.client else "", + ) async def coordinator_history(request: Request) -> JSONResponse: @@ -10105,6 +10107,11 @@ def create_app( spawn_metrics=None, emit_message_queued=True, events_replay=_coord_events_replay, + create_supports_attachments=True, + create_supports_user_id_override=False, + create_validate_request=_coord_create_validate_request, + create_build_kwargs=_coord_create_build_kwargs, + create_post_install=_coord_create_post_install, ) coord_workstream_routes: list[Any] = [] register_session_routes( @@ -10113,7 +10120,10 @@ def create_app( handlers=SharedSessionVerbHandlers( list_workstreams=coordinator_list, list_saved=coordinator_saved, - create=coordinator_create, + create=make_create_handler( # lifted: shared body + coord_endpoint_config, + audit_emit=_audit_coordinator_create, + ), detail=coordinator_detail, open=make_open_handler(coord_endpoint_config), # lifted: shared body close=make_close_handler( # lifted: shared body diff --git a/turnstone/core/attachments.py b/turnstone/core/attachments.py index 9488878f..df772b04 100644 --- a/turnstone/core/attachments.py +++ b/turnstone/core/attachments.py @@ -14,8 +14,12 @@ from __future__ import annotations import collections import os import threading +import uuid from dataclasses import dataclass -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from starlette.responses import JSONResponse # Byte caps — enforced by the server layer at upload time. The # constants live here so the session / tests share the same definitions. @@ -188,6 +192,104 @@ def classify_text_attachment( return "text/plain", None +def validate_and_save_uploaded_files( + files: list[tuple[str, str, bytes]], + ws_id: str, + user_id: str, +) -> tuple[list[str], JSONResponse | None]: + """Classify + save a list of ``(filename, claimed_mime, data)`` tuples. + + Applies the same validation rules as ``upload_attachment`` (magic-byte + image sniffing, UTF-8 text decode, per-kind size cap, per-(ws,user) + pending cap) under the shared :func:`upload_lock`. + + Kind-agnostic: both interactive and coordinator create-with-attachments + paths call into this helper from the lifted ``make_create_handler`` + factory (Stage 2 ``create`` verb lift). The helper does not consult + any kind-specific config — the storage layer is kind-agnostic by + design (P1.5). + + Returns ``(attachment_ids, None)`` on success or ``(ids_saved_so_far, + JSONResponse)`` on the first failure so the caller can roll back any + partial state. + """ + from starlette.responses import JSONResponse as _JSONResponse + + from turnstone.core.memory import list_pending_attachments, save_attachment + + saved_ids: list[str] = [] + if not files: + return saved_ids, None + + lock = upload_lock(ws_id, user_id) + with lock: + pending_count = len(list_pending_attachments(ws_id, user_id)) + for filename, claimed_mime, data in files: + if not data: + return saved_ids, _JSONResponse({"error": "Empty file"}, status_code=400) + sniffed_image = sniff_image_mime(data) + if sniffed_image is not None: + if len(data) > IMAGE_SIZE_CAP: + return saved_ids, _JSONResponse( + { + "error": ( + f"Image too large ({len(data):,} bytes); " + f"cap is {IMAGE_SIZE_CAP:,} bytes." + ), + "code": "too_large", + }, + status_code=413, + ) + kind = "image" + mime = sniffed_image + else: + if len(data) > TEXT_DOC_SIZE_CAP: + return saved_ids, _JSONResponse( + { + "error": ( + f"Text document too large ({len(data):,} bytes); " + f"cap is {TEXT_DOC_SIZE_CAP:,} bytes." + ), + "code": "too_large", + }, + status_code=413, + ) + mime_or_err = classify_text_attachment(filename, claimed_mime, data) + if mime_or_err[0] is None: + return saved_ids, _JSONResponse( + {"error": mime_or_err[1], "code": "unsupported"}, + status_code=400, + ) + kind = "text" + mime = mime_or_err[0] + + if pending_count + 1 > MAX_PENDING_ATTACHMENTS_PER_USER_WS: + return saved_ids, _JSONResponse( + { + "error": ( + f"Too many pending attachments " + f"(max {MAX_PENDING_ATTACHMENTS_PER_USER_WS} pending per workstream)" + ), + "code": "too_many", + }, + status_code=409, + ) + attachment_id = uuid.uuid4().hex + save_attachment( + attachment_id, + ws_id, + user_id, + filename, + mime, + len(data), + kind, + data, + ) + saved_ids.append(attachment_id) + pending_count += 1 + return saved_ids, None + + def unreadable_placeholder(filename: str) -> dict[str, Any]: """Return a content-part placeholder used when an attachment can't be decoded for a given turn. diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 9d3d8db9..6f1f7315 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -148,6 +148,65 @@ class EventsReplay(Protocol): SseExecutorLookup = Callable[["Request"], Any] +# (request, body, uid, uploaded_files) -> JSONResponse | None. +# Optional kind-specific gate the lifted ``create`` body fires after +# body parsing + uid resolution but before skill resolution and +# ``mgr.create``. Returns ``None`` to continue, or a 4xx response +# to short-circuit. Interactive wires gates for ws_id format, +# kind=INTERACTIVE, parent_ws_id ownership, attachments+resume_ws +# combo. Coord wires a 401-on-empty-uid (admin tokens always carry a +# uid in practice; the gate is defensive). Mostly read-only — the +# parent_ws_id ownership gate does a single storage lookup but +# doesn't mutate anything. +CreateRequestValidator = Callable[ + ["Request", dict[str, Any], str, list[tuple[str, str, bytes]]], + "Awaitable[JSONResponse | None]", +] +# (request, body, uid, skill_data, skill_id, applied_skill_version) -> kwargs. +# Builds the kwargs dict for ``mgr.create``. Both kinds call +# ``mgr.create`` with the same callable shape, but the kwargs they +# pass differ (interactive threads model + judge_model + client_type + +# parent_ws_id + ws_id; coord threads only the smaller subset). +# Captured in a per-kind callable rather than a flag-soup so the +# kwargs dict construction stays readable at the wire-up site. +CreateKwargsBuilder = Callable[ + ["Request", dict[str, Any], str, dict[str, Any] | None, str, int], + dict[str, Any], +] +# (request, ws, body, uid, skill_data, applied_skill_version, attachment_ids) -> +# extra response fields. Kind-specific tail end the lifted ``create`` +# body fires after the workstream is built, attachments are saved, +# and audit is emitted. Returns extra fields to merge into the +# response (e.g. interactive returns ``{resumed, message_count}``; +# coord returns ``{}``). May spawn worker threads / register watch +# runners / persist skill session config / dispatch initial messages +# / pin routing. The factory does NOT wrap the call in try/except: +# post-install failures should surface to the caller as 5xx so the +# operator sees the misconfig instead of a half-built workstream. +CreatePostInstall = Callable[ + [ + "Request", + "Workstream", + dict[str, Any], + str, + dict[str, Any] | None, + int, + list[str], + ], + "Awaitable[dict[str, Any]]", +] +# (request, ws, body, uid) -> None. Audit emitter for the create +# event. Interactive emits ``workstream.created`` with +# ``{kind, parent_ws_id}`` detail; coord emits ``coordinator.create`` +# with ``{coord_ws_id, src, name}`` detail. Wrapped in try/except by +# the factory — audit-write failures shouldn't surface as HTTP 500 +# (mirrors the close / cancel / open lift contracts). +CreateAuditEmitter = Callable[ + ["Request", "Workstream", dict[str, Any], str], + None, +] + + @dataclass(frozen=True) class AttachmentUploadHelpers: """Process-local hooks the lifted attachment factories call into. @@ -282,6 +341,43 @@ class SessionEndpointConfig: # body falls through to the default executor. See # :data:`SseExecutorLookup` docstring above. sse_executor_lookup: SseExecutorLookup | None = None + # When ``True``, the lifted ``create`` body parses + # ``multipart/form-data`` (with one ``meta`` JSON field + zero or + # more ``file`` parts) in addition to plain ``application/json``. + # Both kinds wire ``True`` post-create-lift — coord gains + # create-time attachments here (§ Post-P3 reckoning item #1). + # The actual attachment validation+save+rollback always uses the + # storage layer (kind-agnostic since P1.5); this flag only + # toggles whether the multipart parse is attempted at all. + create_supports_attachments: bool = False + # When ``True``, the lifted ``create`` body honours a ``user_id`` + # field in the request body if the caller's auth token comes from + # a trusted service (currently just ``"console"``). Interactive + # wires ``True`` so console-proxied creates can carry the real + # end user's identity through to the workstream owner. Coord + # wires ``False`` — coord create runs only on the console process + # and the operator's auth result is the source of truth. + create_supports_user_id_override: bool = False + # (request, body, uid, uploaded_files) -> JSONResponse | None. + # Per-kind pre-create gate (ws_id format, parent ownership, kind + # validation, etc. on interactive; 401-on-empty-uid on coord). + # ``None`` skips the gate entirely. + create_validate_request: CreateRequestValidator | None = None + # (request, body, uid, skill_data, skill_id, applied_skill_version) + # -> kwargs for ``mgr.create``. Required when the kind mounts a + # ``create`` handler — the lifted body has no opinion on the + # kind-specific kwarg shape and threads whatever this returns + # straight through to ``await asyncio.to_thread(mgr.create, **kwargs)``. + create_build_kwargs: CreateKwargsBuilder | None = None + # (request, ws, body, uid, skill_data, applied_skill_version, + # attachment_ids) -> extra response fields. Kind-specific tail + # end fired after attachments save + audit. Interactive returns + # ``{resumed, message_count}`` and spawns the initial-message + # worker thread; coord returns ``{}`` and dispatches via + # ``coord_adapter.send`` when an initial_message is provided. + # ``None`` skips the post-install entirely (response is just + # ``{ws_id, name, ...}`` with empty parity fields). + create_post_install: CreatePostInstall | None = None @dataclass(frozen=True) @@ -1318,6 +1414,376 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: return events +def make_create_handler( + cfg: SessionEndpointConfig, + *, + audit_emit: CreateAuditEmitter | None = None, +) -> Handler: + """Lifted body for ``POST {prefix}/new`` — workstream creation. + + Both kinds share the create sequence (parse body → resolve uid → + resolve skill → kind-specific validate → ``mgr.create`` → save + attachments → audit → kind-specific post-install → respond). + Per-kind divergence captured by the cfg + ``audit_emit``: + + - ``cfg.create_supports_attachments`` — when ``True``, the body + may arrive as ``multipart/form-data`` with a ``meta`` JSON + field + ``file`` parts; uploads are validated post-create and + the workstream is rolled back if any file fails (interactive's + pre-lift pattern, lifted to coord here for parity). + - ``cfg.create_supports_user_id_override`` — when ``True``, a + ``user_id`` body field overrides the auth-derived uid if the + auth token is from a trusted service. Interactive ``True`` so + console-proxied creates carry the real end-user identity; + coord ``False``. + - ``cfg.create_validate_request`` — kind-specific pre-create + gates (interactive: ws_id format, kind, parent_ws_id ownership, + attachments+resume_ws combo; coord: 401-on-empty-uid). + - ``cfg.create_build_kwargs`` — kind-specific kwargs for + ``mgr.create``. Required when the kind mounts a create handler. + - ``cfg.create_post_install`` — kind-specific tail end (e.g. + interactive's resume + skill_config + initial-message worker + thread; coord's initial_message via coord_adapter.send). + - ``audit_emit`` — ``workstream.created`` on interactive, + ``coordinator.create`` on coord. + + Behavior changes vs the pre-lift handlers (documented in + CHANGELOG, mostly coord-up-to-interactive parity gains): + + - **Coord gains create-time attachments.** Pre-lift + ``coordinator_create`` accepted JSON only and ignored uploads; + the lifted body parses multipart bodies on coord and saves + attachments through the kind-agnostic storage layer (§ Post-P3 + reckoning item #1). Coord's initial_message dispatch does NOT + yet reserve those attachments onto the first turn (coord + adapter's ``send`` doesn't take attachments) — the rows save + as pending and the first ``/send`` picks them up via the + standard send-with-attachments path. Initial-message + create- + time attachments coordination is a follow-up. + - **Coord gains the disabled-skill rejection.** Pre-lift + ``coordinator_create`` silently allowed disabled skills to + flow through to ``mgr.create``; the lifted body returns 400 + ("Skill not found or disabled") matching interactive's + behaviour. Disabled skills are inert by definition; the gate + makes that explicit. + - **Both kinds converge on 200 OK.** Pre-lift interactive + returned 200 (default); coord returned 201. SDK consumers + that were branching on ``response.status == 201`` on coord + should switch to ``response.ok``. 200 was picked over 201 for + response-shape parity with the rest of the v1 surface (every + other shared verb returns 200), at the cost of leaving REST- + strictly-correct semantics on the table — a one-time release + note rather than ongoing client churn. + - **Both kinds converge on the manager-at-capacity 429 + semantic.** Pre-lift interactive translated mgr.create's + ``RuntimeError`` to 400 ("invalid create request"); coord + already translated to 429. RuntimeError on ``SessionManager.create`` + is documented as "manager at capacity" — 429 (rate-limit / + try-later) is the correct shape for both. + - **Both kinds converge on the factory-misconfig 503 + semantic.** Pre-lift interactive let ``ValueError`` (raised by + the session factory on a misconfigured model alias) propagate + as 500; coord already translated to 503. The lifted body uses + 503 with the factory's remediation text on both kinds — + operators get the actionable message instead of a generic + stack-traced 500. + - **Both kinds get a correlation_id'd 500 on unexpected + ``mgr.create`` failure.** Pre-lift interactive let unexpected + exceptions propagate as 500 with a stack-traced response + (potential information leak); coord already returned a + correlation_id'd 500 with the message redacted. The lifted + body adopts coord's safer pattern on both kinds. + - **Audit-emit failures no longer 500.** Pre-lift interactive + audit failures surfaced as HTTP 500 (no try/except); coord + swallowed via try/except + log.debug. The lifted body wraps + ``audit_emit`` in try/except + ``warning`` log, returning the + successful 201 to the caller. Mirrors the close / cancel / + open lift contracts. + - **Always-include response shape.** The lifted body always + returns ``{ws_id, name, resumed, message_count, attachment_ids}``, + with the parity fields defaulting to ``False`` / ``0`` / ``[]`` + on kinds whose post-install doesn't populate them. SDK + consumers don't branch on kind. + + Args: + cfg: per-kind policy bundle. + audit_emit: kind's audit emitter for the create event. + ``None`` skips the audit entirely. + """ + # Lazy-imported at factory call time (mirrors the events lift) so + # ``session_routes.py``'s top-level import graph stays tight. + + async def create(request: Request) -> Response: + import asyncio + import contextlib + import secrets + + from turnstone.core.attachments import ( + IMAGE_SIZE_CAP, + validate_and_save_uploaded_files, + ) + from turnstone.core.web_helpers import ( + read_json_or_400, + read_multipart_create_or_400, + ) + + if cfg.permission_gate is not None: + err = cfg.permission_gate(request) + if err is not None: + return err + mgr_opt, err503 = cfg.manager_lookup(request) + if err503 is not None: + return err503 + # See ``make_approve_handler`` for the cast rationale. + mgr = cast("SessionManager", mgr_opt) + + # --- Body parsing ------------------------------------------------- + # Multipart only when the cfg lights up attachments AND the + # caller actually sent a multipart body. Plain JSON stays the + # default content type for both kinds. + content_type = (request.headers.get("content-type") or "").lower() + uploaded_files: list[tuple[str, str, bytes]] = [] + body: dict[str, Any] + if cfg.create_supports_attachments and content_type.startswith("multipart/form-data"): + # Multipart cap: up to MAX_PENDING × image cap, plus slack + # for JSON meta + multipart framing. Per-file size is + # enforced inside :func:`validate_and_save_uploaded_files` + # against the kind-specific cap. + parsed = await read_multipart_create_or_400( + request, + max_files=10, + max_per_file_bytes=IMAGE_SIZE_CAP, + max_total_bytes=10 * IMAGE_SIZE_CAP, + ) + if isinstance(parsed, JSONResponse): + return parsed + body, uploaded_files = parsed + else: + json_body = await read_json_or_400(request) + if isinstance(json_body, JSONResponse): + return json_body + body = json_body + + # --- User id resolution ------------------------------------------ + # Auth middleware populates ``request.state.auth_result`` for + # every authed request; we just read the user_id off it. + auth = getattr(getattr(request, "state", None), "auth_result", None) + uid: str = getattr(auth, "user_id", "") or "" + if cfg.create_supports_user_id_override: + # Trusted services (currently just ``console``) may forward + # the real end-user's id in the body so console-proxied + # creates carry the right owner. Token sources on end-user + # tokens (including console-proxy tokens that carry the + # real user's identity at the auth layer) are NOT trusted; + # only service identities. The deny-by-default keeps a + # malicious caller from impersonating other users. + body_uid = body.get("user_id") + if ( + isinstance(body_uid, str) + and body_uid + and auth is not None + and getattr(auth, "token_source", "") in {"console"} + ): + uid = body_uid + + # --- Per-kind pre-create validation ------------------------------ + # Interactive validates ws_id format, kind, parent ownership, + # attachments+resume_ws combo. Coord 401s on empty uid. + if cfg.create_validate_request is not None: + err_validate = await cfg.create_validate_request(request, body, uid, uploaded_files) + if err_validate is not None: + return err_validate + + # --- Skill resolution -------------------------------------------- + # Both kinds resolve a body ``skill`` field through + # ``get_skill_by_name`` to the skill_data dict + the next + # applied_skill_version. Interactive previously skipped this + # entirely on resume_ws (the resumed session restores its own + # skill from config); the resume gate is captured by the + # interactive validator above (it returns 400 on the + # attachments+resume combo, but standalone resume_ws + skill + # is still allowed). To preserve that exact pre-lift skip on + # interactive, the validator may stash a sentinel — but + # simplest: re-read resume_ws_id here and skip skill lookup + # when both kinds see a non-empty resume_ws_id (coord doesn't + # support resume_ws today; the field is silently ignored). + # Strip whitespace on the skill name so a caller passing + # ``"skill": " "`` is treated identically to ``"skill": ""`` + # (skip skill resolution). Pre-lift coord explicitly stripped + # via ``(body.get("skill") or "").strip() or None``; pre-lift + # interactive didn't strip but never received whitespace-only + # skill names from the web UI. Convergence on the safer + # behaviour avoids a misleading 400 for an inert payload. + body_skill_raw = body.get("skill") or "" + body_skill = body_skill_raw.strip() if isinstance(body_skill_raw, str) else "" + resume_ws_id_raw = body.get("resume_ws") or "" + skill_data: dict[str, Any] | None = None + applied_skill_version = 0 + + # --- mgr.create (with skill resolution) ------------------------- + # Skill lookup + version count + ``mgr.create`` all live inside + # one try/except so any storage failure during skill resolution + # gets the same correlation_id'd 500 as a ``mgr.create`` + # exception. Pre-lift interactive let storage exceptions + # propagate to a stack-traced 500; the lifted body keeps the + # 500 status but redacts the message (operator gets the + # correlation id; logs carry the full ``exc_info``). The + # ``RuntimeError`` (capacity) and ``ValueError`` (factory + # misconfig) branches stay specific to ``mgr.create``: the + # skill-lookup path doesn't raise either of those. + if cfg.create_build_kwargs is None: + # The cfg required a build_kwargs callback for any kind + # mounting a create handler. Surface the misconfig as 500 + # with a clear log line so the operator sees it instead of + # a confusing AttributeError. + log.error("ws.create.misconfigured_no_build_kwargs") + return JSONResponse( + {"error": "create handler misconfigured"}, + status_code=500, + ) + try: + if body_skill and not (isinstance(resume_ws_id_raw, str) and resume_ws_id_raw): + from turnstone.core.memory import get_skill_by_name + from turnstone.core.storage._registry import get_storage as _get_storage + + skill_data = await asyncio.to_thread(get_skill_by_name, body_skill) + if not skill_data or not skill_data.get("enabled", False): + return JSONResponse( + {"error": f"Skill not found or disabled: {body_skill}"}, + status_code=400, + ) + tid = skill_data.get("template_id") + if tid: + _st = _get_storage() + if _st is not None: + # ``count_skill_versions`` is best-effort: if the + # version count call fails (transient storage + # blip), default to 1 rather than aborting the + # whole create. Persisted skill_version=1 is + # the right semantic for the first applied + # instance even if the count was unobtainable. + try: + applied_skill_version = ( + await asyncio.to_thread(_st.count_skill_versions, str(tid)) + 1 + ) + except Exception: + log.debug( + "ws.create.skill_version_failed skill=%s", + body_skill, + exc_info=True, + ) + applied_skill_version = 1 + skill_id_resolved = ( + str(skill_data["template_id"]) + if skill_data and skill_data.get("template_id") + else "" + ) + kwargs = cfg.create_build_kwargs( + request, body, uid, skill_data, skill_id_resolved, applied_skill_version + ) + ws = await asyncio.to_thread(mgr.create, **kwargs) + except RuntimeError as exc: + # ``SessionManager.create`` documents RuntimeError as + # "manager at capacity" — translate to 429 (rate-limit / + # try-later) on both kinds. + return JSONResponse({"error": str(exc)}, status_code=429) + except ValueError as exc: + # Session factory raises ValueError on misconfigured alias + # (model alias points at a model that no longer exists, + # etc.). Surface the factory's remediation text as 503 so + # operators get the actionable message instead of a + # stack-traced 500. + return JSONResponse({"error": str(exc)}, status_code=503) + except Exception: + # Don't echo the exception text — it can leak internal + # paths / frame names. Log with a correlation id and + # return that to the client so support can match a report + # to the log line. + correlation_id = secrets.token_hex(4) + log.warning( + "ws.create.failed correlation_id=%s", + correlation_id, + exc_info=True, + ) + kind_noun = cfg.audit_action_prefix or "workstream" + return JSONResponse( + { + "error": ( + f"failed to create {kind_noun} (internal error). " + f"correlation_id={correlation_id}" + ) + }, + status_code=500, + ) + + # --- Attachment validation + save + rollback -------------------- + # Interactive's pre-lift pattern: validate post-create so + # ``ws_id`` is bound (the storage layer scopes attachments by + # ws_id). On any failure, mgr.close + delete_workstream the + # workstream so the caller doesn't see a half-built phantom. + # On coord this is a brand-new path — pre-lift coord_create + # had no attachments support. Note: coord's mgr.create already + # fired ``emit_created`` (cluster collector fan-out) by this + # point, so a rollback here produces a phantom create→close + # pair on the cluster events stream. Cluster consumers handle + # this gracefully (same shape as any quick-create-close), and + # the alternative (decoupling emit_created from mgr.create) + # is a bigger refactor that doesn't belong in the verb lift. + attachment_ids: list[str] = [] + if uploaded_files: + saved_ids, save_err = await asyncio.to_thread( + validate_and_save_uploaded_files, uploaded_files, ws.id, uid + ) + if save_err is not None: + from turnstone.core.memory import delete_workstream as _delete_ws + + with contextlib.suppress(Exception): + await asyncio.to_thread(mgr.close, ws.id) + with contextlib.suppress(Exception): + await asyncio.to_thread(_delete_ws, ws.id) + return save_err + attachment_ids = saved_ids + + # --- Audit emit -------------------------------------------------- + if audit_emit is not None: + try: + audit_emit(request, ws, body, uid) + except Exception: + # Mirrors make_close_handler / make_cancel_handler / + # make_open_handler — audit-write failures shouldn't + # surface as HTTP 500. Log + continue. + log.warning( + "ws.create.audit_failed ws=%s", + ws.id[:8] if ws.id else "", + exc_info=True, + ) + + # --- Per-kind post-install --------------------------------------- + extra_response: dict[str, Any] = {} + if cfg.create_post_install is not None: + extra_response = await cfg.create_post_install( + request, + ws, + body, + uid, + skill_data, + applied_skill_version, + attachment_ids, + ) + + return JSONResponse( + { + "ws_id": ws.id, + "name": ws.name, + "resumed": bool(extra_response.get("resumed", False)), + "message_count": int(extra_response.get("message_count", 0)), + "attachment_ids": attachment_ids, + } + ) + + return create + + def make_send_handler(cfg: SessionEndpointConfig) -> Handler: """Lifted body for ``POST {prefix}/{ws_id}/send`` — message dispatch. diff --git a/turnstone/server.py b/turnstone/server.py index f58a15a8..2da299b8 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -66,6 +66,7 @@ from turnstone.core.session_routes import ( make_attachment_handlers, make_cancel_handler, make_close_handler, + make_create_handler, make_dequeue_handler, make_events_handler, make_legacy_body_keyed_adapter, @@ -2013,209 +2014,55 @@ def _reserve_and_resolve_attachments( return resolved, ordered_reserved, dropped -def _validate_and_save_uploaded_files( - files: list[tuple[str, str, bytes]], - ws_id: str, - user_id: str, -) -> tuple[list[str], JSONResponse | None]: - """Classify + save a list of ``(filename, claimed_mime, data)`` tuples. +async def _interactive_create_validate_request( + request: Request, + body: dict[str, Any], + uid: str, + uploaded_files: list[tuple[str, str, bytes]], +) -> JSONResponse | None: + """Per-kind pre-create gates for interactive workstreams. - Applies the same validation rules as ``upload_attachment`` (magic-byte - image sniffing, UTF-8 text decode, per-kind size cap, per-(ws,user) - pending cap) under the shared ``_attachment_upload_lock``. + Wired onto :attr:`SessionEndpointConfig.create_validate_request` + and called by :func:`make_create_handler` after body parsing + but before skill resolution / ``mgr.create``. Returns the + rejection response or ``None`` to continue. - Returns ``(attachment_ids, None)`` on success or ``(ids_saved_so_far, - JSONResponse)`` on the first failure so the caller can roll back any - partial state. + Gates: + - ws_id format must match :data:`_VALID_WS_ID` (32 hex chars) + when supplied. + - attachments + resume_ws combo is disallowed (resume forks an + existing ws; attachments belong on the *fresh* turn — caller + should resume first, then upload via the standard endpoint). + - body kind must be ``INTERACTIVE``: coordinator workstreams + land on the console handler with ``admin.coordinator`` scope, + not this one. Unknown / future kind values 400 rather than + silently coerce. + - parent_ws_id (when supplied) must reference a coordinator + owned by ``uid``. Without this gate an attacker could point a + new interactive workstream at someone else's coordinator and + receive that coordinator's child_ws_* SSE events + (name/state/tokens leak). + - notify_targets (when supplied) must validate. + :func:`_validate_notify_targets` is pure-read and doesn't need + ``ws`` to be built — gating here preserves pre-lift's 400 + semantic for caller-supplied input. Without this pre-create + gate a malformed ``notify_targets`` would land in + ``post_install`` (after ``mgr.create``, audit emit, and the + ``ws_created`` broadcast) and the only available signal is to + raise — which the factory turns into 500. 400 at the gate is + correct shape for client-input validation. """ - from turnstone.core.attachments import ( - IMAGE_SIZE_CAP, - MAX_PENDING_ATTACHMENTS_PER_USER_WS, - TEXT_DOC_SIZE_CAP, - classify_text_attachment, - sniff_image_mime, - upload_lock, - ) - from turnstone.core.memory import list_pending_attachments, save_attachment - - saved_ids: list[str] = [] - if not files: - return saved_ids, None - - lock = upload_lock(ws_id, user_id) - with lock: - pending_count = len(list_pending_attachments(ws_id, user_id)) - for filename, claimed_mime, data in files: - if not data: - return saved_ids, JSONResponse({"error": "Empty file"}, status_code=400) - sniffed_image = sniff_image_mime(data) - if sniffed_image is not None: - if len(data) > IMAGE_SIZE_CAP: - return saved_ids, JSONResponse( - { - "error": ( - f"Image too large ({len(data):,} bytes); " - f"cap is {IMAGE_SIZE_CAP:,} bytes." - ), - "code": "too_large", - }, - status_code=413, - ) - kind = "image" - mime = sniffed_image - else: - if len(data) > TEXT_DOC_SIZE_CAP: - return saved_ids, JSONResponse( - { - "error": ( - f"Text document too large ({len(data):,} bytes); " - f"cap is {TEXT_DOC_SIZE_CAP:,} bytes." - ), - "code": "too_large", - }, - status_code=413, - ) - mime_or_err = classify_text_attachment(filename, claimed_mime, data) - if mime_or_err[0] is None: - return saved_ids, JSONResponse( - {"error": mime_or_err[1], "code": "unsupported"}, - status_code=400, - ) - kind = "text" - mime = mime_or_err[0] - - if pending_count + 1 > MAX_PENDING_ATTACHMENTS_PER_USER_WS: - return saved_ids, JSONResponse( - { - "error": ( - f"Too many pending attachments " - f"(max {MAX_PENDING_ATTACHMENTS_PER_USER_WS} pending per workstream)" - ), - "code": "too_many", - }, - status_code=409, - ) - attachment_id = uuid.uuid4().hex - save_attachment( - attachment_id, - ws_id, - user_id, - filename, - mime, - len(data), - kind, - data, - ) - saved_ids.append(attachment_id) - pending_count += 1 - return saved_ids, None - - -async def create_workstream(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/new — create a new workstream. - - Accepts two content types: - - - ``application/json`` (default): body is a :class:`CreateWorkstreamRequest`. - - ``multipart/form-data``: one ``meta`` field (JSON object, same shape - as the JSON body) plus zero-or-more ``file`` parts. Files are saved - as attachments under the new workstream and reserved onto the first - ``initial_message`` turn (if provided) before the worker dispatches. - """ - from turnstone.core.attachments import IMAGE_SIZE_CAP - from turnstone.core.audit import record_audit - from turnstone.core.memory import get_workstream_display_name - from turnstone.core.web_helpers import ( - read_json_or_400, - read_multipart_create_or_400, - ) - - content_type = (request.headers.get("content-type") or "").lower() - uploaded_files: list[tuple[str, str, bytes]] = [] - body: dict[str, Any] - if content_type.startswith("multipart/form-data"): - # Multipart cap: up to MAX_PENDING × image cap, plus slack for - # JSON meta + multipart framing. Per-file size is enforced in - # _validate_and_save_uploaded_files against the kind-specific cap. - parsed = await read_multipart_create_or_400( - request, - max_files=10, - max_per_file_bytes=IMAGE_SIZE_CAP, - max_total_bytes=10 * IMAGE_SIZE_CAP, - ) - if isinstance(parsed, JSONResponse): - return parsed - body, uploaded_files = parsed - else: - json_body = await read_json_or_400(request) - if isinstance(json_body, JSONResponse): - return json_body - body = json_body - mgr: SessionManager = request.app.state.workstreams - skip: bool = request.app.state.skip_permissions - auth = getattr(getattr(request, "state", None), "auth_result", None) - uid: str = getattr(auth, "user_id", "") or "" - # Trusted services (console) may forward the real user_id in the request - # body when creating workstreams on behalf of a user. Only service - # identities are trusted — end-user tokens (including console-proxy tokens - # that carry the real user's identity) must not override user_id. - trusted_sources = {"console"} - if ( - body.get("user_id") - and isinstance(body["user_id"], str) - and auth is not None - and auth.token_source in trusted_sources - ): - uid = body["user_id"] - body_skill = body.get("skill", "") - resume_ws_id = body.get("resume_ws", "") - # Resolve skill — applies content + session config (model, temperature, etc.) - # Skip when resuming: the resumed session restores its own skill from config. - skill_data: dict[str, Any] | None = None - if body_skill and not resume_ws_id: - from turnstone.core.memory import get_skill_by_name - - skill_data = get_skill_by_name(body_skill) - if not skill_data or not skill_data.get("enabled", False): - return JSONResponse( - {"error": f"Skill not found or disabled: {body_skill}"}, - status_code=400, - ) - resolved_model = body.get("model") or None - if skill_data and skill_data.get("model"): - resolved_model = skill_data["model"] - resolved_skill: str | None = body_skill if skill_data else None - applied_skill_version = 0 - if skill_data: - from turnstone.core.storage import get_storage as _get_storage - - _st = _get_storage() - applied_skill_version = len(_st.list_skill_versions(skill_data["template_id"])) + 1 requested_ws_id = body.get("ws_id", "") or "" if not isinstance(requested_ws_id, str): requested_ws_id = "" if requested_ws_id and not _VALID_WS_ID.match(requested_ws_id): return JSONResponse({"error": "invalid ws_id format"}, status_code=400) - # Disallow attachments + resume_ws in the same request — semantics - # are unclear (resume forks an existing ws, but attachments are for - # the *fresh* turn). Caller should resume first, then upload via - # the standard endpoint. Checked before mgr.create() so we don't - # waste work on a request we'll reject. + resume_ws_id = body.get("resume_ws", "") or "" if uploaded_files and resume_ws_id: return JSONResponse( {"error": "attachments cannot be combined with resume_ws"}, status_code=400, ) - # Kind + parent relationship: coordinator-spawned children forward - # these from the console's CoordinatorClient. Default kind is - # ``INTERACTIVE``; coordinators themselves are created by the console's - # own SessionManager (coordinator kind) and never land in this handler. - # Only - # ``INTERACTIVE`` is accepted here — requests that try to create a - # coordinator via the generic workstream endpoint are rejected, and - # unknown kinds (typos, future-only values) are 400 rather than being - # silently coerced. User-facing edge: the storage_edge and manager - # layers below re-validate, see comments at those sites for why. try: body_kind = WorkstreamKind.from_raw(body.get("kind")) except ValueError: @@ -2235,16 +2082,6 @@ async def create_workstream(request: Request) -> JSONResponse: ) body_parent = body.get("parent_ws_id") or None if body_parent is not None: - # Ownership gate: parent_ws_id is client-supplied in the request - # body, so a malicious caller could previously point a new - # interactive workstream at another tenant's coordinator — the - # coordinator's SSE fan-out would then route child_ws_* events - # (carrying the attacker's name/state/tokens) to that victim. - # Validate against storage: the parent must exist, be a - # coordinator, and belong to the same user. Coordinator-spawned - # children satisfy this by construction (the coordinator's JWT - # `sub` claim is the owning user and parent_ws_id is the coord's - # own ws_id); external clients that fabricate the field get 403. from turnstone.core.storage._registry import get_storage as _get_storage_for_parent _pstorage = _get_storage_for_parent() @@ -2262,251 +2099,295 @@ async def create_workstream(request: Request) -> JSONResponse: {"error": "parent_ws_id must reference a coordinator you own"}, status_code=403, ) - try: - ws = mgr.create( - user_id=uid, - name=body.get("name", ""), - model=resolved_model, - skill=resolved_skill, - skill_id=skill_data["template_id"] if skill_data else "", - skill_version=applied_skill_version, - ws_id=requested_ws_id, - client_type=body.get("client_type", "") or "", - judge_model=body.get("judge_model", "") or None, - parent_ws_id=body_parent, - ) - if not isinstance(ws.ui, WebUI): - raise TypeError(f"Expected WebUI, got {type(ws.ui).__name__}") - if skip or body.get("auto_approve", False): - ws.ui.auto_approve = True - # Register watch runner for this workstream - runner = getattr(request.app.state, "watch_runner", None) - if runner and ws.session: - ws.session.set_watch_runner( - runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui) - ) - gq: queue.Queue[dict[str, Any]] = request.app.state.global_queue + notify_targets_raw = body.get("notify_targets", "[]") + if isinstance(notify_targets_raw, list): + notify_targets_raw = json.dumps(notify_targets_raw) + _, nt_err = _validate_notify_targets(notify_targets_raw) + if nt_err: + return JSONResponse({"error": nt_err}, status_code=400) + return None - # Save attachments BEFORE the ws_created broadcast so failed - # validation doesn't make SSE consumers flash a workstream that - # never really existed. Validate + save happens early; rollback - # is silent (no ws_created → no ws_closed needed). - attachment_ids: list[str] = [] - if uploaded_files: - saved_ids, save_err = _validate_and_save_uploaded_files(uploaded_files, ws.id, uid) - if save_err is not None: - from turnstone.core.memory import delete_workstream as _delete_ws - with contextlib.suppress(Exception): - mgr.close(ws.id) - with contextlib.suppress(Exception): - _delete_ws(ws.id) - return save_err - attachment_ids = saved_ids +def _interactive_create_build_kwargs( + request: Request, + body: dict[str, Any], + uid: str, + skill_data: dict[str, Any] | None, + skill_id: str, + applied_skill_version: int, +) -> dict[str, Any]: + """Build kwargs for ``mgr.create`` from a parsed interactive create body. - # Emit creation event on global queue for SSE consumers (console). - # Deferred until past attachment validation so a rejected create - # doesn't surface a phantom create→close pair. - display_name = get_workstream_display_name(ws.id) or ws.name - with contextlib.suppress(queue.Full): - gq.put_nowait( - { - "type": "ws_created", - "ws_id": ws.id, - "name": display_name, - "model": ws.session.model if ws.session else "", - "model_alias": ws.session.model_alias if ws.session else "", - "kind": ws.kind, - "parent_ws_id": ws.parent_ws_id, - # Owner id is propagated through the cluster event - # stream so console-side fan-out can enforce tenant - # isolation — a coordinator must never receive - # child_ws_* events for workstreams it doesn't own. - "user_id": ws.user_id, - } - ) - # Tamper-evident audit trail. Lives alongside the broadcast - # event (not replacing it — the broadcast is ephemeral UI - # signalling, the audit row survives for forensic review). - _audit_storage = getattr(request.app.state, "auth_storage", None) - if _audit_storage is not None: - _, _audit_ip = _audit_context(request) - record_audit( - _audit_storage, - uid, - "workstream.created", - "workstream", - ws.id, - {"kind": str(ws.kind), "parent_ws_id": ws.parent_ws_id}, - _audit_ip, - ) - # Atomic workstream resume during creation. - resumed = False - message_count = 0 - if resume_ws_id and ws.session is not None: - from turnstone.core.memory import resolve_workstream + Wired onto :attr:`SessionEndpointConfig.create_build_kwargs`. The + factory threads the resolved skill_data + skill_id + version + through; this builder picks the right model (skill override + beats body) and assembles the full kwargs dict that + ``SessionManager.create`` accepts (including the kind-specific + ``judge_model`` / ``client_type`` / ``parent_ws_id`` extras). + """ + resolved_model = body.get("model") or None + if skill_data and skill_data.get("model"): + resolved_model = skill_data["model"] + requested_ws_id = body.get("ws_id", "") or "" + if not isinstance(requested_ws_id, str): + requested_ws_id = "" + return { + "user_id": uid, + "name": body.get("name", ""), + "model": resolved_model, + "skill": body.get("skill", "") if skill_data else None, + "skill_id": skill_id, + "skill_version": applied_skill_version, + "ws_id": requested_ws_id, + "client_type": body.get("client_type", "") or "", + "judge_model": body.get("judge_model", "") or None, + "parent_ws_id": body.get("parent_ws_id") or None, + } - target_id = resolve_workstream(resume_ws_id) - if target_id and ws.session.resume(target_id, fork=True): - resumed = True - message_count = len(ws.session.messages) - # If the user provided a custom name, set it as the fork's alias - # so it takes priority in display. Otherwise keep the - # auto-generated name so auto-title can run fresh. - user_name = body.get("name", "").strip() - if user_name: - from turnstone.core.memory import set_workstream_alias - set_workstream_alias(ws.id, user_name) - ws.name = user_name - ui = ws.ui - if isinstance(ui, WebUI): - ui._enqueue({"type": "clear_ui"}) - history = _build_history(ws.session) - if history: - ui._enqueue({"type": "history", "messages": history}) - # Broadcast a rename so the tab picks up the correct fork name - # (the ws_created event fired before fork with the pre-fork name). - with contextlib.suppress(queue.Full): - gq.put_nowait({"type": "ws_rename", "ws_id": ws.id, "name": ws.name}) +async def _interactive_create_post_install( + request: Request, + ws: Workstream, + body: dict[str, Any], + uid: str, + skill_data: dict[str, Any] | None, + applied_skill_version: int, + attachment_ids: list[str], +) -> dict[str, Any]: + """Tail end of interactive create: per-WebUI bookkeeping + dispatch. - # Apply skill session config (only for new workstreams with a skill) - if skill_data and not resumed and ws.session: - sess = ws.session - # Session settings from skill - if skill_data.get("temperature") is not None: - sess.temperature = skill_data["temperature"] - if skill_data.get("reasoning_effort"): - sess.reasoning_effort = skill_data["reasoning_effort"] - if skill_data.get("max_tokens") is not None: - sess.max_tokens = skill_data["max_tokens"] - if skill_data.get("token_budget", 0) > 0: - sess._token_budget = skill_data["token_budget"] - if skill_data.get("agent_max_turns") is not None: - sess.agent_max_turns = skill_data["agent_max_turns"] - # Approval policy - if skill_data.get("auto_approve"): - ws.ui.auto_approve = True - allowed = skill_data.get("allowed_tools", "") - if allowed and allowed != "[]": - # Parse as JSON array or comma-separated - import json as _json + Wired onto :attr:`SessionEndpointConfig.create_post_install`. + Runs after the workstream is fully built, attachments saved, + and audit emitted. Sequence: - try: - tools_list = _json.loads(allowed) - except (ValueError, TypeError): - tools_list = [t.strip() for t in allowed.split(",") if t.strip()] - if tools_list: - ws.ui.auto_approve_tools = set(tools_list) - # Metadata - sess._notify_on_complete = skill_data.get("notify_on_complete", "{}") - sess._applied_skill_id = skill_data["template_id"] - sess._applied_skill_version = applied_skill_version - if skill_data.get("content"): - sess._applied_skill_content = skill_data["content"] - sess._save_config() + 1. Cast ``ws.ui`` to :class:`WebUI` (defence in depth — the + interactive adapter's session factory is the only path that + reaches this handler). + 2. Apply ``auto_approve`` from server-wide ``skip_permissions`` + or per-request body. + 3. Register the watch runner for the workstream's session. + 4. Broadcast ``ws_created`` on the global SSE queue. Held until + this point so a rejected attachment validation produces no + phantom create→close pair on the SSE stream. + 5. Atomic resume: if ``body["resume_ws"]`` is set, fork the + referenced session into the new ws_id, push history into the + UI listener queue, and rebroadcast ``ws_rename`` so the tab + picks up the fork's display name. + 6. Apply the skill's session config (temperature / reasoning / + max_tokens / approval policy / metadata). + 7. Resolve notify_targets (schedule targets win over skill + fallback). + 8. Pin the workstream's routing to this node when no caller- + supplied ``ws_id`` was provided (direct creates). + 9. Spawn the initial-message worker thread when ``initial_message`` + is set, reserving any uploaded attachments for that first + turn. - # Resolve notify_targets: schedule targets override skill targets - notify_targets_raw = body.get("notify_targets", "[]") - if isinstance(notify_targets_raw, list): - notify_targets_raw = json.dumps(notify_targets_raw) - nt_str, nt_err = _validate_notify_targets(notify_targets_raw) - if nt_err: - return JSONResponse({"error": nt_err}, status_code=400) - # Skill fallback (only if schedule didn't specify targets) - if nt_str == "[]" and skill_data: - skill_notify = skill_data.get("notify_on_complete", "[]") - if skill_notify and skill_notify != "{}" and skill_notify != "[]": - fallback_str, fallback_err = _validate_notify_targets(skill_notify) - if not fallback_err: - nt_str = fallback_str - ws.notify_targets = nt_str + Returns ``{resumed, message_count}`` for the response. On the + no-resume path both default to ``False`` / ``0``. + """ + from turnstone.core.memory import get_workstream_display_name - # Pin locally-created workstreams so the console routes to this node. - # Console-routed creates pass ws_id in the request body — those are - # already bucket-aligned and don't need an override. Direct creates - # (web UI, watch, TurnstoneInit) generate their own ws_id, which may - # hash to a bucket assigned to a different node. - if not requested_ws_id: - node_id = getattr(request.app.state, "node_id", "") - if node_id: - try: - from turnstone.core.storage import get_storage as _gs - - _gs().set_workstream_override(ws.id, node_id, reason="local") - except Exception: - log.debug("Failed to set routing override for %s", ws.id, exc_info=True) - - # If an initial_message was provided, send it as the first user message. - # This replaces the old bridge behavior where CreateWorkstreamMessage - # carried initial_message and the bridge sent it as a follow-up. - initial_message = body.get("initial_message", "").strip() - if initial_message and ws.session is not None: - session = ws.session - # Reserve any attachments uploaded in this request before the - # worker dispatches. Mirrors the /send endpoint pattern: the - # send_id token scopes both the reservation and the eventual - # consume. Unreserve on worker failure so the rows don't stay - # soft-locked forever. - send_id = uuid.uuid4().hex - resolved_atts: list[Any] = [] - if attachment_ids: - resolved_atts, _ord, _drop = _reserve_and_resolve_attachments( - attachment_ids, send_id, ws.id, uid - ) - - def _run_initial() -> None: - try: - session.send( - initial_message, - attachments=resolved_atts or None, - send_id=send_id if resolved_atts else None, - ) - except (Exception, GenerationCancelled): - if attachment_ids: - from turnstone.core.memory import ( - unreserve_attachments as _unreserve, - ) - - with contextlib.suppress(Exception): - _unreserve(send_id, ws.id, uid) - if isinstance(ws.ui, WebUI): - ws.ui.on_stream_end() - ws.ui.on_state_change("idle") - finally: - try: - last_content = _extract_last_assistant_content(session) - _fire_notify_targets(ws, last_content) - except Exception: - log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True) - with ws._lock: - ws._worker_running = False - - # Inlined rather than via ``session_worker.send`` because - # at workstream creation no live worker can exist by - # construction — the enqueue branch of the shared dispatch - # is dead code here. We still set ``_worker_running`` and - # ``ws.worker_thread`` together under ``ws._lock`` so a - # /v1/api/send arriving immediately after creation observes - # the running state via the shared session_worker gate - # instead of racing into a parallel worker. - with ws._lock: - ws._worker_running = True - t = threading.Thread(target=_run_initial, daemon=True, name=f"ws-init-{ws.id[:8]}") - ws.worker_thread = t - t.start() - - return JSONResponse( + if not isinstance(ws.ui, WebUI): + raise TypeError(f"Expected WebUI, got {type(ws.ui).__name__}") + skip: bool = request.app.state.skip_permissions + if skip or body.get("auto_approve", False): + ws.ui.auto_approve = True + runner = getattr(request.app.state, "watch_runner", None) + if runner and ws.session: + ws.session.set_watch_runner(runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui)) + gq: queue.Queue[dict[str, Any]] = request.app.state.global_queue + # Emit ``ws_created`` on the global queue for SSE consumers + # (console). Held until past attachment validation in the + # factory so a rejected upload doesn't flash a workstream that + # never really existed. + display_name = get_workstream_display_name(ws.id) or ws.name + with contextlib.suppress(queue.Full): + gq.put_nowait( { + "type": "ws_created", "ws_id": ws.id, - "name": ws.name, - "resumed": resumed, - "message_count": message_count, - "attachment_ids": attachment_ids, + "name": display_name, + "model": ws.session.model if ws.session else "", + "model_alias": ws.session.model_alias if ws.session else "", + "kind": ws.kind, + "parent_ws_id": ws.parent_ws_id, + # Owner id propagates through the cluster event + # stream so console-side fan-out can enforce tenant + # isolation — a coordinator must never receive + # child_ws_* events for workstreams it doesn't own. + "user_id": ws.user_id, } ) - except RuntimeError as e: - return JSONResponse({"error": str(e)}, status_code=400) + + # Atomic workstream resume during creation. + resumed = False + message_count = 0 + resume_ws_id = body.get("resume_ws", "") or "" + if resume_ws_id and ws.session is not None: + from turnstone.core.memory import resolve_workstream + + target_id = resolve_workstream(resume_ws_id) + if target_id and ws.session.resume(target_id, fork=True): + resumed = True + message_count = len(ws.session.messages) + user_name = body.get("name", "").strip() + if user_name: + from turnstone.core.memory import set_workstream_alias + + set_workstream_alias(ws.id, user_name) + ws.name = user_name + ui = ws.ui + if isinstance(ui, WebUI): + ui._enqueue({"type": "clear_ui"}) + history = _build_history(ws.session) + if history: + ui._enqueue({"type": "history", "messages": history}) + with contextlib.suppress(queue.Full): + gq.put_nowait({"type": "ws_rename", "ws_id": ws.id, "name": ws.name}) + + # Apply skill session config (only for new workstreams with a skill). + if skill_data and not resumed and ws.session: + sess = ws.session + if skill_data.get("temperature") is not None: + sess.temperature = skill_data["temperature"] + if skill_data.get("reasoning_effort"): + sess.reasoning_effort = skill_data["reasoning_effort"] + if skill_data.get("max_tokens") is not None: + sess.max_tokens = skill_data["max_tokens"] + if skill_data.get("token_budget", 0) > 0: + sess._token_budget = skill_data["token_budget"] + if skill_data.get("agent_max_turns") is not None: + sess.agent_max_turns = skill_data["agent_max_turns"] + if skill_data.get("auto_approve"): + ws.ui.auto_approve = True + allowed = skill_data.get("allowed_tools", "") + if allowed and allowed != "[]": + import json as _json + + try: + tools_list = _json.loads(allowed) + except (ValueError, TypeError): + tools_list = [t.strip() for t in allowed.split(",") if t.strip()] + if tools_list: + ws.ui.auto_approve_tools = set(tools_list) + sess._notify_on_complete = skill_data.get("notify_on_complete", "{}") + sess._applied_skill_id = skill_data["template_id"] + sess._applied_skill_version = applied_skill_version + if skill_data.get("content"): + sess._applied_skill_content = skill_data["content"] + sess._save_config() + + # notify_targets: schedule targets override skill targets. The + # validator already gated malformed input as 400; here we just + # canonicalise (list → JSON-encoded string) and apply the skill + # fallback if the caller didn't supply targets. + notify_targets_raw = body.get("notify_targets", "[]") + if isinstance(notify_targets_raw, list): + notify_targets_raw = json.dumps(notify_targets_raw) + nt_str, _ = _validate_notify_targets(notify_targets_raw) + if nt_str == "[]" and skill_data: + skill_notify = skill_data.get("notify_on_complete", "[]") + if skill_notify and skill_notify != "{}" and skill_notify != "[]": + fallback_str, fallback_err = _validate_notify_targets(skill_notify) + if not fallback_err: + nt_str = fallback_str + ws.notify_targets = nt_str + + # Pin locally-created workstreams so the console routes to this node. + requested_ws_id = body.get("ws_id", "") or "" + if not requested_ws_id: + node_id = getattr(request.app.state, "node_id", "") + if node_id: + try: + from turnstone.core.storage import get_storage as _gs + + _gs().set_workstream_override(ws.id, node_id, reason="local") + except Exception: + log.debug("Failed to set routing override for %s", ws.id, exc_info=True) + + # Initial-message worker thread. + initial_message = body.get("initial_message", "").strip() + if initial_message and ws.session is not None: + session = ws.session + send_id = uuid.uuid4().hex + resolved_atts: list[Any] = [] + if attachment_ids: + resolved_atts, _ord, _drop = _reserve_and_resolve_attachments( + attachment_ids, send_id, ws.id, uid + ) + + def _run_initial() -> None: + try: + session.send( + initial_message, + attachments=resolved_atts or None, + send_id=send_id if resolved_atts else None, + ) + except (Exception, GenerationCancelled): + if attachment_ids: + from turnstone.core.memory import ( + unreserve_attachments as _unreserve, + ) + + with contextlib.suppress(Exception): + _unreserve(send_id, ws.id, uid) + if isinstance(ws.ui, WebUI): + ws.ui.on_stream_end() + ws.ui.on_state_change("idle") + finally: + try: + last_content = _extract_last_assistant_content(session) + _fire_notify_targets(ws, last_content) + except Exception: + log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True) + with ws._lock: + ws._worker_running = False + + # Inlined rather than via ``session_worker.send`` because at + # workstream creation no live worker can exist by + # construction — the enqueue branch of the shared dispatch + # is dead code here. ``_worker_running`` + ``ws.worker_thread`` + # are set together under ``ws._lock`` so a /v1/api/send + # arriving immediately after creation observes the running + # state via the shared session_worker gate instead of racing + # into a parallel worker. + with ws._lock: + ws._worker_running = True + t = threading.Thread(target=_run_initial, daemon=True, name=f"ws-init-{ws.id[:8]}") + ws.worker_thread = t + t.start() + + return {"resumed": resumed, "message_count": message_count} + + +def _audit_workstream_created( + request: Request, + ws: Workstream, + body: dict[str, Any], + uid: str, +) -> None: + """Audit emitter for the interactive ``workstream.created`` event. + + Wired onto :func:`make_create_handler` as ``audit_emit``. Runs + after the workstream is built and attachments saved; failures + are caught + logged by the factory (HTTP response stays 201). + """ + from turnstone.core.audit import record_audit + + _audit_storage = getattr(request.app.state, "auth_storage", None) + if _audit_storage is None: + return + _, _audit_ip = _audit_context(request) + record_audit( + _audit_storage, + uid, + "workstream.created", + "workstream", + ws.id, + {"kind": str(ws.kind), "parent_ws_id": ws.parent_ws_id}, + _audit_ip, + ) async def delete_workstream_endpoint(request: Request) -> JSONResponse: @@ -3814,6 +3695,11 @@ def create_app( # the lifted contract — coord wires ``None`` and falls # back to the default executor. sse_executor_lookup=lambda request: request.app.state.sse_executor, + create_supports_attachments=True, + create_supports_user_id_override=True, + create_validate_request=_interactive_create_validate_request, + create_build_kwargs=_interactive_create_build_kwargs, + create_post_install=_interactive_create_post_install, ) approve_handler = make_approve_handler(interactive_endpoint_config) close_handler = make_close_handler( @@ -3830,6 +3716,10 @@ def create_app( send_handler = make_send_handler(interactive_endpoint_config) dequeue_handler = make_dequeue_handler(interactive_endpoint_config) attachment_handlers = make_attachment_handlers(interactive_endpoint_config) + create_handler = make_create_handler( + interactive_endpoint_config, + audit_emit=_audit_workstream_created, + ) v1_routes: list[Any] = [ Route("/api/events", make_legacy_query_keyed_adapter(events_handler)), Route("/api/events/global", global_events_sse), @@ -3840,7 +3730,7 @@ def create_app( handlers=SharedSessionVerbHandlers( list_workstreams=list_workstreams, list_saved=list_saved_workstreams, - create=create_workstream, + create=create_handler, # lifted: shared body close_legacy=make_legacy_body_keyed_adapter(close_handler), delete=delete_workstream_endpoint, open=open_handler, # lifted: shared body