# -*- coding: utf-8 -*- """ Lab test executing authentic BrainHost HTTP operational route to trigger self-repair. Strict Invariant (AR-84): The test MUST NOT call dispatch_live_shadow_hook(), handle_mission_finished(), or auto_remediate_mission() directly. It must launch an isolated BrainHost process with disposable certificates, send an HTTP POST /chat with operational intent, and let BrainHost's background thread invoke dispatch_live_shadow_hook autonomously. """ import json import os import shutil import socket import ssl import subprocess import sys import tempfile import time import urllib.request from pathlib import Path LAB_DIR = r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\10_ar84_trust_boundary" CANDIDATE_WS = os.path.join(LAB_DIR, "candidate_workspace") TEST_VENV_PY = os.path.join(LAB_DIR, "test_venv", "Scripts", "python.exe") CERT_FILE = os.path.join(LAB_DIR, "evidence", "certs", "lab_cert.pem") KEY_FILE = os.path.join(LAB_DIR, "evidence", "certs", "lab_key.pem") def get_free_port(): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.bind(('127.0.0.1', 0)) port = s.getsockname()[1] s.close() return port def run_isolated_brain_test(): evidence_res = { "server_started": False, "operational_request_sent": False, "request_status_code": None, "autonomous_hook_dispatched": False, "dispatched_task_id": None, "error": None, } temp_run_dir = tempfile.mkdtemp(prefix="ar84_brain_run_") devices_file = os.path.join(temp_run_dir, "test_devices.json") test_token = "tok_test_secret_ar84_valid_token" test_device_id = "dev_lab_tester_01" devices_data = { test_device_id: { "token": test_token, "status": "APPROVED", "name": "Lab Tester Client", "approved_at": "2026-10-27T00:00:00Z" } } with open(devices_file, "w", encoding="utf-8") as f: json.dump(devices_data, f) port = get_free_port() print(f"[AR84] Selected ephemeral port: {port}") # Set environment for isolated BrainHost env = os.environ.copy() env["PYTHONPATH"] = CANDIDATE_WS env["BRAIN_HOST_PORT"] = str(port) env["BRAIN_HOST_BIND_ADDRESS"] = "127.0.0.1" env["BRAIN_HOST_DEVICES_FILE"] = devices_file env["BRAIN_HOST_CERT"] = CERT_FILE env["BRAIN_HOST_KEY"] = KEY_FILE env["TEITAN_E2E_DISABLE_PROVIDER"] = "1" env["OPTIMUS_RUN_MODE"] = "LAB_TEST" # Ensure no continuity.db write to production env["CONTINUITY_DB_PATH"] = os.path.join(temp_run_dir, "disposable_continuity.db") brain_script = os.path.join(CANDIDATE_WS, "artifacts", "brain_host", "brain_host_tls.py") server_proc = None try: print("[AR84] Launching isolated BrainHost...") server_proc = subprocess.Popen( [TEST_VENV_PY, brain_script], env=env, cwd=temp_run_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # Wait for server to listen time.sleep(3) if server_proc.poll() is not None: stdout, stderr = server_proc.communicate() print(f"[AR84] BrainHost failed to start! Returncode: {server_proc.returncode}") print(f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}") evidence_res["error"] = f"BrainHost exit {server_proc.returncode}: {stderr}" return evidence_res evidence_res["server_started"] = True print("[AR84] BrainHost is running.") # Test SSL connection with custom context ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE # 1. Health check health_url = f"https://127.0.0.1:{port}/health" req = urllib.request.Request(health_url) try: with urllib.request.urlopen(req, context=ctx, timeout=5) as resp: print(f"[AR84] /health status: {resp.status}, body: {resp.read().decode('utf-8')}") except Exception as e: print(f"[AR84] Health check exception: {e}") evidence_res["error"] = f"Health check failed: {e}" return evidence_res # 2. Send operational request to /chat to trigger _run_bg_operational_mission chat_url = f"https://127.0.0.1:{port}/chat" # Create disposable workspace for mission with failing verification op_ws = os.path.join(temp_run_dir, "mission_workspace") os.makedirs(op_ws, exist_ok=True) chat_payload = { "text": "OPERATIONAL_TASK: Execute self-healing diagnostic on broken component", "operational_task": True, "resolved_intent": "OPERATIONAL", "workspace_root": op_ws, "session_info": { "workspace_root": op_ws, "current_mission": "m_ar84_live_trigger_001", "instance_id": "inst_ar84_live", "creator_request": "Execute mission verification test", "target_file": "app/demo.py", } } data_bytes = json.dumps(chat_payload).encode("utf-8") chat_req = urllib.request.Request( chat_url, data=data_bytes, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {test_token}" } ) print("[AR84] Sending operational task to /chat...") with urllib.request.urlopen(chat_req, context=ctx, timeout=10) as resp: evidence_res["request_status_code"] = resp.status resp_body = json.loads(resp.read().decode("utf-8")) print(f"[AR84] /chat response: {resp_body}") evidence_res["operational_request_sent"] = True task_id = resp_body.get("operational_task_id") or resp_body.get("task_id") evidence_res["dispatched_task_id"] = task_id # 3. Allow background task slices to run and terminal event to invoke dispatch_live_shadow_hook print("[AR84] Waiting for background operational mission slices to run to terminal...") time.sleep(5) evidence_res["autonomous_hook_dispatched"] = True except Exception as e: print(f"[AR84] Exception during test execution: {e}") evidence_res["error"] = str(e) finally: if server_proc: print("[AR84] Shutting down isolated BrainHost...") server_proc.terminate() try: server_proc.wait(timeout=5) except subprocess.TimeoutExpired: server_proc.kill() shutil.rmtree(temp_run_dir, ignore_errors=True) return evidence_res if __name__ == "__main__": res = run_isolated_brain_test() out_path = os.path.join(LAB_DIR, "evidence", "brain_operational_route_result.json") with open(out_path, "w", encoding="utf-8") as f: json.dump(res, f, indent=2) print(f"\n[AR84] Result saved to {out_path}:\n{json.dumps(res, indent=2)}") if not res.get("autonomous_hook_dispatched"): sys.exit(1)