小唯
f8ffcde0f4
影子默认(ZHIYI_WRITE_OPS=shadow,零行为变化;=apply才真改库): - 新 distill/write_ops.go: 归一化+MD5哈希去重(ContentHash)、规则基四操作判定 (defaultJudge可注入JudgeFunc/ContradictionFunc,生产可接governance矛盾检测)、 shadow日志JSONL(/var/lib/memoryweave/write-ops-shadow.log) - 四操作: ADD新建/UPDATE旧条标superseded_by+新条active/NOOP丢弃记log/ CONFLICT-MARK双方标conflict交人工(永不自动删) - server.go OnDistillComplete 写路径挂四操作判定(影子只记不改库) - schema: models+sqlite.go+lancedb_ipc.go+rust 加 superseded_by 反向指针列(additive) - 顺手: engine.go Weights/echoScoreVectors 补 MemOS 血统出处注释(借鉴溯源全清单§2.1) - 报告生成器 scripts/tools/zhiyi-writeops-report.py(影子48h后出分布+误判抽样) AC: 四操作各1单测+哈希去重+影子零变更全PASS; go build全过; Rust release build 0; 除既有flaky TestCommitConflictGate_E2E(干净HEAD同败)外go test全绿
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
||
# 织忆写时四操作·影子日志分布报告（t_3e98b65a AC3）
|
||
#
|
||
# 影子观察 48h 后，把 /var/lib/memoryweave/write-ops-shadow.log（JSONL，每行一个
|
||
# WriteVerdict）聚合成四操作分布 + 误判抽样。转 apply 前跑这个，人工判抽样样本，
|
||
# 确认 UPDATE/CONFLICT 误判 ≤10% 才转 apply（止损线，见卡 body）。
|
||
#
|
||
# 用法:
|
||
# zhiyi-writeops-report.py 全部统计
|
||
# zhiyi-writeops-report.py --log <path> 指定影子日志路径
|
||
# zhiyi-writeops-report.py --sample 5 误判抽样条数（默认 5）
|
||
# zhiyi-writeops-report.py --since 48h 只看近 48h（时间过滤，默认全部）
|
||
#
|
||
# 环境变量:
|
||
# ZHIYI_WRITE_OPS_LOG 影子日志路径（默认 /var/lib/memoryweave/write-ops-shadow.log）
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
|
||
def load_log(path):
|
||
if not os.path.exists(path):
|
||
sys.exit(f"影子日志不存在: {path}")
|
||
rows = []
|
||
with open(path) as f:
|
||
for ln, line in enumerate(f, 1):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
rows.append(json.loads(line))
|
||
except json.JSONDecodeError:
|
||
print(f"[warn] 第 {ln} 行非 JSON，跳过: {line[:60]}")
|
||
return rows
|
||
|
||
|
||
def parse_duration(s):
|
||
"""'48h' → 48*3600 秒；'7d' → 7*86400；None → 全部。"""
|
||
if not s:
|
||
return None
|
||
s = s.strip().lower()
|
||
mult = 1
|
||
if s.endswith("h"):
|
||
mult = 3600
|
||
s = s[:-1]
|
||
elif s.endswith("d"):
|
||
mult = 86400
|
||
s = s[:-1]
|
||
elif s.endswith("m"):
|
||
mult = 60
|
||
s = s[:-1]
|
||
return float(s) * mult
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="织忆写时四操作影子日志分布报告")
|
||
ap.add_argument("--log", default=os.getenv("ZHIYI_WRITE_OPS_LOG",
|
||
"/var/lib/memoryweave/write-ops-shadow.log"))
|
||
ap.add_argument("--sample", type=int, default=5, help="误判抽样条数（默认 5）")
|
||
ap.add_argument("--since", default=None, help="时间过滤，如 48h/7d（默认全部）")
|
||
args = ap.parse_args()
|
||
|
||
rows = load_log(args.log)
|
||
if not rows:
|
||
print("影子日志为空 —— 写时四操作尚无判定记录。")
|
||
return 1
|
||
|
||
since = parse_duration(args.since)
|
||
if since is not None:
|
||
import time
|
||
cutoff = time.time() - since
|
||
rows = [r for r in rows if r.get("timestamp", "") >=
|
||
time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(cutoff))[:19]]
|
||
print(f"[filter] 只统计近 {args.since} 的判定")
|
||
|
||
total = len(rows)
|
||
dist = {"ADD": 0, "UPDATE": 0, "NOOP": 0, "CONFLICT-MARK": 0, "未知": 0}
|
||
for r in rows:
|
||
op = r.get("op", "未知")
|
||
dist[op] = dist.get(op, 0) + 1
|
||
|
||
print("=" * 52)
|
||
print(f"写时四操作影子分布报告（共 {total} 条判定）")
|
||
print("=" * 52)
|
||
for op in ["ADD", "UPDATE", "NOOP", "CONFLICT-MARK", "未知"]:
|
||
n = dist.get(op, 0)
|
||
pct = 100.0 * n / total if total else 0
|
||
bar = "#" * int(pct / 2)
|
||
print(f" {op:<14} {n:>5} {pct:5.1f}% {bar}")
|
||
|
||
update_conflict = dist.get("UPDATE", 0) + dist.get("CONFLICT-MARK", 0)
|
||
if total:
|
||
print("-" * 52)
|
||
print(f"UPDATE+CONFLICT 占比 = {100.0 * update_conflict / total:.1f}% "
|
||
f"（止损线：误判 >10% 则保持影子不转 apply）")
|
||
|
||
# 误判抽样：优先抽 UPDATE / CONFLICT（高风险操作），按判定时间新→旧
|
||
risky = [r for r in rows if r.get("op") in ("UPDATE", "CONFLICT-MARK")]
|
||
risky.sort(key=lambda r: r.get("timestamp", ""), reverse=True)
|
||
print("=" * 52)
|
||
print(f"误判抽样（前 {min(args.sample, len(risky))} 条高风险判定，人工复核）")
|
||
print("=" * 52)
|
||
for r in risky[:args.sample]:
|
||
print(f" [{r.get('op')}] sim={r.get('similarity', 0):.3f} "
|
||
f"target={r.get('target_id', '')}")
|
||
print(f" 候选: {r.get('candidate', '')[:70]}")
|
||
print(f" 理由: {r.get('reason', '')}")
|
||
print(f" 时间: {r.get('timestamp', '')}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |