yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 10:41:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

证据接口认分析结论：封存范围按台账记下分析结论；证据图的 A 条目带算子、判定、谁做的，连「派生自」的边到它的输入（Q…）和下钻的上一层（A…）；片段接口的出处链为判定 → 方法 → 计算 → 输入（每份输入落到哪个 Q、用了哪些列和行、headline 闭合、下钻的上一层，接输入快照的查询步骤并照常遮罩）→ 复算（按 sha256 取回封存的输入、在线程里复算并逐字段比，算子没落地、出意外照实写复算失败，不让接口出错）→ 封存，结论哈希不符的只给判定和封存两步并写明原因；Agent 写的假设标成 Agent 写的假设；片段说明分判定语、具名的数、结果表的格三种。指标步骤展示口径卡记的两期可比性（点得开那份可比性结论）。审计：分析判定、分析结论的叫法，结果表的格按台账认封存，分析说法核对的三种违规单列一行，违规码补上判断词未挂分析结论、判断与分析结论不符、因果说法没有依据，以及原来就缺的依据写在句子中间、缺失断言。报告导出：判定语片段按引用键共用出处链，分析范围里的编号接同一条链。裁判的 loader 把分析结论的输入快照也逐份遮好

Browse Source
This commit is contained in:
ThearchyHelios
2026-10-21 13:07:55 +08:00
parent 4dd6889e69
commit 7474fa5030
2 changed files with 714 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/evidence.py
+392 -15
View File
@@ -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 "")
backend/tests/test_analysis_evidence_api.py
+322
View File
@@ -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'<script type="application/json" id="report-data">(.*?)</script>', 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 '<section class="scope"' in html and "<dt>现象</dt>" in html and "<dt>拆过</dt>" in html
assert "<dt>算不了</dt>" in html and "<dd>渠道</dd>" 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 '<section class="scope"' not in html and 'data-seg="scope:A1"' not in html