diff --git a/config/semantic.example.yaml b/config/semantic.example.yaml index 284d69a..be7984b 100644 --- a/config/semantic.example.yaml +++ b/config/semantic.example.yaml @@ -15,20 +15,31 @@ # 嵌入账本：沿用 HEL-857 系列唯一权威账（embed_ops 表）续账，绝不另起新账； # SDK 内部提取 LLM 记同库 llm_ops 新表（独立 cap，互不挪用） +# +# ── 累计上限算式（HEL-885，写入前照此换算，禁止直接填增量）─────────────── +# 权威账现状：已用 498 条 / 原批上限 500（其中冻结不动 2 条余额）。 +# 白栖知批准的 B 部分新增嵌入额度（例如 180 条）并入同一累计账： +# embed_cap = 500 + 180 = 680 # 原上限 + 新批增量（不是 180！） +# embed_reserve = 2 # 原冻结余额 2 条继续保留不动 +# 可用余量 = 680 − 498 − 2 = 180 +# 错误示例：embed_cap: 180（对已用 498 的账等于瞬间耗尽并掩盖历史）； +# 动到旧冻结 2 条（min_used: 498 会拒绝）。 ledger_path: /absolute/path/to/embed_ledger.sqlite3 embed_cap: 0 # 本产品环境允许使用的嵌入条数累计上限；0=未批准=全部拒绝 embed_daily: 0 # 嵌入条数日顶（执行表 C：建议 60/日，待批）；0=未批准=拒绝 -embed_reserve: 0 # 预留不动的余量（例如为复测保留 40） +embed_reserve: 0 # 预留不动的余量（例如为复测保留 40；冻结口径见上方算式） llm_cap: 0 # Mem0 内部提取 LLM 调用累计上限（执行表 B：建议 60，待批） llm_daily: 0 # Mem0 内部提取 LLM 日顶（执行表 C：建议 40/日，待批） -# 冻结前缀（HEL-882 防绕过）：任一 cap>0 时，ledger_path 必须仍是含冻结前缀的 -# 权威账——指向新空文件重开余额会被 verify_ledger_authority 拒绝。数值由总管 -# 按权威账现状填写（前 N 行逐行 SHA256 与最少已用条数）。 -# authority: -# prefix_rows: 181 -# prefix_sha256: "<权威账前 N 行逐行摘要>" -# min_used: 498 +# 冻结前缀（HEL-885 起为必填，不可注释掉）：只要 embed_cap / llm_cap 任一 >0， +# 缺少、留空或填错的 authority 块都会让 mem0 模式在构造前直接拒绝 +# （ledger_authority_required / ledger_authority_*，fail-closed，不降级）。 +# ledger_path 必须仍是含冻结前缀的权威账——指向新空文件重开余额同样被拒。 +# 数值由总管按权威账现状填写（前 N 行逐行 SHA256 与最少已用条数）： +authority: + prefix_rows: 181 + prefix_sha256: "<权威账前 N 行逐行摘要，总管填写>" + min_used: 498 # 记录作用域：add(user_id=) 与 search(filters={"user_id":…}) 必须一致； # mem0 2.2.1 拒绝无过滤 search，此键一般不需要改 diff --git a/deploy/hel874-release.md b/deploy/hel874-release.md index 4a90962..f768508 100644 --- a/deploy/hel874-release.md +++ b/deploy/hel874-release.md @@ -66,39 +66,61 @@ Mem0 提取都会在发送前被硬闸拒绝（`budget_*_exhausted`）——这 **当前事实（HEL-882 离线核实，tools/check_release_offline_hel882.py R6）：** 固定安装 hel857-venv 是 Python 3.10.12，产品发布 venv 是 3.11.16，两者 site-packages 不能混用；本机 uv/pip 缓存里没有 mem0ai/fastembed/qdrant-client -的 3.11 wheel，本卡不下载。因此发布卡要在 **B 部分批准后** 先落 wheel 再激活： +的 3.11 wheel，本卡不下载。因此发布卡要在 **B 部分批准后** 先落 wheel 再激活。 + +**HEL-885 命令勘误：旧版此处的 `uv pip download …` 在本机 uv（0.13.5）不存在 +（`uv pip download --help` 退出 2：unrecognized subcommand 'download'，已复测， +见 docs/rework/hel885/）。本机确实可用的落 wheel 工具是 pip 的 download 子命令 +（`python3 -m pip download --help` 退出 0）。发布 venv 由 uv 创建、不带 pip， +这不影响：下载在任一装有 pip 的 Python 上执行即可，wheel 按目标平台解析。** ```bash -# 4.1 在有内网源权限的环境把固定栈拉全（发布卡执行并留证；不在本卡执行） -# mem0ai==2.2.1（源码 2c81af3，tarball SHA256 9dcef930…dda23）及其依赖： -# qdrant-client（1.20.x）、fastembed（0.9.x）、openai、pytz、sqlmodel 等 +# 4.1 在有内网源权限的环境把固定栈拉全（发布卡执行并留证；不在本卡执行、 +# 不下载）。先核对目标机平台：uname -m 必须与 --platform 一致（当前 +# 部署机为 x86_64 Linux）。 +uname -m # 预期 x86_64；不符则改 --platform mkdir -p /opt/cybergril-hel874/wheels -uv pip download --python-version 3.11 -r deploy/mem0-stack.lock -d /opt/cybergril-hel874/wheels -# 4.2 同一发布进程离线并入（一个 venv 同时 import 产品与 mem0 —— 唯一被认可形态） +python3 -m pip download -r deploy/mem0-stack.lock -d /opt/cybergril-hel874/wheels \ + --only-binary :all: --implementation cp --python-version 3.11 \ + --platform manylinux2014_x86_64 --abi cp311 +# 4.1b 固定源身份与哈希留证：下载当时逐 wheel 记录 sha256（与发布评论一并 +# 存档；mem0ai 的源码身份 2c81af3 / tarball SHA256 见 deploy/mem0-stack.lock +# 注释——如需从源码 tarball 侧核对，比对 lock 注释中的值）。 +sha256sum /opt/cybergril-hel874/wheels/*.whl | tee /opt/cybergril-hel874/wheels/SHA256SUMS +# 4.2 同一发布进程离线并入（一个 venv 同时 import 产品与 mem0 —— 唯一被认可形态； +# uv pip install 是本机确实存在的子命令，--no-index 离线装） uv pip install --python .venv311/bin/python --no-index \ --find-links /opt/cybergril-hel874/wheels -r deploy/mem0-stack.lock .venv311/bin/python -c "import cybergril.api, mem0, qdrant_client, fastembed; print('co-import ok')" ``` -`deploy/mem0-stack.lock` 是本卡新交的固定清单（§4.1 的 `-r` 输入）。在此之前 -`CYBERGRIL_SEMANTIC` 保持 `stub`/`off`；**不得以 stub/FTS 降级包冒充用户的 -真实完整记忆体验**（体验交付必须等本节全部完成 + 新增预算批准）。 +**准确缺项与如实上报（HEL-885）：** 本机当前没有这些 3.11 wheel，本卡零下载， +`wheels/` 目录与 SHA256SUMS 均不存在——发布卡执行 4.1 时才能产生。若某个固定 +版本在 cp311/manylinux2014_x86_64 没有 wheel（`--only-binary :all:` 会直接报错 +指出包名），按 lock 注释如实上报缺项，不得换版本号冒充固定安装、不得改用 sdist +混装。4.2 的 co-import 是最终判据：任何一个 import 失败即未完成，不得以 stub/ +FTS 降级包冒充用户的真实完整记忆体验（体验交付必须等本节全部完成 + 新增预算 +批准）。发布前可先跑 `python3 tools/check_release_offline_hel882.py` 的 R7 +核验本节两条命令的工具面（pip download 存在、uv pip download 不存在）。 激活步骤（wheel 落好之后）： 1. `cp config/semantic.example.yaml config/semantic.yaml`，总管填批准值： - `embed_cap`/`llm_cap`（B 部分累计）、`embed_daily`/`llm_daily`（C 部分日顶）、 - `embed_reserve`；`ledger_path` 指向唯一权威 `embed_ledger.sqlite3` 续账 - （`llm_ops` 同库新表），并保留 `authority` 冻结前缀块——指向新的空文件 - 绕过 498/500 冻结会被 `verify_ledger_authority` 直接拒绝（HEL-882）。 + `embed_cap`/`llm_cap`（B 部分累计，换算算式见示例文件与执行表 §2—— + 500+增量，不是把增量直接填进累计上限）、`embed_daily`/`llm_daily` + （C 部分日顶）、`embed_reserve`；`ledger_path` 指向唯一权威 + `embed_ledger.sqlite3` 续账（`llm_ops` 同库新表），并**填写** `authority` + 冻结前缀块（HEL-885 起必填：缺/空/错一律在构造前拒绝 + `ledger_authority_required`，指向新空文件绕过 498/500 冻结同样被拒）。 2. `CYBERGRIL_SEMANTIC=mem0` 重启。任何未批调用 fail-closed （`semantic_*_budget_not_approved/exhausted/daily_exhausted`）；SDK 内部 提取 LLM 同时过统一请求闸（`semantic` 类别 + 日 token 池，发送前预占）。 ## 5. 停用与保留数据回退 -- 停语义层：`CYBERGRIL_SEMANTIC=off`（或 stub）重启；`sem_batch/sem_covered` - 按后端区分保留，重开续跑不重发已完成批次，stub 批次不阻塞真实重建。 +- 停语义层：`CYBERGRIL_SEMANTIC=off`（或 stub）重启；`sem_batch/sem_covered/ + sem_row_done` 按后端区分保留，重开续跑不重发已完成行（逐行锚点+逐行完成 + 记录，HEL-885），stub 批次不阻塞真实重建。 - 停主动联系/生活：`CYBERGRIL_R2_ENABLED=0`。 - 整实例回退：`systemctl stop cybergril-hel874` → 数据目录 `/opt/cybergril-hel874/data/` 原样保留（聊天/记忆/生活/情绪/关系全量 + diff --git a/docs/budget-plan-hel874.md b/docs/budget-plan-hel874.md index 9f84db0..114d73a 100644 --- a/docs/budget-plan-hel874.md +++ b/docs/budget-plan-hel874.md @@ -38,6 +38,11 @@ HEL-857 系列冻结余额（真实请求 39/60、嵌入 498/500、HEL-835 闸 哪些已推理，整批保守落账。 - **Mem0 内部 LLM**：独立 `llm_ops` 累计账（B）+ `llm_daily` 日顶（C）+ 统一 请求闸（`semantic` 类别 + 日 token 池，发送前预占）。 +- **未知预占不退款（HEL-885）**：`reserve:` 预占行不再按一小时超时清除—— + 已发送后崩溃未 settle 的消费事实必须保留；唯一删除路径是可证明未发送的 + 显式 `release()`（如 httpx.ConnectError 级未连接）。作业租约过期不是 + 「未发生」的证明。跨日结算/释放一律按预占凭据的 `hold.day` 记回原日， + 不冲新一天、不产生负用量（`token_usage`/`request_usage` 同口径）。 ## 1. A 部分 · 一次性团队整体检查（隔离空库，非用户档案） @@ -71,12 +76,31 @@ HEL-857 系列冻结余额（真实请求 39/60、嵌入 498/500、HEL-835 闸 | 嵌入（提取批 + 实体 + 查询） | ≤ 180 条（`embed_cap`） | | 对话/生活/判定等其余 | 并入 C 部分日账 | +**累计 cap / 保留量算式（HEL-885，写入 config 前照此换算，禁止把增量直接填进累计上限）：** + +``` +权威账现状 ：已用 498 条，原批上限 500，冻结不动余额 2 条 +新批增量 ：白栖知批准的 B 部分嵌入条数（示例 180） +embed_cap = 500 + 180 = 680 # 原批上限 + 新批增量（≠ 180） +embed_reserve= 2 # 原冻结余额继续保留，不得消耗 +可用余量 = 680 − 498 − 2 = 180 +authority = {prefix_rows: <权威账前N行>, prefix_sha256: <逐行摘要>, + min_used: 498} # HEL-885 起任一 cap>0 时必填，缺/空/错即拒绝构造 +``` + +错误示例：对已用 498 的账把 `embed_cap` 填成 180（等于瞬间耗尽并掩盖历史用量）； +以任何方式消耗旧冻结 2 条（`min_used: 498` 保证拒绝）。`authority` 块缺省或 +整段注释时，mem0 模式在构造前 fail-closed（`ledger_authority_required`）， +不再是可以跳过的可选项。 + - 硬闸：`embed_cap`/`llm_cap`（累计）+ `embed_daily`/`llm_daily`（日顶）写入 config/semantic.yaml（默认全 0=拒绝）；预占-落账事务制保证并发不越 cap； `embed_reserve` 预留复测余量；`authority` 冻结前缀块拒绝新文件绕过 498/500。 - 碰上限即停：作业批回到 `queued`（reason=`semantic_*_budget_*`），**不耗重试 - 次数、不丢行**——总管调 cap 后同一批续跑，已入库的行不再重发（向量库级 - 幂等）；检索返回空块、对话回落词面 FTS，主对话不中断；不自动翻页。 + 次数、不丢行**——总管调 cap 后同一批续跑，已完成的行不再重发（向量库逐行 + 锚点 + `sem_row_done` 逐行完成记录，HEL-885：每个向量只证明自己那一行， + 空提取行同样有完成证明）；检索返回空块、对话回落词面 FTS，主对话不中断； + 不自动翻页。 - SDK 吞错已堵住：中途嵌入被预算挡住时 mem0 返回空列表，产品现在识别为预算 拒绝并重排队（不再误写 done/[] 造成永久漏整理）。 - 授权来源：**需白栖知明确批准**（提高嵌入上限或另批独立账；现有 498/500 diff --git a/docs/requirements/handoff-HEL-885.json b/docs/requirements/handoff-HEL-885.json new file mode 100644 index 0000000..2493a9d --- /dev/null +++ b/docs/requirements/handoff-HEL-885.json @@ -0,0 +1,51 @@ +{ + "baseline_version": "1.11", + "read_commit": "59fb06304fb9f230837b43e431e2a1cc303e16c9", + "task": "HEL-885 返工（总管对 HEL-882 59fb063 独立复审五项缺口的一次归拢修正）：逐行恢复与真实来源一致（每次 add 只带自己那一行的锚点，向量只证明自身行；空提取行落 sem_row_done 完成证明；预算失败/崩溃重启/取消租约/并发不漏行、不伪完成、不重发完成行）、跨日结算按预占日记回原日并维护累计账（不再冲出新日负用量）、未知已发送预占永不超时退款（reserve 行唯一删除路径是可证明未发送的 release，llm/embed 同路径）、正 cap 必须验证权威来源（authority 缺/空/错在构造前 fail-closed，示例配置必填并给出 498已用+冻结2+新批增量的累计 cap/保留量算式）、发布依赖补齐命令改为本机确实支持的 python3 -m pip download 并入 R7 离线探针；零真实调用、零真实嵌入、零非回环出站、不部署、不合并、不改视觉、不碰权威三账", + "requirement_ids": [ + "CG-006", + "CG-008", + "CG-011", + "CG-013", + "CG-014", + "CG-015", + "CG-016", + "CG-021", + "AM-001", + "AM-002", + "AM-003", + "UX-004", + "UX-006", + "GOV-001", + "GOV-002", + "GOV-003", + "GOV-005", + "GOV-006", + "DEL-001" + ], + "deliverable": "在统一候选母线 hel-870-assembly（基于 59fb063 追加，PR#11 只追加不强推）上完成五项修正：1) 逐行进度与来源证明——Mem0Backend/StubBackend add_group 改为逐行发送、每次 add 的 metadata 只带该行锚点（turn_ids/turn_roles/input_sha256 均单行），向量存在只证明自身行；新增 sem_row_done 表（r4.sql 增量迁移）在每行 add 完成瞬间于 worker 线程内落持久完成记录（合法空提取=零向量也有完成证明），tick 恢复时以 向量库锚点∪sem_row_done 为跳过集，批次 entry_ids 跨部分尝试合并；吞错检测改逐行口径。2) 跨日账——PersistentBudget.settle/release 一律写 hold.day（token_usage/request_usage 按预占日，budget_total 全局），不再向 state.day 冲账制造负用量；另修 budget_total 空表时 total_requests_used/total_tokens_used 的 TypeError 崩溃（coalesce 子查询），旧用例靠该崩溃“通过”的假绿改为如实入账。3) 未知预占——_atomic_reserve 删除 1 小时过期 DELETE（reserve 行不再按龄清除），未知消费保守计费、跨重启保留，唯一删除路径是可证明未发送的显式 release()；embed/llm 共用路径同修。4) 权威来源必填——Mem0Backend 在 import mem0 之前：任一 cap>0 而 authority 缺/空/非 dict 即抛 ledger_authority_required（构造前拒绝，不发任何请求），错的 authority 仍被 verify_ledger_authority 前缀校验拒绝；config/semantic.example.yaml authority 段改为必填示例并附累计算式（embed_cap=500+增量、embed_reserve=2、min_used=498，禁把 180 直接填成累计 cap、禁动旧冻结 2 条）；docs/budget-plan-hel874.md §2 补同套算式与 HEL-885 记账口径。5) 发布命令——deploy/hel874-release.md §4 改为 python3 -m pip download（--only-binary :all: --platform manylinux2014_x86_64 --python-version 3.11 --implementation cp --abi cp311）+ sha256 留证 + uv pip install --no-index --find-links 离线并入 + 同进程 co-import 判据，如实列明本机无 3.11 wheel、本卡零下载、无包则如实上报不换版本；tools/check_release_offline_hel882.py 新增 R7 双探针（pip download 存在 rc=0、已删除的 uv pip download 保持不存在）。验证：3.11.16 venv+requirements.lock 离线 281 项 pytest 全过（新增 tests/test_rework_hel885.py 9 例：部分批次只认已发行/空提取不重发/进程重启只补缺行/并发 tick 不双发/跨日结算与释放按预占日/未知预占 llm+embed 不退款/缺空错 authority fail-closed；test_resume_never_loses_or_duplicates_rows 重写为真实产品路径，不再手工植入单行锚点），ruff 0 错（src/tests/tools），web tsc+42 vitest+build（npm ci --offline），固定 mem0 2.2.1 安装确定性替身：verify_mem0_real_hel874.py 12/12、新增 verify_semantic_recovery_hel885.py 7/7（S0 缺 authority 拒绝/S1 逐行部分/S2 重启只补缺行以 HTTP 计数证明/S3 租约回收+并发不双发/S4 未知预占不退款），总管 review_hel882_{partial,extra}.py 两脚本原样复跑全过（证据 docs/rework/hel885/），发布离线检查 R1–R7 全过；零真实调用、零真实嵌入、零非回环出站，权威三账与冻结余额未写，未部署未合并未交体验地址，不改视觉。", + "acceptance": [ + "总管 partial 复现修复：首 tick 后 completed_message_ids=[2]（仅伴侣行，向量逐行锚点），补 cap 后 done、entry_ids 2 条、回环 HTTP=2、用户原话确实发出且每行恰一次（manager-review-partial-rerun.json 原样脚本复跑）", + "总管 extra 复现修复：跨日结算次日 used=0、估算 150 被拒、原日记实际 10；未知预占 second_allowed=false、used 保持 1（manager-review-extra-rerun.json 原样脚本复跑）", + "逐行恢复真实产品路径：空提取行有 sem_row_done 完成证明且不再重发；进程重启后只补缺行；并发 tick 不双发（tests/test_rework_hel885.py + recovery-real-sdk.json S1–S3）", + "未知预占保守计费：llm_ops 与 embed_ops 同路径，一小时前的未结算预占跨重启仍占额度，唯一删除路径是显式可证明未发送的 release（S4a/S4b + 对应用例）", + "authority 必填：缺/空/错 authority 在构造前拒绝（ledger_authority_required / 前缀不匹配），示例配置必填并附 500+增量、reserve=2、min_used=498 算式，明确禁把 180 填成累计 cap、禁动旧冻结 2 条（tests + recovery-real-sdk.json S0 + semantic.example.yaml + budget-plan-hel874.md）", + "发布命令真实可用：§4 改为 python3 -m pip download（本机实测 rc=0），R7 双探针入离线检查；本机无 3.11 wheel、零下载、缺包如实上报的口径不变（release-offline-hel885.json）", + "工程检查：281 pytest、ruff 0（src/tests/tools）、web typecheck/42 vitest/build、双目标需求检查 4/4（含本 handoff）；固定 SDK 合同 12/12 复跑通过" + ], + "deferred": [ + "发布与体验地址：由总管独立复审本 head 后交发布卡按 deploy/hel874-release.md 执行（8360 隔离实例）；本卡未部署未合并 PR#11 未交体验地址", + "真实语义激活（CYBERGRIL_SEMANTIC=mem0）：需 B 部分预算批准 + 3.11 mem0 栈 wheel 落位（按 §4 新命令执行并留 sha256 证）；未激活前 stub/FTS 不冒充真实记忆体验", + "A/B/C 执行表数值仍为待批建议：批准前所有 cap 保持 0=拒绝；authority 前缀实际值由总管按权威账现状填写；新增预算由总管向白栖知一次性申请", + "R5 语音/R6 天气、素材、备份/R7 有条件观察承接不变；视觉仍交原作者，本卡未改视觉", + "真实体验效果待体验地址交付后由白栖知判断" + ], + "requirement_changes": [], + "verification": [ + "分支 hel-870-assembly 追加提交（最终 head 见交付评论），PR#11 只追加未强推未合并；视觉总监分支未覆盖", + "python 3.11.16 venv + requirements.lock 离线：281 项测试全过（新增 tests/test_rework_hel885.py 9 例，重写 test_resume_never_loses_or_duplicates_rows）；ruff 0 错（src/tests/tools）；web typecheck/vitest 42/build 通过（npm ci --offline）", + "固定 mem0 2.2.1 安装（hel857-venv 3.10.12）确定性替身实跑：tools/verify_mem0_real_hel874.py 12/12；新增 tools/verify_semantic_recovery_hel885.py 7/7（零真实嵌入、零非回环）；总管 review_hel882_{partial,extra}.py 原样复跑证据 docs/rework/hel885/manager-review-{partial,extra}-rerun.json", + "tools/check_release_offline_hel882.py（echo 替身、随机回环端口、临时目录）R1–R7 全过；mem0 同进程导入缺项（3.11 无 wheel）如实记录，解决路径 deploy/hel874-release.md §4 新命令，本卡零下载", + "python3 scripts/check_requirements.py --base-ref origin/main [--handoff docs/requirements/handoff-HEL-885.json] 与 --base-ref origin/agent/agent/de8874a3803b [--handoff …] 四项通过" + ] +} diff --git a/docs/rework/hel885/README.md b/docs/rework/hel885/README.md new file mode 100644 index 0000000..07c9e34 --- /dev/null +++ b/docs/rework/hel885/README.md @@ -0,0 +1,43 @@ +# HEL-885 · 总管独立复审五项修正证据 + +读取基线：`59fb06304fb9f230837b43e431e2a1cc303e16c9`（HEL-882 最终 head， +总管独立复审「暂不通过发布审核」）。本目录证据均在修正后的代码上实际执行， +边界与 HEL-882 一致：确定性 512 维向量替身（TextEmbedding 构造前安装）、 +socket 拒绝非回环、127.0.0.1 假网关、临时账/库，零真实 LLM、零真实嵌入、 +零下载、零权威三账写入。 + +## 修正对照（复现 → 修复后实测） + +| # | 总管复现（59fb063） | 修复后实测 | 证据 | +| --- | --- | --- | --- | +| 1 | 整批 send_anchor 复制给每次逐行 add；首向量即宣称两行完成；补 cap 后 done/[]、HTTP 仍 1、用户原话从未发出 | 每向量逐行锚点；首 tick 后 completed=[2]；补 cap 后 done 且 entry_ids 2 条、HTTP=2、用户原话确实发出（每行恰一次） | manager-review-partial-rerun.json（脚本原样复跑） | +| 2 | settle 用 state.day：次日 used=-80，估算 150 放行 | 结算记回预占日：次日 used=0，150 被拒，原日记实际 10 | manager-review-extra-rerun.json（脚本原样复跑） | +| 3 | reserve:* 一小时 DELETE：未知预占被删，第二次 reserve 又放行 | 未知预占保留：second_allowed=false，used 保持 1（llm/embed 同路径、跨重启） | manager-review-extra-rerun.json + recovery-real-sdk.json S4a/S4b | +| 4 | 缺 authority 时跳过校验：新空账+正 cap 成功构造并发送 | 缺/空/错 authority 一律构造前拒绝 ledger_authority_required；示例配置 authority 必填+算式 | recovery-real-sdk.json S0 + tests/test_rework_hel885.py | +| 5 | `uv pip download` 本机不存在（rc=2） | 文档改用 `python3 -m pip download`（rc=0 实测）；R7 双探针入离线检查 | release-offline-hel885.json R7 | + +## 文件 + +- `review_hel882_partial.py` / `review_hel882_extra.py`：总管原样脚本（附件同版， + 自 hel-882-bcbb63d27399 工作目录复制），对本 checkout 复跑。 +- `manager-review-partial-rerun.json` / `manager-review-extra-rerun.json`： + 原样脚本复跑结果（head 字段仍为脚本内写定的 59fb063 字符串，实际执行于本 + 修正后的 checkout；JSON 内容即修复后行为）。 +- `recovery-real-sdk.json`：tools/verify_semantic_recovery_hel885.py（新增）—— + 固定 mem0ai 2.2.1 安装上 S0 缺 authority 拒绝 / S1 逐行部分完成 / S2 进程重启 + 只补缺行（HTTP 计数证明，不依赖文本出现次数：mem0 提取提示会合法携带旧行 + 滚动历史）/ S3 租约回收与并发 tick 不双发 / S4a/S4b 未知预占不退款。 +- `real-sdk-contract.json`：tools/verify_mem0_real_hel874.py 复跑 12/12 + （make_ledger/config_for 现带合成 authority 前缀，产品默认未被放宽）。 +- `release-offline-hel885.json`：tools/check_release_offline_hel882.py 全过， + 新增 R7（pip download 存在 rc=0；已删除的 uv pip download 保持不存在）。 + +## 回归 + +- 3.11.16 venv + requirements.lock 离线：281 项 pytest 全过（新增 + tests/test_rework_hel885.py 9 例；test_resume_never_loses_or_duplicates_rows + 重写为真实产品路径，不再手工植入单行锚点）；ruff 0 错（src/tests/tools）； + web tsc + 42 vitest + build（npm ci --offline）。 +- 发现并修复一处被旧测试掩盖的崩溃：budget_total 空表时 + total_tokens_used() 抛 TypeError（旧用例靠该崩溃“通过”）——现 + coalesce+空行处理，用例改为如实入账一次性总量。 diff --git a/docs/rework/hel885/manager-review-extra-rerun.json b/docs/rework/hel885/manager-review-extra-rerun.json new file mode 100644 index 0000000..269057e --- /dev/null +++ b/docs/rework/hel885/manager-review-extra-rerun.json @@ -0,0 +1,33 @@ +{ + "head": "59fb06304fb9f230837b43e431e2a1cc303e16c9", + "real_llm_requests": 0, + "real_embeddings": 0, + "midnight_settlement": { + "daily_cap": 100, + "old_hold": 90, + "actual": 10, + "new_day_used_before_next_request": 0, + "request_estimate": 150, + "over_cap_request_allowed": false, + "rows": [ + { + "day": "2026-10-26", + "scene": "dialogue_main", + "used": 10 + } + ] + }, + "aged_unknown_reservation": { + "cap": 1, + "used_before": 1, + "second_allowed": false, + "first_hold_id": 1, + "second_hold_id": null, + "used_after": 1 + }, + "release_documented_download_command": { + "command": "uv pip download --help", + "returncode": 2, + "stderr": "error: unrecognized subcommand 'download'\n\nUsage: uv pip [OPTIONS] \n\nFor more information, try '--help'." + } +} \ No newline at end of file diff --git a/docs/rework/hel885/manager-review-partial-rerun.json b/docs/rework/hel885/manager-review-partial-rerun.json new file mode 100644 index 0000000..e8776de --- /dev/null +++ b/docs/rework/hel885/manager-review-partial-rerun.json @@ -0,0 +1,48 @@ +{ + "head": "59fb06304fb9f230837b43e431e2a1cc303e16c9", + "real_llm_requests": 0, + "real_embeddings": 0, + "boundary": "deterministic 512 vectors, non-loopback forbidden", + "missing_authority_constructed_with_positive_caps": false, + "public_batch_enqueued": 1, + "batch_after_first_tick": { + "state": "queued", + "attempts": 0, + "reason": "LLM extraction failed: semantic_llm_budget_exhausted", + "entry_ids": null + }, + "loopback_requests_before_resume": 1, + "stored_points_after_first": [ + { + "src_group": "chat_public:mem0:synthetic:1bc8667410", + "turn_ids": [ + "synthetic" + ], + "turn_roles": [ + [ + "synthetic", + "companion", + 2 + ] + ], + "input_sha256": "840dde45eec07416f21c56cefb0f93f4e71af3edc3100c332f84bb1266637862", + "user_id": "cybergril", + "data": "合成提取记忆HEL874-1", + "text_lemmatized": "合成提取记忆HEL874-1", + "hash": "69195446402294f6e77f8ee993ce3456", + "created_at": "2026-10-26T15:56:46.108106+00:00", + "updated_at": "2026-10-26T15:56:46.108106+00:00" + } + ], + "claimed_completed_message_ids": [ + 2 + ], + "batch_after_resume": { + "state": "done", + "attempts": 1, + "reason": null, + "entry_ids": "[\"2bc5a040-cd80-47a6-b865-c3b3edb7c711\", \"fe98487a-e3f4-4633-abda-2680544a9aaf\"]" + }, + "loopback_requests_after_resume": 2, + "unsent_user_row_actually_sent": true +} \ No newline at end of file diff --git a/docs/rework/hel885/real-sdk-contract.json b/docs/rework/hel885/real-sdk-contract.json new file mode 100644 index 0000000..05bb91f --- /dev/null +++ b/docs/rework/hel885/real-sdk-contract.json @@ -0,0 +1,172 @@ +{ + "started": "2026-10-26T15:58:41Z", + "embedding_boundary": "deterministic 512-dimensional vectors installed before TextEmbedding construction; no model loaded; zero real embeddings", + "external_network": "socket.connect rejects all non-loopback hosts", + "install": { + "python": "3.10.12", + "mem0ai": "2.2.1", + "qdrant-client": "1.20.0", + "fastembed": "0.9.1", + "mem0_path": "/home/xiaomo/multica_workspaces/hello-af23464d222e/hel-857-2cf6dc7dfc87/workdir/hel857-venv/lib/python3.10/site-packages/mem0", + "module_sha256": { + "memory/main.py": "f329ca4a17af1fbefb32ac10d4cc7ae7ddbdb7435a87eb74c30915cff98b8b7c", + "memory/storage.py": "5f5ecdadd2b260fd14f36a0f338336eb760275465dd1056a99a47e1ef07702bd", + "embeddings/fastembed.py": "3ec2ae87df5d0bebedb8b5471366c3412975847458499f66097f7b8c9ac8c9b8", + "llms/openai.py": "37e6d3c22b55df0731740f4e0868b95bd480ce44363227fe53461db328c5f67e", + "vector_stores/qdrant.py": "da570579d5cad69e6f123fdfe32b725235a25604e70047666bec763ba6e2bf43" + }, + "source_tarball_sha256": "9dcef9303ebcec116eec6739f01b8b35bd24459de6a569686b4d19b9412dda23" + }, + "adapter_has_llm_guard": true, + "checks": [ + { + "name": "C1_construct", + "ok": true, + "detail": "client_path=sdk-vector-store-client" + }, + { + "name": "C2_add_anchor_payload", + "ok": true, + "detail": "created=True payload_keys=['created_at', 'data', 'hash', 'input_sha256', 'src_group', 'text_lemmatized', 'turn_ids', 'turn_roles']" + }, + { + "name": "C2_add_user_scope", + "ok": true, + "detail": "user_id present in stored payload" + }, + { + "name": "C2_add_ledger_shape", + "ok": true, + "detail": "embed_items=3 llm_calls=1" + }, + { + "name": "C3_extraction_input_set", + "ok": true, + "detail": "old_memory=True new_row=True history_carry=True" + }, + { + "name": "C3b_unshared_canary_never_outbound", + "ok": true, + "detail": "queued_public_batches=2 leaked=False anchor_taint=False" + }, + { + "name": "C4_search_filters", + "ok": true, + "detail": "hits=3" + }, + { + "name": "C4_search_single_debit", + "ok": true, + "detail": "embed_delta=1 (pre-fix=2)" + }, + { + "name": "C6_concurrency_cap", + "ok": true, + "detail": "outcomes=['ok', 'refused'] used=3 cap=3" + }, + { + "name": "C7_reopen_after_close", + "ok": true, + "detail": "hits_after_reopen=3" + }, + { + "name": "C5_embed_zero_refuses", + "ok": true, + "detail": "err=semantic_embed_budget_not_approved llm_sent=0" + }, + { + "name": "C5_llm_zero_refuses", + "ok": true, + "detail": "err=semantic_llm_budget_not_approved llm_sent=0" + } + ], + "embed_rows_after_first_add": [ + { + "kind": "authority_prefix", + "items": 0, + "note": "authority:synthetic-verify-prefix" + }, + { + "kind": "dense", + "items": 1, + "note": "product:search" + }, + { + "kind": "dense", + "items": 1, + "note": "product:batch:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:add" + } + ], + "final_embed_rows": [ + { + "kind": "authority_prefix", + "items": 0, + "note": "authority:synthetic-verify-prefix" + }, + { + "kind": "dense", + "items": 1, + "note": "product:search" + }, + { + "kind": "dense", + "items": 1, + "note": "product:batch:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:search" + }, + { + "kind": "dense", + "items": 1, + "note": "product:batch:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:search" + }, + { + "kind": "dense", + "items": 1, + "note": "product:batch:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:add" + }, + { + "kind": "dense", + "items": 1, + "note": "product:search" + }, + { + "kind": "dense", + "items": 1, + "note": "product:search" + } + ], + "final_llm_used": 3, + "finished": "2026-10-26T15:58:42Z", + "gateway_requests_total": 3, + "failures": [], + "temp_dir": "/tmp/multica-task-4169170952/hel874-real-sdk-zbdu5fkf" +} \ No newline at end of file diff --git a/docs/rework/hel885/recovery-real-sdk.json b/docs/rework/hel885/recovery-real-sdk.json new file mode 100644 index 0000000..51694ae --- /dev/null +++ b/docs/rework/hel885/recovery-real-sdk.json @@ -0,0 +1,45 @@ +{ + "head": "HEL-885 rework", + "real_llm_requests": 0, + "real_embeddings": 0, + "boundary": "deterministic 512 vectors, non-loopback forbidden", + "checks": [ + { + "name": "S0_positive_caps_without_authority_fail_closed", + "ok": true, + "detail": "missing=ledger_authority_required empty=ledger_authority_required" + }, + { + "name": "S1_p1_first_tick_partial", + "ok": true, + "detail": "state=queued reason=LLM extraction failed: semantic_llm_budget_exhausted http=1 completed=[2] points=1 per_row_anchor=True" + }, + { + "name": "S2_p1_restart_sends_only_missing_row", + "ok": true, + "detail": "state=done http=1 user_sent=True completed=[1, 2] entry_ids=[\"0419c0c8-a6b1-4a24-864b-295658e23b19\", \"3d1de06a-bc8c-4f4b-9914-049ca13edb5e\"]" + }, + { + "name": "S3a_lease_expired_batch_recovers", + "ok": true, + "detail": "recovered={'state': 'failed', 'reason': 'lease_expired'} state_after=done http=2" + }, + { + "name": "S3b_concurrent_ticks_single_send", + "ok": true, + "detail": "http=2 states=['done']" + }, + { + "name": "S4a_aged_unknown_llm_reservation_kept", + "ok": true, + "detail": "second_allowed=False used_after_restart=1" + }, + { + "name": "S4b_aged_unknown_embed_reservation_kept", + "ok": true, + "detail": "second_allowed=False used=1 after_release=0" + } + ], + "gateway_requests_total": 6, + "failures": [] +} \ No newline at end of file diff --git a/docs/rework/hel885/release-offline-hel885.json b/docs/rework/hel885/release-offline-hel885.json new file mode 100644 index 0000000..ce18e6e --- /dev/null +++ b/docs/rework/hel885/release-offline-hel885.json @@ -0,0 +1,86 @@ +{ + "started": "2026-10-26T15:58:29Z", + "venv_python": "/home/xiaomo/multica_workspaces/hello-af23464d222e/hel-874-db7635e6b65e/workdir/CyberGril/.venv311/bin/python", + "venv_python_version": "3.11.16", + "checks": [ + { + "name": "R1_checkout_fresh_dir", + "ok": true, + "detail": "cloned tracked source into /tmp/multica-task-4169170952/hel882-release-qd9sq7nv/cybergril-hel874/current" + }, + { + "name": "R2_service_imports_product", + "ok": true, + "detail": "" + }, + { + "name": "R6_mem0_coimport_status_recorded", + "ok": true, + "detail": "import mem0 → ModuleNotFoundError: No module named 'mem0' (resolution path in deploy/hel874-release.md §4; no download in this check)" + }, + { + "name": "R7_pip_download_command_exists", + "ok": true, + "detail": "/usr/bin/python3 -m pip download --help → rc=0" + }, + { + "name": "R7_removed_uv_pip_download_still_absent", + "ok": true, + "detail": "uv pip download --help → rc=2 (kept absent on purpose; do not re-document it)" + }, + { + "name": "R2_service_serves_health", + "ok": true, + "detail": "version=R4-assembly semantic=stub" + }, + { + "name": "R3_create_status_recorded", + "ok": true, + "detail": "create status=201（409=前次运行的档案仍在，也是保留数据的表现）" + }, + { + "name": "R3_dialogue_refused_honestly", + "ok": true, + "detail": "http=200 events=b'data: {\"t\": \"chat.error\", \"epoch\": 1, \"turn_id\": \"00000000-0000-4000-8000-000000000001\", \"phi\": \"\\xe5\\xa5\\xb9\\xe8\\xbf\\x99\\xe4\\xbc\\x9a\\xe5\\x84\\xbf\\xe6\\xb2\\xa1\\xe6\\x8e\\xa5\\xe4\\xb8\\x8a\\xe8\\xaf'" + }, + { + "name": "R4_service_restarts_on_same_data", + "ok": true, + "detail": "epoch=2" + }, + { + "name": "R5_epoch_bumped_on_reopen", + "ok": true, + "detail": "1 → 2" + }, + { + "name": "R4_companion_created_offline", + "ok": true, + "detail": "status=409" + }, + { + "name": "R4_dialogue_stream_http", + "ok": true, + "detail": "status=200 events=b'data: {\"t\": \"chat.delta\", \"epoch\": 2, \"turn_id\": \"00000000-0000-4000-8000-000000000002\", \"delta\": \"\\xe6\\x89\\xb9\"}\\n\\ndata: {\"t\": \"c'" + }, + { + "name": "R4_messages_roundtrip", + "ok": true, + "detail": "roles=['user', 'user', 'companion']" + }, + { + "name": "R5_data_dir_preserved", + "ok": true, + "detail": "['companion.db']" + }, + { + "name": "R5_companion_row_survives_restart", + "ok": true, + "detail": "companion rows=1" + } + ], + "mem0_import_in_release_venv": "ModuleNotFoundError: No module named 'mem0'", + "removed_uv_pip_download_rc": 2, + "failures": [], + "finished": "2026-10-26T15:58:34Z" +} \ No newline at end of file diff --git a/docs/rework/hel885/review_hel882_extra.py b/docs/rework/hel885/review_hel882_extra.py new file mode 100644 index 0000000..92a92f5 --- /dev/null +++ b/docs/rework/hel885/review_hel882_extra.py @@ -0,0 +1,60 @@ +"""Independent regressions, temporary ledgers, no provider or remote access.""" +import json +import sqlite3 +import subprocess +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +sys.path.insert(0, str(ROOT / 'CyberGril/src')) +from cybergril.life.model import PersistentBudget +from cybergril.memory.store import Store +from cybergril.memory.semantic import LlmLedgerGuard +from cybergril.think.budget import BudgetConfig, BudgetGuard + +result = {'head': '59fb06304fb9f230837b43e431e2a1cc303e16c9', + 'real_llm_requests': 0, 'real_embeddings': 0} +with tempfile.TemporaryDirectory(dir=ROOT) as td: + tmp = Path(td) + store = Store(tmp / 'midnight.db') + guard = PersistentBudget(store, BudgetGuard(BudgetConfig( + daily_tokens=100, daily_requests={'dialogue': 10}))) + guard._roll = lambda: None + guard.state.day = '2026-10-26' + hold = guard.acquire('dialogue_main', tokens=90) + guard.state.day = '2026-10-27' + guard.settle(hold, actual_tokens=10) + used = guard.used + try: + guard.acquire('dialogue_main', tokens=150) + allowed = True + except RuntimeError: + allowed = False + result['midnight_settlement'] = { + 'daily_cap': 100, 'old_hold': 90, 'actual': 10, + 'new_day_used_before_next_request': used, + 'request_estimate': 150, 'over_cap_request_allowed': allowed, + 'rows': [dict(r) for r in store.query('SELECT * FROM token_usage')]} + store.close() + ledger = tmp / 'crash.db' + llm = LlmLedgerGuard(ledger, cap=1) + pending = llm.reserve(1, 'sent_but_crashed_before_settlement') + with sqlite3.connect(ledger) as conn: + conn.execute("UPDATE llm_ops SET ts='2000-01-01T00:00:00Z' WHERE id=?", (pending,)) + before = llm.used() + try: + second = llm.reserve(1, 'second_send_after_restart') + allowed = True + except RuntimeError: + second, allowed = None, False + result['aged_unknown_reservation'] = { + 'cap': 1, 'used_before': before, 'second_allowed': allowed, + 'first_hold_id': pending, 'second_hold_id': second, + 'used_after': llm.used()} + probe = subprocess.run(['uv','pip','download','--help'], capture_output=True, text=True) + result['release_documented_download_command'] = { + 'command': 'uv pip download --help', 'returncode': probe.returncode, + 'stderr': probe.stderr.strip()} +(ROOT / 'HEL-882-extra-independent.json').write_text(json.dumps(result, ensure_ascii=False, indent=2)) +print(json.dumps(result, ensure_ascii=False, indent=2)) diff --git a/docs/rework/hel885/review_hel882_partial.py b/docs/rework/hel885/review_hel882_partial.py new file mode 100644 index 0000000..11d7b3d --- /dev/null +++ b/docs/rework/hel885/review_hel882_partial.py @@ -0,0 +1,68 @@ +"""Actual pinned SDK; deterministic vectors and loopback HTTP only.""" +import asyncio +import importlib.util +import json +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +REPO = ROOT / 'CyberGril' +sys.path.insert(0, str(REPO / 'src')) +spec = importlib.util.spec_from_file_location('contract', REPO / 'tools/verify_mem0_real_hel874.py') +contract = importlib.util.module_from_spec(spec) +spec.loader.exec_module(contract) +contract.install_deterministic_boundary() +from cybergril.memory.semantic import Mem0Backend, SemanticMemory +from cybergril.memory.store import Store + +async def main(): + result = {'head': '59fb06304fb9f230837b43e431e2a1cc303e16c9', + 'real_llm_requests': 0, 'real_embeddings': 0, + 'boundary': 'deterministic 512 vectors, non-loopback forbidden'} + with tempfile.TemporaryDirectory(dir=ROOT) as td: + tmp = Path(td) + server = contract.start_gateway() + backend = None + try: + ledger = tmp / 'ledger.db' + contract.make_ledger(ledger, cap=100, llm_cap=1) + cfg = contract.config_for(tmp, ledger, port=server.server_address[1], cap=100, llm_cap=1) + backend = Mem0Backend(cfg) + result['missing_authority_constructed_with_positive_caps'] = 'authority' not in cfg + store = Store(tmp / 'product.db') + with store.conn: + store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" + "(1,'user','合成用户原话甲','synthetic','sent')") + inserted = store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" + "(2,'companion','合成已公开伴侣原话乙','synthetic','sent')") + store.execute("INSERT INTO sharing_attempt(message_id,turn_id,epoch,token,text_hash,expires_at," + "ack_at,candidates,state) VALUES(?,'synthetic',1,'synthetic','synthetic',999999,3,'[]','judged')", + (inserted.lastrowid,)) + live = SemanticMemory(store, backend) + result['public_batch_enqueued'] = live.enqueue_uncovered() + await live.tick() + job = dict(store.query_one('SELECT * FROM sem_batch')) + result['batch_after_first_tick'] = {k: job[k] for k in ('state','attempts','reason','entry_ids')} + result['loopback_requests_before_resume'] = len(contract.FakeGateway.requests) + points, _ = backend._client.scroll(collection_name=backend.collection, with_payload=True, limit=100) + result['stored_points_after_first'] = [dict(p.payload) for p in points] + result['claimed_completed_message_ids'] = sorted(backend.completed_message_ids(job['batch_key'])) + backend.llm_guard.cap = 5 + await live.tick() + result['batch_after_resume'] = dict(store.query_one('SELECT state,attempts,reason,entry_ids FROM sem_batch')) + result['loopback_requests_after_resume'] = len(contract.FakeGateway.requests) + result['unsent_user_row_actually_sent'] = any( + '合成用户原话甲' in json.dumps(json.loads(r['body']), ensure_ascii=False) + for r in contract.FakeGateway.requests) + store.close() + finally: + if backend: + backend._memory.llm.client.close() + await backend.aclose() + server.shutdown() + server.server_close() + (ROOT / 'HEL-882-partial-independent.json').write_text(json.dumps(result, ensure_ascii=False, indent=2)) + print(json.dumps(result, ensure_ascii=False, indent=2)) + +asyncio.run(main()) diff --git a/src/cybergril/life/model.py b/src/cybergril/life/model.py index 069383d..374c120 100644 --- a/src/cybergril/life/model.py +++ b/src/cybergril/life/model.py @@ -50,14 +50,16 @@ class PersistentBudget(BudgetGuard): )["n"] def total_requests_used(self): - return self.store.query_one( - "SELECT coalesce(value,0) v FROM budget_total WHERE key='total_requests'" - )["v"] + row = self.store.query_one( + "SELECT coalesce((SELECT value FROM budget_total WHERE key='total_requests'),0) v" + ) + return row["v"] if row else 0 def total_tokens_used(self): - return self.store.query_one( - "SELECT coalesce(value,0) v FROM budget_total WHERE key='total_tokens'" - )["v"] + row = self.store.query_one( + "SELECT coalesce((SELECT value FROM budget_total WHERE key='total_tokens'),0) v" + ) + return row["v"] if row else 0 def reserve(self, tokens, scene): """Legacy background-scene token reservation (95% stop), kept for compatibility.""" @@ -126,7 +128,14 @@ class PersistentBudget(BudgetGuard): reserved_tokens=tokens, requests=requests, day=day) def settle(self, hold, *, actual_tokens=None): - """Real usage adjusts the reserved tokens; unknown usage keeps the reserve.""" + """Real usage adjusts the reserved tokens; unknown usage keeps the reserve. + + HEL-885: the correction lands on the day the hold was RESERVED on + (``hold.day``) — a request crossing midnight adjusts the old day's row + to its actual spend. Writing to ``self.state.day`` (the HEL-882 bug) + manufactured negative usage on the new day and re-opened headroom the + old day had already consumed. + """ if hold is None: return self._roll() @@ -137,7 +146,7 @@ class PersistentBudget(BudgetGuard): self.store.execute( "INSERT INTO token_usage VALUES(?,?,?) ON CONFLICT(day,scene) DO UPDATE " "SET used=used+excluded.used", - (self.state.day, hold.scene, delta), + (hold.day, hold.scene, delta), ) self.store.execute( "INSERT INTO budget_total VALUES('total_tokens',?) ON CONFLICT(key) DO UPDATE " @@ -145,7 +154,13 @@ class PersistentBudget(BudgetGuard): ) def release(self, hold): - """Refund only provably-unsent reservations (no provider, nothing left).""" + """Refund only provably-unsent reservations (no provider, nothing left). + + HEL-885: the refund returns to ``hold.day`` — the day that actually + paid the reservation — for the daily tables too; the one-shot + ``budget_total`` counters are cross-day and refund in place. Refunding + into ``self.state.day`` inflated the new day with negative usage. + """ if hold is None: return self._roll() @@ -153,12 +168,12 @@ class PersistentBudget(BudgetGuard): self.store.execute( "INSERT INTO token_usage VALUES(?,?,?) ON CONFLICT(day,scene) DO UPDATE " "SET used=used+excluded.used", - (self.state.day, hold.scene, -hold.reserved_tokens), + (hold.day, hold.scene, -hold.reserved_tokens), ) self.store.execute( "INSERT INTO request_usage VALUES(?,?,?) ON CONFLICT(day,category) DO UPDATE " "SET requests=requests+excluded.requests", - (self.state.day, hold.category, -hold.requests), + (hold.day, hold.category, -hold.requests), ) self.store.execute( "INSERT INTO budget_total VALUES('total_requests',?) ON CONFLICT(key) DO UPDATE " diff --git a/src/cybergril/memory/r4.sql b/src/cybergril/memory/r4.sql index 07a925d..ac76857 100644 --- a/src/cybergril/memory/r4.sql +++ b/src/cybergril/memory/r4.sql @@ -22,6 +22,17 @@ CREATE TABLE IF NOT EXISTS sem_covered ( message_id INTEGER PRIMARY KEY REFERENCES chat_message(id), batch_id INTEGER NOT NULL REFERENCES sem_batch(id) ); +-- HEL-885 per-row completion: durable proof one row's add fully finished — +-- including a legitimate EMPTY extraction that stored no vector. Resume skips +-- rows proven done here or by vector-store anchors; finished rows are never +-- re-sent after a crash, restart or budget-refusal cycle. +CREATE TABLE IF NOT EXISTS sem_row_done ( + batch_id INTEGER NOT NULL REFERENCES sem_batch(id), + message_id INTEGER NOT NULL, + entry_ids TEXT NOT NULL DEFAULT '[]', -- JSON list of backend entry ids this row created + done_at INTEGER NOT NULL, + PRIMARY KEY(batch_id, message_id) +); -- Life activity continuity: start / continue / interrupt / resume / finish. CREATE TABLE IF NOT EXISTS life_activity ( diff --git a/src/cybergril/memory/semantic.py b/src/cybergril/memory/semantic.py index 961d59c..e80a851 100644 --- a/src/cybergril/memory/semantic.py +++ b/src/cybergril/memory/semantic.py @@ -41,6 +41,20 @@ Product port of the proven HEL-857/861–869 isolation results (plan-next-HEL-86 refusal re-queues the batch without burning an attempt; once budget returns, the job resumes and skips rows whose entries already live in the vector store (store-level idempotency: no lost rows, no duplicated sends, restart-safe). +- Per-row progress (HEL-885): every ``Memory.add`` carries an anchor for exactly + the ONE row it sends — a stored vector vouches for its own row and nothing + else (the HEL-882 whole-batch anchor let the first vector mark unsent rows + complete; the manager's real-SDK reproduction showed the user row never + leaving the process). A row whose add fully finished — including a + legitimate EMPTY extraction that stored no vector — is recorded in + ``sem_row_done``; resume skips rows proven done by either the vector-store + anchors or that durable row record, so no row is lost, none is falsely + complete, and finished work is never re-sent. +- Unknown reservations are never refunded (HEL-885): a reservation whose + outcome is unknown (worker crashed after the wire, never settled) stays + counted forever — a lease expiring is not proof the call never happened. + Only ``release()`` with provably-unsent evidence (nothing hit the wire) + ever removes a hold. - The stub backend is an orchestration double for engineering verification and isolated tests only; it is not semantic and never claimed as such. """ @@ -144,17 +158,32 @@ class StubBackend: self._entries: dict[str, dict] = {} self._seq = 0 - async def add_group(self, rows: list[dict], anchor: dict, batch_key: str) -> list[dict]: + async def add_group(self, rows: list[dict], anchor: dict, batch_key: str, *, + skip: set[int] | frozenset[int] = frozenset(), + on_row_done=None) -> list[dict]: + """Per-row mirror of the mem0 contract: one entry per row, each with + that row's own anchor (HEL-885) — orchestration progress parity.""" + def _add() -> list[dict]: + done = self.completed_message_ids(batch_key) | set(skip) + send = [r for r in rows if r["message_id"] not in done] created = [] - text = "\n".join(r["text"] for r in rows) - self._seq += 1 - entry_id = f"stub-{self._seq}" - self._entries[entry_id] = { - "id": entry_id, "data": text, - "payload": {"data": text, "src_group": batch_key, **anchor}, - } - created.append({"id": entry_id, "data": text}) + for row in send: + row_anchor = { + "turn_ids": [row["turn_id"]], + "turn_roles": [[row["turn_id"], row["role"], row["message_id"]]], + "input_sha256": anchor_fingerprint([row]), + } + self._seq += 1 + entry_id = f"stub-{self._seq}" + self._entries[entry_id] = { + "id": entry_id, "data": row["text"], + "payload": {"data": row["text"], "src_group": batch_key, **row_anchor}, + } + entry = {"id": entry_id, "data": row["text"]} + created.append(entry) + if on_row_done is not None: + on_row_done(row, [entry]) return created return await asyncio.to_thread(_add) @@ -199,7 +228,9 @@ class EmbedLedgerGuard: HEL-874: debits go through ``reserve``/``settle``/``release`` so the check-and-hold is one serialized SQLite transaction — concurrent workers can - never overspend the cap. Stale holds (crashed workers) are purged by age. + never overspend the cap. Stale holds are NEVER purged (HEL-885): a crashed + worker's reservation has an unknown outcome and stays counted; only a + provably-unsent ``release()`` removes it. HEL-882: ``daily_cap`` adds the execution-table C per-day top on top of the lifetime cap (both enforced, tightest wins; the ts column is UTC ISO so the @@ -290,8 +321,15 @@ def _pid() -> int: return os.getpid() -def _reserve_cutoff() -> str: - """ISO stamp older than any live reservation may be (crashed workers).""" +def _reserve_cutoff() -> str: # noqa: DEE001 — kept for audit tooling reference + """Deprecated (HEL-885): aged ``reserve:`` rows are NEVER auto-deleted. + + A hold whose worker died is an UNKNOWN outcome — the request may well be + sitting answered on the wire (the exact "sent then crashed before settle" + case). The manager's independent reproduction showed the old 1-hour purge + re-opening cumulative caps. ``_atomic_reserve`` no longer calls this; it + stays only so historical audit scripts importing it keep working. + """ import time as _time return _time.strftime("%Y-%m-%dT%H:%M:%SZ", _time.gmtime(_time.time() - 3600)) @@ -306,6 +344,11 @@ def _atomic_reserve(conn, *, table: str, cap: int, keep: int, items: int, note: same serialized transaction — lifetime cap and daily top both apply, the tightest one refuses. Reservation rows carry today's UTC timestamp so the pending holds themselves count toward the day they were taken on. + + HEL-885: aged ``reserve:`` rows are never purged here. A reservation is + removed ONLY by ``release()`` with provably-unsent evidence — a crashed + worker's hold has an unknown outcome and stays counted (conservative: + unknown consumption is billed, never refunded as "never happened"). """ import time as _time @@ -313,10 +356,6 @@ def _atomic_reserve(conn, *, table: str, cap: int, keep: int, items: int, note: now = _time.strftime("%Y-%m-%dT%H:%M:%SZ", _time.gmtime()) conn.execute("BEGIN IMMEDIATE") try: - conn.execute( - f"DELETE FROM {table} WHERE note LIKE ? AND ts < ?", - (EmbedLedgerGuard.RESERVE_NOTE + "%", _reserve_cutoff()), - ) if counted_kinds: marks = ",".join("?" * len(counted_kinds)) row = conn.execute( @@ -552,22 +591,28 @@ class Mem0Backend: # Must precede the mem0 import; same ordering guarantee as the spike. os.environ.setdefault("MEM0_TELEMETRY", "False") - try: - from mem0 import Memory # noqa: PLC0415 — fixed install, lazy by design - except ImportError as exc: # fail closed, never a silent fallback - raise RuntimeError("mem0_not_installed") from exc self.ledger = ledger or EmbedLedgerGuard( config.get("ledger_path", ""), config.get("embed_cap", 0), daily_cap=config.get("embed_daily")) self.llm_guard = llm_guard or LlmLedgerGuard( config.get("ledger_path", ""), config.get("llm_cap", 0), daily_cap=config.get("llm_daily")) - # HEL-882: a fresh file must never re-open the frozen balance — with any - # approved cap the configured ledger has to still be the authoritative - # account carrying the frozen prefix. + # HEL-885: the authority block is REQUIRED, never optional. The + # HEL-882 code skipped verification entirely when ``authority`` was + # absent, so a fresh empty ledger + positive caps constructed and sent + # (manager's real-SDK reproduction). Now: any positive cap without a + # (non-empty, dict) authority block — or with one that does not match + # the authoritative account — fails closed BEFORE the SDK is even + # imported: a fresh file can never re-open the frozen 498/500 headroom. authority = config.get("authority") - if authority and (self.ledger.cap > 0 or self.llm_guard.cap > 0): + if self.ledger.cap > 0 or self.llm_guard.cap > 0: + if not isinstance(authority, dict) or not authority: + raise SemanticBudgetError("ledger_authority_required") verify_ledger_authority(config.get("ledger_path", ""), authority) + try: + from mem0 import Memory # noqa: PLC0415 — fixed install, lazy by design + except ImportError as exc: # fail closed, never a silent fallback + raise RuntimeError("mem0_not_installed") from exc # Unified provider-exit gate: the SDK's internal extraction LLM is a real # provider exit like any other; when the product hands its router guard # over, every extraction call also reserves request count + conservative @@ -717,6 +762,13 @@ class Mem0Backend: Reads the vector store's own payload (src_group anchor) — a crashed or restarted worker finds what earlier attempts persisted, so recovery resends only the missing rows and never duplicates completed ones. + + HEL-885: every add carries a PER-ROW anchor, so each stored point + vouches for exactly the one message in its own ``turn_roles`` — vector + existence never claims other rows of the batch are done (that whole- + batch-anchor claim was the HEL-882 permanent-loss defect). Rows whose + extraction legitimately stored nothing carry no point; their durable + completion proof is the caller-side ``sem_row_done`` record. """ try: from qdrant_client import models as qmodels # noqa: PLC0415 — lazy, pinned SDK @@ -744,48 +796,65 @@ class Mem0Backend: break return done - async def add_group(self, rows: list[dict], anchor: dict, batch_key: str) -> list[dict]: + async def add_group(self, rows: list[dict], anchor: dict, batch_key: str, *, + skip: set[int] | frozenset[int] = frozenset(), + on_row_done=None) -> list[dict]: + """One entry per row, each carrying that row's own anchor (HEL-885). + + ``skip`` holds message_ids already proven complete (vector-store + anchors ∪ the durable ``sem_row_done`` record) — those rows are never + re-sent. ``on_row_done(row, entries)`` fires in the worker thread as + soon as one row's add fully finished (empty extraction included), so + the caller can durably record per-row progress even if the process + dies mid-batch. + """ # Fail closed before ANY real work while either budget is unapproved: # a 2.2.1 add always spends ≥1 embed item and 1 extraction LLM call. self.ledger.check(1) self.llm_guard.check(1) def _add() -> list[dict]: - # Recovery skips rows whose entries already live in the vector store - # (crash/restart/budget-exhaustion resume); the anchor sent with the - # remaining rows equals exactly that send's input set. - done = self.completed_message_ids(batch_key) + # Vector-store anchors are ALWAYS consulted (direct callers get the + # same no-resend guarantee); ``skip`` adds the caller's durable + # row records for rows that legitimately stored no vector. + done = self.completed_message_ids(batch_key) | set(skip) send = [r for r in rows if r["message_id"] not in done] if not send: return [] - send_anchor = { - "turn_ids": [r["turn_id"] for r in send], - "turn_roles": [[r["turn_id"], r["role"], r["message_id"]] for r in send], - "input_sha256": anchor_fingerprint(send), - } - failures_before = list(getattr(self._failures, "list", []) or []) created = [] for row in send: + # Per-row anchor: this add's input set is exactly this row. + row_anchor = { + "turn_ids": [row["turn_id"]], + "turn_roles": [[row["turn_id"], row["role"], row["message_id"]]], + "input_sha256": anchor_fingerprint([row]), + } + failures_before = len(getattr(self._failures, "list", []) or []) result = self._memory.add( - row["text"], metadata={"src_group": batch_key, **send_anchor}, + row["text"], metadata={"src_group": batch_key, **row_anchor}, user_id=self.user_scope, ) + # mem0 2.2.1 swallows per-text embed failures into empty + # results (phase-3/7 fallback loops). An empty extraction is + # legitimate only when no embed/refusal happened during THIS + # row's add: if our guards refused anything, that refusal is + # the truth, not the empty list. + failures = (getattr(self._failures, "list", []) or [])[failures_before:] + if failures: + raise SemanticBudgetError(f"semantic_add_swallowed_failure:{failures[0]}") items = result.get("results") if isinstance(result, dict) else result if isinstance(items, dict): items = [items] + row_entries = [] for item in items or []: if not isinstance(item, dict): continue entry = {"id": item.get("id"), "data": (item.get("memory") or "")} if entry["id"]: - created.append(entry) - # mem0 2.2.1 swallows per-text embed failures into empty results - # (phase-3/7 fallback loops). An empty extraction is legitimate only - # when no embed/refusal happened: if our guards refused anything - # during this add, that refusal is the truth, not the empty list. - failures = (getattr(self._failures, "list", []) or [])[len(failures_before):] - if failures: - raise SemanticBudgetError(f"semantic_add_swallowed_failure:{failures[0]}") + row_entries.append(entry) + created.extend(row_entries) + if on_row_done is not None: + on_row_done(row, row_entries) return created return await asyncio.to_thread(_add) @@ -949,6 +1018,23 @@ class SemanticMemory: "WHERE id=?", (owner, now + 40000, job["id"]), ) state, reason, entry_ids = "failed", "extraction_failed", None + row_entries: list[str] = [] + + def _row_done(row: dict, entries: list[dict]) -> None: + """Durable per-row completion, written inside the worker thread the + moment a row's add fully finished (HEL-885) — a legitimately empty + extraction stored no vector, so this record is its only resend + proof; a crash between rows can never resend a finished row.""" + with self.store.conn: + self.store.execute( + "INSERT OR IGNORE INTO sem_row_done(batch_id,message_id,entry_ids,done_at) " + "VALUES(?,?,?,?)", + (job["id"], row["message_id"], + json.dumps([e["id"] for e in entries], ensure_ascii=False), + int(self.clock() * 1000)), + ) + row_entries.extend(e["id"] for e in entries if e["id"]) + try: anchor = json.loads(job["anchor_json"]) rows = [] @@ -963,8 +1049,15 @@ class SemanticMemory: "role": role, "text": message["text"]}) if anchor_fingerprint(rows) != anchor["input_sha256"]: raise ValueError("source_mismatch") - created = await self.backend.add_group(rows, anchor, job["batch_key"]) - entry_ids = json.dumps([e["id"] for e in created], ensure_ascii=False) + # Resume truth = vector-store anchors ∪ durable per-row records. + done_ids = self.backend.completed_message_ids(job["batch_key"]) + done_ids |= {r["message_id"] for r in self.store.query( + "SELECT message_id FROM sem_row_done WHERE batch_id=?", (job["id"],))} + created = await self.backend.add_group( + rows, anchor, job["batch_key"], skip=done_ids, on_row_done=_row_done) + row_entries.extend(e["id"] for e in created if e["id"] not in row_entries) + entry_ids = json.dumps( + list(dict.fromkeys(row_entries)), ensure_ascii=False) state, reason = "done", None except Exception as exc: # noqa: BLE001 — job isolation, reason recorded reason = str(exc) or exc.__class__.__name__ @@ -983,6 +1076,14 @@ class SemanticMemory: return log.warning("semantic batch %s failed: %s", job["id"], reason) finally: + if state == "done": + # entry_ids = every entry this batch ever created, across all + # its partial attempts (sem_row_done holds the earlier ticks'). + recorded = [json.loads(r["entry_ids"] or "[]") for r in self.store.query( + "SELECT entry_ids FROM sem_row_done WHERE batch_id=?", (job["id"],))] + entry_ids = json.dumps( + list(dict.fromkeys([e for lst in recorded for e in lst] + row_entries)), + ensure_ascii=False) with self.store.conn: self.store.execute( "UPDATE sem_batch SET state=?,reason=?,entry_ids=?,owner=NULL,finished_at=? " diff --git a/tests/test_hard_gates_hel882.py b/tests/test_hard_gates_hel882.py index d5c2b4a..bdaecc9 100644 --- a/tests/test_hard_gates_hel882.py +++ b/tests/test_hard_gates_hel882.py @@ -182,7 +182,15 @@ def test_total_tokens_cap_blocks_before_daily_wall(tmp_path): with TestClient(app) as client: create_companion(client) guard = app.state.life.router.guard - guard.record(4999, "review_almost_total") # one-shot total nearly spent + # HEL-885: the one-shot TOTAL must be booked in budget_total — the old + # setup used record(), which only writes the DAILY table, and passed + # merely because total_tokens_used() CRASHED on the then-empty + # budget_total (TypeError surfaced as chat.error). Book the total + # honestly now that the read is fixed. + with guard.store.conn: + guard.store.execute( + "INSERT INTO budget_total VALUES('total_tokens',4999) " + "ON CONFLICT(key) DO UPDATE SET value=4999") calls_before = len(provider.calls) assert [e["t"] for e in events(turn(client, "还会发吗"))] == ["chat.error"] assert len(provider.calls) == calls_before diff --git a/tests/test_mem0_backend.py b/tests/test_mem0_backend.py index 2e4a25f..b632435 100644 --- a/tests/test_mem0_backend.py +++ b/tests/test_mem0_backend.py @@ -99,13 +99,15 @@ class FakeLlm: class FakeMemory: instances = [] + _seq_counter = 0 # unique entry ids across instances (restart-safe, like UUIDs) def __init__(self, embedder, collection): self.embedding_model = embedder self.vector_store = _FakeVectorStore() self.llm = FakeLlm() self.collection = collection - self.seq = 0 + FakeMemory._seq_counter += 1 + self.seq = FakeMemory._seq_counter @classmethod def from_config(cls, config): @@ -146,13 +148,36 @@ def _sys_modules(): def _make_ledger(path): + """Test ledger with a synthetic authority prefix row (HEL-885). + + The seeded row is deliberately an UNCOUNTED kind with 0 items, so guard + ``used()`` stays 0 — it only anchors the file identity for the required + ``authority`` block (product defaults are NOT relaxed; tests must hand the + backend an explicit synthetic authority, per the HEL-885 review). + """ conn = sqlite3.connect(str(path)) conn.execute("CREATE TABLE embed_ops (id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, pid INTEGER," " kind TEXT, items INTEGER, note TEXT)") + conn.execute( + "INSERT INTO embed_ops(ts,pid,kind,items,note) VALUES" + "('2026-01-01T00:00:00Z',1,'authority_prefix',0,'authority:synthetic-test-prefix')") conn.commit() conn.close() +def synthetic_authority(ledger_path, *, prefix_rows: int = 1, min_used: int = 0) -> dict: + """Explicit synthetic authority over a test ledger's frozen prefix.""" + import hashlib + + rows = sqlite3.connect(str(ledger_path)).execute( + "SELECT ts,pid,kind,items,note FROM embed_ops ORDER BY id LIMIT ?", + (prefix_rows,), + ).fetchall() + digest = hashlib.sha256( + "\n".join("|".join(str(c) for c in row) for row in rows).encode()).hexdigest() + return {"prefix_rows": prefix_rows, "prefix_sha256": digest, "min_used": min_used} + + @pytest.fixture def fake_mem0(monkeypatch, tmp_path): ledger_path = tmp_path / "ledger.sqlite3" @@ -197,6 +222,10 @@ def backend_for(tmp_path, fake_mem0, *, embedder=None, **overrides): "llm_cap": 10, } config.update(overrides) + if config.get("embed_cap", 0) > 0 or config.get("llm_cap", 0) > 0: + # HEL-885: positive caps REQUIRE an explicit (synthetic) authority — + # the product default refuses without one; tests never relax that. + config.setdefault("authority", synthetic_authority(config["ledger_path"])) # each backend wraps a fresh embedder, exactly like real from_config instances sys.modules["mem0"]._TEST_EMBEDDER = embedder or FakeEmbedder() return Mem0Backend(config) diff --git a/tests/test_rework_hel885.py b/tests/test_rework_hel885.py new file mode 100644 index 0000000..d2c938b --- /dev/null +++ b/tests/test_rework_hel885.py @@ -0,0 +1,330 @@ +"""HEL-885 rework regressions: the manager's independent-review failures. + +Each case first mirrors the exact reproduction from HEL-885's review (real +public user+companion queue; cross-day settlement; aged unknown reservation; +missing authority; per-row progress), then proves the fix. Fake-SDK mirrors +run here; the pinned real install is covered live by +tools/verify_semantic_recovery_hel885.py (deterministic boundary, loopback). +""" +import asyncio +import json +import sqlite3 + +import pytest +from test_mem0_backend import POINT_LOG, FakeLlm, FakeMemory, _make_ledger, synthetic_authority +from test_semantic_recovery_hel882 import config_for, install_fake + +from cybergril.life.model import PersistentBudget +from cybergril.memory.semantic import ( + EmbedLedgerGuard, + LlmLedgerGuard, + Mem0Backend, + SemanticBudgetError, + SemanticMemory, +) +from cybergril.memory.store import Store +from cybergril.think.budget import BudgetConfig, BudgetGuard + + +def seed_public_two_row_turn(store, turn_id, user_text, companion_text): + """The manager's partial-repro shape: a sent user row plus a judged+acked + (public) companion row in the same turn.""" + with store.conn: + store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" + "(1,'user',?,?,'sent')", (user_text, turn_id)) + inserted = store.execute( + "INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" + "(2,'companion',?,?,'sent')", (companion_text, turn_id)) + store.execute( + "INSERT INTO sharing_attempt(message_id,turn_id,epoch,token,text_hash,expires_at," + "ack_at,candidates,state) VALUES(?,'"+turn_id+"',1,'t','h',9223372036854775806," + "3,'[]','judged')", (inserted.lastrowid,)) + return store.query_one("SELECT id FROM chat_message WHERE text=?", (user_text,))["id"], \ + store.query_one("SELECT id FROM chat_message WHERE text=?", (companion_text,))["id"] + + +def llm_bodies(backend): + return [json.dumps(m, ensure_ascii=False) for m in backend._memory.llm.calls] + + +async def test_partial_batch_claims_only_the_row_actually_sent(tmp_path, monkeypatch): + """Manager partial repro, fake mirror: llm_cap=1, two public rows — after + the first tick exactly ONE wire call happened, and the completed set is + the companion row ONLY (the HEL-882 whole-batch anchor claimed both). + Raising the cap resumes: done, and the user's original words really go + out — exactly once.""" + install_fake(monkeypatch) + ledger = tmp_path / "partial.sqlite3" + _make_ledger(ledger) + backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=1)) + store = Store(tmp_path / "partial.db") + user_id, companion_id = seed_public_two_row_turn( + store, "turn1", "合成用户原话甲", "合成已公开伴侣原话乙") + live = SemanticMemory(store, backend) + assert live.enqueue_uncovered() == 1 + await live.tick() + + job = dict(store.query_one("SELECT state,reason,entry_ids FROM sem_batch")) + assert job["state"] == "queued" and "budget" in (job["reason"] or "") + assert len(backend._memory.llm.calls) == 1 # one actual wire send, no retries + batch_key = store.query_one("SELECT batch_key FROM sem_batch")["batch_key"] + # THE defect: the first vector must not vouch for the unsent user row + assert backend.completed_message_ids(batch_key) == {companion_id} + points = backend._client.scroll(collection_name=backend.collection, with_payload=True, + limit=100)[0] + for point in points: # per-row anchors: each payload covers exactly one row + roles = point.payload.get("turn_roles") or [] + assert len(roles) == 1 + + backend.llm_guard.cap = 5 # the manager raises the temporary cap + await live.tick() + after = dict(store.query_one("SELECT state,reason,entry_ids FROM sem_batch")) + assert after["state"] == "done" and json.loads(after["entry_ids"]) + bodies = llm_bodies(backend) + assert sum("合成用户原话甲" in b for b in bodies) == 1 # actually sent, once + assert sum("合成已公开伴侣原话乙" in b for b in bodies) == 1 # never resent + assert backend.completed_message_ids(batch_key) == {user_id, companion_id} + store.close() + + +async def test_empty_extraction_row_is_done_and_never_resent(tmp_path, monkeypatch): + """A row whose extraction legitimately returns nothing stores no vector — + it is still provably complete (sem_row_done) and never re-sent.""" + install_fake(monkeypatch) + + class EmptyForOneRow(FakeLlm): + def generate_response(self, messages=None, **kwargs): + user = messages[1]["content"] if messages and len(messages) > 1 else "" + if "甲" in user: + self.calls.append(messages) + return json.dumps({"memory": []}) # legitimate empty extraction + return super().generate_response(messages=messages, **kwargs) + + class EmptyMemory(FakeMemory): + """Parses the extraction like the real SDK: an empty memory list + stores NO point (the legitimate-empty path HEL-885 must prove done).""" + + def __init__(self, embedder, collection): + super().__init__(embedder, collection) + self.llm = EmptyForOneRow() + + def add(self, text, metadata=None, user_id=None): + if not user_id: + raise ValueError("filters must contain at least one of: user_id, agent_id, run_id") + self.embedding_model.embed(text, memory_action="search") + raw = self.llm.generate_response(messages=[ + {"role": "system", "content": "extract"}, + {"role": "user", "content": text}, + ], response_format={"type": "json_object"}) + extracted = json.loads(raw).get("memory") or [] + if not extracted: + return {"results": []} # nothing extracted → nothing stored + texts = [item["text"] for item in extracted] + self.embedding_model.embed_batch(texts, memory_action="add") + results = [] + for one in texts: + self.seq += 1 + entry_id = f"m{self.seq}" + bucket = POINT_LOG.setdefault(self.collection, {}) + bucket[entry_id] = {"data": one, "user_id": user_id, **(metadata or {})} + results.append({"id": entry_id, "memory": one, "event": "ADD"}) + return {"results": results} + + install_fake(monkeypatch, EmptyMemory) + ledger = tmp_path / "empty.sqlite3" + _make_ledger(ledger) + backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=50)) + store = Store(tmp_path / "empty.db") + seed_public_two_row_turn(store, "t9", "这一行提取为空甲", "这一行有提取乙") + live = SemanticMemory(store, backend) + live.enqueue_uncovered() + await live.tick() + batch = store.query_one("SELECT * FROM sem_batch") + assert batch["state"] == "done" + done = {r["message_id"] for r in store.query("SELECT message_id FROM sem_row_done")} + user_id = store.query_one("SELECT id FROM chat_message WHERE text='这一行提取为空甲'")["id"] + assert done == {user_id, store.query_one( + "SELECT id FROM chat_message WHERE text='这一行有提取乙'")["id"]} + # honest accounting: only the row that extracted something has an entry + assert len(json.loads(batch["entry_ids"])) == 1 + + # crash-style resume probe: batch requeued by hand, tick again — the + # empty-extraction row is skipped via sem_row_done, no second LLM send + with store.conn: + store.execute("UPDATE sem_batch SET state='queued',owner=NULL,lease_until=NULL") + calls_before = len(backend._memory.llm.calls) + await live.tick() + assert len(backend._memory.llm.calls) == calls_before # nothing re-sent + assert store.query_one("SELECT state FROM sem_batch")["state"] == "done" + store.close() + + +async def test_process_restart_resumes_only_the_missing_row(tmp_path, monkeypatch): + """Crash after row 1 persisted: a NEW Store + NEW backend on the same + paths (process restart) resumes sending only the missing row.""" + install_fake(monkeypatch) + ledger = tmp_path / "restart.sqlite3" + _make_ledger(ledger) + backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=1)) + db_path = tmp_path / "restart.db" + store = Store(db_path) + seed_public_two_row_turn(store, "tr", "重启后第一行", "重启后第二行") + live = SemanticMemory(store, backend) + live.enqueue_uncovered() + await live.tick() # companion row done, user row budget-refused + batch_key = store.query_one("SELECT batch_key FROM sem_batch")["batch_key"] + store.close() # graceful close; durable state is what a crash leaves anyway + + backend2 = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=50)) + store2 = Store(db_path) + live2 = SemanticMemory(store2, backend2) + await live2.tick() # restart resumes the SAME batch + assert store2.query_one("SELECT state FROM sem_batch")["state"] == "done" + bodies = llm_bodies(backend2) + assert sum("重启后第一行" in b for b in bodies) == 1 # only the missing row + assert backend2.completed_message_ids(batch_key) == { + store2.query_one("SELECT id FROM chat_message WHERE text='重启后第一行'")["id"], + store2.query_one("SELECT id FROM chat_message WHERE text='重启后第二行'")["id"]} + store2.close() + + +async def test_concurrent_ticks_never_double_send(tmp_path, monkeypatch): + """Two ticks racing on one batch: the BEGIN IMMEDIATE claim gives the + whole batch to one worker; every row is sent exactly once.""" + install_fake(monkeypatch) + ledger = tmp_path / "race.sqlite3" + _make_ledger(ledger) + backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=50)) + store = Store(tmp_path / "race.db") + seed_public_two_row_turn(store, "tc", "并发第一行", "并发第二行") + live = SemanticMemory(store, backend) + live.enqueue_uncovered() + await asyncio.gather(live.tick(), live.tick()) + assert store.query_one("SELECT state FROM sem_batch")["state"] == "done" + bodies = llm_bodies(backend) + assert sum("并发第一行" in b for b in bodies) == 1 + assert sum("并发第二行" in b for b in bodies) == 1 + store.close() + + +def test_cross_day_settlement_books_on_the_reservation_day(tmp_path): + """Manager midnight repro: day cap 100, hold 90 on day 1, actual 10 + arriving on day 2 — day 1 settles to 10, day 2 stays 0 (never -80), and + a 150-token request the next day is refused.""" + store = Store(tmp_path / "midnight.db") + guard = PersistentBudget(store, BudgetGuard(BudgetConfig( + daily_tokens=100, daily_requests={"dialogue": 10}))) + guard._roll = lambda: None + guard.state.day = "2026-10-26" + hold = guard.acquire("dialogue_main", tokens=90) + guard.state.day = "2026-10-27" + guard.settle(hold, actual_tokens=10) + assert guard.used == 0 # new day untouched — no manufactured headroom + with pytest.raises(RuntimeError, match="budget_tokens_exhausted"): + guard.acquire("dialogue_main", tokens=150) # 150 > 100 cap → refused + rows = {r["day"]: r["used"] for r in store.query("SELECT * FROM token_usage")} + assert rows == {"2026-10-26": 10} + assert all(v >= 0 for v in rows.values()) + store.close() + + +def test_cross_day_release_refunds_the_paying_day(tmp_path): + """A provably-unsent hold reserved on day 1 and released on day 2 refunds + day 1 (and the one-shot totals) — day 2 never goes negative.""" + store = Store(tmp_path / "release.db") + guard = PersistentBudget(store, BudgetGuard(BudgetConfig( + daily_tokens=100, daily_requests={"dialogue": 10}, total_tokens=500))) + guard._roll = lambda: None + guard.state.day = "2026-10-26" + hold = guard.acquire("dialogue_main", tokens=90) + guard.state.day = "2026-10-27" + guard.release(hold) + rows = {r["day"]: r["used"] for r in store.query("SELECT * FROM token_usage")} + assert rows == {"2026-10-26": 0} + assert guard.used == 0 + assert guard.total_tokens_used() == 0 + req = {r["day"]: r["requests"] for r in store.query("SELECT * FROM request_usage")} + assert req == {"2026-10-26": 0} + store.close() + + +def _backdate(path, table, row_id): + conn = sqlite3.connect(str(path)) + with conn: + conn.execute(f"UPDATE {table} SET ts='2000-01-01T00:00:00Z' WHERE id=?", (row_id,)) + conn.close() + + +def test_aged_unknown_llm_reservation_is_never_refunded(tmp_path): + """Manager extra repro (llm side): a reservation that crashed before + settle is UNKNOWN — an hour (or a restart) later the cap is still spent; + no timeout quietly deletes the charging fact.""" + ledger = tmp_path / "crash-llm.sqlite3" + llm = LlmLedgerGuard(ledger, cap=1) + pending = llm.reserve(1, "sent_but_crashed_before_settlement") + _backdate(ledger, "llm_ops", pending) + with pytest.raises(RuntimeError, match="semantic_llm_budget_exhausted"): + LlmLedgerGuard(ledger, cap=1).reserve(1, "second_send_after_restart") + reopened = LlmLedgerGuard(ledger, cap=1) + assert reopened.used() == 1 # the unknown consumption stayed billed + + +def test_aged_unknown_embed_reservation_is_never_refunded(tmp_path): + """Same shared path, embed side (embed_ops) — unknown holds stay counted, + across process restarts; only an explicit unsent release removes one.""" + ledger = tmp_path / "crash-embed.sqlite3" + _make_ledger(ledger) + guard = EmbedLedgerGuard(ledger, cap=1) + hold = guard.reserve(1, "embed_sent_then_crashed") + _backdate(ledger, "embed_ops", hold) + with pytest.raises(RuntimeError, match="semantic_embed_budget_exhausted"): + EmbedLedgerGuard(ledger, cap=1).reserve(1, "again_after_restart") + assert EmbedLedgerGuard(ledger, cap=1).used() == 1 + # the explicit provably-unsent release still works — the ONLY removal path + fresh = EmbedLedgerGuard(ledger, cap=1) + fresh.release(hold) + assert fresh.used() == 0 + + +async def test_positive_caps_without_authority_fail_closed(tmp_path, monkeypatch): + """Manager partial repro's second finding: a fresh empty ledger + positive + caps + NO authority used to construct and send. Now every missing/empty/ + malformed authority refuses construction before the SDK is touched.""" + install_fake(monkeypatch) + fresh = tmp_path / "fresh-bypass.sqlite3" + conn = sqlite3.connect(str(fresh)) + with conn: + conn.execute("CREATE TABLE embed_ops (id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT," + " pid INTEGER, kind TEXT, items INTEGER, note TEXT)") + conn.execute("CREATE TABLE llm_ops (id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT," + " pid INTEGER, kind TEXT, items INTEGER, note TEXT)") + conn.close() + + def cfg(**extra): + base = config_for(tmp_path, fresh, embed_cap=0, llm_cap=0) + base.update(extra) + return base + + with pytest.raises(SemanticBudgetError, match="ledger_authority_required"): + Mem0Backend(cfg(embed_cap=100, llm_cap=1)) # no authority at all + with pytest.raises(SemanticBudgetError, match="ledger_authority_required"): + Mem0Backend(cfg(embed_cap=100, llm_cap=1, authority={})) # empty block + with pytest.raises(SemanticBudgetError, match="ledger_authority_required"): + Mem0Backend(cfg(llm_cap=1, authority="see-notes")) # malformed + # a WRONG (non-matching) authority is refused by the prefix check + with pytest.raises(SemanticBudgetError): + Mem0Backend(cfg(embed_cap=100, llm_cap=1, + authority={"prefix_rows": 1, "prefix_sha256": "0" * 64, "min_used": 0})) + # zero caps stay constructible (fail-closed on WORK, not on construction) + zero = Mem0Backend(cfg(embed_cap=0, llm_cap=0)) + with pytest.raises(RuntimeError, match="semantic_embed_budget_not_approved"): + await zero.add_group( + [{"message_id": 1, "turn_id": "t", "role": "user", "text": "x"}], + {"turn_ids": ["t"], "turn_roles": [], "input_sha256": "x"}, "z") + # the explicit synthetic authority path constructs (tests never relax the + # product default — they hand it a real prefix, as HEL-885 requires) + seeded = tmp_path / "seeded.sqlite3" + _make_ledger(seeded) # seeds the synthetic prefix row + okay = Mem0Backend(cfg(embed_cap=10, llm_cap=10, ledger_path=str(seeded), + authority=synthetic_authority(seeded))) + assert okay.backend == "mem0" diff --git a/tests/test_semantic.py b/tests/test_semantic.py index bd47829..a5519bc 100644 --- a/tests/test_semantic.py +++ b/tests/test_semantic.py @@ -116,15 +116,19 @@ async def test_gate_rejects_unshared_companion_row(store, sem): make_public_companion(store, "t4") sem.enqueue_uncovered() await sem.tick() - hits = await sem.backend.search("面试") + # per-row anchors (HEL-885): the companion's own entry anchors only her row + hits = await sem.backend.search("辛苦") assert sem.chain_resolve(hits[0])["resolved"] # share state revoked at read time (attempt no longer acked) -> fail closed with store.conn: store.execute("UPDATE sharing_attempt SET ack_at=NULL WHERE turn_id='t4'") chain = sem.chain_resolve(hits[0]) assert chain["reject_reason"] == "companion_unshared" - block = await sem.context("面试怎么样", [], 600) - assert "面试" not in block + block = await sem.context("辛苦了等什么", [], 600) + assert "辛苦" not in block + # her row's rejection never touches the user row's own entry + user_hit = (await sem.backend.search("面试"))[0] + assert sem.chain_resolve(user_hit)["resolved"] async def test_context_respects_budget_and_excludes_turns(store, sem): diff --git a/tests/test_semantic_recovery_hel882.py b/tests/test_semantic_recovery_hel882.py index 6839dd3..b38e234 100644 --- a/tests/test_semantic_recovery_hel882.py +++ b/tests/test_semantic_recovery_hel882.py @@ -21,7 +21,7 @@ import types import pytest from conftest import engineering_budget -from test_mem0_backend import POINT_LOG, FakeEmbedder, FakeMemory, _make_ledger +from test_mem0_backend import POINT_LOG, FakeEmbedder, FakeMemory, _make_ledger, synthetic_authority from cybergril.memory.semantic import ( EmbedLedgerGuard, @@ -117,6 +117,10 @@ def config_for(tmp_path, ledger, *, embed_cap=10, llm_cap=10, **extra): "llm_cap": llm_cap, } config.update(extra) + if config.get("embed_cap", 0) > 0 or config.get("llm_cap", 0) > 0: + # HEL-885: positive caps REQUIRE an explicit synthetic authority; the + # product default (no authority) refuses — see test_rework_hel885. + config.setdefault("authority", synthetic_authority(ledger)) return config @@ -217,40 +221,60 @@ async def test_mid_add_budget_swallow_requeues_then_resumes(tmp_path, monkeypatc async def test_resume_never_loses_or_duplicates_rows(tmp_path, monkeypatch): - """A crash mid-batch (row1 stored, row2 not) resumes sending ONLY row2.""" + """HEL-885 rewrite: the crash-split resume runs through the REAL product + path (real tick, real batch state) — no hand-planted single-row anchors. + The HEL-882 version planted exactly the anchor the product never wrote, + which masked the whole-batch-anchor defect the manager reproduced live.""" install_fake(monkeypatch) ledger = tmp_path / "resume.sqlite3" _make_ledger(ledger) - backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=50)) + backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=1)) store = Store(tmp_path / "resume.db") - rows = seed_turn(store, "t2", "第一条原话") + seed_turn(store, "t2", "第一条原话") with store.conn: # a shared companion row joins the turn store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" "(2,'companion','第二条原话','t2','sent')") row2 = store.query_one("SELECT id FROM chat_message WHERE text='第二条原话'") - rows.append({"message_id": row2["id"], "turn_id": "t2", "role": "companion", "text": "第二条原话"}) - anchor = {"turn_ids": ["t2", "t2"], "turn_roles": [["t2", "user", rows[0]["message_id"]], - ["t2", "companion", rows[1]["message_id"]]], - "input_sha256": "x"} - # simulate: attempt 1 stored row1's entry then died (e.g. crash before commit) - point_id = "pre-stored-1" - POINT_LOG.setdefault("cg_public", {})[point_id] = { - "data": "提取:第一条原话", "user_id": "cybergril", - "src_group": "chat_public:mem0:t2:key", - "turn_roles": [["t2", "user", rows[0]["message_id"]]], - "input_sha256": "x", - } - created = await backend.add_group(rows, anchor, "chat_public:mem0:t2:key") - texts_added = [backend.payload(e["id"])["data"] for e in created] - assert len(created) == 1 # only the missing row was sent - assert "第二条原话" in texts_added[0] - completed = backend.completed_message_ids("chat_public:mem0:t2:key") - assert completed == {rows[0]["message_id"], rows[1]["message_id"]} + store.execute( + "INSERT INTO sharing_attempt(message_id,turn_id,epoch,token,text_hash,expires_at," + "ack_at,candidates,state) VALUES(?,'t2',1,'t','h',9223372036854775806," + "1700000001000,'[]','judged')", (row2["id"],)) + semantic = SemanticMemory(store, backend) + semantic.enqueue_uncovered() + await semantic.tick() # llm_cap=1: companion row spends it, user row refuses + first = dict(store.query_one("SELECT state,reason FROM sem_batch")) + assert first["state"] == "queued" and "budget" in (first["reason"] or "") + # per-row anchors: the ONE stored vector vouches for its own row ONLY + assert backend.completed_message_ids("chat_public:mem0:t2:" + _batch_digest(store)) == { + row2["id"]} + row1_id = store.query_one("SELECT id FROM chat_message WHERE text='第一条原话'")["id"] + done_rows = {r["message_id"] for r in store.query("SELECT message_id FROM sem_row_done")} + assert done_rows == {row2["id"]} + + backend.llm_guard.cap = 50 # budget returns; the SAME batch resumes + await semantic.tick() + after = dict(store.query_one("SELECT state,reason,entry_ids FROM sem_batch")) + assert after["state"] == "done" and json.loads(after["entry_ids"]) + # both rows actually sent, each EXACTLY once — no lost row, no resend + sent = [json.dumps(m, ensure_ascii=False) for m in backend._memory.llm.calls] + assert sum("第一条原话" in body for body in sent) == 1 + assert sum("第二条原话" in body for body in sent) == 1 + completed = backend.completed_message_ids("chat_public:mem0:t2:" + _batch_digest(store)) + assert completed == {row1_id, row2["id"]} # nothing left to send → idempotent no-op - assert await backend.add_group(rows, anchor, "chat_public:mem0:t2:key") == [] + assert await backend.add_group( + [{"message_id": row1_id, "turn_id": "t2", "role": "user", "text": "第一条原话"}, + {"message_id": row2["id"], "turn_id": "t2", "role": "companion", "text": "第二条原话"}], + {"turn_ids": ["t2", "t2"], "turn_roles": [], "input_sha256": "x"}, + "chat_public:mem0:t2:" + _batch_digest(store)) == [] store.close() +def _batch_digest(store): + row = store.query_one("SELECT batch_key FROM sem_batch WHERE backend='mem0'") + return row["batch_key"].rsplit(":", 1)[1] + + async def test_stub_batches_never_block_real_rebuild_and_real_never_resends(tmp_path, monkeypatch): """Review repro 3: stub→mem0 on the same archive reorganizes; a real done batch is never re-enqueued nor re-sent.""" diff --git a/tools/check_release_offline_hel882.py b/tools/check_release_offline_hel882.py index 83934c7..afa8db4 100644 --- a/tools/check_release_offline_hel882.py +++ b/tools/check_release_offline_hel882.py @@ -15,6 +15,9 @@ epoch+1、档案与消息仍在。 R6 mem0 同进程导入现状：lock-only 3.11 venv 的准确缺项（不下载；解决路径 见 deploy/hel874-release.md §4）。 + R7 §4 依赖补齐命令的工具面（HEL-885）：文档使用的落 wheel 命令 + ``python3 -m pip download`` 在本机确实存在（离线 --help 退出 0）， + 且已删除的旧命令 ``uv pip download`` 确实不存在（避免照抄无效命令）。 运行（在仓库根）： @@ -156,6 +159,19 @@ def main() -> None: f"import mem0 → {REPORT['mem0_import_in_release_venv']} " "(resolution path in deploy/hel874-release.md §4; no download in this check)") + # R7 — HEL-885: the §4 dependency-fill commands use tools that exist. + # Offline --help probes only: no network side effects, nothing downloaded. + pip_probe = subprocess.run( + [sys.executable, "-m", "pip", "download", "--help"], capture_output=True, text=True) + check("R7_pip_download_command_exists", pip_probe.returncode == 0, + f"{sys.executable} -m pip download --help → rc={pip_probe.returncode}") + with open(os.devnull, "wb") as devnull: + uv_rc = subprocess.run(["uv", "pip", "download", "--help"], + stdout=devnull, stderr=devnull).returncode + REPORT["removed_uv_pip_download_rc"] = uv_rc + check("R7_removed_uv_pip_download_still_absent", uv_rc != 0, + f"uv pip download --help → rc={uv_rc} (kept absent on purpose; do not re-document it)") + # offline models: echo providers, zero real gateway; ships-as-is caps (all 0) offline_models = current / "config" / "models.offline-check.yaml" offline_models.write_text( diff --git a/tools/verify_mem0_real_hel874.py b/tools/verify_mem0_real_hel874.py index 3261ef6..273fba4 100644 --- a/tools/verify_mem0_real_hel874.py +++ b/tools/verify_mem0_real_hel874.py @@ -196,17 +196,37 @@ def install_facts() -> None: def make_ledger(path: Path, cap: int, llm_cap: int) -> None: + """Temp ledger + SYNTHETIC authority prefix row (HEL-885). + + The seeded row is an uncounted kind with 0 items (guard ``used()`` stays + 0); it exists because the product now REFUSES any positive cap without an + explicit authority block — test/verify harnesses hand it a synthetic one + over this prefix, never relaxing the production default. + """ conn = sqlite3.connect(str(path)) with conn: conn.execute("CREATE TABLE IF NOT EXISTS embed_ops (id INTEGER PRIMARY KEY AUTOINCREMENT," " ts TEXT, pid INTEGER, kind TEXT, items INTEGER, note TEXT)") conn.execute("CREATE TABLE IF NOT EXISTS llm_ops (id INTEGER PRIMARY KEY AUTOINCREMENT," " ts TEXT, pid INTEGER, kind TEXT, items INTEGER, note TEXT)") + if conn.execute("SELECT count(*) FROM embed_ops").fetchone()[0] == 0: + conn.execute( + "INSERT INTO embed_ops(ts,pid,kind,items,note) VALUES" + "('2026-01-01T00:00:00Z',1,'authority_prefix',0,'authority:synthetic-verify-prefix')") conn.close() path.with_name(path.name + ".meta").write_text( json.dumps({"embed_cap": cap, "llm_cap": llm_cap}), encoding="utf-8") +def synthetic_authority(ledger: Path, *, prefix_rows: int = 1, min_used: int = 0) -> dict: + rows = sqlite3.connect(str(ledger)).execute( + "SELECT ts,pid,kind,items,note FROM embed_ops ORDER BY id LIMIT ?", + (prefix_rows,)).fetchall() + digest = hashlib.sha256( + "\n".join("|".join(str(c) for c in row) for row in rows).encode()).hexdigest() + return {"prefix_rows": prefix_rows, "prefix_sha256": digest, "min_used": min_used} + + def embed_used(path: Path) -> int: conn = sqlite3.connect(str(path)) try: @@ -235,7 +255,7 @@ def llm_used(path: Path) -> int: def config_for(tmp: Path, ledger: Path, *, port: int, cap: int, llm_cap: int) -> dict: - return { + config = { "gateway_base_url": f"http://127.0.0.1:{port}/v1", "gateway_model": "hel874-deterministic", "gateway_key": "hel874-local-test", @@ -246,6 +266,11 @@ def config_for(tmp: Path, ledger: Path, *, port: int, cap: int, llm_cap: int) -> "embed_cap": cap, "llm_cap": llm_cap, } + if cap > 0 or llm_cap > 0: + # HEL-885: positive caps require the explicit (synthetic) authority — + # the production default refuses without one. + config["authority"] = synthetic_authority(ledger) + return config async def main_async() -> None: diff --git a/tools/verify_semantic_recovery_hel885.py b/tools/verify_semantic_recovery_hel885.py new file mode 100644 index 0000000..b327eeb --- /dev/null +++ b/tools/verify_semantic_recovery_hel885.py @@ -0,0 +1,266 @@ +#!/usr/bin/env python3 +"""HEL-885 · 实际固定 SDK 恢复语义验证（零真实出站、零真实嵌入、零权威账写入）。 + +在固定 mem0ai 2.2.1（hel857-venv）+ 真实嵌入式 Qdrant + 真实 OpenAI 客户端上， +复现总管独立复审的四类失败并证明修复：确定性 512 维替身（TextEmbedding 构造前 +安装）、127.0.0.1 假网关、临时账/库，全部同 tools/verify_mem0_real_hel874.py +的 HEL-882 边界口径。 + + S0 正 cap 缺 authority 拒绝构造（HEL-882 版曾成功构造并发送）。 + S1 逐行锚点：llm_cap=1 双公开行 → 首个向量只证明自己那行；补预算后续跑， + 用户原话真的发出且每行恰一次（总管 partial 复现的修复形态）。 + S2 进程重启续跑：行1完成后重启（新 Store/新 Backend 同目录），只补缺行。 + S3 租约回收与并发 tick：过期租约回收后处理；两个并发 tick 不双发。 + S4 未知预占不超时退款（llm_ops 与 embed_ops 同路径）：一小时前的未结算 + 预占重启后仍占额度；唯一删除路径是显式可证明未发送的 release。 + +证据 JSON 写 argv[1]。运行（仓库根，固定 venv）： + + /path/to/hel857-venv/bin/python tools/verify_semantic_recovery_hel885.py out.json +""" +from __future__ import annotations + +import asyncio +import importlib.util +import json +import sqlite3 +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +REPO = ROOT +sys.path.insert(0, str(REPO / "src")) +spec = importlib.util.spec_from_file_location('contract', REPO / 'tools/verify_mem0_real_hel874.py') +contract = importlib.util.module_from_spec(spec) +spec.loader.exec_module(contract) +contract.install_deterministic_boundary() +from cybergril.memory.semantic import ( # noqa: E402 — after boundary install + EmbedLedgerGuard, + LlmLedgerGuard, + Mem0Backend, + SemanticBudgetError, + SemanticMemory, +) +from cybergril.memory.store import Store # noqa: E402 + +REPORT: dict = {"head": "HEL-885 rework", "real_llm_requests": 0, "real_embeddings": 0, + "boundary": "deterministic 512 vectors, non-loopback forbidden"} +FAILURES: list[str] = [] + + +def check(name: str, ok: bool, detail: str = "") -> bool: + REPORT.setdefault("checks", []).append({"name": name, "ok": bool(ok), "detail": detail}) + if not ok: + FAILURES.append(f"{name}: {detail}") + return bool(ok) + + +def seed_two_row_turn(store: Store, turn_id: str, user_text: str, companion_text: str): + with store.conn: + store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" + "(1,'user',?,?,'sent')", (user_text, turn_id)) + inserted = store.execute( + "INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES" + "(2,'companion',?,?,'sent')", (companion_text, turn_id)) + store.execute( + "INSERT INTO sharing_attempt(message_id,turn_id,epoch,token,text_hash,expires_at," + "ack_at,candidates,state) VALUES(?,'"+turn_id+"',1,'synthetic','synthetic'," + "999999,3,'[]','judged')", (inserted.lastrowid,)) + return (store.query_one("SELECT id FROM chat_message WHERE text=?", (user_text,))["id"], + store.query_one("SELECT id FROM chat_message WHERE text=?", (companion_text,))["id"]) + + +def bodies_since(mark: int) -> list[str]: + return [r["body"] for r in contract.FakeGateway.requests[mark:]] + + +async def scenario_partial_and_resume(tmp: Path, server, tag: str) -> dict: + """S1 (+S2 restart inside): budget-fail split, resume, restart-safety.""" + ledger = tmp / f"{tag}.sqlite3" + contract.make_ledger(ledger, cap=100, llm_cap=1) + cfg = contract.config_for(tmp, ledger, port=server.server_address[1], + cap=100, llm_cap=1) + cfg["vector_root"] = str(tmp / f"{tag}-vec") + cfg["history_root"] = str(tmp / f"{tag}-hist") + backend = Mem0Backend(cfg) + db = tmp / f"{tag}.db" + store = Store(db) + user_id, companion_id = seed_two_row_turn( + store, f"turn-{tag}", f"用户原话{tag}", f"伴侣公开原话{tag}") + live = SemanticMemory(store, backend) + live.enqueue_uncovered() + mark = len(contract.FakeGateway.requests) + await live.tick() + first_bodies = len(contract.FakeGateway.requests) - mark + first = dict(store.query_one("SELECT state,reason FROM sem_batch")) + batch_key = store.query_one("SELECT batch_key FROM sem_batch")["batch_key"] + completed = sorted(backend.completed_message_ids(batch_key)) + points, _ = backend._client.scroll(collection_name=backend.collection, + with_payload=True, limit=100) + per_row = all(len((p.payload or {}).get("turn_roles") or []) == 1 for p in points) + check(f"S1_{tag}_first_tick_partial", + first["state"] == "queued" and "budget" in (first["reason"] or "") + and first_bodies == 1 and completed == [companion_id] and per_row, + f"state={first['state']} reason={first['reason']} http={first_bodies} " + f"completed={completed} points={len(points)} per_row_anchor={per_row}") + + # S2 restart: close everything, reopen NEW instances on the same paths. + # Precise no-resend proof = HTTP count: only the user row still needs an + # add (one extraction request each). Raw text presence alone cannot count + # sends — mem0 legitimately carries earlier rows in the rolling history + # of every later extraction prompt. + store.close() + await backend.aclose() + cfg2 = dict(cfg) + backend2 = Mem0Backend(cfg2) + backend2.llm_guard.cap = 5 # the manager raises the temporary cap + store2 = Store(db) + live2 = SemanticMemory(store2, backend2) + mark2 = len(contract.FakeGateway.requests) + await live2.tick() + after = dict(store2.query_one("SELECT state,reason,entry_ids FROM sem_batch")) + bodies = bodies_since(mark2) + user_sent = any(f"用户原话{tag}" in b for b in bodies) + completed2 = sorted(backend2.completed_message_ids(batch_key)) + check(f"S2_{tag}_restart_sends_only_missing_row", + after["state"] == "done" and len(bodies) == 1 and user_sent + and completed2 == sorted([user_id, companion_id]) + and json.loads(after["entry_ids"] or "[]"), + f"state={after['state']} http={len(bodies)} user_sent={user_sent} " + f"completed={completed2} entry_ids={after['entry_ids']}") + store2.close() + await backend2.aclose() + return {"first_state": first["state"], "first_reason": first["reason"], + "completed_after_first": completed} + + +async def scenario_lease_and_concurrency(tmp: Path, server) -> None: + ledger = tmp / "lease.sqlite3" + contract.make_ledger(ledger, cap=100, llm_cap=10) + cfg = contract.config_for(tmp, ledger, port=server.server_address[1], + cap=100, llm_cap=10) + cfg["vector_root"] = str(tmp / "lease-vec") + cfg["history_root"] = str(tmp / "lease-hist") + backend = Mem0Backend(cfg) + db = tmp / "lease.db" + store = Store(db) + seed_two_row_turn(store, "turn-lease", "租约恢复第一行", "租约恢复第二行") + live = SemanticMemory(store, backend) + live.enqueue_uncovered() + # a worker died holding the lease: expired pending batch is recovered + with store.conn: + store.execute("UPDATE sem_batch SET state='pending',owner='dead',lease_until=1") + live.recover() + job = dict(store.query_one("SELECT state,reason FROM sem_batch")) + mark = len(contract.FakeGateway.requests) + await live.tick() + after_lease = dict(store.query_one("SELECT state FROM sem_batch")) + bodies = bodies_since(mark) + check("S3a_lease_expired_batch_recovers", + job["state"] == "failed" and job["reason"] == "lease_expired" + and after_lease["state"] == "done" and len(bodies) == 2 + and all(any(t in b for b in bodies) for t in ("租约恢复第一行", "租约恢复第二行")), + f"recovered={job} state_after={after_lease['state']} http={len(bodies)}") + + # concurrency: two ticks racing on one fresh batch never double-send + seed_two_row_turn(store, "turn-race", "并发第一行", "并发第二行") + live.enqueue_uncovered() + mark2 = len(contract.FakeGateway.requests) + await asyncio.gather(live.tick(), live.tick()) + bodies2 = bodies_since(mark2) + races = store.query("SELECT state FROM sem_batch WHERE turn_id='turn-race'") + check("S3b_concurrent_ticks_single_send", + all(r["state"] == "done" for r in races) and len(bodies2) == 2 + and all(any(t in b for b in bodies2) for t in ("并发第一行", "并发第二行")), + f"http={len(bodies2)} states={[r['state'] for r in races]}") + store.close() + await backend.aclose() + + +def scenario_aged_reservations(tmp: Path) -> None: + llm_path = tmp / "aged-llm.sqlite3" + llm = LlmLedgerGuard(llm_path, cap=1) + pending = llm.reserve(1, "sent_but_crashed_before_settlement") + with sqlite3.connect(str(llm_path)) as conn: + conn.execute("UPDATE llm_ops SET ts='2000-01-01T00:00:00Z' WHERE id=?", (pending,)) + second_allowed = True + try: + LlmLedgerGuard(llm_path, cap=1).reserve(1, "second_after_restart") + except RuntimeError: + second_allowed = False + llm_used = LlmLedgerGuard(llm_path, cap=1).used() + check("S4a_aged_unknown_llm_reservation_kept", + (not second_allowed) and llm_used == 1, + f"second_allowed={second_allowed} used_after_restart={llm_used}") + + embed_path = tmp / "aged-embed.sqlite3" + contract.make_ledger(embed_path, cap=1, llm_cap=0) + guard = EmbedLedgerGuard(embed_path, cap=1) + hold = guard.reserve(1, "embed_sent_then_crashed") + with sqlite3.connect(str(embed_path)) as conn: + conn.execute("UPDATE embed_ops SET ts='2000-01-01T00:00:00Z' WHERE id=?", (hold,)) + embed_second = True + try: + EmbedLedgerGuard(embed_path, cap=1).reserve(1, "again") + except RuntimeError: + embed_second = False + embed_used = EmbedLedgerGuard(embed_path, cap=1).used() + EmbedLedgerGuard(embed_path, cap=1).release(hold) # explicit unsent release path + embed_after_release = EmbedLedgerGuard(embed_path, cap=1).used() + check("S4b_aged_unknown_embed_reservation_kept", + (not embed_second) and embed_used == 1 and embed_after_release == 0, + f"second_allowed={embed_second} used={embed_used} after_release={embed_after_release}") + + +def scenario_authority(tmp: Path) -> None: + fresh = tmp / "authority-fresh.sqlite3" + conn = sqlite3.connect(str(fresh)) + with conn: + conn.execute("CREATE TABLE embed_ops (id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT," + " pid INTEGER, kind TEXT, items INTEGER, note TEXT)") + conn.close() + base = {"gateway_base_url": "http://127.0.0.1:9/v1", "gateway_model": "x", + "vector_root": str(tmp / "a-vec"), "history_root": str(tmp / "a-hist"), + "collection": "cg_a", "ledger_path": str(fresh), + "embed_cap": 100, "llm_cap": 1} + refused = "" + try: + Mem0Backend(dict(base)) + except SemanticBudgetError as exc: + refused = str(exc) + empty_block = "" + try: + Mem0Backend({**base, "authority": {}}) + except SemanticBudgetError as exc: + empty_block = str(exc) + check("S0_positive_caps_without_authority_fail_closed", + refused == "ledger_authority_required" and empty_block == "ledger_authority_required", + f"missing={refused} empty={empty_block}") + + +async def main() -> None: + server = contract.start_gateway() + with tempfile.TemporaryDirectory(dir=ROOT / "tools" if (ROOT / "tools").is_dir() + else None) as raw_tmp: + tmp = Path(raw_tmp) + try: + scenario_authority(tmp) + await scenario_partial_and_resume(tmp, server, "p1") + await scenario_lease_and_concurrency(tmp, server) + scenario_aged_reservations(tmp) + finally: + server.shutdown() + server.server_close() + REPORT["gateway_requests_total"] = len(contract.FakeGateway.requests) + REPORT["failures"] = FAILURES + out = Path(sys.argv[1] if len(sys.argv) > 1 else "hel885-recovery-report.json") + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(json.dumps(REPORT, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps({"out": str(out), "failures": FAILURES, + "checks": {c["name"]: c["ok"] for c in REPORT.get("checks", [])}}, + ensure_ascii=False, indent=2)) + sys.exit(1 if FAILURES else 0) + + +asyncio.run(main())