fix(mothership): omit the largest field of an event no cut can bound, and persist closed lanes

- An event whose many short values survive every string and array cut, such
  as an object with thousands of keys, now has its largest bulk field
  replaced by a size note when it would still exceed one replay write,
  instead of ending the turn. Identity fields and client-executed arguments
  are never replaced; such an event is still refused cleanly.
- finalizeResidualToolCalls reports closing an open subagent lane as a
  change, so the error finalizer persists it instead of leaving a stale lane.

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-helpers.test.ts

@@ -27,4 +27,11 @@ describe('finalizeResidualToolCalls', () => {
     expect(finalizeResidualToolCalls(open, 'error')).toBe(true)
     expect(finalizeResidualToolCalls(settled, 'error')).toBe(false)
   })
+
+  it('reports closing an open subagent lane as a change to persist', () => {
+    const blocks: ContentBlock[] = [{ type: 'subagent', content: 'research' }]
+
+    expect(finalizeResidualToolCalls(blocks, 'error')).toBe(true)
+    expect(blocks[0].endedAt).toEqual(expect.any(Number))
+  })
 })

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-helpers.ts

@@ -72,7 +72,7 @@ export function asPayloadRecord(value: unknown): StreamPayload | undefined {
  * Settles every unfinished tool row (running, pending, or awaiting approval) at
  * a turn terminal by propagating the turn's outcome: a clean `complete` settles
  * a straggler `success`, a stop `cancelled`, an error `error`. Also closes any
- * open subagent lane. Returns whether any tool row was settled.
+ * open subagent lane. Returns whether it settled a row or closed a lane.
  */
 export function finalizeResidualToolCalls(
   blocks: ContentBlock[],
@@ -94,6 +94,7 @@ export function finalizeResidualToolCalls(
     // transport-based gating.
     if (block.type === 'subagent' && block.endedAt === undefined) {
       block.endedAt = endedAt
+      settled = true
       continue
     }
     const tc = block.toolCall

apps/sim/lib/mothership/request/session/replay-compaction.test.ts

@@ -14,6 +14,7 @@ import { describe, expect, it } from 'vitest'
 import {
   compactStreamEvent,
   STREAM_EVENT_COMPACTION_THRESHOLD_BYTES,
+  STREAM_EVENT_MAX_PAYLOAD_BYTES,
   STREAM_STRING_PREVIEW_UNITS,
 } from '@/lib/mothership/request/session/replay-compaction'
 import type { StreamEvent } from '@/lib/mothership/request/session/types'
@@ -237,4 +238,49 @@ describe('compactStreamEvent', () => {
     expect(output.data).toEqual({ results: citations })
     expect(toRecord(payloadOf(event).output).rows).toHaveLength(40_000)
   })
+
+  it('omits the largest remaining field when cuts alone leave the event over one replay write', () => {
+    const blocks = Object.fromEntries(
+      Array.from({ length: 5_000 }, (_, index) => [`block-${index}`, 'b'.repeat(300)])
+    )
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'c',
+        toolName: 'cli_workflows_state_get',
+        executor: 'sim',
+        mode: 'async',
+        phase: 'result',
+        success: true,
+        status: 'success',
+        output: { blocks },
+      },
+    }
+
+    const compacted = payloadOf(compactStreamEvent(event))
+
+    expect(Buffer.byteLength(JSON.stringify(compacted))).toBeLessThanOrEqual(
+      STREAM_EVENT_MAX_PAYLOAD_BYTES
+    )
+    expect(compacted.output).toMatch(/^…\[omitted, [\d.]+ MB total\]$/)
+    expect(compacted).toMatchObject({ toolCallId: 'c', success: true, status: 'success' })
+  })
+
+  it('never omits identity to make room for client-executed arguments it must keep whole', () => {
+    const args = { workflowId: 'wf-1', input: 'w'.repeat(1.5 * MB) }
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'c',
+        toolName: 'run_workflow',
+        executor: 'client',
+        mode: 'async',
+        phase: 'call',
+        ui: { clientExecutable: true },
+        arguments: args,
+      },
+    }
+
+    expect(compactStreamEvent(event)).toBe(event)
+  })
 })

apps/sim/lib/mothership/request/session/replay-compaction.ts

@@ -1,4 +1,5 @@
 import { isRecordLike, toRecordOrNull } from '@sim/utils/object'
+import { getRedisBudgetLimits } from '@/lib/core/redis/byte-budget.server'
 import {
   MothershipStreamV1EventType,
   MothershipStreamV1ToolPhase,
@@ -13,6 +14,13 @@ import type { StreamEvent } from './types'
  */
 export const STREAM_EVENT_COMPACTION_THRESHOLD_BYTES = 256 * 1024
 
+/** Room left in a replay write for the envelope around an event's payload. */
+const ENVELOPE_HEADROOM_BYTES = 16 * 1024
+
+/** The largest serialized payload the replay buffer can persist in one write. */
+export const STREAM_EVENT_MAX_PAYLOAD_BYTES =
+  getRedisBudgetLimits('copilot_stream').maxSingleWriteBytes - ENVELOPE_HEADROOM_BYTES
+
 /** UTF-16 units kept at the head of a long string. */
 export const STREAM_STRING_PREVIEW_UNITS = 8 * 1024
 
@@ -100,11 +108,32 @@ function mapLeaves(
   return copy ?? value
 }
 
+/** A field must be at least this large to be omitted as a last resort. */
+const OMITTABLE_FIELD_MIN_BYTES = 64 * 1024
+
+/**
+ * Replaces the payload's largest field with a note of its size: the last resort
+ * for an event whose many short values no cut could bound, such as an object
+ * with thousands of keys.
+ */
+function omitLargestField(payload: Record<string, unknown>, skipKeys: ReadonlySet<string>) {
+  let largest: { key: string; bytes: number } | undefined
+  for (const [key, field] of Object.entries(payload)) {
+    if (skipKeys.has(key)) continue
+    const bytes = Buffer.byteLength(JSON.stringify(field) ?? '', 'utf8')
+    if (!largest || bytes > largest.bytes) largest = { key, bytes }
+  }
+  // Identity fields are never this large, so only bulk is ever replaced.
+  if (!largest || largest.bytes <= OMITTABLE_FIELD_MIN_BYTES) return payload
+  return { ...payload, [largest.key]: `…[omitted, ${formatFileSize(largest.bytes)} total]` }
+}
+
 /**
  * Bounds an outgoing stream event so the replay buffer can persist it. Applied
  * only to the copy the writer delivers and persists; the caller keeps the full
  * event for dispatch. Long strings are cut to their head in place, so every
- * object keeps its shape; if that is not enough, long arrays keep their head.
+ * object keeps its shape; if that is not enough, long arrays keep their head,
+ * and past one replay write the largest field is replaced by a size note.
  * Assistant text, file previews, and the arguments of calls the browser
  * executes are never cut; an event
  * still too large is refused by the buffer, which ends the turn with an error.
@@ -126,5 +155,8 @@ export function compactStreamEvent(event: StreamEvent): StreamEvent {
   if (Buffer.byteLength(JSON.stringify(compacted)) > STREAM_EVENT_COMPACTION_THRESHOLD_BYTES) {
     compacted = trimArrays(compacted, skipKeys)
   }
+  if (Buffer.byteLength(JSON.stringify(compacted)) > STREAM_EVENT_MAX_PAYLOAD_BYTES) {
+    compacted = omitLargestField(toRecordOrNull(compacted) ?? {}, skipKeys)
+  }
   return compacted === payload ? event : ({ ...event, payload: compacted } as StreamEvent)
 }