From 327c551bdbf9d80f6d911e9a5265710a8b210e77 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:07:15 +0800 Subject: [PATCH 1/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20/=20B1=20=E6=9C=8D=E5=8A=A1=E7=AB=AF=EF=BC=9ASQL=20=E5=8F=AA?= =?UTF-8?q?=E5=86=99=E5=9C=A8=20Agent=20=E9=87=8C=E7=9A=84=E6=A8=A1?= =?UTF-8?q?=E6=9D=BF=E4=B9=9F=E6=8F=90=E9=86=92=E7=9B=AE=E5=BD=95=E5=8F=98?= =?UTF-8?q?=E5=8C=96=E2=80=94=E2=80=94=E5=8F=91=E5=B8=83=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E5=88=86=E7=9B=B4=E6=8E=A5=E5=BC=95=E7=94=A8=E5=92=8C=E5=8F=AF?= =?UTF-8?q?=E8=83=BD=E6=B6=89=E5=8F=8A=E4=B8=A4=E7=BB=84=EF=BC=8CAgent=20?= =?UTF-8?q?=E5=92=8C=E5=8D=8F=E4=BD=9C=E6=88=90=E5=91=98=E7=BB=91=E5=AE=9A?= =?UTF-8?q?=E4=BA=86=E6=9F=A5=E8=AF=A2=E5=B7=A5=E5=85=B7=E7=9A=84=E6=BA=90?= =?UTF-8?q?=E8=AE=B0=E4=B8=8B=E6=89=80=E6=9C=89=E6=9C=89=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=E7=9A=84=E8=A1=A8=EF=BC=8C=E6=AD=A3=E5=BC=8F=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E4=B8=A4=E7=BB=84=E9=83=BD=E6=AF=94=E3=80=81=E6=AF=8F=E5=BC=A0?= =?UTF-8?q?=E8=A1=A8=E6=A0=87=E6=98=8E=E5=B1=9E=E4=BA=8E=E5=93=AA=E4=B8=80?= =?UTF-8?q?=E7=BB=84=EF=BC=9B=E5=8F=AA=E8=AE=B0=E4=BA=86=E7=9B=B4=E6=8E=A5?= =?UTF-8?q?=E5=BC=95=E7=94=A8=E7=9A=84=E6=97=A7=E8=AE=B0=E5=BD=95=E7=85=A7?= =?UTF-8?q?=E6=97=A7=E6=8C=89=E7=9B=B4=E6=8E=A5=E5=BC=95=E7=94=A8=E6=AF=94?= =?UTF-8?q?=EF=BC=9B=E5=93=AA=E4=BA=9B=E8=8A=82=E7=82=B9=E7=AE=97=E5=86=99?= =?UTF-8?q?=E6=AD=BB=E7=9A=84=E6=9F=A5=E8=AF=A2=E7=BB=9F=E4=B8=80=E4=BA=A4?= =?UTF-8?q?=E7=BB=99=20tool=5Fqueries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/runs.py | 3 +- backend/app/api/workflows.py | 12 +- backend/app/core/events.py | 3 +- backend/app/data/catalog_impact.py | 251 ++++++++++++++------ backend/tests/test_catalog_publish_drift.py | 171 +++++++++++-- 5 files changed, 351 insertions(+), 89 deletions(-) diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index bae67e6d..fb424fe1 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -186,7 +186,8 @@ async def start_run( await run_manager.note(run.id, "log", **governance_note( governed=governed, workflow_name=name, version=version)) if catalog_changes: - # 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表 + # 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。 + # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的（界面分开说） await run_manager.note(run.id, "catalog.drift", version=version, count=len(catalog_changes), tables=catalog_changes) # 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的" diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index fef036fa..e24a6be8 100644 --- a/backend/app/api/workflows.py +++ b/backend/app/api/workflows.py @@ -10,7 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.errors import graph_error from app.api.runs import actor_of from app.core.artifact_store import graph_hash -from app.data.catalog_impact import catalog_drift, catalog_versions_for +from app.data.catalog_impact import catalog_drift, catalog_versions_for, recorded_versions from app.db.base import get_session from app.db.models import Run, Workflow, WorkflowVersion from app.engine.governance import publish_issues @@ -210,9 +210,11 @@ class VersionDetailOut(VersionOut): #: 画布上改过入口字段时，拿当前画布的字段去跑已发布版本会传错参数 input_fields: list[dict[str, Any]] = Field(default_factory=list) published: bool = False - #: 发布时这一版 SQL 用到的表的数据目录版本 {源名: {表名: 版本}}；没发布过、或者记录之前发布的为 null + #: 发布时记下的数据目录版本 {"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}：direct 写死的 + #: SQL 用到的表，possible Agent 可能查询的表（catalog_impact.recorded_versions，只记了直接引用的旧记录 possible + #: 为空）；没发布过、或者记录之前发布的为 null catalog_versions: dict[str, Any] | None = None - #: 发布之后目录有变化的表：[{source, source_id, table, label, published, current}] + #: 发布之后目录有变化的表：[{source, source_id, table, label, published, current, impact}] catalog_changes: list[dict[str, Any]] = Field(default_factory=list) @@ -248,7 +250,7 @@ async def get_version( created_at=snapshot.created_at, workflow_id=workflow_id, graph=graph, graph_hash=snapshot.graph_hash or graph_hash(graph), input_fields=fields, published=workflow.published_version == snapshot.version, level=snapshot.level, - catalog_versions=snapshot.catalog_versions, + catalog_versions=recorded_versions(snapshot.catalog_versions), catalog_changes=await catalog_drift(session, snapshot.catalog_versions), ) @@ -352,7 +354,7 @@ async def publish_workflow( "version": version, "graph_hash": snapshot.graph_hash, "published_by": workflow.published_by, - "catalog_versions": snapshot.catalog_versions, + "catalog_versions": recorded_versions(snapshot.catalog_versions), "issues": issues, # 剩下的都是警告 } diff --git a/backend/app/core/events.py b/backend/app/core/events.py index c264829d..95c5211a 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -69,7 +69,8 @@ class EventType(StrEnum): # 出具与口径治理 ISSUANCE = "issuance" # 出具契约的判定结果（档位 / 缺数据 / 未回指数字） CALIBER_UPGRADE = "caliber.upgrade" # 钉住的方法卡有新版本，按声明的策略处置 - # 从发布版本发起的正式运行：发布之后这一版 SQL 用到的表的数据目录有变化（data/catalog_impact.catalog_drift）。 + # 从发布版本发起的正式运行：发布之后这一版用到的表的数据目录有变化（data/catalog_impact.catalog_drift）。 + # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的。 # 只提醒、不拦运行；落在封存范围内，出具物上看得到「当时的目录和发布时不一样」 CATALOG_DRIFT = "catalog.drift" # 报告撰写节点的自查结果：报告文档工件 id、统计、违规清单。落在封存范围内， diff --git a/backend/app/data/catalog_impact.py b/backend/app/data/catalog_impact.py index 7fd60f17..f8f3daa5 100644 --- a/backend/app/data/catalog_impact.py +++ b/backend/app/data/catalog_impact.py @@ -5,17 +5,20 @@ - **影响面**：哪些已发布、受管的模板用到了这张表（table_impact）。按每个模板当前的已发布版本算，不看画布上的 草稿——正式运行跑的是已发布的那一版。调用工具节点写死的 SQL 用到了这张表算「直接引用」；Agent 绑定了这个源 的查询工具算「可能涉及」（SQL 是运行时写的，静态看不出）；合并查询按它的输入一路追溯。 -- **发布时的目录版本**：发布时记下模板 SQL 用到的每张表的目录版本 {源名: {表名: 版本}}（catalog_versions_for）， - 从这一版发起正式运行时和当前的比（catalog_drift），有变化只提醒、不拦运行——目录是给人和助手看的说明， - 不是运行的输入；拦下来等于让一次文字修订卡住正式出具。 +- **发布时的目录版本**：发布时记下模板用到的表的目录版本（catalog_versions_for），和影响面同一个说法分两组： + {"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}。direct 是写死的 SQL 用到的表；possible 是 + Agent、协作成员绑定了查询工具的源里所有有目录的表。从这一版发起正式运行时和当前的比（catalog_drift），有变化 + 只提醒、不拦运行——目录是给人和助手看的说明，不是运行的输入；拦下来等于让一次文字修订卡住正式出具。 表名的取法和证据台账同一个口径：evidence.sql_tables 取 FROM / JOIN 后面的表（去掉注释、字符串、CTE）， -去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。 +去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。哪些节点算「写死的查询」以 +sqlcheck.tool_query 为准（tool_queries），发布记录、影响面、SQL 检查和助手的上下文认的是同一种节点。 """ from __future__ import annotations import logging -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Iterator, Mapping +from types import SimpleNamespace from typing import Any from sqlalchemy import select @@ -51,20 +54,43 @@ def sql_table_keys(sql: str, tables: Iterable[str] | None) -> set[str]: return {hit for s in shorts if (hit := lookup.get(name_key(s))) is not None} +def _as_node(raw: Mapping[str, Any]) -> Any: + """字典写法的节点 → tool_query 认的样子（有 type、config）。画布 {data: {config}} 和 GraphSpec {config} 都认。""" + config = raw.get("config") + if not isinstance(config, Mapping): + data = raw.get("data") + config = data.get("config") if isinstance(data, Mapping) else None + return SimpleNamespace(id=raw.get("id"), type=raw.get("type"), config=config if isinstance(config, Mapping) else {}) + + +def tool_queries(nodes: Iterable[Any]) -> Iterator[tuple[Any, str, str]]: + """一张图里调用工具节点写死的查询：(节点, 源名, SQL)，按图里的顺序。 + + 判定一律交给 sqlcheck.tool_query（是不是调用工具节点、是不是数据源查询工具、SQL 是不是非空字符串）：发布时 + 记目录版本、影响面、SQL 检查、助手给全字段的表都走这一份，不各写一遍。节点认 GraphSpec 的节点对象，也认还没 + 校验过的字典（画布写法、GraphSpec 写法）——助手改图时手上的是字典，图不一定完整到能过 GraphSpec。 + """ + from app.data.sqlcheck import tool_query + + for node in nodes or []: + if isinstance(node, Mapping): + found = tool_query(_as_node(node)) + elif node is not None: + found = tool_query(node) + else: + found = None + if found: + yield node, found[0], found[1] + + def graph_sql_tables(spec: Any, tables_by_source: Mapping[str, Iterable[str]] | None = None) -> dict[str, set[str]]: - """一张图里调用工具节点写死的 SQL 用到的表：{源名: {表名}}。Agent 的 SQL 运行时才有，不在这里。 + """一张图里调用工具节点写死的 SQL 用到的表：{源名: {表名}}。Agent 的 SQL 运行时才有，不在这里（见 agent_sources）。 tables_by_source 是各源表结构里的表名（schema_cache 的 tables 的键）：给了的源按它规范表名、对不上的不算； 没给的源用 SQL 里的原名。 """ - from app.data.sqlcheck import tool_query - out: dict[str, set[str]] = {} - for node in getattr(spec, "nodes", []) or []: - found = tool_query(node) - if not found: - continue - source, sql = found + for _, source, sql in tool_queries(getattr(spec, "nodes", []) or []): known = (tables_by_source or {}).get(source) hits = sql_table_keys(sql, list(known) if known is not None else None) if hits: @@ -82,6 +108,36 @@ def _setting(spec: Any, node: Any, key: str) -> Any: return (getattr(spec, "defaults", None) or {}).get(key) if value in (None, "") else value +def _query_sources(tools: Any) -> set[str]: + """工具清单里绑定的数据源查询工具（db_query__<源>）→ 源名。""" + from app.tools.datasource import QUERY_PREFIX + + names = [t.strip() for t in tools if isinstance(t, str)] if isinstance(tools, (list, tuple)) else [] + return {t[len(QUERY_PREFIX):] for t in names if t.startswith(QUERY_PREFIX) and len(t) > len(QUERY_PREFIX)} + + +def agent_bindings(spec: Any) -> list[tuple[Any, str | None, set[str]]]: + """Agent 节点和协作成员各自绑定了哪些源的查询工具：[(节点, 成员名或 None, {源名})]，按图里的顺序。 + + 它们的 SQL 运行时才由模型写出来，静态看不出用哪张表：影响面算「可能涉及」，发布时记下整个源有目录的表。 + """ + out: list[tuple[Any, str | None, set[str]]] = [] + for node in getattr(spec, "nodes", []) or []: + kind = _kind(node) + if kind == "agent": + out.append((node, None, _query_sources(_setting(spec, node, "tools")))) + elif kind == "supervisor": + for member in _setting(spec, node, "agents") or []: + if isinstance(member, Mapping): + out.append((node, str(member.get("name") or ""), _query_sources(member.get("tools")))) + return out + + +def agent_sources(spec: Any) -> set[str]: + """Agent 节点、协作成员绑定了查询工具的数据源名。""" + return {name for _, _, bound in agent_bindings(spec) for name in bound} + + def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | None) -> list[dict[str, Any]]: """一张图里哪些节点用到了某个源的某张表，按图里的顺序。每项 {node_id, label, type, impact, via?, member?}。 @@ -90,10 +146,6 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No - 合并查询：输入（别名 → 节点）里有用到这张表的 → 取输入里最强的那一级，via 写经由哪些输入。合并的合并 一路追到底；成环（校验会报错的图）时停下。 """ - from app.data.sqlcheck import tool_query - from app.tools.datasource import QUERY_PREFIX - - tool = f"{QUERY_PREFIX}{source}" wanted = name_key(table) known = list(tables) if tables is not None else None nodes = list(getattr(spec, "nodes", []) or []) @@ -104,20 +156,16 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No def base(node: Any, impact: str, **extra: Any) -> dict[str, Any]: return {"node_id": node.id, "label": node.title, "type": _kind(node), "impact": impact, **extra} - for node in nodes: - kind = _kind(node) - if kind == "tool": - found = tool_query(node) - if found and found[0] == source and wanted in {name_key(t) for t in sql_table_keys(found[1], known)}: - own[node.id] = base(node, "direct") - elif kind == "agent": - if tool in [str(t).strip() for t in _setting(spec, node, "tools") or [] if isinstance(t, str)]: - own[node.id] = base(node, "possible") - elif kind == "supervisor": - for member in _setting(spec, node, "agents") or []: - if isinstance(member, Mapping) and tool in [str(t).strip() for t in member.get("tools") or [] - if isinstance(t, str)]: - out.append(base(node, "possible", member=str(member.get("name") or ""))) + for node, name, sql in tool_queries(nodes): + if name == source and wanted in {name_key(t) for t in sql_table_keys(sql, known)}: + own[node.id] = base(node, "direct") + for node, member, bound in agent_bindings(spec): + if source not in bound: + continue + if member is None: + own[node.id] = base(node, "possible") + else: + out.append(base(node, "possible", member=member)) resolved: dict[str, dict[str, Any] | None] = {} @@ -224,66 +272,135 @@ def _schema_tables(view: Any) -> list[str] | None: return list(tables) if isinstance(tables, Mapping) and tables else None -async def catalog_versions_for(session: AsyncSession, spec: Any) -> dict[str, dict[str, int]]: - """发布时要记下的目录版本：{源名: {表名: 版本}}，只收模板 SQL 用到的表；还没有目录的表记 0。 +#: 发布记录的两组，和影响面的两级同名：direct 写死的 SQL 用到的表，possible Agent 可能查询的表 +CatalogVersions = dict[str, dict[str, dict[str, int]]] - 记 0 是有意的：发布之后有人给这张表补了目录，同样算「自发布以来有变化」。找不到的源（停用、删了）不记： - 没有目录可比，正式运行时工具自己会报源不存在。读库出错时返回空 dict——记不下版本不该挡住发布。 + +def _pinned_group(value: Any) -> dict[str, dict[str, Any]] | None: + """{源名: {表名: 版本}} 的一组；形状不对返回 None。""" + if not isinstance(value, Mapping) or not all(isinstance(t, Mapping) for t in value.values()): + return None + return {str(k): dict(v) for k, v in value.items()} + + +def recorded_versions(recorded: Any) -> CatalogVersions | None: + """发布时记下的目录版本，统一成 {"direct": {...}, "possible": {...}}。没记过返回 None。 + + 认两种写法： + - 现在的写法：顶层只有 direct、possible 两个键，每组 {源名: {表名: 版本}}。 + - 可能涉及上线之前的写法：{源名: {表名: 版本}}，只记了写死的 SQL 用到的表——整个算 direct，possible 为空： + 那时没记 Agent 可能查询的表，不知道当时是哪一版，不能说它们变了。 + 两种写法靠第二层分得开：旧写法第二层是版本号（整数），新写法第二层是 {表名: 版本}。源名可以叫 direct， + 所以不能只看顶层的键。 + """ + if not isinstance(recorded, Mapping): + return None + if recorded and set(recorded) <= set(IMPACTS): + groups = {k: _pinned_group(recorded.get(k, {})) for k in IMPACTS} + if all(g is not None for g in groups.values()): + return {k: g or {} for k, g in groups.items()} + legacy = _pinned_group(recorded) + return {"direct": legacy, "possible": {}} if legacy is not None else None + + +async def catalog_versions_for(session: AsyncSession, spec: Any) -> CatalogVersions | None: + """发布时要记下的目录版本：{"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}。 + + - direct：调用工具节点写死的 SQL 用到的表；还没有目录的表记 0。记 0 是有意的：发布之后有人给这张表补了 + 目录，同样算「自发布以来有变化」。 + - possible：Agent 节点、协作成员绑定了某个源的查询工具（影响面里的「可能涉及」）。SQL 运行时才写，看不出用哪 + 张表，于是记下这个源里**所有有目录的表**；direct 里已经有的不重复记。发布时还没有目录的表不逐张记 0（一个 + 库几百张表），比对时「之后新建了目录」照样算变化（catalog_drift）。源里一张有目录的表都没有时记一个空的 + {}，表示这个源在看着。 + + 找不到的源（停用、删了）不记：没有目录可比，正式运行时工具自己会报源不存在。读库出错时返回 None（和没记过 + 一样，不提醒）——记不下版本不该挡住发布。 """ from app.data import catalog try: - names = graph_sql_tables(spec).keys() - sources = await _sources_by_name(session, names) + possible_names = agent_sources(spec) + sources = await _sources_by_name(session, [*graph_sql_tables(spec).keys(), *possible_names]) tables_by_source = {name: _schema_tables(view) for name, (_, view) in sources.items()} used = graph_sql_tables(spec, {k: v for k, v in tables_by_source.items() if v is not None}) - out: dict[str, dict[str, int]] = {} + entries_of: dict[str, dict[str, Any]] = {} + + async def entries(name: str) -> dict[str, Any]: + if name not in entries_of: + entries_of[name] = await catalog.read_catalog(session, sources[name][0].id) + return entries_of[name] + + direct: dict[str, dict[str, int]] = {} for name, tables in sorted(used.items()): - if name not in sources: - continue - entries = await catalog.read_catalog(session, sources[name][0].id) - out[name] = {t: (entries[t].version if t in entries else 0) for t in sorted(tables)} - return out + if name in sources: + found = await entries(name) + direct[name] = {t: (found[t].version if t in found else 0) for t in sorted(tables)} + possible: dict[str, dict[str, int]] = {} + for name in sorted(possible_names): + if name in sources: + pinned = direct.get(name, {}) + possible[name] = {t: e.version for t, e in sorted((await entries(name)).items()) if t not in pinned} + return {"direct": direct, "possible": possible} except Exception: # noqa: BLE001 logger.exception("记录发布时的目录版本失败，本次不记") - return {} + return None + + +def _label(entry: Any) -> str | None: + """表现在的中文名；没有或被驳回时为 None。""" + label = (entry.notes.get("label") if entry else None) or {} + return label.get("value") if isinstance(label, dict) and label.get("status") != "rejected" else None async def catalog_drift(session: AsyncSession, recorded: Mapping[str, Any] | None) -> list[dict[str, Any]]: - """发布时记下的目录版本和现在比：变了的表 [{source, source_id, table, label, published, current}]。 + """发布时记下的目录版本和现在比：变了的表 [{source, source_id, table, label, published, current, impact}]。 - 没记过（这个功能之前发布的版本）返回空列表——不知道当时是哪一版，不能说它变了。现在找不到的源跳过（理由同 - catalog_versions_for）。label 是表现在的中文名（没有或被驳回时为 None）。读库出错返回空列表：提醒是附加的。 + impact 是这张表属于哪一组（direct 直接引用 / possible Agent 可能查询的），direct 在前，同组按源名、表名。 + possible 的源里发布时没记、现在有了目录的表也算（published 为 0），direct 里已有的表不重复报。 + + 没记过（这个功能之前发布的版本）返回空列表——不知道当时是哪一版，不能说它变了；只记了直接引用的旧记录 + 同理不报可能涉及（recorded_versions）。现在找不到的源跳过（理由同 catalog_versions_for）。label 是表现在的 + 中文名（没有或被驳回时为 None）。读库出错返回空列表：提醒是附加的。 """ from app.data import catalog - if not isinstance(recorded, Mapping) or not recorded: + groups = recorded_versions(recorded) + if not groups or not any(groups.values()): return [] try: - sources = await _sources_by_name(session, recorded.keys()) + sources = await _sources_by_name(session, [*groups["direct"], *groups["possible"]]) + entries_of: dict[str, dict[str, Any]] = {} + + async def entries(name: str) -> dict[str, Any]: + if name not in entries_of: + entries_of[name] = await catalog.read_catalog(session, sources[name][0].id) + return entries_of[name] + out: list[dict[str, Any]] = [] - for name in sorted(recorded): - pinned = recorded[name] - if name not in sources or not isinstance(pinned, Mapping): - continue - row = sources[name][0] - entries = await catalog.read_catalog(session, row.id) - for table in sorted(pinned): - then = pinned[table] - entry = entries.get(table) - now = entry.version if entry else 0 - if isinstance(then, int) and then == now: + for impact in IMPACTS: + for name in sorted(groups[impact]): + if name not in sources: continue - label = (entry.notes.get("label") if entry else None) or {} - out.append({"source": name, "source_id": row.id, "table": table, - "label": label.get("value") if isinstance(label, dict) and label.get("status") != "rejected" - else None, - "published": then if isinstance(then, int) else None, "current": now}) + pinned = dict(groups[impact][name]) + found = await entries(name) + if impact == "possible": + # 发布时没有目录、之后新建了的表：记录里没有它，按发布时「尚无目录」算 + skip = groups["direct"].get(name, {}) + pinned.update({t: 0 for t in found if t not in pinned and t not in skip}) + for table in sorted(pinned): + then = pinned[table] + entry = found.get(table) + now = entry.version if entry else 0 + if isinstance(then, int) and then == now: + continue + out.append({"source": name, "source_id": sources[name][0].id, "table": table, + "label": _label(entry), "published": then if isinstance(then, int) else None, + "current": now, "impact": impact}) return out except Exception: # noqa: BLE001 logger.exception("比对发布时的目录版本失败，本次不提醒") return [] -__all__ = ["IMPACTS", "catalog_drift", "catalog_versions_for", "graph_sql_tables", "sql_table_keys", "table_impact", - "template_refs"] +__all__ = ["IMPACTS", "CatalogVersions", "agent_bindings", "agent_sources", "catalog_drift", "catalog_versions_for", + "graph_sql_tables", "recorded_versions", "sql_table_keys", "table_impact", "template_refs", "tool_queries"] diff --git a/backend/tests/test_catalog_publish_drift.py b/backend/tests/test_catalog_publish_drift.py index d08ae51e..ea5b1b07 100644 --- a/backend/tests/test_catalog_publish_drift.py +++ b/backend/tests/test_catalog_publish_drift.py @@ -1,10 +1,14 @@ """发布时记下目录版本，正式运行时提示目录的变化（阶段 4A）。 -- 发布（把某一版立为正式版本）时，记下这个模板 SQL 用到的每张表的目录版本 {源名: {表名: 版本}}，存在 - workflow_versions.catalog_versions；还没有目录的表记 0（之后补了目录也算变化）。 -- 从这一版发起正式运行时和当前的目录版本比：有变化发一条 catalog.drift 事件（列出表和前后版本），**不拦运行**； - 目录没变不发。 +- 发布（把某一版立为正式版本）时，记下这个模板用到的表的目录版本，存在 workflow_versions.catalog_versions， + 分两组：{"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}。 + - direct：调用工具节点写死的 SQL 用到的表；还没有目录的表记 0（之后补了目录也算变化）。 + - possible：Agent 节点、协作成员绑定了某个源的查询工具——SQL 运行时才写，发布时记下这个源里所有有目录的表 + （直接引用的不重复记）。之后新建了目录的表同样算变化。 +- 从这一版发起正式运行时和当前的目录版本比：有变化发一条 catalog.drift 事件（列出表、前后版本、属于哪一组）， + **不拦运行**；目录没变不发。 - 版本详情给出发布时的目录版本，以及哪些表在那之后变了。 +- 没有记录的老版本不提醒；只有直接引用那一种写法（{源名: {表名: 版本}}）的记录照旧按直接引用比，不提醒可能涉及。 """ from __future__ import annotations @@ -81,18 +85,52 @@ def node(nid, ntype, **config): return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} +def _chain(*nodes) -> dict: + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + def daily(source: str) -> dict: - nodes = [ + return _chain( node("start", "input"), node("fetch", "tool", tool=f"db_query__{source}", args={"sql": SQL}), node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.fetch }}"}]), - ] - return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + ) -async def _publish(client, source) -> str: +def agent_only(source: str) -> dict: + """SQL 只写在 Agent 里：图上没有一条写死的查询。""" + return _chain( + node("start", "input"), + node("ask", "agent", prompt="查一下昨天各景区的入园人数", tools=[f"db_query__{source}"]), + node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.ask.text }}"}]), + ) + + +def team(source: str) -> dict: + """协作成员里有一个绑定了查询工具。""" + return _chain( + node("start", "input"), + node("crew", "supervisor", agents=[ + {"name": "取数员", "prompt": "查入园人数", "tools": [f"db_query__{source}"]}, + {"name": "撰稿人", "prompt": "写一句总结", "tools": []}, + ]), + node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.crew.text }}"}]), + ) + + +def mixed(source: str) -> dict: + """写死的查询用到入园记录和景区，Agent 还绑着同一个源。""" + return _chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": SQL}), + node("ask", "agent", prompt="再看看别的表", tools=[f"db_query__{source}"]), + node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.ask.text }}"}]), + ) + + +async def _publish(client, source, graph: dict | None = None) -> str: wf = (await client.post("/api/workflows", json={"name": f"入园日报-{uuid.uuid4().hex[:6]}", - "graph": daily(source.name)})).json()["id"] + "graph": graph or daily(source.name)})).json()["id"] source.made.append(wf) body = (await client.post(f"/api/workflows/{wf}/publish", json={"level": "published"})).json() assert body["ok"] is True, body @@ -106,12 +144,13 @@ async def _events(run_id: str, kind: str) -> list[dict]: return [e.data for e in rows] -async def _settle(run_id: str) -> Run: +async def _settle(run_id: str, *, sealed: bool = True) -> Run: + """等运行结束。sealed=False 时不等封存：Agent 没有可用的模型，跑到它那一步就失败，只看开头的事件。""" 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: + if row.status in ("succeeded", "failed") and (row.manifest_seq is not None or not sealed): return row raise AssertionError(f"run {run_id} 没跑完：{row.status}") @@ -125,15 +164,35 @@ async def _edit_visits(source_id: str) -> int: return saved.version +async def _recorded(wf: str) -> dict | None: + async with SessionLocal() as session: + return (await session.execute(select(WorkflowVersion.catalog_versions).where( + WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one() + + +async def _catalogued(source_id: str) -> dict[str, int]: + """这个源现在所有有目录的表：{表名: 版本}。""" + async with SessionLocal() as session: + return {t: e.version for t, e in (await catalog.read_catalog(session, source_id)).items()} + + +async def _formal(client, wf: str, *, sealed: bool = True) -> str: + r = await client.post("/api/runs", json={"workflow_id": wf, "run_class": "formal"}) + assert r.status_code == 201, r.text # 不拦 + run_id = r.json()["id"] + await _settle(run_id, sealed=sealed) + return run_id + + async def test_publish_records_catalog_versions(client, source): wf = await _publish(client, source) async with SessionLocal() as session: snap = (await session.execute(select(WorkflowVersion).where( WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one() - # 入园记录有目录（第 1 版）；景区表还没有目录，记 0 - assert snap.catalog_versions == {source.name: {"parks": 0, "visits": 1}} + # 入园记录有目录（第 1 版）；景区表还没有目录，记 0。没有 Agent 查库，可能涉及的一组是空的 + assert snap.catalog_versions == {"direct": {source.name: {"parks": 0, "visits": 1}}, "possible": {}} detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() - assert detail["catalog_versions"] == {source.name: {"parks": 0, "visits": 1}} + assert detail["catalog_versions"] == {"direct": {source.name: {"parks": 0, "visits": 1}}, "possible": {}} assert detail["catalog_changes"] == [] @@ -149,7 +208,7 @@ async def test_formal_run_warns_on_catalog_change_but_runs(client, source): assert drift["count"] == 1 and drift["version"] == 1 [table] = drift["tables"] assert table == {"source": source.name, "source_id": source.id, "table": "visits", "label": "入园记录", - "published": 1, "current": now} + "published": 1, "current": now, "impact": "direct"} # 版本详情也写明哪些表在发布之后变了 detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() assert [(c["table"], c["published"], c["current"]) for c in detail["catalog_changes"]] == [("visits", 1, now)] @@ -167,6 +226,88 @@ async def test_catalog_added_after_publish_counts_as_change(client, source): assert [(t["table"], t["published"], t["current"], t["label"]) for t in drift["tables"]] == [("parks", 0, 1, "景区")] +# -------------------------------------------------------------------------- +# SQL 只写在 Agent 里：可能涉及的表 +# -------------------------------------------------------------------------- + + +async def test_agent_only_template_records_every_catalogued_table_as_possible(client, source): + """影响面把绑定了查询工具的 Agent 算作「可能涉及」，发布记录得是同一个说法：记下这个源所有有目录的表。""" + wf = await _publish(client, source, agent_only(source.name)) + recorded = await _recorded(wf) + assert recorded == {"direct": {}, "possible": {source.name: await _catalogued(source.id)}} + assert "visits" in recorded["possible"][source.name] and "parks" not in recorded["possible"][source.name] + detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() + assert detail["catalog_versions"] == recorded and detail["catalog_changes"] == [] + + +async def test_agent_only_template_warns_after_a_catalog_change(client, source): + wf = await _publish(client, source, agent_only(source.name)) + now = await _edit_visits(source.id) + run_id = await _formal(client, wf, sealed=False) + [drift] = await _events(run_id, "catalog.drift") + assert drift["count"] == 1 + assert drift["tables"] == [{"source": source.name, "source_id": source.id, "table": "visits", "label": "入园记录", + "published": 1, "current": now, "impact": "possible"}] + detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() + assert [(c["table"], c["impact"]) for c in detail["catalog_changes"]] == [("visits", "possible")] + + +async def test_catalog_created_after_publish_counts_for_possible_tables(client, source): + """发布时没有目录的表不逐张记 0（一个库几百张表），之后有人给它建了目录同样算变化。""" + wf = await _publish(client, source, agent_only(source.name)) + async with SessionLocal() as session: + await catalog.write_entry(session, source.id, "parks", {"label": catalog.make_item("景区", "human")}, + if_version=0, actor="王敏") + run_id = await _formal(client, wf, sealed=False) + [drift] = await _events(run_id, "catalog.drift") + assert [(t["table"], t["published"], t["current"], t["impact"]) for t in drift["tables"]] == [ + ("parks", 0, 1, "possible")] + + +async def test_team_members_with_the_query_tool_count_as_possible(client, source): + wf = await _publish(client, source, team(source.name)) + assert await _recorded(wf) == {"direct": {}, "possible": {source.name: await _catalogued(source.id)}} + + +async def test_direct_tables_are_not_repeated_as_possible(client, source): + """同一个源既有写死的查询、又有 Agent：写死的 SQL 用到的表按直接引用记，可能涉及里不再记一遍。""" + wf = await _publish(client, source, mixed(source.name)) + recorded = await _recorded(wf) + assert recorded["direct"] == {source.name: {"parks": 0, "visits": 1}} + rest = {t: v for t, v in (await _catalogued(source.id)).items() if t not in ("parks", "visits")} + assert recorded["possible"] == {source.name: rest} and rest + now = await _edit_visits(source.id) + async with SessionLocal() as session: + await catalog.write_entry(session, source.id, "parks", {"label": catalog.make_item("景区", "human")}, + if_version=0, actor="王敏") + run_id = await _formal(client, wf, sealed=False) + [drift] = await _events(run_id, "catalog.drift") + # 两张都是直接引用：parks 新建了目录也不会再按可能涉及报一遍 + assert [(t["table"], t["published"], t["current"], t["impact"]) for t in drift["tables"]] == [ + ("parks", 0, 1, "direct"), ("visits", 1, now, "direct")] + + +async def test_direct_only_records_from_before_still_warn_as_direct(client, source): + """可能涉及上线之前发布的版本只记了直接引用（{源名: {表名: 版本}}）：照旧按直接引用比，不补可能涉及。""" + wf = await _publish(client, source, mixed(source.name)) + async with SessionLocal() as session: + snap = (await session.execute(select(WorkflowVersion).where( + WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one() + snap.catalog_versions = {source.name: {"parks": 0, "visits": 1}} + await session.commit() + detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() + assert detail["catalog_versions"] == {"direct": {source.name: {"parks": 0, "visits": 1}}, "possible": {}} + now = await _edit_visits(source.id) + async with SessionLocal() as session: + await catalog.write_entry(session, source.id, "gates", {"label": catalog.make_item("闸机", "human")}, + if_version=0, actor="王敏") + run_id = await _formal(client, wf, sealed=False) + [drift] = await _events(run_id, "catalog.drift") + assert [(t["table"], t["impact"]) for t in drift["tables"]] == [("visits", "direct")] + assert drift["tables"][0]["current"] == now + + async def test_no_change_no_warning(client, source): wf = await _publish(client, source) r = await client.post("/api/runs", json={"workflow_id": wf, "run_class": "formal"}) From c12f0f2852ad1a371d7bda58e8626cc5898f5020 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:07:41 +0800 Subject: [PATCH 2/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20=E4=B8=80=E8=87=B4=E6=80=A7=EF=BC=9A=E5=8A=A9=E6=89=8B?= =?UTF-8?q?=E4=B8=8A=E4=B8=8B=E6=96=87=E9=87=8C=E3=80=8C=E7=8E=B0=E6=9C=89?= =?UTF-8?q?=E5=B7=A5=E4=BD=9C=E6=B5=81=E7=94=A8=E5=88=B0=E7=9A=84=E8=A1=A8?= =?UTF-8?q?=E3=80=8D=E6=94=B9=E7=94=A8=20tool=5Fqueries=20=E8=AE=A4?= =?UTF-8?q?=E8=8A=82=E7=82=B9=EF=BC=88=E4=BB=A5=20sqlcheck.tool=5Fquery=20?= =?UTF-8?q?=E4=B8=BA=E5=87=86=EF=BC=8C=E6=8D=A2=E4=BA=86=E7=B1=BB=E5=9E=8B?= =?UTF-8?q?=E7=9A=84=E8=8A=82=E7=82=B9=E6=AE=8B=E7=95=99=E7=9A=84=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E4=B8=8D=E5=86=8D=E7=AE=97=EF=BC=89=EF=BC=9B=E6=AD=A3?= =?UTF-8?q?=E5=BC=8F=E8=BF=90=E8=A1=8C=E7=9A=84=E7=9B=AE=E5=BD=95=E5=8F=98?= =?UTF-8?q?=E5=8C=96=E4=BA=8B=E4=BB=B6=E6=94=B9=E7=94=A8=20EventType.CATAL?= =?UTF-8?q?OG=5FDRIFT=EF=BC=9B=E5=86=99=E6=98=8E=E5=85=B3=E7=B3=BB?= =?UTF-8?q?=E7=BC=96=E5=8F=B7=E5=8C=BA=E5=88=86=E6=96=B9=E5=90=91=E3=80=81?= =?UTF-8?q?=E6=9C=89=E6=84=8F=E4=B8=8D=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/copilot_context.py | 37 ++++++++++++++-------------- backend/app/api/runs.py | 3 ++- backend/tests/test_catalog_impact.py | 22 +++++++++++++++++ 3 files changed, 42 insertions(+), 20 deletions(-) diff --git a/backend/app/api/copilot_context.py b/backend/app/api/copilot_context.py index d2893c1f..86dc1205 100644 --- a/backend/app/api/copilot_context.py +++ b/backend/app/api/copilot_context.py @@ -212,35 +212,28 @@ def _modes(sources: list[Any]) -> list[Literal["all", "pick"]]: def graph_tables(graph: Mapping[str, Any] | None, sources: Iterable[Any]) -> dict[str, list[str]]: """现有工作流里查库节点的 SQL 用到的表：{源名: [schema_cache 的键]}，按出现顺序。 - 只认调用工具节点上写死的 SQL（db_query__<源>）；Agent 运行时自己写的 SQL 图里看不到。节点认两种写法： - 画布的 {data: {config}} 和 GraphSpec 的 {config}。SQL 解析不了的跳过。 + 只认调用工具节点上写死的 SQL（db_query__<源>）；Agent 运行时自己写的 SQL 图里看不到。哪些节点算写死的查询 + 以 sqlcheck.tool_query 为准（catalog_impact.tool_queries，和发布记录、SQL 检查同一份）：换成别的类型的节点 + 配置里残留的 tool、args 不算。节点认两种写法：画布的 {data: {config}} 和 GraphSpec 的 {config}。SQL 解析 + 不了的跳过。 """ + from app.data.catalog_impact import tool_queries from app.engine.evidence import sql_tables - from app.tools.datasource import QUERY_PREFIX by_name = {getattr(s, "name", None): s for s in sources} out: dict[str, list[str]] = {} - for node in (graph or {}).get("nodes") or []: - if not isinstance(node, dict): - continue - config = node.get("config") - if not isinstance(config, dict): - config = (node.get("data") or {}).get("config") if isinstance(node.get("data"), dict) else None - if not isinstance(config, dict): - continue - tool = str(config.get("tool") or "") - source = by_name.get(tool[len(QUERY_PREFIX):]) if tool.startswith(QUERY_PREFIX) else None - args = config.get("args") - sql = args.get("sql") if isinstance(args, dict) else None - if source is None or not isinstance(sql, str): + nodes = (graph or {}).get("nodes") or [] + for _, source_name, sql in tool_queries(n for n in nodes if isinstance(n, dict)): + source = by_name.get(source_name) + if source is None: continue try: names = sql_tables(sql) except Exception: # noqa: BLE001 - 认不出表就当没用到，只少一个提示 continue found = out.setdefault(source.name, []) - for name in names: - key = catalog.resolve_table_name(source.schema_cache, name) + for table in names: + key = catalog.resolve_table_name(source.schema_cache, table) if key and key not in found: found.append(key) return {k: v for k, v in out.items() if v} @@ -604,7 +597,13 @@ def _edge_line(edge: JoinEdge, qualified) -> str: def _edges_within(graph: Mapping[str, list[JoinEdge]], tables: list[str]) -> list[JoinEdge]: - """两端都在 tables 里的关系，每条一次；一对多的翻过来，从「多」的一端写起（外键所在的表在左）。""" + """两端都在 tables 里的关系，每条一次；一对多的翻过来，从「多」的一端写起（外键所在的表在左）。 + + 「每条一次」按 relation_id 去重，去掉的是同一条关系在关系图里的正反两条边。注意 catalog.relation_id 区分方向： + A 表目录里记的 A→B 和 B 表目录里记的 B→A 是两个编号，两边都记了的话这里会各出一行（内容相同、写法相反）。 + 这是有意不改的：编号存在每条关系上，也是审阅路径 relations.<编号> 的一部分，改成不分方向会让已有目录里的 + 编号全部对不上。 + """ wanted = set(tables) seen: set[str] = set() out = [] diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index fb424fe1..d61bd8a0 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -17,6 +17,7 @@ from sqlalchemy import select as _select from app.api.coded import CodedHTTPException from app.core.bus import bus from app.core import artifact_store +from app.core.events import EventType from app.data.catalog_impact import catalog_drift from app.db.base import SessionLocal, get_session from app.db.models import Approval, Artifact, Run, RunEvent, Workflow, WorkflowVersion @@ -188,7 +189,7 @@ async def start_run( if catalog_changes: # 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。 # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的（界面分开说） - await run_manager.note(run.id, "catalog.drift", version=version, count=len(catalog_changes), + await run_manager.note(run.id, EventType.CATALOG_DRIFT, version=version, count=len(catalog_changes), tables=catalog_changes) # 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的" for ev in upgrade_events: diff --git a/backend/tests/test_catalog_impact.py b/backend/tests/test_catalog_impact.py index a25d23f3..bf932094 100644 --- a/backend/tests/test_catalog_impact.py +++ b/backend/tests/test_catalog_impact.py @@ -183,3 +183,25 @@ def test_graph_sql_tables_uses_the_evidence_rules(): )) tables = {"shop": {"orders": {}, "order_items": {}, "customers": {}}, "other": {"orders": {}}} assert graph_sql_tables(spec, tables) == {"shop": {"orders", "order_items", "customers"}, "other": {"orders"}} + + +def test_assistant_seeds_read_tool_nodes_the_same_way(): + """助手给全字段的「现有工作流用到的表」和发布记录、SQL 检查认的是同一种节点（sqlcheck.tool_query）。 + + 画布上把调用工具节点换成别的类型，配置里的 tool、args 还留着：那已经不是一条会执行的查询，不能当成用到的表。 + """ + from app.api.copilot_context import graph_tables + + shop = SimpleNamespace(name="shop", schema_cache={"tables": { + "orders": {"qualified": "orders"}, "customers": {"qualified": "customers"}, "stores": {"qualified": "stores"}}}) + graph = _graph( + _query("q1", "shop", "SELECT * FROM customers c JOIN orders o ON o.customer_id = c.id"), + # 换成了 Agent 的节点：残留的写死查询不算 + _node("ask", "agent", tool="db_query__shop", args={"sql": "SELECT * FROM stores"}, tools=["db_query__shop"]), + # GraphSpec 的写法（config 在节点上）照样认 + {"id": "q2", "type": "tool", "config": {"tool": "db_query__shop", "args": {"sql": "SELECT * FROM stores"}}}, + _query("q3", "shop", " "), + ) + assert graph_tables(graph, [shop]) == {"shop": ["customers", "orders", "stores"]} + graph["nodes"] = [n for n in graph["nodes"] if n["id"] != "q2"] + assert graph_tables(graph, [shop]) == {"shop": ["customers", "orders"]} From 4c7cfa62b2ea698b1ecd00f1d6e355b9ea10ea73 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:08:42 +0800 Subject: [PATCH 3/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20=E4=B8=80=E8=87=B4=E6=80=A7=EF=BC=9A=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E7=9A=84=E6=88=AA=E6=96=AD=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E6=8A=BD=E6=88=90=20data/engine.py=20=E7=9A=84=20RowCollector?= =?UTF-8?q?=EF=BC=8C=E6=9F=A5=E8=AF=A2=E5=B1=82=E5=92=8C=E5=90=88=E5=B9=B6?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E9=83=BD=E8=B0=83=E5=AE=83=EF=BC=8C=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E5=90=84=E5=86=99=E4=B8=80=E9=81=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/engine.py | 44 +++++++++++++++++++++--------- backend/app/engine/merge_query.py | 19 ++++--------- backend/tests/test_merge_engine.py | 33 ++++++++++++++++++++++ 3 files changed, 70 insertions(+), 26 deletions(-) diff --git a/backend/app/data/engine.py b/backend/app/data/engine.py index b419f2a1..0f4cbf3d 100644 --- a/backend/app/data/engine.py +++ b/backend/app/data/engine.py @@ -879,6 +879,33 @@ async def _bounded( raise +class RowCollector: + """逐行收结果、边收边判上限（行数 max_rows、字节 max_bytes）。截断的判定只有这一份：查询层（run_query）和 + 合并查询（engine/merge_query._read）都调它，不各写一遍。 + + 到了上限之后确实还有下一行才记 truncated。以前一到上限就标截断，恰好 1000 行的完整结果也算截断——截断的结果 + 整组拿去算的指标要降档出具、截断的合并结果下游会拒收，误报就是把完整的结果降了档、拒了。字节数按收下的行累加， + 第一行总是收下：单行就超过字节上限也交回这一行。 + """ + + def __init__(self, limits: QueryLimits) -> None: + self.limits = limits + self.rows: list[list[Any]] = [] + self.size = 0 + self.truncated = False + self._full = False + + def take(self, values: list[Any]) -> bool: + """收一行（已转成能进 JSON 的值）。已经取满还来一行：记截断、不收，返回 False，调用方就此停止读取。""" + if self._full: + self.truncated = True + return False + self.rows.append(values) + self.size += len(json.dumps(values, ensure_ascii=False, default=str)) + self._full = len(self.rows) >= self.limits.max_rows or self.size >= self.limits.max_bytes + return True + + async def run_query( source: Any, sql: str, *, limits: QueryLimits | None = None ) -> QueryResult: @@ -903,23 +930,14 @@ async def run_query( async def _read(conn: AsyncConnection) -> tuple[list[str], list[list[Any]], bool, dict[str, str]]: cursor = await conn.stream(text(statement)) columns = list(cursor.keys()) - rows: list[list[Any]] = [] + collected = RowCollector(limits) seen = _TypeTally(len(columns)) - truncated = full = False - size = 0 + # 逐行累加，边收边判上限：一次性 fetchall 一个亿级表就晚了 async for row in cursor: - if full: - # 到了上限之后还有下一行：确实没取完。以前一到上限就标截断，恰好 1000 行的完整结果 - # 也算截断——截断的结果整组拿去算的指标要降档出具，误报就是把完整的结果降了档 - truncated = True + if not collected.take([_jsonable(v) for v in row]): break - # 逐行累加，边收边判上限：一次性 fetchall 一个亿级表就晚了 seen.add(row) - values = [_jsonable(v) for v in row] - rows.append(values) - size += len(json.dumps(values, ensure_ascii=False, default=str)) - full = len(rows) >= limits.max_rows or size >= limits.max_bytes - return columns, rows, truncated, seen.types(columns) + return columns, collected.rows, collected.truncated, seen.types(columns) columns, rows, truncated, types = await _bounded( engine.connect, kind, limits, diff --git a/backend/app/engine/merge_query.py b/backend/app/engine/merge_query.py index 52a54d59..87d5d6c7 100644 --- a/backend/app/engine/merge_query.py +++ b/backend/app/engine/merge_query.py @@ -40,7 +40,7 @@ from dataclasses import dataclass, field from typing import Any from app.data import guard -from app.data.engine import _jsonable, column_types, open_memory_sqlite +from app.data.engine import RowCollector, _jsonable, column_types, open_memory_sqlite from app.data.guard import QueryLimits, SqlRejected from app.data.names import name_key from app.engine.direct_select import _kw, _lex, _Lex @@ -343,24 +343,17 @@ def _authorize(action: int, *_args: Any) -> int: def _read(conn: sqlite3.Connection, stmt: str, limits: QueryLimits) -> tuple[list[str], list[list[Any]], bool]: - """逐行读、边读边判上限：和查询层（data.engine.run_query）同一个口径——到了行数或字节上限之后 - 确实还有下一行才记 truncated，恰好取满不算（截断的合并结果下游会拒收，误报就是拒了完整的结果）。""" + """逐行读、边读边判上限。判定调查询层的 RowCollector（data/engine.py），和 run_query 是同一份：到了行数或 + 字节上限之后确实还有下一行才记 truncated，恰好取满不算（截断的合并结果下游会拒收，误报就是拒了完整的结果）。""" cursor = conn.execute(stmt) columns = [str(d[0]) for d in cursor.description or []] if not columns: raise MergeError("合并 SQL 没有返回任何列") - rows: list[list[Any]] = [] - size = 0 - truncated = full = False + collected = RowCollector(limits) for row in cursor: - if full: - truncated = True + if not collected.take([_jsonable(v) for v in row]): break - values = [_jsonable(v) for v in row] - rows.append(values) - size += len(json.dumps(values, ensure_ascii=False, default=str)) - full = len(rows) >= limits.max_rows or size >= limits.max_bytes - return columns, rows, truncated + return columns, collected.rows, collected.truncated def _sqlite_reason(e: sqlite3.Error, limits: QueryLimits) -> str: diff --git a/backend/tests/test_merge_engine.py b/backend/tests/test_merge_engine.py index 9acf20c9..65306842 100644 --- a/backend/tests/test_merge_engine.py +++ b/backend/tests/test_merge_engine.py @@ -225,6 +225,39 @@ def test_byte_limit_matches_the_query_layer(): assert out.truncated is True and 0 < len(out.rows) < 16 +def test_row_collector_is_the_one_cutoff_rule(): + """截断判定只有一份（data.engine.RowCollector）：到了行数或字节上限之后确实还有下一行才算截断，恰好取满不算； + 第一行总是收下（单行就超过字节上限也得交回点什么）。""" + from app.data.engine import RowCollector + + full = RowCollector(QueryLimits(max_rows=2)) + assert full.take([1]) and full.take([2]) + assert full.rows == [[1], [2]] and full.truncated is False # 恰好取满、后面没有了 + assert full.take([3]) is False # 取满之后还有一行：截断，这一行不收 + assert full.rows == [[1], [2]] and full.truncated is True + + tiny = RowCollector(QueryLimits(max_bytes=5)) + assert tiny.take(["一行就超过字节上限"]) is True and tiny.take(["下一行"]) is False + assert len(tiny.rows) == 1 and tiny.truncated is True + + +def test_merge_reads_through_the_query_layer_cutoff(monkeypatch): + """合并查询调用查询层的判定，不另写一遍：两边的口径不会再各改各的。""" + import app.engine.merge_query as merge_query + from app.data.engine import RowCollector + + made: list[RowCollector] = [] + + class Spy(RowCollector): + def __init__(self, limits): + super().__init__(limits) + made.append(self) + + monkeypatch.setattr(merge_query, "RowCollector", Spy) + out = execute([sales(), visits()], "SELECT * FROM s CROSS JOIN v", limits=QueryLimits(max_rows=5)) + assert len(made) == 1 and made[0].truncated is True and len(out.rows) == len(made[0].rows) == 5 + + def test_runaway_query_is_stopped(): endless = "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c) SELECT count(*) FROM c" with pytest.raises(MergeError, match="秒"): From 6c30becd94237e4d65787676412234a9ccdbfd07 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:09:12 +0800 Subject: [PATCH 4/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20/=20B8=EF=BC=9A=E4=B8=80=E9=94=AE=E5=8D=87=E7=BA=A7=E7=9A=84?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=E5=B1=82=EF=BC=88assist=5Fupgrade=EF=BC=89?= =?UTF-8?q?=E8=87=AA=E6=9F=A5=E3=80=81=E4=BF=AE=E6=AD=A3=E8=BD=AE=E5=92=8C?= =?UTF-8?q?=E6=9C=80=E7=BB=88=E5=88=A4=E6=96=AD=E9=83=BD=E5=B8=A6=E4=B8=8A?= =?UTF-8?q?=E5=9F=BA=E4=BA=8E=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E7=9A=84?= =?UTF-8?q?=20SQL=20=E6=A3=80=E6=9F=A5=E5=99=A8=EF=BC=8C=E5=92=8C=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E9=97=A8=E7=A6=81=E4=B8=80=E4=B8=AA=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=8A=A9=E6=89=8B=E6=94=B9=E5=87=BA=E4=B8=80?= =?UTF-8?q?=E5=AF=B9=E5=A4=9A=E5=85=B3=E8=81=94=E5=90=8E=E6=B1=82=E5=92=8C?= =?UTF-8?q?=E8=BF=99=E7=B1=BB=20SQL=20=E4=BC=9A=E4=BA=A4=E5=9B=9E=E5=8E=BB?= =?UTF-8?q?=E6=94=B9=EF=BC=8C=E6=94=B9=E4=B8=8D=E5=A5=BD=E5=B0=B1=E4=BD=9C?= =?UTF-8?q?=E5=BA=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/copilot.py | 12 ++++-- backend/tests/test_copilot_sqlcheck.py | 51 ++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 7ef959d0..f253d126 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -2251,13 +2251,18 @@ async def assist_upgrade( if not feeders: out["summary"] = "口径卡的输入均不来自负责计算的沙箱代码，无需交给助手改写" return out + from app.data.sqlcheck import load_checkers + sources = await _sources(session, None) + # 基于数据目录的 SQL 检查器：自查和最终判断都带上，和发布门禁一个口径（受管级别下一对多关联后求和这类 error + # 会挡发布）。不带的话，助手改出这种 SQL 照样被采纳，交出去的图要到发布时才被拦 + checkers = await load_checkers(session, sources) 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()) defaults = base.get("defaults") base_errors = _blocking_issues(before_nodes, before_edges, sources=sources, baseline=baseline, level=level, - defaults=defaults) or [] + defaults=defaults, checkers=checkers) or [] known = {_sig(e) for e in base_errors} out["warnings"] = _kept_code(spec) request = GenerateIn(instruction="升级为可追溯结构", provider=provider, model=model) @@ -2309,7 +2314,8 @@ async def assist_upgrade( await collect(_upgrade_assist_request(base, spec, feeders), first=True) # 自查：和搭图同一套规则，只把这一轮冒出来的 error 交回去改（原来就有的不归这一轮管） for _ in range(_SELF_CHECK_ROUNDS): - now = _blocking_issues(nodes, edges, sources=sources, baseline=baseline, level=level, defaults=defaults) + now = _blocking_issues(nodes, edges, sources=sources, baseline=baseline, level=level, defaults=defaults, + checkers=checkers) fresh = [e for e in now or [] if _sig(e) not in known] if not effective or not fresh: break @@ -2326,7 +2332,7 @@ async def assist_upgrade( 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: diff --git a/backend/tests/test_copilot_sqlcheck.py b/backend/tests/test_copilot_sqlcheck.py index efdd24df..2a6cb072 100644 --- a/backend/tests/test_copilot_sqlcheck.py +++ b/backend/tests/test_copilot_sqlcheck.py @@ -241,3 +241,54 @@ async def test_clean_sql_costs_no_extra_call(scenic_source, monkeypatch): final = events[-1] assert final["autorun"] is True assert not [i for i in final["issues"] if i.get("code") in ("fanout_sum", "missing_valid_filter")] + + +# -------------------------------------------------------------------------- +# 一键升级的语义层（assist_upgrade）：自查和最终判断同样带 SQL 检查器，和发布门禁一个口径 +# -------------------------------------------------------------------------- + + +def _upgrade_graph(source: str) -> dict: + """查库 → 沙箱代码算比率 → 口径卡读代码的产出 → 报告撰写 → 出口。查库的 SQL 本身没问题。""" + def n(nid, ntype, x, **config): + return {"id": nid, "type": ntype, "position": {"x": x, "y": 80}, "data": {"label": nid, "config": config}} + + nodes = [ + n("start", "input", 0), + n("fetch", "tool", 280, tool=f"db_query__{source}", args={"sql": CLEAN_SQL}), + n("calc", "code", 560, language="python", assign_to="calc", code="import json\nprint(json.dumps({'ratio': 3 / 4}))"), + n("card", "metrics", 840, metrics=[{"id": "ratio", "name": "客单价", "expression": "vars.calc.ratio"}]), + n("write", "report", 1120, instructions="写一句话"), + n("out", "output", 1400, fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ] + return {"nodes": nodes, "edges": [{"id": f"e{i}", "source": a["id"], "target": b["id"]} + for i, (a, b) in enumerate(zip(nodes, nodes[1:]))]} + + +async def test_upgrade_assist_checks_sql_like_the_publish_gate(scenic_source, monkeypatch): + """助手改写口径卡时顺手把查询改成了一对多关联后求和：受管级别的发布门禁会挡，升级的自查也得交回去改， + 改不好就作废——以前自查和最终判断都没带 SQL 检查器，这样的改写会被采纳，交出去的图发布时才被拦。""" + import app.api.copilot as copilot + from app.db.base import SessionLocal + + first = [{"op": "plan", "summary": "把纯算术的代码挪进口径卡"}, + {"op": "update_node", "id": "card", "config": {"metrics": [ + {"id": "ratio", "name": "客单价", "expression": "cell(nodes.fetch, 0, 'gmv') / 4"}]}}, + {"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL}}}, + {"op": "done", "explanation": "客单价改由口径卡直接算"}] + fix = [{"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL + " "}}}, + {"op": "done", "explanation": "改了"}] + model = _Scripted(first, fix) + + async def _model(*_a, **_k): + return model, "mock-fast" + + monkeypatch.setattr(copilot, "get_chat_model", _model) + async with SessionLocal() as session: + out = await copilot.assist_upgrade(session, _upgrade_graph(scenic_source), level="governed") + # 自查把一对多求和交回模型，带着给模型的改法 + assert len(model.calls) >= 2 and "上线前自查" in model.calls[1] + assert "orders 关联 order_items 是一对多" in model.calls[1] + # 改不好：整个作废，原因写的是这条 SQL 检查 + assert out["accepted"] is False and out["ok"] is False + assert "一对多" in (out["reason"] or ""), out["reason"] From 52aaaaf94500edcae853f24b0211683c759b92b9 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:11:03 +0800 Subject: [PATCH 5/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20/=20B4=20=E6=9C=8D=E5=8A=A1=E7=AB=AF=EF=BC=9A=E6=B2=A1?= =?UTF-8?q?=E6=8E=A2=E6=9F=A5=E5=88=B0=E7=9A=84=E8=A1=A8=E8=AF=B4=E5=87=BA?= =?UTF-8?q?=E6=9D=A5=E2=80=94=E2=80=94=E7=9B=AE=E5=BD=95=E6=B8=85=E5=8D=95?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=20schema=5Ftruncated=E3=80=81schema=5Ftotal?= =?UTF-8?q?=EF=BC=9B=E5=8A=A9=E6=89=8B=E4=B8=8A=E4=B8=8B=E6=96=87=E7=9A=84?= =?UTF-8?q?=20total=20=E7=94=A8=E8=A1=A8=E7=BB=93=E6=9E=84=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E9=87=8C=E7=9A=84=E7=9C=9F=E5=AE=9E=E6=80=BB=E6=95=B0?= =?UTF-8?q?=EF=BC=8C=E6=88=AA=E6=96=AD=E6=97=B6=20context=20=E6=93=8D?= =?UTF-8?q?=E4=BD=9C=E5=B8=A6=20explored=EF=BC=8C=E6=8F=90=E7=A4=BA?= =?UTF-8?q?=E8=AF=8D=E5=86=99=E6=98=8E=E8=BF=98=E6=9C=89=E5=87=A0=E5=BC=A0?= =?UTF-8?q?=E6=B2=A1=E6=8E=A2=E6=9F=A5=E5=88=B0=E3=80=81=E4=B8=8D=E8=AE=A4?= =?UTF-8?q?=E8=AF=86=E7=9A=84=E8=A1=A8=E5=90=8D=E5=85=88=E6=9F=A5=E7=BB=93?= =?UTF-8?q?=E6=9E=84=EF=BC=8C=E6=8C=91=E8=A1=A8=E8=AF=B7=E6=B1=82=E4=B9=9F?= =?UTF-8?q?=E5=86=99=E6=98=8E=E5=8F=AA=E5=88=97=E4=BA=86=E6=8E=A2=E6=9F=A5?= =?UTF-8?q?=E5=88=B0=E7=9A=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/catalog.py | 13 +- backend/app/api/copilot_context.py | 53 +++++++- .../tests/test_catalog_schema_truncated.py | 116 ++++++++++++++++++ 3 files changed, 176 insertions(+), 6 deletions(-) create mode 100644 backend/tests/test_catalog_schema_truncated.py diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 75159525..e85c63a0 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -87,6 +87,10 @@ def _iso(value: Any) -> str | None: return value.isoformat() if value is not None else None +def _int(value: Any) -> int: + return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else 0 + + def _item_value(notes: dict[str, Any], name: str) -> tuple[Any, str | None]: item = notes.get(name) if not isinstance(item, dict) or item.get("status") == "rejected": @@ -169,7 +173,12 @@ def _invalid(e: catalog.CatalogInvalid) -> HTTPException: @router.get("/{source_id}/catalog") async def list_catalog(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: - """表清单：表结构里的每张表各一行（没有目录的也列），使用次数多的在前；目录还在、表已不在的排最后。""" + """表清单：表结构里的每张表各一行（没有目录的也列），使用次数多的在前；目录还在、表已不在的排最后。 + + 探查结构每个源最多取 200 张表（introspect._MAX_TABLES），多出来的不在表结构里、这里也列不出：schema_truncated + 说明有没有截断，schema_total 是数据库里一共几张表（含视图），界面据此说明「只探查了前 200 张」。老缓存没记 + 总数时按探查到的张数算。 + """ _, source = await _resolved(session, source_id) cache = source.schema_cache or {} tables = cache.get("tables") or {} @@ -184,6 +193,8 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi "system_notes": catalog.system_notes_source(source), # 没有表结构时说明原因（尚未探查 / 探查失败），界面据此引导去探查 "schema_note": None if tables else introspect.why_empty(cache), + "schema_truncated": bool(cache.get("truncated")), + "schema_total": max(_int(cache.get("total")), len(tables)), } diff --git a/backend/app/api/copilot_context.py b/backend/app/api/copilot_context.py index 86dc1205..6c527ebc 100644 --- a/backend/app/api/copilot_context.py +++ b/backend/app/api/copilot_context.py @@ -78,6 +78,14 @@ SECTION_HEAD_PROMPT = "\n\n已接入的数据源（涉及取数时优先用它 NAMES_ONLY_PROMPT = " （对象较多，上面只列了名字。写 SQL 前用 db_schema 工具确认字段，别猜）" INFERRED_PROMPT = (" 标「推断，未确认」的说明和连接条件是按列名、数据库注释或模型推断出来的，没有人确认过：" "照抄前对照字段核实，拿不准就先用 db_schema 工具查") +#: 探查结构截断了的源（每个源最多取 200 张表，introspect._MAX_TABLES）：没探查到的表不在表结构缓存里，上面列不出、 +#: db_schema 也查不到。不说的话模型会以为库里就这么多表，用户说的表名对不上时就近找一张名字像的顶上 +UNEXPLORED_PROMPT = (" （这个库共 {total} 张表，只探查了 {explored} 张，还有 {rest} 张没有探查到、不在上面。用户提到的表名" + "在上面找不到时，先用 db_schema__{source} 查；也查不到就查数据字典（information_schema.columns，" + "SQLite 用 sqlite_master）确认它的字段，别猜，也别拿名字相近的表顶替）") +#: 挑表请求里一个源的标题：截断了的写明只列了探查到的 +PICK_SOURCE_PROMPT = "数据源「{source}」（共 {total} 张表）" +PICK_SOURCE_PARTIAL_PROMPT = "数据源「{source}」（共 {total} 张表，只探查了 {explored} 张，下面只列探查到的）" SQL_RULES_PROMPT = ( "\n\n写 SQL 的硬性要求：\n" # 这三条都是真实踩过的：Oracle 上漏 schema 前缀直接 ORA-00942， @@ -177,6 +185,26 @@ class SourcePlan: def tables(self) -> dict[str, Any]: return (getattr(self.source, "schema_cache", None) or {}).get("tables") or {} + @property + def explored(self) -> int: + """探查到、在表结构缓存里的表数。""" + return len(self.tables) + + @property + def total(self) -> int: + """这个源一共几张表（含视图）：探查时数据库报的总数。老缓存没记总数时按探查到的算。""" + raw = (getattr(self.source, "schema_cache", None) or {}).get("total") + return max(raw if isinstance(raw, int) and not isinstance(raw, bool) else 0, self.explored) + + @property + def unexplored(self) -> int: + """探查截断了的话还有几张没探查到；没截断为 0。 + + 只看 truncated：反射失败跳过的表（skipped）也算在 total 里，但那是「探查了、读不到」，不是没探查到。 + """ + truncated = (getattr(self.source, "schema_cache", None) or {}).get("truncated") + return max(self.total - self.explored, 0) if truncated else 0 + @property def notes(self) -> dict[str, dict[str, Any]]: """给模型看的目录：去掉驳回项。""" @@ -330,7 +358,9 @@ class ContextPlan: continue qualified = _qualifier(p.tables) index = catalog.render_table_index(p.source.schema_cache, p.notes) - blocks.append(f"数据源「{p.source.name}」（共 {len(p.tables)} 张表）：\n{index}") + head = (PICK_SOURCE_PARTIAL_PROMPT if p.unexplored else PICK_SOURCE_PROMPT).format( + source=p.source.name, total=p.total, explored=p.explored) + blocks.append(f"{head}：\n{index}") if p.seeds: used.append(f"数据源「{p.source.name}」的 " + "、".join(qualified(t) for t in p.seeds)) human = PICK_PROMPT.format(need=need.strip(), used=PICK_USED_PROMPT.format(tables="；".join(used)) if used @@ -499,7 +529,13 @@ class SourceContext: @property def total(self) -> int: - return len(self.plan.tables) + """这个源一共几张表：表结构缓存里记的真实总数（探查截断时比探查到的多），见 SourcePlan.total。""" + return self.plan.total + + @property + def explored(self) -> int: + """探查到的表数：没有表结构时为 0。""" + return self.plan.explored def qualified(self, key: str) -> str: return _qualifier(self.plan.tables)(key) @@ -525,15 +561,18 @@ class DatasourceContext: def op(self) -> dict[str, Any] | None: """告诉前端这一轮参考了哪些表：{"op": "context", "sources": [...], "elapsed_ms"?}。没有可查的表时为 None。 - 每个源一项：source 源名，tables 给了全字段的表（全名），selected_by，total 这个源一共几张表；挑表失败的 + 每个源一项：source 源名，tables 给了全字段的表（全名），selected_by，total 这个源一共几张表（表结构缓存里 + 记的真实总数）；探查截断了的再带 explored（探查到几张），界面写「共 N 张，只探查了 200 张」；挑表失败的 带上 reason。只列了名字的表不进 tables：界面上说的「参考了 N 张表」是模型看得到字段的那些。 """ rows = [] for s in self.sources: - if not s.total: + if not s.explored: continue row: dict[str, Any] = {"source": s.source.name, "tables": [s.qualified(t) for t in s.tables], "selected_by": s.selected_by, "total": s.total} + if s.plan.unexplored: + row["explored"] = s.explored if s.selected_by == "fallback" and self.reason: row["reason"] = self.reason rows.append(row) @@ -564,6 +603,10 @@ class DatasourceContext: logger.exception("数据源 %s 的上下文组装失败，本轮只列表名", s.source.name) blocks.append(introspect.summary(s.source, detail=False)) names_only = True + if s.plan.unexplored: + # 探查截断了：上面列的只是探查到的那部分，说清还有多少张没探查到、遇到不认识的表名怎么办 + blocks.append(UNEXPLORED_PROMPT.format(total=s.total, explored=s.explored, rest=s.plan.unexplored, + source=s.source.name)) text = "\n".join(blocks) tail = [] if names_only: @@ -635,7 +678,7 @@ def _detail_block(s: SourceContext, budget: list[int]) -> str: 目录按表放，放不下的表让模型用 db_schema 查（db_schema 的输出里带目录）。 """ lines = [introspect.summary(s.source, detail=True)] - if not s.total: + if not s.explored: return lines[0] joins = _join_lines(s, _relation_graph(s.plan), list(s.plan.tables)) size = sum(len(line) + 1 for line in joins) diff --git a/backend/tests/test_catalog_schema_truncated.py b/backend/tests/test_catalog_schema_truncated.py new file mode 100644 index 00000000..b16489d2 --- /dev/null +++ b/backend/tests/test_catalog_schema_truncated.py @@ -0,0 +1,116 @@ +"""没探查到的表要说出来（数据目录完整性审查 F3 / B4）。 + +探查结构每个数据源最多取 200 张表（introspect._MAX_TABLES）。多出来的表不在表结构缓存里：目录页列不出，助手也 +看不到字段。以前这件事哪儿都没说——目录清单只给探查到的 200 行，助手上下文里的「共几张表」是探查到的张数， +提示词里也不提。这组守的是： + +- 目录清单接口返回 schema_truncated、schema_total（数据库里一共几张），界面据此说明「只探查了前 200 张」 +- 助手上下文的 total 是表结构缓存里的真实总数；context 操作里带上 explored（探查到几张） +- 提示词里写明还有几张没探查到，遇到不认识的表名先查结构 +- 没截断的库照旧，不多说一句 + +合成库：205 张空表，测试里临时建，用完连同数据源一起删掉。 +""" +from __future__ import annotations + +import sqlite3 +import uuid + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.db.base import SessionLocal +from app.db.models import DataSource +from app.main import app +from tests.fixtures.sources import drop_source + +WIDE = 205 + + +def _build(path, count: int) -> str: + db = sqlite3.connect(path) + for i in range(count): + db.execute(f"CREATE TABLE t{i:03d} (id INTEGER PRIMARY KEY, amount INTEGER)") + db.commit() + db.close() + return str(path) + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def _source(client, path: str) -> tuple[str, str]: + name = f"wide_{uuid.uuid4().hex[:8]}" + r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": path, + "readonly": True}) + assert r.status_code in (200, 201), r.text + sid = r.json()["id"] + r = await client.post(f"/api/datasources/{sid}/introspect") + assert r.status_code == 200, r.text + return sid, name + + +@pytest.fixture +async def wide(client, tmp_path): + sid, name = await _source(client, _build(tmp_path / "wide.db", WIDE)) + yield sid, name + await drop_source(sid) + + +@pytest.fixture +async def narrow(client, tmp_path): + sid, name = await _source(client, _build(tmp_path / "narrow.db", 3)) + yield sid, name + await drop_source(sid) + + +async def _row(sid: str) -> DataSource: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + session.expunge(row) + return row + + +async def test_catalog_list_says_how_many_tables_were_not_introspected(client, wide): + sid, _ = wide + body = (await client.get(f"/api/datasources/{sid}/catalog")).json() + assert len(body["tables"]) == 200 + assert body["schema_truncated"] is True and body["schema_total"] == WIDE + + +async def test_catalog_list_of_a_complete_schema(client, narrow): + sid, _ = narrow + body = (await client.get(f"/api/datasources/{sid}/catalog")).json() + assert len(body["tables"]) == 3 + assert body["schema_truncated"] is False and body["schema_total"] == 3 + + +async def test_assistant_context_counts_the_real_total(wide): + from app.api import copilot_context + + sid, name = wide + plan = await copilot_context.plan_context([await _row(sid)]) + ctx = plan.fallback("测试") + (row,) = ctx.op()["sources"] + assert row["total"] == WIDE and row["explored"] == 200 + text = ctx.section() + assert f"共 {WIDE} 张表" in text and "只探查了 200 张" in text and f"还有 {WIDE - 200} 张没有探查到" in text + # 遇到不认识的表名先查结构，别猜 + assert f"db_schema__{name}" in text and "information_schema" in text + # 挑表的请求里也写明只列了探查到的 + pick = plan._pick_messages("各表的金额")[1][1] + assert f"共 {WIDE} 张表，只探查了 200 张" in pick + + +async def test_assistant_context_of_a_complete_schema_says_nothing_extra(narrow): + from app.api import copilot_context + + sid, _ = narrow + plan = await copilot_context.plan_context([await _row(sid)]) + ctx = await plan.resolve(None, need="x") + (row,) = ctx.op()["sources"] + assert row["total"] == 3 and "explored" not in row + assert "没有探查到" not in ctx.section() From 3ece89872bc5bf665ffa93b139c7f84bd8e9e3a2 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:24:50 +0800 Subject: [PATCH 6/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20/=20B1=20=E7=95=8C=E9=9D=A2=EF=BC=9A=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E6=A8=AA=E5=B9=85=E3=80=81=E6=97=B6=E9=97=B4=E7=BA=BF=E5=92=8C?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E5=8E=86=E5=8F=B2=E6=8A=8A=20Agent=20?= =?UTF-8?q?=E5=8F=AF=E8=83=BD=E6=9F=A5=E8=AF=A2=E7=9A=84=E8=A1=A8=E5=92=8C?= =?UTF-8?q?=E7=9B=B4=E6=8E=A5=E5=BC=95=E7=94=A8=E5=88=86=E5=BC=80=E8=AF=B4?= =?UTF-8?q?=E2=80=94=E2=80=94=E7=9B=B4=E6=8E=A5=E5=BC=95=E7=94=A8=E5=86=99?= =?UTF-8?q?=E3=80=8C=E6=9C=89=E5=8F=98=E5=8C=96=E3=80=8D=EF=BC=8C=E5=8F=AF?= =?UTF-8?q?=E8=83=BD=E6=B6=89=E5=8F=8A=E5=86=99=E3=80=8CAgent=20=E5=8F=AF?= =?UTF-8?q?=E8=83=BD=E6=9F=A5=E8=AF=A2=E7=9A=84=E8=A1=A8=E6=9C=89=E5=8F=98?= =?UTF-8?q?=E5=8C=96=E3=80=8D=E5=B9=B6=E5=8F=A6=E8=B5=B7=E4=B8=80=E7=BB=84?= =?UTF-8?q?=EF=BC=9B=E7=89=88=E6=9C=AC=E5=8E=86=E5=8F=B2=E6=8C=89=E6=96=B0?= =?UTF-8?q?=E7=9A=84=E5=8F=91=E5=B8=83=E8=AE=B0=E5=BD=95=E4=B8=A4=E7=BB=84?= =?UTF-8?q?=E5=88=97=E5=87=BA=EF=BC=8C=E5=8F=91=E5=B8=83=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E6=89=8D=E5=BB=BA=E7=9B=AE=E5=BD=95=E7=9A=84=E8=A1=A8=E4=B9=9F?= =?UTF-8?q?=E5=88=97=EF=BC=9B=E8=A1=A5=E5=8F=91=E5=B8=83=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E4=B8=A4=E7=A7=8D=E5=86=99=E6=B3=95=E7=9A=84=E5=8D=95=E6=B5=8B?= =?UTF-8?q?=E5=92=8C=E4=B8=89=E4=B8=AA=E6=A3=80=E6=9F=A5=E8=84=9A=E6=9C=AC?= =?UTF-8?q?=E7=9A=84=E6=96=AD=E8=A8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/test_catalog_impact.py | 15 +++++ frontend/src/canvas/VersionsSheet.tsx | 83 ++++++++++++++++++--------- frontend/src/lib/terms.ts | 18 +++++- frontend/src/pages/runs/Banners.tsx | 57 +++++++++++------- frontend/src/run/decode.ts | 36 +++++++++--- frontend/src/types.ts | 17 +++++- scripts/check-decode.mjs | 20 +++++++ scripts/check-runs.mjs | 19 ++++-- scripts/check-studio.mjs | 26 +++++++-- 9 files changed, 222 insertions(+), 69 deletions(-) diff --git a/backend/tests/test_catalog_impact.py b/backend/tests/test_catalog_impact.py index bf932094..128c1e20 100644 --- a/backend/tests/test_catalog_impact.py +++ b/backend/tests/test_catalog_impact.py @@ -205,3 +205,18 @@ def test_assistant_seeds_read_tool_nodes_the_same_way(): assert graph_tables(graph, [shop]) == {"shop": ["customers", "orders", "stores"]} graph["nodes"] = [n for n in graph["nodes"] if n["id"] != "q2"] assert graph_tables(graph, [shop]) == {"shop": ["customers", "orders"]} + + +def test_recorded_versions_reads_both_shapes(): + """发布记录两种写法：现在的 {direct, possible}，和可能涉及上线之前只记直接引用的 {源名: {表名: 版本}}。 + 源名可以就叫 direct / possible，靠第二层是不是版本号分开。""" + from app.data.catalog_impact import recorded_versions + + now = {"direct": {"shop": {"orders": 2}}, "possible": {"shop": {"customers": 1}, "crm": {}}} + assert recorded_versions(now) == now + assert recorded_versions({"direct": {}, "possible": {}}) == {"direct": {}, "possible": {}} + assert recorded_versions({"shop": {"orders": 2}}) == {"direct": {"shop": {"orders": 2}}, "possible": {}} + # 旧写法里一个叫 direct 的源 + assert recorded_versions({"direct": {"orders": 2}}) == {"direct": {"direct": {"orders": 2}}, "possible": {}} + assert recorded_versions({}) == {"direct": {}, "possible": {}} + assert recorded_versions(None) is None and recorded_versions("x") is None and recorded_versions({"shop": 3}) is None diff --git a/frontend/src/canvas/VersionsSheet.tsx b/frontend/src/canvas/VersionsSheet.tsx index 5ea271f5..7b047a97 100644 --- a/frontend/src/canvas/VersionsSheet.tsx +++ b/frontend/src/canvas/VersionsSheet.tsx @@ -10,7 +10,7 @@ import { formatDateTime, formatTime } from '../lib/format' import { CATALOG_DRIFT_TEXT, WORKFLOW_STATUS_LABEL } from '../lib/terms' import { hintOf } from './shortcuts' import type { Edge } from '@xyflow/react' -import type { CatalogDriftTable, GraphSpec, Workflow, WorkflowVersion } from '../types' +import type { CatalogDriftTable, CatalogVersions as CatalogVersionsRecord, GraphSpec, Workflow, WorkflowVersion } from '../types' /** * 版本历史。盖在右栏上的一层，和属性面板同一种「临时造访」。 @@ -212,38 +212,65 @@ function VersionPreview({ version, target, diff, lock, onRestore }: { const DIFF_COLOR = { add: 'var(--ok)', del: 'var(--err)', mod: 'var(--warn)' } as const /** - * 发布时的目录版本：这一版 SQL 用到的表在发布时是数据目录的第几版，之后改过的标出来——下次从这一版发起正式 - * 运行时会提醒「数据目录有变化」，这里先让人看到是哪几张表 + * 发布时的目录版本：这一版用到的表在发布时是数据目录的第几版，之后改过的标出来——下次从这一版发起正式运行时会 + * 提醒「数据目录有变化」，这里先让人看到是哪几张表。 + * + * 两组分开列：SQL 里写着的表在前；Agent 可能查询的表（Agent 绑定了数据源的查询工具，发布时记下数据源中所有有 + * 目录的表）在小标题下面。可能涉及的数据源里发布之后新建了目录的表，记录里没有它，按变化里给的补进来（发布时 + * 「尚无目录」） */ -function CatalogVersions({ versions, changes }: { - versions: Record> - changes: CatalogDriftTable[] -}) { - const rows = Object.entries(versions).flatMap(([source, tables]) => - Object.entries(tables).map(([table, v]) => ({ source, table, v }))) - const changed = new Map(changes.map((c) => [`${c.source}/${c.table}`, c])) +function CatalogVersions({ versions, changes }: { versions: CatalogVersionsRecord; changes: CatalogDriftTable[] }) { + // 老后端给的变化没有 impact：按直接引用 + const impactOf = (c: CatalogDriftTable) => (c.impact === 'possible' ? 'possible' : 'direct') + const changed = new Map(changes.map((c) => [`${impactOf(c)}/${c.source}/${c.table}`, c])) + const rowsOf = (impact: CatalogDriftTable['impact']) => { + const recorded = Object.entries(versions[impact] ?? {}).flatMap(([source, tables]) => + Object.entries(tables).map(([table, v]) => ({ source, table, v: v as number | null }))) + const known = new Set(recorded.map((r) => `${r.source}/${r.table}`)) + const added = changes.filter((c) => impactOf(c) === impact && !known.has(`${c.source}/${c.table}`)) + .map((c) => ({ source: c.source, table: c.table, v: c.published })) + return [...recorded, ...added] + } + const direct = rowsOf('direct') + const possible = rowsOf('possible') + const bare = Object.entries(versions.possible ?? {}) + .filter(([source, tables]) => !Object.keys(tables).length && !possible.some((r) => r.source === source)) + .map(([source]) => source) + const list = (impact: CatalogDriftTable['impact'], rows: typeof direct) => ( +
    + {rows.map(({ source, table, v }) => { + const c = changed.get(`${impact}/${source}/${table}`) + return ( +
  • + {table} + {CATALOG_DRIFT_TEXT.version(v)} + {c && ( + + {CATALOG_DRIFT_TEXT.changedSince(CATALOG_DRIFT_TEXT.version(c.current))} + + )} +
    • + ) + })} +
+ ) + const empty = !direct.length && !possible.length && !bare.length return (
{CATALOG_DRIFT_TEXT.section}
- {!rows.length ?
{CATALOG_DRIFT_TEXT.empty}
: ( + {empty ?
{CATALOG_DRIFT_TEXT.empty}
: ( <> -
    - {rows.map(({ source, table, v }) => { - const c = changed.get(`${source}/${table}`) - return ( -
  • - {table} - {CATALOG_DRIFT_TEXT.version(v)} - {c && ( - - {CATALOG_DRIFT_TEXT.changedSince(CATALOG_DRIFT_TEXT.version(c.current))} - - )} -
    • - ) - })} -
+ {direct.length > 0 && list('direct', direct)} + {(possible.length > 0 || bare.length > 0) && ( +
+
{CATALOG_DRIFT_TEXT.possibleHead}
+ {possible.length > 0 && list('possible', possible)} + {bare.map((source) => ( +
{CATALOG_DRIFT_TEXT.possibleNone(source)}
+ ))} +
+ )}
{changes.length ? CATALOG_DRIFT_TEXT.changedCount(changes.length) : CATALOG_DRIFT_TEXT.unchanged} diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 3bc567b2..689a7c01 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -2224,26 +2224,40 @@ export const CATALOG_IMPACT_TEXT = { } /** - * 发布时的目录版本：发布时记下这一版 SQL 用到的表的数据目录版本；从这一版发起正式运行时目录有变化，提醒而不拦。 + * 发布时的目录版本：发布时记下这一版用到的表的数据目录版本；从这一版发起正式运行时目录有变化，提醒而不拦。 * 版本号写「第 N 版」，还没有目录的表写「尚无目录」。 + * + * 和影响面（CATALOG_IMPACT_TEXT）同一个分法：SQL 里写着的表是直接引用，说「有变化」；Agent 绑定了数据源的查询工具、 + * SQL 运行时才写的，记的是这个数据源中所有有目录的表，说「Agent 可能查询的表有变化」——两组分开写，不混成一句。 */ export const CATALOG_DRIFT_TEXT = { /** 运行里的提醒：「自发布以来，数据目录中「入园记录」等 2 张表有变化」 */ title: (first: string, n: number) => (n > 1 ? `自发布以来，数据目录中「${first}」等 ${formatNumber(n)} 张表有变化` : `自发布以来，数据目录中「${first}」有变化`), + /** 只有 Agent 可能查询的表变了：「自发布以来，Agent 可能查询的表有变化：「入园记录」等 3 张」 */ + possibleTitle: (first: string, n: number) => (n > 1 + ? `自发布以来，Agent 可能查询的表有变化：「${first}」等 ${formatNumber(n)} 张` + : `自发布以来，Agent 可能查询的表有变化：「${first}」`), + /** 两组都有：直接引用的标题后面补一句 */ + alsoPossible: (n: number) => `；另有 ${formatNumber(n)} 张 Agent 可能查询的表有变化`, sub: '运行未被拦截。请核对这些表的目录修改是否影响本次结果', table: (label: string | null, table: string) => (label ? `${label}（${table}）` : table), version: (v: number | null) => (v ? `第 ${formatNumber(v)} 版` : '尚无目录'), line: (source: string, table: string, from: string, to: string) => `「${source}」${table}：发布时${from}，现为${to}`, open: '查看目录', + /** 可能涉及那一组的小标题和说明（运行横幅、时间线展开、版本历史共用） */ + possibleHead: 'Agent 可能查询的表', + possibleHint: 'Agent 绑定了这些数据源的查询工具，SQL 在运行时生成：发布时记下了数据源中所有有目录的表，不一定都会用到', // ---- 版本历史里的「发布时的目录版本」 section: '发布时的目录版本', - sectionHint: '这一版 SQL 用到的表在发布时的数据目录版本', + sectionHint: '这一版 SQL 用到的表，以及 Agent 可能查询的表，在发布时的数据目录版本', changedSince: (now: string) => `之后有变化，现为${now}`, changedCount: (n: number) => `${formatNumber(n)} 张表在发布之后有变化，下次正式运行时会提醒`, unchanged: '发布之后这些表的目录没有变化', empty: '这一版的 SQL 没有用到数据库表', + /** 可能涉及的数据源在发布时一张有目录的表都没有 */ + possibleNone: (source: string) => `「${source}」发布时还没有目录`, } // =========================================================================== diff --git a/frontend/src/pages/runs/Banners.tsx b/frontend/src/pages/runs/Banners.tsx index 28643dad..06d45f43 100644 --- a/frontend/src/pages/runs/Banners.tsx +++ b/frontend/src/pages/runs/Banners.tsx @@ -5,6 +5,7 @@ import { CopyButton, Spinner, StatusBadge } from '../../components/ui' import { formatDateTime, formatSpan, formatTime } from '../../lib/format' import type { Approval, CatalogDriftTable } from '../../types' import { CATALOG_DRIFT_TEXT, CATALOG_IMPACT_TEXT } from '../../lib/terms' +import { catalogDriftLine } from '../../run/decode' import type { RunErrorExplain } from '../../lib/explain' import { LONG_WAIT_MS, ageMs } from './model' import { copyText } from './parts' @@ -28,12 +29,14 @@ function Shell({ color, children, data }: { color: string; children: ReactNode; // ------------------------------------------------------------------------- /** - * 从发布版本发起的正式运行，开始时发现这一版 SQL 用到的表的数据目录在发布之后改过。运行照常（目录是说明， + * 从发布版本发起的正式运行，开始时发现这一版用到的表的数据目录在发布之后改过。运行照常（目录是说明， * 不是运行的输入），但看结果的人要知道：「在园人数」在发布之后被改成了存量，这一版的 SQL 可能还在跨天相加。 - * 每张表给到数据目录的入口；表多时只列前几张 + * SQL 里写着的表在前；Agent 可能查询的表（Agent 运行时自己写 SQL）另起一组、带小标题，和直接引用分开。 + * 每张表给到数据目录的入口；每组表多时只列前几张 */ -export function CatalogDriftBanner({ drift }: { drift: { title: string; tables: CatalogDriftTable[] } }) { - const shown = drift.tables.slice(0, DRIFT_SHOWN) +export function CatalogDriftBanner({ drift }: { + drift: { title: string; direct: CatalogDriftTable[]; possible: CatalogDriftTable[] } +}) { return (
@@ -41,29 +44,41 @@ export function CatalogDriftBanner({ drift }: { drift: { title: string; tables:

{drift.title}

{CATALOG_DRIFT_TEXT.sub}

-
    - {shown.map((t) => ( -
  • - - {CATALOG_DRIFT_TEXT.line(t.source, CATALOG_DRIFT_TEXT.table(t.label, t.table), - CATALOG_DRIFT_TEXT.version(t.published), CATALOG_DRIFT_TEXT.version(t.current))} - - {t.source_id && ( - - {CATALOG_DRIFT_TEXT.open} - - )} -
    • - ))} - {drift.tables.length > shown.length &&
  • {CATALOG_IMPACT_TEXT.more(drift.tables.length - shown.length)}
    • } -
+ {drift.direct.length > 0 && } + {drift.possible.length > 0 && ( +
+

{CATALOG_DRIFT_TEXT.possibleHead}

+

{CATALOG_DRIFT_TEXT.possibleHint}

+ +
+ )}
) } +function DriftTables({ tables }: { tables: CatalogDriftTable[] }) { + const shown = tables.slice(0, DRIFT_SHOWN) + return ( +
    + {shown.map((t) => ( +
  • + {catalogDriftLine(t)} + {t.source_id && ( + + {CATALOG_DRIFT_TEXT.open} + + )} +
    • + ))} + {tables.length > shown.length &&
  • {CATALOG_IMPACT_TEXT.more(tables.length - shown.length)}
    • } +
+ ) +} + const DRIFT_SHOWN = 6 // ------------------------------------------------------------------------- diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index bf3a959e..8bea3225 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -2556,25 +2556,45 @@ function labelsOf(ops: CopilotOp[], fallback?: (id: string) => string | undefine return (id) => map.get(id) ?? fallback?.(id) } +/** 一张表的那一行：「scenic」入园记录（visits）：发布时第 3 版，现为第 5 版 */ +export function catalogDriftLine(t: CatalogDriftTable): string { + return CATALOG_DRIFT_TEXT.line(t.source, CATALOG_DRIFT_TEXT.table(t.label, t.table), + CATALOG_DRIFT_TEXT.version(t.published), CATALOG_DRIFT_TEXT.version(t.current)) +} + /** - * catalog.drift 事件 → 提醒的标题和逐表明细。标题按第一张表的中文名说（「入园记录」等 2 张表），明细每张表一行： - * 发布时第几版、现在第几版（0 写「尚无目录」）。认不出的形状返回 null + * catalog.drift 事件 → 提醒的标题和逐表明细。每张表带 impact：direct 是 SQL 里写着的表，possible 是 Agent 可能 + * 查询的表（Agent 绑定了数据源的查询工具，SQL 运行时才生成），两组分开说： + * - 标题按第一张直接引用的表说（「入园记录」等 2 张表有变化），还有可能涉及的补一句「另有 N 张 Agent 可能查询的表 + * 有变化」；只有可能涉及的，标题直说「Agent 可能查询的表有变化」 + * - 明细每张表一行：发布时第几版、现在第几版（0 写「尚无目录」）；可能涉及的放在小标题下面 + * 老事件的表没有 impact，按直接引用。认不出的形状返回 null */ -export function catalogDrift(d: Record): { title: string; detail: string; tables: CatalogDriftTable[] } | null { +export function catalogDrift(d: Record): { + title: string; detail: string; tables: CatalogDriftTable[]; direct: CatalogDriftTable[]; possible: CatalogDriftTable[] +} | null { const tables: CatalogDriftTable[] = (Array.isArray(d?.tables) ? d.tables : []) .filter((t: any) => t && typeof t === 'object' && typeof t.table === 'string' && typeof t.source === 'string') .map((t: any) => ({ source: t.source, source_id: String(t.source_id ?? ''), table: t.table, label: typeof t.label === 'string' && t.label ? t.label : null, published: num(t.published) ?? null, current: num(t.current) ?? 0, + impact: t.impact === 'possible' ? 'possible' : 'direct', })) if (!tables.length) return null - const first = tables[0].label ?? tables[0].table + const direct = tables.filter((t) => t.impact === 'direct') + const possible = tables.filter((t) => t.impact === 'possible') + const name = (t: CatalogDriftTable) => t.label ?? t.table + const title = direct.length + ? CATALOG_DRIFT_TEXT.title(name(direct[0]), direct.length) + (possible.length ? CATALOG_DRIFT_TEXT.alsoPossible(possible.length) : '') + : CATALOG_DRIFT_TEXT.possibleTitle(name(possible[0]), possible.length) return { - title: CATALOG_DRIFT_TEXT.title(first, tables.length), - detail: tables.map((t) => CATALOG_DRIFT_TEXT.line(t.source, CATALOG_DRIFT_TEXT.table(t.label, t.table), - CATALOG_DRIFT_TEXT.version(t.published), CATALOG_DRIFT_TEXT.version(t.current))).join('\n'), - tables, + title, + detail: [ + ...direct.map(catalogDriftLine), + ...(possible.length ? [`${CATALOG_DRIFT_TEXT.possibleHead}：`, ...possible.map(catalogDriftLine)] : []), + ].join('\n'), + tables, direct, possible, } } diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 1798b3b3..6fb841b9 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -56,12 +56,23 @@ export interface WorkflowVersion { input_fields?: { name: string; required?: boolean; [key: string]: any }[] /** 是不是当前的已发布版本 */ published?: boolean - /** 发布时这一版 SQL 用到的表的数据目录版本 {源名: {表名: 版本}}，0 表示当时还没有目录；没记过为 null */ - catalog_versions?: Record> | null + /** 发布时记下的数据目录版本，没记过为 null */ + catalog_versions?: CatalogVersions | null /** 发布之后目录有变化的表 */ catalog_changes?: CatalogDriftTable[] } +/** + * 发布时记下的数据目录版本（服务端 catalog_impact.recorded_versions），两组都是 {源名: {表名: 版本}}： + * - direct：SQL 里写着的表，0 表示当时还没有目录 + * - possible：Agent 绑定了查询工具的数据源中所有有目录的表（SQL 运行时才生成）；源下为空表示当时一张有目录的 + * 表都没有。只记了直接引用的老版本这一组为空 + */ +export interface CatalogVersions { + direct: Record> + possible: Record> +} + /** 发布之后数据目录有变化的一张表（服务端 catalog_impact.catalog_drift；运行里的 catalog.drift 事件同形） */ export interface CatalogDriftTable { source: string @@ -72,6 +83,8 @@ export interface CatalogDriftTable { /** 发布时的版本，0 表示当时还没有目录 */ published: number | null current: number + /** direct：SQL 里写着这张表；possible：Agent 可能查询的表。老事件没有这一项，按 direct */ + impact: 'direct' | 'possible' } /** diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index d0d05ee4..0bf2a0a8 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -737,6 +737,26 @@ await section('正式运行：发布之后数据目录有变化（catalog.drift && row.detail.includes('「scenic」parks：发布时尚无目录，现为第 1 版'), row?.detail) const one = mod.catalogDrift({ tables: [{ source: 's', table: 't', label: null, published: 1, current: 2 }] }) check('只有一张表：标题不写「等」', one?.title === '自发布以来，数据目录中「t」有变化', one?.title) + check('老事件的表没有 impact：按直接引用', one?.direct.length === 1 && one?.possible.length === 0 && one?.tables[0].impact === 'direct') + // Agent 可能查询的表（SQL 运行时才写）：和直接引用分开说 + const mixed = mod.catalogDrift({ tables: [ + { source: 'scenic', source_id: 'src-scenic', table: 'visits', label: '入园记录', published: 3, current: 5, impact: 'direct' }, + { source: 'scenic', source_id: 'src-scenic', table: 'orders', label: '订单', published: 2, current: 4, impact: 'possible' }, + { source: 'scenic', source_id: 'src-scenic', table: 'gates', label: null, published: 0, current: 1, impact: 'possible' }, + ] }) + check('两组都有：标题说直接引用的，再补一句 Agent 可能查询的表有变化', + mixed?.title === '自发布以来，数据目录中「入园记录」有变化；另有 2 张 Agent 可能查询的表有变化', mixed?.title) + check('……展开：直接引用在前，可能涉及的在小标题下面', mixed?.detail === [ + '「scenic」入园记录（visits）：发布时第 3 版，现为第 5 版', 'Agent 可能查询的表：', + '「scenic」订单（orders）：发布时第 2 版，现为第 4 版', '「scenic」gates：发布时尚无目录，现为第 1 版'].join('\n'), mixed?.detail) + const agentOnly = mod.catalogDrift({ tables: [ + { source: 'scenic', source_id: 'src-scenic', table: 'orders', label: '订单', published: 2, current: 4, impact: 'possible' }, + ] }) + check('只有可能涉及的：标题直说「Agent 可能查询的表有变化」', agentOnly?.title === '自发布以来，Agent 可能查询的表有变化：「订单」', agentOnly?.title) + const run = mod.decodeRun([{ seq: 1, type: 'catalog.drift', node_id: null, ts: 1, data: { version: 2, count: 1, tables: [ + { source: 'scenic', source_id: 'src-scenic', table: 'orders', label: '订单', published: 2, current: 4, impact: 'possible' }] } }]) + .find((s) => s.code === 'catalog_drift') + check('……时间线里那一行同样是警告', run?.level === 'warn' && run?.title === agentOnly?.title, run?.title) check('形状不对不出行', mod.decodeRun([{ seq: 1, type: 'catalog.drift', node_id: null, ts: 1, data: { tables: 'x' } }]) .every((s) => s.code !== 'catalog_drift')) }) diff --git a/scripts/check-runs.mjs b/scripts/check-runs.mjs index c86ba48c..fc4e9ef4 100644 --- a/scripts/check-runs.mjs +++ b/scripts/check-runs.mjs @@ -706,9 +706,11 @@ await section('正式运行：发布之后数据目录有变化，提醒但不 ], edges: [{ source: 'a', target: 'q' }] } const events = [ { seq: 1, type: 'run.started', node_id: null, ts: t, data: { nodes: 2 } }, - { seq: 2, type: 'catalog.drift', node_id: null, ts: t + 0.01, data: { version: 3, count: 2, tables: [ - { source: 'scenic', source_id: 'src-scenic', table: 'visits', label: '入园记录', published: 3, current: 5 }, - { source: 'scenic', source_id: 'src-scenic', table: 'parks', label: null, published: 0, current: 1 }, + { seq: 2, type: 'catalog.drift', node_id: null, ts: t + 0.01, data: { version: 3, count: 3, tables: [ + { source: 'scenic', source_id: 'src-scenic', table: 'visits', label: '入园记录', published: 3, current: 5, impact: 'direct' }, + { source: 'scenic', source_id: 'src-scenic', table: 'parks', label: null, published: 0, current: 1, impact: 'direct' }, + // Agent 绑定了这个源的查询工具：SQL 运行时才写，记的是源里有目录的表 + { source: 'scenic', source_id: 'src-scenic', table: 'orders', label: '订单', published: 2, current: 4, impact: 'possible' }, ] } }, { seq: 3, type: 'node.started', node_id: 'a', ts: t + 0.1, data: {} }, { seq: 4, type: 'node.finished', node_id: 'a', ts: t + 0.2, data: { duration_ms: 100 } }, @@ -723,11 +725,20 @@ await section('正式运行：发布之后数据目录有变化，提醒但不 const banner = page.locator('[data-run-banner="catalog-drift"]') check('详情显示目录变化的横幅', await banner.waitFor({ timeout: 5000 }).then(() => true, () => false)) const text = await banner.innerText().catch(() => '') - check('横幅标题：「入园记录」等 2 张表有变化', text.includes('自发布以来，数据目录中「入园记录」等 2 张表有变化'), text.split('\n')[0]) + check('横幅标题：「入园记录」等 2 张表有变化，另有 1 张 Agent 可能查询的表有变化', + text.includes('自发布以来，数据目录中「入园记录」等 2 张表有变化；另有 1 张 Agent 可能查询的表有变化'), text.split('\n')[0]) check('写明运行未被拦截', text.includes('运行未被拦截')) check('逐表写发布时和现在的版本，没有目录写「尚无目录」', text.includes('「scenic」入园记录（visits）：发布时第 3 版，现为第 5 版') && text.includes('「scenic」parks：发布时尚无目录，现为第 1 版'), text.replace(/\s+/g, ' ')) check('每张表给到数据目录的入口', (await banner.locator('[data-drift-open="visits"]').getAttribute('href')) === '/data/catalog/src-scenic/visits') + const possible = banner.locator('[data-drift-possible]') + const possibleText = await possible.innerText().catch(() => '') + check('Agent 可能查询的表另起一组：小标题、说明、逐表一行', await possible.count() === 1 + && possibleText.includes('Agent 可能查询的表') && possibleText.includes('SQL 在运行时生成') + && possibleText.includes('「scenic」订单（orders）：发布时第 2 版，现为第 4 版'), possibleText.replace(/\s+/g, ' ')) + check('……直接引用的表不在这一组里', await possible.locator('[data-drift-table="visits"]').count() === 0 + && (await banner.locator('[data-drift-table="orders"]').getAttribute('data-drift-impact')) === 'possible') + check('……也给到数据目录的入口', (await banner.locator('[data-drift-open="orders"]').getAttribute('href')) === '/data/catalog/src-scenic/orders') check('运行照常完成：没有失败横幅', await page.locator('[data-run-banner=failed]').count() === 0) const row = page.locator('[data-step-code="catalog_drift"]') check('时间线里也有一行提醒', await row.count() === 1 && (await row.innerText()).includes('等 2 张表有变化')) diff --git a/scripts/check-studio.mjs b/scripts/check-studio.mjs index 02bd4b84..90cba8cc 100644 --- a/scripts/check-studio.mjs +++ b/scripts/check-studio.mjs @@ -263,11 +263,17 @@ async function open({ width = 1440, height = 900, path = '/studio/st-main', pref if (sub === 'versions' && !v) return json(route, VERSIONS) if (sub === 'versions' && v) { const n = Number(v) - // 已发布的 v2 记着发布时的目录版本：入园记录之后改过（第 3 版 → 第 5 版），景区表没变 + // 已发布的 v2 记着发布时的目录版本：入园记录之后改过（第 3 版 → 第 5 版），景区表没变； + // Agent 绑定了 shop 的查询工具，可能查询的表里订单改过、闸机是发布之后才建的目录，另一个源发布时还没有目录 return json(route, { ...VERSIONS.find((x) => x.version === n), workflow_id: id, graph: n === 1 ? V1 : GRAPH, graph_hash: 'x', input_fields: [{ name: 'goal' }], published: n === 2, - catalog_versions: n === 2 ? { shop: { parks: 1, visits: 3 } } : null, - catalog_changes: n === 2 ? [{ source: 'shop', source_id: 'src-shop', table: 'visits', label: '入园记录', published: 3, current: 5 }] : [] }) + catalog_versions: n === 2 + ? { direct: { shop: { parks: 1, visits: 3 } }, possible: { shop: { orders: 2, channels: 1 }, crm: {} } } : null, + catalog_changes: n === 2 ? [ + { source: 'shop', source_id: 'src-shop', table: 'visits', label: '入园记录', published: 3, current: 5, impact: 'direct' }, + { source: 'shop', source_id: 'src-shop', table: 'orders', label: '订单', published: 2, current: 4, impact: 'possible' }, + { source: 'shop', source_id: 'src-shop', table: 'gates', label: null, published: 0, current: 1, impact: 'possible' }, + ] : [] }) } if (method === 'GET' && !sub) return state.deleted.has(id) ? json(route, { detail: '工作流不存在，可能已被删除' }, 404) : json(route, FAKES[id]) if (method === 'DELETE' && !sub) { @@ -1011,8 +1017,20 @@ await section('版本历史：预览、恢复成一次可撤销的改动', async const catText = await cat.innerText().catch(() => '') check('已发布的版本写「发布时的目录版本」', catText.includes('发布时的目录版本') && catText.includes('parks') && catText.includes('第 1 版'), catText) check('之后改过的表标出来', (await page.locator('[data-version-catalog-table="visits"]').getAttribute('data-changed')) === 'true' - && catText.includes('之后有变化，现为第 5 版') && catText.includes('1 张表在发布之后有变化'), catText.replace(/\s+/g, ' ')) + && catText.includes('之后有变化，现为第 5 版') && catText.includes('3 张表在发布之后有变化'), catText.replace(/\s+/g, ' ')) check('没改过的表不标', (await page.locator('[data-version-catalog-table="parks"]').getAttribute('data-changed')) === null) + // Agent 可能查询的表：另起一组，和 SQL 里写着的表分开 + const possible = cat.locator('[data-version-catalog-possible]') + const possibleText = await possible.innerText().catch(() => '') + check('Agent 可能查询的表另起一组', await possible.count() === 1 && possibleText.includes('Agent 可能查询的表') + && await possible.locator('[data-version-catalog-table="visits"]').count() === 0, possibleText.replace(/\s+/g, ' ')) + check('……改过的标出来，没改过的不标', (await possible.locator('[data-version-catalog-table="orders"]').getAttribute('data-changed')) === 'true' + && (await possible.locator('[data-version-catalog-table="channels"]').getAttribute('data-changed')) === null + && possibleText.includes('之后有变化，现为第 4 版'), possibleText.replace(/\s+/g, ' ')) + check('……发布之后才建目录的表也列出来，发布时写「尚无目录」', + (await possible.locator('[data-version-catalog-table="gates"]').getAttribute('data-changed')) === 'true' + && (await possible.locator('[data-version-catalog-table="gates"]').innerText()).includes('尚无目录')) + check('……发布时一张有目录的表都没有的源写一句', possibleText.includes('「crm」发布时还没有目录'), possibleText.replace(/\s+/g, ' ')) await page.locator('ol[aria-label="版本"] > li').nth(2).locator('button').first().click() await page.waitForTimeout(300) check('没发布过的版本没有目录版本这一栏', await count(page, '[data-version-catalog]') === 0) From 9c0283354425b8b295caf0be82123127abeceeff Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:25:19 +0800 Subject: [PATCH 7/7] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3?= =?UTF-8?q?=20/=20B4=20=E7=95=8C=E9=9D=A2=EF=BC=9A=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=E9=A1=B5=E9=A1=B6=E9=83=A8=E8=AF=B4=E6=98=8E=E3=80=8C=E8=BF=99?= =?UTF-8?q?=E4=B8=AA=E6=95=B0=E6=8D=AE=E6=BA=90=E5=85=B1=E6=9C=89=20N=20?= =?UTF-8?q?=E5=BC=A0=E8=A1=A8=EF=BC=8C=E5=8F=AA=E6=8E=A2=E6=9F=A5=E4=BA=86?= =?UTF-8?q?=E5=89=8D=20200=20=E5=BC=A0=E3=80=8D=EF=BC=8C=E5=86=99=E6=98=8E?= =?UTF-8?q?=E4=B8=BA=E4=BB=80=E4=B9=88=E5=8F=AA=E5=8F=96=E4=BA=86=E8=BF=99?= =?UTF-8?q?=E4=BA=9B=E3=80=81=E5=85=B6=E4=BD=99=E7=9A=84=E5=8A=A9=E6=89=8B?= =?UTF-8?q?=E4=B9=9F=E7=9C=8B=E4=B8=8D=E5=88=B0=EF=BC=8C=E4=B8=8B=E4=B8=80?= =?UTF-8?q?=E6=AD=A5=E5=9B=9E=E5=88=B0=E6=95=B0=E6=8D=AE=E6=BA=90=E5=8D=A1?= =?UTF-8?q?=E7=89=87=E6=8D=A2=20schema=20=E6=88=96=E7=BC=A9=E5=B0=8F?= =?UTF-8?q?=E8=8C=83=E5=9B=B4=E5=90=8E=E9=87=8D=E6=96=B0=E6=8E=A2=E6=9F=A5?= =?UTF-8?q?=EF=BC=9B=E5=8A=A9=E6=89=8B=E3=80=8C=E5=8F=82=E8=80=83=E4=BA=86?= =?UTF-8?q?=E5=93=AA=E4=BA=9B=E8=A1=A8=E3=80=8D=E9=82=A3=E4=B8=80=E8=A1=8C?= =?UTF-8?q?=E6=88=AA=E6=96=AD=E4=BA=86=E7=9A=84=E5=BA=93=E5=86=99=E3=80=8C?= =?UTF-8?q?=E5=85=B1=20N=20=E5=BC=A0=EF=BC=8C=E5=8F=AA=E6=8E=A2=E6=9F=A5?= =?UTF-8?q?=E4=BA=86=20200=20=E5=BC=A0=E3=80=8D=EF=BC=9B=E8=A1=A5=E6=A3=80?= =?UTF-8?q?=E6=9F=A5=E8=84=9A=E6=9C=AC=E7=9A=84=E6=96=AD=E8=A8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend/src/lib/terms.ts | 15 +++++++++ frontend/src/pages/catalog/CatalogPage.tsx | 29 ++++++++++++++++- frontend/src/run/decode.ts | 24 ++++++++++---- frontend/src/types.ts | 4 +++ scripts/check-catalog.mjs | 38 +++++++++++++++++++++- scripts/check-decode.mjs | 10 ++++++ 6 files changed, 111 insertions(+), 9 deletions(-) diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 689a7c01..5820bd34 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -2260,6 +2260,21 @@ export const CATALOG_DRIFT_TEXT = { possibleNone: (source: string) => `「${source}」发布时还没有目录`, } +/** + * 表结构只探查了一部分：探查结构时每个数据源最多取 200 张表（服务端 introspect._MAX_TABLES），多出来的表不在 + * 表结构里，数据目录列不出，助手也看不到它们的字段。数据目录页顶部和助手「参考了哪些表」那一行都要说出来。 + */ +export const SCHEMA_PARTIAL_TEXT = { + /** 数据目录页顶部：「这个数据源共有 205 张表，只探查了前 200 张」 */ + title: (total: number, explored: number) => + `这个数据源共有 ${formatNumber(total)} 张表，只探查了前 ${formatNumber(explored)} 张`, + body: (rest: number) => `探查结构时每个数据源最多取 200 张表（按数据库列出的顺序）：生产库动辄上千张，全部取回既慢也看不完。` + + `其余 ${formatNumber(rest)} 张不在数据目录中，助手也看不到它们的字段。`, + next: '所需的表不在清单中时，请回到数据源卡片换一个 schema，或改用只能看到所需表的数据库账号缩小范围，然后重新「探查结构」。', + /** 助手「参考了哪些表」那一行：「共 205 张，只探查了 200 张」 */ + context: (total: number, explored: number) => `共 ${formatNumber(total)} 张，只探查了 ${formatNumber(explored)} 张`, +} + // =========================================================================== // 数据剖析（数据源设置里的开关和预算；数据目录页的「数据剖析」） // diff --git a/frontend/src/pages/catalog/CatalogPage.tsx b/frontend/src/pages/catalog/CatalogPage.tsx index afcaeed1..24bd2dc9 100644 --- a/frontend/src/pages/catalog/CatalogPage.tsx +++ b/frontend/src/pages/catalog/CatalogPage.tsx @@ -7,7 +7,7 @@ import { ApiError, api } from '../../api/client' import type { CatalogDetail, CatalogList, CatalogTableRow } from '../../types' import { EmptyState, ErrorState, PageHeader, Skeleton, Spinner, toast, useTicker } from '../../components/ui' import { useCatalog, useDatasources, useOnReconnect } from '../../store/catalog' -import { CATALOG_TEXT as CT, PROFILE_STOP_LABEL, PROFILE_TEXT as PT } from '../../lib/terms' +import { CATALOG_TEXT as CT, PROFILE_STOP_LABEL, PROFILE_TEXT as PT, SCHEMA_PARTIAL_TEXT as SPT } from '../../lib/terms' import { StatusLegend, SystemNotesNotice } from './parts' import { DraftDialog } from './DraftDialog' import { ProfileDialog } from './ProfileDialog' @@ -308,6 +308,10 @@ export function CatalogPage() {
)} /> + {hasTables && data?.schema_truncated && ( + r.in_schema).length} + onGoSource={() => navigate(back)} /> + )} {hasTables && usage.used > 0 && !empty && } {body} {drafting && data && ( @@ -347,6 +351,29 @@ export function CatalogPage() { ) } +/** + * 表结构只探查了一部分（每个数据源最多取 200 张表）：没探查到的表不在清单里，助手也看不到。顶部常驻一条说明， + * 写清一共几张、探查了几张、为什么只取了这些，以及下一步——回到数据源卡片换 schema 或缩小范围后重新探查 + */ +function SchemaPartialNotice({ total, explored, onGoSource }: { total: number; explored: number; onGoSource: () => void }) { + const shownTotal = Math.max(total, explored) + return ( +
+ +
+

{SPT.title(shownTotal, explored)}

+

{SPT.body(Math.max(shownTotal - explored, 0))}

+

{SPT.next}

+
+ +
+ ) +} + /** * 「用到但没确认」：运行中查询过的表里还有几张有推断项、几张没有目录。这些表的推断项会被助手当真用，最该先审。 * 两段各是一个按钮，点了切到清单的对应筛选、按使用次数排；都审完了只写一句，不给按钮 diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index 8bea3225..3c697991 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -2,8 +2,8 @@ import type { CatalogDriftTable, RunEvent, SqlCheckItem, TeamMember, TeamRound, import type { NodeState, RunPhase } from './trace' import { formatDuration, formatNumber } from '../lib/format' import { - CATALOG_DRIFT_TEXT, CATALOG_PATCH_TEXT, JUDGE_TEXT, MERGE_TEXT, SQL_CHECK_TEXT, TYPE_LABEL, claimTally, evidenceTally, - issuanceLabel, nodeTypeLabel, + CATALOG_DRIFT_TEXT, CATALOG_PATCH_TEXT, JUDGE_TEXT, MERGE_TEXT, SCHEMA_PARTIAL_TEXT, SQL_CHECK_TEXT, TYPE_LABEL, claimTally, + evidenceTally, issuanceLabel, nodeTypeLabel, } from '../lib/terms' import { catalogPatchOf, patchValueText, patchWhere } from './catalogPatch' import { claimCountsOf, isJudged, statsTally } from '../lib/evidence' @@ -2605,8 +2605,10 @@ export interface CopilotContextSource { tables: string[] /** model：按需求挑的；all：库不大，全部表都带字段；fallback：没能挑出来，只给了表名 */ selectedBy: 'model' | 'all' | 'fallback' - /** 这个库一共几张表 */ + /** 这个库一共几张表（数据库里的真实总数） */ total?: number + /** 探查结构截断了时，探查到几张（服务端只在截断时给）：模型只看得到这几张 */ + explored?: number /** 没能挑出来的原因 */ reason?: string } @@ -2620,26 +2622,34 @@ function contextSources(op: CopilotOp): CopilotContextSource[] { .filter((x): x is Record => !!x && typeof x === 'object' && typeof (x as any).source === 'string') .map((x) => { const total = num(x.total) + const explored = num(x.explored) return { source: x.source, tables: Array.isArray(x.tables) ? x.tables.map(String) : [], selectedBy: x.selected_by === 'model' || x.selected_by === 'fallback' ? x.selected_by : 'all', ...(total != null ? { total } : {}), + ...(total != null && explored != null && explored < total ? { explored } : {}), ...(typeof x.reason === 'string' && x.reason ? { reason: x.reason } : {}), } }) } -/** 一个库在展开区里的那一行：怎么来的 + 表名 */ +/** + * 一个库在展开区里的那一行：怎么来的 + 表名。探查结构截断了的库写明「共 205 张，只探查了 200 张」： + * 模型只看得到探查到的那些，用户说的表不在里面时，先得知道是没探查到，不是库里没有 + */ function contextLine(g: CopilotContextSource): string { const names = g.tables.join('、') - const of = g.total != null ? `共 ${g.total} 张表` : '' + const partial = g.total != null && g.explored != null ? SCHEMA_PARTIAL_TEXT.context(g.total, g.explored) : '' + const of = partial || (g.total != null ? `共 ${g.total} 张表` : '') switch (g.selectedBy) { case 'all': - return `「${g.source}」全部 ${g.tables.length} 张表：${names}` + return `「${g.source}」全部 ${g.tables.length} 张表${partial ? `（${partial}）` : ''}：${names}` case 'model': return g.tables.length - ? `「${g.source}」按需求${g.total != null ? `从 ${g.total} 张表中` : ''}挑出 ${g.tables.length} 张：${names}` + ? partial + ? `「${g.source}」${partial}，按需求挑出 ${g.tables.length} 张：${names}` + : `「${g.source}」按需求${g.total != null ? `从 ${g.total} 张表中` : ''}挑出 ${g.tables.length} 张：${names}` : `「${g.source}」${of ? `${of}，` : ''}未挑中与需求相关的表，已提供全部表名` default: return `「${g.source}」${of ? `${of}，` : ''}${CONTEXT_FALLBACK_TEXT}` diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 6fb841b9..8053d369 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -2671,6 +2671,10 @@ export interface CatalogList { system_notes: boolean /** 没有表结构时的原因（「尚未探查结构」「结构探查失败：…」）；有表结构时为 null */ schema_note: string | null + /** 探查结构截断了（每个数据源最多取 200 张表）：没探查到的表不在清单里，助手也看不到。老后端没有这一项 */ + schema_truncated?: boolean + /** 数据库里一共几张表（含视图）；没截断时等于探查到的张数 */ + schema_total?: number } /** 表结构里的一列 */ diff --git a/scripts/check-catalog.mjs b/scripts/check-catalog.mjs index 65b980f0..2680c5e8 100644 --- a/scripts/check-catalog.mjs +++ b/scripts/check-catalog.mjs @@ -347,7 +347,11 @@ function listOf(state, src) { const all = state.tables[src] const rows = all.filter((t) => !t.missing).sort((a, b) => b.usage - a.usage).map(rowOf) rows.push(...all.filter((t) => t.missing).map(rowOf)) - return { tables: rows, system_notes: isUpload(src), schema_note: all.length ? null : '尚未探查结构' } + // 探查结构截断了的源（state.schemaTotal 里给了数据库里的总数）：清单只有探查到的那部分 + const explored = all.filter((t) => !t.missing).length + const total = state.schemaTotal?.[src] ?? explored + return { tables: rows, system_notes: isUpload(src), schema_note: all.length ? null : '尚未探查结构', + schema_truncated: total > explored, schema_total: total } } const detailOf = (src, t) => ({ table_name: t.name, in_schema: !t.missing, notes: t.notes ?? {}, version: t.version, updated_at: t.updated_at, updated_by: t.updated_by, @@ -1186,6 +1190,38 @@ const profilePhase = (page) => profileBox(page).getAttribute('data-profile-dialo /** 报告里不该露的机读码：停止原因、跳过原因、状态、基数的英文值 */ const PROFILE_CODE_RE = /\b(?:budget|deadline|failed|masked|too_large|high_cardinality|no_data|unsupported|verified|proposed|many_to_one|row_estimate)\b/ +await section('数据目录 · 只探查了一部分表', async () => { + // 探查结构每个数据源最多取 200 张表：多出来的不在清单里，助手也看不到。顶部要说出来，并给出下一步 + const state = freshState() + const explored = state.tables[S1].filter((t) => !t.missing).length + const total = explored + 57 + state.schemaTotal = { [S1]: total } + const { page, sent, errors, natives, close } = await open(`/data/catalog/${S1}`, { handlers: catalogHandlers(state) }) + const bar = page.locator('[data-catalog-schema-partial]') + check('顶部有一条说明', await bar.waitFor({ timeout: 3000 }).then(() => true, () => false)) + check('……数字：一共几张、探查了几张', (await bar.getAttribute('data-catalog-schema-partial')) === `${total},${explored}`, + await bar.getAttribute('data-catalog-schema-partial')) + const text = await bar.innerText().catch(() => '') + check('……写成「共有 N 张表，只探查了前 M 张」', text.includes(`这个数据源共有 ${total} 张表，只探查了前 ${explored} 张`), text) + check('……说明为什么只取了这些、其余的助手也看不到', text.includes('最多取 200 张表') && text.includes('其余 57 张不在数据目录中') + && text.includes('助手也看不到'), text.replace(/\s+/g, ' ')) + check('……给出下一步：换 schema 或缩小范围后重新探查', text.includes('换一个 schema') && text.includes('缩小范围') + && text.includes('探查结构'), text.replace(/\s+/g, ' ')) + await shot(page, 'catalog-schema-partial') + await cleanCopy(page, '只探查了一部分表') + await bar.locator('[data-catalog-schema-partial-go]').click() + await page.waitForURL('**/data/databases', { timeout: 5000 }).catch(() => {}) + check('……按钮回到数据源卡片', new URL(page.url()).pathname === '/data/databases', page.url()) + // 没截断的库不出这一条 + await goto(page, `/data/catalog/${S4}`) + await page.locator('[data-catalog-overview]').waitFor() + check('没截断的库不出这一条', await page.locator('[data-catalog-schema-partial]').count() === 0) + check('只读、不发写请求', writes(sent, /./).length === 0) + check('没有原生对话框', natives.length === 0, natives.join(' | ')) + check('没有运行时报错', errors.length === 0, errors.join(' | ')) + await close() +}) + await section('数据目录 · 数据剖析设置', async () => { const state = freshState() const replies = { patch: [] } diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index 0bf2a0a8..69c59810 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -685,6 +685,16 @@ await section('Copilot：这一轮参考了哪些表（context 操作）', async { source: 'b', tables: [], selected_by: 'model', total: 70 }] }) check('挑表成功但某个库一张没挑中：说清只给了表名', !!none?.detail.includes('「b」共 70 张表，未挑中与需求相关的表，已提供全部表名'), none?.detail) + // 探查结构截断了（每个数据源最多取 200 张表）：total 是数据库里的总数，explored 是探查到的 + const partial = mod.copilotContext({ op: 'context', sources: [ + { source: 'wide', tables: ['t001', 't002'], selected_by: 'model', total: 205, explored: 200 }, + { source: 'wide2', tables: [], selected_by: 'fallback', total: 260, explored: 200, reason: 'r' }] }) + check('截断了的库写「共 N 张，只探查了 200 张」', !!partial?.detail.includes('「wide」共 205 张，只探查了 200 张，按需求挑出 2 张：t001、t002') + && partial.detail.includes('「wide2」共 260 张，只探查了 200 张，未能按需求挑选'), partial?.detail) + const whole = mod.copilotContext({ op: 'context', sources: [ + { source: 'a', tables: ['x'], selected_by: 'model', total: 60, explored: 60 }] }) + check('探查到的和总数一样：照旧写「从 N 张表中挑出」', !!whole?.detail.includes('「a」按需求从 60 张表中挑出 1 张') && !whole.detail.includes('只探查了'), + whole?.detail) check('形状不对不出行、不报错', mod.decodeCopilot([ { op: 'context' }, { op: 'context', sources: 'x' }, { op: 'context', sources: [null, 3, { tables: ['t'] }] }, ]).length === 0)