diff --git a/CHANGELOG.md b/CHANGELOG.md index 21d4b5e0..a1222033 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ ### 新增 Added + - **Round memory peak into the ledger (issue #139 item ①; ops-backlog batch)**: `backend/ipdb/_ops_rounds.py` gains `read_memory_peak()` — cgroup v2 `/sys/fs/cgroup/memory.peak`, v1 `memory.max_usage_in_bytes` fallback, both unreadable → `None` (a missing reading is never fabricated as zero); `run_round` records a baseline at round start and the `finally` closeout (success / failure / cancel / OSError crash-settle — all four paths, courtesy of the #196 try-finally) writes `memory_peak_bytes=N` as its own ledger row through the frozen five-column header (LEDGER_HEADER is T4-frozen, no new column; legacy rows simply lack the field); peak==baseline gets the honest annotation "(=轮始基线 N,容器累计高水位,轮内未破)" — cgroup memory.peak is a lifetime cumulative high-water mark, the annotation marks the upper-bound semantics; ledger write is best-effort (never tears the round chain); spawn/env faces untouched (C4 whitelist not expanded). Closes the cold-start acceptance #5 gap where runbook §7 substituted manual `docker stats`. + - **轮次内存峰值入台账(issue #139 挂账①;ops-backlog 批)**:`backend/ipdb/_ops_rounds.py` 新增 `read_memory_peak()`——cgroup v2 `/sys/fs/cgroup/memory.peak` 优先、v1 `memory.max_usage_in_bytes` 回退、皆不可读返回 `None`(缺读数不虚构为零);`run_round` 轮始记基线,`finally` 收口面(成功/失败/取消/OSError 兑底四路径全收,借 #196 的 try-finally 骨架)写 `memory_peak_bytes=N` 独立台账行——LEDGER_HEADER 是 T4 冻结契约不加列,老行缺该字段=缺省不虚构;终值==基线时附注「(=轮始基线 N,容器累计高水位,轮内未破)」——cgroup memory.peak 是生命周期累计高水位,附注即上界语义的诚实标注;台账写 best-effort 不撕轮次链;spawn/env 面零改动(C4 白名单未扩)。收口冷启动验收 #5 的缺口(runbook §7 原以人工 `docker stats` 替代)。 + - **queue.md cross-process race: bilateral flock (issue #139 item ②; the frozen "agent-face" premise is obsolete)**: python `append_queue`'s read-dedup-write critical section now holds `fcntl.flock(LOCK_EX)`; node writes go through new `ops/agent/tools/queue-lock.mjs` which spawns util-linux `flock(1)` — same lock domain as python's flock(2), chosen over O_EXCL hand-rolling (different lock domain wouldn't interoperate; self-rolled stale-age takeover risks strong-takeover false kills), zero new npm dependencies; the lock-file path constant lives exactly once per side (`QUEUE_LOCK_NAME` / `QUEUE_LOCK_SUFFIX`) with a 17th contract sentinel cross-checking the two resolve to the same path; the lock file is never deleted (deletion would itself be the race); CLI face is envelope-shaped (single-line JSON even on crash). Race evidence: unpatched two-process × 40-append probe loses ~half the rows (40~41 of 80); patched → lost=0; mixed python×node interop (8+8 concurrent, production shape) → exactly 16 rows, 3/3 runs; node t4 adds four cases (cross-process exactly-N / same-key dedup-once / SIGKILL releases via kernel / missing flock binary = fail-closed zero writes). + - **queue.md 跨进程竞态双边锁(issue #139 挂账②;原「agent 冻结面」前提已过时)**:python `append_queue` 读-判重-写临界段持 `fcntl.flock(LOCK_EX)`;node 写口走新 `ops/agent/tools/queue-lock.mjs`,spawn util-linux `flock(1)`——与 python flock(2) 同锁域;弃 O_EXCL 自造锁(异锁域不互锁、自造 stale 判龄有强接误杀竞态),零新 npm 依赖;锁文件路径常量两侧各恰一处(`QUEUE_LOCK_NAME`/`QUEUE_LOCK_SUFFIX`),contract 第 17 哨跨语言对拍同址;锁文件永不删(删除本身即竞态);CLI 面信封化(崩溃也是单行 JSON)。竞态实证:修前双进程×40 追加探针丢约半数行(80 实得 40~41),修后 lost=0;混语互操(python×node 各 8 并发,生产同形)恰 16 行、三连跑 3/3;node t4 增四例(跨子进程恰 N 行/同键判重恰一次/SIGKILL 后内核即释/缺 flock 二进制 fail-closed 零写)。 + - **Assessment-track rc machine-checking + month-round leg separation (issue #193, umbrella #191 arch-review C5)**: `scripts/fusion_replay_gate.py` grows `gate_rc` aggregation — 0 = all legs PASS with sufficient corpus, 1 = any true FAIL (C2 strict/loose mismatch stays a hard gate), 2 = insufficient corpus (empty-corpus vacuous PASS no longer masquerades as green); `main()` returns it instead of unconditionally `0` (the dead gate is alive again; reviewer-D P1-1). `scripts/monthly_eval.sh` separates per-leg rc: all eight legs (LEG_ORDER: model/all/dsem/audit/temporal/readout/anchors/agg-share) capture rc explicitly and never abort siblings — the F4 anchors special case is honestly generalized to a whole-round contract; final exit = first non-zero in leg order; `check_family -maxdepth 1` closes the old find-recursion hole where a same-shape artifact in a subdirectory could substitute for the family verdict; leg-status artifacts (`leg-status-*.json`) dual-write (in-flight snapshot with readout=null + final) and feed the monthly report's leg-status section (missing = section omitted, never fabricated). STANDARD.md D-1 gate hook 47→48. - **评估轨 rc 机检化+月轮全腿分离(issue #193,伞 #191 审审整改 C5)**:`scripts/fusion_replay_gate.py` 新增 `gate_rc` 聚合——0=全腿 PASS 且语料充分,1=任一真 FAIL(C2 strict/loose 失配仍硬门),2=语料不足(空语料的空真 PASS 不再冒充绿);`main()` 返回真实 rc(死门复活,评审D P1-1)。`scripts/monthly_eval.sh` 全腿 rc 分离:八腿(LEG_ORDER)显式捕获不互断——F4 anchors 特例诚实推广为全轮契约;最终退出码=腿序首非零;`check_family -maxdepth 1` 补 find 递归旧洞(子目录同形状件不再能顶替族 verdict 件);腿级状态件(leg-status-*.json)双写(在飞快照+定稿件)入月报腿况节(缺省不出节,不虚构)。STANDARD.md D-1 挂点 47→48。 - **Gate-contract single source + dual cross-language sentinels (issue #194, umbrella #191 arch-review C3)**: `GATE_SEQUENCE`/`GATE_COUNT`/`GATE_NAMES` derive from the gates.mjs registry itself (7 hardcoded `records.length !== 10` sites zeroed; adding/removing/reordering a gate trips the registry-order↔execution-order bijection invariant test); the 19-anchor JS mirror is unified into `ops/agent/anchor-mirror.mjs` (single definition, `anchorRow` constructor throws on unknown/missing keys — the #189 stub-missing-conflict precedent); two new §14 sentinels: SHAPE (anchor row key-set cross-language check) and FOLD (fold-family full-value-domain check: python `_error_family` golden constants AST-extracted vs JS `foldErrorFamily` — previously only net_timeout→timeout was covered). Acceptance replayed by the merger: tampering with gate list / one fold value / one anchor / one row key each trips the sentinels red while the JS-internal tests still look green (the canned-self-consistency blind spot the sentinels exist to catch). diff --git a/backend/ipdb/_ops_rounds.py b/backend/ipdb/_ops_rounds.py index 5e951244..a2879acb 100644 --- a/backend/ipdb/_ops_rounds.py +++ b/backend/ipdb/_ops_rounds.py @@ -21,7 +21,8 @@ 契约违背 → infra 兜底信封; - 一切失败面与 admin 取消落台账(_ops_store.append_ledger;轮次失败≠工件 灭失,规格 §3;九裁决 9:取消的轮次也可能已花真钱,留痕);成功面桥零追加 - (agent 侧 lib.sh 自记真实 usage,双记=回声); + (agent 侧 lib.sh 自记真实 usage,双记=回声);例外=内存峰值独立行 + (#139 挂账①:agent 侧无采集,桥侧 cgroup 读数是唯一记录,非回声); - stderr 进度行([turn]/[tool],契约面)逐行回调 on_log(run_ops_round 接任务链 SSE 的 task_log 事件)。 @@ -499,6 +500,55 @@ def _settle_crash(manager, task_id: str, exc: BaseException, logger.exception("C1 兑底 finish_ops_round 失败(task %s)——车道释放链断,需人工介入", task_id) +# ── #139 挂账①:轮次内存峰值采集(cgroup 读数,桥侧;agent 冻结面零触碰)── +# 冷启动验收 #5 需要轮次内存峰值,此前以 docker stats 人工替代(runbook §7); +# 挂账原判「agent 冻结面不可动」已过时(hotfix #196/C2 均动过本文件),改在 +# 桥侧读容器自身 cgroup。cgroup v2 容器内 /sys/fs/cgroup/memory.peak 即容器 +# 自创建累计高水位(单调不回落);cgroup v1 无此文件,回退 +# memory/memory.max_usage_in_bytes(语义同为累计高水位)。两者皆不可读 +# (开发机裸机无 cgroup 或无权限)→ None 不抛:缺读数≠零用量,绝不虚构 +# (测量建言/宁少算)。路径钉模块常量:测试/P5 特殊拓扑 monkeypatch 此面。 +_CGROUP_V2_PEAK = Path("/sys/fs/cgroup/memory.peak") +_CGROUP_V1_PEAK = Path("/sys/fs/cgroup/memory/memory.max_usage_in_bytes") + + +def read_memory_peak() -> int | None: + """cgroup 内存峰值高水位读数(v2 → v1 回退;皆不可读 → None,不抛)。 + + #139 挂账①(冷启动验收 #5;原 docker stats 人工替代口径见 runbook §7)。 + """ + for path in (_CGROUP_V2_PEAK, _CGROUP_V1_PEAK): + try: + return int(path.read_text(encoding="ascii").strip()) + except (OSError, ValueError): # 缺文件/无权限/非整数(含解码错):回退下一面 + continue + return None + + +def _record_memory_peak(round_type: str, base: int | None) -> None: + """轮次收口面:内存峰值独立行入台账(#139 挂账①;best-effort 不抛)。 + + 行形状决定:独立行(append_ledger 既有 5 列)而非加列——台账表头 T4 冻结 + (node 侧 lib.sh ledger_append 首建同串),加列=改冻结契约;老行/agent 侧行 + 无此列,前向兼容=缺行即缺省,不虚构。「成功面桥零追加」口径限 usage 双记 + =回声;内存峰值 agent 侧无采集(#139 立项缺口),本行是唯一记录,非回声。 + cgroup 峰值为累计高水位:终值>基线 → 轮内破高,终值即轮内峰值;相等 → + 轮内未破,终值是容器历史峰值上界,行内标注(不把上界虚构或轮内峰值)。 + 写盘失败只记日志:收口链(finish/兑底)已走完,不得反向撕线程(C1 同判例)。 + """ + peak = read_memory_peak() + if peak is None: + return # 缺读数不虚构:两代 cgroup 皆不可读(裸机开发机)→ 零行 + note = f"memory_peak_bytes={peak}" + if base is not None and peak == base: + note += f"(=轮始基线 {base},容器累计高水位,轮内未破)" + try: + _ops_store.append_ledger(round_type, "—", "—", note) + except Exception: + logger.warning("#139 内存峰值台账行写入失败(轮 %s,峰值 %d)", + round_type, peak, exc_info=True) + + def run_round(manager, task, round_type: str, input_text: str) -> None: """ops 轮线程入口(路由侧 threading.Thread 拉起):spawn → 收口。 @@ -506,10 +556,13 @@ def run_round(manager, task, round_type: str, input_text: str) -> None: manager.finish_ops_round(done/failed/cancelled;error_code = 信封 error.code 九词闭集透传);stderr 行经 manager.ops_log 进任务链 SSE。 C1(#196):线程入口整体兑底——链上任何未捕获异常(如磁盘满 OSError) - 也落终态+放车道,不得卡 running 撕停摆源 worker。""" + 也落终态+放车道,不得卡 running 撕停摆源 worker。 + #139 挂账①:轮始记 cgroup 峰值基线,收口面 finally(一切含兑底路径) + 写终值入台账(_record_memory_peak;冷启动验收 #5 的机读替代)。""" def on_log(line: str) -> None: manager.ops_log(task.id, line) + peak_base = read_memory_peak() # 轮始基线(容器累计高水位现值) try: result = spawn_round(round_type, input_text, on_log=on_log, cancel_token=task.token) @@ -528,3 +581,6 @@ def on_log(line: str) -> None: logger.exception("ops 轮线程未捕获异常(task %s, round_type=%s)", task.id, round_type) _settle_crash(manager, task.id, e, round_type) + finally: + # #139 挂账①:一切收口路径(成功/失败/取消/兑底)同写终值 + _record_memory_peak(round_type, peak_base) diff --git a/backend/ipdb/_ops_store.py b/backend/ipdb/_ops_store.py index 10bfd198..3ecfbda9 100644 --- a/backend/ipdb/_ops_store.py +++ b/backend/ipdb/_ops_store.py @@ -34,14 +34,16 @@ append_queue 入队(action+source 去重幂等);其余 verdict/动词零队列写 (不发明闭集外动词);幂等路径也调 append_queue —— 去重下真零写,且自愈 「判例已落、队列中途失或多」的崩殘态; -- 竞态窗口:模块锁关进程内窗(先查后写 + 原子写);多进程部署下窗口残留 - (uvicorn 单 worker 单容器形态下不可达,报告已说明); +- 竞态窗口:queue.md 写口持跨进程 flock(_queue_lock,与 node 侧 + tools/queue-lock.mjs 的 flock(1) 同锁域互认,#139 挂账②双边闭环);旧判 + 「多进程部署下窗口残留」已由双边锁关闭; - P1 §3 审阅四键(票 #155):write_ruling 增 item_id 来源(与 package 互斥), 四律同构复用:lineage/hard_gates 指针 = ops/items/.md(证据在条目里 不复制),幂等键首槽 = item:(与包名空间隔离),队列行轮次列 = "review"(审阅动作非轮次产物);包来源路径字节零变化。 """ +import fcntl import hashlib import json import os @@ -50,6 +52,7 @@ import sys import tempfile import threading +from contextlib import contextmanager from datetime import date from pathlib import Path from typing import Any, Sequence @@ -78,6 +81,46 @@ "|---|---|---|---|---|---|---|\n" ) +# #139 挂账②:queue.md 跨进程互斥锁文件名(= queue.md 同名 + .lock)。对称契约: +# node 侧 ops/agent/tools/queue-lock.mjs QUEUE_LOCK_SUFFIX=".lock" 从其 queueFile +# 落位(DOCS_ROOT/ops/queue.md)同式推导;生产形态 P2 桥 _ops_rounds._round_env +# 注入 DOCS_ROOT=ops_root().parent 使两侧落位同址同文件。路径漂移防线 = +# selftest-contract.mjs「锁路径对拍」哨(抽本文件 QUEUE_LOCK_NAME 字面量与 +# node queueLockPath 实跑值对拍,形制同 selftest.sh §14)。 +QUEUE_LOCK_NAME = "queue.md.lock" + + +def queue_lock_path() -> Path: + """queue.md 跨进程互斥锁文件路径(与 queue.md 同目录)。""" + return ops_root() / QUEUE_LOCK_NAME + + +@contextmanager +def _queue_lock(): + """queue.md 跨进程排它锁(fcntl.flock(2),LOCK_EX)。 + + 为什么锁:append_queue 是读-判重-写三段,node agent 的 queue_action(轮次中) + 是同构读-改-写,同容器两进程(uvicorn × node 轮次)并发时互窗存在——旧模块锁 + 只关进程内窗(#139 挂账②原判「单边修是假修」,现双边闭环:node 侧 spawn + util-linux flock(1),与本侧 fcntl.flock 同一 flock(2) 锁域,互认互斥)。 + 为什么这窗口:锁必须覆盖 读 existing → 去重判 → 原子写 全段,缺任何一段都 + 留窗(读后写前的判重结果会被他方写失效)。 + 为什么 flock 不用 O_EXCL 判龄:锁挂在 open file description 上由内核管理, + 进程死锁即自动释放,无 stale 判龄协议的「>30s 强接误杀活锁」竞态;锁文件 + 不存在则 O_CREAT 创建,close 即释放,不删除(空文件常驻无害)。 + fcntl 面向 POSIX:容器 Linux 限定(#139 语境=单容器生产形态),可接受。 + 锁序:write_ruling 先持 threading._ruling_lock 再进本锁(恒 threading→flock, + 本锁内不再取其他锁,无倒置死锁面)。 + """ + lock = queue_lock_path() + lock.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(lock, os.O_CREAT | os.O_RDWR, 0o644) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + yield + finally: + os.close(fd) # close 即释放 flock,不删锁文件 + _RG_TIMEOUT_S = 60 # ops/agent tools.mjs search_precedents 同值 @@ -206,19 +249,22 @@ def append_queue( raise ValueError(f"source 形状非法(^[a-z0-9_]{{1,32}}$): {source!r}") qpath = ops_root() / "queue.md" - existing = _read_if_exists(qpath) - # T4 去重判式逐字:任意日期/轮次的同 action+source 行已存在即去重 - dedup_re = re.compile( - r"\| \d{4}-\d{2}-\d{2} \| [^|]+ \| " + re.escape(action) - + r" \| " + re.escape(source) + r" \|") - dedup = dedup_re.search(existing) is not None - if not dedup: - ev = ";".join(f"{c}:{f}={json.dumps(v, ensure_ascii=False)}" - for c, f, v in evidence) - d = (today or date.today()).isoformat() - row = (f"| {d} | {round_} | {action} | {source} | {predicate if predicate else '—'} " - f"| {ev} | 是(否决=删除本行) |\n") - _append_row_with_header(qpath, QUEUE_HEADER, row) + # #139 挂账②:读-判重-写全段持跨进程锁(窗口论证见 _queue_lock;幂等去重 + # 逻辑不动,只包锁)——无锁窗口 = 他方写使本方判重结果失效,丢行/重行。 + with _queue_lock(): + existing = _read_if_exists(qpath) + # T4 去重判式逐字:任意日期/轮次的同 action+source 行已存在即去重 + dedup_re = re.compile( + r"\| \d{4}-\d{2}-\d{2} \| [^|]+ \| " + re.escape(action) + + r" \| " + re.escape(source) + r" \|") + dedup = dedup_re.search(existing) is not None + if not dedup: + ev = ";".join(f"{c}:{f}={json.dumps(v, ensure_ascii=False)}" + for c, f, v in evidence) + d = (today or date.today()).isoformat() + row = (f"| {d} | {round_} | {action} | {source} | {predicate if predicate else '—'} " + f"| {ev} | 是(否决=删除本行) |\n") + _append_row_with_header(qpath, QUEUE_HEADER, row) return {"appended": not dedup, "dedup": dedup} diff --git a/backend/tests/test_ops_rounds.py b/backend/tests/test_ops_rounds.py index 722d3e2a..1f00a030 100644 --- a/backend/tests/test_ops_rounds.py +++ b/backend/tests/test_ops_rounds.py @@ -73,6 +73,10 @@ def ops_env(tmp_path, monkeypatch): if out: with open(out, "w", encoding="utf-8") as f: json.dump({"argv": sys.argv[1:], "env": dict(os.environ)}, f) +bump = cfg.get("mem_bump") # #139 挂账①:轮中抬写假 cgroup memory.peak(钉终值读时序) +if bump: + with open(bump[0], "w", encoding="utf-8") as f: + f.write(str(bump[1])) print("[turn 1] fake progress", file=sys.stderr) if mode == "sleep": time.sleep(30) @@ -333,6 +337,147 @@ def disk_full(*_a, **_k): assert mgr.logs, "兑底路径留痕 task_log(不吞异常信息)" +# ── #139 挂账①:轮次内存峰值(cgroup 读数三态 + run_round 收口面台账行)── + + +def test_read_memory_peak_cgroup_v2(rounds, tmp_path, monkeypatch): + """cgroup v2 可读:memory.peak(P5 容器内即容器峰值高水位)。""" + v2 = tmp_path / "cgroup2" / "memory.peak" + v2.parent.mkdir() + v2.write_text("838860800\n", encoding="utf-8") + monkeypatch.setattr(rounds, "_CGROUP_V2_PEAK", v2) + monkeypatch.setattr(rounds, "_CGROUP_V1_PEAK", tmp_path / "cgroup1" / "nope") + assert rounds.read_memory_peak() == 838860800 + + +def test_read_memory_peak_cgroup_v1_fallback(rounds, tmp_path, monkeypatch): + """v2 不可读 → 回退 cgroup v1 memory.max_usage_in_bytes(旧内核容器)。""" + v1 = tmp_path / "cgroup1" / "memory" / "memory.max_usage_in_bytes" + v1.parent.mkdir(parents=True) + v1.write_text("1048576\n", encoding="utf-8") + monkeypatch.setattr(rounds, "_CGROUP_V2_PEAK", tmp_path / "cgroup2" / "nope") + monkeypatch.setattr(rounds, "_CGROUP_V1_PEAK", v1) + assert rounds.read_memory_peak() == 1048576 + + +def test_read_memory_peak_absent_returns_none(rounds, tmp_path, monkeypatch): + """两代路径皆不可读(开发机裸机无 cgroup/无权限)→ None 不抛: + 缺读数≠零用量,绝不虚构(测量建言/宁少算)。""" + monkeypatch.setattr(rounds, "_CGROUP_V2_PEAK", tmp_path / "cgroup2" / "nope") + monkeypatch.setattr(rounds, "_CGROUP_V1_PEAK", tmp_path / "cgroup1" / "nope") + assert rounds.read_memory_peak() is None + + +def _fake_cgroup(rounds, tmp_path, monkeypatch, v2_bytes=None, v1_bytes=None): + """假 cgroup 树落 tmp_path + monkeypatch 模块路径常量;v2_bytes/v1_bytes + 均缺省 = 两代皆不可读(None 态)。""" + v2 = tmp_path / "cgroup2" / "memory.peak" + v1 = tmp_path / "cgroup1" / "memory" / "memory.max_usage_in_bytes" + if v2_bytes is not None: + v2.parent.mkdir(parents=True, exist_ok=True) + v2.write_text(f"{v2_bytes}\n", encoding="utf-8") + if v1_bytes is not None: + v1.parent.mkdir(parents=True, exist_ok=True) + v1.write_text(f"{v1_bytes}\n", encoding="utf-8") + monkeypatch.setattr(rounds, "_CGROUP_V2_PEAK", v2) + monkeypatch.setattr(rounds, "_CGROUP_V1_PEAK", v1) + return v2 + + +def _run_round_to_finish(rounds, mgr, round_type="health", input_text="体检输入"): + """拉起 run_round 线程并等待收口退出(#139 测试面;复用 C1 崩溃测试同型)。""" + task, busy = mgr.start_ops_round(round_type) + assert task is not None and busy is None + t = threading.Thread(target=rounds.run_round, + args=(mgr, task, round_type, input_text), daemon=True) + t.start() + t.join(timeout=15) + assert not t.is_alive(), "run_round 线程必须收口退出,不得死挂" + return task + + +def test_run_round_writes_memory_peak_row_success( + rounds, ops_env, fake_agent, monkeypatch): + """#139 挂账①:成功轮收口面写内存峰值独立行(memory_peak_bytes=终值; + 轮内未破高水位时标注上界语义,不把容器历史峰值虚构或轮内峰值)。""" + _fake_cgroup(rounds, ops_env, monkeypatch, v2_bytes=838860800) + mgr = _make_manager([]) + task = _run_round_to_finish(rounds, mgr) + snap = {row["id"]: row for row in mgr.snapshot()["tasks"]} + assert snap[task.id]["state"] == "done" + led = _ledger_text(ops_env) + assert "memory_peak_bytes=838860800" in led, led + # 独立行:既有 5 列形状零改动(轮次=health,model/用量=—);加列=改 T4 冻结表头 + assert any(" | health | — | — | memory_peak_bytes=" in ln + for ln in led.splitlines()), led + assert "(=轮始基线" in led # 静态假文件:终值=基线,标注上界语义 + + +def test_run_round_memory_peak_growth_attributed( + rounds, ops_env, fake_agent, monkeypatch): + """终值>基线(假 agent 轮中抬写假 memory.peak)→ 轮内破高:行记轮内真实 + 峰值且无上界标注;时序钉死——终值读在 spawn 收口之后(子进程写完才读)。""" + v2 = _fake_cgroup(rounds, ops_env, monkeypatch, v2_bytes=100) + cfg_path = fake_agent[0].parent / "fake-agent-cfg.json" + cfg = json.loads(cfg_path.read_text(encoding="utf-8")) + cfg["mem_bump"] = [str(v2), 200] + cfg_path.write_text(json.dumps(cfg), encoding="utf-8") + mgr = _make_manager([]) + task = _run_round_to_finish(rounds, mgr) + snap = {row["id"]: row for row in mgr.snapshot()["tasks"]} + assert snap[task.id]["state"] == "done" + led = _ledger_text(ops_env) + assert "memory_peak_bytes=200" in led, led + assert "(=轮始基线" not in led # 破高:无上界标注 + + +def test_run_round_writes_memory_peak_row_on_failure( + rounds, ops_env, fake_agent, monkeypatch): + """失败信封轮(fail 轮):spawn error 行与内存峰值行同落台账——收口面 + finally 不挑结果(轮次失败≠工件灭失,规格 §3;峰值面同理)。""" + _fake_cgroup(rounds, ops_env, monkeypatch, v2_bytes=838860800) + _fake_agent_set(fake_agent[0], "failure") + mgr = _make_manager([]) + task = _run_round_to_finish(rounds, mgr) + snap = {row["id"]: row for row in mgr.snapshot()["tasks"]} + assert snap[task.id]["state"] == "failed" + assert snap[task.id]["error_code"] == "closure_check" + led = _ledger_text(ops_env) + assert "closure_check" in led and "memory_peak_bytes=838860800" in led, led + + +def test_run_round_writes_memory_peak_row_on_crash_settle( + rounds, ops_env, fake_agent, monkeypatch): + """兜底路径(C1 #196 异常兑底):_settle_crash 台账行 + finally 内存峰值行 + 都在;线程照旧收口退出(内存行写失败也不得反向撕线程,同 C1 判例)。""" + _fake_cgroup(rounds, ops_env, monkeypatch, v2_bytes=838860800) + + def boom(*_a, **_k): + raise OSError(5, "Input/output error") + + monkeypatch.setattr(rounds, "spawn_round", boom) + mgr = _make_manager([]) + task = _run_round_to_finish(rounds, mgr) + snap = {row["id"]: row for row in mgr.snapshot()["tasks"]} + assert snap[task.id]["state"] == "failed" + assert snap[task.id]["error_code"] == "infra" + led = _ledger_text(ops_env) + assert "spawn error_code=infra" in led, led + assert "memory_peak_bytes=838860800" in led, led + + +def test_run_round_no_memory_row_when_cgroup_unreadable( + rounds, ops_env, fake_agent, monkeypatch): + """两代 cgroup 皆缺(开发机裸机)→ 成功轮零内存行:缺读数不虚构(None + 不落行;台账不添「peak=0」式伪证)。""" + _fake_cgroup(rounds, ops_env, monkeypatch) # 两代皆缺 + mgr = _make_manager([]) + task = _run_round_to_finish(rounds, mgr) + snap = {row["id"]: row for row in mgr.snapshot()["tasks"]} + assert snap[task.id]["state"] == "done" + assert "memory_peak_bytes" not in _ledger_text(ops_env) + + # ── A. argv 契约(纯函数;真 node 命令形状)────────────────────────────────── diff --git a/backend/tests/test_ops_store.py b/backend/tests/test_ops_store.py index 1574d42a..ad5c3c43 100644 --- a/backend/tests/test_ops_store.py +++ b/backend/tests/test_ops_store.py @@ -11,7 +11,9 @@ 不从被测模块导入常量,防套套逻辑。 """ +import json import os +import shutil import subprocess import sys from datetime import date @@ -206,6 +208,100 @@ def test_append_queue_validates_frozen_closed_sets(store, ops_env, kwargs): assert not (ops_env / "ops" / "queue.md").exists() +# ── #139 挂账②:queue.md 跨进程竞态双边锁(queue.md.lock) ──────────────── + +# 子进程驱动:每进程串行 append N 个不同 source(复现 POST /api/ops/ruling 与 +# node agent 轮次两进程并发写口的互窗;真子进程而非线程,锁面=进程间 flock) +_QUEUE_RACE_DRIVER = ( + "import sys; from ipdb._ops_store import append_queue\n" + "n = int(sys.argv[1]); tag = sys.argv[2]\n" + "for i in range(n):\n" + " append_queue(round_='discover', action='REJECT', source=f'{tag}{i}',\n" + " predicate='sunset_frozen', evidence=[('c', 'age_days', 300)])\n" +) + + +def _spawn_appenders(ops_env, per_proc, tags): + env = {**os.environ, "IP_RADAR_DATA_DIR": str(ops_env)} + procs = [subprocess.Popen([sys.executable, "-c", _QUEUE_RACE_DRIVER, + str(per_proc), tag], cwd=BACKEND_DIR, env=env) + for tag in tags] + for p in procs: + assert p.wait() == 0, "子进程 append 崩殈(锁/写路径异常)" + + +def _queue_rows(ops_env): + q = (ops_env / "ops" / "queue.md").read_text(encoding="utf-8") + return [l for l in q.splitlines() if l.startswith("| 2026-")] + + +def test_append_queue_concurrent_processes_no_lost_rows(store, ops_env): + """#139 挂账②:两子进程并发 append 不同 source → 行数恰 2N(不丢不重)。 + + 修前实证(2×40 并发,无锁):期望 80 实际 40~41,丢行近半——读-判重-写 + 全段互窗,后写覆盖前写;持锁后全段串行化,行数恰等。锁文件随首次写创建。 + """ + _spawn_appenders(ops_env, 8, ("aaa_", "bbb_")) + rows = _queue_rows(ops_env) + assert len(rows) == 16, f"丢行/重行:期望 16,实际 {len(rows)}" + assert (ops_env / "ops" / "queue.md").read_text(encoding="utf-8").count(QUEUE_HEADER) == 1 + assert (ops_env / "ops" / "queue.md.lock").exists(), "锁文件应随首次 append 创建" + + +def test_append_queue_concurrent_same_key_exactly_one_row(store, ops_env): + """#139 挂账②:两子进程并发 append 同 action+source 键 → 每键恰 1 行。 + + 去重判(读 existing)也须在锁内:无锁时两进程同时读到「不在队列」→ 双写 + 同键两行;判重+写入全段持锁后互斥,键唯一。 + """ + _spawn_appenders(ops_env, 6, ("same_", "same_")) + rows = _queue_rows(ops_env) + assert len(rows) == 6, f"同键双写:期望 6,实际 {len(rows)}" + assert sum(1 for l in rows if "| same_3 |" in l) == 1 + + +# 混语互操(#139 挂账②双边闭环的终验):node 侧临界段经 util-linux flock(1) +# 子进程执行,与本侧 fcntl.flock 同一 flock(2) 锁域——两语言写口真互斥才证明 +# 双边锁闭环(单边锁只关自家窗)。供给面:node 可执行 + queue-lock.mjs 自含 +# (零仓依赖);backend CI 无 node 供给时跳过(ubuntu-latest 预装 node,本地必跑)。 +_NODE_QUEUE_LOCK = BACKEND_DIR.parent / "ops" / "agent" / "tools" / "queue-lock.mjs" + + +@pytest.mark.skipif(shutil.which("node") is None or not _NODE_QUEUE_LOCK.exists() + or shutil.which("flock") is None, + reason="node/flock 或 queue-lock.mjs 不在供给面(backend CI 无 node 供给时跳过)") +def test_append_queue_mixed_python_node_race(store, ops_env): + """#139 挂账②:python append_queue × node flock(1) 临界段并发 → 不丢不重。""" + # 前提同生产:init_ops_tree 先建 ops/ 目录。flock(1) 对锁文件 O_CREAT,但 + # 父目录缺失时 cannot open → fail-closed 不写 queue.md;不预建则 node 子进程 + # 与 python 首笔 append(惰性建目录)赛跑,前者可能先撞缺目录。 + (ops_env / "ops").mkdir(parents=True, exist_ok=True) + per_side, qfile = 8, ops_env / "ops" / "queue.md" + env = {**os.environ, "IP_RADAR_DATA_DIR": str(ops_env)} + py = subprocess.Popen([sys.executable, "-c", _QUEUE_RACE_DRIVER, str(per_side), "py_"], + cwd=BACKEND_DIR, env=env) + # node 侧:每 append 一个 queue-lock.mjs --append 子进程(与生产路径同形: + # registry.mjs 每次入队 spawn 一个 flock 包装的临界段子进程) + def _node_payload(src): + return {"queueFile": str(qfile), "round": "discover", "action": "REJECT", + "source": src, "predicate": "sunset_frozen", + "evidence": [{"call_id": "c", "field": "age_days", "value": 300}]} + nds = [subprocess.Popen( + # 生产路径同形:flock(1) 包装临界段子进程(与 python fcntl.flock 同 + # flock(2) 锁域)——直调 --append 无锁,不测互操面;选项须在锁文件前 + ["flock", "-w", "30", str(qfile) + ".lock", "node", str(_NODE_QUEUE_LOCK), + "--append", json.dumps(_node_payload(f"nd_{i}"), ensure_ascii=False)], + cwd=BACKEND_DIR.parent, env=env, stdout=subprocess.PIPE, text=True) + for i in range(per_side)] + assert py.wait() == 0 + outs = [n.communicate(timeout=60)[0] for n in nds] + assert all(n.returncode == 0 for n in nds), "\n".join(outs) + assert sum(json.loads(o.splitlines()[-1])["appended"] for o in outs) == per_side + rows = _queue_rows(ops_env) + assert len(rows) == 2 * per_side, f"混语丢行:期望 {2 * per_side},实际 {len(rows)}" + assert sum(1 for l in rows if "| py_3 |" in l) == 1 + + # ── 台账(ledger.md,T4 冻结形状) ────────────────────────────────── diff --git a/ops/agent/selftest-contract.mjs b/ops/agent/selftest-contract.mjs index 6b4ca14c..4136560f 100644 --- a/ops/agent/selftest-contract.mjs +++ b/ops/agent/selftest-contract.mjs @@ -36,7 +36,10 @@ const SRCS = { agent: src("agent.mjs"), tools: src("tools.mjs"), models: src("mo // tools/shared.mjs;对拍不变式(表内每项真被读/读取∈表)随面延展覆盖 toolsShared: src("tools/shared.mjs"), toolsEnvelope: src("tools/envelope.mjs"), toolsRegistry: src("tools/registry.mjs"), toolsPrecedents: src("tools/precedents.mjs"), - toolsMeasurement: src("tools/measurement.mjs"), toolsDiscovery: src("tools/discovery.mjs") }; + toolsMeasurement: src("tools/measurement.mjs"), toolsDiscovery: src("tools/discovery.mjs"), + // #139 挂账②:锁基建入扫描(零 env 读取,现状无表项;未来若读 env 会被 + // 「读取∈表」不变式拦下,防静默扩面) + toolsQueueLock: src("tools/queue-lock.mjs") }; // ── fixture(in-process 用;golden writePrecedents 同型:env 先于 agent.mjs import)── const TMP = fs.mkdtempSync(path.join(os.tmpdir(), "contract-selftest-")); @@ -67,7 +70,7 @@ const ERROR_CODES = mod.ERROR_CODES; const computeReads = () => { const reads = new Set(); let unknownTemplate = null; - for (const s of [SRCS.agent, SRCS.tools, SRCS.models, SRCS.pipeline, SRCS.pipelineSteps, SRCS.pipelineGates, SRCS.pipelineFaux, SRCS.pipelineFindingExit, SRCS.toolsShared, SRCS.toolsEnvelope, SRCS.toolsRegistry, SRCS.toolsPrecedents, SRCS.toolsMeasurement, SRCS.toolsDiscovery]) { + for (const s of [SRCS.agent, SRCS.tools, SRCS.models, SRCS.pipeline, SRCS.pipelineSteps, SRCS.pipelineGates, SRCS.pipelineFaux, SRCS.pipelineFindingExit, SRCS.toolsShared, SRCS.toolsEnvelope, SRCS.toolsRegistry, SRCS.toolsPrecedents, SRCS.toolsMeasurement, SRCS.toolsDiscovery, SRCS.toolsQueueLock]) { for (const m of s.matchAll(/process\.env\.([A-Za-z_][A-Za-z0-9_]*)/g)) reads.add(m[1]); // 方括号字符串读形(reviewer Minor-2,T2):process.env["QUOTED_NAME"]/['X']——与点形 // 同名集并入对拍;引号族与反引号族(下方)互不重叠。动态下标读形 process.env[expr] @@ -293,6 +296,31 @@ await t("spawn:OPS_PIPELINE=pipeline 委派可落定(死锁回归:exit 2 用法 assert.ok(!r.stderr.includes("unsettled"), `不得再出现 unsettled top-level await:${r.stderr.slice(0, 200)}`); }); +// ── #139 挂账②:锁路径跨语言对拍哨(形制同 selftest.sh §14 锚集对拍:python 文本解析 ↔ JS 实跑) ── +// 两侧锁文件必须同源同址(python ops_root()/QUEUE_LOCK_NAME ↔ node +// queueFile+".lock"),锁路径推导各自实现会漂移=双边锁开窗(互不互认)。 +// CI frontend job 无 backend python 依赖 → 文本解析 _ops_store.py 抽字面量, +// 不 import ipdb(同 §14 python heredoc 不得依赖的裁例)。 +await t("锁路径对拍:python QUEUE_LOCK_NAME ↔ node queueLockPath 同源同址(#139 挂账②)", async () => { + const py = fs.readFileSync(path.join(AGENT, "..", "..", "backend", "ipdb", "_ops_store.py"), "utf8"); + const m = py.match(/QUEUE_LOCK_NAME = "([^"]+)"/); + assert.ok(m, "_ops_store.py 缺 QUEUE_LOCK_NAME 字面量(锁路径契约漂移)"); + assert.equal(m[1], "queue.md.lock", "python 锁名漂移(须 = queue.md+'.lock')"); + assert.ok(py.includes("ops_root() / QUEUE_LOCK_NAME"), + "queue_lock_path 必须 = ops_root()/QUEUE_LOCK_NAME(与 queue.md 同目录)"); + const { queueLockPath } = await import("./tools/queue-lock.mjs"); + // 生产同址链:P2 桥 _ops_rounds._round_env 注入 DOCS_ROOT=ops_root().parent → + // node 默认 queueFile=DOCS_ROOT/ops/queue.md ≡ python ops_root()/queue.md 同一文件, + // 锁路径同理同址才互认互斥 + const h = new (await import("./tools.mjs")).ToolHost("discover", { artifactDir: TMP }); + assert.equal(h.queueFile, path.join(DOCS, "ops", "queue.md"), "host 默认 queueFile 落位漂移"); + assert.equal(queueLockPath(h.queueFile), path.join(DOCS, "ops", m[1]), + "node 锁路径 = queue.md 同目录同名+.lock,须与 python ops_root()/QUEUE_LOCK_NAME 同址"); + // 接线钉死:registry 面真消费锁封装(漂移 = queue_action 绕开跨进程锁,竞态窗复活) + assert.match(SRCS.toolsRegistry, /appendQueueRowLocked/, + "registry.mjs queueAction 未消费 appendQueueRowLocked(竞态窗复活)"); +}); + fs.rmSync(TMP, { recursive: true, force: true }); console.log(`contract-selftest: PASS=${pass} FAIL=${fail}`); process.exit(fail ? 1 : 0); diff --git a/ops/agent/selftest-t4.mjs b/ops/agent/selftest-t4.mjs index c12dafc8..b664bae8 100644 --- a/ops/agent/selftest-t4.mjs +++ b/ops/agent/selftest-t4.mjs @@ -7,7 +7,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { fileURLToPath } from "node:url"; -import { execFile as execFileCb } from "node:child_process"; +import { execFile as execFileCb, spawn as spawnCb } from "node:child_process"; import { promisify } from "node:util"; const execFileP = promisify(execFileCb); @@ -345,6 +345,72 @@ await t("队列:模板渲染落行/幂等去重/vetoable+veto_cmd", async () => assert.equal(fs.readFileSync(path.join(TMP, "queue.md"), "utf8").split("\n").filter((l) => l.includes("cand_x")).length, 1); }); +// ── #139 挂账②:queue.md 跨进程锁(queue-lock.mjs;python append_queue 对称双边) ── +// 修前实证(python 两子进程 2×40 无锁):丢行近半;临界段 = 读-判重-写全段入 +// flock(1) 子进程(与 fcntl.flock 同 flock(2) 锁域,混语互操测试在 +// backend/tests/test_ops_store.py),本节钉 node 面:真子进程互斥/持锁方死内核 +// 即释/fail-closed 信封。锁路径对拍哨在 selftest-contract.mjs。 +const { appendQueueRowLocked, queueLockPath, LOCK_WAIT_S } = await import("./tools/queue-lock.mjs"); +const QUEUE_PAYLOAD = (tmp, src) => ({ + queueFile: path.join(tmp, "queue.md"), round: "discover", action: "REJECT", + source: src, predicate: "sunset_frozen", + evidence: [{ call_id: "c1", field: "age_days", value: 300 }], +}); + +await t("队列锁:跨子进程并发 append 恰 N 行(不丢不重)+锁文件落位", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "q139-")); + try { + const N = 8; // 每路 = 一个真 flock 子进程(生产路径同形:每次入队 spawn 一个) + const rs = await Promise.all(Array.from({ length: N }, (_, i) => + appendQueueRowLocked(QUEUE_PAYLOAD(tmp, `lock_src_${i}`)))); + assert.ok(rs.every((r) => r.ok === true && r.appended === true), JSON.stringify(rs)); + const rows = fs.readFileSync(path.join(tmp, "queue.md"), "utf8") + .split("\n").filter((l) => /^\| 2026-/.test(l)); + assert.equal(rows.length, N, `丢行/重行:期望 ${N},实际 ${rows.length}`); + assert.ok(fs.existsSync(queueLockPath(path.join(tmp, "queue.md"))), "锁文件应随首次写创建"); + } finally { fs.rmSync(tmp, { recursive: true, force: true }); } +}); + +await t("队列锁:同键跨子进程并发恰 1 行(判重也在锁内)", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "q139-")); + try { + const rs = await Promise.all(Array.from({ length: 6 }, () => + appendQueueRowLocked(QUEUE_PAYLOAD(tmp, "same_key")))); + assert.equal(rs.filter((r) => r.appended).length, 1, JSON.stringify(rs)); + const rows = fs.readFileSync(path.join(tmp, "queue.md"), "utf8") + .split("\n").filter((l) => /^\| 2026-/.test(l)); + assert.equal(rows.length, 1); + } finally { fs.rmSync(tmp, { recursive: true, force: true }); } +}); + +await t("队列锁:持锁进程被杀内核即释(无 stale 判龄,接管不等 30s)", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "q139-")); + try { + const lock = queueLockPath(path.join(tmp, "queue.md")); + // 持锁方全灭模拟:flock(1) fork 后子进程继承锁 fd,杀整进程组才是真死 + // (单杀 flock 父进程,已 fork 的 sleep 子进程仍持 open file description 锁) + const holder = spawnCb("flock", [lock, "sleep", "30"], { detached: true }); + await new Promise((r) => setTimeout(r, 300)); // 等 flock 真拿到锁 + process.kill(-holder.pid, "SIGKILL"); // 内核释放 open file description 上的锁 + const t0 = Date.now(); + const r = await appendQueueRowLocked(QUEUE_PAYLOAD(tmp, "after_kill")); + assert.equal(r.ok, true, JSON.stringify(r)); + assert.ok(Date.now() - t0 < LOCK_WAIT_S * 1000, "接管耗时应远小于锁等待上限"); + } finally { fs.rmSync(tmp, { recursive: true, force: true }); } +}); + +await t("队列锁:锁通道缺失 fail-closed(无锁绝不写,铁律②信封)", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "q139-")); + try { + const r = await appendQueueRowLocked(QUEUE_PAYLOAD(tmp, "no_channel"), + { env: { ...process.env, PATH: "/nonexistent-139" } }); // flock(1) 不可寻 + assert.equal(r.ok, false); + assert.equal(r.code, "io_error"); + assert.match(r.reason, /flock/); + assert.ok(!fs.existsSync(path.join(tmp, "queue.md")), "fail-closed 不得写 queue.md"); + } finally { fs.rmSync(tmp, { recursive: true, force: true }); } +}); + // ── PR #177 工具通道修复(判例 2026-10-24-discover-dual-channel-dead):search_precedents 单源 ── // 事故面:search_precedents 失败不可诊断(trace 硬编码 rc:0/lib.sh 吞 stderr 2>/dev/null/ // ENOENT 塞字符串违反 outputSchema Number 槽)。本节钉:golden 字节等价(JS 核心 = lib.sh diff --git a/ops/agent/tools/queue-lock.mjs b/ops/agent/tools/queue-lock.mjs new file mode 100644 index 00000000..a9f47b54 --- /dev/null +++ b/ops/agent/tools/queue-lock.mjs @@ -0,0 +1,118 @@ +// tools/queue-lock.mjs — queue.md 跨进程互斥基建(#139 挂账②;被 registry.mjs +// 面单向消费,不入门面 re-export——C6 29 名导出面不动;零仓依赖自含:临界段 +// 子进程不require node_modules,python 侧混语互操测试可直接调起)。 +// +// 为什么锁:queue_action(本侧,node agent 轮次中)与 POST /api/ops/ruling 的 +// python append_queue(backend/ipdb/_ops_store.py)是同容器两进程对同一 +// queue.md 的「读-判重-写」——互窗存在,后写覆盖前写丢行(修前实证:python 两 +// 子进程 2×40 并发丢行近半)。原判「单边修是假修」,双边闭环。 +// +// 取舍(两案择一,裁定 = spawn util-linux flock(1)): +// - O_EXCL 原子建锁 + 重试退避 + stale 判龄(>30s 强接):被否——python 侧已裁 +// fcntl.flock(内核管锁生死,进程死即释),O_EXCL 协议与 flock 不互锁(python +// O_CREAT 建锁文件后 node O_EXCL 创建永假),自造 stale 判龄还有「强接误杀 +// 活锁」竞态;协议两侧同构才能互认,所以本侧必须同入 flock(2) 锁域。 +// - spawn util-linux flock(1):采纳——node 无内置 flock,零新依赖红线排除 +// with-lockfile 等包;flock(1) 与 fcntl.flock 同一 flock(2) 锁域,跨语言 +// 互认互斥;python:3.12-slim 生产镜像实测 /usr/bin/flock 2.41.5 在位 +// (util-linux 属 Debian Essential),dev 机/CI 亦在位。 +// 代价:父进程无 flock fd 可持,临界段(读-判重-写)必须整体跑在 flock 包装 +// 的子进程里(= 本文件 --append 模式),stdout 单行 JSON 回传结果。 +// +// 锁路径契约(queue.md 同目录同名 + .lock):与 python QUEUE_LOCK_NAME 对称; +// 两侧各自从自家 queue.md 落位推导(本侧 queueFile 即 host 的 DOCS_ROOT/ops/ +// queue.md,tools.mjs:33;python 侧 ops_root()/queue.md),生产形态 P2 桥 +// _ops_rounds._round_env 注入 DOCS_ROOT=ops_root().parent 使两落位同址同文件。 +// 路径漂移防线 = selftest-contract.mjs「锁路径对拍」哨。 +// +// 失败面 fail-closed(铁律②):锁不可得/临界段崩 → 结构化信封 {ok:false, +// code:"io_error"},绝不无锁写——无锁写 = 重开本票要关的竞态窗。 + +import fs from "node:fs"; +import path from "node:path"; +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import { fileURLToPath } from "node:url"; + +const execFileP = promisify(execFile); +const SELF = fileURLToPath(import.meta.url); + +// 锁后缀(与 python QUEUE_LOCK_NAME="queue.md.lock" 同语义:queueFile + ".lock") +export const QUEUE_LOCK_SUFFIX = ".lock"; +// flock -w 等待上限:临界段毫秒级,30s = 异常兜底(持锁方僵死时快速失败进 +// 人裁信封,不无限挂轮次),非正常等待面 +export const LOCK_WAIT_S = 30; + +export function queueLockPath(queueFile) { + return queueFile + QUEUE_LOCK_SUFFIX; +} + +// T4 冻结表头(逐字;与 python _ops_store.QUEUE_HEADER 跨语言双抄是冻结形状的 +// 既定双源,§14 哨对拍钉死不漂移) +const QUEUE_HEADER = + "# 源生命周期动作队列(queue_action 唯一写口;行由 host 模板渲染,rationale 只落工件)\n\n" + + "| 日期 | 轮次 | action | source | predicate | evidence | vetoable |\n" + + "|---|---|---|---|---|---|---|\n"; + +// 临界段本体(只在 flock 包装下跑;导出供单测直测,生产消费面恒走 +// appendQueueRowLocked):读 queue.md → action+source 判重 → 表头惰性首建 → +// append 行。行模板/去重判式 = T4 冻结形状,与 python append_queue 同构。 +export function appendQueueRowCore(payload) { + let existing = ""; + try { existing = fs.readFileSync(payload.queueFile, "utf8"); } catch { /* 首建 */ } + const dedupRe = new RegExp( + `\\| \\d{4}-\\d{2}-\\d{2} \\| [^|]+ \\| ${payload.action} \\| ${payload.source} \\|`); + const dedup = dedupRe.test(existing); + if (!dedup) { + if (!existing) { + fs.mkdirSync(path.dirname(payload.queueFile), { recursive: true }); + fs.writeFileSync(payload.queueFile, QUEUE_HEADER); + } + const ev = payload.evidence + .map((e) => `${e.call_id}:${e.field}=${JSON.stringify(e.value)}`).join(";"); + const row = `| ${new Date().toISOString().slice(0, 10)} | ${payload.round} ` + + `| ${payload.action} | ${payload.source} | ${payload.predicate ?? "—"} | ${ev} ` + + `| 是(否决=删除本行) |`; + fs.appendFileSync(payload.queueFile, `${row}\n`); + } + return { ok: true, appended: !dedup, dedup }; +} + +// 生产消费面(registry.mjs queueAction 唯一):临界段经 flock(1) 子进程执行, +// 锁文件 = queueLockPath(payload.queueFile);失败 = fail-closed 信封不写。 +export async function appendQueueRowLocked(payload, opts = {}) { + let r; + try { + r = await execFileP( + "flock", + ["-w", String(LOCK_WAIT_S), queueLockPath(payload.queueFile), + process.execPath, SELF, "--append", JSON.stringify(payload)], + { timeout: (LOCK_WAIT_S + 30) * 1000, maxBuffer: 64 * 1024, env: opts.env }); + } catch (e) { + const reason = e.code === "ENOENT" + ? "flock(1) 不可用(镜像须含 util-linux;#139 挂账②锁契约)" + : e.killed || e.code === "ETIMEDOUT" + ? `锁等待超时 >${LOCK_WAIT_S}s(持锁方僵死,人裁区排查 queue.md.lock)` + : `临界段子进程失败 rc=${e.code ?? "?"}:${String(e.stderr ?? e.message).slice(0, 200)}`; + return { ok: false, code: "io_error", reason }; + } + const line = r.stdout.trim().split("\n").pop() || ""; + try { + return JSON.parse(line); + } catch { + return { ok: false, code: "io_error", reason: `临界段回传非 JSON:${line.slice(0, 200)}` }; + } +} + +// CLI 临界段模式(只经 flock(1) 调起,不作公共 API):结果恒走 stdout 单行 +// JSON + rc 0(传输面与业务面分离,父进程统一解析;崩溃也是信封不是裸栈)。 +const invokedAsCli = process.argv[1] && path.resolve(process.argv[1]) === SELF; +if (invokedAsCli && process.argv[2] === "--append") { + let res; + try { + res = appendQueueRowCore(JSON.parse(process.argv[3])); + } catch (e) { + res = { ok: false, code: "io_error", reason: String(e?.message ?? e).slice(0, 200) }; + } + process.stdout.write(`${JSON.stringify(res)}\n`); +} diff --git a/ops/agent/tools/registry.mjs b/ops/agent/tools/registry.mjs index 7cd5e834..d710ccfb 100644 --- a/ops/agent/tools/registry.mjs +++ b/ops/agent/tools/registry.mjs @@ -8,6 +8,7 @@ import path from "node:path"; import { Type } from "@earendil-works/pi-ai"; import { B, ROOT, DOCS_ROOT, ts, artifactPath } from "./shared.mjs"; import { envelope } from "./envelope.mjs"; +import { appendQueueRowLocked } from "./queue-lock.mjs"; export const PREDICATES = ["sunset_frozen", "mirror_echo", "zero_asset"]; // discover 谓词表闭集 export const MEASUREMENT_TOOLS = new Set(["fetch_url", "gh_search", "exclusive_probe", "matrix_scan_summary"]); @@ -147,26 +148,26 @@ export const registryMethods = { // 预算计数走真实路径:queue_action 不经 finish()(非测量工具,不进登记表), // 若不在此计数 spent 永不增长、10/轮上限永不触发(评审 Important-1,2026-10-19) this.budgetSpend("queue_action"); - // 幂等去重:同 action+source 已在队列 → 返回既有事实,不双写 - let existing = ""; - try { existing = fs.readFileSync(this.queueFile, "utf8"); } catch { /* 首建 */ } - const dedupRe = new RegExp(`\\| \\d{4}-\\d{2}-\\d{2} \\| [^|]+ \\| ${p.action} \\| ${p.source} \\|`); - const dedup = dedupRe.test(existing); - let lineN = existing ? existing.trimEnd().split("\n").length : 0; const rel = path.relative(ROOT, this.queueFile); const vetoCmd = `sed -i '/| ${p.action} | ${p.source} |/d' ${rel}`; + // #139 挂账②:临界段(读-判重-写)整体入 flock 子进程——与 python + // append_queue(POST /api/ops/ruling)跨进程互斥,同容器两进程丢行窗 + // 关闭;锁不可得 = fail-closed 信封,绝不无锁写(取舍/契约见 + // tools/queue-lock.mjs 模块头)。锁外仅保留与 queue.md 无争的面: + // veto_cmd 计算/rationale 工件落盘。 + const lock = await appendQueueRowLocked({ + queueFile: this.queueFile, round: this.round, action: p.action, source: p.source, + predicate: p.predicate ?? null, evidence: p.evidence, + }); + if (lock.ok === false) { + // 信封面也不经 finish()(queue_action 非测量工具不入登记表,预钞已手工 + // 计;finish 会二次计数+入登记表,破坏两不变式)——铁律②结构化信封手返 + const envr = envelope(lock.code ?? "io_error", lock.reason); + return { content: [{ type: "text", text: `入队失败信封:${envr.code}: ${envr.reason}(fail-closed,未写 queue.md)` }], + details: {}, structuredContent: { ...envr, action: p.action, source: p.source } }; + } + const dedup = lock.dedup; if (!dedup) { - if (!existing) { - fs.mkdirSync(path.dirname(this.queueFile), { recursive: true }); - fs.writeFileSync(this.queueFile, - "# 源生命周期动作队列(queue_action 唯一写口;行由 host 模板渲染,rationale 只落工件)\n\n" - + "| 日期 | 轮次 | action | source | predicate | evidence | vetoable |\n|---|---|---|---|---|---|---|\n"); - lineN = 4; - } - const ev = p.evidence.map((e) => `${e.call_id}:${e.field}=${JSON.stringify(e.value)}`).join(";"); - const row = `| ${new Date().toISOString().slice(0, 10)} | ${this.round} | ${p.action} | ${p.source} | ${p.predicate ?? "—"} | ${ev} | 是(否决=删除本行) |`; - fs.appendFileSync(this.queueFile, `${row}\n`); - lineN += 1; const rpath = artifactPath(this.artifactDir, "queue-rationale", "md"); fs.appendFileSync(rpath, `## ${ts()} ${p.action} ${p.source}\n- predicate: ${p.predicate ?? "(无)"}\n- evidence: ${JSON.stringify(p.evidence)}\n- rationale(模型原话,仅人裁阅读,不进队列行):\n\n${p.rationale}\n\n`); this.artifacts.push(rpath);