diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 283dde72..a53d6ffe 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -1109,16 +1109,32 @@ def _blocking_issues( **({"defaults": defaults} if isinstance(defaults, dict) else {})}) except Exception: # noqa: BLE001 - 结构不合法留给收尾那一步报 return None + # 带发布级别时 SQL 检查交给门禁（查全图、按级别定挡不挡），自查这边就不再查一遍，免得同一处报两条 + authored = authored_issues(spec, sources or [], baseline, checkers=None if level else checkers) out = ([i.model_dump() for i in validate_graph(spec).issues if i.level == "error"] + _issue_dicts(scope_issues(list(nodes.values()), scope), "datasource_out_of_scope") - + [i for i in authored_issues(spec, sources or [], baseline, checkers=checkers) if i["level"] == "error"]) + + [i for i in authored if i["level"] == "error"]) if level: from app.engine.governance import lint_for_publish - out += [i.model_dump() for i in lint_for_publish(spec, level=level).issues if i.level == "error"] + gate = [i.model_dump() for i in lint_for_publish(spec, level=level, checkers=checkers).issues + if i.level == "error"] + out += _sql_model_hints(gate, spec, checkers) return out +def _sql_model_hints(issues: list[dict[str, Any]], spec: GraphSpec, checkers: dict[str, Any] | None, + ) -> list[dict[str, Any]]: + """门禁报的 SQL 问题只有给人看的那句（ValidationIssue 不带 for_model）；交回模型改时补上给模型的改法。""" + if not checkers or not any(i.get("code") in CHECK_CODES for i in issues): + return issues + from app.data.sqlcheck import graph_checks + + hints = {(node.id, c.code, c.message): c.for_model for node, c in graph_checks(spec.nodes, checkers)} + return [{**i, "for_model": hints[key]} if (key := (i.get("node_id"), i.get("code"), i.get("message"))) in hints + else i for i in issues] + + def _changed(spec: GraphSpec, baseline: list[dict[str, Any]] | None) -> set[str]: """这一轮新加或改过配置的节点。改图时只查这些：用户自己写的、这一轮没碰的 SQL 不替他改。""" before = {n.get("id"): (n.get("data") or {}).get("config") for n in baseline or [] if isinstance(n, dict)} @@ -1887,13 +1903,17 @@ async def assist_publish_fix( """ from app.engine.autofix import forbidden_changes, judge + from app.data.sqlcheck import load_checkers + sources = await _sources(session, None) + # 和发布接口同一套 SQL 检查：受管级别里 SQL 的 error 也是挡住发布的问题，一并交给助手改 + checkers = await load_checkers(session, sources) base = copy.deepcopy(graph) before_nodes = {str(n["id"]): n for n in copy.deepcopy(base.get("nodes") or []) if isinstance(n, dict)} before_edges = [e for e in copy.deepcopy(base.get("edges") or []) if isinstance(e, dict)] baseline = list(copy.deepcopy(before_nodes).values()) errors = _blocking_issues(before_nodes, before_edges, sources=sources, baseline=baseline, level=level, - defaults=base.get("defaults")) or [] + defaults=base.get("defaults"), checkers=checkers) or [] out: dict[str, Any] = {"accepted": False, "reason": None, "summary": "", "questions": [], "graph": graph, "ops": []} if not errors: @@ -1949,7 +1969,7 @@ async def assist_publish_fix( proposed = _proposed_graph(base, nodes, edges) reasons = forbidden_changes(base, proposed, tried) after = _blocking_issues(_nodes_by_id(proposed), proposed.get("edges") or [], sources=sources, - baseline=baseline, level=level, defaults=proposed.get("defaults")) + baseline=baseline, level=level, defaults=proposed.get("defaults"), checkers=checkers) if reasons: out["reason"] = "助手的修改降低了要求，已作废：" + "；".join(reasons) elif after is None: @@ -2217,7 +2237,11 @@ async def upgrade_evidence(payload: UpgradeIn, session: AsyncSession = Depends(g out["notes"] = [n for n in out["notes"] if n["rule"] != "R5" or n["node_id"] in still] elif helped["reason"]: out["rejected"].append({"fix_id": "assist", "reason": helped["reason"]}) - issues = publish_issues(GraphSpec.model_validate(out["graph"]), level=payload.level) + from app.data.sqlcheck import graph_sources, load_checkers_by_name + + upgraded = GraphSpec.model_validate(out["graph"]) + checkers = await load_checkers_by_name(session, graph_sources(upgraded.nodes)) + issues = publish_issues(upgraded, level=payload.level, checkers=checkers) out["issues"] = [i.model_dump() for i in issues] out["ok"] = not any(i.level == "error" for i in issues) return out diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 26924e0d..5038dfc0 100644 --- a/backend/app/api/workflows.py +++ b/backend/app/api/workflows.py @@ -324,7 +324,8 @@ async def publish_workflow( except Exception as e: # noqa: BLE001 return {"ok": False, "issues": [{"level": "error", "message": f"无法解析工作流结构：{graph_error(e)}"}]} - issues = [i.model_dump() for i in publish_issues(spec, level=payload.level)] + checkers = await _sql_checkers(session, spec) + issues = [i.model_dump() for i in publish_issues(spec, level=payload.level, checkers=checkers)] if any(i["level"] == "error" for i in issues): return {"ok": False, "level": payload.level, "version": version, "issues": issues} @@ -400,6 +401,14 @@ async def _gate_context(session: AsyncSession, spec: GraphSpec) -> tuple[dict[st return versions, cards +async def _sql_checkers(session: AsyncSession, spec: GraphSpec) -> dict[str, Any]: + """图里调用工具节点写死的 SQL 用到的数据源，各建一个基于数据目录的 SQL 检查器。发布、发布前检查、 + 自动修复都传这一份给门禁（governance.lint_for_publish）：门禁本身不查库，三处口径才一致。""" + from app.data.sqlcheck import graph_sources, load_checkers_by_name + + return await load_checkers_by_name(session, graph_sources(spec.nodes)) + + def _unreadable(level: str, e: Exception) -> dict[str, Any]: return {"level": level, "ok": False, "fixes": [], "issues": [ {"level": "error", "node_id": None, "edge_id": None, "field": None, "code": None, "fix": None, @@ -422,7 +431,8 @@ async def publish_check( except Exception as e: # noqa: BLE001 return _unreadable(payload.level, e) versions, cards = await _gate_context(session, spec) - return check(spec, level=payload.level, versions=versions, cards=cards) + return check(spec, level=payload.level, versions=versions, cards=cards, + checkers=await _sql_checkers(session, spec)) @router.post("/{workflow_id}/autofix") @@ -446,7 +456,9 @@ async def autofix( {"fix_id": fid, "reason": "无法解析工作流结构，不能应用修复"} for fid in payload.apply], "remaining": bad["issues"], "fixes": [], "handoff": [], "assist": None, "ok": False} versions, cards = await _gate_context(session, spec) - out = apply_fixes(graph, payload.apply, payload.choices, level=payload.level, versions=versions, cards=cards) + checkers = await _sql_checkers(session, spec) + out = apply_fixes(graph, payload.apply, payload.choices, level=payload.level, versions=versions, cards=cards, + checkers=checkers) out["assist"] = None # 人在选项里选了「交给 Copilot」（handoff），和点「交给 Copilot」是一回事 if payload.assist or out["handoff"]: @@ -464,7 +476,8 @@ async def autofix( out["changes"] += diff_changes(out["graph"], fixed, fix_id="assist", label="助手的修改") out["ops"] += helped["ops"] out["applied"].append("assist") - after = apply_fixes(fixed, [], level=payload.level, versions=versions, cards=cards) + after = apply_fixes(fixed, [], level=payload.level, versions=versions, cards=cards, + checkers=await _sql_checkers(session, GraphSpec.model_validate(fixed))) out.update(graph=fixed, remaining=after["remaining"], fixes=after["fixes"], ok=after["ok"]) elif helped["reason"]: out["rejected"].append({"fix_id": "assist", "reason": helped["reason"]}) diff --git a/backend/app/engine/autofix.py b/backend/app/engine/autofix.py index 65e2618f..db2fd86a 100644 --- a/backend/app/engine/autofix.py +++ b/backend/app/engine/autofix.py @@ -559,9 +559,13 @@ def annotate(issues: list[Any], fixes: list[dict[str, Any]]) -> list[dict[str, A def check(spec: GraphSpec, *, level: str, versions: dict[str, Any] | None = None, - cards: dict[str, list[dict[str, Any]]] | None = None) -> dict[str, Any]: - """发布前检查：和真正发布同一套口径的问题，以及能怎么修。只读。""" - issues = publish_issues(spec, level=level) + cards: dict[str, list[dict[str, Any]]] | None = None, + checkers: dict[str, Any] | None = None) -> dict[str, Any]: + """发布前检查：和真正发布同一套口径的问题，以及能怎么修。只读。 + + checkers：基于数据目录的 SQL 检查器（data/sqlcheck），和发布接口传的是同一份（governance.lint_for_publish）。 + """ + issues = publish_issues(spec, level=level, checkers=checkers) fixes = plan_fixes(spec, issues, level=level, versions=versions, cards=cards) return {"level": level, "ok": not any(i.level == "error" for i in issues), "issues": annotate(issues, fixes), "fixes": [public(f) for f in fixes]} @@ -702,7 +706,8 @@ def _op(graph: dict[str, Any], node_id: str | None, edits: list[tuple[str | None def apply_fixes(graph: dict[str, Any] | GraphSpec, fix_ids: list[str], choices: dict[str, Any] | None = None, *, level: str, versions: dict[str, Any] | None = None, - cards: dict[str, list[dict[str, Any]]] | None = None) -> dict[str, Any]: + cards: dict[str, list[dict[str, Any]]] | None = None, + checkers: dict[str, Any] | None = None) -> dict[str, Any]: """在副本上逐个应用修复，每一个都复核：降低要求、错误没变少或者冒出新错误的，丢弃并写明原因。 返回 {graph, changes, ops, applied, rejected, handoff, remaining, fixes, ok}。graph 只是预览，不落库。 @@ -711,7 +716,7 @@ def apply_fixes(graph: dict[str, Any] | GraphSpec, fix_ids: list[str], choices: raw = graph.model_dump(mode="json") if isinstance(graph, GraphSpec) else graph current = copy.deepcopy(raw) spec = GraphSpec.model_validate(current) - issues = publish_issues(spec, level=level) + issues = publish_issues(spec, level=level, checkers=checkers) plan = plan_fixes(spec, issues, level=level, versions=versions, cards=cards) known = {f["id"] for f in plan} rank = {code: i for i, code in enumerate(_ORDER)} @@ -743,7 +748,7 @@ def apply_fixes(graph: dict[str, Any] | GraphSpec, fix_ids: list[str], choices: except Exception: # noqa: BLE001 - 修复规则把图改坏了：丢弃，不交给人一张存不回去的图 rejected.append({"fix_id": fid, "reason": "修复后的工作流结构无法解析，已放弃这项修复"}) continue - trial_issues = publish_issues(trial_spec, level=level) + trial_issues = publish_issues(trial_spec, level=level, checkers=checkers) # choice 里人亲手选的值不算「替人决定」（比如把沙箱代码标成取数）；别的降低要求照样拦 chosen = {(str(nid), path) for nid, path, _, _ in edits} if fix["kind"] == "choice" else set() why = "；".join(forbidden_changes(current, trial, chosen=chosen)) or judge( diff --git a/backend/app/engine/governance.py b/backend/app/engine/governance.py index 9bdd07d3..567b3e45 100644 --- a/backend/app/engine/governance.py +++ b/backend/app/engine/governance.py @@ -2,6 +2,7 @@ from __future__ import annotations import re from collections import defaultdict +from collections.abc import Mapping from typing import Any, Callable from sqlalchemy import func, select @@ -29,7 +30,7 @@ UPGRADE_POLICIES = {"recompute", "dual", "incomparable"} UPGRADE_POLICY_LABELS = {k: option_label("upgrade_policy", k) for k in ("recompute", "dual", "incomparable")} -def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult: +def lint_for_publish(spec: GraphSpec, *, level: str, checkers: Mapping[str, Any] | None = None) -> ValidationResult: """发布门禁。 published：基础可运行性之外只给警告。 @@ -38,6 +39,9 @@ def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult: 每条问题带机器编号（code）：发布前自动修复（engine/autofix.py）按它认问题、出修复， 文案可以改，编号不能随便改。证据门禁 G1–G5 见 _lint_evidence。 + + checkers 是各数据源基于数据目录的 SQL 检查器（data/sqlcheck.load_checkers_by_name），由调用方读好传进来： + 门禁本身不查库。给了的话，调用工具节点里写死的 SQL 对照目录查一遍（_lint_sql）。 """ result = ValidationResult() strict = level == "governed" @@ -100,6 +104,7 @@ def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult: "（那里会保存取数快照）", code="governed.code_network", node_id=node.id, field="network") _lint_evidence(spec, flag, strict=strict) + _lint_sql(spec, flag, checkers) if strict and not has_contract_output: result.add( @@ -109,15 +114,34 @@ def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult: return result -def publish_issues(spec: GraphSpec, *, level: str) -> list[Any]: +def publish_issues(spec: GraphSpec, *, level: str, checkers: Mapping[str, Any] | None = None) -> list[Any]: """发布时拦的全部问题：validate_graph 和 lint_for_publish 的并集，顺序也照发布接口。 - 发布、发布前检查、自动修复复核都走这一个口径——检查说能发的，真发布时不会被拦。 + 发布、发布前检查、自动修复复核都走这一个口径——检查说能发的，真发布时不会被拦。checkers 见 lint_for_publish： + 这几处都要传同一份，否则检查和发布对 SQL 的判断不一致。 """ - gate = lint_for_publish(spec, level=level).issues + gate = lint_for_publish(spec, level=level, checkers=checkers).issues return [*_not_repeated(validate_graph(spec).issues, gate, spec), *gate] +def _lint_sql(spec: GraphSpec, flag: Callable[..., None], checkers: Mapping[str, Any] | None) -> None: + """调用工具节点里写死的 SQL 对照数据目录检查（data/sqlcheck.py）。 + + error 级（依据有确证的重复计算、存量跨期求和）按硬性问题报：受管级别挡住发布，已发布级别只警告——和别的 + 受管要求同一个口径。warning 两档都只提示。info 的依据只是推断（模型起草、命名推断），不进门禁：发布时 + 摆一排「推断，未确认」的提示，只会让人学会无视门禁。编号就是规则编号（fanout_sum…），没有确定性的修复， + 改 SQL 交给人或助手。Agent 节点的 SQL 要到运行时才写出来，不在这里查。 + """ + if not checkers: + return + from app.data.sqlcheck import graph_checks + + for node, check in graph_checks(spec.nodes, checkers): + if check.level == "info": + continue + flag(check.message, code=check.code, node_id=node.id, hard=check.level == "error", field="args.sql") + + #: validate 里和门禁 G4、G5 说同一件事的两条提示（schema.evidence_issues） _REPEATS = frozenset({"evidence.caliber_compute_input", "evidence.cite_fields_off"}) diff --git a/backend/tests/test_publish_sqlcheck.py b/backend/tests/test_publish_sqlcheck.py new file mode 100644 index 00000000..69cb1432 --- /dev/null +++ b/backend/tests/test_publish_sqlcheck.py @@ -0,0 +1,212 @@ +"""发布门禁接上基于数据目录的 SQL 检查（data/sqlcheck.py）。 + +- 受管级别：调用工具节点里写死的 SQL 有 error 级问题（一对多关联后重复计算、存量跨期求和）就挡住发布； +- 已发布级别：同样的问题只是警告，不挡； +- warning 级的问题两档都进提示，info（依据只是推断）不进门禁； +- 发布前检查、自动修复和真正发布同一个口径：检查说能发的，真发布时不会被拦。 +""" +from __future__ import annotations + +import copy +import json +import uuid +from types import SimpleNamespace + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.data import catalog +from app.data.engine import engines +from app.data.introspect import introspect +from app.data.sqlcheck import SqlChecker +from app.engine.governance import lint_for_publish, publish_issues +from app.engine.schema import GraphSpec +from app.main import app +from tests.fixtures.catalog import scenic_notes + +FANOUT_SQL = ("SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id " + "WHERE o.status = 1") +CLEAN_SQL = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1" +WARN_SQL = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id" + +_CACHE: dict[str, dict] = {} + + +async def _schema(path: str) -> dict: + if path not in _CACHE: + probe = SimpleNamespace(id=f"probe-{uuid.uuid4().hex[:6]}", name="scenic", kind="sqlite", database=path, + host=None, port=None, username=None, password=None, options={}, readonly=True, + description="", schema_cache={}, origin="manual") + try: + _CACHE[path] = await introspect(probe) + finally: + await engines.invalidate(probe.id) + return _CACHE[path] + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 10, "y": 20}, "data": {"label": label or nid, "config": config}} + + +def weekly(source: str, sql: str) -> dict: + """input → 取数（调用工具，写死的 SQL）→ 口径卡 → 报告撰写 → 出具，其余受管要求都满足。""" + nodes = [ + node("start", "input", "周期"), + node("fetch", "tool", "取数", tool=f"db_query__{source}", args={"sql": sql}), + node("card", "metrics", "周报口径卡", caliber="周报口径", metrics=[ + {"id": "gmv", "name": "订单金额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"}]), + node("write", "report", "报告撰写", instructions="写周报", + numbers="strict", on_violation="fail", claims="require_citation"), + node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"], "strict": True}), + ] + edges = [{"id": f"e{i}", "source": a["id"], "target": b["id"]} for i, (a, b) in enumerate(zip(nodes, nodes[1:]))] + return {"nodes": nodes, "edges": edges, "defaults": {"approval": "dangerous"}} + + +async def _checkers(scenic_db, status="confirmed"): + cache = await _schema(scenic_db) + return {"scenic": SqlChecker(kind="sqlite", schema_cache=cache, notes=scenic_notes.notes(status), + source_name="scenic")} + + +def _sql_issues(issues): + return [(i.code, i.level, i.node_id, i.field) for i in issues if i.field == "args.sql"] + + +async def test_governed_level_blocks_error_level_sql_problems(scenic_db): + checkers = await _checkers(scenic_db) + spec = GraphSpec.model_validate(weekly("scenic", FANOUT_SQL)) + assert not [i for i in publish_issues(spec, level="governed") if i.level == "error"], "其余受管要求都满足" + issues = publish_issues(spec, level="governed", checkers=checkers) + assert _sql_issues(issues) == [("fanout_sum", "error", "fetch", "args.sql")] + [issue] = [i for i in issues if i.code == "fanout_sum"] + assert issue.message.startswith("「订单」关联「订单明细」是一对多") + # 已发布级别：同一个问题只警告 + assert _sql_issues(publish_issues(spec, level="published", checkers=checkers)) == [ + ("fanout_sum", "warning", "fetch", "args.sql")] + # 依据只是推断：info 不进门禁 + assert _sql_issues(lint_for_publish(spec, level="governed", checkers=await _checkers(scenic_db, "proposed")) + .issues) == [] + # 没有检查器（数据源没有目录）：和以前一样 + assert _sql_issues(lint_for_publish(spec, level="governed").issues) == [] + + +async def test_warnings_are_hints_on_both_levels(scenic_db): + checkers = await _checkers(scenic_db) + spec = GraphSpec.model_validate(weekly("scenic", WARN_SQL)) + for level in ("governed", "published"): + assert _sql_issues(publish_issues(spec, level=level, checkers=checkers)) == [ + ("missing_valid_filter", "warning", "fetch", "args.sql")] + clean = GraphSpec.model_validate(weekly("scenic", CLEAN_SQL)) + assert _sql_issues(publish_issues(clean, level="governed", checkers=checkers)) == [] + + +async def test_autofix_check_and_apply_use_the_same_gate(scenic_db): + from app.engine.autofix import apply_fixes, check + + checkers = await _checkers(scenic_db) + graph = weekly("scenic", FANOUT_SQL) + body = check(GraphSpec.model_validate(graph), level="governed", checkers=checkers) + assert body["ok"] is False and "fanout_sum" in {i["code"] for i in body["issues"]} + out = apply_fixes(copy.deepcopy(graph), [], level="governed", checkers=checkers) + assert out["ok"] is False and "fanout_sum" in {i["code"] for i in out["remaining"]} + + +# -------------------------------------------------------------------------- +# 接口：发布、发布前检查、自动修复 +# -------------------------------------------------------------------------- + + +@pytest.fixture +async def scenic_source(scenic_db): + from app.db.base import SessionLocal + from app.db.models import DataSource + + name = f"scenic_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=scenic_db, readonly=True) + row.schema_cache = await _schema(scenic_db) + session.add(row) + await session.commit() + for table, notes in scenic_notes.notes().items(): + await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏") + source_id = row.id + yield name + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + row.enabled = False + await session.commit() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def test_publish_endpoints_block_governed_fanout(client, scenic_source): + r = await client.post("/api/workflows", json={"name": f"订单周报-{scenic_source}", + "graph": weekly(scenic_source, FANOUT_SQL)}) + assert r.status_code == 201, r.text + wid = r.json()["id"] + + checked = (await client.post(f"/api/workflows/{wid}/publish-check", json={"level": "governed"})).json() + assert checked["ok"] is False + [issue] = [i for i in checked["issues"] if i["code"] == "fanout_sum"] + assert issue["level"] == "error" and issue["node_id"] == "fetch" and issue["field"] == "args.sql" + assert issue["fix"] is None, "SQL 写错没有确定性的修复，交给人或助手改" + + fixed = (await client.post(f"/api/workflows/{wid}/autofix", json={"level": "governed", "apply": []})).json() + assert fixed["ok"] is False and "fanout_sum" in {i["code"] for i in fixed["remaining"]} + + blocked = (await client.post(f"/api/workflows/{wid}/publish", json={"level": "governed"})).json() + assert blocked["ok"] is False and "fanout_sum" in {i.get("code") for i in blocked["issues"]} + # 已发布级别不挡，问题作为警告带回 + published = (await client.post(f"/api/workflows/{wid}/publish", json={"level": "published"})).json() + assert published["ok"] is True + assert [i["level"] for i in published["issues"] if i.get("code") == "fanout_sum"] == ["warning"] + + # 改好 SQL：受管级别发得出去，剩下的 warning 进提示 + await client.patch(f"/api/workflows/{wid}", json={"graph": weekly(scenic_source, WARN_SQL)}) + ok = (await client.post(f"/api/workflows/{wid}/publish", json={"level": "governed"})).json() + assert ok["ok"] is True, ok["issues"] + assert [i["code"] for i in ok["issues"] if i.get("field") == "args.sql"] == ["missing_valid_filter"] + + +class _Scripted: + """按剧本吐操作流的模型；记下每次收到的请求。""" + + def __init__(self, ops): + self.ops, self.calls = ops, [] + + async def astream(self, messages): + from langchain_core.messages import AIMessageChunk + + self.calls.append(messages) + for op in self.ops: + yield AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n") + + +async def test_publish_fix_assistant_gets_the_sql_error_and_its_fix_is_adopted(client, scenic_source, monkeypatch): + """确定性修复修不了 SQL：交给助手。助手看到的是给模型的那句改法，改好的 SQL 复核通过才采纳。""" + import app.api.copilot as copilot + + model = _Scripted([{"op": "plan", "summary": "改 SQL"}, + {"op": "update_node", "id": "fetch", + "config": {"tool": f"db_query__{scenic_source}", "args": {"sql": CLEAN_SQL}}}, + {"op": "done", "explanation": "改为对明细金额求和"}]) + + async def _model(*_a, **_k): + return model, "mock-fast" + + monkeypatch.setattr(copilot, "get_chat_model", _model) + wid = (await client.post("/api/workflows", json={"name": f"订单周报-助手-{scenic_source}", + "graph": weekly(scenic_source, FANOUT_SQL)})).json()["id"] + out = (await client.post(f"/api/workflows/{wid}/autofix", + json={"level": "governed", "apply": [], "assist": True})).json() + human = next(text for role, text in model.calls[0] if role == "human") + assert "orders 关联 order_items 是一对多" in human + assert out["assist"]["ok"] is True and out["applied"] == ["assist"] and out["ok"] is True, out + sql = next(n for n in out["graph"]["nodes"] if n["id"] == "fetch")["data"]["config"]["args"]["sql"] + assert sql == CLEAN_SQL