fix(mothership): hand off a failed replay append again, bounded by the recovery budget

Ending the turn on any leased append error that was not a lost lease turned a
short Redis blip (a failover READONLY or a connection reset that outlasts the
append retries) into a failed Chat turn and a worker stop. Before, that turn was
handed off and a successor completed it.

- A failed append that is not a budget refusal hands off again, as on staging.
  The per-run recovery budget bounds a persistent failure: MAX_RECOVERY_ATTEMPTS
  takeovers with backoff, and then the run ends as stream_recovery_exhausted.
- Drop StreamPersistenceFailedError and stream_persistence_failed. The writer
  only latches budget refusals again, and finalizeAsError surfaces a failed
  terminal append as it did before.
- Keep the change that an unreadable lease is not a hand-off. Its test now
  fails a lease read alone: a corrupted lock key also fails the fenced append,
  which is now a hand-off.
- Storm suite: a single failed append hands off and the next controller
  completes the turn. A persistent append failure exhausts the budget, the same
  as a lease lost with no successor.

Author: waleedlatif1

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

@@ -62,11 +62,7 @@ import {
 } from '@/lib/mothership/request/session/controller-lease'
 import { StreamReplayBudgetExhaustedError } from '@/lib/mothership/request/session/replay-budget'
 import { SSE_RESPONSE_HEADERS } from '@/lib/mothership/request/session/sse'
-import {
-  StreamPersistenceFailedError,
-  type StreamTurnFailure,
-  turnFailure,
-} from '@/lib/mothership/request/session/turn-failure'
+import { type StreamTurnFailure, turnFailure } from '@/lib/mothership/request/session/turn-failure'
 import { TraceCollector } from '@/lib/mothership/request/trace'
 import type { OrchestratorResult } from '@/lib/mothership/request/types'
 import { getMothershipBaseURL } from '@/lib/mothership/server/agent-url'
@@ -448,17 +444,18 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
                   await publisher.publish(event)
                 } catch (error) {
                   /*
-                    Only a lost lease hands the turn to a successor. A refusal, or a
-                    Redis error that outlasted its retries, is terminal: leaving it
-                    recoverable made each replacement re-receive and fail the same
-                    event, re-POSTing the run on every reconnect poll.
+                    A refused write is a terminal failure, not a handoff: leaving it
+                    recoverable made each replacement re-receive and re-refuse the same
+                    event. Any other failure to persist means this controller can no
+                    longer prove ownership of the replay, so a successor takes over; the
+                    run's recovery budget bounds how often that repeats.
                   */
-                  const failure =
-                    error instanceof StreamControllerSupersededError
-                      ? error
-                      : (turnFailure(error) ?? new StreamPersistenceFailedError(error))
-                  if (!abortController.signal.aborted) abortController.abort(failure)
-                  throw failure
+                  if (!abortController.signal.aborted) {
+                    abortController.abort(
+                      turnFailure(error) ?? new StreamControllerSupersededError()
+                    )
+                  }
+                  throw error
                 }
               },
               onAbortObserved: (reason) => {

apps/sim/lib/mothership/request/session/recovery-storm.integration.ts

@@ -2,8 +2,8 @@
  * How often reconnecting to an orphaned Chat run re-POSTs it to the worker, against real
  * Redis and PostgreSQL. The production reconnect route, stream recovery, chat lifecycle
  * and finalization run unmodified; a local HTTP server stands in for the worker. Faults
- * are injected at one seam, the leased replay append: an append that fails for a reason
- * other than a lost lease, or a lease lost with no successor to take over.
+ * are injected at one seam, the leased replay append: an append that fails once or every
+ * time, or a lease lost with no successor to take over.
  */
 import { authMock, authMockFns } from '@sim/testing/mocks/auth.mock'
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
@@ -15,7 +15,10 @@ const { redisUrl, inheritedEnv, worker, faults } = await vi.hoisted(async () =>
     /** Which leased appends fail, and how; `undefined` lets every append through. */
     append: undefined as
       | undefined
-      | { frames: 'any_tool' | 'tool_result'; effect: 'throw' | 'lose_lease' },
+      | {
+          frames: 'any_tool' | 'tool_result'
+          effect: 'throw' | 'throw_once' | 'lose_lease'
+        },
   }
   const worker = {
     posts: [] as Array<{ path: string; at: number }>,
@@ -76,6 +79,11 @@ vi.mock('@/lib/mothership/request/session/buffer', async (importOriginal) => {
               (envelope.payload as { phase?: string }).phase === 'result')
         )
       if (hit && fault.effect === 'throw') throw new Error('simulated Redis write failure')
+      /** A Redis blip that outlasts the append retries, after which Redis is healthy again. */
+      if (hit && fault.effect === 'throw_once') {
+        faults.append = undefined
+        throw new Error('simulated transient Redis write failure')
+      }
       /** The lock expires under a live controller, and nobody else holds it. */
       if (hit && fault.effect === 'lose_lease') await getRedisClient()!.del(lease.key)
       return actual.appendEvents(...args)
@@ -105,10 +113,7 @@ import { isTerminalStreamStatus } from '@/lib/mothership/request/session'
 import { appendEvents } from '@/lib/mothership/request/session/buffer'
 import { chatStreamLockKey } from '@/lib/mothership/request/session/controller-lease'
 import { createEvent } from '@/lib/mothership/request/session/event'
-import {
-  StreamPersistenceFailedError,
-  StreamRecoveryExhaustedError,
-} from '@/lib/mothership/request/session/turn-failure'
+import { StreamRecoveryExhaustedError } from '@/lib/mothership/request/session/turn-failure'
 import { GET as streamGET } from '@/app/api/copilot/chat/stream/route'
 
 const userId = generateId()
@@ -331,56 +336,57 @@ describe.runIf(Boolean(redisUrl))('reconnecting to an orphaned Chat run', () =>
     30_000
   )
 
+  it('hands off a turn whose append fails once, and the next controller completes it', async () => {
+    worker.mode = 'frames'
+    const run = await orphanedRun()
+    faults.append = { frames: 'any_tool', effect: 'throw_once' }
+    try {
+      const { turns, stops } = await reconnect(run, { windowMs: 15_000 })
+
+      expect(turns).toHaveLength(2)
+      expect(stops).toHaveLength(0)
+      expect(await storedRun(run.runId)).toMatchObject({
+        status: 'complete',
+        recoveryBackoff: { attempts: 2 },
+      })
+    } finally {
+      faults.append = undefined
+    }
+  }, 60_000)
+
   it.each([
-    { frames: 'any_tool', tails: 1 },
-    { frames: 'any_tool', tails: 3 },
-    { frames: 'tool_result', tails: 1 },
-    { frames: 'tool_result', tails: 3 },
+    { frames: 'any_tool', effect: 'throw', tails: 1 },
+    { frames: 'any_tool', effect: 'throw', tails: 3 },
+    { frames: 'tool_result', effect: 'throw', tails: 1 },
+    { frames: 'tool_result', effect: 'throw', tails: 3 },
+    { frames: 'any_tool', effect: 'lose_lease', tails: 3 },
   ] as const)(
-    'ends the run after one worker request when a $frames frame cannot be persisted ($tails tails)',
-    async ({ frames, tails }) => {
+    'gives up on a run whose recovered controllers keep failing ($effect on $frames, $tails tails)',
+    async ({ frames, effect, tails }) => {
       worker.mode = 'frames'
       const run = await orphanedRun()
-      faults.append = { frames, effect: 'throw' }
+      faults.append = { frames, effect }
       try {
-        const { turns, stops } = await reconnect(run, { tails, windowMs: 15_000 })
+        const { turns, stops } = await reconnect(run, { tails, windowMs: 60_000 })
 
-        expect(turns).toHaveLength(1)
+        expect(turns).toHaveLength(MAX_RECOVERY_ATTEMPTS)
+        /** Each takeover waits at least the jittered floor of the one before it. */
+        turns.slice(1).forEach((turn, i) => {
+          expect(turn.at - turns[i].at).toBeGreaterThanOrEqual(0.7 * 1_000 * 2 ** i)
+        })
         expect(stops).toHaveLength(1)
         expect(await storedRun(run.runId)).toMatchObject({
           status: 'error',
-          error: new StreamPersistenceFailedError(undefined).userMessage,
+          error: new StreamRecoveryExhaustedError().userMessage,
+          recoveryBackoff: { attempts: MAX_RECOVERY_ATTEMPTS + 1 },
         })
       } finally {
         faults.append = undefined
       }
     },
-    60_000
+    120_000
   )
 
-  it('gives up on a run whose recovered controllers keep losing their lease', async () => {
-    worker.mode = 'frames'
-    const run = await orphanedRun()
-    faults.append = { frames: 'any_tool', effect: 'lose_lease' }
-    try {
-      const { turns, stops } = await reconnect(run, { tails: 3, windowMs: 60_000 })
-
-      expect(turns).toHaveLength(MAX_RECOVERY_ATTEMPTS)
-      /** Each takeover waits at least the jittered floor of the one before it. */
-      turns.slice(1).forEach((turn, i) => {
-        expect(turn.at - turns[i].at).toBeGreaterThanOrEqual(0.7 * 1_000 * 2 ** i)
-      })
-      expect(stops).toHaveLength(1)
-      expect(await storedRun(run.runId)).toMatchObject({
-        status: 'error',
-        error: new StreamRecoveryExhaustedError().userMessage,
-        recoveryBackoff: { attempts: MAX_RECOVERY_ATTEMPTS + 1 },
-      })
-    } finally {
-      faults.append = undefined
-    }
-  }, 120_000)
-
   it('waits out the backoff of a recent takeover, then takes over and completes the run', async () => {
     worker.mode = 'frames'
     const now = Date.now()

apps/sim/lib/mothership/request/session/replay-budget.integration.ts

@@ -17,6 +17,8 @@ const { redisUrl, inheritedEnv, worker } = await vi.hoisted(async () => {
     onAbort: undefined as (() => Promise<void>) | undefined,
     /** The read-only replay's answer; a worker that does not know the run by default. */
     replay: { status: 404, frames: [] as unknown[] },
+    /** How many coming reads of the controller's own lease fail, as a Redis error would. */
+    leaseReadFailures: 0,
   }
   const replayRequests: Array<Record<string, unknown>> = []
   const server = createHttpServer(async (request, response) => {
@@ -72,6 +74,20 @@ const { redisUrl, inheritedEnv, worker } = await vi.hoisted(async () => {
 })
 
 vi.mock('@/lib/auth', () => authMock)
+vi.mock('@/lib/mothership/request/session/controller-lease', async (importOriginal) => {
+  const actual =
+    await importOriginal<typeof import('@/lib/mothership/request/session/controller-lease')>()
+  return {
+    ...actual,
+    assertChatStreamLease: async (...args: Parameters<typeof actual.assertChatStreamLease>) => {
+      if (worker.hooks.leaseReadFailures > 0) {
+        worker.hooks.leaseReadFailures--
+        throw new Error('simulated Redis read failure')
+      }
+      return actual.assertChatStreamLease(...args)
+    },
+  }
+})
 vi.mock('@/lib/mothership/request/lifecycle/run', () => ({
   /**
    * Stands in for the worker leg: forwards each scripted event to the controller's
@@ -147,7 +163,6 @@ import {
   StreamReplayBudgetExhaustedError,
 } from '@/lib/mothership/request/session/replay-budget'
 import { STREAM_STRING_PREVIEW_UNITS } from '@/lib/mothership/request/session/replay-compaction'
-import { STREAM_PERSISTENCE_FAILED_CODE } from '@/lib/mothership/request/session/turn-failure'
 import type { StreamEvent } from '@/lib/mothership/request/session/types'
 import { StreamWriter } from '@/lib/mothership/request/session/writer'
 import type { StreamingContext } from '@/lib/mothership/request/types'
@@ -659,7 +674,7 @@ describe.runIf(Boolean(redisUrl))('a turn whose stream exhausts its replay budge
     }
   )
 
-  it('ends a turn whose append fails while it still holds the lease, and stops the worker', async () => {
+  it('leaves a run it handed off recoverable after a transient append failure', async () => {
     let eventsKey = ''
     const { streamId, runId, frames } = await runTurn(
       [
@@ -674,36 +689,27 @@ describe.runIf(Boolean(redisUrl))('a turn whose stream exhausts its replay budge
       }
     )
 
-    expect(frames.map((frame) => [frame.type, frame.payload.code ?? frame.payload.status])).toEqual(
-      [
-        ['session', undefined],
-        ['error', STREAM_PERSISTENCE_FAILED_CODE],
-        ['complete', 'error'],
-      ]
-    )
+    expect(frames.map((frame) => frame.type)).toEqual(['session'])
+    expect(await redis().ttl(`mothership_stream:${streamId}:events`)).toBeGreaterThan(300)
     const [stored] = await db.select().from(copilotRuns).where(eq(copilotRuns.id, runId))
-    expect(stored.status).toBe('error')
-    expect(worker.abortRequests).toEqual([expect.objectContaining({ messageId: streamId })])
-    expect(worker.runs[0].dispatched).toEqual([])
-    expect(await redis().get(chatStreamLockKey(chatId))).toBeNull()
+    expect(stored.status).toBe('active')
+    expect(worker.abortRequests).toEqual([])
   })
 
-  it('settles a finished turn whose lease became unreadable instead of handing it off', async () => {
-    const { runId, frames } = await runTurn([
-      text('Done.'),
-      async () => {
-        // A lock key of the wrong type makes every read of the lease fail, as a Redis error would.
-        await redis().del(chatStreamLockKey(chatId))
-        await redis().hset(chatStreamLockKey(chatId), 'corrupt', '1')
-      },
-    ])
-
+  it('settles a finished turn whose lease could not be read instead of handing it off', async () => {
     try {
+      const { runId, frames } = await runTurn([
+        text('Done.'),
+        async () => {
+          worker.hooks.leaseReadFailures = 1
+        },
+      ])
+
       expect(frames.at(-1)).toMatchObject({ type: 'complete', payload: { status: 'complete' } })
       const [stored] = await db.select().from(copilotRuns).where(eq(copilotRuns.id, runId))
       expect(stored.status).toBe('complete')
     } finally {
-      await redis().del(chatStreamLockKey(chatId))
+      worker.hooks.leaseReadFailures = 0
     }
   })
 
@@ -1065,7 +1071,7 @@ describe.runIf(Boolean(redisUrl))('a turn whose stream exhausts its replay budge
     )
   }
 
-  it('marks its run terminal even when the final events cannot be persisted', async () => {
+  it('marks its run terminal even when the final events cannot be published', async () => {
     const controllerToken = `owner\n${generateId()}`
     const { streamId, runId } = await pausedRun(controllerToken)
     const lease = { key: chatStreamLockKey(generateId()), value: controllerToken }
@@ -1074,8 +1080,10 @@ describe.runIf(Boolean(redisUrl))('a turn whose stream exhausts its replay budge
     await redis().set(`mothership_stream:${streamId}:events`, 'not a sorted set', 'EX', 60)
     const publisher = new StreamWriter({ streamId, requestId: generateId(), lease })
 
-    await finalizeAsError(publisher, runId)
+    const failure = await finalizeAsError(publisher, runId).catch((error: unknown) => error)
 
+    expect(failure).toBeInstanceOf(Error)
+    expect(failure).not.toBeInstanceOf(StreamControllerSupersededError)
     const [stored] = await db.select().from(copilotRuns).where(eq(copilotRuns.id, runId))
     expect(stored.status).toBe('error')
     await redis().del(lease.key)

apps/sim/lib/mothership/request/session/turn-failure.ts

@@ -1,8 +1,7 @@
 /**
- * A failure that ends the turn as an error instead of handing it to a successor.
- * Only a controller that provably lost its chat lease hands off
- * ({@link StreamControllerSupersededError}); a replacement would meet any other failure
- * again, re-POSTing the same run to the worker on every reconnect poll.
+ * A failure that ends the turn as an error instead of handing it to a successor
+ * ({@link StreamControllerSupersededError}): a replacement would meet it again,
+ * re-POSTing the same run to the worker on every reconnect poll.
  */
 export abstract class StreamTurnFailure extends Error {
   abstract readonly code: string
@@ -15,29 +14,9 @@ export function turnFailure(value: unknown): StreamTurnFailure | undefined {
   return value instanceof StreamTurnFailure ? value : undefined
 }
 
-/** Run-error code for a turn whose controller could not persist or verify its stream. */
-export const STREAM_PERSISTENCE_FAILED_CODE = 'stream_persistence_failed'
-
 /** Run-error code for a run whose controllers kept dying before it could finish. */
 export const STREAM_RECOVERY_EXHAUSTED_CODE = 'stream_recovery_exhausted'
 
-/**
- * The controller could not record an event to the replay buffer, or could not read its
- * own lease, for a reason other than losing it (a Redis error that outlasted the retries).
- */
-export class StreamPersistenceFailedError extends StreamTurnFailure {
-  readonly code = STREAM_PERSISTENCE_FAILED_CODE
-
-  constructor(cause: unknown) {
-    super('Stream persistence failed', { cause })
-    this.name = 'StreamPersistenceFailedError'
-  }
-
-  get userMessage(): string {
-    return 'This response was stopped because it could not be saved while streaming. The work it already completed has been saved — send a message to continue from there.'
-  }
-}
-
 /** Recovery took over the run too many times without a controller staying alive. */
 export class StreamRecoveryExhaustedError extends StreamTurnFailure {
   readonly code = STREAM_RECOVERY_EXHAUSTED_CODE