fix(relay): stop dropping the SSH connection when a terminal outruns a slow link

Two producer paths in the relay still destroyed a client's sink over a
transiently-full write queue instead of applying PTY backpressure:

- `RelayDispatcher.projectToClients` closed any subscriber whose producer
  queue was momentarily full, and still reported the projection as
  delivered, so the PTY was never paused.
- The `pty.replay` branch of `notify()` routed replay to the fatal control
  lane, so a workspace restoring many terminals over a slow link killed the
  link — and reattach regenerated the same replay, looping forever.

Both now back-pressure: a full queue fails the projection so the producer
pauses the native PTY and republishes the memoized span on drain, and a
frame no sink can ever admit is shed with one log line per generation
rather than closing the connection.

`RelayPtySourcePublication.publish` now pumps the source sender even when
the legacy projection is rejected, so a stalled legacy mirror cannot starve
the flow-controlled V1 owner's credited stream.

`PtyHandler` also guards the publish window: the pty's queue is lifted out
of `pendingOutputByPty` during a publish, so a capacity callback re-entering
from the sink could previously let `pty.exit` overtake still-pending output.

Closes #12041

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

src/relay/dispatcher-capacity-degradation.test.ts

@@ -300,7 +300,7 @@ describe('RelayDispatcher bounded-capacity degradation', () => {
     }
   })
 
-  it('closes the client when pty.replay overflows the control queue', () => {
+  it('drops pty.replay when it overflows the control queue instead of closing', () => {
     const primary = makeSaturatingClient(65536)
     const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
     const bounded = new RelayDispatcher(primary.write, primary.options)
@@ -310,10 +310,13 @@ describe('RelayDispatcher bounded-capacity degradation', () => {
         bounded.notifyClient(clientId, `control.${index}`)
       }
       expect(primary.closes).toBe(0)
+      stderr.mockClear()
 
-      // Replay is never retried, so an unnoticed drop would strand the pane on a short buffer.
+      // Reattach regenerates replay, so closing here would only re-kill the reconnect.
       bounded.notify('pty.replay', { paneKey: 'tab-1:pane-1', data: 'x' })
-      expect(primary.closes).toBe(1)
+      expect(primary.closes).toBe(0)
+      expect(stderr).toHaveBeenCalledTimes(1)
+      expect(String(stderr.mock.calls[0][0])).toMatch(/^\[relay\] Dropped pty\.replay \(/)
     } finally {
       stderr.mockRestore()
       bounded.dispose()

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

@@ -0,0 +1,247 @@
+import { describe, expect, it, vi } from 'vitest'
+import { RelayDispatcher, type RelayClientSinkOptions } from './dispatcher'
+import { DISPATCHER_CONTROL_QUEUE_MAX_FRAMES } from './dispatcher-writer-admission'
+
+type SlowClient = {
+  frames: Buffer[]
+  closes: number
+  options: RelayClientSinkOptions
+  write: (data: Buffer) => boolean
+  drain: () => void
+}
+
+// Why: a sink that refuses every write until told otherwise reproduces the ~268ms link in #12041, where
+// full-screen redraws outrun the client's drain rate and saturate the bounded producer lane.
+function makeStallingClient(highWaterMark: number, drainable = false): SlowClient {
+  let stalled = true
+  let drainWaiter: (() => void) | null = null
+  const client: SlowClient = {
+    frames: [],
+    closes: 0,
+    write: (data: Buffer) => {
+      client.frames.push(Buffer.from(data))
+      return !stalled
+    },
+    options: {
+      writableHighWaterMark: () => highWaterMark,
+      writableLength: () => 0,
+      close: () => {
+        client.closes++
+      },
+      ...(drainable
+        ? {
+            waitWriteDrain: (callback: () => void) => {
+              drainWaiter = callback
+            }
+          }
+        : {})
+    },
+    drain: () => {
+      stalled = false
+      const waiter = drainWaiter
+      drainWaiter = null
+      waiter?.()
+    }
+  }
+  return client
+}
+
+function ptyDataFrameCount(frames: Buffer[]): number {
+  return frames.filter((frame) => {
+    const length = frame.readUInt32BE(9)
+    const message = JSON.parse(frame.subarray(13, 13 + length).toString('utf-8')) as {
+      method?: string
+    }
+    return message.method === 'pty.data'
+  }).length
+}
+
+// Fills the 2 MiB producer lane to the byte, so even a bare pty.exit no longer fits.
+function fillProducerQueue(dispatcher: RelayDispatcher, clientId: number): void {
+  for (const size of [40_000, 32]) {
+    while (
+      dispatcher.tryNotifyPtyDataToClient(
+        clientId,
+        { id: 'pty-1', data: 'x'.repeat(size) },
+        () => {}
+      )
+    ) {
+      /* saturate */
+    }
+  }
+}
+
+describe('RelayDispatcher slow-client PTY projection', () => {
+  it('tryNotifyPtyData applies backpressure without closing the client', () => {
+    const primary = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    try {
+      let accepted = 0
+      let rejected = 0
+      for (let index = 0; index < 200; index++) {
+        if (dispatcher.tryNotifyPtyData({ id: 'pty-1', data: 'x'.repeat(40_000) })) {
+          accepted++
+        } else {
+          rejected++
+        }
+      }
+      expect(accepted).toBeGreaterThan(0)
+      expect(rejected).toBeGreaterThan(0)
+      expect(primary.closes).toBe(0)
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
+  it('rejects a projected pty.data frame instead of closing a subscriber 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 projected = dispatcher.projectPtyDataToMatchingClients(() => true, {
+        id: 'pty-1',
+        data: 'y'.repeat(40_000)
+      })
+
+      // A false return is what makes pty-handler pause the PTY and republish the span after drain.
+      expect(projected).toBe(false)
+      expect(primary.closes).toBe(0)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('sheds a projected frame no 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 {
+      // Larger than producerFrameCapacity (49152B), so retrying it forever would stall the PTY.
+      const projected = dispatcher.projectPtyDataToMatchingClients(() => true, {
+        id: 'pty-1',
+        data: 'y'.repeat(60_000)
+      })
+
+      expect(projected).toBe(true)
+      expect(primary.closes).toBe(0)
+      const drops = stderr.mock.calls.filter((call) =>
+        /producer frame capacity/.test(String(call[0]))
+      )
+      expect(drops).toHaveLength(1)
+      expect(String(drops[0][0])).toMatch(
+        /^\[relay\] Dropped pty\.data \(\d+B > producer frame capacity 49152B\)\n$/
+      )
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  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)
+    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 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)
+
+      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)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('holds a projected pty.exit for retry instead of closing a saturated subscriber', () => {
+    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 params = { id: 'pty-1', exitCode: 0 }
+      expect(dispatcher.projectPtyExitToMatchingClients(() => true, params)).toBe(false)
+      expect(primary.closes).toBe(0)
+
+      primary.drain()
+      expect(dispatcher.projectPtyExitToMatchingClients(() => true, params)).toBe(true)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('a >1MiB response while the producer queue is full does not close the client', async () => {
+    const primary = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      dispatcher.onRequest('pty.serialize', async () => ({ data: 'z'.repeat(1_500_000) }))
+      fillProducerQueue(dispatcher, dispatcher.activeClientIds()[0])
+
+      await new Promise<void>((resolve) => setTimeout(resolve, 50))
+      expect(primary.closes).toBe(0)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+})
+
+describe('RelayDispatcher pty.replay under a full control lane', () => {
+  it('drops replay instead of killing a link that would regenerate the same replay', () => {
+    const primary = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      // REPLAY_BUFFER_MAX is 100 KiB per PTY; a restored workspace replays many terminals at once.
+      for (let index = 0; index < 12; index++) {
+        dispatcher.notify('pty.replay', { id: `pty-${index}`, data: 'x'.repeat(100 * 1024) })
+      }
+      expect(primary.closes).toBe(0)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
+  it('logs one dropped-replay line per generation however many panes are stranded', () => {
+    const primary = makeStallingClient(65536)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    try {
+      const clientId = dispatcher.activeClientIds()[0]
+      for (let index = 0; index < DISPATCHER_CONTROL_QUEUE_MAX_FRAMES; index += 1) {
+        dispatcher.notifyClient(clientId, `control.${index}`)
+      }
+      stderr.mockClear()
+
+      dispatcher.notify('pty.replay', { id: 'pty-1', data: 'x' })
+      expect(primary.closes).toBe(0)
+      expect(stderr).toHaveBeenCalledTimes(1)
+      expect(String(stderr.mock.calls[0][0])).toMatch(/^\[relay\] Dropped pty\.replay \(/)
+
+      dispatcher.notify('pty.replay', { id: 'pty-2', data: 'x' })
+      expect(primary.closes).toBe(0)
+      expect(stderr).toHaveBeenCalledTimes(1)
+    } finally {
+      stderr.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+})

src/relay/dispatcher.ts

@@ -530,9 +530,11 @@ export class RelayDispatcher {
           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.
-          this.enqueueFrame(client, msg, 'control', undefined, frameBytes)
+          // Why: reattach regenerates replay, so killing the link over it just re-kills the reconnect;
+          // drop the pane's scrollback instead and let live output repaint it.
+          if (!this.enqueueFrame(client, msg, 'control', undefined, frameBytes, 'reject')) {
+            this.logDroppedProducerNotification(client, method, frameBytes)
+          }
           continue
         }
         // Why: closing can never make an oversized frame sendable — the producer regenerates it after
@@ -1088,13 +1090,15 @@ export class RelayDispatcher {
   private tryPublishToClients(
     clients: readonly RelayClient[],
     msg: JsonRpcNotification,
-    lane: 'interactive' | 'ordinary' | 'bulk'
+    lane: 'interactive' | 'ordinary' | 'bulk',
+    // Why: projection already sized the frame to pick shed targets; avoid a redundant encode.
+    estimatedBytes?: number
   ): boolean {
     return this.runPublicationTransaction(() => {
       if (clients.length === 0) {
         return true
       }
-      const bytes = this.estimateFrameBytes(msg)
+      const bytes = estimatedBytes ?? this.estimateFrameBytes(msg)
       if (clients.some((client) => !client.writer.canEnqueueProducer(bytes))) {
         return false
       }
@@ -1124,19 +1128,21 @@ export class RelayDispatcher {
     msg: JsonRpcNotification,
     lane: 'interactive' | 'ordinary'
   ): boolean {
-    return this.runPublicationTransaction(() => {
-      for (const client of clients) {
-        if (client.closed || this.publishToClient(client, msg, lane)) {
-          continue
-        }
-        this.closeClient(
-          client,
-          new Error('Relay PTY subscriber projection capacity exceeded'),
-          client !== this.primaryClient
-        )
+    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
       }
-      return !this.disposed
+      if (bytes <= client.writer.producerFrameCapacity) {
+        return true
+      }
+      this.logDroppedProducerNotification(client, msg.method, bytes)
+      return false
     })
+    return this.tryPublishToClients(deliverable, msg, lane, bytes)
   }
 
   private publishToClient(

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

@@ -676,16 +676,61 @@ describe('PtyHandler negotiated source publication', () => {
     ])
   })
 
-  it('keeps the native PTY and V1 owner live when one subscriber saturates', async () => {
+  it('does not emit pty.exit ahead of output a failed projection re-queued', async () => {
+    await spawn({})
+    const projectSpy = vi
+      .spyOn(dispatcher, 'projectPtyDataToMatchingClients')
+      .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.
+        exitCallback!({ exitCode: 0 })
+        handler.handleSourcePublicationCapacity('pty-1')
+        return false
+      })
+
+    dataCallback!('abcd')
+    await vi.advanceTimersByTimeAsync(8)
+
+    expect(projectSpy).toHaveBeenCalledTimes(1)
+    expect(exitFrames()).toHaveLength(0)
+
+    projectSpy.mockRestore()
+    handler.handleSourcePublicationCapacity('pty-1')
+    await vi.advanceTimersByTimeAsync(8)
+
+    const ordered = writes
+      .map(notification)
+      .filter(
+        (frame): frame is Notification =>
+          frame?.method === 'pty.data' || frame?.method === 'pty.exit'
+      )
+      .map((frame) => frame.method)
+    expect(ordered.indexOf('pty.data')).toBeGreaterThanOrEqual(0)
+    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 () => {
     await spawn({})
     const detached: number[] = []
     const healthyWrites: Buffer[] = []
+    const saturatedSettlements: ((result: SinkWriteSettlement) => void)[] = []
+    const saturatedDrainWaiters: (() => void)[] = []
+    let saturatedStalled = true
     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) => {
+        saturatedSettlements.push(settle)
+        return !saturatedStalled
+      },
+      {
+        supportsWriteCallback: true,
+        writableLength: () => (saturatedStalled ? 16 * 1024 : 0),
+        writableHighWaterMark: () => 4 * 1024 * 1024,
+        waitWriteDrain: (callback: () => void) => {
+          saturatedDrainWaiters.push(callback)
+        }
+      }
+    )
     const healthyId = dispatcher.attachClient(
       (data, settle) => {
         healthyWrites.push(Buffer.from(data))
@@ -694,6 +739,8 @@ 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 (
@@ -707,13 +754,26 @@ describe('PtyHandler negotiated source publication', () => {
 
     expect(admitted).toBeGreaterThan(100)
     expect(admitted).toBeLessThan(140)
-    expect(detached).toEqual([saturatedId])
+    expect(detached).toEqual([])
     expect(detached).not.toContain(healthyId)
-    expect(
-      healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
-    ).toHaveLength(1)
+    // 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(sourceDataFrames()).toHaveLength(1)
     expect(publication.getDebugSnapshot()).toMatchObject({ sendCommitted: 1 })
-    expect(pausePty).not.toHaveBeenCalled()
+
+    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([])
   })
 })

src/relay/pty-handler.ts

@@ -360,6 +360,9 @@ export class PtyHandler {
   private pendingOutputByPty = new Map<string, PendingPtyOutput[]>()
   private pendingExitByPty = new Map<string, { id: string; code: number; incarnationId: string }>()
   private pausedOutputPtys = new Set<string>()
+  // Why: a publish runs with the pty's queue lifted out of pendingOutputByPty, so a capacity callback
+  // re-entering from the sink would read "no pending output" and let pty.exit overtake that span.
+  private publishingOutputPtys = new Set<string>()
   private consumerPausedOutputPtys = new Set<string>()
   private removeLegacyCapacityListener: (() => void) | null = null
   private sourcePublication: RelayPtySourcePublication | null = null
@@ -996,7 +999,13 @@ export class PtyHandler {
         ...(pending.transformed ? { transformed: true } : {})
       } satisfies RelayPtySourceOutput)
     pending.sourceChunk = sourceChunk
-    const published = this.publishPtyOutput(id, sourceChunk, pending.interactive === true)
+    this.publishingOutputPtys.add(id)
+    let published: boolean
+    try {
+      published = this.publishPtyOutput(id, sourceChunk, pending.interactive === true)
+    } finally {
+      this.publishingOutputPtys.delete(id)
+    }
     if (!published) {
       this.pendingOutputByPty.set(id, queue)
       this.pausePtyOutput(id)
@@ -1081,7 +1090,7 @@ export class PtyHandler {
   }
 
   private publishPendingExit(id: string): void {
-    if (this.pendingOutputByPty.has(id)) {
+    if (this.pendingOutputByPty.has(id) || this.publishingOutputPtys.has(id)) {
       return
     }
     const exit = this.pendingExitByPty.get(id)

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

@@ -312,6 +312,23 @@ describe('RelayPtySourceLegacyExitIndex', () => {
     ).toBe(true)
   })
 
+  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)
+
+    expect(scenario.run()).toBe(false)
+    expect(record.legacyExitAccepted).toBe(false)
+    expect(record.sourceExitState).toBe('idle')
+    expect(scenario.dispatcher.tryNotifyPtyExitToClient).not.toHaveBeenCalled()
+
+    // 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.tryNotifyPtyExitToClient).toHaveBeenCalledOnce()
+  })
+
   it('remembers a subscriber projection when the owner publication throws', () => {
     const record = deliveryRecord({ sealed: true })
     const scenario = createScenario(closedSnapshot({ state: 'sealed-unsettled' }), record)

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

@@ -283,7 +283,7 @@ describe('RelayPtySourcePublication', () => {
     ).toHaveLength(2)
   })
 
-  it('detaches only a saturated subscriber while the V1 owner stays live', async () => {
+  it('backpressures instead of detaching a saturated subscriber while the V1 owner stays live', async () => {
     const harness = await createHarness(8)
     const detached: number[] = []
     const saturatedWrites: Buffer[] = []
@@ -324,15 +324,17 @@ describe('RelayPtySourcePublication', () => {
 
     expect(admitted).toBeGreaterThan(0)
     expect(admitted).toBeLessThan(20)
-    expect(harness.publication.publish('pty-1', { data: saturatedPayload }, false)).toBe(true)
+    // 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)
 
-    expect(detached).toEqual([saturatedId])
+    expect(detached).toEqual([])
     expect(detached).not.toContain(healthyId)
     expect(saturatedWrites).toHaveLength(1)
     expect(heldSettlements).toHaveLength(1)
     expect(
       healthyWrites.map(notification).filter((frame) => frame?.method === 'pty.data')
-    ).toHaveLength(1)
+    ).toHaveLength(0)
+    // The span is already appended to the source ledger, so the credited owner still drains it.
     expect(
       harness.writes.map(notification).filter((frame) => frame?.method === 'pty.data')
     ).toHaveLength(1)

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

@@ -231,11 +231,12 @@ export class RelayPtySourcePublication {
       }
       return false
     }
-    if (!projectPtySourceOutputToLegacy(this.dispatcher, this.session, id, output, interactive)) {
-      return false
-    }
-    this.sender.pump(record)
-    return true
+    const { dispatcher, session, sender } = this
+    const projected = projectPtySourceOutputToLegacy(dispatcher, session, id, output, interactive)
+    // Why: the span is already in the source ledger, so a stalled legacy subscriber must not also
+    // starve the flow-controlled owner — only the PTY pause is shared.
+    sender.pump(record)
+    return projected
   }
 
   sealAndPublishExit = (params: PtyExitParams): boolean =>