feat(ai-chat): ретраи finalizeAssistant + owner-write приоритет + двусторонний reconcile (#487)

Раньше 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) <noreply@anthropic.com>
2026-07-11 05:13:35 +03:00
parent 307ad49324
commit 002c931c6b
9 changed files with 889 additions and 45 deletions
@@ -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<AiChatMessage | undefined> {
+    const db = dbOrTx(this.db, trx);
+    return db
+      .updateTable('aiChatMessages')
+      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      .where('id', '=', id)
+      .where('workspaceId', '=', workspaceId)
+      .where((eb) =>
+        eb.or([
+          eb('status', '=', 'streaming'),
+          eb(sql<string>`(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<AiChatMessage | undefined> {
+    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<number> {
+    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<number> {
    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')
@@ -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<Array<{ id: string; workspaceId: string; chatId: string }>> {
+    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,