perf(ai-chat): дельта-эндпоинт поллинга + run-факт в ответе (#491)

Degraded-poll рефетчил ВСЕ страницы infinite-query каждые 2.5 c с полными parts. Вводим дельта-эндпоинт «строки, изменённые после курсора»: - POST /ai-chat/messages/delta → { rows, cursor, run: {id,status}|null }. Курсор — таймстамп часов БД (now()), клиент эхом возвращает его каждый поллинг. Окно перекрытия now()−5s ловит строку, закоммиченную с updatedAt чуть раньше момента снятия предыдущего курсора на другом autocommit- соединении (одиночные UPDATE, длинных транзакций нет). Окно ГАРАНТИРУЕТ повторы → контракт: клиентский merge идемпотентен по id (mergeById). - Run-факт едет В дельте (отдельный /run-поллинг удвоил бы QPS — отвергнуто). - Все дельта-релевантные записи (message update/finalizeOwner/reconcile/sweep, run update/finalizeIfActive/markStopRequested/sweepRunning) стампуют updatedAt через SQL now(), а не app-clock new Date(): единая монотонная ось курсора, смешанные источники часов были независимым источником пропусков. - Клиент: сервис-функция getAiChatMessagesDelta + фиксация контракта идемпотентности mergeById тестом (свап degraded-поллинга на дельту — в коммите 3, вместе с re-seed-путём attach). Тесты (observable-property на живом Postgres, не моки): - дельта-семантика + монотонность курсора; - RACE «коммит позже с updatedAt раньше курсора» — ловится окном перекрытия (наивный updatedAt > cursor пропустил бы); - окно гарантирует повторы → идемпотентный merge; - updatedAt стампуется часами БД, а не app-clock (фейк системных часов в 2099 — стамп остаётся на времени БД); - контроллер: owner-gate, форма ответа, run-факт только {id,status}. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-11 17:55:06 +03:00
parent 6dad309b51
commit 798a81abfe
9 changed files with 570 additions and 18 deletions
@@ -0,0 +1,108 @@
+import { ForbiddenException } from '@nestjs/common';
+import { AiChatController } from './ai-chat.controller';
+import type { User, Workspace } from '@docmost/db/types/entity.types';
+
+/**
+ * Wiring spec for the #491 delta-poll endpoint (`POST /ai-chat/messages/delta`).
+ * Owner-gated via assertOwnedChat (same gate as the other reads), NOT flag-gated.
+ * The run fact rides IN the delta response (no separate /run poll). Hand-rolled
+ * mocks — no Nest graph, no DB. Constructor order: (aiChatService,
+ * aiChatRunService, aiChatRepo, aiChatMessageRepo, aiTranscription, pageRepo).
+ */
+describe('AiChatController POST /ai-chat/messages/delta (#491)', () => {
+  const user = { id: 'u1' } as User;
+  const workspace = { id: 'ws1' } as Workspace;
+
+  function makeController(opts: {
+    chat?: unknown;
+    delta?: { rows: unknown[]; cursor: string };
+    run?: unknown;
+  }) {
+    const aiChatRunService = {
+      getLatestForChat: jest.fn().mockResolvedValue(opts.run),
+    };
+    const aiChatRepo = {
+      findById: jest.fn().mockResolvedValue(opts.chat),
+    };
+    const aiChatMessageRepo = {
+      findByChatUpdatedAfter: jest
+        .fn()
+        .mockResolvedValue(opts.delta ?? { rows: [], cursor: 'C1' }),
+    };
+    const controller = new AiChatController(
+      {} as never,
+      aiChatRunService as never,
+      aiChatRepo as never,
+      aiChatMessageRepo as never,
+      {} as never,
+      {} as never,
+    );
+    return { controller, aiChatRunService, aiChatRepo, aiChatMessageRepo };
+  }
+
+  it('owner-gates: a chat the user does not own throws, never reaching the repo', async () => {
+    const { controller, aiChatMessageRepo, aiChatRunService } = makeController({
+      chat: { id: 'c1', creatorId: 'someone-else' },
+    });
+    await expect(
+      controller.getMessagesDelta({ chatId: 'c1' }, user, workspace),
+    ).rejects.toBeInstanceOf(ForbiddenException);
+    expect(aiChatMessageRepo.findByChatUpdatedAfter).not.toHaveBeenCalled();
+    expect(aiChatRunService.getLatestForChat).not.toHaveBeenCalled();
+  });
+
+  it('returns { rows, cursor, run:{id,status} } with the run fact inlined', async () => {
+    const rows = [{ id: 'm1' }];
+    const { controller } = makeController({
+      chat: { id: 'c1', creatorId: 'u1' },
+      delta: { rows, cursor: 'C2' },
+      run: { id: 'r1', status: 'running', error: 'ignored', stepCount: 3 },
+    });
+    const res = await controller.getMessagesDelta(
+      { chatId: 'c1', cursor: 'C1' },
+      user,
+      workspace,
+    );
+    expect(res).toEqual({
+      rows,
+      cursor: 'C2',
+      // ONLY id + status — never the whole run row.
+      run: { id: 'r1', status: 'running' },
+    });
+  });
+
+  it('run is null when the chat has never had a run', async () => {
+    const { controller } = makeController({
+      chat: { id: 'c1', creatorId: 'u1' },
+      run: undefined,
+    });
+    const res = await controller.getMessagesDelta(
+      { chatId: 'c1' },
+      user,
+      workspace,
+    );
+    expect(res.run).toBeNull();
+  });
+
+  it('passes cursor through, defaulting a missing cursor to null (first poll)', async () => {
+    const { controller, aiChatMessageRepo } = makeController({
+      chat: { id: 'c1', creatorId: 'u1' },
+    });
+    await controller.getMessagesDelta({ chatId: 'c1' }, user, workspace);
+    expect(aiChatMessageRepo.findByChatUpdatedAfter).toHaveBeenCalledWith(
+      'c1',
+      'ws1',
+      null,
+    );
+    await controller.getMessagesDelta(
+      { chatId: 'c1', cursor: 'CX' },
+      user,
+      workspace,
+    );
+    expect(aiChatMessageRepo.findByChatUpdatedAfter).toHaveBeenLastCalledWith(
+      'c1',
+      'ws1',
+      'CX',
+    );
+  });
+});
@@ -51,6 +51,7 @@ import {
  ChatIdDto,
  ExportChatDto,
  GeneratePageTitleDto,
+  GetChatDeltaDto,
  GetChatMessagesDto,
  GetRunDto,
  RenameChatDto,
@@ -149,6 +150,46 @@ export class AiChatController {
    );
  }

+  /**
+   * Delta poll (#491) — the degraded-poll fallback's payload. Returns the chat's
+   * message rows changed since `cursor` (a DB-clock timestamp from the previous
+   * poll), a FRESH cursor, AND the current run fact `{ id, status } | null`. This
+   * replaces the old degraded poll that refetched ALL infinite-query pages (full
+   * parts) every 2.5s: the client seeds once and thereafter merges only the
+   * deltas by id (the overlap window guarantees repeats — the merge is idempotent,
+   * see mergeById). The run fact rides IN the delta (a separate /run poll would
+   * double the poll QPS), so the client FSM gets the run's status on the same tick.
+   * Owner-gated via assertOwnedChat (same gate as the other read endpoints).
+   */
+  @HttpCode(HttpStatus.OK)
+  @Post('messages/delta')
+  async getMessagesDelta(
+    @Body() dto: GetChatDeltaDto,
+    @AuthUser() user: User,
+    @AuthWorkspace() workspace: Workspace,
+  ): Promise<{
+    rows: AiChatMessage[];
+    cursor: string;
+    run: { id: string; status: string } | null;
+  }> {
+    await this.assertOwnedChat(dto.chatId, user, workspace);
+    const { rows, cursor } =
+      await this.aiChatMessageRepo.findByChatUpdatedAfter(
+        dto.chatId,
+        workspace.id,
+        dto.cursor ?? null,
+      );
+    const run = await this.aiChatRunService.getLatestForChat(
+      dto.chatId,
+      workspace.id,
+    );
+    return {
+      rows,
+      cursor,
+      run: run ? { id: run.id, status: run.status } : null,
+    };
+  }
+
  /**
   * Export a chat to Markdown (#183). The DB is the single source of truth: the
   * whole transcript is loaded (oldest -> newest) and rendered server-side. Now
@@ -37,6 +37,21 @@ export class GetChatMessagesDto {
  cursor?: string;
 }

+/**
+ * Delta poll (#491): pull the chat's rows changed since `cursor` (a DB-clock
+ * timestamp from the previous poll) plus the current run fact — the degraded-poll
+ * fallback's payload, replacing the full infinite-query refetch. Omit `cursor` on
+ * the first poll (returns just a fresh cursor to start the chain).
+ */
+export class GetChatDeltaDto {
+  @IsString()
+  chatId: string;
+
+  @IsOptional()
+  @IsString()
+  cursor?: string;
+}
+
 /** Resolve the chat bound to a document (the page's most-recent owned chat). */
 export class BoundChatDto {
  @IsString()
@@ -0,0 +1,223 @@
+import { randomUUID } from 'crypto';
+import { CamelCasePlugin, Kysely, sql } from 'kysely';
+import { PostgresJSDialect } from 'kysely-postgres-js';
+import postgres from 'postgres';
+import { AiChatMessageRepo } from './ai-chat-message.repo';
+import { AiChatRunRepo } from './ai-chat-run.repo';
+
+/**
+ * #491 delta-poll — OBSERVABLE-PROPERTY tests against a LIVE Postgres (the local
+ * gitmost test DB, docker `gitmost-test-pg` on :5432), not "rows through a mock"
+ * (a mock cannot observe the DB clock nor the overlap-window race — the very
+ * things that matter here). Drives the REAL repo methods (`findByChatUpdatedAfter`,
+ * the now()-stamped `update`) and asserts:
+ *   1. delta-relevant writes stamp `updatedAt` from the DB clock, not the app clock
+ *      (proven by faking the process clock far into the future and observing the
+ *      stamp stays on real DB time);
+ *   2. the poll returns only rows changed after the cursor, ordered, with a fresh
+ *      DB-clock cursor;
+ *   3. the "committed late but stamped earlier than the cursor" RACE is caught by
+ *      the overlap window (a naive `updatedAt > cursor` would MISS it);
+ *   4. the overlap GUARANTEES repeats across close polls — the contract behind the
+ *      client's idempotent merge (mergeById).
+ *
+ * FK triggers are bypassed (`session_replication_role = replica`) so synthetic
+ * chat/workspace ids need no parent fixtures; a single pooled connection (max 1)
+ * keeps that session setting for every query. SKIPS cleanly when the DB is
+ * unreachable so a DB-less CI never breaks.
+ */
+const CONN =
+  process.env.WAL_TEST_DATABASE_URL ??
+  'postgresql://docmost:docmost_dev_pw@localhost:5432/docmost';
+
+let db: Kysely<any>;
+let sqlClient: ReturnType<typeof postgres>;
+let msgRepo: AiChatMessageRepo;
+let runRepo: AiChatRunRepo;
+let reachable = false;
+
+const workspaceId = randomUUID();
+const chatId = randomUUID();
+
+beforeAll(async () => {
+  try {
+    sqlClient = postgres(CONN, { max: 1, onnotice: () => {} });
+    db = new Kysely<any>({
+      dialect: new PostgresJSDialect({ postgres: sqlClient }),
+      plugins: [new CamelCasePlugin()],
+    });
+    // Single connection keeps this session-scoped bypass for the whole suite.
+    await sql`set session_replication_role = replica`.execute(db);
+    await sql`select 1`.execute(db);
+    reachable = true;
+  } catch {
+    reachable = false;
+  }
+  msgRepo = new AiChatMessageRepo(db as never);
+  runRepo = new AiChatRunRepo(db as never);
+});
+
+afterAll(async () => {
+  if (db) {
+    try {
+      await db
+        .deleteFrom('aiChatMessages')
+        .where('workspaceId', '=', workspaceId)
+        .execute();
+      await db
+        .deleteFrom('aiChatRuns')
+        .where('workspaceId', '=', workspaceId)
+        .execute();
+    } catch {
+      /* best-effort cleanup */
+    }
+    await db.destroy();
+  }
+});
+
+afterEach(() => {
+  jest.useRealTimers();
+});
+
+async function seedMessage(overrides: Record<string, unknown> = {}) {
+  return msgRepo.insert({
+    chatId,
+    workspaceId,
+    userId: null as never,
+    role: 'assistant',
+    content: 'x',
+    status: 'streaming',
+    ...overrides,
+  } as never);
+}
+
+async function dbNow(): Promise<string> {
+  const r = await sql<{ now: Date }>`select now() as now`.execute(db);
+  return r.rows[0].now.toISOString();
+}
+
+const maybe = (name: string, fn: () => Promise<void>) =>
+  it(name, async () => {
+    if (!reachable) {
+      console.warn(`SKIP (${name}): test DB unreachable at ${CONN}`);
+      return;
+    }
+    await fn();
+  });
+
+describe('AiChatMessageRepo.findByChatUpdatedAfter (#491 delta poll)', () => {
+  maybe('null cursor returns no rows and a fresh DB-clock cursor', async () => {
+    const before = await dbNow();
+    const { rows, cursor } = await msgRepo.findByChatUpdatedAfter(
+      chatId,
+      workspaceId,
+      null,
+    );
+    expect(rows).toEqual([]);
+    expect(new Date(cursor).getTime()).toBeGreaterThanOrEqual(
+      new Date(before).getTime(),
+    );
+  });
+
+  maybe('returns only rows changed after the cursor', async () => {
+    const { cursor: c0 } = await msgRepo.findByChatUpdatedAfter(
+      chatId,
+      workspaceId,
+      null,
+    );
+    const m = await seedMessage();
+    const { rows, cursor: c1 } = await msgRepo.findByChatUpdatedAfter(
+      chatId,
+      workspaceId,
+      c0,
+    );
+    expect(rows.map((r) => r.id)).toContain(m.id);
+    // Cursor is monotonic (advances).
+    expect(new Date(c1).getTime()).toBeGreaterThanOrEqual(
+      new Date(c0).getTime(),
+    );
+  });
+
+  maybe(
+    'RACE: a row stamped BEFORE the cursor but seen after is caught by the overlap',
+    async () => {
+      // Cursor taken now; then a row appears whose updatedAt is 2s in the PAST
+      // (committed late on another connection but stamped earlier). A naive
+      // `updatedAt > cursor` would MISS it; the 5s overlap window catches it.
+      const cursor = await dbNow();
+      const m = await seedMessage();
+      await sql`update ai_chat_messages set updated_at = now() - interval '2 seconds' where id = ${m.id}`.execute(
+        db,
+      );
+      const { rows } = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        cursor,
+      );
+      expect(rows.map((r) => r.id)).toContain(m.id);
+    },
+  );
+
+  maybe(
+    'overlap GUARANTEES repeats across close polls (idempotent-merge contract)',
+    async () => {
+      const { cursor: c0 } = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        null,
+      );
+      const m = await seedMessage();
+      const first = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        c0,
+      );
+      expect(first.rows.map((r) => r.id)).toContain(m.id);
+      // Immediately re-poll with the JUST-returned cursor: the row is still within
+      // the overlap window, so it is returned AGAIN — the client MUST dedupe by id.
+      const second = await msgRepo.findByChatUpdatedAfter(
+        chatId,
+        workspaceId,
+        first.cursor,
+      );
+      expect(second.rows.map((r) => r.id)).toContain(m.id);
+    },
+  );
+
+  maybe(
+    'update() stamps updatedAt from the DB clock, not the app clock',
+    async () => {
+      const m = await seedMessage();
+      // Fake the PROCESS clock ~73 years into the future. If the stamp came from
+      // `new Date()` the row would read year 2099; sql now() keeps it on DB time.
+      jest.useFakeTimers().setSystemTime(new Date('2099-01-01T00:00:00Z'));
+      const updated = await msgRepo.update(m.id, workspaceId, {
+        content: 'y',
+      });
+      jest.useRealTimers();
+      expect(updated).toBeDefined();
+      expect(new Date(updated!.updatedAt).getFullYear()).toBeLessThan(2099);
+    },
+  );
+
+  maybe(
+    'run update() also stamps updatedAt from the DB clock',
+    async () => {
+      const run = await runRepo.insert({
+        chatId,
+        workspaceId,
+        createdBy: null as never,
+        trigger: 'user',
+        status: 'running',
+        stepCount: 0,
+      } as never);
+      jest.useFakeTimers().setSystemTime(new Date('2099-01-01T00:00:00Z'));
+      const updated = await runRepo.update(run.id, workspaceId, {
+        stepCount: 1,
+      });
+      jest.useRealTimers();
+      expect(updated).toBeDefined();
+      expect(new Date(updated!.updatedAt).getFullYear()).toBeLessThan(2099);
+    },
+  );
+});
@@ -25,6 +25,20 @@ const SWEEP_STREAMING_STALE_MS = 10 * 60 * 1000; // 10 minutes
 // into memory; far above any realistic transcript length.
 const FIND_ALL_BY_CHAT_LIMIT = 5000;

+// Delta-poll overlap (#491): the poll query reaches this far BEHIND the client's
+// echoed cursor, so a row that committed with an `updatedAt` marginally before the
+// previous cursor was taken (on another autocommit connection) is still caught.
+// Sized well above realistic single-row commit skew; the client merge is
+// idempotent by id (mergeById), so the guaranteed repeats the overlap produces are
+// harmless.
+export const DELTA_POLL_OVERLAP_SECONDS = 5;
+
+// Hard cap on rows one delta poll returns — a safety bound (a poll should carry a
+// handful of just-changed rows, never a whole transcript). Ordered by (updatedAt,
+// id) asc, so on the pathological overflow the OLDEST changes win and the newest
+// are picked up by the next poll (its cursor did not advance past them).
+export const DELTA_POLL_MAX_ROWS = 500;
+
@Injectable()
 export class AiChatMessageRepo {
  private readonly logger = new Logger(AiChatMessageRepo.name);
@@ -139,6 +153,72 @@ export class AiChatMessageRepo {
      .executeTakeFirst();
  }

+  /**
+   * Delta read (#491) for the degraded poll: the chat's messages whose row
+   * changed AFTER the client's `cursor`, plus a FRESH cursor taken from the DB
+   * clock. Replaces the old "refetch ALL infinite-query pages every 2.5s with
+   * full parts" poll — the client seeds once (findByChat) and thereafter pulls
+   * only the deltas and merges them by id (mergeById).
+   *
+   * Cursor: a DB-clock timestamp (now()) the client echoes back each poll. All
+   * delta-relevant writes stamp `updatedAt` with now() (see `update` /
+   * `finalizeOwner`), so this is a SINGLE monotonic axis. The query overlaps the
+   * cursor by DELTA_POLL_OVERLAP_SECONDS to catch a row committed with an
+   * `updatedAt` marginally BEFORE the previous cursor was taken on another
+   * connection (single-row autocommit UPDATEs; no long transactions). The overlap
+   * GUARANTEES occasional REPEATS, so the client merge MUST be idempotent by id.
+   *
+   * `cursor === null` (first poll after the full seed) returns NO rows — there is
+   * nothing "new" relative to a just-loaded seed — only the fresh cursor to start
+   * the delta chain. The fresh cursor is read AFTER the rows, so it is >= every
+   * returned row's `updatedAt` (they were read strictly earlier) — a row that
+   * commits between the rows-read and the cursor-read is at most
+   * DELTA_POLL_OVERLAP_SECONDS behind the returned cursor, so the next poll's
+   * overlap window always re-includes it (no miss).
+   */
+  async findByChatUpdatedAfter(
+    chatId: string,
+    workspaceId: string,
+    cursor: string | null,
+  ): Promise<{ rows: AiChatMessage[]; cursor: string }> {
+    if (cursor === null) {
+      const nowRow = await sql<{ now: Date }>`select now() as now`.execute(
+        this.db,
+      );
+      return { rows: [], cursor: nowRow.rows[0].now.toISOString() };
+    }
+    // Overlap the client cursor by DELTA_POLL_OVERLAP_SECONDS, computed in SQL off
+    // the echoed cursor so the whole comparison stays on the DB clock.
+    const rows = await this.db
+      .selectFrom('aiChatMessages')
+      .select(this.baseFields)
+      .where('chatId', '=', chatId)
+      .where('workspaceId', '=', workspaceId)
+      .where('deletedAt', 'is', null)
+      .where(
+        'updatedAt',
+        '>',
+        sql<Date>`${cursor}::timestamptz - make_interval(secs => ${DELTA_POLL_OVERLAP_SECONDS})`,
+      )
+      .orderBy('updatedAt', 'asc')
+      .orderBy('id', 'asc')
+      .limit(DELTA_POLL_MAX_ROWS)
+      .execute();
+    // When the page filled (pathological overflow), DO NOT advance the cursor to
+    // now(): that would skip the changed rows past the cap that this poll did not
+    // return. Resume from the last returned row's updatedAt instead (the next
+    // poll's overlap re-includes ties by id). In the normal case the fresh DB-clock
+    // now() is the cursor.
+    if (rows.length === DELTA_POLL_MAX_ROWS) {
+      return {
+        rows,
+        cursor: rows[rows.length - 1].updatedAt.toISOString(),
+      };
+    }
+    const nowRow = await sql<{ now: Date }>`select now() as now`.execute(this.db);
+    return { rows, cursor: nowRow.rows[0].now.toISOString() };
+  }
+
  async insert(
    insertable: InsertableAiChatMessage,
    trx?: KyselyTransaction,
@@ -172,7 +252,13 @@ export class AiChatMessageRepo {
    const db = dbOrTx(this.db, opts?.trx);
    let query = db
      .updateTable('aiChatMessages')
-      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      // #491: stamp `updatedAt` from the DB clock (sql now()), NOT the app clock
+      // (new Date()). The delta-poll cursor (findByChatUpdatedAfter) is a single
+      // DB-clock axis; a per-step 'streaming' UPDATE stamped with the app clock
+      // would be a SECOND, skewed clock source and could leave a row's updatedAt
+      // just under a cursor taken from now() on another connection — an
+      // independent source of delta MISSES. All delta-relevant writes use now().
+      .set({ ...(patch as Record<string, unknown>), updatedAt: sql`now()` })
      .where('id', '=', id)
      .where('workspaceId', '=', workspaceId);
    // Concurrency guard (#183 review): a per-step 'streaming' update must NEVER
@@ -214,7 +300,9 @@ export class AiChatMessageRepo {
    const db = dbOrTx(this.db, trx);
    return db
      .updateTable('aiChatMessages')
-      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      // #491: DB-clock stamp (see `update`) — this terminal write flips the row's
+      // status, which the delta poll must observe on the shared now() cursor axis.
+      .set({ ...(patch as Record<string, unknown>), updatedAt: sql`now()` })
      .where('id', '=', id)
      .where('workspaceId', '=', workspaceId)
      .where((eb) =>
@@ -249,7 +337,9 @@ export class AiChatMessageRepo {
      .set({
        status,
        metadata: sql`coalesce(metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`,
-        updatedAt: new Date(),
+        // #491: DB-clock stamp (see `update`) so a reconcile status flip lands on
+        // the same now() cursor axis the delta poll reads.
+        updatedAt: sql`now()`,
      })
      .where('id', '=', id)
      .where('workspaceId', '=', workspaceId)
@@ -307,7 +397,9 @@ export class AiChatMessageRepo {
      .set({
        status: 'aborted',
        metadata: sql`coalesce(m.metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`,
-        updatedAt: new Date(),
+        // #491: DB-clock stamp (see `update`). The staleness WHERE below stays on
+        // the app clock — a >minutes window makes the ms-scale skew irrelevant.
+        updatedAt: sql`now()`,
      })
      .where('m.status', '=', 'streaming')
      .where('m.updatedAt', '<', staleBefore)
@@ -351,7 +443,8 @@ export class AiChatMessageRepo {
      .set({
        status: 'aborted',
        metadata: sql`coalesce(metadata, '{}'::jsonb) || jsonb_build_object('finalizeFailed', true)`,
-        updatedAt: new Date(),
+        // #491: DB-clock stamp (see `update`). Staleness WHERE stays app-clock.
+        updatedAt: sql`now()`,
      })
      .where('status', '=', 'streaming')
      .where('updatedAt', '<', staleBefore)
@@ -62,10 +62,17 @@ describe('AiChatRunRepo.sweepRunning', () => {
    // ...but a fresh 'running' run (updatedAt = now) must NOT be skipped: no
    // updatedAt predicate at all on the boot path.
    expect(rec.wheres.some(([col]) => col === 'updatedAt')).toBe(false);
-    // It flips to 'aborted' and stamps finishedAt.
-    expect(rec.set).toEqual(
-      expect.objectContaining({ status: 'aborted', finishedAt: expect.any(Date) }),
-    );
+    // It flips to 'aborted' and stamps finishedAt + updatedAt. #491: the stamps
+    // are now DB-clock `sql now()` expressions (raw builders), NOT app-clock
+    // `new Date()`, so the run row shares the delta poll's single now() cursor axis
+    // — assert they are present and are the sql raw-builder objects (not a Date,
+    // not undefined).
+    expect(rec.set?.status).toBe('aborted');
+    for (const stamp of ['finishedAt', 'updatedAt'] as const) {
+      expect(rec.set?.[stamp]).toBeDefined();
+      expect(rec.set?.[stamp]).not.toBeInstanceOf(Date);
+      expect(typeof rec.set?.[stamp]).toBe('object');
+    }
  });

  it('phase-2 path: an explicit staleMs reintroduces the updatedAt window', async () => {
@@ -136,7 +136,11 @@ export class AiChatRunRepo {
    const db = dbOrTx(this.db, trx);
    return db
      .updateTable('aiChatRuns')
-      .set({ ...(patch as Record<string, unknown>), updatedAt: new Date() })
+      // #491: DB-clock stamp (sql now()) so the run row shares the delta poll's
+      // single now() cursor axis with the assistant message rows — a run-status
+      // change (the run fact the delta carries) must never sit on a skewed app
+      // clock relative to the message updatedAt cursor.
+      .set({ ...(patch as Record<string, unknown>), updatedAt: sql`now()` })
      .where('id', '=', id)
      .where('workspaceId', '=', workspaceId)
      .returning(this.baseFields)
@@ -162,14 +166,15 @@ export class AiChatRunRepo {
    trx?: KyselyTransaction,
  ): Promise<AiChatRun | undefined> {
    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,
+        // #491: DB-clock stamps (finished_at + updated_at) so the terminal run
+        // fact lands on the delta poll's now() cursor axis.
+        finishedAt: sql`now()`,
+        updatedAt: sql`now()`,
      })
      .where('id', '=', id)
      .where('workspaceId', '=', workspaceId)
@@ -192,7 +197,8 @@ export class AiChatRunRepo {
    const db = dbOrTx(this.db, trx);
    return db
      .updateTable('aiChatRuns')
-      .set({ stopRequestedAt: new Date(), updatedAt: new Date() })
+      // #491: DB-clock stamps (see `update`).
+      .set({ stopRequestedAt: sql`now()`, updatedAt: sql`now()` })
      .where('id', '=', id)
      .where('workspaceId', '=', workspaceId)
      .where('status', 'in', ACTIVE_RUN_STATUSES as unknown as string[])
@@ -249,13 +255,14 @@ export class AiChatRunRepo {
    trx?: KyselyTransaction,
  ): Promise<number> {
    const db = dbOrTx(this.db, trx);
-    const now = new Date();
    let query = db
      .updateTable('aiChatRuns')
      .set({
        status: 'aborted',
-        finishedAt: now,
-        updatedAt: now,
+        // #491: DB-clock stamps (see `update`). The staleness WHERE below stays on
+        // the app clock — a >minutes window makes the ms-scale skew irrelevant.
+        finishedAt: sql`now()`,
+        updatedAt: sql`now()`,
        error: sql`coalesce(error, ${'Run interrupted by a server restart.'})`,
      })
      .where('status', 'in', ACTIVE_RUN_STATUSES as unknown as string[]);
@@ -263,7 +270,7 @@ export class AiChatRunRepo {
    // sibling replica's live run is never aborted. Omitted on the phase-1 boot
    // sweep -> unconditional.
    if (typeof opts.staleMs === 'number') {
-      const staleBefore = new Date(now.getTime() - opts.staleMs);
+      const staleBefore = new Date(Date.now() - opts.staleMs);
      query = query.where('updatedAt', '<', staleBefore);
    }
    const rows = await query.returning('id').execute();