fix(ssh): gate provisional source activation

config/reliability-gates.jsonc

@@ -3103,13 +3103,14 @@
         "https://github.com/stablyai/orca/pull/7002",
         "https://github.com/stablyai/orca/pull/7054"
       ],
-      "invariant": "High-volume terminal output must stay bounded across daemon socket writes, SSH relay writer/decoder/source retention/reconnect, main model and projection admission, runtime metadata, detectors, and tail checks, main-to-renderer in-flight bytes, renderer scheduler queues, and hidden-output restore without starving focused input. Negotiated SSH spans remain exact and contiguous through activation, exit, cancellation, recovery, desktop projection, and required remote replacement; exit-time cancellation transfers published projections before proof commit, and generation close fences late proof from final exit. Additional subscribers cannot stall the owner, while the required legacy primary retains backpressure.",
-      "oracle": "The current executable slice injects main-process renderer backlog pressure, then asserts unchanged path-provenance history reuse for pathless output, ordinary terminal-wait detection without a joined-tail allocation and with full prompt-family continuity, per-PTY and total pending-output caps, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, and ACK-gated in-flight bounds. The joined source-intake/multiplex oracle blocks renderer or headless serialization, admits one snapshot-covered span and one trailing span, requires replacement reservation only after the authoritative sequence is known, advances upstream ACK eligibility exactly to the covered boundary after SnapshotEnd, and delivers then ACKs the trailing span through the ordinary live path without cancellation. Source-range ledger and multiplex seams additionally require cumulative byte ACK 40 then 100 for one 100-byte frame to release exact byte credit without early source settlement, admit a contiguous higher-generation recovered token while the prior token remains unsettled, and reject stale stream or source generations. The exit-deadline oracle publishes a source-backed desktop projection, advances a fake clock to cancellation, and requires one projection transfer, proof commit, preparation, and final exit with no provider close or retained obligation; a second controlled proof remains pending across generation close and must never publish final exit. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, invalid-checkpoint restore retention through response settlement followed by fresh-token retry and one live source frame, isolated saturated-subscriber eviction with healthy owner liveness, preserved gate-off primary backpressure, one-write ownership across write(false)/drain, control selection ahead of queued ordinary PTY input with FIFO lanes and bounded fairness, liveness rebasing, failed-exit retention through late ACK or exact recovery, response-settlement cancellation authority, token-scoped exit cancellation, exact checkpoint-to-recoveryEnd continuity, stale-owner one-shot retry, no physical PTY teardown on recovery failure, exact provider-generation closure and once-only cleanup when recovery cancellation publication or proof rejects, 16 MiB-plus-1 MiB decoder caps, charged per-PTY/session retention budgets, and eight-wide isolated reattach. The live Electron perf oracle adds hidden-output floods, renderer scheduler queue depth, dropped-output-zero normal scenarios, and active key latency budgets before promotion.",
+      "invariant": "High-volume terminal output must stay bounded across daemon socket writes, SSH relay writer/decoder/source retention/reconnect, main model and projection admission, runtime metadata, detectors, and tail checks, main-to-renderer in-flight bytes, renderer scheduler queues, and hidden-output restore without starving focused input. Negotiated SSH spans remain exact and contiguous through activation, exit, cancellation, recovery, desktop projection, and required remote replacement; provisional activation data cannot project before authority commits, exit-time cancellation transfers published projections before proof commit, and generation close fences late proof from final exit. Additional subscribers cannot stall the owner, while the required legacy primary retains backpressure.",
+      "oracle": "The current executable slice injects main-process renderer backlog pressure, then asserts unchanged path-provenance history reuse for pathless output, ordinary terminal-wait detection without a joined-tail allocation and with full prompt-family continuity, per-PTY and total pending-output caps, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, and ACK-gated in-flight bounds. The joined source-intake/multiplex oracle blocks renderer or headless serialization, admits one snapshot-covered span and one trailing span, requires replacement reservation only after the authoritative sequence is known, advances upstream ACK eligibility exactly to the covered boundary after SnapshotEnd, and delivers then ACKs the trailing span through the ordinary live path without cancellation. Source-range ledger and multiplex seams additionally require cumulative byte ACK 40 then 100 for one 100-byte frame to release exact byte credit without early source settlement, admit a contiguous higher-generation recovered token while the prior token remains unsettled, and reject stale stream or source generations. The exit-deadline oracle publishes a source-backed desktop projection, advances a fake clock to cancellation, and requires one projection transfer, proof commit, preparation, and final exit with no provider close or retained obligation; a second controlled proof remains pending across generation close and must never publish final exit. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, claim-gated projection under same-decoder-turn response/data delivery, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, invalid-checkpoint restore retention through response settlement followed by fresh-token retry and one live source frame, isolated saturated-subscriber eviction with healthy owner liveness, preserved gate-off primary backpressure, one-write ownership across write(false)/drain, control selection ahead of queued ordinary PTY input with FIFO lanes and bounded fairness, liveness rebasing, failed-exit retention through late ACK or exact recovery, response-settlement cancellation authority, token-scoped exit cancellation, exact checkpoint-to-recoveryEnd continuity, stale-owner one-shot retry, no physical PTY teardown on recovery failure, exact provider-generation closure and once-only cleanup when recovery cancellation publication or proof rejects, 16 MiB-plus-1 MiB decoder caps, charged per-PTY/session retention budgets, and eight-wide isolated reattach. The live Electron perf oracle adds hidden-output floods, renderer scheduler queue depth, dropped-output-zero normal scenarios, and active key latency budgets before promotion.",
       "commands": [
         "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/pty.test.ts",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime-path-candidate-history.test.ts",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/orca-runtime-tail-wait-memo.test.ts",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/providers/ssh-pty-provider.test.ts src/main/ssh/ssh-relay-session.test.ts src/main/ssh/ssh-relay-session-data-delivery.test.ts src/main/ssh/ssh-relay-session-incarnation.test.ts src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts src/main/ssh/ssh-relay-session-terminal-error.test.ts src/main/ssh/ssh-pty-recovery-retention-budget.test.ts src/main/ssh/relay-protocol-backpressure.test.ts src/relay/protocol-backpressure.test.ts src/relay/pty-source-credit-ledger.test.ts src/relay/pty-source-credit-scheduler.test.ts src/relay/relay-pty-source-publication.test.ts src/relay/ssh-pty-source-credit-adapter.test.ts --reporter=dot",
+        "pnpm exec vitest run --config config/vitest.config.ts src/main/providers/ssh-pty-notification-routing.test.ts src/main/providers/ssh-pty-provider-agent-session-create-operation.test.ts src/main/providers/ssh-pty-provider-exit-race.test.ts --reporter=dot",
         "pnpm exec vitest run --config config/vitest.config.ts src/relay/relay-pty-source-recovery-interleavings.test.ts src/relay/relay-pty-source-recovery-completion.test.ts src/relay/relay-pty-source-restore-retry.test.ts --reporter=dot",
         "pnpm exec vitest run --config config/vitest.config.ts src/relay/dispatcher.test.ts src/relay/pty-handler-source-publication.test.ts --reporter=dot",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/ssh/ssh-relay-session-recovery-races.test.ts --reporter=dot",
@@ -3124,6 +3125,9 @@
         "src/main/runtime/orca-runtime.test.ts",
         "src/main/runtime/orca-runtime-tail-wait-memo.test.ts",
         "src/main/providers/ssh-pty-provider.test.ts",
+        "src/main/providers/ssh-pty-notification-routing.test.ts",
+        "src/main/providers/ssh-pty-provider-agent-session-create-operation.test.ts",
+        "src/main/providers/ssh-pty-provider-exit-race.test.ts",
         "src/main/ssh/ssh-relay-session.test.ts",
         "src/main/ssh/ssh-relay-session-data-delivery.test.ts",
         "src/main/ssh/ssh-relay-session-recovery-races.test.ts",
@@ -3190,6 +3194,25 @@
             "resolves tui-idle when a stale Codex prompt is followed by the ready header"
           ]
         },
+        {
+          "file": "src/main/providers/ssh-pty-notification-routing.test.ts",
+          "assertions": [
+            "provisional source frames remain unprojected and are discarded on rollback",
+            "older rollback cancels only its exact token without replacing a newer activation"
+          ]
+        },
+        {
+          "file": "src/main/providers/ssh-pty-provider-agent-session-create-operation.test.ts",
+          "assertions": [
+            "same-decoder-turn source data waits for claim validation, failed claim rollback, and exact cancellation proof while a newer activation remains live"
+          ]
+        },
+        {
+          "file": "src/main/providers/ssh-pty-provider-exit-race.test.ts",
+          "assertions": [
+            "exit before a fresh spawn reply drops provisional source data and settles exact cancellation"
+          ]
+        },
         {
           "file": "src/main/ssh/ssh-relay-session-reconnect-incarnation.test.ts",
           "assertions": [

src/main/providers/ssh-agent-session-create-operation.ts

@@ -141,7 +141,9 @@ export async function spawnFreshSshPty(args: {
     sourceActivationLease?.commit()
     return mappedResult
   } catch (error) {
-    sourceActivationLease?.rollback()
+    if (sourceActivationLease && !(await sourceActivationLease.rollback())) {
+      throw new Error('execution_owner_unavailable')
+    }
     throw error
   } finally {
     args.exitRaceTracker.finish(operation)

src/main/providers/ssh-pty-notification-routing.test.ts

@@ -10,7 +10,7 @@ type MockMux = {
 function createSubscription() {
   const mux: MockMux = {
     onNotification: vi.fn(),
-    request: vi.fn(async () => ({ canceled: true }))
+    request: vi.fn(async () => ({ canceled: true, sentEndSu: 0, creditedEndSu: 0 }))
   }
   const dataListeners = new Set<(payload: { id: string; data: string }) => void>()
   const replayListeners = new Set<(payload: { id: string; data: string }) => void>()
@@ -140,7 +140,7 @@ describe('subscribeSshPtyNotifications', () => {
     installReceivingActivation(
       'pty-1',
       sourceActivation({ checkpointSourceEndSu: 10, recoveryEndSu: 14 })
-    )
+    ).commit()
 
     handler('pty.data', {
       id: 'pty-1',
@@ -200,7 +200,7 @@ describe('subscribeSshPtyNotifications', () => {
     installReceivingActivation(
       'pty-1',
       sourceActivation({ checkpointSourceEndSu: 0, recoveryEndSu: 4 })
-    )
+    ).commit()
     handler('pty.data', {
       id: 'pty-1',
       data: 'data',
@@ -240,7 +240,7 @@ describe('subscribeSshPtyNotifications', () => {
         deliveryToken: 'token-new',
         recoveryEndSu: 3
       })
-    )
+    ).commit()
 
     handler('pty.data', {
       id: 'pty-1',
@@ -290,7 +290,7 @@ describe('subscribeSshPtyNotifications', () => {
       ownerGeneration: 3,
       sourceLengthSu: 3
     }
-    installReceivingActivation('pty-1', sourceActivation({ recoveryEndSu: 3 }))
+    installReceivingActivation('pty-1', sourceActivation({ recoveryEndSu: 3 })).commit()
 
     handler('pty.data', {
       ...sourceParams,
@@ -357,7 +357,7 @@ describe('subscribeSshPtyNotifications', () => {
     installReceivingActivation(
       'pty-1',
       sourceActivation({ deliveryToken: 'token-old', recoveryEndSu: 3 })
-    )
+    ).commit()
     handler('pty.data', frame('old', 'token-old', 2, 3, 3))
     installReceivingActivation(
       'pty-1',
@@ -368,7 +368,7 @@ describe('subscribeSshPtyNotifications', () => {
         checkpointSourceEndSu: 10,
         recoveryEndSu: 13
       })
-    )
+    ).commit()
     handler('pty.data', frame('new', 'token-new', 3, 4, 13))
     handler('pty.data', frame('old', 'token-old', 2, 3, 6))
     handler('pty.data', frame('next', 'token-new', 3, 4, 17))
@@ -411,7 +411,7 @@ describe('subscribeSshPtyNotifications', () => {
       installReceivingActivation(
         'pty-1',
         sourceActivation({ deliveryToken: 'token-current', recoveryEndSu: 3 })
-      )
+      ).commit()
       handler('pty.data', {
         ...base,
         data: 'one',
@@ -482,7 +482,7 @@ describe('subscribeSshPtyNotifications', () => {
     const { handler, dataListeners, installReceivingActivation } = createSubscription()
     const onData = vi.fn()
     dataListeners.add(onData)
-    installReceivingActivation(
+    const lease = installReceivingActivation(
       'pty-1',
       sourceActivation({ checkpointSourceEndSu: 4, recoveryEndSu: 8 })
     )
@@ -498,6 +498,8 @@ describe('subscribeSshPtyNotifications', () => {
       sourceLengthSu: 4
     })
 
+    expect(onData).not.toHaveBeenCalled()
+    lease.commit()
     expect(onData).toHaveBeenCalledWith(
       expect.objectContaining({
         data: 'next',
@@ -538,8 +540,9 @@ describe('subscribeSshPtyNotifications', () => {
     expect(onData).toHaveBeenCalledOnce()
   })
 
-  it('rolls back after provisional frames and cancels only that activation', () => {
-    const { handler, mux, dataListeners, installReceivingActivation } = createSubscription()
+  it('drops provisional frames and settles cancellation before rollback completes', async () => {
+    const { handler, mux, dataListeners, livePtyIds, installReceivingActivation } =
+      createSubscription()
     const onData = vi.fn()
     dataListeners.add(onData)
     const lease = installReceivingActivation(
@@ -557,9 +560,10 @@ describe('subscribeSshPtyNotifications', () => {
       sourceLengthSu: 4
     })
 
-    lease.rollback()
+    await expect(lease.rollback()).resolves.toBe(true)
 
-    expect(onData).toHaveBeenCalledOnce()
+    expect(onData).not.toHaveBeenCalled()
+    expect(livePtyIds).not.toContain('ssh:conn@@pty-1')
     expect(mux.request).toHaveBeenCalledWith('pty.cancelDelivery', {
       id: 'pty-1',
       clientGeneration: 2,
@@ -568,7 +572,7 @@ describe('subscribeSshPtyNotifications', () => {
     })
   })
 
-  it('restores the exact prior cursor when a replacement rolls back after frames', () => {
+  it('restores the exact prior cursor when a replacement rolls back after frames', async () => {
     const { handler, dataListeners, installReceivingActivation } = createSubscription()
     const onData = vi.fn()
     dataListeners.add(onData)
@@ -604,7 +608,7 @@ describe('subscribeSshPtyNotifications', () => {
       sourceLengthSu: 3
     })
 
-    replacement.rollback()
+    await replacement.rollback()
     handler('pty.data', {
       id: 'pty-1',
       data: 'old',
@@ -616,10 +620,10 @@ describe('subscribeSshPtyNotifications', () => {
       sourceLengthSu: 3
     })
 
-    expect(onData.mock.calls.map(([payload]) => payload.data)).toEqual(['pre', 'new', 'old'])
+    expect(onData.mock.calls.map(([payload]) => payload.data)).toEqual(['pre', 'old'])
   })
 
-  it('does not let an older lease rollback replace a newer activation', () => {
+  it('does not let an older lease rollback replace a newer activation', async () => {
     const { handler, mux, dataListeners, installReceivingActivation } = createSubscription()
     const onData = vi.fn()
     dataListeners.add(onData)
@@ -633,7 +637,7 @@ describe('subscribeSshPtyNotifications', () => {
       })
     )
 
-    older.rollback()
+    await older.rollback()
     handler('pty.data', {
       id: 'pty-1',
       data: 'new',

src/main/providers/ssh-pty-notification-routing.ts

@@ -6,21 +6,15 @@ import type {
   SshPtyExitCallback,
   SshPtyReplayCallback
 } from './ssh-pty-provider-contract'
-import { parseSshPtySourceFrame, type SshPtySourceFrame } from './ssh-pty-source-frame'
+import { parseSshPtySourceFrame } from './ssh-pty-source-frame'
+import {
+  SshPtySourceDeliveryLedger,
+  type PendingSshPtySourceData,
+  type SshPtyReceivingActivationLease
+} from './ssh-pty-source-delivery-ledger'
 
 export type { SshPtyDataCallback, SshPtyExitCallback, SshPtyReplayCallback }
-
-type SourceDeliveryState = Readonly<{
-  activation: PtySourceReceivingActivation
-  sourceEndSu: number
-  revision: number
-  provisional: boolean
-}>
-
-export type SshPtyReceivingActivationLease = Readonly<{
-  commit: () => void
-  rollback: () => void
-}>
+export type { SshPtyReceivingActivationLease } from './ssh-pty-source-delivery-ledger'
 
 export type SshPtyNotificationSubscription = Readonly<{
   dispose: () => void
@@ -41,8 +35,29 @@ export function subscribeSshPtyNotifications(args: {
   providerGeneration: number
   resolvePtyIncarnation: (relayPtyId: string, incarnationId?: unknown) => string
 }): SshPtyNotificationSubscription {
-  const sourceDeliveryByPty = new Map<string, SourceDeliveryState>()
-  let nextSourceDeliveryRevision = 1
+  const publishData = (pending: PendingSshPtySourceData): void => {
+    const id = args.toAppPtyId(pending.relayPtyId)
+    const ptyIncarnation = args.resolvePtyIncarnation(
+      pending.relayPtyId,
+      pending.params.ptyIncarnation ?? pending.params.incarnationId
+    )
+    args.livePtyIds.add(id)
+    for (const listener of args.dataListeners) {
+      listener({
+        id,
+        data: pending.data,
+        providerGeneration: args.providerGeneration,
+        ptyIncarnation: pending.source ? (pending.params.ptyIncarnation as string) : ptyIncarnation,
+        ...(typeof pending.params.rawLength === 'number'
+          ? { sequenceChars: pending.params.rawLength }
+          : {}),
+        ...(pending.params.transformed === true ? { transformed: true } : {}),
+        ...(typeof pending.params.seq === 'number' ? { seq: pending.params.seq } : {}),
+        ...(pending.source ? { source: pending.source } : {})
+      })
+    }
+  }
+  const sourceDeliveries = new SshPtySourceDeliveryLedger(args.mux, publishData)
   const dispose = args.mux.onNotification((method, params) => {
     // Why: mux delivers every method to generic handlers; non-PTY payloads
     // (workspace.changed, fs.changed, …) have no `id` and must not reach
@@ -59,9 +74,7 @@ export function subscribeSshPtyNotifications(args: {
       const ptyIncarnation = args.resolvePtyIncarnation(relayPtyId, params.incarnationId)
       args.recordExit(relayPtyId, params.incarnationId)
       args.livePtyIds.delete(id)
-      if (!sourceDeliveryByPty.get(relayPtyId)?.provisional) {
-        sourceDeliveryByPty.delete(relayPtyId)
-      }
+      sourceDeliveries.recordExit(relayPtyId)
       for (const listener of args.exitListeners) {
         listener({
           id,
@@ -85,149 +98,31 @@ export function subscribeSshPtyNotifications(args: {
     }
     const data = typeof params.data === 'string' ? params.data : ''
     const sourceFrame = parseSshPtySourceFrame(params, data, relayPtyId)
-    if (
-      sourceFrame.malformed ||
-      (sourceFrame.source &&
-        !acceptSourceDelivery(sourceDeliveryByPty, relayPtyId, params, sourceFrame.source))
-    ) {
+    if (sourceFrame.malformed) {
       cancelExactSourceDelivery(args.mux, relayPtyId, params)
       return
     }
-    const id = args.toAppPtyId(relayPtyId)
-    const ptyIncarnation = args.resolvePtyIncarnation(
+    const pending = Object.freeze({
       relayPtyId,
-      params.ptyIncarnation ?? params.incarnationId
-    )
-    args.livePtyIds.add(id)
-    for (const listener of args.dataListeners) {
-      listener({
-        id,
-        data,
-        providerGeneration: args.providerGeneration,
-        ptyIncarnation: sourceFrame.source ? (params.ptyIncarnation as string) : ptyIncarnation,
-        ...(typeof params.rawLength === 'number' ? { sequenceChars: params.rawLength } : {}),
-        ...(params.transformed === true ? { transformed: true } : {}),
-        ...(typeof params.seq === 'number' ? { seq: params.seq } : {}),
-        ...(sourceFrame.source ? { source: sourceFrame.source } : {})
-      })
+      params,
+      data,
+      source: sourceFrame.source
+    })
+    if (sourceFrame.source) {
+      if (!sourceDeliveries.admit({ ...pending, source: sourceFrame.source })) {
+        cancelExactSourceDelivery(args.mux, relayPtyId, params)
+      }
+      return
     }
+    publishData(pending)
   })
   return Object.freeze({
     dispose,
     installReceivingActivation: (relayPtyId, activation) =>
-      installReceivingActivation(
-        args.mux,
-        sourceDeliveryByPty,
-        relayPtyId,
-        activation,
-        nextSourceDeliveryRevision++
-      )
+      sourceDeliveries.install(relayPtyId, activation)
   })
 }
 
-function installReceivingActivation(
-  mux: SshChannelMultiplexer,
-  sourceDeliveryByPty: Map<string, SourceDeliveryState>,
-  relayPtyId: string,
-  activation: PtySourceReceivingActivation,
-  revision: number
-): SshPtyReceivingActivationLease {
-  if (!relayPtyId || activation.ptyIncarnation.length === 0) {
-    throw new Error('ssh_source_receiving_activation_invalid')
-  }
-  const previous = sourceDeliveryByPty.get(relayPtyId)
-  if (previous && sameReceivingActivation(previous.activation, activation)) {
-    if (previous.provisional) {
-      throw new Error('ssh_source_receiving_activation_stale')
-    }
-    return settledReceivingActivationLease()
-  }
-  if (
-    previous &&
-    (activation.clientGeneration <= previous.activation.clientGeneration ||
-      activation.ownerGeneration <= previous.activation.ownerGeneration ||
-      activation.deliveryToken === previous.activation.deliveryToken)
-  ) {
-    throw new Error('ssh_source_receiving_activation_stale')
-  }
-  const installed = Object.freeze({
-    activation,
-    sourceEndSu: activation.checkpointSourceEndSu,
-    revision,
-    provisional: true
-  })
-  sourceDeliveryByPty.set(relayPtyId, installed)
-  let settled = false
-  return Object.freeze({
-    commit: () => {
-      if (settled) {
-        return
-      }
-      settled = true
-      const current = sourceDeliveryByPty.get(relayPtyId)
-      if (current?.revision === revision) {
-        sourceDeliveryByPty.set(relayPtyId, Object.freeze({ ...current, provisional: false }))
-      }
-    },
-    rollback: () => {
-      if (settled) {
-        return
-      }
-      settled = true
-      if (sourceDeliveryByPty.get(relayPtyId)?.revision === revision) {
-        if (previous) {
-          sourceDeliveryByPty.set(relayPtyId, previous)
-        } else {
-          sourceDeliveryByPty.delete(relayPtyId)
-        }
-      }
-      cancelExactSourceDelivery(mux, relayPtyId, activation)
-    }
-  })
-}
-
-function settledReceivingActivationLease(): SshPtyReceivingActivationLease {
-  return Object.freeze({ commit: () => {}, rollback: () => {} })
-}
-
-function sameReceivingActivation(
-  left: PtySourceReceivingActivation,
-  right: PtySourceReceivingActivation
-): boolean {
-  return (
-    left.clientGeneration === right.clientGeneration &&
-    left.ownerGeneration === right.ownerGeneration &&
-    left.ptyIncarnation === right.ptyIncarnation &&
-    left.deliveryToken === right.deliveryToken &&
-    left.checkpointSourceEndSu === right.checkpointSourceEndSu &&
-    left.recoveryEndSu === right.recoveryEndSu
-  )
-}
-
-function acceptSourceDelivery(
-  sourceDeliveryByPty: Map<string, SourceDeliveryState>,
-  relayPtyId: string,
-  params: Record<string, unknown>,
-  source: SshPtySourceFrame
-): boolean {
-  const current = sourceDeliveryByPty.get(relayPtyId)
-  if (
-    !current ||
-    current.activation.ptyIncarnation !== params.ptyIncarnation ||
-    current.activation.deliveryToken !== source.deliveryToken ||
-    current.activation.clientGeneration !== source.clientGeneration ||
-    current.activation.ownerGeneration !== source.ownerGeneration ||
-    current.sourceEndSu !== source.sourceStartSu
-  ) {
-    return false
-  }
-  sourceDeliveryByPty.set(
-    relayPtyId,
-    Object.freeze({ ...current, sourceEndSu: source.sourceEndSu })
-  )
-  return true
-}
-
 function cancelExactSourceDelivery(
   mux: SshChannelMultiplexer,
   relayPtyId: string,

src/main/providers/ssh-pty-provider-agent-session-create-operation.test.ts

@@ -1,7 +1,76 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
-import { AGENT_SESSION_CREATE_OPERATION_PROTOCOL_VERSION } from '../../shared/agent-session-host-authority'
+import {
+  AGENT_SESSION_CREATE_OPERATION_PROTOCOL_VERSION,
+  AGENT_SESSION_EXECUTION_OWNER_PROTOCOL_VERSION
+} from '../../shared/agent-session-host-authority'
+import { SshChannelMultiplexer, type MultiplexerTransport } from '../ssh/ssh-channel-multiplexer'
+import { encodeFrame, HEADER_LENGTH, MessageType } from '../ssh/relay-protocol'
 import { SshPtyProvider } from './ssh-pty-provider'
 
+function createTransport(): MultiplexerTransport & {
+  deliver: (data: Buffer) => void
+  written: Buffer[]
+} {
+  let deliver = (_data: Buffer): void => {}
+  const written: Buffer[] = []
+  return {
+    write: (data) => {
+      written.push(data)
+    },
+    onData: (callback) => {
+      deliver = callback
+    },
+    onClose: () => {},
+    deliver: (data) => deliver(data),
+    written
+  }
+}
+
+function rpcFrame(payload: Record<string, unknown>, sequence: number): Buffer {
+  return encodeFrame(MessageType.Regular, sequence, 0, Buffer.from(JSON.stringify(payload)))
+}
+
+function responseFrame(id: number, result: unknown, sequence: number): Buffer {
+  return rpcFrame({ jsonrpc: '2.0', id, result }, sequence)
+}
+
+function notificationFrame(
+  method: string,
+  params: Record<string, unknown>,
+  sequence: number
+): Buffer {
+  return rpcFrame({ jsonrpc: '2.0', method, params }, sequence)
+}
+
+function requestPayloads(transport: ReturnType<typeof createTransport>): Record<string, unknown>[] {
+  return transport.written.flatMap((frame) => {
+    if (frame[0] !== MessageType.Regular) {
+      return []
+    }
+    const payloadLength = frame.readUInt32BE(9)
+    return [
+      JSON.parse(frame.subarray(HEADER_LENGTH, HEADER_LENGTH + payloadLength).toString()) as Record<
+        string,
+        unknown
+      >
+    ]
+  })
+}
+
+async function waitForRequest(
+  transport: ReturnType<typeof createTransport>,
+  method: string
+): Promise<Record<string, unknown>> {
+  for (let turn = 0; turn < 10; turn += 1) {
+    const request = requestPayloads(transport).find((payload) => payload.method === method)
+    if (request) {
+      return request
+    }
+    await Promise.resolve()
+  }
+  throw new Error(`request not dispatched: ${method}`)
+}
+
 describe('SSH fresh agent-session create operations', () => {
   const request = vi.fn()
   let provider: SshPtyProvider
@@ -165,4 +234,163 @@ describe('SSH fresh agent-session create operations', () => {
     })
     expect(request).toHaveBeenCalledTimes(2)
   })
+
+  it('withholds same-turn source data until claim validation and isolates rollback', async () => {
+    const transport = createTransport()
+    const mux = new SshChannelMultiplexer(transport)
+    const exactProvider = new SshPtyProvider('conn-1', mux)
+    const onData = vi.fn()
+    exactProvider.onData(onData)
+    const claim = {
+      digestVersion: 1 as const,
+      keyId: 'key',
+      identityDigest: 'a'.repeat(43),
+      worktreeScopeDigest: 'b'.repeat(43),
+      agent: 'codex' as const
+    }
+    const surface = {
+      worktreeId: 'worktree',
+      tabId: 'tab',
+      leafId: '11111111-1111-4111-8111-111111111111',
+      terminalHandle: 'term_claimed'
+    }
+    const spawn = exactProvider.spawn({
+      cols: 80,
+      rows: 24,
+      agentSessionEnsure: { claim, surface }
+    })
+    const capabilityRequest = await waitForRequest(transport, 'pty.getCapabilities')
+    transport.deliver(
+      responseFrame(
+        capabilityRequest.id as number,
+        { agentSessionClaimVersion: AGENT_SESSION_EXECUTION_OWNER_PROTOCOL_VERSION },
+        1
+      )
+    )
+    const spawnRequest = await waitForRequest(transport, 'pty.spawn')
+    const oldActivation = {
+      status: 'pending',
+      clientGeneration: 2,
+      ownerGeneration: 3,
+      ptyIncarnation: 'incarnation-old',
+      deliveryToken: 'token-old',
+      checkpointSourceEndSu: 0,
+      recoveryEndSu: 3
+    }
+    transport.deliver(
+      Buffer.concat([
+        responseFrame(
+          spawnRequest.id as number,
+          {
+            id: 'pty-1',
+            incarnationId: 'incarnation-old',
+            sourceActivation: oldActivation,
+            agentSessionEnsure: {
+              disposition: 'created',
+              owner: {
+                claim: { ...claim, identityDigest: 'c'.repeat(43) },
+                generation: 'generation-old',
+                phase: 'live',
+                ptyId: 'pty-1',
+                surface
+              }
+            }
+          },
+          2
+        ),
+        notificationFrame(
+          'pty.data',
+          {
+            id: 'pty-1',
+            data: 'old',
+            ptyIncarnation: 'incarnation-old',
+            deliveryToken: 'token-old',
+            clientGeneration: 2,
+            ownerGeneration: 3,
+            sourceEndSu: 3,
+            sourceLengthSu: 3
+          },
+          3
+        )
+      ])
+    )
+
+    expect(onData).not.toHaveBeenCalled()
+    expect(exactProvider.hasPty('ssh:conn-1@@pty-1')).toBe(false)
+    const shutdownRequest = await waitForRequest(transport, 'pty.shutdown')
+    transport.deliver(responseFrame(shutdownRequest.id as number, null, 4))
+    const cancelRequest = await waitForRequest(transport, 'pty.cancelDelivery')
+    let spawnSettled = false
+    void spawn.then(
+      () => {
+        spawnSettled = true
+      },
+      () => {
+        spawnSettled = true
+      }
+    )
+    await Promise.resolve()
+    expect(spawnSettled).toBe(false)
+    expect(cancelRequest.params).toEqual({
+      id: 'pty-1',
+      clientGeneration: 2,
+      ownerGeneration: 3,
+      deliveryToken: 'token-old'
+    })
+
+    const replacement = exactProvider.spawn({ cols: 80, rows: 24 })
+    const replacementRequest = requestPayloads(transport).findLast(
+      (payload) => payload.method === 'pty.spawn'
+    )
+    expect(replacementRequest).toBeDefined()
+    transport.deliver(
+      Buffer.concat([
+        responseFrame(
+          replacementRequest!.id as number,
+          {
+            id: 'pty-1',
+            incarnationId: 'incarnation-new',
+            sourceActivation: {
+              ...oldActivation,
+              clientGeneration: 4,
+              ownerGeneration: 5,
+              ptyIncarnation: 'incarnation-new',
+              deliveryToken: 'token-new'
+            }
+          },
+          5
+        ),
+        notificationFrame(
+          'pty.data',
+          {
+            id: 'pty-1',
+            data: 'new',
+            ptyIncarnation: 'incarnation-new',
+            deliveryToken: 'token-new',
+            clientGeneration: 4,
+            ownerGeneration: 5,
+            sourceEndSu: 3,
+            sourceLengthSu: 3
+          },
+          6
+        )
+      ])
+    )
+    await expect(replacement).resolves.toMatchObject({ incarnationId: 'incarnation-new' })
+    expect(onData.mock.calls.map(([payload]) => payload.data)).toEqual(['new'])
+
+    transport.deliver(
+      responseFrame(
+        cancelRequest.id as number,
+        { canceled: true, sentEndSu: 3, creditedEndSu: 0 },
+        7
+      )
+    )
+    await expect(spawn).rejects.toThrow('agent_session_ownership_unknown')
+    expect(exactProvider.hasPty('ssh:conn-1@@pty-1')).toBe(true)
+    expect(
+      requestPayloads(transport).filter((payload) => payload.method === 'pty.cancelDelivery')
+    ).toHaveLength(1)
+    mux.dispose()
+  })
 })

src/main/providers/ssh-pty-provider-exit-race.test.ts

@@ -52,6 +52,9 @@ it('rejects a fresh SSH PTY whose exit shares the spawn response batch', async (
       })
       return result
     }
+    if (method === 'pty.cancelDelivery') {
+      return { canceled: true, sentEndSu: 4, creditedEndSu: 0 }
+    }
     return undefined
   })
 
@@ -66,7 +69,7 @@ it('rejects a fresh SSH PTY whose exit shares the spawn response batch', async (
     providerGeneration: expect.any(Number),
     ptyIncarnation: 'incarnation-raced'
   })
-  expect(dataListener).toHaveBeenCalledWith(expect.objectContaining({ data: 'data' }))
+  expect(dataListener).not.toHaveBeenCalled()
   expect(mux.request).toHaveBeenCalledWith('pty.cancelDelivery', {
     id: 'pty-raced',
     clientGeneration: 2,
@@ -146,6 +149,9 @@ it('returns a provisional source activation lease to reconnect authority', async
     }
   }
   mux.request.mockImplementation(async (method: string, _params, options) => {
+    if (method === 'pty.cancelDelivery') {
+      return { canceled: true, sentEndSu: 8, creditedEndSu: 4 }
+    }
     if (method !== 'pty.attach') {
       return undefined
     }
@@ -154,7 +160,7 @@ it('returns a provisional source activation lease to reconnect authority', async
   })
 
   const result = await provider.attachForReconnect('ssh:conn-1@@pty-1')
-  result.sourceActivationLease?.rollback()
+  await result.sourceActivationLease?.rollback()
 
   expect(Object.isFrozen(result.sourceActivation)).toBe(true)
   expect(mux.request).toHaveBeenCalledWith('pty.cancelDelivery', {

src/main/providers/ssh-pty-source-delivery-ledger.ts

@@ -0,0 +1,241 @@
+import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
+import type { PtySourceReceivingActivation } from '../../shared/pty-source-receiving-activation'
+import type { SshPtySourceFrame } from './ssh-pty-source-frame'
+
+export type PendingSshPtySourceData = Readonly<{
+  relayPtyId: string
+  params: Record<string, unknown>
+  data: string
+  source?: SshPtySourceFrame
+}>
+
+type SourceDeliveryLeaseState = {
+  phase: 'provisional' | 'committing' | 'committed' | 'retired'
+  pendingData: PendingSshPtySourceData[]
+}
+
+type SourceDeliveryState = Readonly<{
+  activation: PtySourceReceivingActivation
+  sourceEndSu: number
+  lease: SourceDeliveryLeaseState
+  previous?: SourceDeliveryState
+}>
+
+export type SshPtyReceivingActivationLease = Readonly<{
+  commit: () => void
+  rollback: () => Promise<boolean>
+}>
+
+export class SshPtySourceDeliveryLedger {
+  private readonly deliveryByPty = new Map<string, SourceDeliveryState>()
+
+  constructor(
+    private readonly mux: SshChannelMultiplexer,
+    private readonly publishData: (pending: PendingSshPtySourceData) => void
+  ) {}
+
+  install(
+    relayPtyId: string,
+    activation: PtySourceReceivingActivation
+  ): SshPtyReceivingActivationLease {
+    if (!relayPtyId || activation.ptyIncarnation.length === 0) {
+      throw new Error('ssh_source_receiving_activation_invalid')
+    }
+    const previous = this.deliveryByPty.get(relayPtyId)
+    if (previous && sameReceivingActivation(previous.activation, activation)) {
+      if (previous.lease.phase !== 'committed') {
+        throw new Error('ssh_source_receiving_activation_stale')
+      }
+      return settledReceivingActivationLease()
+    }
+    if (
+      previous &&
+      (activation.clientGeneration <= previous.activation.clientGeneration ||
+        activation.ownerGeneration <= previous.activation.ownerGeneration ||
+        activation.deliveryToken === previous.activation.deliveryToken)
+    ) {
+      throw new Error('ssh_source_receiving_activation_stale')
+    }
+    return this.installProvisional(relayPtyId, activation, previous)
+  }
+
+  admit(pending: PendingSshPtySourceData & { source: SshPtySourceFrame }): boolean {
+    const current = this.deliveryByPty.get(pending.relayPtyId)
+    if (!acceptsSourceFrame(current, pending.params, pending.source)) {
+      return false
+    }
+    const accepted = Object.freeze({
+      ...current,
+      sourceEndSu: pending.source.sourceEndSu
+    }) as SourceDeliveryState
+    this.deliveryByPty.set(pending.relayPtyId, accepted)
+    if (accepted.lease.phase !== 'committed') {
+      accepted.lease.pendingData.push(pending)
+    } else {
+      this.publishData(pending)
+    }
+    return true
+  }
+
+  recordExit(relayPtyId: string): void {
+    if (this.deliveryByPty.get(relayPtyId)?.lease.phase === 'committed') {
+      this.deliveryByPty.delete(relayPtyId)
+    }
+  }
+
+  private installProvisional(
+    relayPtyId: string,
+    activation: PtySourceReceivingActivation,
+    previous: SourceDeliveryState | undefined
+  ): SshPtyReceivingActivationLease {
+    const leaseState: SourceDeliveryLeaseState = {
+      phase: 'provisional',
+      pendingData: []
+    }
+    this.deliveryByPty.set(
+      relayPtyId,
+      Object.freeze({
+        activation,
+        sourceEndSu: activation.checkpointSourceEndSu,
+        lease: leaseState,
+        ...(previous ? { previous } : {})
+      })
+    )
+    let settled = false
+    let rollbackSettlement: Promise<boolean> | undefined
+    return Object.freeze({
+      commit: () => {
+        if (settled) {
+          return
+        }
+        settled = true
+        this.commit(relayPtyId, leaseState)
+      },
+      rollback: () => {
+        if (rollbackSettlement) {
+          return rollbackSettlement
+        }
+        if (settled) {
+          return Promise.resolve(false)
+        }
+        settled = true
+        this.retire(relayPtyId, previous, leaseState)
+        rollbackSettlement = settleExactSourceDeliveryCancellation(this.mux, relayPtyId, activation)
+        return rollbackSettlement
+      }
+    })
+  }
+
+  private commit(relayPtyId: string, lease: SourceDeliveryLeaseState): void {
+    if (this.deliveryByPty.get(relayPtyId)?.lease !== lease) {
+      lease.phase = 'retired'
+      lease.pendingData.splice(0)
+      return
+    }
+    lease.phase = 'committing'
+    while (lease.pendingData.length > 0) {
+      this.publishData(lease.pendingData.shift()!)
+    }
+    lease.phase = 'committed'
+    const current = this.deliveryByPty.get(relayPtyId)
+    if (current?.lease === lease && current.previous) {
+      this.deliveryByPty.set(
+        relayPtyId,
+        Object.freeze({
+          activation: current.activation,
+          sourceEndSu: current.sourceEndSu,
+          lease: current.lease
+        })
+      )
+    }
+  }
+
+  private retire(
+    relayPtyId: string,
+    previous: SourceDeliveryState | undefined,
+    lease: SourceDeliveryLeaseState
+  ): void {
+    lease.phase = 'retired'
+    lease.pendingData.splice(0)
+    if (this.deliveryByPty.get(relayPtyId)?.lease !== lease) {
+      return
+    }
+    const predecessor = activePredecessor(previous)
+    if (predecessor) {
+      this.deliveryByPty.set(relayPtyId, predecessor)
+    } else {
+      this.deliveryByPty.delete(relayPtyId)
+    }
+  }
+}
+
+function settledReceivingActivationLease(): SshPtyReceivingActivationLease {
+  return Object.freeze({ commit: () => {}, rollback: async () => true })
+}
+
+function activePredecessor(
+  previous: SourceDeliveryState | undefined
+): SourceDeliveryState | undefined {
+  let candidate = previous
+  while (candidate?.lease.phase === 'retired') {
+    candidate = candidate.previous
+  }
+  return candidate
+}
+
+function sameReceivingActivation(
+  left: PtySourceReceivingActivation,
+  right: PtySourceReceivingActivation
+): boolean {
+  return (
+    left.clientGeneration === right.clientGeneration &&
+    left.ownerGeneration === right.ownerGeneration &&
+    left.ptyIncarnation === right.ptyIncarnation &&
+    left.deliveryToken === right.deliveryToken &&
+    left.checkpointSourceEndSu === right.checkpointSourceEndSu &&
+    left.recoveryEndSu === right.recoveryEndSu
+  )
+}
+
+function acceptsSourceFrame(
+  current: SourceDeliveryState | undefined,
+  params: Record<string, unknown>,
+  source: SshPtySourceFrame
+): current is SourceDeliveryState {
+  return Boolean(
+    current &&
+    current.lease.phase !== 'retired' &&
+    current.activation.ptyIncarnation === params.ptyIncarnation &&
+    current.activation.deliveryToken === source.deliveryToken &&
+    current.activation.clientGeneration === source.clientGeneration &&
+    current.activation.ownerGeneration === source.ownerGeneration &&
+    current.sourceEndSu === source.sourceStartSu
+  )
+}
+
+async function settleExactSourceDeliveryCancellation(
+  mux: SshChannelMultiplexer,
+  relayPtyId: string,
+  activation: PtySourceReceivingActivation
+): Promise<boolean> {
+  try {
+    const result = (await mux.request('pty.cancelDelivery', {
+      id: relayPtyId,
+      clientGeneration: activation.clientGeneration,
+      ownerGeneration: activation.ownerGeneration,
+      deliveryToken: activation.deliveryToken
+    })) as Record<string, unknown>
+    return (
+      result.canceled === true &&
+      nonNegativeSafeInteger(result.sentEndSu) &&
+      nonNegativeSafeInteger(result.creditedEndSu) &&
+      result.creditedEndSu <= result.sentEndSu
+    )
+  } catch {
+    return false
+  }
+}
+
+function nonNegativeSafeInteger(value: unknown): value is number {
+  return Number.isSafeInteger(value) && (value as number) >= 0
+}