From d99888fef6d2e24aa7b96b54328dec3d50086360 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Sat, 1 Aug 2026 01:01:11 -0700 Subject: [PATCH] fix(relay,ssh): restore relay-shed status fields and scope backpressure per client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 3 + 4 review fixes. Watcher chunking is now gated on the *client's* retention reserve rather than the dispatcher-wide one, so one stalled peer no longer forces a healthy client into a full file-tree resync. The relay-lost redeploy ladder no longer burns its 6-attempt budget while the SSH transport itself is down: it holds at the 15s step with a non-terminal status and rearms, so a laptop that slept past the ladder comes back instead of landing on a terminal "give up" banner. The shedFields wire marker had no consumer, so an agent-hook envelope whose subagent roster was dropped to fit the frame read as "roster cleared" on the Orca side: live child rows blanked and a done pane became hibernation-eligible while its teammates were still running. ingestRemote now restores shed fields from the cached payload (interactivePrompt deliberately excluded — a stale answerable question card is worse than none). Also: stream terminal-frame slots are counted per client, since the control queue they protect is per client; the chunking fast path no longer logs a drop for a batch it goes on to deliver in full; -32010 is now RelayErrorCode.ResponseOverCapacity. Test debt from the review: pending-pane eviction, per-client stream isolation, and the reconnect budget are now asserted rather than assumed; four fragile exact-byte pins dropped in favour of the tier comparisons that carry the requirement. --- src/main/agent-hooks/server.test.ts | 45 +++++++ src/main/agent-hooks/server.ts | 15 ++- src/main/ipc/ssh.test.ts | 127 ++++++++++++++++++ src/main/ipc/ssh.ts | 67 +++++++-- src/main/ssh/ssh-connection.test.ts | 8 +- src/main/ssh/ssh-relay-session.ts | 3 + .../agent-hook-envelope-publication.test.ts | 25 ++++ src/relay/agent-hook-envelope-publication.ts | 12 +- .../dispatcher-capacity-degradation.test.ts | 38 +++++- src/relay/dispatcher.ts | 28 +++- src/relay/fs-handler-file-read.ts | 26 +++- src/relay/fs-handler-stream.test.ts | 10 +- src/relay/protocol.ts | 5 +- ...ay-oversized-notification-survival.test.ts | 13 +- src/relay/relay-watcher-event-emitter.test.ts | 110 ++++++++++++++- src/relay/relay-watcher-event-emitter.ts | 17 ++- src/shared/agent-hook-relay.test.ts | 56 ++++++++ src/shared/agent-hook-relay.ts | 38 ++++++ 18 files changed, 599 insertions(+), 44 deletions(-) diff --git a/src/main/agent-hooks/server.test.ts b/src/main/agent-hooks/server.test.ts index c8baaa6e69c..faefe48b223 100644 --- a/src/main/agent-hooks/server.test.ts +++ b/src/main/agent-hooks/server.test.ts @@ -187,6 +187,51 @@ describe('AgentHookServer listener replay', () => { expect(server.getStatusSnapshot()[0]?.subagents).toEqual([child('new-child')]) }) + it('restores a subagent roster the relay shed to fit the frame', () => { + const server = new AgentHookServer() + const roster = [ + { id: 'reviewer-1', agentType: 'reviewer', state: 'working' as const, startedAt: 1 } + ] + server.ingestRemote( + { + paneKey: PANE, + payload: { state: 'working', prompt: 'review', agentType: 'claude', subagents: roster } + }, + 'conn-1' + ) + // The relay dropped the roster to fit an oversized frame and said so on the wire. + server.ingestRemote( + { + paneKey: PANE, + shedFields: ['lastAssistantMessage', 'subagents'], + payload: { state: 'done', prompt: 'review', agentType: 'claude' } + }, + 'conn-1' + ) + expect(server.getStatusSnapshot()[0]).toMatchObject({ state: 'done', subagents: roster }) + }) + + it('lets an unmarked absent roster clear, so a finished team still retires', () => { + const server = new AgentHookServer() + server.ingestRemote( + { + paneKey: PANE, + payload: { + state: 'working', + prompt: 'review', + agentType: 'claude', + subagents: [{ id: 'reviewer-1', agentType: 'reviewer', state: 'working', startedAt: 1 }] + } + }, + 'conn-1' + ) + server.ingestRemote( + { paneKey: PANE, payload: { state: 'done', prompt: 'review', agentType: 'claude' } }, + 'conn-1' + ) + expect(server.getStatusSnapshot()[0]?.subagents).toBeUndefined() + }) + it('retains root Codex identity when relay child events omit it', () => { const server = new AgentHookServer() const providerSession = { key: 'session_id' as const, id: 'root-session' } diff --git a/src/main/agent-hooks/server.ts b/src/main/agent-hooks/server.ts index 8d5e57943b6..cb9d4f247f0 100644 --- a/src/main/agent-hooks/server.ts +++ b/src/main/agent-hooks/server.ts @@ -41,7 +41,7 @@ import { claudeRosterHasWorkingSubagent, claudeRosterToSnapshots } from '../../shared/claude-subagent-roster' -import type { AgentHookSource } from '../../shared/agent-hook-relay' +import { restoreShedStatusFields, type AgentHookSource } from '../../shared/agent-hook-relay' import { CLAUDE_STATUSLINE_PATHNAME, parseClaudeStatusLineBody, @@ -1718,6 +1718,8 @@ export class AgentHookServer { providerSession?: unknown providerSessionOnly?: unknown isReplay?: boolean + /** Payload fields the relay dropped to fit an oversized frame; validated below. */ + shedFields?: unknown payload: unknown }, connectionId: string @@ -1796,10 +1798,17 @@ export class AgentHookServer { : undefined const providerSession = normalizeAgentProviderSession(envelope.providerSession) ?? undefined // Why: relay crosses a trust boundary — re-run the canonical normalizer to enforce caps/invariants (returns null on malformed). - const normalizedPayload = normalizeAgentStatusPayload(envelope.payload) - if (!normalizedPayload) { + const validatedPayload = normalizeAgentStatusPayload(envelope.payload) + if (!validatedPayload) { return } + // Why: a field the relay shed to fit the frame is absent, not cleared; restore it from the + // cached payload or a live subagent roster blanks and its pane becomes hibernation-eligible. + const normalizedPayload = restoreShedStatusFields( + validatedPayload, + envelope.shedFields, + this.state.lastStatusByPaneKey.get(paneKey)?.payload + ) if ( envelope.providerSessionOnly === true && !isValidPiProviderSessionOnly(providerSession, normalizedPayload.agentType) diff --git a/src/main/ipc/ssh.test.ts b/src/main/ipc/ssh.test.ts index 05de50fb2d4..7ae2ba903b2 100644 --- a/src/main/ipc/ssh.test.ts +++ b/src/main/ipc/ssh.test.ts @@ -246,6 +246,7 @@ import { RelayVersionMismatchError } from '../ssh/ssh-relay-version-mismatch-err import { SSH_RELAY_CONFIGURE_GRACE_TIME_METHOD, type SshConnectionState, + type SshConnectionStatus, type SshTarget } from '../../shared/ssh-types' import { PTY_CONSUMER_SESSION_PROTOCOL_VERSION } from '../../shared/pty-consumer-session' @@ -1059,6 +1060,132 @@ describe('SSH IPC handlers', () => { } }) + describe('relay loss while the SSH transport is down', () => { + const relayLostTarget: SshTarget = { + id: 'ssh-1', + label: 'Server', + host: 'example.com', + port: 22, + username: 'deploy' + } + const transportState = (status: SshConnectionStatus): SshConnectionState => ({ + targetId: 'ssh-1', + status, + error: null, + reconnectAttempt: 0 + }) + const setTransportStatus = (status: SshConnectionStatus): void => { + mockConnectionManager.getState.mockReturnValue(transportState(status)) + } + const maxRelayDelayMs = relayReconnectDelaysMs.at(-1)! + const connectWithLiveTransport = async (): Promise => { + mockSshStore.getTarget.mockReturnValue(relayLostTarget) + mockConnectionManager.connect.mockResolvedValue({}) + mockConnectionManager.getConnection.mockReturnValue({}) + setTransportStatus('connected') + await handlers.get('ssh:connect')!(null, { targetId: 'ssh-1' }) + mockDeployAndLaunchRelay.mockClear() + } + + it('does not consume attempts or publish the manual-reconnect banner', async () => { + vi.useFakeTimers() + vi.setSystemTime(0) + try { + await connectWithLiveTransport() + setTransportStatus('reconnecting') + getLatestRelayDisposeCallback()('connection_lost') + + // Well past the whole 6-step ladder: a redeploy cannot ride a dead transport, so nothing is spent. + const fullLadderMs = relayReconnectDelaysMs.reduce((sum, delay) => sum + delay, 0) + await vi.advanceTimersByTimeAsync(fullLadderMs + relayLostStabilizedMs) + + expect(mockDeployAndLaunchRelay).not.toHaveBeenCalled() + expect(handlers.get('ssh:getState')!(null, { targetId: 'ssh-1' })).toEqual({ + targetId: 'ssh-1', + status: 'reconnecting', + error: 'Relay channel lost. Reconnecting...', + reconnectAttempt: 0, + providerEpoch: expect.any(String), + connectionGeneration: 2 + }) + + setTransportStatus('connected') + await vi.advanceTimersByTimeAsync(maxRelayDelayMs) + expect(mockDeployAndLaunchRelay).toHaveBeenCalledTimes(1) + expect( + (handlers.get('ssh:getState')!(null, { targetId: 'ssh-1' }) as SshConnectionState).status + ).toBe('connected') + } finally { + vi.useRealTimers() + } + }) + + it('stops retrying once the transport reaches a terminal state', async () => { + vi.useFakeTimers() + vi.setSystemTime(0) + try { + await connectWithLiveTransport() + setTransportStatus('reconnecting') + getLatestRelayDisposeCallback()('connection_lost') + await vi.advanceTimersByTimeAsync(maxRelayDelayMs) + + setTransportStatus('reconnection-failed') + await vi.advanceTimersByTimeAsync(maxRelayDelayMs) + + // The wait loop is gone: only onStateChange's redeploy may revive the relay after this. + setTransportStatus('connected') + await vi.advanceTimersByTimeAsync(maxRelayDelayMs * 4) + expect(mockDeployAndLaunchRelay).not.toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + }) + + it('still reaches the manual-reconnect banner when the transport is healthy', async () => { + vi.useFakeTimers() + vi.setSystemTime(0) + try { + await connectWithLiveTransport() + mockDeployAndLaunchRelay.mockRejectedValue(new Error('relay refused')) + getLatestRelayDisposeCallback()('connection_lost') + for (const delayMs of relayReconnectDelaysMs) { + await vi.advanceTimersByTimeAsync(delayMs) + } + + expect( + (handlers.get('ssh:getState')!(null, { targetId: 'ssh-1' }) as SshConnectionState).error + ).toBe('Relay channel kept dropping. Click Reconnect on the SSH target before retrying.') + } finally { + vi.useRealTimers() + } + }) + + it('restores the full relay budget once the transport reconnects', async () => { + vi.useFakeTimers() + vi.setSystemTime(0) + try { + await connectWithLiveTransport() + mockDeployAndLaunchRelay.mockRejectedValue(new Error('relay refused')) + getLatestRelayDisposeCallback()('connection_lost') + await vi.advanceTimersByTimeAsync(relayReconnectDelaysMs[0]) + + const callbacks = mockConnectionManager.callbacksRef.current as { + onStateChange: (targetId: string, state: SshConnectionState) => void + } + callbacks.onStateChange('ssh-1', transportState('reconnecting')) + callbacks.onStateChange('ssh-1', transportState('connected')) + await vi.advanceTimersByTimeAsync(0) + + // Budget reset: the next loss waits the base delay again instead of the third ladder step. + mockDeployAndLaunchRelay.mockClear() + await vi.advanceTimersByTimeAsync(relayReconnectDelaysMs[0]) + expect(mockDeployAndLaunchRelay).toHaveBeenCalledTimes(1) + } finally { + vi.useRealTimers() + } + }) + }) + it('reuses a fast relay reconnect after the post-ready stabilization window', async () => { vi.useFakeTimers() vi.setSystemTime(0) diff --git a/src/main/ipc/ssh.ts b/src/main/ipc/ssh.ts index 7e1a54026df..7bb79f588d6 100644 --- a/src/main/ipc/ssh.ts +++ b/src/main/ipc/ssh.ts @@ -295,6 +295,13 @@ const RELAY_LOST_BASE_DELAY_MS = 500 const RELAY_LOST_MAX_DELAY_MS = 15_000 // Why: a reconnect whose mux dies within this window was a flap, not a recovery — don't reset the attempt counter. 5s covers provider re-registration + PTY reattach. const RELAY_LOST_STABILIZED_MS = 5_000 +// Why: transport states the SSH ladder never leaves on its own — waiting for a relay redeploy past one of these is an unbounded loop. +const TRANSPORT_TERMINAL_STATUSES = new Set([ + 'disconnected', + 'auth-failed', + 'reconnection-failed', + 'error' +]) function clearRelayLostBackoff(targetId: string): void { const state = relayLostBackoff.get(targetId) @@ -615,6 +622,8 @@ function createSshConnectionCallbacks(): SshConnectionCallbacks { relayStateOverrides.has(targetId) if (shouldReconnectRelay) { + // Why: this branch redeploys the relay itself over a fresh transport, so any pending relay-lost retry is stale — dropping it also gives the new transport generation a full attempt budget. + clearRelayLostBackoff(targetId) // Why: SSH connects before the relay providers rebuild; keep renderer actions gated until SshRelaySession reaches ready again. publishRelayOverride( getCurrentMainWindow, @@ -714,7 +723,13 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { return } rotateSshProviderAuthority(tid) - if (state.attempts >= RELAY_LOST_MAX_ATTEMPTS) { + + // Why: re-deploying the relay rides the SSH transport, so while the transport is itself down no attempt + // can succeed. Waiting at the max delay without consuming the budget keeps a flapping host off the + // manual-reconnect banner, which would tell the user to act on a link that is still auto-recovering. + const transportStatus = connectionManager?.getState(tid)?.status + const transportConnected = transportStatus === 'connected' + if (transportConnected && state.attempts >= RELAY_LOST_MAX_ATTEMPTS) { console.warn( `[ssh] Relay channel for ${tid} kept dying across ${state.attempts} attempts; giving up. User must reconnect manually.` ) @@ -729,6 +744,45 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { ) return } + + const scheduleRelayRedeploy = (delay: number): void => { + state.reconnectTimer = setTimeout(() => { + state.reconnectTimer = null + relayLostBackoff.set(tid, state) + const liveConn = connectionManager?.getConnection(tid) + if (!liveConn || !activeSessions.has(tid)) { + return + } + const status = connectionManager?.getState(tid)?.status + if (status === 'connected') { + void s.reconnect(liveConn, relayGracePeriodForTarget(t)) + return + } + if (status === undefined || TRANSPORT_TERMINAL_STATUSES.has(status)) { + // Why: the transport gave up for good; its own state is what the user acts on, so stop waiting for a redeploy that can never run. + clearRelayLostBackoff(tid) + return + } + scheduleRelayRedeploy(RELAY_LOST_MAX_DELAY_MS) + }, delay) + relayLostBackoff.set(tid, state) + } + + if (!transportConnected) { + publishRelayOverride( + getCurrentMainWindow, + tid, + 'reconnecting', + 'Relay channel lost. Reconnecting...', + state.attempts + ) + scheduleRelayRedeploy(RELAY_LOST_MAX_DELAY_MS) + console.warn( + `[ssh] Relay channel for ${tid} lost while the SSH transport is ${transportStatus ?? 'unknown'}; waiting ${RELAY_LOST_MAX_DELAY_MS}ms without consuming an attempt` + ) + return + } + const delay = Math.min(RELAY_LOST_BASE_DELAY_MS * 2 ** state.attempts, RELAY_LOST_MAX_DELAY_MS) state.attempts += 1 publishRelayOverride( @@ -738,16 +792,7 @@ function configureRelaySessionCallbacks(session: SshRelaySession): void { 'Relay channel lost. Reconnecting...', state.attempts ) - state.reconnectTimer = setTimeout(() => { - state.reconnectTimer = null - relayLostBackoff.set(tid, state) - const liveConn = connectionManager?.getConnection(tid) - if (!liveConn || !activeSessions.has(tid)) { - return - } - void s.reconnect(liveConn, relayGracePeriodForTarget(t)) - }, delay) - relayLostBackoff.set(tid, state) + scheduleRelayRedeploy(delay) console.warn( `[ssh] Relay channel for ${tid} lost; reconnect attempt ${state.attempts}/${RELAY_LOST_MAX_ATTEMPTS} in ${delay}ms` ) diff --git a/src/main/ssh/ssh-connection.test.ts b/src/main/ssh/ssh-connection.test.ts index 9433ece028d..0314d1c3892 100644 --- a/src/main/ssh/ssh-connection.test.ts +++ b/src/main/ssh/ssh-connection.test.ts @@ -12,6 +12,7 @@ let connectErrorMessage = '' let connectErrorCode = '' let destroyErrorMessage = '' let connectSequence: ('ready' | Error)[] = [] +let connectAttempts = 0 let execBehavior: 'callback' | 'pending' = 'callback' let pendingExecCallback: ((err: Error | undefined, channel: unknown) => void) | null = null let sftpBehavior: 'callback' | 'pending' = 'callback' @@ -59,6 +60,7 @@ vi.mock('ssh2', () => { } } connect(config?: unknown) { + connectAttempts += 1 this.lastConnectConfig = config const hostVerifier = (config as { hostVerifier?: (key: Buffer) => boolean } | undefined) ?.hostVerifier @@ -167,7 +169,7 @@ import { writeFileViaSystemSsh } from './ssh-system-fallback' import { getRemoteHostPlatform } from './ssh-remote-platform' -import { CONNECT_TIMEOUT_MS } from './ssh-connection-utils' +import { CONNECT_TIMEOUT_MS, RECONNECT_BACKOFF_MS } from './ssh-connection-utils' import { MIN_SSH_RELAY_GRACE_PERIOD_SECONDS, type SshTarget } from '../../shared/ssh-types' function createTarget(overrides?: Partial): SshTarget { @@ -275,6 +277,7 @@ describe('SshConnection', () => { connectErrorCode = '' destroyErrorMessage = '' connectSequence = [] + connectAttempts = 0 execBehavior = 'callback' pendingExecCallback = null sftpBehavior = 'callback' @@ -530,6 +533,9 @@ describe('SshConnection', () => { await vi.advanceTimersByTimeAsync(200_000) expect(statuses).toContain('reconnection-failed') + // Pin the budget itself: the initial success plus exactly RECONNECT_BACKOFF_MS.length retries. + // Counting a failure twice, or giving up early, would strand a user on a flaky link. + expect(connectAttempts).toBe(1 + RECONNECT_BACKOFF_MS.length) } finally { vi.useRealTimers() } diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts index d5dba94c80b..a5342f9d0cd 100644 --- a/src/main/ssh/ssh-relay-session.ts +++ b/src/main/ssh/ssh-relay-session.ts @@ -1159,6 +1159,7 @@ export class SshRelaySession { isReplay?: unknown providerSession?: unknown providerSessionOnly?: unknown + shedFields?: unknown payload?: unknown } if (typeof envelope.paneKey !== 'string') { @@ -1187,6 +1188,8 @@ export class SshRelaySession { isReplay: envelope.isReplay === true ? true : undefined, providerSession: envelope.providerSession, providerSessionOnly: envelope.providerSessionOnly === true ? true : undefined, + // Why: names the fields the relay dropped to fit the frame; ingestRemote restores them. + shedFields: envelope.shedFields, payload: envelope.payload }, this.targetId diff --git a/src/relay/agent-hook-envelope-publication.test.ts b/src/relay/agent-hook-envelope-publication.test.ts index 2be4d2f0301..3f7339ea96e 100644 --- a/src/relay/agent-hook-envelope-publication.test.ts +++ b/src/relay/agent-hook-envelope-publication.test.ts @@ -469,6 +469,31 @@ describe('publishAgentHookEnvelope redelivery', () => { } }) + it('evicts the longest-idle pane once the pending map is full, keeping the newest', () => { + const primary = makeBoundedClient(16384) + const dispatcher = new RelayDispatcher(primary.write, primary.options) + try { + saturateProducerQueue(dispatcher, primary) + // 65 panes against a 64-pane cap: the first must be evicted, the last must survive. + for (let pane = 0; pane < 65; pane++) { + const envelope = makeEnvelope({}) + envelope.paneKey = `tab-1:4f1b0f4e-0000-4000-8000-${String(pane).padStart(12, '0')}` + publishAgentHookEnvelope(dispatcher, envelope) + } + + primary.blocked = false + primary.drain() + vi.advanceTimersByTime(1_000) + + const redelivered = decodeEnvelopes(primary).map((envelope) => envelope.paneKey) + expect(redelivered).toHaveLength(64) + expect(redelivered).toContain('tab-1:4f1b0f4e-0000-4000-8000-000000000064') + expect(redelivered).not.toContain('tab-1:4f1b0f4e-0000-4000-8000-000000000000') + } finally { + dispatcher.dispose() + } + }) + it('drops pending redeliveries when the dispatcher is disposed', () => { const primary = makeBoundedClient(16384) const dispatcher = new RelayDispatcher(primary.write, primary.options) diff --git a/src/relay/agent-hook-envelope-publication.ts b/src/relay/agent-hook-envelope-publication.ts index 8aa4b4558ef..66f857c82b7 100644 --- a/src/relay/agent-hook-envelope-publication.ts +++ b/src/relay/agent-hook-envelope-publication.ts @@ -1,5 +1,6 @@ import { AGENT_HOOK_NOTIFICATION_METHOD, + AGENT_HOOK_SHED_FIELDS_KEY, type AgentHookRelayEnvelope } from '../shared/agent-hook-relay' import type { RelayDispatcher } from './dispatcher' @@ -9,13 +10,10 @@ import type { RelayDispatcher } from './dispatcher' // Why: interactivePrompt is last on purpose — a blocking question is load-bearing (without its card // the pane sits on `waiting` with no way to answer from web or mobile). The roster is shed earlier // only because it is cheaper to rebuild, NOT because it is cosmetic: a missing roster blanks live -// subagent rows and unblocks pane hibernation, which is why shed fields are named on the wire. +// subagent rows and unblocks pane hibernation, which is why shed fields are named on the wire — +// `restoreShedStatusFields` re-attaches the restorable ones from Orca's cached payload. const SHED_ORDER = ['lastAssistantMessage', 'subagents', 'interactivePrompt'] as const -/** Wire marker listing the fields dropped to fit the frame, so a consumer can tell "shed" from - * "genuinely absent" and keep its current value instead of overwriting it with a gap. */ -const SHED_FIELDS_KEY = 'shedFields' - // Why: these envelopes are fire-and-forget state snapshots — no ack, no producer-side retry — and // the only other delivery path is the reattach replay. A queue-full drop would otherwise leave the // pane on a stale spinner until the SSH link cycles, so a rejected snapshot is retried briefly. @@ -46,7 +44,9 @@ function toNotificationParams( shedFields: readonly string[] ): Record { const params = envelope as unknown as Record - return shedFields.length === 0 ? params : { ...params, [SHED_FIELDS_KEY]: [...shedFields] } + return shedFields.length === 0 + ? params + : { ...params, [AGENT_HOOK_SHED_FIELDS_KEY]: [...shedFields] } } /** Returns the ids of the clients that rejected the frame; a rejected client is never written to. */ diff --git a/src/relay/dispatcher-capacity-degradation.test.ts b/src/relay/dispatcher-capacity-degradation.test.ts index 5cc96ff88fe..18fc9b1371f 100644 --- a/src/relay/dispatcher-capacity-degradation.test.ts +++ b/src/relay/dispatcher-capacity-degradation.test.ts @@ -4,8 +4,9 @@ import { type RelayClientSinkOptions, type SinkWriteSettlement } from './dispatcher' +import { LEGACY_CLIENT_RETAINED_BYTES_LOW } from './legacy-relay-publication-ledger' import type * as ProtocolModule from './protocol' -import { encodeJsonRpcFrame } from './protocol' +import { encodeJsonRpcFrame, RelayErrorCode } from './protocol' // Counts every frame encode (including the estimate-only ones) so a redundant re-encode is observable. const encodeCalls = vi.hoisted(() => ({ count: 0 })) @@ -338,7 +339,7 @@ describe('RelayDispatcher bounded-capacity degradation', () => { error: { code: number; message: string } } expect(response.id).toBe(77) - expect(response.error.code).toBe(-32010) + expect(response.error.code).toBe(RelayErrorCode.ResponseOverCapacity) expect(response.error.message).toBe('Relay response exceeded the bounded transport capacity') } finally { bounded.dispose() @@ -369,6 +370,39 @@ describe('RelayDispatcher bounded-capacity degradation', () => { } }) + it('producerRetentionBelowLowWater reports the reserve of one client, not the dispatcher', () => { + const stalled = makeSaturatingClient(65536) + const healthy = makeBoundedClient(65536) + const bounded = new RelayDispatcher(stalled.write, stalled.options) + try { + const stalledId = bounded.activeClientIds()[0] + const healthyId = bounded.attachClient(healthy.write, healthy.options) + let parked = 0 + while ( + parked <= LEGACY_CLIENT_RETAINED_BYTES_LOW && + bounded.publishProducerNotification(stalledId, 'pty.data', { + paneId: 'pane', + data: 'x'.repeat(40_000) + }) + ) { + parked += 40_000 + } + expect(parked).toBeGreaterThan(LEGACY_CLIENT_RETAINED_BYTES_LOW) + + expect(bounded.producerRetentionBelowLowWater(stalledId)).toBe(false) + expect(bounded.producerRetentionBelowLowWater(healthyId)).toBe(true) + // The dispatcher-wide signal is the one that lets a stalled peer speak for a healthy client. + expect(bounded.legacyRetentionBelowLowWater).toBe(false) + + // A client that cannot be written to has no headroom at all. + expect(bounded.producerRetentionBelowLowWater(999)).toBe(false) + bounded.detachClient(healthyId) + expect(bounded.producerRetentionBelowLowWater(healthyId)).toBe(false) + } finally { + bounded.dispose() + } + }) + it('sendResponse leaves an already-failed response rejected without closing', async () => { const primary = makeBoundedClient(65536) const bounded = new RelayDispatcher(primary.write, primary.options) diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index a51869f9d8d..d48b5daffe1 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -6,6 +6,7 @@ import { encodeKeepAliveFrame, parseJsonRpcMessage, KEEPALIVE_SEND_MS, + RelayErrorCode, type DecodedFrame, type JsonRpcRequest, type JsonRpcNotification, @@ -192,6 +193,19 @@ export class RelayDispatcher { return this.publicationLedger.belowLowWater(this.activeClientKeys()) } + /** + * Same reserve as legacyRetentionBelowLowWater, but scoped to one client: a paced bulk producer + * gated on the dispatcher-wide signal stops for a peer's stall and degrades a healthy link. + * The relay-wide aggregate still counts — that ceiling is shared by every client. + */ + producerRetentionBelowLowWater(clientId: number): boolean { + const client = this.clients.get(clientId) + if (!client || client.closed) { + return false + } + return this.publicationLedger.belowLowWater([this.clientKey(client)]) + } + writePrimaryBytes(data: Buffer, lane: 'control' | 'ordinary' = 'control'): boolean { if (this.disposed || this.primaryClient.closed) { return false @@ -468,7 +482,10 @@ export class RelayDispatcher { publishProducerNotification( clientId: number, method: string, - params?: Record + params?: Record, + // Why: a caller that recovers from rejection itself (the watcher emitter re-sends the batch in + // chunks) would otherwise log "Dropped" for a frame it goes on to deliver in full. + options?: { logDrop?: boolean } ): boolean { if (this.disposed) { return false @@ -487,7 +504,9 @@ export class RelayDispatcher { return true } // Why: same diagnostics as notify() — a producer that drops here must not do so silently. - this.logDroppedProducerNotification(client, method, frameBytes) + if (options?.logDrop !== false) { + this.logDroppedProducerNotification(client, method, frameBytes) + } return false } @@ -910,7 +929,10 @@ export class RelayDispatcher { { jsonrpc: '2.0', id, - error: { code: -32010, message: RESPONSE_OVER_CAPACITY_MESSAGE } + error: { + code: RelayErrorCode.ResponseOverCapacity, + message: RESPONSE_OVER_CAPACITY_MESSAGE + } }, 'control', // Why: writing the substitute is not delivering the result — a settlement fence must never read diff --git a/src/relay/fs-handler-file-read.ts b/src/relay/fs-handler-file-read.ts index 452bf10c8f0..14672457cc6 100644 --- a/src/relay/fs-handler-file-read.ts +++ b/src/relay/fs-handler-file-read.ts @@ -110,7 +110,7 @@ export async function readRelayFileStreamMetadata( // Why: reserved before the fd opens so a refusal costs nothing, and released only // once the terminal frame settles — see reserveTerminalFrameSlot. - const releaseTerminalFrameSlot = reserveTerminalFrameSlot(registry) + const releaseTerminalFrameSlot = reserveTerminalFrameSlot(registry, context.clientId) let handle: FileHandle | undefined let streamId: number try { @@ -160,22 +160,36 @@ export async function readRelayFileStreamMetadata( * terminal frames at MAX_CONCURRENT_STREAMS, far below the killing threshold; the * overflow now costs one refused read (TooManyStreams, which clients already handle) * instead of the whole connection. + * + * Counted per client, because the control queue this protects is per client: one peer whose + * socket stopped draining must not refuse reads for every other peer on the same relay. */ -const pendingTerminalFrames = new WeakMap() +const pendingTerminalFramesByClient = new WeakMap>() -function reserveTerminalFrameSlot(registry: RelayStreamRegistry): () => void { - const pending = pendingTerminalFrames.get(registry) ?? 0 +function reserveTerminalFrameSlot(registry: RelayStreamRegistry, clientId: number): () => void { + let byClient = pendingTerminalFramesByClient.get(registry) + if (!byClient) { + byClient = new Map() + pendingTerminalFramesByClient.set(registry, byClient) + } + const pending = byClient.get(clientId) ?? 0 if (pending >= MAX_CONCURRENT_STREAMS) { throw new TooManyStreamsError() } - pendingTerminalFrames.set(registry, pending + 1) + byClient.set(clientId, pending + 1) let released = false return () => { if (released) { return } released = true - pendingTerminalFrames.set(registry, (pendingTerminalFrames.get(registry) ?? 1) - 1) + const remaining = (byClient.get(clientId) ?? 1) - 1 + // Drop the entry at zero so a long-lived registry cannot accumulate one per detached client. + if (remaining <= 0) { + byClient.delete(clientId) + return + } + byClient.set(clientId, remaining) } } diff --git a/src/relay/fs-handler-stream.test.ts b/src/relay/fs-handler-stream.test.ts index d387879327e..c770fe72f23 100644 --- a/src/relay/fs-handler-stream.test.ts +++ b/src/relay/fs-handler-stream.test.ts @@ -478,10 +478,18 @@ describe('FsHandler readFileStream', () => { await flush(10) expect(terminalFrames()).toHaveLength(MAX_CONCURRENT_STREAMS) + // The budget is per client, so a second peer on the same relay still reads. + await dispatcher.callRequest( + 'fs.readFileStream', + { filePath }, + { clientId: 9, isStale: () => false } + ) + await waitFor(() => terminalFrames().length === MAX_CONCURRENT_STREAMS + 1) + // Draining the sink settles the queued frames and hands the budget back. dispatcher.settleHeldControlFrames() await dispatcher.callRequest('fs.readFileStream', { filePath }, context) - await waitFor(() => terminalFrames().length === MAX_CONCURRENT_STREAMS + 1) + await waitFor(() => terminalFrames().length === MAX_CONCURRENT_STREAMS + 2) }) it('rejects the 17th concurrent stream with TooManyStreams', async () => { diff --git a/src/relay/protocol.ts b/src/relay/protocol.ts index 0fbd85cc8fb..642f694c526 100644 --- a/src/relay/protocol.ts +++ b/src/relay/protocol.ts @@ -102,7 +102,10 @@ export type GitResponseStreamMarker = { export const RelayErrorCode = { TooManyStreams: -33006, - StreamProtocolError: -33007 + StreamProtocolError: -33007, + /** Substituted for a result too large for the sink's frame capacity; the request fails + * instead of the whole link, so a caller can retry with a narrower scope. */ + ResponseOverCapacity: -32010 } as const export type JsonRpcRequest = { diff --git a/src/relay/relay-oversized-notification-survival.test.ts b/src/relay/relay-oversized-notification-survival.test.ts index 7b1189bea41..41d3dd833e5 100644 --- a/src/relay/relay-oversized-notification-survival.test.ts +++ b/src/relay/relay-oversized-notification-survival.test.ts @@ -101,7 +101,7 @@ function createDispatcher(hwm: number, onClose: () => void, sink: Buffer[]): Rel } describe('relay oversized notification survival', () => { - // Without this the pinned byte counts below could measure a payload the emitter never sends. + // Without this the byte measurements below could describe a payload the emitter never sends. it('measures the params emitRelayWatcherEvents actually publishes', () => { const sink: Buffer[] = [] const dispatcher = createDispatcher(NODE22_HWM, () => {}, sink) @@ -134,20 +134,19 @@ describe('relay oversized notification survival', () => { return lo } + // Capacity is a fixed fraction of the sink's high-water mark, so these are exact. expect(capacityFor(NODE22_HWM)).toBe(49152) expect(capacityFor(NODE21_HWM)).toBe(12288) - expect(hookBytes).toBe(29894) - expect(fullBatchBytes).toBe(465074) - expect(trip(capacityFor(NODE22_HWM))).toBe(527) - expect(trip(capacityFor(NODE21_HWM))).toBe(131) - // agent.hook: fits a modern remote, over-cap on an older one. + // agent.hook: fits a modern remote, over-cap on an older one — which is why the remote's + // Node major, not Orca's, decides whether a maximal envelope has to shed. expect(hookBytes).toBeLessThan(capacityFor(NODE22_HWM)) expect(hookBytes).toBeGreaterThan(capacityFor(NODE21_HWM)) // fs.changed: the batch cap is orders of magnitude above what one frame can carry, on BOTH tiers. expect(fullBatchBytes).toBeGreaterThan(capacityFor(NODE22_HWM)) - expect(trip(capacityFor(NODE22_HWM))).toBeLessThan(MAX_BATCHED_WATCHER_EVENTS) + expect(trip(capacityFor(NODE22_HWM)) * 5).toBeLessThan(MAX_BATCHED_WATCHER_EVENTS) + expect(trip(capacityFor(NODE21_HWM))).toBeLessThan(trip(capacityFor(NODE22_HWM))) }) it('a 5000-event fs.changed no longer closes the client on a Node>=22-sized sink', () => { diff --git a/src/relay/relay-watcher-event-emitter.test.ts b/src/relay/relay-watcher-event-emitter.test.ts index 8b82eefcfeb..3f87d57e266 100644 --- a/src/relay/relay-watcher-event-emitter.test.ts +++ b/src/relay/relay-watcher-event-emitter.test.ts @@ -1,6 +1,6 @@ import { readFileSync } from 'node:fs' import { fileURLToPath } from 'node:url' -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import type { WatcherProcessEvent } from '../main/ipc/parcel-watcher-process-protocol' import { RelayDispatcher } from './dispatcher' import { DEFAULT_PRODUCER_QUEUE_MAX_BYTES } from './dispatcher-writer-admission' @@ -113,10 +113,40 @@ function parkProducerBytes(dispatcher: RelayDispatcher, bytes: number): number { return parked } +// Parks PTY frames in ONE client's producer queue, leaving every peer's retention untouched. +function parkProducerBytesForClient( + dispatcher: RelayDispatcher, + clientId: number, + bytes: number +): number { + const chunk = 'x'.repeat(40_000) + let parked = 0 + while ( + parked < bytes && + dispatcher.publishProducerNotification(clientId, 'pty.data', { paneId: 'pane', data: chunk }) + ) { + parked += chunk.length + } + expect(parked).toBeGreaterThanOrEqual(bytes) + return parked +} + function isOverflowFrame(frame: Buffer): boolean { return frameEvents(frame).some((event) => event.kind === 'overflow') } +function watcherFrames(frames: readonly Buffer[]): Buffer[] { + return frames.filter((frame) => frameMethod(frame) === 'fs.changed') +} + +function deliveredEvents(events: readonly WatcherProcessEvent[]): WatcherFrameEvent[] { + return events.map((event) => ({ + kind: event.type, + absolutePath: event.path, + ...(event.isDirectory === undefined ? {} : { isDirectory: event.isDirectory }) + })) +} + describe('relay watcher writer admission', () => { it('keeps watcher batches on the producer lane while overflow markers ride the control lane', () => { const sink = createRecordingSink() @@ -334,6 +364,27 @@ describe('relay watcher batch chunking', () => { } }) + it('does not report a drop for a batch it goes on to deliver in full', () => { + const sink = createRecordingSink(16384) + const dispatcher = new RelayDispatcher(sink.write, sink.options) + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) + + try { + const events = watcherBatch(5000) + emitRelayWatcherEvents(dispatcher, '/workspace', false, events) + + expect(sink.frames.length).toBeGreaterThan(1) + expect(sink.frames.flatMap(frameEvents)).toHaveLength(events.length) + // A "Dropped fs.changed" line here would both lie and burn the one-per-generation + // suppression slot a genuine over-capacity drop needs. + const lines = stderr.mock.calls.map(([line]) => String(line)) + expect(lines.filter((line) => line.includes('fs.changed'))).toEqual([]) + } finally { + stderr.mockRestore() + dispatcher.dispose() + } + }) + it('sizes chunks near the byte minimum instead of overshooting it', () => { const sink = createRecordingSink(16384) const capacity = 12288 @@ -381,6 +432,63 @@ describe('relay watcher batch chunking', () => { } }) + it('delivers a whole flood to a healthy client while a second client sits past its reserve', () => { + const stalled = createRecordingSink(65536) + const healthy = createRecordingSink(65536) + stalled.blockNextWrite() + const dispatcher = new RelayDispatcher(stalled.write, stalled.options) + const events = watcherBatch(5000) + + try { + dispatcher.attachClient(healthy.write, healthy.options) + const stalledId = dispatcher.activeClientIds()[0] + parkProducerBytesForClient(dispatcher, stalledId, LEGACY_CLIENT_RETAINED_BYTES_LOW) + + emitRelayWatcherEvents(dispatcher, '/workspace', false, events) + stalled.drainWaiters.shift()?.() + + // A resync forces a readDir per directory, so an unrelated peer's stall must never trigger one here. + expect(watcherFrames(healthy.frames).some(isOverflowFrame)).toBe(false) + expect(watcherFrames(healthy.frames).flatMap(frameEvents)).toEqual(deliveredEvents(events)) + expect(watcherFrames(stalled.frames).map(isOverflowFrame)).toEqual([true]) + expect(healthy.closes()).toBe(0) + expect(stalled.closes()).toBe(0) + } finally { + dispatcher.dispose() + } + }) + + it('stops the walk for the congested client while a healthy peer takes the whole flood', () => { + const healthy = createRecordingSink(65536) + const stalled = createRecordingSink(65536) + stalled.blockNextWrite() + const dispatcher = new RelayDispatcher(healthy.write, healthy.options) + const events = watcherBatch(5000) + + try { + // Reversed attach order from the peer-stall case: the gate must read the client being written to. + const stalledId = dispatcher.attachClient(stalled.write, stalled.options) + parkProducerBytesForClient(dispatcher, stalledId, LEGACY_CLIENT_RETAINED_BYTES_LOW) + + emitRelayWatcherEvents(dispatcher, '/workspace', false, events) + stalled.drainWaiters.shift()?.() + + const stalledWatcher = watcherFrames(stalled.frames) + expect(stalledWatcher.map(isOverflowFrame)).toEqual([true]) + const injected = stalledWatcher + .filter((frame) => !isOverflowFrame(frame)) + .reduce((total, frame) => total + frame.length, 0) + // Nothing injected: the congested client keeps its whole reserve for interactive PTY frames. + expect(injected).toBe(0) + expect(watcherFrames(healthy.frames).some(isOverflowFrame)).toBe(false) + expect(watcherFrames(healthy.frames).flatMap(frameEvents)).toEqual(deliveredEvents(events)) + expect(healthy.closes()).toBe(0) + expect(stalled.closes()).toBe(0) + } finally { + dispatcher.dispose() + } + }) + it('still delivers a flood larger than the reserve while the sink keeps draining', () => { const sink = createRecordingSink(16384) const dispatcher = new RelayDispatcher(sink.write, sink.options) diff --git a/src/relay/relay-watcher-event-emitter.ts b/src/relay/relay-watcher-event-emitter.ts index 3805e8b4872..40b31ce6b91 100644 --- a/src/relay/relay-watcher-event-emitter.ts +++ b/src/relay/relay-watcher-event-emitter.ts @@ -71,7 +71,18 @@ function publishWatcherBatchToClient( dispatcher.producerEnvelopeBudget('fs.changed', { events }, clientId) // Fast path: publish the whole batch first — two encodes, the same cost as an unchunked emit. - if (publish(mapped)) { + // logDrop:false because rejection here is a measurement, not an outcome: the batch is re-sent in + // chunks below, so logging it would report a drop for events that all arrive. + if ( + dispatcher.publishProducerNotification( + clientId, + 'fs.changed', + { events: mapped }, + { + logDrop: false + } + ) + ) { return } @@ -92,7 +103,9 @@ function publishWatcherBatchToClient( // Why: the retention ledger covers every producer publication despite its legacy name, and admission // is lane-agnostic: chunks queued past its low-water reserve (half the 2 MB queue) starve interactive // PTY frames until pty-handler pauses every remote pane. A resync costs the user far less. - if (!dispatcher.legacyRetentionBelowLowWater) { + // Per client, never dispatcher-wide: one stalled peer must not cost a healthy client a resync, + // which forces a readDir per directory in its file tree. + if (!dispatcher.producerRetentionBelowLowWater(clientId)) { emitWatcherOverflowToClient(dispatcher, clientId, rootPath) return } diff --git a/src/shared/agent-hook-relay.test.ts b/src/shared/agent-hook-relay.test.ts index 27aec405236..a22655474a5 100644 --- a/src/shared/agent-hook-relay.test.ts +++ b/src/shared/agent-hook-relay.test.ts @@ -3,10 +3,13 @@ import { AGENT_HOOK_INSTALL_PLUGINS_METHOD, AGENT_HOOK_NOTIFICATION_METHOD, AGENT_HOOK_REQUEST_REPLAY_METHOD, + AGENT_HOOK_SHED_FIELDS_KEY, ORCA_FEATURE_REMOTE_AGENT_HOOKS_ENV, isRemoteAgentHooksEnabled, + restoreShedStatusFields, type AgentHookRelayEnvelope } from './agent-hook-relay' +import type { ParsedAgentStatusPayload } from './agent-status-types' describe('agent-hook-relay wire shape', () => { it('encodes/decodes through JSON without losing fields', () => { @@ -55,3 +58,56 @@ describe('isRemoteAgentHooksEnabled', () => { expect(isRemoteAgentHooksEnabled({ [ORCA_FEATURE_REMOTE_AGENT_HOOKS_ENV]: 'true' })).toBe(true) }) }) + +describe('restoreShedStatusFields', () => { + const roster = [{ id: 'child-1', agentType: 'reviewer', state: 'working' as const, startedAt: 1 }] + const cached: ParsedAgentStatusPayload = { + state: 'working', + prompt: 'p', + agentType: 'claude', + subagents: roster, + lastAssistantMessage: 'cached message', + interactivePrompt: '{"questions":["old"]}' + } + const shed: ParsedAgentStatusPayload = { state: 'done', prompt: 'p', agentType: 'claude' } + + it('restores a roster the relay shed, so a done pane is not falsely hibernation-eligible', () => { + const restored = restoreShedStatusFields(shed, ['subagents'], cached) + expect(restored.subagents).toEqual(roster) + expect(restored.state).toBe('done') + // Only the named field comes back. + expect(restored.lastAssistantMessage).toBeUndefined() + }) + + it('restores the assistant message and reads the producer key', () => { + expect(AGENT_HOOK_SHED_FIELDS_KEY).toBe('shedFields') + const restored = restoreShedStatusFields(shed, ['lastAssistantMessage', 'subagents'], cached) + expect(restored.lastAssistantMessage).toBe('cached message') + expect(restored.subagents).toEqual(roster) + }) + + it('never restores interactivePrompt — a stale answerable card is worse than none', () => { + const restored = restoreShedStatusFields(shed, ['interactivePrompt'], cached) + expect(restored.interactivePrompt).toBeUndefined() + expect(restored).toBe(shed) + }) + + it('leaves a genuinely cleared field cleared when the relay shed nothing', () => { + expect(restoreShedStatusFields(shed, undefined, cached)).toBe(shed) + expect(restoreShedStatusFields(shed, [], cached)).toBe(shed) + // A hostile/garbled marker must not throw or restore. + expect(restoreShedStatusFields(shed, 'subagents', cached)).toBe(shed) + expect(restoreShedStatusFields(shed, [{ toString: () => 'subagents' }], cached)).toBe(shed) + }) + + it('does not overwrite a value the envelope still carries', () => { + const fresh: ParsedAgentStatusPayload = { ...shed, lastAssistantMessage: 'fresh' } + expect( + restoreShedStatusFields(fresh, ['lastAssistantMessage'], cached).lastAssistantMessage + ).toBe('fresh') + }) + + it('is a no-op with no cached payload for the pane', () => { + expect(restoreShedStatusFields(shed, ['subagents'], undefined)).toBe(shed) + }) +}) diff --git a/src/shared/agent-hook-relay.ts b/src/shared/agent-hook-relay.ts index 71e7953170a..a5f912023f3 100644 --- a/src/shared/agent-hook-relay.ts +++ b/src/shared/agent-hook-relay.ts @@ -99,6 +99,44 @@ export type AgentHookRelayEnvelope = { /** JSON-RPC notification method name carried over the relay control channel. */ export const AGENT_HOOK_NOTIFICATION_METHOD = 'agent.hook' as const +/** Names the optional payload fields the relay dropped to fit an oversized frame + * (see `src/relay/agent-hook-envelope-publication.ts`), so `ingestRemote` can tell + * "shed in transit" from "the agent cleared it". */ +export const AGENT_HOOK_SHED_FIELDS_KEY = 'shedFields' as const + +/** + * Re-attaches shed fields from Orca's cached payload for this pane, so a transport-level + * truncation cannot read as a cleared field — an absent `subagents` blanks live child rows and + * unblocks hibernation for a pane whose teammates are still running. + * + * `interactivePrompt` is deliberately not restored: the cached card belongs to an earlier + * question, and an answerable card carrying the wrong question is worse than no card. + */ +export function restoreShedStatusFields( + payload: ParsedAgentStatusPayload, + shedFields: unknown, + previous: ParsedAgentStatusPayload | undefined +): ParsedAgentStatusPayload { + if (!previous || !Array.isArray(shedFields) || shedFields.length === 0) { + return payload + } + const shed = new Set(shedFields.filter((field): field is string => typeof field === 'string')) + const subagents = + shed.has('subagents') && payload.subagents === undefined ? previous.subagents : undefined + const lastAssistantMessage = + shed.has('lastAssistantMessage') && payload.lastAssistantMessage === undefined + ? previous.lastAssistantMessage + : undefined + if (subagents === undefined && lastAssistantMessage === undefined) { + return payload + } + return { + ...payload, + ...(subagents === undefined ? {} : { subagents }), + ...(lastAssistantMessage === undefined ? {} : { lastAssistantMessage }) + } +} + /** JSON-RPC request method Orca issues after `--connect` reattach to ask the * relay to replay its per-paneKey last-payload cache. See §5 Path 3 of the * design doc for the race that ruled out push-on-`setWrite`. */