[codex] 修复 Host Bridge 失联恢复、执行租约与慢文件阻塞

agent/host_bridge/client.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import contextlib
+import logging
 import uuid
 from dataclasses import dataclass
 from pathlib import Path
@@ -27,6 +28,26 @@
 from core.common.diagnostic_log import current_diagnostic_context
 
 _HEARTBEAT_INTERVAL_S = 2.0
+logger = logging.getLogger(__name__)
+
+
+class HostBridgeRpcError(RuntimeError):
+    """保留传输状态；只有明确的暂时失联允许恢复探测和心跳。"""
+
+    def __init__(self, method: str, code: grpc.StatusCode, detail: str | None) -> None:
+        self.method = method
+        self.code = code
+        uncertainty = (
+            "；操作可能已生效，不得自动重发"
+            if method in {"Exec", "WriteStdin", "FileTool"}
+            else ""
+        )
+        super().__init__(f"Host Bridge {method} 失败: {code.name}: {detail}{uncertainty}")
+
+    @property
+    def transient(self) -> bool:
+        return self.code in {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED}
+
 
 
 @dataclass(frozen=True)
@@ -121,6 +142,8 @@ def __init__(
         self._stub = rpc.HostBridgeStub(self._channel)
         self._heartbeat_task: asyncio.Task[None] | None = None
         self._lease_error: Exception | None = None
+        self._opened = False
+        self._open_lock = asyncio.Lock()
         self._unconfirmed_owners: dict[str, str] = {}
         self._closed = False
 
@@ -142,6 +165,7 @@ async def probe(self) -> dict[str, Any]:
             pb.ContextRequest(context=self._request_context()),
             method="Probe",
             timeout=5,
+            lease=False,
         )
         return self._identity_reply(reply)
 
@@ -150,6 +174,7 @@ async def inspect(self) -> dict[str, Any]:
             self._stub.Inspect,
             pb.ContextRequest(context=self._request_context()),
             method="Inspect",
+            timeout=5,
             lease=False,
         )
         return self._identity_reply(reply)
@@ -288,6 +313,9 @@ async def terminate_owner(
     async def shutdown(self) -> ExecutionCleanupReport:
         if self._closed:
             return ExecutionCleanupReport((), (), ())
+        if not self._opened:
+            await self.close_transport()
+            return ExecutionCleanupReport((), (), ())
         await self._stop_heartbeat()
         reply: pb.CleanupReply = await self._call(
             self._stub.ShutdownManager,
@@ -358,25 +386,42 @@ async def _call(
         """发起一次 RPC；失败或取消均不重放可能已生效的操作。"""
         if self._closed:
             raise RuntimeError("Host Bridge manager 已关闭")
-        if method not in {"Heartbeat", "ShutdownManager"} and self._lease_error is not None:
-            raise RuntimeError(f"Host Bridge lease 已失效: {self._lease_error}")
+        if lease and self._lease_error is not None:
+            raise self._lease_error
         if lease:
-            self._ensure_heartbeat()
+            await self._open_manager()
         try:
             return await call(
                 request,
                 timeout=timeout,
                 metadata=(("authorization", f"Bearer {self._token}"),),
             )
         except grpc.aio.AioRpcError as exc:
-            uncertainty = (
-                "；操作可能已生效，不得自动重发"
-                if method in {"Exec", "WriteStdin", "FileTool"}
-                else ""
+            error = HostBridgeRpcError(method, exc.code(), exc.details())
+            if self._opened and error.code in {
+                grpc.StatusCode.NOT_FOUND, grpc.StatusCode.PERMISSION_DENIED,
+                grpc.StatusCode.UNAUTHENTICATED, grpc.StatusCode.FAILED_PRECONDITION,
+            }:
+                self._lease_error = error
+            raise error from exc
+
+    async def _open_manager(self) -> None:
+        """业务调用前只登记一次；失联续期不能重新创建已丢失的 manager。"""
+        async with self._open_lock:
+            if self._opened:
+                return
+            reply: pb.HeartbeatReply = await self._call(
+                self._stub.OpenManager,
+                pb.ContextRequest(context=self._request_context()),
+                method="OpenManager",
+                lease=False,
+                timeout=5,
             )
-            raise RuntimeError(
-                f"Host Bridge {method} 失败: {exc.code().name}: {exc.details()}{uncertainty}"
-            ) from exc
+            require_fields(reply, "alive")
+            if not reply.alive:
+                raise RuntimeError("Host Bridge 未确认 manager 登记")
+            self._opened = True
+            self._ensure_heartbeat()
 
     def _ensure_heartbeat(self) -> None:
         if self._heartbeat_task is None:
@@ -385,22 +430,36 @@ def _ensure_heartbeat(self) -> None:
             )
 
     async def _heartbeat_loop(self) -> None:
+        """暂时传输失败继续续期；租约丢失和身份错误终结旧 manager。"""
+        failures = 0
         try:
             while True:
-                await asyncio.sleep(_HEARTBEAT_INTERVAL_S)
-                reply: pb.HeartbeatReply = await self._call(
-                    self._stub.Heartbeat,
-                    pb.ContextRequest(context=self._request_context()),
-                    method="Heartbeat",
-                    timeout=5,
-                )
+                await asyncio.sleep(min(_HEARTBEAT_INTERVAL_S * (2 ** min(failures, 3)), 10))
+                try:
+                    reply: pb.HeartbeatReply = await self._call(
+                        self._stub.Heartbeat,
+                        pb.ContextRequest(context=self._request_context()),
+                        method="Heartbeat",
+                        lease=False,
+                        timeout=5,
+                    )
+                except HostBridgeRpcError as exc:
+                    if not exc.transient:
+                        raise
+                    failures += 1
+                    logger.warning("Host Bridge 心跳暂时失败，继续探测: %s", exc)
+                    continue
                 require_fields(reply, "alive")
                 if not reply.alive:
                     raise RuntimeError("Host Bridge 未确认 lease 存活")
+                if failures:
+                    logger.info("Host Bridge 心跳恢复")
+                failures = 0
         except asyncio.CancelledError:
             raise
         except Exception as exc:
             self._lease_error = exc
+            logger.error("Host Bridge manager 已失效: %s", exc)
 
 
 def _check_client_identity(socket_path: Path, boot_id: str, token: str) -> None:

agent/host_bridge/filesystem.py

@@ -7,8 +7,8 @@
 import difflib
 import logging
 import os
-from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
+from collections.abc import Callable
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, TypeVar
 
@@ -32,6 +32,50 @@ class _FileMutationState:
 _FILE_MUTATION_LOCKS: dict[str, _FileMutationState] = {}
 
 
+@dataclass
+class _FileIoState:
+    slots: asyncio.Semaphore = field(default_factory=lambda: asyncio.Semaphore(4))
+    users: int = 0
+
+
+_FILE_IO_SLOTS: dict[asyncio.AbstractEventLoop, _FileIoState] = {}
+
+
+async def _run_file_io(fn: Callable[[], T]) -> T:
+    """最多四个磁盘操作并行；取消后仍等物理工作结束才归还锁与 owner。"""
+    # 1. 等待名额时可以取消；线程启动后不能把取消当作工作已结束。
+    loop = asyncio.get_running_loop()
+    state = _FILE_IO_SLOTS.setdefault(loop, _FileIoState())
+    state.users += 1
+    try:
+        async with state.slots:
+            work = asyncio.create_task(asyncio.to_thread(fn))
+            cancelled: asyncio.CancelledError | None = None
+            while not work.done():
+                try:
+                    await asyncio.shield(work)
+                except asyncio.CancelledError as exc:
+                    cancelled = exc
+                except Exception:
+                    # 实际错误由 result 取回；同时发生取消时保留两种失败。
+                    break
+            # 2. 到这里线程已结束，外层才可以释放文件锁和 manager operation。
+            try:
+                result = work.result()
+            except Exception as exc:
+                if cancelled is not None:
+                    raise BaseExceptionGroup("文件操作取消且物理工作失败", [cancelled, exc]) from None
+                raise
+            if cancelled is not None:
+                raise cancelled
+            return result
+    finally:
+        state.users -= 1
+        if state.users == 0:
+            del _FILE_IO_SLOTS[loop]
+
+
+
 def _is_inside(path: Path, allowed_dir: Path) -> bool:
     try:
         _ = path.relative_to(allowed_dir)
@@ -106,12 +150,12 @@ def _get_file_mutation_key(file_path: Path) -> str:
 
 
 async def _run_with_file_mutation_lock(
-    file_path: Path, fn: Callable[[], Awaitable[T]]
+    file_path: Path, fn: Callable[[], T]
 ) -> T:
     """按规范化路径串行执行文件变更，并在异常或取消后回收锁状态。"""
 
     # 1. 登记当前调用，等待者也必须计入生命周期
-    key = _get_file_mutation_key(file_path)
+    key = await _run_file_io(lambda: _get_file_mutation_key(file_path))
     state = _FILE_MUTATION_LOCKS.get(key)
     if state is None:
         state = _FileMutationState(lock=asyncio.Lock())
@@ -121,7 +165,7 @@ async def _run_with_file_mutation_lock(
     try:
         # 2. 同一文件串行执行，取消也由 async with 释放底层锁
         async with state.lock:
-            return await fn()
+            return await _run_file_io(fn)
     finally:
         # 3. 最后一个持有者或等待者退出后再移除路径映射
         state.users -= 1
@@ -271,7 +315,7 @@ async def read_raw(self, path: str, **kwargs: Any) -> str | ToolResult:
                 allowed_dir=self._allowed_dir,
                 arguments={"path": path, **kwargs},
             )
-        return self.read_from_disk(path, **kwargs)
+        return await _run_file_io(lambda: self.read_from_disk(path, **kwargs))
 
     def read_from_disk(self, path: str, **kwargs: Any) -> str | ToolResult:
         """Read host bytes without applying the current Turn model projection."""
@@ -363,9 +407,9 @@ async def execute(self, path: str, content: str, **kwargs: Any) -> str | ToolRes
             )
             return result
         try:
-            file_path = _resolve_path(path, self._allowed_dir)
+            file_path = await _run_file_io(lambda: _resolve_path(path, self._allowed_dir))
 
-            async def _write() -> str | ToolResult:
+            def _write() -> str | ToolResult:
                 if file_path.exists() and file_path.is_dir():
                     return ToolResult(
                         text=f"写入文件失败：目标路径是目录：{path}", is_error=True
@@ -401,9 +445,9 @@ async def execute(
             )
             return result
         try:
-            file_path = _resolve_path(path, self._allowed_dir)
+            file_path = await _run_file_io(lambda: _resolve_path(path, self._allowed_dir))
 
-            async def _edit() -> str | ToolResult:
+            def _edit() -> str | ToolResult:
                 if not file_path.exists():
                     return ToolResult(text=f"错误：文件不存在：{path}", is_error=True)
                 if not file_path.is_file():
@@ -468,6 +512,10 @@ async def execute(self, path: str, **kwargs: Any) -> str | ToolResult:
                 arguments={"path": path, **kwargs},
             )
             return result
+        return await _run_file_io(lambda: self._list_from_disk(path))
+
+    def _list_from_disk(self, path: str) -> str | ToolResult:
+        """在线程中完成路径解析、目录遍历和文件类型查询。"""
         try:
             dir_path = _resolve_path(path, self._allowed_dir)
             if not dir_path.exists():

agent/host_bridge/host_bridge.proto

@@ -7,6 +7,7 @@ service HostBridge {
   rpc Inspect(ContextRequest) returns (IdentityReply);
   rpc ClaimBoot(ContextRequest) returns (ClaimBootReply);
   rpc Probe(ContextRequest) returns (IdentityReply);
+  rpc OpenManager(ContextRequest) returns (HeartbeatReply);
   rpc Heartbeat(ContextRequest) returns (HeartbeatReply);
   rpc Exec(ExecRequest) returns (ExecutionReply);
   rpc WriteStdin(WriteStdinRequest) returns (ExecutionReply);