合入完整性修复 F3:Agent 查库的模板也记目录版本并提醒「可能涉及」,只探查了部分表时清单、目录页和助手上下文都说明,升级入口接上 SQL 检查,截断判定和工具节点遍历各收成一处

This commit is contained in:
ThearchyHelios
2026-10-04 01:28:59 +08:00
24 changed files with 1032 additions and 222 deletions
+12 -1
View File
@@ -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)),
}
+9 -3
View File
@@ -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:
+66 -24
View File
@@ -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]]:
"""给模型看的目录:去掉驳回项。"""
@@ -212,35 +240,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}
@@ -337,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
@@ -506,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)
@@ -532,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)
@@ -571,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:
@@ -604,7 +640,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 = []
@@ -636,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)
+4 -2
View File
@@ -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
@@ -186,8 +187,9 @@ async def start_run(
await run_manager.note(run.id, "log", **governance_note(
governed=governed, workflow_name=name, version=version))
if catalog_changes:
# 落在封存范围内:之后回看这次正式出具,能知道当时的目录和发布时不一样、差在哪几张表
await run_manager.note(run.id, "catalog.drift", version=version, count=len(catalog_changes),
# 落在封存范围内:之后回看这次正式出具,能知道当时的目录和发布时不一样、差在哪几张表。
# 每张表带 impact:direct 写死的 SQL 用到的,possible Agent 可能查询的(界面分开说)
await run_manager.note(run.id, EventType.CATALOG_DRIFT, version=version, count=len(catalog_changes),
tables=catalog_changes)
# 已声明策略的升版记入事件流,出具物上能看到"这期换口径了、怎么处置的"
for ev in upgrade_events:
+7 -5
View File
@@ -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, # 剩下的都是警告
}
+2 -1
View File
@@ -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、统计、违规清单。落在封存范围内,
+184 -67
View File
@@ -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"]
+31 -13
View File
@@ -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,
+6 -13
View File
@@ -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:
+37
View File
@@ -183,3 +183,40 @@ 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"]}
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
+156 -15
View File
@@ -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"})
@@ -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()
+51
View File
@@ -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"]
+33
View File
@@ -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="秒"):
+55 -28
View File
@@ -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<string, Record<string, number>>
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) => (
<ul className="mt-1 space-y-0.5 text-2xs">
{rows.map(({ source, table, v }) => {
const c = changed.get(`${impact}/${source}/${table}`)
return (
<li key={`${source}/${table}`} className="flex min-w-0 flex-wrap items-baseline gap-x-1.5"
data-version-catalog-table={table} data-impact={impact} data-changed={c ? 'true' : undefined}>
<span className="mono min-w-0 truncate" title={`${source} · ${table}`}>{table}</span>
<span className="shrink-0 text-faint">{CATALOG_DRIFT_TEXT.version(v)}</span>
{c && (
<span className="shrink-0" style={{ color: 'var(--st-waiting)' }}>
{CATALOG_DRIFT_TEXT.changedSince(CATALOG_DRIFT_TEXT.version(c.current))}
</span>
)}
</li>
)
})}
</ul>
)
const empty = !direct.length && !possible.length && !bare.length
return (
<div className="mt-3 border-t pt-2" data-version-catalog={changes.length}>
<div className="text-2xs font-medium text-dim" title={CATALOG_DRIFT_TEXT.sectionHint}>{CATALOG_DRIFT_TEXT.section}</div>
{!rows.length ? <div className="mt-1 text-2xs text-faint">{CATALOG_DRIFT_TEXT.empty}</div> : (
{empty ? <div className="mt-1 text-2xs text-faint">{CATALOG_DRIFT_TEXT.empty}</div> : (
<>
<ul className="mt-1 space-y-0.5 text-2xs">
{rows.map(({ source, table, v }) => {
const c = changed.get(`${source}/${table}`)
return (
<li key={`${source}/${table}`} className="flex min-w-0 flex-wrap items-baseline gap-x-1.5" data-version-catalog-table={table}
data-changed={c ? 'true' : undefined}>
<span className="mono min-w-0 truncate" title={`${source} · ${table}`}>{table}</span>
<span className="shrink-0 text-faint">{CATALOG_DRIFT_TEXT.version(v)}</span>
{c && (
<span className="shrink-0" style={{ color: 'var(--st-waiting)' }}>
{CATALOG_DRIFT_TEXT.changedSince(CATALOG_DRIFT_TEXT.version(c.current))}
</span>
)}
</li>
)
})}
</ul>
{direct.length > 0 && list('direct', direct)}
{(possible.length > 0 || bare.length > 0) && (
<div className="mt-1.5" data-version-catalog-possible={possible.length}>
<div className="text-2xs text-dim" title={CATALOG_DRIFT_TEXT.possibleHint}>{CATALOG_DRIFT_TEXT.possibleHead}</div>
{possible.length > 0 && list('possible', possible)}
{bare.map((source) => (
<div key={source} className="mt-0.5 text-2xs text-faint">{CATALOG_DRIFT_TEXT.possibleNone(source)}</div>
))}
</div>
)}
<div className="mt-1 text-2xs" style={{ color: changes.length ? 'var(--st-waiting)' : undefined }}>
<span className={changes.length ? undefined : 'text-faint'}>
{changes.length ? CATALOG_DRIFT_TEXT.changedCount(changes.length) : CATALOG_DRIFT_TEXT.unchanged}
+31 -2
View File
@@ -2229,26 +2229,55 @@ 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}」发布时还没有目录`,
}
/**
* 表结构只探查了一部分:探查结构时每个数据源最多取 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)} 张`,
}
// ===========================================================================
+28 -1
View File
@@ -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() {
</div>
)}
/>
{hasTables && data?.schema_truncated && (
<SchemaPartialNotice total={data.schema_total ?? 0} explored={rows.filter((r) => r.in_schema).length}
onGoSource={() => navigate(back)} />
)}
{hasTables && usage.used > 0 && !empty && <UsageSummary {...usage} active={filter} onFocus={focusUsed} />}
{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 (
<div className="flex shrink-0 items-start gap-2 border-b px-4 py-2 text-xs"
style={{ background: 'color-mix(in srgb, var(--st-waiting) 6%, var(--bg-panel))' }}
data-catalog-schema-partial={`${shownTotal},${explored}`}>
<Database size={12} className="mt-0.5 shrink-0" style={{ color: 'var(--st-waiting)' }} aria-hidden />
<div className="min-w-0 flex-1 leading-relaxed">
<p className="font-medium">{SPT.title(shownTotal, explored)}</p>
<p className="text-2xs text-dim">{SPT.body(Math.max(shownTotal - explored, 0))}</p>
<p className="text-2xs text-dim">{SPT.next}</p>
</div>
<button type="button" className="btn btn-sm shrink-0" onClick={onGoSource} data-catalog-schema-partial-go="">
{CT.back}
</button>
</div>
)
}
/**
* 「用到但没确认」:运行中查询过的表里还有几张有推断项、几张没有目录。这些表的推断项会被助手当真用,最该先审。
* 两段各是一个按钮,点了切到清单的对应筛选、按使用次数排;都审完了只写一句,不给按钮
+36 -21
View File
@@ -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 (
<Shell color="var(--st-waiting)" data="catalog-drift">
<div className="flex items-start gap-2.5">
@@ -41,29 +44,41 @@ export function CatalogDriftBanner({ drift }: { drift: { title: string; tables:
<div className="min-w-0 flex-1 text-xs">
<p className="font-semibold" data-drift-title="">{drift.title}</p>
<p className="mt-0.5 text-2xs text-dim">{CATALOG_DRIFT_TEXT.sub}</p>
<ul className="mt-1 space-y-0.5 text-2xs text-dim">
{shown.map((t) => (
<li key={`${t.source}/${t.table}`} className="flex min-w-0 flex-wrap items-baseline gap-x-2" data-drift-table={t.table}>
<span className="min-w-0 break-words">
{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))}
</span>
{t.source_id && (
<Link className="shrink-0 text-[var(--accent)] hover:underline" data-drift-open={t.table}
to={`/data/catalog/${encodeURIComponent(t.source_id)}/${encodeURIComponent(t.table)}`}>
{CATALOG_DRIFT_TEXT.open}
</Link>
)}
</li>
))}
{drift.tables.length > shown.length && <li className="text-faint">{CATALOG_IMPACT_TEXT.more(drift.tables.length - shown.length)}</li>}
</ul>
{drift.direct.length > 0 && <DriftTables tables={drift.direct} />}
{drift.possible.length > 0 && (
<div className="mt-1.5" data-drift-possible={drift.possible.length}>
<p className="text-2xs font-medium text-dim">{CATALOG_DRIFT_TEXT.possibleHead}</p>
<p className="text-2xs text-faint">{CATALOG_DRIFT_TEXT.possibleHint}</p>
<DriftTables tables={drift.possible} />
</div>
)}
</div>
</div>
</Shell>
)
}
function DriftTables({ tables }: { tables: CatalogDriftTable[] }) {
const shown = tables.slice(0, DRIFT_SHOWN)
return (
<ul className="mt-1 space-y-0.5 text-2xs text-dim">
{shown.map((t) => (
<li key={`${t.source}/${t.table}`} className="flex min-w-0 flex-wrap items-baseline gap-x-2" data-drift-table={t.table}
data-drift-impact={t.impact}>
<span className="min-w-0 break-words">{catalogDriftLine(t)}</span>
{t.source_id && (
<Link className="shrink-0 text-[var(--accent)] hover:underline" data-drift-open={t.table}
to={`/data/catalog/${encodeURIComponent(t.source_id)}/${encodeURIComponent(t.table)}`}>
{CATALOG_DRIFT_TEXT.open}
</Link>
)}
</li>
))}
{tables.length > shown.length && <li className="text-faint">{CATALOG_IMPACT_TEXT.more(tables.length - shown.length)}</li>}
</ul>
)
}
const DRIFT_SHOWN = 6
// -------------------------------------------------------------------------
+45 -15
View File
@@ -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'
@@ -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<string, any>): { title: string; detail: string; tables: CatalogDriftTable[] } | null {
export function catalogDrift(d: Record<string, any>): {
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,
}
}
@@ -2585,8 +2605,10 @@ export interface CopilotContextSource {
tables: string[]
/** model:按需求挑的;all:库不大,全部表都带字段;fallback:没能挑出来,只给了表名 */
selectedBy: 'model' | 'all' | 'fallback'
/** 这个库一共几张表 */
/** 这个库一共几张表(数据库里的真实总数) */
total?: number
/** 探查结构截断了时,探查到几张(服务端只在截断时给):模型只看得到这几张 */
explored?: number
/** 没能挑出来的原因 */
reason?: string
}
@@ -2600,26 +2622,34 @@ function contextSources(op: CopilotOp): CopilotContextSource[] {
.filter((x): x is Record<string, any> => !!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}`
+19 -2
View File
@@ -56,12 +56,23 @@ export interface WorkflowVersion {
input_fields?: { name: string; required?: boolean; [key: string]: any }[]
/** 是不是当前的已发布版本 */
published?: boolean
/** 发布时这一版 SQL 用到的表的数据目录版本 {源名: {表名: 版本}},0 表示当时还没有目录;没记过为 null */
catalog_versions?: Record<string, Record<string, number>> | null
/** 发布时记下的数据目录版本,没记过为 null */
catalog_versions?: CatalogVersions | null
/** 发布之后目录有变化的表 */
catalog_changes?: CatalogDriftTable[]
}
/**
* 发布时记下的数据目录版本(服务端 catalog_impact.recorded_versions),两组都是 {源名: {表名: 版本}}:
* - direct:SQL 里写着的表,0 表示当时还没有目录
* - possible:Agent 绑定了查询工具的数据源中所有有目录的表(SQL 运行时才生成);源下为空表示当时一张有目录的
* 表都没有。只记了直接引用的老版本这一组为空
*/
export interface CatalogVersions {
direct: Record<string, Record<string, number>>
possible: Record<string, Record<string, number>>
}
/** 发布之后数据目录有变化的一张表(服务端 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'
}
/**
@@ -2658,6 +2671,10 @@ export interface CatalogList {
system_notes: boolean
/** 没有表结构时的原因(「尚未探查结构」「结构探查失败:…」);有表结构时为 null */
schema_note: string | null
/** 探查结构截断了(每个数据源最多取 200 张表):没探查到的表不在清单里,助手也看不到。老后端没有这一项 */
schema_truncated?: boolean
/** 数据库里一共几张表(含视图);没截断时等于探查到的张数 */
schema_total?: number
}
/** 表结构里的一列 */
+37 -1
View File
@@ -354,7 +354,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,
@@ -1225,6 +1229,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: [] }
+30
View File
@@ -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)
@@ -737,6 +747,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'))
})
+15 -4
View File
@@ -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 张表有变化'))
+22 -4
View File
@@ -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)