From c08ab760d83da5191a99e55345a91f1f020c500f Mon Sep 17 00:00:00 2001 From: Ace <1160684883@qq.com> Date: Sun, 20 Sep 2026 19:17:04 +0800 Subject: [PATCH] fix(summarize): persist distillation cursor (#229) --- dsh-mneme/lib/service.js | 2 + dsh-mneme/lib/store.js | 41 ++++++++++ dsh-mneme/lib/summarize.js | 32 +++++++- dsh-mneme/src/service.js | 2 + dsh-mneme/src/store.js | 41 ++++++++++ dsh-mneme/src/summarize.js | 32 +++++++- dsh-mneme/test/store.test.js | 13 +++ dsh-mneme/test/summarize.test.js | 131 ++++++++++++++++++++++++++++++- 8 files changed, 287 insertions(+), 7 deletions(-) diff --git a/dsh-mneme/lib/service.js b/dsh-mneme/lib/service.js index 3fa59fef..04681451 100644 --- a/dsh-mneme/lib/service.js +++ b/dsh-mneme/lib/service.js @@ -2120,6 +2120,8 @@ export function createService({ store, mirror, config, onWrite, logger }) { toApiList, isVisibleInScope, transaction, + getDistillCursor: (sessionId) => store.getDistillCursor(sessionId), + setDistillCursor: (sessionId, lastSeq) => store.setDistillCursor(sessionId, lastSeq), enqueue, setDreamHook(fn) { dreamHook = fn; }, setSleepHook(fn) { sleepHook = fn; }, diff --git a/dsh-mneme/lib/store.js b/dsh-mneme/lib/store.js index 27a13493..35ae3bf5 100644 --- a/dsh-mneme/lib/store.js +++ b/dsh-mneme/lib/store.js @@ -197,6 +197,14 @@ CREATE TABLE IF NOT EXISTS llm_audit_logs ( CREATE INDEX IF NOT EXISTS idx_llm_audit_timestamp ON llm_audit_logs(timestamp); CREATE INDEX IF NOT EXISTS idx_llm_audit_source ON llm_audit_logs(trigger_source); +-- autoSummarize 的增量蒸馏游标：按 session.id 持久化最近一次成功消费的事件序。 +-- 游标是蒸馏窗口的恢复事实，不与 user_settings 或记忆内容混用。 +CREATE TABLE IF NOT EXISTS distill_cursors ( + session_id TEXT PRIMARY KEY, + last_seq INTEGER NOT NULL, + updated_at TEXT NOT NULL +); + -- entity gene (v0.3.0): named entities mentioned across memories, with -- time-boxed attributes (valid_from → valid_until) and typed relations. -- Attributes follow the snapshot style: saveAttr invalidates the previous @@ -789,6 +797,37 @@ export function createStore(path) { return ts; } + /** 返回指定会话持久化的 autoSummarize 游标；不存在时返回 undefined。 */ + function getDistillCursor(sessionId) { + if (typeof sessionId !== "string" || sessionId.length === 0) return undefined; + const row = db.prepare( + "SELECT session_id, last_seq, updated_at FROM distill_cursors WHERE session_id = ?" + ).get(sessionId); + return row + ? { session_id: row.session_id, last_seq: Number(row.last_seq), updated_at: row.updated_at } + : undefined; + } + + /** + * 单调持久化会话游标。调用方若同时写入记忆，必须放在外层 SQLite 事务内。 + */ + function setDistillCursor(sessionId, lastSeq) { + if (typeof sessionId !== "string" || sessionId.length === 0) { + throw new TypeError("setDistillCursor: sessionId must be a non-empty string"); + } + if (!Number.isSafeInteger(lastSeq)) { + throw new TypeError("setDistillCursor: lastSeq must be a safe integer"); + } + db.prepare(` + INSERT INTO distill_cursors (session_id, last_seq, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + last_seq = MAX(distill_cursors.last_seq, excluded.last_seq), + updated_at = excluded.updated_at + `).run(sessionId, lastSeq, nowIso()); + return getDistillCursor(sessionId); + } + function count(type, { minImportance = null, source = null, includeForgotten = false, includeArchived = false, onlyArchived = false, depositedOnly = false, updatedFrom = null, updatedTo = null, occurredFrom = null, occurredTo = null, visibility = null } = {}) { const clauses = []; const params = []; @@ -2461,6 +2500,8 @@ export function createStore(path) { getRecallEval, listRecallEvals, saveLlmAudit, + getDistillCursor, + setDistillCursor, listLlmAudits, countLlmAudits, getLlmAuditStats, diff --git a/dsh-mneme/lib/summarize.js b/dsh-mneme/lib/summarize.js index 93d9045b..c4f4d88e 100644 --- a/dsh-mneme/lib/summarize.js +++ b/dsh-mneme/lib/summarize.js @@ -405,6 +405,19 @@ export function createSummarizer(ctx, service, config, deps = {}) { } } + function persistCursor(sessionId, nextSeq) { + if (!Number.isFinite(nextSeq)) return; + if (typeof service.setDistillCursor !== "function") { + throw new Error("dsh-mneme: persistent summarization cursors are unavailable"); + } + service.setDistillCursor(sessionId, nextSeq); + } + + function commitCursor(sessionId, nextSeq) { + persistCursor(sessionId, nextSeq); + if (Number.isFinite(nextSeq)) lastDistilledSeq.set(sessionId, nextSeq); + } + /** * Issue #239（第 4 项）：把这一轮蒸馏推迟到最近的「高峰结束」时刻。每会话只挂 * 一个定时器（重复触发不叠加）；被 summarizePeakMaxDeferMinutes 截断时到点照跑 @@ -500,7 +513,10 @@ export function createSummarizer(ctx, service, config, deps = {}) { let abortedRun = false; try { if (!route) return; - const previousSeq = lastDistilledSeq.get(session.id); + const persistedCursor = typeof service.getDistillCursor === "function" + ? service.getDistillCursor(session.id) + : undefined; + const previousSeq = persistedCursor?.last_seq ?? lastDistilledSeq.get(session.id); const triggerSeq = eventSeq(triggerEvent); // 只读取上次成功游标之后、当前 turn/end 之前的事件。旧版 snapshotEvents // 即使忽略范围参数，collectMessages 仍会按事件 seq 二次过滤。 @@ -519,7 +535,7 @@ export function createSummarizer(ctx, service, config, deps = {}) { if (!collected.messages.length) { // 没有可蒸馏的公开文本也算成功消费当前事件窗口，避免每个 turn/end // 都重新扫描同一批无内容事件；没有 seq 时则不提交不可验证的游标。 - if (Number.isFinite(nextSeq)) lastDistilledSeq.set(session.id, nextSeq); + commitCursor(session.id, nextSeq); return; } const messages = collected.messages; @@ -536,7 +552,7 @@ export function createSummarizer(ctx, service, config, deps = {}) { 0 ); if (distillChars < minWindowChars) { - if (Number.isFinite(nextSeq)) lastDistilledSeq.set(session.id, nextSeq); + commitCursor(session.id, nextSeq); writeAudit({ timestamp: new Date().toISOString(), model_id: route ? `${route.provider}:${route.model}` : "unknown", @@ -712,7 +728,11 @@ export function createSummarizer(ctx, service, config, deps = {}) { ...(dup ? { _mergeInto: dup.memory.id } : {}) }); } + persistCursor(session.id, nextSeq); }); + } else { + // 空数组是合法成功：没有记忆写入，但本次事件窗口仍然应被持久消费。 + persistCursor(session.id, nextSeq); } if (audit && (capped > 0 || deduped > 0)) { audit.metadata = { @@ -724,6 +744,12 @@ export function createSummarizer(ctx, service, config, deps = {}) { // 记忆写入和解析都成功后才提交窗口；流失败、中止、解析失败或写入异常 // 都会在此之前退出，从而保留窗口供下一次重试。 if (Number.isFinite(nextSeq)) lastDistilledSeq.set(session.id, nextSeq); + } catch (error) { + if (audit) { + audit.status = "error"; + audit.errorMessage = String(error?.message ?? error); + } + throw error; } finally { // Issue #127：aborted（会话关闭 / 插件 dispose）不占间隔，避免误伤该会话的 // 下一次蒸馏；其余情况（含失败）的打点保留，与 dreamMinIntervalMinutes 一致。 diff --git a/dsh-mneme/src/service.js b/dsh-mneme/src/service.js index 3fa59fef..04681451 100644 --- a/dsh-mneme/src/service.js +++ b/dsh-mneme/src/service.js @@ -2120,6 +2120,8 @@ export function createService({ store, mirror, config, onWrite, logger }) { toApiList, isVisibleInScope, transaction, + getDistillCursor: (sessionId) => store.getDistillCursor(sessionId), + setDistillCursor: (sessionId, lastSeq) => store.setDistillCursor(sessionId, lastSeq), enqueue, setDreamHook(fn) { dreamHook = fn; }, setSleepHook(fn) { sleepHook = fn; }, diff --git a/dsh-mneme/src/store.js b/dsh-mneme/src/store.js index 27a13493..35ae3bf5 100644 --- a/dsh-mneme/src/store.js +++ b/dsh-mneme/src/store.js @@ -197,6 +197,14 @@ CREATE TABLE IF NOT EXISTS llm_audit_logs ( CREATE INDEX IF NOT EXISTS idx_llm_audit_timestamp ON llm_audit_logs(timestamp); CREATE INDEX IF NOT EXISTS idx_llm_audit_source ON llm_audit_logs(trigger_source); +-- autoSummarize 的增量蒸馏游标：按 session.id 持久化最近一次成功消费的事件序。 +-- 游标是蒸馏窗口的恢复事实，不与 user_settings 或记忆内容混用。 +CREATE TABLE IF NOT EXISTS distill_cursors ( + session_id TEXT PRIMARY KEY, + last_seq INTEGER NOT NULL, + updated_at TEXT NOT NULL +); + -- entity gene (v0.3.0): named entities mentioned across memories, with -- time-boxed attributes (valid_from → valid_until) and typed relations. -- Attributes follow the snapshot style: saveAttr invalidates the previous @@ -789,6 +797,37 @@ export function createStore(path) { return ts; } + /** 返回指定会话持久化的 autoSummarize 游标；不存在时返回 undefined。 */ + function getDistillCursor(sessionId) { + if (typeof sessionId !== "string" || sessionId.length === 0) return undefined; + const row = db.prepare( + "SELECT session_id, last_seq, updated_at FROM distill_cursors WHERE session_id = ?" + ).get(sessionId); + return row + ? { session_id: row.session_id, last_seq: Number(row.last_seq), updated_at: row.updated_at } + : undefined; + } + + /** + * 单调持久化会话游标。调用方若同时写入记忆，必须放在外层 SQLite 事务内。 + */ + function setDistillCursor(sessionId, lastSeq) { + if (typeof sessionId !== "string" || sessionId.length === 0) { + throw new TypeError("setDistillCursor: sessionId must be a non-empty string"); + } + if (!Number.isSafeInteger(lastSeq)) { + throw new TypeError("setDistillCursor: lastSeq must be a safe integer"); + } + db.prepare(` + INSERT INTO distill_cursors (session_id, last_seq, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + last_seq = MAX(distill_cursors.last_seq, excluded.last_seq), + updated_at = excluded.updated_at + `).run(sessionId, lastSeq, nowIso()); + return getDistillCursor(sessionId); + } + function count(type, { minImportance = null, source = null, includeForgotten = false, includeArchived = false, onlyArchived = false, depositedOnly = false, updatedFrom = null, updatedTo = null, occurredFrom = null, occurredTo = null, visibility = null } = {}) { const clauses = []; const params = []; @@ -2461,6 +2500,8 @@ export function createStore(path) { getRecallEval, listRecallEvals, saveLlmAudit, + getDistillCursor, + setDistillCursor, listLlmAudits, countLlmAudits, getLlmAuditStats, diff --git a/dsh-mneme/src/summarize.js b/dsh-mneme/src/summarize.js index 93d9045b..c4f4d88e 100644 --- a/dsh-mneme/src/summarize.js +++ b/dsh-mneme/src/summarize.js @@ -405,6 +405,19 @@ export function createSummarizer(ctx, service, config, deps = {}) { } } + function persistCursor(sessionId, nextSeq) { + if (!Number.isFinite(nextSeq)) return; + if (typeof service.setDistillCursor !== "function") { + throw new Error("dsh-mneme: persistent summarization cursors are unavailable"); + } + service.setDistillCursor(sessionId, nextSeq); + } + + function commitCursor(sessionId, nextSeq) { + persistCursor(sessionId, nextSeq); + if (Number.isFinite(nextSeq)) lastDistilledSeq.set(sessionId, nextSeq); + } + /** * Issue #239（第 4 项）：把这一轮蒸馏推迟到最近的「高峰结束」时刻。每会话只挂 * 一个定时器（重复触发不叠加）；被 summarizePeakMaxDeferMinutes 截断时到点照跑 @@ -500,7 +513,10 @@ export function createSummarizer(ctx, service, config, deps = {}) { let abortedRun = false; try { if (!route) return; - const previousSeq = lastDistilledSeq.get(session.id); + const persistedCursor = typeof service.getDistillCursor === "function" + ? service.getDistillCursor(session.id) + : undefined; + const previousSeq = persistedCursor?.last_seq ?? lastDistilledSeq.get(session.id); const triggerSeq = eventSeq(triggerEvent); // 只读取上次成功游标之后、当前 turn/end 之前的事件。旧版 snapshotEvents // 即使忽略范围参数，collectMessages 仍会按事件 seq 二次过滤。 @@ -519,7 +535,7 @@ export function createSummarizer(ctx, service, config, deps = {}) { if (!collected.messages.length) { // 没有可蒸馏的公开文本也算成功消费当前事件窗口，避免每个 turn/end // 都重新扫描同一批无内容事件；没有 seq 时则不提交不可验证的游标。 - if (Number.isFinite(nextSeq)) lastDistilledSeq.set(session.id, nextSeq); + commitCursor(session.id, nextSeq); return; } const messages = collected.messages; @@ -536,7 +552,7 @@ export function createSummarizer(ctx, service, config, deps = {}) { 0 ); if (distillChars < minWindowChars) { - if (Number.isFinite(nextSeq)) lastDistilledSeq.set(session.id, nextSeq); + commitCursor(session.id, nextSeq); writeAudit({ timestamp: new Date().toISOString(), model_id: route ? `${route.provider}:${route.model}` : "unknown", @@ -712,7 +728,11 @@ export function createSummarizer(ctx, service, config, deps = {}) { ...(dup ? { _mergeInto: dup.memory.id } : {}) }); } + persistCursor(session.id, nextSeq); }); + } else { + // 空数组是合法成功：没有记忆写入，但本次事件窗口仍然应被持久消费。 + persistCursor(session.id, nextSeq); } if (audit && (capped > 0 || deduped > 0)) { audit.metadata = { @@ -724,6 +744,12 @@ export function createSummarizer(ctx, service, config, deps = {}) { // 记忆写入和解析都成功后才提交窗口；流失败、中止、解析失败或写入异常 // 都会在此之前退出，从而保留窗口供下一次重试。 if (Number.isFinite(nextSeq)) lastDistilledSeq.set(session.id, nextSeq); + } catch (error) { + if (audit) { + audit.status = "error"; + audit.errorMessage = String(error?.message ?? error); + } + throw error; } finally { // Issue #127：aborted（会话关闭 / 插件 dispose）不占间隔，避免误伤该会话的 // 下一次蒸馏；其余情况（含失败）的打点保留，与 dreamMinIntervalMinutes 一致。 diff --git a/dsh-mneme/test/store.test.js b/dsh-mneme/test/store.test.js index d4ef4889..e9aa50ce 100644 --- a/dsh-mneme/test/store.test.js +++ b/dsh-mneme/test/store.test.js @@ -16,6 +16,15 @@ test("createStore initializes schema and opens db", () => { "SELECT name FROM sqlite_master WHERE type='table' AND name='memories'" ).get(); assert.ok(row, "memories table exists"); + const cursorTable = store.db.prepare( + "SELECT name FROM sqlite_master WHERE type='table' AND name='distill_cursors'" + ).get(); + assert.ok(cursorTable, "distill cursor table exists"); + assert.equal(store.getDistillCursor("session-1"), undefined); + const saved = store.setDistillCursor("session-1", 4); + assert.equal(saved.last_seq, 4); + assert.ok(saved.updated_at); + assert.equal(store.setDistillCursor("session-1", 2).last_seq, 4, "cursor never moves backwards"); store.close(); }); @@ -173,6 +182,10 @@ test("schema migration adds archived column to legacy database", () => { const store = createStore(dbPath); const cols = store.db.prepare("PRAGMA table_info(memories)").all().map((c) => c.name); assert.ok(cols.includes("archived"), "archived column added"); + const cursorTable = store.db.prepare( + "SELECT name FROM sqlite_master WHERE type='table' AND name='distill_cursors'" + ).get(); + assert.ok(cursorTable, "new cursor table is created for legacy databases"); store.close(); } finally { rmSync(dir, { recursive: true, force: true }); diff --git a/dsh-mneme/test/summarize.test.js b/dsh-mneme/test/summarize.test.js index 23e4eadd..61734ff7 100644 --- a/dsh-mneme/test/summarize.test.js +++ b/dsh-mneme/test/summarize.test.js @@ -1,5 +1,8 @@ import test from "node:test"; import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; import { createStore } from "../src/store.js"; import { createService } from "../src/service.js"; import { createSummarizer, parseSummaryJson, parsePeakSpec, isInPeakWindow, nextOffPeakAt } from "../src/summarize.js"; @@ -45,7 +48,7 @@ function fakeClock(start) { } function setup(over = {}, opts = {}) { - const store = createStore(":memory:"); + const store = opts.store ?? createStore(":memory:"); const service = createService({ store, mirror: null, config: {} }); const events = []; const calls = []; @@ -423,6 +426,94 @@ test("does not call the LLM when no event was added after the last successful se assert.equal(ranges[1][0], 2, "the successful event seq is passed as the next snapshot lower bound"); }); +test("resumes a persisted seq cursor after the summarizer is restarted", async () => { + const dir = mkdtempSync(join(tmpdir(), "dsh-mneme-distill-cursor-")); + const dbPath = join(dir, "memory.db"); + let firstStore; + let firstSummarizer; + let secondStore; + let secondSummarizer; + try { + firstStore = createStore(dbPath); + const first = setup({ distillRateLimitIntervalMs: 0 }, { store: firstStore }); + firstSummarizer = first.summarizer; + const session = { + id: "s-restart", + requestHeader: () => ({ config: { provider: "deepseek", model: "deepseek-chat" } }), + events: [userMessage("重启前的窗口", 1), { seq: 2, type: "turn/end" }] + }; + const firstHandler = first.events.find((e) => e.name === "session/event").fn; + await firstHandler(session, { seq: 2, type: "turn/end" }); + assert.equal(first.calls.length, 1); + assert.equal(firstStore.getDistillCursor("s-restart").last_seq, 2); + + firstSummarizer.dispose(); + firstStore.close(); + firstSummarizer = undefined; + firstStore = undefined; + + secondStore = createStore(dbPath); + const second = setup({ distillRateLimitIntervalMs: 0 }, { store: secondStore }); + secondSummarizer = second.summarizer; + const secondHandler = second.events.find((e) => e.name === "session/event").fn; + await secondHandler(session, { seq: 2, type: "turn/end" }); + + assert.equal(second.calls.length, 0, "a restarted summarizer must reuse the persisted cursor"); + assert.equal(secondStore.count(), 2, "the restart must not duplicate memories"); + assert.equal(secondStore.getDistillCursor("s-restart").last_seq, 2); + } finally { + secondSummarizer?.dispose(); + secondStore?.close(); + firstSummarizer?.dispose(); + firstStore?.close(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("treats a valid empty summary as a successful cursor commit", async () => { + const { events, store, calls } = setup( + { distillRateLimitIntervalMs: 0 }, + { stream: streamOf([]) } + ); + const handler = events.find((e) => e.name === "session/event").fn; + const session = { + id: "s-empty-summary", + requestHeader: () => ({ config: { provider: "deepseek", model: "deepseek-chat" } }), + events: [userMessage("这一轮没有可保存的长期记忆", 1), { seq: 2, type: "turn/end" }] + }; + + await handler(session, { seq: 2, type: "turn/end" }); + assert.equal(calls.length, 1); + assert.equal(store.count(), 0); + assert.equal(store.getDistillCursor(session.id).last_seq, 2); + + await handler(session, { seq: 2, type: "turn/end" }); + assert.equal(calls.length, 1, "a valid empty summary must consume the window once"); +}); + +test("advances the cursor to the window end for mixed valid and invalid entries", async () => { + const valid = { type: "history", title: "有效摘要", content: "只保存这一条", importance: 3 }; + const { events, store, calls } = setup( + { distillRateLimitIntervalMs: 0 }, + { stream: streamOf([valid, { type: "unknown", title: "无效摘要", content: "忽略" }, null]) } + ); + const handler = events.find((e) => e.name === "session/event").fn; + const session = { + id: "s-mixed-summary", + requestHeader: () => ({ config: { provider: "deepseek", model: "deepseek-chat" } }), + events: [userMessage("混合摘要窗口", 1), { seq: 2, type: "assistant/message" }, { seq: 3, type: "turn/end" }] + }; + + await handler(session, { seq: 3, type: "turn/end" }); + assert.equal(calls.length, 1); + assert.equal(store.count(), 1); + assert.equal(store.all()[0].title, valid.title); + assert.equal(store.getDistillCursor(session.id).last_seq, 3); + + await handler(session, { seq: 3, type: "turn/end" }); + assert.equal(calls.length, 1, "a mixed summary must consume the window once"); +}); + test("distills only new seq events and keeps the recent tail when a window exceeds distillMaxChars", async () => { const { events, calls } = setup( { distillMaxChars: 80, distillRateLimitIntervalMs: 0 }, @@ -654,10 +745,12 @@ test("rolls back partial memory writes so retrying a failed window does not dupl await handler(session, { seq: 2, type: "turn/end" }); assert.equal(calls.length, 1); assert.equal(store.count(), 0, "a failed write transaction must leave no partial memory"); + assert.equal(store.getDistillCursor("s-partial-write"), undefined, "a failed write must not advance the cursor"); await handler(session, { seq: 2, type: "turn/end" }); assert.equal(calls.length, 2, "the failed seq window must be retried"); assert.equal(store.count(), 2); + assert.equal(store.getDistillCursor("s-partial-write").last_seq, 2); assert.equal(store.all().find((memory) => memory.title === "第一条")?.content, "第一条内容"); await handler(session, { seq: 2, type: "turn/end" }); @@ -665,6 +758,42 @@ test("rolls back partial memory writes so retrying a failed window does not dupl assert.equal(store.count(), 2); }); +test("rolls back memory writes when persisting the cursor fails", async () => { + const { events, store, service, calls } = setup( + { distillRateLimitIntervalMs: 0 }, + { stream: streamOf([{ type: "history", title: "游标失败", content: "事务应回滚", importance: 3 }]) } + ); + const originalSetCursor = service.setDistillCursor.bind(service); + let failOnce = true; + service.setDistillCursor = (sessionId, lastSeq) => { + if (failOnce) { + failOnce = false; + throw new Error("simulated cursor write failure"); + } + return originalSetCursor(sessionId, lastSeq); + }; + const handler = events.find((e) => e.name === "session/event").fn; + const session = { + id: "s-cursor-write-failure", + requestHeader: () => ({ config: { provider: "deepseek", model: "deepseek-chat" } }), + events: [userMessage("游标写入失败也不能丢窗口", 1), { seq: 2, type: "turn/end" }] + }; + + await handler(session, { seq: 2, type: "turn/end" }); + assert.equal(calls.length, 1); + assert.equal(store.count(), 0, "cursor failure must roll back the memory write"); + assert.equal(store.getDistillCursor(session.id), undefined); + const failedAudit = service.listLlmAudits({ source: "autoSummarize" }).find( + (audit) => audit.error_message === "simulated cursor write failure" + ); + assert.equal(failedAudit?.status, "error", "cursor failure must not be audited as success"); + + await handler(session, { seq: 2, type: "turn/end" }); + assert.equal(calls.length, 2, "the failed cursor window must be retryable"); + assert.equal(store.count(), 1); + assert.equal(store.getDistillCursor(session.id).last_seq, 2); +}); + test("uses summarizeProvider/summarizeModel config override when set", async () => { const { events, calls } = setup({ summarizeProvider: "aliyun",