From 8e2329d4f0cc20db97e95d871cf40c26dd3fd8c5 Mon Sep 17 00:00:00 2001 From: Lex Date: Tue, 18 Aug 2026 06:36:05 +0800 Subject: [PATCH] fix(memory): replace wall-clock timeout with per-chunk SSE liveness Retire the whole-call wall clock in favor of activity-based liveness on the memory model seam: switch generateObject -> streamObject and re-arm an idle watchdog on every arriving part (reasoning deltas included), so an actively streaming call is never killed however long the reasoning runs. Add a connect timeout (no first part) and an idle timeout (gap between parts), both pure non-response detectors; generation still terminates via max_output_tokens / a natural stop. Addresses the streaming-liveness half of #324; the periodic prepare lock hardening remains tracked there. --- packages/opencode/src/memory/model.ts | 149 +++++++++++++++---- packages/opencode/test/memory/memory.test.ts | 91 +++++++++++ 2 files changed, 213 insertions(+), 27 deletions(-) diff --git a/packages/opencode/src/memory/model.ts b/packages/opencode/src/memory/model.ts index 88c7bcee5..f8d7c2285 100644 --- a/packages/opencode/src/memory/model.ts +++ b/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,6 +73,101 @@ 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 = (input: { + parts: AsyncIterable + connectTimeout: Duration.Duration + idleTimeout: Duration.Duration + onStall: () => void + errorOf: (part: T) => unknown | undefined +}) => + new Promise((resolve, reject) => { + let timer: ReturnType | 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[0]["model"] + system: string + prompt: string + schema: Schema.Decoder + 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* () { @@ -73,22 +175,15 @@ export const layer = Layer.effect( 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, }) }), }) diff --git a/packages/opencode/test/memory/memory.test.ts b/packages/opencode/test/memory/memory.test.ts index 49d455f2c..2f0a9e4b3 100644 --- a/packages/opencode/test/memory/memory.test.ts +++ b/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", () => {