"""Stable local launcher for versioned RF-001 RX-agent releases.""" from __future__ import annotations import argparse import ctypes from ctypes import wintypes from dataclasses import dataclass from datetime import datetime, timezone import json import ntpath import os from pathlib import Path import secrets import signal import shlex import subprocess import sys import time from typing import Any, Mapping, Sequence, TextIO LAUNCHER_VERSION = "2026.09.29.1" RELEASE_ID_PREFIX = "rf001-agent-" PROCESS_QUERY_TIMEOUT_SECONDS = 8.0 PROCESS_CLEANUP_TIMEOUT_SECONDS = 5.0 PROCESS_CREATION_SKEW_SECONDS = 0.05 LAUNCHER_PHASES = frozenset( { "LEGACY_ACTIVE", "TARGET_VERIFIED", "STARTING_TARGET", "WAITING_FOR_HEALTH", "TARGET_HEALTHY", "TARGET_FAILED", "ROLLING_BACK", "LEGACY_RESTORED", } ) @dataclass(frozen=True) class ProcessIdentity: pid: int parent_pid: int executable: str command_line: str created_at: str @dataclass(frozen=True) class LaunchAttempt: process: subprocess.Popen[str] command: tuple[str, ...] wrapper_pid: int launch_boundary: str wrapper_executable: str child_executable: str launch_nonce: str @dataclass(frozen=True) class ValidatedTarget: mode: str wrapper: ProcessIdentity target: ProcessIdentity @dataclass(frozen=True) class CleanupResult: state: str actions: tuple[str, ...] orphan_verified_target_processes: tuple[int, ...] class StartupFailure(RuntimeError): def __init__(self, message: str, cleanup: CleanupResult) -> None: super().__init__(message) self.cleanup = cleanup def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def atomic_write_json(path: Path, value: Mapping[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + "\n", encoding="utf-8", ) os.replace(temporary, path) def read_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(value, dict): raise ValueError(f"JSON object is required: {path}") return value def redirect_output(path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) stream = path.open("a", encoding="utf-8", buffering=1) sys.stdout = stream sys.stderr = stream def _parse_utc(value: object) -> datetime | None: if not isinstance(value, str): return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _path_key(value: str) -> str: return ntpath.normcase(ntpath.normpath(value.strip().strip('"'))) def _looks_like_windows_path(value: str) -> bool: return len(value) >= 3 and value[1:3] in (":\\", ":/") def _command_arguments(command_line: str) -> list[str] | None: if not command_line.strip(): return None if os.name != "nt": try: return shlex.split(command_line, posix=False) except ValueError: return None argument_count = ctypes.c_int() command_to_argv = ctypes.windll.shell32.CommandLineToArgvW command_to_argv.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(ctypes.c_int)] command_to_argv.restype = ctypes.POINTER(wintypes.LPWSTR) pointer = command_to_argv(command_line, ctypes.byref(argument_count)) if not pointer: return None try: return [pointer[index] for index in range(argument_count.value)] finally: ctypes.windll.kernel32.LocalFree(pointer) def _command_matches(command_line: str, expected: Sequence[str]) -> bool: actual = _command_arguments(command_line) if actual is None or len(actual) != len(expected): return False for actual_value, expected_value in zip(actual, expected, strict=True): if _looks_like_windows_path(actual_value) or _looks_like_windows_path( expected_value ): if _path_key(actual_value) != _path_key(expected_value): return False elif actual_value != expected_value: return False return True def _process_identity_from_json(value: Mapping[str, Any]) -> ProcessIdentity: return ProcessIdentity( pid=int(value["pid"]), parent_pid=int(value["parent_pid"]), executable=str(value.get("executable") or ""), command_line=str(value.get("command_line") or ""), created_at=str(value.get("created_at") or ""), ) def _query_process_identities( wrapper_pid: int, health_pid: int | None = None, ) -> list[ProcessIdentity]: if os.name != "nt": raise RuntimeError("strict process identity requires Windows") process_ids = {wrapper_pid} if health_pid is not None: process_ids.add(health_pid) process_terms = " OR ".join( f"ProcessId = {process_id}" for process_id in sorted(process_ids) ) query = f"({process_terms}) OR ParentProcessId = {wrapper_pid}" script = ( "$ErrorActionPreference='Stop';" "$ProgressPreference='SilentlyContinue';" "[Console]::OutputEncoding=[Text.Encoding]::UTF8;" f"$rows=@(Get-CimInstance Win32_Process -Filter '{query}' | ForEach-Object {{" "[ordered]@{pid=[int]$_.ProcessId;parent_pid=[int]$_.ParentProcessId;" "executable=[string]$_.ExecutablePath;command_line=[string]$_.CommandLine;" "created_at=$_.CreationDate.ToUniversalTime().ToString('o')}});" "[Console]::WriteLine((ConvertTo-Json -InputObject $rows -Compress))" ) system_root = Path(os.environ.get("SystemRoot", r"C:\Windows")) powershell = ( system_root / "System32" / "WindowsPowerShell" / "v1.0" / "powershell.exe" ) executable = str(powershell) if powershell.is_file() else "powershell.exe" result = subprocess.run( [ executable, "-NoLogo", "-NoProfile", "-NonInteractive", "-OutputFormat", "Text", "-Command", script, ], stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8-sig", errors="strict", timeout=PROCESS_QUERY_TIMEOUT_SECONDS, check=False, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) if result.returncode != 0: detail = result.stderr.strip() or f"exit code {result.returncode}" raise RuntimeError(f"Windows process identity query failed: {detail}") raw = result.stdout.strip() if not raw: return [] value = json.loads(raw) rows = value if isinstance(value, list) else [value] if not all(isinstance(row, dict) for row in rows): raise RuntimeError("Windows process identity query returned invalid JSON") return [_process_identity_from_json(row) for row in rows] def _same_process( first: ProcessIdentity, second: ProcessIdentity, *, require_parent: bool = True, ) -> bool: return ( first.pid == second.pid and (not require_parent or first.parent_pid == second.parent_pid) and _path_key(first.executable) == _path_key(second.executable) and first.command_line == second.command_line and first.created_at == second.created_at ) def _created_after_boundary(identity: ProcessIdentity, boundary: str) -> bool: created_at = _parse_utc(identity.created_at) launched_at = _parse_utc(boundary) if created_at is None or launched_at is None: return False return created_at.timestamp() + PROCESS_CREATION_SKEW_SECONDS >= launched_at.timestamp() def _runtime_executables(command_executable: str) -> tuple[str, str]: wrapper = str(Path(command_executable).resolve()) base_value = getattr(sys, "_base_executable", None) if not isinstance(base_value, str) or not base_value: base_value = command_executable base = str(Path(base_value).resolve()) return wrapper, base def _wrapper_matches( identity: ProcessIdentity, attempt: LaunchAttempt, ) -> bool: return ( identity.pid == attempt.wrapper_pid and _path_key(identity.executable) == _path_key(attempt.wrapper_executable) and _created_after_boundary(identity, attempt.launch_boundary) and _command_matches(identity.command_line, attempt.command) ) def _child_matches( identity: ProcessIdentity, attempt: LaunchAttempt, ) -> bool: return ( identity.parent_pid == attempt.wrapper_pid and _path_key(identity.executable) == _path_key(attempt.child_executable) and _created_after_boundary(identity, attempt.launch_boundary) and _command_matches(identity.command_line, attempt.command) ) def _release_directory(config: Mapping[str, Any], release_id: str) -> Path: if not release_id.startswith(RELEASE_ID_PREFIX): raise ValueError("release_id is invalid") releases_root = Path(str(config["releases_root"])).resolve() candidate = (releases_root / release_id).resolve() try: candidate.relative_to(releases_root) except ValueError as error: raise ValueError("release directory escaped releases root") from error return candidate def validate_ready_release( config: Mapping[str, Any], release_id: str, expected_version: str | None = None, ) -> tuple[Path, dict[str, Any]]: release_directory = _release_directory(config, release_id) ready_path = release_directory / "READY.json" ready = read_json(ready_path) if ready.get("status") != "READY" or ready.get("release_id") != release_id: raise ValueError("release READY marker is invalid") agent_version = ready.get("agent_version") if not isinstance(agent_version, str): raise ValueError("release agent_version is invalid") if expected_version is not None and agent_version != expected_version: raise ValueError("release agent_version does not match update request") agent_script = release_directory / "app" / "tools" / "rf001_rx_http_agent.py" if not agent_script.is_file(): raise FileNotFoundError(f"release agent is absent: {agent_script}") return agent_script, ready def write_launcher_status( config: Mapping[str, Any], *, state: str, active_release_id: str, active_agent_version: str | None, error: str | None = None, previous_release_id: str | None = None, phase: str | None = None, cleanup_state: str | None = None, ) -> None: if phase is not None and phase not in LAUNCHER_PHASES: raise ValueError(f"unsupported launcher phase: {phase}") atomic_write_json( Path(str(config["launcher_status"])), { "schema_version": 1, "launcher_version": LAUNCHER_VERSION, "state": state, "active_release_id": active_release_id, "active_agent_version": active_agent_version, "previous_release_id": previous_release_id, "error": error, "phase": phase, "cleanup_state": cleanup_state, "updated_at": utc_now(), "hardware_accessed": False, "rf_tx": 0, }, ) def _agent_command( config: Mapping[str, Any], agent_script: Path, release_id: str, launch_nonce: str, ) -> list[str]: arguments = config.get("agent_arguments") if not isinstance(arguments, list) or not all( isinstance(value, str) for value in arguments ): raise ValueError("agent_arguments must be a list of strings") return [ sys.executable, "-u", "-B", "-X", "utf8", str(agent_script), *arguments, "--release-id", release_id, "--launch-nonce", launch_nonce, "--health-file", str(config["health_file"]), "--launcher-status", str(config["launcher_status"]), "--launcher-request", str(config["launcher_request"]), "--install-root", str(config["install_root"]), ] def _spawn_agent( command: Sequence[str], *, cwd: Path, log_stream: TextIO, ) -> subprocess.Popen[str]: return subprocess.Popen( list(command), cwd=cwd, stdin=subprocess.DEVNULL, stdout=log_stream, stderr=subprocess.STDOUT, text=True, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) def _capture_wrapper_identity( attempt: LaunchAttempt, *, timeout_seconds: float, ) -> ProcessIdentity: deadline = time.monotonic() + timeout_seconds last_error = "wrapper process identity was not observed" while time.monotonic() < deadline: if attempt.process.poll() is not None: raise RuntimeError( f"wrapper exited before identity capture with code " f"{attempt.process.returncode}" ) try: records = _query_process_identities(attempt.wrapper_pid) except (OSError, RuntimeError, subprocess.SubprocessError, json.JSONDecodeError) as error: last_error = f"process identity query failed: {type(error).__name__}: {error}" else: wrappers = [ record for record in records if record.pid == attempt.wrapper_pid and _wrapper_matches(record, attempt) ] if len(wrappers) == 1: return wrappers[0] if wrappers: last_error = "wrapper process identity is ambiguous" else: last_error = "wrapper process identity does not match launch" time.sleep(0.05) raise RuntimeError(last_error) def _health_fields_match( health: Mapping[str, Any], *, release_id: str, agent_version: str, launch_nonce: str, launch_boundary: str, ) -> bool: updated_at = _parse_utc(health.get("updated_at")) boundary = _parse_utc(launch_boundary) return ( health.get("release_id") == release_id and health.get("agent_version") == agent_version and health.get("launch_nonce") == launch_nonce and health.get("state") == "RUNNING" and type(health.get("pid")) is int and updated_at is not None and boundary is not None and updated_at >= boundary ) def _validate_target_identity( attempt: LaunchAttempt, wrapper_identity: ProcessIdentity, health_pid: int, records: Sequence[ProcessIdentity], ) -> ValidatedTarget: if attempt.process.poll() is not None: raise RuntimeError("wrapper died before process identity validation") by_pid = {record.pid: record for record in records} current_wrapper = by_pid.get(attempt.wrapper_pid) if current_wrapper is None or not _same_process(current_wrapper, wrapper_identity): raise RuntimeError("wrapper process identity changed or disappeared") if not _wrapper_matches(current_wrapper, attempt): raise RuntimeError("wrapper process does not match expected launch") target = by_pid.get(health_pid) if target is None: raise RuntimeError("health process is absent") if health_pid == attempt.wrapper_pid: if not _same_process(target, wrapper_identity): raise RuntimeError("direct target process identity changed") return ValidatedTarget( mode="DIRECT", wrapper=current_wrapper, target=target, ) if target.parent_pid != attempt.wrapper_pid: raise RuntimeError("health process is not a direct wrapper child") matching_children = [ record for record in records if record.pid != attempt.wrapper_pid and _child_matches(record, attempt) ] if len(matching_children) != 1 or matching_children[0].pid != health_pid: raise RuntimeError("expected direct wrapper child is absent or ambiguous") return ValidatedTarget( mode="WINDOWS_VENV_WRAPPER", wrapper=current_wrapper, target=target, ) def _validate_target_twice( attempt: LaunchAttempt, wrapper_identity: ProcessIdentity, health_pid: int, ) -> ValidatedTarget: first_records = _query_process_identities(attempt.wrapper_pid, health_pid) first = _validate_target_identity( attempt, wrapper_identity, health_pid, first_records, ) second_records = _query_process_identities(attempt.wrapper_pid, health_pid) second = _validate_target_identity( attempt, wrapper_identity, health_pid, second_records, ) if first.mode != second.mode: raise RuntimeError("target process topology changed during validation") if not _same_process(first.wrapper, second.wrapper) or not _same_process( first.target, second.target, ): raise RuntimeError("target process identity changed during validation") return second def _identity_is_present(identity: ProcessIdentity) -> bool: records = _query_process_identities(identity.pid, identity.pid) current = next((record for record in records if record.pid == identity.pid), None) return current is not None and _same_process(current, identity) def _terminate_verified_identity(identity: ProcessIdentity) -> str: records = _query_process_identities(identity.pid, identity.pid) current = next((record for record in records if record.pid == identity.pid), None) if current is None: return f"PID {identity.pid}: already absent" if not _same_process(current, identity): raise RuntimeError(f"PID {identity.pid}: identity changed before cleanup") os.kill(identity.pid, signal.SIGTERM) deadline = time.monotonic() + PROCESS_CLEANUP_TIMEOUT_SECONDS while time.monotonic() < deadline: if not _identity_is_present(identity): return f"PID {identity.pid}: terminated" time.sleep(0.05) raise RuntimeError(f"PID {identity.pid}: did not terminate within timeout") def _verified_matching_children( attempt: LaunchAttempt, records: Sequence[ProcessIdentity], ) -> list[ProcessIdentity]: return [ record for record in records if record.pid != attempt.wrapper_pid and _child_matches(record, attempt) ] def strict_failure_cleanup( attempt: LaunchAttempt, wrapper_identity: ProcessIdentity | None, ) -> CleanupResult: actions: list[str] = [] try: records = _query_process_identities(attempt.wrapper_pid) except Exception as error: return CleanupResult( state="CLEANUP_INCOMPLETE", actions=(f"identity query failed: {type(error).__name__}: {error}",), orphan_verified_target_processes=(), ) current_wrapper = next( (record for record in records if record.pid == attempt.wrapper_pid), None, ) matching_children = _verified_matching_children(attempt, records) if wrapper_identity is None: if current_wrapper is None and not matching_children: return CleanupResult("COMPLETE", (), ()) return CleanupResult( state="CLEANUP_INCOMPLETE", actions=("wrapper identity was never proven; no process terminated",), orphan_verified_target_processes=tuple( sorted(record.pid for record in matching_children) ), ) if current_wrapper is not None and not _same_process( current_wrapper, wrapper_identity, ): return CleanupResult( state="CLEANUP_INCOMPLETE", actions=("wrapper PID identity changed; no process terminated",), orphan_verified_target_processes=tuple( sorted(record.pid for record in matching_children) ), ) if len(matching_children) > 1: return CleanupResult( state="CLEANUP_INCOMPLETE", actions=("matching direct child identity is ambiguous; no process terminated",), orphan_verified_target_processes=tuple( sorted(record.pid for record in matching_children) ), ) try: if matching_children: actions.append(_terminate_verified_identity(matching_children[0])) refreshed = _query_process_identities(attempt.wrapper_pid) refreshed_wrapper = next( (record for record in refreshed if record.pid == attempt.wrapper_pid), None, ) if refreshed_wrapper is not None: if not _same_process(refreshed_wrapper, wrapper_identity): raise RuntimeError("wrapper identity changed before termination") terminate_process(attempt.process) actions.append(f"PID {attempt.wrapper_pid}: wrapper terminated") except Exception as error: actions.append(f"cleanup error: {type(error).__name__}: {error}") final_records = _query_process_identities(attempt.wrapper_pid) remaining: list[int] = [] final_wrapper = next( (record for record in final_records if record.pid == attempt.wrapper_pid), None, ) if final_wrapper is not None and _same_process(final_wrapper, wrapper_identity): remaining.append(final_wrapper.pid) remaining.extend( record.pid for record in _verified_matching_children(attempt, final_records) ) unique_remaining = tuple(sorted(set(remaining))) return CleanupResult( state="COMPLETE" if not unique_remaining else "CLEANUP_INCOMPLETE", actions=tuple(actions), orphan_verified_target_processes=unique_remaining, ) def start_release( config: Mapping[str, Any], release_id: str, *, expected_version: str | None = None, ) -> tuple[subprocess.Popen[str], str]: agent_script, ready = validate_ready_release( config, release_id, expected_version, ) agent_version = str(ready["agent_version"]) launch_nonce = secrets.token_hex(16) health_file = Path(str(config["health_file"])) health_before = health_file.stat().st_mtime_ns if health_file.exists() else None log_path = Path(str(config["agent_log"])).resolve() log_path.parent.mkdir(parents=True, exist_ok=True) log_stream: TextIO = log_path.open("a", encoding="utf-8", buffering=1) command = _agent_command(config, agent_script, release_id, launch_nonce) launch_boundary = utc_now() try: process = _spawn_agent( command, cwd=agent_script.parents[2], log_stream=log_stream, ) finally: log_stream.close() wrapper_executable, child_executable = _runtime_executables(command[0]) attempt = LaunchAttempt( process=process, command=tuple(command), wrapper_pid=process.pid, launch_boundary=launch_boundary, wrapper_executable=wrapper_executable, child_executable=child_executable, launch_nonce=launch_nonce, ) timeout = float(config.get("startup_timeout_seconds", 30.0)) deadline = time.monotonic() + timeout last_error = "agent health confirmation timed out" wrapper_identity: ProcessIdentity | None = None try: wrapper_identity = _capture_wrapper_identity( attempt, timeout_seconds=min(5.0, max(0.1, timeout)), ) while time.monotonic() < deadline: returncode = process.poll() if returncode is not None: last_error = f"agent wrapper exited during startup with code {returncode}" break if health_file.exists(): changed = ( health_before is None or health_file.stat().st_mtime_ns != health_before ) if changed: try: health = read_json(health_file) except (OSError, ValueError, json.JSONDecodeError) as error: last_error = ( "health file is invalid: " f"{type(error).__name__}: {error}" ) else: if not _health_fields_match( health, release_id=release_id, agent_version=agent_version, launch_nonce=launch_nonce, launch_boundary=launch_boundary, ): last_error = "health fields do not match the current launch" else: try: _validate_target_twice( attempt, wrapper_identity, int(health["pid"]), ) except ( OSError, RuntimeError, subprocess.SubprocessError, json.JSONDecodeError, ) as error: last_error = ( "process identity validation failed: " f"{type(error).__name__}: {error}" ) else: return process, agent_version time.sleep(0.25) except ( OSError, RuntimeError, subprocess.SubprocessError, json.JSONDecodeError, ) as error: last_error = f"startup identity failed: {type(error).__name__}: {error}" cleanup = strict_failure_cleanup(attempt, wrapper_identity) cleanup_detail = ( f"cleanup={cleanup.state}; " f"orphan_verified_target_processes=" f"{list(cleanup.orphan_verified_target_processes)}" ) raise StartupFailure(f"{last_error}; {cleanup_detail}", cleanup) def terminate_process(process: subprocess.Popen[str]) -> None: if process.poll() is not None: return process.terminate() try: process.wait(timeout=10.0) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5.0) def _write_active_release(config: Mapping[str, Any], release_id: str) -> None: atomic_write_json( Path(str(config["active_release_file"])), {"schema_version": 1, "release_id": release_id, "updated_at": utc_now()}, ) def _archive_request(config: Mapping[str, Any], request: Mapping[str, Any]) -> None: request_path = Path(str(config["launcher_request"])) history = request_path.parent / "request_history" history.mkdir(parents=True, exist_ok=True) task_id = str(request.get("task_id", "unknown")) destination = history / f"{task_id}_{int(time.time())}.json" os.replace(request_path, destination) def switch_release( config: Mapping[str, Any], process: subprocess.Popen[str], current_release_id: str, current_version: str, request: Mapping[str, Any], ) -> tuple[subprocess.Popen[str], str, str]: target_release_id = request.get("release_id") target_version = request.get("agent_version") if request.get("safe_to_switch") is not True: raise ValueError("launcher request is not marked safe to switch") if request.get("from_release_id") != current_release_id: raise ValueError("launcher request was created for another active release") if not isinstance(target_release_id, str) or not isinstance(target_version, str): raise ValueError("launcher request target is invalid") health = read_json(Path(str(config["health_file"]))) if health.get("busy") is not False or health.get("active_task_id") is not None: raise RuntimeError("agent is not idle at the update switch point") write_launcher_status( config, state="RUNNING", active_release_id=current_release_id, active_agent_version=current_version, phase="LEGACY_ACTIVE", ) validate_ready_release(config, target_release_id, target_version) write_launcher_status( config, state="RUNNING", active_release_id=current_release_id, active_agent_version=current_version, previous_release_id=target_release_id, phase="TARGET_VERIFIED", ) terminate_process(process) _write_active_release(config, target_release_id) write_launcher_status( config, state="STARTING_UPDATE", active_release_id=target_release_id, active_agent_version=target_version, previous_release_id=current_release_id, phase="STARTING_TARGET", ) try: write_launcher_status( config, state="STARTING_UPDATE", active_release_id=target_release_id, active_agent_version=target_version, previous_release_id=current_release_id, phase="WAITING_FOR_HEALTH", ) new_process, new_version = start_release( config, target_release_id, expected_version=target_version, ) except Exception as error: cleanup_state = ( error.cleanup.state if isinstance(error, StartupFailure) else None ) write_launcher_status( config, state="STARTING_UPDATE", active_release_id=target_release_id, active_agent_version=target_version, previous_release_id=current_release_id, error=f"{type(error).__name__}: {error}", phase="TARGET_FAILED", cleanup_state=cleanup_state, ) _write_active_release(config, current_release_id) write_launcher_status( config, state="STARTING", active_release_id=current_release_id, active_agent_version=current_version, previous_release_id=target_release_id, error=f"{type(error).__name__}: {error}", phase="ROLLING_BACK", cleanup_state=cleanup_state, ) rollback_process, rollback_version = start_release( config, current_release_id, expected_version=current_version, ) write_launcher_status( config, state="ROLLED_BACK", active_release_id=current_release_id, active_agent_version=rollback_version, previous_release_id=target_release_id, error=f"{type(error).__name__}: {error}", phase="LEGACY_RESTORED", cleanup_state=cleanup_state, ) _archive_request(config, request) return rollback_process, current_release_id, rollback_version write_launcher_status( config, state="RUNNING", active_release_id=target_release_id, active_agent_version=new_version, previous_release_id=current_release_id, phase="TARGET_HEALTHY", cleanup_state="COMPLETE", ) _archive_request(config, request) return new_process, target_release_id, new_version def run_launcher(config: Mapping[str, Any], *, once: bool = False) -> None: active = read_json(Path(str(config["active_release_file"]))) release_id = active.get("release_id") if not isinstance(release_id, str): raise ValueError("active release pointer is invalid") write_launcher_status( config, state="STARTING", active_release_id=release_id, active_agent_version=None, phase="STARTING_TARGET", ) write_launcher_status( config, state="STARTING", active_release_id=release_id, active_agent_version=None, phase="WAITING_FOR_HEALTH", ) process, agent_version = start_release(config, release_id) write_launcher_status( config, state="RUNNING", active_release_id=release_id, active_agent_version=agent_version, phase="TARGET_HEALTHY", cleanup_state="COMPLETE", ) if once: terminate_process(process) return request_path = Path(str(config["launcher_request"])) while True: if request_path.exists(): try: request = read_json(request_path) process, release_id, agent_version = switch_release( config, process, release_id, agent_version, request, ) except Exception as error: write_launcher_status( config, state="UPDATE_REQUEST_REJECTED", active_release_id=release_id, active_agent_version=agent_version, error=f"{type(error).__name__}: {error}", ) _archive_request(config, {"task_id": "rejected"}) returncode = process.poll() if returncode is not None: write_launcher_status( config, state="RESTARTING", active_release_id=release_id, active_agent_version=agent_version, error=f"agent exited with code {returncode}", phase="STARTING_TARGET", ) time.sleep(float(config.get("restart_delay_seconds", 2.0))) write_launcher_status( config, state="RESTARTING", active_release_id=release_id, active_agent_version=agent_version, phase="WAITING_FOR_HEALTH", ) process, agent_version = start_release( config, release_id, expected_version=agent_version, ) write_launcher_status( config, state="RUNNING", active_release_id=release_id, active_agent_version=agent_version, phase="TARGET_HEALTHY", cleanup_state="COMPLETE", ) time.sleep(0.5) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", required=True, type=Path) parser.add_argument("--log-file", type=Path) parser.add_argument("--once", action="store_true") arguments = parser.parse_args() if arguments.log_file is not None: redirect_output(arguments.log_file) config = read_json(arguments.config.resolve()) print(f"LAUNCHER VERSION = {LAUNCHER_VERSION}") print("ARBITRARY SHELL = DISABLED") print("RF TX = 0") run_launcher(config, once=arguments.once) if __name__ == "__main__": main()