feat: separate timeline compaction from durable object organization
This commit is contained in:
18962558917
parent
3fc24c1fc1
commit
c373d21e1f
147
OBJECT_MEMORY.md
Normal file
147
OBJECT_MEMORY.md
Normal file
@ -0,0 +1,147 @@
|
||||
# 对象记忆与两阶段整理（2026-10-13）
|
||||
|
||||
本文件描述两阶段对象记忆实现。已于 2026-10-14 00:34 与 Agent 配套部署；实际记录见第 8 节。部署不等于已有记忆全部重新整理完成。
|
||||
|
||||
## 1. 定义
|
||||
|
||||
- 事件：带发生时间的不可变原文，保存在 `timeline`。时间不明确用接入时间。
|
||||
- 热时间轴：Super 的基础上下文。近期是事件原文片段，整理后是概括索引头，children 可逐级展开到原文。
|
||||
- 经历附件：退出热时间轴的事件原文；仍在原表，不删除、不另复制一份。不要求每个事件都提炼出对象。
|
||||
- 长期对象：稳定 ID 对应一份高度概括、当前有效的档案。人物、任务、待办、日程、知识、项目采用相同结构。credential 仅额外保存用途/引用，不保存密钥值。
|
||||
- 项目：也是对象，同时是可交叉的关联集合，不是独占文件夹。日常是内置 `project:daily`。
|
||||
- 附件关系：对象→事件的证据索引，及事件→对象的归档索引。原话、工具流水留在附件，不不断追加进正文。
|
||||
|
||||
建议正文：人物的身份/职责/特点/协作提示；任务的目标/进度/约束/下一步；待办的事项/条件/状态；日程的时间规则/事项/状态；知识的结论/适用条件/经验；项目的范围/目标/现状。缺失信息标记未知，不猜测补全。`object.status` 是记忆生命周期，业务“待办/完成”等写在正文。
|
||||
|
||||
## 2. 两种职责，三个执行模式
|
||||
|
||||
```text
|
||||
timeline 原文 → 热时间轴
|
||||
│ compact
|
||||
├─ 概括头替换原文节点
|
||||
└─ 保留原文 + 持久归档队列
|
||||
│ organize / deep
|
||||
├─ 对象档案及联系 → linked
|
||||
├─ 仅留经历附件 → archive_only
|
||||
└─ 尚未完成 → continue 检查点
|
||||
```
|
||||
|
||||
### compact：时间轴整理
|
||||
|
||||
模型只拿本批节点与一个压缩提交工具，不再加载对象正文、核心文件、目录和对象写入协议。校验覆盖范围和实际缩短后，在同一 SQLite 事务中创建概括头、停用原热节点、登记待归档事件、记录回执。尝试顺便写对象会被拒绝。
|
||||
|
||||
事件最多 32,000 字符，原文热节点每片 2,000 字符。一个大事件可能跨多个批次；只有全部原文片段退出热时间轴，整个事件才进入归档队列。再次压缩概括头不会让已归档事件重新入队。
|
||||
|
||||
阈值任务在低于阈值 60% 或固定范围已处理后结束；主动 compact 处理请求范围内全部节点。整理期间新进入的后缀不被覆盖。
|
||||
|
||||
### organize：对象归档
|
||||
|
||||
每批领取一个事件的一页原文，以及预算内的相关对象。Memory 可以浏览全库、检索、读取对象/来源，再分批写人物、任务、项目、联系等。一次提交可更新多个有关对象。
|
||||
|
||||
每个领取的事件必须明确处置：
|
||||
|
||||
| outcome | 含义及程序约束 |
|
||||
|---|---|
|
||||
| continue | 还有下一页，或还有对象工作；保存 progress。末页继续必须确实更新对象/关系/core，不允许无进展空转 |
|
||||
| linked | 全事件读完，至少有一个持久对象/core 目标；允许只挂附件而不重复改写已有事实 |
|
||||
| archive_only | 全事件读完，没有对象目标，明确说明无需更新对象的原因；仍可按时间查询原文 |
|
||||
|
||||
正文、关系、附件目标、页游标、进度和完成处置同事务提交。后续失败不撤销前一批成功写入。下一批读取当前档案和 progress，继续未完成部分。已有对象目标的事件不能再说完全无长期价值；linked/archive_only 不再参与小时扫描。
|
||||
|
||||
### deep：深度对象审阅
|
||||
|
||||
沿用对象归档协议，另外持久建立首次领取时的全库对象清单，包括归档和长期未提及的对象。每批最多审阅四个，显式确认已审阅的 ID；没变化的对象无需重写，失败不推进清单。
|
||||
|
||||
处理跨项目关系、过时结论、重复与可复用知识；不再改热时间轴，不每天强行修改人格。必要时展开原文，不强制每晚重读全量附件。
|
||||
|
||||
边界：未被 compact 摘取的近期事件不会仅因 hourly/deep 而进入归档。若要立刻把最新对话形成对象，先 compact，再 organize/deep；请求本身也是事件。全库可寻址不代表模型已理解全库每句话。
|
||||
|
||||
## 3. 触发与调度
|
||||
|
||||
| 触发 | 行为 |
|
||||
|---|---|
|
||||
| 达到时间轴阈值 | compact，沿用估算 token 口径 |
|
||||
| Super 主动 | memory_organize(mode, scope, retry_blocked?)，只等待入队回执 |
|
||||
| 每 1 小时 | 持久记录检查时间；存在 pending 事件且无在途 hourly 作业才创建 organize，空队列不调用模型 |
|
||||
| 每日 02:50，五分钟无对话 | deep；按本地日期去重，不安静则延后，启动时已过该时间可补本日一次 |
|
||||
|
||||
Worker 每两秒检查，已经开始的归档作业逐批推进，不是每批再等一小时。Worker 在 Agent 进程内；Agent 停止则不调用模型。
|
||||
|
||||
只有一个有效租约。按上次领取时间轮转，未领取过的任务中主动请求优先；持续压缩不再永久压住夜间任务。Super 忙时不领下一批，已经在后台运行的请求不阻塞前台。scope 是给模型的目标和检索线索，不是硬性权限/范围过滤。
|
||||
|
||||
## 4. 对象和召回
|
||||
|
||||
新关系仅用短标签：member_of（对象→项目，可多项目）、related_to、assigned_to、depends_on、about、supersedes、knows。详情写档案，不用长句当标签。旧长句关系保留可读，可在 deep 中归档后建立标准关系，不由程序猜测映射。
|
||||
|
||||
新项目强制 candidate，Super 询问用户后才可激活。无明确项目可联系日常，但不强制为每条事件建立对象或项目，也不强制每对象只属一个项目。
|
||||
|
||||
档案维护名称、别名、摘要、正文、cues（未来什么问题会用到它）与来源。Super 同步用 memory_find 查询姓名/词项/cues/项目直接关系，再用 memory_read 读档案，memory_evidence 展开附件。没有 embedding，没有自动每轮塞入全部对象。
|
||||
|
||||
常驻仍是三份核心 MD、有限目录、热时间轴及本轮输入。档案增长不直接扩大每轮上下文。原文、历史版本和审计会增加磁盘占用，尚无自动删除策略。
|
||||
|
||||
## 5. 参数与恢复
|
||||
|
||||
| 参数 | 当前值 |
|
||||
|---|---|
|
||||
| compact | 每批最多 8 个节点，容量失败可缩批 |
|
||||
| organize 原文页 | 每批 1 事件，最多 8,000 字符，按字节余量/失败缩小，最小 256 |
|
||||
| deep 对象 | 每批最多 4 个，按容量缩小 |
|
||||
| 领取传输预算 | Worker 默认 96,000 字节，协议允许 24,000–200,000；不是 token |
|
||||
| progress / reason | 最多 2,000 / 1,000 字符 |
|
||||
| 对象正文 / 摘要 | 最多 16,000 / 400 字符；不建议写满 |
|
||||
| 单次图写入 | 最多 32 对象、64 关系、3 core；提案小于 900,000 传输字节 |
|
||||
| 模型输出 | 默认 32,768，截断重试增长至 131,072 token，仍受模型及剩余上下文限制 |
|
||||
| 批次 | 默认 300 秒、16 轮、3 次字段级修正；沿用 MEMORY 环境配置 |
|
||||
| 租约 | Worker 超时 + 60 秒，默认 360 秒 |
|
||||
| 重试 | 同批最多连续 5 次失败领取，退避后终止；成功批次重置计数 |
|
||||
| Markdown 投影 | 每次最多 10 个对象，内容不变不重写，后续 heartbeat 继续 |
|
||||
|
||||
同 token 重投返回原回执；旧图版本、过期租约、未读对象覆盖、越界证据、漏事件或跳过未读页均拒绝。保证数据库幂等提交，不保证模型只调用一次。
|
||||
|
||||
终止失败只将该批事件标为 blocked，原文和先前检查点保留；其它事件仍可由后续小时任务处理。memory_status 给出各状态数量和最早 20 条未完成记录。存在 blocked 的范围只能算 partial，不能伪报全完成。修复原因后显式 retry_blocked=true 解封，重复旧请求不会再次解封。
|
||||
|
||||
阈值失败后抑制自动新建一小时，避免“失败通知入时间轴→再建五次失败作业”的循环。主动请求不受自动冷却影响。
|
||||
|
||||
## 6. 如何亲自阅读
|
||||
|
||||
部署后的默认入口：`/home/kickpi/.deskpaw/memory/objects/README.md`。按名称点击人物/任务/待办/日程/知识/项目档案；底层文件名为稳定 ID 哈希，避免重名与路径问题。各类型位于 person/task/todo/schedule/knowledge/project/credential 子目录。
|
||||
|
||||
一对象一份当前正文、检索线索、关系及附件时间/ID；每文件最多展示 50 条关系/来源，更多可分页。SQLite 权威，对象文件不导入人工修改；投影分批生成可能暂时落后，失败在 status 中显示。三份核心 MD 的人工编辑导入机制不变。
|
||||
|
||||
在 deskpaw-memory 目录只读查看，不调用模型：
|
||||
|
||||
```bash
|
||||
node src/inspect.js status
|
||||
node src/inspect.js list person
|
||||
node src/inspect.js object person:qhy
|
||||
node src/inspect.js events '2026-10-13T12:30:00+08:00' '2026-10-13T13:30:00+08:00'
|
||||
node src/inspect.js event '实际事件ID'
|
||||
```
|
||||
|
||||
对象/事件返回 next_offset，时间列表返回 cursor，可作为最后一个参数续读。person:qhy 是示例，不表示生产库已有此人。
|
||||
|
||||
## 7. 迁移、验证和未完成项
|
||||
|
||||
首次初始化为旧的已摘取事件补 pending 记录及已有对象来源关系；不删除原文/档案/版本。旧协议 running 快照被释放以重新领取；部署必须停止旧 Agent 后切换两端，不混跑。历史 failed 不改成 done。升级代码不会瞬间生成所有人物档案，需要后续模型渐进整理。
|
||||
|
||||
验证覆盖两阶段事务、原文保留、无价值终态、全事件页覆盖、检查点、回执重投、过期租约、隔离恢复、每小时空转、夜间轮转、项目审批、来源防回灌、Markdown 文件，以及真实临时 socket/独立 Worker/Super 同步召回。
|
||||
|
||||
本轮结果：Memory `npm test` **63/63**；Agent memory-recovery、chain-memory、timeline-bridge、super-context、super-session 五组 **25/25**，无跳过；Agent `npm run check` 与两个仓库 `git diff --check` 通过。均使用隔离数据和模拟模型，没有收费调用或真实键鼠动作。
|
||||
|
||||
另使用 SQLite backup 从生产库生成一致性副本，在副本执行迁移及 Markdown 投影：迁移前后均为 876 事件、17 对象、1,145 节点，补出 250 条待归档事件；integrity_check=ok，投影错误为空。该验证未写入生产库，不代表线上已经迁移或语义整理完成。
|
||||
|
||||
未验收：真实模型长期保持档案简洁与正确、跨项目关系质量、连续整夜稳定性、断电和大规模磁盘增长。未实现：embedding、自动物理删除、合并对象的重定向、记忆图形编辑 UI、完整模型用量总账。自动测试不等于语义质量或生产长稳验收。
|
||||
|
||||
## 8. 2026-10-14 实际部署与存量状态
|
||||
|
||||
00:33:43 Memory 启动新代码，00:34:15 Agent 启动新 Worker。停服备份和验证见 [FullGet 配套部署记录](../deskpaw_agent/FULL_GET.md#12-2026-10-14-v2-与对象记忆部署)。部署前追加回归 Memory 63/63，Agent Super/Memory/PawCode 五组 44/44；Agent 构建成功。
|
||||
|
||||
部署前已经有 120 个成功批次回执、29 个对象（人物 2、待办 11、知识 15、项目 1），最后旧对象更新时间 00:18。这证明旧模型确实做过整理，但不是新版整理完成；存量还有 11 个失败 compact、2 个失败 deep，历史状态没有篡改为成功。部分对象标题/正文偏流水账，旧长句关系也仍存在，需后续 organize/deep 逐步维护。
|
||||
|
||||
迁移保留 886 条原始事件和 29 个对象，将已摘取经历补入待归档队列，初始 632 条。Markdown 投影由服务分批生成，objects/README.md 与人物/待办/知识/项目文件已实际存在；不是手工编造样例。投影错误为空，Runtime/Memory quick_check 均为 ok。
|
||||
|
||||
部署后观察到真实模型 compact 成功提交 revision 121、122，时间轴由 322 节点/264,447 估算 token 减至 308 节点/252,642；待归档增长至 641 条，因为新摘取经历转入下一阶段。小时 organize 已入队，不能把队列建立等同完成。失败历史保留，当前 Worker 错误已在成功提交后清除。
|
||||
|
||||
上下文检查：本机 Super 模型登记 contextWindow=1,000,000、maxTokens=384,000；输入预算 800,000、输出预留 150,000、安全余量 50,000。最近一次部署前请求估算 307,716，提供商报告输入 134,700、输出 136；不要把估算当真实计费。Computer 输入预算 128,000、64,000 压缩、近期 16,000、摘要 4,096、输出 8,192。未修改这些配置。
|
||||
|
||||
另有独立时间轴传输限制 max_bytes=350,000，最近快照只装入 241/308 节点、67 节点未进入本次视图；虽然 Super 尚有模型容量，字节传输限额仍会先触发显式截断。当前核心文件实际只有 AGENTS.md（1,907 字符），SOUL.md/USER.md 尚未生成；不能把代码支持三份文件描述成三份已经有完整内容。
|
||||
@ -2,6 +2,10 @@
|
||||
|
||||
## 当前：链式记忆与 Computer 增量经历
|
||||
|
||||
两阶段对象整理已于 **2026-10-14 00:34 配套部署**：[实现、参数、阅读方法与自查](OBJECT_MEMORY.md)。compact 只缩短时间轴并登记待归档事件；每小时 organize 更新对象档案，允许 archive_only 终态；deep 另做全库审阅。对象 Markdown 目录已生成，存量仍在后台整理，不能视作已全部整理完成。下文旧“三入口、compact 同时写对象”是历史基线，以新文档为准。
|
||||
|
||||
2026-10-13 待部署修复位于 `fix/memory-organizer-recovery`：新增分类重试状态、失败历史和自动整理失败通知，允许 Worker 按批次超时设置租约；对象正文上限放宽至 16,000 字符。`memory.organize.notices` 先取待通知作业，以 `{ack: job_id}` 确认；`memory.chain.status` 返回 retry 和最近五次 failures。通知在 Agent 持久接收后才确认，历史自动失败会补通知，但不自动重跑。对应 Agent 输出默认 32,768、截断增长至默认 131,072 token，整批五分钟。完整说明见 [整理恢复](../deskpaw_agent/CHAIN_MEMORY.md#16-2026-10-13-整理失败恢复与输出额度)。下文部署记录为 10 月 12 日基线，不代表本修复已上线。
|
||||
|
||||
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 是历史记录，不代表当前部署或余额状态。
|
||||
@ -10,7 +14,8 @@ Computer 现在直接将执行文本增量写入共享时间轴，不等整个
|
||||
|
||||
完整跨仓库说明与自查位于相邻 Agent 仓库的 `CHAIN_MEMORY.md`（本机 `/home/kickpi/deskpaw_agent/CHAIN_MEMORY.md`）。
|
||||
|
||||
- `chain-store.js`：原始事件分片建链；达到阈值、Super 主动请求、每日 02:50 且五分钟无对话三种触发。compact 局部提取；deep 遍历时间轴和首次领取时的全库对象快照，包括归档及近期无关对象。审阅进度持久化，失败不推进，纯对象批次不制造经历。版本、租约、退避、覆盖校验及事务发布。
|
||||
- `chain-store.js` / `object-organizer.js`：时间轴摘取、持久归档、每小时扫描、夜间对象审阅；分批检查点、失败隔离及轮转。compact 与对象写入是独立事务阶段。
|
||||
- `object-projection.js` / `inspect.js`：一对象一份 Markdown 档案，以及对象和原始事件的只读查询命令。
|
||||
- `chain-graph.js`：人物、任务、待办、日程、知识、交叉项目、凭据引用；来源和版本，别名/词项/cues 索引、有限关系展开、分页读取。新项目为候选，需逐字用户证据确认。精简目录常驻 Super，事实按需召回。
|
||||
- `core-projection.js`：SOUL/USER/AGENTS 三份文件，每份最多 2,400 字符；数据库权威版本与可编辑文件投影，下一轮热加载。凭据真实值禁止存入。
|
||||
|
||||
@ -28,7 +33,7 @@ Computer 现在直接将执行文本增量写入共享时间轴，不等整个
|
||||
| `memory.chain.project_decision` | 当前用户证据确认/放弃候选项目 |
|
||||
| `memory.organize.request/claim/inspect/commit/fail` | 持久请求、领取有限批次、补读旧记忆、原子发布、失败退避 |
|
||||
|
||||
长期内容和摘要同一事务发布；重试回执幂等，旧租约不得覆盖新版本。原文、审计和版本不会因摘要缩短而删除，因此仍需磁盘监控和备份。未实现自动回收、会议采集、任意同义检索保证或密钥库全量元数据同步。Memory 故障保留 Agent 待同步内容，不阻止接收任务。
|
||||
时间轴替换与归档入队同事务，对象写入与事件处置另用独立事务；两者均有幂等回执、版本与租约校验。原文、审计和版本不会因摘要缩短而删除，仍需磁盘监控和备份。未实现自动回收、会议采集、任意同义检索保证或密钥库全量元数据同步。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` 表示深度任务的对象总量和已成功提交的审阅量，不证明模型语义判断必然正确。
|
||||
|
||||
|
||||
@ -3,6 +3,7 @@ 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 const RELATIONS = ["member_of", "related_to", "assigned_to", "depends_on", "about", "supersedes", "knows"];
|
||||
export function key(value) {
|
||||
if (typeof value !== "string" || !/^[\w:./-]{1,240}$/.test(value)) throw new Error("Invalid memory ID");
|
||||
return value;
|
||||
@ -81,7 +82,7 @@ export class ChainGraph {
|
||||
const sourceRefs = refs(item.source_refs);
|
||||
const name = clean(item.name, 160);
|
||||
const summary = clean(item.summary, 400);
|
||||
const body = clean(item.body, 6000);
|
||||
const body = clean(item.body, 16000);
|
||||
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));
|
||||
@ -100,7 +101,7 @@ export class ChainGraph {
|
||||
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 term of words(text, text.length)) 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);
|
||||
@ -108,6 +109,10 @@ export class ChainGraph {
|
||||
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");
|
||||
// Legacy prose relations remain readable and can be retired, but never added anew.
|
||||
if (!RELATIONS.includes(relation) && !(item.status === "archived" && this.db.prepare("SELECT 1 FROM memory_edges WHERE id=?").get(id)))
|
||||
throw new Error("Use a supported directed relation");
|
||||
if (relation === "member_of" && this.object(to).type !== "project") throw new Error("Membership target must be a project");
|
||||
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);
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { ChainGraph, clean, CORE_NAMES, integer, key, sourceKey } from "./chain-graph.js";
|
||||
import { estimateTokens } from "./token-budget.js";
|
||||
import { ObjectOrganizer } from "./object-organizer.js";
|
||||
|
||||
/** One personal timeline; raw evidence is immutable, active heads are versioned projections. */
|
||||
export class ChainStore {
|
||||
@ -24,7 +25,20 @@ export class ChainStore {
|
||||
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_organize_retry (
|
||||
job_id TEXT PRIMARY KEY,output_failures INTEGER NOT NULL DEFAULT 0,shrink INTEGER NOT NULL DEFAULT 0,
|
||||
kind TEXT NOT NULL,error TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS memory_organize_failures (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,job_id TEXT NOT NULL,token TEXT UNIQUE NOT NULL,
|
||||
attempt INTEGER NOT NULL,kind TEXT NOT NULL,error TEXT NOT NULL,at INTEGER NOT NULL);
|
||||
CREATE INDEX IF NOT EXISTS memory_organize_failure_job ON memory_organize_failures(job_id,seq);
|
||||
CREATE TABLE IF NOT EXISTS memory_organize_notices (
|
||||
job_id TEXT PRIMARY KEY,created_at INTEGER NOT NULL,sent INTEGER NOT NULL DEFAULT 0);
|
||||
CREATE TABLE IF NOT EXISTS memory_project_decisions (request_id TEXT PRIMARY KEY,project_id TEXT,status TEXT,source_id TEXT);`);
|
||||
// Backfill previously invisible automatic failures once, without restarting failed work.
|
||||
db.exec(`INSERT OR IGNORE INTO memory_organize_notices(job_id,created_at)
|
||||
SELECT id,created_at FROM memory_organize_jobs WHERE status='failed' AND trigger!='explicit'`);
|
||||
this.organizer = new ObjectOrganizer(this);
|
||||
}
|
||||
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)); }
|
||||
@ -80,11 +94,16 @@ export class ChainStore {
|
||||
}
|
||||
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");
|
||||
if (!["compact", "organize", "deep"].includes(mode) || !["threshold", "explicit", "nightly", "hourly"].includes(trigger)) throw new Error("Invalid organization mode/trigger");
|
||||
const dedupe = key(p.request_id), scope = clean(p.scope ?? "", 2000);
|
||||
const previous = this.db.prepare("SELECT id,status,mode,trigger,cutoff FROM memory_organize_jobs WHERE dedupe=?").get(dedupe);
|
||||
if (previous) return previous;
|
||||
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();
|
||||
// Explicit requests may retry a quarantined attachment; automatic ticks cannot create a failure loop.
|
||||
if (trigger === "explicit" && mode !== "compact" && p.retry_blocked === true)
|
||||
this.db.prepare("UPDATE memory_event_organization SET status='pending',reason='' WHERE status='blocked'").run();
|
||||
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',?,?)`)
|
||||
@ -101,8 +120,14 @@ export class ChainStore {
|
||||
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)
|
||||
if (stats.caught_up && stats.estimated_tokens >= threshold && !pending && now >= Number(this.meta("automatic_retry_after:compact", "0")))
|
||||
this.request({ request_id: `threshold:${stats.revision}:${stats.ingested}`, trigger: "threshold" }, now);
|
||||
const lastHourly = Number(this.meta("last_hourly_check", "0"));
|
||||
if (!lastHourly || now - lastHourly >= 3600000) {
|
||||
this.set("last_hourly_check", now);
|
||||
if (this.organizer.pending() && !this.db.prepare("SELECT 1 FROM memory_organize_jobs WHERE trigger='hourly' AND status IN ('pending','running')").get())
|
||||
this.request({ request_id: `hourly:${Math.floor(now / 3600000)}`, mode: "organize", trigger: "hourly" }, 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) {
|
||||
@ -119,29 +144,34 @@ export class ChainStore {
|
||||
this.sync();
|
||||
if (!this.caughtUp()) return { job: null };
|
||||
const maxBytes = integer(p.max_bytes ?? 96000, 24000, 200000);
|
||||
const leaseMs = integer(p.lease_ms ?? 180000, 30000, 1860000);
|
||||
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`)
|
||||
const job = this.db.prepare(`SELECT j.* FROM memory_organize_jobs j LEFT JOIN memory_job_service s ON s.job_id=j.id
|
||||
WHERE j.status='pending' AND j.available_at<=? AND (j.trigger!='nightly' OR ?=1)
|
||||
ORDER BY coalesce(s.last_claimed,0),CASE j.trigger WHEN 'explicit' THEN 0 WHEN 'threshold' THEN 1 ELSE 2 END,j.created_at,j.id LIMIT 1`)
|
||||
.get(now, Number(!busy && quiet));
|
||||
if (!job) return { job: null };
|
||||
this.db.prepare("INSERT OR REPLACE INTO memory_job_service VALUES (?,?)").run(job.id, now);
|
||||
// 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") {
|
||||
if (!job.cursor && job.trigger === "explicit" && !this.db.prepare("SELECT 1 FROM memory_organize_batches WHERE job_id=? LIMIT 1").get(job.id)) {
|
||||
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();
|
||||
// Output truncation first grows the worker's output allowance. Only capacity/budget
|
||||
// failures shrink input; schema, network and version errors do not drop batch coverage.
|
||||
const retry = this.db.prepare("SELECT * FROM memory_organize_retry WHERE job_id=?").get(job.id);
|
||||
const shrink = Math.min(retry?.shrink ?? 0, 3);
|
||||
const nodeLimit = Math.max(1, Math.floor(8 / 2 ** shrink));
|
||||
const candidates = job.mode === "compact" ? 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 = job.mode === "compact" ? [] : this.graph.core(), catalog = job.mode === "compact" ? {} : 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");
|
||||
@ -152,21 +182,15 @@ export class ChainStore {
|
||||
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) {
|
||||
const prepared = job.mode === "compact" ? { events: [], reviewObjects: [], related: [] } : this.organizer.prepare(job, available, shrink);
|
||||
const { events, reviewObjects, related } = prepared;
|
||||
if (!nodes.length && !reviewObjects.length && !events.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);
|
||||
const blocked = job.mode === "compact" ? 0 : this.db.prepare("SELECT count(*) AS n FROM memory_event_organization WHERE status='blocked' AND event_seq<=?").get(job.cutoff).n;
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status=?,receipt=? WHERE id=?")
|
||||
.run(blocked ? "partial" : "done", JSON.stringify({ revision: this.revision(), noop: true, blocked_events: blocked }), job.id);
|
||||
if (blocked && job.trigger !== "explicit") this.db.prepare("INSERT OR IGNORE INTO memory_organize_notices(job_id,created_at) VALUES (?,?)").run(job.id, now);
|
||||
return { job: null };
|
||||
}
|
||||
const refs = new Set();
|
||||
@ -189,21 +213,17 @@ export class ChainStore {
|
||||
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 maxSummary = job.mode === "compact" ? Math.min(6000, Math.max(300, Math.floor(nodes.reduce((n, row) => n + row.text.length, 0) * 0.45))) : 6000;
|
||||
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 };
|
||||
const snapshot = { job_id: job.id, revision: this.revision(), nodes: nodes.map(n => n.id), events: events.map(({ text, source, targets, ...event }) => event), reviewed_objects: reviewObjects.map(item => item.id), evidence_ids: [...new Set([...events.map(e => e.id), ...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);
|
||||
.run(token, now + leaseMs, JSON.stringify(snapshot), job.id);
|
||||
return { job: { id: job.id, token, mode: job.mode, trigger: job.trigger, scope: job.scope,
|
||||
attempt: job.attempts + 1, output_failures: retry?.output_failures ?? 0,
|
||||
retry_feedback: retry ? { kind: retry.kind, error: retry.error } : null,
|
||||
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,
|
||||
events, 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 } };
|
||||
});
|
||||
}
|
||||
@ -217,8 +237,12 @@ export class ChainStore {
|
||||
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");
|
||||
if (typeof p.summary === "string" && p.summary.length > snapshot.max_summary_chars)
|
||||
throw new Error(`summary: maxLength (limit=${snapshot.max_summary_chars})`);
|
||||
const summary = clean(p.summary, snapshot.max_summary_chars);
|
||||
if (!summary) throw new Error("Empty timeline summary");
|
||||
if (job.mode === "compact" && ["objects", "edges", "core", "catalog", "dispositions", "reviewed_object_ids"].some(name => p[name]?.length))
|
||||
throw new Error("Compaction only replaces timeline heads; object work is queued separately");
|
||||
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");
|
||||
@ -226,27 +250,35 @@ export class ChainStore {
|
||||
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);
|
||||
if (job.mode !== "compact") {
|
||||
this.graph.apply(p, revision, evidence, now);
|
||||
this.organizer.apply(p, snapshot, revision, 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);
|
||||
this.organizer.enqueue(nodes.filter(n => n.event_id).map(n => n.event_id), now);
|
||||
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;
|
||||
const done = job.mode === "compact" ? (job.trigger === "threshold" && belowTarget) || !remaining : !this.organizer.pending(job.cutoff) && !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 };
|
||||
const blocked = job.mode === "compact" ? 0 : this.db.prepare("SELECT count(*) AS n FROM memory_event_organization WHERE status='blocked' AND event_seq<=?").get(job.cutoff).n;
|
||||
const receipt = { job_id: job.id, revision, node_id: id, status: done ? blocked ? "partial" : "done" : "pending", trigger: job.trigger,
|
||||
mode: job.mode, blocked_events: blocked, organization: this.organizer.counts() };
|
||||
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);
|
||||
this.db.prepare("DELETE FROM memory_organize_retry WHERE job_id=?").run(job.id);
|
||||
if (receipt.status === "partial" && job.trigger !== "explicit")
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_organize_notices(job_id,created_at) VALUES (?,?)").run(job.id, now);
|
||||
return receipt;
|
||||
});
|
||||
}
|
||||
@ -271,24 +303,57 @@ export class ChainStore {
|
||||
});
|
||||
}
|
||||
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 };
|
||||
return this.transaction(() => {
|
||||
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 kind = p.kind ?? "unknown";
|
||||
if (!["output", "schema", "evidence", "conflict", "context", "timeout", "transient", "budget", "unknown"].includes(kind)) throw new Error("Invalid failure kind");
|
||||
const error = clean(String(p.error ?? "organization failed").slice(0, 500), 500);
|
||||
const status = job.attempts >= 5 ? "failed" : "pending";
|
||||
if (status === "failed") {
|
||||
this.set(`automatic_retry_after:${job.mode}`, now + 3600000);
|
||||
for (const event of JSON.parse(job.snapshot).events ?? [])
|
||||
this.db.prepare("UPDATE memory_event_organization SET status='blocked',reason=?,updated_at=? WHERE event_id=? AND status='pending'").run(error, now, event.id);
|
||||
}
|
||||
this.db.prepare("INSERT INTO memory_organize_failures(job_id,token,attempt,kind,error,at) VALUES (?,?,?,?,?,?)")
|
||||
.run(job.id, job.token, job.attempts, kind, error, now);
|
||||
this.db.prepare(`INSERT INTO memory_organize_retry(job_id,output_failures,shrink,kind,error) VALUES (?,?,?,?,?)
|
||||
ON CONFLICT(job_id) DO UPDATE SET output_failures=output_failures+excluded.output_failures,
|
||||
shrink=shrink+excluded.shrink,kind=excluded.kind,error=excluded.error`)
|
||||
.run(job.id, Number(kind === "output"), Number(p.shrink_batch === true || kind === "unknown"), kind, error);
|
||||
this.db.prepare("UPDATE memory_organize_jobs SET status=?,token=NULL,lease_until=0,snapshot=NULL,error=?,available_at=? WHERE id=?")
|
||||
.run(status, error, now + Math.min(300000, 5000 * 2 ** job.attempts), job.id);
|
||||
if (status === "failed" && job.trigger !== "explicit")
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_organize_notices(job_id,created_at) VALUES (?,?)").run(job.id, now);
|
||||
return { released: true, status, kind };
|
||||
});
|
||||
}
|
||||
notices(p = {}) {
|
||||
if (p.ack) {
|
||||
this.db.prepare("UPDATE memory_organize_notices SET sent=1 WHERE job_id=?").run(key(p.ack));
|
||||
return { acknowledged: true };
|
||||
}
|
||||
return { jobs: this.db.prepare(`SELECT j.id,j.mode,j.trigger,j.status,j.cursor,j.cutoff,j.attempts,j.error
|
||||
FROM memory_organize_notices n JOIN memory_organize_jobs j ON j.id=n.job_id
|
||||
WHERE n.sent=0 ORDER BY n.created_at,n.job_id LIMIT 20`).all() };
|
||||
}
|
||||
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,
|
||||
retry: this.db.prepare("SELECT output_failures,shrink,kind,error FROM memory_organize_retry WHERE job_id=?").get(job.id) ?? null,
|
||||
failures: this.db.prepare("SELECT attempt,kind,error,at FROM memory_organize_failures WHERE job_id=? ORDER BY seq DESC LIMIT 5").all(job.id),
|
||||
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 };
|
||||
})), organization: this.organizer.counts(), organization_backlog: this.db.prepare("SELECT event_id,status,offset,checkpoints,reason,updated_at FROM memory_event_organization WHERE status IN ('pending','blocked') ORDER BY event_seq LIMIT 20").all(),
|
||||
next_hourly_check: Number(this.meta("last_hourly_check", "0")) + 3600000,
|
||||
object_projection_error: this.meta("object_projection_error") || null, 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) };
|
||||
if (!row) return { ...this.timeline.read(p), organization: this.organizer.detail(p.id) };
|
||||
const offset = integer(p.offset ?? 0, 0, 1000000), limit = integer(p.limit ?? 8000, 1, 32000);
|
||||
return { ...row, text: row.text.slice(offset, offset + limit), next_offset: offset + limit < row.text.length ? offset + limit : null,
|
||||
children: JSON.parse(row.children) };
|
||||
}
|
||||
bootstrapCore(p) {
|
||||
return this.transaction(() => {
|
||||
|
||||
40
src/inspect.js
Normal file
40
src/inspect.js
Normal file
@ -0,0 +1,40 @@
|
||||
#!/usr/bin/env node
|
||||
import { createConnection } from "node:net";
|
||||
import { homedir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
const [command, ...args] = process.argv.slice(2);
|
||||
const time = value => {
|
||||
const n = Date.parse(value);
|
||||
if (!Number.isSafeInteger(n) || n < 0) throw new Error("Use an ISO timestamp including timezone");
|
||||
return n;
|
||||
};
|
||||
let method, params;
|
||||
switch (command) {
|
||||
case "status": method = "memory.chain.status"; params = {}; break;
|
||||
case "list": method = "memory.chain.find"; params = { type: args[0], offset: Number(args[1] ?? 0), limit: 30 }; break;
|
||||
case "object": method = "memory.chain.read"; params = { id: args[0], offset: Number(args[1] ?? 0), limit: 50 }; break;
|
||||
case "event": method = "memory.chain.node"; params = { id: args[0], offset: Number(args[1] ?? 0), limit: 32000 }; break;
|
||||
case "events": method = "timeline.query"; params = { from: time(args[0]), to: time(args[1]), after: Number(args[2] ?? 0), limit: 50 }; break;
|
||||
default: throw new Error("Usage: node src/inspect.js status | list [type] [offset] | object ID [offset] | event ID [offset] | events FROM_ISO TO_ISO [cursor]");
|
||||
}
|
||||
const socketPath = process.env.DESKPAW_MEMORY_SOCKET ?? join(process.env.DESKPAW_STATE_DIR ?? join(homedir(), ".deskpaw"), "runtime", "memory.sock");
|
||||
const result = await new Promise((resolve, reject) => {
|
||||
const socket = createConnection(socketPath); let data = "";
|
||||
socket.setEncoding("utf8");
|
||||
socket.setTimeout(10000, () => socket.destroy(new Error("Memory read timed out")));
|
||||
socket.once("error", reject);
|
||||
socket.once("connect", () => socket.write(JSON.stringify({ id: "inspect", method, params }) + "\n"));
|
||||
socket.on("data", chunk => {
|
||||
data += chunk;
|
||||
if (Buffer.byteLength(data) > 2000000) socket.destroy(new Error("Memory response too large"));
|
||||
});
|
||||
socket.once("end", () => {
|
||||
try {
|
||||
const response = JSON.parse(data);
|
||||
if (!response.ok) throw new Error(response.error);
|
||||
resolve(response.result);
|
||||
} catch (error) { reject(error); }
|
||||
});
|
||||
});
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
154
src/object-organizer.js
Normal file
154
src/object-organizer.js
Normal file
@ -0,0 +1,154 @@
|
||||
import { clean, sourceKey } from "./chain-graph.js";
|
||||
|
||||
/** Durable handoff: detaching timeline text is not the same as organizing its facts. */
|
||||
export class ObjectOrganizer {
|
||||
constructor(chain) {
|
||||
this.chain = chain;
|
||||
this.db = chain.db;
|
||||
this.db.exec(`CREATE TABLE IF NOT EXISTS memory_event_organization (
|
||||
event_id TEXT PRIMARY KEY,event_seq INTEGER NOT NULL,status TEXT NOT NULL DEFAULT 'pending',
|
||||
offset INTEGER NOT NULL DEFAULT 0,progress TEXT NOT NULL DEFAULT '',reason TEXT NOT NULL DEFAULT '',
|
||||
updated_at INTEGER NOT NULL,checkpoints INTEGER NOT NULL DEFAULT 0);
|
||||
CREATE INDEX IF NOT EXISTS memory_event_organization_pending ON memory_event_organization(status,event_seq);
|
||||
CREATE TABLE IF NOT EXISTS memory_event_targets (
|
||||
event_id TEXT,target_id TEXT,target_kind TEXT,PRIMARY KEY(event_id,target_id,target_kind));
|
||||
CREATE TABLE IF NOT EXISTS memory_job_service (job_id TEXT PRIMARY KEY,last_claimed INTEGER NOT NULL);
|
||||
CREATE INDEX IF NOT EXISTS memory_nodes_event ON memory_nodes(event_id,active);`);
|
||||
// Previously compacted evidence was never given a durable completion decision.
|
||||
// Reconcile it once; do not treat an old object reference as proof of complete review.
|
||||
if (chain.meta("object_pipeline_version") !== "1") chain.transaction(() => {
|
||||
this.db.exec(`INSERT OR IGNORE INTO memory_event_organization(event_id,event_seq,updated_at)
|
||||
SELECT t.id,t.seq,t.at FROM timeline t
|
||||
WHERE EXISTS(SELECT 1 FROM memory_nodes n WHERE n.event_id=t.id)
|
||||
AND NOT EXISTS(SELECT 1 FROM memory_nodes n WHERE n.event_id=t.id AND n.active=1);
|
||||
UPDATE memory_organize_jobs SET status='pending',token=NULL,snapshot=NULL,lease_until=0
|
||||
WHERE status='running';`);
|
||||
this.db.exec(`INSERT OR IGNORE INTO memory_event_targets SELECT s.event_id,s.id,'object'
|
||||
FROM memory_object_sources s JOIN memory_event_organization q ON q.event_id=s.event_id;
|
||||
INSERT OR IGNORE INTO memory_event_targets SELECT j.value,c.name,'core'
|
||||
FROM memory_core c,json_each(c.source_refs) j JOIN memory_event_organization q ON q.event_id=j.value;
|
||||
INSERT OR IGNORE INTO memory_event_targets SELECT j.value,e.from_id,'object'
|
||||
FROM memory_edges e,json_each(e.source_refs) j JOIN memory_event_organization q ON q.event_id=j.value;
|
||||
INSERT OR IGNORE INTO memory_event_targets SELECT j.value,e.to_id,'object'
|
||||
FROM memory_edges e,json_each(e.source_refs) j JOIN memory_event_organization q ON q.event_id=j.value;`);
|
||||
chain.set("object_pipeline_version", "1");
|
||||
});
|
||||
}
|
||||
enqueue(ids, now) {
|
||||
for (const id of new Set(ids)) {
|
||||
this.db.prepare(`INSERT OR IGNORE INTO memory_event_organization(event_id,event_seq,updated_at)
|
||||
SELECT id,seq,? FROM timeline t WHERE id=?
|
||||
AND NOT EXISTS(SELECT 1 FROM memory_nodes n WHERE n.event_id=t.id AND n.active=1)`).run(now, id);
|
||||
this.db.prepare(`INSERT OR IGNORE INTO memory_event_targets SELECT event_id,id,'object' FROM memory_object_sources
|
||||
WHERE event_id=? AND EXISTS(SELECT 1 FROM memory_event_organization WHERE event_id=?)`).run(id, id);
|
||||
this.db.prepare(`INSERT OR IGNORE INTO memory_event_targets SELECT j.value,c.name,'core'
|
||||
FROM memory_core c,json_each(c.source_refs) j JOIN memory_event_organization q ON q.event_id=j.value WHERE j.value=?`).run(id);
|
||||
}
|
||||
}
|
||||
counts() {
|
||||
return this.db.prepare("SELECT status,count(*) AS count FROM memory_event_organization GROUP BY status ORDER BY status").all().map(row => ({ ...row }));
|
||||
}
|
||||
pending(cutoff = Number.MAX_SAFE_INTEGER) {
|
||||
return this.db.prepare("SELECT 1 FROM memory_event_organization WHERE status='pending' AND event_seq<=? LIMIT 1").get(cutoff);
|
||||
}
|
||||
detail(id) {
|
||||
const row = this.db.prepare("SELECT * FROM memory_event_organization WHERE event_id=?").get(sourceKey(id));
|
||||
return row ? { ...row, targets: this.targets(id) } : null;
|
||||
}
|
||||
targets(id) {
|
||||
return this.db.prepare("SELECT target_id,target_kind FROM memory_event_targets WHERE event_id=? ORDER BY target_kind,target_id").all(id);
|
||||
}
|
||||
/** Invoked inside the same transaction that grants the lease. */
|
||||
prepare(job, available, shrink) {
|
||||
const events = [], reviewObjects = [], related = [];
|
||||
let bytes = 0;
|
||||
const queued = this.db.prepare(`SELECT q.*,t.at,t.kind,t.source,t.text FROM memory_event_organization q
|
||||
JOIN timeline t ON t.id=q.event_id WHERE q.status='pending' AND q.event_seq<=?
|
||||
ORDER BY q.event_seq LIMIT 1`).get(job.cutoff);
|
||||
if (queued) {
|
||||
const size = Math.max(256, Math.min(Math.floor(8000 / 2 ** shrink), Math.floor((available - 6000) / 6)));
|
||||
const end = Math.min(queued.text.length, queued.offset + size);
|
||||
const targets = this.targets(queued.event_id);
|
||||
const event = { id: queued.event_id, at: queued.at, kind: queued.kind, source: JSON.parse(queued.source),
|
||||
text: queued.text.slice(queued.offset, end), offset: queued.offset, end_offset: end,
|
||||
total_chars: queued.text.length, has_more: end < queued.text.length,
|
||||
progress: queued.progress, targets: targets.slice(0, 10), total_targets: targets.length, checkpoints: queued.checkpoints };
|
||||
while (Buffer.byteLength(JSON.stringify(event)) > available && event.text.length > 256) {
|
||||
event.text = event.text.slice(0, Math.max(256, event.text.length - 512));
|
||||
event.end_offset = event.offset + event.text.length;
|
||||
event.has_more = event.end_offset < queued.text.length;
|
||||
}
|
||||
if (Buffer.byteLength(JSON.stringify(event)) > available) throw new Error("Memory batch budget too small for one event page");
|
||||
bytes += Buffer.byteLength(JSON.stringify(event));
|
||||
events.push(event);
|
||||
}
|
||||
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 ** shrink)))) {
|
||||
let item = this.chain.graph.read({ id: row.object_id, limit: 10 });
|
||||
if (bytes + Buffer.byteLength(JSON.stringify(item)) > available)
|
||||
item = this.chain.graph.read({ id: row.object_id, limit: 1 });
|
||||
const size = Buffer.byteLength(JSON.stringify(item));
|
||||
if (bytes + size > available) break;
|
||||
reviewObjects.push(item); bytes += size;
|
||||
}
|
||||
}
|
||||
// Claimed sources are complete metadata plus a bounded page, not giant replayed tool blobs.
|
||||
for (const hit of this.chain.graph.find({ query: job.scope || events.map(e => e.text).join(" ").slice(0, 1000), limit: 8 }).hits) {
|
||||
if (reviewObjects.some(item => item.id === hit.id)) continue;
|
||||
const item = this.chain.graph.read({ id: hit.id, limit: 10 }), size = Buffer.byteLength(JSON.stringify(item));
|
||||
if (bytes + size > available) continue;
|
||||
related.push(item); bytes += size;
|
||||
}
|
||||
if (!events.length && !reviewObjects.length && this.db.prepare("SELECT 1 FROM memory_job_objects WHERE job_id=? AND reviewed=0").get(job.id))
|
||||
throw new Error("Memory batch budget too small for one object");
|
||||
return { events, reviewObjects, related };
|
||||
}
|
||||
/** Object writes, evidence links, page/checkpoint advancement and completion are one transaction. */
|
||||
apply(p, snapshot, revision, now) {
|
||||
if ((p.covered_node_ids ?? []).length) throw new Error("Object organization cannot change timeline nodes");
|
||||
const dispositions = p.dispositions ?? [], claimed = snapshot.events ?? [];
|
||||
if (!Array.isArray(dispositions) || JSON.stringify(dispositions.map(d => d.event_id).sort()) !== JSON.stringify(claimed.map(e => e.id).sort()))
|
||||
throw new Error("Must account for every claimed event exactly once");
|
||||
const reviewed = p.reviewed_object_ids ?? [];
|
||||
if (!Array.isArray(reviewed) || new Set(reviewed).size !== reviewed.length || reviewed.some(id => !snapshot.reviewed_objects.includes(id)))
|
||||
throw new Error("Review acknowledgement outside claimed objects");
|
||||
for (const item of p.objects ?? []) for (const id of item.source_refs)
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_event_targets VALUES (?,?,'object')").run(id, item.id);
|
||||
for (const item of p.core ?? []) for (const id of item.source_refs)
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_event_targets VALUES (?,?,'core')").run(id, item.name);
|
||||
for (const item of p.edges ?? []) for (const id of item.source_refs) for (const target of [item.from_id, item.to_id])
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_event_targets VALUES (?,?,'object')").run(id, target);
|
||||
for (const disposition of dispositions) {
|
||||
const event = claimed.find(e => e.id === disposition.event_id);
|
||||
const current = this.detail(event.id);
|
||||
if (current?.status !== "pending" || current.offset !== event.offset) throw new Error("Event organization changed");
|
||||
if (!["continue", "linked", "archive_only"].includes(disposition.outcome)) throw new Error("Invalid event disposition");
|
||||
const reason = clean(disposition.reason, 1000), progress = clean(disposition.progress ?? "", 2000);
|
||||
if (!reason) throw new Error("Event disposition requires a reason");
|
||||
if (!Array.isArray(disposition.target_ids ?? []) || (disposition.target_ids ?? []).length > 100) throw new Error("Event target budget exceeded");
|
||||
// Existing facts need not be rewritten just to attach another supporting event.
|
||||
for (const target of disposition.target_ids ?? []) {
|
||||
if (!snapshot.objects_seen.includes(target) && !(p.objects ?? []).some(o => o.id === target)) throw new Error("Inspect attachment target before linking");
|
||||
if (!this.chain.graph.object(target)) throw new Error("Memory object not found");
|
||||
const raw = this.chain.timeline.read({ id: event.id });
|
||||
if (["memory_recall", "memory_organized", "organize_request", "organize_result"].includes(raw.kind)) throw new Error("Recall and maintenance records are not independent evidence");
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_event_targets VALUES (?,?,'object')").run(event.id, target);
|
||||
this.db.prepare("INSERT OR IGNORE INTO memory_object_sources VALUES (?,?)").run(target, event.id);
|
||||
}
|
||||
const targets = this.targets(event.id);
|
||||
if (targets.length > 100) throw new Error("Event target budget exceeded");
|
||||
if (disposition.outcome !== "continue" && event.has_more) throw new Error("Read all event pages before completing organization");
|
||||
if (disposition.outcome === "linked" && !targets.length) throw new Error("Linked event requires a durable target");
|
||||
if (disposition.outcome === "archive_only" && targets.length) throw new Error("Archive-only event cannot have object targets");
|
||||
if (disposition.outcome === "continue" && (!progress || (!event.has_more && (!(p.objects?.length || p.edges?.length || p.core?.length) || progress === current.progress))))
|
||||
throw new Error("Checkpoint must advance an event page or object work");
|
||||
this.db.prepare(`UPDATE memory_event_organization SET status=?,offset=?,progress=?,reason=?,updated_at=?,checkpoints=checkpoints+1 WHERE event_id=?`)
|
||||
.run(disposition.outcome === "continue" ? "pending" : disposition.outcome, event.end_offset, progress, reason, now, event.id);
|
||||
}
|
||||
if (!dispositions.length && !reviewed.length && !(p.objects?.length || p.edges?.length || p.core?.length))
|
||||
throw new Error("Organization must make durable progress");
|
||||
for (const id of reviewed) this.db.prepare("UPDATE memory_job_objects SET reviewed=1 WHERE job_id=? AND object_id=?").run(snapshot.job_id, id);
|
||||
// Link-only writes still produce a new graph/projection revision. Nothing deletes raw evidence.
|
||||
return revision;
|
||||
}
|
||||
}
|
||||
70
src/object-projection.js
Normal file
70
src/object-projection.js
Normal file
@ -0,0 +1,70 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { closeSync, existsSync, fsyncSync, lstatSync, mkdirSync, openSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
|
||||
import { dirname, join } from "node:path";
|
||||
|
||||
const file = item => `${item.type}/${createHash("sha256").update(item.id).digest("hex")}.md`;
|
||||
const label = value => String(value).replace(/[\r\n\[\]<>]/g, " ");
|
||||
|
||||
/** Human-readable, one dossier per object. Never import edits to generated files. */
|
||||
export class ObjectProjection {
|
||||
constructor(chain, directory) {
|
||||
this.chain = chain; this.directory = directory; this.revision = -1;
|
||||
this.pending = null; this.hashes = new Map();
|
||||
}
|
||||
write(relative, text) {
|
||||
const path = join(this.directory, relative), dir = dirname(path);
|
||||
mkdirSync(dir, { recursive: true, mode: 0o700 });
|
||||
for (const checked of [this.directory, dir, path]) if (existsSync(checked) && lstatSync(checked).isSymbolicLink()) throw new Error("Unsafe projection path");
|
||||
const digest = createHash("sha256").update(text).digest("hex");
|
||||
if (this.hashes.get(relative) === digest && existsSync(path)) return;
|
||||
const temp = `${path}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
writeFileSync(temp, text, { flag: "wx", mode: 0o600, flush: true });
|
||||
renameSync(temp, path);
|
||||
const fd = openSync(dir, "r");
|
||||
try { fsyncSync(fd); } finally { closeSync(fd); }
|
||||
this.hashes.set(relative, digest);
|
||||
} finally { if (existsSync(temp)) unlinkSync(temp); }
|
||||
}
|
||||
refresh(maxObjects = 10) {
|
||||
if (!this.pending && this.revision === this.chain.revision()) return;
|
||||
try {
|
||||
if (!this.pending) this.pending = {
|
||||
revision: this.chain.revision(), offset: 0,
|
||||
rows: this.chain.db.prepare("SELECT id,type,name FROM memory_objects ORDER BY type,name,id").all(),
|
||||
index: ["# 长期记忆对象", "", "SQLite 是权威数据；这些文件是只读投影，重新整理会覆盖文件修改。", "",
|
||||
"每个对象是一份当前有效的概括档案；原始经历留在时间轴数据库，不复制进正文。投影分批生成，各文件注明自己的版本。", ""]
|
||||
};
|
||||
const { rows, index } = this.pending;
|
||||
const lookup = new Map(rows.map(row => [row.id, row]));
|
||||
for (const row of rows.slice(this.pending.offset, this.pending.offset + maxObjects)) {
|
||||
const item = this.chain.graph.read({ id: row.id, limit: 50 });
|
||||
const lines = [`# ${label(item.name)}`, "", `类型：${item.type} · 状态：${item.status} · ID：\`${item.id}\` · 版本：${item.revision}`, "",
|
||||
item.summary, "", "## 当前档案", "", item.body, "", "## 检索入口", "",
|
||||
`别名：${item.aliases.join("、") || "无"}`, "", `未来召回线索：${item.cues.join("；") || "无"}`, "", "## 有向关系", ""];
|
||||
for (const edge of item.edges) {
|
||||
const from = lookup.get(edge.from_id), to = lookup.get(edge.to_id);
|
||||
lines.push(`- [${label(from?.name ?? edge.from_id)}](../${from ? file(from) : "README.md"}) → ${label(edge.relation)} → [${label(to?.name ?? edge.to_id)}](../${to ? file(to) : "README.md"})`);
|
||||
}
|
||||
lines.push("", `## 经历附件（共 ${item.total_sources} 条）`, "");
|
||||
for (const id of item.source_refs) {
|
||||
const event = this.chain.db.prepare("SELECT at,kind FROM timeline WHERE id=?").get(id);
|
||||
lines.push(`- ${event ? new Date(event.at).toISOString() : "时间未知"} · ${event?.kind ?? ""} · ${label(id)}`);
|
||||
}
|
||||
if (item.next_offset !== null) lines.push("", "此文件仅展示前 50 条关系/附件；完整来源可用 memory_read 分页查看。");
|
||||
lines.push("", "按附件 ID 使用 `node src/inspect.js event <ID>` 读取原文；也可按时间查事件。", "");
|
||||
this.write(file(item), lines.join("\n"));
|
||||
index.push(`- ${item.type}：[${label(item.name)}](${file(item)})（${item.status}）— ${label(item.summary)}`);
|
||||
this.pending.offset++;
|
||||
}
|
||||
if (this.pending.offset < rows.length) return;
|
||||
this.write("README.md", index.join("\n") + "\n");
|
||||
this.revision = this.pending.revision;
|
||||
this.pending = null;
|
||||
this.chain.set("object_projection_revision", this.revision);
|
||||
this.chain.set("object_projection_error", "");
|
||||
} catch {
|
||||
this.chain.set("object_projection_error", "Object projection failed; SQLite remains authoritative; refresh will retry");
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -8,6 +8,7 @@ import { ContinuityStore } from "./continuity.js";
|
||||
import { Timeline } from "./timeline.js";
|
||||
import { ChainStore } from "./chain-store.js";
|
||||
import { CoreProjection } from "./core-projection.js";
|
||||
import { ObjectProjection } from "./object-projection.js";
|
||||
|
||||
const stateDir = resolve(process.env.DESKPAW_STATE_DIR ?? join(homedir(), ".deskpaw"));
|
||||
const runtimeDir = join(stateDir, "runtime");
|
||||
@ -39,10 +40,12 @@ const timeline = new Timeline(store.db);
|
||||
const chain = new ChainStore(store.db, timeline);
|
||||
const coreProjection = new CoreProjection(chain, join(stateDir, "super"));
|
||||
coreProjection.refresh();
|
||||
const objectProjection = new ObjectProjection(chain, join(memoryDir, "objects"));
|
||||
objectProjection.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.tick") { objectProjection.refresh(); 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; }
|
||||
@ -50,12 +53,13 @@ function call(method, 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.chain.project_decision") { const result = chain.projectDecision(params); objectProjection.refresh(); return result; }
|
||||
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.commit") { const receipt = chain.commit(params); coreProjection.refresh(); objectProjection.refresh(); return receipt; }
|
||||
if (method === "memory.organize.fail") return chain.fail(params);
|
||||
if (method === "memory.organize.notices") return chain.notices(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);
|
||||
|
||||
@ -13,11 +13,18 @@ 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);
|
||||
if (mode === "organize") {
|
||||
chain.request({ request_id: `detach:${Math.random()}`, mode: "compact", trigger: "threshold" }, 5);
|
||||
const compact = chain.claim({}, 6).job;
|
||||
chain.commit({ job_id: compact.id, token: compact.token, covered_node_ids: compact.nodes.map(n => n.id), summary: "已保存经历，等待归档。" }, 7);
|
||||
}
|
||||
chain.request({ request_id: `test:${Math.random()}`, mode, trigger: mode === "organize" ? "hourly" : "explicit" }, 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 });
|
||||
summary: "讨论报销，尚未提交。详细依据可以展开原始事件。", objects: [], edges: [], core: [], catalog: [],
|
||||
dispositions: job.events.map(e => ({ event_id: e.id, outcome: [...(extra.objects ?? []), ...(extra.core ?? [])].some(o => o.source_refs.includes(e.id)) ? "linked" : "archive_only", reason: "已审阅", progress: "", target_ids: [] })),
|
||||
reviewed_object_ids: job.review_objects.map(o => o.id), ...extra });
|
||||
return { db, timeline, chain, append, claim, proposal };
|
||||
}
|
||||
const person = (id = "person:qhy", refs = ["meeting"]) => ({ id, type: "person", name: "钱灏宇", summary: "用户的同事，与报销工作有关。",
|
||||
@ -28,13 +35,16 @@ test("atomic compaction preserves raw evidence and concurrent suffix; receipt re
|
||||
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);
|
||||
const receipt = f.chain.commit(f.proposal(job), 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);
|
||||
f.chain.request({ request_id: "archive", mode: "organize" }, 51);
|
||||
const archive = f.chain.claim({}, 52).job;
|
||||
f.chain.commit(f.proposal(archive, { objects: [person()], catalog: [{ type: "person", summary: "有同事和报销协作人的相关记忆" }] }), 53);
|
||||
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");
|
||||
@ -43,12 +53,13 @@ test("atomic compaction preserves raw evidence and concurrent suffix; receipt re
|
||||
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();
|
||||
f.append("meeting"); const job = f.claim("organize"), before = f.chain.context();
|
||||
const revision = f.chain.revision();
|
||||
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/);
|
||||
assert.equal(f.chain.revision(), revision);
|
||||
assert.throws(() => f.chain.commit({ ...f.proposal(job), dispositions: [] }, 40), /every claimed event/);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
test("core changes invalidate stale leases and expired leases can be reclaimed", () => {
|
||||
@ -99,9 +110,9 @@ test("threshold is not an every-message model trigger; explicit request is an ev
|
||||
test("projects require user confirmation, graph links are cross-project and paginated", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); const job = f.claim();
|
||||
f.append("meeting"); const job = f.claim("organize");
|
||||
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);
|
||||
f.chain.commit(f.proposal(job, { objects: [person(), project], edges: [{ from_id: "person:qhy", relation: "member_of", 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);
|
||||
@ -114,7 +125,8 @@ test("projects require user confirmation, graph links are cross-project and pagi
|
||||
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();
|
||||
f.append("meeting"); f.append("read", "钱灏宇的历史记忆", "memory_recall"); const job = f.claim("organize");
|
||||
f.chain.inspect({job_id:job.id,token:job.token,kind:"evidence",id:"read"},30);
|
||||
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);
|
||||
@ -133,7 +145,7 @@ test("catalog remains bounded, types can be fully paginated, no implicit graph f
|
||||
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();
|
||||
f.append("meeting"); const job = f.claim("organize");
|
||||
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/);
|
||||
@ -146,7 +158,7 @@ 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);
|
||||
const request = f.chain.request({ request_id: "large", mode: "compact" }, 10);
|
||||
let batches = 0;
|
||||
for (let now = 20; now < 100; now++) {
|
||||
const job = f.chain.claim({ max_bytes: 24000 }, now).job;
|
||||
@ -203,10 +215,13 @@ test("organization survives SQLite reopen and a stale worker cannot overwrite a
|
||||
test("deep inspection grants evidence without letting uninspected objects be replaced", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); let job = f.claim();
|
||||
f.append("meeting"); let job = f.claim("organize");
|
||||
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);
|
||||
f.chain.request({ request_id: "detach-new", mode: "compact", trigger: "threshold" }, 50);
|
||||
const detached=f.chain.claim({},51).job;
|
||||
f.chain.commit(f.proposal(detached),52);
|
||||
f.chain.request({ request_id: "inspect", mode: "organize", scope: "完全不相关词 abcxyz" }, 53);
|
||||
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/);
|
||||
@ -274,7 +289,7 @@ test("opaque Chinese evidence IDs remain expandable and credential metadata is n
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("会议记录/十月十二日", "讨论密钥管理：用途是模型调用，引用 provider/example/api_key，未提供真实值。".repeat(10));
|
||||
const job = f.claim();
|
||||
const job = f.claim("organize");
|
||||
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, ["会议记录/十月十二日"]);
|
||||
@ -297,3 +312,60 @@ test("failed model batches shrink on retry without consuming or dropping their o
|
||||
assert.ok(f.chain.context().nodes.some(node=>node.event_id==="event:11"));
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("classified output retries grow output first, preserve feedback and review coverage, then shrink at capacity", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
for (let i=0;i<12;i++) f.append(`raw:${i}`, "需要长期保留的工作事实".repeat(20));
|
||||
const first=f.claim("compact"), before=f.chain.context().nodes;
|
||||
f.chain.fail({job_id:first.id,token:first.token,kind:"output",error:"output truncated",shrink_batch:false},30);
|
||||
const retry=f.chain.claim({lease_ms:360000},30000).job;
|
||||
assert.deepEqual(retry.nodes.map(n=>n.id),first.nodes.map(n=>n.id));
|
||||
assert.equal(retry.output_failures,1); assert.equal(retry.retry_feedback.error,"output truncated");
|
||||
assert.equal(f.db.prepare("SELECT lease_until FROM memory_organize_jobs WHERE id=?").get(first.id).lease_until,390000);
|
||||
f.chain.fail({job_id:retry.id,token:retry.token,kind:"output",error:"capacity reached",shrink_batch:true},30001);
|
||||
const smaller=f.chain.claim({},60000).job;
|
||||
assert.equal(smaller.nodes.length,4); assert.equal(smaller.output_failures,2);
|
||||
assert.deepEqual(f.chain.context().nodes,before);
|
||||
f.chain.commit(f.proposal(smaller),60001);
|
||||
assert.equal(f.chain.status({id:first.id}).jobs[0].retry,null);
|
||||
assert.equal(f.chain.status({id:first.id}).jobs[0].failures.length,2);
|
||||
assert.equal(f.chain.claim({},60002).job.output_failures,0);
|
||||
} finally {f.db.close();}
|
||||
});
|
||||
|
||||
test("nightly failures are durable, acknowledged once and never silently replayed by idle ticks", () => {
|
||||
const f=fixture();
|
||||
try {
|
||||
f.append("meeting");
|
||||
const now=Date.parse("2026-10-12T18:50:00Z");
|
||||
f.chain.tick({busy:false,last_dialogue_at:now-300001},now);
|
||||
let job;
|
||||
for(let attempt=0;attempt<5;attempt++) {
|
||||
job=f.chain.claim({},now+attempt*400000).job;
|
||||
assert.ok(job);
|
||||
f.chain.fail({job_id:job.id,token:job.token,kind:"schema",error:"objects.0.body: maxLength (limit=16000)",shrink_batch:false},now+attempt*400000+1);
|
||||
}
|
||||
assert.equal(f.chain.notices().jobs.length,1);
|
||||
const rebuilt=new ChainStore(f.db,f.timeline);
|
||||
assert.equal(rebuilt.notices().jobs.length,1);
|
||||
rebuilt.tick({busy:false},now+2100000);
|
||||
assert.equal(rebuilt.claim({},now+2100001).job,null);
|
||||
assert.equal(rebuilt.status({id:job.id}).jobs[0].status,"failed");
|
||||
assert.equal(rebuilt.status({id:job.id}).jobs[0].failures.length,5);
|
||||
rebuilt.notices({ack:job.id});
|
||||
assert.equal(new ChainStore(f.db,f.timeline).notices().jobs.length,0);
|
||||
assert.ok(f.timeline.read({id:"meeting"}).text);
|
||||
} finally {f.db.close();}
|
||||
});
|
||||
|
||||
test("larger object bodies are accepted without weakening evidence or core limits", () => {
|
||||
const f=fixture();
|
||||
try {
|
||||
f.append("meeting"); const job=f.claim("organize");
|
||||
const body="工作知识".repeat(3000);
|
||||
f.chain.commit(f.proposal(job,{objects:[{...person(),body}]}),40);
|
||||
assert.equal(f.chain.graph.read({id:"person:qhy"}).body.length,12000);
|
||||
assert.throws(()=>f.chain.graph.saveCore("USER.md","字".repeat(2401),[],2),/length/);
|
||||
} finally {f.db.close();}
|
||||
});
|
||||
|
||||
@ -20,7 +20,7 @@ test("incremental Computer events interleave with user dialogue, remain task-add
|
||||
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);
|
||||
chain.request({ request_id: "organize", mode: "compact" }, 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);
|
||||
|
||||
273
test/object-organizer.test.js
Normal file
273
test/object-organizer.test.js
Normal file
@ -0,0 +1,273 @@
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { mkdtempSync, readFileSync, rmSync, symlinkSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
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 { ObjectProjection } from "../src/object-projection.js";
|
||||
|
||||
function fixture() {
|
||||
const db = new DatabaseSync(":memory:"), timeline = new Timeline(db), chain = new ChainStore(db, timeline);
|
||||
const append = (id, text = "钱灏宇负责报销材料核对。".repeat(50), at = 123) => timeline.append({ id, kind: "user", at, text, source: {} });
|
||||
const compact = () => {
|
||||
const request = chain.request({ request_id: `compact:${Math.random()}`, mode: "compact" }, 1000);
|
||||
while (chain.status({ id: request.id }).jobs[0].status === "pending") {
|
||||
const job = chain.claim({}, 1100).job;
|
||||
chain.commit({ job_id: job.id, token: job.token, summary: "工作经历已保存，待对象归档。", covered_node_ids: job.nodes.map(n => n.id) }, 1200);
|
||||
}
|
||||
};
|
||||
const claim = (mode = "organize", now = 2000) => {
|
||||
chain.request({ request_id: `organize:${Math.random()}`, mode }, now);
|
||||
return chain.claim({}, now + 1).job;
|
||||
};
|
||||
const proposal = (job, extra = {}) => ({ job_id: job.id, token: job.token, summary: "已审阅", covered_node_ids: [], objects: [], edges: [], core: [], catalog: [],
|
||||
reviewed_object_ids: job.review_objects.map(o => o.id), dispositions: job.events.map(e => ({ event_id: e.id, outcome: e.has_more ? "continue" : "archive_only", progress: e.has_more ? `已审阅至${e.end_offset}` : "", reason: "仅保留经历附件，无需更新对象", target_ids: [] })), ...extra });
|
||||
return { db, timeline, chain, append, compact, claim, proposal };
|
||||
}
|
||||
const person = (id = "person:qhy", source = "meeting") => ({ id, type: "person", name: "钱灏宇", summary: "负责核对报销材料的同事", body: "负责报销核对；其它职责待确认。", aliases: ["灏宇"], cues: ["报销找谁核对"], status: "active", source_refs: [source] });
|
||||
|
||||
test("compaction rejects object writes, enqueues atomically and archive-only retains time retrieval", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting");
|
||||
f.chain.request({ request_id: "one" }, 1000);
|
||||
const job = f.chain.claim({}, 1001).job;
|
||||
const compact = { job_id: job.id, token: job.token, summary: "报销会议", covered_node_ids: job.nodes.map(n => n.id) };
|
||||
assert.throws(() => f.chain.commit({ ...compact, objects: [person()] }, 1002), /Compaction only/);
|
||||
assert.equal(f.chain.organizer.counts().length, 0);
|
||||
f.chain.commit(compact, 1003);
|
||||
assert.equal(f.chain.graph.object("person:qhy"), null);
|
||||
const archive = f.claim();
|
||||
const before = f.chain.context().nodes.map(n => n.id);
|
||||
const receipt = f.chain.commit(f.proposal(archive), 2002);
|
||||
assert.equal(f.chain.organizer.detail("meeting").status, "archive_only");
|
||||
assert.deepEqual(f.chain.context().nodes.map(n => n.id), before);
|
||||
assert.equal(f.timeline.query({ from: 120, to: 125 }).entries[0].id, "meeting");
|
||||
assert.ok(f.chain.node({ id: "meeting" }).text.includes("钱灏宇"));
|
||||
assert.deepEqual(f.chain.commit(f.proposal(archive), 999999), receipt);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("large raw event is queued only after its final hot fragment detaches", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("large", "会议原文".repeat(7000));
|
||||
f.chain.request({ request_id: "all", mode: "compact" }, 1000);
|
||||
let batches = 0;
|
||||
for (;;) {
|
||||
const job = f.chain.claim({ max_bytes: 24000 }, 1001).job;
|
||||
if (!job) break;
|
||||
f.chain.commit({ job_id: job.id, token: job.token, summary: "会议片段索引", covered_node_ids: job.nodes.map(n => n.id) }, 1002);
|
||||
const rawActive = f.db.prepare("SELECT count(*) AS n FROM memory_nodes WHERE event_id='large' AND active=1").get().n;
|
||||
assert.equal(Boolean(f.chain.organizer.detail("large")), !rawActive);
|
||||
batches++;
|
||||
}
|
||||
assert.ok(batches > 1);
|
||||
assert.equal(f.chain.organizer.detail("large").offset, 0);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("hourly check creates no empty work and does not restart on every heartbeat", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
const now = Date.parse("2026-10-12T16:00:00Z");
|
||||
f.chain.tick({}, now);
|
||||
assert.equal(f.chain.status().jobs.length, 0);
|
||||
f.append("meeting"); f.compact();
|
||||
f.chain.tick({}, now + 3599999);
|
||||
assert.equal(f.chain.status().jobs.filter(j => j.trigger === "hourly").length, 0);
|
||||
f.chain.tick({}, now + 3600000);
|
||||
f.chain.tick({}, now + 3600001);
|
||||
assert.equal(f.chain.status().jobs.filter(j => j.trigger === "hourly").length, 1);
|
||||
const job = f.chain.claim({}, now + 3600002).job;
|
||||
assert.equal(job.mode, "organize");
|
||||
assert.equal(job.events[0].id, "meeting");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("event pages and multi-write checkpoints survive retries; final outcome cannot skip unread pages", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting", "会议议题与明确职责".repeat(2000)); f.compact();
|
||||
let job = f.claim();
|
||||
assert.ok(job.events[0].has_more);
|
||||
const linked = { event_id: "meeting", outcome: "linked", reason: "归入人物", progress: "", target_ids: ["person:qhy"] };
|
||||
assert.throws(() => f.chain.commit(f.proposal(job, { objects: [person()], dispositions: [linked] }), 2100), /all event pages/);
|
||||
assert.equal(f.chain.graph.object("person:qhy"), null);
|
||||
const page = job.events[0];
|
||||
f.chain.commit(f.proposal(job, { objects: [person()], dispositions: [{ ...linked, outcome: "continue", progress: "人物职责已建档，继续读后文" }] }), 2101);
|
||||
assert.equal(f.chain.organizer.detail("meeting").offset, page.end_offset);
|
||||
job = f.chain.claim({}, 2200).job;
|
||||
assert.equal(job.events[0].offset, page.end_offset);
|
||||
assert.match(job.events[0].progress, /职责已建档/);
|
||||
f.chain.fail({ job_id: job.id, token: job.token, kind: "transient", error: "temporary" }, 2201);
|
||||
job = f.chain.claim({}, 20000).job;
|
||||
assert.equal(job.events[0].offset, page.end_offset);
|
||||
let now = 20001;
|
||||
while (job.events[0]?.has_more) {
|
||||
f.chain.commit(f.proposal(job), now++);
|
||||
job = f.chain.claim({}, now++).job;
|
||||
}
|
||||
assert.throws(() => f.chain.commit(f.proposal(job), now++), /Archive-only/);
|
||||
f.chain.inspect({ job_id: job.id, token: job.token, kind: "object", id: "person:qhy" }, now++);
|
||||
f.chain.commit(f.proposal(job, { dispositions: [linked] }), now++);
|
||||
assert.equal(f.chain.organizer.detail("meeting").status, "linked");
|
||||
assert.equal(f.chain.graph.read({ id: "person:qhy" }).total_sources, 1);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("existing dossier can accept an attachment without rewriting its facts", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("old"); f.append("meeting");
|
||||
f.chain.graph.apply({ objects: [person("person:qhy", "old")] }, 1, new Map([["old", f.timeline.read({ id: "old" })]]), 10);
|
||||
f.compact(); let job = f.claim();
|
||||
f.chain.commit(f.proposal(job, {dispositions:[{event_id:"old",outcome:"linked",reason:"已有证据关系",progress:"",target_ids:[]}]}), 2002);
|
||||
job = f.chain.claim({}, 2003).job;
|
||||
const original = f.chain.graph.object("person:qhy");
|
||||
f.chain.inspect({ job_id: job.id, token: job.token, kind: "object", id: "person:qhy" }, 2004);
|
||||
f.chain.commit(f.proposal(job, { dispositions: [{ event_id: "meeting", outcome: "linked", reason: "佐证已有职责", progress: "", target_ids: ["person:qhy"] }] }), 2005);
|
||||
assert.equal(f.chain.graph.object("person:qhy").body, original.body);
|
||||
assert.equal(f.chain.graph.read({ id: "person:qhy" }).total_sources, 2);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("terminal failure quarantines only current event, explicit recovery preserves checkpoint", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); f.append("next"); f.compact();
|
||||
let job = f.claim(); const requestId = job.id;
|
||||
for (let attempt = 0; attempt < 5; attempt++) {
|
||||
if (attempt) job = f.chain.claim({}, 2000 + attempt * 400000).job;
|
||||
f.chain.fail({ job_id: job.id, token: job.token, kind: "schema", error: "test schema" }, 2002 + attempt * 400000);
|
||||
}
|
||||
assert.equal(f.chain.organizer.detail("meeting").status, "blocked");
|
||||
assert.equal(f.chain.organizer.detail("next").status, "pending");
|
||||
assert.equal(f.chain.status({ id: requestId }).jobs[0].status, "failed");
|
||||
f.chain.request({ request_id: "unblock", mode: "organize", retry_blocked: true }, 2000000);
|
||||
assert.equal(f.chain.organizer.detail("meeting").status, "pending");
|
||||
assert.equal(f.chain.claim({}, 2000001).job.events[0].id, "meeting");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("threshold failure notice cannot immediately spawn another five-attempt job", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("big", "有效工作".repeat(3000));
|
||||
const now = Date.parse("2026-10-12T16:00:00Z");
|
||||
f.chain.tick({ threshold_tokens: 8000 }, now);
|
||||
for (let i = 0; i < 5; i++) {
|
||||
const job = f.chain.claim({}, now + i * 400000).job;
|
||||
f.chain.fail({ job_id: job.id, token: job.token, kind: "schema", error: "invalid proposal" }, now + i * 400000 + 1);
|
||||
}
|
||||
f.timeline.append({ id: "notice", at: now + 1600002, kind: "memory_organized", text: "整理失败", source: {} });
|
||||
f.chain.tick({ threshold_tokens: 8000 }, now + 1600003);
|
||||
assert.equal(f.chain.claim({}, now + 1600004).job, null);
|
||||
assert.equal(f.chain.status().jobs.length, 1);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("round-robin yields to deep review between compaction batches", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("big", "会议原始经历".repeat(5000));
|
||||
f.chain.request({ request_id: "compact", mode: "compact" }, 1000);
|
||||
f.chain.request({ request_id: "deep", mode: "deep" }, 1001);
|
||||
const first = f.chain.claim({}, 1010).job;
|
||||
assert.equal(first.mode, "compact");
|
||||
f.chain.commit({ job_id: first.id, token: first.token, summary: "索引", covered_node_ids: first.nodes.map(n => n.id) }, 1011);
|
||||
assert.equal(f.chain.claim({}, 1012).job.mode, "deep");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("migration backfills detached events once; completed archive-only is never requeued", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting"); f.compact();
|
||||
f.db.exec("DELETE FROM memory_event_organization; DELETE FROM memory_chain_meta WHERE key='object_pipeline_version'");
|
||||
const rebuilt = new ChainStore(f.db, f.timeline);
|
||||
assert.equal(rebuilt.organizer.detail("meeting").status, "pending");
|
||||
const job = f.claim(); f.chain.commit(f.proposal(job), 2002);
|
||||
assert.equal(new ChainStore(f.db, f.timeline).organizer.detail("meeting").status, "archive_only");
|
||||
assert.equal(f.db.prepare("PRAGMA integrity_check").get().integrity_check, "ok");
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("per-object Markdown projection is readable, links evidence, rejects symlinks and reports failure", () => {
|
||||
const f = fixture(), dir = mkdtempSync(join(tmpdir(), "object-projection-"));
|
||||
try {
|
||||
f.append("meeting"); f.compact();
|
||||
const job = f.claim();
|
||||
f.chain.commit(f.proposal(job, { objects: [person()], dispositions: [{ event_id: "meeting", outcome: "linked", reason: "人物职责", progress: "", target_ids: [] }] }), 2002);
|
||||
const projection = new ObjectProjection(f.chain, join(dir, "objects")); projection.refresh();
|
||||
const index = readFileSync(join(dir, "objects", "README.md"), "utf8");
|
||||
const path = index.match(/\[钱灏宇\]\(([^)]+)\)/)[1];
|
||||
const body = readFileSync(join(dir, "objects", path), "utf8");
|
||||
assert.match(body, /负责报销核对/); assert.match(body, /meeting/);
|
||||
assert.equal(f.chain.meta("object_projection_error"), "");
|
||||
symlinkSync(dir, join(dir, "unsafe"));
|
||||
new ObjectProjection(f.chain, join(dir, "unsafe")).refresh();
|
||||
assert.ok(f.chain.meta("object_projection_error"));
|
||||
assert.equal(f.chain.graph.object("person:qhy").name, "钱灏宇");
|
||||
} finally { f.db.close(); rmSync(dir, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
test("node read honors range; long dossier tail remains searchable; graph relations are typed", () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.append("meeting");
|
||||
const item = { ...person(), body: "知".repeat(15980) + "罕见工作尾词锆铪" };
|
||||
f.chain.graph.apply({ objects: [item] }, 1, new Map([["meeting", f.timeline.read({ id: "meeting" })]]), 1);
|
||||
assert.equal(f.chain.graph.find({ query: "锆铪" }).hits[0].id, item.id);
|
||||
f.chain.sync(); const node = f.chain.context().nodes[0];
|
||||
assert.equal(f.chain.node({ id: node.id, offset: 2, limit: 4 }).text, node.text.slice(2, 6));
|
||||
assert.throws(() => f.chain.graph.apply({ edges: [{ from_id: item.id, to_id: "project:daily", relation: "一大段叙述", source_refs: ["meeting"] }] }, 2, new Map(), 2), /supported directed/);
|
||||
} finally { f.db.close(); }
|
||||
});
|
||||
|
||||
test("object checkpoint survives database reopen, expired token and repeated commit", () => {
|
||||
const dir=mkdtempSync(join(tmpdir(),"object-checkpoint-")), path=join(dir,"memory.sqlite");
|
||||
let db=new DatabaseSync(path);
|
||||
try {
|
||||
let timeline=new Timeline(db), chain=new ChainStore(db,timeline);
|
||||
timeline.append({id:"meeting",kind:"user",at:1,text:"钱灏宇参与报销。".repeat(50),source:{}});
|
||||
chain.request({request_id:"compact",mode:"compact",trigger:"threshold"},10);
|
||||
const compact=chain.claim({},11).job;
|
||||
chain.commit({job_id:compact.id,token:compact.token,summary:"报销会议",covered_node_ids:compact.nodes.map(n=>n.id)},12);
|
||||
chain.request({request_id:"organize",mode:"organize"},20);
|
||||
let job=chain.claim({},21).job;
|
||||
const patch={job_id:job.id,token:job.token,summary:"已建人物，待关联项目",covered_node_ids:[],objects:[person()],
|
||||
dispositions:[{event_id:"meeting",outcome:"continue",reason:"还需检查项目",progress:"人物档案已建，下一批检查项目关系",target_ids:[]}],reviewed_object_ids:[]};
|
||||
const receipt=chain.commit(patch,22);
|
||||
db.close(); db=new DatabaseSync(path); timeline=new Timeline(db); chain=new ChainStore(db,timeline);
|
||||
assert.deepEqual(chain.commit(patch,999999),receipt);
|
||||
job=chain.claim({},30).job;
|
||||
assert.equal(job.events[0].text,"");
|
||||
assert.match(job.events[0].progress,/人物档案已建/);
|
||||
const final={job_id:job.id,token:job.token,summary:"归档完成",covered_node_ids:[],dispositions:[{event_id:"meeting",outcome:"linked",reason:"已有档案，无需新项目",progress:"",target_ids:[]}],reviewed_object_ids:[]};
|
||||
assert.throws(()=>chain.commit(final,200000),/Expired/);
|
||||
const next=chain.claim({},200001).job;
|
||||
chain.commit({...final,token:next.token},200002);
|
||||
assert.equal(chain.organizer.detail("meeting").status,"linked");
|
||||
assert.equal(chain.graph.object("person:qhy").body,person().body);
|
||||
} finally {db.close();rmSync(dir,{recursive:true,force:true});}
|
||||
});
|
||||
|
||||
test("all-blocked work is partial, and a replayed recovery request cannot re-unblock it", () => {
|
||||
const f=fixture();
|
||||
try {
|
||||
f.append("meeting"); f.compact();
|
||||
f.db.exec("UPDATE memory_event_organization SET status='blocked'");
|
||||
const req=f.chain.request({request_id:"check",mode:"organize"},2000);
|
||||
assert.equal(f.chain.claim({},2001).job,null);
|
||||
assert.equal(f.chain.status({id:req.id}).jobs[0].status,"partial");
|
||||
const retry={request_id:"retry",mode:"organize",retry_blocked:true};
|
||||
f.chain.request(retry,3000);
|
||||
assert.equal(f.chain.organizer.detail("meeting").status,"pending");
|
||||
f.db.exec("UPDATE memory_event_organization SET status='blocked'");
|
||||
f.chain.request(retry,4000);
|
||||
assert.equal(f.chain.organizer.detail("meeting").status,"blocked");
|
||||
} finally {f.db.close();}
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user