Native chat: record which agent run wrote each journal item (1/5, rebuild of #26867)

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

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

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,18 +28,18 @@ export const LEGACY_MIXED_PARTITIONS = [
   [['', 30]],
   [['', 96]],
   [
-    [':1/2', 117],
-    [':2/2', 5]
+    [':1/2', 116],
+    [':2/2', 6]
   ],
   [['', 70]],
   [
     [':1/2', 36],
     [':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 {