"""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." ), }