diff --git a/docs/http-streaming-requirements.md b/docs/http-streaming-requirements.md
new file mode 100644
index 00000000..4fca70cf
--- /dev/null
+++ b/docs/http-streaming-requirements.md
@@ -0,0 +1,371 @@
+# 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<string, number>`
+- 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. 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` is unauthenticated — wrap it with auth
+
+`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.
+
+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 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",
+    });
+  },
+});
+```
+
+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.
+
+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)
+
+`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.
+
+---
+
+## 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`)?
+
+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/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..eb2d1673
--- /dev/null
+++ b/example/convex/chat/streamingDemo.ts
@@ -0,0 +1,238 @@
+/**
+ * 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, 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 {};
+    },
+  }),
+);
+
+// ============================================================================
+// 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 @@
 /// <reference types="vite/client" />
 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<GenericSchema, boolean>. 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<GenericSchema, boolean>
+  >;
+  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 (
+    <>
+      <header className="sticky top-0 h-16 z-10 bg-white/80 backdrop-blur-sm p-4 flex justify-between items-center border-b">
+        <h1 className="text-xl font-semibold accent-text">
+          Full Streaming Demo
+        </h1>
+      </header>
+      <div className="h-[calc(100vh-8rem)] flex bg-gray-50">
+        {threadId ? (
+          <DemoApp threadId={threadId} reset={() => void resetThread()} />
+        ) : (
+          <div className="flex-1 flex items-center justify-center text-gray-500">
+            Loading...
+          </div>
+        )}
+      </div>
+    </>
+  );
+}
+
+function DemoApp({
+  threadId,
+  reset,
+}: {
+  threadId: string;
+  reset: () => void;
+}) {
+  const [streamMode, setStreamMode] = useState<StreamMode>("delta");
+
+  return (
+    <div className="flex flex-1 overflow-hidden">
+      {/* Left: Chat area */}
+      <div className="flex-1 flex flex-col min-w-0">
+        <ModeSelector mode={streamMode} onChange={setStreamMode} />
+        <ChatPanel threadId={threadId} mode={streamMode} reset={reset} />
+      </div>
+
+      {/* Right: Stream Inspector */}
+      <div className="w-80 border-l bg-white overflow-y-auto">
+        <StreamInspector threadId={threadId} />
+      </div>
+    </div>
+  );
+}
+
+// ============================================================================
+// 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 (
+    <div className="flex gap-2 p-3 bg-gray-100 border-b">
+      {modes.map((m) => (
+        <button
+          key={m.value}
+          onClick={() => onChange(m.value)}
+          className={cn(
+            "px-3 py-1.5 rounded text-sm font-medium transition",
+            mode === m.value
+              ? "bg-indigo-600 text-white"
+              : "bg-white text-gray-700 hover:bg-gray-200",
+          )}
+          title={m.desc}
+        >
+          {m.label}
+        </button>
+      ))}
+    </div>
+  );
+}
+
+// ============================================================================
+// 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<string | null>(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<StreamMode>(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<HTMLDivElement>(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 (
+    <>
+      <div className="flex-1 overflow-y-auto p-6">
+        {messages.length > 0 || httpText ? (
+          <div className="flex flex-col gap-3 max-w-3xl mx-auto">
+            {status === "CanLoadMore" && (
+              <button
+                className="self-center text-sm text-indigo-600 hover:underline"
+                onClick={() => loadMore(10)}
+              >
+                Load more
+              </button>
+            )}
+            {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) => (
+                <MessageBubble
+                  key={m.key}
+                  message={m}
+                  threadId={threadId}
+                  onApproval={handleApproval}
+                />
+              ))}
+            {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 (
+                <div className="flex justify-start">
+                  <div
+                    className={cn(
+                      "rounded-lg px-4 py-2 max-w-lg whitespace-pre-wrap shadow-sm",
+                      httpStreaming
+                        ? "bg-green-100 text-gray-800"
+                        : "bg-gray-200 text-gray-800",
+                    )}
+                  >
+                    <span className="text-xs text-gray-400 block mb-1">
+                      [HTTP stream{httpStreaming ? " - live" : " - done"}]
+                    </span>
+                    {toolParts.map((p: any) => (
+                      <div
+                        key={p.toolCallId}
+                        className="text-xs bg-gray-100 rounded p-1.5 my-1 font-mono"
+                      >
+                        <span className="text-indigo-600">{p.type}</span>
+                        {p.state && (
+                          <span className="text-gray-400 ml-1">({p.state})</span>
+                        )}
+                        {p.output && (
+                          <div className="text-gray-600 mt-0.5 truncate">
+                            {typeof p.output === "string"
+                              ? p.output
+                              : JSON.stringify(p.output)}
+                          </div>
+                        )}
+                      </div>
+                    ))}
+                    <div>{httpText}</div>
+                  </div>
+                </div>
+              );
+            })()}
+            <div ref={messagesEndRef} />
+          </div>
+        ) : (
+          <div className="flex items-center justify-center h-full text-gray-500">
+            Pick a streaming mode above and start chatting.
+          </div>
+        )}
+      </div>
+
+      <div className="border-t bg-white p-4">
+        <form
+          className="flex gap-2 items-center max-w-3xl mx-auto"
+          onSubmit={(e) => {
+            e.preventDefault();
+            void handleSend();
+          }}
+        >
+          <input
+            type="text"
+            value={prompt}
+            onChange={(e) => 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 ? (
+            <button
+              type="button"
+              className="px-4 py-2 rounded-lg bg-red-100 text-red-700 hover:bg-red-200 transition font-medium"
+              onClick={() => {
+                if (httpStreaming) {
+                  httpStream.abort();
+                }
+                const streaming = messages.find(
+                  (m) => m.status === "streaming",
+                );
+                if (streaming) {
+                  void abortByOrder({ threadId, order: streaming.order });
+                }
+              }}
+            >
+              Abort
+            </button>
+          ) : (
+            <button
+              type="submit"
+              className="px-4 py-2 rounded-lg bg-indigo-600 text-white hover:bg-indigo-700 transition font-semibold disabled:opacity-50"
+              disabled={!prompt.trim() || hasPendingApprovals}
+            >
+              Send
+            </button>
+          )}
+          <button
+            type="button"
+            className="px-3 py-2 rounded-lg bg-gray-100 text-gray-600 hover:bg-gray-200 transition text-sm"
+            onClick={() => {
+              httpStream.abort();
+              lastApprovalMessageIdRef.current = null;
+              continuationTriggeredRef.current = false;
+              reset();
+            }}
+          >
+            Reset
+          </button>
+        </form>
+        <div className="max-w-3xl mx-auto mt-2 text-xs text-gray-400">
+          Mode:{" "}
+          <strong>
+            {mode === "delta"
+              ? "Async Delta Streaming"
+              : mode === "http"
+                ? "HTTP Streaming"
+                : "One-Shot Streaming"}
+          </strong>
+        </div>
+      </div>
+    </>
+  );
+}
+
+// ============================================================================
+// Message Bubble
+// ============================================================================
+
+function MessageBubble({
+  message,
+  threadId,
+  onApproval,
+}: {
+  message: UIMessage;
+  threadId: string;
+  onApproval: (args: {
+    threadId: string;
+    approvalId: string;
+    approved: boolean;
+    reason?: string;
+  }) => Promise<unknown>;
+}) {
+  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 (
+    <div className={cn("flex", isUser ? "justify-end" : "justify-start")}>
+      <div
+        className={cn(
+          "rounded-lg px-4 py-2 max-w-lg whitespace-pre-wrap shadow-sm",
+          isUser ? "bg-indigo-100 text-indigo-900" : "bg-gray-200 text-gray-800",
+          {
+            "bg-green-100 border border-green-300":
+              message.status === "streaming",
+            "bg-red-100 border border-red-300": message.status === "failed",
+          },
+        )}
+      >
+        {/* Status badge */}
+        {message.status !== "success" && message.role !== "user" && (
+          <span
+            className={cn("text-[10px] font-mono uppercase tracking-wide", {
+              "text-green-600": message.status === "streaming",
+              "text-orange-600": message.status === "pending",
+              "text-red-600": message.status === "failed",
+            })}
+          >
+            [{message.status}]
+          </span>
+        )}
+
+        {/* Reasoning */}
+        {reasoningText && (
+          <div className="text-xs text-gray-500 italic mb-1 border-l-2 border-gray-300 pl-2">
+            {reasoningText}
+          </div>
+        )}
+
+        {/* Tool calls with approval UI */}
+        {toolParts.map((tool) => (
+          <ToolCallDisplay
+            key={tool.toolCallId}
+            tool={tool}
+            threadId={threadId}
+            onApproval={onApproval}
+          />
+        ))}
+
+        {/* Main text */}
+        <div>{visibleText || (isUser ? message.text : "...")}</div>
+      </div>
+    </div>
+  );
+}
+
+// ============================================================================
+// 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<unknown>;
+}) {
+  const [denialReason, setDenialReason] = useState("");
+  const [showReasonInput, setShowReasonInput] = useState(false);
+  const toolName = tool.type.replace("tool-", "");
+  const approvalId = getToolApprovalId(tool);
+  const approvalReason = getToolApprovalReason(tool);
+
+  return (
+    <div className="mb-2 p-2 rounded bg-white/50 border border-gray-300 text-sm">
+      <div className="font-mono text-xs text-gray-500 mb-1">
+        {toolName}({JSON.stringify(tool.input)})
+      </div>
+
+      {tool.state === "approval-requested" && approvalId && (
+        <div className="mt-2">
+          <div className="text-amber-700 font-medium mb-2">
+            Approval required
+          </div>
+          {showReasonInput ? (
+            <div className="flex gap-2 items-center mb-2">
+              <input
+                type="text"
+                value={denialReason}
+                onChange={(e) => setDenialReason(e.target.value)}
+                placeholder="Reason for denial..."
+                className="flex-1 px-2 py-1 text-sm rounded border border-gray-300"
+              />
+              <button
+                className="px-3 py-1 rounded bg-red-500 text-white text-sm hover:bg-red-600"
+                onClick={() => {
+                  void onApproval({
+                    threadId,
+                    approvalId,
+                    approved: false,
+                    reason: denialReason || undefined,
+                  });
+                }}
+              >
+                Deny
+              </button>
+              <button
+                className="text-xs text-gray-500 hover:underline"
+                onClick={() => setShowReasonInput(false)}
+              >
+                Cancel
+              </button>
+            </div>
+          ) : (
+            <div className="flex gap-2">
+              <button
+                className="px-3 py-1 rounded bg-green-600 text-white text-sm hover:bg-green-700"
+                onClick={() => {
+                  void onApproval({
+                    threadId,
+                    approvalId,
+                    approved: true,
+                  });
+                }}
+              >
+                Approve
+              </button>
+              <button
+                className="px-3 py-1 rounded bg-red-500 text-white text-sm hover:bg-red-600"
+                onClick={() => setShowReasonInput(true)}
+              >
+                Deny
+              </button>
+            </div>
+          )}
+        </div>
+      )}
+
+      {tool.state === "approval-responded" && (
+        <div className="text-green-700 text-xs">Approved - executing...</div>
+      )}
+
+      {tool.state === "output-denied" && (
+        <div className="text-red-600 text-xs">
+          Denied
+          {approvalReason && `: ${approvalReason}`}
+        </div>
+      )}
+
+      {tool.state === "output-available" && (
+        <div className="text-green-700 text-xs mt-1">
+          Result: {JSON.stringify("output" in tool ? tool.output : undefined)}
+        </div>
+      )}
+
+      {tool.state === "output-error" && (
+        <div className="text-red-600 text-xs mt-1">
+          Error: {"errorText" in tool ? tool.errorText : "Unknown error"}
+        </div>
+      )}
+
+      {(tool.state === "input-available" || tool.state === "input-streaming") && (
+        <div className="text-gray-500 text-xs">Processing...</div>
+      )}
+    </div>
+  );
+}
+
+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 (
+    <div className="p-4">
+      <h2 className="text-sm font-bold text-gray-700 uppercase tracking-wide mb-3">
+        Stream Inspector
+      </h2>
+
+      <StreamSection
+        label="Streaming"
+        streams={streaming}
+        color="green"
+      />
+      <StreamSection
+        label="Finished"
+        streams={finished}
+        color="blue"
+      />
+      <StreamSection
+        label="Aborted"
+        streams={aborted}
+        color="red"
+      />
+
+      {allStreams?.length === 0 && (
+        <p className="text-xs text-gray-400 italic">
+          No streams yet. Send a message to see stream lifecycle.
+        </p>
+      )}
+
+      <div className="mt-6 p-3 bg-gray-50 rounded text-xs text-gray-500 space-y-2">
+        <p className="font-semibold text-gray-700">How it works:</p>
+        <ul className="space-y-1 list-disc list-inside">
+          <li>
+            <strong>Delta Streaming:</strong> Mutation saves prompt, schedules
+            an action. The action streams AI response and saves deltas to the
+            database. Clients subscribe via reactive queries.
+          </li>
+          <li>
+            <strong>HTTP Streaming:</strong> Direct text stream over HTTP.
+            Response chunks are decoded by the browser. No database
+            persistence of intermediate deltas.
+          </li>
+          <li>
+            <strong>One-Shot:</strong> Single action call. Simpler but no
+            optimistic updates.
+          </li>
+          <li>
+            <strong>Abort:</strong> Transitions the stream to "aborted" state.
+            Clients see the partial response with failed status.
+          </li>
+          <li>
+            <strong>Fallback:</strong> When streaming finishes, the full
+            message is saved to the database. The deduplication logic prefers
+            finalized messages over streaming ones.
+          </li>
+        </ul>
+      </div>
+    </div>
+  );
+}
+
+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 (
+    <div className="mb-4">
+      <div className="flex items-center gap-1.5 mb-1.5">
+        <div className={cn("w-2 h-2 rounded-full", dotColors[color])} />
+        <span className="text-xs font-semibold text-gray-600">
+          {label} ({streams.length})
+        </span>
+      </div>
+      <div className="space-y-1.5">
+        {streams.map((s: any) => (
+          <div
+            key={s.streamId}
+            className={cn(
+              "text-[11px] font-mono p-2 rounded border",
+              bgColors[color],
+            )}
+          >
+            <div className="truncate text-gray-500">
+              id: {s.streamId.slice(0, 12)}...
+            </div>
+            <div>
+              order: {s.order}, step: {s.stepOrder}
+            </div>
+            {s.agentName && <div>agent: {s.agentName}</div>}
+            {s.model && <div>model: {s.model}</div>}
+          </div>
+        ))}
+      </div>
+    </div>
+  );
+}
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() {
             <Route path="/weather-fashion" element={<WeatherFashion />} />
             <Route path="/stream-array" element={<StreamArray />} />
             <Route path="/chat-approval" element={<ChatApproval />} />
-
+            <Route path="/streaming-demo" element={<StreamingDemo />} />
           </Routes>
         </main>
         <Toaster />
@@ -154,6 +154,20 @@ function Index() {
               with an optional reason.
             </p>
           </li>
+          <li className="border rounded p-4 hover:shadow transition border-indigo-200 bg-indigo-50">
+            <Link
+              to="/streaming-demo"
+              className="text-xl font-semibold text-indigo-700 hover:underline"
+            >
+              Full Streaming Demo
+            </Link>
+            <p className="mt-2 text-gray-700">
+              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.
+            </p>
+          </li>
         </ul>
         <div className="mt-8 text-sm text-gray-500">
           More examples coming soon!
diff --git a/src/client/http.test.ts b/src/client/http.test.ts
new file mode 100644
index 00000000..d21f057b
--- /dev/null
+++ b/src/client/http.test.ts
@@ -0,0 +1,586 @@
+import { describe, expect, test } from "vitest";
+import {
+  Agent,
+  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<typeof schema>;
+const action = actionGeneric as ActionBuilder<DataModel, "public">;
+
+const model = () =>
+  mockModel({
+    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.
+// ============================================================================
+
+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"),
+    };
+  },
+});
+
+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"),
+    };
+  },
+});
+
+// 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 };
+  },
+});
+
+// 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: {},
+  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;
+    testStreamTextNoDeltas: typeof testStreamTextNoDeltas;
+    testHttpStreamTextWithThread: typeof testHttpStreamTextWithThread;
+    testHttpStreamTextCreatesThread: typeof testHttpStreamTextCreatesThread;
+    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;
+    testAsHttpActionIgnoresUnvalidatedBodyThreadId: typeof testAsHttpActionIgnoresUnvalidatedBodyThreadId;
+    testAsHttpActionRejectsMismatchedUserAndThread: typeof testAsHttpActionRejectsMismatchedUserAndThread;
+    testAsHttpActionHonorsAuthorizedThreadId: typeof testAsHttpActionHonorsAuthorizedThreadId;
+  };
+}>["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);
+  });
+});
+
+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);
+  });
+
+  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("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(
+      testApi.testAsHttpActionHonorsAuthorizedThreadId,
+      {},
+    );
+    expect(result.usedAuthorizedThread).toBe(true);
+  });
+});
diff --git a/src/client/http.ts b/src/client/http.ts
new file mode 100644
index 00000000..743040c9
--- /dev/null
+++ b/src/client/http.ts
@@ -0,0 +1,190 @@
+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";
+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;
+  userId?: string | null;
+  threadId?: string;
+  /**
+   * Whether to save incremental data (deltas) from streaming responses
+   * to the database alongside the HTTP stream. Defaults to false.
+   *
+   * 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). */
+  corsHeaders?: Record<string, string>;
+};
+
+type StreamTextInputArgs<
+  TOOLS extends ToolSet,
+  OUTPUT extends Output<any, any, any>,
+> = AgentPrompt &
+  Omit<
+    Parameters<typeof streamTextAi<TOOLS, OUTPUT>>[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<any, any, any> = never,
+>(
+  ctx: ActionCtx,
+  component: AgentComponent,
+  streamTextArgs: StreamTextInputArgs<TOOLS, OUTPUT>,
+  options: HttpStreamOptions,
+): Promise<Response> {
+  const threadId = await resolveThreadId(ctx, component, options);
+
+  const result = await streamText<TOOLS, OUTPUT>(
+    ctx,
+    component,
+    streamTextArgs,
+    {
+      ...options,
+      threadId,
+      saveStreamDeltas: options.saveStreamDeltas,
+    },
+  );
+
+  const response = result.toTextStreamResponse();
+  applyStreamHeaders(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<any, any, any> = never,
+>(
+  ctx: ActionCtx,
+  component: AgentComponent,
+  streamTextArgs: StreamTextInputArgs<TOOLS, OUTPUT>,
+  options: HttpStreamOptions,
+): Promise<Response> {
+  const threadId = await resolveThreadId(ctx, component, options);
+
+  const result = await streamText<TOOLS, OUTPUT>(
+    ctx,
+    component,
+    streamTextArgs,
+    {
+      ...options,
+      threadId,
+      saveStreamDeltas: options.saveStreamDeltas,
+    },
+  );
+
+  const response = result.toUIMessageStreamResponse();
+  applyStreamHeaders(response, result, options.corsHeaders);
+  return response;
+}
+
+async function resolveThreadId(
+  ctx: ActionCtx,
+  component: AgentComponent,
+  options: HttpStreamOptions,
+): Promise<string> {
+  if (options.threadId) return options.threadId;
+  return createThread(ctx, component, {
+    userId: options.userId ?? null,
+  });
+}
+
+/**
+ * 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<string, string>,
+) {
+  if (result.promptMessageId) {
+    response.headers.set(
+      HTTP_STREAM_MESSAGE_ID_HEADER,
+      result.promptMessageId,
+    );
+  }
+  if (result.streamId) {
+    response.headers.set(HTTP_STREAM_STREAM_ID_HEADER, 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..81bce174 100644
--- a/src/client/index.ts
+++ b/src/client/index.ts
@@ -99,9 +99,22 @@ 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,
   toModelMessage,
@@ -1650,6 +1663,174 @@ export class Agent<
     });
   }
 
+  /**
+   * Build a plain handler `(ctx, request) => Promise<Response>` 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({
+   *     // Save deltas in the background and start the response immediately
+   *     // (otherwise `saveStreamDeltas: true` buffers the full generation
+   *     // before the body opens).
+   *     saveStreamDeltas: { returnImmediately: true },
+   *   })),
+   * });
+   * ```
+   */
+  asHttpAction<DataModel extends GenericDataModel>(
+    spec?: MaybeCustomCtx<CustomCtx, DataModel, AgentTools> & {
+      /**
+       * Whether to save incremental data (deltas) from streaming responses
+       * to the database alongside the HTTP stream. Defaults to false.
+       *
+       * - `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;
+      /**
+       * When to stop generating text.
+       * Defaults to the {@link Agent["options"].stopWhen} option.
+       */
+      stopWhen?: StopCondition<AgentTools> | Array<StopCondition<AgentTools>>;
+      /**
+       * 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<string, string>;
+      /**
+       * 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<DataModel>,
+        request: Request,
+        body: HttpStreamRequestBody,
+      ) => Promise<{ userId?: string; threadId?: string } | void>;
+    } & Options,
+  ): (
+    ctx: GenericActionCtx<DataModel>,
+    request: Request,
+  ) => Promise<Response> {
+    return async (ctx_, request) => {
+      // 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: HttpStreamRequestBody;
+      try {
+        body = (await request.json()) as HttpStreamRequestBody;
+      } catch {
+        return new Response(
+          JSON.stringify({ error: "Invalid JSON in request body" }),
+          {
+            status: 400,
+            headers: { "Content-Type": "application/json" },
+          },
+        );
+      }
+
+      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 = {
+        prompt: body.prompt,
+        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
+          ? {
+              ...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,
+        {
+          // Forward all per-request Options from spec (contextOptions,
+          // storageOptions, usageHandler, contextHandler,
+          // rawRequestResponseHandler) so they actually take effect.
+          ...spec,
+        },
+      );
+
+      const response =
+        spec?.format === "ui-messages"
+          ? result.toUIMessageStreamResponse()
+          : result.toTextStreamResponse();
+      applyStreamHeaders(response, result, spec?.corsHeaders);
+      return response;
+    };
+  }
+
   /**
    * @deprecated Use {@link saveMessages} directly instead.
    */
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<void>;
   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/streamText.ts b/src/client/streamText.ts
index 8f96817c..413b943a 100644
--- a/src/client/streamText.ts
+++ b/src/client/streamText.ts
@@ -83,6 +83,15 @@ export async function streamText<
   // Track the final step for atomic save with stream finish (issue #181)
   let pendingFinalStep: StepResult<TOOLS> | undefined;
 
+  // Whether streamText will await stream consumption before returning.
+  // Only true when there's an actual DeltaStreamer to drain — otherwise
+  // we'd block the response with no persister to drain into.
+  const willAwaitStream =
+    Boolean(threadId) &&
+    (options.saveStreamDeltas === true ||
+      (typeof options.saveStreamDeltas === "object" &&
+        !options.saveStreamDeltas.returnImmediately));
+
   const streamer =
     threadId && options.saveStreamDeltas
       ? new DeltaStreamer(
@@ -111,6 +120,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,
@@ -142,10 +158,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);
       }
@@ -155,11 +182,7 @@ export async function streamText<
   const stream = streamer?.consumeStream(
     result.toUIMessageStream<AIUIMessage<TOOLS>>(),
   );
-  if (
-    (typeof options?.saveStreamDeltas === "object" &&
-      !options.saveStreamDeltas.returnImmediately) ||
-    options?.saveStreamDeltas === true
-  ) {
+  if (willAwaitStream) {
     try {
       await stream;
       await result.consumeStream();
@@ -190,6 +213,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/streaming.ts b/src/client/streaming.ts
index c3146b43..6dd9015d 100644
--- a/src/client/streaming.ts
+++ b/src/client/streaming.ts
@@ -80,6 +80,17 @@ export async function syncStreams(
   }
 }
 
+/**
+ * Abort an in-progress stream.
+ *
+ * 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,
@@ -287,6 +298,13 @@ export class DeltaStreamer<T> {
     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 (
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/component/streams.ts b/src/component/streams.ts
index 4c3499fe..faf878f9 100644
--- a/src/component/streams.ts
+++ b/src/component/streams.ts
@@ -159,6 +159,11 @@ function publicStreamMessage(m: Doc<"streamingMessages">): StreamMessage {
   };
 }
 
+// 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() },
   returns: v.boolean(),
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<Uint8Array> {
+  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<Uint8Array>({
+      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<Uint8Array>({
+      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<Uint8Array>,
+  options: {
+    onChunk: (text: string) => void;
+    signal?: AbortSignal;
+  },
+): Promise<void> {
+  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..4187ded7
--- /dev/null
+++ b/src/react/useHttpStream.ts
@@ -0,0 +1,153 @@
+"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";
+
+/**
+ * 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 <token>`.
+   * e.g. from `useAuthToken()` via `@convex-dev/auth/react`.
+   */
+  token?: string;
+  /** Additional headers to include in the request. */
+  headers?: Record<string, string>;
+}): {
+  text: string;
+  isStreaming: boolean;
+  error: Error | null;
+  streamId: string | null;
+  messageId: string | null;
+  send: (body: {
+    threadId?: string;
+    prompt?: string;
+    [key: string]: unknown;
+  }) => Promise<void>;
+  abort: () => void;
+} {
+  const [text, setText] = useState("");
+  const [isStreaming, setIsStreaming] = useState(false);
+  const [error, setError] = useState<Error | null>(null);
+  const [streamId, setStreamId] = useState<string | null>(null);
+  const [messageId, setMessageId] = useState<string | null>(null);
+  const abortControllerRef = useRef<AbortController | null>(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(
+          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);
+        }
+
+        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 };
+}
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" ||