Some checks failed
tamq-ci / test (push) Failing after 36s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02f34-ead4-7a60-85cd-d7f08e22fa0e
43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import tomllib
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .config import config_path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PolicyProfile:
|
|
name: str
|
|
allow: frozenset[str]
|
|
require_human: frozenset[str]
|
|
safety_gated_max_attempts: int = 0
|
|
delivery_ack_mode: str = "injected"
|
|
|
|
|
|
DEFAULT = PolicyProfile(
|
|
"default",
|
|
frozenset({"statehub_read", "repo_inspect", "repo_edit", "local_checks", "state_updates", "coordination_messages", "tmux_wake", "transient_retry"}),
|
|
frozenset({"secrets", "destructive", "live_external", "scope_change", "ambiguous_ownership", "external_publish", "financial", "legal"}),
|
|
)
|
|
|
|
|
|
def load_profile(path: Path | None = None, selected: str = "default") -> PolicyProfile:
|
|
path = path or config_path()
|
|
if not path.exists():
|
|
return DEFAULT
|
|
with path.open("rb") as stream:
|
|
data = tomllib.load(stream)
|
|
profile = data.get("policy", {}).get("profiles", {}).get(selected)
|
|
if not profile:
|
|
if selected == "default":
|
|
return DEFAULT
|
|
raise ValueError(f"unknown policy profile: {selected}")
|
|
attempts = int(profile.get("safety_gated_max_attempts", 0))
|
|
if not 0 <= attempts <= 9:
|
|
raise ValueError("safety_gated_max_attempts must be between 0 and 9")
|
|
ack_mode = profile.get("delivery_ack_mode", "injected")
|
|
if ack_mode not in {"injected", "acknowledged"}:
|
|
raise ValueError("delivery_ack_mode must be injected or acknowledged")
|
|
return PolicyProfile(selected, frozenset(profile.get("allow", [])), frozenset(profile.get("require_human", [])), attempts, ack_mode)
|