From cb2fdd860f3082f472936fbf0f68e8d4e242aa02 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 13:49:09 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=81=E6=8D=AE=E6=8E=A5=E5=85=A5=E6=8C=89?= =?UTF-8?q?=20W1-a=E3=80=81W1-c=20=E7=9A=84=E5=AE=9E=E9=99=85=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=AF=B9=E9=BD=90=EF=BC=9A=E5=A4=8D=E7=AE=97=E5=B0=B1?= =?UTF-8?q?=E6=98=AF=E7=9C=9F=E7=9A=84=20app.analysis.recompute=E2=80=94?= =?UTF-8?q?=E2=80=94load=20=E4=BA=A4=E6=9F=A5=E8=AF=A2=E5=BF=AB=E7=85=A7?= =?UTF-8?q?=E5=B7=A5=E4=BB=B6=E7=9A=84=E5=8E=9F=E5=86=85=E5=AE=B9=E3=80=81?= =?UTF-8?q?load=5Fresult=20=E4=BA=A4=E4=B8=8A=E4=B8=80=E5=B1=82=E7=BB=93?= =?UTF-8?q?=E8=AE=BA=EF=BC=88=E5=8F=AA=E8=AE=A4=E5=B0=81=E5=AD=98=E8=8C=83?= =?UTF-8?q?=E5=9B=B4=E5=86=85=E4=BA=A4=E5=9B=9E=E8=BF=87=E7=9A=84=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E3=80=81=E5=8F=B0=E8=B4=A6=E8=AE=B0=E8=BF=87=E7=9A=84?= =?UTF-8?q?=E7=BB=93=E8=AE=BA=EF=BC=8C=E5=8F=96=E4=B8=8D=E5=9B=9E=E7=85=A7?= =?UTF-8?q?=E5=AE=9E=E8=AF=B4=E5=8F=96=E4=B8=8D=E5=9B=9E=E5=93=AA=E4=B8=80?= =?UTF-8?q?=E4=BB=BD=EF=BC=89=EF=BC=8C=E6=96=B9=E6=B3=95=E5=8D=87=E6=B2=A1?= =?UTF-8?q?=E5=8D=87=E7=89=88=E6=8C=89=E7=BB=93=E8=AE=BA=E9=87=8C=E7=9A=84?= =?UTF-8?q?=20method=20=E5=92=8C=E7=8E=B0=E5=9C=A8=E7=9A=84=20contract.MET?= =?UTF-8?q?HODS=20=E6=AF=94=EF=BC=88=E5=8D=87=E4=BA=86=E7=89=88=E3=80=81?= =?UTF-8?q?=E5=A4=8D=E7=AE=97=E6=88=90=E7=AB=8B=E7=9A=84=E5=86=99=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E5=90=8C=E4=B8=8D=E5=90=8C=EF=BC=8C=E7=8E=B0=E5=9C=A8?= =?UTF-8?q?=E7=9A=84=E6=96=B9=E6=B3=95=E5=A4=8D=E7=AE=97=E4=B8=8D=E4=BA=86?= =?UTF-8?q?=E7=9A=84=E5=86=99=E6=98=8E=E5=8E=9F=E5=9B=A0=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E5=8E=BB=E6=8E=89=E4=B8=BA=E7=AE=97=E5=AD=90=E6=B2=A1=E8=90=BD?= =?UTF-8?q?=E5=9C=B0=E5=87=86=E5=A4=87=E7=9A=84=E9=82=A3=E4=B8=80=E5=A5=97?= =?UTF-8?q?=EF=BC=9B=E5=85=B3=E9=94=AE=E6=95=B0=E7=9A=84=E4=BB=A3=E5=85=A5?= =?UTF-8?q?=E5=BC=8F=E6=8C=89=E7=9C=9F=E7=9A=84=E5=85=B7=E5=90=8D=E7=9A=84?= =?UTF-8?q?=E6=95=B0=E5=86=99=EF=BC=88=E4=B9=98=E7=A7=AF=E6=8B=86=E8=A7=A3?= =?UTF-8?q?=E8=A1=A5=E4=B8=8A=EF=BC=8C=E6=B3=A2=E5=8A=A8=E5=9F=BA=E7=BA=BF?= =?UTF-8?q?=E6=8C=89=E5=8F=98=E5=8C=96=E7=8E=87=E6=AF=94=E7=9A=84=E6=8C=89?= =?UTF-8?q?=E7=99=BE=E5=88=86=E6=AF=94=E5=86=99=EF=BC=8C=E8=B4=9F=E6=95=B0?= =?UTF-8?q?=E5=8A=A0=E6=8B=AC=E5=8F=B7=EF=BC=89=E3=80=82=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E5=8D=A1=E6=8C=87=E6=A0=87=E4=B8=8A=E7=9A=84=E5=8F=AF=E6=AF=94?= =?UTF-8?q?=E6=80=A7=E7=BB=93=E8=AE=BA=EF=BC=88=E4=B8=8D=E8=BF=9B=E5=8F=B0?= =?UTF-8?q?=E8=B4=A6=E3=80=81=E5=8F=AA=E4=BB=8E=E5=8D=A1=E4=B8=8A=E5=BC=95?= =?UTF-8?q?=E7=94=A8=EF=BC=89=E7=82=B9=E5=BE=97=E5=BC=80=EF=BC=9A=E8=AF=81?= =?UTF-8?q?=E6=8D=AE=E9=9D=A2=E6=9D=BF=E7=9A=84=E6=8C=87=E6=A0=87=E6=AD=A5?= =?UTF-8?q?=E9=AA=A4=E5=B8=A6=E5=AE=83=E6=95=B4=E6=9D=A1=E5=87=BA=E5=A4=84?= =?UTF-8?q?=E9=93=BE=EF=BC=88=E5=88=A4=E5=AE=9A=20=E2=86=92=20=E6=96=B9?= =?UTF-8?q?=E6=B3=95=20=E2=86=92=20=E8=AE=A1=E7=AE=97=20=E2=86=92=20?= =?UTF-8?q?=E8=BE=93=E5=85=A5=20=E2=86=92=20=E5=A4=8D=E7=AE=97=20=E2=86=92?= =?UTF-8?q?=20=E5=B0=81=E5=AD=98=EF=BC=8C=E5=B0=81=E5=AD=98=E6=8C=89?= =?UTF-8?q?=E5=8D=A1=E4=B8=8A=E8=AE=B0=E7=9A=84=20sha256=20=E8=AE=A4?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E6=8A=A5=E5=91=8A=E5=AF=BC=E5=87=BA=E6=94=B6?= =?UTF-8?q?=E5=9C=A8=E4=B8=80=E4=B8=AA=E5=8F=AF=E5=B1=95=E5=BC=80=E7=9A=84?= =?UTF-8?q?=E5=9D=97=E9=87=8C=EF=BC=8C=E5=8F=AF=E9=AA=8C=E8=AF=81=E6=8A=A5?= =?UTF-8?q?=E5=91=8A=E9=9A=8F=E9=99=84=E5=B8=A6=E7=9A=84=E5=8D=A1=E4=B8=80?= =?UTF-8?q?=E8=B5=B7=E9=99=84=E4=B8=8A=E3=80=81=E2=91=A3=20=E6=8C=89?= =?UTF-8?q?=E5=B7=B2=E6=A0=B8=E5=AF=B9=E7=9A=84=E5=8D=A1=E8=AE=A4=E5=AE=83?= =?UTF-8?q?=E3=80=81=E6=95=B0=E5=AD=97=E8=AF=A6=E6=83=85=E9=87=8C=E5=B1=95?= =?UTF-8?q?=E7=A4=BA=E5=90=84=E9=A1=B9=E6=A3=80=E6=9F=A5=E5=92=8C=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E7=AB=AF=E5=A4=8D=E7=AE=97=EF=BC=9B=E8=A3=81=E5=88=A4?= =?UTF-8?q?=E7=9A=84=E6=8C=87=E6=A0=87=E6=91=98=E5=BD=95=E3=80=81=E5=86=99?= =?UTF-8?q?=E4=BD=9C=E7=9B=AE=E5=BD=95=E7=85=A7=E5=8D=A1=E4=B8=8A=E8=AE=B0?= =?UTF-8?q?=E7=9A=84=E5=88=A4=E5=AE=9A=E8=AF=AD=E5=86=99=E3=80=82=E7=8E=B0?= =?UTF-8?q?=E6=9F=A5=E7=BB=93=E6=9E=9C=E5=88=86=E6=B8=85=E6=8C=89=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=BD=93=E5=A4=A9=E7=8E=B0=E7=AE=97=E7=9A=84=E7=95=8C?= =?UTF-8?q?=E3=80=81=E4=B8=80=E5=88=97=E4=B8=80=E4=B8=AA=E5=80=BC=E9=83=BD?= =?UTF-8?q?=E6=B2=A1=E6=9C=89=E3=80=81=E8=A1=A8=E7=9A=84=E8=8C=83=E5=9B=B4?= =?UTF-8?q?=EF=BC=9B=E5=8F=AF=E9=AA=8C=E8=AF=81=E6=8A=A5=E5=91=8A=E9=87=8C?= =?UTF-8?q?=E5=88=86=E6=9E=90=E5=B7=A5=E5=85=B7=E7=9A=84=20tool.end?= =?UTF-8?q?=E3=80=81=E5=88=86=E6=9E=90=E5=8D=A1=E7=9A=84=20analysis.end=20?= =?UTF-8?q?=E5=92=8C=E7=BB=93=E8=AE=BA=E5=90=8C=E4=B8=80=E4=B8=AA=E8=AE=A4?= =?UTF-8?q?=E6=B3=95=E6=89=8D=E5=B8=A6=E5=86=85=E5=AE=B9=EF=BC=8C=E8=BE=93?= =?UTF-8?q?=E5=85=A5=E7=9A=84=E8=A7=92=E8=89=B2=E5=86=99=E5=8F=AB=E6=B3=95?= =?UTF-8?q?=E3=80=82=E6=B5=8B=E8=AF=95=E6=94=B9=E7=94=A8=E7=9C=9F=E7=AE=97?= =?UTF-8?q?=E7=9A=84=E7=BB=93=E8=AE=BA=EF=BC=88run=20=E4=BA=A4=E7=9A=84?= =?UTF-8?q?=E8=BE=93=E5=85=A5=E7=85=A7=E5=88=86=E6=9E=90=E5=85=A5=E5=8F=A3?= =?UTF-8?q?=E7=9A=84=E5=86=99=E6=B3=95=EF=BC=8C=E5=A4=8D=E7=AE=97=E4=BA=A4?= =?UTF-8?q?=E5=B7=A5=E4=BB=B6=E5=8E=9F=E5=86=85=E5=AE=B9=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E4=B8=8B=E9=92=BB=E3=80=81=E6=80=BB=E6=95=B0=E9=97=AD=E5=90=88?= =?UTF-8?q?=E3=80=81=E6=96=B9=E6=B3=95=E5=8D=87=E7=89=88=E3=80=81=E5=8F=AF?= =?UTF-8?q?=E6=AF=94=E6=80=A7=E7=AB=AF=E5=88=B0=E7=AB=AF=E9=83=BD=E8=BF=87?= =?UTF-8?q?=E7=9C=9F=E7=9A=84=E5=A4=8D=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/evidence.py | 104 +++++---- backend/app/bundle/build.py | 43 +++- backend/app/bundle/verifier.js | 56 ++++- backend/app/engine/evidence.py | 168 +++++++++------ backend/app/engine/judge.py | 4 + backend/app/report_export/runtime.js | 41 +++- backend/tests/test_analysis_bundle.py | 127 +++++++++-- backend/tests/test_analysis_evidence.py | 223 +++++++++++++++----- backend/tests/test_analysis_evidence_api.py | 209 ++++++++++++++---- backend/tests/test_analysis_judge_ledger.py | 18 ++ backend/tests/test_analysis_pages.py | 62 +++++- 11 files changed, 811 insertions(+), 244 deletions(-) diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 590d14a9..bc76da20 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -113,7 +113,6 @@ from app.engine.evidence import ( RenderError, analysis_card, analysis_formulas, - analysis_input_snapshot, analysis_recheck, build_catalog, closest_entities, @@ -987,6 +986,11 @@ HYPOTHESIS_NOTE = "Agent 写的假设：说明这次在验证什么，不是分 UNSEALED_ANALYSIS = "这份分析结论不属于封存范围内的任何事件（封存前的台账里没有它），不能作为证据展示" TAMPERED_ANALYSIS = "分析结论与哈希不一致，疑似被修改，不展示其中的判定和数" MISSING_ANALYSIS = "无法读取分析结论" +#: 口径卡指标上的可比性结论不进台账（00-总方案 §3.4）：它记在封存过的口径卡指标集里，卡上记着它的 sha256，属于封存链的 +#: 延伸（同推演认 value_artifact）。指标那一步没封存的，这份结论也不算封存 +COMPARABILITY_CHAINED = ("这份可比性结论不进台账：口径卡执行时自动做的，记在封存过的口径卡指标集里（卡上记着它的 sha256），" + "随卡一起封存") +UNSEALED_COMPARABILITY = "这份可比性结论记在口径卡上，可这张卡的指标集不在封存范围内，不能作为证据展示" #: 每个算子用到的阈值（细案 01 §10「用在」一栏）：方法一步只列这几个，判定标准照它们写 _OP_SETTINGS: dict[str, tuple[str, ...]] = { "decompose": ("main_share", "concentration", "offset_ratio", "flat_pct", "min_denominator", "max_drill_depth", @@ -1051,7 +1055,8 @@ def _analysis_rendered(cite: dict[str, Any], result: dict[str, Any]) -> str | No async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[str, Any], doc: dict[str, Any], - sealed: _Sealed, masks: _Masks) -> list[dict[str, Any]]: + sealed: _Sealed, masks: _Masks, *, chained: bool | None = None, + queries: bool = True) -> list[dict[str, Any]]: """分析结论的出处链（00-总方案 §3.3）：判定 → 方法 → 计算 → 输入 → 复算 → 封存。 - 判定（analysis）：判定语、判定标准、判定码和点名的段、附带说明、算不了的原因，这一处引用的是哪一项（判定语、具名标量、 @@ -1064,19 +1069,24 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict - 复算（analysis_recompute）：按 sha256 取回封存的输入、用同一个函数复算，和封存的逐字段比（evidence.analysis_recheck）； 方法升过版、复算不了都照实写 - 封存（analysis_seal）：这件工件是不是封存范围内的台账记过的、哈希复验通过 - 结论取不回、哈希不符的只给判定和封存两步，照实说为什么。""" + 结论取不回、哈希不符的只给判定和封存两步，照实说为什么。 + + 口径卡指标上的可比性结论（_comparability_view）不进台账：chained 是那个指标的一步封存了没有（卡上记着这份结论的 sha256， + 属于封存链的延伸），封存一步照它认；谁做的写口径卡（origin metric）。queries 为假时不接输入快照的查询步骤（可比性的 + 两份输入就是指标输入的那两份，指标链里已经有了）。""" catalog = doc.get("catalog") or {} alias = str(cite.get("alias") or entry.get("alias") or "") artifact = str(entry.get("artifact") or "") result, hash_ok = _analysis_of(artifact) info = sealed.analyses.get(artifact) - in_ledger = sealed.trusted and (str(entry.get("node_id") or ""), artifact) in sealed.ledger + in_ledger = sealed.trusted and (str(entry.get("node_id") or ""), artifact) in sealed.ledger \ + if chained is None else sealed.trusted and chained eid = make_eid(LEDGER_KIND, artifact or None, {}) rendered = _analysis_rendered(cite, result) if result is not None else None judged: dict[str, Any] = { "step": "analysis", "alias": alias, "op": entry.get("op"), "mode": entry.get("mode"), "method_label": method_label(entry.get("op"), entry.get("mode")), - "origin": "card" if analysis_card(entry) else "agent", + "origin": "metric" if chained is not None else "card" if analysis_card(entry) else "agent", **{k: entry[k] for k in ("item", "call_id") if entry.get(k)}, "node_id": entry.get("node_id"), "label": _node_label(sealed, entry.get("node_id")) or None, "artifact": artifact or None, "eid": entry.get("eid"), "eid_ok": entry.get("eid") == eid, @@ -1086,7 +1096,10 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict seal = {"step": "analysis_seal", "alias": alias, "node_id": entry.get("node_id"), "artifact": artifact or None, "ledger": info is not None, "hash_ok": hash_ok, "eid_ok": judged["eid_ok"], "sealed": in_ledger and hash_ok is True and judged["eid_ok"]} - if info is None: + if chained is not None: + # 封存的路子是口径卡的指标集（卡上记着这份结论的 sha256），不是台账 + seal.update(via="metric_set", note=COMPARABILITY_CHAINED if chained else UNSEALED_COMPARABILITY) + elif info is None: seal["note"] = UNSEALED_ANALYSIS if result is None: judged["note"] = TAMPERED_ANALYSIS if hash_ok is False else MISSING_ANALYSIS @@ -1106,12 +1119,12 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict for k in _OP_SETTINGS.get(str(result.get("op")), ()) if k in settings], "probes": [p for p in result.get("probes") or [] if isinstance(p, dict)], } - inputs, queries = await _analysis_inputs(alias, result, catalog, doc, sealed, masks) - hidden = {str(c).lower() for q in queries for c in q.get("masked") or []} + inputs, steps = await _analysis_inputs(alias, result, catalog, doc, sealed, masks, queries=queries) + hidden = {str(c).lower() for q in steps for c in q.get("masked") or []} calc = _analysis_calc(alias, result, judged["cited"], hidden) again = await _analysis_recheck(result, sealed) recheck = {"step": "analysis_recompute", "alias": alias, **again} - return [judged, method, calc, *inputs, *queries, recheck, seal] + return [judged, method, calc, *inputs, *steps, recheck, seal] def _analysis_calc(alias: str, result: dict[str, Any], cited: dict[str, Any], hidden: set[str]) -> dict[str, Any]: @@ -1154,10 +1167,11 @@ def _analysis_calc(alias: str, result: dict[str, Any], cited: dict[str, Any], hi async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str, Any], doc: dict[str, Any], - sealed: _Sealed, masks: _Masks) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + sealed: _Sealed, masks: _Masks, *, queries: bool = True + ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """输入：每份一步（analysis_input），再给每份输入快照一个查询步骤（_query_steps，高亮用到的列；整份用到的给前 - MAX_WINDOW_ROWS 行）。headline 的那一格另核对和结论里的总数闭不闭合（照实写判定是不是 unclosed，不在这里重算容差）； - 下钻的上一层落到报告目录的哪个 A。""" + MAX_WINDOW_ROWS 行；queries 为假时不给）。headline 的那一格另核对和结论里的总数闭不闭合（照实写判定是不是 + unclosed，不在这里重算容差）；下钻的上一层落到报告目录的哪个 A。""" values = result.get("values") if isinstance(result.get("values"), dict) else {} verdict = result.get("verdict") if isinstance(result.get("verdict"), dict) else {} unclosed = verdict.get("code") == "unclosed" @@ -1198,37 +1212,35 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str rows = list(range(MAX_WINDOW_ROWS)) if picked == "all" else \ [r for r in picked if isinstance(r, int)][:MAX_WINDOW_ROWS] if isinstance(picked, list) else [] wanted.extend((r, column) for r in rows) - queries: list[dict[str, Any]] = [] - for artifact, picked in cells.items(): + found: list[dict[str, Any]] = [] + for artifact, picked in cells.items() if queries else (): snap, _ = masks.fetch(artifact) total = len(snap["rows"]) if isinstance(snap, dict) and isinstance(snap.get("rows"), list) else 0 picked = list(dict.fromkeys((r, c) for r, c in picked if not isinstance(r, int) or r < total)) for sub in await _query_steps(artifact, picked, doc, sealed, masks): - queries.append({**sub, "input_of": alias}) - return steps, queries + found.append({**sub, "input_of": alias}) + return steps, found async def _analysis_recheck(result: dict[str, Any], sealed: _Sealed) -> dict[str, Any]: - """复算：按 sha256 取回封存范围内的输入快照（evidence.analysis_input_snapshot）和下钻的上一层结论，在线程里复算、比较 - （evidence.analysis_recheck）。不在封存范围内的输入一律当取不回来。""" - snapshots: dict[str, Any] = {} - parents: dict[str, Any] = {} - for item in result.get("inputs") or []: - artifact = str((item or {}).get("artifact") or "") if isinstance(item, dict) else "" - if not artifact or artifact in snapshots or artifact in parents: - continue - if item.get("role") == "parent": - parents[artifact] = _analysis_of(artifact)[0] if artifact in sealed.analyses else None - continue - info = sealed.queries.get(artifact) - content, ok = _fetch(artifact) if info is not None else (None, None) - snapshots[artifact] = analysis_input_snapshot(content, artifact, ref=item.get("ref"), - node_id=(info or {}).get("node_id")) if ok is True else None + """复算（evidence.analysis_recheck）：输入快照交工件原内容，下钻的上一层交那份结论，都按 sha256 取回、复验哈希，在线程里 + 复算、比较。只认封存范围内的——输入快照要是这次运行交回过的查询（sealed.queries），上一层要是台账记过的分析结论； + 不在范围内、哈希对不上的一律当取不回（复算照实说取不回哪一份）。""" + snapshots = {str(i["artifact"]) for i in result.get("inputs") or [] + if isinstance(i, dict) and i.get("artifact") and i.get("role") != "parent"} + parents = {str(i["artifact"]) for i in result.get("inputs") or [] + if isinstance(i, dict) and i.get("artifact") and i.get("role") == "parent"} - def run() -> dict[str, Any]: - return analysis_recheck(result, snapshots.get, load_result=parents.get) + def load(artifact: str) -> Any: + if artifact not in snapshots or artifact not in sealed.queries: + return None + content, ok = _fetch(artifact) + return content if ok is True else None - return await asyncio.to_thread(run) + def load_result(artifact: str) -> Any: + return _analysis_of(artifact)[0] if artifact in parents and artifact in sealed.analyses else None + + return await asyncio.to_thread(analysis_recheck, result, load, load_result=load_result) def _missing_input(cite: dict[str, Any], entry: dict[str, Any]) -> bool: @@ -1288,7 +1300,9 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s # 跨两份同形查询算的环比、同比：口径卡执行时自动做的可比性检查（analysis.contract.MetricComparability），点得开那份 # 可比性结论（artifact）。没做过检查的指标没有这个键 if isinstance((metric or {}).get("comparability"), dict): - step["comparability"] = _comparability_view(metric["comparability"], sealed, chained=step["sealed"]) + step["comparability"] = await _comparability_view( + metric["comparability"], sealed, masks, chained=step["sealed"], node_id=str(entry.get("node_id") or ""), + doc=doc) card_node = str(entry.get("node_id") or "") if metric is not None and metric.get("inputs"): # 输入追没追到原始数据（engine/lineage.py）：和「有出处」（确定性复算一致）是两件事，面板分开说。追不到的照实说 @@ -1329,23 +1343,33 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s return [step, *inputs, *upstream, *queries] -def _comparability_view(comparability: dict[str, Any], sealed: _Sealed, *, chained: bool) -> dict[str, Any]: +async def _comparability_view(comparability: dict[str, Any], sealed: _Sealed, masks: _Masks, *, chained: bool, + node_id: str, doc: dict[str, Any]) -> dict[str, Any]: """指标步骤里的可比性：判定（comparable / conditional / incomparable）、判定语、原因（一句一个），连同那份可比性结论的 - 工件：取回、复验哈希，带上它的判定标准、方法和现查的表范围（点得开）。sealed：哈希复验通过，而且它记在封存过的口径卡 - 指标集里（chained：指标步骤本身封存了；内容寻址，卡上记着它的 sha256）或者封存范围内的台账里。""" + 工件：取回、复验哈希，带上它的判定标准、方法和现查结果，再给一条点得开的出处链（steps：判定 → 方法 → 计算 → 输入 → + 复算 → 封存，同分析结论的引用，_analysis_chain；两份输入快照的查询步骤指标链里已经有了，不再给）。 + + 这份结论不进台账，只从口径卡的指标集里引用：sealed 认的是哈希复验通过，而且记着它的指标集封存了（chained：指标步骤 + 本身封存了；内容寻址，卡上记着它的 sha256——同推演认 value_artifact），或者它恰好也在封存范围内的台账里。""" artifact = str(comparability.get("artifact") or "") result, hash_ok = _analysis_of(artifact) + scoped = sealed.trusted and bool(artifact) and (chained or artifact in sealed.analyses) view: dict[str, Any] = { "verdict": comparability.get("verdict"), "phrase": comparability.get("phrase"), "reasons": [str(r) for r in comparability.get("reasons") or [] if r], "artifact": artifact or None, - "hash_ok": hash_ok, - "sealed": sealed.trusted and hash_ok is True and (chained or artifact in sealed.analyses), + "hash_ok": hash_ok, "sealed": scoped and hash_ok is True, } if result is not None: view.update(method=result.get("method"), criterion=result.get("criterion"), caveats=[{k: c.get(k) for k in ("code", "text")} for c in result.get("caveats") or [] if isinstance(c, dict)], probes=[p for p in result.get("probes") or [] if isinstance(p, dict)]) + if artifact: + entry = {"kind": LEDGER_KIND, "artifact": artifact, "node_id": node_id, "op": "comparable", "mode": None, + "eid": make_eid(LEDGER_KIND, artifact, {})} + cite = {"kind": LEDGER_KIND, "alias": None, "locator": {}, "status": "resolved"} + view["steps"] = await _analysis_chain({"text": str(comparability.get("phrase") or "")}, cite, entry, doc, + sealed, masks, chained=scoped, queries=False) return view diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index 3ebf22cd..949d3053 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -27,6 +27,9 @@ 下钻的上一层结论；任一份输入快照只给 id（证明不了没碰遮罩列）的，结论本身也只给 id——结果表里的维度值取自输入快照， 这份文件宁可多挡。浏览器只核哈希、核判定语和数是不是结论里写的那些，不在页面里重算统计：结论随文件附带时，服务端在 导出时按封存的输入复算一遍（engine/evidence.analysis_recheck），结果记在这份工件的条目上（recheck），页面照它展示。 + 口径卡指标上的可比性结论（口径卡执行时自动做的，不进台账、只从卡上引用）随附带的卡一起附上，规则同上，引用链（④） + 按「已核对的口径卡记着的可比性」认它。分析工具的 tool.end（预览里是结论的前一截）、分析卡一项的 analysis.end（带着 + 判定语）和结论本身同一个认法：结论随文件附带得了才带规范字符串。 - 报告文档带着系统写的「分析范围」（analysis_scope）的，页面在正文后面照它渲染，编号点得开那条结论。 - 封存之后追加的事件（按需裁判、审批质询这类批注）单列：类型、序号、内容摘要，不进清单。 @@ -63,7 +66,7 @@ from urllib.parse import quote from app.analysis.contract import ARTIFACT_KIND, LEDGER_KIND, NODE_TYPE as ANALYSIS_NODE, is_result from app.core.artifact_store import canonical_json, content_hash from app.core.events import EventType -from app.engine.evidence import analysis_input_snapshot, analysis_recheck, iter_units, query_entry_fields +from app.engine.evidence import analysis_recheck, iter_units, query_entry_fields from app.engine.lineage import in_scope, scoped_id from app.engine.masking import OPAQUE_SQL, SYSTEM_SCHEMAS, indirect_reads from app.engine.merge_query import MERGE_SOURCE @@ -93,6 +96,8 @@ _SHOWN_EVENTS = frozenset({EventType.RUN_STARTED, EventType.REPORT_CHECKED, Even _CITED_KINDS = {"metric": "metric_set", "query": "query", LEDGER_KIND: ARTIFACT_KIND} #: 下钻的上一层结论最多跟几层（同分析设置里下钻层数的上限） _PARENT_DEPTH = 6 +#: 分析卡里一项做完的事件（core/events.EventType.ANALYSIS_END）：带着那一项的判定、判定语和结论工件（analysis_artifact） +_ANALYSIS_END = "analysis.end" #: 没挂在工作流上的运行（画布、问数据当场搭的）：老数据叫「临时图」，新的叫「未保存的工作流」，也可能是空的。 #: 和前端 lib/terms.ts 的 runName 同一个规则 @@ -1066,6 +1071,17 @@ def _analysis_inputs(result: Any) -> list[tuple[str, str]]: return list(out.items()) +def _card_comparability(card: Any) -> list[str]: + """口径卡指标上记着的可比性结论（contract.MetricComparability 的 artifact），按先后、去重。这份结论不进台账，只从卡上 + 引用：卡随文件附带时一并附上，引用链（④）按「已核对的口径卡记着的可比性」认它。""" + out: list[str] = [] + for metric in (card.get("metrics") or []) if isinstance(card, dict) else []: + found = metric.get("comparability") if isinstance(metric, dict) else None + if isinstance(found, dict) and isinstance(found.get("artifact"), str) and found["artifact"]: + out.append(found["artifact"]) + return list(dict.fromkeys(out)) + + def _card_input_queries(card: Any) -> list[tuple[str, str | None]]: """口径卡记着的输入引用的查询快照：(工件 id, 输入来自的节点)，按先后、按工件去重。 @@ -1169,7 +1185,7 @@ def _recheck(result: dict[str, Any], store: _Store) -> dict[str, Any]: """导出时服务端按封存的输入把这份分析结论复算一遍（engine/evidence.analysis_recheck，和证据面板同一个函数）：输入快照、 上一层结论只认这份文件里哈希对得上的原字节。页面照这个结果写「导出时服务端的复算」，不在浏览器里重算统计。""" def load(artifact: str) -> Any: - return analysis_input_snapshot(store.content(artifact), artifact) if store.intact(artifact) else None + return store.content(artifact) if store.intact(artifact) else None def load_result(artifact: str) -> Any: content = store.content(artifact) if store.intact(artifact) else None @@ -1209,10 +1225,24 @@ def _shows_content(etype: str, node_id: str | None, data: Any, node_types: Mappi if etype == EventType.TOOL_END: # 查询类：把查询快照的 id 和运行记录连起来。预览里是查询结果的前一截，数据源带遮罩的只给哈希 artifact = data.get("query_artifact") if isinstance(data, dict) else None - return isinstance(artifact, str) and bool(artifact) and not store.masked(artifact) + if isinstance(artifact, str) and artifact: + return not store.masked(artifact) + # 分析工具：预览里是结论的前一截（判定语、具名的数、结果表），和结论本身同一个认法 + found = data.get("analysis_artifact") if isinstance(data, dict) else None + return isinstance(found, str) and bool(found) and _result_shown(found, store) + if etype == _ANALYSIS_END: + # 分析卡的一项：带着判定、判定语，结论随文件附带得了（输入没碰遮罩列）才带 + found = data.get("analysis_artifact") if isinstance(data, dict) else None + return isinstance(found, str) and bool(found) and _result_shown(found, store) return False +def _result_shown(artifact: str, store: _Store) -> bool: + """一份分析结论的内容能不能出现在页面上（事件里带着它的判定语、预览）：文件在、认得出是分析结论、输入都没碰遮罩列 + （_Store.analysis_masked）。和这份结论随文件附带时同一个认法。""" + return store.raw(artifact) is not None and not store.analysis_masked(artifact) + + def _card_shown(data: Any, store: _Store, masked_run: bool, events: Sequence[tuple[int, str, str | None, Any]], forced: frozenset[str] = frozenset()) -> bool: """口径卡的节点完成事件带不带规范字符串。预览里逐条写着「指标（名称）= 值」（metrics 节点的 text），赋了变量的另有 @@ -1248,7 +1278,7 @@ def _analysis_shown(data: Any, store: _Store) -> bool: results = [str(e["artifact"]) for e in data.get("evidence") or [] if isinstance(e, dict) and e.get("kind") == LEDGER_KIND and isinstance(e.get("artifact"), str) and e["artifact"]] - return bool(results) and all(store.raw(a) is not None and not store.analysis_masked(a) for a in results) + return bool(results) and all(_result_shown(a, store) for a in results) def _masked_cards(doc: dict[str, Any]) -> set[str]: @@ -1420,6 +1450,11 @@ def build_bundle( if kind == "metric_set" and item is not None and "b64" in item and store.intact(artifact): for query, query_owner in _card_input_queries(store.content(artifact)): artifacts.add(query, "query", query_owner) + # 指标上的可比性结论（口径卡执行时自动做的，不进台账、只从卡上引用）同样附上：它的输入快照、服务端的复算 + # 跟着它（_attach_analysis），输入带遮罩的照旧只给 id + for found in _card_comparability(store.content(artifact)): + _attach_analysis(artifacts, store, artifacts.add(found, ARTIFACT_KIND, owner), found, owner, + cited_queries) masked_run = _masked_run(sealed, store, cited_queries, source_masks or {}) shown_events = [] diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index ad50ca01..0ba72100 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -22,6 +22,7 @@ * 分析结论（判定语 [[a:A3]]、具名的数 [[a:A3.键]]、结果表的格 [[v:A3.r0.列]]）在 ③ ④ 里和查询快照一样核哈希、核引用链 * （它的输入快照按「已核对的分析结论的输入」认）；⑤ 里核正文显示的判定语、数是不是结论里写的那几个字（同服务端的渲染 * 规则）。本页不重算统计：结论随文件附带时，出处面板展示导出时服务端按封存的输入复算的结论（工件条目上的 recheck）。 + * 口径卡指标上的可比性结论不进台账、只从卡上引用：④ 按「已核对的口径卡记着的可比性」认它，指标的数字详情里展示它。 * 报告文档带着系统写的「分析范围」（analysis_scope）的，正文后面照它渲染，编号点得开那条结论。 * 每项给出通过 / 不通过 / 无法核对和原因。服务端导出时不会出现的状态（被引用的工件不在清单里、检索快照这类工件标着 * 遮罩）一律算不通过：只可能是文件被改过。报告文档的证据目录里写着取自遮罩列的指标的值时，服务端只给它的哈希 @@ -67,7 +68,8 @@ 'node.failed': '节点失败', 'node.skipped': '节点跳过', 'edge.taken': '分支走向', 'llm.start': '模型调用开始', 'llm.end': '模型调用结束', 'llm.thinking': '模型思考', 'tool.start': '工具调用', 'tool.end': '工具返回', 'tool.error': '工具出错', 'tool.gated': '门控判定', 'retrieve.end': '知识检索', 'memory.end': '记忆读写', - 'merge.end': '合并查询', 'sandbox.start': '沙箱开始', 'sandbox.end': '沙箱结束', 'agent.step.start': 'Agent 步骤开始', + 'merge.end': '合并查询', 'analysis.end': '分析卡的一项', + 'sandbox.start': '沙箱开始', 'sandbox.end': '沙箱结束', 'agent.step.start': 'Agent 步骤开始', 'agent.step.end': 'Agent 步骤结束', 'agent.route.start': '调度决策开始', 'agent.route.end': '调度决策结束', 'human.requested': '请求人工审批', 'human.resolved': '人工审批结果', 'approval.inquiry': '审批质询', 'issuance': '出具判定', 'caliber.upgrade': '口径卡升版处置', 'catalog.drift': '数据目录变化', @@ -259,6 +261,11 @@ recheckBy: '导出时服务端按封存的输入复算的结论；本页只核哈希，不在浏览器里重算统计', recheckNone: '导出时没有复算（结论未随文件附带）', analysisInput: (role, alias) => `${role}${alias ? ` · ${alias}` : ''}`, + // 分析结论输入的角色（analysis.contract.AnalysisInput.role），同证据面板的叫法 + roleLabel: { data: '数据', base: '基期', target: '本期', headline_base: '基期总数', headline_target: '本期总数', + parent: '上一层拆解' }, + dComparability: '两期可比性', comparabilityLabel: { comparable: '两期可比', conditional: '两期有条件可比', incomparable: '两期不可比' }, + dJudged: '判定', dChecks: '各项检查', comparabilityNotAttached: '可比性结论未随文件附带（见核对明细 ③ ④），这里按口径卡的记录展示', scopeTitle: '分析范围', scopeNote: '这一段由系统按这次运行的分析结论写成，写作者改不了；点开编号看判定语、判定标准、输入和复算。', scopeMissing: '没拆', scopeAll: '候选维度都拆过了', scopeUnknown: '候选维度核对不了', scopeTests: '检验', scopeTestsText: (n) => `这次做了 ${n} 次差异检验`, scopeParent: (alias) => `下钻自 ${alias}`, scopeDims: (dims) => `按${dims}`, @@ -1378,6 +1385,13 @@ } } } + // 指标上的可比性结论不进台账、只从卡上引用（卡记着它的 sha256）：已核对的卡记着的，同卡的输入一样认 + for (const metric of Array.isArray(item.json.metrics) ? item.json.metrics : []) { + const c = isObj(metric) && isObj(metric.comparability) ? metric.comparability : null + if (c && typeof c.artifact === 'string' && c.artifact) { + push(byCard, c.artifact, { card: item, metric: metric.id }) + } + } } // 已核对的分析结论记着的输入 → 输入快照、下钻的上一层结论。同口径卡：结论本身核过哈希，它记的输入和文档的证据目录 // 同一个分量；没核对过的（哈希对不上）不认 @@ -1958,6 +1972,7 @@ } parts.push(kv([[T.dRecompute, metric.recompute_ok == null ? T.recomputeNone : metric.recompute_ok ? T.recomputeOk : T.recomputeBad]])) + if (isObj(metric.comparability)) parts.push(...comparabilityBlock(metric.comparability)) } const snap = number.snapshot if (snap) parts.push(...snapshotBlock(snap, number.row, number.column, @@ -1966,6 +1981,42 @@ return parts } + /** 口径卡指标上的可比性（口径卡执行时自动做的可比性检查）：卡上记的判定和原因；那份结论随文件附带、核对通过的，再给 + * 判定标准、方法与版本、各项检查和导出时服务端的复算。没附带的照卡上的记录写，说明为什么没有更多 */ + function comparabilityBlock(c) { + const why = Array.isArray(c.reasons) && c.reasons.length ? `：${c.reasons.map(String).join('；')}` : '' + const label = own(T.comparabilityLabel, c.verdict) || String(c.phrase || '—') + const artifact = typeof c.artifact === 'string' && c.artifact ? c.artifact : null + const art = artifact ? view.out.arts.get(artifact) : null + const ok = art && art.result === PASS && view.out.chained && view.out.chained.has(artifact) + const result = ok && isObj(art.json) ? art.json : null + const parts = [h('p', { class: 'sub' }, T.dComparability), kv([ + [T.dJudged, `${label}${c.verdict === 'comparable' ? '' : why}`], + artifact ? [T.dArtifact, [h('code', {}, short(artifact)), artifactState(artifact)]] : null, + ])] + if (!result) { + if (artifact) parts.push(h('p', { class: 'note' }, T.comparabilityNotAttached)) + return parts + } + parts.push(kv([ + result.criterion ? [T.dCriterion, String(result.criterion)] : null, + [T.dMethod, String(result.method || '—')], + ])) + const table = isObj(result.table) ? result.table : null + const columns = table && Array.isArray(table.columns) ? table.columns.map(String) : [] + const labels = table && isObj(table.labels) ? table.labels : {} + if (columns.length && Array.isArray(table.rows) && table.rows.length) { + parts.push(h('p', { class: 'sub' }, T.dChecks), h('div', { class: 'table-wrap' }, h('table', {}, + h('thead', {}, h('tr', {}, columns.map((col) => h('th', {}, String(own(labels, col) || col))))), + h('tbody', {}, table.rows.filter(Array.isArray).map((row) => h('tr', {}, + columns.map((_, i) => h('td', {}, row[i] == null ? '—' : textOf(row[i]))))))))) + } + const recheck = isObj(art.a.recheck) ? art.a.recheck : null + parts.push(kv([[T.dRecheck, recheck + ? [String(recheck.text || '—'), h('span', { class: 'muted' }, `（${T.recheckBy}）`)] : T.recheckNone]])) + return parts + } + /** 数字详情里分析结论一节：判定语、判定标准、方法与版本、附带说明、Agent 写的假设、输入（各自的核对状态）、导出时服务端 * 的复算结论。结论没随文件附带（遮罩、文件不在）时只写复算一句 */ function analysisBlock(number) { @@ -1987,7 +2038,8 @@ const inputs = (Array.isArray(result.inputs) ? result.inputs : []).filter((i) => isObj(i) && typeof i.artifact === 'string') if (inputs.length) { parts.push(h('p', { class: 'sub' }, T.dAnalysisInputs), h('ul', { class: 'items' }, inputs.map((i) => h('li', {}, - h('span', { class: 'what' }, [T.analysisInput(String(i.role || ''), view.out.queryAliases.get(i.artifact)), ' ', + h('span', { class: 'what' }, [T.analysisInput(own(T.roleLabel, i.role) || String(i.role || ''), + view.out.queryAliases.get(i.artifact)), ' ', h('code', {}, short(i.artifact))]), artifactState(i.artifact))))) } } diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 84d8ba8d..d1aeeecd 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -4255,8 +4255,11 @@ def _comparability_note(entry: dict[str, Any]) -> str: comparability = entry.get("comparability") if not isinstance(comparability, dict): return "" - reasons = "；".join(str(r) for r in comparability.get("reasons") or [] if r) \ - or str(comparability.get("phrase") or "") + reasons = "；".join(str(r) for r in comparability.get("reasons") or [] if r) + if not reasons: + # 没记原因的照判定语写（「两期不可比：原因；原因」），开头的判定不重复 + phrase = str(comparability.get("phrase") or "") + reasons = phrase.split("：", 1)[1].strip() if "：" in phrase else "" if comparability.get("verdict") == "incomparable": return f"（两期不可比：{reasons or '原因见证据面板'}。这个比较不能当结论，引用时须如实说明两期不可比）" if comparability.get("verdict") == "conditional": @@ -4650,59 +4653,61 @@ def describe_violations(violations: list[dict[str, Any]], *, limit: int = 20) -> #: 复算不一致时点名哪几项不同的叫法（analysis.contract.AnalysisResult 的字段） _RESULT_FIELDS = { "op": "算子", "mode": "模式", "method": "方法版本", "params": "参数", "settings": "阈值", "inputs": "输入", - "probes": "现查的表范围", "status": "算成没算成", "reason_code": "算不了的原因", "reason": "算不了的原因", + "probes": "现查结果", "status": "算成没算成", "reason_code": "算不了的原因", "reason": "算不了的原因", "table": "结果表", "values": "具名的数", "verdict": "判定", "criterion": "判定标准", "phrase": "判定语", "caveats": "附带说明", } -#: 复算抛 NotImplementedError：这一版系统里还没有这种分析的算子 -RECHECK_MISSING = "系统里还没有这种分析的复算实现" +#: app.analysis.recompute 取不回输入（CallError 的 code input_missing）时那句以「复算」开头（复算取不回 Q2 的查询快照）： +#: 接在「复算失败：」后面时去掉，不说两遍 +_RECHECK_HEAD = "复算" -def analysis_input_snapshot(content: Any, artifact: str, *, ref: str | None = None, - node_id: str | None = None) -> dict[str, Any] | None: - """一份查询快照的内容 → 复算要的输入快照（analysis.contract.InputSnapshot：数据源查询工具的 to_payload 加上出处）。 - 不能当维度、度量的列（masked）按快照里记下的、查询当时的遮罩：复算按运行当时的条件重算；现在设的遮罩只管展示。 - 不是查询快照（没有 rows）的返回 None。""" - if not isinstance(content, dict) or not isinstance(content.get("rows"), list): - return None - return {"ref": str(ref or ""), "artifact": str(artifact), "node_id": str(node_id or ""), - "source": content.get("source"), "sql": str(content.get("sql") or ""), - "columns": [str(c) for c in content.get("columns") or []], - "column_types": content.get("column_types") if isinstance(content.get("column_types"), dict) else {}, - "rows": content["rows"], "truncated": bool(content.get("truncated")), - "masked": sorted(_mask_names(content.get("mask_columns")))} +def _recheck_reason(error: Any) -> str: + text = str(getattr(error, "message", "") or "").strip() or "原因未知" + if getattr(error, "code", None) == "input_missing" and text.startswith(_RECHECK_HEAD) \ + and len(text) > len(_RECHECK_HEAD): + return text[len(_RECHECK_HEAD):].lstrip() + return text def analysis_recheck(result: dict[str, Any], load: Callable[[str], Any], *, load_result: Callable[[str], Any] | None = None) -> dict[str, Any]: - """把一份封存的分析结论按它记下的输入复算一遍（app.analysis.recompute：按 sha256 取回输入快照，按结论里记下的参数、 - 阈值和现查结果算，不再现查），和封存的逐字段比（contract.same，agent 写的假设不比）。 + """把一份封存的分析结论按它记下的输入复算一遍，和封存的逐字段比（contract.same，agent 写的假设不比）。 + + 复算就是 app.analysis.recompute：load 按 sha256 交回查询快照工件的原内容，load_result 交回下钻的上一层结论的原内容， + 取不回（不在封存范围内、哈希对不上）的交 None；算子按结论里记下的参数、阈值、现查结果算，不再现查、不读目录。 交回 {status, text, method, method_now?, fields?}：status 是 same（一致）/ different（不一致）/ failed（复算不了）； - text 是给人看的一句，证据面板、报告导出、可验证报告照它写。方法升过版的照实写运行时用的是哪一版、现在是哪一版， - 复算结果同不同。复算抛异常（算子还没有、输入取不回、调用的写法现在不认）不往外抛，照实写「复算失败：原因」。""" + text 是给人看的一句，证据面板、报告导出、可验证报告照它写。fields 是和封存的不同的那几项（结论的字段名）。 + + 方法升过版（结论里记的 method 和现在的 contract.METHODS[(op, mode)] 不同）时 method_now 是现在的那一版：复算成立的照实写 + 复算结果同不同（方法版本那一项本来就不同，不算进去），现在的方法按封存的输入不成立（recompute 抛 CallError）的写明原因。 + 复算出错（输入取不回、调用的写法现在不认、出了意外）不往外抛，照实写「复算失败：原因」。""" from app import analysis - from app.analysis.contract import NOT_RECOMPUTED, CallError, same + from app.analysis.contract import METHODS, NOT_RECOMPUTED, CallError, same from app.core.errors import describe_exception method = str(result.get("method") or "") + now = METHODS.get((result.get("op"), result.get("mode"))) + upgraded = isinstance(now, str) and now != method + out: dict[str, Any] = {"method": method, **({"method_now": now} if upgraded else {})} + head = f"运行时用的是 {method}，现在是 {now}" if upgraded else "" try: again = analysis.recompute(result, load, load_result=load_result) # type: ignore[arg-type] - except NotImplementedError: - return {"status": "failed", "method": method, "text": f"复算失败：{RECHECK_MISSING}"} except CallError as e: - return {"status": "failed", "method": method, "text": f"复算失败：{e.message}"} + why = _recheck_reason(e) + text = f"{head}：现在的方法按封存的输入复算不了（{why}）" if upgraded else f"复算失败：{why}" + return {"status": "failed", **out, "text": text} except Exception as e: # noqa: BLE001 - 复算出了意外：照实说复算不了，不能让证据面板、导出跟着出错 - return {"status": "failed", "method": method, "text": f"复算失败：{describe_exception(e)}"} + return {"status": "failed", **out, "text": f"复算失败：{describe_exception(e)}"} if not isinstance(again, dict): - return {"status": "failed", "method": method, "text": "复算失败：复算没有交回结论"} - ok = same(result, again) - fields = sorted(k for k in set(result) | set(again) if k not in NOT_RECOMPUTED and result.get(k) != again.get(k)) - out: dict[str, Any] = {"status": "same" if ok else "different", "method": method} - now = str(again.get("method") or "") - if now != method: - out["method_now"] = now - out["text"] = f"运行时用的是 {method}，现在是 {now}，复算结果{'相同' if ok else '不同'}" + return {"status": "failed", **out, "text": "复算失败：复算没有交回结论"} + ok = same({**result, "method": now} if upgraded else result, again) + skip = NOT_RECOMPUTED | ({"method"} if upgraded else set()) + fields = sorted(k for k in set(result) | set(again) if k not in skip and result.get(k) != again.get(k)) + out["status"] = "same" if ok else "different" + if upgraded: + out["text"] = f"{head}，复算结果{'相同' if ok else '不同'}" elif ok: out["text"] = "复算一致：按封存的输入重新计算，结论逐项相同" else: @@ -4713,31 +4718,56 @@ def analysis_recheck(result: dict[str, Any], load: Callable[[str], Any], *, return out -def _decompose_row(result: dict[str, Any]) -> list[Any] | None: - """变化拆解的结果表里判定点名的那一段（verdict.subjects 的第一段，按维度列 params.by 对），对不上就取第一行（结果表 - 按变化的绝对值从大到小排）。""" +#: 变化拆解结果表里不是「段」的那几列（细案 01 §2：可加、乘积两种模式的数列和说明列） +_DECOMPOSE_NUMBERS = ("base", "target", "delta", "pct", "share", "contrib", "flag") + + +def _decompose_row(result: dict[str, Any]) -> tuple[str, list[Any]] | None: + """变化拆解的结果表里判定点名的那一段：(这一段的叫法, 那一行)。段按维度列（params.by；乘积模式是 factor 一列）对 + verdict.subjects 的第一项，对不上就取第一行（结果表按变化的绝对值从大到小排）。叫法是维度值用「·」连起来。""" table = result.get("table") if isinstance(result.get("table"), dict) else {} rows = [r for r in table.get("rows") or [] if isinstance(r, list)] columns = [str(c) for c in table.get("columns") or []] if not rows: return None - subjects = ((result.get("verdict") or {}).get("subjects") or []) if isinstance(result.get("verdict"), dict) else [] - by = (result.get("params") or {}).get("by") if isinstance(result.get("params"), dict) else None + params = result.get("params") if isinstance(result.get("params"), dict) else {} + by = ["factor"] if result.get("mode") == "factor" else params.get("by") by = [str(b) for b in (by if isinstance(by, list) else [by] if by else []) if str(b) in columns] - if subjects and by and isinstance(subjects[0], list): + if not by: + by = [c for c in columns if c not in _DECOMPOSE_NUMBERS][:1] + at = [columns.index(b) for b in by] + verdict = result.get("verdict") if isinstance(result.get("verdict"), dict) else {} + subjects = verdict.get("subjects") or [] + picked = rows[0] + if subjects and at and isinstance(subjects[0], list): wanted = [str(v) for v in subjects[0]] - for row in rows: - if [str(row[columns.index(b)]) for b in by if columns.index(b) < len(row)] == wanted: - return row - return rows[0] + picked = next((row for row in rows if [str(row[i]) for i in at if i < len(row)] == wanted), rows[0]) + name = "·".join(str(picked[i]) for i in at if i < len(picked) and picked[i] is not None) + return name or "这一段", picked + + +def _paren(text: str) -> str: + """代入式里减号、除号后面的负数加括号：「30.0% − （-1.0%）」。""" + return f"（{text}）" if text.startswith("-") else text def analysis_formulas(result: dict[str, Any]) -> list[dict[str, str]]: """出处链「计算」一步里关键数的代入式：[{label, text}]，数照单元格的规则渲染（系统写的，不是模型写的）。只给拿结论里 - 记下的数就写得出来的几条——变化拆解的总变化、变化率、主要来源的占比，比率拆解的结构效应、自身效应占比，波动基线的 - 常规波动范围和稳健 z；缺了哪个数、渲染不出来的那条不给。比如「华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ -194 = 61.9%」。""" + 记下的数就写得出来的几条： + + - 变化拆解（可加、乘积）：总变化、变化率、点名那一段（因子）的占比——「华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ + （-194） = 61.9%」，乘积模式按那个因子的贡献 + - 变化拆解（比率）：结构效应、自身效应各占比率变化的多少 + - 波动基线：常规波动范围、稳健 z（按变化率比的，几个数按百分比写） + + 量按单元格的规则写（千分位，小数照值本来的样子）；百分比同判定语的写法（app.analysis.phrases.pct：一位小数，不是 0 的值 + 写出来成了 0.0% 的多给小数位）。缺了哪个数、渲染不出来的那条不给，不编。趋势、差异检验、可比性检查的数在结果表和具名的数 + 里，不另写代入式。""" + from app.analysis.phrases import pct + values = result.get("values") if isinstance(result.get("values"), dict) else {} settings = result.get("settings") if isinstance(result.get("settings"), dict) else {} + params = result.get("params") if isinstance(result.get("params"), dict) else {} op, mode = result.get("op"), result.get("mode") out: list[dict[str, str]] = [] @@ -4747,34 +4777,38 @@ def analysis_formulas(result: dict[str, Any]) -> list[dict[str, str]]: except (KeyError, IndexError, ValueError, TypeError, RenderError): pass # 缺一个数、渲染不出来：这条不给，不编 + def cell(raw: Any, conv: str | None = None) -> str: + return pct(raw, signed=False) if conv == "pct" else render_cell(raw) + if result.get("status") != "ok": return out - if op == "decompose" and mode == "additive": - add("总变化", lambda: f"总变化 = 本期合计 − 基期合计 = {render_cell(values['target_total'])} − " - f"{render_cell(values['base_total'])} = {render_cell(values['delta_total'])}") - add("变化率", lambda: f"变化率 = 总变化 ÷ 基期合计 = {render_cell(values['delta_total'])} ÷ " - f"{render_cell(values['base_total'])} = {render_cell(values['pct_total'], 'pct')}") - table = result.get("table") or {} + if op == "decompose" and mode in ("additive", "factor"): + add("总变化", lambda: f"总变化 = 本期合计 − 基期合计 = {cell(values['target_total'])} − " + f"{_paren(cell(values['base_total']))} = {cell(values['delta_total'])}") + add("变化率", lambda: f"变化率 = 总变化 ÷ 基期合计 = {cell(values['delta_total'])} ÷ " + f"{_paren(cell(values['base_total']))} = {cell(values['pct_total'], 'pct')}") + table = result.get("table") if isinstance(result.get("table"), dict) else {} columns = [str(c) for c in table.get("columns") or []] - row = _decompose_row(result) - if row is not None and {"delta", "share"} <= set(columns): - by = [c for c in columns if c not in ("base", "target", "delta", "pct", "share", "flag")] - name = " · ".join(str(row[columns.index(c)]) for c in by if columns.index(c) < len(row)) or "这一段" - add("占比", lambda: f"{name}的占比 = {name}的变化 ÷ 总变化 = " - f"{render_cell(row[columns.index('delta')])} ÷ {render_cell(values['delta_total'])} = " - f"{render_cell(row[columns.index('share')], 'pct')}") + part = "contrib" if mode == "factor" else "delta" + found = _decompose_row(result) + if found is not None and {part, "share"} <= set(columns): + name, row = found + word = "贡献" if mode == "factor" else "变化" + add("占比", lambda: f"{name}的占比 = {name}的{word} ÷ 总变化 = {cell(row[columns.index(part)])} ÷ " + f"{_paren(cell(values['delta_total']))} = {cell(row[columns.index('share')], 'pct')}") elif op == "decompose" and mode == "ratio": for part, label in (("mix", "结构效应"), ("rate", "自身效应")): add(f"{label}占比", lambda part=part, label=label: ( - f"{label}占比 = {label}合计 ÷ 整体的变化 = {render_cell(values[f'{part}_total'])} ÷ " - f"{render_cell(values['delta'])} = {render_cell(values[f'{part}_share'], 'pct')}")) + f"{label}占比 = {label}合计 ÷ 比率的变化 = {cell(values[f'{part}_total'])} ÷ " + f"{_paren(cell(values['delta']))} = {cell(values[f'{part}_share'], 'pct')}")) elif op == "baseline": + conv = "pct" if params.get("basis") == "change" else None add("常规波动范围", lambda: ( - f"常规波动范围 = 中位数 ± {settings['k_warn']:g} 倍稳健标准差 = {render_cell(values['band_lo'])} 至 " - f"{render_cell(values['band_hi'])}（中位数 {render_cell(values['median'])}，" - f"过去 {render_cell(values['n_history'])} 期）")) - add("稳健 z", lambda: f"稳健 z = （本期 − 中位数）÷ 稳健标准差 = （{render_cell(values['x'])} − " - f"{render_cell(values['median'])}）÷ 稳健标准差 = {render_cell(values['z'])}") + f"常规波动范围 = 中位数 ± {settings['k_warn']:g} 倍稳健标准差 = {cell(values['band_lo'], conv)} 至 " + f"{cell(values['band_hi'], conv)}（中位数 {cell(values['median'], conv)}，" + f"过去 {cell(values['n_history'])} 期）")) + add("稳健 z", lambda: f"稳健 z = （本期 − 中位数）÷ 稳健标准差 = （{cell(values['x'], conv)} − " + f"{_paren(cell(values['median'], conv))}）÷ 稳健标准差 = {cell(values['z'])}") return out diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index 5895b195..583aef57 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -575,6 +575,10 @@ def _metric_excerpt(alias: str, entry: dict[str, Any], fetch: _Fetch, catalog: d if metric.get("expression"): lines.append(f"原式：{_clean(str(metric['expression']))}") lines.extend(_metric_inputs(metric, fetch, catalog, masked, mask)) + comparability = metric.get("comparability") + if isinstance(comparability, dict) and str(comparability.get("phrase") or "").strip(): + # 跨两份同形查询算的环比、同比：口径卡执行时自动做的可比性检查（判定语是系统写的，原样给） + lines.append(f"两期可比性（系统检查）：{_clean(str(comparability['phrase']))}") return "\n".join(lines) diff --git a/backend/app/report_export/runtime.js b/backend/app/report_export/runtime.js index 2f5ca262..a98595ad 100644 --- a/backend/app/report_export/runtime.js +++ b/backend/app/report_export/runtime.js @@ -80,14 +80,18 @@ unnamed: '已删除的工作流', // 口径卡指标上的可比性（跨两份同形查询算的环比、同比） dComparability: '两期可比性', comparabilityLabel: { comparable: '两期可比', conditional: '两期有条件可比', incomparable: '两期不可比' }, + comparabilityOpen: '展开这份可比性结论（判定、方法、计算、输入、复算、封存）', // 分析结论：判定 → 方法 → 计算 → 输入 → 复算 → 封存 analysis: '分析结论', dPhrase: '判定语', dCriterion: '判定标准', dSubjects: '点名的段', dDirection: '方向', directionLabel: { up: '上升', down: '下降' }, dCaveats: '附带说明', dReason: '算不了的原因', dCitedPart: '这一处引用的是', citedPhrase: '判定语', citedValue: (k) => `具名的数「${k}」`, - citedCell: (r, c) => `结果表第 ${r + 1} 行「${c}」`, dWho: '谁做的', originLabel: { card: '分析卡（模板定好的分析）', agent: 'Agent 现场做的分析' }, + citedCell: (r, c) => `结果表第 ${r + 1} 行「${c}」`, dWho: '谁做的', + originLabel: { card: '分析卡（模板定好的分析）', agent: 'Agent 现场做的分析', metric: '口径卡执行时自动做的可比性检查' }, dHypothesis: 'Agent 写的假设', refused: '这次分析没有做成', renderOk: '与按结论渲染的结果一致', renderBad: '与按结论渲染的结果不一致', - method: '方法', dMethod: '方法与版本', dParams: '参数', dSettings: '实际用的阈值', dProbes: '现查的表范围', + method: '方法', dMethod: '方法与版本', dParams: '参数', dSettings: '实际用的阈值', dProbes: '现查结果', probeBasis: { live: '现查', profile: '剖析', observed: '取值观察' }, + probeBound: (expr, value) => `按运行当天现算的界 ${expr} = ${value}`, probeEmpty: '这一列一个值都没有', + probeRange: (min, max) => `${min} 至 ${max}`, calc: '计算', dValues: '具名的数', dFormulas: '关键数的代入式', dResultTable: '结果表', resultRows: (shown, total) => `列出 ${shown} 行，共 ${total} 行`, analysisInput: '输入', dRole: '角色', dRef: '调用里写的名字', dColumnsUsed: '用到的列', dRowsUsed: '用到的行', @@ -95,6 +99,7 @@ closureOk: (h, t) => `总数 ${h} 与各段之和 ${t} 闭合`, closureBad: (h, t) => `总数 ${h} 与各段之和 ${t} 对不上`, dParent: '下钻自', recompute: '复算', seal: '封存', inputOfAnalysis: (alias) => `分析结论 ${alias} 的输入`, inLedger: '封存范围内的台账记过这份分析结论', notInLedger: '封存范围内的台账里没有这份分析结论', + viaCard: '记在口径卡的指标集里，随卡封存（不进台账）', } const MASKED = '已遮罩' @@ -235,10 +240,17 @@ const c = step.comparability const why = Array.isArray(c.reasons) && c.reasons.length ? `：${c.reasons.join('；')}` : '' body.push(h('div', { style: 'margin-top:8px' }, kv([ - [T.dComparability, `${T.comparabilityLabel[c.verdict] || c.phrase || NONE}${c.verdict === 'comparable' ? '' : why}`], + [T.dComparability, + `${own(T.comparabilityLabel, c.verdict) || c.phrase || NONE}${c.verdict === 'comparable' ? '' : why}`], c.criterion ? [T.dCriterion, String(c.criterion)] : null, c.artifact ? [T.dArtifact, h('code', {}, short(c.artifact))] : null, ]))) + // 点得开那份可比性结论：服务端给的出处链（同分析结论的引用），收在一个可展开的块里 + const steps = Array.isArray(c.steps) ? c.steps.filter((x) => isObj(x) && own(PARTS, x.step)) : [] + if (steps.length) { + body.push(h('details', { class: 'comparability', 'data-comparability': '' }, + h('summary', {}, T.comparabilityOpen), ...steps.map((x) => PARTS[x.step](x)))) + } } // 双盲复算（期 4）：复核过这个指标才有 if (isObj(step.dual_check)) { @@ -461,7 +473,8 @@ ? [T.dSubjects, verdict.subjects.map((s) => (Array.isArray(s) ? s.join(' · ') : String(s))).join('、')] : null, verdict && verdict.direction ? [T.dDirection, T.directionLabel[verdict.direction] || String(verdict.direction)] : null, [T.dCitedPart, citedText], - [T.dWho, `${T.originLabel[step.origin] || NONE}${step.node_id ? ` · ${step.label || nodeLabel(step.node_id)}` : ''}`], + [T.dWho, `${own(T.originLabel, step.origin) || NONE}` + + `${step.node_id ? ` · ${step.label || nodeLabel(step.node_id)}` : ''}`], ])) if (step.status === 'refused') body.push(waiting(`${T.refused}${step.reason ? `：${step.reason}` : ''}`)) if (Array.isArray(step.caveats) && step.caveats.length) { @@ -475,10 +488,21 @@ if (step.note) body.push(waiting(step.note)) if (typeof step.render_ok === 'boolean') body.push(muted(step.render_ok ? T.renderOk : T.renderBad)) return h('section', { class: 'part', 'data-step': 'analysis' }, - h('h3', {}, `${T.analysis} ${step.alias || ''} · ${step.method_label || NONE}`), h('div', { class: 'part-body' }, ...body)) + h('h3', {}, `${[T.analysis, step.alias].filter(Boolean).join(' ')} · ${step.method_label || NONE}`), + h('div', { class: 'part-body' }, ...body)) } - /** 分析结论 · 方法：方法与版本、参数、实际用的阈值、现查的表范围 */ + /** 现查结果的一条：有 expr 的是按运行当天现算的界（CURRENT_DATE 这类在库里算出的日期），其余是一列的实际范围；现查到的 + * 范围最早、最晚都空，是这一列一个值都没有 */ + function probeText(p) { + const how = `（${own(T.probeBasis, p.basis) || p.basis || NONE}${p.at ? ` · ${p.at}` : ''}）` + if (typeof p.expr === 'string' && p.expr) return `${T.probeBound(p.expr, p.value || NONE)}${how}` + const where = `${p.table || NONE}.${p.column || NONE}` + const empty = p.basis === 'live' && !p.min && !p.max + return `${where}：${empty ? T.probeEmpty : T.probeRange(p.min || NONE, p.max || NONE)}${how}` + } + + /** 分析结论 · 方法：方法与版本、参数、实际用的阈值、现查结果 */ function analysisMethodPart(step) { const body = [kv([[T.dMethod, `${step.method_label || NONE} · ${step.method || NONE}`]])] if (isObj(step.params) && Object.keys(step.params).length) { @@ -489,8 +513,7 @@ } if (Array.isArray(step.probes) && step.probes.length) { body.push(h('p', { class: 'sub' }, T.dProbes), h('ul', { style: 'margin:4px 0 0;padding-left:1.2em;font-size:13px' }, - step.probes.map((p) => h('li', {}, `${p.table || NONE}.${p.column || NONE}：${p.min || NONE} 至 ${p.max || NONE}` - + `（${T.probeBasis[p.basis] || p.basis || NONE}${p.at ? ` · ${p.at}` : ''}）`)))) + step.probes.filter(isObj).map((p) => h('li', {}, probeText(p))))) } return h('section', { class: 'part', 'data-step': 'analysis_method' }, h('h3', {}, T.method), h('div', { class: 'part-body' }, ...body)) } @@ -550,7 +573,7 @@ /** 分析结论 · 封存：台账记没记过、哈希核对 */ function analysisSealPart(step) { const body = [kv([[T.dArtifact, step.artifact ? h('code', {}, short(step.artifact)) : NONE], - [T.seal, step.ledger ? T.inLedger : T.notInLedger]])] + [T.seal, step.via === 'metric_set' ? T.viaCard : step.ledger ? T.inLedger : T.notInLedger]])] if (step.note) body.push(waiting(step.note)) const bits = integrity(step) if (bits) body.push(bits) diff --git a/backend/tests/test_analysis_bundle.py b/backend/tests/test_analysis_bundle.py index b749fab8..d41f31aa 100644 --- a/backend/tests/test_analysis_bundle.py +++ b/backend/tests/test_analysis_bundle.py @@ -1,6 +1,9 @@ """分析结论进可验证报告（第 1 波 W1-b）：bundle/build.py 带上结论工件和它的输入快照，verifier.js 在 node 里核对。 -- 报告引用的分析结论按原字节嵌入，连同输入快照；服务端导出时按封存的输入复算一遍，结果记在工件条目上（recheck） +- 报告引用的分析结论按原字节嵌入，连同输入快照；服务端导出时按封存的输入复算一遍（真的 app.analysis.recompute），结果记在 + 工件条目上（recheck） +- 口径卡指标上的可比性结论（不进台账、只从卡上引用）随卡附带，④ 按「已核对的口径卡记着的可比性」认它 +- 分析工具的 tool.end、分析卡一项的 analysis.end：结论随文件附带得了才带规范字符串 - 页面只核哈希、核判定语和数是不是结论里写的那几个字（不在浏览器里重算统计）：原样导出全部通过，只改显示文字的伪造 点名不通过 - 输入快照只给 id（证明不了没碰遮罩列）的，结论本身也只给 id，引用它的判定语、数无法核对 @@ -14,13 +17,12 @@ from typing import Any import pytest -from app import analysis from app.analysis.contract import ARTIFACT_KIND, ledger_entry -from app.analysis.testing import fake_result from app.bundle.build import build_bundle from app.core.artifact_store import canonical_json, content_hash, manifest_hash from app.engine.evidence import build_catalog, compose_doc, iter_segments -from tests.test_analysis_evidence import SNAPSHOT +from tests.test_analysis_evidence import SNAPSHOT, analyzed, real_result, run_sync +from tests.test_analysis_evidence_api import LAST_WEEK, THIS_WEEK from tests.test_bundle_shown import NODE, node, reseal pytestmark = pytest.mark.skipif(NODE is None, reason="没有 node，无法执行页面上的核对脚本") @@ -47,14 +49,15 @@ class Files(dict): def bundle_of(text: str = TEXT, *, snapshot: dict[str, Any] | None = None, doc_extra: dict[str, Any] | None = None, - source_masks: dict[str, list[str]] | None = None, - graph: dict[str, Any] | None = None) -> tuple[dict[str, Any], Files, str]: - """一次封存好的运行的页面数据：(bundle, 工件, 分析结论的 id)。""" + source_masks: dict[str, list[str]] | None = None, graph: dict[str, Any] | None = None, + extra_events: list[tuple[str, str | None, dict[str, Any]]] = (), + drop: tuple[str, ...] = ()) -> tuple[dict[str, Any], Files, str]: + """一次封存好的运行的页面数据：(bundle, 工件, 分析结论的 id)。结论按快照真算（app.analysis.run）；extra_events 接在 + 分析节点完成之后（{analysis} 换成结论的 id），drop 是导出时服务端已经没有的工件（"query"：输入快照）。""" files = Files() snap = snapshot or SNAPSHOT q = files.put(snap) - result = fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", "columns": ["region", "week", "gmv"], - "rows": "all"}]) + result = real_result(q, snap) a = files.put(result) qentry = {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": q, "tool": "db_query__shop", "source": snap.get("source"), "columns": snap["columns"], "rows": len(snap["rows"]), "truncated": False} @@ -74,10 +77,14 @@ def bundle_of(text: str = TEXT, *, snapshot: dict[str, Any] | None = None, doc_e add("tool.end", "fetch", {"tool": "db_query__shop", "duration_ms": 5, "query_artifact": q, "preview": "…"}) add("node.finished", "fetch", {"duration_ms": 6, "attempt": 1, "evidence": [qentry]}) add("node.finished", "an", {"duration_ms": 8, "attempt": 1, "evidence": [aentry]}) + for etype, node_id, data in extra_events: + add(etype, node_id, json.loads(json.dumps(data).replace("{analysis}", a))) add("report.checked", "write", {"doc_artifact": doc_id, "ok": True}) add("node.finished", "write", {"duration_ms": 9, "attempt": 1, "preview": {"text": "（报告正文）", "doc_artifact": doc_id}}) add("issuance", "out", {"tier": "formal", "mode": "citations"}) add("run.finished", None, {"output": {"周报": "（报告正文）"}}) + if "query" in drop: + files.pop(q) bundle = build_bundle(RUN, events, manifest_seq=len(events), manifest_hash=manifest_hash(events), read=files.get, exported_at="2026-10-21T08:00:00+00:00", graph=graph or GRAPH, source_masks=source_masks or {}) @@ -96,8 +103,9 @@ def test_the_analysis_and_its_input_travel_with_the_file_and_verify(): bundle, files, a = bundle_of() analysis_item = item(bundle, a) assert analysis_item["kind"] == ARTIFACT_KIND and "b64" in analysis_item and analysis_item["node_id"] == "an" - assert analysis_item["recheck"] == {"status": "failed", "method": "decompose.additive@1", - "text": "复算失败：系统里还没有这种分析的复算实现"} + # 导出时服务端真的复算了一遍：按封存的输入快照（这份文件里哈希对得上的原字节）重算，逐字段相同 + assert analysis_item["recheck"] == {"status": "same", "method": "decompose.additive@1", + "text": "复算一致：按封存的输入重新计算，结论逐项相同"} [query] = [x for x in bundle["artifacts"] if x["kind"] == "query"] assert "b64" in query @@ -106,17 +114,17 @@ def test_the_analysis_and_its_input_travel_with_the_file_and_verify(): numbers = {n["ref"]: n for n in got["numbers"]} assert set(numbers) == {"a:A1", "a:A1.top_share|pct", "v:A1.r0.delta"} assert all(n["result"] == "pass" and n["expected"] == n["text"] for n in numbers.values()), numbers - assert numbers["a:A1"]["text"] == "销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）" + assert numbers["a:A1"]["text"] == "销售额下降 194（-8.7%），主要来自华东（-120，占 61.9%）" assert got["overall"]["result"] == "pass" -def test_the_server_recheck_is_recorded_when_the_operators_answer(monkeypatch): - monkeypatch.setattr(analysis, "recompute", lambda result, load, *, load_result=None: dict(result)) - bundle, _, a = bundle_of() - assert item(bundle, a)["recheck"]["text"] == "复算一致：按封存的输入重新计算，结论逐项相同" +def test_an_analysis_whose_input_is_gone_travels_only_as_a_hash(): + """导出时服务端已经没有输入快照：证明不了没碰遮罩列，结论只给 id，也就不顺着它复算、不附带别的。""" + bundle, _, a = bundle_of(drop=("query",)) + assert item(bundle, a) == {"id": a, "kind": ARTIFACT_KIND, "node_id": "an", "masked": True} -@pytest.mark.parametrize("ref, text", [("a:A1", "销售额下降 194 万（-8.7%），主要来自华南（-120 万，占 62%）"), +@pytest.mark.parametrize("ref, text", [("a:A1", "销售额下降 194（-8.7%），主要来自华南（-120，占 61.9%）"), ("a:A1.top_share|pct", "71.9%"), ("v:A1.r0.delta", "-220")]) def test_forging_the_shown_text_fails_by_name(ref, text): bundle, _, _ = bundle_of() @@ -173,3 +181,88 @@ def test_an_analysis_card_event_shows_its_content_only_when_its_conclusions_trav assert not shown(bundle_of(graph=card, snapshot=masked)[0]) got = verdicts(bundle_of(graph=card)[0]) assert all(r == "pass" for key in ("events", "manifest", "artifacts", "chain") for r in got[key]) + + +def test_analysis_tool_and_card_events_show_content_only_when_the_conclusion_travels(): + """分析工具的 tool.end（预览里是结论的前一截）、分析卡一项的 analysis.end（带判定语）：和结论随文件附带同一个认法。""" + events = [("tool.end", "an", {"tool": "analysis_decompose", "duration_ms": 3, "analysis_artifact": "{analysis}", + "preview": "{\"analysis_artifact\": \"{analysis}\"}"}), + ("analysis.end", "an", {"item": "drop", "op": "decompose", "mode": "additive", "status": "ok", + "verdict": "driver", "direction": "down", "phrase": "主要来自华东", + "reason_code": None, "analysis_artifact": "{analysis}"})] + + def shown(bundle: dict[str, Any]) -> dict[str, bool]: + return {e["type"]: "canonical" in e for e in bundle["events"] if e["type"] in ("tool.end", "analysis.end") + and e["node_id"] == "an"} + + open_ = bundle_of(extra_events=events)[0] + assert shown(open_) == {"tool.end": True, "analysis.end": True} + got = verdicts(open_) + assert all(r == "pass" for key in ("events", "manifest", "artifacts", "chain") for r in got[key]), got["why"] + masked = {**SNAPSHOT, "mask_columns": ["phone"], "columns": ["region", "phone", "week", "gmv"], + "sql": "SELECT region, phone, week, SUM(gmv) AS gmv FROM orders GROUP BY 1, 2, 3", + "rows": [[r[0], "000-0001", r[1], r[2]] for r in SNAPSHOT["rows"]]} + assert shown(bundle_of(snapshot=masked, extra_events=events)[0]) == {"tool.end": False, "analysis.end": False} + + +def card_bundle() -> tuple[dict[str, Any], Files, str, str]: + """跨两周算环比的口径卡，指标上记着口径卡执行时自动做的可比性检查（结论落成工件，不进台账）： + (bundle, 工件, 可比性结论的 id, 口径卡的 id)。""" + files = Files() + last, this = files.put(LAST_WEEK), files.put(THIS_WEEK) + result = run_sync(analyzed({"fetch·base": (LAST_WEEK, last), "fetch·target": (THIS_WEEK, this)}, + {"op": "comparable", "mode": None, "params": {}}, + base_query="fetch·base", target_query="fetch·target")) + found = files.put(result) + card = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v1", "metrics": [ + {"id": "wow", "name": "环比", "value": -0.0874, "unit": "", "format": "percent_of_ratio", "decimals": 3, + "expression": "cell(nodes.fetch, 0, 'gmv') / cell(nodes.fetch, 0, 'gmv') - 1", + "substituted": "2026 / 2220 - 1", "recompute_ok": True, "status": "ok", + "inputs": [{"path": "cell(nodes.fetch, 0, 'gmv')", "value": 2026, "node_id": "fetch", "via": "tool_cell", + "artifact": this, "locator": {"row": 0, "column": "gmv"}}, + {"path": "cell(nodes.fetch, 0, 'gmv')", "value": 2220, "node_id": "fetch", "via": "tool_cell", + "artifact": last, "locator": {"row": 0, "column": "gmv"}}], + "comparability": {"artifact": found, "verdict": result["verdict"]["code"], "phrase": result["phrase"], + "reasons": []}}]} + card_id = files.put(card) + qentries = [{"kind": "query", "node_id": "fetch", "exec": 1, "artifact": x, "tool": "db_query__shop", + "source": "shop", "columns": ["gmv"], "rows": 1, "truncated": False} for x in (last, this)] + mentry = {"kind": "metric_set", "node_id": "card", "exec": 1, "artifact": card_id} + catalog = build_catalog(nodes={"card": {**card, "artifact": card_id}}, ledger=[*qentries, mentry], + loader=files.load) + doc = compose_doc("## 环比\n\n销售额环比 [[m:wow]]。", catalog, node_id="write", run_id=RUN_ID, loader=files.load) + doc_id = files.put(doc) + events: list[tuple[int, str, str | None, Any]] = [] + + def add(etype: str, node_id: str | None, data: dict[str, Any]) -> None: + events.append((len(events) + 1, etype, node_id, data)) + + add("run.started", None, {"nodes": 4, "resumed": False, "memory_scope": "default", "collection": "default", + "approval_default": None, "tool_trust": "formal", "agent_limits": {}, + "data_versions": None, "replay_protocol": 2}) + for x in (last, this): + add("tool.end", "fetch", {"tool": "db_query__shop", "duration_ms": 5, "query_artifact": x, "preview": "…"}) + add("node.finished", "fetch", {"duration_ms": 6, "attempt": 1, "evidence": qentries}) + add("node.finished", "card", {"duration_ms": 7, "attempt": 1, "evidence": [mentry]}) + add("report.checked", "write", {"doc_artifact": doc_id, "ok": True}) + add("node.finished", "write", {"duration_ms": 9, "attempt": 1, "preview": {"text": "（报告正文）", "doc_artifact": doc_id}}) + add("run.finished", None, {"output": {"周报": "（报告正文）"}}) + graph = {"nodes": [{"id": n, "type": t, "data": {"label": label}} for n, t, label in [ + ("fetch", "tool", "按周取数"), ("card", "metrics", "周报口径"), ("write", "report", "写周报")]]} + bundle = build_bundle(RUN, events, manifest_seq=len(events), manifest_hash=manifest_hash(events), read=files.get, + exported_at="2026-10-21T08:00:00+00:00", graph=graph, source_masks={}) + return bundle, files, found, card_id + + +def test_the_comparability_on_a_card_travels_with_the_card_and_is_chained_through_it(): + bundle, _, found, card_id = card_bundle() + assert "b64" in item(bundle, card_id) + comparability = item(bundle, found) + assert comparability["kind"] == ARTIFACT_KIND and "b64" in comparability and comparability["node_id"] == "card" + assert comparability["recheck"]["status"] == "same" + got = verdicts(bundle) + assert all(r == "pass" for key in ("events", "manifest", "artifacts", "chain") for r in got[key]), got["why"] + assert {w["kind"] for w in got["why"]["chain"]} >= {"metric_set", ARTIFACT_KIND, "query"} + [wow] = got["numbers"] + assert wow["result"] == "pass", wow + diff --git a/backend/tests/test_analysis_evidence.py b/backend/tests/test_analysis_evidence.py index 0109765f..84c922de 100644 --- a/backend/tests/test_analysis_evidence.py +++ b/backend/tests/test_analysis_evidence.py @@ -8,23 +8,27 @@ - 写作目录的「分析结论」一段：判定语、判定标准、具名的数、结果表前 table_top_k 段、附带说明、算不了的原因、来自哪几个 Q， 放不下时结果表的各段排在查询的行前面补 - 分析结论的输入查询上的 SQL 检查问题随 A 条目出现；口径卡指标上的可比性照实写进写作目录 -- 复算（analysis_recheck）、代入式（analysis_formulas） +- 复算（analysis_recheck）、代入式（analysis_formulas）：用 app.analysis 真算的结论（run 交的输入照分析入口的写法，复算交 + 工件原内容），也用 fake_result 测目录和引用的写法 """ from __future__ import annotations +import asyncio +from concurrent.futures import ThreadPoolExecutor +from datetime import date from typing import Any import pytest from app import analysis -from app.analysis.contract import CallError, ledger_entry +from app.analysis import contract +from app.analysis.contract import CallError, effective_settings, ledger_entry from app.analysis.testing import fake_result from app.core.artifact_store import canonical_json, content_hash from app.engine.evidence import ( ANALYSIS_CELLS_REASON, StreamRenderer, analysis_formulas, - analysis_input_snapshot, analysis_recheck, build_catalog, compose_doc, @@ -58,6 +62,59 @@ class Store(dict): return self.get(artifact) +#: 运行当天（服务端时区）：真算的结论记进 params.today，复算按它 +TODAY = date(2026, 10, 21) +#: 按 SNAPSHOT 真算一份按地区拆的可加拆解 +REQUEST: dict[str, Any] = { + "op": "decompose", "mode": "additive", "hypothesis": "华东拖累", + "params": {"query": "Q2", "by": ["region"], "period": "week", "base": "W41", "target": "W42", "measure": "gmv", + "label": "销售额"}} +#: 华东按渠道再拆一层（下钻）：两个渠道之和 820 → 700，和 A1 里华东那一段闭合 +CHANNELS: dict[str, Any] = { + "columns": ["channel", "week", "gmv"], "column_types": {"channel": "text", "week": "text", "gmv": "number"}, + "rows": [["线上", "W41", 500], ["线上", "W42", 410], ["线下", "W41", 320], ["线下", "W42", 290]], + "sql": "SELECT channel, week, SUM(gmv) AS gmv FROM orders WHERE region = '华东' GROUP BY channel, week", + "source": "shop", "truncated": False, "row_count": 4, +} + + +def input_snapshot(content: dict[str, Any], *, ref: str, artifact: str, node_id: str = "fetch") -> dict[str, Any]: + """照分析入口（tools/analysis.input_snapshot）的写法交给 run：列的类型按列序写成列表，加上出处，遮罩列留空。复算交的是 + 工件原内容（类型是 {列: 类型}）：两边算出来要逐字段相同。""" + types = content.get("column_types") or {} + return {**content, "ref": ref, "artifact": artifact, "node_id": node_id, + "column_types": [types.get(c) for c in content["columns"]], "masked": []} + + +async def analyzed(snapshots: dict[str, tuple[dict[str, Any], str]], request: dict[str, Any] | None = None, *, + parents: dict[str, Any] | None = None, hypothesis: str | None = None, + **params: Any) -> dict[str, Any]: + """app.analysis.run 真算一份结论。snapshots：{调用里写的名字: (快照内容, 工件 id)}；params 盖在请求的参数上，hypothesis + 换掉请求里 agent 写的假设（不进复算的比较，用来让一份结论的内容独一份）。""" + request = request or REQUEST + by_ref = {ref: input_snapshot(content, ref=ref, artifact=sha) for ref, (content, sha) in snapshots.items()} + asked = {**request, "params": {**request["params"], **params}} + if hypothesis is not None: + asked["hypothesis"] = hypothesis + return dict(await analysis.run(asked, by_ref.get, settings=effective_settings(None), context={"today": TODAY}, + parents=(parents or {}).get)) + + +def run_sync(coro: Any) -> Any: + """在同步的地方等一个协程：没有正在跑的事件循环就直接 asyncio.run，协程测试里调用（已经在循环里）换一个线程跑。""" + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(coro) + with ThreadPoolExecutor(1) as pool: + return pool.submit(asyncio.run, coro).result() + + +def real_result(q: str, snapshot: dict[str, Any] | None = None, **params: Any) -> dict[str, Any]: + """同步的 analyzed：按 SNAPSHOT（工件 q）真算 REQUEST。""" + return run_sync(analyzed({"Q2": (snapshot or SNAPSHOT, q)}, **params)) + + def setup(*, item: str | None = "drop", call_id: str | None = None, result: dict[str, Any] | None = None, snapshot: dict[str, Any] | None = None) -> tuple[Store, list[dict[str, Any]], str, str]: """一次查询（Q1）加一份按地区拆的分析结论（A1）：(工件库, 台账, 查询快照, 分析结论)。""" @@ -451,71 +508,139 @@ def test_sql_check_problems_of_the_input_queries_follow_the_analysis(): # -------------------------------------------------------------------------- -def test_recheck_says_plainly_when_the_operators_are_not_there_yet(): - store, ledger, q, a = setup() - out = analysis_recheck(store[a], lambda sha: analysis_input_snapshot(store.get(sha), sha)) - assert out == {"status": "failed", "method": "decompose.additive@1", - "text": "复算失败：系统里还没有这种分析的复算实现"} - - -def test_recheck_compares_field_by_field_and_names_the_method_versions(monkeypatch): - store, _, q, a = setup() - sealed = store[a] - seen: list[Any] = [] - - def same(result, load, *, load_result=None): - seen.append(load(q)) - return {**result, "hypothesis": "复算时没有假设"} - - monkeypatch.setattr(analysis, "recompute", same) - out = analysis_recheck(sealed, lambda sha: analysis_input_snapshot(store.get(sha), sha, ref="Q2", node_id="fetch")) - assert out == {"status": "same", "method": "decompose.additive@1", "text": "复算一致：按封存的输入重新计算，结论逐项相同"} - assert seen[0]["rows"] == SNAPSHOT["rows"] and seen[0]["masked"] == [] and seen[0]["node_id"] == "fetch" - - monkeypatch.setattr(analysis, "recompute", lambda r, load, *, load_result=None: {**r, "phrase": "别的判定"}) - assert analysis_recheck(sealed, store.get)["text"] == "复算不一致：判定语和封存的不同" - - monkeypatch.setattr(analysis, "recompute", - lambda r, load, *, load_result=None: {**r, "method": "decompose.additive@2"}) +def test_recheck_with_the_real_operators_says_a_sealed_result_holds(): + """真算的结论按封存的输入复算（load 交工件原内容），逐字段相同；agent 写的假设不比。""" + store = Store() + q = store.put(SNAPSHOT) + sealed = real_result(q) + assert sealed["phrase"] == "销售额下降 194（-8.7%），主要来自华东（-120，占 61.9%）" out = analysis_recheck(sealed, store.get) - assert out["status"] == "different" and out["method_now"] == "decompose.additive@2" + assert out == {"status": "same", "method": "decompose.additive@1", "text": "复算一致：按封存的输入重新计算，结论逐项相同"} + assert analysis_recheck({**sealed, "hypothesis": "别的假设"}, store.get)["status"] == "same" + + +def test_recheck_follows_the_parent_of_a_drill_down_by_hash(): + store = Store() + q, c = store.put(SNAPSHOT), store.put(CHANNELS) + top = real_result(q) + a = store.put(top) + child = asyncio.run(analyzed({"Q3": (CHANNELS, c)}, parents={"A1": top}, query="Q3", by=["channel"], + parent={"of": "A1", "key": ["华东"]})) + assert [i["role"] for i in child["inputs"]] == ["parent", "data"] and child["inputs"][0]["artifact"] == a + assert analysis_recheck(child, store.get, load_result=store.get)["status"] == "same" + # 上一层取不回：照实说取不回哪一份，不说两遍「复算」 + out = analysis_recheck(child, store.get, load_result=lambda sha: None) + assert out == {"status": "failed", "method": "decompose.additive@1", "text": "复算失败：取不回上一层的分析结论 A1"} + + +def test_recheck_says_what_differs_and_what_is_missing(): + store = Store() + q = store.put(SNAPSHOT) + sealed = real_result(q) + # 封存的那份和按输入算出来的不一样（判定语被改过、哈希也就是另一份）：点名哪几项不同 + out = analysis_recheck({**sealed, "phrase": "销售额下降，主要来自华南"}, store.get) + assert out["status"] == "different" and out["fields"] == ["phrase"] + assert out["text"] == "复算不一致：判定语和封存的不同" + # 输入快照取不回（不在封存范围、哈希对不上的调用方都交 None） + out = analysis_recheck(sealed, lambda sha: None) + assert out == {"status": "failed", "method": "decompose.additive@1", "text": "复算失败：取不回 Q2 的查询快照"} + # 输入快照换了一份（同一个 sha256 交回了别的内容）：复算照样算，结果和封存的不同 + other = {**SNAPSHOT, "rows": [["华东", "W41", 820], ["华东", "W42", 650], *SNAPSHOT["rows"][2:]]} + assert analysis_recheck(sealed, lambda sha: other)["status"] == "different" + + +def test_recheck_names_both_versions_when_the_method_was_upgraded(monkeypatch): + store = Store() + q = store.put(SNAPSHOT) + sealed = real_result(q) + monkeypatch.setitem(contract.METHODS, ("decompose", "additive"), "decompose.additive@2") + out = analysis_recheck(sealed, store.get) + assert out == {"status": "same", "method": "decompose.additive@1", "method_now": "decompose.additive@2", + "text": "运行时用的是 decompose.additive@1，现在是 decompose.additive@2，复算结果相同"} + out = analysis_recheck({**sealed, "phrase": "别的判定"}, store.get) + assert out["status"] == "different" and out["fields"] == ["phrase"] assert out["text"] == "运行时用的是 decompose.additive@1，现在是 decompose.additive@2，复算结果不同" - def refuse(r, load, *, load_result=None): - raise CallError("missing_column", "Q2 里没有列「gmv」") + def refuse(result, load, *, load_result=None): + raise CallError("unknown_mode", "变化拆解要写模式：additive / ratio / factor") monkeypatch.setattr(analysis, "recompute", refuse) - assert analysis_recheck(sealed, store.get)["text"] == "复算失败：Q2 里没有列「gmv」" + out = analysis_recheck(sealed, store.get) + assert out["status"] == "failed" and out["method_now"] == "decompose.additive@2" + assert out["text"] == ("运行时用的是 decompose.additive@1，现在是 decompose.additive@2：现在的方法按封存的输入复算不了" + "（变化拆解要写模式：additive / ratio / factor）") - def broken(r, load, *, load_result=None): - raise RuntimeError("输入快照取不回来") + +def test_recheck_errors_are_said_plainly(monkeypatch): + store = Store() + q = store.put(SNAPSHOT) + sealed = real_result(q) + seen: list[Any] = [] + + def broken(result, load, *, load_result=None): + seen.append(load(q)) + raise ZeroDivisionError("division by zero") monkeypatch.setattr(analysis, "recompute", broken) - assert analysis_recheck(sealed, store.get)["text"] == "复算失败：输入快照取不回来" - - -def test_input_snapshots_for_recompute_keep_the_masks_of_the_query_time(): - snap = analysis_input_snapshot({**SNAPSHOT, "mask_columns": ["Phone", "email"]}, "a" * 64, ref="Q1") - assert snap["masked"] == ["email", "phone"] and snap["artifact"] == "a" * 64 and snap["ref"] == "Q1" - assert analysis_input_snapshot({"columns": []}, "a" * 64) is None + out = analysis_recheck(sealed, store.get) + assert out["status"] == "failed" and out["text"].startswith("复算失败：") and seen == [SNAPSHOT] + monkeypatch.setattr(analysis, "recompute", lambda result, load, *, load_result=None: None) + assert analysis_recheck(sealed, store.get)["text"] == "复算失败：复算没有交回结论" def test_formulas_substitute_the_key_numbers(): - formulas = {f["label"]: f["text"] for f in analysis_formulas(fake_result())} + store = Store() + q = store.put(SNAPSHOT) + formulas = {f["label"]: f["text"] for f in analysis_formulas(real_result(q))} assert formulas == { "总变化": "总变化 = 本期合计 − 基期合计 = 2,026 − 2,220 = -194", "变化率": "变化率 = 总变化 ÷ 基期合计 = -194 ÷ 2,220 = -8.7%", - "占比": "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ -194 = 61.9%", + "占比": "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ （-194） = 61.9%", } - baseline = fake_result("baseline", verdict="outside_band", values={ - "x": "-0.087", "median": "-0.004", "band_lo": "-0.061", "band_hi": "0.053", "z": "-2.9", "n_history": "12"}) - texts = [f["text"] for f in analysis_formulas(baseline)] - assert texts == ["常规波动范围 = 中位数 ± 2 倍稳健标准差 = -0.061 至 0.053（中位数 -0.004，过去 12 期）", - "稳健 z = （本期 − 中位数）÷ 稳健标准差 = （-0.087 − -0.004）÷ 稳健标准差 = -2.9"] assert analysis_formulas(fake_result(status="refused", reason_code="history_short")) == [] assert analysis_formulas(fake_result(values={})) == [] +def test_formulas_for_factor_ratio_and_baseline(): + store = Store() + factor_snap = {"columns": ["week", "gmv", "orders", "aov"], + "rows": [["W41", 1000, 10, 100], ["W42", 1320, 12, 110]], + "sql": "SELECT week, SUM(amount) AS gmv, COUNT(*) AS orders, ROUND(AVG(amount), 2) AS aov " + "FROM orders GROUP BY week", "source": "shop", "truncated": False} + factor = asyncio.run(analyzed({"Q1": (factor_snap, store.put(factor_snap))}, + {"op": "decompose", "mode": "factor", "params": {}}, query="Q1", period="week", + base="W41", target="W42", product="gmv", factors=["orders", "aov"], label="销售额")) + texts = {f["label"]: f["text"] for f in analysis_formulas(factor)} + assert texts["总变化"] == "总变化 = 本期合计 − 基期合计 = 1,320 − 1,000 = 320" + assert texts["占比"].startswith("orders的占比 = orders的贡献 ÷ 总变化 = 210.1449167") and texts["占比"].endswith(" = 65.7%") + + ratio_snap = {"columns": ["channel", "week", "refunds", "orders"], + "rows": [["线上", "W41", 50, 1000], ["线上", "W42", 80, 1000], ["线下", "W41", 20, 1000], + ["线下", "W42", 26, 1000]], + "sql": "SELECT channel, week, SUM(refund_flag) AS refunds, COUNT(*) AS orders FROM orders " + "GROUP BY 1, 2", "source": "shop", "truncated": False} + ratio = asyncio.run(analyzed({"Q1": (ratio_snap, store.put(ratio_snap))}, + {"op": "decompose", "mode": "ratio", "params": {}}, query="Q1", by="channel", + period="week", base="W41", target="W42", numerator="refunds", denominator="orders", + label="退款率")) + assert [f["text"] for f in analysis_formulas(ratio)] == [ + "结构效应占比 = 结构效应合计 ÷ 比率的变化 = 0 ÷ 0.018 = 0.0%", + "自身效应占比 = 自身效应合计 ÷ 比率的变化 = 0.018 ÷ 0.018 = 100.0%"] + + weeks = [f"2026-{m:02d}-{d:02d}" for m, d in ((6, 8), (6, 15), (6, 22), (6, 29), (7, 6), (7, 13), (7, 20), (7, 27), + (8, 3), (8, 10))] + series = {"columns": ["week", "orders"], "column_types": {"week": "text", "orders": "number"}, + "rows": [[w, v] for w, v in zip(weeks, [100, 102, 98, 101, 99, 103, 100, 97, 102, 130])], + "sql": "SELECT week, COUNT(*) AS orders FROM orders GROUP BY 1", "source": "shop", "truncated": False} + baseline = asyncio.run(analyzed({"Q1": (series, store.put(series))}, {"op": "baseline", "mode": None, "params": {}}, + query="Q1", period="week", measure="orders", label="订单数")) + assert baseline["params"]["basis"] == "change" and baseline["status"] == "ok" + band, z = [f["text"] for f in analysis_formulas(baseline)] + assert band == "常规波动范围 = 中位数 ± 2 倍稳健标准差 = -9.0% 至 9.0%（中位数 0.01%，过去 8 期）" + assert z == "稳健 z = （本期 − 中位数）÷ 稳健标准差 = （27.5% − 0.01%）÷ 稳健标准差 = 6.1072657902" + assert analysis_recheck(baseline, store.get)["status"] == "same" + + # -------------------------------------------------------------------------- # 血缘（D16）：分析卡的结果不是口径卡的原料 # -------------------------------------------------------------------------- diff --git a/backend/tests/test_analysis_evidence_api.py b/backend/tests/test_analysis_evidence_api.py index 800e5cd7..3c454dc9 100644 --- a/backend/tests/test_analysis_evidence_api.py +++ b/backend/tests/test_analysis_evidence_api.py @@ -1,13 +1,14 @@ """分析结论进证据链（第 1 波 W1-b）：证据接口、审计、报告导出。 运行是直接落库的虚构封存运行（虚构零售库 shop）：一次查询（Q1）、一份按地区拆的分析结论（A1，分析卡或 Agent 做的）， -报告引用判定语、具名标量、结果表的格。算子第 1 波还没落地，复算按 monkeypatch 换。 +报告引用判定语、具名标量、结果表的格。结论用 app.analysis 真算（tests.test_analysis_evidence.analyzed），出处链里的复算是 +真的 recompute：按 sha256 取回封存的输入快照、下钻的上一层结论，逐字段比。只有要它出错、升版的几处用 monkeypatch。 - 证据图：A1 条目带算子、判定；「派生自」的边 A1 → Q1 - 片段接口：判定 → 方法 → 计算 → 输入（接查询步骤）→ 复算 → 封存；Agent 写的假设标出来；复算不了照实写，不让接口出错 - 审计：分析判定、分析结论的叫法；分析说法核对的违规单列一行 - 报告导出：数据块里有分析结论那几步；文档带「分析范围」的，正文后面渲染、编号点得开 -- 指标步骤带可比性 +- 指标步骤带可比性：口径卡指标上的可比性结论（不进台账）照卡上记的 sha256 取回，点得开（判定 → … → 复算 → 封存） """ from __future__ import annotations @@ -16,21 +17,22 @@ import io import json import re import uuid +from collections.abc import Sequence from typing import Any import pytest from httpx import ASGITransport, AsyncClient from app import analysis +from app.analysis import contract from app.analysis.contract import ARTIFACT_KIND, ledger_entry -from app.analysis.testing import fake_result from app.core import artifact_store from app.core.artifact_store import manifest_hash from app.db.base import SessionLocal from app.db.models import Run, RunEvent from app.engine.evidence import build_catalog, compose_doc, iter_segments from app.main import app -from tests.test_analysis_evidence import SNAPSHOT +from tests.test_analysis_evidence import CHANNELS, SNAPSHOT, analyzed GRAPH = {"nodes": [{"id": n, "type": t, "data": {"label": label}} for n, t, label in [ ("fetch", "tool", "按地区取数"), ("an", "agent", "分析"), ("card", "metrics", "周报口径"), @@ -43,28 +45,38 @@ async def client(): yield c +def _query_entry(content: dict[str, Any], artifact: str) -> dict[str, Any]: + return {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": artifact, "tool": "db_query__shop", + "source": "shop", "columns": content["columns"], "rows": len(content["rows"]), "truncated": False} + + async def sealed_run(text: str, *, item: str | None = "drop", call_id: str | None = None, result: dict[str, Any] | None = None, doc_extra: dict[str, Any] | None = None, - card: dict[str, Any] | None = None) -> dict[str, Any]: - """落一次封存好的运行：查询快照、分析结论、（可选）口径卡、报告文档都按真实的工件写，事件按 runner 的算法封存。""" + card: dict[str, Any] | None = None, queries: Sequence[dict[str, Any]] = (), + parents: Sequence[dict[str, Any]] = ()) -> dict[str, Any]: + """落一次封存好的运行：查询快照（SNAPSHOT，再加 queries）、分析结论（parents 在前，result 在后；result 缺省按 SNAPSHOT + 真算一份按地区拆的可加拆解）、（可选）口径卡、报告文档都按真实的工件写，事件按 runner 的算法封存。""" run_id = uuid.uuid4().hex q = await artifact_store.put_json(SNAPSHOT, kind="query_snapshot", run_id=run_id, node_id="fetch") - res = result or fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", - "columns": ["region", "week", "gmv"], "rows": "all"}]) + more = [await artifact_store.put_json(c, kind="query_snapshot", run_id=run_id, node_id="fetch") for c in queries] + res = result or await analyzed({"Q2": (SNAPSHOT, q)}) + qentries = [_query_entry(SNAPSHOT, q), *(_query_entry(c, a) for c, a in zip(queries, more))] + aentries = [] + for i, parent in enumerate(parents): + found = await artifact_store.put_json(parent, kind=ARTIFACT_KIND, run_id=run_id, node_id="an") + aentries.append(ledger_entry(parent, node_id="an", exec_no=1, artifact=found, item=f"up{i}")) a = await artifact_store.put_json(res, kind=ARTIFACT_KIND, run_id=run_id, node_id="an") - qentry = {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": q, "tool": "db_query__shop", - "source": "shop", "columns": SNAPSHOT["columns"], "rows": len(SNAPSHOT["rows"]), "truncated": False} - aentry = ledger_entry(res, node_id="an", exec_no=1, artifact=a, item=item, call_id=call_id) + aentries.append(ledger_entry(res, node_id="an", exec_no=1, artifact=a, item=item, call_id=call_id)) nodes: dict[str, Any] = {} events: list[tuple[int, str, str | None, dict[str, Any]]] = [ - (1, "run.started", None, {"input": {}}), (2, "node.finished", "fetch", {"evidence": [qentry]}), - (3, "node.finished", "an", {"evidence": [aentry]})] + (1, "run.started", None, {"input": {}}), (2, "node.finished", "fetch", {"evidence": qentries}), + (3, "node.finished", "an", {"evidence": aentries})] if card is not None: card_id = await artifact_store.put_json(card, kind="metric_set", run_id=run_id, node_id="card") nodes["card"] = {**card, "artifact": card_id} events.append((4, "node.finished", "card", {"evidence": [{"kind": "metric_set", "node_id": "card", "exec": 1, "artifact": card_id}]})) - catalog = build_catalog(nodes=nodes, ledger=[qentry, aentry]) + catalog = build_catalog(nodes=nodes, ledger=[*qentries, *aentries]) doc = compose_doc(text, catalog, node_id="write", run_id=run_id) doc.update(doc_extra or {}) doc_id = await artifact_store.put_json(doc, kind="report_doc", run_id=run_id, node_id="write") @@ -78,7 +90,7 @@ async def sealed_run(text: str, *, item: str | None = "drop", call_id: str | Non for s, etype, node, data in events: session.add(RunEvent(run_id=run_id, seq=s, type=etype, node_id=node, ts=float(s), data=data)) await session.commit() - return {"run": run_id, "doc": doc, "query": q, "analysis": a, "result": res} + return {"run": run_id, "doc": doc, "query": q, "analysis": a, "result": res, "queries": more} def seg_id(doc: dict[str, Any], ref: str) -> str: @@ -118,11 +130,13 @@ async def test_the_phrase_opens_judgment_method_calculation_inputs_recompute_and assert [s["step"] for s in chain] == ["analysis", "analysis_method", "analysis_calc", "analysis_input", "query", "analysis_recompute", "analysis_seal"] judged, method, calc, given, query, again, seal = chain - assert judged["phrase"] == made["result"]["phrase"] and judged["criterion"] == "一段占总变化一半以上算主要来源" + assert judged["phrase"] == made["result"]["phrase"] == "销售额下降 194（-8.7%），主要来自华东（-120，占 61.9%）" + assert judged["criterion"].startswith("一段占总变化的一半以上算主要来源") assert judged["verdict"] == {"code": "driver", "subjects": [["华东"]], "direction": "down"} assert judged["cited"] == {"what": "phrase"} and judged["render_ok"] is True and judged["eid_ok"] is True assert judged["origin"] == "card" and judged["item"] == "drop" and judged["label"] == "分析" - assert judged["hypothesis"] == "地区拖累" and judged["hypothesis_note"].startswith("Agent 写的假设") + assert judged["hypothesis"] == "华东拖累" and judged["hypothesis_note"].startswith("Agent 写的假设") + assert {c["code"] for c in judged["caveats"]} == {"measure_unlabeled", "headline_basis"} assert method["method"] == "decompose.additive@1" and method["method_label"] == "变化拆解（可加）" assert {s["key"]: s["label"] for s in method["settings"]}["main_share"] == "「主要来源」的占比" assert {s["key"] for s in method["settings"]} >= {"main_share", "concentration", "table_top_k"} @@ -130,28 +144,30 @@ async def test_the_phrase_opens_judgment_method_calculation_inputs_recompute_and assert calc["table"]["columns"][0] == "region" and calc["table"]["labels"]["share"] == "占变化的比重" assert calc["table"]["total_rows"] == 4 and calc["table"]["rows"][0][:2] == ["华东", "820"] assert {f["label"]: f["text"] for f in calc["formulas"]}["占比"] == \ - "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ -194 = 61.9%" + "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ （-194） = 61.9%" assert given == {"step": "analysis_input", "of": "A1", "ref": "Q2", "role": "data", "role_label": "数据", "artifact": made["query"], "columns": ["region", "week", "gmv"], "rows": "all", "query": "Q1", "node_id": "fetch", "sealed": True} assert query["alias"] == "Q1" and query["input_of"] == "A1" and query["total_rows"] == 8 assert query["highlight"]["cols"] == ["region", "week", "gmv"] - assert again["status"] == "failed" and again["text"] == "复算失败：系统里还没有这种分析的复算实现" + # 真的复算：按 sha256 取回封存的输入快照，逐字段和封存的比 + assert again == {"step": "analysis_recompute", "alias": "A1", "status": "same", "method": "decompose.additive@1", + "text": "复算一致：按封存的输入重新计算，结论逐项相同"} assert seal["sealed"] is True and seal["ledger"] is True and seal["hash_ok"] is True assert body["seal"]["covered"] is True assert body["note"].startswith("判定语由系统按平台的方法算出判定") assert body["segment"]["kind"] == "analysis" -async def test_named_values_and_result_cells_point_at_what_they_cite(client, monkeypatch): - monkeypatch.setattr(analysis, "recompute", lambda result, load, *, load_result=None: dict(result)) +async def test_named_values_and_result_cells_point_at_what_they_cite(client): made = await sealed_run(TEXT) value = (await client.get( f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1.top_share|pct')}")).json() - assert value["chain"][0]["cited"] == {"what": "value", "key": "top_share", "raw": "0.6186", "conv": "pct"} + assert value["chain"][0]["cited"] == {"what": "value", "key": "top_share", "raw": "0.618556701031", "conv": "pct"} + assert value["segment"]["text"] == "61.9%" and value["chain"][0]["render_ok"] is True calc = value["chain"][2] assert next(v for v in calc["values"] if v["key"] == "top_share") == { - "key": "top_share", "raw": "0.6186", "rendered": "0.6186", "cited": True} + "key": "top_share", "raw": "0.618556701031", "rendered": "0.618556701", "cited": True} assert value["chain"][-2]["text"] == "复算一致：按封存的输入重新计算，结论逐项相同" assert value["note"] == "这个数取自分析结论的具名的数，由系统按固定规则显示，未经模型转写" @@ -163,20 +179,41 @@ async def test_named_values_and_result_cells_point_at_what_they_cite(client, mon async def test_the_recompute_gets_the_sealed_inputs_by_hash(client, monkeypatch): + """复算的 load 交工件原内容（不加出处、不改列的类型），只认封存范围内交回过的查询；别的一律当取不回。""" seen: dict[str, Any] = {} + real = analysis.recompute def recompute(result, load, *, load_result=None): seen["input"] = load(result["inputs"][0]["artifact"]) seen["missing"] = load("0" * 64) - return {**result, "method": "decompose.additive@2"} + return real(result, load, load_result=load_result) monkeypatch.setattr(analysis, "recompute", recompute) made = await sealed_run(TEXT) body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() again = next(s for s in body["chain"] if s["step"] == "analysis_recompute") - assert again["text"] == "运行时用的是 decompose.additive@1，现在是 decompose.additive@2，复算结果不同" - assert seen["input"]["rows"] == SNAPSHOT["rows"] and seen["input"]["node_id"] == "fetch" - assert seen["missing"] is None + assert again["status"] == "same" and seen["input"] == SNAPSHOT and seen["missing"] is None + + +async def test_an_input_outside_the_sealed_scope_is_not_recomputed_from(client): + """结论记着的输入快照不是这次运行交回过的（台账里没有）：复算照实说取不回，不去工件库里捞。""" + stray = {**SNAPSHOT, "sql": SNAPSHOT["sql"] + " -- 另一次"} + s_id = await artifact_store.put_json(stray, kind="query_snapshot") + made = await sealed_run(TEXT, result=await analyzed({"Q2": (stray, s_id)})) + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() + again = next(s for s in body["chain"] if s["step"] == "analysis_recompute") + assert again["status"] == "failed" and again["text"] == "复算失败：取不回 Q2 的查询快照" + given = next(s for s in body["chain"] if s["step"] == "analysis_input") + assert given["sealed"] is False and given["query"] is None + + +async def test_an_upgraded_method_names_both_versions(client, monkeypatch): + made = await sealed_run(TEXT) + monkeypatch.setitem(contract.METHODS, ("decompose", "additive"), "decompose.additive@2") + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() + again = next(s for s in body["chain"] if s["step"] == "analysis_recompute") + assert again["method"] == "decompose.additive@1" and again["method_now"] == "decompose.additive@2" + assert again["text"] == "运行时用的是 decompose.additive@1，现在是 decompose.additive@2，复算结果相同" async def test_recompute_errors_do_not_break_the_panel(client, monkeypatch): @@ -194,7 +231,8 @@ async def test_recompute_errors_do_not_break_the_panel(client, monkeypatch): async def test_a_tampered_result_is_said_and_not_shown(client): # 工件按内容寻址：改的是这一次独有的那份结论（别的测试的结论内容相同、是同一个文件） q = await artifact_store.put_json(SNAPSHOT, kind="query_snapshot") - own = fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data"}], criterion=f"只给这一次：{uuid.uuid4().hex}") + own = await analyzed({"Q2": (SNAPSHOT, q)}, hypothesis=f"只给这一次：{uuid.uuid4().hex}") + assert own["hypothesis"].startswith("只给这一次：") made = await sealed_run(TEXT, result=own) path = artifact_store._path_of(made["analysis"]) path.write_text(path.read_text("utf-8").replace("华东", "华南"), "utf-8") @@ -210,27 +248,97 @@ async def test_agent_analyses_are_marked_as_such(client): assert body["chain"][0]["origin"] == "agent" and body["chain"][0]["call_id"] == "call-9" -async def test_headline_inputs_show_the_closure_and_drill_downs_the_parent(client): - base = fake_result() - made = await sealed_run(TEXT) - q, a = made["query"], made["analysis"] - child = fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", "columns": ["region", "gmv"], - "rows": [0, 1]}, - {"ref": "Q2.r1.gmv", "artifact": q, "role": "headline_target", "row": 1, - "column": "gmv", "value": "2026"}, - {"ref": "A3", "artifact": a, "role": "parent"}], - values={**base["values"], "target_total": "2026"}) - drilled = await sealed_run("[[a:A1]]。", result=child) - body = (await client.get(f"/api/runs/{drilled['run']}/evidence/segments/{seg_id(drilled['doc'], 'a:A1')}")).json() +async def test_headline_inputs_show_the_closure(client): + """给了总数（headline）的拆解：两格各是一份输入，和结论里的总数闭合；复算照样按 sha256 取回两份快照。""" + totals = {"columns": ["gmv_base", "gmv"], "column_types": {"gmv_base": "number", "gmv": "number"}, + "rows": [[2220, 2026]], "sql": "SELECT 2220 AS gmv_base, 2026 AS gmv", "source": "shop", + "truncated": False} + t = artifact_store.content_hash(artifact_store.canonical_json(totals)) + q = artifact_store.content_hash(artifact_store.canonical_json(SNAPSHOT)) + result = await analyzed({"Q2": (SNAPSHOT, q), "Q5": (totals, t)}, headline=["Q5.r0.gmv_base", "Q5.r0.gmv"]) + made = await sealed_run("[[a:A1]]。", result=result, queries=[totals]) + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() inputs = [s for s in body["chain"] if s["step"] == "analysis_input"] - assert [s["role"] for s in inputs] == ["data", "headline_target", "parent"] + assert [s["role"] for s in inputs] == ["headline_base", "headline_target", "data"] + assert inputs[0]["closure"] == {"total_key": "base_total", "total": "2,220", "headline": "2,220", "closed": True} assert inputs[1]["closure"] == {"total_key": "target_total", "total": "2,026", "headline": "2,026", "closed": True} - assert inputs[2]["phrase"] == base["phrase"] and inputs[2]["role_label"] == "上一层拆解" - query = next(s for s in body["chain"] if s["step"] == "query") - assert [0, "region"] in query["highlight"]["cells"] and [1, "gmv"] in query["highlight"]["cells"] + assert inputs[1]["query"] == "Q2" and inputs[1]["sealed"] is True and inputs[2]["query"] == "Q1" + assert next(s for s in body["chain"] if s["step"] == "analysis_recompute")["status"] == "same" -async def test_the_metric_step_shows_the_comparability(client): +async def test_a_drill_down_points_at_its_parent_and_recomputes_from_it(client): + """下钻（华东按渠道再拆）：输入里有上一层结论（parent），落到报告目录的 A1；复算按 sha256 取回台账记过的上一层。""" + q = artifact_store.content_hash(artifact_store.canonical_json(SNAPSHOT)) + c = artifact_store.content_hash(artifact_store.canonical_json(CHANNELS)) + top = await analyzed({"Q2": (SNAPSHOT, q)}) + child = await analyzed({"Q3": (CHANNELS, c)}, parents={"A1": top}, query="Q3", by=["channel"], + parent={"of": "A1", "key": ["华东"]}) + made = await sealed_run("[[a:A2]]。", result=child, parents=[top], queries=[CHANNELS]) + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A2')}")).json() + inputs = [s for s in body["chain"] if s["step"] == "analysis_input"] + assert [s["role"] for s in inputs] == ["parent", "data"] + assert inputs[0]["alias"] == "A1" and inputs[0]["phrase"] == top["phrase"] and inputs[0]["sealed"] is True + assert inputs[0]["role_label"] == "上一层拆解" and inputs[1]["query"] == "Q2" + again = next(s for s in body["chain"] if s["step"] == "analysis_recompute") + assert again["status"] == "same", again + graph = (await client.get(f"/api/runs/{made['run']}/evidence")).json() + edges = {(e["from"], e["to"], e["rel"]) for e in graph["edges"]} + assert {("A2", "A1", "derived_from"), ("A2", "Q2", "derived_from"), ("A1", "Q1", "derived_from")} <= edges + + +#: 本周、上周两份同形查询（只差日期）：口径卡跨这两份算环比时自动做可比性检查 +WINDOW = "SELECT SUM(gmv) AS gmv FROM orders WHERE order_date >= '{a}' AND order_date < '{b}'" +LAST_WEEK = {"columns": ["gmv"], "column_types": {"gmv": "number"}, "rows": [[2220]], + "sql": WINDOW.format(a="2026-10-05", b="2026-10-12"), "source": "shop", "truncated": False} +THIS_WEEK = {**LAST_WEEK, "rows": [[2026]], "sql": WINDOW.format(a="2026-10-12", b="2026-10-19")} + + +async def comparability_card() -> tuple[dict[str, Any], dict[str, Any]]: + """一张跨两周算环比的口径卡，指标上记着口径卡执行时自动做的可比性检查（结论落成工件，不进台账）：(卡, 可比性结论)。""" + last = artifact_store.content_hash(artifact_store.canonical_json(LAST_WEEK)) + this = artifact_store.content_hash(artifact_store.canonical_json(THIS_WEEK)) + result = await analyzed({"fetch·base": (LAST_WEEK, last), "fetch·target": (THIS_WEEK, this)}, + {"op": "comparable", "mode": None, "params": {}}, + base_query="fetch·base", target_query="fetch·target") + found = await artifact_store.put_json(result, kind=ARTIFACT_KIND, node_id="card") + verdict = result["verdict"]["code"] + card = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v1", "metrics": [ + {"id": "wow", "name": "环比", "value": -0.0874, "unit": "", + "expression": "cell(nodes.fetch, 0, 'gmv') / cell(nodes.fetch, 0, 'gmv') - 1", + "substituted": "2026 / 2220 - 1", "recompute_ok": True, + "inputs": [{"path": "cell(nodes.fetch, 0, 'gmv')", "value": 2026, "node_id": "fetch", "via": "tool_cell", + "artifact": this, "locator": {"row": 0, "column": "gmv"}}, + {"path": "cell(nodes.fetch, 0, 'gmv')", "value": 2220, "node_id": "fetch", "via": "tool_cell", + "artifact": last, "locator": {"row": 0, "column": "gmv"}}], + "comparability": {"artifact": found, "verdict": verdict, "phrase": result["phrase"], "reasons": []}}]} + return card, result + + +async def test_the_metric_step_opens_the_comparability_result(client): + """可比性结论不进台账、只从口径卡的指标集里引用：照卡上记的 sha256 取回，随卡封存；点得开它的整条出处链，复算是真的。""" + card, result = await comparability_card() + made = await sealed_run("环比 [[m:wow]]。", card=card, queries=[LAST_WEEK, THIS_WEEK]) + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'm:wow')}")).json() + view = body["chain"][0]["comparability"] + assert view["verdict"] == result["verdict"]["code"] and view["phrase"] == result["phrase"] + assert view["sealed"] is True and view["hash_ok"] is True and view["method"] == "comparable@1" + steps = view["steps"] + assert [s["step"] for s in steps] == ["analysis", "analysis_method", "analysis_calc", "analysis_input", + "analysis_input", "analysis_recompute", "analysis_seal"] + judged, method, calc, base, target, again, seal = steps + assert judged["origin"] == "metric" and judged["cited"] == {"what": "phrase"} and judged["render_ok"] is True + assert judged["method_label"] == "可比性检查" and judged["node_id"] == "card" + assert [r[0] for r in calc["table"]["rows"]] == ["同口径", "窗口长度", "整期", "星期构成", "数据问题"] + assert (base["role"], target["role"]) == ("base", "target") + assert base["query"] == "Q2" and target["query"] == "Q3" and base["sealed"] is True + assert again["status"] == "same" and again["text"] == "复算一致：按封存的输入重新计算，结论逐项相同" + assert seal["via"] == "metric_set" and seal["ledger"] is False and seal["sealed"] is True + assert seal["note"].startswith("这份可比性结论不进台账") + # 两份输入快照的查询步骤在指标链里（指标的输入就是这两份），可比性的链里不再给 + assert [s["step"] for s in body["chain"]].count("query") == 2 and "query" not in [s["step"] for s in steps] + + +async def test_a_card_without_the_comparability_artifact_says_only_what_it_recorded(client): card = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v1", "metrics": [ {"id": "wow", "name": "环比", "value": -0.087, "unit": "", "expression": "1", "inputs": [], "comparability": {"artifact": "", "verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据", @@ -242,6 +350,15 @@ async def test_the_metric_step_shows_the_comparability(client): "artifact": None, "hash_ok": None, "sealed": False} +async def test_the_app_metric_endpoint_opens_the_comparability_too(client): + card, _ = await comparability_card() + made = await sealed_run("环比 [[m:wow]]。", card=card, queries=[LAST_WEEK, THIS_WEEK]) + body = (await client.get(f"/api/runs/{made['run']}/evidence/metrics/card/wow")).json() + view = body["chain"][0]["comparability"] + assert view["sealed"] is True and view["steps"][-2]["status"] == "same" + assert [s["query"] for s in view["steps"] if s["step"] == "analysis_input"] == [None, None] + + # -------------------------------------------------------------------------- # 审计 # -------------------------------------------------------------------------- @@ -286,7 +403,7 @@ def test_audit_labels_cover_the_new_violation_codes(): # -------------------------------------------------------------------------- SCOPE = {"analysis_scope": { - "phenomena": [{"alias": "A1", "op": "decompose", "phrase": "销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）"}], + "phenomena": [{"alias": "A1", "op": "decompose", "phrase": "销售额下降 194（-8.7%），主要来自华东（-120，占 61.9%）"}], "decomposed": [{"alias": "A1", "op": "decompose", "mode": "additive", "dims": ["地区"], "verdict": "driver", "phrase": "主要来自华东"}], "excluded": [], "refused": [{"alias": "A1", "op": "baseline", "reason": "历史只有 5 期，至少要 8 期"}], diff --git a/backend/tests/test_analysis_judge_ledger.py b/backend/tests/test_analysis_judge_ledger.py index c3db6bc5..c7fb4360 100644 --- a/backend/tests/test_analysis_judge_ledger.py +++ b/backend/tests/test_analysis_judge_ledger.py @@ -97,3 +97,21 @@ def test_a_report_citing_only_analyses_has_nothing_to_label(): assert ledger_labels.label_doc(doc, {}, {}, lambda _: None) == {} assert ledger_labels.referenced_columns([doc], lambda _: None) == {} assert ledger_labels.whole_of({}, {})["reason"] == "报告里没有可标口径的数" + + +def test_the_judge_and_the_writer_see_the_comparability_of_a_metric(): + """口径卡指标上的可比性（口径卡执行时自动做的检查）：裁判的摘录里原样给判定语；写作目录没记原因的照判定语写，不重复判定。""" + from app.core.artifact_store import canonical_json, content_hash + from app.engine.evidence import writer_catalog + + card = {"kind": "metric_set", "caliber": "周报", "caliber_version": "v1", "metrics": [ + {"id": "wow", "name": "环比", "value": -0.087, "unit": "", "expression": "1", "inputs": [], + "comparability": {"artifact": "d" * 64, "verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据", + "reasons": []}}]} + sha = content_hash(canonical_json(card)) + catalog = build_catalog(nodes={"card": {**card, "artifact": sha}}, ledger=[]) + assert "环比 = -0.087（两期不可比：本期只有 3 天数据。这个比较不能当结论" in writer_catalog(catalog).text + doc = compose_doc("环比 [[m:wow]]，需求走弱。", catalog) + request = judging.prepare(doc, catalog, loader={sha: card}.get) + [excerpt] = request.excerpts.values() + assert "两期可比性（系统检查）：两期不可比：本期只有 3 天数据" in excerpt diff --git a/backend/tests/test_analysis_pages.py b/backend/tests/test_analysis_pages.py index bd2be83a..94e720de 100644 --- a/backend/tests/test_analysis_pages.py +++ b/backend/tests/test_analysis_pages.py @@ -19,8 +19,8 @@ from httpx import ASGITransport, AsyncClient import app.bundle import app.report_export from app.main import app as asgi -from tests.test_analysis_bundle import bundle_of -from tests.test_analysis_evidence_api import SCOPE, TEXT, sealed_run, seg_id +from tests.test_analysis_bundle import bundle_of, card_bundle +from tests.test_analysis_evidence_api import LAST_WEEK, SCOPE, TEXT, THIS_WEEK, comparability_card, sealed_run, seg_id NODE = shutil.which("node") pytestmark = pytest.mark.skipif(NODE is None, reason="没有 node，无法执行页面上的脚本") @@ -129,15 +129,27 @@ async def test_the_export_page_opens_the_analysis_steps_and_the_scope_numbers(): segs = [{"seg": phrase, "kind": "analysis", "state": "deterministic", "text": made["result"]["phrase"]}, {"seg": cell, "kind": "number", "state": "deterministic", "text": "-120"}, {"seg": "scope:A1", "kind": "analysis", "state": "deterministic", "text": "A1"}] - out = run_node(RUN_RUNTIME, RUNTIME, {"data": json.loads(json.dumps(data)), "segs": segs}) + # 现查结果的三种写法：按运行当天现算的界、一列一个值都没有、表的范围（结论里的 probes 原样交给页面） + parsed = json.loads(data) + method = next(s for s in parsed["chains"][parsed["refs"][phrase]] if s["step"] == "analysis_method") + method["probes"] = [ + {"source": "shop", "expr": "date('now', '-7 days')", "value": "2026-10-14", "basis": "live", + "at": "2026-10-21T09:00:00"}, + {"source": "shop", "table": "orders", "column": "refund_date", "min": None, "max": None, "basis": "live", + "at": "2026-10-21T09:00:00"}, + {"source": "shop", "table": "orders", "column": "order_date", "min": "2025-01-01", "max": "2026-10-20", + "basis": "profile", "at": "2026-10-21T09:00:00"}] + out = run_node(RUN_RUNTIME, RUNTIME, {"data": json.dumps(parsed, ensure_ascii=False), "segs": segs}) opened = out[phrase] assert opened["steps"] == ["analysis", "analysis_method", "analysis_calc", "analysis_input", "query", "analysis_recompute", "analysis_seal"] assert opened["title"].startswith("判定「销售额下降") body = opened["body"] - for text in ("判定标准一段占总变化一半以上算主要来源", "Agent 写的假设地区拖累", "实际用的阈值", "「主要来源」的占比", - "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ -194 = 61.9%", "结果表（列出 4 行，共 4 行）", - "复算失败：系统里还没有这种分析的复算实现", "封存范围内的台账记过这份分析结论"): + for text in ("判定标准一段占总变化的一半以上算主要来源", "Agent 写的假设华东拖累", "实际用的阈值", "「主要来源」的占比", + "华东的占比 = 华东的变化 ÷ 总变化 = -120 ÷ （-194） = 61.9%", "结果表（列出 4 行，共 4 行）", + "复算一致：按封存的输入重新计算，结论逐项相同", "封存范围内的台账记过这份分析结论", + "按运行当天现算的界 date('now', '-7 days') = 2026-10-14（现查 · 2026-10-21T09:00:00）", + "orders.refund_date：这一列一个值都没有（现查", "orders.order_date：2025-01-01 至 2026-10-20（剖析"): assert text in body, text assert "结果表第 1 行「delta」" in out[cell]["body"] assert out["scope:A1"]["steps"] == opened["steps"] @@ -151,10 +163,40 @@ async def test_the_bundle_page_shows_the_analysis_detail_and_the_scope(): assert scope.startswith("分析范围") and "现象A1：销售额下降" in scope and "拆过A1（按地区）：主要来自华东" in scope assert "没拆渠道" in scope and "这次做了 2 次差异检验" in scope phrase = next(text for text in out["opened"].values() if "出处分析结论 A1 的判定语" in text) - assert "核对结果✓ 通过" in phrase and "分析结论里的值「销售额下降 194 万" in phrase - assert "判定语销售额下降 194 万" in phrase and "方法与版本decompose.additive@1" in phrase - assert "复算失败：系统里还没有这种分析的复算实现（导出时服务端按封存的输入复算的结论；本页只核哈希" in phrase + assert "核对结果✓ 通过" in phrase and "分析结论里的值「销售额下降 194（-8.7%）" in phrase + assert "判定语销售额下降 194（-8.7%）" in phrase and "方法与版本decompose.additive@1" in phrase + assert "输入数据 · Q1" in phrase # 输入的角色写叫法，不写 data + assert "复算一致：按封存的输入重新计算，结论逐项相同（导出时服务端按封存的输入复算的结论；本页只核哈希" in phrase value = next(text for text in out["opened"].values() if "具名的数「top_share」" in text) - assert "引用的值0.6186" in value and "正文显示61.9%（与按显示规则渲染的结果一致）" in value + assert "引用的值0.618556701031" in value and "正文显示61.9%（与按显示规则渲染的结果一致）" in value scope_item = next(text for key, text in out["opened"].items() if key.endswith("|scope:A1")) assert "核对结果✓ 通过" in scope_item and "正文显示" not in scope_item + + +async def test_the_export_page_opens_the_comparability_on_a_metric(): + """口径卡指标上的可比性：指标那一步里写判定，点得开那份可比性结论（同分析结论的几步，收在一个可展开的块里）。""" + card, result = await comparability_card() + made = await sealed_run("环比 [[m:wow]]。", card=card, queries=[LAST_WEEK, THIS_WEEK]) + async with AsyncClient(transport=ASGITransport(app=asgi), base_url="http://test") as client: + html = (await client.get(f"/api/runs/{made['run']}/report.html")).text + data = re.search(r'', html, re.S).group(1) + wow = seg_id(made["doc"], "m:wow") + out = run_node(RUN_RUNTIME, RUNTIME, {"data": data, "segs": [ + {"seg": wow, "kind": "number", "state": "deterministic", "text": "-0.0874"}]}) + opened = out[wow] + assert opened["steps"][:1] == ["metric"] + assert opened["steps"][opened["steps"].index("analysis"):] [:3] == ["analysis", "analysis_method", "analysis_calc"] + for text in ("两期可比性", "展开这份可比性结论", "可比性检查", "口径卡执行时自动做的可比性检查", result["phrase"], + "同口径", "复算一致：按封存的输入重新计算，结论逐项相同", "记在口径卡的指标集里，随卡封存（不进台账）"): + assert text in opened["body"], text + + +def test_the_bundle_page_shows_the_comparability_in_the_metric_detail(): + bundle, _, _, _ = card_bundle() + out = run_node(RUN_VERIFIER, VERIFIER, {"data": json.dumps(bundle), "seal": bundle["meta"]["manifest_hash"]}) + assert out["verified"] == "" + [detail] = out["opened"].values() + for text in ("两期可比性", "判定两期可比", "判定标准只差日期", "方法与版本comparable@1", "各项检查", "同口径", + "复算一致：按封存的输入重新计算，结论逐项相同（导出时服务端按封存的输入复算的结论"): + assert text in detail, text +