From c4ae4b7120d1737b2bd7e3136e7ed17fa97a0f3c Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Sat, 1 Aug 2026 02:17:39 -0700 Subject: [PATCH] fix(relay,ssh): restore relay-shed status fields and scope backpressure - Oversized relay responses now fail their request instead of closing the connection, preventing one frame from killing every pane on the host - Restore subagent state for correct hibernation; don't resurrect stale prose across turns - Account for relay re-establishment and PTY reattach time in SSH flap delay caps - Only log drops of final unsendable envelopes, not temporary rejections during measurement probes - Fix watcher overflow marker release race when notification admission rejects without settlement; use precise byte counting for event batching --- src/main/ssh/ssh-reconnect-ladder.test.ts | 12 +++- src/main/ssh/ssh-reconnect-ladder.ts | 9 ++- .../agent-hook-envelope-publication.test.ts | 20 ++++++ src/relay/agent-hook-envelope-publication.ts | 26 +++++-- src/relay/agent-hook-integration.test.ts | 63 ++++++++++++++--- .../dispatcher-capacity-degradation.test.ts | 11 ++- src/relay/dispatcher.ts | 6 +- src/relay/protocol.ts | 2 +- src/relay/relay-watcher-event-emitter.test.ts | 30 +++++++++ src/relay/relay-watcher-event-emitter.ts | 67 ++++++++++++------- src/shared/agent-hook-relay.test.ts | 14 ++-- src/shared/agent-hook-relay.ts | 13 ++-- 12 files changed, 206 insertions(+), 67 deletions(-) diff --git a/src/main/ssh/ssh-reconnect-ladder.test.ts b/src/main/ssh/ssh-reconnect-ladder.test.ts index 994ea698fb1..c43d4e9e6e9 100644 --- a/src/main/ssh/ssh-reconnect-ladder.test.ts +++ b/src/main/ssh/ssh-reconnect-ladder.test.ts @@ -1,7 +1,12 @@ import { describe, expect, it } from 'vitest' import { MIN_SSH_RELAY_GRACE_PERIOD_SECONDS } from '../../shared/ssh-types' import { CONNECT_TIMEOUT_MS, RECONNECT_BACKOFF_MS } from './ssh-connection-utils' -import { FLAP_DELAY_CAP_MS, SshReconnectLadder, STABLE_CONNECTION_MS } from './ssh-reconnect-ladder' +import { + FLAP_DELAY_CAP_MS, + RELAY_REESTABLISH_BUDGET_MS, + SshReconnectLadder, + STABLE_CONNECTION_MS +} from './ssh-reconnect-ladder' describe('SshReconnectLadder', () => { it('climbs the table across consecutive post-handshake drops', () => { @@ -37,7 +42,7 @@ describe('SshReconnectLadder', () => { } }) - it('caps every flap delay so a full handshake still beats the shortest relay grace', () => { + it('caps every flap delay so handshake and relay re-establishment beat grace', () => { const ladder = new SshReconnectLadder() let now = 0 @@ -47,12 +52,13 @@ describe('SshReconnectLadder', () => { const decision = ladder.next(now) expect(decision.kind).toBe('retry') if (decision.kind === 'retry') { - expect(decision.delayMs + CONNECT_TIMEOUT_MS).toBeLessThan( + expect(decision.delayMs + CONNECT_TIMEOUT_MS + RELAY_REESTABLISH_BUDGET_MS).toBeLessThan( MIN_SSH_RELAY_GRACE_PERIOD_SECONDS * 1000 ) now += decision.delayMs } } + expect(FLAP_DELAY_CAP_MS).toBe(5_000) }) it('keeps the uncapped table for a host whose handshakes fail', () => { diff --git a/src/main/ssh/ssh-reconnect-ladder.ts b/src/main/ssh/ssh-reconnect-ladder.ts index 00c937bb0eb..bf03154be0e 100644 --- a/src/main/ssh/ssh-reconnect-ladder.ts +++ b/src/main/ssh/ssh-reconnect-ladder.ts @@ -3,15 +3,18 @@ import { CONNECT_TIMEOUT_MS, RECONNECT_BACKOFF_MS } from './ssh-connection-utils // A connection that held this long is treated as healthy, so the next drop restarts the delay ladder. export const STABLE_CONNECTION_MS = 60_000 +// Existing-relay attach and one bounded PTY reattach attempt each allow 10s. +export const RELAY_REESTABLISH_BUDGET_MS = 20_000 /** * Why: a flap leaves the remote relay in grace, and the shortest grace a target can configure is * MIN_SSH_RELAY_GRACE_PERIOD_SECONDS. Retrying later than this cap would let even a full-length - * handshake (CONNECT_TIMEOUT_MS) finish after grace expires, so the relay would shut down and take - * every remote PTY with it. Raising RECONNECT_BACKOFF_MS past this bound therefore changes nothing. + * handshake plus relay/mux re-establishment finish after grace expires, so the relay would shut + * down and take every remote PTY with it. Raising the backoff past this bound changes nothing. */ export const FLAP_DELAY_CAP_MS = (() => { - const budgetMs = MIN_SSH_RELAY_GRACE_PERIOD_SECONDS * 1000 - CONNECT_TIMEOUT_MS + const budgetMs = + MIN_SSH_RELAY_GRACE_PERIOD_SECONDS * 1000 - CONNECT_TIMEOUT_MS - RELAY_REESTABLISH_BUDGET_MS const fitting = RECONNECT_BACKOFF_MS.filter((delayMs) => delayMs < budgetMs) return fitting.length > 0 ? Math.max(...fitting) : RECONNECT_BACKOFF_MS[0] })() diff --git a/src/relay/agent-hook-envelope-publication.test.ts b/src/relay/agent-hook-envelope-publication.test.ts index 3f7339ea96e..de6d7211278 100644 --- a/src/relay/agent-hook-envelope-publication.test.ts +++ b/src/relay/agent-hook-envelope-publication.test.ts @@ -168,6 +168,26 @@ describe('publishAgentHookEnvelope', () => { } }) + it('logs only when the final single-sink envelope is actually dropped', () => { + const primary = makeBoundedClient(16384) + const dispatcher = new RelayDispatcher(primary.write, primary.options) + const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true) + try { + publishAgentHookEnvelope(dispatcher, makeEnvelope({ lastAssistantMessage: 20_000 })) + expect(decodeEnvelopes(primary)).toHaveLength(1) + expect(stderr).not.toHaveBeenCalled() + + const unsendable = makeEnvelope({}) + unsendable.payload.prompt = 'p'.repeat(40_000) + publishAgentHookEnvelope(dispatcher, unsendable) + expect(stderr).toHaveBeenCalledTimes(1) + expect(String(stderr.mock.calls[0][0])).toContain('Dropped agent.hook') + } finally { + stderr.mockRestore() + dispatcher.dispose() + } + }) + it('sheds subagents next when dropping the assistant message is not enough', () => { 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 66f857c82b7..ecbad32af73 100644 --- a/src/relay/agent-hook-envelope-publication.ts +++ b/src/relay/agent-hook-envelope-publication.ts @@ -53,11 +53,19 @@ function toNotificationParams( function publishToClients( dispatcher: RelayDispatcher, params: Record, - clientIds: readonly number[] + clientIds: readonly number[], + logDrop?: boolean ): number[] { const rejected: number[] = [] for (const clientId of clientIds) { - if (!dispatcher.publishProducerNotification(clientId, AGENT_HOOK_NOTIFICATION_METHOD, params)) { + if ( + !dispatcher.publishProducerNotification( + clientId, + AGENT_HOOK_NOTIFICATION_METHOD, + params, + logDrop === undefined ? undefined : { logDrop } + ) + ) { rejected.push(clientId) } } @@ -84,8 +92,17 @@ export function publishAgentHookEnvelope( let step = 0 for (;;) { const params = toNotificationParams(candidate, shedFields) + while (step < SHED_ORDER.length && candidate.payload[SHED_ORDER[step]] === undefined) { + step += 1 + } if (!measureBeforePublish || fitsProducerFrame(dispatcher, params)) { - const rejected = publishToClients(dispatcher, params, clientIds) + // A rejected intermediate probe is not a drop if a smaller envelope is delivered next. + const rejected = publishToClients( + dispatcher, + params, + clientIds, + measureBeforePublish ? undefined : step >= SHED_ORDER.length + ) if (rejected.length === 0) { clearPendingEnvelope(dispatcher, envelope.paneKey) return @@ -96,9 +113,6 @@ export function publishAgentHookEnvelope( return } } - while (step < SHED_ORDER.length && candidate.payload[SHED_ORDER[step]] === undefined) { - step += 1 - } if (step >= SHED_ORDER.length) { // Nothing left to shed and it still does not fit: waiting cannot make it sendable, and this // snapshot supersedes any pending one, which is now stale. diff --git a/src/relay/agent-hook-integration.test.ts b/src/relay/agent-hook-integration.test.ts index 9e3df40a930..3b941e6baee 100644 --- a/src/relay/agent-hook-integration.test.ts +++ b/src/relay/agent-hook-integration.test.ts @@ -26,7 +26,9 @@ import { AGENT_HOOK_NOTIFICATION_METHOD, AGENT_HOOK_REQUEST_REPLAY_METHOD } from '../shared/agent-hook-relay' +import { AGENT_STATUS_ASSISTANT_MESSAGE_MAX_LENGTH } from '../shared/agent-status-types' import { AgentHookServer } from '../main/agent-hooks/server' +import { publishAgentHookEnvelope } from './agent-hook-envelope-publication' const LEAF_7 = '77777777-7777-4777-8777-777777777777' const LEAF_9 = '99999999-9999-4999-8999-999999999999' @@ -57,22 +59,25 @@ describe('Integration: relay hook server → mux → AgentHookServer.ingestRemot onClose: () => {} } - dispatcher = new RelayDispatcher((data: Buffer) => { - setImmediate(() => { - for (const cb of clientDataCallbacks) { - cb(data) - } - }) - }) + dispatcher = new RelayDispatcher( + (data: Buffer) => { + setImmediate(() => { + for (const cb of clientDataCallbacks) { + cb(data) + } + }) + }, + { + writableHighWaterMark: () => 4096, + writableLength: () => 0 + } + ) relayFeedFn = (data: Buffer) => dispatcher.feed(data) hookServer = new RelayAgentHookServer({ endpointDir: tmpDir, forward: (envelope) => { - dispatcher.notify( - AGENT_HOOK_NOTIFICATION_METHOD, - envelope as unknown as Record - ) + publishAgentHookEnvelope(dispatcher, envelope) } }) await hookServer.start() @@ -157,6 +162,42 @@ describe('Integration: relay hook server → mux → AgentHookServer.ingestRemot expect(payload.agentType).toBe('claude') }) + it('sheds an oversized assistant message through the production publication path', async () => { + const events: { payload: { state: string; lastAssistantMessage?: string } }[] = [] + orcaServer.setListener((event) => { + events.push({ payload: event.payload }) + }) + const { port, token } = hookServer.getCoordinates() + const post = (payload: Record): Promise => + fetch(`http://127.0.0.1:${port}/hook/claude`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Orca-Agent-Hook-Token': token + }, + body: JSON.stringify({ paneKey: `tab-7:${LEAF_7}`, payload }) + }) + + await expect( + post({ hook_event_name: 'UserPromptSubmit', prompt: 'bounded publication' }) + ).resolves.toMatchObject({ status: 204 }) + await expect( + post({ + hook_event_name: 'Stop', + last_assistant_message: 'a'.repeat(AGENT_STATUS_ASSISTANT_MESSAGE_MAX_LENGTH) + }) + ).resolves.toMatchObject({ status: 204 }) + + const start = Date.now() + while (events.length < 2 && Date.now() - start < 1500) { + await new Promise((resolve) => setImmediate(resolve)) + } + expect(events).toHaveLength(2) + expect(events[1].payload.state).toBe('done') + expect(events[1].payload.lastAssistantMessage).toBeUndefined() + expect(dispatcher.activeClientIds()).toHaveLength(1) + }) + it('clears remote Claude permission when the approved tool starts with matching identity', async () => { const { port, token } = hookServer.getCoordinates() const postClaude = async (payload: Record): Promise => diff --git a/src/relay/dispatcher-capacity-degradation.test.ts b/src/relay/dispatcher-capacity-degradation.test.ts index 18fc9b1371f..b5f5cc7aa3f 100644 --- a/src/relay/dispatcher-capacity-degradation.test.ts +++ b/src/relay/dispatcher-capacity-degradation.test.ts @@ -403,7 +403,7 @@ describe('RelayDispatcher bounded-capacity degradation', () => { } }) - it('sendResponse leaves an already-failed response rejected without closing', async () => { + it('sendResponse settles an already-failed oversized response without closing', async () => { const primary = makeBoundedClient(65536) const bounded = new RelayDispatcher(primary.write, primary.options) try { @@ -414,7 +414,14 @@ describe('RelayDispatcher bounded-capacity degradation', () => { await vi.advanceTimersByTimeAsync(0) expect(primary.closes).toBe(0) - expect(primary.frames).toHaveLength(0) + expect(primary.frames).toHaveLength(1) + const response = decodePayload(primary.frames[0]) as unknown as { + id: number + error: { code: number; message: string } + } + expect(response.id).toBe(78) + expect(response.error.code).toBe(RelayErrorCode.ResponseOverCapacity) + expect(response.error.message).toBe('Relay response exceeded the bounded transport capacity') } finally { bounded.dispose() } diff --git a/src/relay/dispatcher.ts b/src/relay/dispatcher.ts index d48b5daffe1..05fda848da7 100644 --- a/src/relay/dispatcher.ts +++ b/src/relay/dispatcher.ts @@ -919,10 +919,10 @@ export class RelayDispatcher { const estimatedBytes = this.estimateFrameBytes(msg) const lane = estimatedBytes > DISPATCHER_CONTROL_QUEUE_MAX_BYTES ? 'legacy-response' : 'control' const accepted = this.enqueueFrame(client, msg, lane, onSettled) - if (accepted || error !== undefined) { - return accepted + if (accepted) { + return true } - // Why: an oversized result must fail its own request; closing would kill every pane on the host. + // Why: an oversized response must fail its own request; closing would kill every pane on the host. // A rejected first enqueue either left onSettled untouched or closed the client, so exactly one settlement happens. return this.enqueueFrame( client, diff --git a/src/relay/protocol.ts b/src/relay/protocol.ts index 642f694c526..567166be4fa 100644 --- a/src/relay/protocol.ts +++ b/src/relay/protocol.ts @@ -103,7 +103,7 @@ export type GitResponseStreamMarker = { export const RelayErrorCode = { TooManyStreams: -33006, StreamProtocolError: -33007, - /** Substituted for a result too large for the sink's frame capacity; the request fails + /** Substituted for a response 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 diff --git a/src/relay/relay-watcher-event-emitter.test.ts b/src/relay/relay-watcher-event-emitter.test.ts index 3f87d57e266..fccf85c37d5 100644 --- a/src/relay/relay-watcher-event-emitter.test.ts +++ b/src/relay/relay-watcher-event-emitter.test.ts @@ -203,6 +203,21 @@ describe('relay watcher overflow suppression key', () => { dispatcher.dispose() } }) + + it('releases the outstanding key when notification admission rejects without settlement', () => { + const sink = createRecordingSink(65536) + const dispatcher = new RelayDispatcher(sink.write, sink.options) + const notify = vi.spyOn(dispatcher, 'tryNotifyClient').mockReturnValue(false) + + try { + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + emitRelayWatcherOverflow(dispatcher, '/workspace', false) + expect(notify).toHaveBeenCalledTimes(2) + } finally { + notify.mockRestore() + dispatcher.dispose() + } + }) }) describe('relay watcher batch chunking', () => { @@ -233,6 +248,21 @@ describe('relay watcher batch chunking', () => { } }) + it('encodes the empty envelope once instead of re-encoding candidate chunks', () => { + const sink = createRecordingSink(16384) + const dispatcher = new RelayDispatcher(sink.write, sink.options) + const budget = vi.spyOn(dispatcher, 'producerEnvelopeBudget') + + try { + emitRelayWatcherEvents(dispatcher, '/workspace', false, watcherBatch(5000)) + expect(budget).toHaveBeenCalledTimes(1) + expect(sink.frames.flatMap(frameEvents)).toHaveLength(5000) + } finally { + budget.mockRestore() + dispatcher.dispose() + } + }) + it('falls back to one overflow marker when a single event exceeds the frame budget', () => { const sink = createRecordingSink(65536) 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 40b31ce6b91..fea31dfc524 100644 --- a/src/relay/relay-watcher-event-emitter.ts +++ b/src/relay/relay-watcher-event-emitter.ts @@ -57,6 +57,10 @@ function groupWatcherEventsByDirectory( return Array.from(groups.values()).flat() } +function encodedWatcherEventBytes(event: MappedWatcherEvent): number { + return Buffer.byteLength(JSON.stringify(event)) +} + // Batches are sized to each sink's frame capacity; a batch that cannot be sized degrades to an overflow resync. function publishWatcherBatchToClient( dispatcher: RelayDispatcher, @@ -67,8 +71,6 @@ function publishWatcherBatchToClient( ): void { const publish = (events: readonly MappedWatcherEvent[]): boolean => dispatcher.publishProducerNotification(clientId, 'fs.changed', { events }) - const budgetFor = (events: readonly MappedWatcherEvent[]): number => - dispatcher.producerEnvelopeBudget('fs.changed', { events }, clientId) // Fast path: publish the whole batch first — two encodes, the same cost as an unchunked emit. // logDrop:false because rejection here is a measurement, not an outcome: the batch is re-sent in @@ -87,18 +89,27 @@ function publishWatcherBatchToClient( } // Rejection is ambiguous: an over-capacity frame is chunkable, a full producer queue is real data loss. - const batchBudget = budgetFor(mapped) - if (batchBudget >= 0) { + // The empty envelope is encoded once; event JSON sizes are exact deltas apart from array commas. + const eventsCapacity = dispatcher.producerEnvelopeBudget('fs.changed', { events: [] }, clientId) + if (eventsCapacity < 0) { + emitWatcherOverflowToClient(dispatcher, clientId, rootPath) + return + } + const eventBytes = new Map() + let batchBytes = Math.max(0, mapped.length - 1) + for (const event of mapped) { + const bytes = encodedWatcherEventBytes(event) + eventBytes.set(event, bytes) + batchBytes += bytes + } + if (batchBytes <= eventsCapacity) { emitWatcherOverflowToClient(dispatcher, clientId, rootPath) return } - const eventsCapacity = budgetFor([]) - // Measured average, so the first chunk already lands near capacity instead of at half of it. - const bytesPerEvent = (eventsCapacity - batchBudget) / mapped.length const grouped = groupedByDirectory() + const groupedEventBytes = grouped.map((event) => eventBytes.get(event)!) let index = 0 - let size = Math.max(1, Math.floor(eventsCapacity / bytesPerEvent)) while (index < grouped.length) { // 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 @@ -109,22 +120,21 @@ function publishWatcherBatchToClient( emitWatcherOverflowToClient(dispatcher, clientId, rootPath) return } - size = Math.min(size, grouped.length - index) - let chunk = grouped.slice(index, index + size) - let budget = budgetFor(chunk) - while (budget < 0 && size > 1) { - // Shrink by the measured excess rather than halving, which overshoots the byte minimum ~1.7x. - size = Math.max(1, Math.min(size - 1, size - Math.ceil(-budget / bytesPerEvent))) - chunk = grouped.slice(index, index + size) - budget = budgetFor(chunk) + let end = index + let chunkBytes = 0 + while (end < grouped.length) { + const nextBytes = groupedEventBytes[end] + (end === index ? 0 : 1) + if (chunkBytes + nextBytes > eventsCapacity) { + break + } + chunkBytes += nextBytes + end += 1 } - if (budget < 0 || !publish(chunk)) { + if (end === index || !publish(grouped.slice(index, end))) { emitWatcherOverflowToClient(dispatcher, clientId, rootPath) return } - index += size - // Re-grow into the measured headroom so one shrunk chunk cannot pin the rest of the walk below capacity. - size += Math.floor(budget / bytesPerEvent) + index = end } } @@ -143,18 +153,25 @@ function emitWatcherOverflowToClient( } outstanding.add(key) outstandingOverflowMarkers.set(dispatcher, outstanding) - dispatcher.tryNotifyClient( + const release = (): void => { + outstanding.delete(key) + if (outstanding.size === 0) { + outstandingOverflowMarkers.delete(dispatcher) + } + } + const accepted = dispatcher.tryNotifyClient( clientId, 'fs.changed', { events: [{ kind: 'overflow', absolutePath: rootPath }] }, () => { // Settles on write, drop, or client close, so the slot can never leak. - outstanding.delete(key) - if (outstanding.size === 0) { - outstandingOverflowMarkers.delete(dispatcher) - } + release() } ) + // A close/dispose race may reject between tryNotifyClient's pre-check and enqueue settlement. + if (!accepted) { + release() + } } export function emitRelayWatcherOverflow( diff --git a/src/shared/agent-hook-relay.test.ts b/src/shared/agent-hook-relay.test.ts index a22655474a5..f00443a801c 100644 --- a/src/shared/agent-hook-relay.test.ts +++ b/src/shared/agent-hook-relay.test.ts @@ -79,10 +79,15 @@ describe('restoreShedStatusFields', () => { expect(restored.lastAssistantMessage).toBeUndefined() }) - it('restores the assistant message and reads the producer key', () => { + it('restores shed subagents without resurrecting stale prose for a new state and prompt', () => { expect(AGENT_HOOK_SHED_FIELDS_KEY).toBe('shedFields') - const restored = restoreShedStatusFields(shed, ['lastAssistantMessage', 'subagents'], cached) - expect(restored.lastAssistantMessage).toBe('cached message') + const nextTurn = { ...shed, prompt: 'next prompt' } + const restored = restoreShedStatusFields( + nextTurn, + ['lastAssistantMessage', 'subagents'], + cached + ) + expect(restored.lastAssistantMessage).toBeUndefined() expect(restored.subagents).toEqual(roster) }) @@ -92,9 +97,10 @@ describe('restoreShedStatusFields', () => { expect(restored).toBe(shed) }) - it('leaves a genuinely cleared field cleared when the relay shed nothing', () => { + it('leaves an intentional lastAssistantMessage clear intact when the relay shed nothing', () => { expect(restoreShedStatusFields(shed, undefined, cached)).toBe(shed) expect(restoreShedStatusFields(shed, [], cached)).toBe(shed) + expect(restoreShedStatusFields(shed, [], cached).lastAssistantMessage).toBeUndefined() // A hostile/garbled marker must not throw or restore. expect(restoreShedStatusFields(shed, 'subagents', cached)).toBe(shed) expect(restoreShedStatusFields(shed, [{ toString: () => 'subagents' }], cached)).toBe(shed) diff --git a/src/shared/agent-hook-relay.ts b/src/shared/agent-hook-relay.ts index a5f912023f3..cf2676c8d02 100644 --- a/src/shared/agent-hook-relay.ts +++ b/src/shared/agent-hook-relay.ts @@ -109,8 +109,8 @@ export const AGENT_HOOK_SHED_FIELDS_KEY = 'shedFields' as const * 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. + * `interactivePrompt` and `lastAssistantMessage` are deliberately not restored: cached prose can + * belong to an earlier turn, while subagent continuity is required for correct hibernation. */ export function restoreShedStatusFields( payload: ParsedAgentStatusPayload, @@ -123,17 +123,12 @@ export function restoreShedStatusFields( 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) { + if (subagents === undefined) { return payload } return { ...payload, - ...(subagents === undefined ? {} : { subagents }), - ...(lastAssistantMessage === undefined ? {} : { lastAssistantMessage }) + subagents } }