W3-a 表结构变化：前后差异、影响面、结构版本与 schema.drift（next-round 03 B1）

以前重新探查是整份替换表结构、不比前后：删了一列，那一列上人工确认过的目录项、写死这一列的已发布模板都收不到提醒，
正式运行要到查询那一步才报错。现在：

- data/schema_diff.py：纯函数，同一张表前后两次的结构比出变化（加列、删列、类型、可空、注释、主键与唯一约束、外键）；
  疑似改名（类型相同、夹在同样两列之间、只有一对）只说「可能」；读不到唯一约束、外键的一边不比；升级后第一次读到
  （结构版本 0）只建基线，不把通用写法换成方言完整写法说成类型变了。
- 后台探查顺带记变化（data/introspect_jobs.py）：刷新对象索引记新出现的表、真删除的表（第一次建索引只建基线）；开始
  执行时记离开范围的有目录的表、范围规则和上一次不同；逐批读完拿上一次的结构比，读不到结构的记 table_reflect_failed。
  离开范围、读不到结构不算删除。记变化出错只记日志，不挡探查。
- 变化记录 source_changes（db/models_sources.py 的 SourceChange，格式 agentlab-change/1，seq 自增不复用、按指纹去重），
  记录与读取在 data/source_changes.py；取值观察的表 TableObservation 同文件建好。
- 影响面（data/catalog_impact.structure_impact）：目录里说到被删、被改的列的项（列级项、业务主键、业务日期、有效记录条件
  按 parse_valid_filter 取列、两端的关联关系）带状态，人工确认过的升为要处理；已发布模板写死的 SQL 换上新结构再跑
  sql_unknown_columns，说得出「这条 SQL 现在会报列不存在」。sql_unknown_columns 从 api/copilot.py 挪到
  data/sqlcheck_columns.py 共用，copilot 那边只改 import。
- 发布时记 WorkflowVersion.schema_versions（只记直接引用、结构版本大于 0 的表）；正式运行发起时比对，变了写 schema.drift
  （core/events.py 登记，只提醒、不拦）；版本详情给出 schema_versions、schema_changes；发布前检查把这一版直接引用的表上
  还没处理的结构变化列成警告（schema.change_open），不拦发布。

测试：test_schema_diff（纯函数）、test_schema_changes（SQLite 夹具 ALTER 前后、影响面、去重、真删除与离开范围、
schema.drift、发布前检查、升级后基线）。

backend/app/api/copilot.py

@@ -25,6 +25,7 @@ from app.api.copilot_gen import (  # noqa: F401
 # 起别名：本文件底下有个叫 explain 的接口，generate 里又有个局部变量叫 raw，同名会互相盖掉
 from app.core.errors import explain as explain_error, graph_error, raw as raw_error
 from app.data.sqlcheck import CHECK_CODES
+from app.data.sqlcheck_columns import sql_unknown_columns
 from app.db.base import get_session
 from app.db.models import Workflow
 from app.engine.layout import CORRIDOR_MIN, MIN_ROW_GAP, NODE_W, _height, auto_layout
@@ -1347,80 +1348,9 @@ def _catalog_issues(node_id: str, sql: str, checker: Any) -> list[dict[str, Any]
 
 
 # --------------------------------------------------------------------------
-# SQL 里的列名对照数据源结构
-#
-# 只做有把握的判断，宁可漏判不可误判：这条会把图打回去让模型改。CTE、FROM 里的子查询、
-# 引号括起来的标识符、结构里找不到的表、结构没探查过或者探查被截断的，一律不判。
+# SQL 里的列名对照数据源结构（sql_unknown_columns）在 data/sqlcheck_columns.py：表结构变化的影响面也用它
 # --------------------------------------------------------------------------
 
-#: @ 开头的是变量（@region、@@session），: 开头的是参数，. 后面的是限定名
-_SQL_IDENT = re.compile(r"(?<![\w$#.:@])([A-Za-z_][A-Za-z0-9_$#]*)(?![\w$#])")
-_FROM_ITEMS = re.compile(
-    r"\b(?:from|join)\s+(.+?)(?=\bwhere\b|\bgroup\b|\border\b|\bhaving\b|\blimit\b|\bfetch\b|\boffset\b"
-    r"|\bunion\b|\bintersect\b|\bexcept\b|\bon\b|\busing\b|\b(?:left|right|inner|outer|full|cross|natural)\b"
-    r"|\bjoin\b|\bwindow\b|\)|$)", re.I | re.S)
-_TABLE_ITEM = re.compile(r"^\s*([A-Za-z_][\w$#]*(?:\.[A-Za-z_][\w$#]*)*)(?:\s+(?:as\s+)?([A-Za-z_][\w$#]*))?\s*$",
-                         re.I)
-_ALIAS_AS = re.compile(r"\bas\s+([A-Za-z_][\w$#]*)", re.I)
-#: 不写 AS 的列别名：「SUM(amount) total,」「amount total FROM」
-_ALIAS_BARE = re.compile(r"(\)|\b[A-Za-z_][\w$#]*|\d)\s+([A-Za-z_][\w$#]*)\s*(?=,|\bfrom\b)", re.I)
-_QUALIFIED = re.compile(r"(?<![\w$#.])([A-Za-z_][\w$#]*)\s*\.\s*([A-Za-z_][\w$#]*|\*)(?!\s*\()")
-_SQL_WORDS = frozenset("""
-select distinct all from where and or not in is null like ilike between exists case when then else end as on using
-join inner left right outer full cross natural group by order having limit offset fetch first next rows row only
-union intersect except asc desc nulls last with over partition range preceding following unbounded current true
-false interval year month day hour minute second week quarter dow doy epoch date time timestamp datetime zone
-cast integer int bigint smallint tinyint decimal numeric real float double precision varchar char character text
-boolean bool signed unsigned binary top percent escape collate any some values current_date current_time
-current_timestamp localtime localtimestamp sysdate systimestamp rownum level dual unknown filter within
-at nocase rtrim glob regexp rlike similar to div mod xor isnull notnull minus prior connect start nocycle siblings
-grouping sets rollup cube ignore respect lateral apply qualify window exclude include ties others no groups
-for update of share nowait skip locked lock mode array sounds overlaps language query expansion separator
-charset utf8 utf8mb4 latin1 ascii unicode years months weeks days hours minutes seconds microsecond microseconds
-millisecond milliseconds year_month day_hour day_minute day_second hour_minute hour_second minute_second
-isodow isoyear century decade millennium julian timezone timezone_hour timezone_minute
-sql_calc_found_rows sql_no_cache sql_cache sql_small_result sql_big_result sql_buffer_result high_priority
-straight_join distinctrow unique recursive tablesample bernoulli sample repeatable pivot unpivot keep ordinality
-symmetric asymmetric leading trailing both placing force use index into default without local
-user current_user session_user system_user current_schema current_catalog current_role
-""".split())
-
-#: 后面跟的一定是操作数、不会是别名的词：「SELECT created_at FROM」里 created_at 是列。
-#: END、NULL、TRUE、CURRENT_DATE 这类「一个值到此为止」的词不在里面——「… END flag,」的 flag 是别名
-_OPERAND_BEFORE = frozenset("""
-select distinct distinctrow all unique by as and or not xor is in like ilike glob regexp rlike similar to between
-when then else case on where having exists div mod escape interval collate using any some from join
-sql_calc_found_rows sql_no_cache sql_cache sql_small_result sql_big_result sql_buffer_result high_priority
-straight_join
-""".split())
-#: 不是列的名字：排序规则名（COLLATE x）、字符集名（USING x、CHARSET x、CHARACTER SET x）
-_NAME_AFTER = re.compile(
-    r"\b(?:collate|charset|character\s+set|using(?!\s*\())\s+([A-Za-z_][\w$#]*)", re.I)
-#: DATEADD(dd, …)、DATEDIFF(wk, …) 的第一个参数是日期部分的缩写，不是列。只在这个位置放过：
-#: 放进关键字表的话，别处打错成 wk、dd 的列名就查不出来了
-_DATEPART_ARG = re.compile(
-    r"\b(?:dateadd|datediff|datediff_big|datepart|datename|datetrunc|date_bucket|timestampadd|timestampdiff)"
-    r"\s*\(\s*([A-Za-z_][\w$#]*)", re.I)
-#: 命名窗口：WINDOW w AS (…)、…, w2 AS (…)
-_WINDOW_NAME = re.compile(r"(?:\bwindow|,)\s*([A-Za-z_][\w$#]*)\s+as\s*\(", re.I)
-
-
-def _blank_unicode_identifiers(sql: str) -> str:
-    """带引号、名字里有非 ASCII 字符的标识符（中文别名 AS "月份"、AS `曝光量`）连引号一起换成空格。
-
-    这项核对只认 ASCII 的列名，中文名本来就不判；不抹掉的话，SQL 里一出现带引号的标识符就整条放弃核对，照「结果列
-    起中文别名」写的 SQL 会把猜错的英文列名一起放过去。英文的带引号标识符（"StatDate"）大小写、空格都可能，照旧不判。"""
-    from app.data.guard import _GENERIC, _tokens
-
-    out = []
-    for token in _tokens(sql, _GENERIC):
-        if token.kind == "ident" and not token.text.isascii():
-            out.append(" " * len(token.text))
-        else:
-            out.append(token.text)
-    return "".join(out)
-
-
 def sql_source_prefix(sql: str, source: Any) -> list[str]:
     """SQL 里 FROM / JOIN 后面写成「数据源名.表名」的表（去重、按出现顺序）。
 
@@ -1444,66 +1374,6 @@ def sql_source_prefix(sql: str, source: Any) -> list[str]:
     return list(dict.fromkeys(f"{name}.{m.group(1)}" for m in pattern.finditer(text)))
 
 
-def sql_unknown_columns(sql: str, source: Any) -> tuple[list[str], list[str]] | None:
-    """SQL 里用到、数据源结构里查不到的列：(查不到的列, 这几张表的全部列)。判断不了、或者都查得到时返回 None。"""
-    from app.data import introspect as _introspect
-    from app.data.guard import _blank_quoted, first_verb, strip_comments
-
-    cache = getattr(source, "schema_cache", None) or {}
-    if not cache.get("tables") or cache.get("failed") or first_verb(sql) != "select":
-        return None
-    text = _blank_unicode_identifiers(re.sub(r"\{\{.*?\}\}", " 0 ", strip_comments(sql), flags=re.S))
-    if '"' in text.replace("''", "") and _blank_quoted(text) != _blank_quoted(text, identifiers=False):
-        return None                                   # 引号括起来的标识符：大小写、空格都可能，不猜
-    text = _blank_quoted(text)
-    if re.search(r"\bwith\b", text, re.I) or re.search(r"\b(?:from|join)\s*\(", text, re.I) or "`" in text \
-            or "[" in text:
-        return None                                   # [方括号] 标识符、数组下标：同样不猜
-    tables: dict[str, dict[str, Any]] = {}       # 表名、别名（小写）→ 表结构
-    spans: list[tuple[int, int]] = []
-    for m in _FROM_ITEMS.finditer(text):
-        spans.append(m.span(1))
-        for item in m.group(1).split(","):
-            parsed = _TABLE_ITEM.match(item)
-            if not parsed:
-                return None
-            name, alias = parsed.group(1), parsed.group(2)
-            meta = _introspect.find_table(source, name)
-            if meta is None:
-                return None                           # 结构里找不到的表（视图、同义词、截断了的结构）不判
-            tables[name.lower()] = tables[name.rsplit(".", 1)[-1].lower()] = meta
-            if alias and alias.lower() not in _SQL_WORDS:
-                tables[alias.lower()] = meta
-    if not tables:
-        return None
-    known = list(dict.fromkeys(c["name"] for meta in {id(m): m for m in tables.values()}.values()
-                               for c in meta.get("columns") or []))
-    columns = {c.lower() for c in known}
-    # 「SELECT created_at FROM」里 created_at 前面是关键字，它是列，不是别名
-    aliases = {a.lower() for a in _ALIAS_AS.findall(text)} | {
-        m.group(2).lower() for m in _ALIAS_BARE.finditer(text) if m.group(1).lower() not in _OPERAND_BEFORE} | {
-        a.lower() for a in _WINDOW_NAME.findall(text)}
-    names = {m.start(1) for pattern in (_NAME_AFTER, _DATEPART_ARG) for m in pattern.finditer(text)}
-    unknown: list[str] = []
-    for m in _QUALIFIED.finditer(text):
-        if any(a <= m.start() < b for a, b in spans):
-            continue
-        meta = tables.get(m.group(1).lower())
-        col = m.group(2)
-        if meta is not None and col != "*" and col.lower() not in {c["name"].lower() for c in meta.get("columns") or []}:
-            unknown.append(f"{m.group(1)}.{col}")
-    for m in _SQL_IDENT.finditer(text):
-        word = m.group(1)
-        low = word.lower()
-        if any(a <= m.start() < b for a, b in spans) or low in _SQL_WORDS or low in columns or low in aliases \
-                or low in tables or m.start() in names or re.match(r"\s*[.(]", text[m.end():]) \
-                or text[m.end():m.end() + 1] == "'":      # N'…'、X'…'、_utf8mb4'…'：字面量的前缀
-            continue
-        unknown.append(word)
-    unknown = list(dict.fromkeys(unknown))
-    return (unknown, known) if unknown else None
-
-
 def _source_of(tool: str) -> str | None:
     from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX
 

backend/app/api/workflows.py

@@ -12,7 +12,9 @@ from app.core.errors import graph_error
 from app.api.coded import CodedHTTPException
 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, recorded_versions
+from app.data.catalog_impact import (
+    catalog_drift, catalog_versions_for, open_structure_issues, recorded_versions, schema_drift, schema_versions_for,
+)
 from app.db.base import get_session
 from app.db.models import Run, Workflow, WorkflowVersion
 from app.engine.governance import publish_issues
@@ -345,6 +347,10 @@ class VersionDetailOut(VersionOut):
     catalog_versions: dict[str, Any] | None = None
     #: 发布之后目录有变化的表：[{source, source_id, table, label, published, current, impact}]
     catalog_changes: list[dict[str, Any]] = Field(default_factory=list)
+    #: 发布时记下的表结构版本 {源名: {表名: 结构版本}}（只记写死的 SQL 直接引用的表，next-round 03 B1）；记录之前发布的为 null
+    schema_versions: dict[str, Any] | None = None
+    #: 发布之后结构版本变了的表：[{source, source_id, table, label, published, current, removed, changes, more}]
+    schema_changes: list[dict[str, Any]] = Field(default_factory=list)
     #: 入口字段换成表单（agentlab.form/1）：发起表单、应用入口用同一个渲染器（X2）
     input_schema: dict[str, Any] = Field(default_factory=dict)
     #: 账本（L4）：固定这一版的应用、生效记录数、正式运行数
@@ -388,6 +394,8 @@ async def get_version(
         published=workflow.published_version == snapshot.version, level=snapshot.level,
         catalog_versions=recorded_versions(snapshot.catalog_versions),
         catalog_changes=await catalog_drift(session, snapshot.catalog_versions),
+        schema_versions=snapshot.schema_versions,
+        schema_changes=await schema_drift(session, snapshot.schema_versions),
         input_schema=input_schema(fields), published_at=snapshot.published_at, published_by=snapshot.published_by,
         retired_at=snapshot.retired_at, retired_by=snapshot.retired_by, retire_note=snapshot.retire_note,
         usage=await usage(session, workflow_id, snapshot.version),
@@ -535,7 +543,9 @@ async def gate_issues(session: AsyncSession, graph: Any, level: str) -> list[dic
     checkers = await _sql_checkers(session, spec)
     providers = await _providers(session, spec)
     with with_subgraphs(await load_subgraphs(session, spec)):
-        return [i.model_dump() for i in publish_issues(spec, level=level, checkers=checkers, providers=providers)]
+        issues = [i.model_dump() for i in publish_issues(spec, level=level, checkers=checkers, providers=providers)]
+    # 这一版直接引用的表上还没处理的结构变化（next-round 03 B1）：只是警告，不拦发布
+    return issues + await open_structure_issues(session, spec)
 
 
 async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: WorkflowVersion, level: str,
@@ -554,6 +564,8 @@ async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: Wo
     snapshot.level = level
     # 发布时的目录版本也记在这一版上：重新发布同一版等于按现在的目录重新认可它，跟着更新
     snapshot.catalog_versions = await catalog_versions_for(session, spec)
+    # 表的结构版本同理（next-round 03 B1）：正式运行发起时和现在的比，变了写 schema.drift，只提醒、不拦
+    snapshot.schema_versions = await schema_versions_for(session, spec)
     # 发布人、发布时间、用到的能力版本也记在这一版上（F5）：工作流上的 published_by 只记最近一次。退役过的版本重新
     # 发布等于重新认可它，退役信息清掉
     snapshot.published_at = utcnow()
@@ -673,7 +685,10 @@ async def check_graph(session: AsyncSession, graph: Any, level: str) -> dict[str
     checkers = await _sql_checkers(session, spec)
     providers = await _providers(session, spec)
     with with_subgraphs(await load_subgraphs(session, spec)):
-        return check(spec, level=level, versions=versions, cards=cards, checkers=checkers, providers=providers)
+        out = check(spec, level=level, versions=versions, cards=cards, checkers=checkers, providers=providers)
+    # 这一版直接引用的表上还没处理的结构变化（next-round 03 B1）：发布弹窗照别的警告一样列出来，不拦发布
+    out["issues"] = [*out["issues"], *await open_structure_issues(session, spec)]
+    return out
 
 
 @router.post("/{workflow_id}/autofix")

backend/app/core/events.py

@@ -77,6 +77,10 @@ class EventType(StrEnum):
     # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的。
     # 只提醒、不拦运行；落在封存范围内，出具物上看得到「当时的目录和发布时不一样」
     CATALOG_DRIFT = "catalog.drift"
+    # 从发布版本发起的正式运行：发布之后这一版直接引用的表的结构版本变了（data/catalog_impact.schema_drift，next-round
+    # 03 B1）。每张表带发布时和现在的结构版本、是否已删除、还没处理的结构变化。和 catalog.drift 分开记（老的渲染只认整数的
+    # 目录版本），界面上合到同一块横幅里。只提醒、不拦运行；落在封存范围内
+    SCHEMA_DRIFT = "schema.drift"
     # 报告撰写节点的自查结果：报告文档工件 id、统计、违规清单。落在封存范围内，
     # 证据接口靠它找到文档——不去翻不受封存保护的 artifacts 表
     REPORT_CHECKED = "report.checked"

backend/app/data/catalog_impact.py

@@ -10,6 +10,11 @@
   Agent、协作成员绑定了查询工具的源里所有有目录的表。从这一版发起正式运行时和当前的比（catalog_drift），有变化
   只提醒、不拦运行——目录是给人和助手看的说明，不是运行的输入；拦下来等于让一次文字修订卡住正式出具。
 
+- **表结构变化**（next-round 03 B1）：结构变化的影响面（structure_impact：目录里说到被删、被改的列的项和状态，人工确认过的
+  升为「要处理」；已发布模板写死的 SQL 换上新结构再跑 sql_unknown_columns，说得出「这条 SQL 现在会报列不存在」）；发布时记下
+  直接引用的表的结构版本（schema_versions_for，写法照 catalog_versions_for），正式运行发起时比对（schema_drift，变了写
+  schema.drift，只提醒、不拦）；发布前检查列出这一版用到的表上还没处理的结构变化（open_structure_issues，警告）。
+
 表名的取法和证据台账同一个口径：evidence.sql_tables 取 FROM / JOIN 后面的表（去掉注释、字符串、CTE），
 去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。哪些节点算「写死的查询」以
 sqlcheck.tool_query 为准（tool_queries），发布记录、影响面、SQL 检查和助手的上下文认的是同一种节点。
@@ -402,5 +407,325 @@ async def catalog_drift(session: AsyncSession, recorded: Mapping[str, Any] | Non
         return []
 
 
-__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"]
+# ==========================================================================
+# 表结构变化（next-round 03 B1）：影响面、发布时的结构版本、正式运行发起时比对、发布前检查
+# ==========================================================================
+
+#: 发布时记下的结构版本：{源名: {表名: 结构版本}}，只记写死的 SQL 直接引用的表（写法照 catalog_versions_for）
+SchemaVersions = dict[str, dict[str, int]]
+
+
+async def published_specs(session: AsyncSession) -> list[tuple[Any, Any, Any]]:
+    """每个发布过的模板、它当前的已发布版本、解析好的图：[(Workflow, WorkflowVersion, GraphSpec)]。图解析不了的跳过（发起
+    不了正式运行，也就谈不上影响）。一次后台探查里各张表的影响面共用这一份，不每张表重读、重解析一遍。"""
+    from app.engine.schema import GraphSpec
+
+    out: list[tuple[Any, Any, Any]] = []
+    for workflow, version in await _published(session):
+        try:
+            spec = GraphSpec.model_validate(version.graph or {})
+        except Exception:  # noqa: BLE001
+            continue
+        out.append((workflow, version, spec))
+    return out
+
+
+def _rel_columns(rel: Mapping[str, Any], side: str) -> set[str]:
+    return {str(c).lower() for c in rel.get(side) or []}
+
+
+def catalog_refs(notes_by_table: Mapping[str, Mapping[str, Any]], table: str, columns: set[str] | None, *,
+                 kind: str | None = None, before_meta: Mapping[str, Any] | None = None,
+                 source_kind: str | None = None) -> tuple[list[dict[str, Any]], bool]:
+    """目录里说到这张表这几列的项：([{table, path, label, status, source}], 有效记录条件用没用到它们)。
+
+    columns 为 None 是整张表（表删了）：这张表的每一项，外加别的表指向它的关联关系。否则是说到这几列的：
+    columns.<列>.*、业务主键、业务日期、有效记录条件（按 sqlcheck.parse_valid_filter 取它用到的列，按变化之前的结构解析）、
+    这张表的关联关系里用到这几列的，以及别的表指向这几列的关联关系。kind 是 fk_changed 时只看外键约束来的关联关系。
+    列名按不分大小写比。"""
+    from app.data import catalog, catalog_review, sqlcheck
+
+    own = notes_by_table.get(table) or {}
+    wanted = {c.lower() for c in columns} if columns is not None else None
+    out: list[dict[str, Any]] = []
+    rules_touched = False
+
+    def add(owner: str, key: tuple[str, ...], item: Mapping[str, Any]) -> None:
+        out.append({"table": owner, "path": catalog_review.slot_path(key), "label": catalog_review.item_label(key, item),
+                    "status": item.get("status"), "source": item.get("source")})
+
+    for key, item in catalog._slots(own).items():
+        if wanted is None:
+            add(table, key, item)
+            if key == ("t", "valid_filter"):
+                rules_touched = True
+            continue
+        if kind == "fk_changed":
+            if key[0] == "r" and item.get("source") == "fk" and _rel_columns(item, "columns") & wanted:
+                add(table, key, item)
+            continue
+        hit = False
+        if key[0] == "c":
+            hit = key[1].lower() in wanted
+        elif key == ("t", "keys"):
+            hit = any(str(c).lower() in wanted for c in item.get("value") or [])
+        elif key == ("t", "business_date"):
+            hit = str((item.get("value") or {}).get("column") or "").lower() in wanted
+        elif key == ("t", "valid_filter"):
+            parsed = sqlcheck.parse_valid_filter(item.get("value"), before_meta, kind=source_kind, table=table)
+            hit = any(c.lower() in wanted for c in parsed.columns)
+            rules_touched = rules_touched or hit
+        elif key[0] == "r":
+            hit = bool(_rel_columns(item, "columns") & wanted)
+        if hit:
+            add(table, key, item)
+    target = name_key(table)
+    for other, notes in notes_by_table.items():
+        if other == table:
+            continue
+        for key, item in catalog._slots(notes).items():
+            if key[0] != "r" or name_key(str(item.get("to_table") or "").rsplit(".", 1)[-1]) != target:
+                continue
+            if kind == "fk_changed":
+                continue
+            if wanted is None or _rel_columns(item, "to_columns") & wanted:
+                add(other, key, item)
+    return out, rules_touched
+
+
+def _short(column: str) -> str:
+    return str(column).rsplit(".", 1)[-1].lower()
+
+
+def template_hits(specs: list[tuple[Any, Any, Any]], source_name: str, table: str, *,
+                  tables_after: Mapping[str, Any] | None, removed_columns: set[str] | None,
+                  table_removed: bool = False) -> list[dict[str, Any]]:
+    """已发布模板里用到这张表的：[{workflow_id, name, version, level, impact, nodes: [{node_id, label, type, impact,
+    breaks, unknown_columns?}]}]，直接引用的在前。
+
+    直接引用（写死的 SQL）的那几个节点，拿新结构（tables_after）再跑一遍 sql_unknown_columns（data/sqlcheck_columns）：报出的
+    列里有被删的列（removed_columns），就是「这条 SQL 现在会报列不存在」（breaks）。表真删了，直接引用的节点都 breaks。
+    判断不了的 SQL（CTE、引号标识符……）不说会坏。"""
+    from types import SimpleNamespace
+
+    from app.data.sqlcheck_columns import sql_unknown_columns
+
+    removed = {c.lower() for c in removed_columns or set()}
+    known = list(tables_after) if tables_after is not None else None
+    probe = SimpleNamespace(name=source_name, schema_cache={"tables": dict(tables_after or {})})
+    out: list[dict[str, Any]] = []
+    for workflow, version, spec in specs:
+        refs = template_refs(spec, source_name, table, known if not table_removed else None)
+        if not refs:
+            continue
+        sqls = {getattr(node, "id", None): sql for node, name, sql in tool_queries(getattr(spec, "nodes", []) or [])
+                if name == source_name}
+        nodes: list[dict[str, Any]] = []
+        for ref in refs:
+            entry = {k: ref[k] for k in ("node_id", "label", "type", "impact")}
+            entry["breaks"] = False
+            if ref["impact"] == "direct" and ref["node_id"] in sqls:
+                if table_removed:
+                    entry["breaks"] = True
+                elif removed:
+                    found = sql_unknown_columns(sqls[ref["node_id"]], probe)
+                    hit = [c for c in (found[0] if found else []) if _short(c) in removed]
+                    if hit:
+                        entry.update(breaks=True, unknown_columns=hit)
+            nodes.append(entry)
+        level = version.level or (workflow.status if workflow.status in ("published", "governed") else "published")
+        out.append({"workflow_id": workflow.id, "name": workflow.name, "version": version.version, "level": level,
+                    "impact": "direct" if any(n["impact"] == "direct" for n in nodes) else "possible", "nodes": nodes})
+    out.sort(key=lambda t: (t["impact"] != "direct", t["name"]))
+    return out
+
+
+async def structure_impact(session: AsyncSession, row: Any, table: str, diffs: list[Any], *,
+                           before_meta: Mapping[str, Any] | None, tables_after: Mapping[str, Any] | None,
+                           specs: list[tuple[Any, Any, Any]] | None = None,
+                           notes_by_table: Mapping[str, Mapping[str, Any]] | None = None,
+                           ) -> list[tuple[dict[str, Any], str]]:
+    """一张表的每条结构变化（schema_diff.Diff，或者表级的那几种）→ (影响面, 严重程度)，和 diffs 一一对应。
+
+    影响面 {catalog: [{table, path, label, status, source, needs_action}], templates: [...], rules_touched}：
+    - 删了表、删了列、类型或可空变了：目录里说到它们的项（catalog_refs），已发布模板里用到这张表的（template_hits）；
+    - 外键变了：外键约束来的关联关系；其余（加列、注释、离开范围、读不到结构）没有影响面。
+    严重程度：删了表、删了列时，人工确认过的项还说着它（needs_action）、或者写死的 SQL 现在会报列不存在，是 action（要处理）；
+    否则按种类的缺省（schema_diff.DEFAULT_SEVERITY）。读库出错只记日志、影响面为空：影响面是提示，不挡探查。"""
+    from app.data import catalog
+    from app.data import schema_diff as sd
+
+    try:
+        if notes_by_table is None:
+            notes_by_table = {t: e.notes for t, e in (await catalog.read_catalog(session, row.id)).items()}
+        if specs is None:
+            specs = await published_specs(session)
+    except Exception:  # noqa: BLE001
+        logger.exception("计算 %s 的结构变化影响面时读库出错，本次不给影响面", table)
+        return [({"catalog": [], "templates": [], "rules_touched": False}, d.severity) for d in diffs]
+    out: list[tuple[dict[str, Any], str]] = []
+    for d in diffs:
+        catalog_items: list[dict[str, Any]] = []
+        templates: list[dict[str, Any]] = []
+        rules = False
+        destructive = d.kind in sd.DESTRUCTIVE
+        if d.kind == sd.TABLE_REMOVED:
+            catalog_items, rules = catalog_refs(notes_by_table, table, None, before_meta=before_meta,
+                                                source_kind=getattr(row, "kind", None))
+            templates = template_hits(specs, row.name, table, tables_after=tables_after, removed_columns=None,
+                                      table_removed=True)
+        elif d.kind in (sd.COLUMN_REMOVED, sd.COLUMN_TYPE, sd.COLUMN_NULLABLE, sd.FK) and d.column:
+            catalog_items, rules = catalog_refs(notes_by_table, table, {d.column}, kind=d.kind, before_meta=before_meta,
+                                                source_kind=getattr(row, "kind", None))
+            if d.kind in (sd.COLUMN_REMOVED, sd.COLUMN_TYPE):
+                templates = template_hits(specs, row.name, table, tables_after=tables_after,
+                                          removed_columns={d.column} if d.kind == sd.COLUMN_REMOVED else None)
+        for item in catalog_items:
+            item["needs_action"] = destructive and item.get("status") == "confirmed"
+        breaks = any(n.get("breaks") for t in templates for n in t["nodes"])
+        severity = "action" if destructive and (breaks or any(i["needs_action"] for i in catalog_items)) else d.severity
+        out.append(({"catalog": catalog_items, "templates": templates, "rules_touched": rules}, severity))
+    return out
+
+
+async def _index_rows(session: AsyncSession, row: Any, names: Iterable[str]) -> dict[str, Any]:
+    """对象索引里这个源（配置的 schema 下）这几张表的行：{name_key: 行}。"""
+    from app.data import source_objects
+    from app.db.models_sources import SourceObject
+
+    keys = {name_key(str(n).rsplit(".", 1)[-1]) for n in names}
+    if not keys:
+        return {}
+    rows = (await session.execute(select(SourceObject).where(
+        SourceObject.source_id == row.id, SourceObject.schema_name == source_objects.index_schema(row),
+        SourceObject.name_key.in_(list(keys))))).scalars()
+    return {r.name_key: r for r in rows}
+
+
+async def schema_versions_for(session: AsyncSession, spec: Any) -> SchemaVersions | None:
+    """发布时要记下的结构版本：{源名: {表名: 结构版本}}，只记调用工具节点写死的 SQL 直接引用的表（写法照
+    catalog_versions_for）。对象索引里没有的表、结构版本还是 0 的表（升级后还没按对象索引读过）不记：不知道那时是哪一版，
+    之后不能说它变了。读库出错返回 None（和没记过一样，不提醒）——记不下版本不该挡住发布。"""
+    try:
+        sources = await _sources_by_name(session, graph_sql_tables(spec).keys())
+        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: SchemaVersions = {}
+        for name, tables in sorted(used.items()):
+            if name not in sources:
+                continue
+            found = await _index_rows(session, sources[name][0], tables)
+            pinned = {t: int(found[name_key(t)].struct_version) for t in sorted(tables)
+                      if name_key(t) in found and (found[name_key(t)].struct_version or 0) > 0}
+            if pinned:
+                out[name] = pinned
+        return out
+    except Exception:  # noqa: BLE001
+        logger.exception("记录发布时的结构版本失败，本次不记")
+        return None
+
+
+async def _pending_structure(session: AsyncSession, source_id: str, table: str) -> list[Any]:
+    from app.data import schema_diff as sd
+    from app.data.source_changes import PENDING
+    from app.db.models_sources import SourceChange
+
+    return list((await session.execute(select(SourceChange).where(
+        SourceChange.source_id == source_id, SourceChange.table_name == table,
+        SourceChange.kind.in_(sd.STRUCT_KINDS), SourceChange.status.in_(PENDING))
+        .order_by(SourceChange.seq))).scalars())
+
+
+#: 漂移里每张表最多列几条没处理的变化
+DRIFT_CHANGES_SHOWN = 5
+
+
+async def schema_drift(session: AsyncSession, recorded: Mapping[str, Any] | None) -> list[dict[str, Any]]:
+    """发布时记下的结构版本和现在比：变了的表 [{source, source_id, table, label, published, current, removed, changes,
+    more}]。current 是现在的结构版本（表已删除时为 None、removed 为真）；changes 是这张表上在发布之后测到、还没处理的结构
+    变化（[{id, kind, column, severity, summary}]，最多 DRIFT_CHANGES_SHOWN 条，more 是另外还有几条）。
+
+    没记过（加这一列之前发布的版本）返回空列表；现在找不到的源跳过；对象索引里找不到的表跳过（换了 schema、索引重建）：
+    不知道它现在是哪一版，不能说它变了。读库出错返回空列表：提醒是附加的。"""
+    from app.data import catalog, source_changes
+
+    if not isinstance(recorded, Mapping) or not recorded:
+        return []
+    try:
+        sources = await _sources_by_name(session, [str(k) for k in recorded])
+        out: list[dict[str, Any]] = []
+        for name in sorted(recorded):
+            pinned = recorded.get(name)
+            if name not in sources or not isinstance(pinned, Mapping):
+                continue
+            row = sources[name][0]
+            found = await _index_rows(session, row, pinned)
+            entries = None
+            for table in sorted(pinned):
+                then = pinned[table]
+                obj = found.get(name_key(table))
+                if obj is None or not isinstance(then, int):
+                    continue
+                removed = obj.removed_at is not None
+                now = None if removed else int(obj.struct_version or 0)
+                if not removed and now == then:
+                    continue
+                pending = [c for c in await _pending_structure(session, row.id, obj.name)
+                           if (c.struct_version_after or 0) > then or c.kind == "table_removed"]
+                if entries is None:
+                    entries = await catalog.read_catalog(session, row.id)
+                out.append({"source": name, "source_id": row.id, "table": obj.name,
+                            "label": _label(entries.get(obj.name)), "published": then, "current": now,
+                            "removed": removed,
+                            "changes": [{"id": c.id, "kind": c.kind, "column": c.column_name, "severity": c.severity,
+                                         "summary": source_changes.summary(c)} for c in pending[:DRIFT_CHANGES_SHOWN]],
+                            "more": max(0, len(pending) - DRIFT_CHANGES_SHOWN)})
+        return out
+    except Exception:  # noqa: BLE001
+        logger.exception("比对发布时的结构版本失败，本次不提醒")
+        return []
+
+
+#: 发布前检查里「这张表上还有没处理的结构变化」的机读码
+SCHEMA_CHANGE_OPEN = "schema.change_open"
+
+
+async def open_structure_issues(session: AsyncSession, spec: Any) -> list[dict[str, Any]]:
+    """发布前检查要列的：这一版写死的 SQL 直接引用的表上，还没处理的结构变化（警告、要处理两级；说明性的不列）。每张表
+    一条警告，挂在第一个直接引用它的节点上，不拦发布：结构变了不一定影响这条 SQL，真删了列查询本身会照实报错。读库出错
+    返回空列表。"""
+    from app.data import source_changes
+
+    try:
+        sources = await _sources_by_name(session, graph_sql_tables(spec).keys())
+        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})
+        issues: list[dict[str, Any]] = []
+        for name, tables in sorted(used.items()):
+            if name not in sources:
+                continue
+            row = sources[name][0]
+            for table in sorted(tables):
+                pending = [c for c in await _pending_structure(session, row.id, table)
+                           if c.severity in ("warning", "action")]
+                if not pending:
+                    continue
+                refs = [r for r in template_refs(spec, name, table, tables_by_source.get(name)) if r["impact"] == "direct"]
+                shown = "；".join(source_changes.summary(c) for c in pending[:3])
+                more = f"等 {len(pending)} 条" if len(pending) > 3 else ""
+                issues.append({
+                    "level": "warning", "node_id": refs[0]["node_id"] if refs else None, "edge_id": None, "field": None,
+                    "code": SCHEMA_CHANGE_OPEN, "fix": None,
+                    "message": (f"数据源「{name}」的表 {table} 有还没处理的结构变化：{shown}{more}。这一版写死的 SQL 用到了"
+                                "这张表，发布前请核对这几处 SQL 是否还对（不拦发布）"),
+                })
+        return issues
+    except Exception:  # noqa: BLE001
+        logger.exception("发布前检查结构变化时出错，本次不列")
+        return []
+
+
+__all__ = ["IMPACTS", "SCHEMA_CHANGE_OPEN", "CatalogVersions", "SchemaVersions", "agent_bindings", "agent_sources",
+           "catalog_drift", "catalog_refs", "catalog_versions_for", "graph_sql_tables", "open_structure_issues",
+           "published_specs", "recorded_versions", "schema_drift", "schema_versions_for", "sql_table_keys",
+           "structure_impact", "table_impact", "template_hits", "template_refs", "tool_queries"]

backend/app/data/introspect_jobs.py

@@ -19,6 +19,12 @@
 - **保留**（prune）：每个源保留最近 KEEP_JOBS 个结束了的探查任务。
 
 任务最后的状态：有表读不到是 partial，否则 succeeded；整体出错（数据源没了、连不上）是 failed。
+
+**结构变化**（第 3 波，next-round 03 B1）：探查顺带把前后差异记进 source_changes（data/source_changes.py），每条带影响面
+（data/catalog_impact.structure_impact）。刷新对象索引时记新出现的表、真删除的表（第一次建索引只建基线）；开始执行时记
+离开范围的、有数据目录的表，以及范围规则和上一次探查不同（scope_changed）；逐批读完时拿上一次的结构和这一次的比
+（data/schema_diff.diff_table），在范围里却读不到结构的记 table_reflect_failed（不算删除）。升级后第一次读到（结构版本 0）
+只建基线。记变化出错只记日志，不挡探查。任务上记 trigger（manual 人点的 / schedule 定时维护）和定时的 fire_at。
 """
 from __future__ import annotations
 
@@ -271,6 +277,7 @@ async def preview(session: AsyncSession, row: Any, *, schema: str | None = None,
         return out
     async with so.source_lock(row.id):
         stats = await so.refresh_index(session, row.id, configured, result, legacy=row.schema_cache)
+        await _record_index_changes(session, row, stats, _detected(None, "manual", None))
         await session.commit()
     plan = await so.plan_for(session, row, rules)
     out.update(stored=True, objects=[o.name for o in result.objects], scope=plan.summary(), added=stats.added,
@@ -278,9 +285,10 @@ async def preview(session: AsyncSession, row: Any, *, schema: str | None = None,
     return out
 
 
-async def prepare(session: AsyncSession, row: Any, *, actor: str | None,
-                  trigger: str = "manual") -> tuple[IntrospectJob, int]:
-    """建一个排队的探查任务（还没开始，调用方接着 manager.start）。返回 (任务, 几张表)。
+async def prepare(session: AsyncSession, row: Any, *, actor: str | None, trigger: str = "manual",
+                  fire_at: str | None = None, schedule_id: str | None = None) -> tuple[IntrospectJob, int]:
+    """建一个排队的探查任务（还没开始，调用方接着 manager.start）。返回 (任务, 几张表)。trigger 是 manual（人点的）或
+    schedule（定时维护：fire_at、schedule_id 记进任务的 scope，留痕这一次是哪一次触发建的）。
 
     上传表格、记录表抛 IntrospectRefused；同源已有排队或进行中的探查任务抛 IntrospectBusy；对象清单拿不到抛
     IntrospectFailed（失败照实记进 schema_cache）；规则命中超过上限抛 ScopeTooLarge（对象索引照样刷新了）。
@@ -306,6 +314,7 @@ async def prepare(session: AsyncSession, row: Any, *, actor: str | None,
         rules = so.ScopeRules.from_options(row.options)
         plan = await so.plan_for(session, row, rules)
         if plan.over:
+            await _record_index_changes(session, row, stats, _detected(None, trigger, actor, fire_at))
             await session.commit()
             raise ScopeTooLarge(plan)
         summary = {**plan.summary(rules), "index_method": stats.method, "index_capped": stats.capped,
@@ -313,6 +322,10 @@ async def prepare(session: AsyncSession, row: Any, *, actor: str | None,
                    "adopted": stats.adopted}
         if candidates:
             summary["available_schemas"] = candidates
+        if fire_at:
+            summary["fire_at"] = fire_at
+        if schedule_id:
+            summary["schedule_id"] = schedule_id
         job = IntrospectJob(id=new_id(), source_id=source_id, status="queued", trigger=trigger, schema_name=schema,
                             scope=summary, created_by=actor)
         session.add(job)
@@ -323,6 +336,7 @@ async def prepare(session: AsyncSession, row: Any, *, actor: str | None,
             raise IntrospectBusy(busy_message(SimpleNamespace(name=name), None)) from e
         session.add_all([IntrospectTask(job_id=job.id, seq=i, table_name=table, reason=reason)
                          for i, (table, reason) in enumerate(plan.in_scope.items())])
+        await _record_index_changes(session, row, stats, _detected(job.id, trigger, actor, fire_at))
         await session.commit()
     try:
         await prune(session, source_id)
@@ -345,6 +359,93 @@ async def prune(session: AsyncSession, source_id: str, *, keep: int = KEEP_JOBS)
     return len(victims)
 
 
+# ==========================================================================
+# 结构变化（next-round 03 B1）
+# ==========================================================================
+
+#: 一次刷新对象索引最多记几条「新出现的表」（几千个按月新建的分区表这类：照实记前面的，其余只进日志）
+ADDED_RECORDED = 200
+
+
+def _detected(job_id: str | None, trigger: str, actor: str | None, fire_at: str | None = None,
+              task_seq: int | None = None) -> Any:
+    from app.data.source_changes import Detected
+
+    return Detected(job_kind="introspect", job_id=job_id, task_seq=task_seq, trigger=trigger or "manual",
+                    actor=actor, fire_at=fire_at)
+
+
+async def _record_index_changes(session: AsyncSession, row: Any, stats: so.RefreshStats, detected: Any) -> None:
+    """刷新对象索引得出的表级变化（不提交）：新出现的表（table_added；这个 schema 第一次建索引时全是新的，只建基线）、
+    真删除的表（table_removed，带影响面：目录里这张表的项、别的表指向它的关联关系、写死它的已发布模板）。出错只记日志。"""
+    if not stats.removed_names and not (stats.added_names and not stats.first_build):
+        return
+    from app.data import catalog_impact
+    from app.data import schema_diff as sd
+    from app.data import source_changes as sc
+
+    try:
+        now = utcnow().isoformat()
+        changes: list[sc.NewChange] = []
+        added = [] if stats.first_build else list(stats.added_names)
+        if len(added) > ADDED_RECORDED:
+            logger.warning("数据源 %s 这次新出现 %d 个对象，变化记录只记前 %d 个", row.name, len(added), ADDED_RECORDED)
+        for name in added[:ADDED_RECORDED]:
+            changes.append(sc.NewChange(kind=sd.TABLE_ADDED, table=name, severity="info", key=now))
+        if stats.removed_names:
+            schema = so.index_schema(row)
+            gone = {r.name: r for r in (await session.execute(select(SourceObject).where(
+                SourceObject.source_id == row.id, SourceObject.schema_name == schema,
+                SourceObject.name.in_(stats.removed_names)))).scalars()}
+            tables_after = dict((row.schema_cache or {}).get("tables") or {})
+            specs = await catalog_impact.published_specs(session)
+            for name in stats.removed_names:
+                obj = gone.get(name)
+                version = int(obj.struct_version or 0) if obj is not None else 0
+                meta = obj.meta if obj is not None and isinstance(obj.meta, dict) else None
+                [(affects, severity)] = await catalog_impact.structure_impact(
+                    session, row, name, [sd.Diff(sd.TABLE_REMOVED)], before_meta=meta,
+                    tables_after={k: v for k, v in tables_after.items() if k != name}, specs=specs)
+                changes.append(sc.NewChange(
+                    kind=sd.TABLE_REMOVED, table=name, severity=severity, affects=affects,
+                    before={"kind": obj.kind if obj is not None else "table"}, struct_version_before=version or None,
+                    key=(obj.removed_at.isoformat() if obj is not None and obj.removed_at else now)))
+        await sc.record_all(session, row, changes, detected)
+    except Exception:  # noqa: BLE001 - 记变化出错不挡探查
+        logger.exception("数据源 %s 记录表级的结构变化时出错", getattr(row, "name", "?"))
+
+
+async def _record_scope_changes(session: AsyncSession, row: Any, job: IntrospectJob, leaving: list[str],
+                                entering: int) -> None:
+    """开始执行、把范围写进对象索引时（不提交）：有数据目录的表离开了范围（table_left_scope，不算删除），和范围规则跟上一次
+    探查不同（scope_changed：前后的规则、进出各几张）。出错只记日志。"""
+    from app.data import catalog
+    from app.data import schema_diff as sd
+    from app.data import source_changes as sc
+
+    try:
+        detected = _detected(job.id, job.trigger, job.created_by, (job.scope or {}).get("fire_at"))
+        changes: list[sc.NewChange] = []
+        if leaving:
+            noted = set((await catalog.read_catalog(session, row.id)).keys())
+            for name in leaving:
+                if name in noted:
+                    changes.append(sc.NewChange(kind=sd.TABLE_LEFT_SCOPE, table=name, severity="info", key=job.id))
+        previous = (await session.execute(
+            select(IntrospectJob).where(IntrospectJob.source_id == job.source_id, IntrospectJob.id != job.id,
+                                        IntrospectJob.started_at.is_not(None),
+                                        IntrospectJob.created_at <= job.created_at)
+            .order_by(IntrospectJob.created_at.desc()).limit(1))).scalar_one_or_none()
+        before = (previous.scope or {}).get("rules") if previous is not None else None
+        after = (job.scope or {}).get("rules")
+        if isinstance(before, dict) and isinstance(after, dict) and before != after:
+            changes.append(sc.NewChange(kind=sd.SCOPE, table="", severity="info", before={"rules": before},
+                                        after={"rules": after, "entered": entering, "left": len(leaving)}, key=job.id))
+        await sc.record_all(session, row, changes, detected)
+    except Exception:  # noqa: BLE001
+        logger.exception("数据源 %s 记录探查范围的变化时出错", getattr(row, "name", "?"))
+
+
 # ==========================================================================
 # 收尾、停止、启动时的整理
 # ==========================================================================
@@ -425,12 +526,32 @@ async def startup() -> int:
 # ==========================================================================
 
 
+def _side(diff: Any, which: str) -> Any:
+    """一条差异的前（before）或后（after），疑似改名的另一半记在这一侧（{…, rename}）。"""
+    value = getattr(diff, which)
+    if diff.rename:
+        base = dict(value) if isinstance(value, dict) else ({"value": value} if value is not None else {})
+        return {**base, "rename": dict(diff.rename)}
+    return value
+
+
+def json_key(value: Any) -> str:
+    import json
+
+    return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str)
+
+
 class _IntrospectRun(runtime.JobRun):
     """一个探查任务在后台的一次执行（进队、停止、收尾的框架在 runtime.JobRun）。"""
 
     kind = "introspect"
     model = IntrospectJob
 
+    def __init__(self, job_id: str) -> None:
+        super().__init__(job_id)
+        #: 已发布模板和解析好的图（catalog_impact.published_specs）：这次探查里各批的影响面共用，第一次用到时读
+        self._specs: list[Any] | None = None
+
     async def _stopped(self) -> None:
         if self.stop_reason == "user":
             await self._end("cancelled", None, STOPPED_ERROR)
@@ -510,8 +631,15 @@ class _IntrospectRun(runtime.JobRun):
             counts = {r: 0 for r in so.SCOPE_REASONS}
             for reason in reasons.values():
                 counts[reason] = counts.get(reason, 0) + 1
+            before = set((await session.execute(select(SourceObject.name).where(
+                SourceObject.source_id == source_id, SourceObject.schema_name == schema,
+                SourceObject.in_scope.is_(True)))).scalars())
             plan = so.ScopePlan(in_scope=reasons, counts=counts, cap=0, index_total=0, out_of_scope=0)
             await so.apply_scope(session, source_id, schema, plan)
+            row = await session.get(DataSource, source_id)
+            if job is not None and row is not None:
+                await _record_scope_changes(session, row, job, sorted(before - set(reasons)),
+                                            len(set(reasons) - before))
             await session.commit()
 
     async def _recheck(self, session: AsyncSession, source_id: str) -> Any:
@@ -540,8 +668,15 @@ class _IntrospectRun(runtime.JobRun):
 
     async def _save_batch(self, session: AsyncSession, source_id: str, schema: str, batch: list[tuple[int, str]],
                           got: so.Reflected) -> None:
-        """一批读完：读到的结构写进对象索引，逐表记 ok / failed 和结构版本。"""
+        """一批读完：读到的结构写进对象索引，逐表记 ok / failed 和结构版本；拿上一次的结构和这一次的比，变化记进
+        source_changes（同一个事务）。"""
+        old = {r.name: r for r in (await session.execute(
+            select(SourceObject.name, SourceObject.meta, SourceObject.struct_version, SourceObject.fingerprint,
+                   SourceObject.reflect_error).where(
+                SourceObject.source_id == source_id, SourceObject.schema_name == schema,
+                SourceObject.name.in_([name for _, name in batch])))).all()}
         versions = await so.save_reflected(session, source_id, schema, got)
+        await self._record_diffs(session, source_id, batch, got, old, versions)
         now = utcnow()
         for seq, name in batch:
             if name in got.tables:
@@ -554,6 +689,60 @@ class _IntrospectRun(runtime.JobRun):
                                   .execution_options(synchronize_session=False))
         await session.commit()
 
+    async def _record_diffs(self, session: AsyncSession, source_id: str, batch: list[tuple[int, str]],
+                            got: so.Reflected, old: dict[str, Any], versions: dict[str, int]) -> None:
+        """这一批的结构差异（不提交）：读到的表指纹变了、又不是基线（schema_diff.is_baseline）的，逐条比出变化、算影响面；
+        以前读得到、这次读不到的记 table_reflect_failed。出错只记日志，不挡探查。"""
+        from app.data import catalog_impact
+        from app.data import schema_diff as sd
+        from app.data import source_changes as sc
+
+        try:
+            todo: list[tuple[int, str, list[Any], dict[str, Any] | None]] = []
+            failed: list[tuple[int, str]] = []
+            for seq, name in batch:
+                prev = old.get(name)
+                if prev is None:
+                    continue
+                if name in got.tables:
+                    meta = got.tables[name]
+                    if sd.is_baseline(prev.struct_version, prev.meta) or so.fingerprint(meta) == prev.fingerprint:
+                        continue
+                    diffs = sd.diff_table(prev.meta, meta)
+                    if diffs:
+                        todo.append((seq, name, diffs, prev.meta))
+                elif name in got.skipped and isinstance(prev.meta, dict) and prev.meta and not prev.reflect_error:
+                    failed.append((seq, name))
+            if not todo and not failed:
+                return
+            job = await session.get(IntrospectJob, self.job_id)
+            row = await session.get(DataSource, source_id)
+            if job is None or row is None:
+                return
+            fire_at = (job.scope or {}).get("fire_at")
+            if self._specs is None:
+                self._specs = await catalog_impact.published_specs(session)
+            tables_after = {**dict((row.schema_cache or {}).get("tables") or {}), **got.tables}
+            for seq, name, diffs, before in todo:
+                impacts = await catalog_impact.structure_impact(session, row, name, diffs, before_meta=before,
+                                                                tables_after=tables_after, specs=self._specs)
+                after_v, before_v = versions.get(name), old[name].struct_version
+                changes = [sc.NewChange(kind=d.kind, table=name, column=d.column, severity=severity,
+                                        before=_side(d, "before"), after=_side(d, "after"), affects=affects,
+                                        struct_version_before=before_v, struct_version_after=after_v,
+                                        key=(after_v, d.key, json_key(d.before), json_key(d.after)))
+                           for d, (affects, severity) in zip(diffs, impacts)]
+                await sc.record_all(session, row, changes,
+                                    _detected(self.job_id, job.trigger, job.created_by, fire_at, seq))
+            for seq, name in failed:
+                await sc.record(session, row, sc.NewChange(
+                    kind=sd.TABLE_REFLECT_FAILED, table=name, severity="warning",
+                    after={"error": got.skipped.get(name)}, struct_version_before=old[name].struct_version,
+                    key=(old[name].struct_version, got.skipped.get(name))),
+                    _detected(self.job_id, job.trigger, job.created_by, fire_at, seq))
+        except Exception:  # noqa: BLE001
+            logger.exception("探查任务 %s 记录结构变化时出错", self.job_id)
+
     async def _finish(self, session: AsyncSession, source_id: str) -> None:
         """读完：物化 schema_cache、盖同步时间；有表读不到是 partial，否则 succeeded。"""
         failed = int((await session.execute(select(func.count()).select_from(IntrospectTask).where(

backend/app/data/schema_diff.py

@@ -0,0 +1,216 @@
+"""表结构的前后差异（docs/design/next-round/03-数据接入与目录维护.md B1）：纯函数，不连库、不读目录。
+
+**为什么要有它。** 以前重新探查是整份替换 schema_cache，不比前后：删了一列，那一列上人工确认过的目录项、写死了这一列的
+已发布模板都收不到提醒，正式运行要到查询那一步才报错。现在每张表有结构版本（data/source_objects：指纹变了才加 1），
+后台探查读到新结构时拿上一次的结构和这一次的比（diff_table），变化写进 source_changes（data/source_changes.py），再由
+影响面（data/catalog_impact.structure_impact）说清哪些目录项、哪些写死的 SQL 会受影响。
+
+**表级**的四种由调用方判（它知道对象索引和范围）：table_added（对象索引里新出现的）、table_removed（对象索引里也没了，
+真删除）、table_left_scope（离开了探查范围）、table_reflect_failed（在范围里、这次读不到结构）。后两种**不算删除**
+（is_deletion）：范围外、读不到结构时，目录和模板里说到它的照旧有效，只是这次没看到。
+
+**列级**按列名比：column_added、column_removed、column_type_changed（比的是按方言编译的完整写法，VARCHAR2(20 CHAR)、
+ENUM('a','b')，见 introspect.type_text）、column_nullable_changed、comment_changed（列注释，表注释的 column 为 None）。
+**键**：主键、唯一约束的增删（key_changed），外键的增删（fk_changed）。反射时读不到唯一约束、外键的表不写那两个键：
+两边有一边没有就不比（不知道 ≠ 删了）。
+
+**疑似改名**：被删的列和新加的列类型相同、在列清单里的位置相同（夹在同样的两列之间），而且这样的配对只有一对时，两条变化都带 rename，
+说法只写「可能是改名（未确认）」——库里只看得到删了一列、加了一列，不下结论。
+
+**基线**（is_baseline）：升级后第一次按对象索引读到这张表（结构版本还是 0：升级前同步探查的老缓存接进来的那份、或者
+从没读过）只建基线、不出变化。列类型的写法第 1 波从通用改成了方言完整写法（目录-D10），拿老缓存的 VARCHAR2 和新读的
+VARCHAR2(20 CHAR) 比，所有 Oracle 列都会被说成「类型变了」。
+"""
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from typing import Any
+
+#: 表级的四种（调用方判）
+TABLE_ADDED = "table_added"
+TABLE_REMOVED = "table_removed"
+TABLE_LEFT_SCOPE = "table_left_scope"
+TABLE_REFLECT_FAILED = "table_reflect_failed"
+#: 列级、键
+COLUMN_ADDED = "column_added"
+COLUMN_REMOVED = "column_removed"
+COLUMN_TYPE = "column_type_changed"
+COLUMN_NULLABLE = "column_nullable_changed"
+COMMENT = "comment_changed"
+KEY = "key_changed"
+FK = "fk_changed"
+#: 探查范围的规则改了（源级，table 为空）
+SCOPE = "scope_changed"
+
+STRUCT_KINDS = (TABLE_ADDED, TABLE_REMOVED, TABLE_LEFT_SCOPE, TABLE_REFLECT_FAILED, COLUMN_ADDED, COLUMN_REMOVED,
+                COLUMN_TYPE, COLUMN_NULLABLE, KEY, FK, COMMENT, SCOPE)
+
+#: 没有影响面时的严重程度（影响面会升级：已确认的项、写死的 SQL 指向被删的列时是 action）
+DEFAULT_SEVERITY = {
+    TABLE_ADDED: "info", TABLE_REMOVED: "warning", TABLE_LEFT_SCOPE: "info", TABLE_REFLECT_FAILED: "warning",
+    COLUMN_ADDED: "info", COLUMN_REMOVED: "warning", COLUMN_TYPE: "warning", COLUMN_NULLABLE: "info",
+    KEY: "warning", FK: "warning", COMMENT: "info", SCOPE: "info",
+}
+
+#: 会让说到它的目录项、写死的 SQL 失效的：表真删了、列删了
+DESTRUCTIVE = frozenset({TABLE_REMOVED, COLUMN_REMOVED})
+
+
+def is_deletion(kind: str) -> bool:
+    """这种变化是不是删除。离开范围、读不到结构都不是：范围外的表还在库里，读不到结构的上一次的结构照旧留着。"""
+    return kind in DESTRUCTIVE
+
+
+def is_baseline(old_version: int | None, old_meta: Mapping[str, Any] | None) -> bool:
+    """这一次读到的结构只建基线、不和以前比：结构版本还是 0（升级前的老缓存接进来的、从没按对象索引读过），或者上一次的
+    结构不在手上（离开过范围，结构清掉了）。"""
+    return not old_version or not isinstance(old_meta, Mapping) or not old_meta.get("columns")
+
+
+@dataclass(frozen=True)
+class Diff:
+    """一条变化。column 是列级变化的那一列（表注释、键、外键为 None）；before / after 是前后各是什么（列的类型、可空、注释，
+    键的列组，外键的两端）；rename 是疑似改名的另一半 {from, to}（只说可能）；key 是键的种类（primary / unique）。"""
+
+    kind: str
+    column: str | None = None
+    before: Any = None
+    after: Any = None
+    rename: dict[str, str] | None = None
+    key: str | None = None
+    extra: dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def severity(self) -> str:
+        return DEFAULT_SEVERITY.get(self.kind, "info")
+
+    def to_dict(self) -> dict[str, Any]:
+        out: dict[str, Any] = {"kind": self.kind, "column": self.column, "before": self.before, "after": self.after}
+        if self.rename:
+            out["rename"] = dict(self.rename)
+        if self.key:
+            out["key"] = self.key
+        if self.extra:
+            out.update(self.extra)
+        return out
+
+
+def _columns(meta: Mapping[str, Any] | None) -> list[dict[str, Any]]:
+    return [c for c in (meta or {}).get("columns") or [] if isinstance(c, Mapping) and c.get("name") is not None]
+
+
+def _text(value: Any) -> str | None:
+    text = str(value).strip() if value is not None else ""
+    return text or None
+
+
+def _groups(raw: Any) -> set[tuple[str, ...]]:
+    return {tuple(str(c) for c in g) for g in raw or [] if isinstance(g, (list, tuple)) and g}
+
+
+def _fks(raw: Any) -> dict[tuple[Any, ...], dict[str, Any]]:
+    out: dict[tuple[Any, ...], dict[str, Any]] = {}
+    for fk in raw or []:
+        if not isinstance(fk, Mapping):
+            continue
+        cols = tuple(str(c) for c in fk.get("columns") or [])
+        key = (cols, str(fk.get("to_table") or ""), tuple(str(c) for c in fk.get("to_columns") or []))
+        out[key] = {"columns": list(key[0]), "to_table": key[1], "to_columns": list(key[2])}
+    return out
+
+
+def _slots_of(names: list[str], common: set[str]) -> dict[str, tuple[str | None, str | None]]:
+    """每一列在列清单里的「位置」：它前后最近的、两次都在的那两列。按下标比会被别处的删列、加列带偏（删了前面一列，后面
+    的下标全都前移一位），按前后的邻居比才说得出「还在原来那个位置」。"""
+    out: dict[str, tuple[str | None, str | None]] = {}
+    for i, name in enumerate(names):
+        before = next((n for n in reversed(names[:i]) if n in common), None)
+        after = next((n for n in names[i + 1:] if n in common), None)
+        out[name] = (before, after)
+    return out
+
+
+def diff_table(before: Mapping[str, Any] | None, after: Mapping[str, Any] | None) -> list[Diff]:
+    """同一张表前后两次的结构 → 变化（列级、键）。顺序：删掉的列、加上的列（各按列清单的顺序），再是各列的类型、可空、
+    注释，表注释，主键、唯一约束，外键。两边一样返回空列表。是不是基线（该不该比）由调用方先用 is_baseline 判。"""
+    old_cols = _columns(before)
+    new_cols = _columns(after)
+    old_by = {str(c["name"]): c for c in old_cols}
+    new_by = {str(c["name"]): c for c in new_cols}
+    removed = [n for n in old_by if n not in new_by]
+    added = [n for n in new_by if n not in old_by]
+    common = set(old_by) & set(new_by)
+    old_pos = _slots_of([str(c["name"]) for c in old_cols], common)
+    new_pos = _slots_of([str(c["name"]) for c in new_cols], common)
+
+    # 疑似改名：被删的和新加的类型相同、位置相同（夹在同样的两列之间）；一个被删的只配得上一个新加的、反过来也是，才算
+    pairs: dict[str, str] = {}
+    for r in removed:
+        hits = [a for a in added if old_pos[r] == new_pos[a] and old_by[r].get("type") == new_by[a].get("type")]
+        if len(hits) == 1:
+            back = [x for x in removed if old_pos[x] == new_pos[hits[0]]
+                    and old_by[x].get("type") == new_by[hits[0]].get("type")]
+            if back == [r]:
+                pairs[r] = hits[0]
+    reverse = {a: r for r, a in pairs.items()}
+
+    out: list[Diff] = []
+    for name in removed:
+        col = old_by[name]
+        out.append(Diff(COLUMN_REMOVED, name, before={"type": col.get("type"), "nullable": col.get("nullable", True)},
+                        rename={"from": name, "to": pairs[name]} if name in pairs else None))
+    for name in added:
+        col = new_by[name]
+        out.append(Diff(COLUMN_ADDED, name, after={"type": col.get("type"), "nullable": col.get("nullable", True)},
+                        rename={"from": reverse[name], "to": name} if name in reverse else None))
+    for name, col in new_by.items():
+        old = old_by.get(name)
+        if old is None:
+            continue
+        if (old.get("type") or "") != (col.get("type") or ""):
+            out.append(Diff(COLUMN_TYPE, name, before=old.get("type"), after=col.get("type")))
+        if bool(old.get("nullable", True)) != bool(col.get("nullable", True)):
+            out.append(Diff(COLUMN_NULLABLE, name, before=bool(old.get("nullable", True)),
+                            after=bool(col.get("nullable", True))))
+        if _text(old.get("comment")) != _text(col.get("comment")):
+            out.append(Diff(COMMENT, name, before=_text(old.get("comment")), after=_text(col.get("comment"))))
+    if _text((before or {}).get("comment")) != _text((after or {}).get("comment")):
+        out.append(Diff(COMMENT, None, before=_text((before or {}).get("comment")),
+                        after=_text((after or {}).get("comment"))))
+
+    old_pk = [str(c) for c in (before or {}).get("primary_key") or []]
+    new_pk = [str(c) for c in (after or {}).get("primary_key") or []]
+    if old_pk != new_pk:
+        out.append(Diff(KEY, None, before=old_pk or None, after=new_pk or None, key="primary"))
+    # 唯一约束、外键：反射读不到时不写这个键，两边都有才比
+    if "unique" in (before or {}) and "unique" in (after or {}):
+        old_u, new_u = _groups(before.get("unique")), _groups(after.get("unique"))  # type: ignore[union-attr]
+        for g in sorted(old_u - new_u):
+            out.append(Diff(KEY, g[0] if len(g) == 1 else None, before=list(g), after=None, key="unique"))
+        for g in sorted(new_u - old_u):
+            out.append(Diff(KEY, g[0] if len(g) == 1 else None, before=None, after=list(g), key="unique"))
+    if "foreign_keys" in (before or {}) and "foreign_keys" in (after or {}):
+        old_f, new_f = _fks(before.get("foreign_keys")), _fks(after.get("foreign_keys"))  # type: ignore[union-attr]
+        for k in sorted(set(old_f) - set(new_f)):
+            cols = old_f[k]["columns"]
+            out.append(Diff(FK, cols[0] if len(cols) == 1 else None, before=old_f[k], after=None))
+        for k in sorted(set(new_f) - set(old_f)):
+            cols = new_f[k]["columns"]
+            out.append(Diff(FK, cols[0] if len(cols) == 1 else None, before=None, after=new_f[k]))
+    return out
+
+
+def touched_columns(diffs: list[Diff]) -> dict[str, set[str]]:
+    """各条变化说到的列，按种类分：{"removed": {…}, "changed": {…}}。removed 是被删的列（含疑似改名的旧名），changed 是
+    类型、可空变了的列。影响面按它找目录里说到这些列的项。"""
+    removed = {d.column for d in diffs if d.kind == COLUMN_REMOVED and d.column}
+    changed = {d.column for d in diffs if d.kind in (COLUMN_TYPE, COLUMN_NULLABLE) and d.column}
+    return {"removed": removed, "changed": changed}
+
+
+__all__ = [
+    "COLUMN_ADDED", "COLUMN_NULLABLE", "COLUMN_REMOVED", "COLUMN_TYPE", "COMMENT", "DEFAULT_SEVERITY", "DESTRUCTIVE",
+    "Diff", "FK", "KEY", "SCOPE", "STRUCT_KINDS", "TABLE_ADDED", "TABLE_LEFT_SCOPE", "TABLE_REFLECT_FAILED",
+    "TABLE_REMOVED", "diff_table", "is_baseline", "is_deletion", "touched_columns",
+]

backend/app/data/source_changes.py

@@ -0,0 +1,492 @@
+"""数据源的变化记录（docs/design/next-round/03-数据接入与目录维护.md B3）：结构变化、取值变化的记录、去重、处理和拉取。
+
+**为什么要有它。** 以前表结构变了、码值多了一个新取值都没有地方记：审阅页看不到，通知发不了，口径账本的召回也没有可以
+订阅的源。现在全站只有 source_changes 这一份记录（格式 agentlab-change/1）：
+
+- **记**（record）：后台探查读到的结构差异（data/schema_diff → introspect_jobs）、取值观察和同口径历史的比较
+  （data/value_drift → catalog_profile_observe）都经这里写。fingerprint 去重：同一个变化再次测到只更新 last_seen_at、
+  seen_count（还开着的顺带换上最新的影响面和严重程度），不另起一条。
+- **处理**（handle / resolve）：动作都落到目录已有的写入接口上——清理目录里说到被删的列的项（catalog.write_entry，
+  整份写、乐观锁）、加入码值（catalog.save_patch：apply_patch 配 _merge_codes，含义待填写，依据写这条变化的编号）、重新
+  起草（建起草任务）、重新剖析（建剖析任务）、知道了、误报（必须写理由）。每个动作写 resolution，动了目录的记下目录版本，
+  建了任务的记下任务编号。
+- **拉取**（list_changes / affecting）：给召回按 seq 游标拉；seq 自增、不复用。触发召回的只有结构类、处理时改了目录的、
+  以及 rules_touched 的新码值（triggers_recall）；纯取值漂移只算数据质量提醒。
+- **说法**（summary）：一条变化给人看的一句（接口、日志用）；通知里不放它（不放表名、列名、取值，D9），见
+  notify/service.notify_source_changed。
+
+不记遮罩列的取值（观察查询本来就不碰遮罩列）。
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.data import schema_diff as sd
+from app.data.names import name_key
+from app.db.base import utcnow
+from app.db.models_sources import SourceChange
+
+logger = logging.getLogger(__name__)
+
+SCHEMA = "agentlab-change/1"
+
+#: 取值类的变化（data/value_drift）
+CODES_NEW = "codes_new"
+CODES_SHIFT = "codes_shift"
+DATE_STALLED = "date_stalled"
+DATE_GAP = "date_gap"
+VOLUME_DROP = "volume_drop"
+ROWS_DROP = "rows_drop"
+ROWS_JUMP = "rows_jump"
+COVERAGE_DROP = "coverage_drop"
+VALUE_KINDS = (CODES_NEW, CODES_SHIFT, DATE_STALLED, DATE_GAP, VOLUME_DROP, ROWS_DROP, ROWS_JUMP, COVERAGE_DROP)
+KINDS = sd.STRUCT_KINDS + VALUE_KINDS
+
+SEVERITIES = ("info", "warning", "action")
+SEVERITY_RANK = {"info": 1, "warning": 2, "action": 3}
+STATUSES = ("open", "acknowledged", "resolved", "dismissed")
+#: 还要人看的：开着的、知道了但还没处理的
+PENDING = ("open", "acknowledged")
+
+#: 处理动作：clean_catalog 清理目录里说到这些列的项、add_codes 加入码值、redraft 重新起草这张表、reprofile 重新剖析、
+#: acknowledge 知道了、dismiss 误报（必须写理由）
+ACTIONS = ("clean_catalog", "add_codes", "redraft", "reprofile", "acknowledge", "dismiss")
+#: 动作之后的状态
+ACTION_STATUS = {"clean_catalog": "resolved", "add_codes": "resolved", "redraft": "resolved", "reprofile": "resolved",
+                 "acknowledge": "acknowledged", "dismiss": "dismissed"}
+#: 理由最长几个字
+NOTE_MAX = 500
+
+#: 变化的叫法（接口给人看的 summary 用；前端第 4 波的变化标签页在 lib/terms.ts 另有一份，逐字对照）
+KIND_LABEL = {
+    sd.TABLE_ADDED: "新出现的表", sd.TABLE_REMOVED: "表已删除", sd.TABLE_LEFT_SCOPE: "表离开了探查范围",
+    sd.TABLE_REFLECT_FAILED: "这次读不到表结构", sd.COLUMN_ADDED: "新加了列", sd.COLUMN_REMOVED: "删除了列",
+    sd.COLUMN_TYPE: "列的类型变了", sd.COLUMN_NULLABLE: "列的可空变了", sd.KEY: "键变了", sd.FK: "外键变了",
+    sd.COMMENT: "注释变了", sd.SCOPE: "探查范围的规则变了",
+    CODES_NEW: "出现了新码值", CODES_SHIFT: "码值的比例变化大", DATE_STALLED: "最新业务日期没有前进",
+    DATE_GAP: "有的日子没有数据", VOLUME_DROP: "按日行数大幅下跌", ROWS_DROP: "行数大幅下跌", ROWS_JUMP: "行数大幅上涨",
+    COVERAGE_DROP: "关联的覆盖率下降",
+}
+
+
+class ChangeNotFound(Exception):
+    pass
+
+
+class ChangeFinished(Exception):
+    """这条变化已经处理过（resolved / dismissed）了。"""
+
+
+class ChangeInvalid(ValueError):
+    """处理动作不对：不认识的动作、误报没写理由、这种变化做不了这个动作。message 给人看。"""
+
+
+# ==========================================================================
+# 记
+# ==========================================================================
+
+
+def fingerprint(*parts: Any) -> str:
+    """去重用的指纹：几段内容的规范 JSON 的 sha256。"""
+    text = json.dumps(list(parts), sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str)
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+@dataclass
+class Detected:
+    """谁、经什么测到的（detected_by）：job_kind 是 introspect / profile / on_demand，trigger 是 manual / schedule /
+    on_demand，actor 是署名。"""
+
+    job_kind: str
+    job_id: str | None = None
+    task_seq: int | None = None
+    trigger: str = "manual"
+    actor: str | None = None
+    fire_at: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        out = {"job_kind": self.job_kind, "job_id": self.job_id, "task_seq": self.task_seq, "trigger": self.trigger,
+               "actor": self.actor}
+        if self.fire_at:
+            out["fire_at"] = self.fire_at
+        return out
+
+
+@dataclass
+class NewChange:
+    """要记的一条变化（还没去重）。key 是去重指纹的组成部分（除了源、种类、表、列以外的那部分：结构版本、码值……）。"""
+
+    kind: str
+    table: str
+    column: str | None = None
+    severity: str = "info"
+    before: Any = None
+    after: Any = None
+    basis: dict[str, Any] | None = None
+    affects: dict[str, Any] | None = None
+    struct_version_before: int | None = None
+    struct_version_after: int | None = None
+    key: Any = None
+
+
+async def record(session: AsyncSession, source: Any, change: NewChange, detected: Detected) -> tuple[SourceChange, bool]:
+    """记一条变化（不提交）。同一个指纹已经有了：只更新 last_seen_at、seen_count（还开着的另换上最新的影响面、前后值、
+    严重程度），返回 (那一条, False)；新的返回 (新的一条, True)。"""
+    if change.kind not in KINDS:
+        raise ValueError(f"不认识的变化种类：{change.kind}")
+    if change.severity not in SEVERITIES:
+        raise ValueError(f"不认识的严重程度：{change.severity}")
+    fp = fingerprint(change.kind, change.table, change.column, change.key)
+    now = utcnow()
+    found = (await session.execute(select(SourceChange).where(
+        SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one_or_none()
+    if found is not None:
+        found.last_seen_at = now
+        found.seen_count = int(found.seen_count or 0) + 1
+        if found.status == "open":
+            found.affects, found.before, found.after = change.affects, change.before, change.after
+            if SEVERITY_RANK[change.severity] > SEVERITY_RANK.get(found.severity, 0):
+                found.severity = change.severity
+        return found, False
+    row = SourceChange(source_id=source.id, source_name=str(getattr(source, "name", "") or ""),
+                       table_name=change.table or "", column_name=change.column, kind=change.kind,
+                       severity=change.severity, detected_at=now, last_seen_at=now, seen_count=1,
+                       detected_by=detected.to_dict(), basis=change.basis, before=change.before, after=change.after,
+                       affects=change.affects, struct_version_before=change.struct_version_before,
+                       struct_version_after=change.struct_version_after, fingerprint=fp, status="open",
+                       schema_tag=SCHEMA)
+    try:
+        async with session.begin_nested():
+            session.add(row)
+    except IntegrityError:
+        # 同一个指纹刚被别处写进去（单进程下几乎不会）：按已有的那条更新
+        found = (await session.execute(select(SourceChange).where(
+            SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one()
+        found.last_seen_at, found.seen_count = now, int(found.seen_count or 0) + 1
+        return found, False
+    return row, True
+
+
+async def record_all(session: AsyncSession, source: Any, changes: Iterable[NewChange],
+                     detected: Detected) -> list[SourceChange]:
+    """一批变化逐条 record（不提交），返回这次新记的那几条。"""
+    fresh: list[SourceChange] = []
+    for change in changes:
+        row, new = await record(session, source, change, detected)
+        if new:
+            fresh.append(row)
+    return fresh
+
+
+# ==========================================================================
+# 读
+# ==========================================================================
+
+
+def _iso(value: Any) -> str | None:
+    return value.isoformat() if value is not None else None
+
+
+def summary(row: Any) -> str:
+    """一条变化给人看的一句（接口、日志）：「orders：删除了列 amount（可能是改名为 total，未确认）」。"""
+    kind = row.kind
+    label = KIND_LABEL.get(kind, kind)
+    table = row.table_name or ""
+    column = row.column_name
+    where = f"{table}.{column}" if table and column else table
+    text = f"{where}：{label}" if where else label
+    after = row.after if isinstance(row.after, Mapping) else {}
+    before = row.before if isinstance(row.before, Mapping) else {}
+    rename = after.get("rename") or before.get("rename")
+    if isinstance(rename, Mapping):
+        text += f"（可能是 {rename.get('from')} 改名为 {rename.get('to')}，未确认）"
+    elif kind == sd.COLUMN_TYPE:
+        text += f"（{row.before} → {row.after}）"
+    elif kind == CODES_NEW and isinstance(after.get("codes"), list):
+        text += f"（{'、'.join(str(c) for c in after['codes'][:8])}）"
+    return text
+
+
+def change_out(row: SourceChange) -> dict[str, Any]:
+    """接口给的一条变化（agentlab-change/1）。"""
+    return {
+        "schema": row.schema_tag or SCHEMA, "seq": row.seq, "id": row.id, "source_id": row.source_id,
+        "source_name": row.source_name, "table": row.table_name or None, "column": row.column_name, "kind": row.kind,
+        "severity": row.severity, "summary": summary(row), "detected_at": _iso(row.detected_at),
+        "last_seen_at": _iso(row.last_seen_at), "seen_count": row.seen_count or 1,
+        "detected_by": dict(row.detected_by or {}), "basis": row.basis, "before": row.before, "after": row.after,
+        "affects": row.affects or {"catalog": [], "templates": [], "rules_touched": False},
+        "struct_version_before": row.struct_version_before, "struct_version_after": row.struct_version_after,
+        "status": row.status, "resolution": row.resolution,
+    }
+
+
+async def load(session: AsyncSession, source_id: str, change_id: str) -> SourceChange | None:
+    row = (await session.execute(select(SourceChange).where(SourceChange.id == change_id))).scalar_one_or_none()
+    return row if row is not None and row.source_id == source_id else None
+
+
+async def counts(session: AsyncSession, source_id: str) -> dict[str, Any]:
+    """一个源还要人看的变化（开着的、知道了还没处理的）：{pending, by_severity: {action, warning, info}, open}。数据源卡片的
+    角标「N 个变化要处理」按 by_severity.action（第 4 波）。"""
+    rows = (await session.execute(select(SourceChange.status, SourceChange.severity, func.count()).where(
+        SourceChange.source_id == source_id, SourceChange.status.in_(PENDING))
+        .group_by(SourceChange.status, SourceChange.severity))).all()
+    by = dict.fromkeys(SEVERITIES, 0)
+    opened = 0
+    for status, severity, n in rows:
+        by[severity] = by.get(severity, 0) + n
+        if status == "open":
+            opened += n
+    return {"pending": sum(by.values()), "open": opened, "by_severity": by}
+
+
+async def list_for_source(session: AsyncSession, source: Any, *, status: str | None = "pending",
+                          kinds: Iterable[str] | None = None, table: str | None = None, limit: int = 100,
+                          offset: int = 0) -> tuple[list[SourceChange], int]:
+    """一个源的变化（审阅页「变化」标签页）：按「使用次数权重 × 严重程度」排（catalog_review 的权重），同分新的在前。
+    status：pending 开着的和知道了的（缺省）、all、或者某一个状态。返回 (这一页, 一共几条)。"""
+    from app.data import catalog_review
+
+    filt = [SourceChange.source_id == source.id]
+    if status == "pending":
+        filt.append(SourceChange.status.in_(PENDING))
+    elif status and status != "all":
+        filt.append(SourceChange.status == status)
+    if kinds:
+        filt.append(SourceChange.kind.in_(list(kinds)))
+    if table:
+        filt.append(SourceChange.table_name == table)
+    total = int((await session.execute(select(func.count()).select_from(SourceChange).where(*filt))).scalar() or 0)
+    rows = list((await session.execute(select(SourceChange).where(*filt))).scalars())
+    usage = (await catalog_review.table_usage_counts(session, source)).queries if rows else {}
+
+    def rank(r: SourceChange) -> tuple[float, int]:
+        weight = catalog_review.usage_weight(usage.get(r.table_name, 0)) * SEVERITY_RANK.get(r.severity, 1)
+        return -weight, -int(r.seq or 0)
+
+    rows.sort(key=rank)
+    return rows[offset:offset + limit], total
+
+
+async def list_changes(session: AsyncSession, *, since_seq: int = 0, source_id: str | None = None,
+                       kinds: Iterable[str] | None = None, status: Iterable[str] | str | None = None,
+                       limit: int = 200) -> list[dict[str, Any]]:
+    """按 seq 游标拉取（给召回）：seq 大于 since_seq 的，按 seq 从小到大，最多 limit 条。拉到的最后一条的 seq 就是下一次的
+    since_seq。"""
+    filt = [SourceChange.seq > int(since_seq or 0)]
+    if source_id:
+        filt.append(SourceChange.source_id == source_id)
+    if kinds:
+        filt.append(SourceChange.kind.in_(list(kinds)))
+    if isinstance(status, str):
+        filt.append(SourceChange.status == status)
+    elif status:
+        filt.append(SourceChange.status.in_(list(status)))
+    rows = (await session.execute(select(SourceChange).where(*filt).order_by(SourceChange.seq)
+                                  .limit(max(1, min(int(limit), 1000))))).scalars()
+    return [change_out(r) for r in rows]
+
+
+async def affecting(session: AsyncSession, source_name: str, tables: Iterable[str], since: int | datetime | None = None,
+                    ) -> list[dict[str, Any]]:
+    """动到这个源这几张表的变化（给召回反查「这次答案用到的表之后有没有变」）：表名按 name_key 比；since 是 seq（拉取的
+    游标）或时刻（测到的时间晚于它的）。源级的变化（范围规则改了）不算。"""
+    keys = {name_key(str(t).rsplit(".", 1)[-1]) for t in tables if t}
+    if not keys:
+        return []
+    filt = [SourceChange.source_name == source_name, SourceChange.table_name != ""]
+    if isinstance(since, datetime):
+        filt.append(or_(SourceChange.detected_at > since, SourceChange.last_seen_at > since))
+    elif since:
+        filt.append(SourceChange.seq > int(since))
+    rows = (await session.execute(select(SourceChange).where(*filt).order_by(SourceChange.seq))).scalars()
+    return [change_out(r) for r in rows if name_key(r.table_name) in keys]
+
+
+def triggers_recall(change: Mapping[str, Any]) -> bool:
+    """这条变化触发不触发召回：结构类、处理时改了目录的（resolution 里有目录版本）、触及有效记录条件的新码值。纯取值漂移
+    只算数据质量提醒。"""
+    kind = change.get("kind")
+    if kind in sd.STRUCT_KINDS:
+        return True
+    if (change.get("resolution") or {}).get("catalog_version"):
+        return True
+    return kind == CODES_NEW and bool((change.get("affects") or {}).get("rules_touched"))
+
+
+# ==========================================================================
+# 处理
+# ==========================================================================
+
+
+def resolve(row: SourceChange, *, action: str, by: str | None, note: str | None = None,
+            catalog_version: int | None = None, job_id: str | None = None) -> SourceChange:
+    """把一条变化记成处理过（不提交）：状态按动作（ACTION_STATUS），resolution 写谁、什么时候、什么动作、理由、目录版本、
+    任务编号。已经 resolved / dismissed 的抛 ChangeFinished；误报没写理由抛 ChangeInvalid。"""
+    if action not in ACTIONS:
+        raise ChangeInvalid(f"不认识的处理动作：{action}。可选：{'、'.join(ACTIONS)}")
+    if row.status in ("resolved", "dismissed"):
+        raise ChangeFinished("这条变化已经处理过了")
+    text = (note or "").strip()
+    if action == "dismiss" and not text:
+        raise ChangeInvalid("标为误报要写明理由")
+    if len(text) > NOTE_MAX:
+        raise ChangeInvalid(f"理由最多 {NOTE_MAX} 个字")
+    resolution: dict[str, Any] = {"by": by, "at": utcnow().isoformat(), "action": action, "note": text or None}
+    if catalog_version is not None:
+        resolution["catalog_version"] = catalog_version
+    if job_id is not None:
+        resolution["job_id"] = job_id
+    row.status = ACTION_STATUS[action]
+    row.resolution = resolution
+    return row
+
+
+def _catalog_paths(row: SourceChange) -> list[dict[str, Any]]:
+    """影响面里这张表上的目录项（清理目录时删的就是它们）。"""
+    items = ((row.affects or {}).get("catalog") or [])
+    return [i for i in items if isinstance(i, Mapping) and i.get("table") == row.table_name and i.get("path")]
+
+
+def without_paths(notes: Mapping[str, Any], paths: Iterable[str]) -> dict[str, Any]:
+    """目录去掉这几项（路径同审阅接口：字段名、columns.<列>.<字段>、relations.<编号>）。"""
+    from app.data import catalog
+
+    slots = catalog._slots(notes)
+    for path in paths:
+        try:
+            slots.pop(catalog.parse_path(path), None)
+        except catalog.CatalogPathError:
+            continue
+    return catalog._assemble(slots)
+
+
+async def handle(session: AsyncSession, source: Any, row: SourceChange, action: str, *, actor: str | None,
+                 note: str | None = None, codes: Mapping[str, str] | None = None) -> dict[str, Any]:
+    """执行一个处理动作并记下 resolution（提交）。返回 {change, catalog_version?, job?}。
+
+    - clean_catalog：把影响面里这张表上说到这些列的项从目录里去掉（整份写入，乐观锁），记目录版本。影响面里没有目录项的
+      抛 ChangeInvalid。
+    - add_codes：码值的新取值加进目录（catalog.save_patch，含义待填写；依据写「数据源变化 #编号」），只给 codes_new。codes
+      给了就按它加（{码: 含义}，含义可以空），不给就加这条变化测到的新码值。
+    - redraft：建一个只含这张表的起草任务；reprofile：建一个只含这张表的剖析任务。建不了（模型没配、剖析没开、同源正忙）
+      抛 ChangeInvalid，原因照实。
+    - acknowledge、dismiss（必须写理由）只改状态。
+    """
+    from app.data import catalog
+
+    if action not in ACTIONS:
+        raise ChangeInvalid(f"不认识的处理动作：{action}。可选：{'、'.join(ACTIONS)}")
+    if row.status in ("resolved", "dismissed"):
+        raise ChangeFinished("这条变化已经处理过了")
+    if action == "dismiss" and not (note or "").strip():
+        raise ChangeInvalid("标为误报要写明理由")
+    out: dict[str, Any] = {}
+    version: int | None = None
+    job_id: str | None = None
+    table = row.table_name
+    if action == "clean_catalog":
+        paths = [str(i["path"]) for i in _catalog_paths(row)]
+        if not paths:
+            raise ChangeInvalid("这条变化的影响面里没有目录项，不需要清理")
+        entry = await catalog.read_entry(session, source.id, table)
+        if entry is None:
+            raise ChangeInvalid(f"{table} 已经没有数据目录了")
+        notes = without_paths(entry.notes, paths)
+        written = await catalog.write_entry(session, source.id, table, notes, if_version=entry.version, actor=actor,
+                                            commit=False)
+        version = written.version
+    elif action == "add_codes":
+        if row.kind != CODES_NEW or not row.column_name:
+            raise ChangeInvalid("只有「出现了新码值」能加入码值")
+        wanted = dict(codes) if codes else {str(c): "" for c in ((row.after or {}).get("codes") or [])}
+        if not wanted or not all(isinstance(k, str) and isinstance(v, str) for k, v in wanted.items()):
+            raise ChangeInvalid("要加入的码值应为 {码值: 含义} 的对照，含义可以先空着")
+        entry = await catalog.read_entry(session, source.id, table)
+        current = entry.version if entry else 0
+        meta_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or None
+        changes = [{"path": f"columns.{row.column_name}.codes", "value": wanted,
+                    "reason": f"数据源变化 #{row.id[:6]}：取值观察测到的新码值，含义待填写"}]
+        try:
+            notes = catalog.apply_patch(entry.notes if entry else {}, changes, table=table, tables=meta_tables)
+        except catalog.CatalogInvalid as e:
+            raise ChangeInvalid("；".join(e.problems) if hasattr(e, "problems") else str(e)) from e
+        written = await catalog.write_entry(session, source.id, table, notes, if_version=current, actor=actor,
+                                            commit=False)
+        version = written.version
+    elif action == "reprofile":
+        job_id = await _reprofile(session, source, table, actor)
+        out["job"] = {"kind": "profile", "id": job_id}
+    elif action == "redraft":
+        job_id = await _redraft(session, source, table, actor, row.id)
+        out["job"] = {"kind": "draft", "id": job_id}
+    resolve(row, action=action, by=actor, note=note, catalog_version=version, job_id=job_id)
+    await session.commit()
+    if version is not None:
+        out["catalog_version"] = version
+    out["change"] = change_out(row)
+    return out
+
+
+async def _reprofile(session: AsyncSession, source: Any, table: str, actor: str | None) -> str:
+    from app.data import catalog_profile, catalog_profile_jobs, table_versions
+
+    try:
+        settings = catalog_profile.ensure_enabled(source)
+    except catalog_profile.ProfileDisabled as e:
+        raise ChangeInvalid(str(e)) from e
+    try:
+        view = await table_versions.resolve_source(session, source)
+    except table_versions.SnapshotMissing as e:
+        raise ChangeInvalid(str(e)) from e
+    try:
+        job, _tasks = await catalog_profile_jobs.create_job(session, source, view, tables=[table], scope="selected",
+                                                            settings=settings, actor=actor)
+    except (catalog_profile_jobs.ProfileBusy, catalog_profile_jobs.ProfileNothing) as e:
+        raise ChangeInvalid(str(e)) from e
+    catalog_profile_jobs.manager.start(job.id)
+    return job.id
+
+
+async def _redraft(session: AsyncSession, source: Any, table: str, actor: str | None, change_id: str) -> str:
+    """建一个只含这张表的起草任务。起草模型用不了时只按表结构起草（数据库注释、外键、命名推断），不挡这个动作。"""
+    from app.data import catalog, catalog_jobs, table_versions
+
+    try:
+        view = await table_versions.resolve_source(session, source)
+    except table_versions.SnapshotMissing as e:
+        raise ChangeInvalid(str(e)) from e
+    if not ((getattr(view, "schema_cache", None) or {}).get("tables") or {}):
+        raise ChangeInvalid("这个数据源还没有表结构，不能起草")
+    use_model = True
+    try:
+        await catalog_jobs.check_model(session)
+    except catalog.CatalogModelUnavailable:
+        use_model = False
+    try:
+        job, _tasks = await catalog_jobs.create_job(session, source, view, tables=[table], use_model=use_model,
+                                                    note=f"数据源变化 #{change_id[:6]}", actor=actor)
+    except catalog_jobs.DraftBusy as e:
+        raise ChangeInvalid(str(e)) from e
+    catalog_jobs.manager.start(job.id)
+    return job.id
+
+
+__all__ = [
+    "ACTIONS", "ACTION_STATUS", "CODES_NEW", "CODES_SHIFT", "COVERAGE_DROP", "ChangeFinished", "ChangeInvalid",
+    "ChangeNotFound", "DATE_GAP", "DATE_STALLED", "Detected", "KINDS", "KIND_LABEL", "NewChange", "PENDING", "ROWS_DROP",
+    "ROWS_JUMP", "SCHEMA", "SEVERITIES", "SEVERITY_RANK", "STATUSES", "VALUE_KINDS", "VOLUME_DROP", "affecting",
+    "change_out", "counts", "fingerprint", "handle", "list_changes", "list_for_source", "load", "record", "record_all",
+    "resolve", "summary", "triggers_recall", "without_paths",
+]

backend/app/data/source_objects.py

@@ -446,6 +446,11 @@ class RefreshStats:
     note: str | None = None
     #: 第一次建对象索引时从升级前的老缓存接进来几张（legacy_seed）
     adopted: int = 0
+    #: 这次新出现的、不在了（真删除）的对象名（结构变化按它记 table_added / table_removed，next-round 03 B1）；
+    #: first_build：这个 schema 的对象索引是这一次才建起来的——全是新的，只建基线、不记「新出现的表」
+    added_names: list[str] = field(default_factory=list)
+    removed_names: list[str] = field(default_factory=list)
+    first_build: bool = False
 
 
 def _chunks(items: list[Any], size: int = 500) -> Iterable[list[Any]]:
@@ -548,14 +553,18 @@ async def refresh_index(session: AsyncSession, source_id: str, schema: str, resu
         await session.execute(update(_SO).where(_SO.id.in_(chunk)).values(removed_at=None)
                               .execution_options(synchronize_session=False))
     gone: list[str] = []
+    gone_names: list[str] = []
     if not result.capped:
-        gone = [r.id for name, r in existing.items() if name not in seen and r.removed_at is None]
+        gone_names = [name for name, r in existing.items() if name not in seen and r.removed_at is None]
+        gone = [existing[name].id for name in gone_names]
         for chunk in _chunks(gone):
-            # 真删除：不在范围里了，结构（meta）留着——下一波的结构差异要拿它和以前比
+            # 真删除：不在范围里了，结构（meta）留着——结构差异（data/schema_diff）要拿它和以前比、算影响面
             await session.execute(update(_SO).where(_SO.id.in_(chunk)).values(removed_at=now, in_scope=False)
                                   .execution_options(synchronize_session=False))
     return RefreshStats(total=len(seen), added=len(new_rows), removed=len(gone), method=result.method,
-                        capped=result.capped, note=result.note, adopted=len(seeds))
+                        capped=result.capped, note=result.note, adopted=len(seeds),
+                        added_names=[r["name"] for r in new_rows] if existing else [], removed_names=gone_names,
+                        first_build=not existing)
 
 
 async def live_objects(session: AsyncSession, source_id: str, schema: str) -> list[tuple[str, str]]:

backend/app/data/sqlcheck_columns.py

@@ -0,0 +1,142 @@
+"""SQL 里的列名对照数据源结构（sql_unknown_columns）：助手自查（api/copilot.py）和表结构变化的影响面
+（data/catalog_impact.py：已发布模板写死的 SQL 换上新结构再跑一遍，说得出「这条 SQL 现在会报列不存在」）共用这一份。
+
+以前它在 api/copilot.py 里，影响面要用就得从 data 包反过来导入接口层；挪到这里，copilot 那边只改 import。
+
+只做有把握的判断，宁可漏判不可误判：助手自查会把图打回去让模型改，影响面会把一条变化升为「要处理」。CTE、FROM 里的
+子查询、引号括起来的标识符、结构里找不到的表、结构没探查过或者探查被截断的，一律不判。
+"""
+from __future__ import annotations
+
+import re
+from typing import Any
+
+#: @ 开头的是变量（@region、@@session），: 开头的是参数，. 后面的是限定名
+_SQL_IDENT = re.compile(r"(?<![\w$#.:@])([A-Za-z_][A-Za-z0-9_$#]*)(?![\w$#])")
+_FROM_ITEMS = re.compile(
+    r"\b(?:from|join)\s+(.+?)(?=\bwhere\b|\bgroup\b|\border\b|\bhaving\b|\blimit\b|\bfetch\b|\boffset\b"
+    r"|\bunion\b|\bintersect\b|\bexcept\b|\bon\b|\busing\b|\b(?:left|right|inner|outer|full|cross|natural)\b"
+    r"|\bjoin\b|\bwindow\b|\)|$)", re.I | re.S)
+_TABLE_ITEM = re.compile(r"^\s*([A-Za-z_][\w$#]*(?:\.[A-Za-z_][\w$#]*)*)(?:\s+(?:as\s+)?([A-Za-z_][\w$#]*))?\s*$",
+                         re.I)
+_ALIAS_AS = re.compile(r"\bas\s+([A-Za-z_][\w$#]*)", re.I)
+#: 不写 AS 的列别名：「SUM(amount) total,」「amount total FROM」
+_ALIAS_BARE = re.compile(r"(\)|\b[A-Za-z_][\w$#]*|\d)\s+([A-Za-z_][\w$#]*)\s*(?=,|\bfrom\b)", re.I)
+_QUALIFIED = re.compile(r"(?<![\w$#.])([A-Za-z_][\w$#]*)\s*\.\s*([A-Za-z_][\w$#]*|\*)(?!\s*\()")
+_SQL_WORDS = frozenset("""
+select distinct all from where and or not in is null like ilike between exists case when then else end as on using
+join inner left right outer full cross natural group by order having limit offset fetch first next rows row only
+union intersect except asc desc nulls last with over partition range preceding following unbounded current true
+false interval year month day hour minute second week quarter dow doy epoch date time timestamp datetime zone
+cast integer int bigint smallint tinyint decimal numeric real float double precision varchar char character text
+boolean bool signed unsigned binary top percent escape collate any some values current_date current_time
+current_timestamp localtime localtimestamp sysdate systimestamp rownum level dual unknown filter within
+at nocase rtrim glob regexp rlike similar to div mod xor isnull notnull minus prior connect start nocycle siblings
+grouping sets rollup cube ignore respect lateral apply qualify window exclude include ties others no groups
+for update of share nowait skip locked lock mode array sounds overlaps language query expansion separator
+charset utf8 utf8mb4 latin1 ascii unicode years months weeks days hours minutes seconds microsecond microseconds
+millisecond milliseconds year_month day_hour day_minute day_second hour_minute hour_second minute_second
+isodow isoyear century decade millennium julian timezone timezone_hour timezone_minute
+sql_calc_found_rows sql_no_cache sql_cache sql_small_result sql_big_result sql_buffer_result high_priority
+straight_join distinctrow unique recursive tablesample bernoulli sample repeatable pivot unpivot keep ordinality
+symmetric asymmetric leading trailing both placing force use index into default without local
+user current_user session_user system_user current_schema current_catalog current_role
+""".split())
+
+#: 后面跟的一定是操作数、不会是别名的词：「SELECT created_at FROM」里 created_at 是列。
+#: END、NULL、TRUE、CURRENT_DATE 这类「一个值到此为止」的词不在里面——「… END flag,」的 flag 是别名
+_OPERAND_BEFORE = frozenset("""
+select distinct distinctrow all unique by as and or not xor is in like ilike glob regexp rlike similar to between
+when then else case on where having exists div mod escape interval collate using any some from join
+sql_calc_found_rows sql_no_cache sql_cache sql_small_result sql_big_result sql_buffer_result high_priority
+straight_join
+""".split())
+#: 不是列的名字：排序规则名（COLLATE x）、字符集名（USING x、CHARSET x、CHARACTER SET x）
+_NAME_AFTER = re.compile(
+    r"\b(?:collate|charset|character\s+set|using(?!\s*\())\s+([A-Za-z_][\w$#]*)", re.I)
+#: DATEADD(dd, …)、DATEDIFF(wk, …) 的第一个参数是日期部分的缩写，不是列。只在这个位置放过：
+#: 放进关键字表的话，别处打错成 wk、dd 的列名就查不出来了
+_DATEPART_ARG = re.compile(
+    r"\b(?:dateadd|datediff|datediff_big|datepart|datename|datetrunc|date_bucket|timestampadd|timestampdiff)"
+    r"\s*\(\s*([A-Za-z_][\w$#]*)", re.I)
+#: 命名窗口：WINDOW w AS (…)、…, w2 AS (…)
+_WINDOW_NAME = re.compile(r"(?:\bwindow|,)\s*([A-Za-z_][\w$#]*)\s+as\s*\(", re.I)
+
+
+def _blank_unicode_identifiers(sql: str) -> str:
+    """带引号、名字里有非 ASCII 字符的标识符（中文别名 AS "月份"、AS `曝光量`）连引号一起换成空格。
+
+    这项核对只认 ASCII 的列名，中文名本来就不判；不抹掉的话，SQL 里一出现带引号的标识符就整条放弃核对，照「结果列
+    起中文别名」写的 SQL 会把猜错的英文列名一起放过去。英文的带引号标识符（"StatDate"）大小写、空格都可能，照旧不判。"""
+    from app.data.guard import _GENERIC, _tokens
+
+    out = []
+    for token in _tokens(sql, _GENERIC):
+        if token.kind == "ident" and not token.text.isascii():
+            out.append(" " * len(token.text))
+        else:
+            out.append(token.text)
+    return "".join(out)
+
+
+def sql_unknown_columns(sql: str, source: Any) -> tuple[list[str], list[str]] | None:
+    """SQL 里用到、数据源结构里查不到的列：(查不到的列, 这几张表的全部列)。判断不了、或者都查得到时返回 None。"""
+    from app.data import introspect as _introspect
+    from app.data.guard import _blank_quoted, first_verb, strip_comments
+
+    cache = getattr(source, "schema_cache", None) or {}
+    if not cache.get("tables") or cache.get("failed") or first_verb(sql) != "select":
+        return None
+    text = _blank_unicode_identifiers(re.sub(r"\{\{.*?\}\}", " 0 ", strip_comments(sql), flags=re.S))
+    if '"' in text.replace("''", "") and _blank_quoted(text) != _blank_quoted(text, identifiers=False):
+        return None                                   # 引号括起来的标识符：大小写、空格都可能，不猜
+    text = _blank_quoted(text)
+    if re.search(r"\bwith\b", text, re.I) or re.search(r"\b(?:from|join)\s*\(", text, re.I) or "`" in text \
+            or "[" in text:
+        return None                                   # [方括号] 标识符、数组下标：同样不猜
+    tables: dict[str, dict[str, Any]] = {}       # 表名、别名（小写）→ 表结构
+    spans: list[tuple[int, int]] = []
+    for m in _FROM_ITEMS.finditer(text):
+        spans.append(m.span(1))
+        for item in m.group(1).split(","):
+            parsed = _TABLE_ITEM.match(item)
+            if not parsed:
+                return None
+            name, alias = parsed.group(1), parsed.group(2)
+            meta = _introspect.find_table(source, name)
+            if meta is None:
+                return None                           # 结构里找不到的表（视图、同义词、截断了的结构）不判
+            tables[name.lower()] = tables[name.rsplit(".", 1)[-1].lower()] = meta
+            if alias and alias.lower() not in _SQL_WORDS:
+                tables[alias.lower()] = meta
+    if not tables:
+        return None
+    known = list(dict.fromkeys(c["name"] for meta in {id(m): m for m in tables.values()}.values()
+                               for c in meta.get("columns") or []))
+    columns = {c.lower() for c in known}
+    # 「SELECT created_at FROM」里 created_at 前面是关键字，它是列，不是别名
+    aliases = {a.lower() for a in _ALIAS_AS.findall(text)} | {
+        m.group(2).lower() for m in _ALIAS_BARE.finditer(text) if m.group(1).lower() not in _OPERAND_BEFORE} | {
+        a.lower() for a in _WINDOW_NAME.findall(text)}
+    names = {m.start(1) for pattern in (_NAME_AFTER, _DATEPART_ARG) for m in pattern.finditer(text)}
+    unknown: list[str] = []
+    for m in _QUALIFIED.finditer(text):
+        if any(a <= m.start() < b for a, b in spans):
+            continue
+        meta = tables.get(m.group(1).lower())
+        col = m.group(2)
+        if meta is not None and col != "*" and col.lower() not in {c["name"].lower() for c in meta.get("columns") or []}:
+            unknown.append(f"{m.group(1)}.{col}")
+    for m in _SQL_IDENT.finditer(text):
+        word = m.group(1)
+        low = word.lower()
+        if any(a <= m.start() < b for a, b in spans) or low in _SQL_WORDS or low in columns or low in aliases \
+                or low in tables or m.start() in names or re.match(r"\s*[.(]", text[m.end():]) \
+                or text[m.end():m.end() + 1] == "'":      # N'…'、X'…'、_utf8mb4'…'：字面量的前缀
+            continue
+        unknown.append(word)
+    unknown = list(dict.fromkeys(unknown))
+    return (unknown, known) if unknown else None
+
+
+__all__ = ["sql_unknown_columns"]

backend/app/db/models_sources.py

@@ -19,7 +19,7 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import Boolean, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text
+from sqlalchemy import JSON, Boolean, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base, UTCDateTime, new_id, utcnow
@@ -134,4 +134,101 @@ class IntrospectTask(Base):
     finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
 
 
-__all__ = ["IntrospectJob", "IntrospectTask", "SourceObject"]
+# --------------------------------------------------------------------------
+# 第 3 波：取值观察、变化记录（docs/design/next-round/03-数据接入与目录维护.md B2、B3）
+# --------------------------------------------------------------------------
+
+
+class TableObservation(Base):
+    """一次取值观察（剖析引擎的四类观察检查，data/catalog_profile_observe.py）：码值列的取值分布（codes_observe）、最新业务
+    日期（freshness）、最近 35 天按天的行数（daily_volume）、行数（rows）。
+
+    **为什么单独一张表。** 剖析任务的逐表一行只留最近 50 个任务，而且一张表一次剖析只有一份发现；取值漂移要拿同一口径的
+    历史比（新码值、比例剧变、断档、缺天、量跌、行数跌），得按「表 × 种类 × 列」留一串。每张表每种观察（码值按列）留最近
+    KEEP 条，更早的删掉。
+
+    **口径（basis）。** 每条都记它看的是哪一部分数据：full 整表、first_rows（按存储顺序的前 N 行）、window（业务日期最近 N 天）、
+    sample（块采样 p%）、stats / count（行数的来源）。basis_key 是口径的规范写法：只和 basis_key 相同的历史比，口径不同的照实说
+    不比。遮罩列不出现在任何一条观察查询里，取值也就不会进这张表。
+
+    挂外键、级联删除：观察是数据源的附属记录。"""
+
+    __tablename__ = "table_observations"
+    __table_args__ = (
+        Index("ix_table_observations_lookup", "source_id", "table_name", "kind", "column_name", "at"),
+    )
+
+    #: 每张表每种观察（码值按列）留几条
+    KEEP = 26
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
+    source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE"))
+    table_name: Mapped[str] = mapped_column(String(255))
+    #: 码值、业务日期观察的那一列；表级的（行数）是空串
+    column_name: Mapped[str] = mapped_column(String(255), default="")
+    #: codes_observe | freshness | daily_volume | rows
+    kind: Mapped[str] = mapped_column(String(20))
+    basis: Mapped[dict[str, Any]] = mapped_column(default=dict)
+    basis_key: Mapped[str] = mapped_column(String(200), default="")
+    #: 观察到的：码值 {values: [[码, 行数]…], rows, truncated}；最新业务日期 {max}；按日 {since, until, days: {日: 行数}}；
+    #: 行数 {rows, at_least, method}
+    data: Mapped[dict[str, Any]] = mapped_column(default=dict)
+    #: 哪个剖析任务的第几张表测到的
+    job_id: Mapped[str | None] = mapped_column(String(32), default=None)
+    task_seq: Mapped[int | None] = mapped_column(Integer, default=None)
+    at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+
+class SourceChange(Base):
+    """数据源的一条变化（data/source_changes.py，格式 agentlab-change/1）：结构变化（重新探查后的前后差异）和取值变化（取值
+    观察和同口径历史的比较）。全站只有这一份记录：审阅页的「变化」、数据源卡片的角标、外部通知、口径账本的召回都读它。
+
+    - seq：自增，不复用（sqlite_autoincrement），召回按它当游标拉取；id 是给界面、链接用的编号；
+    - kind：结构类 table_added / table_removed / table_left_scope / table_reflect_failed / column_added / column_removed /
+      column_type_changed / column_nullable_changed / key_changed / fk_changed / comment_changed / scope_changed；取值类
+      codes_new / codes_shift / date_stalled / date_gap / volume_drop / rows_drop / rows_jump / coverage_drop；
+    - severity：info / warning / action（现在大概率已经不对了：已确认的项、写死的 SQL 指向被删的列，或者「已列全」被撤销）；
+    - detected_by：{job_kind, job_id, task_seq, trigger: manual|schedule|on_demand, actor}；basis：{method, query_refs}；
+    - before / after：前后各是什么（码值只记非遮罩列的）；affects：{catalog: [{table, path, status, source}], templates:
+      [{workflow_id, name, version, impact, nodes}], rules_touched}；struct_version_before / after：结构变化前后的结构版本；
+    - fingerprint：去重——同一个变化再次测到只更新 last_seen_at 和 seen_count；
+    - status：open / acknowledged / resolved / dismissed；resolution：{by, at, action, note, catalog_version, job_id}。
+
+    挂外键、级联删除：变化是数据源的附属记录。"""
+
+    __tablename__ = "source_changes"
+    __table_args__ = (
+        UniqueConstraint("source_id", "fingerprint"),
+        Index("ix_source_changes_source_status", "source_id", "status"),
+        Index("ix_source_changes_source_table", "source_id", "table_name"),
+        {"sqlite_autoincrement": True},
+    )
+
+    seq: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    id: Mapped[str] = mapped_column(String(32), unique=True, default=new_id)
+    source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE"))
+    #: 测到时这个源叫什么（之后改了名，这条记录说的仍是当时的名字）
+    source_name: Mapped[str] = mapped_column(String(100), default="")
+    #: 源级的变化（范围规则改了）是空串
+    table_name: Mapped[str] = mapped_column(String(255), default="")
+    column_name: Mapped[str | None] = mapped_column(String(255), default=None)
+    kind: Mapped[str] = mapped_column(String(40))
+    severity: Mapped[str] = mapped_column(String(10), default="info")
+    detected_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+    last_seen_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+    seen_count: Mapped[int] = mapped_column(Integer, default=1)
+    detected_by: Mapped[dict[str, Any]] = mapped_column(default=dict)
+    basis: Mapped[dict[str, Any] | None] = mapped_column(default=None)
+    before: Mapped[Any] = mapped_column(JSON, nullable=True, default=None)
+    after: Mapped[Any] = mapped_column(JSON, nullable=True, default=None)
+    affects: Mapped[dict[str, Any] | None] = mapped_column(default=None)
+    struct_version_before: Mapped[int | None] = mapped_column(Integer, default=None)
+    struct_version_after: Mapped[int | None] = mapped_column(Integer, default=None)
+    fingerprint: Mapped[str] = mapped_column(String(64))
+    status: Mapped[str] = mapped_column(String(20), default="open")
+    resolution: Mapped[dict[str, Any] | None] = mapped_column(default=None)
+    #: 记录的格式（agentlab-change/1）：召回按它读，字段以后改要升版本
+    schema_tag: Mapped[str] = mapped_column("schema", String(30), default="agentlab-change/1")
+
+
+__all__ = ["IntrospectJob", "IntrospectTask", "SourceChange", "SourceObject", "TableObservation"]

backend/app/engine/launch.py

@@ -19,7 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core import artifact_store
 from app.core.events import EventType
-from app.data.catalog_impact import catalog_drift
+from app.data.catalog_impact import catalog_drift, schema_drift
 from app.apps.versions import is_published, subgraph_pin_problems
 from app.db.models import Run, Workflow, WorkflowVersion
 from app.engine.governance import governance_note, unresolved_caliber_upgrades
@@ -165,9 +165,13 @@ async def launch_run(
         # 数据目录和口径卡不同：目录是给人和助手看的说明，不是运行的输入。发布之后变了只提醒、不拦——
         # 拦下来等于让一次文字修订卡住正式出具
         catalog_changes = await catalog_drift(session, snapshot.catalog_versions)
+        # 表结构同理（next-round 03 B1）：发布之后结构版本变了只提醒、不拦——只是加了一列不该卡住正式出具，真删了列查询
+        # 本身会照实报错
+        schema_changes = await schema_drift(session, snapshot.schema_versions)
     else:
         upgrade_events = []
         catalog_changes = []
+        schema_changes = []
         if workflow_id:
             workflow = await session.get(Workflow, workflow_id)
             if not workflow:
@@ -197,6 +201,10 @@ async def launch_run(
         # SQL 用到的，possible Agent 可能查询的（界面分开说）
         notes.append((EventType.CATALOG_DRIFT, None,
                       {"version": run_version, "count": len(catalog_changes), "tables": catalog_changes}))
+    if schema_changes:
+        # 回看这次正式出具，能知道当时的表结构和发布时不一样：哪几张表、发布时和现在的结构版本、还没处理的变化
+        notes.append((EventType.SCHEMA_DRIFT, None,
+                      {"version": run_version, "count": len(schema_changes), "tables": schema_changes}))
     if capability_changes:
         # 能力的治理（读写效果、审批、遮罩……）发布之后改过：同目录漂移，只提醒、不拦，之后回看这次运行知道差在哪
         notes.append(("log", None, {

backend/tests/test_schema_changes.py

@@ -0,0 +1,263 @@
+"""表结构变化：前后差异进变化记录、影响面、结构版本与 schema.drift、发布前检查（next-round 03 B1）。
+
+用一个 SQLite 夹具（虚构的店铺库：stores、orders、refunds）走一遍真实路径：后台探查建基线 → 写目录（人工确认的码值、
+有效记录条件、关联关系）→ 发布一个写死 SQL 的模板（记下结构版本）→ ALTER（删列、加列、改名、改类型、加注释）→ 重新探查：
+
+- 变化恰好就是这几条，疑似改名只说「可能」；第一次探查只建基线、不报变化；
+- 删掉的列上人工确认过的项、写死这一列的已发布模板都列进影响面，这条变化是「要处理」（action），有效记录条件用到它记
+  rules_touched；
+- 再探查一次（结构没变）不重复记；表真删了记 table_removed，范围外 ≠ 删除；
+- 从这一版发起的正式运行写一条 schema.drift、照常发起；发布前检查列出这张表上还没处理的结构变化（警告，不拦）。
+"""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+import uuid
+from pathlib import Path
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import select
+
+from app.data import catalog
+from app.db.base import SessionLocal
+from app.db.models import Run, RunEvent, Workflow, WorkflowVersion
+from app.db.models_sources import SourceChange, SourceObject
+from app.engine.runner import run_manager
+from app.main import app
+from tests.fixtures.sources import drop_source, introspect_now
+
+SQL = "SELECT o.status, o.amount FROM orders o WHERE o.status = 'paid'"
+
+
+def _build(path: Path) -> None:
+    db = sqlite3.connect(path)
+    db.executescript("""
+        CREATE TABLE stores (id INTEGER PRIMARY KEY, name VARCHAR(40));
+        CREATE TABLE orders (id INTEGER PRIMARY KEY, store_id INTEGER REFERENCES stores(id), status VARCHAR(10),
+                             amount NUMERIC(12, 2), note VARCHAR(40), paid_on DATE);
+        CREATE TABLE refunds (id INTEGER PRIMARY KEY, order_id INTEGER, amount NUMERIC(12, 2));
+        INSERT INTO stores VALUES (1, 'east'), (2, 'west');
+        INSERT INTO orders VALUES (1, 1, 'paid', 10.5, 'a', '2026-10-01'), (2, 2, 'void', 3, 'b', '2026-10-02');
+    """)
+    db.commit()
+    db.close()
+
+
+def _alter(path: Path) -> None:
+    """删 status、加 channel、note 改名为 remark（同位置同类型）、amount 改类型（SQLite 没有改类型：整表重建）。"""
+    db = sqlite3.connect(path)
+    db.executescript("""
+        CREATE TABLE orders_new (id INTEGER PRIMARY KEY, store_id INTEGER REFERENCES stores(id),
+                                 amount NUMERIC(14, 2), remark VARCHAR(40), paid_on DATE, channel VARCHAR(8));
+        INSERT INTO orders_new SELECT id, store_id, amount, note, paid_on, 'web' FROM orders;
+        DROP TABLE orders;
+        ALTER TABLE orders_new RENAME TO orders;
+    """)
+    db.commit()
+    db.close()
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+async def shop(client, tmp_path):
+    path = tmp_path / "shop.db"
+    _build(path)
+    name = f"shop_{uuid.uuid4().hex[:8]}"
+    r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": str(path),
+                                                    "readonly": True})
+    assert r.status_code == 201, r.text
+    sid = r.json()["id"]
+    made: list[str] = []
+    yield {"id": sid, "name": name, "path": path, "made": made}
+    async with SessionLocal() as session:
+        for wid in made:
+            if (wf := await session.get(Workflow, wid)) is not None:
+                await session.delete(wf)
+        await session.commit()
+    await drop_source(sid)
+
+
+async def _changes(sid: str) -> list[SourceChange]:
+    async with SessionLocal() as session:
+        return list((await session.execute(select(SourceChange).where(SourceChange.source_id == sid)
+                                           .order_by(SourceChange.seq))).scalars())
+
+
+async def _write_notes(sid: str) -> None:
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, sid, "orders", {
+            "label": catalog.make_item("订单", "human"),
+            "valid_filter": catalog.make_item("status <> 'void'", "human"),
+            "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human", complete=True)},
+                        "note": {"label": catalog.make_item("备注", "llm")}},
+        }, if_version=0, actor="测试员")
+        await catalog.write_entry(session, sid, "refunds", {
+            "relations": [{"id": catalog.relation_id("refunds", ["order_id"], "orders", ["id"]),
+                           "columns": ["order_id"], "to_table": "orders", "to_columns": ["id"],
+                           "cardinality": "many_to_one", "source": "human", "status": "confirmed"}],
+        }, if_version=0, actor="测试员")
+
+
+def _graph(source: str) -> dict:
+    def node(nid, ntype, **config):
+        return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+    nodes = [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, shop) -> str:
+    r = await client.post("/api/workflows", json={"name": f"订单日报-{uuid.uuid4().hex[:6]}",
+                                                  "graph": _graph(shop["name"])})
+    wf = r.json()["id"]
+    shop["made"].append(wf)
+    body = (await client.post(f"/api/workflows/{wf}/publish", json={"level": "published"})).json()
+    assert body["ok"] is True, body
+    return wf
+
+
+async def _settle(run_id: str) -> Run:
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            return row
+    raise AssertionError(f"run {run_id} 没跑完：{row.status}")
+
+
+async def test_first_introspect_is_baseline_and_alter_reports_exactly_the_changes(client, shop):
+    sid = shop["id"]
+    await introspect_now(client, sid)
+    assert await _changes(sid) == [], "第一次探查只建基线：不报「新出现的表」，也不报列"
+    async with SessionLocal() as session:
+        orders = (await session.execute(select(SourceObject).where(
+            SourceObject.source_id == sid, SourceObject.name == "orders"))).scalar_one()
+    assert orders.struct_version == 1
+    await _write_notes(sid)
+    wf = await _publish(client, shop)
+    async with SessionLocal() as session:
+        snap = (await session.execute(select(WorkflowVersion).where(
+            WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one()
+    assert snap.schema_versions == {shop["name"]: {"orders": 1}}, "只记写死的 SQL 直接引用的表"
+    detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
+    assert detail["schema_versions"] == {shop["name"]: {"orders": 1}} and detail["schema_changes"] == []
+
+    _alter(shop["path"])
+    await introspect_now(client, sid)
+    got = [(c.kind, c.column_name) for c in await _changes(sid) if c.table_name == "orders"]
+    assert sorted(got) == sorted([("column_removed", "status"), ("column_removed", "note"), ("column_added", "remark"),
+                                  ("column_added", "channel"), ("column_type_changed", "amount")]), got
+    by = {(c.kind, c.column_name): c for c in await _changes(sid)}
+    removed = by[("column_removed", "status")]
+    assert removed.struct_version_before == 1 and removed.struct_version_after == 2
+    assert removed.severity == "action", "人工确认过的码值、有效记录条件、写死它的模板都还说着这一列"
+    paths = {(i["table"], i["path"]) for i in removed.affects["catalog"]}
+    assert ("orders", "columns.status.codes") in paths and ("orders", "valid_filter") in paths
+    assert removed.affects["rules_touched"] is True, "有效记录条件用到了被删的列"
+    assert all(i["needs_action"] for i in removed.affects["catalog"] if i["status"] == "confirmed")
+    [tpl] = removed.affects["templates"]
+    assert tpl["workflow_id"] == wf and tpl["impact"] == "direct"
+    [fetch] = [n for n in tpl["nodes"] if n["node_id"] == "fetch"]
+    assert fetch["breaks"] is True and any(c.endswith("status") for c in fetch["unknown_columns"]), fetch
+    # 疑似改名只说「可能」：note → remark 同位置同类型
+    note = by[("column_removed", "note")]
+    assert note.before["rename"] == {"from": "note", "to": "remark"} and note.kind == "column_removed"
+    assert note.severity != "action", "只有推断的中文名说着这一列：不是要处理"
+    assert by[("column_type_changed", "amount")].before == "NUMERIC(12, 2)"
+    assert by[("column_type_changed", "amount")].after == "NUMERIC(14, 2)"
+    assert all(c.detected_by["job_kind"] == "introspect" and c.detected_by["trigger"] == "manual"
+               for c in await _changes(sid))
+
+    # 结构没再变：再探查一次不重复记
+    count = len(await _changes(sid))
+    await introspect_now(client, sid)
+    assert len(await _changes(sid)) == count
+
+    # 正式运行：写一条 schema.drift，照常发起
+    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)
+    async with SessionLocal() as session:
+        drift = [e.data for e in (await session.execute(select(RunEvent).where(
+            RunEvent.run_id == run_id, RunEvent.type == "schema.drift"))).scalars()]
+    assert len(drift) == 1 and drift[0]["count"] == 1
+    [table] = drift[0]["tables"]
+    assert table["table"] == "orders" and table["published"] == 1 and table["current"] == 2 and not table["removed"]
+    assert {c["kind"] for c in table["changes"]} >= {"column_removed", "column_type_changed"}
+    detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
+    assert [(c["table"], c["published"], c["current"]) for c in detail["schema_changes"]] == [("orders", 1, 2)]
+
+    # 发布前检查：这一版直接引用的 orders 上有还没处理的结构变化（警告，不拦）
+    check = (await client.post(f"/api/workflows/{wf}/publish-check", json={"level": "published"})).json()
+    [issue] = [i for i in check["issues"] if i.get("code") == "schema.change_open"]
+    assert issue["level"] == "warning" and issue["node_id"] == "fetch" and "orders" in issue["message"]
+
+
+async def test_table_removed_is_a_deletion_with_impact(client, shop):
+    sid = shop["id"]
+    await introspect_now(client, sid)
+    await _write_notes(sid)
+    db = sqlite3.connect(shop["path"])
+    db.execute("DROP TABLE refunds")
+    db.commit()
+    db.close()
+    await introspect_now(client, sid)
+    [gone] = [c for c in await _changes(sid) if c.kind == "table_removed"]
+    assert gone.table_name == "refunds" and gone.struct_version_before == 1
+    assert gone.severity == "action", "人工确认过的关联关系还在这张表上"
+    assert ("refunds", f"relations.{catalog.relation_id('refunds', ['order_id'], 'orders', ['id'])}") in {
+        (i["table"], i["path"]) for i in gone.affects["catalog"]}
+
+
+async def test_leaving_scope_is_not_a_deletion(client, shop):
+    sid = shop["id"]
+    await introspect_now(client, sid)
+    await _write_notes(sid)
+    r = await client.put(f"/api/datasources/{sid}/introspect-scope",
+                         json={"scope": {"mode": "rules", "include": ["stores", "orders"], "removed": ["refunds"]}})
+    assert r.status_code == 200, r.text
+    await introspect_now(client, sid)
+    kinds = {(c.kind, c.table_name) for c in await _changes(sid)}
+    assert ("table_left_scope", "refunds") in kinds, "有目录的表离开了范围"
+    assert ("table_removed", "refunds") not in kinds, "范围外 ≠ 删除"
+    assert any(k == "scope_changed" for k, _ in kinds), "范围规则和上一次探查不同"
+
+
+async def test_first_introspect_after_upgrade_only_builds_a_baseline(client, shop):
+    """升级前同步探查写的老缓存（类型是通用写法 VARCHAR、没有 scope 摘要）：第一次按对象索引读到只建基线，不把写法的差异
+    说成「类型变了」。"""
+    from app.db.models import DataSource
+
+    sid = shop["id"]
+    legacy = {"tables": {"orders": {"qualified": "orders", "schema": None, "comment": None, "is_view": False,
+                                    "primary_key": ["id"],
+                                    "columns": [{"name": "id", "type": "INTEGER", "nullable": False},
+                                                {"name": "status", "type": "VARCHAR", "nullable": True}]}},
+              "truncated": False, "total": 1, "schema": None, "synced_at": "2026-09-01T00:00:00+00:00"}
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        row.schema_cache = legacy
+        await session.commit()
+    await introspect_now(client, sid)
+    assert await _changes(sid) == [], "老缓存接进来的结构版本是 0：这一次只建基线"
+    async with SessionLocal() as session:
+        orders = (await session.execute(select(SourceObject).where(
+            SourceObject.source_id == sid, SourceObject.name == "orders"))).scalar_one()
+    assert orders.struct_version == 1

backend/tests/test_schema_diff.py

@@ -0,0 +1,124 @@
+"""表结构的前后差异（app/data/schema_diff.py，next-round 03 B1）：纯函数。
+
+守的是：加列、删列、类型、可空、注释（列和表）、主键、唯一约束、外键各出一条；疑似改名只说「可能」（类型和位置都对上、
+而且只有一对）；读不到唯一约束、外键的一边不比；离开范围、读不到结构不算删除；升级后第一次读到（版本 0）只建基线。
+表、列都是虚构的通用名字。
+"""
+from __future__ import annotations
+
+import copy
+
+from app.data import schema_diff as sd
+
+
+def _meta(**over):
+    meta = {
+        "qualified": "orders", "schema": None, "comment": "订单", "is_view": False,
+        "columns": [
+            {"name": "id", "type": "INTEGER", "nullable": False, "comment": None},
+            {"name": "status", "type": "VARCHAR(10)", "nullable": True, "comment": "状态"},
+            {"name": "amount", "type": "NUMERIC(12, 2)", "nullable": True, "comment": None},
+            {"name": "paid_at", "type": "TIMESTAMP", "nullable": True, "comment": None},
+        ],
+        "primary_key": ["id"],
+        "unique": [],
+        "foreign_keys": [{"columns": ["store_id"], "to_table": "stores", "to_columns": ["id"]}],
+    }
+    meta.update(over)
+    return meta
+
+
+def _kinds(diffs):
+    return [(d.kind, d.column) for d in diffs]
+
+
+def test_same_structure_has_no_changes():
+    assert sd.diff_table(_meta(), _meta()) == []
+
+
+def test_added_removed_type_nullable_and_comment_each_get_one_change():
+    before = _meta()
+    after = copy.deepcopy(before)
+    cols = after["columns"]
+    cols[1]["type"] = "VARCHAR(20)"          # 类型变长
+    cols[2]["nullable"] = False               # 改成非空
+    cols[3]["comment"] = "支付时间"            # 补了注释
+    cols.append({"name": "channel", "type": "VARCHAR(8)", "nullable": True, "comment": None})
+    after["columns"] = [c for c in cols if c["name"] != "amount"] + []
+    after["columns"].insert(2, {"name": "amount", "type": "NUMERIC(12, 2)", "nullable": False, "comment": None})
+    after["comment"] = "订单明细"
+    diffs = sd.diff_table(before, after)
+    assert _kinds(diffs) == [
+        ("column_added", "channel"),
+        ("column_type_changed", "status"), ("column_nullable_changed", "amount"),
+        ("comment_changed", "paid_at"), ("comment_changed", None),
+    ]
+    by = {(d.kind, d.column): d for d in diffs}
+    assert by[("column_type_changed", "status")].before == "VARCHAR(10)"
+    assert by[("column_type_changed", "status")].after == "VARCHAR(20)"
+    assert by[("column_nullable_changed", "amount")].before is True and by[("column_nullable_changed", "amount")].after is False
+    assert by[("comment_changed", None)].after == "订单明细"
+
+
+def test_removed_column_is_a_deletion_and_left_scope_is_not():
+    after = _meta(columns=[c for c in _meta()["columns"] if c["name"] != "amount"])
+    [d] = sd.diff_table(_meta(), after)
+    assert d.kind == "column_removed" and d.column == "amount" and d.before["type"] == "NUMERIC(12, 2)"
+    assert d.rename is None, "没有新加的列，谈不上改名"
+    assert sd.is_deletion("column_removed") and sd.is_deletion("table_removed")
+    assert not sd.is_deletion("table_left_scope") and not sd.is_deletion("table_reflect_failed"), "范围外、读不到不算删除"
+    assert d.severity == "warning"
+
+
+def test_same_type_same_position_is_only_a_possible_rename():
+    after = copy.deepcopy(_meta())
+    after["columns"][1] = {"name": "state", "type": "VARCHAR(10)", "nullable": True, "comment": "状态"}
+    diffs = sd.diff_table(_meta(), after)
+    assert _kinds(diffs) == [("column_removed", "status"), ("column_added", "state")]
+    assert diffs[0].rename == {"from": "status", "to": "state"} == diffs[1].rename
+    assert diffs[0].kind == "column_removed", "改名只说「可能」：照样是一条删列、一条加列"
+
+
+def test_rename_needs_same_type_and_same_position():
+    moved = copy.deepcopy(_meta())
+    moved["columns"] = [moved["columns"][0], moved["columns"][2], moved["columns"][3],
+                        {"name": "state", "type": "VARCHAR(10)", "nullable": True, "comment": None}]
+    diffs = sd.diff_table(_meta(), moved)
+    assert all(d.rename is None for d in diffs), "位置不同：不说改名"
+    retyped = copy.deepcopy(_meta())
+    retyped["columns"][1] = {"name": "state", "type": "INTEGER", "nullable": True, "comment": None}
+    assert all(d.rename is None for d in sd.diff_table(_meta(), retyped)), "类型不同：不说改名"
+
+
+def test_keys_and_foreign_keys():
+    after = _meta(primary_key=["id", "status"], unique=[["paid_at"]],
+                  foreign_keys=[{"columns": ["store_id"], "to_table": "shops", "to_columns": ["id"]}])
+    diffs = sd.diff_table(_meta(), after)
+    assert [(d.kind, d.key, d.column) for d in diffs] == [
+        ("key_changed", "primary", None), ("key_changed", "unique", "paid_at"),
+        ("fk_changed", None, "store_id"), ("fk_changed", None, "store_id")]
+    fk_gone, fk_new = diffs[2], diffs[3]
+    assert fk_gone.before["to_table"] == "stores" and fk_gone.after is None
+    assert fk_new.after["to_table"] == "shops" and fk_new.before is None
+
+
+def test_constraints_not_read_on_one_side_are_not_compared():
+    """反射读不到唯一约束、外键时不写那两个键：不知道 ≠ 删了。"""
+    after = _meta()
+    del after["unique"], after["foreign_keys"]
+    assert sd.diff_table(_meta(), after) == []
+
+
+def test_baseline_after_upgrade_reports_nothing():
+    """升级前的老缓存接进来的结构版本是 0：第一次按对象索引读到只建基线，通用写法换成方言完整写法不算类型变了。"""
+    legacy = _meta()
+    assert sd.is_baseline(0, legacy) and sd.is_baseline(None, legacy)
+    assert sd.is_baseline(3, None) and sd.is_baseline(3, {"columns": []}), "上一次的结构不在手上：没法比"
+    assert not sd.is_baseline(1, legacy)
+
+
+def test_to_dict_keeps_rename_and_key():
+    d = sd.Diff("key_changed", None, before=["id"], after=None, key="primary")
+    assert d.to_dict() == {"kind": "key_changed", "column": None, "before": ["id"], "after": None, "key": "primary"}
+    assert sd.touched_columns([sd.Diff("column_removed", "a"), sd.Diff("column_type_changed", "b"),
+                               sd.Diff("column_added", "c")]) == {"removed": {"a"}, "changed": {"b"}}