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.
2026-11-03 18:47:40 +00:00 · 2026-10-12 15:06:46 -07:00
parent be4cea1584
commit d438377157
34 changed files with 534 additions and 70 deletions
@@ -153,12 +153,20 @@ describe('Grok roster ownership after more than 32 spawning groups', () => {
        {
          id: 'old',
          label: 'Original label',
+          ownerFence: 1,
          startedAt: 2,
          settledAt: 301,
          state: 'completed',
          tokens: 12
        },
-        { id: 'sibling', label: 'Original label 2', startedAt: 3, settledAt: 4, state: 'completed' }
+        {
+          id: 'sibling',
+          label: 'Original label 2',
+          ownerFence: 1,
+          startedAt: 3,
+          settledAt: 4,
+          state: 'completed'
+        }
      ])
      expect(
        (await roster(fixture)).flatMap(({ group }) =>
@@ -192,6 +200,7 @@ describe('Grok roster ownership after more than 32 spawning groups', () => {
        {
          id: 'old',
          label: 'Original label',
+          ownerFence: 1,
          startedAt: 2,
          settledAt: 301,
          state: kind === 'output' ? 'completed' : 'stopped'
@@ -62,6 +62,7 @@ describe('Grok subagents through the shared timeline', () => {
      {
        id: 'subagent-1',
        label: 'Count note lines',
+        ownerFence: 1,
        state: 'completed',
        tokens: 13778,
        startedAt: expect.any(Number),
@@ -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' })

@@ -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)]
@@ -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)
@@ -27,8 +27,8 @@ 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`). */
+  fences: { fence: number; ownerFence?: number }
 ): void {
  const tombstoned = state.tombstones.get(itemId)
  if (tombstoned !== undefined && revision <= tombstoned) {
@@ -41,10 +41,14 @@ 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
  }
+  // A revision keeps the provenance it found unless it states one: an owner's observation does.
+  if (fences.ownerFence !== undefined) {
+    state.itemFences.set(itemId, fences.ownerFence)
+  }
  // Creation sequence is the ordering key; a revision refreshes content only.
  // `observedAt` is pinned with it: clients sort the timeline by that timestamp,
  // so letting a revision advance it makes the row jump past everything that
@@ -0,0 +1,186 @@
+// Execution provenance (`JournalItemRow.ownerFence`): the generation whose execution produced an
+// item, kept apart from each row's writer fence. Every writer states it; the fold prefers 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,
+  insertTestJournalRowJson,
+  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 a provider observation's owner, and a host revision keeps it", async () => {
+    const journal = await open()
+    await journal.appendItem(TURN, RUNNING, { fence: 4, ownerFence: 3, turnScope: SCOPE })
+    // Host bookkeeping under a later writer fence states no owner of its own.
+    await journal.appendItem(TURN, SETTLED, { fence: 5, turnScope: SCOPE })
+
+    expect(journal.itemFence(TURN_ID)).toBe(3)
+    // Every row states it: none leaves the reader to guess from the writer fence.
+    expect(storedOwnerFences()).toEqual([3, 3])
+  })
+
+  it('a first write with no stated owner is its writer generation’s', async () => {
+    const journal = await open()
+    await journal.appendItem(
+      item(2),
+      { kind: 'status', text: 'note' },
+      { fence: 6, turnScope: SCOPE }
+    )
+    expect(journal.itemFence(agentJournalItemKey(item(2)))).toBe(6)
+    expect(storedOwnerFences()).toEqual([6])
+  })
+
+  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('survives a reopen from storage', async () => {
+    const journal = await open()
+    await journal.appendItem(TURN, RUNNING, { fence: 4, ownerFence: 3, turnScope: SCOPE })
+    await journals.closeAll()
+
+    expect((await open()).itemFence(TURN_ID)).toBe(3)
+  })
+
+  it('epoch replacement carries each item’s provenance', async () => {
+    const journal = await open()
+    await journal.appendItem(TURN, RUNNING, { fence: 2, ownerFence: 2, turnScope: SCOPE })
+    const before = journal.epoch
+
+    await journal.replaceEpochItems('legacy_import', 5, [
+      { identity: TURN, body: RUNNING, ownerFence: journal.itemFence(TURN_ID) ?? 0 },
+      { identity: item(2), body: { kind: 'status', text: 'new' }, ownerFence: 5 }
+    ])
+
+    expect(journal.epoch).not.toBe(before)
+    // Written at fence 5, the dead generation's turn is still fence 2's work.
+    expect(journal.itemFence(TURN_ID)).toBe(2)
+    expect(journal.itemFence(agentJournalItemKey(item(2)))).toBe(5)
+  })
+
+  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)
+  })
+
+  it('a stored provenance that is not a fence is dropped, and the row reads as its writer’s', async () => {
+    const journal = await open()
+    await journal.appendItem(TURN, RUNNING, { fence: 2, ownerFence: 2, turnScope: SCOPE })
+    const rows = liveTestJournalRows(database.db, IDENTITY.sessionId)
+    const stored = rows.find((row) => row.rowJson.includes('"kind":"item"'))
+    const last = rows.at(-1)?.seq
+    if (!stored || last === undefined) {
+      throw new Error('expected the item row')
+    }
+    await journals.closeAll()
+    database = openTestJournalHostDatabase(root)
+    const parsed: Record<string, unknown> = JSON.parse(stored.rowJson)
+    insertTestJournalRowJson(
+      database.db,
+      IDENTITY.sessionId,
+      last + 1,
+      JSON.stringify({
+        ...parsed,
+        seq: last + 1,
+        revision: Number(parsed.revision) + 1,
+        fence: 7,
+        ownerFence: -1
+      })
+    )
+
+    expect((await open()).itemFence(TURN_ID)).toBe(2)
+  })
+})
@@ -0,0 +1,129 @@
+// Execution provenance: the generation whose execution produced an item (`JournalItemRow.ownerFence`)
+// and each entry of a container row.
+//
+// Each entry of a container row: every subagent roster entry and every
+// background task. One roster row is revised in place across generations (a resumed provider keeps
+// the group's identity), so the row's own provenance cannot say which generation each child
+// belongs to. A provider's observation stamps the entries it changed or added with its own
+// generation; an entry it carries unchanged keeps the one it had (or, never stamped, the row's
+// earlier provenance). Host bookkeeping states no provenance and stamps nothing.
+
+import type { AgentJournalItemBody } from '../../../shared/agent-session-journal-types'
+import type { JournalReducerState } from './journal-reducer'
+import {
+  isBackgroundTaskBlock,
+  isSubagentGroupBlock,
+  type NativeChatBlock
+} from '../../../shared/native-chat-types'
+
+type Entry = { ownerFence?: number }
+
+/** An entry's observed fields, in a fixed order: its provenance is not part of what was seen. */
+function observationKey(entry: Entry): string {
+  const { ownerFence: _ownerFence, ...fields } = entry
+  return JSON.stringify(Object.entries(fields).sort(([left], [right]) => left.localeCompare(right)))
+}
+
+function stamp<T extends Entry>(
+  entry: T,
+  previous: T | undefined,
+  previousItemFence: number | undefined,
+  ownerFence: number
+): T {
+  if (previous && observationKey(previous) === observationKey(entry)) {
+    const kept = previous.ownerFence ?? previousItemFence
+    return kept === undefined || kept === entry.ownerFence ? entry : { ...entry, ownerFence: kept }
+  }
+  // No earlier entry to compare: one that already names its generation (history carried into a new
+  // epoch) keeps it.
+  if (!previous && entry.ownerFence !== undefined) {
+    return entry
+  }
+  return entry.ownerFence === ownerFence ? entry : { ...entry, ownerFence }
+}
+
+function hasEntries(body: AgentJournalItemBody): boolean {
+  return (
+    body.kind === 'message' &&
+    body.blocks.some((block) => isSubagentGroupBlock(block) || isBackgroundTaskBlock(block))
+  )
+}
+
+/** The body as written by an observation of generation `ownerFence`, each entry stamped. */
+export function stampJournalEntryProvenance(
+  body: AgentJournalItemBody,
+  previous: { body: AgentJournalItemBody; ownerFence?: number } | undefined,
+  ownerFence: number
+): AgentJournalItemBody {
+  if (body.kind !== 'message' || !hasEntries(body)) {
+    return body
+  }
+  const earlier: readonly NativeChatBlock[] =
+    previous?.body.kind === 'message' ? previous.body.blocks : []
+  const blocks = body.blocks.map((block): NativeChatBlock => {
+    if (isSubagentGroupBlock(block)) {
+      const group = earlier.find(
+        (candidate) => isSubagentGroupBlock(candidate) && candidate.groupId === block.groupId
+      )
+      const agents = group && isSubagentGroupBlock(group) ? group.agents : []
+      return {
+        ...block,
+        agents: block.agents.map((agent) =>
+          stamp(
+            agent,
+            agents.find((candidate) => candidate.id === agent.id),
+            previous?.ownerFence,
+            ownerFence
+          )
+        )
+      }
+    }
+    if (isBackgroundTaskBlock(block)) {
+      const task = earlier.find(
+        (candidate) => isBackgroundTaskBlock(candidate) && candidate.taskId === block.taskId
+      )
+      return stamp(
+        block,
+        task && isBackgroundTaskBlock(task) ? task : undefined,
+        previous?.ownerFence,
+        ownerFence
+      )
+    }
+    return block
+  })
+  return { ...body, blocks }
+}
+
+/** The provenance a row states: the writer's own claim when it has one — a provider's observation,
+ *  or history carried into a new epoch — else host bookkeeping's, which keeps the item's, and for
+ *  an item it creates is the fence it writes at. Every item row states it. */
+export function journalItemOwnerFence(
+  state: Pick<JournalReducerState, 'itemFences' | 'items'>,
+  resolvedItemId: string,
+  write: { fence: number; ownerFence?: number }
+): number {
+  if (write.ownerFence !== undefined) {
+    return write.ownerFence
+  }
+  const kept = state.items.has(resolvedItemId) ? state.itemFences.get(resolvedItemId) : undefined
+  return kept ?? write.fence
+}
+
+/** A provider's observation stamps each roster entry and background task it saw
+ *  (`stampJournalEntryProvenance`); host bookkeeping states no provenance and stamps nothing. */
+export function journalObservedBody(
+  state: Pick<JournalReducerState, 'itemFences' | 'items'>,
+  resolvedItemId: string,
+  body: AgentJournalItemBody,
+  ownerFence: number | undefined
+): AgentJournalItemBody {
+  if (ownerFence === undefined) {
+    return body
+  }
+  const previous = state.items.get(resolvedItemId)
+  return stampJournalEntryProvenance(
+    body,
+    previous && { body: previous.body, ownerFence: state.itemFences.get(resolvedItemId) },
+    ownerFence
+  )
+}
@@ -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 () => {
@@ -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 }
@@ -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 writes.
+    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
@@ -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],
@@ -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
      )
    )
@@ -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
  )
 }
@@ -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 }
@@ -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 {
@@ -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>
@@ -109,15 +110,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)) {
@@ -141,13 +143,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)
      }
@@ -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, journalObservedBody } 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,
@@ -183,10 +190,22 @@ export function journalLifecycleBatchRowBuilder(
      const revision = nextJournalItemRevision(current, itemId, runningRevisions)
      return journalLifecycleMutationRow(
        mutation.kind === 'item'
-          ? { ...mutation, body: turnEndAfterStop(current, resolved, mutation.body) }
+          ? {
+              ...mutation,
+              body: journalObservedBody(
+                current,
+                resolved,
+                turnEndAfterStop(current, resolved, mutation.body),
+                mutation.ownerFence
+              )
+            }
          : mutation,
        itemId,
-        revision
+        revision,
+        journalItemOwnerFence(current, resolved, {
+          fence: options.fence,
+          ownerFence: mutation.kind === 'item' ? mutation.ownerFence : undefined
+        })
      )
    })
    const row: JournalLifecycleBatchRow = {
@@ -225,6 +244,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
@@ -237,13 +258,19 @@ export function buildJournalItemRow(
  // A tombstoned row keeps its revision in `tombstones`, and the reducer drops
  // any item at or below it — so a re-add has to outrank the tombstone too.
  const revision = nextJournalItemRevision(input.state, itemId, input.revisions)
-  const body = turnEndAfterStop(input.state, resolved, input.body)
+  const body = journalObservedBody(
+    input.state,
+    resolved,
+    turnEndAfterStop(input.state, resolved, input.body),
+    input.ownerFence
+  )
  return {
    kind: 'item',
    itemId,
    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)
@@ -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'
@@ -196,13 +202,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. */
@@ -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. */
@@ -59,6 +59,9 @@ export type JournalAppendResult = {

 export type JournalItemAppendOptions = AgentJournalRowAttribution & {
  fence: number
+  /** The generation whose execution produced the item: set by a provider's observation, and by
+   *  nothing else. Host bookkeeping leaves it out and keeps the item's (`journalItemOwnerFence`). */
+  ownerFence?: number
  observedAt?: number
  recovered?: true
 }
@@ -76,6 +79,9 @@ export type JournalLifecycleBatchInput = {
  settlementId: string
  mutations: readonly JournalLifecycleMutationInput[]
  fence: number
+  /** A provider's own batch: the generation that produced every item it writes
+   *  (`JournalItemAppendOptions.ownerFence`). */
+  ownerFence?: number
  recovered?: true
  /** Submission verdicts belonging to this settlement, committed before its item rows. */
  dispatches?: readonly ResolveDispatchInput[]
@@ -46,7 +46,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 }
  )
 }

@@ -215,9 +215,14 @@ describe('a background task', () => {
    rig.assembler.apply({ type: 'item.open', item: 'ran', body: backgroundTask('ran', 'working') })
    rig.assembler.apply({ type: 'item.close', item: 'ran', body: backgroundTask('ran', 'done') })
    rig.assembler.apply({ type: 'session.ended', verdict: { state: 'unverifiable' } })
-    expect((await rig.row(providerItemId('item', 'bg')))?.body).toEqual(
-      backgroundTask('bg', 'unverifiable')
-    )
+    // The entry keeps the generation that ran it (`ownerFence`), stamped when it was observed.
+    const unverifiable = backgroundTask('bg', 'unverifiable')
+    expect((await rig.row(providerItemId('item', 'bg')))?.body).toEqual({
+      ...unverifiable,
+      blocks: unverifiable.blocks.map((block) =>
+        block.type === 'background-task' ? { ...block, ownerFence: 1 } : block
+      )
+    })
    expect(await backgroundTaskState(rig, 'ran')).toBe('done')
  })
 })
@@ -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()
  })
 })
@@ -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.
+            ownerFence: bound.fence
          })
      },
      { ...options, lifecycle: true }
@@ -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 (`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)
@@ -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
+    }
  ])
 }

@@ -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 (
@@ -443,7 +443,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')
@@ -119,6 +119,7 @@ function transitionAppend(
                  batch: {
                    settlementId: step.settlementId,
                    fence,
+                    ownerFence: fence,
                    recovered: step.recovered,
                    resolve: () => step.resolve(journal)
                  }
@@ -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 }),
@@ -70,7 +70,8 @@ const SubagentEntry = z.object({
  state: z.string().min(1),
  tokens: z.number().optional(),
  startedAt: z.number().optional(),
-  settledAt: z.number().optional()
+  settledAt: z.number().optional(),
+  ownerFence: z.number().int().nonnegative().optional()
 })

 /** Renderers select blocks by `type` equality and skip what they cannot draw,
@@ -125,7 +126,8 @@ const Block = openDiscriminatedUnion(
      outputFile: z.string().optional(),
      tokens: z.number().optional(),
      startedAt: z.number().optional(),
-      settledAt: z.number().optional()
+      settledAt: z.number().optional(),
+      ownerFence: z.number().int().nonnegative().optional()
    })
  ])
 )
@@ -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
      })
    )
@@ -151,6 +151,9 @@ export type NativeChatSubagentEntry = {
  startedAt?: number
  /** Epoch ms the entry latched terminal. */
  settledAt?: number
+  /** The agent generation (lease fence) whose execution this entry last observed: stamped by the
+   *  journal on a fresh observation, kept when carried unchanged. Absent: the row's own. */
+  ownerFence?: number
 }

 /** One spawn group's roster, revised in place as its children report activity.
@@ -195,6 +198,8 @@ export type NativeChatBackgroundTaskBlock = {
  startedAt?: number
  /** Epoch ms the row latched terminal. */
  settledAt?: number
+  /** As `NativeChatSubagentEntry.ownerFence`. */
+  ownerFence?: number
 }

 export type NativeChatBlock =