ops-backlog #139:轮次内存峰值入台账 + queue.md 双边 flock

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).

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,17 +500,69 @@ 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 → 收口。
 
     ops 树形状初始化在 spawn_round 内(单一拼写,幂等)。任务态收口走
     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)

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/<id>.md(证据在条目里
   不复制),幂等键首槽 = item:<id>(与包名空间隔离),队列行轮次列 =
   "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}