HEL-886 rework: a live worker is never treated as dead, and a fully-completed batch finalizes without new budget - SemanticMemory gains an in-process live-worker registry (batch_id -> _LiveWorker, registered BEFORE the thread starts, so the batch is provably active from the moment the claim commits) and the batch's whole execution (row building, anchor verify, resume set, real adds, terminal sem_batch write) moves INTO the worker thread function, because asyncio.to_thread cannot interrupt a running call: cancelling the tick cancels only the waiter, and the old finally that marked the still-running batch failed and cleared its owner was the exact double-send the manager reproduced live with the pinned SDK (first HTTP accepted but unanswered -> cancel tick -> failed written while the add was on the wire -> second tick re-claimed and sent the same sentence again: 2 HTTP, 2 vectors, 1 row_done entry, the late worker's result dropped); recover() now fails only expired pending batches WITHOUT a live worker, so a lease past 40s with the thread still inside its add is never reclaimed and concurrent ticks cannot re-send an active row (same 1-HTTP/1-vector/1-record outcome on the manager's clock-injection lease variant), while a genuinely dead owner (crash/restart, no registered worker) keeps the honest conservative recovery: unknown wire outcome, ledger holds stay billed (HEL-885), the retry sends only rows without a completion proof, and strict-once is NOT claimed for arbitrary crashes without remote idempotency (module docstring + runbook say so in words); new SemanticMemory.aclose() drains live workers (bounded wait, honest warning on timeout) BEFORE backend/store close so shutdown never orphans durable writes, and the app lifespan now drains through it; Mem0Backend/StubBackend extract the sync core _add_group_sync (async add_group contract unchanged) where the completion set (vector-store anchors + sem_row_done + skip) is proven BEFORE any budget gate: a batch whose every row is already complete returns done with ZERO new HTTP/embed/LLM after restart (the manager's third example: llm_cap=1 fully spent, row really completed, 'row persisted, final status did not' crash window -> was queued/semantic_llm_budget_exhausted, now done with entry_ids intact, cap unchanged, nothing refunded), while any missing row keeps the hard fail-closed gate - finalization never buys headroom; verification: 287 python tests pass on 3.11.16 venv + requirements.lock offline (6 new tests/test_rework_hel886.py cases: in-flight cancel never double-sends / expired lease with live worker never reclaims + concurrent ticks single-send / dead-owner lease still recovers / shutdown drains the live worker before close proven by post-aclose durable writes / all-rows-done restart finalizes with zero new HTTP / partially-done keeps refusing), ruff clean (src/tests/tools), web tsc + 42 vitest + build via npm ci --offline, pinned mem0 2.2.1 on the deterministic boundary (512-dim stand-in installed before TextEmbedding construction, loopback-only): the manager's review_hel885_inflight.py re-run VERBATIM now fails exactly at its worker_done 2-add barrier (the assertion that a second send must happen; before the fix the same script passed with 2 HTTP/2 vectors - baseline JSON kept), the same-shape rerun with the barrier at ONE completed add passes both scenarios with 1 HTTP/1 vector/1 sem_row_done, batch pending while in flight, done with full entry_ids, and the all-done restart finalizes done with zero new HTTP and llm_used still 1; new tools/verify_inflight_lifecycle_hel886.py 6/6 (L1 cancel single-send full record, L2 lease-expired-live no reclaim, L3 shutdown drain before close, L4a all-done finalize without budget, L4b partial keeps the hard gate, L5 dead-owner recovery still works); regressions all green: verify_semantic_recovery_hel885.py 7/7, verify_mem0_real_hel874.py 12/12, check_release_offline_hel882.py R1-R7 15/15, both manager review_hel882_{partial,extra}.py scripts re-run verbatim pass; dual-target requirement checks 4/4 with handoff-HEL-886.json; zero real LLM, zero real embeddings, zero non-loopback outbound, authoritative ledgers and freezes untouched, not deployed, PR not merged, no visual changes
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
zonggong
co-authored by
multica-agent
multica-agent
parent
78b6267d3a
commit
f1b50ad91a
@@ -0,0 +1,51 @@
|
||||
{
|
||||
"baseline_version": "1.11",
|
||||
"read_commit": "78b6267d3a9cc9adc1136fc4bf22c4178a38cab2",
|
||||
"task": "HEL-886 返工（总管对 HEL-885 候选 78b6267 独立复审确认的两个缺口一次修完）：处理中取消与租约回收不得把仍在运行的真实 worker 当死亡而双发同一行（总管实际固定 SDK 复现：取消只取消等待协程、to_thread 中 HTTP 仍在跑时 finally 立即 failed/清 owner，第二 tick 重领并发第二 HTTP；不取消、注入时钟前进 41 秒 recover 同样回收活 worker 重发；两例同一句 2 HTTP/2 向量而批 done 只记 1 entry）、已全完成批次无新余额也须可收尾（总管第三例：llm_cap=1 单行已真实完成，落『行完成已持久、批状态未写』过期 pending 后重启，add_group 在计算 skip 前先 check(1) 导致全完成仍 queued/budget_exhausted）；零真实调用、零真实嵌入、零非回环出站、不部署、不合并、不改视觉、不碰权威三账、不新增任何 cap",
|
||||
"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（基于 78b6267 追加，PR#11 只追加不强推）完成两项修正：1) 活跃 worker 生命周期——SemanticMemory 增进程内活跃注册表（batch_id→_LiveWorker，登记先于线程启动）；批次的行构建/续跑集/真实 add/终态写入全部移入 worker 线程函数 _run（Mem0Backend/StubBackend 抽出同步核心 _add_group_sync，add_group 保持异步合同仅包装）；取消等待协程不再触碰批次状态（旧缺陷：finally 把活跃批次写成 failed 供重领）；recover() 只回收无活跃 worker 的过期 pending（真死 owner 语义保持保守：结果未知、预占不退、重试只发无证明行、不宣称严格一次）；并发 tick 对活跃批次不可领不重发；新增 SemanticMemory.aclose() 排空仍运行 worker 后才关 backend，app lifespan 改走该路径先排空再关 store。2) 完成证明先于预算闸——_add_group_sync 先算 完成集(向量锚点∪sem_row_done)∩剩余工作，全完成零新 HTTP/嵌入即返回完成（重启收尾 done、entry_ids 跨尝试完整、cap 不变不退款）；仍有缺行时维持 ledger/llm_guard 原硬闸 fail-closed。验证：3.11.16 venv+requirements.lock 离线 287 pytest 全过（新增 tests/test_rework_hel886.py 6 例：处理中取消不双发/跨租约仍活跃不回收+并发 tick 不重发/真死 owner 租约照常回收/关闭排空先于关库/全完成零新预算收尾/部分未完成仍硬闸拒绝，全部用屏障阻塞的真实产品路径驱动，不再用预置死租约或快任务并发冒充），ruff 0 错（src/tests/tools），固定 mem0 2.2.1 确定性替身：总管 review_hel885_inflight.py 原样复跑在 2-add 屏障处如预期断言失败（修复前该屏障通过且 2 HTTP/2 向量——双发已消失，唯一调整是把屏障改为 1 次完成的同形脚本 review_hel886_inflight_rerun.py：两场景 1 HTTP/1 向量/1 sem_row_done、飞行中批次保持 pending、终态 done 且 entry_ids 完整；第三例重启后 done、零新 HTTP、llm_used 仍 1、entry_ids 原样）；新增 tools/verify_inflight_lifecycle_hel886.py 6/6（L1 取消单发全记录/L2 跨租约活跃不重发/L3 关闭排空先于关库/L4a 全完成零新预算收尾/L4b 缺行仍硬闸/L5 真死租约照常回收）；verify_semantic_recovery_hel885.py 7/7、verify_mem0_real_hel874.py 12/12、check_release_offline_hel882.py R1–R7 15/15、总管 review_hel882_{partial,extra}.py 原样复跑全过；零真实调用、零真实嵌入、零非回环出站，权威三账与冻结余额未写，未部署未合并未交体验地址，不改视觉。",
|
||||
"acceptance": [
|
||||
"总管 inflight 复现修复（取消例）：第一 HTTP 已接受未返回时取消 tick，批次保持 pending/owner 不被写 failed；并发第二/三 tick 不重发；worker 完成后 1 HTTP、1 向量、1 条 sem_row_done，批 done 且 entry_ids 与行记录一致（docs/rework/hel886/HEL-886-inflight-rerun.json + inflight-real-sdk.json L1）",
|
||||
"总管 inflight 复现修复（租约例）：注入时钟前进 41 秒、worker 仍在第一请求中，recover 不回收活跃批次，并发 tick 不重发；最终同样 1 HTTP/1 向量/1 行记录（L2）",
|
||||
"晚到写入不漏记：飞行中批次不被重领，最终 entry_ids 与 sem_row_done 完整一致，无重复向量、无漏记完成条目（两场景 JSON）",
|
||||
"关闭先排空：aclose 等待仍运行 worker 的持久写落盘后才关向量库 client（L3：排空期间完成的行 done_at/finished_at 证明写在 aclose 开始之后）",
|
||||
"总管第三例修复：llm_cap=1 已用完且单行已真实完成，重启后零新 HTTP、cap 不变即 done、entry_ids 原样；部分未完成仍按原硬闸拒绝（L4a/L4b + tests/test_rework_hel886.py）",
|
||||
"真死进程恢复语义未停用：无活跃 worker 的过期 pending 照常 failed/lease_expired 并重试缺行；未知预占不退款口径不变（L5 + S4a/S4b 复跑）",
|
||||
"工程检查：287 pytest（含 6 新例）、ruff 0（src/tests/tools）、web typecheck/42 vitest/build、双目标需求检查 4/4（含本 handoff）；固定 SDK 合同 12/12、HEL-885 恢复 7/7、发布离线 R1–R7 15/15、总管两脚本原样复跑通过"
|
||||
],
|
||||
"deferred": [
|
||||
"发布与体验地址：由总管独立复审本 head 后交发布卡按 deploy/hel874-release.md 执行（8360 隔离实例）；本卡未部署未合并 PR#11 未交体验地址",
|
||||
"真实语义激活（CYBERGRIL_SEMANTIC=mem0）：需 B 部分预算批准 + 3.11 mem0 栈 wheel 落位；未激活前 stub/FTS 不冒充真实记忆体验",
|
||||
"A/B/C 执行表数值仍为待批建议：批准前所有 cap 保持 0=拒绝；新增预算由总管统一处理，不作为本卡离线修复阻塞",
|
||||
"真死进程『严格一次』不在本卡宣称范围：远端无幂等支持时崩溃结果未知，恢复保守留账、只发无证明行（运行手册 §4 已如实写明）",
|
||||
"R5 语音/R6 天气、素材、备份/R7 有条件观察承接不变；视觉仍交原作者，本卡未改视觉；真实体验效果待体验地址交付后由白栖知判断"
|
||||
],
|
||||
"requirement_changes": [],
|
||||
"verification": [
|
||||
"分支 hel-870-assembly 追加提交（最终 head 见交付评论），PR#11 只追加未强推未合并；视觉总监分支未覆盖",
|
||||
"python 3.11.16 venv + requirements.lock 离线：287 项测试全过（新增 tests/test_rework_hel886.py 6 例，全部屏障阻塞真实产品路径）；ruff 0 错（src/tests/tools）；web typecheck/vitest 42/build 通过（npm ci --offline）",
|
||||
"固定 mem0 2.2.1 安装（hel857-venv 3.10.12）确定性替身实跑（docs/rework/hel886/）：总管 review_hel885_inflight.py 原样复跑在 worker_done 2-add 屏障断言失败（original-verbatim-after-fix.log，修复前基线 HEL-885-inflight-baseline-78b6267.json 为 2 HTTP/2 向量）；调整屏障为 1 的同形脚本 review_hel886_inflight_rerun.py 全过（HEL-886-inflight-rerun.json）；新增 tools/verify_inflight_lifecycle_hel886.py 6/6（inflight-real-sdk.json）",
|
||||
"回归：tools/verify_semantic_recovery_hel885.py 7/7、tools/verify_mem0_real_hel874.py 12/12、tools/check_release_offline_hel882.py R1–R7 15/15、总管 review_hel882_{partial,extra}.py 原样复跑通过（docs/rework/hel886/）",
|
||||
"python3 scripts/check_requirements.py --base-ref origin/main [--handoff docs/requirements/handoff-HEL-886.json] 与 --base-ref origin/agent/agent/de8874a3803b [--handoff …] 四项通过"
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
{
|
||||
"head": "78b6267d3a9cc9adc1136fc4bf22c4178a38cab2",
|
||||
"real_llm_requests": 0,
|
||||
"real_embeddings": 0,
|
||||
"scenarios": [
|
||||
{
|
||||
"kind": "cancel",
|
||||
"before": {
|
||||
"state": "pending",
|
||||
"owner": "21ce5e01a2534b601718ffa14f66a303",
|
||||
"lease_until": 1040000
|
||||
},
|
||||
"accepted_http_before_first_finished": 2,
|
||||
"batch_before_first_finished": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"2edd0645-1f95-453d-b139-a9d80c984363\"]"
|
||||
},
|
||||
"accepted_http_total": 2,
|
||||
"stored_points": 2,
|
||||
"batch_final": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"2edd0645-1f95-453d-b139-a9d80c984363\"]"
|
||||
},
|
||||
"row_done": [
|
||||
{
|
||||
"batch_id": 1,
|
||||
"message_id": 1,
|
||||
"entry_ids": "[\"2edd0645-1f95-453d-b139-a9d80c984363\"]",
|
||||
"done_at": 1000000
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"kind": "lease",
|
||||
"before": {
|
||||
"state": "pending",
|
||||
"owner": "a39018d7e1f4deaa850b01b2df3a2de0",
|
||||
"lease_until": 1040000
|
||||
},
|
||||
"accepted_http_before_first_finished": 2,
|
||||
"batch_before_first_finished": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"2d4a7ff2-3636-4486-9305-f3ae03f6c5fa\"]"
|
||||
},
|
||||
"accepted_http_total": 2,
|
||||
"stored_points": 2,
|
||||
"batch_final": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"2d4a7ff2-3636-4486-9305-f3ae03f6c5fa\"]"
|
||||
},
|
||||
"row_done": [
|
||||
{
|
||||
"batch_id": 1,
|
||||
"message_id": 1,
|
||||
"entry_ids": "[\"2d4a7ff2-3636-4486-9305-f3ae03f6c5fa\"]",
|
||||
"done_at": 1041000
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"all_rows_already_completed": {
|
||||
"before_crash": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"f663dace-fdb1-463d-b76f-fa0fc3c56b31\"]"
|
||||
},
|
||||
"completed_rows": 1,
|
||||
"after_restart": {
|
||||
"state": "queued",
|
||||
"reason": "semantic_llm_budget_exhausted",
|
||||
"entry_ids": "[\"f663dace-fdb1-463d-b76f-fa0fc3c56b31\"]"
|
||||
},
|
||||
"total_http": 1,
|
||||
"llm_cap": 1,
|
||||
"llm_used": 1
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
{
|
||||
"head": "HEL-886 fixed checkout (see git rev in delivery)",
|
||||
"real_llm_requests": 0,
|
||||
"real_embeddings": 0,
|
||||
"scenarios": [
|
||||
{
|
||||
"kind": "cancel",
|
||||
"before": {
|
||||
"state": "pending",
|
||||
"owner": "f2e64f5989b6c88bed2563882663a99d",
|
||||
"lease_until": 1040000
|
||||
},
|
||||
"http_before_first_finished": 1,
|
||||
"batch_before_first_finished": {
|
||||
"state": "pending",
|
||||
"entry_ids": null
|
||||
},
|
||||
"http_total": 1,
|
||||
"stored_points": 1,
|
||||
"batch_final": {
|
||||
"state": "done",
|
||||
"reason": null,
|
||||
"entry_ids": "[\"f8b9effb-b536-4aec-82da-ff3f742ccca1\"]"
|
||||
},
|
||||
"row_done": [
|
||||
{
|
||||
"batch_id": 1,
|
||||
"message_id": 1,
|
||||
"entry_ids": "[\"f8b9effb-b536-4aec-82da-ff3f742ccca1\"]",
|
||||
"done_at": 1000000
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"kind": "lease",
|
||||
"before": {
|
||||
"state": "pending",
|
||||
"owner": "be4ab54a53b5bcafdd3ef27caa5a0656",
|
||||
"lease_until": 1040000
|
||||
},
|
||||
"http_before_first_finished": 1,
|
||||
"batch_before_first_finished": {
|
||||
"state": "pending",
|
||||
"entry_ids": null
|
||||
},
|
||||
"http_total": 1,
|
||||
"stored_points": 1,
|
||||
"batch_final": {
|
||||
"state": "done",
|
||||
"reason": null,
|
||||
"entry_ids": "[\"b14ec3a2-316a-452d-b913-d03980eb35ac\"]"
|
||||
},
|
||||
"row_done": [
|
||||
{
|
||||
"batch_id": 1,
|
||||
"message_id": 1,
|
||||
"entry_ids": "[\"b14ec3a2-316a-452d-b913-d03980eb35ac\"]",
|
||||
"done_at": 1041000
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"all_rows_already_completed": {
|
||||
"before_crash": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"4ffc3d18-c745-42f4-be03-647c24aa05ca\"]"
|
||||
},
|
||||
"completed_rows": 1,
|
||||
"after_restart": {
|
||||
"state": "done",
|
||||
"reason": null,
|
||||
"entry_ids": "[\"4ffc3d18-c745-42f4-be03-647c24aa05ca\"]"
|
||||
},
|
||||
"total_http": 1,
|
||||
"llm_cap": 1,
|
||||
"llm_used": 1
|
||||
},
|
||||
"failures": []
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
{
|
||||
"started": "2026-10-26T16:27:52Z",
|
||||
"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-26T16:27:53Z",
|
||||
"gateway_requests_total": 3,
|
||||
"failures": [],
|
||||
"temp_dir": "/tmp/multica-task-693257945/hel874-real-sdk-1l8rzjt7"
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
{
|
||||
"head": "HEL-886 inflight lifecycle",
|
||||
"real_llm_requests": 0,
|
||||
"real_embeddings": 0,
|
||||
"boundary": "deterministic 512 vectors, non-loopback forbidden",
|
||||
"checks": [
|
||||
{
|
||||
"name": "L1_cancel_single_send_and_full_record",
|
||||
"ok": true,
|
||||
"detail": "arrived=1 concurrent=1 http_total=1 points=1 batch_live={'state': 'pending', 'owner': '55b76b8ed25bdca43442bd451c02e080'} batch_concurrent=pending final={'state': 'done', 'reason': None, 'entry_ids': '[\"d1dfe2cb-7215-47de-972c-33df30263dea\"]'} row_done=1"
|
||||
},
|
||||
{
|
||||
"name": "L2_lease_single_send_and_full_record",
|
||||
"ok": true,
|
||||
"detail": "arrived=1 concurrent=1 http_total=1 points=1 batch_live={'state': 'pending', 'owner': '724fbb7d85503b0b1b1d3854d0b0dcdf'} batch_concurrent=pending final={'state': 'done', 'reason': None, 'entry_ids': '[\"63cd1007-b5dd-499f-8976-a69a52f3065f\"]'} row_done=1"
|
||||
},
|
||||
{
|
||||
"name": "L3_shutdown_drains_live_worker_before_close",
|
||||
"ok": true,
|
||||
"detail": "arrived=1 waited=0.21s row_done_at=2000000 batch={'state': 'done', 'finished_at': 2000000}"
|
||||
},
|
||||
{
|
||||
"name": "L4a_all_done_finalizes_without_new_budget",
|
||||
"ok": true,
|
||||
"detail": "before={'state': 'done', 'entry_ids': '[\"70b6e734-eb26-447e-988e-9fe32c82e930\"]'} after={'state': 'done', 'reason': None, 'entry_ids': '[\"70b6e734-eb26-447e-988e-9fe32c82e930\"]'} http=1 llm_used=1"
|
||||
},
|
||||
{
|
||||
"name": "L4b_partial_keeps_the_hard_gate",
|
||||
"ok": true,
|
||||
"detail": "after={'state': 'queued', 'reason': 'semantic_llm_budget_exhausted'} http=0"
|
||||
},
|
||||
{
|
||||
"name": "L5_dead_owner_lease_still_recovers",
|
||||
"ok": true,
|
||||
"detail": "recovered={'state': 'failed', 'reason': 'lease_expired'} after={'state': 'done'} http=2"
|
||||
}
|
||||
],
|
||||
"gateway_requests_total": 7,
|
||||
"failures": []
|
||||
}
|
||||
@@ -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] <COMMAND>\n\nFor more information, try '--help'."
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
Failed to load spaCy lemma model: spaCy is not installed. Install it with: pip install mem0ai[nlp]
|
||||
Failed to load BM25 encoder: verify boundary forbids non-loopback connection: 198.18.7.201
|
||||
Failed to load spaCy full model: spaCy is not installed. Install it with: pip install mem0ai[nlp]
|
||||
LLM extraction failed: semantic_llm_budget_exhausted
|
||||
semantic batch 1 requeued (budget): LLM extraction failed: semantic_llm_budget_exhausted
|
||||
{
|
||||
"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-26T16:28:34.461368+00:00",
|
||||
"updated_at": "2026-10-26T16:28:34.461368+00:00"
|
||||
}
|
||||
],
|
||||
"claimed_completed_message_ids": [
|
||||
2
|
||||
],
|
||||
"batch_after_resume": {
|
||||
"state": "done",
|
||||
"attempts": 1,
|
||||
"reason": null,
|
||||
"entry_ids": "[\"8c05da20-12aa-439b-8f49-e8860b5fc6ed\", \"41ae5dca-d46f-49cc-a839-9da12632eecf\"]"
|
||||
},
|
||||
"loopback_requests_after_resume": 2,
|
||||
"unsent_user_row_actually_sent": true
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
Failed to load spaCy lemma model: spaCy is not installed. Install it with: pip install mem0ai[nlp]
|
||||
Failed to load BM25 encoder: verify boundary forbids non-loopback connection: 198.18.7.201
|
||||
Failed to load spaCy full model: spaCy is not installed. Install it with: pip install mem0ai[nlp]
|
||||
Traceback (most recent call last):
|
||||
File "/home/xiaomo/multica_workspaces/hello-af23464d222e/hel-886-7c1997665cac/workdir/review_hel885_inflight.py", line 143, in <module>
|
||||
asyncio.run(main())
|
||||
File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
|
||||
return loop.run_until_complete(main)
|
||||
File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
|
||||
return future.result()
|
||||
File "/home/xiaomo/multica_workspaces/hello-af23464d222e/hel-886-7c1997665cac/workdir/review_hel885_inflight.py", line 102, in main
|
||||
'scenarios': [await scenario('cancel'), await scenario('lease')],
|
||||
File "/home/xiaomo/multica_workspaces/hello-af23464d222e/hel-886-7c1997665cac/workdir/review_hel885_inflight.py", line 84, in scenario
|
||||
assert await asyncio.to_thread(worker_done.wait, 8)
|
||||
AssertionError
|
||||
@@ -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=[\"2bee5aad-59b3-4af2-be3a-fd02c0abac74\", \"7767fe14-4a8f-4275-a012-ecfce4d794a8\"]"
|
||||
},
|
||||
{
|
||||
"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": []
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
{
|
||||
"started": "2026-10-26T16:28:24Z",
|
||||
"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": "reused ."
|
||||
},
|
||||
{
|
||||
"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-26T16:28:26Z"
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
Failed to load spaCy lemma model: spaCy is not installed. Install it with: pip install mem0ai[nlp]
|
||||
Failed to load BM25 encoder: verify boundary forbids non-loopback connection: 198.18.7.201
|
||||
Failed to load spaCy full model: spaCy is not installed. Install it with: pip install mem0ai[nlp]
|
||||
Failed to load BM25 encoder: verify boundary forbids non-loopback connection: 198.18.7.201
|
||||
Failed to load BM25 encoder: verify boundary forbids non-loopback connection: 198.18.7.201
|
||||
{
|
||||
"head": "HEL-886 fixed checkout (see git rev in delivery)",
|
||||
"real_llm_requests": 0,
|
||||
"real_embeddings": 0,
|
||||
"scenarios": [
|
||||
{
|
||||
"kind": "cancel",
|
||||
"before": {
|
||||
"state": "pending",
|
||||
"owner": "f2e64f5989b6c88bed2563882663a99d",
|
||||
"lease_until": 1040000
|
||||
},
|
||||
"http_before_first_finished": 1,
|
||||
"batch_before_first_finished": {
|
||||
"state": "pending",
|
||||
"entry_ids": null
|
||||
},
|
||||
"http_total": 1,
|
||||
"stored_points": 1,
|
||||
"batch_final": {
|
||||
"state": "done",
|
||||
"reason": null,
|
||||
"entry_ids": "[\"f8b9effb-b536-4aec-82da-ff3f742ccca1\"]"
|
||||
},
|
||||
"row_done": [
|
||||
{
|
||||
"batch_id": 1,
|
||||
"message_id": 1,
|
||||
"entry_ids": "[\"f8b9effb-b536-4aec-82da-ff3f742ccca1\"]",
|
||||
"done_at": 1000000
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"kind": "lease",
|
||||
"before": {
|
||||
"state": "pending",
|
||||
"owner": "be4ab54a53b5bcafdd3ef27caa5a0656",
|
||||
"lease_until": 1040000
|
||||
},
|
||||
"http_before_first_finished": 1,
|
||||
"batch_before_first_finished": {
|
||||
"state": "pending",
|
||||
"entry_ids": null
|
||||
},
|
||||
"http_total": 1,
|
||||
"stored_points": 1,
|
||||
"batch_final": {
|
||||
"state": "done",
|
||||
"reason": null,
|
||||
"entry_ids": "[\"b14ec3a2-316a-452d-b913-d03980eb35ac\"]"
|
||||
},
|
||||
"row_done": [
|
||||
{
|
||||
"batch_id": 1,
|
||||
"message_id": 1,
|
||||
"entry_ids": "[\"b14ec3a2-316a-452d-b913-d03980eb35ac\"]",
|
||||
"done_at": 1041000
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"all_rows_already_completed": {
|
||||
"before_crash": {
|
||||
"state": "done",
|
||||
"entry_ids": "[\"4ffc3d18-c745-42f4-be03-647c24aa05ca\"]"
|
||||
},
|
||||
"completed_rows": 1,
|
||||
"after_restart": {
|
||||
"state": "done",
|
||||
"reason": null,
|
||||
"entry_ids": "[\"4ffc3d18-c745-42f4-be03-647c24aa05ca\"]"
|
||||
},
|
||||
"total_http": 1,
|
||||
"llm_cap": 1,
|
||||
"llm_used": 1
|
||||
},
|
||||
"failures": []
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
"""HEL-886 rerun of the manager's review_hel885_inflight.py — same scenarios,
|
||||
barrier adjusted to the FIXED contract.
|
||||
|
||||
Adjustment vs the original (fix-before/after documented):
|
||||
- the original sets ``worker_done`` when TWO Memory.add calls complete — that
|
||||
barrier encodes the defect (a second send must happen). Before the fix the
|
||||
script passed with HTTP=2/vectors=2/row_done=1 (see
|
||||
HEL-885-inflight-baseline-78b6267.json); after the fix only ONE add ever
|
||||
happens, so the original's barrier times out (original-verbatim-after-fix.log,
|
||||
AssertionError at worker_done.wait — the exact double-send assertion).
|
||||
This rerun sets the barrier at ONE completed add, keeps all three negative
|
||||
scenarios and adds explicit no-duplicate/no-loss assertions.
|
||||
- cleanup (server shutdown) is unconditional so the process exits.
|
||||
|
||||
Everything else — real pinned SDK, real Qdrant, real OpenAI client,
|
||||
deterministic 512-dim stand-in installed before TextEmbedding construction,
|
||||
loopback-only gateway, temp ledgers — is unchanged.
|
||||
"""
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import json
|
||||
import tempfile
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
REPO = ROOT / 'CyberGril'
|
||||
import sys
|
||||
sys.path.insert(0, str(REPO / 'src'))
|
||||
spec = importlib.util.spec_from_file_location('contract', REPO / 'tools/verify_mem0_real_hel874.py')
|
||||
c = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(c)
|
||||
c.install_deterministic_boundary()
|
||||
from cybergril.memory.semantic import Mem0Backend, SemanticMemory
|
||||
from cybergril.memory.store import Store
|
||||
|
||||
FAILURES = []
|
||||
|
||||
|
||||
async def scenario_fixed(kind):
|
||||
first_arrived, release_first, worker_done = (threading.Event() for _ in range(3))
|
||||
accepted = []
|
||||
lock = threading.Lock()
|
||||
|
||||
class Gateway(c.FakeGateway):
|
||||
def do_POST(self):
|
||||
with lock:
|
||||
accepted.append(self.path)
|
||||
n = len(accepted)
|
||||
if n == 1:
|
||||
first_arrived.set()
|
||||
if not release_first.wait(15):
|
||||
raise RuntimeError('test barrier timed out')
|
||||
super().do_POST()
|
||||
|
||||
server = c._Server(('127.0.0.1', 0), Gateway)
|
||||
server_thread = threading.Thread(target=server.serve_forever)
|
||||
server_thread.start()
|
||||
with tempfile.TemporaryDirectory(dir=ROOT) as td:
|
||||
tmp = Path(td)
|
||||
c.make_ledger(tmp / 'ledger.db', cap=100, llm_cap=10)
|
||||
cfg = c.config_for(tmp, tmp / 'ledger.db', port=server.server_address[1], cap=100, llm_cap=10)
|
||||
backend = Mem0Backend(cfg)
|
||||
store = Store(tmp / 'product.db')
|
||||
clock = [1000.0]
|
||||
live = SemanticMemory(store, backend, clock=lambda: clock[0])
|
||||
add = backend._memory.add
|
||||
completed_calls = []
|
||||
|
||||
def counted_add(*args, **kwargs):
|
||||
try:
|
||||
return add(*args, **kwargs)
|
||||
finally:
|
||||
with lock:
|
||||
completed_calls.append(1)
|
||||
if len(completed_calls) == 1:
|
||||
worker_done.set()
|
||||
|
||||
backend._memory.add = counted_add
|
||||
with store.conn:
|
||||
store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES(1,'user','正在整理的合成原话','test','sent')")
|
||||
live.enqueue_uncovered()
|
||||
task = asyncio.create_task(live.tick())
|
||||
try:
|
||||
assert await asyncio.to_thread(first_arrived.wait, 5)
|
||||
before = dict(store.query_one('SELECT state,owner,lease_until FROM sem_batch'))
|
||||
if kind == 'cancel':
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
else:
|
||||
clock[0] += 41 # deterministic elapsed time, no real sleep
|
||||
await asyncio.wait_for(live.tick(), 8)
|
||||
await asyncio.wait_for(live.tick(), 8)
|
||||
result = {'kind': kind, 'before': before,
|
||||
'http_before_first_finished': len(accepted),
|
||||
'batch_before_first_finished': dict(store.query_one('SELECT state,entry_ids FROM sem_batch'))}
|
||||
finally:
|
||||
release_first.set()
|
||||
if not task.cancelled():
|
||||
await task
|
||||
assert await asyncio.to_thread(worker_done.wait, 8)
|
||||
await asyncio.sleep(0.05)
|
||||
points, _ = backend._client.scroll(collection_name=backend.collection, with_payload=True, limit=100)
|
||||
row_done = [dict(r) for r in store.query('SELECT * FROM sem_row_done')]
|
||||
batch_final = dict(store.query_one('SELECT state,reason,entry_ids FROM sem_batch'))
|
||||
result.update(http_total=len(accepted), stored_points=len(points),
|
||||
batch_final=batch_final, row_done=row_done)
|
||||
await backend.aclose()
|
||||
store.close()
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
server_thread.join()
|
||||
# fixed-contract assertions: one send, one vector, one durable record
|
||||
ok = (len(accepted) == 1 and len(points) == 1 and len(row_done) == 1
|
||||
and batch_final['state'] == 'done'
|
||||
and json.loads(batch_final['entry_ids'] or '[]')
|
||||
== json.loads(row_done[0]['entry_ids']))
|
||||
if not ok:
|
||||
FAILURES.append(f"{kind}: http={len(accepted)} points={len(points)} "
|
||||
f"row_done={len(row_done)} batch={batch_final}")
|
||||
return result
|
||||
|
||||
|
||||
async def all_done_recovery():
|
||||
server = c.start_gateway()
|
||||
with tempfile.TemporaryDirectory(dir=ROOT) as td:
|
||||
tmp = Path(td)
|
||||
c.make_ledger(tmp / 'ledger.db', cap=100, llm_cap=1)
|
||||
cfg = c.config_for(tmp, tmp / 'ledger.db', port=server.server_address[1], cap=100, llm_cap=1)
|
||||
backend = Mem0Backend(cfg)
|
||||
store = Store(tmp / 'product.db')
|
||||
with store.conn:
|
||||
store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES(1,'user','完整处理的合成原话','done','sent')")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
mark = len(c.FakeGateway.requests)
|
||||
await live.tick()
|
||||
original = dict(store.query_one('SELECT state,entry_ids FROM sem_batch'))
|
||||
with store.conn:
|
||||
# Crash window: row completion persisted; final batch status did not.
|
||||
store.execute("UPDATE sem_batch SET state='pending',owner='crashed',lease_until=1")
|
||||
await backend.aclose()
|
||||
store.close()
|
||||
backend = Mem0Backend(cfg)
|
||||
store = Store(tmp / 'product.db')
|
||||
live = SemanticMemory(store, backend)
|
||||
await live.tick()
|
||||
result = {'before_crash': original,
|
||||
'completed_rows': len(store.query('SELECT * FROM sem_row_done')),
|
||||
'after_restart': dict(store.query_one('SELECT state,reason,entry_ids FROM sem_batch')),
|
||||
'total_http': len(c.FakeGateway.requests) - mark,
|
||||
'llm_cap': 1, 'llm_used': backend.llm_guard.used()}
|
||||
await backend.aclose()
|
||||
store.close()
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
ok = (result['after_restart']['state'] == 'done'
|
||||
and result['after_restart']['reason'] is None
|
||||
and result['total_http'] == 1
|
||||
and result['llm_used'] == 1
|
||||
and json.loads(result['after_restart']['entry_ids'] or '[]')
|
||||
== json.loads(original['entry_ids']))
|
||||
if not ok:
|
||||
FAILURES.append(f"all_rows_already_completed: {result}")
|
||||
return result
|
||||
|
||||
|
||||
async def main():
|
||||
result = {'head': 'HEL-886 fixed checkout (see git rev in delivery)',
|
||||
'real_llm_requests': 0, 'real_embeddings': 0,
|
||||
'scenarios': [await scenario_fixed('cancel'), await scenario_fixed('lease')],
|
||||
'all_rows_already_completed': await all_done_recovery(),
|
||||
'failures': FAILURES}
|
||||
(ROOT / 'HEL-886-inflight-rerun.json').write_text(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
if FAILURES:
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -62,6 +62,19 @@ HEL-882 追加：SDK 隐藏重试禁用（max_retries=0）与被应答失败照
|
||||
拒绝重排队不丢行；stub→mem0 覆盖按后端区分；嵌入证据口径勘正见
|
||||
docs/rework/hel874/erratum-hel882.json（旧 runs 有真实本地嵌入，下限 12 条）。
|
||||
|
||||
**HEL-886 处理中生命周期与收尾语义（docs/rework/hel886/，固定安装本体验证）**：
|
||||
|
||||
- 取消/租约不再把「仍在运行的 worker」当死亡。批次的最终状态由 worker 线程
|
||||
自己写（`asyncio.to_thread` 中的真实调用不可中断，取消只取消等待协程）；
|
||||
活跃批次登记在进程内注册表，租约到期但 worker 仍活跃时不回收、并发 tick
|
||||
不重发同一行；关闭时先排空仍在运行的 worker 再关向量库 client 与 SQLite。
|
||||
- 真死进程（崩溃/重启后租约过期）的恢复语义如实保守：旧尝试的出站结果未知
|
||||
——其账本预占保持计费（HEL-885 口径），重试只发没有完成证明的行（向量锚点
|
||||
∪ `sem_row_done`）。不宣称远端无幂等支持下任意崩溃的严格一次。
|
||||
- 已全完成的批次收尾不再要新余额：所有行都有完成证明时，重启后零新
|
||||
HTTP/零新嵌入/cap 不变即 done 且 entry_ids 完整；仍有缺行时维持原硬闸拒绝，
|
||||
不为收尾加 cap 或退款未知预占。
|
||||
|
||||
## 5. 体验链路清单（按用户实际体验顺序；只有实际能力标已接）
|
||||
|
||||
| # | 步骤 | 状态 | 说明 |
|
||||
|
||||
@@ -262,7 +262,9 @@ def create_app(
|
||||
app.state.active["cancel"].set()
|
||||
await app.state.active["finished"].wait()
|
||||
if app.state.semantic:
|
||||
await app.state.semantic.backend.aclose()
|
||||
# HEL-886: drain live batch workers BEFORE the vector-store
|
||||
# client / SQLite store close under their durable writes.
|
||||
await app.state.semantic.aclose()
|
||||
for provider in set(router.provider_for(t) for t in router.available_tiers()):
|
||||
await provider.aclose()
|
||||
store.close()
|
||||
|
||||
+266
-117
@@ -55,6 +55,22 @@ Product port of the proven HEL-857/861–869 isolation results (plan-next-HEL-86
|
||||
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.
|
||||
- Live-worker lifecycle (HEL-886): the batch's durable fate is written by its
|
||||
worker THREAD, not by the coroutine that started it. Cancelling the waiter
|
||||
cancels only the wait — the thread finishes, records ``sem_row_done`` per
|
||||
row and the terminal ``sem_batch`` state itself. A lease that expires while
|
||||
its worker is still registered LIVE is never reclaimed (a blocked HTTP call
|
||||
is not a dead owner) and no concurrent tick resends rows of an active
|
||||
batch; shutdown drains live workers before closing the vector-store client
|
||||
and the SQLite store under them. For genuinely dead processes the recovery
|
||||
stays honest: the crashed attempt's wire outcome is UNKNOWN (its ledger
|
||||
holds stay billed), the retry sends only rows without a completion proof,
|
||||
and strict-once delivery is NOT claimed for arbitrary crashes.
|
||||
- Completed-work finalization needs no new budget (HEL-886): a batch whose
|
||||
every row already carries a completion proof finalizes to ``done`` with
|
||||
zero new HTTP/embed/LLM — recovery of the "row persisted, final status did
|
||||
not" crash window no longer demands balance. Rows still missing keep the
|
||||
hard fail-closed gate; finalization never raises a cap or refunds 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.
|
||||
"""
|
||||
@@ -65,6 +81,7 @@ import hashlib
|
||||
import json
|
||||
import logging
|
||||
import secrets
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -158,11 +175,13 @@ class StubBackend:
|
||||
self._entries: dict[str, dict] = {}
|
||||
self._seq = 0
|
||||
|
||||
async def add_group(self, rows: list[dict], anchor: dict, batch_key: str, *,
|
||||
def _add_group_sync(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."""
|
||||
that row's own anchor (HEL-885) — orchestration progress parity. The
|
||||
sync core runs inside the HEL-886 batch worker thread like the real
|
||||
backend's."""
|
||||
|
||||
def _add() -> list[dict]:
|
||||
done = self.completed_message_ids(batch_key) | set(skip)
|
||||
@@ -186,7 +205,13 @@ class StubBackend:
|
||||
on_row_done(row, [entry])
|
||||
return created
|
||||
|
||||
return await asyncio.to_thread(_add)
|
||||
return _add()
|
||||
|
||||
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]:
|
||||
return await asyncio.to_thread(
|
||||
self._add_group_sync, rows, anchor, batch_key, skip=skip, on_row_done=on_row_done)
|
||||
|
||||
async def search(self, query: str, limit: int = 5) -> list[dict]:
|
||||
def _search() -> list[dict]:
|
||||
@@ -796,6 +821,61 @@ class Mem0Backend:
|
||||
break
|
||||
return done
|
||||
|
||||
def _add_group_sync(self, rows: list[dict], anchor: dict, batch_key: str, *,
|
||||
skip: set[int] | frozenset[int] = frozenset(),
|
||||
on_row_done=None) -> list[dict]:
|
||||
"""Synchronous core of ``add_group`` — callable from an already-running
|
||||
worker thread (the HEL-886 batch worker owns the whole execution)."""
|
||||
# HEL-886: prove the remaining work BEFORE any budget gate. A batch
|
||||
# whose every row already carries a completion proof (vector anchor or
|
||||
# sem_row_done) finalizes with ZERO new HTTP/embed/LLM — recovery of a
|
||||
# crash window where "row done persisted, final status did not" must
|
||||
# not demand new balance. Rows still missing keep the hard gate below:
|
||||
# finalization never buys headroom, refunds nothing, raises no cap.
|
||||
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 []
|
||||
# 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)
|
||||
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, **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"]:
|
||||
row_entries.append(entry)
|
||||
created.extend(row_entries)
|
||||
if on_row_done is not None:
|
||||
on_row_done(row, row_entries)
|
||||
return created
|
||||
|
||||
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]:
|
||||
@@ -808,56 +888,8 @@ class Mem0Backend:
|
||||
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]:
|
||||
# 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 []
|
||||
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, **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"]:
|
||||
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)
|
||||
return await asyncio.to_thread(
|
||||
self._add_group_sync, rows, anchor, batch_key, skip=skip, on_row_done=on_row_done)
|
||||
|
||||
async def search(self, query: str, limit: int = 5) -> list[dict]:
|
||||
self.ledger.check(1) # the query embed is accounted inside the wrapper
|
||||
@@ -918,12 +950,37 @@ def _ledger_debit(self, items: int, note: str) -> None: # pragma: no cover —
|
||||
EmbedLedgerGuard.debit = _ledger_debit
|
||||
|
||||
|
||||
class _LiveWorker:
|
||||
"""In-process handle on a batch worker thread that is REALLY running.
|
||||
|
||||
HEL-886: the batch's durable fate belongs to its worker thread, never to
|
||||
the coroutine that happened to start it. A task cancellation cancels only
|
||||
the waiter (``asyncio.to_thread`` cannot interrupt the running thread);
|
||||
the thread finishes its rows, writes ``sem_row_done`` and the terminal
|
||||
``sem_batch`` state itself, then signals ``done`` and unregisters. As long
|
||||
as this handle is registered, the batch is actively owned — lease expiry
|
||||
must NOT reclaim it (a slow/blocked call is not a dead owner) and no
|
||||
concurrent tick may resend its rows.
|
||||
"""
|
||||
|
||||
__slots__ = ("batch_id", "owner", "done")
|
||||
|
||||
def __init__(self, batch_id: int, owner: str) -> None:
|
||||
self.batch_id, self.owner = batch_id, owner
|
||||
self.done = threading.Event()
|
||||
|
||||
|
||||
class SemanticMemory:
|
||||
"""Owns the batch queue, the anchors and the fail-closed read gate."""
|
||||
|
||||
def __init__(self, store, backend, *, clock=time.time) -> None:
|
||||
self.store, self.backend, self.clock = store, backend, clock
|
||||
self.public = PublicView(store)
|
||||
# HEL-886 live-worker registry: batch_id → handle. Guarded by its own
|
||||
# lock — the registry is read/written from event loop AND worker
|
||||
# threads while the SQLite paths keep their own transactions.
|
||||
self._live: dict[int, _LiveWorker] = {}
|
||||
self._live_lock = threading.Lock()
|
||||
|
||||
# ------------------------------------------------------------ enqueue
|
||||
def public_rows_for_turn(self, turn_id: str) -> list[dict]:
|
||||
@@ -997,10 +1054,34 @@ class SemanticMemory:
|
||||
|
||||
# ------------------------------------------------------------ job loop
|
||||
def recover(self) -> None:
|
||||
"""Fail expired pending batches whose owner is GENUINELY gone.
|
||||
|
||||
HEL-886: a lease expiring while THIS process still runs the batch's
|
||||
worker thread proves nothing about death — the worker may be blocked
|
||||
mid-HTTP and will write its durable fate when it returns. Batches with
|
||||
a registered live worker are never touched here. For the truly dead
|
||||
case (process crash / restart) the semantics stay honest and
|
||||
conservative: the old attempt's wire outcome is UNKNOWN — its ledger
|
||||
reservations stay billed (HEL-885), the retry sends only rows without
|
||||
a completion proof (vector anchor ∪ ``sem_row_done``), and no
|
||||
strict-once delivery is claimed for arbitrary crashes.
|
||||
"""
|
||||
now = int(self.clock() * 1000)
|
||||
expired = [r["id"] for r in self.store.query(
|
||||
"SELECT id FROM sem_batch WHERE state='pending' AND lease_until IS NOT NULL "
|
||||
"AND lease_until<?", (now,))]
|
||||
if not expired:
|
||||
return
|
||||
with self._live_lock:
|
||||
dead = [batch_id for batch_id in expired if batch_id not in self._live]
|
||||
if not dead:
|
||||
return
|
||||
marks = ",".join("?" * len(dead))
|
||||
with self.store.conn:
|
||||
self.store.execute(
|
||||
"UPDATE sem_batch SET state='failed',reason='lease_expired',owner=NULL,finished_at=? "
|
||||
"WHERE state='pending' AND lease_until<?", (int(self.clock() * 1000), int(self.clock() * 1000)),
|
||||
f"UPDATE sem_batch SET state='failed',reason='lease_expired',owner=NULL,"
|
||||
f"finished_at=? WHERE state='pending' AND lease_until<? AND id IN ({marks})",
|
||||
(now, now, *dead),
|
||||
)
|
||||
|
||||
async def tick(self) -> None:
|
||||
@@ -1008,17 +1089,31 @@ class SemanticMemory:
|
||||
owner, now = secrets.token_hex(16), int(self.clock() * 1000)
|
||||
with self.store.conn:
|
||||
self.store.execute("BEGIN IMMEDIATE")
|
||||
job = self.store.query_one(
|
||||
"SELECT * FROM sem_batch WHERE state IN ('queued','failed') AND attempts<3 ORDER BY id LIMIT 1"
|
||||
)
|
||||
if not job:
|
||||
return
|
||||
job, offset = None, 0
|
||||
while job is None:
|
||||
candidate = self.store.query_one(
|
||||
"SELECT * FROM sem_batch WHERE state IN ('queued','failed') AND attempts<3 "
|
||||
"ORDER BY id LIMIT 1 OFFSET ?", (offset,),
|
||||
)
|
||||
if not candidate:
|
||||
return
|
||||
with self._live_lock:
|
||||
# Defensive: a live worker's batch is unclaimable whatever
|
||||
# its row state says — active ownership beats the table.
|
||||
if candidate["id"] not in self._live:
|
||||
job = candidate
|
||||
else:
|
||||
offset += 1
|
||||
self.store.execute(
|
||||
"UPDATE sem_batch SET state='pending',attempts=attempts+1,owner=?,lease_until=?,reason=NULL "
|
||||
"WHERE id=?", (owner, now + 40000, job["id"]),
|
||||
)
|
||||
state, reason, entry_ids = "failed", "extraction_failed", None
|
||||
row_entries: list[str] = []
|
||||
# Registered BEFORE the thread exists, so the batch is provably active
|
||||
# from the moment the claim commits — no recover() in a concurrent
|
||||
# tick can reclaim it in the gap.
|
||||
handle = _LiveWorker(job["id"], owner)
|
||||
with self._live_lock:
|
||||
self._live[job["id"]] = handle
|
||||
|
||||
def _row_done(row: dict, entries: list[dict]) -> None:
|
||||
"""Durable per-row completion, written inside the worker thread the
|
||||
@@ -1033,63 +1128,117 @@ class SemanticMemory:
|
||||
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"])
|
||||
|
||||
def _run() -> None:
|
||||
"""The WHOLE batch execution, owned by the worker thread (HEL-886).
|
||||
|
||||
Row building, resume-set computation, the real adds and the
|
||||
terminal ``sem_batch`` write all live here, so cancelling the
|
||||
awaiting coroutine can no longer mark a still-running batch
|
||||
failed/abandoned — that was the double-send defect: the cancelled
|
||||
waiter wrote ``failed``, a second tick reclaimed the batch and
|
||||
re-sent rows whose first HTTP was still in flight.
|
||||
"""
|
||||
state, reason, entry_ids = "failed", "extraction_failed", None
|
||||
row_entries: list[str] = []
|
||||
try:
|
||||
anchor = json.loads(job["anchor_json"])
|
||||
rows = []
|
||||
for turn_id, role, message_id in anchor["turn_roles"]:
|
||||
message = self.store.query_one(
|
||||
"SELECT * FROM chat_message WHERE id=? AND turn_id=? AND role=? AND status='sent'",
|
||||
(message_id, turn_id, role),
|
||||
)
|
||||
if not message:
|
||||
raise ValueError("archive_missing")
|
||||
rows.append({"message_id": message_id, "turn_id": turn_id,
|
||||
"role": role, "text": message["text"]})
|
||||
if anchor_fingerprint(rows) != anchor["input_sha256"]:
|
||||
raise ValueError("source_mismatch")
|
||||
# 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 = self.backend._add_group_sync(
|
||||
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__
|
||||
if "budget" in reason:
|
||||
# A refused budget is not an extraction failure: the batch goes
|
||||
# back to queued WITHOUT burning an attempt — raising the cap
|
||||
# later resumes it (skipping rows already in the vector store),
|
||||
# so exhausted budget can never permanently lose these rows.
|
||||
with self.store.conn:
|
||||
self.store.execute(
|
||||
"UPDATE sem_batch SET state='queued',reason=?,owner=NULL,"
|
||||
"attempts=MAX(0,attempts-1),lease_until=NULL WHERE id=? AND owner=?",
|
||||
(reason, job["id"], owner),
|
||||
)
|
||||
log.warning("semantic batch %s requeued (budget): %s", job["id"], reason)
|
||||
else:
|
||||
log.warning("semantic batch %s failed: %s", job["id"], reason)
|
||||
finally:
|
||||
try:
|
||||
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=? "
|
||||
"WHERE id=? AND owner=? AND state='pending'",
|
||||
(state, reason, entry_ids, int(self.clock() * 1000), job["id"], owner),
|
||||
)
|
||||
finally:
|
||||
# Unregister LAST: until this runs, recover() and every
|
||||
# concurrent tick treat the batch as actively owned.
|
||||
handle.done.set()
|
||||
with self._live_lock:
|
||||
self._live.pop(job["id"], None)
|
||||
|
||||
try:
|
||||
anchor = json.loads(job["anchor_json"])
|
||||
rows = []
|
||||
for turn_id, role, message_id in anchor["turn_roles"]:
|
||||
message = self.store.query_one(
|
||||
"SELECT * FROM chat_message WHERE id=? AND turn_id=? AND role=? AND status='sent'",
|
||||
(message_id, turn_id, role),
|
||||
)
|
||||
if not message:
|
||||
raise ValueError("archive_missing")
|
||||
rows.append({"message_id": message_id, "turn_id": turn_id,
|
||||
"role": role, "text": message["text"]})
|
||||
if anchor_fingerprint(rows) != anchor["input_sha256"]:
|
||||
raise ValueError("source_mismatch")
|
||||
# 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__
|
||||
if "budget" in reason:
|
||||
# A refused budget is not an extraction failure: the batch goes
|
||||
# back to queued WITHOUT burning an attempt — raising the cap
|
||||
# later resumes it (skipping rows already in the vector store),
|
||||
# so exhausted budget can never permanently lose these rows.
|
||||
with self.store.conn:
|
||||
self.store.execute(
|
||||
"UPDATE sem_batch SET state='queued',reason=?,owner=NULL,"
|
||||
"attempts=MAX(0,attempts-1),lease_until=NULL WHERE id=? AND owner=?",
|
||||
(reason, job["id"], owner),
|
||||
)
|
||||
log.warning("semantic batch %s requeued (budget): %s", job["id"], reason)
|
||||
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=? "
|
||||
"WHERE id=? AND owner=? AND state='pending'",
|
||||
(state, reason, entry_ids, int(self.clock() * 1000), job["id"], owner),
|
||||
)
|
||||
await asyncio.to_thread(_run)
|
||||
except asyncio.CancelledError:
|
||||
# Only the WAITER was cancelled. The worker thread keeps running
|
||||
# and writes the batch's durable fate itself; abandoning it as
|
||||
# "failed" here would let the next tick double-send rows whose
|
||||
# first HTTP is still in flight (the HEL-886 defect). Nothing to
|
||||
# clean up — _run owns the registry entry.
|
||||
raise
|
||||
|
||||
async def aclose(self, *, drain_timeout: float = 120.0) -> None:
|
||||
"""Shutdown order (HEL-886): drain OUR still-running workers first.
|
||||
|
||||
A live worker holds an open HTTP call and will write ``sem_row_done``
|
||||
and the terminal batch state when it returns; closing the vector-store
|
||||
client or the SQLite store under it would orphan exactly those durable
|
||||
writes. Waits up to ``drain_timeout`` seconds, then closes the backend
|
||||
anyway with an honest warning — the workers' registry entries stay
|
||||
conservative (their rows resend only without a completion proof).
|
||||
"""
|
||||
deadline = time.monotonic() + drain_timeout
|
||||
while True:
|
||||
with self._live_lock:
|
||||
handles = list(self._live.values())
|
||||
if not handles:
|
||||
break
|
||||
left = deadline - time.monotonic()
|
||||
if left <= 0:
|
||||
log.warning("semantic shutdown drain: %d live batch worker(s) still running "
|
||||
"after %.0fs — closing anyway, their outcome stays unknown",
|
||||
len(handles), drain_timeout)
|
||||
break
|
||||
await asyncio.to_thread(
|
||||
lambda hs=handles, lv=left: [h.done.wait(min(lv, 1.0)) for h in hs])
|
||||
await self.backend.aclose()
|
||||
|
||||
# ------------------------------------------------------------ read gate
|
||||
def chain_resolve(self, hit: dict) -> dict:
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
"""HEL-886 rework regressions: the manager's independent-review failures.
|
||||
|
||||
Two defects against the real pinned SDK (review_hel885_inflight.py):
|
||||
|
||||
P1 — a cancelled waiter or an expired lease treated a STILL-RUNNING worker as
|
||||
dead: the finally wrote ``failed``, the next tick reclaimed the batch and
|
||||
re-sent the same row while its first HTTP was in flight (2 HTTP, 2 vectors,
|
||||
1 row_done entry — the late worker's result dropped).
|
||||
P2 — ``add_group`` checked the ledger BEFORE proving remaining work, so a
|
||||
batch whose every row was already complete could not finalize after restart
|
||||
without new balance (``semantic_llm_budget_exhausted`` instead of ``done``).
|
||||
|
||||
These cases drive the REAL product path with barrier-blocked workers (an
|
||||
in-flight HTTP call), not preseeded dead leases or fast racing tasks. The
|
||||
pinned real install is covered live by tools/verify_inflight_lifecycle_hel886.py.
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from contextlib import suppress
|
||||
|
||||
from test_mem0_backend import FakeLlm, FakeMemory, _make_ledger
|
||||
from test_rework_hel885 import seed_public_two_row_turn
|
||||
from test_semantic_recovery_hel882 import config_for, install_fake
|
||||
|
||||
from cybergril.memory.semantic import Mem0Backend, SemanticMemory
|
||||
from cybergril.memory.store import Store
|
||||
|
||||
|
||||
class BlockingLlm(FakeLlm):
|
||||
"""The manager's repro shape: the extraction HTTP reached the gateway but
|
||||
the answer has not come back. ``attempts`` records every wire entry (the
|
||||
in-flight one included); ``calls`` only completed ones, like FakeLlm."""
|
||||
|
||||
def __init__(self, arrived: threading.Event, release: threading.Event) -> None:
|
||||
super().__init__()
|
||||
self.arrived, self.release = arrived, release
|
||||
self.attempts: list[list] = []
|
||||
|
||||
def generate_response(self, messages=None, **kwargs):
|
||||
self.attempts.append(messages)
|
||||
self.arrived.set()
|
||||
if not self.release.wait(10):
|
||||
raise RuntimeError("test barrier timed out")
|
||||
return super().generate_response(messages=messages, **kwargs)
|
||||
|
||||
|
||||
class BlockingMemory(FakeMemory):
|
||||
gate: tuple[threading.Event, threading.Event] | None = None
|
||||
|
||||
def __init__(self, embedder, collection):
|
||||
super().__init__(embedder, collection)
|
||||
self.llm = BlockingLlm(*BlockingMemory.gate)
|
||||
|
||||
|
||||
def install_blocking(monkeypatch, arrived, release):
|
||||
BlockingMemory.gate = (arrived, release)
|
||||
return install_fake(monkeypatch, BlockingMemory)
|
||||
|
||||
|
||||
def seed_user_row(store, turn_id, text):
|
||||
with store.conn:
|
||||
store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES"
|
||||
"(1,'user',?,?,'sent')", (text, turn_id))
|
||||
return store.query_one("SELECT id FROM chat_message WHERE text=?", (text,))["id"]
|
||||
|
||||
|
||||
def batch_row(store):
|
||||
return dict(store.query_one(
|
||||
"SELECT state,reason,owner,entry_ids,finished_at FROM sem_batch"))
|
||||
|
||||
|
||||
async def test_inflight_cancel_never_double_sends(tmp_path, monkeypatch):
|
||||
"""Cancel the tick WHILE the row's HTTP is in flight: the worker thread
|
||||
owns the batch fate, the second tick must not resend, and the late worker
|
||||
result is durably recorded — one send, one vector, one row_done."""
|
||||
arrived, release = threading.Event(), threading.Event()
|
||||
install_blocking(monkeypatch, arrived, release)
|
||||
ledger = tmp_path / "cancel.sqlite3"
|
||||
_make_ledger(ledger)
|
||||
backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=10))
|
||||
store = Store(tmp_path / "cancel.db")
|
||||
seed_user_row(store, "t-cancel", "处理中取消的合成原话")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
task = asyncio.create_task(live.tick())
|
||||
assert await asyncio.to_thread(arrived.wait, 5) # HTTP accepted, not answered
|
||||
|
||||
task.cancel() # cancels the WAITER only — the worker thread keeps running
|
||||
with suppress(asyncio.CancelledError):
|
||||
await task
|
||||
after_cancel = batch_row(store)
|
||||
# THE defect: the cancelled waiter used to write failed here, freeing the
|
||||
# batch for reclaim while its add was still on the wire.
|
||||
assert after_cancel["state"] == "pending" and after_cancel["owner"]
|
||||
|
||||
# ticks while the worker is live: nothing claimable, nothing re-sent
|
||||
await asyncio.gather(live.tick(), live.tick())
|
||||
assert len(backend._memory.llm.attempts) == 1
|
||||
|
||||
release.set()
|
||||
await live.aclose(drain_timeout=10) # shutdown drains the live worker
|
||||
final = batch_row(store)
|
||||
rows = store.query("SELECT * FROM sem_row_done")
|
||||
assert final["state"] == "done" and final["reason"] is None
|
||||
assert len(backend._memory.llm.attempts) == 1 # still exactly one send
|
||||
assert len(backend._memory.llm.calls) == 1
|
||||
assert len(rows) == 1 and len(json.loads(rows[0]["entry_ids"])) == 1
|
||||
assert json.loads(final["entry_ids"]) == json.loads(rows[0]["entry_ids"])
|
||||
store.close()
|
||||
|
||||
|
||||
async def test_expired_lease_with_live_worker_never_reclaims(tmp_path, monkeypatch):
|
||||
"""Lease past 40s while the worker is REALLY still in the add: recover()
|
||||
leaves the batch alone, concurrent ticks cannot resend the same row."""
|
||||
arrived, release = threading.Event(), threading.Event()
|
||||
install_blocking(monkeypatch, arrived, release)
|
||||
ledger = tmp_path / "lease.sqlite3"
|
||||
_make_ledger(ledger)
|
||||
backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=10))
|
||||
store = Store(tmp_path / "lease.db")
|
||||
seed_user_row(store, "t-lease", "跨租约仍活跃的合成原话")
|
||||
clock = [1000.0]
|
||||
live = SemanticMemory(store, backend, clock=lambda: clock[0])
|
||||
live.enqueue_uncovered()
|
||||
tick_task = asyncio.create_task(live.tick())
|
||||
assert await asyncio.to_thread(arrived.wait, 5)
|
||||
clock[0] += 41 # lease expired on paper — the worker is NOT dead
|
||||
|
||||
live.recover()
|
||||
assert batch_row(store)["state"] == "pending" # never failed while live
|
||||
|
||||
await asyncio.gather(live.tick(), live.tick(), live.tick())
|
||||
assert len(backend._memory.llm.attempts) == 1
|
||||
|
||||
release.set()
|
||||
await live.aclose(drain_timeout=10)
|
||||
with suppress(asyncio.CancelledError):
|
||||
await tick_task
|
||||
final = batch_row(store)
|
||||
assert final["state"] == "done"
|
||||
assert len(backend._memory.llm.attempts) == 1
|
||||
assert len(store.query("SELECT * FROM sem_row_done")) == 1
|
||||
assert len(json.loads(final["entry_ids"])) == 1
|
||||
store.close()
|
||||
|
||||
|
||||
def test_dead_owner_lease_still_recovers(tmp_path, monkeypatch):
|
||||
"""Guard the other direction: a lease held by a GENUINELY dead owner (no
|
||||
live worker — the crash/restart case) still recovers and retries."""
|
||||
install_fake(monkeypatch)
|
||||
ledger = tmp_path / "dead.sqlite3"
|
||||
_make_ledger(ledger)
|
||||
backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=50))
|
||||
store = Store(tmp_path / "dead.db")
|
||||
seed_public_two_row_turn(store, "t-dead", "真正死亡第一行", "真正死亡第二行")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
with store.conn: # a worker died holding the lease
|
||||
store.execute("UPDATE sem_batch SET state='pending',owner='dead',lease_until=1")
|
||||
live.recover()
|
||||
job = batch_row(store)
|
||||
assert job["state"] == "failed" and job["reason"] == "lease_expired"
|
||||
# unknown-outcome honesty: the retry sends only rows without a proof;
|
||||
# here that is both rows (nothing was proven complete)
|
||||
calls_before = len(backend._memory.llm.calls)
|
||||
asyncio.run(_drain_tick(live))
|
||||
after = batch_row(store)
|
||||
assert after["state"] == "done"
|
||||
assert len(backend._memory.llm.calls) - calls_before == 2
|
||||
store.close()
|
||||
|
||||
|
||||
async def _drain_tick(live):
|
||||
await live.tick()
|
||||
await live.aclose(drain_timeout=5)
|
||||
|
||||
|
||||
async def test_shutdown_drains_live_worker_before_close(tmp_path, monkeypatch):
|
||||
"""aclose() waits for the still-running worker's durable writes BEFORE
|
||||
the vector-store client closes under them."""
|
||||
arrived, release = threading.Event(), threading.Event()
|
||||
install_blocking(monkeypatch, arrived, release)
|
||||
ledger = tmp_path / "drain.sqlite3"
|
||||
_make_ledger(ledger)
|
||||
backend = Mem0Backend(config_for(tmp_path, ledger, embed_cap=50, llm_cap=10))
|
||||
store = Store(tmp_path / "drain.db")
|
||||
seed_user_row(store, "t-drain", "关闭排空的合成原话")
|
||||
clock = [1000.0]
|
||||
live = SemanticMemory(store, backend, clock=lambda: clock[0])
|
||||
live.enqueue_uncovered()
|
||||
task = asyncio.create_task(live.tick())
|
||||
assert await asyncio.to_thread(arrived.wait, 5)
|
||||
|
||||
def _finish(): # the wire answer arrives while shutdown is already waiting
|
||||
clock[0] = 2000 # any durable write after this proves it postdates aclose
|
||||
release.set()
|
||||
|
||||
timer = threading.Timer(0.2, _finish)
|
||||
timer.start()
|
||||
started = time.monotonic()
|
||||
await live.aclose(drain_timeout=10)
|
||||
assert time.monotonic() >= started + 0.15 # aclose actually waited
|
||||
row = store.query_one("SELECT * FROM sem_row_done")
|
||||
batch = batch_row(store)
|
||||
assert row is not None and row["done_at"] >= 2000 # written after aclose began
|
||||
assert batch["state"] == "done" and batch["finished_at"] >= 2000
|
||||
assert backend._client.closed # closed only after the drain
|
||||
with suppress(asyncio.CancelledError, Exception):
|
||||
await task
|
||||
timer.join()
|
||||
store.close()
|
||||
|
||||
|
||||
async def test_all_rows_done_batch_finalizes_without_new_budget(tmp_path, monkeypatch):
|
||||
"""P2 (manager third example): llm_cap=1 fully spent, the single row
|
||||
completed for real; a restart into the crash window 'row done persisted,
|
||||
final status did not' finalizes done with ZERO new HTTP and the complete
|
||||
entry_ids — no cap raise, no refund."""
|
||||
install_fake(monkeypatch)
|
||||
ledger = tmp_path / "alldone.sqlite3"
|
||||
_make_ledger(ledger)
|
||||
cfg = config_for(tmp_path, ledger, embed_cap=50, llm_cap=1)
|
||||
backend = Mem0Backend(cfg)
|
||||
db = tmp_path / "alldone.db"
|
||||
store = Store(db)
|
||||
seed_user_row(store, "t-done", "完整处理的合成原话")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
await live.tick()
|
||||
original = batch_row(store)
|
||||
assert original["state"] == "done" and json.loads(original["entry_ids"])
|
||||
with store.conn: # crash window: final batch status never landed
|
||||
store.execute("UPDATE sem_batch SET state='pending',owner='crashed',lease_until=1")
|
||||
store.close()
|
||||
await backend.aclose()
|
||||
|
||||
backend2 = Mem0Backend(cfg) # same paths, same caps — nothing new approved
|
||||
store2 = Store(db)
|
||||
live2 = SemanticMemory(store2, backend2)
|
||||
calls_before = len(backend2._memory.llm.calls)
|
||||
await live2.tick()
|
||||
after = batch_row(store2)
|
||||
assert after["state"] == "done" and after["reason"] is None
|
||||
assert len(backend2._memory.llm.calls) == calls_before # zero new HTTP
|
||||
assert json.loads(after["entry_ids"]) == json.loads(original["entry_ids"])
|
||||
assert backend2.llm_guard.used() == 1 # cap unchanged, nothing refunded
|
||||
store2.close()
|
||||
await backend2.aclose()
|
||||
|
||||
|
||||
async def test_partially_done_batch_keeps_the_hard_gate(tmp_path, monkeypatch):
|
||||
"""The P2 fix must not buy headroom: with a row still missing and the
|
||||
budget exhausted, restart keeps refusing (queued/budget) — never a fake
|
||||
done, never an extra send past the cap."""
|
||||
install_fake(monkeypatch)
|
||||
ledger = tmp_path / "partial.sqlite3"
|
||||
_make_ledger(ledger)
|
||||
cfg = config_for(tmp_path, ledger, embed_cap=50, llm_cap=1)
|
||||
backend = Mem0Backend(cfg)
|
||||
db = tmp_path / "partial.db"
|
||||
store = Store(db)
|
||||
seed_public_two_row_turn(store, "t-partial", "部分完成用户行", "部分完成伴侣行")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
await live.tick() # one row really sent; the other budget-refused
|
||||
first = batch_row(store)
|
||||
assert first["state"] == "queued" and "budget" in (first["reason"] or "")
|
||||
with store.conn: # crash window on top of the budget refusal
|
||||
store.execute("UPDATE sem_batch SET state='pending',owner='crashed',lease_until=1")
|
||||
store.close()
|
||||
await backend.aclose()
|
||||
|
||||
backend2 = Mem0Backend(cfg)
|
||||
store2 = Store(db)
|
||||
live2 = SemanticMemory(store2, backend2)
|
||||
calls_before = len(backend2._memory.llm.calls)
|
||||
await live2.tick()
|
||||
after = batch_row(store2)
|
||||
assert after["state"] == "queued" and "budget" in (after["reason"] or "")
|
||||
assert len(backend2._memory.llm.calls) == calls_before # gate held the wire
|
||||
store2.close()
|
||||
await backend2.aclose()
|
||||
@@ -0,0 +1,326 @@
|
||||
#!/usr/bin/env python3
|
||||
"""HEL-886 · 实际固定 SDK 处理中生命周期验证（零真实出站、零真实嵌入、零权威账写入）。
|
||||
|
||||
在固定 mem0ai 2.2.1（hel857-venv）+ 真实嵌入式 Qdrant + 真实 OpenAI 客户端上复现
|
||||
总管独立复审（review_hel885_inflight.py）的两个缺口并证明修复。边界口径与
|
||||
tools/verify_mem0_real_hel874.py 相同：确定性 512 维替身（TextEmbedding 构造前
|
||||
安装）、127.0.0.1 假网关、临时账/库。回环网关会阻塞第一份 HTTP 应答，模拟
|
||||
「第一请求已被服务端接受但尚未返回」，全部场景在同一进程内推进。
|
||||
|
||||
L1 处理中取消：取消只取消等待协程；worker 线程完成并发送终态。第二/第三次
|
||||
tick 不重发；最终 1 HTTP、1 向量、1 条 sem_row_done、批次 done 且 entry_ids
|
||||
完整（修复前：2 HTTP、2 向量、只记 1 条）。
|
||||
L2 跨租约仍活跃：注入时钟前进 41 秒，worker 仍在第一请求里；recover 不回收
|
||||
活跃批次，并发 tick 不重发（修复前：同样双发）。
|
||||
L3 关闭排空：aclose 先排空仍在运行的 worker（其后才关 vector client），
|
||||
worker 的持久写在排空等待期间落盘。
|
||||
L4a 全行已完成的重启收尾：llm_cap=1 全部用完且单行已真实完成，落「行完成已
|
||||
持久、批状态未写」的崩溃窗口后重启——零新 HTTP、cap 不变即 done，
|
||||
entry_ids 完整（修复前：queued/semantic_llm_budget_exhausted）。
|
||||
L4b 部分未完成仍走硬闸：还有缺行且额度耗尽时，重启后依旧拒绝，不为收尾放行。
|
||||
L5 真死 owner 的租约回收仍有效：无活跃 worker 的过期 pending 照常 failed/
|
||||
lease_expired 并重试（未因修复而停用恢复）。
|
||||
|
||||
证据 JSON 写 argv[1]。运行（仓库根，固定 venv）：
|
||||
|
||||
/path/to/hel857-venv/bin/python tools/verify_inflight_lifecycle_hel886.py out.json
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _suppress_cancel():
|
||||
with contextlib.suppress(asyncio.CancelledError, Exception):
|
||||
yield
|
||||
|
||||
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 Mem0Backend, SemanticMemory # noqa: E402 — after boundary
|
||||
from cybergril.memory.store import Store # noqa: E402
|
||||
|
||||
REPORT: dict = {"head": "HEL-886 inflight lifecycle", "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_turn(store: Store, turn_id: str, user_text: str, companion_text: str | None = None):
|
||||
with store.conn:
|
||||
store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES"
|
||||
"(1,'user',?,?,'sent')", (user_text, turn_id))
|
||||
ids = [store.query_one("SELECT id FROM chat_message WHERE text=?", (user_text,))["id"]]
|
||||
if companion_text is not None:
|
||||
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,))
|
||||
ids.append(store.query_one(
|
||||
"SELECT id FROM chat_message WHERE text=?", (companion_text,))["id"])
|
||||
return ids
|
||||
|
||||
|
||||
async def blocked_first_http(kind: str, tmp: Path):
|
||||
"""L1/L2/L3 share the shape: one real add accepted by the loopback gateway,
|
||||
answer withheld; the manager's exact repro."""
|
||||
first_arrived, release = threading.Event(), threading.Event()
|
||||
|
||||
class Gateway(contract.FakeGateway):
|
||||
def do_POST(self): # noqa: N802 — http.server contract
|
||||
if len(getattr(self.server, "accepted", [])) == 0:
|
||||
self.server.accepted.append(1)
|
||||
first_arrived.set()
|
||||
if not release.wait(15):
|
||||
raise RuntimeError("test barrier timed out")
|
||||
super().do_POST()
|
||||
|
||||
server = contract._Server(('127.0.0.1', 0), Gateway)
|
||||
server.accepted = []
|
||||
thread = threading.Thread(target=server.serve_forever)
|
||||
thread.start()
|
||||
try:
|
||||
ledger = tmp / f"{kind}.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 / f"{kind}-vec")
|
||||
cfg["history_root"] = str(tmp / f"{kind}-hist")
|
||||
backend = Mem0Backend(cfg)
|
||||
store = Store(tmp / f"{kind}.db")
|
||||
clock = [1000.0]
|
||||
live = SemanticMemory(store, backend, clock=lambda: clock[0])
|
||||
seed_turn(store, f"turn-{kind}", f"处理中原话{kind}")
|
||||
live.enqueue_uncovered()
|
||||
mark = len(contract.FakeGateway.requests)
|
||||
task = asyncio.create_task(live.tick())
|
||||
assert await asyncio.to_thread(first_arrived.wait, 5)
|
||||
# "arrived" = accepted by the loopback server, answer still withheld
|
||||
# (FakeGateway.requests only logs on completion, after the barrier).
|
||||
http_arrived = len(server.accepted)
|
||||
batch_live = dict(store.query_one("SELECT state,owner FROM sem_batch"))
|
||||
|
||||
if kind == "cancel":
|
||||
task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await task
|
||||
elif kind == "lease":
|
||||
clock[0] += 41
|
||||
live.recover()
|
||||
# concurrent scheduler ticks while the first HTTP is unanswered
|
||||
await asyncio.wait_for(live.tick(), 8)
|
||||
await asyncio.wait_for(live.tick(), 8)
|
||||
http_concurrent = len(server.accepted)
|
||||
batch_concurrent = dict(store.query_one("SELECT state FROM sem_batch"))
|
||||
|
||||
if kind == "drain":
|
||||
def _answer(): # the wire answer arrives while shutdown is waiting
|
||||
clock[0] = 2000
|
||||
release.set()
|
||||
timer = threading.Timer(0.2, _answer)
|
||||
timer.start()
|
||||
started = time.monotonic()
|
||||
await live.aclose(drain_timeout=10)
|
||||
waited = time.monotonic() - started
|
||||
row = store.query_one("SELECT * FROM sem_row_done")
|
||||
batch = dict(store.query_one("SELECT state,finished_at FROM sem_batch"))
|
||||
check("L3_shutdown_drains_live_worker_before_close",
|
||||
http_concurrent == 1 and waited >= 0.15 and row is not None
|
||||
and row["done_at"] >= 2000 and batch["state"] == "done"
|
||||
and batch["finished_at"] >= 2000,
|
||||
f"arrived={http_concurrent} waited={waited:.2f}s row_done_at="
|
||||
f"{row and row['done_at']} batch={batch}")
|
||||
with _suppress_cancel():
|
||||
await task
|
||||
timer.join()
|
||||
store.close()
|
||||
return {}
|
||||
|
||||
release.set()
|
||||
if not task.cancelled():
|
||||
await task
|
||||
for _ in range(300): # the WORKER thread writes the terminal state
|
||||
if store.query_one("SELECT state FROM sem_batch")["state"] != "pending":
|
||||
break
|
||||
await asyncio.sleep(0.02)
|
||||
# the drain path waits for the worker's durable writes, then closes
|
||||
points, _ = backend._client.scroll(collection_name=backend.collection,
|
||||
with_payload=True, limit=100)
|
||||
await live.aclose(drain_timeout=10)
|
||||
await asyncio.sleep(0.05)
|
||||
http_total = len(contract.FakeGateway.requests) - mark
|
||||
rows = store.query("SELECT * FROM sem_row_done")
|
||||
batch = dict(store.query_one("SELECT state,reason,entry_ids FROM sem_batch"))
|
||||
store.close()
|
||||
check(f"L{'1' if kind == 'cancel' else '2'}_{kind}_single_send_and_full_record",
|
||||
http_arrived == 1 and http_concurrent == 1 and http_total == 1
|
||||
and len(points) == 1
|
||||
and batch_live["state"] == "pending" and batch_live["owner"]
|
||||
and batch_concurrent["state"] == "pending"
|
||||
and batch["state"] == "done" and batch["reason"] is None
|
||||
and len(rows) == 1
|
||||
and json.loads(batch["entry_ids"] or "[]") == json.loads(rows[0]["entry_ids"]),
|
||||
f"arrived={http_arrived} concurrent={http_concurrent} http_total={http_total} "
|
||||
f"points={len(points)} batch_live={batch_live} "
|
||||
f"batch_concurrent={batch_concurrent['state']} final={batch} row_done={len(rows)}")
|
||||
return {}
|
||||
finally:
|
||||
release.set()
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join()
|
||||
|
||||
|
||||
async def scenario_all_done_restart(tmp: Path):
|
||||
"""L4a: the manager's third example — fully-completed batch finalizes after
|
||||
restart with zero new budget. L4b: a missing row keeps the hard gate."""
|
||||
ledger = tmp / "alldone.sqlite3"
|
||||
contract.make_ledger(ledger, cap=100, llm_cap=1)
|
||||
cfg = contract.config_for(tmp, ledger, port=_gateway_port, cap=100, llm_cap=1)
|
||||
cfg["vector_root"] = str(tmp / "alldone-vec")
|
||||
cfg["history_root"] = str(tmp / "alldone-hist")
|
||||
backend = Mem0Backend(cfg)
|
||||
db = tmp / "alldone.db"
|
||||
store = Store(db)
|
||||
seed_turn(store, "turn-done", "完整处理的合成原话")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
mark = len(contract.FakeGateway.requests)
|
||||
await live.tick()
|
||||
original = dict(store.query_one("SELECT state,entry_ids FROM sem_batch"))
|
||||
with store.conn: # crash window: row persisted, final status did not
|
||||
store.execute("UPDATE sem_batch SET state='pending',owner='crashed',lease_until=1")
|
||||
store.close()
|
||||
await backend.aclose()
|
||||
|
||||
backend2 = Mem0Backend(cfg) # same caps — nothing new approved
|
||||
store2 = Store(db)
|
||||
live2 = SemanticMemory(store2, backend2)
|
||||
await live2.tick()
|
||||
after = dict(store2.query_one("SELECT state,reason,entry_ids FROM sem_batch"))
|
||||
http = len(contract.FakeGateway.requests) - mark
|
||||
used = backend2.llm_guard.used()
|
||||
check("L4a_all_done_finalizes_without_new_budget",
|
||||
original["state"] == "done" and after["state"] == "done"
|
||||
and after["reason"] is None and http == 1 and used == 1
|
||||
and json.loads(after["entry_ids"] or "[]") == json.loads(original["entry_ids"]),
|
||||
f"before={original} after={after} http={http} llm_used={used}")
|
||||
store2.close()
|
||||
await backend2.aclose()
|
||||
|
||||
|
||||
async def scenario_partial_keeps_gate(tmp: Path):
|
||||
"""L4b: one row still missing + budget exhausted → restart keeps refusing."""
|
||||
ledger = tmp / "partial.sqlite3"
|
||||
contract.make_ledger(ledger, cap=100, llm_cap=1)
|
||||
cfg = contract.config_for(tmp, ledger, port=_gateway_port, cap=100, llm_cap=1)
|
||||
cfg["vector_root"] = str(tmp / "partial-vec")
|
||||
cfg["history_root"] = str(tmp / "partial-hist")
|
||||
backend = Mem0Backend(cfg)
|
||||
db = tmp / "partial.db"
|
||||
store = Store(db)
|
||||
seed_turn(store, "turn-partial", "部分完成用户原话", "部分完成伴侣原话")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
await live.tick() # one row sent, the other budget-refused
|
||||
with store.conn:
|
||||
store.execute("UPDATE sem_batch SET state='pending',owner='crashed',lease_until=1")
|
||||
store.close()
|
||||
await backend.aclose()
|
||||
|
||||
backend2 = Mem0Backend(cfg)
|
||||
store2 = Store(db)
|
||||
live2 = SemanticMemory(store2, backend2)
|
||||
mark = len(contract.FakeGateway.requests)
|
||||
await live2.tick()
|
||||
after = dict(store2.query_one("SELECT state,reason FROM sem_batch"))
|
||||
http = len(contract.FakeGateway.requests) - mark
|
||||
check("L4b_partial_keeps_the_hard_gate",
|
||||
after["state"] == "queued" and "budget" in (after["reason"] or "") and http == 0,
|
||||
f"after={after} http={http}")
|
||||
store2.close()
|
||||
await backend2.aclose()
|
||||
|
||||
|
||||
async def scenario_dead_lease(tmp: Path):
|
||||
"""L5: recovery of a genuinely dead owner is NOT disabled by the fix."""
|
||||
ledger = tmp / "deadlease.sqlite3"
|
||||
contract.make_ledger(ledger, cap=100, llm_cap=10)
|
||||
cfg = contract.config_for(tmp, ledger, port=_gateway_port, cap=100, llm_cap=10)
|
||||
cfg["vector_root"] = str(tmp / "dead-vec")
|
||||
cfg["history_root"] = str(tmp / "dead-hist")
|
||||
backend = Mem0Backend(cfg)
|
||||
store = Store(tmp / "dead.db")
|
||||
seed_turn(store, "turn-dead", "真正死亡第一行", "真正死亡第二行")
|
||||
live = SemanticMemory(store, backend)
|
||||
live.enqueue_uncovered()
|
||||
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()
|
||||
await live.aclose(drain_timeout=5)
|
||||
after = dict(store.query_one("SELECT state FROM sem_batch"))
|
||||
http = len(contract.FakeGateway.requests) - mark
|
||||
check("L5_dead_owner_lease_still_recovers",
|
||||
job["state"] == "failed" and job["reason"] == "lease_expired"
|
||||
and after["state"] == "done" and http == 2,
|
||||
f"recovered={job} after={after} http={http}")
|
||||
store.close()
|
||||
|
||||
|
||||
_gateway_port: int = 0
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
global _gateway_port
|
||||
server = contract.start_gateway()
|
||||
_gateway_port = server.server_address[1]
|
||||
with tempfile.TemporaryDirectory(dir=ROOT / "tools") as raw_tmp:
|
||||
tmp = Path(raw_tmp)
|
||||
try:
|
||||
await blocked_first_http("cancel", tmp)
|
||||
await blocked_first_http("lease", tmp)
|
||||
await blocked_first_http("drain", tmp)
|
||||
await scenario_all_done_restart(tmp)
|
||||
await scenario_partial_keeps_gate(tmp)
|
||||
await scenario_dead_lease(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 "hel886-inflight-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())
|
||||
Reference in New Issue
Block a user