From b8aac9635bec9c59054e2e0836f992694d9cbc50 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sat, 11 Jul 2026 05:13:35 +0300 Subject: [PATCH] =?UTF-8?q?feat(ai-chat):=20=D1=80=D0=B5=D1=82=D1=80=D0=B0?= =?UTF-8?q?=D0=B8=20finalizeAssistant=20+=20owner-write=20=D0=BF=D1=80?= =?UTF-8?q?=D0=B8=D0=BE=D1=80=D0=B8=D1=82=D0=B5=D1=82=20+=20=D0=B4=D0=B2?= =?UTF-8?q?=D1=83=D1=81=D1=82=D0=BE=D1=80=D0=BE=D0=BD=D0=BD=D0=B8=D0=B9=20?= =?UTF-8?q?reconcile=20(#487)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Раньше finalizeAssistant ставил finalized=true ДО записи и не ретраил → один неудачный UPDATE = строка вечно 'streaming'; свип был boot-only; run-свип безусловный — асимметрия «run succeeded / message streaming навсегда». - finalizeAssistant: bounded-ретраи; once-гейт закрывается ТОЛЬКО после успешной записи; возвращает ok. Правило owner-write: терминальная запись owner'а условна на status='streaming' OR metadata.finalizeFailed (repo.finalizeOwner) — перетирает reconcile-штамп, но не проставленный терминал. ВСЕ status-only штампы reconcile (stampTerminalIfStreaming, sweepStreaming) пишут строго onlyIfStreaming И мёржат metadata.finalizeFailed:true (иначе поздний owner-write не перетрёт). - Порядок: попытка message-finalize → ран финализируется ВСЕГДА; при провале message onFinish помечает ран 'error' (не 'completed'). Ран не гейтится на message. - Периодический reconcile-джоб (setInterval, env-tunable) клаузами по порядку: (a) пере-драйв зомби; (b) message streaming + ран терминален → штамп по статусу рана (succeeded-ран + зависшая строка → 'aborted'+finalizeFailed, НЕ 'completed'-empty); (c) run running + НЕТ entry И НЕТ zombie + staleness → aborted (гейт «нет entry» первичный, staleness от last-progress updatedAt, X=max(2×per-call cap,15мин)+boot-warn); (d) message streaming + возраст>X + нет активной run-строки → aborted (двойной гейт). isInterruptResume исключает finalizeFailed-строки. Оппортунистический одно-чатовый reconcile при старте хода (best-effort, не фейлит ход). sweepStreaming boot-only → периодический. Тесты (реальная БД): owner finalizeOwner чистит finalizeFailed; штамп не перетирает терминал; поздний owner-write перетирает aborted-штамп; клаузы b/c/d (+живой entry не трогать, двойной гейт d); «убить БД на finish → после восстановления ни строка, ни ран не застряли». Юниты: finalizeFailed исключает interrupt-resume; reconcileStaleRuns «нет entry». Co-Authored-By: Claude Opus 4.8 (1M context) --- .../core/ai-chat/ai-chat-run.service.spec.ts | 32 ++ .../src/core/ai-chat/ai-chat-run.service.ts | 46 +++ .../ai-chat/ai-chat.controller.export.spec.ts | 13 +- .../ai-chat/ai-chat.service.run-race.spec.ts | 12 +- .../src/core/ai-chat/ai-chat.service.spec.ts | 51 ++- .../src/core/ai-chat/ai-chat.service.ts | 302 +++++++++++++++-- .../repos/ai-chat/ai-chat-message.repo.ts | 148 ++++++++- .../repos/ai-chat/ai-chat-run.repo.ts | 25 ++ .../integration/ai-chat-reconcile.int-spec.ts | 305 ++++++++++++++++++ 9 files changed, 889 insertions(+), 45 deletions(-) create mode 100644 apps/server/test/integration/ai-chat-reconcile.int-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 1518dc73..ade4563c 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 @@ -704,6 +704,38 @@ describe('#487 AiChatRunService.supersede (CAS)', () => { expect(await svc.supersede(chat, 'run-1', ws)).toEqual({ kind: 'degrade' }); }); + it('reconcileStaleRuns: aborts a stale run with NO entry/zombie; NEVER touches a live entry', async () => { + const finalizeIfActive = jest.fn(async () => ({ id: 'x', status: 'aborted' })); + const repo = makeRepo({ + insert: jest.fn(async (v: any) => ({ + id: 'live-1', + status: 'running', + chatId: v.chatId, + workspaceId: v.workspaceId, + })), + finalizeIfActive, + findStaleActive: jest.fn(async () => [ + { id: 'orphan-1', workspaceId: ws, chatId: 'c-orphan' }, + { id: 'live-1', workspaceId: ws, chatId: 'c-live' }, + ]), + }); + const svc = new AiChatRunService(repo as never, makeEnv() as never); + // A LIVE run this replica owns (in the `active` map). + await svc.beginRun({ chatId: 'c-live', workspaceId: ws, userId: 'u1' }); + expect(svc.isLocallyActive('live-1')).toBe(true); + + const aborted = await svc.reconcileStaleRuns(15 * 60 * 1000); + expect(aborted).toBe(1); + // The orphan (no entry) was aborted; the live entry was NEVER passed to the DB. + expect(finalizeIfActive).toHaveBeenCalledTimes(1); + expect(finalizeIfActive).toHaveBeenCalledWith( + 'orphan-1', + ws, + expect.objectContaining({ status: 'aborted' }), + ); + expect(svc.isLocallyActive('live-1')).toBe(true); + }); + it('gave-up zombie: supersede applies the intended status (settleZombie) then is ready', async () => { let healthy = false; let active: unknown = { 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 8fd577c8..1530c3b5 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 @@ -507,6 +507,52 @@ export class AiChatRunService implements OnModuleInit { } } + /** + * #487 reconcile clause (c): abort runs the DB still shows active (pending| + * running) but that this replica does NOT own — NO live entry AND NO zombie — + * and that have been UNTOUCHED past `staleMs` (from last-progress `updated_at`, + * NOT startedAt, so a legit long marathon is never a candidate). "No entry" is + * the PRIMARY gate: a live entry (an actively-executing run on this replica) is + * NEVER aborted, whatever its age. Returns the number aborted. Best-effort — + * never throws (a periodic-job failure must not crash the process). + */ + async reconcileStaleRuns(staleMs: number): Promise { + let candidates: Array<{ id: string; workspaceId: string; chatId: string }>; + try { + candidates = await this.runRepo.findStaleActive(staleMs); + } catch (err) { + this.logger.warn( + `Reconcile (stale runs) query failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + return 0; + } + let aborted = 0; + for (const c of candidates) { + // PRIMARY gate: never touch a live entry, and never race a zombie we are + // already re-driving (settleZombie owns those). + if (this.active.has(c.id) || this.zombies.has(c.id)) continue; + try { + const row = await this.runRepo.finalizeIfActive(c.id, c.workspaceId, { + status: 'aborted', + error: 'Run aborted by reconcile: no live runner (stale).', + }); + if (row) { + aborted += 1; + this.settled.add(c.id); + } + } catch (err) { + this.logger.warn( + `Reconcile abort of stale run ${c.id} failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + } + } + return aborted; + } + /** * #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). diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.export.spec.ts b/apps/server/src/core/ai-chat/ai-chat.controller.export.spec.ts index bfbc64c0..6afefc61 100644 --- a/apps/server/src/core/ai-chat/ai-chat.controller.export.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat.controller.export.spec.ts @@ -115,7 +115,7 @@ describe('finalizeAssistant dispatch (planFinalizeAssistant + applyFinalize)', ( // Drive the SAME applyFinalize the service calls (no duplicated logic). async function dispatchFinalize( - repo: { insert: jest.Mock; update: jest.Mock }, + repo: { insert: jest.Mock; finalizeOwner: jest.Mock }, assistantId: string | undefined, flushed: AssistantFlush, ): Promise { @@ -135,21 +135,22 @@ describe('finalizeAssistant dispatch (planFinalizeAssistant + applyFinalize)', ( expect(planFinalizeAssistant(undefined)).toEqual({ kind: 'insert' }); }); - it('(a) upfront insert succeeded -> finalize UPDATEs the row by id', async () => { - const repo = { insert: jest.fn(), update: jest.fn() }; + it('(a) upfront insert succeeded -> finalize CONDITIONALLY updates the row by id (#487 owner-write)', async () => { + const repo = { insert: jest.fn(), finalizeOwner: jest.fn() }; const flushed = flushAssistant([], 'final answer', 'completed', { finishReason: 'stop', }); await dispatchFinalize(repo, 'a1', flushed); - expect(repo.update).toHaveBeenCalledWith('a1', workspaceId, flushed); + // #487: the owner write is the CONDITIONAL finalizeOwner, not a raw update. + expect(repo.finalizeOwner).toHaveBeenCalledWith('a1', workspaceId, flushed); expect(repo.insert).not.toHaveBeenCalled(); }); it('(b) upfront insert failed -> finalize INSERTs the terminal payload', async () => { - const repo = { insert: jest.fn(), update: jest.fn() }; + const repo = { insert: jest.fn(), finalizeOwner: jest.fn() }; const flushed = flushAssistant([], 'partial', 'error', { error: 'boom' }); await dispatchFinalize(repo, undefined, flushed); - expect(repo.update).not.toHaveBeenCalled(); + expect(repo.finalizeOwner).not.toHaveBeenCalled(); expect(repo.insert).toHaveBeenCalledTimes(1); const arg = repo.insert.mock.calls[0][0]; // The fallback insert carries the terminal content/status/metadata. diff --git a/apps/server/src/core/ai-chat/ai-chat.service.run-race.spec.ts b/apps/server/src/core/ai-chat/ai-chat.service.run-race.spec.ts index d167abbe..8b170c11 100644 --- a/apps/server/src/core/ai-chat/ai-chat.service.run-race.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat.service.run-race.spec.ts @@ -155,6 +155,8 @@ describe('AiChatService.stream — abortSignal wiring (#184 F3)', () => { insert: jest.fn(async () => ({ id: 'msg-1' })), findAllByChat: jest.fn(async () => []), update: jest.fn(async () => ({ id: 'msg-1' })), + finalizeOwner: jest.fn(async () => ({ id: 'msg-1' })), + findStreamingWithTerminalRun: jest.fn(async () => []), }; const aiSettings = { resolve: jest.fn(async () => ({})) }; const tools = { forUser: jest.fn(async () => ({})) }; @@ -332,7 +334,13 @@ describe('AiChatService.stream — abortSignal wiring (#184 F3)', () => { usage: {}, steps: [], }); - expect(runHooks.onSettled).toHaveBeenCalledWith('run-1', 'completed'); + // #487: onFinish passes the (undefined) error slot so a message-finalize + // failure could error-mark the run; on the success path it is undefined. + expect(runHooks.onSettled).toHaveBeenCalledWith( + 'run-1', + 'completed', + undefined, + ); }); it('F9: onAbort settles the run "aborted"', async () => { @@ -415,6 +423,8 @@ describe('AiChatService.stream — begin-failure fails the turn (#184 F14 / #486 insert: jest.fn(async () => ({ id: 'msg-1' })), findAllByChat: jest.fn(async () => []), update: jest.fn(async () => ({ id: 'msg-1' })), + finalizeOwner: jest.fn(async () => ({ id: 'msg-1' })), + findStreamingWithTerminalRun: jest.fn(async () => []), }; const aiSettings = { resolve: jest.fn(async () => ({})) }; const tools = { forUser: jest.fn(async () => ({})) }; diff --git a/apps/server/src/core/ai-chat/ai-chat.service.spec.ts b/apps/server/src/core/ai-chat/ai-chat.service.spec.ts index 46e37f61..cdfa80e9 100644 --- a/apps/server/src/core/ai-chat/ai-chat.service.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat.service.spec.ts @@ -1315,8 +1315,12 @@ describe('AiChatService page-change lifecycle (#274)', () => { describe('isInterruptResume', () => { // history tail is the just-inserted user row; [len-2] is the previous turn. const withPrev = ( - prev: { role: string; status?: string | null } | null, - ): Array<{ role: string; status?: string | null }> => + prev: { + role: string; + status?: string | null; + metadata?: unknown; + } | null, + ): Array<{ role: string; status?: string | null; metadata?: unknown }> => prev ? [prev, { role: 'user', status: null }] : [{ role: 'user', status: null }]; @@ -1357,6 +1361,33 @@ describe('isInterruptResume', () => { it('false when there is no preceding turn (only the new user row)', () => { expect(isInterruptResume(withPrev(null), true)).toBe(false); }); + + it('#487 EXCLUDES a reconcile stamp (finalizeFailed) — not a genuine interruption', () => { + // A row a reconcile settled to 'aborted' carries metadata.finalizeFailed. It + // must NOT be treated as an interrupt-resume (that would inject a false + // "you were interrupted" note), even though its status is 'aborted'. + expect( + isInterruptResume( + withPrev({ + role: 'assistant', + status: 'aborted', + metadata: { finalizeFailed: true }, + }), + true, + ), + ).toBe(false); + // A genuine abort (no finalizeFailed) still counts. + expect( + isInterruptResume( + withPrev({ + role: 'assistant', + status: 'aborted', + metadata: { parts: [] }, + }), + true, + ), + ).toBe(true); + }); }); /** @@ -1419,6 +1450,9 @@ describe('AiChatService.stream — resumable pipe options (#184 phase 1.5)', () insert: jest.fn(async () => ({ id: 'msg-1' })), findAllByChat: jest.fn(async () => []), update: jest.fn(async () => ({ id: 'msg-1' })), + // #487: the terminal owner-write + the opportunistic reconcile query. + finalizeOwner: jest.fn(async () => ({ id: 'msg-1' })), + findStreamingWithTerminalRun: jest.fn(async () => []), }; const aiSettings = { resolve: jest.fn(async () => ({})) }; const tools = { forUser: jest.fn(async () => ({})) }; @@ -1623,6 +1657,19 @@ describe('AiChatService.stream — token-degeneration reaction (#444)', () => { return { id }; }, ), + // #487: the terminal owner-write records into the SAME `updated` recorder so + // assertions on the terminal 'completed'/'error'/'aborted' write still hold. + finalizeOwner: jest.fn( + async ( + id: string, + workspaceId: string, + patch: Record, + ) => { + updated.push({ id, workspaceId, patch }); + return { id }; + }, + ), + findStreamingWithTerminalRun: jest.fn(async () => []), }; const aiSettings = { resolve: jest.fn(async () => ({})) }; const tools = { forUser: jest.fn(async () => ({})) }; 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 4ccca58a..f0b90f5b 100644 --- a/apps/server/src/core/ai-chat/ai-chat.service.ts +++ b/apps/server/src/core/ai-chat/ai-chat.service.ts @@ -3,6 +3,7 @@ import { ForbiddenException, Injectable, Logger, + OnModuleDestroy, OnModuleInit, ServiceUnavailableException, } from '@nestjs/common'; @@ -42,7 +43,11 @@ import { makeLoadToolsTool, buildExternalToolCatalog, } from './tools/tool-tiers'; -import { RunAlreadyActiveError } from './ai-chat-run.service'; +import { + RunAlreadyActiveError, + AiChatRunService, +} from './ai-chat-run.service'; +import { inAppToolCallCapMs } from './tools/ai-chat-tools.service'; import { computePageChange } from './page-change/page-change.util'; import { sanitizeSelection, @@ -250,15 +255,23 @@ export function cleanGeneratedTitle(text: string): string { * partial output is already in history thanks to the step-granular write path). */ export function isInterruptResume( - history: Array<{ role: string; status?: string | null }>, + history: Array<{ + role: string; + status?: string | null; + metadata?: unknown; + }>, clientInterrupted: boolean | undefined, ): boolean { if (clientInterrupted !== true) return false; const prev = history[history.length - 2]; - return ( - prev?.role === 'assistant' && - (prev.status === 'aborted' || prev.status === 'streaming') - ); + if (prev?.role !== 'assistant') return false; + // #487: a reconcile STAMP (metadata.finalizeFailed) is NOT a genuine user + // interruption — the previous turn's process died and a reconcile settled the + // row as 'aborted'. Treating it as an interrupt-resume would inject a false + // "you were interrupted" note. Exclude any finalizeFailed row. + const meta = prev.metadata as { finalizeFailed?: unknown } | null | undefined; + if (meta && meta.finalizeFailed === true) return false; + return prev.status === 'aborted' || prev.status === 'streaming'; } /** @@ -457,7 +470,7 @@ export interface AiChatStreamArgs { * can be rebuilt for `convertToModelMessages`. */ @Injectable() -export class AiChatService implements OnModuleInit { +export class AiChatService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(AiChatService.name); constructor( @@ -478,8 +491,17 @@ export class AiChatService implements OnModuleInit { // constructions (int-specs) compile unchanged; Nest always injects the real // provider in production. Only ever touched on the run-wrapped + flag-on path. private readonly streamRegistry?: AiChatStreamRegistryService, + // #487: the run lifecycle service, for the periodic + opportunistic reconcile + // (zombie re-drive + stale-run abort). OPTIONAL so positional test + // constructions compile unchanged; Nest always injects the real singleton, so + // reconcile sees the SAME in-memory active/zombie maps the runner mutates. + private readonly aiChatRunService?: AiChatRunService, ) {} + // #487: periodic reconcile timer (single-process phase 1). Started in + // onModuleInit, cleared in onModuleDestroy. + private reconcileTimer?: ReturnType; + /** * Crash-recovery sweep on server start (#183): any assistant row left in the * 'streaming' state is the relic of a turn whose process died before it @@ -504,6 +526,158 @@ export class AiChatService implements OnModuleInit { }`, ); } + + // #487: start the PERIODIC reconcile (was boot-only). It heals both directions + // of the run<->message lifecycle asymmetry that a boot sweep alone left to the + // NEXT restart. Single-process phase 1: the in-memory active/zombie maps are + // authoritative, so "no live entry" is a safe primary gate. + const staleMs = this.reconcileStalenessMs(); + // boot-warn if the per-call cap is configured so high the derived staleness is + // unusually long (a stale run then lingers longer before reconcile aborts it). + if (staleMs > 30 * 60 * 1000) { + this.logger.warn( + `#487 reconcile staleness is ${Math.round(staleMs / 60000)}min ` + + `(derived from max(2 x per-call cap, 15min)); a per-call cap this high ` + + `delays stale-run recovery. Review AI_CHAT_INAPP_TOOL_CALL_CAP_MS.`, + ); + } + const intervalMs = this.reconcileIntervalMs(); + this.reconcileTimer = setInterval(() => { + void this.reconcile().catch((err) => { + this.logger.warn( + `Periodic reconcile failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + }); + }, intervalMs); + this.reconcileTimer.unref?.(); + } + + /** #487: stop the periodic reconcile timer on shutdown. */ + onModuleDestroy(): void { + if (this.reconcileTimer) { + clearInterval(this.reconcileTimer); + this.reconcileTimer = undefined; + } + } + + /** + * #487: reconcile staleness threshold X — a run/message is only a "no live + * runner" abort candidate once UNTOUCHED past this. Derived as + * max(2 x per-call cap, 15min): 2x the longest legitimate single tool call plus + * a floor, so a marathon turn making steady progress (updatedAt bumped each + * step) is never swept. + */ + private reconcileStalenessMs(): number { + return Math.max(2 * inAppToolCallCapMs(), 15 * 60 * 1000); + } + + /** #487: how often the periodic reconcile runs (env-tunable, default 2min). */ + private reconcileIntervalMs(): number { + const raw = Number(process.env.AI_CHAT_RECONCILE_INTERVAL_MS); + return Number.isFinite(raw) && raw > 0 ? raw : 2 * 60 * 1000; + } + + /** + * #487: the periodic BIDIRECTIONAL reconcile. Runs the clauses IN ORDER; each is + * best-effort (a failure of one never blocks the others). Single-process phase 1 + * — the run service's in-memory maps are authoritative for "live entry". + * + * (a) re-drive ZOMBIE runs (a terminal write that gave up) — apply the intended + * status via the conditional UPDATE; + * (b) message 'streaming' + its RUN terminal -> stamp the message by the run's + * status (succeeded-run + stuck row -> 'aborted'+finalizeFailed, NOT + * 'completed' with empty parts — the final text lived only in the dead + * process's memory, a documented loss); + * (c) run active + NO live entry + NO zombie + stale -> aborted (the run + * service applies the "no entry" primary gate + last-progress staleness); + * (d) message 'streaming' + age>X + NO active run on the chat -> aborted + * (historical-row safety, double-gated). + */ + async reconcile(): Promise { + const staleMs = this.reconcileStalenessMs(); + + // (a) zombie re-drive. + if (this.aiChatRunService) { + for (const runId of this.aiChatRunService.zombieRunIds()) { + try { + await this.aiChatRunService.settleZombie(runId); + } catch (err) { + this.logger.warn( + `Reconcile (a) zombie ${runId} re-drive failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + } + } + } + + // (b) message streaming + run terminal -> stamp message by run status. + try { + const stuck = await this.aiChatMessageRepo.findStreamingWithTerminalRun(); + for (const s of stuck) { + // succeeded-run -> 'aborted' (NOT 'completed'-empty); failed -> 'error'; + // aborted -> 'aborted'. All via the finalizeFailed stamp. + const status = s.runStatus === 'failed' ? 'error' : 'aborted'; + await this.aiChatMessageRepo.stampTerminalIfStreaming( + s.messageId, + s.workspaceId, + status, + ); + } + } catch (err) { + this.logger.warn( + `Reconcile (b) message<-run failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + } + + // (c) stale active run with no live runner -> aborted. + if (this.aiChatRunService) { + try { + await this.aiChatRunService.reconcileStaleRuns(staleMs); + } catch (err) { + this.logger.warn( + `Reconcile (c) stale-run abort failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + } + } + + // (d) historical streaming row, no active run on the chat, stale -> aborted. + try { + await this.aiChatMessageRepo.sweepStreamingWithoutActiveRun(staleMs); + } catch (err) { + this.logger.warn( + `Reconcile (d) historical-row sweep failed: ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + } + } + + /** + * #487: OPPORTUNISTIC single-chat reconcile at the start of a turn (beginRun / + * supersede path), so a user who returns to a chat with a stuck streaming row + * (its run already terminal) sees it settled WITHOUT waiting for the periodic + * job. Best-effort — a failure NEVER fails the turn (swallowed by the caller). + */ + async reconcileChat(chatId: string, workspaceId: string): Promise { + const stuck = await this.aiChatMessageRepo.findStreamingWithTerminalRun(50, { + chatId, + workspaceId, + }); + for (const s of stuck) { + const status = s.runStatus === 'failed' ? 'error' : 'aborted'; + await this.aiChatMessageRepo.stampTerminalIfStreaming( + s.messageId, + s.workspaceId, + status, + ); + } } /** @@ -842,6 +1016,20 @@ export class AiChatService implements OnModuleInit { } } + // #487: opportunistic single-chat reconcile — settle any streaming row on this + // chat whose run is already terminal BEFORE this turn's history load, so the + // user never waits on the periodic job and the new turn's model history is not + // polluted by a stuck 'streaming' row. Best-effort: it must NEVER fail the turn. + try { + await this.reconcileChat(chatId, workspace.id); + } catch (err) { + this.logger.debug( + `Opportunistic reconcile for chat ${chatId} failed (ignored): ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + } + try { // Extract the incoming user turn (the last user message from useChat). const incoming = lastUserMessage(body.messages); @@ -1204,29 +1392,59 @@ export class AiChatService implements OnModuleInit { // callbacks — mirroring the pre-#183 persist-at-most-once guard for the // TERMINAL status (the row may be updated many times with 'streaming' before // this fires once). + // #487: the once-gate closes ONLY AFTER a successful write, and the write is + // BOUNDED-RETRIED. Previously `finalized` was set BEFORE the write and never + // retried, so a single failed UPDATE stranded the row 'streaming' forever + // (the boot-only sweep was the only recovery). Now a transient blip is ridden + // out in place; a total give-up leaves the gate OPEN and logs, and the + // periodic reconcile (clauses b/d) later settles the row. Returns whether the + // terminal write LANDED, so the caller can error-mark the RUN on a message + // failure (the run is finalized regardless — never gated on the message). let finalized = false; + const FINALIZE_MSG_MAX_ATTEMPTS = 3; const finalizeAssistant = async ( flushed: AssistantFlush, - ): Promise => { - if (finalized) return; - finalized = true; + ): Promise => { + if (finalized) return true; const plan = planFinalizeAssistant(assistantId); - try { - // Shared dispatch (see applyFinalize): UPDATE the upfront row, or — when - // the upfront insert failed (kind 'insert') — INSERT the terminal row as - // the only safety against losing the turn entirely. - await applyFinalize( - this.aiChatMessageRepo, - plan, - { chatId, workspaceId: workspace.id, userId: user.id }, - flushed, - ); - } catch (err) { - this.logger.error( - `Failed to finalize assistant message (kind=${plan.kind})`, - err as Error, - ); + let lastError: unknown; + for (let attempt = 1; attempt <= FINALIZE_MSG_MAX_ATTEMPTS; attempt++) { + try { + // Shared dispatch (see applyFinalize): conditionally UPDATE the upfront + // row (owner-write priority), or — when the upfront insert failed (kind + // 'insert') — INSERT the terminal row as the only safety against losing + // the turn entirely. + await applyFinalize( + this.aiChatMessageRepo, + plan, + { chatId, workspaceId: workspace.id, userId: user.id }, + flushed, + ); + finalized = true; // gate closes ONLY after a successful write + return true; + } catch (err) { + lastError = err; + this.logger.warn( + `Assistant message finalize attempt ${attempt}/${FINALIZE_MSG_MAX_ATTEMPTS} ` + + `failed (kind=${plan.kind}): ${ + err instanceof Error ? err.message : 'unknown error' + }`, + ); + if (attempt < FINALIZE_MSG_MAX_ATTEMPTS) { + await new Promise((r) => setTimeout(r, 50 * attempt)); + } + } } + // Gave up: leave the gate OPEN (no in-process second settler exists — the + // terminal callbacks are mutually exclusive) and log. The periodic reconcile + // settles the stranded row; a late owner-write is impossible for this turn, + // so the reconcile stamp (aborted+finalizeFailed) is the final state. + this.logger.error( + `Assistant message finalize GAVE UP after ${FINALIZE_MSG_MAX_ATTEMPTS} ` + + `attempts (row left 'streaming', chat ${chatId}); reconcile will settle it`, + lastError as Error, + ); + return false; }; // DIAGNOSTIC (Safari stream-drop investigation) — temporary. Measure @@ -1371,7 +1589,7 @@ export class AiChatService implements OnModuleInit { const stepExhausted = steps.length >= MAX_AGENT_STEPS; const emptyTurnMarker = !producedText && stepExhausted ? STEP_LIMIT_NO_ANSWER_MARKER : ''; - await finalizeAssistant( + const msgOk = await finalizeAssistant( flushAssistant(steps as StepLike[], emptyTurnMarker, 'completed', { finishReason: finishReason as string, usage: totalUsage as StreamUsage, @@ -1385,9 +1603,19 @@ export class AiChatService implements OnModuleInit { pageChanged, }), ); - // #184: settle the RUN as succeeded (best-effort, after the projection - // is finalized above). - if (runId) await runHooks?.onSettled?.(runId, 'completed'); + // #184/#487: the RUN is finalized ALWAYS (never gated on the message). + // If the message finalize GAVE UP, error-mark the run so the asymmetry + // "run succeeded / message streaming forever" cannot arise; the + // periodic reconcile then settles the stuck message from this run. + if (runId) { + await runHooks?.onSettled?.( + runId, + msgOk ? 'completed' : 'error', + msgOk + ? undefined + : 'Assistant message could not be persisted (finalize failed).', + ); + } // Lifecycle: release the external MCP clients leased for this turn. await closeExternalClients(); @@ -2128,7 +2356,10 @@ export function planFinalizeAssistant( * a test mock both satisfy it). */ export interface FinalizeRepo { insert(insertable: Record): Promise; - update( + // #487: the OWNER terminal write is CONDITIONAL (status='streaming' OR + // metadata.finalizeFailed) so the owner overwrites a reconcile stamp but never + // an already-proper terminal row (owner-write priority). + finalizeOwner( id: string, workspaceId: string, patch: AssistantFlush, @@ -2137,10 +2368,11 @@ export interface FinalizeRepo { /** * Apply a finalize `plan` to the repo with the terminal `flushed` payload (#183): - * UPDATE the upfront row, or INSERT a fresh terminal row as the fallback when the - * upfront insert failed. The SINGLE dispatch shared by the service's - * finalizeAssistant and its test, so the test exercises the real path instead of - * a copy (#186 review). Pure of error handling — the caller wraps it. + * conditionally UPDATE the upfront row (owner-write priority, #487), or INSERT a + * fresh terminal row as the fallback when the upfront insert failed. The SINGLE + * dispatch shared by the service's finalizeAssistant and its test, so the test + * exercises the real path instead of a copy (#186 review). Pure of error + * handling — the caller wraps it (and RETRIES it, #487). */ export async function applyFinalize( repo: FinalizeRepo, @@ -2149,7 +2381,7 @@ export async function applyFinalize( flushed: AssistantFlush, ): Promise { if (plan.kind === 'update') { - await repo.update(plan.id, base.workspaceId, flushed); + await repo.finalizeOwner(plan.id, base.workspaceId, flushed); return; } await repo.insert({ diff --git a/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.ts b/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.ts index 78cc7064..96070e77 100644 --- a/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.ts +++ b/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.ts @@ -1,5 +1,6 @@ import { Injectable, Logger } from '@nestjs/common'; import { InjectKysely } from 'nestjs-kysely'; +import { sql } from 'kysely'; import { KyselyDB, KyselyTransaction } from '../../types/kysely.types'; import { dbOrTx } from '../../utils'; import { @@ -188,6 +189,144 @@ export class AiChatMessageRepo { return query.returning(this.baseFields).executeTakeFirst(); } + /** + * #487 OWNER terminal write — the streamText terminal callback's finalize. Like + * `update` but CONDITIONAL on `status='streaming' OR metadata.finalizeFailed`: + * the owner writes its real content EITHER when the row is still streaming (the + * normal case) OR when a reconcile stamp already flipped it to a terminal status + * but marked `finalizeFailed:true` — the owner's real content OVERWRITES that + * placeholder stamp (owner-write priority, #487). A row that is properly terminal + * (no finalizeFailed) is left untouched (undefined) — idempotent. The `patch` + * carries the real metadata WITHOUT finalizeFailed, so a successful write CLEARS + * the flag. Returns the updated row, or undefined when nothing matched. + */ + async finalizeOwner( + id: string, + workspaceId: string, + patch: Partial<{ + content: string | null; + toolCalls: unknown; + metadata: unknown; + status: string | null; + }>, + trx?: KyselyTransaction, + ): Promise { + const db = dbOrTx(this.db, trx); + return db + .updateTable('aiChatMessages') + .set({ ...(patch as Record), updatedAt: new Date() }) + .where('id', '=', id) + .where('workspaceId', '=', workspaceId) + .where((eb) => + eb.or([ + eb('status', '=', 'streaming'), + eb(sql`(metadata->>'finalizeFailed')`, '=', 'true'), + ]), + ) + .returning(this.baseFields) + .executeTakeFirst(); + } + + /** + * #487 RECONCILE status-only stamp — settle a stuck 'streaming' row to a + * terminal status WITHOUT the owner's real content (which lived only in the + * dead process's memory — a documented loss). CONDITIONAL on `status='streaming'` + * (never touches an already-terminal row) AND it MERGES `finalizeFailed:true` + * into metadata (preserving the partial `parts` already persisted) so a LATER + * owner-write (finalizeOwner) can still OVERWRITE this placeholder with real + * content, and so `isInterruptResume` can EXCLUDE this row (a reconcile stamp is + * not a genuine user interruption). Returns the updated row, or undefined. + */ + async stampTerminalIfStreaming( + id: string, + workspaceId: string, + status: 'aborted' | 'error' | 'completed', + trx?: KyselyTransaction, + ): Promise { + const db = dbOrTx(this.db, trx); + return db + .updateTable('aiChatMessages') + .set({ + status, + metadata: sql`coalesce(metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`, + updatedAt: new Date(), + }) + .where('id', '=', id) + .where('workspaceId', '=', workspaceId) + .where('status', '=', 'streaming') + .returning(this.baseFields) + .executeTakeFirst(); + } + + /** + * #487 reconcile clause (b): streaming assistant rows whose linked RUN has + * already reached a terminal status — an asymmetry ("run settled / message + * streaming forever") the periodic reconcile heals by stamping the message. + * Returns the message id + its run's terminal status, bounded. + */ + async findStreamingWithTerminalRun( + limit = 200, + // #487: scope to ONE chat for the opportunistic per-turn reconcile (removes + // reconcile latency from the user-visible path); omit for the periodic sweep. + chat?: { chatId: string; workspaceId: string }, + ): Promise< + Array<{ messageId: string; workspaceId: string; runStatus: string }> + > { + let query = this.db + .selectFrom('aiChatMessages as m') + .innerJoin('aiChatRuns as r', 'r.assistantMessageId', 'm.id') + .select([ + 'm.id as messageId', + 'm.workspaceId as workspaceId', + 'r.status as runStatus', + ]) + .where('m.status', '=', 'streaming') + .where('r.status', 'in', ['succeeded', 'failed', 'aborted']); + if (chat) { + query = query + .where('m.chatId', '=', chat.chatId) + .where('m.workspaceId', '=', chat.workspaceId); + } + return query.limit(limit).execute(); + } + + /** + * #487 reconcile clause (d) — historical-row safety: streaming rows older than + * `staleMs` whose chat has NO active run row (double-gated). Settle them to + * 'aborted' + finalizeFailed (so a late owner-write could still overwrite). + * Returns the count. Used ONLY by the periodic reconcile, never at boot. + */ + async sweepStreamingWithoutActiveRun( + staleMs: number, + trx?: KyselyTransaction, + ): Promise { + const db = dbOrTx(this.db, trx); + const staleBefore = new Date(Date.now() - staleMs); + const rows = await db + .updateTable('aiChatMessages as m') + .set({ + status: 'aborted', + metadata: sql`coalesce(m.metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`, + updatedAt: new Date(), + }) + .where('m.status', '=', 'streaming') + .where('m.updatedAt', '<', staleBefore) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom('aiChatRuns as r') + .select('r.id') + .whereRef('r.chatId', '=', 'm.chatId') + .where('r.status', 'in', ['pending', 'running']), + ), + ), + ) + .returning('m.id') + .execute(); + return rows.length; + } + /** * Crash-recovery sweep (#183): flip every assistant row still left in the * 'streaming' state (a turn that died mid-write before reaching a terminal @@ -200,13 +339,20 @@ export class AiChatMessageRepo { * step, so an actively-streaming row never matches; this prevents a fresh * replica's boot-sweep from aborting a turn another replica is still streaming * in a multi-instance deploy. + * + * #487: the sweep now ALSO marks `finalizeFailed:true` so a late owner-write can + * overwrite this placeholder with real content (owner-write priority). */ async sweepStreaming(trx?: KyselyTransaction): Promise { const db = dbOrTx(this.db, trx); const staleBefore = new Date(Date.now() - SWEEP_STREAMING_STALE_MS); const rows = await db .updateTable('aiChatMessages') - .set({ status: 'aborted', updatedAt: new Date() }) + .set({ + status: 'aborted', + metadata: sql`coalesce(metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`, + updatedAt: new Date(), + }) .where('status', '=', 'streaming') .where('updatedAt', '<', staleBefore) .returning('id') 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 e7976ebe..c27480bf 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 @@ -219,6 +219,31 @@ export class AiChatRunRepo { * sweeps only runs UNTOUCHED past the window. Phase 1 is single-process, so the * boot path supplies no window. */ + /** + * #487 reconcile clause (c): active (pending|running) runs UNTOUCHED past + * `staleMs` — candidates for "no live runner" abort. Staleness is measured from + * `updated_at` (the LAST-PROGRESS timestamp — recordStep bumps it), NOT + * `started_at`, so a legitimate long-running marathon (11–25 min of steady + * progress) is never a candidate. The caller filters these against its in-memory + * `active` / zombie maps ("no entry" is the PRIMARY gate — a live entry is never + * aborted) before settling any of them. Bounded. + */ + async findStaleActive( + staleMs: number, + limit = 200, + trx?: KyselyTransaction, + ): Promise> { + const db = dbOrTx(this.db, trx); + const staleBefore = new Date(Date.now() - staleMs); + return db + .selectFrom('aiChatRuns') + .select(['id', 'workspaceId', 'chatId']) + .where('status', 'in', ACTIVE_RUN_STATUSES as unknown as string[]) + .where('updatedAt', '<', staleBefore) + .limit(limit) + .execute(); + } + async sweepRunning( opts: { staleMs?: number } = {}, trx?: KyselyTransaction, diff --git a/apps/server/test/integration/ai-chat-reconcile.int-spec.ts b/apps/server/test/integration/ai-chat-reconcile.int-spec.ts new file mode 100644 index 00000000..4a38f1c2 --- /dev/null +++ b/apps/server/test/integration/ai-chat-reconcile.int-spec.ts @@ -0,0 +1,305 @@ +import { Kysely } from 'kysely'; +import { AiChatMessageRepo } from '@docmost/db/repos/ai-chat/ai-chat-message.repo'; +import { AiChatRunRepo } from '@docmost/db/repos/ai-chat/ai-chat-run.repo'; +import { AiChatRunService } from '../../src/core/ai-chat/ai-chat-run.service'; +import { + getTestDb, + destroyTestDb, + createWorkspace, + createUser, + createChat, + createMessage, +} from './db'; + +/** + * #487 commit 4 — bidirectional reconcile + owner-write priority, real SQL. + * + * Proves the OBSERVABLE recovery properties against docmost_test: + * - the CONDITIONAL owner-write beats a reconcile stamp, and a stamp never + * clobbers a proper terminal row; + * - a LATE owner-finalize with real content OVERWRITES a reconcile 'aborted' + * stamp (finalizeFailed); + * - each reconcile clause (b message<-run, c stale-run, d historical row) settles + * the stuck row/run, and a LIVE run entry is never touched; + * - the "kill DB on finish" recovery: after the DB comes back, neither the + * message row nor the run row stays stuck. + */ +describe('#487 reconcile + owner-write priority [integration]', () => { + let db: Kysely; + let messageRepo: AiChatMessageRepo; + let runRepo: AiChatRunRepo; + let runService: AiChatRunService; + let workspaceId: string; + let userId: string; + + beforeAll(async () => { + db = getTestDb(); + messageRepo = new AiChatMessageRepo(db as any); + runRepo = new AiChatRunRepo(db as any); + runService = new AiChatRunService(runRepo, { isCloud: () => false } as never); + workspaceId = (await createWorkspace(db)).id; + userId = (await createUser(db, workspaceId)).id; + }); + + afterAll(async () => { + await destroyTestDb(); + }); + + const newChat = async () => + (await createChat(db, { workspaceId, creatorId: userId })).id; + + const metaOf = async (id: string): Promise | null> => { + const row = await messageRepo.findById(id, workspaceId); + return (row?.metadata as Record | null) ?? null; + }; + + it('owner finalizeOwner writes a streaming row and CLEARS finalizeFailed', async () => { + const chatId = await newChat(); + const m = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'streaming', + metadata: { parts: [] }, + }); + const wrote = await messageRepo.finalizeOwner(m.id, workspaceId, { + content: 'final answer', + status: 'completed', + metadata: { parts: [{ type: 'text', text: 'final answer' }] }, + } as never); + expect(wrote!.status).toBe('completed'); + expect((await metaOf(m.id))?.finalizeFailed).toBeUndefined(); + }); + + it('a reconcile stamp NEVER clobbers a proper terminal row (finalizeOwner is a no-op there)', async () => { + const chatId = await newChat(); + const m = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'completed', + content: 'real', + metadata: { parts: [] }, + }); + // The reconcile stamp is onlyIfStreaming -> no-op on a completed row. + const stamped = await messageRepo.stampTerminalIfStreaming( + m.id, + workspaceId, + 'aborted', + ); + expect(stamped).toBeUndefined(); + expect((await messageRepo.findById(m.id, workspaceId))!.status).toBe( + 'completed', + ); + }); + + it('LATE owner-finalize with real content OVERWRITES a reconcile aborted stamp', async () => { + const chatId = await newChat(); + const m = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'streaming', + metadata: { parts: [{ type: 'text', text: 'partial' }] }, + }); + // Reconcile stamps it aborted + finalizeFailed (final text lived only in mem). + const stamped = await messageRepo.stampTerminalIfStreaming( + m.id, + workspaceId, + 'aborted', + ); + expect(stamped!.status).toBe('aborted'); + expect((await metaOf(m.id))?.finalizeFailed).toBe(true); + + // A LATE owner-write (finalizeFailed=true satisfies the OR) overwrites it with + // real content, clearing the flag — owner-write priority. + const wrote = await messageRepo.finalizeOwner(m.id, workspaceId, { + content: 'the real final answer', + status: 'completed', + metadata: { parts: [{ type: 'text', text: 'the real final answer' }] }, + } as never); + expect(wrote!.status).toBe('completed'); + expect(wrote!.content).toBe('the real final answer'); + expect((await metaOf(m.id))?.finalizeFailed).toBeUndefined(); + }); + + it('clause (c): a stale active run with NO live entry -> aborted; a LIVE entry is untouched', async () => { + // Stale run, NOT owned by this replica (no entry) -> reconcile aborts it. + const staleChat = await newChat(); + const stale = await runRepo.insert({ + chatId: staleChat, + workspaceId, + createdBy: userId, + status: 'running', + }); + await db + .updateTable('aiChatRuns') + .set({ updatedAt: new Date(Date.now() - 60 * 60 * 1000) }) + .where('id', '=', stale.id) + .execute(); + + // A live run OWNED by this replica (beginRun registers an in-memory entry), + // ALSO backdated stale — the "no entry" primary gate must protect it. + const liveChat = await newChat(); + const live = await runService.beginRun({ + chatId: liveChat, + workspaceId, + userId, + }); + await db + .updateTable('aiChatRuns') + .set({ updatedAt: new Date(Date.now() - 60 * 60 * 1000) }) + .where('id', '=', live.runId) + .execute(); + + const aborted = await runService.reconcileStaleRuns(15 * 60 * 1000); + expect(aborted).toBeGreaterThanOrEqual(1); + expect((await runRepo.findById(stale.id, workspaceId))!.status).toBe( + 'aborted', + ); + // The live entry is NEVER aborted, however stale its row looks. + expect((await runRepo.findById(live.runId, workspaceId))!.status).toBe( + 'running', + ); + expect(runService.isLocallyActive(live.runId)).toBe(true); + + // cleanup the live run + await runService.finalizeRun(live.runId, workspaceId, 'aborted'); + }); + + it('clause (b): a streaming message whose RUN is terminal is stamped by run status (succeeded -> aborted, NOT completed-empty)', async () => { + const chatId = await newChat(); + const msg = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'streaming', + metadata: { parts: [] }, + }); + // A SUCCEEDED run linked to the still-streaming message (the asymmetry). + const run = await runRepo.insert({ + chatId, + workspaceId, + createdBy: userId, + status: 'running', + assistantMessageId: msg.id, + }); + await runRepo.finalizeIfActive(run.id, workspaceId, { + status: 'succeeded', + error: null, + }); + + const stuck = await messageRepo.findStreamingWithTerminalRun(); + const mine = stuck.find((s) => s.messageId === msg.id); + expect(mine?.runStatus).toBe('succeeded'); + // Reconcile clause (b): succeeded run -> message 'aborted' (NOT 'completed'), + // the final text lived only in memory (documented loss), +finalizeFailed. + const status = mine!.runStatus === 'failed' ? 'error' : 'aborted'; + await messageRepo.stampTerminalIfStreaming(msg.id, workspaceId, status); + const row = await messageRepo.findById(msg.id, workspaceId); + expect(row!.status).toBe('aborted'); + expect((row!.metadata as Record).finalizeFailed).toBe(true); + }); + + it('clause (d): a stale streaming row with NO active run on the chat -> aborted+finalizeFailed', async () => { + const chatId = await newChat(); + const msg = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'streaming', + metadata: { parts: [] }, + }); + await db + .updateTable('aiChatMessages') + .set({ updatedAt: new Date(Date.now() - 60 * 60 * 1000) }) + .where('id', '=', msg.id) + .execute(); + + const swept = await messageRepo.sweepStreamingWithoutActiveRun( + 15 * 60 * 1000, + ); + expect(swept).toBeGreaterThanOrEqual(1); + const row = await messageRepo.findById(msg.id, workspaceId); + expect(row!.status).toBe('aborted'); + expect((row!.metadata as Record).finalizeFailed).toBe(true); + }); + + it('clause (d) is DOUBLE-GATED: a stale streaming row WITH an active run on the chat is left alone', async () => { + const chatId = await newChat(); + const msg = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'streaming', + metadata: { parts: [] }, + }); + await db + .updateTable('aiChatMessages') + .set({ updatedAt: new Date(Date.now() - 60 * 60 * 1000) }) + .where('id', '=', msg.id) + .execute(); + // An ACTIVE run on the same chat -> clause (d) must NOT touch the message. + const run = await runRepo.insert({ + chatId, + workspaceId, + createdBy: userId, + status: 'running', + }); + + await messageRepo.sweepStreamingWithoutActiveRun(15 * 60 * 1000); + expect((await messageRepo.findById(msg.id, workspaceId))!.status).toBe( + 'streaming', + ); + await runRepo.finalizeIfActive(run.id, workspaceId, { + status: 'aborted', + error: null, + }); + }); + + it('"kill DB on finish" recovery: after the DB is back, reconcile leaves NEITHER the row nor the run stuck', async () => { + // Simulate a process that seeded the assistant row + run, then died before + // finalizing EITHER (a mid-turn crash): a streaming message + a running run, + // both stale, with no in-memory entry (fresh service = fresh maps). + const chatId = await newChat(); + const msg = await createMessage(db, { + workspaceId, + chatId, + role: 'assistant', + status: 'streaming', + metadata: { parts: [{ type: 'text', text: 'partial' }] }, + }); + const run = await runRepo.insert({ + chatId, + workspaceId, + createdBy: userId, + status: 'running', + assistantMessageId: msg.id, + }); + await db + .updateTable('aiChatRuns') + .set({ updatedAt: new Date(Date.now() - 60 * 60 * 1000) }) + .where('id', '=', run.id) + .execute(); + await db + .updateTable('aiChatMessages') + .set({ updatedAt: new Date(Date.now() - 60 * 60 * 1000) }) + .where('id', '=', msg.id) + .execute(); + + // Reconcile (as the periodic job would): (c) aborts the orphan run, then + // (b) settles the message from the now-terminal run. + await runService.reconcileStaleRuns(15 * 60 * 1000); + const stuck = await messageRepo.findStreamingWithTerminalRun(); + for (const s of stuck) { + const status = s.runStatus === 'failed' ? 'error' : 'aborted'; + await messageRepo.stampTerminalIfStreaming(s.messageId, s.workspaceId, status); + } + + // Neither is stuck: the run is terminal AND the message is terminal. + expect((await runRepo.findById(run.id, workspaceId))!.status).toBe('aborted'); + const row = await messageRepo.findById(msg.id, workspaceId); + expect(row!.status).toBe('aborted'); + expect((row!.metadata as Record).finalizeFailed).toBe(true); + }); +});