fix(mothership): honest replay dedupe on execute + headless flush skip

The execute NDJSON forwarder guessed delta-vs-cumulative by string prefix
and could slice real characters off a delta that happened to begin with
the forwarded content. The wire's ordering key decides now: seq rides the
StreamEvent projection (it was dropped at construction), and the forwarder
skips only seqs it has already sent.

The per-event macrotask yield exists to flush the HTTP response buffer;
headless legs (no caller sink) now opt out via flushAfterEvent, declared
where the sink is owned.

Claude-Session: https://claude.ai/code/session_01CgaxNAaeD3taGdghbXn17w

Author: Sg312

apps/sim/app/api/mothership/execute/route.ts

@@ -350,7 +350,7 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
 
       const stream = new ReadableStream<Uint8Array>({
         start(controller) {
-          let forwardedAssistantContent = ''
+          let lastForwardedTextSeq = -1
           const send = (event: unknown) => {
             if (!cancelled) {
               controller.enqueue(encodeNdjson(event))
@@ -372,14 +372,15 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
                   event.payload.channel === MothershipStreamV1TextChannel.assistant &&
                   event.payload.text
                 ) {
-                  const text = event.payload.text
-                  const content = text.startsWith(forwardedAssistantContent)
-                    ? text.slice(forwardedAssistantContent.length)
-                    : text
-                  if (content) {
-                    forwardedAssistantContent += content
-                    send({ type: 'chunk', content })
+                  /* The wire carries text DELTAS with monotone seqs; a transport-retry
+                     replay re-delivers earlier seqs. Dedupe replays by seq — the old
+                     string-prefix guess sliced characters off a genuine delta that
+                     happened to begin with the already-forwarded content. */
+                  if (typeof event.seq === 'number') {
+                    if (event.seq <= lastForwardedTextSeq) return
+                    lastForwardedTextSeq = event.seq
                   }
+                  send({ type: 'chunk', content: event.payload.text })
                 }
               })
               allowExplicitAbort = false

apps/sim/lib/mothership/request/go/stream.ts

@@ -405,7 +405,10 @@ export async function runStreamLoop(
         // Yield a macrotask so Node.js flushes the HTTP response buffer to
         // the browser. Microtask yields (await Promise.resolve()) are not
         // enough — the I/O layer needs a full event loop tick to write.
-        await new Promise<void>((resolve) => setImmediate(resolve))
+        // Headless legs (no client response attached) opt out via flushAfterEvent.
+        if (options.flushAfterEvent !== false) {
+          await new Promise<void>((resolve) => setImmediate(resolve))
+        }
 
         if (options.onBeforeDispatch?.(streamEvent, context)) {
           return context.streamComplete || undefined

apps/sim/lib/mothership/request/lifecycle/run.ts

@@ -912,6 +912,10 @@ async function runCheckpointLoop(
 
     const loopOptions = {
       ...options,
+      /* The wrapper below always exists (checkpoint bookkeeping), so the forwarder can't
+         infer "headless" from onEvent's absence — declare it: only a caller-attached sink
+         has an HTTP buffer worth a per-event macrotask flush. */
+      flushAfterEvent: options.flushAfterEvent ?? Boolean(callerOnEvent),
       onEvent: async (event: StreamEvent) => {
         if (
           event.type === MothershipStreamV1EventType.run &&

apps/sim/lib/mothership/request/session/contract.ts

@@ -33,7 +33,13 @@ type EnvelopeToStreamEvent<T> = T extends {
   payload: infer TPayload
   scope?: infer TScope
 }
-  ? { type: TType; payload: TPayload; scope?: Exclude<TScope, undefined> }
+  ? {
+      type: TType
+      payload: TPayload
+      scope?: Exclude<TScope, undefined>
+      /** Wire ordering key, carried off the envelope; absent on synthetic events. */
+      seq?: number
+    }
   : never
 
 export type SyntheticFilePreviewPhase = (typeof FILE_PREVIEW_PHASE)[keyof typeof FILE_PREVIEW_PHASE]

apps/sim/lib/mothership/request/session/event.test.ts

@@ -53,6 +53,7 @@ describe('createEvent', () => {
     const streamEvent = eventToStreamEvent(envelope)
     expect(streamEvent).toEqual({
       type: MothershipStreamV1EventType.tool,
+      seq: 2,
       payload: {
         previewPhase: 'file_preview_start',
         toolCallId: 'preview-1',

apps/sim/lib/mothership/request/session/event.ts

@@ -69,6 +69,7 @@ export function eventToStreamEvent<TEnvelope extends PersistedStreamEventEnvelop
   return {
     type: envelope.type,
     payload: envelope.payload,
+    seq: envelope.seq,
     ...(envelope.scope ? { scope: envelope.scope } : {}),
   } as StreamEventFromEnvelope<TEnvelope>
 }

apps/sim/lib/mothership/request/types.ts

@@ -220,6 +220,12 @@ export interface OrchestratorOptions {
   autoExecuteTools?: boolean
   timeout?: number
   onEvent?: (event: StreamEvent) => void | Promise<void>
+  /**
+   * Whether the per-event macrotask yield (which lets Node flush the HTTP response buffer)
+   * should run. The sink owner sets this: legs with no client response attached have
+   * nothing to flush, and the yield only slows the forwarder. Defaults to true.
+   */
+  flushAfterEvent?: boolean
   onComplete?: (result: OrchestratorResult) => void | Promise<void>
   onError?: (error: Error, result?: OrchestratorResult) => void | Promise<void>
   abortSignal?: AbortSignal