diff --git a/src/main/ipc/ssh.test.ts b/src/main/ipc/ssh.test.ts
index 2a95134e712..f967fceb629 100644
--- a/src/main/ipc/ssh.test.ts
+++ b/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 }
               }
diff --git a/src/main/ssh/ssh-owner-admission-blocked-error.ts b/src/main/ssh/ssh-owner-admission-blocked-error.ts
new file mode 100644
index 00000000000..35055cb4880
--- /dev/null
+++ b/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
+}
diff --git a/src/main/ssh/ssh-owner-recovery-retry.test.ts b/src/main/ssh/ssh-owner-recovery-retry.test.ts
index 9c8f4bd6a14..5a2ac3e92e9 100644
--- a/src/main/ssh/ssh-owner-recovery-retry.test.ts
+++ b/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
diff --git a/src/main/ssh/ssh-owner-recovery-retry.ts b/src/main/ssh/ssh-owner-recovery-retry.ts
index e5e78d9a4f2..9c54841c328 100644
--- a/src/main/ssh/ssh-owner-recovery-retry.ts
+++ b/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)
diff --git a/src/main/ssh/ssh-pty-consumer-session.test.ts b/src/main/ssh/ssh-pty-consumer-session.test.ts
index 76dc25ffacb..a1fc612f1dd 100644
--- a/src/main/ssh/ssh-pty-consumer-session.test.ts
+++ b/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
     })
   })
 
diff --git a/src/main/ssh/ssh-pty-consumer-session.ts b/src/main/ssh/ssh-pty-consumer-session.ts
index 8bb272d7273..aa58534909b 100644
--- a/src/main/ssh/ssh-pty-consumer-session.ts
+++ b/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!
   }
 }
diff --git a/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts b/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts
index 1a384a32f72..f54afef3f89 100644
--- a/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts
+++ b/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 }
     }
diff --git a/src/main/ssh/ssh-relay-session-data-delivery.test.ts b/src/main/ssh/ssh-relay-session-data-delivery.test.ts
index 56f6c6eeff2..0966b07fd2a 100644
--- a/src/main/ssh/ssh-relay-session-data-delivery.test.ts
+++ b/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([
diff --git a/src/main/ssh/ssh-relay-session-model-migration.test.ts b/src/main/ssh/ssh-relay-session-model-migration.test.ts
index 9bd2216357d..ae5db55e0cc 100644
--- a/src/main/ssh/ssh-relay-session-model-migration.test.ts
+++ b/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))
diff --git a/src/main/ssh/ssh-relay-session-recovery-durability.test.ts b/src/main/ssh/ssh-relay-session-recovery-durability.test.ts
index 6e1fe81c5a5..50e87d91e9c 100644
--- a/src/main/ssh/ssh-relay-session-recovery-durability.test.ts
+++ b/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
 
diff --git a/src/main/ssh/ssh-relay-session-recovery-races.test.ts b/src/main/ssh/ssh-relay-session-recovery-races.test.ts
index 81039d901ec..e44359ba5c2 100644
--- a/src/main/ssh/ssh-relay-session-recovery-races.test.ts
+++ b/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([
       {
diff --git a/src/main/ssh/ssh-relay-session-test-fixtures.ts b/src/main/ssh/ssh-relay-session-test-fixtures.ts
index b0ed98eaf2e..14bbb76a64a 100644
--- a/src/main/ssh/ssh-relay-session-test-fixtures.ts
+++ b/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')
-}
diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts
index 34bdf498843..9889b1484af 100644
--- a/src/main/ssh/ssh-relay-session.ts
+++ b/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)
+      )
     }
   }
 
diff --git a/src/relay/ssh-pty-consumer-session-adapter.test.ts b/src/relay/ssh-pty-consumer-session-adapter.test.ts
index a0f040814ed..496fab8ab76 100644
--- a/src/relay/ssh-pty-consumer-session-adapter.test.ts
+++ b/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 })
   })
 
diff --git a/src/shared/pty-consumer-owner-recovery.ts b/src/shared/pty-consumer-owner-recovery.ts
index dfac94d7e9c..f107d23444f 100644
--- a/src/shared/pty-consumer-owner-recovery.ts
+++ b/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
diff --git a/src/shared/pty-consumer-session-capabilities.ts b/src/shared/pty-consumer-session-capabilities.ts
new file mode 100644
index 00000000000..89962f6b3b3
--- /dev/null
+++ b/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)
+      }
+    }
+  }
+}
diff --git a/src/shared/pty-consumer-session-contract.ts b/src/shared/pty-consumer-session-contract.ts
index 1e359a11f53..4e4cbcfeab1 100644
--- a/src/shared/pty-consumer-session-contract.ts
+++ b/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
diff --git a/src/shared/pty-consumer-session-hello.ts b/src/shared/pty-consumer-session-hello.ts
index c8119da88ee..9dc54adf30c 100644
--- a/src/shared/pty-consumer-session-hello.ts
+++ b/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
-  })
-}
diff --git a/src/shared/pty-consumer-session.test.ts b/src/shared/pty-consumer-session.test.ts
index ce6282e8664..44945d3bfa8 100644
--- a/src/shared/pty-consumer-session.test.ts
+++ b/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 })
diff --git a/src/shared/pty-consumer-session.ts b/src/shared/pty-consumer-session.ts
index 3baf502acd8..215ec1181fb 100644
--- a/src/shared/pty-consumer-session.ts
+++ b/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' &&