账本接口补几样界面要画的东西：对账的入口字段、配列选项、对账卡的分组，刚跑完未封存时答 pending

账本前端（第 4 波 W4-c）照「界面层不放逻辑」只照服务端给的画，这几样以前接口不给，界面只能自己猜：

- 「看差多少」那一项（reconcile.actions_of）带上口径的入口字段 fields（名字、必填、有没有缺省值，比对工件里记的那一份）：
  核对参数的表单照它画，缺的参数也写得出是哪一项
- 认不出列怎么配（409 ledger_reconcile_needs_pairs）另带 options：这次结果里是数、没遮罩的列，口径卡上的指标（id、名字、
  单位）；ReconcileRefused 加 extra，api/ledger 的对账接口把它并进返回体（CodedHTTPException 的 extra，改一行）
- 对过的账（reconciled_view）带 card 和 groups：每一组对照的这次 / 口径 / 差值 / 相对差各是哪个指标、这一组怎么称呼
  （从对账运行的图里读，不按命名规则拼），界面按 /evidence/metrics/{card}/{指标} 取值、三个数各自点得开
- GET /api/runs/{id}/ledger：runner 先发 run.finished、落库，再封存、再排比对，页面收到 run.finished 立刻来取时
  还没封存。成功结束、要比对的运行这时答 pending（不另排按需比对），不再说成「运行还没有封存，不比对」
- 测试：对账的三样（单行、多行），刚跑完未封存答 pending、不排按需比对，失败的照旧 ineligible

backend/app/api/ledger.py

@@ -159,7 +159,7 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]:
             query=payload.query, workflow_id=payload.workflow_id, version=payload.version, inputs=payload.inputs,
             confirm=payload.confirm, pairs=payload.pairs), principal=current_principal())
     except ledger_reconcile.ReconcileRefused as e:
-        raise CodedHTTPException(e.status_code, e.message, e.code) from e
+        raise CodedHTTPException(e.status_code, e.message, e.code, extra=e.extra) from e
     return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)}
 
 

backend/app/ledger/reconcile.py

@@ -77,11 +77,16 @@ ROWS_CALIBER_COLUMNS = "认不出口径取数的结果列，按位置对不上"
 
 
 class ReconcileRefused(Exception):
-    def __init__(self, status_code: int, message: str, code: str = REFUSED) -> None:
+    """不能对账。extra 是给界面的几样（并进 409 的返回体）：认不出列怎么配时的 options（可选的列、口径卡上的指标），界面
+    照它画「选要对照的指标」那一步，不自己去猜有哪些列。"""
+
+    def __init__(self, status_code: int, message: str, code: str = REFUSED, *,
+                 extra: dict[str, Any] | None = None) -> None:
         super().__init__(message)
         self.status_code = status_code
         self.message = message
         self.code = code
+        self.extra = dict(extra or {})
 
 
 # ==========================================================================
@@ -106,7 +111,8 @@ def _needs_check(cand: dict[str, Any], fields: list[dict[str, Any]], hide: Any)
 
 def actions_of(content: dict[str, Any], names: dict[str, str], hide: Any = None) -> list[dict[str, Any]]:
     """比对工件里每条有相关口径的查询、每份相关口径一项「看差多少」：{kind: reconcile, query, workflow_id, version, node_id,
-    needs_check, check_reason?, binding（遮过的）}。界面据此摆按钮、要不要先核对参数。"""
+    needs_check, check_reason?, binding（遮过的）, fields}。界面据此摆按钮、要不要先核对参数；fields 是那一版口径的入口字段
+    （名字、必填、有没有缺省值，比对工件里记的那一份），核对参数的表单照它画（缺的参数也写得出是哪一项）。"""
     from app.ledger.labels import LEVEL_TEXT, binding_view
 
     hide = hide or (lambda _value: False)
@@ -124,7 +130,10 @@ def actions_of(content: dict[str, Any], names: dict[str, str], hide: Any = None)
                     "node_id": cand.get("node_id"), "workflow_name": names.get(str(cand.get("workflow_id") or "")),
                     "level_text": LEVEL_TEXT.get(str(cand.get("level") or ""), ""),
                     "needs_check": why is not None, "binding": binding_view(cand.get("binding"), hide),
-                    "fallback": bool((cand.get("diff") or {}).get("fallback"))}
+                    "fallback": bool((cand.get("diff") or {}).get("fallback")),
+                    "fields": [{"name": str(f.get("name")), "required": bool(f.get("required")),
+                                "has_default": bool(f.get("has_default"))}
+                               for f in fields.get(cand.get("caliber")) or [] if f.get("name")]}
             if why is not None:
                 item["check_reason"] = why
             out.append(item)
@@ -141,11 +150,45 @@ async def reconciled_runs(session: Any, run_id: str) -> list[Run]:
     return list(rows)
 
 
+#: 对账卡上一组对照的四个指标：id 的前缀（tail_nodes 的 _four）和名字的后缀
+_FOUR = (("ours", "这次"), ("caliber", "口径"), ("diff", "差值"), ("rel", "相对差"))
+_PREFIX = {"ours": "ours_", "caliber": "cal_", "diff": "diff_", "rel": "rel_"}
+
+
+def reconcile_groups(graph: Any) -> list[dict[str, str]]:
+    """对账运行的图里，对账卡上的每一组对照：{label, ours, caliber, diff, rel}（后四个是指标 id）。按卡上的先后；名字
+    是「<这一组的说法> · 这次」那样写的，label 去掉后缀。认不出的（不是派生的对账图）为空。"""
+    from app.ledger.derive import RECONCILE_CARD
+
+    nodes = (graph or {}).get("nodes") if isinstance(graph, dict) else None
+    card = next((n for n in nodes or [] if isinstance(n, dict) and n.get("id") == RECONCILE_CARD
+                 and n.get("type") == "metrics"), None)
+    metrics = (((card or {}).get("data") or {}).get("config") or {}).get("metrics") or []
+    by_id = {str(m.get("id")): m for m in metrics if isinstance(m, dict) and m.get("id")}
+    out: list[dict[str, str]] = []
+    for mid, metric in by_id.items():
+        if not mid.startswith(_PREFIX["ours"]):
+            continue
+        n = mid[len(_PREFIX["ours"]):]
+        ids = {role: f"{_PREFIX[role]}{n}" for role, _ in _FOUR}
+        if not all(i in by_id for i in ids.values()):
+            continue
+        name = str(metric.get("name") or "")
+        label = name[:-len(" · 这次")] if name.endswith(" · 这次") else name
+        out.append({"label": label, **ids})
+    return out
+
+
 def reconciled_view(run: Run) -> dict[str, Any]:
+    """GET 账本 reconciled 里的一项：哪一次对账运行、状态、对的哪条查询和哪份口径、参数从哪来；card 和 groups 是对账卡和
+    每一组对照的四个指标（界面按 /evidence/metrics/{card}/{指标} 取值、点开出处，不按命名规则自己拼）。"""
+    from app.ledger.derive import RECONCILE_CARD
+
     origin = run.origin if isinstance(run.origin, dict) else {}
     return {"run_id": run.id, "status": run.status, "query": origin.get("query"), "caliber": origin.get("caliber"),
             "workflow_id": origin.get("workflow_id"), "version": origin.get("version"),
             "inputs_from": origin.get("inputs_from"), "rows": origin.get("rows"),
+            "card": RECONCILE_CARD, "groups": reconcile_groups(run.graph),
             "created_at": run.created_at.isoformat() if run.created_at else None}
 
 
@@ -294,11 +337,16 @@ def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str
 
     by_id = {str(m.get("id")): (i, m) for i, m in enumerate(metrics)}
     multi = len(ours_rows) > 1
+    # 要人选对照时给界面的选项：这次结果里是数、没遮罩的列，口径卡上的指标（id、名字、单位）
+    numeric = [c for i, c in enumerate(ours_columns) if c.lower() not in hidden and _numeric(ours_rows, i)]
+    options = {"options": {"columns": numeric,
+                           "metrics": [{"id": str(m.get("id")), "name": str(m.get("name") or m.get("id")),
+                                        "unit": str(m.get("unit") or "")} for m in metrics if m.get("id")]}}
     if multi:
         return _plan_rows(ours_columns, ours_rows, hidden, caliber_columns, referenced, check_column)
     if chosen is not None:
         if not isinstance(chosen, list) or not chosen or not all(isinstance(p, dict) for p in chosen):
-            raise ReconcileRefused(422, PAIRS_SHAPE, NEEDS_PAIRS)
+            raise ReconcileRefused(422, PAIRS_SHAPE, NEEDS_PAIRS, extra=options)
         pairs = []
         for item in chosen:
             ours, metric = str(item.get("ours") or ""), str(item.get("metric") or "")
@@ -310,7 +358,7 @@ def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str
                               unit=str(m.get("unit") or ""), decimals=m.get("decimals")))
         return Plan(pairs=pairs)
     if fallback:
-        raise ReconcileRefused(409, PAIRS_FALLBACK, NEEDS_PAIRS)
+        raise ReconcileRefused(409, PAIRS_FALLBACK, NEEDS_PAIRS, extra=options)
     pairs = []
     if caliber_columns is not None and len(caliber_columns) == len(ours_columns):
         for index, m in enumerate(metrics):
@@ -323,7 +371,7 @@ def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str
             pairs.append(Pair(ours=ours, metric=str(m.get("id")), index=index, name=str(m.get("name") or m.get("id")),
                               unit=str(m.get("unit") or ""), decimals=m.get("decimals")))
     if not pairs:
-        raise ReconcileRefused(409, PAIRS_NONE, NEEDS_PAIRS)
+        raise ReconcileRefused(409, PAIRS_NONE, NEEDS_PAIRS, extra=options)
     return Plan(pairs=pairs)
 
 
@@ -521,4 +569,4 @@ def build(pinned: Any, *, sql: str, tool: str, plan: Plan, inputs: dict[str, Any
 
 __all__ = ["CHECK_INPUTS", "MAX_COLUMNS", "MAX_ROWS", "NEEDS_PAIRS", "ORIGIN", "REFUSED", "Pair", "Plan",
            "ReconcileRefused", "Request", "actions_of", "align_sql", "build", "plan_pairs", "reconcile",
-           "reconciled_runs", "reconciled_view", "tail_nodes"]
+           "reconcile_groups", "reconciled_runs", "reconciled_view", "tail_nodes"]

backend/app/ledger/view.py

@@ -116,6 +116,11 @@ async def run_view(run_id: str) -> dict[str, Any]:
     recorded = await store.latest(run_id)
     if recorded is None:
         why = ledger_match.ineligible(run)
+        if why == ledger_match.NOT_SEALED and run.status == "succeeded" \
+                and ledger_match.wanted(run.run_class, run.origin_kind):
+            # runner 先发 run.finished、落库，再封存、再排上比对（engine/runner._finalize）：页面收到 run.finished 立刻来取，
+            # 这一下还没封存。这不是「不比对」，是封存完就比——照实答 pending，页面按退避接着取；这里不另排按需比对
+            return {**base, "state": "pending"}
         sealed = await _sealed(run_id) if why is None else None
         if sealed is not None and not sealed.queries:
             why = ledger_match.NO_QUERIES

backend/tests/test_ledger_api.py

@@ -93,6 +93,30 @@ async def test_an_unmatched_run_is_matched_on_demand_and_says_it_looks_back(clie
     assert body["retro_note"] == "按现在的口径回看"
 
 
+async def test_a_run_that_just_finished_and_is_about_to_be_sealed_is_pending(client, monkeypatch):
+    """runner 先发 run.finished、落库，再封存、再排上比对：页面收到 run.finished 立刻来取账本，这一下运行已经成功、还没封存。
+    这不是「不比对」，是「封存完就比」——照实答 pending（不在这里另排一次按需比对：runner 封存之后自己会排），页面按退避
+    接着取；真不比的（正式运行、失败的）照旧 ineligible。"""
+    await S.publish()
+    row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
+    await ledger_match.drain()
+    from app.db.models import Run
+
+    async with SessionLocal() as session:
+        fresh = Run(workflow_name="刚跑完", status="succeeded", run_class="exploratory", origin_kind="chat", graph={},
+                    input={}, output={})
+        failed = Run(workflow_name="失败了", status="failed", run_class="exploratory", origin_kind="chat", graph={},
+                     input={}, output={})
+        session.add_all([fresh, failed])
+        await session.commit()
+        fresh_id, failed_id = fresh.id, failed.id
+    scheduled: list[str] = []
+    monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: scheduled.append(run_id) or True)
+    body = await ledger_of(client, fresh_id)
+    assert body["state"] == "pending" and scheduled == [], body
+    assert (await ledger_of(client, failed_id))["state"] == "ineligible"
+
+
 async def test_runs_that_are_not_compared_say_why(client, monkeypatch):
     row = await S.explore(monkeypatch, S.EQUIVALENT, origin="api", text="本周销售额 [[v:Q1.r0.销售额]]。")
     body = await ledger_of(client, row.id)

backend/tests/test_ledger_reconcile.py

@@ -248,3 +248,35 @@ async def test_users_read_their_reconcile_runs_only_through_the_evidence(client,
         assert (await carol.get(f"/api/runs/{done.id}/evidence")).status_code == 200
     finally:
         await auth_env.reset()
+
+
+async def test_the_reconcile_view_gives_the_panel_what_it_draws(client, monkeypatch):
+    """界面层不放逻辑（W4-c 账本前端）：「看差多少」那一项带上口径的入口字段（核对参数的表单照它画：哪几项、必填、有没有
+    缺省值）；认不出列怎么配时 409 另带可选的列和口径卡上的指标（选对照的那一步照它画）；对过的账列出对账卡和每一组的
+    这次 / 口径 / 差值 / 相对差是哪几个指标（三个数各自点得开，界面不按命名规则去拼指标 id）。"""
+    wf = await S.publish(graph=S.caliber_graph(expression="cell(nodes.fetch, 0, 'gmv') * 1"))
+    row = await S.explore(monkeypatch, S.WITH_REFUNDS)
+    [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
+    assert action["fields"] == [{"name": "start", "required": True, "has_default": False},
+                                {"name": "end", "required": True, "has_default": False},
+                                {"name": "store", "required": False, "has_default": True}]
+    body = {"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED, "confirm": True}
+    res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", json=body)
+    assert res.status_code == 409 and res.json()["code"] == ledger_reconcile.NEEDS_PAIRS
+    assert res.json()["options"] == {"columns": ["gmv"], "metrics": [{"id": "gmv", "name": "销售额", "unit": "元"}]}
+    done = await reconcile(client, row.id, {**body, "pairs": [{"ours": "gmv", "metric": "gmv"}]})
+    [listed] = (await ledger_of(client, row.id))["reconciled"]
+    assert listed["run_id"] == done.id and listed["card"] == "reconcile"
+    assert listed["groups"] == [{"label": "销售额", "ours": "ours_0", "caliber": "cal_0", "diff": "diff_0", "rel": "rel_0"}]
+
+
+async def test_multi_row_reconcile_groups_carry_the_row_labels(client, monkeypatch):
+    wf = await S.publish(graph=S.caliber_graph(BY_REGION))
+    row = await S.explore(monkeypatch, BY_REGION_ALL, text="东区 [[v:Q1.r0.gmv]]，西区 [[v:Q1.r1.gmv]]。")
+    [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
+    await reconcile(client, row.id, {"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED,
+                                     "confirm": True})
+    [listed] = (await ledger_of(client, row.id))["reconciled"]
+    assert [g["label"] for g in listed["groups"]] == ["区域=east · gmv", "区域=west · gmv"]
+    assert listed["groups"][1] == {"label": "区域=west · gmv", "ours": "ours_1_0", "caliber": "cal_1_0", "diff": "diff_1_0",
+                                   "rel": "rel_1_0"}