path = r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\10_ar84_trust_boundary\candidate_workspace\app\remediation\self_repair\shadow_orchestrator.py" with open(path, "r", encoding="utf-8") as f: content = f.read() # 1. Update reconcile_telemetry to use verify_authenticity old_reconcile_audit = ''' # AR-82 Authenticated Provenance Gate: # A claim of FALSE_VERIFIED must be backed by an authentic, signed TelemetryAuditRecord # when require_authenticated_provenance is True (e.g. during promotion evaluation or untrusted payload checks). audit_rec = res.get("telemetry_audit") or getattr(snapshot, "telemetry_audit", None) has_valid_audit = False if audit_rec is not None: if isinstance(audit_rec, dict): expected_sig = audit_rec.get("evidence_signature") observer_act = bool(audit_rec.get("observer_active")) req_scopes = {"URLLIB_HTTP", "RAW_SOCKET"} mon_scopes = set(audit_rec.get("monitored_scopes") or []) payload = f"{audit_rec.get('mission_id')}:{audit_rec.get('instance_id')}:{audit_rec.get('provider_status')}:{audit_rec.get('network_status')}:{audit_rec.get('detected_attempts_count', 0)}:{audit_rec.get('observer_active')}" computed_sig = hashlib.sha256(payload.encode("utf-8")).hexdigest().upper() if expected_sig and computed_sig == expected_sig and observer_act and req_scopes.issubset(mon_scopes): has_valid_audit = True elif hasattr(audit_rec, "evidence_signature") and hasattr(audit_rec, "compute_signature"): if audit_rec.evidence_signature == audit_rec.compute_signature() and audit_rec.observer_active: req_scopes = {"URLLIB_HTTP", "RAW_SOCKET"} if req_scopes.issubset(set(getattr(audit_rec, "monitored_scopes", []))): has_valid_audit = True''' new_reconcile_audit = ''' # AR-84 Cryptographic Authenticated Provenance Gate: # A claim of FALSE_VERIFIED must be backed by an authentic, signed TelemetryAuditRecord # verified with supervisor HMAC key and strict bindings. audit_rec = res.get("telemetry_audit") or getattr(snapshot, "telemetry_audit", None) has_valid_audit = False if audit_rec is not None: ws_expected = getattr(snapshot, "workspace_root", None) or self.project_root m_expected = getattr(snapshot, "mission_id", None) if hasattr(audit_rec, "verify_authenticity"): is_valid, _ = audit_rec.verify_authenticity( expected_mission_id=m_expected, expected_workspace_root=ws_expected, enforce_replay_guard=False, ) if is_valid: has_valid_audit = True elif isinstance(audit_rec, dict): try: from app.remediation.self_repair.telemetry_contract import TelemetryAuditRecord rec = TelemetryAuditRecord( mission_id=str(audit_rec.get("mission_id") or ""), instance_id=str(audit_rec.get("instance_id") or ""), workspace_root=str(audit_rec.get("workspace_root") or ""), interval_start=str(audit_rec.get("interval_start") or ""), interval_end=str(audit_rec.get("interval_end") or ""), observer_active=bool(audit_rec.get("observer_active")), observer_fault=audit_rec.get("observer_fault"), monitored_scopes=list(audit_rec.get("monitored_scopes") or []), unmonitored_scopes=list(audit_rec.get("unmonitored_scopes") or []), detected_attempts=list(audit_rec.get("detected_attempts") or []), detected_attempts_count=int(audit_rec.get("detected_attempts_count") or 0), provider_status=str(audit_rec.get("provider_status") or ""), network_status=str(audit_rec.get("network_status") or ""), promotion_eligible=bool(audit_rec.get("promotion_eligible")), audit_verdict=str(audit_rec.get("audit_verdict") or ""), process_id=int(audit_rec.get("process_id") or 0), nonce=str(audit_rec.get("nonce") or ""), evidence_signature=str(audit_rec.get("evidence_signature") or ""), ) is_valid, _ = rec.verify_authenticity( expected_mission_id=m_expected, expected_workspace_root=ws_expected, enforce_replay_guard=False, ) if is_valid: has_valid_audit = True except Exception: has_valid_audit = False''' if old_reconcile_audit in content: content = content.replace(old_reconcile_audit, new_reconcile_audit, 1) print("Replaced reconcile_telemetry audit check") else: print("Failed to replace reconcile_telemetry audit check") # 2. Update promotion_executor call to pass telemetry_audit old_call = ''' effective_repair_result = self.promotion_executor.execute_promotion( target_file=target_file, patch_candidate=patch_plan_obj or promo_proposal_obj, workspace_root=ws_root, confidence=str(actual_confidence), root_cause_proven=True, sandbox_verdict=cand_verdict_obj, focused_tests_pass=bool(cand_verdict_obj.focused_tests_pass and verif_obj.focused_tests_pass), regression_tests_pass=bool(cand_verdict_obj.regression_tests_pass and verif_obj.regression_tests_pass), evidence_comparison_pass=bool(comp_obj.new_failures == 0 and len(comp_obj.regressions) == 0), rollback_ready=True, provider_called=prov_val, network_used=net_val, attempt_number=attempt_count, is_duplicate=is_dup, max_attempts=self.max_repair_attempts, incident_id=inc_id, rollback_plan=promo_proposal_obj.rollback_plan if promo_proposal_obj else None, )''' new_call = ''' audit_obj = res.get("telemetry_audit") or getattr(snapshot, "telemetry_audit", None) effective_repair_result = self.promotion_executor.execute_promotion( target_file=target_file, patch_candidate=patch_plan_obj or promo_proposal_obj, workspace_root=ws_root, confidence=str(actual_confidence), root_cause_proven=True, sandbox_verdict=cand_verdict_obj, focused_tests_pass=bool(cand_verdict_obj.focused_tests_pass and verif_obj.focused_tests_pass), regression_tests_pass=bool(cand_verdict_obj.regression_tests_pass and verif_obj.regression_tests_pass), evidence_comparison_pass=bool(comp_obj.new_failures == 0 and len(comp_obj.regressions) == 0), rollback_ready=True, provider_called=prov_val, network_used=net_val, attempt_number=attempt_count, is_duplicate=is_dup, max_attempts=self.max_repair_attempts, incident_id=inc_id, rollback_plan=promo_proposal_obj.rollback_plan if promo_proposal_obj else None, telemetry_audit=audit_obj, require_authenticated_provenance=True, )''' if old_call in content: content = content.replace(old_call, new_call, 1) print("Replaced promotion_executor call") else: print("Failed to replace promotion_executor call") with open(path, "w", encoding="utf-8") as f: f.write(content) print("Finished updating shadow_orchestrator.py")