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

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' })
 

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)]

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)

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
   }

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<AgentSessionJournal> {
+  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)
+  })
+})

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<JournalReducerState, 'itemFences' | 'items'>,
+  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
+}

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 () => {

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

src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts

@@ -75,12 +75,22 @@ export class JournalLifecycleBatchAppender {
   }
 
   private planMutations(
-    input: Pick<JournalLifecycleBatchInput, 'settlementId' | 'fence' | 'recovered'>,
-    mutations: readonly JournalLifecycleMutationInput[]
+    input: Pick<JournalLifecycleBatchInput, 'settlementId' | 'fence' | 'recovered' | 'ownerFence'>,
+    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

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],

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
       )
     )

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
   )
 }

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 }

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<JournalReducerState, 'providerActivityAt'>,
   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 {

src/main/native-chat/agent-session-journal/journal-reducer.ts

@@ -51,7 +51,8 @@ export type JournalReducerState = {
   oldestSequence: number
   highestFence: number
   items: Map<string, AgentJournalRenderItem>
-  /** 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<string, number>
   /** Latest saved output per owner, rebuilt during the existing row fold. */
   providerActivityAt: Map<number, number>
@@ -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)
       }

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<string, number>
 ): RowBuilder<JournalItemRow> {
   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)

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. */

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<string, unknown>): 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<string, unknown>): void {
   }
 }
 
+/** Provenance this build cannot read, removed like linkage: the item's creating fence stands in. */
+function dropUnusableOwnerFence(record: Record<string, unknown>): 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. */

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[]

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 }
   )
 }
 

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()
   })
 })

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 }

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)

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<void> {
 async function rewindKeepingPrompt(itemId: string): Promise<void> {
   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
+    }
   ])
 }
 

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 (

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')

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)
                   }

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 }),

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
       })
     )