surface.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Active-slice spend-awareness report builder (§PC.7).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from dataclasses import dataclass |
| 6 | from pathlib import Path |
| 7 | |
| 8 | from adapters.config import OverseerConfig |
| 9 | from cli.paths import PathEscapeError, confine_path |
| 10 | from tools.cost_awareness.derive import derive_cost_view |
| 11 | from tools.cost_awareness.normalize import gate_for_phase, model_label_for_phase, normalize_phase_tier |
| 12 | from tools.governance_gates import scan_governance_gates |
| 13 | from tools.model_routing.labels import ( |
| 14 | RoutingPolicyError, |
| 15 | load_label_ids, |
| 16 | load_model_tier_cost_bands, |
| 17 | ) |
| 18 | from tools.model_routing.policy import load_routing_policy |
| 19 | from tools.model_routing.resolve import resolve_route |
| 20 | from tools.model_routing.types import RouteSelector |
| 21 | |
| 22 | |
| 23 | @dataclass(frozen=True) |
| 24 | class CostSlice: |
| 25 | phase_id: str |
| 26 | phase_tier: str | None |
| 27 | gate: str | None |
| 28 | route_id: str |
| 29 | model_tier: str |
| 30 | cost_class: str |
| 31 | paid_step_before_spend: bool |
| 32 | |
| 33 | |
| 34 | @dataclass(frozen=True) |
| 35 | class CostAwarenessReport: |
| 36 | enabled: bool |
| 37 | policy: str | None = None |
| 38 | slices: tuple[CostSlice, ...] = () |
| 39 | invalid: bool = False |
| 40 | violation: str | None = None |
| 41 | exit_code: int | None = None |
| 42 | |
| 43 | |
| 44 | def build_cost_awareness_report( |
| 45 | config: OverseerConfig, |
| 46 | repo_root: Path, |
| 47 | *, |
| 48 | kit_root: Path, |
| 49 | handover_text: str | None = None, |
| 50 | roadmap_text: str | None = None, |
| 51 | fail_closed: bool = False, |
| 52 | ) -> CostAwarenessReport: |
| 53 | """Build the active-slice spend-awareness report. |
| 54 | |
| 55 | When ``fail_closed`` is True (``overseer route`` path), malformed cost metadata |
| 56 | raises ``RoutingPolicyError`` with exit ``32``. On informational surfaces |
| 57 | (``status``, ``governance-sync``), malformed metadata degrades to a warning. |
| 58 | """ |
| 59 | if not config.cost_awareness.enabled: |
| 60 | return CostAwarenessReport(enabled=False) |
| 61 | |
| 62 | policy_rel = config.model_routing.policy |
| 63 | try: |
| 64 | policy_path = confine_path(repo_root, policy_rel) |
| 65 | except PathEscapeError: |
| 66 | violation = "model_routing.policy path escapes repo root" |
| 67 | if fail_closed: |
| 68 | raise RoutingPolicyError(violation, exit_code=31, citation=policy_rel) |
| 69 | return CostAwarenessReport( |
| 70 | enabled=True, |
| 71 | policy=policy_rel, |
| 72 | invalid=True, |
| 73 | violation=violation, |
| 74 | ) |
| 75 | |
| 76 | try: |
| 77 | cost_bands = load_model_tier_cost_bands(kit_root, fail_closed=fail_closed) |
| 78 | except RoutingPolicyError as exc: |
| 79 | if fail_closed: |
| 80 | raise |
| 81 | return CostAwarenessReport( |
| 82 | enabled=True, |
| 83 | policy=policy_rel, |
| 84 | invalid=True, |
| 85 | violation=exc.message, |
| 86 | ) |
| 87 | |
| 88 | try: |
| 89 | policy = load_routing_policy(policy_path, kit_root=kit_root) |
| 90 | except RoutingPolicyError as exc: |
| 91 | if fail_closed: |
| 92 | raise |
| 93 | return CostAwarenessReport( |
| 94 | enabled=True, |
| 95 | policy=policy_rel, |
| 96 | invalid=exc.exit_code != 31, |
| 97 | violation=exc.message, |
| 98 | exit_code=exc.exit_code, |
| 99 | ) |
| 100 | |
| 101 | gate_scan = scan_governance_gates( |
| 102 | config, |
| 103 | repo_root, |
| 104 | handover_text=handover_text, |
| 105 | roadmap_text=roadmap_text, |
| 106 | ) |
| 107 | label_ids = load_label_ids(kit_root) |
| 108 | roadmap = roadmap_text |
| 109 | if roadmap is None: |
| 110 | roadmap_path = repo_root / config.repo.root_relative_docs / config.docs.roadmap |
| 111 | if roadmap_path.is_file(): |
| 112 | roadmap = roadmap_path.read_text(encoding="utf-8") |
| 113 | |
| 114 | slices: list[CostSlice] = [] |
| 115 | for phase_id in gate_scan.active_phases: |
| 116 | model_label = model_label_for_phase(roadmap, phase_id) |
| 117 | phase_tier = ( |
| 118 | normalize_phase_tier(model_label, label_ids=label_ids) if model_label else None |
| 119 | ) |
| 120 | gate = gate_for_phase(gate_scan.pending, phase_id) |
| 121 | query = RouteSelector(position=None, phase_tier=phase_tier, gate=gate) |
| 122 | decision = resolve_route(policy, query) |
| 123 | cost_class, paid = derive_cost_view(decision.model_tier, cost_bands) |
| 124 | slices.append( |
| 125 | CostSlice( |
| 126 | phase_id=phase_id, |
| 127 | phase_tier=phase_tier, |
| 128 | gate=gate, |
| 129 | route_id=decision.route_id, |
| 130 | model_tier=decision.model_tier, |
| 131 | cost_class=cost_class, |
| 132 | paid_step_before_spend=paid, |
| 133 | ) |
| 134 | ) |
| 135 | |
| 136 | return CostAwarenessReport(enabled=True, policy=policy_rel, slices=tuple(slices)) |
| 137 | |
| 138 | |
| 139 | def cost_awareness_payload(report: CostAwarenessReport) -> dict: |
| 140 | """JSON payload for ``overseer status --json``.""" |
| 141 | if not report.enabled: |
| 142 | return {"enabled": False} |
| 143 | if report.invalid: |
| 144 | return { |
| 145 | "enabled": True, |
| 146 | "policy": report.policy, |
| 147 | "invalid": True, |
| 148 | "violation": report.violation, |
| 149 | } |
| 150 | return { |
| 151 | "enabled": True, |
| 152 | "policy": report.policy, |
| 153 | "slices": [ |
| 154 | { |
| 155 | "phase_id": slice_.phase_id, |
| 156 | "phase_tier": slice_.phase_tier, |
| 157 | "gate": slice_.gate, |
| 158 | "route_id": slice_.route_id, |
| 159 | "model_tier": slice_.model_tier, |
| 160 | "cost_class": slice_.cost_class, |
| 161 | "paid_step_before_spend": slice_.paid_step_before_spend, |
| 162 | } |
| 163 | for slice_ in report.slices |
| 164 | ], |
| 165 | } |