313 lines
14 KiB
Python
313 lines
14 KiB
Python
# ============================================================
|
||
# 兜底验证器 v1（moduleId: core-fallback-verify, 可重生 ✅）
|
||
# 《Pi-Agent兜底能力详细方案》§4.1/§4.5 + GOAL-P2 交付 4 + P2-DESIGN §4：
|
||
# 兜底执行后的 world diff、验证规则（行数/数量对账）、报告生成。
|
||
# 铁律：报告里的每个数字都只许来自冻结快照（cp_before/cp_after 的
|
||
# world 深拷贝），绝不引用 Pi 报告文本或 Pi 自述。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# 对账覆盖的业务主数据表（分表 diff 的固定口径）
|
||
_DIFF_TABLES = (
|
||
"salesOrders", "flexOrders", "materials", "flexMaterials",
|
||
"flexEquipment", "flexMolds", "flexOperations", "flexRoutings",
|
||
"flexBom", "productionOrders", "workOrders",
|
||
)
|
||
|
||
# 数量对账类规则的数据源字段（存在即计入 quantityDelta）
|
||
_QUANTITY_FIELDS = ("quantity", "stock")
|
||
|
||
|
||
def _canonical(item: Any) -> str:
|
||
"""条目的规范 JSON（排序键 + 紧凑分隔符），modified 判定与指纹共用口径。"""
|
||
return json.dumps(item, ensure_ascii=True, sort_keys=True,
|
||
separators=(",", ":"), default=str)
|
||
|
||
|
||
def _entry_key(item: dict) -> tuple:
|
||
"""条目身份键：数值 id 优先，其次 orderNo/code，最后整条 canonical（防无键表）。"""
|
||
if isinstance(item, dict):
|
||
if isinstance(item.get("id"), int):
|
||
return ("id", item["id"])
|
||
if item.get("orderNo"):
|
||
return ("orderNo", str(item["orderNo"]))
|
||
if item.get("code"):
|
||
return ("code", str(item["code"]))
|
||
return ("json", _canonical(item))
|
||
|
||
|
||
def _quantity_of(item: dict) -> float:
|
||
total = 0.0
|
||
for field in _QUANTITY_FIELDS:
|
||
value = item.get(field) if isinstance(item, dict) else None
|
||
if isinstance(value, (int, float)):
|
||
total += float(value)
|
||
return total
|
||
|
||
|
||
def world_diff(before: dict, after: dict) -> dict:
|
||
"""分表 diff：{table: {"added","removed","modified","quantityDelta"}}。
|
||
|
||
modified 判定 = 同身份键条目内容变化（canonical JSON 不等）；
|
||
quantityDelta 对含 quantity/stock 字段的条目求和（added 减 removed）。
|
||
"""
|
||
diff: dict[str, dict] = {}
|
||
for table in _DIFF_TABLES:
|
||
before_rows = before.get(table) or []
|
||
after_rows = after.get(table) or []
|
||
before_map = {_entry_key(r): r for r in before_rows if isinstance(r, dict)}
|
||
after_map = {_entry_key(r): r for r in after_rows if isinstance(r, dict)}
|
||
added_keys = [k for k in after_map if k not in before_map]
|
||
removed_keys = [k for k in before_map if k not in after_map]
|
||
modified = sum(
|
||
1 for k in before_map.keys() & after_map.keys()
|
||
if _canonical(before_map[k]) != _canonical(after_map[k])
|
||
)
|
||
qty_delta = (
|
||
sum(_quantity_of(after_map[k]) for k in added_keys)
|
||
- sum(_quantity_of(before_map[k]) for k in removed_keys)
|
||
)
|
||
if added_keys or removed_keys or modified:
|
||
diff[table] = {
|
||
"added": len(added_keys),
|
||
"removed": len(removed_keys),
|
||
"modified": modified,
|
||
"quantityDelta": qty_delta,
|
||
}
|
||
return diff
|
||
|
||
|
||
def check_expectations(plan: dict, diff: dict) -> list[dict]:
|
||
"""逐条比对计划 expected（结构化字段）与实际 diff。
|
||
|
||
返回 [{"step","expect","actual","ok"} ...]；容差 = 0——任一声明字段不等即 ok=False，
|
||
调用方据此把报告 verdict 判为 MISMATCH（显式，不圆场）。
|
||
"""
|
||
checks: list[dict] = []
|
||
for step in plan.get("steps") or []:
|
||
seq = step.get("seq")
|
||
for expect in step.get("expected") or []:
|
||
if not isinstance(expect, dict) or not expect.get("table"):
|
||
continue
|
||
table = str(expect["table"])
|
||
actual = (diff.get(table) or {}).copy()
|
||
actual.setdefault("added", 0)
|
||
actual.setdefault("removed", 0)
|
||
actual.setdefault("modified", 0)
|
||
ok = True
|
||
for field in ("added", "removed", "modified"):
|
||
if field in expect and expect[field] is not None \
|
||
and int(expect[field]) != int(actual.get(field) or 0):
|
||
ok = False
|
||
checks.append({
|
||
"step": seq,
|
||
"expect": {k: expect[k] for k in ("table", "added", "removed", "modified")
|
||
if k in expect},
|
||
"actual": {"table": table,
|
||
**{k: actual.get(k, 0) for k in ("added", "removed", "modified")}},
|
||
"ok": ok,
|
||
})
|
||
return checks
|
||
|
||
|
||
def diff_summary_lines(diff: dict) -> list[str]:
|
||
"""分表 diff 的人类可读摘要行（回复文案与报告共用）。"""
|
||
lines = []
|
||
for table, d in diff.items():
|
||
parts = []
|
||
if d["added"]:
|
||
parts.append(f"+{d['added']}")
|
||
if d["removed"]:
|
||
parts.append(f"-{d['removed']}")
|
||
if d["modified"]:
|
||
parts.append(f"~{d['modified']}")
|
||
line = f"{table} {'/'.join(parts)}"
|
||
if d.get("quantityDelta"):
|
||
line += f"（数量净变化 {d['quantityDelta']:g}）"
|
||
lines.append(line)
|
||
return lines or ["（无业务主数据变化）"]
|
||
|
||
|
||
def build_report(run_dir: Path, plan: dict, *,
|
||
cp_before_id: str, cp_after_id: str,
|
||
before_world: dict, after_world: dict,
|
||
checks: list[dict]) -> Path:
|
||
"""生成 outbox/verify-report.md：计划摘要 / 分表 diff 表 / 对账结论 / 证据引用。
|
||
|
||
数字来源 = 两个冻结快照的 world（调用方保证传入的是 checkpoint 仓内深拷贝），
|
||
返回报告路径。verdict：全部对账通过 = PASS；任一不符 = MISMATCH（显式标注）。
|
||
"""
|
||
diff = world_diff(before_world, after_world)
|
||
verdict = "PASS" if all(c["ok"] for c in checks) else "MISMATCH"
|
||
|
||
lines = [
|
||
"# 兜底执行验证报告",
|
||
"",
|
||
f"- 运行：{plan.get('runId') or ''}",
|
||
f"- 场景：{plan.get('scenario') or ''} · 步骤数 {len(plan.get('steps') or [])}",
|
||
f"- 检查点：执行前 `{cp_before_id}` → 执行后 `{cp_after_id}`",
|
||
"",
|
||
"## 分表 diff（before → after）",
|
||
"",
|
||
"| 表 | 新增 | 移除 | 修改 | 数量净变化 |",
|
||
"|----|------|------|------|-----------|",
|
||
]
|
||
for table in _DIFF_TABLES:
|
||
d = diff.get(table)
|
||
if not d:
|
||
continue
|
||
lines.append(f"| {table} | {d['added']} | {d['removed']} | {d['modified']} "
|
||
f"| {d['quantityDelta']:g} |")
|
||
if not diff:
|
||
lines.append("| （无变化） | 0 | 0 | 0 | 0 |")
|
||
lines += ["", "## 对账结论", ""]
|
||
if checks:
|
||
for c in checks:
|
||
mark = "✅" if c["ok"] else "❌"
|
||
lines.append(f"- {mark} 步骤{c['step']} 期望 {json.dumps(c['expect'], ensure_ascii=False)}"
|
||
f" · 实际 {json.dumps(c['actual'], ensure_ascii=False)}")
|
||
else:
|
||
lines.append("- （计划未声明结构化预期，仅呈现实际 diff）")
|
||
lines += [
|
||
"",
|
||
f"**verdict: {verdict}**",
|
||
"",
|
||
f"本报告全部数字来自检查点 {cp_before_id} 与 {cp_after_id} 的冻结快照。",
|
||
]
|
||
out = Path(run_dir) / "outbox" / "verify-report.md"
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
out.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return out
|
||
|
||
|
||
def report_fingerprint(text: str) -> str:
|
||
"""报告文本 sha256（审计 rationale 引用用，不落全文）。"""
|
||
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# P3：S4 沙盒「草稿不碰主干」物理判决 + S6 恢复后对账（P3-DESIGN §5.2/§6.5）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def verify_sandbox_no_main_writes(*, before_fp: str, after_world: dict,
|
||
new_audit_events: list[dict]) -> dict:
|
||
"""S4 沙盒执行验证器断言（§5.2）：前后世界指纹相等 + 执行区间零 WORLD_WRITE。
|
||
|
||
指纹口径 = harness.world_fingerprint（auditEvents 等 append-only 键已被排除，
|
||
审计自然增长不误报）；WORLD_WRITE 扫描区间由调用方切片（执行前水位之后）。
|
||
返回 {"ok", "fingerprintEqual", "worldWrites", "offenders"}——任一不满足
|
||
ok=False，调用方按执行失败处理（自动回滚 + 显式文案）。
|
||
"""
|
||
from server.agent_core import harness
|
||
|
||
after_fp = harness.world_fingerprint(after_world)
|
||
writes = [e for e in new_audit_events if e.get("category") == "WORLD_WRITE"]
|
||
return {
|
||
"ok": after_fp == before_fp and not writes,
|
||
"fingerprintEqual": after_fp == before_fp,
|
||
"worldWrites": len(writes),
|
||
"offenders": [str(e.get("action") or "") for e in writes][:10],
|
||
}
|
||
|
||
|
||
def reconcile_external(world: dict, domain: dict, external: dict) -> list[dict]:
|
||
"""S6 对账比对（§6.5）：对账域 × 外部快照 → 三档结论。
|
||
|
||
domain = {externalWoId: woId | None}（编排器确定的双来源并集）；
|
||
external = {externalWoId: 外部响应 dict | None}（None = 拉取失败/接口缺失）。
|
||
比对字段：status / progressPct / qtyDone（世界侧 workOrders 投影）。
|
||
三档：MATCH（一致）/ DRIFT（列差异字段）/ MISSING（单边存在）。
|
||
每行附带 pendingSync：本地存在 syncStatus=PENDING_SYNC 的离线落账报工记录
|
||
（断连补录的可识别状态）——对账如实呈现待同步事实；自动补推属后续轮次。
|
||
"""
|
||
local_index: dict[str, dict] = {}
|
||
for table in ("flexWorkOrders", "workOrders"):
|
||
for wo in world.get(table) or []:
|
||
ext = wo.get("mesExternalId") if isinstance(wo, dict) else None
|
||
if ext:
|
||
local_index[str(ext)] = wo
|
||
pending_index: dict[str, int] = {}
|
||
for link in world.get("mesLinks") or []:
|
||
if isinstance(link, dict) and link.get("kind") == "report" \
|
||
and link.get("syncStatus") == "PENDING_SYNC" and link.get("externalWoId"):
|
||
key = str(link["externalWoId"])
|
||
pending_index[key] = pending_index.get(key, 0) + 1
|
||
rows: list[dict] = []
|
||
for ext_id in sorted(domain):
|
||
local = local_index.get(ext_id)
|
||
snap = external.get(ext_id)
|
||
base = {"externalWoId": ext_id, "woId": domain.get(ext_id)
|
||
or (local or {}).get("id"),
|
||
"pendingSync": pending_index.get(ext_id, 0)}
|
||
if snap is None:
|
||
rows.append({**base, "verdict": "MISSING",
|
||
"side": "external-unavailable", "fields": []})
|
||
continue
|
||
if local is None:
|
||
rows.append({**base, "verdict": "MISSING",
|
||
"side": "local-missing", "fields": []})
|
||
continue
|
||
fields: list[dict] = []
|
||
for field in ("status", "progressPct", "qtyDone"):
|
||
local_value = local.get(field, 0 if field != "status" else None)
|
||
external_value = snap.get(field)
|
||
if external_value is None:
|
||
continue
|
||
if str(local_value) != str(external_value):
|
||
fields.append({"field": field, "local": local_value,
|
||
"external": external_value})
|
||
rows.append({**base,
|
||
"verdict": "DRIFT" if fields else "MATCH",
|
||
"side": "both", "fields": fields})
|
||
return rows
|
||
|
||
|
||
def build_reconcile_report(run_dir: Path, rows: list[dict], manifest: dict) -> Path:
|
||
"""outbox/reconcile-report.md：逐工单一行 + 汇总计数（机器生成）。
|
||
|
||
铁律：报告数字只许来自对账域的冻结比对结果与 manifest 所列冻结文件。
|
||
"""
|
||
counts = {
|
||
"MATCH": sum(1 for r in rows if r["verdict"] == "MATCH"),
|
||
"DRIFT": sum(1 for r in rows if r["verdict"] == "DRIFT"),
|
||
"MISSING": sum(1 for r in rows if r["verdict"] == "MISSING"),
|
||
}
|
||
pending_total = sum(int(r.get("pendingSync") or 0) for r in rows)
|
||
lines = [
|
||
"# S6 恢复后对账报告",
|
||
"",
|
||
f"- 运行：{manifest.get('runId') or ''}",
|
||
(f"- 对账域 {len(rows)} 项：一致 {counts['MATCH']} / "
|
||
f"漂移 {counts['DRIFT']} / 单边缺失 {counts['MISSING']}"),
|
||
(f"- 待同步（PENDING_SYNC）{pending_total} 笔：断连期间本地落账、"
|
||
"尚未推送 MES 的报工记录（自动补推属后续轮次，本轮如实呈现）"),
|
||
(f"- 证据：外部响应原文 {len(manifest.get('items') or [])} 份"
|
||
"（sha256 见 reconcile-manifest.json）"),
|
||
"",
|
||
"| 外部工单 | 本地工单 | 结论 | 差异 |",
|
||
"|----------|----------|------|------|",
|
||
]
|
||
for r in rows:
|
||
diffs = "；".join(
|
||
f"{f['field']} 本地={f['local']} 外部={f['external']}"
|
||
for f in r.get("fields") or []) or r.get("side") or "—"
|
||
if r.get("pendingSync"):
|
||
diffs += f"；本地待同步（PENDING_SYNC）×{int(r['pendingSync'])}"
|
||
lines.append(f"| {r.get('externalWoId')} | {r.get('woId') or '—'} "
|
||
f"| {r['verdict']} | {diffs} |")
|
||
lines += [
|
||
"",
|
||
("本报告全部数字来自对账域的冻结比对结果与 reconcile-manifest.json "
|
||
"所列冻结文件；差异纠正请回到逐笔补录确认卡（对账本身永远不写）。"),
|
||
]
|
||
out = Path(run_dir) / "outbox" / "reconcile-report.md"
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
out.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return out
|