抽出数据目录后台任务的共用运行时，起草任务改用它：为剖析改成逐表的后台任务做准备

data/catalog_job_runtime.py 从 catalog_jobs.py 抽出进程内任务的执行与停止（JobManager：start / stop(grace) / wait / running / shutdown；JobRun 把进队、等轮到自己、执行、按停止原因收尾、出队写一次）、同一任务的库写入一条一条来并 shield（SerialWriter）、启动时标中断（interrupt_on_startup、close_job）、状态词常量，以及每个源一条先来先到的队（SourceLanes：任务阻塞地占，同步写操作不阻塞地试占）。剖析和起草是同一类问题（关页面就没了、不能停、不留痕），生命周期各写一份迟早一边修了另一边没修；共用的是生命周期代码，表各建各的。

起草的行为不变（test_catalog_draft_jobs.py 一条不改全过），只是：ensure_idle 拆成「同种任务互斥」（ensure_draft_free，建起草任务、重试用，剖析任务不挡起草，起草任务排在它后面）和「同步写操作要求源空闲」（ensure_idle，同步起草、导入用，剖析任务也算）；同步起草、导入目录执行期间占着这个源的队，期间新建的任务排在后面；任务和清单顶上的进度带 waiting_for（起草排在剖析后面时为 profile）；排队中被停止的任务连同在队里等着的后台执行一起停掉、出队。

db/models.py 加 CatalogProfileJob、CatalogProfileTask 两张表（create_all 建，同源排队或进行中的部分唯一索引照起草的写法）；catalog_profile_jobs.py 先放剖析任务的忙碌判定，下一步补全。

backend/app/api/catalog.py

@@ -42,7 +42,8 @@ from app.api.coded import (
     SNAPSHOT_TAMPERED, CodedHTTPException,
 )
 from app.api.runs import actor_of
-from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, introspect, table_versions
+from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, catalog_profile_jobs, introspect
+from app.data import table_versions
 from app.data import catalog_assistant, catalog_review, catalog_transfer
 from app.data.engine import SnapshotTampered
 from app.db.base import get_session
@@ -317,13 +318,14 @@ async def import_catalog(source_id: str, payload: CatalogImportIn, x_actor: str
     409 catalog_draft_busy；文件格式不对 422；没有表结构 409。"""
     row, source = await _import_source(session, source_id)
     try:
-        await catalog_jobs.ensure_idle(session, row)
+        async with catalog_jobs.import_write(session, row):
+            report = await catalog_transfer.run_import(session, row, source, payload.content,
+                                                       include_guidance=payload.include_guidance,
+                                                       actor=actor_of(x_actor))
     except catalog_jobs.DraftBusy as e:
         raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
-    try:
-        report = await catalog_transfer.run_import(session, row, source, payload.content,
-                                                   include_guidance=payload.include_guidance,
-                                                   actor=actor_of(x_actor))
+    except catalog_profile_jobs.ProfileBusy as e:
+        raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e
     except catalog_transfer.TransferInvalid as e:
         raise HTTPException(422, str(e)) from e
     return report.as_dict()
@@ -398,7 +400,7 @@ async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | N
         raise HTTPException(422, str(e)) from e
     row, source = await _resolved(session, source_id)
     try:
-        await catalog_jobs.ensure_idle(session, row)
+        await catalog_jobs.ensure_draft_free(session, row)
     except catalog_jobs.DraftBusy as e:
         raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
     cache = source.schema_cache or {}
@@ -412,7 +414,7 @@ async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | N
     except catalog_jobs.DraftBusy as e:
         raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
     catalog_jobs.manager.start(job.id)
-    return {"job": catalog_jobs.job_dict(job, tasks)}
+    return {"job": catalog_jobs.job_dict(job, tasks, waiting_for=await catalog_jobs.waiting_for(session, job))}
 
 
 @router.get("/{source_id}/catalog/draft-jobs/active")
@@ -466,7 +468,7 @@ async def retry_draft_job(source_id: str, job_id: str, x_actor: str | None = Hea
     _, old = await _job_or_404(session, source_id, job_id)
     row, source = await _resolved(session, source_id)
     try:
-        await catalog_jobs.ensure_idle(session, row)
+        await catalog_jobs.ensure_draft_free(session, row)
     except catalog_jobs.DraftBusy as e:
         raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
     try:
@@ -484,7 +486,7 @@ async def retry_draft_job(source_id: str, job_id: str, x_actor: str | None = Hea
     except catalog_jobs.DraftBusy as e:
         raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
     catalog_jobs.manager.start(job.id)
-    return {"job": catalog_jobs.job_dict(job, tasks)}
+    return {"job": catalog_jobs.job_dict(job, tasks, waiting_for=await catalog_jobs.waiting_for(session, job))}
 
 
 @router.put("/{source_id}/catalog/{table}")
@@ -529,6 +531,8 @@ async def draft_catalog(source_id: str, payload: CatalogDraftIn,
                                                  guidance=catalog_jobs.guidance_of(row))
     except catalog_jobs.DraftBusy as e:
         raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
+    except catalog_profile_jobs.ProfileBusy as e:
+        raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e
     rows = [{"table_name": r.table, "added": r.added, "updated": r.updated, "removed": r.removed,
              "version": r.version, "error": r.error, "model_error": r.model_error, "model_items": r.model_items}
             for r in report.tables]

backend/app/data/catalog_job_runtime.py

@@ -0,0 +1,400 @@
+"""数据目录后台任务的共用运行时：起草（data/catalog_jobs.py）和剖析（data/catalog_profile_jobs.py）各自建表、各自的步骤，
+生命周期是同一套。
+
+**为什么抽出来。** 剖析以前是一次同步长请求，和起草改成后台任务之前是同一类问题：关掉页面结果就没了、不能停、不留痕。
+两边要的是同一样东西——进程内跑、能停、重启标中断、写库不被停止打断、同源排队——各写一份，迟早一边修了另一边没修。
+共用的是生命周期代码，不是表：两种任务的进度完全不同（起草按段、块、模型用量，剖析按检查项、查询），合成一张
+「kind + JSON」的通用任务表要丢掉起草那些强类型列，还要迁移刚稳定的起草数据。
+
+- **状态词**（ACTIVE、FINISHED、TASK_STATUSES、TASK_DONE、RETRYABLE、OPEN）：任务 queued → running → succeeded / partial
+  / failed / cancelled / interrupted；逐表 pending → running → ok / partial / failed / skipped / cancelled。两种任务、
+  接口、界面都用这一套词。
+- **SerialWriter**：同一个任务的库写入一条一条来，shield 起来——停止任务时不会打断写到一半的库操作；收尾前 settle 等它们
+  写完。
+- **JobRun / JobManager**：进程内一个 asyncio 任务一件事（照 app/scheduler 的写法）：start / stop(grace) / wait / running
+  / shutdown。JobRun.run 把「进队、等轮到自己、执行、按停止的原因收尾、出队」写一次，各种任务只写执行和收尾的那几步。
+- **SourceLanes**：每个数据源一条先来先到的队。起草、剖析都打业务库或模型、都写同一个源的目录：同源串行（先建的先跑），
+  不同源并行。任务阻塞地占：建出来是 queued，轮到了才 running。同步起草、导入目录这类同步写操作不阻塞地试占：占不到
+  说明源上正有别的事，接口回 409；占到了，执行期间新建的任务排在它后面。
+- **close_job / interrupt_on_startup**：收尾（还没结束的表记成某个状态，任务记成某个状态），和服务启动时的整理：排队、
+  进行中的标 interrupted，没结束的表记 cancelled。**不自动续跑**：复制出来的数据目录（检查脚本、验收副本）启动时不能
+  自己去调模型、查业务库，界面让人点「重试」接着做。
+
+进程内登记、进程内的队：服务端是单进程部署，多进程部署时要换成库里的锁。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from typing import Any
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.base import SessionLocal, utcnow
+
+logger = logging.getLogger(__name__)
+
+#: 任务还没结束的状态（同一个源同一种任务只能有一个）、结束了的状态
+ACTIVE = ("queued", "running")
+FINISHED = ("succeeded", "partial", "failed", "cancelled", "interrupted")
+#: 逐表的状态，其中算「已结束」的、重试时要带上的、还没结束的
+TASK_STATUSES = ("pending", "running", "ok", "partial", "failed", "skipped", "cancelled")
+TASK_DONE = ("ok", "partial", "failed", "skipped", "cancelled")
+RETRYABLE = ("failed", "partial", "cancelled")
+OPEN = ("pending", "running")
+
+#: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace）
+STOP_GRACE_S = 10.0
+SHUTDOWN_GRACE_S = 3.0
+
+
+class JobFailed(Exception):
+    """任务整体出错（数据源没了、快照没了、模型用不了……）：message 就是任务的 error。"""
+
+
+# ==========================================================================
+# 库写入
+# ==========================================================================
+
+
+class SerialWriter:
+    """一个任务的库写入：一条一条来，每条开一个会话；shield 起来，停止任务时不会打断写到一半的库操作。"""
+
+    def __init__(self) -> None:
+        self._writes: set[asyncio.Task[Any]] = set()
+        self._lock = asyncio.Lock()
+
+    async def __call__(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any:
+        async def go() -> Any:
+            async with self._lock:
+                async with SessionLocal() as session:
+                    return await fn(session, *args)
+
+        task = asyncio.ensure_future(go())
+        self._writes.add(task)
+        task.add_done_callback(self._writes.discard)
+        return await asyncio.shield(task)
+
+    async def settle(self) -> None:
+        """等手上的库写入都写完（停止、出错收尾之前）。"""
+        while self._writes:
+            await asyncio.gather(*list(self._writes), return_exceptions=True)
+
+
+# ==========================================================================
+# 每个源一条队
+# ==========================================================================
+
+
+class LaneTicket:
+    """在一个源的队里的一个位置。kind 是占着它的是什么：draft 起草任务、profile 剖析任务、sync_draft 同步起草、
+    import 导入目录；job_id 是任务的（同步写操作没有）。轮到它时 granted 置位。"""
+
+    def __init__(self, lanes: SourceLanes, source_id: str, kind: str, job_id: str | None,
+                 loop: asyncio.AbstractEventLoop) -> None:
+        self._lanes = lanes
+        self.source_id = source_id
+        self.kind = kind
+        self.job_id = job_id
+        self.loop = loop
+        self.granted = asyncio.Event()
+        self.released = False
+
+    async def wait(self) -> None:
+        """等轮到自己。"""
+        await self.granted.wait()
+
+    def release(self) -> None:
+        """出队（轮到了、没轮到都行，重复调用无害）：排在后面的第一个接着开始。"""
+        self._lanes._release(self)
+
+
+class LaneBusy(Exception):
+    """不阻塞地试占时这个源的队里已经有人：holder 是排在最前面（正在做）的那一个。"""
+
+    def __init__(self, holder: LaneTicket) -> None:
+        super().__init__(holder.kind)
+        self.holder = holder
+
+
+class SourceLanes:
+    """每个数据源一条先来先到的队。"""
+
+    def __init__(self) -> None:
+        self._queues: dict[str, list[LaneTicket]] = {}
+
+    def _queue(self, source_id: str) -> list[LaneTicket]:
+        """这个源的队。别的事件循环留下的（测试每个用例一个循环）去掉：那个循环里没人会再出队了。"""
+        queue = self._queues.get(source_id) or []
+        loop = asyncio.get_running_loop()
+        alive = [t for t in queue if t.loop is loop and not t.released]
+        if alive:
+            self._queues[source_id] = alive
+            alive[0].granted.set()
+        else:
+            self._queues.pop(source_id, None)
+        return alive
+
+    def enter(self, source_id: str, kind: str, job_id: str | None = None) -> LaneTicket:
+        """排进队尾，马上返回（不等）。排在最前面的立刻 granted；调用方 await ticket.wait() 等轮到自己。"""
+        queue = self._queue(source_id)
+        ticket = LaneTicket(self, source_id, kind, job_id, asyncio.get_running_loop())
+        queue.append(ticket)
+        self._queues[source_id] = queue
+        if len(queue) == 1:
+            ticket.granted.set()
+        return ticket
+
+    def try_enter(self, source_id: str, kind: str) -> LaneTicket:
+        """队空着才占（马上就轮到）；有人在队里抛 LaneBusy。"""
+        queue = self._queue(source_id)
+        if queue:
+            raise LaneBusy(queue[0])
+        return self.enter(source_id, kind)
+
+    def holders(self, source_id: str) -> list[LaneTicket]:
+        """队里的各位，排在最前面（正在做）的在前。"""
+        return list(self._queue(source_id))
+
+    def ahead(self, source_id: str, job_id: str) -> list[str] | None:
+        """排在这个任务前面的各位是什么（kind，正在做的在前）；这个任务不在队里（刚建、后台还没排进来）返回 None。"""
+        queue = self._queue(source_id)
+        for i, ticket in enumerate(queue):
+            if ticket.job_id == job_id:
+                return [t.kind for t in queue[:i]]
+        return None
+
+    def _release(self, ticket: LaneTicket) -> None:
+        if ticket.released:
+            return
+        ticket.released = True
+        queue = self._queues.get(ticket.source_id)
+        if not queue or ticket not in queue:
+            return
+        head = queue[0] is ticket
+        queue.remove(ticket)
+        if not queue:
+            self._queues.pop(ticket.source_id, None)
+        elif head:
+            queue[0].granted.set()
+
+    @contextlib.asynccontextmanager
+    async def hold(self, source_id: str, kind: str, job_id: str | None = None) -> AsyncIterator[LaneTicket]:
+        """阻塞地占：排队，轮到了再进去，出来时出队。"""
+        ticket = self.enter(source_id, kind, job_id)
+        try:
+            await ticket.wait()
+            yield ticket
+        finally:
+            ticket.release()
+
+    @contextlib.contextmanager
+    def try_hold(self, source_id: str, kind: str) -> Iterator[LaneTicket]:
+        """不阻塞地试占（同步起草、导入目录）：占不到抛 LaneBusy；占到了，执行期间新建的任务排在后面。"""
+        ticket = self.try_enter(source_id, kind)
+        try:
+            yield ticket
+        finally:
+            ticket.release()
+
+
+lanes = SourceLanes()
+
+
+# ==========================================================================
+# 收尾、启动时的整理
+# ==========================================================================
+
+
+async def close_job(session: AsyncSession, job_model: Any, task_model: Any, job_id: str, job_status: str,
+                    job_error: str | None, task_status: str, task_error: str | None) -> None:
+    """收尾：还没结束的表记 task_status（原因 task_error），任务（还没结束的话）记 job_status。提交。"""
+    now = utcnow()
+    await session.execute(
+        update(task_model).where(task_model.job_id == job_id, task_model.status.in_(OPEN))
+        .values(status=task_status, error=task_error, finished_at=now).execution_options(synchronize_session=False))
+    await session.execute(
+        update(job_model).where(job_model.id == job_id, job_model.status.in_(ACTIVE))
+        .values(status=job_status, error=job_error, finished_at=now).execution_options(synchronize_session=False))
+    await session.commit()
+
+
+async def cancel_queued(session: AsyncSession, job_model: Any, task_model: Any, job_id: str) -> bool:
+    """停止一个排队中的任务：任务和没结束的表直接记 cancelled。已经被后台认领（开始了）返回 False，交给调用方按进行中
+    的停。"""
+    now = utcnow()
+    claimed = await session.execute(
+        update(job_model).where(job_model.id == job_id, job_model.status == "queued")
+        .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False))
+    if claimed.rowcount != 1:
+        await session.rollback()
+        return False
+    await session.execute(
+        update(task_model).where(task_model.job_id == job_id, task_model.status.in_(OPEN))
+        .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False))
+    await session.commit()
+    return True
+
+
+async def claim(session: AsyncSession, job_model: Any, job_id: str) -> bool:
+    """认领：排队中 → 进行中。开始之前就被停止了返回 False。"""
+    result = await session.execute(
+        update(job_model).where(job_model.id == job_id, job_model.status == "queued")
+        .values(status="running", started_at=utcnow()).execution_options(synchronize_session=False))
+    await session.commit()
+    return result.rowcount == 1
+
+
+async def interrupt_on_startup(job_model: Any, close: Callable[[AsyncSession, str], Awaitable[None]], *,
+                               running: Callable[[str], bool]) -> int:
+    """服务启动时：排队、进行中、而且不在这个进程里跑的任务逐个 close（各自记成 interrupted、没结束的表记 cancelled）。
+    不自动续跑。返回整理了几个。"""
+    async with SessionLocal() as session:
+        ids = [job_id for job_id in (await session.execute(
+            select(job_model.id).where(job_model.status.in_(ACTIVE)))).scalars() if not running(job_id)]
+        for job_id in ids:
+            await close(session, job_id)
+    return len(ids)
+
+
+# ==========================================================================
+# 在后台执行
+# ==========================================================================
+
+
+class JobRun:
+    """一个任务在后台的一次执行。子类给 kind（进队时的身份）、model（任务表，进队前读源和状态），写 _run（执行）和
+    _stopped / _failed / _crashed（停止、整体出错、意外出错时的收尾）。"""
+
+    kind = ""
+    model: Any = None
+
+    def __init__(self, job_id: str) -> None:
+        self.job_id = job_id
+        self.task: asyncio.Task[None] | None = None
+        #: 为什么停：user 使用者点了停止，shutdown 服务关停（标中断）
+        self.stop_reason: str | None = None
+        self.writer = SerialWriter()
+        self.ticket: LaneTicket | None = None
+
+    async def _write(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any:
+        """一条库写入（SerialWriter：一条一条来，停止时不被打断）。"""
+        return await self.writer(fn, *args)
+
+    async def _settle(self) -> None:
+        await self.writer.settle()
+
+    async def run(self) -> None:
+        try:
+            try:
+                await self._queue()
+                await self._run()
+            except asyncio.CancelledError:
+                current = asyncio.current_task()
+                if current is not None:
+                    current.uncancel()
+                await self._stopped()
+            except JobFailed as e:
+                await self._failed(str(e))
+            except Exception as e:  # noqa: BLE001 - 任务整体出错：照实记下，不留在「进行中」
+                await self._crashed(e)
+        finally:
+            # 收尾写完才出队：排在后面的任务开始时，这个任务已经记成结束了
+            if self.ticket is not None:
+                self.ticket.release()
+
+    async def _queue(self) -> None:
+        """进这个源的队，等轮到自己。任务不在了、已经不是排队中（开始之前就被停止了）的不进队：_run 认领不到，直接结束。"""
+        async with SessionLocal() as session:
+            job = await session.get(self.model, self.job_id)
+            source_id = job.source_id if job is not None and job.status == "queued" else None
+        if source_id is None:
+            return
+        self.ticket = lanes.enter(source_id, self.kind, self.job_id)
+        await self.ticket.wait()
+
+    async def _run(self) -> None:
+        raise NotImplementedError
+
+    async def _stopped(self) -> None:
+        raise NotImplementedError
+
+    async def _failed(self, reason: str) -> None:
+        raise NotImplementedError
+
+    async def _crashed(self, e: Exception) -> None:
+        raise NotImplementedError
+
+
+class JobManager:
+    """进程内的一种后台任务：一个 asyncio 任务一件事，停的时候给宽限。factory 按任务编号造一次执行（JobRun）；label 是
+    日志里的叫法（「数据目录起草任务」），name 是 asyncio 任务名的前缀。"""
+
+    def __init__(self, factory: Callable[[str], JobRun], *, label: str, name: str) -> None:
+        self._runs: dict[str, JobRun] = {}
+        self._factory = factory
+        self.label = label
+        self.name = name
+
+    def _alive(self, job_id: str) -> JobRun | None:
+        run = self._runs.get(job_id)
+        if run is None or run.task is None or run.task.done():
+            return None
+        try:
+            if run.task.get_loop() is not asyncio.get_running_loop():
+                self._runs.pop(job_id, None)   # 别的事件循环留下的（测试每个用例一个循环）
+                return None
+        except RuntimeError:
+            return None
+        return run
+
+    def start(self, job_id: str) -> None:
+        """开始在后台执行这个（排队中的）任务：先进这个源的队，轮到了才开始。"""
+        run = self._factory(job_id)
+        run.task = asyncio.get_running_loop().create_task(run.run(), name=f"{self.name}-{job_id}")
+        self._runs[job_id] = run
+
+        def forget(_: asyncio.Task[None]) -> None:
+            if self._runs.get(job_id) is run:
+                self._runs.pop(job_id, None)
+
+        run.task.add_done_callback(forget)
+
+    def running(self, job_id: str) -> bool:
+        return self._alive(job_id) is not None
+
+    async def stop(self, job_id: str, reason: str, *, grace: float = STOP_GRACE_S) -> bool:
+        """停掉这个进程里在跑（或在排队等）的任务，等它收尾（最多 grace 秒）。不在这个进程里返回 False。"""
+        run = self._alive(job_id)
+        if run is None:
+            return False
+        if run.stop_reason is None:
+            run.stop_reason = reason
+            run.task.cancel()  # type: ignore[union-attr]
+        await asyncio.wait([run.task], timeout=grace)  # type: ignore[list-item]
+        return True
+
+    async def wait(self, job_id: str, timeout: float | None = None) -> None:
+        """等这个任务做完（测试、调试用）。不在跑直接返回。"""
+        run = self._alive(job_id)
+        if run is not None:
+            await asyncio.wait([run.task], timeout=timeout)  # type: ignore[list-item]
+
+    async def shutdown(self, *, grace: float = SHUTDOWN_GRACE_S) -> None:
+        """关停：停掉进行中、排队等着的任务（标中断），最多等 grace 秒收尾。"""
+        runs = [run for job_id in list(self._runs) if (run := self._alive(job_id)) is not None]
+        for run in runs:
+            if run.stop_reason is None:
+                run.stop_reason = "shutdown"
+                run.task.cancel()  # type: ignore[union-attr]
+        if runs:
+            await asyncio.wait([run.task for run in runs], timeout=grace)  # type: ignore[misc]
+            logger.info("已停止 %d 个%s", len(runs), self.label)
+
+
+__all__ = ["ACTIVE", "FINISHED", "JobFailed", "JobManager", "JobRun", "LaneBusy", "LaneTicket", "OPEN", "RETRYABLE",
+           "SHUTDOWN_GRACE_S", "STOP_GRACE_S", "SerialWriter", "SourceLanes", "TASK_DONE", "TASK_STATUSES", "cancel_queued",
+           "claim", "close_job", "interrupt_on_startup", "lanes"]

backend/app/data/catalog_jobs.py

@@ -13,15 +13,20 @@
   接口读的是库里的状态。
 - **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没开始和没写完的表记 cancelled，已经写进
   目录的（含宽表已经写出来的段）不回滚，任务最后是 cancelled。写过段的表 model_error 说清写进了几段、剩几段没起草。
-- **和同步起草互斥**（sync_draft）：同步起草（POST /catalog/draft）进行期间，同一个源建任务、重试、导入（ensure_idle）
-  都是 DraftBusy；有排队或进行中的任务时同步起草也是。进程内登记（单进程部署）。
+- **同种任务互斥**（ensure_draft_free）：同一个源同时只有一个排队或进行中的起草任务，同步起草（POST /catalog/draft）
+  进行期间也算在起草，建任务、重试都是 DraftBusy。剖析任务不挡起草：同源有剖析任务时起草任务照建，排在它后面
+  （waiting_for = "profile"），剖析结束后自动开始（每个源一条队，见 data/catalog_job_runtime.py）。
+- **同步写操作要求源空闲**（ensure_idle、sync_write）：同步起草、导入目录要的是这个源上没有别的事——有排队或进行中的
+  起草任务（DraftBusy）、剖析任务（catalog_profile_jobs.ProfileBusy）都不开始；开始了就占着这个源的队，执行期间
+  新建的任务排在它后面。进程内登记（单进程部署）。
 - **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表；用不用模型、本次说明沿用，
   起草要求取数据源当前的。
 - **服务重启**：启动时（startup）把排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明「服务重启，起草
   中断」。不自动续跑：复制出来的数据目录（检查、验收）启动时不能自己去调模型；界面让人点「重试」续上。关停时
   （manager.shutdown）停掉进行中的任务，同样标成 interrupted。
 
-任务的库写入一律经 _JobRun._write：一条一条来，shield 起来——停止任务时不会打断写到一半的库操作，收尾前等它们写完。
+任务的生命周期（进程内执行、停止、同源排队、库写入一条一条来并 shield、启动时标中断、状态词）在
+data/catalog_job_runtime.py，剖析任务用的是同一套；这里只写起草自己的步骤。
 
 **起草要求**存在数据源 options 的 catalog_draft_guidance 里（上传表格的源存在它自己的 DataSource 行上），一段文字，
 去首尾空白、最多 GUIDANCE_MAX 字，空串等于没有。只经 PUT /catalog/guidance 改：数据源接口不显示它、整份回写 options
@@ -40,20 +45,15 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.data import catalog, catalog_draft
+from app.data import catalog_job_runtime as runtime
+from app.data.catalog_job_runtime import ACTIVE, FINISHED, RETRYABLE, STOP_GRACE_S, TASK_DONE, TASK_STATUSES
 from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION
 from app.db.base import SessionLocal, new_id, utcnow
-from app.db.models import CatalogDraftJob, CatalogDraftTask, DataSource
+from app.db.models import CatalogDraftJob, CatalogDraftTask, CatalogProfileJob, DataSource
 
 logger = logging.getLogger(__name__)
 
-#: 任务还没结束的状态（同一个源只能有一个）、结束了的状态
-ACTIVE = ("queued", "running")
-FINISHED = ("succeeded", "partial", "failed", "cancelled", "interrupted")
-#: 逐表的状态，以及其中算「已结束」的、重试时要带上的
-TASK_STATUSES = ("pending", "running", "ok", "partial", "failed", "skipped", "cancelled")
-TASK_DONE = ("ok", "partial", "failed", "skipped", "cancelled")
-RETRYABLE = ("failed", "partial", "cancelled")
-_OPEN = ("pending", "running")
+_OPEN = runtime.OPEN
 
 #: 一个任务最多几张表；起草要求、本次说明最多几个字
 MAX_TABLES = 500
@@ -63,9 +63,6 @@ NOTE_MAX = 1000
 LAST_DRAFT_JOBS = 100
 #: 逐表最多记几个模型留空的列名
 BLANK_SAMPLE = 5
-#: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace）
-STOP_GRACE_S = 10.0
-
 INTERRUPTED_ERROR = "服务重启，起草中断"
 #: 停止、中断时写过段的表（宽表每写出一段就并入目录）：写进了几段、剩几段没起草
 STOPPED_WRITTEN = "已停止：已写进目录 {n} 段，剩下 {m} 段没起草"
@@ -93,7 +90,7 @@ class TextTooLong(ValueError):
     """起草要求、本次说明超长。message 给人看。"""
 
 
-class _JobFailed(Exception):
+class _JobFailed(runtime.JobFailed):
     """任务整体出错（数据源没了、快照没了、模型用不了）：message 就是任务的 error。"""
 
 
@@ -200,13 +197,15 @@ def _counts(statuses: list[str]) -> dict[str, int]:
     return counts
 
 
-def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, Any]:
-    """接口返回的任务（CatalogDraftJob）：逐表的状态按起草顺序；计数、合计都按逐表的算。"""
+def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask], *, waiting_for: str | None = None) -> dict[str, Any]:
+    """接口返回的任务（CatalogDraftJob）：逐表的状态按起草顺序；计数、合计都按逐表的算。waiting_for：排队中、排在同源的
+    剖析任务后面时是 "profile"（见 waiting_for），否则 null。"""
     counts = _counts([t.status for t in tasks])
     return {
         "id": job.id,
         "source_id": job.source_id,
         "status": job.status,
+        "waiting_for": waiting_for,
         "use_model": bool(job.use_model),
         "model": job.model,
         "guidance": job.guidance,
@@ -235,12 +234,32 @@ def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, A
     }
 
 
+async def waiting_for(session: AsyncSession, job: CatalogDraftJob) -> str | None:
+    """排队中的起草任务在等什么：排在同源的剖析任务后面时是 "profile"，否则 None（进行中、结束了的都是 None）。
+
+    先看这个源的队（进程内，排在它前面的是谁）；任务刚建、后台还没排进队时看库：同源有比它先建的、排队或进行中的剖析
+    任务。"""
+    if job.status != "queued":
+        return None
+    ahead = runtime.lanes.ahead(job.source_id, job.id)
+    if ahead is not None:
+        return "profile" if "profile" in ahead else None
+    earlier = (await session.execute(
+        select(CatalogProfileJob.id).where(CatalogProfileJob.source_id == job.source_id,
+                                           CatalogProfileJob.status.in_(ACTIVE),
+                                           CatalogProfileJob.created_at <= job.created_at).limit(1)
+    )).first()
+    return "profile" if earlier is not None else None
+
+
 async def job_out(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None:
-    return None if job is None else job_dict(job, await tasks_of(session, job.id))
+    if job is None:
+        return None
+    return job_dict(job, await tasks_of(session, job.id), waiting_for=await waiting_for(session, job))
 
 
 async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None:
-    """目录清单顶上的进度：{id, status, total, done}。"""
+    """目录清单顶上的进度：{id, status, total, done, waiting_for}。"""
     if job is None:
         return None
     rows = (await session.execute(
@@ -249,7 +268,7 @@ async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[
     )).all()
     counts = {status: n for status, n in rows}
     return {"id": job.id, "status": job.status, "total": sum(counts.values()),
-            "done": sum(counts.get(s, 0) for s in TASK_DONE)}
+            "done": sum(counts.get(s, 0) for s in TASK_DONE), "waiting_for": await waiting_for(session, job)}
 
 
 async def last_drafts(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]:
@@ -302,6 +321,10 @@ def sync_busy_message(row: Any) -> str:
     return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录（同步起草），请等这次起草结束再开始新的起草"
 
 
+def import_busy_message(row: Any) -> str:
+    return f"数据源「{getattr(row, 'name', '')}」正在导入数据目录，请等导入完成后再试"
+
+
 #: 正在同步起草（POST /catalog/draft）的数据源。进程内登记：服务端是单进程部署，多进程部署时要换成库里的锁
 _SYNC: set[str] = set()
 
@@ -311,30 +334,82 @@ def sync_drafting(source_id: str) -> bool:
     return source_id in _SYNC
 
 
-async def ensure_idle(session: AsyncSession, row: Any) -> None:
-    """这个源没有在起草：没有排队或进行中的任务，也没有在同步起草。有就抛 DraftBusy。建任务、重试、导入目录之前都调它。"""
+async def ensure_draft_free(session: AsyncSession, row: Any) -> None:
+    """同种任务互斥：这个源没有在起草——没有排队或进行中的起草任务，也没有在同步起草。有就抛 DraftBusy。建起草任务、
+    重试之前调它。剖析任务不算：同源有剖析任务时起草任务照建，排在它后面。"""
     if row.id in _SYNC:
         raise DraftBusy(sync_busy_message(row))
     if await active_job(session, row.id) is not None:
         raise DraftBusy(busy_message(row))
 
 
+async def ensure_idle(session: AsyncSession, row: Any) -> None:
+    """同步写操作（导入目录、同步起草）要求源空闲：没有在起草（DraftBusy，同 ensure_draft_free），也没有排队或进行中的
+    剖析任务（catalog_profile_jobs.ProfileBusy，说法里写明进行到第几张）。"""
+    from app.data import catalog_profile_jobs
+
+    await ensure_draft_free(session, row)
+    await catalog_profile_jobs.ensure_free(session, row)
+
+
+def _lane_busy(row: Any, holder: runtime.LaneTicket) -> Exception:
+    """库里查过没有任务、这个源的队却被占着（同步起草、导入目录正在做，或者一个任务刚好在收尾）：按占着的是谁说。"""
+    if holder.kind == "profile":
+        from app.data import catalog_profile_jobs
+
+        return catalog_profile_jobs.ProfileBusy(catalog_profile_jobs.busy_message(row, None))
+    if holder.kind == "import":
+        return DraftBusy(import_busy_message(row))
+    return DraftBusy(sync_busy_message(row) if holder.kind == "sync_draft" else busy_message(row))
+
+
+@contextlib.asynccontextmanager
+async def _sync_write(row: Any, kind: str, check: Callable[[], Awaitable[None]]) -> AsyncIterator[None]:
+    """同步写操作期间占着这个源的队（不阻塞地试占）：先查库（check：源上有排队或进行中的任务抛 DraftBusy / ProfileBusy），
+    再占队——别的同步写操作正在做时占不到，同样抛忙，什么都不做。占到了，执行期间新建的任务排在它后面，做完才开始。"""
+    await check()
+    try:
+        ticket = runtime.lanes.try_enter(row.id, kind)
+    except runtime.LaneBusy as e:
+        raise _lane_busy(row, e.holder) from e
+    try:
+        yield
+    finally:
+        ticket.release()
+
+
+async def _ensure_idle_but_sync(session: AsyncSession, row: Any) -> None:
+    """同 ensure_idle，只是不看 _SYNC：同步起草自己刚登记在里面。"""
+    from app.data import catalog_profile_jobs
+
+    if await active_job(session, row.id) is not None:
+        raise DraftBusy(busy_message(row))
+    await catalog_profile_jobs.ensure_free(session, row)
+
+
 @contextlib.asynccontextmanager
 async def sync_draft(session: AsyncSession, row: Any) -> AsyncIterator[None]:
-    """同步起草期间登记这个源：同一个源建任务、重试、导入、再开一次同步起草都抛 DraftBusy；有排队或进行中的任务时
-    这里就抛 DraftBusy。先登记、再查库：查库要让出事件循环，先登记才不会和同时建的任务互相看不见（建任务那边写库之后
-    也再看一次，见 create_job）。"""
+    """同步起草期间登记这个源、占着它的队：同一个源建起草任务、重试、导入、再开一次同步起草都抛 DraftBusy，期间新建的
+    剖析任务排在它后面；有排队或进行中的起草任务时这里抛 DraftBusy，有剖析任务时抛 ProfileBusy。先登记、再查库：查库要
+    让出事件循环，先登记才不会和同时建的任务互相看不见（建任务那边写库之后也再看一次，见 create_job）。"""
     if row.id in _SYNC:
         raise DraftBusy(sync_busy_message(row))
     _SYNC.add(row.id)
     try:
-        if await active_job(session, row.id) is not None:
-            raise DraftBusy(busy_message(row))
-        yield
+        async with _sync_write(row, "sync_draft", lambda: _ensure_idle_but_sync(session, row)):
+            yield
     finally:
         _SYNC.discard(row.id)
 
 
+@contextlib.asynccontextmanager
+async def import_write(session: AsyncSession, row: Any) -> AsyncIterator[None]:
+    """导入目录期间占着这个源的队：源上有排队或进行中的起草、剖析任务（或者正在同步起草、导入）时抛 DraftBusy /
+    ProfileBusy；导入期间新建的任务排在它后面。"""
+    async with _sync_write(row, "import", lambda: ensure_idle(session, row)):
+        yield
+
+
 async def check_model(session: AsyncSession) -> None:
     """起草模型能不能用：不能用抛 catalog.CatalogModelUnavailable（只建模型对象，不发请求）。"""
     await catalog.resolve_draft_model(session)
@@ -344,8 +419,9 @@ async def create_job(session: AsyncSession, row: Any, source: Any, *, tables: li
                      note: str | None, actor: str | None,
                      retry_of: str | None = None) -> tuple[CatalogDraftJob, list[CatalogDraftTask]]:
     """建一个排队的任务（还没开始，调用方接着 manager.start）。row 是数据源记录（起草要求取它的），source 是读表结构
-    用的源（上传源是绑定当前快照的视图）。有排队或进行中的任务抛 DraftBusy。"""
-    await ensure_idle(session, row)
+    用的源（上传源是绑定当前快照的视图）。有排队或进行中的起草任务（或正在同步起草）抛 DraftBusy；同源有剖析任务时照建，
+    排在它后面。"""
+    await ensure_draft_free(session, row)
     cache = getattr(source, "schema_cache", None) or {}
     all_tables = cache.get("tables") or {}
     wanted = list(tables) if tables else await catalog.default_tables(session, source)
@@ -395,22 +471,15 @@ async def retry_tables(session: AsyncSession, job: CatalogDraftJob) -> list[str]
 
 
 async def cancel(session: AsyncSession, job: CatalogDraftJob) -> None:
-    """停止一个任务。排队的直接 cancelled；进行中的交给 manager 停（停掉模型调用、没写完的表记 cancelled）。
-    已经结束的抛 DraftFinished。"""
+    """停止一个任务。排队的直接 cancelled（在队里等着的后台执行跟着停掉、出队）；进行中的交给 manager 停（停掉模型调用、
+    没写完的表记 cancelled）。已经结束的抛 DraftFinished。"""
     if job.status in FINISHED:
         raise DraftFinished("这次起草已经结束了，不需要停止")
     if job.status == "queued":
-        now = utcnow()
-        claimed = await session.execute(
-            update(CatalogDraftJob).where(CatalogDraftJob.id == job.id, CatalogDraftJob.status == "queued")
-            .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False))
-        if claimed.rowcount == 1:
-            await session.execute(
-                update(CatalogDraftTask).where(CatalogDraftTask.job_id == job.id, CatalogDraftTask.status.in_(_OPEN))
-                .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False))
-            await session.commit()
+        if await runtime.cancel_queued(session, CatalogDraftJob, CatalogDraftTask, job.id):
+            await manager.stop(job.id, "user", grace=STOP_GRACE_S)
             return
-        await session.rollback()         # 刚被后台认领、开始了：按进行中的停
+        # 刚被后台认领、开始了：按进行中的停
     if await manager.stop(job.id, "user", grace=STOP_GRACE_S):
         return
     # 库里说在进行，这个进程里却没有在跑它（记录对不上）：直接记停止
@@ -434,26 +503,19 @@ async def _close(session: AsyncSession, job_id: str, job_status: str, job_error:
                 .values(status=task_status, error=task_error, finished_at=now,
                         model_error=written.format(n=n, m=max(0, int(total or n) - int(n))))
                 .execution_options(synchronize_session=False))
-    await session.execute(
-        update(CatalogDraftTask).where(CatalogDraftTask.job_id == job_id, CatalogDraftTask.status.in_(_OPEN))
-        .values(status=task_status, error=task_error, finished_at=now).execution_options(synchronize_session=False))
-    await session.execute(
-        update(CatalogDraftJob).where(CatalogDraftJob.id == job_id, CatalogDraftJob.status.in_(ACTIVE))
-        .values(status=job_status, error=job_error, finished_at=now).execution_options(synchronize_session=False))
-    await session.commit()
+    await runtime.close_job(session, CatalogDraftJob, CatalogDraftTask, job_id, job_status, job_error, task_status,
+                            task_error)
 
 
 async def startup() -> int:
     """服务启动时：排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明服务重启。不自动续跑。返回标了几个。"""
-    async with SessionLocal() as session:
-        ids = [job_id for job_id in (await session.execute(
-            select(CatalogDraftJob.id).where(CatalogDraftJob.status.in_(ACTIVE)))).scalars()
-            if not manager.running(job_id)]
-        for job_id in ids:
-            await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
-    if ids:
-        logger.warning("服务重启：%d 个数据目录起草任务没有做完，已标为中断，可在数据目录页重试", len(ids))
-    return len(ids)
+    async def interrupt(session: AsyncSession, job_id: str) -> None:
+        await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
+
+    n = await runtime.interrupt_on_startup(CatalogDraftJob, interrupt, running=manager.running)
+    if n:
+        logger.warning("服务重启：%d 个数据目录起草任务没有做完，已标为中断，可在数据目录页重试", n)
+    return n
 
 
 # ==========================================================================
@@ -461,60 +523,35 @@ async def startup() -> int:
 # ==========================================================================
 
 
-class _JobRun:
-    """一个任务在后台的一次执行。"""
+class _JobRun(runtime.JobRun):
+    """一个起草任务在后台的一次执行（进队、停止、收尾的框架在 runtime.JobRun）。"""
+
+    kind = "draft"
+    model = CatalogDraftJob
 
     def __init__(self, job_id: str) -> None:
-        self.job_id = job_id
-        self.task: asyncio.Task[None] | None = None
-        #: 为什么停：user 使用者点了停止，shutdown 服务关停（标中断）
-        self.stop_reason: str | None = None
-        self._writes: set[asyncio.Task[Any]] = set()
-        self._lock = asyncio.Lock()
+        super().__init__(job_id)
         self._started: set[str] = set()
         self.source: Any = None
         self.actor: str | None = None
 
-    # ---------------------------------------------------------------- 库写入
+    # ---------------------------------------------------------------- 停止、出错时收尾
 
-    async def _write(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any:
-        """一条库写入：同一个任务的写入一条一条来；shield 起来，停止任务时不会打断写到一半的库操作。"""
-        async def go() -> Any:
-            async with self._lock:
-                async with SessionLocal() as session:
-                    return await fn(session, *args)
+    async def _stopped(self) -> None:
+        if self.stop_reason == "user":
+            await self._end("cancelled", None, "cancelled", None)
+        else:
+            await self._end("interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
 
-        task = asyncio.ensure_future(go())
-        self._writes.add(task)
-        task.add_done_callback(self._writes.discard)
-        return await asyncio.shield(task)
+    async def _failed(self, reason: str) -> None:
+        await self._end("failed", reason, "failed", reason)
 
-    async def _settle(self) -> None:
-        """等手上的库写入都写完（停止、出错收尾之前）。"""
-        while self._writes:
-            await asyncio.gather(*list(self._writes), return_exceptions=True)
+    async def _crashed(self, e: Exception) -> None:
+        from app.core.errors import describe_exception, first_line
 
-    # ---------------------------------------------------------------- 主流程
-
-    async def run(self) -> None:
-        try:
-            await self._run()
-        except asyncio.CancelledError:
-            current = asyncio.current_task()
-            if current is not None:
-                current.uncancel()
-            if self.stop_reason == "user":
-                await self._end("cancelled", None, "cancelled", None)
-            else:
-                await self._end("interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
-        except _JobFailed as e:
-            await self._end("failed", str(e), "failed", str(e))
-        except Exception as e:  # noqa: BLE001 - 任务整体出错：照实记下，不留在「起草中」
-            from app.core.errors import describe_exception, first_line
-
-            logger.exception("数据目录起草任务 %s 出错", self.job_id)
-            reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e))
-            await self._end("failed", reason, "failed", reason)
+        logger.exception("数据目录起草任务 %s 出错", self.job_id)
+        reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e))
+        await self._end("failed", reason, "failed", reason)
 
     async def _end(self, job_status: str, job_error: str | None, task_status: str, task_error: str | None) -> None:
         """停止、出错时收尾：等手上的库写入写完，再把没结束的表和任务记下来。收尾本身写不进去只记日志（下次启动时
@@ -603,11 +640,7 @@ class _JobRun:
             .execution_options(synchronize_session=False)
 
     async def _claim(self, session: AsyncSession) -> bool:
-        result = await session.execute(
-            update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id, CatalogDraftJob.status == "queued")
-            .values(status="running", started_at=utcnow()).execution_options(synchronize_session=False))
-        await session.commit()
-        return result.rowcount == 1
+        return await runtime.claim(session, CatalogDraftJob, self.job_id)
 
     async def _set_model(self, session: AsyncSession, model_id: str) -> None:
         await session.execute(update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id)
@@ -746,66 +779,11 @@ async def write_part(session: AsyncSession, source: Any, table: str, notes: dict
     return None
 
 
-class DraftJobs:
-    """进程内的起草任务（照 app/scheduler 的写法：一个 asyncio 任务一件事，停的时候给宽限）。"""
+class DraftJobs(runtime.JobManager):
+    """进程内的起草任务（start / stop / wait / running / shutdown 见 runtime.JobManager）。"""
 
     def __init__(self) -> None:
-        self._runs: dict[str, _JobRun] = {}
-
-    def _alive(self, job_id: str) -> _JobRun | None:
-        run = self._runs.get(job_id)
-        if run is None or run.task is None or run.task.done():
-            return None
-        try:
-            if run.task.get_loop() is not asyncio.get_running_loop():
-                self._runs.pop(job_id, None)   # 别的事件循环留下的（测试每个用例一个循环）
-                return None
-        except RuntimeError:
-            return None
-        return run
-
-    def start(self, job_id: str) -> None:
-        """开始在后台起草这个（排队中的）任务。"""
-        run = _JobRun(job_id)
-        run.task = asyncio.get_running_loop().create_task(run.run(), name=f"catalog-draft-{job_id}")
-        self._runs[job_id] = run
-
-        def forget(_: asyncio.Task[None]) -> None:
-            if self._runs.get(job_id) is run:
-                self._runs.pop(job_id, None)
-
-        run.task.add_done_callback(forget)
-
-    def running(self, job_id: str) -> bool:
-        return self._alive(job_id) is not None
-
-    async def stop(self, job_id: str, reason: str, *, grace: float = STOP_GRACE_S) -> bool:
-        """停掉这个进程里在跑的任务，等它收尾（最多 grace 秒）。不在这个进程里跑返回 False。"""
-        run = self._alive(job_id)
-        if run is None:
-            return False
-        if run.stop_reason is None:
-            run.stop_reason = reason
-            run.task.cancel()  # type: ignore[union-attr]
-        await asyncio.wait([run.task], timeout=grace)  # type: ignore[list-item]
-        return True
-
-    async def wait(self, job_id: str, timeout: float | None = None) -> None:
-        """等这个任务做完（测试、调试用）。不在跑直接返回。"""
-        run = self._alive(job_id)
-        if run is not None:
-            await asyncio.wait([run.task], timeout=timeout)  # type: ignore[list-item]
-
-    async def shutdown(self, *, grace: float = 3.0) -> None:
-        """关停：停掉进行中的任务（标中断），最多等 grace 秒收尾。"""
-        runs = [run for job_id in list(self._runs) if (run := self._alive(job_id)) is not None]
-        for run in runs:
-            if run.stop_reason is None:
-                run.stop_reason = "shutdown"
-                run.task.cancel()  # type: ignore[union-attr]
-        if runs:
-            await asyncio.wait([run.task for run in runs], timeout=grace)  # type: ignore[misc]
-            logger.info("已停止 %d 个数据目录起草任务", len(runs))
+        super().__init__(_JobRun, label="数据目录起草任务", name="catalog-draft")
 
 
 manager = DraftJobs()
@@ -814,6 +792,7 @@ manager = DraftJobs()
 __all__ = ["ACTIVE", "BLANK_SAMPLE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX",
            "INTERRUPTED_ERROR", "INTERRUPTED_WRITTEN", "MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE",
            "STOPPED_WRITTEN", "TASK_DONE", "TASK_STATUSES", "TextTooLong",
-           "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_idle", "guidance_of", "job_brief",
-           "job_dict", "job_out", "last_drafts", "latest_job", "load", "manager", "retry_tables", "save_guidance",
-           "startup", "sync_busy_message", "sync_draft", "sync_drafting", "tasks_of", "write_part"]
+           "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_draft_free", "ensure_idle",
+           "guidance_of", "import_busy_message", "import_write", "job_brief", "job_dict", "job_out", "last_drafts",
+           "latest_job", "load", "manager", "retry_tables", "save_guidance", "startup", "sync_busy_message", "sync_draft",
+           "sync_drafting", "tasks_of", "waiting_for", "write_part"]

backend/app/data/catalog_profile_jobs.py

@@ -0,0 +1,37 @@
+"""数据目录的后台剖析任务（草稿，下一步补全）。"""
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.data.catalog_job_runtime import ACTIVE, TASK_DONE
+from app.db.models import CatalogProfileJob, CatalogProfileTask
+
+
+class ProfileBusy(Exception):
+    """这个数据源已经有排队或进行中的剖析任务。message 给人看。"""
+
+
+async def active_job(session: AsyncSession, source_id: str) -> CatalogProfileJob | None:
+    return (await session.execute(
+        select(CatalogProfileJob).where(CatalogProfileJob.source_id == source_id,
+                                        CatalogProfileJob.status.in_(ACTIVE))
+        .order_by(CatalogProfileJob.created_at.desc()).limit(1)
+    )).scalar_one_or_none()
+
+
+def busy_message(row: Any, progress: tuple[int, int] | None) -> str:
+    name = getattr(row, "name", "") or ""
+    return f"数据源「{name}」正在剖析，请等这次剖析结束（或先停止它）再开始新的剖析"
+
+
+async def ensure_free(session: AsyncSession, row: Any) -> None:
+    job = await active_job(session, row.id)
+    if job is not None:
+        rows = (await session.execute(
+            select(CatalogProfileTask.status, func.count()).where(CatalogProfileTask.job_id == job.id)
+            .group_by(CatalogProfileTask.status))).all()
+        counts = dict(rows)
+        raise ProfileBusy(busy_message(row, (sum(counts.get(s, 0) for s in TASK_DONE), sum(counts.values()))))

backend/app/db/models.py

@@ -435,6 +435,96 @@ class CatalogDraftTask(Base):
     finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
 
 
+class CatalogProfileJob(Base, TimestampMixin):
+    """数据目录的一次后台剖析任务（data/catalog_profile_jobs.py）。逐表的结果在 catalog_profile_tasks。
+
+    以前剖析是一次同步长请求：预算按整次剖析算（选 50 张表只做得完两三张）、不能停、关掉页面报告就没了，对业务库查了什么
+    也无从查。现在建任务后在后台逐表剖析，每张表各有自己的查询次数和时长，进度、发过的查询（字面量去掉）随时落库。生命周期
+    和起草任务是同一套（data/catalog_job_runtime.py），表各建各的：进度字段完全不同。
+
+    同一个数据源同时只有一个排队或进行中的剖析任务：接口先查一遍，部分唯一索引兜底。和同源的起草任务按建立的先后排队。
+    服务重启时还没结束的标成 interrupted，不自动续跑。挂外键、级联删除：任务是数据源的附属记录。
+    """
+
+    __tablename__ = "catalog_profile_jobs"
+    __table_args__ = (
+        Index("ix_catalog_profile_jobs_source_created", "source_id", "created_at"),
+        Index("ux_catalog_profile_jobs_active", "source_id", unique=True,
+              sqlite_where=text("status IN ('queued', 'running')")),
+    )
+
+    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"))
+    # queued | running | succeeded | partial | failed | cancelled | interrupted
+    status: Mapped[str] = mapped_column(String(20), default="queued")
+    # selected 指定的表 | needed 需要剖析的表 | all 全部表
+    scope: Mapped[str] = mapped_column(String(20), default="selected")
+    # 建任务时的剖析设置快照（catalog_profile.ProfileSettings.to_dict）
+    settings: Mapped[dict[str, Any]] = mapped_column(default=dict)
+    # 署名（开了登录是认证过的用户名，否则请求头 X-Actor 的自报）
+    created_by: Mapped[str | None] = mapped_column(String(100), default=None)
+    started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
+    finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
+    # 整个任务发出的查询数、其中超时的、失败的（守卫拒掉的没发出去，不算）
+    queries: Mapped[int] = mapped_column(Integer, default=0)
+    timeouts: Mapped[int] = mapped_column(Integer, default=0)
+    errors: Mapped[int] = mapped_column(Integer, default=0)
+    # 任务级止损停下的原因：failed 连续失败、slow 超时比例；没有为 NULL
+    stop_reason: Mapped[str | None] = mapped_column(String(20), default=None)
+    # 提醒：最近的查询多数超时时的一句话，好转后清掉
+    notice: Mapped[str | None] = mapped_column(Text, default=None)
+    # 任务整体出错、止损、数据源中途停用或关掉剖析时的整句原因
+    error: Mapped[str | None] = mapped_column(Text, default=None)
+    # 重试建的任务记下是重试哪一个
+    retry_of: Mapped[str | None] = mapped_column(String(32), default=None)
+
+
+class CatalogProfileTask(Base):
+    """剖析任务里的一张表。按 seq（剖析顺序）排。状态：pending 排队、running 剖析中、ok 每项都有结论、partial 有项因为
+    运行原因没查成、failed 写入一直冲突或整张表出错、skipped 视图 / 表结构里没有 / 空表 / 没有可剖析的项、cancelled
+    （停止、止损或服务重启）。
+
+    检查项、发现、跳过的项、日期范围、发过的查询都是 JSON：条数受每张表的查询上限约束，大小有界。发过的查询里的 SQL
+    字面量都换成了「?」（sql_semantics.normalize_sql）：抽样到的键值不进 AgentLab 自己的库。"""
+
+    __tablename__ = "catalog_profile_tasks"
+    __table_args__ = (UniqueConstraint("job_id", "seq"),)
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
+    job_id: Mapped[str] = mapped_column(ForeignKey("catalog_profile_jobs.id", ondelete="CASCADE"), index=True)
+    seq: Mapped[int] = mapped_column(Integer)
+    # 表结构里的键；表结构里没有的（skipped）是请求里的原样名字
+    table_name: Mapped[str] = mapped_column(String(255))
+    status: Mapped[str] = mapped_column(String(20), default="pending")
+    # 检查项：[{kind, target, path, state, reason, detail, queries}]。建任务时按当时的目录列一遍，开始剖析这张表时按最新的
+    # 目录重列
+    checks: Mapped[list[Any] | None] = mapped_column(default=None)
+    queries: Mapped[int] = mapped_column(Integer, default=0)
+    elapsed_ms: Mapped[int] = mapped_column(Integer, default=0)
+    # 进行中时正在做哪一项的整句（「正在核对 orders.customer_id → customers.id」）
+    current: Mapped[str | None] = mapped_column(Text, default=None)
+    # 行数估算 {rows, method, at_least}
+    row_estimate: Mapped[dict[str, Any] | None] = mapped_column(default=None)
+    findings: Mapped[list[Any] | None] = mapped_column(default=None)
+    skipped: Mapped[list[Any] | None] = mapped_column(default=None)
+    date_ranges: Mapped[list[Any] | None] = mapped_column(default=None)
+    # 发过的查询：[{at, kind, target, sql, ms, rows, outcome, detail}]
+    query_log: Mapped[list[Any] | None] = mapped_column(default=None)
+    # 发现的合计 {relations_checked, relations_verified, codes, codes_pending, business_dates}；跳过的项数、其中因为运行
+    # 原因没查成的（表清单的「上次剖析」读这几列，不必读整份发现）
+    summary: Mapped[dict[str, Any] | None] = mapped_column(default=None)
+    skipped_count: Mapped[int] = mapped_column(Integer, default=0)
+    unfinished: Mapped[int] = mapped_column(Integer, default=0)
+    added: Mapped[int] = mapped_column(Integer, default=0)
+    updated: Mapped[int] = mapped_column(Integer, default=0)
+    removed: Mapped[int] = mapped_column(Integer, default=0)
+    version: Mapped[int] = mapped_column(Integer, default=0)
+    # 整张表的原因：表结构里没有、视图、空表、没有可剖析的项、写入一直冲突、停止 / 中断 / 止损
+    error: Mapped[str | None] = mapped_column(Text, default=None)
+    started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
+    finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
+
+
 class Setting(Base, TimestampMixin):
     """键值形式的用户设置，前端 Settings 页直接读写。"""