aaronrene / overseer-kit public
surface.py python
165 lines 5.2 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 2 days ago
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 }
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 2 days ago