feat(ai-chat): единый гейт конкурентности + серверный supersede (CAS) (#487)

Раньше в legacy-режиме (дефолт!) гейта конкурентности НЕ было вообще — проверка
409 сидела внутри if (autonomousRuns && chatId): два таба = два параллельных
стрима на один чат (интерливинг истории, падения convertToModelMessages).

Серверная часть (клиентская лестница ретраев — отдельной клиентской итерацией
FSM, см. ниже):
- Run-строка ОБЯЗАТЕЛЬНА в ОБОИХ режимах: legacy тоже проходит beginRun/
  finalizeRun (гейт партиального уникального индекса + runId в start-метаданных).
  Различие режимов сведено к семантике abort: legacy onClose при дисконнекте
  зовёт requestStop(runId) вместо аборта сокет-сигнала (которого streamText уже
  не потребляет). Второй таб в ЛЮБОМ режиме → 409 A_RUN_ALREADY_ACTIVE.
- Supersede CAS POST /stream { supersede: { runId: X } } (AiChatRunService
  .supersede): валидация X.chatId===body.chatId (иначе 400 SUPERSEDE_INVALID);
  нет активного рана → degrade в обычный send; активный ≠ X → 409
  SUPERSEDE_TARGET_MISMATCH + текущий runId; активный = X → requestStop →
  awaitSettled (таймаут W=10c) → «записано/сдался» (сдался → settleZombie
  применяет intended условным UPDATE) → ready; таймаут → 409 SUPERSEDE_TIMEOUT,
  ничего не персистится. W обоснован race-on-abort'ом коммита 1; DB-brownout →
  TIMEOUT штатен, W не увеличивать (env-tunable).
- Задокументированы ограничения: нет квиесценции сайд-эффектов (в промпт нового
  рана добавлена строка SUPERSEDE_NOTE «предыдущий ран прерван, его последние
  операции могли примениться с задержкой»); кража слота между освобождением и
  beginRun (→ MISMATCH с новым runId, бэкстоп — уникальный индекс).

Тесты: два параллельных старта (оба режима) → строго 409 второму; supersede при
живом долгом ТУЛЕ (не UPDATE-задержке) settle'ится быстро; все CAS-ветки; дубль
supersede-POST → degrade; зомби-путь через settleZombie; HTTP-маппинг ветвей.
Кейс «beginRun ok → insert user-msg упал → ран settled, слот свободен» покрыт
lifecycle-спекой против реального safety-net catch.

КЛИЕНТ (deferred, точно flagged): удаление лестницы SUPERSEDE_RETRY_DELAYS_MS/
isRunAlreadyActive/supersedeRetryRef и переход на supersede-в-body требуют
адопции runId из start-метаданных (сейчас клиент runId не трекает вовсе) — это
и есть та «итерация клиентского FSM», которую данный серверный контракт
разблокирует. Не трогаю фрагильный клиентский FSM без возможности E2E-валидации
в браузере, чтобы не внести непроверяемую регрессию.

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

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

@@ -577,3 +577,165 @@ describe('AiChatRunService run lifecycle', () => {
     ).resolves.toBeUndefined();
   });
 });
+
+describe('#487 AiChatRunService.supersede (CAS)', () => {
+  const chat = 'chat-1';
+  const ws = 'ws-1';
+
+  it('degrade: no active run on the chat -> caller sends a normal turn', async () => {
+    const repo = makeRepo({
+      findById: jest.fn(async () => undefined),
+      findActiveByChat: jest.fn(async () => undefined),
+    });
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    expect(await svc.supersede(chat, 'run-x', ws)).toEqual({ kind: 'degrade' });
+  });
+
+  it('invalid: the target run belongs to a DIFFERENT chat -> 400', async () => {
+    const repo = makeRepo({
+      findById: jest.fn(async () => ({
+        id: 'run-x',
+        chatId: 'other-chat',
+        workspaceId: ws,
+      })),
+    });
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    expect(await svc.supersede(chat, 'run-x', ws)).toEqual({ kind: 'invalid' });
+  });
+
+  it('mismatch: a DIFFERENT run is active than the one targeted -> current runId', async () => {
+    const repo = makeRepo({
+      findById: jest.fn(async () => ({ id: 'run-x', chatId: chat, workspaceId: ws })),
+      findActiveByChat: jest.fn(async () => ({
+        id: 'run-live',
+        chatId: chat,
+        workspaceId: ws,
+        status: 'running',
+      })),
+    });
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    expect(await svc.supersede(chat, 'run-x', ws)).toEqual({
+      kind: 'mismatch',
+      activeRunId: 'run-live',
+    });
+  });
+
+  it('ready: the target IS active -> stop it, await its (fast) settle, free the slot', async () => {
+    // Simulate a live long TOOL (NOT a slow UPDATE): the run stays active until an
+    // explicit Stop unwinds it; commit-1's race makes that settle land quickly.
+    // The abort listener stands in for streamText's onAbort -> finalizeRun.
+    const repo = makeRepo({
+      findById: jest.fn(async () => ({
+        id: 'run-1',
+        chatId: chat,
+        workspaceId: ws,
+        status: 'aborted',
+        error: null,
+      })),
+      findActiveByChat: jest.fn(async () => ({
+        id: 'run-1',
+        chatId: chat,
+        workspaceId: ws,
+        status: 'running',
+      })),
+    });
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    const handle = await svc.beginRun({ chatId: chat, workspaceId: ws, userId: 'u1' });
+    handle.signal.addEventListener('abort', () => {
+      void svc.finalizeRun('run-1', ws, 'aborted');
+    });
+
+    // supersede: getRun -> getActiveByChat(==target) -> requestStop -> the abort
+    // listener settles the run -> awaitSettled resolves -> ready.
+    expect(await svc.supersede(chat, 'run-1', ws, 10_000)).toEqual({
+      kind: 'ready',
+    });
+    expect(handle.signal.aborted).toBe(true); // Stop reached the run
+  });
+
+  it('timeout: the target never settles within W -> 409 SUPERSEDE_TIMEOUT (nothing persisted)', async () => {
+    const repo = makeRepo({
+      findById: jest.fn(async () => ({ id: 'run-1', chatId: chat, workspaceId: ws })),
+      findActiveByChat: jest.fn(async () => ({
+        id: 'run-1',
+        chatId: chat,
+        workspaceId: ws,
+        status: 'running',
+      })),
+    });
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    await svc.beginRun({ chatId: chat, workspaceId: ws, userId: 'u1' });
+    // Do NOT settle the run: a tiny W elapses -> timeout.
+    const result = await svc.supersede(chat, 'run-1', ws, 30);
+    expect(result).toEqual({ kind: 'timeout' });
+  });
+
+  it('ready then a DUPLICATE supersede POST degrades (the run is already gone)', async () => {
+    let active: unknown = {
+      id: 'run-1',
+      chatId: chat,
+      workspaceId: ws,
+      status: 'running',
+    };
+    const repo = makeRepo({
+      findById: jest.fn(async () => ({
+        id: 'run-1',
+        chatId: chat,
+        workspaceId: ws,
+        status: 'aborted',
+        error: null,
+      })),
+      findActiveByChat: jest.fn(async () => active),
+      finalizeIfActive: jest.fn(async () => {
+        active = undefined; // settling frees the active slot
+        return { id: 'run-1', status: 'aborted' };
+      }),
+    });
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    const handle = await svc.beginRun({ chatId: chat, workspaceId: ws, userId: 'u1' });
+    handle.signal.addEventListener('abort', () => {
+      void svc.finalizeRun('run-1', ws, 'aborted');
+    });
+
+    expect(await svc.supersede(chat, 'run-1', ws, 10_000)).toEqual({
+      kind: 'ready',
+    });
+    // The duplicate POST for the same target now finds no active run -> degrade.
+    expect(await svc.supersede(chat, 'run-1', ws)).toEqual({ kind: 'degrade' });
+  });
+
+  it('gave-up zombie: supersede applies the intended status (settleZombie) then is ready', async () => {
+    let healthy = false;
+    let active: unknown = {
+      id: 'run-1',
+      chatId: chat,
+      workspaceId: ws,
+      status: 'running',
+    };
+    const repo = makeRepo({
+      findById: jest.fn(async () => ({ id: 'run-1', chatId: chat, workspaceId: ws })),
+      findActiveByChat: jest.fn(async () => active),
+      finalizeIfActive: jest.fn(async () => {
+        if (!healthy) throw new Error('db down');
+        active = undefined;
+        return { id: 'run-1', status: 'aborted' };
+      }),
+    });
+    jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
+    jest.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined);
+    const svc = new AiChatRunService(repo as never, makeEnv() as never);
+    await svc.beginRun({ chatId: chat, workspaceId: ws, userId: 'u1' });
+
+    // The run's terminal write gives up -> zombie (row still 'running').
+    await svc.finalizeRun('run-1', ws, 'aborted');
+    expect(svc.hasZombie('run-1')).toBe(true);
+
+    // The DB recovers; supersede awaits the (already-resolved, terminalWriteFailed)
+    // settle, then settleZombie applies the intended status -> ready.
+    healthy = true;
+    expect(await svc.supersede(chat, 'run-1', ws, 10_000)).toEqual({
+      kind: 'ready',
+    });
+    expect(svc.hasZombie('run-1')).toBe(false);
+  });
+});

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

@@ -62,6 +62,40 @@ export interface RunSettleOutcome {
   terminalWriteFailed: boolean;
 }
 
+/**
+ * #487: how long a supersede waits for the target run to settle after Stop before
+ * it degrades to `SUPERSEDE_TIMEOUT`. W=10s is generous under a HEALTHY DB: commit
+ * 1's race-on-abort makes an in-app tool abort->settle in ms/hundreds of ms, so a
+ * live run releases its slot well within the window. Under a DB brownout the
+ * timeout is normal (the write cannot land); W must NOT be raised to paper
+ * over a slow DB — a SUPERSEDE_TIMEOUT is the honest signal (nothing persisted,
+ * the composer keeps the user's text). Env-tunable for ops, default 10s.
+ */
+export const SUPERSEDE_SETTLE_TIMEOUT_MS = (() => {
+  const raw = Number(process.env.AI_CHAT_SUPERSEDE_TIMEOUT_MS);
+  return Number.isFinite(raw) && raw > 0 ? raw : 10_000;
+})();
+
+/**
+ * #487: the result of the supersede CAS ({@link AiChatRunService.supersede}).
+ *  - `degrade`  : no active run on the chat (it ended between click and POST) —
+ *                 the caller sends a NORMAL turn (NOT a mismatch);
+ *  - `invalid`  : the target runId belongs to a DIFFERENT chat (malformed CAS 400);
+ *  - `mismatch` : a DIFFERENT run is active than the one the client targeted —
+ *                 409 SUPERSEDE_TARGET_MISMATCH carrying the current `activeRunId`
+ *                 (the client does NOT auto-retry);
+ *  - `timeout`  : the target did not settle within W — 409 SUPERSEDE_TIMEOUT,
+ *                 nothing persisted;
+ *  - `ready`    : the target was stopped AND settled (or its zombie's intended was
+ *                 applied) — the slot is free; the caller may beginRun the new run.
+ */
+export type SupersedeResult =
+  | { kind: 'degrade' }
+  | { kind: 'invalid' }
+  | { kind: 'mismatch'; activeRunId: string }
+  | { kind: 'timeout' }
+  | { kind: 'ready' };
+
 /** A one-shot settle notifier (#487): `resolve` is called EXACTLY ONCE. */
 interface Deferred<T> {
   promise: Promise<T>;
@@ -494,6 +528,95 @@ export class AiChatRunService implements OnModuleInit {
     return undefined;
   }
 
+  /**
+   * #487: await a run's settle outcome, bounded by `timeoutMs`. Returns the
+   * outcome on settle, or undefined on TIMEOUT (or when this replica has no record
+   * of the run and its row is not terminal). Uses the LIVE settle notifier / the
+   * zombie synth when present; else reads the row (the DB is the source of truth
+   * once the in-memory record is gone). The subscriber (supersede) grabs this
+   * right after Stop; commit 1's race makes the settle land in ms on a healthy DB.
+   */
+  async awaitSettled(
+    runId: string,
+    workspaceId: string,
+    timeoutMs: number,
+  ): Promise<RunSettleOutcome | undefined> {
+    const pending = this.peekSettled(runId);
+    if (pending) {
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      const timeout = new Promise<undefined>((resolve) => {
+        timer = setTimeout(() => resolve(undefined), timeoutMs);
+        timer.unref?.();
+      });
+      try {
+        return await Promise.race([pending, timeout]);
+      } finally {
+        if (timer) clearTimeout(timer);
+      }
+    }
+    // No live notifier and no zombie: read the row (already settled-and-written,
+    // or unknown here). A terminal row is an outcome; anything else -> undefined.
+    const row = await this.runRepo.findById(runId, workspaceId);
+    if (row && isRunTerminal(row.status)) {
+      return {
+        status: row.status as RunTerminalStatus,
+        error: row.error ?? null,
+        terminalWriteFailed: false,
+      };
+    }
+    return undefined;
+  }
+
+  /**
+   * #487: the SERVER supersede CAS for `POST /stream { supersede: { runId: X } }`.
+   * Atomically transitions "X is the chat's active run" -> "X is stopped, settled,
+   * slot free" so the caller can start a replacement run. See {@link
+   * SupersedeResult} for the branch semantics.
+   *
+   * On a `ready` result the caller MUST still go through the normal beginRun gate
+   * (the partial unique index) — between the slot freeing here and beginRun a
+   * neighbouring tab's ordinary POST can win the slot (documented SLOT-THEFT: the
+   * loser then gets a MISMATCH carrying the NEW runId). There is also NO side-
+   * effect quiescence: an in-flight write of the stopped run may still land AFTER
+   * the new run starts (commit 1 stops the NEXT call, not one already committing),
+   * so the caller adds a prompt note to the new run.
+   */
+  async supersede(
+    chatId: string,
+    targetRunId: string,
+    workspaceId: string,
+    timeoutMs: number = SUPERSEDE_SETTLE_TIMEOUT_MS,
+  ): Promise<SupersedeResult> {
+    // Validate the target belongs to THIS chat (a CAS targeting another chat's run
+    // is malformed -> 400). A missing row is NOT invalid: the run may have ended
+    // and been pruned; the active-run check below decides degrade vs mismatch.
+    const target = await this.getRun(targetRunId, workspaceId);
+    if (target && target.chatId !== chatId) return { kind: 'invalid' };
+
+    const active = await this.getActiveForChat(chatId, workspaceId);
+    // No active run: it ended between the client's click and this POST — this is a
+    // DEGRADE to a normal send, NOT a mismatch (the user's intent still holds).
+    if (!active) return { kind: 'degrade' };
+    // A DIFFERENT run is active than the one the client saw -> mismatch. The
+    // client does not auto-retry; it surfaces the new runId.
+    if (active.id !== targetRunId) {
+      return { kind: 'mismatch', activeRunId: active.id };
+    }
+
+    // The target IS active: stop it, then await its settle within W.
+    await this.requestStop(targetRunId, workspaceId);
+    const outcome = await this.awaitSettled(targetRunId, workspaceId, timeoutMs);
+    if (!outcome) return { kind: 'timeout' };
+    // Gave up (terminal write failed): apply the intended status via the
+    // conditional UPDATE so the slot actually frees. If that ALSO fails, the row
+    // is still stranded -> treat as a timeout (nothing persisted for the new run).
+    if (outcome.terminalWriteFailed) {
+      const settled = await this.settleZombie(targetRunId);
+      if (!settled) return { kind: 'timeout' };
+    }
+    return { kind: 'ready' };
+  }
+
   /** #487 test/diagnostic seam: whether a give-up zombie is held for this run. */
   hasZombie(runId: string): boolean {
     return this.zombies.has(runId);

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

@@ -0,0 +1,221 @@
+import {
+  BadRequestException,
+  ConflictException,
+  HttpException,
+} from '@nestjs/common';
+import { AiChatController } from './ai-chat.controller';
+import type { User, Workspace } from '@docmost/db/types/entity.types';
+
+/**
+ * #487 commit 3 — the single concurrency GATE (both modes) + the server supersede
+ * CAS, at the controller boundary. The gate + CAS run BEFORE res.hijack(), so a
+ * rejected concurrent start / a CAS branch returns clean JSON (an HttpException
+ * the controller's post-hijack catch re-serializes). These assert the OBSERVABLE
+ * HTTP contract against the real controller + a stubbed run service.
+ */
+describe('#487 AiChatController.stream — gate + supersede', () => {
+  const user = { id: 'u1' } as User;
+
+  function wsWith(autonomousRuns: boolean): Workspace {
+    return {
+      id: 'ws1',
+      settings: { ai: { chat: true, autonomousRuns } },
+    } as unknown as Workspace;
+  }
+
+  function makeReqRes(body: Record<string, unknown>) {
+    const req = {
+      raw: { sessionId: 'sess', once: jest.fn(), destroyed: false },
+      body,
+    };
+    const res = {
+      raw: {
+        writableEnded: false,
+        headersSent: false,
+        on: jest.fn(),
+        once: jest.fn(),
+        setHeader: jest.fn(),
+        end: jest.fn(),
+        statusCode: 200,
+        flushHeaders: jest.fn(),
+      },
+      hijack: jest.fn(),
+      status: jest.fn().mockReturnThis(),
+      send: jest.fn(),
+    };
+    return { req, res };
+  }
+
+  function makeController(runServiceOverrides: Record<string, jest.Mock>) {
+    const aiChatService = {
+      resolveRoleForRequest: jest.fn().mockResolvedValue(null),
+      getChatModel: jest.fn().mockResolvedValue({}),
+      stream: jest.fn().mockResolvedValue(undefined),
+    };
+    const aiChatRunService = {
+      getActiveForChat: jest.fn().mockResolvedValue(undefined),
+      supersede: jest.fn(),
+      beginRun: jest.fn().mockResolvedValue({
+        runId: 'run-new',
+        signal: new AbortController().signal,
+      }),
+      linkAssistantMessage: jest.fn(),
+      recordStep: jest.fn(),
+      finalizeRun: jest.fn(),
+      requestStop: jest.fn(),
+      ...runServiceOverrides,
+    };
+    const controller = new AiChatController(
+      aiChatService as never,
+      aiChatRunService as never,
+      {} as never, // aiChatRepo
+      {} as never, // aiChatMessageRepo
+      {} as never, // aiTranscription
+      {} as never, // pageRepo
+    );
+    return { controller, aiChatService, aiChatRunService };
+  }
+
+  const codeOf = (err: unknown) =>
+    (((err as HttpException).getResponse() as Record<string, unknown>) ?? {})
+      .code;
+
+  describe('single concurrency gate — BOTH modes reject the second tab with 409', () => {
+    for (const autonomousRuns of [true, false]) {
+      it(`rejects a concurrent start with 409 A_RUN_ALREADY_ACTIVE (autonomousRuns=${autonomousRuns})`, async () => {
+        const { controller, aiChatRunService } = makeController({
+          getActiveForChat: jest
+            .fn()
+            .mockResolvedValue({ id: 'run-live', chatId: 'c1' }),
+        });
+        const { req, res } = makeReqRes({ chatId: 'c1' });
+        let thrown: unknown;
+        try {
+          await controller.stream(
+            req as never,
+            res as never,
+            user,
+            wsWith(autonomousRuns),
+          );
+        } catch (e) {
+          thrown = e;
+        }
+        expect(thrown).toBeInstanceOf(ConflictException);
+        expect((thrown as HttpException).getStatus()).toBe(409);
+        expect(codeOf(thrown)).toBe('A_RUN_ALREADY_ACTIVE');
+        // Rejected BEFORE committing to the stream (no hijack, no service.stream).
+        expect(res.hijack).not.toHaveBeenCalled();
+        expect(aiChatRunService.getActiveForChat).toHaveBeenCalledWith(
+          'c1',
+          'ws1',
+        );
+      });
+    }
+  });
+
+  it('supersede MISMATCH -> 409 SUPERSEDE_TARGET_MISMATCH carrying the current runId', async () => {
+    const { controller } = makeController({
+      supersede: jest
+        .fn()
+        .mockResolvedValue({ kind: 'mismatch', activeRunId: 'run-other' }),
+    });
+    const { req, res } = makeReqRes({
+      chatId: 'c1',
+      supersede: { runId: 'run-x' },
+    });
+    let thrown: unknown;
+    try {
+      await controller.stream(req as never, res as never, user, wsWith(true));
+    } catch (e) {
+      thrown = e;
+    }
+    expect(thrown).toBeInstanceOf(ConflictException);
+    expect(codeOf(thrown)).toBe('SUPERSEDE_TARGET_MISMATCH');
+    expect(
+      ((thrown as HttpException).getResponse() as Record<string, unknown>)
+        .activeRunId,
+    ).toBe('run-other');
+    expect(res.hijack).not.toHaveBeenCalled();
+  });
+
+  it('supersede TIMEOUT -> 409 SUPERSEDE_TIMEOUT, nothing streamed', async () => {
+    const { controller } = makeController({
+      supersede: jest.fn().mockResolvedValue({ kind: 'timeout' }),
+    });
+    const { req, res } = makeReqRes({
+      chatId: 'c1',
+      supersede: { runId: 'run-x' },
+    });
+    let thrown: unknown;
+    try {
+      await controller.stream(req as never, res as never, user, wsWith(false));
+    } catch (e) {
+      thrown = e;
+    }
+    expect(thrown).toBeInstanceOf(ConflictException);
+    expect(codeOf(thrown)).toBe('SUPERSEDE_TIMEOUT');
+    expect(res.hijack).not.toHaveBeenCalled();
+  });
+
+  it('supersede INVALID (target on another chat) -> 400 SUPERSEDE_INVALID', async () => {
+    const { controller } = makeController({
+      supersede: jest.fn().mockResolvedValue({ kind: 'invalid' }),
+    });
+    const { req, res } = makeReqRes({
+      chatId: 'c1',
+      supersede: { runId: 'run-x' },
+    });
+    let thrown: unknown;
+    try {
+      await controller.stream(req as never, res as never, user, wsWith(true));
+    } catch (e) {
+      thrown = e;
+    }
+    expect(thrown).toBeInstanceOf(BadRequestException);
+    expect(codeOf(thrown)).toBe('SUPERSEDE_INVALID');
+  });
+
+  it('supersede without chatId -> 400 SUPERSEDE_INVALID', async () => {
+    const { controller, aiChatRunService } = makeController({});
+    const { req, res } = makeReqRes({ supersede: { runId: 'run-x' } });
+    let thrown: unknown;
+    try {
+      await controller.stream(req as never, res as never, user, wsWith(true));
+    } catch (e) {
+      thrown = e;
+    }
+    expect(thrown).toBeInstanceOf(BadRequestException);
+    expect(codeOf(thrown)).toBe('SUPERSEDE_INVALID');
+    expect(aiChatRunService.supersede).not.toHaveBeenCalled();
+  });
+
+  it('supersede READY -> proceeds to stream with superseded=true', async () => {
+    const { controller, aiChatService } = makeController({
+      supersede: jest.fn().mockResolvedValue({ kind: 'ready' }),
+      getActiveForChat: jest.fn().mockResolvedValue(undefined), // slot free after CAS
+    });
+    const { req, res } = makeReqRes({
+      chatId: 'c1',
+      supersede: { runId: 'run-x' },
+    });
+    await controller.stream(req as never, res as never, user, wsWith(true));
+    expect(res.hijack).toHaveBeenCalled();
+    expect(aiChatService.stream).toHaveBeenCalledTimes(1);
+    expect(aiChatService.stream.mock.calls[0][0].superseded).toBe(true);
+    // The run hooks are always present now (both modes).
+    expect(aiChatService.stream.mock.calls[0][0].runHooks).toBeDefined();
+  });
+
+  it('supersede DEGRADE -> proceeds to a normal send (superseded=false)', async () => {
+    const { controller, aiChatService } = makeController({
+      supersede: jest.fn().mockResolvedValue({ kind: 'degrade' }),
+    });
+    const { req, res } = makeReqRes({
+      chatId: 'c1',
+      supersede: { runId: 'run-x' },
+    });
+    await controller.stream(req as never, res as never, user, wsWith(false));
+    expect(aiChatService.stream).toHaveBeenCalledTimes(1);
+    expect(aiChatService.stream.mock.calls[0][0].superseded).toBe(false);
+  });
+});

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

@@ -432,12 +432,66 @@ export class AiChatController {
     // HttpException) instead of breaking mid-stream.
     const model = await this.aiChatService.getChatModel(workspace.id, role);
 
-    // #184: one active run per chat. For an EXISTING chat reject a concurrent
-    // start with a clean 409 BEFORE hijack (the common double-submit / second-tab
-    // case), so the user gets JSON, not a mid-stream error. A brand-new chat
-    // (no chatId) cannot have a prior run, and the DB partial unique index is the
-    // backstop against any race that slips past this check.
-    if (autonomousRuns && body.chatId) {
+    // #487: server-side supersede CAS ("interrupt and send now"). When the client
+    // asks to replace a live run, atomically STOP it and wait for it to settle
+    // before this turn claims the slot. Runs BEFORE hijack so every branch returns
+    // clean JSON (the client keeps the composer text on a 409). See
+    // AiChatRunService.supersede for the branch semantics.
+    let superseded = false;
+    const supersedeRunId = body.supersede?.runId;
+    if (supersedeRunId) {
+      if (!body.chatId) {
+        throw new BadRequestException({
+          message: 'supersede requires chatId',
+          code: 'SUPERSEDE_INVALID',
+        });
+      }
+      const result = await this.aiChatRunService.supersede(
+        body.chatId,
+        supersedeRunId,
+        workspace.id,
+      );
+      switch (result.kind) {
+        case 'invalid':
+          throw new BadRequestException({
+            message: 'The run to supersede does not belong to this chat',
+            code: 'SUPERSEDE_INVALID',
+          });
+        case 'mismatch':
+          // A DIFFERENT run is active than the one the client targeted. Surface
+          // the CURRENT runId; the client does NOT auto-retry (a stale CAS).
+          throw new ConflictException({
+            message: 'A different agent run is now active on this chat',
+            code: 'SUPERSEDE_TARGET_MISMATCH',
+            activeRunId: result.activeRunId,
+          });
+        case 'timeout':
+          // The target did not settle within W — nothing was persisted, the
+          // composer keeps the text. NOT a rollback: the stop is already issued.
+          throw new ConflictException({
+            message:
+              'The previous run did not stop in time; nothing was sent — please try again',
+            code: 'SUPERSEDE_TIMEOUT',
+          });
+        case 'ready':
+          // The target stopped and settled: the slot is free. Prompt the new run
+          // that the old run's last operations may still be applying.
+          superseded = true;
+          break;
+        case 'degrade':
+          // The run already ended between click and POST — send normally.
+          break;
+      }
+    }
+
+    // #487: one active run per chat — ENFORCED IN BOTH MODES now (legacy mode used
+    // to have NO gate, so two tabs streamed two parallel turns on one chat, which
+    // interleaved history and crashed convertToModelMessages). Reject a concurrent
+    // start with a clean pre-hijack 409 (double-submit / second-tab). A brand-new
+    // chat (no chatId) cannot have a prior run, and the DB partial unique index in
+    // beginRun is the authoritative backstop for any race that slips past here
+    // (including a slot stolen between a supersede release and beginRun).
+    if (body.chatId) {
       const active = await this.aiChatRunService.getActiveForChat(
         body.chatId,
         workspace.id,
@@ -446,107 +500,94 @@ export class AiChatController {
         throw new ConflictException({
           message: 'An agent run is already in progress for this chat',
           code: 'A_RUN_ALREADY_ACTIVE',
+          activeRunId: active.id,
         });
       }
     }
 
-    // Run-lifecycle hooks (#184), only when the flag is on. They wrap the turn in
-    // a durable run whose abort is governed by the run (explicit stop), persist
-    // its progress, and settle its terminal status — see AiChatRunService.
-    const runHooks: AiChatRunHooks | undefined = autonomousRuns
-      ? {
-          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,
-              workspace.id,
-              messageId,
-            ),
-          onStep: (runId, stepCount) =>
-            void this.aiChatRunService.recordStep(
-              runId,
-              workspace.id,
-              stepCount,
-            ),
-          onSettled: (runId, status, error) =>
-            this.aiChatRunService.finalizeRun(
-              runId,
-              workspace.id,
-              status,
-              error,
-            ),
+    // #487: the turn is ALWAYS a first-class RUN now (both modes). The mode
+    // difference is only the abort semantics on a browser disconnect (onClose
+    // below). currentRunId is captured at begin so a legacy disconnect can stop
+    // the run through its stop lever.
+    let currentRunId: string | undefined;
+    const runHooks: AiChatRunHooks = {
+      begin: async (chatId) => {
+        const handle = await this.aiChatRunService.beginRun({
+          chatId,
+          workspaceId: workspace.id,
+          userId: user.id,
+          trigger: 'user',
+        });
+        currentRunId = handle?.runId;
+        // #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.
+        if (
+          handle?.runId &&
+          this.environment?.isAiChatResumableStreamEnabled?.()
+        ) {
+          this.streamRegistry?.open(chatId, handle.runId);
         }
-      : undefined;
+        return handle;
+      },
+      onAssistantSeeded: (runId, messageId) =>
+        this.aiChatRunService.linkAssistantMessage(
+          runId,
+          workspace.id,
+          messageId,
+        ),
+      onStep: (runId, stepCount) =>
+        void this.aiChatRunService.recordStep(runId, workspace.id, stepCount),
+      onSettled: (runId, status, error) =>
+        this.aiChatRunService.finalizeRun(runId, workspace.id, status, error),
+    };
 
-    // Abort the agent loop when the client disconnects. `close` also fires on
-    // normal completion, so only abort when the response has not finished
-    // writing (a genuine disconnect). `once` fires at most once and self-removes;
-    // we also drop it on response `finish` so it never lingers after the stream
-    // completes normally (the AI SDK pipes the response fire-and-forget, so we
-    // cannot simply remove it once `stream()` returns).
+    // Handle a client disconnect. `close` also fires on normal completion, so only
+    // act when the response has not finished writing (a genuine disconnect). `once`
+    // fires at most once and self-removes; we also drop it on response `finish`.
     // DIAGNOSTIC (Safari stream-drop investigation) — temporary: wall-clock at
     // which a Safari disconnect is observed, measured from request receipt.
     const reqStartedAt = Date.now();
     const controller = new AbortController();
     const onClose = (): void => {
-      // A genuine disconnect leaves the response unfinished (unlike a normal
-      // completion, which also fires `close`). Such a drop — e.g. a reverse
-      // proxy cutting the SSE mid-answer — is otherwise invisible server-side,
-      // so log it here.
       if (!res.raw.writableEnded) {
         if (autonomousRuns) {
-          // #184: the turn is a DETACHED run. A disconnect must NOT abort it —
-          // the run keeps executing and persisting server-side; the client
-          // reconnects via /ai-chat/run (or re-stops via /ai-chat/stop). Log only.
+          // #184: a DETACHED run — a disconnect must NOT stop it. The run keeps
+          // executing and persisting server-side; the client reconnects via
+          // /ai-chat/run (or re-stops via /ai-chat/stop). Log only.
           this.logger.log(
             `AI chat stream: client disconnected; run continues server-side ` +
               `(elapsed=${Date.now() - reqStartedAt}ms since request received)`,
           );
         } else {
+          // #487: legacy — a disconnect ENDS the turn, but the turn is now a RUN,
+          // so stop it through the run's stop lever (requestStop). streamText no
+          // longer consumes the socket signal (effectiveSignal is the run signal),
+          // so aborting `controller` would do nothing; requestStop aborts the run.
           this.logger.warn(
-            `AI chat stream: client disconnected before completion; aborting turn ` +
-              `(elapsed=${Date.now() - reqStartedAt}ms since request received)`,
+            `AI chat stream: client disconnected before completion; stopping the ` +
+              `run (elapsed=${Date.now() - reqStartedAt}ms since request received)`,
           );
-          controller.abort();
+          if (currentRunId) {
+            void this.aiChatRunService.requestStop(currentRunId, workspace.id);
+          }
         }
       }
     };
     req.raw.once('close', onClose);
     res.raw.once('finish', () => req.raw.off('close', onClose));
 
-    // #184: in detached mode the turn is NOT aborted on disconnect, so the SDK's
-    // pipe keeps writing to a socket the client may have dropped — for the rest of
-    // the (continuing) run. A write to the dead socket can emit an 'error' on the
-    // raw response; without a listener that surfaces as an unhandled error event.
-    // Swallow it (the run continues server-side regardless). Legacy mode aborts on
-    // disconnect, so it does not need this and keeps its exact prior behavior.
-    if (autonomousRuns) {
-      res.raw.on('error', (err) => {
-        this.logger.debug(
-          `AI chat detached stream: post-disconnect socket error swallowed: ${
-            err instanceof Error ? err.message : String(err)
-          }`,
-        );
-      });
-    }
+    // #184/#487: the run/pipe can outlive the socket in BOTH modes now (autonomous
+    // keeps going; legacy keeps going until requestStop's abort unwinds the turn).
+    // The SDK's pipe may then write to a dropped socket and emit an 'error' on the
+    // raw response — swallow it so it never surfaces as an unhandled error event.
+    res.raw.on('error', (err) => {
+      this.logger.debug(
+        `AI chat stream: post-disconnect socket error swallowed: ${
+          err instanceof Error ? err.message : String(err)
+        }`,
+      );
+    });
 
     // Commit to streaming: hijack so Fastify stops managing the response and
     // the AI SDK can write the UI-message stream directly to the Node socket.
@@ -562,8 +603,10 @@ export class AiChatController {
         signal: controller.signal,
         model,
         role,
-        // #184: present only when the flag is on; wraps the turn in a durable run.
+        // #487: the turn is always run-wrapped now (both modes).
         runHooks,
+        // #487: warn the new run that a superseded run's last ops may still apply.
+        superseded,
       });
     } catch (err) {
       // Any failure AFTER hijack can no longer go through Nest's exception

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

@@ -101,6 +101,22 @@ const INTERRUPT_NOTE =
   'assume your previous response was complete, and do not silently restart the ' +
   'partial work — build on it or follow the new instruction.';
 
+/**
+ * #487: injected on a turn started by SUPERSEDING a previous run (the user hit
+ * "interrupt and send now" while a run was live). The previous run was Stopped,
+ * but there is NO side-effect quiescence — a write it had already committed, or
+ * one committing at the moment of Stop, may land with a small delay AFTER this new
+ * run starts. So the model is told its picture of the page/state may be a beat
+ * stale and to re-read before assuming an edit did or did not apply.
+ */
+const SUPERSEDE_NOTE =
+  'NOTE: A previous agent run in this conversation was just interrupted so this ' +
+  'new turn could start. That run was stopped, but any operation it had already ' +
+  'begun (e.g. a page edit) may still be applied with a short delay. Do not ' +
+  'assume the document/state is exactly as the interrupted run left it — if you ' +
+  'need to rely on the current content, RE-READ it with the page tools before ' +
+  'acting rather than trusting a cached view.';
+
 /**
  * Injected on a turn where the open page was hand-edited by the user (or anyone
  * else) AFTER the agent's previous response ended (#274). The server takes a
@@ -203,6 +219,14 @@ export interface BuildSystemPromptInput {
    * (partial) answer was cut off by the user's new message.
    */
   interrupted?: boolean;
+  /**
+   * #487: true when THIS turn was started by superseding a still-live previous run
+   * ("interrupt and send now"). Adds SUPERSEDE_NOTE so the model knows the previous
+   * run's last operations may still be applying and to re-read state it depends on.
+   * Distinct from `interrupted` (which is about a PARTIAL prior answer in history);
+   * both can be set together. Self-clears — set only for the superseding turn.
+   */
+  superseded?: boolean;
   /**
    * Set only when the open page was edited by the user AFTER the agent's previous
    * turn ended (#274), confirmed server-side by diffing the current page against
@@ -311,6 +335,7 @@ export function buildSystemPrompt({
   openedPage,
   mcpInstructions,
   interrupted,
+  superseded,
   pageChanged,
   deferredToolsEnabled,
   toolCatalog,
@@ -360,6 +385,13 @@ export function buildSystemPrompt({
     context += `\n${INTERRUPT_NOTE}`;
   }
 
+  // Supersede note (#487): present only for a turn that stopped and replaced a
+  // still-live previous run — warns the model the previous run's last operations
+  // may still be applying (no side-effect quiescence).
+  if (superseded) {
+    context += `\n${SUPERSEDE_NOTE}`;
+  }
+
   // Per-turn page-change note (#274). Added to the context section (inside the
   // safety sandwich), present only when the server detected that the open page
   // was edited by the user since the agent's last turn ended. The diff content is

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

@@ -379,6 +379,14 @@ export interface AiChatStreamBody {
   // it against persisted history (`isInterruptResume`) before injecting the
   // interrupt note, so a spoofed/stale flag on an ordinary turn is ignored.
   interrupted?: boolean;
+  // #487: server-side supersede CAS. When present, this POST asks the server to
+  // STOP the run `supersede.runId` (which the client saw as the chat's active run)
+  // and, once it has settled, start THIS turn in its place. The server validates
+  // the target against the chat and answers 400 (wrong chat) / 409
+  // SUPERSEDE_TARGET_MISMATCH / 409 SUPERSEDE_TIMEOUT, or proceeds normally
+  // (degrade / ready). Absent => an ordinary send (rejected with 409
+  // A_RUN_ALREADY_ACTIVE if a run is already active on the chat).
+  supersede?: { runId?: string } | null;
   // useChat sends the full UIMessage list; the last one is the new user turn.
   messages?: UIMessage[];
 }
@@ -428,6 +436,11 @@ export interface AiChatStreamArgs {
   // chat row (existing chat) or the request body (new chat). null => universal
   // assistant. Carried here so the turn never re-loads it.
   role: AiAgentRole | null;
+  // #487: true when this turn was started by SUPERSEDING a still-live previous run
+  // (the controller ran the supersede CAS to a `ready` result). Adds the
+  // SUPERSEDE_NOTE to the system prompt (the previous run's last ops may still be
+  // applying — no side-effect quiescence). Absent on an ordinary send.
+  superseded?: boolean;
 }
 
 /**
@@ -727,6 +740,7 @@ export class AiChatService implements OnModuleInit {
     model,
     role,
     runHooks,
+    superseded,
   }: AiChatStreamArgs): Promise<void> {
     // Resolve / create the chat. A new chat is created when no valid chatId is
     // supplied or the supplied one does not belong to this workspace.
@@ -1033,6 +1047,9 @@ export class AiChatService implements OnModuleInit {
           // History-confirmed interrupt-resume flag (#198): adds the interrupt note
           // so the model treats the partial answer above as cut off, not finished.
           interrupted,
+          // #487: this turn superseded a still-live run — warn the model the
+          // previous run's last ops may still be applying (no quiescence).
+          superseded,
           // Detected between-turns human edit to the open page (#274): adds the
           // page_changed note + unified diff so the agent doesn't overwrite it.
           pageChanged,