fix(ssh): make PTY owner admission explicit and non-destructive

An owner-capable `pty.openClient` had two ways to fail that looked like
something else. If the relay still held an owner record but the request
carried no matching resume proof, admission fell through to a *subscriber*
grant — a success-shaped response the client cannot use, which it then
rejected with "did not grant an authenticated PTY session owner". And if
the relay no longer held the record the client named, admission threw a
stale-recovery error, which the client answered by deleting its own
recovery row and reopening: two round trips, and the deletion took the
`clientInstanceId` with it.

Admission now returns an explicit outcome. Every `session-owner` grant
carries `resumed`, so a client learns in one round trip whether its
checkpoints survived. A resume proof for a record the relay has forgotten
is not an error — it mints a fresh claim with `resumed: false`. A held
claim produces one of two coded refusals instead of a subscriber grant:
`PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR` is blocked and exits retry at
once, `PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR` is transient and
retried under its own bounded budget, separate from the publication-pending
one so a settling publication cannot spend it.

Nothing here destroys state on suspicion. A refusal mutates no checkpoint,
owner, migration, or persisted recovery state, and `resumed: false` voids
only the checkpoints taken under the lease the relay just replaced — the
recovery row and `clientInstanceId` survive, guarded so a losing attempt
cannot void a winner's. Shortening a disconnected incumbent's grace to a
250ms floor is gated on positive proof: that record reaches 'disconnected'
only from an observed transport close, expiring it transfers admission
rather than stopping any remote PTY, and the incumbent's own matching proof
still wins outright because it never reaches that branch.

A duplicate `pty.openClient` on one connection is now rejected even when
identical. Two responses settle their publications independently, so one
shared admission cannot make one response's rollback and the other's commit
atomic; recovery from an RPC timeout opens a new connection instead.

An attached-holder refusal is converted to a typed
`SshOwnerAdmissionBlockedError` and routed through the existing terminal
relay-error callback rather than transport reconnect classification — the
relay answered and refused on purpose, so redeploy backoff would spend its
whole budget on a link that is working.

Verified by mutation: reverting each behavior turns a named test red —
vacant-resume grant (1), held-owner refusals (5), grace clamp (1),
duplicate rejection (2), `resumed` validation (4), refusal-triggered row
deletion (1), the losing-attempt guard (1), attached-holder retry (2),
the split retry budgets (1), and the blocked-error routing (1). Negative
tests assert live sessions survive: a resumed claim keeps its checkpoints,
the incumbent reclaims its lease during the shortened floor, a newer
owner's record is untouched by a stale attempt, and a blocked refusal
removes no recovery state.

Refs STA-3365.

src/main/ipc/ssh.test.ts

@@ -421,6 +421,7 @@ describe('SSH IPC handlers', () => {
               role: 'session-owner',
               ownerGeneration: 1,
               ownerLease: 'ipc-test-owner',
+              resumed: false,
               capabilities: {
                 outputFlowControl: { version: 1, windowSu: DEFAULT_PTY_SOURCE_WINDOW_SU }
               }

src/main/ssh/ssh-owner-admission-blocked-error.ts

@@ -0,0 +1,21 @@
+// Why a typed error instead of raw transport classification: another authenticated connection holds
+// the PTY session owner claim, and no relay redeploy or backoff attempt can reconcile that. The
+// reconnect ladder has to stop and say what actually happened rather than blame the link.
+
+export class SshOwnerAdmissionBlockedError extends Error {
+  readonly name = 'SshOwnerAdmissionBlockedError'
+
+  constructor(targetId: string, options?: { cause?: unknown }) {
+    super(
+      `Another connection currently owns the remote terminals for ${targetId}. ` +
+        `Disconnect that client, or wait for it to release ownership, before reconnecting.`,
+      options
+    )
+  }
+}
+
+export function isSshOwnerAdmissionBlockedError(
+  err: unknown
+): err is SshOwnerAdmissionBlockedError {
+  return err instanceof SshOwnerAdmissionBlockedError
+}

src/main/ssh/ssh-owner-recovery-retry.test.ts

@@ -1,9 +1,15 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import {
+  PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR
 } from '../../shared/pty-consumer-session'
-import { retrySshOwnerRecoveryWhileBlocked } from './ssh-owner-recovery-retry'
+import {
+  isSshOwnerAdmissionBlocked,
+  retrySshOwnerRecoveryWhileBlocked,
+  SSH_OWNER_HELD_DISCONNECTED_WAIT_MS
+} from './ssh-owner-recovery-retry'
 
 function publicationPendingError(): Error & { code: number } {
   return Object.assign(new Error('Owner grant publication is still pending'), {
@@ -17,6 +23,10 @@ function supersededError(): Error & { code: number } {
   })
 }
 
+function heldError(code: number): Error & { code: number } {
+  return Object.assign(new Error('PTY session owner is held'), { code })
+}
+
 function openGate() {
   return {
     isCurrent: () => true,
@@ -79,6 +89,82 @@ describe('SSH owner recovery retry', () => {
     expect(attempt).toHaveBeenCalledTimes(2)
   })
 
+  it('retries a disconnected holder until it releases admission', async () => {
+    vi.useFakeTimers()
+    const attempt = vi
+      .fn<() => Promise<string>>()
+      .mockRejectedValueOnce(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR))
+      .mockRejectedValueOnce(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR))
+      .mockResolvedValue('recovered')
+
+    const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate())
+    await vi.advanceTimersByTimeAsync(75)
+
+    await expect(recovery).resolves.toBe('recovered')
+    expect(attempt).toHaveBeenCalledTimes(3)
+  })
+
+  it('never retries an attached holder', async () => {
+    const error = heldError(PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR)
+    const attempt = vi.fn<() => Promise<never>>().mockRejectedValue(error)
+
+    // Why blocked, not transient: another connection is live on the claim, so no amount of waiting
+    // inside this admission changes the answer.
+    await expect(retrySshOwnerRecoveryWhileBlocked(attempt, openGate())).rejects.toBe(error)
+    expect(attempt).toHaveBeenCalledOnce()
+    expect(isSshOwnerAdmissionBlocked(error)).toBe(true)
+    expect(isSshOwnerAdmissionBlocked(publicationPendingError())).toBe(false)
+  })
+
+  it('gives a disconnected holder its own budget rather than the publication one', async () => {
+    vi.useFakeTimers()
+    let failures = 0
+    const attempt = vi.fn<() => Promise<string>>().mockImplementation(async () => {
+      if (failures++ < 6) {
+        throw heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR)
+      }
+      return 'recovered'
+    })
+
+    // Why a 60ms publication budget: six backoff waits run well past it, so a single shared deadline
+    // would give up before the incumbent's clamped grace floor could ever elapse.
+    const recovery = retrySshOwnerRecoveryWhileBlocked(attempt, openGate(), 60)
+    await vi.advanceTimersByTimeAsync(SSH_OWNER_HELD_DISCONNECTED_WAIT_MS)
+
+    await expect(recovery).resolves.toBe('recovered')
+    expect(attempt).toHaveBeenCalledTimes(7)
+  })
+
+  it('reports each exhausted retry budget under its own reason', async () => {
+    vi.useFakeTimers()
+    const exhausted: string[] = []
+    const gate = { ...openGate(), onRetryExhausted: (reason: string) => exhausted.push(reason) }
+
+    const pending = retrySshOwnerRecoveryWhileBlocked(
+      vi.fn<() => Promise<never>>().mockRejectedValue(publicationPendingError()),
+      gate,
+      60
+    )
+    const pendingRejection = expect(pending).rejects.toThrow('publication')
+    await vi.advanceTimersByTimeAsync(60)
+    await pendingRejection
+
+    const held = retrySshOwnerRecoveryWhileBlocked(
+      vi
+        .fn<() => Promise<never>>()
+        .mockRejectedValue(heldError(PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR)),
+      gate,
+      60
+    )
+    const heldRejection = expect(held).rejects.toThrow('held')
+    // Why the disconnected budget and not the 60ms one: each reason carries its own deadline so a
+    // settling publication cannot spend the budget that waits out a grace floor.
+    await vi.advanceTimersByTimeAsync(SSH_OWNER_HELD_DISCONNECTED_WAIT_MS)
+    await heldRejection
+
+    expect(exhausted).toEqual(['publication-pending', 'disconnected-holder'])
+  })
+
   it('stops waiting when the relay channel closes', async () => {
     vi.useFakeTimers()
     let current = true

src/main/ssh/ssh-owner-recovery-retry.ts

@@ -1,16 +1,46 @@
 import {
+  PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR
 } from '../../shared/pty-consumer-session'
 
 // Why: bound polling when publication is settling or a superseded attempt is closing its transport.
 export const SSH_OWNER_RECOVERY_WAIT_MS = 3_000
+// Why separate and longer than the relay's grace floor: a disconnected incumbent releases admission
+// only after that floor elapses, so this budget must outlast it without borrowing the pending budget.
+export const SSH_OWNER_HELD_DISCONNECTED_WAIT_MS = 2_000
+
+export type SshOwnerRecoveryRetryReason = 'publication-pending' | 'disconnected-holder'
+
+// Why exported: an attached holder is a decision, not a transport fault, so callers must be able to
+// report it as blocked instead of feeding it to reconnect classification as an unexplained failure.
+export function isSshOwnerAdmissionBlocked(error: unknown): boolean {
+  return (
+    (error as { code?: unknown } | null | undefined)?.code ===
+    PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR
+  )
+}
+
+function retryReasonFor(error: unknown): SshOwnerRecoveryRetryReason | null {
+  const code = (error as { code?: unknown } | null | undefined)?.code
+  if (
+    code === PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR ||
+    code === PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR
+  ) {
+    return 'publication-pending'
+  }
+  // Why an attached holder is deliberately absent: it is blocked, not transient — retrying cannot
+  // change the answer while another connection is live on the claim.
+  return code === PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR ? 'disconnected-holder' : null
+}
 const SSH_OWNER_RECOVERY_INITIAL_DELAY_MS = 25
 const SSH_OWNER_RECOVERY_MAX_DELAY_MS = 250
 
 type SshOwnerRecoveryRetryGate = {
   isCurrent: () => boolean
   onClosed: (listener: () => void) => () => void
+  onRetryExhausted?: (reason: SshOwnerRecoveryRetryReason) => void
 }
 
 function waitForRetry(delayMs: number, gate: SshOwnerRecoveryRetryGate): Promise<void> {
@@ -41,22 +71,22 @@ export async function retrySshOwnerRecoveryWhileBlocked<T>(
   gate: SshOwnerRecoveryRetryGate,
   waitMs: number = SSH_OWNER_RECOVERY_WAIT_MS
 ): Promise<T> {
-  const deadline = Date.now() + waitMs
+  const deadlineByReason: Record<SshOwnerRecoveryRetryReason, number> = {
+    'publication-pending': Date.now() + waitMs,
+    'disconnected-holder': Date.now() + SSH_OWNER_HELD_DISCONNECTED_WAIT_MS
+  }
   let delayMs = SSH_OWNER_RECOVERY_INITIAL_DELAY_MS
   while (true) {
     try {
       return await attempt()
     } catch (error) {
-      const code = (error as { code?: unknown } | null | undefined)?.code
-      if (
-        (code !== PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR &&
-          code !== PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR) ||
-        !gate.isCurrent()
-      ) {
+      const reason = retryReasonFor(error)
+      if (reason === null || !gate.isCurrent()) {
         throw error
       }
-      const remainingMs = deadline - Date.now()
+      const remainingMs = deadlineByReason[reason] - Date.now()
       if (remainingMs <= 0) {
+        gate.onRetryExhausted?.(reason)
         throw error
       }
       await waitForRetry(Math.min(delayMs, remainingMs), gate)

src/main/ssh/ssh-pty-consumer-session.test.ts

@@ -18,6 +18,7 @@ function legacyOwnerGrant(overrides: Record<string, unknown> = {}): Record<strin
     role: 'session-owner',
     ownerGeneration: 7,
     ownerLease: 'lease-a',
+    resumed: false,
     ...overrides
   }
 }
@@ -32,11 +33,14 @@ describe('openSshPtyConsumerSession', () => {
         expectedServerBuildId: 'build-a'
       })
     ).resolves.toEqual({
-      mode: 'negotiated',
-      clientInstanceId: 'client-a',
-      clientGeneration: 3,
-      ownerGeneration: 7,
-      ownerLease: 'lease-a'
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: 'client-a',
+        clientGeneration: 3,
+        ownerGeneration: 7,
+        ownerLease: 'lease-a'
+      },
+      resumed: false
     })
     expect(request).toHaveBeenCalledWith(
       'pty.openClient',
@@ -51,9 +55,9 @@ describe('openSshPtyConsumerSession', () => {
 
   it('carries recovery generation and lease on reconnect', async () => {
     const { mux, request } = muxReturning(
-      legacyOwnerGrant({ ownerGeneration: 8, ownerLease: 'lease-b' })
+      legacyOwnerGrant({ ownerGeneration: 8, ownerLease: 'lease-a', resumed: true })
     )
-    await openSshPtyConsumerSession(mux, {
+    const admission = await openSshPtyConsumerSession(mux, {
       clientInstanceId: 'client-a',
       expectedServerBuildId: 'build-a',
       resume: { ownerGeneration: 7, ownerLease: 'lease-a' }
@@ -62,8 +66,30 @@ describe('openSshPtyConsumerSession', () => {
     expect(request.mock.calls[0][1]).toMatchObject({
       resume: { ownerGeneration: 7, ownerLease: 'lease-a' }
     })
+    expect(admission.resumed).toBe(true)
   })
 
+  it.each([undefined, 'yes', 1, null])(
+    'rejects an owner grant that does not state whether the claim was resumed',
+    async (resumed) => {
+      const grant = legacyOwnerGrant()
+      // Why not a legacy peer: the build id already matched, and client and relay ship together.
+      if (resumed === undefined) {
+        delete grant.resumed
+      } else {
+        grant.resumed = resumed
+      }
+      const { mux } = muxReturning(grant)
+
+      await expect(
+        openSshPtyConsumerSession(mux, {
+          clientInstanceId: 'client-a',
+          expectedServerBuildId: 'build-a'
+        })
+      ).rejects.toThrow('whether the claim was resumed')
+    }
+  )
+
   it('rejects a prior or mismatched relay build', async () => {
     const { mux } = muxReturning(legacyOwnerGrant({ serverBuildId: 'old-build' }))
 
@@ -126,9 +152,12 @@ describe('openSshPtyConsumerSession', () => {
         outputFlowControl: { requestedWindowSu: 64 }
       })
     ).resolves.toEqual({
-      mode: 'legacy-fallback',
-      clientInstanceId: 'client-a',
-      serverBuildId: 'build-a'
+      state: {
+        mode: 'legacy-fallback',
+        clientInstanceId: 'client-a',
+        serverBuildId: 'build-a'
+      },
+      resumed: false
     })
   })
 

src/main/ssh/ssh-pty-consumer-session.ts

@@ -27,6 +27,13 @@ export type SshPtyLegacyFallbackState = {
 
 export type SshPtyConsumerSessionState = SshPtyConsumerOwnerState | SshPtyLegacyFallbackState
 
+export type SshPtyConsumerAdmission = {
+  state: SshPtyConsumerSessionState
+  // Why not on the owner state itself: this describes one admission's outcome, not the persisted
+  // claim, and it must never round-trip through the recovery record.
+  resumed: boolean
+}
+
 export type OpenSshPtyConsumerSessionOptions = {
   clientInstanceId: string
   expectedServerBuildId: string | undefined
@@ -68,6 +75,11 @@ function validateGrant(
   ) {
     throw new Error('Remote relay did not grant an authenticated PTY session owner')
   }
+  // Why not treated as a legacy relay: client and relay ship in one build, and the build id was already
+  // matched above — a missing `resumed` here is corruption, not an older peer.
+  if (typeof grant.resumed !== 'boolean') {
+    throw new Error('Remote relay owner grant did not state whether the claim was resumed')
+  }
   const requestedFlow = options.outputFlowControl
   const grantedFlow = grant.capabilities?.outputFlowControl
   if (requestedFlow) {
@@ -88,7 +100,7 @@ function validateGrant(
 export async function openSshPtyConsumerSession(
   mux: SshChannelMultiplexer,
   options: OpenSshPtyConsumerSessionOptions
-): Promise<SshPtyConsumerSessionState> {
+): Promise<SshPtyConsumerAdmission> {
   let result: unknown
   try {
     result = await mux.request(
@@ -119,23 +131,29 @@ export async function openSshPtyConsumerSession(
       typeof options.expectedServerBuildId === 'string' &&
       options.expectedServerBuildId.length > 0
     ) {
-      return Object.freeze({
-        mode: 'legacy-fallback',
-        clientInstanceId: options.clientInstanceId,
-        serverBuildId: options.expectedServerBuildId
-      })
+      return {
+        state: Object.freeze({
+          mode: 'legacy-fallback',
+          clientInstanceId: options.clientInstanceId,
+          serverBuildId: options.expectedServerBuildId
+        }),
+        resumed: false
+      }
     }
     throw error
   }
   const grant = validateGrant(result, options)
   return {
-    mode: 'negotiated',
-    clientInstanceId: options.clientInstanceId,
-    clientGeneration: grant.clientGeneration,
-    ownerGeneration: grant.ownerGeneration!,
-    ownerLease: grant.ownerLease!,
-    ...(grant.capabilities?.outputFlowControl
-      ? { outputFlowControl: grant.capabilities.outputFlowControl }
-      : {})
+    state: {
+      mode: 'negotiated',
+      clientInstanceId: options.clientInstanceId,
+      clientGeneration: grant.clientGeneration,
+      ownerGeneration: grant.ownerGeneration!,
+      ownerLease: grant.ownerLease!,
+      ...(grant.capabilities?.outputFlowControl
+        ? { outputFlowControl: grant.capabilities.outputFlowControl }
+        : {})
+    },
+    resumed: grant.resumed!
   }
 }

src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts

@@ -111,6 +111,7 @@ function createFakeRelay(): FakeRelay {
         ? (params.resume as { ownerGeneration: number }).ownerGeneration + 1
         : 1,
     ownerLease: 'test-owner-lease',
+    resumed: params.resume !== undefined,
     capabilities: {
       outputFlowControl: { version: 1, windowSu: DEFAULT_PTY_SOURCE_WINDOW_SU }
     }

src/main/ssh/ssh-relay-session-data-delivery.test.ts

@@ -1,10 +1,6 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { SshRelaySession } from './ssh-relay-session'
-import {
-  createMismatchedOwnerRecoveryError,
-  createMockDeps,
-  mockDeploySuccess
-} from './ssh-relay-session-test-fixtures'
+import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
 
 const {
   acceptOutputDataMock,
@@ -145,16 +141,22 @@ describe('SshRelaySession data delivery', () => {
     vi.mocked(applySshPtySourceCancellationProof).mockReturnValue(true)
     vi.mocked(applySshPtySourceRecoveryCancellationProof).mockReturnValue(true)
     openConsumerSessionMock.mockImplementation(async (_mux, options) => ({
-      mode: 'negotiated',
-      clientInstanceId: options.clientInstanceId,
-      clientGeneration: 1,
-      ownerGeneration: 1,
-      ownerLease: 'test-owner-lease',
-      ...(options.outputFlowControl
-        ? {
-            outputFlowControl: { version: 1, windowSu: options.outputFlowControl.requestedWindowSu }
-          }
-        : {})
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: options.clientInstanceId,
+        clientGeneration: 1,
+        ownerGeneration: 1,
+        ownerLease: 'test-owner-lease',
+        ...(options.outputFlowControl
+          ? {
+              outputFlowControl: {
+                version: 1,
+                windowSu: options.outputFlowControl.requestedWindowSu
+              }
+            }
+          : {})
+      },
+      resumed: options.resume !== undefined
     }))
     muxRequestMock.mockResolvedValue([])
     mockDeploySuccess()
@@ -173,12 +175,15 @@ describe('SshRelaySession data delivery', () => {
     )
     let generation = 0
     openConsumerSessionMock.mockImplementation(async (_mux, options) => ({
-      mode: 'negotiated',
-      clientInstanceId: options.clientInstanceId,
-      clientGeneration: ++generation,
-      ownerGeneration: generation,
-      ownerLease: `owner-lease-${generation}`,
-      outputFlowControl: { version: 1, windowSu: 256 * 1024 }
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: options.clientInstanceId,
+        clientGeneration: ++generation,
+        ownerGeneration: generation,
+        ownerLease: `owner-lease-${generation}`,
+        outputFlowControl: { version: 1, windowSu: 256 * 1024 }
+      },
+      resumed: options.resume !== undefined
     }))
     vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([
       {
@@ -342,7 +347,7 @@ describe('SshRelaySession data delivery', () => {
     expect(mockStore.removeSshPtyConsumerRecovery).toHaveBeenCalledWith(targetId)
   })
 
-  it('clears stale owner recovery and retries a fresh same-build relay once without resume', async () => {
+  it('voids checkpoints for a fresh claim without a second owner request', async () => {
     const targetId = 'fresh-relay-retry'
     const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
     vi.mocked(deployAndLaunchRelay).mockResolvedValue({
@@ -365,16 +370,17 @@ describe('SshRelaySession data delivery', () => {
     await first.establish(mockConn)
     first.detach()
 
-    openConsumerSessionMock
-      .mockRejectedValueOnce(createMismatchedOwnerRecoveryError())
-      .mockImplementationOnce(async (_mux, options) => ({
+    openConsumerSessionMock.mockImplementationOnce(async (_mux, options) => ({
+      state: {
         mode: 'negotiated',
         clientInstanceId: options.clientInstanceId,
         clientGeneration: 1,
         ownerGeneration: 1,
         ownerLease: 'fresh-owner-lease',
         outputFlowControl: { version: 1, windowSu: 256 * 1024 }
-      }))
+      },
+      resumed: false
+    }))
     vi.mocked(getPtyIdsForConnection).mockReturnValue([`ssh:${targetId}@@pty-1`])
     vi.mocked(getSshPtyProvider).mockImplementation(
       () => vi.mocked(registerSshPtyProvider).mock.calls.at(-1)?.[1]
@@ -391,9 +397,10 @@ describe('SshRelaySession data delivery', () => {
     const retryCalls = openConsumerSessionMock.mock.calls
       .slice(openCallCountBeforeRetry)
       .map(([, options]) => options)
-    expect(retryCalls).toHaveLength(2)
+    // Why one call: the relay answers a proof it cannot match with a fresh claim, so the client never
+    // needs a second, resume-less request to get owner authority back.
+    expect(retryCalls).toHaveLength(1)
     expect(retryCalls[0]).toHaveProperty('resume')
-    expect(retryCalls[1]).not.toHaveProperty('resume')
     expect(attachForReconnectMock).toHaveBeenCalledWith(
       'pty-1',
       undefined,
@@ -644,12 +651,15 @@ describe('SshRelaySession data delivery', () => {
     openConsumerSessionMock.mockImplementation(async (_mux, options) => {
       generation++
       return {
-        mode: 'negotiated',
-        clientInstanceId: options.clientInstanceId,
-        clientGeneration: generation,
-        ownerGeneration: generation,
-        ownerLease: `owner-lease-${generation}`,
-        outputFlowControl: { version: 1, windowSu: 256 * 1024 }
+        state: {
+          mode: 'negotiated',
+          clientInstanceId: options.clientInstanceId,
+          clientGeneration: generation,
+          ownerGeneration: generation,
+          ownerLease: `owner-lease-${generation}`,
+          outputFlowControl: { version: 1, windowSu: 256 * 1024 }
+        },
+        resumed: options.resume !== undefined
       }
     })
     vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([

src/main/ssh/ssh-relay-session-model-migration.test.ts

@@ -1,10 +1,6 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { SshRelaySession } from './ssh-relay-session'
-import {
-  createMismatchedOwnerRecoveryError,
-  createMockDeps,
-  mockDeploySuccess
-} from './ssh-relay-session-test-fixtures'
+import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
 
 type SettledMigration = {
   status: 'settled'
@@ -135,16 +131,22 @@ describe('SshRelaySession model migration', () => {
     vi.mocked(getPtyIdsForConnection).mockReturnValue([])
     vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([])
     openConsumerSessionMock.mockImplementation(async (_mux, options) => ({
-      mode: 'negotiated',
-      clientInstanceId: options.clientInstanceId,
-      clientGeneration: 1,
-      ownerGeneration: 1,
-      ownerLease: 'owner-lease-1',
-      ...(options.outputFlowControl
-        ? {
-            outputFlowControl: { version: 1, windowSu: options.outputFlowControl.requestedWindowSu }
-          }
-        : {})
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: options.clientInstanceId,
+        clientGeneration: 1,
+        ownerGeneration: 1,
+        ownerLease: 'owner-lease-1',
+        ...(options.outputFlowControl
+          ? {
+              outputFlowControl: {
+                version: 1,
+                windowSu: options.outputFlowControl.requestedWindowSu
+              }
+            }
+          : {})
+      },
+      resumed: options.resume !== undefined
     }))
     muxRequestMock.mockResolvedValue([])
     mockDeploySuccess()
@@ -209,7 +211,7 @@ describe('SshRelaySession model migration', () => {
     )
   })
 
-  it('waits for a stale-owner migration before requesting restore', async () => {
+  it('waits for a voided-checkpoint migration before requesting restore', async () => {
     const targetId = 'migration-stale-owner'
     const appPtyId = `ssh:${targetId}@@pty-1`
     const migration = pendingMigration()
@@ -232,23 +234,24 @@ describe('SshRelaySession model migration', () => {
     vi.mocked(getSshPtyProvider).mockImplementation(
       () => vi.mocked(registerSshPtyProvider).mock.calls.at(-1)?.[1]
     )
-    openConsumerSessionMock
-      .mockRejectedValueOnce(createMismatchedOwnerRecoveryError())
-      .mockImplementationOnce(async (_mux, options) => ({
+    openConsumerSessionMock.mockImplementationOnce(async (_mux, options) => ({
+      state: {
         mode: 'negotiated',
         clientInstanceId: options.clientInstanceId,
         clientGeneration: 2,
         ownerGeneration: 2,
         ownerLease: 'fresh-owner-lease',
         outputFlowControl: { version: 1, windowSu: 256 * 1024 }
-      }))
+      },
+      resumed: false
+    }))
     attachForReconnectMock.mockResolvedValue({
       incarnationId: 'incarnation-1',
       sourceRecovery: { status: 'restoreRequired', reason: 'checkpointUnavailable' }
     })
 
     const reconnect = session.reconnect(deps.mockConn)
-    await vi.waitFor(() => expect(openConsumerSessionMock).toHaveBeenCalledTimes(3))
+    await vi.waitFor(() => expect(openConsumerSessionMock).toHaveBeenCalledTimes(2))
     expect(attachForReconnectMock).not.toHaveBeenCalled()
 
     migration.resolve(settledMigration(appPtyId, 8))

src/main/ssh/ssh-relay-session-recovery-durability.test.ts

@@ -1,13 +1,11 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import {
+  PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR
 } from '../../shared/pty-consumer-session'
 import { SshRelaySession } from './ssh-relay-session'
-import {
-  createMismatchedOwnerRecoveryError,
-  createMockDeps
-} from './ssh-relay-session-test-fixtures'
+import { createMockDeps } from './ssh-relay-session-test-fixtures'
 import { getSshPtyConsumerRecovery } from './ssh-pty-consumer-recovery'
 
 const { muxRequestMock, openConsumerSessionMock } = vi.hoisted(() => ({
@@ -108,11 +106,14 @@ describe('SshRelaySession consumer recovery durability', () => {
     vi.clearAllMocks()
     muxRequestMock.mockResolvedValue([])
     openConsumerSessionMock.mockImplementation(async (_mux, options) => ({
-      mode: 'negotiated',
-      clientInstanceId: options.clientInstanceId,
-      clientGeneration: 1,
-      ownerGeneration: 1,
-      ownerLease: 'test-owner-lease'
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: options.clientInstanceId,
+        clientGeneration: 1,
+        ownerGeneration: 1,
+        ownerLease: 'test-owner-lease'
+      },
+      resumed: false
     }))
     vi.mocked(deployAndLaunchRelay).mockResolvedValue({
       transport: { write: vi.fn(), onData: vi.fn(), onClose: vi.fn() },
@@ -277,8 +278,38 @@ describe('SshRelaySession consumer recovery durability', () => {
     expect(completed).toBe(true)
   })
 
-  it('does not retry a stale owner after disposal wins the recovery-removal race', async () => {
-    const targetId = 'target-stale-owner-disposal'
+  it('reports an attached owner as terminal instead of a lost relay', async () => {
+    const targetId = 'target-owner-held-attached'
+    const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
+    const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward)
+    const onTerminal = vi.fn()
+    const onRelayLost = vi.fn()
+    session.setOnTerminalRelayError(onTerminal)
+    session.setOnRelayLost(onRelayLost)
+    await session.establish(mockConn)
+
+    openConsumerSessionMock.mockRejectedValueOnce(
+      Object.assign(new Error('PTY session owner is held by an attached connection'), {
+        code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR
+      })
+    )
+    await session.reconnect(mockConn)
+
+    // Why not relay-lost: the relay answered and refused on purpose, so redeploy backoff would spend
+    // its whole budget on a link that is working.
+    expect(onRelayLost).not.toHaveBeenCalled()
+    expect(onTerminal).toHaveBeenCalledWith(
+      targetId,
+      expect.objectContaining({ name: 'SshOwnerAdmissionBlockedError' })
+    )
+    expect(onTerminal.mock.calls[0]?.[1]?.message).toContain('owns the remote terminals')
+    expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled()
+    expect(openConsumerSessionMock).toHaveBeenCalledTimes(2)
+    session.dispose()
+  })
+
+  it('recovers a forgotten owner record in one request without deleting recovery identity', async () => {
+    const targetId = 'target-forgotten-owner-record'
     const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
     vi.mocked(mockStore.getSshPtyConsumerRecovery).mockReturnValue({
       targetId,
@@ -286,30 +317,75 @@ describe('SshRelaySession consumer recovery durability', () => {
       serverBuildId: 'test-relay-build',
       clientGeneration: 1,
       ownerGeneration: 1,
-      ownerLease: 'stale-owner'
+      ownerLease: 'forgotten-owner'
     })
-    let signalRemovalStarted!: () => void
-    const removalStarted = new Promise<void>((resolve) => {
-      signalRemovalStarted = resolve
-    })
-    let settleRemoval!: () => void
-    vi.mocked(mockStore.removeSshPtyConsumerRecovery).mockImplementationOnce(() => {
-      signalRemovalStarted()
-      return new Promise<void>((resolve) => {
-        settleRemoval = resolve
-      })
-    })
-    openConsumerSessionMock.mockRejectedValueOnce(createMismatchedOwnerRecoveryError())
+    openConsumerSessionMock.mockImplementationOnce(async () => ({
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: 'persisted-client',
+        clientGeneration: 4,
+        ownerGeneration: 9,
+        ownerLease: 'fresh-owner'
+      },
+      resumed: false
+    }))
     const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward)
+    getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.set('pty-1', {
+      id: 'pty-1'
+    } as unknown as never)
 
-    const establishing = session.establish(mockConn)
-    const failed = expect(establishing).rejects.toThrow('Session disposed during establish')
-    await removalStarted
-    const disposal = session.disposeAndPersist()
-    settleRemoval()
-    await Promise.all([failed, disposal])
+    await session.establish(mockConn)
 
     expect(openConsumerSessionMock).toHaveBeenCalledTimes(1)
+    expect(openConsumerSessionMock.mock.calls[0]?.[1]).toMatchObject({
+      resume: { ownerGeneration: 1, ownerLease: 'forgotten-owner' }
+    })
+    // Why both: the fresh claim voids checkpoints taken under the old lease, but the recovery identity
+    // is what lets this client keep resuming the target at all — a refusal must never cost it.
+    expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.size).toBe(0)
+    expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled()
+    expect(mockStore.upsertSshPtyConsumerRecovery).toHaveBeenCalledWith(
+      expect.objectContaining({
+        targetId,
+        clientInstanceId: 'persisted-client',
+        ownerGeneration: 9,
+        ownerLease: 'fresh-owner'
+      })
+    )
+    session.dispose()
+  })
+
+  it('keeps checkpoints when a resumed claim comes back', async () => {
+    const targetId = 'target-resumed-owner-record'
+    const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
+    vi.mocked(mockStore.getSshPtyConsumerRecovery).mockReturnValue({
+      targetId,
+      clientInstanceId: 'persisted-client',
+      serverBuildId: 'test-relay-build',
+      clientGeneration: 1,
+      ownerGeneration: 1,
+      ownerLease: 'persisted-owner'
+    })
+    openConsumerSessionMock.mockImplementationOnce(async () => ({
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: 'persisted-client',
+        clientGeneration: 4,
+        ownerGeneration: 2,
+        ownerLease: 'persisted-owner'
+      },
+      resumed: true
+    }))
+    const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward)
+    getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.set('pty-1', {
+      id: 'pty-1'
+    } as unknown as never)
+
+    await session.establish(mockConn)
+
+    expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.has('pty-1')).toBe(true)
+    expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled()
+    session.dispose()
   })
 
   it('leaves recovery state alone when a newer owner already claimed the target record', async () => {
@@ -330,28 +406,84 @@ describe('SshRelaySession consumer recovery durability', () => {
       ownerGeneration: 5,
       ownerLease: 'winner-owner'
     }
-    openConsumerSessionMock.mockImplementationOnce(() => {
-      // Why inside the rejection: the record is target-scoped, so the winner can land while this
-      // attempt is still unwinding its own resume.
+    openConsumerSessionMock.mockImplementationOnce(async () => {
+      // Why inside the open: the record is target-scoped, so the winner can land while this attempt is
+      // still waiting on its own resume, leaving this attempt's `previousOwner` snapshot stale.
       const record = getSshPtyConsumerRecovery(targetId)!
       record.owner = winner
-      record.checkpointsByAppPtyId.set('pty-1', {
-        id: 'pty-1'
-      } as unknown as never)
-      return Promise.reject(createMismatchedOwnerRecoveryError())
+      record.checkpointsByAppPtyId.set('pty-1', { id: 'pty-1' } as unknown as never)
+      return {
+        state: {
+          mode: 'negotiated',
+          clientInstanceId: 'persisted-client',
+          clientGeneration: 3,
+          ownerGeneration: 1,
+          ownerLease: 'loser-owner'
+        },
+        resumed: false
+      }
     })
-    openConsumerSessionMock.mockRejectedValueOnce(new Error('fresh open failed'))
     const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward)
 
-    await expect(session.establish(mockConn)).rejects.toThrow('fresh open failed')
+    await session.establish(mockConn)
 
-    const record = getSshPtyConsumerRecovery(targetId)!
-    expect(record.owner).toBe(winner)
-    expect(record.checkpointsByAppPtyId.has('pty-1')).toBe(true)
+    expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.has('pty-1')).toBe(true)
     expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled()
     session.dispose()
   })
 
+  it('mutates no recovery state after a local attempt loses its authority', async () => {
+    const targetId = 'target-owner-attempt-superseded'
+    const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
+    vi.mocked(mockStore.getSshPtyConsumerRecovery).mockReturnValue({
+      targetId,
+      clientInstanceId: 'persisted-client',
+      serverBuildId: 'test-relay-build',
+      clientGeneration: 1,
+      ownerGeneration: 1,
+      ownerLease: 'persisted-owner'
+    })
+    let signalOpenStarted!: () => void
+    const openStarted = new Promise<void>((resolve) => {
+      signalOpenStarted = resolve
+    })
+    let finishOpen!: (value: unknown) => void
+    openConsumerSessionMock.mockImplementationOnce(() => {
+      signalOpenStarted()
+      return new Promise((resolve) => {
+        finishOpen = resolve
+      })
+    })
+    const session = new SshRelaySession(targetId, getMainWindow, mockStore, mockPortForward)
+    getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.set('pty-1', {
+      id: 'pty-1'
+    } as unknown as never)
+
+    const superseded = session.establish(mockConn)
+    const failed = expect(superseded).rejects.toThrow('Session disposed during establish')
+    await openStarted
+    // Why detach and not disposal: disposal legitimately removes the record, which would hide whether
+    // the attempt that lost its authority mutated anything on its way out.
+    await session.detachAndPersist()
+    finishOpen({
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: 'persisted-client',
+        clientGeneration: 3,
+        ownerGeneration: 3,
+        ownerLease: 'superseded-owner'
+      },
+      resumed: false
+    })
+    await failed
+
+    expect(getSshPtyConsumerRecovery(targetId)!.checkpointsByAppPtyId.has('pty-1')).toBe(true)
+    expect(mockStore.upsertSshPtyConsumerRecovery).not.toHaveBeenCalledWith(
+      expect.objectContaining({ ownerLease: 'superseded-owner' })
+    )
+    expect(mockStore.removeSshPtyConsumerRecovery).not.toHaveBeenCalled()
+  })
+
   it('does not remember a consumer opened after establish was disposed', async () => {
     const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
     let signalOpenStarted!: () => void
@@ -377,11 +509,14 @@ describe('SshRelaySession consumer recovery durability', () => {
     await openStarted
     await session.disposeAndPersist()
     finishOpen({
-      mode: 'negotiated',
-      clientInstanceId: 'late-client',
-      clientGeneration: 1,
-      ownerGeneration: 1,
-      ownerLease: 'late-owner'
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: 'late-client',
+        clientGeneration: 1,
+        ownerGeneration: 1,
+        ownerLease: 'late-owner'
+      },
+      resumed: false
     })
     await failed
 

src/main/ssh/ssh-relay-session-recovery-races.test.ts

@@ -191,12 +191,15 @@ describe('SshRelaySession recovery race fencing', () => {
   }> {
     let generation = 0
     openConsumerSessionMock.mockImplementation(async (_mux, options) => ({
-      mode: 'negotiated',
-      clientInstanceId: options.clientInstanceId,
-      clientGeneration: ++generation,
-      ownerGeneration: generation,
-      ownerLease: `owner-lease-${generation}`,
-      outputFlowControl: { version: 1, windowSu: 256 * 1024 }
+      state: {
+        mode: 'negotiated',
+        clientInstanceId: options.clientInstanceId,
+        clientGeneration: ++generation,
+        ownerGeneration: generation,
+        ownerLease: `owner-lease-${generation}`,
+        outputFlowControl: { version: 1, windowSu: 256 * 1024 }
+      },
+      resumed: options.resume !== undefined
     }))
     vi.mocked(getSshPtyAcceptedSourceCheckpoints).mockReturnValue([
       {

src/main/ssh/ssh-relay-session-test-fixtures.ts

@@ -1,6 +1,5 @@
 import { vi, type Mock } from 'vitest'
 import type { BrowserWindow } from 'electron'
-import { PtyConsumerSession } from '../../shared/pty-consumer-session'
 import type { SshConnection } from './ssh-connection'
 import type { Store } from '../persistence'
 import type { SshPortForwardManager } from './ssh-port-forward'
@@ -53,39 +52,3 @@ export function mockDeploySuccess(): void {
     platform: 'linux-x64'
   })
 }
-
-export function createMismatchedOwnerRecoveryError(): unknown {
-  const stateMachine = new PtyConsumerSession({
-    serverBuildId: 'test-relay-build',
-    createLease: () => 'retained-owner-lease'
-  })
-  const owner = stateMachine.admit(
-    { clientInstanceId: 'retained-client', requestedRole: 'session-owner' },
-    {
-      connectionId: 'retained-connection',
-      principal: 'retained-principal',
-      authenticated: true,
-      allowSessionOwner: true
-    }
-  )
-  owner.commitPublication()
-  stateMachine.close('retained-connection')
-  try {
-    stateMachine.admit(
-      {
-        clientInstanceId: 'retained-client',
-        requestedRole: 'session-owner',
-        resume: { ownerGeneration: 1, ownerLease: 'retained-owner-lease' }
-      },
-      {
-        connectionId: 'stale-connection',
-        principal: 'stale-principal',
-        authenticated: true,
-        allowSessionOwner: true
-      }
-    )
-  } catch (error) {
-    return error
-  }
-  throw new Error('Expected mismatched owner recovery to fail')
-}

src/main/ssh/ssh-relay-session.ts

@@ -6,7 +6,6 @@ import type { BrowserWindow } from 'electron'
 import { deployAndLaunchRelay } from './ssh-relay-deploy'
 import { execCommand } from './ssh-relay-deploy-helpers'
 import { isRelayVersionMismatchError } from './ssh-relay-version-mismatch-error'
-import type { RelayVersionMismatchError } from './ssh-relay-version-mismatch-error'
 import { SshChannelMultiplexer } from './ssh-channel-multiplexer'
 import { SshPtyProvider } from '../providers/ssh-pty-provider'
 import type { SshPtyAttachResult } from '../providers/ssh-pty-session-reattach'
@@ -80,8 +79,14 @@ import {
 import type { Store } from '../persistence'
 import type { OrcaRuntimeService } from '../runtime/orca-runtime'
 import { DEFAULT_PTY_SOURCE_WINDOW_SU } from '../../shared/pty-source-credit-contract'
-import { PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR } from '../../shared/pty-consumer-session'
-import { retrySshOwnerRecoveryWhileBlocked } from './ssh-owner-recovery-retry'
+import {
+  isSshOwnerAdmissionBlocked,
+  retrySshOwnerRecoveryWhileBlocked
+} from './ssh-owner-recovery-retry'
+import {
+  isSshOwnerAdmissionBlockedError,
+  SshOwnerAdmissionBlockedError
+} from './ssh-owner-admission-blocked-error'
 import { runRemoteOrcaCli } from './ssh-remote-orca-cli'
 import {
   acknowledgeRemoteOrcaCliPostOutput,
@@ -97,6 +102,8 @@ import { isTerminalLeafId, makePaneKey } from '../../shared/stable-pane-id'
 import { isValidTerminalTabId } from '../../shared/terminal-tab-id'
 import {
   openSshPtyConsumerSession,
+  type OpenSshPtyConsumerSessionOptions,
+  type SshPtyConsumerAdmission,
   type SshPtyConsumerOwnerState,
   type SshPtyConsumerSessionState
 } from './ssh-pty-consumer-session'
@@ -112,8 +119,7 @@ import {
   detachSshPtyConsumerRecovery,
   forgetSshPtyConsumerRecovery,
   getSshPtyConsumerRecovery,
-  rememberSshPtyConsumerRecovery,
-  removeSshPtyConsumerOwnerRecovery
+  rememberSshPtyConsumerRecovery
 } from './ssh-pty-consumer-recovery'
 
 export type RelaySessionState = 'idle' | 'deploying' | 'ready' | 'reconnecting' | 'disposed'
@@ -270,10 +276,9 @@ export class SshRelaySession {
   private muxNotificationCleanup: (() => void) | null = null
   // Why: onStateChange never fires when the relay channel closes but SSH stays up; this callback lets ssh.ts drive relay-level reconnect.
   private _onRelayLost: ((targetId: string) => void) | null = null
-  // Why: version mismatch is terminal, so it needs a separate callback from _onRelayLost (which expects a recoverable transport drop).
-  private _onTerminalRelayError:
-    | ((targetId: string, err: RelayVersionMismatchError) => void)
-    | null = null
+  // Why: a version mismatch or a blocked owner admission is terminal, so it needs a separate callback
+  // from _onRelayLost (which expects a recoverable transport drop).
+  private _onTerminalRelayError: ((targetId: string, err: Error) => void) | null = null
   private _onReady: ((targetId: string) => void) | null = null
   private portScanner: PortScanner | null = null
   private currentConnection: SshConnection | null = null
@@ -340,7 +345,7 @@ export class SshRelaySession {
     this._onRelayLost = cb
   }
 
-  setOnTerminalRelayError(cb: (targetId: string, err: RelayVersionMismatchError) => void): void {
+  setOnTerminalRelayError(cb: (targetId: string, err: Error) => void): void {
     this._onTerminalRelayError = cb
   }
 
@@ -540,10 +545,11 @@ export class SshRelaySession {
         )
         this._state = 'idle'
       }
-      // Why: a version mismatch on first connect is terminal (deployed binary vs. a still-running legacy daemon); notify the callback but still rethrow.
-      if (isRelayVersionMismatchError(err)) {
+      // Why: terminal on first connect — a deployed binary against a still-running legacy daemon, or a
+      // claim another connection holds. Notify the callback but still rethrow.
+      if (isRelayVersionMismatchError(err) || isSshOwnerAdmissionBlockedError(err)) {
         console.warn(
-          `[ssh-relay-session] Terminal relay version mismatch on initial connect for ${this.targetId}: ${err.message}`
+          `[ssh-relay-session] Terminal relay error on initial connect for ${this.targetId}: ${err.message}`
         )
         this._onTerminalRelayError?.(this.targetId, err)
       }
@@ -688,10 +694,11 @@ export class SshRelaySession {
             : 'connection_lost'
         )
       }
-      // Why: version-mismatch is terminal — fire the typed callback and drop out of 'reconnecting' since backoff retry can't reconcile it.
-      if (isRelayVersionMismatchError(err)) {
+      // Why terminal: neither a version mismatch nor a blocked owner claim is reconcilable by backoff
+      // retry, so fire the typed callback and drop out of 'reconnecting'.
+      if (isRelayVersionMismatchError(err) || isSshOwnerAdmissionBlockedError(err)) {
         console.warn(
-          `[ssh-relay-session] Terminal relay version mismatch for ${this.targetId}: ${err.message}`
+          `[ssh-relay-session] Terminal relay error for ${this.targetId}: ${err.message}`
         )
         if (this.abortController === abortController && !this.isDisposed()) {
           this._state = 'idle'
@@ -1027,6 +1034,19 @@ export class SshRelaySession {
       allowSameBuildLegacyFallback: true,
       outputFlowControl: { requestedWindowSu: DEFAULT_PTY_SOURCE_WINDOW_SU }
     }
+    const admission = await this.admitPtyConsumerOwner(mux, previousOwner, options, ownsAttempt)
+    if (previousOwner && !admission.resumed) {
+      this.voidPtyConsumerCheckpoints(previousOwner, ownsAttempt)
+    }
+    return admission.state
+  }
+
+  private async admitPtyConsumerOwner(
+    mux: SshChannelMultiplexer,
+    previousOwner: SshPtyConsumerOwnerState | null,
+    options: OpenSshPtyConsumerSessionOptions,
+    ownsAttempt: () => boolean
+  ): Promise<SshPtyConsumerAdmission> {
     try {
       return await retrySshOwnerRecoveryWhileBlocked(
         () =>
@@ -1047,48 +1067,47 @@ export class SshRelaySession {
         }
       )
     } catch (error) {
-      if (
-        !previousOwner ||
-        (error as { code?: unknown }).code !== PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR
-      ) {
-        throw error
+      // Why converted here: past this point the failure travels the same path as a dropped transport,
+      // where backoff would keep redeploying a relay that is working fine and refusing on purpose.
+      if (isSshOwnerAdmissionBlocked(error)) {
+        throw new SshOwnerAdmissionBlockedError(this.targetId, { cause: error })
       }
-      const recovery = getSshPtyConsumerRecovery(this.targetId)
-      // Why identity-guarded: the record is target-scoped and its clientInstanceId is shared by every
-      // session for that target, so only a record still describing the owner this attempt tried to
-      // resume is ours to drop — otherwise a loser wipes the winner's checkpoints.
-      const ownsRecoveryRecord =
-        ownsAttempt() &&
-        (!recovery?.owner ||
-          (recovery.owner.ownerGeneration === previousOwner.ownerGeneration &&
-            recovery.owner.ownerLease === previousOwner.ownerLease))
-      if (ownsRecoveryRecord) {
-        if (recovery) {
-          delete recovery.owner
-          recovery.checkpointsByAppPtyId.clear()
-          for (const [ptyId, migration] of recovery.modelMigrationsByAppPtyId) {
-            recovery.modelMigrationsByAppPtyId.set(
-              ptyId,
-              migration.then(() =>
-                Object.freeze({
-                  status: 'checkpoint-unavailable' as const,
-                  reason: 'completion-failed' as const
-                })
-              )
-            )
-          }
-        }
-        await removeSshPtyConsumerOwnerRecovery(
-          this.targetId,
-          this.ptyConsumerClientInstanceId,
-          this.store
+      throw error
+    }
+  }
+
+  // Why the recovery row and clientInstanceId survive: the relay minted a fresh claim, which voids the
+  // checkpoints taken under the old one but says nothing about our identity for this target. The caller
+  // durably records the new lease before ready, so removal here would only lose the identity.
+  private voidPtyConsumerCheckpoints(
+    previousOwner: SshPtyConsumerOwnerState,
+    ownsAttempt: () => boolean
+  ): void {
+    const recovery = getSshPtyConsumerRecovery(this.targetId)
+    // Why identity-guarded: the record is target-scoped and its clientInstanceId is shared by every
+    // session for that target, so only a record still describing the owner this attempt tried to
+    // resume is ours to void — otherwise a loser wipes the winner's checkpoints.
+    if (
+      !recovery ||
+      !ownsAttempt() ||
+      (recovery.owner &&
+        (recovery.owner.ownerGeneration !== previousOwner.ownerGeneration ||
+          recovery.owner.ownerLease !== previousOwner.ownerLease))
+    ) {
+      return
+    }
+    delete recovery.owner
+    recovery.checkpointsByAppPtyId.clear()
+    for (const [ptyId, migration] of recovery.modelMigrationsByAppPtyId) {
+      recovery.modelMigrationsByAppPtyId.set(
+        ptyId,
+        migration.then(() =>
+          Object.freeze({
+            status: 'checkpoint-unavailable' as const,
+            reason: 'completion-failed' as const
+          })
         )
-      }
-      if (!ownsAttempt()) {
-        throw new Error('Session disposed during establish')
-      }
-      this.ptyConsumerSessionState = null
-      return openSshPtyConsumerSession(mux, options)
+      )
     }
   }
 

src/relay/ssh-pty-consumer-session-adapter.test.ts

@@ -1,5 +1,8 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { PTY_CONSUMER_OWNER_GRACE_MS } from '../shared/pty-consumer-session'
+import {
+  PTY_CONSUMER_OWNER_GRACE_MS,
+  PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR
+} from '../shared/pty-consumer-session'
 import { RelayDispatcher, type RelayClientSessionIdentity } from './dispatcher'
 import { encodeJsonRpcFrame, MessageType } from './protocol'
 import { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter'
@@ -29,10 +32,18 @@ function openFrame(id: number, overrides: Record<string, unknown> = {}): Buffer
   )
 }
 
-function responseResult(buffer: Buffer): Record<string, unknown> {
+function responsePayload(buffer: Buffer): Record<string, unknown> {
   expect(buffer[0]).toBe(MessageType.Regular)
   const length = buffer.readUInt32BE(9)
-  return JSON.parse(buffer.subarray(13, 13 + length).toString('utf8')).result
+  return JSON.parse(buffer.subarray(13, 13 + length).toString('utf8'))
+}
+
+function responseResult(buffer: Buffer): Record<string, unknown> {
+  return responsePayload(buffer).result as Record<string, unknown>
+}
+
+function responseError(buffer: Buffer): Record<string, unknown> {
+  return responsePayload(buffer).error as Record<string, unknown>
 }
 
 async function flushRequests(): Promise<void> {
@@ -80,9 +91,14 @@ describe('SshPtyConsumerSessionAdapter', () => {
 
     expect(responseResult(firstWrites[0])).toMatchObject({
       role: 'session-owner',
-      ownerGeneration: 1
+      ownerGeneration: 1,
+      resumed: false
+    })
+    // Why a coded refusal reaches the wire: the dispatcher transports code and message only, so the
+    // competitor has to be able to tell "still settling" from "held" without parsing prose.
+    expect(responseError(secondWrites[0])).toMatchObject({
+      code: PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR
     })
-    expect(responseResult(secondWrites[0])).toMatchObject({ role: 'subscriber' })
     firstSettlements[0]({ ok: true })
   })
 

src/shared/pty-consumer-owner-recovery.ts

@@ -19,6 +19,23 @@ function throwRecoveryError(message: string, code: number): never {
   throw Object.assign(new Error(message), { code })
 }
 
+// Why identity excludes ownerGeneration: the generation fences the data path, but a reconnecting
+// owner legitimately arrives holding whatever generation it last persisted. Matching on the logical
+// triple is what lets one claim survive reconnects.
+export function matchesPtyConsumerOwnerClaim(
+  hello: PtyConsumerSessionHello,
+  authentication: PtyConsumerAuthentication,
+  current: IncumbentOwner
+): boolean {
+  const resume = hello.resume
+  return (
+    resume !== undefined &&
+    resume.ownerLease === current.lease &&
+    hello.clientInstanceId === current.clientInstanceId &&
+    authentication.principal === current.principal
+  )
+}
+
 export function assertPtyConsumerOwnerRecovery(
   hello: PtyConsumerSessionHello,
   authentication: PtyConsumerAuthentication,
@@ -28,11 +45,7 @@ export function assertPtyConsumerOwnerRecovery(
   if (!resume) {
     throw new Error('Owner recovery proof is required')
   }
-  if (
-    resume.ownerLease !== current.lease ||
-    hello.clientInstanceId !== current.clientInstanceId ||
-    authentication.principal !== current.principal
-  ) {
+  if (!matchesPtyConsumerOwnerClaim(hello, authentication, current)) {
     throwRecoveryError(
       'Owner recovery lease is stale or belongs to another principal',
       PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR

src/shared/pty-consumer-session-capabilities.ts

@@ -0,0 +1,45 @@
+import type {
+  PtyConsumerSessionGrant,
+  PtyConsumerSessionHello,
+  PtyConsumerSessionOptions
+} from './pty-consumer-session-contract'
+import { assertNonEmptyString, MAX_CAPABILITY_VERSIONS } from './pty-consumer-session-hello'
+
+export function assertPtyConsumerSessionOptions(options: PtyConsumerSessionOptions): void {
+  assertNonEmptyString(options.serverBuildId, 'serverBuildId')
+  if (
+    options.outputFlowControl &&
+    (!Number.isSafeInteger(options.outputFlowControl.maxWindowSu) ||
+      options.outputFlowControl.maxWindowSu <= 0 ||
+      options.outputFlowControl.versions.length > MAX_CAPABILITY_VERSIONS ||
+      options.outputFlowControl.versions.some(
+        (version) => !Number.isSafeInteger(version) || version <= 0
+      ))
+  ) {
+    throw new Error('outputFlowControl support is invalid')
+  }
+  if (
+    options.ownerGraceMs !== undefined &&
+    (!Number.isSafeInteger(options.ownerGraceMs) || options.ownerGraceMs < 0)
+  ) {
+    throw new Error('ownerGraceMs must be a non-negative safe integer')
+  }
+}
+
+export function intersectPtyConsumerCapabilities(
+  hello: PtyConsumerSessionHello,
+  support: PtyConsumerSessionOptions['outputFlowControl']
+): Pick<PtyConsumerSessionGrant, 'capabilities'> {
+  const offer = hello.capabilities?.outputFlowControl
+  if (!offer || !support || !offer.versions.includes(1) || !support.versions.includes(1)) {
+    return {}
+  }
+  return {
+    capabilities: {
+      outputFlowControl: {
+        version: 1,
+        windowSu: Math.min(offer.requestedWindowSu, support.maxWindowSu)
+      }
+    }
+  }
+}

src/shared/pty-consumer-session-contract.ts

@@ -5,6 +5,13 @@ export const PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR = -32041
 // window bounded by one response write, so the client may retry within a short budget.
 export const PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR = -32042
 export const PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR = -32043
+// Why two codes, not one message: the dispatcher transports only code and message, and the two
+// holders need opposite client behavior — an attached incumbent blocks, a disconnected one is transient.
+export const PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR = -32044
+export const PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR = -32045
+// Why: a disconnected incumbent keeps at most this much of its remaining grace once a different
+// owner-capable client asks, so admission converges inside one bounded retry instead of the full grace.
+export const PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS = 250
 
 export type PtyConsumerRole = 'session-owner' | 'subscriber'
 
@@ -30,6 +37,9 @@ export type PtyConsumerSessionGrant = {
   role: PtyConsumerRole
   ownerGeneration?: number
   ownerLease?: string
+  // Why: always present on a 'session-owner' grant, absent on a subscriber grant. `false` means the
+  // relay minted a fresh claim, so the client's checkpoints for the previous claim no longer apply.
+  resumed?: boolean
   capabilities?: {
     outputFlowControl?: {
       version: 1

src/shared/pty-consumer-session-hello.ts

@@ -34,19 +34,3 @@ export function validateHello(hello: PtyConsumerSessionHello): void {
     throw new Error('outputFlowControl.requestedWindowSu must be a positive safe integer')
   }
 }
-
-// Why: a duplicate open on one connection is only idempotent if it asks for exactly the same thing.
-export function helloFingerprint(hello: PtyConsumerSessionHello): string {
-  const flow = hello.capabilities?.outputFlowControl
-  return JSON.stringify({
-    clientInstanceId: hello.clientInstanceId,
-    requestedRole: hello.requestedRole,
-    resume: hello.resume,
-    outputFlowControl: flow
-      ? {
-          versions: [...flow.versions].sort((a, b) => a - b),
-          requestedWindowSu: flow.requestedWindowSu
-        }
-      : undefined
-  })
-}

src/shared/pty-consumer-session.test.ts

@@ -1,8 +1,10 @@
 import { describe, expect, it } from 'vitest'
 import {
+  PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS,
   PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR,
   PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR,
-  PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR,
   PtyConsumerSession,
   type PtyConsumerAuthentication,
   type PtyConsumerSessionHello
@@ -40,38 +42,51 @@ function createSession(options: { now?: () => number } = {}): PtyConsumerSession
 }
 
 describe('PtyConsumerSession', () => {
-  it('types a resume against a fresh relay without weakening other owner refusals', () => {
+  it('grants a fresh claim when the relay no longer holds the resumed record', () => {
     const session = createSession()
 
-    expect(() =>
-      session.admit(
-        ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'stale' } }),
-        auth('connection-1')
-      )
-    ).toThrow(
-      expect.objectContaining({
-        code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR,
-        message: expect.stringContaining('stale')
-      })
+    const admission = session.admit(
+      ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'forgotten' } }),
+      auth('connection-1')
     )
+
+    // Why one round trip: the client named a record this relay does not have, which is a fresh claim,
+    // not a refusal — `resumed: false` is what tells it the old checkpoints no longer apply.
+    expect(admission.grant).toMatchObject({
+      role: 'session-owner',
+      ownerGeneration: 1,
+      ownerLease: 'lease-1',
+      resumed: false
+    })
   })
 
   it('activates an authenticated owner only after its publication fence', () => {
     const session = createSession()
     const first = session.admit(ownerHello(), auth('connection-1'))
-    const competitor = session.admit(
-      ownerHello({ clientInstanceId: 'client-b' }),
-      auth('connection-2', { principal: 'other' })
-    )
 
     expect(first.grant).toMatchObject({
       clientGeneration: 1,
       role: 'session-owner',
       ownerGeneration: 1,
-      ownerLease: 'lease-1'
+      ownerLease: 'lease-1',
+      resumed: false
     })
-    expect(competitor.grant).toMatchObject({ clientGeneration: 2, role: 'subscriber' })
+    // Why a coded refusal and not a subscriber grant: a subscriber grant is unusable to a client that
+    // asked to own the PTY, and it arrives shaped like success.
+    expect(() =>
+      session.admit(
+        ownerHello({ clientInstanceId: 'client-b' }),
+        auth('connection-2', { principal: 'other' })
+      )
+    ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR }))
     first.commitPublication()
+
+    expect(() =>
+      session.admit(
+        ownerHello({ clientInstanceId: 'client-b' }),
+        auth('connection-3', { principal: 'other' })
+      )
+    ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR }))
   })
 
   it('rolls back an unpublished owner without consuming authority', () => {
@@ -86,14 +101,15 @@ describe('PtyConsumerSession', () => {
     })
   })
 
-  it('returns the same generation and lease for duplicate opens on one connection', () => {
+  it('rejects an identical duplicate open before it registers a second publication', () => {
     const session = createSession()
     const first = session.admit(ownerHello(), auth('connection-1'))
-    const duplicate = session.admit(ownerHello(), auth('connection-1'))
 
-    expect(duplicate.grant).toBe(first.grant)
+    // Why even an identical repeat: two responses settle independently, so one admission cannot make
+    // one response's rollback and the other's commit atomic.
+    expect(() => session.admit(ownerHello(), auth('connection-1'))).toThrow('only once')
     first.commitPublication()
-    duplicate.commitPublication()
+    expect(session.activeGrant('connection-1')).toBe(first.grant)
   })
 
   it('rejects a second, different open on one connection', () => {
@@ -339,48 +355,83 @@ describe('PtyConsumerSession', () => {
     expect(session.activeGrant('connection-2')).toBeNull()
   })
 
-  it('types mismatched recovery without disturbing principal or lease ownership', () => {
-    const session = createSession()
+  it('separates a disconnected holder from the owner it belongs to', () => {
+    let now = 10
+    const session = createSession({ now: () => now })
     const first = session.admit(ownerHello(), auth('connection-1'))
     first.commitPublication()
     session.close('connection-1')
 
-    expect(() =>
-      session.admit(
-        ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } }),
-        auth('connection-2', { principal: 'stale-desktop' })
-      )
-    ).toThrow(
-      expect.objectContaining({
-        code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR,
-        message: expect.stringContaining('principal')
-      })
-    )
-    expect(() =>
-      session.admit(
-        ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'wrong' } }),
-        auth('connection-3')
-      )
-    ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR }))
-
-    const staleFresh = session.admit(
-      ownerHello(),
-      auth('connection-2', { principal: 'stale-desktop' })
-    )
-    expect(staleFresh.grant).toMatchObject({ role: 'subscriber' })
-    expect(staleFresh.grant.ownerLease).toBeUndefined()
+    for (const [connectionId, hello] of [
+      ['connection-2', ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } })],
+      ['connection-3', ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'wrong' } })],
+      ['connection-4', ownerHello()]
+    ] as const) {
+      expect(() =>
+        session.admit(hello, auth(connectionId, { principal: 'other-desktop' }))
+      ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR }))
+    }
 
+    // Why the incumbent still wins: a matching proof is routed as a replacement and never reaches the
+    // held-owner branch, so shortening the grace cannot cost the real owner its lease.
     const recovered = session.admit(
       ownerHello({ resume: { ownerGeneration: 1, ownerLease: 'lease-1' } }),
-      auth('connection-4')
+      auth('connection-5')
     )
     expect(recovered.grant).toMatchObject({
       role: 'session-owner',
       ownerGeneration: 2,
-      ownerLease: 'lease-1'
+      ownerLease: 'lease-1',
+      resumed: true
     })
   })
 
+  it('clamps a refused disconnected holder to the shared grace floor', () => {
+    let now = 10
+    const session = createSession({ now: () => now })
+    const first = session.admit(ownerHello(), auth('connection-1'))
+    first.commitPublication()
+    session.close('connection-1')
+
+    const rival = ownerHello({ clientInstanceId: 'client-b' })
+    expect(() => session.admit(rival, auth('connection-2', { principal: 'other' }))).toThrow(
+      expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR })
+    )
+    now += PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS - 1
+    expect(() => session.admit(rival, auth('connection-3', { principal: 'other' }))).toThrow(
+      expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR })
+    )
+
+    now += 1
+    const promoted = session.admit(rival, auth('connection-4', { principal: 'other' }))
+
+    expect(promoted.grant).toMatchObject({
+      role: 'session-owner',
+      ownerGeneration: 2,
+      ownerLease: 'lease-2',
+      resumed: false
+    })
+  })
+
+  it('never converts an owner-capable request into a subscriber grant', () => {
+    const session = createSession()
+    const first = session.admit(ownerHello(), auth('connection-1'))
+    first.commitPublication()
+
+    const ineligible = session.admit(
+      ownerHello(),
+      auth('connection-2', { allowSessionOwner: false })
+    )
+
+    // Why this is the only subscriber outcome left: the request was not owner-capable in the first
+    // place, so no refusal code applies and the grant carries no `resumed`.
+    expect(ineligible.grant.role).toBe('subscriber')
+    expect(ineligible.grant).not.toHaveProperty('resumed')
+    expect(() =>
+      session.admit(ownerHello({ clientInstanceId: 'client-b' }), auth('connection-3'))
+    ).toThrow(expect.objectContaining({ code: PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR }))
+  })
+
   it('elects a new owner after disconnected-owner grace expires', () => {
     let now = 10
     const session = createSession({ now: () => now })

src/shared/pty-consumer-session.ts

@@ -1,8 +1,11 @@
 import { randomUUID } from 'node:crypto'
 import {
   PTY_CONSUMER_OWNER_GRACE_MS,
+  PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS,
+  PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR,
   PTY_CONSUMER_SESSION_PROTOCOL_VERSION,
-  PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR,
   type PtyConsumerAuthentication,
   type PtyConsumerDisplacedOwner,
   type PtyConsumerSessionAdmission,
@@ -10,18 +13,19 @@ import {
   type PtyConsumerSessionHello,
   type PtyConsumerSessionOptions
 } from './pty-consumer-session-contract'
+import { assertNonEmptyString, validateHello } from './pty-consumer-session-hello'
 import {
-  assertNonEmptyString,
-  helloFingerprint,
-  MAX_CAPABILITY_VERSIONS,
-  validateHello
-} from './pty-consumer-session-hello'
-import { assertPtyConsumerOwnerRecovery } from './pty-consumer-owner-recovery'
+  assertPtyConsumerSessionOptions,
+  intersectPtyConsumerCapabilities
+} from './pty-consumer-session-capabilities'
+import {
+  assertPtyConsumerOwnerRecovery,
+  matchesPtyConsumerOwnerClaim
+} from './pty-consumer-owner-recovery'
 
 export * from './pty-consumer-session-contract'
 
 type ClientRecord = {
-  fingerprint: string
   principal: string
   clientInstanceId: string
   grant: Readonly<PtyConsumerSessionGrant>
@@ -35,11 +39,16 @@ type OwnerRecord = {
   clientInstanceId: string
   generation: number
   lease: string
+  resumed: boolean
   state: 'pending' | 'active' | 'disconnected'
   disconnectedAt?: number
   replaces?: OwnerRecord
 }
 
+function throwOwnerError(message: string, code: number): never {
+  throw Object.assign(new Error(message), { code })
+}
+
 export class PtyConsumerSession {
   private readonly clients = new Map<string, ClientRecord>()
   private readonly now: () => number
@@ -50,24 +59,7 @@ export class PtyConsumerSession {
   private owner: OwnerRecord | null = null
 
   constructor(private readonly options: PtyConsumerSessionOptions) {
-    assertNonEmptyString(options.serverBuildId, 'serverBuildId')
-    if (
-      options.outputFlowControl &&
-      (!Number.isSafeInteger(options.outputFlowControl.maxWindowSu) ||
-        options.outputFlowControl.maxWindowSu <= 0 ||
-        options.outputFlowControl.versions.length > MAX_CAPABILITY_VERSIONS ||
-        options.outputFlowControl.versions.some(
-          (version) => !Number.isSafeInteger(version) || version <= 0
-        ))
-    ) {
-      throw new Error('outputFlowControl support is invalid')
-    }
-    if (
-      options.ownerGraceMs !== undefined &&
-      (!Number.isSafeInteger(options.ownerGraceMs) || options.ownerGraceMs < 0)
-    ) {
-      throw new Error('ownerGraceMs must be a non-negative safe integer')
-    }
+    assertPtyConsumerSessionOptions(options)
     this.now = options.now ?? Date.now
     this.createLease = options.createLease ?? randomUUID
     this.ownerGraceMs = options.ownerGraceMs ?? PTY_CONSUMER_OWNER_GRACE_MS
@@ -85,16 +77,11 @@ export class PtyConsumerSession {
     }
     this.expireOwner()
 
-    const fingerprint = helloFingerprint(hello)
-    const duplicate = this.clients.get(authentication.connectionId)
-    if (duplicate) {
-      if (
-        duplicate.fingerprint !== fingerprint ||
-        duplicate.principal !== authentication.principal
-      ) {
-        throw new Error('pty.openClient may be used only once per transport connection')
-      }
-      return this.admissionFor(duplicate)
+    // Why even an identical repeat is rejected: the two responses settle their publications
+    // independently, so one shared admission cannot make one response's rollback and the other's
+    // commit atomic. A client recovering from an RPC timeout opens a new connection instead.
+    if (this.clients.has(authentication.connectionId)) {
+      throw new Error('pty.openClient may be used only once per transport connection')
     }
 
     const owner = this.selectOwner(hello, authentication)
@@ -103,11 +90,12 @@ export class PtyConsumerSession {
       serverBuildId: this.options.serverBuildId,
       clientGeneration: this.nextClientGeneration++,
       role: owner ? ('session-owner' as const) : ('subscriber' as const),
-      ...(owner ? { ownerGeneration: owner.generation, ownerLease: owner.lease } : {}),
-      ...this.intersectCapabilities(hello)
+      ...(owner
+        ? { ownerGeneration: owner.generation, ownerLease: owner.lease, resumed: owner.resumed }
+        : {}),
+      ...intersectPtyConsumerCapabilities(hello, this.options.outputFlowControl)
     })
     const client: ClientRecord = {
-      fingerprint,
       principal: authentication.principal,
       clientInstanceId: hello.clientInstanceId,
       grant,
@@ -217,16 +205,14 @@ export class PtyConsumerSession {
       return null
     }
     const current = this.owner
+    // Why resume proof for a vacant record is not an error: the relay simply no longer has the record
+    // the client is naming. Minting a fresh claim here resolves it in one round trip, and `resumed:
+    // false` tells the client its checkpoints are void without making it delete its identity first.
     if (!current) {
-      if (hello.resume) {
-        throw Object.assign(new Error('Owner recovery lease is stale'), {
-          code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR
-        })
-      }
       return this.newOwner(hello, authentication, null)
     }
-    if (!hello.resume) {
-      return null
+    if (!matchesPtyConsumerOwnerClaim(hello, authentication, current)) {
+      this.refuseHeldOwner(current)
     }
     assertPtyConsumerOwnerRecovery(hello, authentication, current)
     // Why an active owner is displaced rather than refused: the resume proof matched this owner's
@@ -236,6 +222,41 @@ export class PtyConsumerSession {
     return this.newOwner(hello, authentication, current)
   }
 
+  // Why an owner-capable request is refused rather than demoted: a subscriber grant is unusable to a
+  // client that needs to drive the PTY, and it arrives shaped like success. A coded refusal lets the
+  // caller retry the transient case and stop on the blocked one.
+  private refuseHeldOwner(current: OwnerRecord): never {
+    if (current.state === 'pending') {
+      throwOwnerError(
+        'Owner grant publication is still pending',
+        PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR
+      )
+    }
+    if (current.state === 'active') {
+      throwOwnerError(
+        'PTY session owner is held by an attached connection',
+        PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR
+      )
+    }
+    this.clampDisconnectedOwnerGrace(current)
+    throwOwnerError(
+      'PTY session owner is held by a disconnected connection within its grace period',
+      PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR
+    )
+  }
+
+  // Why this is safe to shorten: the incumbent reached 'disconnected' only by an observed transport
+  // close, and expiring an owner record hands over admission — it never stops the remote PTY. The
+  // incumbent's own reconnect still wins outright, because a matching proof never reaches this branch.
+  private clampDisconnectedOwnerGrace(current: OwnerRecord): void {
+    const floorStart =
+      this.now() - Math.max(this.ownerGraceMs - PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS, 0)
+    if ((current.disconnectedAt ?? 0) <= floorStart) {
+      return
+    }
+    this.owner = { ...current, disconnectedAt: floorStart }
+  }
+
   private displacedOwnerFor(
     owner: OwnerRecord | null
   ): Readonly<PtyConsumerDisplacedOwner> | undefined {
@@ -275,29 +296,12 @@ export class PtyConsumerSession {
       clientInstanceId: hello.clientInstanceId,
       generation: this.nextOwnerGeneration++,
       lease,
+      resumed: replaces !== null,
       state: 'pending',
       ...(replaces ? { replaces } : {})
     }
   }
 
-  private intersectCapabilities(
-    hello: PtyConsumerSessionHello
-  ): Pick<PtyConsumerSessionGrant, 'capabilities'> {
-    const offer = hello.capabilities?.outputFlowControl
-    const support = this.options.outputFlowControl
-    if (!offer || !support || !offer.versions.includes(1) || !support.versions.includes(1)) {
-      return {}
-    }
-    return {
-      capabilities: {
-        outputFlowControl: {
-          version: 1,
-          windowSu: Math.min(offer.requestedWindowSu, support.maxWindowSu)
-        }
-      }
-    }
-  }
-
   private expireOwner(): void {
     if (
       this.owner?.state === 'disconnected' &&