"""Isolated Windows diagnostic for the RF-001 launcher/health handshake. The probe imports the production launcher and calls its ``start_release`` path, but replaces only the child command with this module's ``health-child`` mode. That child imports ``write_health`` from the immutable canonical release, writes one diagnostic health file, never polls HTTP, never touches SDR, and exits after the parent has preserved process evidence. """ from __future__ import annotations import argparse import base64 from dataclasses import asdict, dataclass from datetime import datetime, timezone import importlib.util import json import os from pathlib import Path import signal import subprocess import sys import threading import time from types import ModuleType from typing import Any, Mapping, Sequence EXPECTED_STATE = "RUNNING" NO_HARDWARE = False NO_RF_TX = 0 _ORIGINAL_POPEN = subprocess.Popen def utc_now() -> str: return datetime.now(timezone.utc).isoformat() 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: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) 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, sort_keys=True, separators=(",", ":")) + "\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 @dataclass(frozen=True) class ProcessRecord: pid: int parent_pid: int name: str | None executable: str | None command_line: str | None created_at: str | None @dataclass(frozen=True) class HandshakeExpectation: release_id: str agent_version: str launch_nonce: str state: str launcher_pid: int popen_pid: int launched_at: str diagnostic_script: str def _powershell_json(script: str) -> object: encoded = base64.b64encode(script.encode("utf-16-le")).decode("ascii") process = _ORIGINAL_POPEN( [ "powershell.exe", "-NoProfile", "-NonInteractive", "-EncodedCommand", encoded, ], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False, ) stdout, stderr = process.communicate(timeout=20) if process.returncode != 0: detail = stderr.decode("utf-8", errors="replace").strip() raise RuntimeError(f"Windows process inventory failed: {detail}") text = stdout.decode("utf-8-sig", errors="strict").strip() if not text: return [] return json.loads(text) def capture_windows_processes() -> list[ProcessRecord]: if os.name != "nt": raise RuntimeError("Windows process inventory is required") script = r''' $ErrorActionPreference = "Stop" $ProgressPreference = "SilentlyContinue" [Console]::OutputEncoding = New-Object Text.UTF8Encoding($false) $rows = @(Get-CimInstance Win32_Process | ForEach-Object { $created = $null if ($null -ne $_.CreationDate) { $created = $_.CreationDate.ToUniversalTime().ToString("o") } [ordered]@{ pid = [int]$_.ProcessId parent_pid = [int]$_.ParentProcessId name = $_.Name executable = $_.ExecutablePath command_line = $_.CommandLine created_at = $created } }) ConvertTo-Json -InputObject $rows -Depth 4 -Compress ''' raw = _powershell_json(script) if isinstance(raw, dict): raw = [raw] if not isinstance(raw, list): raise ValueError("Windows process inventory is not a list") records: list[ProcessRecord] = [] for row in raw: if not isinstance(row, dict): continue records.append( ProcessRecord( pid=int(row["pid"]), parent_pid=int(row["parent_pid"]), name=row.get("name") if isinstance(row.get("name"), str) else None, executable=( row.get("executable") if isinstance(row.get("executable"), str) else None ), command_line=( row.get("command_line") if isinstance(row.get("command_line"), str) else None ), created_at=( row.get("created_at") if isinstance(row.get("created_at"), str) else None ), ) ) return records def process_map(records: Sequence[ProcessRecord]) -> dict[int, ProcessRecord]: return {record.pid: record for record in records} def is_descendant( child_pid: int, ancestor_pid: int, records: Sequence[ProcessRecord], ) -> bool: by_pid = process_map(records) current = by_pid.get(child_pid) visited: set[int] = set() while current is not None and current.pid not in visited: if current.parent_pid == ancestor_pid: return True visited.add(current.pid) current = by_pid.get(current.parent_pid) return False def _record_is_fresh(record: ProcessRecord, launched_at: datetime) -> bool: created_at = parse_utc(record.created_at) if created_at is None: return False return created_at >= launched_at def _record_matches_launch( record: ProcessRecord, expectation: HandshakeExpectation, ) -> bool: command_line = record.command_line or "" return ( expectation.launch_nonce in command_line and expectation.diagnostic_script.casefold() in command_line.casefold() ) def compare_health_fields( health: Mapping[str, Any] | None, expectation: HandshakeExpectation, ) -> dict[str, bool]: if health is None: return { "release_id_match": False, "agent_version_match": False, "launch_nonce_match": False, "state_match": False, "pid_match": False, } return { "release_id_match": health.get("release_id") == expectation.release_id, "agent_version_match": health.get("agent_version") == expectation.agent_version, "launch_nonce_match": health.get("launch_nonce") == expectation.launch_nonce, "state_match": health.get("state") == expectation.state, "pid_match": health.get("pid") == expectation.popen_pid, } def evaluate_identity( health: Mapping[str, Any] | None, expectation: HandshakeExpectation, records: Sequence[ProcessRecord], *, popen_alive: bool, health_process_alive: bool, ) -> dict[str, Any]: fields = compare_health_fields(health, expectation) result: dict[str, Any] = { **fields, "identity_valid": False, "identity_model": "REJECTED", "reason": None, } if health is None: result["reason"] = "HEALTH_ABSENT" return result if not all( fields[name] for name in ( "release_id_match", "agent_version_match", "launch_nonce_match", "state_match", ) ): result["reason"] = "HEALTH_FIELDS_MISMATCH" return result updated_at = parse_utc(health.get("updated_at")) launched_at = parse_utc(expectation.launched_at) if updated_at is None or launched_at is None or updated_at < launched_at: result["reason"] = "STALE_HEALTH" return result health_pid = health.get("pid") if type(health_pid) is not int: result["reason"] = "HEALTH_PID_INVALID" return result by_pid = process_map(records) health_record = by_pid.get(health_pid) if health_record is None: result["reason"] = "HEALTH_PROCESS_ABSENT" return result if not health_process_alive: result["reason"] = "HEALTH_PROCESS_DIED" return result if not _record_is_fresh(health_record, launched_at): result["reason"] = "PROCESS_CREATION_TIME_MISMATCH" return result if not _record_matches_launch(health_record, expectation): result["reason"] = "PROCESS_COMMAND_MISMATCH" return result if health_pid == expectation.popen_pid: if not popen_alive: result["reason"] = "TARGET_DIED" return result result.update( identity_valid=True, identity_model="EXACT_POPEN_PROCESS", reason="PASS", ) return result popen_record = by_pid.get(expectation.popen_pid) if popen_record is None: result["reason"] = "POPEN_PROCESS_NOT_CAPTURED" return result direct_children = [ record for record in records if record.parent_pid == expectation.popen_pid and _record_is_fresh(record, launched_at) and _record_matches_launch(record, expectation) ] if health_record.parent_pid != expectation.popen_pid: if is_descendant(health_pid, expectation.popen_pid, records): result["reason"] = "UNEXPECTED_GRANDCHILD" else: result["reason"] = "UNRELATED_PID" return result if len(direct_children) != 1 or direct_children[0].pid != health_pid: result["reason"] = "AMBIGUOUS_DIRECT_CHILD" return result if not _record_matches_launch(popen_record, expectation): result["reason"] = "POPEN_COMMAND_MISMATCH" return result result.update( identity_valid=True, identity_model="UNIQUE_DIRECT_CHILD_WITH_NONCE_AND_CREATION_TIME", reason="PASS", ) return result def strict_owned_process_ids( expectation: HandshakeExpectation, records: Sequence[ProcessRecord], ) -> list[int]: launched_at = parse_utc(expectation.launched_at) if launched_at is None: return [] result: list[int] = [] for record in records: if record.pid == expectation.launcher_pid: continue if not _record_is_fresh(record, launched_at): continue if not _record_matches_launch(record, expectation): continue if record.pid == expectation.popen_pid or is_descendant( record.pid, expectation.popen_pid, records ): result.append(record.pid) return sorted(set(result), reverse=True) def relevant_process_tree( records: Sequence[ProcessRecord], expectation: HandshakeExpectation, health_pid: int | None, ) -> list[dict[str, Any]]: by_pid = process_map(records) selected: set[int] = {expectation.launcher_pid, expectation.popen_pid} if health_pid is not None: selected.add(health_pid) for record in records: if is_descendant(record.pid, expectation.popen_pid, records): selected.add(record.pid) for seed in tuple(selected): current = by_pid.get(seed) while current is not None and current.parent_pid not in selected: if current.parent_pid == 0: break selected.add(current.parent_pid) current = by_pid.get(current.parent_pid) return [asdict(by_pid[pid]) for pid in sorted(selected) if pid in by_pid] def _load_module(path: Path, name: str) -> ModuleType: specification = importlib.util.spec_from_file_location(name, path) if specification is None or specification.loader is None: raise ImportError(f"Could not load Python module: {path}") module = importlib.util.module_from_spec(specification) sys.modules[name] = module specification.loader.exec_module(module) return module def run_health_child(arguments: argparse.Namespace) -> int: agent_source = arguments.agent_source.resolve() agent = _load_module(agent_source, f"rf001_diag_agent_{os.getpid()}") namespace = argparse.Namespace( health_file=arguments.health_file.resolve(), release_id=arguments.release_id, launch_nonce=arguments.launch_nonce, ) child_report = arguments.child_report.resolve() started_at = utc_now() atomic_write_json( child_report, { "schema_version": 1, "phase": "BEFORE_HEALTH", "pid": os.getpid(), "parent_pid": os.getppid(), "sys_executable": sys.executable, "command_line": list(sys.argv), "started_at": started_at, "network_task_polling": False, "hardware_accessed": NO_HARDWARE, "rf_tx": NO_RF_TX, }, ) agent.write_health(namespace, None) health_written_at = utc_now() atomic_write_json( child_report, { "schema_version": 1, "phase": "HEALTH_WRITTEN", "pid": os.getpid(), "parent_pid": os.getppid(), "sys_executable": sys.executable, "command_line": list(sys.argv), "started_at": started_at, "health_written_at": health_written_at, "network_task_polling": False, "hardware_accessed": NO_HARDWARE, "rf_tx": NO_RF_TX, }, ) time.sleep(arguments.hold_seconds) return 0 def _process_is_present(pid: int, records: Sequence[ProcessRecord]) -> bool: return any(record.pid == pid for record in records) def _terminate_owned_processes( expectation: HandshakeExpectation, records: Sequence[ProcessRecord], ) -> list[dict[str, Any]]: actions: list[dict[str, Any]] = [] for pid in strict_owned_process_ids(expectation, records): fresh = capture_windows_processes() current = process_map(fresh).get(pid) original = process_map(records).get(pid) if current is None or original is None: continue if current.created_at != original.created_at or not _record_matches_launch( current, expectation ): actions.append({"pid": pid, "status": "IDENTITY_CHANGED_NOT_TERMINATED"}) continue try: os.kill(pid, signal.SIGTERM) except ProcessLookupError: actions.append({"pid": pid, "status": "ALREADY_EXITED"}) else: actions.append({"pid": pid, "status": "TERMINATED"}) return actions def run_probe_once( launcher: ModuleType, *, canonical_release: Path, output_directory: Path, startup_timeout: float, hold_seconds: float, ) -> dict[str, Any]: output_directory.mkdir(parents=True, exist_ok=False) ready = read_json(canonical_release / "READY.json") release_id = str(ready["release_id"]) agent_version = str(ready["agent_version"]) agent_source = canonical_release / "app" / "tools" / "rf001_rx_http_agent.py" health_path = output_directory / "diagnostic_health.json" child_report_path = output_directory / "diagnostic_child.json" config: dict[str, object] = { "install_root": str(output_directory / "isolated_install_root"), "releases_root": str(canonical_release.parent), "active_release_file": str(output_directory / "unused_active_release.json"), "launcher_status": str(output_directory / "unused_launcher_status.json"), "launcher_request": str(output_directory / "unused_launcher_request.json"), "health_file": str(health_path), "agent_log": str(output_directory / "diagnostic_child.log"), "startup_timeout_seconds": startup_timeout, "agent_arguments": [], } captured: dict[str, Any] = {} process_ready = threading.Event() health_observed = threading.Event() observer_snapshot: list[ProcessRecord] = [] original_command = launcher._agent_command original_spawn_agent = launcher._spawn_agent launcher_pid = os.getpid() launched_at = utc_now() def diagnostic_command( _config: Mapping[str, Any], actual_agent_source: Path, actual_release_id: str, launch_nonce: str, ) -> list[str]: captured["launch_nonce"] = launch_nonce command = [ sys.executable, "-u", "-B", "-X", "utf8", str(Path(__file__).resolve()), "health-child", "--agent-source", str(actual_agent_source), "--release-id", actual_release_id, "--launch-nonce", launch_nonce, "--health-file", str(health_path), "--child-report", str(child_report_path), "--hold-seconds", str(hold_seconds), ] captured["command"] = command return command def recording_spawn_agent( command: Sequence[str], *, cwd: Path, log_stream: Any, ) -> subprocess.Popen[str]: process = original_spawn_agent(command, cwd=cwd, log_stream=log_stream) captured["process"] = process captured["popen_pid"] = process.pid captured["popen_created_at"] = utc_now() process_ready.set() return process def observer() -> None: nonlocal observer_snapshot if not process_ready.wait(timeout=10): return deadline = time.monotonic() + max(10.0, startup_timeout + 5.0) while time.monotonic() < deadline: if health_path.exists() and child_report_path.exists(): observer_snapshot = capture_windows_processes() health_observed.set() return time.sleep(0.02) observer_thread = threading.Thread(target=observer, name="rf001-health-observer") observer_thread.start() launcher._agent_command = diagnostic_command launcher._spawn_agent = recording_spawn_agent start_result = "UNKNOWN" start_error: str | None = None returned_process: subprocess.Popen[str] | None = None try: returned_process, returned_version = launcher.start_release( config, release_id, expected_version=agent_version, ) start_result = "PASS" captured["returned_version"] = returned_version except Exception as error: start_result = "FAIL" start_error = f"{type(error).__name__}: {error}" finally: launcher._agent_command = original_command launcher._spawn_agent = original_spawn_agent observer_thread.join(timeout=15) popen_pid = int(captured.get("popen_pid", -1)) launch_nonce = str(captured.get("launch_nonce", "")) expectation = HandshakeExpectation( release_id=release_id, agent_version=agent_version, launch_nonce=launch_nonce, state=EXPECTED_STATE, launcher_pid=launcher_pid, popen_pid=popen_pid, launched_at=launched_at, diagnostic_script=str(Path(__file__).resolve()), ) health = read_json(health_path) if health_path.exists() else None child_report = read_json(child_report_path) if child_report_path.exists() else None evidence_snapshot = observer_snapshot or capture_windows_processes() health_pid = health.get("pid") if isinstance(health, dict) else None typed_health_pid = health_pid if type(health_pid) is int else None popen_process = captured.get("process") popen_alive_at_evaluation = ( isinstance(popen_process, subprocess.Popen) and popen_process.poll() is None ) health_process_alive = ( typed_health_pid is not None and _process_is_present(typed_health_pid, evidence_snapshot) ) evaluation = evaluate_identity( health, expectation, evidence_snapshot, popen_alive=popen_alive_at_evaluation, health_process_alive=health_process_alive, ) post_failure_snapshot = capture_windows_processes() cleanup_actions = _terminate_owned_processes(expectation, post_failure_snapshot) deadline = time.monotonic() + 5.0 final_snapshot = capture_windows_processes() while strict_owned_process_ids(expectation, final_snapshot) and time.monotonic() < deadline: time.sleep(0.1) final_snapshot = capture_windows_processes() orphan_pids = strict_owned_process_ids(expectation, final_snapshot) health_created_at = None if health_path.exists(): health_created_at = datetime.fromtimestamp( health_path.stat().st_ctime, tz=timezone.utc, ).isoformat() report: dict[str, Any] = { "schema_version": 1, "mode": "ISOLATED_LAUNCHER_HEALTH_DIAGNOSTIC", "launcher_source": str(Path(launcher.__file__).resolve()), "canonical_release": str(canonical_release.resolve()), "canonical_release_id": release_id, "canonical_agent_version": agent_version, "launcher_pid": launcher_pid, "launcher_executable": sys.executable, "launcher_command_line": list(sys.argv), "popen_pid": popen_pid, "popen_command": captured.get("command"), "popen_created_at": captured.get("popen_created_at"), "launch_nonce": launch_nonce, "launched_at": launched_at, "start_release_result": start_result, "start_release_error": start_error, "health_observed_before_cleanup": health_observed.is_set(), "health_path": str(health_path.resolve()), "health_file_created_at": health_created_at, "health": health, "child_report": child_report, "field_comparison": compare_health_fields(health, expectation), "strict_identity_evaluation": evaluation, "process_tree_at_health": relevant_process_tree( evidence_snapshot, expectation, typed_health_pid, ), "popen_alive_at_evaluation": popen_alive_at_evaluation, "health_process_alive_at_evaluation": health_process_alive, "cleanup_actions": cleanup_actions, "orphan_pids_after_cleanup": orphan_pids, "network_task_polling": False, "hardware_accessed": NO_HARDWARE, "rf_tx": NO_RF_TX, "completed_at": utc_now(), } atomic_write_json(output_directory / "diagnostic_report.json", report) return report def run_probe(arguments: argparse.Namespace) -> int: canonical_release = arguments.canonical_release.resolve() launcher_source = arguments.launcher_source.resolve() output_root = arguments.output_root.resolve() if output_root.exists(): raise FileExistsError(f"diagnostic output root already exists: {output_root}") output_root.mkdir(parents=True) launcher = _load_module(launcher_source, "rf001_diagnostic_launcher") reports = [] for index in range(1, arguments.runs + 1): reports.append( run_probe_once( launcher, canonical_release=canonical_release, output_directory=output_root / f"run_{index:02d}", startup_timeout=arguments.startup_timeout, hold_seconds=arguments.hold_seconds, ) ) nonces = [str(report["launch_nonce"]) for report in reports] summary = { "schema_version": 1, "mode": "ISOLATED_LAUNCHER_HEALTH_DIAGNOSTIC", "runs": reports, "run_count": len(reports), "nonces_unique": len(nonces) == len(set(nonces)), "orphan_pids_after_test": sorted( { pid for report in reports for pid in report["orphan_pids_after_cleanup"] } ), "network_task_polling": False, "hardware_accessed": NO_HARDWARE, "rf_tx": NO_RF_TX, "completed_at": utc_now(), } atomic_write_json(output_root / "diagnostic_summary.json", summary) print(f"DIAGNOSTIC_REPORT={output_root / 'diagnostic_summary.json'}") print(f"RUNS={len(reports)}") print(f"NONCES_UNIQUE={summary['nonces_unique']}") print(f"ORPHAN_PIDS={summary['orphan_pids_after_test']}") print("NETWORK_TASK_POLLING=false") print("HARDWARE_ACCESSED=false") print("RF_TX=0") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) child = subparsers.add_parser("health-child") child.add_argument("--agent-source", required=True, type=Path) child.add_argument("--release-id", required=True) child.add_argument("--launch-nonce", required=True) child.add_argument("--health-file", required=True, type=Path) child.add_argument("--child-report", required=True, type=Path) child.add_argument("--hold-seconds", required=True, type=float) probe = subparsers.add_parser("probe") probe.add_argument("--launcher-source", required=True, type=Path) probe.add_argument("--canonical-release", required=True, type=Path) probe.add_argument("--output-root", required=True, type=Path) probe.add_argument("--runs", type=int, default=2) probe.add_argument("--startup-timeout", type=float, default=2.0) probe.add_argument("--hold-seconds", type=float, default=10.0) return parser def main() -> None: arguments = build_parser().parse_args() if arguments.command == "health-child": raise SystemExit(run_health_child(arguments)) if arguments.runs < 1 or arguments.runs > 4: raise ValueError("runs must be 1 to 4") if arguments.startup_timeout <= 0 or arguments.hold_seconds <= 0: raise ValueError("diagnostic timeouts must be positive") raise SystemExit(run_probe(arguments)) if __name__ == "__main__": main()