# -*- coding: utf-8 -*- """ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ QSAP INTEGRATION MODULE ║ ║ quasar_main4.py ←→ quasar_selective_attention.py ║ ║ ║ ║ Responsibilities ║ ║ ─────────────── ║ ║ 1. TRAINING BLOCK — monkey-patches every training entry-point in ║ ║ quasar_main4.py to no-op while QSAP is active. Any of the four ║ ║ training paths (AVN continuous thread, AVN._train, AsyncTraining ║ ║ Executor, QuantumSystemTrainer.train_step) will log a suppression ║ ║ message and return immediately without gradient steps. ║ ║ ║ ║ 2. QSAP INSTALL — primary path calls install_qsap_v4() which wires ║ ║ the full enforcement stack (FrozenTransaction, LatestStateLatch, ║ ║ ShardedInferenceEngine, hard-block on _process_latest_features). ║ ║ If install_qsap_v4 is unavailable or raises AT RUNTIME (e.g. a ║ ║ transitive import like torchvision is missing), we fall through to ║ ║ install_qsap (v3) with a loud warning banner so operators know ║ ║ enforcement is NOT active. Specifically: ║ ║ a. Registers MessageContext in the QSAP module namespace so ║ ║ _qsap_on_feature_message can find it via sys.modules scan. ║ ║ b. Confirms quantum_bridge is fully initialised before handing it ║ ║ to the QSAP engine. ║ ║ c. Tries install_qsap_v4(system) first — on ANY exception ║ ║ (ImportError, RuntimeError, missing transitive dep), logs ║ ║ the failure and falls through to install_qsap(system). This ║ ║ robustness was ADDED in v2.1.0 after a production incident ║ ║ where v4 raised `No module named 'torchvision'` at runtime and ║ ║ the previous catch-ImportError-only handler returned False ║ ║ without trying the v3 fallback, leaving the engine with no ║ ║ QSAP install at all. ║ ║ d. Stores the engine handle on system._qsap_engine for diagnostics. ║ ║ e. Sets system._qsap_enforcement_active = True iff v4 succeeded ║ ║ (so the qsap_status() helper can report whether inference is ║ ║ truly routed through ShardedInferenceEngine). ║ ║ ║ ║ 3. HEALTH PRINTER — background thread logs print_qsap_health() every ║ ║ QSAP_HEALTH_INTERVAL_S seconds so you can see drop rates and ║ ║ freshness deltas in the console while running. ║ ║ ║ ║ Usage (in quasar_main4.py) ║ ║ ────────────────────────── ║ ║ Step A — import (place with other try/except module imports): ║ ║ ║ ║ try: ║ ║ from qsap_integration import ( ║ ║ QSAP_TRAINING_BLOCKED, ║ ║ apply_qsap_training_block, ║ ║ apply_qsap_integration, ║ ║ ) ║ ║ QSAP_INTEGRATION_AVAILABLE = True ║ ║ except ImportError as _e: ║ ║ QSAP_INTEGRATION_AVAILABLE = False ║ ║ print(f"⚠️ [QSAP] qsap_integration not found: {_e}") ║ ║ ║ ║ Step B — apply training block BEFORE AVNTrainer is first instantiated ║ ║ (place right before _init_avn_system() / IntegratedSignal ║ ║ System construction so the continuous-training thread is never ║ ║ started in the first place): ║ ║ ║ ║ if QSAP_INTEGRATION_AVAILABLE: ║ ║ apply_qsap_training_block() ║ ║ ║ ║ Step C — install QSAP onto the live system object (place AFTER ║ ║ install_state_fix(system) and AFTER quantum_bridge is ready): ║ ║ ║ ║ if QSAP_INTEGRATION_AVAILABLE: ║ ║ apply_qsap_integration(system) ║ ║ ║ ║ To re-BLOCK training later (e.g. for debugging or incident response): ║ ║ import qsap_integration ║ ║ qsap_integration.QSAP_TRAINING_BLOCKED = True ║ ║ ║ ║ Note: As of 2026-04-18, training is UNBLOCKED by default. The rewards- ║ ║ channel metadata transport fix made the RL reward loop live ║ ║ end-to-end, so the paranoid-safe default no longer applies. ║ ║ ║ ╚══════════════════════════════════════════════════════════════════════════════╝ Author : ENG Karl + QSAP integration layer (2026-04) Version : 2.1.0 — v4 enforcement primary, v3 graceful fallback on ANY runtime exception (not just ImportError). Prevents the failure mode where v4 raised mid-init for a missing transitive dep and left the engine with no QSAP at all. """ from __future__ import annotations import sys import time import threading import logging from typing import Optional # ── PATCH 1 (qsap-latch-deadlock-fix): torch must be in scope before # install_qsap_v4() is called. Without this, any torch reference inside the # call chain raises NameError: name 'torch' is not defined, silently falling # back to v3 and leaving _process_latest_features as a live bypass path. try: import torch # noqa: F401 — used by install_qsap_v4 internals import torch.nn as nn # noqa: F401 except ImportError: torch = None # type: ignore nn = None # type: ignore logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # §1 GLOBAL TRAINING-BLOCK FLAG # # Single source of truth. All monkey-patched training methods check this flag # at entry and return immediately when True. # # Default is False (training UNBLOCKED) as of 2026-04-18 — the rewards-channel # metadata transport fix made the RL reward loop live end-to-end, so the # paranoid-safe default no longer applies. Patches A–D are still installed at # startup; they simply pass through to their _original_* implementations while # this flag is False. # # Set to True at runtime to re-block training without a restart (e.g. for # debugging, dry runs, or incident response): # import qsap_integration; qsap_integration.QSAP_TRAINING_BLOCKED = True # ───────────────────────────────────────────────────────────────────────────── QSAP_TRAINING_BLOCKED: bool = False # How often (seconds) to print QSAP health to console QSAP_HEALTH_INTERVAL_S: float = 120.0 # ───────────────────────────────────────────────────────────────────────────── # §2 TRAINING BLOCK — monkey-patches # # Four entry-points are patched: # # A. AVNTrainer.start_continuous_training() # Called in AVNTrainer.__init__ when continuous_training=True. # Patched to no-op so the background training thread is never started. # # B. AVNTrainer._train() # The actual gradient-step method. Patched to early-return so that # any path that calls _train (including any residual threads) is safe. # # C. AsyncTrainingExecutor.submit_training() # The non-blocking executor used by IntegratedSignalSystem. # Patched to log and skip. # # D. QuantumSystemTrainer.train_step() # The TD3 / quantum trainer step. Patched to return None silently. # # All originals are saved as _original_* so the block is reversible. # ───────────────────────────────────────────────────────────────────────────── _training_block_applied: bool = False def apply_qsap_training_block() -> None: """ Monkey-patch all training entry-points to no-op while QSAP_TRAINING_BLOCKED is True. Safe to call multiple times — subsequent calls are idempotent. Must be called BEFORE AVNTrainer is first instantiated so that the continuous-training thread is never launched. """ global _training_block_applied if _training_block_applied: return # ── A. AVNTrainer.start_continuous_training ──────────────────────────── _patch_avn_start_continuous_training() # ── B. AVNTrainer._train ─────────────────────────────────────────────── _patch_avn_train() # ── C. AsyncTrainingExecutor.submit_training ─────────────────────────── _patch_async_training_executor() # ── D. QuantumSystemTrainer.train_step ──────────────────────────────── _patch_quantum_system_trainer() _training_block_applied = True print() if QSAP_TRAINING_BLOCKED: print("╔══════════════════════════════════════════════════════════════════════╗") print("║ QSAP TRAINING BLOCK APPLIED (v1.0.0) ║") print("╠══════════════════════════════════════════════════════════════════════╣") print("║ All training entry-points monkey-patched: ║") print("║ A. AVNTrainer.start_continuous_training → no-op ║") print("║ B. AVNTrainer._train → early-return ║") print("║ C. AsyncTrainingExecutor.submit_training → skip ║") print("║ D. QuantumSystemTrainer.train_step → return None ║") print("║ ║") print("║ To re-enable: import qsap_integration ║") print("║ qsap_integration.QSAP_TRAINING_BLOCKED = False ║") print("╚══════════════════════════════════════════════════════════════════════╝") else: print("╔══════════════════════════════════════════════════════════════════════╗") print("║ QSAP TRAINING PATCHES INSTALLED — TRAINING UNBLOCKED (v1.0.0) ║") print("╠══════════════════════════════════════════════════════════════════════╣") print("║ All training entry-points wrapped (pass-through while flag=False): ║") print("║ A. AVNTrainer.start_continuous_training → passes through ║") print("║ B. AVNTrainer._train → passes through ║") print("║ C. AsyncTrainingExecutor.submit_training → passes through ║") print("║ D. QuantumSystemTrainer.train_step → passes through ║") print("║ ║") print("║ To re-block: import qsap_integration ║") print("║ qsap_integration.QSAP_TRAINING_BLOCKED = True ║") print("╚══════════════════════════════════════════════════════════════════════╝") print() # ── Patch helpers ───────────────────────────────────────────────────────────── def _patch_avn_start_continuous_training() -> None: """Patch A: AVNTrainer.start_continuous_training → no-op.""" _main = sys.modules.get("__main__") or sys.modules.get("quasar_main4") AVNTrainer = getattr(_main, "AVNTrainer", None) if AVNTrainer is None: logger.warning("[QSAP-Block] AVNTrainer not found in __main__ — patch A skipped") return _orig = AVNTrainer.start_continuous_training def _blocked_start_continuous_training(self_avn): import qsap_integration as _qi if _qi.QSAP_TRAINING_BLOCKED: print( f"[{time.strftime('%H:%M:%S')}] 🔒 [QSAP-Block] " f"AVNTrainer.start_continuous_training suppressed " f"(QSAP_TRAINING_BLOCKED=True)" ) return # no-op — thread never starts return _orig(self_avn) AVNTrainer.start_continuous_training = _blocked_start_continuous_training AVNTrainer._original_start_continuous_training = _orig logger.debug("[QSAP-Block] Patch A applied: AVNTrainer.start_continuous_training") def _patch_avn_train() -> None: """Patch B: AVNTrainer._train → early-return when blocked.""" _main = sys.modules.get("__main__") or sys.modules.get("quasar_main4") AVNTrainer = getattr(_main, "AVNTrainer", None) if AVNTrainer is None: logger.warning("[QSAP-Block] AVNTrainer not found — patch B skipped") return _orig = AVNTrainer._train def _blocked_train(self_avn): import qsap_integration as _qi if _qi.QSAP_TRAINING_BLOCKED: # Only log once every 60 s so console stays clean _now = time.time() if _now - getattr(self_avn, "_qsap_block_last_log", 0) > 60: print( f"[{time.strftime('%H:%M:%S')}] 🔒 [QSAP-Block] " f"AVNTrainer._train suppressed (QSAP_TRAINING_BLOCKED=True)" ) self_avn._qsap_block_last_log = _now return None return _orig(self_avn) AVNTrainer._train = _blocked_train AVNTrainer._original_train = _orig logger.debug("[QSAP-Block] Patch B applied: AVNTrainer._train") def _patch_async_training_executor() -> None: """Patch C: AsyncTrainingExecutor.submit_training → skip when blocked.""" _main = sys.modules.get("__main__") or sys.modules.get("quasar_main4") AsyncTrainingExecutor = getattr(_main, "AsyncTrainingExecutor", None) if AsyncTrainingExecutor is None: logger.warning("[QSAP-Block] AsyncTrainingExecutor not found — patch C skipped") return _orig = AsyncTrainingExecutor.submit_training # ── §P2-fix-17 (2026-04-19): Wrapper signature alignment ──────────── # The real AsyncTrainingExecutor.submit_training signature is: # submit_training(self, trainer, buffer, batch_size=32, training_number=0) # It has NO `callback` parameter. The previous wrapper accepted # `callback=None` as a 4th keyword arg and then forwarded it positionally # to _orig as: # _orig(self_exec, trainer, buffer, batch_size, callback, **kwargs) # Since _orig's 5th positional is `training_number`, callback (None) got # bound to training_number. Then **kwargs ALSO carried training_number # from the real caller (quasar_main4.py:L34534 passes training_number as # a keyword), producing: # TypeError: submit_training() got multiple values for argument 'training_number' # Observed impact: 58 TypeErrors in the error log, zero successful # training submits, Training Steps frozen for the entire run. Every # single 🎓🎓🎓🎓 pre-submit banner was immediately followed by this # crash — the body of _run_training_safe never executed. # Fix: mirror the real signature exactly, forward all arguments by name. def _blocked_submit_training(self_exec, trainer, buffer, batch_size=32, training_number=0, **kwargs): import qsap_integration as _qi if _qi.QSAP_TRAINING_BLOCKED: _now = time.time() if _now - getattr(self_exec, "_qsap_block_last_log", 0) > 60: print( f"[{time.strftime('%H:%M:%S')}] 🔒 [QSAP-Block] " f"AsyncTrainingExecutor.submit_training skipped " f"(QSAP_TRAINING_BLOCKED=True)" ) self_exec._qsap_block_last_log = _now return None return _orig(self_exec, trainer, buffer, batch_size=batch_size, training_number=training_number, **kwargs) AsyncTrainingExecutor.submit_training = _blocked_submit_training AsyncTrainingExecutor._original_submit_training = _orig logger.debug("[QSAP-Block] Patch C applied: AsyncTrainingExecutor.submit_training") def _patch_quantum_system_trainer() -> None: """Patch D: QuantumSystemTrainer.train_step → return None when blocked.""" _main = sys.modules.get("__main__") or sys.modules.get("quasar_main4") QuantumSystemTrainer = getattr(_main, "QuantumSystemTrainer", None) if QuantumSystemTrainer is None: logger.warning("[QSAP-Block] QuantumSystemTrainer not found — patch D skipped") return _orig = QuantumSystemTrainer.train_step def _blocked_train_step(self_qt, batch_size=None): import qsap_integration as _qi if _qi.QSAP_TRAINING_BLOCKED: _now = time.time() if _now - getattr(self_qt, "_qsap_block_last_log", 0) > 60: print( f"[{time.strftime('%H:%M:%S')}] 🔒 [QSAP-Block] " f"QuantumSystemTrainer.train_step suppressed " f"(QSAP_TRAINING_BLOCKED=True)" ) self_qt._qsap_block_last_log = _now return None return _orig(self_qt, batch_size) QuantumSystemTrainer.train_step = _blocked_train_step QuantumSystemTrainer._original_train_step = _orig logger.debug("[QSAP-Block] Patch D applied: QuantumSystemTrainer.train_step") def stop_existing_avn_training_thread(system) -> None: """ If the AVNTrainer continuous-training thread is already running (because the block was applied after AVNTrainer.__init__), signal it to stop. Signals the thread; does NOT join — the training loop checks _continuous_training_active every train_interval_seconds. """ avn_trainer = getattr(system, "avn_trainer", None) if avn_trainer is None: return was_active = getattr(avn_trainer, "_continuous_training_active", False) if was_active: avn_trainer._continuous_training_active = False print( f"[{time.strftime('%H:%M:%S')}] 🔒 [QSAP-Block] " f"AVNTrainer continuous-training thread signalled to stop" ) # ───────────────────────────────────────────────────────────────────────────── # §3 QSAP INSTALL # ───────────────────────────────────────────────────────────────────────────── def apply_qsap_integration( system, alpha_threshold: float = 0.02, tau_min_ms: float = 50.0, budget_slots: Optional[int] = None, commit_wait_max_ms: float = 150.0, abstain_threshold: float = 0.1, ) -> bool: """ Install the QSAP perception-action loop onto a live IntegratedSignalSystem. Pre-conditions (callers must guarantee): 1. install_state_fix(system) has already run — _aoi_tracker attached. 2. system.quantum_bridge is fully initialised (not None). 3. system.agents is populated (non-empty dict). 4. ably_manager is attached and subscriptions are active (so the patched on_feature_message starts receiving real ticks immediately). Post-conditions: • system.on_feature_message → _qsap_on_feature_message • system._process_latest_features → legacy (callable as fallback) • system._qsap_engine → SelectiveBindingEngine handle • system.training_enabled → False (belt-and-suspenders block) • Background health-printer thread started (logs every QSAP_HEALTH_INTERVAL_S seconds) Returns True on success, False if QSAP module unavailable or quantum_bridge not ready. """ # ── Pre-flight checks ───────────────────────────────────────────────── if system is None: print("[QSAP-Install] ❌ system is None — aborting") return False if not getattr(system, "agents", None): print("[QSAP-Install] ❌ system.agents is empty — aborting") return False if not getattr(system, "quantum_bridge", None): print("[QSAP-Install] ❌ system.quantum_bridge is None — aborting") print(" QSAP needs quantum_bridge for per-agent inference.") return False # ── Import QSAP module ──────────────────────────────────────────────── # # We need two installers: the v3 primitive (install_qsap) and the v4 # orchestrator (install_qsap_v4). v3 is the unconditional fallback — # if it's unavailable, we can't do anything at all. v4 is opportunistic — # we try it first and fall through on ANY error (not just ImportError), # because in production we've seen it raise at runtime for a missing # transitive dependency (torchvision) AFTER passing the import check. try: import quasar_selective_attention as _qsap_mod from quasar_selective_attention import install_qsap, print_qsap_health except ImportError as e: print(f"[QSAP-Install] ❌ Cannot import quasar_selective_attention: {e}") print(" Place quasar_selective_attention.py in the same") print(" directory as quasar_main4.py and retry.") return False # Optional v4 symbols. We look them up but do NOT fail if unavailable — # the v3 fallback is always usable. install_qsap_v4 = getattr(_qsap_mod, "install_qsap_v4", None) apply_enforcement_patch = getattr(_qsap_mod, "apply_enforcement_patch", None) print_enforcement_health = getattr(_qsap_mod, "print_enforcement_health", None) enforcement_diagnostics = getattr(_qsap_mod, "enforcement_diagnostics", None) _V4_AVAILABLE = all((install_qsap_v4, apply_enforcement_patch, print_enforcement_health, enforcement_diagnostics)) if not _V4_AVAILABLE: print("[QSAP-Install] ⚠️ v4 entry points not exported by SA module — " "will use v3 install_qsap. Enforcement patch will NOT be active.") # ── Register MessageContext in QSAP's namespace ─────────────────────── # HISTORY: before MessageContext was defined directly in # quasar_selective_attention.py (§0b), it was looked up at runtime via a # sys.modules scan in each dispatcher. This injection was added as a # belt-and-suspenders measure so that scan would find it. # # F6 FIX: MessageContext is now a proper class defined at the top of # quasar_selective_attention.py (§0b). Injecting a version from __main__ # overwrites the correct immutable class (with MappingProxyType-sealed # features_snapshot and __setattr__ guard) with whatever quasar_main4 # happens to have under that name — which may be missing, a different # class, or the old SimpleNamespace-based version from a prior session. # # Guard: only inject if the SA module does NOT already have MessageContext # at module level (i.e. we're running against an older SA module that # pre-dates §0b). If it's present in the SA module, trust that definition. _sa_has_mc = hasattr(_qsap_mod, 'MessageContext') and isinstance( getattr(_qsap_mod, 'MessageContext', None), type ) if _sa_has_mc: print( "[QSAP-Install] ✅ MessageContext already defined in SA module (§0b) — " "skipping injection from __main__ (F6 guard active)" ) else: # SA module pre-dates §0b — inject from __main__ as before. _main = sys.modules.get("__main__") or sys.modules.get("quasar_main4") _MainMC = getattr(_main, "MessageContext", None) if _MainMC is not None: _qsap_mod.MessageContext = _MainMC print("[QSAP-Install] ✅ MessageContext injected from __main__ (legacy SA module)") else: print( "[QSAP-Install] ⚠️ MessageContext not found in __main__ and not in SA module — " "QSAP will use SimpleNamespace fallback (harmless but upgrade SA module)" ) # ── Block training on the live system instance ──────────────────────── # Belt-and-suspenders: when QSAP_TRAINING_BLOCKED is True, ensure # system.training_enabled is False and the AVN continuous-training # thread (spawned during AVNTrainer.__init__) is signalled to stop. # # §P2-fix-3 (2026-04-19): Previously these kill-switches ran # unconditionally, which meant that even when the global flag was # False (training UNBLOCKED), the hard-wired kill-switches still # fired — stopping the CT thread that AVNTrainer.__init__ had # started 19 s earlier, and forcing system.training_enabled to # False. The pass-through logic in patches A–D could not recover # because re-spawn only happens on a fresh call to # start_continuous_training(), which never occurred. # # Now the kill-switches are gated on the flag. When training is # unblocked we instead leave system.training_enabled at True and # re-kick AVN CT if it somehow got cleared before we got here. import qsap_integration as _qi if _qi.QSAP_TRAINING_BLOCKED: system.training_enabled = False stop_existing_avn_training_thread(system) print(f"[QSAP-Install] 🔒 system.training_enabled = False") else: system.training_enabled = True _avn_trainer = getattr(system, "avn_trainer", None) if _avn_trainer is not None: _ct_alive = getattr(_avn_trainer, "_continuous_training_active", False) if not _ct_alive: # AVN CT was killed or never started — re-kick it. The # patched wrapper (patch A) will now pass through to the # original implementation because the flag is False. try: _avn_trainer.start_continuous_training() print( f"[{time.strftime('%H:%M:%S')}] ✅ [QSAP-Block] " f"AVN continuous-training re-kicked " f"(QSAP_TRAINING_BLOCKED=False)" ) except Exception as _ct_err: print( f"[{time.strftime('%H:%M:%S')}] ⚠️ [QSAP-Block] " f"AVN continuous-training re-kick failed: {_ct_err}" ) print(f"[QSAP-Install] ✅ system.training_enabled = True (training UNBLOCKED)") # ── Determine budget slots ──────────────────────────────────────────── # Only used by the v3 fallback path; v4 derives this internally from # len(system.agents). _budget = budget_slots or len(system.agents) # ── Install engine (v4 attempt → v3 fallback on ANY failure) ────────── # # Robustness requirement: install_qsap_v4 may fail AT RUNTIME even after # its entry-point imports resolve. Example from production: v4 called # into something that lazy-imported torchvision, which wasn't installed # on the Space. The old handler caught only ImportError at entry and # then returned False without trying v3 — leaving the engine with no # QSAP install at all, which is strictly worse than degraded v3. # # New behaviour: catch Exception (but not KeyboardInterrupt etc.) from # the v4 call and fall through to v3 with a loud warning. The fallback # is transparent to downstream code — system._qsap_engine is always # set, on_feature_message is always rebound, training is always blocked. engine = None _v4_succeeded = False if _V4_AVAILABLE: try: _v4_ok = install_qsap_v4( system = system, alpha_threshold = alpha_threshold, tau_min_ms = tau_min_ms, commit_wait_max_ms = commit_wait_max_ms, abstain_threshold = abstain_threshold, health_interval_s = int(QSAP_HEALTH_INTERVAL_S), ) if _v4_ok: engine = getattr(system, "_qsap_engine", None) if engine is not None: _v4_succeeded = True else: print("[QSAP-Install] ⚠️ install_qsap_v4 returned True but " "system._qsap_engine is None — falling back to v3") else: print("[QSAP-Install] ⚠️ install_qsap_v4 returned False — " "falling back to v3") except Exception as _v4_exc: # Catch ANYTHING — ImportError from transitive deps, RuntimeError # from misconfigured sub-components, KeyError from version drift. # We'd rather run in degraded v3 mode than not at all. print(f"[QSAP-Install] ⚠️ install_qsap_v4() raised: {_v4_exc}") print(f"[QSAP-Install] ⚠️ Falling back to v3 install_qsap. " f"Enforcement patch will NOT be active.") import traceback traceback.print_exc() # Clean up any partial state v4 may have left on the system # before retrying v3 from a known baseline. for _stale_attr in ("_qsap_engine", "_qsap_enforcement_active"): if hasattr(system, _stale_attr): try: delattr(system, _stale_attr) except Exception: pass # v3 fallback path — runs if v4 was unavailable, returned False, or raised. if not _v4_succeeded: try: engine = install_qsap( system_instance = system, alpha_threshold = alpha_threshold, tau_min_ms = tau_min_ms, budget_slots = _budget, commit_wait_max_ms = commit_wait_max_ms, abstain_threshold = abstain_threshold, ) system._qsap_engine = engine # Apply enforcement patch even on v3 fallback to block legacy path if apply_enforcement_patch is not None: try: apply_enforcement_patch(system) print('[QSAP-Install] ✅ Enforcement patch applied on v3 fallback path') except Exception as _ep_err: print(f'[QSAP-Install] ⚠️ Enforcement patch failed on v3 path: {_ep_err}') except Exception as e: # If v3 ALSO raises, we're out of options — report and bail. print(f"[QSAP-Install] ❌ install_qsap() (v3 fallback) also raised: {e}") import traceback; traceback.print_exc() return False # Flag for downstream callers and qsap_status() to distinguish the two # installation modes without sniffing method identities. system._qsap_enforcement_active = bool(_v4_succeeded) # ── QSAP-3.0.4: Fix double Q-value print ───────────────────────────── # quasar_main4.py's patched_predict_single_agent (line ~28713) calls # qb._orig_predict_single_agent (which prints Q-values at line 22751), # then the wrapper ALSO prints the same Q-values via print_diag. # Result: every agent's Q-values appear twice in the console. # # Fix: replace the wrapper with a version that preserves the safety # checks (None guards, exception handling) but removes the duplicate # print. The original _orig_predict_single_agent already prints. try: import numpy as _np qb = getattr(system, 'quantum_bridge', None) if qb is not None and hasattr(qb, '_orig_predict_single_agent'): _real_orig = qb._orig_predict_single_agent def _deduped_predict_single_agent(agent_name, states_dict=None): try: if getattr(qb, 'quantum_system', None) is None: return _np.array([0.5, 0.5], dtype=_np.float32) if getattr(qb, 'state_cache', None) is None: return _np.array([0.5, 0.5], dtype=_np.float32) # _real_orig already prints Q-values — no wrapper print needed return _real_orig(agent_name, states_dict) except Exception as _pred_err: print(f"[QSAP] predict_single_agent failed for {agent_name}: {_pred_err}") return _np.array([0.5, 0.5], dtype=_np.float32) qb.predict_single_agent = _deduped_predict_single_agent print("[QSAP-Install] ✅ De-duplicated predict_single_agent Q-value print") else: print("[QSAP-Install] ⚠️ No _orig_predict_single_agent found — " "double Q-value print may persist") except Exception as _dedup_err: print(f"[QSAP-Install] ⚠️ Q-value dedup patch failed: {_dedup_err}") # ── Health printer ───────────────────────────────────────────────────── # When v4 succeeded, install_qsap_v4 already started its own health # printer (print_enforcement_health, which internally covers # print_qsap_health). Starting a second thread would double the log # output. Only start the legacy QSAP-only printer when we're on the v3 # fallback path. if not _v4_succeeded: _start_qsap_health_printer(system, print_qsap_health) print("[QSAP-Install] ℹ️ Started v3 QSAP health printer " "(v4 not active — upgrade SA module or fix missing deps " "to enable enforcement diagnostics)") else: print(f"[QSAP-Install] ✅ v4 enforcement health printer active " f"(reports every {int(QSAP_HEALTH_INTERVAL_S)}s via print_enforcement_health)") print() if _v4_succeeded: print("╔══════════════════════════════════════════════════════════════════════╗") print("║ QSAP INTEGRATION COMPLETE — v4 ENFORCEMENT ACTIVE ║") print("╠══════════════════════════════════════════════════════════════════════╣") print(f"║ on_feature_message → _qsap_enforced_on_feature_message ║") print(f"║ _process_latest_features → _legacy_blocked_stub (HARD-BLOCKED) ║") print(f"║ ShardedInferenceEngine → instrumented with per-agent timings ║") print(f"║ system._qsap_engine → {type(engine).__name__:<41}║") _t_state_str = "True (UNBLOCKED)" if not _qi.QSAP_TRAINING_BLOCKED else "False (BLOCKED)" print(f"║ training_enabled → {_t_state_str:<41}║") print(f"║ Health report every → {int(QSAP_HEALTH_INTERVAL_S)}s (enforcement + engine) ║") print("╠══════════════════════════════════════════════════════════════════════╣") print("║ Diagnostic tags now active in log/stdout: ║") print("║ [QSAP-ENFORCE] ENTER — per-tick dispatch proof ║") print("║ [QSAP-SHARD] agent=… — per-agent inference proof ║") print("║ [QSAP-VIOLATION] — legacy bypass attempt (should be 0) ║") print("╚══════════════════════════════════════════════════════════════════════╝") else: print("╔══════════════════════════════════════════════════════════════════════╗") print("║ QSAP INTEGRATION COMPLETE — ⚠️ v3 FALLBACK (no enforcement) ║") print("╠══════════════════════════════════════════════════════════════════════╣") print(f"║ on_feature_message → _qsap_on_feature_message (v3) ║") print(f"║ _process_latest_features → legacy (fallback only — NOT blocked) ║") print(f"║ system._qsap_engine → {type(engine).__name__:<41}║") _t_state_str = "True (UNBLOCKED)" if not _qi.QSAP_TRAINING_BLOCKED else "False (BLOCKED)" print(f"║ training_enabled → {_t_state_str:<41}║") print(f"║ Health report every → {int(QSAP_HEALTH_INTERVAL_S)}s (engine only, no enforcement) ║") print("╠══════════════════════════════════════════════════════════════════════╣") print("║ ⚠️ v4 enforcement did NOT install (missing dep or runtime error). ║") print("║ _process_latest_features remains a LIVE BYPASS PATH. ║") print("║ Fix: pip install torchvision (or whichever transitive dep ║") print("║ was named in the exception above), then restart. ║") print("╚══════════════════════════════════════════════════════════════════════╝") print() return True # ───────────────────────────────────────────────────────────────────────────── # §4 BACKGROUND HEALTH PRINTER # ───────────────────────────────────────────────────────────────────────────── def _start_qsap_health_printer(system, print_qsap_health_fn) -> None: """ Daemon thread that calls print_qsap_health() every QSAP_HEALTH_INTERVAL_S seconds, so freshness stats and superseded-drop counts are visible in the console without any manual call. """ def _health_loop(): import qsap_integration as _qi while True: time.sleep(_qi.QSAP_HEALTH_INTERVAL_S) try: engine = getattr(system, "_qsap_engine", None) if engine is not None: print_qsap_health_fn(engine) except Exception as _e: print(f"[QSAP-Health] ⚠️ Health report failed: {_e}") t = threading.Thread( target=_health_loop, daemon=True, name="QSAP-HealthPrinter", ) t.start() print(f"[QSAP-Install] ✅ Health-printer thread started " f"(reports every {QSAP_HEALTH_INTERVAL_S:.0f}s)") # ───────────────────────────────────────────────────────────────────────────── # §5 CONVENIENCE DIAGNOSTIC # ───────────────────────────────────────────────────────────────────────────── def qsap_status(system) -> dict: """ Return a dict summarising the current QSAP+training-block state. Useful for dashboard widgets or one-off checks. Under v4 enforcement, the returned dict includes enforcement counters pulled from enforcement_diagnostics(). A healthy system reports: enforcement_active = True legacy_violation_count = 0 qsap_control_rate_pct ≈ 100.0 Example: from qsap_integration import qsap_status print(qsap_status(system)) """ import qsap_integration as _qi engine = getattr(system, "_qsap_engine", None) engine_report = engine.health_report() if engine is not None else {} status = { "qsap_installed": engine is not None, "enforcement_active": bool(getattr(system, "_qsap_enforcement_active", False)), "training_blocked": _qi.QSAP_TRAINING_BLOCKED, "system_training_enabled": getattr(system, "training_enabled", "?"), "avn_ct_active": getattr( getattr(system, "avn_trainer", None), "_continuous_training_active", False ), "engine_cycles": engine_report.get("cycles_total", 0), "engine_publishes": engine_report.get("publishes", 0), "engine_superseded_drops": engine_report.get("superseded_drops", {}), "freshness_stats": engine_report.get("freshness_stats", {}), } # Pull enforcement counters if v4 symbols are reachable. Failure here # is non-fatal — it just means we report None for those fields. try: from quasar_selective_attention import enforcement_diagnostics diag = enforcement_diagnostics() status.update({ "qsap_dispatch_count": diag.get("qsap_dispatch_count", 0), "legacy_bypass_count": diag.get("legacy_bypass_count", 0), "legacy_violation_count": diag.get("legacy_violation_count", 0), "qsap_control_rate_pct": diag.get("qsap_control_rate_pct"), "sealed_txn_publish_count": diag.get("sealed_txn_publish_count", 0), "invalid_msg_drops": diag.get("invalid_msg_drops", 0), "dispatch_latency_ms": diag.get("dispatch_latency_ms", {}), "per_agent_inference_ms": diag.get("per_agent_inference_ms", {}), }) except (ImportError, AttributeError): status.update({ "qsap_dispatch_count": None, "legacy_bypass_count": None, "legacy_violation_count": None, "qsap_control_rate_pct": None, "sealed_txn_publish_count": None, }) return status