fix(memory): replace wall-clock timeout with per-chunk SSE liveness

packages/opencode/src/memory/model.ts

@@ -2,14 +2,16 @@ export * as MemoryModel from "./model"
 
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { Context, Duration, Effect, Layer, Schema } from "effect"
-import { generateObject } from "ai"
+import { streamObject } from "ai"
 import { Provider } from "@/provider/provider"
 
-// Last-resort guard for a provider that never responds at all. This is not an
-// activity budget: maintenance reasoning runs are long, generation terminates
-// on its own via max_output_tokens, and transport timers own dead-connection
-// detection, so a call that keeps working finishes before this ever fires.
-const RESPONSE_TIMEOUT = Duration.minutes(5)
+// Liveness is judged per-chunk, never by a whole-call wall clock: a stream
+// that keeps delivering parts is alive, however long the reasoning runs.
+// CONNECT_TIMEOUT bounds the wait for the FIRST part; IDLE_TIMEOUT bounds the
+// silence BETWEEN parts and is re-armed by every arriving part. Generation
+// still terminates on its own via max_output_tokens / a natural stop.
+const CONNECT_TIMEOUT = Duration.seconds(60)
+const IDLE_TIMEOUT = Duration.seconds(60)
 
 const JSON_HINT = "Respond with a JSON object matching the provided schema."
 
@@ -48,14 +50,19 @@ export function make(input: {
   readonly timeout?: Duration.Input
 }) {
   return Service.of({
-    generate: Effect.fn("MemoryModel.generate")((request) =>
-      input.execute(requireJsonToken(request)).pipe(
+    generate: Effect.fn("MemoryModel.generate")((request) => {
+      const effect = input.execute(requireJsonToken(request))
+      // An injected timeout (tests) stays a hard deadline; production relies on
+      // the per-chunk liveness below, so an actively streaming call is never
+      // killed by a wall clock.
+      if (input.timeout === undefined) return effect
+      return effect.pipe(
         Effect.timeoutOrElse({
-          duration: input.timeout ?? RESPONSE_TIMEOUT,
+          duration: input.timeout,
           orElse: () => Effect.fail(new TimeoutError()),
         }),
-      ),
-    ),
+      )
+    }),
   })
 }
 
@@ -66,29 +73,117 @@ function requireJsonToken(request: Request): Request {
   return { ...request, system: `${request.system}\n${JSON_HINT}` }
 }
 
+// Signals that the stream went silent past the liveness window.
+export class Stalled extends Error {}
+
+// Drains `parts`, re-arming the idle watchdog on EVERY part (so a live stream
+// that keeps delivering — reasoning deltas included — never trips the timer).
+// Arms `connectTimeout` until the first part and `idleTimeout` between parts;
+// a silent window invokes `onStall` (abort the request) and fails with
+// `Stalled`, while an `errorOf` hit fails with that part's error.
+export const drainWithLiveness = <T>(input: {
+  parts: AsyncIterable<T>
+  connectTimeout: Duration.Duration
+  idleTimeout: Duration.Duration
+  onStall: () => void
+  errorOf: (part: T) => unknown | undefined
+}) =>
+  new Promise<void>((resolve, reject) => {
+    let timer: ReturnType<typeof setTimeout> | undefined
+    let settled = false
+    const finish = (action: () => void) => {
+      if (settled) return
+      settled = true
+      if (timer) clearTimeout(timer)
+      action()
+    }
+    const arm = (duration: Duration.Duration) => {
+      if (timer) clearTimeout(timer)
+      timer = setTimeout(
+        () =>
+          finish(() => {
+            input.onStall()
+            reject(new Stalled())
+          }),
+        Duration.toMillis(duration),
+      )
+    }
+    arm(input.connectTimeout)
+    void (async () => {
+      try {
+        for await (const part of input.parts) {
+          arm(input.idleTimeout)
+          const error = input.errorOf(part)
+          if (error !== undefined) throw error
+        }
+        finish(resolve)
+      } catch (cause) {
+        finish(() => reject(cause))
+      }
+    })()
+  })
+
+const streamGenerate = (input: {
+  language: Parameters<typeof streamObject>[0]["model"]
+  system: string
+  prompt: string
+  schema: Schema.Decoder<unknown>
+  temperature?: number
+  maxOutputTokens: number
+  connectTimeout: Duration.Duration
+  idleTimeout: Duration.Duration
+}) =>
+  Effect.tryPromise({
+    try: (signal) =>
+      (async () => {
+        const controller = new AbortController()
+        const forwardAbort = () => controller.abort()
+        signal.addEventListener("abort", forwardAbort)
+        try {
+          const result = streamObject({
+            model: input.language,
+            system: input.system,
+            prompt: input.prompt,
+            schema: Object.assign(
+              Schema.toStandardSchemaV1(input.schema),
+              Schema.toStandardJSONSchemaV1(input.schema),
+            ),
+            temperature: input.temperature,
+            maxOutputTokens: input.maxOutputTokens,
+            abortSignal: controller.signal,
+            onError: () => {},
+          })
+          await drainWithLiveness({
+            parts: result.fullStream,
+            connectTimeout: input.connectTimeout,
+            idleTimeout: input.idleTimeout,
+            onStall: () => controller.abort(),
+            errorOf: (part) => (part.type === "error" ? part.error : undefined),
+          })
+          return await result.object
+        } finally {
+          signal.removeEventListener("abort", forwardAbort)
+        }
+      })(),
+    catch: (cause) => (cause instanceof Stalled ? new TimeoutError() : new GenerateError({ cause })),
+  })
+
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
     const provider = yield* Provider.Service
     return make({
       execute: Effect.fnUntraced(function* (input) {
         const language = yield* provider.getLanguage(input.model)
-        const schema = Object.assign(
-          Schema.toStandardSchemaV1(input.schema),
-          Schema.toStandardJSONSchemaV1(input.schema),
-        )
-        return yield* Effect.tryPromise({
-          try: (signal) =>
-            generateObject({
-              model: language,
-              system: input.system,
-              prompt: input.prompt,
-              schema,
-              temperature: input.model.capabilities.temperature ? 0 : undefined,
-              maxOutputTokens: input.maxOutputTokens,
-              abortSignal: signal,
-            }).then((result) => result.object),
-          catch: (cause) => new GenerateError({ cause }),
+        return yield* streamGenerate({
+          language,
+          system: input.system,
+          prompt: input.prompt,
+          schema: input.schema,
+          temperature: input.model.capabilities.temperature ? 0 : undefined,
+          maxOutputTokens: input.maxOutputTokens,
+          connectTimeout: CONNECT_TIMEOUT,
+          idleTimeout: IDLE_TIMEOUT,
         })
       }),
     })

packages/opencode/test/memory/memory.test.ts

@@ -1410,6 +1410,97 @@ describe("memory hidden model", () => {
       expect(seen[1].system).toBe("Propose updates as a JSON object.")
     }),
   )
+
+  it.live("keeps an actively streaming call alive regardless of total duration", () =>
+    Effect.gen(function* () {
+      // Thirty parts arriving every 8ms: ~240ms total, far past the 40ms
+      // idle window — every arrival re-arms the watchdog, so the call lives.
+      async function* streaming() {
+        for (let index = 0; index < 30; index++) {
+          await new Promise((resolve) => setTimeout(resolve, 8))
+          yield { type: index % 2 === 0 ? "reasoning" : "text-delta" }
+        }
+      }
+      yield* Effect.promise(() =>
+        MemoryModel.drainWithLiveness({
+          parts: streaming(),
+          connectTimeout: Duration.millis(40),
+          idleTimeout: Duration.millis(40),
+          onStall: () => {},
+          errorOf: () => undefined,
+        }),
+      )
+    }),
+  )
+
+  it.live("stalls a silent stream after the idle window and invokes the abort hook", () =>
+    Effect.gen(function* () {
+      let aborts = 0
+      async function* onePartThenSilence() {
+        yield { type: "reasoning" }
+        await new Promise(() => {})
+      }
+      const error = yield* Effect.tryPromise({
+        try: () =>
+          MemoryModel.drainWithLiveness({
+            parts: onePartThenSilence(),
+            connectTimeout: Duration.millis(250),
+            idleTimeout: Duration.millis(40),
+            onStall: () => {
+              aborts++
+            },
+            errorOf: () => undefined,
+          }),
+        catch: (cause) => cause,
+      }).pipe(Effect.flip)
+      expect(error instanceof MemoryModel.Stalled).toBe(true)
+      expect(aborts).toBe(1)
+    }),
+  )
+
+  it.live("fails on a dead connection that never delivers a first part", () =>
+    Effect.gen(function* () {
+      async function* nothing() {
+        await new Promise(() => {})
+      }
+      const started = Date.now()
+      const error = yield* Effect.tryPromise({
+        try: () =>
+          MemoryModel.drainWithLiveness({
+            parts: nothing(),
+            connectTimeout: Duration.millis(40),
+            idleTimeout: Duration.millis(250),
+            onStall: () => {},
+            errorOf: () => undefined,
+          }),
+        catch: (cause) => cause,
+      }).pipe(Effect.flip)
+      expect(error instanceof MemoryModel.Stalled).toBe(true)
+      expect(Date.now() - started).toBeLessThan(200)
+    }),
+  )
+
+  it.live("propagates a stream error part without treating it as a stall", () =>
+    Effect.gen(function* () {
+      const boom = new Error("provider stream error")
+      async function* erroring() {
+        yield { type: "text-delta" }
+        yield { type: "error", error: boom }
+      }
+      const error = yield* Effect.tryPromise({
+        try: () =>
+          MemoryModel.drainWithLiveness({
+            parts: erroring(),
+            connectTimeout: Duration.millis(250),
+            idleTimeout: Duration.millis(250),
+            onStall: () => {},
+            errorOf: (part: { type: string; error?: unknown }) => (part.type === "error" ? part.error : undefined),
+          }),
+        catch: (cause) => cause,
+      }).pipe(Effect.flip)
+      expect(error).toBe(boom)
+    }),
+  )
 })
 
 describe("memory maintenance budgets", () => {