From 89fcc607828db7a6d7ab267ee303cc33296b1073 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sat, 11 Jul 2026 04:33:39 +0300 Subject: [PATCH] =?UTF-8?q?feat(ai-chat):=20=D0=B7=D0=BE=D0=BC=D0=B1=D0=B8?= =?UTF-8?q?-=D0=BC=D0=B5=D1=85=D0=B0=D0=BD=D0=B8=D0=BA=D0=B0=20finalizeRun?= =?UTF-8?q?=20+=20settledPromise=20+=20=D1=83=D1=81=D0=BB=D0=BE=D0=B2?= =?UTF-8?q?=D0=BD=D1=8B=D0=B5=20=D1=82=D0=B5=D1=80=D0=BC=D0=B8=D0=BD=D0=B0?= =?UTF-8?q?=D0=BB=D1=8C=D0=BD=D1=8B=D0=B5=20UPDATE=20(#487)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Раньше give-up-путь finalizeRun ВОССТАНАВЛИВАЛ entry (зомби неотличим от живого рана), а терминальный runRepo.update был БЕЗУСЛОВНЫМ (последний писатель перетирал терминальный статус). - Все терминальные UPDATE ранов теперь УСЛОВНЫЕ: новый repo.finalizeIfActive пишет только пока строка pending|running (зеркально onlyIfStreaming у сообщений) → двойной settle схлопывается в benign no-op, терминальный статус не перетереть. - При исчерпании ретраев finalizeRun НЕ восстанавливает entry, а оставляет ЗОМБИ-запись { terminalWriteFailed, intended:{status,error} } в отдельной map; settleZombie пере-драйвит intended условным UPDATE (зовётся reconcile/supersede/ boot sweep). Зомби удаляется после успешного UPDATE или обнаружения строки уже терминальной. - Per-run settledPromise в ОТДЕЛЬНОЙ map runId→deferred (переживает active.delete), создаётся в beginRun, резолвится ровно один раз исходом (записано/сдался); поздний подписчик через ранее взятую ссылку получает резолвленный; peekSettled: live deferred → зомби-синтез → undefined (читать строку). Обе map bounded. - Задокументирована потеря (single-process): рестарт до пере-драйва → boot sweep пишет 'aborted' поверх фактического intended — неустранимо в phase 1. Тесты: юниты give-up-пути (зомби вместо restore + notifier terminalWriteFailed + settleZombie), двойного settle, позднего подписчика; интеграционные против реальной Б, что условный finalizeIfActive не перетирает терминал и двойной settle схлопывается. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../core/ai-chat/ai-chat-run.service.spec.ts | 134 +++++--- .../src/core/ai-chat/ai-chat-run.service.ts | 302 ++++++++++++++---- .../ai-chat/ai-chat.service.lifecycle.spec.ts | 12 +- .../repos/ai-chat/ai-chat-run.repo.ts | 35 ++ .../test/integration/ai-chat-run.int-spec.ts | 46 +++ 5 files changed, 424 insertions(+), 105 deletions(-) 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 4843a979..4a24909b 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 @@ -43,6 +43,9 @@ function makeRepo(overrides: Record = {}) { workspaceId: v.workspaceId, })), update: jest.fn(async () => ({ id: 'run-1' })), + // #487: terminal finalize now goes through the CONDITIONAL write. Default + // returns a truthy row (the run WAS active -> this call wrote it). + finalizeIfActive: jest.fn(async () => ({ id: 'run-1', status: 'succeeded' })), markStopRequested: jest.fn(async () => ({ id: 'run-1' })), findActiveByChat: jest.fn(async () => undefined), findLatestByChat: jest.fn(async () => undefined), @@ -336,14 +339,12 @@ describe('AiChatRunService run lifecycle', () => { await svc.finalizeRun('run-1', 'ws-1', 'error', 'provider blew up'); expect(svc.isLocallyActive('run-1')).toBe(false); - expect(repo.update).toHaveBeenCalledWith( + // #487: the terminal write is CONDITIONAL (finalizeIfActive); finishedAt is + // stamped inside the repo method, so the service passes just status + error. + expect(repo.finalizeIfActive).toHaveBeenCalledWith( 'run-1', 'ws-1', - expect.objectContaining({ - status: 'failed', - error: 'provider blew up', - finishedAt: expect.any(Date), - }), + expect.objectContaining({ status: 'failed', error: 'provider blew up' }), ); }); @@ -366,8 +367,8 @@ describe('AiChatRunService run lifecycle', () => { // A second settle (e.g. a streamText callback firing after the catch) no-ops. await svc.finalizeRun('run-1', 'ws-1', 'completed', undefined); - expect(repo.update).toHaveBeenCalledTimes(1); - expect(repo.update).toHaveBeenCalledWith( + expect(repo.finalizeIfActive).toHaveBeenCalledTimes(1); + expect(repo.finalizeIfActive).toHaveBeenCalledWith( 'run-1', 'ws-1', expect.objectContaining({ status: 'failed', error: 'first' }), @@ -389,8 +390,8 @@ describe('AiChatRunService run lifecycle', () => { const updateGate = new Promise((res) => { resolveUpdate = res; }); - const update = jest.fn(() => updateGate); - const repo = makeRepo({ update }); + const finalizeIfActive = jest.fn(() => updateGate); + const repo = makeRepo({ finalizeIfActive }); const svc = new AiChatRunService(repo as never, makeEnv() as never); await svc.beginRun({ chatId: 'chat-1', @@ -399,23 +400,23 @@ describe('AiChatRunService run lifecycle', () => { }); // Fire both before the (pending) update resolves. The first synchronously - // claims the entry (active.delete) and awaits update; the second, started in - // the same macrotask, finds the entry already gone and returns at the claim - // WITHOUT ever calling update. + // claims the entry (active.delete) and awaits the write; the second, started + // in the same macrotask, finds the entry already gone and returns at the claim + // WITHOUT ever writing. const p1 = svc.finalizeRun('run-1', 'ws-1', 'completed'); const p2 = svc.finalizeRun('run-1', 'ws-1', 'error', 'safety-net'); // The decisive assertion: exactly one caller reached the terminal UPDATE. - expect(update).toHaveBeenCalledTimes(1); + expect(finalizeIfActive).toHaveBeenCalledTimes(1); // Let the single in-flight update land; both calls resolve cleanly. - resolveUpdate({ id: 'run-1' }); + resolveUpdate({ id: 'run-1', status: 'succeeded' }); await Promise.all([p1, p2]); - expect(update).toHaveBeenCalledTimes(1); + expect(finalizeIfActive).toHaveBeenCalledTimes(1); // The winner is the FIRST caller ('completed' -> 'succeeded'); the late // 'error' settle never wrote, so it could not clobber the real status. - expect(update).toHaveBeenCalledWith( + expect(finalizeIfActive).toHaveBeenCalledWith( 'run-1', 'ws-1', expect.objectContaining({ status: 'succeeded' }), @@ -431,10 +432,10 @@ describe('AiChatRunService run lifecycle', () => { // 409s until a restart. The fix updates FIRST and retries. let calls = 0; const repo = makeRepo({ - update: jest.fn(async () => { + finalizeIfActive: jest.fn(async () => { calls += 1; if (calls === 1) throw new Error('deadlock detected'); - return { id: 'run-1' }; + return { id: 'run-1', status: 'succeeded' }; }), }); jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); @@ -447,26 +448,29 @@ describe('AiChatRunService run lifecycle', () => { await svc.finalizeRun('run-1', 'ws-1', 'completed'); - // The retry landed the terminal write: the entry is dropped (slot freed) and - // the row carries the real terminal status — NOT stranded at 'running'. + // The retry landed the terminal write: the entry is dropped (slot freed), no + // zombie left, and the row carries the real terminal status. expect(svc.isLocallyActive('run-1')).toBe(false); - expect(repo.update).toHaveBeenCalledTimes(2); - expect(repo.update).toHaveBeenLastCalledWith( + expect(svc.hasZombie('run-1')).toBe(false); + expect(repo.finalizeIfActive).toHaveBeenCalledTimes(2); + expect(repo.finalizeIfActive).toHaveBeenLastCalledWith( 'run-1', 'ws-1', expect.objectContaining({ status: 'succeeded' }), ); }); - it('F6: if the terminal write keeps failing, the entry is RETAINED and a LATER settle completes it (chat not permanently 409d)', async () => { + it('#487 give-up: if the terminal write keeps failing, finalizeRun leaves a ZOMBIE (does NOT restore the entry) and settleZombie re-drives it', async () => { // Worst case: the DB is down for the whole first finalize (all attempts fail). - // The run must NOT be silently lost — the entry stays so a subsequent settle - // (a streamText callback, requestStop -> onAbort, or a future sweep) can retry. + // #487 changes the give-up behaviour: the entry is NOT restored (a restored + // entry is indistinguishable from a live run). Instead a ZOMBIE record holds + // the intended terminal status, and a re-drive (settleZombie — called by the + // reconcile / supersede / opportunistic paths) applies it later. let healthy = false; const repo = makeRepo({ - update: jest.fn(async () => { + finalizeIfActive: jest.fn(async () => { if (!healthy) throw new Error('pool exhausted'); - return { id: 'run-1' }; + return { id: 'run-1', status: 'succeeded' }; }), }); jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); @@ -480,35 +484,83 @@ describe('AiChatRunService run lifecycle', () => { userId: 'user-1', }); - // First settle: every bounded attempt fails -> entry retained, NOT settled. + // First settle: every bounded attempt fails -> ZOMBIE, entry NOT restored. await svc.finalizeRun('run-1', 'ws-1', 'completed'); - expect(svc.isLocallyActive('run-1')).toBe(true); - // F12: the give-up emits ONE explicit, greppable ERROR (run + chat context) - // so an operator can tell "gave up, run held in memory" from a per-attempt - // blip — distinct from the per-attempt warns. + expect(svc.isLocallyActive('run-1')).toBe(false); // NOT a live entry + expect(svc.hasZombie('run-1')).toBe(true); + expect(svc.zombieRunIds()).toContain('run-1'); + // The give-up emits ONE explicit, greppable ERROR mentioning the zombie. const gaveUp = errorSpy.mock.calls.some( (c) => /NON-TERMINAL/.test(String(c[0])) && + /ZOMBIE/.test(String(c[0])) && /run-1/.test(String(c[0])) && /chat-1/.test(String(c[0])), ); expect(gaveUp).toBe(true); + // The settle notifier resolved as terminalWriteFailed (a subscriber learns the + // slot still needs the intended status applied). + const outcome = await svc.peekSettled('run-1'); + expect(outcome).toEqual({ + status: 'succeeded', + error: null, + terminalWriteFailed: true, + }); - // The DB recovers; a later settle now succeeds and frees the slot. + // The DB recovers; a re-drive settles the zombie via the conditional UPDATE. healthy = true; - await svc.finalizeRun('run-1', 'ws-1', 'completed'); - expect(svc.isLocallyActive('run-1')).toBe(false); - expect(repo.update).toHaveBeenLastCalledWith( + const redriven = await svc.settleZombie('run-1'); + expect(redriven).toBe(true); + expect(svc.hasZombie('run-1')).toBe(false); + expect(repo.finalizeIfActive).toHaveBeenLastCalledWith( 'run-1', 'ws-1', expect.objectContaining({ status: 'succeeded' }), ); - // And it is now idempotent: a further settle no-ops (terminal row already - // written), so a double-settle can never clobber the real status. - const callsBefore = repo.update.mock.calls.length; + // A later finalizeRun is idempotent (row already terminal): it no-ops at the + // once-gate, never re-writing. + const callsBefore = repo.finalizeIfActive.mock.calls.length; await svc.finalizeRun('run-1', 'ws-1', 'error', 'late'); - expect(repo.update).toHaveBeenCalledTimes(callsBefore); + expect(repo.finalizeIfActive).toHaveBeenCalledTimes(callsBefore); + }); + + it('#487 double-settle collapses to a benign no-op (conditional write; notifier resolves once)', async () => { + // A second concurrent settle is stopped at the synchronous active.delete + // claim, so the terminal write runs exactly once and the notifier resolves + // exactly once with the FIRST settler's outcome. + const repo = makeRepo(); + const svc = new AiChatRunService(repo as never, makeEnv() as never); + await svc.beginRun({ chatId: 'chat-1', workspaceId: 'ws-1', userId: 'u1' }); + + await svc.finalizeRun('run-1', 'ws-1', 'aborted'); + await svc.finalizeRun('run-1', 'ws-1', 'error', 'late'); // no-op + + expect(repo.finalizeIfActive).toHaveBeenCalledTimes(1); + const outcome = await svc.peekSettled('run-1'); + // peekSettled after resolve+delete falls through (notifier dropped, no zombie) + // -> undefined; the FIRST settler already resolved any earlier subscriber. + expect(outcome).toBeUndefined(); + }); + + it('#487 late settledPromise subscriber gets the resolved outcome', async () => { + const repo = makeRepo(); + const svc = new AiChatRunService(repo as never, makeEnv() as never); + await svc.beginRun({ chatId: 'chat-1', workspaceId: 'ws-1', userId: 'u1' }); + + // Subscribe BEFORE settle: hold the promise reference (as supersede does). + const early = svc.peekSettled('run-1'); + expect(early).toBeDefined(); + + await svc.finalizeRun('run-1', 'ws-1', 'completed'); + + // The reference grabbed before settle resolves with the written outcome, even + // though the notifier was dropped from the map on resolve (bounded). + await expect(early).resolves.toEqual({ + status: 'succeeded', + error: null, + terminalWriteFailed: false, + }); }); it('recordStep / linkAssistantMessage are best-effort: a repo failure is swallowed', async () => { 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 b2e0d393..2c2c379d 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 @@ -34,6 +34,54 @@ export class RunAlreadyActiveError extends Error { export type TurnTerminalStatus = 'completed' | 'error' | 'aborted'; export type RunTerminalStatus = 'succeeded' | 'failed' | 'aborted'; +/** The terminal run statuses — the row is done once it reads one of these. */ +export const RUN_TERMINAL_STATUSES: readonly RunTerminalStatus[] = [ + 'succeeded', + 'failed', + 'aborted', +]; + +/** Whether a persisted run status is terminal (settled). */ +export function isRunTerminal(status: string | null | undefined): boolean { + return ( + status === 'succeeded' || status === 'failed' || status === 'aborted' + ); +} + +/** + * #487: the outcome a run's {@link AiChatRunService.finalizeRun} settled with. + * `terminalWriteFailed` = the terminal write GAVE UP after the bounded retry, so + * the row is still non-terminal ('running') and a ZOMBIE record holds the + * `intended` status for a later re-drive (reconcile / supersede / boot sweep). A + * subscriber (supersede, #487 commit 3) uses this to decide whether the slot is + * genuinely free or must first have the intended status applied. + */ +export interface RunSettleOutcome { + status: RunTerminalStatus; + error: string | null; + terminalWriteFailed: boolean; +} + +/** A one-shot settle notifier (#487): `resolve` is called EXACTLY ONCE. */ +interface Deferred { + promise: Promise; + resolve: (value: T) => void; +} + +/** + * #487: a run whose terminal write GAVE UP (every bounded attempt failed). The + * row is stranded non-terminal ('running'); this record is the ONLY thing that + * distinguishes it from a live run, and carries the `intended` terminal status so + * a re-drive can apply it via the conditional UPDATE. Process-local (phase-1 + * single-process assumption): a restart drops it, and the boot sweep then writes + * 'aborted' over the intended — a documented loss (see finalizeRun). + */ +interface ZombieRun { + workspaceId: string; + chatId: string; + intended: { status: RunTerminalStatus; error: string | null }; +} + export function mapTurnStatusToRun( status: TurnTerminalStatus, ): RunTerminalStatus { @@ -101,6 +149,22 @@ export class AiChatRunService implements OnModuleInit { // uptime — negligible in phase 1's single process. private readonly settled = new Set(); + // #487 runId -> one-shot settle notifier. Kept in a SEPARATE map from `active` + // ON PURPOSE: it must OUTLIVE the `active.delete` claim inside finalizeRun (the + // claim frees the slot the instant finalize starts), so a subscriber can still + // await the outcome after the entry is gone. Created in beginRun, resolved + // EXACTLY ONCE in finalizeRun, then removed (bounded). Absence => this replica + // has no live notifier: a subscriber falls back to the zombie map, then to the + // row (see peekSettled). Process-local (phase-1 single-process assumption). + private readonly settledPromises = new Map>(); + + // #487 runId -> ZOMBIE record: a run whose terminal write gave up (row stranded + // non-terminal). BOUNDED — an entry is added only on give-up and removed on a + // successful re-drive (settleZombie) or when the row is found already terminal; + // a process restart clears it (and the boot sweep settles the stranded row). + // Process-local (phase-1 single-process assumption). + private readonly zombies = new Map(); + // Bounded retry for the terminal write (F6): a single PK UPDATE can fail // transiently under many fire-and-forget writes (pool exhaustion, deadlock, a // brief connection blip). Riding out that blip in-place matters because the @@ -224,6 +288,10 @@ export class AiChatRunService implements OnModuleInit { chatId: args.chatId, workspaceId: args.workspaceId, }); + // #487: arm the one-shot settle notifier BEFORE returning, so a subscriber + // that races in immediately after begin always finds a promise to await. It + // is resolved exactly once when the run settles (or gives up). + this.settledPromises.set(run.id, this.makeDeferred()); return { runId: run.id, signal: controller.signal }; } @@ -263,47 +331,43 @@ export class AiChatRunService implements OnModuleInit { } /** - * Finalize a run to its terminal status (succeeded / failed / aborted), - * stamping finishedAt + any error. Best-effort, but ROBUST against a transient - * terminal-write failure (F6) AND atomically safe against a concurrent settle. + * Finalize a run to its terminal status (succeeded / failed / aborted) via a + * CONDITIONAL UPDATE, stamping finishedAt + any error. Atomically safe against a + * concurrent settle AND robust against a transient terminal-write failure. * * ATOMIC ONCE-CLAIM (the gate must close in ONE synchronous tick): two * finalizeRun calls for the SAME run can race — the documented real path is * AiChatService.stream's safety-net catch settling the turn to 'error' while a * streamText terminal callback (onFinish/onAbort/onError) ALSO settles it. The - * `settled.has` check alone is NOT a gate: it is read BEFORE the awaited UPDATE, - * so two callers can both see `false` and both write the row (last-write-wins - * clobbers the real terminal status, and the bounded retry only widens that - * window). The claim therefore happens via `active.delete`, a SYNCHRONOUS - * check-and-clear with NO await between the gate and the entry removal: the - * second concurrent caller finds the entry already gone and returns in the same - * tick, before any UPDATE. The transition "nobody is finalizing" -> "I am - * finalizing" is thus a single atomic step. + * claim happens via `active.delete`, a SYNCHRONOUS check-and-clear with NO await + * between the gate and the entry removal: the second concurrent caller finds the + * entry already gone and returns in the same tick, before any UPDATE. * - * ORDER MATTERS (F6): once we own the claim, the terminal UPDATE happens FIRST; - * only once it SUCCEEDS do we record the run as settled. If the UPDATE fails on - * every bounded attempt we RESTORE the in-memory entry, leave the run UNsettled, - * and emit an ERROR signal that the row is left non-terminal 'running' (which - * would 409 every future turn in the chat until recovery). An in-process retry - * by a LATER settle is only POSSIBLE, never guaranteed: it needs (a) the entry - * to have been restored at the give-up path AND (b) a fresh settler to arrive - * AFTER that restore. A concurrent settler that arrives DURING the retry window - * — while the entry is deleted for backoff and not yet restored — is consumed at - * the synchronous `active.delete` claim (it finds nothing to delete and returns - * a no-op), so it does NOT become an in-process retrier. The NO-streamText path - * (the turn threw before streamText was wired, so ONLY the safety-net ever - * settles) likewise has no second in-process settler at all. The UNCONDITIONAL - * backstop in every case is the boot sweep on the next restart (phase 1 has no - * periodic in-process sweep); the retained entry is bounded (cleared on restart) - * and harmless meanwhile. + * ALL TERMINAL WRITES ARE CONDITIONAL (#487): `finalizeIfActive` only flips a + * row still in pending|running (mirror of the assistant message's + * `onlyIfStreaming`). So even a settle that DID reach the UPDATE (e.g. a + * reconcile stamp racing an owner finalize) can never clobber a terminal status + * — the loser matches nothing and is a benign no-op. `active.delete` is the + * fast, in-process gate; the conditional WHERE is the authoritative one. * - * IDEMPOTENT on SUCCESS (#184 review): the terminal write happens AT MOST ONCE - * per run. After a successful write the once-gate keys off {@link settled} (the - * terminal row already written) so a settle arriving AFTER the entry was already - * dropped-and-settled returns early; a settle racing the in-flight write is - * stopped earlier still, by the `active.delete` claim. Either way a genuine - * double-settle collapses to a single write and a late settle can never clobber - * the real terminal status or double-write the row. + * ZOMBIE ON GIVE-UP (#487): if every bounded attempt THROWS (the DB is down for + * the whole finalize), we do NOT restore the entry. The row is stranded + * non-terminal ('running'); we record a ZOMBIE `{ terminalWriteFailed, intended + * }` (the ONLY thing distinguishing this dead run from a live one) and resolve + * the settle notifier with `terminalWriteFailed: true`. A restore would make the + * zombie indistinguishable from a live run to every reader; instead a re-drive + * (settleZombie, called by the periodic reconcile / supersede / opportunistic + * paths) applies the intended status later via the same conditional UPDATE. + * + * DOCUMENTED LOSS (#487, single-process phase 1): if the process RESTARTS before + * a zombie is re-driven, the in-memory zombie map is gone and the boot sweep + * (unconditional) writes 'aborted' over the ACTUAL intended status. This is + * unavoidable while the run lifecycle is single-process — there is no durable + * record of `intended`; a cross-process durable intent is deferred to phase 2. + * + * IDEMPOTENT: the settle notifier resolves EXACTLY ONCE; a second settle is + * stopped at `settled.has` or the `active.delete` claim, so a double-settle + * collapses to a single write and can never double-resolve or clobber the row. */ async finalizeRun( runId: string, @@ -314,13 +378,17 @@ export class AiChatRunService implements OnModuleInit { // ---- Atomic once-claim (synchronous; NO await before the gate closes) ---- // Already terminally written -> idempotent no-op. if (this.settled.has(runId)) return; - // Capture the entry BEFORE the delete so a total-failure path can restore it. + // Capture the entry BEFORE the delete for the give-up log context. const entry = this.active.get(runId); // SYNCHRONOUS check-and-clear: the FIRST caller deletes (claims) the entry; // any concurrent SECOND caller finds nothing to delete and returns HERE, in // the same tick, before any await — so it can never reach the UPDATE. if (!this.active.delete(runId)) return; + const status = mapTurnStatusToRun(turnStatus); + const err = error ?? null; + const chatId = entry?.chatId ?? 'unknown'; + let lastError: unknown; for ( let attempt = 1; @@ -328,47 +396,159 @@ export class AiChatRunService implements OnModuleInit { attempt++ ) { try { - await this.runRepo.update(runId, workspaceId, { - status: mapTurnStatusToRun(turnStatus), - finishedAt: new Date(), - error: error ?? null, + const row = await this.runRepo.finalizeIfActive(runId, workspaceId, { + status, + error: err, }); - // Terminal write landed: arm the once-gate. The entry is already gone - // (claimed above); we do NOT restore it. The slot is now free. + // No throw => the row is now terminal (we wrote it, or it was ALREADY + // terminal — another writer won the conditional UPDATE, a benign no-op). this.settled.add(runId); + this.zombies.delete(runId); + // Resolve with the persisted outcome: our status when WE wrote it, else + // the row's real terminal status (re-read on the already-terminal path so + // a subscriber never sees a status we did not actually persist). + const outcome: RunSettleOutcome = row + ? { status, error: err, terminalWriteFailed: false } + : await this.readTerminalOutcome(runId, workspaceId, status, err); + this.resolveSettled(runId, outcome); return; - } catch (err) { - lastError = err; + } catch (err2) { + lastError = err2; this.logger.warn( `Failed to finalize run ${runId} (attempt ${attempt}/${ AiChatRunService.FINALIZE_MAX_ATTEMPTS - }): ${err instanceof Error ? err.message : 'unknown error'}`, + }): ${err2 instanceof Error ? err2.message : 'unknown error'}`, ); if (attempt < AiChatRunService.FINALIZE_MAX_ATTEMPTS) { await this.delay(AiChatRunService.FINALIZE_RETRY_BASE_MS * attempt); } } } - // Every attempt failed: this is a give-up, materially worse than a per-attempt - // blip — the row is left NON-TERMINAL ('running'), so emit ONE explicit, - // greppable ERROR so an operator can tell "survived a blip" from "gave up, run - // held in memory until recovery" (the last warn alone says only "attempt 3/3"). + // Every attempt threw: GIVE UP. The row is stranded non-terminal ('running'). + // Do NOT restore the entry (a restored entry is indistinguishable from a live + // run); leave a ZOMBIE record instead, and resolve the notifier as + // terminalWriteFailed so a subscriber knows the slot still needs the intended + // status applied. One explicit, greppable ERROR so an operator can tell a + // give-up from a per-attempt blip. this.logger.error( - `Run ${runId} (chat ${entry?.chatId ?? 'unknown'}) left NON-TERMINAL ` + - `('running'): terminal write failed after ${ - AiChatRunService.FINALIZE_MAX_ATTEMPTS - } attempts; entry retained in memory, recovery deferred to next settle / ` + - `boot sweep`, + `Run ${runId} (chat ${chatId}) left NON-TERMINAL ('running'): terminal ` + + `write failed after ${AiChatRunService.FINALIZE_MAX_ATTEMPTS} attempts; ` + + `ZOMBIE recorded (intended '${status}'), recovery deferred to reconcile / ` + + `supersede / boot sweep`, lastError, ); - // RESTORE the claimed entry (and leave the run UNsettled) so a LATER settle - // that arrives AFTER this restore MAY retry the terminal write — but that - // in-process retry is NOT guaranteed (a concurrent settler caught in the retry - // window above is consumed at the `active.delete` claim, and the no-streamText - // path has no second settler at all). The UNCONDITIONAL backstop in every case - // is the boot sweep on the next restart; the restored entry is bounded and - // cleared on restart. - if (entry) this.active.set(runId, entry); + this.zombies.set(runId, { + workspaceId, + chatId, + intended: { status, error: err }, + }); + this.resolveSettled(runId, { status, error: err, terminalWriteFailed: true }); + } + + /** + * #487: re-drive a zombie run's intended terminal write (the conditional + * UPDATE). Called by the periodic reconcile (commit 4), an opportunistic + * single-chat reconcile, and supersede (commit 3). On success — the row is now + * terminal (written OR found already terminal) — the zombie is cleared and the + * once-gate armed; on another failure the zombie is kept for a later retry. + * Returns true when the row is now terminal. Best-effort; never throws. + */ + async settleZombie(runId: string): Promise { + const z = this.zombies.get(runId); + if (!z) return false; + try { + await this.runRepo.finalizeIfActive(runId, z.workspaceId, { + status: z.intended.status, + error: z.intended.error, + }); + this.zombies.delete(runId); + this.settled.add(runId); + return true; + } catch (err) { + this.logger.warn( + `Re-drive of zombie run ${runId} (chat ${z.chatId}) failed; will retry ` + + `later: ${err instanceof Error ? err.message : 'unknown error'}`, + ); + return false; + } + } + + /** + * #487: the run's settle outcome as seen by THIS replica, or undefined when it + * has no record (the caller then reads the row — the DB is the source of truth). + * A LIVE deferred (still settling, or resolved-but-not-yet-consumed) wins; a + * ZOMBIE synthesizes the give-up outcome. A subscriber (supersede) races this + * against a timeout. + */ + peekSettled(runId: string): Promise | undefined { + const d = this.settledPromises.get(runId); + if (d) return d.promise; + const z = this.zombies.get(runId); + if (z) { + return Promise.resolve({ + status: z.intended.status, + error: z.intended.error, + terminalWriteFailed: true, + }); + } + return undefined; + } + + /** #487 test/diagnostic seam: whether a give-up zombie is held for this run. */ + hasZombie(runId: string): boolean { + return this.zombies.has(runId); + } + + /** #487: every zombie runId held on this replica (reconcile clause a, commit 4). */ + zombieRunIds(): string[] { + return [...this.zombies.keys()]; + } + + /** #487: create a one-shot deferred (resolve captured for a later single call). */ + private makeDeferred(): Deferred { + let resolve!: (value: T) => void; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; + } + + /** #487: resolve a run's settle notifier EXACTLY ONCE, then drop it (bounded). + * A subscriber that already grabbed the promise still resolves; a later one + * falls back to the zombie map / the row (see peekSettled). */ + private resolveSettled(runId: string, outcome: RunSettleOutcome): void { + const d = this.settledPromises.get(runId); + if (!d) return; + this.settledPromises.delete(runId); + d.resolve(outcome); + } + + /** #487: read the persisted terminal outcome when the conditional finalize was a + * no-op (the row was already terminal). Falls back to the intended status when + * the read fails or the row is unexpectedly missing/non-terminal. */ + private async readTerminalOutcome( + runId: string, + workspaceId: string, + fallbackStatus: RunTerminalStatus, + fallbackError: string | null, + ): Promise { + try { + 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, + }; + } + } catch { + // Fall through to the intended status — best-effort only. + } + return { + status: fallbackStatus, + error: fallbackError, + terminalWriteFailed: false, + }; } /** Small async backoff between terminal-write retries (F6). Isolated so it is diff --git a/apps/server/src/core/ai-chat/ai-chat.service.lifecycle.spec.ts b/apps/server/src/core/ai-chat/ai-chat.service.lifecycle.spec.ts index 66eb5b42..a6f3e75c 100644 --- a/apps/server/src/core/ai-chat/ai-chat.service.lifecycle.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat.service.lifecycle.spec.ts @@ -89,6 +89,11 @@ describe('AiChatService.stream run-lifecycle safety net (#184)', () => { const runRepo = { insert: jest.fn().mockResolvedValue({ id: 'run-1', status: 'running' }), update: jest.fn().mockResolvedValue({ id: 'run-1' }), + // #487: the terminal settle now goes through the CONDITIONAL write. + finalizeIfActive: jest + .fn() + .mockResolvedValue({ id: 'run-1', status: 'failed' }), + findById: jest.fn().mockResolvedValue(undefined), }; const runService = new AiChatRunService(runRepo as never, { isCloud: () => false } as never); @@ -148,9 +153,10 @@ describe('AiChatService.stream run-lifecycle safety net (#184)', () => { // The run was begun... expect(runRepo.insert).toHaveBeenCalledTimes(1); - // ...then settled to a terminal FAILED status by the safety net... - expect(runRepo.update).toHaveBeenCalledTimes(1); - expect(runRepo.update).toHaveBeenCalledWith( + // ...then settled to a terminal FAILED status by the safety net (via the + // #487 conditional write)... + expect(runRepo.finalizeIfActive).toHaveBeenCalledTimes(1); + expect(runRepo.finalizeIfActive).toHaveBeenCalledWith( 'run-1', 'ws1', expect.objectContaining({ status: 'failed' }), diff --git a/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.ts b/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.ts index 7bb6bcdb..e7976ebe 100644 --- a/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.ts +++ b/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.ts @@ -143,6 +143,41 @@ export class AiChatRunRepo { .executeTakeFirst(); } + /** + * #487: CONDITIONAL terminal finalize — flip a run to a terminal status and + * stamp `finished_at` ONLY while it is still active (pending|running), mirroring + * the assistant message's `onlyIfStreaming` guard. A double-settle (a late or + * second writer, a supersede applying a zombie's intended, a reconcile stamp) + * matches NOTHING once the row is terminal and is a benign no-op — so a terminal + * status can never be clobbered by a later writer (last-writer-wins is gone). + * + * Returns the updated row when it WAS active (this call wrote it), else + * undefined (the row was already terminal — another writer won). The caller + * distinguishes the two to resolve the correct settle outcome. + */ + async finalizeIfActive( + id: string, + workspaceId: string, + patch: { status: string; error: string | null }, + trx?: KyselyTransaction, + ): Promise { + const db = dbOrTx(this.db, trx); + const now = new Date(); + return db + .updateTable('aiChatRuns') + .set({ + status: patch.status, + error: patch.error, + finishedAt: now, + updatedAt: now, + }) + .where('id', '=', id) + .where('workspaceId', '=', workspaceId) + .where('status', 'in', ACTIVE_RUN_STATUSES as unknown as string[]) + .returning(this.baseFields) + .executeTakeFirst(); + } + /** * Mark an EXPLICIT stop request on an active run (distinct from a browser * disconnect, which never stops a run). Stamps `stop_requested_at` ONLY while diff --git a/apps/server/test/integration/ai-chat-run.int-spec.ts b/apps/server/test/integration/ai-chat-run.int-spec.ts index f6a753a1..aca31c03 100644 --- a/apps/server/test/integration/ai-chat-run.int-spec.ts +++ b/apps/server/test/integration/ai-chat-run.int-spec.ts @@ -281,6 +281,52 @@ describe('AiChatRun durable lifecycle [integration]', () => { }); }); + it('#487 finalizeIfActive is CONDITIONAL: a late terminal write cannot clobber the settled status (real SQL)', async () => { + const c = (await createChat(db, { workspaceId, creatorId: userId })).id; + const run = await runRepo.insert({ + chatId: c, + workspaceId, + createdBy: userId, + status: 'running', + }); + + // First terminal write: the run IS active, so it flips + returns the row. + const first = await runRepo.finalizeIfActive(run.id, workspaceId, { + status: 'succeeded', + error: null, + }); + expect(first!.status).toBe('succeeded'); + expect(first!.finishedAt).toBeTruthy(); + + // A late/second writer tries to flip it to 'aborted' — the WHERE status IN + // ('pending','running') guard matches NOTHING now, so it is a benign no-op. + const second = await runRepo.finalizeIfActive(run.id, workspaceId, { + status: 'aborted', + error: 'late clobber attempt', + }); + expect(second).toBeUndefined(); + + // The persisted terminal status is UNCHANGED — last-writer-wins is gone. + const row = await runRepo.findById(run.id, workspaceId); + expect(row!.status).toBe('succeeded'); + expect(row!.error).toBeNull(); + }); + + it('#487 double-settle through the service collapses to one write at the SQL gate', async () => { + const c = (await createChat(db, { workspaceId, creatorId: userId })).id; + const handle = await service.beginRun({ chatId: c, workspaceId, userId }); + + // First settle writes 'aborted' via the conditional write. + await service.finalizeRun(handle.runId, workspaceId, 'aborted'); + // A late safety-net settle to 'error' is a no-op (row already terminal). + await service.finalizeRun(handle.runId, workspaceId, 'error', 'late'); + + const row = await runRepo.findById(handle.runId, workspaceId); + expect(row!.status).toBe('aborted'); + expect(service.isLocallyActive(handle.runId)).toBe(false); + expect(service.hasZombie(handle.runId)).toBe(false); + }); + it('sweepRunning() with NO args (boot sweep / variant C) aborts even a FRESH running run', async () => { // F1/DECISION C at the SQL level: the unconditional boot sweep has NO // staleness window, so a run updated just now (a fast restart) is settled too