#!/usr/bin/env python3 """Focused bridge integration regressions for the Assurance Kernel upgrade.""" from __future__ import annotations import json import os import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch HARNESS_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(HARNESS_ROOT / "scripts" / "python")) import agentic_bridge as bridge class BridgeUpgradeIntegrationTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.state = str(Path(self.temp.name) / "state") self.project = str(HARNESS_ROOT.parent.parent) self.base_env = { "CASAN_STATE_ROOT": self.state, "CASAN_AGENTIC_ENFORCEMENT_MODE": "enforce", "CASAN_AGENT": "boss", "CASAN_ACTOR": "boss", "CASAN_PROFILE": "test", } def tearDown(self) -> None: self.temp.cleanup() def begin(self, actor: str = "boss") -> dict[str, object]: return bridge.op_begin({ "op": "begin", "client": "codex", "project": self.project, "session": "integration", "prompt": "edit a source file safely", "integration_mode": "project_hook", "actor": actor, }) def fixture_gate(self, body: str) -> str: path = Path(self.temp.name) / "h2-gate.sh" path.write_text("#!/usr/bin/env bash\n" + body + "\n", encoding="utf-8") path.chmod(0o700) return str(path) def pre_edit(self, admission_id: str) -> dict[str, object]: return bridge.op_pre_tool({ "op": "pre-tool", "admission_id": admission_id, "tool": "Edit", "tool_input": "update a source file", "project": self.project, }) def test_missing_h2_gate_denies_enforce_with_structured_evidence(self) -> None: missing = str(Path(self.temp.name) / "missing.sh") with patch.dict(os.environ, {**self.base_env, "CASAN_H2_GATE_PATH": missing}, clear=False): begin = self.begin() result = self.pre_edit(str(begin["admission_id"])) record = bridge.load_admission(str(begin["admission_id"])) self.assertEqual(result["decision"], "deny") self.assertEqual(result["reason"], "h2_gate_unavailable") evidence = next(item for item in record["evidence"] if item["kind"] == "tool-registry") self.assertEqual(evidence["facts"]["mode"], "enforce") self.assertEqual(evidence["facts"]["actor"], "boss") self.assertEqual(evidence["facts"]["tool"], "Edit") self.assertEqual(evidence["facts"]["execution_id"], begin["trace_id"]) self.assertEqual(evidence["facts"]["enforcement_path"], "agentic_bridge.pre_tool.h2_registry") def test_missing_h2_gate_observe_allows_only_degraded_non_certifiable_execution(self) -> None: missing = str(Path(self.temp.name) / "missing.sh") env = {**self.base_env, "CASAN_AGENTIC_ENFORCEMENT_MODE": "observe", "CASAN_H2_GATE_PATH": missing} with patch.dict(os.environ, env, clear=False): begin = self.begin() result = self.pre_edit(str(begin["admission_id"])) finalized = bridge.op_finalize({"op": "finalize", "admission_id": begin["admission_id"], "stop_reason": "completed"}) self.assertEqual(result["decision"], "allow") self.assertEqual(finalized["decision"], "non_certified") self.assertEqual(finalized["certification_strength"], "observed_only") def test_registry_defaults_on_and_development_bypass_is_visible(self) -> None: missing = str(Path(self.temp.name) / "missing.sh") with patch.dict(os.environ, {**self.base_env, "CASAN_H2_GATE_PATH": missing}, clear=False): begin = self.begin() default_result = self.pre_edit(str(begin["admission_id"])) self.assertEqual(default_result["reason"], "h2_gate_unavailable") with patch.dict(os.environ, {**self.base_env, "CASAN_AGENTIC_H2_REGISTRY": "0", "CASAN_H2_GATE_PATH": missing}, clear=False): begin = self.begin() bypass = self.pre_edit(str(begin["admission_id"])) record = bridge.load_admission(str(begin["admission_id"])) self.assertEqual(bypass["decision"], "allow") self.assertTrue(any("HIGH" in warning for warning in bypass["warnings"])) self.assertEqual(record["certification_strength"], "observed_only") def test_benign_deploy_requires_approval_from_action_floor(self) -> None: gate = self.fixture_gate("echo 'TOOL_APPROVED tool=deploy reason=registered'") with patch.dict(os.environ, {**self.base_env, "CASAN_H2_GATE_PATH": gate}, clear=False): begin = self.begin() result = bridge.op_pre_tool({ "op": "pre-tool", "admission_id": begin["admission_id"], "tool": "Bash", "tool_input": "deploy harmless documentation", "project": self.project, }) self.assertEqual(result["decision"], "require_approval") risk = result["policy_decision"] self.assertEqual(risk["action_class"], "deployment") self.assertEqual(risk["risk_factors"]["content_risk"], "medium") self.assertEqual(risk["effective_risk"], "high") def test_structured_shell_payload_classifies_the_inner_read_only_command(self) -> None: with patch.dict(os.environ, self.base_env, clear=False): begin = self.begin() result = bridge.op_pre_tool({ "op": "pre-tool", "admission_id": begin["admission_id"], "tool": "Bash", "tool_input": {"command": "ls"}, "project": self.project, }) self.assertEqual(result["decision"], "allow") self.assertEqual(result["policy_decision"]["action_class"], "read_only") def test_failed_side_effect_halts_and_cannot_finalize_successfully(self) -> None: gate = self.fixture_gate("echo 'TOOL_APPROVED tool=write_file reason=registered'") with patch.dict(os.environ, {**self.base_env, "CASAN_H2_GATE_PATH": gate}, clear=False): begin = self.begin() admitted = self.pre_edit(str(begin["admission_id"])) post = bridge.op_post_tool({ "op": "post-tool", "admission_id": begin["admission_id"], "tool": "Edit", "status": "error", "duration_ms": 2, "result": "write failed", }) finalized = bridge.op_finalize({ "op": "finalize", "admission_id": begin["admission_id"], "stop_reason": "completed", "assistant_summary": "completed", }) self.assertEqual(admitted["decision"], "allow") self.assertEqual(post["decision"], "halt") self.assertFalse(post["assurance_may_continue"]) self.assertEqual(finalized["decision"], "non_certified") self.assertEqual(finalized["execution_outcome"], "failed") self.assertEqual(finalized["assurance_outcome"], "failed") self.assertIn("failed_tool_outcome", finalized["reason"]) kernel_files = list((Path(self.state) / "logs" / "kernel").glob("*.json")) self.assertEqual(len(kernel_files), 1) kernel = json.loads(kernel_files[0].read_text(encoding="utf-8")) self.assertEqual(kernel["run"]["outcome"]["execution_result"], "failed") self.assertEqual(kernel["run"]["outcome"]["certification_result"], "non_certified") def test_production_without_external_trust_root_never_certifies(self) -> None: env = {**self.base_env, "CASAN_PROFILE": "production", "CASAN_AGENTIC_ENFORCEMENT_MODE": "enforce"} with patch.dict(os.environ, env, clear=False): begin = self.begin() finalized = bridge.op_finalize({"op": "finalize", "admission_id": begin["admission_id"], "stop_reason": "completed"}) self.assertEqual(finalized["decision"], "non_certified") self.assertIn("external_signing_trust_root_required", finalized["reason"]) if __name__ == "__main__": unittest.main(verbosity=2)