Skip to content

fix(chat): recover lost execution workers without replay #7970

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
loopx-agent merged 2 commits into from
Nov 3, 2026
Merged

fix(chat): recover lost execution workers without replay #7970

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
84c0f28
fix(chat): recover lost execution workers without replay
loopx-agent Nov 3, 2026
f4fc03b
test(chat): verify worker recovery scope across session changes
loopx-agent Nov 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions loopx/chat_ingress.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,72 @@ class ChatIngressStore:
def runtime_activation_lock(self):
return chat_runtime_activation_lock(self.root.parent)

def queued_turns(self, session_id: str) -> list[dict[str, Any]]:
from .chat_store import _read_json, CHAT_TURN_SCHEMA_VERSION

session = self.load_session(session_id)
if session is None:
raise KeyError("chat session was not found")
active_turn_id = str(session.get("active_turn_id") or "")
turns_dir = self._session_dir(session_id) / "turns"
rows: list[dict[str, Any]] = []
for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
if path.name.endswith(".events.json"):
continue
payload = _read_json(path)
if (
payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
and payload.get("status") == "queued"
and str(payload.get("turn_id") or "") != active_turn_id
):
rows.append(payload)
return sorted(
rows,
key=lambda item: (
str(item.get("created_at") or ""),
str(item.get("turn_id") or ""),
),
)

def complete_recovered_terminalization(
self, session_id: str, turn_id: str,
session: dict[str, Any] | None, turn: dict[str, Any] | None,
*, recovery_messages: dict[str, str],
) -> None:
"""Finish the lifecycle owner's already-persisted failure, never dispatch.

State precedes the event, error message and claim release. If one of
those writes failed, retain the exact recovery result and idempotently
finish only its missing writes under the caller's execution fence.
"""
from .chat_store import CHAT_SESSION_MODE_MANAGED, utc_now

if (session is None or turn is None
or session.get("session_mode") != CHAT_SESSION_MODE_MANAGED
or turn.get("status") != "failed"):
return
code = str(turn.get("error_code") or "")
message = recovery_messages.get(code)
if message is None or (code != "server_restarted" and turn.get("error") != message):
return # Match this owner's recorded recovery, not arbitrary failures.
if not any(
event.get("kind") == "turn.failed"
and (event.get("payload") or {}).get("error_code") == code
and (event.get("payload") or {}).get("message") == message
for event in self.events_after(session_id, turn_id, None)
):
self.append_event(session_id, turn_id, kind="turn.failed",
payload={"error_code": code, "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.messages(session_id)
):
self.append_message(session_id, role="error", text=message, turn_id=turn_id)
if session.get("active_turn_id") == turn_id:
self.release_active_turn(session_id, turn_id,
last_activity_at=str(turn.get("completed_at") or utc_now()),
last_error_code=code)

def create_queued_turn(
self,
session_id: str,
Expand Down
119 changes: 52 additions & 67 deletions loopx/chat_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,10 @@
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 .file_lock import (
LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED,
_lock_id, lock_holder_host_label, lock_holder_liveness, try_exclusive_file_lock,
)
from .kiro_cli_goal_mode import (
KIRO_CLI_BIN,
KIRO_CLI_CHAT_AGENT_ID,
Expand Down Expand Up @@ -108,6 +111,11 @@
"Your conversation is saved; this turn was not retried automatically. "
"You can continue with a follow-up in this session."
)
WORKER_EXITED_MESSAGE = (
"The LoopX Chat execution worker exited before this turn finished. "
"Your conversation is saved; execution may have occurred. "
"This turn was not retried automatically."
)


class ChatTurnAcceptanceUnavailableError(Exception):
Expand Down Expand Up @@ -1420,6 +1428,13 @@ def _drain_session_queue(
return
active_turn_id = str(session.get("active_turn_id") or "")
if active_turn_id:
try:
self.recover_exited_turn(session_id, active_turn_id)
except OSError:
# Storage can still be unavailable. Retry only the
# non-executing lifecycle repair, never the old input.
self.closed.wait(0.25)
continue
with self.lock:
attached = self.adapters.get(session_id)
if attached is None or not attached.healthcheck():
Expand Down Expand Up @@ -1474,14 +1489,20 @@ def _drain_session_queue(
done_event = threading.Event()
with self.lock:
self.turn_done_events[(session_id, turn_id)] = done_event
self._run_turn(
session_id=session_id,
turn_id=turn_id,
message=str(turn.get("message") or ""),
attachments=turn.get("attachments") or [],
adapter=adapter,
done_event=done_event,
)
try:
self._run_turn(
session_id=session_id,
turn_id=turn_id,
message=str(turn.get("message") or ""),
attachments=turn.get("attachments") or [],
adapter=adapter,
done_event=done_event,
)
except OSError:
# _run_turn has released its execution fence/done event.
# A failed error checkpoint must not kill this queue worker;
# the next iteration reconciles the original durable state.
self.closed.wait(0.25)
finally:
with self.lock:
worker = self.session_queue_threads.get(session_id)
Expand Down Expand Up @@ -1527,10 +1548,11 @@ def _fail_queue_preparation(
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.
"""Reconcile a lost local execution 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.
The kernel fence spans the worker, while the PID spans the service.
A failed terminal write can end one worker without ending that PID.
Unknown/foreign records and a held execution fence never prove loss.
"""
session = self.store.load_session(session_id)
if not session:
Expand All @@ -1539,84 +1561,47 @@ def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> No
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:
state, observed = lock_holder_liveness(path)
if (state not in {LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED}
or observed.get("host") != lock_holder_host_label()
or type(observed.get("pid")) is not int or observed["pid"] <= 0
or observed.get("lock_id") != _lock_id(path)
or observed.get("agent_id") != "loopx-chat"
or observed.get("operation") != "execute_chat_turn"
or observed.get("policy") != "single_flight"):
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:
state, current = lock_holder_liveness(path)
if (ownership is None or current != observed
or state not in {LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED}
or current.get("host") != lock_holder_host_label()):
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,
):
code, message = (("server_restarted", SERVER_RESTARTED_MESSAGE)
if state == LOCK_HOLDER_DEAD else ("runtime_error", WORKER_EXITED_MESSAGE))
self._fail_turn(
session_id, turn_id, "server_restarted", SERVER_RESTARTED_MESSAGE,
session_id, turn_id, code, 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(
self.store.complete_recovered_terminalization(
session_id, turn_id, session, turn,
recovery_messages={"server_restarted": SERVER_RESTARTED_MESSAGE,
"runtime_error": WORKER_EXITED_MESSAGE},
)

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,
*,
Expand Down
34 changes: 6 additions & 28 deletions loopx/chat_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,7 @@ def _replace_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
class ChatSessionStore(ChatIngressStore):
"""Filesystem store kept outside project and public LoopX run history."""

def __init__(self, runtime_root: Path) -> None:
def __init__(self, runtime_root: Path, *, startup_maintenance: bool = True) -> None:
self.root = runtime_root.expanduser().resolve() / "chat"
self.sessions_root = self.root / "sessions"
self._session_lock_guard = threading.Lock()
Expand All @@ -220,8 +220,11 @@ def __init__(self, runtime_root: Path) -> None:
self.sessions_root.mkdir(parents=True, exist_ok=True, mode=0o700)
os.chmod(self.root, 0o700)
os.chmod(self.sessions_root, 0o700)
self.compact_completed_events()
self._recover_turn_completions()
# A scoped lifecycle repair must not compact unrelated history or finish
# other pending completions merely by opening their data owner.
if startup_maintenance:
self.compact_completed_events()
self._recover_turn_completions()

def _session_dir(self, session_id: str) -> Path:
token = _opaque_id(session_id, field="session_id")
Expand Down Expand Up @@ -1199,31 +1202,6 @@ def accept_managed_turn(
dispatch_reason=plan.dispatch_reason,
)

def queued_turns(self, session_id: str) -> list[dict[str, Any]]:
session = self.load_session(session_id)
if session is None:
raise KeyError("chat session was not found")
active_turn_id = str(session.get("active_turn_id") or "")
turns_dir = self._session_dir(session_id) / "turns"
rows: list[dict[str, Any]] = []
for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
if path.name.endswith(".events.json"):
continue
payload = _read_json(path)
if (
payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
and payload.get("status") == "queued"
and str(payload.get("turn_id") or "") != active_turn_id
):
rows.append(payload)
return sorted(
rows,
key=lambda item: (
str(item.get("created_at") or ""),
str(item.get("turn_id") or ""),
),
)

def _settle_expired_queued_turns(
self,
session_id: str,
Expand Down
17 changes: 14 additions & 3 deletions loopx/runtime_activation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,12 +130,23 @@ def managed_chat_activation_guard(
with ExitStack() as locks:
for root in roots:
locks.enter_context(chat_runtime_activation_lock(root, timeout_seconds=2.0))
for root in roots:
if chat_runtime_activation_state(root)["nonterminal_turns"]:
raise RuntimeActivationDeferred("chat_work_pending", labels)
for plist, (label, pid) in (expected_services or {}).items():
if plist not in plists or _managed_service_pid(plist, label=label) != pid:
raise ValueError("chat_service_route_unresolved")
for root in roots:
if chat_runtime_activation_state(root)["nonterminal_turns"]:
from .chat_runtime import ChatRuntimeController
from .chat_store import ChatSessionStore

# Data/lifecycle recovery only: this controller opens no adapter,
# starts no queue worker and never resends accepted input. Keep
# unrelated completion/compaction out of this scoped repair.
store = ChatSessionStore(root, startup_maintenance=False)
recovery = ChatRuntimeController(store=store, codex_bin="codex")
for session in store.list_sessions():
recovery.recover_exited_turn(str(session["session_id"]))
if chat_runtime_activation_state(root)["nonterminal_turns"]:
raise RuntimeActivationDeferred("chat_work_pending", labels)
yield


Expand Down
Loading
Loading