aaronrene / overseer-kit public
resolve.py python
212 lines 7.2 KB
Raw
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 3 days ago
1 """Stamp resolution and freeze Auto-authorization state (§FRV.3.4 / §FRV.6.4)."""
2
3 from __future__ import annotations
4
5 from dataclasses import dataclass
6 from pathlib import Path
7 from typing import Any, Literal
8
9 from adapters.config import OverseerConfig
10 from cli.paths import PathEscapeError, confine_path
11 from tools.freeze_reviewer.artifact import (
12 artifact_digest,
13 extract_existing_stamp,
14 parse_artifact,
15 )
16 from tools.honesty.gate import honesty_module_disabled
17 from tools.honesty.ledger import verify_chain
18 from tools.honesty.ledger_io import read_ledger_entries
19 from tools.honesty.validate import find_matching_freeze_review
20
21 # §FRV.3.4 — kit-version property; flipping to False is a separate phase (not FRV-b).
22 # No config knob. Window: every kit version below 0.3.0.
23 ACCEPT_LEGACY_VERDICT_STAMP = True
24
25 AuthState = Literal[
26 "substantive",
27 "mechanical_only",
28 "non_pass",
29 "blocked_by_operator",
30 "absent",
31 ]
32
33 ResolvedKind = Literal["mechanical", "forged_substantive", "unreadable", "none"]
34
35
36 @dataclass(frozen=True)
37 class ResolvedStamp:
38 """Result of §FRV.3.4 stamp mapping resolution."""
39
40 kind: ResolvedKind
41 verdict: str | None = None
42 advisory: str | None = None
43
44
45 @dataclass(frozen=True)
46 class FreezeAuthorization:
47 """Shared authorization state for next_regen + governance_gates (§FRV.6.4)."""
48
49 state: AuthState
50 advisory: str | None = None
51 matched_entry_hash: str | None = None
52
53
54 def resolve_stamp_record(stamp_mapping: dict[str, Any] | None) -> ResolvedStamp:
55 """Resolve a stamp mapping per §FRV.3.4 (six-branch order)."""
56 if not isinstance(stamp_mapping, dict):
57 return ResolvedStamp(kind="none")
58
59 if "gate" in stamp_mapping:
60 gate = stamp_mapping.get("gate")
61 if gate == "mechanical":
62 verdict = stamp_mapping.get("mechanical_verdict")
63 verdict_s = str(verdict).strip() if verdict is not None else ""
64 return ResolvedStamp(kind="mechanical", verdict=verdict_s or None)
65 if gate == "substantive":
66 return ResolvedStamp(
67 kind="forged_substantive",
68 advisory="forged_substantive_gate",
69 )
70 if not isinstance(gate, str):
71 return ResolvedStamp(kind="unreadable", advisory="unreadable_gate")
72 return ResolvedStamp(kind="unreadable", advisory="unreadable_gate")
73
74 if "mechanical_verdict" in stamp_mapping:
75 verdict = stamp_mapping.get("mechanical_verdict")
76 verdict_s = str(verdict).strip() if verdict is not None else ""
77 return ResolvedStamp(kind="mechanical", verdict=verdict_s or None)
78
79 if "verdict" in stamp_mapping and ACCEPT_LEGACY_VERDICT_STAMP:
80 verdict = stamp_mapping.get("verdict")
81 verdict_s = str(verdict).strip() if verdict is not None else ""
82 return ResolvedStamp(kind="mechanical", verdict=verdict_s or None)
83
84 return ResolvedStamp(kind="none")
85
86
87 def _resolve_ledger_path_safe(config: OverseerConfig, repo_root: Path) -> Path | None:
88 """Mirror honesty.ledger._resolve_ledger_path; failures → None (non-authorizing)."""
89 try:
90 honesty = getattr(config, "honesty", None)
91 if honesty is None:
92 return None
93 ledger = getattr(honesty, "ledger", None)
94 if ledger is None or not str(ledger).strip():
95 return None
96 return confine_path(repo_root, str(ledger))
97 except (PathEscapeError, ValueError, TypeError, AttributeError, OSError):
98 return None
99
100
101 def freeze_authorization_state(
102 repo_root: Path,
103 artifact_path: Path,
104 *,
105 phase_id: str,
106 config: OverseerConfig,
107 ) -> FreezeAuthorization:
108 """Evaluate Auto-authorization state per §FRV.6.4 (fail-closed, never raise)."""
109 try:
110 return _freeze_authorization_state_impl(
111 repo_root, artifact_path, phase_id=phase_id, config=config
112 )
113 except Exception:
114 return FreezeAuthorization(state="absent")
115
116
117 def _freeze_authorization_state_impl(
118 repo_root: Path,
119 artifact_path: Path,
120 *,
121 phase_id: str,
122 config: OverseerConfig,
123 ) -> FreezeAuthorization:
124 # 1. Parse artifact
125 try:
126 rel = artifact_path.resolve().relative_to(repo_root.resolve()).as_posix()
127 except (OSError, ValueError):
128 try:
129 rel = artifact_path.as_posix()
130 except Exception:
131 return FreezeAuthorization(state="absent")
132
133 try:
134 parsed = parse_artifact(artifact_path, rel_path=rel)
135 except (ValueError, OSError, UnicodeError):
136 return FreezeAuthorization(state="absent")
137
138 # 2. auto_may_start (before ledger)
139 freeze_mapping = parsed.freeze_mapping
140 if isinstance(freeze_mapping, dict) and "auto_may_start" in freeze_mapping:
141 auto_val = freeze_mapping.get("auto_may_start")
142 if auto_val is True:
143 pass # no effect
144 elif auto_val is False:
145 return FreezeAuthorization(
146 state="blocked_by_operator",
147 advisory="operator_block",
148 )
149 else:
150 return FreezeAuthorization(
151 state="blocked_by_operator",
152 advisory="operator_block_malformed",
153 )
154
155 # 3. Digest
156 try:
157 digest = artifact_digest(parsed)
158 except Exception:
159 return FreezeAuthorization(state="absent")
160
161 frozen_spec = rel
162 advisory: str | None = None
163
164 # 4–6. Ledger match (only when honesty enabled)
165 if not honesty_module_disabled(config):
166 ledger_path = _resolve_ledger_path_safe(config, repo_root)
167 if ledger_path is not None and ledger_path.is_file():
168 try:
169 entries = read_ledger_entries(ledger_path)
170 except (ValueError, OSError):
171 entries = None
172 if entries is not None:
173 chain_code = verify_chain(entries)
174 if chain_code != 0:
175 advisory = "ledger_chain_broken"
176 else:
177 match = find_matching_freeze_review(
178 entries,
179 phase_id=phase_id,
180 frozen_spec=frozen_spec,
181 artifact_digest=digest,
182 )
183 if match is not None:
184 return FreezeAuthorization(
185 state="substantive",
186 matched_entry_hash=str(match.get("entry_hash") or "") or None,
187 )
188
189 # 7. Stamp resolve
190 stamp = extract_existing_stamp(parsed)
191 resolved = resolve_stamp_record(stamp)
192 if resolved.kind == "forged_substantive":
193 return FreezeAuthorization(
194 state="absent",
195 advisory=resolved.advisory or "forged_substantive_gate",
196 )
197 if resolved.kind == "unreadable":
198 return FreezeAuthorization(
199 state="absent",
200 advisory=resolved.advisory or "unreadable_gate",
201 )
202 if resolved.kind == "mechanical":
203 if resolved.verdict == "pass":
204 return FreezeAuthorization(
205 state="mechanical_only",
206 advisory=advisory,
207 )
208 if resolved.verdict:
209 return FreezeAuthorization(state="non_pass", advisory=advisory)
210 return FreezeAuthorization(state="absent", advisory=advisory)
211
212 return FreezeAuthorization(state="absent", advisory=advisory)
File History 1 commit
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 3 days ago