审批暂停与继续：按锚点恢复准入 + 不批准终止本轮

- langgraph_dev/http.py：_compatible_checkpoint/_history_admission 支持带锚点准入
  （以承载 __interrupt__ 的检查点为父状态，校验中断标识），payload 保留锚点；
  不带锚点时行为与原来完全一致
- middleware/dynamic_review.py：_abort_requested/_finish_turn + hook_config(can_jump_to=["end"])，
  在库消费中断后结束本轮（工具不执行、不再发起模型调用）；无标记时休眠

EvoScientist/langgraph_dev/http.py

@@ -275,8 +275,46 @@ async def bind_workspace_run(request: Request) -> JSONResponse:
     return JSONResponse(_run_payload(run))
 
 
+def _interrupt_ids(value: Any) -> set[str]:
+    """从 __interrupt__ 写入值里取出中断标识（Interrupt 对象或字典都兼容）。"""
+
+    items = value if isinstance(value, (list, tuple)) else [value]
+    found: set[str] = set()
+    for item in items:
+        ident = getattr(item, "id", None)
+        if ident is None and isinstance(item, dict):
+            ident = item.get("id") or item.get("interrupt_id")
+        if ident:
+            found.add(str(ident))
+    return found
+
+
+def _anchor_has_interrupt(checkpoint: Any, expected: str) -> bool:
+    """锚点处是否仍承载待审批中断（并核对中断标识）。
+
+    这是"按锚点恢复"的准入判据：只要承载该中断的检查点写入还在 PG，暂停就仍然
+    有效 —— 与运行时进程是否重启过、距暂停多久都无关。
+    """
+
+    found: set[str] = set()
+    for write in getattr(checkpoint, "pending_writes", None) or ():
+        try:
+            if len(write) < 3 or str(write[1]) != "__interrupt__":
+                continue
+            found |= _interrupt_ids(write[2])
+        except TypeError:
+            continue
+    if not found:
+        return False
+    # 网关未提供标识（或回退值 default）时只做存在性判定。
+    if not expected or expected == "default":
+        return True
+    return expected in found
+
+
 async def _compatible_checkpoint(conn, thread_id: str, assistant_id: str,
-                                 config: dict) -> tuple[bool, bool]:
+                                 config: dict, anchor: dict | None = None
+                                 ) -> tuple[bool, bool]:
     """Read through the API-owned saver and graph factory, never execute here."""
     from langgraph_api._checkpointer import get_checkpointer
     from langgraph_api.graph import get_graph, graph_exists
@@ -285,10 +323,15 @@ async def _compatible_checkpoint(conn, thread_id: str, assistant_id: str,
     saver = await get_checkpointer(conn=conn)
     read_config = {**config, "configurable": {
         **config.get("configurable", {}), "thread_id": thread_id,
-        "checkpoint_ns": "",
+        "checkpoint_ns": str((anchor or {}).get("checkpoint_ns") or ""),
     }}
-    # Admission always checks the current head, never a caller-selected ancestor.
-    read_config["configurable"].pop("checkpoint_id", None)
+    if anchor and str(anchor.get("checkpoint_id") or ""):
+        # 按锚点恢复：调用方（网关）指定了承载该中断的祖先检查点，暂停时就已落库。
+        # 只有该锚点处确实还有待审批写入才放行 —— 不依赖运行时当前头部。
+        read_config["configurable"]["checkpoint_id"] = str(anchor["checkpoint_id"])
+    else:
+        # Admission always checks the current head, never a caller-selected ancestor.
+        read_config["configurable"].pop("checkpoint_id", None)
     checkpoint = await saver.aget_tuple(read_config)
     if checkpoint is None:
         return False, False
@@ -300,6 +343,10 @@ async def _compatible_checkpoint(conn, thread_id: str, assistant_id: str,
         graph_id = assistant["graph_id"]
     if checkpoint.metadata.get("graph_id", graph_id) != graph_id:
         raise ValueError("checkpoint graph mismatch")
+    if anchor and str(anchor.get("checkpoint_id") or ""):
+        return True, _anchor_has_interrupt(
+            checkpoint, str(anchor.get("interrupt_id") or "")
+        )
     # get_graph enters coroutine/async-context-manager factories and binds the
     # same API saver used by the worker. aget_state also validates delta seeds.
     async with get_graph(graph_id, read_config, checkpointer=saver,
@@ -337,8 +384,11 @@ async def _has_legacy_history(conn, thread_id: str) -> bool:
     return True
 
 
-async def _history_admission(conn, thread_id, assistant_id, config, operation, history):
-    exists, pending = await _compatible_checkpoint(conn, thread_id, assistant_id, config)
+async def _history_admission(conn, thread_id, assistant_id, config, operation, history,
+                             anchor: dict | None = None):
+    exists, pending = await _compatible_checkpoint(
+        conn, thread_id, assistant_id, config, anchor
+    )
     if operation == "resume":
         return "resume" if exists and pending else "CHECKPOINT_RESUME_UNAVAILABLE"
     if pending:
@@ -398,6 +448,11 @@ async def create_recoverable_run(request: Request) -> JSONResponse:
             return JSONResponse({"code": "INVALID_RESUME_REQUEST"}, status_code=400)
     elif command is not None:
         return JSONResponse({"code": "INVALID_START_REQUEST"}, status_code=400)
+    # 按锚点恢复（可选，仅 resume）：网关在暂停时把"承载该中断的检查点"落库，
+    # 继续时回传。这样续接只依赖该检查点仍在 PG，而不依赖运行时当前头部。
+    anchor = value.get("anchor") if operation == "resume" else None
+    if not isinstance(anchor, dict) or not str(anchor.get("checkpoint_id") or ""):
+        anchor = None
     history = value.get("history")
     # History validation is performed under the create lock, after idempotent
     # lookup. Only that branch can attest this request did not create a Run.
@@ -427,12 +482,18 @@ async def create_recoverable_run(request: Request) -> JSONResponse:
         "run_request_id": run_request_id,
         "request_hash": request_hash,
     }
-    # The admission read and the worker must address the same current head.
-    # Forking from an ancestor is not part of the recoverable-run contract.
+    # The admission read and the worker must address the same checkpoint.
+    # Without an anchor that is the current head (forking from an ancestor is
+    # not part of the recoverable-run contract); with an anchor it is the
+    # ancestor recorded at pause time — which is exactly what makes the
+    # continuation independent of restarts and elapsed time.
     configurable = dict(payload["config"].get("configurable", {}))
     for key in ("checkpoint_id", "checkpoint_map"):
         configurable.pop(key, None)
     configurable.update(thread_id=thread_id, checkpoint_ns="")
+    if anchor is not None:
+        configurable["checkpoint_id"] = str(anchor["checkpoint_id"])
+        configurable["checkpoint_ns"] = str(anchor.get("checkpoint_ns") or "")
     payload["config"] = {**payload["config"], "configurable": configurable}
     async with _recoverable_run_lock:
         async with connect() as conn:
@@ -459,6 +520,7 @@ async def create_recoverable_run(request: Request) -> JSONResponse:
             try:
                 admission = await _history_admission(
                     conn, thread_id, assistant_id, payload["config"], operation, history,
+                    anchor,
                 )
             except Exception:
                 return JSONResponse({

EvoScientist/middleware/dynamic_review.py

@@ -9,7 +9,7 @@ from typing import Annotated, Any, NotRequired
 import httpx
 
 from EvoScientist.internal_service import internal_service_headers
-from langchain.agents.middleware import HumanInTheLoopMiddleware
+from langchain.agents.middleware import HumanInTheLoopMiddleware, hook_config
 from langchain.agents.middleware.types import AgentState, OmitFromSchema
 from langgraph.config import get_config
 
@@ -137,6 +137,31 @@ class DynamicReviewMiddleware(HumanInTheLoopMiddleware):
 
     state_schema = DynamicReviewState
 
+    @staticmethod
+    def _abort_requested() -> bool:
+        """本轮是否被要求"终止"（网关放弃某个待审批时注入的标记）。
+
+        方案 A：中断仍由库正常消费（该工具不执行），但随后不再发起模型调用，
+        直接把本轮跳到 end —— 即用户口径里的"不批准 → 停止这条对话"。
+        """
+
+        try:
+            config = get_config()
+        except RuntimeError:
+            return False
+        configurable = config.get("configurable") if isinstance(config, Mapping) else None
+        if not isinstance(configurable, Mapping):
+            return False
+        return bool(configurable.get("ai4sci_abort_turn"))
+
+    @classmethod
+    def _finish_turn(cls, update: dict[str, Any] | None) -> dict[str, Any] | None:
+        """终止本轮：保留库的决议结果（中断被消费掉），再结束本轮。"""
+
+        if not cls._abort_requested():
+            return update
+        return {**(update or {}), "jump_to": "end"}
+
     def before_agent(self, state: DynamicReviewState, runtime: Any) -> dict[str, Any]:
         del state, runtime
         run_id, review = _review_context()
@@ -153,6 +178,7 @@ class DynamicReviewMiddleware(HumanInTheLoopMiddleware):
             return {"_verified_review_mode": _manual_state(run_id, review)}
         return {"_verified_review_mode": await _resolve_async(run_id, review)}
 
+    @hook_config(can_jump_to=["end"])
     def after_model(
         self, state: DynamicReviewState, runtime: Any
     ) -> dict[str, Any] | None:
@@ -177,12 +203,12 @@ class DynamicReviewMiddleware(HumanInTheLoopMiddleware):
                 return None
             if review.get("requested_mode") != "auto":
                 # The gateway now requires manual approval for this turn.
-                return super().after_model(state, runtime)
+                return self._finish_turn(super().after_model(state, runtime))
             try:
                 _resolve_sync(current_run_id, review)
             except AutoReviewVerificationError:
                 # Auto approval could not be re-verified; fall back to review.
-                return super().after_model(state, runtime)
+                return self._finish_turn(super().after_model(state, runtime))
             return None
         if mode == "manual":
             # A LangGraph resume continues at this interrupted node and does not
@@ -197,11 +223,12 @@ class DynamicReviewMiddleware(HumanInTheLoopMiddleware):
                 try:
                     _resolve_sync(current_run_id, review)
                 except AutoReviewVerificationError:
-                    return super().after_model(state, runtime)
+                    return self._finish_turn(super().after_model(state, runtime))
                 return None
-            return super().after_model(state, runtime)
+            return self._finish_turn(super().after_model(state, runtime))
         raise AutoReviewVerificationError("REVIEW_MODE_STATE_INVALID")
 
+    @hook_config(can_jump_to=["end"])
     async def aafter_model(
         self, state: DynamicReviewState, runtime: Any
     ) -> dict[str, Any] | None:
@@ -218,11 +245,11 @@ class DynamicReviewMiddleware(HumanInTheLoopMiddleware):
             if review is None:
                 return None
             if review.get("requested_mode") != "auto":
-                return super().after_model(state, runtime)
+                return self._finish_turn(super().after_model(state, runtime))
             try:
                 await _resolve_async(current_run_id, review)
             except AutoReviewVerificationError:
-                return super().after_model(state, runtime)
+                return self._finish_turn(super().after_model(state, runtime))
             return None
         if mode == "manual":
             # Mirror the sync path: an injected auto context on a resume child
@@ -232,7 +259,7 @@ class DynamicReviewMiddleware(HumanInTheLoopMiddleware):
                 try:
                     await _resolve_async(current_run_id, review)
                 except AutoReviewVerificationError:
-                    return super().after_model(state, runtime)
+                    return self._finish_turn(super().after_model(state, runtime))
                 return None
-            return super().after_model(state, runtime)
+            return self._finish_turn(super().after_model(state, runtime))
         raise AutoReviewVerificationError("REVIEW_MODE_STATE_INVALID")