多进程协调补齐几处：执行时长从拿到名额算起（排队等名额不算执行，墙钟本来就从开始执行算）；蓝图留痕、蓝图助手开轮先占住那一行再取最大值往下接；问数据等别的进程接着跑的运行时回库里等封存；按配方导入的 AI 起草、知识库重建索引另取跨进程的锁（多进程时重建进度记在设置表里，别的进程读得到、属主不在了照实说没做完）；调度循环一拍的间隔可配（AGENTLAB_SCHEDULER_TICK_S）

backend/app/api/copilot_blueprint.py

@@ -207,9 +207,12 @@ async def _conversation(session: AsyncSession, blueprint: Any, actor: str | None
 async def _open_turn(session: AsyncSession, conversation: Any, question: str) -> str:
     from sqlalchemy import func, select
 
+    from app.chat.job import hold_conversation
     from app.db.base import utcnow
     from app.db.models import ConversationTurn
 
+    # 先占住这段对话再取最大值往下接（同 api/conversations 开一轮）：多进程时同时两次开轮不撞号
+    await hold_conversation(session, conversation.id)
     seq = int((await session.execute(select(func.coalesce(func.max(ConversationTurn.seq), -1))
                                      .where(ConversationTurn.conversation_id == conversation.id))).scalar_one()) + 1
     turn = ConversationTurn(conversation_id=conversation.id, seq=seq, question=question, status="running")

backend/app/api/knowledge.py

@@ -477,8 +477,10 @@ class _ReindexJob:
         }
 
 
-#: 最近一次重建。只在进程里记：重建是幂等的，服务重启丢了进度，再点一次就是
+#: 最近一次重建（本进程里的）。重建是幂等的，服务重启丢了进度，再点一次就是。多进程时进度另记在设置表里（REINDEX_KEY，
+#: 带上做它的进程），别的进程上的「有没有在跑」「到哪了」读那一份；同一时间只跑一个靠跨进程的锁（kb-reindex）
 _REINDEX: _ReindexJob | None = None
+REINDEX_KEY = "cluster:kb_reindex"
 #: 后台任务要有人拿着引用，否则跑到一半可能被回收
 _REINDEX_TASK: Any = None
 
@@ -487,9 +489,59 @@ def reindex_job() -> _ReindexJob | None:
     return _REINDEX
 
 
-async def _run_reindex(job: _ReindexJob) -> None:
+async def _save_state(job: _ReindexJob) -> None:
+    """把进度记进设置表（多进程时别的进程读它；SQLite 只有一个进程，不记）。写不进去只记日志。"""
+    from app.cluster import counters, identity
+
+    if not counters.shared():
+        return
+    from app.db.base import SessionLocal
+    from app.db.models import Setting
+
+    try:
+        async with SessionLocal() as session:
+            row = await session.get(Setting, REINDEX_KEY)
+            value = {**job.view(), "owner": identity.owner()}
+            if row is None:
+                session.add(Setting(key=REINDEX_KEY, value=value))
+            else:
+                row.value = value
+            await session.commit()
+    except Exception:  # noqa: BLE001
+        logger.exception("重建索引的进度没记进库")
+
+
+async def _stored_state() -> dict[str, Any] | None:
+    """别的进程记的那一份：还写着在跑、做它的进程却不在了的，照实说没做完。SQLite 只有一个进程，没有这一份。"""
+    from app.cluster import counters, state
+
+    if not counters.shared():
+        return None
+    from app.db.base import SessionLocal
+    from app.db.models import Setting
+
+    async with SessionLocal() as session:
+        row = await session.get(Setting, REINDEX_KEY)
+    value = dict(row.value) if row is not None and isinstance(row.value, dict) else None
+    if value is None:
+        return None
+    owner = value.pop("owner", None)
+    if value.get("state") == "running" and not (await state.live_for([owner])).alive(owner, local=False):
+        value.update(state="failed", error="重建索引未完成：做这件事的服务进程停止了", hint="再点一次重建就是")
+    return value
+
+
+async def _run_reindex(job: _ReindexJob, lock: Any = None) -> None:
+    import asyncio
+
     from app.db.base import SessionLocal, utcnow
 
+    async def report() -> None:
+        while True:
+            await _save_state(job)
+            await asyncio.sleep(1.0)
+
+    reporter = asyncio.ensure_future(report())
     try:
         async with SessionLocal() as session:
             out = await kb.reindex(session, job.collection, on_progress=job.chunk_step)
@@ -503,6 +555,10 @@ async def _run_reindex(job: _ReindexJob) -> None:
         job.state, job.error, job.hint = "failed", f"重建索引未完成：{reason}", hint or None
     finally:
         job.finished_at = utcnow()
+        reporter.cancel()
+        await _save_state(job)
+        if lock is not None:
+            await lock.give_back()
 
 
 @kb_router.post("/reindex")
@@ -542,12 +598,22 @@ async def reindex(
             409, f"正在重建索引（{_REINDEX.collection or '全部集合'}，"
                  f"已完成 {running['done']}/{running['total']}），请等待完成后再操作",
         )
+    # 多进程：别的进程上正在重建的，跨进程的锁占不到（app/cluster/mutex），同样 409，说法照它记在库里的进度
+    from app.cluster import mutex
+
+    lock = mutex.lock("kb-reindex")
+    if not await lock.try_take():
+        other = await _stored_state() or {}
+        raise HTTPException(
+            409, f"正在重建索引（{other.get('collection') or '全部集合'}，"
+                 f"已完成 {other.get('done', 0)}/{other.get('total', 0)}），请等待完成后再操作",
+        )
     job = _REINDEX = _ReindexJob(collection, chunks, memories)
     if background:
-        _REINDEX_TASK = asyncio.create_task(_run_reindex(job))
+        _REINDEX_TASK = asyncio.create_task(_run_reindex(job, lock))
         return JSONResponse(job.view(), status_code=202)
 
-    await _run_reindex(job)
+    await _run_reindex(job, lock)
     if job.state == "failed":
         raise HTTPException(502, job.error + (f"。{job.hint}" if job.hint else ""))
     return job.result
@@ -557,8 +623,13 @@ async def reindex(
 async def reindex_status() -> dict[str, Any]:
     """最近一次重建的进度：{state, phase, total, done, chunks, memories, result, error…}。
 
-    还没重建过返回 {state: "idle"}。
+    还没重建过返回 {state: "idle"}。多进程时本进程没在做的，读做它的进程记在库里的那一份。
     """
+    if _REINDEX is not None and _REINDEX.state == "running":
+        return _REINDEX.view()
+    stored = await _stored_state()
+    if stored is not None and (_REINDEX is None or str(stored.get("started_at") or "") > _REINDEX.view()["started_at"]):
+        return stored
     return _REINDEX.view() if _REINDEX is not None else {"state": "idle"}
 
 

backend/app/api/settings.py

@@ -914,7 +914,8 @@ def _internal(key: str) -> bool:
     """通知自己记的账（每天汇总的认领行、自动建过邮件渠道的人，notify/digest.STATE_PREFIX）：不是用户设置，不列、不收。"""
     from app.notify.digest import STATE_PREFIX
 
-    return key.startswith(STATE_PREFIX)
+    # cluster: 开头的是多进程协调自己记的（重建索引的进度之类，app/cluster），同样不是用户设置
+    return key.startswith(STATE_PREFIX) or key.startswith("cluster:")
 
 
 DEFAULT_SETTINGS: dict[str, Any] = {

backend/app/blueprints/store.py

@@ -152,8 +152,14 @@ async def _verified_reuse(session: AsyncSession, op: dict[str, Any]) -> dict[str
 
 
 async def _next_seq(session: AsyncSession, blueprint_id: str) -> int:
-    from app.db.models_blueprints import BlueprintEvent
+    """这个蓝图下一条留痕的序号。先占住蓝图那一行（PostgreSQL 上是行锁，SQLite 上拿到写锁）再取最大值往下接：多进程时同一个
+    蓝图同时两处记留痕（卡片上改、助手提的），一个接一个，不撞 (blueprint_id, seq) 的唯一约束。updated_at 不动。"""
+    from sqlalchemy import update
 
+    from app.db.models_blueprints import Blueprint, BlueprintEvent
+
+    await session.execute(update(Blueprint).where(Blueprint.id == blueprint_id)
+                          .values(updated_at=Blueprint.updated_at).execution_options(synchronize_session=False))
     return int((await session.execute(select(func.coalesce(func.max(BlueprintEvent.seq), 0))
                                       .where(BlueprintEvent.blueprint_id == blueprint_id))).scalar_one()) + 1
 

backend/app/chat/drive.py

@@ -100,6 +100,20 @@ def add_usage(into: dict[str, Any] | None, more: Mapping[str, Any] | None) -> di
     return out
 
 
+async def _sealed(run_id: str, timeout: float) -> None:
+    """等终态的运行封存好（manifest_hash 写上），最多 timeout 秒。"""
+    from app.db.base import SessionLocal
+    from app.db.models import Run
+
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        async with SessionLocal() as session:
+            if await session.scalar(select(Run.manifest_hash).where(Run.id == run_id)):
+                return
+        await asyncio.sleep(0.1)
+
+
 class TurnDriver:
     """一轮问数据。drive() 跑完一轮，结果在 turn 上；on_phase(阶段) 在每一步开始时调。子类给差异（见模块说明）。
 
@@ -244,6 +258,9 @@ class TurnDriver:
             if row is None or row.status in ("succeeded", "failed", "cancelled", "interrupted"):
                 # 封存在终态之后：等执行任务收完尾（含封存），评分只读封存范围
                 await run_manager.wait_idle(run_id, timeout=CANCEL_GRACE_S)
+                if row is not None and row.status != "interrupted" and not run_manager.is_active(run_id):
+                    # 多进程：运行在别的进程里执行（审批之后被别的进程接着跑了），本进程等不到它的执行任务，回库里等封存
+                    await _sealed(run_id, CANCEL_GRACE_S)
                 async with SessionLocal() as session:
                     return await session.get(Run, run_id)
             await asyncio.sleep(0.2)

backend/app/cluster/mutex.py

@@ -45,6 +45,40 @@ class ClusterLock:
     def _waiters(self) -> Any:
         return self._local._waiters
 
+    async def try_take(self) -> bool:
+        """不等：本进程、别的进程都没人占着就占上，交回占没占到。占到了用 give_back 放（可以在另一个任务里放：后台任务做完才放）。"""
+        if self._local.locked():
+            return False
+        await self._local.acquire()
+        if settings.db_dialect != "postgresql":
+            return True
+        try:
+            from sqlalchemy import text
+
+            from app.db.base import engine
+
+            conn = await engine.connect()
+            try:
+                conn = await conn.execution_options(isolation_level="AUTOCOMMIT")
+                got = bool(await conn.scalar(text("SELECT pg_try_advisory_lock(:k)"), {"k": self.key}))
+            except BaseException:
+                await conn.close()
+                raise
+            if not got:
+                await conn.close()
+                self._local.release()
+                return False
+            self._conn = conn
+        except BaseException:
+            if self._local.locked() and self._conn is None:
+                self._local.release()
+            raise
+        return True
+
+    async def give_back(self) -> None:
+        """放掉 try_take 占到的。"""
+        await self.__aexit__(None, None, None)
+
     async def __aenter__(self) -> ClusterLock:
         await self._local.acquire()
         if settings.db_dialect != "postgresql":

backend/app/core/config.py

@@ -135,6 +135,8 @@ class Settings(BaseSettings):
     # （检查脚本的 --stacks、沙箱）必须关：库里的定时任务会照常到点发起运行、发出 webhook，
     # 拷贝里的任务和原库的是同一批，等于同一件事做两遍
     scheduler: bool = True
+    # 调度循环一拍的间隔（秒）：定时认领、事件派发、流程的时限和核实都按它查一次。缺省 15；验收脚本（多进程端到端）调小
+    scheduler_tick_s: float = 15.0
 
     # --- 工具 ---
     http_tool_timeout: int = 20

backend/app/data/recipe_imports.py

@@ -1703,8 +1703,9 @@ def _adopt_ai_work(p: Pipeline, book: _Book, snap: _Snap, work: Any) -> _Eval:
     return _evaluate(p, book, snap, work)
 
 
-#: 正在 AI 起草的暂存区（进程内）：同一暂存区不并发起草，否则两次发送各花一份钱，同意记录、用量还会互相覆盖
-#: （SE-11）。事件循环单线程，查和加之间没有 await，不会有两个请求同时通过；进程重启后自然清空
+#: 正在 AI 起草的暂存区（本进程里）：同一暂存区不并发起草，否则两次发送各花一份钱，同意记录、用量还会互相覆盖
+#: （SE-11）。事件循环单线程，查和加之间没有 await，不会有两个请求同时通过；进程重启后自然清空。多进程时另取跨进程的锁
+#: （run_ai_draft）
 _AI_RUNNING: set[str] = set()
 
 
@@ -1716,9 +1717,18 @@ async def run_ai_draft(session: AsyncSession, staging: ImportStaging, *, consent
         raise ImportRefused(422, "ai_consent_required", "需要先查看并同意发送的内容，才能让 AI 起草")
     if staging.id in _AI_RUNNING:
         raise ImportRefused(409, "ai_in_progress", "这次导入正在由 AI 起草，请等它完成后再试")
+    # 多进程：另一个进程上正在起草同一个暂存区的，跨进程的锁占不到，同样回 409（app/cluster/mutex）
+    from app.cluster import mutex
+
+    lock = mutex.lock(f"recipe-ai:{staging.id}")
     _AI_RUNNING.add(staging.id)
     try:
-        return await _run_ai_draft(session, staging, preview_sha256=preview_sha256, signed_by=signed_by)
+        if not await lock.try_take():
+            raise ImportRefused(409, "ai_in_progress", "这次导入正在由 AI 起草，请等它完成后再试")
+        try:
+            return await _run_ai_draft(session, staging, preview_sha256=preview_sha256, signed_by=signed_by)
+        finally:
+            await lock.give_back()
     finally:
         _AI_RUNNING.discard(staging.id)
 

backend/app/engine/runner.py

@@ -792,6 +792,8 @@ class RunManager:
         # active + wait 比墙钟还长（机器一忙就差出几十毫秒）
         paused: float | None = None
         async with slots.hold(run_id, origin_kind):
+            # 执行时长从拿到名额算起：排队等名额的那一段不算执行（墙钟从 started_at 算，started_at 也是拿到名额之后才写）
+            started = time.perf_counter()
             try:
                 async with SessionLocal() as session:
                     row = await session.get(Run, run_id)

backend/app/scheduler/__init__.py

@@ -34,9 +34,9 @@ async def start(*, interval: float | None = None, config: Settings | None = None
     if not scheduler_enabled(config):
         logger.info("定时任务调度已关闭（AGENTLAB_SCHEDULER），不会自动发起运行")
         return False
-    from app.scheduler.loop import TICK_SECONDS, runner
+    from app.scheduler.loop import runner
 
-    await runner.start(interval if interval is not None else TICK_SECONDS)
+    await runner.start(interval if interval is not None else float((config or settings).scheduler_tick_s))
     return True