From f5b906c845cca86a0d73ab5fc00e1277f830377d Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 13:03:46 -0600 Subject: [PATCH 1/9] Add httpStreamText and useHttpStream Adds first-class HTTP streaming helpers so apps can return AI SDK streams from a Convex httpAction with auth, thread auto-creation, optional delta persistence, and CORS in a single call. The companion React hook consumes the stream and surfaces streamId/messageId for deduplication with useUIMessages. streamText now eagerly creates the delta streamId when saveStreamDeltas is set, so the HTTP X-Stream-Id header is populated even when the response returns before the stream is fully consumed. Co-Authored-By: Claude Opus 4.7 --- docs/http-streaming-requirements.md | 302 ++++++++++++++++++++++++++++ src/client/http.test.ts | 262 ++++++++++++++++++++++++ src/client/http.ts | 175 ++++++++++++++++ src/client/index.ts | 5 + src/client/streamText.ts | 8 + src/client/types.ts | 8 + src/react/httpStreamUtils.test.ts | 108 ++++++++++ src/react/httpStreamUtils.ts | 61 ++++++ src/react/index.ts | 8 +- src/react/useHttpStream.ts | 145 +++++++++++++ 10 files changed, 1081 insertions(+), 1 deletion(-) create mode 100644 docs/http-streaming-requirements.md create mode 100644 src/client/http.test.ts create mode 100644 src/client/http.ts create mode 100644 src/react/httpStreamUtils.test.ts create mode 100644 src/react/httpStreamUtils.ts create mode 100644 src/react/useHttpStream.ts diff --git a/docs/http-streaming-requirements.md b/docs/http-streaming-requirements.md new file mode 100644 index 00000000..5050d41e --- /dev/null +++ b/docs/http-streaming-requirements.md @@ -0,0 +1,302 @@ +# Technical Requirements: HTTP Streaming for @convex-dev/agent + +## Status: Draft +## Date: 2026-02-23 + +--- + +## 1. Executive Summary + +The current streaming architecture relies exclusively on Convex's reactive query system (WebSocket-based delta polling). This document specifies requirements for adding HTTP streaming support, including delta filtering logic, stream ID lifecycle management, and backwards compatibility constraints. + +--- + +## 2. Current Architecture + +### 2.1 Streaming Transport (WebSocket Delta Polling) + +The existing system persists stream data as discrete deltas in the database, which clients poll via Convex reactive queries. There is no HTTP streaming transport. + +**Flow:** +1. `DeltaStreamer` (client action) writes compressed parts via `streams.addDelta` mutations +2. React hooks (`useDeltaStreams`) issue two reactive queries per render cycle: + - `kind: "list"` — discovers active `streamingMessages` for the thread + - `kind: "deltas"` — fetches new deltas using per-stream cursors +3. `deriveUIMessagesFromDeltas()` materializes `UIMessage[]` from accumulated deltas + +**Key files:** +- `src/client/streaming.ts` — `DeltaStreamer` class, compression, `syncStreams()` +- `src/component/streams.ts` — Backend mutations/queries (`create`, `addDelta`, `listDeltas`, `finish`, `abort`) +- `src/react/useDeltaStreams.ts` — Client-side cursor tracking and delta accumulation +- `src/deltas.ts` — Delta-to-UIMessage materialization + +### 2.2 Stream State Machine + +``` + create() addDelta() (with heartbeat) + │ │ + ▼ ▼ +┌──────────┐ ┌──────────┐ +│ streaming │─────▶│ streaming │──── heartbeat every ~2.5 min +└──────────┘ └──────────┘ + │ │ + │ finish() │ abort() / timeout (10 min) + ▼ ▼ +┌──────────┐ ┌─────────┐ +│ finished │ │ aborted │ +└──────────┘ └─────────┘ + │ + │ cleanup (5 min delay) + ▼ + [deleted] +``` + +### 2.3 Data Formats + +Two delta formats are supported, declared per-stream: + +| Format | Description | Primary Use | +|--------|-------------|-------------| +| `UIMessageChunk` | AI SDK v6 native format (`text-delta`, `tool-input-delta`, `reasoning-delta`, etc.) | Default for new streams | +| `TextStreamPart` | Legacy AI SDK format | Backwards compatibility | + +--- + +## 3. HTTP Streaming Requirements + +### 3.1 Transport Layer + +**REQ-HTTP-1**: Provide an HTTP streaming endpoint that emits deltas as Server-Sent Events (SSE) or newline-delimited JSON (NDJSON), enabling clients that cannot use Convex WebSocket subscriptions (e.g., non-JS environments, CLI tools, third-party integrations). + +**REQ-HTTP-2**: The HTTP endpoint must support resumption. A client that disconnects and reconnects with a cursor value must receive only deltas it hasn't seen, not replay the full stream. + +**REQ-HTTP-3**: The HTTP endpoint must respect the same rate-limiting constants as the WebSocket path: +- `MAX_DELTAS_PER_REQUEST = 1000` (total across all streams) +- `MAX_DELTAS_PER_STREAM = 100` (per stream per request) + +**REQ-HTTP-4**: The HTTP endpoint must support filtering by stream status (`streaming`, `finished`, `aborted`) matching the existing `listStreams` query interface. + +**REQ-HTTP-5**: The HTTP endpoint must emit a terminal event when the stream reaches `finished` or `aborted` state, so clients know to stop polling/listening. + +### 3.2 Response Format + +**REQ-HTTP-6**: Each SSE/NDJSON frame must include: +```typescript +{ + streamId: string; // ID of the streaming message + start: number; // Inclusive cursor position + end: number; // Exclusive cursor position + parts: any[]; // Delta parts (UIMessageChunk[] or TextStreamPart[]) +} +``` + +This matches the existing `StreamDelta` type (`src/validators.ts:628-634`). + +**REQ-HTTP-7**: Stream metadata must be available either as an initial frame or via a separate endpoint, containing: +```typescript +{ + streamId: string; + status: "streaming" | "finished" | "aborted"; + format: "UIMessageChunk" | "TextStreamPart" | undefined; + order: number; + stepOrder: number; + userId?: string; + agentName?: string; + model?: string; + provider?: string; + providerOptions?: ProviderOptions; +} +``` + +This matches the existing `StreamMessage` type (`src/validators.ts:607-626`). + +--- + +## 4. Delta Stream Filtering Logic + +### 4.1 Server-Side Filtering + +**REQ-FILT-1**: The `listDeltas` query must continue to filter by stream ID + cursor position using the `streamId_start_end` index: +``` +.withIndex("streamId_start_end", (q) => + q.eq("streamId", cursor.streamId).gte("start", cursor.cursor)) +``` + +**REQ-FILT-2**: Stream discovery (`list` query) must filter by: +- `threadId` (required) — scoped to a single thread +- `state.kind` (optional, defaults to `["streaming"]`) — which statuses to include +- `startOrder` (optional, defaults to 0) — minimum message order position + +This uses the compound index `threadId_state_order_stepOrder`. + +**REQ-FILT-3**: For HTTP streaming, add support for filtering deltas by a single `streamId` (not requiring `threadId`), for clients that already know which stream they want to follow. + +### 4.2 Client-Side Filtering + +**REQ-FILT-4**: The `useDeltaStreams` hook's cursor management must be preserved: +- Per-stream cursor tracking via `Record` +- Gap detection: assert `previousEnd === delta.start` for consecutive deltas +- Stale delta rejection: skip deltas where `delta.start < oldCursor` +- Cache-friendly `startOrder` rounding (round down to nearest 10) + +**REQ-FILT-5**: Support `skipStreamIds` filtering to allow callers to exclude specific streams (used when streams are already materialized from stored messages). + +### 4.3 Delta Compression + +**REQ-FILT-6**: Delta compression must happen before persistence (in `DeltaStreamer.#createDelta`). Two compression strategies: + +1. **UIMessageChunk compression** (`compressUIMessageChunks`): + - Merge consecutive `text-delta` parts with same `id` by concatenating `.delta` + - Merge consecutive `reasoning-delta` parts with same `id` by concatenating `.delta` + +2. **TextStreamPart compression** (`compressTextStreamParts`): + - Merge consecutive `text-delta` parts with same `id` by concatenating `.text` + - Merge consecutive `reasoning-delta` parts with same `id` by concatenating `.text` + - Strip `Uint8Array` data from `file` parts (not suitable for delta transport) + +**REQ-FILT-7**: Throttling must remain configurable per-stream: +- Default: `250ms` between delta writes +- Configurable via `StreamingOptions.throttleMs` +- Chunking granularity: `"word"`, `"line"`, `RegExp`, or custom `ChunkDetector` (default: `/[\p{P}\s]/u` — punctuation + whitespace) + +--- + +## 5. Stream ID Tracking + +### 5.1 Stream ID Lifecycle + +**REQ-SID-1**: Stream IDs are Convex document IDs (`Id<"streamingMessages">`) generated lazily on first delta write: +- `DeltaStreamer.getStreamId()` creates the stream via `streams.create` mutation +- Race-condition safe: only one creation promise via `#creatingStreamIdPromise` +- Stream ID is `undefined` until the first `addParts()` call + +**REQ-SID-2**: The `streams.create` mutation must: +1. Insert a `streamingMessages` document with `state: { kind: "streaming", lastHeartbeat: Date.now() }` +2. Schedule a timeout function at `TIMEOUT_INTERVAL` (10 minutes) +3. Patch the document with the `timeoutFnId` + +**REQ-SID-3**: Stream IDs must be passed to `addMessages` via `finishStreamId` for atomic stream finish + message persistence (prevents UI flicker from separate mutations). + +### 5.2 Client-Side Stream ID Management + +**REQ-SID-4**: React hooks must track multiple concurrent streams per thread: +- `useDeltaStreams` returns `Array<{ streamMessage: StreamMessage; deltas: StreamDelta[] }>` +- Each stream accumulates deltas independently +- Streams are sorted by `[order, stepOrder]` for display + +**REQ-SID-5**: When a thread changes (`threadId` differs from previous render): +- Clear all accumulated delta streams (`state.deltaStreams = undefined`) +- Reset all cursors (`setCursors({})`) +- Reset `startOrder` + +**REQ-SID-6**: Stream identity in UIMessages uses the convention `id: "stream:{streamId}"` to distinguish streaming messages from persisted messages. + +### 5.3 Heartbeat & Timeout + +**REQ-SID-7**: Heartbeat behavior: +- Triggered on every `addDelta` call +- Debounced: only writes if >2.5 minutes since last heartbeat (`TIMEOUT_INTERVAL / 4`) +- Updates `state.lastHeartbeat` and reschedules the timeout function + +**REQ-SID-8**: Timeout behavior: +- After 10 minutes of inactivity, `timeoutStream` internal mutation fires +- Checks if `lastHeartbeat + TIMEOUT_INTERVAL < Date.now()` +- If expired: aborts the stream with reason `"timeout"` +- If not expired: reschedules for the remaining time + +**REQ-SID-9**: Cleanup behavior: +- `finish()` schedules `deleteStream` after `DELETE_STREAM_DELAY` (5 minutes) +- `deleteStream` removes the `streamingMessages` document and all associated `streamDeltas` +- 5-minute delay allows clients to fetch final deltas before cleanup + +--- + +## 6. Backwards Compatibility Requirements + +### 6.1 Transport Compatibility + +**REQ-BC-1**: The existing WebSocket/reactive-query streaming path must remain the default and primary transport. HTTP streaming is additive, not a replacement. + +**REQ-BC-2**: All existing public APIs must remain unchanged: +- `syncStreams()` function signature and return type (`SyncStreamsReturnValue`) +- `listStreams()` function signature +- `abortStream()` function signature +- `vStreamMessagesReturnValue` validator + +**REQ-BC-3**: The `StreamArgs` union type must be extended (not replaced) to support HTTP streaming parameters: +```typescript +// Existing (preserved): +type StreamArgs = + | { kind: "list"; startOrder: number } + | { kind: "deltas"; cursors: Array<{ streamId: string; cursor: number }> } +// New (additive): + | { kind: "http"; streamId: string; cursor?: number } +``` + +### 6.2 Data Format Compatibility + +**REQ-BC-4**: Both `UIMessageChunk` and `TextStreamPart` delta formats must be supported in perpetuity. The `format` field on `streamingMessages` is `v.optional(...)`, so streams created before format tracking was added (format = `undefined`) must default to `TextStreamPart` behavior. + +**REQ-BC-5**: Forward compatibility for new `TextStreamPart` types from future AI SDK versions must be maintained via the `default` case in `updateFromTextStreamParts` (`src/deltas.ts:520-527`): +```typescript +default: { + console.warn(`Received unexpected part: ${JSON.stringify(part)}`); + break; +} +``` + +**REQ-BC-6**: The `readUIMessageStream` error suppression for `"no tool invocation found"` must be preserved (`src/deltas.ts:77-81`). This handles tool approval continuation streams that have `tool-result` without the original `tool-call`. + +### 6.3 React Hook Compatibility + +**REQ-BC-7**: Existing React hooks must not change behavior: +- `useThreadMessages` — paginated messages + streaming +- `useUIMessages` — UIMessage-first with metadata +- `useSmoothText` — animated text rendering + +**REQ-BC-8**: New HTTP-streaming React hooks (if any) must be additive exports from `@convex-dev/agent/react`, not replacements. + +### 6.4 Schema Compatibility + +**REQ-BC-9**: No breaking changes to the component schema tables: +- `streamingMessages` — no field removals or type changes +- `streamDeltas` — no field removals or type changes +- Indexes must not be dropped (can add new ones) + +**REQ-BC-10**: The `vStreamDelta` and `vStreamMessage` validators must remain structurally compatible. New optional fields may be added but existing fields must not change type or be removed. + +### 6.5 Export Surface Compatibility + +**REQ-BC-11**: All four export surfaces must remain stable: +- `@convex-dev/agent` — main exports +- `@convex-dev/agent/react` — React hooks +- `@convex-dev/agent/validators` — Convex validators +- `@convex-dev/agent/test` — testing utilities + +HTTP streaming additions should be exported from the main surface or a new `@convex-dev/agent/http` surface (not mixed into existing surfaces that would break tree-shaking). + +--- + +## 7. Non-Functional Requirements + +**REQ-NF-1**: HTTP streaming latency must not exceed the WebSocket path latency by more than 100ms for equivalent payload sizes. + +**REQ-NF-2**: HTTP streaming must support concurrent streams per thread (matching current behavior of up to 100 active streams per thread, per the `list` query's `.take(100)`). + +**REQ-NF-3**: HTTP streaming must gracefully handle client disconnection without leaving orphaned streams (existing heartbeat/timeout mechanism applies). + +**REQ-NF-4**: Delta writes must remain throttled at the configured `throttleMs` regardless of transport, to avoid excessive database writes. + +--- + +## 8. Open Questions + +1. **SSE vs NDJSON**: Should the HTTP transport use SSE (native browser support, automatic reconnection) or NDJSON (simpler, works with `fetch` + `ReadableStream`)? + +2. **Authentication**: How should HTTP streaming endpoints authenticate? Convex actions have auth context, but raw HTTP endpoints may need token-based auth. + +3. **Multi-stream HTTP**: Should a single HTTP connection support multiplexed streams (like the current WebSocket path with multi-cursor queries), or should each HTTP connection follow a single stream? + +4. **Convex HTTP actions**: Should HTTP streaming be implemented as Convex HTTP actions (which have a 2-minute timeout and limited streaming support), or as a separate server/proxy? + +5. **Atomic finish over HTTP**: The current `finishStreamId` pattern enables atomic stream finish + message save. How should this translate to the HTTP transport where the client may not be the writer? diff --git a/src/client/http.test.ts b/src/client/http.test.ts new file mode 100644 index 00000000..adb12bcd --- /dev/null +++ b/src/client/http.test.ts @@ -0,0 +1,262 @@ +import { describe, expect, test } from "vitest"; +import { createThread, httpStreamText, httpStreamUIMessages } from "./index.js"; +import { + anyApi, + actionGeneric, + defineSchema, + type DataModelFromSchemaDefinition, +} from "convex/server"; +import type { ApiFromModules, ActionBuilder } from "convex/server"; +import { components, initConvexTest } from "./setup.test.js"; +import { mockModel } from "./mockModel.js"; +import { streamText } from "./streamText.js"; + +const schema = defineSchema({}); +type DataModel = DataModelFromSchemaDefinition; +const action = actionGeneric as ActionBuilder; + +const model = () => + mockModel({ + content: [{ type: "text", text: "Hello from mock" }], + }); + +// ============================================================================ +// Test action exports — convex-test requires these to live in test files so +// that t.action(api...) can dispatch into them. +// ============================================================================ + +export const testStreamTextStreamId = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const result = await streamText( + ctx, + components.agent, + { model: model(), prompt: "test prompt" }, + { + agentName: "stream-test", + threadId, + saveStreamDeltas: { returnImmediately: true }, + }, + ); + // Drain the stream so the test can observe streamId after generation + // is set up (but the stream itself may not be fully finished). + for await (const _ of result.textStream) { + // consume + } + return { + streamId: result.streamId, + promptMessageId: result.promptMessageId, + }; + }, +}); + +export const testStreamTextNoDeltas = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const result = await streamText( + ctx, + components.agent, + { model: model(), prompt: "test prompt" }, + { + agentName: "stream-test", + threadId, + }, + ); + for await (const _ of result.textStream) { + // consume + } + return { streamId: result.streamId }; + }, +}); + +export const testHttpStreamTextWithThread = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const response = await httpStreamText( + ctx, + components.agent, + { model: model(), prompt: "Hello" }, + { + agentName: "http-test", + threadId, + }, + ); + const text = await response.text(); + return { + status: response.status, + hasText: text.length > 0, + hasMessageId: response.headers.has("X-Message-Id"), + hasStreamId: response.headers.has("X-Stream-Id"), + }; + }, +}); + +export const testHttpStreamTextCreatesThread = action({ + args: {}, + handler: async (ctx) => { + const response = await httpStreamText( + ctx, + components.agent, + { model: model(), prompt: "Hello" }, + { + agentName: "http-test", + userId: "user-abc", + }, + ); + const text = await response.text(); + return { + status: response.status, + hasText: text.length > 0, + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +export const testHttpStreamTextWithCors = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const response = await httpStreamText( + ctx, + components.agent, + { model: model(), prompt: "Hello" }, + { + agentName: "http-test", + threadId, + corsHeaders: { + "Access-Control-Allow-Origin": "*", + "Access-Control-Expose-Headers": "X-Message-Id, X-Stream-Id", + }, + }, + ); + await response.text(); + return { + status: response.status, + corsOrigin: response.headers.get("Access-Control-Allow-Origin"), + corsExpose: response.headers.get("Access-Control-Expose-Headers"), + }; + }, +}); + +export const testHttpStreamTextSavesDeltas = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const response = await httpStreamText( + ctx, + components.agent, + { model: model(), prompt: "Hello" }, + { + agentName: "http-test", + threadId, + saveStreamDeltas: true, + }, + ); + await response.text(); + return { + status: response.status, + hasStreamId: response.headers.has("X-Stream-Id"), + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +export const testHttpStreamUIMessages = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const response = await httpStreamUIMessages( + ctx, + components.agent, + { model: model(), prompt: "Hello" }, + { + agentName: "http-test", + threadId, + }, + ); + const text = await response.text(); + return { + status: response.status, + hasText: text.length > 0, + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +const testApi: ApiFromModules<{ + fns: { + testStreamTextStreamId: typeof testStreamTextStreamId; + testStreamTextNoDeltas: typeof testStreamTextNoDeltas; + testHttpStreamTextWithThread: typeof testHttpStreamTextWithThread; + testHttpStreamTextCreatesThread: typeof testHttpStreamTextCreatesThread; + testHttpStreamTextWithCors: typeof testHttpStreamTextWithCors; + testHttpStreamTextSavesDeltas: typeof testHttpStreamTextSavesDeltas; + testHttpStreamUIMessages: typeof testHttpStreamUIMessages; + }; +}>["fns"] = anyApi["http.test"] as any; + +// ============================================================================ +// Tests +// ============================================================================ + +describe("streamText streamId metadata", () => { + test("returns streamId when saveStreamDeltas is enabled", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testStreamTextStreamId, {}); + expect(result.streamId).toBeDefined(); + expect(result.promptMessageId).toBeDefined(); + }); + + test("streamId is undefined when saveStreamDeltas is not set", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testStreamTextNoDeltas, {}); + expect(result.streamId).toBeUndefined(); + }); +}); + +describe("httpStreamText", () => { + test("streams text and sets X-Message-Id header", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testHttpStreamTextWithThread, {}); + expect(result.status).toBe(200); + expect(result.hasText).toBe(true); + expect(result.hasMessageId).toBe(true); + }); + + test("creates a thread when threadId is omitted", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testHttpStreamTextCreatesThread, {}); + expect(result.status).toBe(200); + expect(result.hasText).toBe(true); + expect(result.hasMessageId).toBe(true); + }); + + test("applies corsHeaders to the response", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testHttpStreamTextWithCors, {}); + expect(result.status).toBe(200); + expect(result.corsOrigin).toBe("*"); + expect(result.corsExpose).toBe("X-Message-Id, X-Stream-Id"); + }); + + test("sets X-Stream-Id when saveStreamDeltas is enabled", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testHttpStreamTextSavesDeltas, {}); + expect(result.status).toBe(200); + expect(result.hasStreamId).toBe(true); + expect(result.hasMessageId).toBe(true); + }); +}); + +describe("httpStreamUIMessages", () => { + test("returns a UI message stream response", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testHttpStreamUIMessages, {}); + expect(result.status).toBe(200); + expect(result.hasText).toBe(true); + expect(result.hasMessageId).toBe(true); + }); +}); diff --git a/src/client/http.ts b/src/client/http.ts new file mode 100644 index 00000000..dbc219a7 --- /dev/null +++ b/src/client/http.ts @@ -0,0 +1,175 @@ +import type { streamText as streamTextAi, ToolSet } from "ai"; +import { streamText } from "./streamText.js"; +import { createThread } from "./threads.js"; +import type { + ActionCtx, + AgentComponent, + AgentPrompt, + Options, + Output, +} from "./types.js"; +import type { StreamingOptions } from "./streaming.js"; + +export type HttpStreamOptions = Options & { + agentName: string; + userId?: string | null; + threadId?: string; + /** + * Whether to save incremental data (deltas) from streaming responses + * to the database alongside the HTTP stream. Defaults to false. + * + * NOTE: For HTTP flows, `true` is normalized to `{ returnImmediately: true }` + * so the response body starts streaming without waiting for completion. + */ + saveStreamDeltas?: boolean | StreamingOptions; + /** Extra headers to add to the response (e.g. CORS headers). */ + corsHeaders?: Record; +}; + +type StreamTextInputArgs< + TOOLS extends ToolSet, + OUTPUT extends Output, +> = AgentPrompt & + Omit< + Parameters>[0], + "model" | "prompt" | "messages" + > & { + tools?: TOOLS; + }; + +/** + * Stream text over HTTP, returning a standard `Response` with a readable + * text stream body. Wraps {@link streamText} and uses + * `toTextStreamResponse()` for the body. + * + * Response headers include: + * - `X-Message-Id` — the prompt message ID + * - `X-Stream-Id` — the delta stream ID (only when `saveStreamDeltas` is set) + * + * @example + * ```ts + * export const chat = httpAction(async (ctx, request) => { + * const { prompt, threadId } = await request.json(); + * return httpStreamText(ctx, components.agent, { prompt }, { + * agentName: "myAgent", + * threadId, + * model: openai.chat("gpt-4o-mini"), + * }); + * }); + * ``` + */ +export async function httpStreamText< + TOOLS extends ToolSet = ToolSet, + OUTPUT extends Output = never, +>( + ctx: ActionCtx, + component: AgentComponent, + streamTextArgs: StreamTextInputArgs, + options: HttpStreamOptions, +): Promise { + const threadId = await resolveThreadId(ctx, component, options); + + const result = await streamText( + ctx, + component, + streamTextArgs, + { + ...options, + threadId, + saveStreamDeltas: normalizeHttpSaveStreamDeltas(options.saveStreamDeltas), + }, + ); + + const response = result.toTextStreamResponse(); + applyHeaders(response, result, options.corsHeaders); + return response; +} + +/** + * Stream UI messages over HTTP, returning a standard `Response` + * using AI SDK's `toUIMessageStreamResponse()` format. This provides + * richer streaming data including tool calls, reasoning, and sources. + * + * @example + * ```ts + * export const chat = httpAction(async (ctx, request) => { + * const { prompt, threadId } = await request.json(); + * return httpStreamUIMessages(ctx, components.agent, { prompt }, { + * agentName: "myAgent", + * threadId, + * model: openai.chat("gpt-4o-mini"), + * }); + * }); + * ``` + */ +export async function httpStreamUIMessages< + TOOLS extends ToolSet = ToolSet, + OUTPUT extends Output = never, +>( + ctx: ActionCtx, + component: AgentComponent, + streamTextArgs: StreamTextInputArgs, + options: HttpStreamOptions, +): Promise { + const threadId = await resolveThreadId(ctx, component, options); + + const result = await streamText( + ctx, + component, + streamTextArgs, + { + ...options, + threadId, + saveStreamDeltas: normalizeHttpSaveStreamDeltas(options.saveStreamDeltas), + }, + ); + + const response = result.toUIMessageStreamResponse(); + applyHeaders(response, result, options.corsHeaders); + return response; +} + +async function resolveThreadId( + ctx: ActionCtx, + component: AgentComponent, + options: HttpStreamOptions, +): Promise { + if (options.threadId) return options.threadId; + return createThread(ctx, component, { + userId: options.userId ?? null, + }); +} + +/** + * If callers pass `saveStreamDeltas: true` for an HTTP flow, force + * `returnImmediately: true` — otherwise `streamText` consumes the stream + * before returning, which buffers the entire response and defeats the + * purpose of streaming over HTTP. + */ +function normalizeHttpSaveStreamDeltas( + saveStreamDeltas?: boolean | StreamingOptions, +): boolean | StreamingOptions | undefined { + if (saveStreamDeltas === true) return { returnImmediately: true }; + if (saveStreamDeltas && typeof saveStreamDeltas === "object") { + return { ...saveStreamDeltas, returnImmediately: true }; + } + return saveStreamDeltas; +} + +function applyHeaders( + response: Response, + result: { promptMessageId?: string; streamId?: string }, + corsHeaders?: Record, +) { + if (result.promptMessageId) { + response.headers.set("X-Message-Id", result.promptMessageId); + } + if (result.streamId) { + response.headers.set("X-Stream-Id", result.streamId); + } + if (corsHeaders) { + for (const [key, value] of Object.entries(corsHeaders)) { + response.headers.set(key, value); + } + } +} diff --git a/src/client/index.ts b/src/client/index.ts index 874b5648..40b4d52d 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -102,6 +102,11 @@ import { streamText } from "./streamText.js"; import { errorToString, willContinue } from "./utils.js"; export { stepCountIs } from "ai"; +export { + httpStreamText, + httpStreamUIMessages, + type HttpStreamOptions, +} from "./http.js"; export { docsToModelMessages, toModelMessage, diff --git a/src/client/streamText.ts b/src/client/streamText.ts index 8f96817c..8dbe23cb 100644 --- a/src/client/streamText.ts +++ b/src/client/streamText.ts @@ -111,6 +111,13 @@ export async function streamText< ) : undefined; + // Eagerly create the streamId so it's available on the returned metadata + // (e.g. so HTTP responses can set an X-Stream-Id header) without having + // to wait for stream consumption to begin. + if (streamer) { + await streamer.getStreamId(); + } + const result = streamTextAi({ ...args, abortSignal: streamer?.abortController.signal ?? args.abortSignal, @@ -190,6 +197,7 @@ export async function streamText< promptMessageId, order, savedMessages: call.getSavedMessages(), + streamId: streamer?.streamId, messageId: promptMessageId, }; return Object.assign(result, metadata); diff --git a/src/client/types.ts b/src/client/types.ts index ea07439a..cdfb5ea1 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -268,6 +268,14 @@ export type GenerationOutputMetadata = { * If you passed promptMessageId, it will not include that message. */ savedMessages?: MessageDoc[]; + /** + * The ID of the delta stream, if `saveStreamDeltas` was enabled. + * Useful for HTTP streaming deduplication: pass this to `skipStreamIds` + * on `useUIMessages` / `useThreadMessages` to avoid showing the same + * content twice (once from the HTTP stream and once from the persisted + * delta stream). + */ + streamId?: string; /** * @deprecated Use promptMessageId instead. * The ID of the prompt message for the generation. diff --git a/src/react/httpStreamUtils.test.ts b/src/react/httpStreamUtils.test.ts new file mode 100644 index 00000000..547052d1 --- /dev/null +++ b/src/react/httpStreamUtils.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, test } from "vitest"; +import { consumeTextStream, supportsStreaming } from "./httpStreamUtils.js"; + +function makeReadableStream( + chunks: string[], +): ReadableStream { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)); + } + controller.close(); + }, + }); +} + +describe("consumeTextStream", () => { + test("decodes chunks and calls onChunk", async () => { + const chunks: string[] = []; + const stream = makeReadableStream(["Hello ", "world", "!"]); + await consumeTextStream(stream.getReader(), { + onChunk: (text) => chunks.push(text), + }); + expect(chunks).toEqual(["Hello ", "world", "!"]); + }); + + test("handles empty stream", async () => { + const chunks: string[] = []; + const stream = makeReadableStream([]); + await consumeTextStream(stream.getReader(), { + onChunk: (text) => chunks.push(text), + }); + expect(chunks).toEqual([]); + }); + + test("handles single large chunk", async () => { + const longText = "A".repeat(10000); + const chunks: string[] = []; + const stream = makeReadableStream([longText]); + await consumeTextStream(stream.getReader(), { + onChunk: (text) => chunks.push(text), + }); + expect(chunks.join("")).toBe(longText); + }); + + test("stops when abort signal is already aborted", async () => { + const chunks: string[] = []; + const controller = new AbortController(); + controller.abort(); + + const stream = makeReadableStream(["first ", "second"]); + + await consumeTextStream(stream.getReader(), { + onChunk: (text) => chunks.push(text), + signal: controller.signal, + }); + + expect(chunks).toEqual([]); + }); + + test("aborting mid-stream cancels a pending read", async () => { + // Stream that never enqueues — reader.read() will hang until cancel. + const stream = new ReadableStream({ + start() { + // intentionally idle + }, + }); + const controller = new AbortController(); + const chunks: string[] = []; + + const consume = consumeTextStream(stream.getReader(), { + onChunk: (text) => chunks.push(text), + signal: controller.signal, + }); + + // Abort while consumeTextStream is blocked on reader.read() + setTimeout(() => controller.abort(), 10); + + await consume; // Should resolve, not hang + expect(chunks).toEqual([]); + }); + + test("handles multi-byte UTF-8 characters split across chunks", async () => { + const emoji = new Uint8Array([0xf0, 0x9f, 0x98, 0x80]); + const chunks: string[] = []; + + const stream = new ReadableStream({ + start(ctrl) { + ctrl.enqueue(emoji.slice(0, 2)); + ctrl.enqueue(emoji.slice(2, 4)); + ctrl.close(); + }, + }); + + await consumeTextStream(stream.getReader(), { + onChunk: (text) => chunks.push(text), + }); + + expect(chunks.join("")).toContain("\u{1F600}"); + }); +}); + +describe("supportsStreaming", () => { + test("returns true in environments with ReadableStream and fetch", () => { + expect(supportsStreaming()).toBe(true); + }); +}); diff --git a/src/react/httpStreamUtils.ts b/src/react/httpStreamUtils.ts new file mode 100644 index 00000000..b3768e01 --- /dev/null +++ b/src/react/httpStreamUtils.ts @@ -0,0 +1,61 @@ +/** + * Consume a ReadableStream of Uint8Array chunks, decoding them as text + * and calling `onChunk` for each decoded segment. + * + * Handles multi-byte characters correctly via + * `decoder.decode(value, { stream: true })`. + * + * If `signal` aborts during a pending `reader.read()`, the read is + * unblocked by cancelling the reader so the caller does not have to + * wait for the next chunk. + */ +export async function consumeTextStream( + reader: ReadableStreamDefaultReader, + options: { + onChunk: (text: string) => void; + signal?: AbortSignal; + }, +): Promise { + const decoder = new TextDecoder(); + const onAbort = () => { + void reader.cancel().catch(() => { + // best-effort + }); + }; + options.signal?.addEventListener("abort", onAbort); + + try { + while (true) { + if (options.signal?.aborted) break; + const { done, value } = await reader.read(); + if (done) break; + const text = decoder.decode(value, { stream: true }); + options.onChunk(text); + } + // Flush any remaining bytes in the decoder + const remaining = decoder.decode(); + if (remaining) { + options.onChunk(remaining); + } + } finally { + options.signal?.removeEventListener("abort", onAbort); + try { + reader.releaseLock(); + } catch { + // reader may already be released after cancel() + } + } +} + +/** + * Returns `true` when running in a browser that supports + * `ReadableStream` on `Response.body`. Returns `false` during SSR + * or in environments where streaming fetch is unavailable. + */ +export function supportsStreaming(): boolean { + return ( + typeof globalThis !== "undefined" && + typeof globalThis.ReadableStream !== "undefined" && + typeof globalThis.fetch !== "undefined" + ); +} diff --git a/src/react/index.ts b/src/react/index.ts index 2ede9a07..c49bdb2f 100644 --- a/src/react/index.ts +++ b/src/react/index.ts @@ -12,9 +12,15 @@ export { } from "./useThreadMessages.js"; export { type UIMessagesQuery, useUIMessages } from "./useUIMessages.js"; export { useStreamingUIMessages } from "./useStreamingUIMessages.js"; +export { useHttpStream } from "./useHttpStream.js"; +export { consumeTextStream, supportsStreaming } from "./httpStreamUtils.js"; /** - * @deprecated use useThreadMessages or useStreamingThreadMessages instead + * @deprecated Use {@link useHttpStream} instead for HTTP streaming with + * deduplication support (`streamId`, `messageId`, `abort()`), or use + * `useUIMessages` / `useThreadMessages` with `stream: true` for + * WebSocket delta streaming. + * * Use this hook to stream text from a server action, using the * toTextStreamResponse or equivalent HTTP streaming endpoint returning text. * @param url The URL of the server action to stream text from. diff --git a/src/react/useHttpStream.ts b/src/react/useHttpStream.ts new file mode 100644 index 00000000..b8d6b123 --- /dev/null +++ b/src/react/useHttpStream.ts @@ -0,0 +1,145 @@ +"use client"; +import { useCallback, useRef, useState } from "react"; +import { consumeTextStream } from "./httpStreamUtils.js"; + +/** + * React hook for consuming an HTTP text stream from a Convex HTTP action. + * + * Returns `streamId` and `messageId` from response headers so you can + * pass them to `useUIMessages` via `skipStreamIds` for deduplication. + * + * @example + * ```tsx + * const httpStream = useHttpStream({ url: `${siteUrl}/chat` }); + * const messages = useUIMessages(api.chat.listMessages, { threadId }, { + * stream: true, + * skipStreamIds: httpStream.streamId ? [httpStream.streamId] : [], + * }); + * + * await httpStream.send({ threadId, prompt: "Hello!" }); + * ``` + */ +export function useHttpStream(options: { + /** The full URL of the HTTP streaming endpoint. */ + url: string; + /** + * Auth token to send as `Authorization: Bearer `. + * e.g. from `useAuthToken()` via `@convex-dev/auth/react`. + */ + token?: string; + /** Additional headers to include in the request. */ + headers?: Record; +}): { + text: string; + isStreaming: boolean; + error: Error | null; + streamId: string | null; + messageId: string | null; + send: (body: { + threadId?: string; + prompt?: string; + [key: string]: unknown; + }) => Promise; + abort: () => void; +} { + const [text, setText] = useState(""); + const [isStreaming, setIsStreaming] = useState(false); + const [error, setError] = useState(null); + const [streamId, setStreamId] = useState(null); + const [messageId, setMessageId] = useState(null); + const abortControllerRef = useRef(null); + // Each call to send() bumps this; only the latest request is allowed + // to flip the streaming/error/text state in its finally block. + const requestIdRef = useRef(0); + + const abort = useCallback(() => { + abortControllerRef.current?.abort(); + abortControllerRef.current = null; + }, []); + + const send = useCallback( + async (body: { + threadId?: string; + prompt?: string; + [key: string]: unknown; + }) => { + // Abort any existing stream + abort(); + + const controller = new AbortController(); + abortControllerRef.current = controller; + const requestId = ++requestIdRef.current; + + setText(""); + setError(null); + setStreamId(null); + setMessageId(null); + setIsStreaming(true); + + try { + const response = await fetch(options.url, { + method: "POST", + headers: { + "Content-Type": "application/json", + ...(options.token + ? { Authorization: `Bearer ${options.token}` } + : {}), + ...options.headers, + }, + body: JSON.stringify(body), + signal: controller.signal, + }); + + if (!response.ok) { + throw new Error(`HTTP error! status: ${response.status}`); + } + + const responseStreamId = response.headers.get("X-Stream-Id"); + const responseMessageId = response.headers.get("X-Message-Id"); + if (requestId === requestIdRef.current) { + if (responseStreamId) setStreamId(responseStreamId); + if (responseMessageId) setMessageId(responseMessageId); + } + + if (!response.body) { + throw new Error("Response body is not readable"); + } + + const reader = response.body.getReader(); + let accumulated = ""; + + await consumeTextStream(reader, { + onChunk: (chunk) => { + // Stale chunks from a superseded request must not bleed into + // the current view. + if (requestId !== requestIdRef.current) return; + accumulated += chunk; + setText(accumulated); + }, + signal: controller.signal, + }); + } catch (e) { + if (e instanceof Error && e.name === "AbortError") { + return; + } + if (requestId === requestIdRef.current) { + const err = e instanceof Error ? e : new Error(String(e)); + setError(err); + } + } finally { + // Only the latest send() should flip streaming state. Otherwise a + // stale request that finishes after a newer one has started would + // mark the live stream as finished. + if (requestId === requestIdRef.current) { + setIsStreaming(false); + } + if (abortControllerRef.current === controller) { + abortControllerRef.current = null; + } + } + }, + [options.url, options.token, options.headers, abort], + ); + + return { text, isStreaming, error, streamId, messageId, send, abort }; +} From 363563faf932d651681c9c56f054a9de9b860e19 Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 13:17:34 -0600 Subject: [PATCH 2/9] Add Agent.asHttpAction and full streaming demo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit asHttpAction wraps httpStreamText for the common pattern of returning the agent's stream from a Convex httpAction. It parses the JSON body, auto-creates a thread when missing, applies CORS headers, and supports an authorize callback that can override userId/threadId. Adds a comprehensive streaming demo (route: /streaming-demo) showing the three streaming patterns side-by-side — async delta streaming, HTTP streaming, one-shot streaming — plus tool approval flow with a stream inspector panel. Also fixes a pre-existing typecheck failure in example/convex/setup.test.ts: workflow and rate-limiter packages still type `register` against the pre-generic SchemaDefinition, so cast through the broader type so example/convex typechecks regardless of which version is resolved. Co-Authored-By: Claude Opus 4.7 --- example/convex/_generated/api.d.ts | 4 + example/convex/agents/streamingDemo.ts | 58 ++ example/convex/chat/streamingDemo.ts | 215 +++++++ example/convex/http.ts | 14 +- example/convex/setup.test.ts | 14 +- example/ui/chat/StreamingDemo.tsx | 761 +++++++++++++++++++++++++ example/ui/main.tsx | 18 +- src/client/http.test.ts | 195 ++++++- src/client/http.ts | 2 +- src/client/index.ts | 133 +++++ 10 files changed, 1406 insertions(+), 8 deletions(-) create mode 100644 example/convex/agents/streamingDemo.ts create mode 100644 example/convex/chat/streamingDemo.ts create mode 100644 example/ui/chat/StreamingDemo.tsx diff --git a/example/convex/_generated/api.d.ts b/example/convex/_generated/api.d.ts index b7cf3c58..9f706627 100644 --- a/example/convex/_generated/api.d.ts +++ b/example/convex/_generated/api.d.ts @@ -13,12 +13,14 @@ import type * as agents_config from "../agents/config.js"; import type * as agents_fashion from "../agents/fashion.js"; import type * as agents_simple from "../agents/simple.js"; import type * as agents_story from "../agents/story.js"; +import type * as agents_streamingDemo from "../agents/streamingDemo.js"; import type * as agents_weather from "../agents/weather.js"; import type * as chat_approval from "../chat/approval.js"; import type * as chat_basic from "../chat/basic.js"; import type * as chat_human from "../chat/human.js"; import type * as chat_streamAbort from "../chat/streamAbort.js"; import type * as chat_streaming from "../chat/streaming.js"; +import type * as chat_streamingDemo from "../chat/streamingDemo.js"; import type * as chat_streamingReasoning from "../chat/streamingReasoning.js"; import type * as chat_withoutAgent from "../chat/withoutAgent.js"; import type * as crons from "../crons.js"; @@ -62,12 +64,14 @@ declare const fullApi: ApiFromModules<{ "agents/fashion": typeof agents_fashion; "agents/simple": typeof agents_simple; "agents/story": typeof agents_story; + "agents/streamingDemo": typeof agents_streamingDemo; "agents/weather": typeof agents_weather; "chat/approval": typeof chat_approval; "chat/basic": typeof chat_basic; "chat/human": typeof chat_human; "chat/streamAbort": typeof chat_streamAbort; "chat/streaming": typeof chat_streaming; + "chat/streamingDemo": typeof chat_streamingDemo; "chat/streamingReasoning": typeof chat_streamingReasoning; "chat/withoutAgent": typeof chat_withoutAgent; crons: typeof crons; diff --git a/example/convex/agents/streamingDemo.ts b/example/convex/agents/streamingDemo.ts new file mode 100644 index 00000000..441ee48e --- /dev/null +++ b/example/convex/agents/streamingDemo.ts @@ -0,0 +1,58 @@ +// Agent for the streaming demo. Reuses the same tools as the approval demo +// so the UI can show streaming patterns alongside human-in-the-loop approval. +import { Agent, createTool, stepCountIs } from "@convex-dev/agent"; +import { components } from "../_generated/api"; +import { defaultConfig } from "./config"; +import { z } from "zod/v4"; + +const deleteFileTool = createTool({ + description: "Delete a file from the system", + inputSchema: z.object({ + filename: z.string().describe("The name of the file to delete"), + }), + needsApproval: () => true, + execute: async (_ctx, input) => { + return `Successfully deleted file: ${input.filename}`; + }, +}); + +const transferMoneyTool = createTool({ + description: "Transfer money to an account", + inputSchema: z.object({ + amount: z.number().describe("The amount to transfer"), + toAccount: z.string().describe("The destination account"), + }), + needsApproval: async (_ctx, input) => { + return input.amount > 100; + }, + execute: async (_ctx, input) => { + return `Transferred $${input.amount} to account ${input.toAccount}`; + }, +}); + +const checkBalanceTool = createTool({ + description: "Check the account balance", + inputSchema: z.object({ + accountId: z.string().describe("The account to check"), + }), + execute: async (_ctx, _input) => { + return `Balance: $1,234.56`; + }, +}); + +export const streamingDemoAgent = new Agent(components.agent, { + name: "Streaming Demo Agent", + instructions: + "You are a concise assistant who responds with emojis " + + "and abbreviations like lmao, lol, iirc, afaik, etc. where appropriate. " + + "You can delete files, transfer money, and check account balances. " + + "Always confirm what action you took after it completes.", + tools: { + deleteFile: deleteFileTool, + transferMoney: transferMoneyTool, + checkBalance: checkBalanceTool, + }, + stopWhen: stepCountIs(5), + ...defaultConfig, + callSettings: { ...defaultConfig.callSettings, temperature: 0 }, +}); diff --git a/example/convex/chat/streamingDemo.ts b/example/convex/chat/streamingDemo.ts new file mode 100644 index 00000000..611ae044 --- /dev/null +++ b/example/convex/chat/streamingDemo.ts @@ -0,0 +1,215 @@ +/** + * Full Streaming Demo + * + * Demonstrates ALL streaming patterns in one place: + * 1. Async delta streaming (recommended) - mutation saves prompt, action streams + * 2. HTTP streaming - direct text stream over HTTP response + * 3. One-shot streaming - single action call with delta persistence + * 4. Stream lifecycle management - abort, status transitions, cleanup + * 5. Tool approval - pauses generation, resumes after approve/deny + */ +import { paginationOptsValidator } from "convex/server"; +import { + listUIMessages, + syncStreams, + abortStream, + listStreams, + vStreamArgs, +} from "@convex-dev/agent"; +import { components, internal } from "../_generated/api"; +import { + action, + httpAction, + internalAction, + mutation, + query, +} from "../_generated/server"; +import { v } from "convex/values"; +import { authorizeThreadAccess } from "../threads"; +import { streamingDemoAgent } from "../agents/streamingDemo"; + +// ============================================================================ +// Pattern 1: Async Delta Streaming (RECOMMENDED) +// +// Two-phase approach: +// Phase 1 (mutation): Save the user message and schedule the action. +// Phase 2 (action): Stream the AI response, saving deltas to the DB. +// +// Clients subscribe via `useUIMessages` with `stream: true` and see real-time +// delta updates through Convex's reactive query system. +// ============================================================================ + +export const sendMessage = mutation({ + args: { prompt: v.string(), threadId: v.string() }, + handler: async (ctx, { prompt, threadId }) => { + await authorizeThreadAccess(ctx, threadId); + const { messageId } = await streamingDemoAgent.saveMessage(ctx, { + threadId, + prompt, + skipEmbeddings: true, + }); + await ctx.scheduler.runAfter( + 0, + internal.chat.streamingDemo.streamResponse, + { threadId, promptMessageId: messageId }, + ); + }, +}); + +export const streamResponse = internalAction({ + args: { promptMessageId: v.string(), threadId: v.string() }, + handler: async (ctx, { promptMessageId, threadId }) => { + const result = await streamingDemoAgent.streamText( + ctx, + { threadId }, + { promptMessageId }, + { saveStreamDeltas: { chunking: "word", throttleMs: 100 } }, + ); + await result.consumeStream(); + }, +}); + +// ============================================================================ +// Pattern 2: HTTP Streaming +// +// Streams text directly over an HTTP response using `agent.asHttpAction()`. +// The handler parses the JSON body, creates a thread if needed, streams +// the response, and sets X-Message-Id / X-Stream-Id headers. +// ============================================================================ + +export const streamOverHttp = httpAction(streamingDemoAgent.asHttpAction()); + +// ============================================================================ +// Pattern 3: One-Shot Streaming +// +// Single action call that both streams and persists deltas. Simpler than +// the two-phase approach but does not support optimistic client updates. +// ============================================================================ + +export const streamOneShot = action({ + args: { prompt: v.string(), threadId: v.string() }, + handler: async (ctx, { prompt, threadId }) => { + await authorizeThreadAccess(ctx, threadId); + await streamingDemoAgent.streamText( + ctx, + { threadId }, + { prompt }, + { saveStreamDeltas: true }, + ); + }, +}); + +// ============================================================================ +// Tool Approval +// +// When the model calls a tool with `needsApproval`, generation pauses. +// The client shows Approve/Deny buttons; once resolved, the client triggers +// continuation via delta streaming. +// ============================================================================ + +export const submitApproval = mutation({ + args: { + threadId: v.string(), + approvalId: v.string(), + approved: v.boolean(), + reason: v.optional(v.string()), + }, + returns: v.object({ messageId: v.string() }), + handler: async (ctx, { threadId, approvalId, approved, reason }) => { + await authorizeThreadAccess(ctx, threadId); + const { messageId } = approved + ? await streamingDemoAgent.approveToolCall(ctx, { + threadId, + approvalId, + reason, + }) + : await streamingDemoAgent.denyToolCall(ctx, { + threadId, + approvalId, + reason, + }); + return { messageId }; + }, +}); + +export const triggerContinuation = mutation({ + args: { threadId: v.string(), lastApprovalMessageId: v.string() }, + handler: async (ctx, { threadId, lastApprovalMessageId }) => { + await authorizeThreadAccess(ctx, threadId); + await ctx.scheduler.runAfter( + 0, + internal.chat.streamingDemo.continueAfterApprovals, + { threadId, lastApprovalMessageId }, + ); + }, +}); + +export const continueAfterApprovals = internalAction({ + args: { threadId: v.string(), lastApprovalMessageId: v.string() }, + handler: async (ctx, { threadId, lastApprovalMessageId }) => { + const result = await streamingDemoAgent.streamText( + ctx, + { threadId }, + { promptMessageId: lastApprovalMessageId }, + { saveStreamDeltas: { chunking: "word", throttleMs: 100 } }, + ); + await result.consumeStream(); + }, +}); + +// ============================================================================ +// Queries: Messages + Stream Sync +// ============================================================================ + +export const listThreadMessages = query({ + args: { + threadId: v.string(), + paginationOpts: paginationOptsValidator, + streamArgs: vStreamArgs, + }, + handler: async (ctx, args) => { + const { threadId, streamArgs } = args; + await authorizeThreadAccess(ctx, threadId); + const streams = await syncStreams(ctx, components.agent, { + threadId, + streamArgs, + }); + const paginated = await listUIMessages(ctx, components.agent, args); + return { ...paginated, streams }; + }, +}); + +export const listActiveStreams = query({ + args: { threadId: v.string() }, + handler: async (ctx, { threadId }) => { + await authorizeThreadAccess(ctx, threadId); + return listStreams(ctx, components.agent, { threadId }); + }, +}); + +// ============================================================================ +// Stream Lifecycle Management +// ============================================================================ + +export const abortStreamByOrder = mutation({ + args: { threadId: v.string(), order: v.number() }, + handler: async (ctx, { threadId, order }) => { + await authorizeThreadAccess(ctx, threadId); + return abortStream(ctx, components.agent, { + threadId, + order, + reason: "User requested abort", + }); + }, +}); + +export const listAllStreams = query({ + args: { threadId: v.string() }, + handler: async (ctx, { threadId }) => { + await authorizeThreadAccess(ctx, threadId); + return listStreams(ctx, components.agent, { + threadId, + includeStatuses: ["streaming", "finished", "aborted"], + }); + }, +}); diff --git a/example/convex/http.ts b/example/convex/http.ts index 39e109e5..6bfaa60f 100644 --- a/example/convex/http.ts +++ b/example/convex/http.ts @@ -1,5 +1,6 @@ import { httpRouter } from "convex/server"; import { streamOverHttp } from "./chat/streaming"; +import { streamOverHttp as streamOverHttpDemo } from "./chat/streamingDemo"; import { corsRouter } from "convex-helpers/server/cors"; const http = httpRouter(); @@ -7,7 +8,12 @@ const http = httpRouter(); const cors = corsRouter(http, { allowCredentials: true, allowedHeaders: ["Authorization", "Content-Type"], - exposedHeaders: ["Content-Type", "Content-Length", "X-Message-Id"], + exposedHeaders: [ + "Content-Type", + "Content-Length", + "X-Message-Id", + "X-Stream-Id", + ], }); cors.route({ @@ -16,5 +22,11 @@ cors.route({ handler: streamOverHttp, }); +cors.route({ + path: "/streamTextDemo", + method: "POST", + handler: streamOverHttpDemo, +}); + // Convex expects the router to be the default export of `convex/http.js`. export default http; diff --git a/example/convex/setup.test.ts b/example/convex/setup.test.ts index b6c0a4f7..1b4aacdf 100644 --- a/example/convex/setup.test.ts +++ b/example/convex/setup.test.ts @@ -1,6 +1,7 @@ /// import { test } from "vitest"; -import { convexTest } from "convex-test"; +import { convexTest, type TestConvex } from "convex-test"; +import type { GenericSchema, SchemaDefinition } from "convex/server"; import schema from "./schema.js"; import agent from "@convex-dev/agent/test"; import workflow from "@convex-dev/workflow/test"; @@ -10,8 +11,15 @@ export const modules = import.meta.glob("./**/*.*s"); export function initConvexTest() { const t = convexTest(schema, modules); agent.register(t); - workflow.register(t); - rateLimiter.register(t); + // workflow and rate-limiter still type their `register` against the + // pre-generic SchemaDefinition. Cast through + // the broader type so this file typechecks regardless of which version + // of those packages is resolved. + const generic = t as unknown as TestConvex< + SchemaDefinition + >; + workflow.register(generic); + rateLimiter.register(generic); return t; } diff --git a/example/ui/chat/StreamingDemo.tsx b/example/ui/chat/StreamingDemo.tsx new file mode 100644 index 00000000..94f93b43 --- /dev/null +++ b/example/ui/chat/StreamingDemo.tsx @@ -0,0 +1,761 @@ +/** + * Full Streaming Demo UI + * + * Demonstrates ALL streaming patterns with a comprehensive UI: + * + * - Async delta streaming with real-time message updates + * - HTTP streaming via fetch with text decoding + * - Stream lifecycle visualization (streaming / finished / aborted) + * - Abort in-progress streams + * - Stream inspector panel showing active/finished/aborted streams + * - Smooth text animation via useSmoothText + * - Optimistic message sending + * - Tool approval flow (approve/deny buttons, auto-continuation) + */ +import { useAction, useMutation, useQuery } from "convex/react"; +import { api } from "../../convex/_generated/api"; +import { + optimisticallySendMessage, + useHttpStream, + useSmoothText, + useUIMessages, + type UIMessage, +} from "@convex-dev/agent/react"; +import { useCallback, useEffect, useRef, useState } from "react"; +import { cn } from "@/lib/utils"; +import { useDemoThread } from "@/hooks/use-demo-thread"; +import type { ToolUIPart } from "ai"; + +type StreamMode = "delta" | "http" | "oneshot"; + +export default function StreamingDemo() { + const { threadId, resetThread } = useDemoThread("Streaming Demo"); + + return ( + <> +
+

+ Full Streaming Demo +

+
+
+ {threadId ? ( + void resetThread()} /> + ) : ( +
+ Loading... +
+ )} +
+ + ); +} + +function DemoApp({ + threadId, + reset, +}: { + threadId: string; + reset: () => void; +}) { + const [streamMode, setStreamMode] = useState("delta"); + + return ( +
+ {/* Left: Chat area */} +
+ + +
+ + {/* Right: Stream Inspector */} +
+ +
+
+ ); +} + +// ============================================================================ +// Mode Selector +// ============================================================================ + +function ModeSelector({ + mode, + onChange, +}: { + mode: StreamMode; + onChange: (m: StreamMode) => void; +}) { + const modes: { value: StreamMode; label: string; desc: string }[] = [ + { + value: "delta", + label: "Delta Streaming", + desc: "Async mutation + action with delta persistence (recommended)", + }, + { + value: "http", + label: "HTTP Streaming", + desc: "Direct text stream over HTTP response", + }, + { + value: "oneshot", + label: "One-Shot", + desc: "Single action call with delta persistence", + }, + ]; + + return ( +
+ {modes.map((m) => ( + + ))} +
+ ); +} + +// ============================================================================ +// Chat Panel +// ============================================================================ + +function ChatPanel({ + threadId, + mode, + reset, +}: { + threadId: string; + mode: StreamMode; + reset: () => void; +}) { + const convexUrl = import.meta.env.VITE_CONVEX_URL as string; + if (!convexUrl.endsWith(".cloud")) { + console.warn("Unexpected Convex URL format; HTTP streaming may not work:", convexUrl); + } + const httpUrl = convexUrl.replace(/\.cloud$/, ".site"); + + const httpStream = useHttpStream({ url: `${httpUrl}/streamTextDemo` }); + + const { + results: messages, + status, + loadMore, + } = useUIMessages( + api.chat.streamingDemo.listThreadMessages, + { threadId }, + { + initialNumItems: 20, + stream: true, + skipStreamIds: httpStream.streamId ? [httpStream.streamId] : [], + }, + ); + + const sendDelta = useMutation( + api.chat.streamingDemo.sendMessage, + ).withOptimisticUpdate( + optimisticallySendMessage(api.chat.streamingDemo.listThreadMessages), + ); + const sendOneShot = useAction(api.chat.streamingDemo.streamOneShot); + const abortByOrder = useMutation( + api.chat.streamingDemo.abortStreamByOrder, + ); + + // Tool approval mutations + const submitApproval = useMutation(api.chat.streamingDemo.submitApproval); + const triggerContinuation = useMutation(api.chat.streamingDemo.triggerContinuation); + + // Track the last approval messageId so we can use it for continuation. + const lastApprovalMessageIdRef = useRef(null); + // Track whether we've already triggered continuation for this batch. + const continuationTriggeredRef = useRef(false); + // Track the mode used when the request was sent, so continuation uses the same mode. + const requestModeRef = useRef(mode); + + const hasPendingApprovals = messages.some((m) => + m.parts.some( + (p) => p.type.startsWith("tool-") && (p as ToolUIPart).state === "approval-requested", + ), + ); + + // When all approvals are resolved (hasPendingApprovals goes false) + // and we have a saved messageId, trigger continuation. + // In HTTP mode, continuation also goes over HTTP. Otherwise, delta streaming. + useEffect(() => { + if ( + !hasPendingApprovals && + lastApprovalMessageIdRef.current && + !continuationTriggeredRef.current + ) { + continuationTriggeredRef.current = true; + const messageId = lastApprovalMessageIdRef.current; + lastApprovalMessageIdRef.current = null; + if (requestModeRef.current === "http") { + void httpStream.send({ threadId, promptMessageId: messageId }); + } else { + void triggerContinuation({ + threadId, + lastApprovalMessageId: messageId, + }); + } + } + if (hasPendingApprovals) { + continuationTriggeredRef.current = false; + } + }, [hasPendingApprovals, threadId, triggerContinuation, httpStream]); + + async function handleApproval(args: { + threadId: string; + approvalId: string; + approved: boolean; + reason?: string; + }) { + const { messageId } = await submitApproval(args); + lastApprovalMessageIdRef.current = messageId; + } + + const [prompt, setPrompt] = useState("Delete the file important.txt"); + const messagesEndRef = useRef(null); + + const httpText = httpStream.text; + const httpStreaming = httpStream.isStreaming; + + const scrollToBottom = useCallback(() => { + messagesEndRef.current?.scrollIntoView({ behavior: "smooth" }); + }, []); + + useEffect(() => { + scrollToBottom(); + }, [messages, httpText, scrollToBottom]); + + const isStreaming = messages.some((m) => m.status === "streaming"); + + async function handleSend() { + const text = prompt.trim(); + if (!text) return; + setPrompt(""); + requestModeRef.current = mode; + + if (mode === "delta") { + await sendDelta({ threadId, prompt: text }); + } else if (mode === "oneshot") { + // Don't await — the action runs server-side while deltas stream + // to the client via reactive queries. + sendOneShot({ threadId, prompt: text }).catch((e) => + console.error("oneshot error:", e), + ); + } else if (mode === "http") { + await httpStream.send({ threadId, prompt: text }); + } + } + + return ( + <> +
+ {messages.length > 0 || httpText ? ( +
+ {status === "CanLoadMore" && ( + + )} + {messages + .filter( + (m) => + // While HTTP streaming, hide the pending assistant message — + // its content is shown in the HTTP stream bubble instead. + !(httpStreaming && httpText && m.role === "assistant" && m.status === "pending"), + ) + .map((m) => ( + + ))} + {httpStreaming && httpText && (() => { + // Grab tool parts from the pending assistant message + const pending = messages.find( + (m) => m.role === "assistant" && m.status === "pending", + ); + const toolParts = pending?.parts.filter((p) => + p.type.startsWith("tool-"), + ) ?? []; + return ( +
+
+ + [HTTP stream{httpStreaming ? " - live" : " - done"}] + + {toolParts.map((p: any) => ( +
+ {p.type} + {p.state && ( + ({p.state}) + )} + {p.output && ( +
+ {typeof p.output === "string" + ? p.output + : JSON.stringify(p.output)} +
+ )} +
+ ))} +
{httpText}
+
+
+ ); + })()} +
+
+ ) : ( +
+ Pick a streaming mode above and start chatting. +
+ )} +
+ +
+
{ + e.preventDefault(); + void handleSend(); + }} + > + setPrompt(e.target.value)} + className="flex-1 px-4 py-2 rounded-lg border border-gray-300 focus:outline-none focus:ring-2 focus:ring-indigo-400 bg-gray-50 disabled:opacity-50 disabled:cursor-not-allowed" + placeholder={hasPendingApprovals ? "Respond to pending approvals first..." : "Type a message..."} + disabled={hasPendingApprovals} + /> + {isStreaming || httpStreaming ? ( + + ) : ( + + )} + +
+
+ Mode:{" "} + + {mode === "delta" + ? "Async Delta Streaming" + : mode === "http" + ? "HTTP Streaming" + : "One-Shot Streaming"} + +
+
+ + ); +} + +// ============================================================================ +// Message Bubble +// ============================================================================ + +function MessageBubble({ + message, + threadId, + onApproval, +}: { + message: UIMessage; + threadId: string; + onApproval: (args: { + threadId: string; + approvalId: string; + approved: boolean; + reason?: string; + }) => Promise; +}) { + const isUser = message.role === "user"; + const [visibleText] = useSmoothText(message.text, { + startStreaming: message.status === "streaming", + }); + const [reasoningText] = useSmoothText( + message.parts + .filter((p) => p.type === "reasoning") + .map((p) => p.text) + .join("\n") ?? "", + { startStreaming: message.status === "streaming" }, + ); + + const toolParts = message.parts.filter( + (p): p is ToolUIPart => p.type.startsWith("tool-"), + ); + + return ( +
+
+ {/* Status badge */} + {message.status !== "success" && message.role !== "user" && ( + + [{message.status}] + + )} + + {/* Reasoning */} + {reasoningText && ( +
+ {reasoningText} +
+ )} + + {/* Tool calls with approval UI */} + {toolParts.map((tool) => ( + + ))} + + {/* Main text */} +
{visibleText || (isUser ? message.text : "...")}
+
+
+ ); +} + +// ============================================================================ +// Tool Call Display with Approval UI +// ============================================================================ + +function ToolCallDisplay({ + tool, + threadId, + onApproval, +}: { + tool: ToolUIPart; + threadId: string; + onApproval: (args: { + threadId: string; + approvalId: string; + approved: boolean; + reason?: string; + }) => Promise; +}) { + const [denialReason, setDenialReason] = useState(""); + const [showReasonInput, setShowReasonInput] = useState(false); + const toolName = tool.type.replace("tool-", ""); + const approvalId = getToolApprovalId(tool); + const approvalReason = getToolApprovalReason(tool); + + return ( +
+
+ {toolName}({JSON.stringify(tool.input)}) +
+ + {tool.state === "approval-requested" && approvalId && ( +
+
+ Approval required +
+ {showReasonInput ? ( +
+ setDenialReason(e.target.value)} + placeholder="Reason for denial..." + className="flex-1 px-2 py-1 text-sm rounded border border-gray-300" + /> + + +
+ ) : ( +
+ + +
+ )} +
+ )} + + {tool.state === "approval-responded" && ( +
Approved - executing...
+ )} + + {tool.state === "output-denied" && ( +
+ Denied + {approvalReason && `: ${approvalReason}`} +
+ )} + + {tool.state === "output-available" && ( +
+ Result: {JSON.stringify("output" in tool ? tool.output : undefined)} +
+ )} + + {tool.state === "output-error" && ( +
+ Error: {"errorText" in tool ? tool.errorText : "Unknown error"} +
+ )} + + {(tool.state === "input-available" || tool.state === "input-streaming") && ( +
Processing...
+ )} +
+ ); +} + +function getToolApprovalId(tool: ToolUIPart): string | undefined { + if (tool.state !== "approval-requested" || !("approval" in tool)) { + return undefined; + } + const approval = tool.approval as { id?: unknown } | undefined; + return typeof approval?.id === "string" ? approval.id : undefined; +} + +function getToolApprovalReason(tool: ToolUIPart): string | undefined { + if ( + (tool.state !== "output-denied" && tool.state !== "approval-requested") || + !("approval" in tool) + ) { + return undefined; + } + const approval = tool.approval as { reason?: unknown } | undefined; + return typeof approval?.reason === "string" ? approval.reason : undefined; +} + +// ============================================================================ +// Stream Inspector Panel +// ============================================================================ + +function StreamInspector({ threadId }: { threadId: string }) { + const allStreams = useQuery(api.chat.streamingDemo.listAllStreams, { + threadId, + }); + + const streaming = allStreams?.filter((s) => s.status === "streaming") ?? []; + const finished = allStreams?.filter((s) => s.status === "finished") ?? []; + const aborted = allStreams?.filter((s) => s.status === "aborted") ?? []; + + return ( +
+

+ Stream Inspector +

+ + + + + + {allStreams?.length === 0 && ( +

+ No streams yet. Send a message to see stream lifecycle. +

+ )} + +
+

How it works:

+
    +
  • + Delta Streaming: Mutation saves prompt, schedules + an action. The action streams AI response and saves deltas to the + database. Clients subscribe via reactive queries. +
  • +
  • + HTTP Streaming: Direct text stream over HTTP. + Response chunks are decoded by the browser. No database + persistence of intermediate deltas. +
  • +
  • + One-Shot: Single action call. Simpler but no + optimistic updates. +
  • +
  • + Abort: Transitions the stream to "aborted" state. + Clients see the partial response with failed status. +
  • +
  • + Fallback: When streaming finishes, the full + message is saved to the database. The deduplication logic prefers + finalized messages over streaming ones. +
  • +
+
+
+ ); +} + +function StreamSection({ + label, + streams, + color, +}: { + label: string; + streams: any[]; + color: "green" | "blue" | "red"; +}) { + if (streams.length === 0) return null; + + const dotColors = { + green: "bg-green-400", + blue: "bg-blue-400", + red: "bg-red-400", + }; + const bgColors = { + green: "bg-green-50 border-green-200", + blue: "bg-blue-50 border-blue-200", + red: "bg-red-50 border-red-200", + }; + + return ( +
+
+
+ + {label} ({streams.length}) + +
+
+ {streams.map((s: any) => ( +
+
+ id: {s.streamId.slice(0, 12)}... +
+
+ order: {s.order}, step: {s.stepOrder} +
+ {s.agentName &&
agent: {s.agentName}
} + {s.model &&
model: {s.model}
} +
+ ))} +
+
+ ); +} diff --git a/example/ui/main.tsx b/example/ui/main.tsx index 636d9429..595ff3b7 100644 --- a/example/ui/main.tsx +++ b/example/ui/main.tsx @@ -12,7 +12,7 @@ import RagBasic from "./rag/RagBasic"; import { StrictMode } from "react"; import StreamArray from "./objects/StreamArray"; import ChatApproval from "./chat/ChatApproval"; - +import StreamingDemo from "./chat/StreamingDemo"; const convex = new ConvexReactClient(import.meta.env.VITE_CONVEX_URL as string); @@ -49,7 +49,7 @@ export function App() { } /> } /> } /> - + } /> @@ -154,6 +154,20 @@ function Index() { with an optional reason.

+
  • + + Full Streaming Demo + +

    + Comprehensive demo of all streaming patterns: async delta + streaming, HTTP streaming, one-shot streaming, stream abort, + lifecycle visualization, and tool approval. Includes a stream + inspector panel. +

    +
  • More examples coming soon! diff --git a/src/client/http.test.ts b/src/client/http.test.ts index adb12bcd..31dd8cd4 100644 --- a/src/client/http.test.ts +++ b/src/client/http.test.ts @@ -1,5 +1,10 @@ import { describe, expect, test } from "vitest"; -import { createThread, httpStreamText, httpStreamUIMessages } from "./index.js"; +import { + Agent, + createThread, + httpStreamText, + httpStreamUIMessages, +} from "./index.js"; import { anyApi, actionGeneric, @@ -20,6 +25,12 @@ const model = () => content: [{ type: "text", text: "Hello from mock" }], }); +const agent = new Agent(components.agent, { + name: "http-test-agent", + instructions: "You are a test agent for HTTP streaming", + languageModel: model(), +}); + // ============================================================================ // Test action exports — convex-test requires these to live in test files so // that t.action(api...) can dispatch into them. @@ -186,6 +197,130 @@ export const testHttpStreamUIMessages = action({ }, }); +export const testAsHttpActionParsesBody = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const handler = agent.asHttpAction(); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ threadId, prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + const text = await response.text(); + return { + status: response.status, + hasText: text.length > 0, + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +export const testAsHttpActionCreatesThread = action({ + args: {}, + handler: async (ctx) => { + const handler = agent.asHttpAction(); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + const text = await response.text(); + return { + status: response.status, + hasText: text.length > 0, + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +export const testAsHttpActionWithCorsHeaders = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const handler = agent.asHttpAction({ + corsHeaders: { + "Access-Control-Allow-Origin": "*", + "Access-Control-Expose-Headers": "X-Message-Id, X-Stream-Id", + }, + }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ threadId, prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + await response.text(); + return { + status: response.status, + corsOrigin: response.headers.get("Access-Control-Allow-Origin"), + corsExpose: response.headers.get("Access-Control-Expose-Headers"), + }; + }, +}); + +export const testAsHttpActionWithSaveDeltas = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const handler = agent.asHttpAction({ saveStreamDeltas: true }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ threadId, prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + await response.text(); + return { + status: response.status, + hasStreamId: response.headers.has("X-Stream-Id"), + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +export const testAsHttpActionUIMessages = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const handler = agent.asHttpAction({ format: "ui-messages" }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ threadId, prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + const text = await response.text(); + return { + status: response.status, + hasText: text.length > 0, + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + +export const testAsHttpActionAuthorizeOverridesUserId = action({ + args: {}, + handler: async (ctx) => { + const handler = agent.asHttpAction({ + authorize: async () => ({ userId: "user-from-auth" }), + }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + await response.text(); + return { + status: response.status, + hasMessageId: response.headers.has("X-Message-Id"), + }; + }, +}); + const testApi: ApiFromModules<{ fns: { testStreamTextStreamId: typeof testStreamTextStreamId; @@ -195,6 +330,12 @@ const testApi: ApiFromModules<{ testHttpStreamTextWithCors: typeof testHttpStreamTextWithCors; testHttpStreamTextSavesDeltas: typeof testHttpStreamTextSavesDeltas; testHttpStreamUIMessages: typeof testHttpStreamUIMessages; + testAsHttpActionParsesBody: typeof testAsHttpActionParsesBody; + testAsHttpActionCreatesThread: typeof testAsHttpActionCreatesThread; + testAsHttpActionWithCorsHeaders: typeof testAsHttpActionWithCorsHeaders; + testAsHttpActionWithSaveDeltas: typeof testAsHttpActionWithSaveDeltas; + testAsHttpActionUIMessages: typeof testAsHttpActionUIMessages; + testAsHttpActionAuthorizeOverridesUserId: typeof testAsHttpActionAuthorizeOverridesUserId; }; }>["fns"] = anyApi["http.test"] as any; @@ -260,3 +401,55 @@ describe("httpStreamUIMessages", () => { expect(result.hasMessageId).toBe(true); }); }); + +describe("agent.asHttpAction()", () => { + test("parses JSON body and streams text", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testAsHttpActionParsesBody, {}); + expect(result.status).toBe(200); + expect(result.hasText).toBe(true); + expect(result.hasMessageId).toBe(true); + }); + + test("creates a thread when threadId is omitted", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testAsHttpActionCreatesThread, {}); + expect(result.status).toBe(200); + expect(result.hasText).toBe(true); + expect(result.hasMessageId).toBe(true); + }); + + test("applies corsHeaders to the response", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testAsHttpActionWithCorsHeaders, {}); + expect(result.status).toBe(200); + expect(result.corsOrigin).toBe("*"); + expect(result.corsExpose).toBe("X-Message-Id, X-Stream-Id"); + }); + + test("sets X-Stream-Id when saveStreamDeltas is enabled", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testAsHttpActionWithSaveDeltas, {}); + expect(result.status).toBe(200); + expect(result.hasStreamId).toBe(true); + expect(result.hasMessageId).toBe(true); + }); + + test("returns a UI message stream when format=ui-messages", async () => { + const t = initConvexTest(schema); + const result = await t.action(testApi.testAsHttpActionUIMessages, {}); + expect(result.status).toBe(200); + expect(result.hasText).toBe(true); + expect(result.hasMessageId).toBe(true); + }); + + test("authorize callback can supply userId for thread creation", async () => { + const t = initConvexTest(schema); + const result = await t.action( + testApi.testAsHttpActionAuthorizeOverridesUserId, + {}, + ); + expect(result.status).toBe(200); + expect(result.hasMessageId).toBe(true); + }); +}); diff --git a/src/client/http.ts b/src/client/http.ts index dbc219a7..5306e2e5 100644 --- a/src/client/http.ts +++ b/src/client/http.ts @@ -146,7 +146,7 @@ async function resolveThreadId( * before returning, which buffers the entire response and defeats the * purpose of streaming over HTTP. */ -function normalizeHttpSaveStreamDeltas( +export function normalizeHttpSaveStreamDeltas( saveStreamDeltas?: boolean | StreamingOptions, ): boolean | StreamingOptions | undefined { if (saveStreamDeltas === true) return { returnImmediately: true }; diff --git a/src/client/index.ts b/src/client/index.ts index 40b4d52d..25bc6561 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -99,6 +99,7 @@ import type { Output, } from "./types.js"; import { streamText } from "./streamText.js"; +import { normalizeHttpSaveStreamDeltas } from "./http.js"; import { errorToString, willContinue } from "./utils.js"; export { stepCountIs } from "ai"; @@ -1655,6 +1656,138 @@ export class Agent< }); } + /** + * Build a plain handler `(ctx, request) => Promise` for an + * HTTP streaming endpoint. Wrap the return value in your app's + * `httpAction()`. + * + * The handler parses the JSON body for + * `{ threadId?, prompt?, promptMessageId?, messages? }`, creates a + * thread if one isn't supplied, streams the response, and returns it + * with `X-Message-Id` and (if `saveStreamDeltas` is on) `X-Stream-Id` + * headers. + * + * @example + * ```ts + * // convex/http.ts + * import { httpAction } from "./_generated/server"; + * http.route({ + * path: "/chat", + * method: "POST", + * handler: httpAction(myAgent.asHttpAction()), + * }); + * ``` + */ + asHttpAction( + spec?: MaybeCustomCtx & { + /** + * Whether to save incremental data (deltas) from streaming responses + * to the database alongside the HTTP stream. Defaults to false. + * + * For HTTP flows, `true` is normalized so the response body starts + * streaming without buffering the entire generation first. + */ + saveStreamDeltas?: boolean | StreamingOptions; + /** + * When to stop generating text. + * Defaults to the {@link Agent["options"].stopWhen} option. + */ + stopWhen?: StopCondition | Array>; + /** + * Response format: + * - `"text"` (default) — plain text via `toTextStreamResponse()`. + * - `"ui-messages"` — rich AI SDK UI message stream via + * `toUIMessageStreamResponse()` (tool calls, reasoning, sources). + */ + format?: "text" | "ui-messages"; + /** Extra headers to add to the response (e.g. CORS headers). */ + corsHeaders?: Record; + /** + * Optional authorization callback. Receives the raw request and may + * return `{ userId?, threadId? }` to override values from the body. + * Throw to reject the request. + */ + authorize?: ( + ctx: GenericActionCtx, + request: Request, + ) => Promise<{ userId?: string; threadId?: string } | void>; + } & Options, + ): ( + ctx: GenericActionCtx, + request: Request, + ) => Promise { + return async (ctx_, request) => { + const body = (await request.json()) as { + threadId?: string; + prompt?: string; + promptMessageId?: string; + messages?: unknown[]; + }; + + let userId: string | undefined; + let threadId: string | undefined = body.threadId; + + if (spec?.authorize) { + const authResult = await spec.authorize(ctx_, request); + if (authResult?.userId) userId = authResult.userId; + if (authResult?.threadId) threadId = authResult.threadId; + } + + const targetArgs = { userId, threadId }; + const llmArgs = { + prompt: body.prompt, + promptMessageId: body.promptMessageId, + messages: body.messages?.map((m) => toModelMessage(m as never)), + stopWhen: spec?.stopWhen, + }; + const ctx = ( + spec?.customCtx + ? { + ...ctx_, + ...spec.customCtx(ctx_, targetArgs, llmArgs as never), + } + : ctx_ + ) as unknown as ActionCtx & CustomCtx; + + if (!threadId) { + threadId = await createThread(ctx, this.component, { + userId: userId ?? null, + }); + } + + const result = await this.streamText( + ctx, + { threadId, userId }, + llmArgs, + { + contextOptions: spec?.contextOptions, + storageOptions: spec?.storageOptions, + saveStreamDeltas: normalizeHttpSaveStreamDeltas( + spec?.saveStreamDeltas, + ), + }, + ); + + const response = + spec?.format === "ui-messages" + ? result.toUIMessageStreamResponse() + : result.toTextStreamResponse(); + + if (result.promptMessageId) { + response.headers.set("X-Message-Id", result.promptMessageId); + } + if (result.streamId) { + response.headers.set("X-Stream-Id", result.streamId); + } + if (spec?.corsHeaders) { + for (const [key, value] of Object.entries(spec.corsHeaders)) { + response.headers.set(key, value); + } + } + return response; + }; + } + /** * @deprecated Use {@link saveMessages} directly instead. */ From 0317e7136c53784679217c0e4491b7cde2aa22f0 Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 15:12:53 -0600 Subject: [PATCH 3/9] Address Codex review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - streamText: when saveStreamDeltas runs in returnImmediately mode (the HTTP streaming path), the deferred final-step save would never run — streamText returns before consumption finishes. Now save the final step inline using the streamId for atomic finish, and mark the streamer finished externally so its background consumeStream doesn't call finish() after the action context has gone away. Fixes the case where HTTP streams left messages unsaved and stream records stuck in "streaming" state. - asHttpAction: forward all per-request Options from spec (usageHandler, contextHandler, rawRequestResponseHandler) instead of silently dropping them. - asHttpAction: wrap request.json() in try/catch and return 400 for malformed JSON rather than leaking an opaque error. Co-Authored-By: Claude Opus 4.7 --- src/client/index.ts | 19 ++++++++++++++++--- src/client/streamText.ts | 31 +++++++++++++++++++++++-------- 2 files changed, 39 insertions(+), 11 deletions(-) diff --git a/src/client/index.ts b/src/client/index.ts index 25bc6561..a71cdf78 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -1717,12 +1717,23 @@ export class Agent< request: Request, ) => Promise { return async (ctx_, request) => { - const body = (await request.json()) as { + let body: { threadId?: string; prompt?: string; promptMessageId?: string; messages?: unknown[]; }; + try { + body = (await request.json()) as typeof body; + } catch { + return new Response( + JSON.stringify({ error: "Invalid JSON in request body" }), + { + status: 400, + headers: { "Content-Type": "application/json" }, + }, + ); + } let userId: string | undefined; let threadId: string | undefined = body.threadId; @@ -1760,8 +1771,10 @@ export class Agent< { threadId, userId }, llmArgs, { - contextOptions: spec?.contextOptions, - storageOptions: spec?.storageOptions, + // Forward all per-request Options from spec (contextOptions, + // storageOptions, usageHandler, contextHandler, + // rawRequestResponseHandler) so they actually take effect. + ...spec, saveStreamDeltas: normalizeHttpSaveStreamDeltas( spec?.saveStreamDeltas, ), diff --git a/src/client/streamText.ts b/src/client/streamText.ts index 8dbe23cb..73a06e58 100644 --- a/src/client/streamText.ts +++ b/src/client/streamText.ts @@ -83,6 +83,14 @@ export async function streamText< // Track the final step for atomic save with stream finish (issue #181) let pendingFinalStep: StepResult | undefined; + // Whether streamText will await stream consumption before returning. + // When false (saveStreamDeltas with returnImmediately, or no deltas), + // we cannot defer the final-step save — the deferred work would never run. + const willAwaitStream = + options.saveStreamDeltas === true || + (typeof options.saveStreamDeltas === "object" && + !options.saveStreamDeltas.returnImmediately); + const streamer = threadId && options.saveStreamDeltas ? new DeltaStreamer( @@ -149,10 +157,21 @@ export async function streamText< steps.push(step); const createPendingMessage = await willContinue(steps, args.stopWhen); if (!createPendingMessage && streamer) { - // This is the final step with streaming enabled. - // Defer saving until stream consumption completes for atomic finish (issue #181). + // Final step with streaming enabled: save the message and finish + // the stream atomically (issue #181). When `willAwaitStream` is + // true we defer the save to after consumption (so any post-step + // delta writes land on the saved row). When false (e.g. + // `returnImmediately` for HTTP), we save inline using the streamId + // and mark the streamer finished externally — otherwise the + // background `consumeStream` would call `finish()` after the + // caller has moved on. streamer.markFinishedExternally(); - pendingFinalStep = step; + if (willAwaitStream) { + pendingFinalStep = step; + } else { + const finishStreamId = await streamer.getOrCreateStreamId(); + await call.save({ step }, false, finishStreamId); + } } else { await call.save({ step }, createPendingMessage); } @@ -162,11 +181,7 @@ export async function streamText< const stream = streamer?.consumeStream( result.toUIMessageStream>(), ); - if ( - (typeof options?.saveStreamDeltas === "object" && - !options.saveStreamDeltas.returnImmediately) || - options?.saveStreamDeltas === true - ) { + if (willAwaitStream) { try { await stream; await result.consumeStream(); From 161cf676d8384161d659e35dd26096da6e67472c Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 16:19:46 -0600 Subject: [PATCH 4/9] Address second-pass review feedback - streamText: guard willAwaitStream so it's only true when threadId is present. Without a thread, no DeltaStreamer is created, and forcing full consumption would block the live response with no place to persist deltas. - asHttpAction: run authorize before consuming the request body, so callbacks that need to read the raw request (HMAC/signature verification, etc.) still have an unread body to work with. - asHttpAction: forward request.signal as abortSignal so a client disconnect halts model execution and tool work instead of running to completion. Co-Authored-By: Claude Opus 4.7 --- src/client/index.ts | 23 +++++++++++++++-------- src/client/streamText.ts | 11 ++++++----- 2 files changed, 21 insertions(+), 13 deletions(-) diff --git a/src/client/index.ts b/src/client/index.ts index a71cdf78..5cebe79c 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -1717,6 +1717,17 @@ export class Agent< request: Request, ) => Promise { return async (ctx_, request) => { + // Run authorize FIRST, before consuming the body, so callbacks that + // need to read the raw request (e.g. HMAC signature verification) + // still have an unread body to work with. + let userId: string | undefined; + let threadId: string | undefined; + if (spec?.authorize) { + const authResult = await spec.authorize(ctx_, request); + if (authResult?.userId) userId = authResult.userId; + if (authResult?.threadId) threadId = authResult.threadId; + } + let body: { threadId?: string; prompt?: string; @@ -1735,14 +1746,7 @@ export class Agent< ); } - let userId: string | undefined; - let threadId: string | undefined = body.threadId; - - if (spec?.authorize) { - const authResult = await spec.authorize(ctx_, request); - if (authResult?.userId) userId = authResult.userId; - if (authResult?.threadId) threadId = authResult.threadId; - } + threadId = threadId ?? body.threadId; const targetArgs = { userId, threadId }; const llmArgs = { @@ -1750,6 +1754,9 @@ export class Agent< promptMessageId: body.promptMessageId, messages: body.messages?.map((m) => toModelMessage(m as never)), stopWhen: spec?.stopWhen, + // Forward the request abort signal so a client disconnect halts + // model execution and tool work instead of running to completion. + abortSignal: request.signal, }; const ctx = ( spec?.customCtx diff --git a/src/client/streamText.ts b/src/client/streamText.ts index 73a06e58..413b943a 100644 --- a/src/client/streamText.ts +++ b/src/client/streamText.ts @@ -84,12 +84,13 @@ export async function streamText< let pendingFinalStep: StepResult | undefined; // Whether streamText will await stream consumption before returning. - // When false (saveStreamDeltas with returnImmediately, or no deltas), - // we cannot defer the final-step save — the deferred work would never run. + // Only true when there's an actual DeltaStreamer to drain — otherwise + // we'd block the response with no persister to drain into. const willAwaitStream = - options.saveStreamDeltas === true || - (typeof options.saveStreamDeltas === "object" && - !options.saveStreamDeltas.returnImmediately); + Boolean(threadId) && + (options.saveStreamDeltas === true || + (typeof options.saveStreamDeltas === "object" && + !options.saveStreamDeltas.returnImmediately)); const streamer = threadId && options.saveStreamDeltas From 028e62a0ccf6b6da648324a25c3a46609db09b6a Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 16:52:28 -0600 Subject: [PATCH 5/9] Drop silent HTTP saveStreamDeltas normalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `saveStreamDeltas: true` was silently rewritten to `{ returnImmediately: true }` for HTTP helpers, so the same option had weaker durability semantics on HTTP than elsewhere — and it was documented in only one type-def comment. Drop the normalization and let the option mean the same thing everywhere: `true` consumes the stream before returning, `{ returnImmediately: true }` opts into background-save streaming. Update HttpStreamOptions and asHttpAction docstrings to spell out the trade-off, and switch the streaming-demo HTTP route to the returnImmediately variant so the demo uses the recommended pattern. Also harden DeltaStreamer.addParts to no-op once markFinishedExternally has been called — the stream record is already "finished" in the DB, so late delta writes would just be dropped by streams.addDelta. Skipping them avoids wasted mutation calls when trailing UI message chunks arrive after onStepFinish has finished the stream atomically with the message save. Co-Authored-By: Claude Opus 4.7 --- example/convex/chat/streamingDemo.ts | 11 ++++++++++- src/client/http.ts | 29 +++++++++------------------- src/client/index.ts | 19 +++++++++++------- src/client/streaming.ts | 7 +++++++ 4 files changed, 38 insertions(+), 28 deletions(-) diff --git a/example/convex/chat/streamingDemo.ts b/example/convex/chat/streamingDemo.ts index 611ae044..fc641640 100644 --- a/example/convex/chat/streamingDemo.ts +++ b/example/convex/chat/streamingDemo.ts @@ -75,9 +75,18 @@ export const streamResponse = internalAction({ // Streams text directly over an HTTP response using `agent.asHttpAction()`. // The handler parses the JSON body, creates a thread if needed, streams // the response, and sets X-Message-Id / X-Stream-Id headers. +// +// `saveStreamDeltas: { returnImmediately: true }` saves deltas in the +// background so `useUIMessages` can dedupe by `streamId` AND the response +// body starts streaming immediately. Plain `true` would buffer the full +// generation before opening the body. // ============================================================================ -export const streamOverHttp = httpAction(streamingDemoAgent.asHttpAction()); +export const streamOverHttp = httpAction( + streamingDemoAgent.asHttpAction({ + saveStreamDeltas: { returnImmediately: true }, + }), +); // ============================================================================ // Pattern 3: One-Shot Streaming diff --git a/src/client/http.ts b/src/client/http.ts index 5306e2e5..1e2a0572 100644 --- a/src/client/http.ts +++ b/src/client/http.ts @@ -18,8 +18,13 @@ export type HttpStreamOptions = Options & { * Whether to save incremental data (deltas) from streaming responses * to the database alongside the HTTP stream. Defaults to false. * - * NOTE: For HTTP flows, `true` is normalized to `{ returnImmediately: true }` - * so the response body starts streaming without waiting for completion. + * Behaviour matches {@link streamText}'s `saveStreamDeltas`: + * - `true` — save deltas; `streamText` consumes the model output before + * returning, so the HTTP response body buffers until generation is + * done. Stronger durability, slower first-byte. + * - `{ returnImmediately: true, ...}` — save deltas in the background + * and return the response immediately so the body actually streams. + * This is the typical HTTP setting. */ saveStreamDeltas?: boolean | StreamingOptions; /** Extra headers to add to the response (e.g. CORS headers). */ @@ -76,7 +81,7 @@ export async function httpStreamText< { ...options, threadId, - saveStreamDeltas: normalizeHttpSaveStreamDeltas(options.saveStreamDeltas), + saveStreamDeltas: options.saveStreamDeltas, }, ); @@ -120,7 +125,7 @@ export async function httpStreamUIMessages< { ...options, threadId, - saveStreamDeltas: normalizeHttpSaveStreamDeltas(options.saveStreamDeltas), + saveStreamDeltas: options.saveStreamDeltas, }, ); @@ -140,22 +145,6 @@ async function resolveThreadId( }); } -/** - * If callers pass `saveStreamDeltas: true` for an HTTP flow, force - * `returnImmediately: true` — otherwise `streamText` consumes the stream - * before returning, which buffers the entire response and defeats the - * purpose of streaming over HTTP. - */ -export function normalizeHttpSaveStreamDeltas( - saveStreamDeltas?: boolean | StreamingOptions, -): boolean | StreamingOptions | undefined { - if (saveStreamDeltas === true) return { returnImmediately: true }; - if (saveStreamDeltas && typeof saveStreamDeltas === "object") { - return { ...saveStreamDeltas, returnImmediately: true }; - } - return saveStreamDeltas; -} - function applyHeaders( response: Response, result: { promptMessageId?: string; streamId?: string }, diff --git a/src/client/index.ts b/src/client/index.ts index 5cebe79c..6ebb4592 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -99,7 +99,6 @@ import type { Output, } from "./types.js"; import { streamText } from "./streamText.js"; -import { normalizeHttpSaveStreamDeltas } from "./http.js"; import { errorToString, willContinue } from "./utils.js"; export { stepCountIs } from "ai"; @@ -1674,7 +1673,12 @@ export class Agent< * http.route({ * path: "/chat", * method: "POST", - * handler: httpAction(myAgent.asHttpAction()), + * handler: httpAction(myAgent.asHttpAction({ + * // Save deltas in the background and start the response immediately + * // (otherwise `saveStreamDeltas: true` buffers the full generation + * // before the body opens). + * saveStreamDeltas: { returnImmediately: true }, + * })), * }); * ``` */ @@ -1684,8 +1688,12 @@ export class Agent< * Whether to save incremental data (deltas) from streaming responses * to the database alongside the HTTP stream. Defaults to false. * - * For HTTP flows, `true` is normalized so the response body starts - * streaming without buffering the entire generation first. + * - `true` — save deltas; `streamText` consumes the model output before + * returning, so the HTTP response body buffers until generation is + * done. Stronger durability, slower first-byte. + * - `{ returnImmediately: true, ...}` — save deltas in the background + * and return the response immediately so the body actually streams. + * This is the typical HTTP setting. */ saveStreamDeltas?: boolean | StreamingOptions; /** @@ -1782,9 +1790,6 @@ export class Agent< // storageOptions, usageHandler, contextHandler, // rawRequestResponseHandler) so they actually take effect. ...spec, - saveStreamDeltas: normalizeHttpSaveStreamDeltas( - spec?.saveStreamDeltas, - ), }, ); diff --git a/src/client/streaming.ts b/src/client/streaming.ts index c3146b43..b737cd36 100644 --- a/src/client/streaming.ts +++ b/src/client/streaming.ts @@ -287,6 +287,13 @@ export class DeltaStreamer { if (this.abortController.signal.aborted) { return; } + // Once the stream has been finished externally (e.g. by an atomic save + // in `streamText`'s onStepFinish), the stream record is already in the + // "finished" state in the DB. Late deltas would be silently dropped by + // `streams.addDelta`, so skip the work. + if (this.#finishedExternally) { + return; + } await this.getStreamId(); this.#nextParts.push(...parts); if ( From 48a7752109560a41082a08b475ee962d96b4c399 Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 17:15:38 -0600 Subject: [PATCH 6/9] Default-deny body.threadId, formalize HTTP header contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Security fix: asHttpAction's authorize callback no longer optionally overrides body values — it now exclusively decides them. body.threadId is ignored unless authorize validates ownership and returns it. A consumer who writes the simplest authorize ({ userId }) gets a fresh thread, not access to whatever thread an attacker passed in the body. The authorize signature changes to (ctx, request, body) — the request is a clone so HMAC users can still read the raw body, and the parsed body is supplied so threadId validation doesn't require re-parsing. Refactor the X-Message-Id / X-Stream-Id header convention into shared constants (HTTP_STREAM_MESSAGE_ID_HEADER, HTTP_STREAM_STREAM_ID_HEADER) in src/shared.ts and a single applyStreamHeaders helper in src/client/http.ts. asHttpAction and useHttpStream both use these so the on-the-wire contract has one source of truth. Also documents the related concerns in docs/http-streaming-requirements.md §8: concurrent-stream interleaving (component-level TODO) and unauthenticated abortStream — both pre-existing but worth flagging now that streamId is a first-class client value. Co-Authored-By: Claude Opus 4.7 --- docs/http-streaming-requirements.md | 61 ++++++++++++++++++- example/convex/chat/streamingDemo.ts | 18 +++++- src/client/http.test.ts | 89 ++++++++++++++++++++++++++++ src/client/http.ts | 36 +++++++++-- src/client/index.ts | 82 +++++++++++++++---------- src/react/useHttpStream.ts | 12 +++- src/shared.ts | 14 +++++ 7 files changed, 270 insertions(+), 42 deletions(-) diff --git a/docs/http-streaming-requirements.md b/docs/http-streaming-requirements.md index 5050d41e..b825817c 100644 --- a/docs/http-streaming-requirements.md +++ b/docs/http-streaming-requirements.md @@ -289,7 +289,66 @@ HTTP streaming additions should be exported from the main surface or a new `@con --- -## 8. Open Questions +## 8. Security Considerations + +HTTP streaming widens the surface area for cross-tenant data exposure compared to the WebSocket path, which sits behind Convex auth on every reactive query. Three invariants must be enforced by consumers: + +### 8.1 thread ownership must be validated by `authorize` + +`Agent.asHttpAction()` will only honor a `threadId` supplied via `authorize`'s return value, not from the JSON body. Without an `authorize` callback (or with one that doesn't return `threadId`), the helper creates a fresh thread and ignores `body.threadId` entirely. This is the default-deny path. + +To accept caller-supplied `threadId` from the body, `authorize` MUST validate ownership before returning it: + +```ts +agent.asHttpAction({ + authorize: async (ctx, _request, body) => { + const userId = await getUserIdFromAuth(ctx); + if (body.threadId) { + // Throws or returns falsy if userId doesn't own the thread. + await assertThreadOwnedBy(ctx, body.threadId, userId); + return { userId, threadId: body.threadId }; + } + return { userId }; + }, +}) +``` + +A consumer who omits validation and returns `body.threadId` unconditionally re-introduces the cross-tenant hole. + +### 8.2 Concurrent streams on the same thread are not deduplicated + +The component schema does not enforce uniqueness on `(threadId, order, stepOrder)` for `streamingMessages` (see `streams.create` — there's an explicit TODO). Two simultaneous streams against the same thread are allowed; both write deltas; subscribers see the interleaved output. + +In practice this matters when: +- the same client triggers two generations before the first finishes, or +- two authenticated callers with thread access stream concurrently. + +If your application can have multiple writers per thread, either: +- single-flight per-thread on your side (mutex / debounce), or +- abort the previous stream before starting a new one (`abortStream` from `@convex-dev/agent`). + +### 8.3 `abortStream` does not authorize the caller + +The component-level `streams.abort` and `streams.abortByOrder` mutations do not check that the caller owns the stream — they accept any valid `streamId` or `(threadId, order)`. The `abortStream` client helper plumbs straight through. If you re-export it as a public mutation, **wrap it in your own ownership check**: + +```ts +export const stopStream = mutation({ + args: { streamId: v.string() }, + handler: async (ctx, { streamId }) => { + const userId = await getUserIdFromAuth(ctx); + await assertStreamOwnedBy(ctx, streamId, userId); + return abortStream(ctx, components.agent, { + streamId, reason: "user", + }); + }, +}); +``` + +The `X-Stream-Id` response header on the HTTP path normalizes `streamId` as a known-to-clients value, which strengthens the case for adding ownership checks at the component layer in a follow-up. Until then, treat `streamId` as a capability and only expose abort behind an auth check. + +--- + +## 9. Open Questions 1. **SSE vs NDJSON**: Should the HTTP transport use SSE (native browser support, automatic reconnection) or NDJSON (simpler, works with `fetch` + `ReadableStream`)? diff --git a/example/convex/chat/streamingDemo.ts b/example/convex/chat/streamingDemo.ts index fc641640..eb2d1673 100644 --- a/example/convex/chat/streamingDemo.ts +++ b/example/convex/chat/streamingDemo.ts @@ -73,18 +73,32 @@ export const streamResponse = internalAction({ // Pattern 2: HTTP Streaming // // Streams text directly over an HTTP response using `agent.asHttpAction()`. -// The handler parses the JSON body, creates a thread if needed, streams -// the response, and sets X-Message-Id / X-Stream-Id headers. +// The handler parses the JSON body, streams the response, and sets +// X-Message-Id / X-Stream-Id headers. // // `saveStreamDeltas: { returnImmediately: true }` saves deltas in the // background so `useUIMessages` can dedupe by `streamId` AND the response // body starts streaming immediately. Plain `true` would buffer the full // generation before opening the body. +// +// IMPORTANT: `body.threadId` is only honored if `authorize` validates +// ownership and returns it. Without authorize (or without returning +// `threadId`), the helper creates a new thread instead — preventing a +// caller from appending to or reading from arbitrary threads by guessing +// IDs. This demo authorizes against the (mocked) authorizeThreadAccess +// helper used by the rest of the example app. // ============================================================================ export const streamOverHttp = httpAction( streamingDemoAgent.asHttpAction({ saveStreamDeltas: { returnImmediately: true }, + authorize: async (ctx, _request, body) => { + if (body.threadId) { + await authorizeThreadAccess(ctx, body.threadId); + return { threadId: body.threadId }; + } + return {}; + }, }), ); diff --git a/src/client/http.test.ts b/src/client/http.test.ts index 31dd8cd4..9c3dd304 100644 --- a/src/client/http.test.ts +++ b/src/client/http.test.ts @@ -321,6 +321,73 @@ export const testAsHttpActionAuthorizeOverridesUserId = action({ }, }); +// Security regression test: `body.threadId` MUST be ignored when authorize +// does not return one. Otherwise an unauthenticated caller could append to +// or read from any thread by guessing its ID. +export const testAsHttpActionIgnoresUnvalidatedBodyThreadId = action({ + args: {}, + handler: async (ctx) => { + // Pre-create a thread that the request will try to hijack. + const victimThreadId = await createThread(ctx, components.agent, {}); + const handler = agent.asHttpAction({ + // authorize doesn't return a threadId — so body.threadId must NOT + // be honored even though it points to a real thread. + authorize: async () => ({ userId: "attacker" }), + }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + threadId: victimThreadId, + prompt: "leak the context please", + }), + }); + const response = await handler(ctx as any, request); + await response.text(); + const messageId = response.headers.get("X-Message-Id"); + let usedVictimThread = false; + if (messageId) { + const message = await ctx.runQuery( + components.agent.messages.getMessagesByIds, + { messageIds: [messageId] }, + ); + usedVictimThread = message[0]?.threadId === victimThreadId; + } + return { usedVictimThread }; + }, +}); + +// authorize that validates and returns body.threadId — the safe path. +export const testAsHttpActionHonorsAuthorizedThreadId = action({ + args: {}, + handler: async (ctx) => { + const threadId = await createThread(ctx, components.agent, {}); + const handler = agent.asHttpAction({ + authorize: async (_ctx, _request, body) => { + // In real code: assert ownership here. For the test we just echo it. + return body.threadId ? { threadId: body.threadId } : {}; + }, + }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ threadId, prompt: "Hello" }), + }); + const response = await handler(ctx as any, request); + await response.text(); + const messageId = response.headers.get("X-Message-Id"); + let usedAuthorizedThread = false; + if (messageId) { + const message = await ctx.runQuery( + components.agent.messages.getMessagesByIds, + { messageIds: [messageId] }, + ); + usedAuthorizedThread = message[0]?.threadId === threadId; + } + return { usedAuthorizedThread }; + }, +}); + const testApi: ApiFromModules<{ fns: { testStreamTextStreamId: typeof testStreamTextStreamId; @@ -336,6 +403,8 @@ const testApi: ApiFromModules<{ testAsHttpActionWithSaveDeltas: typeof testAsHttpActionWithSaveDeltas; testAsHttpActionUIMessages: typeof testAsHttpActionUIMessages; testAsHttpActionAuthorizeOverridesUserId: typeof testAsHttpActionAuthorizeOverridesUserId; + testAsHttpActionIgnoresUnvalidatedBodyThreadId: typeof testAsHttpActionIgnoresUnvalidatedBodyThreadId; + testAsHttpActionHonorsAuthorizedThreadId: typeof testAsHttpActionHonorsAuthorizedThreadId; }; }>["fns"] = anyApi["http.test"] as any; @@ -452,4 +521,24 @@ describe("agent.asHttpAction()", () => { expect(result.status).toBe(200); expect(result.hasMessageId).toBe(true); }); + + test("body.threadId is ignored unless authorize returns it", async () => { + const t = initConvexTest(schema); + const result = await t.action( + testApi.testAsHttpActionIgnoresUnvalidatedBodyThreadId, + {}, + ); + // Without authorize returning a threadId, the helper must create a + // fresh thread instead of writing into the victim's. + expect(result.usedVictimThread).toBe(false); + }); + + test("authorize-returned threadId is honored", async () => { + const t = initConvexTest(schema); + const result = await t.action( + testApi.testAsHttpActionHonorsAuthorizedThreadId, + {}, + ); + expect(result.usedAuthorizedThread).toBe(true); + }); }); diff --git a/src/client/http.ts b/src/client/http.ts index 1e2a0572..743040c9 100644 --- a/src/client/http.ts +++ b/src/client/http.ts @@ -9,6 +9,23 @@ import type { Output, } from "./types.js"; import type { StreamingOptions } from "./streaming.js"; +import { + HTTP_STREAM_MESSAGE_ID_HEADER, + HTTP_STREAM_STREAM_ID_HEADER, +} from "../shared.js"; + +export { + HTTP_STREAM_MESSAGE_ID_HEADER, + HTTP_STREAM_STREAM_ID_HEADER, +} from "../shared.js"; + +/** JSON request body shape consumed by HTTP streaming helpers. */ +export type HttpStreamRequestBody = { + threadId?: string; + prompt?: string; + promptMessageId?: string; + messages?: unknown[]; +}; export type HttpStreamOptions = Options & { agentName: string; @@ -86,7 +103,7 @@ export async function httpStreamText< ); const response = result.toTextStreamResponse(); - applyHeaders(response, result, options.corsHeaders); + applyStreamHeaders(response, result, options.corsHeaders); return response; } @@ -130,7 +147,7 @@ export async function httpStreamUIMessages< ); const response = result.toUIMessageStreamResponse(); - applyHeaders(response, result, options.corsHeaders); + applyStreamHeaders(response, result, options.corsHeaders); return response; } @@ -145,16 +162,25 @@ async function resolveThreadId( }); } -function applyHeaders( +/** + * Set the standard HTTP-streaming response headers (`X-Message-Id`, + * `X-Stream-Id`) plus any caller-supplied CORS headers. Shared between + * the standalone helpers and `Agent.asHttpAction()` so the on-the-wire + * contract has a single source of truth. + */ +export function applyStreamHeaders( response: Response, result: { promptMessageId?: string; streamId?: string }, corsHeaders?: Record, ) { if (result.promptMessageId) { - response.headers.set("X-Message-Id", result.promptMessageId); + response.headers.set( + HTTP_STREAM_MESSAGE_ID_HEADER, + result.promptMessageId, + ); } if (result.streamId) { - response.headers.set("X-Stream-Id", result.streamId); + response.headers.set(HTTP_STREAM_STREAM_ID_HEADER, result.streamId); } if (corsHeaders) { for (const [key, value] of Object.entries(corsHeaders)) { diff --git a/src/client/index.ts b/src/client/index.ts index 6ebb4592..81bce174 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -99,13 +99,21 @@ import type { Output, } from "./types.js"; import { streamText } from "./streamText.js"; +import { + applyStreamHeaders, + type HttpStreamRequestBody, +} from "./http.js"; import { errorToString, willContinue } from "./utils.js"; export { stepCountIs } from "ai"; export { + HTTP_STREAM_MESSAGE_ID_HEADER, + HTTP_STREAM_STREAM_ID_HEADER, + applyStreamHeaders, httpStreamText, httpStreamUIMessages, type HttpStreamOptions, + type HttpStreamRequestBody, } from "./http.js"; export { docsToModelMessages, @@ -1711,13 +1719,34 @@ export class Agent< /** Extra headers to add to the response (e.g. CORS headers). */ corsHeaders?: Record; /** - * Optional authorization callback. Receives the raw request and may - * return `{ userId?, threadId? }` to override values from the body. + * Authorization callback. Receives the action ctx, a clone of the + * raw request (so HMAC verification can still read the body), and + * the parsed JSON body. + * + * IMPORTANT: `body.threadId` is **only** honored if you return a + * `threadId` from this callback after validating that the caller + * owns the thread. Without an explicit return, body.threadId is + * ignored and a new thread is created. This prevents a caller from + * appending to or reading from an arbitrary thread by guessing IDs. + * * Throw to reject the request. + * + * @example + * ```ts + * authorize: async (ctx, _request, body) => { + * const userId = await getUserIdFromAuth(ctx); + * if (body.threadId) { + * await assertThreadOwnedBy(ctx, body.threadId, userId); + * return { userId, threadId: body.threadId }; + * } + * return { userId }; + * } + * ``` */ authorize?: ( ctx: GenericActionCtx, request: Request, + body: HttpStreamRequestBody, ) => Promise<{ userId?: string; threadId?: string } | void>; } & Options, ): ( @@ -1725,25 +1754,13 @@ export class Agent< request: Request, ) => Promise { return async (ctx_, request) => { - // Run authorize FIRST, before consuming the body, so callbacks that - // need to read the raw request (e.g. HMAC signature verification) - // still have an unread body to work with. - let userId: string | undefined; - let threadId: string | undefined; - if (spec?.authorize) { - const authResult = await spec.authorize(ctx_, request); - if (authResult?.userId) userId = authResult.userId; - if (authResult?.threadId) threadId = authResult.threadId; - } + // Clone before parsing so `authorize` can read the raw body if it + // needs to (e.g. HMAC signature verification). + const cloned = request.clone(); - let body: { - threadId?: string; - prompt?: string; - promptMessageId?: string; - messages?: unknown[]; - }; + let body: HttpStreamRequestBody; try { - body = (await request.json()) as typeof body; + body = (await request.json()) as HttpStreamRequestBody; } catch { return new Response( JSON.stringify({ error: "Invalid JSON in request body" }), @@ -1754,7 +1771,19 @@ export class Agent< ); } - threadId = threadId ?? body.threadId; + let userId: string | undefined; + let authorizedThreadId: string | undefined; + if (spec?.authorize) { + const authResult = await spec.authorize(ctx_, cloned, body); + if (authResult?.userId) userId = authResult.userId; + if (authResult?.threadId) authorizedThreadId = authResult.threadId; + } + + // Default-deny: only honor a thread the authorize callback has + // explicitly returned (after validating ownership). If authorize + // didn't run or didn't return a threadId, we ignore body.threadId + // and create a fresh thread below. + let threadId: string | undefined = authorizedThreadId; const targetArgs = { userId, threadId }; const llmArgs = { @@ -1797,18 +1826,7 @@ export class Agent< spec?.format === "ui-messages" ? result.toUIMessageStreamResponse() : result.toTextStreamResponse(); - - if (result.promptMessageId) { - response.headers.set("X-Message-Id", result.promptMessageId); - } - if (result.streamId) { - response.headers.set("X-Stream-Id", result.streamId); - } - if (spec?.corsHeaders) { - for (const [key, value] of Object.entries(spec.corsHeaders)) { - response.headers.set(key, value); - } - } + applyStreamHeaders(response, result, spec?.corsHeaders); return response; }; } diff --git a/src/react/useHttpStream.ts b/src/react/useHttpStream.ts index b8d6b123..4187ded7 100644 --- a/src/react/useHttpStream.ts +++ b/src/react/useHttpStream.ts @@ -1,5 +1,9 @@ "use client"; import { useCallback, useRef, useState } from "react"; +import { + HTTP_STREAM_MESSAGE_ID_HEADER, + HTTP_STREAM_STREAM_ID_HEADER, +} from "../shared.js"; import { consumeTextStream } from "./httpStreamUtils.js"; /** @@ -94,8 +98,12 @@ export function useHttpStream(options: { throw new Error(`HTTP error! status: ${response.status}`); } - const responseStreamId = response.headers.get("X-Stream-Id"); - const responseMessageId = response.headers.get("X-Message-Id"); + const responseStreamId = response.headers.get( + HTTP_STREAM_STREAM_ID_HEADER, + ); + const responseMessageId = response.headers.get( + HTTP_STREAM_MESSAGE_ID_HEADER, + ); if (requestId === requestIdRef.current) { if (responseStreamId) setStreamId(responseStreamId); if (responseMessageId) setMessageId(responseMessageId); diff --git a/src/shared.ts b/src/shared.ts index 79452e53..7952902d 100644 --- a/src/shared.ts +++ b/src/shared.ts @@ -17,6 +17,20 @@ import type { Message, MessageContentParts } from "./validators.js"; export const DEFAULT_RECENT_MESSAGES = 100; +/** + * Header name carrying the prompt message ID on HTTP streaming responses. + * Clients (e.g. `useHttpStream`) read this to associate the streamed + * response with a persisted message. + */ +export const HTTP_STREAM_MESSAGE_ID_HEADER = "X-Message-Id"; + +/** + * Header name carrying the delta stream ID on HTTP streaming responses. + * Clients pass this to `useUIMessages`'s `skipStreamIds` to dedupe HTTP + * stream content against the persisted delta stream. + */ +export const HTTP_STREAM_STREAM_ID_HEADER = "X-Stream-Id"; + export function isTool(message: Message | ModelMessage) { return ( message.role === "tool" || From 08f45f9ee7209772c659a97614f905dfd4ebdbd4 Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 17:21:34 -0600 Subject: [PATCH 7/9] Add userId-based defense-in-depth for abort and tenant mismatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit streams.abort and streams.abortByOrder now accept an optional userId. When supplied, the component verifies the stream's thread is owned by that user and rejects the mutation otherwise. The abortStream client helper plumbs userId through. Internal abort paths (DeltaStreamer's own cleanup) skip the check by not passing userId — they're already in an authorized context. startGeneration now rejects calls where opts.userId is set, the thread is owned by a different user, and the two don't match. This catches a misconfigured authorize callback in asHttpAction (or any caller that hands streamText a userId/threadId pair it shouldn't have) at the component layer, even if upstream auth missed it. Adds a regression test for the tenant-mismatch case and updates docs/http-streaming-requirements.md §8.3-§8.4 to reflect the new defense-in-depth surfaces. Co-Authored-By: Claude Opus 4.7 --- docs/http-streaming-requirements.md | 15 +++++--- src/client/http.test.ts | 42 +++++++++++++++++++++++ src/client/start.ts | 26 ++++++++++---- src/client/streaming.ts | 15 +++++++- src/component/_generated/component.ts | 8 ++++- src/component/streams.ts | 49 +++++++++++++++++++++++++-- 6 files changed, 140 insertions(+), 15 deletions(-) diff --git a/docs/http-streaming-requirements.md b/docs/http-streaming-requirements.md index b825817c..dedfaa15 100644 --- a/docs/http-streaming-requirements.md +++ b/docs/http-streaming-requirements.md @@ -327,24 +327,29 @@ If your application can have multiple writers per thread, either: - single-flight per-thread on your side (mutex / debounce), or - abort the previous stream before starting a new one (`abortStream` from `@convex-dev/agent`). -### 8.3 `abortStream` does not authorize the caller +### 8.3 `abortStream` accepts an optional `userId` for defense-in-depth -The component-level `streams.abort` and `streams.abortByOrder` mutations do not check that the caller owns the stream — they accept any valid `streamId` or `(threadId, order)`. The `abortStream` client helper plumbs straight through. If you re-export it as a public mutation, **wrap it in your own ownership check**: +The `abortStream` client helper now accepts an optional `userId`. When supplied, the component verifies the stream's thread is owned by that user and rejects the mutation otherwise. Pass it whenever you have a userId in scope: ```ts export const stopStream = mutation({ args: { streamId: v.string() }, handler: async (ctx, { streamId }) => { const userId = await getUserIdFromAuth(ctx); - await assertStreamOwnedBy(ctx, streamId, userId); return abortStream(ctx, components.agent, { - streamId, reason: "user", + streamId, + reason: "user", + userId, // component verifies thread ownership against this }); }, }); ``` -The `X-Stream-Id` response header on the HTTP path normalizes `streamId` as a known-to-clients value, which strengthens the case for adding ownership checks at the component layer in a follow-up. Until then, treat `streamId` as a capability and only expose abort behind an auth check. +The check is opt-in (a missing `userId` skips it) so internal abort paths inside the agent itself — which run inside an action that has already authorized the thread — are not regressed. For any consumer code that re-exposes `abortStream`, always pass `userId`. + +### 8.4 streamText / generateText reject mismatched (userId, threadId) + +`startGeneration` validates that, when both `opts.userId` and a thread userId are present, they match. A misconfigured `authorize` callback that pairs the requester's userId with a thread owned by a different user is rejected at the component layer with a clear error rather than silently leaking cross-tenant context. This is the second line of defense behind the `body.threadId` invariant in §8.1. --- diff --git a/src/client/http.test.ts b/src/client/http.test.ts index 9c3dd304..d21f057b 100644 --- a/src/client/http.test.ts +++ b/src/client/http.test.ts @@ -357,6 +357,38 @@ export const testAsHttpActionIgnoresUnvalidatedBodyThreadId = action({ }, }); +// Defense-in-depth: a misconfigured authorize that pairs a requester's +// userId with a thread owned by a different user must be rejected at the +// component layer rather than silently leaking cross-tenant context. +export const testAsHttpActionRejectsMismatchedUserAndThread = action({ + args: {}, + handler: async (ctx) => { + const victimThreadId = await createThread(ctx, components.agent, { + userId: "victim-user", + }); + const handler = agent.asHttpAction({ + // Misconfigured: returns attacker's userId AND victim's threadId. + authorize: async () => ({ + userId: "attacker-user", + threadId: victimThreadId, + }), + }); + const request = new Request("https://example.com/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ prompt: "leak" }), + }); + let threwTenantMismatch = false; + try { + await handler(ctx as any, request); + } catch (e) { + threwTenantMismatch = + e instanceof Error && /not owned|refusing/.test(e.message); + } + return { threwTenantMismatch }; + }, +}); + // authorize that validates and returns body.threadId — the safe path. export const testAsHttpActionHonorsAuthorizedThreadId = action({ args: {}, @@ -404,6 +436,7 @@ const testApi: ApiFromModules<{ testAsHttpActionUIMessages: typeof testAsHttpActionUIMessages; testAsHttpActionAuthorizeOverridesUserId: typeof testAsHttpActionAuthorizeOverridesUserId; testAsHttpActionIgnoresUnvalidatedBodyThreadId: typeof testAsHttpActionIgnoresUnvalidatedBodyThreadId; + testAsHttpActionRejectsMismatchedUserAndThread: typeof testAsHttpActionRejectsMismatchedUserAndThread; testAsHttpActionHonorsAuthorizedThreadId: typeof testAsHttpActionHonorsAuthorizedThreadId; }; }>["fns"] = anyApi["http.test"] as any; @@ -533,6 +566,15 @@ describe("agent.asHttpAction()", () => { expect(result.usedVictimThread).toBe(false); }); + test("rejects a misconfigured authorize pairing wrong user with thread", async () => { + const t = initConvexTest(schema); + const result = await t.action( + testApi.testAsHttpActionRejectsMismatchedUserAndThread, + {}, + ); + expect(result.threwTenantMismatch).toBe(true); + }); + test("authorize-returned threadId is honored", async () => { const t = initConvexTest(schema); const result = await t.action( diff --git a/src/client/start.ts b/src/client/start.ts index b09ee766..c4748dec 100644 --- a/src/client/start.ts +++ b/src/client/start.ts @@ -117,12 +117,26 @@ export async function startGeneration< fail: (reason: string) => Promise; getSavedMessages: () => MessageDoc[]; }> { - const userId = - opts.userId ?? - (threadId && - (await ctx.runQuery(component.threads.getThread, { threadId })) - ?.userId) ?? - undefined; + const thread = threadId + ? await ctx.runQuery(component.threads.getThread, { threadId }) + : undefined; + + // Defense-in-depth tenant check: if the caller asserted a userId AND + // the thread is owned by a (different) user, refuse the operation. + // Catches a misconfigured `authorize` callback that pairs the requester's + // userId with a thread it doesn't own. + if ( + opts.userId != null && + thread?.userId != null && + thread.userId !== opts.userId + ) { + throw new Error( + `Thread ${threadId} is owned by ${thread.userId}; ` + + `refusing to operate on it as user ${opts.userId}`, + ); + } + + const userId = opts.userId ?? thread?.userId ?? undefined; const context = await fetchContextWithPrompt(ctx, component, { ...opts, diff --git a/src/client/streaming.ts b/src/client/streaming.ts index b737cd36..7d83b77d 100644 --- a/src/client/streaming.ts +++ b/src/client/streaming.ts @@ -80,10 +80,21 @@ export async function syncStreams( } } +/** + * Abort an in-progress stream. + * + * Pass `userId` whenever the caller has authenticated a user — the + * component will verify the stream's thread is owned by that user and + * reject the mutation otherwise. This is defense-in-depth: it catches + * a misconfigured wrapper mutation that forgets to validate ownership + * before re-exposing this helper. Skip `userId` only when the calling + * code has already validated ownership and is acting on its own state + * (e.g. internal cleanup paths). + */ export async function abortStream( ctx: MutationCtx | ActionCtx, component: AgentComponent, - args: { reason: string } & ( + args: { reason: string; userId?: string } & ( | { streamId: string } | { threadId: string; order: number } ), @@ -92,12 +103,14 @@ export async function abortStream( return await ctx.runMutation(component.streams.abort, { reason: args.reason, streamId: args.streamId, + userId: args.userId, }); } else { return await ctx.runMutation(component.streams.abortByOrder, { reason: args.reason, threadId: args.threadId, order: args.order, + userId: args.userId, }); } } diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index e808be92..4c7e8c9a 100644 --- a/src/component/_generated/component.ts +++ b/src/component/_generated/component.ts @@ -4467,6 +4467,7 @@ export type ComponentApi = }; reason: string; streamId: string; + userId?: string; }, boolean, Name @@ -4474,7 +4475,12 @@ export type ComponentApi = abortByOrder: FunctionReference< "mutation", "internal", - { order: number; reason: string; threadId: string }, + { + order: number; + reason: string; + threadId: string; + userId?: string; + }, boolean, Name >; diff --git a/src/component/streams.ts b/src/component/streams.ts index 4c3499fe..972aeabb 100644 --- a/src/component/streams.ts +++ b/src/component/streams.ts @@ -160,9 +160,23 @@ function publicStreamMessage(m: Doc<"streamingMessages">): StreamMessage { } export const abortByOrder = mutation({ - args: { threadId: v.id("threads"), order: v.number(), reason: v.string() }, + args: { + threadId: v.id("threads"), + order: v.number(), + reason: v.string(), + /** + * Defense-in-depth: when set, the mutation throws if the thread is + * not owned by this userId. Skip only when the calling code has + * already validated ownership (e.g. the agent's own internal abort + * paths run inside an action that already authorized the thread). + */ + userId: v.optional(v.string()), + }, returns: v.boolean(), handler: async (ctx, args) => { + if (args.userId !== undefined) { + await assertThreadOwnedBy(ctx, args.threadId, args.userId); + } const streams = await ctx.db .query("streamingMessages") .withIndex("threadId_state_order_stepOrder", (q) => @@ -184,11 +198,42 @@ export const abort = mutation({ streamId: v.id("streamingMessages"), reason: v.string(), finalDelta: v.optional(deltaValidator), + /** + * Defense-in-depth: when set, the mutation throws if the stream's + * thread is not owned by this userId. Skip only when the calling + * code has already validated ownership. + */ + userId: v.optional(v.string()), }, returns: v.boolean(), - handler: abortById, + handler: async (ctx, args) => { + if (args.userId !== undefined) { + const stream = await ctx.db.get(args.streamId); + if (!stream) { + throw new Error(`Stream not found: ${args.streamId}`); + } + await assertThreadOwnedBy(ctx, stream.threadId, args.userId); + } + return abortById(ctx, args); + }, }); +async function assertThreadOwnedBy( + ctx: MutationCtx, + threadId: Id<"threads">, + userId: string, +) { + const thread = await ctx.db.get(threadId); + if (!thread) { + throw new Error(`Thread not found: ${threadId}`); + } + if (thread.userId !== userId) { + throw new Error( + `Thread ${threadId} is not owned by ${userId}; refusing operation`, + ); + } +} + async function abortById( ctx: MutationCtx, args: { From 4fb89cb826e350e5ec4c6cb75a3b1be1aaa932fd Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 17:26:55 -0600 Subject: [PATCH 8/9] Make abort ownership enforcement mandatory, not opt-in MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously the userId check on streams.abort / streams.abortByOrder was opt-in: a caller who simply didn't pass userId got the same behavior as before. That made the "defense-in-depth" toothless — any consumer who wraps abortStream without auth and forgets userId is unprotected. Tighten the contract: when the stream / thread has a userId set, the caller MUST supply a matching userId or the mutation throws. Anonymous streams (no userId on the resource) are still abortable without a caller userId, since by definition they're public. Internal abort paths (DeltaStreamer.fail and its abortSignal handler) now plumb metadata.userId through, so the agent's own cleanup keeps working for tenant-bound streams. Adds 3 component-level tests: - mismatched userId rejected - missing userId on tenant-bound stream rejected (with descriptive error message) - anonymous stream still aborts without userId Co-Authored-By: Claude Opus 4.7 --- docs/http-streaming-requirements.md | 15 +++- src/client/streaming.integration.test.ts | 99 ++++++++++++++++++++++++ src/client/streaming.ts | 2 + src/component/streams.ts | 73 +++++++++-------- 4 files changed, 154 insertions(+), 35 deletions(-) diff --git a/docs/http-streaming-requirements.md b/docs/http-streaming-requirements.md index dedfaa15..afcd5696 100644 --- a/docs/http-streaming-requirements.md +++ b/docs/http-streaming-requirements.md @@ -327,9 +327,14 @@ If your application can have multiple writers per thread, either: - single-flight per-thread on your side (mutex / debounce), or - abort the previous stream before starting a new one (`abortStream` from `@convex-dev/agent`). -### 8.3 `abortStream` accepts an optional `userId` for defense-in-depth +### 8.3 `abortStream` enforces ownership at the component layer -The `abortStream` client helper now accepts an optional `userId`. When supplied, the component verifies the stream's thread is owned by that user and rejects the mutation otherwise. Pass it whenever you have a userId in scope: +The component-level `streams.abort` and `streams.abortByOrder` mutations now enforce ownership: + +- If the stream / thread has a `userId`, the caller MUST supply a matching `userId`. Otherwise the mutation throws. +- If the stream / thread is anonymous (no `userId`), the check is skipped — anonymous streams are public. + +This means a consumer who re-exposes `abortStream` as a public mutation without auth can only kill anonymous streams. For any tenant-bound stream, the caller must prove ownership by supplying the userId. ```ts export const stopStream = mutation({ @@ -339,13 +344,15 @@ export const stopStream = mutation({ return abortStream(ctx, components.agent, { streamId, reason: "user", - userId, // component verifies thread ownership against this + userId, // required when the stream was created with a userId }); }, }); ``` -The check is opt-in (a missing `userId` skips it) so internal abort paths inside the agent itself — which run inside an action that has already authorized the thread — are not regressed. For any consumer code that re-exposes `abortStream`, always pass `userId`. +Internal abort paths inside the agent itself (e.g. DeltaStreamer's own cleanup) plumb the originating userId through automatically, so they don't regress. + +The residual risk: an attacker who knows BOTH the streamId and the victim's userId can still abort. userId is typically not secret. If you need stronger isolation, add an additional capability check in your wrapper (e.g. require a per-stream signed token). ### 8.4 streamText / generateText reject mismatched (userId, threadId) diff --git a/src/client/streaming.integration.test.ts b/src/client/streaming.integration.test.ts index 45fbc6fd..37cb8ece 100644 --- a/src/client/streaming.integration.test.ts +++ b/src/client/streaming.integration.test.ts @@ -833,6 +833,105 @@ describe("Fallback Behavior", () => { }); }); + test("abort throws when caller's userId doesn't match owner", async () => { + await t.run(async (ctx) => { + // Owner-bound thread + stream + const ownedThread = await ctx.runMutation( + components.agent.threads.createThread, + { userId: "alice" }, + ); + const ownedThreadId = ownedThread._id; + const streamer = new DeltaStreamer( + components.agent, + ctx, + { ...defaultTestOptions }, + { ...testMetadata, threadId: ownedThreadId, userId: "alice" }, + ); + const streamId = await streamer.getStreamId(); + + // Wrong user (case 1: missing userId) + await expect( + ctx.runMutation(components.agent.streams.abort, { + streamId, + reason: "evil", + }), + ).rejects.toThrow(/userId must be supplied/); + + // Wrong user (case 2: mismatched userId) + await expect( + ctx.runMutation(components.agent.streams.abort, { + streamId, + reason: "evil", + userId: "mallory", + }), + ).rejects.toThrow(/not owned/); + + // Correct user — succeeds + const ok = await ctx.runMutation(components.agent.streams.abort, { + streamId, + reason: "user", + userId: "alice", + }); + expect(ok).toBe(true); + }); + }); + + test("abort permits no-userId calls on anonymous streams", async () => { + await t.run(async (ctx) => { + const streamer = new DeltaStreamer( + components.agent, + ctx, + { ...defaultTestOptions }, + { ...testMetadata, threadId }, + ); + const streamId = await streamer.getStreamId(); + const ok = await ctx.runMutation(components.agent.streams.abort, { + streamId, + reason: "user", + }); + expect(ok).toBe(true); + }); + }); + + test("abortByOrder throws when caller's userId doesn't match thread", async () => { + await t.run(async (ctx) => { + const ownedThread = await ctx.runMutation( + components.agent.threads.createThread, + { userId: "alice" }, + ); + const ownedThreadId = ownedThread._id; + const streamer = new DeltaStreamer( + components.agent, + ctx, + { ...defaultTestOptions }, + { + ...testMetadata, + threadId: ownedThreadId, + userId: "alice", + order: 7, + }, + ); + await streamer.getStreamId(); + + await expect( + ctx.runMutation(components.agent.streams.abortByOrder, { + threadId: ownedThreadId, + order: 7, + reason: "evil", + }), + ).rejects.toThrow(/userId must be supplied/); + + await expect( + ctx.runMutation(components.agent.streams.abortByOrder, { + threadId: ownedThreadId, + order: 7, + reason: "evil", + userId: "mallory", + }), + ).rejects.toThrow(/not owned/); + }); + }); + test("fail on already-aborted stream is a no-op", async () => { await t.run(async (ctx) => { const streamer = new DeltaStreamer( diff --git a/src/client/streaming.ts b/src/client/streaming.ts index 7d83b77d..fe07ebfe 100644 --- a/src/client/streaming.ts +++ b/src/client/streaming.ts @@ -271,6 +271,7 @@ export class DeltaStreamer { await this.ctx.runMutation(this.component.streams.abort, { streamId: this.streamId, reason: "abortSignal", + userId: this.metadata.userId, }); } } catch { @@ -428,6 +429,7 @@ export class DeltaStreamer { await this.ctx.runMutation(this.component.streams.abort, { streamId: this.streamId, reason, + userId: this.metadata.userId, }); } } diff --git a/src/component/streams.ts b/src/component/streams.ts index 972aeabb..88a64c26 100644 --- a/src/component/streams.ts +++ b/src/component/streams.ts @@ -159,24 +159,53 @@ function publicStreamMessage(m: Doc<"streamingMessages">): StreamMessage { }; } +/** + * Enforce that a caller-provided userId matches the owner of the + * resource being aborted. The check is required when the resource is + * owned by some user; for anonymous (no-userId) resources, no check is + * performed. This means a consumer who re-exposes `abortStream` without + * auth can only kill anonymous streams — for any tenant-bound stream, + * the caller MUST supply the matching userId. + */ +function assertOwner( + resourceUserId: string | undefined, + callerUserId: string | undefined, + resourceLabel: string, +) { + if (resourceUserId === undefined) { + return; + } + if (callerUserId === undefined) { + throw new Error( + `${resourceLabel} is owned by a user; userId must be supplied to abort it.`, + ); + } + if (callerUserId !== resourceUserId) { + throw new Error( + `${resourceLabel} is not owned by ${callerUserId}; refusing operation.`, + ); + } +} + export const abortByOrder = mutation({ args: { threadId: v.id("threads"), order: v.number(), reason: v.string(), /** - * Defense-in-depth: when set, the mutation throws if the thread is - * not owned by this userId. Skip only when the calling code has - * already validated ownership (e.g. the agent's own internal abort - * paths run inside an action that already authorized the thread). + * Required to abort a stream on a tenant-bound thread (one with a + * userId). Anonymous threads (no userId) may be aborted without + * supplying a userId. */ userId: v.optional(v.string()), }, returns: v.boolean(), handler: async (ctx, args) => { - if (args.userId !== undefined) { - await assertThreadOwnedBy(ctx, args.threadId, args.userId); + const thread = await ctx.db.get(args.threadId); + if (!thread) { + throw new Error(`Thread not found: ${args.threadId}`); } + assertOwner(thread.userId, args.userId, `Thread ${args.threadId}`); const streams = await ctx.db .query("streamingMessages") .withIndex("threadId_state_order_stepOrder", (q) => @@ -199,41 +228,23 @@ export const abort = mutation({ reason: v.string(), finalDelta: v.optional(deltaValidator), /** - * Defense-in-depth: when set, the mutation throws if the stream's - * thread is not owned by this userId. Skip only when the calling - * code has already validated ownership. + * Required to abort a stream that was created with a userId. + * Anonymous streams (no userId) may be aborted without supplying + * a userId. */ userId: v.optional(v.string()), }, returns: v.boolean(), handler: async (ctx, args) => { - if (args.userId !== undefined) { - const stream = await ctx.db.get(args.streamId); - if (!stream) { - throw new Error(`Stream not found: ${args.streamId}`); - } - await assertThreadOwnedBy(ctx, stream.threadId, args.userId); + const stream = await ctx.db.get(args.streamId); + if (!stream) { + throw new Error(`Stream not found: ${args.streamId}`); } + assertOwner(stream.userId, args.userId, `Stream ${args.streamId}`); return abortById(ctx, args); }, }); -async function assertThreadOwnedBy( - ctx: MutationCtx, - threadId: Id<"threads">, - userId: string, -) { - const thread = await ctx.db.get(threadId); - if (!thread) { - throw new Error(`Thread not found: ${threadId}`); - } - if (thread.userId !== userId) { - throw new Error( - `Thread ${threadId} is not owned by ${userId}; refusing operation`, - ); - } -} - async function abortById( ctx: MutationCtx, args: { From 6568c5eac147aa010ecd8e175d42bfeb10337121 Mon Sep 17 00:00:00 2001 From: Seth Raphael Date: Sat, 9 May 2026 18:37:42 -0600 Subject: [PATCH 9/9] Drop abort userId check; auth is the consumer's job MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous "userId required" enforcement was security theatre. streams.list exposes streamId AND userId pairs to anyone with thread read access, the HTTP path returns the requester's own streamId in X-Stream-Id, and userId is generally not a secret. Requiring those two values to abort doesn't actually prevent the attack the check was supposed to defend against. Components in Convex can't authenticate callers — they have no access to ctx.auth. Real auth has to happen in the consumer's wrapping mutation. Add a clear note in the docs (§8.3) and the abortStream JSDoc spelling this out, with the recommended pattern. Reverts: - streams.abort / streams.abortByOrder no longer take userId - abortStream client helper no longer takes userId - DeltaStreamer's internal abort callsites no longer pass it - _generated/component.ts back to original shape - 3 abort-enforcement tests removed The tenant-mismatch check in startGeneration stays — that one is checking that authorize's returned (userId, threadId) pair is internally consistent against the thread's actual owner. It catches a real misconfig pattern (authorize pairing a userId with someone else's threadId), not a guessable-IDs attack. Co-Authored-By: Claude Opus 4.7 --- docs/http-streaming-requirements.md | 18 ++--- src/client/streaming.integration.test.ts | 99 ------------------------ src/client/streaming.ts | 20 ++--- src/component/_generated/component.ts | 8 +- src/component/streams.ts | 65 ++-------------- 5 files changed, 24 insertions(+), 186 deletions(-) diff --git a/docs/http-streaming-requirements.md b/docs/http-streaming-requirements.md index afcd5696..4fca70cf 100644 --- a/docs/http-streaming-requirements.md +++ b/docs/http-streaming-requirements.md @@ -327,32 +327,30 @@ If your application can have multiple writers per thread, either: - single-flight per-thread on your side (mutex / debounce), or - abort the previous stream before starting a new one (`abortStream` from `@convex-dev/agent`). -### 8.3 `abortStream` enforces ownership at the component layer +### 8.3 `abortStream` is unauthenticated — wrap it with auth -The component-level `streams.abort` and `streams.abortByOrder` mutations now enforce ownership: +`streams.abort`, `streams.abortByOrder`, and the `abortStream` client helper do **not** authenticate the caller. Components in Convex have no access to `ctx.auth`; the only args we could require (`userId`, `streamId`, `threadId`) are not secret — they leak via `streams.list`, profile pages, URL hashes, and the `X-Stream-Id` response header. Adding an opt-in `userId` check at the component layer would be theatre, not a security boundary. -- If the stream / thread has a `userId`, the caller MUST supply a matching `userId`. Otherwise the mutation throws. -- If the stream / thread is anonymous (no `userId`), the check is skipped — anonymous streams are public. - -This means a consumer who re-exposes `abortStream` as a public mutation without auth can only kill anonymous streams. For any tenant-bound stream, the caller must prove ownership by supplying the userId. +Auth is the **consumer's** responsibility. Always wrap `abortStream` in a public mutation that runs `ctx.auth.getUserIdentity()` (or your project's equivalent) and validates the caller owns the stream BEFORE calling it: ```ts export const stopStream = mutation({ args: { streamId: v.string() }, handler: async (ctx, { streamId }) => { - const userId = await getUserIdFromAuth(ctx); + const userId = await getAuthUserIdOrThrow(ctx); + // Look up the stream's thread, assert this user owns it. + await assertStreamOwnedBy(ctx, streamId, userId); return abortStream(ctx, components.agent, { streamId, reason: "user", - userId, // required when the stream was created with a userId }); }, }); ``` -Internal abort paths inside the agent itself (e.g. DeltaStreamer's own cleanup) plumb the originating userId through automatically, so they don't regress. +If you re-export `abortStream` (or a thin wrapper around it) as a public mutation without an auth + ownership check first, **any caller who can guess or enumerate `streamId` can kill any stream**. `streams.list` exposes streamIds for any thread the caller has read access to, and the HTTP path returns the requester's own streamId in the `X-Stream-Id` header — so streamIds should be treated as known-to-clients values, never as capabilities. -The residual risk: an attacker who knows BOTH the streamId and the victim's userId can still abort. userId is typically not secret. If you need stronger isolation, add an additional capability check in your wrapper (e.g. require a per-stream signed token). +If you need a true capability-style abort that doesn't depend on consumer-level auth (e.g. for unauthenticated streaming flows), that's a separate design — a per-stream secret token returned at create time and required at abort time. Not implemented here. ### 8.4 streamText / generateText reject mismatched (userId, threadId) diff --git a/src/client/streaming.integration.test.ts b/src/client/streaming.integration.test.ts index 37cb8ece..45fbc6fd 100644 --- a/src/client/streaming.integration.test.ts +++ b/src/client/streaming.integration.test.ts @@ -833,105 +833,6 @@ describe("Fallback Behavior", () => { }); }); - test("abort throws when caller's userId doesn't match owner", async () => { - await t.run(async (ctx) => { - // Owner-bound thread + stream - const ownedThread = await ctx.runMutation( - components.agent.threads.createThread, - { userId: "alice" }, - ); - const ownedThreadId = ownedThread._id; - const streamer = new DeltaStreamer( - components.agent, - ctx, - { ...defaultTestOptions }, - { ...testMetadata, threadId: ownedThreadId, userId: "alice" }, - ); - const streamId = await streamer.getStreamId(); - - // Wrong user (case 1: missing userId) - await expect( - ctx.runMutation(components.agent.streams.abort, { - streamId, - reason: "evil", - }), - ).rejects.toThrow(/userId must be supplied/); - - // Wrong user (case 2: mismatched userId) - await expect( - ctx.runMutation(components.agent.streams.abort, { - streamId, - reason: "evil", - userId: "mallory", - }), - ).rejects.toThrow(/not owned/); - - // Correct user — succeeds - const ok = await ctx.runMutation(components.agent.streams.abort, { - streamId, - reason: "user", - userId: "alice", - }); - expect(ok).toBe(true); - }); - }); - - test("abort permits no-userId calls on anonymous streams", async () => { - await t.run(async (ctx) => { - const streamer = new DeltaStreamer( - components.agent, - ctx, - { ...defaultTestOptions }, - { ...testMetadata, threadId }, - ); - const streamId = await streamer.getStreamId(); - const ok = await ctx.runMutation(components.agent.streams.abort, { - streamId, - reason: "user", - }); - expect(ok).toBe(true); - }); - }); - - test("abortByOrder throws when caller's userId doesn't match thread", async () => { - await t.run(async (ctx) => { - const ownedThread = await ctx.runMutation( - components.agent.threads.createThread, - { userId: "alice" }, - ); - const ownedThreadId = ownedThread._id; - const streamer = new DeltaStreamer( - components.agent, - ctx, - { ...defaultTestOptions }, - { - ...testMetadata, - threadId: ownedThreadId, - userId: "alice", - order: 7, - }, - ); - await streamer.getStreamId(); - - await expect( - ctx.runMutation(components.agent.streams.abortByOrder, { - threadId: ownedThreadId, - order: 7, - reason: "evil", - }), - ).rejects.toThrow(/userId must be supplied/); - - await expect( - ctx.runMutation(components.agent.streams.abortByOrder, { - threadId: ownedThreadId, - order: 7, - reason: "evil", - userId: "mallory", - }), - ).rejects.toThrow(/not owned/); - }); - }); - test("fail on already-aborted stream is a no-op", async () => { await t.run(async (ctx) => { const streamer = new DeltaStreamer( diff --git a/src/client/streaming.ts b/src/client/streaming.ts index fe07ebfe..6dd9015d 100644 --- a/src/client/streaming.ts +++ b/src/client/streaming.ts @@ -83,18 +83,18 @@ export async function syncStreams( /** * Abort an in-progress stream. * - * Pass `userId` whenever the caller has authenticated a user — the - * component will verify the stream's thread is owned by that user and - * reject the mutation otherwise. This is defense-in-depth: it catches - * a misconfigured wrapper mutation that forgets to validate ownership - * before re-exposing this helper. Skip `userId` only when the calling - * code has already validated ownership and is acting on its own state - * (e.g. internal cleanup paths). + * IMPORTANT: This is an unauthenticated primitive. The component layer + * has no access to `ctx.auth` and cannot verify the caller. Consumers + * MUST authenticate and authorize the caller (e.g. via + * `ctx.auth.getUserIdentity()` and a thread-ownership check) BEFORE + * calling this helper from a public mutation. Re-exposing `abortStream` + * without an auth wrapper lets any caller kill any stream whose ID they + * can guess or enumerate. */ export async function abortStream( ctx: MutationCtx | ActionCtx, component: AgentComponent, - args: { reason: string; userId?: string } & ( + args: { reason: string } & ( | { streamId: string } | { threadId: string; order: number } ), @@ -103,14 +103,12 @@ export async function abortStream( return await ctx.runMutation(component.streams.abort, { reason: args.reason, streamId: args.streamId, - userId: args.userId, }); } else { return await ctx.runMutation(component.streams.abortByOrder, { reason: args.reason, threadId: args.threadId, order: args.order, - userId: args.userId, }); } } @@ -271,7 +269,6 @@ export class DeltaStreamer { await this.ctx.runMutation(this.component.streams.abort, { streamId: this.streamId, reason: "abortSignal", - userId: this.metadata.userId, }); } } catch { @@ -429,7 +426,6 @@ export class DeltaStreamer { await this.ctx.runMutation(this.component.streams.abort, { streamId: this.streamId, reason, - userId: this.metadata.userId, }); } } diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index 4c7e8c9a..e808be92 100644 --- a/src/component/_generated/component.ts +++ b/src/component/_generated/component.ts @@ -4467,7 +4467,6 @@ export type ComponentApi = }; reason: string; streamId: string; - userId?: string; }, boolean, Name @@ -4475,12 +4474,7 @@ export type ComponentApi = abortByOrder: FunctionReference< "mutation", "internal", - { - order: number; - reason: string; - threadId: string; - userId?: string; - }, + { order: number; reason: string; threadId: string }, boolean, Name >; diff --git a/src/component/streams.ts b/src/component/streams.ts index 88a64c26..faf878f9 100644 --- a/src/component/streams.ts +++ b/src/component/streams.ts @@ -159,53 +159,15 @@ function publicStreamMessage(m: Doc<"streamingMessages">): StreamMessage { }; } -/** - * Enforce that a caller-provided userId matches the owner of the - * resource being aborted. The check is required when the resource is - * owned by some user; for anonymous (no-userId) resources, no check is - * performed. This means a consumer who re-exposes `abortStream` without - * auth can only kill anonymous streams — for any tenant-bound stream, - * the caller MUST supply the matching userId. - */ -function assertOwner( - resourceUserId: string | undefined, - callerUserId: string | undefined, - resourceLabel: string, -) { - if (resourceUserId === undefined) { - return; - } - if (callerUserId === undefined) { - throw new Error( - `${resourceLabel} is owned by a user; userId must be supplied to abort it.`, - ); - } - if (callerUserId !== resourceUserId) { - throw new Error( - `${resourceLabel} is not owned by ${callerUserId}; refusing operation.`, - ); - } -} - +// Abort mutations are unauthenticated primitives — components have no +// access to ctx.auth, and the only "ownership" args we could require +// (userId, streamId) are not secret. Authentication and ownership +// validation MUST happen in the consumer's wrapping mutation. See +// docs/http-streaming-requirements.md §8.3 for the recommended pattern. export const abortByOrder = mutation({ - args: { - threadId: v.id("threads"), - order: v.number(), - reason: v.string(), - /** - * Required to abort a stream on a tenant-bound thread (one with a - * userId). Anonymous threads (no userId) may be aborted without - * supplying a userId. - */ - userId: v.optional(v.string()), - }, + args: { threadId: v.id("threads"), order: v.number(), reason: v.string() }, returns: v.boolean(), handler: async (ctx, args) => { - const thread = await ctx.db.get(args.threadId); - if (!thread) { - throw new Error(`Thread not found: ${args.threadId}`); - } - assertOwner(thread.userId, args.userId, `Thread ${args.threadId}`); const streams = await ctx.db .query("streamingMessages") .withIndex("threadId_state_order_stepOrder", (q) => @@ -227,22 +189,9 @@ export const abort = mutation({ streamId: v.id("streamingMessages"), reason: v.string(), finalDelta: v.optional(deltaValidator), - /** - * Required to abort a stream that was created with a userId. - * Anonymous streams (no userId) may be aborted without supplying - * a userId. - */ - userId: v.optional(v.string()), }, returns: v.boolean(), - handler: async (ctx, args) => { - const stream = await ctx.db.get(args.streamId); - if (!stream) { - throw new Error(`Stream not found: ${args.streamId}`); - } - assertOwner(stream.userId, args.userId, `Stream ${args.streamId}`); - return abortById(ctx, args); - }, + handler: abortById, }); async function abortById(