diff --git a/docs/findings/memory-batch-findings.md b/docs/findings/memory-batch-findings.md new file mode 100644 index 000000000..426b1beb0 --- /dev/null +++ b/docs/findings/memory-batch-findings.md @@ -0,0 +1,79 @@ +# MEMORY 批次（验收遗留）Findings Register + +- 验收 primary source：`docs/audit-dag-memory-goal-2026-08-18.md`（MEMORY 章节）+ 产品→代码验收（f1c2c8c33→11cfafe9c）Spec 轴三项遗留 +- 分支：`fix/memory-fence-scope` → PR `dev` +- 收敛判据：连续两轮独立审阅（Spec 镜 + Standards 镜）零 findings + 模块门禁全绿 +- 规格：`workflows/audit-fix-loop.md` + +## 处置项 + +| ID | 来源 | 处置 | 状态 | 提交 | +|---|---|---|---|---| +| MEM-01 后半（P1） | 审计 + 验收 (a)2 | prepareUnsafe `shouldMatch` 分支的 select matcher 移出 fence/lock：matcher 无锁跑，markMatched 经 `applyUpdate`（fence+lock 只包提交） | 完成（红-绿-变异通过） | 70291fbe4 | +| MEM-02（P2） | 审计 + 验收 (a)1 | `search` 的 identity fence 缩到 markMatched 提交；同查询合并从「持锁阻塞后来者」改为 turn 内 per-(session,turn,key) in-flight coalescing（进程内 Deferred），语义等价（后来者 reused:true、不耗 slot；失败/中断/retired 时唤醒降级而非挂起）且不再跨模型调用持 fence/lock | 完成（红-绿-变异通过；旧 coalescing 回归保持绿） | 70291fbe4 + f08548d6d + ef8d6b8d1 | +| MEM-03（P3） | 验收 (a)3 | 处置记录：已被 PR #333 的 MEM-01 重构结构性抵消——维护恒为后台（kickMaintenance），失败折入其 catchCause，维护前渲染是已声明设计（CONTEXT.md「render the pre-maintenance snapshot」）。旧的「维护前快照渲染注入」窗口随 inline maintain 一起消失。本行即处置记录。 | 已记录 | — | + +## 设计要点 + +- `applyUpdate` 泛型化：`Update` 透传结果，select 的 markMatched 提交经它返回 matched topics（供 render）。 +- `select` 拆两半：`match()`（无 fence/lock）+ markMatched 提交（`applyUpdate`）。 +- `search`：短临界区（进程内 lock）只做缓存 re-read/limit/queryCount++ 与 in-flight 登记；matcher 在任何 fence/lock 之外；同 key 后来者 await in-flight Deferred（进程内合并，语义与旧「锁内阻塞」等价）；提交走 `applyUpdate`。 +- `prepareUnsafe` shouldMatch 分支：matcher 出 fence/lock；identity retired（applyUpdate 返回 None）时仍 clearSession（fail-closed 不变）。 +- CONTEXT.md 最后一条 invariant 改写：fence 只包提交；同查询合并显式声明为进程内 in-flight coalescing。 +- 锁序不变量：全程不出现 lock 内嵌 fence（KeyedMutex 不可重入 + 与 checkpoint 的 fence>lock 序相反会死锁）。 + +## 模块门禁 +- 未开始 + +## 审阅轮次 + +### Round 1 +- Spec 镜：**BLOCKING**（R-1 P1 + R-2/R-3 INFO）；Standards 镜：**BLOCKING**（F1 P1 与 R-1 同源 + F2-F5 INFO）。处置： + - R-1/F1（P1）：in-flight 泄漏——失败/中断/retired 路径 deferred 永不完成 → coalesced awaiter 永久挂起、(session,key) 进程级 wedge。→ **已修**：runner 分支 `Effect.onExit`（对齐 kickMaintenance 槽位纪律）——每个退出路径先 `releaseIfOwner` 再把真实 Exit 打包进 deferred（deferred 永不失败，Exit 载荷即全部消息）；awaiter 按 Exit 分支：失败→`failed`、interrupt→failCause 传播、retired→`unavailable`。新增红绿测试「a failed first query never wedges the session key or its coalesced awaiter」（有界等待断言无挂起、无 wedge）+ 变异验证（去掉 onExit → 新测试与旧 coalescing 测试双红）。 + - R-2（INFO）：dereg→缓存写入窗口内第三个同查询会多耗一次调用。→ 已修：缓存写入提前到 `Effect.tap`（releaseIfOwner 之前），窗口闭合。 + - R-3（INFO）：等价声明只覆盖 happy path。→ CONTEXT.md 措辞已含失败路径降级语义。 + - F2（INFO）：CONTEXT.md「outside every fence/lock」对短注册临界区不真。→ 已改为「outside every fence + SHORT project-lock critical section (registration)」。 + - F3/F4/F5（INFO）：裸块、`!` 断言、命名不一致。→ 已修（裸块展开、get-then-check、统一 `selected`）。 +- 结论：修复后进入 Round 2。 + +### Round 2 +- Spec 镜：**BLOCKING**（Issue 1 P1 + Issue 2 INFO）；Standards 镜：**BLOCKING**（Issue 1 P1 同源 + 2/3/4 INFO）。处置： + - Issue 1（P1）：onExit 括号只覆盖 select 管道——注册后的 `store.readTopics` 挂起（生产为 flock 磁盘 IO）期间被中断/失败会在括号附着前 unwind → 泄漏条目 + wedge。→ **已修**：整个尾部（readTopics + select 管道）包进同一 `Effect.gen(...).pipe(tap, onExit, exit)` 括号；mock 的 `readTopics` 加 `parkReads` 挂起钩子；新增红绿测试「an interrupted topics read releases the in-flight entry and never wedges the key」+ 变异验证（readTopics 挪出括号 → 翻红）。 + - Spec Issue 2（INFO）：in-flight key 无 turn 分量 → 新轮次的同文查询可能骑上一轮的 deferred，attached 但自身缓存不填充。→ **已修**：key 加入 turn origin（messageID）——合并严格 turn 内，跨轮重跑。 + - Standards 2（INFO）：awaiter 注释的 interrupt 机制描述不准（failCause 重抛在 awaiter，failed 映射在 search wrapper）。→ 已改写。 + - Standards 3（INFO）：测试注释宣称 interrupted 覆盖但原先无此测试。→ R2 新增的中断测试已补齐该覆盖。 + - Standards 4（INFO）：`undefined as void` 多余 cast。→ 已删。 +- 结论：修复后进入 Round 3。 + +### Round 3 +- Spec 镜：**PASS**，1 条 INFO；Standards 镜：**PASS**，2 条 INFO。处置： + - R3-1（INFO）：CONTEXT.md 与 searchUnsafe 注释仍写 per-(session,key)，key 已 turn-scoped。→ 两处已改为 per-(session,turn,key)。 + - R3-2（INFO）：awaiter 注释「shares the cancellation」不符合 v4 语义（failCause 重抛由 wrapper catchCause 吸收，awaiter 仍以 failed 完成）。→ 已改写为准确机制。 + - register「待提交」歧义。→ 已改为实际提交哈希。 +- 结论：非干净轮（INFO）。修复后进入 Round 4。 + +### Round 4 +- Spec 镜：**PASS，no findings**（干净轮候选）。 +- Standards 镜：**PASS**，2 条 INFO：R4-1 awaiter 注释把非中断失败的映射错归 wrapper catchCause（实际直接返回 failed）；R4-2 测试注释「flock'd disk I/O」不准（readTopics 无 flock，是 async fs）。→ 均已改写。 +- 结论：非干净轮。修复后进入 Round 5。 + +### Round 5 +- Spec 镜：**PASS，no findings**；Standards 镜：**PASS，no findings**（R4 两处措辞独立复核为准确）。 +- 结论：**干净轮 2/2**。连续两轮零 findings → **MEMORY 验收遗留批次收敛**。 + +## 收敛结论 + +R1（双镜 BLOCKING：in-flight 泄漏 P1）→ R2（双镜 BLOCKING：括号窗口 P1 + turn 作用域）→ R3（双 PASS，3 INFO 措辞）→ R4（Spec 干净 + Standards 2 INFO 措辞）→ **R4+R5 连续两轮零 findings**。findings 轨迹：两轮 P1 并发缺陷（实装修复+回归测试）→ 纯措辞 → 零。 + +## 模块门禁（终态） + +- memory 测试簇 95/95 绿（新增 5 条回归：fence-free ×2、failed-wedge、interrupted-read-wedge、原 coalescing 保持） +- `bun typecheck` + pre-push turbo 29/29 绿 +- 全量套件 4164 tests：仅 2 失败为历次批次已在干净基线证实的 darwin 环境既有失败（help-snapshots、project-copy），与本批无因果 +- 变异验证：fence 回置翻红 ×2（search/prepare）、onExit 移除翻红 ×2（wedge + coalescing）、readTopics 出括号翻红 ×1 + +## 交付 + +- 分支：`fix/memory-fence-scope`（基于 origin/dev 11cfafe9c） +- 提交链：70291fbe4（MEM-01/02 主体）→ f08548d6d（R1 exit-safe）→ ef8d6b8d1（R2 整尾括号 + turn key）→ 9094abcaa/04385af79/0c1919465（记账与措辞） +- PR → dev（Typecheck 门禁） diff --git a/packages/opencode/src/memory/CONTEXT.md b/packages/opencode/src/memory/CONTEXT.md index 450cf6a30..140467d01 100644 --- a/packages/opencode/src/memory/CONTEXT.md +++ b/packages/opencode/src/memory/CONTEXT.md @@ -52,7 +52,7 @@ The domain runs on the existing seams; the elaborate `ProjectMemoryAuthority` re - Legacy files are re-read and compared immediately before deletion; content that changed after the scan is preserved and surfaced as a conflict. - Every writer of a MEMORY config file serializes on the file's cross-process lock; byte-atomicity is not undermined by whole-document last-writer-wins. - Maintenance model calls never run under the identity fence or the project lock: prepare and checkpoint render the pre-maintenance snapshot, then kick maintenance in the background, gated on identity liveness so a retired identity never starts a job (one job in flight per project, the commit-only write back under the fence). -- Bounded matcher calls deliberately hold the fence across one model call: the search matcher to coalesce concurrent identical queries, the prepare and checkpoint matchers because their match result feeds an atomic read-match-write under the project lock. Only unbounded-class work (maintenance) is excluded from the fence; a bounded matcher is at most one call per fence acquisition. +- Matcher model calls never run under the identity fence either (issue #324 acceptance): search, prepare, and checkpoint run their matcher outside every fence, and only the markMatched commit acquires the fence (`applyUpdate`). Search keeps a SHORT project-lock critical section (stale/cache/limit check + in-flight registration); concurrent identical queries within one turn coalesce through a process-local per-(session,turn,key) in-flight Deferred — the second caller awaits the first caller's exit (reused, no extra query slot) and degrades to `failed`/`unavailable` when the first call fails or its identity retires, instead of blocking a lock across the model call. ## Boundaries diff --git a/packages/opencode/src/memory/memory.ts b/packages/opencode/src/memory/memory.ts index 564a5de77..89b2f99da 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, Effect, Layer, Option, Ref, Schema, Scope, Semaphore } from "effect" +import { Cause, Context, Deferred, Effect, Exit, Layer, Option, Ref, Schema, Scope, Semaphore } from "effect" import { stringify } from "yaml" import { Config } from "@/config/config" import { Provider } from "@/provider/provider" @@ -93,7 +93,23 @@ export const layer: Layer.Layer< const state = yield* InstanceState.make(() => Effect.succeed({ sessions: new Map() })) const scope = yield* Scope.Scope const maintenanceInFlight = yield* Ref.make(new Set()) - + // MEM-02: per-(session,turn,key) in-flight matcher registrations — the + // turn origin (messageID) in the key keeps coalescing turn-scoped, so a + // new turn's identical query re-runs instead of riding a previous turn's + // result that could never populate its cache. Replaces the old "hold the + // fence/lock across the model call so the second caller blocks and + // re-reads the cache" coalescing: the second identical query now awaits + // the first one's Deferred instead — same observable semantics (reused: + // true, no extra query slot) without a model call under the fence/lock. + // The runner brackets EVERYTHING after registration in an exit guard + // (onExit), so a failed or interrupted first call wakes its coalesced + // awaiter (degraded to "failed") instead of parking it forever. + // Process-local by design (the fence is the cross-process seam, and it + // now covers only the markMatched commit). + type MatchRun = { count: number; rendered: string[] } + const matchInFlight = yield* Ref.make( + new Map>>(), + ) const availableModels = Effect.fn("Memory.availableModels")(function* () { const providers = yield* provider.list() return new Set( @@ -283,7 +299,12 @@ export const layer: Layer.Layer< // 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) => + // The update callback's result is passed through, so a commit like + // markMatched can hand the caller the post-commit topics to render. + const applyUpdate = ( + projectID: ProjectV2.ID, + update: (topics: MemorySchema.Topic[]) => MemoryStore.Update, + ) => fence.withLiveIdentity( projectID, lock.withProject(projectID)(store.updateTopics(projectID, update)), @@ -328,6 +349,9 @@ export const layer: Layer.Layer< return decoded.value.actions }) + // MEM-01/02: the matcher model call runs OUTSIDE the fence/lock; only the + // markMatched commit acquires them (applyUpdate). The matched topics come + // back from the commit for rendering. const select = Effect.fn("Memory.select")(function* (input: { model: Provider.Model config: MemorySchema.Config @@ -336,11 +360,20 @@ export const layer: Layer.Layer< projectID: Project.Info["id"] }) { const topicIDs = yield* match(input) - const matched = yield* store.updateTopics(input.projectID, (topics) => ({ - applied: MemoryStore.markMatched(topics, topicIDs), - result: undefined, - })) - const byID = new Map(matched.topics.map((topic) => [topic.id, topic])) + const committed = yield* applyUpdate(input.projectID, (topics) => { + // Re-filter against the post-read topics: the matcher filtered on the + // snapshot it saw; a topic deleted since then must not resurrect. + const live = new Set(topics.map((topic) => topic.id)) + return { + applied: MemoryStore.markMatched(topics, topicIDs.filter((id) => live.has(id))), + result: undefined, + } + }) + // Identity retired between the model call and the commit: nothing was + // written; there is no matched set to render. + if (Option.isNone(committed)) return undefined + const matched = committed.value.topics + const byID = new Map(matched.map((topic) => [topic.id, topic])) const selected = topicIDs.flatMap((id) => { const topic = byID.get(id) return topic ? [topic] : [] @@ -470,33 +503,28 @@ export const layer: Layer.Layer< return } - // The fence and the project lock cover the topic read plus the bounded - // first-turn matcher (declared tradeoff, see CONTEXT.md). Due maintenance - // is kicked AFTER the fence releases, so a long reasoning call never - // holds it: this turn renders the pre-maintenance topics and the - // committed update surfaces on a later prepare. - const live = yield* fence.withLiveIdentity( - current.project.id, - lock.withProject(current.project.id)( - Effect.gen(function* () { - const topics = yield* store.readTopics(current.project.id) - const rendered = (yield* select({ - model: current.model, - config: current.loaded.config, - topics, - text: user.text, - projectID: current.project.id, - })).rendered - const entry = data.sessions.get(input.sessionID) - if (entry?.turn.messageID !== user.info.id) return - entry.turn = { ...entry.turn, completedTurns: turns, rendered } - }), - ), - ) - if (Option.isNone(live)) { + // MEM-01: the first-turn matcher runs OUTSIDE the fence/lock; only its + // markMatched commit acquires them (inside select → applyUpdate). An + // identity retired mid-call surfaces as select === undefined — fail + // closed by dropping the cached session state. Due maintenance is + // kicked AFTERwards, so a long reasoning call never holds the fence: + // this turn renders the pre-maintenance topics and the committed update + // surfaces on a later prepare. + const selected = yield* select({ + model: current.model, + config: current.loaded.config, + topics: yield* store.readTopics(current.project.id), + text: user.text, + projectID: current.project.id, + }) + if (!selected) { yield* clearSession(input.sessionID) return } + const entry = data.sessions.get(input.sessionID) + if (entry?.turn.messageID === user.info.id) { + entry.turn = { ...entry.turn, completedTurns: turns, rendered: selected.rendered } + } if (!due) return if (Option.isNone(yield* kickMaintenance(maintenance))) yield* clearSession(input.sessionID) }) @@ -559,51 +587,123 @@ export const layer: Layer.Layer< } const origin = user.info.id - // Declared tradeoff (issue #324, see CONTEXT.md): unlike maintenance, the - // matcher model call runs INSIDE the fence/lock. That serialization is - // what coalesces concurrent identical queries — the second caller blocks, - // re-reads `queries` under the lock, and reuses the first result instead - // of spending another model call. The lock also covers markMatched. - const live = yield* fence.withLiveIdentity( - current.project.id, + // MEM-02 (issue #324 acceptance): the matcher model call runs OUTSIDE + // the fence/lock. Concurrent identical queries coalesce through the + // per-(session,turn,key) in-flight Deferred instead of lock-blocking: + // the second caller re-checks the cache under a SHORT project-lock + // critical section (stale/cache/limit check + registration), awaits + // the first caller's result, and reports reused without spending + // another model call or query slot. Only the markMatched commit + // (inside select → applyUpdate) acquires the fence. + const inFlightKey = `${input.sessionID}\0${origin}\0${key}` + const deferred = yield* Deferred.make>() + + const releaseIfOwner = Effect.fnUntraced(function* () { + yield* Ref.update(matchInFlight, (map) => { + if (map.get(inFlightKey) !== deferred) return map + const next = new Map(map) + next.delete(inFlightKey) + return next + }) + }) + + const outcome = yield* lock.withProject(current.project.id)( Effect.gen(function* () { - return yield* lock.withProject(current.project.id)( - Effect.gen(function* () { - const activeTurn = data.sessions.get(input.sessionID)?.turn - if (activeTurn?.messageID !== origin) return { status: "stale" as const } - const repeated = activeTurn.queries.get(key) - if (repeated) { - activeTurn.rendered = repeated.rendered - return repeated.count > 0 - ? { status: "attached" as const, count: repeated.count, reused: true } - : { status: "empty" as const, reused: true } - } - if (activeTurn.queryCount >= 2) return { status: "limit" as const } - activeTurn.queryCount++ - const topics = yield* store.readTopics(current.project.id) - const selected = yield* select({ - model: current.model, - config: current.loaded.config, - topics, - text: query, - projectID: current.project.id, - }) - const latest = data.sessions.get(input.sessionID)?.turn - if (latest?.messageID !== origin) return { status: "stale" as const } - latest.queries.set(key, selected) - latest.rendered = selected.rendered - return selected.count > 0 - ? { status: "attached" as const, count: selected.count, reused: false } - : { status: "empty" as const, reused: false } - }), - ) + const activeTurn = data.sessions.get(input.sessionID)?.turn + if (activeTurn?.messageID !== origin) return { status: "stale" as const } + const repeated = activeTurn.queries.get(key) + if (repeated) { + activeTurn.rendered = repeated.rendered + return repeated.count > 0 + ? { status: "attached" as const, count: repeated.count, reused: true } + : { status: "empty" as const, reused: true } + } + if (activeTurn.queryCount >= 2) return { status: "limit" as const } + const running = yield* Ref.modify(matchInFlight, (map) => { + const existing = map.get(inFlightKey) + if (existing) return [existing, map] as const + return [deferred, new Map(map).set(inFlightKey, deferred)] as const + }) + if (running !== deferred) return { kind: "await-first" as const, first: running } + activeTurn.queryCount++ + return { kind: "run" as const } }), ) - if (Option.isNone(live)) { - yield* clearSession(input.sessionID) - return { status: "unavailable" as const } + if ("kind" in outcome) { + if (outcome.kind === "await-first") { + // The awaiter rides the runner's exit: the runner's exit bracket + // packs every outcome — success, failure, interrupt, retired — + // into the deferred payload, so the await always wakes. Failures + // surface as this caller's "failed" (returned directly here); + // the runner's interrupt cause is re-raised via failCause so the + // wrapper's log carries it — the awaiter itself still completes + // with "failed". Never a permanent park. + const first = yield* Deferred.await(outcome.first) + if (Exit.isFailure(first)) { + if (Cause.hasInterrupts(first.cause)) return yield* Effect.failCause(first.cause) + return { status: "failed" as const } + } + const selected = first.value + if (!selected) return { status: "unavailable" as const } + const latest = data.sessions.get(input.sessionID)?.turn + if (latest?.messageID !== origin) return { status: "stale" as const } + return selected.count > 0 + ? { status: "attached" as const, count: selected.count, reused: true } + : { status: "empty" as const, reused: true } + } + // This caller owns the matcher run. EVERYTHING after registration — + // the topics read and the select pipeline (model call, fenced + // markMatched commit) — runs inside one exit bracket: onExit fires + // on success, failure, interrupt, and identity-retired alike, + // deregistering the map entry and packing the real exit into the + // deferred so a coalesced awaiter wakes instead of parking forever + // (kickMaintenance's slot discipline, applied from the moment the + // entry exists — an interrupt during the topics read would otherwise + // unwind before the bracket attaches and wedge the (turn,key) + // forever). The deferred itself never fails — the Exit payload is + // the whole message. + const runExit = yield* Effect.gen(function* () { + return yield* select({ + model: current.model, + config: current.loaded.config, + topics: yield* store.readTopics(current.project.id), + text: query, + projectID: current.project.id, + }) + }).pipe( + Effect.tap((selected) => { + // Publish the cache entry BEFORE the in-flight deregistration in + // onExit: a third identical caller entering between the two + // would otherwise miss both the cache and the in-flight entry + // and burn a second model call + query slot where the old + // lock-blocking design guaranteed reuse. Stale-origin runs skip + // the write; the stale check below still governs the response. + if (!selected) return Effect.void + const latest = data.sessions.get(input.sessionID)?.turn + if (latest?.messageID !== origin) return Effect.void + latest.queries.set(key, selected) + latest.rendered = selected.rendered + return Effect.void + }), + Effect.onExit((exit) => releaseIfOwner().pipe(Effect.andThen(Deferred.succeed(deferred, exit)))), + Effect.exit, + ) + if (Exit.isFailure(runExit)) return yield* Effect.failCause(runExit.cause) + const selected = runExit.value + // Identity retired between model call and commit — fail closed. The + // deferred already carries the same (succeeded-undefined) exit, so a + // coalesced awaiter degrades to "unavailable" rather than hanging. + if (!selected) { + yield* clearSession(input.sessionID) + return { status: "unavailable" as const } + } + const latest = data.sessions.get(input.sessionID)?.turn + if (latest?.messageID !== origin) return { status: "stale" as const } + return selected.count > 0 + ? { status: "attached" as const, count: selected.count, reused: false } + : { status: "empty" as const, reused: false } } - return live.value + return outcome }) const search: Interface["search"] = Effect.fn("Memory.search")((input) => @@ -627,22 +727,18 @@ export const layer: Layer.Layer< return [] } const user = latestRealUser(input.messages) - const live = yield* fence.withLiveIdentity( - current.project.id, - 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)) { + // MEM-01: the render matcher runs OUTSIDE the fence/lock; its + // markMatched commit is fenced inside select → applyUpdate. An identity + // retired mid-call (select === undefined) fails closed to an empty + // render, same as the retired-fence outcome before. + const selected = yield* select({ + model: current.model, + config: current.loaded.config, + topics: yield* store.readTopics(current.project.id), + text: user?.text ?? "", + projectID: current.project.id, + }) + if (!selected) { yield* clearSession(input.sessionID) return [] } @@ -657,7 +753,7 @@ export const layer: Layer.Layer< projectID: current.project.id, }) if (Option.isNone(kicked)) yield* clearSession(input.sessionID) - return live.value + return selected.rendered }) const checkpoint: Interface["checkpoint"] = Effect.fn("Memory.checkpoint")((input) => diff --git a/packages/opencode/test/memory/memory.test.ts b/packages/opencode/test/memory/memory.test.ts index 182503e44..9b711d389 100644 --- a/packages/opencode/test/memory/memory.test.ts +++ b/packages/opencode/test/memory/memory.test.ts @@ -318,6 +318,8 @@ function recallFixture() { projectInitialized: number matcher?: (query: string) => Effect.Effect maintenanceHook?: () => Effect.Effect + /** Parks the runner's pre-select topics read (interrupt-window probe). */ + parkReads?: { started: Deferred.Deferred; release: Deferred.Deferred } } = { queries: [], reads: 0, @@ -376,8 +378,12 @@ function recallFixture() { MemoryLock.defaultLayer, Layer.mock(MemoryStore.Service, { readTopics: () => - Effect.sync(() => { + Effect.gen(function* () { state.reads++ + if (state.parkReads) { + yield* Deferred.succeed(state.parkReads.started, undefined) + yield* Deferred.await(state.parkReads.release) + } return state.topics }), updateTopics: (_projectID, update) => @@ -413,6 +419,7 @@ function recallFixture() { state.projectInitialized = 1 state.matcher = undefined state.maintenanceHook = undefined + state.parkReads = undefined }, it: testEffect(layer), systemIt: testEffect(systemLayer), @@ -1382,6 +1389,179 @@ describe("memory turn-scoped retrieval", () => { }), { git: true }, ) + + // MEM-02 follow-up (acceptance): the cross-process identity fence must not + // be held across the SEARCH matcher model call — only the markMatched + // commit is fenced. While the matcher is parked mid-call, a concurrent + // checkpoint (whose render select needs the same identity fence) must not + // starve. Under the old shape the fence wrapped the whole lock block, so + // the checkpoint waited on the streaming matcher. + recall.it.instance( + "keeps the identity fence free while the search matcher streams", + () => + Effect.gen(function* () { + recall.reset() + const started = yield* Deferred.make() + const release = yield* Deferred.make() + // Only the SLOW search query parks; the checkpoint's own render match + // must sail through — that is the assertion. + recall.state.matcher = (query) => + query === "慢架构查询" + ? Effect.gen(function* () { + yield* Deferred.succeed(started, undefined) + yield* Deferred.await(release) + return { topic_ids: [recall.state.topics[0]?.id ?? ""] } + }) + : Effect.succeed({ topic_ids: [recall.state.topics[0]?.id ?? ""] }) + const memory = yield* Memory.Service + const sessionID = SessionID.make("ses_memory_search_fence_free") + const messages = [ + user(MessageID.ascending(), sessionID, "先处理当前问题"), + user(MessageID.ascending(), sessionID, "召回相关历史"), + ] + + const pending = yield* memory.search({ sessionID, messages, query: "慢架构查询" }).pipe(Effect.forkChild) + yield* Deferred.await(started) + + const rendered = yield* awaitWithTimeout( + memory.checkpoint({ sessionID, messages }), + "checkpoint starved by the search matcher — fence held across the model call (MEM-02)", + ) + expect(rendered.length).toBeGreaterThan(0) + + yield* Deferred.succeed(release, undefined) + expect(yield* Fiber.join(pending)).toEqual({ status: "attached", count: 1, reused: false }) + }), + { git: true }, + ) + + // MEM-01 follow-up (acceptance): same discipline for the prepare + // first-turn (shouldMatch) branch — the bounded matcher moved out of the + // fence/lock, so a parked first-turn matcher cannot starve the fence. + recall.it.instance( + "keeps the identity fence free while the first-turn prepare matcher streams", + () => + Effect.gen(function* () { + recall.reset() + const started = yield* Deferred.make() + const release = yield* Deferred.make() + // Only the FIRST matcher call (the first-turn prepare) parks; the + // concurrent checkpoint's render match — same text, so query-based + // discrimination is impossible — must sail through on its own call. + let matcherCalls = 0 + recall.state.matcher = () => + Effect.gen(function* () { + matcherCalls++ + if (matcherCalls > 1) return { topic_ids: [recall.state.topics[0]?.id ?? ""] } + yield* Deferred.succeed(started, undefined) + yield* Deferred.await(release) + return { topic_ids: [recall.state.topics[0]?.id ?? ""] } + }) + const memory = yield* Memory.Service + const sessionID = SessionID.make("ses_memory_prepare_fence_free") + const messages = [user(MessageID.ascending(), sessionID, "首次真实用户输入关于架构")] + + const pending = yield* memory.prepare({ sessionID, messages }).pipe(Effect.forkChild) + yield* Deferred.await(started) + + const rendered = yield* awaitWithTimeout( + memory.checkpoint({ sessionID, messages }), + "checkpoint starved by the first-turn prepare matcher — fence held across the model call (MEM-01)", + ) + expect(rendered.length).toBeGreaterThan(0) + + yield* Deferred.succeed(release, undefined) + yield* Fiber.join(pending) + // Both the first-turn prepare and the concurrent checkpoint render + // matched the same text; each recorded exactly its own call. + expect(recall.state.queries).toEqual(["首次真实用户输入关于架构", "首次真实用户输入关于架构"]) + }), + { git: true }, + ) + + // Review R-1: the runner must complete the in-flight deferred on EVERY + // exit. A failed first matcher call must not wedge the (session,key): a + // coalesced awaiter wakes (degraded) and a later identical query re-runs + // the matcher instead of parking on a leaked in-flight entry. + recall.it.instance( + "a failed first query never wedges the session key or its coalesced awaiter", + () => + Effect.gen(function* () { + recall.reset() + const started = yield* Deferred.make() + let matcherCalls = 0 + recall.state.matcher = () => + Effect.gen(function* () { + matcherCalls++ + if (matcherCalls === 1) { + yield* Deferred.succeed(started, undefined) + throw new Error("matcher exploded") + } + return { topic_ids: [recall.state.topics[0]?.id ?? ""] } + }) + const memory = yield* Memory.Service + const sessionID = SessionID.make("ses_memory_failed_first") + const messages = [ + user(MessageID.ascending(), sessionID, "先处理当前问题"), + user(MessageID.ascending(), sessionID, "召回相关历史"), + ] + + const failing = yield* memory.search({ sessionID, messages, query: "易碎查询" }).pipe(Effect.forkChild) + yield* Deferred.await(started) + // Concurrent identical query — either it coalesced onto the failing + // run (wakes degraded to "failed") or it raced past the in-flight + // window and re-runs (attached). Both are non-hang outcomes; a + // parked-forever awaiter or a wedged key fails the bounded waits. + const coalesced = yield* memory.search({ sessionID, messages, query: "易碎查询" }).pipe(Effect.forkChild) + const failingResult = yield* awaitWithTimeout(Fiber.join(failing), "failing query hung") + const coalescedResult = yield* awaitWithTimeout(Fiber.join(coalesced), "coalesced awaiter hung on the failing run") + expect(failingResult).toEqual({ status: "failed" }) + expect(["failed", "attached", "empty"]).toContain(coalescedResult.status) + // The (session,key) is NOT wedged: the next identical query resolves + // within the bounded window (fresh run or cached reuse from the + // coalesced fiber's successful re-run — either proves liveness). + expect(yield* awaitWithTimeout(memory.search({ sessionID, messages, query: "易碎查询" }), "later identical query wedged on the leaked in-flight entry")).toMatchObject({ + status: "attached", + }) + }), + { git: true }, + ) + + // Review R2 issue 1: the exit bracket must start at REGISTRATION, not at + // the select pipeline — an interrupt during the runner's pre-select topics + // read (a real async fs suspension in production) previously unwound + // before onExit attached, leaking the in-flight entry and wedging the + // (turn,key) forever. + recall.it.instance( + "an interrupted topics read releases the in-flight entry and never wedges the key", + () => + Effect.gen(function* () { + recall.reset() + const memory = yield* Memory.Service + const sessionID = SessionID.make("ses_memory_interrupted_read") + const messages = [ + user(MessageID.ascending(), sessionID, "先处理当前问题"), + user(MessageID.ascending(), sessionID, "召回相关历史"), + ] + const started = yield* Deferred.make() + const release = yield* Deferred.make() + recall.state.parkReads = { started, release } + + const runner = yield* memory.search({ sessionID, messages, query: "可中断查询" }).pipe(Effect.forkChild) + yield* Deferred.await(started) + yield* Fiber.interrupt(runner).pipe(Effect.ignore) + + // The entry must be released despite the interrupt landing before + // the select pipeline attached its bracket: a later identical query + // resolves within the bounded window instead of parking forever + // (empty: the query text matches no topic — liveness is the point). + yield* Deferred.succeed(release, undefined) + expect(yield* awaitWithTimeout(memory.search({ sessionID, messages, query: "可中断查询" }), "later identical query wedged on the entry leaked by the interrupted read")).toMatchObject({ + status: "empty", + }) + }), + { git: true }, + ) }) describe("memory project config Git exclusions", () => {