Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
<a href="LICENSE"><img src="https://img.shields.io/badge/license-MIT-3E63DD?style=flat-square" alt="license"></a>
<a href="https://github.com/slow-stack/mneme/actions"><img src="https://img.shields.io/github/actions/workflow/status/slow-stack/mneme/ci.yml?style=flat-square&label=CI" alt="CI"></a>
<a href="https://nodejs.org"><img src="https://img.shields.io/badge/node-22%2B-3E63DD?style=flat-square&logo=nodedotjs&logoColor=white" alt="node"></a>
<a href="https://github.com/slow-stack/mneme"><img src="https://img.shields.io/badge/tests-2103%20passed-3E63DD?style=flat-square" alt="tests"></a>
<a href="https://github.com/slow-stack/mneme"><img src="https://img.shields.io/badge/tests-2107%20passed-3E63DD?style=flat-square" alt="tests"></a>
<a href="https://codecov.io/gh/slow-stack/mneme"><img src="https://img.shields.io/codecov/c/github/slow-stack/mneme/main?style=flat-square" alt="coverage"></a>
<a href="https://github.com/awesome-dsh-plugin/awesome-dsh-plugin"><img src="https://awesome-dsh-plugin.com/badge.svg" alt="Awesome"></a>
</p>
Expand Down Expand Up @@ -209,7 +209,7 @@ dsh-mneme-serve # 默认 ~/.dsh/memory + 127.0.0.1:8790

```bash
cd dsh-mneme && npm install
npm test # 2103 个测试
npm test # 2107 个测试
npm run stress # 三轴线压测
npm run sync # src → lib 同步
```
Expand Down Expand Up @@ -430,7 +430,7 @@ Community projects that wire mneme into more platforms — not part of this repo

```bash
cd dsh-mneme && npm install
npm test # 2103 tests
npm test # 2107 tests
npm run stress # three-axis stress test
npm run sync # src → lib sync
```
Expand Down
6 changes: 3 additions & 3 deletions dsh-mneme/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
[![npm version](https://img.shields.io/npm/v/@modusensus/dsh-mneme?color=blue&label=npm)](https://www.npmjs.com/package/@modusensus/dsh-mneme)
[![license](https://img.shields.io/badge/license-MIT-green)](LICENSE)
[![Awesome](https://awesome-dsh-plugin.com/badge.svg)](https://github.com/awesome-dsh-plugin/awesome-dsh-plugin)
[![tests](https://img.shields.io/badge/tests-2103%20passed-success)](https://github.com/slow-stack/mneme)
[![tests](https://img.shields.io/badge/tests-2107%20passed-success)](https://github.com/slow-stack/mneme)
[![CI](https://img.shields.io/github/actions/workflow/status/slow-stack/mneme/ci.yml)](https://github.com/slow-stack/mneme/actions)
[![node](https://img.shields.io/badge/node-22%2B-blue)](https://nodejs.org)
[![npm downloads](https://img.shields.io/npm/d18m/@modusensus/dsh-mneme.svg?color=blue&label=downloads)](https://www.npmjs.com/package/@modusensus/dsh-mneme)
Expand Down Expand Up @@ -618,7 +618,7 @@ src/
├── api.js # HTTP 路由(Web 面板数据通道,含 /conflicts 冲突队列)
└── index.js # 插件接线
lib/ # src 的同步分发产物(npm run sync;发布前由 root prepack 的 check-sync.js 校验一致性;唯一手写例外 lib/client.js——Web 面板 bundle,sync 不覆盖)
test/ # 2103 个 node:test 测试(审计与三轴线压测不变量;src↔lib 一致性由 scripts/check-sync.js 发布闸门校验)
test/ # 2107 个 node:test 测试(审计与三轴线压测不变量;src↔lib 一致性由 scripts/check-sync.js 发布闸门校验)
scripts/ # e2e-dsh.js 端到端演示 · stress-dsh.js 三轴线压测 · sync-lib.js 同步 · check-sync.js 发布闸门 · benchmark-recall.js / benchmark-embed.js / benchmark-rerank.js 基准 · sync-test-badge.mjs 测试徽章 · build-runtime-manifest.mjs 运行时清单
```

Expand All @@ -627,7 +627,7 @@ scripts/ # e2e-dsh.js 端到端演示 · stress-dsh.js 三轴线压
```bash
cd dsh-mneme
npm install # 安装 peer 依赖(以 devDependencies 形式,用于本地测试)
npm test # 运行 2103 个测试
npm test # 运行 2107 个测试
npm run stress # 三轴线压测:长会话检索 / 冲突仲裁 / 多 Agent 并发(离线 mock LLM)
npm run sync # 把 src/ 同步到 lib/(发布时由 prepack 钩子自动执行)
```
Expand Down
7 changes: 7 additions & 0 deletions dsh-mneme/lib/config.js
Original file line number Diff line number Diff line change
Expand Up @@ -415,6 +415,13 @@ export const Config = z.object({
// in the background after a short delay, rate-limited in batches. On by
// default; set false to keep the backfill manual only.
autoReindexOnBoot: z.boolean().default(true),
// Backfill pacing (previously hardcoded 500 / 10 / 200 in semantic.js). A
// fixed 500-per-boot cap meant a 4k-row backlog needed nine restarts to
// clear, and nothing reported that the run had been capped.
// reindexMaxPerBoot = 0 means "no cap, drain the whole backlog this run".
reindexMaxPerBoot: z.natural().default(500),
reindexBatchSize: z.natural().min(1).max(64).default(10),
reindexRateLimitMs: z.natural().default(200),
// Semantic-first injection (Bug4): when enabled, injectCandidates with a
// non-empty query recalls via the vector index first and falls back to the
// rule-based pick to fill/dedupe. Empty query / no vector → legacy behavior.
Expand Down
131 changes: 116 additions & 15 deletions dsh-mneme/lib/embedding.js
Original file line number Diff line number Diff line change
Expand Up @@ -69,9 +69,51 @@ export async function embedText({ baseUrl, apiKey, model }, text) {
* successful embed the model that produced the vectors is recorded, so the
* index can detect drift and the auto-reindex backfill knows what to rebuild.
*/
/**
* Bounded receipt ring for embedding attempts.
*
* Why a ring instead of llm_audit_logs: a full backfill is thousands of
* attempts, and one audit row each would bury the LLM calls that table exists
* to trace. What the user actually needs in order to answer "is the vector
* layer silently dead right now?" is the recent window plus running totals —
* which costs nothing and cannot flood.
*/
const RECEIPT_LIMIT = 200;

function createReceiptLedger() {
const recent = [];
const totals = { ok: 0, failed: 0, skipped: 0 };
let lastOkAt = null;
let lastFailAt = null;
let lastFailReason = null;
return {
record(entry) {
recent.push(entry);
if (recent.length > RECEIPT_LIMIT) recent.shift();
if (entry.status in totals) totals[entry.status]++;
if (entry.status === "ok") lastOkAt = entry.at;
else if (entry.status === "failed") {
lastFailAt = entry.at;
lastFailReason = entry.reason ?? null;
}
},
snapshot() {
return {
totals: { ...totals },
lastOkAt,
lastFailAt,
lastFailReason,
recent: recent.slice(-RECEIPT_LIMIT)
};
}
};
}

export function createEmbedder({ store, settings, logger, vectorIndex }) {
// Dimension of the most recent successful embed, exposed for fingerprinting.
let _dimension = 0;
// Per-write embed receipts (see createReceiptLedger above).
const ledger = createReceiptLedger();

/** Record the producing model fingerprint in vector_meta (best-effort). */
function markModel(cfg, dimension) {
Expand All @@ -85,19 +127,52 @@ export function createEmbedder({ store, settings, logger, vectorIndex }) {
// ready/configured 两个 getter 共用,避免以后改一处漏一处。
const configComplete = (cfg) => !!(cfg?.enabled && cfg.baseUrl && cfg.apiKey && cfg.model);

/**
* Embed one memory and store the vector. Never throws — a flaky endpoint
* must not break memory writes — but it now REPORTS what happened instead
* of returning undefined on both success and failure. Callers that ignore
* the result are unaffected; callers that care get a receipt.
*
* @returns {Promise<{status:"ok"|"failed"|"skipped", id:string,
* at:string, reason?:string, dim?:number, model?:string}>}
*/
async function embedFor(id, title, content) {
const at = new Date().toISOString();
const cfg = settings.getVectorConfig();
if (!configComplete(cfg)) return;
if (!configComplete(cfg)) {
const entry = { status: "skipped", id, at, reason: "vector-config-incomplete" };
ledger.record(entry);
return entry;
}
const text = [title, content].filter(Boolean).join("\n");
const vector = await embedText(cfg, text);
if (vector) {
store.setEmbedding(id, vector);
_dimension = vector.length;
// Bug3: record which model produced the current vectors so the index can
// detect drift and skip a redundant backfill when nothing changed.
markModel(cfg, vector.length);
logger?.info?.(`[dsh-mneme] embedded memory ${id} (dim=${vector.length})`);
let vector = null;
try {
vector = await embedText(cfg, text);
} catch (error) {
const entry = { status: "failed", id, at, reason: String(error) };
ledger.record(entry);
logger?.warn?.(`[dsh-mneme] embed memory ${id} failed: ${String(error)}`);
return entry;
}
if (!vector) {
// embedText swallows the HTTP detail (404 / timeout / malformed body all
// resolve to null). The receipt still has to say "no vector", which is
// exactly the fact the write path never used to report.
const entry = { status: "failed", id, at, reason: "embedding-request-returned-no-vector" };
ledger.record(entry);
logger?.warn?.(`[dsh-mneme] embed memory ${id} produced no vector`);
return entry;
}
const model = modelHashOf(cfg.model);
store.setEmbedding(id, vector, model);
_dimension = vector.length;
// Bug3: record which model produced the current vectors so the index can
// detect drift and skip a redundant backfill when nothing changed.
markModel(cfg, vector.length);
const entry = { status: "ok", id, at, dim: vector.length, model };
ledger.record(entry);
logger?.info?.(`[dsh-mneme] embedded memory ${id} (dim=${vector.length})`);
return entry;
}

return {
Expand All @@ -120,10 +195,26 @@ export function createEmbedder({ store, settings, logger, vectorIndex }) {
get configured() {
return configComplete(settings.getVectorConfig());
},
/** Fire-and-forget re-embed of a memory after any write. */
/**
* Re-embed a memory after any write. Still non-blocking — the caller does
* not have to await it — but it now RETURNS the receipt promise so a
* caller that wants to know can. Previously this returned undefined and
* swallowed the rejection, so "did this write get a vector?" had no answer
* anywhere in the process. Swallowing still applies to the rejection
* itself: a vector failure must never break the memory write.
*/
schedule(memory) {
if (!memory?.id) return;
embedFor(memory.id, memory.title, memory.content).catch(() => {});
if (!memory?.id) {
return Promise.resolve({ status: "skipped", id: null, at: new Date().toISOString(), reason: "no-id" });
}
return embedFor(memory.id, memory.title, memory.content).catch((error) => ({
status: "failed", id: memory.id, at: new Date().toISOString(), reason: String(error)
}));
},

/** Recent embed receipts + running totals (see createReceiptLedger). */
embedReceipts() {
return ledger.snapshot();
},

/** Embed one text and return its vector (null on failure/disabled). */
Expand Down Expand Up @@ -155,17 +246,27 @@ export function createEmbedder({ store, settings, logger, vectorIndex }) {
return _dimension || undefined;
},

/** Batch re-index rows still missing an embedding. */
/**
* Batch re-index rows whose vector is missing OR came from another model.
* Uses needsEmbeddingFor (not needsEmbedding) so a model switch actually
* re-embeds the old vectors instead of leaving them eligible for "has a
* vector" forever while they poison similarity ranking.
*/
async reindexMissing(limit = 50) {
const cfg = settings.getVectorConfig();
if (!cfg?.enabled) return { indexed: 0, skipped: 0 };
const rows = store.needsEmbedding(limit);
const model = modelHashOf(cfg.model);
// Fall back to the plain selector for stores that predate the
// provenance column (older hosts, test doubles).
const rows = typeof store.needsEmbeddingFor === "function"
? store.needsEmbeddingFor(model, limit)
: store.needsEmbedding(limit);
let indexed = 0;
for (const row of rows) {
const text = [row.title, row.content].filter(Boolean).join("\n");
const vector = await embedText(cfg, text);
if (vector) {
store.setEmbedding(row.id, vector);
store.setEmbedding(row.id, vector, model);
_dimension = vector.length;
indexed++;
}
Expand Down
76 changes: 66 additions & 10 deletions dsh-mneme/lib/semantic.js
Original file line number Diff line number Diff line change
Expand Up @@ -28,29 +28,59 @@ import { LocalReranker } from "./reranker.js";
*/
export async function backfillMissingEmbeddings({
store, embedder, vectorIndex, logger,
// maxTotal: 0 = no cap (drain the whole backlog this run). The default stays
// modest so a boot never floods a provider, but the cap is a caller choice
// now instead of a hardcoded one — the fixed 500 meant a 4k-row backlog
// needed nine restarts to clear, and nothing reported that it had been cut.
maxTotal = 500, batchSize = 10, rateLimitMs = 200
}) {
const startedAt = new Date().toISOString();
// Model-aware selector: rows whose vector came from another model (or
// predate per-row provenance) must be re-embedded too, otherwise a model
// switch leaves uncomparable vectors in the index forever.
const model = embedder.modelHash;
const pick = (n) => (typeof store.needsEmbeddingFor === "function"
? store.needsEmbeddingFor(model, n)
: store.needsEmbedding(n));
const unlimited = !(maxTotal > 0);
let indexed = 0;
for (let done = 0; done < maxTotal;) {
const rows = store.needsEmbedding(batchSize);
let processed = 0;
let failed = 0;
for (let done = 0; unlimited || done < maxTotal;) {
const rows = pick(batchSize);
if (!rows.length) break;
for (const row of rows) {
try {
const text = [row.title, row.content].filter(Boolean).join("\n");
const vector = await embedder.embedSingle(text);
if (vector?.length) {
store.setEmbedding(row.id, vector);
store.setEmbedding(row.id, vector, model);
indexed++;
} else {
failed++;
}
} catch { /* skip the bad row */ }
} catch { failed++; /* skip the bad row, but count it */ }
}
done += rows.length;
processed += rows.length;
// Rate limit: space out batches so the provider is not hammered.
if (store.needsEmbedding(1).length) await new Promise((r) => setTimeout(r, rateLimitMs));
if (pick(1).length) await new Promise((r) => setTimeout(r, rateLimitMs));
}
if (indexed > 0 && embedder.modelHash) vectorIndex.markModel?.(embedder.modelHash, embedder.dimension);
logger?.info?.(`[dsh-mneme] auto-reindex backfilled ${indexed} embeddings on boot`);
return indexed;
if (indexed > 0 && model) vectorIndex.markModel?.(model, embedder.dimension);
// stillStale answers "did it actually finish?". The old return value was a
// bare indexed count, so a run that hit the cap looked exactly like a run
// that drained the backlog.
const stillStale = pick(1).length > 0;
const finishReason = stillStale ? "capped" : (failed > 0 && indexed === 0 ? "all-failed" : "drained");
logger?.info?.(
`[dsh-mneme] auto-reindex backfilled ${indexed} embeddings (processed=${processed}, failed=${failed}, finish=${finishReason})`
);
return {
indexed, processed, failed, finishReason, stillStale,
startedAt,
finishedAt: new Date().toISOString(),
model: model ?? null
};
}

/**
Expand All @@ -74,6 +104,9 @@ export function createSemantic({ store, service, settings, cfg, logger, vectorIn
// boot 回填首查计时器。搬移前在 index.js 是裸 setTimeout(不参与卸载清理);
// 纳入 dispose 是修悬挂,不改变启动行为。
let reindexTimer = null;
// Last boot-backfill summary, exposed via backfillStatus() so "did it
// finish?" has an answer other than a log line.
let lastBackfill = null;

if (lightMode) {
// Light mode: the whole vector pipeline stays off — no embedder (nothing
Expand Down Expand Up @@ -202,12 +235,31 @@ export function createSemantic({ store, service, settings, cfg, logger, vectorIn
if (tries > 0) reindexTimer = setTimeout(() => attempt(tries - 1), 2000);
return;
}
if (!store.needsEmbedding(1).length) return; // nothing to backfill
const model = embedder.modelHash;
const stale = typeof store.hasStaleEmbedding === "function"
? store.hasStaleEmbedding(model)
: store.needsEmbedding(1).length > 0;
if (!stale) return; // nothing missing and nothing from another model
// Issue #128: no fingerprint gate here anymore — a matching fingerprint
// used to return early and permanently orphan rows whose embed failed
// at write time. See backfillMissingEmbeddings().
backfillMissingEmbeddings({ store, embedder, vectorIndex, logger })
// Cap and pacing are configuration now (reindexMaxPerBoot /
// reindexBatchSize / reindexRateLimitMs) instead of hardcoded 500/10/200.
lastBackfill = {
state: "running", startedAt: new Date().toISOString(),
maxTotal: cfg.reindexMaxPerBoot, batchSize: cfg.reindexBatchSize
};
backfillMissingEmbeddings({
store, embedder, vectorIndex, logger,
maxTotal: cfg.reindexMaxPerBoot,
batchSize: cfg.reindexBatchSize,
rateLimitMs: cfg.reindexRateLimitMs
})
.then((summary) => {
lastBackfill = { ...summary, state: "finished" };
})
.catch((error) => {
lastBackfill = { state: "error", error: String(error), startedAt: lastBackfill?.startedAt };
logger?.warn?.(`[dsh-mneme] auto-reindex failed: ${String(error)}`);
});
} catch (error) {
Expand All @@ -221,6 +273,10 @@ export function createSemantic({ store, service, settings, cfg, logger, vectorIn
return {
embedder,
reranker,
/** Last backfill run: { state, indexed, failed, finishReason, stillStale, ... }. */
backfillStatus() {
return lastBackfill;
},
dispose() {
if (embedRetryTimer !== null) { clearTimeout(embedRetryTimer); embedRetryTimer = null; }
if (reindexTimer !== null) { clearTimeout(reindexTimer); reindexTimer = null; }
Expand Down
Loading
Loading