diff --git a/src/main/providers/ssh-pty-notification-rejection-routing.test.ts b/src/main/providers/ssh-pty-notification-rejection-routing.test.ts
new file mode 100644
index 000000000..79314ae9e
--- /dev/null
+++ b/src/main/providers/ssh-pty-notification-rejection-routing.test.ts
@@ -0,0 +1,183 @@
+import { expect, it, vi } from 'vitest'
+import { subscribeSshPtyNotifications } from './ssh-pty-notification-routing'
+
+it('routes malformed and unadmitted source frames only to rejection diagnostics', async () => {
+  const mux = {
+    onNotification: vi.fn(),
+    request: vi.fn(async (_method: string, params: Record<string, unknown>) => {
+      if (params.deliveryToken !== 'token-1') {
+        throw new Error('Unknown or stale PTY source delivery cancellation')
+      }
+      return { canceled: true, sentEndSu: 0, creditedEndSu: 0 }
+    })
+  }
+  const dataListeners = new Set()
+  const rejectedDataListeners = new Set()
+  const subscription = subscribeSshPtyNotifications({
+    mux: mux as never,
+    toAppPtyId: (id) => `ssh:conn@@${id}`,
+    dataListeners: dataListeners as never,
+    rejectedDataListeners: rejectedDataListeners as never,
+    replayListeners: new Set() as never,
+    exitListeners: new Set() as never,
+    livePtyIds: new Set(),
+    recordExit: vi.fn(),
+    providerGeneration: 7,
+    resolvePtyIncarnation: (id) => `incarnation:${id}`,
+    peekPtyIncarnation: () => undefined
+  })
+  const handler = mux.onNotification.mock.calls[0]?.[0] as (
+    method: string,
+    params: Record<string, unknown>
+  ) => void
+  const onData = vi.fn()
+  const onRejectedData = vi.fn()
+  dataListeners.add(onData)
+  rejectedDataListeners.add(onRejectedData)
+  subscription
+    .installReceivingActivation('pty-1', {
+      status: 'pending',
+      clientGeneration: 2,
+      ownerGeneration: 3,
+      ptyIncarnation: 'incarnation-1',
+      deliveryToken: 'token-1',
+      checkpointSourceEndSu: 0,
+      recoveryEndSu: 4
+    })
+    .commit()
+
+  handler('pty.data', {
+    id: 'pty-1',
+    data: 'bad',
+    deliveryToken: 'token-1',
+    clientGeneration: 2,
+    ownerGeneration: 3,
+    ptyIncarnation: 'incarnation-1',
+    sourceEndSu: 4,
+    sourceLengthSu: 4
+  })
+  handler('pty.data', {
+    id: 'pty-2',
+    data: 'old',
+    deliveryToken: 'token-old',
+    clientGeneration: 1,
+    ownerGeneration: 1,
+    ptyIncarnation: 'incarnation-1',
+    sourceEndSu: 3,
+    sourceLengthSu: 3
+  })
+
+  await vi.waitFor(() => expect(onRejectedData).toHaveBeenCalledTimes(2))
+
+  expect(onData).not.toHaveBeenCalled()
+  expect(onRejectedData).toHaveBeenNthCalledWith(
+    1,
+    expect.objectContaining({ sourceMalformed: true })
+  )
+  expect(onRejectedData).toHaveBeenNthCalledWith(
+    2,
+    expect.objectContaining({
+      sourceRejected: true,
+      source: expect.objectContaining({
+        deliveryToken: 'token-old',
+        clientGeneration: 1,
+        ownerGeneration: 1
+      })
+    })
+  )
+})
+
+it('never resolves an incarnation for a rejected frame', async () => {
+  const mux = {
+    onNotification: vi.fn(),
+    request: vi.fn(async () => ({ canceled: true, sentEndSu: 0, creditedEndSu: 0 }))
+  }
+  const rejectedDataListeners = new Set()
+  const resolvePtyIncarnation = vi.fn((id: string) => `incarnation:${id}`)
+  subscribeSshPtyNotifications({
+    mux: mux as never,
+    toAppPtyId: (id) => `ssh:conn@@${id}`,
+    dataListeners: new Set() as never,
+    rejectedDataListeners: rejectedDataListeners as never,
+    replayListeners: new Set() as never,
+    exitListeners: new Set() as never,
+    livePtyIds: new Set(),
+    recordExit: vi.fn(),
+    providerGeneration: 7,
+    resolvePtyIncarnation,
+    peekPtyIncarnation: () => undefined
+  })
+  const handler = mux.onNotification.mock.calls[0]?.[0] as (
+    method: string,
+    params: Record<string, unknown>
+  ) => void
+  const onRejectedData = vi.fn()
+  rejectedDataListeners.add(onRejectedData)
+
+  // Why this matters: resolvePtyIncarnation caches what it mints and rememberPtyIncarnation is
+  // first-write-wins, so a synthetic id minted here would fence the PTY's real incarnation off for
+  // the rest of the generation.
+  handler('pty.data', { id: 'pty-1', data: 'bad', deliveryToken: 'token-1', sourceEndSu: 4 })
+
+  await vi.waitFor(() => expect(onRejectedData).toHaveBeenCalledOnce())
+
+  expect(onRejectedData).toHaveBeenCalledOnce()
+  expect(onRejectedData).toHaveBeenCalledWith(expect.objectContaining({ sourceMalformed: true }))
+  expect(resolvePtyIncarnation).not.toHaveBeenCalled()
+})
+
+it('coalesces repeated exact rejections into one fresh-activation recovery', async () => {
+  const mux = {
+    onNotification: vi.fn(),
+    request: vi.fn(async () => ({ canceled: true, sentEndSu: 3, creditedEndSu: 0 }))
+  }
+  const rejectedDataListeners = new Set()
+  const subscription = subscribeSshPtyNotifications({
+    mux: mux as never,
+    toAppPtyId: (id) => `ssh:conn@@${id}`,
+    dataListeners: new Set() as never,
+    rejectedDataListeners: rejectedDataListeners as never,
+    replayListeners: new Set() as never,
+    exitListeners: new Set() as never,
+    livePtyIds: new Set(),
+    recordExit: vi.fn(),
+    providerGeneration: 7,
+    resolvePtyIncarnation: (id) => `incarnation:${id}`,
+    peekPtyIncarnation: () => 'incarnation-1'
+  })
+  subscription
+    .installReceivingActivation('pty-1', {
+      status: 'pending',
+      clientGeneration: 2,
+      ownerGeneration: 3,
+      ptyIncarnation: 'incarnation-1',
+      deliveryToken: 'token-1',
+      checkpointSourceEndSu: 0,
+      recoveryEndSu: 0
+    })
+    .commit()
+  const handler = mux.onNotification.mock.calls[0]?.[0] as (
+    method: string,
+    params: Record<string, unknown>
+  ) => void
+  const onRejectedData = vi.fn()
+  rejectedDataListeners.add(onRejectedData)
+  const rejected = {
+    id: 'pty-1',
+    data: 'gap',
+    deliveryToken: 'token-1',
+    clientGeneration: 2,
+    ownerGeneration: 3,
+    ptyIncarnation: 'incarnation-1',
+    sourceEndSu: 6,
+    sourceLengthSu: 3
+  }
+
+  handler('pty.data', rejected)
+  handler('pty.data', rejected)
+
+  await vi.waitFor(() => expect(onRejectedData).toHaveBeenCalledOnce())
+  expect(onRejectedData).toHaveBeenCalledWith(
+    expect.objectContaining({ rejectedSourceRecovery: 'fresh-activation' })
+  )
+})
diff --git a/src/main/providers/ssh-pty-notification-routing.test.ts b/src/main/providers/ssh-pty-notification-routing.test.ts
index 5fb495536..fdb6776bf 100644
--- a/src/main/providers/ssh-pty-notification-routing.test.ts
+++ b/src/main/providers/ssh-pty-notification-routing.test.ts
@@ -29,7 +29,8 @@ function createSubscription() {
     livePtyIds,
     recordExit,
     providerGeneration: 7,
-    resolvePtyIncarnation
+    resolvePtyIncarnation,
+    peekPtyIncarnation: () => undefined
   })
 
   const handler = mux.onNotification.mock.calls[0]?.[0] as (
diff --git a/src/main/providers/ssh-pty-notification-routing.ts b/src/main/providers/ssh-pty-notification-routing.ts
index 8cd5c2cb8..b6063799f 100644
--- a/src/main/providers/ssh-pty-notification-routing.ts
+++ b/src/main/providers/ssh-pty-notification-routing.ts
@@ -7,10 +7,11 @@ import type {
   SshPtyReplayCallback
 } from './ssh-pty-provider-contract'
 import { parseSshPtySourceFrame } from './ssh-pty-source-frame'
-import {
-  SshPtySourceDeliveryLedger,
-  type PendingSshPtySourceData
-} from './ssh-pty-source-delivery-ledger'
+import { SshPtySourceDeliveryLedger } from './ssh-pty-source-delivery-ledger'
+import type {
+  PendingSshPtySourceData,
+  SshPtyRejectedSourceRecovery
+} from './ssh-pty-source-delivery-state'
 
 export type { SshPtyDataCallback, SshPtyExitCallback, SshPtyReplayCallback }
 export type SshPtyRecoveryActivationLease = Readonly<{
@@ -35,18 +36,25 @@ export function subscribeSshPtyNotifications(args: {
   mux: SshChannelMultiplexer
   toAppPtyId: (id: string) => string
   dataListeners: Set<SshPtyDataCallback>
+  rejectedDataListeners?: Set<SshPtyDataCallback>
   replayListeners: Set<SshPtyReplayCallback>
   exitListeners: Set<SshPtyExitCallback>
   livePtyIds: Set<string>
   recordExit: (relayPtyId: string, incarnationId: unknown) => void
   providerGeneration: number
   resolvePtyIncarnation: (relayPtyId: string, incarnationId?: unknown) => string
+  peekPtyIncarnation: (relayPtyId: string) => string | undefined
 }): SshPtyNotificationSubscription {
-  const toDataPayload = (pending: PendingSshPtySourceData): Parameters<SshPtyDataCallback>[0] => {
+  const toDataPayload = (
+    pending: PendingSshPtySourceData,
+    incarnationOverride?: string
+  ): Parameters<SshPtyDataCallback>[0] => {
     const id = args.toAppPtyId(pending.relayPtyId)
-    const ptyIncarnation = pending.source
-      ? (pending.params.ptyIncarnation as string)
-      : args.resolvePtyIncarnation(pending.relayPtyId, pending.params.incarnationId)
+    const ptyIncarnation =
+      incarnationOverride ??
+      (pending.source
+        ? (pending.params.ptyIncarnation as string)
+        : args.resolvePtyIncarnation(pending.relayPtyId, pending.params.incarnationId))
     return {
       id,
       data: pending.data,
@@ -67,7 +75,80 @@ export function subscribeSshPtyNotifications(args: {
       listener(payload)
     }
   }
+  // Why: a rejected frame is diagnostic and must never mint an incarnation. resolvePtyIncarnation
+  // caches what it synthesizes and rememberPtyIncarnation is first-write-wins, so a malformed frame
+  // that lands before the PTY's first good one would pin a `legacy:` id the real attach can never
+  // displace — fencing every later frame of that generation off as a mismatch.
+  const rejectedPtyIncarnation = (pending: PendingSshPtySourceData): string => {
+    const offered = pending.params.ptyIncarnation
+    if (typeof offered === 'string' && offered.length > 0) {
+      return offered
+    }
+    return args.peekPtyIncarnation(pending.relayPtyId) ?? ''
+  }
+  const publishRejectedData = (
+    pending: PendingSshPtySourceData,
+    rejection: 'malformed' | 'unadmitted',
+    recovery: SshPtyRejectedSourceRecovery
+  ): void => {
+    const listeners = args.rejectedDataListeners
+    if (!listeners || listeners.size === 0) {
+      return
+    }
+    const payload = {
+      ...toDataPayload(pending, rejectedPtyIncarnation(pending)),
+      ...(rejection === 'malformed' ? { sourceMalformed: true } : {}),
+      ...(rejection === 'unadmitted' ? { sourceRejected: true } : {}),
+      rejectedSourceRecovery: recovery
+    }
+    for (const listener of listeners) {
+      listener(payload)
+    }
+  }
   const sourceDeliveries = new SshPtySourceDeliveryLedger(args.mux, publishData)
+  const rejectedPublications = new Map<
+    string,
+    {
+      pending: number
+      payload: PendingSshPtySourceData
+      rejection: 'malformed' | 'unadmitted'
+      recovery?: SshPtyRejectedSourceRecovery
+    }
+  >()
+  const rejectSourceData = (
+    pending: PendingSshPtySourceData,
+    rejection: 'malformed' | 'unadmitted'
+  ): void => {
+    const batch = rejectedPublications.get(pending.relayPtyId) ?? {
+      pending: 0,
+      payload: pending,
+      rejection
+    }
+    batch.pending++
+    rejectedPublications.set(pending.relayPtyId, batch)
+    void sourceDeliveries
+      .reject(pending.relayPtyId, rejectedSourceIdentity(pending.params))
+      .then((recovery) => {
+        if (
+          !batch.recovery ||
+          rejectedRecoveryPriority(recovery) > rejectedRecoveryPriority(batch.recovery)
+        ) {
+          batch.payload = pending
+          batch.rejection = rejection
+          batch.recovery = recovery
+        }
+      })
+      .finally(() => {
+        batch.pending--
+        queueMicrotask(() => {
+          if (batch.pending > 0 || rejectedPublications.get(pending.relayPtyId) !== batch) {
+            return
+          }
+          rejectedPublications.delete(pending.relayPtyId)
+          publishRejectedData(batch.payload, batch.rejection, batch.recovery ?? 'reconnect-channel')
+        })
+      })
+  }
   const dispose = args.mux.onNotification((method, params) => {
     // Why: mux delivers every method to generic handlers; non-PTY payloads
     // (workspace.changed, fs.changed, …) have no `id` and must not reach
@@ -82,6 +163,7 @@ export function subscribeSshPtyNotifications(args: {
     if (method === 'pty.exit') {
       const id = args.toAppPtyId(relayPtyId)
       const ptyIncarnation = args.resolvePtyIncarnation(relayPtyId, params.incarnationId)
+      rejectedPublications.delete(relayPtyId)
       args.recordExit(relayPtyId, params.incarnationId)
       args.livePtyIds.delete(id)
       sourceDeliveries.recordExit(relayPtyId)
@@ -109,7 +191,8 @@ export function subscribeSshPtyNotifications(args: {
     const data = typeof params.data === 'string' ? params.data : ''
     const sourceFrame = parseSshPtySourceFrame(params, data, relayPtyId)
     if (sourceFrame.malformed) {
-      cancelExactSourceDelivery(args.mux, relayPtyId, params)
+      const pending = Object.freeze({ relayPtyId, params, data })
+      rejectSourceData(pending, 'malformed')
       return
     }
     const pending = Object.freeze({
@@ -120,14 +203,17 @@ export function subscribeSshPtyNotifications(args: {
     })
     if (sourceFrame.source) {
       if (!sourceDeliveries.admit({ ...pending, source: sourceFrame.source })) {
-        cancelExactSourceDelivery(args.mux, relayPtyId, params)
+        rejectSourceData(pending, 'unadmitted')
       }
       return
     }
     publishData(pending)
   })
   return Object.freeze({
-    dispose,
+    dispose: () => {
+      rejectedPublications.clear()
+      dispose()
+    },
     installReceivingActivation: (relayPtyId, activation) => {
       const lease = sourceDeliveries.install(relayPtyId, activation)
       return Object.freeze({
@@ -140,33 +226,42 @@ export function subscribeSshPtyNotifications(args: {
   })
 }
 
-function cancelExactSourceDelivery(
-  mux: SshChannelMultiplexer,
-  relayPtyId: string,
-  params: {
-    deliveryToken?: unknown
-    clientGeneration?: unknown
-    ownerGeneration?: unknown
+function rejectedRecoveryPriority(recovery: SshPtyRejectedSourceRecovery): number {
+  if (recovery === 'reconnect-channel') {
+    return 3
   }
-): void {
+  return recovery === 'fresh-activation' ? 2 : 1
+}
+
+function rejectedSourceIdentity(params: {
+  deliveryToken?: unknown
+  clientGeneration?: unknown
+  ownerGeneration?: unknown
+  ptyIncarnation?: unknown
+}):
+  | Readonly<{
+      deliveryToken: string
+      clientGeneration: number
+      ownerGeneration: number
+      ptyIncarnation: string
+    }>
+  | undefined {
   if (
     typeof params.deliveryToken !== 'string' ||
     params.deliveryToken.length === 0 ||
     !positiveSafeInteger(params.clientGeneration) ||
-    !positiveSafeInteger(params.ownerGeneration)
+    !positiveSafeInteger(params.ownerGeneration) ||
+    typeof params.ptyIncarnation !== 'string' ||
+    params.ptyIncarnation.length === 0
   ) {
-    return
+    return undefined
   }
-  try {
-    void mux
-      .request('pty.cancelDelivery', {
-        id: relayPtyId,
-        clientGeneration: params.clientGeneration,
-        ownerGeneration: params.ownerGeneration,
-        deliveryToken: params.deliveryToken
-      })
-      .catch(() => {})
-  } catch {}
+  return Object.freeze({
+    deliveryToken: params.deliveryToken,
+    clientGeneration: params.clientGeneration,
+    ownerGeneration: params.ownerGeneration,
+    ptyIncarnation: params.ptyIncarnation
+  })
 }
 
 function positiveSafeInteger(value: unknown): value is number {
diff --git a/src/main/providers/ssh-pty-provider-contract.ts b/src/main/providers/ssh-pty-provider-contract.ts
index 18c21bdb7..4ac6f0fb4 100644
--- a/src/main/providers/ssh-pty-provider-contract.ts
+++ b/src/main/providers/ssh-pty-provider-contract.ts
@@ -27,6 +27,8 @@ export type SshPtyDataCallback = (payload: {
     sourceEndSu: number
   }>
   sourceMalformed?: boolean
+  sourceRejected?: boolean
+  rejectedSourceRecovery?: 'confirm-existing' | 'fresh-activation' | 'reconnect-channel'
 }) => void
 export type SshPtyReplayCallback = (payload: { id: string; data: string }) => void
 export type SshPtyExitCallback = (payload: {
diff --git a/src/main/providers/ssh-pty-provider-output-state.ts b/src/main/providers/ssh-pty-provider-output-state.ts
index fd9bde525..4d98097b9 100644
--- a/src/main/providers/ssh-pty-provider-output-state.ts
+++ b/src/main/providers/ssh-pty-provider-output-state.ts
@@ -14,6 +14,7 @@ import type { PtySourceReceivingActivation } from '../../shared/pty-source-recei
 
 export class SshPtyProviderOutputState {
   private readonly dataListeners = new Set<SshPtyDataCallback>()
+  private readonly rejectedDataListeners = new Set<SshPtyDataCallback>()
   private readonly replayListeners = new Set<SshPtyReplayCallback>()
   private readonly exitListeners = new Set<SshPtyExitCallback>()
   private readonly incarnationByRelayPtyId = new Map<string, string>()
@@ -34,11 +35,13 @@ export class SshPtyProviderOutputState {
     this.subscription = subscribeSshPtyNotifications({
       ...args,
       dataListeners: this.dataListeners,
+      rejectedDataListeners: this.rejectedDataListeners,
       replayListeners: this.replayListeners,
       exitListeners: this.exitListeners,
       providerGeneration,
       resolvePtyIncarnation: (relayPtyId, incarnationId) =>
         this.resolvePtyIncarnation(relayPtyId, incarnationId),
+      peekPtyIncarnation: (relayPtyId) => this.incarnationByRelayPtyId.get(relayPtyId),
       recordExit: (relayPtyId, incarnationId) => {
         args.recordExit(relayPtyId, incarnationId)
         this.incarnationByRelayPtyId.delete(relayPtyId)
@@ -52,6 +55,7 @@ export class SshPtyProviderOutputState {
     this.subscription?.dispose()
     this.subscription = null
     this.dataListeners.clear()
+    this.rejectedDataListeners.clear()
     this.replayListeners.clear()
     this.exitListeners.clear()
     this.incarnationByRelayPtyId.clear()
@@ -63,6 +67,11 @@ export class SshPtyProviderOutputState {
     return () => this.dataListeners.delete(callback)
   }
 
+  onRejectedData(callback: SshPtyDataCallback): () => void {
+    this.rejectedDataListeners.add(callback)
+    return () => this.rejectedDataListeners.delete(callback)
+  }
+
   onReplay(callback: SshPtyReplayCallback): () => void {
     this.replayListeners.add(callback)
     return () => this.replayListeners.delete(callback)
diff --git a/src/main/providers/ssh-pty-provider.ts b/src/main/providers/ssh-pty-provider.ts
index 5811a19d5..2bc52fe49 100644
--- a/src/main/providers/ssh-pty-provider.ts
+++ b/src/main/providers/ssh-pty-provider.ts
@@ -314,6 +314,8 @@ export class SshPtyProvider implements IPtyProvider {
   }
 
   onData = (callback: SshPtyDataCallback): (() => void) => this.outputState.onData(callback)
+  onRejectedData = (callback: SshPtyDataCallback): (() => void) =>
+    this.outputState.onRejectedData(callback)
   onReplay = (callback: SshPtyReplayCallback): (() => void) => this.outputState.onReplay(callback)
   onExit = (callback: SshPtyExitCallback): (() => void) => this.outputState.onExit(callback)
 
diff --git a/src/main/providers/ssh-pty-source-delivery-ledger.test.ts b/src/main/providers/ssh-pty-source-delivery-ledger.test.ts
index 4040b56a0..7b1f7aa2b 100644
--- a/src/main/providers/ssh-pty-source-delivery-ledger.test.ts
+++ b/src/main/providers/ssh-pty-source-delivery-ledger.test.ts
@@ -2,6 +2,17 @@ import { describe, expect, it, vi } from 'vitest'
 import { SshPtySourceDeliveryLedger } from './ssh-pty-source-delivery-ledger'
 
 describe('SshPtySourceDeliveryLedger', () => {
+  const activation = (deliveryToken: string) =>
+    Object.freeze({
+      status: 'pending' as const,
+      clientGeneration: 2,
+      ownerGeneration: 3,
+      ptyIncarnation: 'incarnation-1',
+      deliveryToken,
+      checkpointSourceEndSu: 0,
+      recoveryEndSu: 0
+    })
+
   it('retains cancellation ownership when recovery transfer is superseded', async () => {
     const request = vi.fn(async () => ({ canceled: true, sentEndSu: 0, creditedEndSu: 0 }))
     const ledger = new SshPtySourceDeliveryLedger({ request } as never, vi.fn())
@@ -41,4 +52,44 @@ describe('SshPtySourceDeliveryLedger', () => {
       deliveryToken: 'token-old'
     })
   })
+
+  it('retires an exact rejected activation only after cancellation is proven', async () => {
+    const request = vi.fn(async () => ({ canceled: true, sentEndSu: 4, creditedEndSu: 0 }))
+    const ledger = new SshPtySourceDeliveryLedger({ request } as never, vi.fn())
+    ledger.install('pty-1', activation('token-old')).commit()
+
+    await expect(ledger.reject('pty-1', activation('token-old'))).resolves.toBe('fresh-activation')
+
+    expect(() => ledger.install('pty-1', activation('token-new')).commit()).not.toThrow()
+  })
+
+  it('keeps the current activation when a stale rejected token cannot cancel it', async () => {
+    const publish = vi.fn()
+    const request = vi.fn(async () => {
+      throw new Error('Unknown or stale PTY source delivery cancellation')
+    })
+    const ledger = new SshPtySourceDeliveryLedger({ request } as never, publish)
+    ledger.install('pty-1', activation('token-current')).commit()
+
+    await expect(ledger.reject('pty-1', activation('token-stale'))).resolves.toBe(
+      'confirm-existing'
+    )
+    expect(
+      ledger.admit({
+        relayPtyId: 'pty-1',
+        params: { ptyIncarnation: 'incarnation-1' },
+        data: 'healthy',
+        source: {
+          relayPtyId: 'pty-1',
+          spanId: 'span-1',
+          clientGeneration: 2,
+          ownerGeneration: 3,
+          deliveryToken: 'token-current',
+          sourceStartSu: 0,
+          sourceEndSu: 7
+        }
+      })
+    ).toBe(true)
+    expect(publish).toHaveBeenCalledWith(expect.objectContaining({ data: 'healthy' }))
+  })
 })
diff --git a/src/main/providers/ssh-pty-source-delivery-ledger.ts b/src/main/providers/ssh-pty-source-delivery-ledger.ts
index 9e101345c..717e8554f 100644
--- a/src/main/providers/ssh-pty-source-delivery-ledger.ts
+++ b/src/main/providers/ssh-pty-source-delivery-ledger.ts
@@ -1,40 +1,21 @@
 import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
 import type { PtySourceReceivingActivation } from '../../shared/pty-source-receiving-activation'
 import type { SshPtySourceFrame } from './ssh-pty-source-frame'
-
-export type PendingSshPtySourceData = Readonly<{
-  relayPtyId: string
-  params: Record<string, unknown>
-  data: string
-  source?: SshPtySourceFrame
-}>
-
-type SourceDeliveryLeaseState = {
-  phase: 'provisional' | 'recovery' | 'committing' | 'committed' | 'retired'
-  pendingData: PendingSshPtySourceData[]
-  recoverySink?: (pending: PendingSshPtySourceData) => void
-  exited: boolean
-}
-
-type SourceDeliveryState = Readonly<{
-  activation: PtySourceReceivingActivation
-  sourceEndSu: number
-  lease: SourceDeliveryLeaseState
-  previous?: SourceDeliveryState
-}>
-
-export type SshPtyRecoveryActivationLease = Readonly<{
-  commit: () => void
-  retire: () => void
-}>
-
-export type SshPtySourceDeliveryLease = Readonly<{
-  commit: () => void
-  rollback: () => Promise<boolean>
-  transferToRecovery: (
-    sink: (pending: PendingSshPtySourceData) => void
-  ) => SshPtyRecoveryActivationLease
-}>
+import {
+  acceptsSourceFrame,
+  activePredecessor,
+  sameReceivingActivation,
+  sameRejectedSourceIdentity,
+  settleExactSourceDeliveryCancellation,
+  settledReceivingActivationLease,
+  type PendingSshPtySourceData,
+  type RejectedSourceIdentity,
+  type SourceDeliveryLeaseState,
+  type SourceDeliveryState,
+  type SshPtyRecoveryActivationLease,
+  type SshPtyRejectedSourceRecovery,
+  type SshPtySourceDeliveryLease
+} from './ssh-pty-source-delivery-state'
 
 export class SshPtySourceDeliveryLedger {
   private readonly deliveryByPty = new Map<string, SourceDeliveryState>()
@@ -95,6 +76,32 @@ export class SshPtySourceDeliveryLedger {
     }
   }
 
+  async reject(
+    relayPtyId: string,
+    identity: RejectedSourceIdentity | undefined
+  ): Promise<SshPtyRejectedSourceRecovery> {
+    if (!identity) {
+      return 'reconnect-channel'
+    }
+    const offeredCurrent = this.deliveryByPty.get(relayPtyId)
+    const matched = Boolean(
+      offeredCurrent && sameRejectedSourceIdentity(offeredCurrent.activation, identity)
+    )
+    const canceled = await settleExactSourceDeliveryCancellation(this.mux, relayPtyId, identity)
+    if (matched && canceled) {
+      const current = this.deliveryByPty.get(relayPtyId)
+      if (current && sameRejectedSourceIdentity(current.activation, identity)) {
+        this.retire(relayPtyId, current.previous, current.lease)
+        return 'fresh-activation'
+      }
+      return current ? 'reconnect-channel' : 'fresh-activation'
+    }
+    if (!matched && !canceled) {
+      return 'confirm-existing'
+    }
+    return 'reconnect-channel'
+  }
+
   private installProvisional(
     relayPtyId: string,
     activation: PtySourceReceivingActivation,
@@ -245,76 +252,3 @@ export class SshPtySourceDeliveryLedger {
     }
   }
 }
-
-function settledReceivingActivationLease(): SshPtySourceDeliveryLease {
-  return Object.freeze({
-    commit: () => {},
-    rollback: async () => true,
-    transferToRecovery: () => Object.freeze({ commit: () => {}, retire: () => {} })
-  })
-}
-
-function activePredecessor(previous?: SourceDeliveryState): SourceDeliveryState | undefined {
-  while (previous?.lease.phase === 'retired') {
-    previous = previous.previous
-  }
-  return previous
-}
-
-function sameReceivingActivation(
-  left: PtySourceReceivingActivation,
-  right: PtySourceReceivingActivation
-): boolean {
-  return (
-    left.clientGeneration === right.clientGeneration &&
-    left.ownerGeneration === right.ownerGeneration &&
-    left.ptyIncarnation === right.ptyIncarnation &&
-    left.deliveryToken === right.deliveryToken &&
-    left.checkpointSourceEndSu === right.checkpointSourceEndSu &&
-    left.recoveryEndSu === right.recoveryEndSu
-  )
-}
-
-function acceptsSourceFrame(
-  current: SourceDeliveryState | undefined,
-  params: Record<string, unknown>,
-  source: SshPtySourceFrame
-): current is SourceDeliveryState {
-  return Boolean(
-    current &&
-    current.lease.phase !== 'retired' &&
-    !current.lease.exited &&
-    current.activation.ptyIncarnation === params.ptyIncarnation &&
-    current.activation.deliveryToken === source.deliveryToken &&
-    current.activation.clientGeneration === source.clientGeneration &&
-    current.activation.ownerGeneration === source.ownerGeneration &&
-    current.sourceEndSu === source.sourceStartSu
-  )
-}
-
-async function settleExactSourceDeliveryCancellation(
-  mux: SshChannelMultiplexer,
-  relayPtyId: string,
-  activation: PtySourceReceivingActivation
-): Promise<boolean> {
-  try {
-    const result = (await mux.request('pty.cancelDelivery', {
-      id: relayPtyId,
-      clientGeneration: activation.clientGeneration,
-      ownerGeneration: activation.ownerGeneration,
-      deliveryToken: activation.deliveryToken
-    })) as Record<string, unknown>
-    return (
-      result.canceled === true &&
-      nonNegativeSafeInteger(result.sentEndSu) &&
-      nonNegativeSafeInteger(result.creditedEndSu) &&
-      result.creditedEndSu <= result.sentEndSu
-    )
-  } catch {
-    return false
-  }
-}
-
-function nonNegativeSafeInteger(value: unknown): value is number {
-  return Number.isSafeInteger(value) && (value as number) >= 0
-}
diff --git a/src/main/providers/ssh-pty-source-delivery-state.ts b/src/main/providers/ssh-pty-source-delivery-state.ts
new file mode 100644
index 000000000..efdcbfb58
--- /dev/null
+++ b/src/main/providers/ssh-pty-source-delivery-state.ts
@@ -0,0 +1,134 @@
+import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
+import type { PtySourceReceivingActivation } from '../../shared/pty-source-receiving-activation'
+import type { SshPtySourceFrame } from './ssh-pty-source-frame'
+
+export type PendingSshPtySourceData = Readonly<{
+  relayPtyId: string
+  params: Record<string, unknown>
+  data: string
+  source?: SshPtySourceFrame
+}>
+
+export type SourceDeliveryLeaseState = {
+  phase: 'provisional' | 'recovery' | 'committing' | 'committed' | 'retired'
+  pendingData: PendingSshPtySourceData[]
+  recoverySink?: (pending: PendingSshPtySourceData) => void
+  exited: boolean
+}
+
+export type SourceDeliveryState = Readonly<{
+  activation: PtySourceReceivingActivation
+  sourceEndSu: number
+  lease: SourceDeliveryLeaseState
+  previous?: SourceDeliveryState
+}>
+
+export type SshPtyRecoveryActivationLease = Readonly<{
+  commit: () => void
+  retire: () => void
+}>
+
+export type SshPtySourceDeliveryLease = Readonly<{
+  commit: () => void
+  rollback: () => Promise<boolean>
+  transferToRecovery: (
+    sink: (pending: PendingSshPtySourceData) => void
+  ) => SshPtyRecoveryActivationLease
+}>
+
+export type SshPtyRejectedSourceRecovery =
+  | 'confirm-existing'
+  | 'fresh-activation'
+  | 'reconnect-channel'
+
+export type RejectedSourceIdentity = Readonly<{
+  clientGeneration: number
+  ownerGeneration: number
+  ptyIncarnation: string
+  deliveryToken: string
+}>
+
+export function settledReceivingActivationLease(): SshPtySourceDeliveryLease {
+  return Object.freeze({
+    commit: () => {},
+    rollback: async () => true,
+    transferToRecovery: () => Object.freeze({ commit: () => {}, retire: () => {} })
+  })
+}
+
+export function activePredecessor(previous?: SourceDeliveryState): SourceDeliveryState | undefined {
+  while (previous?.lease.phase === 'retired') {
+    previous = previous.previous
+  }
+  return previous
+}
+
+export function sameReceivingActivation(
+  left: PtySourceReceivingActivation,
+  right: PtySourceReceivingActivation
+): boolean {
+  return (
+    left.clientGeneration === right.clientGeneration &&
+    left.ownerGeneration === right.ownerGeneration &&
+    left.ptyIncarnation === right.ptyIncarnation &&
+    left.deliveryToken === right.deliveryToken &&
+    left.checkpointSourceEndSu === right.checkpointSourceEndSu &&
+    left.recoveryEndSu === right.recoveryEndSu
+  )
+}
+
+export function sameRejectedSourceIdentity(
+  left: PtySourceReceivingActivation,
+  right: RejectedSourceIdentity
+): boolean {
+  return (
+    left.clientGeneration === right.clientGeneration &&
+    left.ownerGeneration === right.ownerGeneration &&
+    left.ptyIncarnation === right.ptyIncarnation &&
+    left.deliveryToken === right.deliveryToken
+  )
+}
+
+export function acceptsSourceFrame(
+  current: SourceDeliveryState | undefined,
+  params: Record<string, unknown>,
+  source: SshPtySourceFrame
+): current is SourceDeliveryState {
+  return Boolean(
+    current &&
+    current.lease.phase !== 'retired' &&
+    !current.lease.exited &&
+    current.activation.ptyIncarnation === params.ptyIncarnation &&
+    current.activation.deliveryToken === source.deliveryToken &&
+    current.activation.clientGeneration === source.clientGeneration &&
+    current.activation.ownerGeneration === source.ownerGeneration &&
+    current.sourceEndSu === source.sourceStartSu
+  )
+}
+
+export async function settleExactSourceDeliveryCancellation(
+  mux: SshChannelMultiplexer,
+  relayPtyId: string,
+  activation: RejectedSourceIdentity
+): Promise<boolean> {
+  try {
+    const result = (await mux.request('pty.cancelDelivery', {
+      id: relayPtyId,
+      clientGeneration: activation.clientGeneration,
+      ownerGeneration: activation.ownerGeneration,
+      deliveryToken: activation.deliveryToken
+    })) as Record<string, unknown>
+    return (
+      result.canceled === true &&
+      nonNegativeSafeInteger(result.sentEndSu) &&
+      nonNegativeSafeInteger(result.creditedEndSu) &&
+      result.creditedEndSu <= result.sentEndSu
+    )
+  } catch {
+    return false
+  }
+}
+
+function nonNegativeSafeInteger(value: unknown): value is number {
+  return Number.isSafeInteger(value) && (value as number) >= 0
+}
diff --git a/src/main/ssh/ssh-pty-frame-rejection.ts b/src/main/ssh/ssh-pty-frame-rejection.ts
new file mode 100644
index 000000000..07dcf3827
--- /dev/null
+++ b/src/main/ssh/ssh-pty-frame-rejection.ts
@@ -0,0 +1,115 @@
+import type { SshPtyDataCallback } from '../providers/ssh-pty-provider-contract'
+import type { SshPtyConsumerOwnerState } from './ssh-pty-consumer-session'
+
+type SshPtyDataPayload = Parameters<SshPtyDataCallback>[0]
+
+const SSH_PTY_FRAME_REJECTION_LOG_KEY_LIMIT = 12
+
+export type SshPtyFrameRejectionReason =
+  | 'source-missing'
+  | 'source-malformed'
+  | 'client-generation-stale'
+  | 'owner-generation-stale'
+  | 'client-and-owner-generation-stale'
+  | 'source-range-invalid'
+
+export type SshPtyFrameRejection = Readonly<{
+  reason: SshPtyFrameRejectionReason
+  action: 'quarantine-legacy-frame' | 'retire-and-reattach-delivery'
+}>
+
+// Why a source-less frame is quarantined rather than reattached: it is the signature of legacy
+// output already in flight when the flow-control grant landed, not of a broken delivery. The relay's
+// own admission stops publishing those once the grant is live, and source recovery replays from
+// sourceStartSu, so dropping them loses nothing. Reattaching instead would restart the PTY on every
+// openClient handshake.
+export function classifySshPtyFrameRejection(
+  payload: SshPtyDataPayload,
+  owner: SshPtyConsumerOwnerState | null
+): SshPtyFrameRejection | null {
+  if (payload.sourceMalformed) {
+    return Object.freeze({
+      reason: 'source-malformed',
+      action: 'retire-and-reattach-delivery'
+    })
+  }
+  const source = payload.source
+  if (!source) {
+    return owner?.outputFlowControl
+      ? Object.freeze({
+          reason: 'source-missing',
+          action: 'quarantine-legacy-frame'
+        })
+      : null
+  }
+  if (!owner?.outputFlowControl) {
+    return payload.sourceRejected
+      ? Object.freeze({
+          reason: 'source-range-invalid',
+          action: 'retire-and-reattach-delivery'
+        })
+      : null
+  }
+  const staleClient = source.clientGeneration !== owner.clientGeneration
+  const staleOwner = source.ownerGeneration !== owner.ownerGeneration
+  if (staleClient || staleOwner) {
+    return Object.freeze({
+      reason:
+        staleClient && staleOwner
+          ? 'client-and-owner-generation-stale'
+          : staleClient
+            ? 'client-generation-stale'
+            : 'owner-generation-stale',
+      action: 'retire-and-reattach-delivery'
+    })
+  }
+  if (payload.sourceRejected) {
+    return Object.freeze({
+      reason: 'source-range-invalid',
+      action: 'retire-and-reattach-delivery'
+    })
+  }
+  return null
+}
+
+/**
+ * Why bounded: a quarantined frame carries no delivery token to retire, so nothing dedupes it — a
+ * relay that keeps publishing legacy output to a flow-control owner would log once per frame and
+ * bury every other diagnostic. One line per PTY and reason per generation names the fault; the
+ * generation is itself the counter for how often it recurs.
+ */
+export class SshPtyFrameRejectionLog {
+  private generation: number | null = null
+  private readonly loggedKeys = new Set<string>()
+
+  record(
+    payload: SshPtyDataPayload,
+    owner: SshPtyConsumerOwnerState | null,
+    rejection: SshPtyFrameRejection
+  ): void {
+    if (this.generation !== payload.providerGeneration) {
+      this.generation = payload.providerGeneration
+      this.loggedKeys.clear()
+    }
+    const key = `${payload.id}\0${rejection.reason}`
+    if (this.loggedKeys.has(key) || this.loggedKeys.size >= SSH_PTY_FRAME_REJECTION_LOG_KEY_LIMIT) {
+      return
+    }
+    this.loggedKeys.add(key)
+    console.warn('[ssh-relay-session] Rejected PTY delivery', {
+      ptyId: payload.id,
+      providerGeneration: payload.providerGeneration,
+      expectedClientGeneration: owner?.clientGeneration ?? null,
+      expectedOwnerGeneration: owner?.ownerGeneration ?? null,
+      offeredClientGeneration: payload.source?.clientGeneration ?? null,
+      offeredOwnerGeneration: payload.source?.ownerGeneration ?? null,
+      sourceState: rejection.reason,
+      recoveryAction: rejection.action
+    })
+  }
+
+  clear(): void {
+    this.generation = null
+    this.loggedKeys.clear()
+  }
+}
diff --git a/src/main/ssh/ssh-pty-targeted-reattach-queue.ts b/src/main/ssh/ssh-pty-targeted-reattach-queue.ts
new file mode 100644
index 000000000..23329abf5
--- /dev/null
+++ b/src/main/ssh/ssh-pty-targeted-reattach-queue.ts
@@ -0,0 +1,57 @@
+type TargetedReattach = Readonly<{ start: () => void }>
+
+// Why bounded: every rejected frame asks for its own attach round trip, so a relay that starts
+// rejecting across many PTYs at once fans out one concurrent reattach per PTY. The bulk reconnect
+// path caps the identical work, and each attach also costs a lease read and a lease write.
+export class SshPtyTargetedReattachQueue {
+  private readonly running = new Map<string, TargetedReattach>()
+  private readonly waiting: TargetedReattach[] = []
+  private active = 0
+
+  constructor(private readonly maxConcurrency: number) {}
+
+  has(key: string): boolean {
+    return this.running.has(key)
+  }
+
+  run(key: string, task: () => Promise<boolean>): Promise<boolean> {
+    return new Promise<boolean>((resolve, reject) => {
+      const entry: TargetedReattach = {
+        start: () => {
+          this.active++
+          task().then(
+            (recovered) => {
+              this.settle(key, entry)
+              resolve(recovered)
+            },
+            (error: unknown) => {
+              this.settle(key, entry)
+              reject(error instanceof Error ? error : new Error(String(error)))
+            }
+          )
+        }
+      }
+      this.running.set(key, entry)
+      if (this.active < this.maxConcurrency) {
+        entry.start()
+        return
+      }
+      this.waiting.push(entry)
+    })
+  }
+
+  // Why queued entries are dropped rather than started: each one is keyed to the provider generation
+  // the teardown just ended, so running it would attach onto a mux that is already gone.
+  clear(): void {
+    this.waiting.length = 0
+    this.running.clear()
+  }
+
+  private settle(key: string, entry: TargetedReattach): void {
+    if (this.running.get(key) === entry) {
+      this.running.delete(key)
+    }
+    this.active--
+    this.waiting.shift()?.start()
+  }
+}
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 0966b07fd..2eca6802f 100644
--- a/src/main/ssh/ssh-relay-session-data-delivery.test.ts
+++ b/src/main/ssh/ssh-relay-session-data-delivery.test.ts
@@ -519,7 +519,7 @@ describe('SshRelaySession data delivery', () => {
     )
   })
 
-  it('rejects missing negotiated source identity before main admission', async () => {
+  it('quarantines missing negotiated source identity before main admission', async () => {
     const { mockConn, mockStore, mockPortForward, getMainWindow } = createMockDeps()
     const session = new SshRelaySession('target-1', getMainWindow, mockStore, mockPortForward)
     await session.establish(mockConn)
@@ -536,10 +536,8 @@ describe('SshRelaySession data delivery', () => {
     })
 
     expect(acceptOutputDataMock).not.toHaveBeenCalled()
-    expect(closeSshPtyOutputGeneration).toHaveBeenCalledWith(
-      23,
-      'ssh_source_frame_malformed_or_missing'
-    )
+    expect(closeSshPtyOutputGeneration).not.toHaveBeenCalled()
+    expect(muxDisposeMock).not.toHaveBeenCalled()
   })
 
   it('keeps unoffered source metadata out of legacy intake', async () => {
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 e44359ba5..26dd798ac 100644
--- a/src/main/ssh/ssh-relay-session-recovery-races.test.ts
+++ b/src/main/ssh/ssh-relay-session-recovery-races.test.ts
@@ -420,6 +420,7 @@ describe('SshRelaySession recovery race fencing', () => {
     })
     await session.reconnect(deps.mockConn)
     const closeCount = vi.mocked(closeSshPtyOutputGeneration).mock.calls.length
+    const muxDisposeCount = muxDisposeMock.mock.calls.length
 
     emitSourceFrame({
       targetId,
@@ -431,11 +432,9 @@ describe('SshRelaySession recovery race fencing', () => {
     })
 
     expect(acceptOutputDataMock).not.toHaveBeenCalled()
-    expect(closeSshPtyOutputGeneration).toHaveBeenCalledTimes(closeCount + 1)
-    expect(closeSshPtyOutputGeneration).toHaveBeenLastCalledWith(
-      23,
-      'ssh_source_frame_stale_or_non_contiguous'
-    )
+    expect(closeSshPtyOutputGeneration).toHaveBeenCalledTimes(closeCount)
+    expect(muxDisposeMock).toHaveBeenCalledTimes(muxDisposeCount)
+    await vi.waitFor(() => expect(attachForReconnectMock).toHaveBeenCalledTimes(2))
   })
 
   it('drops late frames from a token after its cancellation proof is validated', async () => {
diff --git a/src/main/ssh/ssh-relay-session-rejected-delivery.test.ts b/src/main/ssh/ssh-relay-session-rejected-delivery.test.ts
new file mode 100644
index 000000000..e0f1572d4
--- /dev/null
+++ b/src/main/ssh/ssh-relay-session-rejected-delivery.test.ts
@@ -0,0 +1,496 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { SshPtyDataCallback } from '../providers/ssh-pty-provider-contract'
+import type { SshPtyProvider } from '../providers/ssh-pty-provider'
+import type { SshChannelMultiplexer } from './ssh-channel-multiplexer'
+import type { SshPtyConsumerSessionState } from './ssh-pty-consumer-session'
+import { SshRelaySession } from './ssh-relay-session'
+import { createMockDeps } from './ssh-relay-session-test-fixtures'
+
+const { acceptOutputDataMock, getSshPtyProviderMock } = vi.hoisted(() => ({
+  acceptOutputDataMock: vi.fn().mockResolvedValue(undefined),
+  getSshPtyProviderMock: vi.fn()
+}))
+
+vi.mock('../ipc/ssh-pty-output-intake-registry', async (importOriginal) => {
+  const original = (await importOriginal()) as object
+  return { ...original, acceptSshPtyOutputData: acceptOutputDataMock }
+})
+
+vi.mock('../ipc/pty', async (importOriginal) => {
+  const original = (await importOriginal()) as object
+  return { ...original, getSshPtyProvider: getSshPtyProviderMock }
+})
+
+type SshPtyDataPayload = Parameters<SshPtyDataCallback>[0]
+
+type RejectedDeliverySession = {
+  mux: SshChannelMultiplexer | null
+  activePtyProviderGeneration: number | null
+  ptyConsumerSessionState: SshPtyConsumerSessionState | null
+  acceptPtyData: (payload: SshPtyDataPayload) => Promise<unknown>
+  reattachKnownPty: (args: {
+    ptyId: string
+    activeLeaseByPtyId: Map<string, unknown>
+    expectedIdentityByPtyId: Map<string, unknown>
+    attachedLeaseIds: Set<string>
+    targetedDeliveryRecovery?: 'confirm-existing' | 'fresh-activation'
+  }) => Promise<void>
+  reattachRejectedPty: (
+    relayPtyId: string,
+    mux: SshChannelMultiplexer,
+    providerGeneration: number,
+    targetedDeliveryRecovery: 'confirm-existing' | 'fresh-activation'
+  ) => Promise<boolean>
+  sourceRecoveryRequest: (appPtyId: string) => Promise<
+    | {
+        status: 'checkpoint'
+        clientGeneration: number
+        ownerGeneration: number
+        ptyIncarnation: string
+        deliveryToken: string
+        acceptedSourceEndSu: number
+      }
+    | undefined
+  >
+  rejectedPtyRecoveryAttempts: Map<string, unknown>
+  sourceIdentityByRelayPtyId: Map<string, unknown>
+  retireExitedPty: (payload: {
+    id: string
+    code: number
+    providerGeneration: number
+    ptyIncarnation: string
+  }) => void
+}
+
+function source(overrides: Partial<NonNullable<SshPtyDataPayload['source']>> = {}) {
+  return {
+    relayPtyId: 'pty-bad',
+    spanId: 'token-bad:0:4',
+    clientGeneration: 1,
+    ownerGeneration: 1,
+    deliveryToken: 'token-bad',
+    sourceStartSu: 0,
+    sourceEndSu: 4,
+    ...overrides
+  }
+}
+
+function rejectedPayload(overrides: Partial<SshPtyDataPayload> = {}): SshPtyDataPayload {
+  return {
+    id: 'ssh:target-1@@pty-bad',
+    data: 'rejected',
+    providerGeneration: 23,
+    ptyIncarnation: 'incarnation-bad',
+    source: source(),
+    sourceRejected: true,
+    ...overrides
+  }
+}
+
+function prepareSession() {
+  const deps = createMockDeps()
+  const mux = {
+    isDisposed: vi.fn(() => false),
+    dispose: vi.fn()
+  } as unknown as SshChannelMultiplexer
+  const session = new SshRelaySession(
+    'target-1',
+    deps.getMainWindow,
+    deps.mockStore,
+    deps.mockPortForward
+  )
+  const internals = session as unknown as RejectedDeliverySession
+  internals.mux = mux
+  internals.activePtyProviderGeneration = 23
+  internals.ptyConsumerSessionState = {
+    mode: 'negotiated',
+    clientInstanceId: 'client-1',
+    clientGeneration: 1,
+    ownerGeneration: 1,
+    ownerLease: 'owner-lease',
+    outputFlowControl: { version: 1, windowSu: 64 }
+  }
+  return { deps, internals, mux, session }
+}
+
+describe('SshRelaySession rejected PTY delivery recovery', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('quarantines source-less output queued before a flow-control grant', async () => {
+    const { internals, mux } = prepareSession()
+    const reattach = vi.fn().mockResolvedValue(true)
+    internals.reattachRejectedPty = reattach
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+    await internals.acceptPtyData(rejectedPayload({ source: undefined, sourceRejected: undefined }))
+
+    expect(warn).toHaveBeenCalledWith(
+      '[ssh-relay-session] Rejected PTY delivery',
+      expect.objectContaining({
+        sourceState: 'source-missing',
+        recoveryAction: 'quarantine-legacy-frame'
+      })
+    )
+    expect(reattach).not.toHaveBeenCalled()
+    expect(acceptOutputDataMock).not.toHaveBeenCalled()
+    expect(mux.dispose).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    ['client', 9, 1],
+    ['owner', 1, 9]
+  ] as const)(
+    'retires a superseded %s generation without disposing the mux',
+    async (_generation, clientGeneration, ownerGeneration) => {
+      const { internals, mux } = prepareSession()
+      const reattach = vi.fn().mockResolvedValue(true)
+      internals.reattachRejectedPty = reattach
+      vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+      await internals.acceptPtyData(
+        rejectedPayload({ source: source({ clientGeneration, ownerGeneration }) })
+      )
+
+      expect(reattach).toHaveBeenCalledWith('pty-bad', mux, 23, 'confirm-existing')
+      expect(acceptOutputDataMock).not.toHaveBeenCalled()
+      expect(mux.dispose).not.toHaveBeenCalled()
+    }
+  )
+
+  it('logs malformed source metadata and reattaches from the exact outer PTY identity', async () => {
+    const { internals } = prepareSession()
+    const reattach = vi.fn().mockResolvedValue(true)
+    internals.reattachRejectedPty = reattach
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+    await internals.acceptPtyData(
+      rejectedPayload({ source: undefined, sourceMalformed: true, sourceRejected: undefined })
+    )
+
+    expect(warn).toHaveBeenCalledWith(
+      '[ssh-relay-session] Rejected PTY delivery',
+      expect.objectContaining({
+        ptyId: 'ssh:target-1@@pty-bad',
+        providerGeneration: 23,
+        expectedClientGeneration: 1,
+        expectedOwnerGeneration: 1,
+        sourceState: 'source-malformed',
+        recoveryAction: 'retire-and-reattach-delivery'
+      })
+    )
+    expect(reattach).toHaveBeenCalledWith('pty-bad', expect.anything(), 23, 'confirm-existing')
+    expect(acceptOutputDataMock).not.toHaveBeenCalled()
+  })
+
+  it('reattaches only the rejected PTY while a healthy sibling keeps delivering', async () => {
+    const { deps, internals, mux } = prepareSession()
+    const provider = {} as SshPtyProvider
+    getSshPtyProviderMock.mockReturnValue(provider)
+    vi.mocked(deps.mockStore.getSshRemotePtyLeases).mockReturnValue([
+      {
+        targetId: 'target-1',
+        ptyId: 'pty-bad',
+        state: 'detached',
+        createdAt: 1,
+        updatedAt: 1
+      },
+      {
+        targetId: 'target-1',
+        ptyId: 'pty-healthy',
+        state: 'attached',
+        createdAt: 1,
+        updatedAt: 1
+      }
+    ])
+    const reattachKnownPty = vi.fn(
+      async (args: Parameters<RejectedDeliverySession['reattachKnownPty']>[0]) => {
+        args.attachedLeaseIds.add(args.ptyId)
+      }
+    )
+    internals.reattachKnownPty = reattachKnownPty
+
+    await internals.acceptPtyData(rejectedPayload({ rejectedSourceRecovery: 'fresh-activation' }))
+    await vi.waitFor(() => expect(reattachKnownPty).toHaveBeenCalledOnce())
+    await internals.acceptPtyData(
+      rejectedPayload({
+        id: 'ssh:target-1@@pty-healthy',
+        data: 'healthy',
+        ptyIncarnation: 'incarnation-healthy',
+        source: source({
+          relayPtyId: 'pty-healthy',
+          spanId: 'token-healthy:0:7',
+          deliveryToken: 'token-healthy',
+          sourceEndSu: 7
+        }),
+        sourceRejected: undefined
+      })
+    )
+
+    const recovery = reattachKnownPty.mock.calls[0]?.[0]
+    expect(recovery?.ptyId).toBe('pty-bad')
+    expect(Array.from(recovery?.activeLeaseByPtyId.keys() ?? [])).toEqual(['pty-bad'])
+    expect(Array.from(recovery?.expectedIdentityByPtyId.keys() ?? [])).toEqual([])
+    expect(recovery?.targetedDeliveryRecovery).toBe('fresh-activation')
+    expect(deps.mockStore.markSshRemotePtyLeasesAttachedAsync).toHaveBeenCalledWith('target-1', [
+      'pty-bad'
+    ])
+    expect(acceptOutputDataMock).toHaveBeenCalledOnce()
+    expect(acceptOutputDataMock).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'ssh:target-1@@pty-healthy', data: 'healthy' })
+    )
+    expect(mux.dispose).not.toHaveBeenCalled()
+  })
+
+  it('reconnects instead of canceling an unprovable malformed delivery', async () => {
+    const { internals, mux } = prepareSession()
+    const reattach = vi.fn().mockResolvedValue(true)
+    internals.reattachRejectedPty = reattach
+    vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+    await internals.acceptPtyData(
+      rejectedPayload({
+        source: undefined,
+        sourceMalformed: true,
+        sourceRejected: undefined,
+        rejectedSourceRecovery: 'reconnect-channel'
+      })
+    )
+
+    expect(mux.dispose).toHaveBeenCalledWith('connection_lost')
+    expect(reattach).not.toHaveBeenCalled()
+  })
+
+  it('opens a fresh activation after exact rejected-delivery cancellation', async () => {
+    const { deps, internals, mux } = prepareSession()
+    internals.sourceIdentityByRelayPtyId.set('pty-bad', {
+      clientGeneration: 1,
+      ownerGeneration: 1,
+      ptyIncarnation: 'incarnation-bad',
+      deliveryToken: 'token-old',
+      nextSourceSu: 4
+    })
+    const commit = vi.fn()
+    const attachForReconnect = vi.fn(async () => ({
+      incarnationId: 'incarnation-bad',
+      sourceActivation: {
+        status: 'pending' as const,
+        clientGeneration: 1,
+        ownerGeneration: 1,
+        ptyIncarnation: 'incarnation-bad',
+        deliveryToken: 'token-fresh',
+        checkpointSourceEndSu: 0,
+        recoveryEndSu: 0
+      },
+      sourceActivationLease: { commit, rollback: vi.fn(async () => true) }
+    }))
+    getSshPtyProviderMock.mockReturnValue({ attachForReconnect } as unknown as SshPtyProvider)
+
+    await expect(
+      internals.reattachRejectedPty('pty-bad', mux, 23, 'fresh-activation')
+    ).resolves.toBe(true)
+
+    expect(attachForReconnect).toHaveBeenCalledWith('pty-bad')
+    expect(commit).toHaveBeenCalledOnce()
+    expect(deps.mockStore.markSshRemotePtyLeasesAttachedAsync).toHaveBeenCalledWith('target-1', [
+      'pty-bad'
+    ])
+    await internals.acceptPtyData(
+      rejectedPayload({
+        data: 'fresh',
+        sourceRejected: undefined,
+        source: source({ deliveryToken: 'token-fresh' })
+      })
+    )
+    expect(acceptOutputDataMock).toHaveBeenCalledWith(expect.objectContaining({ data: 'fresh' }))
+    expect(mux.dispose).not.toHaveBeenCalled()
+  })
+
+  it('accepts an exact existing activation as stale-frame confirmation', async () => {
+    const { internals, mux } = prepareSession()
+    const checkpoint = {
+      status: 'checkpoint' as const,
+      clientGeneration: 1,
+      ownerGeneration: 1,
+      ptyIncarnation: 'incarnation-bad',
+      deliveryToken: 'token-current',
+      acceptedSourceEndSu: 7
+    }
+    internals.sourceRecoveryRequest = vi.fn(async () => checkpoint)
+    const commit = vi.fn()
+    const attachForReconnect = vi.fn(async () => ({
+      incarnationId: 'incarnation-bad',
+      sourceActivation: {
+        status: 'pending' as const,
+        clientGeneration: 1,
+        ownerGeneration: 1,
+        ptyIncarnation: 'incarnation-bad',
+        deliveryToken: 'token-current',
+        checkpointSourceEndSu: 0,
+        recoveryEndSu: 0
+      },
+      sourceActivationLease: { commit, rollback: vi.fn(async () => true) }
+    }))
+    getSshPtyProviderMock.mockReturnValue({ attachForReconnect } as unknown as SshPtyProvider)
+
+    await expect(
+      internals.reattachRejectedPty('pty-bad', mux, 23, 'confirm-existing')
+    ).resolves.toBe(true)
+
+    expect(attachForReconnect).toHaveBeenCalledWith('pty-bad', undefined, checkpoint)
+    expect(commit).toHaveBeenCalledOnce()
+    expect(mux.dispose).not.toHaveBeenCalled()
+  })
+
+  it('forgets rejected-delivery recovery history when the PTY exits', () => {
+    const { internals } = prepareSession()
+    internals.rejectedPtyRecoveryAttempts.set('ssh:target-1@@pty-bad', {})
+
+    internals.retireExitedPty({
+      id: 'ssh:target-1@@pty-bad',
+      code: 0,
+      providerGeneration: 23,
+      ptyIncarnation: 'incarnation-bad'
+    })
+
+    expect(internals.rejectedPtyRecoveryAttempts).toHaveLength(0)
+  })
+
+  // Why a channel drop rather than a terminal relay error: a terminal error clears the reconnect
+  // backoff and rotates provider authority, aborting every fs and git request on the target, so one
+  // PTY's undeliverable output would strand the whole connection in manual recovery.
+  it('bounds failed targeted recovery and escalates to a recoverable relay reconnect', async () => {
+    const { internals, mux, session } = prepareSession()
+    const reattach = vi.fn().mockResolvedValue(false)
+    internals.reattachRejectedPty = reattach
+    getSshPtyProviderMock.mockReturnValue({ hasPty: () => true } as unknown as SshPtyProvider)
+    const onTerminalError = vi.fn()
+    session.setOnTerminalRelayError(onTerminalError)
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+    await internals.acceptPtyData(rejectedPayload())
+    await vi.waitFor(() => expect(mux.dispose).toHaveBeenCalledOnce(), { timeout: 2000 })
+
+    expect(reattach).toHaveBeenCalledTimes(2)
+    expect(reattach.mock.calls).toEqual([
+      ['pty-bad', mux, 23, 'confirm-existing'],
+      ['pty-bad', mux, 23, 'confirm-existing']
+    ])
+    expect(mux.dispose).toHaveBeenCalledWith('connection_lost')
+    expect(warn).toHaveBeenCalledWith(
+      expect.stringContaining('PTY pty-bad delivery recovery exhausted')
+    )
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(acceptOutputDataMock).not.toHaveBeenCalled()
+  })
+
+  it('stops without an error when the rejected PTY exited during recovery', async () => {
+    const { internals, mux, session } = prepareSession()
+    const reattach = vi.fn().mockResolvedValue(false)
+    internals.reattachRejectedPty = reattach
+    // Why: reattachKnownPty resolves without claiming the lease when the PTY exits mid-attach, so a
+    // plain "not recovered" is indistinguishable from a failure until liveness is checked.
+    getSshPtyProviderMock.mockReturnValue({ hasPty: () => false } as unknown as SshPtyProvider)
+    const onTerminalError = vi.fn()
+    session.setOnTerminalRelayError(onTerminalError)
+
+    await internals.acceptPtyData(rejectedPayload())
+    await vi.waitFor(() => expect(reattach).toHaveBeenCalledOnce())
+    await new Promise((resolve) => setTimeout(resolve, 400))
+
+    expect(reattach).toHaveBeenCalledOnce()
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(mux.dispose).not.toHaveBeenCalled()
+  })
+
+  // Why bounded: the bulk reconnect path caps the identical attach work at 8, and a relay that
+  // starts rejecting across every PTY at once would otherwise open one attach round trip per PTY.
+  it('caps concurrent targeted reattaches and coalesces repeats for one PTY', async () => {
+    const { internals } = prepareSession()
+    getSshPtyProviderMock.mockReturnValue({ hasPty: () => true } as unknown as SshPtyProvider)
+    vi.spyOn(console, 'warn').mockImplementation(() => {})
+    const release: (() => void)[] = []
+    let active = 0
+    let peak = 0
+    const reattach = vi.fn(async () => {
+      active++
+      peak = Math.max(peak, active)
+      await new Promise<void>((resolve) => release.push(resolve))
+      active--
+      return true
+    })
+    internals.reattachRejectedPty = reattach
+
+    for (let index = 0; index < 20; index++) {
+      const relayPtyId = `pty-${index}`
+      await internals.acceptPtyData(
+        rejectedPayload({
+          id: `ssh:target-1@@${relayPtyId}`,
+          source: source({ relayPtyId, deliveryToken: `token-${index}` })
+        })
+      )
+      // Why twice: a repeat for a PTY already recovering must not consume a second slot.
+      await internals.acceptPtyData(
+        rejectedPayload({
+          id: `ssh:target-1@@${relayPtyId}`,
+          source: source({ relayPtyId, deliveryToken: `token-${index}-repeat` })
+        })
+      )
+    }
+
+    expect(peak).toBe(8)
+    expect(reattach).toHaveBeenCalledTimes(8)
+    for (const resolve of release.splice(0)) {
+      resolve()
+    }
+    await vi.waitFor(() => expect(reattach).toHaveBeenCalledTimes(16))
+    expect(peak).toBe(8)
+    for (const resolve of release.splice(0)) {
+      resolve()
+    }
+  })
+
+  it('does not let accepted frames refill the recovery budget indefinitely', async () => {
+    const { internals, mux, session } = prepareSession()
+    const reattach = vi.fn().mockResolvedValue(true)
+    internals.reattachRejectedPty = reattach
+    getSshPtyProviderMock.mockReturnValue({ hasPty: () => true } as unknown as SshPtyProvider)
+    const onTerminalError = vi.fn()
+    session.setOnTerminalRelayError(onTerminalError)
+
+    // Why alternating, with a fresh bad token each round: every rejection retires its own delivery,
+    // so a flapping PTY only keeps asking for recovery by moving onto new ones, and the accepted
+    // frame in between is what used to clear the budget outright. Each reattach here succeeds, so
+    // the consecutive budget is cleared legitimately too — only the per-generation ceiling can stop
+    // it.
+    for (let round = 0; round < 40; round++) {
+      await internals.acceptPtyData(
+        rejectedPayload({
+          source: source({
+            spanId: `token-bad-${round}:0:4`,
+            deliveryToken: `token-bad-${round}`,
+            clientGeneration: 9
+          })
+        })
+      )
+      await internals.acceptPtyData(
+        rejectedPayload({
+          sourceRejected: undefined,
+          source: source({
+            spanId: `token-good:${round * 4}:${round * 4 + 4}`,
+            deliveryToken: 'token-good',
+            sourceStartSu: round * 4,
+            sourceEndSu: round * 4 + 4
+          })
+        })
+      )
+      await Promise.resolve()
+    }
+    await vi.waitFor(() => expect(mux.dispose).toHaveBeenCalledOnce(), { timeout: 2000 })
+
+    expect(onTerminalError).not.toHaveBeenCalled()
+    expect(acceptOutputDataMock).toHaveBeenCalledTimes(40)
+    expect(reattach.mock.calls.length).toBeLessThanOrEqual(12)
+  })
+})
diff --git a/src/main/ssh/ssh-relay-session-test-fixtures.ts b/src/main/ssh/ssh-relay-session-test-fixtures.ts
index c5f084a87..48461af68 100644
--- a/src/main/ssh/ssh-relay-session-test-fixtures.ts
+++ b/src/main/ssh/ssh-relay-session-test-fixtures.ts
@@ -1,5 +1,6 @@
 import { vi, type Mock } from 'vitest'
 import type { BrowserWindow } from 'electron'
+import { PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR } from '../../shared/pty-consumer-session'
 import type { SshConnection } from './ssh-connection'
 import type { Store } from '../persistence'
 import type { SshPortForwardManager } from './ssh-port-forward'
@@ -53,3 +54,9 @@ export function mockDeploySuccess(): void {
     platform: 'linux-x64'
   })
 }
+
+export function createMismatchedOwnerRecoveryError(): unknown {
+  return Object.assign(new Error('Owner recovery lease is stale'), {
+    code: PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR
+  })
+}
diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts
index 2e8816dd8..f44d2a77d 100644
--- a/src/main/ssh/ssh-relay-session.ts
+++ b/src/main/ssh/ssh-relay-session.ts
@@ -79,6 +79,7 @@ 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 {
   isSshOwnerAdmissionBlocked,
   retrySshOwnerRecoveryWhileBlocked
@@ -121,6 +122,8 @@ import {
   getSshPtyConsumerRecovery,
   rememberSshPtyConsumerRecovery
 } from './ssh-pty-consumer-recovery'
+import { classifySshPtyFrameRejection, SshPtyFrameRejectionLog } from './ssh-pty-frame-rejection'
+import { SshPtyTargetedReattachQueue } from './ssh-pty-targeted-reattach-queue'
 
 export type RelaySessionState = 'idle' | 'deploying' | 'ready' | 'reconnecting' | 'disposed'
 
@@ -131,6 +134,12 @@ const SSH_PTY_REATTACH_MAX_CONCURRENCY = 8
 const SSH_PTY_REATTACH_ATTEMPT_TIMEOUT_MS = 10_000
 const SSH_PTY_REATTACH_RETRY_MIN_DELAY_MS = 50
 const SSH_PTY_REATTACH_RETRY_JITTER_MS = 200
+const SSH_REJECTED_PTY_RECOVERY_MAX_ATTEMPTS = 2
+// Why a second ceiling: the consecutive budget resets whenever a reattach succeeds, so a PTY that
+// alternates recovered and rejected frames would otherwise reattach forever — each one costs a
+// store read, an attach round trip and a store write.
+const SSH_REJECTED_PTY_RECOVERY_MAX_GENERATION_ATTEMPTS = 12
+const SSH_REJECTED_PTY_RECOVERY_RETRY_DELAY_MS = 150
 const SSH_SOURCE_RECOVERY_CANCELLATION_FAILED = 'ssh_source_recovery_cancellation_failed'
 
 // Why: superseded attempts stop quietly; a dead mux still owned by this attempt must enter recovery.
@@ -179,6 +188,7 @@ type RemoteCliBridgeEnv = {
 }
 
 type ExpectedPtyIdentity = { paneKey?: string; tabId?: string }
+type TargetedDeliveryRecovery = 'confirm-existing' | 'fresh-activation'
 
 function expectedIdentityForLease(lease: {
   tabId?: string
@@ -324,6 +334,20 @@ export class SshRelaySession {
     }>
   >()
   private readonly retiredSourceDeliveries = new SshPtyRetiredSourceDeliveries()
+  private readonly rejectedPtyRecoveryAttempts = new Map<
+    string,
+    {
+      providerGeneration: number
+      attempts: number
+      generationAttempts: number
+      reported: boolean
+    }
+  >()
+  private readonly rejectedPtyRecoveryRetries = new Set<ReturnType<typeof setTimeout>>()
+  private readonly rejectedPtyReattaches = new SshPtyTargetedReattachQueue(
+    SSH_PTY_REATTACH_MAX_CONCURRENCY
+  )
+  private readonly ptyFrameRejectionLog = new SshPtyFrameRejectionLog()
   private readonly ptyConsumerClientInstanceId: string
   private ptyConsumerSessionState: SshPtyConsumerSessionState | null = null
   private activeCompatibilityAttachmentIds = new Set<string>()
@@ -1072,13 +1096,28 @@ export class SshRelaySession {
     ownsAttempt: () => boolean
   ): Promise<SshPtyConsumerSessionState> {
     const previousOwner = this.recoverablePtyConsumerOwner(serverBuildId)
-    const options = {
+    const options: OpenSshPtyConsumerSessionOptions = {
       clientInstanceId: this.ptyConsumerClientInstanceId,
       expectedServerBuildId: serverBuildId,
       allowSameBuildLegacyFallback: true,
       outputFlowControl: { requestedWindowSu: DEFAULT_PTY_SOURCE_WINDOW_SU }
     }
-    const admission = await this.admitPtyConsumerOwner(mux, previousOwner, options, ownsAttempt)
+    let admission: SshPtyConsumerAdmission
+    try {
+      admission = await this.admitPtyConsumerOwner(mux, previousOwner, options, ownsAttempt)
+    } catch (error) {
+      if (
+        !previousOwner ||
+        (error as { code?: unknown }).code !== PTY_CONSUMER_STALE_OWNER_RECOVERY_ERROR
+      ) {
+        throw error
+      }
+      this.voidPtyConsumerCheckpoints(previousOwner, ownsAttempt)
+      if (!ownsAttempt()) {
+        throw new Error('Session disposed during owner recovery')
+      }
+      admission = await openSshPtyConsumerSession(mux, options)
+    }
     if (previousOwner && !admission.resumed) {
       this.voidPtyConsumerCheckpoints(previousOwner, ownsAttempt)
     }
@@ -1535,6 +1574,13 @@ export class SshRelaySession {
     unregisterSshGitProvider(this.targetId)
     this.sourceIdentityByRelayPtyId.clear()
     this.retiredSourceDeliveries.clear()
+    this.rejectedPtyRecoveryAttempts.clear()
+    for (const timer of this.rejectedPtyRecoveryRetries) {
+      clearTimeout(timer)
+    }
+    this.rejectedPtyRecoveryRetries.clear()
+    this.rejectedPtyReattaches.clear()
+    this.ptyFrameRejectionLog.clear()
     for (const pending of this.pendingPtyReattaches.values()) {
       for (const resolve of pending.recoveryWaiters) {
         resolve()
@@ -1637,6 +1683,24 @@ export class SshRelaySession {
       }
       void this.acceptPtyData(payload).catch(() => {})
     })
+    ptyProvider.onRejectedData?.((payload) => {
+      if (
+        this.mux !== mux ||
+        this.activePtyProviderGeneration !== providerGeneration ||
+        payload.providerGeneration !== providerGeneration
+      ) {
+        return
+      }
+      const pending = this.pendingPtyReattaches.get(payload.id)
+      if (pending) {
+        pending.restoreRequired = payload.sourceMalformed
+          ? 'recoverySourceMalformed'
+          : 'recoverySourceUnadmitted'
+        this.wakeRecovery(pending)
+        return
+      }
+      void this.acceptPtyData(payload).catch(() => {})
+    })
     ptyProvider.onReplay((payload) => {
       if (this.mux !== mux || this.activePtyProviderGeneration !== providerGeneration) {
         return
@@ -1677,22 +1741,19 @@ export class SshRelaySession {
     ) {
       return Promise.resolve()
     }
-    if (
-      consumerOwner?.outputFlowControl &&
-      (!offeredSource ||
-        payload.sourceMalformed ||
-        offeredSource.clientGeneration !== consumerOwner.clientGeneration ||
-        offeredSource.ownerGeneration !== consumerOwner.ownerGeneration)
-    ) {
-      closeSshPtyOutputGeneration(
-        payload.providerGeneration,
-        'ssh_source_frame_malformed_or_missing'
-      )
-      this.mux?.dispose('connection_lost')
-      return Promise.reject(new Error('ssh_source_frame_malformed_or_missing'))
+    const rejection = classifySshPtyFrameRejection(payload, consumerOwner)
+    if (rejection) {
+      if (offeredSource) {
+        this.retiredSourceDeliveries.retire(payload.providerGeneration, offeredSource)
+      }
+      this.ptyFrameRejectionLog.record(payload, consumerOwner, rejection)
+      if (rejection.action === 'retire-and-reattach-delivery') {
+        this.recoverRejectedPtyDelivery(payload, offeredSource)
+      }
+      return Promise.resolve()
     }
     const source = consumerOwner?.outputFlowControl ? offeredSource : undefined
-    if (source) {
+    if (source && consumerOwner) {
       const current = this.sourceIdentityByRelayPtyId.get(source.relayPtyId)
       if (
         source.sourceEndSu <= source.sourceStartSu ||
@@ -1703,12 +1764,14 @@ export class SshRelaySession {
             current.ptyIncarnation !== payload.ptyIncarnation ||
             (current.nextSourceSu !== undefined && current.nextSourceSu !== source.sourceStartSu)))
       ) {
-        closeSshPtyOutputGeneration(
-          payload.providerGeneration,
-          'ssh_source_frame_stale_or_non_contiguous'
-        )
-        this.mux?.dispose('connection_lost')
-        return Promise.reject(new Error('ssh_source_frame_stale_or_non_contiguous'))
+        const rejection = {
+          reason: 'source-range-invalid',
+          action: 'retire-and-reattach-delivery'
+        } as const
+        this.retiredSourceDeliveries.retire(payload.providerGeneration, source)
+        this.ptyFrameRejectionLog.record(payload, consumerOwner, rejection)
+        this.recoverRejectedPtyDelivery(payload, source)
+        return Promise.resolve()
       }
       this.sourceIdentityByRelayPtyId.set(source.relayPtyId, {
         deliveryToken: source.deliveryToken,
@@ -1731,6 +1794,163 @@ export class SshRelaySession {
     })
   }
 
+  private recoverRejectedPtyDelivery(
+    payload: SshPtyDataPayload,
+    source: SshPtyDataPayload['source']
+  ): void {
+    const mux = this.mux
+    const providerGeneration = this.activePtyProviderGeneration
+    let relayPtyId: string
+    try {
+      relayPtyId = source?.relayPtyId ?? toRelaySshPtyId(this.targetId, payload.id)
+    } catch {
+      return
+    }
+    const appPtyId = toAppSshPtyId(this.targetId, relayPtyId)
+    if (
+      payload.id !== appPtyId ||
+      !mux ||
+      mux.isDisposed() ||
+      providerGeneration !== payload.providerGeneration ||
+      this.pendingPtyReattaches.has(appPtyId) ||
+      this.rejectedPtyReattaches.has(appPtyId)
+    ) {
+      return
+    }
+    if (payload.rejectedSourceRecovery === 'reconnect-channel') {
+      console.warn(
+        `[ssh-relay-session] PTY ${relayPtyId} delivery identity could not be retired safely for ${this.targetId}; dropping the relay channel to reconnect`
+      )
+      mux.dispose('connection_lost')
+      return
+    }
+    const previous = this.rejectedPtyRecoveryAttempts.get(appPtyId)
+    const attempt =
+      previous?.providerGeneration === providerGeneration
+        ? previous
+        : { providerGeneration, attempts: 0, generationAttempts: 0, reported: false }
+    if (
+      attempt.attempts >= SSH_REJECTED_PTY_RECOVERY_MAX_ATTEMPTS ||
+      attempt.generationAttempts >= SSH_REJECTED_PTY_RECOVERY_MAX_GENERATION_ATTEMPTS
+    ) {
+      if (!attempt.reported) {
+        attempt.reported = true
+        console.warn(
+          `[ssh-relay-session] PTY ${relayPtyId} delivery recovery exhausted for ${this.targetId}; dropping the relay channel to reconnect`
+        )
+        // Why a channel drop and not a terminal relay error: a terminal error clears the reconnect
+        // backoff, rotates provider authority (aborting every in-flight fs and git request on the
+        // target) and parks the target in a manual-recovery state — over one PTY's delivery. Losing
+        // the channel is the recoverable escalation, and it is what this path did before targeted
+        // recovery existed.
+        mux.dispose('connection_lost')
+      }
+      return
+    }
+    attempt.attempts++
+    attempt.generationAttempts++
+    this.rejectedPtyRecoveryAttempts.set(appPtyId, attempt)
+    void this.rejectedPtyReattaches
+      .run(appPtyId, () =>
+        this.reattachRejectedPty(
+          relayPtyId,
+          mux,
+          providerGeneration,
+          payload.rejectedSourceRecovery === 'fresh-activation'
+            ? 'fresh-activation'
+            : 'confirm-existing'
+        )
+      )
+      .then(
+        (recovered) => {
+          if (recovered) {
+            // Why only a completed reattach clears this: an accepted frame proves nothing about the
+            // delivery that was rejected, and resetting on one lets a flapping PTY reattach forever.
+            attempt.attempts = 0
+            return
+          }
+          this.retryRejectedPtyDelivery(payload, source, appPtyId)
+        },
+        (error: unknown) => {
+          console.warn(`[ssh-relay-session] PTY ${relayPtyId} targeted delivery recovery failed`, {
+            providerGeneration,
+            error: error instanceof Error ? error.message : String(error)
+          })
+          this.retryRejectedPtyDelivery(payload, source, appPtyId)
+        }
+      )
+  }
+
+  // Why liveness is checked before retrying: reattachKnownPty resolves without claiming the lease
+  // when the PTY exited mid-attach, which is indistinguishable from a failed reattach at the call
+  // site. Retrying that race twice would drop the relay channel over an ordinary PTY exit.
+  private retryRejectedPtyDelivery(
+    payload: SshPtyDataPayload,
+    source: SshPtyDataPayload['source'],
+    appPtyId: string
+  ): void {
+    const ptyProvider = getSshPtyProvider(this.targetId) as SshPtyProvider | undefined
+    if (!ptyProvider || typeof ptyProvider.hasPty !== 'function' || !ptyProvider.hasPty(appPtyId)) {
+      this.rejectedPtyRecoveryAttempts.delete(appPtyId)
+      return
+    }
+    const timer = setTimeout(() => {
+      this.rejectedPtyRecoveryRetries.delete(timer)
+      this.recoverRejectedPtyDelivery(payload, source)
+    }, SSH_REJECTED_PTY_RECOVERY_RETRY_DELAY_MS)
+    timer.unref?.()
+    this.rejectedPtyRecoveryRetries.add(timer)
+  }
+
+  private async reattachRejectedPty(
+    relayPtyId: string,
+    mux: SshChannelMultiplexer,
+    providerGeneration: number,
+    targetedDeliveryRecovery: TargetedDeliveryRecovery
+  ): Promise<boolean> {
+    const shouldContinue = () =>
+      this.mux === mux &&
+      !mux.isDisposed() &&
+      this.activePtyProviderGeneration === providerGeneration
+    const ptyProvider = getSshPtyProvider(this.targetId) as SshPtyProvider | undefined
+    // Why re-checked here: this can have waited for a queue slot, and a superseded generation must
+    // not pay for a lease read or an attach round trip.
+    if (!ptyProvider || !shouldContinue()) {
+      return false
+    }
+    const activeLease = this.store
+      .getSshRemotePtyLeases(this.targetId)
+      .find(
+        (lease) =>
+          lease.ptyId === relayPtyId && lease.state !== 'terminated' && lease.state !== 'expired'
+      )
+    const activeLeaseByPtyId = activeLease
+      ? new Map<string, SshPtyLease>([[relayPtyId, activeLease]])
+      : new Map<string, SshPtyLease>()
+    const expectedIdentity = activeLease ? expectedIdentityForLease(activeLease) : undefined
+    const attachedLeaseIds = new Set<string>()
+    await this.reattachKnownPty({
+      ptyProvider,
+      ptyId: relayPtyId,
+      activeLeaseByPtyId,
+      expectedIdentityByPtyId: expectedIdentity
+        ? new Map([[relayPtyId, expectedIdentity]])
+        : new Map(),
+      attachedLeaseIds,
+      mux,
+      providerGeneration,
+      shouldContinue,
+      targetedDeliveryRecovery
+    })
+    if (attachedLeaseIds.size > 0 && shouldContinue()) {
+      await this.store.markSshRemotePtyLeasesAttachedAsync(
+        this.targetId,
+        Array.from(attachedLeaseIds)
+      )
+    }
+    return attachedLeaseIds.has(relayPtyId)
+  }
+
   private quarantineReattachData(pending: PendingPtyReattach, payload: SshPtyDataPayload): void {
     this.observePrivateRecoveryFrame(pending, payload)
     if (pending.restoreRequired) {
@@ -1943,6 +2163,7 @@ export class SshRelaySession {
     this.retiredSourceDeliveries.activate(relayPtyId)
     clearProviderPtyState(payload.id)
     deletePtyOwnership(payload.id)
+    this.rejectedPtyRecoveryAttempts.delete(payload.id)
     getSshPtyConsumerRecovery(this.targetId)?.checkpointsByAppPtyId.delete(payload.id)
     getSshPtyConsumerRecovery(this.targetId)?.checkpointsByAppPtyId.delete(
       toRelaySshPtyId(this.targetId, payload.id)
@@ -2051,6 +2272,7 @@ export class SshRelaySession {
     mux: SshChannelMultiplexer
     providerGeneration: number
     shouldContinue: () => boolean
+    targetedDeliveryRecovery?: TargetedDeliveryRecovery
   }): Promise<void> {
     const {
       ptyProvider,
@@ -2060,7 +2282,8 @@ export class SshRelaySession {
       attachedLeaseIds,
       mux,
       providerGeneration,
-      shouldContinue
+      shouldContinue,
+      targetedDeliveryRecovery
     } = args
     const appPtyId = toAppSshPtyId(this.targetId, ptyId)
     const pendingReattach: PendingPtyReattach = {
@@ -2079,7 +2302,10 @@ export class SshRelaySession {
     let sourceActivationLease: SshPtyAttachResult['sourceActivationLease']
     let recoveryActivationLease: SshPtyRecoveryActivationLease | undefined
     try {
-      const recoveryRequest = await this.sourceRecoveryRequest(appPtyId)
+      const recoveryRequest =
+        targetedDeliveryRecovery === 'fresh-activation'
+          ? undefined
+          : await this.sourceRecoveryRequest(appPtyId)
       const attachResult = await this.attachPtyWithRetry(
         ptyProvider,
         ptyId,
@@ -2105,7 +2331,27 @@ export class SshRelaySession {
         await this.acceptPtyExit(exitDuringAttach)
         return
       }
-      if (recoveryRequest) {
+      const existingDeliveryConfirmed =
+        targetedDeliveryRecovery === 'confirm-existing' &&
+        recoveryRequest?.status === 'checkpoint' &&
+        !attachResult.sourceRecovery &&
+        Boolean(
+          attachResult.sourceActivation &&
+          this.sameSourceDelivery(attachResult.sourceActivation, recoveryRequest)
+        )
+      if (targetedDeliveryRecovery) {
+        const owner = this.activePtyConsumerOwner()
+        const activation = attachResult.sourceActivation
+        if (
+          !owner?.outputFlowControl ||
+          !activation ||
+          activation.clientGeneration !== owner.clientGeneration ||
+          activation.ownerGeneration !== owner.ownerGeneration
+        ) {
+          return
+        }
+      }
+      if (recoveryRequest && !existingDeliveryConfirmed) {
         const recovered = await this.finishSourceRecovery(
           ptyId,
           appPtyId,
@@ -2177,6 +2423,18 @@ export class SshRelaySession {
       pendingReattach.activated = true
       recoveryActivationLease?.commit()
       recoveryActivationLease = undefined
+      if (targetedDeliveryRecovery) {
+        if (targetedDeliveryRecovery === 'fresh-activation') {
+          this.retiredSourceDeliveries.activate(ptyId)
+          this.sourceIdentityByRelayPtyId.delete(ptyId)
+          getSshPtyConsumerRecovery(this.targetId)?.checkpointsByAppPtyId.delete(appPtyId)
+          getSshPtyConsumerRecovery(this.targetId)?.checkpointsByAppPtyId.delete(ptyId)
+        }
+        while (pendingReattach.queuedData.length > 0) {
+          await this.acceptPtyData(pendingReattach.queuedData.shift()!)
+        }
+        pendingReattach.livePassthrough = true
+      }
       const exitAfterActivation = pendingReattach.exits.find(
         (exit) =>
           !exit.incarnationId ||
@@ -2187,7 +2445,7 @@ export class SshRelaySession {
         await this.acceptPtyExit(exitAfterActivation)
         return
       }
-      if (!recoveryRequest) {
+      if (!recoveryRequest && !targetedDeliveryRecovery) {
         this.forwardReattachReplay(appPtyId, attachResult.replay ?? '')
       }
       sourceActivationLease?.commit()
diff --git a/src/relay/dispatcher-client-writer.ts b/src/relay/dispatcher-client-writer.ts
index dd9a2e3b2..514ee47b8 100644
--- a/src/relay/dispatcher-client-writer.ts
+++ b/src/relay/dispatcher-client-writer.ts
@@ -5,6 +5,7 @@ import {
   type DispatcherWriterEntry,
   type DispatcherWriterLane
 } from './dispatcher-writer-admission'
+import { DispatcherWriterDrainArm } from './dispatcher-writer-drain-arm'
 import { DispatcherWriterLaneScheduler } from './dispatcher-writer-lane-scheduler'
 import {
   DispatcherWriterSink,
@@ -33,11 +34,11 @@ export class DispatcherClientWriter {
   private readonly settleOnDrain = new Set<DispatcherWriterEntry>()
   private readonly capacityListeners = new Set<() => void>()
   private readonly idleWaiters = new Set<() => void>()
+  private readonly drain = new DispatcherWriterDrainArm()
   private saturated = false
-  private drainArmed = false
-  private removeDrainListener: (() => void) | null = null
   private livenessBypassOutstanding = false
   private pumping = false
+  private retiredCapacity = false
   private closed = false
   private closeNotified = false
 
@@ -88,7 +89,8 @@ export class DispatcherClientWriter {
     encode: () => Buffer,
     estimatedBytes: number,
     onSettled: (result: SinkWriteSettlement) => void = () => {},
-    overflowIsNonFatal = false
+    overflowIsNonFatal = false,
+    isStillAdmitted?: () => boolean
   ): boolean {
     if (this.closed) {
       onSettled({ ok: false, error: new Error('Relay writer is closed') })
@@ -97,6 +99,7 @@ export class DispatcherClientWriter {
     const entry: DispatcherWriterEntry = {
       lane,
       encode,
+      isStillAdmitted,
       estimatedBytes,
       onSettled: onceDispatcherWriterSettlement(onSettled),
       settled: false,
@@ -126,8 +129,7 @@ export class DispatcherClientWriter {
     }
     this.closed = true
     this.saturated = false
-    this.drainArmed = false
-    this.clearDrainListener()
+    this.drain.disarm()
     for (const entry of this.admission.takeQueued()) {
       this.releaseEntry(entry, { ok: false, error })
     }
@@ -167,6 +169,12 @@ export class DispatcherClientWriter {
     } finally {
       this.pumping = false
     }
+    // Why deferred: retiring a deep queue drops every entry in one pass, and notifying per drop
+    // fans out to each listener O(queue) times for capacity that only changed once.
+    if (this.retiredCapacity && !this.closed) {
+      this.retiredCapacity = false
+      this.notifyCapacity()
+    }
   }
 
   private selectNext(): DispatcherWriterEntry | undefined {
@@ -197,6 +205,11 @@ export class DispatcherClientWriter {
   }
 
   private writeEntry(entry: DispatcherWriterEntry): void {
+    if (entry.isStillAdmitted && !entry.isStillAdmitted()) {
+      this.releaseEntry(entry, { ok: false, error: new Error('PTY publication retired') })
+      this.retiredCapacity = true
+      return
+    }
     this.laneScheduler.recordWrite(entry.lane)
     this.inFlight.add(entry)
     let callbackResult: SinkWriteSettlement | undefined
@@ -247,30 +260,21 @@ export class DispatcherClientWriter {
   }
 
   private armDrain(): void {
-    if (this.drainArmed) {
-      return
-    }
-    this.drainArmed = true
     try {
-      const registration = this.sink.registerDrain(() => this.handleDrain())
-      if (!registration.registered) {
-        this.drainArmed = false
-      } else if (this.drainArmed) {
-        this.removeDrainListener = registration.remove
-      } else {
-        registration.remove()
-      }
+      this.drain.arm(
+        (onDrain) => this.sink.registerDrain(onDrain),
+        () => this.handleDrain()
+      )
     } catch (error) {
       this.close(error instanceof Error ? error : new Error(String(error)))
     }
   }
 
   private handleDrain(): void {
-    if (this.closed || !this.drainArmed) {
+    if (this.closed || !this.drain.isArmed) {
       return
     }
-    this.drainArmed = false
-    this.clearDrainListener()
+    this.drain.disarm()
     this.saturated = false
     this.livenessBypassOutstanding = false
     for (const entry of Array.from(this.settleOnDrain)) {
@@ -299,12 +303,6 @@ export class DispatcherClientWriter {
     }
   }
 
-  private clearDrainListener(): void {
-    const remove = this.removeDrainListener
-    this.removeDrainListener = null
-    remove?.()
-  }
-
   private isIdle = (): boolean => this.inFlight.size === 0 && this.admission.queuedEntries === 0
 
   private notifyIdle(): void {
diff --git a/src/relay/dispatcher-writer-admission.ts b/src/relay/dispatcher-writer-admission.ts
index 27504697f..71865b3c1 100644
--- a/src/relay/dispatcher-writer-admission.ts
+++ b/src/relay/dispatcher-writer-admission.ts
@@ -12,6 +12,7 @@ export type DispatcherWriterSettlement = { ok: true } | { ok: false; error: Erro
 export type DispatcherWriterEntry = {
   lane: DispatcherWriterLane
   encode: () => Buffer
+  isStillAdmitted?: () => boolean
   estimatedBytes: number
   onSettled: (result: DispatcherWriterSettlement) => void
   settled: boolean
diff --git a/src/relay/dispatcher-writer-drain-arm.ts b/src/relay/dispatcher-writer-drain-arm.ts
new file mode 100644
index 000000000..d0076e295
--- /dev/null
+++ b/src/relay/dispatcher-writer-drain-arm.ts
@@ -0,0 +1,39 @@
+type DrainRegistration = { registered: boolean; remove: () => void }
+
+/**
+ * Holds the writer's single outstanding drain registration on its sink.
+ *
+ * Why it is re-checked after registering: a sink may invoke the drain callback synchronously from
+ * inside registerDrain, which disarms this arm before registration returns. Storing the remove
+ * handle then would cancel a later, still-wanted arm instead of the one that already fired.
+ */
+export class DispatcherWriterDrainArm {
+  private armed = false
+  private remove: (() => void) | null = null
+
+  get isArmed(): boolean {
+    return this.armed
+  }
+
+  arm(register: (onDrain: () => void) => DrainRegistration, onDrain: () => void): void {
+    if (this.armed) {
+      return
+    }
+    this.armed = true
+    const registration = register(onDrain)
+    if (!registration.registered) {
+      this.armed = false
+    } else if (this.armed) {
+      this.remove = registration.remove
+    } else {
+      registration.remove()
+    }
+  }
+
+  disarm(): void {
+    this.armed = false
+    const remove = this.remove
+    this.remove = null
+    remove?.()
+  }
+}
diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts
index c3fc87609..c95eef321 100644
--- a/src/relay/dispatcher.ts
+++ b/src/relay/dispatcher.ts
@@ -55,6 +55,11 @@ export type RelayClientSourceOptions = {
   resumeReads?: () => void
 }
 
+export type PtyDataPublicationAdmission = (
+  clientId: number,
+  params: Readonly<Record<string, unknown>>
+) => boolean
+
 export type MethodHandler = (
   params: Record<string, unknown>,
   context: RequestContext
@@ -108,6 +113,7 @@ export class RelayDispatcher {
   private disposeListeners = new Set<() => void>()
   private legacyCapacityListeners = new Set<() => void>()
   private clientCapacityListeners = new Map<number, Set<() => void>>()
+  private ptyDataPublicationAdmission: PtyDataPublicationAdmission | null = null
   private publicationTransactionDepth = 0
   private deferredLegacyCapacity = false
   private deferredForcedLegacyCapacity = false
@@ -205,6 +211,20 @@ export class RelayDispatcher {
     return () => this.disposeListeners.delete(listener)
   }
 
+  // Why single-slot rather than a listener set: admission is a veto, so two registrations would have
+  // to agree on precedence. One owner (the PTY consumer session) holds it for the dispatcher's life.
+  registerPtyDataPublicationAdmission(admission: PtyDataPublicationAdmission): () => void {
+    if (this.ptyDataPublicationAdmission) {
+      throw new Error('PTY data publication admission is already registered')
+    }
+    this.ptyDataPublicationAdmission = admission
+    return () => {
+      if (this.ptyDataPublicationAdmission === admission) {
+        this.ptyDataPublicationAdmission = null
+      }
+    }
+  }
+
   onLegacyPtyCapacity(listener: () => void): () => void {
     this.legacyCapacityListeners.add(listener)
     return () => this.legacyCapacityListeners.delete(listener)
@@ -281,7 +301,9 @@ export class RelayDispatcher {
     data: string,
     limit = data.length
   ): number {
-    const clients = this.activeClients()
+    const clients = this.activeClients().filter((client) =>
+      this.admitsPtyDataPublication(client.id, params)
+    )
     const max = Math.min(data.length, limit)
     if (clients.length === 0) {
       return max
@@ -331,7 +353,7 @@ export class RelayDispatcher {
       params
     }
     return this.tryPublishToClients(
-      this.activeClients(),
+      this.activeClients().filter((client) => this.admitsPtyDataPublication(client.id, params)),
       msg,
       options.interactive ? 'interactive' : 'ordinary'
     )
@@ -346,7 +368,9 @@ export class RelayDispatcher {
       return false
     }
     return this.tryPublishToClients(
-      this.activeClients().filter((client) => matchesClient(client.id)),
+      this.activeClients().filter(
+        (client) => matchesClient(client.id) && this.admitsPtyDataPublication(client.id, params)
+      ),
       { jsonrpc: '2.0', method: 'pty.data', params },
       options.interactive ? 'interactive' : 'ordinary'
     )
@@ -361,7 +385,9 @@ export class RelayDispatcher {
       return false
     }
     return this.projectToClients(
-      this.activeClients().filter((client) => matchesClient(client.id)),
+      this.activeClients().filter(
+        (client) => matchesClient(client.id) && this.admitsPtyDataPublication(client.id, params)
+      ),
       { jsonrpc: '2.0', method: 'pty.data', params },
       options.interactive ? 'interactive' : 'ordinary'
     )
@@ -381,6 +407,10 @@ export class RelayDispatcher {
       onSettled({ ok: false, error: new Error('Relay client is not connected') })
       return false
     }
+    if (!this.admitsPtyDataPublication(clientId, params)) {
+      onSettled({ ok: false, error: new Error('PTY publication is not admitted') })
+      return false
+    }
     return this.publishToClient(
       client,
       { jsonrpc: '2.0', method: 'pty.data', params },
@@ -547,6 +577,9 @@ export class RelayDispatcher {
         if (client.closed) {
           continue
         }
+        if (method === 'pty.data' && !this.admitsPtyDataPublication(client.id, params ?? {})) {
+          continue
+        }
         if (method === 'pty.replay') {
           // Why: replay is never re-sent, so it takes the control lane where overflow is fatal — the
           // writer closes the client and reconnect reloads history rather than stranding a short buffer.
@@ -584,6 +617,9 @@ export class RelayDispatcher {
       method,
       ...(params !== undefined ? { params } : {})
     }
+    if (method === 'pty.data' && !this.admitsPtyDataPublication(client.id, params ?? {})) {
+      return false
+    }
     const frameBytes = this.estimateFrameBytes(msg)
     if (this.publishToClient(client, msg, 'ordinary', undefined, frameBytes)) {
       return true
@@ -1055,12 +1091,17 @@ export class RelayDispatcher {
       const seq = client.nextOutgoingSeq++
       return encodeJsonRpcFrame(msg, seq, client.highestReceivedSeq)
     }
+    const isStillAdmitted =
+      'method' in msg && msg.method === 'pty.data'
+        ? () => this.admitsPtyDataPublication(client.id, msg.params ?? {})
+        : undefined
     return client.writer.enqueue(
       lane,
       encode,
       frameBytes,
       onSettled,
-      lane === 'control' && controlOverflow === 'reject'
+      lane === 'control' && controlOverflow === 'reject',
+      isStillAdmitted
     )
   }
 
@@ -1091,6 +1132,13 @@ export class RelayDispatcher {
     return Array.from(this.clients.values()).filter((client) => !client.closed)
   }
 
+  private admitsPtyDataPublication(
+    clientId: number,
+    params: Readonly<Record<string, unknown>>
+  ): boolean {
+    return this.ptyDataPublicationAdmission?.(clientId, params) ?? true
+  }
+
   private activeClientKeys(): string[] {
     return this.activeClients().map((client) => this.clientKey(client))
   }
diff --git a/src/relay/pty-handler-source-publication.test.ts b/src/relay/pty-handler-source-publication.test.ts
index 48a5f7490..04e45d2b3 100644
--- a/src/relay/pty-handler-source-publication.test.ts
+++ b/src/relay/pty-handler-source-publication.test.ts
@@ -221,12 +221,18 @@ describe('PtyHandler negotiated source publication', () => {
     expect(exitFrames()).toHaveLength(1)
   })
 
-  function attachSubscriber(): Buffer[] {
+  async function attachSubscriber(
+    holdDataSettlement?: (settle: (result: SinkWriteSettlement) => void) => boolean
+  ): Promise<Buffer[]> {
     const subscriberWrites: Buffer[] = []
-    dispatcher.attachClient(
+    const clientId = dispatcher.attachClient(
       (data, settle) => {
         subscriberWrites.push(Buffer.from(data))
-        if (notification(data)?.method === 'pty.exit') {
+        const frame = notification(data)
+        if (frame?.method === 'pty.data' && holdDataSettlement?.(settle)) {
+          return true
+        }
+        if (frame?.method === 'pty.exit') {
           // Why: real sockets never settle inside write(); see the primary sink above.
           queueMicrotask(() => settle({ ok: true }))
           return true
@@ -234,8 +240,18 @@ describe('PtyHandler negotiated source publication', () => {
         settle({ ok: true })
         return true
       },
-      { supportsWriteCallback: true }
+      { supportsWriteCallback: true },
+      endpointIdentity
     )
+    dispatcher.feedClient(
+      clientId,
+      requestFrame(20, 'pty.openClient', {
+        protocolVersion: 1,
+        clientInstanceId: 'legacy-subscriber',
+        requestedRole: 'subscriber'
+      })
+    )
+    await vi.advanceTimersByTimeAsync(0)
     return subscriberWrites
   }
 
@@ -248,7 +264,7 @@ describe('PtyHandler negotiated source publication', () => {
   it('never re-delivers the exit to subscribers when a cancel retires the record', async () => {
     await spawn({})
     const spawnResult = writes.map((buffer) => responseResult(buffer, 2)).find(Boolean)!
-    const subscriberWrites = attachSubscriber()
+    const subscriberWrites = await attachSubscriber()
     dataCallback!('prompt')
     await vi.advanceTimersByTimeAsync(8)
 
@@ -293,6 +309,13 @@ describe('PtyHandler negotiated source publication', () => {
     await spawn({})
     const spawnResult = writes.map((buffer) => responseResult(buffer, 2)).find(Boolean)!
     await cancelSourceDelivery(spawnResult)
+    const subscriberWrites = await attachSubscriber((settle) => {
+      if (!holdDataSettlements) {
+        return false
+      }
+      heldDataSettlements.push(settle)
+      return true
+    })
     holdDataSettlements = true
 
     dataCallback!('first')
@@ -311,7 +334,7 @@ describe('PtyHandler negotiated source publication', () => {
     heldDataSettlements[0]({ ok: true })
     await vi.advanceTimersByTimeAsync(8)
 
-    const frames = writes
+    const frames = subscriberWrites
       .map(notification)
       .filter(
         (frame): frame is Notification =>
@@ -351,7 +374,7 @@ describe('PtyHandler negotiated source publication', () => {
     await spawn({})
     const spawnResult = writes.map((buffer) => responseResult(buffer, 2)).find(Boolean)!
     const id = String(spawnResult.id)
-    const subscriberWrites = attachSubscriber()
+    const subscriberWrites = await attachSubscriber()
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     const publishOwnerExit = vi
       .spyOn(dispatcher, 'tryNotifyPtyExitToClient')
@@ -381,7 +404,7 @@ describe('PtyHandler negotiated source publication', () => {
     await spawn({})
     const spawnResult = writes.map((buffer) => responseResult(buffer, 2)).find(Boolean)!
     const id = String(spawnResult.id)
-    const subscriberWrites = attachSubscriber()
+    const subscriberWrites = await attachSubscriber()
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     const settleOwnerExit = vi.spyOn(adapter, 'settleExitPublication').mockImplementation(() => {
       throw new Error('exit settlement failed')
@@ -408,7 +431,7 @@ describe('PtyHandler negotiated source publication', () => {
   it('lets a retired record re-target its own exit instead of broadcasting a duplicate', async () => {
     await spawn({})
     const spawnResult = writes.map((buffer) => responseResult(buffer, 2)).find(Boolean)!
-    const subscriberWrites = attachSubscriber()
+    const subscriberWrites = await attachSubscriber()
     const publishExitAfterRetire = vi.fn(() => true)
     handler.setSourcePublication(stubPublication({ accepts: () => false, publishExitAfterRetire }))
 
@@ -680,20 +703,50 @@ describe('PtyHandler negotiated source publication', () => {
     await spawn({})
     const detached: number[] = []
     const healthyWrites: Buffer[] = []
+    let saturateSubscriber = false
     dispatcher.onClientDetached((clientId) => detached.push(clientId))
-    const saturatedId = dispatcher.attachClient(() => false, {
-      supportsWriteCallback: true,
-      writableLength: () => 16 * 1024,
-      writableHighWaterMark: () => 4 * 1024 * 1024
-    })
+    const saturatedId = dispatcher.attachClient(
+      (_data, settle) => {
+        if (saturateSubscriber) {
+          return false
+        }
+        settle({ ok: true })
+        return true
+      },
+      {
+        supportsWriteCallback: true,
+        writableLength: () => 16 * 1024,
+        writableHighWaterMark: () => 4 * 1024 * 1024
+      },
+      endpointIdentity
+    )
     const healthyId = dispatcher.attachClient(
       (data, settle) => {
         healthyWrites.push(Buffer.from(data))
         settle({ ok: true })
         return true
       },
-      { supportsWriteCallback: true }
+      { supportsWriteCallback: true },
+      endpointIdentity
     )
+    dispatcher.feedClient(
+      saturatedId,
+      requestFrame(20, 'pty.openClient', {
+        protocolVersion: 1,
+        clientInstanceId: 'saturated-subscriber',
+        requestedRole: 'subscriber'
+      })
+    )
+    dispatcher.feedClient(
+      healthyId,
+      requestFrame(21, 'pty.openClient', {
+        protocolVersion: 1,
+        clientInstanceId: 'healthy-subscriber',
+        requestedRole: 'subscriber'
+      })
+    )
+    await vi.advanceTimersByTimeAsync(0)
+    saturateSubscriber = true
     const payload = 's'.repeat(16 * 1024)
     let admitted = 0
     while (
diff --git a/src/relay/relay-pty-consumer-owner-displacement.test.ts b/src/relay/relay-pty-consumer-owner-displacement.test.ts
index 89059119f..155abc542 100644
--- a/src/relay/relay-pty-consumer-owner-displacement.test.ts
+++ b/src/relay/relay-pty-consumer-owner-displacement.test.ts
@@ -121,7 +121,7 @@ describe('relay PTY consumer owner displacement', () => {
     expect(adapter.deliveryMode(reconnectClientId)).toBe('source-owner')
     // Why: revocation is what bounds the takeover — the stale owner cannot open or drive deliveries again
     // even though its socket never closed.
-    expect(adapter.deliveryMode(1)).toBe('subscriber')
+    expect(adapter.deliveryMode(1)).toBe('unadmitted')
     expect(adapter.openDelivery(1, 'pty-2', 'incarnation-1')).toBeNull()
     expect(detached).toContain(1)
     expect(closeStaleTransport).toHaveBeenCalledOnce()
@@ -244,7 +244,7 @@ describe('relay PTY consumer owner displacement', () => {
 
     grantSettlement!({ ok: false, error: new Error('reconnect socket closed mid-response') })
     expect(adapter.deliveryMode(1)).toBe('source-owner')
-    expect(adapter.deliveryMode(reconnectClientId)).toBe('subscriber')
+    expect(adapter.deliveryMode(reconnectClientId)).toBe('unadmitted')
 
     // Why: the rolled-back attempt must leave the incumbent's lease reclaimable by the next reconnect.
     const retryWrites: Buffer[] = []
@@ -275,7 +275,7 @@ describe('relay PTY consumer owner displacement', () => {
       role: 'session-owner',
       ownerGeneration: 3
     })
-    expect(adapter.deliveryMode(1)).toBe('subscriber')
+    expect(adapter.deliveryMode(1)).toBe('unadmitted')
     expect(detached).toContain(1)
   })
 
@@ -349,7 +349,7 @@ describe('relay PTY consumer owner displacement', () => {
       code: PTY_CONSUMER_OWNER_RECOVERY_SUPERSEDED_ERROR
     })
     expect(adapter.deliveryMode(firstReconnectClientId)).toBe('source-owner')
-    expect(adapter.deliveryMode(retryClientId)).toBe('subscriber')
+    expect(adapter.deliveryMode(retryClientId)).toBe('unadmitted')
 
     dispatcher.detachClient(firstReconnectClientId)
     dispatcher.feedClient(
diff --git a/src/relay/relay-pty-publication-admission.test.ts b/src/relay/relay-pty-publication-admission.test.ts
new file mode 100644
index 000000000..d1cb9551c
--- /dev/null
+++ b/src/relay/relay-pty-publication-admission.test.ts
@@ -0,0 +1,219 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import {
+  RelayDispatcher,
+  type RelayClientSessionIdentity,
+  type RelayClientSinkOptions,
+  type SinkWriteSettlement
+} from './dispatcher'
+import { encodeJsonRpcFrame, MessageType } from './protocol'
+import { RelayPtySourcePublication } from './relay-pty-source-publication'
+import { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter'
+
+const endpointIdentity: RelayClientSessionIdentity = {
+  principal: 'endpoint-principal',
+  authenticated: true,
+  allowSessionOwner: true,
+  authenticationKind: 'endpoint-credential'
+}
+
+type RpcMessage = {
+  id?: number
+  method?: string
+  params?: Record<string, unknown>
+  result?: Record<string, unknown>
+}
+
+function message(buffer: Buffer): RpcMessage | null {
+  if (buffer[0] !== MessageType.Regular) {
+    return null
+  }
+  const length = buffer.readUInt32BE(9)
+  return JSON.parse(buffer.subarray(13, 13 + length).toString('utf8'))
+}
+
+function openFrame(
+  id: number,
+  requestedRole: 'session-owner' | 'subscriber',
+  outputFlowControl = false
+): Buffer {
+  return encodeJsonRpcFrame(
+    {
+      jsonrpc: '2.0',
+      id,
+      method: 'pty.openClient',
+      params: {
+        protocolVersion: 1,
+        clientInstanceId: `client-${id}`,
+        requestedRole,
+        ...(outputFlowControl
+          ? {
+              capabilities: {
+                outputFlowControl: { versions: [1], requestedWindowSu: 16 }
+              }
+            }
+          : {})
+      }
+    },
+    id,
+    0
+  )
+}
+
+async function flushRequests(): Promise<void> {
+  await new Promise((resolve) => setImmediate(resolve))
+}
+
+class SaturatedSink {
+  readonly writes: Buffer[] = []
+  private readonly drainWaiters = new Set<() => void>()
+  private writableBytes = 0
+  saturateNext = false
+
+  readonly options: RelayClientSinkOptions = {
+    supportsWriteCallback: true,
+    writableLength: () => this.writableBytes,
+    writableHighWaterMark: () => 8 * 1024 * 1024,
+    waitWriteDrain: (callback) => {
+      this.drainWaiters.add(callback)
+      return () => this.drainWaiters.delete(callback)
+    }
+  }
+
+  write = (data: Buffer, onSettled: (result: SinkWriteSettlement) => void): boolean => {
+    this.writes.push(Buffer.from(data))
+    this.writableBytes += data.length
+    onSettled({ ok: true })
+    if (!this.saturateNext) {
+      return true
+    }
+    this.saturateNext = false
+    return false
+  }
+
+  drain(): void {
+    this.writableBytes = 0
+    for (const callback of Array.from(this.drainWaiters)) {
+      callback()
+    }
+  }
+}
+
+describe('relay PTY publication admission', () => {
+  let dispatcher: RelayDispatcher | null = null
+
+  afterEach(() => {
+    dispatcher?.dispose()
+    dispatcher = null
+  })
+
+  it('retires queued legacy data when the source-credit grant response overtakes it', async () => {
+    const sink = new SaturatedSink()
+    dispatcher = new RelayDispatcher(sink.write, sink.options, endpointIdentity)
+    sink.saturateNext = true
+    dispatcher.notify('test.blocker')
+    const settled = vi.fn<(result: SinkWriteSettlement) => void>()
+    const legacyData = 'x'.repeat(1024 * 1024 + 128)
+
+    expect(dispatcher.tryNotifyPtyDataToClient(1, { id: 'pty-1', data: legacyData }, settled)).toBe(
+      true
+    )
+    expect(dispatcher.legacyRetentionBelowLowWater).toBe(false)
+    const adapter = new SshPtyConsumerSessionAdapter(dispatcher, 'build-a')
+    expect(adapter.deliveryMode(1)).toBe('unadmitted')
+    dispatcher.feed(openFrame(1, 'session-owner', true))
+    await flushRequests()
+
+    sink.drain()
+
+    expect(sink.writes.map(message).some((entry) => entry?.method === 'pty.data')).toBe(false)
+    expect(sink.writes.map(message).find((entry) => entry?.id === 1)?.result).toMatchObject({
+      role: 'session-owner'
+    })
+    expect(settled).toHaveBeenCalledOnce()
+    expect(settled.mock.calls[0][0]).toMatchObject({ ok: false })
+    expect(dispatcher.legacyRetentionBelowLowWater).toBe(true)
+  })
+
+  it('does not publish legacy data to a source-credit owner before PTY attachment', async () => {
+    const writes: Buffer[] = []
+    dispatcher = new RelayDispatcher(
+      (data, onSettled) => {
+        writes.push(Buffer.from(data))
+        onSettled({ ok: true })
+        return true
+      },
+      { supportsWriteCallback: true },
+      endpointIdentity
+    )
+    const adapter = new SshPtyConsumerSessionAdapter(dispatcher, 'build-a')
+    dispatcher.feed(openFrame(1, 'session-owner', true))
+    await flushRequests()
+
+    expect(adapter.deliveryMode(1)).toBe('source-owner')
+    expect(dispatcher.tryNotifyPtyData({ id: 'pty-1', data: 'legacy' })).toBe(true)
+
+    expect(writes.map(message).filter((entry) => entry?.method === 'pty.data')).toEqual([])
+  })
+
+  it('publishes current source output to the admitted source-credit owner', async () => {
+    const writes: Buffer[] = []
+    dispatcher = new RelayDispatcher(
+      (data, onSettled) => {
+        writes.push(Buffer.from(data))
+        onSettled({ ok: true })
+        return true
+      },
+      { supportsWriteCallback: true },
+      endpointIdentity
+    )
+    const adapter = new SshPtyConsumerSessionAdapter(dispatcher, 'build-a')
+    const publication = new RelayPtySourcePublication(dispatcher, adapter, () => {})
+    dispatcher.feed(openFrame(1, 'session-owner', true))
+    await flushRequests()
+    const activationSettlements: ((result: SinkWriteSettlement) => void)[] = []
+
+    expect(
+      publication.activate('pty-1', 'incarnation-1', {
+        clientId: 1,
+        isStale: () => false,
+        sessionIdentity: endpointIdentity,
+        onResponseSettled: (callback) => activationSettlements.push(callback)
+      })
+    ).toBe('opened')
+    activationSettlements[0]({ ok: true })
+    expect(publication.publish('pty-1', { data: 'current' }, false)).toBe(true)
+
+    const data = writes.map(message).find((entry) => entry?.method === 'pty.data')?.params
+    expect(data).toMatchObject({
+      id: 'pty-1',
+      data: 'current',
+      clientGeneration: 1,
+      ownerGeneration: 1,
+      ptyIncarnation: 'incarnation-1'
+    })
+    expect(data?.deliveryToken).toEqual(expect.any(String))
+  })
+
+  it('publishes legacy output to an admitted subscriber', async () => {
+    const writes: Buffer[] = []
+    dispatcher = new RelayDispatcher(
+      (data, onSettled) => {
+        writes.push(Buffer.from(data))
+        onSettled({ ok: true })
+        return true
+      },
+      { supportsWriteCallback: true },
+      endpointIdentity
+    )
+    const adapter = new SshPtyConsumerSessionAdapter(dispatcher, 'build-a')
+    dispatcher.feed(openFrame(1, 'subscriber'))
+    await flushRequests()
+
+    expect(adapter.deliveryMode(1)).toBe('subscriber')
+    expect(dispatcher.tryNotifyPtyData({ id: 'pty-1', data: 'legacy' })).toBe(true)
+
+    expect(writes.map(message).find((entry) => entry?.method === 'pty.data')?.params).toMatchObject(
+      { id: 'pty-1', data: 'legacy' }
+    )
+  })
+})
diff --git a/src/relay/relay-pty-source-output.ts b/src/relay/relay-pty-source-output.ts
index bbc57d1c6..0f6ea798a 100644
--- a/src/relay/relay-pty-source-output.ts
+++ b/src/relay/relay-pty-source-output.ts
@@ -58,7 +58,10 @@ export function projectPtySourceOutputToLegacy(
   interactive: boolean
 ): boolean {
   return dispatcher.projectPtyDataToMatchingClients(
-    (clientId) => session.deliveryMode(clientId) !== 'source-owner',
+    (clientId) => {
+      const mode = session.deliveryMode(clientId)
+      return mode === 'legacy-owner' || mode === 'subscriber'
+    },
     {
       id,
       data: output.data,
diff --git a/src/relay/relay-pty-source-publication.test.ts b/src/relay/relay-pty-source-publication.test.ts
index 1d30e4e85..af75613de 100644
--- a/src/relay/relay-pty-source-publication.test.ts
+++ b/src/relay/relay-pty-source-publication.test.ts
@@ -223,14 +223,24 @@ describe('RelayPtySourcePublication', () => {
   it('keeps mixed legacy and V1 clients on distinct frame authority', async () => {
     const harness = await createHarness(8)
     const legacyWrites: Buffer[] = []
-    dispatcher!.attachClient(
+    const legacyClientId = dispatcher!.attachClient(
       (data, onSettled) => {
         legacyWrites.push(Buffer.from(data))
         onSettled({ ok: true })
         return true
       },
-      { supportsWriteCallback: true }
+      { supportsWriteCallback: true },
+      endpointIdentity
     )
+    dispatcher!.feedClient(
+      legacyClientId,
+      requestFrame(2, 'pty.openClient', {
+        protocolVersion: 1,
+        clientInstanceId: 'legacy-client',
+        requestedRole: 'subscriber'
+      })
+    )
+    await flushRequests()
 
     harness.publication.publish('pty-1', { data: 'data' }, false)
 
@@ -289,10 +299,15 @@ describe('RelayPtySourcePublication', () => {
     const saturatedWrites: Buffer[] = []
     const healthyWrites: Buffer[] = []
     const heldSettlements: ((result: SinkWriteSettlement) => void)[] = []
+    let saturateSubscriber = false
     dispatcher!.onClientDetached((clientId) => detached.push(clientId))
     const saturatedId = dispatcher!.attachClient(
       (data, onSettled) => {
         saturatedWrites.push(Buffer.from(data))
+        if (!saturateSubscriber) {
+          onSettled({ ok: true })
+          return true
+        }
         heldSettlements.push(onSettled)
         return false
       },
@@ -300,7 +315,8 @@ describe('RelayPtySourcePublication', () => {
         supportsWriteCallback: true,
         writableLength: () => 128 * 1024,
         writableHighWaterMark: () => 4 * 1024 * 1024
-      }
+      },
+      endpointIdentity
     )
     const healthyId = dispatcher!.attachClient(
       (data, onSettled) => {
@@ -308,8 +324,27 @@ describe('RelayPtySourcePublication', () => {
         onSettled({ ok: true })
         return true
       },
-      { supportsWriteCallback: true }
+      { supportsWriteCallback: true },
+      endpointIdentity
     )
+    dispatcher!.feedClient(
+      saturatedId,
+      requestFrame(2, 'pty.openClient', {
+        protocolVersion: 1,
+        clientInstanceId: 'saturated-subscriber',
+        requestedRole: 'subscriber'
+      })
+    )
+    dispatcher!.feedClient(
+      healthyId,
+      requestFrame(3, 'pty.openClient', {
+        protocolVersion: 1,
+        clientInstanceId: 'healthy-subscriber',
+        requestedRole: 'subscriber'
+      })
+    )
+    await flushRequests()
+    saturateSubscriber = true
     const saturatedPayload = 's'.repeat(128 * 1024)
     let admitted = 0
     while (
@@ -328,7 +363,9 @@ describe('RelayPtySourcePublication', () => {
 
     expect(detached).toEqual([saturatedId])
     expect(detached).not.toContain(healthyId)
-    expect(saturatedWrites).toHaveLength(1)
+    expect(
+      saturatedWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
+    ).toHaveLength(1)
     expect(heldSettlements).toHaveLength(1)
     expect(
       healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
diff --git a/src/relay/relay-pty-source-publication.ts b/src/relay/relay-pty-source-publication.ts
index b21e8a79c..bad396b5b 100644
--- a/src/relay/relay-pty-source-publication.ts
+++ b/src/relay/relay-pty-source-publication.ts
@@ -71,7 +71,7 @@ export class RelayPtySourcePublication {
     }
     const mode = this.session.deliveryMode(context.clientId)
     let current = this.deliveries.get(id)
-    if (mode === 'subscriber') {
+    if (mode === 'unadmitted' || mode === 'subscriber') {
       this.sender.releaseRotationFence(current)
       return false
     }
diff --git a/src/relay/ssh-pty-consumer-session-adapter.ts b/src/relay/ssh-pty-consumer-session-adapter.ts
index 1b3bf7bad..83e61f17f 100644
--- a/src/relay/ssh-pty-consumer-session-adapter.ts
+++ b/src/relay/ssh-pty-consumer-session-adapter.ts
@@ -2,8 +2,7 @@ import {
   PTY_CONSUMER_SESSION_PROTOCOL_VERSION,
   PtyConsumerSession,
   type PtyConsumerSessionAdmission,
-  type PtyConsumerSessionGrant,
-  type PtyConsumerSessionHello
+  type PtyConsumerSessionGrant
 } from '../shared/pty-consumer-session'
 import { DEFAULT_PTY_SOURCE_WINDOW_SU } from '../shared/pty-source-credit-contract'
 import type {
@@ -12,63 +11,18 @@ import type {
   PtySourceSpan,
   PtySourceTransform
 } from '../shared/pty-source-credit-contract'
-import type { RelayClientSessionIdentity, RelayDispatcher, RequestContext } from './dispatcher'
+import type { RelayDispatcher, RequestContext } from './dispatcher'
 import type { PtySourceSendReservation } from './pty-source-credit-ledger'
 import { SshPtySourceCreditAdapter } from './ssh-pty-source-credit-adapter'
+import {
+  admitsSshPtyDataPublication,
+  sshPtyDeliveryMode,
+  type SshPtyDeliveryMode
+} from './ssh-pty-data-publication-admission'
+import { parseOpenClientParams, requireIdentity } from './ssh-pty-open-client-request'
 
 export const SSH_PTY_OPEN_CLIENT_METHOD = 'pty.openClient'
 
-type OpenClientParams = PtyConsumerSessionHello & {
-  protocolVersion: number
-}
-
-function parseOpenClientParams(params: Record<string, unknown>): OpenClientParams {
-  const resume =
-    typeof params.resume === 'object' && params.resume !== null
-      ? (params.resume as Record<string, unknown>)
-      : undefined
-  const capabilities =
-    typeof params.capabilities === 'object' && params.capabilities !== null
-      ? (params.capabilities as Record<string, unknown>)
-      : undefined
-  const outputFlowControl =
-    typeof capabilities?.outputFlowControl === 'object' && capabilities.outputFlowControl !== null
-      ? (capabilities.outputFlowControl as Record<string, unknown>)
-      : undefined
-  return {
-    protocolVersion: Number(params.protocolVersion),
-    clientInstanceId: String(params.clientInstanceId ?? ''),
-    requestedRole: String(params.requestedRole ?? '') as PtyConsumerSessionHello['requestedRole'],
-    ...(resume
-      ? {
-          resume: {
-            ownerGeneration: Number(resume.ownerGeneration),
-            ownerLease: String(resume.ownerLease ?? '')
-          }
-        }
-      : {}),
-    ...(outputFlowControl
-      ? {
-          capabilities: {
-            outputFlowControl: {
-              versions: Array.isArray(outputFlowControl.versions)
-                ? outputFlowControl.versions.map(Number)
-                : [],
-              requestedWindowSu: Number(outputFlowControl.requestedWindowSu)
-            }
-          }
-        }
-      : {})
-  }
-}
-
-function requireIdentity(context: RequestContext): RelayClientSessionIdentity {
-  if (!context.sessionIdentity) {
-    throw new Error('SSH PTY consumer transport identity is unavailable')
-  }
-  return context.sessionIdentity
-}
-
 export class SshPtyConsumerSessionAdapter {
   private readonly session: PtyConsumerSession
   private readonly sourceCredit: SshPtySourceCreditAdapter
@@ -92,15 +46,26 @@ export class SshPtyConsumerSessionAdapter {
       serverBuildId,
       outputFlowControl: { versions: [1], maxWindowSu: DEFAULT_PTY_SOURCE_WINDOW_SU }
     })
+    // Why the admission is consulted again at drain time (see isStillAdmitted): a frame can sit
+    // queued behind a saturated socket long enough for the grant or the delivery to be retired, and
+    // publishing it then hands the client output from an owner it no longer is.
+    dispatcher.registerPtyDataPublicationAdmission((clientId, params) =>
+      admitsSshPtyDataPublication(
+        this.session.activeGrant(String(clientId)),
+        params,
+        this.sourceCredit
+      )
+    )
     dispatcher.onRequest(SSH_PTY_OPEN_CLIENT_METHOD, (params, context) =>
       this.openClient(params, context)
     )
     dispatcher.onClientDetached((clientId, cause) => {
-      const grant = this.session.activeGrant(String(clientId))
+      const connectionKey = String(clientId)
+      const grant = this.session.activeGrant(connectionKey)
       if (grant) {
         this.clearPausedForGrant(grant)
       }
-      this.session.close(String(clientId), cause)
+      this.session.close(connectionKey, cause)
       if (grant) {
         this.sourceCredit.retainOrCloseOnDetach(grant)
       }
@@ -239,12 +204,8 @@ export class SshPtyConsumerSessionAdapter {
     return this.sourceCredit.retentionSnapshot()
   }
 
-  deliveryMode(clientId: number): 'source-owner' | 'legacy-owner' | 'subscriber' {
-    const grant = this.session.activeGrant(String(clientId))
-    if (grant?.role !== 'session-owner') {
-      return 'subscriber'
-    }
-    return grant.capabilities?.outputFlowControl ? 'source-owner' : 'legacy-owner'
+  deliveryMode(clientId: number): SshPtyDeliveryMode {
+    return sshPtyDeliveryMode(this.session.activeGrant(String(clientId)))
   }
 
   private async openClient(
diff --git a/src/relay/ssh-pty-data-publication-admission.ts b/src/relay/ssh-pty-data-publication-admission.ts
new file mode 100644
index 000000000..95c86ee53
--- /dev/null
+++ b/src/relay/ssh-pty-data-publication-admission.ts
@@ -0,0 +1,38 @@
+import type { PtyConsumerSessionGrant } from '../shared/pty-consumer-session'
+import type { SshPtySourceCreditAdapter } from './ssh-pty-source-credit-adapter'
+
+export type SshPtyDeliveryMode = 'unadmitted' | 'source-owner' | 'legacy-owner' | 'subscriber'
+
+export function sshPtyDeliveryMode(
+  grant: Readonly<PtyConsumerSessionGrant> | null
+): SshPtyDeliveryMode {
+  if (!grant) {
+    return 'unadmitted'
+  }
+  if (grant.role !== 'session-owner') {
+    return 'subscriber'
+  }
+  return grant.capabilities?.outputFlowControl ? 'source-owner' : 'legacy-owner'
+}
+
+export function admitsSshPtyDataPublication(
+  grant: Readonly<PtyConsumerSessionGrant> | null,
+  params: Readonly<Record<string, unknown>>,
+  sourceCredit: Pick<SshPtySourceCreditAdapter, 'ownsDelivery'>
+): boolean {
+  if (!grant) {
+    return false
+  }
+  if (grant.role !== 'session-owner' || !grant.capabilities?.outputFlowControl) {
+    return true
+  }
+  const id = typeof params.id === 'string' ? params.id : ''
+  const token = typeof params.deliveryToken === 'string' ? params.deliveryToken : ''
+  const identity = token ? sourceCredit.ownsDelivery(token, grant, id) : null
+  return (
+    identity !== null &&
+    params.clientGeneration === identity.clientGeneration &&
+    params.ownerGeneration === identity.ownerGeneration &&
+    params.ptyIncarnation === identity.ptyIncarnation
+  )
+}
diff --git a/src/relay/ssh-pty-open-client-request.ts b/src/relay/ssh-pty-open-client-request.ts
new file mode 100644
index 000000000..48d6d2551
--- /dev/null
+++ b/src/relay/ssh-pty-open-client-request.ts
@@ -0,0 +1,56 @@
+import type { PtyConsumerSessionHello } from '../shared/pty-consumer-session'
+import type { RelayClientSessionIdentity, RequestContext } from './dispatcher'
+
+export type OpenClientParams = PtyConsumerSessionHello & {
+  protocolVersion: number
+}
+
+// Why every field is coerced rather than trusted: these arrive off the wire from a peer that may be
+// a different build, so a missing or wrong-typed member has to degrade into a value the session can
+// reject by contract instead of throwing somewhere further in.
+export function parseOpenClientParams(params: Record<string, unknown>): OpenClientParams {
+  const resume =
+    typeof params.resume === 'object' && params.resume !== null
+      ? (params.resume as Record<string, unknown>)
+      : undefined
+  const capabilities =
+    typeof params.capabilities === 'object' && params.capabilities !== null
+      ? (params.capabilities as Record<string, unknown>)
+      : undefined
+  const outputFlowControl =
+    typeof capabilities?.outputFlowControl === 'object' && capabilities.outputFlowControl !== null
+      ? (capabilities.outputFlowControl as Record<string, unknown>)
+      : undefined
+  return {
+    protocolVersion: Number(params.protocolVersion),
+    clientInstanceId: String(params.clientInstanceId ?? ''),
+    requestedRole: String(params.requestedRole ?? '') as PtyConsumerSessionHello['requestedRole'],
+    ...(resume
+      ? {
+          resume: {
+            ownerGeneration: Number(resume.ownerGeneration),
+            ownerLease: String(resume.ownerLease ?? '')
+          }
+        }
+      : {}),
+    ...(outputFlowControl
+      ? {
+          capabilities: {
+            outputFlowControl: {
+              versions: Array.isArray(outputFlowControl.versions)
+                ? outputFlowControl.versions.map(Number)
+                : [],
+              requestedWindowSu: Number(outputFlowControl.requestedWindowSu)
+            }
+          }
+        }
+      : {})
+  }
+}
+
+export function requireIdentity(context: RequestContext): RelayClientSessionIdentity {
+  if (!context.sessionIdentity) {
+    throw new Error('SSH PTY consumer transport identity is unavailable')
+  }
+  return context.sessionIdentity
+}
diff --git a/src/shared/pty-consumer-owner-admission.ts b/src/shared/pty-consumer-owner-admission.ts
new file mode 100644
index 000000000..580567508
--- /dev/null
+++ b/src/shared/pty-consumer-owner-admission.ts
@@ -0,0 +1,84 @@
+import {
+  PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR,
+  PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS,
+  PTY_CONSUMER_OWNER_HELD_SELF_ERROR,
+  type PtyConsumerCloseCause,
+  PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR
+} from './pty-consumer-session-contract'
+
+type HeldOwner = {
+  state: 'pending' | 'active' | 'disconnected'
+  disconnectedAt?: number
+  disconnectCause?: PtyConsumerCloseCause
+}
+
+export type RefuseHeldPtyConsumerOwnerOptions = {
+  ownerGraceMs: number
+  now: number
+  sameClient: boolean
+  // Why a callback instead of mutating the record: the owner record is reachable from `replaces`
+  // chains and from displaced-owner snapshots already handed to callers. Clamping in place would
+  // rewrite those retroactively, so the session that owns the record applies it copy-on-write.
+  clampGraceTo: (disconnectedAt: number) => void
+}
+
+function refuse(message: string, code: number): never {
+  throw Object.assign(new Error(message), { code })
+}
+
+/**
+ * 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.
+ */
+export function refuseHeldPtyConsumerOwner(
+  owner: Readonly<HeldOwner>,
+  options: RefuseHeldPtyConsumerOwnerOptions
+): never {
+  if (owner.state === 'pending') {
+    refuse('Owner grant publication is still pending', PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR)
+  }
+  if (owner.state === 'active') {
+    // Why identity without the lease: a client that lost its proof — a fresh process, a dropped
+    // recovery record — still knows who it is. Against an incumbent carrying its own instance id
+    // the honest answer is "your other connection is still registered", which resolves itself once
+    // the relay notices that socket. Blocking here strands the single-app case forever.
+    if (options.sameClient) {
+      refuse(
+        "PTY session owner is held by this client's own earlier connection",
+        PTY_CONSUMER_OWNER_HELD_SELF_ERROR
+      )
+    }
+    refuse(
+      'PTY session owner is held by an attached connection',
+      PTY_CONSUMER_OWNER_HELD_ATTACHED_ERROR
+    )
+  }
+  clampDisconnectedOwnerGrace(owner, options)
+  refuse(
+    'PTY session owner is held by a disconnected connection within its grace period',
+    PTY_CONSUMER_OWNER_HELD_DISCONNECTED_ERROR
+  )
+}
+
+// Why only a peer-closed disconnect may shorten this: the floor is a bet that the incumbent is gone,
+// and the relay tears a client's socket down for its own reasons too — a full lane queue is the
+// signature of an owner that is alive but not draining fast enough. No owner completes a reconnect
+// ladder in 250 ms, so clamping on a teardown we initiated hands a live owner's admission away and
+// it can never get it back. Expiring a record never stops the remote PTY, but it does cost the user
+// every route back to it.
+function clampDisconnectedOwnerGrace(
+  owner: Readonly<HeldOwner>,
+  options: RefuseHeldPtyConsumerOwnerOptions
+): void {
+  if (owner.disconnectCause !== 'peer-closed') {
+    return
+  }
+  const floorStart =
+    options.now - Math.max(options.ownerGraceMs - PTY_CONSUMER_OWNER_HELD_GRACE_FLOOR_MS, 0)
+  if ((owner.disconnectedAt ?? 0) <= floorStart) {
+    return
+  }
+  options.clampGraceTo(floorStart)
+}
diff --git a/src/shared/pty-consumer-session.ts b/src/shared/pty-consumer-session.ts
index 31e6d4dce..6b040be47 100644
--- a/src/shared/pty-consumer-session.ts
+++ b/src/shared/pty-consumer-session.ts
@@ -1,11 +1,6 @@
 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_HELD_SELF_ERROR,
-  PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR,
   PTY_CONSUMER_SESSION_PROTOCOL_VERSION,
   type PtyConsumerAuthentication,
   type PtyConsumerCloseCause,
@@ -25,6 +20,7 @@ import {
   isPtyConsumerOwnerSameClient,
   matchesPtyConsumerOwnerClaim
 } from './pty-consumer-owner-recovery'
+import { refuseHeldPtyConsumerOwner } from './pty-consumer-owner-admission'
 
 export * from './pty-consumer-session-contract'
 
@@ -49,10 +45,6 @@ type OwnerRecord = {
   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
@@ -224,7 +216,14 @@ export class PtyConsumerSession {
       return this.newOwner(hello, authentication, null)
     }
     if (!matchesPtyConsumerOwnerClaim(hello, authentication, current)) {
-      this.refuseHeldOwner(hello, authentication, current)
+      refuseHeldPtyConsumerOwner(current, {
+        ownerGraceMs: this.ownerGraceMs,
+        now: this.now(),
+        sameClient: isPtyConsumerOwnerSameClient(hello, authentication, current),
+        clampGraceTo: (disconnectedAt) => {
+          this.owner = { ...current, disconnectedAt }
+        }
+      })
     }
     assertPtyConsumerOwnerRecovery(hello, authentication, current)
     // Why an active owner is displaced rather than refused: the resume proof matched this owner's
@@ -234,61 +233,6 @@ 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(
-    hello: PtyConsumerSessionHello,
-    authentication: PtyConsumerAuthentication,
-    current: OwnerRecord
-  ): never {
-    if (current.state === 'pending') {
-      throwOwnerError(
-        'Owner grant publication is still pending',
-        PTY_CONSUMER_OWNER_RECOVERY_PENDING_ERROR
-      )
-    }
-    if (current.state === 'active') {
-      // Why identity without the lease: a client that lost its proof — a fresh process, a dropped
-      // recovery record — still knows who it is. Against an incumbent carrying its own instance id
-      // the honest answer is "your other connection is still registered", which resolves itself once
-      // the relay notices that socket. Blocking here strands the single-app case forever.
-      if (isPtyConsumerOwnerSameClient(hello, authentication, current)) {
-        throwOwnerError(
-          "PTY session owner is held by this client's own earlier connection",
-          PTY_CONSUMER_OWNER_HELD_SELF_ERROR
-        )
-      }
-      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 only a peer-closed disconnect may shorten this: the floor is a bet that the incumbent is gone,
-  // and the relay tears a client's socket down for its own reasons too — a full lane queue is the
-  // signature of an owner that is alive but not draining fast enough. No owner completes a reconnect
-  // ladder in 250 ms, so clamping on a teardown we initiated hands a live owner's admission away and
-  // it can never get it back. Expiring a record never stops the remote PTY, but it does cost the user
-  // every route back to it.
-  private clampDisconnectedOwnerGrace(current: OwnerRecord): void {
-    if (current.disconnectCause !== 'peer-closed') {
-      return
-    }
-    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 {