取数 SQL 不再把数据源名写成库名，试运行失败的蓝图不能发布：填关节的说明以前把表写成「数据源名.表名」，模型照抄进 SQL，数据库把数据源名当成库名（MySQL 报 Unknown database），应用每次出数都失败；说明改成只写数据源里的表名，并写明不要加数据源名。助手自查加 sql_source_prefix：表名前面写了数据源名就交回模型改（认列的检查找表时去掉前缀，拦不住它；数据源名恰好是库名、schema 的不报）。发布单遇到最近一次试运行失败改为挡住（plan.trial_failed，带出错的那一句），不再只是「还没有成功的试运行」的提醒

backend/app/api/copilot.py

@@ -1248,6 +1248,8 @@ def authored_issues(spec: GraphSpec, sources: list[Any], baseline: list[dict[str
 
     - sql_unknown_column：调用工具节点的 SQL 里有数据源结构（schema_cache）里查不到的列——凭空
       猜的列名跑起来只会「no such column」，还白跑前面的步骤
+    - sql_source_prefix：表名前面写了数据源名（sql_source_prefix）——数据源名被数据库当成库名，跑起来报库不存在；
+      认列的检查按表名找表时会去掉前缀，拦不住它
     - parse_model_text：整形节点按 JSON 解析 agent / llm 写的文字（validate 里只是 warning：旧图照跑）
     - report_no_source：报告撰写节点的上游没有任何证据来源（运行时只给一条警告，报告里的数全没有出处）。
       metrics_from 指向不是口径卡的节点，validate 本来就报 error，自查照样交回去
@@ -1278,6 +1280,16 @@ def authored_issues(spec: GraphSpec, sources: list[Any], baseline: list[dict[str
         if source is None or not str(node.config.get("tool") or "").startswith("db_query__") \
                 or not isinstance(sql, str):
             continue
+        if prefixed := sql_source_prefix(sql, source):
+            name, bare = source.name, prefixed[0].split(".", 1)[1]
+            out.append({"level": "error", "node_id": node.id, "edge_id": None, "field": "args.sql",
+                        "code": "sql_source_prefix",
+                        "message": f"SQL 里的表 {'、'.join(prefixed[:5])} 前面写了数据源名「{name}」：数据库会把它当成库名，"
+                                   f"运行时报库不存在。去掉「{name}.」，表名照表结构写（如 {bare}）",
+                        "for_model": f"SQL 里的表 {'、'.join(prefixed[:5])} 前面写了数据源名「{name}」。工具 db_query__{name} "
+                                     f"已经连到这个数据源，数据库里没有叫「{name}」的库或 schema，这样写跑起来会报库不存在。"
+                                     f"去掉「{name}.」，表名照表结构里的写（如 {bare}）"})
+            continue
         found = sql_unknown_columns(sql, source)
         if not found:
             out += _catalog_issues(node.id, sql, (checkers or {}).get(source.name))
@@ -1407,6 +1419,29 @@ def _blank_unicode_identifiers(sql: str) -> str:
     return "".join(out)
 
 
+def sql_source_prefix(sql: str, source: Any) -> list[str]:
+    """SQL 里 FROM / JOIN 后面写成「数据源名.表名」的表（去重、按出现顺序）。
+
+    数据源名是在 AgentLab 里给连接起的名字，不是库名、schema：取数工具 db_query__<数据源名> 已经连到这个数据源，再写
+    前缀，数据库会把它当成库名（MySQL 的 Unknown database、SQLite 的 unknown database、PostgreSQL 的 schema 不存在）。
+    认列的检查（introspect.find_table）找表时会去掉前缀，拦不住它。数据源名恰好也是库名、schema 的不算：连接里填的库、
+    探查的 schema、候选 schema、表结构里本来就带这个前缀的表。"""
+    from app.data.guard import _blank_quoted, strip_comments
+
+    name = str(getattr(source, "name", "") or "")
+    if not name:
+        return []
+    cache = getattr(source, "schema_cache", None) or {}
+    real = {str(x).lower() for x in (getattr(source, "database", None), cache.get("schema"),
+                                     *(cache.get("available_schemas") or [])) if x}
+    if name.lower() in real or any(str(t).lower().startswith(f"{name.lower()}.") for t in cache.get("tables") or {}):
+        return []
+    # 抹掉注释、模板写法和字符串，带引号的标识符留着（`shop`.`orders` 一样要认）
+    text = _blank_quoted(re.sub(r"\{\{.*?\}\}", " 0 ", strip_comments(sql), flags=re.S), identifiers=False)
+    pattern = re.compile(rf"\b(?:from|join)\s+[`\"\[]?{re.escape(name)}[`\"\]]?\s*\.\s*[`\"\[]?([^\s`\"\[\].,()]+)", re.I)
+    return list(dict.fromkeys(f"{name}.{m.group(1)}" for m in pattern.finditer(text)))
+
+
 def sql_unknown_columns(sql: str, source: Any) -> tuple[list[str], list[str]] | None:
     """SQL 里用到、数据源结构里查不到的列：(查不到的列, 这几张表的全部列)。判断不了、或者都查得到时返回 None。"""
     from app.data import introspect as _introspect

backend/app/blueprints/publish_plan.py

@@ -69,6 +69,7 @@ GRANT_ROLES = ("use", "observe", "manage")
 NOT_INSTANTIATED = "plan.not_instantiated"
 GRAPH_CHANGED = "plan.graph_changed"
 UNTRIALED = "plan.untrialed"
+TRIAL_FAILED = "plan.trial_failed"
 TRIAL_DIFFERS = "plan.trial_differs"
 ADMIN_REQUIRED = "plan.admin_required"
 DOWNGRADE_ADMIN = "plan.downgrade_admin"
@@ -347,9 +348,13 @@ async def _preflight(session: AsyncSession, bp: Blueprint, target: _Target, *, l
     _, problem = await grants_problem(session, grant_items(bp))
     if problem:
         out.append(_problem(GRANT_REFUSED, problem, source="grants"))
-    # 试运行：没用真数试过、试过之后又改了，照实说（不拦）
+    # 试运行：最近一次失败了的挡住——用真数跑过、出了错，照样发布出去每次出数都是同一个错（取数 SQL 把数据源名写成了
+    # 库名这一类，2026-10 撞过）；没用真数试过、试过之后又改了，照实说（不拦）
     trial = await session.get(Run, bp.last_trial_run_id) if bp.last_trial_run_id else None
-    if not is_trial_of(trial, bp) or trial.status != "succeeded":
+    if is_trial_of(trial, bp) and trial.status == "failed":
+        why = (trial.error or "").strip().splitlines()[0][:200] if (trial.error or "").strip() else ""
+        out.append(_problem(TRIAL_FAILED, f"最近一次试运行失败了{f'：{why}' if why else ''}", source="plan"))
+    elif not is_trial_of(trial, bp) or trial.status != "succeeded":
         out.append(_problem(UNTRIALED, "还没有成功的试运行：发布之前没用真数看过结果", "warning", source="plan"))
     elif graph_hash(trial.graph or {}) != target.digest:
         out.append(_problem(TRIAL_DIFFERS, "发布的内容和最近一次试运行不同：试运行之后又改过", "warning", source="plan"))

backend/app/blueprints/templates/_common.py

@@ -36,7 +36,8 @@ START = "start"
 # ==========================================================================
 
 SQL_BRIEF_HEAD_PROMPT = ("取数节点 {node}（工具 db_query__{source}）：写一条只读 SQL（只能 SELECT，一条语句），结果{shape}。"
-                         "{period}")
+                         "这个工具已经连到数据源「{source}」：表名照下面写的（就是这个数据源里的表名），前面不要加「{source}.」，"
+                         "数据库里没有叫这个名字的库。{period}")
 SQL_PERIOD_PROMPT = ("统计期用入口字段：从 '{{ input.period_start }}' 到 '{{ input.period_end }}'（日期，含两端，"
                      "写在 SQL 里要带单引号）。")
 SQL_ONE_ROW_PROMPT = "只出一行，每个要算的数一列（或者算它要用的基础量各一列），列名用英文小写和下划线"
@@ -598,7 +599,9 @@ def item_brief(item: dict[str, Any]) -> str:
 
     return SQL_ITEM_PROMPT.format(
         label=value(item.get("label")) or item["id"], id=item["id"], unit=f"，单位 {unit}" if unit else "",
-        measure=text("measure"), table=f"{table.get('source')}.{table.get('table')}" if table else UNKNOWN_PROMPT,
+        # 只写数据源里的表名（多 schema 的库，目录里的名字本身带 schema 前缀）：以前写成「数据源名.表名」，模型照抄进 SQL，
+        # 数据源名被数据库当成库名，跑起来报库不存在
+        measure=text("measure"), table=str(table.get("table") or UNKNOWN_PROMPT) if table else UNKNOWN_PROMPT,
         date=(f"{date.get('column')}" + (f"（{date['rule']}）" if date.get("rule") else "")) if date else UNKNOWN_PROMPT,
         filter=text("filter"), grain=text("grain"))
 

backend/app/plain.py

@@ -178,6 +178,9 @@ PLAIN: dict[str, Plain] = {
         "蓝图还没生成工作流", "没有图就没有东西可以发布。", _Y, "先把卡片补齐，用真数试运行一次。"),
     "plan.graph_changed": Plain(
         "出单之后蓝图又改过", "要发布的已经不是这份发布单看过的那张图。", _Y, "重新出一份发布单。"),
+    "plan.trial_failed": Plain(
+        "最近一次试运行失败了", "用真数运行出了错，这样发布出去，每次出数都会是同一个错。", _Y,
+        "回到试运行看出错的原因（可以交给助手修），修好后再用真数试一次。"),
     "plan.untrialed": Plain(
         "还没用真数试运行过", "没看过真数就发布，可能不是你要的结果。", _Y, "先试运行一次，看过结果再发布。"),
     "plan.trial_differs": Plain(

backend/tests/test_copilot_sqlcheck.py

@@ -292,3 +292,20 @@ async def test_upgrade_assist_checks_sql_like_the_publish_gate(scenic_source, mo
     # 改不好：整个作废，原因写的是这条 SQL 检查
     assert out["accepted"] is False and out["ok"] is False
     assert "一对多" in (out["reason"] or ""), out["reason"]
+
+
+def test_a_source_name_written_as_a_database_is_sent_back():
+    """表名前面写了数据源名（scenic.orders）：数据库把它当成库名，跑起来报库不存在；认列的检查找表时会去掉前缀，拦不住它，
+    这里单独交回模型改。数据源名恰好也是库名的不报。"""
+    from app.api.copilot import authored_issues
+
+    cache = {"schema": "main", "tables": {"orders": {"columns": [{"name": "id"}, {"name": "total_amount"}]}}}
+    source = SimpleNamespace(name="scenic", database="/data/scenic.db", schema_cache=cache)
+    sql = "SELECT SUM(o.total_amount) AS gmv FROM `scenic`.orders o WHERE o.id > '{{ input.since }}' -- from scenic.x"
+    spec = _spec(_node("start", "input"), _fetch(sql), _node("out", "output", fields=[]))
+    (issue,) = authored_issues(spec, [source])
+    assert (issue["code"], issue["level"], issue["node_id"]) == ("sql_source_prefix", "error", "fetch")
+    assert "scenic.orders" in issue["message"] and "去掉「scenic.」" in issue["for_model"]
+    # 数据源名就是库名：照写，不报
+    same = SimpleNamespace(name="scenic", database="scenic", schema_cache=cache)
+    assert all(i["code"] != "sql_source_prefix" for i in authored_issues(spec, [same]))

backend/tests/test_publish_plan.py

@@ -167,6 +167,25 @@ async def test_preflight_blocks_and_nothing_is_written(client, tmp_path):
     assert (await s.blueprint(bp["id"])).workflow_id is None
 
 
+async def test_a_failed_trial_blocks_the_plan(client, tmp_path):
+    """最近一次试运行失败了：挡住发布（用真数跑出了错，发布出去每次出数都是同一个错），写明出错的那一句；不再只是
+    「还没有成功的试运行」的提醒。"""
+    from sqlalchemy import update
+
+    bp = await _board_from_run(client, tmp_path)
+    async with SessionLocal() as session:
+        await session.execute(update(Run).where(Run.id == bp["last_trial_run_id"]).values(
+            status="failed", error="工具 db_query__shop 查询失败：(1049, \"Unknown database 'shop'\")\n第二行不要"))
+        await session.commit()
+    plan = (await client.post(f"/api/blueprints/{bp['id']}/publish-plan", json={"level": "published"})).json()
+    assert plan["ok"] is False
+    codes = {p["code"]: p for p in plan["problems"]}
+    failed = codes["plan.trial_failed"]
+    assert failed["level"] == "error" and failed["who"] == "you" and failed["plain"]["title"] == "最近一次试运行失败了"
+    assert failed["message"] == "最近一次试运行失败了：工具 db_query__shop 查询失败：(1049, \"Unknown database 'shop'\")"
+    assert "plan.untrialed" not in codes
+
+
 async def test_a_process_with_an_unbound_step_stops_at_preflight(client):
     wf_id = await s.published_workflow(s.expense_graph(), f"报销-{uuid.uuid4().hex[:4]}", description="报销审批")
     bp = (await client.post("/api/blueprints", json={"source": {"kind": "workflow_version", "id": wf_id},