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

@@ -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', () => {

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]
 })()

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)

src/relay/agent-hook-envelope-publication.ts

@@ -53,11 +53,19 @@ function toNotificationParams(
 function publishToClients(
   dispatcher: RelayDispatcher,
   params: Record<string, unknown>,
-  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.

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<string, unknown>
-        )
+        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<string, unknown>): Promise<Response> =>
+      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<string, unknown>): Promise<Response> =>

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()
     }

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,

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

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)

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<MappedWatcherEvent, number>()
+  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(

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)

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
   }
 }