W3-a 变化记录的接口与处理、定时维护（next-round 03 B3、B4）

变化没有统一的记录：进不了审阅，发不了通知，口径账本的召回也没有可以订阅的接口；目录只能手动探查、手动剖析。现在：

- api/source_changes.py（骨架已在 datasources 通配之前登记，搭建者可用，访问密钥不放）：GET /api/datasources/{id}/changes
  （缺省开着的和知道了的，按「使用次数权重 × 严重程度」排，带各级别计数）、GET /changes/{change_id}、POST
  /changes/{change_id}/resolve。处理动作落到目录已有的写入接口：清理目录里说到被删的列的项（整份写、乐观锁）、加入码值
  （apply_patch 配 _merge_codes，含义待填写，依据写变化编号）、重新起草这张表（模型用不了时只按表结构起草）、重新剖析、
  知道了、误报（必须写理由）；每个动作写 resolution，动了目录的记目录版本，建了任务的记任务编号。
- data/source_changes.py 给召回：list_changes（按 seq 游标拉）、affecting（按源和表反查）、resolve、triggers_recall（只认
  结构类、处理时改了目录的、触及有效记录条件的新码值）。
- 通知 source.changed 的发送函数 notify/service.notify_source_changed：只推还要人看的「要处理」级变化，按源汇总，消息里
  只有源名、各级别条数和链接（D9）；探查、剖析任务结束时有新的要处理变化就调。这一行的订阅范围第 4 波有了界面才打开，
  现在推不出去（照实）。
- 定时维护（data/source_maintenance.py，X4：就是一条 target_kind=source_maintenance 的定时，不另建表）：fire 按顺序建
  探查任务 → 剖析任务（范围 needed / observe / maintenance 按选的步骤），任务上记 trigger=schedule、fire_at、schedule_id；
  剖析等探查进了同源队才建，靠 SourceLanes 串行；剖析没开、数据源停用 / 不存在 / 是上传表格时照实记没执行，同源还有
  探查或剖析在做时记跳过；触发记录的 inputs 写明建了哪几个任务、哪几步没执行。check_target 核对目标。维护步骤存在
  options.catalog_maintenance，经 GET / PUT /api/datasources/{id}/maintenance 改，数据源表单不显示它（api/datasources.py
  的 _MANAGED_OPTIONS 加了这一个键）。

测试：test_source_changes（列表与计数、误报要理由、处理过的 409、清理目录与加入码值记目录版本、剖析没开照实 422、
去重、游标拉取与反查、召回触发、通知只有源名条数链接）、test_source_maintenance（按顺序串行、trigger 与 fire_at、
剖析没开记未执行、停用 / 不存在 / 忙、调度循环只补最近一次且积压只记一条、步骤接口与核对目标）。

backend/app/api/datasources.py

@@ -140,8 +140,9 @@ def _import_mode(row: DataSource) -> str | None:
 #: options 里由别的接口维护的项：数据源表单不显示、不提交它们，整份回写 options 时原样留着（同模型接入 extra 里的
 #: last_check）。数据目录的起草要求只经 PUT /catalog/guidance 改（api/catalog.py）：表单把认不得的项放进「高级参数」
 #: 的单行输入框，多行的要求会被压成一行
-#: 探查范围（introspect_scope）同理：只经 PUT /introspect-scope 改，表单整份回写 options 时原样留着
-_MANAGED_OPTIONS = (CATALOG_DRAFT_GUIDANCE_OPTION, source_objects.SCOPE_OPTION)
+#: 探查范围（introspect_scope）同理：只经 PUT /introspect-scope 改，表单整份回写 options 时原样留着；定时维护的步骤
+#: （catalog_maintenance）只经 PUT /maintenance 改（api/source_changes.py）
+_MANAGED_OPTIONS = (CATALOG_DRAFT_GUIDANCE_OPTION, source_objects.SCOPE_OPTION, "catalog_maintenance")
 
 
 def _shown_options(options: dict[str, Any] | None) -> dict[str, Any]:

backend/app/api/source_changes.py

@@ -1,9 +1,125 @@
-"""数据源的变化记录（docs/design/next-round/03-数据接入与目录维护.md B3）：结构变化、取值变化，查看与处理。
+"""数据源的变化记录（docs/design/next-round/03-数据接入与目录维护.md B3）：结构变化、取值变化，查看与处理；以及同一前缀下
+数据源定时维护要做哪几步（B4）。
 
-挂在 /api/datasources/{id}/changes 下，要在 datasources 的通配 /{source_id} 之前登记（main.py）。第 3 波 W3-a 填。
+挂在 /api/datasources/{id}/… 下，要在 datasources 的通配 /{source_id} 之前登记（main.py）：
+
+- GET  /{id}/changes                 变化列表（缺省开着的和知道了的；status=all 全部）：按「使用次数权重 × 严重程度」排，
+                                     同分新的在前；带各级别的计数（数据源卡片的角标「N 个变化要处理」按它，第 4 波）
+- GET  /{id}/changes/{change_id}     一条变化
+- POST /{id}/changes/{change_id}/resolve  处理：{action, note?, codes?}。动作落到目录已有的写入接口并记目录版本
+                                     （data/source_changes.handle）；误报必须写理由（422）；已经处理过的 409
+- GET  /{id}/maintenance             定时维护要做哪几步、剖析开没开、指向这个源的维护定时
+- PUT  /{id}/maintenance             改要做哪几步：{steps: [introspect | profile_needed | observe]}。定时本身（cron、时区、开关）
+                                     在 /api/schedules 里建（target_kind=source_maintenance），这里只管步骤
+
+访问密钥默认不放（auth/keys.py 的 KEY_ROUTES 没有登记这几条）。界面（审阅页「变化」标签页、数据源角标、定时维护设置）在
+第 4 波。
 """
 from __future__ import annotations
 
-from fastapi import APIRouter
+from typing import Any
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Query
+from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import actor_of
+from app.data import catalog, source_changes as sc, source_maintenance
+from app.db.base import get_session
+from app.db.models import DataSource
 
 router = APIRouter(prefix="/api/datasources", tags=["source-changes"])
+
+#: 一页最多几条
+LIMIT_MAX = 200
+
+
+async def _source(session: AsyncSession, source_id: str) -> DataSource:
+    row = await session.get(DataSource, source_id)
+    if row is None:
+        raise HTTPException(404, "数据源不存在，可能已被删除")
+    return row
+
+
+@router.get("/{source_id}/changes")
+async def list_changes(source_id: str, status: str = Query("pending"), kind: list[str] | None = Query(None),
+                       table: str | None = Query(None), limit: int = Query(100, ge=1, le=LIMIT_MAX),
+                       offset: int = Query(0, ge=0), session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """{items: [变化（agentlab-change/1）], total, counts: {pending, open, by_severity}}。"""
+    if status not in ("pending", "all", *sc.STATUSES):
+        raise HTTPException(422, f"不认识的状态：{status}。可选：pending、all、{'、'.join(sc.STATUSES)}")
+    if kind and (bad := [k for k in kind if k not in sc.KINDS]):
+        raise HTTPException(422, f"不认识的变化种类：{'、'.join(bad)}")
+    row = await _source(session, source_id)
+    rows, total = await sc.list_for_source(session, row, status=status, kinds=kind, table=table, limit=limit,
+                                           offset=offset)
+    return {"items": [sc.change_out(r) for r in rows], "total": total, "counts": await sc.counts(session, row.id)}
+
+
+@router.get("/{source_id}/changes/{change_id}")
+async def get_change(source_id: str, change_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    row = await _source(session, source_id)
+    found = await sc.load(session, row.id, change_id)
+    if found is None:
+        raise HTTPException(404, "这条变化不存在，可能已随数据源删除")
+    return {"change": sc.change_out(found)}
+
+
+class ResolveIn(BaseModel):
+    action: str
+    #: 理由：误报必须写
+    note: str | None = Field(default=None, max_length=sc.NOTE_MAX)
+    #: 加入码值时的 {码值: 含义}（含义可以空）；不给就加这条变化测到的新码值
+    codes: dict[str, str] | None = None
+
+
+@router.post("/{source_id}/changes/{change_id}/resolve")
+async def resolve_change(source_id: str, change_id: str, payload: ResolveIn, x_actor: str | None = Header(default=None),
+                         session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """处理一条变化：clean_catalog 清理目录里说到这些列的项、add_codes 加入码值、redraft 重新起草这张表、reprofile 重新
+    剖析、acknowledge 知道了、dismiss 误报（必须写理由）。返回 {change, catalog_version?, job?}。目录被别人刚改过 409。"""
+    row = await _source(session, source_id)
+    found = await sc.load(session, row.id, change_id)
+    if found is None:
+        raise HTTPException(404, "这条变化不存在，可能已随数据源删除")
+    try:
+        return await sc.handle(session, row, found, payload.action, actor=actor_of(x_actor), note=payload.note,
+                               codes=payload.codes)
+    except sc.ChangeInvalid as e:
+        await session.rollback()
+        raise HTTPException(422, str(e)) from e
+    except sc.ChangeFinished as e:
+        await session.rollback()
+        raise HTTPException(409, str(e)) from e
+    except catalog.CatalogConflict as e:
+        await session.rollback()
+        raise HTTPException(409, "这张表的数据目录刚被别人改过，请刷新后再处理") from e
+
+
+@router.get("/{source_id}/maintenance")
+async def get_maintenance(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """{steps, profile_enabled, schedules}：要做哪几步、剖析开没开（没开时剖析、观察两步到点照实记没执行）、指向这个源的
+    维护定时。"""
+    from app.data import catalog_profile
+
+    row = await _source(session, source_id)
+    return {"steps": source_maintenance.steps_of(row.options),
+            "profile_enabled": catalog_profile.profile_settings(row.options).enabled,
+            "schedules": await source_maintenance.schedules_for(session, row.id)}
+
+
+class MaintenanceIn(BaseModel):
+    steps: list[Any]
+
+
+@router.put("/{source_id}/maintenance")
+async def put_maintenance(source_id: str, payload: MaintenanceIn, x_actor: str | None = Header(default=None),
+                          session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    row = await _source(session, source_id)
+    if (row.origin or "manual") in ("upload", "records"):
+        raise HTTPException(409, source_maintenance.NOT_MAINTAINED)
+    if problem := source_maintenance.steps_problem(payload.steps):
+        raise HTTPException(422, problem)
+    row.options = source_maintenance.with_steps(row.options, [str(s) for s in payload.steps], actor=actor_of(x_actor))
+    await session.commit()
+    return await get_maintenance(source_id, session)

backend/app/data/catalog_profile_jobs.py

@@ -904,6 +904,16 @@ class _ProfileRun(runtime.JobRun):
             await self._write(_close, self.job_id, job_status, job_error, task_status, task_error)
         except Exception:  # noqa: BLE001
             logger.exception("数据剖析任务 %s 收尾时写库出错", self.job_id)
+        await self._notify()
+
+    async def _notify(self) -> None:
+        """这次取值观察新记了「要处理」的变化（撤销了已列全）：按源推一条 source.changed。"""
+        from app.data import source_changes
+
+        async with SessionLocal() as session:
+            job = await session.get(CatalogProfileJob, self.job_id)
+        if job is not None and job.scope in OBSERVING_SCOPES:
+            await source_changes.notify_after_job(job.source_id, self.job_id)
 
     def stop_text(self) -> str:
         return STOPPED_ERROR if self.stop_reason == "user" else INTERRUPTED_ERROR
@@ -966,6 +976,7 @@ class _ProfileRun(runtime.JobRun):
                 break
         await self._settle()
         await self._write(self._finish)
+        await self._notify()
 
     # ---------------------------------------------------------------- 进度回调（引擎调）
 

backend/app/data/introspect_jobs.py

@@ -572,6 +572,16 @@ class _IntrospectRun(runtime.JobRun):
             await self._write(_close, self.job_id, job_status, job_error, task_error)
         except Exception:  # noqa: BLE001
             logger.exception("探查任务 %s 收尾时写库出错", self.job_id)
+        await self._notify()
+
+    async def _notify(self) -> None:
+        """这次新记了「要处理」的结构变化：按源推一条 source.changed（data/source_changes.notify_after_job）。"""
+        from app.data import source_changes
+
+        async with SessionLocal() as session:
+            job = await session.get(IntrospectJob, self.job_id)
+        if job is not None:
+            await source_changes.notify_after_job(job.source_id, self.job_id)
 
     async def _run(self) -> None:
         if not await self._write(self._claim):
@@ -600,6 +610,7 @@ class _IntrospectRun(runtime.JobRun):
             await self._write(self._save_batch, source_id, schema, batch, got)
         await self._settle()
         await self._write(self._finish, source_id)
+        await self._notify()
 
     # ---------------------------------------------------------------- 库操作
 

backend/app/data/source_changes.py

@@ -325,6 +325,27 @@ def triggers_recall(change: Mapping[str, Any]) -> bool:
     return kind == CODES_NEW and bool((change.get("affects") or {}).get("rules_touched"))
 
 
+async def notify_after_job(source_id: str, job_id: str) -> int:
+    """一次探查、剖析任务结束时：这次新记了还开着的「要处理」级变化，就在后台按源推一条 source.changed（不等它推完）。返回
+    这次新记了几条要处理的。出错只记日志：通知是附加的。"""
+    from app.db.base import SessionLocal
+
+    try:
+        async with SessionLocal() as session:
+            n = int((await session.execute(select(func.count()).select_from(SourceChange).where(
+                SourceChange.source_id == source_id, SourceChange.severity == "action",
+                SourceChange.status == "open",
+                func.json_extract(SourceChange.detected_by, "$.job_id") == job_id))).scalar() or 0)
+        if n:
+            from app.notify import service
+
+            service.spawn(service.notify_source_changed(source_id))
+        return n
+    except Exception:  # noqa: BLE001
+        logger.exception("任务 %s 结束后推送数据源变化的通知时出错", job_id)
+        return 0
+
+
 # ==========================================================================
 # 处理
 # ==========================================================================
@@ -487,6 +508,7 @@ __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",
+    "change_out", "counts", "fingerprint", "handle", "list_changes", "list_for_source", "load", "notify_after_job",
+    "record", "record_all",
     "resolve", "summary", "triggers_recall", "without_paths",
 ]

backend/app/data/source_maintenance.py

@@ -1,21 +1,242 @@
-"""数据源的定时维护：探查 → 剖析需要的表 → 取值观察（docs/design/next-round/03-数据接入与目录维护.md B4）。
+"""数据源的定时维护：探查 → 剖析需要的表 → 取值观察（docs/design/next-round/03-数据接入与目录维护.md B4，目录-D5）。
 
-scheduler/service.py 按目标种类导入本模块（_TARGET_MODULES）：fire 到点发起一次，check_target 建任务时核对目标。
-第 3 波 W3-a 填；填之前照实记「当前版本还不能按时发起这类目标」，不发起。
+**为什么要有它。** 以前只能手动探查、手动剖析，目录只在点按钮那一刻是对的；结构变化、取值漂移要靠定期观察才看得到。
+
+**定时只有一套**（总方案 X4）：维护就是一条 Schedule，target_kind=source_maintenance、target_id=数据源 id。认领、错过的只补
+最近一次、积压只记一条、重叠跳过、触发记录、通知都是 scheduler 那一套（scheduler/service.py 按目标种类导入本模块：fire 到点
+发起一次、check_target 建任务时核对目标），不另建表、不另写 cron。按数据源开、缺省关（没有这条定时就不维护），只在设定的
+时段跑（cron 自己定）。
+
+**做哪几步**存在数据源的 options.catalog_maintenance（{steps: [...], updated_by, updated_at}，由 PUT
+/api/datasources/{id}/maintenance 维护，数据源表单不显示、不提交它）：introspect 探查结构、profile_needed 剖析需要的表、
+observe 取值观察，可多选，没设过是三步都做。
+
+**到点**（fire，持着 scheduler 的 launch_lock）：按顺序建探查任务 → 剖析任务（范围 needed / observe / maintenance，按选的
+步骤），任务上记 trigger=schedule 和 fire_at，靠 SourceLanes 串行：剖析排在同一次触发的探查后面（等探查进了同源队才建剖析
+任务），开始时按那时的表结构和目录挑表（挑出 0 张也照建，任务以 0 张表结束、提醒里写明为什么）。预算、止损照剖析设置。
+照实记没执行的：数据源不存在、已停用、是上传表格或记录表时整次不执行（failed_to_start）；剖析没开时剖析、观察两步不执行、
+探查照做；上一次的维护（或别的探查、剖析）还没做完时这次跳过（skipped_overlap）。每次触发记下建了哪几个任务、哪几步没执行、
+为什么（触发记录的 inputs：{steps, jobs, skipped}）。
+
+AGENTLAB_SCHEDULER 关着时调度循环不起，到点不会触发（拷来的数据目录不替原库查业务库）；「立即运行」是人点的，照常执行。
 """
 from __future__ import annotations
 
+import asyncio
+import logging
+from collections.abc import Mapping
 from datetime import datetime
 from typing import Any
 
+from sqlalchemy import select, update
+
+from app.db.base import SessionLocal, utcnow
+
+logger = logging.getLogger(__name__)
+
+#: options 里维护设置的键
+MAINTENANCE_OPTION = "catalog_maintenance"
+STEPS = ("introspect", "profile_needed", "observe")
+STEP_LABEL = {"introspect": "探查结构", "profile_needed": "剖析需要的表", "observe": "取值观察"}
+DEFAULT_STEPS = STEPS
+#: 等同一次触发的探查任务进同源队最多等几秒（进了队，剖析任务才排得到它后面）
+LANE_WAIT_S = 5.0
+
+#: 先记下这次触发、还没建完任务时的说法：进程在这中间没了，留下的记录照实说没有执行
+INTERRUPTED = "服务在建维护任务前停止，这次维护没有执行"
+SOURCE_GONE = "数据源不存在，可能已被删除"
+SOURCE_DISABLED = "数据源「{name}」已停用，这次维护没有执行"
+NOT_MAINTAINED = "上传的表格、记录表的结构随版本或表单走，不需要定时维护"
+NO_STEPS = "没有选要做的维护步骤（探查结构、剖析需要的表、取值观察至少选一步）"
+BUSY = "数据源「{name}」上一次的维护（或别的探查、剖析）还没做完：{what}，这次触发已跳过"
+NOT_RUN = "{step}没有执行：{why}"
+
+
+def steps_of(options: Mapping[str, Any] | None) -> list[str]:
+    """这个源定时维护要做哪几步（按 STEPS 的顺序）；没设过是三步都做。"""
+    raw = (options or {}).get(MAINTENANCE_OPTION)
+    steps = raw.get("steps") if isinstance(raw, Mapping) else None
+    if not isinstance(steps, list):
+        return list(DEFAULT_STEPS)
+    picked = {s for s in steps if s in STEPS}
+    return [s for s in STEPS if s in picked]
+
+
+def steps_problem(raw: Any) -> str | None:
+    """PUT /maintenance 交来的步骤合不合规：不合规返回一句中文。"""
+    if not isinstance(raw, list) or not all(isinstance(s, str) for s in raw):
+        return "维护步骤要是一组文字：introspect（探查结构）、profile_needed（剖析需要的表）、observe（取值观察）"
+    unknown = [s for s in raw if s not in STEPS]
+    if unknown:
+        return f"不认识的维护步骤：{'、'.join(unknown)}。可选：{'、'.join(f'{k}（{v}）' for k, v in STEP_LABEL.items())}"
+    if not raw:
+        return NO_STEPS
+    return None
+
+
+def with_steps(options: Mapping[str, Any] | None, steps: list[str], *, actor: str | None) -> dict[str, Any]:
+    """options 换上新的维护步骤（不改别的项）。"""
+    ordered = [s for s in STEPS if s in set(steps)]
+    return {**dict(options or {}),
+            MAINTENANCE_OPTION: {"steps": ordered, "updated_by": actor, "updated_at": utcnow().isoformat()}}
+
+
+def _origin_problem(row: Any) -> str | None:
+    return NOT_MAINTAINED if (getattr(row, "origin", None) or "manual") in ("upload", "records") else None
+
+
+async def check_target(session: Any, target_id: str) -> str | None:
+    """建定时任务、改目标时核对：数据源要在、不是上传表格或记录表、选了至少一步。停用了、没开剖析不拦（到点照实记没执行）。"""
+    from app.db.models import DataSource
+
+    row = await session.get(DataSource, target_id)
+    if row is None:
+        return SOURCE_GONE
+    if problem := _origin_problem(row):
+        return problem
+    if not steps_of(row.options):
+        return NO_STEPS
+    return None
+
+
+async def _busy(session: Any, source_id: str) -> str | None:
+    """同源有排队或进行中的探查、剖析任务：说明是哪一种。"""
+    from app.data import catalog_profile_jobs, introspect_jobs
+
+    if await introspect_jobs.active_job(session, source_id) is not None:
+        return "正在探查结构"
+    if await catalog_profile_jobs.active_job(session, source_id) is not None:
+        return "正在剖析"
+    return None
+
+
+async def _entered(source_id: str, job_id: str) -> None:
+    """等这个探查任务进了同源队（或者已经不在跑了）：剖析任务要排在它后面。最多等 LANE_WAIT_S 秒。"""
+    from app.core import job_runtime as runtime
+    from app.data import introspect_jobs
+
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + LANE_WAIT_S
+    while loop.time() < deadline:
+        if runtime.lanes.ahead(source_id, job_id) is not None or not introspect_jobs.manager.running(job_id):
+            return
+        await asyncio.sleep(0.01)
+    logger.warning("定时维护：探查任务 %s 过了 %.0f 秒还没进同源队，剖析任务照建", job_id, LANE_WAIT_S)
+
+
+async def _settle_fire(fire_id: str, status: str, reason: str | None, inputs: dict[str, Any]) -> Any:
+    from app.db.models import ScheduleFire
+
+    async with SessionLocal() as session:
+        await session.execute(update(ScheduleFire).where(ScheduleFire.id == fire_id)
+                              .values(status=status, reason=reason, inputs=inputs))
+        await session.commit()
+        return await session.get(ScheduleFire, fire_id)
+
 
 async def fire(*, schedule_id: str, fire_at: datetime, kind: str, actor: str | None, target_id: str, name: str,
                tz: str, cron_expr: str) -> Any:
+    """到点维护一次（持着 launch_lock 调用）：建探查任务、剖析任务并在后台开始，记下触发。同一次触发已经记过就交回那一条。"""
+    from app.db.models import DataSource, ScheduleFire
     from app.scheduler import service
 
-    reason = service.TARGET_UNSUPPORTED.format(kind=service.TARGET_LABEL["source_maintenance"])
-    return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", reason)
+    async with SessionLocal() as session:
+        existing = (await session.execute(select(ScheduleFire).where(
+            ScheduleFire.schedule_id == schedule_id, ScheduleFire.fire_at == fire_at,
+            ScheduleFire.kind == kind))).scalar_one_or_none()
+        if existing is not None:
+            return existing
+        row = await session.get(DataSource, target_id)
+        if row is None:
+            return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", SOURCE_GONE)
+        if problem := _origin_problem(row):
+            return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", problem)
+        if row.enabled is False:
+            return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start",
+                                             SOURCE_DISABLED.format(name=row.name))
+        steps = steps_of(row.options)
+        if not steps:
+            return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", NO_STEPS)
+        if what := await _busy(session, row.id):
+            return await service.record_fire(schedule_id, fire_at, kind, "skipped_overlap",
+                                             BUSY.format(name=row.name, what=what), {"steps": steps})
+    placeholder = await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", INTERRUPTED,
+                                            {"steps": steps})
+    if placeholder is None:
+        return None
+    try:
+        jobs, skipped = await _start(target_id, steps, actor=actor or service.actor_label(name),
+                                     fire_at=fire_at.isoformat(), schedule_id=schedule_id)
+    except Exception:  # noqa: BLE001 - 建任务出意外不能让调度循环停下：照实记下，详细信息进日志
+        logger.exception("定时维护（%s）建任务时出错", schedule_id)
+        return await _settle_fire(placeholder.id, "failed_to_start", "建维护任务时出错，详细信息见服务日志",
+                                  {"steps": steps})
+    reason = "；".join(NOT_RUN.format(step=STEP_LABEL[s["step"]], why=s["reason"]) for s in skipped) or None
+    return await _settle_fire(placeholder.id, "started" if jobs else "failed_to_start", reason,
+                              {"steps": steps, "jobs": jobs, "skipped": skipped})
 
 
-async def check_target(session: Any, target_id: str) -> str | None:
-    return "当前版本还不能按时发起「数据源维护」这类目标"
+async def _start(source_id: str, steps: list[str], *, actor: str, fire_at: str,
+                 schedule_id: str) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
+    """按顺序建任务并开始：探查（introspect）→ 剖析 / 观察（一个剖析任务：范围按选的步骤）。返回 (建了的任务, 没执行的步骤
+    和原因)。"""
+    from app.data import catalog_profile, catalog_profile_jobs, introspect_jobs, table_versions
+    from app.db.models import DataSource
+
+    jobs: list[dict[str, Any]] = []
+    skipped: list[dict[str, str]] = []
+    introspect_id: str | None = None
+    if "introspect" in steps:
+        async with SessionLocal() as session:
+            row = await session.get(DataSource, source_id)
+            try:
+                job, n = await introspect_jobs.prepare(session, row, actor=actor, trigger="schedule", fire_at=fire_at,
+                                                       schedule_id=schedule_id)
+            except (introspect_jobs.IntrospectRefused, introspect_jobs.IntrospectBusy, introspect_jobs.IntrospectFailed,
+                    introspect_jobs.ScopeTooLarge) as e:
+                skipped.append({"step": "introspect", "reason": str(e)})
+            else:
+                introspect_jobs.manager.start(job.id)
+                introspect_id = job.id
+                jobs.append({"step": "introspect", "kind": "introspect", "id": job.id, "tables": n})
+    wanted = [s for s in steps if s in ("profile_needed", "observe")]
+    if not wanted:
+        return jobs, skipped
+    if introspect_id is not None:
+        await _entered(source_id, introspect_id)
+    scope = "maintenance" if len(wanted) == 2 else ("needed" if wanted == ["profile_needed"] else "observe")
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, source_id)
+        try:
+            settings = catalog_profile.ensure_enabled(row)
+        except catalog_profile.ProfileDisabled as e:
+            skipped += [{"step": s, "reason": str(e)} for s in wanted]
+            return jobs, skipped
+        try:
+            view = await table_versions.resolve_source(session, row)
+            job, tasks = await catalog_profile_jobs.create_job(
+                session, row, view, tables=None, scope=scope, settings=settings, actor=actor, trigger="schedule",
+                fire_at=fire_at, schedule_id=schedule_id, allow_empty=True)
+        except (catalog_profile_jobs.ProfileBusy, catalog_profile_jobs.ProfileNothing,
+                table_versions.SnapshotMissing) as e:
+            skipped += [{"step": s, "reason": str(e)} for s in wanted]
+            return jobs, skipped
+        catalog_profile_jobs.manager.start(job.id)
+        jobs.append({"step": "+".join(wanted), "kind": "profile", "id": job.id, "scope": scope, "tables": len(tasks)})
+    return jobs, skipped
+
+
+async def schedules_for(session: Any, source_id: str) -> list[dict[str, Any]]:
+    """指向这个数据源的维护定时（数据源设置里「定时维护」的快捷入口读它，第 4 波）：[{id, name, cron, timezone, enabled,
+    next_run_at}]。"""
+    from app.db.models import Schedule
+
+    rows = (await session.execute(select(Schedule).where(
+        Schedule.target_kind == "source_maintenance", Schedule.target_id == source_id)
+        .order_by(Schedule.created_at))).scalars()
+    return [{"id": r.id, "name": r.name, "cron": r.cron, "timezone": r.timezone, "enabled": r.enabled,
+             "next_run_at": r.next_run_at.isoformat() if r.next_run_at else None} for r in rows]
+
+
+__all__ = ["DEFAULT_STEPS", "MAINTENANCE_OPTION", "STEPS", "STEP_LABEL", "check_target", "fire", "schedules_for",
+           "steps_of", "steps_problem", "with_steps"]

backend/app/notify/service.py

@@ -357,3 +357,33 @@ async def notify_action(action_id: str, event: str) -> int:
                "subject": {"action_id": action_id, "run_id": rec.run_id, "task_id": task_id}}
     return await _push(channels, message, subject_kind="action", subject_id=action_id, event=event,
                        run_id=rec.run_id, task_id=task_id)
+
+
+async def notify_source_changed(source_id: str) -> int:
+    """数据源有要处理的变化（source.changed，03 B3 ④、目录-D9）：只推还要人看（开着、知道了还没处理）的「要处理」级变化，
+    按源汇总；消息里只有源名、各级别条数和链接，不放表名、列名、取值。探查、剖析任务结束时有新的「要处理」变化才调。
+
+    这一行的 scopes 第 4 波有了「变化」标签页才打开：现在 _channels 推不出去，这里返回 0（照实：没有界面看不了的通知不发）。"""
+    from app.data import source_changes
+    from app.db.models import DataSource
+
+    event = "source.changed"
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, source_id)
+        if row is None:
+            return 0
+        counts = await source_changes.counts(session, source_id)
+        by = counts["by_severity"]
+        if not by.get("action"):
+            return 0
+        channels = await _channels(session, event, None, [])
+        if not channels:
+            return 0
+        base = await _base_url(session)
+    lines = [("数据源", f"「{row.name}」"), ("要处理", f"{by['action']} 条"), ("提醒", f"{by.get('warning', 0)} 条"),
+             ("说明", f"{by.get('info', 0)} 条")]
+    message = {"event": event, "status": "action", "title": f"数据源「{row.name[:80]}」有 {by['action']} 个变化要处理",
+               "lines": lines, "links": [("去看看", f"{base}/data/catalog/{source_id}")] if base else [],
+               "subject": {"source_id": source_id}}
+    return await _push(channels, message, subject_kind="source", subject_id=source_id, event=event, run_id=None,
+                       task_id=None)

backend/tests/test_source_changes.py

@@ -0,0 +1,239 @@
+"""数据源的变化记录：接口、处理动作、给召回的拉取、通知（next-round 03 B3）。
+
+SQLite 夹具（虚构的店铺库）探查建基线、写目录，再删一列、重新探查得到真实的结构变化；取值变化（新码值）直接经
+source_changes.record 记一条。守的是：
+
+- GET /changes 缺省列开着的和知道了的，带各级别计数；状态、种类筛选；不认识的状态 422；
+- 处理：误报必须写理由（422），处理过的再处理 409；清理目录把影响面里的项从目录里去掉、记目录版本；加入码值走
+  apply_patch（含义待填写、依据写变化编号）、记目录版本；知道了只改状态；剖析没开时「重新剖析」照实 422；
+- 同一个变化再次测到只更新 last_seen_at、seen_count；
+- list_changes 按 seq 游标拉、affecting 按表反查、triggers_recall 只认结构类和触及规则的新码值；
+- source.changed 的推送：订阅还没打开时推不出去；打开后（测试里临时打开）消息里只有源名、条数和链接。
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+import uuid
+
+import httpx
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import select
+
+from app.data import catalog, source_changes as sc
+from app.db.base import SessionLocal
+from app.db.models import DataSource
+from app.db.models_sources import SourceChange
+from app.main import app
+from app.notify import core, service
+from tests.fixtures.sources import drop_source, introspect_now
+
+
+@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"
+    db = sqlite3.connect(path)
+    db.executescript("""
+        CREATE TABLE orders (id INTEGER PRIMARY KEY, status VARCHAR(10), channel VARCHAR(8), amount NUMERIC(10, 2));
+        INSERT INTO orders VALUES (1, 'paid', 'web', 3), (2, 'void', 'shop', 4);
+    """)
+    db.commit()
+    db.close()
+    name = f"chg_{uuid.uuid4().hex[:8]}"
+    r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": str(path),
+                                                    "readonly": True})
+    sid = r.json()["id"]
+    await introspect_now(client, sid)
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, sid, "orders", {
+            "label": catalog.make_item("订单", "human"),
+            "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human"),
+                                   "label": catalog.make_item("状态", "human")},
+                        "channel": {"codes": catalog.make_item({"web": "线上"}, "profile", "proposed")}},
+        }, if_version=0, actor="测试员")
+    db = sqlite3.connect(path)
+    db.execute("ALTER TABLE orders DROP COLUMN status")
+    db.commit()
+    db.close()
+    await introspect_now(client, sid)
+    yield {"id": sid, "name": name}
+    await drop_source(sid)
+
+
+async def _change(sid: str, kind: str) -> SourceChange:
+    async with SessionLocal() as session:
+        return (await session.execute(select(SourceChange).where(
+            SourceChange.source_id == sid, SourceChange.kind == kind))).scalars().first()
+
+
+async def _codes_new(sid: str, *, rules: bool = False) -> SourceChange:
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        change, new = await sc.record(session, row, sc.NewChange(
+            kind=sc.CODES_NEW, table="orders", column="channel", severity="warning",
+            after={"codes": ["app"], "shares": {"app": 0.4}}, basis={"method": "full", "key": "full"},
+            affects={"catalog": [], "templates": [], "rules_touched": rules}, key=("full", ("app",))),
+            sc.Detected(job_kind="profile", job_id="job-x", trigger="schedule"))
+        await session.commit()
+        assert new
+        return change
+
+
+async def test_list_counts_and_filters(client, shop):
+    sid = shop["id"]
+    r = await client.get(f"/api/datasources/{sid}/changes")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    [removed] = [c for c in body["items"] if c["kind"] == "column_removed"]
+    assert removed["schema"] == "agentlab-change/1" and removed["column"] == "status" and removed["status"] == "open"
+    assert removed["severity"] == "action" and "删除了列" in removed["summary"]
+    assert body["counts"]["pending"] == body["total"] and body["counts"]["by_severity"]["action"] >= 1
+    r = await client.get(f"/api/datasources/{sid}/changes", params={"kind": ["column_added"]})
+    assert r.json()["items"] == []
+    assert (await client.get(f"/api/datasources/{sid}/changes", params={"status": "nope"})).status_code == 422
+    one = (await client.get(f"/api/datasources/{sid}/changes/{removed['id']}")).json()["change"]
+    assert one["id"] == removed["id"]
+
+
+async def test_dismiss_needs_a_reason_and_finished_changes_are_final(client, shop):
+    sid = shop["id"]
+    change = await _change(sid, "column_removed")
+    url = f"/api/datasources/{sid}/changes/{change.id}/resolve"
+    r = await client.post(url, json={"action": "dismiss"})
+    assert r.status_code == 422 and "理由" in r.json()["detail"]
+    r = await client.post(url, json={"action": "dismiss", "note": "测试库里故意删的列"}, headers={"X-Actor": "tester"})
+    assert r.status_code == 200, r.text
+    got = r.json()["change"]
+    assert got["status"] == "dismissed" and got["resolution"]["note"] == "测试库里故意删的列"
+    assert got["resolution"]["action"] == "dismiss" and got["resolution"]["by"] == "tester"
+    assert (await client.post(url, json={"action": "acknowledge"})).status_code == 409
+    counts = (await client.get(f"/api/datasources/{sid}/changes")).json()["counts"]
+    assert counts["by_severity"]["action"] == 0, "误报不再算要处理"
+
+
+async def test_clean_catalog_removes_the_items_and_records_the_catalog_version(client, shop):
+    sid = shop["id"]
+    change = await _change(sid, "column_removed")
+    r = await client.post(f"/api/datasources/{sid}/changes/{change.id}/resolve", json={"action": "clean_catalog"})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    async with SessionLocal() as session:
+        entry = await catalog.read_entry(session, sid, "orders")
+    assert "status" not in (entry.notes.get("columns") or {}), "说到被删的列的项都清掉了"
+    assert entry.notes["label"]["value"] == "订单", "别的项不动"
+    assert body["catalog_version"] == entry.version == body["change"]["resolution"]["catalog_version"]
+    assert body["change"]["status"] == "resolved"
+
+
+async def test_add_codes_goes_through_apply_patch(client, shop):
+    sid = shop["id"]
+    change = await _codes_new(sid)
+    r = await client.post(f"/api/datasources/{sid}/changes/{change.id}/resolve", json={"action": "add_codes"})
+    assert r.status_code == 200, r.text
+    async with SessionLocal() as session:
+        entry = await catalog.read_entry(session, sid, "orders")
+    item = entry.notes["columns"]["channel"]["codes"]
+    assert item["value"] == {"web": "线上", "app": ""}, "码值是补充：原来的留着，新码值含义待填写"
+    assert (item["source"], item["status"]) == ("human", "confirmed")
+    assert change.id[:6] in item["reason"], "确认时的依据写这条变化的编号"
+    assert r.json()["catalog_version"] == entry.version
+    removed = await _change(sid, "column_removed")
+    r = await client.post(f"/api/datasources/{sid}/changes/{removed.id}/resolve", json={"action": "add_codes"})
+    assert r.status_code == 422, "只有新码值能加入码值"
+
+
+async def test_acknowledge_and_reprofile_when_profiling_is_off(client, shop):
+    sid = shop["id"]
+    change = await _codes_new(sid)
+    url = f"/api/datasources/{sid}/changes/{change.id}/resolve"
+    r = await client.post(url, json={"action": "reprofile"})
+    assert r.status_code == 422 and "剖析" in r.json()["detail"], "剖析没开：照实说做不了"
+    r = await client.post(url, json={"action": "acknowledge"})
+    assert r.status_code == 200 and r.json()["change"]["status"] == "acknowledged"
+    counts = (await client.get(f"/api/datasources/{sid}/changes")).json()["counts"]
+    assert counts["pending"] > counts["open"], "知道了还没处理的照样算要人看"
+
+
+async def test_same_change_detected_again_only_bumps_seen_count(shop):
+    sid = shop["id"]
+    first = await _codes_new(sid)
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        again, new = await sc.record(session, row, sc.NewChange(
+            kind=sc.CODES_NEW, table="orders", column="channel", severity="action", key=("full", ("app",))),
+            sc.Detected(job_kind="profile"))
+        await session.commit()
+    assert not new and again.id == first.id and again.seen_count == 2 and again.severity == "action"
+
+
+async def test_cursor_pull_affecting_and_recall_triggers(shop):
+    sid, name = shop["id"], shop["name"]
+    codes = await _codes_new(sid, rules=True)
+    async with SessionLocal() as session:
+        everything = await sc.list_changes(session, source_id=sid)
+        seqs = [c["seq"] for c in everything]
+        assert seqs == sorted(seqs) and len(seqs) >= 2
+        rest = await sc.list_changes(session, source_id=sid, since_seq=seqs[0])
+        assert [c["seq"] for c in rest] == seqs[1:], "按 seq 游标：大于 since_seq 的"
+        hit = await sc.affecting(session, name, ["ORDERS"])
+        assert {c["id"] for c in hit} == {c["id"] for c in everything}, "表名不分大小写"
+        assert await sc.affecting(session, name, ["refunds"]) == []
+        assert await sc.affecting(session, name, ["orders"], since=seqs[-1]) == []
+    by = {c["kind"]: c for c in everything}
+    assert sc.triggers_recall(by["column_removed"]), "结构类触发召回"
+    assert sc.triggers_recall(next(c for c in everything if c["id"] == codes.id)), "触及有效记录条件的新码值触发召回"
+    assert not sc.triggers_recall({"kind": "volume_drop", "affects": {}}), "纯取值漂移只算数据质量提醒"
+
+
+async def test_source_changed_message_has_only_name_counts_and_link(shop, monkeypatch):
+    sid, name = shop["id"], shop["name"]
+    assert await service.notify_source_changed(sid) == 0, "订阅还没打开（第 4 波）：推不出去"
+    got: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        got.append(request)
+        return httpx.Response(200, json={"ok": True})
+
+    monkeypatch.setattr(core, "_transport", httpx.MockTransport(handler))
+    monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0, 0))
+    monkeypatch.setitem(service.EVENTS, "source.changed", service.Event("数据源有变化要处理", ("global",)))
+    async with SessionLocal() as session:
+        from app.core.crypto import encrypt
+        from app.db.models import NotifyChannel
+
+        channel = NotifyChannel(name=f"变化-{uuid.uuid4().hex[:6]}", scope="global", format="generic",
+                                url=encrypt("https://hooks.example.com/agentlab/changes-7a1c"),
+                                events=["source.changed"])
+        session.add(channel)
+        await session.commit()
+        channel_id = channel.id
+    try:
+        assert await service.notify_source_changed(sid) == 1
+        [request] = got
+        body = json.loads(request.content)
+        text = body["title"] + body["text"]
+        assert body["event"] == "source.changed" and name in body["title"]
+        assert "要处理" in text and "orders" not in text and "status" not in text and "app" not in text, \
+            "不放表名、列名、取值"
+        assert body["subject"] == {"source_id": sid}
+    finally:
+        async with SessionLocal() as session:
+            from app.db.models import NotifyChannel
+
+            row = await session.get(NotifyChannel, channel_id)
+            await session.delete(row)
+            await session.commit()
+
+
+def test_maintenance_option_is_hidden_from_the_datasource_form():
+    from app.api import datasources
+    from app.data import source_maintenance
+
+    assert source_maintenance.MAINTENANCE_OPTION in datasources._MANAGED_OPTIONS

backend/tests/test_source_maintenance.py

@@ -0,0 +1,210 @@
+"""数据源的定时维护（app/data/source_maintenance.py，next-round 03 B4、总方案 X4）。
+
+定时只有一套：维护就是一条 target_kind=source_maintenance 的 Schedule，认领、补跑、积压只记一条都是 scheduler 那一套。
+这里守：
+
+- 到点按顺序建探查任务 → 剖析任务（范围按选的步骤），任务上记 trigger=schedule 和 fire_at；剖析排在同一次触发的探查后面
+  （同源队串行：探查做完剖析才开始）；触发记录写明建了哪几个任务；
+- 照实记没执行的：剖析没开时剖析、观察两步不执行（探查照做）；数据源停用、不存在、是上传表格时整次不执行；上一次的探查
+  还没做完时这次跳过；
+- 调度循环的认领：错过的只补最近一次、积压只记一条；
+- 维护步骤的接口：不合规 422，合规存进 options、数据源表单不显示它；核对目标（check_target）。
+"""
+from __future__ import annotations
+
+import sqlite3
+import uuid
+from datetime import timedelta
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import delete, select
+
+from app.data import catalog, catalog_profile_jobs, introspect_jobs, source_maintenance as sm
+from app.db.base import SessionLocal, utcnow
+from app.db.models import CatalogProfileJob, DataSource, Schedule, ScheduleFire
+from app.db.models_sources import IntrospectJob
+from app.main import app
+from app.scheduler import loop, service
+from tests.fixtures.sources import drop_source, introspect_now
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def _source(client, path, *, profile: bool = True) -> str:
+    db = sqlite3.connect(path)
+    db.executescript("""
+        CREATE TABLE orders (id INTEGER PRIMARY KEY, status VARCHAR(10), paid_on DATE);
+        INSERT INTO orders VALUES (1, 'paid', '2026-10-01'), (2, 'void', '2026-10-02');
+    """)
+    db.commit()
+    db.close()
+    r = await client.post("/api/datasources", json={"name": f"mnt_{uuid.uuid4().hex[:8]}", "kind": "sqlite",
+                                                    "database": str(path), "readonly": True})
+    sid = r.json()["id"]
+    await introspect_now(client, sid)
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        if profile:
+            row.options = {**(row.options or {}), "catalog_profile": {"enabled": True}}
+        await session.commit()
+        await catalog.write_entry(session, sid, "orders", {
+            "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human")}},
+        }, if_version=0, actor="测试员")
+    return sid
+
+
+async def _schedule(sid: str, **over) -> str:
+    async with SessionLocal() as session:
+        fields = {"name": f"维护-{uuid.uuid4().hex[:6]}", "cron": "0 3 * * *", "timezone": "Asia/Shanghai",
+                  **service.target_columns("source_maintenance", sid), **over}
+        row = Schedule(**fields)
+        session.add(row)
+        await session.commit()
+        return row.id
+
+
+@pytest.fixture
+async def made():
+    ids: list[str] = []
+    yield ids
+    async with SessionLocal() as session:
+        await session.execute(delete(Schedule).where(Schedule.id.in_(ids)))
+        await session.commit()
+
+
+async def _wait_jobs(fire: ScheduleFire) -> None:
+    for job in (fire.inputs or {}).get("jobs") or []:
+        manager = introspect_jobs.manager if job["kind"] == "introspect" else catalog_profile_jobs.manager
+        await manager.wait(job["id"], timeout=60)
+
+
+async def test_fire_builds_introspect_then_profile_in_order(client, tmp_path, made):
+    sid = await _source(client, tmp_path / "a.db")
+    try:
+        schedule_id = await _schedule(sid)
+        made.append(schedule_id)
+        at = utcnow().replace(microsecond=0)
+        fire = await service.fire_once(schedule_id, at, "manual", actor="测试员")
+        assert fire.status == "started" and fire.reason is None, fire.reason
+        jobs = fire.inputs["jobs"]
+        assert [j["kind"] for j in jobs] == ["introspect", "profile"] and jobs[1]["scope"] == "maintenance"
+        assert fire.inputs["steps"] == list(sm.STEPS)
+        await _wait_jobs(fire)
+        async with SessionLocal() as session:
+            intro = await session.get(IntrospectJob, jobs[0]["id"])
+            prof = await session.get(CatalogProfileJob, jobs[1]["id"])
+        assert intro.trigger == "schedule" and intro.scope["fire_at"] == at.isoformat()
+        assert intro.scope["schedule_id"] == schedule_id and intro.created_by == "测试员"
+        assert catalog_profile_jobs.trigger_of(prof) == {"trigger": "schedule", "fire_at": at.isoformat(),
+                                                         "schedule_id": schedule_id}
+        assert prof.started_at >= intro.finished_at, "同源串行：探查做完，剖析才开始"
+        assert prof.status in ("succeeded", "partial"), prof.error
+        # 同一次触发再来一遍：交回已有的那一条，不再建任务
+        again = await service.fire_once(schedule_id, at, "manual", actor="测试员")
+        assert again.id == fire.id
+    finally:
+        await drop_source(sid)
+
+
+async def test_profile_off_records_steps_not_run_and_still_introspects(client, tmp_path, made):
+    sid = await _source(client, tmp_path / "b.db", profile=False)
+    try:
+        schedule_id = await _schedule(sid)
+        made.append(schedule_id)
+        fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual")
+        assert fire.status == "started"
+        assert [j["kind"] for j in fire.inputs["jobs"]] == ["introspect"]
+        assert {s["step"] for s in fire.inputs["skipped"]} == {"profile_needed", "observe"}
+        assert "剖析需要的表没有执行" in fire.reason and "取值观察没有执行" in fire.reason and "数据剖析" in fire.reason
+        await _wait_jobs(fire)
+        # 只选观察一步：剖析没开就整次没执行
+        async with SessionLocal() as session:
+            row = await session.get(DataSource, sid)
+            row.options = sm.with_steps(row.options, ["observe"], actor="测试员")
+            await session.commit()
+        fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=1), "manual")
+        assert fire.status == "failed_to_start" and "取值观察没有执行" in fire.reason
+    finally:
+        await drop_source(sid)
+
+
+async def test_disabled_missing_and_busy_sources(client, tmp_path, made):
+    sid = await _source(client, tmp_path / "c.db")
+    try:
+        schedule_id = await _schedule(sid)
+        made.append(schedule_id)
+        async with SessionLocal() as session:
+            row = await session.get(DataSource, sid)
+            row.enabled = False
+            await session.commit()
+        fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual")
+        assert fire.status == "failed_to_start" and "已停用" in fire.reason
+        async with SessionLocal() as session:
+            row = await session.get(DataSource, sid)
+            row.enabled = True
+            session.add(IntrospectJob(source_id=sid, status="queued", scope={}))
+            await session.commit()
+        fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=2), "manual")
+        assert fire.status == "skipped_overlap" and "正在探查结构" in fire.reason
+        async with SessionLocal() as session:
+            await session.execute(delete(IntrospectJob).where(IntrospectJob.source_id == sid,
+                                                              IntrospectJob.status == "queued"))
+            await session.commit()
+    finally:
+        await drop_source(sid)
+    fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=3), "manual")
+    assert fire.status == "failed_to_start" and fire.reason == sm.SOURCE_GONE
+
+
+async def test_loop_claims_and_only_the_latest_missed_fire_runs(client, tmp_path, made):
+    sid = await _source(client, tmp_path / "d.db")
+    try:
+        async with SessionLocal() as session:
+            row = await session.get(DataSource, sid)
+            row.options = sm.with_steps(row.options, ["introspect"], actor="测试员")
+            await session.commit()
+        now = utcnow().replace(second=0, microsecond=0)
+        # 每 10 分钟一次，停机错过了好几次：只补最近一次，更早的积压只记一条
+        schedule_id = await _schedule(sid, cron="*/10 * * * *", timezone="UTC",
+                                      next_run_at=now - timedelta(minutes=40), misfire_grace_s=3600)
+        made.append(schedule_id)
+        fires = await loop.fire_due(now + timedelta(minutes=1))
+        assert sorted(f.status for f in fires) == ["skipped_misfire", "started"], [(f.kind, f.status) for f in fires]
+        started = next(f for f in fires if f.status == "started")
+        assert [j["kind"] for j in started.inputs["jobs"]] == ["introspect"]
+        await _wait_jobs(started)
+        async with SessionLocal() as session:
+            sched = await session.get(Schedule, schedule_id)
+            count = len(list((await session.execute(select(ScheduleFire).where(
+                ScheduleFire.schedule_id == schedule_id))).scalars()))
+        assert sched.next_run_at > now and count == 2
+    finally:
+        await drop_source(sid)
+
+
+async def test_steps_api_and_check_target(client, tmp_path):
+    sid = await _source(client, tmp_path / "e.db")
+    try:
+        url = f"/api/datasources/{sid}/maintenance"
+        body = (await client.get(url)).json()
+        assert body["steps"] == list(sm.STEPS) and body["profile_enabled"] is True and body["schedules"] == []
+        assert (await client.put(url, json={"steps": ["nope"]})).status_code == 422
+        assert (await client.put(url, json={"steps": []})).status_code == 422
+        r = await client.put(url, json={"steps": ["observe", "introspect"]})
+        assert r.status_code == 200 and r.json()["steps"] == ["introspect", "observe"], "按固定顺序存"
+        [shown] = [s["options"] for s in (await client.get("/api/datasources")).json() if s["id"] == sid]
+        assert sm.MAINTENANCE_OPTION not in shown, "数据源表单不显示、不提交它"
+        async with SessionLocal() as session:
+            assert await sm.check_target(session, sid) is None
+            assert await sm.check_target(session, "missing") == sm.SOURCE_GONE
+            row = await session.get(DataSource, sid)
+            row.origin = "upload"
+            await session.commit()
+            assert await sm.check_target(session, sid) == sm.NOT_MAINTAINED
+    finally:
+        await drop_source(sid)