from pathlib import Path # 1. Update brain_host_tls.py bh_path = Path(r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\12_ar88_authority_operational_closure\candidate_workspace\artifacts\brain_host\brain_host_tls.py") bh_content = bh_path.read_text(encoding="utf-8") old_hook_normal = """ # Phase 4B: One-way non-blocking Shadow Self-Repair hook if dispatch_live_shadow_hook: try: with OPERATIONAL_TASKS_LOCK: _snap = dict(OPERATIONAL_TASKS.get(tid, {})) if "activity" not in _snap: _snap["activity"] = [] _snap["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_DISPATCHED", "status": term_st, "message": "Live shadow repair hook dispatched autonomously" }) if tid in OPERATIONAL_TASKS: OPERATIONAL_TASKS[tid]["shadow_hook_dispatched"] = True if "activity" not in OPERATIONAL_TASKS[tid]: OPERATIONAL_TASKS[tid]["activity"] = [] OPERATIONAL_TASKS[tid]["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_DISPATCHED", "status": term_st, "message": "Live shadow repair hook dispatched autonomously" })""" new_hook_normal = """ # Phase 4B: One-way non-blocking Shadow Self-Repair hook (AR-88 Multi-State Hook Contract) if dispatch_live_shadow_hook: try: with OPERATIONAL_TASKS_LOCK: _snap = dict(OPERATIONAL_TASKS.get(tid, {})) if "activity" not in _snap: _snap["activity"] = [] _snap["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_REQUESTED", "status": term_st, "message": "Live shadow repair hook requested autonomously" }) if tid in OPERATIONAL_TASKS: OPERATIONAL_TASKS[tid]["shadow_hook_requested"] = True if "activity" not in OPERATIONAL_TASKS[tid]: OPERATIONAL_TASKS[tid]["activity"] = [] OPERATIONAL_TASKS[tid]["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_REQUESTED", "status": term_st, "message": "Live shadow repair hook requested autonomously" })""" old_hook_exc = """ # Phase 4B: One-way non-blocking Shadow Self-Repair hook (exception terminal) if dispatch_live_shadow_hook: try: with OPERATIONAL_TASKS_LOCK: _snap = dict(OPERATIONAL_TASKS.get(tid, {})) if "activity" not in _snap: _snap["activity"] = [] _snap["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_DISPATCHED", "status": "FAILED", "message": "Live shadow repair hook dispatched autonomously on exception" }) if tid in OPERATIONAL_TASKS: OPERATIONAL_TASKS[tid]["shadow_hook_dispatched"] = True if "activity" not in OPERATIONAL_TASKS[tid]: OPERATIONAL_TASKS[tid]["activity"] = [] OPERATIONAL_TASKS[tid]["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_DISPATCHED", "status": "FAILED", "message": "Live shadow repair hook dispatched autonomously on exception" })""" new_hook_exc = """ # Phase 4B: One-way non-blocking Shadow Self-Repair hook (exception terminal, AR-88) if dispatch_live_shadow_hook: try: with OPERATIONAL_TASKS_LOCK: _snap = dict(OPERATIONAL_TASKS.get(tid, {})) if "activity" not in _snap: _snap["activity"] = [] _snap["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_REQUESTED", "status": "FAILED", "message": "Live shadow repair hook requested on terminal exception" }) if tid in OPERATIONAL_TASKS: OPERATIONAL_TASKS[tid]["shadow_hook_requested"] = True if "activity" not in OPERATIONAL_TASKS[tid]: OPERATIONAL_TASKS[tid]["activity"] = [] OPERATIONAL_TASKS[tid]["activity"].append({ "timestamp": time.time(), "event": "SHADOW_HOOK_REQUESTED", "status": "FAILED", "message": "Live shadow repair hook requested on terminal exception" })""" assert old_hook_normal in bh_content, "old_hook_normal not found in brain_host_tls.py" bh_content = bh_content.replace(old_hook_normal, new_hook_normal, 1) assert old_hook_exc in bh_content, "old_hook_exc not found in brain_host_tls.py" bh_content = bh_content.replace(old_hook_exc, new_hook_exc, 1) bh_path.write_text(bh_content, encoding="utf-8") print("SUCCESS: Updated brain_host_tls.py with SHADOW_HOOK_REQUESTED!") # 2. Update live_shadow_hook.py to emit SHADOW_HOOK_ACCEPTED, SHADOW_HOOK_STARTED, SHADOW_HOOK_COMPLETED / FAILED lsh_path = Path(r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\12_ar88_authority_operational_closure\candidate_workspace\app\remediation\self_repair\live_shadow_hook.py") lsh_content = lsh_path.read_text(encoding="utf-8") # In dispatch_live_shadow_hook: after snapshot building and lock acquisition, update task activity with SHADOW_HOOK_ACCEPTED and SHADOW_HOOK_STARTED old_dispatch_entry = """ orch = orchestrator or get_default_shadow_orchestrator( project_root=snapshot.workspace_root or None )""" new_dispatch_entry = """ orch = orchestrator or get_default_shadow_orchestrator( project_root=snapshot.workspace_root or None ) # AR-88 State Transition: Record SHADOW_HOOK_ACCEPTED in task state tid = terminal_rec.get("task_id") or snapshot.mission_id try: default_canonical_task_state_updater({ "task_id": tid, "mission_id": snapshot.mission_id, "activity_event": "SHADOW_HOOK_ACCEPTED", "message": "Live shadow hook accepted by engine", }) except Exception: pass""" assert old_dispatch_entry in lsh_content, "old_dispatch_entry not found in live_shadow_hook.py" lsh_content = lsh_content.replace(old_dispatch_entry, new_dispatch_entry, 1) # In _async_worker(): emit SHADOW_HOOK_STARTED, then on completion emit SHADOW_HOOK_COMPLETED or SHADOW_HOOK_FAILED old_async_worker = """ if async_dispatch: def _async_worker(): try: rep = orch.handle_mission_finished( snapshot, execute_resume_if_eligible=effective_execute_resume, allow_auto_promotion=effective_allow_promotion, mission_runner_callable=effective_runner, final_verifier_callable=effective_verifier, ) _post_execution_sync(rep) except Exception as ex_worker: logger.exception( f"[LiveShadowHook] Async shadow execution failed for mission {snapshot.mission_id}: {ex_worker}" ) finally: _cleanup_scoped_bridges()""" new_async_worker = """ if async_dispatch: def _async_worker(): try: # AR-88 State Transition: SHADOW_HOOK_STARTED try: default_canonical_task_state_updater({ "task_id": tid, "mission_id": snapshot.mission_id, "activity_event": "SHADOW_HOOK_STARTED", "message": "Live shadow repair execution started", }) except Exception: pass rep = orch.handle_mission_finished( snapshot, execute_resume_if_eligible=effective_execute_resume, allow_auto_promotion=effective_allow_promotion, mission_runner_callable=effective_runner, final_verifier_callable=effective_verifier, ) _post_execution_sync(rep) # AR-88 State Transition: SHADOW_HOOK_COMPLETED try: default_canonical_task_state_updater({ "task_id": tid, "mission_id": snapshot.mission_id, "activity_event": "SHADOW_HOOK_COMPLETED", "status": getattr(rep, "shadow_status", "UNKNOWN"), "incident_id": getattr(rep, "incident_id", None), "shadow_run_id": getattr(rep, "shadow_run_id", None), "message": "Live shadow repair cycle completed", }) except Exception: pass except Exception as ex_worker: logger.exception( f"[LiveShadowHook] Async shadow execution failed for mission {snapshot.mission_id}: {ex_worker}" ) # AR-88 State Transition: SHADOW_HOOK_FAILED try: default_canonical_task_state_updater({ "task_id": tid, "mission_id": snapshot.mission_id, "activity_event": "SHADOW_HOOK_FAILED", "error": str(ex_worker), "message": f"Live shadow repair failed with exception: {ex_worker}", }) except Exception: pass finally: _cleanup_scoped_bridges()""" assert old_async_worker in lsh_content, "old_async_worker not found in live_shadow_hook.py" lsh_content = lsh_content.replace(old_async_worker, new_async_worker, 1) # In default_canonical_task_state_updater, handle activity_event old_updater_body = """ activities = t_entry.setdefault("activity", []) if isinstance(activities, list): for act_msg in ("SELF_REPAIR_PROMOTED", "MISSION_RESUMED", "FINAL_VERIFICATION_PASS"): if act_msg not in activities: activities.append(act_msg)""" new_updater_body = """ activities = t_entry.setdefault("activity", []) if isinstance(activities, list): act_ev = update_payload.get("activity_event") if act_ev: activities.append({ "timestamp": time.time(), "event": act_ev, "status": update_payload.get("status") or t_entry.get("status"), "message": update_payload.get("message") or act_ev, "incident_id": update_payload.get("incident_id"), "shadow_run_id": update_payload.get("shadow_run_id"), }) t_entry[act_ev.lower()] = True for act_msg in ("SELF_REPAIR_PROMOTED", "MISSION_RESUMED", "FINAL_VERIFICATION_PASS"): if act_msg not in activities: activities.append(act_msg)""" assert old_updater_body in lsh_content, "old_updater_body not found in live_shadow_hook.py" lsh_content = lsh_content.replace(old_updater_body, new_updater_body, 1) lsh_path.write_text(lsh_content, encoding="utf-8") print("SUCCESS: Updated live_shadow_hook.py with AR-88 5-state lifecycle transitions!")