Skip to content

fix(summarize): 持久化增量蒸馏游标（#229） #274

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
modusensus merged 1 commit into from
Sep 21, 2026
Merged

fix(summarize): 持久化增量蒸馏游标（#229） #274

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 2 additions & 0 deletions dsh-mneme/lib/service.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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; },
Expand Down
41 changes: 41 additions & 0 deletions dsh-mneme/lib/store.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = [];
Expand Down Expand Up @@ -2461,6 +2500,8 @@ export function createStore(path) {
getRecallEval,
listRecallEvals,
saveLlmAudit,
getDistillCursor,
setDistillCursor,
listLlmAudits,
countLlmAudits,
getLlmAuditStats,
Expand Down
32 changes: 29 additions & 3 deletions dsh-mneme/lib/summarize.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 截断时到点照跑
Expand Down Expand Up @@ -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 二次过滤。
Expand All @@ -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;
Expand All @@ -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",
Expand Down Expand Up @@ -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 = {
Expand All @@ -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 一致。
Expand Down
2 changes: 2 additions & 0 deletions dsh-mneme/src/service.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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; },
Expand Down
41 changes: 41 additions & 0 deletions dsh-mneme/src/store.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = [];
Expand Down Expand Up @@ -2461,6 +2500,8 @@ export function createStore(path) {
getRecallEval,
listRecallEvals,
saveLlmAudit,
getDistillCursor,
setDistillCursor,
listLlmAudits,
countLlmAudits,
getLlmAuditStats,
Expand Down
32 changes: 29 additions & 3 deletions dsh-mneme/src/summarize.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 截断时到点照跑
Expand Down Expand Up @@ -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 二次过滤。
Expand All @@ -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;
Expand All @@ -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",
Expand Down Expand Up @@ -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 = {
Expand All @@ -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 一致。
Expand Down
13 changes: 13 additions & 0 deletions dsh-mneme/test/store.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
});

Expand Down Expand Up @@ -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 });
Expand Down
Loading
Loading