From 83686aa39b608c4d20d9db7ee342e48e10cdbb00 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:18:51 +0000 Subject: [PATCH 1/8] perf(conversations): Avoid full history reload after append commits Make append return the inserted delta and live history cursor so commitMessages can advance committedSeq/messageSeqs without a second loadCurrentHistory of the entire active version. Co-Authored-By: David Cramer --- .../junior/src/chat/conversations/README.md | 2 + .../junior/src/chat/conversations/history.ts | 18 +- .../src/chat/conversations/projection.ts | 71 ++++---- .../src/chat/conversations/sql/history.ts | 163 +++++++++++------- .../conversation-storage-sql.test.ts | 33 +++- .../unit/conversations/turn-lifecycle.test.ts | 15 +- 6 files changed, 191 insertions(+), 111 deletions(-) diff --git a/packages/junior/src/chat/conversations/README.md b/packages/junior/src/chat/conversations/README.md index 8b664b1757..99bd4acf9e 100644 --- a/packages/junior/src/chat/conversations/README.md +++ b/packages/junior/src/chat/conversations/README.md @@ -59,6 +59,8 @@ older source-thread context; it does not replace Pi history. - Persist inbound `message` events before agent execution. - Persist assistant `message` events only after destination acceptance. - Append stable native agent-history events in sequence order. +- `append` returns the inserted delta and active history cursor so commit + paths can advance without reloading the full current history. - Reject attempts to mutate an already committed agent-history prefix. - Replace agent history only through explicit compaction or handoff. - Restore transcripts and agent history directly from conversation events. diff --git a/packages/junior/src/chat/conversations/history.ts b/packages/junior/src/chat/conversations/history.ts index 9af8a247de..fc32542ebc 100644 --- a/packages/junior/src/chat/conversations/history.ts +++ b/packages/junior/src/chat/conversations/history.ts @@ -503,6 +503,22 @@ export const newConversationEventSchema = z /** An event to append; the store assigns `seq` and current history version. */ export type NewConversationEvent = z.output; +/** + * Result of an append: newly inserted events plus the active history version. + * + * Callers that need a post-write cursor use this instead of reloading the full + * current history. Idempotent no-ops still report the live history version and + * an empty `inserted` list. + */ +export interface ConversationEventAppendResult { + /** Active model-history version after the append. */ + historyVersion: number; + /** Events that were newly inserted, in `seq` order. */ + inserted: ConversationEvent[]; + /** Next free `seq` after the append. */ + nextSeq: number; +} + /** Bounded observational page over the durable conversation event log. */ export interface ConversationEventQuery { /** Exclusive lower bound on `seq`. */ @@ -531,7 +547,7 @@ export interface ConversationEventStore { conversationId: string, events: NewConversationEvent[], options?: { activity?: "preserve" }, - ): Promise; + ): Promise; /** Replace active model history with a compaction or handoff event. */ replaceHistory( conversationId: string, diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index e0c15190e7..3d68a45558 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -334,47 +334,46 @@ async function commitMessagesLocked( ? { newMessageProvenance: args.newMessageProvenance } : {}), }); - if (matchingPrefix === current.messages.length) { - const newMessages = nextLocalMessages.slice(matchingPrefix); - const turnContext = args.turnContext; - const turnContextEvents = - turnContext?.contexts.map((context, index) => ({ - idempotencyKey: - `turn:${turnContext.turnId}:context:` + - `${context.pluginName}:${index}`, - createdAtMs: context.loadedAtMs, - data: { - type: "turn_context" as const, - turnId: turnContext.turnId, - pluginName: context.pluginName, - kind: context.kind, - version: context.version, - content: context.content, - }, - })) ?? []; - await eventStore.append(args.conversationId, [ - ...newMessages.map((message, index) => ({ - data: historyItemFromPiMessage( - message, - nextLocalProvenance[matchingPrefix + index]!, - ), - createdAtMs: messageTimestamp(message), - })), - ...turnContextEvents, - ]); - } else { + if (matchingPrefix !== current.messages.length) { throw new Error( `Agent history for ${args.conversationId} changed before its committed boundary`, ); } - const committedEvents = await eventStore.loadCurrentHistory( - args.conversationId, - ); - const committed = projectConversationEvents(committedEvents); + const newMessages = nextLocalMessages.slice(matchingPrefix); + const turnContext = args.turnContext; + const turnContextEvents = + turnContext?.contexts.map((context, index) => ({ + idempotencyKey: + `turn:${turnContext.turnId}:context:` + + `${context.pluginName}:${index}`, + createdAtMs: context.loadedAtMs, + data: { + type: "turn_context" as const, + turnId: turnContext.turnId, + pluginName: context.pluginName, + kind: context.kind, + version: context.version, + content: context.content, + }, + })) ?? []; + // Append returns only the inserted delta. Build the committed cursor from the + // pre-append projection plus those rows instead of reloading full history. + const appendResult = await eventStore.append(args.conversationId, [ + ...newMessages.map((message, index) => ({ + data: historyItemFromPiMessage( + message, + nextLocalProvenance[matchingPrefix + index]!, + ), + createdAtMs: messageTimestamp(message), + })), + ...turnContextEvents, + ]); + const appended = projectConversationEvents(appendResult.inserted); + const lastInserted = appendResult.inserted.at(-1); return { - committedSeq: committedEvents.at(-1)?.seq ?? -1, - historyVersion: committedEvents.at(-1)?.historyVersion ?? 0, - messageSeqs: committed.seqs, + committedSeq: lastInserted?.seq ?? appendResult.nextSeq - 1, + historyVersion: lastInserted?.historyVersion ?? appendResult.historyVersion, + messageSeqs: [...current.seqs, ...appended.seqs], messages: nextLocalMessages, provenance: nextLocalProvenance, }; diff --git a/packages/junior/src/chat/conversations/sql/history.ts b/packages/junior/src/chat/conversations/sql/history.ts index 1850aa96be..4647427829 100644 --- a/packages/junior/src/chat/conversations/sql/history.ts +++ b/packages/junior/src/chat/conversations/sql/history.ts @@ -18,6 +18,7 @@ import { historyReplacementSchema, newConversationEventSchema, type ConversationEvent, + type ConversationEventAppendResult, type ConversationEventPage, type ConversationEventQuery, type ConversationEventStore, @@ -99,79 +100,94 @@ class SqlConversationEventStore implements ConversationEventStore { conversationId: string, events: NewConversationEvent[], options: { activity?: "preserve" } = {}, - ): Promise { + ): Promise { const parsed = events.map((event) => newConversationEventSchema.parse(event), ); - if (parsed.length === 0) { - return; - } - await withConversationEventLock(this.executor, conversationId, async () => { - const existingKeys = parsed - .map((event) => event.idempotencyKey) - .filter((key): key is string => key !== undefined); - const persistedKeys = - existingKeys.length === 0 - ? new Set() - : new Set( - ( - await this.executor - .db() - .select({ key: juniorConversationEvents.idempotencyKey }) - .from(juniorConversationEvents) - .where( - and( - eq( - juniorConversationEvents.conversationId, - conversationId, - ), - inArray( - juniorConversationEvents.idempotencyKey, - existingKeys, + return await withConversationEventLock( + this.executor, + conversationId, + async () => { + if (parsed.length === 0) { + return this.appendResultFromCursor(conversationId, []); + } + const existingKeys = parsed + .map((event) => event.idempotencyKey) + .filter((key): key is string => key !== undefined); + const persistedKeys = + existingKeys.length === 0 + ? new Set() + : new Set( + ( + await this.executor + .db() + .select({ key: juniorConversationEvents.idempotencyKey }) + .from(juniorConversationEvents) + .where( + and( + eq( + juniorConversationEvents.conversationId, + conversationId, + ), + inArray( + juniorConversationEvents.idempotencyKey, + existingKeys, + ), ), - ), - ) - ).flatMap((row) => (row.key ? [row.key] : [])), + ) + ).flatMap((row) => (row.key ? [row.key] : [])), + ); + const acceptedKeys = new Set(persistedKeys); + const pending = parsed.filter((event) => { + if (event.idempotencyKey === undefined) return true; + if (acceptedKeys.has(event.idempotencyKey)) return false; + acceptedKeys.add(event.idempotencyKey); + return true; + }); + if (pending.length === 0) { + return this.appendResultFromCursor(conversationId, []); + } + const newestCreatedAtMs = Math.max( + ...pending.map((event) => event.createdAtMs), + ); + await ensureConversationRow( + this.executor, + conversationId, + newestCreatedAtMs, + options, + ); + if (options.activity !== "preserve") { + await this.executor + .db() + .update(juniorConversations) + .set({ archivedAt: null }) + .where( + and( + eq(juniorConversations.conversationId, conversationId), + isNotNull(juniorConversations.archivedAt), + ), ); - const acceptedKeys = new Set(persistedKeys); - const pending = parsed.filter((event) => { - if (event.idempotencyKey === undefined) return true; - if (acceptedKeys.has(event.idempotencyKey)) return false; - acceptedKeys.add(event.idempotencyKey); - return true; - }); - if (pending.length === 0) { - return; - } - const newestCreatedAtMs = Math.max( - ...pending.map((event) => event.createdAtMs), - ); - await ensureConversationRow( - this.executor, - conversationId, - newestCreatedAtMs, - options, - ); - if (options.activity !== "preserve") { - await this.executor + } + const cursor = await this.readCursor(conversationId); + const historyVersion = cursor.maxHistoryVersion ?? 0; + let seq = cursor.nextSeq; + const rows = pending.map((event) => + insertFromEvent(conversationId, seq++, historyVersion, event), + ); + const insertedRows = await this.executor .db() - .update(juniorConversations) - .set({ archivedAt: null }) - .where( - and( - eq(juniorConversations.conversationId, conversationId), - isNotNull(juniorConversations.archivedAt), - ), - ); - } - const cursor = await this.readCursor(conversationId); - const historyVersion = cursor.maxHistoryVersion ?? 0; - let seq = cursor.nextSeq; - const rows = pending.map((event) => - insertFromEvent(conversationId, seq++, historyVersion, event), - ); - await this.executor.db().insert(juniorConversationEvents).values(rows); - }); + .insert(juniorConversationEvents) + .values(rows) + .returning(); + return { + historyVersion, + inserted: insertedRows + .map(eventFromRow) + .sort((left, right) => left.seq - right.seq), + nextSeq: seq, + }; + }, + ); } async replaceHistory( @@ -452,6 +468,19 @@ class SqlConversationEventStore implements ConversationEventStore { return row !== undefined; } + /** Build an append result from the live cursor without inserting rows. */ + private async appendResultFromCursor( + conversationId: string, + inserted: ConversationEvent[], + ): Promise { + const cursor = await this.readCursor(conversationId); + return { + historyVersion: cursor.maxHistoryVersion ?? 0, + inserted, + nextSeq: cursor.nextSeq, + }; + } + /** Read the next sequence and active model-history version. */ private async readCursor( conversationId: string, diff --git a/packages/junior/tests/component/conversation-storage-sql.test.ts b/packages/junior/tests/component/conversation-storage-sql.test.ts index c13a51180b..5f1b6a10e5 100644 --- a/packages/junior/tests/component/conversation-storage-sql.test.ts +++ b/packages/junior/tests/component/conversation-storage-sql.test.ts @@ -372,7 +372,7 @@ describe("SQL conversation storage", () => { await seedConversation(fixture, CONVERSATION_ID); const store = createSqlConversationEventStore(fixture.sql); - await store.append(CONVERSATION_ID, [ + const first = await store.append(CONVERSATION_ID, [ { data: userMessageEvent("one"), createdAtMs: 1_000, @@ -382,12 +382,26 @@ describe("SQL conversation storage", () => { createdAtMs: 2_000, }, ]); - await store.append(CONVERSATION_ID, [ + expect(first.historyVersion).toBe(0); + expect(first.nextSeq).toBe(2); + expect(first.inserted.map((event) => event.seq)).toEqual([0, 1]); + expect(first.inserted.map((event) => event.data.type)).toEqual([ + "user_message", + "user_message", + ]); + + const second = await store.append(CONVERSATION_ID, [ { data: { type: "mcp_provider_connected", provider: "github" }, createdAtMs: 3_000, }, ]); + expect(second.historyVersion).toBe(0); + expect(second.nextSeq).toBe(3); + expect(second.inserted.map((event) => event.seq)).toEqual([2]); + expect(second.inserted.map((event) => event.data.type)).toEqual([ + "mcp_provider_connected", + ]); const history = await store.loadHistory(CONVERSATION_ID); expect(history.map((event) => event.seq)).toEqual([0, 1, 2]); @@ -494,13 +508,18 @@ describe("SQL conversation storage", () => { }; const archived = await readConversationTimestamps(); - await store.append(CONVERSATION_ID, [ + const duplicate = await store.append(CONVERSATION_ID, [ { ...firstEvent, createdAtMs: 9_000 }, ]); + expect(duplicate).toEqual({ + historyVersion: 0, + inserted: [], + nextSeq: 1, + }); expect(await readConversationTimestamps()).toEqual(archived); - await store.append(CONVERSATION_ID, [ + const mixed = await store.append(CONVERSATION_ID, [ { ...firstEvent, createdAtMs: 10_000 }, { data: userMessageEvent("second"), @@ -508,6 +527,12 @@ describe("SQL conversation storage", () => { createdAtMs: 8_000, }, ]); + expect(mixed.historyVersion).toBe(0); + expect(mixed.nextSeq).toBe(2); + expect(mixed.inserted.map((event) => event.seq)).toEqual([1]); + expect(mixed.inserted.map((event) => event.data.type)).toEqual([ + "user_message", + ]); expect(await readConversationTimestamps()).toEqual({ archivedAt: null, diff --git a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts index 974977e649..9cd073eb9b 100644 --- a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts +++ b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import type { ConversationEvent, + ConversationEventAppendResult, ConversationEventPage, ConversationEventQuery, ConversationEventStore, @@ -17,7 +18,8 @@ class MemoryConversationEventStore implements ConversationEventStore { async append( _conversationId: string, events: NewConversationEvent[], - ): Promise { + ): Promise { + const inserted: ConversationEvent[] = []; for (const event of events) { if ( event.idempotencyKey && @@ -28,7 +30,7 @@ class MemoryConversationEventStore implements ConversationEventStore { if (event.idempotencyKey) { this.idempotencyKeys.add(event.idempotencyKey); } - this.history.push({ + const next: ConversationEvent = { schemaVersion: 1, seq: this.history.length, historyVersion: 0, @@ -37,8 +39,15 @@ class MemoryConversationEventStore implements ConversationEventStore { : {}), createdAtMs: event.createdAtMs, data: event.data, - }); + }; + this.history.push(next); + inserted.push(next); } + return { + historyVersion: this.history.at(-1)?.historyVersion ?? 0, + inserted, + nextSeq: this.history.length, + }; } async replaceHistory( From 6c9e5910df36ac78b0a4785f63930d6da72d6ffa Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 18:11:07 +0000 Subject: [PATCH 2/8] ref(conversations): Narrow append result to event identities Co-Authored-By: David Cramer --- .../junior/src/chat/conversations/README.md | 4 ++-- .../junior/src/chat/conversations/history.ts | 12 +++++------ .../src/chat/conversations/projection.ts | 17 +++++++++------- .../src/chat/conversations/sql/history.ts | 20 ++++++++----------- .../conversation-storage-sql.test.ts | 13 ++++-------- .../unit/conversations/turn-lifecycle.test.ts | 7 +++++-- 6 files changed, 35 insertions(+), 38 deletions(-) diff --git a/packages/junior/src/chat/conversations/README.md b/packages/junior/src/chat/conversations/README.md index 99bd4acf9e..b4ec0957fa 100644 --- a/packages/junior/src/chat/conversations/README.md +++ b/packages/junior/src/chat/conversations/README.md @@ -59,8 +59,8 @@ older source-thread context; it does not replace Pi history. - Persist inbound `message` events before agent execution. - Persist assistant `message` events only after destination acceptance. - Append stable native agent-history events in sequence order. -- `append` returns the inserted delta and active history cursor so commit - paths can advance without reloading the full current history. +- `append` returns inserted event identities and the active history cursor so + commit paths can advance without reloading the full current history. - Reject attempts to mutate an already committed agent-history prefix. - Replace agent history only through explicit compaction or handoff. - Restore transcripts and agent history directly from conversation events. diff --git a/packages/junior/src/chat/conversations/history.ts b/packages/junior/src/chat/conversations/history.ts index fc32542ebc..f2f9f2e785 100644 --- a/packages/junior/src/chat/conversations/history.ts +++ b/packages/junior/src/chat/conversations/history.ts @@ -504,17 +504,17 @@ export const newConversationEventSchema = z export type NewConversationEvent = z.output; /** - * Result of an append: newly inserted events plus the active history version. + * Result of an append: inserted event identities plus the active cursor. * - * Callers that need a post-write cursor use this instead of reloading the full - * current history. Idempotent no-ops still report the live history version and - * an empty `inserted` list. + * Callers can advance after a write without reloading current history. The + * inserted identities stay aligned with accepted input order; idempotent + * no-ops return an empty list and the live cursor. */ export interface ConversationEventAppendResult { /** Active model-history version after the append. */ historyVersion: number; - /** Events that were newly inserted, in `seq` order. */ - inserted: ConversationEvent[]; + /** Identities assigned to newly inserted events. */ + inserted: Array<{ historyVersion: number; seq: number }>; /** Next free `seq` after the append. */ nextSeq: number; } diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index 3d68a45558..72d253a62c 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -356,8 +356,9 @@ async function commitMessagesLocked( content: context.content, }, })) ?? []; - // Append returns only the inserted delta. Build the committed cursor from the - // pre-append projection plus those rows instead of reloading full history. + // Append returns assigned identities in input order. Native messages precede + // retry-stable turn context, so their sequence suffix stays aligned without + // reloading full history. const appendResult = await eventStore.append(args.conversationId, [ ...newMessages.map((message, index) => ({ data: historyItemFromPiMessage( @@ -368,12 +369,16 @@ async function commitMessagesLocked( })), ...turnContextEvents, ]); - const appended = projectConversationEvents(appendResult.inserted); const lastInserted = appendResult.inserted.at(-1); return { committedSeq: lastInserted?.seq ?? appendResult.nextSeq - 1, historyVersion: lastInserted?.historyVersion ?? appendResult.historyVersion, - messageSeqs: [...current.seqs, ...appended.seqs], + messageSeqs: [ + ...current.seqs, + ...appendResult.inserted + .slice(0, newMessages.length) + .map((event) => event.seq), + ], messages: nextLocalMessages, provenance: nextLocalProvenance, }; @@ -502,9 +507,7 @@ async function recordAuthenticationAccountChange( actorId: args.actorId, provider: args.provider, ...(args.accountLabel ? { accountLabel: args.accountLabel } : {}), - ...(args.authorizationId - ? { authorizationId: args.authorizationId } - : {}), + ...(args.authorizationId ? { authorizationId: args.authorizationId } : {}), ...(args.providerLabel ? { providerLabel: args.providerLabel } : {}), }); await getConversationEventStore().append(args.conversationId, [ diff --git a/packages/junior/src/chat/conversations/sql/history.ts b/packages/junior/src/chat/conversations/sql/history.ts index 4647427829..171d03a1ba 100644 --- a/packages/junior/src/chat/conversations/sql/history.ts +++ b/packages/junior/src/chat/conversations/sql/history.ts @@ -109,7 +109,7 @@ class SqlConversationEventStore implements ConversationEventStore { conversationId, async () => { if (parsed.length === 0) { - return this.appendResultFromCursor(conversationId, []); + return this.appendResultFromCursor(conversationId); } const existingKeys = parsed .map((event) => event.idempotencyKey) @@ -145,7 +145,7 @@ class SqlConversationEventStore implements ConversationEventStore { return true; }); if (pending.length === 0) { - return this.appendResultFromCursor(conversationId, []); + return this.appendResultFromCursor(conversationId); } const newestCreatedAtMs = Math.max( ...pending.map((event) => event.createdAtMs), @@ -174,16 +174,13 @@ class SqlConversationEventStore implements ConversationEventStore { const rows = pending.map((event) => insertFromEvent(conversationId, seq++, historyVersion, event), ); - const insertedRows = await this.executor - .db() - .insert(juniorConversationEvents) - .values(rows) - .returning(); + await this.executor.db().insert(juniorConversationEvents).values(rows); return { historyVersion, - inserted: insertedRows - .map(eventFromRow) - .sort((left, right) => left.seq - right.seq), + inserted: rows.map((row) => ({ + historyVersion: row.historyVersion, + seq: row.seq, + })), nextSeq: seq, }; }, @@ -471,12 +468,11 @@ class SqlConversationEventStore implements ConversationEventStore { /** Build an append result from the live cursor without inserting rows. */ private async appendResultFromCursor( conversationId: string, - inserted: ConversationEvent[], ): Promise { const cursor = await this.readCursor(conversationId); return { historyVersion: cursor.maxHistoryVersion ?? 0, - inserted, + inserted: [], nextSeq: cursor.nextSeq, }; } diff --git a/packages/junior/tests/component/conversation-storage-sql.test.ts b/packages/junior/tests/component/conversation-storage-sql.test.ts index 5f1b6a10e5..493e043ab4 100644 --- a/packages/junior/tests/component/conversation-storage-sql.test.ts +++ b/packages/junior/tests/component/conversation-storage-sql.test.ts @@ -385,9 +385,8 @@ describe("SQL conversation storage", () => { expect(first.historyVersion).toBe(0); expect(first.nextSeq).toBe(2); expect(first.inserted.map((event) => event.seq)).toEqual([0, 1]); - expect(first.inserted.map((event) => event.data.type)).toEqual([ - "user_message", - "user_message", + expect(first.inserted.map((event) => event.historyVersion)).toEqual([ + 0, 0, ]); const second = await store.append(CONVERSATION_ID, [ @@ -399,9 +398,7 @@ describe("SQL conversation storage", () => { expect(second.historyVersion).toBe(0); expect(second.nextSeq).toBe(3); expect(second.inserted.map((event) => event.seq)).toEqual([2]); - expect(second.inserted.map((event) => event.data.type)).toEqual([ - "mcp_provider_connected", - ]); + expect(second.inserted.map((event) => event.historyVersion)).toEqual([0]); const history = await store.loadHistory(CONVERSATION_ID); expect(history.map((event) => event.seq)).toEqual([0, 1, 2]); @@ -530,9 +527,7 @@ describe("SQL conversation storage", () => { expect(mixed.historyVersion).toBe(0); expect(mixed.nextSeq).toBe(2); expect(mixed.inserted.map((event) => event.seq)).toEqual([1]); - expect(mixed.inserted.map((event) => event.data.type)).toEqual([ - "user_message", - ]); + expect(mixed.inserted.map((event) => event.historyVersion)).toEqual([0]); expect(await readConversationTimestamps()).toEqual({ archivedAt: null, diff --git a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts index 9cd073eb9b..8f82a057ee 100644 --- a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts +++ b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts @@ -19,7 +19,7 @@ class MemoryConversationEventStore implements ConversationEventStore { _conversationId: string, events: NewConversationEvent[], ): Promise { - const inserted: ConversationEvent[] = []; + const inserted: ConversationEventAppendResult["inserted"] = []; for (const event of events) { if ( event.idempotencyKey && @@ -41,7 +41,10 @@ class MemoryConversationEventStore implements ConversationEventStore { data: event.data, }; this.history.push(next); - inserted.push(next); + inserted.push({ + historyVersion: next.historyVersion, + seq: next.seq, + }); } return { historyVersion: this.history.at(-1)?.historyVersion ?? 0, From 8a7b858b5831696d006173098da6e579053c5fa1 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:02:44 +0000 Subject: [PATCH 3/8] perf(conversations): Fence commits on caller-owned history cursors Separate message and turn-context appends, return committedSeq from append, and let turn-session checkpoints pass an already-materialized base so commitMessages only writes the delta after verifying the live cursor. Co-Authored-By: David Cramer --- .../junior/src/chat/conversations/README.md | 4 + .../junior/src/chat/conversations/history.ts | 20 ++- .../src/chat/conversations/projection.ts | 147 +++++++++++++----- .../src/chat/conversations/sql/history.ts | 10 +- .../src/chat/services/turn-session-record.ts | 2 + .../junior/src/chat/state/turn-session.ts | 73 +++++++-- .../conversation-storage-sql.test.ts | 13 +- .../conversations/commit-messages.test.ts | 123 +++++++++++++++ .../unit/conversations/turn-lifecycle.test.ts | 3 +- 9 files changed, 323 insertions(+), 72 deletions(-) create mode 100644 packages/junior/tests/component/conversations/commit-messages.test.ts diff --git a/packages/junior/src/chat/conversations/README.md b/packages/junior/src/chat/conversations/README.md index b4ec0957fa..357c43314b 100644 --- a/packages/junior/src/chat/conversations/README.md +++ b/packages/junior/src/chat/conversations/README.md @@ -61,6 +61,10 @@ older source-thread context; it does not replace Pi history. - Append stable native agent-history events in sequence order. - `append` returns inserted event identities and the active history cursor so commit paths can advance without reloading the full current history. +- Prefer cursor-fenced commits: callers that already hold a committed base pass + it to `commitMessages`, which verifies the live cursor and appends only the + delta. Message events and host-only turn context are appended separately so + message sequence assignment does not depend on mixed-event order. - Reject attempts to mutate an already committed agent-history prefix. - Replace agent history only through explicit compaction or handoff. - Restore transcripts and agent history directly from conversation events. diff --git a/packages/junior/src/chat/conversations/history.ts b/packages/junior/src/chat/conversations/history.ts index f2f9f2e785..977a3d4b7e 100644 --- a/packages/junior/src/chat/conversations/history.ts +++ b/packages/junior/src/chat/conversations/history.ts @@ -503,20 +503,26 @@ export const newConversationEventSchema = z /** An event to append; the store assigns `seq` and current history version. */ export type NewConversationEvent = z.output; +/** Identity assigned to one newly inserted conversation event. */ +export interface ConversationEventIdentity { + /** Sequence assigned to the inserted event. */ + seq: number; +} + /** * Result of an append: inserted event identities plus the active cursor. * - * Callers can advance after a write without reloading current history. The - * inserted identities stay aligned with accepted input order; idempotent - * no-ops return an empty list and the live cursor. + * Callers can advance after a write without reloading current history. + * Identities stay aligned with accepted input order. Idempotent no-ops return + * an empty list and the live cursor. */ export interface ConversationEventAppendResult { /** Active model-history version after the append. */ historyVersion: number; - /** Identities assigned to newly inserted events. */ - inserted: Array<{ historyVersion: number; seq: number }>; - /** Next free `seq` after the append. */ - nextSeq: number; + /** Identities assigned to newly inserted events, in input order. */ + inserted: ConversationEventIdentity[]; + /** `seq` of the latest event after the append, or -1 when none exist. */ + committedSeq: number; } /** Bounded observational page over the durable conversation event log. */ diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index 72d253a62c..3e2ad3421e 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -221,13 +221,37 @@ function messageTimestamp(message: PiMessage): number { } /** - * Append newly stable native history items. A shorter or changed prefix indicates - * that a caller persisted volatile Pi state; only compaction and handoff may - * intentionally replace active model history. + * Already-committed agent history known to the caller. + * + * When present, the commit path fences on this cursor and appends only the + * delta instead of reloading and deep-comparing the full active history. + */ +export interface CommitMessagesBase { + /** `seq` of the last event already committed for this base. */ + committedSeq: number; + /** History version that owns `committedSeq`. */ + historyVersion: number; + /** Event sequence for every projected agent-history item already committed. */ + messageSeqs: number[]; + /** Durable messages already committed for this base. */ + messages: PiMessage[]; + /** Provenance aligned one-to-one with `messages`. */ + provenance: ConversationMessageProvenance[]; +} + +/** + * Append newly stable native history items. + * + * Prefer supplying `base` from an already-loaded turn projection so checkpoints + * only write the delta. Without `base`, the store loads current history and + * rejects a shorter or changed committed prefix. Only compaction and handoff + * may intentionally replace active model history. */ export async function commitMessages(args: { conversationId: string; messages: PiMessage[]; + /** Already-committed cursor/projection used to fence and append the delta. */ + base?: CommitMessagesBase; /** Explicit per-message provenance aligned one-to-one with `messages`. */ provenance?: ConversationMessageProvenance[]; /** Explicit provenance for the trailing newly committed messages. */ @@ -239,7 +263,7 @@ export async function commitMessages(args: { contexts: PluginTurnContext[]; turnId: string; }; - /** SQL authority for the atomic commit; defaults to the process executor. */ + /** SQL executor for the atomic commit; defaults to the process executor. */ executor?: JuniorSqlDatabase; }): Promise<{ committedSeq: number; @@ -303,27 +327,70 @@ export async function commitAcceptedReply(args: { ); } -async function commitMessagesLocked( +async function resolveCommitBase( args: Parameters[0], - executor: JuniorSqlDatabase, -): ReturnType { - const eventStore = createSqlConversationEventStore(executor); + eventStore: ReturnType, + nextLocalMessages: PiMessage[], +): Promise { + if (args.base) { + const matchingPrefix = countMatchingPrefix( + args.base.messages, + nextLocalMessages, + ); + if (matchingPrefix !== args.base.messages.length) { + throw new Error( + `Agent history for ${args.conversationId} changed before its committed boundary`, + ); + } + const live = await eventStore.append(args.conversationId, []); + if ( + live.historyVersion !== args.base.historyVersion || + live.committedSeq !== args.base.committedSeq + ) { + throw new Error( + `Agent history for ${args.conversationId} changed before its committed boundary`, + ); + } + return args.base; + } + const currentEvents = await eventStore.loadCurrentHistory( args.conversationId, ); const current = projectConversationEvents(currentEvents); + const matchingPrefix = countMatchingPrefix( + current.messages, + nextLocalMessages, + ); + if (matchingPrefix !== current.messages.length) { + throw new Error( + `Agent history for ${args.conversationId} changed before its committed boundary`, + ); + } + return { + committedSeq: currentEvents.at(-1)?.seq ?? -1, + historyVersion: currentEvents.at(-1)?.historyVersion ?? 0, + messageSeqs: current.seqs, + messages: current.messages, + provenance: current.provenance, + }; +} + +async function commitMessagesLocked( + args: Parameters[0], + executor: JuniorSqlDatabase, +): ReturnType { + const eventStore = createSqlConversationEventStore(executor); // Runtime bootstrap is per-run input, not durable agent history. Session // records may retain it while a turn is live, but event replay must not need // a compensating history rewrite when that bootstrap changes. const nextLocalMessages = stripRuntimeTurnContext(args.messages).map( normalizeDurableMessage, ); - const matchingPrefix = countMatchingPrefix( - current.messages, - nextLocalMessages, - ); + const base = await resolveCommitBase(args, eventStore, nextLocalMessages); + const matchingPrefix = base.messages.length; const nextLocalProvenance = resolveCommitProvenance({ - existing: current, + existing: base, nextMessages: nextLocalMessages, matchingPrefix, ...(args.provenance ? { explicitProvenance: args.provenance } : {}), @@ -334,11 +401,6 @@ async function commitMessagesLocked( ? { newMessageProvenance: args.newMessageProvenance } : {}), }); - if (matchingPrefix !== current.messages.length) { - throw new Error( - `Agent history for ${args.conversationId} changed before its committed boundary`, - ); - } const newMessages = nextLocalMessages.slice(matchingPrefix); const turnContext = args.turnContext; const turnContextEvents = @@ -356,28 +418,37 @@ async function commitMessagesLocked( content: context.content, }, })) ?? []; - // Append returns assigned identities in input order. Native messages precede - // retry-stable turn context, so their sequence suffix stays aligned without - // reloading full history. - const appendResult = await eventStore.append(args.conversationId, [ - ...newMessages.map((message, index) => ({ - data: historyItemFromPiMessage( - message, - nextLocalProvenance[matchingPrefix + index]!, - ), - createdAtMs: messageTimestamp(message), - })), - ...turnContextEvents, - ]); - const lastInserted = appendResult.inserted.at(-1); + + // Append native messages and host-only turn context separately so message + // sequence assignment never depends on mixed-event ordering assumptions. + const messageAppend = + newMessages.length === 0 + ? { + historyVersion: base.historyVersion, + inserted: [] as Array<{ seq: number }>, + committedSeq: base.committedSeq, + } + : await eventStore.append( + args.conversationId, + newMessages.map((message, index) => ({ + data: historyItemFromPiMessage( + message, + nextLocalProvenance[matchingPrefix + index]!, + ), + createdAtMs: messageTimestamp(message), + })), + ); + const contextAppend = + turnContextEvents.length === 0 + ? messageAppend + : await eventStore.append(args.conversationId, turnContextEvents); + return { - committedSeq: lastInserted?.seq ?? appendResult.nextSeq - 1, - historyVersion: lastInserted?.historyVersion ?? appendResult.historyVersion, + committedSeq: contextAppend.committedSeq, + historyVersion: contextAppend.historyVersion, messageSeqs: [ - ...current.seqs, - ...appendResult.inserted - .slice(0, newMessages.length) - .map((event) => event.seq), + ...base.messageSeqs, + ...messageAppend.inserted.map((event) => event.seq), ], messages: nextLocalMessages, provenance: nextLocalProvenance, diff --git a/packages/junior/src/chat/conversations/sql/history.ts b/packages/junior/src/chat/conversations/sql/history.ts index 171d03a1ba..a37e851d59 100644 --- a/packages/junior/src/chat/conversations/sql/history.ts +++ b/packages/junior/src/chat/conversations/sql/history.ts @@ -175,13 +175,11 @@ class SqlConversationEventStore implements ConversationEventStore { insertFromEvent(conversationId, seq++, historyVersion, event), ); await this.executor.db().insert(juniorConversationEvents).values(rows); + const inserted = rows.map((row) => ({ seq: row.seq })); return { historyVersion, - inserted: rows.map((row) => ({ - historyVersion: row.historyVersion, - seq: row.seq, - })), - nextSeq: seq, + inserted, + committedSeq: inserted.at(-1)?.seq ?? seq - 1, }; }, ); @@ -473,7 +471,7 @@ class SqlConversationEventStore implements ConversationEventStore { return { historyVersion: cursor.maxHistoryVersion ?? 0, inserted: [], - nextSeq: cursor.nextSeq, + committedSeq: cursor.nextSeq - 1, }; } diff --git a/packages/junior/src/chat/services/turn-session-record.ts b/packages/junior/src/chat/services/turn-session-record.ts index 7c8f82693b..dcb5252b59 100644 --- a/packages/junior/src/chat/services/turn-session-record.ts +++ b/packages/junior/src/chat/services/turn-session-record.ts @@ -148,6 +148,7 @@ export async function persistRunningSessionRecord(args: { sliceId: args.sliceId, state: "running", piMessages: args.messages, + ...(latestSessionRecord ? { existing: latestSessionRecord } : {}), ...(args.trailingMessageProvenance ? { trailingMessageProvenance: args.trailingMessageProvenance } : {}), @@ -308,6 +309,7 @@ export async function persistCompletedSessionRecord(args: { latestSessionRecord?.turnStartMessageIndex, } : {}), + ...(latestSessionRecord ? { existing: latestSessionRecord } : {}), }; await persistWithRetry(async () => { await upsertAgentTurnSessionRecord(target); diff --git a/packages/junior/src/chat/state/turn-session.ts b/packages/junior/src/chat/state/turn-session.ts index d248c3b8ec..713424ea77 100644 --- a/packages/junior/src/chat/state/turn-session.ts +++ b/packages/junior/src/chat/state/turn-session.ts @@ -68,6 +68,7 @@ export type AgentDispatchOutcome = "blocked" | "completed" | "failed"; interface ConversationMessageProjection { messages: PiMessage[]; provenance: ConversationMessageProvenance[]; + seqs: number[]; } export interface AgentTurnSessionRecord { @@ -90,6 +91,15 @@ export interface AgentTurnSessionRecord { piMessages: PiMessage[]; /** Per-message provenance aligned one-to-one with `piMessages`. */ piMessageProvenance: ConversationMessageProvenance[]; + /** + * `seq` of the last durable event whose projection reproduces `piMessages` + * without volatile bootstrap; -1 when nothing was committed. + */ + committedSeq: number; + /** History version that owns `committedSeq` and any volatile bootstrap. */ + historyVersion: number; + /** Event sequence for every projected durable agent-history item. */ + messageSeqs: number[]; /** * All distinct actors annotated on this run's committed instruction-authority * messages, in first-seen order. Persisted as an attribution handle so a @@ -117,21 +127,29 @@ export type AgentTurnSessionSummary = Omit< | "actors" | "piMessages" | "piMessageProvenance" + | "committedSeq" + | "historyVersion" + | "messageSeqs" | "turnStartMessageIndex" >; interface StoredAgentTurnSessionRecord extends Omit< AgentTurnSessionRecord, - "actors" | "piMessages" | "piMessageProvenance" | "turnStartMessageIndex" + | "actors" + | "piMessages" + | "piMessageProvenance" + | "messageSeqs" + | "historyVersion" + | "turnStartMessageIndex" > { actors?: Actor[]; - /** - * `seq` of the last event in `junior_conversation_events` whose projection reproduces - * this record's committed Pi messages; -1 when nothing was committed. - */ - committedSeq: number; /** History version that owns `committedSeq` and any volatile bootstrap. */ historyVersion?: number; + /** + * Event sequence for every projected durable agent-history item. Optional on + * older session records that predate cursor-fenced commits. + */ + messageSeqs?: number[]; /** * `seq` boundary where this turn's fresh prompt starts: the seq of the last * projected message before the prompt, or -1 when the turn starts the epoch. @@ -199,6 +217,7 @@ const storedAgentTurnSessionRecordSchema = agentTurnSessionSummarySchema actors: z.array(actorSchema).optional(), committedSeq: seqCursorSchema, historyVersion: z.number().int().nonnegative().optional(), + messageSeqs: z.array(seqCursorSchema).optional(), errorMessage: z.string().optional(), turnStartSeq: seqCursorSchema.optional(), runtimeContext: z.array(piMessageSchema).optional(), @@ -320,6 +339,9 @@ function materializeAgentTurnSessionRecord( updatedAtMs: stored.updatedAtMs, piMessages, piMessageProvenance: piProjection.provenance, + committedSeq: stored.committedSeq, + historyVersion: stored.historyVersion ?? 0, + messageSeqs: stored.messageSeqs ?? piProjection.seqs, actors: stored.actors ?? instructionActors(piProjection.provenance), cumulativeDurationMs: stored.cumulativeDurationMs, ...(stored.destination ? { destination: stored.destination } : {}), @@ -488,6 +510,7 @@ function buildStoredRecord(args: { source?: Source; committedSeq: number; historyVersion?: number; + messageSeqs?: number[]; lastProgressAtMs?: number; loadedSkillNames?: string[]; modelId?: string; @@ -522,6 +545,7 @@ function buildStoredRecord(args: { ...(args.historyVersion !== undefined ? { historyVersion: args.historyVersion } : {}), + ...(args.messageSeqs ? { messageSeqs: args.messageSeqs } : {}), ...(args.turnStartSeq !== undefined ? { turnStartSeq: args.turnStartSeq } : {}), @@ -591,6 +615,7 @@ async function setStoredRecord(args: { { messages: [...args.piMessages], provenance: [...args.piMessageProvenance], + seqs: args.record.messageSeqs ?? [], }, args.turnStartMessageIndex, ); @@ -629,6 +654,7 @@ async function updateAgentTurnSessionState(args: { ...(parsed.historyVersion !== undefined ? { historyVersion: parsed.historyVersion } : {}), + ...(parsed.messageSeqs ? { messageSeqs: parsed.messageSeqs } : {}), ...(parsed.turnStartSeq !== undefined ? { turnStartSeq: parsed.turnStartSeq } : {}), @@ -714,13 +740,19 @@ export async function upsertAgentTurnSessionRecord(args: { turnContexts?: PluginTurnContext[]; turnStartMessageIndex?: number; ttlMs?: number; + /** Already-materialized session used to fence commits without another reload. */ + existing?: AgentTurnSessionRecord; }): Promise { - const existingRecord = await getStoredAgentTurnSessionRecord( + const existingRecord = + args.existing ?? + (await getAgentTurnSessionRecord(args.conversationId, args.sessionId)); + const storedRecord = await getStoredAgentTurnSessionRecord( args.conversationId, args.sessionId, ); const existingDispatchId = existingRecord?.dispatchId ?? + storedRecord?.dispatchId ?? ( await listAgentTurnSessionSummariesForConversation(args.conversationId) ).find((summary) => summary.sessionId === args.sessionId)?.dispatchId; @@ -738,9 +770,29 @@ export async function upsertAgentTurnSessionRecord(args: { // store reuses committed provenance for the unchanged prefix and defaults the // rest to context. Platform-neutral so local identities are preserved too. const instructionActor = args.actor ?? existingRecord?.actor; + const durableExistingMessages = existingRecord + ? stripRuntimeTurnContext(existingRecord.piMessages) + : undefined; + const commitBase = + existingRecord && + durableExistingMessages && + existingRecord.messageSeqs.length === durableExistingMessages.length && + existingRecord.piMessageProvenance.length === durableExistingMessages.length + ? { + committedSeq: existingRecord.committedSeq, + historyVersion: existingRecord.historyVersion, + messageSeqs: existingRecord.messageSeqs, + messages: durableExistingMessages, + provenance: existingRecord.piMessageProvenance.slice( + 0, + durableExistingMessages.length, + ), + } + : undefined; const commit = await commitMessages({ conversationId: args.conversationId, messages: args.piMessages, + ...(commitBase ? { base: commitBase } : {}), ...(instructionActor ? { newMessageProvenance: instructionProvenanceFor(instructionActor) } : {}), @@ -767,13 +819,13 @@ export async function upsertAgentTurnSessionRecord(args: { runtimeContext.length > 0 ? runtimeContext : existingRecord?.historyVersion === commit.historyVersion - ? existingRecord.runtimeContext + ? storedRecord?.runtimeContext : undefined; // Flip the caller's message-index cursor into a durable seq reference: the // seq of the last committed message before the turn's fresh prompt. const turnStartSeq = durableTurnStartMessageIndex === undefined - ? existingRecord?.turnStartSeq + ? storedRecord?.turnStartSeq : durableTurnStartMessageIndex <= 0 ? -1 : (commit.messageSeqs[durableTurnStartMessageIndex - 1] ?? @@ -807,11 +859,12 @@ export async function upsertAgentTurnSessionRecord(args: { : {}), committedSeq: commit.committedSeq, historyVersion: commit.historyVersion, + messageSeqs: commit.messageSeqs, ...(turnStartSeq !== undefined ? { turnStartSeq } : {}), ...(retainedRuntimeContext ? { runtimeContext: retainedRuntimeContext } : {}), - previousVersion: existingRecord?.version, + previousVersion: storedRecord?.version ?? existingRecord?.version, cumulativeDurationMs: args.cumulativeDurationMs ?? existingRecord?.cumulativeDurationMs ?? 0, ...(args.cumulativeUsage diff --git a/packages/junior/tests/component/conversation-storage-sql.test.ts b/packages/junior/tests/component/conversation-storage-sql.test.ts index 493e043ab4..47e8cdffaa 100644 --- a/packages/junior/tests/component/conversation-storage-sql.test.ts +++ b/packages/junior/tests/component/conversation-storage-sql.test.ts @@ -383,11 +383,8 @@ describe("SQL conversation storage", () => { }, ]); expect(first.historyVersion).toBe(0); - expect(first.nextSeq).toBe(2); + expect(first.committedSeq).toBe(1); expect(first.inserted.map((event) => event.seq)).toEqual([0, 1]); - expect(first.inserted.map((event) => event.historyVersion)).toEqual([ - 0, 0, - ]); const second = await store.append(CONVERSATION_ID, [ { @@ -396,9 +393,8 @@ describe("SQL conversation storage", () => { }, ]); expect(second.historyVersion).toBe(0); - expect(second.nextSeq).toBe(3); + expect(second.committedSeq).toBe(2); expect(second.inserted.map((event) => event.seq)).toEqual([2]); - expect(second.inserted.map((event) => event.historyVersion)).toEqual([0]); const history = await store.loadHistory(CONVERSATION_ID); expect(history.map((event) => event.seq)).toEqual([0, 1, 2]); @@ -511,7 +507,7 @@ describe("SQL conversation storage", () => { expect(duplicate).toEqual({ historyVersion: 0, inserted: [], - nextSeq: 1, + committedSeq: 0, }); expect(await readConversationTimestamps()).toEqual(archived); @@ -525,9 +521,8 @@ describe("SQL conversation storage", () => { }, ]); expect(mixed.historyVersion).toBe(0); - expect(mixed.nextSeq).toBe(2); + expect(mixed.committedSeq).toBe(1); expect(mixed.inserted.map((event) => event.seq)).toEqual([1]); - expect(mixed.inserted.map((event) => event.historyVersion)).toEqual([0]); expect(await readConversationTimestamps()).toEqual({ archivedAt: null, diff --git a/packages/junior/tests/component/conversations/commit-messages.test.ts b/packages/junior/tests/component/conversations/commit-messages.test.ts new file mode 100644 index 0000000000..74727b73ea --- /dev/null +++ b/packages/junior/tests/component/conversations/commit-messages.test.ts @@ -0,0 +1,123 @@ +import { describe, expect, it } from "vitest"; +import { commitMessages } from "@/chat/conversations/projection"; +import { createSqlConversationEventStore } from "@/chat/conversations/sql/history"; +import { migrateSchema } from "@/chat/conversations/sql/migrations"; +import type { PiMessage } from "@/chat/pi/messages"; +import { createLocalJuniorSqlFixture } from "../../fixtures/sql"; + +const CONVERSATION_ID = "slack:CCOMMIT:1718123456.000000"; + +function user(text: string, timestamp: number): PiMessage { + return { + role: "user", + content: [{ type: "text", text }], + timestamp, + } as PiMessage; +} + +function assistant(text: string, timestamp: number): PiMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "openai-responses", + provider: "openai", + model: "test-model", + usage: {}, + stopReason: "stop", + timestamp, + } as PiMessage; +} + +describe("commitMessages cursor fencing", () => { + it("appends only the delta from a fenced base without rewriting message seqs", async () => { + const fixture = await createLocalJuniorSqlFixture(); + try { + await migrateSchema(fixture.sql); + + const first = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1), assistant("hi", 2)], + executor: fixture.sql, + }); + expect(first.committedSeq).toBe(1); + expect(first.messageSeqs).toEqual([0, 1]); + expect(first.historyVersion).toBe(0); + + const second = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1), assistant("hi", 2), user("follow up", 3)], + base: { + committedSeq: first.committedSeq, + historyVersion: first.historyVersion, + messageSeqs: first.messageSeqs, + messages: first.messages, + provenance: first.provenance, + }, + turnContext: { + turnId: "turn-1", + contexts: [ + { + pluginName: "test-plugin", + kind: "note", + version: 1, + content: { text: "context" }, + loadedAtMs: 4, + }, + ], + }, + executor: fixture.sql, + }); + + expect(second.committedSeq).toBe(3); + expect(second.messageSeqs).toEqual([0, 1, 2]); + expect(second.historyVersion).toBe(0); + + const store = createSqlConversationEventStore(fixture.sql); + const history = await store.loadHistory(CONVERSATION_ID); + expect(history.map((event) => [event.seq, event.data.type])).toEqual([ + [0, "user_message"], + [1, "assistant_message"], + [2, "user_message"], + [3, "turn_context"], + ]); + } finally { + await fixture.close(); + } + }); + + it("rejects a stale base cursor before writing", async () => { + const fixture = await createLocalJuniorSqlFixture(); + try { + await migrateSchema(fixture.sql); + + const first = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1)], + executor: fixture.sql, + }); + + await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1), assistant("hi", 2)], + executor: fixture.sql, + }); + + await expect( + commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1), assistant("hi", 2), user("stale", 3)], + base: { + committedSeq: first.committedSeq, + historyVersion: first.historyVersion, + messageSeqs: first.messageSeqs, + messages: first.messages, + provenance: first.provenance, + }, + executor: fixture.sql, + }), + ).rejects.toThrow(/changed before its committed boundary/); + } finally { + await fixture.close(); + } + }); +}); diff --git a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts index 8f82a057ee..9c9933a2f8 100644 --- a/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts +++ b/packages/junior/tests/unit/conversations/turn-lifecycle.test.ts @@ -42,14 +42,13 @@ class MemoryConversationEventStore implements ConversationEventStore { }; this.history.push(next); inserted.push({ - historyVersion: next.historyVersion, seq: next.seq, }); } return { historyVersion: this.history.at(-1)?.historyVersion ?? 0, inserted, - nextSeq: this.history.length, + committedSeq: this.history.at(-1)?.seq ?? -1, }; } From 99e9fa35ae0ef0cd8f82346b9ce596cb8e99334f Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:18:51 +0000 Subject: [PATCH 4/8] fix(conversations): Allow host-only events past commit fences Exact global-cursor equality rejected checkpoints after MCP connects, auth requests, and other non-agent events advanced seq. Fence on agent message prefix and message seqs instead, and adopt the live cursor when only host-only facts moved it. --- .../junior/src/chat/conversations/README.md | 9 ++-- .../src/chat/conversations/projection.ts | 45 ++++++++++++---- .../conversations/commit-messages.test.ts | 52 +++++++++++++++++++ 3 files changed, 92 insertions(+), 14 deletions(-) diff --git a/packages/junior/src/chat/conversations/README.md b/packages/junior/src/chat/conversations/README.md index 357c43314b..cc5927bd90 100644 --- a/packages/junior/src/chat/conversations/README.md +++ b/packages/junior/src/chat/conversations/README.md @@ -62,9 +62,12 @@ older source-thread context; it does not replace Pi history. - `append` returns inserted event identities and the active history cursor so commit paths can advance without reloading the full current history. - Prefer cursor-fenced commits: callers that already hold a committed base pass - it to `commitMessages`, which verifies the live cursor and appends only the - delta. Message events and host-only turn context are appended separately so - message sequence assignment does not depend on mixed-event order. + it to `commitMessages`, which verifies the live agent-history prefix and + appends only the delta. Host-only events may advance the global cursor after + the base; the fence still holds when projected agent messages and message + seqs are unchanged. Concurrent agent-history writes still fail closed. + Message events and host-only turn context are appended separately so message + sequence assignment does not depend on mixed-event order. - Reject attempts to mutate an already committed agent-history prefix. - Replace agent history only through explicit compaction or handoff. - Restore transcripts and agent history directly from conversation events. diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index 3e2ad3421e..891e457044 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -327,6 +327,12 @@ export async function commitAcceptedReply(args: { ); } +function throwCommittedBoundaryChanged(conversationId: string): never { + throw new Error( + `Agent history for ${conversationId} changed before its committed boundary`, + ); +} + async function resolveCommitBase( args: Parameters[0], eventStore: ReturnType, @@ -338,20 +344,39 @@ async function resolveCommitBase( nextLocalMessages, ); if (matchingPrefix !== args.base.messages.length) { - throw new Error( - `Agent history for ${args.conversationId} changed before its committed boundary`, - ); + throwCommittedBoundaryChanged(args.conversationId); } + // Empty append is the cheap live cursor read under the conversation lock. const live = await eventStore.append(args.conversationId, []); if ( live.historyVersion !== args.base.historyVersion || - live.committedSeq !== args.base.committedSeq + live.committedSeq < args.base.committedSeq ) { - throw new Error( - `Agent history for ${args.conversationId} changed before its committed boundary`, - ); + throwCommittedBoundaryChanged(args.conversationId); } - return args.base; + if (live.committedSeq === args.base.committedSeq) { + return args.base; + } + // Global cursor advanced after the caller's base. Host-only facts (MCP + // connect, turn_context, native events) do that without changing agent + // history; concurrent agent-message writes must still fail closed. + const currentEvents = await eventStore.loadCurrentHistory( + args.conversationId, + ); + const current = projectConversationEvents(currentEvents); + if ( + current.messages.length !== args.base.messages.length || + current.seqs.length !== args.base.messageSeqs.length || + countMatchingPrefix(args.base.messages, current.messages) !== + args.base.messages.length || + args.base.messageSeqs.some((seq, index) => current.seqs[index] !== seq) + ) { + throwCommittedBoundaryChanged(args.conversationId); + } + return { + ...args.base, + committedSeq: live.committedSeq, + }; } const currentEvents = await eventStore.loadCurrentHistory( @@ -363,9 +388,7 @@ async function resolveCommitBase( nextLocalMessages, ); if (matchingPrefix !== current.messages.length) { - throw new Error( - `Agent history for ${args.conversationId} changed before its committed boundary`, - ); + throwCommittedBoundaryChanged(args.conversationId); } return { committedSeq: currentEvents.at(-1)?.seq ?? -1, diff --git a/packages/junior/tests/component/conversations/commit-messages.test.ts b/packages/junior/tests/component/conversations/commit-messages.test.ts index 74727b73ea..a24a4a1bf9 100644 --- a/packages/junior/tests/component/conversations/commit-messages.test.ts +++ b/packages/junior/tests/component/conversations/commit-messages.test.ts @@ -120,4 +120,56 @@ describe("commitMessages cursor fencing", () => { await fixture.close(); } }); + + it("keeps a fenced base when only host-only events advanced the cursor", async () => { + const fixture = await createLocalJuniorSqlFixture(); + try { + await migrateSchema(fixture.sql); + + const first = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1), assistant("hi", 2)], + executor: fixture.sql, + }); + + const store = createSqlConversationEventStore(fixture.sql); + await store.append(CONVERSATION_ID, [ + { + data: { type: "mcp_provider_connected", provider: "github" }, + createdAtMs: 3, + }, + ]); + + const second = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [ + user("hello", 1), + assistant("hi", 2), + user("follow up", 4), + ], + base: { + committedSeq: first.committedSeq, + historyVersion: first.historyVersion, + messageSeqs: first.messageSeqs, + messages: first.messages, + provenance: first.provenance, + }, + executor: fixture.sql, + }); + + expect(second.messageSeqs).toEqual([0, 1, 3]); + expect(second.committedSeq).toBe(3); + expect(second.historyVersion).toBe(0); + + const history = await store.loadHistory(CONVERSATION_ID); + expect(history.map((event) => [event.seq, event.data.type])).toEqual([ + [0, "user_message"], + [1, "assistant_message"], + [2, "mcp_provider_connected"], + [3, "user_message"], + ]); + } finally { + await fixture.close(); + } + }); }); From 1f4b1276cfe46ceb7934c9ff7e3e2fda8a7d36c9 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:20:48 +0000 Subject: [PATCH 5/8] fix(conversations): Strip messageSeqs from turn session summaries Session summaries use a strict schema without fence fields. messageSeqs was left on the summary payload, so index readers rejected every new summary and hid sessions from operational listings. --- packages/junior/src/chat/state/turn-session.ts | 1 + .../tests/component/services/turn-session-record.test.ts | 3 +++ packages/junior/tests/integration/reporting-support.test.ts | 3 +++ 3 files changed, 7 insertions(+) diff --git a/packages/junior/src/chat/state/turn-session.ts b/packages/junior/src/chat/state/turn-session.ts index 713424ea77..905b99c10e 100644 --- a/packages/junior/src/chat/state/turn-session.ts +++ b/packages/junior/src/chat/state/turn-session.ts @@ -604,6 +604,7 @@ async function setStoredRecord(args: { actors: _actors, committedSeq: _committedSeq, historyVersion: _historyVersion, + messageSeqs: _messageSeqs, errorMessage: _errorMessage, turnStartSeq: _turnStartSeq, runtimeContext: _runtimeContext, diff --git a/packages/junior/tests/component/services/turn-session-record.test.ts b/packages/junior/tests/component/services/turn-session-record.test.ts index 0b6f48090b..ea899d79a9 100644 --- a/packages/junior/tests/component/services/turn-session-record.test.ts +++ b/packages/junior/tests/component/services/turn-session-record.test.ts @@ -517,6 +517,9 @@ describe("persistAuthPauseSessionRecord", () => { "conversation-turn-scope", ); expect(summaries[0]).not.toHaveProperty("turnStartMessageIndex"); + expect(summaries[0]).not.toHaveProperty("messageSeqs"); + expect(summaries[0]).not.toHaveProperty("committedSeq"); + expect(summaries[0]).not.toHaveProperty("historyVersion"); }); it("persists and materializes per-message provenance aligned to piMessages", async () => { diff --git a/packages/junior/tests/integration/reporting-support.test.ts b/packages/junior/tests/integration/reporting-support.test.ts index 3ccf42ff0c..dd0ee151a8 100644 --- a/packages/junior/tests/integration/reporting-support.test.ts +++ b/packages/junior/tests/integration/reporting-support.test.ts @@ -67,6 +67,9 @@ describe("reporting support", () => { loadedSkillNames: ["triage"], }); expect(matching[0]).not.toHaveProperty("errorMessage"); + expect(matching[0]).not.toHaveProperty("messageSeqs"); + expect(matching[0]).not.toHaveProperty("committedSeq"); + expect(matching[0]).not.toHaveProperty("historyVersion"); }); it("lists recent conversations through the conversation reporting API", async () => { From 117c7e4d9d04a0fe87f3aac3e4d02dd39b48a34a Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:19:08 +0000 Subject: [PATCH 6/8] fix(conversations): Harden cursor fences for handoff and races Reject non-prefix session bases so handoff/compaction take the cold path, adopt concurrent same-prefix agent commits past a stale fence, and cover the product checkpoint/resume path with integration tests. Co-Authored-By: David Cramer --- .../src/chat/conversations/projection.ts | 38 ++++- .../junior/src/chat/state/turn-session.ts | 36 +++- .../conversations/commit-messages.test.ts | 50 +++++- .../integration/local-agent-runner.test.ts | 161 +++++++++++++++++- .../integration/reporting-support.test.ts | 89 ++++++++-- 5 files changed, 347 insertions(+), 27 deletions(-) diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index 891e457044..6402ea6d7f 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -357,25 +357,47 @@ async function resolveCommitBase( if (live.committedSeq === args.base.committedSeq) { return args.base; } - // Global cursor advanced after the caller's base. Host-only facts (MCP - // connect, turn_context, native events) do that without changing agent - // history; concurrent agent-message writes must still fail closed. + // Global cursor advanced after the caller's base. That can be: + // - host-only facts (MCP connect, turn_context, tool_execution_started) + // - a concurrent checkpoint that already committed part of `nextLocalMessages` + // (turn_end persist racing a timeout/yield continuation) + // Divergent agent-history rewrites still fail closed. const currentEvents = await eventStore.loadCurrentHistory( args.conversationId, ); const current = projectConversationEvents(currentEvents); + const basePrefix = countMatchingPrefix( + args.base.messages, + current.messages, + ); if ( - current.messages.length !== args.base.messages.length || - current.seqs.length !== args.base.messageSeqs.length || - countMatchingPrefix(args.base.messages, current.messages) !== - args.base.messages.length || + basePrefix !== args.base.messages.length || args.base.messageSeqs.some((seq, index) => current.seqs[index] !== seq) ) { throwCommittedBoundaryChanged(args.conversationId); } + if (current.messages.length === args.base.messages.length) { + return { + ...args.base, + committedSeq: live.committedSeq, + }; + } + // Live agent history already extends the caller's base. Adopt it only when + // those extras are exactly the prefix of what this commit still wants. + const adoptedMessages = current.messages; + const adoptedPrefix = countMatchingPrefix( + adoptedMessages, + nextLocalMessages, + ); + if (adoptedPrefix !== adoptedMessages.length) { + throwCommittedBoundaryChanged(args.conversationId); + } return { - ...args.base, committedSeq: live.committedSeq, + historyVersion: live.historyVersion, + messageSeqs: current.seqs, + messages: adoptedMessages, + provenance: current.provenance, }; } diff --git a/packages/junior/src/chat/state/turn-session.ts b/packages/junior/src/chat/state/turn-session.ts index 905b99c10e..20f9d049c5 100644 --- a/packages/junior/src/chat/state/turn-session.ts +++ b/packages/junior/src/chat/state/turn-session.ts @@ -7,6 +7,7 @@ * `junior_conversation_events` so resumes can materialize the exact continuable * boundary without duplicating the event history. */ +import { isDeepStrictEqual } from "node:util"; import { THREAD_STATE_TTL_MS, type StateAdapter } from "chat"; import { actorSchema, @@ -321,6 +322,15 @@ function materializeAgentTurnSessionRecord( piProjection: ConversationMessageProjection, turnStartMessageIndex?: number, restoreVolatileContext = true, + /** + * Fence cursor for a newer replacement epoch. The stored record still points + * at the pre-handoff boundary; without this override it would look like a + * valid commit base for the replacement messages. + */ + replacementFence?: { + committedSeq: number; + historyVersion: number; + }, ): AgentTurnSessionRecord { const piMessages = restoreVolatileContext && @@ -339,9 +349,12 @@ function materializeAgentTurnSessionRecord( updatedAtMs: stored.updatedAtMs, piMessages, piMessageProvenance: piProjection.provenance, - committedSeq: stored.committedSeq, - historyVersion: stored.historyVersion ?? 0, - messageSeqs: stored.messageSeqs ?? piProjection.seqs, + committedSeq: replacementFence?.committedSeq ?? stored.committedSeq, + historyVersion: + replacementFence?.historyVersion ?? stored.historyVersion ?? 0, + messageSeqs: replacementFence + ? piProjection.seqs + : (stored.messageSeqs ?? piProjection.seqs), actors: stored.actors ?? instructionActors(piProjection.provenance), cumulativeDurationMs: stored.cumulativeDurationMs, ...(stored.destination ? { destination: stored.destination } : {}), @@ -470,6 +483,12 @@ async function materializeStoredAgentTurnSessionRecord( !followsReplacement && (parsed.historyVersion === undefined || parsed.historyVersion === currentHistoryVersion), + followsReplacement + ? { + committedSeq: currentHistory.at(-1)?.seq ?? -1, + historyVersion: currentHistoryVersion, + } + : undefined, ); } @@ -774,11 +793,20 @@ export async function upsertAgentTurnSessionRecord(args: { const durableExistingMessages = existingRecord ? stripRuntimeTurnContext(existingRecord.piMessages) : undefined; + const nextDurableMessages = stripRuntimeTurnContext(args.piMessages); + // Only fence on the prior session base when this write is a true prefix + // extension. Compaction/handoff replacements intentionally rewrite history + // and must take the cold path so commitMessages can adopt the new epoch. const commitBase = existingRecord && durableExistingMessages && existingRecord.messageSeqs.length === durableExistingMessages.length && - existingRecord.piMessageProvenance.length === durableExistingMessages.length + existingRecord.piMessageProvenance.length === + durableExistingMessages.length && + nextDurableMessages.length >= durableExistingMessages.length && + durableExistingMessages.every((message, index) => + isDeepStrictEqual(message, nextDurableMessages[index]), + ) ? { committedSeq: existingRecord.committedSeq, historyVersion: existingRecord.historyVersion, diff --git a/packages/junior/tests/component/conversations/commit-messages.test.ts b/packages/junior/tests/component/conversations/commit-messages.test.ts index a24a4a1bf9..9e351b50ac 100644 --- a/packages/junior/tests/component/conversations/commit-messages.test.ts +++ b/packages/junior/tests/component/conversations/commit-messages.test.ts @@ -85,7 +85,7 @@ describe("commitMessages cursor fencing", () => { } }); - it("rejects a stale base cursor before writing", async () => { + it("rejects a stale base when live history diverges from the next commit", async () => { const fixture = await createLocalJuniorSqlFixture(); try { await migrateSchema(fixture.sql); @@ -96,9 +96,11 @@ describe("commitMessages cursor fencing", () => { executor: fixture.sql, }); + // Concurrent checkpoint wrote a different assistant reply than this + // caller still wants to commit. Same-prefix races may adopt; rewrites must not. await commitMessages({ conversationId: CONVERSATION_ID, - messages: [user("hello", 1), assistant("hi", 2)], + messages: [user("hello", 1), assistant("other path", 2)], executor: fixture.sql, }); @@ -172,4 +174,48 @@ describe("commitMessages cursor fencing", () => { await fixture.close(); } }); + + it("adopts a concurrent same-prefix agent commit past a stale fence base", async () => { + const fixture = await createLocalJuniorSqlFixture(); + try { + await migrateSchema(fixture.sql); + + const first = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1)], + executor: fixture.sql, + }); + + // A racing turn_end checkpoint already wrote the tool boundary while the + // session record still points at the pre-tool fence. + const concurrent = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [user("hello", 1), assistant("working", 2), user("done", 3)], + executor: fixture.sql, + }); + + const second = await commitMessages({ + conversationId: CONVERSATION_ID, + messages: [ + user("hello", 1), + assistant("working", 2), + user("done", 3), + ], + base: { + committedSeq: first.committedSeq, + historyVersion: first.historyVersion, + messageSeqs: first.messageSeqs, + messages: first.messages, + provenance: first.provenance, + }, + executor: fixture.sql, + }); + + expect(second.messageSeqs).toEqual(concurrent.messageSeqs); + expect(second.committedSeq).toBe(concurrent.committedSeq); + expect(second.historyVersion).toBe(concurrent.historyVersion); + } finally { + await fixture.close(); + } + }); }); diff --git a/packages/junior/tests/integration/local-agent-runner.test.ts b/packages/junior/tests/integration/local-agent-runner.test.ts index ad0ae18a56..006727717c 100644 --- a/packages/junior/tests/integration/local-agent-runner.test.ts +++ b/packages/junior/tests/integration/local-agent-runner.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vitest"; import type { AgentRunResult } from "@/chat/services/turn-result"; import { getAssistantReplyText } from "@/chat/services/assistant-reply"; import { + createLocalSource, defineJuniorPlugin, type PluginRunContext, } from "@sentry/junior-plugin-api"; @@ -17,7 +18,10 @@ import { import type { PiMessage } from "@/chat/pi/messages"; import type { AssistantMessage } from "@earendil-works/pi-ai"; import type { AgentRunner } from "@/chat/runtime/agent-runner"; -import { persistRunningSessionRecord } from "@/chat/services/turn-session-record"; +import { + completeDeliveredTurn, + persistRunningSessionRecord, +} from "@/chat/services/turn-session-record"; import { getPersistedSandboxState, getPersistedThreadState, @@ -25,7 +29,12 @@ import { import { commitMessages, loadProjection, + recordMcpProviderConnected, } from "@/chat/conversations/projection"; +import { + getAgentTurnSessionRecord, + listAgentTurnSessionSummariesForConversation, +} from "@/chat/state/turn-session"; import { coerceThreadConversationState } from "@/chat/state/conversation"; import { hydrateConversationMessages } from "@/chat/conversations/messages"; import { coerceThreadArtifactsState } from "@/chat/state/artifacts"; @@ -1140,6 +1149,156 @@ describe("local agent runner", () => { expect(contexts[0]?.piMessages).toEqual([generatedMessages[0]]); }); + it("keeps mid-turn checkpoints durable after host-only MCP events and resumes from them", async () => { + const conversationId = normalizeLocalConversationId({ + alias: "checkpoint-host-only", + cwd: "/tmp/local-agent-runner-checkpoint-host-only", + }); + expect(conversationId).toBeDefined(); + + const destination = { + platform: "local" as const, + conversationId: conversationId!, + }; + const source = createLocalSource(destination.conversationId); + const userMessage = userPiMessage("connect github and continue", 1); + const assistantPartial = assistantPiMessage("connecting github", 2); + const toolResult = { + role: "toolResult", + toolCallId: "tool-call-github", + toolName: "searchMcpTools", + content: [{ type: "text", text: "github tools ready" }], + isError: false, + timestamp: 3, + } as PiMessage; + const checkpointMessages: PiMessage[] = [ + userMessage, + assistantPartial, + toolResult, + ]; + const finalAssistant = assistantPiMessage("github is connected", 4); + const finalMessages: PiMessage[] = [...checkpointMessages, finalAssistant]; + const sessionId = "turn-checkpoint-host-only"; + + // Product path: checkpoint a running turn, let a host-only MCP connect + // advance the global cursor, checkpoint again, complete, then prove the + // next local turn still loads the durable history. + expect( + await persistRunningSessionRecord({ + modelId: "fake-local-agent", + conversationId: conversationId!, + destination, + source, + sessionId, + sliceId: 1, + messages: checkpointMessages, + surface: "internal", + turnStartMessageIndex: 0, + }), + ).toBe(true); + + await recordMcpProviderConnected({ + conversationId: conversationId!, + provider: "github", + }); + + expect( + await persistRunningSessionRecord({ + modelId: "fake-local-agent", + conversationId: conversationId!, + destination, + source, + sessionId, + sliceId: 1, + messages: checkpointMessages, + surface: "internal", + turnStartMessageIndex: 0, + }), + ).toBe(true); + + const running = await getAgentTurnSessionRecord(conversationId!, sessionId); + expect(running).toMatchObject({ + state: "running", + messageSeqs: [0, 1, 2], + piMessages: checkpointMessages, + }); + expect(running?.committedSeq).toBeGreaterThanOrEqual(3); + + const historyAfterHostOnly = await getConversationEventStore().loadHistory( + conversationId!, + ); + expect( + historyAfterHostOnly.map((event) => [event.seq, event.data.type]), + ).toEqual([ + [0, "user_message"], + [1, "assistant_message"], + [2, "tool_result"], + [3, "mcp_provider_connected"], + ]); + + await completeDeliveredTurn({ + conversationId: conversationId!, + destination, + source, + sessionId, + sliceId: 1, + messages: finalMessages, + modelId: "fake-local-agent", + surface: "internal", + turnStartMessageIndex: 0, + }); + + expect(await loadProjection({ conversationId: conversationId! })).toEqual( + finalMessages, + ); + + const completed = await getAgentTurnSessionRecord( + conversationId!, + sessionId, + ); + expect(completed).toMatchObject({ + state: "completed", + messageSeqs: [0, 1, 2, 4], + piMessages: finalMessages, + }); + expect(completed?.committedSeq).toBeGreaterThanOrEqual(4); + + const summaries = await listAgentTurnSessionSummariesForConversation( + conversationId!, + ); + expect(summaries).toEqual([ + expect.objectContaining({ + conversationId: conversationId!, + sessionId, + state: "completed", + }), + ]); + expect(summaries[0]).not.toHaveProperty("messageSeqs"); + expect(summaries[0]).not.toHaveProperty("committedSeq"); + expect(summaries[0]).not.toHaveProperty("historyVersion"); + + const followUpContexts: FlatAgentRunRequest[] = []; + await runLocalAgentTurn( + { + conversationId: conversationId!, + message: "what did you connect?", + }, + { + deliverReply: async () => undefined, + agentRunner: { + run: async (request) => { + followUpContexts.push(flattenAgentRunRequestForTest(request)); + return completedAgentRun(successReply("github")); + }, + }, + }, + ); + + // Local follow-ups load durable Pi history with trailing assistant output + // trimmed so the new user turn can continue from a safe boundary. + expect(followUpContexts[0]?.piMessages).toEqual(checkpointMessages); + }); + it("keeps the delivered local reply successful when a background task fails", async () => { const conversationId = normalizeLocalConversationId({ alias: "background-task-failure", diff --git a/packages/junior/tests/integration/reporting-support.test.ts b/packages/junior/tests/integration/reporting-support.test.ts index dd0ee151a8..40c756b403 100644 --- a/packages/junior/tests/integration/reporting-support.test.ts +++ b/packages/junior/tests/integration/reporting-support.test.ts @@ -1,4 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createSlackSource } from "@sentry/junior-plugin-api"; const ORIGINAL_ENV = { ...process.env }; const TEST_DATABASE_URL = ORIGINAL_ENV.DATABASE_URL; @@ -29,26 +30,90 @@ describe("reporting support", () => { }); it("indexes only the latest safe turn-session summary", async () => { - const { listAgentTurnSessionSummaries, upsertAgentTurnSessionRecord } = + const { getConversationStore } = await import("@/chat/db"); + const { completeDeliveredTurn, persistRunningSessionRecord } = + await import("@/chat/services/turn-session-record"); + const { listAgentTurnSessionSummaries } = await import("@/chat/state/turn-session"); - const conversationId = "slack:C-reporting-support:summary-index"; + const conversationId = "slack:CREPORTINGSUPPORT:summary-index"; + const destination = { + platform: "slack" as const, + teamId: "TREPORTINGSUPPORT", + channelId: "CREPORTINGSUPPORT", + }; + const source = createSlackSource({ + teamId: "TREPORTINGSUPPORT", + channelId: "CREPORTINGSUPPORT", + messageTs: "1700000000.100", + threadTs: "1700000000.100", + visibility: "public", + }); + const userMessage = { + role: "user" as const, + content: [{ type: "text" as const, text: "summarize this" }], + timestamp: 1, + }; + const assistantMessage = { + role: "assistant" as const, + content: [{ type: "text" as const, text: "done" }], + api: "openai-responses", + provider: "openai", + model: "test/model", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: "stop" as const, + timestamp: 2, + }; - await upsertAgentTurnSessionRecord({ - modelId: "test/model", + // Product session writes also mirror execution into the conversation store. + await getConversationStore().recordActivity({ conversationId, - sessionId: "reporting-support-turn", - sliceId: 1, - state: "running", - piMessages: [], + channelName: "reporting-support-summary", + destination, + nowMs: Date.now(), + source: "slack", + title: "Reporting support summary", + visibility: "public", }); - await upsertAgentTurnSessionRecord({ + + expect( + await persistRunningSessionRecord({ + modelId: "test/model", + conversationId, + destination, + destinationVisibility: "public", + source, + sessionId: "reporting-support-turn", + sliceId: 1, + messages: [userMessage], + surface: "slack", + loadedSkillNames: ["triage"], + }), + ).toBe(true); + + await completeDeliveredTurn({ modelId: "test/model", conversationId, + destination, + destinationVisibility: "public", + source, sessionId: "reporting-support-turn", sliceId: 2, - state: "completed", - piMessages: [], - cumulativeDurationMs: 1_200, + messages: [userMessage, assistantMessage], + surface: "slack", + durationMs: 1_200, errorMessage: "provider failed with sensitive details", loadedSkillNames: ["triage"], }); From ab58929f483c981a0486c56d0726ac9386b1b99c Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:34:43 +0000 Subject: [PATCH 7/8] fix(conversations): Reject stale session writes after history races History fences can adopt concurrent same-prefix commits, but session metadata must not regress. Re-check the live turn session after commit and refuse delayed running checkpoints that would overwrite awaiting_resume/completed/failed state. Co-Authored-By: David Cramer --- .../junior/src/chat/state/turn-session.ts | 58 +++++++++++++- .../services/turn-session-record.test.ts | 78 +++++++++++++++++++ 2 files changed, 135 insertions(+), 1 deletion(-) diff --git a/packages/junior/src/chat/state/turn-session.ts b/packages/junior/src/chat/state/turn-session.ts index 20f9d049c5..f6f996b692 100644 --- a/packages/junior/src/chat/state/turn-session.ts +++ b/packages/junior/src/chat/state/turn-session.ts @@ -61,6 +61,21 @@ export type AgentTurnSessionStatus = | "failed" | "abandoned"; +/** Lifecycle rank for concurrent session writes. Higher ranks must not regress. */ +function agentTurnSessionStateRank(state: AgentTurnSessionStatus): number { + switch (state) { + case "running": + return 0; + case "awaiting_resume": + return 1; + case "failed": + case "abandoned": + return 2; + case "completed": + return 3; + } +} + export type AgentTurnSurface = "slack" | "api" | "scheduler" | "internal"; export type AgentTurnResumeReason = "timeout" | "auth" | "yield" | "retry"; @@ -865,6 +880,47 @@ export async function upsertAgentTurnSessionRecord(args: { ? undefined : commit.messageSeqs.filter((seq) => seq <= turnStartSeq).length); + // History commits can adopt concurrent same-prefix writes. Re-check the + // session record before overwriting metadata so a delayed running checkpoint + // cannot clobber awaiting_resume/completed/failed that landed meanwhile. + const liveAfterCommit = await getStoredAgentTurnSessionRecord( + args.conversationId, + args.sessionId, + ); + const expectedVersion = storedRecord?.version ?? existingRecord?.version; + if ( + liveAfterCommit && + expectedVersion !== undefined && + liveAfterCommit.version !== expectedVersion + ) { + const nextRank = agentTurnSessionStateRank(args.state); + const liveRank = agentTurnSessionStateRank(liveAfterCommit.state); + const regressesLifecycle = + nextRank < liveRank || + (nextRank === liveRank && args.sliceId < liveAfterCommit.sliceId); + if (regressesLifecycle) { + // Completed delivery retries are idempotent once the terminal record exists. + if (args.state === "completed" && liveAfterCommit.state === "completed") { + const liveRecord = await getAgentTurnSessionRecord( + args.conversationId, + args.sessionId, + ); + if (liveRecord) { + return liveRecord; + } + } + throw new Error( + `Turn session ${args.sessionId} changed before its session write ` + + `(${liveAfterCommit.state}@v${liveAfterCommit.version}/slice ${liveAfterCommit.sliceId} vs ` + + `${args.state}/slice ${args.sliceId})`, + ); + } + } + const previousVersion = + liveAfterCommit?.version ?? + storedRecord?.version ?? + existingRecord?.version; + return await setStoredRecord({ conversationStore: args.conversationStore, destinationVisibility: args.destinationVisibility, @@ -893,7 +949,7 @@ export async function upsertAgentTurnSessionRecord(args: { ...(retainedRuntimeContext ? { runtimeContext: retainedRuntimeContext } : {}), - previousVersion: storedRecord?.version ?? existingRecord?.version, + previousVersion, cumulativeDurationMs: args.cumulativeDurationMs ?? existingRecord?.cumulativeDurationMs ?? 0, ...(args.cumulativeUsage diff --git a/packages/junior/tests/component/services/turn-session-record.test.ts b/packages/junior/tests/component/services/turn-session-record.test.ts index ea899d79a9..ac200b0d01 100644 --- a/packages/junior/tests/component/services/turn-session-record.test.ts +++ b/packages/junior/tests/component/services/turn-session-record.test.ts @@ -1174,6 +1174,84 @@ describe("persistAuthPauseSessionRecord", () => { ).resolves.toBe(false); }); + it("rejects a delayed running checkpoint after awaiting_resume wins the race", async () => { + const { persistContinuationSessionRecord, persistRunningSessionRecord } = + await import("@/chat/services/turn-session-record"); + const { + getAgentTurnSessionRecord, + upsertAgentTurnSessionRecord, + } = await import("@/chat/state/turn-session"); + + const request = userMessage("help me"); + const toolBoundary = assistantMessage("working", Date.now() + 1); + const toolResult = userMessage("tool done"); + + // Mid-turn checkpoint at the user prompt. + expect( + await persistRunningSessionRecord({ + modelId: "test-model", + conversationId: "conversation-stale-running", + sessionId: "turn-stale-running", + sliceId: 1, + messages: [request], + }), + ).toBe(true); + + const staleRunningBase = await getAgentTurnSessionRecord( + "conversation-stale-running", + "turn-stale-running", + ); + expect(staleRunningBase?.state).toBe("running"); + + // Timeout path already committed the tool boundary and parked the turn. + await persistContinuationSessionRecord({ + resumeReason: "timeout", + modelId: "test-model", + conversationId: "conversation-stale-running", + sessionId: "turn-stale-running", + currentSliceId: 1, + messages: [request, toolBoundary, toolResult], + errorMessage: "provider stream interrupted", + }); + + await expect( + getAgentTurnSessionRecord( + "conversation-stale-running", + "turn-stale-running", + ), + ).resolves.toMatchObject({ + state: "awaiting_resume", + sliceId: 2, + resumeReason: "timeout", + }); + + // A delayed running checkpoint still holds the pre-timeout session base. + // History adoption may succeed, but session metadata must not regress. + await expect( + upsertAgentTurnSessionRecord({ + modelId: "test-model", + conversationId: "conversation-stale-running", + sessionId: "turn-stale-running", + sliceId: 1, + state: "running", + piMessages: [request, toolBoundary, toolResult], + existing: staleRunningBase!, + }), + ).rejects.toThrow(/changed before its session write/); + + await expect( + getAgentTurnSessionRecord( + "conversation-stale-running", + "turn-stale-running", + ), + ).resolves.toMatchObject({ + state: "awaiting_resume", + sliceId: 2, + resumeReason: "timeout", + piMessages: [request, toolBoundary, toolResult], + }); + }); + it("promotes the latest running record when timeout capture has no messages", async () => { const { persistContinuationSessionRecord, persistRunningSessionRecord } = await import("@/chat/services/turn-session-record"); From 487ab6677444a0647ab291dd8807340d531b2dbf Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:44:05 +0000 Subject: [PATCH 8/8] fix(conversations): Fence session writes on caller-owned versions expectedVersion must prefer the stale caller base over a fresh store read at upsert entry; otherwise delayed running checkpoints skip the lifecycle guard and can still overwrite awaiting_resume. Co-Authored-By: David Cramer --- packages/junior/src/chat/state/turn-session.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/junior/src/chat/state/turn-session.ts b/packages/junior/src/chat/state/turn-session.ts index f6f996b692..7c032675f1 100644 --- a/packages/junior/src/chat/state/turn-session.ts +++ b/packages/junior/src/chat/state/turn-session.ts @@ -883,11 +883,14 @@ export async function upsertAgentTurnSessionRecord(args: { // History commits can adopt concurrent same-prefix writes. Re-check the // session record before overwriting metadata so a delayed running checkpoint // cannot clobber awaiting_resume/completed/failed that landed meanwhile. + // Prefer the caller-owned base version when present: a fresh store read at + // upsert entry would mask a stale `existing` and skip this guard entirely. const liveAfterCommit = await getStoredAgentTurnSessionRecord( args.conversationId, args.sessionId, ); - const expectedVersion = storedRecord?.version ?? existingRecord?.version; + const expectedVersion = + args.existing?.version ?? storedRecord?.version ?? existingRecord?.version; if ( liveAfterCommit && expectedVersion !== undefined &&