__init__.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """AFF fixture helpers (§AFF.14).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import shutil |
| 7 | from pathlib import Path |
| 8 | |
| 9 | import yaml |
| 10 | |
| 11 | from adapters.config import OverseerConfig, load_config |
| 12 | from tests.support import seed_honesty_repo, seed_muse_substrate |
| 13 | |
| 14 | FIXTURES = Path(__file__).resolve().parent |
| 15 | AFF_ENTRIES = FIXTURES / "entries" |
| 16 | AFF_FREEZE_REL = "docs/archive/phases/PHASE-AFF-ADVERSARIAL-FREEZE-HONESTY-GATE.md" |
| 17 | AFF_DIGEST_PLACEHOLDER = "sha256:" + ("a" * 64) |
| 18 | |
| 19 | |
| 20 | def load_aff_entry(name: str) -> dict: |
| 21 | """Load an adversarial_freeze entry fixture by filename.""" |
| 22 | return json.loads((AFF_ENTRIES / name).read_text(encoding="utf-8")) |
| 23 | |
| 24 | |
| 25 | def seed_aff_repo( |
| 26 | repo_root: Path, |
| 27 | *, |
| 28 | adversarial_freeze: str | None = "suggest", |
| 29 | honesty_enabled: bool = True, |
| 30 | human_escalation: list[str] | None = None, |
| 31 | omit_adversarial_freeze_key: bool = False, |
| 32 | regime_config: str | None = None, |
| 33 | write_freeze_artifact: bool = True, |
| 34 | ) -> OverseerConfig: |
| 35 | """Seed honesty repo with AFF gate config and archive freeze path.""" |
| 36 | seed_honesty_repo(repo_root) |
| 37 | cfg_path = repo_root / ".overseer" / "config.yaml" |
| 38 | data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) |
| 39 | data["honesty"]["enabled"] = honesty_enabled |
| 40 | if omit_adversarial_freeze_key: |
| 41 | data["honesty"].pop("adversarial_freeze", None) |
| 42 | elif adversarial_freeze is not None: |
| 43 | data["honesty"]["adversarial_freeze"] = adversarial_freeze |
| 44 | escalation = human_escalation if human_escalation is not None else ["security"] |
| 45 | data["freeze_contract"] = { |
| 46 | "enabled": True, |
| 47 | "reviewer": { |
| 48 | "mode": "agent", |
| 49 | "model": "thinking-high", |
| 50 | "provider": "local", |
| 51 | "fallback": "human", |
| 52 | }, |
| 53 | "human_escalation": list(escalation), |
| 54 | } |
| 55 | if "modules" in data and isinstance(data["modules"], dict): |
| 56 | honesty_mod = data["modules"].setdefault("honesty", {}) |
| 57 | if isinstance(honesty_mod, dict): |
| 58 | honesty_mod["enabled"] = honesty_enabled |
| 59 | if regime_config is not None: |
| 60 | regime_src = Path(__file__).resolve().parents[2] / "fixtures" / Path(regime_config).name |
| 61 | regime_data = yaml.safe_load(regime_src.read_text(encoding="utf-8")) |
| 62 | for key in ("vcs", "repo", "docs", "thresholds", "freeze_contract"): |
| 63 | if key in regime_data: |
| 64 | data[key] = regime_data[key] |
| 65 | data["freeze_contract"] = { |
| 66 | **data.get("freeze_contract", {}), |
| 67 | "enabled": True, |
| 68 | "human_escalation": list(escalation), |
| 69 | } |
| 70 | if regime_data.get("vcs", {}).get("regime", "").startswith("muse"): |
| 71 | seed_muse_substrate(repo_root) |
| 72 | cfg_path.write_text(yaml.safe_dump(data), encoding="utf-8") |
| 73 | docs = repo_root / "docs" |
| 74 | docs.mkdir(parents=True, exist_ok=True) |
| 75 | archive = docs / "archive" / "phases" |
| 76 | archive.mkdir(parents=True, exist_ok=True) |
| 77 | if write_freeze_artifact: |
| 78 | art = repo_root / AFF_FREEZE_REL |
| 79 | if not art.is_file(): |
| 80 | art.write_text( |
| 81 | "# AFF freeze\n\n" |
| 82 | "```yaml\n" |
| 83 | "phase: AFF\n" |
| 84 | "outputs:\n" |
| 85 | " - id: a\n" |
| 86 | " path: docs/a.md\n" |
| 87 | " frozen: true\n" |
| 88 | "```\n\n" |
| 89 | "Ground truth frozen: true. seven-tier matrix. file+line citation.\n", |
| 90 | encoding="utf-8", |
| 91 | ) |
| 92 | return load_config(cfg_path) |
| 93 | |
| 94 | |
| 95 | def copy_aff_entries(repo_root: Path) -> None: |
| 96 | """Copy entry fixtures into repo for CLI append tests.""" |
| 97 | dest = repo_root / "entries" |
| 98 | if dest.exists(): |
| 99 | shutil.rmtree(dest) |
| 100 | shutil.copytree(AFF_ENTRIES, dest) |
| 101 | |
| 102 | |
| 103 | def aff_body( |
| 104 | name: str = "aff-pass.json", |
| 105 | *, |
| 106 | artifact_digest: str | None = None, |
| 107 | **overrides, |
| 108 | ) -> dict: |
| 109 | """Load fixture and optionally override digest / fields.""" |
| 110 | body = load_aff_entry(name) |
| 111 | if artifact_digest is not None: |
| 112 | body["artifact_digest"] = artifact_digest |
| 113 | body.update(overrides) |
| 114 | return body |
| 115 | |
| 116 | |
| 117 | def write_mechanical_stamp(repo_root: Path, rel: str = AFF_FREEZE_REL) -> str: |
| 118 | """Stamp freeze artifact mechanically; return current artifact_digest.""" |
| 119 | from tools.freeze_reviewer.artifact import artifact_digest, parse_artifact |
| 120 | from tools.freeze_reviewer.stamp import build_stamp, write_stamp |
| 121 | from tools.freeze_reviewer.types import ReviewerSettings |
| 122 | |
| 123 | path = repo_root / rel |
| 124 | parsed = parse_artifact(path, rel_path=rel) |
| 125 | stamp = build_stamp( |
| 126 | parsed, |
| 127 | reviewer=ReviewerSettings("agent", "thinking-high", "local", "human"), |
| 128 | kit_version="0.1.0", |
| 129 | produced_by="checklist_engine", |
| 130 | provider_kind="rule_engine", |
| 131 | checklist_ids=["C1"], |
| 132 | checklist_source="builtin", |
| 133 | findings_count=0, |
| 134 | ) |
| 135 | write_stamp(path, parsed, stamp) |
| 136 | reparsed = parse_artifact(path, rel_path=rel) |
| 137 | return artifact_digest(reparsed) |
| 138 | |
| 139 | |
| 140 | def current_artifact_digest(repo_root: Path, rel: str = AFF_FREEZE_REL) -> str: |
| 141 | """Return FRV artifact_digest for the freeze path.""" |
| 142 | from tools.freeze_reviewer.artifact import artifact_digest, parse_artifact |
| 143 | |
| 144 | path = repo_root / rel |
| 145 | return artifact_digest(parse_artifact(path, rel_path=rel)) |