From c373d21e1f011e6b9c68155dcd9274e96eef10ae Mon Sep 17 00:00:00 2001 From: 18962558917 <18962558917@189.cn> Date: Thu, 15 Oct 2026 02:03:15 +0800 Subject: [PATCH] feat: separate timeline compaction from durable object organization --- OBJECT_MEMORY.md | 147 ++++++++++++++++++ README.md | 9 +- src/chain-graph.js | 9 +- src/chain-store.js | 159 ++++++++++++++------ src/inspect.js | 40 +++++ src/object-organizer.js | 154 +++++++++++++++++++ src/object-projection.js | 70 +++++++++ src/server.js | 10 +- test/chain.test.js | 100 +++++++++++-- test/computer-events.test.js | 2 +- test/object-organizer.test.js | 273 ++++++++++++++++++++++++++++++++++ 11 files changed, 904 insertions(+), 69 deletions(-) create mode 100644 OBJECT_MEMORY.md create mode 100644 src/inspect.js create mode 100644 src/object-organizer.js create mode 100644 src/object-projection.js create mode 100644 test/object-organizer.test.js diff --git a/OBJECT_MEMORY.md b/OBJECT_MEMORY.md new file mode 100644 index 0000000..f2d3d05 --- /dev/null +++ b/OBJECT_MEMORY.md @@ -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 尚未生成；不能把代码支持三份文件描述成三份已经有完整内容。 diff --git a/README.md b/README.md index c0c821c..dc122f8 100644 --- a/README.md +++ b/README.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` 表示深度任务的对象总量和已成功提交的审阅量，不证明模型语义判断必然正确。 diff --git a/src/chain-graph.js b/src/chain-graph.js index b95ddae..c769929 100644 --- a/src/chain-graph.js +++ b/src/chain-graph.js @@ -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); diff --git a/src/chain-store.js b/src/chain-store.js index 245b9f2..7ced8ac 100644 --- a/src/chain-store.js +++ b/src/chain-store.js @@ -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(() => { diff --git a/src/inspect.js b/src/inspect.js new file mode 100644 index 0000000..d660053 --- /dev/null +++ b/src/inspect.js @@ -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)); diff --git a/src/object-organizer.js b/src/object-organizer.js new file mode 100644 index 0000000..f16642e --- /dev/null +++ b/src/object-organizer.js @@ -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; + } +} diff --git a/src/object-projection.js b/src/object-projection.js new file mode 100644 index 0000000..5d5a133 --- /dev/null +++ b/src/object-projection.js @@ -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 ` 读取原文；也可按时间查事件。", ""); + 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"); + } + } +} diff --git a/src/server.js b/src/server.js index aa0a7dd..d6a5580 100644 --- a/src/server.js +++ b/src/server.js @@ -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); diff --git a/test/chain.test.js b/test/chain.test.js index 8425952..a7ba130 100644 --- a/test/chain.test.js +++ b/test/chain.test.js @@ -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();} +}); diff --git a/test/computer-events.test.js b/test/computer-events.test.js index 1a0ddf5..c433d1b 100644 --- a/test/computer-events.test.js +++ b/test/computer-events.test.js @@ -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); diff --git a/test/object-organizer.test.js b/test/object-organizer.test.js new file mode 100644 index 0000000..494f7d1 --- /dev/null +++ b/test/object-organizer.test.js @@ -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();} +});