复核 R-b1（卡住但没死的进程被当成没了）：续租约挪到单独的任务、专用连接（不走连接池，不被同一拍别的活拖住）；进程间请求另起任务办，不在心跳里等收尾；新增 cluster/fence 自我隔离——离上次续上租约超过一个租约时长，或者这段执行属于上一次隔离之前，执行在写事件、写断点、开审批、发外部写入、后台任务写库之前就停手（Fenced 是 BaseException，节点里的 except Exception 拦不住），续租时发现续晚了就隔离：停掉手上的运行和后台任务、不写终态，换一个身份重新登记；运行收尾先占住那一行、属主不是自己就不写；别的进程按租约过期再多等一个租约时长才算不在了（两边中间留一段）；run_events 的 (run_id, seq) 改成唯一索引（存量库先数重复，有就照实记日志、这一次不建，不让启动失败）。补测试：卡住期间被接手之后不盖终态、不写事件、序号不重；被接手的那次在别处接着跑完只结束一次、后面那一步只跑一次；收尾按属主写；撞号被库拒绝；多进程端到端加 6b、6c（真 SIGSTOP / SIGCONT）

backend/app/cluster/fence.py

@@ -0,0 +1,116 @@
+"""自我隔离（fencing）：本进程续不上租约的时候，当作自己认领的行可能已经被别的进程接手，停下手上的执行、不再写。
+
+**为什么。** 进程卡住了但没死（SIGSTOP、虚拟机暂停、事件循环被长同步代码卡住、库一时连不上）：别的进程看它的租约过期了，把它在跑
+的运行标中断、交给别人继续运行。它一缓过来，手上的执行接着往下走：照样写事件（和接手的那一边撞号）、照样写终态（把接手的结果
+盖掉）。所以：
+
+- 续租约在单独的任务、单独的连接上做（cluster/service.py），不被同一拍别的活拖住；最近一次续上的时刻记在这里（renewed）。
+- 本进程的执行在动手写之前问一句 check()：离上次续上已经超过一个租约时长（别的进程按两个租约时长才接手，中间留着一段），或者这段
+  执行是在上一次隔离之前开始的（世代不对），就抛 Fenced——写事件、写断点、写终态、写待办和动作、发外部写入之前都问。
+- 一旦发现续晚了（trip）：世代加一，手上的执行全部停掉（停的时候不写终态，那些行交给接手的进程按「属主不在了」处理），换一个新的
+  身份重新登记，之后发起的照常。
+
+Fenced 是 BaseException：节点、工具里常见的 except Exception 拦不住它，一路退到执行的最外层。没起协调（测试、脚本）时什么都不查。
+"""
+from __future__ import annotations
+
+import contextvars
+import logging
+import time
+from collections.abc import Callable
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class Fenced(BaseException):
+    """本进程续不上租约了（或者这段执行属于上一次隔离之前）：不许再写。"""
+
+
+_generation = 0
+_renewed: float | None = None
+_active = False
+_handlers: list[Callable[[], Any]] = []
+_bound: contextvars.ContextVar[int | None] = contextvars.ContextVar("agentlab_fence_generation", default=None)
+
+
+def activate(renewed_at: float | None = None) -> None:
+    """起协调、续上了第一次租约时（PostgreSQL；SQLite 只有一个进程，不用隔离）。"""
+    global _active, _renewed
+    _active = True
+    _renewed = time.monotonic() if renewed_at is None else renewed_at
+
+
+def deactivate() -> None:
+    global _active, _renewed
+    _active, _renewed = False, None
+
+
+def renewed(at: float | None = None) -> None:
+    """续上了一次租约。"""
+    global _renewed
+    _renewed = time.monotonic() if at is None else at
+
+
+def generation() -> int:
+    return _generation
+
+
+def bind() -> None:
+    """这段执行属于现在这一世代（运行的一段执行、后台任务、问数据的一轮开始时调）。"""
+    _bound.set(_generation)
+
+
+def stale(now: float | None = None) -> bool:
+    """离上次续上租约是不是已经超过一个租约时长。"""
+    if not _active or _renewed is None:
+        return False
+    from app.cluster.workers import lease_s
+
+    return (time.monotonic() if now is None else now) - _renewed > lease_s()
+
+
+def fenced_here() -> bool:
+    """当前这段执行还能不能写：世代不对（隔离之前开始的）或者租约续晚了。"""
+    if not _active:
+        return False
+    bound = _bound.get()
+    return (bound is not None and bound != _generation) or stale()
+
+
+def check() -> None:
+    """写之前问一句：不许写就抛 Fenced（续晚了的话顺手隔离）。"""
+    if not _active:
+        return
+    if stale():
+        trip("执行要写的时候发现租约已经续晚了")
+        raise Fenced()
+    bound = _bound.get()
+    if bound is not None and bound != _generation:
+        raise Fenced()
+
+
+def on_trip(handler: Callable[[], Any]) -> None:
+    if handler not in _handlers:
+        _handlers.append(handler)
+
+
+def trip(why: str) -> bool:
+    """隔离：世代加一，叫各处停掉手上的执行（cluster/service 再换身份重新登记）。已经在这一世代隔离过的不重复。交回这一次有没有动手。"""
+    global _generation, _renewed
+    if not _active or _renewed is None:
+        return False
+    _generation += 1
+    _renewed = None
+    logger.error("本进程的租约续晚了（%s）：当作认领的运行、问数据、后台任务可能已被别的进程接手，停掉手上的执行、不写终态，"
+                 "换一个身份重新登记", why)
+    for handler in list(_handlers):
+        try:
+            handler()
+        except Exception:  # noqa: BLE001
+            logger.exception("隔离时停执行出错")
+    return True
+
+
+__all__ = ["Fenced", "activate", "bind", "check", "deactivate", "fenced_here", "generation", "on_trip", "renewed",
+           "stale", "trip"]

backend/app/cluster/identity.py

@@ -33,6 +33,14 @@ def owner() -> str:
     return OWNER
 
 
+def regenerate() -> str:
+    """换一个身份（隔离之后重新登记用，cluster/fence.py）：同一个 pid、新的随机串。之前认领的行算「属主不在了」。"""
+    global OWNER, _PID
+    _PID = os.getpid()
+    OWNER = _make(_PID)
+    return OWNER
+
+
 def host() -> str:
     return _HOST
 
@@ -45,4 +53,4 @@ def describe(value: str | None) -> str:
     return f"{parts[0]} 上的进程 {parts[1]}" if len(parts) >= 2 else value
 
 
-__all__ = ["OWNER", "describe", "host", "owner"]
+__all__ = ["OWNER", "describe", "host", "owner", "regenerate"]

backend/app/cluster/leader.py

@@ -32,10 +32,16 @@ class Leadership:
         self.until: datetime | None = None
         self.epoch = 0
 
+    def single(self) -> None:
+        """只有本进程（SQLite）：直接算当班，不写库。"""
+        self.until = datetime.max.replace(tzinfo=None)
+
     def held(self, now: datetime | None = None) -> bool:
         """本地看：还在当班（租约离到期还有余量）。"""
         if self.until is None:
             return False
+        if self.until.tzinfo is None:
+            return True   # single()：只有本进程
         # 余量不超过租约的四分之一：租约调得很短（验收脚本）时也留得出当班的时间
         margin = min(MARGIN_S, lease_s() / 4)
         return (now or utcnow()) + timedelta(seconds=margin) < self.until
@@ -85,7 +91,10 @@ class Leadership:
 
         if self.until is None:
             return
+        single = self.until.tzinfo is None
         self.until = None
+        if single:
+            return
         async with SessionLocal() as session:
             await session.execute(update(ClusterLease).where(
                 ClusterLease.name == self.name, ClusterLease.owner == identity.owner())

backend/app/cluster/requests.py

@@ -80,14 +80,32 @@ async def _drain() -> int:
         if fn is None:
             logger.warning("收到不认识的进程间请求 %s/%s，不办", row.kind, row.target)
             continue
-        try:
-            await fn(row.target_id, row.actor, dict(row.payload or {}))
-            done += 1
-        except Exception:  # noqa: BLE001
-            logger.exception("办进程间请求 %s/%s %s 出错", row.kind, row.target, row.target_id)
+        # 各办各的、另起任务：停止要等执行收尾（最多十来秒），不能拖住心跳和后面几件
+        task = asyncio.ensure_future(_handle(fn, row.kind, row.target, row.target_id, row.actor, dict(row.payload or {})))
+        _running.add(task)
+        task.add_done_callback(_running.discard)
+        done += 1
     return done
 
 
+#: 正在办的请求（另起的任务要有人拿着引用）
+_running: set[asyncio.Future[Any]] = set()
+
+
+async def _handle(fn: Callable[..., Awaitable[Any]], kind: str, target: str, target_id: str, actor: str | None,
+                  payload: dict[str, Any]) -> None:
+    try:
+        await fn(target_id, actor, payload)
+    except Exception:  # noqa: BLE001
+        logger.exception("办进程间请求 %s/%s %s 出错", kind, target, target_id)
+
+
+async def settle(timeout: float = 5.0) -> None:
+    """等手上正在办的请求办完（测试、关停用）。"""
+    if _running:
+        await asyncio.wait(list(_running), timeout=timeout)
+
+
 async def discard_for(owners: set[str]) -> int:
     """收件人已经不在了的请求删掉（接手的进程做：那些行按属主不在了的规矩处理）。"""
     from app.db.models_cluster import ClusterRequest

backend/app/cluster/service.py

@@ -1,7 +1,11 @@
 """本进程在协调里的那一份：启动时登记、起心跳，关停时交班、删登记。
 
-心跳每一拍（AGENTLAB_WORKER_HEARTBEAT_S，缺省 5 秒）：
-1. 续自己的租约（登记不见了就重新登记）；
+续租约在单独的任务、单独的连接上（不走连接池，不被同一拍别的活拖住）；续晚了（离上次续上超过一个租约时长）就隔离：停掉手上的
+执行、不写终态，换一个身份重新登记（cluster/fence.py）。SQLite 只有一个进程：登记只写一次、不续租、本进程直接当班、接手只在
+启动时做，心跳只做本进程自己攒着的活。
+
+心跳每一拍（AGENTLAB_WORKER_HEARTBEAT_S，缺省 5 秒；PostgreSQL）：
+1. （续租约另有任务）；
 2. 刷新「别的进程谁还活着」；
 3. 取走发给本进程的请求照办（停止、往运行里记事件，cluster/requests.py）；
 4. 本进程自己的周期活（on_beat 登记的：访问密钥的请求记录写库……——这些是每个进程攒在自己内存里的，只能自己写）；
@@ -52,36 +56,48 @@ def leading() -> bool:
 class _Service:
     def __init__(self) -> None:
         self.task: asyncio.Task[None] | None = None
+        self.lease_task: asyncio.Task[None] | None = None
         self.wake: asyncio.Event | None = None
         self.reap_now = False
         self.last_reap = 0.0
         self.others: frozenset[str] = frozenset()
         self.started = False
+        #: SQLite：守卫已经确认只有本进程——登记只写一次、不续租、本进程直接当班、接手只在启动时做（不为协调多写库）
+        self.single = False
+        self._lease_conn: Any = None
 
     async def start(self) -> None:
         """登记、开始收发通知、起心跳。SQLite 时先取单进程的守卫（拿不到抛 guard.SingleProcessOnly）。"""
         if self.started:
             return
         guard.acquire()
-        exclusive = settings.db_dialect == "sqlite"
-        await workers.register(exclusive=exclusive)
+        self.single = settings.db_dialect == "sqlite"
+        await workers.register(exclusive=self.single)
         live = await workers.liveness()
         state.remember(live)
         self.others = live.others
         await signals.start()
         signals.on("q", self._on_request)
         signals.on("gone", self._on_gone)
-        if settings.db_dialect == "postgresql":
+        if not self.single:
             await signals.ready(timeout=5.0)
         self.wake = asyncio.Event()
         self.started = True
-        from app.cluster import caches
+        from app.cluster import caches, fence
 
         await caches.start()
-        await self._lead()
-        self.task = asyncio.create_task(self._loop(), name="agentlab-cluster-heartbeat")
+        if self.single:
+            # 只有本进程：当班不用争（不写库），心跳只做本进程自己攒着的活（访问密钥的请求记录）
+            leader.scheduler.single()
+            self.task = asyncio.create_task(self._light_loop(), name="agentlab-cluster-beats")
+        else:
+            await self._lead()
+            fence.activate()
+            fence.on_trip(self._on_fenced)
+            self.lease_task = asyncio.create_task(self._lease_loop(), name="agentlab-cluster-lease")
+            self.task = asyncio.create_task(self._loop(), name="agentlab-cluster-heartbeat")
         logger.info("工作进程 %s 已登记（%s，租约 %g 秒、心跳 %g 秒；现在另有 %d 个进程）", identity.describe(identity.owner()),
-                    "PostgreSQL，可以多进程" if settings.db_dialect == "postgresql" else "SQLite，单进程",
+                    "SQLite，单进程" if self.single else "PostgreSQL，可以多进程",
                     workers.lease_s(), workers.heartbeat_s(), len(live.others))
 
     def _on_request(self, parts: list[str]) -> None:
@@ -93,6 +109,75 @@ class _Service:
         if self.wake is not None:
             self.wake.set()
 
+    # ---------------- 续租约：单独的任务、单独的连接 ----------------
+
+    async def _lease_conn_open(self) -> Any:
+        import asyncpg
+
+        from app.cluster.signal import listen_dsn
+        from app.db.base import APPLICATION_NAME
+
+        if self._lease_conn is None or self._lease_conn.is_closed():
+            self._lease_conn = await asyncpg.connect(listen_dsn(), server_settings={"application_name": APPLICATION_NAME},
+                                                     timeout=max(1.0, workers.heartbeat_s()))
+        return self._lease_conn
+
+    async def renew(self) -> bool:
+        """续一拍租约（在专用连接上，不走连接池：池子被占满时取连接要等，等得比租约还长）。续晚了（离上次续上超过一个租约时长）
+        就不续这个身份了：隔离（cluster/fence.py），换身份重新登记。交回续没续上。"""
+        from datetime import timedelta
+
+        from app.cluster import fence
+        from app.db.base import utcnow
+
+        if fence.stale():
+            fence.trip("续租约的时候发现已经续晚了")
+            return False
+        now = utcnow()
+        until = now + timedelta(seconds=workers.lease_s())
+        try:
+            conn = await self._lease_conn_open()
+            done = await asyncio.wait_for(conn.execute(
+                "UPDATE cluster_workers SET heartbeat_at = $2, lease_until = $3 WHERE id = $1",
+                identity.owner(), now.replace(tzinfo=None), until.replace(tzinfo=None)), timeout=workers.heartbeat_s())
+        except Exception:  # noqa: BLE001 - 库一时连不上：这一拍没续上，下一拍再试（续晚了的话下一拍隔离）
+            logger.warning("续租约没续上，下一拍再试", exc_info=True)
+            with contextlib.suppress(Exception):
+                if self._lease_conn is not None:
+                    await self._lease_conn.close()
+            self._lease_conn = None
+            return False
+        if done.endswith(" 0"):
+            logger.warning("本进程的登记不见了（可能被当成没了），重新登记")
+            await workers.register()
+        fence.renewed()
+        return True
+
+    async def _lease_loop(self) -> None:
+        while True:
+            await self.renew()
+            await asyncio.sleep(workers.heartbeat_s())
+
+    def _on_fenced(self) -> None:
+        """隔离了：手上的执行由各处自己停（runner、job_runtime 登记的处理），这里换身份重新登记。"""
+        asyncio.ensure_future(self._reidentify())
+
+    async def _reidentify(self) -> None:
+        from app.cluster import fence
+
+        old = identity.owner()
+        new = identity.regenerate()
+        leader.scheduler.until = None
+        try:
+            await workers.register()
+            fence.renewed()
+            logger.warning("隔离之后换了身份重新登记：%s → %s（之前认领的行由当班进程按「属主不在了」接手）",
+                           identity.describe(old), identity.describe(new))
+        except Exception:  # noqa: BLE001
+            logger.exception("隔离之后重新登记失败，下一拍再试")
+
+    # ---------------- 当班、心跳的其余活 ----------------
+
     async def _lead(self) -> bool:
         was = leader.scheduler.held()
         try:
@@ -113,10 +198,10 @@ class _Service:
         return now_leading
 
     async def beat(self) -> None:
-        """心跳的一拍（测试里直接调）。"""
-        if not await workers.renew():
-            logger.warning("本进程的登记不见了（可能被当成没了），重新登记")
-            await workers.register()
+        """心跳的一拍（续租约另在 _lease_loop；测试里直接调）。SQLite 只做本进程自己攒着的活。"""
+        if self.single:
+            await self._own_work()
+            return
         live = await workers.liveness()
         state.remember(live)
         if self.others - live.others:
@@ -132,14 +217,17 @@ class _Service:
             await caches.poll()
         except Exception:  # noqa: BLE001
             logger.exception("读别的进程的缓存变化时出错")
+        await self._own_work()
+        if await self._lead() and (self.reap_now or time.monotonic() - self.last_reap >= workers.lease_s()):
+            # 隔一会儿的这一次信得过自己：属主是本进程的行（刚建、还没交给后台任务的）不当成没人管
+            await self.reap(live.trusting())
+
+    async def _own_work(self) -> None:
         for name, fn in list(_beats):
             try:
                 await fn()
             except Exception:  # noqa: BLE001
                 logger.exception("心跳里的「%s」出错，下一拍再试", name)
-        if await self._lead() and (self.reap_now or time.monotonic() - self.last_reap >= workers.lease_s()):
-            # 隔一会儿的这一次信得过自己：属主是本进程的行（刚建、还没交给后台任务的）不当成没人管
-            await self.reap(live.trusting())
 
     async def reap(self, live: workers.Liveness | None = None) -> dict[str, Any]:
         self.reap_now = False
@@ -173,20 +261,33 @@ class _Service:
             except Exception:  # noqa: BLE001 - 一拍出错（库一时不通）不该让心跳停下
                 logger.exception("心跳这一拍出错，下一拍照常进行")
 
+    async def _light_loop(self) -> None:
+        while True:
+            await asyncio.sleep(workers.heartbeat_s())
+            await self._own_work()
+
     async def stop(self) -> None:
         """关停：停心跳、交班、删登记（别的进程马上接手本进程留下的行），停收发通知。"""
         if not self.started:
             return
         self.started = False
-        if self.task is not None:
-            self.task.cancel()
-            with contextlib.suppress(asyncio.CancelledError, Exception):
-                await self.task
-        self.task = None
+        from app.cluster import fence
+
+        for task in (self.lease_task, self.task):
+            if task is not None:
+                task.cancel()
+                with contextlib.suppress(asyncio.CancelledError, Exception):
+                    await task
+        self.task = self.lease_task = None
+        fence.deactivate()
         with contextlib.suppress(Exception):
             await leader.scheduler.release()
         with contextlib.suppress(Exception):
             await workers.unregister()
+        if self._lease_conn is not None:
+            with contextlib.suppress(Exception):
+                await self._lease_conn.close()
+            self._lease_conn = None
         signals.publish("gone", identity.owner())
         await signals.stop()
         state.forget()

backend/app/cluster/workers.py

@@ -2,7 +2,7 @@
 
 - register / renew / unregister：启动时登记、隔一拍心跳续租约、平稳关停时删掉自己那一行（别的进程马上就知道它不在了，
   不必等租约到期）。
-- liveness()：现在哪些别的进程还活着（租约没过期）。各种「启动整理」「接手」都只碰属主不在了的行：
+- liveness()：现在哪些别的进程还活着（租约过期不到一个租约时长：再晚才算不在了，见 cluster/fence.py）。各种「启动整理」「接手」都只碰属主不在了的行：
   Liveness.alive(owner, local=…) —— 属主是别的进程：看它的租约；属主是自己：看这一行在本进程里是不是真的在跑（local），
   同一个进程里前一次起停留下的（测试里一个进程起停多次服务）照样算不在了；没有属主（升级前的行）算不在了。
 
@@ -65,9 +65,12 @@ async def liveness(now: datetime | None = None) -> Liveness:
 
     now = now or utcnow()
     me = identity.owner()
+    # 租约过期之后再多等一个租约时长才算不在了：卡住了但没死的进程（SIGSTOP、虚拟机暂停、库一时连不上）自己在一个租约时长时就
+    # 停手（cluster/fence.py），两边中间留着一段，不会它还在写、这边已经接手
+    floor = now - timedelta(seconds=lease_s())
     async with SessionLocal() as session:
         ids = (await session.execute(select(ClusterWorker.id).where(
-            ClusterWorker.lease_until > now, ClusterWorker.id != me))).scalars().all()
+            ClusterWorker.lease_until > floor, ClusterWorker.id != me))).scalars().all()
     return Liveness(frozenset(ids))
 
 

backend/app/core/job_runtime.py

@@ -78,6 +78,11 @@ class SerialWriter:
         self._lock = asyncio.Lock()
 
     async def __call__(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any:
+        from app.cluster import fence
+
+        # 本进程续不上租约了（cluster/fence.py）：这个任务可能已经被别的进程接手，不再写
+        fence.check()
+
         async def go() -> Any:
             async with self._lock:
                 async with SessionLocal() as session:
@@ -509,14 +514,22 @@ class JobRun:
         await self.writer.settle()
 
     async def run(self) -> None:
+        from app.cluster import fence
+
+        fence.bind()
         try:
             try:
                 await self._queue()
                 await self._run()
+            except fence.Fenced:
+                # 本进程续不上租约了：停手、不写收尾（这个任务由接手的进程按「属主不在了」处理）
+                logger.error("%s %s：本进程续不上租约，停手、不写收尾", self.kind, self.job_id)
             except asyncio.CancelledError:
                 current = asyncio.current_task()
                 if current is not None:
                     current.uncancel()
+                if fence.fenced_here():
+                    return
                 await self._stopped()
             except JobFailed as e:
                 await self._failed(str(e))
@@ -568,9 +581,18 @@ class JobManager:
         self._factory = factory
         self.label = label
         self.name = name
-        from app.cluster import requests
+        from app.cluster import fence, requests
 
         requests.handle("stop", name)(self._stop_requested)
+        fence.on_trip(self._stop_fenced)
+
+    def _stop_fenced(self) -> None:
+        """本进程隔离了（cluster/fence.py）：停掉手上的任务，停的时候不写收尾。"""
+        for job_id in list(self._runs):
+            run = self._runs.get(job_id)
+            if run is not None and run.task is not None and not run.task.done():
+                run.stop_reason = run.stop_reason or "fenced"
+                run.task.cancel()
 
     def model(self) -> Any:
         """任务表（停止时读属主用）：缺省取 factory 上的 model。"""

backend/app/db/base.py

@@ -571,6 +571,7 @@ def _migrate(conn) -> None:  # pragma: no cover - 同步回调
         cols = columns.get(table) or set()
         if all(c in cols for c in wanted):
             conn.execute(text(f"CREATE INDEX IF NOT EXISTS {name} ON {table} ({', '.join(wanted)})"))
+    _unique_run_event_seq(conn, columns)
     for name, table, wanted, where, dedupe in _UNIQUE_INDEX_MIGRATIONS:
         cols = columns.get(table) or set()
         if not all(c in cols for c in wanted):
@@ -583,6 +584,29 @@ def _migrate(conn) -> None:  # pragma: no cover - 同步回调
     _backfill_not_null(conn, columns)
 
 
+def _unique_run_event_seq(conn: Any, columns: dict[str, set[str]]) -> None:  # pragma: no cover - 同步回调
+    """run_events 的 (run_id, seq) 补唯一索引（ux_run_events_run_seq，换掉原来不唯一的 ix_run_events_run_seq）。存量库里可能已经
+    有撞了号的（多进程之前的进程内计数、按需裁判的老写法）：先数一数，有就照实记日志、这一次不建（照旧用不唯一的索引），不让启动
+    失败——那几次运行的封存核对本来就对不上，删哪一条都不对，要人看。"""
+    from sqlalchemy import inspect, text
+
+    if "run_events" not in columns:
+        return
+    names = {ix["name"] for ix in inspect(conn).get_indexes("run_events")}
+    if "ux_run_events_run_seq" in names:
+        return
+    dupes = conn.execute(text("SELECT run_id, seq, COUNT(*) FROM run_events GROUP BY run_id, seq HAVING COUNT(*) > 1 "
+                              "LIMIT 20")).all()
+    if dupes:
+        runs = sorted({str(r[0]) for r in dupes})
+        logger.warning("run_events 里有撞了号的事件（%d 处以上，涉及运行 %s）：这一次不建 (run_id, seq) 的唯一索引。"
+                       "这几次运行的封存核对对不上，请人工核对后处理", len(dupes), "、".join(runs[:5]))
+        return
+    conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ux_run_events_run_seq ON run_events (run_id, seq)"))
+    if "ix_run_events_run_seq" in names:
+        conn.execute(text("DROP INDEX IF EXISTS ix_run_events_run_seq"))
+
+
 #: init_db 在 PostgreSQL 上拿的会话级咨询锁的号：几个进程同时启动时，建表和迁移一个一个来（同一个号别处别用）
 _INIT_LOCK = 0x4167656E744C6162  # "AgentLab"
 

backend/app/db/models.py

@@ -721,7 +721,9 @@ class RunEvent(Base):
     """完整事件流落库 —— 这既是 trace，也是页面刷新后重建时间线的依据。"""
 
     __tablename__ = "run_events"
-    __table_args__ = (Index("ix_run_events_run_seq", "run_id", "seq"),)
+    # (run_id, seq) 唯一（编排完善第 2 波 W2-b）：撞号时报错，不悄悄写两条——封存按 seq 排序算哈希，撞了号核对时对时不对。
+    # 存量库由 db/base._migrate 补建（先查有没有重复，有就照实记日志、不建）
+    __table_args__ = (Index("ux_run_events_run_seq", "run_id", "seq", unique=True),)
 
     id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
     run_id: Mapped[str] = mapped_column(ForeignKey("runs.id", ondelete="CASCADE"))

backend/app/engine/runner.py

@@ -15,7 +15,8 @@ from langgraph.errors import GraphRecursionError
 from langgraph.types import Command
 from sqlalchemy import case, func, select, update
 
-from app.cluster import identity
+from app.cluster import fence, identity
+from app.cluster.fence import Fenced
 from app.core.bus import bus
 from app.core.config import settings
 from app.core.errors import describe_exception, raw_detail
@@ -92,7 +93,7 @@ class RunManager:
         from app.engine.checkpoints import open_saver
 
         self._cm = open_saver()
-        self._checkpointer = await self._cm.__aenter__()
+        self._checkpointer = _fenced_saver(await self._cm.__aenter__())
         # 上次进程被强杀时留下的记录。running 和 queued 要分开处理：
         # running 至少跑过一步、有 checkpoint，可以续；queued 还在排队，
         # 一个 checkpoint 都没写过，"恢复"它只会从 START 用默认值重跑一遍，
@@ -209,6 +210,8 @@ class RunManager:
         persist: bool = True,
         ts: float | None = None,
     ) -> None:
+        # 执行这次运行的那段（cluster/fence 绑定过世代的）写之前问一句：租约续晚了就不写（可能已经被别的进程接手）
+        fence.check()
         event = self._event(run_id, event_type, node_id=node_id, data=data, ts=ts)
         # 先落库，再广播。顺序反过来会漏事件，而且是结构性的漏：
         #
@@ -791,6 +794,11 @@ class RunManager:
         # 等待从 run.interrupted 算起——以前执行一直计到 finally，收尾那一截执行和等待两边都算，
         # active + wait 比墙钟还长（机器一忙就差出几十毫秒）
         paused: float | None = None
+        # 这段执行属于现在这一世代（cluster/fence.py）：本进程隔离之后，它不再写
+        fence.bind()
+        mine = identity.owner()
+        #: 被隔离了（续不上租约）：停手，不写终态（这次运行由接手的进程按「属主不在了」处理）
+        fenced = False
         # 先要并发名额（app/cluster/slots.py）。等名额的时候被停止（或者关停）也要照实收尾，见 _stopped_in_queue
         slot = slots.hold(run_id, origin_kind)
         try:
@@ -806,7 +814,7 @@ class RunManager:
                     row = await session.get(Run, run_id)
                     if row is not None:
                         row.status = "running"
-                        row.owner = identity.owner()
+                        row.owner = mine
                         # started_at 只在第一次开始时写。以前每段 _drive 都覆盖它，
                         # 审批恢复、接着跑之后，运行的"开始时间"就成了最后一段的开始
                         if row.started_at is None:
@@ -891,7 +899,13 @@ class RunManager:
                 elif snapshot.next:
                     status = "interrupted"
 
+            except Fenced:
+                fenced = True
+                logger.error("运行 %s：本进程续不上租约，停手、不写终态（由接手的进程按「属主不在了」处理）", run_id)
             except asyncio.CancelledError:
+                if fence.fenced_here():
+                    fenced = True
+                    raise
                 if self._closing:
                     # 服务关停（dev.sh 的 --reload 改一个文件就是一次）不是用户取消。
                     # checkpoint 完好，记成中断、重启后从断点接着跑——和强杀后启动
@@ -956,11 +970,12 @@ class RunManager:
                 elapsed = int((stop - started) * 1000)
                 self._finalizing.add(run_id)
                 try:
-                    await self._finalize(run_id, status, error, final_state, elapsed,
-                                         terminal=terminal, error_node=error_node, end_reason=end_reason)
+                    if not fenced:
+                        await self._finalize(run_id, status, error, final_state, elapsed, terminal=terminal,
+                                             error_node=error_node, end_reason=end_reason, owner=mine)
                     # 记 action.unknown、安排核实也算收尾：做完才把自己摘掉。以前摘掉之后才做，等运行「不在跑了」
                     # 再去读事件的一方（接口、测试）有一小段读不到这一条；库慢一点（PostgreSQL 满载时）就读不到
-                    if pending_action is not None:
+                    if pending_action is not None and not fenced:
                         from app.process.actions import on_unknown
 
                         try:
@@ -978,7 +993,7 @@ class RunManager:
                 # 不收的话每跑一次带代码节点的图就多一台常驻 microVM（几百 MB），
                 # 而且没有任何自动回收路径会碰它——闲置回收只在"下次有人执行
                 # 代码"时才触发。interrupted 要留着，人回来还得接着跑。
-                if status in ("succeeded", "failed", "cancelled"):
+                if status in ("succeeded", "failed", "cancelled") and not fenced:
                     with contextlib.suppress(Exception):
                         from app.sandbox.manager import sandbox_manager
 
@@ -992,11 +1007,13 @@ class RunManager:
         try:
             if self._closing:
                 return
+            if fence.fenced_here():
+                return
             actor = self._stopped_by.pop(run_id, None)
             self._finalizing.add(run_id)
             await self._finalize(run_id, "cancelled", "用户取消", {}, 0,
                                  terminal=(EventType.RUN_CANCELLED, {"reason": END_STOPPED, "actor": actor}),
-                                 end_reason=END_STOPPED)
+                                 end_reason=END_STOPPED, owner=identity.owner())
         finally:
             self._finalizing.discard(run_id)
             self._tasks.pop(run_id, None)
@@ -1021,6 +1038,7 @@ class RunManager:
         interrupts = chunk.get("__interrupt__")
         if not interrupts:
             return False
+        fence.check()
 
         for item in interrupts:
             value = getattr(item, "value", item)
@@ -1082,7 +1100,9 @@ class RunManager:
         terminal: tuple[EventType, dict[str, Any]] | None = None,
         error_node: str | None = None,
         end_reason: str | None = None,
+        owner: str | None = None,
     ) -> None:
+        """收尾。owner：执行这一段的属主——运行那一行的属主已经不是它（被别的进程当成没了接手了，cluster/fence.py）就什么都不写。"""
         usage = dict(state.get("usage") or {})
         output = dict(state.get("output") or {})
         now = datetime.now(timezone.utc)
@@ -1100,7 +1120,13 @@ class RunManager:
         run_class: str | None = None
         origin_kind: str | None = None
         async with SessionLocal() as session:
-            run = await session.get(Run, run_id)
+            # 先占住这一行再看属主（PostgreSQL 上 FOR UPDATE；SQLite 只有一个进程）：接手的进程改属主用的是带条件的 UPDATE，
+            # 两边一个接一个，被接手了的这边不写
+            run = (await session.execute(select(Run).where(Run.id == run_id).with_for_update())).scalar_one_or_none()
+            if run is not None and owner is not None and run.owner != owner:
+                logger.error("运行 %s 已经被别的进程接手（属主 %s），本进程不写终态", run_id, identity.describe(run.owner))
+                await session.rollback()
+                return
             if run:
                 app_id = run.app_id
                 run_class, origin_kind = run.run_class, run.origin_kind
@@ -1759,6 +1785,31 @@ async def _timing(session: Any, run: Run, elapsed_ms: int, now: datetime) -> dic
 run_manager = RunManager()
 
 
+def _fenced_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver:
+    """断点库的写也要过 cluster/fence：本进程续不上租约之后，手上的执行不再写断点（接手的进程会从断点接着跑，两边都写就分叉了）。"""
+    for name in ("aput", "aput_writes"):
+        original = getattr(saver, name, None)
+        if original is None:
+            continue
+
+        def guarded(*args: Any, __original: Any = original, **kwargs: Any) -> Any:
+            fence.check()
+            return __original(*args, **kwargs)
+
+        setattr(saver, name, guarded)
+    return saver
+
+
+def _stop_fenced() -> None:
+    """隔离了：停掉本进程在执行的运行（停的时候不写终态，见 _drive 的 fenced）。"""
+    for task in list(run_manager._tasks.values()):
+        if not task.done():
+            task.cancel()
+
+
+fence.on_trip(_stop_fenced)
+
+
 def _register_requests() -> None:
     """别的进程发来的：停止本进程在执行的运行、往它里面记一条事件（app/cluster/requests.py）。"""
     from app.cluster import requests

backend/app/process/actions.py

@@ -399,6 +399,10 @@ async def send(prep: Prepared, capability: Any, *, run_id: str, node_id: str, ex
              else "演练：探索运行里端点没有测试环境，没有发送请求")
         return _output(action_id, prep, "simulated", None, receipt)
 
+    # 本进程续不上租约了（cluster/fence.py）就不发：这次运行可能已经被别的进程接手
+    from app.cluster import fence
+
+    fence.check()
     # 先记 sent 再发：发到一半进程没了，重启后看到的是 sent，只能核实，不会当成没发过再发一次
     if not await _set(action_id, ("pending",), state="sent", request_at=utcnow()):
         raise ActionPending(action_id)

backend/tests/test_cluster_fence.py

@@ -0,0 +1,149 @@
+"""自我隔离（cluster/fence.py，复核 R-b1）：本进程卡住了、续不上租约，别的进程已经把它在跑的运行接手了——它缓过来之后不再写：
+不写事件（不和接手的一边撞号）、不写断点、不写终态（不把接手的结果盖掉）。在一个进程里装出来：把「上次续上租约」拨到一个租约
+时长以前，再把运行改成接手之后的样子。两种库都跑。"""
+from __future__ import annotations
+
+import asyncio
+import time
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+
+from app.cluster import fence
+from app.cluster.workers import lease_s
+from app.db.base import SessionLocal, new_id
+from app.db.models import Run, RunEvent
+from app.engine import compiler
+from app.engine.runner import run_manager
+from app.engine.schema import NodeType
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    fence.activate()
+    yield
+    fence.deactivate()
+    await run_manager.shutdown()
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+GRAPH = {"nodes": [node("start", "input", fields=[{"name": "question"}]),
+                   node("slow", "transform", mode="template", template="慢的这步"),
+                   node("next", "transform", mode="template", template="后面这步"),
+                   node("out", "output", fields=[{"name": "r", "value": "{{ nodes.next }}"}])],
+         "edges": [{"source": "start", "target": "slow"}, {"source": "slow", "target": "next"},
+                   {"source": "next", "target": "out"}]}
+
+
+@pytest.fixture
+def slow(monkeypatch):
+    """slow 那一步：第一次跑时卡住，等测试放行（好在它卡着的时候装出「续不上租约、被接手」）；之后立刻返回。"""
+    calls = {"n": 0}
+    gate = asyncio.Event()
+    entered = asyncio.Event()
+    original = compiler.RUNNERS[NodeType.TRANSFORM]
+
+    async def runner(state, ctx):
+        if ctx.node.id == "slow":
+            calls["n"] += 1
+            if calls["n"] == 1:
+                entered.set()
+                await gate.wait()
+        return await original(state, ctx)
+
+    monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, runner)
+    return {"calls": calls, "gate": gate, "entered": entered}
+
+
+async def _taken_over(run_id: str) -> None:
+    """别的进程看它的租约过期了、接手：标中断、属主清空（同 run_manager.reap 做的）。"""
+    from sqlalchemy import update
+
+    async with SessionLocal() as session:
+        await session.execute(update(Run).where(Run.id == run_id).values(
+            status="interrupted", owner=None, error="服务重启，运行已挂起，可从断点继续运行"))
+        await session.commit()
+
+
+async def _events(run_id: str) -> list[RunEvent]:
+    async with SessionLocal() as session:
+        return list((await session.execute(select(RunEvent).where(RunEvent.run_id == run_id)
+                                           .order_by(RunEvent.seq))).scalars())
+
+
+async def _no_repeated_seq(run_id: str) -> None:
+    async with SessionLocal() as session:
+        rows = (await session.execute(select(RunEvent.seq, func.count()).where(RunEvent.run_id == run_id)
+                                      .group_by(RunEvent.seq).having(func.count() > 1))).all()
+    assert rows == [], f"撞号：{rows}"
+
+
+async def _stall_and_take_over(slow) -> str:
+    run = await run_manager.start(graph=GRAPH, input_payload={"question": "q"})
+    await asyncio.wait_for(slow["entered"].wait(), timeout=15)
+    before = fence.generation()
+    fence.renewed(time.monotonic() - lease_s() - 1)    # 卡了一个多租约时长
+    await _taken_over(run.id)
+    slow["gate"].set()                                 # 缓过来，手上的执行接着往下走
+    for _ in range(200):
+        if not run_manager.is_active(run.id):
+            break
+        await asyncio.sleep(0.05)
+    assert not run_manager.is_active(run.id), "执行停了"
+    assert fence.generation() == before + 1, "发现续晚了：隔离了一次"
+    return run.id
+
+
+async def test_a_stalled_executor_does_not_overwrite_the_takeover(slow):
+    run_id = await _stall_and_take_over(slow)
+    async with SessionLocal() as session:
+        row = await session.get(Run, run_id)
+    assert row.status == "interrupted" and row.owner is None, "接手之后的样子没被盖掉"
+    events = await _events(run_id)
+    assert not [e for e in events if e.type in ("run.finished", "run.failed", "run.cancelled")]
+    assert not [e for e in events if e.node_id == "next"], "缓过来之后一条都没写"
+    await _no_repeated_seq(run_id)
+
+
+async def test_a_run_continued_elsewhere_finishes_once_and_the_stalled_side_stays_quiet(slow):
+    run_id = await _stall_and_take_over(slow)
+    fence.renewed()                                    # 换了身份重新登记之后，这一世代照常
+    await run_manager.continue_failed(run_id)          # 接手的那一边接着跑
+    for _ in range(300):
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in ("succeeded", "failed") and row.manifest_hash:
+            break
+        await asyncio.sleep(0.05)
+    assert row.status == "succeeded", row.error
+    events = await _events(run_id)
+    assert len([e for e in events if e.type == "run.finished"]) == 1
+    assert len([e for e in events if e.type == "node.started" and e.node_id == "next"]) == 1, "后面那一步只跑了一次"
+    await _no_repeated_seq(run_id)
+
+
+async def test_the_final_write_checks_the_owner():
+    async with SessionLocal() as session:
+        run = Run(status="running", owner="peer-host:1:abcd", graph={}, input={})
+        session.add(run)
+        await session.commit()
+    await run_manager._finalize(run.id, "succeeded", None, {}, 10, owner="mine:1:ffff")
+    async with SessionLocal() as session:
+        row = await session.get(Run, run.id)
+    assert row.status == "running" and row.finished_at is None, "属主已经不是自己：不写终态"
+
+
+async def test_a_repeated_seq_is_refused_by_the_database():
+    async with SessionLocal() as session:
+        run = Run(status="running", graph={}, input={})
+        session.add(run)
+        await session.flush()
+        session.add_all([RunEvent(id=new_id(), run_id=run.id, seq=1, type="log", ts=0, data={}),
+                         RunEvent(id=new_id(), run_id=run.id, seq=1, type="log", ts=0, data={})])
+        with pytest.raises(IntegrityError):
+            await session.commit()

backend/tests/test_multiprocess_e2e.py

@@ -11,7 +11,9 @@
    全站 2 个、给交互留底 1 个：评测借满了别的名额、还有题排着，问数据一来马上拿到留底，不等评测跑完；
 6. 每日花费上限跨进程不越过：两个进程同时往同一本账上记，记下的合计不超过上限；
 7. 强杀 B：租约过期后，B 上在跑的运行被照实标中断（可从断点继续），停在审批上的运行照常能在 A 上批、接着跑完；B 当着调度班的
-   话 A 接班。
+   话 A 接班；
+8. 卡住但没死（复核 R-b1）：A 在跑一次运行时被 SIGSTOP，B 等它的租约过期两倍时长后接手、标中断；A 缓过来（SIGCONT）之后发现续晚了，
+   停手、不写终态、不写事件，换一个身份重新登记——不管 B 有没有在这期间把那次运行接着跑完，都不双写、不盖掉终态、序号不重。
 
 缺省不跑：要 AGENTLAB_MULTIPROCESS_E2E=1 和 PostgreSQL 测试库（scripts/test-multiprocess.sh 起临时容器、设好这两样）。
 """
@@ -88,8 +90,16 @@ class Worker:
             self.proc.send_signal(signal.SIGKILL)
             self.proc.wait(timeout=10)
 
+    def pause(self) -> None:
+        self.proc.send_signal(signal.SIGSTOP)
+
+    def resume(self) -> None:
+        if self.proc.poll() is None:
+            self.proc.send_signal(signal.SIGCONT)
+
     def stop(self) -> None:
         if self.proc.poll() is None:
+            self.proc.send_signal(signal.SIGCONT)
             self.proc.send_signal(signal.SIGTERM)
             try:
                 self.proc.wait(timeout=15)
@@ -98,7 +108,7 @@ class Worker:
 
     def tail(self) -> str:
         try:
-            return self.log.read_text(encoding="utf-8", errors="replace")[-4000:]
+            return self.log.read_text(encoding="utf-8", errors="replace")[-12000:]
         except OSError:
             return ""
 
@@ -420,7 +430,8 @@ async def test_5b_the_interactive_reserve_is_there_when_evaluations_fill_the_res
         put = await client.put(f"{a.base}/api/settings", json={"values": {"concurrency": {
             "max_total": 2, "interactive_reserve": 1, "classes": {}}}})
         assert put.status_code == 200, put.text
-        assert put.json()["concurrency"]["classes"]["system"]["limit"] == 1, "别的类缺省借满到 全站 − 留底"
+        got = put.json()["concurrency"]
+        assert got["max_total"] - got["interactive_reserve"] == 1, "别的类最多借得到 全站 − 留底 = 1 个"
         slots.forget_policy()
         monkeypatch.setattr(settings, "sandbox_backend", "local")
         await cluster.start()
@@ -480,6 +491,88 @@ async def _spent() -> float:
     return float((await Ledger("e2e_spend").daily())["usd"])
 
 
+async def _stall_a_during_a_run(client, a: Worker, b: Worker, seconds: float) -> str:
+    """A 上发起一次运行，跑起来之后 SIGSTOP A，等 B 接手（标中断）。交回运行 id（A 还停着，调用方负责 SIGCONT）。"""
+    run_id = (await client.post(f"{a.base}/api/runs", json={"graph": slow_graph(seconds), "input": {"question": "q"},
+                                                             "origin": "canvas"})).json()["id"]
+    await _until(lambda: _owner_while_active(run_id), what="A 认领了运行")
+    await asyncio.sleep(0.5)
+    a.pause()
+
+    async def taken_over() -> Any:
+        row = await _run(client, b, run_id)
+        return row if row["status"] == "interrupted" else None
+
+    await _until(taken_over, timeout=LEASE_S * 6 + 10, what="B 接手了卡住的 A 上的运行")
+    return run_id
+
+
+async def _a_reidentified(client, a: Worker) -> None:
+    """A 缓过来之后换了身份重新登记：fixture 里记的 owner 跟着换。"""
+    old = a.owner
+
+    async def fresh() -> Any:
+        try:
+            me = (await client.get(f"{a.base}/api/system")).json()["cluster"]["me"]
+        except Exception:  # noqa: BLE001
+            return None
+        return me if me != old else None
+
+    a.owner = await _until(fresh, timeout=15, what="A 换身份重新登记")
+
+
+async def _settled_events(client, worker: Worker, run_id: str) -> list[dict[str, Any]]:
+    return (await client.get(f"{worker.base}/api/runs/{run_id}/events")).json()
+
+
+def _seqs_unique(events: list[dict[str, Any]]) -> None:
+    seqs = [e["seq"] for e in events]
+    assert len(seqs) == len(set(seqs)), f"撞号：{sorted(seqs)}"
+
+
+async def test_6b_a_stalled_worker_does_not_write_after_it_was_taken_over(workers):
+    a, b = workers["A"], workers["B"]
+    async with await _client() as client:
+        try:
+            run_id = await _stall_a_during_a_run(client, a, b, 6)
+        finally:
+            a.resume()
+        await _a_reidentified(client, a)
+        await asyncio.sleep(max(4.0, 3 * HEARTBEAT_S))     # A 手上那一步本来再过几秒就跑完：给它机会（它不该写）
+        row = await _run(client, b, run_id)
+        assert row["status"] == "interrupted", f"接手之后的状态被盖掉了：{row['status']}"
+        events = await _settled_events(client, b, run_id)
+        assert not [e for e in events if e["type"] in ("run.finished", "run.failed", "run.cancelled")]
+        assert any(e["type"] == "log" and e["data"].get("code") == "server_shutdown" for e in events)
+        _seqs_unique(events)
+
+
+async def test_6c_a_run_continued_by_the_other_worker_is_not_written_twice(workers):
+    a, b = workers["A"], workers["B"]
+    async with await _client() as client:
+        try:
+            run_id = await _stall_a_during_a_run(client, a, b, 3)
+            cont = await client.post(f"{b.base}/api/runs/{run_id}/continue", json={})
+            assert cont.status_code == 200, cont.text
+        finally:
+            a.resume()
+        await _a_reidentified(client, a)
+
+        async def done() -> Any:
+            row = await _run(client, b, run_id)
+            return row if row["status"] in ("succeeded", "failed") and row.get("manifest_hash") else None
+
+        row = await _until(done, timeout=60, what="B 接着跑完")
+        await asyncio.sleep(2.0)
+        assert row["status"] == "succeeded", row
+        assert (await _run(client, b, run_id))["status"] == "succeeded", "A 缓过来之后没有把终态改掉"
+        events = await _settled_events(client, b, run_id)
+        assert len([e for e in events if e["type"] == "run.finished"]) == 1
+        _seqs_unique(events)
+        verified = (await client.get(f"{b.base}/api/runs/{run_id}/verify")).json()
+        assert verified["ok"] is True, verified
+
+
 async def test_7_kill_a_worker(workers):
     a, b = workers["A"], workers["B"]
     async with await _client() as client: