From d72e1012167e9a0b2031e2608de4c7b57c7eebf2 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sat, 11 Jul 2026 04:51:44 +0300 Subject: [PATCH] =?UTF-8?q?feat(ai-chat):=20=D0=B5=D0=B4=D0=B8=D0=BD=D1=8B?= =?UTF-8?q?=D0=B9=20=D0=B3=D0=B5=D0=B9=D1=82=20=D0=BA=D0=BE=D0=BD=D0=BA?= =?UTF-8?q?=D1=83=D1=80=D0=B5=D0=BD=D1=82=D0=BD=D0=BE=D1=81=D1=82=D0=B8=20?= =?UTF-8?q?+=20=D1=81=D0=B5=D1=80=D0=B2=D0=B5=D1=80=D0=BD=D1=8B=D0=B9=20su?= =?UTF-8?q?persede=20(CAS)=20(#487)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Раньше в 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) --- .../core/ai-chat/ai-chat-run.service.spec.ts | 162 +++++++++++++ .../src/core/ai-chat/ai-chat-run.service.ts | 123 ++++++++++ .../ai-chat.controller.supersede.spec.ts | 221 ++++++++++++++++++ .../src/core/ai-chat/ai-chat.controller.ts | 207 +++++++++------- .../server/src/core/ai-chat/ai-chat.prompt.ts | 32 +++ .../src/core/ai-chat/ai-chat.service.ts | 17 ++ 6 files changed, 680 insertions(+), 82 deletions(-) create mode 100644 apps/server/src/core/ai-chat/ai-chat.controller.supersede.spec.ts diff --git a/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts b/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts index 4a24909b..1518dc73 100644 --- a/apps/server/src/core/ai-chat/ai-chat-run.service.spec.ts +++ b/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); + }); +}); diff --git a/apps/server/src/core/ai-chat/ai-chat-run.service.ts b/apps/server/src/core/ai-chat/ai-chat-run.service.ts index 2c2c379d..8fd577c8 100644 --- a/apps/server/src/core/ai-chat/ai-chat-run.service.ts +++ b/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 { promise: Promise; @@ -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 { + const pending = this.peekSettled(runId); + if (pending) { + let timer: ReturnType | undefined; + const timeout = new Promise((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 { + // 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); diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.supersede.spec.ts b/apps/server/src/core/ai-chat/ai-chat.controller.supersede.spec.ts new file mode 100644 index 00000000..69aa84b5 --- /dev/null +++ b/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) { + 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) { + 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) ?? {}) + .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) + .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); + }); +}); diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.ts b/apps/server/src/core/ai-chat/ai-chat.controller.ts index 8cc7b03d..6dbe8420 100644 --- a/apps/server/src/core/ai-chat/ai-chat.controller.ts +++ b/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 diff --git a/apps/server/src/core/ai-chat/ai-chat.prompt.ts b/apps/server/src/core/ai-chat/ai-chat.prompt.ts index 5107a77d..33c222fd 100644 --- a/apps/server/src/core/ai-chat/ai-chat.prompt.ts +++ b/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 diff --git a/apps/server/src/core/ai-chat/ai-chat.service.ts b/apps/server/src/core/ai-chat/ai-chat.service.ts index dfa1bb75..4ccca58a 100644 --- a/apps/server/src/core/ai-chat/ai-chat.service.ts +++ b/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 { // 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,