双盲复算：盲算 agent 的任务里带上本次运行的输入参数

原流程的 SQL 常常是拿运行输入拼出来的（统计期的起止、地区），盲算 agent 只看口径说明不知道算哪一期，就算不出同一个数。任务里
加一行本次运行输入里的标量参数（最多 12 项、每项截到 40 字），只给参数、不给 SQL。分组结果的测试顺带核对任务里有统计期参数、
没有原流程算出的数。

backend/app/engine/nodes/dual_check.py

@@ -68,9 +68,10 @@ DIFF_TIMEOUT_S = 15.0
 #: 交给盲算 agent 的数据目录最多这么多字（表清单、各表的说明分开算），再多就请它自己用 db_schema 查
 _INDEX_CHARS = 3_000
 _NOTES_CHARS = 6_000
-#: 交给盲算 agent 的同一行其他列最多几列、每个值最多几个字
+#: 交给盲算 agent 的同一行其他列、运行输入最多几项、每个值最多几个字
 _DIMS = 6
 _DIM_CHARS = 40
+_INPUTS = 12
 #: 卡上给容器输入记的大概（metrics._brief）：「[12 项]」「{3 个字段}」，写进计算式时不当成字面量
 _BRIEF = re.compile(r"^(?:\[\d+ 项\]|\{\d+ 个字段\})$")
 
@@ -102,13 +103,15 @@ BLIND_TASK_PROMPT = (
     "口径说明：{definition}\n"
     "计算式：{formula}\n"
     "你要算的是计算式里的 {label}{whole}：原流程里它是一次查询结果中「{column}」列的值{dims}。只交这一个数。\n"
-    "数据源：{source}\n\n"
+    "数据源：{source}\n"
+    "{inputs}\n"
     "{catalog}"
 )
 BLIND_UNIT_PROMPT = "（单位：{unit}）"
 BLIND_NO_DEFINITION_PROMPT = "（未提供，按指标名称和数据目录理解）"
 BLIND_WHOLE_PROMPT = "（这个指标就是它本身）"
 BLIND_DIMS_PROMPT = "，所在行的其他列：{dims}"
+BLIND_INPUTS_PROMPT = "本次运行的输入（统计期之类的参数在这里）：{inputs}\n"
 BLIND_INDEX_PROMPT = "数据源里的表（表名｜中文名｜粒度）：\n{index}"
 BLIND_NOTES_PROMPT = "数据目录（标「推断，未确认」的项是推断，使用前请核实）：\n{notes}"
 BLIND_NO_CATALOG_PROMPT = "这个数据源还没有数据目录，请用 db_schema 工具查看表结构。"
@@ -351,10 +354,13 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An
     sources = _Sources()
     load, outputs = _loader(state)
     aliases = _original_aliases(state)
+    inputs = _run_inputs(state)
     rows: list[dict[str, Any]] = []
     jobs: dict[str, dict[str, Any]] = {}
     for metric in targets:
         row, metric_jobs = await _plan_metric(state, metric, options, sources, aliases, load, outputs)
+        for job in metric_jobs.values():
+            job["inputs"] = inputs
         rows.append(row)
         for key, job in metric_jobs.items():
             # 两个指标用到同一格只盲算一次，按先出现的那个指标说清要算什么；先出现的没写口径说明、后一个写了，用后一个
@@ -373,6 +379,19 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An
     return {"rows": rows, "results": results, "model": model, "same_model": same, "usage": usage}
 
 
+def _run_inputs(state: GraphState) -> list[str]:
+    """这次运行的输入里的标量（统计期、地区这类参数）：原流程的 SQL 常常是拿它们拼出来的，盲算 agent 不知道统计期
+    就算不出同一个数。只给参数，不给 SQL。"""
+    out: list[str] = []
+    for key, value in (state.get("input") or {}).items():
+        if isinstance(key, str) and key and (value is None or isinstance(value, (str, int, float, bool))):
+            text = str(value).replace("\n", " ")
+            out.append(f"{key} = {text[:_DIM_CHARS]}")
+        if len(out) >= _INPUTS:
+            break
+    return out
+
+
 def _original_aliases(state: GraphState) -> dict[str, str]:
     """原流程的查询快照在这次运行的证据台账里的编号（Q1、Q2…，和报告取全部上游查询时的编号一致）。"""
     ledger = [e for e in state.get("evidence") or [] if isinstance(e, dict)]
@@ -576,7 +595,8 @@ def _task_text(job: dict[str, Any], catalog: str) -> str:
         definition=job["definition"] or BLIND_NO_DEFINITION_PROMPT, formula=job["formula"], label=job["label"],
         whole=BLIND_WHOLE_PROMPT if job["formula"].strip() == job["label"] else "", column=job["column"],
         dims=BLIND_DIMS_PROMPT.format(dims="、".join(job["dims"])) if job["dims"] else "",
-        source=job["source"], catalog=catalog)
+        source=job["source"], catalog=catalog,
+        inputs=BLIND_INPUTS_PROMPT.format(inputs="；".join(job["inputs"])) if job.get("inputs") else "")
 
 
 async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Options, settings: Mapping[str, Any],

backend/tests/test_dual_check_node.py

@@ -106,8 +106,8 @@ async def wait(run_id: str, statuses=("succeeded", "failed", "interrupted")) ->
     raise AssertionError(f"运行 {run_id} 停在 {row.status}：{row.error}")
 
 
-async def start(spec: dict, **kwargs) -> Run:
-    run = await run_manager.start(graph=spec, input_payload={}, **kwargs)
+async def start(spec: dict, inputs: dict | None = None, **kwargs) -> Run:
+    run = await run_manager.start(graph=spec, input_payload=inputs or {}, **kwargs)
     return await wait(run.id)
 
 
@@ -185,14 +185,17 @@ async def test_agreeing_check_passes_without_asking(monkeypatch, shop):
 
 
 async def test_grouped_rows_tell_the_blind_agent_which_group(monkeypatch, shop):
-    """结果分了组：盲算 agent 要知道算的是哪一组（同一行的维度列），但看不到同一行别的数。"""
+    """结果分了组：盲算 agent 要知道算的是哪一组（同一行的维度列），但看不到同一行别的数；统计期这类参数在运行输入里。"""
     name, _ = shop
     metric = {**GMV, "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
     blind = Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")})
-    run = await start(graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}}))
+    spec = graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}})
+    spec["nodes"][0]["data"]["config"]["fields"] = [{"name": "week_start"}, {"name": "week_end"}]
+    run = await start(spec, {"week_start": "2026-09-01", "week_end": "2026-09-07"})
     assert run.status == "succeeded", run.error
     [task] = blind.tasks
-    assert "region 为「华东」" in task and "（未提供" in task
+    assert "region 为「华东」" in task and "（未提供" in task and "300" not in task
+    assert "week_start = 2026-09-01" in task and "week_end = 2026-09-07" in task
     out = dual_output(run, await events(run.id))
     assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed"