合并查询节点（三）：画图时校验输入（不存在、不是查询、不在上游、Agent 的结果）、别名和合并 SQL；助手的节点说明和搭图规则写清合并查询的使用时机

backend/app/api/copilot.py

@@ -163,6 +163,18 @@ NODE_REFERENCE = """\
 - supervisor：多 agent 协作。config: {goal, agents:[{name, description, system, tools, model}], max_rounds}
   成员停不下来等人：需要审批的 MCP / 自定义工具在成员手里不会执行。要用这类工具，交给团队外的 agent 节点
 - tool：直接调一个工具。config: {tool: 工具名, args: {...}, assign_to}
+- merge：合并查询，把几次查询的结果在库外按键合并成一张表。config: {inputs: {别名: 上游查询节点id}, sql, assign_to}
+  **只在数据分在不同的库、要按行对齐时用**：先在各自的库里用 SQL 聚合到相同的键和粒度（比如「日期 + 门店」），
+  再由 merge 按键合并。同一个库里的数据写成一条 SQL（JOIN / WITH），不要拆成两次查询再合并；只组合几个单值时
+  用口径卡的 cell()，不用 merge。比率、增幅、占比这类派生计算仍然写在口径卡里，不写进合并 SQL。
+  inputs 的值只能是调用 db_query__ 工具的 tool 节点或另一个 merge 节点，而且要连在 merge 之前；agent 查过的库
+  不能当输入（它可能查了好几次），要合并的那条查询单独放进 tool 节点。别名就是 sql 里的表名，只用英文字母、数字
+  和下划线（例如 s、v）。sql 是一条 SQLite 的 SELECT 或 WITH，例如：SELECT s.日期, s.门店, s.订单数, v.到店人数
+  FROM s JOIN v ON s.日期 = v.日期 AND s.门店 = v.门店。合并读的是每个输入的完整结果：任何一个输入被截断（超过
+  查询的行数上限）merge 就会失败，所以源查询要先聚合、缩小范围；两边的键类型要一致（文本 '001' 对数值 1 会给警告）；
+  结果比行数最多的输入还多，说明合并键不唯一（也会给警告）。结果和查询同形：报告按先后编号引用 [[v:Qn.r0.列名]]，
+  口径卡写 cell(nodes.merge节点id, 行, '列名')。选取项直接写输入的列（可以用 AS 改名）时，报告里的数点得开、能追到
+  输入的那一格；表达式算出来的列只有表级来历
 - code：沙箱里跑代码。config: {language: python|bash|node, code, timeout, network, assign_to}
   evidence_role：source（取数：产出本身就是源数据）/ compute（计算，默认）。负责取数的代码写 source；
   compute 的产出喂给口径卡会被提示「核对不了出处」——业务计算写进口径卡的表达式。
@@ -194,7 +206,7 @@ NODE_REFERENCE = """\
 - report：报告撰写（带引用）。config: {instructions, system, metrics_from:[口径卡id], numbers, on_violation, max_repairs,
   claims, entities, assign_to}
   **凡是给人看的、带数字的报告、分析、结论，都用 report，不要用 llm**。它自动收集上游的口径卡指标、查询结果
-  （tool 节点和 agent 查过的库，按查询先后编成 Q1、Q2…）、查库当时的表结构、知识库检索（按先后编成 K1、K2…）
+  （tool 节点和 agent 查过的库、merge 的合并结果，按查询先后编成 Q1、Q2…）、查库当时的表结构、知识库检索（按先后编成 K1、K2…）
   和运行输入，模型只能写引用标记，数值由系统从证据里取出来渲染；模型自己写的数字会被打回重写：
   - [[m:指标id]]：口径卡指标。比率、增幅、占比、差值这类派生计算只能先在口径卡里登记再引用
   - [[v:Q1.r0.列名]]：第 1 次查询结果里第 0 行那一列的原值（行号从 0 数）
@@ -234,6 +246,8 @@ NODE_REFERENCE = """\
 2. 能用 SUM / COUNT 做的聚合放在 SQL 里，口径卡只做标量运算
 3. 不要用 code 节点做业务计算；code 只做格式转换（负责取数的 code 标 evidence_role: source）
 4. 不要用 transform 解析 agent / llm 的文字：口径卡读 cell(nodes.查库节点id, 行, '列名') 或 vars.<agent 的 assign_to>.字段
+5. 数据在两个库、要按行对齐时：两个 tool 节点各自聚合到相同的键和粒度 → merge 按键合并 → 口径卡算派生指标 → report。
+   同一个库写成一条 SQL；只组合几个单值用口径卡的 cell()；不要用 code 节点合并（出处会断，喂给口径卡时核对不了）
 
 模型字段（llm / agent / supervisor / report 的 config.model）：
 - **不要填**。留空表示跟随当前供应商的默认模型，这几乎总是对的。
@@ -1154,7 +1168,8 @@ def _evidence_source(node: Any) -> bool:
     if node.type == NodeType.INPUT:
         fields = node.config.get("fields")
         return isinstance(fields, list) and any(_numeric_input(f) for f in fields)
-    return node.type in (NodeType.METRICS, NodeType.TOOL, NodeType.RETRIEVE, NodeType.SUBGRAPH, NodeType.SUPERVISOR)
+    return node.type in (NodeType.METRICS, NodeType.TOOL, NodeType.MERGE, NodeType.RETRIEVE, NodeType.SUBGRAPH,
+                         NodeType.SUPERVISOR)
 
 
 def _unsourced_reports(spec: GraphSpec, changed: set[str]) -> list[dict[str, Any]]:

backend/app/engine/schema.py

@@ -417,6 +417,70 @@ def _check_report_sources(node: GraphNode, spec: GraphSpec, result: ValidationRe
                        node_id=node.id, field="metrics_from", code=code)
 
 
+def _produces_query(node: GraphNode) -> bool:
+    """这个节点的产出是一份存成快照的查询结果：调用数据库查询工具的「调用工具」节点，或另一个合并查询。"""
+    if node.type == NodeType.MERGE:
+        return True
+    return node.type == NodeType.TOOL and str(node.config.get("tool") or "").startswith("db_query__")
+
+
+def _check_merge(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> None:
+    """合并查询的输入和合并 SQL。规则和执行时（engine/merge_query.py）同一套，画图时就说破，不等运行到这一步才失败。
+
+    输入只认能稳定取到一份查询快照的节点。Agent 查过的库不行：一次循环里可能查好几次，取哪一次说不准，
+    拿错一次合并出来的数照样像模像样——所以报 error，并说清楚把那条查询单独放进「调用工具」节点。
+    """
+    from app.engine.merge_query import MAX_INPUTS, alias_problem, sql_problem
+
+    cfg = node.config
+    inputs = cfg.get("inputs")
+    if not isinstance(inputs, dict) or not inputs:
+        result.add("「合并查询」还没有配置输入：至少选择一个上游的查询节点，并给它起一个别名（合并 SQL 里的表名）",
+                   node_id=node.id, field="inputs", code="merge.inputs_missing")
+    else:
+        if len(inputs) > MAX_INPUTS:
+            result.add(f"「合并查询」最多 {MAX_INPUTS} 个输入，这里有 {len(inputs)} 个。请先在源库里合并一部分",
+                       node_id=node.id, field="inputs", code="merge.inputs_missing")
+        nodes = spec.node_map()
+        upstream = _ancestors(spec, node.id)
+        seen: dict[str, str] = {}
+        for alias, source in inputs.items():
+            field = f"inputs.{alias}"
+            if problem := alias_problem(alias):
+                result.add(f"「合并查询」的{problem}", node_id=node.id, field=field, code="merge.alias_invalid")
+            elif alias.lower() in seen:
+                result.add(f"「合并查询」的别名「{seen[alias.lower()]}」和「{alias}」只差大小写：SQLite 的表名不分大小写，"
+                           "会被当成同一张表。请换一个别名", node_id=node.id, field=field, code="merge.alias_invalid")
+            else:
+                seen[alias.lower()] = alias
+            target = nodes.get(source) if isinstance(source, str) else None
+            who = f"「合并查询」的输入「{alias}」"
+            if not isinstance(source, str) or not source:
+                result.add(f"{who}还没有选择节点", node_id=node.id, field=field, code="merge.input_invalid")
+            elif target is None:
+                result.add(f"{who}选择了「{source}」，但工作流中不存在这个节点", node_id=node.id, field=field,
+                           code="merge.input_invalid")
+            elif target.type == NodeType.AGENT:
+                result.add(f"{who}选择了「{target.title}」（Agent）：Agent 可能查询多次，无法确定合并哪一次的结果。"
+                           "请把要合并的那条查询单独放进「调用工具」节点（选择数据库查询工具），再把它选作输入",
+                           node_id=node.id, field=field, code="merge.input_invalid")
+            elif target.type == NodeType.TOOL and not _produces_query(target):
+                result.add(f"{who}选择的「{target.title}」调用的不是数据库查询工具：合并查询只能合并数据库查询的结果",
+                           node_id=node.id, field=field, code="merge.input_invalid")
+            elif not _produces_query(target):
+                result.add(f"{who}选择的「{target.title}」是「{type_label(target.type)}」节点，不产出查询结果。"
+                           "输入只能是调用数据库查询工具的「调用工具」节点，或另一个「合并查询」节点",
+                           node_id=node.id, field=field, code="merge.input_invalid")
+            elif source not in upstream:
+                result.add(f"{who}选择的「{target.title}」不在合并查询的上游：合并时它还没有运行。请把它连到合并查询之前",
+                           node_id=node.id, field=field, code="merge.input_invalid")
+    sql = cfg.get("sql")
+    if not isinstance(sql, str) or not sql.strip():
+        result.add("「合并查询」还没有填写合并 SQL", node_id=node.id, field="sql", code="merge.sql_invalid")
+    elif problem := sql_problem(sql):
+        result.add(problem, node_id=node.id, field="sql", code="merge.sql_invalid")
+
+
 #: 报告撰写节点的结论句策略（claims）。off：结论句不参与判档（默认，保持以前的行为）；
 #: require_citation：没挂引用的结论句计入缺口、按出具档位降档。这两档是确定性的（发布前自动修复
 #: 给人选的也只有这两档）
@@ -605,6 +669,8 @@ def validate_graph(spec: GraphSpec) -> ValidationResult:
         elif node.type == NodeType.REPORT:
             _check_report_sources(node, spec, result)
             _check_report_claims(node, spec, result)
+        elif node.type == NodeType.MERGE:
+            _check_merge(node, spec, result)
         elif node.type == NodeType.OUTPUT:
             contract = cfg.get("contract")
             if contract and not isinstance(contract, dict):

backend/tests/test_merge_node.py

@@ -260,6 +260,22 @@ def test_merge_inputs_count_as_references_to_the_upstream_nodes():
     assert any(r.node_id == "merge" for r in refs["nodes.q_sales"])
 
 
+def test_publish_checks_treat_the_merge_as_traceable_evidence():
+    """受管级别的发布检查：合并结果喂给口径卡、报告，和数据源查询一样算可追溯的证据，不在合并节点上报任何问题。"""
+    from app.engine.governance import publish_issues
+
+    graph = store_graph()
+    for n in graph["nodes"]:
+        if n["id"] == "write":
+            n["data"]["config"].update(numbers="strict", on_violation="fail", claims="require_citation")
+        if n["id"] == "out":
+            n["data"]["config"]["contract"] = {"report_from": "write", "metrics_from": ["card"],
+                                               "required": ["conversion"], "strict": True}
+    issues = publish_issues(GraphSpec.model_validate(graph), level="governed")
+    assert [i for i in issues if i.node_id == "merge"] == []
+    assert [i for i in issues if i.level == "error"] == [], [i.message for i in issues if i.level == "error"]
+
+
 def test_labels_and_governance_know_the_type():
     from app.engine.governance import _traceable
     from app.engine.labels import field_label, type_label