@@ -31,9 +31,9 @@
       "dict_any_count": 0
     },
     "loopx/chat_runtime.py": {
-      "any_count": 35,
+      "any_count": 36,
       "dict_any_count": 0,
-      "lines": 2024
+      "lines": 2128
     },
     "loopx/chat_server.py": {
       "any_count": 25,
 
@@ -32,6 +32,7 @@ def __init__(
         super().__init__(message)
         self.gate = gate
         self.error_code = error_code
+        self.native_error_info: dict[str, Any] | str | None = None
 
 
 class CodexChatTimeoutError(CodexChatAgentError):
@@ -1340,6 +1341,84 @@ def _control_items(self, method: str, params: dict[str, Any], deadline: float) -
                 raise self._runtime_error("Codex returned an invalid control inventory cursor.")
             seen_cursors.add(cursor)
 
+    def transport_retry_readback(self, turn_id: str) -> dict[str, Any] | None:
+        """Read native terminal/effect facts; notification history is insufficient.
+
+        Only a failed, latest Turn with a typed connection error and a complete
+        text-only inventory is eligible. Unknown items, tools (including reads),
+        delivery, additional user input and unreadable inventories stop recovery.
+        This is adapter IO evidence, never permission to reopen a Core Turn.
+        """
+        deadline = time.monotonic() + min(self.response_timeout_sec, 5.0)
+
+        def failed_turn(*, await_terminal: bool = False) -> dict[str, Any] | None:
+            while True:
+                turns = list(self._control_items("thread/turns/list", {
+                    "sortDirection": "desc", "itemsView": "summary",
+                }, deadline))
+                ids = [row.get("id") for row in turns]
+                if (not ids or any(not isinstance(value, str) or not value for value in ids)
+                        or len(set(ids)) != len(ids) or ids[0] != turn_id
+                        or any(row.get("status") not in {"completed", "interrupted", "failed"}
+                               for row in turns[1:])):
+                    return None
+                if turns[0].get("status") == "failed":
+                    break
+                # The native error notification can precede its persisted
+                # failed Turn. A provisional interrupted snapshot has no
+                # completion timestamp or error yet. Wait within the same
+                # readback deadline; never dispatch from this interim state.
+                if (not await_terminal or turns[0].get("status") not in {"inProgress", "interrupted"}
+                        or turns[0].get("completedAt") is not None or turns[0].get("error") is not None):
+                    return None
+                self._control_wait(deadline)
+            error = turns[0].get("error")
+            info = _retry_error_details(error).get("retry", {}).get("codex_error_info")
+            if not isinstance(info, dict) or len(info) != 1:
+                return None
+            variant, detail = next(iter(info.items()))
+            if not isinstance(detail, dict) or "httpStatusCode" not in detail:
+                return None
+            status = detail["httpStatusCode"]
+            if status is not None and not (type(status) is int and 500 <= status <= 599):
+                return None
+            return {"codex_error_info": {variant: {"httpStatusCode": status}}}
+
+        try:
+            terminal = failed_turn(await_terminal=True)
+            if terminal is None:
+                return None
+            with self._command_items_lock:
+                if self._command_items.get(turn_id):
+                    return None
+            items = list(self._control_items("thread/items/list", {"turnId": turn_id}, deadline))
+            item_ids: set[str] = set()
+            user_count = 0
+            for entry in items:
+                item = entry.get("item")
+                if entry.get("turnId") != turn_id or not isinstance(item, dict):
+                    return None
+                item_id = item.get("id")
+                if not isinstance(item_id, str) or not item_id or item_id in item_ids:
+                    return None
+                item_ids.add(item_id)
+                if item.get("type") not in {"userMessage", "agentMessage", "reasoning", "plan"}:
+                    return None
+                if item.get("type") == "userMessage":
+                    user_count += 1
+                if item.get("type") == "agentMessage" and (
+                    item.get("delivery") is not None or item.get("questions")
+                ):
+                    return None
+            # An accepted user item binds the complete inventory to this native
+            # request. A second user item can be a correction; never drop it.
+            if user_count != 1 or failed_turn() != terminal:
+                return None
+            return {"thread_id": self.thread_id, "turn_id": turn_id,
+                    "item_count": len(items), **terminal}
+        except (CodexChatAgentError, OSError, TypeError, ValueError):
+            return None
+
     def _control_wait(self, deadline: float) -> None:
         remaining = deadline - time.monotonic()
         if remaining <= 0:
@@ -1541,10 +1620,12 @@ def send(
                     str(turn.get("status") or "") if isinstance(turn, dict) else ""
                 )
                 if turn_status == "failed":
-                    raise _terminal_turn_error(
+                    error = _terminal_turn_error(
                         turn.get("error"),
                         "Codex app-server reported a terminal turn failure.",
                     )
+                    error.native_error_info = _retry_error_details(turn.get("error")).get("retry", {}).get("codex_error_info")
+                    raise error
                 if turn_status == "interrupted":
                     raise CodexChatAgentError(
                         "Codex app-server reported an interrupted turn.",
@@ -1567,10 +1648,12 @@ def send(
                             },
                         )
                     continue
-                raise _terminal_turn_error(
+                error = _terminal_turn_error(
                     params.get("error") if isinstance(params, dict) else None,
                     "Codex app-server reported a turn error.",
                 )
+                error.native_error_info = _retry_error_details(params.get("error")).get("retry", {}).get("codex_error_info") if isinstance(params, dict) else None
+                raise error
         visible_tail = display_filter.finish()
         if visible_tail and on_event:
             visible_delta_count += 1
 
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+import hashlib
 import json
 import logging
 import os
@@ -1515,14 +1517,13 @@ def _run_turn(
             # This caches no Session, Turn or authority facts: each checkpoint
             # still reads current state, including after a long provider wait.
             with effect_runtime_request_scope():
-                self._run_started_turn(
-                    session_id=session_id,
-                    turn_id=turn_id,
-                    message=message,
-                    attachments=attachments,
-                    adapter=adapter,
-                    loopx_execution=loopx_execution,
-                )
+                retry_binding: dict[str, str] = {}
+                while self._run_started_turn(
+                    session_id=session_id, turn_id=turn_id, message=message,
+                    attachments=attachments, adapter=adapter,
+                    loopx_execution=loopx_execution, retry_binding=retry_binding,
+                ):
+                    pass
         finally:
             done_event.set()
             with self.lock:
@@ -1538,7 +1539,8 @@ def _run_started_turn(
         attachments: list[AttachmentPayload],
         adapter: ChatRuntimeAdapter,
         loopx_execution: bool,
-    ) -> None:
+        retry_binding: dict[str, str],
+    ) -> bool | None:
         """The body of a Turn whose `queued -> starting` fact is already durable.
 
         `_run_turn` owns the single-flight release, so nothing here needs the
@@ -1652,6 +1654,14 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None:
                     adapter.goal_driver = None
             else:
                 from .usage_goal import observe_goal_execution
+                if isinstance(adapter, CodexAppServerAdapter):
+                    binding = self._transport_retry_binding(session_id, turn_id, adapter)
+                    if retry_binding and binding != retry_binding["sha256"]:
+                        raise CodexChatAgentError(
+                            "The original request configuration changed; transport recovery stopped.",
+                            error_code="host_gate", gate=None,
+                        )
+                    retry_binding["sha256"] = binding
                 # Manager/native/external conversations have different execution
                 # and waiting boundaries. Do not invent Goal timing for them.
                 observed_goal = str(session.get("goal_id") or "") if scope["kind"] == "owner_goal" else ""
@@ -1718,6 +1728,15 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None:
             self._fail_turn(session_id, turn_id, exc.error_code, str(exc), status="timed_out")
         except CodexChatAgentError as exc:
             event_buffer.close()
+            if execution_ended():
+                return
+            if (not loopx_execution and parse_native_goal_command(message) is None
+                    and isinstance(adapter, CodexAppServerAdapter)
+                    and exc.error_code in {"host_gate", "upstream_transport_failure"}
+                    and self._prepare_transport_retry(
+                        session_id, turn_id, adapter, retry_binding, exc.native_error_info,
+                    )):
+                return True
             if execution_ended():
                 return
             self._fail_turn(session_id, turn_id, exc.error_code, str(exc), status="failed", gate=exc.gate)
@@ -1740,6 +1759,91 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None:
             with self.lock:
                 self.turn_event_buffers.pop(key, None)
 
+    def _transport_retry_binding(
+        self, session_id: str, turn_id: str, adapter: CodexAppServerAdapter,
+    ) -> str:
+        """Compare the original request and current grants/config, not prose."""
+        session = self.store.load_session(session_id) or {}
+        turn = self.store.load_turn(session_id, turn_id) or {}
+        self._check_codex_home(session)
+        if session.get("project_context") is not None or session.get("steward_context") is not None:
+            self.project_contexts.session_context(session)
+        if session.get("project_context") is not None:
+            project_context_policy.validate_project_executor_scope(
+                str(session["agent_id"]), session["project_context"], self.capabilities,
+            )
+        manager = (self.manager_runtime_profile(
+            str(session.get("channel_id") or "manager"),
+            steward_context=session.get("steward_context"),
+        ) if is_manager_channel(session.get("channel_id")) else None)
+        native = adapter.session
+        facts = {
+            "session": {key: value for key, value in session.items() if key not in {
+                "status", "active_turn_id", "last_activity_at", "last_error_code", "updated_at",
+            }},
+            "request": {key: turn.get(key) for key in (
+                "turn_id", "client_turn_id", "host_claim_id", "message", "attachments", "origin",
+            )},
+            "native": {key: getattr(native, key) for key in (
+                "thread_id", "model", "reasoning_effort", "sandbox", "permissions_profile",
+                "runtime_profile", "project_context", "execution_mode",
+            )},
+            "manager": manager,
+            "allocation": manager_session_model_allocation(self, session) if manager else None,
+        }
+        return hashlib.sha256(json.dumps(facts, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
+
+    def _prepare_transport_retry(
+        self, session_id: str, turn_id: str, adapter: CodexAppServerAdapter,
+        retry_binding: dict[str, str], native_error_info: dict[str, Any] | str | None,
+    ) -> bool:
+        """One same-worker retry, reserved durably before backoff/dispatch.
+
+        The existing queued CAS and Core active claim own execution. The event
+        ledger retains the attempt across restart/compaction; a new worker never
+        reopens starting/terminal Turns or resets this allowance. Lark delivery
+        and other adapters do not decide whether to replay a model request.
+        """
+        from .control_plane.chat_turn_acceptance import managed_turn_execution_allowed
+
+        def current() -> bool:
+            with self.lock:
+                attached = self.adapters.get(session_id) is adapter
+            return (attached and not self.closed.is_set() and adapter.healthcheck()
+                    and managed_turn_execution_allowed(
+                        self.store.load_session(session_id), self.store.load_turn(session_id, turn_id),
+                    ) and retry_binding.get("sha256") == self._transport_retry_binding(session_id, turn_id, adapter)
+                    and not any(row.get("kind") == "turn.steered"
+                                for row in self.store.events_after(session_id, turn_id, None)))
+
+        try:
+            if not isinstance(native_error_info, dict):
+                return False
+            if retry_binding.get("attempt") or not current() or any(row.get("kind") == "transport.retry_planned"
+                                    for row in self.store.events_after(session_id, turn_id, None)):
+                return False
+            original = str((self.store.load_turn(session_id, turn_id) or {}).get("upstream_turn_id") or "")
+            proof = adapter.session.transport_retry_readback(original) if original else None
+            if proof is None or proof["codex_error_info"] != native_error_info or not current():
+                return False
+            delay = 2.0
+            retry_binding["attempt"] = "1"
+            self.store.append_event(session_id, turn_id, kind="transport.retry_planned", payload={
+                "attempt": 1, "upstream_turn_id": original,
+                "binding_sha256": retry_binding["sha256"], "readback": proof,
+                "not_before": (datetime.now(timezone.utc) + timedelta(seconds=delay)).isoformat(),
+            })
+            self.store.append_event(session_id, turn_id, kind="agent.phase", payload={
+                "phase": "retrying", "label": "连接暂时中断，已核验本轮未执行工具；正在原会话继续。",
+            })
+            if self.closed.wait(delay) or not current():
+                return False
+            # Stop/config changes during the final native read also win. A tool
+            # or new input appearing late cannot borrow the earlier proof.
+            return adapter.session.transport_retry_readback(original) == proof and current()
+        except (CodexChatAgentError, KeyError, OSError, TypeError, ValueError):
+            return False
+
     def _fail_turn(
         self,
         session_id: str,
 
@@ -114,6 +114,77 @@ def write(packet):
     return session, state
 
 
+@pytest.mark.parametrize("fault", [
+    None, "other", "cyberPolicy", "unauthorized", "usageLimitExceeded", "rateLimitExceeded",
+    "http401", "http429", "bad-status", "commandExecution", "fileChange", "dynamicToolCall",
+    "mcpToolCall", "webSearch", "collabAgentToolCall", "futureItem", "delivery", "correction",
+    "wrong-turn", "missing-data", "cursor-cycle", "new-turn", "late-tool", "changed-error",
+    "native-error-before-terminal", "pending-terminal", "interrupted", "completed",
+])
+def test_transport_recovery_requires_complete_native_terminal_and_effect_readback(tmp_path, monkeypatch, fault):
+    session = chat_agent.CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None),
+        messages=queue.Queue(), thread_id="recovery-thread", work_dir=tmp_path,
+        response_timeout_sec=.2)
+    calls = []
+    reads = 0
+
+    def write(packet):
+        nonlocal reads
+        method, params = packet["method"], packet["params"]
+        calls.append((method, params))
+        assert params["threadId"] == "recovery-thread"
+        if method == "thread/turns/list":
+            reads += 1
+            info = {"responseStreamDisconnected": {"httpStatusCode": None}}
+            if fault in {"other", "cyberPolicy", "unauthorized", "usageLimitExceeded", "rateLimitExceeded"}:
+                info = fault
+            if fault in {"http401", "http429", "bad-status"}:
+                info = {"responseStreamDisconnected": {"httpStatusCode": {
+                    "http401": 401, "http429": 429, "bad-status": True,
+                }[fault]}}
+            if fault == "changed-error" and reads > 1:
+                info = "other"
+            result = {"data": [{"id": "failed-turn", "status": "failed", "error": {
+                "codexErrorInfo": info, "message": "private fixture detail",
+            }}]}
+            if fault == "new-turn":
+                result["data"].insert(0, {"id": "new-turn", "status": "inProgress"})
+            if fault == "pending-terminal" or (fault == "native-error-before-terminal" and reads == 1):
+                result["data"][0].update(status="interrupted", error=None, completedAt=None)
+            if fault in {"interrupted", "completed"}:
+                result["data"][0].update(status=fault, error=None, completedAt=1)
+        else:
+            assert method == "thread/items/list"
+            item = {"id": "request", "type": "userMessage"}
+            if params.get("cursor"):
+                item = {"id": "answer", "type": "agentMessage", "delivery": None}
+                if fault == "late-tool":
+                    item["type"] = "commandExecution"
+                if fault == "correction":
+                    item["type"] = "userMessage"
+            elif fault in {"commandExecution", "fileChange", "dynamicToolCall", "mcpToolCall",
+                           "webSearch", "collabAgentToolCall", "futureItem"}:
+                item["type"] = fault
+            if fault == "delivery":
+                item.update(type="agentMessage", delivery={"status": "delivered"})
+            result = {"data": [{"turnId": "wrong" if fault == "wrong-turn" else "failed-turn", "item": item}],
+                "nextCursor": "second" if not params.get("cursor") or fault == "cursor-cycle" else None}
+            if fault == "missing-data":
+                result.pop("data")
+        session.messages.put({"id": packet["id"], "result": result})
+
+    monkeypatch.setattr(session, "_write", write)
+    proof = session.transport_retry_readback("failed-turn")
+    if fault in {None, "native-error-before-terminal"}:
+        assert proof == {"thread_id": "recovery-thread", "turn_id": "failed-turn", "item_count": 2,
+                         "codex_error_info": {"responseStreamDisconnected": {"httpStatusCode": None}}}
+        assert reads == (3 if fault else 2) and len(calls) == reads + 2
+        assert "private fixture" not in json.dumps(proof)
+    else:
+        assert proof is None
+    assert all(method in {"thread/turns/list", "thread/items/list"} for method, _ in calls)
+
+
 def test_stop_cancels_completed_and_late_command_items_but_preserves_older_terminal(tmp_path, monkeypatch):
     session, state = _stop_server(tmp_path, monkeypatch)
     session.interrupt("current-turn")