aaronrene / overseer-kit public

authorize.py file-level

at sha256:8 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 💥 blast risk
sha256:2 fix(muse): use rev-parse for muse-only branch status · · Sep 22, 2026
1 """Shared AFF authorization helper (§AFF.5.6 / §AFF.7.1)."""
2
3 from __future__ import annotations
4
5 import re
6 from dataclasses import dataclass
7 from pathlib import Path
8 from typing import Literal
9
10 from adapters.config import OverseerConfig
11 from cli.paths import PathEscapeError, confine_path
12 from tools.freeze_authorization.resolve import freeze_authorization_state
13 from tools.freeze_reviewer.artifact import artifact_digest, parse_artifact
14 from tools.honesty.gate import honesty_module_disabled
15 from tools.honesty.ledger import verify_chain
16 from tools.honesty.ledger_io import read_ledger_entries
17 from tools.honesty.validate import find_latest_adversarial_freeze_verdict
18
19 AffAuthState = Literal["off", "pass", "skipped", "pending", "absent"]
20
21 _DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
22
23 _AGGREGATE_RANK = {
24 "absent": 0,
25 "pending": 1,
26 "skipped": 2,
27 "pass": 3,
28 }
29
30
31 @dataclass(frozen=True)
32 class AdversarialAuthorization:
33 """Shared AFF authorization state for next_regen + status (§AFF.5.6)."""
34
35 state: AffAuthState
36 matched_entry_hash: str | None = None
37 matched_via: str | None = None
38
39
40 def aff_hold_bypassed(config: OverseerConfig) -> bool:
41 """True when honesty is disabled or resolved adversarial_freeze is off."""
42 if honesty_module_disabled(config):
43 return True
44 return config.honesty.adversarial_freeze == "off"
45
46
47 def author_loop_complete(
48 frv_state: str,
49 *,
50 stamp_digest: str | None,
51 current_digest: str,
52 ) -> bool:
53 """Trigger-B fresh-stamp predicate (§AFF.7.1 ADV2-M1)."""
54 if frv_state == "substantive":
55 return True
56 if frv_state == "mechanical_only":
57 if not isinstance(stamp_digest, str) or not _DIGEST_RE.fullmatch(stamp_digest):
58 return False
59 return stamp_digest == current_digest
60 return False
61
62
63 def frv_authorizing_freeze_paths(
64 repo_root: Path,
65 candidates: list[Path],
66 *,
67 phase_id: str,
68 config: OverseerConfig,
69 ) -> list[Path]:
70 """Discover-order paths whose freeze_authorization_state is substantive."""
71 authorizing: list[Path] = []
72 for path in candidates:
73 auth = freeze_authorization_state(
74 repo_root, path, phase_id=phase_id, config=config
75 )
76 if auth.state == "substantive":
77 authorizing.append(path)
78 return authorizing
79
80
81 def aggregate_adversarial_authorization_states(states: list[str]) -> str:
82 """Worst-state aggregate: absent > pending > skipped > pass (§AFF.5.6)."""
83 ranked = [s for s in states if s in _AGGREGATE_RANK]
84 if not ranked:
85 raise ValueError("aggregate requires at least one non-off state")
86 return min(ranked, key=lambda s: _AGGREGATE_RANK[s])
87
88
89 def adversarial_authorization_state(
90 repo_root: Path,
91 artifact_path: Path,
92 *,
93 config: OverseerConfig,
94 ) -> AdversarialAuthorization:
95 """Evaluate AFF authorization for a freeze artifact (§AFF.5.6)."""
96 try:
97 return _adversarial_authorization_state_impl(
98 repo_root, artifact_path, config=config
99 )
100 except Exception:
101 mode = config.honesty.adversarial_freeze
102 if aff_hold_bypassed(config):
103 return AdversarialAuthorization(state="off")
104 if mode in {"suggest", "require"}:
105 return AdversarialAuthorization(state="absent")
106 return AdversarialAuthorization(state="off")
107
108
109 def _resolve_ledger_path_safe(config: OverseerConfig, repo_root: Path) -> Path | None:
110 try:
111 ledger = config.honesty.ledger
112 if ledger is None or not str(ledger).strip():
113 return None
114 return confine_path(repo_root, str(ledger))
115 except (PathEscapeError, ValueError, TypeError, AttributeError, OSError):
116 return None
117
118
119 def _adversarial_authorization_state_impl(
120 repo_root: Path,
121 artifact_path: Path,
122 *,
123 config: OverseerConfig,
124 ) -> AdversarialAuthorization:
125 mode = config.honesty.adversarial_freeze
126 if aff_hold_bypassed(config):
127 return AdversarialAuthorization(state="off")
128
129 try:
130 rel = artifact_path.resolve().relative_to(repo_root.resolve()).as_posix()
131 except (OSError, ValueError):
132 return AdversarialAuthorization(state="absent")
133
134 try:
135 parsed = parse_artifact(artifact_path, rel_path=rel)
136 digest = artifact_digest(parsed)
137 except (ValueError, OSError, UnicodeError, Exception):
138 return AdversarialAuthorization(state="absent")
139
140 ledger_path = _resolve_ledger_path_safe(config, repo_root)
141 if ledger_path is None or not ledger_path.is_file() or ledger_path.stat().st_size == 0:
142 return AdversarialAuthorization(state="pending")
143
144 try:
145 entries = read_ledger_entries(ledger_path)
146 except (ValueError, OSError):
147 return AdversarialAuthorization(state="absent")
148
149 chain_code = verify_chain(
150 entries,
151 regime=config.vcs.regime,
152 require_agent_signature=config.honesty.require_agent_signature,
153 )
154 if chain_code != 0:
155 return AdversarialAuthorization(state="absent")
156
157 winner = find_latest_adversarial_freeze_verdict(
158 entries,
159 frozen_spec=rel,
160 artifact_digest=digest,
161 )
162 return _map_winner(winner, mode=mode)
163
164
165 def _map_winner(
166 winner: dict | None,
167 *,
168 mode: str,
169 ) -> AdversarialAuthorization:
170 if winner is None:
171 return AdversarialAuthorization(state="pending")
172 verdict = winner.get("aff_verdict")
173 entry_hash = winner.get("entry_hash")
174 hash_s = entry_hash if isinstance(entry_hash, str) else None
175 if verdict == "pass":
176 return AdversarialAuthorization(
177 state="pass", matched_entry_hash=hash_s, matched_via="pass"
178 )
179 if verdict == "skip":
180 if mode == "suggest":
181 return AdversarialAuthorization(
182 state="skipped", matched_entry_hash=hash_s, matched_via="skip"
183 )
184 return AdversarialAuthorization(state="pending")
185 # findings | blocked
186 return AdversarialAuthorization(state="pending")