diff --git a/dsh-mneme/lib/inject.js b/dsh-mneme/lib/inject.js index 014332c7..0613ad3a 100644 --- a/dsh-mneme/lib/inject.js +++ b/dsh-mneme/lib/inject.js @@ -44,13 +44,17 @@ function visibleSeqsOf(session) { // 候选问的是同一件事——「当时到底想干什么」；这条取原话的路只该有一份实现。 // 复制一份的代价本仓库付过几次（src/lib 平价、snapshotEvents 兼容层），所以 // 这里只加 export，不改行为。 -export function lastUserQuery(ctx) { +// +// #617：多带一个 seq——注入器要拿它判断「待认领票据是否比事件流更新」（见 +// createInjector 的 pendingQueries）。seq 是宿主事件序号，取不到时为 null +// （老版本宿主 / 测试夹具），调用方按 null 参与相等比较，不做真假推断。 +export function lastUserMessageInfo(ctx) { try { const session = ctx?.agent?.session; // DSH ≥0.1.2-rc only exposes events via snapshotEvents(); older builds // still have the .events property, so fall back to it. const events = session?.snapshotEvents?.() ?? session?.events; - if (!Array.isArray(events) || events.length === 0) return ""; + if (!Array.isArray(events) || events.length === 0) return { text: "", seq: null }; const visibleSeqs = visibleSeqsOf(session); for (let i = events.length - 1; i >= 0; i--) { const event = events[i]; @@ -61,16 +65,44 @@ export function lastUserQuery(ctx) { if (kind !== undefined && kind !== "user") continue; const parts = event.data?.content; if (!Array.isArray(parts) || parts.length === 0) continue; - return safeSlice( - parts - .map((p) => (typeof p === "string" ? p : p?.text ?? "")) - .filter(Boolean) - .join("\n"), - 500 - ); + return { + text: safeSlice( + parts + .map((p) => (typeof p === "string" ? p : p?.text ?? "")) + .filter(Boolean) + .join("\n"), + 500 + ), + seq: event?.seq ?? null + }; } } catch { /* session internals unavailable: degrade to no query */ } - return ""; + return { text: "", seq: null }; +} + +export function lastUserQuery(ctx) { + return lastUserMessageInfo(ctx).text; +} + +// 认领消息 → 查询文本。#617 的补位票据就是这里算出来的字符串：判定口径与 +// user/message 事件同源（content parts + source.kind），于是「插件注入的上下文」 +// 这类非人类消息照样被跳过，票据不会拿插件自己的话当用户问题。 +function inboxMessageQuery(message) { + try { + const parts = message?.content ?? message?.data?.content; + if (!Array.isArray(parts) || parts.length === 0) return ""; + const kind = message?.source?.kind ?? message?.data?.source?.kind; + if (kind !== undefined && kind !== "user") return ""; + return safeSlice( + parts + .map((p) => (typeof p === "string" ? p : p?.text ?? "")) + .filter(Boolean) + .join("\n"), + 500 + ); + } catch { + return ""; + } } // Hot-memory round extraction (v0.5.0 1.3): pairs each user/message with the @@ -291,21 +323,117 @@ export function createInjector(ctx, service, settings, config, gate = null, sess // first render after a new user message may still fall back to the rule-based // pick; later assemblies in the same session reuse the cached vector. Bounded // cache (cap 8, drop oldest) so a long session never grows it unbounded. + // + // #617 根因 B：旧实现用一个 lastPrefetched 字符串做去重，且它在 await 之前 + // 就置位——同一条查询第一次失败（模型没下好、超时）之后**永不重试**，而那 + // 恰好是最需要重试的一次。这里换成三件套：inflight Map 去重（渲染同一步会 + // 被求值两次：assemble 一次、refreshContext 一次，同一条查询只该嵌一次）、 + // 失败冷却 15s（失败可重试，但不许每次渲染都拿一次失败去撞模型）、返回 + // Promise（调用方/测试能等到它，而不是靠魔法延时）。 const QUERY_VECTOR_CACHE_MAX = 8; + const PREFETCH_FAIL_COOLDOWN_MS = 15000; const queryVectorCache = new Map(); - let lastPrefetched = ""; + const inflightPrefetches = new Map(); // query -> Promise + const prefetchFailedAt = new Map(); // query -> ms function prefetchQueryVector(query) { - if (!query || query === lastPrefetched || queryVectorCache.has(query)) return; - lastPrefetched = query; - service.embedQuery(query).then((vec) => { + if (!query) return null; + const cached = queryVectorCache.get(query); + if (cached) return Promise.resolve(cached); + const inflight = inflightPrefetches.get(query); + if (inflight) return inflight; + const failedAt = prefetchFailedAt.get(query); + if (failedAt !== undefined && Date.now() - failedAt < PREFETCH_FAIL_COOLDOWN_MS) return null; + const pending = service.embedQuery(query).then((vec) => { if (Array.isArray(vec) && vec.length) { queryVectorCache.set(query, vec); if (queryVectorCache.size > QUERY_VECTOR_CACHE_MAX) { queryVectorCache.delete(queryVectorCache.keys().next().value); } + prefetchFailedAt.delete(query); + return vec; } - }).catch(() => { /* prefetch is best-effort */ }); + prefetchFailedAt.set(query, Date.now()); + return null; + }).catch(() => { + prefetchFailedAt.set(query, Date.now()); + return null; /* prefetch is best-effort */ + }).finally(() => { + inflightPrefetches.delete(query); + }); + inflightPrefetches.set(query, pending); + return pending; + } + + // #617 根因 A 的补位票据：宿主（agent-loop 0.2.x）在每个 step 的 pre-step 里 + // 先 inbox.claim()（同步 emit agent/inbox/claimed），再 assemble()，而 + // user/message 要到 step() 末尾（:865）才 append——注入的 text 回调在 + // assemble / refreshContext 里求值，**永远比当轮用户输入早一拍**：Step 1 拿到 + // ""，第 N 轮的 Step 1 只能拿到第 N-1 轮的旧文本（实测 46 个渲染点错 3 次）。 + // 票据在 claim 那一刻记下原文；渲染时若事件流里最后一条用户消息还不是它， + // 说明本轮 user/message 尚未落盘，用票据；一旦落盘，事件流才是权威（它对 + // rewind / compact 的可见面判定更准，票据不看 surface）。 + // + // 为什么只在 claimed 上记、不从 agent/inbox/spliced 推：claim 自己的两次 + // mutate 都落 outcome:"canceled" 的 spliced（inserted 为空），拿 removedCount + // 当撤回信号会把每次正常认领误判成取消；claimed 只在真有消息被认领时触发， + // 是唯一无歧义的信号源。Cancel/Rewind 穿透因此也不存在——它们不产生 claimed。 + const PENDING_QUERY_MAX = 32; + // 票据超过这个岁数就不再采信：一个 step 的 assemble 不可能跑这么久，更久说明 + // 那条消息压根没落盘（认领后整轮被取消），票据已是过期证据。 + const PENDING_QUERY_TTL_MS = 60000; + const pendingQueries = new Map(); // sessionId -> { text, at } + const offClaimed = typeof ctx?.on === "function" + ? ctx.on("agent/inbox/claimed", (payload) => { + try { + const message = payload?.message; + const text = inboxMessageQuery(message); + if (!text) return; + const session = payload?.agent?.session; + const sessionId = session?.id; + if (typeof sessionId !== "string" || !sessionId) return; + // #539 无痕闸门必须挡在这里，不能只挡渲染侧：claim 在 pre-step 里同步触发， + // 早于 assemble；这条路径既写票据（票据会变成注入查询文本）又预取向量 + // （embedQuery 直接 POST 外部 /embeddings），闸门命中时两者都是该会话数据的 + // 外流。嵌入层自身没有任何会话维度判断，这里是唯一的拦截点。 + if (sessionGate?.isOff(session)) return; + pendingQueries.delete(sessionId); // 重新插入 = 移到队尾 + pendingQueries.set(sessionId, { text, at: Date.now() }); + if (pendingQueries.size > PENDING_QUERY_MAX) { + pendingQueries.delete(pendingQueries.keys().next().value); + } + // 认领即预取：embed 在 assemble→step 之间就已经在跑，Step 1 的第二次 + // 求值（refreshContext）有机会拿到向量。拿不到也无妨——查询文本已经在 + // 手上，BM25 词法路（#198）能顶上。 + prefetchQueryVector(text); + } catch { /* the ticket is best-effort; injection must never break */ } + }) + : null; + + // 当轮查询：票据优先于事件流（见上）。判据是**文本比对**而不是事件条数—— + // 从 claim 到 assemble 之间还夹着 await tools.resolve(...)，这期间任何插件 + // 追加一个事件都会让条数增长，按条数判会把「消息还没落盘」误判成「已落盘」， + // 票据提前退休、Step 1 又退回空串（正是本 issue 要修的那一拍）。文本相等才是 + // 「这条消息真的进事件流了」的直接证据；连续两轮同一句话时两者相等，取哪个都对。 + function liveStepQuery(ctx) { + const info = lastUserMessageInfo(ctx); + let sessionId = null; + try { + const id = ctx?.agent?.session?.id; + if (typeof id === "string" && id) sessionId = id; + } catch { /* session internals unavailable */ } + if (!sessionId) return info.text; + const ticket = pendingQueries.get(sessionId); + if (!ticket) return info.text; + if (info.text === ticket.text) { + pendingQueries.delete(sessionId); // 已落盘，票据退休（顺带保证 Map 不被历史会话撑大） + return info.text; + } + if (Date.now() - ticket.at > PENDING_QUERY_TTL_MS) { + pendingQueries.delete(sessionId); // 过期证据，不采信 + return info.text; + } + return ticket.text; } // 会话身份解析（v1.2 人格寻址用）：读 session.header.agentPreset——冻结的 @@ -395,10 +523,12 @@ export function createInjector(ctx, service, settings, config, gate = null, sess text: (ctx) => { if (sessionGate?.isOff(sessionOf(ctx))) return ""; // Bug4: pass the latest user query so injection prefers semantically - // relevant memories; lastUserQuery is best-effort (empty → legacy). + // relevant memories; liveStepQuery is best-effort (empty → legacy). // The query vector is prefetched asynchronously (cached) because the // render itself must stay synchronous. - const query = lastUserQuery(ctx); + // #617：liveStepQuery 而不是 lastUserQuery——宿主在 step 末尾才 append + // user/message，而注入在这之前求值，只看事件流会拿到上一轮的输入。 + const query = liveStepQuery(ctx); if (query) prefetchQueryVector(query); const queryVector = queryVectorCache.get(query); // v0.8.0 A3：scope 随请求解析（解析器恒返回对象，身份取不到时两维为 @@ -493,6 +623,12 @@ export function createInjector(ctx, service, settings, config, gate = null, sess ]; return () => { + // #617：认领票据是监听宿主事件的活句柄，卸载时必须成对摘掉，否则注入器重装 + // （面板开关、宿主热重载）会留下一个继续写 Map 的孤儿监听。 + offClaimed?.(); + pendingQueries.clear(); + inflightPrefetches.clear(); + prefetchFailedAt.clear(); queryVectorCache.clear(); rotationHistory.clear(); injectionSnapshot = null; // 注入器卸载即失效：快照不得跨生命周期存留 diff --git a/dsh-mneme/lib/service.js b/dsh-mneme/lib/service.js index 01d3a47b..b8463659 100644 --- a/dsh-mneme/lib/service.js +++ b/dsh-mneme/lib/service.js @@ -532,6 +532,25 @@ export function createService({ store, mirror, config, onWrite, logger, document return dot / (Math.sqrt(na) * Math.sqrt(nb)); } + // BM25 语料索引缓存（#617 审计 O2）。此前每次渲染都 store.list 500 行 + 重新 + // tokenize 全语料，实测 4.318ms/次、每个 step 一次；其中 store.list 占 4.318ms + // 里的绝大部分，而语料在两次写入之间根本没变。 + // + // 失效键用 mirror_state.generation，不另建计数器：改正文/可见性的写入（insert / + // update / remove / setArchived / setForgotten / setContent …）都已经 bump 它， + // 而 last_accessed_at 与 embedding 这两类写入**刻意不 bump**（store.js 里写明 + // reads must not mark the mirror dirty）——这两类恰好也不改 title+content 语料， + // 正是我们要的失效粒度，不用额外判断。 + // + // 水位每次从库里现读，不在进程内记本地计数：多宿主共用一个 memoryDir 是支持形态 + // （busy_timeout + WAL），另一个进程写完 bump 的是库里的 generation，本地计数 + // 永远看不见，只有现读才收得到。 + // + // 只缓存索引、不缓存查询结果：search() 返回的是 {...row, score} 副本，索引里 + // 留着的原始行不会被调用方改到；缓存结果则会跨 step 串味（同一 query 在不同 + // 池子状态下该给不同答案）。 + let bm25IndexCache = null; // { generation, index } | null + /** * BM25 third recall path (v0.5.0 1.1). Scores the query tokens against the * live non-archived rows and returns the top `limit` hits with scores @@ -541,14 +560,40 @@ export function createService({ store, mirror, config, onWrite, logger, document function bm25Recall(q, limit) { if (config?.bm25SearchEnabled === false) return []; try { - const docs = store.list({ limit: 500, includeForgotten: false }).filter((m) => !m.archived); - if (!docs.length) return []; - return createBM25Index(docs).search(q, { limit }); + const index = bm25CorpusIndex(); + if (!index) return []; + return index.search(q, { limit }); } catch { return []; } } + /** + * 取（必要时重建）BM25 语料索引。水位读不到时退化成每次重建而不是报错：这条 + * 通路是 recall booster，读不到 mirror_state 只该慢一点，不该让词法召回整批消失。 + */ + function bm25CorpusIndex() { + let generation = null; + try { + generation = store.getMirrorState().generation; + } catch { /* 水位不可读：本次退回重建，且不写缓存 */ } + if (bm25IndexCache && generation !== null && bm25IndexCache.generation === generation) { + return bm25IndexCache.index; + } + const docs = store.list({ limit: 500, includeForgotten: false }).filter((m) => !m.archived); + if (!docs.length) { + bm25IndexCache = null; + return null; + } + const index = createBM25Index(docs); + // 水位读不到时**不碰缓存**，保留上一份「已知有效 generation」的条目：读失败 + // 若只是偶发（库锁、迁移窗口）而期间没有写入，恢复后水位与条目一致，直接命中； + // 期间真有写入，水位已经变了，下一轮自然重建。反过来把 null 当键存进去则是纯 + // 噪声——那样的条目永远命不中，还把有效的那份挤掉了。 + if (generation !== null) bm25IndexCache = { generation, index }; + return index; + } + /** * Graph fourth recall path (issue #219 → #24 块1 锚定层). When the query text * mentions a known entity name, memories linked to that entity (attrs + @@ -922,18 +967,13 @@ export function createService({ store, mirror, config, onWrite, logger, document } } - // Mode contract, auto (aligns rrf/minmax with blend): keyword leads, the - // recipe fills the remaining slots. blend.auto already front-loads keyword; - // rrf/minmax rank across sources, so re-apply the same "keyword first" - // ordering here to preserve the pre-fusion auto contract — the keyword - // hit list keeps its power, and only slots it couldn't fill go to the - // recipe's ranking. - if (recipe !== "blend" && mode === "auto" && keyword.length) { - const head = keyword.slice(0, lim); - const seen = new Set(head.map((m) => m.id)); - const tail = merged.filter((m) => !seen.has(m.id)).slice(0, Math.max(0, lim - head.length)); - merged = head.concat(tail); - } + // #617 第 3 条：这里原本还有一段「recipe !== "blend" 且 mode === "auto" 时用 + // keyword.slice(0,lim) 覆盖 merged」的收尾，注释自称是「保留 blend 的 auto + // 契约」。但 rrf/minmax 的全部意义就是跨源排序——把它们算出来的序整个换成 + // 「关键词命中先上」，等于配置了 rrf 却在 auto 下拿到 keyword 的顺序，配方 + // 白选。auto 的「关键词领位」语义只属于 blend（它在 :906-922 里自己实现了， + // 且带 seen 去重与容量控制），不该由收尾段替别的配方代劳。删掉后 rrf/minmax + // 的 auto 排序就是它们各自算出来的排序。 return { merged, signals }; } @@ -1778,62 +1818,60 @@ export function createService({ store, mirror, config, onWrite, logger, document (effImportance(a) * qualityWeight(a) * heatOf(a) * scopeMultOf(a)); }); let candidates = items; + // #617 第 2 条（前半）：向量路与 BM25 路此前是**互斥**的——`if (!semanticItems.length)` + // 包住 BM25，于是「有向量」等于「词法命中整批消失」。可两条路测的不是同一件事： + // 向量管「意思像」，BM25 管「词面撞上」，专有名词、issue 编号、函数名这类查询 + // 恰恰是向量最容易糊、词法最准的地方。现在两路都跑，用 RRF（k=60）融合排名： + // 某一路的第 1 名恒得 1/(60+1)，两路都靠前的行自然走高，不再由「谁先跑」决定。 + // 另外先按 fetchLimit 多取再过滤——旧代码先按 poolSize 截断再筛门控，筛掉的 + // 名额永远补不回来（缩水），池子越脏越明显。 + let fusedRrf = null; // id -> RRF 原始分；selectiveInject 复用它做二次重排 if (config.hybridInject !== false && q) { - // Bug4: semantic-first recall. Vector hits (queryVector, cached by the - // injector's async prefetch) lead when present; otherwise the last - // searchMemories recall for the exact same query is reused. Rule-based - // items fill + dedupe the remaining slots. Empty query / no vector / - // no cached recall → pure legacy rule-based selection. - const semanticItems = []; + const fetchLimit = Math.max(poolSize * 4, 40); + const passesGate = (m) => Boolean(m) && !m.archived && injectTypes.has(m.type) && !m.forgotten && + (m.source !== "narrative" || allowNarrative) && + codingGate(m) && + (m.type === "summary" || m.type === "preference" || m.importance >= threshold); + const vectorHits = []; if (Array.isArray(queryVector) && queryVector.length && vectorIndex) { try { - const hits = vectorIndex.search(queryVector, { limit: poolSize, threshold: 0 }); - for (const m of hits) { - if (m && !m.archived && injectTypes.has(m.type) && !m.forgotten && - (m.source !== "narrative" || allowNarrative) && - codingGate(m) && - (m.type === "summary" || m.type === "preference" || m.importance >= threshold)) { - semanticItems.push(m); - } + for (const m of vectorIndex.search(queryVector, { limit: fetchLimit, threshold: 0 })) { + if (passesGate(m)) vectorHits.push(m); } } catch { /* vector unavailable: fall through to the recall cache */ } } - if (!semanticItems.length && lastSemanticRecall?.query === q && lastSemanticRecall.items?.length) { - // CodeRabbit on #404：这条回填分支此前独漏重要度门槛——向量路（上方）与 - // BM25 路（下方 :1655 注释「门控与向量路径一致（含 importance 阈值）」） + if (!vectorHits.length && lastSemanticRecall?.query === q && lastSemanticRecall.items?.length) { + // CodeRabbit on #404：这条回填分支此前独漏重要度门槛——向量路与 BM25 路 // 都带着它。#369 收编 history 后缺口会被放大：库里低于闸门的行（实测 // history 378 条中有 27 条 i<=2）走一次检索进缓存就能绕开门槛占注入位。 - for (const m of lastSemanticRecall.items) { - if (m && !m.archived && injectTypes.has(m.type) && !m.forgotten && - (m.source !== "narrative" || allowNarrative) && codingGate(m) && - (m.type === "summary" || m.type === "preference" || m.importance >= threshold)) semanticItems.push(m); - } - } - // Issue #198：首轮（无向量、无缓存召回）的同步兜底——BM25 词法召回领位。 - // 注入渲染是同步的（宿主 systemPrompt 的 contexts 不支持异步 text）， - // 查询向量只能异步 prefetch 给下一轮，首轮必 miss；此时纯静态排序会让 - // 老的高重要性 summary/preference 占满槽位，当前话题的 decision/project - // 进不来。查询文本本身就在手上，BM25 是纯进程内同步计算，词法相关性 - // 足以把当前话题顶到前排；门控与向量路径一致（含 importance 阈值）。 - // 语义向量命中时本分支不参与，行为不变。 - if (!semanticItems.length) { - for (const hit of bm25Recall(q, poolSize)) { - if (hit && !hit.archived && injectTypes.has(hit.type) && !hit.forgotten && - (hit.source !== "narrative" || allowNarrative) && - codingGate(hit) && - (hit.type === "summary" || hit.type === "preference" || hit.importance >= threshold)) { - semanticItems.push(hit); - } - } - // issue #339（CodeRabbit review on #350）：首轮 BM25 兜底领跑合并池， - // 而 selectiveInject 相似度重排要等查询向量就绪才执行——软加权必须 - // 在这里就位，否则 foreign 命中绕过降权直接占注入位。 - if (softScopeActive) { - semanticItems.sort((a, b) => - (b.score ?? 0) * scopeMultOf(b) - (a.score ?? 0) * scopeMultOf(a)); - } + for (const m of lastSemanticRecall.items) if (passesGate(m)) vectorHits.push(m); } - if (semanticItems.length) { + // Issue #198：BM25 词法路。首轮（无向量、无缓存召回）这是唯一的同步相关性 + // 信号；有向量时它补上向量糊掉的那一类命中。#617 之后它不再被向量挤掉。 + const bm25Hits = []; + for (const hit of bm25Recall(q, fetchLimit)) if (passesGate(hit)) bm25Hits.push(hit); + if (vectorHits.length || bm25Hits.length) { + const RRF_K = 60; + const rrf = new Map(); + const rowById = new Map(); + const addList = (list) => { + list.forEach((m, idx) => { + rowById.set(m.id, m); + rrf.set(m.id, (rrf.get(m.id) ?? 0) + 1 / (RRF_K + idx + 1)); + }); + }; + addList(vectorHits); + addList(bm25Hits); + fusedRrf = rrf; + // 软加权（issue #339 / CodeRabbit on #350）作用在融合分上：foreign 行不再 + // 靠一路的原始 score 绕过降权，也不需要像旧代码那样单独给 BM25 排一次序。 + const ranked = [...rrf.entries()] + .map(([id, score]) => { + const m = rowById.get(id); + return { m, score: score * scopeMultOf(m) }; + }) + .sort((a, b) => b.score - a.score) + .map((row) => row.m); const seen = new Set(); const merged = []; const push = (m) => { @@ -1841,7 +1879,7 @@ export function createService({ store, mirror, config, onWrite, logger, document seen.add(m.id); merged.push(m); }; - for (const m of semanticItems) { + for (const m of ranked) { push(m); if (merged.length >= poolSize) break; } @@ -1927,16 +1965,29 @@ export function createService({ store, mirror, config, onWrite, logger, document // vector, so the injected slots go to memories on the current topic // rather than to the rule-based order. Rows the index did not return // keep their relative order after the scored ones. + // + // #617 第 2 条（后半）：这段此前是**纯余弦**全量 sort，于是上游用 RRF 顶上来的 + // BM25 命中（词面精确、余弦却低）会被重新压到 maxItems 之后，再被 general 槽 + // 的 `general.length >= maxItems` 丢掉——融合白做。现在余弦只占一半权重，另一半 + // 给 RRF 分（同一轮融合算出来的那份）：纯余弦命中仍是主体，但词法精确命中 + // 不会因为「意思不够像」被挤出去。 + // RRF 归一化：单路第 1 名原始分 1/(60+1)，乘 (60+1)/2 得 0.5，双路都在第 1 名 + // 得 1.0 —— 与余弦同量纲，且天然落在 [0,1]。 if (config?.selectiveInjectEnabled !== false && Array.isArray(queryVector) && queryVector.length && vectorIndex) { try { const hits = vectorIndex.search(queryVector, { limit: 200, threshold: 0 }); const sim = new Map(hits.map((m) => [m.id, m.score ?? 0])); if (sim.size) { - // 软加权作用在真实 sim 上；未命中的规则候选保持 -1 沉底（缺失项 - // 乘乘数会让 foreign 缺失项反而排到自己的缺失项之上，故不乘）。 + const RRF_NORM = (60 + 1) / 2; + // 软加权乘在融合分上（不是乘在某一路的原始分上）；两路都没有的规则 + // 候选保持 -1 沉底（缺失项乘乘数会让 foreign 缺失项反而排到自己的缺失 + // 项之上，故不乘）。 const scopedSim = (m) => { const s = sim.get(m.id); - return s === undefined ? -1 : s * scopeMultOf(m); + const rrf = fusedRrf?.get(m.id); + if (s === undefined && rrf === undefined) return -1; + const fused = 0.5 * (s ?? 0) + 0.5 * (rrf ?? 0) * RRF_NORM; + return fused * scopeMultOf(m); }; candidates = [...candidates].sort((a, b) => scopedSim(b) - scopedSim(a)); } diff --git a/dsh-mneme/src/inject.js b/dsh-mneme/src/inject.js index 014332c7..0613ad3a 100644 --- a/dsh-mneme/src/inject.js +++ b/dsh-mneme/src/inject.js @@ -44,13 +44,17 @@ function visibleSeqsOf(session) { // 候选问的是同一件事——「当时到底想干什么」；这条取原话的路只该有一份实现。 // 复制一份的代价本仓库付过几次（src/lib 平价、snapshotEvents 兼容层），所以 // 这里只加 export，不改行为。 -export function lastUserQuery(ctx) { +// +// #617：多带一个 seq——注入器要拿它判断「待认领票据是否比事件流更新」（见 +// createInjector 的 pendingQueries）。seq 是宿主事件序号，取不到时为 null +// （老版本宿主 / 测试夹具），调用方按 null 参与相等比较，不做真假推断。 +export function lastUserMessageInfo(ctx) { try { const session = ctx?.agent?.session; // DSH ≥0.1.2-rc only exposes events via snapshotEvents(); older builds // still have the .events property, so fall back to it. const events = session?.snapshotEvents?.() ?? session?.events; - if (!Array.isArray(events) || events.length === 0) return ""; + if (!Array.isArray(events) || events.length === 0) return { text: "", seq: null }; const visibleSeqs = visibleSeqsOf(session); for (let i = events.length - 1; i >= 0; i--) { const event = events[i]; @@ -61,16 +65,44 @@ export function lastUserQuery(ctx) { if (kind !== undefined && kind !== "user") continue; const parts = event.data?.content; if (!Array.isArray(parts) || parts.length === 0) continue; - return safeSlice( - parts - .map((p) => (typeof p === "string" ? p : p?.text ?? "")) - .filter(Boolean) - .join("\n"), - 500 - ); + return { + text: safeSlice( + parts + .map((p) => (typeof p === "string" ? p : p?.text ?? "")) + .filter(Boolean) + .join("\n"), + 500 + ), + seq: event?.seq ?? null + }; } } catch { /* session internals unavailable: degrade to no query */ } - return ""; + return { text: "", seq: null }; +} + +export function lastUserQuery(ctx) { + return lastUserMessageInfo(ctx).text; +} + +// 认领消息 → 查询文本。#617 的补位票据就是这里算出来的字符串：判定口径与 +// user/message 事件同源（content parts + source.kind），于是「插件注入的上下文」 +// 这类非人类消息照样被跳过，票据不会拿插件自己的话当用户问题。 +function inboxMessageQuery(message) { + try { + const parts = message?.content ?? message?.data?.content; + if (!Array.isArray(parts) || parts.length === 0) return ""; + const kind = message?.source?.kind ?? message?.data?.source?.kind; + if (kind !== undefined && kind !== "user") return ""; + return safeSlice( + parts + .map((p) => (typeof p === "string" ? p : p?.text ?? "")) + .filter(Boolean) + .join("\n"), + 500 + ); + } catch { + return ""; + } } // Hot-memory round extraction (v0.5.0 1.3): pairs each user/message with the @@ -291,21 +323,117 @@ export function createInjector(ctx, service, settings, config, gate = null, sess // first render after a new user message may still fall back to the rule-based // pick; later assemblies in the same session reuse the cached vector. Bounded // cache (cap 8, drop oldest) so a long session never grows it unbounded. + // + // #617 根因 B：旧实现用一个 lastPrefetched 字符串做去重，且它在 await 之前 + // 就置位——同一条查询第一次失败（模型没下好、超时）之后**永不重试**，而那 + // 恰好是最需要重试的一次。这里换成三件套：inflight Map 去重（渲染同一步会 + // 被求值两次：assemble 一次、refreshContext 一次，同一条查询只该嵌一次）、 + // 失败冷却 15s（失败可重试，但不许每次渲染都拿一次失败去撞模型）、返回 + // Promise（调用方/测试能等到它，而不是靠魔法延时）。 const QUERY_VECTOR_CACHE_MAX = 8; + const PREFETCH_FAIL_COOLDOWN_MS = 15000; const queryVectorCache = new Map(); - let lastPrefetched = ""; + const inflightPrefetches = new Map(); // query -> Promise + const prefetchFailedAt = new Map(); // query -> ms function prefetchQueryVector(query) { - if (!query || query === lastPrefetched || queryVectorCache.has(query)) return; - lastPrefetched = query; - service.embedQuery(query).then((vec) => { + if (!query) return null; + const cached = queryVectorCache.get(query); + if (cached) return Promise.resolve(cached); + const inflight = inflightPrefetches.get(query); + if (inflight) return inflight; + const failedAt = prefetchFailedAt.get(query); + if (failedAt !== undefined && Date.now() - failedAt < PREFETCH_FAIL_COOLDOWN_MS) return null; + const pending = service.embedQuery(query).then((vec) => { if (Array.isArray(vec) && vec.length) { queryVectorCache.set(query, vec); if (queryVectorCache.size > QUERY_VECTOR_CACHE_MAX) { queryVectorCache.delete(queryVectorCache.keys().next().value); } + prefetchFailedAt.delete(query); + return vec; } - }).catch(() => { /* prefetch is best-effort */ }); + prefetchFailedAt.set(query, Date.now()); + return null; + }).catch(() => { + prefetchFailedAt.set(query, Date.now()); + return null; /* prefetch is best-effort */ + }).finally(() => { + inflightPrefetches.delete(query); + }); + inflightPrefetches.set(query, pending); + return pending; + } + + // #617 根因 A 的补位票据：宿主（agent-loop 0.2.x）在每个 step 的 pre-step 里 + // 先 inbox.claim()（同步 emit agent/inbox/claimed），再 assemble()，而 + // user/message 要到 step() 末尾（:865）才 append——注入的 text 回调在 + // assemble / refreshContext 里求值，**永远比当轮用户输入早一拍**：Step 1 拿到 + // ""，第 N 轮的 Step 1 只能拿到第 N-1 轮的旧文本（实测 46 个渲染点错 3 次）。 + // 票据在 claim 那一刻记下原文；渲染时若事件流里最后一条用户消息还不是它， + // 说明本轮 user/message 尚未落盘，用票据；一旦落盘，事件流才是权威（它对 + // rewind / compact 的可见面判定更准，票据不看 surface）。 + // + // 为什么只在 claimed 上记、不从 agent/inbox/spliced 推：claim 自己的两次 + // mutate 都落 outcome:"canceled" 的 spliced（inserted 为空），拿 removedCount + // 当撤回信号会把每次正常认领误判成取消；claimed 只在真有消息被认领时触发， + // 是唯一无歧义的信号源。Cancel/Rewind 穿透因此也不存在——它们不产生 claimed。 + const PENDING_QUERY_MAX = 32; + // 票据超过这个岁数就不再采信：一个 step 的 assemble 不可能跑这么久，更久说明 + // 那条消息压根没落盘（认领后整轮被取消），票据已是过期证据。 + const PENDING_QUERY_TTL_MS = 60000; + const pendingQueries = new Map(); // sessionId -> { text, at } + const offClaimed = typeof ctx?.on === "function" + ? ctx.on("agent/inbox/claimed", (payload) => { + try { + const message = payload?.message; + const text = inboxMessageQuery(message); + if (!text) return; + const session = payload?.agent?.session; + const sessionId = session?.id; + if (typeof sessionId !== "string" || !sessionId) return; + // #539 无痕闸门必须挡在这里，不能只挡渲染侧：claim 在 pre-step 里同步触发， + // 早于 assemble；这条路径既写票据（票据会变成注入查询文本）又预取向量 + // （embedQuery 直接 POST 外部 /embeddings），闸门命中时两者都是该会话数据的 + // 外流。嵌入层自身没有任何会话维度判断，这里是唯一的拦截点。 + if (sessionGate?.isOff(session)) return; + pendingQueries.delete(sessionId); // 重新插入 = 移到队尾 + pendingQueries.set(sessionId, { text, at: Date.now() }); + if (pendingQueries.size > PENDING_QUERY_MAX) { + pendingQueries.delete(pendingQueries.keys().next().value); + } + // 认领即预取：embed 在 assemble→step 之间就已经在跑，Step 1 的第二次 + // 求值（refreshContext）有机会拿到向量。拿不到也无妨——查询文本已经在 + // 手上，BM25 词法路（#198）能顶上。 + prefetchQueryVector(text); + } catch { /* the ticket is best-effort; injection must never break */ } + }) + : null; + + // 当轮查询：票据优先于事件流（见上）。判据是**文本比对**而不是事件条数—— + // 从 claim 到 assemble 之间还夹着 await tools.resolve(...)，这期间任何插件 + // 追加一个事件都会让条数增长，按条数判会把「消息还没落盘」误判成「已落盘」， + // 票据提前退休、Step 1 又退回空串（正是本 issue 要修的那一拍）。文本相等才是 + // 「这条消息真的进事件流了」的直接证据；连续两轮同一句话时两者相等，取哪个都对。 + function liveStepQuery(ctx) { + const info = lastUserMessageInfo(ctx); + let sessionId = null; + try { + const id = ctx?.agent?.session?.id; + if (typeof id === "string" && id) sessionId = id; + } catch { /* session internals unavailable */ } + if (!sessionId) return info.text; + const ticket = pendingQueries.get(sessionId); + if (!ticket) return info.text; + if (info.text === ticket.text) { + pendingQueries.delete(sessionId); // 已落盘，票据退休（顺带保证 Map 不被历史会话撑大） + return info.text; + } + if (Date.now() - ticket.at > PENDING_QUERY_TTL_MS) { + pendingQueries.delete(sessionId); // 过期证据，不采信 + return info.text; + } + return ticket.text; } // 会话身份解析（v1.2 人格寻址用）：读 session.header.agentPreset——冻结的 @@ -395,10 +523,12 @@ export function createInjector(ctx, service, settings, config, gate = null, sess text: (ctx) => { if (sessionGate?.isOff(sessionOf(ctx))) return ""; // Bug4: pass the latest user query so injection prefers semantically - // relevant memories; lastUserQuery is best-effort (empty → legacy). + // relevant memories; liveStepQuery is best-effort (empty → legacy). // The query vector is prefetched asynchronously (cached) because the // render itself must stay synchronous. - const query = lastUserQuery(ctx); + // #617：liveStepQuery 而不是 lastUserQuery——宿主在 step 末尾才 append + // user/message，而注入在这之前求值，只看事件流会拿到上一轮的输入。 + const query = liveStepQuery(ctx); if (query) prefetchQueryVector(query); const queryVector = queryVectorCache.get(query); // v0.8.0 A3：scope 随请求解析（解析器恒返回对象，身份取不到时两维为 @@ -493,6 +623,12 @@ export function createInjector(ctx, service, settings, config, gate = null, sess ]; return () => { + // #617：认领票据是监听宿主事件的活句柄，卸载时必须成对摘掉，否则注入器重装 + // （面板开关、宿主热重载）会留下一个继续写 Map 的孤儿监听。 + offClaimed?.(); + pendingQueries.clear(); + inflightPrefetches.clear(); + prefetchFailedAt.clear(); queryVectorCache.clear(); rotationHistory.clear(); injectionSnapshot = null; // 注入器卸载即失效：快照不得跨生命周期存留 diff --git a/dsh-mneme/src/service.js b/dsh-mneme/src/service.js index 01d3a47b..b8463659 100644 --- a/dsh-mneme/src/service.js +++ b/dsh-mneme/src/service.js @@ -532,6 +532,25 @@ export function createService({ store, mirror, config, onWrite, logger, document return dot / (Math.sqrt(na) * Math.sqrt(nb)); } + // BM25 语料索引缓存（#617 审计 O2）。此前每次渲染都 store.list 500 行 + 重新 + // tokenize 全语料，实测 4.318ms/次、每个 step 一次；其中 store.list 占 4.318ms + // 里的绝大部分，而语料在两次写入之间根本没变。 + // + // 失效键用 mirror_state.generation，不另建计数器：改正文/可见性的写入（insert / + // update / remove / setArchived / setForgotten / setContent …）都已经 bump 它， + // 而 last_accessed_at 与 embedding 这两类写入**刻意不 bump**（store.js 里写明 + // reads must not mark the mirror dirty）——这两类恰好也不改 title+content 语料， + // 正是我们要的失效粒度，不用额外判断。 + // + // 水位每次从库里现读，不在进程内记本地计数：多宿主共用一个 memoryDir 是支持形态 + // （busy_timeout + WAL），另一个进程写完 bump 的是库里的 generation，本地计数 + // 永远看不见，只有现读才收得到。 + // + // 只缓存索引、不缓存查询结果：search() 返回的是 {...row, score} 副本，索引里 + // 留着的原始行不会被调用方改到；缓存结果则会跨 step 串味（同一 query 在不同 + // 池子状态下该给不同答案）。 + let bm25IndexCache = null; // { generation, index } | null + /** * BM25 third recall path (v0.5.0 1.1). Scores the query tokens against the * live non-archived rows and returns the top `limit` hits with scores @@ -541,14 +560,40 @@ export function createService({ store, mirror, config, onWrite, logger, document function bm25Recall(q, limit) { if (config?.bm25SearchEnabled === false) return []; try { - const docs = store.list({ limit: 500, includeForgotten: false }).filter((m) => !m.archived); - if (!docs.length) return []; - return createBM25Index(docs).search(q, { limit }); + const index = bm25CorpusIndex(); + if (!index) return []; + return index.search(q, { limit }); } catch { return []; } } + /** + * 取（必要时重建）BM25 语料索引。水位读不到时退化成每次重建而不是报错：这条 + * 通路是 recall booster，读不到 mirror_state 只该慢一点，不该让词法召回整批消失。 + */ + function bm25CorpusIndex() { + let generation = null; + try { + generation = store.getMirrorState().generation; + } catch { /* 水位不可读：本次退回重建，且不写缓存 */ } + if (bm25IndexCache && generation !== null && bm25IndexCache.generation === generation) { + return bm25IndexCache.index; + } + const docs = store.list({ limit: 500, includeForgotten: false }).filter((m) => !m.archived); + if (!docs.length) { + bm25IndexCache = null; + return null; + } + const index = createBM25Index(docs); + // 水位读不到时**不碰缓存**，保留上一份「已知有效 generation」的条目：读失败 + // 若只是偶发（库锁、迁移窗口）而期间没有写入，恢复后水位与条目一致，直接命中； + // 期间真有写入，水位已经变了，下一轮自然重建。反过来把 null 当键存进去则是纯 + // 噪声——那样的条目永远命不中，还把有效的那份挤掉了。 + if (generation !== null) bm25IndexCache = { generation, index }; + return index; + } + /** * Graph fourth recall path (issue #219 → #24 块1 锚定层). When the query text * mentions a known entity name, memories linked to that entity (attrs + @@ -922,18 +967,13 @@ export function createService({ store, mirror, config, onWrite, logger, document } } - // Mode contract, auto (aligns rrf/minmax with blend): keyword leads, the - // recipe fills the remaining slots. blend.auto already front-loads keyword; - // rrf/minmax rank across sources, so re-apply the same "keyword first" - // ordering here to preserve the pre-fusion auto contract — the keyword - // hit list keeps its power, and only slots it couldn't fill go to the - // recipe's ranking. - if (recipe !== "blend" && mode === "auto" && keyword.length) { - const head = keyword.slice(0, lim); - const seen = new Set(head.map((m) => m.id)); - const tail = merged.filter((m) => !seen.has(m.id)).slice(0, Math.max(0, lim - head.length)); - merged = head.concat(tail); - } + // #617 第 3 条：这里原本还有一段「recipe !== "blend" 且 mode === "auto" 时用 + // keyword.slice(0,lim) 覆盖 merged」的收尾，注释自称是「保留 blend 的 auto + // 契约」。但 rrf/minmax 的全部意义就是跨源排序——把它们算出来的序整个换成 + // 「关键词命中先上」，等于配置了 rrf 却在 auto 下拿到 keyword 的顺序，配方 + // 白选。auto 的「关键词领位」语义只属于 blend（它在 :906-922 里自己实现了， + // 且带 seen 去重与容量控制），不该由收尾段替别的配方代劳。删掉后 rrf/minmax + // 的 auto 排序就是它们各自算出来的排序。 return { merged, signals }; } @@ -1778,62 +1818,60 @@ export function createService({ store, mirror, config, onWrite, logger, document (effImportance(a) * qualityWeight(a) * heatOf(a) * scopeMultOf(a)); }); let candidates = items; + // #617 第 2 条（前半）：向量路与 BM25 路此前是**互斥**的——`if (!semanticItems.length)` + // 包住 BM25，于是「有向量」等于「词法命中整批消失」。可两条路测的不是同一件事： + // 向量管「意思像」，BM25 管「词面撞上」，专有名词、issue 编号、函数名这类查询 + // 恰恰是向量最容易糊、词法最准的地方。现在两路都跑，用 RRF（k=60）融合排名： + // 某一路的第 1 名恒得 1/(60+1)，两路都靠前的行自然走高，不再由「谁先跑」决定。 + // 另外先按 fetchLimit 多取再过滤——旧代码先按 poolSize 截断再筛门控，筛掉的 + // 名额永远补不回来（缩水），池子越脏越明显。 + let fusedRrf = null; // id -> RRF 原始分；selectiveInject 复用它做二次重排 if (config.hybridInject !== false && q) { - // Bug4: semantic-first recall. Vector hits (queryVector, cached by the - // injector's async prefetch) lead when present; otherwise the last - // searchMemories recall for the exact same query is reused. Rule-based - // items fill + dedupe the remaining slots. Empty query / no vector / - // no cached recall → pure legacy rule-based selection. - const semanticItems = []; + const fetchLimit = Math.max(poolSize * 4, 40); + const passesGate = (m) => Boolean(m) && !m.archived && injectTypes.has(m.type) && !m.forgotten && + (m.source !== "narrative" || allowNarrative) && + codingGate(m) && + (m.type === "summary" || m.type === "preference" || m.importance >= threshold); + const vectorHits = []; if (Array.isArray(queryVector) && queryVector.length && vectorIndex) { try { - const hits = vectorIndex.search(queryVector, { limit: poolSize, threshold: 0 }); - for (const m of hits) { - if (m && !m.archived && injectTypes.has(m.type) && !m.forgotten && - (m.source !== "narrative" || allowNarrative) && - codingGate(m) && - (m.type === "summary" || m.type === "preference" || m.importance >= threshold)) { - semanticItems.push(m); - } + for (const m of vectorIndex.search(queryVector, { limit: fetchLimit, threshold: 0 })) { + if (passesGate(m)) vectorHits.push(m); } } catch { /* vector unavailable: fall through to the recall cache */ } } - if (!semanticItems.length && lastSemanticRecall?.query === q && lastSemanticRecall.items?.length) { - // CodeRabbit on #404：这条回填分支此前独漏重要度门槛——向量路（上方）与 - // BM25 路（下方 :1655 注释「门控与向量路径一致（含 importance 阈值）」） + if (!vectorHits.length && lastSemanticRecall?.query === q && lastSemanticRecall.items?.length) { + // CodeRabbit on #404：这条回填分支此前独漏重要度门槛——向量路与 BM25 路 // 都带着它。#369 收编 history 后缺口会被放大：库里低于闸门的行（实测 // history 378 条中有 27 条 i<=2）走一次检索进缓存就能绕开门槛占注入位。 - for (const m of lastSemanticRecall.items) { - if (m && !m.archived && injectTypes.has(m.type) && !m.forgotten && - (m.source !== "narrative" || allowNarrative) && codingGate(m) && - (m.type === "summary" || m.type === "preference" || m.importance >= threshold)) semanticItems.push(m); - } - } - // Issue #198：首轮（无向量、无缓存召回）的同步兜底——BM25 词法召回领位。 - // 注入渲染是同步的（宿主 systemPrompt 的 contexts 不支持异步 text）， - // 查询向量只能异步 prefetch 给下一轮，首轮必 miss；此时纯静态排序会让 - // 老的高重要性 summary/preference 占满槽位，当前话题的 decision/project - // 进不来。查询文本本身就在手上，BM25 是纯进程内同步计算，词法相关性 - // 足以把当前话题顶到前排；门控与向量路径一致（含 importance 阈值）。 - // 语义向量命中时本分支不参与，行为不变。 - if (!semanticItems.length) { - for (const hit of bm25Recall(q, poolSize)) { - if (hit && !hit.archived && injectTypes.has(hit.type) && !hit.forgotten && - (hit.source !== "narrative" || allowNarrative) && - codingGate(hit) && - (hit.type === "summary" || hit.type === "preference" || hit.importance >= threshold)) { - semanticItems.push(hit); - } - } - // issue #339（CodeRabbit review on #350）：首轮 BM25 兜底领跑合并池， - // 而 selectiveInject 相似度重排要等查询向量就绪才执行——软加权必须 - // 在这里就位，否则 foreign 命中绕过降权直接占注入位。 - if (softScopeActive) { - semanticItems.sort((a, b) => - (b.score ?? 0) * scopeMultOf(b) - (a.score ?? 0) * scopeMultOf(a)); - } + for (const m of lastSemanticRecall.items) if (passesGate(m)) vectorHits.push(m); } - if (semanticItems.length) { + // Issue #198：BM25 词法路。首轮（无向量、无缓存召回）这是唯一的同步相关性 + // 信号；有向量时它补上向量糊掉的那一类命中。#617 之后它不再被向量挤掉。 + const bm25Hits = []; + for (const hit of bm25Recall(q, fetchLimit)) if (passesGate(hit)) bm25Hits.push(hit); + if (vectorHits.length || bm25Hits.length) { + const RRF_K = 60; + const rrf = new Map(); + const rowById = new Map(); + const addList = (list) => { + list.forEach((m, idx) => { + rowById.set(m.id, m); + rrf.set(m.id, (rrf.get(m.id) ?? 0) + 1 / (RRF_K + idx + 1)); + }); + }; + addList(vectorHits); + addList(bm25Hits); + fusedRrf = rrf; + // 软加权（issue #339 / CodeRabbit on #350）作用在融合分上：foreign 行不再 + // 靠一路的原始 score 绕过降权，也不需要像旧代码那样单独给 BM25 排一次序。 + const ranked = [...rrf.entries()] + .map(([id, score]) => { + const m = rowById.get(id); + return { m, score: score * scopeMultOf(m) }; + }) + .sort((a, b) => b.score - a.score) + .map((row) => row.m); const seen = new Set(); const merged = []; const push = (m) => { @@ -1841,7 +1879,7 @@ export function createService({ store, mirror, config, onWrite, logger, document seen.add(m.id); merged.push(m); }; - for (const m of semanticItems) { + for (const m of ranked) { push(m); if (merged.length >= poolSize) break; } @@ -1927,16 +1965,29 @@ export function createService({ store, mirror, config, onWrite, logger, document // vector, so the injected slots go to memories on the current topic // rather than to the rule-based order. Rows the index did not return // keep their relative order after the scored ones. + // + // #617 第 2 条（后半）：这段此前是**纯余弦**全量 sort，于是上游用 RRF 顶上来的 + // BM25 命中（词面精确、余弦却低）会被重新压到 maxItems 之后，再被 general 槽 + // 的 `general.length >= maxItems` 丢掉——融合白做。现在余弦只占一半权重，另一半 + // 给 RRF 分（同一轮融合算出来的那份）：纯余弦命中仍是主体，但词法精确命中 + // 不会因为「意思不够像」被挤出去。 + // RRF 归一化：单路第 1 名原始分 1/(60+1)，乘 (60+1)/2 得 0.5，双路都在第 1 名 + // 得 1.0 —— 与余弦同量纲，且天然落在 [0,1]。 if (config?.selectiveInjectEnabled !== false && Array.isArray(queryVector) && queryVector.length && vectorIndex) { try { const hits = vectorIndex.search(queryVector, { limit: 200, threshold: 0 }); const sim = new Map(hits.map((m) => [m.id, m.score ?? 0])); if (sim.size) { - // 软加权作用在真实 sim 上；未命中的规则候选保持 -1 沉底（缺失项 - // 乘乘数会让 foreign 缺失项反而排到自己的缺失项之上，故不乘）。 + const RRF_NORM = (60 + 1) / 2; + // 软加权乘在融合分上（不是乘在某一路的原始分上）；两路都没有的规则 + // 候选保持 -1 沉底（缺失项乘乘数会让 foreign 缺失项反而排到自己的缺失 + // 项之上，故不乘）。 const scopedSim = (m) => { const s = sim.get(m.id); - return s === undefined ? -1 : s * scopeMultOf(m); + const rrf = fusedRrf?.get(m.id); + if (s === undefined && rrf === undefined) return -1; + const fused = 0.5 * (s ?? 0) + 0.5 * (rrf ?? 0) * RRF_NORM; + return fused * scopeMultOf(m); }; candidates = [...candidates].sort((a, b) => scopedSim(b) - scopedSim(a)); } diff --git a/dsh-mneme/test/bm25-corpus-cache.test.js b/dsh-mneme/test/bm25-corpus-cache.test.js new file mode 100644 index 00000000..839dfe1f --- /dev/null +++ b/dsh-mneme/test/bm25-corpus-cache.test.js @@ -0,0 +1,145 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createStore } from "../src/store.js"; +import { createService } from "../src/service.js"; + +// #617 审计 O2 回归：BM25 语料索引按 mirror_state.generation 缓存。 +// +// 断言观测点选 store.list({ limit: 500 }) 的调用次数，不选耗时——耗时在 CI 上 +// 抖动大，而「这一轮到底有没有重新取语料重建索引」是确定性事实。 +// +// 一正一反两把尺子：正向锁「缓存真的命中了」，反向（写入 / 归档 / 开关）锁 +// 「该失效的时候真的失效了」——只有正向会让「永不失效」的实现也全绿。 + +function setup(over = {}) { + const store = createStore(":memory:"); + const config = { entitySearchEnabled: false, bm25SearchEnabled: true, ...over }; + const service = createService({ store, mirror: null, config, logger: null }); + // 只数语料取数：searchMemories 里还有别的 list 调用，按 limit 精确匹配。 + let corpusLoads = 0; + const origList = store.list.bind(store); + store.list = (opts) => { + if (opts?.limit === 500) corpusLoads += 1; + return origList(opts); + }; + return { store, service, loads: () => corpusLoads }; +} + +// auto 且不装 embedder：向量路不启动，词法结果按 blend 的 auto 分支领位后由 +// BM25 补位——这条路径下 BM25 命中真的会进返回值，断言才钉在用户可见的行为上， +// 而不是「索引建了但没人用」。 +const recall = (service, q) => service.searchMemories(q, { mode: "auto", topK: 5, useRerank: false, recordRecall: false }); + +test("#617-O2: 同一水位下多条查询共用一份语料索引，不重复重建", async () => { + const { store, service, loads } = setup(); + store.save({ type: "project", title: "异步并发模式", content: "async runtime 选用 tokio", importance: 3 }); + + const first = await recall(service, "rust 异步"); + assert.equal(loads(), 1, "首查要建一次索引"); + assert.ok(first.length > 0, "夹具前提：这条查询本来就有命中"); + + // 换一条查询：索引是按语料缓存的，不该因为 query 变了就重建。 + await recall(service, "tokio"); + await recall(service, "完全不相干的词"); + assert.equal(loads(), 1, "语料没变，三次查询只该取一次语料"); +}); + +test("#617-O2: 写入新记忆后索引失效，新行立刻可召回", async () => { + const { store, service, loads } = setup(); + store.save({ type: "project", title: "夜间 ETL 脚本", content: "离线批处理", importance: 3 }); + await recall(service, "离线批处理"); + assert.equal(loads(), 1); + + const fresh = store.save({ type: "project", title: "令牌桶限流", content: "ratelimit 令牌桶", importance: 3 }); + const ids = (await recall(service, "令牌桶")).map((r) => r.id); + assert.equal(loads(), 2, "写了新行，语料变了，必须重建"); + assert.ok(ids.includes(fresh.id), "重建后新行必须能被词法召回（陈旧索引会漏掉它）"); +}); + +test("#617-O2: 归档后索引失效，归档行退出词法召回", async () => { + const { store, service, loads } = setup(); + const m = store.save({ type: "project", title: "令牌桶限流", content: "ratelimit 令牌桶", importance: 3 }); + assert.ok((await recall(service, "令牌桶")).some((r) => r.id === m.id), "夹具前提：归档前能召回"); + + store.setArchived(m.id, true); + const ids = (await recall(service, "令牌桶")).map((r) => r.id); + assert.equal(loads(), 2, "可见性变了要重建"); + assert.ok(!ids.includes(m.id), "归档行不得留在索引里"); +}); + +test("#617-O2: 触达与重嵌不改语料，不该让索引失效", async () => { + const { store, service, loads } = setup(); + const m = store.save({ type: "project", title: "异步并发模式", content: "async runtime 选用 tokio", importance: 3 }); + await recall(service, "rust 异步"); + assert.equal(loads(), 1); + + // 这两类写入刻意不 bump generation（store.js：reads must not mark the mirror + // dirty），也不改 title+content。缓存若跟着它们失效，就等于每轮注入都白重建 + // ——正是 O2 要消除的成本。 + store.touchLastAccess(m.id); + store.setEmbedding(m.id, [0.1, 0.2, 0.3]); + await recall(service, "rust 异步"); + assert.equal(loads(), 1, "触达 / 重嵌不构成语料变化"); +}); + +test("#617-O2: bm25SearchEnabled=false 时根本不取语料", async () => { + const { store, service, loads } = setup({ bm25SearchEnabled: false }); + store.save({ type: "project", title: "异步并发模式", content: "async runtime 选用 tokio", importance: 3 }); + const rows = await recall(service, "rust 异步"); + assert.equal(loads(), 0, "开关关着连语料都不该取"); + assert.ok(!rows.some((r) => r.source === "bm25")); +}); + +test("#617-O2: 空库不缓存空索引，后续写入仍可召回", async () => { + const { store, service, loads } = setup(); + assert.deepEqual(await recall(service, "令牌桶"), [], "夹具前提：空库没有命中"); + + const m = store.save({ type: "project", title: "令牌桶限流", content: "ratelimit 令牌桶", importance: 3 }); + const ids = (await recall(service, "令牌桶")).map((r) => r.id); + assert.ok(ids.includes(m.id), "空库那一轮不得留下「空索引」这种缓存"); + assert.equal(loads(), 2); +}); + +test("#617-O2: 水位偶发读不到时，不拿旧索引冒充当前语料", async () => { + const { store, service, loads } = setup(); + store.save({ type: "project", title: "夜间 ETL 脚本", content: "离线批处理", importance: 3 }); + await recall(service, "离线批处理"); + assert.equal(loads(), 1, "夹具前提：正常路径已经缓存了一份索引"); + + // 模拟 mirror_state 读不出来（库锁 / 迁移窗口）。此时无法证明缓存新鲜，必须 + // 退回重建，而不是把 generation 当成 null 去和缓存里的旧值比对后命中。 + const realGet = store.getMirrorState.bind(store); + store.getMirrorState = () => { throw new Error("mirror_state unavailable"); }; + const fresh = store.save({ type: "project", title: "令牌桶限流", content: "ratelimit 令牌桶", importance: 3 }); + const ids = (await recall(service, "令牌桶")).map((r) => r.id); + assert.ok(ids.includes(fresh.id), "水位不可读时仍要看到新写入的行"); + assert.equal(loads(), 2); + + // 恢复可读：期间确实写过，水位已经变了，于是再重建一次拿到的才是与水位对齐 + // 的索引——宁可多建一次，也不能让缓存条目声称自己代表某个它没见过的版本。 + store.getMirrorState = realGet; + await recall(service, "令牌桶"); + assert.equal(loads(), 3, "水位已前进，恢复后须重建一次把条目对齐到新水位"); + + // 对齐之后再查就不该再重建了。 + await recall(service, "离线批处理"); + assert.equal(loads(), 3, "对齐后回到正常缓存路径"); +}); + +test("#617-O2: 水位读不到但期间没有写入时，恢复后直接命中旧索引", async () => { + const { store, service, loads } = setup(); + store.save({ type: "project", title: "夜间 ETL 脚本", content: "离线批处理", importance: 3 }); + await recall(service, "离线批处理"); + assert.equal(loads(), 1); + + // 读失败但语料没变：此时丢掉缓存纯属白干。保留上一份「已知有效 generation」 + // 的条目，恢复后水位一致就直接命中。 + const realGet = store.getMirrorState.bind(store); + store.getMirrorState = () => { throw new Error("mirror_state unavailable"); }; + await recall(service, "离线批处理"); + assert.equal(loads(), 2, "读不到水位只能重建（无法证明缓存新鲜）"); + + store.getMirrorState = realGet; + await recall(service, "离线批处理"); + assert.equal(loads(), 2, "期间无写入，恢复后不该再白建一次"); +}); diff --git a/dsh-mneme/test/issue-617-step1-query-and-fusion.test.js b/dsh-mneme/test/issue-617-step1-query-and-fusion.test.js new file mode 100644 index 00000000..11ac77b6 --- /dev/null +++ b/dsh-mneme/test/issue-617-step1-query-and-fusion.test.js @@ -0,0 +1,361 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createStore } from "../src/store.js"; +import { createService } from "../src/service.js"; +import { createInjector } from "../src/inject.js"; +import { createSettings } from "../src/settings.js"; +import { createVectorIndex } from "../src/vector-index.js"; +import { createSessionGate } from "../src/session-gate.js"; + +// #617 回归。三条缺陷各自独立成组，组内都用「一正一反」两把尺子：正向锁 +// 「改对了」，反向（关掉对应开关 / 换成对照配方）锁「这条断言真的是在测那条 +// 通路」，免得断言退化成常量真。 + +// --- 组 1：Step 1 的当轮查询（根因 A）------------------------------------- +// +// 宿主 agent-loop 的每个 step 是 claim() → assemble() → step()，而 user/message +// 要到 step() 末尾才 append。注入的 text 回调在 assemble 里求值，所以只看事件流 +// 必然慢一拍：Step 1 拿到 ""，第 N 轮的 Step 1 拿到第 N-1 轮的旧输入。 +// 票据（agent/inbox/claimed）在 claim 那一刻记原文，渲染时事件流还没长就用它。 +function setupInjector(over = {}) { + const store = createStore(":memory:"); + const service = createService({ store, mirror: null, config: {} }); + const settings = createSettings(store.db); + const contexts = []; + const handlers = new Map(); + const ctx = { + // 真实宿主 ctx 有 on()；旧夹具没有，所以旧夹具永远测不到这条路径。 + on(event, fn) { + handlers.set(event, fn); + return () => handlers.delete(event); + }, + systemPrompt: { + context(def) { + contexts.push(def); + return () => {}; + } + } + }; + const config = { maxInjectedItems: 3, importanceThreshold: 3, ...over }; + const injector = createInjector(ctx, service, settings, config); + // 查询词是注入链路里唯一能直接观测的输入：包一层记下每次渲染实际传了什么。 + const queries = []; + const orig = service.injectCandidates.bind(service); + service.injectCandidates = (args) => { + queries.push(args.query); + return orig(args); + }; + const fireClaim = (message, session) => handlers.get("agent/inbox/claimed")?.({ message, agent: { session } }); + return { store, service, contexts, injector, queries, fireClaim, handlers }; +} + +const userMessage = (text) => ({ type: "user/message", data: { source: { kind: "user" }, content: [{ type: "text", text }] } }); +const claimed = (text) => ({ content: [{ type: "text", text }], source: { kind: "user" }, id: "m1" }); + +test("#617-A: Step 1 在 user/message 落盘之前，用认领票据拿到当轮输入", () => { + const { contexts, queries, fireClaim } = setupInjector(); + const session = { id: "s-617", events: [], snapshotEvents() { return this.events; } }; + // claim 发生在 assemble 之前，此刻事件流里还没有本轮消息 + fireClaim(claimed("夜航App令牌桶限流方案怎么设计"), session); + assert.equal(session.events.length, 0, "夹具前提：认领时 user/message 尚未落盘"); + contexts[0].text({ agent: { session } }); + assert.equal(queries.at(-1), "夜航App令牌桶限流方案怎么设计", "Step 1 必须拿到当轮输入，而不是空串"); +}); + +test("#617-A: claim 与 assemble 之间插入无关事件，票据不得提前退休", () => { + // 宿主 preStep 的顺序是 claim() → await tools.resolve() → assemble()。中间那个 + // await 给了别的插件追加事件的机会；若按「事件条数变多」判票据失效，Step 1 就 + // 又退回空串——正是本 issue 要修的那一拍。判据必须是文本，不是条数。 + const { contexts, queries, fireClaim } = setupInjector(); + const session = { id: "s-617i", events: [], snapshotEvents() { return this.events; } }; + session.events.push(userMessage("上一轮的问题")); + fireClaim(claimed("这一轮的新问题"), session); + // tools.resolve 期间，别的插件追加了一个与用户消息无关的事件 + session.events.push({ type: "system/message", data: { content: [{ type: "text", text: "插件上下文" }] } }); + contexts[0].text({ agent: { session } }); + assert.equal(queries.at(-1), "这一轮的新问题", "无关事件不得让票据提前失效"); +}); + +test("#617-A: 事件流长过票据记录后，票据退休、以事件流为准", () => { + const { contexts, queries, fireClaim } = setupInjector(); + const session = { id: "s-617b", events: [], snapshotEvents() { return this.events; } }; + fireClaim(claimed("第一个问题"), session); + contexts[0].text({ agent: { session } }); + assert.equal(queries.at(-1), "第一个问题"); + // step() 末尾 append → 事件流变长 + session.events.push(userMessage("第一个问题")); + contexts[0].text({ agent: { session } }); + assert.equal(queries.at(-1), "第一个问题", "落盘后两者一致（此轮无差异）"); + // 下一轮：新的 claim 覆盖票据，事件流里仍是上一轮的文本 + fireClaim(claimed("第二个问题"), session); + contexts[0].text({ agent: { session } }); + assert.equal(queries.at(-1), "第二个问题", "新一轮的 Step 1 不得拿到上一轮的旧输入"); +}); + +test("#617-A: 非人类消息（source.kind 非 user）不产生票据", () => { + const { contexts, queries, fireClaim } = setupInjector(); + const session = { id: "s-617c", events: [], snapshotEvents() { return this.events; } }; + fireClaim({ content: [{ type: "text", text: "插件注入的上下文" }], source: { kind: "system" } }, session); + contexts[0].text({ agent: { session } }); + assert.equal(queries.at(-1), "", "插件自己的话不得被当成用户问题"); +}); + +test("#617-A: 卸载后摘掉 claimed 监听，不再写票据", () => { + const { contexts, injector, queries, fireClaim, handlers } = setupInjector(); + const session = { id: "s-617d", events: [], snapshotEvents() { return this.events; } }; + assert.ok(handlers.has("agent/inbox/claimed"), "挂载时注册监听"); + injector(); + assert.equal(handlers.has("agent/inbox/claimed"), false, "卸载即摘，不留孤儿监听"); + fireClaim(claimed("卸载后不该记"), session); + // 注入器已卸载，context 也不再生效；这里只确认没有票据泄漏到下一次装配 + assert.equal(queries.length, 0); + void contexts; +}); + +// --- 组 2：prefetch 去重与失败冷却（根因 B）------------------------------- +// +// 旧实现用 lastPrefetched 字符串去重，且置位在 await 之前 —— 同一条查询第一次 +// 失败（模型没下好、超时）之后永不重试，而那恰好是最需要重试的一次。 +function setupPrefetch(embedImpl) { + const store = createStore(":memory:"); + const service = createService({ store, mirror: null, config: {} }); + const settings = createSettings(store.db); + const contexts = []; + const ctx = { systemPrompt: { context(def) { contexts.push(def); return () => {}; } } }; + const calls = []; + service.setEmbedder({ + embedSingle: async (text) => { + calls.push(text); + return embedImpl(text); + } + }); + const injector = createInjector(ctx, service, settings, { maxInjectedItems: 3, importanceThreshold: 3 }); + return { store, service, contexts, injector, calls }; +} + +const renderWith = (contexts, session) => contexts[0].text({ agent: { session } }); +const tick = () => new Promise((r) => setTimeout(r, 0)); + +test("#617-B: 同一步的两次求值（assemble + refreshContext）只嵌一次", async () => { + const { contexts, calls } = setupPrefetch(async () => [1, 0, 0]); + const session = { id: "p1", events: [userMessage("查询文本")], snapshotEvents() { return this.events; } }; + renderWith(contexts, session); // 第一次求值：发起 inflight + renderWith(contexts, session); // 第二次求值：必须复用 inflight，不再嵌 + await tick(); + renderWith(contexts, session); // 第三次：命中缓存 + assert.equal(calls.length, 1, "同一条查询只应嵌入一次"); +}); + +test("#617-B: 失败后进入冷却期，不在同一帧反复撞模型", async () => { + const { contexts, calls } = setupPrefetch(async () => { throw new Error("model not ready"); }); + const session = { id: "p2", events: [userMessage("失败查询")], snapshotEvents() { return this.events; } }; + renderWith(contexts, session); + await tick(); + renderWith(contexts, session); + await tick(); + renderWith(contexts, session); + assert.equal(calls.length, 1, "冷却期内不得重试（15s 内）"); +}); + +test("#617-B: 冷却按查询分别计，别的查询照常嵌入", async () => { + const { contexts, calls } = setupPrefetch(async (text) => { + if (text.includes("失败")) throw new Error("boom"); + return [1, 0, 0]; + }); + const s1 = { id: "p3", events: [userMessage("失败查询")], snapshotEvents() { return this.events; } }; + renderWith(contexts, s1); + await tick(); + renderWith(contexts, s1); + const s2 = { id: "p4", events: [userMessage("另一条查询")], snapshotEvents() { return this.events; } }; + renderWith(contexts, s2); + await tick(); + assert.equal(calls.length, 2, "一条查询的失败冷却不应连坐其它查询"); +}); + +// --- 组 3：向量路与 BM25 路并存（根因 C，前半）---------------------------- +function seedStalePreferences(service, n = 8) { + const ids = []; + for (let i = 0; i < n; i++) { + ids.push(service.saveWithDedupe({ + type: "preference", + title: `旧偏好条目${i}`, + content: `用户喜欢深色界面与紧凑排版，这是第${i}条历史偏好存档`, + importance: 5 + }).memory.id); + } + return ids; +} + +function setupDualPath(config = {}) { + const store = createStore(":memory:"); + const service = createService({ store, mirror: null, config }); + return { store, service }; +} + +test("#617-C: 有向量命中时，BM25 的词法精确命中不再整批消失", () => { + const { service, store } = setupDualPath(); + seedStalePreferences(service, 8); + // 与查询词法完全不相交，但向量最近 + const vecOnly = service.saveWithDedupe({ + type: "project", title: "星图渲染重构", content: "绘制管线迁移与图层合并", importance: 3 + }).memory; + // 词法精确命中，但没有向量（真实库里大量老行就是这样） + const lexHit = service.saveWithDedupe({ + type: "decision", title: "夜航限流选型", content: "决定采用令牌桶做夜航App的限流方案", importance: 3 + }).memory; + const vectorIndex = createVectorIndex({ store }); + service.setVectorIndex(vectorIndex); + vectorIndex.saveEmbedding(vecOnly.id, [1, 0, 0]); + + const picked = service.injectCandidates({ + query: "夜航App令牌桶限流方案怎么设计的", maxItems: 5, threshold: 3, queryVector: [1, 0, 0] + }); + assert.equal(picked[0].id, vecOnly.id, "纯余弦命中仍是首位"); + assert.ok(picked.some((m) => m.id === lexHit.id), "BM25 命中不得被向量路挤掉（旧代码：有向量则 BM25 整段跳过）"); + assert.ok(picked.length <= 5, "池子仍按 maxItems 封顶"); + store.close(); +}); + +test("#617-C: 关掉 BM25 后同一条词法命中确实消失（反向对照）", () => { + const { service, store } = setupDualPath({ bm25SearchEnabled: false }); + seedStalePreferences(service, 8); + const vecOnly = service.saveWithDedupe({ + type: "project", title: "星图渲染重构", content: "绘制管线迁移与图层合并", importance: 3 + }).memory; + const lexHit = service.saveWithDedupe({ + type: "decision", title: "夜航限流选型", content: "决定采用令牌桶做夜航App的限流方案", importance: 3 + }).memory; + const vectorIndex = createVectorIndex({ store }); + service.setVectorIndex(vectorIndex); + vectorIndex.saveEmbedding(vecOnly.id, [1, 0, 0]); + + const picked = service.injectCandidates({ + query: "夜航App令牌桶限流方案怎么设计的", maxItems: 5, threshold: 3, queryVector: [1, 0, 0] + }); + assert.equal(picked[0].id, vecOnly.id); + assert.equal(picked.some((m) => m.id === lexHit.id), false, "BM25 关 = 词法路不参与，正向用例才不是在测别的东西"); + store.close(); +}); + +// --- 组 4：auto 不再被 keyword 覆盖（根因 C，后半）------------------------ +// +// 旧代码在 fuseRecall 收尾处，只要 recipe !== "blend" 且 mode === "auto"，就用 +// keyword.slice(0,lim) 覆盖 merged —— 于是配了 rrf 却拿到 keyword 的顺序，配方 +// 白选。auto 的「关键词领位」语义只属于 blend。 +function setupFusion(recipe) { + const store = createStore(":memory:"); + const service = createService({ store, mirror: null, config: { recallFusion: recipe } }); + // 标题命中（keyword 第一），但没有向量 + const kwTitle = service.saveWithDedupe({ + type: "preference", title: "令牌桶限流笔记", content: "随手记的限流笔记", importance: 2 + }).memory; + // 正文命中（keyword 第二）+ 向量第一 —— 两路都靠前 + const both = service.saveWithDedupe({ + type: "decision", title: "夜航限流选型", content: "决定采用令牌桶限流方案，阈值按等级分层", importance: 5 + }).memory; + const vectorIndex = createVectorIndex({ store }); + service.setVectorIndex(vectorIndex); + vectorIndex.saveEmbedding(both.id, [1, 0, 0]); + service.setEmbedder({ embedSingle: async () => [1, 0, 0] }); + return { service, store, kwTitle, both }; +} + +test("#617-C: recallFusion=rrf 时 auto 用 RRF 自己的排序，不再被 keyword 覆盖", async () => { + const { service, store, kwTitle, both } = setupFusion("rrf"); + const rows = await service.searchMemories("令牌桶限流", { mode: "auto", topK: 2, useRerank: false, recordRecall: false }); + assert.equal(rows.length, 2); + assert.equal(rows[0].id, both.id, "两路都命中的行必须排第一（keyword 只命中标题的行不该领位）"); + assert.equal(rows[1].id, kwTitle.id); + store.close(); +}); + +test("#617-C: blend 的 auto 关键词领位契约保持不变（反向对照）", async () => { + const { service, store, kwTitle, both } = setupFusion("blend"); + const rows = await service.searchMemories("令牌桶限流", { mode: "auto", topK: 2, useRerank: false, recordRecall: false }); + assert.equal(rows.length, 2); + assert.equal(rows[0].id, kwTitle.id, "blend 下 keyword 领位是既有契约，本次改动不该碰它"); + assert.equal(rows[1].id, both.id); + store.close(); +}); + +// --- 组 5：无痕会话的认领路径不得外发（#539 × #617）----------------------- +// +// claim 在宿主 pre-step 里同步触发，早于 assemble，因此也早于渲染侧的闸门判断。 +// 这条路径既写票据（票据会成为注入查询文本）又预取向量（embedQuery 直接 POST +// 外部 /embeddings），闸门命中时两者都是该会话数据的外流；嵌入层自身没有任何 +// 会话维度判断（embedding.js / service.js 里 sessionGate 零命中），这里是唯一 +// 拦截点。两条反向对照（未命中名单 / 总开关关闭）保证断言不是退化成「嵌入器 +// 恰好没跑」。 +function setupGated(over = {}) { + const store = createStore(":memory:"); + const service = createService({ store, mirror: null, config: {} }); + const settings = createSettings(store.db); + const contexts = []; + const handlers = new Map(); + const ctx = { + on(event, fn) { + handlers.set(event, fn); + return () => handlers.delete(event); + }, + systemPrompt: { context(def) { contexts.push(def); return () => {}; } } + }; + const calls = []; + service.setEmbedder({ + embedSingle: async (text) => { + calls.push(text); + return [1, 0, 0]; + } + }); + const queries = []; + const orig = service.injectCandidates.bind(service); + service.injectCandidates = (args) => { + queries.push(args.query); + return orig(args); + }; + const sessionGate = createSessionGate({ + sessionGateEnabled: true, + sessionGateSessions: ["s-gated"], + sessionGateSubagents: true, + ...over + }); + const injector = createInjector( + ctx, service, settings, { maxInjectedItems: 3, importanceThreshold: 3 }, null, sessionGate + ); + const fireClaim = (message, session) => handlers.get("agent/inbox/claimed")?.({ message, agent: { session } }); + return { store, service, contexts, injector, calls, queries, fireClaim, sessionGate }; +} + +test("#617-D: 无痕会话的认领消息既不预取向量，也不留票据", async () => { + const { contexts, calls, queries, fireClaim, sessionGate } = setupGated(); + const session = { id: "s-gated", events: [], snapshotEvents() { return this.events; } }; + assert.equal(sessionGate.isOff(session), true, "夹具前提：闸门认得出这个会话"); + fireClaim(claimed("桉桉的私人日记：我今天去了协和医院复查"), session); + await tick(); + assert.equal(calls.length, 0, "闸门命中时不得把原文发往嵌入端点"); + // 票据若被写下，渲染时事件流还没长，Step 1 就会拿到这段原文 + contexts[0].text({ agent: { session } }); + assert.deepEqual(queries, [], "闸门命中时连 injectCandidates 都不该被调到"); +}); + +test("#617-D: 未命中名单的会话照常预取（反向对照）", async () => { + const { contexts, calls, queries, fireClaim, sessionGate } = setupGated(); + const session = { id: "s-normal", events: [], snapshotEvents() { return this.events; } }; + assert.equal(sessionGate.isOff(session), false); + fireClaim(claimed("夜航App令牌桶限流方案怎么设计"), session); + await tick(); + assert.equal(calls.length, 1, "普通会话的预取行为不变"); + assert.equal(calls[0], "夜航App令牌桶限流方案怎么设计"); + // 票据确实写下了：渲染拿到的就是认领的原文 + contexts[0].text({ agent: { session } }); + assert.deepEqual(queries, ["夜航App令牌桶限流方案怎么设计"]); +}); + +test("#617-D: 总开关关闭时闸门不生效（反向对照）", async () => { + const { calls, fireClaim, sessionGate } = setupGated({ sessionGateEnabled: false }); + const session = { id: "s-gated", events: [], snapshotEvents() { return this.events; } }; + assert.equal(sessionGate.isEnabled(), false); + assert.equal(sessionGate.isOff(session), false, "总开关关时名单不得生效"); + fireClaim(claimed("夜航App令牌桶限流方案怎么设计"), session); + await tick(); + assert.equal(calls.length, 1, "未开启 #539 时不得凭空拦截（默认行为须与基线一致）"); +});