diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py
index 1a88fc7d..590d14a9 100644
--- a/backend/app/api/evidence.py
+++ b/backend/app/api/evidence.py
@@ -9,6 +9,7 @@
- 运行输入：入口节点 node.finished 事件里的 node_output 工件
- 表结构快照：tool.end 事件里的 schema_artifact，或者 node.finished.evidence 里的 schema 条目
- 检索快照：retrieve.end 事件里的 artifact，或者 node.finished.evidence 里的 retrieval 条目
+- 分析结论：node.finished.evidence 里的 analysis 条目（Agent 分析工具、分析卡记的），它的输入快照照查询快照认
artifacts 表可以事后插行，run.output 是可以改写的一列，封存之后追加的事件不在
核对范围里——这三样都不当来源。解析数字层的链：口径卡指标 → 它的输入 → 查询快照里
@@ -70,6 +71,7 @@ from fastapi.responses import JSONResponse, Response
from sqlalchemy import select, update
from sqlalchemy.exc import StatementError
+from app.analysis.contract import CATALOG_PREFIX, LABELS as ANALYSIS_LABELS, LEDGER_KIND, is_result, method_label
from app.api.coded import CodedHTTPException
from app.core.artifact_store import load
from app.core.events import EventType
@@ -109,6 +111,10 @@ from app.engine.evidence import (
UNKNOWN_ENTITY_REASON,
UNVERIFIED_ENTITY_REASON,
RenderError,
+ analysis_card,
+ analysis_formulas,
+ analysis_input_snapshot,
+ analysis_recheck,
build_catalog,
closest_entities,
find_segment,
@@ -119,6 +125,7 @@ from app.engine.evidence import (
make_eid,
normalize_quote,
query_entry_fields,
+ render_cell,
render_metric,
schema_column,
schema_table,
@@ -127,6 +134,7 @@ from app.engine.evidence import (
uncited_claims,
withhold_metrics,
)
+from app.engine.expressions import locate_cell
from app.engine.judge import FIELDS_MAX, SETTLED, UNSUPPORTED, VERDICTS, candidates, outdated
from app.engine.lineage import untraced
from app.engine.merge_query import MERGE_SOURCE, trace_cell
@@ -196,6 +204,8 @@ class _Sealed:
schemas: dict[str, dict[str, Any]] = field(default_factory=dict)
#: 封存范围内交回过的检索快照：工件 id → {node_id, source}
retrievals: dict[str, dict[str, Any]] = field(default_factory=dict)
+ #: 封存范围内台账里记过的分析结论：工件 id → {node_id, exec, call_id / item, inputs}
+ analyses: dict[str, dict[str, Any]] = field(default_factory=dict)
#: 封存时记下的清单哈希（没封存的是 None）：旧答案的猜测按 (run_id, 它) 缓存
manifest_hash: str | None = None
#: 运行记录上的 manifest_seq 原值（老运行是 None）：按需裁判追加判定前核对封存没换过
@@ -304,6 +314,10 @@ async def _sealed(run_id: str) -> _Sealed:
out.schemas[artifact] = {"node_id": node_id, "source": entry.get("source")}
elif entry.get("kind") == "retrieval":
out.retrievals[artifact] = {"node_id": node_id, "source": entry.get("source")}
+ elif entry.get("kind") == LEDGER_KIND:
+ out.analyses.setdefault(artifact, {
+ "node_id": node_id, **{k: entry[k] for k in ("exec", "call_id", "item", "inputs")
+ if entry.get(k) is not None}})
return out
@@ -517,6 +531,14 @@ def _graph_of(report: _Report, sealed: _Sealed, inputs: dict[str, Any]) -> tuple
"report": report.node_id}
for source in _merge_inputs(entry.get("artifact"))
if (target := _query_alias(catalog, source.get("artifact"))))
+ elif entry.get("kind") == LEDGER_KIND:
+ # 分析结论：哪种分析、算成没有、判定码、谁做的（分析卡的那一项 / Agent 的那次调用）；连向它的输入（「派生自」：
+ # 输入快照的 Q…，下钻的上一层 A…），图上看得出 A3 是从 Q2 算出来的
+ item.update({k: entry[k] for k in ("op", "mode", "status", "verdict", "item", "call_id")
+ if entry.get(k) is not None})
+ edges.extend({"from": alias, "to": target, "rel": "derived_from", "report": report.node_id}
+ for target in dict.fromkeys(_catalog_alias_of(catalog, a) for a in entry.get("inputs") or [])
+ if target and target != alias)
elif entry.get("kind") in ENTITY_KINDS:
# 表和字段：叫什么、出现在哪几次查询里、来历（表结构快照 / SQL / 结果列）各自封存了没有
item.update({k: entry[k] for k in ("name", "table", "tables", "qualified", "source") if entry.get(k)})
@@ -934,6 +956,9 @@ async def _chain(seg: dict[str, Any], doc: dict[str, Any], sealed: _Sealed,
entry, doc, sealed, masks)
if seg.get("state") != "deterministic" or cite.get("status") != "resolved":
return []
+ if cite.get("kind") == LEDGER_KIND or (cite.get("kind") == "cell" and entry.get("kind") == LEDGER_KIND):
+ # 分析结论：判定语、具名标量、结果表的一格都接同一条链（判定 → 方法 → 计算 → 输入 → 复算 → 封存）
+ return await _analysis_chain(seg, cite, entry, doc, sealed, masks)
if cite.get("kind") == "cell":
locator = cite.get("locator") or {}
# 只有直接引用的单元格才可能推断来源（P4-SPEC 2.8.2）。指标链里的查询步骤不带：指标片段的推断来源只会
@@ -953,6 +978,259 @@ async def _chain(seg: dict[str, Any], doc: dict[str, Any], sealed: _Sealed,
return await _metric_chain(seg, cite, entry, doc, sealed, masks)
+# --------------------------------------------------------------------------
+# 分析结论的出处链
+# --------------------------------------------------------------------------
+
+#: Agent 写的假设（hypothesis）旁边那句：它不是结论，只在证据面板显示（细案 01 §7.3）
+HYPOTHESIS_NOTE = "Agent 写的假设：说明这次在验证什么，不是分析结论，也不进报告正文"
+UNSEALED_ANALYSIS = "这份分析结论不属于封存范围内的任何事件（封存前的台账里没有它），不能作为证据展示"
+TAMPERED_ANALYSIS = "分析结论与哈希不一致，疑似被修改，不展示其中的判定和数"
+MISSING_ANALYSIS = "无法读取分析结论"
+#: 每个算子用到的阈值（细案 01 §10「用在」一栏）：方法一步只列这几个，判定标准照它们写
+_OP_SETTINGS: dict[str, tuple[str, ...]] = {
+ "decompose": ("main_share", "concentration", "offset_ratio", "flat_pct", "min_denominator", "max_drill_depth",
+ "table_top_k"),
+ "baseline": ("k_warn", "k_alert", "min_history", "max_history"),
+ "trend": ("trend_min_points", "trend_alpha", "min_history", "max_history"),
+ "differ": ("differ_alpha", "min_n", "min_expected"),
+ "comparable": (),
+}
+#: 输入的角色（analysis.contract.AnalysisInput.role）给人看的叫法
+_INPUT_ROLES = {"data": "数据", "base": "基期", "target": "本期", "headline_base": "基期总数",
+ "headline_target": "本期总数", "parent": "上一层拆解"}
+#: 闭合核对：headline 的那一格对着结论里的哪个总数（细案 01 §2.1）
+_HEADLINE_TOTALS = {"headline_base": "base_total", "headline_target": "target_total"}
+#: 计算一步最多给结果表的几行（前 table_top_k 段，加上被引用的那一行前后各 WINDOW 行）
+MAX_ANALYSIS_ROWS = MAX_WINDOW_ROWS
+
+
+def _analysis_of(artifact: str) -> tuple[dict[str, Any] | None, bool | None]:
+ """取回一份分析结论：(结论, hash_ok)。哈希对不上的 (None, False)，取不回来的 (None, None)，内容不是分析结论的 (None, None)。"""
+ content, ok = _fetch(artifact) if artifact else (None, None)
+ return (content, ok) if ok is True and is_result(content) else (None, False if ok is False else None)
+
+
+def _catalog_alias_of(catalog: dict[str, Any], artifact: Any) -> str | None:
+ """这份工件在报告目录里的编号：查询快照是 Q…，分析结论是 A…。"""
+ if not artifact:
+ return None
+ return next((alias for alias, e in catalog.items()
+ if isinstance(e, dict) and e.get("kind") in ("query", LEDGER_KIND) and e.get("artifact") == artifact), None)
+
+
+def _analysis_cited(cite: dict[str, Any], result: dict[str, Any] | None) -> dict[str, Any]:
+ """这一处引用的是结论的哪一项：判定语（phrase）、具名标量（value：键、原值、换算）、结果表的一格（cell：行、列、原值）。"""
+ locator = cite.get("locator") or {}
+ if cite.get("kind") == "cell":
+ row, column = locator.get("row"), locator.get("column")
+ raw = None
+ table = (result or {}).get("table") if isinstance((result or {}).get("table"), dict) else None
+ if table is not None:
+ with contextlib.suppress(Exception):
+ raw, column = locate_cell(table, row, column)
+ return {"what": "cell", "row": row, "column": column, "raw": raw}
+ if locator.get("key"):
+ values = (result or {}).get("values") if isinstance((result or {}).get("values"), dict) else {}
+ return {"what": "value", "key": locator["key"], "raw": values.get(locator["key"]),
+ **({"conv": cite["conv"]} if cite.get("conv") else {})}
+ return {"what": "phrase"}
+
+
+def _analysis_rendered(cite: dict[str, Any], result: dict[str, Any]) -> str | None:
+ """按封存的结论重新渲染这一处引用（判定语按文本格、具名标量和结果表的格按单元格的规则）。渲染不出来是 None。"""
+ try:
+ cited = _analysis_cited(cite, result)
+ if cited["what"] == "phrase":
+ return render_cell(result.get("phrase"), kind="text") if isinstance(result.get("phrase"), str) else None
+ if cited["what"] == "value":
+ return render_cell(cited["raw"], cite.get("conv"))
+ return render_cell(cited["raw"], cite.get("conv"), plain=cite.get("plain") is True)
+ except RenderError:
+ return None
+
+
+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]]:
+ """分析结论的出处链（00-总方案 §3.3）：判定 → 方法 → 计算 → 输入 → 复算 → 封存。
+
+ - 判定（analysis）：判定语、判定标准、判定码和点名的段、附带说明、算不了的原因，这一处引用的是哪一项（判定语、具名标量、
+ 结果表的一格）；Agent 写的假设另列，标成「Agent 写的假设」（HYPOTHESIS_NOTE）
+ - 方法（analysis_method）：方法与版本、规范化后的参数、实际用的阈值（只列这个算子用到的）、现查的表范围（probes）
+ - 计算（analysis_calc）：结果表（前 table_top_k 段，加上被引用的那一行附近）、具名的数、关键数的代入式
+ （evidence.analysis_formulas）
+ - 输入（analysis_input，每份一步）：落到报告目录的哪个 Q、用了哪些列和行、headline 那一格和结论里总数的闭合核对、
+ 下钻的上一层是哪个 A；后面接上每份输入快照的查询步骤（只认封存范围内交回过的，遮罩照查询步骤的办法）
+ - 复算（analysis_recompute）：按 sha256 取回封存的输入、用同一个函数复算，和封存的逐字段比（evidence.analysis_recheck）；
+ 方法升过版、复算不了都照实写
+ - 封存（analysis_seal）：这件工件是不是封存范围内的台账记过的、哈希复验通过
+ 结论取不回、哈希不符的只给判定和封存两步，照实说为什么。"""
+ 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
+ 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",
+ **{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,
+ "cited": _analysis_cited(cite, result), "rendered": seg.get("text"),
+ "render_ok": rendered is not None and rendered == seg.get("text"), "hash_ok": hash_ok,
+ }
+ 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:
+ seal["note"] = UNSEALED_ANALYSIS
+ if result is None:
+ judged["note"] = TAMPERED_ANALYSIS if hash_ok is False else MISSING_ANALYSIS
+ return [judged, seal]
+ verdict = result.get("verdict") if isinstance(result.get("verdict"), dict) else None
+ judged.update(
+ status=result.get("status"), reason_code=result.get("reason_code"), reason=result.get("reason"),
+ phrase=result.get("phrase"), criterion=result.get("criterion"), verdict=verdict,
+ caveats=[{k: c.get(k) for k in ("code", "text")} for c in result.get("caveats") or [] if isinstance(c, dict)])
+ if isinstance(result.get("hypothesis"), str) and result["hypothesis"].strip():
+ judged.update(hypothesis=result["hypothesis"], hypothesis_note=HYPOTHESIS_NOTE)
+ settings = result.get("settings") if isinstance(result.get("settings"), dict) else {}
+ method = {
+ "step": "analysis_method", "alias": alias, "method": result.get("method"),
+ "method_label": judged["method_label"], "params": result.get("params") or {},
+ "settings": [{"key": k, "label": ANALYSIS_LABELS.get(k, k), "value": settings[k]}
+ 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 []}
+ 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]
+
+
+def _analysis_calc(alias: str, result: dict[str, Any], cited: dict[str, Any], hidden: set[str]) -> dict[str, Any]:
+ """计算一步：结果表（前 table_top_k 段加上被引用的那一行附近，最多 MAX_ANALYSIS_ROWS 行，高亮被引用的格）、具名的数（原值、
+ 渲染出来的字）、关键数的代入式。结果表里万一有取自遮罩列的列（输入查询步骤里遮着的列名），照查询步骤写「已遮罩」。"""
+ values = result.get("values") if isinstance(result.get("values"), dict) else {}
+ shown_values = []
+ for key, raw in values.items():
+ try:
+ text = render_cell(raw)
+ except RenderError:
+ text = MISSING
+ shown_values.append({"key": str(key), "raw": raw, "rendered": text,
+ **({"cited": True} if cited.get("what") == "value" and cited.get("key") == key else {})})
+ step: dict[str, Any] = {"step": "analysis_calc", "alias": alias, "values": shown_values,
+ "formulas": analysis_formulas(result)}
+ table = result.get("table") if isinstance(result.get("table"), dict) else None
+ if table is None or not isinstance(table.get("rows"), list):
+ return step
+ columns = [str(c) for c in table.get("columns") or []]
+ rows = table["rows"]
+ settings = result.get("settings") if isinstance(result.get("settings"), dict) else {}
+ top = settings.get("table_top_k") if isinstance(settings.get("table_top_k"), int) else 10
+ index = set(range(min(top, len(rows))))
+ row = cited.get("row") if cited.get("what") == "cell" else None
+ if isinstance(row, int):
+ index |= {i for i in range(row - WINDOW, row + WINDOW + 1) if 0 <= i < len(rows)}
+ picked = sorted(index)[:MAX_ANALYSIS_ROWS]
+ masked = {i for i, c in enumerate(columns) if c.lower() in hidden}
+ step["table"] = {
+ "columns": columns, "labels": dict(table.get("labels") or {}) if isinstance(table.get("labels"), dict) else {},
+ "total_rows": len(rows), "row_index": picked,
+ "rows": [[MASKED if i in masked else v for i, v in enumerate(rows[r])] if isinstance(rows[r], list) else rows[r]
+ for r in picked],
+ "masked": [columns[i] for i in sorted(masked)],
+ "highlight": {"cells": [[row, str(cited.get("column"))]] if isinstance(row, int) else []},
+ **({"window_truncated": True} if len(index) > MAX_ANALYSIS_ROWS else {}),
+ }
+ return step
+
+
+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]]]:
+ """输入：每份一步（analysis_input），再给每份输入快照一个查询步骤（_query_steps，高亮用到的列；整份用到的给前
+ MAX_WINDOW_ROWS 行）。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"
+ steps: list[dict[str, Any]] = []
+ cells: dict[str, list[tuple[Any, Any]]] = {}
+ for item in result.get("inputs") or []:
+ if not isinstance(item, dict):
+ continue
+ artifact = str(item.get("artifact") or "")
+ role = str(item.get("role") or "")
+ step: dict[str, Any] = {
+ "step": "analysis_input", "of": alias, "ref": item.get("ref"), "role": role,
+ "role_label": _INPUT_ROLES.get(role, role or "输入"), "artifact": artifact or None,
+ **{k: item[k] for k in ("columns", "rows", "row", "column", "value") if k in item},
+ }
+ if role == "parent":
+ step["alias"] = _catalog_alias_of(catalog, artifact)
+ parent, ok = _analysis_of(artifact)
+ step.update(sealed=sealed.trusted and artifact in sealed.analyses, hash_ok=ok,
+ **({"phrase": parent.get("phrase")} if parent else {}))
+ steps.append(step)
+ continue
+ query = sealed.queries.get(artifact)
+ step.update(query=_catalog_alias_of(catalog, artifact), node_id=(query or {}).get("node_id"),
+ sealed=sealed.trusted and query is not None)
+ if role in _HEADLINE_TOTALS and _HEADLINE_TOTALS[role] in values:
+ total = values[_HEADLINE_TOTALS[role]]
+ with contextlib.suppress(RenderError):
+ step["closure"] = {"total_key": _HEADLINE_TOTALS[role], "total": render_cell(total),
+ "headline": render_cell(item.get("value")), "closed": not unclosed}
+ steps.append(step)
+ if artifact and query is not None:
+ wanted = cells.setdefault(artifact, [])
+ if isinstance(item.get("row"), int) and item.get("column"):
+ wanted.append((item["row"], item["column"]))
+ for column in item.get("columns") or []:
+ picked = item.get("rows")
+ 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():
+ 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
+
+
+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
+
+ def run() -> dict[str, Any]:
+ return analysis_recheck(result, snapshots.get, load_result=parents.get)
+
+ return await asyncio.to_thread(run)
+
+
def _missing_input(cite: dict[str, Any], entry: dict[str, Any]) -> bool:
"""引用的指标存在，只是这次缺输入没有值。引用写错（目录里没有、同名指标有歧义）的不算。"""
return (cite.get("kind") == "metric" and cite.get("status") == "unresolved" and entry.get("kind") == "metric"
@@ -1007,6 +1285,10 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
# 期 4：这个指标做没做过双盲复算、结果如何。没有双盲复算复核过这份指标集时没有这个键，老运行一字不差
if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id, metric):
step["dual_check"] = checked
+ # 跨两份同形查询算的环比、同比：口径卡执行时自动做的可比性检查（analysis.contract.MetricComparability），点得开那份
+ # 可比性结论（artifact）。没做过检查的指标没有这个键
+ if isinstance((metric or {}).get("comparability"), dict):
+ step["comparability"] = _comparability_view(metric["comparability"], sealed, chained=step["sealed"])
card_node = str(entry.get("node_id") or "")
if metric is not None and metric.get("inputs"):
# 输入追没追到原始数据（engine/lineage.py）：和「有出处」（确定性复算一致）是两件事，面板分开说。追不到的照实说
@@ -1047,6 +1329,26 @@ 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]:
+ """指标步骤里的可比性：判定（comparable / conditional / incomparable）、判定语、原因（一句一个），连同那份可比性结论的
+ 工件：取回、复验哈希，带上它的判定标准、方法和现查的表范围（点得开）。sealed：哈希复验通过，而且它记在封存过的口径卡
+ 指标集里（chained：指标步骤本身封存了；内容寻址，卡上记着它的 sha256）或者封存范围内的台账里。"""
+ artifact = str(comparability.get("artifact") or "")
+ result, hash_ok = _analysis_of(artifact)
+ 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),
+ }
+ 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)])
+ return view
+
+
async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Masks", catalog: dict[str, Any],
wanted: dict[str, list[tuple[Any, Any]]], seen: set[tuple[str, str]],
depth: int = 1, *, card_node: str = "") -> list[dict[str, Any]]:
@@ -2905,16 +3207,51 @@ async def evidence_provenance(run_id: str, segment_id: str, report: str | None =
def _export_cite_key(seg: dict[str, Any]) -> tuple | None:
- """数字 / 取值片段的引用键：同一处引用（同一个指标、同一格、同样的换算）共用一条出处链。
- 别的片段（表名、引文）的链跟着片段走，不归并。"""
- if seg.get("kind") not in ("number", "value"):
+ """数字 / 取值 / 判定语片段的引用键：同一处引用（同一个指标、同一格、同一条分析结论的同一项、同样的换算）共用一条
+ 出处链。别的片段（表名、引文）的链跟着片段走，不归并。具名标量的键另接在后面：没有它的引用键和以前一模一样。"""
+ if seg.get("kind") not in ("number", "value", LEDGER_KIND):
return None
cite = seg.get("cite")
if not isinstance(cite, dict):
return None
loc = cite.get("locator") if isinstance(cite.get("locator"), dict) else {}
- return (cite.get("kind"), cite.get("alias"), cite.get("status"), cite.get("conv"),
- loc.get("metric"), loc.get("row"), loc.get("column"), loc.get("field"), seg.get("issue"))
+ key = (cite.get("kind"), cite.get("alias"), cite.get("status"), cite.get("conv"),
+ loc.get("metric"), loc.get("row"), loc.get("column"), loc.get("field"), seg.get("issue"))
+ return (*key, loc["key"]) if loc.get("key") else key
+
+
+#: 报告文档里系统写的「分析范围」（analysis.contract.ScopeBlock）：导出页、可验证报告按它渲染，编号点得开。各组的键
+SCOPE_GROUPS = ("phenomena", "decomposed", "excluded", "refused")
+#: 分析范围里一个编号在导出页数据块里的片段键：refs、notes 按它找出处链（不和片段 id s0、s1… 撞）
+SCOPE_REF = "scope:{alias}"
+
+
+def scope_aliases(doc: dict[str, Any]) -> list[str]:
+ """报告文档的「分析范围」里点名的分析结论编号，按出现的先后、去重。没有这一段的文档是空列表。"""
+ scope = doc.get("analysis_scope") if isinstance(doc.get("analysis_scope"), dict) else None
+ if scope is None:
+ return []
+ out = [str(item["alias"]) for group in SCOPE_GROUPS for item in scope.get(group) or []
+ if isinstance(item, dict) and isinstance(item.get("alias"), str) and item["alias"]]
+ return list(dict.fromkeys(out))
+
+
+async def _scope_chains(doc: dict[str, Any], sealed: _Sealed, report_node: str) -> dict[str, list[dict[str, Any]]]:
+ """分析范围里每个编号的出处链（和点开正文里的判定语同一条 _chain，按那条结论的判定语认）：{编号: 链}。目录里没有的
+ 编号不给（文档里写着、目录里却没有，只能是文档有问题，范围那一项点不开）。"""
+ catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {}
+ out: dict[str, list[dict[str, Any]]] = {}
+ for alias in scope_aliases(doc):
+ entry = catalog.get(alias)
+ if not isinstance(entry, dict) or entry.get("kind") != LEDGER_KIND:
+ continue
+ result, _ = _analysis_of(str(entry.get("artifact") or ""))
+ seg = {"id": SCOPE_REF.format(alias=alias), "kind": LEDGER_KIND, "state": "deterministic",
+ "text": render_cell(result.get("phrase"), kind="text") if result and result.get("phrase") else alias,
+ "cite": {"alias": alias, "kind": LEDGER_KIND, "status": "resolved", "locator": {},
+ "eid": entry.get("eid")}}
+ out[alias] = await _chain(seg, doc, sealed, report_node)
+ return out
async def _export_withheld(sealed: _Sealed, doc: dict[str, Any]) -> frozenset[str]:
@@ -3022,6 +3359,14 @@ async def export_report_html(run_id: str, report: str | None = None) -> Response
refs[sid] = str(ck)
if key is not None:
memo[key] = str(ck)
+ # 系统写的「分析范围」：每个编号点得开，链和点开正文里的判定语是同一条（引用键相同的共用）
+ for alias, chain in (await _scope_chains(doc, sealed, chosen.node_id)).items():
+ sid = SCOPE_REF.format(alias=alias)
+ ck = (LEDGER_KIND, alias, "resolved", None, None, None, None, None, None)
+ if ck not in memo:
+ memo[ck] = str(ck)
+ chains[str(ck)] = chain
+ refs[sid] = memo[ck]
async with SessionLocal() as session:
run = await session.get(Run, run_id)
if run is None: # 两次读之间被删掉了
@@ -3739,6 +4084,14 @@ def _note(seg: dict[str, Any], unit: dict[str, Any]) -> str:
"仅核对它存在且本次用到，不核对其业务含义")
if seg.get("state") == "deterministic" and seg.get("kind") == "quote":
return "这段原话在知识库检索命中的片段中逐字出现（空白归一化后比对），已由系统核对"
+ if seg.get("state") == "deterministic" and cite.get("kind") == LEDGER_KIND:
+ if (cite.get("locator") or {}).get("key"):
+ return "这个数取自分析结论的具名的数，由系统按固定规则显示，未经模型转写"
+ return ("判定语由系统按平台的方法算出判定、按模板写成，未经模型转写；判定标准、计算过程、输入和复算见出处链。"
+ "判定只说明变化来自哪里，不说明原因")
+ if seg.get("state") == "deterministic" and cite.get("kind") == "cell" \
+ and str(cite.get("alias") or "").startswith(CATALOG_PREFIX):
+ return "该值由系统从分析结论的结果表中取出、按固定规则显示，未经模型转写"
if seg.get("state") == "deterministic" and cite.get("kind") == "cell":
return "该值由系统从查询快照中取出、按固定规则显示，未经模型转写"
if seg.get("state") == "deterministic":
@@ -3994,21 +4347,30 @@ async def _judge_loader(doc: dict[str, Any], catalog: dict[str, Any], units: lis
read = loader or load
masks = _Masks()
shown: dict[str, Any] = {}
+ wanted: list[tuple[str, Any]] = []
for alias in dict.fromkeys(a for c in candidates(doc, units) for a in c.aliases):
entry = catalog.get(alias)
- if not isinstance(entry, dict) or entry.get("kind") != "query" or not entry.get("artifact"):
+ if not isinstance(entry, dict) or not entry.get("artifact"):
continue
- artifact = str(entry["artifact"])
+ if entry.get("kind") == "query":
+ wanted.append((str(entry["artifact"]), entry.get("source")))
+ elif entry.get("kind") == LEDGER_KIND:
+ # 分析结论的摘录给它输入查询的 SQL：输入快照同样逐份遮（结论本身不经这里，摘录照它自己的办法取）
+ with contextlib.suppress(Exception):
+ result = read(str(entry["artifact"]))
+ wanted.extend((str(i["artifact"]), None) for i in (result or {}).get("inputs") or []
+ if isinstance(i, dict) and i.get("artifact") and i.get("role") != "parent")
+ for artifact, source in wanted:
if artifact in shown:
continue
try:
snap = read(artifact)
except Exception: # noqa: BLE001 - 读不出来、不在封存范围里：摘录本来就不给它，交给 prepare 照它的办法处理
continue
- if not isinstance(snap, dict):
+ if not isinstance(snap, dict) or not isinstance(snap.get("rows"), list):
continue
recorded = _mask_names(snap.get("mask_columns"))
- sure = set((await masks.sets(artifact, entry.get("source") or snap.get("source"), snap)).sure)
+ sure = set((await masks.sets(artifact, source or snap.get("source"), snap)).sure)
shown[artifact] = {**snap, "mask_columns": sorted(recorded | sure)} if sure - recorded else snap
def reader(artifact: str) -> Any:
@@ -4262,8 +4624,11 @@ GROUPS = (("cited", "有出处"), ("none", "无证据"), ("suspicious", "可疑
_INTEGRITY = frozenset({"bad_schema", "segment_mismatch", "structural_text", "render_mismatch", "eid_mismatch",
"state_mismatch"})
#: 落在已经列出的片段上也单列一行的违规：文档对不上的，和引用了遮罩列的格子的（masked_cell：引用的那一格落在的正是一个
-#: 「有出处」的单元格片段，不单列的话审计表、导出的 CSV 里只有一行普通的「有出处」）
-_LISTED_ALWAYS = _INTEGRITY | {"masked_cell"}
+#: 「有出处」的单元格片段，不单列的话审计表、导出的 CSV 里只有一行普通的「有出处」）；分析说法核对的三种（判断词没挂分析
+#: 结论、判断和结论对不上、因果说法没有依据，engine/analysis_claims.py）：句子里的判定语、数照样是「有出处」的片段，说法本身
+#: 有问题要单独看得到
+ANALYSIS_CLAIM_CODES = frozenset({"analysis_claim_uncited", "analysis_claim_mismatch", "causal_unsupported"})
+_LISTED_ALWAYS = _INTEGRITY | {"masked_cell"} | ANALYSIS_CLAIM_CODES
UNCITED_CLAIM = "这句结论没有附依据（既没有引用，也没有注明依据）"
UNCITED_CLAIM_COUNTED = "这句结论没有附依据。本报告要求结论句附依据，出具时计入缺口"
UNCITED_CLAIM_WITHHELD = "这句结论没有附依据，按出具契约不予出具"
@@ -4297,15 +4662,17 @@ SEAL_NONE = "未封存：运行尚未结束，或正停在人工审批"
_CSV_LABELS: dict[str, dict[Any, str]] = {
"group": dict(GROUPS),
"kind": {"number": "数字", "value": "值", "entity": "表名、字段名", "quote": "引文", "claim": "结论句",
- "violation": "违规"},
+ "violation": "违规", LEDGER_KIND: "分析判定"},
"state": {"deterministic": "有出处", "none": "无证据", "candidate": "候选"},
"issue": {"uncited_number": "没有出处的数字", "unresolved_ref": "引用无法解析", "unknown_entity": "疑似不存在的名称",
"unverified_entity": "无法核实", "uncited_claim": "结论句未附依据", "no_candidate": "未找到候选",
"contradicted_claim": "证据相矛盾", "insufficient_claim": "证据不足", "unsupported_claim": "证据不支持",
- "masked_cell": "引用了遮罩列",
+ "masked_cell": "引用了遮罩列", "inline_see": "依据写在句子中间", "absence_claim": "缺失断言",
+ "analysis_claim_uncited": "判断词未挂分析结论", "analysis_claim_mismatch": "判断与分析结论不符",
+ "causal_unsupported": "因果说法没有依据",
**{code: "文档不一致" for code in _INTEGRITY}},
"evidence_kind": {"metric": "口径卡指标", "cell": "查询单元格", "input": "运行输入", "table": "表", "column": "字段",
- "quote": "知识库原话", "query": "查询结果"},
+ "quote": "知识库原话", "query": "查询结果", LEDGER_KIND: "分析结论"},
"sealed": {True: "是", False: "否", None: "—"},
}
#: 电子表格会把以这些字符开头的格子当公式算：报告、答案里的字来自模型，照原样写进去就能被人利用
@@ -4507,10 +4874,20 @@ def _segment_row(group: str, report: _Report, field_name: str | None, seg: dict[
if group != "cited":
return row
kind = cite.get("kind")
- if kind == "cell":
+ if kind == "cell" and entry.get("kind") == LEDGER_KIND:
+ # 分析结论结果表里的一格：证据种类记分析结论，封存按台账认（结论不是查询快照）
+ row.update(evidence=_cell_name(alias, cite.get("locator") or {}), evidence_kind=LEDGER_KIND,
+ artifact=entry.get("artifact"), node_id=entry.get("node_id"),
+ sealed=_entry_sealed(sealed, entry, inputs))
+ elif kind == "cell":
artifact = str(entry.get("artifact") or "")
row.update(evidence=_cell_name(alias, cite.get("locator") or {}), artifact=artifact or None,
node_id=entry.get("node_id"), sealed=sealed.trusted and artifact in sealed.queries)
+ elif kind == LEDGER_KIND:
+ # 判定语写结论的叫法（「变化拆解（可加） · 判定语」），具名标量写到键（A3.top_share）
+ key = (cite.get("locator") or {}).get("key")
+ row.update(evidence=f"{alias}.{key}" if key else entry.get("label") or alias, artifact=entry.get("artifact"),
+ node_id=entry.get("node_id"), sealed=_entry_sealed(sealed, entry, inputs) if entry else False)
elif kind == "quote":
source = cite.get("source") or {}
artifact = str(source.get("artifact") or entry.get("artifact") or "")
diff --git a/backend/tests/test_analysis_evidence_api.py b/backend/tests/test_analysis_evidence_api.py
new file mode 100644
index 00000000..800e5cd7
--- /dev/null
+++ b/backend/tests/test_analysis_evidence_api.py
@@ -0,0 +1,322 @@
+"""分析结论进证据链（第 1 波 W1-b）：证据接口、审计、报告导出。
+
+运行是直接落库的虚构封存运行（虚构零售库 shop）：一次查询（Q1）、一份按地区拆的分析结论（A1，分析卡或 Agent 做的），
+报告引用判定语、具名标量、结果表的格。算子第 1 波还没落地，复算按 monkeypatch 换。
+
+- 证据图：A1 条目带算子、判定；「派生自」的边 A1 → Q1
+- 片段接口：判定 → 方法 → 计算 → 输入（接查询步骤）→ 复算 → 封存；Agent 写的假设标出来；复算不了照实写，不让接口出错
+- 审计：分析判定、分析结论的叫法；分析说法核对的违规单列一行
+- 报告导出：数据块里有分析结论那几步；文档带「分析范围」的，正文后面渲染、编号点得开
+- 指标步骤带可比性
+"""
+from __future__ import annotations
+
+import csv
+import io
+import json
+import re
+import uuid
+from typing import Any
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app import analysis
+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
+
+GRAPH = {"nodes": [{"id": n, "type": t, "data": {"label": label}} for n, t, label in [
+ ("fetch", "tool", "按地区取数"), ("an", "agent", "分析"), ("card", "metrics", "周报口径"),
+ ("write", "report", "写周报")]]}
+
+
+@pytest.fixture
+async def client():
+ async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+ yield c
+
+
+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 的算法封存。"""
+ 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"}])
+ 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)
+ 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]})]
+ 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])
+ 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")
+ seq = len(events)
+ events += [(seq + 1, "report.checked", "write", {"doc_artifact": doc_id, "ok": True}),
+ (seq + 2, "run.finished", None, {"output": {}})]
+ async with SessionLocal() as session:
+ session.add(Run(id=run_id, workflow_name="分析周报", status="succeeded", run_class="exploratory",
+ graph=GRAPH, agent_limits={}, manifest_hash=manifest_hash(events), manifest_seq=events[-1][0]))
+ await session.flush()
+ 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}
+
+
+def seg_id(doc: dict[str, Any], ref: str) -> str:
+ return next(s["id"] for s in iter_segments(doc) if s.get("ref") == ref)
+
+
+TEXT = "[[a:A1]]。华东占 [[a:A1.top_share|pct]]，变化 [[v:A1.r0.delta]]。[[see:A1]]"
+
+
+# --------------------------------------------------------------------------
+# 证据图
+# --------------------------------------------------------------------------
+
+
+async def test_the_graph_lists_the_analysis_and_what_it_was_derived_from(client):
+ made = await sealed_run(TEXT)
+ body = (await client.get(f"/api/runs/{made['run']}/evidence")).json()
+ assert body["mode"] == "cited"
+ item = next(e for e in body["evidence"] if e["alias"] == "A1")
+ assert item["kind"] == "analysis" and item["op"] == "decompose" and item["verdict"] == "driver"
+ assert item["item"] == "drop" and item["sealed"] is True and item["artifact"] == made["analysis"]
+ assert item["label"].startswith("变化拆解（可加） · ")
+ edges = {(e["from"], e["to"], e["rel"]) for e in body["edges"]}
+ assert ("A1", "Q1", "derived_from") in edges
+ assert ("A1.r0.delta", "A1", "cell_of") in edges
+
+
+# --------------------------------------------------------------------------
+# 片段接口：出处链
+# --------------------------------------------------------------------------
+
+
+async def test_the_phrase_opens_judgment_method_calculation_inputs_recompute_and_seal(client):
+ made = await sealed_run(TEXT)
+ body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()
+ chain = body["chain"]
+ 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["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 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"}
+ assert "k_warn" not in {s["key"] for s in method["settings"]}
+ 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%"
+ 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"] == "复算失败：系统里还没有这种分析的复算实现"
+ 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))
+ 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"}
+ 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}
+ assert value["chain"][-2]["text"] == "复算一致：按封存的输入重新计算，结论逐项相同"
+ assert value["note"] == "这个数取自分析结论的具名的数，由系统按固定规则显示，未经模型转写"
+
+ cell = (await client.get(
+ f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'v:A1.r0.delta')}")).json()
+ assert cell["chain"][0]["cited"] == {"what": "cell", "row": 0, "column": "delta", "raw": "-120"}
+ assert cell["chain"][2]["table"]["highlight"] == {"cells": [[0, "delta"]]}
+ assert cell["note"] == "该值由系统从分析结论的结果表中取出、按固定规则显示，未经模型转写"
+
+
+async def test_the_recompute_gets_the_sealed_inputs_by_hash(client, monkeypatch):
+ seen: dict[str, Any] = {}
+
+ 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"}
+
+ 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
+
+
+async def test_recompute_errors_do_not_break_the_panel(client, monkeypatch):
+ def broken(result, load, *, load_result=None):
+ raise ZeroDivisionError("division by zero")
+
+ monkeypatch.setattr(analysis, "recompute", broken)
+ made = await sealed_run(TEXT)
+ response = await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")
+ assert response.status_code == 200
+ again = next(s for s in response.json()["chain"] if s["step"] == "analysis_recompute")
+ assert again["status"] == "failed" and again["text"].startswith("复算失败：")
+
+
+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}")
+ 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")
+ body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()
+ judged, seal = body["chain"]
+ assert judged["note"] == "分析结论与哈希不一致，疑似被修改，不展示其中的判定和数" and "phrase" not in judged
+ assert seal["hash_ok"] is False and seal["sealed"] is False and body["seal"]["covered"] is False
+
+
+async def test_agent_analyses_are_marked_as_such(client):
+ made = await sealed_run(TEXT, item=None, call_id="call-9")
+ body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json()
+ 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()
+ inputs = [s for s in body["chain"] if s["step"] == "analysis_input"]
+ assert [s["role"] for s in inputs] == ["data", "headline_target", "parent"]
+ 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"]
+
+
+async def test_the_metric_step_shows_the_comparability(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 天数据",
+ "reasons": ["本期只有 3 天数据"]}}]}
+ made = await sealed_run("环比 [[m:wow]]。", card=card)
+ body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'm:wow')}")).json()
+ assert body["chain"][0]["comparability"] == {
+ "verdict": "incomparable", "phrase": "两期不可比：本期只有 3 天数据", "reasons": ["本期只有 3 天数据"],
+ "artifact": None, "hash_ok": None, "sealed": False}
+
+
+# --------------------------------------------------------------------------
+# 审计
+# --------------------------------------------------------------------------
+
+
+async def test_the_audit_names_analysis_evidence_and_lists_claim_violations(client):
+ violation = {"code": "analysis_claim_mismatch", "message": "A1 判的是变化主要来自华东，不能写「主要来自华南」",
+ "segment": "s0", "unit": "u0", "span": [0, 4], "text": "主要来自"}
+ made = await sealed_run(TEXT, doc_extra={"violations": [violation]})
+ body = (await client.get(f"/api/runs/{made['run']}/evidence/audit")).json()
+ rows = [r for g in body["groups"] for r in g["rows"]]
+ phrase = next(r for r in rows if r.get("ref") == "a:A1")
+ assert phrase["kind"] == "analysis" and phrase["evidence_kind"] == "analysis" and phrase["sealed"] is True
+ assert phrase["evidence"].startswith("变化拆解（可加） · ")
+ value = next(r for r in rows if r.get("ref") == "a:A1.top_share|pct")
+ assert value["evidence"] == "A1.top_share" and value["kind"] == "number"
+ cell = next(r for r in rows if r.get("ref") == "v:A1.r0.delta")
+ assert cell["evidence"] == "A1.r0.delta" and cell["evidence_kind"] == "analysis" and cell["sealed"] is True
+ flagged = next(r for r in rows if r.get("issue") == "analysis_claim_mismatch")
+ assert flagged["group"] == "none" and flagged["kind"] == "violation"
+
+ text = (await client.get(f"/api/runs/{made['run']}/evidence/audit?format=csv")).text
+ table = list(csv.reader(io.StringIO(text.lstrip(""))))
+ head = table[0]
+ lines = [dict(zip(head, line)) for line in table[1:]]
+ assert any(r["种类"] == "分析判定" and r["证据种类"] == "分析结论" for r in lines)
+ assert any(r["问题"] == "判断与分析结论不符" for r in lines)
+
+
+def test_audit_labels_cover_the_new_violation_codes():
+ from app.api.evidence import _CSV_LABELS
+
+ issues = _CSV_LABELS["issue"]
+ assert issues["analysis_claim_uncited"] == "判断词未挂分析结论"
+ assert issues["analysis_claim_mismatch"] == "判断与分析结论不符"
+ assert issues["causal_unsupported"] == "因果说法没有依据"
+ assert issues["inline_see"] == "依据写在句子中间" and issues["absence_claim"] == "缺失断言"
+
+
+# --------------------------------------------------------------------------
+# 报告导出
+# --------------------------------------------------------------------------
+
+SCOPE = {"analysis_scope": {
+ "phenomena": [{"alias": "A1", "op": "decompose", "phrase": "销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）"}],
+ "decomposed": [{"alias": "A1", "op": "decompose", "mode": "additive", "dims": ["地区"], "verdict": "driver",
+ "phrase": "主要来自华东"}],
+ "excluded": [], "refused": [{"alias": "A1", "op": "baseline", "reason": "历史只有 5 期，至少要 8 期"}],
+ "not_decomposed": ["渠道"], "tests": 2}}
+
+
+def data_block(html: str) -> dict[str, Any]:
+ raw = re.search(r'', html, re.S).group(1)
+ return json.loads(raw)
+
+
+async def test_the_html_export_carries_the_analysis_chain_and_the_scope(client):
+ made = await sealed_run(TEXT, doc_extra=SCOPE)
+ response = await client.get(f"/api/runs/{made['run']}/report.html")
+ assert response.status_code == 200
+ html = response.text
+ data = data_block(html)
+ phrase = seg_id(made["doc"], "a:A1")
+ chain = data["chains"][data["refs"][phrase]]
+ assert [s["step"] for s in chain][:3] == ["analysis", "analysis_method", "analysis_calc"]
+ assert data["notes"][phrase].startswith("判定语由系统按平台的方法算出判定")
+ # 分析范围：正文后面渲染，编号点得开、和点开判定语是同一条链
+ assert '现象" in html and "拆过" in html
+ assert "算不了" in html and "渠道" in html and "这次做了 2 次差异检验" in html
+ assert 'data-seg="scope:A1"' in html and "（按地区）" in html
+ assert data["refs"]["scope:A1"] == data["refs"][phrase]
+ assert f'data-seg="{phrase}"' in html and 'data-kind="analysis"' in html
+
+
+async def test_docs_without_a_scope_render_no_scope(client):
+ made = await sealed_run(TEXT)
+ html = (await client.get(f"/api/runs/{made['run']}/report.html")).text
+ assert '