diff --git a/sdk/typescript/src/base.ts b/sdk/typescript/src/base.ts index 18a73686..7d09cfa8 100644 --- a/sdk/typescript/src/base.ts +++ b/sdk/typescript/src/base.ts @@ -29,6 +29,12 @@ export interface ClientOptions { export interface RequestOptions { json?: object; params?: Record; + /** + * When set, send as multipart form-data with this body. The runtime's + * fetch sets the Content-Type + boundary itself, so we deliberately do + * not include a Content-Type header in this case. + */ + form?: FormData; } export class BaseClient { @@ -47,36 +53,34 @@ export class BaseClient { path: string, options?: RequestOptions, ): Promise { - const headers: Record = { - "Content-Type": "application/json", - }; + const headers: Record = {}; + if (!options?.form) { + headers["Content-Type"] = "application/json"; + } if (this.token) { headers["Authorization"] = `Bearer ${this.token}`; } - let url = `${this.baseUrl}${path}`; - if (options?.params) { - const searchParams = new URLSearchParams(); - for (const [key, value] of Object.entries(options.params)) { - if (value !== undefined && value !== "") { - searchParams.set(key, String(value)); - } - } - const qs = searchParams.toString(); - if (qs) url += `?${qs}`; + const url = this._buildUrl(path, options?.params); + + let body: BodyInit | undefined; + if (options?.form) { + body = options.form; + } else if (options?.json) { + body = JSON.stringify(options.json); } const resp = await this.fetchFn(url, { method, headers, - body: options?.json ? JSON.stringify(options.json) : undefined, + body, }); if (!resp.ok) { let msg = ""; try { - const body = (await resp.json()) as Record; - msg = (body.error as string) ?? (body.detail as string) ?? ""; + const errBody = (await resp.json()) as Record; + msg = (errBody.error as string) ?? (errBody.detail as string) ?? ""; } catch { msg = await resp.text().catch(() => ""); } @@ -86,6 +90,55 @@ export class BaseClient { return (await resp.json()) as T; } + protected async requestBytes( + method: string, + path: string, + options?: { params?: Record }, + ): Promise<{ bytes: Uint8Array; contentType: string; filename: string }> { + const headers: Record = {}; + if (this.token) { + headers["Authorization"] = `Bearer ${this.token}`; + } + + const url = this._buildUrl(path, options?.params); + const resp = await this.fetchFn(url, { method, headers }); + if (!resp.ok) { + let msg = ""; + try { + const errBody = (await resp.json()) as Record; + msg = (errBody.error as string) ?? (errBody.detail as string) ?? ""; + } catch { + msg = await resp.text().catch(() => ""); + } + throw new TurnstoneAPIError(resp.status, msg || `HTTP ${resp.status}`); + } + const contentType = + resp.headers.get("content-type") ?? "application/octet-stream"; + const disposition = resp.headers.get("content-disposition") ?? ""; + const match = /filename="?([^";]+)"?/.exec(disposition); + const filename = match ? match[1] : ""; + const buf = await resp.arrayBuffer(); + return { bytes: new Uint8Array(buf), contentType, filename }; + } + + private _buildUrl( + path: string, + params?: Record, + ): string { + let url = `${this.baseUrl}${path}`; + if (params) { + const searchParams = new URLSearchParams(); + for (const [key, value] of Object.entries(params)) { + if (value !== undefined && value !== "") { + searchParams.set(key, String(value)); + } + } + const qs = searchParams.toString(); + if (qs) url += `?${qs}`; + } + return url; + } + protected async *streamSSE>( path: string, params?: Record, diff --git a/sdk/typescript/src/console.ts b/sdk/typescript/src/console.ts index c4efdfc2..f0169547 100644 --- a/sdk/typescript/src/console.ts +++ b/sdk/typescript/src/console.ts @@ -4,6 +4,8 @@ import type { AdminListMemoriesOptions, AdminMemoryInfo, AdminSearchMemoriesOptions, + AttachmentContent, + AttachmentUpload, AuditQueryOptions, AuditResponse, AuthLoginResponse, @@ -16,6 +18,9 @@ import type { ConsoleCreateWsRequest, ConsoleCreateWsResponse, ConsoleHealthResponse, + CreateWorkstreamRequest, + CreateWorkstreamResponse, + ListAttachmentsResponse, CreateMcpServerRequest, CreatePolicyOptions, CreateRoleOptions, @@ -55,12 +60,37 @@ import type { UpdateScheduleRequest, UpdateSettingOptions, UpdateSkillRequest, + UploadAttachmentResponse, UsageQueryOptions, UsageResponse, UserRoleInfo, WorkstreamsOptions, } from "./types.js"; +function generateConsoleWsId(): string { + // 16 bytes => 32 hex chars; matches `secrets.token_hex(16)` server-side. + const buf = new Uint8Array(16); + crypto.getRandomValues(buf); + return Array.from(buf, (b) => b.toString(16).padStart(2, "0")).join(""); +} + +function consoleAttachmentToBlob(att: AttachmentUpload): Blob { + if (att.data instanceof Blob) { + return att.mimeType + ? new Blob([att.data], { type: att.mimeType }) + : att.data; + } + // Copy bytes into a fresh ArrayBuffer-backed Uint8Array. The Blob + // BlobPart type rejects ArrayBufferLike views (could be backed by + // SharedArrayBuffer); a freshly allocated buffer is plainly ArrayBuffer. + const src = att.data; + const fresh = new Uint8Array(new ArrayBuffer(src.byteLength)); + fresh.set(src); + return new Blob([fresh], { + type: att.mimeType ?? "application/octet-stream", + }); +} + /** Async client for the turnstone console API. */ export class TurnstoneConsole extends BaseClient { constructor(options: ClientOptions) { @@ -113,6 +143,92 @@ export class TurnstoneConsole extends BaseClient { }); } + // -- Routing proxy -------------------------------------------------------- + + /** + * Create a workstream via the console hash-ring router. + * + * When `attachments` is non-empty the request is sent as + * multipart/form-data and the console routes via `?ws_id=` + * (auto-generated when not supplied) so the body lands on the + * owning node directly. + */ + async routeCreateWorkstream( + opts?: CreateWorkstreamRequest & { target_node?: string }, + ): Promise< + CreateWorkstreamResponse & { node_url?: string; node_id?: string } + > { + const attachments = opts?.attachments; + if (attachments && attachments.length > 0) { + // The console's multipart route_create routes by `?ws_id=` only — + // it does not parse the body to honor `target_node`. Refuse the + // combination at the SDK boundary so callers don't silently get + // routed to the wrong node. + if (opts?.target_node) { + throw new Error( + "target_node is not supported with attachments; " + + "use ws_id (caller-generated to hash to the desired node) instead", + ); + } + const meta: Record = { ...opts }; + delete (meta as { attachments?: unknown }).attachments; + let wsId = (meta.ws_id as string | undefined) ?? ""; + if (!wsId) { + wsId = generateConsoleWsId(); + meta.ws_id = wsId; + } + const form = new FormData(); + form.append("meta", JSON.stringify(meta)); + for (const att of attachments) { + form.append("file", consoleAttachmentToBlob(att), att.filename); + } + return this.request("POST", "/v1/api/route/workstreams/new", { + form, + params: { ws_id: wsId }, + }); + } + return this.request("POST", "/v1/api/route/workstreams/new", { + json: opts ?? {}, + }); + } + + async routeUploadAttachment( + wsId: string, + file: AttachmentUpload, + ): Promise { + const form = new FormData(); + form.append("file", consoleAttachmentToBlob(file), file.filename); + return this.request( + "POST", + `/v1/api/route/workstreams/${wsId}/attachments`, + { form }, + ); + } + + async routeListAttachments(wsId: string): Promise { + return this.request("GET", `/v1/api/route/workstreams/${wsId}/attachments`); + } + + async routeGetAttachmentContent( + wsId: string, + attachmentId: string, + ): Promise { + return this.requestBytes( + "GET", + `/v1/api/route/workstreams/${wsId}/attachments/${attachmentId}/content`, + ); + } + + async routeDeleteAttachment( + wsId: string, + attachmentId: string, + ): Promise { + return this.request( + "DELETE", + `/v1/api/route/workstreams/${wsId}/attachments/${attachmentId}`, + ); + } + // -- Streaming ------------------------------------------------------------ async *clusterEvents(): AsyncIterableIterator { diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 0c3746c5..971ea9a7 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -183,6 +183,12 @@ export type { SkillInstallRequest, SkillInstallResponse, SkillInstallSkipped, + // Attachment types + AttachmentUpload, + AttachmentInfo, + UploadAttachmentResponse, + ListAttachmentsResponse, + AttachmentContent, } from "./types.js"; // SSE parser (for advanced usage) diff --git a/sdk/typescript/src/server.ts b/sdk/typescript/src/server.ts index b8ae41b0..8fb9609a 100644 --- a/sdk/typescript/src/server.ts +++ b/sdk/typescript/src/server.ts @@ -1,6 +1,8 @@ import { BaseClient, type ClientOptions } from "./base.js"; import type { ServerEvent } from "./events.js"; import type { + AttachmentContent, + AttachmentUpload, AuthLoginResponse, AuthSetupResponse, AuthStatusResponse, @@ -9,20 +11,46 @@ import type { DashboardResponse, DeleteMemoryOptions, HealthResponse, + ListAttachmentsResponse, ListMemoriesOptions, ListMemoriesResponse, ListSavedWorkstreamsResponse, - SkillSummary, ListWorkstreamsResponse, MemoryInfo, SaveMemoryRequest, SearchMemoriesRequest, SendAndWaitOptions, SendResponse, + SkillSummary, StatusResponse, TurnResult, + UploadAttachmentResponse, } from "./types.js"; +function generateWsId(): string { + // 16 bytes => 32 hex chars; matches `secrets.token_hex(16)` server-side. + const buf = new Uint8Array(16); + crypto.getRandomValues(buf); + return Array.from(buf, (b) => b.toString(16).padStart(2, "0")).join(""); +} + +function attachmentToBlob(att: AttachmentUpload): Blob { + if (att.data instanceof Blob) { + return att.mimeType + ? new Blob([att.data], { type: att.mimeType }) + : att.data; + } + // Copy bytes into a fresh ArrayBuffer-backed Uint8Array. The Blob + // BlobPart type rejects ArrayBufferLike views (could be backed by + // SharedArrayBuffer); a freshly allocated buffer is plainly ArrayBuffer. + const src = att.data; + const fresh = new Uint8Array(new ArrayBuffer(src.byteLength)); + fresh.set(src); + return new Blob([fresh], { + type: att.mimeType ?? "application/octet-stream", + }); +} + /** Async client for the turnstone server API. */ export class TurnstoneServer extends BaseClient { constructor(options: ClientOptions) { @@ -42,7 +70,27 @@ export class TurnstoneServer extends BaseClient { async createWorkstream( opts?: CreateWorkstreamRequest, ): Promise { - return this.request("POST", "/v1/api/workstreams/new", { json: opts }); + const attachments = opts?.attachments; + if (attachments && attachments.length > 0) { + // Multipart variant: pre-generate ws_id so cluster routers can + // hash to the owning node before this body lands. Server accepts + // either a server-generated id (when meta.ws_id is empty) or the + // caller-supplied one. + const meta: Record = { ...opts }; + delete (meta as { attachments?: unknown }).attachments; + if (!meta.ws_id) { + meta.ws_id = generateWsId(); + } + const form = new FormData(); + form.append("meta", JSON.stringify(meta)); + for (const att of attachments) { + form.append("file", attachmentToBlob(att), att.filename); + } + return this.request("POST", "/v1/api/workstreams/new", { form }); + } + return this.request("POST", "/v1/api/workstreams/new", { + json: opts ?? {}, + }); } async closeWorkstream(wsId: string): Promise { @@ -53,12 +101,55 @@ export class TurnstoneServer extends BaseClient { // -- Chat interaction ----------------------------------------------------- - async send(message: string, wsId: string): Promise { - return this.request("POST", "/v1/api/send", { - json: { message, ws_id: wsId }, + async send( + message: string, + wsId: string, + opts?: { attachmentIds?: string[] }, + ): Promise { + const body: Record = { message, ws_id: wsId }; + if (opts?.attachmentIds !== undefined) { + body.attachment_ids = opts.attachmentIds; + } + return this.request("POST", "/v1/api/send", { json: body }); + } + + // -- Attachments ---------------------------------------------------------- + + async uploadAttachment( + wsId: string, + file: AttachmentUpload, + ): Promise { + const form = new FormData(); + form.append("file", attachmentToBlob(file), file.filename); + return this.request("POST", `/v1/api/workstreams/${wsId}/attachments`, { + form, }); } + async listAttachments(wsId: string): Promise { + return this.request("GET", `/v1/api/workstreams/${wsId}/attachments`); + } + + async getAttachmentContent( + wsId: string, + attachmentId: string, + ): Promise { + return this.requestBytes( + "GET", + `/v1/api/workstreams/${wsId}/attachments/${attachmentId}/content`, + ); + } + + async deleteAttachment( + wsId: string, + attachmentId: string, + ): Promise { + return this.request( + "DELETE", + `/v1/api/workstreams/${wsId}/attachments/${attachmentId}`, + ); + } + async approve(opts: { wsId: string; approved?: boolean; diff --git a/sdk/typescript/src/types.ts b/sdk/typescript/src/types.ts index dbfcbb33..049368ce 100644 --- a/sdk/typescript/src/types.ts +++ b/sdk/typescript/src/types.ts @@ -50,10 +50,67 @@ export interface AuthSetupResponse { export interface SendRequest { message: string; ws_id: string; + /** + * Explicit list of pending attachment ids to inject into this turn. + * When omitted, any pending attachments for the caller on the + * workstream are auto-consumed; an empty list disables auto-consume. + */ + attachment_ids?: string[]; } export interface SendResponse { + /** "ok" | "busy" | "queued" | "queue_full". */ status: string; + /** + * Attachment ids actually reserved onto this turn. Subset of the + * request's `attachment_ids` (or the auto-consumed pending set). + */ + attached_ids?: string[]; + /** + * 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. + */ + dropped_attachment_ids?: string[]; + /** Set on "queued" responses: relative priority of the queued message. */ + priority?: string | null; + /** Set on "queued" responses: id used to dequeue the message. */ + msg_id?: string | null; +} + +// --------------------------------------------------------------------------- +// Server API — Attachments +// --------------------------------------------------------------------------- + +/** A file to upload as an attachment. */ +export interface AttachmentUpload { + filename: string; + /** Raw file bytes; use a `Blob` in browsers and a `Uint8Array` in Node. */ + data: Blob | Uint8Array; + /** Optional advisory MIME type; the server applies its own validation. */ + mimeType?: string; +} + +export interface AttachmentInfo { + attachment_id: string; + filename: string; + mime_type: string; + size_bytes: number; + /** "image" or "text". */ + kind: string; +} + +export type UploadAttachmentResponse = AttachmentInfo; + +export interface ListAttachmentsResponse { + attachments: AttachmentInfo[]; +} + +/** Raw bytes returned from the attachment `/content` endpoint. */ +export interface AttachmentContent { + bytes: Uint8Array; + contentType: string; + filename: string; } export interface ApproveRequest { @@ -79,6 +136,20 @@ export interface CreateWorkstreamRequest { auto_approve?: boolean; resume_ws?: string; skill?: string; + /** First user message dispatched in a background worker after creation. */ + initial_message?: string; + /** + * Caller-supplied workstream id (32-hex). Auto-generated when omitted. + * Required for cluster-routed multipart creates so the console can + * hash to the owning node before the body lands. + */ + ws_id?: string; + /** + * Files to attach to the first turn. When non-empty the request is + * sent as multipart/form-data and (with `initial_message`) reserved + * onto that turn before the worker dispatches. + */ + attachments?: AttachmentUpload[]; } export interface CreateWorkstreamResponse { @@ -86,6 +157,8 @@ export interface CreateWorkstreamResponse { name: string; resumed?: boolean; message_count?: number; + /** Ids of attachments saved by this request (multipart variant only). */ + attachment_ids?: string[]; } export interface CloseWorkstreamRequest { diff --git a/sdk/typescript/tests/console.test.ts b/sdk/typescript/tests/console.test.ts index 96093920..400f108d 100644 --- a/sdk/typescript/tests/console.test.ts +++ b/sdk/typescript/tests/console.test.ts @@ -62,6 +62,28 @@ describe("TurnstoneConsole", () => { expect(url).toContain("page=2"); }); + it("routeCreateWorkstream rejects attachments + target_node", async () => { + const fetchFn = vi.fn().mockResolvedValue( + new Response("{}", { + status: 500, + headers: { "content-type": "application/json" }, + }), + ); + const client = new TurnstoneConsole({ + baseUrl: "http://test", + fetch: fetchFn, + }); + const data = new TextEncoder().encode("hi"); + await expect( + client.routeCreateWorkstream({ + name: "x", + target_node: "n1", + attachments: [{ filename: "a.txt", data }], + }), + ).rejects.toThrow(/target_node/); + expect(fetchFn).not.toHaveBeenCalled(); + }); + it("health returns parsed response", async () => { const fetchFn = mockFetch({ status: "ok", diff --git a/sdk/typescript/tests/server-attachments.test.ts b/sdk/typescript/tests/server-attachments.test.ts new file mode 100644 index 00000000..5c28baac --- /dev/null +++ b/sdk/typescript/tests/server-attachments.test.ts @@ -0,0 +1,168 @@ +import { describe, expect, it, vi } from "vitest"; +import { TurnstoneServer } from "../src/server.js"; + +function mockFetch(response: object, status = 200): typeof globalThis.fetch { + return vi.fn().mockResolvedValue( + new Response(JSON.stringify(response), { + status, + headers: { "content-type": "application/json" }, + }), + ); +} + +function mockFetchBytes( + body: Uint8Array, + contentType: string, + filename = "", +): typeof globalThis.fetch { + const headers: Record = { "content-type": contentType }; + if (filename) + headers["content-disposition"] = `inline; filename="${filename}"`; + return vi + .fn() + .mockResolvedValue(new Response(body, { status: 200, headers })); +} + +describe("TurnstoneServer attachments", () => { + it("uploadAttachment sends multipart with filename", async () => { + const fetchFn = mockFetch({ + attachment_id: "att-1", + filename: "a.txt", + mime_type: "text/plain", + size_bytes: 5, + kind: "text", + }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + const data = new TextEncoder().encode("hello"); + const result = await client.uploadAttachment("ws-X", { + filename: "a.txt", + data, + mimeType: "text/plain", + }); + expect(result.attachment_id).toBe("att-1"); + + const [url, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(url).toBe("http://test/v1/api/workstreams/ws-X/attachments"); + expect(init.method).toBe("POST"); + expect(init.body).toBeInstanceOf(FormData); + // Browser/Node fetch sets the Content-Type header from FormData itself + expect(init.headers["Content-Type"]).toBeUndefined(); + }); + + it("listAttachments hits the GET endpoint", async () => { + const fetchFn = mockFetch({ attachments: [] }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + const resp = await client.listAttachments("ws-X"); + expect(resp.attachments).toEqual([]); + const [url, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(url).toBe("http://test/v1/api/workstreams/ws-X/attachments"); + expect(init.method).toBe("GET"); + }); + + it("getAttachmentContent returns raw bytes + parsed headers", async () => { + const bytes = new TextEncoder().encode("hello world"); + const fetchFn = mockFetchBytes( + bytes, + "text/plain; charset=utf-8", + "notes.md", + ); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + const result = await client.getAttachmentContent("ws-X", "att-1"); + expect(new TextDecoder().decode(result.bytes)).toBe("hello world"); + expect(result.contentType).toBe("text/plain; charset=utf-8"); + expect(result.filename).toBe("notes.md"); + }); + + it("deleteAttachment hits the DELETE endpoint", async () => { + const fetchFn = mockFetch({ status: "deleted" }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + const resp = await client.deleteAttachment("ws-X", "att-1"); + expect(resp.status).toBe("deleted"); + const [, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(init.method).toBe("DELETE"); + }); + + it("send threads attachment_ids when provided", async () => { + const fetchFn = mockFetch({ status: "ok" }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + await client.send("hi", "ws-X", { attachmentIds: ["a1", "a2"] }); + const [, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(JSON.parse(init.body)).toEqual({ + message: "hi", + ws_id: "ws-X", + attachment_ids: ["a1", "a2"], + }); + }); + + it("send omits attachment_ids when not supplied", async () => { + const fetchFn = mockFetch({ status: "ok" }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + await client.send("hi", "ws-X"); + const [, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(JSON.parse(init.body)).toEqual({ message: "hi", ws_id: "ws-X" }); + }); + + it("createWorkstream with attachments sends multipart and auto-generates ws_id", async () => { + const fetchFn = mockFetch({ + ws_id: "00ff00000000000000000000000000ff", + name: "demo", + attachment_ids: ["att-1"], + }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + const data = new TextEncoder().encode("hello"); + const resp = await client.createWorkstream({ + name: "demo", + initial_message: "describe", + attachments: [{ filename: "a.txt", data, mimeType: "text/plain" }], + }); + expect(resp.attachment_ids).toEqual(["att-1"]); + + const [url, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(url).toBe("http://test/v1/api/workstreams/new"); + expect(init.method).toBe("POST"); + expect(init.body).toBeInstanceOf(FormData); + + const form = init.body as FormData; + const meta = JSON.parse(form.get("meta") as string); + expect(meta.name).toBe("demo"); + expect(meta.initial_message).toBe("describe"); + expect(meta.ws_id).toMatch(/^[0-9a-f]{32}$/); + expect(meta.attachments).toBeUndefined(); + + const file = form.get("file"); + expect(file).toBeInstanceOf(Blob); + }); + + it("createWorkstream without attachments uses JSON body", async () => { + const fetchFn = mockFetch({ ws_id: "ws-json", name: "j" }); + const client = new TurnstoneServer({ + baseUrl: "http://test", + fetch: fetchFn, + }); + await client.createWorkstream({ name: "j" }); + const [, init] = (fetchFn as ReturnType).mock.calls[0]; + expect(init.headers["Content-Type"]).toBe("application/json"); + expect(JSON.parse(init.body)).toEqual({ name: "j" }); + }); +}); diff --git a/tests/test_console_route_attachments.py b/tests/test_console_route_attachments.py new file mode 100644 index 00000000..019311e5 --- /dev/null +++ b/tests/test_console_route_attachments.py @@ -0,0 +1,353 @@ +"""Tests for console routing of attachment endpoints + multipart route_create. + +Covers the cluster-routing surface added alongside the workstream +attachment-on-create feature: the multipart variant of route_create and +the four ws-id-keyed attachment proxies under /v1/api/route/. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +import httpx +from starlette.testclient import TestClient + +from turnstone.console.collector import ClusterCollector +from turnstone.console.router import ConsoleRouter, NodeRef + +_TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!" + + +def _test_jwt() -> str: + from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt + + return create_jwt( + user_id="test-routing", + scopes=frozenset({"read", "write", "approve", "service"}), + source="test", + secret=_TEST_JWT_SECRET, + audience=JWT_AUD_CONSOLE, + ) + + +_AUTH: dict[str, str] = {"Authorization": f"Bearer {_test_jwt()}"} + + +def _make_app(router: Any) -> Any: + from turnstone.console.server import _load_static, create_app + + _load_static() + collector = MagicMock(spec=ClusterCollector) + return create_app( + collector=collector, + jwt_secret=_TEST_JWT_SECRET, + router=router, + ) + + +def _make_router() -> MagicMock: + router = MagicMock(spec=ConsoleRouter) + router.is_ready.return_value = True + router.route.return_value = NodeRef("node-a", "http://a:8080") + return router + + +# --------------------------------------------------------------------------- +# route_create multipart +# --------------------------------------------------------------------------- + + +class TestRouteCreateMultipart: + def test_multipart_requires_ws_id_query(self): + router = _make_router() + app = _make_app(router=router) + app.state.proxy_client = MagicMock(spec=httpx.AsyncClient) + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.post( + "/v1/api/route/workstreams/new", + files=[("file", ("a.txt", b"hello", "text/plain"))], + data={"meta": "{}"}, + headers=_AUTH, + ) + assert resp.status_code == 400 + assert "ws_id" in resp.json()["error"] + finally: + client.close() + + def test_multipart_forwards_raw_body_to_routed_node(self): + router = _make_router() + app = _make_app(router=router) + + captured: dict[str, Any] = {} + + async def _mock_post(*args: Any, **kwargs: Any) -> httpx.Response: + captured["url"] = args[0] if args else "" + captured["headers"] = kwargs.get("headers") or {} + captured["content"] = kwargs.get("content") + return httpx.Response( + 200, + json={"ws_id": "00ff" + "0" * 28, "name": "demo"}, + request=httpx.Request("POST", args[0] if args else "http://test"), + ) + + mock_proxy = MagicMock(spec=httpx.AsyncClient) + mock_proxy.post = MagicMock(side_effect=_mock_post) + app.state.proxy_client = mock_proxy + client = TestClient(app, raise_server_exceptions=False) + try: + ws_id = "00ff" + "0" * 28 + resp = client.post( + f"/v1/api/route/workstreams/new?ws_id={ws_id}", + files=[("file", ("a.txt", b"hello", "text/plain"))], + data={"meta": '{"name":"demo"}'}, + headers=_AUTH, + ) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["node_id"] == "node-a" + # Forwarded multipart Content-Type + assert captured["headers"].get("Content-Type", "").startswith("multipart/form-data") + # Body bytes were forwarded raw + assert isinstance(captured["content"], (bytes, bytearray)) + assert b"hello" in bytes(captured["content"]) + router.route.assert_called_with(ws_id) + finally: + client.close() + + def test_multipart_preserves_mixed_case_boundary(self): + """The boundary= param is case-sensitive — must match body bytes verbatim. + + Regression for an earlier bug where route_create lowercased the + whole Content-Type header before forwarding, mangling boundaries + like ``WebKitFormBoundary7MA4YWxkTrZu0gW``. + """ + router = _make_router() + app = _make_app(router=router) + + captured: dict[str, Any] = {} + + async def _mock_post(*args: Any, **kwargs: Any) -> httpx.Response: + captured["headers"] = kwargs.get("headers") or {} + captured["content"] = kwargs.get("content") + return httpx.Response( + 200, + json={"ws_id": "00ff" + "0" * 28, "name": "ok"}, + request=httpx.Request("POST", args[0] if args else "http://test"), + ) + + mock_proxy = MagicMock(spec=httpx.AsyncClient) + mock_proxy.post = MagicMock(side_effect=_mock_post) + app.state.proxy_client = mock_proxy + client = TestClient(app, raise_server_exceptions=False) + try: + ws_id = "00ff" + "0" * 28 + boundary = "WebKitFormBoundary7MA4YWxkTrZu0gW" # mixed-case + body = ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="meta"\r\n\r\n' + f'{{"name":"demo"}}\r\n' + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="file"; filename="a.txt"\r\n' + f"Content-Type: text/plain\r\n\r\n" + f"hello\r\n" + f"--{boundary}--\r\n" + ).encode() + resp = client.post( + f"/v1/api/route/workstreams/new?ws_id={ws_id}", + content=body, + headers={ + **_AUTH, + "Content-Type": f"multipart/form-data; boundary={boundary}", + }, + ) + assert resp.status_code == 200, resp.text + forwarded = captured["headers"].get("Content-Type", "") + assert boundary in forwarded, ( + f"boundary mangled in upstream Content-Type: {forwarded!r}" + ) + # Body bytes still contain the mixed-case boundary + assert boundary.encode() in bytes(captured["content"]) + finally: + client.close() + + def test_json_path_unchanged(self): + """Existing JSON callers should continue to work as before.""" + router = _make_router() + app = _make_app(router=router) + + async def _mock_post(*args: Any, **kwargs: Any) -> httpx.Response: + return httpx.Response( + 200, + json={"ws_id": "abc123", "name": "json"}, + request=httpx.Request("POST", args[0] if args else "http://test"), + ) + + mock_proxy = MagicMock(spec=httpx.AsyncClient) + mock_proxy.post = MagicMock(side_effect=_mock_post) + app.state.proxy_client = mock_proxy + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.post( + "/v1/api/route/workstreams/new", + json={"name": "json"}, + headers=_AUTH, + ) + assert resp.status_code == 200 + assert resp.json()["ws_id"] == "abc123" + # JSON path uses json= kwarg, not content= + call_kwargs = mock_proxy.post.call_args.kwargs + assert "json" in call_kwargs + assert "content" not in call_kwargs + finally: + client.close() + + +# --------------------------------------------------------------------------- +# route_attachment_proxy +# --------------------------------------------------------------------------- + + +class TestRouteAttachmentProxy: + def _wire(self, mock_request_fn) -> tuple[Any, MagicMock]: + router = _make_router() + app = _make_app(router=router) + mock_proxy = MagicMock(spec=httpx.AsyncClient) + mock_proxy.request = MagicMock(side_effect=mock_request_fn) + mock_proxy.get = MagicMock(side_effect=mock_request_fn) + mock_proxy.post = MagicMock(side_effect=mock_request_fn) + app.state.proxy_client = mock_proxy + return app, mock_proxy + + def test_upload_proxies_multipart(self): + captured: dict[str, Any] = {} + + async def _mock(*args: Any, **kwargs: Any) -> httpx.Response: + captured["method"] = args[0] if args else kwargs.get("method") + captured["url"] = args[1] if len(args) > 1 else kwargs.get("url", "") + captured["headers"] = kwargs.get("headers") or {} + captured["content"] = kwargs.get("content") + return httpx.Response( + 200, + json={ + "attachment_id": "att-1", + "filename": "a.txt", + "mime_type": "text/plain", + "size_bytes": 5, + "kind": "text", + }, + request=httpx.Request("POST", "http://a:8080/x"), + ) + + app, _ = self._wire(_mock) + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.post( + "/v1/api/route/workstreams/ws-X/attachments", + files=[("file", ("a.txt", b"hello", "text/plain"))], + headers=_AUTH, + ) + assert resp.status_code == 200 + assert resp.json()["attachment_id"] == "att-1" + assert "/v1/api/workstreams/ws-X/attachments" in captured["url"] + assert "/route/" not in captured["url"] + assert captured["headers"].get("Content-Type", "").startswith("multipart/form-data") + finally: + client.close() + + def test_list_proxies_get(self): + async def _mock(*args: Any, **kwargs: Any) -> httpx.Response: + return httpx.Response( + 200, + json={"attachments": []}, + request=httpx.Request("GET", "http://a:8080/x"), + ) + + app, mock_proxy = self._wire(_mock) + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.get( + "/v1/api/route/workstreams/ws-X/attachments", + headers=_AUTH, + ) + assert resp.status_code == 200 + assert resp.json() == {"attachments": []} + mock_proxy.get.assert_called() + finally: + client.close() + + def test_get_content_preserves_upstream_headers(self): + async def _mock(*args: Any, **kwargs: Any) -> httpx.Response: + return httpx.Response( + 200, + content=b"hello world", + headers={ + "Content-Type": "text/plain; charset=utf-8", + "Content-Disposition": 'inline; filename="notes.md"', + "X-Content-Type-Options": "nosniff", + }, + request=httpx.Request("GET", "http://a:8080/x"), + ) + + app, _ = self._wire(_mock) + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.get( + "/v1/api/route/workstreams/ws-X/attachments/att-1/content", + headers=_AUTH, + ) + assert resp.status_code == 200 + assert resp.content == b"hello world" + assert resp.headers.get("X-Content-Type-Options") == "nosniff" + assert "filename" in resp.headers.get("Content-Disposition", "") + finally: + client.close() + + def test_delete_proxies_method(self): + captured: dict[str, Any] = {} + + async def _mock(*args: Any, **kwargs: Any) -> httpx.Response: + captured["method"] = args[0] if args else "" + captured["url"] = args[1] if len(args) > 1 else "" + return httpx.Response( + 200, + json={"status": "deleted"}, + request=httpx.Request("DELETE", "http://a:8080/x"), + ) + + app, _ = self._wire(_mock) + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.delete( + "/v1/api/route/workstreams/ws-X/attachments/att-1", + headers=_AUTH, + ) + assert resp.status_code == 200 + assert resp.json() == {"status": "deleted"} + assert captured["method"] == "DELETE" + finally: + client.close() + + +# --------------------------------------------------------------------------- +# Routing-failure paths +# --------------------------------------------------------------------------- + + +class TestRoutingFailures: + def test_router_not_ready_returns_503(self): + router = MagicMock(spec=ConsoleRouter) + router.is_ready.return_value = False + router.refresh_cache.return_value = None + app = _make_app(router=router) + app.state.proxy_client = MagicMock(spec=httpx.AsyncClient) + client = TestClient(app, raise_server_exceptions=False) + try: + resp = client.get( + "/v1/api/route/workstreams/ws-X/attachments", + headers=_AUTH, + ) + assert resp.status_code == 503 + finally: + client.close() diff --git a/tests/test_sdk_console.py b/tests/test_sdk_console.py index aa35569d..e395c9b8 100644 --- a/tests/test_sdk_console.py +++ b/tests/test_sdk_console.py @@ -467,6 +467,27 @@ async def test_route_create_workstream(): assert captured_body["user_id"] == "u1" +@pytest.mark.anyio +async def test_route_create_workstream_rejects_attachments_with_target_node(): + """Regression: target_node has no effect on multipart route_create + (which routes by ?ws_id=) — refuse the combination at the SDK boundary + instead of silently routing to the wrong node. + """ + from turnstone.sdk._types import AttachmentUpload + + transport = httpx.MockTransport( + lambda req: _json_response({"error": "should not be called"}, status=500) + ) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + with pytest.raises(ValueError, match="target_node"): + await client.route_create_workstream( + name="x", + target_node="n1", + attachments=[AttachmentUpload(filename="a.txt", data=b"hi")], + ) + + @pytest.mark.anyio async def test_route_create_workstream_omits_defaults(): captured_body: dict = {} diff --git a/tests/test_sdk_server_attachments.py b/tests/test_sdk_server_attachments.py new file mode 100644 index 00000000..6d697b24 --- /dev/null +++ b/tests/test_sdk_server_attachments.py @@ -0,0 +1,229 @@ +"""Tests for the attachment surface of turnstone.sdk.server (async + sync). + +Uses ``httpx.MockTransport`` to record what the SDK sends so we can +assert on multipart bodies, the auto-generated ws_id, etc. +""" + +from __future__ import annotations + +import json +import re + +import httpx +import pytest + +from turnstone.sdk._types import AttachmentUpload +from turnstone.sdk.server import AsyncTurnstoneServer + +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\xaeB`\x82" +) + + +def _capturing_transport(response: httpx.Response) -> tuple[httpx.MockTransport, list]: + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return response + + return httpx.MockTransport(handler), captured + + +# --------------------------------------------------------------------------- +# upload / list / get_content / delete +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_upload_attachment_sends_multipart(): + response = httpx.Response( + 200, + json={ + "attachment_id": "att-1", + "filename": "tiny.png", + "mime_type": "image/png", + "size_bytes": len(PNG_1x1), + "kind": "image", + }, + ) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + result = await client.upload_attachment("ws-X", "tiny.png", PNG_1x1, mime_type="image/png") + assert result.attachment_id == "att-1" + assert result.kind == "image" + assert len(captured) == 1 + req = captured[0] + assert req.method == "POST" + assert req.url.path == "/v1/api/workstreams/ws-X/attachments" + ct = req.headers.get("content-type", "") + assert ct.startswith("multipart/form-data") + body = bytes(req.content) + assert b"tiny.png" in body + assert PNG_1x1 in body + + +@pytest.mark.anyio +async def test_list_attachments_returns_pending(): + response = httpx.Response( + 200, + json={ + "attachments": [ + { + "attachment_id": "att-1", + "filename": "a.txt", + "mime_type": "text/plain", + "size_bytes": 5, + "kind": "text", + } + ] + }, + ) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + result = await client.list_attachments("ws-X") + assert len(result.attachments) == 1 + assert result.attachments[0].attachment_id == "att-1" + assert captured[0].method == "GET" + + +@pytest.mark.anyio +async def test_get_attachment_content_returns_bytes(): + response = httpx.Response( + 200, + content=b"hello world", + headers={"Content-Type": "text/plain; charset=utf-8"}, + ) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + data = await client.get_attachment_content("ws-X", "att-1") + assert data == b"hello world" + assert captured[0].url.path == "/v1/api/workstreams/ws-X/attachments/att-1/content" + + +@pytest.mark.anyio +async def test_delete_attachment(): + response = httpx.Response(200, json={"status": "deleted"}) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + result = await client.delete_attachment("ws-X", "att-1") + assert result.status == "deleted" + assert captured[0].method == "DELETE" + + +# --------------------------------------------------------------------------- +# send(attachment_ids=...) +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_send_with_attachment_ids(): + response = httpx.Response(200, json={"status": "ok"}) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + await client.send("hi", "ws-X", attachment_ids=["a1", "a2"]) + body = json.loads(bytes(captured[0].content)) + assert body["attachment_ids"] == ["a1", "a2"] + assert body["message"] == "hi" + + +@pytest.mark.anyio +async def test_send_omits_attachment_ids_when_none(): + response = httpx.Response(200, json={"status": "ok"}) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + await client.send("hi", "ws-X") + body = json.loads(bytes(captured[0].content)) + assert "attachment_ids" not in body + + +# --------------------------------------------------------------------------- +# create_workstream(attachments=...) +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_create_workstream_with_attachments_sends_multipart(): + response = httpx.Response( + 200, + json={ + "ws_id": "00ff" + "0" * 28, + "name": "demo", + "resumed": False, + "message_count": 0, + "attachment_ids": ["att-1"], + }, + ) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + resp = await client.create_workstream( + name="demo", + initial_message="describe", + attachments=[AttachmentUpload(filename="hi.png", data=PNG_1x1, mime_type="image/png")], + ) + assert resp.ws_id + assert resp.attachment_ids == ["att-1"] + req = captured[0] + assert req.method == "POST" + assert req.url.path == "/v1/api/workstreams/new" + ct = req.headers.get("content-type", "") + assert ct.startswith("multipart/form-data") + + body = bytes(req.content) + # `meta` field carries the JSON metadata including the auto-generated ws_id + meta_match = re.search(rb'name="meta"\r\n\r\n(\{[^}]*\})', body) + assert meta_match, body + meta = json.loads(meta_match.group(1)) + assert meta["name"] == "demo" + assert meta["initial_message"] == "describe" + assert re.fullmatch(r"[0-9a-f]{32}", meta["ws_id"]) + # PNG bytes appear in the body as a file part + assert PNG_1x1 in body + + +@pytest.mark.anyio +async def test_create_workstream_caller_supplied_ws_id_used(): + response = httpx.Response( + 200, + json={ + "ws_id": "deadbeef" * 4, + "name": "demo", + "resumed": False, + "message_count": 0, + "attachment_ids": [], + }, + ) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + await client.create_workstream( + name="demo", + ws_id="deadbeef" * 4, + attachments=[AttachmentUpload(filename="a.txt", data=b"hi")], + ) + body = bytes(captured[0].content) + meta_match = re.search(rb'name="meta"\r\n\r\n(\{[^}]*\})', body) + assert meta_match + meta = json.loads(meta_match.group(1)) + assert meta["ws_id"] == "deadbeef" * 4 + + +@pytest.mark.anyio +async def test_create_workstream_without_attachments_uses_json(): + """Back-compat: callers that don't pass attachments still get the JSON path.""" + response = httpx.Response(200, json={"ws_id": "ws-json", "name": "j", "attachment_ids": []}) + transport, captured = _capturing_transport(response) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + await client.create_workstream(name="j") + req = captured[0] + assert req.headers.get("content-type", "").startswith("application/json") diff --git a/tests/test_server_attachments_on_create.py b/tests/test_server_attachments_on_create.py new file mode 100644 index 00000000..ac13725e --- /dev/null +++ b/tests/test_server_attachments_on_create.py @@ -0,0 +1,425 @@ +"""Tests for the multipart variant of POST /v1/api/workstreams/new. + +Exercises: +- The pure helpers `_validate_and_save_uploaded_files` and + `_reserve_and_resolve_attachments` (added alongside the multipart path). +- The full create endpoint via TestClient with a FakeSession factory so + the initial-message dispatch thread runs end-to-end without an LLM. +""" + +from __future__ import annotations + +import json +import queue +import threading +import time + +import pytest +from starlette.testclient import TestClient + +# Magic-byte-valid 1x1 PNG (matches the fixture in 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\xaeB`\x82" +) + +_TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!" + + +def _make_jwt(user_id: str) -> str: + from turnstone.core.auth import JWT_AUD_SERVER, create_jwt + + return create_jwt( + user_id=user_id, + scopes=frozenset({"read", "write"}), + source="test", + secret=_TEST_JWT_SECRET, + audience=JWT_AUD_SERVER, + ) + + +def _auth(user: str) -> dict[str, str]: + return {"Authorization": f"Bearer {_make_jwt(user)}"} + + +# --------------------------------------------------------------------------- +# Pure helper tests +# --------------------------------------------------------------------------- + + +class TestValidateAndSaveUploadedFiles: + def test_saves_image_and_text(self, tmp_path): + 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) + try: + files = [ + ("hi.png", "image/png", PNG_1x1), + ("notes.md", "text/markdown", b"# Hello\n"), + ] + ids, err = _validate_and_save_uploaded_files(files, "ws-X", "userA") + assert err is None + assert len(ids) == 2 + pending = list_pending_attachments("ws-X", "userA") + assert len(pending) == 2 + kinds = {p["kind"] for p in pending} + assert kinds == {"image", "text"} + finally: + reset_storage() + + def test_rejects_oversized_image(self, tmp_path): + from turnstone.core.attachments import IMAGE_SIZE_CAP + 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) + try: + # Magic-byte-valid PNG header padded past the cap. + oversized = PNG_1x1 + b"\x00" * (IMAGE_SIZE_CAP + 1) + files = [("big.png", "image/png", oversized)] + ids, err = _validate_and_save_uploaded_files(files, "ws-X", "userA") + assert err is not None + assert err.status_code == 413 + assert ids == [] + finally: + reset_storage() + + def test_rejects_unsupported_text(self, tmp_path): + 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) + try: + # No image magic, MIME isn't text/*, extension not allowlisted + files = [("evil.bin", "application/octet-stream", b"\x00\x01\x02")] + ids, err = _validate_and_save_uploaded_files(files, "ws-X", "userA") + assert err is not None + assert err.status_code == 400 + assert ids == [] + finally: + reset_storage() + + def test_pending_cap_returns_409(self, tmp_path): + from turnstone.core.attachments import MAX_PENDING_ATTACHMENTS_PER_USER_WS + 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) + try: + # Saturate the pending cap + for i in range(MAX_PENDING_ATTACHMENTS_PER_USER_WS): + save_attachment( + f"pre-{i}", "ws-X", "userA", f"f{i}.txt", "text/plain", 1, "text", b"x" + ) + files = [("notes.md", "text/markdown", b"hello")] + ids, err = _validate_and_save_uploaded_files(files, "ws-X", "userA") + assert err is not None + assert err.status_code == 409 + assert ids == [] + finally: + reset_storage() + + +class TestReserveAndResolveAttachments: + def test_reserves_and_returns_attachments(self, tmp_path): + from turnstone.core.attachments import Attachment + from turnstone.core.memory import save_attachment + from turnstone.core.storage import init_storage, reset_storage + from turnstone.server import _reserve_and_resolve_attachments + + reset_storage() + init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) + try: + save_attachment("a1", "ws-X", "userA", "a.txt", "text/plain", 5, "text", b"hello") + save_attachment("a2", "ws-X", "userA", "b.png", "image/png", 91, "image", PNG_1x1) + resolved, ordered, dropped = _reserve_and_resolve_attachments( + ["a1", "a2"], "send-1", "ws-X", "userA" + ) + assert ordered == ["a1", "a2"] + assert dropped == [] + assert len(resolved) == 2 + assert all(isinstance(a, Attachment) for a in resolved) + kinds = [a.kind for a in resolved] + assert kinds == ["text", "image"] + finally: + reset_storage() + + def test_double_reserve_drops_second(self, tmp_path): + from turnstone.core.memory import save_attachment + from turnstone.core.storage import init_storage, reset_storage + from turnstone.server import _reserve_and_resolve_attachments + + reset_storage() + init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) + try: + save_attachment("a1", "ws-X", "userA", "a.txt", "text/plain", 5, "text", b"hello") + r1, ord1, _ = _reserve_and_resolve_attachments(["a1"], "send-A", "ws-X", "userA") + assert len(r1) == 1 + r2, ord2, drop2 = _reserve_and_resolve_attachments(["a1"], "send-B", "ws-X", "userA") + assert r2 == [] + assert ord2 == [] + assert drop2 == ["a1"] + finally: + reset_storage() + + +# --------------------------------------------------------------------------- +# End-to-end create endpoint tests (multipart variant) +# --------------------------------------------------------------------------- + + +class _FakeSession: + """Minimal stand-in: records send() invocations from the dispatch thread. + + Knows its own ``ws_id`` and ``user_id`` so it can faithfully simulate the + real ChatSession's attachment-consume step against storage. Tests then + assert that pending attachments are gone after dispatch. + """ + + def __init__(self, ws_id: str = "", user_id: str = ""): + self.ws_id = ws_id + self.user_id = user_id + self.model = "test-model" + self.model_alias = "test-model" + self.messages = [] + self.sends: list[tuple[str, list, str | None]] = [] + self._lock = threading.Lock() + self._cancel_event = threading.Event() + self.notify_targets = "" + self._notify_on_complete = "[]" + + def send(self, text, attachments=None, send_id=None): + with self._lock: + self.sends.append((text, list(attachments or []), send_id)) + # Simulate the real ChatSession's consume step against storage + # so callers can assert the lifecycle landed. + if attachments and send_id and self.ws_id and self.user_id: + import uuid as _uuid + + from turnstone.core.memory import mark_attachments_consumed + + ids = [a.attachment_id for a in attachments] + mark_attachments_consumed( + ids, + _uuid.uuid4().hex, # synthetic conversation message id + self.ws_id, + self.user_id, + reserved_for_msg_id=send_id, + ) + + # Methods the create handler may call but we don't care about + def set_watch_runner(self, *_a, **_kw): + pass + + def queue_message(self, *_a, **_kw): + return ("", "normal", "msg-x") + + def request_title_refresh(self, *_a, **_kw): + pass + + def resume(self, *_a, **_kw): + return False + + +class _FakeUI: + def __init__(self, ws_id="", user_id=""): + self.ws_id = ws_id + self._user_id = user_id + self.auto_approve = False + self.auto_approve_tools: set[str] = set() + self.events: list[dict] = [] + self._enqueued: list[dict] = [] + + def _enqueue(self, ev): + self._enqueued.append(ev) + + def on_stream_end(self): + pass + + def on_state_change(self, state): + self.events.append({"type": "state_change", "state": state}) + + def on_error(self, msg): + self.events.append({"type": "error", "message": msg}) + + +@pytest.fixture +def app_client(tmp_path, monkeypatch): + """End-to-end app with a fake session factory + WorkstreamManager.""" + from turnstone.core.metrics import MetricsCollector + from turnstone.core.storage import init_storage, reset_storage + from turnstone.core.workstream import WorkstreamManager + from turnstone.server import create_app + + reset_storage() + init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False) + + metrics = MetricsCollector() + metrics.model = "test-model" + monkeypatch.setattr("turnstone.server._metrics", metrics) + # Replace WebUI with our fake so the create handler's isinstance check passes. + monkeypatch.setattr("turnstone.server.WebUI", _FakeUI) + + fake_sessions: list[_FakeSession] = [] + + def _factory(ui, _model, ws_id, **_kw): + # The user_id rides on the WebUI factory closure; pull it off the + # ui instance so the FakeSession's consume step uses the right scope. + user_id = getattr(ui, "_user_id", "") + s = _FakeSession(ws_id=ws_id, user_id=user_id) + fake_sessions.append(s) + return s + + mgr = WorkstreamManager(_factory, max_workstreams=10, node_id="node-test") + + gq: queue.Queue[dict] = queue.Queue() + app = create_app( + workstreams=mgr, + global_queue=gq, + global_listeners=[], + global_listeners_lock=threading.Lock(), + skip_permissions=False, + jwt_secret=_TEST_JWT_SECRET, + ) + client = TestClient(app, raise_server_exceptions=False) + try: + yield client, fake_sessions, gq + finally: + client.close() + reset_storage() + + +class TestCreateMultipart: + def test_create_with_image_and_initial_message(self, app_client): + from turnstone.core.memory import list_pending_attachments + + client, sessions, _gq = app_client + meta = {"name": "demo", "initial_message": "describe this image"} + resp = client.post( + "/v1/api/workstreams/new", + data={"meta": json.dumps(meta)}, + files=[("file", ("tiny.png", PNG_1x1, "image/png"))], + headers=_auth("userA"), + ) + assert resp.status_code == 200, resp.text + data = resp.json() + ws_id = data["ws_id"] + assert ws_id + assert len(data["attachment_ids"]) == 1 + + # Wait briefly for the dispatch thread + deadline = time.time() + 2.0 + while time.time() < deadline and not sessions: + time.sleep(0.02) + deadline = time.time() + 2.0 + while time.time() < deadline and not sessions[0].sends: + time.sleep(0.02) + assert sessions + assert sessions[0].sends, "session.send was not invoked" + text, atts, send_id = sessions[0].sends[0] + assert text == "describe this image" + assert send_id # reservation token threaded through + assert len(atts) == 1 + assert atts[0].kind == "image" + + # Lifecycle: the FakeSession marks them consumed via storage — + # so the pending-list for this ws should be empty after dispatch. + assert list_pending_attachments(ws_id, "userA") == [] + + def test_create_with_attachments_no_initial_message_keeps_pending(self, app_client): + from turnstone.core.memory import list_pending_attachments + + client, _, _gq = app_client + meta = {"name": "stash"} + resp = client.post( + "/v1/api/workstreams/new", + data={"meta": json.dumps(meta)}, + files=[("file", ("notes.md", b"# hello\n", "text/markdown"))], + headers=_auth("userA"), + ) + assert resp.status_code == 200, resp.text + data = resp.json() + ws_id = data["ws_id"] + pending = list_pending_attachments(ws_id, "userA") + assert len(pending) == 1 + assert pending[0]["filename"] == "notes.md" + + def test_create_rejects_oversized_image_and_rolls_back(self, app_client): + from turnstone.core.attachments import IMAGE_SIZE_CAP + + client, _, gq = app_client + oversized = PNG_1x1 + b"\x00" * (IMAGE_SIZE_CAP + 1) + meta = {"name": "fails"} + resp = client.post( + "/v1/api/workstreams/new", + data={"meta": json.dumps(meta)}, + files=[("file", ("big.png", oversized, "image/png"))], + headers=_auth("userA"), + ) + assert resp.status_code == 413 + # Regression: ws_created must NOT have been emitted for a request + # that's about to be rejected. Otherwise SSE consumers see a + # phantom workstream flash on dashboards. + events: list[dict] = [] + while not gq.empty(): + events.append(gq.get_nowait()) + kinds = {e.get("type") for e in events} + assert "ws_created" not in kinds, f"phantom ws_created emitted for failed create: {events}" + + def test_create_missing_meta_returns_400(self, app_client): + client, _, _gq = app_client + resp = client.post( + "/v1/api/workstreams/new", + files=[("file", ("notes.md", b"hello", "text/markdown"))], + headers=_auth("userA"), + ) + assert resp.status_code == 400 + + def test_create_invalid_meta_json_returns_400(self, app_client): + client, _, _gq = app_client + resp = client.post( + "/v1/api/workstreams/new", + data={"meta": "{not json}"}, + files=[], + headers=_auth("userA"), + ) + assert resp.status_code == 400 + + def test_attachments_with_resume_ws_returns_400(self, app_client): + from turnstone.core.memory import register_workstream + + client, _, _gq = app_client + register_workstream("ws-resume-target", name="resume target") + meta = {"name": "fork", "resume_ws": "ws-resume-target"} + resp = client.post( + "/v1/api/workstreams/new", + data={"meta": json.dumps(meta)}, + files=[("file", ("notes.md", b"hello", "text/markdown"))], + headers=_auth("userA"), + ) + assert resp.status_code == 400 + + +class TestCreateJsonStillWorks: + """The JSON path must remain byte-for-byte identical (back-compat).""" + + def test_create_json_no_attachments(self, app_client): + client, _, _gq = app_client + resp = client.post( + "/v1/api/workstreams/new", + json={"name": "json-only"}, + headers=_auth("userA"), + ) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["ws_id"] + # New optional field, but always emitted (empty list when absent) + assert data["attachment_ids"] == [] diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index 428f7bd2..18a76972 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -26,7 +26,35 @@ class SendRequest(BaseModel): class SendResponse(BaseModel): - status: str = Field(description="'ok' or 'busy'", examples=["ok", "busy"]) + status: str = Field( + description="'ok', 'busy', 'queued', or 'queue_full'", + examples=["ok", "busy", "queued", "queue_full"], + ) + attached_ids: list[str] = Field( + default_factory=list, + 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." + ), + ) + dropped_attachment_ids: list[str] = Field( + default_factory=list, + 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; the " + "client can retry uploads or surface a partial-attach warning." + ), + ) + priority: str | None = Field( + default=None, + description="Set on `queued` responses: relative priority of the queued message.", + ) + msg_id: str | None = Field( + default=None, + description="Set on `queued` responses: id used to dequeue the message.", + ) class AttachmentInfo(BaseModel): @@ -95,6 +123,23 @@ class CreateWorkstreamRequest(BaseModel): default="", description="Client surface type (web, cli, chat). Defaults to web for server-created sessions.", ) + initial_message: str = Field( + default="", + description=( + "Optional first user message dispatched as a background turn after " + "the workstream is created. When attachments are also provided " + "(via the multipart variant), they are reserved onto this turn." + ), + ) + ws_id: str = Field( + default="", + description=( + "Optional caller-supplied workstream id (32-hex). Required when " + "creating with attachments via the cluster routing layer so the " + "console can hash to the owning node before the multipart body " + "lands. Auto-generated when omitted." + ), + ) class CreateWorkstreamResponse(BaseModel): @@ -104,6 +149,14 @@ class CreateWorkstreamResponse(BaseModel): message_count: int = Field( default=0, description="Number of messages in the resumed workstream" ) + attachment_ids: list[str] = Field( + default_factory=list, + description=( + "Ids of attachments saved by this request (multipart variant only). " + "Already reserved onto the initial_message turn when one was provided; " + "otherwise left pending for a follow-up POST /v1/api/send." + ), + ) class CloseWorkstreamRequest(BaseModel): diff --git a/turnstone/api/server_spec.py b/turnstone/api/server_spec.py index 7c3cc3d9..9ff91154 100644 --- a/turnstone/api/server_spec.py +++ b/turnstone/api/server_spec.py @@ -64,9 +64,19 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [ "/v1/api/workstreams/new", "POST", "Create a new workstream", + description=( + "Accepts two content types. Default is `application/json` with a " + "`CreateWorkstreamRequest` body. Alternatively, `multipart/form-data` " + "with one `meta` field (JSON-encoded `CreateWorkstreamRequest` shape) " + "plus zero-or-more `file` parts saves each file as an attachment " + "under the new workstream. When `initial_message` is also set, " + "attachments are reserved onto that turn before the worker thread " + "dispatches; otherwise they remain pending for a follow-up " + "`POST /v1/api/send`." + ), request_model=CreateWorkstreamRequest, response_model=CreateWorkstreamResponse, - error_codes=[400], + error_codes=[400, 409, 413], tags=["Workstreams"], ), EndpointSpec( diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 185c62a4..1017aed4 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -625,7 +625,13 @@ def _record_route( async def route_create(request: Request) -> Response: - """POST /v1/api/route/workstreams/new — create via hash-ring routing.""" + """POST /v1/api/route/workstreams/new — create via hash-ring routing. + + Accepts both `application/json` and `multipart/form-data`. Multipart + callers must include ``?ws_id=`` in the URL query string so the + console can hash to the owning node before the multipart body lands — + we do not parse the body just to peek at the metadata. + """ t0 = time.monotonic() router: ConsoleRouter | None = request.app.state.router ring_ready = router is not None and router.is_ready() @@ -649,88 +655,102 @@ async def route_create(request: Request) -> Response: ) assert router is not None - try: - body = await request.json() - except Exception: - return _record_route( - request, - "create", - 400, - t0, - JSONResponse( - {"error": "Invalid JSON body"}, - status_code=400, - ), - ) - + raw_content_type = request.headers.get("content-type") or "" + is_multipart = raw_content_type.lower().startswith("multipart/form-data") client: httpx.AsyncClient = request.app.state.proxy_client headers = _proxy_auth_headers(request) pin = False + body: dict[str, Any] = {} + raw_body: bytes = b"" - try: - if body.get("resume_ws"): - ref = router.route(body["resume_ws"]) - elif body.get("target_node"): - ws_id = router.generate_ws_id_for_node(body["target_node"]) - body["ws_id"] = ws_id - ref = router.route(ws_id) - pin = True - else: - ws_id = secrets.token_hex(16) - body["ws_id"] = ws_id - ref = router.route(ws_id) - except NoAvailableNodeError: - return _record_route( - request, - "create", - 503, - t0, - JSONResponse( - {"error": "No available node for routing"}, - status_code=503, - ), - ) - - try: - resp = await client.post(f"{ref.url}/v1/api/workstreams/new", json=body, headers=headers) - except httpx.HTTPError: - return _record_route( - request, - "create", - 502, - t0, - JSONResponse( - {"error": f"upstream node {ref.node_id} unreachable"}, - status_code=502, - ), - ) - - # 503 retry with a new ws_id that hashes to a different node - if resp.status_code == 503 and not pin and not body.get("resume_ws"): - failed_node = ref.node_id - found_alt = False - for _ in range(10): - ws_id = secrets.token_hex(16) - try: - ref = router.route(ws_id) - except NoAvailableNodeError: - break - if ref.node_id != failed_node: - found_alt = True - break - if not found_alt: + if is_multipart: + # Multipart: caller must pass ws_id as a query param so we can + # route without parsing the body. Stream the raw bytes through + # to the upstream so we don't lose the multipart framing. + ws_id = request.query_params.get("ws_id", "").strip() + if not ws_id: return _record_route( request, "create", - resp.status_code, + 400, t0, - Response( - content=resp.content, - status_code=resp.status_code, - headers=dict(resp.headers), + JSONResponse( + {"error": "ws_id query parameter required for multipart create"}, + status_code=400, ), ) - body["ws_id"] = ws_id + try: + ref = router.route(ws_id) + except NoAvailableNodeError: + return _record_route( + request, + "create", + 503, + t0, + JSONResponse( + {"error": "No available node for routing"}, + status_code=503, + ), + ) + raw_body = await request.body() + # Forward the raw header verbatim — the multipart `boundary=` parameter + # is case-sensitive and must match the bytes in the body exactly. + upstream_headers = {**headers, "Content-Type": raw_content_type} + try: + resp = await client.post( + f"{ref.url}/v1/api/workstreams/new", + content=raw_body, + headers=upstream_headers, + ) + except httpx.HTTPError: + return _record_route( + request, + "create", + 502, + t0, + JSONResponse( + {"error": f"upstream node {ref.node_id} unreachable"}, + status_code=502, + ), + ) + else: + try: + body = await request.json() + except Exception: + return _record_route( + request, + "create", + 400, + t0, + JSONResponse( + {"error": "Invalid JSON body"}, + status_code=400, + ), + ) + try: + if body.get("resume_ws"): + ref = router.route(body["resume_ws"]) + elif body.get("target_node"): + ws_id = router.generate_ws_id_for_node(body["target_node"]) + body["ws_id"] = ws_id + ref = router.route(ws_id) + pin = True + else: + ws_id = secrets.token_hex(16) + body["ws_id"] = ws_id + ref = router.route(ws_id) + except NoAvailableNodeError: + return _record_route( + request, + "create", + 503, + t0, + JSONResponse( + {"error": "No available node for routing"}, + status_code=503, + ), + ) + try: resp = await client.post( f"{ref.url}/v1/api/workstreams/new", json=body, headers=headers @@ -747,6 +767,50 @@ async def route_create(request: Request) -> Response: ), ) + # 503 retry with a new ws_id that hashes to a different node. + # Multipart variant skips this branch — the body is bound to the + # ws_id the caller chose, so re-routing would mean re-uploading. + if resp.status_code == 503 and not pin and not body.get("resume_ws"): + failed_node = ref.node_id + found_alt = False + for _ in range(10): + ws_id = secrets.token_hex(16) + try: + ref = router.route(ws_id) + except NoAvailableNodeError: + break + if ref.node_id != failed_node: + found_alt = True + break + if not found_alt: + return _record_route( + request, + "create", + resp.status_code, + t0, + Response( + content=resp.content, + status_code=resp.status_code, + headers=dict(resp.headers), + ), + ) + body["ws_id"] = ws_id + try: + resp = await client.post( + f"{ref.url}/v1/api/workstreams/new", json=body, headers=headers + ) + except httpx.HTTPError: + return _record_route( + request, + "create", + 502, + t0, + JSONResponse( + {"error": f"upstream node {ref.node_id} unreachable"}, + status_code=502, + ), + ) + if resp.status_code == 200: data = resp.json() data["node_url"] = ref.url @@ -765,6 +829,132 @@ async def route_create(request: Request) -> Response: ) +async def route_attachment_proxy(request: Request) -> Response: + """Proxy ws-id-keyed attachment endpoints through the hash-ring router. + + Handles all four shapes mounted under + ``/v1/api/route/workstreams/{ws_id}/attachments[/...]``: + + - ``POST .../attachments`` — multipart upload (raw-body forward) + - ``GET .../attachments`` — list pending (JSON pass-through) + - ``GET .../attachments/{attachment_id}/content`` — raw bytes + - ``DELETE .../attachments/{attachment_id}`` — JSON pass-through + + All variants forward ``Content-Type`` + auth headers so multipart + framing survives, and propagate upstream response headers so the + ``Content-Disposition`` / ``X-Content-Type-Options`` set by + ``get_attachment_content`` reach the original caller intact. + """ + method = "attach" + t0 = time.monotonic() + router: ConsoleRouter | None = request.app.state.router + ring_ready = router is not None and router.is_ready() + if not ring_ready: + if router is not None: + await asyncio.to_thread(router.refresh_cache) + ring_ready = router.is_ready() + if not ring_ready: + return _record_route( + request, + method, + 503, + t0, + JSONResponse( + {"error": "Cluster routing not initialized"}, + status_code=503, + ), + ) + assert router is not None + + ws_id = request.path_params.get("ws_id", "").strip() + if not ws_id: + return _record_route( + request, + method, + 400, + t0, + JSONResponse({"error": "ws_id required"}, status_code=400), + ) + try: + ref = router.route(ws_id) + except (NoAvailableNodeError, ValueError): + return _record_route( + request, + method, + 503, + t0, + JSONResponse({"error": "routing failed"}, status_code=503), + ) + + upstream_path = request.url.path.replace("/api/route/", "/api/", 1) + if request.url.query: + upstream_path += f"?{request.url.query}" + + client: httpx.AsyncClient = request.app.state.proxy_client + headers = _proxy_auth_headers(request) + upstream_headers: dict[str, str] = dict(headers) + if request.method in ("POST", "PUT", "DELETE"): + upstream_headers["Content-Type"] = request.headers.get( + "content-type", "application/octet-stream" + ) + body = await request.body() + try: + resp = await client.request( + request.method, + f"{ref.url}{upstream_path}", + content=body, + headers=upstream_headers, + ) + except httpx.HTTPError: + return _record_route( + request, + method, + 502, + t0, + JSONResponse( + {"error": f"upstream node {ref.node_id} unreachable"}, + status_code=502, + ), + ) + else: + try: + resp = await client.get(f"{ref.url}{upstream_path}", headers=upstream_headers) + except httpx.HTTPError: + return _record_route( + request, + method, + 502, + t0, + JSONResponse( + {"error": f"upstream node {ref.node_id} unreachable"}, + status_code=502, + ), + ) + + # Preserve upstream headers — Content-Disposition + CSP set by the + # /content handler must reach the original caller, and the upstream + # already produced the correct Content-Type for both JSON and binary + # payloads. Drop hop-by-hop headers that the underlying transport + # will manage itself. + response_headers = { + k: v + for k, v in resp.headers.items() + if k.lower() + not in {"transfer-encoding", "content-encoding", "connection", "content-length"} + } + return _record_route( + request, + method, + resp.status_code, + t0, + Response( + content=resp.content, + status_code=resp.status_code, + headers=response_headers, + ), + ) + + async def route_proxy(request: Request) -> Response: """Generic routing proxy for send/approve/cancel/command/close.""" t0 = time.monotonic() @@ -7453,6 +7643,21 @@ def create_app( Route("/api/route/command", route_proxy, methods=["POST"]), Route("/api/route/plan", route_proxy, methods=["POST"]), Route("/api/route/workstreams/close", route_proxy, methods=["POST"]), + Route( + "/api/route/workstreams/{ws_id}/attachments", + route_attachment_proxy, + methods=["POST", "GET"], + ), + Route( + "/api/route/workstreams/{ws_id}/attachments/{attachment_id}", + route_attachment_proxy, + methods=["DELETE"], + ), + Route( + "/api/route/workstreams/{ws_id}/attachments/{attachment_id}/content", + route_attachment_proxy, + methods=["GET"], + ), Route("/api/route", route_lookup, methods=["GET"]), Route("/api/models", list_available_models), Route("/api/skills", list_skills_summary), diff --git a/turnstone/core/web_helpers.py b/turnstone/core/web_helpers.py index b2ed6834..1828a1f2 100644 --- a/turnstone/core/web_helpers.py +++ b/turnstone/core/web_helpers.py @@ -37,6 +37,120 @@ async def read_json_or_400(request: Request) -> dict[str, Any] | JSONResponse: return _JSONResponse({"error": "Failed to read request body"}, status_code=500) +async def read_multipart_create_or_400( + request: Request, + *, + meta_field: str = "meta", + file_field: str = "file", + max_files: int = 10, + max_per_file_bytes: int | None = None, + max_total_bytes: int | None = None, +) -> tuple[dict[str, Any], list[tuple[str, str, bytes]]] | JSONResponse: + """Parse a multipart create-with-attachments body. + + Expects one ``meta`` field (JSON-encoded object with create metadata) + and zero-or-more ``file`` parts (standard UploadFile objects). Returns + ``(meta_dict, [(filename, content_type, bytes), ...])`` on success or a + ``JSONResponse`` (400/413) on failure. + + Enforces a cheap ``Content-Length`` pre-check against *max_total_bytes* + when the header is sensible, and (when ``max_per_file_bytes`` is set) a + generic per-file cap as a defense-in-depth gate. The caller still + classifies each file and applies any kind-specific cap on top. + """ + from starlette.datastructures import UploadFile + from starlette.responses import JSONResponse as _JSONResponse + + if max_total_bytes is not None: + cl_raw = request.headers.get("content-length") + if cl_raw: + try: + cl = int(cl_raw) + except ValueError: + cl = -1 + if cl > int(max_total_bytes * 1.1): + return _JSONResponse( + { + "error": ( + f"Request body too large ({cl:,} bytes by Content-Length); " + f"cap is {max_total_bytes:,} bytes." + ), + "code": "too_large", + }, + status_code=413, + ) + + try: + form = await request.form() + except Exception: + import structlog + + structlog.get_logger(__name__).warning( + "read_multipart_create_or_400.parse_failed", exc_info=True + ) + return _JSONResponse({"error": "Invalid multipart body"}, status_code=400) + + meta_raw = form.get(meta_field) + if not isinstance(meta_raw, str): + return _JSONResponse({"error": f"Missing '{meta_field}' JSON field"}, status_code=400) + try: + meta: dict[str, Any] = json.loads(meta_raw) + except (ValueError, json.JSONDecodeError): + return _JSONResponse({"error": f"'{meta_field}' field must be valid JSON"}, status_code=400) + if not isinstance(meta, dict): + return _JSONResponse({"error": f"'{meta_field}' must be a JSON object"}, status_code=400) + + uploads = [v for v in form.getlist(file_field) if isinstance(v, UploadFile)] + if len(uploads) > max_files: + return _JSONResponse( + { + "error": (f"Too many files ({len(uploads)}); max {max_files} per request"), + "code": "too_many", + }, + status_code=400, + ) + + files: list[tuple[str, str, bytes]] = [] + running_total = 0 + try: + for upload in uploads: + filename = upload.filename or "" + content_type = upload.content_type or "application/octet-stream" + try: + data = await upload.read() + except Exception: + return _JSONResponse({"error": "Failed to read upload"}, status_code=400) + if max_per_file_bytes is not None and len(data) > max_per_file_bytes: + return _JSONResponse( + { + "error": ( + f"File too large ({len(data):,} bytes); " + f"cap is {max_per_file_bytes:,} bytes." + ), + "code": "too_large", + }, + status_code=413, + ) + running_total += len(data) + if max_total_bytes is not None and running_total > max_total_bytes: + return _JSONResponse( + { + "error": ( + f"Request body too large ({running_total:,} bytes total); " + f"cap is {max_total_bytes:,} bytes." + ), + "code": "too_large", + }, + status_code=413, + ) + files.append((filename, content_type, data)) + finally: + for upload in uploads: + await upload.close() + + return meta, files + + async def read_multipart_file_or_400( request: Request, field: str = "file", diff --git a/turnstone/sdk/__init__.py b/turnstone/sdk/__init__.py index 8c131221..8e965a47 100644 --- a/turnstone/sdk/__init__.py +++ b/turnstone/sdk/__init__.py @@ -12,7 +12,7 @@ Quick start:: from __future__ import annotations -from turnstone.sdk._types import TurnResult, TurnstoneAPIError +from turnstone.sdk._types import AttachmentUpload, TurnResult, TurnstoneAPIError from turnstone.sdk.console import AsyncTurnstoneConsole, TurnstoneConsole from turnstone.sdk.events import ( ApproveRequestEvent, @@ -54,6 +54,7 @@ __all__ = [ "AsyncTurnstoneConsole", "TurnstoneConsole", # Result types + "AttachmentUpload", "TurnResult", "TurnstoneAPIError", # Server events diff --git a/turnstone/sdk/_base.py b/turnstone/sdk/_base.py index f4947eed..13b069c2 100644 --- a/turnstone/sdk/_base.py +++ b/turnstone/sdk/_base.py @@ -83,6 +83,8 @@ class _BaseClient: *, json_body: dict[str, Any] | None = ..., params: dict[str, Any] | None = ..., + files: list[tuple[str, tuple[str, bytes, str]]] | None = ..., + data: dict[str, Any] | None = ..., response_model: type[T], ) -> T: ... @@ -94,6 +96,8 @@ class _BaseClient: *, json_body: dict[str, Any] | None = ..., params: dict[str, Any] | None = ..., + files: list[tuple[str, tuple[str, bytes, str]]] | None = ..., + data: dict[str, Any] | None = ..., response_model: None = ..., ) -> dict[str, Any]: ... @@ -104,22 +108,40 @@ class _BaseClient: *, json_body: dict[str, Any] | None = None, params: dict[str, Any] | None = None, + files: list[tuple[str, tuple[str, bytes, str]]] | None = None, + data: dict[str, Any] | None = None, response_model: type[Any] | None = None, ) -> Any: """Execute an HTTP request and return parsed response data. - Raises :class:`TurnstoneAPIError` on non-2xx responses. + When *files* is provided, the request is sent as + ``multipart/form-data`` with the named file parts (and any + *data* fields as plain form fields). Mutually exclusive with + *json_body*. Raises :class:`TurnstoneAPIError` on non-2xx + responses. """ headers: dict[str, str] | None = None if self._token_factory is not None: headers = {"Authorization": f"Bearer {self._token_factory()}"} - resp = await self._client.request( - method, - path, - json=json_body, - params=params, - headers=headers, - ) + if files is not None: + # httpx infers multipart from the files= kwarg and sets the + # Content-Type + boundary itself; do not pass json= alongside. + resp = await self._client.request( + method, + path, + files=files, + data=data, + params=params, + headers=headers, + ) + else: + resp = await self._client.request( + method, + path, + json=json_body, + params=params, + headers=headers, + ) if resp.status_code >= 400: # Try to extract error message from JSON body msg = "" @@ -129,10 +151,37 @@ class _BaseClient: if not msg: msg = resp.text[:200] raise TurnstoneAPIError(resp.status_code, msg) - data: dict[str, Any] = resp.json() + body_data: dict[str, Any] = resp.json() if response_model is not None: - return response_model.model_validate(data) - return data + return response_model.model_validate(body_data) + return body_data + + async def _request_bytes( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + ) -> bytes: + """Execute a request and return the raw response bytes. + + Used for endpoints like the attachment ``/content`` route that + return arbitrary binary or text payloads with their own + ``Content-Type``. Raises :class:`TurnstoneAPIError` on non-2xx. + """ + headers: dict[str, str] | None = None + if self._token_factory is not None: + headers = {"Authorization": f"Bearer {self._token_factory()}"} + resp = await self._client.request(method, path, params=params, headers=headers) + if resp.status_code >= 400: + msg = "" + with contextlib.suppress(Exception): + body = resp.json() + msg = body.get("error", body.get("detail", "")) + if not msg: + msg = resp.text[:200] + raise TurnstoneAPIError(resp.status_code, msg) + return resp.content async def _stream_sse( self, diff --git a/turnstone/sdk/_types.py b/turnstone/sdk/_types.py index 48cba6cb..2365adbc 100644 --- a/turnstone/sdk/_types.py +++ b/turnstone/sdk/_types.py @@ -5,6 +5,21 @@ from __future__ import annotations from dataclasses import dataclass, field +@dataclass +class AttachmentUpload: + """A file to upload as an attachment. + + Used by ``upload_attachment`` and by ``create_workstream(attachments=...)``. + ``mime_type`` is advisory — the server applies its own magic-byte + sniffing for images and UTF-8 validation for text documents and + rejects anything that doesn't match its allowlist. + """ + + filename: str + data: bytes + mime_type: str | None = None + + @dataclass class TurnResult: """Aggregated result of a send_and_wait call. diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py index c7d4d0a5..8eac8b5c 100644 --- a/turnstone/sdk/console.py +++ b/turnstone/sdk/console.py @@ -11,6 +11,7 @@ Usage:: from __future__ import annotations +import secrets from typing import TYPE_CHECKING, Any from turnstone.api.console_schemas import ( @@ -57,6 +58,10 @@ from turnstone.api.schemas import ( ScheduleInfo, StatusResponse, ) +from turnstone.api.server_schemas import ( + ListAttachmentsResponse, + UploadAttachmentResponse, +) from turnstone.sdk._base import _BaseClient from turnstone.sdk._sync import _SyncRunner from turnstone.sdk.events import ClusterEvent @@ -68,6 +73,8 @@ if TYPE_CHECKING: import httpx + from turnstone.sdk._types import AttachmentUpload + class AsyncTurnstoneConsole(_BaseClient): """Async client for the turnstone console API.""" @@ -209,11 +216,16 @@ class AsyncTurnstoneConsole(_BaseClient): target_node: str = "", user_id: str = "", client_type: str = "", + ws_id: str = "", + attachments: list[AttachmentUpload] | None = None, ) -> dict[str, Any]: """Create a workstream via the console's routing proxy. - Posts to /v1/api/route/workstreams/new. Returns the full response - dict including node_url and node_id. + Posts to /v1/api/route/workstreams/new. When *attachments* is + non-empty, the request is sent as multipart and the console + routes via ``?ws_id=`` (auto-generated when not supplied) + so the body lands on the owning node directly. Returns the + full response dict including ``node_url`` and ``node_id``. """ body: dict[str, Any] = {} if name: @@ -236,8 +248,88 @@ class AsyncTurnstoneConsole(_BaseClient): body["user_id"] = user_id if client_type: body["client_type"] = client_type + + if attachments: + # The console's multipart route_create routes by `?ws_id=` only — + # it does not parse the body to honor `target_node`. Refuse the + # combination at the SDK boundary so callers don't silently get + # routed to the wrong node. + if target_node: + raise ValueError( + "target_node is not supported with attachments; " + "use ws_id (caller-generated to hash to the desired node) instead" + ) + if not ws_id: + ws_id = secrets.token_hex(16) + body["ws_id"] = ws_id + import json as _json + + files: list[tuple[str, tuple[str, bytes, str]]] = [ + ( + "file", + ( + att.filename, + att.data, + att.mime_type or "application/octet-stream", + ), + ) + for att in attachments + ] + return await self._request( + "POST", + "/v1/api/route/workstreams/new", + files=files, + data={"meta": _json.dumps(body)}, + params={"ws_id": ws_id}, + ) + + if ws_id: + body["ws_id"] = ws_id return await self._request("POST", "/v1/api/route/workstreams/new", json_body=body) + # -- routing proxy: attachments ----------------------------------------- + + async def route_upload_attachment( + self, + ws_id: str, + filename: str, + data: bytes, + *, + mime_type: str | None = None, + ) -> UploadAttachmentResponse: + files: list[tuple[str, tuple[str, bytes, str]]] = [ + ( + "file", + (filename, data, mime_type or "application/octet-stream"), + ) + ] + return await self._request( + "POST", + f"/v1/api/route/workstreams/{ws_id}/attachments", + files=files, + response_model=UploadAttachmentResponse, + ) + + async def route_list_attachments(self, ws_id: str) -> ListAttachmentsResponse: + return await self._request( + "GET", + f"/v1/api/route/workstreams/{ws_id}/attachments", + response_model=ListAttachmentsResponse, + ) + + async def route_get_attachment_content(self, ws_id: str, attachment_id: str) -> bytes: + return await self._request_bytes( + "GET", + f"/v1/api/route/workstreams/{ws_id}/attachments/{attachment_id}/content", + ) + + async def route_delete_attachment(self, ws_id: str, attachment_id: str) -> StatusResponse: + return await self._request( + "DELETE", + f"/v1/api/route/workstreams/{ws_id}/attachments/{attachment_id}", + response_model=StatusResponse, + ) + async def route_send(self, message: str, ws_id: str) -> dict[str, Any]: """Send a message via the routing proxy.""" return await self._request( @@ -1079,6 +1171,8 @@ class TurnstoneConsole: target_node: str = "", user_id: str = "", client_type: str = "", + ws_id: str = "", + attachments: list[AttachmentUpload] | None = None, ) -> dict[str, Any]: return self._runner.run( self._async.route_create_workstream( @@ -1092,12 +1186,37 @@ class TurnstoneConsole: target_node=target_node, user_id=user_id, client_type=client_type, + ws_id=ws_id, + attachments=attachments, ) ) def route_send(self, message: str, ws_id: str) -> dict[str, Any]: return self._runner.run(self._async.route_send(message, ws_id)) + # -- routing proxy: attachments ----------------------------------------- + + def route_upload_attachment( + self, + ws_id: str, + filename: str, + data: bytes, + *, + mime_type: str | None = None, + ) -> UploadAttachmentResponse: + return self._runner.run( + self._async.route_upload_attachment(ws_id, filename, data, mime_type=mime_type) + ) + + def route_list_attachments(self, ws_id: str) -> ListAttachmentsResponse: + return self._runner.run(self._async.route_list_attachments(ws_id)) + + def route_get_attachment_content(self, ws_id: str, attachment_id: str) -> bytes: + return self._runner.run(self._async.route_get_attachment_content(ws_id, attachment_id)) + + def route_delete_attachment(self, ws_id: str, attachment_id: str) -> StatusResponse: + return self._runner.run(self._async.route_delete_attachment(ws_id, attachment_id)) + def route_approve( self, *, diff --git a/turnstone/sdk/server.py b/turnstone/sdk/server.py index 74aeab1b..58a47a73 100644 --- a/turnstone/sdk/server.py +++ b/turnstone/sdk/server.py @@ -14,6 +14,7 @@ from __future__ import annotations import asyncio import contextlib +import secrets from typing import TYPE_CHECKING, Any from turnstone.api.schemas import ( @@ -26,6 +27,7 @@ from turnstone.api.server_schemas import ( CreateWorkstreamResponse, DashboardResponse, HealthResponse, + ListAttachmentsResponse, ListAvailableModelsResponse, ListMemoriesResponse, ListSavedWorkstreamsResponse, @@ -33,10 +35,11 @@ from turnstone.api.server_schemas import ( ListWorkstreamsResponse, MemoryInfo, SendResponse, + UploadAttachmentResponse, ) from turnstone.sdk._base import _BaseClient from turnstone.sdk._sync import _SyncRunner -from turnstone.sdk._types import TurnResult +from turnstone.sdk._types import AttachmentUpload, TurnResult from turnstone.sdk.events import ( ClusterEvent, ContentEvent, @@ -108,7 +111,19 @@ class AsyncTurnstoneServer(_BaseClient): ws_id: str = "", client_type: str = "", notify_targets: str = "", + attachments: list[AttachmentUpload] | None = None, ) -> CreateWorkstreamResponse: + """Create a new workstream. + + When *attachments* is non-empty the request is sent as + ``multipart/form-data`` with the metadata in a ``meta`` JSON + field and one ``file`` part per attachment. A ws_id is + auto-generated client-side when not supplied so cluster-routed + callers can bind the body to the owning node up front. When + *initial_message* is also set, the server reserves the + attachments onto that turn before its background worker + dispatches. + """ body: dict[str, Any] = {} if name: body["name"] = name @@ -126,12 +141,38 @@ class AsyncTurnstoneServer(_BaseClient): body["auto_approve_tools"] = auto_approve_tools if user_id: body["user_id"] = user_id - if ws_id: - body["ws_id"] = ws_id if client_type: body["client_type"] = client_type if notify_targets and notify_targets != "[]": body["notify_targets"] = notify_targets + + if attachments: + if not ws_id: + ws_id = secrets.token_hex(16) + body["ws_id"] = ws_id + import json as _json + + files: list[tuple[str, tuple[str, bytes, str]]] = [ + ( + "file", + ( + att.filename, + att.data, + att.mime_type or "application/octet-stream", + ), + ) + for att in attachments + ] + return await self._request( + "POST", + "/v1/api/workstreams/new", + files=files, + data={"meta": _json.dumps(body)}, + response_model=CreateWorkstreamResponse, + ) + + if ws_id: + body["ws_id"] = ws_id return await self._request( "POST", "/v1/api/workstreams/new", @@ -149,14 +190,76 @@ class AsyncTurnstoneServer(_BaseClient): # -- chat interaction ---------------------------------------------------- - async def send(self, message: str, ws_id: str) -> SendResponse: + async def send( + self, + message: str, + ws_id: str, + *, + attachment_ids: list[str] | None = None, + ) -> SendResponse: + body: dict[str, Any] = {"message": message, "ws_id": ws_id} + if attachment_ids is not None: + body["attachment_ids"] = list(attachment_ids) return await self._request( "POST", "/v1/api/send", - json_body={"message": message, "ws_id": ws_id}, + json_body=body, response_model=SendResponse, ) + # -- attachments --------------------------------------------------------- + + async def upload_attachment( + self, + ws_id: str, + filename: str, + data: bytes, + *, + mime_type: str | None = None, + ) -> UploadAttachmentResponse: + """Upload one file as a pending attachment for this workstream. + + The server validates size + MIME (magic-byte sniff for images, + UTF-8 decode for text) and rejects with 400/413 on any mismatch. + Returns the persisted ``AttachmentInfo`` so the caller can pass + the id into a subsequent ``send(attachment_ids=...)``. + """ + files: list[tuple[str, tuple[str, bytes, str]]] = [ + ( + "file", + (filename, data, mime_type or "application/octet-stream"), + ) + ] + return await self._request( + "POST", + f"/v1/api/workstreams/{ws_id}/attachments", + files=files, + response_model=UploadAttachmentResponse, + ) + + async def list_attachments(self, ws_id: str) -> ListAttachmentsResponse: + """List the caller's pending (unconsumed) attachments for *ws_id*.""" + return await self._request( + "GET", + f"/v1/api/workstreams/{ws_id}/attachments", + response_model=ListAttachmentsResponse, + ) + + async def get_attachment_content(self, ws_id: str, attachment_id: str) -> bytes: + """Return the raw bytes of an attachment.""" + return await self._request_bytes( + "GET", + f"/v1/api/workstreams/{ws_id}/attachments/{attachment_id}/content", + ) + + async def delete_attachment(self, ws_id: str, attachment_id: str) -> StatusResponse: + """Remove a pending attachment. Consumed attachments return 404.""" + return await self._request( + "DELETE", + f"/v1/api/workstreams/{ws_id}/attachments/{attachment_id}", + response_model=StatusResponse, + ) + async def approve( self, *, @@ -494,6 +597,7 @@ class TurnstoneServer: ws_id: str = "", client_type: str = "", notify_targets: str = "", + attachments: list[AttachmentUpload] | None = None, ) -> CreateWorkstreamResponse: return self._runner.run( self._async.create_workstream( @@ -508,6 +612,7 @@ class TurnstoneServer: ws_id=ws_id, client_type=client_type, notify_targets=notify_targets, + attachments=attachments, ) ) @@ -516,8 +621,37 @@ class TurnstoneServer: # -- chat interaction ---------------------------------------------------- - def send(self, message: str, ws_id: str) -> SendResponse: - return self._runner.run(self._async.send(message, ws_id)) + def send( + self, + message: str, + ws_id: str, + *, + attachment_ids: list[str] | None = None, + ) -> SendResponse: + return self._runner.run(self._async.send(message, ws_id, attachment_ids=attachment_ids)) + + # -- attachments --------------------------------------------------------- + + def upload_attachment( + self, + ws_id: str, + filename: str, + data: bytes, + *, + mime_type: str | None = None, + ) -> UploadAttachmentResponse: + return self._runner.run( + self._async.upload_attachment(ws_id, filename, data, mime_type=mime_type) + ) + + def list_attachments(self, ws_id: str) -> ListAttachmentsResponse: + return self._runner.run(self._async.list_attachments(ws_id)) + + def get_attachment_content(self, ws_id: str, attachment_id: str) -> bytes: + return self._runner.run(self._async.get_attachment_content(ws_id, attachment_id)) + + def delete_attachment(self, ws_id: str, attachment_id: str) -> StatusResponse: + return self._runner.run(self._async.delete_attachment(ws_id, attachment_id)) def approve( self, diff --git a/turnstone/server.py b/turnstone/server.py index 3d1f0104..dbec4e43 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -2081,14 +2081,196 @@ def _deliver_notification( time.sleep(1.0 if attempt == 0 else 3.0) -async def create_workstream(request: Request) -> JSONResponse: - """POST /v1/api/workstreams/new — create a new workstream.""" - from turnstone.core.memory import get_workstream_display_name - from turnstone.core.web_helpers import read_json_or_400 +def _reserve_and_resolve_attachments( + requested_ids: list[str], + send_id: str, + ws_id: str, + user_id: str, +) -> tuple[list[Any], list[str], list[str]]: + """Reserve attachment ids for ``send_id`` and resolve to Attachment objects. - body = await read_json_or_400(request) - if isinstance(body, JSONResponse): - return body + Returns ``(resolved, ordered_reserved, dropped)``. ``dropped`` is the + subset of *requested_ids* that could not be reserved (already consumed, + lost a race, or cross-scope). Used by the create-with-attachments + path; ``send_message`` has its own inlined variant with an + auto-consume fast path that reuses bytes fetched during selection. + """ + from turnstone.core.attachments import Attachment + from turnstone.core.memory import get_attachments as _get_attachments + from turnstone.core.memory import reserve_attachments as _reserve + + if not requested_ids: + return [], [], [] + + reserved_ids: list[str] = _reserve(requested_ids, send_id, ws_id, user_id) + reserved_set = set(reserved_ids) + ordered_reserved: list[str] = [aid for aid in requested_ids if aid in reserved_set] + dropped: list[str] = [aid for aid in requested_ids if aid not in reserved_set] + + resolved: list[Any] = [] + if ordered_reserved: + rows = _get_attachments(ordered_reserved) + rows_by_id = {str(r["attachment_id"]): r for r in rows} + for aid in ordered_reserved: + r = rows_by_id.get(aid) + if not r: + continue + if ( + r.get("ws_id") != ws_id + or r.get("user_id") != user_id + or r.get("message_id") is not None + or r.get("reserved_for_msg_id") != send_id + ): + continue + content = r.get("content") + if not isinstance(content, bytes): + continue + resolved.append( + Attachment( + attachment_id=str(r["attachment_id"]), + filename=str(r.get("filename") or ""), + mime_type=str(r.get("mime_type") or "application/octet-stream"), + kind=str(r.get("kind") or ""), + content=content, + ) + ) + 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. + + 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``. + + 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 turnstone.core.attachments import ( + IMAGE_SIZE_CAP, + MAX_PENDING_ATTACHMENTS_PER_USER_WS, + TEXT_DOC_SIZE_CAP, + ) + from turnstone.core.memory import list_pending_attachments, save_attachment + + saved_ids: list[str] = [] + if not files: + return saved_ids, None + + lock = _attachment_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.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: WorkstreamManager = request.app.state.workstreams skip: bool = request.app.state.skip_permissions auth = getattr(getattr(request, "state", None), "auth_result", None) @@ -2134,6 +2316,16 @@ async def create_workstream(request: Request) -> JSONResponse: 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. + if uploaded_files and resume_ws_id: + return JSONResponse( + {"error": "attachments cannot be combined with resume_ws"}, + status_code=400, + ) try: ws = mgr.create( name=body.get("name", ""), @@ -2156,9 +2348,29 @@ async def create_workstream(request: Request) -> JSONResponse: ws.session.set_watch_runner( runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui) ) - # Emit creation event on global queue for SSE consumers (console) - display_name = get_workstream_display_name(ws.id) or ws.name gq: queue.Queue[dict[str, Any]] = request.app.state.global_queue + + # 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 + + # 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( { @@ -2284,11 +2496,33 @@ async def create_workstream(request: Request) -> JSONResponse: 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) + 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") @@ -2309,6 +2543,7 @@ async def create_workstream(request: Request) -> JSONResponse: "name": ws.name, "resumed": resumed, "message_count": message_count, + "attachment_ids": attachment_ids, } ) except RuntimeError as e: diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js index f3a834f1..7964c505 100644 --- a/turnstone/ui/static/app.js +++ b/turnstone/ui/static/app.js @@ -3147,6 +3147,141 @@ function switchTab(wsId) { var _newWsTrapHandler = null; var _forkFromWsId = ""; +// Staged files for the new-workstream modal. Distinct from the pane's +// chip strip: there's no ws_id yet, so we hold File objects in memory +// and ship them all in one multipart create request on submit. +var _newWsStagedFiles = []; + +// Per-kind size caps (mirrored from turnstone/core/attachments.py so the +// browser can fail fast before uploading). Keep in sync. +var _NEW_WS_IMAGE_CAP = 4 * 1024 * 1024; +var _NEW_WS_TEXT_CAP = 512 * 1024; +var _NEW_WS_MAX_FILES = 10; + +function _newWsRenderChips() { + var chipsEl = document.getElementById("new-ws-attach-chips"); + if (!chipsEl) return; + chipsEl.textContent = ""; + for (var i = 0; i < _newWsStagedFiles.length; i++) { + (function (idx) { + var f = _newWsStagedFiles[idx]; + var chip = document.createElement("span"); + chip.className = "new-ws-attach-chip"; + chip.setAttribute("role", "listitem"); + var label = document.createElement("span"); + label.className = "new-ws-attach-chip-name"; + label.textContent = f.name; + label.title = f.name + " (" + f.size + " bytes)"; + chip.appendChild(label); + var size = document.createElement("span"); + size.className = "new-ws-attach-chip-size"; + size.textContent = _formatAttachSize(f.size); + chip.appendChild(size); + var rm = document.createElement("button"); + rm.type = "button"; + rm.className = "new-ws-attach-chip-remove"; + rm.setAttribute("aria-label", "Remove " + f.name); + rm.textContent = "\u00d7"; + rm.onclick = function () { + _newWsStagedFiles.splice(idx, 1); + _newWsRenderChips(); + }; + chip.appendChild(rm); + chipsEl.appendChild(chip); + })(i); + } +} + +// Mirrors turnstone/server.py classifier — magic-byte image allowlist plus +// text/* MIMEs, allowlisted application/* MIMEs, and known text extensions. +// Surfaces unsupported types client-side so the user sees a clear error +// instead of a generic create failure after the server rejects. +var _ATTACH_IMAGE_MIMES = [ + "image/png", + "image/jpeg", + "image/gif", + "image/webp", +]; +var _ATTACH_TEXT_APP_MIMES = [ + "application/json", + "application/xml", + "application/x-yaml", + "application/yaml", + "application/toml", +]; +var _ATTACH_TEXT_EXTENSIONS = [ + ".c", + ".conf", + ".cpp", + ".css", + ".go", + ".h", + ".hpp", + ".html", + ".ini", + ".java", + ".js", + ".json", + ".jsx", + ".md", + ".py", + ".rs", + ".sh", + ".sql", + ".toml", + ".ts", + ".tsx", + ".txt", + ".xml", + ".yaml", + ".yml", +]; + +function _isAttachmentAllowed(file) { + var mime = (file.type || "").toLowerCase(); + if (_ATTACH_IMAGE_MIMES.indexOf(mime) !== -1) return true; + if (mime.indexOf("text/") === 0) return true; + if (_ATTACH_TEXT_APP_MIMES.indexOf(mime) !== -1) return true; + var name = (file.name || "").toLowerCase(); + var dot = name.lastIndexOf("."); + if (dot >= 0 && _ATTACH_TEXT_EXTENSIONS.indexOf(name.substr(dot)) !== -1) { + return true; + } + return false; +} + +function _newWsAddFiles(files) { + var errEl = document.getElementById("new-ws-error"); + for (var i = 0; i < files.length; i++) { + var f = files[i]; + if (_newWsStagedFiles.length >= _NEW_WS_MAX_FILES) { + errEl.textContent = + "At most " + _NEW_WS_MAX_FILES + " attachments per workstream"; + errEl.style.display = "block"; + return; + } + if (!_isAttachmentAllowed(f)) { + errEl.textContent = + "Unsupported file type: " + + f.name + + " (allowed: png/jpeg/gif/webp images, text)"; + errEl.style.display = "block"; + return; + } + var isImage = (f.type || "").indexOf("image/") === 0; + var cap = isImage ? _NEW_WS_IMAGE_CAP : _NEW_WS_TEXT_CAP; + if (f.size > cap) { + errEl.textContent = + f.name + " exceeds the " + _formatAttachSize(cap) + " cap"; + errEl.style.display = "block"; + return; + } + _newWsStagedFiles.push(f); + } + errEl.style.display = "none"; + _newWsRenderChips(); +} + function newWorkstream() { showNewWsModal(); } @@ -3244,6 +3379,8 @@ function showNewWsModal(forkFromWsId) { }); document.getElementById("new-ws-name").value = ""; + var initEl = document.getElementById("new-ws-initial-message"); + if (initEl) initEl.value = ""; var errEl = document.getElementById("new-ws-error"); errEl.style.display = "none"; errEl.textContent = ""; @@ -3251,6 +3388,28 @@ function showNewWsModal(forkFromWsId) { submitBtn.disabled = false; submitBtn.textContent = _forkFromWsId ? "Fork" : "Create"; + // Reset attachment staging. Forks don't carry attachments — + // disable the attach UI in that case (the fork inherits its + // parent's history; new attachments go on the next manual send). + _newWsStagedFiles = []; + var attachRow = document.getElementById("new-ws-attach-row"); + var attachInput = document.getElementById("new-ws-attach-input"); + var attachBtn = document.getElementById("new-ws-attach-btn"); + if (attachRow) attachRow.style.display = _forkFromWsId ? "none" : ""; + if (attachInput) attachInput.value = ""; + _newWsRenderChips(); + if (attachBtn && attachInput) { + attachBtn.onclick = function () { + attachInput.click(); + }; + attachInput.onchange = function () { + if (attachInput.files && attachInput.files.length) { + _newWsAddFiles(attachInput.files); + } + attachInput.value = ""; + }; + } + document.getElementById("new-ws-cancel").onclick = hideNewWsModal; submitBtn.onclick = submitNewWs; @@ -3317,20 +3476,37 @@ function submitNewWs() { var model = document.getElementById("new-ws-model").value.trim(); var judge_model = document.getElementById("new-ws-judge-model").value.trim(); var skill = document.getElementById("new-ws-skill").value; + var initEl = document.getElementById("new-ws-initial-message"); + var initial_message = initEl ? initEl.value.trim() : ""; if (name) body.name = name; if (model) body.model = model; if (judge_model) body.judge_model = judge_model; if (skill && !_forkFromWsId) body.skill = skill; if (_forkFromWsId) body.resume_ws = _forkFromWsId; + if (initial_message) body.initial_message = initial_message; var errEl = document.getElementById("new-ws-error"); errEl.style.display = "none"; - authFetch("/v1/api/workstreams/new", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify(body), - }) + var fetchOpts; + var staged = _forkFromWsId ? [] : _newWsStagedFiles.slice(); + if (staged.length > 0) { + var form = new FormData(); + form.append("meta", JSON.stringify(body)); + for (var i = 0; i < staged.length; i++) { + form.append("file", staged[i], staged[i].name); + } + // Don't set Content-Type — the browser adds the correct boundary. + fetchOpts = { method: "POST", body: form }; + } else { + fetchOpts = { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }; + } + + authFetch("/v1/api/workstreams/new", fetchOpts) .then(function (r) { return r.json(); }) @@ -3344,6 +3520,7 @@ function submitNewWs() { } if (data.ws_id) { workstreams[data.ws_id] = { name: data.name, state: "idle" }; + _newWsStagedFiles = []; hideNewWsModal(); switchTab(data.ws_id); } @@ -3542,6 +3719,8 @@ function showDashboard() { document.getElementById("tab-bar").inert = true; document.getElementById("split-root").inert = true; loadDashboard(); + _loadDashboardOptionsLists(); + _refreshDashboardSubmitLabel(); setTimeout(function () { document.getElementById("dashboard-input").focus(); }, 50); @@ -3554,6 +3733,9 @@ function hideDashboard() { document.getElementById("tab-bar").inert = false; document.getElementById("split-root").inert = false; document.getElementById("dashboard-input").value = ""; + _dashboardStagedFiles = []; + _renderDashboardChips(); + _refreshDashboardSubmitLabel(); var pane = getFocusedPane(); if (pane) pane.inputEl.focus(); } @@ -4323,57 +4505,253 @@ function dashboardResumeSession(wsId) { }); } -function dashboardNewChat() { - hideDashboard(); - newWorkstream(); +// Staged files for the dashboard composer. Reuses the same file-list pattern +// as the new-workstream modal but lives independently so the two flows don't +// stomp on each other's state. +var _dashboardStagedFiles = []; + +// Per-kind size caps mirrored from turnstone/core/attachments.py — keep in sync. +var _DASH_IMAGE_CAP = 4 * 1024 * 1024; +var _DASH_TEXT_CAP = 512 * 1024; +var _DASH_MAX_FILES = 10; + +function _renderDashboardChips() { + var chipsEl = document.getElementById("dashboard-attach-chips"); + if (!chipsEl) return; + chipsEl.textContent = ""; + for (var i = 0; i < _dashboardStagedFiles.length; i++) { + (function (idx) { + var f = _dashboardStagedFiles[idx]; + var chip = document.createElement("span"); + chip.className = "new-ws-attach-chip"; + chip.setAttribute("role", "listitem"); + var label = document.createElement("span"); + label.className = "new-ws-attach-chip-name"; + label.textContent = f.name; + label.title = f.name + " (" + f.size + " bytes)"; + chip.appendChild(label); + var size = document.createElement("span"); + size.className = "new-ws-attach-chip-size"; + size.textContent = _formatAttachSize(f.size); + chip.appendChild(size); + var rm = document.createElement("button"); + rm.type = "button"; + rm.className = "new-ws-attach-chip-remove"; + rm.setAttribute("aria-label", "Remove " + f.name); + rm.textContent = "\u00d7"; + rm.onclick = function () { + _dashboardStagedFiles.splice(idx, 1); + _renderDashboardChips(); + _refreshDashboardSubmitLabel(); + }; + chip.appendChild(rm); + chipsEl.appendChild(chip); + })(i); + } } -function dashboardSendMessage() { +function _addDashboardFiles(files) { + for (var i = 0; i < files.length; i++) { + var f = files[i]; + if (_dashboardStagedFiles.length >= _DASH_MAX_FILES) { + _dashboardError( + "At most " + _DASH_MAX_FILES + " attachments per workstream", + ); + return; + } + // Drag-drop bypasses the filter, so re-check + // against the server's allowlist before the upload roundtrip. + if (!_isAttachmentAllowed(f)) { + _dashboardError( + "Unsupported file type: " + + f.name + + " (allowed: png/jpeg/gif/webp images, text)", + ); + return; + } + var isImage = (f.type || "").indexOf("image/") === 0; + var cap = isImage ? _DASH_IMAGE_CAP : _DASH_TEXT_CAP; + if (f.size > cap) { + _dashboardError( + f.name + " exceeds the " + _formatAttachSize(cap) + " cap", + ); + return; + } + _dashboardStagedFiles.push(f); + } + _renderDashboardChips(); + _refreshDashboardSubmitLabel(); +} + +var _dashboardErrorTimer = null; + +function _dashboardError(msg) { + // Live-region message + outline. title= alone is invisible to screen + // readers and on touch devices, so we surface the message visibly + // beneath the textarea via aria-live="polite". var input = document.getElementById("dashboard-input"); + var errEl = document.getElementById("dashboard-error"); + if (errEl) { + errEl.textContent = msg; + } + if (input) { + input.classList.add("dashboard-input-error"); + } + if (_dashboardErrorTimer) clearTimeout(_dashboardErrorTimer); + _dashboardErrorTimer = setTimeout(function () { + if (input) input.classList.remove("dashboard-input-error"); + if (errEl) errEl.textContent = ""; + _dashboardErrorTimer = null; + }, 5000); +} + +function _refreshDashboardSubmitLabel() { + var btn = document.getElementById("dashboard-submit-btn"); + if (!btn) return; + var input = document.getElementById("dashboard-input"); + var hasText = input && input.value.trim().length > 0; + var hasFiles = _dashboardStagedFiles.length > 0; + btn.textContent = hasText || hasFiles ? "Send" : "Create"; +} + +function _loadDashboardOptionsLists() { + // Models + var modelSel = document.getElementById("dashboard-model"); + var judgeSel = document.getElementById("dashboard-judge-model"); + if (modelSel && modelSel.options.length <= 1) { + authFetch("/v1/api/models") + .then(function (r) { + return r.json(); + }) + .then(function (data) { + (data.models || []).forEach(function (m) { + var opt = document.createElement("option"); + opt.value = m.alias; + opt.textContent = + m.alias === m.model ? m.alias : m.alias + " (" + m.model + ")"; + modelSel.appendChild(opt); + if (judgeSel) { + var jOpt = document.createElement("option"); + jOpt.value = m.alias; + jOpt.textContent = opt.textContent; + judgeSel.appendChild(jOpt); + } + }); + }) + .catch(function () { + /* default model still works */ + }); + } + // Skills + var skillSel = document.getElementById("dashboard-skill"); + if (skillSel && skillSel.options.length <= 1) { + authFetch("/v1/api/skills") + .then(function (r) { + return r.json(); + }) + .then(function (data) { + (data.skills || []).forEach(function (t) { + var opt = document.createElement("option"); + opt.value = t.name; + var label = t.name; + if (t.is_default) label += " (default)"; + if (t.origin === "mcp") label += " [MCP]"; + opt.textContent = label; + skillSel.appendChild(opt); + }); + }) + .catch(function () { + /* ignore */ + }); + } +} + +function _toggleDashboardOptions() { + var panel = document.getElementById("dashboard-options"); + var btn = document.getElementById("dashboard-options-btn"); + if (!panel || !btn) return; + var open = !panel.hasAttribute("hidden"); + if (open) { + panel.setAttribute("hidden", ""); + btn.setAttribute("aria-expanded", "false"); + } else { + panel.removeAttribute("hidden"); + btn.setAttribute("aria-expanded", "true"); + } +} + +// Unified dashboard submit. Replaces the old "click button → modal" + +// "press Enter → quick-send-empty-config" split. One path: build the +// create payload from text + attachments + options, send it, switch. +function dashboardSubmit() { + var input = document.getElementById("dashboard-input"); + var btn = document.getElementById("dashboard-submit-btn"); var text = input.value.trim(); - if (!text) return; + var staged = _dashboardStagedFiles.slice(); + + var body = {}; + var model = document.getElementById("dashboard-model").value.trim(); + var judge = document.getElementById("dashboard-judge-model").value.trim(); + var skill = document.getElementById("dashboard-skill").value; + if (model) body.model = model; + if (judge) body.judge_model = judge; + if (skill) body.skill = skill; + if (text) body.initial_message = text; + input.disabled = true; - var btn = document.querySelector(".dashboard-new-btn"); btn.disabled = true; - authFetch("/v1/api/workstreams/new", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: "{}", - }) + + var fetchOpts; + if (staged.length > 0) { + var form = new FormData(); + form.append("meta", JSON.stringify(body)); + for (var i = 0; i < staged.length; i++) { + form.append("file", staged[i], staged[i].name); + } + fetchOpts = { method: "POST", body: form }; + } else { + fetchOpts = { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }; + } + + authFetch("/v1/api/workstreams/new", fetchOpts) .then(function (r) { return r.json(); }) .then(function (data) { - if (!data.ws_id) { - input.disabled = false; - btn.disabled = false; + input.disabled = false; + btn.disabled = false; + if (data.error || !data.ws_id) { + _dashboardError(data.error || "Failed to create workstream"); return; } workstreams[data.ws_id] = { name: data.name, state: "idle" }; switchTab(data.ws_id); hideDashboard(); - input.disabled = false; - btn.disabled = false; - var pane = getFocusedPane(); - if (pane) { - pane.setBusy(true); - pane.addUserMessage(text); - } - authFetch("/v1/api/send", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ message: text, ws_id: data.ws_id }), - }).catch(function (err) { - var p = getFocusedPane(); - if (p) { - p.addErrorMessage("Connection error: " + err.message); - p.setBusy(false); + // If we sent an initial_message, the server's worker thread already + // dispatched it. Echo into the pane so the user sees their own text + // immediately rather than waiting for SSE to backfill. + if (text) { + var pane = getFocusedPane(); + if (pane) { + pane.setBusy(true); + pane.addUserMessage(text); } - }); + } }) - .catch(function () { + .catch(function (err) { input.disabled = false; btn.disabled = false; + // authFetch throws Error("auth") when the user is signed out and the + // login modal has already been surfaced; suppress the redundant + // error toast in that case. Otherwise fall back to a generic + // string so we never render "Connection error: undefined". + if (err && err.message === "auth") return; + var detail = (err && err.message) || "Unable to reach the server"; + _dashboardError("Connection error: " + detail); }); } @@ -5229,14 +5607,81 @@ document .getElementById("plan-feedback") .addEventListener("input", _updatePlanRejectBtn); -document - .getElementById("dashboard-input") - .addEventListener("keydown", function (e) { - if (e.key === "Enter" && !e.shiftKey) { +// Dashboard composer wiring — Enter (no shift) submits, input refreshes the +// button label, paperclip + drag-drop + paste-image stage files, options +// toggle expands the dropdown panel. +(function () { + var input = document.getElementById("dashboard-input"); + var attachBtn = document.getElementById("dashboard-attach-btn"); + var attachInput = document.getElementById("dashboard-attach-input"); + var optionsBtn = document.getElementById("dashboard-options-btn"); + var composer = document.getElementById("dashboard-composer"); + if (!input) return; + + input.addEventListener("keydown", function (e) { + if (e.key === "Enter" && !e.shiftKey && !e.altKey) { e.preventDefault(); - dashboardSendMessage(); + dashboardSubmit(); } }); + input.addEventListener("input", _refreshDashboardSubmitLabel); + input.addEventListener("paste", function (e) { + if (!e.clipboardData) return; + var items = e.clipboardData.items || []; + var pasted = []; + for (var i = 0; i < items.length; i++) { + if (items[i].kind === "file") { + var f = items[i].getAsFile(); + if (f) pasted.push(f); + } + } + if (pasted.length) { + e.preventDefault(); + _addDashboardFiles(pasted); + } + }); + + if (attachBtn && attachInput) { + attachBtn.addEventListener("click", function () { + attachInput.click(); + }); + attachInput.addEventListener("change", function () { + if (attachInput.files && attachInput.files.length) { + _addDashboardFiles(attachInput.files); + } + attachInput.value = ""; + }); + } + if (optionsBtn) { + optionsBtn.addEventListener("click", _toggleDashboardOptions); + } + if (composer) { + composer.addEventListener("dragover", function (e) { + if ( + e.dataTransfer && + Array.from(e.dataTransfer.types || []).includes("Files") + ) { + e.preventDefault(); + composer.classList.add("dashboard-composer-drop"); + } + }); + composer.addEventListener("dragleave", function (e) { + if (e.target === composer) + composer.classList.remove("dashboard-composer-drop"); + }); + composer.addEventListener("drop", function (e) { + composer.classList.remove("dashboard-composer-drop"); + if ( + e.dataTransfer && + e.dataTransfer.files && + e.dataTransfer.files.length + ) { + e.preventDefault(); + _addDashboardFiles(e.dataTransfer.files); + } + }); + } +})(); document.addEventListener("keydown", function (e) { // Defer to modal's own keydown handler when any modal is open diff --git a/turnstone/ui/static/index.html b/turnstone/ui/static/index.html index 67bfc499..5030163d 100644 --- a/turnstone/ui/static/index.html +++ b/turnstone/ui/static/index.html @@ -28,10 +28,38 @@