按口径出具也按 SQL 里参数的位置校验绑定值；去掉 certify 没用到的导入

为什么：出具用的参数来自比对时的绑定，运行时要渲染进口径的 SQL 模板。写在引号里的值带着引号、裸写的值不是数，渲染出来就不是那一版口径的 SQL 了。组合时校验助手给的参数用的是同一个判断（ledger/compose.slot_contexts / value_problem），这里共用；那张图里别的节点引用运行时变量的不归出具管，跳过不报。宁可少出具，照实说为什么不代填。

backend/app/ledger/certify.py

@@ -35,7 +35,6 @@ from typing import Any
 
 from sqlalchemy import select
 
-from app.db.base import SessionLocal
 from app.db.models import Run
 
 logger = logging.getLogger(__name__)

backend/app/ledger/compose.py

@@ -125,10 +125,18 @@ def _names(items: list[str]) -> str:
 
 def _slot_contexts(pinned: Pinned) -> dict[str, set[str]]:
     """取数 SQL 模板里每个入口字段写在哪：{字段: {quoted / bare / comment}}。认不出的抛 ComposeRefused。"""
+    return slot_contexts(pinned.fetches)
+
+
+def slot_contexts(nodes: Any, *, strict: bool = True) -> dict[str, set[str]]:
+    """几个调用工具节点的 args.sql 里每个入口字段写在哪（见 _slot_contexts）。按口径正式出具（ledger/issue）也按它查参数：
+    strict=False 时认不出写法的那条（引用了运行时的变量）跳过，不抛——那张图里别的节点怎么写不归出具管。"""
     from app.data.sql_semantics import _NotComparable, _template_slots
 
     out: dict[str, set[str]] = {}
-    for node in pinned.fetches:
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
         args = ((node.get("data") or {}).get("config") or {}).get("args")
         sql = args.get("sql") if isinstance(args, dict) else None
         if not isinstance(sql, str):
@@ -136,6 +144,8 @@ def _slot_contexts(pinned: Pinned) -> dict[str, set[str]]:
         try:
             spans, _ = _template_slots(sql)
         except _NotComparable as e:
+            if not strict:
+                continue
             raise ComposeRefused(TEMPLATE_UNREADABLE.format(reason=str(e))) from None
         for _, _, slot in spans:
             out.setdefault(slot.field, set()).add(slot.context)
@@ -239,4 +249,6 @@ class Composer:
 
 
 __all__ = ["OP", "REFUSED", "ComposeRefused", "Composed", "Composer", "Option", "check_inputs", "instantiate",
-           "signature"]
+           "signature", "slot_contexts", "value_problem"]
+
+value_problem = _value_problem

backend/app/ledger/issue.py

@@ -251,6 +251,13 @@ async def plan_issue(session: Any, run: Run | None, content: dict[str, Any] | No
     plan.inputs = values
     if errors := validate_form(input_schema(fields), values):
         return plan.refuse(INPUTS.format(problems="；".join(e["message"] for e in errors)))
+    # 参数运行时要渲染进口径的 SQL 模板：写在引号里的不能带引号、裸写的只能是数（ledger/compose 同一个判断）
+    from app.ledger.compose import slot_contexts, value_problem
+
+    contexts = slot_contexts([n for n in (row.graph or {}).get("nodes") or [] if n.get("type") == "tool"],
+                             strict=False)
+    if problem := next((p for k, v in values.items() if (p := value_problem(k, v, contexts.get(k, set())))), None):
+        return plan.refuse(f"{problem}，不代填出具")
     if principal.mode != "off" and not principal.at_least("builder"):
         plan.granted_by_app = await _granting_app(session, principal, workflow_id, int(version))
         if plan.granted_by_app is None:

backend/tests/test_ledger_issue.py

@@ -333,3 +333,21 @@ async def test_auto_issue_goes_by_the_askers_permissions_when_login_is_on(monkey
     launched = await ledger_issue.auto_issue(row.id)
     [formal] = await settle(row.id)
     assert launched == formal.id and formal.started_by == f"erin_{row.id[:6]}" and formal.origin["auto"] is True
+
+
+async def test_binding_values_that_would_break_the_sql_are_not_filled_in(client, monkeypatch):
+    """绑定出来的值要渲染进口径的 SQL 模板：写在引号里的值带着引号（这次的 SQL 里写的是转义过的 ''，绑定照原样取出）就不代填
+    出具——和组合时校验助手给的参数是同一个判断，宁可少出具。"""
+    template = "SELECT SUM(amount) AS gmv FROM orders WHERE status = '{{ input.status }}'"
+    graph = S.caliber_graph(template)
+    graph["nodes"][0]["data"]["config"]["fields"] = [{"name": "status", "required": True}]
+    wf = await S.publish(graph=graph, level="published")
+    two_writers(monkeypatch, "本周销售额 [[v:Q1.r0.gmv]]。")
+    run = await run_manager.start(graph=S.explore_graph("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'pa''id'"),
+                                  input_payload={}, origin_kind="canvas")
+    row = await S.wait(run.id)
+    await ledger_match.drain()
+    whole = (await ledger_match.store.latest(row.id)).content["whole"]
+    assert whole["equivalent"] is True and whole["binding"] == {"status": "pa''id"}
+    res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1})
+    assert res.status_code == 409 and "不能有引号" in res.json()["detail"] and "不代填出具" in res.json()["detail"]