"""task-2553+34 TRACK C2 — independent adversarial regression for the policy profile engine. Two-phase design (task md §3, 9-R.2): * Phase-A (engine-independent, ALWAYS runs): validates that the five adversarial fixtures are well-formed and internally consistent. This makes the suite non-vacuous even while C1 (task-2553+33) is unsettled. * Phase-B (engine-consuming): if the C1 core `anu_v3.policy_profile_engine` is importable with a usable decision entrypoint, drive every fixture through it and assert the fail-closed contract. If the engine is absent, the Phase-B tests SKIP (not fail) and the suite records DEFERRED_PENDING_C1. C2 scope invariant: this file only READS C1 artifacts. It never imports for side effects, never writes engine/schema files, never mutates the profile registry. Absence of C1 ⇒ DEFERRED, never HOLD, never FAIL (task md §8). """ from __future__ import annotations import importlib import json import unittest from pathlib import Path # Resolve the live workspace root regardless of CWD. _THIS = Path(__file__).resolve() WS_ROOT = _THIS.parents[2] # tests/regression/ -> workspace root FIXTURE_DIR = WS_ROOT / "memory" / "fixtures" FIXTURES = { "profile_mismatch": "task-2553+34.profile-mismatch.json", "missing_profile": "task-2553+34.missing-profile.json", "stale_profile": "task-2553+34.stale-profile.json", "forbidden_boundary": "task-2553+34.forbidden-boundary.json", "allow_vs_forbid_conflict": "task-2553+34.allow-vs-forbid-conflict.json", } def _load(name: str) -> dict: with open(FIXTURE_DIR / FIXTURES[name], "r", encoding="utf-8") as fh: return json.load(fh) # --------------------------------------------------------------------------- # # C1 engine discovery (read-only, no side effects). # --------------------------------------------------------------------------- # def _discover_engine(): """Return (module, callable) for the C1 policy profile engine, or (None, None). The exact public entrypoint name is owned by C1 (task-2553+33). We probe a set of plausible callables so Phase-B activates the moment C1 settles without this C2 file having to write into C1 scope. """ try: mod = importlib.import_module("anu_v3.policy_profile_engine") except Exception: return None, None for attr in ("evaluate", "decide", "resolve", "run", "evaluate_goal_request"): fn = getattr(mod, attr, None) if callable(fn): return mod, fn cls = getattr(mod, "PolicyProfileEngine", None) if cls is not None: return mod, cls return mod, None _ENGINE_MOD, _ENGINE_ENTRY = _discover_engine() ENGINE_PRESENT = _ENGINE_MOD is not None and _ENGINE_ENTRY is not None DEFER_REASON = ( "DEFERRED_PENDING_C1: anu_v3.policy_profile_engine (task-2553+33 C1 core) " "not present / no usable entrypoint. Phase-B re-runs automatically once C1 settles. " "This is an explicit terminal deferral, NOT a HOLD and NOT a FAIL (task md §3/§8/9-R.2)." ) # --------------------------------------------------------------------------- # # Phase-A — engine-independent. ALWAYS runs. # --------------------------------------------------------------------------- # class PhaseAFixtureContractTests(unittest.TestCase): """The fixtures themselves must be well-formed adversarial contracts.""" def test_all_fixtures_present_and_parseable(self): for key, fname in FIXTURES.items(): p = FIXTURE_DIR / fname self.assertTrue(p.exists(), f"missing fixture file: {fname}") data = _load(key) self.assertEqual(data["task"], "task-2553+34") self.assertEqual(data["track"], "C2") def test_each_fixture_declares_fail_closed_expectation(self): for key in FIXTURES: data = _load(key) beh = data["expected_engine_behavior"] self.assertEqual( beh["gate"], "HOLD", f"{key}: adversarial fixture must expect fail-closed HOLD", ) self.assertEqual(beh["decision_status"], "HOLD") self.assertEqual( beh["allowed_actions_expansion"], [], f"{key}: no privilege expansion permitted under attack", ) self.assertTrue(beh["forbidden_enforced"]) self.assertTrue( len(beh["must_not"]) >= 1, f"{key}: must enumerate explicit false-positive guards", ) def test_each_fixture_has_machine_checkable_phase_b_assertions(self): for key in FIXTURES: data = _load(key) self.assertIn("phase_b_assertions", data) self.assertTrue( len(data["phase_b_assertions"]) >= 2, f"{key}: needs >=2 Phase-B predicates", ) for a in data["phase_b_assertions"]: self.assertIn("predicate", a) self.assertIn("rationale", a) def test_required_adversarial_classes_are_covered(self): classes = {_load(k)["adversarial_class"] for k in FIXTURES} required = { "profile_identity_mismatch", "missing_profile_fail_open", "stale_profile_silent_consume", "forbidden_boundary_breach", "allow_vs_forbid_precedence_conflict", } self.assertEqual( classes, required, "task md §2 mandates exactly these five adversarial classes", ) def test_conflict_fixture_pins_forbid_precedence(self): data = _load("allow_vs_forbid_conflict") body = data["profile_registry_state"]["resolved_profile_body"] self.assertIn("merge_pr", body["allowed_actions"]) self.assertIn("merge_pr", body["forbidden_actions"]) self.assertEqual( data["expected_engine_behavior"]["precedence_rule_applied"], "FORBID_DOMINATES_ALLOW", ) # --------------------------------------------------------------------------- # # Phase-B — engine-consuming. SKIPS (DEFERRED) when C1 absent. # --------------------------------------------------------------------------- # @unittest.skipUnless(ENGINE_PRESENT, DEFER_REASON) class PhaseBEngineAdversarialTests(unittest.TestCase): """Drive each adversarial fixture through the real C1 engine. These assertions are intentionally strict: a true positive requires the SPECIFIC hold reason, not merely any HOLD, so a precedence/coverage bug cannot hide behind an unrelated failure. """ def _decide(self, fixture: dict) -> dict: gr = fixture["input"]["goal_request"] ctx = { "goal_request": gr, "profile_registry_state": fixture.get("profile_registry_state"), "planned_effective_actions": fixture["input"].get("planned_effective_actions"), "effective_diff_files": fixture["input"].get("effective_diff_files"), } entry = _ENGINE_ENTRY if entry is None: # pragma: no cover - guarded by skipUnless self.skipTest(DEFER_REASON) if isinstance(entry, type): # PolicyProfileEngine class inst = entry() for m in ("evaluate", "decide", "resolve", "run"): fn = getattr(inst, m, None) if callable(fn): return _normalize(fn(ctx)) self.skipTest("engine class present but no decision method (DEFERRED-shaped)") return _normalize(entry(ctx)) def _assert_no_residual_permission(self, d: dict, token: str | None = None): """Codex MED resolution: a fail-closed HOLD must not leak permission through ANY resolved permission surface, not just `allowed_actions`.""" for surface in ("allowed_actions", "effective_allowed_actions", "resolved_allowed_actions", "expanded_actions"): if token is None: self.assertEqual( d.get(surface, []), [], f"fail-closed HOLD leaked permissions via '{surface}': {d.get(surface)}", ) else: self.assertNotIn( token, d.get(surface, []), f"forbidden token '{token}' leaked via '{surface}'", ) def test_profile_mismatch_holds_fail_closed(self): f = _load("profile_mismatch") d = self._decide(f) self.assertEqual(d["status"], "HOLD") self.assertIn("profile_identity_mismatch", d.get("hold_reasons", [])) self._assert_no_residual_permission(d) def test_missing_profile_fails_closed_no_side_effect(self): f = _load("missing_profile") d = self._decide(f) self.assertEqual(d["status"], "HOLD") self.assertIn("missing_profile", d.get("hold_reasons", [])) self._assert_no_residual_permission(d) # Codex HIGH resolution: fail-closed must not plant a persistent # escalation path. The engine must not have created/cached a profile # entry as a side effect of a missing-profile lookup. self.assertFalse( d.get("registry_mutated", False), "missing-profile HOLD must not mutate/create a registry entry", ) self.assertFalse( d.get("profile_created", False), "missing-profile HOLD must not auto-create a profile", ) created = d.get("created_profiles") or d.get("registry_writes") or [] self.assertEqual( list(created), [], "missing-profile HOLD must leave the profile registry unchanged", ) def test_stale_profile_holds_with_refresh_hint(self): f = _load("stale_profile") d = self._decide(f) self.assertEqual(d["status"], "HOLD") self.assertIn("stale_profile", d.get("hold_reasons", [])) self._assert_no_residual_permission(d) def test_forbidden_boundary_breach_all_violations_reported(self): f = _load("forbidden_boundary") d = self._decide(f) self.assertEqual(d["status"], "HOLD") self.assertIn("forbidden_boundary_breach", d.get("hold_reasons", [])) violations = " ".join(map(str, d.get("boundary_violations", []))) for forbidden_file in f["input"]["effective_diff_files"]: self.assertIn( forbidden_file, violations, f"forbidden file not surfaced as a violation: {forbidden_file}", ) def test_allow_vs_forbid_conflict_forbid_dominates(self): f = _load("allow_vs_forbid_conflict") d = self._decide(f) self.assertEqual(d["status"], "HOLD") # Codex CRITICAL resolution: prove FORBID>ALLOW *on the action axis*, # independently of the path breach that the same fixture also carries. # 1. The conflict itself must be a recorded hold cause (not only the # forbidden-path breach), so an ALLOW-biased engine that merely # HOLDs on the path cannot pass this test. self.assertIn( "allow_vs_forbid_conflict", d.get("hold_reasons", []), "the action-axis ALLOW/FORBID collision must be an explicit hold " "cause, not masked by the co-located forbidden-path breach", ) # 2. Deterministic precedence marker must be present and fail-closed — # not order/luck dependent. self.assertEqual( d.get("precedence_rule"), "FORBID_DOMINATES_ALLOW", "precedence must be an explicit deterministic fail-closed rule", ) # 3. The colliding token must not survive in ANY resolved permission # surface — closes the hidden privilege-escalation primitive. self._assert_no_residual_permission(d, token="merge_pr") def _normalize(decision) -> dict: """Coerce a variety of plausible engine return shapes into a dict view. C1 owns the exact decision schema; we accept dicts, objects with a `to_dict`/`as_dict`, or attribute-style decision objects. """ if isinstance(decision, dict): return decision for m in ("to_dict", "as_dict", "asdict"): fn = getattr(decision, m, None) if callable(fn): res = fn() return res if isinstance(res, dict) else dict(res) # type: ignore[call-overload] out: dict = {} for k in ( "status", "hold_reasons", "allowed_actions", "effective_allowed_actions", "boundary_violations", ): if hasattr(decision, k): out[k] = getattr(decision, k) return out if __name__ == "__main__": if not ENGINE_PRESENT: print(DEFER_REASON) print(f"engine_present={ENGINE_PRESENT} (Phase-A still runs; Phase-B deferred)") unittest.main(verbosity=2)