From 7f07fc370a6ea333bc1be36a9b0d69fa58e8f8ad Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 3 Oct 2026 10:32:54 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=202B=EF=BC=9A=E8=BF=90=E8=A1=8C=E6=97=B6=E6=8E=A5?= =?UTF-8?q?=E4=B8=8A=20SQL=20=E6=A3=80=E6=9F=A5=E2=80=94=E2=80=94=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E7=BB=93=E6=9E=9C=E5=92=8C=E5=BF=AB=E7=85=A7=E5=B8=A6?= =?UTF-8?q?=20checks=E3=80=81=E8=AF=81=E6=8D=AE=E6=8E=A5=E5=8F=A3=E7=9A=84?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E6=AD=A5=E9=AA=A4=E5=B8=A6=E4=B8=8A=EF=BC=8C?= =?UTF-8?q?=E5=A5=91=E7=BA=A6=E6=8C=87=E6=A0=87=E7=9A=84=E6=9D=A5=E6=BA=90?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E6=9C=89=20error=20=E6=97=B6=E5=87=BA?= =?UTF-8?q?=E5=85=B7=E9=99=8D=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/evidence.py | 21 +- backend/app/engine/evidence.py | 6 + backend/app/engine/issuance.py | 11 + backend/app/engine/nodes/io.py | 4 +- backend/app/engine/nodes/metrics.py | 85 +++++++- backend/app/tools/datasource.py | 33 +++ backend/tests/test_runtime_sqlcheck.py | 288 +++++++++++++++++++++++++ 7 files changed, 445 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_runtime_sqlcheck.py diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 7b3ba4c7..bbb9b38b 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -825,6 +825,9 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s # 拿截断的查询结果整组算出来的：面板标「结果不完整」并写明原因。完整的指标没有这两个键 **({"incomplete": True, "incomplete_reason": source.get("incomplete_reason") or ""} if source.get("incomplete") else {}), + # 来源查询没通过 SQL 检查：面板标出来并写明原因（问题和改法在查询步骤的 checks 里）。没问题的没有这两个键 + **({"sql_check_failed": True, "sql_check_reason": source.get("sql_check_reason") or ""} + if source.get("sql_check_failed") else {}), "node_id": entry.get("node_id"), "artifact": artifact or None, "eid": cite.get("eid"), "eid_ok": cite.get("eid") == entry.get("eid") == make_eid("metric", artifact or None, {"metric": metric_id}), "hash_ok": hash_ok and metric is not None, @@ -904,6 +907,8 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str hint（期 4，P4-SPEC 2.8.2）：调用方是带文档标记的单元格片段时为真。这时快照若是上传表格的（有 data_version）， 步骤多一个 `provenance: true`，界面据此才去请求推断来源接口；其余情况一个键都不加，形状和期 4 之前完全相同 （期 4 之前的文档、手工源、指标链都不多发请求）。不看封存状态：未封存的运行照常推断，响应里照实标明。 + + checks：快照里记下的 SQL 检查结果（数据目录阶段 2B，见 _query_checks），只在查出问题时有这个键。 """ catalog = doc.get("catalog") or {} alias = _query_alias(catalog, artifact) @@ -948,7 +953,8 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str masked=[columns[i] for i in sorted(masked)], # 查询时从驱动的原始值记下的列类型（老快照没有）：文本列里的 "2026" 按文本显示，不当数 column_types=snap.get("column_types") if isinstance(snap.get("column_types"), dict) else {}, - **({"provenance": True} if hint and snap.get("data_version") else {})) + **({"provenance": True} if hint and snap.get("data_version") else {}), + **({"checks": checks} if (checks := _query_checks(snap)) else {})) if len(index) > MAX_WINDOW_ROWS: index, step["window_truncated"] = index[:MAX_WINDOW_ROWS], True step["row_index"] = index @@ -958,6 +964,19 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str return step +#: 查询步骤里每条 SQL 检查结果交给界面的字段（data/sqlcheck.SqlCheck.as_dict 去掉 for_model） +_CHECK_FIELDS = ("code", "level", "message", "table", "column", "relation_id", "sql_excerpt") + + +def _query_checks(snap: dict[str, Any]) -> list[dict[str, Any]]: + """查询快照里记下的 SQL 检查结果（数据源查询工具执行后对照数据目录查的）。给模型的那句改法不上界面。 + 快照里没有（没查出问题、或者是升级前的快照）返回空列表，步骤里就不加这个键。""" + checks = snap.get("checks") + if not isinstance(checks, list): + return [] + return [{k: c[k] for k in _CHECK_FIELDS if k in c} for c in checks if isinstance(c, dict) and c.get("code")] + + # -------------------------------------------------------------------------- # GET /api/runs/{run_id}/evidence/segments/{segment_id}/provenance（期 4：推断的来源） # -------------------------------------------------------------------------- diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index 198bb816..44abc1bd 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -902,6 +902,9 @@ def build_catalog( # 只在不完整时有这两个键，完整的指标条目和以前一字不差 **({"incomplete": True, "incomplete_reason": metric.get("incomplete_reason") or ""} if metric.get("incomplete") else {}), + # 来源查询没通过 SQL 检查（data/sqlcheck.py 的 error 级问题）：同样只在有问题时有这两个键 + **({"sql_check_failed": True, "sql_check_reason": metric.get("sql_check_reason") or ""} + if metric.get("sql_check_failed") else {}), } queries = retrievals = 0 @@ -3109,6 +3112,9 @@ def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowe # 不完整的指标照样能引用（单格、口径都对），但写作者得知道它不是全量：不写行数，免得照抄成裸数字 caveat = "（基于被截断的查询结果计算，结果不完整：引用时须如实说明，不要写成全部数据）" \ if e.get("incomplete") else "" + # 取数的 SQL 没通过检查（重复计算、存量跨期加总）：数是算错的可能很大，写作者引用时得交代 + if e.get("sql_check_failed"): + caveat += "（所依据的查询未通过 SQL 检查，结果存疑：引用时须如实说明，不要当作确定的结论）" lines.append(f"- [[{e['alias']}]] {e.get('name')} = {e.get('rendered')}{caveat}") inputs = [e for e in catalog.values() if e.get("kind") == "input"] if inputs: diff --git a/backend/app/engine/issuance.py b/backend/app/engine/issuance.py index c7aad257..5b030f3a 100644 --- a/backend/app/engine/issuance.py +++ b/backend/app/engine/issuance.py @@ -462,6 +462,17 @@ def incomplete_gaps(metrics: list[dict[str, Any]]) -> list[str]: for m in metrics if isinstance(m, dict) and m.get("incomplete")] +def sql_check_gaps(metrics: list[dict[str, Any]]) -> list[str]: + """契约收来的指标里来源查询没通过 SQL 检查的（口径卡标了 sql_check_failed：取数的那条 SQL 对照数据目录查出 + error 级问题，比如一对多关联之后对「一」那一侧求和），每个一句缺口。 + + 和 incomplete_gaps 同一个道理：数字照样回指得上口径卡，口径卡照样复算一致，但算它的那条查询本身就把数算错了 + （重复计算、存量跨期加总），盖完整出具的章等于替一个错数作保。原因用口径卡记下的那句，写明是哪条检查。 + """ + return [f"指标「{m.get('name') or m.get('id')}」{m.get('sql_check_reason') or '所依据的查询未通过 SQL 检查'}" + for m in metrics if isinstance(m, dict) and m.get("sql_check_failed")] + + def _claims_effect( policy: str | dict[str, Any] | None, unsupported: list[Any] | int | None, diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index 6ed8c676..4d79ba03 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -147,7 +147,7 @@ def _apply_contract( ) -> dict[str, Any]: from datetime import datetime, timezone - from app.engine.issuance import decide_tier, incomplete_gaps, trace_numbers + from app.engine.issuance import decide_tier, incomplete_gaps, sql_check_gaps, trace_numbers nodes = state.get("nodes") or {} # 写了 report_from 就是引用模式：按报告文档逐段复核，不再按数值回指叙述 @@ -188,6 +188,8 @@ def _apply_contract( gaps.append(f"指标来源未解析：{'、'.join(unresolved[:5])}") # 拿截断的查询结果整组算出来的指标：数回指得上，但算的只是取回的那一截，不能完整出具 gaps.extend(incomplete_gaps(metrics)) + # 来源查询没通过 SQL 检查（error 级：重复计算、存量跨期加总）的指标：口径卡复算得一致，算的却是错数 + gaps.extend(sql_check_gaps(metrics)) # 引用模式下，缺输入记成空值的指标（口径卡 on_missing=null）就是缺：报告里引用不了它， # 也不能因为 id 在清单里就算「齐了」。旧模式照旧只看 id，行为不变 diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index d3826b12..c6b73c9c 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -49,6 +49,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 表达式把截断的查询结果当成整组用了（len、sum、max……，见 expressions.TruncatedUse）的指标， 值照算，但标 incomplete 并写明原因：它算的只是取回的那一截。出具契约据此降档（issuance.incomplete_gaps）。 + + 取数的那条 SQL 对照数据目录查出 error 级问题的（查询快照的 checks，见 data/sqlcheck.py：一对多关联后重复计算、 + 存量跨期加总），同样值照算，标 sql_check_failed 并写明原因，出具契约据此降档（issuance.sql_check_gaps）。 """ definitions = ctx.cfg("metrics", []) or [] caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1") @@ -71,6 +74,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: metrics: list[dict[str, Any]] = [] errors: list[str] = [] + #: 查询快照工件 id → 它的 error 级检查结果。一张卡里多个指标常常取自同一次查询，快照只读一次 + checked: dict[str, list[dict[str, Any]]] = {} for definition in definitions: metric_id = str(definition.get("id") or "").strip() expr = str(definition.get("expression") or "").strip() @@ -127,6 +132,12 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: sources = list(use.sources.values()) incomplete = {"incomplete": True, "incomplete_reason": _incomplete_reason(sources), "truncated_sources": sources} + inputs = [_input_of(path, values[path], state, ctx, tctx) for path in paths] + sql_check: dict[str, Any] = {} + if failed := _failed_queries(inputs, list(use.sources.values()), paths, tctx, checked): + sql_check = {"sql_check_failed": True, "sql_check_reason": _sql_check_reason(failed), + "sql_check_sources": [{"artifact": artifact, "codes": list(dict.fromkeys( + str(c.get("code")) for c in found))} for artifact, found in failed]} metrics.append( { "id": metric_id, @@ -137,7 +148,7 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "decimals": decimals, "format": display["format"], "rendered": _rendered(value, definition, display), - "inputs": [_input_of(path, values[path], state, ctx, tctx) for path in paths], + "inputs": inputs, "substituted": substituted, "recompute_ok": _recompute(substituted, tctx, value, decimals), # 值是空的就是缺输入，不论 on_missing 是哪一种：fail 模式下直接取到空值的 @@ -145,6 +156,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "status": "ok" if value is not None else "missing_input", # 只在不完整时出现：完整的指标和以前一字不差，老运行复核时目录也对得上 **incomplete, + # 只在来源查询没通过 SQL 检查时出现，同上 + **sql_check, } ) @@ -158,12 +171,16 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: else f"- {m['id']}（{m['name']}）= {MISSING}（缺少输入，无值）") # 不写行数：叙述节点照抄进正文就是一个没有出处的数 + ("（不完整：基于被截断的查询结果计算）" if m.get("incomplete") else "") + + ("（存疑：所依据的查询未通过 SQL 检查）" if m.get("sql_check_failed") else "") for m in metrics ] for m in metrics: if m.get("incomplete"): ctx.emit(EventType.LOG, level="warn", code="metric_incomplete", metric=m["id"], message=f"指标「{m['name']}」{m['incomplete_reason']}") + if m.get("sql_check_failed"): + ctx.emit(EventType.LOG, level="warn", code="metric_sql_check", metric=m["id"], + message=f"指标「{m['name']}」{m['sql_check_reason']}") card = { "kind": "metric_set", "caliber": caliber, @@ -331,6 +348,72 @@ def _incomplete_reason(sources: list[dict[str, Any]]) -> str: return f"基于 {len(fetched)} 份被截断的查询结果计算（分别只取回了{'、'.join(fetched)}），结果不完整" +#: 取值链的最后一段：.字段 或 [下标] +_LAST_STEP = re.compile(r"(?:\.[^.\[\]]+|\[[^\[\]]*\])$") + + +def _failed_queries(inputs: list[dict[str, Any]], truncated: list[dict[str, Any]], paths: list[str], + tctx: dict[str, Any], checked: dict[str, list[dict[str, Any]]], + ) -> list[tuple[str, list[dict[str, Any]]]]: + """这个指标用到的查询里，SQL 检查有 error 的：[(查询快照工件 id, error 级检查结果)]。 + + 用到了哪些查询，三处来历合起来认：输入的出处（cell() 取的格、Agent 经 cite_fields 交来的字段都带着快照 id）、 + 截断跟踪记下的整组、取值链上的查询结果本身（数据整形解析过的 vars.res.rows[0][0]，往上找到 vars.res 那份 + 带 artifact 的结果）。检查结果从快照里读，不信内存里那份：快照按哈希寻址，就是查询当时存下的那份。 + 读不出来的快照不算（另有别的检查报快照的问题）。Agent 没开 cite_fields 时说不清数出自哪次查询，不算。 + """ + artifacts = [item.get("artifact") for item in inputs] + [s.get("artifact") for s in truncated] + artifacts += [_result_artifact(path, tctx) for path in paths if cell_parts(path) is None] + out = [] + for artifact in dict.fromkeys(a for a in artifacts if isinstance(a, str) and a): + if artifact not in checked: + checked[artifact] = _error_checks(artifact) + if checked[artifact]: + out.append((artifact, checked[artifact])) + return out + + +def _result_artifact(path: str, tctx: dict[str, Any]) -> str | None: + """顺着取值链往上找（vars.res.rows[0][0] → vars.res.rows[0] → vars.res.rows → vars.res），第一个查询结果的快照 id。 + 数据源工具交回的是 JSON 文本（调用工具节点的输出），解析过的是对象（数据整形之后），两种都认。""" + current = path + for _ in range(8): + value = _value_of(current, tctx) + if isinstance(value, str) and value.lstrip().startswith("{") and '"artifact"' in value: + try: + value = json.loads(value) + except ValueError: + value = None + if isinstance(value, dict) and isinstance(value.get("artifact"), str) and isinstance(value.get("rows"), list): + return value["artifact"] + shorter = _LAST_STEP.sub("", current) + if not shorter or shorter == current: + return None + current = shorter + return None + + +def _error_checks(artifact: str) -> list[dict[str, Any]]: + try: + snapshot = artifact_store.load(artifact) + except Exception: # noqa: BLE001 - 快照读不出来、被改过：这里不报，证据层另有说法 + return [] + checks = snapshot.get("checks") if isinstance(snapshot, dict) else None + return [c for c in checks if isinstance(c, dict) and c.get("level") == "error"] if isinstance(checks, list) else [] + + +def _sql_check_reason(failed: list[tuple[str, list[dict[str, Any]]]]) -> str: + """「所依据的查询未通过 SQL 检查（「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算）， + 结果不可靠」：出具声明、运行日志、证据面板都用这句。只摘问题那半句，改法留在查询步骤里看。""" + problems = list(dict.fromkeys(str(c.get("message") or "").split("。")[0] for _, found in failed for c in found)) + problems = [p for p in problems if p] + if not problems: + return "所依据的查询未通过 SQL 检查，结果不可靠" + if len(problems) == 1: + return f"所依据的查询未通过 SQL 检查（{problems[0]}），结果不可靠" + return f"所依据的查询有 {len(problems)} 处未通过 SQL 检查（{problems[0]}等），结果不可靠" + + def _truncated_field(path: str, state: GraphState, ctx: NodeContext) -> dict[str, Any] | None: """agent 经 cite_fields 交来、被截断切开的数组字段：{rows, artifact}。别的取值链返回 None。 diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index b7c5da17..575fc847 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -33,6 +33,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.events import EventType from app.data import catalog as data_catalog from app.data import introspect, table_versions +from app.data.sqlcheck import SqlChecker from app.data.engine import SnapshotTampered, masked_columns, query_timeout, run_query from app.data.guard import QueryLimits, SqlRejected, is_write from app.data.tabular import UNSHAPED_NOTE @@ -53,6 +54,10 @@ DATA_UNAVAILABLE = "查询失败（数据不可用）：" #: 运行中途补固定一个源时发的那条日志事件的 code（_pin_late） LATE_PIN_CODE = "data_version_pinned" +#: 每次查询之后基于数据目录的 SQL 检查最多等这么久。正常一条 SQL 几毫秒；等不到就放弃这一次检查， +#: 查询结果照常交回——检查是附加的，不能拖慢查询 +SQL_CHECK_TIMEOUT_S = 2.0 + class _QueryArgs(BaseModel): sql: str = Field(description="要执行的 SQL，一次一条") @@ -445,6 +450,15 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, # 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句， # 引擎按它告诉界面上限是多少（metadata["timeout_s"]） seconds = query_timeout(source) + # 基于数据目录的 SQL 检查器：和冻结进表结构快照的是同一版目录。第一次查询时才建（关系图也是用到才建） + checker: list[SqlChecker] = [] + + def _checker() -> SqlChecker: + if not checker: + checker.append(SqlChecker(kind=source.kind, schema_cache=source.schema_cache, + notes={name: entry.notes for name, entry in (catalog or {}).items()}, + source_name=source.name)) + return checker[0] async def _run(sql: str, limit: int | None = None) -> str: if gone := _vanished(source, fixed): @@ -470,6 +484,11 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, payload = result.to_payload() payload["source"] = source.name + # 对照数据目录检查这条 SQL（一对多关联后重复计算、存量跨期求和……）。结果交给模型（Agent 据此改写）， + # 也随查询快照存下：证据面板看得到，口径卡据此给指标标「存疑」、出具降档。没查出问题不加这个键， + # 快照和以前一字不差 + if checks := await _sql_checks(_checker(), sql): + payload["checks"] = checks # 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源 # 重新探查、改了结构，已经跑完的运行核对的还是当时那一份。上传源冻结的是绑定快照里那份 if schema := await _store_schema(source, ctx, catalog=catalog): @@ -511,6 +530,20 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, ) +async def _sql_checks(checker: SqlChecker, sql: str) -> list[dict[str, Any]]: + """一条查询的 SQL 检查结果（as_dict 的形状）。在线程里跑、限时 SQL_CHECK_TIMEOUT_S：解析是纯计算，放在事件循环里 + 会挡住同一进程里别的运行。出错、超时都返回空列表——检查不能让一次成功的查询变成失败。""" + if not checker.enabled: + return [] + try: + return await asyncio.wait_for(asyncio.to_thread(checker.check_dicts, sql), timeout=SQL_CHECK_TIMEOUT_S) + except asyncio.TimeoutError: + logger.warning("SQL 检查超过 %.1f 秒，本次跳过（数据源 %s）", SQL_CHECK_TIMEOUT_S, checker.source_name) + except Exception: # noqa: BLE001 + logger.exception("SQL 检查出错，本次跳过（数据源 %s）", checker.source_name) + return [] + + async def _store_schema(source: Any, ctx: ToolContext, *, catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> str | None: """把数据源此刻的 schema_cache 存成 schema_snapshot 工件，返回工件 id。 diff --git a/backend/tests/test_runtime_sqlcheck.py b/backend/tests/test_runtime_sqlcheck.py new file mode 100644 index 00000000..5a8029b4 --- /dev/null +++ b/backend/tests/test_runtime_sqlcheck.py @@ -0,0 +1,288 @@ +"""运行时的 SQL 检查：数据源查询工具每次执行成功后对照数据目录检查一遍。 + +- 结果写进查询结果（模型看得到，Agent 能据此改写 SQL）和查询快照的 checks 字段；没查出问题时不加这个键， + 快照和以前一字不差。检查出错、超时都不影响查询本身。 +- 证据接口的查询步骤带上 checks（不带给模型的那句 for_model）。 +- 契约指标的来源查询有 error 级问题：口径卡把这个指标标 sql_check_failed 并写明原因，出具按 gaps 降档； + 写作目录提醒写作者，证据面板的指标步骤标出来。和阶段 0 的「结果不完整」走同一条路。 +""" +from __future__ import annotations + +import asyncio +import json +import time +import uuid +from types import SimpleNamespace + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.core import artifact_store +from app.data import catalog +from app.data.engine import engines +from app.data.introspect import introspect +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, RunEvent +from app.engine.evidence import iter_segments +from app.engine.issuance import decide_tier, sql_check_gaps +from app.engine.runner import run_manager +from app.main import app +from app.tools.datasource import build_datasource_tools +from app.tools.registry import ToolContext +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" +REASON = ("所依据的查询未通过 SQL 检查（「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算），" + "结果不可靠") + +_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] + + +async def _source(path: str, *, with_catalog: bool = True) -> tuple[str, str]: + name = f"scenic_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=path, readonly=True, enabled=True) + row.schema_cache = await _schema(path) + session.add(row) + await session.commit() + if with_catalog: + for table, notes in scenic_notes.notes().items(): + await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏") + return row.id, name + + +@pytest.fixture +async def scenic_source(scenic_db): + source_id, name = await _source(scenic_db) + yield name + await engines.invalidate(source_id) + + +async def _query_tool(name: str): + ctx = ToolContext(run_id=uuid.uuid4().hex, node_id="n1", data_versions=None) + async with SessionLocal() as session: + [tool] = await build_datasource_tools([f"db_query__{name}"], ctx, session) + return tool + + +# -------------------------------------------------------------------------- +# 数据源查询工具 +# -------------------------------------------------------------------------- + + +async def test_query_result_and_snapshot_carry_checks(scenic_source): + tool = await _query_tool(scenic_source) + result = json.loads(await tool.coroutine(sql=FANOUT_SQL)) + assert result["rows"] and result["artifact"] + [check] = result["checks"] + assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders" + assert check["message"].startswith("「订单」关联「订单明细」是一对多") and "order_items" in check["for_model"] + assert artifact_store.load(result["artifact"])["checks"] == result["checks"] + + clean = json.loads(await tool.coroutine(sql=CLEAN_SQL)) + assert "checks" not in clean and "checks" not in artifact_store.load(clean["artifact"]) + + +async def test_sources_without_catalog_are_not_checked(scenic_db): + source_id, name = await _source(scenic_db, with_catalog=False) + try: + result = json.loads(await (await _query_tool(name)).coroutine(sql=FANOUT_SQL)) + assert result["rows"] and "checks" not in result + finally: + await engines.invalidate(source_id) + + +async def test_a_failing_or_slow_check_never_breaks_the_query(scenic_source, monkeypatch): + from app.data import sqlcheck + from app.tools import datasource + + tool = await _query_tool(scenic_source) + + def boom(self, sql): + raise RuntimeError("规则缺陷") + + monkeypatch.setattr(sqlcheck.SqlChecker, "check", boom) + result = json.loads(await tool.coroutine(sql=FANOUT_SQL)) + assert result["rows"] and "checks" not in result + + def slow(self, sql): + time.sleep(1.0) + return [] + + monkeypatch.setattr(sqlcheck.SqlChecker, "check", slow) + monkeypatch.setattr(datasource, "SQL_CHECK_TIMEOUT_S", 0.05) + started = time.monotonic() + result = json.loads(await tool.coroutine(sql=FANOUT_SQL)) + assert result["rows"] and "checks" not in result + assert time.monotonic() - started < 0.9, "检查超时就放弃，不等它跑完" + + +# -------------------------------------------------------------------------- +# 出具：契约指标的来源查询有 error 级问题时降档 +# -------------------------------------------------------------------------- + + +def test_sql_check_failures_become_readable_gaps(): + metrics = [{"id": "gmv", "name": "订单金额", "value": 1, "sql_check_failed": True, "sql_check_reason": REASON}, + {"id": "n", "name": "订单数", "value": 2}] + assert sql_check_gaps(metrics) == [f"指标「订单金额」{REASON}"] + assert sql_check_gaps(metrics[1:]) == [] + tier = decide_tier(missing_required=[], missing_expected=[], unmatched=[], strict=False, + gaps=sql_check_gaps(metrics)) + assert tier == "degraded" + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def chain(nodes: list[dict]) -> dict: + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +GMV = {"id": "gmv", "name": "订单金额", "unit": "元", "format": "plain", "expression": "cell(nodes.fetch, 0, 'gmv')"} +#: 同一份结果经数据整形解析后按下标取：来历记成 transform，要顺着取值链找到那份查询结果 +GMV2 = {"id": "gmv2", "name": "订单金额（整形后）", "unit": "元", "format": "plain", + "expression": "vars.res.rows[0][0]"} +WEEKLY = "本周订单金额 [[m:gmv]]。" + + +def weekly(source: str, sql: str) -> dict: + return chain([ + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}), + node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="res"), + node("card", "metrics", caliber="订单口径", caliber_version="v1", metrics=[GMV, GMV2]), + node("write", "report", instructions="写订单周报"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"]}), + ]) + + +class Writer: + """报告写作者：每次都照 text 写；记下看到的提示。""" + + def __init__(self, monkeypatch, text: str = WEEKLY): + from app.providers import mock_model + + self.prompts: list[str] = [] + + def decide(model, messages): + self.prompts.append("\n".join(str(m.content) for m in messages)) + return AIMessage(content=text) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + + +async def wait(run_id: str) -> Run: + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + return row + raise AssertionError(f"run {run_id} 没有结束：{row.status}") + + +async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype) + if node_id: + q = q.where(RunEvent.node_id == node_id) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +async def output_of(run_id: str, node_id: str) -> dict: + [finished] = await events(run_id, "node.finished", node_id) + return artifact_store.load(finished.data["artifact"]) + + +async def test_a_fanout_behind_a_contract_metric_degrades_the_issuance(scenic_source, engine_up, monkeypatch): + writer = Writer(monkeypatch) + run = await run_manager.start(graph=weekly(scenic_source, FANOUT_SQL), input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + + card = await output_of(row.id, "card") + metrics = {m["id"]: m for m in card["metrics"]} + [query_end] = [e.data for e in await events(row.id, "tool.end", "fetch")] + for mid in ("gmv", "gmv2"): + assert metrics[mid]["sql_check_failed"] is True and metrics[mid]["sql_check_reason"] == REASON + assert metrics[mid]["sql_check_sources"] == [{"artifact": query_end["query_artifact"], + "codes": ["fanout_sum"]}] + assert "（存疑：所依据的查询未通过 SQL 检查）" in card["text"] + warns = [e.data for e in await events(row.id, "log", "card") if e.data.get("code") == "metric_sql_check"] + assert [w["message"] for w in warns] == [f"指标「订单金额」{REASON}", f"指标「订单金额（整形后）」{REASON}"] + + issuance = row.output["_issuance"] + assert issuance["tier"] == "degraded", issuance + assert f"指标「订单金额」{REASON}" in issuance["gaps"] + # 写作目录提醒写作者：这个数存疑，引用时要说明 + assert "所依据的查询未通过 SQL 检查，结果存疑" in writer.prompts[0] + + # 证据面板：指标步骤标出来，查询步骤带着检查结果（不带给模型的那句） + doc = artifact_store.load(row.output["_evidence"]["doc_artifact"]) + assert doc["catalog"]["m:gmv"]["sql_check_failed"] is True + seg = next(s for s in iter_segments(doc) if s.get("ref") == "m:gmv") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json() + metric = body["chain"][0] + assert metric["step"] == "metric" and metric["sql_check_failed"] is True + assert metric["sql_check_reason"] == REASON + [query] = [s for s in body["chain"] if s["step"] == "query"] + [check] = query["checks"] + assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders" + assert check["message"].startswith("「订单」关联「订单明细」是一对多") and "for_model" not in check + + +async def test_the_same_card_on_a_clean_query_issues_formally(scenic_source, engine_up, monkeypatch): + Writer(monkeypatch) + run = await run_manager.start(graph=weekly(scenic_source, CLEAN_SQL), input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + card = await output_of(row.id, "card") + assert all("sql_check_failed" not in m for m in card["metrics"]) + issuance = row.output["_issuance"] + assert issuance["tier"] == "formal" and issuance["gaps"] == [], issuance + doc = artifact_store.load(row.output["_evidence"]["doc_artifact"]) + seg = next(s for s in iter_segments(doc) if s.get("ref") == "m:gmv") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json() + assert "sql_check_failed" not in body["chain"][0] + assert all("checks" not in s for s in body["chain"] if s["step"] == "query") + + +async def test_warnings_alone_do_not_degrade(scenic_source, engine_up, monkeypatch): + """只有 warning（这里是没按订单状态筛）：查询结果和快照里照样带着，但不影响出具档位。""" + Writer(monkeypatch) + sql = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id" + run = await run_manager.start(graph=weekly(scenic_source, sql), input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + [query_end] = [e.data for e in await events(row.id, "tool.end", "fetch")] + snapshot = artifact_store.load(query_end["query_artifact"]) + assert [c["code"] for c in snapshot["checks"]] == ["missing_valid_filter"] + assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]