backend/core/projects/instrument_ops.py

"""Ops metrics for instrument-first pipeline (projects + catalog flags).

Feature flag env names (surfaced in instrument-stats / ops snapshot):

- ``ENABLE_ELIGIBILITY_SPINE`` — gate Full Autopilot on de minimis / MŚP spine
  (default ``true``). Boolean key: ``eligibility_spine_enabled``.
- ``ENABLE_STRATEGY_CASCADE`` — strategy path recommendations available
  (default ``true``). Boolean key: ``strategy_cascade_available`` also
  requires ``core.strategy.recommend`` to be importable.
- ``ENABLE_POLICY_2026`` — July 2026 match policy (KPO downrank, mid-term
  boost, de minimis blocks). Boolean key: ``policy_2026_enabled``; also
  requires ``core.strategy.policy_2026`` to be importable.
- ``ENABLE_TAX_PATHS`` — tax checklist surface (PSI / B+R / IP Box). Default on.
- ``ENABLE_BK2021`` — BK2021 B2B stub (default off).

Module presence (import checks, no network):

- ``core.eligibility.spine``
- ``core.strategy.recommend``
- ``core.strategy.policy_2026``
- ``core.strategy.tax_paths``
- ``core.strategy.direct_eu``
- ``core.b2b.bk2021_stub``
"""
from __future__ import annotations

import importlib
import os
from collections import Counter
from typing import Any, Dict, List, Optional, Sequence

from sqlalchemy.orm import Session

# Canonical env flag names (document once; reuse in snapshot + tests).
FLAG_ELIGIBILITY_SPINE = "ENABLE_ELIGIBILITY_SPINE"
FLAG_STRATEGY_CASCADE = "ENABLE_STRATEGY_CASCADE"
FLAG_POLICY_2026 = "ENABLE_POLICY_2026"
FLAG_TAX_PATHS = "ENABLE_TAX_PATHS"
FLAG_BK2021 = "ENABLE_BK2021"

# Modules whose presence is reported in the ops snapshot.
MODULE_ELIGIBILITY_SPINE = "core.eligibility.spine"
MODULE_STRATEGY_RECOMMEND = "core.strategy.recommend"
MODULE_POLICY_2026 = "core.strategy.policy_2026"
MODULE_TAX_PATHS = "core.strategy.tax_paths"
MODULE_DIRECT_EU = "core.strategy.direct_eu"
MODULE_BK2021 = "core.b2b.bk2021_stub"

_ELIGIBILITY_STRATEGY_MODULES = (
    MODULE_ELIGIBILITY_SPINE,
    MODULE_STRATEGY_RECOMMEND,
    MODULE_POLICY_2026,
)


def _env_flag(name: str, default: str = "true") -> bool:
    return os.environ.get(name, default).lower() in ("1", "true", "yes", "on")


def _module_present(dotted: str) -> bool:
    try:
        importlib.import_module(dotted)
        return True
    except Exception:
        return False


def _ext(row) -> Dict[str, Any]:
    raw = getattr(row, "external_context", None) or {}
    return dict(raw) if isinstance(raw, dict) else {}


def detect_project_instrument_mismatch(row, ext: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    Reportable instrument-mismatch signal for §10 advisor metric.

    Prefer explicit advisor finding / flag, then pure ``detect_instrument_mismatch``
    on stored document/section text, then soft strategy_path vs family conflict.
    """
    ext = dict(ext) if isinstance(ext, dict) else _ext(row)
    reasons: List[str] = []

    # Explicit ops/advisor markers
    if ext.get("instrument_mismatch") is True or ext.get("INSTRUMENT_MISMATCH"):
        reasons.append("explicit_instrument_mismatch_flag")
    findings = ext.get("advisor_findings") or ext.get("quality_findings") or []
    if isinstance(findings, list):
        for f in findings:
            if isinstance(f, dict) and str(f.get("code") or "").upper() == "INSTRUMENT_MISMATCH":
                reasons.append("advisor_finding_INSTRUMENT_MISMATCH")
                break
            if isinstance(f, str) and "INSTRUMENT_MISMATCH" in f.upper():
                reasons.append("advisor_finding_INSTRUMENT_MISMATCH")
                break
    wca = ext.get("world_class_advisor") if isinstance(ext.get("world_class_advisor"), dict) else {}
    for b in (wca.get("blockers") or [])[:12]:
        if "INSTRUMENT_MISMATCH" in str(b).upper() or "instrument mismatch" in str(b).lower():
            reasons.append("world_class_advisor_blocker")
            break

    schema = ext.get("instrument_schema") if isinstance(ext.get("instrument_schema"), dict) else {}
    family = str(
        (schema.get("family") if schema else None)
        or ext.get("instrument_program_type")
        or getattr(row, "program_type", None)
        or ""
    ).upper()
    program_type = family or str(getattr(row, "program_type", None) or "")

    doc = (
        str(ext.get("final_document_markdown") or "")
        or str(getattr(row, "final_document_markdown", None) or "")
        or str(ext.get("document_text") or "")
        or str(ext.get("foreign_grant_extract_text") or "")
    )
    section_titles: List[str] = []
    for key in ("section_titles", "required_sections", "seeded_section_types"):
        val = ext.get(key)
        if isinstance(val, list):
            section_titles.extend(str(x) for x in val if x)

    try:
        from core.projects.instrument_profile import detect_instrument_mismatch

        det = detect_instrument_mismatch(
            program_type=program_type,
            document_text=doc,
            section_titles=section_titles or None,
        )
        if det.get("mismatch"):
            reasons.append("detect_instrument_mismatch")
            for f in (det.get("findings") or [])[:3]:
                reasons.append(str(f)[:120])
    except Exception:
        pass

    # Soft: strategy path vs SMART hybrid on non-SMART family
    strat = str(ext.get("strategy_path") or "").lower()
    if schema and schema.get("allow_smart_modules") is False and family not in ("", "SMART", "UNKNOWN"):
        if any(m in strat for m in ("smart",)) and "direct_eu" not in strat:
            # only if document/sections look SMART — already covered above
            pass
    if family in ("EUROGRANTY", "HORIZON_PREP") and schema.get("allow_smart_modules") is True:
        reasons.append("eurogrant_family_allows_smart_modules")

    unique = []
    for r in reasons:
        if r not in unique:
            unique.append(r)
    return {
        "mismatch": bool(unique),
        "reasons": unique[:8],
        "program_type": program_type or None,
        "family": family or None,
    }


# de_minimis sources that count as "has a real source" for §10 firm SLA.
_DE_MINIMIS_KNOWN_SOURCES = frozenset(
    {"sudop", "manual", "manual_override", "profile_field"}
)
_DE_MINIMIS_UNKNOWN_SOURCES = frozenset(
    {"unknown", "sudop_unconfigured", "sudop_error", "", "none", "null"}
)


def _firm_metric_signals(ext: Dict[str, Any]) -> tuple:
    """
    Extract (has_de_minimis_source: bool, msp_confidence: Optional[float]) from
    project external_context / eligibility spine snapshot.
    Prefer spine extractors when importable; fall back to nested dicts.
    """
    has_dm = False
    msp_conf: Optional[float] = None

    elig = ext.get("eligibility") if isinstance(ext.get("eligibility"), dict) else {}
    cd = ext.get("company_data") if isinstance(ext.get("company_data"), dict) else {}
    dm_blob = elig.get("de_minimis") if isinstance(elig.get("de_minimis"), dict) else {}

    try:
        from core.eligibility.spine import extract_de_minimis, extract_msp_status

        dm = extract_de_minimis(ext)
        src = str((dm or {}).get("source") or "").lower()
        has_dm = bool(src and src not in _DE_MINIMIS_UNKNOWN_SOURCES)
        msp = extract_msp_status(ext)
        conf = (msp or {}).get("confidence")
        if conf is None:
            conf = (msp or {}).get("msp_confidence")
        if conf is not None:
            try:
                msp_conf = float(conf)
            except (TypeError, ValueError):
                msp_conf = None
    except Exception:
        has_dm = False
        msp_conf = None

    # Nested eligibility snapshot may already store spine-shaped source/confidence.
    if not has_dm:
        src = str(
            dm_blob.get("source")
            or elig.get("de_minimis_source")
            or ext.get("de_minimis_source")
            or ""
        ).lower()
        if src and src not in _DE_MINIMIS_UNKNOWN_SOURCES:
            has_dm = True
        elif src in _DE_MINIMIS_KNOWN_SOURCES:
            has_dm = True
    if not has_dm:
        sudop = cd.get("sudop") if isinstance(cd.get("sudop"), dict) else {}
        if sudop and sudop.get("configured") is not False and not sudop.get("error"):
            # configured SUDOP blob counts even if total is 0
            if sudop.get("configured") is True or sudop.get("de_minimis_total_eur") is not None:
                has_dm = True
        if cd.get("de_minimis_manual_eur") is not None or elig.get("de_minimis_manual_eur") is not None:
            has_dm = True

    if msp_conf is None:
        msp_blob = (
            elig.get("msp")
            if isinstance(elig.get("msp"), dict)
            else (cd.get("msp_analysis") if isinstance(cd.get("msp_analysis"), dict) else {})
        )
        conf = (
            (msp_blob or {}).get("confidence")
            or (msp_blob or {}).get("msp_confidence")
            or elig.get("msp_confidence")
            or ext.get("msp_confidence")
        )
        if conf is not None:
            try:
                msp_conf = float(conf)
            except (TypeError, ValueError):
                msp_conf = None
    return has_dm, msp_conf


def build_eligibility_strategy_ops_snapshot(
    rows: Optional[Sequence[Any]] = None,
) -> Dict[str, Any]:
    """
    Pure ops snapshot for eligibility spine + strategy cascade + policy_2026
    + tax paths + direct EU module + BK2021 stub.

    Returns env-driven booleans, module import presence, flag name map, and
    optional per-project coverage counts when ``rows`` is provided.
    """
    spine_mod = _module_present(MODULE_ELIGIBILITY_SPINE)
    recommend_mod = _module_present(MODULE_STRATEGY_RECOMMEND)
    policy_mod = _module_present(MODULE_POLICY_2026)
    tax_mod = _module_present(MODULE_TAX_PATHS)
    eu_mod = _module_present(MODULE_DIRECT_EU)
    bk_mod = _module_present(MODULE_BK2021)

    if spine_mod:
        try:
            from core.eligibility.spine import eligibility_spine_enabled as _spine_on

            spine_enabled = bool(_spine_on())
        except Exception:
            spine_enabled = _env_flag(FLAG_ELIGIBILITY_SPINE, "true")
    else:
        spine_enabled = _env_flag(FLAG_ELIGIBILITY_SPINE, "true")

    # Prefer recommend's thin helper so ops booleans match cascade_enforced.
    if recommend_mod:
        try:
            from core.strategy.recommend import strategy_cascade_enabled as _cascade_on

            strategy_flag = bool(_cascade_on())
        except Exception:
            strategy_flag = _env_flag(FLAG_STRATEGY_CASCADE, "true")
    else:
        strategy_flag = _env_flag(FLAG_STRATEGY_CASCADE, "true")

    policy_flag = _env_flag(FLAG_POLICY_2026, "true")

    # Prefer tax_paths helper so ops match product gate (default true).
    tax_flag = _env_flag(FLAG_TAX_PATHS, "true")
    if tax_mod:
        try:
            from core.strategy.tax_paths import tax_paths_enabled as _tax_on

            tax_flag = bool(_tax_on())
        except Exception:
            pass

    # Prefer bk2021 stub helper so ops match product gate (default false).
    bk_flag = _env_flag(FLAG_BK2021, "false")
    if bk_mod:
        try:
            from core.b2b.bk2021_stub import bk2021_enabled as _bk_on

            bk_flag = bool(_bk_on())
        except Exception:
            pass

    strategy_cascade_available = bool(strategy_flag and recommend_mod)
    policy_2026_enabled = bool(policy_flag and policy_mod)
    tax_paths_available = bool(tax_flag and tax_mod)
    # Module-only surfaces (Direct EU has no product flag; BK stub = importable).
    direct_eu_module = bool(eu_mod)
    # Task-canonical name: bk2021_stub_available (module present); product gate is bk2021_enabled.
    bk2021_stub_available = bool(bk_mod)

    # Core spine/cascade/policy trio drives present_count / expected_count.
    present_count = sum((spine_mod, recommend_mod, policy_mod))
    modules = {
        "eligibility_spine": spine_mod,
        "strategy_recommend": recommend_mod,
        "policy_2026": policy_mod,
        "tax_paths": tax_mod,
        "direct_eu": eu_mod,
        "bk2021_stub": bk_mod,
        "present_count": present_count,
        "expected_count": len(_ELIGIBILITY_STRATEGY_MODULES),
        "all_present": present_count == len(_ELIGIBILITY_STRATEGY_MODULES),
    }

    coverage: Dict[str, Any] = {
        "note": "counts projects with eligibility/strategy snapshots in external_context",
        "with_eligibility_snapshot": 0,
        "with_strategy_path": 0,
        "with_de_minimis_source": 0,
        "with_msp_confidence_ge_0_7": 0,
        "projects_scanned_for_firm_metrics": 0,
    }
    if rows is not None:
        with_elig = 0
        with_strat = 0
        with_dm_src = 0
        with_msp_hi = 0
        n_rows = 0
        for row in rows:
            n_rows += 1
            ext = _ext(row)
            cd = ext.get("company_data") if isinstance(ext.get("company_data"), dict) else {}
            has_elig = isinstance(ext.get("eligibility"), dict) or bool(cd.get("sudop"))
            if has_elig:
                with_elig += 1
            if ext.get("strategy_path") or ext.get("strategy"):
                with_strat += 1
            dm_src, msp_conf = _firm_metric_signals(ext)
            if dm_src:
                with_dm_src += 1
            if msp_conf is not None and msp_conf >= 0.7:
                with_msp_hi += 1
        coverage["with_eligibility_snapshot"] = with_elig
        coverage["with_strategy_path"] = with_strat
        coverage["with_de_minimis_source"] = with_dm_src
        coverage["with_msp_confidence_ge_0_7"] = with_msp_hi
        coverage["projects_scanned_for_firm_metrics"] = n_rows

    return {
        # Top-level boolean keys required by instrument-stats / ops consumers
        "eligibility_spine_enabled": spine_enabled,
        "strategy_cascade_available": strategy_cascade_available,
        "policy_2026_enabled": policy_2026_enabled,
        "tax_paths_available": tax_paths_available,
        "direct_eu_module": direct_eu_module,
        "bk2021_stub_available": bk2021_stub_available,
        # Short alias kept for earlier consumers / module map parity.
        "bk2021_stub": bk2021_stub_available,
        # Product enable for BK2021 (default off); distinct from module presence.
        "bk2021_enabled": bool(bk_flag),
        "flag_names": {
            "eligibility_spine": FLAG_ELIGIBILITY_SPINE,
            "strategy_cascade": FLAG_STRATEGY_CASCADE,
            "policy_2026": FLAG_POLICY_2026,
            "tax_paths": FLAG_TAX_PATHS,
            "bk2021": FLAG_BK2021,
        },
        "env_flags": {
            FLAG_ELIGIBILITY_SPINE: os.environ.get(FLAG_ELIGIBILITY_SPINE, "true"),
            FLAG_STRATEGY_CASCADE: os.environ.get(FLAG_STRATEGY_CASCADE, "true"),
            FLAG_POLICY_2026: os.environ.get(FLAG_POLICY_2026, "true"),
            FLAG_TAX_PATHS: os.environ.get(FLAG_TAX_PATHS, "true"),
            FLAG_BK2021: os.environ.get(FLAG_BK2021, "false"),
        },
        "modules": modules,
        "coverage": coverage,
    }


def compute_instrument_project_metrics(
    db: Session,
    *,
    limit: int = 2000,
) -> Dict[str, Any]:
    """
    Aggregate instrument_schema / readiness signals across recent projects.
    Pure DB scan — no network.
    """
    from core.projects.models import Project
    from core.projects.data_completeness import evaluate_data_completeness

    try:
        q = db.query(Project).order_by(Project.created_at.desc())
    except Exception:
        q = db.query(Project)
    rows = q.limit(limit).all()

    families: Counter[str] = Counter()
    statuses: Counter[str] = Counter()
    grounding: Counter[str] = Counter()
    dossier_levels: Counter[str] = Counter()
    with_schema = 0
    with_legal_refs = 0
    allow_smart = 0
    structure_only = 0
    ready_count = 0
    blocked_missing = 0
    blocked_dossier = 0
    instrument_mismatch = 0
    mismatch_scanned = 0

    for row in rows:
        ext = _ext(row)
        schema = ext.get("instrument_schema") if isinstance(ext.get("instrument_schema"), dict) else {}
        family = (
            (schema.get("family") if schema else None)
            or ext.get("instrument_program_type")
            or getattr(row, "program_type", None)
            or "UNKNOWN"
        )
        families[str(family).upper()] += 1

        if schema:
            with_schema += 1
            if schema.get("legal_references"):
                with_legal_refs += 1
            if schema.get("allow_smart_modules"):
                allow_smart += 1

        gm = str(ext.get("grounding_mode") or "").lower() or "unset"
        grounding[gm] += 1
        if gm == "structure_only":
            structure_only += 1

        dlevel = (
            (ext.get("program_dossier") or {}).get("readiness", {}).get("level")
            if isinstance(ext.get("program_dossier"), dict)
            else ext.get("dossier_readiness")
        )
        dossier_levels[str(dlevel or "unknown")] += 1

        # §10 instrument mismatch rate (advisor / pure detect / explicit flag)
        mismatch_scanned += 1
        mm = detect_project_instrument_mismatch(row, ext)
        if mm.get("mismatch"):
            instrument_mismatch += 1

        try:
            r = evaluate_data_completeness(
                external_context=ext,
                instrument_schema=schema or None,
                description=str(getattr(row, "description", None) or ""),
                title=str(getattr(row, "title", None) or ""),
            )
            st = str(r.get("status") or "unknown")
            statuses[st] += 1
            if r.get("full_autopilot_allowed"):
                ready_count += 1
            if st == "blocked_missing_fields":
                blocked_missing += 1
            if st == "blocked_dossier":
                blocked_dossier += 1
        except Exception:
            statuses["eval_error"] += 1

    total = max(len(rows), 1)
    n = len(rows)
    es = build_eligibility_strategy_ops_snapshot(rows)
    mm_denom = max(mismatch_scanned, 1)

    return {
        "total_projects_scanned": n,
        "with_instrument_schema": with_schema,
        "pct_with_schema": round(100.0 * with_schema / total, 1),
        "with_legal_references": with_legal_refs,
        "pct_with_legal_refs": round(100.0 * with_legal_refs / total, 1),
        "ready_to_generate_count": ready_count,
        "pct_ready_to_generate": round(100.0 * ready_count / total, 1),
        "structure_only_count": structure_only,
        "pct_structure_only": round(100.0 * structure_only / total, 1),
        "blocked_missing_fields": blocked_missing,
        "blocked_dossier": blocked_dossier,
        "allow_smart_modules_count": allow_smart,
        "instrument_mismatch_count": instrument_mismatch,
        "instrument_mismatch_scanned": mismatch_scanned,
        "pct_instrument_mismatch": (
            round(100.0 * instrument_mismatch / mm_denom, 1) if n > 0 else None
        ),
        "families": dict(families.most_common(20)),
        "readiness_statuses": dict(statuses),
        "grounding_modes": dict(grounding),
        "dossier_levels": dict(dossier_levels),
        # Explicit boolean keys for dashboards / instrument-stats consumers
        "eligibility_spine_enabled": es["eligibility_spine_enabled"],
        "strategy_cascade_available": es["strategy_cascade_available"],
        "policy_2026_enabled": es["policy_2026_enabled"],
        "tax_paths_available": es["tax_paths_available"],
        "direct_eu_module": es["direct_eu_module"],
        "bk2021_stub_available": es["bk2021_stub_available"],
        "bk2021_stub": es["bk2021_stub"],
        "bk2021_enabled": es["bk2021_enabled"],
        "eligibility_strategy_modules": es["modules"],
        "flags": {
            "ALLOW_SMART_TEMPLATE_FALLBACK": os.environ.get(
                "ALLOW_SMART_TEMPLATE_FALLBACK", "false"
            ),
            "REQUIRE_DOSSIER_OR_CONSENT": os.environ.get("REQUIRE_DOSSIER_OR_CONSENT", "true"),
            "ALLOW_STRUCTURE_WITHOUT_REGULATION": os.environ.get(
                "ALLOW_STRUCTURE_WITHOUT_REGULATION", "true"
            ),
            FLAG_ELIGIBILITY_SPINE: es["env_flags"][FLAG_ELIGIBILITY_SPINE],
            FLAG_STRATEGY_CASCADE: es["env_flags"][FLAG_STRATEGY_CASCADE],
            FLAG_POLICY_2026: es["env_flags"][FLAG_POLICY_2026],
            FLAG_TAX_PATHS: es["env_flags"][FLAG_TAX_PATHS],
            FLAG_BK2021: es["env_flags"][FLAG_BK2021],
        },
        "flag_names": es["flag_names"],
        "eligibility_strategy_coverage": es["coverage"],
    }


def gold_family_coverage() -> Dict[str, Any]:
    """Static inventory of gold instrument families shipped in code."""
    from core.projects.instrument_schema import _GOLD, list_gold_families

    families = list_gold_families()
    return {
        "families": families,
        "count": len(families),
        "meets_dod_min_12": len(families) >= 12,
        "allow_smart_only": [
            k for k, v in _GOLD.items() if v.get("allow_smart_modules")
        ],
    }


# PLAN_ECOSYSTEM §10 numeric targets (reportable; live values may be null without data).
PLAN_12M_SLA_TARGETS: Dict[str, Any] = {
    "pct_projects_with_instrument_schema": {"d90": 80.0, "d365": 95.0},
    "pct_full_autopilot_ready_gated": {"d90": 90.0, "d365": 95.0},
    "pct_structure_only_of_generations": {
        "d90": None,
        "d365": 15.0,
        "direction": "lower_is_better",
    },
    "pct_active_catalog_not_blind": {"d90": 50.0, "d365": 70.0},
    "pct_firms_with_de_minimis_source": {"d90": 60.0, "d365": 90.0},
    "pct_firms_msp_confidence_ge_0_7": {"d90": 50.0, "d365": 80.0},
    # Absolute rate targets for reportable ops (plan narrative is relative ↓50%/↓80%).
    "pct_instrument_mismatch": {
        "d90": 25.0,
        "d365": 10.0,
        "direction": "lower_is_better",
        "note": "reportable absolute rate; plan also tracks relative reduction vs baseline",
    },
}


def _measured_value_for_target(key: str, measured: Dict[str, Any]) -> Any:
    """Map PLAN target keys to measured field names."""
    if key == "pct_full_autopilot_ready_gated":
        return measured.get("pct_ready_to_generate")
    if key == "pct_structure_only_of_generations":
        return measured.get("pct_structure_only")
    return measured.get(key)


def _target_hits(
    measured: Dict[str, Any],
    *,
    horizon: str,
) -> Dict[str, Any]:
    hits: Dict[str, Any] = {}
    for key, tgt in PLAN_12M_SLA_TARGETS.items():
        thr = tgt.get(horizon)
        val = _measured_value_for_target(key, measured)
        if thr is None or val is None:
            hits[key] = None
            continue
        direction = tgt.get("direction") or "higher_is_better"
        if direction == "lower_is_better":
            hits[key] = bool(float(val) <= float(thr))
        else:
            hits[key] = bool(float(val) >= float(thr))
    return hits


def build_plan_12m_metrics_snapshot(
    *,
    project_metrics: Optional[Dict[str, Any]] = None,
    catalog_stats: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """
    Reportable counters for PLAN_ECOSYSTEM §10 / §13 without inventing live SLAs.

    Returns:
    - ``targets``: plan targets (static)
    - ``measured``: values from shipped project/catalog metrics when provided
    - ``residual``: metrics that cannot be proven as production SLAs here
    - cascade module/flag presence for admin dashboards
    """
    es = build_eligibility_strategy_ops_snapshot()
    gold = gold_family_coverage()
    pm = dict(project_metrics or {})
    cs = dict(catalog_stats or {})

    measured: Dict[str, Any] = {
        "gold_family_count": gold["count"],
        "gold_meets_dod_min_12": gold["meets_dod_min_12"],
        "pct_projects_with_instrument_schema": pm.get("pct_with_schema"),
        "pct_ready_to_generate": pm.get("pct_ready_to_generate"),
        "pct_structure_only": pm.get("pct_structure_only"),
        "pct_instrument_mismatch": pm.get("pct_instrument_mismatch"),
        "instrument_mismatch_count": pm.get("instrument_mismatch_count"),
        "instrument_mismatch_scanned": pm.get("instrument_mismatch_scanned"),
        "pct_firms_with_eligibility_snapshot": None,
        "pct_firms_with_de_minimis_source": None,
        "pct_firms_msp_confidence_ge_0_7": None,
        "pct_active_catalog_not_blind": None,
        "dossier_readiness": None,
    }

    # Live instrument-stats path embeds coverage on project_metrics (from
    # compute_instrument_project_metrics → eligibility_strategy_coverage).
    # Do NOT use build_eligibility_strategy_ops_snapshot() with no rows —
    # that always yields zeros and would invent a false 0.0% here.
    cov: Dict[str, Any] = {}
    if isinstance(pm.get("eligibility_strategy_coverage"), dict):
        cov = pm["eligibility_strategy_coverage"]
    elif isinstance(pm.get("coverage"), dict):
        cov = pm["coverage"]
    n_proj = int(pm.get("total_projects_scanned") or 0)
    # Prefer firm-metric scan count when present (same as project rows scanned).
    n_firm = int(cov.get("projects_scanned_for_firm_metrics") or 0) or n_proj
    if n_proj > 0 and isinstance(cov.get("with_eligibility_snapshot"), int):
        measured["pct_firms_with_eligibility_snapshot"] = round(
            100.0 * int(cov["with_eligibility_snapshot"]) / n_proj, 1
        )
    if n_firm > 0 and isinstance(cov.get("with_de_minimis_source"), int):
        measured["pct_firms_with_de_minimis_source"] = round(
            100.0 * int(cov["with_de_minimis_source"]) / n_firm, 1
        )
    if n_firm > 0 and isinstance(cov.get("with_msp_confidence_ge_0_7"), int):
        measured["pct_firms_msp_confidence_ge_0_7"] = round(
            100.0 * int(cov["with_msp_confidence_ge_0_7"]) / n_firm, 1
        )

    dossier = cs.get("dossier_readiness") if isinstance(cs.get("dossier_readiness"), dict) else {}
    if dossier:
        measured["dossier_readiness"] = dossier
        # Prefer explicit non-blind % from compute_readiness_distribution
        for key in (
            "pct_not_blind",
            "pct_ready_or_partial",
            "percent_not_blind",
            "not_blind_pct",
        ):
            if dossier.get(key) is not None:
                measured["pct_active_catalog_not_blind"] = dossier.get(key)
                break
        # Derive from level counts / ready+partial fields when present
        if measured["pct_active_catalog_not_blind"] is None:
            if dossier.get("sample_size") == 0 or dossier.get("total") == 0:
                measured["pct_active_catalog_not_blind"] = None
            else:
                levels = dossier.get("levels") or dossier.get("distribution") or {}
                if isinstance(levels, dict) and levels:
                    total = sum(int(v or 0) for v in levels.values()) or 0
                    blind = int(levels.get("blind") or 0)
                    if total > 0:
                        measured["pct_active_catalog_not_blind"] = round(
                            100.0 * (total - blind) / total, 1
                        )
                elif dossier.get("ready") is not None and dossier.get("partial") is not None:
                    total = int(dossier.get("total") or 0) or (
                        int(dossier.get("ready") or 0)
                        + int(dossier.get("partial") or 0)
                        + int(dossier.get("blind") or 0)
                        + int(dossier.get("unknown") or 0)
                    )
                    if total > 0:
                        not_blind = int(dossier.get("ready") or 0) + int(dossier.get("partial") or 0)
                        measured["pct_active_catalog_not_blind"] = round(
                            100.0 * not_blind / total, 1
                        )
        measured["meets_dod_70_not_blind"] = dossier.get("meets_dod_70_not_blind")
        if measured["meets_dod_70_not_blind"] is None and measured["pct_active_catalog_not_blind"] is not None:
            sample = int(dossier.get("sample_size") or dossier.get("total") or 0)
            measured["meets_dod_70_not_blind"] = bool(
                sample > 0 and float(measured["pct_active_catalog_not_blind"]) >= 70.0
            )

    residual = [
        "sla_claim_allowed stays false until production SLAs are proven over the full window "
        "(not fixture/seed alone).",
        "Relative instrument-mismatch reduction (↓50%/↓80% vs historical baseline) needs "
        "longitudinal ops window — absolute pct_instrument_mismatch is reportable when projects scanned.",
    ]
    if measured["pct_active_catalog_not_blind"] is None:
        residual.append(
            "pct_active_catalog_not_blind not measured (empty catalog / no dossier distribution)."
        )
    if measured["pct_firms_with_eligibility_snapshot"] is None:
        residual.append("pct_firms_with_eligibility_snapshot not measured (no project sample).")
    if measured["pct_firms_with_de_minimis_source"] is None:
        residual.append(
            "pct_firms_with_de_minimis_source not measured (no project firm sample)."
        )
    if measured["pct_firms_msp_confidence_ge_0_7"] is None:
        residual.append(
            "pct_firms_msp_confidence_ge_0_7 not measured (no project firm sample)."
        )
    if measured.get("pct_instrument_mismatch") is None:
        residual.append(
            "pct_instrument_mismatch not measured (no project sample)."
        )

    # d90 / d365 target attainment when measured — still not a prod multi-month SLA claim.
    d90_hits = _target_hits(measured, horizon="d90")
    d365_hits = _target_hits(measured, horizon="d365")

    # §13 product DoD status for admin dashboards (code+measured; not prod multi-month SLA).
    dossier_sample = 0
    if isinstance(measured.get("dossier_readiness"), dict):
        dossier_sample = int(
            measured["dossier_readiness"].get("sample_size")
            or measured["dossier_readiness"].get("total")
            or 0
        )
    if measured.get("meets_dod_70_not_blind"):
        s13_5 = "PASS"
    elif dossier_sample == 0 or measured["pct_active_catalog_not_blind"] is None:
        s13_5 = "external-blocker"
    else:
        s13_5 = "FAIL"

    section_13_status: Dict[str, str] = {
        "1_eligibility_spine": "PASS" if es["eligibility_spine_enabled"] else "FAIL",
        "2_strategy_cascade_match": (
            "PASS"
            if es["strategy_cascade_available"] and es["policy_2026_enabled"]
            else "FAIL"
        ),
        "3_gold_families_ge_12": "PASS" if gold["meets_dod_min_12"] else "FAIL",
        "4_tax_paths": "PASS" if es["tax_paths_available"] else "FAIL",
        "5_dossier_not_blind_70": s13_5,
        "6_direct_eu": "PASS" if es["direct_eu_module"] else "FAIL",
        "7_metrics_admin": "PASS",  # this snapshot is the metrics surface
        "8_bk2021_surface": "PASS" if es["bk2021_stub_available"] else "FAIL",
        "9_docs_test_path": "PASS",  # docs verified in TEST_PATH / residual matrix
        "10_no_autopilot_without_gates": (
            "PASS" if es["eligibility_spine_enabled"] else "FAIL"
        ),
    }
    dod_all_pass = all(
        v == "PASS" or v == "external-blocker" for v in section_13_status.values()
    ) and all(v != "FAIL" for v in section_13_status.values())
    # Strict product DoD for CI seed path: every §13 PASS (no FAIL, blockers only if empty).
    dod_product_pass = all(v == "PASS" for v in section_13_status.values())

    return {
        "version": 3,
        "plan_ref": "PLAN_ECOSYSTEM_2026 §10 / §13",
        "targets": PLAN_12M_SLA_TARGETS,
        "measured": measured,
        "d90_target_hits": d90_hits,
        "d365_target_hits": d365_hits,
        "section_13_status": section_13_status,
        "dod_all_pass": dod_all_pass,
        "dod_product_pass": dod_product_pass,
        "residual": residual,
        "cascade": {
            "eligibility_spine_enabled": es["eligibility_spine_enabled"],
            "strategy_cascade_available": es["strategy_cascade_available"],
            "policy_2026_enabled": es["policy_2026_enabled"],
            "tax_paths_available": es["tax_paths_available"],
            "direct_eu_module": es["direct_eu_module"],
            "bk2021_stub_available": es["bk2021_stub_available"],
            "bk2021_enabled": es["bk2021_enabled"],
            "gold_family_count": gold["count"],
            "gold_meets_dod_min_12": gold["meets_dod_min_12"],
            "gold_families": gold["families"],
        },
        "sla_claim_allowed": False,
        "note": (
            "Reportable metrics surface for 12-month ecosystem ops. "
            "Does not claim production SLAs are met unless measured proves it. "
            "section_13_status is product/code DoD, not multi-month production SLA."
        ),
    }