diff --git a/src/main/native-chat/agent-session-journal/journal-damage.test.ts b/src/main/native-chat/agent-session-journal/journal-damage.test.ts index 8b59940a6a19..168e74484b69 100644 --- a/src/main/native-chat/agent-session-journal/journal-damage.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-damage.test.ts @@ -206,8 +206,8 @@ describe('a row the reader would reject', () => { await expect( journal.replaceEpochItems('legacy_import', 1, [ - { identity: item(5), body: body('replacement') }, - { identity: item(6), body: blankCallId } + { identity: item(5), body: body('replacement'), ownerFence: 1 }, + { identity: item(6), body: blankCallId, ownerFence: 1 } ]) ).rejects.toMatchObject({ code: 'journal_row_rejected' }) diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts index 36247e2571ff..c64027b94555 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts @@ -96,7 +96,7 @@ describe('journal epoch replacement', () => { const published: JournalLoad[] = [] replace({ - items: [{ identity: item(1), body: { kind: 'status', text: 'republished' } }], + items: [{ identity: item(1), body: { kind: 'status', text: 'republished' }, ownerFence: 1 }], onPublished: (loaded) => published.push(loaded) }) @@ -112,7 +112,7 @@ describe('journal epoch replacement', () => { 'a replace', (journal: AgentSessionJournal) => journal.replaceEpochItems('legacy_import', 1, [ - { identity: item(9), body: { kind: 'status', text: 'republished' } } + { identity: item(9), body: { kind: 'status', text: 'republished' }, ownerFence: 1 } ]) ], ['a rollover', (journal: AgentSessionJournal) => journal.rollEpoch('handle_forked', 1)] diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts index 67927b401588..17b9f5e3b956 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts @@ -38,6 +38,9 @@ export type JournalReplacementItem = AgentJournalProducerLinkage & { /** Absent for history rebuilt from a source that never stated one: derived from position, as * a legacy row's is, and then written down. */ turnScope?: AgentJournalTurnScope + /** Carried from the epoch it came from (`JournalItemRow.ownerFence`), never the replacing + * writer's fence: replacing history does not make it the current generation's work. */ + ownerFence: number } export function replaceJournalEpoch(input: { @@ -71,6 +74,7 @@ export function replaceJournalEpoch(input: { body: item.body, seq: state.lastSequence + 1, fence: input.fence, + ownerFence: item.ownerFence, ts: item.observedAt ?? input.now(), linkage: item, turnScope: item.turnScope ?? state.derivedTurnScope.scopeFor(item.body) diff --git a/src/main/native-chat/agent-session-journal/journal-item-fold.ts b/src/main/native-chat/agent-session-journal/journal-item-fold.ts index 30554ab29937..c45d8975b25d 100644 --- a/src/main/native-chat/agent-session-journal/journal-item-fold.ts +++ b/src/main/native-chat/agent-session-journal/journal-item-fold.ts @@ -28,8 +28,9 @@ export function upsertJournalItem( itemId: string, revision: number, next: AgentJournalRenderItem, - /** The creating write's fence: the generation a running turn belongs to. */ - fence: number + /** The write's fence, and the provenance it states (`JournalItemRow.ownerFence`), which only + * creation reads: a revision keeps the item's. */ + fences: { fence: number; ownerFence?: number } ): void { const tombstoned = state.tombstones.get(itemId) if (tombstoned !== undefined && revision <= tombstoned) { @@ -42,7 +43,7 @@ export function upsertJournalItem( state.derivedTurnScope.observe(itemId, isRootAgentJournalItem(next), existing?.body, next.body) if (!existing) { state.items.set(itemId, next) - state.itemFences.set(itemId, fence) + state.itemFences.set(itemId, fences.ownerFence ?? fences.fence) state.tombstones.delete(itemId) return } diff --git a/src/main/native-chat/agent-session-journal/journal-item-provenance.test.ts b/src/main/native-chat/agent-session-journal/journal-item-provenance.test.ts new file mode 100644 index 000000000000..8e94ef30ac9a --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-item-provenance.test.ts @@ -0,0 +1,128 @@ +// Execution provenance (`JournalItemRow.ownerFence`): the generation whose execution produced an +// item, kept apart from each row's writer fence. Creation states it; every revision keeps it. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemBody, + type AgentJournalItemIdentity, + type AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { retainedRowReplacement } from '../agent-session-wire/structured-rewind-retained-host-rows' +import type { JournalHostDatabase } from './journal-host-database' +import { + createTrackedJournalOpener, + liveTestJournalRows, + openTestJournalHostDatabase +} from './journal-host-database-test-support' +import type { AgentSessionJournal } from './journal-store' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: codexProviderHandle('thread-1') +} +const SCOPE = AGENT_JOURNAL_THREAD_SCOPE +const TURN: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-1', + ordinal: 1 +} +const TURN_ID = agentJournalItemKey(TURN) +const RUNNING: AgentJournalItemBody = { kind: 'turn', turnId: 'turn-1', state: 'running' } +const SETTLED: AgentJournalItemBody = { kind: 'turn', turnId: 'turn-1', state: 'interrupted' } + +let root: string +let database: JournalHostDatabase +const journals = createTrackedJournalOpener() + +function item(ordinal: number): AgentJournalItemIdentity { + return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal } +} + +type StoredItemWrite = { kind?: string; ownerFence?: number; mutations?: StoredItemWrite[] } + +/** Each stored item write's stated provenance, in write order, a lifecycle batch's included. */ +function storedOwnerFences(): (number | undefined)[] { + return liveTestJournalRows(database.db, IDENTITY.sessionId).flatMap((row) => { + const parsed: StoredItemWrite = JSON.parse(row.rowJson) + const writes = parsed.kind === 'lifecycle-batch' ? (parsed.mutations ?? []) : [parsed] + return writes.flatMap((write) => (write.kind === 'item' ? [write.ownerFence] : [])) + }) +} + +async function open(): Promise { + return journals.open({ identity: IDENTITY, stateDirectory: root }) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-provenance-')) + database = openTestJournalHostDatabase(root) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +describe('execution provenance', () => { + it("records the creating observation's owner, every revision keeps it, and a reopen reads it", async () => { + const journal = await open() + await journal.appendItem(TURN, RUNNING, { fence: 4, ownerFence: 3, turnScope: SCOPE }) + // A newer run's provider revising it (a start's context-usage report on the newest saved turn). + await journal.appendItem(TURN, RUNNING, { fence: 5, ownerFence: 5, turnScope: SCOPE }) + // Host bookkeeping under a later writer fence states no owner of its own. + await journal.appendItem(TURN, SETTLED, { fence: 5, turnScope: SCOPE }) + // A first write with no stated owner is its writer generation's. + await journal.appendItem( + item(2), + { kind: 'status', text: 'note' }, + { fence: 6, turnScope: SCOPE } + ) + + expect(journal.itemFence(TURN_ID)).toBe(3) + expect(journal.itemFence(agentJournalItemKey(item(2)))).toBe(6) + // Every row states it: none leaves the reader to guess from the writer fence. + expect(storedOwnerFences()).toEqual([3, 3, 3, 6]) + await journals.closeAll() + expect((await open()).itemFence(TURN_ID)).toBe(3) + }) + + it('a lifecycle batch states it on each item: the batch owner, else the kept value', async () => { + const journal = await open() + await journal.appendItem(TURN, RUNNING, { fence: 2, ownerFence: 2, turnScope: SCOPE }) + // A provider batch at the owner's fence creates one item. + await journal.appendLifecycleBatch({ + settlementId: 'provider', + fence: 3, + ownerFence: 3, + mutations: [{ kind: 'item', identity: item(2), body: RUNNING, turnScope: SCOPE }] + }) + // A settlement at a later fence revises the first: it keeps that item's owner. + await journal.appendLifecycleBatch({ + settlementId: 'settlement', + fence: 4, + recovered: true, + mutations: [{ kind: 'item', identity: TURN, body: SETTLED, turnScope: SCOPE }] + }) + + expect(journal.itemFence(TURN_ID)).toBe(2) + expect(journal.itemFence(agentJournalItemKey(item(2)))).toBe(3) + expect(storedOwnerFences()).toEqual([2, 3, 2]) + }) + + it("a rewind's retained row carries the old fold's provenance; an older record's is never current", () => { + const row = { itemId: TURN_ID, body: RUNNING, observedAt: 1, ownerFence: 2 } + expect(retainedRowReplacement(row).ownerFence).toBe(2) + const { ownerFence: _stated, ...older } = row + expect(retainedRowReplacement(older).ownerFence).toBe(0) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-item-provenance.ts b/src/main/native-chat/agent-session-journal/journal-item-provenance.ts new file mode 100644 index 000000000000..afb1df346d1e --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-item-provenance.ts @@ -0,0 +1,16 @@ +// Execution provenance: the generation whose execution produced an item (`JournalItemRow.ownerFence`), +// kept apart from each row's writer fence. The write that creates an item states it; every revision +// keeps it. + +import type { JournalReducerState } from './journal-reducer' + +/** An existing item keeps its owner whoever revises it: a newer run's start revises the newest saved + * turn. Creation states the writer's claim (an observation, carried history), else its fence. */ +export function journalItemOwnerFence( + state: Pick, + resolvedItemId: string, + write: { fence: number; ownerFence?: number } +): number { + const kept = state.items.has(resolvedItemId) ? state.itemFences.get(resolvedItemId) : undefined + return kept ?? write.ownerFence ?? write.fence +} diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts index f4cad1f9edc0..9da96daf490c 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts @@ -245,6 +245,10 @@ describe('claude import', () => { uuid: 'e3a5b7c9-2d13-4f88-94ce-6b0d8f2a4e31' }) ]) + // Imported at fence 1, yet no generation's work: provenance 0, so nothing in it is current. + expect(journal.snapshot().items.map((entry) => journal.itemFence(entry.itemId))).toEqual([ + 0, 0, 0, 0 + ]) }) it('stays aligned when the decoder drops lines the tracker still walks', async () => { diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts index 1e6171e94aa9..7c8b67e440a2 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -138,7 +138,9 @@ export async function prepareLegacyTranscriptImport(input: { replacement.push({ identity, body: legacyItemBody(message, limits), - observedAt: message.timestamp ?? undefined + observedAt: message.timestamp ?? undefined, + // Imported history is no generation's work: never current, whatever it says still runs. + ownerFence: 0 }) } return { ok: true, items: replacement } diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts index f942f0185d05..f906fadfbed4 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts @@ -75,12 +75,22 @@ export class JournalLifecycleBatchAppender { } private planMutations( - input: Pick, - mutations: readonly JournalLifecycleMutationInput[] + input: Pick, + stated: readonly JournalLifecycleMutationInput[] ): ((seq: number, ts: number) => JournalRow)[] { if (this.wasApplied(input.settlementId)) { return [] } + // A provider's batch states its generation on every item it creates. + const { ownerFence } = input + const mutations = + ownerFence === undefined + ? stated + : stated.map((mutation) => + mutation.kind === 'item' && mutation.ownerFence === undefined + ? { ...mutation, ownerFence } + : mutation + ) const current = this.deps.state() const options = { ...input, epoch: current.epoch } // Every row in this transaction must advance past the rows planned before it. @@ -141,6 +151,7 @@ export class JournalLifecycleBatchAppender { ...only.linkage, turnScope: only.turnScope, fence: input.fence, + ownerFence: only.ownerFence, recovered: input.recovered }, revisions diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test-fixture.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test-fixture.ts index 509850c78841..cc94120fe538 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test-fixture.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test-fixture.ts @@ -28,8 +28,8 @@ export const LEGACY_MIXED_PARTITIONS = [ [['', 30]], [['', 96]], [ - [':1/2', 117], - [':2/2', 5] + [':1/2', 116], + [':2/2', 6] ], [['', 70]], [ @@ -37,9 +37,9 @@ export const LEGACY_MIXED_PARTITIONS = [ [':2/2', 34] ], [ - [':1/3', 66], - [':2/3', 66], - [':3/3', 32] + [':1/3', 65], + [':2/3', 65], + [':3/3', 34] ], [ [':1/4', 62], diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test.ts index 25b6f5b96531..1943f30dd673 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.test.ts @@ -57,6 +57,7 @@ function previousProbe(settlementId: string, mutations: readonly JournalLifecycl ? { ...mutation, body: { ...mutation.body, outcome: 'cancellation' } } : mutation, journalLifecycleMutationItemId(mutation), + Number.MAX_SAFE_INTEGER, Number.MAX_SAFE_INTEGER ) ) diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts index 4e01628c6d58..525605b9ec3b 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts @@ -136,6 +136,8 @@ function toLifecycleMutationRow(mutation: JournalLifecycleMutationInput): Journa return journalLifecycleMutationRow( sizedAsStopped(mutation), journalLifecycleMutationItemId(mutation), + Number.MAX_SAFE_INTEGER, + // The widest provenance any row can state, so the chunk built from it still fits. Number.MAX_SAFE_INTEGER ) } diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-mutation.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-mutation.ts index 1340b99f3106..9b939c267436 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-mutation.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-mutation.ts @@ -26,6 +26,8 @@ export type JournalLifecycleMutationInput = JournalItemAddress & linkage?: AgentJournalProducerLinkage /** Which turn the row belongs to. Kept from the write that creates the row. */ turnScope: AgentJournalTurnScope + /** The writer's provenance claim (`journalItemOwnerFence`); absent for host bookkeeping. */ + ownerFence?: number } | { kind: 'tombstone' } ) @@ -61,7 +63,9 @@ export function journalLifecycleItemMutation( export function journalLifecycleMutationRow( mutation: JournalLifecycleMutationInput, itemId: string, - revision: number + revision: number, + /** What an item mutation states (`journalItemOwnerFence`). */ + ownerFence: number ): JournalLifecycleMutation { return mutation.kind === 'item' ? { @@ -70,6 +74,7 @@ export function journalLifecycleMutationRow( revision, body: mutation.body, turnScope: mutation.turnScope, + ownerFence, ...agentJournalLinkageFields(mutation.linkage) } : { kind: 'tombstone', itemId, revision } diff --git a/src/main/native-chat/agent-session-journal/journal-provider-activity.ts b/src/main/native-chat/agent-session-journal/journal-provider-activity.ts index 09a965b624ef..5a7c29ef361b 100644 --- a/src/main/native-chat/agent-session-journal/journal-provider-activity.ts +++ b/src/main/native-chat/agent-session-journal/journal-provider-activity.ts @@ -4,13 +4,15 @@ import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { JournalReducerState } from './journal-reducer' import type { JournalRow } from './journal-row-schema' -/** Saved output extends a turn; client actions and recovery writes do not. */ +/** Saved output extends a turn; client actions and recovery writes do not. Keyed by the fence of + * the generation that produced the item (`AgentSessionJournal.itemFence`), as a verdict reads it. */ export function observeJournalProviderActivity( state: Pick, row: JournalRow, itemId: string, body: AgentJournalItemBody, - savedAt: number + savedAt: number, + fence: number ): void { if (row.recovered || !isProviderActivity(body)) { return @@ -25,10 +27,7 @@ export function observeJournalProviderActivity( ) { return } - state.providerActivityAt.set( - row.fence, - Math.max(state.providerActivityAt.get(row.fence) ?? 0, savedAt) - ) + state.providerActivityAt.set(fence, Math.max(state.providerActivityAt.get(fence) ?? 0, savedAt)) } function isProviderActivity(body: AgentJournalItemBody): boolean { diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 4acb175e82ab..b929ebb24510 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -51,7 +51,8 @@ export type JournalReducerState = { oldestSequence: number highestFence: number items: Map - /** Fence of the writer that created each item: the generation a running turn belongs to. */ + /** Each item's execution provenance (`JournalItemRow.ownerFence`): the generation whose work it + * is, which decides whether it is current. */ itemFences: Map /** Latest saved output per owner, rebuilt during the existing row fold. */ providerActivityAt: Map @@ -112,15 +113,16 @@ export function applyJournalRow( return } const itemId = resolveJournalItemId(state, row.itemId, row.body) - observeJournalProviderActivity(state, row, row.itemId, row.body, savedAt) acceptSubmissionFromProviderItem(state, row.itemId, itemId, row) upsertJournalItem( state, itemId, row.revision, journalRenderItem(itemId, row.revision, row.body, row, statedOrDerivedTurnScope(state, row)), - row.fence + { fence: row.fence, ownerFence: row.ownerFence } ) + const fence = state.itemFences.get(itemId) ?? row.fence + observeJournalProviderActivity(state, row, row.itemId, row.body, savedAt, fence) return } if (isJournalStopOrResumeRow(row)) { @@ -144,13 +146,17 @@ export function applyJournalRow( continue } const { revision, body } = mutation - observeJournalProviderActivity(state, row, mutation.itemId, body, savedAt) const itemId = resolveJournalItemId(state, mutation.itemId, body) acceptSubmissionFromProviderItem(state, mutation.itemId, itemId, row) const producer = journalBatchMutationProducer(row, mutation) const scope = statedOrDerivedTurnScope(state, mutation) const item = journalRenderItem(itemId, revision, body, row, scope, producer, sequenceIndex) - upsertJournalItem(state, itemId, revision, item, row.fence) + upsertJournalItem(state, itemId, revision, item, { + fence: row.fence, + ownerFence: mutation.ownerFence + }) + const fence = state.itemFences.get(itemId) ?? row.fence + observeJournalProviderActivity(state, row, mutation.itemId, body, savedAt, fence) } else { removeJournalItem(state, resolveItemId(state, mutation.itemId), mutation.revision) } diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts index ea08945a4d1d..33c4e86cf49f 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-builders.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -13,6 +13,7 @@ import { agentSessionJournalProviderHandle } from '../../../shared/agent-session import { journalRowSchemaVersion } from '../../../shared/agent-session-journal-types' import { agentJournalLinkageFields } from '../../../shared/agent-session-journal-producer' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { journalItemOwnerFence } from './journal-item-provenance' import type { JournalReducerState } from './journal-reducer' import type { JournalDispatchRow, @@ -51,7 +52,12 @@ export function journalItemRowBuilder( state: () => JournalReducerState, address: AgentJournalItemIdentity | string, body: AgentJournalItemBody, - options: AgentJournalRowAttribution & { fence: number; observedAt?: number; recovered?: true }, + options: AgentJournalRowAttribution & { + fence: number + ownerFence?: number + observedAt?: number + recovered?: true + }, revisions?: Map ): RowBuilder { return (seq, ts) => @@ -61,6 +67,7 @@ export function journalItemRowBuilder( body, seq, fence: options.fence, + ownerFence: options.ownerFence, ts: options.observedAt ?? ts, recovered: options.recovered, linkage: options, @@ -187,7 +194,11 @@ export function journalLifecycleBatchRowBuilder( ? { ...mutation, body: turnEndAfterStop(current, resolved, mutation.body) } : mutation, itemId, - revision + revision, + journalItemOwnerFence(current, resolved, { + fence: options.fence, + ownerFence: mutation.kind === 'item' ? mutation.ownerFence : undefined + }) ) }) const row: JournalLifecycleBatchRow = { @@ -226,6 +237,8 @@ export function buildJournalItemRow( body: AgentJournalItemBody seq: number fence: number + /** The writer's provenance claim (`journalItemOwnerFence`); absent for host bookkeeping. */ + ownerFence?: number ts: number recovered?: true linkage?: AgentJournalProducerLinkage @@ -245,6 +258,7 @@ export function buildJournalItemRow( revision, body, ...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts, [body]), + ownerFence: journalItemOwnerFence(input.state, resolved, input), ...(input.recovered ? { recovered: input.recovered } : {}), turnScope: input.turnScope, ...agentJournalLinkageFields(input.linkage) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 477e8549ae64..de246e0567d0 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -80,12 +80,18 @@ export const AGENT_JOURNAL_EPOCH_REASONS = [ ] as const export type AgentJournalEpochReason = (typeof AGENT_JOURNAL_EPOCH_REASONS)[number] -export type JournalItemRow = JournalRowBase & { - kind: 'item' - itemId: string - revision: number - body: AgentJournalItemBody -} +/** The generation whose execution produced an item: execution provenance, distinct from the + * row's `fence`, which is writer authority. Rides the row like linkage, with no `v` bump: an older + * host ignores it. Absent on rows from before it: the item's creating fence stands in. */ +type JournalItemProvenance = { ownerFence?: number } + +export type JournalItemRow = JournalRowBase & + JournalItemProvenance & { + kind: 'item' + itemId: string + revision: number + body: AgentJournalItemBody + } export type JournalTombstoneRow = JournalRowBase & { kind: 'tombstone' @@ -199,13 +205,14 @@ export type JournalDispatchRow = JournalRowBase & { * Inline like the row base, and for the same reason no `v` bump: an older host * ignores the unknown keys and reads the mutation as root, as it always did. */ export type JournalLifecycleMutation = - | (AgentJournalProducerLinkage & { - kind: 'item' - itemId: string - revision: number - body: AgentJournalItemBody - turnScope?: AgentJournalTurnScope - }) + | (AgentJournalProducerLinkage & + JournalItemProvenance & { + kind: 'item' + itemId: string + revision: number + body: AgentJournalItemBody + turnScope?: AgentJournalTurnScope + }) | { kind: 'tombstone'; itemId: string; revision: number } /** One durable append whose nested mutations share the outer ordering facts. */ diff --git a/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts b/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts index 6c8811eb4a5b..e6d25b21df9e 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-unusable-annotations.ts @@ -6,11 +6,13 @@ import { isAdmissibleAgentSessionContextUsage } from '../../../shared/agent-sess export function dropUnusableRowAnnotations(record: Record): void { dropUnusableProducerLinkage(record) dropUnusableTurnScope(record) + dropUnusableOwnerFence(record) if (record.kind === 'lifecycle-batch' && Array.isArray(record.mutations)) { for (const mutation of record.mutations) { if (isPlainObject(mutation)) { dropUnusableProducerLinkage(mutation) dropUnusableTurnScope(mutation) + dropUnusableOwnerFence(mutation) } } } @@ -56,6 +58,14 @@ function dropUnusableTurnScope(record: Record): void { } } +/** Provenance this build cannot read, removed like linkage: the item's creating fence stands in. */ +function dropUnusableOwnerFence(record: Record): void { + const value = record.ownerFence + if (value !== undefined && !(Number.isSafeInteger(value) && Number(value) >= 0)) { + delete record.ownerFence + } +} + /** Context facts this build cannot read, removed from the turn row that carries * them. Same reasoning as linkage: they are an annotation on the turn, and * rejecting the row for them would truncate the journal from that row on. */ diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index 3259d488b22e..4295f0287069 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -59,6 +59,9 @@ export type JournalAppendResult = { export type JournalItemAppendOptions = AgentJournalRowAttribution & { fence: number + /** The generation whose execution produced the item: set by a provider's observation, and by + * nothing else. Host bookkeeping leaves it out and keeps the item's (`journalItemOwnerFence`). */ + ownerFence?: number observedAt?: number recovered?: true } @@ -76,6 +79,9 @@ export type JournalLifecycleBatchInput = { settlementId: string mutations: readonly JournalLifecycleMutationInput[] fence: number + /** A provider's own batch: the generation that produced every item it writes + * (`JournalItemAppendOptions.ownerFence`). */ + ownerFence?: number recovered?: true /** Submission verdicts belonging to this settlement, committed before its item rows. */ dispatches?: readonly ResolveDispatchInput[] diff --git a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts index 2a71be5515c8..01d9e47ac7b0 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts @@ -49,7 +49,8 @@ export function applyJournalSubmission( itemId, 0, journalRenderItem(itemId, 0, row.body, row, turnScope), - row.fence + // A person's message is no generation's work: its provenance is the fence that accepted it. + { fence: row.fence } ) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts index 476089292a93..3a443559e2e6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts @@ -577,7 +577,7 @@ describe('producer linkage reaches the journal through every append path', () => await deferred.drained() expect(journalAppendOptions).toEqual([ - { fence: 5, turnScope: AGENT_JOURNAL_THREAD_SCOPE, ...LINKAGE } + { fence: 5, ownerFence: 5, turnScope: AGENT_JOURNAL_THREAD_SCOPE, ...LINKAGE } ]) deferred.close() } @@ -601,7 +601,7 @@ describe('producer linkage reaches the journal through every append path', () => await deferred.drained() expect(journalAppendOptions).toEqual([ - { fence: 5, turnScope: AGENT_JOURNAL_THREAD_SCOPE, ...LINKAGE } + { fence: 5, ownerFence: 5, turnScope: AGENT_JOURNAL_THREAD_SCOPE, ...LINKAGE } ]) deferred.close() } @@ -637,7 +637,10 @@ describe('producer linkage reaches the journal through every append path', () => // A control, not a pin. Absence is the claim, so the keys must be missing // rather than present-and-undefined: a reader holding this options object // would read `agentId: undefined` as a key that exists. - expect(journalAppendOptions).toEqual([{ fence: 5, turnScope: AGENT_JOURNAL_THREAD_SCOPE }]) + // A provider observation names the generation that produced it: the bound fence. + expect(journalAppendOptions).toEqual([ + { fence: 5, ownerFence: 5, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ]) deferred.close() }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index 1e2d2c64d70a..880a7dfa86b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -246,7 +246,9 @@ export function createDeferredStructuredAgentSessionEventSink(deps: { settlementId, mutations, // No row-level linkage: each mutation names its own (see the batch row builder). - fence: bound.fence + fence: bound.fence, + // The provider's own observation: its child's generation produced every item it creates. + ownerFence: bound.fence }) }, { ...options, lifecycle: true } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts index e7ce667ed2b6..26bec39b190f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts @@ -3,6 +3,9 @@ // Every such path calls this, so a row-level field added to the sink's options // reaches the durable row through one edit rather than through whichever spread // the next change remembers. Lifecycle batches carry producers per mutation. +// +// The sink carries only the provider's own observations, bound at its child's fence: that fence is +// the generation whose execution produced each item it creates (`JournalItemRow.ownerFence`). import { agentJournalLinkageFields } from '../../../shared/agent-session-journal-producer' import type { JournalItemAppendOptions } from '../agent-session-journal/journal-store-contracts' @@ -14,6 +17,7 @@ export function structuredAgentSessionJournalAppendOptions( ): JournalItemAppendOptions { return { fence, + ownerFence: fence, ...(options.observedAt === undefined ? {} : { observedAt: options.observedAt }), turnScope: options.turnScope, ...agentJournalLinkageFields(options) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts index 1b1ab589de65..a5bc09e798b5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts @@ -115,7 +115,13 @@ async function reviseFromProvider(itemId: string): Promise { async function rewindKeepingPrompt(itemId: string): Promise { const fence = store.getRecord(SESSION)!.lease.runtimeFence await journal().replaceEpochItems('handle_forked', fence, [ - { identity: PROMPT, body: promptBody(itemId), turnScope: AGENT_JOURNAL_THREAD_SCOPE } + // The prompt keeps the generation that raised it (this one), as a rewind carries it. + { + identity: PROMPT, + body: promptBody(itemId), + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + ownerFence: fence + } ]) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts index bd819d09f2ca..5c1f9df0ce41 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts @@ -2,7 +2,6 @@ import { agentSessionProviderHandleChainHead } from '../../../shared/agent-sessi import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import { agentJournalLinkageFields } from '../../../shared/agent-session-journal-producer' import { - agentJournalItemKey, agentJournalSubmissionKey, parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' @@ -23,7 +22,10 @@ import { conversationCommandBlocked } from './structured-conversation-command-ad import { rewindRefusal } from './structured-rewind-refusal' import { renameRewindTurnOpener } from './structured-rewind-journal-body' import { persistRewindRecord, recoverStructuredRewind } from './structured-rewind-recovery' -import { mergeRetainedHostLifecycleRows } from './structured-rewind-retained-host-rows' +import { + mergeRetainedHostLifecycleRows, + rewindProviderRows +} from './structured-rewind-retained-host-rows' export async function rewindStructuredAgentSession( context: StructuredAgentSessionMutationContext, @@ -165,11 +167,13 @@ export async function rewindStructuredAgentSession( if (!body) { throw new Error('agent_session_rewind:missing-retained-item') } + const ownerFence = ctx.journal.itemFence(itemId) return { itemId: providerKey(itemId), body: renameRewindTurnOpener(body, providerKey), observedAt, ...(turnScope ? { turnScope } : {}), + ...(ownerFence === undefined ? {} : { ownerFence }), ...agentJournalLinkageFields(linkage) } }) @@ -204,11 +208,7 @@ export async function rewindStructuredAgentSession( onPrepared: async (items) => { const retained = mergeRetainedHostLifecycleRows( prepared.retained, - items.map(({ identity, body }) => ({ - itemId: agentJournalItemKey(identity), - body, - observedAt: ctx.now() - })) + rewindProviderRows(items, ctx.now(), ctx.fence) ) if ( retained.length > 10_000 || @@ -241,11 +241,7 @@ export async function rewindStructuredAgentSession( const confirmed = provider.items ? mergeRetainedHostLifecycleRows( prepared.retained, - provider.items.map(({ identity, body }) => ({ - itemId: agentJournalItemKey(identity), - body, - observedAt: ctx.now() - })) + rewindProviderRows(provider.items, ctx.now(), ctx.fence) ) : prepared.retained if ( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts index 3f1cf21d3dbb..d8435b85b926 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts @@ -508,7 +508,8 @@ describe('a rewind that restates a turnless Stop', () => { await journal().replaceEpochItems('handle_forked', 1, [ { identity: stoppedTurn, - body: { ...ended, completedAt: Date.now() + 1, outcome: 'cancellation' } + body: { ...ended, completedAt: Date.now() + 1, outcome: 'cancellation' }, + ownerFence: 1 } ]) const mail = rig.send('mail after the rewind') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts index d1cfdec43356..a6b5b9b30be9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts @@ -119,6 +119,7 @@ function transitionAppend( batch: { settlementId: step.settlementId, fence, + ownerFence: fence, recovered: step.recovered, resolve: () => step.resolve(journal) } diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts b/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts index f41e83522d2c..5fc0f7f184a7 100644 --- a/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts +++ b/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts @@ -5,7 +5,10 @@ // each keeps the ones its retained row held. import { parseCodexGoalJournalItemId } from '../../codex/codex-goal-journal-identity' -import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { + agentJournalItemKey, + parseAgentJournalItemKey +} from '../../../shared/agent-session-journal-item-key' import { AGENT_JOURNAL_ITEM_BODY_KINDS, isAdmissibleAgentJournalMessageBody @@ -64,12 +67,28 @@ export function mergeRetainedHostLifecycleRows( ) } +/** The live agent's rewound history as retained rows: its own observation, at the fence it + * answered at. A row the old epoch held keeps that row's provenance when merged. */ +export function rewindProviderRows( + items: readonly { identity: AgentJournalItemIdentity; body: AgentJournalItemBody }[], + observedAt: number, + ownerFence: number +): RetainedRow[] { + return items.map(({ identity, body }) => ({ + itemId: agentJournalItemKey(identity), + body, + observedAt, + ownerFence + })) +} + /** The rebuilt epoch's item for one retained row, with the scope and producer it was written with. */ export function retainedRowReplacement(row: RetainedRow): AgentJournalProducerLinkage & { identity: AgentJournalItemIdentity body: AgentJournalItemBody observedAt: number turnScope?: AgentJournalTurnScope + ownerFence: number } { const identity = parseAgentJournalItemKey(row.itemId) if (!identity) { @@ -87,6 +106,9 @@ export function retainedRowReplacement(row: RetainedRow): AgentJournalProducerLi body: restoreRewindJournalBody(row.body), observedAt: row.observedAt, ...(scope ? { turnScope: scope } : {}), + // The old fold's provenance, so a rewind never makes a dead generation's work current. A record + // from before rows carried it states none: 0, which is never current. + ownerFence: row.ownerFence ?? 0, ...(agentId === undefined ? {} : { agentId }), ...(parentAgentId === undefined ? {} : { parentAgentId }), ...(providerParentRef === undefined ? {} : { providerParentRef }), diff --git a/src/shared/agent-session-rewind.ts b/src/shared/agent-session-rewind.ts index e7d864390cca..36e30f9b79a4 100644 --- a/src/shared/agent-session-rewind.ts +++ b/src/shared/agent-session-rewind.ts @@ -55,6 +55,9 @@ export const AgentSessionRewindRecordSchema = z.object({ observedAt: z.number().finite(), // Absent on records written before rows stated them: the rebuild then places the row by position. turnScope: AgentJournalTurnScopeSchema.optional(), + // The row's execution provenance, carried into the rebuilt epoch. Absent on records written + // before rows carried it. + ownerFence: z.number().int().nonnegative().optional(), ...AgentJournalProducerLinkageFields }) )