diff --git a/apps/client/src/features/ai-chat/services/ai-chat-service.ts b/apps/client/src/features/ai-chat/services/ai-chat-service.ts index 34e6aa71..d5a6bfac 100644 --- a/apps/client/src/features/ai-chat/services/ai-chat-service.ts +++ b/apps/client/src/features/ai-chat/services/ai-chat-service.ts @@ -57,6 +57,31 @@ export async function stopRun( return req.data; } +/** + * Delta poll (#491): the chat's message rows changed since `cursor` (a DB-clock + * timestamp echoed from the previous poll) plus the current run fact, in ONE + * round-trip — the degraded-poll fallback's payload, replacing the old "refetch + * ALL infinite-query pages every 2.5s with full parts" poll. Omit `cursor` on the + * first poll (returns just a fresh cursor, no rows, to start the chain). The + * overlap window guarantees occasional REPEATS, so the caller MUST merge rows + * idempotently by id (mergeById). Owner-gated server-side. + */ +export async function getAiChatMessagesDelta( + chatId: string, + cursor?: string, +): Promise<{ + rows: IAiChatMessageRow[]; + cursor: string; + run: { id: string; status: string } | null; +}> { + const req = await api.post<{ + rows: IAiChatMessageRow[]; + cursor: string; + run: { id: string; status: string } | null; + }>("/ai-chat/messages/delta", { chatId, cursor }); + return req.data; +} + /** * #488: the run-fact — "is a run active on this chat?" — first-class from the * server (POST /ai-chat/run). Called on mount to seed the client FSM's run-fact diff --git a/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts b/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts index 8a4dc7ef..9278d004 100644 --- a/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts +++ b/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts @@ -109,4 +109,37 @@ describe("mergeById", () => { expect(mergeById(prev, null)).toBe(prev); expect(mergeById(prev, undefined)).toBe(prev); }); + + // #491 CONTRACT: the delta poll's overlap window GUARANTEES the same row is + // re-delivered across close polls, so merging must be IDEMPOTENT by id — merging + // the same row (or an equal-length list of rows) twice must not duplicate or + // reorder. This is the property the whole delta-poll design leans on; a + // regression here would re-introduce duplicate assistant bubbles on every poll. + it("is idempotent by id: re-merging the same row does not duplicate or reorder", () => { + const seed = [makeMsg("u1", "hi"), makeMsg("a1", "step 1")]; + const repeat = makeMsg("a1", "step 1"); // the SAME row the overlap re-delivers + const once = mergeById(seed, repeat); + const twice = mergeById(once, repeat); + const thrice = mergeById(twice, repeat); + // Length is stable (no growth), order is stable (user then assistant). + expect(once.map((m) => m.id)).toEqual(["u1", "a1"]); + expect(twice.map((m) => m.id)).toEqual(["u1", "a1"]); + expect(thrice.map((m) => m.id)).toEqual(["u1", "a1"]); + // The repeated merge converges: the row is replaced in place, never appended. + expect(twice[1]).toBe(repeat); + }); + + it("is idempotent across a batch of repeated + grown rows (delta re-delivery)", () => { + // A delta poll re-delivers a1 (unchanged) and a2 (grown one step). Applying the + // batch twice must equal applying it once — the poll can re-send either. + const start = [makeMsg("u1", "hi"), makeMsg("a1", "done")]; + const batch = [makeMsg("a1", "done"), makeMsg("a2", "grown step 2")]; + const apply = (list: typeof start) => + batch.reduce((acc, row) => mergeById(acc, row), list); + const once = apply(start); + const twice = apply(once); + expect(once.map((m) => m.id)).toEqual(["u1", "a1", "a2"]); + expect(twice.map((m) => m.id)).toEqual(["u1", "a1", "a2"]); + expect(twice).toEqual(once); + }); }); diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.delta.spec.ts b/apps/server/src/core/ai-chat/ai-chat.controller.delta.spec.ts new file mode 100644 index 00000000..2d418766 --- /dev/null +++ b/apps/server/src/core/ai-chat/ai-chat.controller.delta.spec.ts @@ -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', + ); + }); +}); diff --git a/apps/server/src/core/ai-chat/ai-chat.controller.ts b/apps/server/src/core/ai-chat/ai-chat.controller.ts index f9b58b98..49267968 100644 --- a/apps/server/src/core/ai-chat/ai-chat.controller.ts +++ b/apps/server/src/core/ai-chat/ai-chat.controller.ts @@ -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 diff --git a/apps/server/src/core/ai-chat/dto/ai-chat.dto.ts b/apps/server/src/core/ai-chat/dto/ai-chat.dto.ts index 1fd3c8ff..ec09c6aa 100644 --- a/apps/server/src/core/ai-chat/dto/ai-chat.dto.ts +++ b/apps/server/src/core/ai-chat/dto/ai-chat.dto.ts @@ -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() diff --git a/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.spec.ts b/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.spec.ts new file mode 100644 index 00000000..e3086de8 --- /dev/null +++ b/apps/server/src/database/repos/ai-chat/ai-chat-message.repo.delta.spec.ts @@ -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; +let sqlClient: ReturnType; +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({ + 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 = {}) { + return msgRepo.insert({ + chatId, + workspaceId, + userId: null as never, + role: 'assistant', + content: 'x', + status: 'streaming', + ...overrides, + } as never); +} + +async function dbNow(): Promise { + const r = await sql<{ now: Date }>`select now() as now`.execute(db); + return r.rows[0].now.toISOString(); +} + +const maybe = (name: string, fn: () => Promise) => + 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); + }, + ); +}); 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 96070e77..57e82baa 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 @@ -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`${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), 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), 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), 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), 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) diff --git a/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.spec.ts b/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.spec.ts index d03a43e3..5d6faa9d 100644 --- a/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.spec.ts +++ b/apps/server/src/database/repos/ai-chat/ai-chat-run.repo.spec.ts @@ -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 () => { 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 c27480bf..5758867e 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 @@ -136,7 +136,11 @@ export class AiChatRunRepo { const db = dbOrTx(this.db, trx); return db .updateTable('aiChatRuns') - .set({ ...(patch as Record), 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), updatedAt: sql`now()` }) .where('id', '=', id) .where('workspaceId', '=', workspaceId) .returning(this.baseFields) @@ -162,14 +166,15 @@ export class AiChatRunRepo { 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, + // #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 { 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();