Files
AgentLab/backend/tests/bundle_samples.py
T
ThearchyHelios 6ff9d7155b M3 可验证报告核对正文显示的字:按服务端的显示规则把引用的值重新渲染、逐字比对
验收发现:页面只核「引用的值 = 口径卡的值或快照的格子」,不核正文显示的字。只改显示文字(74,613.65元 改成
84,613.65元、华南改成华东)、引用值不动、重算全部哈希的伪造页照样「全部通过」,读者看到的数和核过的数不是同一个。

- verifier.js 移植 engine/evidence.py 的 render_number / render_cell:千分位、不分组、小数位(含负数位)、比率显示成
  百分数、万 / 亿换算、pct、int、.N、单位后缀、负数、0,文字压成一行(按 Python 的空白),竖线换成 ¦,ISO 日期时间按
  文档记下的单元格显示规则版本。数照 Python 的 Decimal 算:系数用 BigInt,ROUND_HALF_UP,浮点取最短写法,整数取 JSON
  原文(新写的 numberTokens 读出每个数的原文,超过 2^53 的订单号一位不差)
- ⑤ 在引用的值对上之后再核显示文字(只去掉首尾空白):对不上判不通过,原因点名「正文显示「X」,引用的值是「Y」」,
  数字另写按显示规则应显示的字;渲染不出来的(不认识的换算、显示规则版本、对象和列表、非 ASCII 数字、缺 BigInt 的
  浏览器)判「无法核对显示文字」,不判通过。出处面板改成列出按显示规则应为的字,一致时注明,页脚说明同步
- 在 node 里载入时(没有 document)交出核对用的纯函数,tests/test_bundle_shown.py 拿服务端的渲染逐条对照约八万个
  组合,并在 node 里整份核对:经营月报样本每种格式原样通过,只改显示文字的伪造逐个变红并点名
- 样本页加经营月报(formats);check-bundle 在真导出上只改一个数字(有文字取值时也改一个)的显示文字,在经营月报上
  改文字取值、数字、换算成万的口径卡引用,各自变红并点名,原样导出照旧全绿
2026-10-07 00:17:22 +08:00

306 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""可验证报告的虚构样本页:给 scripts/check-bundle.mjs 在浏览器里核对(pytest 不执行页面上的核对脚本)。
python -m tests.bundle_samples <目录>
在目录里写下几份页面(<名字>.html)。每份都是 build_bundle + render_html 真实生成的,封存哈希按 runner 的算法
(artifact_store.manifest_hash)对事件真实算出,报告文档用 compose_doc 按证据目录真实组装(标题、段落、列表、
表格、粗体、斜体,数字和取值的引用都和真实运行一样)。check-bundle 断网打开它们,逐项断言核对结果和原因,再在
这几份上做篡改。这样遮罩快照、多份报告、早期运行、服务端缺文件这些分支,不论真数据里有没有,都有人在浏览器里核过。
数据全是虚构的:门店、会员、金额都是编的。
- base:一份报告,引用口径卡、门店流水(没有遮罩)、会员消费(有遮罩)各一处,一个没有出处的数,一张门店流水整表
(数字列右对齐,数字与取值共 21 个,超过折叠的条数);一条封存之后追加的按需裁判
- two-reports:另有一份月报,两份报告各自的引用链
- legacy:早期运行,封存范围按结束事件推算(运行结束事件在范围外,运行结果无法核对)
- failed:失败结束的运行(没有出具判定),封存说明多一句「继续运行会重新封存」
- continued:早期运行先失败、经继续运行跑到成功:范围里有那条运行失败,之后运行又继续了,成功结束的那条在范围外
- degraded:两条出具判定,先完整出具、后降档出具,认最后一条
- missing-doc / missing-card:导出时服务端已没有报告文档 / 口径卡的文件
- masked-card:口径卡上一个指标的输入取自会员消费的号码列(设了遮罩),卡上记着号码原值:整张卡只给 id,引用它的
数字无法核对
- masked-source:会员库设了号码遮罩(查询当时记下了,数据源现在也设着)。报告引用会员按地区汇总的查询(没碰遮罩列:
快照随文件附带,数字核对得过)和选出了号码的那份会员消费(只给哈希)
- nan:运行开始事件和门店流水快照里有 json.dumps 写出的裸 NaN(快照里那一列报告没有引用)
- stray:报告自查记下的文档是另一次运行写的
- formats:经营月报,每种显示格式各引用一处:口径卡的千分位加货币单位、换算成万 / 亿、比率显示成百分数、负数、不分组、
小数位与 |.N / |int、pct;流水快照里的文字取值(地区)、DECIMAL 落成的数字文本、负数、换算成万、超过 2^53 的订单号、
ISO 日期时间(文档记下 2 版显示规则)。check-bundle 在它上面只改显示文字做伪造,tests/test_bundle_shown.py 在 node 里
逐个格式核一遍
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
from typing import Any
from app.bundle.build import build_bundle, render_html
from app.core.artifact_store import canonical_json, content_hash, manifest_hash
from app.engine.evidence import build_catalog, compose_doc
RUN_ID = "5a3c0f4b9d2e4e1f8b7a6c5d4e3f2a1b"
RUN = {"id": RUN_ID, "workflow_id": "wf-sample-weekly", "workflow_name": "门店周报", "version": 3,
"version_hash": "ab" * 32, "run_class": "formal", "status": "succeeded",
"started_at": "2026-09-07T01:00:00+00:00", "finished_at": "2026-09-07T01:02:03+00:00"}
GRAPH = {"nodes": [{"id": n, "type": t, "data": {"label": label}} for n, t, label in [
("start", "input", "开始"), ("fetch", "tool", "门店流水"), ("members", "tool", "会员消费"),
("regions", "tool", "会员地区汇总"), ("card", "metrics", "周报口径"), ("write", "report", "写周报"),
("monthly", "report", "写月报"), ("out", "output", "成果")]]}
EXPORTED_AT = "2026-10-05T08:00:00+00:00"
STORES = [["青石店", 1280.5], ["白沙店", "960.00"], ["临川店", 732], ["望江店", 1105.25], ["枫桥店", 418],
["云岭店", 2210], ["石桥店", 655.5], ["柳湾店", 980]]
CRM = {"columns": ["name", "phone", "spend"], "column_types": {"name": "text", "phone": "text", "spend": "number"},
"rows": [["林一", "000-0001", 320.0], ["周二", "000-0002", 188.5]], "source": "crm",
"sql": "SELECT name, phone, spend FROM members ORDER BY id", "truncated": False, "mask_columns": ["phone"]}
#: 会员按地区汇总:同一个设了号码遮罩的会员库,SQL 没碰号码列
REGIONS = {"columns": ["region", "spend"], "column_types": {"region": "text", "spend": "number"},
"rows": [["华东", 320.0], ["华南", 188.5]], "source": "crm",
"sql": "SELECT region, SUM(spend) AS spend FROM members GROUP BY region ORDER BY region", "truncated": False,
"mask_columns": ["phone"]}
CARD = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v2",
"metrics": [{"id": "gmv", "name": "销售额", "value": 8381.25, "unit": "元", "decimals": 2, "format": "thousands"}]}
WEEKLY = ("## 门店周报\n\n本周销售额 [[m:gmv]],*按发生周统计*。青石店流水 [[v:Q1.r0.amount]],白沙店 [[v:Q1.r1.amount]],"
"**会员消费** [[v:Q2.r0.spend]],另有 42 笔退款待入账。[[see:Q1]]\n\n- 云岭店流水最高\n- 枫桥店流水最低\n\n"
"[[table:Q1 cols=store,amount rows=0-7]]")
MONTHLY = "## 门店月报\n\n本月销售额 [[m:gmv]],青石店流水 [[v:Q1.r0.amount]]。[[see:Q1]]"
#: 经营月报的口径卡:每个指标一种显示格式(千分位加货币单位、不分组、比率显示成百分数、负数、小数位、不写小数位)
FORMATS_CARD = {"kind": "metric_set", "caliber": "经营月报口径", "caliber_version": "v1", "metrics": [
{"id": "gmv", "name": "销售额", "value": 74613.65, "unit": "元", "decimals": 2, "format": "thousands"},
{"id": "annual", "name": "年度销售额", "value": 1234567890.5, "unit": "元", "decimals": 2, "format": "thousands"},
{"id": "refund", "name": "退款", "value": -3120.5, "unit": "元", "decimals": 2, "format": "thousands"},
{"id": "conv_rate", "name": "转化率", "value": 0.0235, "unit": "", "decimals": 4, "format": "percent_of_ratio"},
{"id": "growth", "name": "环比", "value": -0.0815, "unit": "", "decimals": 3, "format": "percent_of_ratio"},
{"id": "orders", "name": "订单数", "value": 12840, "unit": "单", "decimals": 0, "format": "plain"},
{"id": "avg", "name": "客单价", "value": 5.8113, "unit": "元", "decimals": 2, "format": "thousands"},
{"id": "share", "name": "线上占比", "value": 0.346, "unit": "", "decimals": None, "format": "thousands"},
{"id": "export", "name": "出口额", "value": 15230.4, "unit": "美元", "decimals": 1, "format": "thousands"},
]}
#: 经营流水:地区是文字,金额一格是 DECIMAL 落成的数字文本、一格是负数,订单号超过 2^53,付款时间是 ISO 日期时间
LEDGER = {"columns": ["region", "gmv", "order_no", "paid_at"],
"column_types": {"region": "text", "gmv": "number", "order_no": "number", "paid_at": "datetime"},
"rows": [["华南", "45678.50", 202609071234567890, "2026-09-07T08:21:39"],
["华东", -1200.75, 202609071234567891, "2026-09-08T19:02:11"]],
"source": "shop", "sql": "SELECT region, gmv, order_no, paid_at FROM ledger ORDER BY region DESC",
"truncated": False}
FORMATS = ("## 经营月报\n\n本月销售额 [[m:gmv]](约 [[m:gmv|万]]),年度累计 [[m:annual|亿]],退款 [[m:refund]]。"
"转化率 [[m:conv_rate]],环比 [[m:growth]],订单 [[m:orders]]。客单价 [[m:avg]],精确些是 [[m:avg|.3]],"
"取整是 [[m:avg|int]];线上占比 [[m:share|pct]],出口额 [[m:export]]。\n\n"
"[[v:Q1.r0.region]]流水 [[v:Q1.r0.gmv]](约 [[v:Q1.r0.gmv|万]]),[[v:Q1.r1.region]]冲减 [[v:Q1.r1.gmv]]。"
"最大一单的订单号 [[v:Q1.r0.order_no]],付款时间 [[v:Q1.r0.paid_at]]。[[see:Q1]]")
#: 会员周报:Q3 是会员按地区汇总(没碰遮罩列),Q2 是选出了号码的会员消费
MEMBERS = ("## 会员周报\n\n华东会员消费 [[v:Q3.r0.spend]],华南 [[v:Q3.r1.spend]]。[[see:Q3]]\n\n"
"会员林一消费 [[v:Q2.r0.spend]]。")
def stored(obj: Any) -> tuple[str, bytes]:
"""照工件库的样子落一份:规范 JSON 的字节,id 是它的 sha256。"""
raw = canonical_json(obj).encode()
return content_hash(raw), raw
def shop(nan: bool = False) -> dict[str, Any]:
"""门店流水的查询快照。ratio 那一列报告不引用;nan=True 时第一行是浮点 NaN(json.dumps 写成裸的 NaN)。"""
rows = [[name, amount, float("nan") if nan and i == 0 else round(0.05 * (i + 1), 2)]
for i, (name, amount) in enumerate(STORES)]
return {"columns": ["store", "amount", "ratio"],
"column_types": {"store": "text", "amount": "number", "ratio": "number"}, "rows": rows, "source": "shop",
"sql": "SELECT store, amount, ratio FROM orders ORDER BY id", "truncated": False}
def masked_card(crm_id: str) -> dict[str, Any]:
"""口径卡多一个指标:联系电话位数,输入是会员消费第 0 行的号码(phone 设了遮罩),卡上记着号码原值和代入式。"""
cell = "cell(nodes.members, 0, 'phone')"
contact = {"id": "contact", "name": "联系电话位数", "value": 8, "unit": "位", "decimals": 0, "format": "thousands",
"expression": f"len({cell})", "substituted": f"len('{CRM['rows'][0][1]}')",
"inputs": [{"path": cell, "value": CRM["rows"][0][1], "node_id": "members", "via": "tool_cell",
"artifact": crm_id, "locator": {"row": 0, "column": "phone"}, "status": "verified"}]}
return {**CARD, "metrics": [*CARD["metrics"], contact]}
class Sample:
"""一次虚构的已封存运行:工件(原字节)、证据目录、报告文档、事件。masked=True 时口径卡带着取自遮罩列的输入;
regions=True 时会员库另有一次按地区汇总的查询(目录里排在 Q3)。card、table 换掉口径卡和门店流水的快照(经营月报)。"""
def __init__(self, *, nan: bool = False, masked: bool = False, regions: bool = False,
card: dict[str, Any] | None = None, table: dict[str, Any] | None = None) -> None:
self.files: dict[str, bytes] = {}
snapshot = table or shop(nan)
self.shop_id = self.put(snapshot)
self.crm_id = self.put(CRM)
self.card = card or (masked_card(self.crm_id) if masked else CARD)
self.card_id = self.put(self.card)
self.entries = {
"fetch": {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": self.shop_id,
"tool": "db_query__shop", "source": "shop", "columns": snapshot["columns"],
"rows": len(snapshot["rows"]), "truncated": False},
"members": {"kind": "query", "node_id": "members", "exec": 1, "artifact": self.crm_id,
"tool": "db_query__crm", "source": "crm", "columns": CRM["columns"], "rows": len(CRM["rows"]),
"truncated": False},
}
if regions:
self.entries["regions"] = {"kind": "query", "node_id": "regions", "exec": 1,
"artifact": self.put(REGIONS), "tool": "db_query__crm", "source": "crm",
"columns": REGIONS["columns"], "rows": len(REGIONS["rows"]), "truncated": False}
self.entries["card"] = {"kind": "metric_set", "node_id": "card", "artifact": self.card_id}
self.ledger = list(self.entries.values())
self.catalog = build_catalog(nodes={"card": self.card}, ledger=self.ledger)
def put(self, obj: Any) -> str:
artifact, raw = stored(obj)
self.files[artifact] = raw
return artifact
def load(self, artifact: str) -> Any:
raw = self.files.get(artifact)
return json.loads(raw) if raw is not None else None
def doc(self, text: str, node_id: str, run_id: str = RUN_ID) -> str:
return self.put(compose_doc(text, self.catalog, node_id=node_id, run_id=run_id, loader=self.load))
def events(self, reports: dict[str, str], *, ending: str = "finished", tiers: tuple[str, ...] = ("formal",),
nan: bool = False) -> tuple[list[tuple[int, str, str | None, Any]], int]:
"""(全部事件, 封存范围)。封存范围到结束事件为止,后面再追加一条按需裁判。"""
rows: list[tuple[int, str, str | None, Any]] = []
def add(etype: str, node_id: str | None, data: dict[str, Any]) -> None:
rows.append((len(rows) + 1, etype, node_id, data))
add("run.started", None, {"nodes": len(GRAPH["nodes"]), "resumed": False, "memory_scope": "default",
"collection": "default", "approval_default": None, "tool_trust": "formal",
"agent_limits": None, "data_versions": None, "replay_protocol": 2,
**({"sample_rate": float("nan")} if nan else {})})
add("node.started", "start", {"node_type": "input", "label": "开始"})
add("node.finished", "start", {"duration_ms": 1, "attempt": 1, "preview": {"week": "2026-W37"}})
add("node.started", "fetch", {"node_type": "tool", "label": "门店流水"})
add("tool.end", "fetch", {"tool": "db_query__shop", "duration_ms": 12, "query_artifact": self.shop_id,
"preview": json.dumps({"columns": ["store", "amount"], "rows": STORES[:2]},
ensure_ascii=False)})
add("node.finished", "fetch", {"duration_ms": 14, "attempt": 1, "preview": "{\"rows\": …}",
"evidence": [self.entries["fetch"]]})
add("tool.end", "members", {"tool": "db_query__crm", "duration_ms": 9, "query_artifact": self.crm_id,
"preview": json.dumps(CRM["rows"][:1], ensure_ascii=False)})
add("node.finished", "members", {"duration_ms": 10, "attempt": 1, "evidence": [self.entries["members"]]})
if "regions" in self.entries:
add("tool.end", "regions", {"tool": "db_query__crm", "duration_ms": 7,
"query_artifact": self.entries["regions"]["artifact"],
"preview": json.dumps(REGIONS["rows"], ensure_ascii=False)})
add("node.finished", "regions", {"duration_ms": 8, "attempt": 1, "evidence": [self.entries["regions"]]})
add("node.finished", "card", {"duration_ms": 3, "attempt": 1, "preview": {"kind": "metric_set"},
"evidence": [self.entries["card"]]})
for node_id, doc_id in reports.items():
add("llm.end", node_id, {"model": "mock", "output": "(写作草稿)"})
add("report.checked", node_id, {"doc_artifact": doc_id, "ok": True})
add("node.finished", node_id, {"duration_ms": 20, "attempt": 1,
"preview": {"text": "(报告正文)", "doc_artifact": doc_id}})
if ending == "finished":
for tier in tiers:
add("issuance", "out", {"tier": tier, "mode": "citations"})
add("node.finished", "out", {"duration_ms": 2, "attempt": 1, "preview": {"周报": "(报告正文)"}})
add("run.finished", None, {"output": {"周报": "(报告正文)"}})
elif ending == "continued":
# 先失败,再从成果节点继续运行到成功(runner.continue_failed 记一条 run.resumed)
add("node.failed", "out", {"error": "出口复核没有通过"})
add("run.failed", None, {"error": "出口复核没有通过"})
add("run.resumed", None, {"from": ["out"], "message": "从「成果」继续运行,保留前面 6 个节点的结果",
"actor": "值班分析员"})
for tier in tiers:
add("issuance", "out", {"tier": tier, "mode": "citations"})
add("node.finished", "out", {"duration_ms": 2, "attempt": 1, "preview": {"周报": "(报告正文)"}})
add("run.finished", None, {"output": {"周报": "(报告正文)"}})
else:
add("node.failed", "out", {"error": "出口复核没有通过"})
add("run.failed", None, {"error": "出口复核没有通过"})
bound = len(rows)
add("evidence.judged", next(iter(reports), None),
{"verdicts": {"u1": {"verdict": "supported"}}, "model": "judge-x", "post_seal": True})
return rows, bound
def build(sample: Sample, events: list[tuple[int, str, str | None, Any]], bound: int, *,
run: dict[str, Any] | None = None, legacy: bool = False, drop: tuple[str, ...] = (),
source_masks: dict[str, list[str]] | None = None) -> dict[str, Any]:
"""真实封存哈希的页面数据(bundle-data)。drop:导出时服务端已没有文件的工件;source_masks:数据源现在设的遮罩。"""
sealed = [e for e in events if e[0] <= bound]
files = {k: v for k, v in sample.files.items() if k not in drop}
return build_bundle(run or RUN, events, manifest_seq=bound, manifest_hash=manifest_hash(sealed), read=files.get,
exported_at=EXPORTED_AT, legacy=legacy, graph=GRAPH, source_masks=source_masks or {})
def formats_bundle() -> dict[str, Any]:
"""经营月报(formats)的页面数据:每种显示格式各引用一处。"""
f = Sample(card=FORMATS_CARD, table=LEDGER)
events, bound = f.events({"write": f.doc(FORMATS, "write")})
return build(f, events, bound)
def bundles() -> dict[str, dict[str, Any]]:
"""{页面名: 页面数据}"""
out: dict[str, dict[str, Any]] = {}
s = Sample()
weekly = s.doc(WEEKLY, "write")
events, bound = s.events({"write": weekly})
out["base"] = build(s, events, bound)
out["missing-doc"] = build(s, events, bound, drop=(weekly,))
out["missing-card"] = build(s, events, bound, drop=(s.card_id,))
# 早期运行:运行结束事件落库之前封存,范围到它前一条为止(runner.verify_manifest 的老口径)
finished = next(seq for seq, etype, _, _ in events if etype == "run.finished")
out["legacy"] = build(s, events, finished - 1, legacy=True)
monthly = s.doc(MONTHLY, "monthly")
events, bound = s.events({"write": weekly, "monthly": monthly})
out["two-reports"] = build(s, events, bound)
events, bound = s.events({"write": weekly}, ending="failed")
out["failed"] = build(s, events, bound, run={**RUN, "status": "failed"})
# 早期运行先失败、经继续运行跑到成功:封存范围同样按结束事件推算,到成功结束的前一条为止
events, _ = s.events({"write": weekly}, ending="continued")
finished = next(seq for seq, etype, _, _ in events if etype == "run.finished")
out["continued"] = build(s, events, finished - 1, legacy=True)
events, bound = s.events({"write": weekly}, tiers=("formal", "degraded"))
out["degraded"] = build(s, events, bound)
stray = s.doc(WEEKLY, "write", run_id="another-run-0000")
events, bound = s.events({"write": stray})
out["stray"] = build(s, events, bound)
n = Sample(nan=True)
events, bound = n.events({"write": n.doc(WEEKLY, "write")}, nan=True)
out["nan"] = build(n, events, bound)
m = Sample(masked=True)
events, bound = m.events({"write": m.doc(WEEKLY, "write")})
out["masked-card"] = build(m, events, bound)
r = Sample(regions=True)
events, bound = r.events({"write": r.doc(MEMBERS, "write")})
out["masked-source"] = build(r, events, bound, source_masks={"crm": ["phone"]})
out["formats"] = formats_bundle()
return out
def pages() -> dict[str, str]:
"""{页面名: HTML}"""
return {name: render_html(bundle) for name, bundle in bundles().items()}
def main(argv: list[str]) -> int:
if len(argv) != 2:
print("用法:python -m tests.bundle_samples <目录>", file=sys.stderr)
return 2
target = Path(argv[1])
target.mkdir(parents=True, exist_ok=True)
for name, page in pages().items():
(target / f"{name}.html").write_text(page, encoding="utf-8")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))