fix(relay): stop a bystander mirror from gating every PTY, drop the stall timer

Co-authored-by: Orca <help@stably.ai>

src/relay/dispatcher-client-writer.ts

@@ -33,7 +33,6 @@ export class DispatcherClientWriter {
   private readonly settleOnDrain = new Set<DispatcherWriterEntry>()
   private readonly capacityListeners = new Set<() => void>()
   private readonly idleWaiters = new Set<() => void>()
-  private settledBytes = 0
   private saturated = false
   private drainArmed = false
   private removeDrainListener: (() => void) | null = null
@@ -56,14 +55,6 @@ export class DispatcherClientWriter {
     return this.admission.retainedProducerBytes
   }
 
-  /**
-   * Monotonic bytes the sink has settled. Only real writes move it — a liveness entry replaced in the
-   * queue never reaches releaseEntry — so a flat counter across a window means the peer moved nothing.
-   */
-  get drainProgressBytes(): number {
-    return this.settledBytes
-  }
-
   get producerFrameCapacity(): number {
     return this.sink.producerFrameCapacity
   }
@@ -295,7 +286,6 @@ export class DispatcherClientWriter {
       return
     }
     entry.settled = true
-    this.settledBytes += entry.estimatedBytes
     this.inFlight.delete(entry)
     this.settleOnDrain.delete(entry)
     this.admission.release(entry)

src/relay/dispatcher-projection-stall.ts

@@ -1,86 +0,0 @@
-/**
- * Bounds the cost of back-pressuring a refused PTY projection.
- *
- * A projection is all-or-nothing, so one subscriber that never drains holds every PTY paused for every
- * viewer. Nothing else reaps such a sink: the writer only closes on a write *error*, and a peer that is
- * alive but not reading never raises one. This detaches a subscriber that made zero drain progress for a
- * full window — it reconnects and replays, while the sinks that are merely slow keep their back-pressure.
- */
-export const PROJECTION_STALL_EVICTION_MS = 30_000
-
-export type ProjectionStallProbe = {
-  // Retaining producer bytes: the sink has accepted nothing back since they were queued.
-  stalled: boolean
-  // Monotonic settled bytes; unchanged across a window means the sink moved nothing at all.
-  drainProgressBytes: number
-}
-
-export class ProjectionStallEvictor<T> {
-  private readonly armed = new Map<T, number>()
-  private timer: ReturnType<typeof setTimeout> | null = null
-
-  constructor(
-    private readonly probe: (target: T) => ProjectionStallProbe,
-    private readonly evict: (target: T) => void,
-    private readonly evictionMs: number = PROJECTION_STALL_EVICTION_MS
-  ) {}
-
-  noteRefusedProjection(targets: readonly T[]): void {
-    for (const target of targets) {
-      const probe = this.probe(target)
-      if (!probe.stalled) {
-        this.armed.delete(target)
-        continue
-      }
-      if (!this.armed.has(target)) {
-        this.armed.set(target, probe.drainProgressBytes)
-      }
-    }
-    this.schedule()
-  }
-
-  forget(target: T): void {
-    this.armed.delete(target)
-  }
-
-  dispose(): void {
-    this.armed.clear()
-    this.clearTimer()
-  }
-
-  private schedule(): void {
-    if (this.timer || this.armed.size === 0) {
-      return
-    }
-    this.timer = setTimeout(() => {
-      this.timer = null
-      this.sweep()
-    }, this.evictionMs)
-    this.timer.unref?.()
-  }
-
-  private clearTimer(): void {
-    if (!this.timer) {
-      return
-    }
-    clearTimeout(this.timer)
-    this.timer = null
-  }
-
-  private sweep(): void {
-    for (const [target, progressBytes] of Array.from(this.armed)) {
-      const probe = this.probe(target)
-      if (!probe.stalled) {
-        this.armed.delete(target)
-        continue
-      }
-      if (probe.drainProgressBytes !== progressBytes) {
-        this.armed.set(target, probe.drainProgressBytes)
-        continue
-      }
-      this.armed.delete(target)
-      this.evict(target)
-    }
-    this.schedule()
-  }
-}

src/relay/dispatcher-slow-client.test.ts

@@ -1,6 +1,10 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { RelayDispatcher, type RelayClientSinkOptions } from './dispatcher'
-import { PROJECTION_STALL_EVICTION_MS } from './dispatcher-projection-stall'
+import {
+  RelayDispatcher,
+  type LegacyProjectionRole,
+  type RelayClientSinkOptions,
+  type SinkWriteSettlement
+} from './dispatcher'
 import { DISPATCHER_CONTROL_QUEUE_MAX_FRAMES } from './dispatcher-writer-admission'
 
 type SlowClient = {
@@ -72,6 +76,18 @@ function fillProducerQueue(dispatcher: RelayDispatcher, clientId: number): void
   }
 }
 
+function roles(
+  byId: Record<number, LegacyProjectionRole>
+): (clientId: number) => LegacyProjectionRole {
+  return (clientId) => byId[clientId] ?? 'skip'
+}
+
+function queueFullDrops(stderr: { mock: { calls: unknown[][] } }): string[] {
+  return stderr.mock.calls
+    .map((call) => String(call[0]))
+    .filter((line) => /producer queue full/.test(line))
+}
+
 describe('RelayDispatcher slow-client PTY projection', () => {
   it('tryNotifyPtyData applies backpressure without closing the client', () => {
     const primary = makeStallingClient(65536)
@@ -94,14 +110,15 @@ describe('RelayDispatcher slow-client PTY projection', () => {
     }
   })
 
-  it('rejects a projected pty.data frame instead of closing a subscriber that cannot drain', () => {
+  it('rejects a projected pty.data frame instead of closing an owner that cannot drain', () => {
     const primary = makeStallingClient(65536)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     try {
-      fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0])
+      const primaryId = dispatcher.activeClientIds()[0]
+      fillProducerQueue(dispatcher, primaryId)
 
-      const projected = dispatcher.projectPtyDataToMatchingClients(() => true, {
+      const projected = dispatcher.projectPtyDataToLegacyClients(roles({ [primaryId]: 'owner' }), {
         id: 'pty-1',
         data: 'y'.repeat(40_000)
       })
@@ -115,13 +132,48 @@ describe('RelayDispatcher slow-client PTY projection', () => {
     }
   })
 
-  it('sheds a projected frame no sink can ever admit rather than wedging the PTY', () => {
+  it('sheds a saturated mirror instead of pausing the PTY for every other viewer', () => {
+    const owner = makeStallingClient(65536)
+    owner.drain()
+    const mirror = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(owner.write, owner.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      const ownerId = dispatcher.activeClientIds()[0]
+      const mirrorId = dispatcher.attachClient(mirror.write, mirror.options)
+      fillProducerQueue(dispatcher, mirrorId)
+      const mirrorBefore = ptyDataFrameCount(mirror.frames)
+      stderr.mockClear()
+
+      const projected = dispatcher.projectPtyDataToLegacyClients(
+        roles({ [ownerId]: 'owner', [mirrorId]: 'mirror' }),
+        { id: 'pty-1', data: 'y'.repeat(40_000) }
+      )
+
+      // A bystander mirror (agent hook, orca CLI, second window) must not gate the native PTY.
+      expect(projected).toBe(true)
+      expect(ptyDataFrameCount(owner.frames)).toBe(1)
+      expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore)
+      expect(mirror.closes).toBe(0)
+      expect(owner.closes).toBe(0)
+      expect(queueFullDrops(stderr)).toHaveLength(1)
+      expect(queueFullDrops(stderr)[0]).toMatch(
+        /^\[relay\] Dropped pty\.data \(\d+B, producer queue full/
+      )
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('sheds a projected frame no owner sink can ever admit rather than wedging the PTY', () => {
     const primary = makeStallingClient(65536)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     try {
+      const primaryId = dispatcher.activeClientIds()[0]
       // Larger than producerFrameCapacity (49152B), so retrying it forever would stall the PTY.
-      const projected = dispatcher.projectPtyDataToMatchingClients(() => true, {
+      const projected = dispatcher.projectPtyDataToLegacyClients(roles({ [primaryId]: 'owner' }), {
         id: 'pty-1',
         data: 'y'.repeat(60_000)
       })
@@ -141,53 +193,57 @@ describe('RelayDispatcher slow-client PTY projection', () => {
     }
   })
 
-  it('delivers a rejected projection exactly once to every subscriber after the stalled sink drains', () => {
-    const stalled = makeStallingClient(65536, true)
-    const healthy = makeStallingClient(65536, true)
-    healthy.drain()
-    const dispatcher = new RelayDispatcher(stalled.write, stalled.options)
+  it('delivers a rejected projection exactly once to every mirror after the owner drains', () => {
+    const owner = makeStallingClient(65536, true)
+    const mirror = makeStallingClient(65536, true)
+    mirror.drain()
+    const dispatcher = new RelayDispatcher(owner.write, owner.options)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     try {
-      dispatcher.attachClient(healthy.write, healthy.options)
-      fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0])
-      const healthyBefore = ptyDataFrameCount(healthy.frames)
+      const ownerId = dispatcher.activeClientIds()[0]
+      const mirrorId = dispatcher.attachClient(mirror.write, mirror.options)
+      fillProducerQueue(dispatcher, ownerId)
+      const mirrorBefore = ptyDataFrameCount(mirror.frames)
+      const roleOf = roles({ [ownerId]: 'owner', [mirrorId]: 'mirror' })
 
       const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
-      // All-or-nothing: the healthy subscriber must not get a copy the retry would duplicate.
-      expect(ptyDataFrameCount(healthy.frames)).toBe(healthyBefore)
+      expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false)
+      // The mirror must not get a copy the owner's retry would duplicate.
+      expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore)
 
-      stalled.drain()
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
-      expect(ptyDataFrameCount(healthy.frames)).toBe(healthyBefore + 1)
-      expect(stalled.closes).toBe(0)
-      expect(healthy.closes).toBe(0)
+      owner.drain()
+      expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(true)
+      expect(ptyDataFrameCount(mirror.frames)).toBe(mirrorBefore + 1)
+      expect(owner.closes).toBe(0)
+      expect(mirror.closes).toBe(0)
     } finally {
       stderr.mockRestore()
       dispatcher.dispose()
     }
   })
 
-  it('holds a projected pty.exit for retry instead of closing a saturated subscriber', () => {
+  it('holds a projected pty.exit for retry instead of closing a saturated owner', () => {
     const primary = makeStallingClient(65536, true)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     try {
-      fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0])
+      const primaryId = dispatcher.activeClientIds()[0]
+      fillProducerQueue(dispatcher, primaryId)
+      const roleOf = roles({ [primaryId]: 'owner' })
 
       const params = { id: 'pty-1', exitCode: 0 }
-      expect(dispatcher.projectPtyExitToMatchingClients(() => true, params)).toBe(false)
+      expect(dispatcher.projectPtyExitToLegacyClients(roleOf, params)).toBe(false)
       expect(primary.closes).toBe(0)
 
       primary.drain()
-      expect(dispatcher.projectPtyExitToMatchingClients(() => true, params)).toBe(true)
+      expect(dispatcher.projectPtyExitToLegacyClients(roleOf, params)).toBe(true)
     } finally {
       stderr.mockRestore()
       dispatcher.dispose()
     }
   })
 
-  it('never encodes a projection with no subscribers', () => {
+  it('never encodes a projection with no targets', () => {
     const primary = makeStallingClient(65536)
     primary.drain()
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
@@ -201,11 +257,14 @@ describe('RelayDispatcher slow-client PTY projection', () => {
           return 'y'.repeat(40_000)
         }
       }
+      const primaryId = dispatcher.activeClientIds()[0]
 
-      expect(dispatcher.projectPtyDataToMatchingClients(() => false, params)).toBe(true)
+      expect(dispatcher.projectPtyDataToLegacyClients(roles({}), params)).toBe(true)
       expect(reads).toBe(0)
 
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
+      expect(
+        dispatcher.projectPtyDataToLegacyClients(roles({ [primaryId]: 'owner' }), params)
+      ).toBe(true)
       expect(reads).toBeGreaterThan(0)
     } finally {
       dispatcher.dispose()
@@ -229,84 +288,58 @@ describe('RelayDispatcher slow-client PTY projection', () => {
   })
 })
 
-describe('RelayDispatcher stalled-subscriber eviction', () => {
+describe('RelayDispatcher projection never reaps a client on a timer', () => {
   afterEach(() => {
     vi.useRealTimers()
   })
 
-  it('detaches a subscriber that never drains so the paused PTY can resume', () => {
+  it('never closes a peer that is merely mid-write while another target back-pressures', () => {
     vi.useFakeTimers()
-    const primary = makeStallingClient(65536)
-    primary.drain()
-    const stalled = makeStallingClient(65536)
-    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const owner = makeStallingClient(65536)
+    const neverDrains = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(owner.write, owner.options)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    let midWriteCloses = 0
     try {
-      const stalledId = dispatcher.attachClient(stalled.write, stalled.options)
-      fillProducerQueue(dispatcher, stalledId)
-      let capacityNotifications = 0
-      dispatcher.onLegacyPtyCapacity(() => {
-        capacityNotifications++
+      const ownerId = dispatcher.activeClientIds()[0]
+      const neverDrainsId = dispatcher.attachClient(neverDrains.write, neverDrains.options)
+      const midWriteId = dispatcher.attachClient(
+        (_data, settle: (result: SinkWriteSettlement) => void) => {
+          // Accepted by the sink but not acknowledged yet: normal for any real socket mid-write.
+          void settle
+          return true
+        },
+        {
+          supportsWriteCallback: true,
+          writableHighWaterMark: () => 65536,
+          writableLength: () => 0,
+          close: () => {
+            midWriteCloses++
+          }
+        }
+      )
+      fillProducerQueue(dispatcher, ownerId)
+      fillProducerQueue(dispatcher, neverDrainsId)
+      const roleOf = roles({
+        [ownerId]: 'owner',
+        [neverDrainsId]: 'mirror',
+        [midWriteId]: 'mirror'
       })
-
       const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
-      expect(stalled.closes).toBe(0)
 
-      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS + 1)
+      expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false)
+      // Just inside the 30s window an earlier revision watched, so the peer armed here would be
+      // judged over ~1ms of observation.
+      vi.advanceTimersByTime(29_999)
+      dispatcher.tryNotifyPtyDataToClient(midWriteId, { id: 'pty-1', data: 'z' }, () => {})
+      expect(dispatcher.projectPtyDataToLegacyClients(roleOf, params)).toBe(false)
+      vi.advanceTimersByTime(10 * 60_000)
 
-      expect(stalled.closes).toBe(1)
-      // The eviction is only worth anything if it un-wedges the producer it was blocking.
-      expect(capacityNotifications).toBeGreaterThan(0)
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
-      expect(primary.closes).toBe(0)
-    } finally {
-      stderr.mockRestore()
-      dispatcher.dispose()
-    }
-  })
-
-  it('never evicts the primary link, however long it back-pressures', () => {
-    vi.useFakeTimers()
-    const primary = makeStallingClient(65536)
-    const dispatcher = new RelayDispatcher(primary.write, primary.options)
-    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
-    try {
-      fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0])
-
-      const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
-
-      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS * 10)
-
-      expect(primary.closes).toBe(0)
-      expect(dispatcher.activeClientIds()).toHaveLength(1)
-    } finally {
-      stderr.mockRestore()
-      dispatcher.dispose()
-    }
-  })
-
-  it('spares a subscriber that drains inside the stall window', () => {
-    vi.useFakeTimers()
-    const primary = makeStallingClient(65536)
-    primary.drain()
-    const slow = makeStallingClient(65536, true)
-    const dispatcher = new RelayDispatcher(primary.write, primary.options)
-    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
-    try {
-      const slowId = dispatcher.attachClient(slow.write, slow.options)
-      fillProducerQueue(dispatcher, slowId)
-
-      const params = { id: 'pty-1', data: 'y'.repeat(40_000) }
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(false)
-
-      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS - 1)
-      slow.drain()
-      vi.advanceTimersByTime(PROJECTION_STALL_EVICTION_MS * 4)
-
-      expect(slow.closes).toBe(0)
-      expect(dispatcher.projectPtyDataToMatchingClients(() => true, params)).toBe(true)
+      // Reaping a peer on a timer is exactly the disconnect #12041 reported.
+      expect(midWriteCloses).toBe(0)
+      expect(neverDrains.closes).toBe(0)
+      expect(owner.closes).toBe(0)
+      expect(dispatcher.activeClientIds()).toHaveLength(3)
     } finally {
       stderr.mockRestore()
       dispatcher.dispose()

src/relay/dispatcher.ts

@@ -27,7 +27,6 @@ import {
   LegacyRelayPublicationLedger,
   type LegacyPublicationLease
 } from './legacy-relay-publication-ledger'
-import { ProjectionStallEvictor } from './dispatcher-projection-stall'
 
 export type {
   RelayClientSinkOptions,
@@ -55,6 +54,15 @@ export type RelayClientSourceOptions = {
   resumeReads?: () => void
 }
 
+/**
+ * How the legacy PTY projection serves a client whose producer queue is full.
+ * - `skip`: not a projection target (a flow-controlled source owner has its own credited lane).
+ * - `owner`: the projection is this client's only delivery lane, so a full queue back-pressures the PTY.
+ * - `mirror`: a bystander copy that recovers by reopening the pty, so a full queue sheds this one frame for
+ *   that client alone — a mirror must never hold the native PTY paused for every other viewer.
+ */
+export type LegacyProjectionRole = 'skip' | 'owner' | 'mirror'
+
 export type MethodHandler = (
   params: Record<string, unknown>,
   context: RequestContext
@@ -101,15 +109,6 @@ export class RelayDispatcher {
   private notificationHandlers = new Map<string, NotificationHandler>()
   private readonly requestAborts = new ClientRequestAborts()
   private readonly publicationLedger = new LegacyRelayPublicationLedger()
-  private readonly projectionStalls = new ProjectionStallEvictor<RelayClient>(
-    (client) => ({
-      stalled: !client.closed && client.writer.retainedProducerBytes > 0,
-      drainProgressBytes: client.writer.drainProgressBytes
-    }),
-    (client) => {
-      this.closeClient(client, new Error('Relay PTY subscriber stalled without draining'), true)
-    }
-  )
   private pendingRelayRequests = new Map<number, PendingRelayRequest>()
   private clientDetachListeners = new Set<(clientId: number) => void>()
   private disposeListeners = new Set<() => void>()
@@ -344,8 +343,8 @@ export class RelayDispatcher {
     )
   }
 
-  projectPtyDataToMatchingClients(
-    matchesClient: (clientId: number) => boolean,
+  projectPtyDataToLegacyClients(
+    roleOf: (clientId: number) => LegacyProjectionRole,
     params: Record<string, unknown>,
     options: { interactive?: boolean } = {}
   ): boolean {
@@ -353,7 +352,7 @@ export class RelayDispatcher {
       return false
     }
     return this.projectToClients(
-      this.activeClients().filter((client) => matchesClient(client.id)),
+      roleOf,
       { jsonrpc: '2.0', method: 'pty.data', params },
       options.interactive ? 'interactive' : 'ordinary'
     )
@@ -410,18 +409,14 @@ export class RelayDispatcher {
     )
   }
 
-  projectPtyExitToMatchingClients(
-    matchesClient: (clientId: number) => boolean,
+  projectPtyExitToLegacyClients(
+    roleOf: (clientId: number) => LegacyProjectionRole,
     params: Record<string, unknown>
   ): boolean {
     if (this.disposed) {
       return false
     }
-    return this.projectToClients(
-      this.activeClients().filter((client) => matchesClient(client.id)),
-      { jsonrpc: '2.0', method: 'pty.exit', params },
-      'ordinary'
-    )
+    return this.projectToClients(roleOf, { jsonrpc: '2.0', method: 'pty.exit', params }, 'ordinary')
   }
 
   tryNotifyPtyExitToClient(
@@ -768,7 +763,6 @@ export class RelayDispatcher {
       return
     }
     this.disposed = true
-    this.projectionStalls.dispose()
     if (this.keepaliveTimer) {
       clearInterval(this.keepaliveTimer)
       this.keepaliveTimer = null
@@ -1135,38 +1129,54 @@ export class RelayDispatcher {
   }
 
   private projectToClients(
-    clients: readonly RelayClient[],
+    roleOf: (clientId: number) => LegacyProjectionRole,
     msg: JsonRpcNotification,
     lane: 'interactive' | 'ordinary'
   ): boolean {
+    const owners: RelayClient[] = []
+    const mirrors: RelayClient[] = []
+    for (const client of this.clients.values()) {
+      if (client.closed) {
+        continue
+      }
+      const role = roleOf(client.id)
+      if (role === 'owner') {
+        owners.push(client)
+      } else if (role === 'mirror') {
+        mirrors.push(client)
+      }
+    }
     // Why: having no projection target is the norm (a flow-controlled owner is excluded from the legacy
     // mirror), and that is the PTY hot path — never encode the frame just to discard it.
-    if (clients.length === 0) {
+    if (owners.length === 0 && mirrors.length === 0) {
       return true
     }
     const bytes = this.estimateFrameBytes(msg)
-    // Why: closing over a transiently full queue just makes the producer regenerate the same span for a
-    // fresh sink of identical capacity, so back-pressure instead; only a frame that can never be admitted
-    // is shed, because retrying it forever would wedge the PTY.
-    const deliverable = clients.filter((client) => {
-      if (client.closed) {
-        return false
-      }
+    // Why: closing over a transiently full queue only makes the producer regenerate the same span for a
+    // fresh sink of identical capacity, so an owner — which has no second delivery lane — back-pressures
+    // instead. A frame past its frame capacity can never be admitted, so shed that one or the retry wedges.
+    const blocking = owners.filter((client) => {
       if (bytes <= client.writer.producerFrameCapacity) {
         return true
       }
       this.logDroppedProducerNotification(client, msg.method, bytes)
       return false
     })
-    if (this.tryPublishToClients(deliverable, msg, lane, bytes)) {
-      return true
-    }
-    // Why: the pause this rejection triggers is dispatcher-wide, so a sink that never drains would freeze
-    // every PTY for every viewer; arm the bounded reap that detaches it (it reconnects and replays).
-    this.projectionStalls.noteRefusedProjection(
-      deliverable.filter((client) => client !== this.primaryClient)
-    )
-    return false
+    return this.runPublicationTransaction(() => {
+      if (!this.tryPublishToClients(blocking, msg, lane, bytes)) {
+        return false
+      }
+      // Why: a mirror is a bystander view that recovers on its own (reopening the pty replays it), so it
+      // must never hold the native PTY paused for every other viewer. Publishing only after the owners
+      // committed also keeps a back-pressured span from reaching a mirror twice when it is retried.
+      for (const client of mirrors) {
+        if (client.closed || this.publishToClient(client, msg, lane, undefined, bytes)) {
+          continue
+        }
+        this.logDroppedProducerNotification(client, msg.method, bytes)
+      }
+      return !this.disposed
+    })
   }
 
   private publishToClient(
@@ -1294,7 +1304,6 @@ export class RelayDispatcher {
       return
     }
     client.closed = true
-    this.projectionStalls.forget(client)
     this.requestAborts.abortClient(client.id)
     client.writer.close(error)
     client.generation++

src/relay/pty-handler-source-publication.test.ts

@@ -652,7 +652,7 @@ describe('PtyHandler negotiated source publication', () => {
     await spawn({})
     const appendSpy = vi.spyOn(adapter, 'appendSource')
     const projectSpy = vi
-      .spyOn(dispatcher, 'projectPtyDataToMatchingClients')
+      .spyOn(dispatcher, 'projectPtyDataToLegacyClients')
       .mockReturnValueOnce(false)
     // Why: a larger capacity result on the retry must not move the memoized span boundary.
     const maxCharsSpy = vi.spyOn(dispatcher, 'maxLegacyPtyDataChars')
@@ -679,7 +679,7 @@ describe('PtyHandler negotiated source publication', () => {
   it('does not emit pty.exit ahead of output a failed projection re-queued', async () => {
     await spawn({})
     const projectSpy = vi
-      .spyOn(dispatcher, 'projectPtyDataToMatchingClients')
+      .spyOn(dispatcher, 'projectPtyDataToLegacyClients')
       .mockImplementationOnce(() => {
         // Why: capacity fans out from a bare socket callback mid-drain, when the captured queue has
         // already been removed from pendingOutputByPty and the exit guard would read it as empty.
@@ -709,28 +709,17 @@ describe('PtyHandler negotiated source publication', () => {
     expect(ordered.indexOf('pty.data')).toBeLessThan(ordered.lastIndexOf('pty.exit'))
   })
 
-  it('pauses the PTY instead of detaching a subscriber that saturates, and resumes on drain', async () => {
+  it('keeps the native PTY, the V1 owner and a saturated subscriber all live', async () => {
     await spawn({})
     const detached: number[] = []
     const healthyWrites: Buffer[] = []
-    const saturatedSettlements: ((result: SinkWriteSettlement) => void)[] = []
-    const saturatedDrainWaiters: (() => void)[] = []
-    let saturatedStalled = true
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     dispatcher.onClientDetached((clientId) => detached.push(clientId))
-    const saturatedId = dispatcher.attachClient(
-      (_data, settle) => {
-        saturatedSettlements.push(settle)
-        return !saturatedStalled
-      },
-      {
-        supportsWriteCallback: true,
-        writableLength: () => (saturatedStalled ? 16 * 1024 : 0),
-        writableHighWaterMark: () => 4 * 1024 * 1024,
-        waitWriteDrain: (callback: () => void) => {
-          saturatedDrainWaiters.push(callback)
-        }
-      }
-    )
+    const saturatedId = dispatcher.attachClient(() => false, {
+      supportsWriteCallback: true,
+      writableLength: () => 16 * 1024,
+      writableHighWaterMark: () => 4 * 1024 * 1024
+    })
     const healthyId = dispatcher.attachClient(
       (data, settle) => {
         healthyWrites.push(Buffer.from(data))
@@ -739,8 +728,6 @@ describe('PtyHandler negotiated source publication', () => {
       },
       { supportsWriteCallback: true }
     )
-    const healthyPtyData = (): unknown[] =>
-      healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
     const payload = 's'.repeat(16 * 1024)
     let admitted = 0
     while (
@@ -754,26 +741,18 @@ describe('PtyHandler negotiated source publication', () => {
 
     expect(admitted).toBeGreaterThan(100)
     expect(admitted).toBeLessThan(140)
+    // Neither closed (that is #12041) nor gating: a bystander subscriber sheds its own copy only.
     expect(detached).toEqual([])
     expect(detached).not.toContain(healthyId)
-    // All-or-nothing: the healthy mirror waits rather than taking a copy the retry would duplicate.
-    expect(healthyPtyData()).toHaveLength(0)
-    expect(pausePty).toHaveBeenCalled()
-    // The span is already in the source ledger, so the credited owner keeps draining it.
+    expect(
+      healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
+    ).toHaveLength(1)
     expect(sourceDataFrames()).toHaveLength(1)
     expect(publication.getDebugSnapshot()).toMatchObject({ sendCommitted: 1 })
-
-    saturatedStalled = false
-    for (const waiter of saturatedDrainWaiters.splice(0)) {
-      waiter()
-    }
-    while (saturatedSettlements.length > 0) {
-      saturatedSettlements.shift()!({ ok: true })
-    }
-    await vi.advanceTimersByTimeAsync(8)
-
-    expect(healthyPtyData()).toHaveLength(1)
-    expect(sourceDataFrames()).toHaveLength(1)
-    expect(detached).toEqual([])
+    expect(pausePty).not.toHaveBeenCalled()
+    expect(
+      stderr.mock.calls.filter((call) => /Dropped pty\.data/.test(String(call[0])))
+    ).toHaveLength(1)
+    stderr.mockRestore()
   })
 })

src/relay/relay-pty-source-exit-publication.test.ts

@@ -106,7 +106,7 @@ function createScenario(
     tryNotifyPtyExitToMatchingClients: vi.fn(
       (_matches: (clientId: number) => boolean, _params: unknown) => notifyAccepted
     ),
-    projectPtyExitToMatchingClients: vi.fn(() => true),
+    projectPtyExitToLegacyClients: vi.fn(() => true),
     tryNotifyPtyExitToClient: vi.fn(() => true)
   }
   const sender = { pump: vi.fn(), wakeSendWaiters: vi.fn() }
@@ -315,7 +315,7 @@ describe('RelayPtySourceLegacyExitIndex', () => {
   it('keeps the exit pending when the subscriber projection is refused', () => {
     const record = deliveryRecord({ sealed: true })
     const scenario = createScenario(closedSnapshot({ state: 'sealed-unsettled' }), record)
-    scenario.dispatcher.projectPtyExitToMatchingClients.mockReturnValueOnce(false)
+    scenario.dispatcher.projectPtyExitToLegacyClients.mockReturnValueOnce(false)
 
     expect(scenario.run()).toBe(false)
     expect(record.legacyExitAccepted).toBe(false)
@@ -325,7 +325,7 @@ describe('RelayPtySourceLegacyExitIndex', () => {
     // The handler's retry after drain lands the exit once, with no duplicate legacy projection.
     expect(scenario.run()).toBe(true)
     expect(record.legacyExitAccepted).toBe(true)
-    expect(scenario.dispatcher.projectPtyExitToMatchingClients).toHaveBeenCalledTimes(2)
+    expect(scenario.dispatcher.projectPtyExitToLegacyClients).toHaveBeenCalledTimes(2)
     expect(scenario.dispatcher.tryNotifyPtyExitToClient).toHaveBeenCalledOnce()
   })
 
@@ -338,7 +338,7 @@ describe('RelayPtySourceLegacyExitIndex', () => {
     })
 
     expect(() => scenario.runTracked(index)).toThrow('owner write failed')
-    expect(scenario.dispatcher.projectPtyExitToMatchingClients).toHaveBeenCalledOnce()
+    expect(scenario.dispatcher.projectPtyExitToLegacyClients).toHaveBeenCalledOnce()
     expect(scenario.session.cancelDelivery).toHaveBeenCalledWith(
       record.identity,
       'exit-publication-failed'

src/relay/relay-pty-source-exit-publication.ts

@@ -1,4 +1,5 @@
 import type { RelayDispatcher } from './dispatcher'
+import { legacyProjectionRole } from './relay-pty-source-output'
 import {
   onceSinkSettlement,
   type RelayPtySourceDeliveryRecord,
@@ -177,8 +178,8 @@ export function sealAndPublishPtySourceExit(options: PtySourceExitOptions): bool
     return false
   }
   if (!record.legacyExitAccepted) {
-    record.legacyExitAccepted = dispatcher.projectPtyExitToMatchingClients(
-      (clientId) => session.deliveryMode(clientId) !== 'source-owner',
+    record.legacyExitAccepted = dispatcher.projectPtyExitToLegacyClients(
+      (clientId) => legacyProjectionRole(session, clientId),
       params
     )
     options.legacyExits?.remember(params, record.legacyExitAccepted)

src/relay/relay-pty-source-output.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from 'node:crypto'
 import type { PtySourceDeliveryIdentity } from '../shared/pty-source-credit-contract'
-import type { RelayDispatcher } from './dispatcher'
+import type { LegacyProjectionRole, RelayDispatcher } from './dispatcher'
 import type { RelayPtySourceDeliveryRecord } from './relay-pty-source-send-scheduler'
 import type { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter'
 
@@ -50,6 +50,22 @@ export function appendPtySourceOutput(
   return true
 }
 
+/**
+ * Only a session owner is served solely by the legacy projection, so only it may back-pressure the native
+ * PTY. Every other target — agent hooks, `orca` CLI processes, a second window, a client still completing
+ * pty.openClient — is a bystander mirror that would otherwise wedge the terminal for everyone.
+ */
+export function legacyProjectionRole(
+  session: SshPtyConsumerSessionAdapter,
+  clientId: number
+): LegacyProjectionRole {
+  const mode = session.deliveryMode(clientId)
+  if (mode === 'source-owner') {
+    return 'skip'
+  }
+  return mode === 'legacy-owner' ? 'owner' : 'mirror'
+}
+
 export function projectPtySourceOutputToLegacy(
   dispatcher: RelayDispatcher,
   session: SshPtyConsumerSessionAdapter,
@@ -57,8 +73,8 @@ export function projectPtySourceOutputToLegacy(
   output: RelayPtySourceOutput,
   interactive: boolean
 ): boolean {
-  return dispatcher.projectPtyDataToMatchingClients(
-    (clientId) => session.deliveryMode(clientId) !== 'source-owner',
+  return dispatcher.projectPtyDataToLegacyClients(
+    (clientId) => legacyProjectionRole(session, clientId),
     {
       id,
       data: output.data,

src/relay/relay-pty-source-publication.test.ts

@@ -283,7 +283,7 @@ describe('RelayPtySourcePublication', () => {
     ).toHaveLength(2)
   })
 
-  it('backpressures instead of detaching a saturated subscriber while the V1 owner stays live', async () => {
+  it('sheds a saturated subscriber without detaching it or stalling the V1 owner', async () => {
     const harness = await createHarness(8)
     const detached: number[] = []
     const saturatedWrites: Buffer[] = []
@@ -324,8 +324,8 @@ describe('RelayPtySourcePublication', () => {
 
     expect(admitted).toBeGreaterThan(0)
     expect(admitted).toBeLessThan(20)
-    // False tells pty-handler to pause the PTY and republish the span once the sink drains.
-    expect(harness.publication.publish('pty-1', { data: saturatedPayload }, false)).toBe(false)
+    // A bystander subscriber is neither closed (#12041) nor allowed to pause the PTY for everyone.
+    expect(harness.publication.publish('pty-1', { data: saturatedPayload }, false)).toBe(true)
 
     expect(detached).toEqual([])
     expect(detached).not.toContain(healthyId)
@@ -333,8 +333,7 @@ describe('RelayPtySourcePublication', () => {
     expect(heldSettlements).toHaveLength(1)
     expect(
       healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
-    ).toHaveLength(0)
-    // The span is already appended to the source ledger, so the credited owner still drains it.
+    ).toHaveLength(1)
     expect(
       harness.writes.map(notification).filter((frame) => frame?.method === 'pty.data')
     ).toHaveLength(1)