"""Tests for M3 (promotion gate + active pointer) and M4 (auto-rollback)."""

from __future__ import annotations

from datetime import UTC, datetime

import numpy as np
import pytest

from opensquilla.gateway.config import RouterSelfLearningConfig, SquillaRouterConfig
from opensquilla.squilla_router.self_learning import encode_features, write_sample
from opensquilla.squilla_router.self_learning.dataset import TrainingDataset
from opensquilla.squilla_router.self_learning.evaluate import (
    decide_promotion,
    route_metrics,
    session_holdout_splits,
)
from opensquilla.squilla_router.self_learning.orchestrator import (
    in_process_trainer,
    maybe_run_update_router,
)
from opensquilla.squilla_router.self_learning.promotion import (
    learned_bundle_dir,
    promote_candidate,
    quarantine_candidate,
    read_active,
    resolve_active_bundle_dir,
    rollback_active,
    should_rollback,
    write_active_atomic,
)
from opensquilla.squilla_router.self_learning.schema import RouterTrainSample
from opensquilla.squilla_router.self_learning.state import (
    TrainState,
    load_train_state,
    save_train_state,
)

NOW = datetime(2026, 6, 6, 12, 0, 0, tzinfo=UTC)


def _cfg(**kw) -> RouterSelfLearningConfig:
    base = dict(
        enabled=True,
        train_min_samples=4,
        idle_hours=2.0,
        cooldown_hours=72.0,
        holdout_min_size=4,
        holdout_pct=0.4,
        holdout_repeats=2,
        max_critical_under_routing=0.5,
        cost_tolerance_pct=25.0,
    )
    base.update(kw)
    return RouterSelfLearningConfig(**base)


# --------------------------------------------------------------------------- #
# Active pointer primitives
# --------------------------------------------------------------------------- #


def test_active_pointer_defaults_to_baseline(tmp_path) -> None:
    assert read_active(tmp_path) == "baseline"
    assert resolve_active_bundle_dir(tmp_path) is None


def test_promote_and_resolve(tmp_path) -> None:
    bundle = learned_bundle_dir("v1-x", tmp_path)
    bundle.mkdir(parents=True)
    (bundle / "lgbm_main.bin").write_text("model", encoding="utf-8")
    prev = promote_candidate("v1-x", tmp_path)
    assert prev == "baseline"
    assert read_active(tmp_path) == "learned/v1-x"
    assert resolve_active_bundle_dir(tmp_path) == bundle


def test_resolve_falls_back_when_bundle_incomplete(tmp_path) -> None:
    write_active_atomic("learned/ghost", tmp_path)  # no such bundle on disk
    assert resolve_active_bundle_dir(tmp_path) is None


def test_rollback_reverts_to_baseline(tmp_path) -> None:
    write_active_atomic("learned/v1-x", tmp_path)
    prev = rollback_active(tmp_path)
    assert prev == "learned/v1-x"
    assert read_active(tmp_path) == "baseline"


def test_quarantine_moves_bundle_out(tmp_path) -> None:
    bundle = learned_bundle_dir("v1-x", tmp_path)
    bundle.mkdir(parents=True)
    (bundle / "lgbm_main.bin").write_text("m", encoding="utf-8")
    dest = quarantine_candidate("v1-x", tmp_path)
    assert dest is not None and dest.exists()
    assert not bundle.exists()


def test_should_rollback_rules() -> None:
    cfg = _cfg(min_monitor_samples=10, complaint_regression_delta=0.05)
    # regression beyond delta with enough samples -> rollback
    assert should_rollback(pre_complaint_rate=0.1, post_complaint_rate=0.3, post_n=20, config=cfg)
    # not enough samples yet
    assert not should_rollback(
        pre_complaint_rate=0.1, post_complaint_rate=0.9, post_n=5, config=cfg
    )
    # within delta
    assert not should_rollback(
        pre_complaint_rate=0.1, post_complaint_rate=0.12, post_n=50, config=cfg
    )
    # no baseline recorded
    assert not should_rollback(
        pre_complaint_rate=None, post_complaint_rate=0.9, post_n=50, config=cfg
    )
    # auto_rollback disabled
    off = _cfg(auto_rollback=False)
    assert not should_rollback(
        pre_complaint_rate=0.0, post_complaint_rate=0.9, post_n=99, config=off
    )


# --------------------------------------------------------------------------- #
# Evaluation
# --------------------------------------------------------------------------- #


def test_route_metrics_basic() -> None:
    pred = np.array([1, 2, 0, 3])
    target = np.array([1, 3, 0, 3])  # under-routes the 2nd (pred 2 < target 3)
    served = np.array([1, 1, 0, 3])
    m = route_metrics(pred, target, served)
    assert m.n == 4
    assert m.agreement == 0.75
    # critical = target>=2 -> indices 1,3; pred<target only at idx1 -> 0.5
    assert m.critical_under_routing_rate == 0.5


def test_holdout_splits_are_session_whole_and_floored(tmp_path) -> None:
    ds = TrainingDataset(
        X=np.zeros((10, 390), np.float32),
        y=np.zeros(10, np.int64),
        w=np.ones(10, np.float32),
        session_keys=[f"s{i % 2}" for i in range(10)],  # 2 sessions
    )
    splits = session_holdout_splits(ds, holdout_pct=0.4, repeats=2, min_size=2)
    assert splits
    for train_idx, hold_idx in splits:
        # no session appears on both sides
        train_sessions = {ds.session_keys[i] for i in train_idx}
        hold_sessions = {ds.session_keys[i] for i in hold_idx}
        assert not (train_sessions & hold_sessions)
    # too few sessions -> no splits
    single = TrainingDataset(
        X=np.zeros((4, 390), np.float32),
        y=np.zeros(4, np.int64),
        w=np.ones(4, np.float32),
        session_keys=["s0"] * 4,
    )
    assert session_holdout_splits(single, holdout_pct=0.4, repeats=2, min_size=2) == []


def test_decide_promotion_paths() -> None:
    cfg = _cfg(holdout_min_size=4, max_critical_under_routing=0.3, cost_tolerance_pct=10.0)
    good_cv = {
        "agreement": 0.9,
        "critical_under_routing_rate": 0.1,
        "mean_pred_idx": 1.0,
        "served_mean_idx": 1.0,
        "n_holdout": 20,
    }
    assert decide_promotion(good_cv, golden=None, baseline_golden=None, config=cfg).promote
    # quality regression
    bad_q = {**good_cv, "critical_under_routing_rate": 0.6}
    d = decide_promotion(bad_q, golden=None, baseline_golden=None, config=cfg)
    assert not d.promote and "quality_regression" in d.reason
    # cost regression (predicts much higher than served)
    bad_c = {**good_cv, "mean_pred_idx": 2.5, "served_mean_idx": 1.0}
    d = decide_promotion(bad_c, golden=None, baseline_golden=None, config=cfg)
    assert not d.promote and "cost_regression" in d.reason
    # insufficient eval (no cv, no golden)
    empty = {"agreement": None, "n_holdout": 0, "served_mean_idx": 0.0}
    assert decide_promotion(empty, golden=None, baseline_golden=None, config=cfg).reason == (
        "insufficient_eval"
    )


# --------------------------------------------------------------------------- #
# Strategy integration (cache)
# --------------------------------------------------------------------------- #


def test_invalidate_strategy_cache(monkeypatch) -> None:
    from opensquilla.engine.steps import squilla_router as step

    step._strategy = object()
    step._strategy_key = ("x",)
    step.invalidate_strategy_cache()
    assert step._strategy is None and step._strategy_key is None


def test_cache_key_tracks_active_bundle(monkeypatch) -> None:
    from opensquilla.engine.steps import squilla_router as step

    cfg = SquillaRouterConfig(self_learning=_cfg())
    monkeypatch.setattr(step, "_active_bundle_dir", lambda _c: "learned/v1")
    key1 = step._strategy_cache_key(cfg)
    monkeypatch.setattr(step, "_active_bundle_dir", lambda _c: "learned/v2")
    key2 = step._strategy_cache_key(cfg)
    assert key1 != key2


def test_active_bundle_dir_none_when_disabled() -> None:
    from opensquilla.engine.steps import squilla_router as step

    cfg = SquillaRouterConfig(self_learning=RouterSelfLearningConfig(enabled=False))
    assert step._active_bundle_dir(cfg) is None


# --------------------------------------------------------------------------- #
# Orchestrator: promote / reject / rollback
# --------------------------------------------------------------------------- #


def _write_separable_store(tmp_path, agent="agp", n=36) -> None:
    """Confidence-gate (high-value) turns with features cleanly separable by
    final class, so CV agreement is high and cost does not regress.

    Across 3 sessions this leaves each whole-session holdout fold ~12 training
    rows (6 per class) — enough for LightGBM to split past ``min_data_in_leaf``.
    """
    rng = np.random.RandomState(1)
    for i in range(n):
        cls = i % 2  # 0 -> R1, 1 -> R2
        fc = "R2" if cls else "R1"
        feats = (rng.randn(390) * 0.1).astype(np.float32)
        feats[0] = 5.0 if cls else -5.0
        write_sample(
            RouterTrainSample(
                session_key=f"s{i % 3}",
                turn_index=i,
                ts=f"2026-06-01T00:00:{i:02d}Z",
                feature_schema_version="v1",
                features_390_b64=encode_features(feats),
                route_class=fc,
                final_route_class=fc,
                routed_tier="c2" if cls else "c1",
                confidence_gate_applied=True,
            ),
            agent,
            home=tmp_path,
        )


def test_orchestrator_promotes_good_candidate(tmp_path) -> None:
    pytest.importorskip("lightgbm")
    base = tmp_path / "base"
    base.mkdir()
    (base / "router.runtime.yaml").write_text("k: v\n", encoding="utf-8")
    _write_separable_store(tmp_path)

    res = maybe_run_update_router(
        "agp",
        router_cfg=SquillaRouterConfig(self_learning=_cfg()),
        home=tmp_path,
        now=NOW,
        trainer=in_process_trainer,
        base_dir=base,
    )
    assert res.ran and res.promoted and res.reason == "promoted", res
    assert read_active(tmp_path) == f"learned/{res.version}"
    state = load_train_state("agp", tmp_path)
    assert state.active_version == res.version
    assert state.promoted_at is not None
    assert list((tmp_path / "router" / ".receipts").glob("agp-*-promoted.json"))


def test_orchestrator_rejects_on_golden_floor(tmp_path) -> None:
    pytest.importorskip("lightgbm")
    base = tmp_path / "base"
    base.mkdir()
    (base / "router.runtime.yaml").write_text("k: v\n", encoding="utf-8")
    _write_separable_store(tmp_path, agent="agr")

    # Golden set whose labels contradict the separable signal -> low agreement.
    rng = np.random.RandomState(2)
    gx = (rng.randn(20, 390) * 0.1).astype(np.float32)
    gy = np.zeros(20, np.int64)
    for i in range(20):
        gx[i, 0] = 5.0 if i % 2 else -5.0
        gy[i] = 0 if i % 2 else 2  # inverted vs training signal
    golden = tmp_path / "golden.npz"
    np.savez(golden, X=gx, y=gy)

    res = maybe_run_update_router(
        "agr",
        router_cfg=SquillaRouterConfig(
            self_learning=_cfg(golden_eval_path=str(golden), min_golden_agreement=0.9)
        ),
        home=tmp_path,
        now=NOW,
        trainer=in_process_trainer,
        base_dir=base,
    )
    assert res.ran and not res.promoted
    assert "golden_below_floor" in res.reason
    assert read_active(tmp_path) == "baseline"  # never swapped
    # candidate quarantined
    assert (tmp_path / "router" / "learned" / ".quarantine" / res.version).exists()


def test_orchestrator_auto_rolls_back_regressed_candidate(tmp_path) -> None:
    # Pre-promoted state with a clean baseline complaint rate.
    bundle = learned_bundle_dir("vBad", tmp_path)
    bundle.mkdir(parents=True)
    (bundle / "lgbm_main.bin").write_text("m", encoding="utf-8")
    write_active_atomic("learned/vBad", tmp_path)
    save_train_state(
        TrainState(
            active_version="vBad",
            promoted_at="2026-06-05T00:00:00Z",
            pre_promotion_complaint_rate=0.0,
        ),
        "agx",
        tmp_path,
    )
    # Post-swap traffic with high complaint rate, recent (so the agent looks
    # active and no training runs after the rollback).
    for i in range(30):
        write_sample(
            RouterTrainSample(
                session_key="s",
                turn_index=i,
                ts=f"2026-06-06T11:5{i % 10}:00Z",
                feature_schema_version="v1",
                features_390_b64=encode_features(np.zeros(390, np.float32)),
                route_class="R1",
                final_route_class="R2",
                complaint_detected=(i < 20),  # 20/30 complaints
            ),
            "agx",
            home=tmp_path,
        )

    res = maybe_run_update_router(
        "agx",
        router_cfg=SquillaRouterConfig(
            self_learning=_cfg(min_monitor_samples=10, complaint_regression_delta=0.05)
        ),
        home=tmp_path,
        now=NOW,
        trainer=in_process_trainer,
        base_dir=tmp_path / "base",
    )
    assert res.rolled_back
    assert read_active(tmp_path) == "baseline"
    state = load_train_state("agx", tmp_path)
    assert state.active_version is None
    assert (tmp_path / "router" / "learned" / ".quarantine" / "vBad").exists()
    assert list((tmp_path / "router" / ".receipts").glob("agx-*-rollback.json"))


# --------------------------------------------------------------------------- #
# Base-upgrade detach guard (verify_active_bundle)
# --------------------------------------------------------------------------- #


@pytest.fixture(autouse=True)
def _reset_verify_memo():
    """The verify memo is process-global; isolate it per test."""
    import opensquilla.squilla_router.self_learning.promotion as promo

    promo._verify_key = None
    promo._verify_result = None
    promo._fp_cache = None
    yield
    promo._verify_key = None
    promo._verify_result = None
    promo._fp_cache = None


def _make_learned(tmp_path, version: str, *, base_fingerprint: str | None) -> None:
    import json

    bundle = learned_bundle_dir(version, tmp_path)
    bundle.mkdir(parents=True)
    (bundle / "lgbm_main.bin").write_bytes(b"learned-head")
    manifest: dict = {"version": version}
    if base_fingerprint is not None:
        manifest["base_fingerprint"] = base_fingerprint
    (bundle / "learned_manifest.json").write_text(json.dumps(manifest), encoding="utf-8")


def _make_base(tmp_path, content: bytes = b"base-model-v1"):
    base = tmp_path / "base"
    base.mkdir(exist_ok=True)
    (base / "lgbm_main.bin").write_bytes(content)
    return base


def test_verify_detaches_on_base_upgrade(tmp_path) -> None:
    from opensquilla.squilla_router.self_learning.promotion import verify_active_bundle
    from opensquilla.squilla_router.self_learning.train import base_bundle_fingerprint

    base = _make_base(tmp_path)
    old_fp = base_bundle_fingerprint(base)
    _make_learned(tmp_path, "v1", base_fingerprint=old_fp)
    write_active_atomic("learned/v1", tmp_path)

    # Same base: nothing happens.
    check = verify_active_bundle(base, tmp_path)
    assert not check.detached and read_active(tmp_path) == "learned/v1"

    # Base replaced (package upgrade): detach + quarantine + baseline.
    (base / "lgbm_main.bin").write_bytes(b"base-model-v2-NEW-WEIGHTS")
    check = verify_active_bundle(base, tmp_path)
    assert check.detached and check.reason == "base_upgraded"
    assert read_active(tmp_path) == "baseline"
    assert (tmp_path / "router" / "learned" / ".quarantine" / "v1").exists()


def test_verify_trusts_legacy_bundle_without_fingerprint(tmp_path) -> None:
    """Pre-fingerprint candidates must not be mass-detached on upgrade."""
    from opensquilla.squilla_router.self_learning.promotion import verify_active_bundle

    base = _make_base(tmp_path)
    _make_learned(tmp_path, "vLegacy", base_fingerprint=None)
    write_active_atomic("learned/vLegacy", tmp_path)

    check = verify_active_bundle(base, tmp_path)
    assert not check.detached
    assert read_active(tmp_path) == "learned/vLegacy"


def test_verify_memoizes_per_pointer_and_base(tmp_path, monkeypatch) -> None:
    """The 39MB hash must not run on every strategy-cache-key computation."""
    import opensquilla.squilla_router.self_learning.train as train_mod
    from opensquilla.squilla_router.self_learning.promotion import verify_active_bundle

    base = _make_base(tmp_path)
    fp = train_mod.base_bundle_fingerprint(base)
    _make_learned(tmp_path, "v1", base_fingerprint=fp)
    write_active_atomic("learned/v1", tmp_path)

    calls = {"n": 0}
    real = train_mod.base_bundle_fingerprint

    def counting(base_dir):
        calls["n"] += 1
        return real(base_dir)

    monkeypatch.setattr(train_mod, "base_bundle_fingerprint", counting)
    import opensquilla.squilla_router.self_learning.promotion as promo

    promo._fp_cache = None
    verify_active_bundle(base, tmp_path)
    verify_active_bundle(base, tmp_path)
    verify_active_bundle(base, tmp_path)
    # Repeated calls are stable, never detach, and the expensive hash is
    # stat-gated to a single computation (see the dedicated hash-once test).
    assert read_active(tmp_path) == "learned/v1"
    assert calls["n"] == 1
    promo._fp_cache = None


def test_verify_noop_on_baseline_pointer(tmp_path) -> None:
    from opensquilla.squilla_router.self_learning.promotion import verify_active_bundle

    base = _make_base(tmp_path)
    check = verify_active_bundle(base, tmp_path)
    assert not check.detached
    assert read_active(tmp_path) == "baseline"


def test_orchestrator_reconciles_detached_candidate(tmp_path) -> None:
    """After a base upgrade the offline pass clears promotion-monitor state."""
    from opensquilla.squilla_router.self_learning.train import base_bundle_fingerprint

    base = _make_base(tmp_path)
    old_fp = base_bundle_fingerprint(base)
    _make_learned(tmp_path, "vOld", base_fingerprint=old_fp)
    write_active_atomic("learned/vOld", tmp_path)
    save_train_state(
        TrainState(
            active_version="vOld",
            promoted_at="2026-06-05T00:00:00Z",
            pre_promotion_complaint_rate=0.10,
        ),
        "agd",
        tmp_path,
    )
    # Upgrade the base.
    (base / "lgbm_main.bin").write_bytes(b"base-model-v2")

    res = maybe_run_update_router(
        "agd",
        router_cfg=SquillaRouterConfig(self_learning=_cfg()),
        home=tmp_path,
        now=NOW,
        trainer=in_process_trainer,
        base_dir=base,
    )
    # No training data -> gates fail, but the detach must have reconciled.
    assert not res.rolled_back  # detach is not a regression rollback
    assert read_active(tmp_path) == "baseline"
    state = load_train_state("agd", tmp_path)
    assert state.active_version is None and state.promoted_at is None
    assert state.pre_promotion_complaint_rate is None
    assert list((tmp_path / "router" / ".receipts").glob("agd-*-detached.json"))
    assert (tmp_path / "router" / "learned" / ".quarantine" / "vOld").exists()


def test_candidate_manifest_records_base_fingerprint(tmp_path) -> None:
    import json

    pytest.importorskip("lightgbm")
    from types import SimpleNamespace

    from opensquilla.squilla_router.self_learning.train import (
        base_bundle_fingerprint,
        build_candidate_bundle,
        train_booster,
    )

    base = tmp_path / "base"
    base.mkdir()
    booster, _ = train_booster(
        _mini_dataset(), base_model_path=None, config=SimpleNamespace(num_boost_round=8)
    )
    booster.save_model(str(base / "lgbm_main.bin"))
    expected = base_bundle_fingerprint(base)

    info = build_candidate_bundle(
        _mini_dataset(),
        base_dir=base,
        learned_root=tmp_path / "learned",
        config=SimpleNamespace(num_boost_round=4),
    )
    assert info.base_fingerprint == expected
    manifest = json.loads(
        (tmp_path / "learned" / info.version / "learned_manifest.json").read_text()
    )
    assert manifest["base_fingerprint"] == expected


def _mini_dataset() -> TrainingDataset:
    rng = np.random.RandomState(0)
    n = 24
    return TrainingDataset(
        X=rng.rand(n, 390).astype(np.float32),
        y=(np.arange(n) % 3).astype(np.int64),
        w=np.ones(n, dtype=np.float32),
        served=(np.arange(n) % 3).astype(np.int64),
        session_keys=[f"s{i // 4}" for i in range(n)],
        turn_indices=[i % 4 for i in range(n)],
        days=["2026-06-01"] * n,
        reasons=["normal"] * n,
        feature_schema_version="v1",
        n_sessions=6,
    )


# --------------------------------------------------------------------------- #
# Engine fallback chain: learned -> baseline -> heuristic
# --------------------------------------------------------------------------- #


def test_broken_learned_bundle_falls_back_to_baseline(tmp_path, monkeypatch) -> None:
    """A corrupt learned bundle must degrade to the shipped ML baseline, not
    straight to heuristic tiering."""
    from opensquilla.engine.steps import squilla_router as step

    built = []

    class _FakeStrategy:
        source = "v4_phase3"
        _available = True

        def __init__(self, bundle_dir=None, **_kw):
            built.append(bundle_dir)
            if bundle_dir == "/learned/broken":
                raise RuntimeError("incomplete V4 router artifact bundle")

    import opensquilla.squilla_router.v4_phase3 as v4mod

    monkeypatch.setattr(v4mod, "V4Phase3Strategy", _FakeStrategy)
    monkeypatch.setattr(step, "_active_bundle_dir", lambda _c: "/learned/broken")
    step.invalidate_strategy_cache()

    cfg = SquillaRouterConfig(self_learning=_cfg())
    strategy = step._get_strategy(cfg)
    # First attempt hit the learned dir, second the baseline (None -> packaged).
    assert built == ["/learned/broken", None]
    assert isinstance(strategy, _FakeStrategy)
    step.invalidate_strategy_cache()


def test_learned_and_baseline_both_broken_degrades_to_heuristic(
    tmp_path, monkeypatch
) -> None:
    from opensquilla.engine.routing.heuristic import HeuristicRouterStrategy
    from opensquilla.engine.steps import squilla_router as step

    class _AlwaysBroken:
        source = "v4_phase3"

        def __init__(self, **_kw):
            raise RuntimeError("no runtime")

    import opensquilla.squilla_router.v4_phase3 as v4mod

    monkeypatch.setattr(v4mod, "V4Phase3Strategy", _AlwaysBroken)
    monkeypatch.setattr(step, "_active_bundle_dir", lambda _c: "/learned/broken")
    monkeypatch.setattr(step, "_router_runtime_warning_emitted", False)
    step.invalidate_strategy_cache()

    cfg = SquillaRouterConfig(self_learning=_cfg())
    strategy = step._get_strategy(cfg)
    assert isinstance(strategy, HeuristicRouterStrategy)
    step.invalidate_strategy_cache()


def test_failed_detach_is_not_memoized_and_retries(tmp_path, monkeypatch) -> None:
    """A transient detach failure must not permanently trust the stale bundle."""
    import opensquilla.squilla_router.self_learning.promotion as promo
    from opensquilla.squilla_router.self_learning.promotion import verify_active_bundle
    from opensquilla.squilla_router.self_learning.train import base_bundle_fingerprint

    base = _make_base(tmp_path)
    old_fp = base_bundle_fingerprint(base)
    _make_learned(tmp_path, "v1", base_fingerprint=old_fp)
    write_active_atomic("learned/v1", tmp_path)
    (base / "lgbm_main.bin").write_bytes(b"base-model-v2-UPGRADED")

    calls = {"n": 0}
    real_rollback = promo.rollback_active

    def flaky_rollback(home=None, **kw):
        calls["n"] += 1
        if calls["n"] == 1:
            raise OSError("disk full")
        return real_rollback(home, **kw)

    monkeypatch.setattr(promo, "rollback_active", flaky_rollback)

    first = verify_active_bundle(base, tmp_path)
    assert not first.detached  # fail-open on the transient error
    assert read_active(tmp_path) == "learned/v1"

    second = verify_active_bundle(base, tmp_path)  # must RETRY, not trust memo
    assert second.detached and second.reason == "base_upgraded"
    assert read_active(tmp_path) == "baseline"
    assert calls["n"] == 2


def test_fingerprint_hash_runs_once_per_base_file_change(tmp_path, monkeypatch) -> None:
    """The 39MB sha256 must be stat-gated, not recomputed per call."""
    import opensquilla.squilla_router.self_learning.promotion as promo
    import opensquilla.squilla_router.self_learning.train as train_mod
    from opensquilla.squilla_router.self_learning.promotion import verify_active_bundle

    base = _make_base(tmp_path)
    fp = train_mod.base_bundle_fingerprint(base)
    _make_learned(tmp_path, "v1", base_fingerprint=fp)
    write_active_atomic("learned/v1", tmp_path)
    promo._fp_cache = None

    hashes = {"n": 0}
    real = train_mod.base_bundle_fingerprint

    def counting(base_dir):
        hashes["n"] += 1
        return real(base_dir)

    monkeypatch.setattr(train_mod, "base_bundle_fingerprint", counting)
    for _ in range(5):
        verify_active_bundle(base, tmp_path)
    assert hashes["n"] == 1  # one hash; four stat-gated cache hits

    promo._fp_cache = None


def test_engine_active_bundle_dir_invokes_verify_and_detaches(
    tmp_path, monkeypatch
) -> None:
    """The real engine path must run the base-upgrade guard, not just resolve.

    No monkeypatching of _active_bundle_dir itself: config points v4_bundle_dir
    at a synthetic base, the state home holds a promoted-but-stale candidate,
    and resolving the bundle through the engine must detach it.
    """
    from opensquilla.engine.steps import squilla_router as step
    from opensquilla.squilla_router.self_learning.train import base_bundle_fingerprint

    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path))
    base = _make_base(tmp_path)
    old_fp = base_bundle_fingerprint(base)
    _make_learned(tmp_path, "vStale", base_fingerprint=old_fp)
    write_active_atomic("learned/vStale", tmp_path)
    (base / "lgbm_main.bin").write_bytes(b"base-model-v2-UPGRADED")

    cfg = SquillaRouterConfig(self_learning=_cfg(), v4_bundle_dir=str(base))
    resolved = step._active_bundle_dir(cfg)

    assert resolved is None  # stale candidate detached -> baseline
    assert read_active(tmp_path) == "baseline"
    assert (tmp_path / "router" / "learned" / ".quarantine" / "vStale").exists()