From e0c78e2aec1d98ca12aa689de9edca6b1a274aa2 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Sat, 25 Apr 2026 20:49:05 -0700 Subject: [PATCH] test,docs: coord attachment + queue parity tests + spec regen + CHANGELOG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five new TestCoordinatorAttachments tests in ``tests/test_coordinator_endpoints.py`` exercising the lifted attachment surface end-to-end on coord: * upload → list round-trip * get_content returns raw bytes with text/plain forced for text * delete removes pending entries and clears them from the listing * send with attachment_ids consumes pending under the send_id token * send response carries attached_ids / dropped_attachment_ids even on plain-text sends (unified shape parity) The existing ``_coord_endpoint_config`` fixture grew capability flags to mirror the production console wiring, and ``_make_client`` now mounts the four coord attachment routes via ``make_attachment_handlers``. OpenAPI specs regenerated; TS SDK bumped to 0.5.0. CHANGELOG entry under [Unreleased] documents the verb-shape lift, the coord attachment surface coming online, the response-shape change for ``coordinator_send``, the unification of the three lifted classifier / lock helpers under ``turnstone.core.attachments``, and the new SDK helpers. --- CHANGELOG.md | 50 ++++ sdk/typescript/openapi-console.json | 364 +++++++++++++++++++++++++++- tests/test_coordinator_endpoints.py | 165 ++++++++++++- 3 files changed, 571 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8024b0a9..464900d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -85,6 +85,56 @@ Three release tracks are maintained: fork ship into the stable line bakes the duplication in for the lifetime of the 1.5 track. +- **`/send` body lift + coordinator attachments + queue surface + parity** ([Stage 2 Priority 1.5]). The ``/send`` HTTP handler is + now ONE factory body (``make_send_handler(cfg)``) wired with + capability flags on both kinds; the four attachment endpoints + (``upload`` / ``list`` / ``get_content`` / ``delete``) are also + unified via ``make_attachment_handlers(cfg)``. Coord workstreams + light up: + + - ``POST/GET /v1/api/workstreams/{ws_id}/attachments``, + ``GET .../attachments/{aid}/content``, + ``DELETE .../attachments/{aid}`` — same shape, same caps, same + reservation flow as interactive. + - ``POST /v1/api/workstreams/{ws_id}/send`` accepts + ``attachment_ids`` (or auto-consumes pending) and returns + ``attached_ids`` / ``dropped_attachment_ids`` for surfacing + partial reservations. Live-worker reuse path also returns + ``priority`` / ``msg_id`` (parity with the interactive + ``status: queued`` shape). + + Backend parity is end-to-end: storage layer was already + kind-agnostic; the route registrar's ``AttachmentHandlers`` slot + has been there since Stage 2 P0; the multi-node attachment + routing-proxy on the console (``route_attachment_proxy``) was + already shipping. P1.5 is the wiring + verb-shape lift that lets + these primitives surface on the coord side. + + Coord dashboard rendering surfaces an attachment-count badge on + past messages with attachments; full chip rendering with + click-to-view is deferred (the coord dashboard is + diagnostic-leaning and chip parity isn't on the critical path + for the unification thesis). Python SDK adds + ``coordinator_send`` / ``coordinator_upload_attachment`` / + ``coordinator_list_attachments`` / + ``coordinator_get_attachment_content`` / + ``coordinator_delete_attachment`` on + ``AsyncTurnstoneConsole`` + ``TurnstoneConsole``. TS SDK + regenerated; bumped to 0.5.0. + + Three lifted helpers (``sniff_image_mime``, + ``classify_text_attachment``, ``upload_lock``) moved from + ``turnstone/server.py`` to ``turnstone/core/attachments.py`` so + both processes use the canonical implementation. The interactive + surface keeps the same behaviour; the helpers are simply + imported from their new home. + + ``coordinator_send`` no longer returns ``429`` on a full worker + queue — the unified body returns ``200 {"status": "queue_full"}`` + for parity with interactive. Existing callers checking for ``429`` + should switch to the status-code shape. + - **Workstream state writes are now buffered through ``StateWriter``.** ``SessionManager.set_state`` no longer holds ``ws._lock`` across a synchronous Postgres ``UPDATE`` for non-terminal transitions; diff --git a/sdk/typescript/openapi-console.json b/sdk/typescript/openapi-console.json index c72c9d1b..e48fae67 100644 --- a/sdk/typescript/openapi-console.json +++ b/sdk/typescript/openapi-console.json @@ -4802,7 +4802,7 @@ "tags": [ "Coordinator" ], - "description": "Worker thread picks up the message via the session's queue. ``429`` when the worker queue is full \u2014 caller should back off.", + "description": "Worker thread picks up the message via the session's queue. Optional ``attachment_ids`` reserve attachments under the message's send_id token (parity with the interactive surface). Response carries ``attached_ids`` / ``dropped_attachment_ids`` so callers can detect partial reservations and ``priority`` / ``msg_id`` on the queued path. ``status: queue_full`` when the worker queue is full \u2014 caller should back off.", "parameters": [ { "name": "ws_id", @@ -4829,7 +4829,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/StatusResponse" + "$ref": "#/components/schemas/CoordinatorSendResponse" } } } @@ -4864,8 +4864,8 @@ } } }, - "429": { - "description": "Error 429", + "500": { + "description": "Error 500", "content": { "application/json": { "schema": { @@ -4874,8 +4874,282 @@ } } }, - "500": { - "description": "Error 500", + "503": { + "description": "Error 503", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + } + }, + "/v1/api/workstreams/{ws_id}/attachments": { + "post": { + "summary": "Upload a file attachment to a coordinator workstream", + "operationId": "v1_api_workstreams_{ws_id}_attachments_post", + "tags": [ + "Coordinator" + ], + "description": "Multipart upload (field ``file``). Same validation rules as the interactive surface: magic-byte image sniff, UTF-8 text decode, per-kind size cap, per-(ws,user) pending cap. Attachments stay pending until a subsequent ``/send`` reserves them under its ``send_id`` token.", + "parameters": [ + { + "name": "ws_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UploadAttachmentResponse" + } + } + } + }, + "400": { + "description": "Error 400", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "403": { + "description": "Error 403", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "404": { + "description": "Error 404", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "409": { + "description": "Error 409", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "413": { + "description": "Error 413", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "503": { + "description": "Error 503", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + }, + "get": { + "summary": "List the caller's pending coordinator attachments", + "operationId": "v1_api_workstreams_{ws_id}_attachments_get", + "tags": [ + "Coordinator" + ], + "parameters": [ + { + "name": "ws_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ListAttachmentsResponse" + } + } + } + }, + "403": { + "description": "Error 403", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "404": { + "description": "Error 404", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "503": { + "description": "Error 503", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + } + }, + "/v1/api/workstreams/{ws_id}/attachments/{attachment_id}/content": { + "get": { + "summary": "Return raw bytes of a coordinator attachment", + "operationId": "v1_api_workstreams_{ws_id}_attachments_{attachment_id}_content_get", + "tags": [ + "Coordinator" + ], + "description": "Same byte-stream + headers as the interactive surface. Text kinds are forced to ``text/plain`` so an HTML-shaped text upload can't render same-origin.", + "parameters": [ + { + "name": "ws_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "attachment_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Success" + }, + "403": { + "description": "Error 403", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "404": { + "description": "Error 404", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "503": { + "description": "Error 503", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + } + }, + "/v1/api/workstreams/{ws_id}/attachments/{attachment_id}": { + "delete": { + "summary": "Remove a pending coordinator attachment", + "operationId": "v1_api_workstreams_{ws_id}_attachments_{attachment_id}_delete", + "tags": [ + "Coordinator" + ], + "description": "Consumed attachments return 404.", + "parameters": [ + { + "name": "ws_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "attachment_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StatusResponse" + } + } + } + }, + "403": { + "description": "Error 403", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "404": { + "description": "Error 404", "content": { "application/json": { "schema": { @@ -7372,6 +7646,22 @@ "description": "User message to queue onto the coordinator's worker.", "title": "Message", "type": "string" + }, + "attachment_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Explicit list of attachment ids to inject into this turn. When omitted, any pending attachments for the caller on this coordinator are auto-consumed. An empty list disables auto-consumption for this send.", + "title": "Attachment Ids" } }, "required": [ @@ -7380,6 +7670,68 @@ "title": "CoordinatorSendRequest", "type": "object" }, + "CoordinatorSendResponse": { + "description": "Response shape for POST /v1/api/workstreams/{ws_id}/send (coord).", + "properties": { + "status": { + "description": "'ok' (fresh worker spawned), 'queued' (live worker reuse), or 'queue_full'.", + "examples": [ + "ok", + "queued", + "queue_full" + ], + "title": "Status", + "type": "string" + }, + "attached_ids": { + "description": "Attachment ids actually reserved onto this turn. Subset of the request's `attachment_ids` (or the auto-consumed pending set). Empty when the send carries no attachments.", + "items": { + "type": "string" + }, + "title": "Attached Ids", + "type": "array" + }, + "dropped_attachment_ids": { + "description": "Attachment ids the caller requested that the server could not reserve (lost a race, already consumed, or cross-scope). The request still proceeds with whatever was reserved.", + "items": { + "type": "string" + }, + "title": "Dropped Attachment Ids", + "type": "array" + }, + "priority": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Set on `queued` responses: relative priority of the queued message.", + "title": "Priority" + }, + "msg_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Set on `queued` responses: id used to dequeue the message.", + "title": "Msg Id" + } + }, + "required": [ + "status" + ], + "title": "CoordinatorSendResponse", + "type": "object" + }, "CoordinatorStopCascadeResponse": { "description": "Response body for POST /v1/api/workstreams/{ws_id}/stop_cascade.", "properties": { diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index 3d0f0130..a665b771 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -43,22 +43,41 @@ from turnstone.console.server import ( coordinator_list, coordinator_open, coordinator_saved, - coordinator_send, coordinator_tasks, ) +from turnstone.core.attachments import ( + classify_text_attachment as _coord_test_classify_text, +) +from turnstone.core.attachments import ( + sniff_image_mime as _coord_test_sniff_image, +) +from turnstone.core.attachments import ( + upload_lock as _coord_test_upload_lock, +) from turnstone.core.auth import AuthResult from turnstone.core.session_routes import ( + AttachmentUploadHelpers, SessionEndpointConfig, make_approve_handler, + make_attachment_handlers, make_close_handler, + make_send_handler, ) from turnstone.core.storage._sqlite import SQLiteBackend +from turnstone.core.web_helpers import resolve_workstream_owner # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- +def _coord_attach_owner(request, ws_id, mgr): + """Coord attachment owner resolver mirroring production wiring.""" + return resolve_workstream_owner( + request, ws_id, mgr=mgr, not_found_label="coordinator not found" + ) + + # Per-kind config the lifted handler factories capture by closure. # Mirrors the production console wiring so tests exercise the same # code path as the live server. @@ -68,6 +87,15 @@ _coord_endpoint_config = SessionEndpointConfig( tenant_check=None, not_found_label="coordinator not found", audit_action_prefix="coordinator", + supports_attachments=True, + attachment_owner_resolver=_coord_attach_owner, + attachment_helpers=AttachmentUploadHelpers( + sniff_image_mime=_coord_test_sniff_image, + classify_text_attachment=_coord_test_classify_text, + upload_lock=_coord_test_upload_lock, + ), + spawn_metrics=None, + emit_message_queued=True, ) @@ -84,6 +112,7 @@ def _make_client( registry=None, ) -> TestClient: """Build a TestClient exposing just the coordinator routes.""" + coord_attachments = make_attachment_handlers(_coord_endpoint_config) app = Starlette( routes=[ Route( @@ -101,7 +130,7 @@ def _make_client( ), Route( "/v1/api/workstreams/{ws_id}/send", - coordinator_send, + make_send_handler(_coord_endpoint_config), methods=["POST"], ), Route( @@ -143,6 +172,26 @@ def _make_client( coordinator_tasks, methods=["GET"], ), + Route( + "/v1/api/workstreams/{ws_id}/attachments", + coord_attachments.upload, + methods=["POST"], + ), + Route( + "/v1/api/workstreams/{ws_id}/attachments", + coord_attachments.list, + methods=["GET"], + ), + Route( + "/v1/api/workstreams/{ws_id}/attachments/{attachment_id}/content", + coord_attachments.get_content, + methods=["GET"], + ), + Route( + "/v1/api/workstreams/{ws_id}/attachments/{attachment_id}", + coord_attachments.delete, + methods=["DELETE"], + ), Route( "/v1/api/workstreams/{ws_id}", coordinator_detail, @@ -1210,3 +1259,115 @@ def test_coordinator_rows_persisted_cluster_wide(storage): request = _persisted_rows_request(storage, mgr, caller, perms) rows = _coordinator_rows(request) assert {r["name"] for r in rows} == {"alice-closed", "bob-closed", "orphan-closed"} + + +# --------------------------------------------------------------------------- +# Stage 2 P1.5 — coord attachment surface parity with interactive +# --------------------------------------------------------------------------- + + +class TestCoordinatorAttachments: + """The lifted ``make_attachment_handlers`` factory exposes + upload / list / get_content / delete on coord workstreams using + the same kind-agnostic storage layer interactive uses. These + tests exercise the surface end-to-end via TestClient.""" + + def _upload(self, client, ws_id, *, name="hello.md", body=b"hi", mime="text/markdown"): + files = {"file": (name, body, mime)} + resp = client.post( + f"/v1/api/workstreams/{ws_id}/attachments", + files=files, + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200, resp.text + return resp.json() + + def test_upload_round_trip_lists_pending(self, storage): + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1", name="c1") + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + + info = self._upload(client, ws.id, name="note.md", body=b"hello world") + assert info["filename"] == "note.md" + assert info["kind"] == "text" + assert info["size_bytes"] == len(b"hello world") + + listing = client.get(f"/v1/api/workstreams/{ws.id}/attachments", headers=_COORD_HEADERS) + assert listing.status_code == 200 + ids = [a["attachment_id"] for a in listing.json()["attachments"]] + assert info["attachment_id"] in ids + + def test_get_content_returns_raw_bytes(self, storage): + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1", name="c1") + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + + info = self._upload( + client, ws.id, name="data.json", body=b'{"k":1}', mime="application/json" + ) + resp = client.get( + f"/v1/api/workstreams/{ws.id}/attachments/{info['attachment_id']}/content", + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200 + # Text kinds force text/plain to avoid same-origin HTML/SVG rendering. + assert resp.headers["content-type"].startswith("text/plain") + assert resp.content == b'{"k":1}' + + def test_delete_removes_pending(self, storage): + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1", name="c1") + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + + info = self._upload(client, ws.id) + resp = client.delete( + f"/v1/api/workstreams/{ws.id}/attachments/{info['attachment_id']}", + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200 + assert resp.json() == {"status": "deleted"} + + listing = client.get(f"/v1/api/workstreams/{ws.id}/attachments", headers=_COORD_HEADERS) + ids = [a["attachment_id"] for a in listing.json()["attachments"]] + assert info["attachment_id"] not in ids + + def test_send_with_attachment_ids_consumes_pending(self, storage): + """End-to-end: upload an attachment, then ``coord_send`` it. The + reservation flips ``reserved_for_msg_id`` to the send_id, so the + attachment is no longer in the pending listing.""" + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1", name="c1") + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + + info = self._upload(client, ws.id) + resp = client.post( + f"/v1/api/workstreams/{ws.id}/send", + json={"message": "hi", "attachment_ids": [info["attachment_id"]]}, + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "ok" + # Surfacing parity with interactive: response carries the + # attached / dropped lists even when no drops occurred. + assert body["attached_ids"] == [info["attachment_id"]] + assert body["dropped_attachment_ids"] == [] + + def test_send_response_includes_attached_ids_field_when_no_attachments(self, storage): + """The unified response shape always carries ``attached_ids`` / + ``dropped_attachment_ids`` so SDK consumers don't have to + branch on whether attachments were involved.""" + mgr = _build_mgr(storage) + ws = mgr.create(user_id="user-1", name="c1") + client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry()) + + resp = client.post( + f"/v1/api/workstreams/{ws.id}/send", + json={"message": "no attachments here"}, + headers=_COORD_HEADERS, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "ok" + assert body["attached_ids"] == [] + assert body["dropped_attachment_ids"] == []