# -*- coding: utf-8 -*- """ AR-88 Operational BrainHook & Correlated Engine Verification Test Strictly complies with AR-88 Requirements 4 & 5: 1. Strict environment whitelist: No production keys, DPAPI, auth tokens or prod paths leaked. 2. Isolated BrainHost process on ephemeral port (not 8443). 3. Sends authentic HTTP POST /chat with operational task intent. 4. Tracks explicit 5-state lifecycle transitions: - SHADOW_HOOK_REQUESTED - SHADOW_HOOK_ACCEPTED - SHADOW_HOOK_STARTED - SHADOW_HOOK_COMPLETED / FAILED 5. Verifies correlation across task_id, mission_id, incident_id, shadow_run_id, and diagnostic report. """ 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 = Path(__file__).resolve().parent CANDIDATE_WS = str(LAB_DIR / "candidate_workspace") TEST_VENV_PY = str(LAB_DIR / "test_venv" / "Scripts" / "python.exe") CERT_FILE = str(LAB_DIR / "evidence" / "certs" / "lab_cert.pem") KEY_FILE = str(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 build_whitelisted_test_env(temp_run_dir: str, port: int, devices_file: str) -> dict: """ AR-88 Requirement 5: Strict minimal environment whitelist. Prevents leaking credentials, production paths, DPAPI, external tokens. """ whitelist_vars = [ "SystemRoot", "COMSPEC", "PATH", "TEMP", "TMP", "PATHEXT", "SystemDrive" ] env = {k: os.environ[k] for k in whitelist_vars if k in os.environ} env["PYTHONPATH"] = CANDIDATE_WS env["PYTHONNOUSERSITE"] = "1" 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["TEITAN_DISABLE_PROVIDER"] = "1" env["OPTIMUS_RUN_MODE"] = "LAB_TEST" env["OPTIMUS_AUTHORITY_ROLE"] = "LAB_SUPERVISOR" env["CONTINUITY_DB_PATH"] = os.path.join(temp_run_dir, "disposable_continuity.db") return env def run_isolated_brain_test(): evidence_res = { "server_started": False, "operational_request_sent": False, "request_status_code": None, "dispatched_task_id": None, "task_terminal_status": None, "lifecycle_states_detected": [], "shadow_hook_requested": False, "shadow_hook_accepted": False, "shadow_hook_started": False, "shadow_hook_completed_or_failed": False, "correlated_incident_id": None, "correlated_shadow_run_id": None, "shadow_repair_report_found": False, "error": None, } temp_run_dir = tempfile.mkdtemp(prefix="ar88_brain_run_") devices_file = os.path.join(temp_run_dir, "test_devices.json") test_token = "tok_test_secret_ar88_valid_token" test_device_id = "dev_lab_tester_88" devices_data = { test_device_id: { "token": test_token, "status": "APPROVED", "name": "Lab Tester Client AR88", "approved_at": "2026-10-28T00:00:00Z" } } with open(devices_file, "w", encoding="utf-8") as f: json.dump(devices_data, f) port = get_free_port() assert port != 8443, "Security check: must NOT use production port 8443" print(f"[AR88] Ephemeral port: {port}") env = build_whitelisted_test_env(temp_run_dir, port, devices_file) brain_script = os.path.join(CANDIDATE_WS, "artifacts", "brain_host", "brain_host_tls.py") server_proc = None try: print("[AR88] Launching isolated BrainHost with whitelisted environment...") server_log_out = open(os.path.join(temp_run_dir, "brain_server.log"), "w", encoding="utf-8") server_proc = subprocess.Popen( [TEST_VENV_PY, "-u", brain_script], env=env, cwd=temp_run_dir, stdout=server_log_out, stderr=subprocess.STDOUT, text=True ) time.sleep(3) if server_proc.poll() is not None: evidence_res["error"] = f"BrainHost exit code {server_proc.returncode}" return evidence_res evidence_res["server_started"] = True 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) with urllib.request.urlopen(req, context=ctx, timeout=5) as resp: print(f"[AR88] /health check: {resp.status}") # 2. Dispatch operational task via POST /chat chat_url = f"https://127.0.0.1:{port}/chat" 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_ar88_live_trigger_001", "instance_id": "inst_ar88_live", "creator_request": "Execute mission verification test", "target_file": "app/demo.py", } } chat_req = urllib.request.Request( chat_url, data=json.dumps(chat_payload).encode("utf-8"), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {test_token}" } ) print("[AR88] 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"[AR88] /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. Poll /developer/tasks/ until terminal status and all lifecycle events recorded task_url = f"https://127.0.0.1:{port}/developer/tasks/{task_id}" poll_req = urllib.request.Request( task_url, headers={"Authorization": f"Bearer {test_token}"} ) print(f"[AR88] Polling task endpoint {task_url} for terminal status and lifecycle events...") terminal_status_reached = False for attempt in range(16): time.sleep(0.5) try: with urllib.request.urlopen(poll_req, context=ctx, timeout=3) as t_resp: t_data = json.loads(t_resp.read().decode("utf-8")) task_info = t_data.get("task") or {} curr_st = task_info.get("task_status") or task_info.get("status") activities = task_info.get("activity") or [] events_in_act = [] for act in activities: if isinstance(act, dict): ev = act.get("event") if ev: events_in_act.append(ev) if ev == "SHADOW_HOOK_REQUESTED": evidence_res["shadow_hook_requested"] = True elif ev == "SHADOW_HOOK_ACCEPTED": evidence_res["shadow_hook_accepted"] = True elif ev == "SHADOW_HOOK_STARTED": evidence_res["shadow_hook_started"] = True elif ev in ("SHADOW_HOOK_COMPLETED", "SHADOW_HOOK_FAILED"): evidence_res["shadow_hook_completed_or_failed"] = True if act.get("incident_id"): evidence_res["correlated_incident_id"] = act.get("incident_id") if act.get("shadow_run_id"): evidence_res["correlated_shadow_run_id"] = act.get("shadow_run_id") evidence_res["lifecycle_states_detected"] = list(set(events_in_act)) print(f"[AR88] Poll {attempt+1}: status={curr_st} events={events_in_act}") if curr_st in ("FAILED", "SUCCESS", "COMPLETED", "BLOCKED"): evidence_res["task_terminal_status"] = curr_st terminal_status_reached = True if evidence_res["shadow_hook_requested"] and evidence_res["shadow_hook_accepted"]: break except Exception as poll_ex: print(f"[AR88] Poll attempt {attempt+1} exception: {poll_ex}") # Check for correlated shadow repair report in reports directory reports_dir = os.path.join(CANDIDATE_WS, "reports", "self_repair") if os.path.isdir(reports_dir): for inc in os.listdir(reports_dir): rep_f = os.path.join(reports_dir, inc, "shadow_repair_report.json") if os.path.isfile(rep_f): evidence_res["shadow_repair_report_found"] = True break except Exception as e: print(f"[AR88] Exception during test execution: {e}") evidence_res["error"] = str(e) finally: if server_proc: print("[AR88] Shutting down isolated BrainHost...") server_proc.terminate() try: server_proc.wait(timeout=5) except subprocess.TimeoutExpired: server_proc.kill() try: server_log_out.close() log_p = os.path.join(temp_run_dir, "brain_server.log") if os.path.exists(log_p): with open(log_p, "r", encoding="utf-8", errors="ignore") as lf: lines = lf.readlines() print(f"[AR88] BrainHost server log tail ({len(lines)} lines):") for l in lines[-15:]: print(" [SRV]", l.rstrip()) except Exception as e_log: print(f"[AR88] Error reading server log: {e_log}") shutil.rmtree(temp_run_dir, ignore_errors=True) return evidence_res if __name__ == "__main__": res = run_isolated_brain_test() out_path = os.path.join(str(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[AR88] Result saved to {out_path}:\n{json.dumps(res, indent=2)}") if not (res.get("shadow_hook_requested") and res.get("shadow_hook_accepted")): sys.exit(1)