From 073b71c7bb4964071a7c513325ebbe4a8f53e419 Mon Sep 17 00:00:00 2001 From: Lex Date: Mon, 17 Aug 2026 22:36:26 +0800 Subject: [PATCH] fix(memory): repair the MEMORY write path for thinking-mode providers Guarantee the literal json token at the MemoryModel seam (providers reject json_object calls without it), size match/maintain output budgets so structured replies survive reasoning, and retire the wall-clock prepare/checkpoint/in-model timeouts that killed active reasoning calls (issue #324 tracks the remaining streaming-liveness debt). Checkpoint maintenance runs in the background with an atomic per-project in-flight guard and commits under fence+lock only; prepare/search semantics unchanged. --- packages/opencode/src/memory/memory.ts | 174 ++++++++++++++----- packages/opencode/src/memory/model.ts | 19 +- packages/opencode/test/memory/memory.test.ts | 77 ++++++++ 3 files changed, 220 insertions(+), 50 deletions(-) diff --git a/packages/opencode/src/memory/memory.ts b/packages/opencode/src/memory/memory.ts index 6e081178ee..4a33e01c77 100644 --- a/packages/opencode/src/memory/memory.ts +++ b/packages/opencode/src/memory/memory.ts @@ -3,7 +3,7 @@ export * as Memory from "./memory" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { ProjectV2 } from "@opencode-ai/core/project" import { SessionV1 } from "@opencode-ai/core/v1/session" -import { Context, Duration, Effect, Layer, Option, Ref, Schema, Semaphore } from "effect" +import { Context, Effect, Layer, Option, Ref, Schema, Scope, Semaphore } from "effect" import { stringify } from "yaml" import { Config } from "@/config/config" import { Provider } from "@/provider/provider" @@ -22,8 +22,10 @@ import { MemoryStore } from "./store" const EVIDENCE_MESSAGES = 16 const EVIDENCE_CHARS = 8_000 -const PREPARE_TIMEOUT = Duration.seconds(5) -const CHECKPOINT_TIMEOUT = Duration.seconds(8) +// Reasoning-heavy models spend thinking tokens against max_output_tokens; size +// the budgets so the structured reply survives the thinking phase. +const MATCH_OUTPUT_TOKENS = 2_048 +const MAINTAIN_OUTPUT_TOKENS = 16_384 type TurnCache = { readonly completedTurns: number @@ -89,6 +91,8 @@ export const layer: Layer.Layer< const globalStarted = yield* Ref.make(false) const initializationLock = Semaphore.makeUnsafe(1) const state = yield* InstanceState.make(() => Effect.succeed({ sessions: new Map() })) + const scope = yield* Scope.Scope + const maintenanceInFlight = yield* Ref.make(new Set()) const availableModels = Effect.fn("Memory.availableModels")(function* () { const providers = yield* provider.list() @@ -265,7 +269,7 @@ export const layer: Layer.Layer< topics: MemoryStore.indexes(input.topics), }), schema: MemorySchema.MatchResponse, - maxOutputTokens: 256, + maxOutputTokens: MATCH_OUTPUT_TOKENS, }) const decoded = Schema.decodeUnknownOption(MemorySchema.MatchResponse)(output) if (Option.isNone(decoded)) @@ -276,15 +280,25 @@ export const layer: Layer.Layer< .slice(0, input.config.injection.max_topics) }) - const maintain = Effect.fn("Memory.maintain")(function* (input: { + // Serialize the identity-liveness recheck and the per-project lock around + // the store write only; the model calls that produce the update run + // outside the fence/lock so a long reasoning call cannot wedge or leak it. + const applyUpdate = (projectID: ProjectV2.ID, update: (topics: MemorySchema.Topic[]) => MemoryStore.Update) => + fence.withLiveIdentity( + projectID, + lock.withProject(projectID)(store.updateTopics(projectID, update)), + ) + + // Model-only half of maintenance: evidence → inspect match → maintenance + // proposal. Performs no persistence; callers own admission and the commit. + const proposeMaintenance = Effect.fn("Memory.proposeMaintenance")(function* (input: { model: Provider.Model config: MemorySchema.Config topics: MemorySchema.Topic[] messages: SessionV1.WithParts[] - projectID: Project.Info["id"] }) { const evidence = maintenanceEvidence(input.messages) - if (!evidence) return input.topics + if (!evidence) return const inspect = yield* match({ model: input.model, config: input.config, @@ -306,16 +320,28 @@ export const layer: Layer.Layer< }), }), schema: MemorySchema.MaintenanceResponse, - maxOutputTokens: 2_048, + maxOutputTokens: MAINTAIN_OUTPUT_TOKENS, }) const decoded = Schema.decodeUnknownOption(MemorySchema.MaintenanceResponse)(output) if (Option.isNone(decoded)) return yield* new ControllerError({ message: "MEMORY maintenance returned invalid output" }) + return decoded.value.actions + }) + + const maintain = Effect.fn("Memory.maintain")(function* (input: { + model: Provider.Model + config: MemorySchema.Config + topics: MemorySchema.Topic[] + messages: SessionV1.WithParts[] + projectID: Project.Info["id"] + }) { + const actions = yield* proposeMaintenance(input) + if (!actions) return input.topics return yield* store .updateTopics(input.projectID, (topics) => ({ applied: MemoryStore.applyActions({ topics, - actions: decoded.value.actions, + actions, topicLimit: input.config.topic_limit, }), result: undefined, @@ -343,6 +369,76 @@ export const layer: Layer.Layer< return renderSelection(selected, input.config) }) + // Self-contained background maintenance for the checkpoint path: the + // matcher and maintenance model run OUTSIDE the fence/lock, and only the + // topic commit acquires them (applyUpdate), so a long reasoning call + // cannot wedge the lock, leak it on interruption, or block the caller. + const backgroundMaintain = Effect.fn("Memory.backgroundMaintain")(function* (input: { + model: Provider.Model + config: MemorySchema.Config + messages: SessionV1.WithParts[] + projectID: ProjectV2.ID + }) { + const topics = yield* store.readTopics(input.projectID) + const actions = yield* proposeMaintenance({ + model: input.model, + config: input.config, + topics, + messages: input.messages, + }) + if (!actions) return + yield* applyUpdate(input.projectID, (current) => ({ + applied: MemoryStore.applyActions({ + topics: current, + actions, + topicLimit: input.config.topic_limit, + }), + result: undefined, + })) + }) + + const releaseMaintenanceSlot = (projectID: ProjectV2.ID) => + Ref.update(maintenanceInFlight, (set) => { + if (!set.has(projectID)) return set + const next = new Set(set) + next.delete(projectID) + return next + }) + + const kickMaintenance = Effect.fn("Memory.kickMaintenance")(function* (input: { + model: Provider.Model + config: MemorySchema.Config + messages: SessionV1.WithParts[] + projectID: ProjectV2.ID + }) { + const job = backgroundMaintain(input).pipe( + Effect.catchCause((cause) => Effect.logWarning("background MEMORY maintenance failed", { cause })), + Effect.ensuring(releaseMaintenanceSlot(input.projectID)), + ) + // Reserve and fork atomically: an interruption between the two would + // leak the in-flight slot and silently skip every later maintenance for + // this process; a fork into a closing scope must hand the slot back. + yield* Effect.uninterruptible( + Effect.gen(function* () { + const reserved = yield* Ref.modify(maintenanceInFlight, (set) => + set.has(input.projectID) + ? ([false, set] as const) + : ([true, new Set(set).add(input.projectID)] as const), + ) + if (!reserved) return + yield* job.pipe( + Effect.forkIn(scope), + Effect.catchCause((cause) => + Effect.gen(function* () { + yield* releaseMaintenanceSlot(input.projectID) + yield* Effect.logWarning("background MEMORY maintenance fork failed", { cause }) + }), + ), + ) + }), + ) + }) + const prepareUnsafe = Effect.fn("Memory.prepareUnsafe")(function* (input: { sessionID: SessionID messages: SessionV1.WithParts[] @@ -422,7 +518,6 @@ export const layer: Layer.Layer< const prepare: Interface["prepare"] = Effect.fn("Memory.prepare")((input) => prepareUnsafe(input).pipe( - Effect.timeout(PREPARE_TIMEOUT), Effect.catchCause((cause) => Effect.logWarning("MEMORY prepare failed", { cause })), ), ) @@ -525,7 +620,6 @@ export const layer: Layer.Layer< const search: Interface["search"] = Effect.fn("Memory.search")((input) => searchUnsafe(input).pipe( - Effect.timeout(PREPARE_TIMEOUT), Effect.catchCause((cause) => Effect.gen(function* () { yield* Effect.logWarning("MEMORY search failed", { cause }) @@ -545,54 +639,40 @@ export const layer: Layer.Layer< return [] } const user = latestRealUser(input.messages) - // Cross-process identity guard: a concurrent upgrade may retire this - // identity (row deleted, Home renamed away) while this write is in - // flight. Serialize on the identity lock and re-check liveness inside it; - // writing after retirement would re-create the retired Home and orphan - // the new content permanently (the identity cache already points at the - // successor, so no migration would ever run for this pair again). const live = yield* fence.withLiveIdentity( current.project.id, - Effect.gen(function* () { - return yield* lock.withProject(current.project.id)( - Effect.gen(function* () { - const topics = yield* store.readTopics(current.project.id) - const maintained = yield* maintain({ - model: current.model, - config: current.loaded.config, - topics, - messages: input.messages, - projectID: current.project.id, - }).pipe( - Effect.catchCause((cause) => - Effect.gen(function* () { - yield* Effect.logWarning("pre-compaction MEMORY maintenance failed", { cause }) - return topics - }), - ), - ) - const rendered = (yield* select({ - model: current.model, - config: current.loaded.config, - topics: maintained, - text: user?.text ?? "", - projectID: current.project.id, - })).rendered - return rendered - }), - ) - }), + lock.withProject(current.project.id)( + Effect.gen(function* () { + const topics = yield* store.readTopics(current.project.id) + return (yield* select({ + model: current.model, + config: current.loaded.config, + topics, + text: user?.text ?? "", + projectID: current.project.id, + })).rendered + }), + ), ) if (Option.isNone(live)) { yield* clearSession(input.sessionID) return [] } + // Maintenance runs in the background AFTER the identity fence: compaction + // must not wait on a long reasoning call, a retired identity never burns + // model calls, and the injection above rendered the pre-maintenance + // topics. At most one job per project is in flight. + yield* kickMaintenance({ + model: current.model, + config: current.loaded.config, + messages: input.messages, + projectID: current.project.id, + }) return live.value }) const checkpoint: Interface["checkpoint"] = Effect.fn("Memory.checkpoint")((input) => checkpointUnsafe(input).pipe( - Effect.timeout(CHECKPOINT_TIMEOUT), Effect.catchCause((cause) => Effect.gen(function* () { yield* Effect.logWarning("MEMORY checkpoint failed", { cause }) diff --git a/packages/opencode/src/memory/model.ts b/packages/opencode/src/memory/model.ts index c328460387..88c7bcee5e 100644 --- a/packages/opencode/src/memory/model.ts +++ b/packages/opencode/src/memory/model.ts @@ -5,7 +5,13 @@ import { Context, Duration, Effect, Layer, Schema } from "effect" import { generateObject } from "ai" import { Provider } from "@/provider/provider" -const DEFAULT_TIMEOUT = Duration.seconds(8) +// 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) + +const JSON_HINT = "Respond with a JSON object matching the provided schema." export interface Request { readonly model: Provider.Model @@ -43,9 +49,9 @@ export function make(input: { }) { return Service.of({ generate: Effect.fn("MemoryModel.generate")((request) => - input.execute(request).pipe( + input.execute(requireJsonToken(request)).pipe( Effect.timeoutOrElse({ - duration: input.timeout ?? DEFAULT_TIMEOUT, + duration: input.timeout ?? RESPONSE_TIMEOUT, orElse: () => Effect.fail(new TimeoutError()), }), ), @@ -53,6 +59,13 @@ export function make(input: { }) } +// Providers serving response_format json_object reject prompts that do not +// contain the literal word "json"; the maintenance prompts never mention it. +function requireJsonToken(request: Request): Request { + if (/json/i.test(request.system) || /json/i.test(request.prompt)) return request + return { ...request, system: `${request.system}\n${JSON_HINT}` } +} + export const layer = Layer.effect( Service, Effect.gen(function* () { diff --git a/packages/opencode/test/memory/memory.test.ts b/packages/opencode/test/memory/memory.test.ts index 40c0f0105b..49d455f2c0 100644 --- a/packages/opencode/test/memory/memory.test.ts +++ b/packages/opencode/test/memory/memory.test.ts @@ -313,6 +313,7 @@ function recallFixture() { topics: MemorySchema.Topic[] failQueries: Set maintenance: number + budgets: number[] config: MemorySchema.Config projectInitialized: number matcher?: (query: string) => Effect.Effect @@ -322,6 +323,7 @@ function recallFixture() { topics: [topic()], failQueries: new Set(), maintenance: 0, + budgets: [], config, projectInitialized: 1, } @@ -349,6 +351,7 @@ function recallFixture() { Layer.mock(MemoryModel.Service, { generate: (input) => Effect.gen(function* () { + state.budgets.push(input.maxOutputTokens) if (input.system === MemoryPrompts.MATCH_SYSTEM) { const request: unknown = JSON.parse(input.prompt) const query = @@ -403,6 +406,7 @@ function recallFixture() { state.topics = [topic()] state.failQueries.clear() state.maintenance = 0 + state.budgets.length = 0 state.config = config state.projectInitialized = 1 state.matcher = undefined @@ -1375,6 +1379,79 @@ describe("memory hidden model", () => { expect(interrupted).toBe(true) }), ) + + it.live("guarantees the json token reaches the model and leaves json-aware prompts untouched", () => + Effect.gen(function* () { + const seen: MemoryModel.Request[] = [] + const service = MemoryModel.make({ + execute: (request) => + Effect.sync(() => { + seen.push(request) + return {} + }), + }) + + yield* service.generate({ + model: ProviderTest.model(), + system: "Select relevant topics.", + prompt: "plain evidence without the token", + schema: MemorySchema.MatchResponse, + maxOutputTokens: 32, + }) + yield* service.generate({ + model: ProviderTest.model(), + system: "Propose updates as a JSON object.", + prompt: "evidence", + schema: MemorySchema.MaintenanceResponse, + maxOutputTokens: 32, + }) + + expect(seen[0].system).toContain("JSON") + expect(seen[1].system).toBe("Propose updates as a JSON object.") + }), + ) +}) + +describe("memory maintenance budgets", () => { + const recall = recallFixture() + + recall.it.instance( + "sizes match and maintenance output for reasoning-heavy models during checkpoint", + () => + Effect.gen(function* () { + recall.reset() + const memory = yield* Memory.Service + const sessionID = SessionID.make("ses_memory_budget") + const userID = MessageID.ascending() + const messages: SessionV1.WithParts[] = [ + user(userID, sessionID, "确认一条长期偏好：回复保持简洁"), + { + info: assistant(userID, sessionID, ProviderV2.ID.make("test"), ModelV2.ID.make("test-model"), "end_turn"), + parts: [], + }, + ] + + yield* memory.checkpoint({ sessionID, messages }) + // Checkpoint select runs synchronously; the kicked maintenance job + // completes in the background (its inspect-match + maintain calls). + yield* pollWithTimeout( + Effect.sync(() => (recall.state.budgets.length === 3 ? (true as const) : undefined)), + "background maintenance never completed", + ) + expect([...recall.state.budgets].sort((a, b) => a - b)).toEqual([2_048, 2_048, 16_384]) + // A second checkpoint must run a second maintenance: the in-flight + // slot is released when the first job finishes, never wedged. + yield* pollWithTimeout( + Effect.gen(function* () { + if (recall.state.budgets.filter((budget) => budget === 16_384).length >= 2) return true as const + yield* memory.checkpoint({ sessionID, messages }) + return undefined + }), + "second background maintenance never ran — in-flight slot wedged", + ) + }), + { git: true }, + ) }) describe("memory bootstrap", () => {