loopx-agent · Oct 18, 2026 · Oct 17, 2026 · Oct 17, 2026 · Oct 17, 2026 · Oct 17, 2026
@@ -123,6 +123,36 @@ try {
   assert.equal(exhausted.payload.turn_id, "t");
   assert.equal(exhausted.payload.events_url, eventsUrl);
   assert.equal(exhausted.payload.reconnect_attempts, 4);
+
+  // A process exit can close a healthy stream, then refuse connections while
+  // its listener restarts. Recover the stored terminal event from the original
+  // cursor; none of these requests may start a new Turn.
+  requests.length = 0;
+  serve = () => {
+    if (requests.length === 1) return new Response(sseBlock("before-exit", "turn.started", {}));
+    if (requests.length < 6) throw new TypeError("Load failed");
+    return new Response(sseBlock("after-exit", "turn.failed", {
+      error_code: "server_restarted", message: "Saved Turn was interrupted by service restart.",
+    }));
+  };
+  const restarted = await resumeChatTurnStreaming("s", "t").then(() => null, (error: unknown) => error);
+  assert.ok(restarted instanceof ChatApiError);
+  assert.equal(restarted.payload.error_code, "server_restarted", "the stored outcome survives the listener outage");
+  assert.deepEqual(requests, ["", ...Array(5).fill("?after=before-exit")], "EOF and refused reads retain the original event cursor");
+
+  // A longer outage remains bounded and offers the existing read-only recovery
+  // route, even when the browser throws an untyped network exception.
+  requests.length = 0;
+  serve = () => { throw new TypeError("Load failed"); };
+  const offline = await resumeChatTurnStreaming("s", "t").then(() => null, (error: unknown) => error);
+  assert.ok(offline instanceof ChatApiError);
+  assert.notEqual(offline.message, "Load failed");
+  assert.equal(offline.payload.reconnectable, true);
+  assert.equal(offline.payload.session_id, "s");
+  assert.equal(offline.payload.turn_id, "t");
+  assert.equal(offline.payload.events_url, eventsUrl);
+  assert.equal(offline.payload.reconnect_attempts, 6);
+  assert.equal(requests.length, 6, "a refused listener cannot cause unbounded reconnects");
 } finally {
   globalThis.fetch = originalFetch;
   globalThis.setTimeout = originalSetTimeout;

@@ -1044,8 +1044,12 @@ export async function streamChatTurn(
   const stallTimeoutMs = options.stallTimeoutMs ?? CHAT_STREAM_STALL_TIMEOUT_MS;
   let cursor = "";
   let attempts = 0;
+  let stalls = 0;
   let terminal = false;
-  while (!terminal && attempts < 4) {
+  // An owned desktop service can need several seconds to restart. Read-only
+  // reconnects cover that window (7.75s backoff), without resubmitting a Turn.
+  // Keep the existing four-watchdog limit for genuinely silent connections.
+  while (!terminal && attempts < 6) {
     signal?.throwIfAborted();
     const origin = typeof window === "undefined" ? "http://127.0.0.1" : window.location.origin;
     const url = new URL(chatApiUrl(eventsUrl), origin);
@@ -1094,20 +1098,20 @@ export async function streamChatTurn(
         }
         if (done || terminal) break;
       }
-      attempts = terminal ? attempts : attempts + 1;
+      if (!terminal) throw new ChatApiError("Agent 事件流连接已断开。", {});
     } catch (error) {
       if (signal?.aborted) throw error;
       attempts += 1;
-      if (attempts >= 4) {
-        // A stalled connection is a transport failure, not a caller abort, so
-        // it ends with the same typed error as any other exhausted reconnect.
-        if (stalled) {
-          throw new ChatApiError("Agent 事件流连接已断开。", {
-            reconnect_attempts: attempts,
-            stall_timeout_ms: stallTimeoutMs,
-          });
-        }
-        throw error;
+      if (stalled) stalls += 1;
+      if (attempts >= 6 || stalls >= 4) {
+        // Fetch refusals (including WebKit's raw "Load failed") and clean EOF
+        // are transport failures too. Preserve the Session/Turn recovery route
+        // through the receive wrapper instead of presenting them as an outcome.
+        throw new ChatApiError("Agent 事件流连接已断开。", {
+          ...(error instanceof ChatApiError ? error.payload : {}),
+          reconnect_attempts: attempts,
+          ...(stalled ? { stall_timeout_ms: stallTimeoutMs } : {}),
+        });
       }
       // A local phase keeps the pending reply honest while the reader resumes
       // from its cursor. It carries no event id, so the cursor is unchanged.

@@ -33,14 +33,14 @@
     "loopx/chat_runtime.py": {
       "any_count": 37,
       "dict_any_count": 0,
-      "lines": 2210
+      "lines": 2312
     },
     "loopx/chat_server.py": {
       "any_count": 25,
       "dict_any_count": 0
     },
     "loopx/chat_store.py": {
-      "lines": 2015
+      "lines": 2016
     },
     "loopx/codex_cli_probe.py": {
       "any_count": 37,

@@ -49,13 +49,16 @@
 from .chat_endpoints import AgentEndpointRegistry
 from .control_plane.turn_driver.host_binding import MANAGED_TURN_HOST
 from .control_plane.turn_driver.execution_profile import managed_execution_profile
+from .control_plane.chat_turn_acceptance import managed_turn_recovery_allowed
+from .file_lock import LOCK_HOLDER_DEAD, lock_holder_liveness, try_exclusive_file_lock
 from .kiro_cli_goal_mode import (
     KIRO_CLI_BIN,
     KIRO_CLI_CHAT_AGENT_ID,
     kiro_cli_chat_command,
 )
 from .chat_store import (
     CHAT_SESSION_MODE_ATTACHED,
+    CHAT_SESSION_MODE_MANAGED,
     TERMINAL_TURN_STATES,
     ChatSessionStore,
     normalize_agent_model_id,
@@ -95,6 +98,12 @@
 TRANSPORT_RETRY_MAX_RETRIES = 3
 TRANSPORT_RETRY_BACKOFF_SECONDS = 2.0
 
+SERVER_RESTARTED_MESSAGE = (
+    "LoopX Chat restarted before this turn finished. "
+    "Your conversation is saved; this turn was not retried automatically. "
+    "You can continue with a follow-up in this session."
+)
+
 
 class ChatTurnAcceptanceUnavailableError(Exception):
     """The durable acceptance attempt can be retried with the same request."""
@@ -903,9 +912,7 @@ def _ensure_adapter_locked(
                         expected_statuses={"queued", "starting", "running", "interrupting"},
                         status="failed",
                         error_code="server_restarted",
-                        error=("LoopX Chat restarted before this turn finished. "
-                               "Your conversation is saved; this turn was not retried automatically. "
-                               "You can continue with a follow-up in this session."),
+                        error=SERVER_RESTARTED_MESSAGE,
                         completed_at=utc_now(),
                     )
                     if failed is not None:
@@ -1543,6 +1550,97 @@ def _fail_queue_preparation(
                     if self.turn_done_events.get(key) is done_event:
                         self.turn_done_events.pop(key, None)
 
+    def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> None:
+        """Reconcile a killed local worker without starting a provider or replay.
+
+        The execution lock records its host and PID for the worker's lifetime.
+        No record, a live/foreign holder, or an attached host proves no loss.
+        """
+        session = self.store.load_session(session_id)
+        if not session:
+            return
+        turn_id = turn_id or session.get("active_turn_id")
+        if not turn_id:
+            return
+        path = self.store.turn_execution_lock_path(session_id, turn_id)
+        if lock_holder_liveness(path)[0] != LOCK_HOLDER_DEAD:
+            return
+        # Take the execution fence without consuming its death evidence, so
+        # dependency failures remain retryable. Recheck under the fence: a
+        # replacement may have changed the holder since the first liveness read.
+        with try_exclusive_file_lock(path, preserve_holder=True) as ownership:
+            if ownership is None or lock_holder_liveness(path)[0] != LOCK_HOLDER_DEAD:
+                return
+            session = self.store.load_session(session_id)
+            turn = self.store.load_turn(session_id, turn_id)
+            if managed_turn_recovery_allowed(
+                session, turn, managed=bool(session and session.get("session_mode") == CHAT_SESSION_MODE_MANAGED),
+                owner_dead=True,
+            ):
+                self._fail_turn(
+                    session_id, turn_id, "server_restarted", SERVER_RESTARTED_MESSAGE,
+                    status="failed", expected_statuses={"starting", "running", "interrupting"},
+                )
+            else:
+                # An earlier recovery may have persisted the exact failed result
+                # and then lost its event, message or claim write. The typed
+                # owner rejects `failed`, so only this narrow completion may
+                # finish that result; it never reopens or replays the Turn.
+                self._complete_recovered_terminalization(
+                    session_id, turn_id, session, turn,
+                )
+
+    def _complete_recovered_terminalization(
+        self,
+        session_id: str,
+        turn_id: str,
+        session: dict[str, Any] | None,
+        turn: dict[str, Any] | None,
+    ) -> None:
+        """Idempotently finish a recovery that already persisted `failed`.
+
+        `_fail_turn` writes the state first and then the terminal event, error
+        message and claim release. A transient store error in those follow-up
+        writes leaves the exact recovery result partially applied, and the
+        typed owner no longer admits the Turn. Repair only that exact result.
+        """
+        if (
+            session is None
+            or turn is None
+            or session.get("session_mode") != CHAT_SESSION_MODE_MANAGED
+            or turn.get("status") != "failed"
+            or turn.get("error_code") != "server_restarted"
+        ):
+            return
+        message = SERVER_RESTARTED_MESSAGE
+        if not any(
+            event.get("kind") == "turn.failed"
+            and (event.get("payload") or {}).get("error_code") == "server_restarted"
+            for event in self.store.events_after(session_id, turn_id, None)
+        ):
+            self.store.append_event(
+                session_id,
+                turn_id,
+                kind="turn.failed",
+                payload={"error_code": "server_restarted", "message": message},
+            )
+        if not any(
+            item.get("turn_id") == turn_id
+            and item.get("role") == "error"
+            and item.get("text") == message
+            for item in self.store.messages(session_id)
+        ):
+            self.store.append_message(
+                session_id, role="error", text=message, turn_id=turn_id,
+            )
+        if session.get("active_turn_id") == turn_id:
+            self.store.release_active_turn(
+                session_id,
+                turn_id,
+                last_activity_at=str(turn.get("completed_at") or utc_now()),
+                last_error_code="server_restarted",
+            )
+
     def _run_turn(
         self,
         *,
@@ -1567,30 +1665,36 @@ def _run_turn(
         # back here rather than left holding the key, which would make every
         # later dispatch of this same Turn a silent no-op.
         try:
-            started_turn = self.store.update_turn(
-                session_id,
-                turn_id,
-                expected_statuses={"queued"},
-                status="starting",
-                started_at=utc_now(),
-            )
-            if started_turn is None:
-                with self.lock:
-                    self.cancelled_turns.discard(key)
-                return
-            from .control_plane.effect_runtime import effect_runtime_request_scope
-
-            # Reuse one code revision across this worker's typed decisions.
-            # 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():
-                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
+            with try_exclusive_file_lock(
+                self.store.turn_execution_lock_path(session_id, turn_id),
+                agent_id="loopx-chat", operation="execute_chat_turn",
+            ) as ownership:
+                if ownership is None:
+                    return
+                started_turn = self.store.update_turn(
+                    session_id,
+                    turn_id,
+                    expected_statuses={"queued"},
+                    status="starting",
+                    started_at=utc_now(),
+                )
+                if started_turn is None:
+                    with self.lock:
+                        self.cancelled_turns.discard(key)
+                    return
+                from .control_plane.effect_runtime import effect_runtime_request_scope
+
+                # Reuse one code revision across this worker's typed decisions.
+                # 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():
+                    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()
             self.turn_release_wakeup.notify()

@@ -785,6 +785,7 @@ def _record_projection_exchange(self) -> None:
         )
 
     def _session_turn(self, session_id: str) -> None:
+        self.server.runtime_controller.recover_exited_turn(session_id)
         session = self.server.chat_store.load_session(session_id)
         if session is None or session.get("status") == "closed":
             self._send_error(
@@ -908,6 +909,7 @@ def _session_turn(self, session_id: str) -> None:
 
     def _session_snapshot(self, session_id: str) -> None:
         try:
+            self.server.runtime_controller.recover_exited_turn(session_id)
             self._send_json(project_chat_session_snapshot(
                 self.server.runtime_root, self.server.chat_store, session_id,
                 registry=self.server.registry_path))
@@ -934,6 +936,7 @@ def _list_sessions(self) -> None:
     def _turn_events(self, session_id: str, turn_id: str) -> None:
         from .presentation.codex_activity import public_activity_event
 
+        self.server.runtime_controller.recover_exited_turn(session_id, turn_id)
         if self.server.chat_store.load_turn(session_id, turn_id) is None:
             self._send_error("chat turn was not found", status=404)
             return

@@ -231,6 +231,9 @@ def _session_lock(self, session_id: str) -> threading.Lock:
     def _turn_path(self, session_id: str, turn_id: str) -> Path:
         return self._session_dir(session_id) / "turns" / f"{_opaque_id(turn_id, field='turn_id')}.json"
 
+    def turn_execution_lock_path(self, session_id: str, turn_id: str) -> Path:
+        return self._turn_path(session_id, turn_id).with_suffix(".execution")
+
     def _event_path(self, session_id: str, turn_id: str) -> Path:
         return self._session_dir(session_id) / "turns" / f"{_opaque_id(turn_id, field='turn_id')}.events.jsonl"
 

@@ -191,6 +191,23 @@ def managed_turn_execution_allowed(
     return _required_bool(result, "allowed")
 
 
+def managed_turn_recovery_allowed(
+    session: dict[str, Any] | None, turn: dict[str, Any] | None,
+    *, managed: bool, owner_dead: bool,
+) -> bool:
+    """Project process observations to the same typed Turn lifecycle owner."""
+    result = effect_runtime_result("chat.turn.recovery_allowed", {
+        "session": ({"status": session.get("status"),
+                     "active_turn_id": session.get("active_turn_id")}
+                    if session is not None else None),
+        "turn": ({"turn_id": turn.get("turn_id"), "status": turn.get("status")}
+                 if turn is not None else None),
+        "managed": managed,
+        "owner_dead": owner_dead,
+    })
+    return _required_bool(result, "allowed")
+
+
 def _raise_rejection(result: dict[str, Any]) -> None:
     code = result.get("code")
     if code in {"session_not_found", "session_closed"}:

@@ -629,6 +629,7 @@ export function createEffectRuntimeHandlers(
     ["collaboration.conversation.reply_context", lazyHandler(() => import("./collaboration/conversation_reply_context.ts"), ({projectConversationReplyContext}) => projectConversationReplyContext)],
     ["chat.turn.accept", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({planChatTurnAcceptance}) => planChatTurnAcceptance)],
     ["chat.turn.execution_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayContinueChatTurn}) => mayContinueChatTurn)],
+    ["chat.turn.recovery_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayRecoverChatTurn}) => mayRecoverChatTurn)],
     ["collaboration.delegation.observe", lazyHandler(() => import("./collaboration/delegation.ts"), ({transitionDelegationObservation}) => transitionDelegationObservation)],
     ["collaboration.delegation.observe_wake", lazyHandler(() => import("./collaboration/delegation.ts"), ({decideDelegationWakeObservation}) => decideDelegationWakeObservation)],
     ["collaboration.delegation.recover_validated_settlement", lazyHandler(() => import("./collaboration/delegation.ts"), ({recoverValidatedDelegationSettlement}) => recoverValidatedDelegationSettlement)],

@@ -284,6 +284,22 @@ export function mayContinueChatTurn(input: JsonObject): { allowed: boolean } {
   };
 }
 
+/** Only a positively observed dead local executor may be recovered on read.
+ * Missing local adapters or lost HTTP connections are not execution evidence. */
+export function mayRecoverChatTurn(input: JsonObject): { allowed: boolean } {
+  const session = decodeSession(input.session);
+  const turn = decodeActiveTurn(input.turn);
+  const managed = requireBoolean(input.managed, "managed");
+  const ownerDead = requireBoolean(input.owner_dead, "owner_dead");
+  return {
+    allowed: managed && ownerDead && session !== null && turn !== null
+      && session.status !== "closed"
+      && session.activeTurnId === turn.turnId
+      && (turn.status === "starting" || turn.status === "running"
+        || turn.status === "interrupting"),
+  };
+}
+
 function decodeAcceptance(value: unknown): AcceptanceCapsuleFacts | null {
   if (value === null || value === undefined) return null;
   const acceptance = requireJsonObject(value, "matching_turn.acceptance");