feat: implement chain memory organization and computer evidence recall
This commit is contained in:
18962558917
parent
cd386329b3
commit
3fc24c1fc1
46
README.md
46
README.md
@ -1,14 +1,52 @@
|
||||
# deskpaw-memory
|
||||
|
||||
## Super Pi v2 时间轴
|
||||
## 当前：链式记忆与 Computer 增量经历
|
||||
|
||||
SQLite 保存原始事件、活动时间轴节点、长期对象图、检索词项、核心文件版本和持久整理队列。Super 主动同步召回；Memory 独占长期记忆语义整理权。无 embedding、无旧索引依赖、无联系人隔离。2026-10-12 21:26 已从原目录重新部署，累计修改整理合入本地 main，未推送远端。
|
||||
|
||||
最近 Agent/Memory 所有改动、实际参数、备份、验证与 main 合并记录见相邻 Agent 仓库的 [发布汇总](../deskpaw_agent/AGENT_MEMORY_RELEASE_2026-10-12.md)（本机 `/home/kickpi/deskpaw_agent/AGENT_MEMORY_RELEASE_2026-10-12.md`）。本轮 Memory 47/47、Agent 91/91 回归通过。下文旧分支、旧测试数量和 HTTP 402 是历史记录，不代表当前部署或余额状态。
|
||||
|
||||
Computer 现在直接将执行文本增量写入共享时间轴，不等整个任务结束；本机持久队列由 Agent 管理，断线重试。普通事件不唤醒 Super，长期整理仍采用阈值、主动和夜间三类入口。`timeline.query` 新增可选 `task_id` 及对应 SQLite 索引；Computer 可只读使用长期记忆，任务内检查点不是本服务的长期整理。
|
||||
|
||||
完整跨仓库说明与自查位于相邻 Agent 仓库的 `CHAIN_MEMORY.md`（本机 `/home/kickpi/deskpaw_agent/CHAIN_MEMORY.md`）。
|
||||
|
||||
- `chain-store.js`：原始事件分片建链；达到阈值、Super 主动请求、每日 02:50 且五分钟无对话三种触发。compact 局部提取；deep 遍历时间轴和首次领取时的全库对象快照，包括归档及近期无关对象。审阅进度持久化，失败不推进，纯对象批次不制造经历。版本、租约、退避、覆盖校验及事务发布。
|
||||
- `chain-graph.js`：人物、任务、待办、日程、知识、交叉项目、凭据引用；来源和版本，别名/词项/cues 索引、有限关系展开、分页读取。新项目为候选，需逐字用户证据确认。精简目录常驻 Super，事实按需召回。
|
||||
- `core-projection.js`：SOUL/USER/AGENTS 三份文件，每份最多 2,400 字符；数据库权威版本与可编辑文件投影，下一轮热加载。凭据真实值禁止存入。
|
||||
|
||||
整理模型由 Agent 进程中的独立 `MemoryWorker` 调用，不是 Super 会话，也不是 Computer Pi。默认沿用 Super 模型；可分别配置 `DESKPAW_MEMORY_MODEL_PROVIDER/MODEL/BASE_URL/API_KEY`，日期使用 `DESKPAW_MEMORY_TIMEZONE`（默认 Asia/Shanghai）。本服务负责读取、校验和提交，不直接调用模型。
|
||||
|
||||
新增 JSON-line 方法：
|
||||
|
||||
| 方法 | 用途 |
|
||||
|---|---|
|
||||
| `memory.chain.context` | max_tokens 估算 token 预算与独立 max_bytes 传输限额；返回最近节点、truncated/omitted_nodes。仅同步未就绪返回 ready=false |
|
||||
| `memory.chain.find/read/node` | 对象目录检索、详情与关系、原始证据按范围展开 |
|
||||
| `memory.chain.browse` | after/limit 按稳定 ID 分页浏览全库，包括归档对象；供 Memory 全局整理使用 |
|
||||
| `memory.chain.tick/status` | 触发检查、整理状态和容量/投影诊断 |
|
||||
| `memory.chain.core.bootstrap/update` | 首次核心导入、受信管理端按 expected_revision 更新 |
|
||||
| `memory.chain.project_decision` | 当前用户证据确认/放弃候选项目 |
|
||||
| `memory.organize.request/claim/inspect/commit/fail` | 持久请求、领取有限批次、补读旧记忆、原子发布、失败退避 |
|
||||
|
||||
长期内容和摘要同一事务发布；重试回执幂等，旧租约不得覆盖新版本。原文、审计和版本不会因摘要缩短而删除，因此仍需磁盘监控和备份。未实现自动回收、会议采集、任意同义检索保证或密钥库全量元数据同步。Memory 故障保留 Agent 待同步内容，不阻止接收任务。
|
||||
|
||||
`token-budget.js` 与 Agent 使用同一 portable-v1 文本 token 估算规则，未引入模型专属 tokenizer。`tick.threshold_tokens` 和 `stats.estimated_tokens` 使用 token 单位，`stats.bytes` 仅为存储/传输诊断，二者不能相加。Super 超限时截断输入视图并向模型明确说明，原始记忆不删除，语义整理仍由 Memory 完成。`status.jobs[].object_review` 表示深度任务的对象总量和已成功提交的审阅量，不证明模型语义判断必然正确。
|
||||
|
||||
测试：`npm test`，使用隔离数据库、临时核心文件和 socket；包括原子回滚、重启租约、夜间安静窗口、大事件跨批不丢尾部、项目确认、中文召回、敏感值拒绝及投影热更新。真实模型语义质量、长稳与断电需另外验收。
|
||||
|
||||
2026-10-12：部署前本仓库 42 项测试全部通过，相邻 Agent 10 组定向测试 76 项通过。部署实测后新增失败缩批回归，本仓库 **43 项**全部通过；Agent 完整链路补测输出截断重试和同轮多查询。服务已重启，未推送仓库。compact 初始最多八个节点、deep 最多十二个，失败重试会进一步缩小，不丢弃未处理片段。
|
||||
|
||||
后续修改加入 token 口径、显式截断和全库深度审阅，**46 项隔离测试全部通过**，现已按用户要求切换到线上进程。服务健康不等于真实模型验收通过；本次接口返回余额不足，详情见 Agent 的部署记录。
|
||||
|
||||
## 原始时间轴协议（仍为当前链式记忆的数据底座）
|
||||
|
||||
`feat/super-agent-v2` 新增独立经历时间轴，使用 `timeline` 与 `timeline_terms` 表，不查询下文的旧任务/话题索引，不使用 embedding。Super v2 不再启动下层夜间旧记忆整理；以下原有接口保留供旧调用方和数据审计使用。
|
||||
|
||||
新 JSON-line 接口为 `timeline.append/query/read/expand`。append 必须包含稳定 `id`、`kind`、毫秒时间 `at` 和不超过 32,000 字符的 `text`，`source` 仅接收来源/任务/文件引用白名单字段。敏感文本经过过滤，重复 ID 不覆盖原始记录。长工具结果保留在 Agent 作业表，时间轴只保存有限摘要和引用。
|
||||
JSON-line 接口为 `timeline.append/query/read/expand`。append 包含稳定 `id`、`kind`、毫秒时间 `at`（省略时使用进入时间）和不超过 32,000 字符的 `text`，`source` 仅接收来源/任务/文件引用白名单字段。敏感文本经过过滤，重复 ID 不覆盖原始记录。长工具结果保留在 Agent 作业表，时间轴只保存有限摘要和引用。
|
||||
|
||||
query 支持 `query/from/to/after/limit`：中英文词项倒排检索、时间范围、最多 50 个索引头，每头最多 1,200 字符；命中数和时间排序。无查询词时按 seq 正序分页。read/expand 接受 `id/offset/limit` 返回原文范围与 next_offset。没有用户分区。
|
||||
query 支持 `query/from/to/after/limit/task_id`：中英文词项倒排检索、时间范围、可选任务过滤、最多 50 个索引头，每头最多 1,200 字符；命中数和时间排序。无查询词时按 seq 正序分页。read/expand 接受 `id/offset/limit` 返回原文范围与 next_offset。没有用户分区。
|
||||
|
||||
预留 parent_id、covered_from、covered_to、version，尚未实现自动会议归并、记忆图提取、压缩稀疏化、删除/遗忘、磁盘容量策略。原始记录会增长，必须监控磁盘。SQLite 是唯一权威来源。Agent 的持久待同步表保证服务离线后可补写；不阻塞上层聊天。
|
||||
原始行保持不可变，压缩与子节点关系使用新的 `memory_nodes`，不改写原文。旧 `parent_id/covered_from/covered_to` 字段不承担当前链式节点发布。Agent 的持久待同步表支持离线补写。
|
||||
|
||||
服务启动会检查已有 socket 所有者，不删除活跃实例的 socket；单请求连接限制 1 MB、5 秒和最多 64 并发连接。测试执行 `npm test`，新增时间轴存储、敏感过滤和真实 Unix socket 的隔离测试，不操作生产状态。
|
||||
|
||||
|
||||
188
src/chain-graph.js
Normal file
188
src/chain-graph.js
Normal file
@ -0,0 +1,188 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { words } from "./policy.js";
|
||||
|
||||
export const OBJECT_TYPES = ["person", "task", "todo", "schedule", "knowledge", "project", "credential"];
|
||||
export const CORE_NAMES = ["SOUL.md", "USER.md", "AGENTS.md"];
|
||||
export function key(value) {
|
||||
if (typeof value !== "string" || !/^[\w:./-]{1,240}$/.test(value)) throw new Error("Invalid memory ID");
|
||||
return value;
|
||||
}
|
||||
export function sourceKey(value) {
|
||||
if (typeof value !== "string" || !value || value.length > 240) throw new Error("Invalid evidence ID");
|
||||
return value;
|
||||
}
|
||||
export function integer(value, min, max) {
|
||||
if (!Number.isSafeInteger(value) || value < min || value > max) throw new Error("Invalid memory bound");
|
||||
return value;
|
||||
}
|
||||
// Names of credentials and vault references are allowed; credential VALUES are not.
|
||||
export function clean(value, max = 8000) {
|
||||
if (typeof value !== "string" || value.length > max) throw new Error("Invalid memory text length");
|
||||
if (/-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----|\bsk-[\w.-]{12,}|\bBearer\s+[\w./+=-]{12,}|(?:password|passwd|api[_ -]?key|access[_ -]?token|refresh[_ -]?token|client[_ -]?secret|密码|口令|密钥)["']?\s*(?:[=:：]|是|为)\s*["']?[^\s"',}]{4,}/i.test(value))
|
||||
throw new Error("Credential values cannot enter memory");
|
||||
return value.trim();
|
||||
}
|
||||
function list(value, max) {
|
||||
if (!Array.isArray(value) || value.length > max) throw new Error("Invalid memory list");
|
||||
return value;
|
||||
}
|
||||
function normalize(value) { return value.normalize("NFKC").toLowerCase().trim(); }
|
||||
|
||||
/** Typed objects and edges are addressing structures, not additional prompt layers. */
|
||||
export class ChainGraph {
|
||||
constructor(db) {
|
||||
this.db = db;
|
||||
db.exec(`CREATE TABLE IF NOT EXISTS memory_objects (
|
||||
id TEXT PRIMARY KEY,type TEXT NOT NULL,name TEXT NOT NULL,summary TEXT NOT NULL,body TEXT NOT NULL,
|
||||
aliases TEXT NOT NULL,cues TEXT NOT NULL,source_refs TEXT NOT NULL,status TEXT NOT NULL,
|
||||
revision INTEGER NOT NULL,updated_at INTEGER NOT NULL,secret_ref TEXT);
|
||||
CREATE TABLE IF NOT EXISTS memory_object_versions (id TEXT,revision INTEGER,data TEXT NOT NULL,PRIMARY KEY(id,revision));
|
||||
CREATE TABLE IF NOT EXISTS memory_object_sources (id TEXT,event_id TEXT,PRIMARY KEY(id,event_id));
|
||||
CREATE TABLE IF NOT EXISTS memory_aliases (alias TEXT,id TEXT,PRIMARY KEY(alias,id));
|
||||
CREATE TABLE IF NOT EXISTS memory_search_terms (term TEXT,id TEXT,channel TEXT,PRIMARY KEY(term,id,channel));
|
||||
CREATE INDEX IF NOT EXISTS memory_search_object ON memory_search_terms(id);
|
||||
CREATE TABLE IF NOT EXISTS memory_edges (id TEXT PRIMARY KEY,from_id TEXT NOT NULL,relation TEXT NOT NULL,
|
||||
to_id TEXT NOT NULL,source_refs TEXT NOT NULL,status TEXT NOT NULL,revision INTEGER NOT NULL);
|
||||
CREATE INDEX IF NOT EXISTS memory_edges_from ON memory_edges(from_id,status);
|
||||
CREATE INDEX IF NOT EXISTS memory_edges_to ON memory_edges(to_id,status);
|
||||
CREATE TABLE IF NOT EXISTS memory_edge_versions (id TEXT,revision INTEGER,data TEXT,PRIMARY KEY(id,revision));
|
||||
CREATE TABLE IF NOT EXISTS memory_core (name TEXT PRIMARY KEY,body TEXT NOT NULL,revision INTEGER NOT NULL,source_refs TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS memory_core_versions (name TEXT,revision INTEGER,body TEXT,source_refs TEXT,PRIMARY KEY(name,revision));
|
||||
CREATE TABLE IF NOT EXISTS memory_catalog (type TEXT PRIMARY KEY,summary TEXT NOT NULL,revision INTEGER NOT NULL);`);
|
||||
db.prepare("INSERT OR IGNORE INTO memory_objects VALUES ('project:daily','project','日常','日常事务与尚未明确项目归属的内容','这是内置关联集合，不代表用户批准了新的独立项目。','[]','[]','[]','active',0,0,NULL)").run();
|
||||
db.prepare("INSERT OR IGNORE INTO memory_aliases VALUES ('日常','project:daily')").run();
|
||||
db.exec("INSERT OR IGNORE INTO memory_object_sources SELECT o.id,j.value FROM memory_objects o,json_each(o.source_refs) j");
|
||||
}
|
||||
object(id) {
|
||||
const row = this.db.prepare("SELECT * FROM memory_objects WHERE id=?").get(key(id));
|
||||
return row ? { ...row, aliases: JSON.parse(row.aliases), cues: JSON.parse(row.cues), source_refs: JSON.parse(row.source_refs) } : null;
|
||||
}
|
||||
core() { return this.db.prepare("SELECT * FROM memory_core ORDER BY name").all(); }
|
||||
saveCore(name, body, refs, revision) {
|
||||
if (!CORE_NAMES.includes(name)) throw new Error("Invalid core file");
|
||||
body = clean(body, 2400);
|
||||
this.db.prepare("INSERT OR REPLACE INTO memory_core VALUES (?,?,?,?)").run(name, body, revision, JSON.stringify(refs));
|
||||
this.db.prepare("INSERT INTO memory_core_versions VALUES (?,?,?,?)").run(name, revision, body, JSON.stringify(refs));
|
||||
}
|
||||
apply(patch, revision, evidence, now) {
|
||||
const refs = (values) => {
|
||||
const result = [...new Set(list(values, 100).map(sourceKey))];
|
||||
if (!result.length || result.some(id => !evidence.has(id))) throw new Error("Evidence outside claimed batch");
|
||||
if (result.some(id => ["memory_recall", "memory_organized", "organize_request", "organize_result"].includes(evidence.get(id).kind)))
|
||||
throw new Error("Recall and maintenance records are not independent evidence");
|
||||
return result;
|
||||
};
|
||||
const objects = list(patch.objects ?? [], 32);
|
||||
for (const item of objects) {
|
||||
const id = key(item.id);
|
||||
if (!OBJECT_TYPES.includes(item.type)) throw new Error("Invalid memory object type");
|
||||
const previous = this.object(id);
|
||||
if (previous && previous.type !== item.type) throw new Error("Object type is immutable");
|
||||
const sourceRefs = refs(item.source_refs);
|
||||
const name = clean(item.name, 160);
|
||||
const summary = clean(item.summary, 400);
|
||||
const body = clean(item.body, 6000);
|
||||
if (!name || !summary) throw new Error("Name and summary required");
|
||||
const aliases = list(item.aliases ?? [], 12).map(x => clean(x, 120));
|
||||
const cues = list(item.cues ?? [], 8).map(x => clean(x, 160));
|
||||
let status = item.status ?? "active";
|
||||
if (!["active", "candidate", "archived"].includes(status)) throw new Error("Invalid object status");
|
||||
// A model cannot approve a project on the user's behalf.
|
||||
if (item.type === "project" && id !== "project:daily" && previous?.status !== "active") status = "candidate";
|
||||
if (item.type === "project" && previous?.status === "archived" && this.db.prepare("SELECT 1 FROM memory_project_decisions WHERE project_id=? AND status='archived' LIMIT 1").get(id)) status = "archived";
|
||||
const secret = item.type === "credential" && item.secret_ref ? key(clean(item.secret_ref, 240)) : null;
|
||||
const allRefs = [...new Set([...(previous?.source_refs ?? []), ...sourceRefs])].slice(-100);
|
||||
for (const ref of sourceRefs) this.db.prepare("INSERT OR IGNORE INTO memory_object_sources VALUES (?,?)").run(id, ref);
|
||||
this.db.prepare("INSERT OR REPLACE INTO memory_objects VALUES (?,?,?,?,?,?,?,?,?,?,?,?)")
|
||||
.run(id, item.type, name, summary, body, JSON.stringify(aliases), JSON.stringify(cues), JSON.stringify(allRefs), status, revision, now, secret);
|
||||
this.db.prepare("INSERT INTO memory_object_versions VALUES (?,?,?)").run(id, revision, JSON.stringify(this.object(id)));
|
||||
this.db.prepare("DELETE FROM memory_aliases WHERE id=?").run(id);
|
||||
this.db.prepare("DELETE FROM memory_search_terms WHERE id=?").run(id);
|
||||
for (const alias of [name, ...aliases]) this.db.prepare("INSERT OR IGNORE INTO memory_aliases VALUES (?,?)").run(normalize(alias), id);
|
||||
for (const [channel, text] of [["fact", [name, ...aliases, summary, body].join(" ")], ["cue", cues.join(" ")]])
|
||||
for (const term of words(text)) this.db.prepare("INSERT OR IGNORE INTO memory_search_terms VALUES (?,?,?)").run(term, id, channel);
|
||||
}
|
||||
for (const item of list(patch.edges ?? [], 64)) {
|
||||
const from = key(item.from_id), to = key(item.to_id);
|
||||
if (!this.object(from) || !this.object(to)) throw new Error("Dangling memory edge");
|
||||
const relation = clean(item.relation, 80);
|
||||
if (!relation) throw new Error("Empty relation");
|
||||
const id = createHash("sha256").update(JSON.stringify([from, relation, to])).digest("hex");
|
||||
if (!["active", "archived"].includes(item.status ?? "active")) throw new Error("Invalid edge status");
|
||||
this.db.prepare("INSERT OR REPLACE INTO memory_edges VALUES (?,?,?,?,?,?,?)")
|
||||
.run(id, from, relation, to, JSON.stringify(refs(item.source_refs)), item.status ?? "active", revision);
|
||||
this.db.prepare("INSERT INTO memory_edge_versions VALUES (?,?,?)").run(id, revision, JSON.stringify(this.db.prepare("SELECT * FROM memory_edges WHERE id=?").get(id)));
|
||||
}
|
||||
for (const item of list(patch.core ?? [], 3)) {
|
||||
const sourceRefs = refs(item.source_refs);
|
||||
// Exact user evidence is required for identity/personality; environment may use executor evidence.
|
||||
if (item.name !== "AGENTS.md" && (!item.user_quote || !sourceRefs.some(id => {
|
||||
const event = evidence.get(id);
|
||||
return event.kind === "user" && event.text.includes(clean(item.user_quote, 1000));
|
||||
}))) throw new Error("Core identity change requires exact user evidence");
|
||||
this.saveCore(item.name, item.body, sourceRefs, revision);
|
||||
}
|
||||
for (const item of list(patch.catalog ?? [], 7)) {
|
||||
if (!OBJECT_TYPES.includes(item.type)) throw new Error("Invalid catalog type");
|
||||
this.db.prepare("INSERT OR REPLACE INTO memory_catalog VALUES (?,?,?)").run(item.type, clean(item.summary, 200), revision);
|
||||
}
|
||||
}
|
||||
catalog() {
|
||||
const entries = this.db.prepare(`SELECT type,count(*) AS count,sum(status='candidate') AS candidates
|
||||
FROM memory_objects WHERE status!='archived' GROUP BY type ORDER BY type`).all();
|
||||
const catalog = { entries: entries.map(row => ({ ...row,
|
||||
summary: this.db.prepare("SELECT summary FROM memory_catalog WHERE type=?").get(row.type)?.summary ?? "已有相关记忆，可按此类型检索",
|
||||
// A bounded preview makes existence concrete; counts and paginated listing cover the rest.
|
||||
examples: this.db.prepare("SELECT id,name,status FROM memory_objects WHERE type=? AND status!='archived' ORDER BY updated_at DESC,id LIMIT 3").all(row.type)
|
||||
})), instruction: "目录仅表示存在性，不是完整事实。用 memory_find 按名称、问题、type 或 project_id 同步召回；用 memory_read 读详情和证据。候选项目需先询问用户。" };
|
||||
while (Buffer.byteLength(JSON.stringify(catalog)) > 8000) {
|
||||
const row = [...catalog.entries].reverse().find(entry => entry.examples.length);
|
||||
if (!row) break;
|
||||
row.examples.pop();
|
||||
}
|
||||
return catalog;
|
||||
}
|
||||
find(p = {}) {
|
||||
const limit = integer(p.limit ?? 8, 1, 30), offset = integer(p.offset ?? 0, 0, 1000000);
|
||||
const query = clean(p.query ?? "", 1000);
|
||||
if (p.type && !OBJECT_TYPES.includes(p.type)) throw new Error("Invalid object type");
|
||||
const clauses = ["o.status!='archived'"], args = [];
|
||||
if (p.type) { clauses.push("o.type=?"); args.push(p.type); }
|
||||
if (p.project_id) {
|
||||
clauses.push("(o.id=? OR EXISTS(SELECT 1 FROM memory_edges e WHERE e.status='active' AND ((e.from_id=o.id AND e.to_id=?) OR (e.to_id=o.id AND e.from_id=?))))");
|
||||
args.push(key(p.project_id), p.project_id, p.project_id);
|
||||
}
|
||||
const terms = [...words(query)].slice(0, 64);
|
||||
let rows;
|
||||
if (!query) rows = this.db.prepare(`SELECT o.*,0 AS score FROM memory_objects o WHERE ${clauses.join(" AND ")} ORDER BY o.type,o.name,o.id LIMIT ? OFFSET ?`).all(...args, limit, offset);
|
||||
else {
|
||||
const placeholders = terms.map(() => "?").join(",") || "NULL";
|
||||
rows = this.db.prepare(`WITH candidates AS (
|
||||
SELECT id,1000 AS weight FROM memory_aliases WHERE alias=?
|
||||
UNION ALL SELECT id,CASE channel WHEN 'fact' THEN 3 ELSE 1 END FROM memory_search_terms WHERE term IN (${placeholders})
|
||||
) SELECT o.*,sum(c.weight) AS score FROM candidates c JOIN memory_objects o ON o.id=c.id
|
||||
WHERE ${clauses.join(" AND ")} GROUP BY o.id ORDER BY score DESC,o.id LIMIT ? OFFSET ?`)
|
||||
.all(normalize(query), ...terms, ...args, limit, offset);
|
||||
}
|
||||
return { hits: rows.map(({ id, type, name, summary, status, revision, score }) => ({ id, type, name, summary, status, revision, score })),
|
||||
next_offset: rows.length === limit ? offset + limit : null, method: "aliases_terms_cues_graph", embedding: false };
|
||||
}
|
||||
browse(p = {}) {
|
||||
const limit = integer(p.limit ?? 8, 1, 20);
|
||||
const after = p.after ? key(p.after) : "";
|
||||
const rows = this.db.prepare("SELECT id,type,name,summary,status,revision FROM memory_objects WHERE id>? ORDER BY id LIMIT ?").all(after, limit + 1);
|
||||
return { entries: rows.slice(0, limit), next_after: rows.length > limit ? rows[limit - 1].id : null,
|
||||
includes_archived: true, revision: Number(this.db.prepare("SELECT value FROM memory_chain_meta WHERE key='revision'").get()?.value ?? 0) };
|
||||
}
|
||||
read(p) {
|
||||
const item = this.object(p.id);
|
||||
if (!item) throw new Error("Memory object not found");
|
||||
const offset = integer(p.offset ?? 0, 0, 1000000), limit = integer(p.limit ?? 20, 1, 50);
|
||||
const edges = this.db.prepare("SELECT * FROM memory_edges WHERE status='active' AND (from_id=? OR to_id=?) ORDER BY id LIMIT ? OFFSET ?").all(item.id, item.id, limit, offset);
|
||||
// Evidence lists can grow; never return the whole list implicitly.
|
||||
const total = this.db.prepare("SELECT count(*) AS n FROM memory_object_sources WHERE id=?").get(item.id).n;
|
||||
return { ...item, source_refs: this.db.prepare("SELECT event_id FROM memory_object_sources WHERE id=? ORDER BY event_id LIMIT ? OFFSET ?").all(item.id, limit, offset).map(row => row.event_id), total_sources: total,
|
||||
edges: edges.map(row => ({ ...row, source_refs: JSON.parse(row.source_refs).slice(0, 20) })),
|
||||
next_offset: edges.length === limit || offset + limit < total ? offset + limit : null };
|
||||
}
|
||||
}
|
||||
331
src/chain-store.js
Normal file
331
src/chain-store.js
Normal file
@ -0,0 +1,331 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { ChainGraph, clean, CORE_NAMES, integer, key, sourceKey } from "./chain-graph.js";
|
||||
import { estimateTokens } from "./token-budget.js";
|
||||
|
||||
/** One personal timeline; raw evidence is immutable, active heads are versioned projections. */
|
||||
export class ChainStore {
|
||||
constructor(db, timeline) {
|
||||
this.db = db;
|
||||
this.timeline = timeline;
|
||||
this.graph = new ChainGraph(db);
|
||||
db.function("memory_token_cost", { deterministic: true }, value => estimateTokens(value));
|
||||
db.exec(`CREATE TABLE IF NOT EXISTS memory_chain_meta (key TEXT PRIMARY KEY,value TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS memory_nodes (
|
||||
id TEXT PRIMARY KEY,start_seq INTEGER NOT NULL,end_seq INTEGER NOT NULL,at INTEGER NOT NULL,
|
||||
text TEXT NOT NULL,event_id TEXT,children TEXT NOT NULL,active INTEGER NOT NULL,revision INTEGER NOT NULL);
|
||||
CREATE INDEX IF NOT EXISTS memory_nodes_active ON memory_nodes(active,start_seq,id);
|
||||
CREATE TABLE IF NOT EXISTS memory_organize_jobs (
|
||||
id TEXT PRIMARY KEY,dedupe TEXT UNIQUE NOT NULL,mode TEXT NOT NULL,trigger TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,status TEXT NOT NULL,cutoff INTEGER NOT NULL,cursor INTEGER NOT NULL DEFAULT 0,
|
||||
token TEXT,lease_until INTEGER NOT NULL DEFAULT 0,available_at INTEGER NOT NULL DEFAULT 0,
|
||||
attempts INTEGER NOT NULL DEFAULT 0,snapshot TEXT,receipt TEXT,error TEXT,created_at INTEGER NOT NULL);
|
||||
CREATE INDEX IF NOT EXISTS memory_organize_pending ON memory_organize_jobs(status,available_at);
|
||||
CREATE TABLE IF NOT EXISTS memory_organize_batches (token TEXT PRIMARY KEY,job_id TEXT,receipt TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS memory_job_nodes (job_id TEXT,node_id TEXT,PRIMARY KEY(job_id,node_id));
|
||||
CREATE TABLE IF NOT EXISTS memory_job_objects (job_id TEXT,object_id TEXT,reviewed INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(job_id,object_id));
|
||||
CREATE TABLE IF NOT EXISTS memory_job_inventory (job_id TEXT PRIMARY KEY);
|
||||
CREATE TABLE IF NOT EXISTS memory_project_decisions (request_id TEXT PRIMARY KEY,project_id TEXT,status TEXT,source_id TEXT);`);
|
||||
}
|
||||
meta(name, fallback = "") { return this.db.prepare("SELECT value FROM memory_chain_meta WHERE key=?").get(name)?.value ?? fallback; }
|
||||
set(name, value) { this.db.prepare("INSERT OR REPLACE INTO memory_chain_meta VALUES (?,?)").run(name, String(value)); }
|
||||
revision() { return Number(this.meta("revision", "0")); }
|
||||
transaction(fn) {
|
||||
this.db.exec("BEGIN IMMEDIATE");
|
||||
try { const value = fn(); this.db.exec("COMMIT"); return value; }
|
||||
catch (error) { this.db.exec("ROLLBACK"); throw error; }
|
||||
}
|
||||
sync() {
|
||||
// Bounded migration/backfill; called until caught up before serving a snapshot or claiming a job.
|
||||
const rows = this.db.prepare("SELECT * FROM timeline WHERE seq>? ORDER BY seq LIMIT 100").all(Number(this.meta("ingested", "0")));
|
||||
this.transaction(() => {
|
||||
for (const row of rows) {
|
||||
for (let offset = 0; offset < Math.max(1, row.text.length); offset += 2000) {
|
||||
const id = `event:${row.seq}:${String(offset).padStart(6, "0")}`;
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_nodes VALUES (?,?,?,?,?,?,?,?,?)")
|
||||
.run(id, row.seq, row.seq, row.at, row.text.slice(offset, offset + 2000), row.id, "[]", 1, this.revision());
|
||||
}
|
||||
this.set("ingested", row.seq);
|
||||
}
|
||||
});
|
||||
return rows.length;
|
||||
}
|
||||
caughtUp() { return Number(this.meta("ingested", "0")) === Number(this.db.prepare("SELECT coalesce(max(seq),0) AS n FROM timeline").get().n); }
|
||||
stats() {
|
||||
const row = this.db.prepare(`SELECT count(*) AS nodes,coalesce(sum(length(cast(json_quote(n.text) AS BLOB))+
|
||||
coalesce(length(cast(t.source AS BLOB)),0)+length(n.id)+240),0) AS bytes,
|
||||
coalesce(sum(memory_token_cost(n.text)+memory_token_cost(coalesce(t.source,''))+memory_token_cost(n.id)+96),0) AS estimated_tokens
|
||||
FROM memory_nodes n LEFT JOIN timeline t ON t.id=n.event_id WHERE n.active=1`).get();
|
||||
return { ...row, estimator: "portable-v1", revision: this.revision(), ingested: Number(this.meta("ingested", "0")), caught_up: this.caughtUp() };
|
||||
}
|
||||
context(p = {}) {
|
||||
this.sync();
|
||||
const budget = integer(p.max_bytes ?? 128000, 4096, 350000);
|
||||
const tokenBudget = integer(p.max_tokens ?? 128000, 512, 1000000);
|
||||
const stats = this.stats();
|
||||
const catalog = this.graph.catalog(), core = this.graph.core();
|
||||
if (!stats.caught_up) return { ...stats, ready: false, catalog, core, reason: "backfill_pending" };
|
||||
const nodes = [];
|
||||
let bytes = 2, tokens = 2;
|
||||
for (const row of this.db.prepare(`SELECT n.id,n.at,n.text,n.event_id,n.revision,t.kind,t.source
|
||||
FROM memory_nodes n LEFT JOIN timeline t ON t.id=n.event_id WHERE n.active=1 ORDER BY n.at DESC,n.start_seq DESC,n.id DESC`).iterate()) {
|
||||
const node = { ...row, kind: row.kind ?? "summary", source: row.source ? JSON.parse(row.source) : null };
|
||||
const size = Buffer.byteLength(JSON.stringify(node)) + 1, cost = estimateTokens(node) + 1;
|
||||
if (bytes + size > budget || tokens + cost > tokenBudget) break;
|
||||
nodes.push(node); bytes += size; tokens += cost;
|
||||
}
|
||||
const truncated = nodes.length < stats.nodes;
|
||||
return { ...stats, ready: true, catalog, core, nodes: nodes.reverse(), truncated,
|
||||
omitted_nodes: stats.nodes - nodes.length, view_estimated_tokens: tokens,
|
||||
reason: truncated ? "context_capacity" : null };
|
||||
}
|
||||
request(p, now = Date.now()) {
|
||||
const mode = p.mode ?? "compact", trigger = p.trigger ?? "explicit";
|
||||
if (!["compact", "deep"].includes(mode) || !["threshold", "explicit", "nightly"].includes(trigger)) throw new Error("Invalid organization mode/trigger");
|
||||
const dedupe = key(p.request_id), scope = clean(p.scope ?? "", 2000);
|
||||
if (trigger === "explicit") this.timeline.append({ id: `organize:${createHash("sha256").update(dedupe).digest("hex")}`,
|
||||
kind: "organize_request", at: now, text: JSON.stringify({ mode, scope }), source: { actor: "super" } });
|
||||
this.sync();
|
||||
const cutoff = Number(this.db.prepare("SELECT coalesce(max(seq),0) AS n FROM timeline").get().n);
|
||||
this.db.prepare(`INSERT OR IGNORE INTO memory_organize_jobs
|
||||
(id,dedupe,mode,trigger,scope,status,cutoff,created_at) VALUES (?,?,?,?,?,'pending',?,?)`)
|
||||
.run(randomUUID(), dedupe, mode, trigger, scope, cutoff, now);
|
||||
const job = this.db.prepare("SELECT id,status,mode,trigger,cutoff FROM memory_organize_jobs WHERE dedupe=?").get(dedupe);
|
||||
return job;
|
||||
}
|
||||
tick(p = {}, now = Date.now()) {
|
||||
this.sync();
|
||||
const threshold = integer(p.threshold_tokens ?? 96000, 8000, 300000);
|
||||
this.set("threshold_tokens", threshold);
|
||||
this.set("busy", p.busy === true ? 1 : 0);
|
||||
this.set("heartbeat", now);
|
||||
if (Number.isSafeInteger(p.last_dialogue_at)) this.set("last_dialogue", Math.max(Number(this.meta("last_dialogue", "0")), Math.min(p.last_dialogue_at, now)));
|
||||
const stats = this.stats();
|
||||
const pending = this.db.prepare("SELECT 1 FROM memory_organize_jobs WHERE status IN ('pending','running') AND trigger='threshold'").get();
|
||||
if (stats.caught_up && stats.estimated_tokens >= threshold && !pending)
|
||||
this.request({ request_id: `threshold:${stats.revision}:${stats.ingested}`, trigger: "threshold" }, now);
|
||||
const zone = p.timezone ?? "Asia/Shanghai";
|
||||
const parts = Object.fromEntries(new Intl.DateTimeFormat("en-CA", { timeZone: zone, year: "numeric", month: "2-digit", day: "2-digit", hour: "2-digit", minute: "2-digit", hourCycle: "h23" }).formatToParts(now).map(x => [x.type, x.value]));
|
||||
if (Number(parts.hour) * 60 + Number(parts.minute) >= 170) {
|
||||
const date = `${parts.year}-${parts.month}-${parts.day}`;
|
||||
// No all-history replay after several offline days: one current cutoff includes the backlog.
|
||||
const dedupe = `nightly:${date}`;
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status='superseded' WHERE trigger='nightly' AND status='pending' AND dedupe!=?").run(dedupe);
|
||||
this.request({ request_id: dedupe, mode: "deep", trigger: "nightly" }, now);
|
||||
// Deep review also covers dormant objects; no-new-events is not a reason to skip it.
|
||||
}
|
||||
return stats;
|
||||
}
|
||||
claim(p = {}, now = Date.now()) {
|
||||
this.sync();
|
||||
if (!this.caughtUp()) return { job: null };
|
||||
const maxBytes = integer(p.max_bytes ?? 96000, 24000, 200000);
|
||||
return this.transaction(() => {
|
||||
if (this.db.prepare("SELECT 1 FROM memory_organize_jobs WHERE status='running' AND lease_until>?").get(now)) return { job: null };
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status='pending',token=NULL,snapshot=NULL WHERE status='running' AND lease_until<=?").run(now);
|
||||
const busy = this.meta("busy") === "1", quiet = now - Number(this.meta("last_dialogue", "0")) >= 300000;
|
||||
if (busy) return { job: null };
|
||||
const job = this.db.prepare(`SELECT * FROM memory_organize_jobs WHERE status='pending' AND available_at<=?
|
||||
AND (trigger!='nightly' OR ?=1) ORDER BY CASE trigger WHEN 'threshold' THEN 0 WHEN 'explicit' THEN 1 ELSE 2 END,created_at LIMIT 1`)
|
||||
.get(now, Number(!busy && quiet));
|
||||
if (!job) return { job: null };
|
||||
// Inventory and cursor initialization share the lease transaction: a crash cannot
|
||||
// publish an incomplete inventory as a successfully reviewed library.
|
||||
if (job.mode === "deep" && this.db.prepare("INSERT OR IGNORE INTO memory_job_inventory VALUES (?)").run(job.id).changes)
|
||||
this.db.prepare("INSERT INTO memory_job_objects(job_id,object_id) SELECT ?,id FROM memory_objects").run(job.id);
|
||||
if (!job.cursor && job.trigger === "explicit") {
|
||||
job.cutoff = Number(this.meta("ingested", "0"));
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET cutoff=? WHERE id=?").run(job.cutoff, job.id);
|
||||
}
|
||||
// Leave output room for graph proposals, not only the input window. Failed batches
|
||||
// shrink deterministically across retries; consumed nodes are never skipped.
|
||||
const nodeLimit = Math.max(1, Math.floor((job.mode === "compact" ? 8 : 12) / 2 ** Math.min(job.attempts, 3)));
|
||||
const candidates = this.db.prepare(`SELECT n.* FROM memory_nodes n WHERE n.active=1 AND n.end_seq<=?
|
||||
AND NOT EXISTS(SELECT 1 FROM memory_job_nodes j WHERE j.job_id=? AND j.node_id=n.id) ORDER BY n.start_seq,n.id LIMIT ?`).all(job.cutoff, job.id, nodeLimit);
|
||||
const core = this.graph.core(), catalog = this.graph.catalog();
|
||||
const fixedBytes = Buffer.byteLength(JSON.stringify({ core, catalog, scope: job.scope })) + 2500;
|
||||
const available = maxBytes - fixedBytes;
|
||||
if (available < 8000) throw new Error("Core/catalog exceed Memory batch budget");
|
||||
const nodes = [];
|
||||
let bytes = 0;
|
||||
for (const row of candidates) {
|
||||
const size = Buffer.byteLength(JSON.stringify(row));
|
||||
if (bytes + size > available * 0.65) break;
|
||||
nodes.push(row); bytes += size;
|
||||
}
|
||||
const reviewObjects = [];
|
||||
if (job.mode === "deep") {
|
||||
for (const row of this.db.prepare("SELECT object_id FROM memory_job_objects WHERE job_id=? AND reviewed=0 ORDER BY object_id LIMIT ?").all(job.id, Math.max(1, Math.floor(4 / 2 ** Math.min(job.attempts, 2))))) {
|
||||
const item = this.graph.read({ id: row.object_id, limit: 10 });
|
||||
const size = Buffer.byteLength(JSON.stringify(item));
|
||||
// Leave room for source metadata; reviewed bodies must not crowd out
|
||||
// the evidence needed to safely extract facts from claimed nodes.
|
||||
if (bytes + size > available * (nodes.length ? 0.8 : 1)) break;
|
||||
reviewObjects.push(item); bytes += size;
|
||||
}
|
||||
}
|
||||
if (!nodes.length && !reviewObjects.length) {
|
||||
if (candidates.length) throw new Error("Memory batch budget too small for one node");
|
||||
if (this.db.prepare("SELECT 1 FROM memory_job_objects WHERE job_id=? AND reviewed=0 LIMIT 1").get(job.id)) throw new Error("Memory batch budget too small for one object");
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status='done',receipt=? WHERE id=?").run(JSON.stringify({ revision: this.revision(), noop: true }), job.id);
|
||||
return { job: null };
|
||||
}
|
||||
const refs = new Set();
|
||||
const visit = [...nodes];
|
||||
// Follow summary references with a bounded walk; full evidence remains separately expandable.
|
||||
for (let i = 0; i < visit.length && i < 500 && refs.size < 80; i++) {
|
||||
const node = visit[i];
|
||||
if (node.event_id) refs.add(node.event_id);
|
||||
else for (const child of JSON.parse(node.children)) {
|
||||
const found = this.db.prepare("SELECT * FROM memory_nodes WHERE id=?").get(child);
|
||||
if (found) visit.push(found);
|
||||
}
|
||||
}
|
||||
const evidence = [];
|
||||
for (const id of refs) {
|
||||
const row = this.timeline.read({ id, limit: 32000 });
|
||||
const entry = { id: row.id, kind: row.kind, source: row.source, at: row.at,
|
||||
text: nodes.some(n => n.event_id === id) ? "See claimed nodes with this event_id" : row.text.slice(0, 600) };
|
||||
const size = Buffer.byteLength(JSON.stringify(entry));
|
||||
if (bytes + size > available * 0.95) break;
|
||||
evidence.push(entry); bytes += size;
|
||||
}
|
||||
const related = [];
|
||||
for (const hit of this.graph.find({ query: job.scope || nodes.map(n => n.text).join(" ").slice(0, 1000), limit: 8 }).hits) {
|
||||
const item = this.graph.read({ id: hit.id, limit: 10 }), size = Buffer.byteLength(JSON.stringify(item));
|
||||
if (bytes + size > available) continue;
|
||||
related.push(item); bytes += size;
|
||||
}
|
||||
const token = randomUUID();
|
||||
const maxSummary = Math.min(6000, Math.max(300, Math.floor(nodes.reduce((n, row) => n + row.text.length, 0) * 0.45)));
|
||||
const seen = [...related, ...reviewObjects];
|
||||
const snapshot = { revision: this.revision(), nodes: nodes.map(n => n.id), reviewed_objects: reviewObjects.map(item => item.id), evidence_ids: [...new Set([...evidence.map(e => e.id), ...seen.flatMap(item => item.source_refs)])], objects_seen: seen.map(item => item.id), end_seq: Math.max(job.cursor, ...nodes.map(n => n.end_seq)), max_summary_chars: maxSummary };
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status='running',token=?,lease_until=?,attempts=attempts+1,snapshot=? WHERE id=?")
|
||||
.run(token, now + 180000, JSON.stringify(snapshot), job.id);
|
||||
return { job: { id: job.id, token, mode: job.mode, trigger: job.trigger, scope: job.scope,
|
||||
revision: snapshot.revision, nodes: nodes.map(({ id, at, text, event_id }) => ({ id, at, text, event_id })),
|
||||
evidence, related, review_objects: reviewObjects, core, catalog, max_summary_chars: maxSummary,
|
||||
review_remaining: this.db.prepare("SELECT count(*) AS n FROM memory_job_objects WHERE job_id=? AND reviewed=0").get(job.id).n } };
|
||||
});
|
||||
}
|
||||
commit(p, now = Date.now()) {
|
||||
return this.transaction(() => {
|
||||
const prior = this.db.prepare("SELECT receipt FROM memory_organize_batches WHERE token=? AND job_id=?").get(key(p.token), key(p.job_id));
|
||||
if (prior) return JSON.parse(prior.receipt);
|
||||
const job = this.db.prepare("SELECT * FROM memory_organize_jobs WHERE id=? AND token=? AND status='running'").get(p.job_id, p.token);
|
||||
if (!job || job.lease_until <= now) throw new Error("Expired or stale memory lease");
|
||||
const snapshot = JSON.parse(job.snapshot);
|
||||
if (snapshot.revision !== this.revision()) throw new Error("Memory changed during organization; retry against current version");
|
||||
const nodes = snapshot.nodes.map(id => this.db.prepare("SELECT * FROM memory_nodes WHERE id=? AND active=1").get(id));
|
||||
if (nodes.some(n => !n)) throw new Error("Timeline prefix changed");
|
||||
const summary = clean(p.summary, snapshot.max_summary_chars);
|
||||
if (!summary) throw new Error("Empty timeline summary");
|
||||
const oldTokens = nodes.reduce((n, row) => n + estimateTokens(row.text) + 96, 0);
|
||||
if (job.mode === "compact" && estimateTokens(summary) + 96 >= oldTokens) throw new Error("Compaction must reduce context size");
|
||||
if (!Array.isArray(p.covered_node_ids) || JSON.stringify([...p.covered_node_ids].sort()) !== JSON.stringify([...snapshot.nodes].sort())) throw new Error("Must account for every claimed node exactly once");
|
||||
const evidence = new Map(snapshot.evidence_ids.map(id => [id, this.timeline.read({ id, limit: 32000 })]));
|
||||
for (const object of p.objects ?? []) if (this.graph.object(object.id) && !snapshot.objects_seen.includes(object.id))
|
||||
throw new Error("Inspect existing object before replacing it");
|
||||
const revision = this.revision() + 1;
|
||||
this.graph.apply(p, revision, evidence, now);
|
||||
const id = nodes.length ? `summary:${randomUUID()}` : null;
|
||||
if (id) {
|
||||
const children = nodes.length === 1 && !nodes[0].event_id ? JSON.parse(nodes[0].children) : snapshot.nodes;
|
||||
this.db.prepare("INSERT INTO memory_nodes VALUES (?,?,?,?,?,NULL,?,1,?)")
|
||||
.run(id, Math.min(...nodes.map(n => n.start_seq)), Math.max(...nodes.map(n => n.end_seq)), Math.min(...nodes.map(n => n.at)), summary, JSON.stringify(children), revision);
|
||||
for (const node of nodes) this.db.prepare("UPDATE memory_nodes SET active=0 WHERE id=?").run(node.id);
|
||||
for (const nodeId of [...snapshot.nodes, id]) this.db.prepare("INSERT OR IGNORE INTO memory_job_nodes VALUES (?,?)").run(job.id, nodeId);
|
||||
}
|
||||
for (const objectId of snapshot.reviewed_objects ?? []) this.db.prepare("UPDATE memory_job_objects SET reviewed=1 WHERE job_id=? AND object_id=?").run(job.id, objectId);
|
||||
this.set("revision", revision);
|
||||
const belowTarget = this.stats().estimated_tokens < Number(this.meta("threshold_tokens", "96000")) * 0.6;
|
||||
const remaining = this.db.prepare(`SELECT 1 FROM memory_nodes n WHERE n.active=1 AND n.end_seq<=?
|
||||
AND NOT EXISTS(SELECT 1 FROM memory_job_nodes j WHERE j.job_id=? AND j.node_id=n.id) LIMIT 1`).get(job.cutoff, job.id);
|
||||
const objectsRemaining = this.db.prepare("SELECT 1 FROM memory_job_objects WHERE job_id=? AND reviewed=0 LIMIT 1").get(job.id);
|
||||
const done = job.mode === "compact" ? belowTarget || !remaining : !remaining && !objectsRemaining;
|
||||
if (done && job.trigger === "nightly") this.set("nightly_watermark", job.cutoff);
|
||||
const receipt = { job_id: job.id, revision, node_id: id, status: done ? "done" : "pending", trigger: job.trigger };
|
||||
this.db.prepare("INSERT INTO memory_organize_batches VALUES (?,?,?)").run(p.token, job.id, JSON.stringify(receipt));
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status=?,cursor=?,token=NULL,snapshot=NULL,lease_until=0,attempts=0,receipt=?,error=NULL WHERE id=?")
|
||||
.run(receipt.status, snapshot.end_seq, JSON.stringify(receipt), job.id);
|
||||
return receipt;
|
||||
});
|
||||
}
|
||||
inspect(p, now = Date.now()) {
|
||||
return this.transaction(() => {
|
||||
const job = this.db.prepare("SELECT * FROM memory_organize_jobs WHERE id=? AND token=? AND status='running' AND lease_until>?").get(key(p.job_id), key(p.token), now);
|
||||
if (!job) throw new Error("Stale inspection lease");
|
||||
const snapshot = JSON.parse(job.snapshot);
|
||||
if (snapshot.revision !== this.revision()) throw new Error("Memory changed during inspection");
|
||||
let result, refs;
|
||||
if (p.kind === "object") {
|
||||
result = this.graph.read({ id: p.id, offset: p.offset ?? 0, limit: 20 }); refs = result.source_refs;
|
||||
snapshot.objects_seen = [...new Set([...snapshot.objects_seen, result.id])];
|
||||
} else if (p.kind === "evidence") {
|
||||
result = this.node({ id: p.id, offset: p.offset ?? 0, limit: 8000 });
|
||||
refs = result.kind ? [result.id] : result.event_id ? [result.event_id] : [];
|
||||
} else throw new Error("Invalid inspection kind");
|
||||
snapshot.evidence_ids = [...new Set([...snapshot.evidence_ids, ...refs])];
|
||||
if (snapshot.evidence_ids.length > 200 || snapshot.objects_seen.length > 32) throw new Error("Inspection budget exceeded");
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET snapshot=? WHERE id=?").run(JSON.stringify(snapshot), job.id);
|
||||
return result;
|
||||
});
|
||||
}
|
||||
fail(p, now = Date.now()) {
|
||||
const job = this.db.prepare("SELECT * FROM memory_organize_jobs WHERE id=? AND token=? AND status='running'").get(key(p.job_id), key(p.token));
|
||||
if (!job) return { released: false };
|
||||
const status = job.attempts >= 5 ? "failed" : "pending";
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status=?,token=NULL,lease_until=0,snapshot=NULL,error=?,available_at=? WHERE id=?")
|
||||
.run(status, clean(String(p.error ?? "organization failed").slice(0, 500), 500), now + Math.min(300000, 5000 * 2 ** job.attempts), job.id);
|
||||
return { released: true, status };
|
||||
}
|
||||
status(p = {}) {
|
||||
const rows = p.id ? this.db.prepare("SELECT id,mode,trigger,status,cursor,cutoff,attempts,error,receipt FROM memory_organize_jobs WHERE id=?").all(key(p.id))
|
||||
: this.db.prepare("SELECT id,mode,trigger,status,cursor,cutoff,attempts,error FROM memory_organize_jobs ORDER BY created_at DESC LIMIT 20").all();
|
||||
return { ...this.stats(), jobs: rows.map(job => ({ ...job,
|
||||
object_review: this.db.prepare("SELECT count(*) AS total,coalesce(sum(reviewed),0) AS reviewed FROM memory_job_objects WHERE job_id=?").get(job.id)
|
||||
})), core_projection_error: this.meta("core_projection_error") || null };
|
||||
}
|
||||
node(p) {
|
||||
const row = this.db.prepare("SELECT * FROM memory_nodes WHERE id=?").get(sourceKey(p.id));
|
||||
if (!row) return this.timeline.read(p);
|
||||
return { ...row, children: JSON.parse(row.children) };
|
||||
}
|
||||
bootstrapCore(p) {
|
||||
return this.transaction(() => {
|
||||
const revision = this.revision() + 1;
|
||||
let changed = false;
|
||||
for (const name of CORE_NAMES) if (typeof p[name] === "string" && !this.graph.core().some(row => row.name === name)) {
|
||||
this.graph.saveCore(name, p[name], [], revision); changed = true;
|
||||
}
|
||||
if (changed) this.set("revision", revision);
|
||||
return { revision: this.revision(), core: this.graph.core() };
|
||||
});
|
||||
}
|
||||
editCore(p) {
|
||||
return this.transaction(() => {
|
||||
if (p.expected_revision !== this.revision()) throw new Error("Core revision conflict");
|
||||
const revision = this.revision() + 1;
|
||||
this.graph.saveCore(p.name, p.body, [], revision);
|
||||
this.set("revision", revision);
|
||||
return { revision, core: this.graph.core() };
|
||||
});
|
||||
}
|
||||
projectDecision(p) {
|
||||
return this.transaction(() => {
|
||||
const request = key(p.request_id), item = this.graph.object(p.id);
|
||||
const previous = this.db.prepare("SELECT * FROM memory_project_decisions WHERE request_id=?").get(request);
|
||||
if (previous) return previous;
|
||||
if (item?.type !== "project") throw new Error("Project not found");
|
||||
const event = this.timeline.read({ id: sourceKey(p.source_id), limit: 32000 });
|
||||
const quote = clean(p.user_quote, 1000);
|
||||
if (event.kind !== "user" || !quote || !event.text.includes(quote)) throw new Error("Project decision requires exact user evidence");
|
||||
if (!["active", "archived"].includes(p.status)) throw new Error("Invalid project decision");
|
||||
const revision = this.revision() + 1;
|
||||
this.db.prepare("UPDATE memory_objects SET status=?,revision=? WHERE id=?").run(p.status, revision, item.id);
|
||||
this.db.prepare("INSERT INTO memory_object_versions VALUES (?,?,?)").run(item.id, revision, JSON.stringify(this.graph.object(item.id)));
|
||||
this.db.prepare("INSERT INTO memory_project_decisions VALUES (?,?,?,?)").run(request, item.id, p.status, event.id);
|
||||
this.set("revision", revision);
|
||||
return { id: item.id, status: p.status, revision };
|
||||
});
|
||||
}
|
||||
}
|
||||
43
src/core-projection.js
Normal file
43
src/core-projection.js
Normal file
@ -0,0 +1,43 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { closeSync, existsSync, fsyncSync, lstatSync, mkdirSync, openSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { CORE_NAMES } from "./chain-graph.js";
|
||||
|
||||
const hash = text => createHash("sha256").update(text).digest("hex");
|
||||
|
||||
/** SQLite is authoritative; explicit human file edits are imported as versioned edits. */
|
||||
export class CoreProjection {
|
||||
constructor(chain, directory) { this.chain = chain; this.directory = directory; }
|
||||
refresh() {
|
||||
try {
|
||||
mkdirSync(this.directory, { recursive: true, mode: 0o700 });
|
||||
for (const name of CORE_NAMES) {
|
||||
const path = join(this.directory, name), meta = `core_file_hash:${name}`;
|
||||
let disk;
|
||||
if (existsSync(path)) {
|
||||
const stat = lstatSync(path);
|
||||
if (!stat.isFile() || stat.isSymbolicLink() || stat.size > 12000) throw new Error("Core file must be a small regular file");
|
||||
disk = readFileSync(path, "utf8");
|
||||
const current = this.chain.graph.core().find(row => row.name === name);
|
||||
if (hash(disk) !== this.chain.meta(meta) && disk !== current?.body)
|
||||
this.chain.editCore({ name, body: disk, expected_revision: this.chain.revision() });
|
||||
}
|
||||
const row = this.chain.graph.core().find(item => item.name === name);
|
||||
if (!row) continue;
|
||||
if (disk !== row.body) {
|
||||
const temp = join(this.directory, `.${name}.${randomUUID()}.tmp`);
|
||||
try {
|
||||
writeFileSync(temp, row.body, { mode: 0o600, flag: "wx", flush: true });
|
||||
renameSync(temp, path);
|
||||
const fd = openSync(this.directory, "r");
|
||||
try { fsyncSync(fd); } finally { closeSync(fd); }
|
||||
} finally { if (existsSync(temp)) unlinkSync(temp); }
|
||||
}
|
||||
this.chain.set(meta, hash(row.body));
|
||||
}
|
||||
this.chain.set("core_projection_error", "");
|
||||
} catch (error) {
|
||||
this.chain.set("core_projection_error", error instanceof Error ? error.message.slice(0, 200) : "Projection failure");
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,4 +1,5 @@
|
||||
const SECRET_PATTERNS = [
|
||||
/\bsk-[\w.-]{12,}/i,
|
||||
/-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----/i,
|
||||
/\b(?:password|passwd|api[_ -]?key|access[_ -]?token|refresh[_ -]?token|bearer|client[_ -]?secret)\b/i,
|
||||
/(?:密码|口令|令牌|密钥|私钥|验证码)/,
|
||||
@ -15,6 +16,13 @@ export function safeMemoryText(value, maxLength = 8_000) {
|
||||
return text;
|
||||
}
|
||||
|
||||
// The new timeline may discuss credential names/references, never their values.
|
||||
export function safeTimelineText(value, maxLength = 32000) {
|
||||
const text = boundedText(value, maxLength);
|
||||
if (/\bsk-[\w.-]{12,}|-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----|\bBearer\s+[\w./+=-]{12,}|(?:password|passwd|api[_ -]?key|access[_ -]?token|refresh[_ -]?token|client[_ -]?secret|密码|口令|密钥)["']?\s*(?:[=:：]|是|为)\s*["']?[^\s"',}]{4,}/i.test(text)) return "[敏感内容未写入记忆]";
|
||||
return text;
|
||||
}
|
||||
|
||||
export function words(value, maxLength = 16_000) {
|
||||
const lowered = boundedText(value, maxLength).toLowerCase();
|
||||
const terms = new Set(lowered.match(/[a-z0-9_-]{2,}|[\u3400-\u9fff]{2}/g) ?? []);
|
||||
|
||||
@ -6,6 +6,8 @@ import { dirname, join, resolve } from "node:path";
|
||||
import { MemoryStore } from "./store.js";
|
||||
import { ContinuityStore } from "./continuity.js";
|
||||
import { Timeline } from "./timeline.js";
|
||||
import { ChainStore } from "./chain-store.js";
|
||||
import { CoreProjection } from "./core-projection.js";
|
||||
|
||||
const stateDir = resolve(process.env.DESKPAW_STATE_DIR ?? join(homedir(), ".deskpaw"));
|
||||
const runtimeDir = join(stateDir, "runtime");
|
||||
@ -34,15 +36,33 @@ if (process.platform !== "win32") {
|
||||
const store = new MemoryStore(databasePath, bundlesDir);
|
||||
const continuity = new ContinuityStore(store.db);
|
||||
const timeline = new Timeline(store.db);
|
||||
const chain = new ChainStore(store.db, timeline);
|
||||
const coreProjection = new CoreProjection(chain, join(stateDir, "super"));
|
||||
coreProjection.refresh();
|
||||
|
||||
function call(method, params) {
|
||||
if (method === "memory.chain.context") { coreProjection.refresh(); return chain.context(params); }
|
||||
if (method === "memory.chain.tick") return chain.tick(params);
|
||||
if (method === "memory.chain.status") return chain.status(params);
|
||||
if (method === "memory.chain.core.bootstrap") return chain.bootstrapCore(params);
|
||||
if (method === "memory.chain.core.update") { const result = chain.editCore(params); coreProjection.refresh(); return result; }
|
||||
if (method === "memory.chain.find") return chain.graph.find(params);
|
||||
if (method === "memory.chain.browse") return chain.graph.browse(params);
|
||||
if (method === "memory.chain.read") return chain.graph.read(params);
|
||||
if (method === "memory.chain.node") return chain.node(params);
|
||||
if (method === "memory.chain.project_decision") return chain.projectDecision(params);
|
||||
if (method === "memory.organize.request") return chain.request({ ...params, trigger: "explicit" });
|
||||
if (method === "memory.organize.claim") return chain.claim(params);
|
||||
if (method === "memory.organize.inspect") return chain.inspect(params);
|
||||
if (method === "memory.organize.commit") { const receipt = chain.commit(params); coreProjection.refresh(); return receipt; }
|
||||
if (method === "memory.organize.fail") return chain.fail(params);
|
||||
if (method === "memory.events.ingest") return continuity.ingest(params);
|
||||
if (method === "memory.observation.claim") return continuity.claim(params);
|
||||
if (method === "memory.observation.commit") return continuity.commit(params);
|
||||
if (method === "memory.observation.fail") return continuity.fail(params);
|
||||
if (method === "memory.continuity.recall") return continuity.recall(params);
|
||||
if (method === "memory.evidence.get") return continuity.evidence(params);
|
||||
if (method === "timeline.append") return timeline.append(params);
|
||||
if (method === "timeline.append") return timeline.append({ ...params, at: params.at ?? Date.now() });
|
||||
if (method === "timeline.query") return timeline.query(params);
|
||||
if (method === "timeline.read" || method === "timeline.expand") return timeline.read(params);
|
||||
if (method === "health") return { status: "ok", version: "0.1.0", ...store.status() };
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
import { safeMemoryText, words } from "./policy.js";
|
||||
import { safeTimelineText, words } from "./policy.js";
|
||||
|
||||
/** Append-only experience ledger. Old indexes/embeddings are not consulted. */
|
||||
export class Timeline {
|
||||
@ -9,6 +9,7 @@ export class Timeline {
|
||||
at INTEGER NOT NULL,text TEXT NOT NULL,source TEXT NOT NULL,
|
||||
parent_id TEXT,covered_from INTEGER,covered_to INTEGER,version INTEGER NOT NULL DEFAULT 1);
|
||||
CREATE INDEX IF NOT EXISTS timeline_time ON timeline(at,seq);
|
||||
CREATE INDEX IF NOT EXISTS timeline_task ON timeline(json_extract(source,'$.task_id'),seq);
|
||||
CREATE TABLE IF NOT EXISTS timeline_terms (term TEXT NOT NULL,seq INTEGER NOT NULL,PRIMARY KEY(term,seq));
|
||||
CREATE INDEX IF NOT EXISTS timeline_terms_seq ON timeline_terms(seq);`);
|
||||
}
|
||||
@ -19,9 +20,9 @@ export class Timeline {
|
||||
if (!Number.isSafeInteger(p.at) || p.at < 0) throw new Error("Invalid timestamp");
|
||||
const source = {};
|
||||
for (const key of ["channel", "conversation_id", "sender", "actor", "message_id", "job_id", "task_id", "artifact_ref"]) {
|
||||
if (typeof p.source?.[key] === "string") source[key] = safeMemoryText(p.source[key], 240);
|
||||
if (typeof p.source?.[key] === "string") source[key] = safeTimelineText(p.source[key], 240);
|
||||
}
|
||||
const text = safeMemoryText(p.text, 32000);
|
||||
const text = safeTimelineText(p.text, 32000);
|
||||
this.db.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
const result = this.db.prepare("INSERT OR IGNORE INTO timeline(id,kind,at,text,source) VALUES (?,?,?,?,?)").run(p.id, p.kind, p.at, text, JSON.stringify(source));
|
||||
@ -32,19 +33,20 @@ export class Timeline {
|
||||
}
|
||||
this.db.exec("COMMIT");
|
||||
} catch (error) { this.db.exec("ROLLBACK"); throw error; }
|
||||
return { id: p.id, stored: true };
|
||||
return { id: p.id, stored: true, seq: Number(this.db.prepare("SELECT seq FROM timeline WHERE id=?").get(p.id).seq) };
|
||||
}
|
||||
query(p = {}) {
|
||||
const limit = p.limit ?? 10;
|
||||
const after = p.after ?? 0;
|
||||
const from = p.from ?? 0;
|
||||
const to = p.to ?? Number.MAX_SAFE_INTEGER;
|
||||
if (p.task_id !== undefined && (typeof p.task_id !== 'string' || !p.task_id || p.task_id.length > 240)) throw new Error('Invalid task_id');
|
||||
if (![limit, after, from, to].every(Number.isSafeInteger) || limit < 1 || limit > 50 || after < 0 || from < 0 || to < from) throw new Error("Invalid query bounds");
|
||||
const terms = [...words(String(p.query ?? ""))].slice(0, 32);
|
||||
const rows = terms.length ? this.db.prepare(`SELECT t.*,count(*) AS score FROM timeline t
|
||||
JOIN timeline_terms x ON x.seq=t.seq WHERE x.term IN (${terms.map(() => "?").join(",")})
|
||||
AND t.seq>? AND t.at BETWEEN ? AND ? GROUP BY t.seq ORDER BY score DESC,t.at DESC,t.seq DESC LIMIT ?`).all(...terms, after, from, to, limit)
|
||||
: this.db.prepare("SELECT * FROM timeline WHERE seq>? AND at BETWEEN ? AND ? ORDER BY seq LIMIT ?").all(after, from, to, limit);
|
||||
AND t.seq>? AND t.at BETWEEN ? AND ? ${p.task_id ? "AND json_extract(t.source,'$.task_id')=?" : ""} GROUP BY t.seq ORDER BY score DESC,t.at DESC,t.seq DESC LIMIT ?`).all(...terms, after, from, to, ...(p.task_id ? [p.task_id] : []), limit)
|
||||
: this.db.prepare(`SELECT * FROM timeline WHERE seq>? AND at BETWEEN ? AND ? ${p.task_id ? "AND json_extract(source,'$.task_id')=?" : ""} ORDER BY seq LIMIT ?`).all(after, from, to, ...(p.task_id ? [p.task_id] : []), limit);
|
||||
return { entries: rows.map((row) => ({ ...row, text: row.text.slice(0, 1200), source: JSON.parse(row.source), expandable: row.text.length > 1200 })), cursor: terms.length ? null : rows.at(-1)?.seq ?? after, method: "lexical_time", embedding: false };
|
||||
}
|
||||
read(p) {
|
||||
|
||||
11
src/token-budget.js
Normal file
11
src/token-budget.js
Normal file
@ -0,0 +1,11 @@
|
||||
// portable-v1; keep identical to Agent token-budget.ts. This is an estimate, not a tokenizer.
|
||||
export function estimateTokens(value) {
|
||||
const text = typeof value === "string" ? value : (JSON.stringify(value) ?? "");
|
||||
let tokens = 0;
|
||||
for (const run of text.matchAll(/[a-zA-Z0-9_]+|\s+|[^a-zA-Z0-9_\s]/gu)) {
|
||||
const part = run[0], point = part.codePointAt(0);
|
||||
if (/^[a-zA-Z0-9_\s]/u.test(part)) tokens += Math.ceil(part.length / 3);
|
||||
else tokens += point < 128 ? 1 : point >= 0x2e80 && point <= 0x9fff ? 2 : 4;
|
||||
}
|
||||
return tokens;
|
||||
}
|
||||
299
test/chain.test.js
Normal file
299
test/chain.test.js
Normal file
@ -0,0 +1,299 @@
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
import { ChainStore } from "../src/chain-store.js";
|
||||
import { Timeline } from "../src/timeline.js";
|
||||
import { CoreProjection } from "../src/core-projection.js";
|
||||
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { estimateTokens } from "../src/token-budget.js";
|
||||
|
||||
function fixture() {
|
||||
const db = new DatabaseSync(":memory:"), timeline = new Timeline(db), chain = new ChainStore(db, timeline);
|
||||
const append = (id, text = "用户讨论报销工作与发票整理。".repeat(100), kind = "user", at = 1) => timeline.append({ id, text, kind, at, source: { channel: "test", sender: "owner" } });
|
||||
const claim = (mode = "compact") => {
|
||||
chain.request({ request_id: `test:${Math.random()}`, mode }, 10);
|
||||
return chain.claim({}, 20).job;
|
||||
};
|
||||
const proposal = (job, extra = {}) => ({ job_id: job.id, token: job.token, covered_node_ids: job.nodes.map(n => n.id),
|
||||
summary: "讨论报销，尚未提交。详细依据可以展开原始事件。", objects: [], edges: [], core: [], catalog: [], ...extra });
|
||||
return { db, timeline, chain, append, claim, proposal };
|
||||
}
|
||||
const person = (id = "person:qhy", refs = ["meeting"]) => ({ id, type: "person", name: "钱灏宇", summary: "用户的同事，与报销工作有关。",
|
||||
body: "已确认是同事。具体职责尚未确认。", aliases: ["灏宇"], cues: ["发票整理找谁沟通", "同事交接报销材料"], source_refs: refs, status: "active" });
|
||||
|
||||
test("atomic compaction preserves raw evidence and concurrent suffix; receipt retries are idempotent", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); const job = f.claim();
|
||||
f.append("new", "正在整理时用户的新消息", "user", 30); f.chain.sync();
|
||||
const receipt = f.chain.commit(f.proposal(job, { objects: [person()], catalog: [{ type: "person", summary: "有同事和报销协作人的相关记忆" }] }), 40);
|
||||
assert.equal(f.chain.commit(f.proposal(job), 50).revision, receipt.revision);
|
||||
const context = f.chain.context(); assert.equal(context.ready, true);
|
||||
assert.ok(context.nodes.some(n => n.text.includes("新消息")));
|
||||
assert.ok(context.nodes.some(n => n.id === receipt.node_id));
|
||||
assert.ok(f.timeline.read({ id: "meeting" }).text.length > 1000);
|
||||
assert.ok(f.chain.node({ id: receipt.node_id }).children.length);
|
||||
assert.equal(f.chain.graph.find({ query: "钱灏宇" }).hits[0].id, "person:qhy");
|
||||
assert.equal(f.chain.graph.find({ query: "灏宇" }).hits[0].id, "person:qhy");
|
||||
assert.equal(f.chain.graph.find({ query: "发票整理找谁沟通" }).hits[0].id, "person:qhy");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("invalid evidence rolls back graph, timeline and revision together", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); const job = f.claim(), before = f.chain.context();
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { objects: [person(), person("person:bad", ["not-in-batch"])] }), 40), /Evidence/);
|
||||
assert.equal(f.chain.graph.object("person:qhy"), null);
|
||||
assert.deepEqual(f.chain.context().nodes, before.nodes);
|
||||
assert.equal(f.chain.revision(), 0);
|
||||
assert.throws(() => f.chain.commit({ ...f.proposal(job), covered_node_ids: [] }, 40), /every claimed node/);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("core changes invalidate stale leases and expired leases can be reclaimed", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); const job = f.claim();
|
||||
f.chain.bootstrapCore({ "SOUL.md": "做可靠的工作同事。" });
|
||||
assert.throws(() => f.chain.commit(f.proposal(job), 40), /Memory changed/);
|
||||
assert.equal(f.chain.fail({ job_id: job.id, token: job.token, error: "revision conflict" }, 50).released, true);
|
||||
const next = f.chain.claim({}, 30000).job; assert.equal(next.id, job.id); assert.notEqual(next.token, job.token);
|
||||
assert.throws(() => f.chain.commit(f.proposal(next), 300001), /Expired/);
|
||||
assert.equal(f.chain.claim({}, 300002).job.id, job.id);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("02:50 local nightly job waits five quiet minutes, is unique, and yields between batches", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting");
|
||||
const now = Date.parse("2026-10-11T18:50:00Z");
|
||||
f.chain.tick({ last_dialogue_at: now - 240000, busy: false }, now);
|
||||
assert.equal(f.chain.claim({}, now).job, null);
|
||||
f.chain.tick({ busy: true }, now + 120000);
|
||||
assert.equal(f.chain.claim({}, now + 120000).job, null);
|
||||
f.chain.tick({ busy: false }, now + 120001);
|
||||
const job = f.chain.claim({}, now + 120001).job;
|
||||
assert.equal(job.trigger, "nightly"); assert.equal(job.mode, "deep");
|
||||
f.chain.commit(f.proposal(job), now + 120010);
|
||||
f.chain.tick({ busy: false }, now + 240000);
|
||||
assert.equal(f.chain.status().jobs.filter(j => j.trigger === "nightly").length, 1);
|
||||
assert.equal(f.chain.claim({}, now + 240000).job, null);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("threshold is not an every-message model trigger; explicit request is an event", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting", "普通聊天");
|
||||
const now = Date.parse("2026-10-11T17:00:00Z");
|
||||
f.chain.tick({ threshold_tokens: 8000 }, now);
|
||||
assert.equal(f.chain.status().jobs.length, 0);
|
||||
f.append("big", "财务材料".repeat(2000));
|
||||
f.chain.tick({ threshold_tokens: 8000 }, now + 1);
|
||||
assert.equal(f.chain.status().jobs[0].trigger, "threshold");
|
||||
f.chain.request({ request_id: "explicit:one", scope: "整理钱灏宇有关的记忆", mode: "deep" }, now + 2);
|
||||
f.chain.request({ request_id: "explicit:one", scope: "整理钱灏宇有关的记忆", mode: "deep" }, now + 3);
|
||||
assert.equal(f.db.prepare("SELECT count(*) AS n FROM timeline WHERE kind='organize_request'").get().n, 1);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("projects require user confirmation, graph links are cross-project and paginated", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); const job = f.claim();
|
||||
const project = { ...person("project:expense"), type: "project", name: "报销", summary: "每月报销与日常发票收集" };
|
||||
f.chain.commit(f.proposal(job, { objects: [person(), project], edges: [{ from_id: "person:qhy", relation: "参与", to_id: project.id, source_refs: ["meeting"] }] }), 40);
|
||||
assert.equal(f.chain.graph.object(project.id).status, "candidate");
|
||||
assert.equal(f.chain.graph.find({ project_id: project.id, type: "person" }).hits[0].id, "person:qhy");
|
||||
assert.equal(f.chain.graph.read({ id: "person:qhy" }).edges.length, 1);
|
||||
f.append("yes", "同意创建报销项目", "user", 60);
|
||||
assert.throws(() => f.chain.projectDecision({ id: project.id, request_id: "decision", status: "active", source_id: "yes", user_quote: "不是用户原话" }), /exact user evidence/);
|
||||
f.chain.projectDecision({ id: project.id, request_id: "decision", status: "active", source_id: "yes", user_quote: "同意创建报销项目" });
|
||||
assert.equal(f.chain.graph.object(project.id).status, "active");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("same name does not merge identities; read/cues never become new factual evidence", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); f.append("read", "钱灏宇的历史记忆", "memory_recall"); const job = f.claim();
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { objects: [person("person:false", ["read"])] }), 40), /independent evidence/);
|
||||
f.chain.commit(f.proposal(job, { objects: [person(), person("person:another")] }), 40);
|
||||
assert.equal(f.chain.graph.find({ query: "钱灏宇" }).hits.length, 2);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("catalog remains bounded, types can be fully paginated, no implicit graph fan-out", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
for (let i = 0; i < 300; i++) f.chain.graph.apply({ objects: [person(`person:${i}`)] }, i + 1, new Map([["meeting", { kind: "user", text: "" }]]), i);
|
||||
assert.ok(JSON.stringify(f.chain.graph.catalog()).length < 3000);
|
||||
assert.equal(f.chain.graph.catalog().entries[0].count, 300);
|
||||
assert.equal(f.chain.graph.find({ type: "person", offset: 290, limit: 20 }).hits.length, 10);
|
||||
assert.throws(() => f.chain.graph.find({ limit: 300 }));
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("credential metadata permits vault refs but rejects values; core identity needs user evidence", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); const job = f.claim();
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { core: [{ name: "USER.md", body: "用户喜欢自动付款", source_refs: ["meeting"] }] }), 40), /exact user evidence/);
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { objects: [{ ...person(), body: "sk-example-secret-01234567890" }] }), 40), /Credential/);
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { objects: [{ ...person("credential:bad"), type: "credential", secret_ref: "sk-example-secret-01234567890" }] }), 40), /Credential/);
|
||||
f.chain.commit(f.proposal(job, { objects: [{ ...person("credential:model"), type: "credential", name: "模型服务", body: "用于模型调用；真实值仅存密钥库", secret_ref: "provider/example/api_key" }] }), 40);
|
||||
assert.equal(f.chain.graph.object("credential:model").secret_ref, "provider/example/api_key");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("large events span bounded batches without skipping sibling chunks", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("big", "原始会议内容".repeat(5000));
|
||||
const request = f.chain.request({ request_id: "large", mode: "deep" }, 10);
|
||||
let batches = 0;
|
||||
for (let now = 20; now < 100; now++) {
|
||||
const job = f.chain.claim({ max_bytes: 24000 }, now).job;
|
||||
if (!job) break;
|
||||
assert.ok(Buffer.byteLength(JSON.stringify(job)) <= 24000);
|
||||
f.chain.commit(f.proposal(job), now); batches++;
|
||||
}
|
||||
assert.ok(batches > 1);
|
||||
assert.equal(f.chain.status({ id: request.id }).jobs[0].status, "done");
|
||||
assert.equal(f.db.prepare("SELECT count(*) AS n FROM memory_nodes WHERE active=1 AND event_id='big'").get().n, 0);
|
||||
assert.equal(f.timeline.read({ id: "big", limit: 32000 }).text.length, 30000);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("core files import human edits, project committed versions, and reject oversized/secret edits without data loss", () => {
|
||||
const f = fixture(), dir = mkdtempSync(join(tmpdir(), "memory-core-"));
|
||||
try {
|
||||
writeFileSync(join(dir, "USER.md"), "用户喜欢简洁沟通。");
|
||||
const projection = new CoreProjection(f.chain, dir); projection.refresh();
|
||||
assert.equal(f.chain.graph.core()[0].body, "用户喜欢简洁沟通。");
|
||||
const revision = f.chain.revision();
|
||||
f.chain.editCore({ name: "USER.md", body: "用户喜欢简洁且有依据的回复。", expected_revision: revision });
|
||||
projection.refresh();
|
||||
assert.equal(readFileSync(join(dir, "USER.md"), "utf8"), "用户喜欢简洁且有依据的回复。");
|
||||
writeFileSync(join(dir, "USER.md"), "用户自己修改的新偏好。"); projection.refresh();
|
||||
assert.equal(f.chain.graph.core()[0].body, "用户自己修改的新偏好。");
|
||||
writeFileSync(join(dir, "USER.md"), "password=never-store-this-value"); projection.refresh();
|
||||
assert.equal(f.chain.graph.core()[0].body, "用户自己修改的新偏好。");
|
||||
assert.ok(f.chain.meta("core_projection_error"));
|
||||
assert.equal(readFileSync(join(dir, "USER.md"), "utf8"), "password=never-store-this-value");
|
||||
assert.throws(() => f.chain.editCore({ name: "USER.md", body: "stale", expected_revision: revision }), /revision conflict/);
|
||||
} finally { f.db.close(); rmSync(dir, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
test("organization survives SQLite reopen and a stale worker cannot overwrite a reclaimed batch", () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), "memory-restart-")), path = join(dir, "memory.sqlite");
|
||||
let db = new DatabaseSync(path);
|
||||
try {
|
||||
let timeline = new Timeline(db), chain = new ChainStore(db, timeline);
|
||||
timeline.append({ id: "event", kind: "user", at: 1, text: "旧事件需要保留".repeat(100), source: {} });
|
||||
chain.request({ request_id: "restart" }, 10); const first = chain.claim({}, 20).job;
|
||||
db.close(); db = new DatabaseSync(path); timeline = new Timeline(db); chain = new ChainStore(db, timeline);
|
||||
assert.equal(chain.claim({}, 30).job, null);
|
||||
const next = chain.claim({}, 180021).job;
|
||||
assert.equal(next.id, first.id); assert.notEqual(next.token, first.token);
|
||||
const proposal = { job_id: first.id, token: first.token, summary: "保留经历", covered_node_ids: first.nodes.map(n => n.id) };
|
||||
assert.throws(() => chain.commit(proposal, 180022), /stale/);
|
||||
chain.commit({ ...proposal, token: next.token, covered_node_ids: next.nodes.map(n => n.id) }, 180022);
|
||||
assert.equal(db.prepare("PRAGMA integrity_check").get().integrity_check, "ok");
|
||||
assert.equal(timeline.read({ id: "event" }).text, "旧事件需要保留".repeat(100));
|
||||
} finally { db.close(); rmSync(dir, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
test("deep inspection grants evidence without letting uninspected objects be replaced", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); let job = f.claim();
|
||||
f.chain.commit(f.proposal(job, { objects: [person()] }), 40);
|
||||
f.append("new", "用户进一步说明了同事职责".repeat(100));
|
||||
f.chain.request({ request_id: "inspect", mode: "compact", scope: "完全不相关词 abcxyz" }, 50);
|
||||
job = f.chain.claim({}, 60).job;
|
||||
const change = { ...person(), body: "更正后的同事信息", source_refs: ["new"] };
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { objects: [change] }), 70), /Inspect existing/);
|
||||
const inspected = f.chain.inspect({ job_id: job.id, token: job.token, kind: "object", id: "person:qhy" }, 70);
|
||||
assert.equal(inspected.name, "钱灏宇");
|
||||
f.chain.commit(f.proposal(job, { objects: [change] }), 80);
|
||||
assert.equal(f.chain.graph.object("person:qhy").body, "更正后的同事信息");
|
||||
assert.equal(f.chain.graph.read({ id: "person:qhy" }).total_sources, 2);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("timeline clipping returns newest nodes, explicit omission and separate byte/token accounting without deleting evidence", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
for (let i=0;i<10;i++) f.append(`raw:${i}`, `事件${i}：${"中文正文".repeat(200)}`, "user", i);
|
||||
const full=f.chain.context({max_tokens:100000,max_bytes:350000});
|
||||
const view=f.chain.context({max_tokens:4000,max_bytes:350000});
|
||||
assert.equal(view.ready,true); assert.equal(view.truncated,true);
|
||||
assert.equal(view.nodes.at(-1).event_id,"raw:9");
|
||||
assert.ok(view.view_estimated_tokens<=4000);
|
||||
assert.equal(view.view_estimated_tokens,2+view.nodes.reduce((n,node)=>n+estimateTokens(node)+1,0));
|
||||
assert.equal(view.omitted_nodes,full.nodes.length-view.nodes.length);
|
||||
assert.deepEqual(f.chain.context({max_tokens:100000,max_bytes:350000}).nodes,full.nodes);
|
||||
assert.ok(f.timeline.read({id:"raw:0"}).text.includes("中文正文"));
|
||||
} finally {f.db.close();}
|
||||
});
|
||||
|
||||
test("deep review covers dormant and archived objects across durable batches and never invents timeline events", () => {
|
||||
const f=fixture();
|
||||
try {
|
||||
f.append("meeting");
|
||||
for(let i=0;i<13;i++) f.chain.graph.apply({objects:[{...person(`person:old-${i}`),status:i===0?"archived":"active"}]},i+1,new Map([["meeting",{kind:"user",text:"证据"}]]),i);
|
||||
let after, ids=[];
|
||||
do { const page=f.chain.graph.browse({after,limit:3}); ids.push(...page.entries.map(x=>x.id)); after=page.next_after; } while(after);
|
||||
assert.equal(ids.length,14); assert.ok(ids.includes("person:old-0"));
|
||||
const req=f.chain.request({request_id:"global-review",mode:"deep",scope:"与旧人物无关的新工作"},10);
|
||||
const reviewed=new Set(); let emptyBatches=0;
|
||||
for(let now=20;now<40;now++) {
|
||||
const job=f.chain.claim({},now).job; if(!job)break;
|
||||
for(const obj of job.review_objects) reviewed.add(obj.id);
|
||||
if(!job.nodes.length)emptyBatches++;
|
||||
const before=f.chain.context().nodes;
|
||||
f.chain.commit(f.proposal(job),now);
|
||||
if(!job.nodes.length)assert.deepEqual(f.chain.context().nodes,before);
|
||||
}
|
||||
assert.equal(reviewed.size,14); assert.ok(emptyBatches>0);
|
||||
const status=f.chain.status({id:req.id}).jobs[0];
|
||||
assert.equal(status.status,"done"); assert.deepEqual({...status.object_review},{total:14,reviewed:14});
|
||||
assert.equal(f.chain.graph.object("person:old-0").status,"archived");
|
||||
}finally{f.db.close();}
|
||||
});
|
||||
|
||||
test("failed deep batch does not advance review coverage",()=>{
|
||||
const f=fixture();
|
||||
try {
|
||||
f.append("meeting"); const first=f.claim("deep");
|
||||
f.chain.fail({job_id:first.id,token:first.token,error:"test failure"},30);
|
||||
assert.equal(f.chain.status({id:first.id}).jobs[0].object_review.reviewed,0);
|
||||
const retry=f.chain.claim({},30000).job;
|
||||
assert.deepEqual(retry.review_objects.map(x=>x.id),first.review_objects.map(x=>x.id));
|
||||
}finally{f.db.close();}
|
||||
});
|
||||
|
||||
test("opaque Chinese evidence IDs remain expandable and credential metadata is not a secret", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("会议记录/十月十二日", "讨论密钥管理：用途是模型调用，引用 provider/example/api_key，未提供真实值。".repeat(10));
|
||||
const job = f.claim();
|
||||
assert.ok(f.chain.node({ id: "会议记录/十月十二日" }).text.includes("provider/example/api_key"));
|
||||
f.chain.commit(f.proposal(job, { objects: [person("person:qhy", ["会议记录/十月十二日"])] }), 40);
|
||||
assert.deepEqual(f.chain.graph.read({ id: "person:qhy" }).source_refs, ["会议记录/十月十二日"]);
|
||||
f.append("quoted-secret", '{"api_key":"sk-example-secret-01234567890"}');
|
||||
assert.equal(f.chain.node({ id: "quoted-secret" }).text, "[敏感内容未写入记忆]");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("failed model batches shrink on retry without consuming or dropping their original nodes", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
for(let i=0;i<12;i++) f.append(`event:${i}`, "有效工作事实".repeat(20));
|
||||
const first=f.claim(), before=f.chain.context().nodes;
|
||||
assert.equal(first.nodes.length,8);
|
||||
f.chain.fail({job_id:first.id,token:first.token,error:"Memory output truncated"},30);
|
||||
const next=f.chain.claim({},30000).job;
|
||||
assert.equal(next.id,first.id); assert.equal(next.nodes.length,4);
|
||||
assert.deepEqual(f.chain.context().nodes,before);
|
||||
f.chain.commit(f.proposal(next),30001);
|
||||
assert.ok(f.chain.context().nodes.some(node=>node.event_id==="event:11"));
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
30
test/computer-events.test.js
Normal file
30
test/computer-events.test.js
Normal file
@ -0,0 +1,30 @@
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
import { Timeline } from "../src/timeline.js";
|
||||
import { ChainStore } from "../src/chain-store.js";
|
||||
|
||||
test("incremental Computer events interleave with user dialogue, remain task-addressable and reach Memory", () => {
|
||||
const db = new DatabaseSync(":memory:"), timeline = new Timeline(db), chain = new ChainStore(db, timeline);
|
||||
try {
|
||||
const event = { id: "computer:task:1", kind: "work_record", at: 10, text: "阅读文章第一屏，作者提出以下方法。".repeat(100), source: { actor: "computer", task_id: "task", message_id: "observation-1" } };
|
||||
timeline.append(event);
|
||||
timeline.append({ id: "user:1", kind: "user", at: 11, text: "先别发布，继续阅读", source: { actor: "owner" } });
|
||||
timeline.append({ ...event, id: "computer:task:2", at: 12, text: "阅读文章第二屏" });
|
||||
timeline.append({ ...event, id: "other-task", at: 13, source: { task_id: "other" } });
|
||||
timeline.append(event);
|
||||
assert.equal(timeline.query({ task_id: "task" }).entries.length, 2);
|
||||
assert.equal(timeline.query({ task_id: "task", query: "阅读文章" }).entries.length, 2);
|
||||
const page = timeline.query({ task_id: "task", limit: 1 });
|
||||
assert.equal(timeline.query({ task_id: "task", after: page.cursor }).entries[0].id, "computer:task:2");
|
||||
assert.equal(chain.context().nodes.find(node => node.event_id === event.id).source.actor, "computer");
|
||||
assert.ok(chain.context().nodes.some(node => node.event_id === "user:1"));
|
||||
assert.equal(chain.status().jobs.length, 0);
|
||||
chain.request({ request_id: "organize", mode: "deep" }, 20);
|
||||
const job = chain.claim({}, 30).job;
|
||||
assert.ok(job.evidence.some(item => item.source.task_id === "task"));
|
||||
chain.commit({ job_id: job.id, token: job.token, covered_node_ids: job.nodes.map(node => node.id), summary: "阅读文章，用户要求暂不发布。正文仍可展开。", objects: [] }, 40);
|
||||
assert.equal(timeline.read({ id: event.id }).text, event.text);
|
||||
assert.throws(() => timeline.query({ task_id: 123 }), /task_id/);
|
||||
} finally { db.close(); }
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user