From fd4a802b3ceae034af2cfa365c87acdafa67354d Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 14 Oct 2026 12:06:21 -0700 Subject: [PATCH] Native chat: record which agent run wrote each journal item (1/5, rebuild of #26867) (#28092) * feat(native-chat): record which agent run wrote each journal item Each journal item row, subagent roster entry and background-task entry now records ownerFence: the lease fence of the agent run whose output produced it. Optional, persisted, no row version bump and no migration. The journal's itemFence() returns it, so the turn-verdict rules and per-run provider activity read the producing run rather than the writer's fence. * test(native-chat): fewer provenance tests: fold reopen and unstated-owner checks into the main case Drops the epoch-replacement case (Part 2 rewind test reads the same values through the same rebuild) and the corrupt-value case (no version writes one). * test(native-chat): an unstamped roster entry from an older build reads as its row provenance Moved here from Part 4 so it sits beside stampJournalEntryProvenance. * test(native-chat): build the older-build roster row inline, after main removed its body helper Main #27947 writes one row per subagent and deleted journal-subagent-group-body; a roster row from an older build is still what this case reads. * refactor(native-chat): provenance is per item only; drop per-entry stamping Main's #27947 writes one saved row per subagent, and each background task already has its own row, so an item's own ownerFence names the run of every entry a producer still writes. Per-entry stamps inside a row mattered only for the shared roster rows saved before #27947, and no run writes into those any more, so the row's own provenance settles them correctly. Removes the entry stamping, the optional ownerFence on roster entries and background tasks, and their tests. * fix(native-chat): an item's provenance is set when it is created; revisions keep it A newer run's provider can revise an item an older run created: Pi's start and an ACP session/load both report context usage onto the newest saved turn. Restamping that turn with the newer run let the newer run's exit settle the older run's unverifiable turn as interrupted. Only creation states an owner now, as main's itemFences rule did. * test(native-chat): keep the provenance test below the production lines it covers --- .../journal-damage.test.ts | 4 +- .../journal-epoch-replacement.test.ts | 4 +- .../journal-epoch-replacement.ts | 4 + .../journal-item-fold.ts | 7 +- .../journal-item-provenance.test.ts | 128 ++++++++++++++++++ .../journal-item-provenance.ts | 16 +++ .../journal-legacy-import.test.ts | 4 + .../journal-legacy-import.ts | 4 +- .../journal-lifecycle-batch-appender.ts | 15 +- ...-lifecycle-batch-partition.test-fixture.ts | 10 +- .../journal-lifecycle-batch-partition.test.ts | 1 + .../journal-lifecycle-batch-partition.ts | 2 + .../journal-lifecycle-mutation.ts | 7 +- .../journal-provider-activity.ts | 11 +- .../agent-session-journal/journal-reducer.ts | 16 ++- .../journal-row-builders.ts | 18 ++- .../journal-row-schema.ts | 33 +++-- .../journal-row-unusable-annotations.ts | 10 ++ .../journal-store-contracts.ts | 6 + .../journal-submission-fold.ts | 3 +- ...tructured-agent-session-event-sink.test.ts | 9 +- .../structured-agent-session-event-sink.ts | 4 +- ...ed-agent-session-journal-append-options.ts | 4 + ...ent-session-prompt-answer-receipts.test.ts | 8 +- .../structured-agent-session-rewind.ts | 20 ++- ...d-agent-session-stop-event-binding.test.ts | 3 +- .../structured-agent-session-transition.ts | 1 + .../structured-rewind-retained-host-rows.ts | 24 +++- src/shared/agent-session-rewind.ts | 3 + 29 files changed, 317 insertions(+), 62 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-item-provenance.test.ts create mode 100644 src/main/native-chat/agent-session-journal/journal-item-provenance.ts 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 8b59940a6a1..168e74484b6 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 36247e2571f..c64027b9455 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 67927b40158..17b9f5e3b95 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 30554ab2993..c45d8975b25 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 00000000000..8e94ef30ac9 --- /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 00000000000..afb1df346d1 --- /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 f4cad1f9edc..9da96daf490 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 f9fc7cfab80..b934e2a5dc7 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 @@ -175,7 +175,9 @@ function replacementItems( 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 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 f942f0185d0..f906fadfbed 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 509850c7884..cc94120fe53 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 25b6f5b9653..1943f30dd67 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 4e01628c6d5..525605b9ec3 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 1340b99f310..9b939c26743 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 09a965b624e..5a7c29ef361 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 0f5c1b8a807..ecb9858b0bc 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 @@ -115,15 +116,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)) { @@ -147,13 +149,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 c55a8398cad..4b5ce901bfa 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, @@ -188,7 +195,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 = { @@ -227,6 +238,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 @@ -246,6 +259,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 90c449c1b2f..91b3eb43a8e 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' @@ -202,13 +208,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 6c8811eb4a5..e6d25b21df9 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 64dfe7625fe..c0d637fdef7 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 @@ -64,6 +64,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 } @@ -81,6 +84,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 2a71be5515c..01d9e47ac7b 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 476089292a9..3a443559e2e 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 1e2d2c64d70..880a7dfa86b 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 e7ce667ed2b..26bec39b190 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 1b1ab589de6..a5bc09e798b 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 bd819d09f2c..5c1f9df0ce4 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 3f1cf21d3db..d8435b85b92 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 d1cfdec4335..a6b5b9b30be 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 b7c98c97a2c..53f2b20cc52 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 e7d864390cc..36e30f9b79a 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 }) )