AgentLab/backend/tests/test_runtime_sqlcheck.py

"""运行时的 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
from tests.fixtures.sources import drop_source

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]


_CREATED: list[str] = []


@pytest.fixture(autouse=True)
async def _drop_created_sources():
    yield
    while _CREATED:
        await drop_source(_CREATED.pop())


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()
        _CREATED.append(row.id)
        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):
    _, name = await _source(scenic_db)
    yield name


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"]