From d6ecaaa7a3c8f255aafee923d0c3812724158775 Mon Sep 17 00:00:00 2001
From: ThearchyHelios <work@thearchyhelios.com>
Date: Mon, 19 Oct 2026 14:29:12 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=20R-b8=EF=BC=88=E9=9A=94?=
 =?UTF-8?q?=E7=A6=BB=E6=97=B6=20asyncio=20=E8=AE=B0=E4=B8=80=E6=9D=A1?=
 =?UTF-8?q?=E5=9B=9E=E8=B0=83=E5=BC=82=E5=B8=B8=EF=BC=89=EF=BC=9A=E6=96=AD?=
 =?UTF-8?q?=E7=82=B9=E5=86=99=E5=85=A5=E7=9A=84=E5=AE=88=E5=8D=AB=E6=94=B9?=
 =?UTF-8?q?=E6=88=90=E5=8D=8F=E7=A8=8B=E5=8C=85=E8=A3=85=E3=80=81=E5=9C=A8?=
 =?UTF-8?q?=E5=8D=8F=E7=A8=8B=E9=87=8C=E6=9F=A5=20fence=E2=80=94=E2=80=94L?=
 =?UTF-8?q?angGraph=20=E5=9C=A8=E8=8A=82=E7=82=B9=E7=9A=84=E5=AE=8C?=
 =?UTF-8?q?=E6=88=90=E5=9B=9E=E8=B0=83=E9=87=8C=E8=B0=83=20aput=5Fwrites?=
 =?UTF-8?q?=20=E6=8B=BF=E5=8D=8F=E7=A8=8B=E4=BA=A4=E7=BB=99=E5=90=8E?=
 =?UTF-8?q?=E5=8F=B0=E4=BB=BB=E5=8A=A1=EF=BC=8C=E4=BB=A5=E5=89=8D=E5=90=8C?=
 =?UTF-8?q?=E6=AD=A5=E5=8C=85=E8=A3=85=E5=9C=A8=E5=9B=9E=E8=B0=83=E9=87=8C?=
 =?UTF-8?q?=E5=B0=B1=E6=8A=9B=20Fenced=EF=BC=8C=E6=AF=8F=E6=AC=A1=E9=9A=94?=
 =?UTF-8?q?=E7=A6=BB=E9=83=BD=E6=89=93=E4=B8=80=E6=9D=A1=E3=80=8CException?=
 =?UTF-8?q?=20in=20callback=20=E2=80=A6=20FuturesDict.on=5Fdone=E3=80=8D?=
 =?UTF-8?q?=E5=8A=A0=20traceback=EF=BC=9B=E7=8E=B0=E5=9C=A8=E5=BC=82?=
 =?UTF-8?q?=E5=B8=B8=E7=95=99=E5=9C=A8=E5=86=99=E6=96=AD=E7=82=B9=E7=9A=84?=
 =?UTF-8?q?=E9=82=A3=E4=B8=AA=E4=BB=BB=E5=8A=A1=E9=87=8C=EF=BC=8C=E7=94=B1?=
 =?UTF-8?q?=20LangGraph=20=E7=85=A7=E5=B8=B8=E6=94=B6=E8=B5=B0=E3=80=82?=
 =?UTF-8?q?=E8=A1=A5=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=8D=A1=E4=BD=8F=E8=A2=AB?=
 =?UTF-8?q?=E6=8E=A5=E6=89=8B=E9=82=A3=E4=B8=80=E6=9D=A1=E9=87=8C=20asynci?=
 =?UTF-8?q?o=20=E4=B8=8D=E8=AE=B0=E5=9B=9E=E8=B0=83=E5=BC=82=E5=B8=B8?=
 =?UTF-8?q?=E3=80=81=E4=B8=8D=E8=AE=B0=E6=B2=A1=E4=BA=BA=E5=8F=96=E7=9A=84?=
 =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E5=BC=82=E5=B8=B8=EF=BC=88=E6=94=B9=E5=89=8D?=
 =?UTF-8?q?=E7=9A=84=E5=86=99=E6=B3=95=E8=BF=99=E6=9D=A1=E4=B8=8D=E8=BF=87?=
 =?UTF-8?q?=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/app/engine/runner.py        | 10 +++++++---
 backend/tests/test_cluster_fence.py | 11 ++++++++++-
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py
index 6599ca5b..141b9b41 100644
--- a/backend/app/engine/runner.py
+++ b/backend/app/engine/runner.py
@@ -1786,15 +1786,19 @@ run_manager = RunManager()
 
 
 def _fenced_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver:
-    """断点库的写也要过 cluster/fence：本进程续不上租约之后，手上的执行不再写断点（接手的进程会从断点接着跑，两边都写就分叉了）。"""
+    """断点库的写也要过 cluster/fence：本进程续不上租约之后，手上的执行不再写断点（接手的进程会从断点接着跑，两边都写就分叉了）。
+
+    包成协程、在协程里查（复核 R-b8）：LangGraph 在节点的完成回调里调 aput_writes 拿协程交给后台任务——同步包装在回调里就抛了，
+    每次隔离 asyncio 都记一条「Exception in callback … FuturesDict.on_done」；在协程里抛，异常留在写断点的那个任务里，由 LangGraph
+    照常收走。"""
     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:
+        async def guarded(*args: Any, __original: Any = original, **kwargs: Any) -> Any:
             fence.check()
-            return __original(*args, **kwargs)
+            return await __original(*args, **kwargs)
 
         setattr(saver, name, guarded)
     return saver
diff --git a/backend/tests/test_cluster_fence.py b/backend/tests/test_cluster_fence.py
index b48aaba6..18f99fd1 100644
--- a/backend/tests/test_cluster_fence.py
+++ b/backend/tests/test_cluster_fence.py
@@ -111,8 +111,17 @@ async def _stall_and_take_over(slow) -> str:
     return run.id
 
 
-async def test_a_stalled_executor_does_not_overwrite_the_takeover(slow):
+async def test_a_stalled_executor_does_not_overwrite_the_takeover(slow, caplog):
+    import gc
+    import logging
+
+    caplog.set_level(logging.ERROR, logger="asyncio")
     run_id = await _stall_and_take_over(slow)
+    await asyncio.sleep(0.1)
+    gc.collect()
+    await asyncio.sleep(0)
+    noisy = [r.getMessage() for r in caplog.records if r.name == "asyncio"]
+    assert noisy == [], f"隔离时 asyncio 不该记回调异常、没人取的任务异常（复核 R-b8）：{noisy}"
     async with SessionLocal() as session:
         row = await session.get(Run, run_id)
     assert row.status == "interrupted" and row.owner is None, "接手之后的样子没被盖掉"