Harden structured agent-session wire recovery

src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts

@@ -123,6 +123,23 @@ describe('readAgentSessionHistory', () => {
     expect(page.page.window.nextCursor.sequence).toBeGreaterThan(cursor.sequence)
   })
 
+  it('carries tombstones in a forward catch-up page', async () => {
+    await appendItems(1)
+    const cursor = journal.cursor()
+    await journal.appendTombstone(item(1), { fence: 1 })
+
+    const page = readAgentSessionHistory(journal, {
+      sessionId: 'session-1',
+      direction: 'after',
+      cursor
+    })
+    if (!page.ok) {
+      throw new Error(`expected a page, got reset ${page.reset}`)
+    }
+    expect(page.page.items).toHaveLength(0)
+    expect(page.page.removedItemIds).toEqual(['codex:thread-1:turn-1:1'])
+  })
+
   it('reports a forward read with no cursor as cursor_ahead rather than serving the tail', async () => {
     await appendItems(1)
     expect(

src/main/native-chat/agent-session-wire/agent-session-history-page.ts

@@ -102,6 +102,7 @@ function readForward(
       snapshot,
       direction: 'after',
       items: projected.batch.items,
+      removedItemIds: projected.batch.removedItemIds,
       // Reading after a position means there is something before it.
       hasOlder: cursor.sequence > 0,
       hasNewer: since.rows.length > rows.length,
@@ -115,6 +116,7 @@ function buildPage(input: {
   snapshot: AgentJournalSnapshot
   direction: AgentSessionHistoryDirection
   items: AgentJournalRenderItem[]
+  removedItemIds?: string[]
   hasOlder: boolean
   hasNewer: boolean
   fallbackCursor: AgentJournalCursor
@@ -129,6 +131,7 @@ function buildPage(input: {
     epoch,
     direction: input.direction,
     items: input.items,
+    removedItemIds: input.removedItemIds ?? [],
     submissions: input.snapshot.submissions.filter((submission) =>
       pageItemIds.has(agentJournalSubmissionKey(submission.clientMessageId))
     ),

src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts

@@ -5,10 +5,12 @@
 // the decisions that must not be client-supplied — the spawn token, the claim
 // key, the owner probe — and passes them in.
 
+import { isDeepStrictEqual } from 'node:util'
 import type {
   AgentSessionAttachResult,
   AgentSessionMutationResult
 } from '../../../shared/agent-session-wire'
+import { agentSessionLeaseAdmitsWriter } from '../../../shared/agent-session-lease-adjudication'
 import type { AgentSessionRecord } from '../../../shared/agent-session-record'
 import {
   admitAttachOrRefuse,
@@ -62,7 +64,7 @@ export async function performAttach(
     )
     record = reserved.record
     replayed = reserved.disposition === 'replayed'
-    if (record.lease.ownerProcess === null) {
+    if (!agentSessionLeaseAdmitsWriter(record.lease)) {
       record = await acquireOwner(input, record)
     }
   } catch (error) {
@@ -107,17 +109,26 @@ async function acquireOwner(
   record: AgentSessionRecord
 ): Promise<AgentSessionRecord> {
   const fence = record.lease.runtimeFence
+  const spawnToken = record.lease.reservedSpawnToken
+  if (!spawnToken) {
+    throw new Error('agent_session_ownership_unknown')
+  }
   const acquired = await input.adapter.acquire({
     identity: journalIdentityFor(record.sessionId, input.params),
     fence,
-    spawnToken: input.authority.spawnToken
-  })
-  await input.store.commitProcessIdentity({
-    sessionId: record.sessionId,
-    fence,
-    process: acquired.process,
-    now: input.now()
+    // Retries must recover the original reservation, not mint a second child.
+    spawnToken
   })
+  if (record.lease.ownerProcess === null) {
+    await input.store.commitProcessIdentity({
+      sessionId: record.sessionId,
+      fence,
+      process: acquired.process,
+      now: input.now()
+    })
+  } else if (!isDeepStrictEqual(record.lease.ownerProcess, acquired.process)) {
+    throw new Error('agent_session_ownership_unknown')
+  }
   return input.store.proveOwner({
     sessionId: record.sessionId,
     fence,

src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts

@@ -50,7 +50,7 @@ export type AgentSessionAttachParams = {
   agent: AgentType
   accountHome: AgentSessionAccountHome
   runtimeKind: AgentSessionOwnerRuntimeKind
-  providerHandle: AgentSessionProviderHandle
+  providerHandle: Exclude<AgentSessionProviderHandle, { kind: 'opaque' }>
 }
 
 /** Host-supplied half of the reservation. */
@@ -81,6 +81,15 @@ export function attachFingerprintFields(params: AgentSessionAttachParams): Recor
 export function admitAttachOrRefuse(
   params: AgentSessionAttachParams
 ): { ok: true; fingerprint: string } | { ok: false; refusal: AgentSessionWireRefusal } {
+  if (params.providerHandle.kind !== params.provider) {
+    return {
+      ok: false,
+      refusal: {
+        code: 'agent_session_operation_invalid',
+        message: `A ${params.provider} session requires a ${params.provider} provider handle.`
+      }
+    }
+  }
   const fingerprint = computeAgentSessionPayloadFingerprint({
     method: 'agentSession.attach',
     sessionId: params.envelope.sessionId,

src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts

@@ -238,6 +238,18 @@ describe('attach', () => {
     })
   })
 
+  it('refuses a provider handle that belongs to a different provider', async () => {
+    const params = attachParams({
+      providerHandle: { kind: 'claude', sessionId: 'claude-session', leafUuid: null }
+    })
+
+    expect(await host.attach(CALLER, params)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_invalid' }
+    })
+    expect(store.getRecord(SESSION)).toBeNull()
+  })
+
   it('refuses a second create against a live session', async () => {
     await attach()
     expect(await host.attach(CALLER, attachParams())).toMatchObject({ ok: false })
@@ -249,6 +261,99 @@ describe('attach', () => {
     const retry = await host.attach(CALLER, params)
     expect(retry).toMatchObject({ ok: true, replayed: true })
   })
+
+  it('reuses the persisted spawn token when acquisition is retried', async () => {
+    const spawnTokens: string[] = []
+    const acquire = vi
+      .fn<StructuredAgentSessionAdapter['acquire']>()
+      .mockImplementationOnce(async ({ spawnToken }) => {
+        spawnTokens.push(spawnToken)
+        throw new Error('reply lost')
+      })
+      .mockImplementation(async ({ fence, spawnToken }) => {
+        spawnTokens.push(spawnToken)
+        return {
+          process: {
+            hostId: 'local',
+            pid: 4242,
+            processStartTimeMs: 1_700_000_000_000,
+            spawnToken
+          },
+          link: {
+            linkId: `link-${fence}`,
+            handle: { provider: 'codex', threadId: THREAD },
+            origin: 'created',
+            mintedAtFence: fence,
+            observedAt: NOW
+          }
+        }
+      })
+    let token = 0
+    host = new StructuredAgentSessionHost({
+      store,
+      adapter: { ...adapter(), acquire },
+      journalRoot: root,
+      claimKeyId: 'key-1',
+      mintSpawnToken: () => `spawn-${++token}`,
+      now: () => NOW
+    })
+    const params = attachParams()
+
+    await expect(host.attach(CALLER, params)).rejects.toThrow('reply lost')
+    expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
+    expect(spawnTokens).toEqual(['spawn-1', 'spawn-1'])
+  })
+
+  it('finishes proof when a retry finds the process identity already committed', async () => {
+    const acquire = vi
+      .fn<StructuredAgentSessionAdapter['acquire']>()
+      .mockImplementationOnce(async ({ fence, spawnToken }) => ({
+        process: {
+          hostId: 'local',
+          pid: 4242,
+          processStartTimeMs: 1_700_000_000_000,
+          spawnToken
+        },
+        link: {
+          linkId: 'stale-link',
+          handle: { provider: 'codex', threadId: THREAD },
+          origin: 'created',
+          mintedAtFence: fence + 1,
+          observedAt: NOW
+        }
+      }))
+      .mockImplementation(async ({ fence, spawnToken }) => ({
+        process: {
+          hostId: 'local',
+          pid: 4242,
+          processStartTimeMs: 1_700_000_000_000,
+          spawnToken
+        },
+        link: {
+          linkId: `link-${fence}`,
+          handle: { provider: 'codex', threadId: THREAD },
+          origin: 'created',
+          mintedAtFence: fence,
+          observedAt: NOW
+        }
+      }))
+    host = new StructuredAgentSessionHost({
+      store,
+      adapter: { ...adapter(), acquire },
+      journalRoot: root,
+      claimKeyId: 'key-1',
+      mintSpawnToken: () => 'spawn-a',
+      now: () => NOW
+    })
+    const params = attachParams()
+
+    await expect(host.attach(CALLER, params)).rejects.toThrow(
+      'agent_session_provider_handle_stale_fence'
+    )
+    expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
+    expect(acquire).toHaveBeenCalledTimes(2)
+    expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live')
+  })
 })
 
 describe('send', () => {
@@ -429,6 +534,37 @@ describe('respondToPrompt', () => {
     expect(answerPrompt).not.toHaveBeenCalled()
   })
 
+  it("does not turn a recorded refusal into another client's successful answer", async () => {
+    const prompt = await seedApproval()
+    await attach()
+    const rejectedFields = {
+      itemId: prompt.itemId,
+      expectedRevision: prompt.revision,
+      optionId: 'deny'
+    }
+    const rejected = {
+      envelope: envelope('agentSession.respondTo:approval', rejectedFields),
+      kind: 'approval' as const,
+      ...rejectedFields
+    }
+    await host.respondToPrompt(CALLER, rejected)
+
+    const acceptedFields = { ...rejectedFields, optionId: 'allow' }
+    await host.respondToPrompt(
+      { callerKey: 'client-2' },
+      {
+        envelope: envelope('agentSession.respondTo:approval', acceptedFields),
+        kind: 'approval',
+        ...acceptedFields
+      }
+    )
+
+    expect(await host.respondToPrompt(CALLER, rejected)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_invalid' }
+    })
+  })
+
   it('keeps the answer and reports it undelivered when the provider callback throws', async () => {
     const prompt = await seedApproval()
     await attach()
@@ -462,6 +598,23 @@ describe('setOption', () => {
     const page = host.history({ sessionId: SESSION, direction: 'tail' })
     expect(page.ok && page.page.items).toHaveLength(0)
   })
+
+  it('does not turn an unknown provider outcome into a successful replay', async () => {
+    await attach()
+    setOption.mockRejectedValueOnce(new Error('reply lost'))
+    const fields = { key: 'model', value: 'gpt-5' }
+    const params = {
+      envelope: envelope('agentSession.setOption', fields),
+      ...fields
+    }
+
+    await expect(host.setOption(CALLER, params)).rejects.toThrow('reply lost')
+    expect(await host.setOption(CALLER, params)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_unknown' }
+    })
+    expect(setOption).toHaveBeenCalledTimes(1)
+  })
 })
 
 describe('restart', () => {
@@ -589,6 +742,33 @@ describe('subscribe', () => {
     expect(events.some((event) => event.type === 'snapshot')).toBe(false)
   })
 
+  it('drops a failed transport without aborting the mutation or other subscribers', async () => {
+    await attach()
+    const events: AgentSessionSubscribeEvent[] = []
+    host.subscribe({
+      id: 'dead-sub',
+      sessionId: SESSION,
+      emit: () => {
+        throw new Error('socket closed')
+      }
+    })
+    host.subscribe({
+      id: 'live-sub',
+      sessionId: SESSION,
+      emit: (event) => events.push(event)
+    })
+    const body = message('survive subscriber failure')
+
+    const result = await host.send(CALLER, {
+      envelope: envelope('agentSession.send', { body }),
+      body
+    })
+
+    expect(result).toMatchObject({ ok: true, value: { submission: { dispatchState: 'accepted' } } })
+    expect(dispatch).toHaveBeenCalledTimes(1)
+    expect(events.some((event) => event.type === 'batch')).toBe(true)
+  })
+
   it('resets a subscriber whose epoch is gone', async () => {
     await attach()
     const events: AgentSessionSubscribeEvent[] = []
@@ -598,6 +778,28 @@ describe('subscribe', () => {
       emit: (event) => events.push(event),
       cursor: { epoch: 'epoch-from-a-previous-life', sequence: 3 }
     })
-    expect(events[0]).toMatchObject({ type: 'reset', reset: 'epoch_changed' })
+    expect(events[0]).toMatchObject({ type: 'reset', reset: 'epoch_changed', fence: 1 })
+  })
+
+  it('publishes the replacement fence when the owner generation changes', async () => {
+    const record = await attach()
+    const events: AgentSessionSubscribeEvent[] = []
+    host.subscribe({
+      id: 'sub-4',
+      sessionId: SESSION,
+      emit: (event) => events.push(event)
+    })
+    const released = await store.evictProvenDeadOwner({
+      sessionId: SESSION,
+      expectedFence: record?.lease.runtimeFence ?? 1,
+      probe: { outcome: 'pid-absent' },
+      now: NOW
+    })
+
+    const replacement = await host.attach(CALLER, ensureParams(released.lease.runtimeFence))
+    if (!replacement.ok) {
+      throw new Error(`expected replacement owner, got ${replacement.refusal.code}`)
+    }
+    expect(events.at(-1)).toMatchObject({ type: 'snapshot', fence: replacement.fence })
   })
 })

src/main/native-chat/agent-session-wire/structured-agent-session-host.ts

@@ -45,11 +45,13 @@ import {
   setOptionPlan,
   type MutationPlan
 } from './structured-agent-session-mutation-plans'
+import { runSettledAgentSessionMutation } from './structured-agent-session-operation-settlement'
+import { resolveAgentSessionReplayOutcome } from './structured-agent-session-replay-outcome'
 import {
   AgentSessionSubscribers,
   type AgentSessionSubscriberEmit
 } from './structured-agent-session-subscribers'
-import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
+import type { AgentSessionTurnContext } from './structured-agent-session-turns'
 
 export type StructuredAgentSessionCaller = {
   /** Stable per-client identity; scopes the operation ledger and records who
@@ -68,7 +70,11 @@ export type StructuredAgentSessionHostDeps = {
   now?: () => number
 }
 
-type SessionState = { journal: AgentSessionJournal; params: AgentSessionAttachParams }
+type SessionState = {
+  journal: AgentSessionJournal
+  params: AgentSessionAttachParams
+  fence: number
+}
 
 function refuse(refusal: AgentSessionWireRefusal): { ok: false; refusal: AgentSessionWireRefusal } {
   return { ok: false, refusal }
@@ -93,9 +99,7 @@ export class StructuredAgentSessionHost {
     })
   }
 
-  private now(): number {
-    return this.deps.now?.() ?? Date.now()
-  }
+  private now = (): number => this.deps.now?.() ?? Date.now()
 
   hasSession(sessionId: string): boolean {
     return this.sessions.has(sessionId)
@@ -140,9 +144,13 @@ export class StructuredAgentSessionHost {
         params,
         now: () => this.now(),
         onAttached: (attached) => {
-          this.sessions.set(sessionId, { journal: attached.journal, params })
+          const fence = this.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0
+          const previousFence = this.sessions.get(sessionId)?.fence
+          this.sessions.set(sessionId, { journal: attached.journal, params, fence })
           if (attached.recovery) {
-            this.subscribers.reset(sessionId, attached.journal, attached.recovery.reset)
+            this.subscribers.reset(sessionId, attached.journal, attached.recovery.reset, fence)
+          } else if (previousFence !== undefined && previousFence !== fence) {
+            this.subscribers.snapshot(sessionId, attached.journal, fence)
           } else {
             this.subscribers.publish(sessionId, attached.journal)
           }
@@ -254,9 +262,22 @@ export class StructuredAgentSessionHost {
     const fence = record.lease.runtimeFence
     const ctx = this.contextFor(caller, session, envelope.sessionId, fence)
     if (admission.decision === 'replay') {
-      const recorded = plan.replay(ctx)
-      if (recorded) {
-        return { ok: true, replayed: true, fence, cursor: ctx.journal.cursor(), value: recorded }
+      const replay = resolveAgentSessionReplayOutcome({
+        operationId: envelope.clientOperationId,
+        outcome: admission.row.outcome,
+        reconstruct: () => plan.replay(ctx, admission.row.outcome)
+      })
+      if (replay.decision === 'refuse') {
+        return refuse(replay.refusal)
+      }
+      if (replay.decision === 'replay') {
+        return {
+          ok: true,
+          replayed: true,
+          fence,
+          cursor: ctx.journal.cursor(),
+          value: replay.value
+        }
       }
       // Nothing durable landed, so this id is about to run for the first time.
       // A refused call leaves its ledger row behind, and replaying past the
@@ -273,7 +294,13 @@ export class StructuredAgentSessionHost {
       }
     }
 
-    const outcome = await this.finish(caller, envelope, plan, ctx)
+    const outcome = await runSettledAgentSessionMutation({
+      store: this.deps.store,
+      callerKey: caller.callerKey,
+      envelope,
+      plan,
+      context: ctx
+    })
     return outcome.ok
       ? { ok: true, replayed: false, fence, cursor: ctx.journal.cursor(), value: outcome.value }
       : refuse(outcome.refusal)
@@ -296,36 +323,6 @@ export class StructuredAgentSessionHost {
     }
   }
 
-  /** Runs the effect and settles its ledger row, including on a throw — a row
-   *  left `pending` would replay as unresolvable forever. */
-  private async finish<TValue>(
-    caller: StructuredAgentSessionCaller,
-    envelope: AgentSessionMutationEnvelope,
-    plan: MutationPlan<TValue>,
-    ctx: AgentSessionTurnContext
-  ): Promise<TurnOutcome<TValue>> {
-    const settle = (
-      outcome: Parameters<AgentSessionRecordStore['recordOperationOutcome']>[0]['outcome']
-    ) =>
-      this.deps.store.recordOperationOutcome({
-        callerKey: caller.callerKey,
-        operationId: envelope.clientOperationId,
-        outcome
-      })
-    try {
-      const outcome = await plan.run(ctx)
-      await settle(
-        outcome.ok
-          ? { status: 'succeeded', sessionId: envelope.sessionId }
-          : { status: 'failed', code: outcome.refusal.code }
-      )
-      return outcome
-    } catch (error) {
-      await settle({ status: 'unknown' })
-      throw error
-    }
-  }
-
   /**
    * Paged read. Throws `agent_session_ownership_unknown` for a session this host
    * has not attached — a reset would tell the client to reload something that

src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts

@@ -7,6 +7,7 @@
 // which is exactly right when the crash landed before the journal write.
 
 import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
+import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger'
 import type {
   AgentSessionCancelResult,
   AgentSessionMutationEnvelope,
@@ -27,7 +28,7 @@ export type MutationPlan<TValue> = {
   method: string
   fields: Record<string, unknown>
   run: (ctx: AgentSessionTurnContext) => Promise<TurnOutcome<TValue>>
-  replay: (ctx: AgentSessionTurnContext) => TValue | null
+  replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null
 }
 
 export function sendPlan(params: {
@@ -108,6 +109,9 @@ export function setOptionPlan(params: {
     method: 'agentSession.setOption',
     fields: { key: params.key, value: params.value },
     run: (ctx) => performSetOption(ctx, params),
-    replay: () => ({ key: params.key, value: params.value })
+    // A pending row may have crashed before the adapter call. Reapplying the
+    // same assignment is safe; only a settled success can be answered directly.
+    replay: (_ctx, outcome) =>
+      outcome.status === 'succeeded' ? { key: params.key, value: params.value } : null
   }
 }

src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts

@@ -0,0 +1,33 @@
+import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire'
+import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import type { MutationPlan } from './structured-agent-session-mutation-plans'
+import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
+
+export async function runSettledAgentSessionMutation<TValue>(input: {
+  store: AgentSessionRecordStore
+  callerKey: string
+  envelope: AgentSessionMutationEnvelope
+  plan: MutationPlan<TValue>
+  context: AgentSessionTurnContext
+}): Promise<TurnOutcome<TValue>> {
+  const settle = (
+    outcome: Parameters<AgentSessionRecordStore['recordOperationOutcome']>[0]['outcome']
+  ) =>
+    input.store.recordOperationOutcome({
+      callerKey: input.callerKey,
+      operationId: input.envelope.clientOperationId,
+      outcome
+    })
+  try {
+    const outcome = await input.plan.run(input.context)
+    await settle(
+      outcome.ok
+        ? { status: 'succeeded', sessionId: input.envelope.sessionId }
+        : { status: 'failed', code: outcome.refusal.code }
+    )
+    return outcome
+  } catch (error) {
+    await settle({ status: 'unknown' })
+    throw error
+  }
+}

src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts

@@ -0,0 +1,53 @@
+import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger'
+import {
+  AGENT_SESSION_WIRE_REFUSAL_CODES,
+  type AgentSessionWireRefusal,
+  type AgentSessionWireRefusalCode
+} from '../../../shared/agent-session-wire'
+
+export type AgentSessionReplayOutcomeDecision<TValue> =
+  | { decision: 'replay'; value: TValue }
+  | { decision: 'rerun' }
+  | { decision: 'refuse'; refusal: AgentSessionWireRefusal }
+
+export function resolveAgentSessionReplayOutcome<TValue>(input: {
+  operationId: string
+  outcome: AgentSessionOperationOutcome
+  reconstruct: () => TValue | null
+}): AgentSessionReplayOutcomeDecision<TValue> {
+  const { operationId, outcome } = input
+  if (outcome.status === 'failed') {
+    const code = (AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(outcome.code)
+      ? (outcome.code as AgentSessionWireRefusalCode)
+      : 'agent_session_operation_invalid'
+    return {
+      decision: 'refuse',
+      refusal: {
+        code,
+        message: `Operation ${operationId} was already refused: ${outcome.code}.`
+      }
+    }
+  }
+  if (outcome.status === 'unknown') {
+    return {
+      decision: 'refuse',
+      refusal: {
+        code: 'agent_session_operation_unknown',
+        message: `The outcome of operation ${operationId} is unknown; it was not run again.`
+      }
+    }
+  }
+  const recorded = input.reconstruct()
+  if (recorded) {
+    return { decision: 'replay', value: recorded }
+  }
+  return outcome.status === 'succeeded'
+    ? {
+        decision: 'refuse',
+        refusal: {
+          code: 'agent_session_operation_unknown',
+          message: `Operation ${operationId} succeeded, but its result is no longer reconstructable.`
+        }
+      }
+    : { decision: 'rerun' }
+}

src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts

@@ -0,0 +1,29 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { AgentSessionRecord } from '../../../shared/agent-session-record'
+import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import { createRestartReconciler } from './structured-agent-session-restart-reconcile'
+
+describe('createRestartReconciler', () => {
+  it('reruns after an external store refresh introduces unreconciled leases', async () => {
+    let record = { sessionId: 'session-1', lease: { unreconciled: true } } as AgentSessionRecord
+    const reconcileOnRestart = vi.fn(async () => {
+      record = { ...record, lease: { ...record.lease, unreconciled: false } }
+      return new Map()
+    })
+    const store = {
+      listRecords: () => [record],
+      getRecord: () => record,
+      reconcileOnRestart
+    } as unknown as AgentSessionRecordStore
+    const reconcile = createRestartReconciler({
+      store,
+      probe: async () => ({ outcome: 'pid-absent' }),
+      now: () => 1
+    })
+
+    expect(await reconcile('session-1')).toBeNull()
+    record = { ...record, lease: { ...record.lease, unreconciled: true } }
+    expect(await reconcile('session-1')).toBeNull()
+    expect(reconcileOnRestart).toHaveBeenCalledTimes(2)
+  })
+})

src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts

@@ -9,9 +9,8 @@ import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire
 import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
 import { classifyStoreFailure } from './structured-agent-session-attach'
 
-/** Adjudicates the loaded leases once per host, on the first attach. Answers
- *  with the refusal that attach owes its caller, or null once the leases are
- *  settled. */
+/** Adjudicates leases loaded by this process or refreshed from another writer.
+ *  Answers with the refusal attach owes its caller, or null once settled. */
 export function createRestartReconciler(deps: {
   store: AgentSessionRecordStore
   probe: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
@@ -19,15 +18,15 @@ export function createRestartReconciler(deps: {
 }): (sessionId: string) => Promise<AgentSessionWireRefusal | null> {
   let pending: Promise<void> | null = null
   return async (sessionId) => {
+    if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) {
+      return null
+    }
     if (!pending) {
       const run = deps.store
         .reconcileOnRestart({ probe: deps.probe, now: deps.now() })
         .then(() => undefined)
-      // A failed adjudication must not be remembered as done, or one unlucky
-      // startup would strand every persisted session for this host's lifetime.
-      pending = run.catch((error: unknown) => {
+      pending = run.finally(() => {
         pending = null
-        throw error
       })
     }
     try {

src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts

@@ -19,6 +19,7 @@ type Subscriber = {
   sessionId: string
   emit: AgentSessionSubscriberEmit
   cursor: AgentJournalCursor
+  fence: number
 }
 
 export class AgentSessionSubscribers {
@@ -39,7 +40,8 @@ export class AgentSessionSubscribers {
       id: input.id,
       sessionId: input.sessionId,
       emit: input.emit,
-      cursor: input.cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 }
+      cursor: input.cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 },
+      fence: input.fence
     }
     const session = this.bySession.get(input.sessionId) ?? new Map<string, Subscriber>()
     session.set(input.id, subscriber)
@@ -48,7 +50,7 @@ export class AgentSessionSubscribers {
     if (input.cursor) {
       this.deliver(subscriber, input.journal)
     } else {
-      subscriber.emit({
+      this.emit(subscriber, {
         type: 'snapshot',
         sessionId: input.sessionId,
         snapshot,
@@ -65,11 +67,12 @@ export class AgentSessionSubscribers {
     if (!session || !subscriber) {
       return
     }
-    session.delete(id)
-    if (session.size === 0) {
-      this.bySession.delete(sessionId)
+    this.drop(subscriber)
+    try {
+      subscriber.emit({ type: 'end' })
+    } catch {
+      // The transport is already gone; teardown must remain idempotent.
     }
-    subscriber.emit({ type: 'end' })
   }
 
   /** Fan out whatever each subscriber has not yet seen. */
@@ -81,11 +84,26 @@ export class AgentSessionSubscribers {
 
   /** Force every subscriber back to a clean snapshot — recovery, epoch
    *  rollover, an unreadable schema. */
-  reset(sessionId: string, journal: AgentSessionJournal, reason: AgentJournalResetReason): void {
+  reset(
+    sessionId: string,
+    journal: AgentSessionJournal,
+    reason: AgentJournalResetReason,
+    fence: number
+  ): void {
     const snapshot = journal.snapshot()
     for (const subscriber of this.subscribers(sessionId)) {
-      subscriber.emit({ type: 'reset', sessionId, reset: reason, snapshot })
+      this.emit(subscriber, { type: 'reset', sessionId, reset: reason, snapshot, fence })
       subscriber.cursor = snapshot.cursor
+      subscriber.fence = fence
+    }
+  }
+
+  snapshot(sessionId: string, journal: AgentSessionJournal, fence: number): void {
+    const snapshot = journal.snapshot()
+    for (const subscriber of this.subscribers(sessionId)) {
+      this.emit(subscriber, { type: 'snapshot', sessionId, snapshot, fence })
+      subscriber.cursor = snapshot.cursor
+      subscriber.fence = fence
     }
   }
 
@@ -97,11 +115,12 @@ export class AgentSessionSubscribers {
     const since = journal.readSince(subscriber.cursor)
     const snapshot = journal.snapshot()
     if (!since.ok) {
-      subscriber.emit({
+      this.emit(subscriber, {
         type: 'reset',
         sessionId: subscriber.sessionId,
         reset: since.reset,
-        snapshot
+        snapshot,
+        fence: subscriber.fence
       })
       subscriber.cursor = snapshot.cursor
       return
@@ -115,16 +134,39 @@ export class AgentSessionSubscribers {
       afterSequence: subscriber.cursor.sequence
     })
     if (!projected.ok) {
-      subscriber.emit({
+      this.emit(subscriber, {
         type: 'reset',
         sessionId: subscriber.sessionId,
         reset: projected.reset,
-        snapshot
+        snapshot,
+        fence: subscriber.fence
       })
       subscriber.cursor = snapshot.cursor
       return
     }
-    subscriber.emit({ type: 'batch', sessionId: subscriber.sessionId, batch: projected.batch })
+    this.emit(subscriber, {
+      type: 'batch',
+      sessionId: subscriber.sessionId,
+      batch: projected.batch
+    })
     subscriber.cursor = projected.batch.cursor
   }
+
+  /** A dead transport cannot be allowed to turn a durable mutation into an
+   *  unknown outcome or poison every later publication. */
+  private emit(subscriber: Subscriber, event: AgentSessionSubscribeEvent): void {
+    try {
+      subscriber.emit(event)
+    } catch {
+      this.drop(subscriber)
+    }
+  }
+
+  private drop(subscriber: Subscriber): void {
+    const session = this.bySession.get(subscriber.sessionId)
+    session?.delete(subscriber.id)
+    if (session?.size === 0) {
+      this.bySession.delete(subscriber.sessionId)
+    }
+  }
 }

src/main/runtime/rpc/methods/structured-agent-session-schemas.ts

@@ -55,13 +55,6 @@ const ProviderHandle = z.discriminatedUnion('kind', [
       sessionId: Identifier('Invalid provider session id'),
       leafUuid: Identifier('Invalid leaf uuid').nullable()
     })
-    .strict(),
-  z
-    .object({
-      kind: z.literal('opaque'),
-      agent: Identifier('Invalid agent'),
-      value: Identifier('Invalid provider handle')
-    })
     .strict()
 ])
 

src/main/runtime/rpc/methods/structured-agent-session.test.ts

@@ -232,6 +232,13 @@ describe('parameter validation', () => {
     )
   })
 
+  it('rejects a journal-only opaque provider handle', async () => {
+    await rejects(
+      'agentSession.create',
+      attachParams({ providerHandle: { kind: 'opaque', agent: 'codex', value: 'thread-1' } })
+    )
+  })
+
   it('requires a sha256 fingerprint and a positive fence', async () => {
     await rejects(
       'agentSession.send',

src/shared/agent-session-wire.ts

@@ -37,6 +37,8 @@ export type AgentSessionHistoryPage = {
   epoch: string
   direction: AgentSessionHistoryDirection
   items: AgentJournalRenderItem[]
+  /** Populated by `after` reads so a disconnected client can apply tombstones. */
+  removedItemIds: string[]
   /** Submissions overlapping this page, so an unconfirmed bubble renders with
    *  its dispatch state instead of as a plain message. */
   submissions: AgentJournalSubmission[]
@@ -75,6 +77,7 @@ export type AgentSessionSubscribeEvent =
       sessionId: string
       reset: AgentJournalResetReason
       snapshot: AgentJournalSnapshot
+      fence: number
     }
   | { type: 'end' }
 
@@ -103,6 +106,7 @@ export const AGENT_SESSION_WIRE_REFUSAL_CODES = [
   'agent_session_operation_expired',
   'agent_session_operation_capacity',
   'agent_session_operation_invalid',
+  'agent_session_operation_unknown',
   'agent_session_item_revision_stale',
   'agent_session_already_resolved',
   'agent_session_identity_required',