fix(ai-chat): #381 PR1 review round 1 — open()-gate test + paused-pending byte-cap

Do 1 [test-coverage]: контроллерный тест на флаг-гейт begin-hook open() — при
OFF beginRun зовётся (durable-ран независим), а streamRegistry.open НЕ зовётся
(закрывает регресс: пустая entry → non-null paused attach → зависший SSE вместо
204); при ON — open зовётся с (chatId, runId).

Do 2 [stability]: байт-кап очереди pending paused-подписчика. pendingBytes +
overflowed на Subscriber; в paused-ветке ingestFrame при превышении
SUBSCRIBER_MAX_BUFFERED_BYTES (8MB) подписчик помечается overflowed, pending
чистится, он выбрасывается из entry.subscribers (как overflowed-entry). start()
на overflowed → onEnd (чистый 204-эквивалент, без частичного реплея). Контракт
«start() в том же тике, что attach()» задокументирован в коде — кап это
структурный бэкстоп для phase-2 Redis-await шва. Юнит-тест: paused A + live B,
9×1MB > cap → A выброшен (0 доставок), B получает все 9 живьём, поздний start(A)
→ один onEnd без реплея.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.github/workflows/develop.yml

@@ -151,6 +151,12 @@ jobs:
       - name: Build editor-ext
         run: pnpm --filter @docmost/editor-ext build
 
+      # @docmost/prosemirror-markdown is an ESM workspace package the server
+      # imports at runtime; its build/ is gitignored and test:e2e has no pretest
+      # hook, so build it before the e2e run (mirrors the test.yml job).
+      - name: Build prosemirror-markdown
+        run: pnpm --filter @docmost/prosemirror-markdown build
+
       - name: Run migrations
         run: pnpm --filter ./apps/server migration:latest
 

apps/client/src/features/editor/gitmost/gitmost-global-bridge.tsx

@@ -24,6 +24,7 @@ import {
   GitmostListPagesResult,
   GitmostListSpacesResult,
   gitmostDecodePayloadToFile,
+  gitmostInsertTranscriptIntoEditor,
   gitmostUploadFileToEditor,
 } from "@/features/editor/gitmost/gitmost-recording.ts";
 
@@ -281,6 +282,18 @@ export default function GitmostGlobalBridge() {
             pageId: page.id,
           };
         }
+
+        // Best-effort: append the transcript (heading + one paragraph per line)
+        // below the just-inserted audio node. The audio insert already
+        // succeeded, so a transcript failure must NOT turn this into an error —
+        // wrap it and, on any throw, log and still return ok. A missing/empty/
+        // non-string transcript is a no-op inside the helper (audio only).
+        try {
+          gitmostInsertTranscriptIntoEditor(editor, payload?.transcript);
+        } catch (err) {
+          console.error("[gitmost] transcript insert failed", err);
+        }
+
         return { ok: true, pageId: page.id };
       } catch (err: any) {
         console.error("[gitmost] createPageWithRecording failed", err);

apps/client/src/features/editor/gitmost/gitmost-recording.test.ts

@@ -0,0 +1,150 @@
+import { describe, it, expect } from "vitest";
+import { Editor } from "@tiptap/core";
+import { Document } from "@tiptap/extension-document";
+import { Paragraph } from "@tiptap/extension-paragraph";
+import { Text } from "@tiptap/extension-text";
+import { Heading } from "@tiptap/extension-heading";
+import { Bold } from "@tiptap/extension-bold";
+import { Italic } from "@tiptap/extension-italic";
+import { Link } from "@tiptap/extension-link";
+import { gitmostInsertTranscriptIntoEditor } from "./gitmost-recording.ts";
+
+const ZWSP = "​"; // U+200B, the helper's block-trigger neutralizer
+
+/**
+ * #377 — the web-side bridge must append the native host's transcript below the
+ * recording. These exercise the pure insert helper through a REAL Tiptap editor
+ * (Document/Paragraph/Text/Heading + Bold/Italic/Link marks so an HTML-parsing
+ * regression would be caught), asserting the resulting document rather than
+ * mocking the editor: transcript present -> "Transcript" heading + one paragraph
+ * per non-empty line; content is inserted as LITERAL TEXT (no HTML/markdown
+ * parsing); col-0 markdown block triggers are neutralized so git-sync keeps them
+ * paragraphs; absent/empty/non-string -> no-op.
+ */
+describe("gitmostInsertTranscriptIntoEditor", () => {
+  const makeEditor = () =>
+    new Editor({
+      // Bold/Italic/Link are registered specifically so that IF the helper ever
+      // regressed to inserting an HTML/markdown string (instead of a text node),
+      // TipTap would parse `<b>`/`*..*`/`[..](..)` into marks and the literal-
+      // text assertions below would fail.
+      extensions: [Document, Paragraph, Text, Heading, Bold, Italic, Link],
+      // Start from a single empty paragraph (a fresh page's baseline). The
+      // helper appends at the end of the doc, i.e. below existing content.
+      content: { type: "doc", content: [{ type: "paragraph" }] },
+    });
+
+  it("inserts a Transcript heading + one paragraph per non-empty line, verbatim", () => {
+    const editor = makeEditor();
+
+    const inserted = gitmostInsertTranscriptIntoEditor(
+      editor,
+      "You: hello there\nSpeaker 1: hi\n\nYou: bye",
+    );
+
+    expect(inserted).toBe(true);
+
+    const nodes = (editor.getJSON().content ?? []) as any[];
+    // A level-2 "Transcript" heading is present.
+    const heading = nodes.find((n) => n.type === "heading");
+    expect(heading?.attrs?.level).toBe(2);
+    expect(heading?.content?.[0]?.text).toBe("Transcript");
+
+    // Every non-empty transcript line becomes a paragraph, in order, verbatim;
+    // the blank line between them is dropped.
+    const texts = nodes
+      .filter((n) => n.type === "paragraph")
+      .map((n) => n.content?.[0]?.text)
+      .filter((t) => typeof t === "string");
+    expect(texts).toEqual(["You: hello there", "Speaker 1: hi", "You: bye"]);
+
+    editor.destroy();
+  });
+
+  it("inserts HTML + markdown metacharacters as LITERAL text (no injection / no mark parsing)", () => {
+    const editor = makeEditor();
+    const line =
+      "You: <b>bold</b> <script>alert(1)</script> and *stars* and [link](x)";
+
+    const inserted = gitmostInsertTranscriptIntoEditor(editor, line);
+    expect(inserted).toBe(true);
+
+    const paras = (editor.getJSON().content ?? []).filter(
+      (n: any) => n.type === "paragraph",
+    ) as any[];
+    // The transcript line is exactly ONE paragraph holding a SINGLE text node
+    // whose text is the verbatim string — not split into bold/link/other nodes,
+    // not carrying any marks, not raw HTML. This FAILS if the helper switched to
+    // insertContent(htmlString): TipTap would then parse <b>/[link](x)/*stars*.
+    const content = paras[paras.length - 1].content;
+    expect(content).toHaveLength(1);
+    expect(content[0].type).toBe("text");
+    expect(content[0].marks ?? []).toEqual([]);
+    expect(content[0].text).toBe(line);
+    // And no bold/italic/link mark exists anywhere in the document.
+    const html = editor.getHTML();
+    expect(html).not.toMatch(/<(strong|b|em|i|a)\b/);
+    // The angle brackets survived as escaped entities (literal text), not a live
+    // <script>/<b> element.
+    expect(html).not.toMatch(/<script/i);
+
+    editor.destroy();
+  });
+
+  it("neutralizes col-0 markdown block triggers with a leading ZWSP (git-sync safety)", () => {
+    const editor = makeEditor();
+    // Trigger lines (some with a leaked indent) + a normal prefixed line.
+    const inserted = gitmostInsertTranscriptIntoEditor(
+      editor,
+      [
+        "- dash",
+        "  > quote", // leading indent must be trimmed then neutralized
+        "# hash",
+        "1. one",
+        "> [!info] note",
+        "```js",
+        "---", // solid thematic break -> horizontalRule (text-losing) if unneutralized
+        "***",
+        "___",
+        "You: normal line",
+      ].join("\n"),
+    );
+    expect(inserted).toBe(true);
+
+    const texts = (editor.getJSON().content ?? [])
+      .filter((n: any) => n.type === "paragraph")
+      .map((n: any) => n.content?.[0]?.text)
+      .filter((t: any) => typeof t === "string") as string[];
+
+    // Every block-trigger line is prefixed with the invisible ZWSP (indent
+    // trimmed first); the normal `You:` line is left byte-exact.
+    expect(texts).toEqual([
+      ZWSP + "- dash",
+      ZWSP + "> quote",
+      ZWSP + "# hash",
+      ZWSP + "1. one",
+      ZWSP + "> [!info] note",
+      ZWSP + "```js",
+      ZWSP + "---",
+      ZWSP + "***",
+      ZWSP + "___",
+      "You: normal line",
+    ]);
+
+    editor.destroy();
+  });
+
+  it("is a no-op for undefined / empty / whitespace-only / non-string transcripts", () => {
+    for (const value of [undefined, "", "   \n  \n", 42, {}, null]) {
+      const editor = makeEditor();
+      const before = JSON.stringify(editor.getJSON());
+
+      const inserted = gitmostInsertTranscriptIntoEditor(editor, value as any);
+
+      expect(inserted).toBe(false);
+      // Document is untouched (audio-only behavior preserved).
+      expect(JSON.stringify(editor.getJSON())).toBe(before);
+      editor.destroy();
+    }
+  });
+});

apps/client/src/features/editor/gitmost/gitmost-recording.ts

@@ -65,6 +65,11 @@ export interface GitmostCreatePagePayload {
   base64: string;
   filename: string;
   mimeType: string;
+  // Optional transcript for the recording: plain text, `\n`-separated, each
+  // line already formatted as `You: ...` / `Speaker N: ...` by the native host
+  // (ready to insert, no parsing needed). Omitted (no speech / no models) ->
+  // audio only.
+  transcript?: string;
 }
 
 export interface GitmostCreatePageResult {
@@ -235,6 +240,83 @@ export async function gitmostUploadFileToEditor(
   }
 }
 
+// Zero-width space (U+200B). Prepended to a transcript line that begins with a
+// markdown BLOCK trigger: it is invisible in the rendered doc but shifts the
+// trigger off column 0, so the git-sync doc->markdown->doc round-trip keeps the
+// line a plain paragraph (see GITMOST_MD_BLOCK_TRIGGER_RE).
+const GITMOST_ZWSP = "​";
+
+// A markdown BLOCK-level construct that, sitting at column 0 of a paragraph
+// line, the git-sync markdown serializer (packages/prosemirror-markdown
+// markdown-converter.ts, `case "paragraph"`) would re-parse into a NON-paragraph
+// block on the doc->markdown->doc cycle. That serializer emits paragraph text
+// verbatim with NO block-escape (the pre-existing root cause), so a leading
+// `#`/`-`/`*`/`+`/`>`, an ordered-list `N.`/`N)`, a code fence ```/~~~, a table
+// `|`, or a `> [!info]` callout opener would silently become a heading / list /
+// quote / code block / table / callout. The final alternative matches a WHOLE-
+// LINE thematic break — solid `---`/`***`/`___` or spaced `- - -`/`_ _ _` (3+ of
+// the same `-`/`*`/`_`) — which round-trips into a `horizontalRule`; because
+// that node carries NO text, an un-neutralized separator line would LOSE its
+// text entirely (worse than the list/quote case). This matches a TRIMMED line's
+// start; the transcript's own `You:` / `Speaker N:` prefix begins with a letter
+// and never matches, so prefixed lines are left byte-exact.
+const GITMOST_MD_BLOCK_TRIGGER_RE =
+  /^(?:#{1,6}(?:\s|$)|[-*+](?:\s|$)|>|\d+[.)](?:\s|$)|```|~~~|\||([-*_])(?:\s*\1){2,}\s*$)/;
+
+// Append a transcript block BELOW the recording's audio node in a live editor:
+// a "Transcript" heading followed by one paragraph per non-empty transcript
+// line. The transcript is plain text, `\n`-separated, each line already
+// formatted as `You: ...` / `Speaker N: ...` by the native host — line text is
+// inserted as a TEXT node (never HTML/markdown), so there is no injection or
+// mark-parsing surface. Each kept line is trimmed (drops an indent that would
+// both leak into the display and, at col 0, form a markdown block trigger) and,
+// if it still begins with a col-0 markdown block trigger, gets an invisible
+// zero-width space prepended so the git-sync round-trip cannot turn it into a
+// list/quote/heading/callout/code/table (defensive boundary against the
+// serializer's missing block-escape). This is best-effort and meant to run
+// AFTER the audio has already been inserted; the caller must guard against a
+// throw so a transcript failure never fails the (already successful) recording.
+// Returns true when a block was inserted, false when there was nothing to
+// insert (transcript undefined/empty/not-a-string). A non-string value is a
+// no-op, not an error.
+export function gitmostInsertTranscriptIntoEditor(
+  editor: Editor,
+  transcript: unknown,
+): boolean {
+  if (typeof transcript !== "string") return false;
+  const lines = transcript
+    .split("\n")
+    // Trim each line and drop blank (whitespace-only) ones.
+    .map((line) => line.trim())
+    .filter((line) => line.length > 0)
+    // Neutralize a col-0 markdown block trigger with an invisible ZWSP so the
+    // git-sync round-trip keeps the line a paragraph. Host lines (`You:` /
+    // `Speaker N:`) never match and stay byte-exact.
+    .map((line) =>
+      GITMOST_MD_BLOCK_TRIGGER_RE.test(line) ? GITMOST_ZWSP + line : line,
+    );
+  if (lines.length === 0) return false;
+
+  const content = [
+    {
+      type: "heading",
+      attrs: { level: 2 },
+      content: [{ type: "text", text: "Transcript" }],
+    },
+    ...lines.map((line) => ({
+      type: "paragraph",
+      content: [{ type: "text", text: line }],
+    })),
+  ];
+
+  // Append at the end of the document. On a freshly-created recording page the
+  // audio node is the last block, so the end position places the transcript
+  // directly below it.
+  const endPos = editor.state.doc.content.size;
+  editor.chain().focus().insertContentAt(endPos, content).run();
+  return true;
+}
+
 // Full insert path used by the open-page bridge (insertRecording): guard the
 // editor, validate/decode the payload, then upload. Never throws — resolves to
 // a result code.

apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts

@@ -0,0 +1,329 @@
+import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
+
+/**
+ * In-memory run-stream registry (#184 phase 1.5). A durable agent run tees its
+ * SSE frames here (via `pipeUIMessageStreamToResponse({ consumeSseStream })`)
+ * so a LATE tab — one that reloaded, or opened after the starter dropped — can
+ * attach through `GET /ai-chat/runs/:chatId/stream`, replay the frames buffered
+ * so far, and then follow the live tail as a normal streamer.
+ *
+ * This is deliberately single-process and best-effort: it holds nothing the DB
+ * does not (the run + assistant row are the source of truth), so a process
+ * restart simply drops in-flight entries and the client falls back to its
+ * restore + degraded-poll path. The async `attach` return type is the seam for a
+ * future phase-2 cross-process backend (Redis) — the interface does not change.
+ */
+
+/** How long a finished entry is retained for late attach (replay + immediate end). */
+export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000;
+
+/** Per-run replay buffer cap. Past this the buffer is dropped (attach -> 204). */
+export const RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024;
+
+// 2x the replay cap: a just-written 4MB replay burst alone can never trip the
+// per-subscriber cap (see controller); only a genuinely stalled socket can.
+export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES;
+
+export interface RunStreamCallbacks {
+  onFrame: (frame: string) => void;
+  onEnd: () => void;
+}
+
+export interface RunStreamAttachment {
+  replay: string[];
+  finished: boolean;
+  start(): void; // drain pending frames (order preserved) and go live
+  unsubscribe(): void; // safe to call at any point, idempotent
+}
+
+interface Subscriber extends RunStreamCallbacks {
+  started: boolean;
+  pending: string[];
+  // Byte size of `pending`, capped at SUBSCRIBER_MAX_BUFFERED_BYTES. `start()` is
+  // called in the SAME tick as `attach()` today (see attach), so `pending` never
+  // holds more than one microtask of frames — but the async `attach` signature is
+  // a phase-2 seam: an await between attach and start would let a stalled paused
+  // subscriber buffer the WHOLE run here. The cap is the structural backstop.
+  pendingBytes: number;
+  overflowed: boolean;
+  pendingEnd: boolean;
+}
+
+interface Entry {
+  runId: string;
+  // The persisted assistant row id of this run (set at bind; undefined if the
+  // seed failed). Used by the attach anchor check (invariant 6).
+  assistantMessageId?: string;
+  frames: string[];
+  bytes: number;
+  overflowed: boolean;
+  finished: boolean;
+  subscribers: Set<Subscriber>;
+  retainTimer?: NodeJS.Timeout;
+}
+
+@Injectable()
+export class AiChatStreamRegistryService implements OnModuleDestroy {
+  private readonly logger = new Logger(AiChatStreamRegistryService.name);
+  private readonly entries = new Map<string, Entry>(); // key: chatId
+
+  /**
+   * Register a fresh entry at the START of a run (before any frame), so a tab
+   * that attaches in the begin->seed window finds an entry to wait on. If an
+   * entry already exists for this chat (a previous, possibly still-live run whose
+   * tee loop is draining), it is terminated MIRRORING the done-path (invariant 3)
+   * so its subscribers are released and its retention timer is cleared; a late
+   * `done` from that old tee then fires against the closed-over old reference and,
+   * thanks to identity checks, never touches this new entry.
+   */
+  open(chatId: string, runId: string): void {
+    const existing = this.entries.get(chatId);
+    if (existing) {
+      if (existing.retainTimer) {
+        clearTimeout(existing.retainTimer);
+        existing.retainTimer = undefined;
+      }
+      // Started subscribers get exactly one onEnd() and are removed; paused ones
+      // are marked pendingEnd (their start() will end them). finished=true guards
+      // any later done from the old tee loop from double-notifying.
+      this.terminateSubscribers(existing);
+    }
+    this.entries.set(chatId, {
+      runId,
+      frames: [],
+      bytes: 0,
+      overflowed: false,
+      finished: false,
+      subscribers: new Set<Subscriber>(),
+    });
+  }
+
+  /**
+   * Tee a run's SSE frame stream into its entry (called from consumeSseStream).
+   * No-op with a warning when there is no entry or the entry belongs to a
+   * different run (invariant 1). The reader loop is fire-and-forget: the tee
+   * branch outlives the client socket by design.
+   */
+  bind(
+    chatId: string,
+    runId: string,
+    assistantMessageId: string | undefined,
+    stream: ReadableStream<string>,
+  ): void {
+    const entry = this.entries.get(chatId);
+    if (!entry || entry.runId !== runId) {
+      // Invariant 1: only the matching run may mutate the entry.
+      this.logger.warn(
+        `bind: no matching run-stream entry for chat=${chatId} run=${runId}`,
+      );
+      return;
+    }
+    entry.assistantMessageId = assistantMessageId;
+    const reader = stream.getReader();
+    const pump = async (): Promise<void> => {
+      try {
+        for (;;) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          this.ingestFrame(entry, value);
+        }
+        this.finalizeEntry(chatId, entry);
+      } catch {
+        // A read error is a terminal event too — release subscribers.
+        this.finalizeEntry(chatId, entry);
+      }
+    };
+    void pump();
+  }
+
+  /**
+   * Terminate a run's entry from the OUTER catch of the stream method (a failure
+   * before/while wiring the pipe, so `done` will never arrive). Identity-checked
+   * on runId (invariant 1); the shared terminal path is idempotent.
+   */
+  abortEntry(chatId: string, runId: string): void {
+    const entry = this.entries.get(chatId);
+    if (!entry || entry.runId !== runId) return;
+    this.finalizeEntry(chatId, entry);
+  }
+
+  /**
+   * Attach to a run's stream. Async only for the phase-2 Redis seam — the body
+   * runs synchronously so the replay snapshot and the subscriber registration
+   * happen in ONE tick with no await between them (invariant 4): a frame ingested
+   * concurrently cannot slip into the gap and be lost or duplicated.
+   *
+   * Returns null (-> the caller answers 204) when:
+   *  - there is no entry, or it overflowed (replay is gone);
+   *  - expect=live with an anchor that does not match this run's assistant id
+   *    (invariant 6: a stripped tab must never replay a FOREIGN run's transcript);
+   *  - the run finished and the caller did not expect a live tail.
+   * A finished run with expect=live yields a replay-only attachment (no
+   * subscriber registered). Otherwise a paused subscriber is registered and the
+   * caller replays `replay`, then calls start() to drain and go live.
+   */
+  async attach(
+    chatId: string,
+    expectLive: boolean,
+    anchor: string | undefined,
+    cb: RunStreamCallbacks,
+  ): Promise<RunStreamAttachment | null> {
+    const entry = this.entries.get(chatId);
+    if (!entry || entry.overflowed) return null;
+    // Invariant 6: cross-run replay is forbidden. Before bind, assistantMessageId
+    // is undefined and mismatches any anchor -> 204 -> client restore+poll path.
+    if (expectLive && anchor && entry.assistantMessageId !== anchor) return null;
+    if (entry.finished && !expectLive) return null;
+    if (entry.finished && expectLive) {
+      // Replay-only: the run is done, no subscriber is registered.
+      return {
+        replay: entry.frames.slice(),
+        finished: true,
+        start: () => undefined,
+        unsubscribe: () => undefined,
+      };
+    }
+
+    const sub: Subscriber = {
+      onFrame: cb.onFrame,
+      onEnd: cb.onEnd,
+      started: false,
+      pending: [],
+      pendingBytes: 0,
+      overflowed: false,
+      pendingEnd: false,
+    };
+    entry.subscribers.add(sub);
+    // Snapshot in the SAME synchronous block as the registration (invariant 4).
+    const replay = entry.frames.slice();
+    // CONTRACT: the caller MUST call start() in the SAME tick as this attach()
+    // returns — no await between them. While a subscriber is paused, every frame
+    // is buffered in sub.pending; a delayed start() lets a whole run accumulate
+    // there. The pendingBytes cap (see ingestFrame) is the structural backstop if
+    // that contract is ever broken (e.g. the phase-2 Redis await seam).
+    return {
+      replay,
+      finished: false,
+      start: () => {
+        if (sub.overflowed) {
+          // The pending buffer overflowed while paused: end the stream instead of
+          // replaying a partial (a 204-equivalent post-attach degrade).
+          try {
+            sub.onEnd();
+          } catch {
+            // The socket is gone; nothing to end.
+          }
+          entry.subscribers.delete(sub);
+          return;
+        }
+        // Deliver frames buffered while paused, in order, then go live.
+        for (const frame of sub.pending) {
+          try {
+            sub.onFrame(frame);
+          } catch {
+            entry.subscribers.delete(sub);
+            return;
+          }
+        }
+        sub.pending = [];
+        sub.started = true;
+        if (sub.pendingEnd) {
+          try {
+            sub.onEnd();
+          } catch {
+            // The socket is gone; nothing to end.
+          }
+          entry.subscribers.delete(sub);
+        }
+      },
+      unsubscribe: () => {
+        entry.subscribers.delete(sub);
+      },
+    };
+  }
+
+  onModuleDestroy(): void {
+    for (const entry of this.entries.values()) {
+      if (entry.retainTimer) clearTimeout(entry.retainTimer);
+    }
+    this.entries.clear();
+  }
+
+  /** Buffer + fan-out a single frame. See invariant/overflow semantics inline. */
+  private ingestFrame(entry: Entry, frame: string): void {
+    entry.bytes += Buffer.byteLength(frame);
+    if (!entry.overflowed) {
+      entry.frames.push(frame);
+      if (entry.bytes > RUN_STREAM_MAX_BUFFER_BYTES) {
+        // The crossing frame was already counted AND (below) fanned out; only the
+        // replay buffer is dropped. After overflow no more frames are buffered,
+        // but live fan-out continues.
+        entry.overflowed = true;
+        entry.frames = [];
+        this.logger.warn(
+          `run-stream buffer overflow for run=${entry.runId}; ` +
+            `late attach will 204 until the run ends`,
+        );
+      }
+    }
+    for (const sub of entry.subscribers) {
+      if (sub.started) {
+        try {
+          sub.onFrame(frame);
+        } catch {
+          entry.subscribers.delete(sub);
+        }
+      } else {
+        sub.pending.push(frame);
+        sub.pendingBytes += Buffer.byteLength(frame);
+        if (sub.pendingBytes > SUBSCRIBER_MAX_BUFFERED_BYTES) {
+          // The paused subscriber's buffer overflowed — only possible if start()
+          // was delayed past the same-tick contract (the phase-2 await seam).
+          // Drop it rather than buffer the whole run; on start() it degrades to an
+          // immediate end (a 204-equivalent) instead of replaying a partial.
+          sub.overflowed = true;
+          sub.pending = [];
+          entry.subscribers.delete(sub);
+        }
+      }
+    }
+  }
+
+  /**
+   * Shared terminal path for done / read-error / external-abort. Idempotent: a
+   * second call (already finished) is a no-op, so an open()-replaced or
+   * abort-then-done entry is never double-armed or double-ended.
+   */
+  private finalizeEntry(chatId: string, entry: Entry): void {
+    if (entry.finished) return;
+    this.terminateSubscribers(entry);
+    const timer = setTimeout(() => {
+      // Invariant 2: only delete OUR entry (a replacement may already own the key).
+      if (this.entries.get(chatId) === entry) this.entries.delete(chatId);
+    }, RUN_STREAM_RETAIN_FINISHED_MS);
+    timer.unref?.();
+    entry.retainTimer = timer;
+  }
+
+  /**
+   * Mark the entry finished and release its subscribers, mirroring the done-path:
+   * started subscribers get exactly one onEnd() and are removed; paused ones are
+   * flagged pendingEnd so their start() ends them. Deleting the current element
+   * during Set iteration is safe.
+   */
+  private terminateSubscribers(entry: Entry): void {
+    entry.finished = true;
+    for (const sub of entry.subscribers) {
+      if (sub.started) {
+        try {
+          sub.onEnd();
+        } catch {
+          // The socket is gone; nothing to end.
+        }
+        entry.subscribers.delete(sub);
+      } else {
+        sub.pendingEnd = true;
+      }
+    }
+  }
+}

apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts

@@ -0,0 +1,388 @@
+import {
+  AiChatStreamRegistryService,
+  RUN_STREAM_MAX_BUFFER_BYTES,
+  RUN_STREAM_RETAIN_FINISHED_MS,
+  RunStreamCallbacks,
+} from './ai-chat-stream-registry.service';
+
+/**
+ * Unit tests for the in-memory run-stream registry (#184 phase 1.5). The registry
+ * is the whole of the resumable-transport contract: replay ordering, paused ->
+ * live hand-off, overflow, retention, the anchor check (invariant 6), and the
+ * mirror-the-done-path replace semantics (invariant 3). Every enumerated case in
+ * the issue's task 1.5 has a test here.
+ */
+
+// A ReadableStream whose frames the test pushes explicitly, plus close/error.
+function makePushStream(): {
+  stream: ReadableStream<string>;
+  push: (f: string) => void;
+  close: () => void;
+  error: (e?: unknown) => void;
+} {
+  let controller!: ReadableStreamDefaultController<string>;
+  const stream = new ReadableStream<string>({
+    start(c) {
+      controller = c;
+    },
+  });
+  return {
+    stream,
+    push: (f) => controller.enqueue(f),
+    close: () => controller.close(),
+    error: (e) => controller.error(e ?? new Error('read error')),
+  };
+}
+
+// Let the fire-and-forget pump drain queued frames (reader.read() resolves on a
+// macrotask boundary for an already-enqueued value).
+const flush = (): Promise<void> => new Promise((r) => setTimeout(r, 0));
+
+function collector(): {
+  cb: RunStreamCallbacks;
+  frames: string[];
+  ended: () => number;
+} {
+  const frames: string[] = [];
+  let ends = 0;
+  return {
+    frames,
+    ended: () => ends,
+    cb: {
+      onFrame: (f) => frames.push(f),
+      onEnd: () => {
+        ends += 1;
+      },
+    },
+  };
+}
+
+describe('AiChatStreamRegistryService', () => {
+  const CHAT = 'chat-1';
+  let registry: AiChatStreamRegistryService;
+
+  beforeEach(() => {
+    registry = new AiChatStreamRegistryService();
+    jest.spyOn((registry as any).logger, 'warn').mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    registry.onModuleDestroy();
+  });
+
+  it('replays frames in arrival order (live attach)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    src.push('b');
+    src.push('c');
+    await flush();
+
+    const c = collector();
+    const att = await registry.attach(CHAT, false, undefined, c.cb);
+    expect(att).not.toBeNull();
+    expect(att!.replay).toEqual(['a', 'b', 'c']);
+    expect(att!.finished).toBe(false);
+  });
+
+  it('late attach gets the full prefix as replay plus the live tail', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    src.push('b');
+    await flush();
+
+    const c = collector();
+    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
+    expect(att.replay).toEqual(['a', 'b']);
+    att.start();
+    // Live tail arrives after start().
+    src.push('c');
+    src.push('d');
+    await flush();
+    expect(c.frames).toEqual(['c', 'd']);
+  });
+
+  it('a paused subscriber receives frames buffered during pause in order, then live (no loss/reorder)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    await flush();
+
+    const c = collector();
+    // Attach (paused). Frames that arrive BEFORE start() must queue, not drop.
+    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
+    expect(att.replay).toEqual(['a']);
+    src.push('b'); // arrives while paused -> pending
+    src.push('c');
+    await flush();
+    expect(c.frames).toEqual([]); // nothing delivered yet (paused)
+    att.start(); // drains pending in order
+    expect(c.frames).toEqual(['b', 'c']);
+    src.push('d'); // now live
+    await flush();
+    expect(c.frames).toEqual(['b', 'c', 'd']);
+  });
+
+  it('a run that finishes while a subscriber is paused ends it on start()', async () => {
+    registry.open(CHAT, 'run-1');
+    const c = collector();
+    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
+    // Terminate the run while the subscriber is still paused.
+    registry.abortEntry(CHAT, 'run-1');
+    expect(c.ended()).toBe(0); // paused: not ended yet
+    att.start();
+    expect(c.ended()).toBe(1); // start() drains + ends
+  });
+
+  it('finished + expect=live returns a replay WITHOUT registering a subscriber', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    src.push('b');
+    src.close();
+    await flush();
+
+    const c = collector();
+    const att = (await registry.attach(CHAT, true, undefined, c.cb))!;
+    expect(att.finished).toBe(true);
+    expect(att.replay).toEqual(['a', 'b']);
+    // No subscriber registered: start()/unsubscribe are no-ops and the entry has
+    // zero subscribers.
+    const entry = (registry as any).entries.get(CHAT);
+    expect(entry.subscribers.size).toBe(0);
+    att.start();
+    expect(c.frames).toEqual([]);
+  });
+
+  it('finished WITHOUT expect=live returns null', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    src.close();
+    await flush();
+
+    const c = collector();
+    expect(await registry.attach(CHAT, false, undefined, c.cb)).toBeNull();
+  });
+
+  it('anchor mismatch with expect=live returns null (and null before bind sets assistantMessageId)', async () => {
+    registry.open(CHAT, 'run-1');
+    const c = collector();
+    // Before bind: assistantMessageId is undefined -> mismatches any anchor.
+    expect(
+      await registry.attach(CHAT, true, 'assist-1', c.cb),
+    ).toBeNull();
+
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    await flush();
+    // Wrong anchor -> null (cross-run replay forbidden, invariant 6).
+    expect(await registry.attach(CHAT, true, 'other-id', c.cb)).toBeNull();
+  });
+
+  it('matching anchor with expect=live attaches', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    await flush();
+
+    const c = collector();
+    const att = await registry.attach(CHAT, true, 'assist-1', c.cb);
+    expect(att).not.toBeNull();
+    expect(att!.replay).toEqual(['a']);
+  });
+
+  it('overflow: attach returns null, but the LIVE subscriber keeps receiving (incl. the crossing frame)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+
+    // A live (started) subscriber attached before the flood.
+    const c = collector();
+    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
+    att.start();
+
+    const oneMb = 'x'.repeat(1024 * 1024);
+    // 5 x 1MB = 5MB > 4MB cap; the 5th frame is the one that crosses.
+    for (let i = 0; i < 5; i++) src.push(oneMb + i);
+    await flush();
+
+    const entry = (registry as any).entries.get(CHAT);
+    expect(entry.overflowed).toBe(true);
+    expect(entry.bytes).toBeGreaterThan(RUN_STREAM_MAX_BUFFER_BYTES);
+    // The live subscriber received ALL 5 frames, including the crossing one.
+    expect(c.frames).toHaveLength(5);
+    expect(c.frames[4]).toBe(oneMb + 4);
+
+    // A NEW attach after overflow gets null (replay buffer is gone).
+    const c2 = collector();
+    expect(await registry.attach(CHAT, false, undefined, c2.cb)).toBeNull();
+  });
+
+  it('a paused subscriber whose pending buffer overflows is dropped and ends on start(); other subscribers keep receiving', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+
+    // A: paused (start() deliberately delayed to simulate the phase-2 await seam).
+    const a = collector();
+    const attA = (await registry.attach(CHAT, false, undefined, a.cb))!;
+    // B: live (started) — its delivery must be unaffected by A's overflow.
+    const b = collector();
+    const attB = (await registry.attach(CHAT, false, undefined, b.cb))!;
+    attB.start();
+
+    const oneMb = 'x'.repeat(1024 * 1024);
+    // 9 x 1MB = 9MB > 8MB per-subscriber cap; A's pending overflows, B streams live.
+    for (let i = 0; i < 9; i++) src.push(oneMb + i);
+    await flush();
+
+    const entry = (registry as any).entries.get(CHAT);
+    // A was dropped from the subscriber set on overflow; B (started) remains.
+    expect(entry.subscribers.size).toBe(1);
+    expect(a.frames).toEqual([]); // paused + overflowed: nothing was delivered
+    // B received every frame live (delivery unaffected by A's overflow).
+    expect(b.frames).toHaveLength(9);
+
+    // A's start() (arriving late) degrades to an immediate end, not a partial replay.
+    attA.start();
+    expect(a.frames).toEqual([]);
+    expect(a.ended()).toBe(1);
+  });
+
+  it('open() over a LIVE entry ends started subscribers exactly once and a late done does not touch the new entry (invariant 3)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+    src.push('a');
+    await flush();
+
+    const c = collector();
+    const att = (await registry.attach(CHAT, false, undefined, c.cb))!;
+    att.start(); // started subscriber on run-1
+
+    // run-2 starts on the same chat while run-1's tee is still reading.
+    registry.open(CHAT, 'run-2');
+    expect(c.ended()).toBe(1); // exactly one onEnd from the replace
+
+    const newEntry = (registry as any).entries.get(CHAT);
+    expect(newEntry.runId).toBe('run-2');
+    expect(newEntry.finished).toBe(false);
+
+    // The old tee now completes: its late done must NOT double-end nor delete the
+    // new entry.
+    src.push('b');
+    src.close();
+    await flush();
+    expect(c.ended()).toBe(1); // still exactly one
+    const still = (registry as any).entries.get(CHAT);
+    expect(still).toBe(newEntry);
+    expect(still.runId).toBe('run-2');
+  });
+
+  it('bind with a foreign runId is a no-op (invariant 1)', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'WRONG-run', 'assist-x', src.stream);
+    src.push('a');
+    await flush();
+    const entry = (registry as any).entries.get(CHAT);
+    // Frames were NOT ingested (bind bailed), assistantMessageId untouched.
+    expect(entry.frames).toEqual([]);
+    expect(entry.assistantMessageId).toBeUndefined();
+  });
+
+  it('abortEntry with a foreign runId is a no-op (invariant 1)', async () => {
+    registry.open(CHAT, 'run-1');
+    registry.abortEntry(CHAT, 'WRONG-run');
+    const entry = (registry as any).entries.get(CHAT);
+    expect(entry.finished).toBe(false);
+  });
+
+  it('a throwing onFrame ejects only that subscriber; the ingest loop stays alive', async () => {
+    registry.open(CHAT, 'run-1');
+    const src = makePushStream();
+    registry.bind(CHAT, 'run-1', 'assist-1', src.stream);
+
+    const bad = collector();
+    const badAtt = (await registry.attach(CHAT, false, undefined, {
+      onFrame: () => {
+        throw new Error('boom');
+      },
+      onEnd: bad.cb.onEnd,
+    }))!;
+    badAtt.start();
+
+    const good = collector();
+    const goodAtt = (await registry.attach(CHAT, false, undefined, good.cb))!;
+    goodAtt.start();
+
+    src.push('a'); // bad throws on this frame -> ejected
+    src.push('b'); // good still receives both
+    await flush();
+
+    const entry = (registry as any).entries.get(CHAT);
+    expect(entry.subscribers.size).toBe(1); // bad ejected, good remains
+    expect(good.frames).toEqual(['a', 'b']);
+  });
+});
+
+/**
+ * Retention + replace timer behavior. Fake timers, and entries are finalized via
+ * the synchronous abortEntry() path so no stream pump / microtask juggling is
+ * needed.
+ */
+describe('AiChatStreamRegistryService retention timers', () => {
+  const CHAT = 'chat-r';
+  let registry: AiChatStreamRegistryService;
+
+  beforeEach(() => {
+    jest.useFakeTimers();
+    registry = new AiChatStreamRegistryService();
+    jest.spyOn((registry as any).logger, 'warn').mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    registry.onModuleDestroy();
+    jest.useRealTimers();
+  });
+
+  it('a finished entry is removed after the retention window', () => {
+    registry.open(CHAT, 'run-1');
+    registry.abortEntry(CHAT, 'run-1'); // finalize -> retention armed
+    expect((registry as any).entries.get(CHAT)).toBeDefined();
+    jest.advanceTimersByTime(RUN_STREAM_RETAIN_FINISHED_MS + 1);
+    expect((registry as any).entries.get(CHAT)).toBeUndefined();
+  });
+
+  it('retention deletes ONLY its own entry (invariant 2)', () => {
+    registry.open(CHAT, 'run-1');
+    registry.abortEntry(CHAT, 'run-1'); // arm retention for entry A
+    // Simulate the race where the key was replaced without clearing A's timer.
+    const sentinel = { marker: true };
+    (registry as any).entries.set(CHAT, sentinel);
+    jest.advanceTimersByTime(RUN_STREAM_RETAIN_FINISHED_MS + 1);
+    // A's timer saw entries.get(CHAT) !== A, so it did NOT delete the successor.
+    expect((registry as any).entries.get(CHAT)).toBe(sentinel);
+  });
+
+  it('open() over a retained entry clears its timer and the successor survives', () => {
+    registry.open(CHAT, 'run-1');
+    registry.abortEntry(CHAT, 'run-1'); // retained, timer armed
+    const clearSpy = jest.spyOn(global, 'clearTimeout');
+    registry.open(CHAT, 'run-2'); // must clear run-1's retain timer
+    expect(clearSpy).toHaveBeenCalled();
+    jest.advanceTimersByTime(RUN_STREAM_RETAIN_FINISHED_MS + 1);
+    const entry = (registry as any).entries.get(CHAT);
+    expect(entry).toBeDefined();
+    expect(entry.runId).toBe('run-2');
+  });
+});

apps/server/src/core/ai-chat/ai-chat.controller.attach.spec.ts

@@ -0,0 +1,423 @@
+import { ForbiddenException } from '@nestjs/common';
+import { AiChatController } from './ai-chat.controller';
+import type {
+  RunStreamAttachment,
+  RunStreamCallbacks,
+} from './ai-chat-stream-registry.service';
+import { SUBSCRIBER_MAX_BUFFERED_BYTES } from './ai-chat-stream-registry.service';
+import type { User, Workspace } from '@docmost/db/types/entity.types';
+
+/**
+ * Wiring spec for the #184 phase 1.5 attach endpoint
+ * (`GET /ai-chat/runs/:chatId/stream`). Owner-gated via assertOwnedChat; the
+ * registry is mocked so this exercises ONLY the controller's replay/live/204/
+ * cleanup wiring against a fake raw socket. Constructor order is (aiChatService,
+ * aiChatRunService, aiChatRepo, aiChatMessageRepo, aiTranscription, pageRepo,
+ * streamRegistry, environment).
+ */
+describe('AiChatController attach endpoint (#184 phase 1.5)', () => {
+  const user = { id: 'u1' } as User;
+  const workspace = { id: 'ws1' } as Workspace;
+
+  function makeRawRes() {
+    const raw: any = {
+      writableEnded: false,
+      writableLength: 0,
+      destroyed: false,
+      written: [] as string[],
+      head: null as any,
+      write: jest.fn((f: string) => {
+        raw.written.push(f);
+        return true;
+      }),
+      writeHead: jest.fn((code: number, headers: any) => {
+        raw.head = { code, headers };
+        return raw;
+      }),
+      flushHeaders: jest.fn(),
+      end: jest.fn(() => {
+        raw.writableEnded = true;
+      }),
+      destroy: jest.fn(() => {
+        raw.destroyed = true;
+      }),
+      on: jest.fn(),
+      once: jest.fn(),
+    };
+    const res: any = {
+      raw,
+      status: jest.fn(() => res),
+      send: jest.fn(),
+      hijack: jest.fn(),
+    };
+    return { res, raw };
+  }
+
+  function makeReq(destroyed = false) {
+    const handlers: Record<string, () => void> = {};
+    const raw: any = {
+      destroyed,
+      once: jest.fn((ev: string, fn: () => void) => {
+        handlers[ev] = fn;
+      }),
+    };
+    return { req: { raw } as any, raw, fireClose: () => handlers['close']?.() };
+  }
+
+  function makeAttachment(
+    over: Partial<RunStreamAttachment> = {},
+  ): RunStreamAttachment {
+    return {
+      replay: [],
+      finished: false,
+      start: jest.fn(),
+      unsubscribe: jest.fn(),
+      ...over,
+    };
+  }
+
+  function makeController(opts: {
+    chat?: unknown;
+    attachment?: RunStreamAttachment | null;
+  }) {
+    const aiChatRepo = { findById: jest.fn().mockResolvedValue(opts.chat) };
+    let capturedCb: RunStreamCallbacks | undefined;
+    const streamRegistry = {
+      attach: jest.fn(
+        (
+          _chatId: string,
+          _live: boolean,
+          _anchor: string | undefined,
+          cb: RunStreamCallbacks,
+        ) => {
+          capturedCb = cb;
+          return Promise.resolve(
+            opts.attachment === undefined ? makeAttachment() : opts.attachment,
+          );
+        },
+      ),
+    };
+    const environment = { isAiChatResumableStreamEnabled: () => true };
+    const controller = new AiChatController(
+      {} as never, // aiChatService
+      {} as never, // aiChatRunService
+      aiChatRepo as never,
+      {} as never, // aiChatMessageRepo
+      {} as never, // aiTranscription
+      {} as never, // pageRepo
+      streamRegistry as never,
+      environment as never,
+    );
+    return {
+      controller,
+      aiChatRepo,
+      streamRegistry,
+      getCb: () => capturedCb!,
+    };
+  }
+
+  const owned = { id: 'c1', creatorId: 'u1' };
+
+  it('owner-gates: a foreign chat throws ForbiddenException and never attaches', async () => {
+    const { controller, streamRegistry } = makeController({
+      chat: { id: 'c1', creatorId: 'someone-else' },
+    });
+    const { res } = makeRawRes();
+    const { req } = makeReq();
+    await expect(
+      controller.attachRunStream(
+        'c1',
+        undefined,
+        undefined,
+        req,
+        res,
+        user,
+        workspace,
+      ),
+    ).rejects.toBeInstanceOf(ForbiddenException);
+    expect(streamRegistry.attach).not.toHaveBeenCalled();
+  });
+
+  it('answers 204 when the registry has nothing to resume (no entry / finished / anchor-mismatch)', async () => {
+    const { controller } = makeController({ chat: owned, attachment: null });
+    const { res } = makeRawRes();
+    const { req } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      undefined,
+      undefined,
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(res.status).toHaveBeenCalledWith(204);
+    expect(res.send).toHaveBeenCalled();
+    expect(res.hijack).not.toHaveBeenCalled();
+  });
+
+  it('threads expect=live and anchor through to the registry', async () => {
+    const { controller, streamRegistry } = makeController({
+      chat: owned,
+      attachment: null,
+    });
+    const { res } = makeRawRes();
+    const { req } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      'live',
+      'anchor-1',
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(streamRegistry.attach).toHaveBeenCalledWith(
+      'c1',
+      true,
+      'anchor-1',
+      expect.anything(),
+    );
+  });
+
+  it('passes expect=false when the query is absent', async () => {
+    const { controller, streamRegistry } = makeController({
+      chat: owned,
+      attachment: null,
+    });
+    const { res } = makeRawRes();
+    const { req } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      undefined,
+      undefined,
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(streamRegistry.attach).toHaveBeenCalledWith(
+      'c1',
+      false,
+      undefined,
+      expect.anything(),
+    );
+  });
+
+  it('hijacks, writes headers + replay, registers a close cleanup, then goes live', async () => {
+    const start = jest.fn();
+    const attachment = makeAttachment({
+      replay: ['f1', 'f2'],
+      finished: false,
+      start,
+    });
+    const { controller } = makeController({ chat: owned, attachment });
+    const { res, raw } = makeRawRes();
+    const { req } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      undefined,
+      undefined,
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(res.hijack).toHaveBeenCalled();
+    expect(raw.writeHead).toHaveBeenCalledWith(
+      200,
+      expect.objectContaining({ 'content-type': 'text/event-stream' }),
+    );
+    expect(raw.written).toEqual(['f1', 'f2']); // replay
+    expect(start).toHaveBeenCalled(); // go live after replay
+    expect(req.raw.once).toHaveBeenCalledWith('close', expect.any(Function));
+  });
+
+  it('finished replay ends the response immediately without going live', async () => {
+    const start = jest.fn();
+    const attachment = makeAttachment({
+      replay: ['f1'],
+      finished: true,
+      start,
+    });
+    const { controller } = makeController({ chat: owned, attachment });
+    const { res, raw } = makeRawRes();
+    const { req } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      'live',
+      'a1',
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(raw.written).toEqual(['f1']);
+    expect(raw.end).toHaveBeenCalled();
+    expect(start).not.toHaveBeenCalled(); // finished -> returns before start()
+  });
+
+  it('a close during the awaits (req.raw.destroyed) unsubscribes and writes nothing', async () => {
+    const attachment = makeAttachment({ replay: ['f1'] });
+    const { controller } = makeController({ chat: owned, attachment });
+    const { res, raw } = makeRawRes();
+    const { req } = makeReq(true); // destroyed already at registration time
+    await controller.attachRunStream(
+      'c1',
+      undefined,
+      undefined,
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(attachment.unsubscribe).toHaveBeenCalled();
+    expect(raw.writeHead).not.toHaveBeenCalled();
+    expect(raw.written).toEqual([]);
+  });
+
+  it('the registered close handler unsubscribes the attachment', async () => {
+    const attachment = makeAttachment({ replay: [] });
+    const { controller } = makeController({ chat: owned, attachment });
+    const { res } = makeRawRes();
+    const { req, fireClose } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      undefined,
+      undefined,
+      req,
+      res,
+      user,
+      workspace,
+    );
+    expect(attachment.unsubscribe).not.toHaveBeenCalled();
+    fireClose(); // socket closed
+    expect(attachment.unsubscribe).toHaveBeenCalled();
+  });
+
+  it('onFrame destroys the socket when the buffered length exceeds the cap', async () => {
+    const attachment = makeAttachment({ replay: [] });
+    const { controller, getCb } = makeController({ chat: owned, attachment });
+    const { res, raw } = makeRawRes();
+    const { req } = makeReq();
+    await controller.attachRunStream(
+      'c1',
+      undefined,
+      undefined,
+      req,
+      res,
+      user,
+      workspace,
+    );
+    const cb = getCb();
+    // Normal live frame writes.
+    raw.writableLength = 0;
+    cb.onFrame('live-1');
+    expect(raw.written).toContain('live-1');
+    // A stalled socket over the cap is destroyed instead of buffering.
+    raw.writableLength = SUBSCRIBER_MAX_BUFFERED_BYTES + 1;
+    raw.write.mockClear();
+    cb.onFrame('too-much');
+    expect(raw.destroy).toHaveBeenCalled();
+    expect(raw.write).not.toHaveBeenCalled();
+  });
+});
+
+/**
+ * The begin-hook `open()` flag gate (#184 phase 1.5). `open()` lives ONLY in the
+ * stream() begin-hook, gated on the resumable flag. If it regressed, a flag-off
+ * turn would create an EMPTY registry entry (never bound, never finished) and a
+ * later attach would find a non-null paused attachment -> a hung SSE that never
+ * gets a frame and never ends, instead of a clean 204. These drive stream() only
+ * far enough to capture the runHooks it hands to the service, then invoke the
+ * begin-hook and assert whether the registry was opened.
+ */
+describe('AiChatController begin-hook open() flag gate (#184 phase 1.5)', () => {
+  const user = { id: 'u1' } as User;
+  const workspace = {
+    id: 'ws1',
+    settings: { ai: { chat: true, autonomousRuns: true } },
+  } as unknown as Workspace;
+
+  function makeController(opts: { resumable: boolean }) {
+    let capturedArgs: any;
+    const aiChatService = {
+      resolveRoleForRequest: jest.fn(async () => null),
+      getChatModel: jest.fn(async () => ({})),
+      stream: jest.fn(async (args: any) => {
+        capturedArgs = args;
+      }),
+    };
+    const aiChatRunService = {
+      getActiveForChat: jest.fn(async () => undefined),
+      beginRun: jest.fn(async () => ({
+        runId: 'run-1',
+        signal: new AbortController().signal,
+      })),
+    };
+    const streamRegistry = { open: jest.fn(), attach: jest.fn() };
+    const environment = {
+      isAiChatResumableStreamEnabled: () => opts.resumable,
+    };
+    const controller = new AiChatController(
+      aiChatService as never,
+      aiChatRunService as never,
+      {} as never, // aiChatRepo
+      {} as never, // aiChatMessageRepo
+      {} as never, // aiTranscription
+      {} as never, // pageRepo
+      streamRegistry as never,
+      environment as never,
+    );
+    return {
+      controller,
+      streamRegistry,
+      aiChatRunService,
+      getRunHooks: () => capturedArgs?.runHooks,
+    };
+  }
+
+  function makeReqRes() {
+    const req: any = {
+      raw: { sessionId: 'sess-1', once: jest.fn() },
+      body: {
+        messages: [
+          { id: 'm1', role: 'user', parts: [{ type: 'text', text: 'hi' }] },
+        ],
+      },
+    };
+    const res: any = {
+      raw: { once: jest.fn(), on: jest.fn(), headersSent: false, writableEnded: false },
+      hijack: jest.fn(),
+    };
+    return { req, res };
+  }
+
+  it('flag OFF: the begin-hook does NOT open a registry entry (no hung empty entry)', async () => {
+    const { controller, streamRegistry, aiChatRunService, getRunHooks } =
+      makeController({ resumable: false });
+    const { req, res } = makeReqRes();
+    await controller.stream(req, res, user, workspace);
+
+    const runHooks = getRunHooks();
+    expect(runHooks).toBeDefined();
+    const handle = await runHooks.begin('chat-1');
+    // The run still begins (the durable-run feature is independent of resume)...
+    expect(aiChatRunService.beginRun).toHaveBeenCalled();
+    expect(handle).toEqual({ runId: 'run-1', signal: expect.anything() });
+    // ...but with the flag off the registry entry is NEVER opened.
+    expect(streamRegistry.open).not.toHaveBeenCalled();
+  });
+
+  it('flag ON: the begin-hook opens the registry entry with (chatId, runId)', async () => {
+    const { controller, streamRegistry, getRunHooks } = makeController({
+      resumable: true,
+    });
+    const { req, res } = makeReqRes();
+    await controller.stream(req, res, user, workspace);
+
+    const runHooks = getRunHooks();
+    await runHooks.begin('chat-1');
+    expect(streamRegistry.open).toHaveBeenCalledWith('chat-1', 'run-1');
+  });
+});

apps/server/src/core/ai-chat/ai-chat.controller.ts

@@ -4,11 +4,15 @@ import {
   ConflictException,
   Controller,
   ForbiddenException,
+  Get,
   HttpCode,
   HttpException,
   HttpStatus,
   Logger,
+  Param,
+  ParseUUIDPipe,
   Post,
+  Query,
   Req,
   Res,
   ServiceUnavailableException,
@@ -54,6 +58,12 @@ import {
 } from './dto/ai-chat.dto';
 import { describeProviderError } from '../../integrations/ai/ai-error.util';
 import { buildChatMarkdown } from './chat-markdown.util';
+import {
+  AiChatStreamRegistryService,
+  SUBSCRIBER_MAX_BUFFERED_BYTES,
+} from './ai-chat-stream-registry.service';
+import { startSseHeartbeat } from './sse-resilience';
+import { EnvironmentService } from '../../integrations/environment/environment.service';
 
 /**
  * Per-user AI chat API (§6.1). Routes are POST to match this codebase's
@@ -72,6 +82,11 @@ export class AiChatController {
     private readonly aiChatMessageRepo: AiChatMessageRepo,
     private readonly aiTranscription: AiTranscriptionService,
     private readonly pageRepo: PageRepo,
+    // #184 phase 1.5. OPTIONAL so existing positional constructions (controller
+    // specs) compile unchanged; Nest always injects the real providers in
+    // production. Only touched on the resumable-stream (flag-on) path.
+    private readonly streamRegistry?: AiChatStreamRegistryService,
+    private readonly environment?: EnvironmentService,
   ) {}
 
   /** List the requesting user's chats in this workspace (paginated). */
@@ -233,6 +248,102 @@ export class AiChatController {
     return { stopped };
   }
 
+  /**
+   * Attach to a chat's live run stream (#184 phase 1.5). A late/reloaded tab
+   * replays the frames buffered so far and then follows the live tail as a normal
+   * streamer. Owner-gated via assertOwnedChat (same gate as getRun). When there is
+   * nothing to resume — no entry, a finished run without expect=live, an
+   * overflowed buffer, or an anchor that pins a DIFFERENT run — the endpoint
+   * answers 204, the ONLY "nothing to resume" signal the AI SDK's reconnect
+   * accepts (it maps 204 to a silent no-op). With AI_CHAT_RESUMABLE_STREAM off the
+   * registry is never populated, so attach always 204s.
+   *
+   * `expect=live` opts into replaying a finished-but-retained run (safe only when
+   * the client stripped the streaming tail); `anchor` is the client's assistant
+   * row id, which must match this run's (invariant 6) or a foreign run's
+   * transcript would be replayed into the store.
+   */
+  @SkipTransform()
+  @UseGuards(JwtAuthGuard, UserThrottlerGuard)
+  @Throttle({ [AI_CHAT_THROTTLER]: { limit: 60, ttl: 60000 } })
+  @Get('runs/:chatId/stream')
+  async attachRunStream(
+    @Param('chatId', new ParseUUIDPipe()) chatId: string,
+    @Query('expect') expect: string | undefined,
+    @Query('anchor') anchor: string | undefined,
+    @Req() req: FastifyRequest,
+    @Res() res: FastifyReply,
+    @AuthUser() user: User,
+    @AuthWorkspace() workspace: Workspace,
+  ): Promise<void> {
+    await this.assertOwnedChat(chatId, user, workspace); // same gate as getRun
+    let stopHeartbeat: () => void = () => undefined;
+    const attachment = await this.streamRegistry?.attach(
+      chatId,
+      expect === 'live',
+      anchor,
+      {
+        onFrame: (frame) => {
+          // Backpressure guard: 2x the replay cap, so the initial replay burst
+          // alone can never trip it; only a genuinely stalled socket can.
+          try {
+            if (res.raw.writableLength > SUBSCRIBER_MAX_BUFFERED_BYTES) {
+              res.raw.destroy(); // 'close' fires -> unsubscribe below
+              return;
+            }
+            if (!res.raw.writableEnded) res.raw.write(frame);
+          } catch {
+            res.raw.destroy();
+          }
+        },
+        onEnd: () => {
+          stopHeartbeat();
+          if (!res.raw.writableEnded) res.raw.end();
+        },
+      },
+    );
+    if (!attachment) {
+      res.status(204).send(); // the ONLY "nothing to resume" signal the SDK accepts
+      return;
+    }
+    res.hijack();
+    // Cleanup BEFORE any write (invariant 5): a torn-down socket must not orphan
+    // a paused subscriber whose pending queue would buffer the whole run.
+    req.raw.once('close', () => {
+      attachment.unsubscribe();
+      stopHeartbeat();
+    });
+    // A close emitted DURING the awaits above was missed by the listener — check.
+    // (Healthy pending GETs have req.raw.destroyed === false, so no false
+    // positives; returning without end() is fine — the socket is gone.)
+    if (req.raw.destroyed) {
+      attachment.unsubscribe();
+      return;
+    }
+    res.raw.on('error', () => undefined);
+    try {
+      res.raw.writeHead(200, {
+        'content-type': 'text/event-stream',
+        'cache-control': 'no-cache',
+        'x-vercel-ai-ui-message-stream': 'v1',
+        'x-accel-buffering': 'no',
+        // deliberately NO Connection/Keep-Alive (hop-by-hop; Safari/HTTP2)
+      });
+      res.raw.flushHeaders?.();
+      for (const frame of attachment.replay) res.raw.write(frame);
+      if (attachment.finished) {
+        res.raw.end();
+        return;
+      }
+      stopHeartbeat = startSseHeartbeat(res.raw, 15_000);
+      attachment.start(); // drain pending accumulated during replay, go live
+    } catch {
+      attachment.unsubscribe();
+      stopHeartbeat();
+      res.raw.destroy();
+    }
+  }
+
   /** Rename a chat. */
   @HttpCode(HttpStatus.OK)
   @Post('rename')
@@ -344,13 +455,25 @@ export class AiChatController {
     // its progress, and settle its terminal status — see AiChatRunService.
     const runHooks: AiChatRunHooks | undefined = autonomousRuns
       ? {
-          begin: (chatId) =>
-            this.aiChatRunService.beginRun({
+          begin: async (chatId) => {
+            const handle = await this.aiChatRunService.beginRun({
               chatId,
               workspaceId: workspace.id,
               userId: user.id,
               trigger: 'user',
-            }),
+            });
+            // #184 phase 1.5: register the run-stream entry at BEGIN (before any
+            // frame) so a tab that attaches in the begin->seed window finds an
+            // entry to wait on. Gated on AI_CHAT_RESUMABLE_STREAM: with the flag
+            // off nothing is registered and attach always 204s.
+            if (
+              handle?.runId &&
+              this.environment?.isAiChatResumableStreamEnabled?.()
+            ) {
+              this.streamRegistry?.open(chatId, handle.runId);
+            }
+            return handle;
+          },
           onAssistantSeeded: (runId, messageId) =>
             this.aiChatRunService.linkAssistantMessage(
               runId,

apps/server/src/core/ai-chat/ai-chat.module.ts

@@ -4,6 +4,7 @@ import { TokenModule } from '../auth/token.module';
 import { AiChatController } from './ai-chat.controller';
 import { AiChatService } from './ai-chat.service';
 import { AiChatRunService } from './ai-chat-run.service';
+import { AiChatStreamRegistryService } from './ai-chat-stream-registry.service';
 import { AiTranscriptionService } from './ai-transcription.service';
 import { AiChatToolsService } from './tools/ai-chat-tools.service';
 import { EmbeddingModule } from './embedding/embedding.module';
@@ -44,6 +45,7 @@ import { PublicShareChatToolsService } from './tools/public-share-chat-tools.ser
   providers: [
     AiChatService,
     AiChatRunService,
+    AiChatStreamRegistryService,
     AiTranscriptionService,
     AiChatToolsService,
     PublicShareChatService,

apps/server/src/core/ai-chat/ai-chat.service.spec.ts

@@ -1,4 +1,13 @@
-import { ForbiddenException } from '@nestjs/common';
+import { ForbiddenException, Logger } from '@nestjs/common';
+// Mock ONLY streamText so a driven stream() call can capture the pipe-options
+// object (consumeSseStream / generateMessageId). Everything else in the AI SDK
+// stays REAL (requireActual), so the pure-helper suites in this file are
+// unaffected — none of them call stream()/streamText.
+jest.mock('ai', () => ({
+  ...jest.requireActual('ai'),
+  streamText: jest.fn(),
+}));
+import { streamText } from 'ai';
 import {
   AiChatService,
   compactToolOutput,
@@ -1059,3 +1068,181 @@ describe('isInterruptResume', () => {
     expect(isInterruptResume(withPrev(null), true)).toBe(false);
   });
 });
+
+/**
+ * #184 phase 1.5 — the run-wrapped pipe options (unit). Drives stream() to the
+ * pipe call with streamText mocked, capturing the options object, and asserts:
+ *  - flag OFF while a runId IS present -> the LEGACY option shape (no
+ *    consumeSseStream, no generateMessageId), and the registry is never touched.
+ *    This is the exact dormancy guarantee this PR rests on.
+ *  - flag ON + runId -> consumeSseStream tees into the registry and
+ *    generateMessageId returns the seeded assistant DB row id.
+ *  - flag ON but no runHooks (runId undefined) -> legacy (the runId gate).
+ *  - flag ON + runId -> the outer catch releases the entry via abortEntry.
+ */
+describe('AiChatService.stream — resumable pipe options (#184 phase 1.5)', () => {
+  const streamTextMock = streamText as unknown as jest.Mock;
+  let pipeMock: jest.Mock;
+
+  beforeEach(() => {
+    streamTextMock.mockReset();
+    pipeMock = jest.fn();
+    streamTextMock.mockReturnValue({
+      consumeStream: jest.fn(),
+      pipeUIMessageStreamToResponse: pipeMock,
+    });
+    // Silence the service's diagnostic logging for a clean test run.
+    jest.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined as never);
+    jest
+      .spyOn(Logger.prototype, 'error')
+      .mockImplementation(() => undefined as never);
+    jest
+      .spyOn(Logger.prototype, 'warn')
+      .mockImplementation(() => undefined as never);
+  });
+
+  afterEach(() => jest.restoreAllMocks());
+
+  // A raw-response stub sufficient for the post-streamText wiring.
+  function makeRes() {
+    return {
+      raw: {
+        writeHead: jest.fn(),
+        write: jest.fn(),
+        once: jest.fn(),
+        on: jest.fn(),
+        flushHeaders: jest.fn(),
+        writableEnded: false,
+        destroyed: false,
+      },
+    };
+  }
+
+  // Wire only the deps reached on the way to the pipe call, plus a spy registry.
+  function makeService(opts: { resumable: boolean }) {
+    const aiChatRepo = {
+      findById: jest.fn(async () => ({ id: 'chat-1', workspaceId: 'ws-1' })),
+      insert: jest.fn(),
+    };
+    const aiChatMessageRepo = {
+      // Both the user insert and the assistant seed return the same row id.
+      insert: jest.fn(async () => ({ id: 'msg-1' })),
+      findAllByChat: jest.fn(async () => []),
+      update: jest.fn(async () => ({ id: 'msg-1' })),
+    };
+    const aiSettings = { resolve: jest.fn(async () => ({})) };
+    const tools = { forUser: jest.fn(async () => ({})) };
+    const mcpClients = {
+      toolsFor: jest.fn(async () => ({
+        tools: {},
+        clients: [],
+        outcomes: [],
+        instructions: [],
+      })),
+    };
+    const streamRegistry = {
+      open: jest.fn(),
+      bind: jest.fn(),
+      abortEntry: jest.fn(),
+    };
+    const svc = new AiChatService(
+      {} as never, // ai (model is injected)
+      aiChatRepo as never,
+      aiChatMessageRepo as never,
+      {} as never, // aiChatPageSnapshotRepo
+      aiSettings as never,
+      tools as never,
+      mcpClients as never,
+      {} as never, // aiAgentRoleRepo
+      {} as never, // pageRepo (openPage undefined -> never touched)
+      {} as never, // pageAccess
+      {
+        isAiChatDeferredToolsEnabled: () => false,
+        isAiChatResumableStreamEnabled: () => opts.resumable,
+      } as never,
+      streamRegistry as never,
+    );
+    return { svc, streamRegistry };
+  }
+
+  const body = {
+    chatId: 'chat-1',
+    messages: [
+      { id: 'm1', role: 'user', parts: [{ type: 'text', text: 'hi' }] },
+    ],
+  };
+
+  const makeRunHooks = () => ({
+    begin: jest.fn(async () => ({
+      runId: 'run-1',
+      signal: new AbortController().signal,
+    })),
+    onAssistantSeeded: jest.fn(),
+    onStep: jest.fn(),
+    onSettled: jest.fn(),
+  });
+
+  async function drive(svc: AiChatService, hooks: unknown): Promise<void> {
+    await svc.stream({
+      user: { id: 'u1' } as never,
+      workspace: { id: 'ws-1' } as never,
+      sessionId: 's1',
+      body: body as never,
+      res: makeRes() as never,
+      signal: new AbortController().signal,
+      model: {} as never,
+      role: null,
+      runHooks: hooks as never,
+    });
+  }
+
+  it('flag OFF + runId present: LEGACY option shape (no consumeSseStream / generateMessageId); registry untouched', async () => {
+    const { svc, streamRegistry } = makeService({ resumable: false });
+    await drive(svc, makeRunHooks());
+    expect(pipeMock).toHaveBeenCalledTimes(1);
+    const options = pipeMock.mock.calls[0][1];
+    // The dormancy guarantee: a live run with the flag off tees NOTHING and does
+    // not stamp a message id — byte-for-byte the pre-1.5 wire.
+    expect(options.consumeSseStream).toBeUndefined();
+    expect(options.generateMessageId).toBeUndefined();
+    expect(streamRegistry.bind).not.toHaveBeenCalled();
+    expect(streamRegistry.abortEntry).not.toHaveBeenCalled();
+  });
+
+  it('flag ON + runId: consumeSseStream tees into the registry; generateMessageId returns the seeded row id', async () => {
+    const { svc, streamRegistry } = makeService({ resumable: true });
+    await drive(svc, makeRunHooks());
+    const options = pipeMock.mock.calls[0][1];
+    expect(typeof options.consumeSseStream).toBe('function');
+    expect(typeof options.generateMessageId).toBe('function');
+    // generateMessageId stamps the seeded assistant DB row id.
+    expect(options.generateMessageId()).toBe('msg-1');
+    // consumeSseStream binds the tee: (chatId, runId, assistantId, stream).
+    const fakeStream = {} as ReadableStream<string>;
+    options.consumeSseStream({ stream: fakeStream });
+    expect(streamRegistry.bind).toHaveBeenCalledWith(
+      'chat-1',
+      'run-1',
+      'msg-1',
+      fakeStream,
+    );
+  });
+
+  it('flag ON but NO runHooks (runId undefined): pipe options stay legacy (the runId gate)', async () => {
+    const { svc, streamRegistry } = makeService({ resumable: true });
+    await drive(svc, undefined);
+    const options = pipeMock.mock.calls[0][1];
+    expect(options.consumeSseStream).toBeUndefined();
+    expect(options.generateMessageId).toBeUndefined();
+    expect(streamRegistry.bind).not.toHaveBeenCalled();
+  });
+
+  it('flag ON + runId: the outer catch calls abortEntry when the stream throws', async () => {
+    const { svc, streamRegistry } = makeService({ resumable: true });
+    streamTextMock.mockImplementation(() => {
+      throw new Error('boom');
+    });
+    await expect(drive(svc, makeRunHooks())).rejects.toThrow('boom');
+    expect(streamRegistry.abortEntry).toHaveBeenCalledWith('chat-1', 'run-1');
+  });
+});

apps/server/src/core/ai-chat/ai-chat.service.ts

@@ -32,6 +32,7 @@ import {
 import { AiChatToolsService } from './tools/ai-chat-tools.service';
 import { McpClientsService } from './external-mcp/mcp-clients.service';
 import { EnvironmentService } from '../../integrations/environment/environment.service';
+import { AiChatStreamRegistryService } from './ai-chat-stream-registry.service';
 import { buildSystemPrompt } from './ai-chat.prompt';
 import {
   CORE_TOOL_KEYS,
@@ -276,6 +277,10 @@ export class AiChatService implements OnModuleInit {
     // Reads the AI_CHAT_DEFERRED_TOOLS toggle (#332). Injected last so existing
     // positional constructor callers (tests) only append one stub.
     private readonly environment: EnvironmentService,
+    // #184 phase 1.5 run-stream registry. OPTIONAL so existing positional
+    // constructions (int-specs) compile unchanged; Nest always injects the real
+    // provider in production. Only ever touched on the run-wrapped + flag-on path.
+    private readonly streamRegistry?: AiChatStreamRegistryService,
   ) {}
 
   /**
@@ -1174,6 +1179,35 @@ export class AiChatService implements OnModuleInit {
         // as the cumulative authoritative usage so the client never jumps DOWN.
         let cumulativeStepUsage: ChatStreamUsage | undefined;
         result.pipeUIMessageStreamToResponse(res.raw, {
+          // #184 phase 1.5: run-wrapped mode only — the legacy path (flag off) stays
+          // byte-for-byte identical, including the absence of start.messageId. Both
+          // fields are gated on `runId` (present only for a durable run) AND the
+          // AI_CHAT_RESUMABLE_STREAM flag; the seed `assistantId` is unconditional,
+          // so gating on `assistantId` alone would change the legacy wire.
+          ...(runId && this.environment?.isAiChatResumableStreamEnabled?.()
+            ? {
+                // Tee the SSE frames into the run-stream registry so late tabs can
+                // attach (replay + live tail).
+                consumeSseStream: ({
+                  stream,
+                }: {
+                  stream: ReadableStream<string>;
+                }) =>
+                  this.streamRegistry?.bind(
+                    chatId,
+                    runId!,
+                    assistantId,
+                    stream,
+                  ),
+                // Stamp the persisted assistant row's DB id onto the streamed
+                // message so every tab renders the SAME id as the DB row (id-based
+                // reconciliation). Seeding is best-effort: when it failed, let the
+                // client generate the id.
+                ...(assistantId
+                  ? { generateMessageId: () => assistantId }
+                  : {}),
+              }
+            : {}),
           headers: { 'X-Accel-Buffering': 'no' },
           // Surface the authoritative chatId on the streamed assistant UI message so
           // the client adopts the REAL id of the row we created, instead of guessing
@@ -1239,6 +1273,12 @@ export class AiChatService implements OnModuleInit {
       // finalizeRun (onSettled) is idempotent — a settle here and a settle from a
       // streamText callback collapse to a single terminal write.
       if (runId) {
+        // #184 phase 1.5: a failure here means the tee `done` will never arrive,
+        // so release the registry entry's subscribers explicitly — otherwise an
+        // attached tab hangs forever. Same flag gate as the tee wiring above.
+        if (this.environment?.isAiChatResumableStreamEnabled?.()) {
+          this.streamRegistry?.abortEntry(chatId, runId);
+        }
         await runHooks?.onSettled?.(
           runId,
           'error',

apps/server/src/database/listeners/page.listener.ts

@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
 import { InjectQueue } from '@nestjs/bullmq';
 import { QueueJob, QueueName } from '../../integrations/queue/constants';
 import { Queue } from 'bullmq';
-import { EnvironmentService } from '../../integrations/environment/environment.service';
 
 /**
  * Thin snapshot of a page node carried inside domain events so the WebSocket
@@ -112,48 +111,24 @@ export class PageListener {
   private readonly logger = new Logger(PageListener.name);
 
   constructor(
-    private readonly environmentService: EnvironmentService,
-    @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
     @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
   ) {}
 
   @OnEvent(EventName.PAGE_CREATED)
   async handlePageCreated(event: PageEvent) {
     const { pageIds, workspaceId } = event;
-    if (this.isTypesense()) {
-      await this.searchQueue.add(QueueJob.PAGE_CREATED, {
-        pageIds,
-      });
-    }
-
     await this.aiQueue.add(QueueJob.PAGE_CREATED, { pageIds, workspaceId });
   }
 
-  @OnEvent(EventName.PAGE_UPDATED)
-  async handlePageUpdated(event: PageEvent) {
-    const { pageIds } = event;
-
-    await this.searchQueue.add(QueueJob.PAGE_UPDATED, { pageIds });
-  }
-
   @OnEvent(EventName.PAGE_DELETED)
   async handlePageDeleted(event: PageEvent) {
     const { pageIds, workspaceId } = event;
-    if (this.isTypesense()) {
-      await this.searchQueue.add(QueueJob.PAGE_DELETED, { pageIds });
-    }
-
     await this.aiQueue.add(QueueJob.PAGE_DELETED, { pageIds, workspaceId });
   }
 
   @OnEvent(EventName.PAGE_SOFT_DELETED)
   async handlePageSoftDeleted(event: PageEvent) {
     const { pageIds, workspaceId } = event;
-
-    if (this.isTypesense()) {
-      await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
-    }
-
     await this.aiQueue.add(QueueJob.PAGE_SOFT_DELETED, {
       pageIds,
       workspaceId,
@@ -163,14 +138,6 @@ export class PageListener {
   @OnEvent(EventName.PAGE_RESTORED)
   async handlePageRestored(event: PageEvent) {
     const { pageIds, workspaceId } = event;
-    if (this.isTypesense()) {
-      await this.searchQueue.add(QueueJob.PAGE_RESTORED, { pageIds });
-    }
-
     await this.aiQueue.add(QueueJob.PAGE_RESTORED, { pageIds, workspaceId });
   }
-
-  isTypesense(): boolean {
-    return this.environmentService.getSearchDriver() === 'typesense';
-  }
 }

apps/server/src/database/listeners/space.listener.ts

@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
 import { InjectQueue } from '@nestjs/bullmq';
 import { QueueJob, QueueName } from '../../integrations/queue/constants';
 import { Queue } from 'bullmq';
-import { EnvironmentService } from '../../integrations/environment/environment.service';
 
 export class SpaceEvent {
   spaceId: string;
@@ -15,22 +14,12 @@ export class SpaceListener {
   private readonly logger = new Logger(SpaceListener.name);
 
   constructor(
-    private readonly environmentService: EnvironmentService,
-    @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
     @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
   ) {}
 
   @OnEvent(EventName.SPACE_DELETED)
   async handleSpaceDeleted(event: SpaceEvent) {
     const { spaceId } = event;
-    if (this.isTypesense()) {
-      await this.searchQueue.add(QueueJob.SPACE_DELETED, { spaceId });
-    }
-
     await this.aiQueue.add(QueueJob.SPACE_DELETED, { spaceId });
   }
-
-  isTypesense(): boolean {
-    return this.environmentService.getSearchDriver() === 'typesense';
-  }
 }

apps/server/src/database/listeners/workspace.listener.ts

@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
 import { InjectQueue } from '@nestjs/bullmq';
 import { QueueJob, QueueName } from '../../integrations/queue/constants';
 import { Queue } from 'bullmq';
-import { EnvironmentService } from '../../integrations/environment/environment.service';
 
 export class WorkspaceEvent {
   workspaceId: string;
@@ -15,22 +14,12 @@ export class WorkspaceListener {
   private readonly logger = new Logger(WorkspaceListener.name);
 
   constructor(
-    private readonly environmentService: EnvironmentService,
-    @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
     @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
   ) {}
 
   @OnEvent(EventName.WORKSPACE_DELETED)
   async handlePageDeleted(event: WorkspaceEvent) {
     const { workspaceId } = event;
-    if (this.isTypesense()) {
-      await this.searchQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
-    }
-
     await this.aiQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
   }
-
-  isTypesense(): boolean {
-    return this.environmentService.getSearchDriver() === 'typesense';
-  }
 }

apps/server/src/integrations/environment/environment.service.ts

@@ -292,6 +292,23 @@ export class EnvironmentService {
     return enabled === 'true';
   }
 
+  /**
+   * Resumable SSE transport for durable agent runs (#184 phase 1.5). When
+   * enabled, a run tees its SSE frames into the in-memory run-stream registry so
+   * a late/reloaded tab can attach (replay + live tail) via
+   * `GET /ai-chat/runs/:chatId/stream`. Defaults to DISABLED: PR 1 ships the
+   * server code dormant — with the flag off, `open`/`bind`/`generateMessageId`
+   * are never called and attach always answers 204, so the legacy and #184
+   * phase-1 wire paths stay byte-for-byte identical. Set
+   * AI_CHAT_RESUMABLE_STREAM=true to activate it (paired with the PR 2 client).
+   */
+  isAiChatResumableStreamEnabled(): boolean {
+    const enabled = this.configService
+      .get<string>('AI_CHAT_RESUMABLE_STREAM', 'false')
+      .toLowerCase();
+    return enabled === 'true';
+  }
+
   getPostHogHost(): string {
     return this.configService.get<string>('POSTHOG_HOST');
   }

apps/server/src/integrations/metrics/http-metrics.hook.ts

@@ -2,15 +2,36 @@ import { FastifyReply, FastifyRequest } from 'fastify';
 import { isStreamingResponse } from './metrics.constants';
 import { observeHttp } from './metrics.registry';
 
+// URL path prefixes served by @fastify/static (client build output under
+// client/dist). `/assets/` holds the content-hashed bundle (index-*.js,
+// chunk-*.js) — a NEW set of names every deploy, i.e. an UNBOUNDED label set
+// (#362); the others (/vad/, /brand/, /locales/, /icons/ — copied verbatim from
+// public/) have stable names, so they are merely repetitive per-file labels
+// rather than unbounded. Either way none of these belong in the API-route
+// histogram: collapse them all to one bounded `static` label. (Edge latency for
+// static is already measured by Traefik's traefik_router_request_duration_*.)
+const STATIC_PATH_PREFIXES = [
+  '/assets/',
+  '/vad/',
+  '/brand/',
+  '/locales/',
+  '/icons/',
+];
+
 /**
  * Resolve the BOUNDED route label for an HTTP response.
  *
- * HARD REQUIREMENT (#355): use the ROUTE TEMPLATE (`/pages/:id`), NEVER the raw
- * URL (`/pages/abc-123`), so label cardinality stays finite. Fastify exposes the
- * matched template on `req.routeOptions.url`. On 404s (no route matched) that is
- * missing → collapse to the literal `unknown`.
+ * HARD REQUIREMENT (#355): use the ROUTE TEMPLATE (`/pages/:id`), NEVER a raw
+ * URL (`/pages/abc-123` or `/assets/index-CAbxDtto.js`), so label cardinality
+ * stays finite. Fastify exposes the matched template on `req.routeOptions.url`,
+ * BUT @fastify/static serves each file through a route whose matched url is the
+ * raw (hashed) file path — so for static assets that value is itself unbounded.
+ * Detect static requests by their path prefix FIRST and collapse to `static`;
+ * otherwise use the route template; on a 404 (no route matched) → `unknown`.
  */
 export function resolveRouteLabel(req: FastifyRequest): string {
+  const path = (req.url ?? '').split('?', 1)[0];
+  if (STATIC_PATH_PREFIXES.some((p) => path.startsWith(p))) return 'static';
   const url = req.routeOptions?.url;
   return typeof url === 'string' && url.length > 0 ? url : 'unknown';
 }

apps/server/src/integrations/metrics/metrics-bull.service.ts

@@ -46,7 +46,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
     @InjectQueue(QueueName.GENERAL_QUEUE) generalQueue: Queue,
     @InjectQueue(QueueName.BILLING_QUEUE) billingQueue: Queue,
     @InjectQueue(QueueName.FILE_TASK_QUEUE) fileTaskQueue: Queue,
-    @InjectQueue(QueueName.SEARCH_QUEUE) searchQueue: Queue,
     @InjectQueue(QueueName.AI_QUEUE) aiQueue: Queue,
     @InjectQueue(QueueName.HISTORY_QUEUE) historyQueue: Queue,
     @InjectQueue(QueueName.NOTIFICATION_QUEUE) notificationQueue: Queue,
@@ -58,7 +57,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
       { label: 'general', queue: generalQueue },
       { label: 'billing', queue: billingQueue },
       { label: 'file-task', queue: fileTaskQueue },
-      { label: 'search', queue: searchQueue },
       { label: 'ai', queue: aiQueue },
       { label: 'history', queue: historyQueue },
       { label: 'notification', queue: notificationQueue },

apps/server/src/integrations/metrics/metrics.spec.ts

@@ -25,6 +25,61 @@ describe('resolveRouteLabel (histogram route label)', () => {
     const req = { url: '/x' } as unknown as FastifyRequest;
     expect(resolveRouteLabel(req)).toBe('unknown');
   });
+
+  it.each([
+    '/assets/index-CAbxDtto.js',
+    '/assets/chunk-3OPIFGDE-CJOt9nr5.js',
+    '/assets/excalidraw-menu-DpsI0kFW.js',
+    '/vad/silero_vad_v5.onnx',
+    '/brand/logo.svg',
+    '/locales/en.json',
+    '/icons/app-icon-192x192.png',
+  ])('collapses hashed/static asset %p to "static" (#362 cardinality)', (url) => {
+    // @fastify/static serves each file through a route whose matched url is the
+    // raw (hashed) file path, so routeOptions.url is itself unbounded here.
+    const req = {
+      url,
+      routeOptions: { url },
+    } as unknown as FastifyRequest;
+    const label = resolveRouteLabel(req);
+    expect(label).toBe('static');
+    expect(label).not.toContain('.js');
+    expect(label).not.toContain('index-');
+  });
+
+  it('strips the query string before the static-prefix check', () => {
+    const req = {
+      url: '/assets/index-CAbxDtto.js?v=2',
+      routeOptions: { url: '/assets/index-CAbxDtto.js' },
+    } as unknown as FastifyRequest;
+    expect(resolveRouteLabel(req)).toBe('static');
+  });
+
+  it('does NOT collapse a real API route that merely mentions assets', () => {
+    // A templated API route is kept as-is; only the static path PREFIXES collapse.
+    const req = {
+      url: '/api/pages/assets-guide',
+      routeOptions: { url: '/api/pages/:id' },
+    } as unknown as FastifyRequest;
+    expect(resolveRouteLabel(req)).toBe('/api/pages/:id');
+  });
+
+  it.each([
+    // The TRAILING SLASH on the prefix is the anti-false-collapse guard: a path
+    // that is the prefix WITHOUT its slash, or merely shares the prefix as a
+    // substring of a longer segment, must NOT collapse. These would collapse
+    // under a buggy `includes('/assets/')` / slashless-prefix impl.
+    '/assets',
+    '/assetsx/foo.js',
+    '/iconset/x.png',
+  ])('does NOT collapse the prefix-boundary case %p', (url) => {
+    const req = {
+      url,
+      routeOptions: { url: '/some/:route' },
+    } as unknown as FastifyRequest;
+    expect(resolveRouteLabel(req)).not.toBe('static');
+    expect(resolveRouteLabel(req)).toBe('/some/:route');
+  });
 });
 
 describe('isStreamingResponse (SSE exclusion)', () => {

apps/server/src/integrations/queue/constants/queue.constants.ts

@@ -4,7 +4,6 @@ export enum QueueName {
   GENERAL_QUEUE = '{general-queue}',
   BILLING_QUEUE = '{billing-queue}',
   FILE_TASK_QUEUE = '{file-task-queue}',
-  SEARCH_QUEUE = '{search-queue}',
   AI_QUEUE = '{ai-queue}',
   HISTORY_QUEUE = '{history-queue}',
   NOTIFICATION_QUEUE = '{notification-queue}',
@@ -32,12 +31,6 @@ export enum QueueJob {
   IMPORT_TASK = 'import-task',
   EXPORT_TASK = 'export-task',
 
-  SEARCH_INDEX_PAGE = 'search-index-page',
-  SEARCH_INDEX_PAGES = 'search-index-pages',
-  SEARCH_INDEX_COMMENT = 'search-index-comment',
-  SEARCH_INDEX_COMMENTS = 'search-index-comments',
-  SEARCH_INDEX_ATTACHMENT = 'search-index-attachment',
-  SEARCH_INDEX_ATTACHMENTS = 'search-index-attachments',
   SEARCH_REMOVE_PAGE = 'search-remove-page',
   SEARCH_REMOVE_ASSET = 'search-remove-attachment',
   SEARCH_REMOVE_FACE = 'search-remove-comment',

apps/server/src/integrations/queue/queue.module.ts

@@ -57,14 +57,6 @@ import { GeneralQueueProcessor } from './processors/general-queue.processor';
         attempts: 1,
       },
     }),
-    BullModule.registerQueue({
-      name: QueueName.SEARCH_QUEUE,
-      defaultJobOptions: {
-        removeOnComplete: true,
-        removeOnFail: true,
-        attempts: 2,
-      },
-    }),
     BullModule.registerQueue({
       name: QueueName.AI_QUEUE,
       defaultJobOptions: {

apps/server/test/integration/ai-chat-attach.int-spec.ts

@@ -0,0 +1,564 @@
+import * as http from 'node:http';
+import { Kysely } from 'kysely';
+import {
+  MockLanguageModelV3,
+  convertArrayToReadableStream,
+} from 'ai/test';
+import { AiChatRepo } from '@docmost/db/repos/ai-chat/ai-chat.repo';
+import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo';
+import { AiChatRunRepo } from '@docmost/db/repos/ai-chat/ai-chat-run.repo';
+import { AiChatService } from 'src/core/ai-chat/ai-chat.service';
+import { AiChatRunService } from 'src/core/ai-chat/ai-chat-run.service';
+import {
+  AiChatStreamRegistryService,
+  RunStreamCallbacks,
+} from 'src/core/ai-chat/ai-chat-stream-registry.service';
+import {
+  getTestDb,
+  destroyTestDb,
+  createWorkspace,
+  createUser,
+  createChat,
+} from './db';
+
+/**
+ * #184 phase 1.5 — the resumable transport end to end against REAL Postgres,
+ * the REAL `streamText` (seeded via MockLanguageModelV3) and a REAL Node
+ * ServerResponse, driving the REAL `AiChatService.stream` run-wrapped path with a
+ * REAL `AiChatStreamRegistryService`. The run-hooks mirror the controller: they
+ * begin a durable run and `open()` the registry entry at begin, and the service
+ * tees the SSE frames into it via `consumeSseStream` while stamping the DB row id
+ * via `generateMessageId` (both gated on runId + the resumable flag).
+ *
+ * Proven here: a finished run's replay is the full frame sequence incl `[DONE]`
+ * with `start.messageId` == the seeded DB row id; the anchor check (invariant 6);
+ * an attach opened BEFORE the first frame follows the live stream from frame 0; an
+ * explicit stop surfaces `{"type":"abort"}` + `[DONE]` + end to the subscriber;
+ * and the legacy (non-run) path tees nothing.
+ */
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+async function waitFor(
+  cond: () => Promise<boolean> | boolean,
+  { timeoutMs = 15_000, stepMs = 25 } = {},
+): Promise<void> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (await cond()) return;
+    await sleep(stepMs);
+  }
+  throw new Error('waitFor: condition not met within timeout');
+}
+
+// A real Node ServerResponse wired to a live socket (as in the stream int-spec).
+function makeRealResponse(): Promise<{
+  res: http.ServerResponse;
+  cleanup: () => Promise<void>;
+}> {
+  return new Promise((resolve) => {
+    const server = http.createServer((_req, res) => {
+      resolve({
+        res,
+        cleanup: () =>
+          new Promise<void>((done) => {
+            try {
+              if (!res.writableEnded) res.end();
+            } catch {
+              /* socket already gone */
+            }
+            server.close(() => done());
+          }),
+      });
+    });
+    server.listen(0, () => {
+      const port = (server.address() as any).port;
+      const creq = http.request({ port, method: 'GET' }, (cres) => {
+        cres.resume();
+      });
+      creq.on('error', () => undefined);
+      creq.end();
+    });
+  });
+}
+
+// A full, successful single-step turn.
+function successStream() {
+  return convertArrayToReadableStream([
+    { type: 'stream-start', warnings: [] },
+    { type: 'text-start', id: 't1' },
+    { type: 'text-delta', id: 't1', delta: 'Hello' },
+    { type: 'text-delta', id: 't1', delta: ' there' },
+    { type: 'text-end', id: 't1' },
+    {
+      type: 'finish',
+      finishReason: 'stop',
+      usage: { inputTokens: 10, outputTokens: 5, totalTokens: 15 },
+    },
+  ] as any);
+}
+
+// A stream the test feeds chunk by chunk (to attach mid-flight / drive a stop).
+function makeControlledStream() {
+  let controller!: ReadableStreamDefaultController<any>;
+  const stream = new ReadableStream<any>({
+    start(c) {
+      controller = c;
+    },
+  });
+  return {
+    stream,
+    emit: (chunk: any) => controller.enqueue(chunk),
+    close: () => controller.close(),
+  };
+}
+
+// Collect replay + live frames from an attachment.
+function liveSink(): {
+  cb: RunStreamCallbacks;
+  frames: string[];
+  ended: () => boolean;
+} {
+  const frames: string[] = [];
+  let ended = false;
+  return {
+    frames,
+    ended: () => ended,
+    cb: {
+      onFrame: (f) => frames.push(f),
+      onEnd: () => {
+        ended = true;
+      },
+    },
+  };
+}
+
+// The SSE `start` frame carries the message id; pull it out of a `data: {...}`.
+function parseStartMessageId(frames: string[]): string | undefined {
+  for (const f of frames) {
+    const m = /^data: (\{.*\})\s*$/m.exec(f.trim());
+    if (!m) continue;
+    try {
+      const json = JSON.parse(m[1]);
+      if (json.type === 'start') return json.messageId;
+    } catch {
+      /* not this frame */
+    }
+  }
+  return undefined;
+}
+
+describe('AiChatService run-stream attach [integration]', () => {
+  let db: Kysely<any>;
+  let aiChatRepo: AiChatRepo;
+  let msgRepo: AiChatMessageRepo;
+  let runRepo: AiChatRunRepo;
+  let workspaceId: string;
+  let userId: string;
+
+  const mcpClients = {
+    toolsFor: async () => ({
+      tools: {},
+      clients: [],
+      outcomes: [],
+      instructions: [],
+    }),
+  };
+
+  // Build the service with the run-stream registry wired and the resumable flag
+  // ON (the property under test). Deferred tools OFF (irrelevant here).
+  function buildService(registry: AiChatStreamRegistryService): AiChatService {
+    return new AiChatService(
+      { getChatModel: async () => null } as any,
+      aiChatRepo,
+      msgRepo,
+      {} as any,
+      { resolve: async () => null } as any,
+      { forUser: async () => ({}) } as any,
+      mcpClients as any,
+      {} as any,
+      {} as any,
+      {} as any,
+      {
+        isAiChatDeferredToolsEnabled: () => false,
+        isAiChatResumableStreamEnabled: () => true,
+      } as any,
+      registry,
+    );
+  }
+
+  // Run-hooks mirroring the controller: begin the durable run AND open() the
+  // registry entry at begin. Captures the runId so a test can stop it.
+  function makeRunHooks(
+    runService: AiChatRunService,
+    registry: AiChatStreamRegistryService,
+    box: { runId?: string },
+  ) {
+    return {
+      begin: async (chatId: string) => {
+        const handle = await runService.beginRun({
+          chatId,
+          workspaceId,
+          userId,
+          trigger: 'user',
+        });
+        box.runId = handle.runId;
+        registry.open(chatId, handle.runId);
+        return handle;
+      },
+      onAssistantSeeded: (runId: string, messageId: string) =>
+        runService.linkAssistantMessage(runId, workspaceId, messageId),
+      onStep: (runId: string, n: number) =>
+        void runService.recordStep(runId, workspaceId, n),
+      onSettled: (runId: string, status: any, error?: string) =>
+        runService.finalizeRun(runId, workspaceId, status, error),
+    };
+  }
+
+  function userUiMessage(text: string) {
+    return {
+      id: `u-${Math.random()}`,
+      role: 'user',
+      parts: [{ type: 'text', text }],
+    };
+  }
+
+  async function startRun(opts: {
+    registry: AiChatStreamRegistryService;
+    runService?: AiChatRunService;
+    model: MockLanguageModelV3;
+    chatId: string;
+    body: any;
+    box?: { runId?: string };
+  }): Promise<{ res: http.ServerResponse; cleanup: () => Promise<void> }> {
+    const service = buildService(opts.registry);
+    const { res, cleanup } = await makeRealResponse();
+    const runHooks = opts.runService
+      ? makeRunHooks(opts.runService, opts.registry, opts.box ?? {})
+      : undefined;
+    await service.stream({
+      user: { id: userId, workspaceId } as any,
+      workspace: { id: workspaceId, name: 'WS' } as any,
+      sessionId: 'sess-1',
+      body: opts.body,
+      res: { raw: res } as any,
+      signal: new AbortController().signal,
+      model: opts.model as any,
+      role: null,
+      runHooks,
+    } as any);
+    return { res, cleanup };
+  }
+
+  async function assistantRowId(chatId: string): Promise<string> {
+    const rows = await msgRepo.findAllByChat(chatId, workspaceId);
+    const row = rows.find((r: any) => r.role === 'assistant');
+    return row!.id as string;
+  }
+
+  beforeAll(async () => {
+    db = getTestDb();
+    aiChatRepo = new AiChatRepo(db as any);
+    msgRepo = new AiChatMessageRepo(db as any);
+    runRepo = new AiChatRunRepo(db as any);
+    workspaceId = (await createWorkspace(db)).id;
+    userId = (await createUser(db, workspaceId)).id;
+  });
+
+  afterAll(async () => {
+    await destroyTestDb();
+  });
+
+  it('run-wrapped: replay is the full frame sequence incl [DONE], start.messageId == the seeded DB row id', async () => {
+    const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+    const registry = new AiChatStreamRegistryService();
+    const runService = new AiChatRunService(runRepo, {
+      isCloud: () => false,
+    } as never);
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: successStream() }),
+    } as any);
+
+    const box: { runId?: string } = {};
+    const { cleanup } = await startRun({
+      registry,
+      runService,
+      model,
+      chatId,
+      body: { chatId, messages: [userUiMessage('Hi')] },
+      box,
+    });
+    try {
+      // Wait for the assistant row to settle (terminal callbacks run async).
+      await waitFor(async () => {
+        const rows = await msgRepo.findAllByChat(chatId, workspaceId);
+        return rows.some(
+          (r: any) =>
+            r.role === 'assistant' &&
+            ['completed', 'error', 'aborted'].includes(r.status),
+        );
+      });
+      const rowId = await assistantRowId(chatId);
+
+      // Finished-run replay with expect=live + the correct anchor.
+      const sink = liveSink();
+      const att = await registry.attach(chatId, true, rowId, sink.cb);
+      expect(att).not.toBeNull();
+      expect(att!.finished).toBe(true);
+      // The tee captured frames (consumeSseStream was wired).
+      expect(att!.replay.length).toBeGreaterThan(0);
+      // generateMessageId stamped the DB row id onto the streamed start frame.
+      expect(parseStartMessageId(att!.replay)).toBe(rowId);
+      // The full sequence includes the streamed text and the terminal marker.
+      const joined = att!.replay.join('');
+      expect(joined).toContain('Hello');
+      expect(att!.replay.some((f) => f.includes('[DONE]'))).toBe(true);
+    } finally {
+      registry.onModuleDestroy();
+      await cleanup();
+    }
+  });
+
+  it('anchor mismatch with expect=live returns null (invariant 6)', async () => {
+    const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+    const registry = new AiChatStreamRegistryService();
+    const runService = new AiChatRunService(runRepo, {
+      isCloud: () => false,
+    } as never);
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: successStream() }),
+    } as any);
+
+    const { cleanup } = await startRun({
+      registry,
+      runService,
+      model,
+      chatId,
+      body: { chatId, messages: [userUiMessage('Hi')] },
+    });
+    try {
+      await waitFor(async () => {
+        const rows = await msgRepo.findAllByChat(chatId, workspaceId);
+        return rows.some(
+          (r: any) => r.role === 'assistant' && r.status === 'completed',
+        );
+      });
+      const sink = liveSink();
+      // A foreign anchor must NOT replay this run's transcript.
+      expect(
+        await registry.attach(chatId, true, 'a-different-run-row', sink.cb),
+      ).toBeNull();
+    } finally {
+      registry.onModuleDestroy();
+      await cleanup();
+    }
+  });
+
+  it('an attach opened BEFORE the first frame follows the live stream from frame 0', async () => {
+    const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+    const registry = new AiChatStreamRegistryService();
+    const runService = new AiChatRunService(runRepo, {
+      isCloud: () => false,
+    } as never);
+    const controlled = makeControlledStream();
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: controlled.stream }),
+    } as any);
+
+    const { cleanup } = await startRun({
+      registry,
+      runService,
+      model,
+      chatId,
+      body: { chatId, messages: [userUiMessage('Slow please')] },
+    });
+    try {
+      // Attach while the entry exists (opened at begin) but before any frame.
+      const sink = liveSink();
+      const att = (await registry.attach(chatId, false, undefined, sink.cb))!;
+      expect(att.replay).toEqual([]); // nothing streamed yet -> replay from 0
+      att.start(); // go live (drains nothing, then follows)
+
+      // Now emit the whole turn.
+      controlled.emit({ type: 'stream-start', warnings: [] });
+      controlled.emit({ type: 'text-start', id: 't1' });
+      controlled.emit({ type: 'text-delta', id: 't1', delta: 'Zero' });
+      controlled.emit({ type: 'text-end', id: 't1' });
+      controlled.emit({
+        type: 'finish',
+        finishReason: 'stop',
+        usage: { inputTokens: 1, outputTokens: 1, totalTokens: 2 },
+      });
+      controlled.close();
+
+      await waitFor(() => sink.frames.some((f) => f.includes('[DONE]')));
+      // The subscriber saw the stream from the very first frame (`start`) through
+      // the terminal marker, with the streamed text present.
+      expect(sink.frames.some((f) => f.includes('"type":"start"'))).toBe(true);
+      expect(sink.frames.join('')).toContain('Zero');
+      expect(sink.frames[sink.frames.length - 1]).toContain('[DONE]');
+      expect(sink.ended()).toBe(true);
+    } finally {
+      registry.onModuleDestroy();
+      await cleanup();
+    }
+  });
+
+  it('requestStop surfaces {"type":"abort"} + [DONE] + end to the attached subscriber', async () => {
+    const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+    const registry = new AiChatStreamRegistryService();
+    const runService = new AiChatRunService(runRepo, {
+      isCloud: () => false,
+    } as never);
+    // An abort-AWARE model: it streams some partial output, then errors the model
+    // stream with an AbortError when the run signal aborts — exactly as a real
+    // provider network stream is torn down on abort (a plain in-memory stream
+    // would just stall, so streamText would never observe the stop).
+    const model = new MockLanguageModelV3({
+      doStream: async ({ abortSignal }: any) => {
+        const stream = new ReadableStream<any>({
+          start(controller) {
+            controller.enqueue({ type: 'stream-start', warnings: [] });
+            controller.enqueue({ type: 'text-start', id: 't1' });
+            controller.enqueue({ type: 'text-delta', id: 't1', delta: 'partial' });
+            abortSignal?.addEventListener('abort', () => {
+              try {
+                controller.error(
+                  new DOMException('Aborted', 'AbortError'),
+                );
+              } catch {
+                /* already errored/closed */
+              }
+            });
+          },
+        });
+        return { stream };
+      },
+    } as any);
+
+    const box: { runId?: string } = {};
+    const { cleanup } = await startRun({
+      registry,
+      runService,
+      model,
+      chatId,
+      body: { chatId, messages: [userUiMessage('Start then stop')] },
+      box,
+    });
+    try {
+      const sink = liveSink();
+      const att = (await registry.attach(chatId, false, undefined, sink.cb))!;
+      att.start();
+
+      // Give streamText a beat to begin consuming the partial output.
+      await sleep(250);
+      // User presses Stop -> the run signal aborts -> the SDK emits an abort chunk.
+      await runService.requestStop(box.runId!, workspaceId);
+
+      await waitFor(() => sink.ended());
+      expect(sink.frames.some((f) => f.includes('"type":"abort"'))).toBe(true);
+      expect(sink.frames.some((f) => f.includes('[DONE]'))).toBe(true);
+      expect(sink.ended()).toBe(true);
+    } finally {
+      registry.onModuleDestroy();
+      await cleanup();
+    }
+  });
+
+  it('the outer catch calls abortEntry so an open entry is released (finished)', async () => {
+    const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+    const registry = new AiChatStreamRegistryService();
+    const runService = new AiChatRunService(runRepo, {
+      isCloud: () => false,
+    } as never);
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: successStream() }),
+    } as any);
+
+    // A msgRepo whose user-row insert throws: the turn fails AFTER begin (the
+    // entry is already open) but BEFORE the pipe, exercising the outer catch.
+    const throwingMsgRepo = {
+      insert: async () => {
+        throw new Error('db boom');
+      },
+      findAllByChat: (...a: any[]) => (msgRepo as any).findAllByChat(...a),
+      update: (...a: any[]) => (msgRepo as any).update(...a),
+      findById: (...a: any[]) => (msgRepo as any).findById(...a),
+    };
+    const service = new AiChatService(
+      { getChatModel: async () => null } as any,
+      aiChatRepo,
+      throwingMsgRepo as any,
+      {} as any,
+      { resolve: async () => null } as any,
+      { forUser: async () => ({}) } as any,
+      mcpClients as any,
+      {} as any,
+      {} as any,
+      {} as any,
+      {
+        isAiChatDeferredToolsEnabled: () => false,
+        isAiChatResumableStreamEnabled: () => true,
+      } as any,
+      registry,
+    );
+    const { res, cleanup } = await makeRealResponse();
+    const box: { runId?: string } = {};
+    try {
+      await expect(
+        service.stream({
+          user: { id: userId, workspaceId } as any,
+          workspace: { id: workspaceId, name: 'WS' } as any,
+          sessionId: 'sess-1',
+          body: { chatId, messages: [userUiMessage('will throw')] },
+          res: { raw: res } as any,
+          signal: new AbortController().signal,
+          model: model as any,
+          role: null,
+          runHooks: makeRunHooks(runService, registry, box),
+        } as any),
+      ).rejects.toThrow();
+      // The entry opened at begin was terminated by abortEntry (from the catch),
+      // so it is finished and a plain attach returns null instead of hanging.
+      const entry = (registry as any).entries.get(chatId);
+      expect(entry).toBeDefined();
+      expect(entry.finished).toBe(true);
+      const sink = liveSink();
+      expect(await registry.attach(chatId, false, undefined, sink.cb)).toBeNull();
+    } finally {
+      registry.onModuleDestroy();
+      await cleanup();
+    }
+  });
+
+  it('legacy (no run-hooks): the registry is never populated', async () => {
+    const chatId = (await createChat(db, { workspaceId, creatorId: userId })).id;
+    const registry = new AiChatStreamRegistryService();
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: successStream() }),
+    } as any);
+
+    // No runHooks -> runId undefined -> the run-wrapped tee is never wired.
+    const { cleanup } = await startRun({
+      registry,
+      model,
+      chatId,
+      body: { chatId, messages: [userUiMessage('Legacy hi')] },
+    });
+    try {
+      await waitFor(async () => {
+        const rows = await msgRepo.findAllByChat(chatId, workspaceId);
+        return rows.some(
+          (r: any) => r.role === 'assistant' && r.status === 'completed',
+        );
+      });
+      const sink = liveSink();
+      // No entry was ever opened; attach always yields null.
+      expect(await registry.attach(chatId, false, undefined, sink.cb)).toBeNull();
+      expect(await registry.attach(chatId, true, 'anything', sink.cb)).toBeNull();
+    } finally {
+      registry.onModuleDestroy();
+      await cleanup();
+    }
+  });
+});

apps/server/test/integration/db.ts

@@ -38,6 +38,24 @@ export const TEST_DATABASE_URL =
   process.env.TEST_DATABASE_URL ??
   'postgresql://docmost:docmost_dev_pw@localhost:5432/docmost_test';
 
+// Build the raw postgres.js client (mirrors database.module.ts: max pool,
+// silenced notices, bigint-as-number parsing). Kept separate so the singleton
+// can hold a reference to bound its shutdown in destroyTestDb.
+function buildTestSql(url: string = TEST_DATABASE_URL) {
+  return postgres(url, {
+    max: 5,
+    onnotice: () => {},
+    types: {
+      bigint: {
+        to: 20,
+        from: [20, 1700],
+        serialize: (value: number) => value.toString(),
+        parse: (value: string) => Number.parseInt(value),
+      },
+    },
+  });
+}
+
 /**
  * Build a Kysely instance that MIRRORS the app's setup in database.module.ts:
  * PostgresJSDialect over postgres(), CamelCasePlugin, and the bigint type
@@ -47,38 +65,40 @@ export const TEST_DATABASE_URL =
  */
 export function buildTestDb(url: string = TEST_DATABASE_URL): Kysely<any> {
   return new Kysely<any>({
-    dialect: new PostgresJSDialect({
-      postgres: postgres(url, {
-        max: 5,
-        onnotice: () => {},
-        types: {
-          bigint: {
-            to: 20,
-            from: [20, 1700],
-            serialize: (value: number) => value.toString(),
-            parse: (value: string) => Number.parseInt(value),
-          },
-        },
-      }),
-    }),
+    dialect: new PostgresJSDialect({ postgres: buildTestSql(url) }),
     plugins: [new CamelCasePlugin()],
   });
 }
 
 let singleton: Kysely<any> | undefined;
+let singletonSql: ReturnType<typeof buildTestSql> | undefined;
 
 /** Lazily-built shared Kysely for the test suite (one per worker; maxWorkers=1). */
 export function getTestDb(): Kysely<any> {
   if (!singleton) {
-    singleton = buildTestDb();
+    singletonSql = buildTestSql();
+    singleton = new Kysely<any>({
+      dialect: new PostgresJSDialect({ postgres: singletonSql }),
+      plugins: [new CamelCasePlugin()],
+    });
   }
   return singleton;
 }
 
 export async function destroyTestDb(): Promise<void> {
-  if (singleton) {
-    await singleton.destroy();
-    singleton = undefined;
+  if (!singleton) return;
+  const sql = singletonSql;
+  // Clear the refs first so a hung end() cannot leave a half-closed singleton.
+  singleton = undefined;
+  singletonSql = undefined;
+  // postgres.js .end() waits indefinitely for in-flight queries by default; a
+  // leaked/stuck pooled connection would hang the afterAll hook (a 60s hook
+  // timeout in CI). Bound the shutdown: the { timeout } grace period lets
+  // active queries drain, then force-closes lingering sockets so teardown
+  // always completes. We close the pool directly instead of Kysely.destroy()
+  // (which would call sql.end() again with no timeout).
+  if (sql) {
+    await sql.end({ timeout: 5 });
   }
 }
 

apps/server/test/jest-e2e.json

@@ -4,10 +4,14 @@
   "testEnvironment": "node",
   "testRegex": ".e2e-spec.ts$",
   "transform": {
+    "prosemirror-markdown/build/.+\\.js$": [
+      "babel-jest",
+      { "presets": [["@babel/preset-env", { "targets": { "node": "current" } }]] }
+    ],
     "^.+\\.(t|j)sx?$": ["ts-jest", { "tsconfig": { "allowJs": true } }]
   },
   "transformIgnorePatterns": [
-    "/node_modules/(?!(\\.pnpm/)?(nanoid|uuid|image-dimensions|marked|happy-dom|lib0|@sindresorhus[+/][a-z0-9-]+|escape-string-regexp|p-limit|yocto-queue)(@|/))"
+    "/node_modules/(?!(\\.pnpm/)?(nanoid|uuid|image-dimensions|marked|happy-dom|lib0|@sindresorhus[+/][a-z0-9-]+|escape-string-regexp|p-limit|yocto-queue|@docmost/prosemirror-markdown)(@|/))"
   ],
   "moduleNameMapper": {
     "^@docmost/db/(.*)$": "<rootDir>/../src/database/$1",

apps/server/test/jest-integration.json

@@ -4,14 +4,19 @@
   "testRegex": ".*\\.int-spec\\.ts$",
   "testPathIgnorePatterns": ["/node_modules/"],
   "transform": {
+    "prosemirror-markdown/build/.+\\.js$": [
+      "babel-jest",
+      { "presets": [["@babel/preset-env", { "targets": { "node": "current" } }]] }
+    ],
     "^.+\\.(t|j)sx?$": "ts-jest"
   },
   "transformIgnorePatterns": [
-    "/node_modules/(?!(\\.pnpm/)?(nanoid|uuid|image-dimensions|marked|happy-dom|lib0)(@|/))"
+    "/node_modules/(?!(\\.pnpm/)?(nanoid|uuid|image-dimensions|marked|happy-dom|lib0|@docmost/prosemirror-markdown)(@|/))"
   ],
   "testEnvironment": "node",
   "testTimeout": 60000,
   "maxWorkers": 1,
+  "forceExit": true,
   "globalSetup": "<rootDir>/test/integration/global-setup.ts",
   "globalTeardown": "<rootDir>/test/integration/global-teardown.ts",
   "moduleNameMapper": {

packages/prosemirror-markdown/test/gitmost-transcript-neutralization.test.ts

@@ -0,0 +1,124 @@
+import { describe, expect, it } from "vitest";
+// Import DIRECTLY from src (NOT the docmost-client barrel, which pulls in
+// collaboration.ts and mutates global DOM at import time).
+import { convertProseMirrorToMarkdown } from "../src/lib/markdown-converter.js";
+import { markdownToProseMirror } from "../src/lib/markdown-to-prosemirror.js";
+
+/**
+ * gitmost #377 (round-1 review, finding #1) — proof, against the REAL
+ * converter, that the transcript-insert boundary defense survives git-sync.
+ *
+ * The web bridge (apps/client .../gitmost/gitmost-recording.ts,
+ * `gitmostInsertTranscriptIntoEditor`) appends each transcript line as a
+ * PARAGRAPH text node. The paragraph serializer here (`case "paragraph"`) emits
+ * that text VERBATIM with no block-escape, so a line whose text begins with a
+ * col-0 markdown block trigger would, on the doc -> markdown -> doc git-sync
+ * cycle, silently re-parse into a heading / list / quote / callout / code block.
+ * That missing block-escape is the pre-existing root cause; the bridge's
+ * boundary defense prepends an invisible zero-width space (U+200B) to a line
+ * that begins with such a trigger, shifting it off column 0.
+ *
+ * This test keeps a COPY of the bridge's trigger regex (the bridge is in a
+ * different package and can't be imported here) and asserts:
+ *   1. bare trigger lines DO corrupt (documents the root cause), and
+ *   2. the ZWSP-neutralized form round-trips as a single PARAGRAPH with the
+ *      text byte-preserved.
+ */
+
+const ZWSP = "​"; // U+200B
+
+// MUST stay in sync with GITMOST_MD_BLOCK_TRIGGER_RE in the client bridge.
+const MD_BLOCK_TRIGGER_RE =
+  /^(?:#{1,6}(?:\s|$)|[-*+](?:\s|$)|>|\d+[.)](?:\s|$)|```|~~~|\||([-*_])(?:\s*\1){2,}\s*$)/;
+
+const doc = (...nodes: any[]) => ({ type: "doc", content: nodes });
+const para = (t: string) => ({
+  type: "paragraph",
+  content: [{ type: "text", text: t }],
+});
+
+const roundtrip = async (text: string) => {
+  const md = convertProseMirrorToMarkdown(doc(para(text)));
+  const back = await markdownToProseMirror(md);
+  return back.content as any[];
+};
+
+describe("gitmost transcript neutralization (git-sync round-trip)", () => {
+  // Lines that, at column 0, the serializer's missing block-escape would let
+  // git-sync re-parse into a non-paragraph block.
+  const triggerLines = [
+    "- dash",
+    "* star",
+    "+ plus",
+    "> quote",
+    "# hash",
+    "1. one",
+    "1) one",
+    "> [!info] note",
+    "```js",
+    "~~~",
+    // Solid + spaced thematic breaks — these re-parse into a `horizontalRule`,
+    // which carries NO text, so a bare separator line LOSES its text entirely
+    // (round-2 finding). `_` also only forms a block via this construct.
+    "---",
+    "***",
+    "___",
+    "- - -", // spaced dash break (solid form is caught by [-*+]\s too, but this is the break)
+    "_ _ _",
+  ];
+
+  it("BARE trigger lines corrupt into non-paragraph blocks (root cause)", async () => {
+    for (const line of triggerLines) {
+      const blocks = await roundtrip(line);
+      // At least one produced block is NOT a paragraph — i.e. corruption.
+      const allParagraphs = blocks.every((b) => b.type === "paragraph");
+      expect(
+        allParagraphs,
+        `expected "${line}" to corrupt when inserted bare`,
+      ).toBe(false);
+    }
+  });
+
+  it("BARE solid thematic breaks corrupt into a text-LOSING horizontalRule", async () => {
+    // The severe case: no text node survives. Documents why neutralization
+    // matters more here than for list/quote (where the text survived).
+    for (const line of ["---", "***", "___"]) {
+      const blocks = await roundtrip(line);
+      expect(blocks.map((b) => b.type)).toContain("horizontalRule");
+      // No block carries the original text anywhere.
+      const flat = JSON.stringify(blocks);
+      expect(flat).not.toContain(line);
+    }
+  });
+
+  it("ZWSP-neutralized trigger lines round-trip as a single paragraph, text preserved", async () => {
+    for (const line of triggerLines) {
+      // The regex must actually classify each as a trigger.
+      expect(MD_BLOCK_TRIGGER_RE.test(line), `regex missed "${line}"`).toBe(
+        true,
+      );
+      const neutralized = ZWSP + line;
+      const blocks = await roundtrip(neutralized);
+
+      expect(blocks).toHaveLength(1);
+      expect(blocks[0].type).toBe("paragraph");
+      // Text is byte-preserved (ZWSP + original line), so the display is the
+      // original line with only an invisible leading character.
+      expect(blocks[0].content[0].text).toBe(neutralized);
+    }
+  });
+
+  it("normal host-prefixed lines never match the trigger regex and round-trip byte-exact", async () => {
+    for (const line of [
+      "You: hello there",
+      "Speaker 1: - and then a dash mid-line",
+      "Speaker 2: 1. not a list",
+    ]) {
+      expect(MD_BLOCK_TRIGGER_RE.test(line)).toBe(false);
+      const blocks = await roundtrip(line);
+      expect(blocks).toHaveLength(1);
+      expect(blocks[0].type).toBe("paragraph");
+      expect(blocks[0].content[0].text).toBe(line);
+    }
+  });
+});