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>

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/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/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();
+    }
+  });
+});

packages/git-sync/test/stabilize.test.ts

@@ -55,15 +55,10 @@ describe('stabilizePageFile — normalize-on-write fixpoint (SPEC §11)', () =>
     const file2 = await stabilizePageFile(doc2, meta);
     expect(file2).toBe(file1);
 
-    // The drawio node was materialized to its canonical HTML form by the
-    // convergence pass — a bare `{ src }` doc node becomes the full
-    // `<div data-type="drawio" data-src=...>` — proof the pass actually ran, not
-    // just two naive exports happening to match. Assert on the stable canonical
-    // markers rather than `data-align="center"`: center is a schema default the
-    // converter may omit (see prosemirror-markdown media-html.ts), so it is not
-    // a reliable convergence proof.
-    expect(body1).toContain('data-type="drawio"');
-    expect(body1).toContain('data-src="/d.drawio"');
+    // The materialized diagram default is present in the stabilized body (proof
+    // that the convergence pass actually ran, not just that two naive exports
+    // happened to match).
+    expect(body1).toContain('data-align="center"');
   });
 
   it('already-stable content is unchanged by the pass (idempotent)', async () => {

packages/prosemirror-markdown/src/lib/markdown-converter.ts

@@ -48,52 +48,6 @@ export interface ConvertProseMirrorToMarkdownOptions {
   dropResolvedCommentAnchors?: boolean;
 }
 
-/**
- * Adjacent sibling lists that share a markdown MARKER FAMILY re-parse as ONE
- * merged list — bulletList and taskList both emit `- ` markers (→ a single
- * `<ul>`), and two orderedLists both emit `1.` markers (→ a single `<ol>`). The
- * cross-type case is real data loss the editor CAN produce (e.g. a taskList
- * followed by a bulletList: the merged `<ul>` has a mix of checkbox and plain
- * items, so `bridgeTaskLists` refuses to convert it and every taskItem loses its
- * checkbox). Between two such adjacent list children we emit an empty HTML comment
- * `<!-- -->`: marked renders it as its own HTML block that interrupts the list, so
- * the two lists stay distinct; on import the comment is inert (parseAttachedComment
- * → null) and dropped by generateJSON, and re-export re-inserts it, so the marker
- * is byte-stable. It fires ONLY between two adjacent same-family list nodes — no
- * separator is emitted for any other join, so non-list output is unchanged.
- */
-const LIST_MARKER_SEPARATOR = "<!-- -->";
-function listMarkerFamily(type: string | undefined): "ul" | "ol" | null {
-  if (type === "bulletList" || type === "taskList") return "ul";
-  if (type === "orderedList") return "ol";
-  return null;
-}
-function adjacentListsMerge(
-  prevType: string | undefined,
-  curType: string | undefined,
-): boolean {
-  const a = listMarkerFamily(prevType);
-  return a !== null && a === listMarkerFamily(curType);
-}
-/**
- * Render each block child, inserting a `<!-- -->` separator entry between any two
- * adjacent same-marker-family list nodes (see LIST_MARKER_SEPARATOR). Callers
- * join the returned strings with their own context separator.
- */
-function renderBlockChildren(
-  children: any[],
-  render: (n: any) => string,
-): string[] {
-  const out: string[] = [];
-  let prevType: string | undefined;
-  for (const child of children) {
-    if (adjacentListsMerge(prevType, child?.type)) out.push(LIST_MARKER_SEPARATOR);
-    out.push(render(child));
-    prevType = child?.type;
-  }
-  return out;
-}
-
 /**
  * Convert ProseMirror/TipTap JSON content to Markdown
  * Supports all Docmost-specific node types and extensions
@@ -393,15 +347,9 @@ export function convertProseMirrorToMarkdown(
         // lossless (the body survives) and byte-stable (it re-exports identically),
         // so it is deliberately not treated as data loss.
         const parts: string[] = [];
-        let prevDocType: string | undefined;
         for (const child of nodeContent) {
           if (child?.type === "footnotesList") continue;
-          // Keep adjacent same-family sibling lists distinct (see renderBlockChildren).
-          if (adjacentListsMerge(prevDocType, child?.type)) {
-            parts.push(LIST_MARKER_SEPARATOR);
-          }
           parts.push(processNode(child));
-          prevDocType = child?.type;
         }
         for (const [id, def] of footnoteDefs) {
           if (!referencedFootnoteIds.has(id)) {
@@ -651,34 +599,20 @@ export function convertProseMirrorToMarkdown(
         return processTaskItem(node);
 
       case "listItem":
-        // Direct-listItem path (lists normally render via processListItem, which
-        // handles the marker + indentation). Blank line between block children so
-        // multiple paragraphs do not merge on re-parse; a `<!-- -->` entry
-        // (renderBlockChildren) separates adjacent sibling lists.
-        return renderBlockChildren(nodeContent, processNode).join("\n\n");
+        return nodeContent.map(processNode).join("\n");
 
-      case "blockquote": {
+      case "blockquote":
         // Prefix EVERY line of EVERY child with "> " and separate block-level
         // children with a blank ">" line so code blocks / multi-paragraph
-        // quotes round-trip correctly. A `> <!-- -->` separator line is inserted
-        // between two adjacent same-family sibling lists (renderBlockChildren)
-        // so they stay distinct inside the quote.
-        const bqParts: string[] = [];
-        let prevBqType: string | undefined;
-        for (const n of nodeContent) {
-          if (adjacentListsMerge(prevBqType, n?.type)) {
-            bqParts.push(`> ${LIST_MARKER_SEPARATOR}`);
-          }
-          bqParts.push(
+        // quotes round-trip correctly.
+        return nodeContent
+          .map((n: any) =>
             processNode(n)
               .split("\n")
               .map((line: string) => (line.length ? `> ${line}` : ">"))
               .join("\n"),
-          );
-          prevBqType = n?.type;
-        }
-        return bqParts.join("\n>\n");
-      }
+          )
+          .join("\n>\n");
 
       case "horizontalRule":
         return "---";
@@ -853,12 +787,9 @@ export function convertProseMirrorToMarkdown(
         // blockquote-prefixed; a blank line becomes a bare `>` so the callout is
         // not split.
         const calloutType = (node.attrs?.type || "info").toLowerCase();
-        const calloutBody = renderBlockChildren(nodeContent, processNode)
-          // Blank line between block children (rendered as a bare `>` after the
-          // prefix pass below) so multiple paragraphs stay separate nodes instead
-          // of merging on re-parse — same rule blockquote already uses. A
-          // `<!-- -->` entry (renderBlockChildren) separates adjacent sibling lists.
-          .join("\n\n")
+        const calloutBody = nodeContent
+          .map(processNode)
+          .join("\n")
           .split("\n")
           .map((l: string) => (l.length ? `> ${l}` : ">"))
           .join("\n");
@@ -878,10 +809,7 @@ export function convertProseMirrorToMarkdown(
         return `<summary>${renderInlineChildren(nodeContent)}</summary>\n\n`;
 
       case "detailsContent":
-        // Blank line between block children so multiple paragraphs in a details
-        // body survive as separate nodes (a single "\n" merges them on re-parse);
-        // a `<!-- -->` entry (renderBlockChildren) separates adjacent sibling lists.
-        return `${renderBlockChildren(nodeContent, processNode).join("\n\n")}\n`;
+        return `${nodeContent.map(processNode).join("\n")}\n`;
 
       case "mathInline": {
         // #293 canon #6: inline math serializes as Obsidian-native `$LaTeX$`
@@ -1410,19 +1338,11 @@ export function convertProseMirrorToMarkdown(
         // The code itself is element TEXT content (between <code> tags), so it
         // must escape < > & — NOT the attribute escaper. The language rides in
         // a class ATTRIBUTE, so it uses escapeAttr.
-        //
-        // Read the child text RAW (as `case "codeBlock"` does) and keep it
-        // VERBATIM — do NOT strip the trailing newline. Unlike the markdown fence
-        // path (which strips then relies on marked re-adding one `\n`), the schema
-        // codeBlock parseHTML reads the `<code>` text content back byte-for-byte,
-        // so stripping here would drop a trailing newline the node legitimately
-        // carries and break the round trip inside a column/cell.
         const code = escapeHtmlText(
           children
-            .map((child: any) =>
-              typeof child?.text === "string" ? child.text : "",
-            )
-            .join(""),
+            .map(processNode)
+            .join("")
+            .replace(/\n+$/, ""),
         );
         const cls = lang ? ` class="language-${escapeAttr(lang)}"` : "";
         return `<pre><code${cls}>${code}</code></pre>`;
@@ -1564,13 +1484,6 @@ export function convertProseMirrorToMarkdown(
     const indent = " ".repeat(indentWidth);
     const lines: string[] = [];
     childStrings.forEach((child, childIndex) => {
-      // Separate consecutive block children with a BLANK line so the item is a
-      // CommonMark "loose" list item and each block stays its own node. Without
-      // it, a second paragraph (`- a\n  b`) is re-parsed as a lazy continuation
-      // of the first and the two merge into one paragraph — silent data loss.
-      // The blank line still sits INSIDE the item (the following block keeps the
-      // continuation indent), so nested lists/code blocks remain nested.
-      if (childIndex > 0) lines.push("");
       child.split("\n").forEach((line, lineIndex) => {
         if (childIndex === 0 && lineIndex === 0) {
           // First physical line of the first block gets the marker.
@@ -1587,9 +1500,7 @@ export function convertProseMirrorToMarkdown(
 
   const processListItem = (item: any, prefix: string): string => {
     const itemContent = item.content || [];
-    // A `<!-- -->` entry separates two adjacent same-family sublists inside the
-    // item so they do not merge on re-parse (see renderBlockChildren).
-    const childStrings = renderBlockChildren(itemContent, processNode);
+    const childStrings = itemContent.map(processNode);
     if (childStrings.length === 0) return prefix;
     // The rendered marker is `${prefix} ` (prefix + one space), so its width —
     // and thus the continuation indent — is prefix.length + 1. This is correct
@@ -1603,9 +1514,7 @@ export function convertProseMirrorToMarkdown(
     const checkbox = checked ? "[x]" : "[ ]";
     const prefix = `- ${checkbox}`;
     const itemContent = item.content || [];
-    // A `<!-- -->` entry separates two adjacent same-family sublists inside the
-    // item so they do not merge on re-parse (see renderBlockChildren).
-    const childStrings = renderBlockChildren(itemContent, processNode);
+    const childStrings = itemContent.map(processNode);
     // An empty task item still needs its checkbox marker; without this guard
     // the indent below produces "" and the "- [ ]"/"- [x]" row disappears.
     if (childStrings.length === 0) return prefix;

packages/prosemirror-markdown/src/lib/markdown-to-prosemirror.ts

@@ -270,10 +270,9 @@ const CALLOUT_CLOSE_RE = /^:::\s*$/;
  * optional title after the type is allowed but ignored (the Docmost callout
  * schema has no title). The body is the following contiguous blockquote lines.
  */
-// The callout's own `>` marker may be preceded by an ENCLOSING container prefix:
-// list-item indentation (`  `) and/or blockquote markers (`> `). Group 1 captures
-// that prefix (lazily, so the LAST `>` before `[!type]` is the callout's own).
-const CALLOUT_BQ_OPEN_RE = /^([>\s]*?)>\s*\[!(\w+)\]/;
+const CALLOUT_BQ_OPEN_RE = /^>\s*\[!(\w+)\]/;
+/** Matches any blockquote continuation line (`>` … ). */
+const BLOCKQUOTE_LINE_RE = /^>/;
 /** Matches the start/end of a code fence (``` or ~~~), capturing the marker. */
 const CODE_FENCE_RE = /^(\s*)(`{3,}|~{3,})/;
 
@@ -403,47 +402,20 @@ async function preprocessCallouts(markdown: string): Promise<string> {
       // recurse so nested callouts (`> > [!type]`) are handled, then emit the same
       // callout div the `:::` path produces. A normal blockquote (no `[!type]` on
       // its first line) does not match and stays a blockquote.
-      //
-      // PREFIX-aware: a callout nested inside a list item and/or a blockquote is
-      // serialized with the enclosing container prefix in front of its own `>`
-      // marker — `  > [!type]` (list indent) or `> > [!type]` (blockquote). We
-      // capture that prefix, take only continuation lines carrying `prefix>`, strip
-      // it, and re-apply the prefix to the emitted HTML block so the callout div
-      // stays WITHIN its container (an unprefixed div would escape and re-parse as
-      // a top-level callout / plain blockquote — silent structure loss).
       const bqOpen = line.match(CALLOUT_BQ_OPEN_RE);
       if (bqOpen) {
-        const prefix = bqOpen[1];
-        const type = bqOpen[2].toLowerCase();
-        const cont = prefix + ">"; // a body line = prefix + the callout's own `>`
+        const type = bqOpen[1].toLowerCase();
         const bodyLines: string[] = [];
         let j = i + 1;
         for (; j < lines.length; j++) {
-          if (!lines[j].startsWith(cont)) break;
-          // Drop the prefix + `>` + one optional space, leaving the body content.
-          bodyLines.push(lines[j].slice(prefix.length).replace(/^>\s?/, ""));
+          if (!BLOCKQUOTE_LINE_RE.test(lines[j])) break;
+          bodyLines.push(lines[j].replace(/^>\s?/, ""));
         }
         const inner = await transform(bodyLines);
         const renderedInner = await markedInstance.parse(inner);
-        const block = `<div data-type="callout" data-callout-type="${type}">${renderedInner}</div>`;
-        if (prefix.length === 0) {
-          // Top-level callout: blank lines isolate the HTML block.
-          out.push(`\n${block}\n`);
-        } else if (prefix.includes(">")) {
-          // Enclosing BLOCKQUOTE: prefix every line and add NO surrounding blank
-          // lines — a blank line would terminate the blockquote and split the
-          // callout out of it.
-          out.push(block.split("\n").map((l) => prefix + l).join("\n"));
-        } else {
-          // Pure LIST-ITEM indentation: re-indent and keep the blank-line
-          // separators (a loose list item), so the div sits at the marker column.
-          out.push(
-            `\n${block
-              .split("\n")
-              .map((l) => (l.length ? prefix + l : l))
-              .join("\n")}\n`,
-          );
-        }
+        out.push(
+          `\n<div data-type="callout" data-callout-type="${type}">${renderedInner}</div>\n`,
+        );
         i = j;
         continue;
       }
@@ -625,40 +597,6 @@ function bridgeTaskLists(html: string): string {
  * (null from parseAttachedComment), an unknown name, a wrong-position comment, or
  * an unknown/empty attr value is ignored.
  */
-/**
- * A directive comment is in ATTACHED position when it sits inside a `<p>`/`<hN>`
- * textblock — bound to that block's text (the `attrs`/`img` conventions). Every
- * other parent (body, document level, a block container like blockquote/details/
- * li/column div) is STANDALONE position, where a lone-block directive
- * (subpages/pagebreak/pageembed/transclusion) is materialized. Broadening
- * standalone beyond body/document is what lets these nodes survive NESTED inside
- * a blockquote/callout/details/list item (previously dropped -> silent data loss).
- */
-function isAttachedPosition(tag: string): boolean {
-  return tag === "p" || /^h[1-6]$/.test(tag);
-}
-
-/**
- * Place a materialized standalone-directive element in the DOM: replace the
- * comment IN PLACE when it has a real element parent inside <body> (body itself
- * or a nested block container), preserving document order; queue it as a leading
- * div only when the comment is at document level (no parentElement) or directly
- * under `<html>` (outside <body>, which `document.body.innerHTML` would drop).
- */
-function placeStandalone(
-  comment: any,
-  el: any,
-  tag: string,
-  leadingDivs: any[],
-): void {
-  if (comment.parentElement && tag !== "html") {
-    comment.replaceWith(el);
-  } else {
-    comment.remove();
-    leadingDivs.push(el);
-  }
-}
-
 function applyCommentDirectives(html: string): string {
   // Cheap early-out: no comments at all -> nothing to intercept.
   if (!html.includes("<!--")) return html;
@@ -726,13 +664,13 @@ function applyCommentDirectives(html: string): string {
 
     if (parsed.name === "subpages" || parsed.name === "pagebreak") {
       // #293 canon #5 STANDALONE machinery. A lone comment line is rendered by
-      // marked as its own HTML block; the parser places it under <body>, at
-      // document level (leading), or — when the directive is NESTED — inside a
-      // block CONTAINER (`<blockquote>` for blockquote/callout, `<details>`,
-      // `<li>`, a column `<div>`, …). All of those are STANDALONE position. Only a
-      // comment ATTACHED inside a `<p>`/`<hN>` (bound to that block's text) is
-      // attached position -> INERT.
-      if (isAttachedPosition(tag)) continue; // wrong position -> inert
+      // marked as an HTML block; the parser places it either directly under
+      // <body> (when other content surrounds it) or at document level (when it
+      // leads the output). Both are STANDALONE position. A `subpages`/`pagebreak`
+      // comment sitting inside a `<p>`/`<hN>` (or any other element) is attached
+      // position -> INERT.
+      const standalone = tag === "" || tag === "body" || tag === "html";
+      if (!standalone) continue; // wrong position -> inert
       const div = document.createElement("div");
       if (parsed.name === "pagebreak") {
         div.setAttribute("data-type", "pageBreak");
@@ -742,18 +680,26 @@ function applyCommentDirectives(html: string): string {
           div.setAttribute("data-recursive", "true");
         }
       }
-      placeStandalone(comment, div, tag, leadingDivs);
+      if (tag === "body") {
+        // In-body: replace in place so surrounding content keeps its order.
+        comment.replaceWith(div);
+      } else {
+        // Document-level (leading): drop the stray comment and queue the div to
+        // be prepended into body below.
+        comment.remove();
+        leadingDivs.push(div);
+      }
       continue;
     }
 
     if (parsed.name === "pageembed" || parsed.name === "transclusion") {
       // #293 canon #8 STANDALONE media. Like subpages/pagebreak: a lone comment
-      // line placed under <body>, at document level (leading), or NESTED inside a
-      // block container (blockquote/callout/details/li/column). An ATTACHED-
-      // position comment (inside a `<p>`/`<hN>`) is INERT. We rebuild the schema
-      // div the raw-HTML path emits (media-html.ts) from the decoded attrs so
-      // serialize/parse stay in sync.
-      if (isAttachedPosition(tag)) continue; // wrong position -> inert
+      // line placed under <body> or at document level (leading). An attached-
+      // position comment (inside a <p>/<hN> with a sibling) is INERT. We rebuild
+      // the schema div the raw-HTML path emits (media-html.ts) from the decoded
+      // attrs so serialize/parse stay in sync.
+      const standalone = tag === "" || tag === "body" || tag === "html";
+      if (!standalone) continue; // wrong position -> inert
       const el = buildElement(
         parsed.name === "pageembed"
           ? pageEmbedToHtml({ sourcePageId: parsed.attrs.sourcePageId })
@@ -763,7 +709,12 @@ function applyCommentDirectives(html: string): string {
             }),
       );
       if (!el) continue; // defensive: builder always yields an element
-      placeStandalone(comment, el, tag, leadingDivs);
+      if (tag === "body") {
+        comment.replaceWith(el);
+      } else {
+        comment.remove();
+        leadingDivs.push(el);
+      }
       continue;
     }
 
@@ -855,36 +806,18 @@ function applyCommentDirectives(html: string): string {
 
     if (!parent) continue; // attrs comment must have an element parent
     if (parsed.name !== "attrs") continue; // unknown name -> inert
-    const align = parsed.attrs.textAlign;
     // #293 canon #9 ATTACHED attrs: honored only in attached position.
-    if (tag === "p" || /^h[1-6]$/.test(tag)) {
-      // A real <p>/<hN> host (loose list item, top-level block, …): re-express as
-      // an inline style; the schema's textAlign parseHTML reads `el.style.textAlign`
-      // back onto the paragraph/heading node.
-      if (typeof align === "string" && align) parent.style.textAlign = align;
-      comment.remove();
-    } else if (tag === "li" || tag === "td" || tag === "th") {
-      // TIGHT list item / GFM table cell: marked emits the paragraph's inline
-      // content DIRECTLY inside the <li>/<td>/<th> with NO <p> wrapper, so there
-      // is no element to carry the style — generateJSON materializes the
-      // paragraph later. Wrap the host's LEADING inline content (everything up to
-      // the comment; any trailing block child such as a nested list stays put) in
-      // a <p> carrying the alignment, so the materialized paragraph re-reads it.
-      if (typeof align === "string" && align) {
-        const p = document.createElement("p");
-        p.style.textAlign = align;
-        while (parent.firstChild && parent.firstChild !== comment) {
-          p.appendChild(parent.firstChild);
-        }
-        parent.insertBefore(p, comment);
-      }
-      comment.remove();
-    } else {
-      // Misplaced `attrs` comment (not a textblock/li/cell host): inert. Consume
-      // it anyway so no attached marker ever survives into the parsed body
-      // (matches the pre-existing "consume regardless" behaviour).
-      comment.remove();
+    const isBlock = tag === "p" || /^h[1-6]$/.test(tag);
+    if (!isBlock) continue; // misplaced comment -> inert
+    const align = parsed.attrs.textAlign;
+    if (typeof align === "string" && align) {
+      // Re-express as an inline style; the schema's textAlign parseHTML reads
+      // `el.style.textAlign` back onto the paragraph/heading node.
+      parent.style.textAlign = align;
     }
+    // Consume the marker regardless (unknown keys are simply ignored) so no
+    // attached comment ever survives into the parsed body.
+    comment.remove();
   }
   // Prepend any document-level (leading) standalone divs into body, preserving
   // their document order relative to each other and ahead of existing content.

packages/prosemirror-markdown/src/lib/media-html.ts

@@ -46,11 +46,7 @@ export function videoToHtml(attrs: Record<string, any>): string {
   if (attrs.width != null) parts.push(`width="${escapeAttr(attrs.width)}"`);
   if (attrs.height != null) parts.push(`height="${escapeAttr(attrs.height)}"`);
   if (attrs.size != null) parts.push(`data-size="${escapeAttr(attrs.size)}"`);
-  // align default is "center" (schema): OMIT it so a bare/center node stays
-  // clean and parse's re-materialized "center" default is not a P2 churn — only
-  // a genuinely non-default left/right emits data-align (mirrors imageToHtml).
-  if (attrs.align && attrs.align !== "center")
-    parts.push(`data-align="${escapeAttr(attrs.align)}"`);
+  if (attrs.align) parts.push(`data-align="${escapeAttr(attrs.align)}"`);
   if (attrs.aspectRatio != null)
     parts.push(`data-aspect-ratio="${escapeAttr(attrs.aspectRatio)}"`);
   return `<div><video ${parts.join(" ")}></video></div>`;
@@ -66,9 +62,7 @@ export function youtubeToHtml(attrs: Record<string, any>): string {
     parts.push(`data-width="${escapeAttr(attrs.width)}"`);
   if (attrs.height != null)
     parts.push(`data-height="${escapeAttr(attrs.height)}"`);
-  // "center" is the schema default -> omit (see videoToHtml rationale).
-  if (attrs.align && attrs.align !== "center")
-    parts.push(`data-align="${escapeAttr(attrs.align)}"`);
+  if (attrs.align) parts.push(`data-align="${escapeAttr(attrs.align)}"`);
   return `<div ${parts.join(" ")}></div>`;
 }
 
@@ -103,9 +97,7 @@ export function diagramToHtml(
   if (attrs.size != null) parts.push(`data-size="${escapeAttr(attrs.size)}"`);
   if (attrs.aspectRatio != null)
     parts.push(`data-aspect-ratio="${escapeAttr(attrs.aspectRatio)}"`);
-  // "center" is the schema default -> omit (see videoToHtml rationale).
-  if (attrs.align && attrs.align !== "center")
-    parts.push(`data-align="${escapeAttr(attrs.align)}"`);
+  if (attrs.align) parts.push(`data-align="${escapeAttr(attrs.align)}"`);
   if (attrs.attachmentId)
     parts.push(`data-attachment-id="${escapeAttr(attrs.attachmentId)}"`);
   return `<div ${parts.join(" ")}></div>`;
@@ -118,18 +110,10 @@ export function embedToHtml(attrs: Record<string, any>): string {
     `data-src="${escapeAttr(attrs.src ?? "")}"`,
     `data-provider="${escapeAttr(attrs.provider ?? "")}"`,
   ];
-  // "center" is the schema default -> omit (see videoToHtml rationale).
-  if (attrs.align && attrs.align !== "center")
-    parts.push(`data-align="${escapeAttr(attrs.align)}"`);
-  // embed width/height default to the NUMBERS 800/600 (schema). Getting a data-
-  // attribute back always yields a STRING, so emitting the default here would
-  // round-trip 800 -> "800" (a number->string P1 divergence canonicalize does
-  // NOT normalize). OMIT the defaults so parse re-materializes the numeric
-  // default instead — mirrors the top-level embed path (markdown-converter.ts),
-  // which also emits width/height only when they differ from 800/600.
-  if (attrs.width != null && attrs.width !== 800)
+  if (attrs.align) parts.push(`data-align="${escapeAttr(attrs.align)}"`);
+  if (attrs.width != null)
     parts.push(`data-width="${escapeAttr(attrs.width)}"`);
-  if (attrs.height != null && attrs.height !== 600)
+  if (attrs.height != null)
     parts.push(`data-height="${escapeAttr(attrs.height)}"`);
   return `<div ${parts.join(" ")}></div>`;
 }

packages/prosemirror-markdown/test/generative/doc-generator.ts

@@ -1,392 +0,0 @@
-/**
- * Nested whole-document generator (#351, PR 2) — "variant A": a random walk over
- * the schema's ContentMatch automaton.
- *
- * Where the FLAT generator (node-generators.ts) emits `{doc:[ <one target> ]}`,
- * this module produces arbitrarily DEEP, valid ProseMirror documents. Validity is
- * NOT hand-asserted: it comes straight from the schema. To fill a container's
- * block content we start at `nodeType.contentMatch` and walk the automaton —
- * enumerate the legal next node types (`match.edge(i).type`), let fast-check pick
- * one (or STOP once `match.validEnd`), generate that child RECURSIVELY, then
- * advance the automaton via `match.matchType(childType)`. A bad walk therefore
- * cannot emit a structurally-invalid doc (the `generator validity` test guards
- * this with `schema.nodeFromJSON(json).check()`).
- *
- * ── What the walk drives, and what it delegates ──────────────────────────────
- * The walk owns BLOCK STRUCTURE (which blocks nest inside which containers, in
- * what order and how deep). Two things it deliberately delegates, to avoid
- * REGRESSING the byte-stable space the flat suite already proved empirically:
- *
- *   - INLINE content of textblocks (paragraph/heading/codeBlock/detailsSummary)
- *     is filled from text-arbitraries.ts — the exact hostile-but-byte-stable
- *     inline corpus the flat suite established. Walking the inline ContentMatch
- *     instead would re-derive (and re-fail) the text-space limitations the flat
- *     suite already pins, which is not this generator's job.
- *   - Node ATTRS come from nodeAttrsArb(type, 'p1') — the round-trip-safe
- *     attribute space. Attribute-degenerate fuzzing (P2/P3 over 'fuzz') is the
- *     flat suite's concern; the nested suite isolates STRUCTURAL round-trip, so
- *     it stays in 'p1' and does not re-litigate frozen/pinned attributes.
- *
- * ── Coordination the automaton cannot express ────────────────────────────────
- * A ContentMatch guarantees a child SEQUENCE is legal, but not cross-sibling
- * invariants. Two nodes need coordination the walk injects by hand:
- *   - `table`: GFM needs a RECTANGULAR grid with column-consistent alignment.
- *     The automaton happily allows ragged rows / per-cell align, which are not a
- *     converter bug — just a malformed table. So a table is generated ATOMICALLY
- *     (same shape the flat suite proved) rather than walked.
- *   - `columns`: the `layout` attr must agree with the column COUNT. We pick the
- *     layout, derive the count, then walk each column's block body normally — so
- *     columns still gain real nested content, only the count is coordinated.
- *
- * ── Excluded from the nested walk ────────────────────────────────────────────
- * Footnote nodes (footnoteReference / footnotesList / footnoteDefinition) need a
- * DOCUMENT-GLOBAL id match between a reference and its definition. That
- * coordination is owned by the flat suite's `footnotes` generator; placing them
- * independently here would fabricate id mismatches that look like converter bugs
- * but are generator defects. They are filtered out of the walk (documented in
- * EXCLUDED). The completeness contract lives in the FLAT suite and is unaffected.
- *
- * ── Termination / budgets ────────────────────────────────────────────────────
- * Two bounds keep every doc finite and the suite fast:
- *   - MAX_DEPTH — a hard cap on block-nesting depth. A precomputed `minDepth`
- *     fixpoint (the minimum extra nesting a subtree of each type needs to be
- *     valid) lets the walk pick a CONTAINER child only when there is depth
- *     headroom to complete it — so the walk can never paint itself into a corner
- *     where a required child cannot fit (no invalid docs, guaranteed termination).
- *   - NODE_BUDGET — a soft cap on total nodes; as it runs low the walk biases
- *     toward STOP (when validEnd) or toward cheap terminating children.
- */
-import fc from 'fast-check';
-import { getSchema } from '@tiptap/core';
-import { docmostExtensions } from '../../src/lib/docmost-schema.js';
-import { nodeAttrsArb } from './attr-arbitraries.js';
-import {
-  inlineContentArb,
-  headingInlineContentArb,
-  plainInlineContentArb,
-  phraseArb,
-} from './text-arbitraries.js';
-
-/** The exact ProseMirror schema the converter targets (built per the issue). */
-export const schema = getSchema(docmostExtensions as never);
-
-/** Hard cap on block-nesting depth (doc = depth 0). Kept in the issue's 4–5 band. */
-export const MAX_DEPTH = 4;
-/**
- * Soft cap on total node count per generated document. Kept moderate: every P1/P2
- * run parses the emitted markdown through jsdom (heavy), so 100+ node docs across
- * hundreds of runs exhaust the worker heap. 60 still yields deeply-nested docs
- * (depth 4) while keeping the suite within memory.
- */
-export const NODE_BUDGET = 60;
-
-/**
- * Nodes kept OUT of the nested walk: footnote nodes need a doc-global id match a
- * local walk cannot coordinate (owned by the flat suite's `footnotes` generator).
- */
-const EXCLUDED = new Set<string>([
-  'footnoteReference',
-  'footnotesList',
-  'footnoteDefinition',
-]);
-
-/**
- * Structural-only children that carry `group: "block"` in the schema and so leak
- * into EVERY block container's ContentMatch, even though the editor only ever
- * places them inside their one true parent. Choosing them freely (e.g. a bare
- * `column` at the document root) fabricates documents no editor produces and that
- * the converter is not designed to round-trip — a GENERATOR artifact, not a
- * converter bug. They are admitted ONLY when the container being filled is their
- * dedicated parent. (`column` is in fact always built inside columnsArb, so this
- * just double-guards it.)
- */
-const DEDICATED_PARENT: Record<string, string> = {
-  column: 'columns',
-  detailsSummary: 'details',
-  detailsContent: 'details',
-};
-
-/** Is child type `t` legal as a freely-chosen child of container `parentType`? */
-function childAllowedUnder(t: string, parentType: string): boolean {
-  const dedicated = DEDICATED_PARENT[t];
-  return dedicated === undefined || dedicated === parentType;
-}
-
-/** Textblock (inlineContent) types — filled from the proven inline corpus. */
-function isTextblock(typeName: string): boolean {
-  return !!schema.nodes[typeName]?.isTextblock;
-}
-/** Leaf/atom types — no content, only generated attrs. */
-function isLeaf(typeName: string): boolean {
-  return !!schema.nodes[typeName]?.isLeaf;
-}
-
-// ---------------------------------------------------------------------------
-// minDepth fixpoint: the minimum EXTRA block-nesting depth a valid subtree
-// rooted at each node type requires. Leaves and textblocks need 0 (a textblock
-// is satisfied by inline content, no block recursion). A container needs
-// 1 + the cheapest way to satisfy its ContentMatch. Computed as a min–max path
-// to `validEnd` over the automaton, iterated to a fixpoint over node types.
-// ---------------------------------------------------------------------------
-
-/** Cheapest (min over reachable validEnd of max child minDepth) to complete a match. */
-function minCompletion(
-  match: any,
-  md: Record<string, number>,
-  seen: Set<any>,
-  parentType: string,
-): number {
-  let best = match.validEnd ? 0 : Infinity;
-  if (seen.has(match)) return best; // a cycle never completes more cheaply
-  seen.add(match);
-  for (let i = 0; i < match.edgeCount; i++) {
-    const edge = match.edge(i);
-    const t = edge.type.name;
-    if (EXCLUDED.has(t) || t === 'text') continue;
-    if (!childAllowedUnder(t, parentType)) continue;
-    const childCost = md[t];
-    if (childCost === undefined || childCost === Infinity) continue;
-    const rest = minCompletion(edge.next, md, seen, parentType);
-    if (rest === Infinity) continue;
-    best = Math.min(best, Math.max(childCost, rest));
-  }
-  seen.delete(match);
-  return best;
-}
-
-function computeMinDepth(): Record<string, number> {
-  const md: Record<string, number> = {};
-  for (const name of Object.keys(schema.nodes)) {
-    md[name] = isLeaf(name) || isTextblock(name) ? 0 : Infinity;
-  }
-  let changed = true;
-  while (changed) {
-    changed = false;
-    for (const name of Object.keys(schema.nodes)) {
-      if (md[name] === 0) continue; // leaves/textblocks fixed at 0
-      const nt: any = schema.nodes[name];
-      const completion = minCompletion(nt.contentMatch, md, new Set(), name);
-      const next = completion === Infinity ? Infinity : 1 + completion;
-      if (next < md[name]) {
-        md[name] = next;
-        changed = true;
-      }
-    }
-  }
-  return md;
-}
-
-const MIN_DEPTH = computeMinDepth();
-
-// ---------------------------------------------------------------------------
-// Leaf / textblock builders (attrs from 'p1', inline from the proven corpus).
-// ---------------------------------------------------------------------------
-
-function attachAttrs(typeName: string, base: Record<string, unknown> = {}) {
-  return nodeAttrsArb(typeName, 'p1', base).map((attrs) => {
-    const node: any = { type: typeName };
-    if (Object.keys(attrs).length) node.attrs = attrs;
-    return node;
-  });
-}
-
-/** A leaf/atom block: attrs only, no content. */
-function leafArb(typeName: string): fc.Arbitrary<any> {
-  return attachAttrs(typeName);
-}
-
-/** A textblock, inline content taken from the byte-stable flat corpus. */
-function textblockArb(typeName: string): fc.Arbitrary<any> {
-  if (typeName === 'codeBlock') {
-    return fc
-      .tuple(
-        nodeAttrsArb('codeBlock', 'p1'),
-        // Fenced code re-imports with a TRAILING NEWLINE (flat suite finding);
-        // author it so the doc is already at the round-trip fixpoint.
-        fc.array(phraseArb, { minLength: 1, maxLength: 3 }).map((l) => l.join('\n') + '\n'),
-      )
-      .map(([attrs, code]) => ({
-        type: 'codeBlock',
-        ...(Object.keys(attrs).length ? { attrs } : {}),
-        content: [{ type: 'text', text: code }],
-      }));
-  }
-  const inline =
-    typeName === 'heading'
-      ? headingInlineContentArb
-      : typeName === 'detailsSummary'
-        ? plainInlineContentArb
-        : inlineContentArb;
-  return fc
-    .tuple(nodeAttrsArb(typeName, 'p1'), inline)
-    .map(([attrs, content]) => ({
-      type: typeName,
-      ...(Object.keys(attrs).length ? { attrs } : {}),
-      content,
-    }));
-}
-
-// ---------------------------------------------------------------------------
-// Coordinated builders: table (atomic, rectangular, column-consistent align)
-// and columns (layout coupled to count, bodies walked).
-// ---------------------------------------------------------------------------
-
-/** A rectangular GFM-safe table (mirrors the flat suite's proven shape). */
-function tableArb(): fc.Arbitrary<any> {
-  return fc.integer({ min: 1, max: 3 }).chain((cols) => {
-    // One alignment per COLUMN, identical on header + every body cell, so the
-    // second export cannot re-align and churn.
-    const alignsArb = fc.array(fc.constantFrom(undefined, 'left', 'center', 'right'), {
-      minLength: cols,
-      maxLength: cols,
-    });
-    const cell = (header: boolean, align?: string) =>
-      phraseArb.map((t) => ({
-        type: header ? 'tableHeader' : 'tableCell',
-        attrs: { colspan: 1, rowspan: 1, ...(align ? { align } : {}) },
-        content: [{ type: 'paragraph', content: [{ type: 'text', text: t }] }],
-      }));
-    return alignsArb.chain((aligns) => {
-      const headerRow = fc
-        .tuple(...aligns.map((a) => cell(true, a)))
-        .map((cells) => ({ type: 'tableRow', content: cells }));
-      const bodyRow = fc
-        .tuple(...aligns.map((a) => cell(false, a)))
-        .map((cells) => ({ type: 'tableRow', content: cells }));
-      return fc
-        .tuple(headerRow, fc.array(bodyRow, { minLength: 1, maxLength: 2 }))
-        .map(([h, body]) => ({ type: 'table', content: [h, ...body] }));
-    });
-  });
-}
-
-/** A columns block: layout ↔ count coupled, each column body walked as blocks. */
-function columnsArb(depth: number, budget: number): fc.Arbitrary<any> {
-  return fc
-    .constantFrom('two_equal', 'three_equal', 'left_sidebar', 'right_sidebar')
-    .chain((layout) => {
-      const count = layout === 'three_equal' ? 3 : 2;
-      const columnType: any = schema.nodes.column;
-      // Split the remaining budget across the fixed number of columns.
-      const per = Math.max(2, Math.floor((budget - 1) / count));
-      return nodeAttrsArb('columns', 'p1', { layout, widthMode: 'normal' }).chain((attrs) =>
-        fc
-          .tuple(
-            ...Array.from({ length: count }, () =>
-              fillMatch(columnType.contentMatch, depth + 1, per, 'column').map(({ children }) => ({
-                type: 'column',
-                content: children,
-              })),
-            ),
-          )
-          .map((cols) => ({ type: 'columns', attrs, content: cols })),
-      );
-    });
-}
-
-// ---------------------------------------------------------------------------
-// The ContentMatch walk.
-// ---------------------------------------------------------------------------
-
-/** Count every node in a subtree (block + inline), for budget accounting. */
-function countNodes(node: any): number {
-  let n = 1;
-  for (const c of node.content ?? []) n += countNodes(c);
-  return n;
-}
-
-/** Build a single child node of a given type at `depth`, within `budget`. */
-function blockNode(typeName: string, depth: number, budget: number): fc.Arbitrary<any> {
-  if (typeName === 'table') return tableArb();
-  if (typeName === 'columns') return columnsArb(depth, budget);
-  if (isTextblock(typeName)) return textblockArb(typeName);
-  if (isLeaf(typeName)) return leafArb(typeName);
-  // Generic container: attrs from 'p1', block content from the automaton walk.
-  const nt: any = schema.nodes[typeName];
-  return nodeAttrsArb(typeName, 'p1').chain((attrs) =>
-    fillMatch(nt.contentMatch, depth, budget - 1, typeName).map(({ children }) => {
-      const node: any = { type: typeName };
-      if (Object.keys(attrs).length) node.attrs = attrs;
-      if (children.length) node.content = children;
-      return node;
-    }),
-  );
-}
-
-/**
- * Fill a container's block content by walking its ContentMatch automaton from
- * `match`. Returns the children array plus the budget left after them.
- */
-function fillMatch(
-  match: any,
-  depth: number,
-  budget: number,
-  parentType: string,
-): fc.Arbitrary<{ children: any[]; budget: number }> {
-  const canStop = match.validEnd;
-  // A child lives at depth+1; only pick it if its subtree can complete within
-  // MAX_DEPTH. This headroom rule is what makes the walk deadlock-free.
-  const headroom = MAX_DEPTH - (depth + 1);
-  const edges: { t: string; next: any }[] = [];
-  if (headroom >= 0) {
-    for (let i = 0; i < match.edgeCount; i++) {
-      const edge = match.edge(i);
-      const t = edge.type.name;
-      if (EXCLUDED.has(t) || t === 'text') continue;
-      if (!childAllowedUnder(t, parentType)) continue;
-      if ((MIN_DEPTH[t] ?? Infinity) > headroom) continue;
-      edges.push({ t, next: edge.next });
-    }
-  }
-
-  // Decide the next action: STOP (if allowed) or extend with one more child.
-  // Bias toward stopping when the budget is spent; force a child only when the
-  // match is not yet at a valid end.
-  const pool: { weight: number; arbitrary: fc.Arbitrary<{ t: string; next: any } | null> }[] = [];
-  const canGo = edges.length > 0 && (budget > 0 || !canStop);
-  if (canStop) {
-    // Stop is weighted higher when the budget is low so docs stay bounded.
-    pool.push({ weight: budget > 0 ? 2 : 5, arbitrary: fc.constant(null) });
-  }
-  if (canGo && !(canStop && budget <= 0)) {
-    pool.push({ weight: 3, arbitrary: fc.constantFrom(...edges) });
-  }
-  // Forced continuation: not a valid end yet and (budget exhausted) — must place
-  // a mandatory child regardless of budget.
-  if (pool.length === 0) {
-    if (edges.length > 0) {
-      pool.push({ weight: 1, arbitrary: fc.constantFrom(...edges) });
-    } else {
-      // No legal child and not required to place one: stop with what we have.
-      return fc.constant({ children: [], budget });
-    }
-  }
-
-  return fc.oneof(...pool).chain((choice) => {
-    if (choice === null) return fc.constant({ children: [], budget });
-    return blockNode(choice.t, depth + 1, budget).chain((node) => {
-      const cost = countNodes(node);
-      return fillMatch(choice.next, depth, budget - cost, parentType).map(
-        ({ children, budget: left }) => ({
-          children: [node, ...children],
-          budget: left,
-        }),
-      );
-    });
-  });
-}
-
-/**
- * The nested-document arbitrary: a valid, arbitrarily-deep ProseMirror doc built
- * by walking the schema from the document root. Attrs stay in the round-trip-safe
- * 'p1' space; inline content reuses the byte-stable flat corpus.
- */
-export const docArb: fc.Arbitrary<any> = fillMatch(
-  schema.nodes.doc.contentMatch,
-  0,
-  NODE_BUDGET,
-  'doc',
-).map(({ children }) => ({ type: 'doc', content: children }));
-
-/** The precomputed minDepth table, exported for inspection/debugging. */
-export { MIN_DEPTH };

packages/prosemirror-markdown/test/generative/nested-roundtrip.property.test.ts

@@ -1,185 +0,0 @@
-import { describe, expect, it, vi } from 'vitest';
-import fc from 'fast-check';
-// Real converters. Importing markdownToProseMirror (transitively, via index)
-// mutates the global DOM via jsdom at module load — expected, required for
-// @tiptap/html's generateJSON under Node (same as the flat sibling suite).
-import {
-  convertProseMirrorToMarkdown,
-  markdownToProseMirror,
-  docsCanonicallyEqual,
-  canonicalizeContent,
-} from '../../src/lib/index.js';
-import { firstDivergence } from '../roundtrip-helpers.js';
-import { schema, docArb } from './doc-generator.js';
-
-// Each run does a real convert + jsdom parse; give ample headroom so the suite
-// is deterministic under parallel worker load (matching the flat sibling suite).
-vi.setConfig({ testTimeout: 60000 });
-
-// ---------------------------------------------------------------------------
-// #351 PR 2 — GENERATIVE round-trip over NESTED (whole-document) docs produced
-// by the ContentMatch random walk (doc-generator.ts). The invariants mirror the
-// flat suite, plus a parser-fuzz totality property (P4):
-//
-//   P1 — semantic round-trip: docsCanonicallyEqual(mdToPm(pmToMd(d)), d)
-//   P2 — byte fixpoint (2nd pass): pmToMd(mdToPm(pmToMd(d))) === pmToMd(d)
-//        (the FIRST pass may normalize once; the SECOND pass must be a fixpoint)
-//   P3 — totality: neither converter throws; bounded.
-//   P4 — parser fuzz totality: for ANY string, markdownToProseMirror does NOT
-//        throw and returns a SCHEMA-VALID document.
-//
-// GUARDRAIL: a P1/P2/P3/P4 failure means the generator FOUND A REAL CONVERTER
-// BUG. These invariants are kept STRICT — no it.fails / skip / weakening. A
-// failure prints the shrunk minimal counterexample for triage.
-// ---------------------------------------------------------------------------
-
-const SEED = 20250705;
-// The nested walk builds far heavier docs than the flat suite (each P1/P2 run
-// parses the emitted markdown through jsdom), so keep the run count moderate to
-// hold runtime and worker memory in budget while still exercising deep
-// structures. P4 (cheap string parsing) runs at a higher count below.
-const NUM_RUNS = 100;
-
-const pmToMd = (doc: unknown): string => convertProseMirrorToMarkdown(doc);
-const mdToPm = (md: string): Promise<any> => markdownToProseMirror(md);
-
-async function roundTrip(doc: unknown): Promise<{ md1: string; md2: string; doc2: any }> {
-  const md1 = pmToMd(doc);
-  const doc2 = await mdToPm(md1);
-  const md2 = pmToMd(doc2);
-  return { md1, md2, doc2 };
-}
-
-describe('#351 nested generative round-trip — generator validity', () => {
-  it('every generated nested doc passes schema.nodeFromJSON(...).check()', () => {
-    // A nested generator that emits an invalid ProseMirror document is a
-    // GENERATOR bug — the ContentMatch walk must only produce schema-valid docs.
-    fc.assert(
-      fc.property(docArb, (doc) => {
-        schema.nodeFromJSON(doc).check(); // throws on an invalid doc
-        return true;
-      }),
-      { numRuns: NUM_RUNS * 2, seed: SEED },
-    );
-  });
-});
-
-// ── STATUS: P1/P2/P3/P4 all GREEN. The nested generator originally surfaced a
-// batch of real converter bugs; all were fixed in the serializer/parser (see the
-// #351 hand-off). For the record, the classes it found and that are now fixed:
-//   • Loose (multi-block) list items / task items / callouts / details bodies were
-//     joined with a single "\n", so every block after the first merged into the
-//     first paragraph on re-parse (silent content loss) — now blank-line separated.
-//   • Paragraph `textAlign` was dropped inside a TIGHT list item (no <p> host).
-//   • A nested codeBlock lost its trailing newline on the raw-HTML path.
-//   • Media (embed/video/youtube/drawio/excalidraw) inside `columns` churned a
-//     default `data-align` and coerced embed's numeric width/height to strings.
-//   • pageBreak / pageEmbed / subpages / transclusion were dropped when nested in
-//     blockquote / callout / details / list item (standalone-comment position).
-//   • Callouts nested in a list item or a blockquote (`  > [!type]` / `> > [!type]`)
-//     were re-parsed as plain blockquotes (prefix-unaware callout preprocessor).
-//   • Two adjacent sibling lists sharing a marker family (bulletList/taskList →
-//     `<ul>`; orderedList → `<ol>`) merged into one list on re-parse — and for the
-//     cross-type case (taskList beside bulletList) the merged `<ul>` LOST every
-//     taskItem checkbox. The serializer now emits a `<!-- -->` separator between
-//     such adjacent lists (markdown-converter.ts renderBlockChildren), so they stay
-//     distinct and round-trip; the generator therefore emits them freely again.
-describe('#351 nested generative round-trip — properties', () => {
-  it('P1 — semantic round-trip: docsCanonicallyEqual(mdToPm(pmToMd(d)), d)', async () => {
-    await fc.assert(
-      fc.asyncProperty(docArb, async (doc) => {
-        const { doc2 } = await roundTrip(doc);
-        if (!docsCanonicallyEqual(doc2, doc)) {
-          const div = firstDivergence(
-            JSON.parse(JSON.stringify(canonicalizeContent(doc2))),
-            JSON.parse(JSON.stringify(canonicalizeContent(doc))),
-          );
-          throw new Error(
-            `P1 divergence @ ${div?.path}: got=${JSON.stringify(div?.a)} want=${JSON.stringify(div?.b)}`,
-          );
-        }
-      }),
-      { numRuns: NUM_RUNS, seed: SEED },
-    );
-  });
-
-  it('P2 — byte fixpoint: pmToMd(mdToPm(pmToMd(d))) === pmToMd(d)', async () => {
-    await fc.assert(
-      fc.asyncProperty(docArb, async (doc) => {
-        const { md1, md2 } = await roundTrip(doc);
-        expect(md2).toBe(md1);
-      }),
-      { numRuns: NUM_RUNS, seed: SEED },
-    );
-  });
-
-  it('P3 — totality: neither converter throws', async () => {
-    await fc.assert(
-      fc.asyncProperty(docArb, async (doc) => {
-        await roundTrip(doc);
-      }),
-      { numRuns: NUM_RUNS, seed: SEED },
-    );
-  });
-});
-
-// ---------------------------------------------------------------------------
-// P4 — parser fuzz. Independent of the doc generator: for ANY input string the
-// PARSER (markdownToProseMirror) must be TOTAL — never throw — and must always
-// return a schema-valid document. The corpus mixes raw unicode strings with
-// strings assembled from markdown-significant fragments (headings, list bullets,
-// fences, pipes, thematic breaks, HTML-ish snippets) to probe the block/inline
-// parsers on hostile but plausible input.
-// ---------------------------------------------------------------------------
-
-const mdFragmentArb: fc.Arbitrary<string> = fc.constantFrom(
-  '# ', '## ', '### ###', '- ', '* ', '+ ', '1. ', '> ', '>> ',
-  '```', '```js', '~~~', '---', '***', '___', '| a | b |', '|---|---|',
-  '[link](http://x)', '![img](http://x)', '**', '__', '~~', '`code`',
-  '<div>', '</div>', '<b>', '<!-- c -->', '<table>', '<br>', '&amp;',
-  '\t', '\n', '  ', '\\', '^[fn]', '[^1]:', '- [ ] ', '- [x] ',
-  '$$', '$x$', ':::', '{.class}', '\u0000', '\uFEFF', '😀', 'مرحبا',
-);
-
-// Full-unicode strings (fast-check v4 replaced fullUnicodeString with the
-// `unit: 'binary'` string option, which draws over the whole code-point range).
-const fullUnicodeStringArb = (max?: number) =>
-  fc.string({ unit: 'binary', ...(max !== undefined ? { maxLength: max } : {}) });
-
-const assembledMarkdownArb: fc.Arbitrary<string> = fc
-  .array(fc.oneof(mdFragmentArb, fc.string(), fullUnicodeStringArb(8)), {
-    minLength: 1,
-    maxLength: 12,
-  })
-  .map((parts) => parts.join(''));
-
-const parserInputArb: fc.Arbitrary<string> = fc.oneof(
-  { weight: 2, arbitrary: fc.string() },
-  { weight: 2, arbitrary: fullUnicodeStringArb() },
-  { weight: 3, arbitrary: assembledMarkdownArb },
-  { weight: 1, arbitrary: fc.array(mdFragmentArb, { minLength: 1, maxLength: 8 }).map((p) => p.join('\n')) },
-);
-
-describe('#351 parser fuzz — totality on arbitrary input (P4)', () => {
-  it('P4 — markdownToProseMirror never throws and always returns a schema-valid doc', async () => {
-    await fc.assert(
-      fc.asyncProperty(parserInputArb, async (s) => {
-        let result: any;
-        try {
-          result = await mdToPm(s);
-        } catch (e: any) {
-          throw new Error(`P4 parser THREW on input ${JSON.stringify(s)}: ${e?.message ?? e}`);
-        }
-        try {
-          schema.nodeFromJSON(result).check();
-        } catch (e: any) {
-          throw new Error(
-            `P4 parser produced an INVALID doc for input ${JSON.stringify(s)}: ${e?.message ?? e}\n` +
-              `doc=${JSON.stringify(result).slice(0, 600)}`,
-          );
-        }
-      }),
-      { numRuns: NUM_RUNS * 2, seed: SEED },
-    );
-  });
-});

packages/prosemirror-markdown/test/markdown-converter-gaps.test.ts

@@ -374,9 +374,7 @@ describe('converter gap coverage — emission branches (specs 1–11)', () => {
         ],
       }),
     );
-    // Block children of a task item are blank-line separated (loose list) per the
-    // #351 fix; the sublist stays at the fixed 2-column continuation indent.
-    expect(out).toBe('- [ ] top\n\n  - child');
+    expect(out).toBe('- [ ] top\n  - child');
   });
 
   // 10. A bulletList inside a blockquote: each list line independently prefixed.

packages/prosemirror-markdown/test/markdown-converter.test.ts

@@ -198,11 +198,7 @@ describe('convertProseMirrorToMarkdown', () => {
         }),
       );
       // First line carries the marker; the nested list is indented 2 columns.
-      // Block children of a list item are separated by a BLANK line (loose list):
-      // this is the #351 fix — a single "\n" let a following block merge into the
-      // first paragraph on re-parse (silent content loss). The blank line stays
-      // inside the item, so the sublist remains nested at the 2-col marker column.
-      expect(out).toBe('- parent\n\n  - child');
+      expect(out).toBe('- parent\n  - child');
     });
 
     it('nested ordered list indents by the wider 3-col marker width', () => {
@@ -223,9 +219,8 @@ describe('convertProseMirrorToMarkdown', () => {
           ],
         }),
       );
-      // "1. " is 3 columns wide, so the continuation indent is 3 spaces. Block
-      // children are blank-line separated (loose list) per the #351 fix.
-      expect(out).toBe('1. parent\n\n   1. child');
+      // "1. " is 3 columns wide, so the continuation indent is 3 spaces.
+      expect(out).toBe('1. parent\n   1. child');
     });
   });
 
@@ -544,12 +539,11 @@ describe('convertProseMirrorToMarkdown', () => {
       );
 
       // The 10th marker is the 4-column "10. "; the nested sublist line must be
-      // indented exactly 4 spaces (prefix.length 3 + 1), NOT 3. Block children are
-      // blank-line separated (loose list) per the #351 fix.
-      expect(out).toContain('10. j\n\n    1. x');
+      // indented exactly 4 spaces (prefix.length 3 + 1), NOT 3.
+      expect(out).toContain('10. j\n    1. x');
       // Guard against the off-by-one (3-space) regression that would re-parse
       // the sublist as loose/sibling content on import.
-      expect(out).not.toContain('10. j\n\n   1. x');
+      expect(out).not.toContain('10. j\n   1. x');
       // And the single-digit items keep the narrower 3-column marker (no body
       // continuation here, but the marker itself must stay "1. ".."9. ").
       expect(out.startsWith('1. a\n2. b\n')).toBe(true);
@@ -586,18 +580,17 @@ describe('convertProseMirrorToMarkdown', () => {
           content: [para(text('line1')), para(text('line2'))],
         }),
       );
-      // The converter emits an Obsidian-native callout: a `> [!type]` opener plus
-      // one `>`-prefixed body line per content line. Block children are separated
-      // by a blank `>` line (#351 fix): a single '\n' let the two paragraphs merge
-      // into one on re-parse. We pin the lowercasing (WARNING -> warning) and the
-      // blank-line-separated multi-child join.
-      expect(out).toBe('> [!warning]\n> line1\n>\n> line2');
+      // NOTE(review): the spec predicted ':::warning\nline1\n\nline2\n:::' (a
+      // The converter joins the callout's rendered children with a single '\n'
+      // and emits an Obsidian-native callout: a `> [!type]` opener plus one
+      // `>`-prefixed body line per content line. We pin the lowercasing
+      // (WARNING -> warning) and the multi-child join.
+      expect(out).toBe('> [!warning]\n> line1\n> line2');
       // The type is lowercased (an uppercase `[!WARNING]` would not re-import).
       expect(out.startsWith('> [!warning]\n')).toBe(true);
       expect(out).not.toContain('[!WARNING]');
-      // Both paragraph children are present, each blockquote-prefixed, blank-`>`
-      // separated so they stay distinct paragraphs on re-parse.
-      expect(out).toContain('> line1\n>\n> line2');
+      // Both paragraph children are present, each blockquote-prefixed.
+      expect(out).toContain('> line1\n> line2');
     });
 
     // Spec 4 — blockquote per-line prefixer over a multi-line nested callout.
@@ -614,11 +607,13 @@ describe('convertProseMirrorToMarkdown', () => {
           ],
         }),
       );
-      // The nested callout renders as an Obsidian callout '> [!info]\n> a\n>\n> b'
-      // (blank-`>` separated children per the #351 fix). The outer blockquote
-      // prefixer then prefixes each of those lines with '> ' again, yielding a
-      // doubly-nested blockquote — the per-line-prefix loop over a multi-line child.
-      expect(out).toBe('> > [!info]\n> > a\n> >\n> > b');
+      // NOTE(review): the spec predicted '> :::info\n> a\n>\n> b\n> :::',
+      // assuming the nested callout body contains a blank line between 'a' and
+      // The nested callout renders as an Obsidian callout '> [!info]\n> a\n> b'
+      // (single-'\n' join, no blank line). The outer blockquote prefixer then
+      // prefixes each of those lines with '> ' again, yielding a doubly-nested
+      // blockquote — the realistic per-line-prefix loop over a multi-line child.
+      expect(out).toBe('> > [!info]\n> > a\n> > b');
       // Every produced line carries the '> ' prefix (no line escapes to col 0).
       for (const line of out.split('\n')) {
         expect(line.startsWith('>')).toBe(true);