fix(mothership): refresh the stream byte counter with its buffer, and never revive a closed buffer

- The chat-lock heartbeat now slides the owner byte counter's TTL along with the
  replay buffer's. Otherwise, after a park longer than an hour, the counter expired
  while the ring survived: the ring could grow to about twice its target, and its
  refunds drained the user counter.
- Scheduling a finished stream's cleanup now marks it closed, and the refresh leaves
  a closed stream alone. A heartbeat still in flight at teardown can no longer
  re-extend a buffer whose cleanup was already scheduled.
- Describe the worker idle timeout in terms of network intermediaries, and state
  that the delegation TTL reuses the long-running tool watchdog's cap.

Author: waleedlatif1

apps/sim/lib/mothership/auth/application-delegation.ts

@@ -2,8 +2,9 @@ import type { DelegatedPrincipal, OrganizationDelegatedPrincipal } from '@sim/au
 import { TOOL_WATCHDOG_LONG_RUNNING_MS } from '@/lib/mothership/constants'
 
 /**
- * Delegated authority is minted per operation, so it must outlive the longest
- * single tool call it authorizes, never the whole run.
+ * Delegated authority is minted per operation and must outlive the longest single
+ * tool call it authorizes, never the whole run, so it reuses the long-running tool
+ * watchdog's cap rather than a lifetime of its own.
  */
 export const COPILOT_APPLICATION_DELEGATION_TTL_MS = TOOL_WATCHDOG_LONG_RUNNING_MS
 

apps/sim/lib/mothership/constants.ts

@@ -14,9 +14,9 @@ export const SIM_AGENT_API_URL =
  * How long a worker SSE leg may stay silent before Sim treats the connection as
  * lost and re-attaches. The worker writes a keepalive comment whenever a leg has
  * been quiet for 10 s (checked every 15 s), independent of model or tool progress,
- * so a healthy leg is never silent for more than about 25 s. This must stay under
- * the NAT's fixed 350 s idle timeout, which drops a silent connection without
- * closing it.
+ * so a healthy leg is never silent for more than about 25 s. It stays well under
+ * the idle timeouts of network intermediaries, which can drop a silent connection
+ * without closing it.
  */
 export const WORKER_STREAM_IDLE_TIMEOUT_MS = 120_000
 

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

@@ -1027,7 +1027,8 @@ describe('copilot go stream helpers', () => {
   })
 
   describe('worker stream liveness without a caller deadline', () => {
-    const NAT_IDLE_MS = 350_000
+    /** Well past the idle timeout, and under common intermediary idle cuts. */
+    const INTERMEDIARY_IDLE_MS = 300_000
     const encoder = new TextEncoder()
     const frame = (event: unknown) => encoder.encode(`data: ${JSON.stringify(event)}\n\n`)
     const firstText = createEvent({
@@ -1112,7 +1113,7 @@ describe('copilot go stream helpers', () => {
       expect(context.streamComplete).toBe(true)
     })
 
-    it('fails a silent leg as a retryable interruption before the NAT drops it', async () => {
+    it('fails a silent leg as a retryable interruption before an intermediary drops it', async () => {
       vi.useFakeTimers()
       vi.mocked(fetch).mockResolvedValueOnce(
         new Response(
@@ -1136,13 +1137,13 @@ describe('copilot go stream helpers', () => {
 
       await vi.advanceTimersByTimeAsync(60_000)
       expect(state.done).toBe(false)
-      await vi.advanceTimersByTimeAsync(NAT_IDLE_MS - 60_000)
+      await vi.advanceTimersByTimeAsync(INTERMEDIARY_IDLE_MS - 60_000)
 
       expect(state.done).toBe(true)
       expect(state.error).toBeInstanceOf(WorkerStreamInterruptedError)
     })
 
-    it('fails a worker that never answers as unreachable before the NAT drops it', async () => {
+    it('fails a worker that never answers as unreachable before an intermediary drops it', async () => {
       vi.useFakeTimers()
       vi.mocked(fetch).mockImplementationOnce(
         (_url, options) =>
@@ -1163,7 +1164,7 @@ describe('copilot go stream helpers', () => {
         )
       )
 
-      await vi.advanceTimersByTimeAsync(NAT_IDLE_MS)
+      await vi.advanceTimersByTimeAsync(INTERMEDIARY_IDLE_MS)
 
       expect(state.done).toBe(true)
       expect(state.error).toBeInstanceOf(WorkerUnreachableError)

apps/sim/lib/mothership/request/session/buffer-ttl.integration.ts

@@ -17,6 +17,7 @@ const { redisUrl } = await vi.hoisted(async () => {
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
+import { getRedisBudgetKeys } from '@/lib/core/redis/byte-budget.server'
 import { MothershipStreamV1EventType } from '@/lib/mothership/generated/mothership-stream-v1'
 import {
   acquirePendingChatStream,
@@ -28,6 +29,8 @@ import {
   appendEvents,
   getLatestSeq,
   readEvents,
+  refreshBufferTtl,
+  scheduleBufferCleanup,
 } from '@/lib/mothership/request/session/buffer'
 import { createEvent } from '@/lib/mothership/request/session/event'
 import { checkForReplayGap } from '@/lib/mothership/request/session/recovery'
@@ -52,11 +55,15 @@ describe.runIf(Boolean(redisUrl))('replay buffer lifetime', () => {
     await closeRedisConnection()
   })
 
-  it('keeps a live run’s buffer through a park longer than its TTL', async () => {
+  it('keeps a live run’s buffer and its byte counter through a park longer than their TTLs', async () => {
     const chatId = generateId()
     const streamId = generateId()
     expect(await acquirePendingChatStream(chatId, streamId, 0)).toBe(true)
     await appendText(streamId, 'before the park')
+    const [ownerBudgetKey] = getRedisBudgetKeys({ kind: 'copilot_stream', id: streamId })
+    const chargedBytes = await getRedisClient()!.get(ownerBudgetKey)
+    /** The counter's own TTL is an hour; shortening it stands in for a park that long. */
+    await getRedisClient()!.expire(ownerBudgetKey, 2)
 
     vi.useFakeTimers({ toFake: ['Date'] })
     const poller = startAbortPoller(streamId, new AbortController(), { chatId, pollMs: 50 })
@@ -73,9 +80,23 @@ describe.runIf(Boolean(redisUrl))('replay buffer lifetime', () => {
 
     expect(await getLatestSeq(streamId)).toBe(1)
     expect((await readEvents(streamId, '0')).map((event) => event.seq)).toEqual([1])
+    expect(chargedBytes).not.toBeNull()
+    expect(await getRedisClient()!.get(ownerBudgetKey)).toBe(chargedBytes)
     expect(await appendText(streamId, 'after the park')).toBe(2)
   })
 
+  it('never re-extends a finished stream’s buffer after its cleanup was scheduled', async () => {
+    const streamId = generateId()
+    await appendText(streamId, 'done')
+    await scheduleBufferCleanup(streamId)
+
+    await refreshBufferTtl(streamId)
+
+    const redis = getRedisClient()!
+    expect(await redis.ttl(`mothership_stream:${streamId}:events`)).toBeGreaterThan(250)
+    expect(await redis.ttl(`mothership_stream:${streamId}:seq`)).toBeGreaterThan(250)
+  })
+
   it('reports a gap to a cursor ahead of a buffer whose numbering restarted', async () => {
     const streamId = generateId()
     for (let index = 0; index < 5; index++) await appendText(streamId, `part ${index}`)

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

@@ -42,6 +42,11 @@ function getAbortKey(streamId: string) {
   return `${STREAM_OUTBOX_PREFIX}${streamId}:abort`
 }
 
+/** Marks a stream whose cleanup is scheduled, so a late heartbeat cannot revive it. */
+function getClosedKey(streamId: string) {
+  return `${STREAM_OUTBOX_PREFIX}${streamId}:closed`
+}
+
 export type StreamConfig = {
   ttlSeconds: number
   eventLimit: number
@@ -127,23 +132,46 @@ export async function clearBuffer(streamId: string, operation = 'clear_outbox'):
       getEventsKey(streamId),
       getSeqKey(streamId),
       getAbortKey(streamId),
+      getClosedKey(streamId),
       ownerBudgetKey
     )
   })
 }
 
 /**
- * Slides a live stream's replay TTLs without an append. The TTLs otherwise move only
- * when an event lands, so a run parked on a long tool call or approval would lose its
- * replay history and restart its numbering while it is still running.
+ * KEYS: [events, seq, ownerBudget, closed]
+ * ARGV: [ttlSeconds, budgetTtlSeconds]
+ */
+const REFRESH_BUFFER_TTL_SCRIPT = `
+if redis.call('EXISTS', KEYS[4]) == 1 then return 0 end
+redis.call('EXPIRE', KEYS[1], ARGV[1])
+redis.call('EXPIRE', KEYS[2], ARGV[1])
+redis.call('EXPIRE', KEYS[3], ARGV[2])
+return 1
+`
+
+/**
+ * Slides a live stream's replay TTLs, and its byte counter's, without an append. They
+ * otherwise move only when an event lands, so a run parked on a long tool call or
+ * approval would lose its replay history and restart its numbering while it still
+ * runs, or keep its history after the counter that accounts for it expired. A stream
+ * whose cleanup is already scheduled is left to expire.
  */
 export async function refreshBufferTtl(streamId: string): Promise<void> {
   const { ttlSeconds } = getStreamConfig()
+  const [ownerBudgetKey] = getRedisBudgetKeys({ kind: 'copilot_stream', id: streamId })
+  const budgetTtlSeconds = Math.max(getRedisBudgetLimits('copilot_stream').ttlSeconds, ttlSeconds)
   await withRedisRetry({ operation: 'refresh_outbox_ttl', streamId }, async (redis) => {
-    const pipeline = redis.pipeline()
-    pipeline.expire(getEventsKey(streamId), ttlSeconds)
-    pipeline.expire(getSeqKey(streamId), ttlSeconds)
-    await pipeline.exec()
+    await redis.eval(
+      REFRESH_BUFFER_TTL_SCRIPT,
+      4,
+      getEventsKey(streamId),
+      getSeqKey(streamId),
+      ownerBudgetKey,
+      getClosedKey(streamId),
+      ttlSeconds,
+      budgetTtlSeconds
+    )
   })
 }
 
@@ -157,6 +185,7 @@ export async function scheduleBufferCleanup(
       pipeline.expire(getEventsKey(streamId), ttlSeconds)
       pipeline.expire(getSeqKey(streamId), ttlSeconds)
       pipeline.expire(getAbortKey(streamId), ttlSeconds)
+      pipeline.set(getClosedKey(streamId), '1', 'EX', ttlSeconds)
       await pipeline.exec()
     })
   } catch (error) {