fix(relay,ssh): restore relay-shed status fields and scope backpressure per client

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.
This commit is contained in:
Jinjing
2026-08-01 01:01:11 -07:00
parent cac09e8a7e
commit d99888fef6
18 changed files with 599 additions and 44 deletions
+45
View File
@@ -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' }
+12 -3
View File
@@ -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)
+127
View File
@@ -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<void> => {
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)
+56 -11
View File
@@ -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<SshConnectionStatus>([
'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`
)
+7 -1
View File
@@ -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>): 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()
}
+3
View File
@@ -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
@@ -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)
+6 -6
View File
@@ -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<string, unknown> {
const params = envelope as unknown as Record<string, unknown>
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. */
@@ -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)
+25 -3
View File
@@ -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<string, unknown>
params?: Record<string, unknown>,
// 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
+20 -6
View File
@@ -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<RelayStreamRegistry, number>()
const pendingTerminalFramesByClient = new WeakMap<RelayStreamRegistry, Map<number, number>>()
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)
}
}
+9 -1
View File
@@ -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 () => {
+4 -1
View File
@@ -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 = {
@@ -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', () => {
+109 -1
View File
@@ -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)
+15 -2
View File
@@ -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
}
+56
View File
@@ -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)
})
})
+38
View File
@@ -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`. */