qsap_integration.py

# -*- 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