From cd386329b30c06b2cfdf56e3d2215cf2fa88e625 Mon Sep 17 00:00:00 2001 From: 18962558917 <18962558917@189.cn> Date: Sun, 11 Oct 2026 22:06:28 +0800 Subject: [PATCH] feat: add durable continuity and indexed experience timeline --- .gitignore | 10 ++ README.md | 43 +++++-- THIRD_PARTY_NOTICES.md | 28 +++++ bin/run-memory | 7 ++ docs/architecture.md | 150 ++++++++++++++++++++++ docs/implementation-v1.md | 136 ++++++++++++++++++++ src/continuity.js | 238 +++++++++++++++++++++++++++++++++++ src/policy.js | 9 +- src/rank-fusion.js | 15 +++ src/server.js | 49 +++++++- src/store.js | 12 +- src/timeline.js | 59 +++++++++ test/continuity.test.js | 128 +++++++++++++++++++ test/protocol.test.js | 71 +++++++++++ test/server-timeline.test.js | 31 +++++ test/timeline.test.js | 29 +++++ test/topics.test.js | 3 +- 17 files changed, 996 insertions(+), 22 deletions(-) create mode 100644 THIRD_PARTY_NOTICES.md create mode 100644 docs/architecture.md create mode 100644 docs/implementation-v1.md create mode 100644 src/continuity.js create mode 100644 src/rank-fusion.js create mode 100644 src/timeline.js create mode 100644 test/continuity.test.js create mode 100644 test/protocol.test.js create mode 100644 test/server-timeline.test.js create mode 100644 test/timeline.test.js diff --git a/.gitignore b/.gitignore index 0424f2d..baee03e 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,13 @@ *.sqlite-shm *.sqlite-wal runtime/ + +# Local reference clones, dependencies and private runtime configuration +/other project/ +node_modules/ +.env +.env.* +*.sqlite3 +*.sqlite3-shm +*.sqlite3-wal +*.log diff --git a/README.md b/README.md index 6748dc8..2edf0b0 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,19 @@ # deskpaw-memory +## Super Pi v2 时间轴 + +`feat/super-agent-v2` 新增独立经历时间轴，使用 `timeline` 与 `timeline_terms` 表，不查询下文的旧任务/话题索引，不使用 embedding。Super v2 不再启动下层夜间旧记忆整理；以下原有接口保留供旧调用方和数据审计使用。 + +新 JSON-line 接口为 `timeline.append/query/read/expand`。append 必须包含稳定 `id`、`kind`、毫秒时间 `at` 和不超过 32,000 字符的 `text`，`source` 仅接收来源/任务/文件引用白名单字段。敏感文本经过过滤，重复 ID 不覆盖原始记录。长工具结果保留在 Agent 作业表，时间轴只保存有限摘要和引用。 + +query 支持 `query/from/to/after/limit`：中英文词项倒排检索、时间范围、最多 50 个索引头，每头最多 1,200 字符；命中数和时间排序。无查询词时按 seq 正序分页。read/expand 接受 `id/offset/limit` 返回原文范围与 next_offset。没有用户分区。 + +预留 parent_id、covered_from、covered_to、version，尚未实现自动会议归并、记忆图提取、压缩稀疏化、删除/遗忘、磁盘容量策略。原始记录会增长，必须监控磁盘。SQLite 是唯一权威来源。Agent 的持久待同步表保证服务离线后可补写；不阻塞上层聊天。 + +服务启动会检查已有 socket 所有者，不删除活跃实例的 socket；单请求连接限制 1 MB、5 秒和最多 64 并发连接。测试执行 `npm test`，新增时间轴存储、敏感过滤和真实 Unix socket 的隔离测试，不操作生产状态。 + +## 以下为 v1 存量机制 + DeskPaw 的本地记忆与上下文整理模块。它作为 `deskpaw-agent` 的平级服务运行，通过本地 socket 提供： - 任务卡和执行结果的持久化； @@ -8,6 +22,7 @@ DeskPaw 的本地记忆与上下文整理模块。它作为 `deskpaw-agent` 的 - 每轮执行前的有限上下文包； - 按 03:30 工作日组织的话题经历、长期事实和技能草稿； - 主动整理记录与可观测事件。 +- 版本化事件接入、可恢复观察批次、带证据的经历与中文倒排索引。 模块不保存隐藏思考，不接管日程执行，不直接操作电脑或发送消息。Agent 不依赖本服务才能运行；服务离线时自动降级。 @@ -40,8 +55,11 @@ DESKPAW_MEMORY_BUNDLES=/path/to/bundles ## 顶层记忆 -DeskPaw 使用一个 Session。顶层 Agent 每轮先同步已完成的问答，再调用 `memory.recall`； -Memory 返回相关与最近的问答、任务卡以及资料包，供顶层优先回答“之前的任务”“刚才怎么样”等问题。 +DeskPaw 使用一个 Session。`feat/supervisor-memory-v1` 将问答和任务结果同步移至持久事件驱动的后台队列， +上层不再先等写入完成。每次召回有持久请求 ID，本地检索最多占用上层 75 毫秒的等待预算； +未完成返回 `pending`，后台继续，不能解释为查无记忆。完成结果缓存并进入上层只读收件箱。 +核心画像保留小型缓存；实时任务状态始终从 Runtime 读取。 +Memory 返回相关问答、任务卡、资料包和新经历，供顶层回答历史问题。 来源仅作为 `source` 元数据保存，不作为检索过滤条件。同步按 `interaction_id` 幂等写入， 并经 `safeMemoryText` 过滤敏感文本；不要在记忆资料中放入凭据或隐藏推理。 @@ -64,14 +82,16 @@ Memory 返回相关与最近的问答、任务卡以及资料包，供顶层优 记忆中的 `topic_id` 是话题单元，既可表示执行任务，也可表示闲聊和讨论；它与 Agent 执行队列的 `task_id` 不同。一个话题可跨工作日继续。工作日固定为北京时间 03:30 至次日 03:29:59， 与实际压缩作业是否延期无关。SQLite 中的 `topic_interactions` 只建立关联，不改写原始交互； -`topic_segments` 保存每日极简概括。最近三个有话题片段的工作日进入顶层概览，旧话题仍可检索。 +`topic_segments` 保存每日极简概括。旧接口仍返回最近三个有话题片段的工作日； +新上层不再固定注入三天全部概览，按问题相关性选择内容，旧话题仍可检索。 原始执行日志仍由 Agent 运行时 SQLite 保存；Agent 按话题读取详情时，利用关联的执行任务 ID 补充有限的执行日志，不在 Memory 中复制无界工具输出。 -下层 Agent 在无执行任务时分批调用 `memory.maintenance.plan`，使用执行模型归类话题、概括结果与问题， +后台 Worker 在空闲时分批调用 `memory.maintenance.plan`，使用记忆角色模型归类话题、概括结果与问题， 然后用 `memory.maintenance.commit` 原子写入。每条交互必须恰好分配一次；重启中断不会删除原始记录。 一小时无用户或执行活动后会整理未归档交互；跨过 03:30 时，在下层空闲后复核前一工作日。 -整理期间上层 Supervisor 仍可交流，新执行任务排队等待。Memory 服务本身不调用模型或 Executor。 +自动整理不再占用 Pi 的执行队列，也不清空正在运行的 Pi 上下文。 +旧的已排队 `memory_maintenance` 项保留兼容处理。Memory 服务本身不调用模型或 Executor。 长期事实必须附可逐字核对的证据，但不要求每批对话都提炼事实。`profile` 只收来自用户原话的 稳定身份、交流偏好和职业工作方式，并须标明 `profile_kind`；当前推进的工作用 `project`， @@ -84,8 +104,17 @@ Memory 返回相关与最近的问答、任务卡以及资料包，供顶层优 `set_status` 为 `active`；执行层可以读取草稿用于审阅，但只有已启用技能可作为执行参考。 不要把凭据、隐藏思考或无界工具输出写入记忆。 -服务故障不会暂停执行队列。顶层会显示实际的记忆请求错误；恢复服务后， -后续交互会继续同步最近的已完成问答。 +服务故障不会暂停执行队列。上层区分待查询、可用和不可用，后台按持久事件游标重试同步。 + +## 连续记忆 v1 + +完整设计见 [架构](docs/architecture.md)，已实现范围、协议和验证方法见 +[第一阶段说明](docs/implementation-v1.md)。 + +新接口：`memory.events.ingest`、`memory.observation.claim/commit/fail`、 +`memory.continuity.recall`、`memory.evidence.get`。它们与旧话题/事实接口并存，不删除旧记忆。 +新经历通过中文/英文倒排索引召回，不受旧任务卡的最近 100 条窗口限制。 +事件所属设备由 `owner` 限定；当前部署仍是受信本地 socket、单设备统一会话，不是多租户权限系统。 ## 资料夹 diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md new file mode 100644 index 0000000..db420b6 --- /dev/null +++ b/THIRD_PARTY_NOTICES.md @@ -0,0 +1,28 @@ +# Third-party notices + +`src/rank-fusion.js` adapts `reciprocal_rank_fusion` from +[T-Mem](https://github.com/Sherlockwz/T-Mem), revision `dd9e152`, +`T_mem/main/stage6_retrieval.py`. Changes: JavaScript implementation, stable ID +tie-breaking, duplicate suppression within each retrieval channel. + +MIT License + +Copyright (c) 2026 Weidong Guo, Dakai Wang, Zixuan Wang, Hui Liu, Yu Xu (Tencent) + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/bin/run-memory b/bin/run-memory index 9384534..156c5eb 100755 --- a/bin/run-memory +++ b/bin/run-memory @@ -8,6 +8,13 @@ if [[ -f "$device_config" ]]; then source "$device_config" set +a fi +config_dir="${DESKPAW_CONFIG_DIR:-$(dirname "$device_config")}" +memory_config="$config_dir/memory.env" +if [[ -f "$memory_config" ]]; then + set -a + source "$memory_config" + set +a +fi node_bin="${DESKPAW_NODE_BIN:-$(command -v node || true)}" if [[ -z "$node_bin" || ! -x "$node_bin" ]]; then echo "DeskPaw memory cannot find Node.js" >&2 diff --git a/docs/architecture.md b/docs/architecture.md new file mode 100644 index 0000000..1d997ec --- /dev/null +++ b/docs/architecture.md @@ -0,0 +1,150 @@ +# DeskPaw 连续智能体与记忆架构 + +DeskPaw 对用户提供一个持续存在的智能体身份。上层负责理解、承诺、决策和调度；Pi 等执行者负责受限任务；记忆系统维护可追溯的经历、事实和约束，并在相关时提供给上层。会话身份持久存在，不要求任何一个模型的上下文无限增长。 + +## 改造前的实现 + +上层目前通过一次 `completeSimple` 调用产生包含 `action`、`reply`、`objective` 等字段的 JSON。运行时从文本的第一对外层花括号之间截取内容，解析并路由到九种固定动作。对话回复与控制指令耦合，回复又被截断到 2000 字符。它没有独立工具循环，也不直接共享 Pi 的全部消息列表。 + +Pi 使用持久化 JSONL 会话文件；Runtime SQLite 保存交互、执行队列、日程和事件。Memory SQLite 保存任务卡、话题、事实和资料夹，Markdown 是投影。模型输出 JSON、JSONL 审计文件、JSON Lines 本地通信是三个不同问题，不能通过替换文件格式来解决语义和调度耦合。 + +当前上层先等待问答同步和记忆召回，然后调用模型。较完整的整理由执行队列中的 `memory_maintenance` 完成，依赖下层空闲。任务完成、阻塞等事件主要进入执行日志和展示渠道，尚未形成有版本校验的上层事件收件箱。自然语言暂停还会等待 Pi 停止；直接 `/v1/stop` 则已经能够先确认请求。 + +## 架构原则 + +1. 上层是记忆的主要消费者和任务意图的管理者。后台整理者不能改变用户目标、授予操作权限或自行恢复暂停任务。 +2. Runtime 是任务状态的权威，Executor 是动作回执的来源，Memory 是记忆记录的权威。检索索引和 Markdown 都可重建，不形成第二份可独立修改的事实库。 +3. 上层不等待电脑任务或后台整理完成。不能安全继续的任务可以等待证据，整个上层仍可接收新消息和控制请求。 +4. 原始观察、执行者报告、用户陈述、验证结果和模型推断分开记录。动作成功不等于业务完成。 +5. 所有跨异步边界的结果带来源、对象、请求标识和版本。迟到结果可以归档，不能覆盖更新的约束或作用于另一任务。 +6. 至少一次投递与幂等提交构成可靠性基础，不假定跨数据库事务或端到端恰好一次。 + +## 组件与责任 + +```text +用户和消息入口 → 上层收件箱 → Supervisor → 经校验的控制命令 + ↑ ↓ + 事件筛选与回调策略 Runtime 与日程调度 + ↑ ↓ + 记忆召回结果 Pi 等执行者 → Executor + ↑ ↓ + 上层上下文投影 ← Memory ← 有界事件与证据引用 + ↑ + 后台观察和整理 Worker +``` + +Supervisor 只持有控制和只读查询能力。发送任务后立即获得持久化任务回执。对用户的文字与工具参数分离；JSON 继续作为通信和模式校验格式，但不再要求每次聊天都是一个控制 JSON 对象。 + +Runtime 管理队列、取消、任务版本和执行资源。一个物理桌面同一时刻只能有一个有副作用的执行所有者。未来研究、文件解析和记忆 Worker 可以并行，但不能绕过桌面资源所有权。已发出的原子操作无法保证撤回；不确定结果应暂停后续动作并核验。 + +Memory 分为接入、状态投影、观察提取、合并复核、检索和上下文组装。模型只生成受校验的记忆提案，程序负责事务、来源校验、生命周期、索引和预算。 + +这些是逻辑边界。首阶段后台 Worker 复用 Agent 的模型适配层，但拥有独立队列，不占 Pi 执行队列；Memory 仍通过本地协议独立运行。后续可将模型 Worker 和 Pi Runner 迁到独立进程。首阶段不宣称提供进程级实时隔离。 + +## 对象与记忆模型 + +对象标识与会话标识分离。对象包括长期责任领域、有限项目、持续维护资产、可复用流程、周期计划和一次执行。项目休眠只表示当前不活跃，不表示其记忆失效。 + +报销领域包含报销规则和资料入口；每月报销是独立项目；每日整理和每月提交是不同计划；每次操作是执行实例。软件资产保留仓库、部署和故障经历，某次修复则是关联该资产的独立任务。没有下一次计划时间的资产也可以在事件或问题命中时恢复上下文。 + +事件可以关联多个对象，保留一个主要归属以支持低成本路由。多设备、多用户扩展必须增加明确的所有者和授权范围；当前统一会话不能被解释为可以跨用户共享全部记忆。 + +记忆分为以下类型： + +- 当前状态：当前讨论、活跃任务、未兑现承诺、阻塞和等待确认。任务状态由程序投影。 +- 经历：目标、关键操作、结果、失败方法、剩余事项及证据。经历按对象和事件段组织，工作日只是浏览维度。 +- 事实和偏好：带范围、来源、有效时间与状态的原子陈述。 +- 规则和纠正：注明约束对象、何时生效、是否取代旧规则。一次例外不自动成为全局规则。 +- 操作经验：适用条件、步骤、失败条件和验证方法。候选经验不自动成为已批准技能。 +- 召回线索：说明未来何种问题可能需要这段记忆。线索不是事实、命令或日程。 + +记录至少包含 `id`、`owner_scope`、`subject_id`、`kind`、`content`、`source_refs`、`observed_at`、`recorded_at`、`status`、`revision`。稳定条目再提供有效时间、复核时间和 `supersedes`。对象版本按对象递增，避免一个无关项目更新导致全库写入冲突。 + +## 记忆形成 + +接入用户消息、上层控制决定和承诺、Pi 阶段报告、Executor 有意义的观察与产物、Runtime 和日程事件。外部网页和屏幕内容只作为数据，不提升为用户指令。 + +普通鼠标动作和重复成功回执留在审计层。程序合并重复观察和同一错误，保留次数和时间范围；关键变化、错误、用户纠正、产物和终态进入记忆事件。图片、大段 UIA 和模型供应商原始响应不复制进记忆。来源引用不等于授权读取任意文件，读取证据仍需范围校验与脱敏。 + +事件先持久化，再异步发送。接收位置、已整理位置和已发布版本分开维护。高优先级批次可以先处理，但不能越过尚未处理的事件而使其永久丢失。以事件序号和 ID 去重，不以毫秒时间作为唯一游标。 + +状态通道立即更新确定性的任务变化。用户最新原话进入上层上下文，不等待语义整理。自然语言理解本身仍需模型；明确停止按钮和控制接口走快速通道。 + +语义通道按任务或对象构造有界工作包。首版普通事件按约 30 秒、20 条有效事件或输入预算触发；阻塞、结束和明确纠正优先。Worker 无新数据不调用模型。初始并发为 1，使用已配置文本模型，模型名称和超时独立可配。 + +观察者提取经历与稀疏候选条目；合并者在任务边界或对象记录过多时合并。每个候选都要能回到原始来源。缺少来源、扩大范围、试图覆盖较新纠正或包含敏感信息的提案不能生效。失败保留待处理事件并退避，成功提交后才消费来源。 + +用户明确陈述与模型推断保留不同标签。数次复述同一来源不增加独立证据数量。环境状态需要时重新检查，旧故障不能成为永远成立的操作前提。压缩后的文本不替代原始证据；重要条目仍能按引用展开。 + +## 记忆管理 + +候选、有效、被取代、过期和归档是记忆生命周期；活跃、休眠、完成是对象生命周期，两者独立。 + +纠正以新增版本和关联旧版本实现。最新规则优先的前提是同一对象、同一适用范围和相应有效时间；不能用简单的全库最后写入胜出。冲突无法消除时保留冲突标记，不暗中选择一个结果。 + +暂态观察按类型设置复核和保留策略。未完成承诺不能仅因长时间未访问被遗忘。用户要求删除时，应同时失效相关索引、上下文缓存和派生记录，按保留策略清理内容；审计机制不能成为拒绝删除的理由。 + +数据库与证据保留量可以增长，上层上下文必须有界。核心画像只保留少量稳定条目，休眠对象通过索引找回，不把全对象目录长期放入提示词。SQLite 负责权威记录；Markdown 用于查看与导出，外部编辑通过显式导入和版本校验。 + +## 召回与结果回调 + +召回是检索历史知识，回调是异步结果返回到接收者，两者用不同协议表示。 + +一次召回请求包含请求 ID、所有者范围、问题、对象或任务引用、用途、预算、发起意图版本和有效期。结果包含命中条目、证据引用、记忆版本、检索原因、缺口和完成状态。查不到、暂时不可用和记录已过期必须区分。 + +检索顺序为：权限和有效性过滤；确定对象及强制适用约束；关键词或结构化 ID 召回；必要时加入语义及关联线索召回；合并排序；展开少量证据；在预算内组装。强制约束不依赖相似度排名。未来混合通道可以使用 RRF 融合，向量索引可替换，不影响权威数据。 + +上层使用已发布快照加最新 Runtime 变化。缓存未命中时发起异步检索，不能因此把未知解释为不存在。涉及不可逆动作而依据不足时，仅让该任务等待上下文，不阻塞其他对话。 + +| 时机 | 召回内容 | 上层处理 | +| --- | --- | --- | +| 新用户消息或话题切换 | 当前对象、相关规则和经历 | 读取缓存，必要时异步补查 | +| 派发任务前 | 约束、验收条件、相关错误经验 | 形成带版本的任务工作包 | +| 下层遇到阻塞或重复失败 | 同对象历史原因和修复经验 | 合并事件，重新判断或请求用户输入 | +| 日程到期 | 对应计划、周期项目和未完成事项 | Runtime 触发，必要时让上层判断变化 | +| 重要观察与休眠资产相关 | 相关资产摘要和证据 | 后台匹配后按必要性唤醒 | +| 用户纠正或相关记忆更新 | 新约束及被取代版本 | 失效旧缓存，必要时改向正在执行的任务 | +| 普通点击或无变化观察 | 通常无 | 不唤醒上层模型 | + +回调进入持久化上层收件箱，包含因果事件 ID、任务 ID、任务版本、请求 ID和优先级。用户消息优先，重复进度合并。回调过期或目标已变化时归档，不对新任务应用旧结果。模型生成控制命令后，执行前还要检查目标与版本，避免给任务 A 的修改落到刚接替的任务 B。 + +控制回执区分已接收、已开始、正在停止和已停止。上层不能在命令未持久化前声称已安排，也不能在原子操作结束前声称已经停止。后台记忆更新通常只替换快照，不每次触发聊天或继续执行。 + +## 三个项目的复用边界 + +源码基线为 T-Mem `dd9e152`、Letta Code `253a3bc8` 和 Mastra `e832eb4870`。 + +- T-Mem：复用原子条目到经历的证据关联、写入时形成召回线索以及 `stage6_retrieval.py` 的 RRF 算法。Python 数据集脚本和默认英文分词不作为生产接入层；其倾向完整提取的提示词需要改成有预算的稀疏筛选。触发线索不进入事实证据路径。 +- Letta Code：复用常驻与按需加载的分层、按引用交接以及后台修改成功落地后才消费历史的规则。`memory-format.ts` 的路径判断可用于 Markdown 投影，`memory-handoff.ts` 的交接协议需换成我们的来源接口。当前本地初始化使用 MemFS v2，不直接沿用旧 `system/` 布局；按英文字符估 token 的方法不适合作为中文硬预算。 +- Mastra：复用 `anchor-ids.ts`、观察分组与来源维护等可拆出的纯函数，以及安全批次边界、后台准备后激活和异步问答协议。`ObservationalMemory` 也提供独立 API，但直接采用仍需适配其消息与存储模型。首阶段摘取需要的模块和回归案例，不迁移整个 Agent 框架。部分激活路径可能等待后台作业，不能置于上层响应路径；提醒检索的英文正则必须补中文处理。 + +移植代码记录上游版本和对应许可证，保留要求的声明。T-Mem 为 MIT；所选 Letta Code 和 Mastra 非企业模块为 Apache 2.0。Mastra 的 `ee` 内容不纳入本次复用。 + +## 第一阶段实现边界 + +第一阶段实现事件接入、后台整理、来源可追溯的经历、基本索引召回、有界上层上下文与异步控制和结果收件箱。先沿用现有模型适配层与 SQLite，本地 JSON Lines 协议继续使用。 + +实施顺序： + +1. 增加版本化事件协议、幂等接入和可恢复批次；保留现有原始记录。 +2. 增加独立于 Pi 队列的观察 Worker，模型只处理有界事件并提交受校验提案。 +3. 组装上层缓存和异步召回结果，当前任务状态始终覆盖旧快照。 +4. 分离模型正文与原生控制工具，暂停先返回控制回执，补充任务目标校验。 +5. 接通关键执行事件的上层收件箱，限制自动回调的权限和重试，避免自动执行循环。 +6. 通过离线模型、隔离数据库和假 Executor 验证，不操作真实桌面。 + +首阶段不建设完整财务台账、发票解析、向量基础设施、多设备调度或自主技能发布。全量项目管理、语义索引和进程级隔离通过现有协议继续扩展。正式环境启用须在测试后单独部署；源码修改与部署是不同操作。 + +该阶段已在 `feat/supervisor-memory-v1` 分支实施一版。具体参数、协议、实测范围及尚未实现的边界见 +[第一阶段说明](implementation-v1.md)。尤其要区分只读结果收件箱与自动续做：本阶段实现前者，不让回调擅自创建操作。 + +## 验收 + +- 上层聊天和暂停控制不等待记忆模型或 Pi 原子动作结束。 +- 任务完成、阻塞和记忆召回结果有可恢复、幂等的回调记录。 +- 迟到结果不覆盖更新约束，不对不同任务实施旧命令。 +- 重启或模型错误不会使未完成整理的事件永久丢失。 +- 普通操作洪流经过筛选，模型批次和上层上下文有明确大小上限。 +- 中文报销、跨月份经历和休眠软件资产可以按相关对象或关键词找回。 +- 工具成功与业务验证结果分开，网页指令、密码、隐藏推理不提升为长期规则。 +- 已有任务、日程、技能和会话身份保持可用。测试记录明确哪些属于单元、协议、集成和实际硬件验证。 diff --git a/docs/implementation-v1.md b/docs/implementation-v1.md new file mode 100644 index 0000000..58d7704 --- /dev/null +++ b/docs/implementation-v1.md @@ -0,0 +1,136 @@ +# 连续记忆第一阶段：实现与验证 + +状态：2026-10-09，源码分支 `feat/supervisor-memory-v1`。涉及 `deskpaw_agent` 和 +`deskpaw-memory`；尚未替换或重启正式服务。本文描述已实现边界，不把架构目标当作已有能力。 + +## 执行路径 + +```text +Runtime 持久事件 + ├─ 上层收件箱：终态、阻塞、需要输入 → 下次决策上下文（只读） + └─ MemoryBridge 后台同步 → Memory 事件表 → 观察批次租约 + → 模型提案 → 校验并原子发布 +用户消息 → 限时索引召回／缓存 → 上层自然语言＋控制工具 → Runtime 校验与执行 + └─ 迟到结果持久化 → 上层收件箱／按请求 ID 查询 +``` + +上层是主要消费者。Pi 在派发后的执行准备阶段也能读取有限工作包，并通过 +`memory_evidence_get` 按 `source_refs` 展开证据；证据读取没有电脑操作权限。 + +上层不等待观察模型、记忆写入或 Pi 停止完成。本地只读检索有 **75 毫秒等待预算**； +超时继续后台处理，上层得到 `pending`，不能据此回答“没有记忆”。这是共享 Node 进程中的 +等待上限设计，不是 CPU 饱和时的硬实时保证。执行器已发出的原子操作仍需合作式停止。 + +## 谁在什么时候处理什么 + +| 部分 | 输入与触发 | 模型与限制 | +| --- | --- | --- | +| 事件筛选和任务投影 | 每秒检查持久事件；任务结束主动唤醒 | 程序，不调用模型 | +| 新观察 Worker | 最多 20 条事件；最早事件等约 30 秒；结束、失败、阻塞、需输入提前触发 | 独立记忆角色，`reasoning=low`、输出上限 3000、90 秒超时、并发 1 | +| 旧话题／事实整理 | 新观察队列没有可处理批次时；每分钟检查；闲置或夜间复核 | 同一记忆角色；保留旧协议，90 秒超时；不占 Pi FIFO | +| 上层决策 | 用户消息 | 当前上层角色；自然语言正文和一次原生控制调用分开 | +| 召回与来源展开 | 用户问题、Pi 任务准备、显式证据读取 | 中文／英文索引、ID 与 RRF，不另开模型 | + +本机检查到的执行和上层模型配置均为 `deepseek / deepseek-v4-flash`；本次不修改运行配置。 +新观察角色默认回退到执行角色，允许独立指定进程环境： + +```text +DESKPAW_MEMORY_MODEL_PROVIDER=deepseek +DESKPAW_MEMORY_MODEL=deepseek-v4-flash +DESKPAW_SUPERVISOR_CONTEXT_BYTES=32000 +``` + +这些是 **Agent 进程**使用的变量，不是要求 Memory SQLite 服务直接连接模型。 +新变量暂未加入设备配置工具的可编辑白名单。现有上层模型／reasoning／timeout 变量继续有效。 +是否换更小的观察模型，应以带来源提取、遗漏率、错误归属率的离线样本评估决定；本轮没有调用付费模型做质量测评。 + +## 事件与持久性 + +来源包括已完成的用户交互及上层回应、任务启动与终态、Pi 的失败／改策略／里程碑／阻塞／需输入报告。 +没有把 `agent_message_end`、隐藏推理、普通点击、图片和原始工具大对象交给观察模型。 +无变化的日程终态和日程启动被过滤。重复内容在同一批次内合并计数，仍保留每个原始事件 ID。 +敏感文本做关键词防线式脱敏；这不是能识别任意未标注凭据的完整 DLP 系统。 + +Runtime 的 `events.seq` 是发送位置；`memory_event_cursor_v1` 只在 Memory 确认所有 ID 后推进。 +Memory 接入以 `event_id` 幂等，重复 ID 不同内容会被拒绝。旧任务卡投影也会跳过完全相同的重复写入。 + +事件每条正文上限 2400 UTF-8 字节，观察工作包选取预算 12000 字节。队列状态为： + +```text +待处理事件 → 带 token 的 120 秒租约 → 成功事务：经历＋索引＋来源已消费＋回执 + └→ 失败：保留来源，退避重试 +``` + +租约过期可重领同一批；新 token 会拒绝旧 Worker 的迟到提交。提交成功而响应丢失时， +同一租约重试返回原回执，不重复发布。模型必须引用或明确丢弃所有本批事件，程序核对完整覆盖。 +来源不在批次、对象无相应来源、候选偏好缺少用户逐字证据的提案均不能消费批次。 + +首版经历为只追加记录，带设备发布序号；未实现可覆盖事实的对象级 CAS 和自动冲突消解。 +反复失败的批次会持续退避，不会默默丢弃；死信队列和人工重放界面是后续管理工作。 + +## 召回、回调和安全边界 + +新经历使用倒排表查找，不限于最近 100 次任务。融合正文、召回线索和显式对象 ID；正文证据的 +排序权重高于模型生成的线索。召回结果携带 `reported`／`candidate` 状态，不能被当作已验证事实或已批准规则。 +新经历与旧资料库并存；旧话题仍有全表排序成本，后续需要迁移，不宣称所有旧数据已完成索引化。 + +上层总文本预算默认 32000 UTF-8 字节，先扣除固定提示词和工具模式，再保留完整当前请求与实时状态。 +日志、历史和记忆有单独裁剪及省略标记。过长的当前请求拒绝派发，不静默截断后执行。 +这不是供应商精确 token 计数；中文不能用“4 字符等于 1 token”来保证预算。 + +暂停返回“已暂停队列、正在停止”，恢复返回“已接收，待检查”。改向／取消当前任务在模型调用前 +捕获任务 ID 和控制修订号，持久化决定后再核对；A 完成、B 接替或 A 已被另一路修改时，旧指令不执行。 + +关键执行事件和召回结果进入 `supervisor_inbox`，重复阻塞按任务控制版本合并，用户交互优先消费。 +`interaction_seq` 区分迟到召回；迟到结果保留，但不自动操作新的任务。 + +新增只读 Agent 接口： + +- `GET /v2/attention`：未消费的上层注意事项。 +- `GET /v2/memory/requests/`：召回状态及有界结果。 + +**这一步没有新增主动聊天通知，也没有让回调自行调用模型继续执行。** +现有任务结果仍走原来的 Gateway 输出渠道；新记忆的迟到查询结果进入缓存和收件箱， +在后续上层交互中使用，也可按 ID 查询。下一步才加入可中断的只读回调评估、通知去重和渠道投递确认。 +首次查询超过等待预算时，当前回复可能只说明查询中，不承诺已自动补发最终回答。 + +当前 `owner=device:` 约束新索引与证据接口。旧全局资料仍按单设备共享会话使用。 +这不是多租户身份认证；本地 socket 的受信边界没有改变。 + +## 开源复用落点 + +- T-Mem：实际移植 RRF 算法到 `src/rank-fusion.js`，保留 MIT 声明及版本；增加通道内去重和正文加权。 +- Mastra：借鉴观察批次、显式来源引用、准备／发布分离与异步结果归属；没有把整个框架或其可能等待后台工作的激活路径嵌进上层。 +- Letta Code：借鉴小型常驻区、按需证据展开、提案成功落地后才消费输入，以及偏好／经历／操作技能分开；没有复制 Git 工作树管理来替代 SQLite。 + +Mastra 与 Letta 本阶段属于协议和流程适配，不是宣称原样导入它们的模块。 + +## 后续顺序 + +1. 对象注册和关联：责任领域、跨月项目、软件资产与多个日程共用稳定 ID；目前适配器自动归属仅到会话／任务／日程，未自动识别所有项目。 +2. 只读回调评估及主动通知：用户消息能抢占，过期回调归档，通知有去重和送达确认；操作仍须满足原授权。 +3. 候选复核、纠正取代、遗忘／删除、保留期限与人工重放；然后迁移旧话题索引和补充向量通道。 +4. 将观察 Worker／Pi 隔离到独立进程，做持续负载和模型限流测试，保证 H618 上的调度公平性。 + +财务示例中，发票日常整理与月度提交应归同一报销领域，月度提交使用每月实例；维护软件则绑定 +资产和独立修复实例，休眠时只保留索引，相关报错再次出现才展开经历。这些对象关系是下一步，而不是在 +首版按时间摘要上贴一个“项目记忆”名称。 + +## 验证 + +2026-10-09 本机结果：Memory 25 项通过；Agent 上述四个测试文件共 54 项通过； +Agent 完整 `npm run check` 通过；两个仓库的 `git diff --check` 通过。 + +Memory：`npm test`，隔离数据库与临时 Unix socket，覆盖幂等、租约恢复、来源校验、中文跨窗口检索、 +重复观察、敏感文本、跨日复核，以及真实服务进程重启后的回执与证据读取。 + +Agent：仓库根目录 `npm run check`（包括格式、锁文件、TS、浏览器冒烟检查）； +在 `packages/coding-agent` 执行： + +```bash +node ../../node_modules/vitest/dist/cli.js --run test/device-continuity.test.ts test/device-runtime.test.ts test/device-memory.test.ts test/device-session.test.ts +``` + +这些测试使用离线模型与假 Executor，不操作真实桌面，不访问真实模型 API。覆盖慢记忆服务、 +后台观察与前台解耦、暂停回执、过期目标、天然语言与控制分离、会话恢复、日程以及证据筛选。 +正式环境验证和部署仍需单独进行；部署前备份 Runtime／Memory SQLite（含 WAL 一致性），先更新 Memory 协议再更新 Agent。 diff --git a/src/continuity.js b/src/continuity.js new file mode 100644 index 0000000..6fa786f --- /dev/null +++ b/src/continuity.js @@ -0,0 +1,238 @@ +import { randomUUID } from "node:crypto"; +import { safeMemoryText, words } from "./policy.js"; +import { reciprocalRankFusion } from "./rank-fusion.js"; + +const KINDS = new Set(["user_message", "manager_reply", "task_started", "task_completed", "task_failed", + "task_cancelled", "task_uncertain", "task_expired", "attempt_failed", "strategy_changed", "milestone", "blocked", "need_user_input"]); +const URGENT = new Set(["task_completed", "task_failed", "task_uncertain", "blocked", "need_user_input"]); +const SOURCES = new Set(["user", "runtime", "pi", "supervisor"]); +const REDACTED = "[敏感内容未写入记忆]"; + +function id(value) { + if (typeof value !== "string" || !/^[\w:.-]{1,240}$/.test(value)) throw new Error("Invalid continuity ID"); + return value; +} + +function bounded(value, bytes) { + const text = safeMemoryText(value, bytes); + let result = ""; + let used = 0; + for (const char of text) { + used += Buffer.byteLength(char); + if (used > bytes) break; + result += char; + } + return result; +} + +function groupsOf(events) { + const groups = new Map(); + for (const event of events) { + const key = JSON.stringify([event.subject_id, event.task_id, event.kind, event.source, event.text]); + if (!groups.has(key)) groups.set(key, []); + groups.get(key).push(event); + } + return [...groups.values()]; +} + +/** Append-only evidence and leased observation jobs. No model or Agent dependency. */ +export class ContinuityStore { + constructor(db) { + this.db = db; + db.exec(` + CREATE TABLE IF NOT EXISTS continuity_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT UNIQUE NOT NULL, owner TEXT NOT NULL, + subject_id TEXT NOT NULL, task_id TEXT, kind TEXT NOT NULL, source TEXT NOT NULL, + observed_at INTEGER NOT NULL, received_at INTEGER NOT NULL, text TEXT NOT NULL, + job_id TEXT, consumed INTEGER NOT NULL DEFAULT 0); + CREATE INDEX IF NOT EXISTS continuity_pending ON continuity_events(consumed,job_id,seq); + CREATE TABLE IF NOT EXISTS continuity_jobs ( + job_id TEXT PRIMARY KEY, owner TEXT NOT NULL, token TEXT NOT NULL, lease_until INTEGER NOT NULL, + available_at INTEGER NOT NULL, attempts INTEGER NOT NULL, status TEXT NOT NULL, receipt TEXT); + CREATE INDEX IF NOT EXISTS continuity_jobs_ready ON continuity_jobs(status,available_at,lease_until); + CREATE TABLE IF NOT EXISTS continuity_records ( + record_id TEXT PRIMARY KEY, owner TEXT NOT NULL, subject_id TEXT NOT NULL, kind TEXT NOT NULL, + content TEXT NOT NULL, cues TEXT NOT NULL, source_refs TEXT NOT NULL, status TEXT NOT NULL, + revision INTEGER NOT NULL, created_at INTEGER NOT NULL); + CREATE INDEX IF NOT EXISTS continuity_records_subject ON continuity_records(owner,subject_id,created_at); + CREATE TABLE IF NOT EXISTS continuity_terms ( + owner TEXT NOT NULL, term TEXT NOT NULL, record_id TEXT NOT NULL, channel TEXT NOT NULL, + PRIMARY KEY(owner,term,record_id,channel)); + CREATE TABLE IF NOT EXISTS continuity_versions (owner TEXT PRIMARY KEY, revision INTEGER NOT NULL); + `); + } + + transaction(fn) { + this.db.exec("BEGIN IMMEDIATE"); + try { + const result = fn(); + this.db.exec("COMMIT"); + return result; + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + } + + ingest({ version, owner, events }) { + if (version !== 1 || !Array.isArray(events) || events.length > 100) throw new Error("Invalid event envelope"); + owner = id(owner); + return this.transaction(() => { + let stored = 0; + for (const event of events) { + if (!KINDS.has(event.kind) || !SOURCES.has(event.source) || !Number.isSafeInteger(event.observed_at)) + throw new Error("Invalid memory event"); + if ((event.kind === "user_message") !== (event.source === "user")) throw new Error("Invalid user provenance"); + const values = [id(event.event_id), owner, id(event.subject_id), event.task_id ? id(event.task_id) : null, + event.kind, event.source, event.observed_at, bounded(event.text, 2_400)]; + const previous = this.db.prepare(`SELECT event_id,owner,subject_id,task_id,kind,source,observed_at,text + FROM continuity_events WHERE event_id=?`).get(values[0]); + if (previous) { + if (JSON.stringify(Object.values(previous)) !== JSON.stringify(values)) throw new Error("Event ID collision"); + continue; + } + this.db.prepare(`INSERT INTO continuity_events + (event_id,owner,subject_id,task_id,kind,source,observed_at,text,received_at) VALUES (?,?,?,?,?,?,?,?,?)`) + .run(...values, Date.now()); + stored++; + } + return { stored, acknowledged: events.map((event) => event.event_id) }; + }); + } + + claim({ owner }, now = Date.now()) { + owner = id(owner); + return this.transaction(() => { + // Only one model job per owner; an expired lease reclaims exactly the same evidence. + let job = this.db.prepare("SELECT * FROM continuity_jobs WHERE owner=? AND status='pending' ORDER BY rowid LIMIT 1").get(owner); + if (job && (Number(job.lease_until) > now || Number(job.available_at) > now)) return null; + if (!job) { + const pending = this.db.prepare(`SELECT * FROM continuity_events WHERE owner=? AND consumed=0 + AND job_id IS NULL ORDER BY seq LIMIT 20`).all(owner); + if (!pending.length) return null; + if (pending.length < 20 && now - Number(pending[0].received_at) < 30_000 && !pending.some((event) => URGENT.has(event.kind))) return null; + const selected = []; + let size = 0; + for (const event of pending) { + const bytes = Buffer.byteLength(JSON.stringify(event)); + if (size + bytes > 12_000) break; + selected.push(event); + size += bytes; + } + const jobId = randomUUID(); + this.db.prepare("INSERT INTO continuity_jobs VALUES (?,?,?,?,?,?,?,NULL)").run(jobId, owner, "", 0, now, 0, "pending"); + for (const event of selected) this.db.prepare("UPDATE continuity_events SET job_id=? WHERE seq=?").run(jobId, event.seq); + job = { job_id: jobId, attempts: 0 }; + } + const token = randomUUID(); + const leaseUntil = now + 120_000; + this.db.prepare("UPDATE continuity_jobs SET token=?,lease_until=?,attempts=attempts+1 WHERE job_id=?") + .run(token, leaseUntil, job.job_id); + const events = this.db.prepare(`SELECT event_id,subject_id,task_id,kind,source,observed_at,text + FROM continuity_events WHERE job_id=? ORDER BY seq`).all(job.job_id); + return { job_id: job.job_id, token, owner, lease_until: leaseUntil, + events: groupsOf(events).map((group) => ({ ...group[0], repeat_count: group.length, + last_observed_at: group.at(-1).observed_at })) }; + }); + } + + fail({ job_id, token }, now = Date.now()) { + const job = this.db.prepare("SELECT * FROM continuity_jobs WHERE job_id=? AND token=? AND status='pending'").get(id(job_id), id(token)); + if (!job) return { released: false }; + const delay = Math.min(600_000, 5_000 * 2 ** Math.min(7, Number(job.attempts))); + this.db.prepare("UPDATE continuity_jobs SET lease_until=0,available_at=? WHERE job_id=? AND token=?").run(now + delay, job_id, token); + return { released: true, retry_at: now + delay }; + } + + commit({ job_id, token, episodes, candidates = [], discarded_event_ids = [] }, now = Date.now()) { + return this.transaction(() => { + const job = this.db.prepare("SELECT * FROM continuity_jobs WHERE job_id=? AND token=?").get(id(job_id), id(token)); + if (!job) throw new Error("Stale observation lease"); + if (job.status === "done") return JSON.parse(job.receipt); + if (Number(job.lease_until) <= now) throw new Error("Expired observation lease"); + if (!Array.isArray(episodes) || episodes.length > 6 || !Array.isArray(candidates) || candidates.length > 4 || !Array.isArray(discarded_event_ids)) + throw new Error("Invalid observation proposal"); + const events = this.db.prepare("SELECT * FROM continuity_events WHERE job_id=? ORDER BY seq").all(job_id); + const sources = new Map(events.map((event) => [event.event_id, event])); + const aliases = new Map(groupsOf(events).flatMap((group) => group.map((event) => + [event.event_id, group.map((item) => item.event_id)]))); + const accounted = new Set(); + const records = []; + for (const [kind, items] of [["episode", episodes], ["candidate", candidates]]) { + for (const item of items) { + if (!Array.isArray(item.source_refs) || !item.source_refs.length || item.source_refs.length > 20) throw new Error("Missing evidence"); + const requestedRefs = item.source_refs.map(id); + const refs = [...new Set(requestedRefs.flatMap((ref) => aliases.get(ref) ?? [ref]))]; + if (refs.some((ref) => !sources.has(ref))) throw new Error("Evidence outside observation job"); + const subjectId = id(item.subject_id); + if (!refs.some((ref) => sources.get(ref).subject_id === subjectId)) throw new Error("Subject has no evidence"); + const content = bounded(item.content, 1_600); + if (!content || content === REDACTED) throw new Error("Unsafe or empty observation"); + if (kind === "candidate") { + // Never promote model inferences, executor text or assistant repetitions to user rules. + const quote = bounded(item.user_quote, 1_000); + if (!quote || quote === REDACTED || !refs.some((ref) => sources.get(ref).source === "user" && sources.get(ref).text.includes(quote))) + throw new Error("Candidate requires an exact user quote"); + if (!["preference", "correction"].includes(item.kind)) throw new Error("Invalid candidate kind"); + } + const cues = Array.isArray(item.cues) ? item.cues.slice(0, 3).map((cue) => bounded(cue, 150)).filter((cue) => cue && cue !== REDACTED) : []; + refs.forEach((ref) => accounted.add(ref)); + records.push({ record_id: randomUUID(), subject_id: subjectId, kind: kind === "episode" ? kind : item.kind, + content, cues, source_refs: refs, status: kind === "episode" ? "reported" : "candidate" }); + } + } + for (const ref of discarded_event_ids) { + if (!sources.has(ref)) throw new Error("Invalid discarded evidence"); + for (const alias of aliases.get(ref)) { + if (accounted.has(alias)) throw new Error("Invalid discarded evidence"); + accounted.add(alias); + } + } + if (accounted.size !== sources.size) throw new Error("Proposal does not account for all source events"); + const revision = this.revision(job.owner) + 1; + for (const record of records) { + this.db.prepare("INSERT INTO continuity_records VALUES (?,?,?,?,?,?,?,?,?,?)") + .run(record.record_id, job.owner, record.subject_id, record.kind, record.content, JSON.stringify(record.cues), JSON.stringify(record.source_refs), record.status, revision, now); + for (const [channel, text] of [["content", record.content], ["cue", record.cues.join(" ")]]) { + for (const term of [...words(text)].slice(0, 600)) { + this.db.prepare("INSERT OR IGNORE INTO continuity_terms VALUES (?,?,?,?)").run(job.owner, term, record.record_id, channel); + } + } + } + this.db.prepare("INSERT OR REPLACE INTO continuity_versions VALUES (?,?)").run(job.owner, revision); + this.db.prepare("UPDATE continuity_events SET consumed=1 WHERE job_id=?").run(job_id); + const receipt = { job_id, revision, records: records.length, consumed: events.length }; + this.db.prepare("UPDATE continuity_jobs SET status='done',receipt=? WHERE job_id=?").run(JSON.stringify(receipt), job_id); + return receipt; + }); + } + + revision(owner) { + return Number(this.db.prepare("SELECT revision FROM continuity_versions WHERE owner=?").get(owner)?.revision ?? 0); + } + + recall({ owner, query = "", subject_id, limit = 6 }) { + owner = id(owner); + limit = Number.isSafeInteger(limit) ? Math.min(10, Math.max(1, limit)) : 6; + const terms = [...words(query)].slice(0, 32); + const rankings = ["content", "cue"].map((channel) => terms.length ? this.db.prepare(`SELECT record_id,count(*) AS matches + FROM continuity_terms WHERE owner=? AND channel=? AND term IN (${terms.map(() => "?").join(",")}) + GROUP BY record_id ORDER BY matches DESC,record_id LIMIT 50`).all(owner, channel, ...terms).map((row) => row.record_id) : []); + if (subject_id) rankings.push(this.db.prepare(`SELECT record_id FROM continuity_records WHERE owner=? AND subject_id=? + ORDER BY created_at DESC,record_id LIMIT 30`).all(owner, id(subject_id)).map((row) => row.record_id)); + // Direct text evidence has twice the vote of speculative recall cues. Otherwise + // many generic cues can tie with and crowd out a precise dormant-asset match. + const hits = reciprocalRankFusion([rankings[0], ...rankings], limit).map(({ id: recordId, score }) => { + const row = this.db.prepare("SELECT * FROM continuity_records WHERE record_id=? AND owner=?").get(recordId, owner); + return { ...row, cues: JSON.parse(row.cues), source_refs: JSON.parse(row.source_refs), score }; + }); + return { state: "available", revision: this.revision(owner), hits }; + } + + evidence({ owner, event_ids }) { + owner = id(owner); + if (!Array.isArray(event_ids) || event_ids.length > 20) throw new Error("Invalid evidence request"); + return event_ids.map((eventId) => this.db.prepare(`SELECT event_id,subject_id,kind,source,observed_at,text + FROM continuity_events WHERE event_id=? AND owner=?`).get(id(eventId), owner)).filter(Boolean); + } +} diff --git a/src/policy.js b/src/policy.js index 9d0a0af..5dfbd9d 100644 --- a/src/policy.js +++ b/src/policy.js @@ -15,11 +15,12 @@ export function safeMemoryText(value, maxLength = 8_000) { return text; } -export function words(value) { - const lowered = boundedText(value, 16_000).toLowerCase(); +export function words(value, maxLength = 16_000) { + const lowered = boundedText(value, maxLength).toLowerCase(); const terms = new Set(lowered.match(/[a-z0-9_-]{2,}|[\u3400-\u9fff]{2}/g) ?? []); - const cjk = [...lowered.matchAll(/[\u3400-\u9fff]/g)].map((match) => match[0]).join(""); - for (let index = 0; index + 1 < cjk.length; index++) terms.add(cjk.slice(index, index + 2)); + for (const [cjk] of lowered.matchAll(/[\u3400-\u9fff]+/g)) { + for (let index = 0; index + 1 < cjk.length; index++) terms.add(cjk.slice(index, index + 2)); + } return terms; } diff --git a/src/rank-fusion.js b/src/rank-fusion.js new file mode 100644 index 0000000..61560fb --- /dev/null +++ b/src/rank-fusion.js @@ -0,0 +1,15 @@ +// Adapted from T-Mem dd9e152, T_mem/main/stage6_retrieval.py: +// reciprocal_rank_fusion. See THIRD_PARTY_NOTICES.md (MIT). +export function reciprocalRankFusion(rankings, limit = 8, k = 60) { + const scores = new Map(); + for (const ranking of rankings) { + const seen = new Set(); + ranking.forEach((id, rank) => { + if (seen.has(id)) return; + seen.add(id); + scores.set(id, (scores.get(id) ?? 0) + 1 / (k + rank + 1)); + }); + } + return [...scores].sort((a, b) => b[1] - a[1] || a[0].localeCompare(b[0])) + .slice(0, limit).map(([id, score]) => ({ id, score })); +} diff --git a/src/server.js b/src/server.js index f1890e3..66fd481 100644 --- a/src/server.js +++ b/src/server.js @@ -1,9 +1,11 @@ #!/usr/bin/env node -import { chmodSync, existsSync, mkdirSync, rmSync } from "node:fs"; -import { createServer } from "node:net"; +import { chmodSync, existsSync, lstatSync, mkdirSync, rmSync } from "node:fs"; +import { createConnection, createServer } from "node:net"; import { homedir } from "node:os"; import { dirname, join, resolve } from "node:path"; import { MemoryStore } from "./store.js"; +import { ContinuityStore } from "./continuity.js"; +import { Timeline } from "./timeline.js"; const stateDir = resolve(process.env.DESKPAW_STATE_DIR ?? join(homedir(), ".deskpaw")); const runtimeDir = join(stateDir, "runtime"); @@ -13,10 +15,36 @@ const databasePath = resolve(process.env.DESKPAW_MEMORY_DB ?? join(memoryDir, "d const bundlesDir = resolve(process.env.DESKPAW_MEMORY_BUNDLES ?? join(memoryDir, "bundles")); process.umask(0o077); mkdirSync(dirname(databasePath), { recursive: true, mode: 0o700 }); -if (process.platform !== "win32" && existsSync(socketPath)) rmSync(socketPath); +if (process.platform !== "win32") { + mkdirSync(dirname(socketPath), { recursive: true, mode: 0o700 }); + if (existsSync(socketPath)) { + if (!lstatSync(socketPath).isSocket()) throw new Error("Refusing to replace non-socket memory path"); + const active = await new Promise((resolve, reject) => { + const socket = createConnection(socketPath); + socket.once("connect", () => { socket.destroy(); resolve(true); }); + socket.once("error", error => { + if (["ENOENT", "ECONNREFUSED"].includes(error.code)) resolve(false); else reject(error); + }); + socket.setTimeout(1000, () => { socket.destroy(); reject(new Error("Cannot verify memory socket owner")); }); + }); + if (active) throw new Error("Memory socket already has an owner"); + if (existsSync(socketPath)) rmSync(socketPath); + } +} const store = new MemoryStore(databasePath, bundlesDir); +const continuity = new ContinuityStore(store.db); +const timeline = new Timeline(store.db); function call(method, params) { + if (method === "memory.events.ingest") return continuity.ingest(params); + if (method === "memory.observation.claim") return continuity.claim(params); + if (method === "memory.observation.commit") return continuity.commit(params); + if (method === "memory.observation.fail") return continuity.fail(params); + if (method === "memory.continuity.recall") return continuity.recall(params); + if (method === "memory.evidence.get") return continuity.evidence(params); + if (method === "timeline.append") return timeline.append(params); + if (method === "timeline.query") return timeline.query(params); + if (method === "timeline.read" || method === "timeline.expand") return timeline.read(params); if (method === "health") return { status: "ok", version: "0.1.0", ...store.status() }; if (method === "memory.recall") return store.recall(params); if (method === "memory.maintenance.status") return store.maintenanceStatus(Number(params.now ?? Date.now())); @@ -28,7 +56,9 @@ function call(method, params) { if (method === "memory.skill.set_status") return store.setSkillStatus(params); if (method === "interaction.remember") return store.rememberInteractions(params); if (method === "transition.assess") return store.assessTransition(params); - if (method === "context.prepare") return store.prepareContext(params); + if (method === "context.prepare") return { ...store.prepareContext(params), + ...(params.owner ? { continuity: continuity.recall({ owner: params.owner, query: params.objective, + subject_id: params.schedule_id ? `schedule:${params.schedule_id}` : undefined, limit: 4 }) } : {}) }; if (method === "task.complete") return store.completeTask(params); if (method === "consolidation.record") return store.recordConsolidation(params); if (method === "bundle.get") return store.bundle(String(params.scope ?? ""), String(params.object_id ?? "")); @@ -40,23 +70,30 @@ function call(method, params) { const server = createServer((socket) => { socket.setEncoding("utf8"); + socket.setTimeout(5000, () => socket.destroy()); let input = ""; + let handled = false; socket.on("error", () => {}); socket.on("data", (chunk) => { + if (handled) return; input += chunk; - if (input.length > 1_000_000) socket.destroy(new Error("Request too large")); + if (Buffer.byteLength(input) > 1_000_000) { handled = true; socket.destroy(); return; } const newline = input.indexOf("\n"); if (newline < 0) return; + handled = true; const line = input.slice(0, newline); + let id = null; try { const request = JSON.parse(line); + id = request.id ?? null; const result = call(String(request.method ?? ""), request.params ?? {}); socket.end(JSON.stringify({ id: request.id ?? null, ok: true, result }) + "\n"); } catch (error) { - socket.end(JSON.stringify({ id: null, ok: false, error: error instanceof Error ? error.message : String(error) }) + "\n"); + socket.end(JSON.stringify({ id, ok: false, error: error instanceof Error ? error.message : String(error) }) + "\n"); } }); }); +server.maxConnections = 64; server.listen(socketPath, () => { if (process.platform !== "win32") chmodSync(socketPath, 0o600); diff --git a/src/store.js b/src/store.js index 7cc2144..a4112aa 100644 --- a/src/store.js +++ b/src/store.js @@ -392,13 +392,15 @@ export class MemoryStore { completeTask(params) { const taskId = boundedText(params.task_id, 200); - const objective = boundedText(params.objective, 16_000); + const objective = safeMemoryText(params.objective, 16_000); const status = boundedText(params.status, 40) || "completed"; const result = safeMemoryText(params.result, 4_000); const scheduleId = boundedText(params.schedule_id, 200) || null; const contextEpoch = boundedText(params.context_epoch, 200) || randomUUID(); const now = Date.now(); - const existing = this.db.prepare("SELECT created_at FROM task_cards WHERE task_id=?").get(taskId); + const existing = this.db.prepare("SELECT * FROM task_cards WHERE task_id=?").get(taskId); + if (existing && existing.objective === objective && existing.status === status && + existing.result === result && existing.schedule_id === scheduleId) return { task_id: taskId, stored: false }; this.db.prepare("INSERT OR REPLACE INTO task_cards VALUES (?,?,?,?,?,?,?,?)") .run(taskId, objective, status, result, scheduleId, contextEpoch, Number(existing?.created_at ?? now), now); if (scheduleId) { @@ -598,10 +600,12 @@ export class MemoryStore { this.db.prepare(`UPDATE memory_facts SET status='inactive',updated_at=? WHERE status='candidate' AND (SELECT max(workday) FROM memory_fact_evidence WHERE fact_id=memory_facts.fact_id) <= date(?, '-3 day')`) - .run(now, workdayAt(now)); + .run(now, workday); if (params.nightly === true) this.db.prepare("INSERT OR REPLACE INTO memory_meta VALUES ('last_reviewed_workday',?)") - .run(workdayAt(now)); + // Bind the receipt to the planned workday, not a clock that may have + // crossed the 03:30 boundary while the model was preparing its result. + .run(workday); this.event("maintenance.committed", { workday, assigned: assigned.length, facts: storedFacts, skills: storedSkills }); this.db.exec("COMMIT"); return { assigned: assigned.length, facts: storedFacts, skills: storedSkills }; diff --git a/src/timeline.js b/src/timeline.js new file mode 100644 index 0000000..85f66f0 --- /dev/null +++ b/src/timeline.js @@ -0,0 +1,59 @@ +import { safeMemoryText, words } from "./policy.js"; + +/** Append-only experience ledger. Old indexes/embeddings are not consulted. */ +export class Timeline { + constructor(db) { + this.db = db; + db.exec(`CREATE TABLE IF NOT EXISTS timeline ( + seq INTEGER PRIMARY KEY AUTOINCREMENT,id TEXT UNIQUE NOT NULL,kind TEXT NOT NULL, + at INTEGER NOT NULL,text TEXT NOT NULL,source TEXT NOT NULL, + parent_id TEXT,covered_from INTEGER,covered_to INTEGER,version INTEGER NOT NULL DEFAULT 1); + CREATE INDEX IF NOT EXISTS timeline_time ON timeline(at,seq); + CREATE TABLE IF NOT EXISTS timeline_terms (term TEXT NOT NULL,seq INTEGER NOT NULL,PRIMARY KEY(term,seq)); + CREATE INDEX IF NOT EXISTS timeline_terms_seq ON timeline_terms(seq);`); + } + append(p) { + if (typeof p.id !== "string" || !p.id || p.id.length > 240) throw new Error("Invalid timeline id"); + if (typeof p.kind !== "string" || !/^[a-z_]{1,40}$/.test(p.kind)) throw new Error("Invalid timeline kind"); + if (typeof p.text !== "string" || p.text.length > 32000) throw new Error("Timeline text exceeds 32000 characters; use artifact references"); + if (!Number.isSafeInteger(p.at) || p.at < 0) throw new Error("Invalid timestamp"); + const source = {}; + for (const key of ["channel", "conversation_id", "sender", "actor", "message_id", "job_id", "task_id", "artifact_ref"]) { + if (typeof p.source?.[key] === "string") source[key] = safeMemoryText(p.source[key], 240); + } + const text = safeMemoryText(p.text, 32000); + this.db.exec("BEGIN IMMEDIATE"); + try { + const result = this.db.prepare("INSERT OR IGNORE INTO timeline(id,kind,at,text,source) VALUES (?,?,?,?,?)").run(p.id, p.kind, p.at, text, JSON.stringify(source)); + if (result.changes) { + const seq = Number(result.lastInsertRowid); + const insert = this.db.prepare("INSERT OR IGNORE INTO timeline_terms(term,seq) VALUES (?,?)"); + for (const term of words(text, 32000)) insert.run(term, seq); + } + this.db.exec("COMMIT"); + } catch (error) { this.db.exec("ROLLBACK"); throw error; } + return { id: p.id, stored: true }; + } + query(p = {}) { + const limit = p.limit ?? 10; + const after = p.after ?? 0; + const from = p.from ?? 0; + const to = p.to ?? Number.MAX_SAFE_INTEGER; + if (![limit, after, from, to].every(Number.isSafeInteger) || limit < 1 || limit > 50 || after < 0 || from < 0 || to < from) throw new Error("Invalid query bounds"); + const terms = [...words(String(p.query ?? ""))].slice(0, 32); + const rows = terms.length ? this.db.prepare(`SELECT t.*,count(*) AS score FROM timeline t + JOIN timeline_terms x ON x.seq=t.seq WHERE x.term IN (${terms.map(() => "?").join(",")}) + AND t.seq>? AND t.at BETWEEN ? AND ? GROUP BY t.seq ORDER BY score DESC,t.at DESC,t.seq DESC LIMIT ?`).all(...terms, after, from, to, limit) + : this.db.prepare("SELECT * FROM timeline WHERE seq>? AND at BETWEEN ? AND ? ORDER BY seq LIMIT ?").all(after, from, to, limit); + return { entries: rows.map((row) => ({ ...row, text: row.text.slice(0, 1200), source: JSON.parse(row.source), expandable: row.text.length > 1200 })), cursor: terms.length ? null : rows.at(-1)?.seq ?? after, method: "lexical_time", embedding: false }; + } + read(p) { + if (typeof p.id !== "string") throw new Error("id required"); + const offset = p.offset ?? 0; + const limit = p.limit ?? 8000; + if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(limit) || limit < 1 || limit > 32000) throw new Error("Invalid read range"); + const row = this.db.prepare("SELECT * FROM timeline WHERE id=?").get(p.id); + if (!row) throw new Error("Timeline entry not found"); + return { ...row, source: JSON.parse(row.source), text: row.text.slice(offset, offset + limit), next_offset: offset + limit < row.text.length ? offset + limit : null }; + } +} diff --git a/test/continuity.test.js b/test/continuity.test.js new file mode 100644 index 0000000..2f87ea8 --- /dev/null +++ b/test/continuity.test.js @@ -0,0 +1,128 @@ +import assert from "node:assert/strict"; +import { DatabaseSync } from "node:sqlite"; +import { test } from "node:test"; +import { ContinuityStore } from "../src/continuity.js"; +import { reciprocalRankFusion } from "../src/rank-fusion.js"; + +const owner = "device:test"; +function event(eventId = "e1", overrides = {}) { + return { event_id: eventId, subject_id: "task:reimburse", task_id: "reimburse", kind: "task_completed", + source: "runtime", observed_at: Date.now(), text: "十月报销的发票已经整理到报销区，尚未提交审批。", ...overrides }; +} +function setup(t) { + const db = new DatabaseSync(":memory:"); + const store = new ContinuityStore(db); + t.after(() => db.close()); + return store; +} +function ingest(store, events) { return store.ingest({ version: 1, owner, events }); } +function proposal(job, content = "十月报销发票已整理，尚未提交审批。") { + return { job_id: job.job_id, token: job.token, episodes: [{ subject_id: job.events[0].subject_id, content, + cues: ["每月报销整理发票"], source_refs: job.events.map((item) => item.event_id) }], candidates: [], discarded_event_ids: [] }; +} + +test("idempotent ingest rejects conflicting IDs and rolls back the whole envelope", (t) => { + const store = setup(t); + const first = event(); + assert.equal(ingest(store, [first]).stored, 1); + assert.equal(ingest(store, [first]).stored, 0); + assert.throws(() => ingest(store, [event("e2"), { ...first, text: "changed" }]), /collision/); + assert.equal(store.db.prepare("SELECT count(*) AS n FROM continuity_events").get().n, 1); + assert.throws(() => ingest(store, [event("forged", { source: "user" })]), /provenance/); +}); + +test("leases recover the same batch, reject stale workers, commit atomically and replay receipts", (t) => { + const store = setup(t); + ingest(store, [event()]); + const now = Date.now(); + const job = store.claim({ owner }, now); + assert.equal(store.claim({ owner }, now + 1), null); + const restored = new ContinuityStore(store.db).claim({ owner }, now + 120_001); + assert.equal(job.job_id, restored.job_id); + assert.notEqual(job.token, restored.token); + assert.throws(() => store.commit(proposal(job), now + 120_002), /Stale/); + const receipt = store.commit(proposal(restored), now + 120_002); + assert.equal(receipt.consumed, 1); + assert.deepEqual(store.commit(proposal(restored), now + 999_999), receipt); + assert.equal(store.claim({ owner }, now + 1_000_000), null); +}); + +test("missing or invented evidence never consumes an observation batch", (t) => { + const store = setup(t); + ingest(store, [event("e1"), event("e2", { text: "另一项没有引用的独立结果" })]); + const job = store.claim({ owner }); + const draft = proposal(job); + draft.episodes[0].source_refs = ["e1"]; + assert.throws(() => store.commit(draft), /all source events/); + draft.episodes[0].source_refs = ["invented"]; + assert.throws(() => store.commit(draft), /outside/); + assert.equal(store.db.prepare("SELECT count(*) AS n FROM continuity_records").get().n, 0); + assert.equal(store.db.prepare("SELECT sum(consumed) AS n FROM continuity_events").get().n, 0); + assert.equal(store.fail(job).released, true); + assert.equal(store.claim({ owner }), null); +}); + +test("background proposals cannot promote executor text to preferences or change another subject", (t) => { + const store = setup(t); + ingest(store, [event()]); + const job = store.claim({ owner }); + const draft = proposal(job); + draft.candidates = [{ ...draft.episodes[0], kind: "preference", user_quote: "报销" }]; + assert.throws(() => store.commit(draft), /exact user quote/); + draft.candidates = []; + draft.episodes[0].subject_id = "project:other"; + assert.throws(() => store.commit(draft), /no evidence/); +}); + +test("explicit corrections remain sourced candidates, not automatically active global rules", (t) => { + const store = setup(t); + ingest(store, [event("correction", { source: "user", kind: "user_message", text: "报销前先让我确认发票清单。" })]); + const job = store.claim({ owner }, Date.now() + 31_000); + store.commit({ job_id: job.job_id, token: job.token, episodes: [], candidates: [{ subject_id: "task:reimburse", + kind: "correction", content: "报销前须确认发票清单", user_quote: "报销前先让我确认发票清单。", + source_refs: ["correction"], cues: ["报销确认"] }], discarded_event_ids: [] }, Date.now() + 31_001); + const record = store.recall({ owner, query: "报销确认" }).hits[0]; + assert.equal(record.status, "candidate"); + assert.deepEqual(record.source_refs, ["correction"]); +}); + +test("Chinese recall finds dormant software beyond a recent-history window and respects owner scope", (t) => { + const store = setup(t); + ingest(store, [event("software", { subject_id: "asset:billing", text: "账单软件升级后导出发票失败，需要回滚模板。" })]); + store.commit(proposal(store.claim({ owner }), "账单软件导出发票故障：模板字段变更，回滚后恢复。")); + for (let i = 0; i < 110; i++) { + ingest(store, [event(`noise${i}`, { subject_id: `task:noise${i}` })]); + store.commit(proposal(store.claim({ owner }), `天气查看任务编号${i}`)); + } + const recalled = store.recall({ owner, query: "账单软件导出发票又失败" }); + assert.match(recalled.hits[0].content, /模板字段/); + assert.equal(recalled.hits[0].subject_id, "asset:billing"); + assert.deepEqual(store.recall({ owner: "device:other", query: "账单软件" }).hits, []); + assert.deepEqual(store.evidence({ owner: "device:other", event_ids: ["software"] }), []); + assert.equal(store.evidence({ owner, event_ids: ["software"] })[0].source, "runtime"); +}); + +test("secrets are redacted before the model, and job size is bounded in UTF-8 bytes", (t) => { + const store = setup(t); + ingest(store, [event("secret", { text: "密码=do-not-retain" }), ...Array.from({ length: 19 }, (_, i) => event(`long${i}`, { text: "中文观察".repeat(2000) }))]); + const job = store.claim({ owner }); + assert.ok(!JSON.stringify(job).includes("do-not-retain")); + assert.ok(Buffer.byteLength(JSON.stringify(job.events)) < 12_000); + assert.ok(job.events.length < 20); +}); + +test("RRF fuses channels deterministically without rewarding duplicates within one channel", () => { + const ranked = reciprocalRankFusion([["b", "a"], ["a", "c"]]); + assert.equal(ranked[0].id, "a"); + assert.equal(reciprocalRankFusion([["a", "a"]])[0].score, 1 / 61); +}); + +test("repeated observations collapse for the model but retain every evidence reference", (t) => { + const store = setup(t); + ingest(store, [event("one"), event("two")]); + const job = store.claim({ owner }); + assert.equal(job.events.length, 1); + assert.equal(job.events[0].repeat_count, 2); + assert.equal(store.commit(proposal(job)).consumed, 2); + assert.deepEqual(store.recall({ owner, query: "报销" }).hits[0].source_refs, ["one", "two"]); +}); diff --git a/test/protocol.test.js b/test/protocol.test.js new file mode 100644 index 0000000..1ea8eaf --- /dev/null +++ b/test/protocol.test.js @@ -0,0 +1,71 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createConnection } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { test } from "node:test"; + +test("real socket protocol preserves observations and provenance across service restart", async (t) => { + const dir = mkdtempSync(join(tmpdir(), "dp-memory-protocol-")); + const socketPath = join(dir, "memory.sock"); + let child; + async function start() { + child = spawn(process.execPath, [new URL("../src/server.js", import.meta.url).pathname], { + env: { DESKPAW_STATE_DIR: dir, DESKPAW_MEMORY_SOCKET: socketPath, + DESKPAW_MEMORY_DB: join(dir, "data.sqlite"), DESKPAW_MEMORY_BUNDLES: join(dir, "bundles") }, + stdio: ["ignore", "pipe", "pipe"], + }); + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error("Isolated memory server did not start")), 5000); + child.once("error", (error) => { clearTimeout(timeout); reject(error); }); + child.once("exit", (code) => { clearTimeout(timeout); reject(new Error(`Server exited: ${code}`)); }); + child.stdout.once("data", () => { clearTimeout(timeout); resolve(); }); + }); + } + async function stop() { + if (!child || child.exitCode !== null) return; + const exited = new Promise((resolve) => child.once("exit", resolve)); + child.kill("SIGTERM"); + const timeout = setTimeout(() => child.kill("SIGKILL"), 2000); + await exited; + clearTimeout(timeout); + } + function call(method, params) { + return new Promise((resolve, reject) => { + const socket = createConnection(socketPath); + let text = ""; + socket.setTimeout(3000, () => socket.destroy(new Error("Test socket timed out"))); + socket.once("connect", () => socket.write(JSON.stringify({ id: "test", method, params }) + "\n")); + socket.on("data", (chunk) => { text += chunk; }); + socket.once("end", () => { + try { + const response = JSON.parse(text); + if (!response.ok) throw new Error(response.error); + resolve(response.result); + } catch (error) { reject(error); } + }); + socket.once("error", reject); + }); + } + t.after(async () => { await stop(); rmSync(dir, { recursive: true, force: true }); }); + await start(); + const owner = "device:protocol"; + const event = { event_id: "run:1", subject_id: "schedule:monthly", task_id: "run1", + kind: "task_completed", source: "runtime", observed_at: Date.now(), text: "十月发票归档完成，等待确认。" }; + assert.equal((await call("memory.events.ingest", { version: 1, owner, events: [event] })).stored, 1); + const job = await call("memory.observation.claim", { owner }); + const proposal = { job_id: job.job_id, token: job.token, episodes: [{ subject_id: event.subject_id, + content: "十月发票已整理，等待用户确认。", source_refs: [event.event_id], cues: ["每月报销"] }], candidates: [], discarded_event_ids: [] }; + const receipt = await call("memory.observation.commit", proposal); + await stop(); + await start(); + assert.deepEqual(await call("memory.observation.commit", proposal), receipt); + const recalled = await call("memory.continuity.recall", { owner, query: "发票报销" }); + assert.equal(recalled.hits.length, 1); + const context = await call("context.prepare", { owner, task_id: "run2", objective: "继续报销", schedule_id: "monthly" }); + assert.equal(context.continuity.hits[0].source_refs[0], "run:1"); + assert.equal((await call("memory.evidence.get", { owner, event_ids: ["run:1"] }))[0].text, event.text); + assert.deepEqual(await call("memory.evidence.get", { owner: "device:another", event_ids: ["run:1"] }), []); + assert.equal(await call("memory.observation.claim", { owner }), null); +}); diff --git a/test/server-timeline.test.js b/test/server-timeline.test.js new file mode 100644 index 0000000..8b25401 --- /dev/null +++ b/test/server-timeline.test.js @@ -0,0 +1,31 @@ +import { spawn } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createConnection } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { once } from "node:events"; +import assert from "node:assert/strict"; +import { test } from "node:test"; + +test("real socket supports timeline calls and refuses a second live owner", { timeout: 15000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), "timeline-server-")); + const env = { ...process.env, DESKPAW_STATE_DIR: dir, DESKPAW_MEMORY_SOCKET: join(dir, "runtime", "memory.sock"), DESKPAW_MEMORY_DB: join(dir, "memory.sqlite"), DESKPAW_MEMORY_BUNDLES: join(dir, "bundles") }; + const server = spawn(process.execPath, ["src/server.js"], { env, stdio: ["ignore", "pipe", "pipe"] }); + const exited = once(server, "exit"); + const call = (method, params) => new Promise((resolve, reject) => { + const socket = createConnection(env.DESKPAW_MEMORY_SOCKET); let data = ""; + socket.setTimeout(2000, () => socket.destroy(new Error("timeout"))); + socket.on("error", reject); socket.on("connect", () => socket.write(JSON.stringify({ id: "test", method, params }) + "\n")); + socket.on("data", chunk => { data += chunk; }); + socket.on("end", () => { try { resolve(JSON.parse(data)); } catch (error) { reject(error); } }); + }); + try { + await Promise.race([once(server.stdout, "data"), exited.then(() => { throw new Error("Server exited before ready"); })]); + assert.equal((await call("timeline.append", { id: "meeting", kind: "user", at: 1, text: "报销会议", source: { sender: "owner" } })).ok, true); + assert.equal((await call("timeline.query", { query: "报销" })).result.entries[0].id, "meeting"); + const invalid = await call("timeline.read", { id: "missing" }); assert.equal(invalid.ok, false); assert.equal(invalid.id, "test"); + const second = spawn(process.execPath, ["src/server.js"], { env, stdio: "ignore" }); + assert.notEqual((await once(second, "exit"))[0], 0); + assert.equal((await call("health", {})).ok, true); + } finally { server.kill("SIGTERM"); await exited; rmSync(dir, { recursive: true, force: true }); } +}); diff --git a/test/timeline.test.js b/test/timeline.test.js new file mode 100644 index 0000000..568d739 --- /dev/null +++ b/test/timeline.test.js @@ -0,0 +1,29 @@ +import { DatabaseSync } from "node:sqlite"; +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { Timeline } from "../src/timeline.js"; + +test("timeline is idempotent, indexed across old history, bounded and expandable", () => { + const db = new DatabaseSync(":memory:"); const timeline = new Timeline(db); + try { + timeline.append({ id: "old", kind: "user", at: 1, text: "财务报销会议" + "材料".repeat(1000), source: { channel: "feishu", sender: "A" } }); + timeline.append({ id: "old", kind: "user", at: 2, text: "overwrite", source: {} }); + for (let i = 0; i < 250; i++) timeline.append({ id: `new-${i}`, kind: "assistant", at: i + 2, text: "other", source: {} }); + const result = timeline.query({ query: "报销", limit: 5 }); + assert.equal(result.entries[0].id, "old"); assert.equal(result.embedding, false); + assert.equal(result.entries[0].expandable, true); assert.equal(timeline.read({ id: "old" }).at, 1); + assert.equal(timeline.read({ id: "old", offset: 0, limit: 2 }).text, "财务"); + assert.equal(timeline.query({ from: 2, to: 2 }).entries.length, 1); + assert.throws(() => timeline.query({ limit: 999 })); + timeline.append({ id: "tail", kind: "user", at: 500, text: "padding ".repeat(2500) + "uniqueTail", source: {} }); + assert.equal(timeline.query({ query: "uniqueTail" }).entries[0].id, "tail"); + } finally { db.close(); } +}); +test("timeline excludes secrets and rejects unbounded data", () => { + const db = new DatabaseSync(":memory:"); const timeline = new Timeline(db); + try { + timeline.append({ id: "secret", kind: "user", at: 1, text: "password=example", source: {} }); + assert.equal(timeline.read({ id: "secret" }).text, "[敏感内容未写入记忆]"); + assert.throws(() => timeline.append({ id: "large", kind: "user", at: 1, text: "x".repeat(32001) })); + } finally { db.close(); } +}); diff --git a/test/topics.test.js b/test/topics.test.js index fcefcc1..e047193 100644 --- a/test/topics.test.js +++ b/test/topics.test.js @@ -85,9 +85,10 @@ test("facts require verbatim user evidence and skills remain drafts", () => with assert.equal(store.recall({ query: "简短回答" }).relevant_skills[0].skill_id, skill.skill_id); })); -test("nightly review has a fixed prior workday and does not rewrite raw interactions", () => withStore((store) => { +test("nightly review has a fixed prior workday and does not rewrite raw interactions", (t) => withStore((store) => { const eventAt = Date.parse("2026-09-22T12:00:00Z"); const reviewAt = Date.parse("2026-09-22T19:31:00Z"); + t.mock.method(Date, "now", () => reviewAt); store.rememberInteractions({ interactions: [ { interaction_id: "night", source: "terminal", user: "讨论一下话题", manager: "先记录", created_at: eventAt }, ] });