#!/usr/bin/env python3 """Long-lived controller for one root Codex app-server host and thread.""" from __future__ import annotations import contextlib import hashlib import secrets import signal import sys import threading import time from collections.abc import Mapping from pathlib import Path from typing import Any from mmo_app_server import ( APP_SERVER_INITIALIZE_TIMEOUT_SECONDS, APP_SERVER_LIFECYCLE_TIMEOUT_SECONDS, APP_SERVER_RECOVERY_DELAYS_SECONDS, AppServerClient, AppServerError, PersistentThreadHost, app_server_listen_command, bounded_goal_objective, completed_turn_presentable_text, last_agent_message, normalize_turn_failure, require_app_server_codex_version, retain_partial_evidence, serve_control_socket, turn_input, validate_server_request_response, ) from mmo_codex_home import session_environment from mmo_snapshot import load_snapshot from mmo_state import ( append_audit, load_session_capabilities, publish_session_record, read_session_record, revoke_session_capabilities, runtime_lock_path, session_dir, terminate_recorded_process_group, ) from mmo_util import ( atomic_write_json, atomic_write_text, file_lock, package_version, process_matches, process_start_token, utc_now, ) from mmo_version import SWITCHYARD_MCP_NAMESPACE_BRIDGE_VERSION _SHUTDOWN = threading.Event() def _signal_handler(_signum: int, _frame: object) -> None: _SHUTDOWN.set() class RootRunner: """Own one durable root host; operator clients may attach and detach freely.""" def __init__(self, session_id: str) -> None: self.directory = session_dir(session_id) self.session_id = session_id self.session = read_session_record(self.directory) self.snapshot = load_snapshot(str(self.session["snapshot_hash"])) self.agent_id = str(self.session["root_agent"]) self.agent = self.snapshot["resolved"]["agents"][self.agent_id] self.events_path = self.directory / "root-events.jsonl" self.stderr_path = self.directory / "root-stderr.log" self.result_path = self.directory / "root-result.md" self.partial_path = self.directory / "root-partial-result.md" self.socket_path = Path(str(self.session["root_app_server_socket"])) self.control_path = Path(str(self.session["root_control_socket"])) self.state_lock = threading.RLock() self.state: dict[str, Any] = { "thread_id": self.session.get("root_thread_id"), "active_turn_id": self.session.get("active_root_turn_id"), "last_turn_id": self.session.get("root_last_turn_id"), "turn_start_pending": bool(self.session.get("root_turn_start_pending", False)), "control_revision": int(self.session.get("root_control_revision", 0)), "finalize_requested": bool(self.session.get("root_finalizing", False)), } self.host = PersistentThreadHost( state=self.state, state_lock=self.state_lock, on_state_change=self._on_state_change, ) self.client: AppServerClient | None = None self.stop_status: str | None = None self.pause_exit = False self.retryable_failure_detach_exit = False self.control_stop = threading.Event() self.last_progress = time.monotonic() self.stall_reported = False self.lifecycle_timeout = float( self.session.get( "root_app_server_lifecycle_timeout_seconds", APP_SERVER_LIFECYCLE_TIMEOUT_SECONDS, ) ) self.finalize_turn_id: str | None = None self.finalize_deadline: float | None = None self.finalize_status = "completed" self.bootstrap_goal_pending = bool(self.session.get("root_goal_bootstrap_pending", False)) def _has_durable_work(self) -> bool: return bool( self.host.active_turn_id or self.host.last_turn_id or self.state.get("turn_start_pending") is True or isinstance(self.host.completed_turn, Mapping) or isinstance(self.state.get("goal"), Mapping) ) def _attached_root_client(self) -> bool: """Return whether the recorded foreground client still owns this session.""" current = read_session_record(self.directory) return process_matches( current.get("root_client_pid"), current.get("root_client_start_token"), ) def _interactive_goal_objective(self) -> str: return bounded_goal_objective( "Continue the operator's ongoing work in this persistent Codex MMO session. " "Treat each accepted interactive message as guidance within that ongoing project. " "Keep ownership of the critical path and integrate delegated evidence. Do not mark " "the goal complete merely because one turn ends; complete it only when the overall " "objective is genuinely achieved." ) def _eligible_root_thread(self, thread: Mapping[str, Any]) -> bool: """Return whether one Codex thread is a persistent top-level root context.""" thread_id = thread.get("id") cwd = thread.get("cwd") return bool( isinstance(thread_id, str) and thread_id and thread.get("ephemeral") is False and thread.get("parentThreadId") is None and thread.get("forkedFromId") is None and thread.get("agentRole") is None and thread.get("agentNickname") is None and isinstance(cwd, str) and Path(cwd).resolve() == Path(str(self.session["cwd"])).resolve() ) @staticmethod def _thread_sort_key(thread: Mapping[str, Any]) -> tuple[float, str]: created = thread.get("createdAt") timestamp = ( float(created) if isinstance(created, (int, float)) and not isinstance(created, bool) else 0.0 ) return timestamp, str(thread.get("id") or "") @staticmethod def _thread_projection(thread: Mapping[str, Any]) -> dict[str, Any]: turns = [value for value in thread.get("turns", []) if isinstance(value, Mapping)] active = next( (value for value in reversed(turns) if value.get("status") == "inProgress"), None, ) last = turns[-1] if turns else None goal = thread.get("goal") projection: dict[str, Any] = { "root_thread_status": thread.get("status"), "active_root_turn_id": (active.get("id") if isinstance(active, Mapping) else None), "root_last_turn_id": last.get("id") if isinstance(last, Mapping) else None, "root_turn_start_pending": False, "root_pending_request_count": 0, "root_token_usage": None, "root_last_item_id": None, "root_last_item_type": None, "root_last_observability_event": None, } if isinstance(goal, Mapping): projection.update( root_goal_status=goal.get("status"), root_goal_objective=goal.get("objective"), root_goal_token_budget=goal.get("tokenBudget"), root_goal_tokens_used=goal.get("tokensUsed", 0), root_goal_time_used_seconds=goal.get("timeUsedSeconds", 0), ) else: projection.update( root_goal_status=None, root_goal_objective=None, root_goal_token_budget=None, root_goal_tokens_used=0, root_goal_time_used_seconds=0, ) return projection def _commit_root_thread( self, thread: Mapping[str, Any], *, reason: str, require_attached_client: bool, ) -> bool: """Atomically make one top-level Codex thread the current logical root.""" if not self._eligible_root_thread(thread): return False thread_id = str(thread["id"]) now = utc_now() with file_lock(runtime_lock_path()): current = read_session_record(self.directory) if current.get("status") not in { "starting", "running", "detached", "paused", "suspended", }: return False current_id = current.get("root_thread_id") if current_id == thread_id: current.update(self._thread_projection(thread)) current["last_active_at"] = now publish_session_record(self.directory, current, mirror_run=True) self.session = current return False lineage = [dict(value) for value in current.get("root_thread_lineage", [])] if thread_id in {value.get("thread_id") for value in lineage}: return False if require_attached_client: with self.state_lock: host_thread_id = self.state.get("thread_id") host_active_turn_id = self.state.get("active_turn_id") host_turn_start_pending = bool(self.state.get("turn_start_pending")) if ( not isinstance(current_id, str) or current_id != host_thread_id or host_active_turn_id is not None or host_turn_start_pending or not process_matches( current.get("root_client_pid"), current.get("root_client_start_token") ) ): return False generation = int(current.get("root_thread_generation", 0)) + 1 needs_interactive_goal = bool( reason != "initial" and current.get("session_kind") == "interactive" and self.agent["execution_mode"] == "goal" and not isinstance(thread.get("goal"), Mapping) ) transition = { "from_thread_id": current_id, "to_thread_id": thread_id, "generation": generation, "observed_at": now, "reason": reason, } current["root_thread_transition"] = transition publish_session_record(self.directory, current, mirror_run=True) if lineage: lineage[-1]["superseded_at"] = now lineage[-1]["successor_thread_id"] = thread_id lineage.append( { "generation": generation, "thread_id": thread_id, "codex_session_id": str(thread.get("sessionId") or thread_id), "adopted_at": now, "reason": reason, "created_at": thread.get("createdAt"), } ) current.update( root_thread_id=thread_id, root_thread_generation=generation, root_thread_lineage=lineage, root_thread_transition=None, root_control_revision=int(current.get("root_control_revision", 0)) + 1, root_last_progress_at=now, last_active_at=now, root_goal_bootstrap_pending=needs_interactive_goal, **self._thread_projection(thread), ) publish_session_record(self.directory, current, mirror_run=True) with self.state_lock: self.state["control_revision"] = int(current["root_control_revision"]) self.session = current self.host.adopt_thread(thread) self.last_progress = time.monotonic() self.stall_reported = False if needs_interactive_goal: self.bootstrap_goal_pending = True append_audit( self.session_id, "root_thread_adopted", previous_thread_id=current_id, root_thread_id=thread_id, generation=generation, reason=reason, ) return True def _record_opened_thread(self, thread: Mapping[str, Any], *, mode: str) -> None: current = read_session_record(self.directory) if current.get("root_thread_id") is None: if not self._commit_root_thread( thread, reason="initial", require_attached_client=False, ): raise AppServerError("initial root thread could not be recorded") return if current.get("root_thread_id") != thread.get("id"): raise AppServerError("opened root thread disagrees with canonical session state") self._commit_root_thread( thread, reason="resume" if mode == "resume" else "initial", require_attached_client=False, ) def _list_root_threads(self, client: AppServerClient) -> list[dict[str, Any]]: cursor: str | None = None seen: set[str] = set() threads: list[dict[str, Any]] = [] while True: params: dict[str, Any] = { "archived": False, "limit": 100, "sortDirection": "asc", "sortKey": "created_at", "useStateDbOnly": True, } if cursor is not None: params["cursor"] = cursor response = client.request("thread/list", params, timeout=self.lifecycle_timeout) data = response.get("data") if isinstance(response, Mapping) else None if not isinstance(data, list) or not all(isinstance(value, Mapping) for value in data): raise AppServerError("thread/list returned invalid root-thread data") threads.extend(dict(value) for value in data if self._eligible_root_thread(value)) next_cursor = response.get("nextCursor") if next_cursor is None: return sorted(threads, key=self._thread_sort_key) if not isinstance(next_cursor, str) or next_cursor in seen: raise AppServerError("thread/list returned an invalid root-thread cursor") seen.add(next_cursor) cursor = next_cursor def _read_root_thread( self, client: AppServerClient, summary: Mapping[str, Any], ) -> dict[str, Any]: """Resolve one list summary to authoritative turn and goal state.""" thread_id = str(summary["id"]) response = client.request( "thread/read", {"threadId": thread_id, "includeTurns": True}, timeout=self.lifecycle_timeout, ) thread = response.get("thread") if isinstance(response, Mapping) else None if ( not isinstance(thread, Mapping) or thread.get("id") != thread_id or not self._eligible_root_thread(thread) ): raise AppServerError("thread/read returned an invalid root successor") return dict(thread) def _recover_root_successors(self, client: AppServerClient) -> None: current_id = self.session.get("root_thread_id") if not isinstance(current_id, str): return threads = self._list_root_threads(client) by_id = {str(value["id"]): value for value in threads} current_thread = by_id.get(current_id) if current_thread is not None: current_key = self._thread_sort_key(current_thread) else: lineage = self.session.get("root_thread_lineage", []) current_row = next( ( value for value in reversed(lineage) if isinstance(value, Mapping) and value.get("thread_id") == current_id ), None, ) created_at = current_row.get("created_at") if isinstance(current_row, Mapping) else None if not isinstance(created_at, (int, float)) or isinstance(created_at, bool): # State-db indexing may briefly lag the rollout that remains # directly resumable by identity. Never trade that canonical # path for an inferred successor without an ordering anchor. return current_key = float(created_at), current_id transition = self.session.get("root_thread_transition") transition_id = transition.get("to_thread_id") if isinstance(transition, Mapping) else None if isinstance(transition_id, str): # A staged transition is stronger evidence than timestamp ordering. # Never skip or overwrite it merely because the state-db index has # not exposed its target yet. if transition_id not in by_id: return self._commit_root_thread( self._read_root_thread(client, by_id[transition_id]), reason="crash_recovery", require_attached_client=False, ) current_key = self._thread_sort_key(by_id[transition_id]) successors = [value for value in threads if self._thread_sort_key(value) > current_key] for successor in successors: self._commit_root_thread( self._read_root_thread(client, successor), reason="crash_recovery", require_attached_client=False, ) def recoverable_descendants(self) -> list[dict[str, Any]]: """Return durable descendants that still require a live session.""" # Imported lazily to keep the standalone runner's transport boundary # acyclic during module initialization. from mmo_runtime import iter_jobs from mmo_state import RECOVERABLE_JOB_STATUSES run_id = self.session.get("current_run_id") return [ job for job in iter_jobs(strict=True) if job.get("session_id") == self.session_id and job.get("run_id") == run_id and job.get("status") in RECOVERABLE_JOB_STATUSES ] def refresh_native_runs(self) -> list[dict[str, Any]]: """Project every native descendant thread into stable MMO run identity.""" client = self.client root_thread_id = self.host.thread_id if client is None or root_thread_id is None: return [] cursor: str | None = None threads: list[dict[str, Any]] = [] seen_cursors: set[str] = set() while True: params: dict[str, Any] = { "ancestorThreadId": root_thread_id, "archived": False, "limit": 100, "sortDirection": "asc", "sortKey": "created_at", "sourceKinds": [ "subAgent", "subAgentReview", "subAgentCompact", "subAgentThreadSpawn", "subAgentOther", ], "useStateDbOnly": True, } if cursor is not None: params["cursor"] = cursor response = client.request("thread/list", params) data = response.get("data") if isinstance(response, Mapping) else None if not isinstance(data, list) or not all(isinstance(item, Mapping) for item in data): raise AppServerError("thread/list returned invalid native-agent data") threads.extend(dict(item) for item in data) next_cursor = response.get("nextCursor") if next_cursor is None: break if not isinstance(next_cursor, str) or next_cursor in seen_cursors: raise AppServerError("thread/list returned an invalid cursor sequence") seen_cursors.add(next_cursor) cursor = next_cursor native_names = { str(agent.get("native_name")): agent_id for agent_id, agent in self.snapshot["resolved"]["agents"].items() if "native" in agent.get("backends", []) } existing = self.session.get("root_native_runs", {}) existing_by_thread = ( { str(value.get("thread_id")): dict(value) for value in existing.values() if isinstance(value, Mapping) and isinstance(value.get("thread_id"), str) } if isinstance(existing, Mapping) else {} ) projected: dict[str, dict[str, Any]] = {} projected_thread_ids: set[str] = set() for thread in threads: thread_id = thread.get("id") role = native_names.get(str(thread.get("agentRole"))) if not isinstance(thread_id, str) or role is None: continue prior = existing_by_thread.get(thread_id, {}) run_ref = prior.get("agent_run_ref") if not isinstance(run_ref, str) or not run_ref.startswith("ar_"): run_ref = "ar_" + secrets.token_urlsafe(24) reported_status = thread.get("status") reported_status_type = ( reported_status.get("type") if isinstance(reported_status, Mapping) else None ) row = { "agent_run_ref": run_ref, "agent": role, "backend": "native", "thread_id": thread_id, "parent_thread_id": thread.get("parentThreadId"), "nickname": thread.get("agentNickname"), "status": ( prior.get("status") if prior.get("status") in {"paused", "stopped"} or (prior.get("status") == "detached" and reported_status_type == "active") else reported_status ), "can_accept_direct_input": thread.get("canAcceptDirectInput"), "created_at": thread.get("createdAt"), "updated_at": thread.get("updatedAt"), "control_revision": int(prior.get("control_revision", 0)), } projected[run_ref] = row projected_thread_ids.add(thread_id) for run_ref, row in existing.items() if isinstance(existing, Mapping) else (): if ( isinstance(row, Mapping) and row.get("status") == "stopped" and row.get("thread_id") not in projected_thread_ids ): projected[str(run_ref)] = dict(row) self._update(root_native_runs=projected) return list(projected.values()) def _update(self, **changes: Any) -> dict[str, Any]: with file_lock(runtime_lock_path()): current = read_session_record(self.directory) if current.get("status") in {"finishing", "stopping", "cancelling"} and changes.get( "status" ) not in {"completed", "stopped", "failed", "cancelled"}: changes.pop("status", None) if ( current.get("status") in {"completed", "stopped", "failed", "cancelled"} and "status" in changes and changes.get("status") != current.get("status") ): self.session = current return current current.update(changes) publish_session_record(self.directory, current, mirror_run=True) self.session = current return current def _on_state_change( self, changes: dict[str, Any], message: Mapping[str, Any], ) -> None: started = changes.get("thread_started") if isinstance(started, Mapping): adopted = self._commit_root_thread( started, reason="fresh_context", require_attached_client=True, ) if not adopted: current_id = read_session_record(self.directory).get("root_thread_id") candidate_id = started.get("id") if isinstance(current_id, str) and candidate_id != current_id: append_audit( self.session_id, "root_thread_candidate_ignored", root_thread_id=current_id, candidate_thread_id=candidate_id, ) return now = utc_now() persisted: dict[str, Any] = {"root_last_progress_at": now, "last_active_at": now} self.last_progress = time.monotonic() self.stall_reported = False if "active_turn_id" in changes: persisted["active_root_turn_id"] = changes["active_turn_id"] if "last_turn_id" in changes: persisted["root_last_turn_id"] = changes["last_turn_id"] if "turn_start_pending" in changes: persisted["root_turn_start_pending"] = bool(changes["turn_start_pending"]) if "pending_request_count" in changes: persisted["root_pending_request_count"] = changes["pending_request_count"] if "thread_status" in changes: persisted["root_thread_status"] = changes["thread_status"] if "token_usage" in changes: persisted["root_token_usage"] = changes["token_usage"] if "turn_failure" in changes: persisted["failure"] = changes["turn_failure"] if changes["turn_failure"] is None: persisted["error"] = None goal = changes.get("goal") if isinstance(goal, Mapping): persisted.update( root_goal_status=goal.get("status"), root_goal_objective=goal.get("objective"), root_goal_token_budget=goal.get("tokenBudget"), root_goal_tokens_used=goal.get("tokensUsed", 0), root_goal_time_used_seconds=goal.get("timeUsedSeconds", 0), ) goal_status = goal.get("status") try: session_status = str(read_session_record(self.directory).get("status")) except Exception: # Observability callbacks must not kill the transport reader if # a concurrent atomic session replacement is briefly unreadable. session_status = str(self.session.get("status")) transition_active = session_status in {"finishing", "stopping", "cancelling"} if ( goal_status == "active" and session_status not in {"detached", "paused", "suspended"} and not transition_active ): persisted["status"] = "running" elif ( goal_status == "paused" and session_status not in {"detached", "suspended"} and not self.bootstrap_goal_pending and not transition_active ): persisted["status"] = "paused" elif ( goal_status == "blocked" and session_status != "detached" and not transition_active and not isinstance(self.state.get("turn_failure"), Mapping) ): persisted["status"] = "paused" elif ( goal_status in {"usageLimited", "budgetLimited"} and session_status != "detached" and not transition_active ): persisted["status"] = "suspended" if isinstance(changes.get("active_turn_id"), str): try: session_status = str(read_session_record(self.directory).get("status")) except Exception: session_status = str(self.session.get("status")) if session_status not in { "detached", "paused", "suspended", "finishing", "stopping", "cancelling", }: persisted["status"] = "running" if "last_item" in changes: item = changes["last_item"] if isinstance(item, Mapping): persisted["root_last_item_id"] = item.get("id") persisted["root_last_item_type"] = item.get("type") if "last_observability_event" in changes: persisted["root_last_observability_event"] = changes["last_observability_event"] method = message.get("method") if method in {"mmo/turn/reset", "mmo/turn/accepted"}: # These synthetic transitions participate in exact crash # reconciliation. A turn must not be sent, or reported accepted, # unless its recovery marker is durable. self._update(**persisted) else: with contextlib.suppress(Exception): self._update(**persisted) if isinstance(method, str) and method in { "thread/goal/updated", "thread/status/changed", "model/rerouted", "model/verification", "warning", "error", "account/rateLimits/updated", }: with contextlib.suppress(Exception): append_audit(self.session_id, "root_app_server_event", method=method) def _publish_app_server(self, pid: int) -> None: token = process_start_token(pid) if token is None: raise AppServerError("root app-server process could not be fingerprinted") with file_lock(runtime_lock_path()): current = read_session_record(self.directory) if current.get("status") not in { "starting", "running", "detached", "paused", "suspended", }: raise AppServerError("root admission closed during app-server bootstrap") current.update( root_app_server_pid=pid, root_app_server_pgid=pid, root_app_server_start_token=token, root_app_server_process_group_isolated=True, ) publish_session_record(self.directory, current, mirror_run=True) self.session = current def _new_client(self, *, start_host: bool) -> AppServerClient: environment = session_environment(self.session, self.agent_id) command = None if start_host: flags = self.session["homes"][self.agent_id].get("command_flags", []) command = app_server_listen_command( str(self.session["codex_binary"]), flags, self.socket_path, ) return AppServerClient( socket_path=self.socket_path, command=command, cwd=Path(str(self.session["cwd"])), env=environment, events_path=self.events_path, stderr_path=self.stderr_path, approval_policy=str(self.agent["approval_policy"]), on_message=self.host.on_message, ) def connect(self) -> None: pid = self.session.get("root_app_server_pid") token = self.session.get("root_app_server_start_token") host_alive = process_matches(pid, token) if not host_alive: # The executable at a pinned path can be replaced while a durable # session is detached. Every replacement host must re-check the # reviewed protocol release before it can touch the saved thread. require_app_server_codex_version(str(self.session["codex_binary"])) if not host_alive and (self.socket_path.exists() or self.socket_path.is_symlink()): if self.socket_path.is_symlink() or not self.socket_path.is_socket(): raise AppServerError("recorded root app-server socket is unsafe") self.socket_path.unlink() client = self._new_client(start_host=not host_alive) try: client.start( timeout=APP_SERVER_INITIALIZE_TIMEOUT_SECONDS, on_started=self._publish_app_server if not host_alive else None, ) self.host.attach_client(client) self.client = client self._recover_root_successors(client) self.session = read_session_record(self.directory) dynamic_tools: list[dict[str, Any]] = [] if self.agent["driver"] == "switchyard": observed_switchyard = self.session.get("switchyard_version") if not isinstance(observed_switchyard, str): raise AppServerError("Switchyard session has no pinned gateway version") if observed_switchyard == SWITCHYARD_MCP_NAMESPACE_BRIDGE_VERSION: dynamic_tools = client.install_switchyard_mcp_bridge( timeout=self.lifecycle_timeout ) mode = "resume" if isinstance(self.session.get("root_thread_id"), str) else "start" params: dict[str, Any] if mode == "resume": params = { "threadId": self.session["root_thread_id"], "cwd": self.session["cwd"], "sandbox": self.session.get("root_sandbox_mode", self.agent["permissions"]), "approvalPolicy": self.agent["approval_policy"], "excludeTurns": False, } else: params = { "cwd": self.session["cwd"], "sandbox": self.session.get("root_sandbox_mode", self.agent["permissions"]), "approvalPolicy": self.agent["approval_policy"], "allowProviderModelFallback": False, "ephemeral": False, "historyMode": "paginated", } if dynamic_tools: params["dynamicTools"] = dynamic_tools prior_turn_id = self.state.get("last_turn_id") or self.state.get("active_turn_id") thread = self.host.open_thread( mode, params, timeout=self.lifecycle_timeout, prior_turn_id=prior_turn_id, expected_thread_id=( str(self.session["root_thread_id"]) if mode == "resume" else None ), ) self._record_opened_thread(thread, mode=mode) self._update( active_root_turn_id=self.state.get("active_turn_id"), root_last_turn_id=self.state.get("last_turn_id"), root_turn_start_pending=bool(self.state.get("turn_start_pending")), root_app_server_protocol="codex-app-server-v2-unix", root_recovery_error=None, ) except BaseException: if client.process is not None: client.stop_host() else: client.close() raise def start_initial_work(self) -> None: prompt = self.session.get("root_initial_prompt") if not isinstance(prompt, str) or not prompt.strip(): if ( self.agent["execution_mode"] == "goal" and self.session.get("session_kind") == "interactive" and not isinstance(self.state.get("goal"), Mapping) ): self.bootstrap_goal_pending = True goal = self.host.set_goal( objective=self._interactive_goal_objective(), status="paused", token_budget=int(self.agent["goal_token_budget"]), timeout=self.lifecycle_timeout, ) self._update( root_goal_status=goal["status"], root_goal_objective=goal["objective"], root_goal_token_budget=goal.get("tokenBudget"), root_goal_bootstrap_pending=True, ) self._resume_turn_work_if_needed(str(self.session.get("status"))) return attachments = [ str(item) for item in self.session.get("root_initial_attachments", []) if isinstance(item, str) ] if self.agent["execution_mode"] == "goal": goal = self.host.set_goal( objective=bounded_goal_objective( str(self.session.get("root_goal_objective") or prompt) ), status="paused", token_budget=int(self.agent["goal_token_budget"]), timeout=self.lifecycle_timeout, ) if self.host.active_turn_id is None and self.host.completed_turn is None: self.host.start_turn( turn_input(prompt.strip(), attachments), effort=self.agent.get("reasoning"), ) goal = self.host.set_goal( status="active", timeout=self.lifecycle_timeout, ) self._update( status="running", root_goal_status=goal["status"], root_goal_objective=goal["objective"], root_goal_token_budget=goal.get("tokenBudget"), ) else: if self.host.active_turn_id is None and self.host.completed_turn is None: self.host.start_turn( turn_input(prompt.strip(), attachments), effort=self.agent.get("reasoning"), ) self._update(status="running") self._update(root_initial_prompt=None, root_initial_attachments=[]) def _activate_bootstrap_goal(self) -> None: if not self.bootstrap_goal_pending or self.host.active_turn_id is None: return current = read_session_record(self.directory) if current.get("status") in { "detached", "paused", "suspended", "finishing", "stopping", "cancelling", }: return goal = self.state.get("goal") if not isinstance(goal, Mapping): return if goal.get("status") == "paused": goal = self.host.set_goal(status="active", timeout=self.lifecycle_timeout) self.bootstrap_goal_pending = False self._update( status="running", root_goal_status=goal.get("status"), root_goal_bootstrap_pending=False, ) def _ensure_bootstrap_goal(self) -> None: """Restore the ongoing interactive objective after adopting a fresh root thread.""" if not self.bootstrap_goal_pending or isinstance(self.state.get("goal"), Mapping): return current = read_session_record(self.directory) if current.get("status") in { "finishing", "stopping", "cancelling", "completed", "stopped", "failed", "cancelled", }: return goal = self.host.set_goal( objective=self._interactive_goal_objective(), status="paused", token_budget=int(self.agent["goal_token_budget"]), timeout=self.lifecycle_timeout, ) self._update( root_goal_status=goal["status"], root_goal_objective=goal["objective"], root_goal_token_budget=goal.get("tokenBudget"), root_goal_bootstrap_pending=True, ) def _resume_turn_work_if_needed(self, lifecycle_status: str) -> None: """Continue an unfinished turn after host recovery without duplicating work.""" if ( self.agent["execution_mode"] != "turn" or lifecycle_status == "paused" or self.host.active_turn_id is not None or self.host.completed_turn is not None ): return self.host.start_turn( turn_input( "Continue the original root task from this persisted app-server thread after " "transport recovery. Use retained evidence and complete the original request.", [], ), effort=self.agent.get("reasoning"), ) def _claim_revision(self, request: Mapping[str, Any], target_thread_id: str) -> int: expected = request.get("expected_revision") root_thread_id = self.host.thread_id if target_thread_id == root_thread_id: with self.state_lock: current = int(self.state.get("control_revision", 0)) if ( not isinstance(expected, int) or isinstance(expected, bool) or expected != current ): raise RuntimeError(f"control revision conflict: expected {current}") current += 1 self.state["control_revision"] = current self._update(root_control_revision=current) return current runs = self.session.get("root_native_runs", {}) if not isinstance(runs, Mapping): raise RuntimeError("native-agent run registry is unavailable") selected = next( ( (run_ref, dict(row)) for run_ref, row in runs.items() if isinstance(row, Mapping) and row.get("thread_id") == target_thread_id ), None, ) if selected is None: raise RuntimeError("native-agent run is unavailable") run_ref, row = selected current = int(row.get("control_revision", 0)) if not isinstance(expected, int) or isinstance(expected, bool) or expected != current: raise RuntimeError(f"control revision conflict: expected {current}") current += 1 row["control_revision"] = current updated = dict(runs) updated[str(run_ref)] = row self._update(root_native_runs=updated) return current def _native_row(self, target_thread_id: str) -> tuple[str, dict[str, Any]]: runs = self.session.get("root_native_runs", {}) if isinstance(runs, Mapping): for run_ref, row in runs.items(): if isinstance(row, Mapping) and row.get("thread_id") == target_thread_id: return str(run_ref), dict(row) raise RuntimeError("native-agent run is unavailable") def _update_native_row(self, target_thread_id: str, **changes: Any) -> dict[str, Any]: run_ref, row = self._native_row(target_thread_id) row.update(changes) runs = dict(self.session.get("root_native_runs", {})) runs[run_ref] = row self._update(root_native_runs=runs) return row @staticmethod def _start_native_turn( client: AppServerClient, thread_id: str, prompt: str, effort: Any, ) -> str: response = client.request( "turn/start", { "threadId": thread_id, "input": turn_input(prompt, []), "effort": effort, }, ) turn = response.get("turn") if isinstance(response, Mapping) else None if not isinstance(turn, Mapping) or not isinstance(turn.get("id"), str): raise RuntimeError("native turn/start returned no turn") return str(turn["id"]) def _execute_control(self, request: Mapping[str, Any]) -> Any: client = self.client if client is None: raise RuntimeError("root app-server client is unavailable") action = str(request.get("action", "")) mutating = action not in {"list", "inspect", "trace"} thread_id = self.host.thread_id if thread_id is None: raise RuntimeError("root app-server thread is unavailable") target_thread_id = str(request.get("target_thread_id") or thread_id) root_target = target_thread_id == thread_id if action == "list": return {"agents": self.refresh_native_runs()} if action == "finalize" and root_target: terminal_status = str(request.get("terminal_status", "completed")) if terminal_status not in {"completed", "stopped"}: raise ValueError("root terminal_status must be completed or stopped") native_row = None if root_target else self._native_row(target_thread_id)[1] if root_target: target_agent_id = self.agent_id else: if native_row is None: raise RuntimeError("native-agent run is unavailable") target_agent_id = str(native_row["agent"]) target_agent = self.snapshot["resolved"]["agents"][target_agent_id] if root_target and action == "fork": raise RuntimeError("the immutable root run cannot be forked as another root") if ( not root_target and mutating and isinstance(native_row, Mapping) and native_row.get("status") == "stopped" ): raise RuntimeError("cannot control a stopped native run") revision = ( self._claim_revision(request, target_thread_id) if mutating else int( self.state.get("control_revision", 0) if root_target else (native_row or {}).get("control_revision", 0) ) ) thread_response = client.request( "thread/read", {"threadId": target_thread_id, "includeTurns": True}, ) target_thread = ( thread_response.get("thread") if isinstance(thread_response, Mapping) else None ) if not isinstance(target_thread, Mapping) or target_thread.get("id") != target_thread_id: raise RuntimeError("target app-server thread is unavailable") active = next( ( turn for turn in reversed(target_thread.get("turns", [])) if isinstance(turn, Mapping) and turn.get("status") == "inProgress" ), None, ) # The host owns the authoritative live root turn. Persisted thread # history is authoritative for native children, but may lag an active # root notification and must not cause a second root turn to start. turn_id = ( self.host.active_turn_id if root_target else active.get("id") if isinstance(active, Mapping) else None ) if action == "inspect": pending_requests = client.pending_server_requests_for_thread(target_thread_id) return { "thread_id": target_thread_id, "thread_status": target_thread.get("status"), "agent": target_agent_id, "backend": "root" if root_target else "native", "agent_run_ref": ( self.session.get("root_agent_run_ref") if root_target else (native_row or {}).get("agent_run_ref") ), "goal": self.state.get("goal") if root_target else None, "active_turn_id": turn_id, "pending_requests": pending_requests, "control_revision": revision, } if action == "trace": return { "events_path": str(self.events_path), "stderr_path": str(self.stderr_path), "result_path": str(self.result_path), } if action == "steer": if turn_id is None: raise RuntimeError("root has no active turn to steer") return client.request( "turn/steer", { "threadId": target_thread_id, "expectedTurnId": str(request.get("expected_turn_id") or turn_id), "input": turn_input(str(request["input"]), []), }, ) if action in {"interrupt", "pause"}: if action == "pause" and root_target and self.agent["execution_mode"] == "goal": self.host.set_goal(status="paused") result: Any = {"interrupted": False} if turn_id is not None: result = client.request( "turn/interrupt", {"threadId": target_thread_id, "turnId": turn_id}, ) if action == "pause": if root_target: self.bootstrap_goal_pending = False self.finalize_deadline = None self.finalize_turn_id = None self.finalize_status = "completed" with self.state_lock: self.state["finalize_requested"] = False self._update( status="paused", root_goal_status=( "paused" if self.agent["execution_mode"] == "goal" else None ), root_finalizing=False, root_goal_bootstrap_pending=False, ) if request.get("retire_host") is True: self.pause_exit = True _SHUTDOWN.set() else: self._update_native_row(target_thread_id, status="paused") elif target_agent["execution_mode"] == "turn": if root_target: self._update(status="paused") else: self._update_native_row(target_thread_id, status="paused") return result if action == "continue": text = str(request.get("input") or "Continue the active objective.") if root_target and self.agent["execution_mode"] == "goal": self.bootstrap_goal_pending = False token_budget = request.get("goal_token_budget") goal = self.host.set_goal( status="active", token_budget=( int(token_budget) if isinstance(token_budget, int) and not isinstance(token_budget, bool) else None ), ) self.finalize_deadline = None self.finalize_turn_id = None self.finalize_status = "completed" with self.state_lock: self.state["finalize_requested"] = False self._update( status="running", root_goal_status=goal["status"], root_finalizing=False, root_completion_deferred=False, root_completion_deferred_jobs=[], root_goal_bootstrap_pending=False, ) return {"goal": goal} if turn_id is not None: raise RuntimeError("root already has an active turn") if root_target: new_turn = self.host.start_turn( turn_input(text, []), effort=self.agent.get("reasoning"), ) else: response = client.request( "turn/start", { "threadId": target_thread_id, "input": turn_input(text, []), "effort": target_agent.get("reasoning"), }, ) turn = response.get("turn") if isinstance(response, Mapping) else None if not isinstance(turn, Mapping) or not isinstance(turn.get("id"), str): raise RuntimeError("native turn/start returned no turn") new_turn = str(turn["id"]) if root_target: self.finalize_deadline = None self.finalize_turn_id = None self.finalize_status = "completed" with self.state_lock: self.state["finalize_requested"] = False self._update( status="running", root_finalizing=False, root_completion_deferred=False, root_completion_deferred_jobs=[], ) else: self._update_native_row(target_thread_id, status="running") return {"turn_id": new_turn} if action == "detach": if root_target: current = read_session_record(self.directory) failure = current.get("failure") if ( isinstance(failure, Mapping) and failure.get("retryable") is True and self.host.active_turn_id is None ): now = utc_now() self.retryable_failure_detach_exit = True self._update( status="suspended", suspended_at=now, detached_at=now, last_active_at=now, ) _SHUTDOWN.set() return {"detached": True, "status": "suspended"} self._update(status="detached", detached_at=utc_now()) else: self._update_native_row( target_thread_id, status="detached", detached_at=utc_now(), ) return {"detached": True} if action == "stop": if root_target and self.agent["execution_mode"] == "goal": self.host.set_goal(status="paused") if turn_id is not None: with contextlib.suppress(AppServerError): client.request( "turn/interrupt", {"threadId": target_thread_id, "turnId": turn_id}, ) if root_target: self.stop_status = "stopped" _SHUTDOWN.set() else: client.request("thread/archive", {"threadId": target_thread_id}) self._update_native_row( target_thread_id, status="stopped", stopped_at=utc_now(), ) return {"stopping": True} if action == "finalize": if root_target and self.agent["execution_mode"] == "goal": self.host.set_goal(status="paused") prompt = str( request.get("input") or "Return the best supported final result from evidence already in this thread." ) if root_target: self.finalize_status = terminal_status self.finalize_deadline = time.monotonic() + int( self.agent["finalization_grace_seconds"] ) with self.state_lock: self.state["finalize_requested"] = True self._update( root_finalizing=True, root_finalization_started_at=utc_now(), ) if turn_id is not None: result = client.request( "turn/steer", { "threadId": target_thread_id, "expectedTurnId": turn_id, "input": turn_input(prompt, []), }, ) if root_target: self.finalize_turn_id = str(turn_id) return result new_turn_id = ( self.host.start_turn( turn_input(prompt, []), effort=self.agent.get("reasoning"), ) if root_target else self._start_native_turn( client, target_thread_id, prompt, target_agent.get("reasoning"), ) ) if root_target: self.finalize_turn_id = str(new_turn_id) return {"turn_id": new_turn_id} if action == "compact": return client.request("thread/compact/start", {"threadId": target_thread_id}) if action == "respond": request_id = request["request_id"] pending = next( ( item for item in client.pending_server_requests_for_thread(target_thread_id) if type(item.get("id")) is type(request_id) and item.get("id") == request_id ), None, ) if not isinstance(pending, Mapping) or not isinstance(pending.get("method"), str): raise RuntimeError("app-server request is not pending") response = request.get("response") if not isinstance(response, Mapping): raise ValueError("app-server response must be an object") validate_server_request_response(str(pending["method"]), response) client.respond(request_id, response) if root_target: self._update( root_pending_request_count=len( client.pending_server_requests_for_thread(target_thread_id) ) ) return {"responded": True} if action == "set_effort": effort = str(request["effort"]) if effort not in target_agent["allowed_reasoning_efforts"]: raise ValueError("reasoning effort is outside the profile grant") return client.request( "thread/settings/update", {"threadId": target_thread_id, "effort": effort}, ) if action == "fork": runs = self.refresh_native_runs() active_count = sum( isinstance(row.get("status"), Mapping) and row["status"].get("type") == "active" for row in runs ) native_limit = int( self.snapshot["resolved"]["coordination"]["native_max_concurrent_threads"] ) if active_count >= native_limit: raise RuntimeError( f"native fork would exceed the compiled thread limit ({native_limit})" ) response = client.request( "thread/fork", { "threadId": target_thread_id, "cwd": self.session["cwd"], "sandbox": target_agent["permissions"], "approvalPolicy": target_agent["approval_policy"], "ephemeral": False, "deferGoalContinuation": True, "excludeTurns": False, }, ) forked = response.get("thread") if isinstance(response, Mapping) else None if not isinstance(forked, Mapping) or not isinstance(forked.get("id"), str): raise RuntimeError("native thread/fork returned no thread") fork_thread_id = str(forked["id"]) fork_ref = "ar_" + secrets.token_urlsafe(24) fork_row = { "agent_run_ref": fork_ref, "agent": target_agent_id, "backend": "native", "thread_id": fork_thread_id, "parent_thread_id": forked.get("parentThreadId"), "nickname": forked.get("agentNickname"), "status": forked.get("status"), "can_accept_direct_input": forked.get("canAcceptDirectInput"), "created_at": forked.get("createdAt"), "updated_at": forked.get("updatedAt"), "control_revision": 0, } updated = dict(self.session.get("root_native_runs", {})) updated[fork_ref] = fork_row self._update(root_native_runs=updated) new_turn_id = self._start_native_turn( client, fork_thread_id, str(request["input"]), target_agent.get("reasoning"), ) return {"agent": fork_row, "turn_id": new_turn_id} raise ValueError(f"unsupported root control action: {action}") def serve_control(self) -> None: def publish_ready() -> None: self._update(root_control_socket_ready=True) serve_control_socket( self.control_path, stop_event=self.control_stop, handler=self._execute_control, on_ready=publish_ready, backlog=16, connection_timeout=None, ) def _result_from_turns(self, turns: list[dict[str, Any]]) -> str: for turn in reversed(turns): if turn.get("status") != "completed": continue turn_id = turn.get("id") if isinstance(turn_id, str): result = completed_turn_presentable_text(self.events_path, turn_id) if result.strip(): return result return last_agent_message(turn) return "" def _publish_terminal(self, status: str, *, error: str | None = None) -> str: if error is None and status == "failed" and isinstance(self.session.get("error"), str): error = str(self.session["error"]) turns: list[dict[str, Any]] = [] history_timeout = self.lifecycle_timeout if self.finalize_deadline is not None: history_timeout = min( history_timeout, max(0.0, self.finalize_deadline - time.monotonic()), ) if history_timeout > 0: with contextlib.suppress(AppServerError): turns = self.host.complete_history(timeout=history_timeout) atomic_write_json( self.directory / "root-terminal-history.json", { "thread_id": self.host.thread_id, "captured_at": utc_now(), "turns": turns, }, ) result = self._result_from_turns(turns) if status == "completed" and not result: status = "failed" error = error or "root completed without a readable terminal result" if status == "completed" and result: atomic_write_text(self.result_path, result, 0o600) result_kind = "final" elif self._has_durable_work(): partial = retain_partial_evidence( self.session, self.directory, reason=error or f"root host ended with status {status}", events_filename=self.events_path.name, result_filename=self.result_path.name, partial_filename=self.partial_path.name, title="Partial root result", ) result_kind = "partial" self._update(**partial) else: result_kind = "none" now = utc_now() current = self._update( status=status, finished_at=now, last_active_at=now, exit_code=0 if status == "completed" else 130 if status == "cancelled" else 1, error=error, result_path=( str(self.result_path) if status == "completed" and result else str(self.partial_path) if result_kind == "partial" else None ), result_kind=result_kind, root_finalizing=False, root_completion_deferred=False, root_completion_deferred_jobs=[], root_control_socket_ready=False, failure=None if status == "completed" else self.session.get("failure"), suspended_at=None if status == "completed" else self.session.get("suspended_at"), ) status = str(current["status"]) with file_lock(runtime_lock_path()): current = read_session_record(self.directory) if current.get("current_run_id"): current["last_run_id"] = current["current_run_id"] current["current_run_id"] = None publish_session_record(self.directory, current, mirror_run=False) revoke_session_capabilities(self.session_id) append_audit(self.session_id, "root_host_terminal", status=status, error=error) return status def recover(self) -> bool: if not self._has_durable_work(): return False previous = self.client if previous is not None: previous.close() for attempt, delay in enumerate(APP_SERVER_RECOVERY_DELAYS_SECONDS, start=1): if _SHUTDOWN.wait(delay): return False self.session = read_session_record(self.directory) prior_status = str(self.session.get("status")) if prior_status in { "finishing", "stopping", "cancelling", "completed", "stopped", "failed", "cancelled", }: return False try: self.connect() self._resume_turn_work_if_needed(prior_status) # Publish recovery under the same lock as resume admission. # The attached client may already have advanced detached to # running after observing the new app-server identity; never # restore the stale pre-recovery status over that decision. with file_lock(runtime_lock_path()): current = read_session_record(self.directory) current["root_recovery_attempts"] = attempt current_status = str(current.get("status")) if current_status not in { "finishing", "stopping", "cancelling", "completed", "stopped", "failed", "cancelled", }: current["status"] = ( current_status if current_status in {"paused", "detached"} else "running" ) publish_session_record(self.directory, current, mirror_run=True) self.session = current append_audit(self.session_id, "root_app_server_recovered", attempt=attempt) return True except (AppServerError, OSError) as exc: self._update(root_recovery_error=f"{type(exc).__name__}: {exc}") self._update(status="suspended") return False def _stop_recorded_app_server(self) -> None: current = read_session_record(self.directory) terminate_recorded_process_group( current, prefix="root_app_server", grace_seconds=8.0, ) def run(self) -> int: load_session_capabilities(self.directory) self._update( root_runtime_package_version=package_version(), root_runtime_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), ) try: self.connect() except (AppServerError, OSError): if ( not isinstance(self.session.get("root_thread_id"), str) or not self._has_durable_work() ): raise if not self.recover(): if isinstance(self.session.get("root_thread_id"), str): return 75 raise resume_status = self.session.get("resume_requested_from") if resume_status in {"detached", "paused", "suspended"}: failure = self.session.get("failure") completed_turn = self.host.completed_turn retrying_failed_turn = bool( resume_status == "suspended" and isinstance(failure, Mapping) and failure.get("retryable") is True and isinstance(completed_turn, Mapping) and completed_turn.get("status") == "failed" and completed_turn.get("id") == failure.get("turn_id") ) if retrying_failed_turn: # The failed turn remains immutable history, but it must not be # reclassified as a fresh failure before the controller can # accept an explicit continuation on the replacement host. self.host.take_completed_turn() with self.state_lock: self.state["turn_failure"] = None self._update( status="paused", active_root_turn_id=None, resume_requested_from=None, ) else: self._update( status="paused" if resume_status == "paused" else "running", resume_requested_from=None, ) control = threading.Thread(target=self.serve_control, name="mmo-root-control", daemon=True) control.start() try: while True: try: self.start_initial_work() break except AppServerError: if not self._has_durable_work(): raise if not self.recover(): return 75 atomic_write_json( self.directory / "root-host-ready.json", { "thread_id": self.host.thread_id, "socket": str(self.socket_path), "ready_at": utc_now(), }, ) stall_seconds = int(self.agent["stall_warning_seconds"]) while not _SHUTDOWN.wait(0.5): client = self.client if client is None or not client.alive: if not self.recover(): # Exit only the lightweight controller. The session is # suspended with its exact thread/evidence identity and # can start a fresh controller on explicit continue. return 75 continue self._ensure_bootstrap_goal() self._activate_bootstrap_goal() if ( not self.stall_reported and time.monotonic() - self.last_progress >= stall_seconds ): self.stall_reported = True self._update(root_stall_warning_at=utc_now()) append_audit( self.session_id, "root_stall_warning", warning_seconds=stall_seconds, ) completed_turn = self.host.completed_turn if isinstance(completed_turn, Mapping) and completed_turn.get("status") == "failed": failure = normalize_turn_failure(completed_turn) or { "kind": "turn_failed", "source": "turn/completed", "turn_id": completed_turn.get("id"), "turn_status": "failed", "message": "app-server turn failed", "retryable": False, "observed_at": utc_now(), } partial = retain_partial_evidence( self.session, self.directory, reason=str(failure["message"]), events_filename="root-events.jsonl", result_filename="root-result.md", partial_filename="root-partial-result.md", title="Partial root result", ) if failure.get("retryable"): if self._attached_root_client(): # A failed turn is immutable thread history, not a # reason to destroy a healthy app-server underneath # an attached TUI. Consume only this controller's # completion marker; a later accepted turn clears # the persisted failure through turn/started. self.host.take_completed_turn() now = utc_now() self._update( status="running", last_active_at=now, error=str(failure["message"]), failure=failure, **partial, ) append_audit( self.session_id, "root_retryable_failure_client_retained", turn_id=failure.get("turn_id"), failure_kind=failure.get("kind"), ) continue self._update( status="suspended", suspended_at=utc_now(), error=str(failure["message"]), failure=failure, **partial, ) return 75 self._update(error=str(failure["message"]), failure=failure, **partial) self.stop_status = "failed" break finalize_requested = bool(self.state.get("finalize_requested")) if finalize_requested and isinstance(completed_turn, Mapping): if ( self.finalize_turn_id is None or completed_turn.get("id") == self.finalize_turn_id ): self.stop_status = ( self.finalize_status if completed_turn.get("status") == "completed" else "stopped" if self.finalize_status == "stopped" else "failed" ) break if ( finalize_requested and self.finalize_deadline is not None and time.monotonic() >= self.finalize_deadline ): self.stop_status = "stopped" break goal = self.state.get("goal") if isinstance(goal, Mapping) and goal.get("status") == "complete": # Codex may publish goal completion from update_goal while # the terminal turn is still producing its final message. # Goal state is not a substitute for authoritative turn # completion; retiring the host here would discard that # message and leave only an in-progress history record. if self.host.active_turn_id is not None: continue descendants = self.recoverable_descendants() if descendants: recoverable_jobs = sorted(str(item["job_id"]) for item in descendants) if ( not self.session.get("root_completion_deferred") or sorted( str(item) for item in self.session.get("root_completion_deferred_jobs", []) ) != recoverable_jobs ): now = utc_now() self._update( status="detached", detached_at=now, last_active_at=now, root_completion_deferred=True, root_completion_deferred_jobs=recoverable_jobs, ) append_audit( self.session_id, "root_completion_deferred", recoverable_jobs=recoverable_jobs, ) continue self.stop_status = "completed" break if self.agent["execution_mode"] == "turn" and completed_turn: turn = completed_turn if ( self.session.get("status") == "paused" and isinstance(turn, Mapping) and turn.get("status") == "interrupted" ): self.host.take_completed_turn() continue self.stop_status = ( "completed" if isinstance(turn, Mapping) and turn.get("status") == "completed" else "failed" ) break if self.retryable_failure_detach_exit: self._update( status="suspended", active_root_turn_id=None, ) return 75 if self.pause_exit: partial = retain_partial_evidence( self.session, self.directory, reason="root cold-paused by operator", events_filename="root-events.jsonl", result_filename="root-result.md", partial_filename="root-partial-result.md", title="Partial root result", ) self._update( status="paused", paused_at=utc_now(), root_goal_status=("paused" if self.agent["execution_mode"] == "goal" else None), active_root_turn_id=None, **partial, ) return 75 status = self.stop_status or "cancelled" status = self._publish_terminal(status) return 0 if status == "completed" else 130 if status == "cancelled" else 1 finally: self.control_stop.set() control.join(timeout=2.0) if self.client is not None: if self.client.process is not None: self.client.stop_host() else: self.client.close() self._stop_recorded_app_server() self._update( root_control_socket_ready=False, root_app_server_pid=None, root_app_server_pgid=None, root_app_server_start_token=None, ) def main() -> int: if len(sys.argv) != 2: print("usage: root_runner.py SESSION_ID", file=sys.stderr) return 2 signal.signal(signal.SIGTERM, _signal_handler) signal.signal(signal.SIGINT, _signal_handler) runner = RootRunner(sys.argv[1]) try: return runner.run() except Exception as exc: with contextlib.suppress(Exception): runner._publish_terminal("failed", error=f"{type(exc).__name__}: {exc}") print(f"root runner failed: {type(exc).__name__}: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())