diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 993e6313..2f8766a5 100644 --- a/backend/app/api/copilot.py +++ b/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 diff --git a/backend/app/blueprints/publish_plan.py b/backend/app/blueprints/publish_plan.py index 79e00d61..71e470ab 100644 --- a/backend/app/blueprints/publish_plan.py +++ b/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")) diff --git a/backend/app/blueprints/templates/_common.py b/backend/app/blueprints/templates/_common.py index 822ebff4..b8820fcb 100644 --- a/backend/app/blueprints/templates/_common.py +++ b/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")) diff --git a/backend/app/plain.py b/backend/app/plain.py index 959f61ab..4c724cab 100644 --- a/backend/app/plain.py +++ b/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( diff --git a/backend/tests/test_copilot_sqlcheck.py b/backend/tests/test_copilot_sqlcheck.py index 2a6cb072..2263ccc3 100644 --- a/backend/tests/test_copilot_sqlcheck.py +++ b/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])) diff --git a/backend/tests/test_publish_plan.py b/backend/tests/test_publish_plan.py index bc838b1f..4ae1ef8a 100644 --- a/backend/tests/test_publish_plan.py +++ b/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},