门禁 G3 加一条：受管级别的报告撰写节点要把「分析说法核对」设为严格（REPORT_POLICY 加 analysis_claims: strict，没写时运行时是 check、只标注）。受管级别挡住发布、已发布级别只警告，写错的值交给校验；已经发布的版本照常运行（按缺省的 check 核），下次发布才按新规则查。同一份要求自动跟到自动修复（一键写明）、一键升级成受管、按口径出具起草的报告节点；自动修复「不许放宽」的判断认这一项（从严格改回去算放宽）；白话补一句分析说法；模板 ⑨ 的报告节点补上这一项，拿来照旧能发布成受管。各处「受管要求都满足」的图（测试夹具）跟着补上这一项

backend/app/engine/autofix.py

@@ -1338,9 +1338,10 @@ def forbidden_changes(before: dict[str, Any], after: dict[str, Any],
     return reasons
 
 
-#: 报告撰写节点三项设置从严到宽的次序：没写时运行时的缺省值（on_violation 没写时探索运行按 flag）
+#: 报告撰写节点几项设置从严到宽的次序：没写时运行时的缺省值（on_violation 没写时探索运行按 flag，analysis_claims 没写是
+#: check）
 _REPORT_STRICT = {"numbers": ("strict", "strict"), "on_violation": ("fail", "flag"),
-                  "claims": ("require_citation", "off")}
+                  "claims": ("require_citation", "off"), "analysis_claims": ("strict", "check")}
 #: claims 从宽到严：off < require_citation < judge（在挂引用之外再请模型逐句裁判，不支持的按 on_unsupported
 #: 判档）。不在这里的写法不比较（validate 报 report.claims_invalid）
 _CLAIMS_ORDER = ("off", "require_citation", CLAIMS_JUDGE)

backend/app/engine/governance.py

@@ -18,6 +18,7 @@ from app.engine.schema import (
     ValidationResult,
     _ancestors,
     _caliber_inputs,
+    analysis_claims_problem,
     claims_problem,
     contract_needs_metrics,
     judge_problems,
@@ -322,7 +323,8 @@ def _lint_contract(
 #
 # G1 带契约的出口核对报告撰写节点的文档，成果字段的文字只能来自报告撰写节点
 # G2 模型写的文字不绕过报告撰写节点直接流进出口（从图上追）
-# G3 报告撰写节点写明 numbers: strict、on_violation: fail、claims: require_citation（或 judge，带上预算）
+# G3 报告撰写节点写明 numbers: strict、on_violation: fail、claims: require_citation（或 judge，带上预算）、
+#    analysis_claims: strict（分析能力一轮加的：判断词没挂对分析结论的句子重写后仍有的计入缺口）
 # G4 口径卡的输入不来自计算角色的沙箱代码
 # G5 给口径卡供数的 agent 配 output_schema 并开 cite_fields（模型调用节点不能给口径卡供数）
 #
@@ -330,13 +332,18 @@ def _lint_contract(
 # 下次发布才按这些规则查。受管级别挡住发布，已发布级别只给警告。
 # --------------------------------------------------------------------------
 
-#: 报告撰写节点在受管级别里要写明的三项：(配置键, 要求的值, 为什么)。给人看的时候键和值都换成
-#: labels 里的界面叫法
+#: 报告撰写节点在受管级别里要写明的几项：(配置键, 要求的值, 为什么)。给人看的时候键和值都换成
+#: labels 里的界面叫法。自动修复、一键升级、按口径出具起草的报告节点都照这一份写（autofix、upgrade、ledger/certify）。
+#: analysis_claims 没写时运行时是 check（只标注、不影响档位），受管级别要写明 strict；已经发布的版本照常运行，下次发布才按
+#: 这一条查（和别的 G 规则一样只在发布、发布前检查时起作用）
 REPORT_POLICY = (
     ("numbers", "strict", "没有出处的数字判为违规"),
     ("on_violation", "fail", "重写后仍有违规时节点失败，不带着问题出具"),
     ("claims", "require_citation", "未附依据的结论句计入缺口，出具按档位降档"),
+    ("analysis_claims", "strict", "判断词没挂对分析结论、原因没有依据的句子重写后仍有的计入缺口，出具按档位降档"),
 )
+#: 取值写错了由 validate 报（report.claims_invalid、report.analysis_claims_invalid），G3 不重复
+_POLICY_PROBLEMS = {"claims": claims_problem, "analysis_claims": analysis_claims_problem}
 #: 除了 REPORT_POLICY 里写的那个值，还认的更严的写法。claims: judge 在挂引用之外再请另一个模型按证据逐句
 #: 裁判，不支持的按 on_unsupported 判档——比 require_citation 严，但要写预算（见 judge_budget_missing）
 REPORT_POLICY_ALSO: dict[str, tuple[str, ...]] = {"claims": (CLAIMS_JUDGE,)}
@@ -666,8 +673,8 @@ def _lint_evidence(spec: GraphSpec, flag: Callable[..., None], *, strict: bool)
         loose = []
         for key, want, why in REPORT_POLICY:
             value = _effective(spec, node, key)
-            if key == "claims" and claims_problem(value):
-                continue        # 写了不认识的值，validate 的 report.claims_invalid 报
+            if key in _POLICY_PROBLEMS and _POLICY_PROBLEMS[key](value):
+                continue        # 写了不认识的值，validate 的 report.claims_invalid / analysis_claims_invalid 报
             also = REPORT_POLICY_ALSO.get(key, ())
             if value != want and value not in also:
                 now = "当前未设置" if value in (None, "") else f"当前为「{option_label(key, value)}」"

backend/app/plain.py

@@ -80,7 +80,9 @@ PLAIN: dict[str, Plain] = {
     "governed.text_bypass": Plain(
         "模型写的文字绕过了报告直接进了成果", "这些文字里的数和表名都没法追溯。", _A, "改由报告撰写节点来写，成果取报告正文。"),
     "governed.report_policy": Plain(
-        "报告的核对规则不够严", "受管级别要求报告用最严的核对规则。", _A, "把报告撰写节点的核对规则调到最严。"),
+        "报告的核对规则不够严",
+        "受管级别要求报告用最严的核对规则：数字要有出处、结论句要挂依据，「主要来自」「异常」这类判断要挂对分析结论。", _A,
+        "把报告撰写节点的核对规则调到最严，分析说法核对也设为严格。"),
     "governed.judge_budget": Plain(
         "模型裁判没设金额上限", "裁判的花费不能沿用默认值，要事先说清。", _Y, "写一个每份报告的裁判金额上限，或明确选择不限。"),
     "governed.caliber_compute_input": Plain(

backend/app/seed.py

@@ -266,9 +266,9 @@ TEMPLATES: list[dict[str, Any]] = [
                     "expression": "cell(nodes.fetch, 0, 'refund_cnt') / cell(nodes.fetch, 0, 'order_cnt')"
                                   " if cell(nodes.fetch, 0, 'order_cnt') else None"},
                ]),
-            # 受管出具要求报告撰写节点写明最严的三项（发布门禁 G3），模板照着写，拿来就能发布成受管
+            # 受管出具要求报告撰写节点写明最严的几项（发布门禁 G3，含分析说法核对设为严格），模板照着写，拿来就能发布成受管
             _n("write", "report", "报告撰写", metrics_from=["caliber"],
-               numbers="strict", on_violation="fail", claims="require_citation",
+               numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict",
                system="你是周报撰写人：结论先行，只写有数据支撑的判断。",
                instructions="为 {{ input.week }} 写一份简短的周报：先用一两句话总结本周，"
                             "再分点说销售额、订单、客单价和退款的变化。"),

backend/tests/test_approval_default.py

@@ -159,7 +159,8 @@ def _governable(tool: str):
              metrics=[{"id": "done", "name": "已执行", "unit": "次", "expression": "1"}]),
         # 受管契约要核对报告撰写节点的文档（门禁 G1），叙述模式发不成受管
         node("write", "report", instructions="写执行结果",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation",
+             analysis_claims="strict"),
         node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.q }}"}],
              contract={"metrics_from": ["caliber"], "report_from": "write", "required": ["done"], "strict": True}),
     )

backend/tests/test_autofix_assist.py

@@ -40,7 +40,7 @@ def two_exits() -> dict:
         node("card", "metrics", "周报口径卡", caliber="周报口径", metrics=[
             {"id": "gmv", "name": "销售额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"}]),
         node("write", "report", "报告撰写", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
         node("raw", "output", "原始数", fields=[{"name": "数", "value": "{{ nodes.fetch }}"}]),
     ]
@@ -230,7 +230,7 @@ def guarded() -> dict:
         node("team", "supervisor", "复核团队", goal="复核", agents=[{"name": "checker", "tools": []}]),
         node("gate", "human", "人工把关", prompt="确认数字"),
         node("write", "report", "报告撰写", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], contract=CONTRACT),
     ]
     edges = [{"id": f"e{i}", "source": a, "target": b} for i, (a, b) in enumerate(

backend/tests/test_autofix_governance.py

@@ -18,7 +18,7 @@ from app.engine.governance import publish_issues
 from app.engine.schema import GraphSpec
 from app.main import app
 
-STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation"}
+STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation", "analysis_claims": "strict"}
 SCHEMA = {"type": "object", "properties": {"gmv": {"type": "number"}}}
 
 
@@ -88,20 +88,21 @@ def test_report_policy_is_filled_in_with_one_auto_fix():
     fix = plan(graph)[fid]
     assert fix["kind"] == "auto" and fix["node_id"] == "write"
     assert fix["preview"]["after"] == STRICT and fix["preview"]["before"] == {
-        "numbers": "off", "on_violation": None, "claims": None}
+        "numbers": "off", "on_violation": None, "claims": None, "analysis_claims": None}
     out = apply_fixes(graph, [fid], level="governed")
     assert out["applied"] == [fid] and out["rejected"] == [] and out["ok"] is True
     assert {k: config(out["graph"], "write")[k] for k in STRICT} == STRICT
     assert config(out["graph"], "write")["instructions"] == "写周报"          # 别的配置原样留着
     assert [(c["field"], c["before"], c["after"]) for c in out["changes"]] == [
-        ("numbers", "off", "strict"), ("on_violation", None, "fail"), ("claims", None, "require_citation")]
+        ("numbers", "off", "strict"), ("on_violation", None, "fail"), ("claims", None, "require_citation"),
+        ("analysis_claims", None, "strict")]
     assert out["ops"] == [{"op": "update_node", "id": "write", "config": STRICT}]
     # 坐标、视口这些前端的东西原样留着
     assert find(out["graph"], "write")["position"] == {"x": 10, "y": 20} and out["graph"]["viewport"]["zoom"] == 1
 
 
 def test_only_the_missing_settings_are_written():
-    graph = traceable(report={"numbers": "strict", "claims": "require_citation"})
+    graph = traceable(report={"numbers": "strict", "claims": "require_citation", "analysis_claims": "strict"})
     fix = plan(graph)["governed.report_policy:write"]
     assert fix["preview"] == {"field": "on_violation", "before": None, "after": "fail"}
     out = apply_fixes(graph, [fix["id"]], level="governed")

backend/tests/test_autofix_graph_defaults.py

@@ -40,7 +40,7 @@ def weekly(default: str | None = "never", **ask_config) -> dict:
         node("card", "metrics", "周报口径卡", caliber="周报口径", metrics=[
             {"id": "gmv", "name": "销售额", "unit": "元", "expression": "vars.gmv"}]),
         node("write", "report", "报告撰写", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
              contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"], "strict": True}),
     ]

backend/tests/test_autofix_rules.py

@@ -33,7 +33,7 @@ def weekly(**contract_overrides) -> dict:
             {"id": "gmv", "name": "销售额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"},
             {"id": "orders", "name": "订单数", "unit": "单", "expression": "cell(nodes.fetch, 0, 'orders')"}]),
         node("write", "report", "报告撰写", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
              contract={k: v for k, v in contract.items() if v is not None}),
     ]
@@ -72,7 +72,8 @@ def add_card(graph: dict, nid: str = "card2") -> dict:
 
 def add_report(graph: dict, nid: str = "write2") -> dict:
     graph["nodes"].insert(4, node(nid, "report", "第二份报告", instructions="写月报",
-                                  numbers="strict", on_violation="fail", claims="require_citation"))
+                                  numbers="strict", on_violation="fail", claims="require_citation",
+                                  analysis_claims="strict"))
     graph["edges"] += [{"source": "card", "target": nid}, {"source": nid, "target": "done"}]
     return graph
 

backend/tests/test_contract_cells.py

@@ -344,7 +344,7 @@ def cells_only(**contract):
         node("fetch", "tool", tool="db_query__shop",
              args={"sql": "SELECT SUM(amount) AS gmv, COUNT(*) AS orders FROM orders"}),
         node("write", "report", instructions="写周报", numbers="strict", on_violation="fail",
-             claims="require_citation"),
+             claims="require_citation", analysis_claims="strict"),
         node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], contract=contract),
     ]
     return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}

backend/tests/test_copilot_prompt_evidence.py

@@ -202,7 +202,7 @@ def ask_report_graph() -> dict:
             g("start", "input", "问题"),
             g("ask", "agent", "查数", tools=["db_query__shop"], prompt="查活跃用户数和用户总数", approval="dangerous"),
             g("report", "report", "写答案", instructions="回答：{{ input.question }}", numbers="strict",
-              on_violation="fail", claims="require_citation"),
+              on_violation="fail", claims="require_citation", analysis_claims="strict"),
             g("out", "output", "输出答案", fields=[{"name": "answer", "value": "{{ nodes.report.text }}"}]),
         ],
         "edges": [{"id": f"e{i}", "source": a, "target": b}

backend/tests/test_governance_analysis_claims.py

@@ -0,0 +1,182 @@
+"""门禁 G3 加的一条：受管级别的报告撰写节点要把分析说法核对设成严格（00-总方案 §3.5，D8）。
+
+- 没写（运行时是重写一次后标出）、写了 check 或 off：受管级别挡住发布（error），已发布级别只警告；写错的值交给校验
+- 自动修复一键写明；一键升级成受管、按口径出具起草的报告节点照同一份要求写（REPORT_POLICY）
+- 已经发布的受管版本照常发起正式运行（报告节点按缺省的 check 核、只标注），下次发布才按新规则查
+- 白话里说到分析说法
+"""
+from __future__ import annotations
+
+import asyncio
+import copy
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage
+
+from app.db.base import SessionLocal
+from app.db.models import Run, Workflow, WorkflowVersion
+from app.engine.autofix import apply_fixes, plan_fixes
+from app.engine.governance import REPORT_POLICY, lint_for_publish, publish_issues
+from app.engine.labels import field_label, option_label
+from app.engine.runner import run_manager
+from app.engine.schema import GraphSpec, validate_graph
+from app.main import app
+from app.plain import plain_of
+
+STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation", "analysis_claims": "strict"}
+
+
+def node(nid, ntype, label=None, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}
+
+
+def traceable(report: dict) -> dict:
+    """input → 查库 → 口径卡 → 报告撰写 → 出具（引用模式契约）：别的受管要求都满足。"""
+    nodes = [
+        node("start", "input", "周期", fields=[{"name": "week", "default": "2026-W37"}]),
+        node("fetch", "tool", "取数", tool="db_query__shop", args={"sql": "SELECT SUM(amount) AS gmv FROM orders"}),
+        node("card", "metrics", "周报口径卡", caliber="周报口径", metrics=[
+            {"id": "gmv", "name": "销售额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"}]),
+        node("write", "report", "报告撰写", instructions="写周报", **report),
+        node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
+             contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"], "strict": True}),
+    ]
+    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+def policy_issues(graph: dict, level: str = "governed") -> list:
+    return [i for i in lint_for_publish(GraphSpec.model_validate(graph), level=level).issues
+            if i.code == "governed.report_policy"]
+
+
+def test_the_rule_is_part_of_the_report_policy():
+    assert ("analysis_claims", "strict") in [(k, v) for k, v, _ in REPORT_POLICY]
+
+
+@pytest.mark.parametrize("value, now", [(None, "当前未设置"), ("check", "当前为「重写一次后标出」"),
+                                        ("off", "当前为「不核对」")])
+def test_a_governed_report_must_check_analysis_claims_strictly(value, now):
+    report = {k: v for k, v in STRICT.items() if k != "analysis_claims"}
+    if value is not None:
+        report["analysis_claims"] = value
+    [issue] = policy_issues(traceable(report))
+    assert issue.level == "error" and issue.node_id == "write" and issue.field == "analysis_claims"
+    assert f"「{field_label('analysis_claims')}」设为「{option_label('analysis_claims', 'strict')}」（{now}）" \
+        in issue.message
+    assert "「未引用的数字」" not in issue.message, "别的几项都写明了，不再点名"
+    assert [i.level for i in policy_issues(traceable(report), "published")] == ["warning"]
+
+
+def test_strict_passes_and_graph_defaults_count():
+    assert policy_issues(traceable(STRICT)) == []
+    graph = traceable({k: v for k, v in STRICT.items() if k != "analysis_claims"})
+    graph["defaults"] = {"analysis_claims": "strict"}
+    assert policy_issues(graph) == []
+
+
+def test_a_bad_value_is_left_to_validation():
+    graph = traceable({**STRICT, "analysis_claims": "loose"})
+    assert policy_issues(graph) == []
+    [bad] = [i for i in validate_graph(GraphSpec.model_validate(graph)).issues
+             if i.code == "report.analysis_claims_invalid"]
+    assert bad.level == "error" and bad.field == "analysis_claims"
+
+
+def test_the_autofix_writes_it_in():
+    graph = traceable({k: v for k, v in STRICT.items() if k != "analysis_claims"})
+    spec = GraphSpec.model_validate(graph)
+    fixes = {f["id"]: f for f in plan_fixes(spec, publish_issues(spec, level="governed"), level="governed")}
+    fix = fixes["governed.report_policy:write"]
+    assert fix["kind"] == "auto" and fix["preview"] == {"field": "analysis_claims", "before": None, "after": "strict"}
+    out = apply_fixes(graph, [fix["id"]], level="governed")
+    assert out["ok"] is True
+    write = next(n for n in out["graph"]["nodes"] if n["id"] == "write")
+    assert write["data"]["config"]["analysis_claims"] == "strict"
+
+
+def test_a_governed_upgrade_writes_the_strict_setting():
+    from app.engine.upgrade import _policy
+
+    assert _policy("governed", {"numbers": "strict"}, {})["analysis_claims"] == "strict"
+    assert _policy("governed", {"analysis_claims": "strict"}, {}).get("analysis_claims") is None
+    assert _policy("published", {}, {}) == {}
+
+
+def test_the_plain_wording_mentions_analysis_claims():
+    plain = plain_of("governed.report_policy")
+    assert "分析结论" in plain.why and "分析说法核对" in plain.action
+
+
+def test_template_nine_stays_publishable_as_governed():
+    from app.seed import TEMPLATES
+
+    tpl = next(t for t in TEMPLATES if t["name"].startswith("⑨"))
+    graph = copy.deepcopy({"nodes": tpl["nodes"], "edges": tpl["edges"]})
+    assert next(n for n in graph["nodes"] if n["type"] == "report")["data"]["config"]["analysis_claims"] == "strict"
+    assert policy_issues(graph) == []
+
+
+@pytest.fixture
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def test_an_already_governed_version_without_the_setting_keeps_running(client, engine_up, monkeypatch):
+    """升级前按受管发布的版本（报告节点没写分析说法核对）照常发起正式运行，按缺省的 check 核；再发布时才按新规则查。"""
+    from app.providers import mock_model
+
+    monkeypatch.setattr(mock_model.MockChatModel, "_decide",
+                        lambda self, messages: AIMessage(content="本周销售额 [[m:gmv]]，下降主要来自华东。[[see:m:gmv]]"))
+    graph = traceable({k: v for k, v in STRICT.items() if k != "analysis_claims"})
+    graph["nodes"][1] = node("fetch", "transform", "取数", mode="expression", expression="{'gmv': 300.5}",
+                             assign_to="kpi")
+    graph["nodes"][2]["data"]["config"]["metrics"] = [
+        {"id": "gmv", "name": "销售额", "unit": "元", "expression": "vars.kpi.gmv"}]
+    async with SessionLocal() as session:
+        wf = Workflow(name="分析说法-老受管版本", graph=graph, status="governed", published_version=1)
+        session.add(wf)
+        await session.flush()
+        session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph, level="governed"))
+        await session.commit()
+        wf_id = wf.id
+    r = await client.post("/api/runs", json={"workflow_id": wf_id, "run_class": "formal"})
+    assert r.status_code == 201, r.text
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, r.json()["id"])
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            break
+    assert row.status == "succeeded", row.error
+    declared = row.output["_issuance"]["analysis_claims"]
+    assert declared["mode"] == "check" and declared["counted"] is False
+    check = (await client.post(f"/api/workflows/{wf_id}/publish-check", json={"level": "governed"})).json()
+    assert check["ok"] is False
+    assert any(i["code"] == "governed.report_policy" and i["field"] == "analysis_claims"
+               for i in check["issues"] if i["level"] == "error")
+
+
+@pytest.mark.parametrize("after", ["check", "off", None])
+def test_loosening_analysis_claims_from_strict_is_refused(after):
+    from app.engine.autofix import forbidden_changes
+
+    old = traceable(STRICT)
+    new = copy.deepcopy(old)
+    write = next(n for n in new["nodes"] if n["id"] == "write")["data"]["config"]
+    if after is None:
+        write.pop("analysis_claims")
+    else:
+        write["analysis_claims"] = after
+    [reason] = forbidden_changes(old, new)
+    assert "「报告撰写」" in reason and "「分析说法核对」" in reason and "「重写后仍有的计入缺口」" in reason
+    # 往严里改不算
+    assert forbidden_changes(new, old) == []

backend/tests/test_governance_judge.py

@@ -19,7 +19,7 @@ from app.engine.governance import lint_for_publish, publish_issues
 from app.engine.judge import JUDGE_DEFAULTS
 from app.engine.schema import GraphSpec, validate_graph
 
-STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation"}
+STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation", "analysis_claims": "strict"}
 JUDGED = {**STRICT, "claims": "judge", "judge": {"max_cost_usd": 0.05}}
 BUDGET = "governed.judge_budget"
 
@@ -98,7 +98,7 @@ def test_the_budget_follows_the_same_config_rule_as_run_time():
                       defaults={"judge": {"max_cost_usd": 0.02}})
     assert [i.field for i in by_code(graph, BUDGET)] == ["judge.max_cost_usd"]
     # claims 本身也可以来自图级 defaults
-    graph = traceable(report={"numbers": "strict", "on_violation": "fail"},
+    graph = traceable(report={"numbers": "strict", "on_violation": "fail", "analysis_claims": "strict"},
                       defaults={"claims": "judge", "judge": {"max_cost_usd": None}})
     assert by_code(graph, BUDGET) == [] and by_code(graph, "governed.report_policy") == []
 

backend/tests/test_governance_report_lint.py

@@ -25,7 +25,7 @@ from app.main import app
 from app.seed import TEMPLATES
 
 #: 报告撰写节点在受管模板里要写明的三项
-STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation"}
+STRICT = {"numbers": "strict", "on_violation": "fail", "claims": "require_citation", "analysis_claims": "strict"}
 
 G_CODES = {
     "governed.report_from_required", "governed.exit_text_source", "governed.text_bypass",
@@ -277,7 +277,7 @@ def test_a_looser_setting_is_named(key, value):
 
 def test_graph_defaults_count_like_they_do_at_run_time():
     graph = traceable(report={"claims": "require_citation"})
-    graph["defaults"] = {"numbers": "strict", "on_violation": "fail"}
+    graph["defaults"] = {"numbers": "strict", "on_violation": "fail", "analysis_claims": "strict"}
     assert by_code(graph, "governed.report_policy") == []
     graph["defaults"]["on_violation"] = "flag"
     assert [i.field for i in by_code(graph, "governed.report_policy")] == ["on_violation"]

backend/tests/test_merge_node.py

@@ -267,7 +267,8 @@ def test_publish_checks_treat_the_merge_as_traceable_evidence():
     graph = store_graph()
     for n in graph["nodes"]:
         if n["id"] == "write":
-            n["data"]["config"].update(numbers="strict", on_violation="fail", claims="require_citation")
+            n["data"]["config"].update(numbers="strict", on_violation="fail", claims="require_citation",
+                                        analysis_claims="strict")
         if n["id"] == "out":
             n["data"]["config"]["contract"] = {"report_from": "write", "metrics_from": ["card"],
                                                "required": ["conversion"], "strict": True}

backend/tests/test_publish_check_api.py

@@ -38,7 +38,7 @@ def weekly(**contract_overrides) -> dict:
             {"id": "gmv", "name": "销售额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"},
             {"id": "orders", "name": "订单数", "unit": "单", "expression": "cell(nodes.fetch, 0, 'orders')"}]),
         node("write", "report", "报告撰写", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
              contract={k: v for k, v in contract.items() if v is not None}),
     ]

backend/tests/test_publish_sqlcheck.py

@@ -57,7 +57,7 @@ def weekly(source: str, sql: str) -> dict:
         node("card", "metrics", "周报口径卡", caliber="周报口径", metrics=[
             {"id": "gmv", "name": "订单金额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"}]),
         node("write", "report", "报告撰写", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
              contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"], "strict": True}),
     ]

backend/tests/test_version_level.py

@@ -52,7 +52,7 @@ def weekly() -> dict:
         node("card", "metrics", caliber="周报口径", metrics=[
             {"id": "gmv", "name": "销售额", "unit": "元", "expression": "vars.kpi.gmv"}]),
         node("write", "report", instructions="写周报",
-             numbers="strict", on_violation="fail", claims="require_citation"),
+             numbers="strict", on_violation="fail", claims="require_citation", analysis_claims="strict"),
         node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
              contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"], "strict": True}),
     ]