From b5e3983fc0194d3458db8416bcc8b27635b97053 Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sun, 2 Aug 2026 22:14:03 -0700 Subject: [PATCH] fix(relay): stop dropping the SSH connection when a terminal outruns a slow link MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../dispatcher-capacity-degradation.test.ts | 9 +- src/relay/dispatcher-slow-client.test.ts | 247 ++++++++++++++++++ src/relay/dispatcher.ts | 38 +-- .../pty-handler-source-publication.test.ts | 82 +++++- src/relay/pty-handler.ts | 13 +- .../relay-pty-source-exit-publication.test.ts | 17 ++ .../relay-pty-source-publication.test.ts | 10 +- src/relay/relay-pty-source-publication.ts | 11 +- 8 files changed, 386 insertions(+), 41 deletions(-) create mode 100644 src/relay/dispatcher-slow-client.test.ts diff --git a/src/relay/dispatcher-capacity-degradation.test.ts b/src/relay/dispatcher-capacity-degradation.test.ts index a546989d5df..f40fd6e5685 100644 --- a/src/relay/dispatcher-capacity-degradation.test.ts +++ b/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() diff --git a/src/relay/dispatcher-slow-client.test.ts b/src/relay/dispatcher-slow-client.test.ts new file mode 100644 index 00000000000..f89d22a9d29 --- /dev/null +++ b/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((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() + } + }) +}) diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index 98e8959547f..c5b531db005 100644 --- a/src/relay/dispatcher.ts +++ b/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( diff --git a/src/relay/pty-handler-source-publication.test.ts b/src/relay/pty-handler-source-publication.test.ts index 48a5f74904f..ffe7f3c45a6 100644 --- a/src/relay/pty-handler-source-publication.test.ts +++ b/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([]) }) }) diff --git a/src/relay/pty-handler.ts b/src/relay/pty-handler.ts index c928bd4a106..085d6c9486e 100644 --- a/src/relay/pty-handler.ts +++ b/src/relay/pty-handler.ts @@ -360,6 +360,9 @@ export class PtyHandler { private pendingOutputByPty = new Map() private pendingExitByPty = new Map() private pausedOutputPtys = new Set() + // 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() private consumerPausedOutputPtys = new Set() 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) diff --git a/src/relay/relay-pty-source-exit-publication.test.ts b/src/relay/relay-pty-source-exit-publication.test.ts index a7820c4aa5f..f5a567a33fc 100644 --- a/src/relay/relay-pty-source-exit-publication.test.ts +++ b/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) diff --git a/src/relay/relay-pty-source-publication.test.ts b/src/relay/relay-pty-source-publication.test.ts index 1d30e4e8540..b79895f2713 100644 --- a/src/relay/relay-pty-source-publication.test.ts +++ b/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) diff --git a/src/relay/relay-pty-source-publication.ts b/src/relay/relay-pty-source-publication.ts index b21e8a79cf2..32167b3b327 100644 --- a/src/relay/relay-pty-source-publication.ts +++ b/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 =>