feat: workstream attachments at creation time + SDK + UI parity (#362)

* feat: workstream attachments at creation time + SDK + UI parity

Closes the two big deferred items from PR #356: attaching files as part
of the initial workstream-creation request, and full SDK coverage of the
attachment surface.

Server: POST /v1/api/workstreams/new now accepts multipart/form-data
(meta JSON + 0..N file parts).  Files are validated and saved as pending
under the new ws; when initial_message is also set the create handler
reserves them onto that turn before the dispatch worker fires, mirroring
the /v1/api/send pattern.  Validation failure rolls back the workstream
via delete_workstream so we don't leak orphan rows or emit a phantom
ws_created/ws_closed pair on SSE.  JSON path is unchanged.

Console routing: route_create accepts multipart with ?ws_id=<hex> as a
query parameter (the console hashes the id before the body lands).
Added /v1/api/route/workstreams/{ws_id}/attachments POST/GET/DELETE +
.../{attachment_id}/content GET proxies that forward raw bytes and
preserve upstream headers (Content-Disposition, X-Content-Type-Options,
CSP sandbox).

Python + TypeScript SDKs: AttachmentUpload type, upload_attachment,
list_attachments, get_attachment_content, delete_attachment, and
send(attachment_ids=...).  create_workstream(attachments=...) sends
multipart and pre-generates a ws_id client-side so cluster routing
works.  SDKs reject attachments+target_node combinations since the
multipart route doesn't honor target_node.

Web UI: dashboard composer refactored to a single unified create flow.
Replaced the inconsistent split (Enter created+sent raw, "New Chat"
opened a modal) with one rich composer carrying a textarea, paperclip
+ chip strip, drag-drop, paste-image, and a collapsible Options panel
for model/judge_model/skill.  Submit button dynamically labels Create
vs Send.  New-workstream modal also gained the same paperclip + chip
strip + first-message field for the tab-bar + entry point.

Tests: 30 new tests across server multipart create, console route
multipart + attachment proxies, Python + TS SDK attachment surfaces,
plus regressions for the three review-flagged bugs (Content-Type
boundary preservation, attachments+target_node rejection, no phantom
ws_created on validation failure).

* fix: address Copilot review feedback on PR #362

- web_helpers: docstring now matches behaviour — read_multipart_create_or_400
  does enforce the optional max_per_file_bytes cap as defense-in-depth.
- app.js: drop the duplicated _formatAttachSize definition (one already
  exists earlier for pane chips); add a shared _isAttachmentAllowed helper
  that mirrors the server's classifier (png/jpeg/gif/webp images, text/*
  MIMEs, allowlisted application/* MIMEs, known text extensions) and call
  it from both _newWsAddFiles and _addDashboardFiles so unsupported files
  fail fast client-side instead of after a server roundtrip.
- app.js: dashboardSubmit catch now suppresses the redundant error toast
  on authFetch's "auth" Error and falls back to a generic message when
  err.message is undefined, instead of rendering "Connection error: undefined".
- SendResponse (Pydantic + TS): document and expose attached_ids,
  dropped_attachment_ids, priority, and msg_id so attachment-aware SDK
  callers can detect partial reservations and dequeue queued messages.
- test_server_attachments_on_create: drop the dual `import turnstone.server`
  + `from turnstone.server import` style — use monkeypatch.setattr by
  dotted path for module-level mutation and `from … import …` for the
  helpers, keeping a single import style.

sdk/typescript/src/base.ts

@@ -29,6 +29,12 @@ export interface ClientOptions {
 export interface RequestOptions {
   json?: object;
   params?: Record<string, string | number>;
+  /**
+   * 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<T> {
-    const headers: Record<string, string> = {
-      "Content-Type": "application/json",
-    };
+    const headers: Record<string, string> = {};
+    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<string, unknown>;
-        msg = (body.error as string) ?? (body.detail as string) ?? "";
+        const errBody = (await resp.json()) as Record<string, unknown>;
+        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<string, string | number> },
+  ): Promise<{ bytes: Uint8Array; contentType: string; filename: string }> {
+    const headers: Record<string, string> = {};
+    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<string, unknown>;
+        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, string | number>,
+  ): 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<T = Record<string, unknown>>(
     path: string,
     params?: Record<string, string | number>,

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=<hex>`
+   * (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<string, unknown> = { ...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<UploadAttachmentResponse> {
+    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<ListAttachmentsResponse> {
+    return this.request("GET", `/v1/api/route/workstreams/${wsId}/attachments`);
+  }
+
+  async routeGetAttachmentContent(
+    wsId: string,
+    attachmentId: string,
+  ): Promise<AttachmentContent> {
+    return this.requestBytes(
+      "GET",
+      `/v1/api/route/workstreams/${wsId}/attachments/${attachmentId}/content`,
+    );
+  }
+
+  async routeDeleteAttachment(
+    wsId: string,
+    attachmentId: string,
+  ): Promise<StatusResponse> {
+    return this.request(
+      "DELETE",
+      `/v1/api/route/workstreams/${wsId}/attachments/${attachmentId}`,
+    );
+  }
+
   // -- Streaming ------------------------------------------------------------
 
   async *clusterEvents(): AsyncIterableIterator<ClusterEvent> {

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)

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<CreateWorkstreamResponse> {
-    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<string, unknown> = { ...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<StatusResponse> {
@@ -53,12 +101,55 @@ export class TurnstoneServer extends BaseClient {
 
   // -- Chat interaction -----------------------------------------------------
 
-  async send(message: string, wsId: string): Promise<SendResponse> {
-    return this.request("POST", "/v1/api/send", {
-      json: { message, ws_id: wsId },
+  async send(
+    message: string,
+    wsId: string,
+    opts?: { attachmentIds?: string[] },
+  ): Promise<SendResponse> {
+    const body: Record<string, unknown> = { 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<UploadAttachmentResponse> {
+    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<ListAttachmentsResponse> {
+    return this.request("GET", `/v1/api/workstreams/${wsId}/attachments`);
+  }
+
+  async getAttachmentContent(
+    wsId: string,
+    attachmentId: string,
+  ): Promise<AttachmentContent> {
+    return this.requestBytes(
+      "GET",
+      `/v1/api/workstreams/${wsId}/attachments/${attachmentId}/content`,
+    );
+  }
+
+  async deleteAttachment(
+    wsId: string,
+    attachmentId: string,
+  ): Promise<StatusResponse> {
+    return this.request(
+      "DELETE",
+      `/v1/api/workstreams/${wsId}/attachments/${attachmentId}`,
+    );
+  }
+
   async approve(opts: {
     wsId: string;
     approved?: boolean;

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 {

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",

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<string, string> = { "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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).mock.calls[0];
+    expect(init.headers["Content-Type"]).toBe("application/json");
+    expect(JSON.parse(init.body)).toEqual({ name: "j" });
+  });
+});

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()

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 = {}

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")

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"] == []

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):

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(

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=<hex>`` 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),

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",

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

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,

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.

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=<hex>`` (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,
         *,

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,

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:

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 <input accept="..."> 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

turnstone/ui/static/index.html

@@ -28,10 +28,38 @@
 
 <div id="dashboard" class="dashboard-overlay" role="dialog" aria-modal="true" aria-label="Dashboard">
   <div class="dashboard-content">
-    <div class="dashboard-input-row">
-      <input type="text" id="dashboard-input" class="dashboard-input"
-             placeholder="What are you working on?" aria-label="Start a new conversation">
-      <button class="dashboard-new-btn" onclick="dashboardNewChat()" aria-label="New empty chat">New Chat</button>
+    <div class="dashboard-composer" id="dashboard-composer">
+      <textarea id="dashboard-input" class="dashboard-input" rows="3"
+                placeholder="What are you working on?"
+                aria-label="Start a new conversation"></textarea>
+      <div class="dashboard-composer-row">
+        <div class="dashboard-composer-actions">
+          <button id="dashboard-attach-btn" class="dashboard-icon-btn" type="button"
+                  title="Attach files" aria-label="Attach files">
+            <span aria-hidden="true">&#128206;</span>
+          </button>
+          <input id="dashboard-attach-input" type="file" multiple style="display:none"
+                 accept="image/png,image/jpeg,image/gif,image/webp,text/*,.md,.txt,.json,.yaml,.yml,.toml,.xml,.html,.css,.js,.jsx,.ts,.tsx,.py,.go,.rs,.java,.c,.cpp,.h,.hpp,.sh,.sql,.ini,.conf">
+          <button id="dashboard-options-btn" class="dashboard-icon-btn dashboard-options-btn" type="button"
+                  title="Options" aria-label="Toggle options" aria-expanded="false"
+                  aria-controls="dashboard-options">
+            Options <span aria-hidden="true" class="dashboard-options-caret">&#9662;</span>
+          </button>
+        </div>
+        <button id="dashboard-submit-btn" class="dashboard-new-btn" type="button"
+                onclick="dashboardSubmit()" aria-label="Create workstream">Create</button>
+      </div>
+      <div id="dashboard-error" class="dashboard-error" role="status" aria-live="polite"></div>
+      <div id="dashboard-attach-chips" role="list" aria-label="Pending attachments" aria-live="polite"></div>
+      <div id="dashboard-options" class="dashboard-options"
+           role="region" aria-labelledby="dashboard-options-btn" hidden>
+        <label for="dashboard-model">Model</label>
+        <select id="dashboard-model"><option value="">Default model</option></select>
+        <label for="dashboard-judge-model">Judge Model</label>
+        <select id="dashboard-judge-model"><option value="">Default (agent model)</option></select>
+        <label for="dashboard-skill">Skill</label>
+        <select id="dashboard-skill"><option value="">Use defaults</option></select>
+      </div>
     </div>
     <div class="dash-header">
       <span class="dash-header-title">WORKSTREAMS</span>
@@ -82,6 +110,15 @@
     <select id="new-ws-judge-model"><option value="">Default (agent model)</option></select>
     <label for="new-ws-skill">Skill <span class="nws-hint">optional</span></label>
     <select id="new-ws-skill"><option value="">Use defaults</option></select>
+    <label for="new-ws-initial-message">First message <span class="nws-hint">optional</span></label>
+    <textarea id="new-ws-initial-message" rows="3" placeholder="Sent as the first turn after the workstream is created"></textarea>
+    <div id="new-ws-attach-row">
+      <button id="new-ws-attach-btn" type="button" title="Attach files to the first turn" aria-label="Attach files">
+        <span aria-hidden="true">&#128206;</span> Attach
+      </button>
+      <input id="new-ws-attach-input" type="file" multiple style="display:none" accept="image/png,image/jpeg,image/gif,image/webp,text/*,.md,.txt,.json,.yaml,.yml,.toml,.xml,.html,.css,.js,.jsx,.ts,.tsx,.py,.go,.rs,.java,.c,.cpp,.h,.hpp,.sh,.sql,.ini,.conf">
+      <div id="new-ws-attach-chips" role="list" aria-label="Pending attachments"></div>
+    </div>
     <div id="new-ws-buttons">
       <button id="new-ws-cancel" type="button">Cancel</button>
       <button id="new-ws-submit" type="button">Create</button>

turnstone/ui/static/style.css

@@ -1025,6 +1025,78 @@ body { position: static; }
 .pane-attach-chip-remove:hover { color: var(--red, #c94040); background: var(--bg-surface); }
 .pane-attach-chip-remove:focus-visible { outline: 2px solid var(--accent); outline-offset: 1px; }
 
+/* New-workstream modal: attachment row + chips */
+#new-ws-attach-row {
+  display: flex;
+  flex-direction: column;
+  gap: 6px;
+  margin: 6px 0 4px;
+}
+#new-ws-attach-btn {
+  align-self: flex-start;
+  padding: 4px 10px;
+  font-family: var(--font-mono);
+  font-size: 12px;
+  background: var(--bg);
+  color: var(--fg);
+  border: 1px solid var(--border-strong);
+  border-radius: 4px;
+  cursor: pointer;
+}
+#new-ws-attach-btn:hover { background: var(--bg-surface); }
+#new-ws-attach-btn:focus-visible { outline: 2px solid var(--accent); outline-offset: 1px; }
+#new-ws-attach-chips {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 6px;
+}
+#new-ws-attach-chips:empty { display: none; }
+.new-ws-attach-chip {
+  display: inline-flex;
+  align-items: center;
+  gap: 6px;
+  background: var(--bg);
+  color: var(--fg);
+  border: 1px solid var(--border-strong);
+  border-radius: 999px;
+  padding: 3px 8px 3px 10px;
+  font-family: var(--font-mono);
+  font-size: 11px;
+  max-width: 280px;
+}
+.new-ws-attach-chip-name {
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+  max-width: 180px;
+}
+.new-ws-attach-chip-size { color: var(--fg-dim, var(--fg)); opacity: 0.65; font-size: 10px; }
+.new-ws-attach-chip-remove {
+  background: transparent;
+  color: var(--fg-dim, var(--fg));
+  border: none;
+  padding: 0 4px;
+  font-size: 14px;
+  line-height: 1;
+  cursor: pointer;
+  border-radius: 50%;
+}
+.new-ws-attach-chip-remove:hover { color: var(--red, #c94040); background: var(--bg-surface); }
+.new-ws-attach-chip-remove:focus-visible { outline: 2px solid var(--accent); outline-offset: 1px; }
+#new-ws-initial-message {
+  width: 100%;
+  font-family: var(--font-mono);
+  font-size: 13px;
+  resize: vertical;
+  min-height: 60px;
+  background: var(--bg);
+  color: var(--fg);
+  border: 1px solid var(--border-strong);
+  border-radius: 4px;
+  padding: 6px 8px;
+  box-sizing: border-box;
+}
+
 /* Drag-and-drop visual state on the pane */
 .pane.pane-drop-target {
   outline: 2px dashed var(--accent);
@@ -1538,21 +1610,132 @@ audio.media-player {
 .dashboard-overlay { display: none; position: fixed; inset: 0; background: var(--bg); z-index: 50; overflow-y: auto; }
 .dashboard-overlay.active { display: flex; justify-content: center; align-items: flex-start; }
 .dashboard-content { width: 100%; max-width: 960px; padding: 32px 20px 24px; }
-.dashboard-input-row { display: flex; gap: 8px; margin-bottom: 24px; }
-.dashboard-input {
-  flex: 1;
+.dashboard-composer {
+  display: flex;
+  flex-direction: column;
+  gap: 8px;
+  margin-bottom: 24px;
   background: var(--bg-surface);
-  color: var(--fg);
   border: 1px solid var(--border-strong);
   border-radius: var(--radius);
-  padding: 12px 16px;
+  padding: 10px;
+  transition: border-color 0.15s, box-shadow 0.15s;
+}
+.dashboard-composer:focus-within {
+  border-color: var(--accent);
+  box-shadow: 0 0 0 3px var(--accent-dim);
+}
+.dashboard-composer-drop {
+  border-style: dashed;
+  border-color: var(--accent);
+  box-shadow: 0 0 0 3px var(--accent-glow-strong);
+  background: var(--accent-glow);
+}
+.dashboard-input {
+  background: transparent;
+  color: var(--fg);
+  border: none;
+  /* Hairline below the textarea so it reads as an input even before focus */
+  border-bottom: 1px solid var(--border);
+  border-radius: 0;
+  padding: 6px 8px;
   font: inherit;
   font-size: 14px;
   outline: none;
-  transition: border-color 0.15s, box-shadow 0.15s;
+  resize: vertical;
+  min-height: 60px;
+  font-family: var(--font-mono);
 }
-.dashboard-input:focus { border-color: var(--accent); box-shadow: 0 0 0 3px var(--accent-dim); }
 .dashboard-input::placeholder { color: var(--fg-dim); }
+.dashboard-input.dashboard-input-error {
+  outline: 2px solid var(--red);
+  outline-offset: -1px;
+}
+.dashboard-error {
+  color: var(--red);
+  font-family: var(--font-mono);
+  font-size: 11px;
+  min-height: 14px;
+  padding: 0 4px;
+}
+.dashboard-error:empty { display: none; }
+.dashboard-options-caret {
+  font-size: 9px;
+  opacity: 0.7;
+  transition: transform 0.15s;
+}
+.dashboard-options-btn[aria-expanded="true"] .dashboard-options-caret {
+  transform: rotate(180deg);
+  display: inline-block;
+}
+.dashboard-composer-row {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 8px;
+  padding: 0 4px;
+}
+.dashboard-composer-actions {
+  display: flex;
+  align-items: center;
+  gap: 6px;
+}
+.dashboard-icon-btn {
+  background: transparent;
+  color: var(--fg-dim, var(--fg));
+  border: 1px solid var(--border-strong);
+  border-radius: var(--radius-sm);
+  padding: 6px 10px;
+  font-family: var(--font-mono);
+  font-size: 12px;
+  cursor: pointer;
+  display: inline-flex;
+  align-items: center;
+  gap: 4px;
+  /* Larger comfortable hit target without growing the visible chrome */
+  min-height: 28px;
+}
+.dashboard-icon-btn:hover { color: var(--fg); background: var(--bg); }
+.dashboard-icon-btn:focus-visible { outline: 2px solid var(--accent); outline-offset: 1px; }
+.dashboard-options-btn[aria-expanded="true"] {
+  color: var(--fg);
+  background: var(--bg);
+}
+#dashboard-attach-chips {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 6px;
+  padding: 0 4px;
+}
+#dashboard-attach-chips:empty { display: none; }
+.dashboard-options {
+  display: grid;
+  grid-template-columns: auto 1fr;
+  gap: 6px 12px;
+  align-items: center;
+  padding: 8px;
+  background: var(--bg);
+  border: 1px solid var(--border);
+  border-radius: var(--radius-sm);
+  font-family: var(--font-mono);
+  font-size: 12px;
+}
+.dashboard-options[hidden] { display: none; }
+.dashboard-options label {
+  color: var(--fg-dim, var(--fg));
+  text-transform: uppercase;
+  font-size: 10px;
+  letter-spacing: 0.05em;
+}
+.dashboard-options select {
+  background: var(--bg-surface);
+  color: var(--fg);
+  border: 1px solid var(--border-strong);
+  border-radius: var(--radius-sm);
+  padding: 4px 6px;
+  font-family: var(--font-mono);
+  font-size: 12px;
+}
 .dashboard-new-btn {
   background: var(--accent);
   color: var(--bg);
@@ -1779,8 +1962,15 @@ audio.media-player {
   .dash-row-sub { padding-left: 76px; }
 }
 @media (max-width: 600px) {
-  .dashboard-input-row { flex-direction: column; }
-  .dashboard-new-btn { width: 100%; }
+  /* Drop the row constraint and let composer children flow vertically so
+     Send can naturally come after chips + options instead of breaking the
+     reading order with a giant button mid-card. */
+  .dashboard-composer-row { display: contents; }
+  .dashboard-composer-actions { order: 2; justify-content: flex-start; }
+  .dashboard-composer #dashboard-error { order: 3; }
+  .dashboard-composer #dashboard-attach-chips { order: 4; }
+  .dashboard-composer .dashboard-options { order: 5; }
+  .dashboard-composer .dashboard-new-btn { order: 6; width: 100%; }
 }
 @media (max-width: 480px) {
   .dashboard-content { padding: 24px 12px 16px; }