"""G8 release agent fail-closed orchestration tests.""" from __future__ import annotations import importlib.util import json import sys import tarfile import tempfile import unittest from dataclasses import dataclass, field from pathlib import Path from typing import Any SCRIPT_PATH = Path(__file__).with_name("run-outcome-os-release-agent.py") def load_release_agent(): spec = importlib.util.spec_from_file_location("outcome_os_release_agent", SCRIPT_PATH) if spec is None or spec.loader is None: raise RuntimeError("release agent could not be loaded") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module @dataclass class FakeRunner: module: Any fail_stage: str | None = None builder_shas: tuple[str, str] = ("a" * 64, "a" * 64) tracked_status: str = "" clean_archive_payloads: tuple[bytes, bytes] = (b"clean archive", b"clean archive") clean_head: str = "b" * 40 clean_tree: str = "c" * 40 calls: list[str] = field(default_factory=list) commands: dict[str, list[str]] = field(default_factory=dict) working_directories: dict[str, Path | None] = field(default_factory=dict) environments: dict[str, dict[str, str] | None] = field(default_factory=dict) timeouts: dict[str, int | None] = field(default_factory=dict) event_log: list[str] | None = None def run(self, stage: str, argv: list[str], **kwargs: Any): self.calls.append(stage) self.commands[stage] = list(argv) self.working_directories[stage] = kwargs.get("cwd") self.environments[stage] = kwargs.get("env") self.timeouts[stage] = kwargs.get("timeout") if self.event_log is not None: self.event_log.append(f"runner:{stage}") if stage == self.fail_stage: raise self.module.StageFailure(stage, "synthetic failure") if stage.startswith("release_patch_run_"): run_index = int(stage.rsplit("_", 1)[1]) - 1 payload = { "ok": True, "base_commit": "b" * 40, "output_patch": "docs/ops/evidence/release.patch", "patch_sha256": self.builder_shas[run_index], "patch_bytes": 123, "patch_files": 9, "git_apply_check": "passed-on-clean-temporary-head", "git_apply_cached_check": "passed-on-canonical-clean-index", "patch_line_endings": "lf-only", "source_worktree": "unchanged-except-output-patch", "api_generation": {"sha256": "c" * 64}, } return self.module.CommandResult(0, json.dumps(payload), "") if stage == "release_manifest": return self.module.CommandResult(0, json.dumps({"ok": True}), "") if stage == "clean_head_repository": return self.module.CommandResult(0, str(kwargs["cwd"].resolve()) + "\n", "") if stage.startswith("clean_head_worktree_"): return self.module.CommandResult(0, self.tracked_status, "") if stage.startswith("clean_head_head_"): return self.module.CommandResult(0, self.clean_head + "\n", "") if stage.startswith("clean_head_tree_"): return self.module.CommandResult(0, self.clean_tree + "\n", "") if stage.startswith("clean_head_archive_run_"): run_index = int(stage.rsplit("_", 1)[1]) - 1 output_index = argv.index("--output") Path(argv[output_index + 1]).write_bytes(self.clean_archive_payloads[run_index]) return self.module.CommandResult(0, "", "") return self.module.CommandResult(0, f"{stage}: passed", "") @dataclass class FakeDeployment: module: Any active_sha: str | None rollback_matches: bool = True tracked_api_image: str = "sha256:" + "1" * 64 calls: list[str] = field(default_factory=list) event_log: list[str] | None = None def _record(self, event: str) -> None: self.calls.append(event) if self.event_log is not None: self.event_log.append(f"deployment:{event}") def read_active_state(self): self._record("read_active_state") if self.active_sha is None: return None return { "schema": "vignette.release-agent-active.v1", "active_sha": self.active_sha, "compose_project": "vignette-preview-20260807", "remote_root": "/volume1/docker/vignette-preview-20260807", "compose_file": "/volume1/docker/vignette-preview-20260807/infra/docker-compose.yml", "env_file": "/volume1/docker/vignette-preview-20260807/infra/.env", "api_image": self.tracked_api_image, "web_image": "sha256:" + "2" * 64, } def snapshot(self): self._record("snapshot") return self.module.DeploymentSnapshot( api_image="sha256:" + "1" * 64, web_image="sha256:" + "2" * 64, compose_file="/volume1/docker/vignette-preview-20260807/infra/docker-compose.yml", env_file="/volume1/docker/vignette-preview-20260807/infra/.env", active_sha=self.active_sha, ) def promote(self, candidate: Path, desired_sha: str, snapshot: Any): self._record("promote") return { "active_sha": desired_sha, "api_image": "sha256:" + "3" * 64, "web_image": "sha256:" + "4" * 64, "candidate": str(candidate), "previous_api_image": snapshot.api_image, } def commit_active_state(self, state: dict[str, Any]): self._record("commit_active_state") self.active_sha = state["active_sha"] def rollback(self, snapshot: Any): self._record("rollback") self.active_sha = snapshot.active_sha return {"api_image": snapshot.api_image, "web_image": snapshot.web_image} def verify_rollback(self, snapshot: Any): self._record("verify_rollback") return { "ok": self.rollback_matches, "api_image": snapshot.api_image if self.rollback_matches else "sha256:" + "9" * 64, "web_image": snapshot.web_image, } @dataclass class FakeRuntimeProbe: module: Any fail_on_call: int | None = None calls: int = 0 def verify(self): self.calls += 1 if self.calls == self.fail_on_call: raise self.module.StageFailure("runtime_contract", "synthetic unhealthy runtime") return { "health": {"status": "ok", "db": True, "engine": True}, "auth_status": 401, "openapi_paths": 122, "goals": [f"G{number}" for number in range(9)], } @dataclass class TransientRuntimeProbe: module: Any transient_failures: int calls: int = 0 def verify(self): self.calls += 1 if self.calls <= self.transient_failures: raise self.module.StageFailure( "runtime_contract", "/api/health status=502, expected=200", ) return { "health": {"status": "ok", "db": True, "engine": True}, "auth_status": 401, "openapi_paths": 122, "goals": [f"G{number}" for number in range(9)], } class FakeCandidate: def __init__(self, root: Path): self.root = root self.cleaned = False self.materialization_mode: str | None = None self.archive_sha256: str | None = None self.archive_bytes: bytes | None = None def materialize(self, base_commit: str, patch_path: str): del base_commit, patch_path self.materialization_mode = "patch" self.root.mkdir(parents=True, exist_ok=True) return self.root def materialize_archive(self, archive_path: Path, expected_sha256: str): self.materialization_mode = "clean-head" self.archive_sha256 = expected_sha256 self.archive_bytes = archive_path.read_bytes() self.root.mkdir(parents=True, exist_ok=True) return self.root def cleanup(self): self.cleaned = True @dataclass class CaptureRunner: module: Any calls: list[tuple[str, list[str]]] = field(default_factory=list) def run(self, stage: str, argv: list[str], **_: Any): self.calls.append((stage, argv)) if stage == "nas_promote": return self.module.CommandResult( 0, "sha256:" + "3" * 64 + "\n" + "sha256:" + "4" * 64 + "\n", "", ) return self.module.CommandResult(0, "", "") @dataclass class ActiveStateRunner: module: Any payload: dict[str, Any] calls: list[str] = field(default_factory=list) def run(self, stage: str, argv: list[str], **_: Any): del argv self.calls.append(stage) return self.module.CommandResult(0, json.dumps(self.payload), "") @dataclass class CandidateArchiveRunner: module: Any calls: list[tuple[str, list[str]]] = field(default_factory=list) def run(self, stage: str, argv: list[str], **_: Any): self.calls.append((stage, argv)) if stage == "candidate_archive": output_index = argv.index("--output") archive_path = Path(argv[output_index + 1]) with tarfile.open(archive_path, "w") as archive: marker = archive_path.parent / "README.md" marker.write_bytes(b"candidate\r\n") archive.add(marker, arcname="README.md") generated = archive_path.parent / "api.gen.ts" generated.write_bytes(b"export type A = 1;\r\nexport type B = 2;\r\n") archive.add(generated, arcname="apps/web/src/lib/api.gen.ts") shell_script = archive_path.parent / "99_app_role.sh" shell_script.write_bytes(b"#!/usr/bin/env bash\r\necho ready\r\n") archive.add(shell_script, arcname="infra/db/init/99_app_role.sh") return self.module.CommandResult(0, "", "") class OutcomeReleaseAgentTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.agent_module = load_release_agent() def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) manifest = { "classifications": { "related": [ { "disposition": "candidate-whole-file", "paths": list(self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS), } ] }, "release_assembly": { "patch_sha256": "a" * 64, "output_patch": "docs/ops/evidence/release.patch", } } manifest_path = self.root / "docs/ops/manifest.json" manifest_path.parent.mkdir(parents=True) manifest_path.write_text(json.dumps(manifest), encoding="utf-8") self.manifest_path = manifest_path self.milestone = self.agent_module.MaterialMilestone( milestone_id="g8-release-agent-close-loop", material=True, goals=("G8",), evidence_refs=("tests:g8-release-agent",), ) def tearDown(self) -> None: self.temp.cleanup() def make_agent( self, *, active_sha: str | None, execute: bool, fail_stage: str | None = None, builder_shas: tuple[str, str] = ("a" * 64, "a" * 64), source_mode: str = "patch", tracked_status: str = "", clean_archive_payloads: tuple[bytes, bytes] = ( b"clean archive", b"clean archive", ), source_repo_root: Path | None = None, runtime_fail_on_call: int | None = None, rollback_matches: bool = True, ): event_log: list[str] = [] runner = FakeRunner( module=self.agent_module, fail_stage=fail_stage, builder_shas=builder_shas, tracked_status=tracked_status, clean_archive_payloads=clean_archive_payloads, event_log=event_log, ) deployment = FakeDeployment( self.agent_module, active_sha, rollback_matches, event_log=event_log, ) runtime = FakeRuntimeProbe(self.agent_module, runtime_fail_on_call) candidate = FakeCandidate(self.root / "candidate") config = self.agent_module.ReleaseAgentConfig( repo_root=source_repo_root or self.root, manifest_path=self.manifest_path, hunk_map_path=self.root / "docs/ops/hunk-map.json", source_mode=source_mode, execute=execute, milestone=self.milestone, target=self.agent_module.NAS_PREVIEW_TARGET, evidence_out=self.root / "evidence.json", nas_env_file=self.root / "infra/.env.nas-preview", ) agent = self.agent_module.ReleaseAgent( config, runner=runner, deployment=deployment, runtime_probe=runtime, candidate_manager=candidate, ) return agent, runner, deployment, runtime, candidate def test_rejects_any_target_outside_the_isolated_preview(self) -> None: wrong = self.agent_module.DeploymentTarget( base_url="http://100.116.83.60:8080", compose_project="vignette", remote_root="/volume1/docker/vignette", ) with self.assertRaisesRegex(ValueError, "isolated NAS preview"): self.agent_module.validate_target(wrong) def test_mobile_shell_and_late_migrations_are_required_release_payloads(self) -> None: manifest = { "classifications": { "related": [ { "disposition": "candidate-whole-file", "paths": [ path for path in self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS if path != "apps/web/src/components/shell/shell.css" ], } ] } } with self.assertRaisesRegex( self.agent_module.StageFailure, "apps/web/src/components/shell/shell.css", ): self.agent_module.validate_required_release_payload(manifest) def test_admin_usage_index_migration_is_idempotent_and_release_required(self) -> None: migration_path = ( SCRIPT_PATH.parent.parent / "infra" / "db" / "init" / "18_admin_usage_ledger_index.sql" ) migration = migration_path.read_text(encoding="utf-8") self.assertNotIn("BEGIN;", migration) self.assertIn( "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_turns_admin_usage_created_at", migration, ) self.assertIn("ON app.turns (created_at DESC)", migration) self.assertIn("WHERE speaker = 'client';", migration) self.assertNotIn("COMMIT;", migration) self.assertIn( "18_admin_usage_ledger_index.sql", self.agent_module.RELEASE_DB_MIGRATIONS, ) self.assertIn( "18_admin_usage_ledger_index.sql", self.agent_module.RELEASE_DB_ONLINE_MIGRATIONS, ) self.assertIn( "infra/db/init/18_admin_usage_ledger_index.sql", self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS, ) def test_single_active_session_index_is_online_and_release_required(self) -> None: migration_path = ( SCRIPT_PATH.parent.parent / "infra" / "db" / "init" / "21_single_active_session.sql" ) migration = migration_path.read_text(encoding="utf-8") self.assertNotIn("BEGIN;", migration) self.assertIn( "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS uq_sessions_one_active_learner_persona", migration, ) self.assertIn("ON app.sessions (learner_id, persona_id)", migration) self.assertIn("WHERE ended_at IS NULL AND persona_id IS NOT NULL;", migration) self.assertIn( "duplicate active learner-persona sessions exist", migration, ) self.assertIn( "target index exists but is invalid or has a different definition", migration, ) self.assertIn("index_meta.indisvalid", migration) self.assertIn("index_meta.indisready", migration) self.assertIn("valid target unique index was not created", migration) self.assertNotIn("COMMIT;", migration) self.assertIn( "21_single_active_session.sql", self.agent_module.RELEASE_DB_MIGRATIONS, ) self.assertIn( "21_single_active_session.sql", self.agent_module.RELEASE_DB_ONLINE_MIGRATIONS, ) self.assertIn( "infra/db/init/21_single_active_session.sql", self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS, ) self.assertIn( "20_public_bootstrap_ticket_events.sql", self.agent_module.RELEASE_DB_MIGRATIONS, ) self.assertNotIn( "20_public_bootstrap_ticket_events.sql", self.agent_module.RELEASE_DB_ONLINE_MIGRATIONS, ) self.assertIn( "infra/db/init/20_public_bootstrap_ticket_events.sql", self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS, ) def test_case_profile_multi_case_migration_is_online_safe_and_release_required( self, ) -> None: migration_path = ( SCRIPT_PATH.parent.parent / "infra" / "db" / "init" / "22_case_profile_multi_case.sql" ) migration = migration_path.read_text(encoding="utf-8") self.assertNotIn("BEGIN;", migration) self.assertNotIn("COMMIT;", migration) self.assertIn("legacy_constraints text[];", migration) self.assertIn("constraint_meta.contype = 'u'", migration) self.assertIn( "ARRAY['persona_id', 'learner_id']::text[]", migration, ) self.assertIn( "multiple legacy case_profile persona-learner unique constraints exist", migration, ) self.assertIn( "ALTER TABLE app.case_profile DROP CONSTRAINT %I", migration, ) self.assertIn( "legacy case_profile persona-learner unique constraint remains", migration, ) self.assertIn( "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_case_profile_learner_persona_activity", migration, ) self.assertIn( "ON app.case_profile (learner_id, persona_id, updated_at DESC, case_id);", migration, ) self.assertIn("index_meta.indisvalid", migration) self.assertIn("index_meta.indisready", migration) self.assertIn("NOT index_meta.indisunique", migration) self.assertIn("index_meta.indnkeyatts = 4", migration) self.assertIn( "target index exists but is invalid or has a different definition", migration, ) self.assertIn("valid case activity index was not created", migration) self.assertIn( "22_case_profile_multi_case.sql", self.agent_module.RELEASE_DB_MIGRATIONS, ) self.assertIn( "22_case_profile_multi_case.sql", self.agent_module.RELEASE_DB_ONLINE_MIGRATIONS, ) self.assertIn( "infra/db/init/22_case_profile_multi_case.sql", self.agent_module.REQUIRED_RELEASE_PAYLOAD_PATHS, ) def test_two_patch_runs_must_be_byte_identical_before_any_deploy(self) -> None: agent, runner, deployment, _, _ = self.make_agent( active_sha=None, execute=True, builder_shas=("a" * 64, "d" * 64), ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("release_determinism", raised.exception.report["failed_stage"]) self.assertNotIn("snapshot", deployment.calls) self.assertNotIn("promote", deployment.calls) self.assertEqual( ["release_patch_run_1", "release_patch_run_2"], runner.calls, ) def test_parser_keeps_patch_and_controller_repo_defaults(self) -> None: parser = self.agent_module.build_parser() default_args = parser.parse_args(["--milestone", "milestone.json"]) detached_root = self.root / "detached" clean_args = parser.parse_args( [ "--milestone", "milestone.json", "--source-mode", "clean-head", "--source-repo-root", str(detached_root), ] ) self.assertEqual("patch", default_args.source_mode) self.assertEqual(self.agent_module.REPO_ROOT, default_args.source_repo_root) self.assertEqual("clean-head", clean_args.source_mode) self.assertEqual(detached_root, clean_args.source_repo_root) def test_source_repo_root_must_be_an_absolute_existing_directory(self) -> None: relative = Path("detached-source") missing = self.root / "missing" source_file = self.root / "source-file" source_file.write_text("not a directory\n", encoding="utf-8") for source_root, message in ( (relative, "must be absolute"), (missing, "does not exist"), (source_file, "not a directory"), ): with self.subTest(source_root=source_root): with self.assertRaisesRegex(self.agent_module.StageFailure, message): self.agent_module.validate_source_repo_root(source_root) def test_clean_head_override_runs_every_git_command_in_source_repo(self) -> None: detached_root = self.root / "detached-source" detached_root.mkdir() archive_payload = b"detached clean-head archive" agent, runner, _, _, candidate = self.make_agent( active_sha="e" * 64, execute=True, source_mode="clean-head", clean_archive_payloads=(archive_payload, archive_payload), source_repo_root=detached_root, ) report = agent.run() git_stages = [ stage for stage in runner.calls if stage == "clean_head_repository" or stage.startswith("clean_head_worktree_") or stage.startswith("clean_head_head_") or stage.startswith("clean_head_tree_") or stage.startswith("clean_head_archive_run_") ] self.assertEqual(9, len(git_stages)) self.assertEqual( {detached_root}, {runner.working_directories[stage] for stage in git_stages}, ) self.assertEqual(str(detached_root), report["source"]["repo_root"]) self.assertEqual("clean-head", candidate.materialization_mode) def test_clean_head_rejects_tracked_changes_and_ignores_untracked_in_query(self) -> None: agent, runner, deployment, _, candidate = self.make_agent( active_sha="e" * 64, execute=True, source_mode="clean-head", tracked_status=" M docs/TODO.md\n", ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("clean_head_worktree", raised.exception.report["failed_stage"]) self.assertEqual( ["clean_head_repository", "clean_head_worktree_before"], runner.calls, ) self.assertIn( "--untracked-files=no", runner.commands["clean_head_worktree_before"], ) self.assertEqual([], deployment.calls) self.assertIsNone(candidate.materialization_mode) def test_clean_head_non_git_source_fails_before_identity_or_deployment(self) -> None: agent, runner, deployment, _, candidate = self.make_agent( active_sha="e" * 64, execute=True, source_mode="clean-head", fail_stage="clean_head_repository", ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("clean_head_repository", raised.exception.report["failed_stage"]) self.assertEqual(["clean_head_repository"], runner.calls) self.assertEqual([], deployment.calls) self.assertIsNone(candidate.materialization_mode) def test_clean_head_archive_runs_must_be_byte_identical(self) -> None: agent, runner, deployment, _, candidate = self.make_agent( active_sha="e" * 64, execute=True, source_mode="clean-head", clean_archive_payloads=(b"first archive", b"second archive"), ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("clean_head_determinism", raised.exception.report["failed_stage"]) self.assertEqual(2, len([stage for stage in runner.calls if "archive_run" in stage])) self.assertEqual([], deployment.calls) self.assertIsNone(candidate.materialization_mode) def test_clean_head_evidence_binds_exact_head_tree_and_archive(self) -> None: archive_payload = b"deterministic clean-head archive" expected_sha = self.agent_module.sha256_bytes(archive_payload) agent, runner, deployment, _, candidate = self.make_agent( active_sha="e" * 64, execute=True, source_mode="clean-head", clean_archive_payloads=(archive_payload, archive_payload), ) report = agent.run() self.assertTrue(report["ok"]) self.assertEqual(expected_sha, report["desired_sha"]) self.assertEqual("clean-head", report["source_mode"]) self.assertEqual("b" * 40, report["source"]["head"]) self.assertEqual("c" * 40, report["source"]["tree"]) self.assertEqual(expected_sha, report["source"]["archive"]["sha256"]) self.assertEqual(len(archive_payload), report["source"]["archive"]["bytes"]) self.assertEqual(2, report["source"]["archive"]["deterministic_runs"]) self.assertEqual("clean-head", candidate.materialization_mode) self.assertEqual(expected_sha, candidate.archive_sha256) self.assertEqual(archive_payload, candidate.archive_bytes) self.assertNotIn("release_manifest", runner.calls) self.assertFalse(any(stage.startswith("release_patch_run_") for stage in runner.calls)) self.assertIn("promote", deployment.calls) evidence = json.loads((self.root / "evidence.json").read_text(encoding="utf-8")) self.assertEqual(report["source"], evidence["source"]) self.assertEqual("clean-head", evidence["source_mode"]) def test_same_deployment_sha_is_a_runtime_checked_noop(self) -> None: agent, runner, deployment, runtime, candidate = self.make_agent( active_sha="a" * 64, execute=True, ) report = agent.run() self.assertTrue(report["ok"]) self.assertEqual("noop_same_sha", report["status"]) self.assertEqual(1, runtime.calls) self.assertIn("snapshot", deployment.calls) self.assertNotIn("promote", deployment.calls) self.assertNotIn("api_tests", runner.calls) self.assertFalse(candidate.cleaned) def test_same_sha_runtime_image_drift_is_not_adopted_as_a_noop(self) -> None: agent, _, deployment, runtime, _ = self.make_agent( active_sha="a" * 64, execute=True, ) deployment.tracked_api_image = "sha256:" + "9" * 64 with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("deployment_state", raised.exception.report["failed_stage"]) self.assertEqual(0, runtime.calls) self.assertNotIn("promote", deployment.calls) def test_default_dry_run_reports_changed_sha_without_mutation(self) -> None: agent, runner, deployment, runtime, _ = self.make_agent( active_sha="e" * 64, execute=False, ) report = agent.run() self.assertTrue(report["ok"]) self.assertEqual("dry_run_changed_sha", report["status"]) self.assertEqual("a" * 64, report["desired_sha"]) self.assertEqual("e" * 64, report["active_sha"]) self.assertEqual(0, runtime.calls) self.assertNotIn("snapshot", deployment.calls) self.assertNotIn("promote", deployment.calls) self.assertNotIn("api_tests", runner.calls) def test_untracked_remote_state_blocks_execute_before_snapshot(self) -> None: agent, runner, deployment, _, _ = self.make_agent( active_sha=None, execute=True, ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("deployment_state", raised.exception.report["failed_stage"]) self.assertNotIn("snapshot", deployment.calls) self.assertNotIn("api_tests", runner.calls) def test_changed_sha_runs_all_gates_then_promotion_and_browser_review(self) -> None: agent, runner, deployment, runtime, candidate = self.make_agent( active_sha="e" * 64, execute=True, ) report = agent.run() self.assertTrue(report["ok"]) self.assertEqual("deployed_ssot_sync_required", report["status"]) self.assertEqual("patch", report["source_mode"]) self.assertEqual("a" * 64, report["source"]["patch"]["sha256"]) self.assertEqual("a" * 64, report["active_sha"]) self.assertEqual("e" * 64, report["previous_sha"]) self.assertEqual( [ "release_patch_run_1", "release_patch_run_2", "release_manifest", "nas_preflight", "api_tests", "web_install", "web_api_contract", "web_typecheck", "web_build", "source_insecure_context_e2e", "session_e2e", "postdeploy_browser_review", ], runner.calls, ) self.assertEqual( ["read_active_state", "snapshot", "promote", "commit_active_state"], deployment.calls, ) self.assertEqual(1, runtime.calls) self.assertTrue(candidate.cleaned) postdeploy_command = runner.commands["postdeploy_browser_review"] test_index = postdeploy_command.index("test") first_flag = next( index for index in range(test_index + 1, len(postdeploy_command)) if postdeploy_command[index].startswith("--") ) self.assertEqual( self.agent_module.POSTDEPLOY_NAS_E2E_SPECS, tuple(postdeploy_command[test_index + 1 : first_flag]), ) self.assertNotIn("--grep", postdeploy_command) self.assertEqual( {f"--project={project}" for project in self.agent_module.RELEASE_BROWSER_PROJECTS}, {value for value in postdeploy_command if value.startswith("--project=")}, ) self.assertEqual( self.agent_module.RELEASE_E2E_TIMEOUT_SECONDS, runner.timeouts["postdeploy_browser_review"], ) self.assertEqual( self.agent_module.NAS_PREVIEW_TARGET.base_url, runner.environments["postdeploy_browser_review"]["PLAYWRIGHT_BASE_URL"], ) self.assertEqual( "1", runner.environments["postdeploy_browser_review"]["PLAYWRIGHT_SKIP_WEB_SERVER"], ) self.assertIn( "e2e/self-directed-learning-loop.spec.ts", self.agent_module.POSTDEPLOY_NAS_E2E_SPECS, ) self.assertIn( "e2e/deliberate-practice.spec.ts", self.agent_module.POSTDEPLOY_NAS_E2E_SPECS, ) self.assertNotIn( "e2e/insecure-context-uuid.spec.ts", self.agent_module.POSTDEPLOY_NAS_E2E_SPECS, ) self.assertNotIn( "e2e/returned-practice-db-closed-loop.spec.ts", self.agent_module.POSTDEPLOY_NAS_E2E_SPECS, ) self.assertEqual( set(self.agent_module.RELEASE_UI_E2E_SPECS), set(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS) | set(self.agent_module.POSTDEPLOY_SOURCE_ONLY_E2E_SPECS), ) self.assertEqual( set(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS), set(self.agent_module.POSTDEPLOY_NAS_REAL_API_E2E_SPECS) | set(self.agent_module.POSTDEPLOY_NAS_ROUTE_FIXTURE_E2E_SPECS), ) self.assertFalse( set(self.agent_module.POSTDEPLOY_NAS_REAL_API_E2E_SPECS) & set(self.agent_module.POSTDEPLOY_NAS_ROUTE_FIXTURE_E2E_SPECS) ) self.assertLess( runner.event_log.index("runner:postdeploy_browser_review"), runner.event_log.index("deployment:commit_active_state"), ) self.assertTrue( set(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS).issubset( set(runner.commands["session_e2e"]) ) ) self.assertNotIn( "e2e/insecure-context-uuid.spec.ts", runner.commands["session_e2e"], ) source_command = runner.commands["source_insecure_context_e2e"] self.assertIn("e2e/insecure-context-uuid.spec.ts", source_command) self.assertIn("--project=chromium-desktop", source_command) self.assertIn("--project=chromium-mobile", source_command) source_environment = runner.environments["source_insecure_context_e2e"] self.assertNotIn("PLAYWRIGHT_BASE_URL", source_environment) self.assertNotIn("PLAYWRIGHT_SKIP_WEB_SERVER", source_environment) self.assertEqual("127.0.0.1", source_environment["PLAYWRIGHT_HOST"]) self.assertEqual( str(self.agent_module.SOURCE_ONLY_E2E_PORT), source_environment["PLAYWRIGHT_PORT"], ) self.assertEqual("1", source_environment["CI"]) self.assertEqual(15 * 60, self.agent_module.RELEASE_E2E_TIMEOUT_SECONDS) self.assertIn( "e2e/insecure-context-uuid.spec.ts", self.agent_module.RELEASE_UI_E2E_SPECS, ) self.assertEqual("patch", candidate.materialization_mode) self.assertEqual( ( "14_continuous_improvement.sql", "15_self_directed_practice_runtime.sql", "16_calibration_transfer_actual_execution.sql", "17_improvement_workbook_contracts.sql", "18_admin_usage_ledger_index.sql", "19_auth_identity_alias.sql", "20_public_bootstrap_ticket_events.sql", "21_single_active_session.sql", "22_case_profile_multi_case.sql", ), self.agent_module.RELEASE_DB_MIGRATIONS, ) evidence = json.loads((self.root / "evidence.json").read_text(encoding="utf-8")) self.assertEqual("a" * 64, evidence["active_sha"]) self.assertEqual("e" * 64, evidence["previous_sha"]) self.assertEqual("a" * 64, evidence["deployment"]["desired_sha"]) self.assertEqual("required", evidence["ssot_sync"]["status"]) self.assertEqual( list(self.agent_module.POSTDEPLOY_NAS_E2E_SPECS), evidence["browser_review_proof"]["specs"], ) self.assertEqual( ["e2e/insecure-context-uuid.spec.ts"], evidence["browser_review_proof"]["source_only_candidate_specs"], ) self.assertEqual( ["e2e/returned-practice-db-closed-loop.spec.ts"], evidence["browser_review_proof"]["separate_disposable_db_specs"], ) self.assertEqual( { "real_api_db_specs": [ "e2e/session-layout.spec.ts", "e2e/session-persistence.spec.ts", ], "route_fixture_specs": list( self.agent_module.POSTDEPLOY_NAS_ROUTE_FIXTURE_E2E_SPECS ), }, evidence["browser_review_proof"]["runtime_scope"], ) self.assertEqual( "not_run_by_release_agent", evidence["browser_review_proof"]["separate_disposable_db_status"], ) self.assertEqual( [ "docs/dev_dashboard.html", "docs/TODO.md", "docs/ops/backlog-2026-06-26.md", "docs/ops/outcome-os-clean-release-manifest-2026-08-07.json", ], evidence["ssot_sync"]["required_paths"], ) def test_postdeploy_failure_rolls_back_and_proves_previous_images(self) -> None: agent, _, deployment, runtime, candidate = self.make_agent( active_sha="e" * 64, execute=True, runtime_fail_on_call=1, ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() report = raised.exception.report self.assertEqual("runtime_contract", report["failed_stage"]) self.assertEqual("verified", report["rollback"]["status"]) self.assertEqual( [ "read_active_state", "snapshot", "promote", "rollback", "verify_rollback", ], deployment.calls, ) self.assertEqual(2, runtime.calls) self.assertTrue(candidate.cleaned) def test_rollback_image_mismatch_remains_a_hard_failure(self) -> None: agent, runner, deployment, _, _ = self.make_agent( active_sha="e" * 64, execute=True, fail_stage="postdeploy_browser_review", rollback_matches=False, ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("failed", raised.exception.report["rollback"]["status"]) self.assertIn("rollback image proof mismatch", raised.exception.report["rollback"]["error"]) self.assertIn("verify_rollback", deployment.calls) self.assertNotIn("commit_active_state", deployment.calls) self.assertLess( runner.event_log.index("runner:postdeploy_browser_review"), runner.event_log.index("deployment:rollback"), ) def test_postpromotion_runtime_retries_only_transient_readiness_failures(self) -> None: agent, _, deployment, _, candidate = self.make_agent( active_sha="e" * 64, execute=True, ) runtime = TransientRuntimeProbe(self.agent_module, transient_failures=2) agent.runtime_probe = runtime agent.config.runtime_ready_timeout_seconds = 1 agent.config.runtime_retry_interval_seconds = 0 report = agent.run() self.assertTrue(report["ok"]) self.assertEqual(3, runtime.calls) self.assertEqual(3, report["runtime"]["readiness_attempts"]) self.assertIn("promote", deployment.calls) self.assertNotIn("rollback", deployment.calls) self.assertTrue(candidate.cleaned) def test_non_material_milestone_fails_before_release_commands(self) -> None: agent, runner, deployment, _, _ = self.make_agent( active_sha=None, execute=True, ) agent.config.milestone = self.agent_module.MaterialMilestone( milestone_id="cosmetic-copy-only", material=False, goals=("G8",), evidence_refs=("test:copy",), ) with self.assertRaises(self.agent_module.ReleaseAgentFailure) as raised: agent.run() self.assertEqual("material_milestone", raised.exception.report["failed_stage"]) self.assertEqual([], runner.calls) self.assertEqual([], deployment.calls) def test_remote_rollback_recreates_only_api_web_and_proxy(self) -> None: runner = CaptureRunner(self.agent_module) driver = self.agent_module.NasPreviewDeploymentDriver( runner, ssh_target="release-agent@100.116.83.60", ) snapshot = self.agent_module.DeploymentSnapshot( api_image="sha256:" + "1" * 64, web_image="sha256:" + "2" * 64, compose_file=( "/volume1/docker/vignette-preview-20260807/infra/docker-compose.yml" ), env_file="/volume1/docker/vignette-preview-20260807/infra/.env", active_sha="e" * 64, ) driver.rollback(snapshot) remote_call = next(argv for stage, argv in runner.calls if stage == "nas_rollback") remote_command = remote_call[-1] self.assertIn("up -d --no-build --no-deps --force-recreate api web proxy", remote_command) self.assertNotIn(" compose down", remote_command) self.assertNotIn("docker volume", remote_command) self.assertNotIn("docker network", remote_command) self.assertNotIn("--volumes", remote_command) self.assertIn( "--env-file /volume1/docker/vignette-preview-20260807/infra/.env", remote_command, ) def test_remote_promotion_reuses_snapshot_env_before_upload(self) -> None: root = "/volume1/docker/vignette-preview-20260807" runner = CaptureRunner(self.agent_module) driver = self.agent_module.NasPreviewDeploymentDriver( runner, ssh_target="release-agent@100.116.83.60", ) candidate = self.root / "candidate" candidate.mkdir() (candidate / "README.md").write_text("candidate\n", encoding="utf-8") snapshot = self.agent_module.DeploymentSnapshot( api_image="sha256:" + "1" * 64, web_image="sha256:" + "2" * 64, compose_file=f"{root}/release/infra/docker-compose.yml", env_file=f"{root}/release/infra/.env", active_sha="e" * 64, ) state = driver.promote(candidate, "a" * 64, snapshot) stages = [stage for stage, _ in runner.calls] self.assertLess(stages.index("nas_prepare_release"), stages.index("nas_upload_archive")) prepare_command = next( argv[-1] for stage, argv in runner.calls if stage == "nas_prepare_release" ) promote_command = next( argv[-1] for stage, argv in runner.calls if stage == "nas_promote" ) self.assertIn("required env file missing", prepare_command) self.assertIn(f"--env-file {root}/release/infra/.env", promote_command) self.assertNotIn(f"--env-file {root}/infra/.env", promote_command) transactional_migration_command = next( line for line in promote_command.splitlines() if "17_improvement_workbook_contracts.sql" in line and "psql -v ON_ERROR_STOP=1" in line ) online_migration_command = next( line for line in promote_command.splitlines() if "18_admin_usage_ledger_index.sql" in line and "psql -v ON_ERROR_STOP=1" in line ) active_session_online_migration_command = next( line for line in promote_command.splitlines() if "21_single_active_session.sql" in line and "psql -v ON_ERROR_STOP=1" in line ) case_profile_online_migration_command = next( line for line in promote_command.splitlines() if "22_case_profile_multi_case.sql" in line and "psql -v ON_ERROR_STOP=1" in line ) self.assertIn("--single-transaction", transactional_migration_command) self.assertNotIn("--single-transaction", online_migration_command) self.assertNotIn( "--single-transaction", active_session_online_migration_command ) self.assertNotIn( "--single-transaction", case_profile_online_migration_command ) self.assertEqual(f"{root}/release/infra/.env", state["env_file"]) def test_adopted_legacy_state_infers_compose_adjacent_env_file(self) -> None: root = "/volume1/docker/vignette-preview-20260807" runner = ActiveStateRunner( self.agent_module, { "schema": "vignette.release-agent-active.v1", "active_sha": "e" * 64, "compose_project": "vignette-preview-20260807", "remote_root": root, "compose_file": f"{root}/release/infra/docker-compose.yml", "api_image": "sha256:" + "1" * 64, "web_image": "sha256:" + "2" * 64, }, ) driver = self.agent_module.NasPreviewDeploymentDriver( runner, ssh_target="release-agent@100.116.83.60", ) state = driver.read_active_state() self.assertEqual(f"{root}/release/infra/.env", state["env_file"]) def test_remote_copy_uses_synology_compatible_legacy_scp_protocol(self) -> None: runner = CaptureRunner(self.agent_module) driver = self.agent_module.NasPreviewDeploymentDriver( runner, ssh_target="release-agent@100.116.83.60", ) source = self.root / "state.json" source.write_text("{}\n", encoding="utf-8") driver._scp( "nas_upload_test", source, "/volume1/docker/vignette-preview-20260807/incoming/state.json", ) argv = next(argv for stage, argv in runner.calls if stage == "nas_upload_test") self.assertEqual("scp.exe", argv[0]) self.assertIn("-O", argv) def test_candidate_normalizes_generated_contract_and_shell_scripts_to_lf(self) -> None: runner = CandidateArchiveRunner(self.agent_module) manager = self.agent_module.CleanCandidateManager(self.root, runner) candidate = manager.materialize("b" * 40, "release.patch") argv = next(argv for stage, argv in runner.calls if stage == "candidate_patch_apply") self.assertEqual( ["git", "apply", "--whitespace=nowarn"], argv[:-1], ) self.assertEqual( b"export type A = 1;\nexport type B = 2;\n", (candidate / "apps/web/src/lib/api.gen.ts").read_bytes(), ) self.assertEqual( b"#!/usr/bin/env bash\necho ready\n", (candidate / "infra/db/init/99_app_role.sh").read_bytes(), ) self.assertEqual(b"candidate\r\n", (candidate / "README.md").read_bytes()) manager.cleanup() def test_clean_head_candidate_is_archive_only_with_platform_normalization(self) -> None: runner = CandidateArchiveRunner(self.agent_module) manager = self.agent_module.CleanCandidateManager(self.root, runner) archive_path = self.root / "clean-head.tar" source = self.root / "source.txt" source.write_bytes(b"exact committed bytes\r\n") generated_api = self.root / "api.gen.ts" generated_api.write_bytes(b"export type Exact = true;\r\n") shell_script = self.root / "99_app_role.sh" shell_script.write_bytes(b"#!/usr/bin/env bash\r\necho ready\r\n") with tarfile.open(archive_path, "w") as archive: archive.add(source, arcname="source.txt") archive.add(generated_api, arcname="apps/web/src/lib/api.gen.ts") archive.add(shell_script, arcname="infra/db/init/99_app_role.sh") expected_sha = self.agent_module.sha256_bytes(archive_path.read_bytes()) candidate = manager.materialize_archive(archive_path, expected_sha) self.assertEqual(b"exact committed bytes\r\n", (candidate / "source.txt").read_bytes()) self.assertEqual( b"export type Exact = true;\n", (candidate / "apps/web/src/lib/api.gen.ts").read_bytes(), ) self.assertEqual( b"#!/usr/bin/env bash\necho ready\n", (candidate / "infra/db/init/99_app_role.sh").read_bytes(), ) self.assertEqual([], runner.calls) manager.cleanup() def test_clean_head_candidate_rejects_archive_sha_drift(self) -> None: runner = CandidateArchiveRunner(self.agent_module) manager = self.agent_module.CleanCandidateManager(self.root, runner) archive_path = self.root / "clean-head.tar" archive_path.write_bytes(b"not the bound archive") with self.assertRaisesRegex( self.agent_module.StageFailure, "archive SHA drifted", ): manager.materialize_archive(archive_path, "a" * 64) self.assertEqual([], runner.calls) def test_active_state_rejects_noncanonical_compose_path_before_snapshot(self) -> None: root = "/volume1/docker/vignette-preview-20260807" runner = ActiveStateRunner( self.agent_module, { "schema": "vignette.release-agent-active.v1", "active_sha": "e" * 64, "compose_project": "vignette-preview-20260807", "remote_root": root, "compose_file": f"{root}/releases/../other/infra/docker-compose.yml", "api_image": "sha256:" + "1" * 64, "web_image": "sha256:" + "2" * 64, }, ) driver = self.agent_module.NasPreviewDeploymentDriver( runner, ssh_target="release-agent@100.116.83.60", ) with self.assertRaisesRegex(self.agent_module.StageFailure, "Compose file"): driver.read_active_state() self.assertEqual(["nas_read_active_state"], runner.calls) def test_active_state_accepts_only_the_exact_audited_legacy_release_path(self) -> None: root = "/volume1/docker/vignette-preview-20260807" payload = { "schema": "vignette.release-agent-active.v1", "active_sha": "e" * 64, "compose_project": "vignette-preview-20260807", "remote_root": root, "compose_file": f"{root}/release/infra/docker-compose.yml", "api_image": "sha256:" + "1" * 64, "web_image": "sha256:" + "2" * 64, } runner = ActiveStateRunner(self.agent_module, payload) driver = self.agent_module.NasPreviewDeploymentDriver( runner, ssh_target="release-agent@100.116.83.60", ) state = driver.read_active_state() self.assertEqual(payload, {key: state[key] for key in payload}) self.assertEqual(f"{root}/release/infra/.env", state["env_file"]) self.assertEqual(["nas_read_active_state"], runner.calls) def test_candidate_archive_excludes_release_agent_secret_env(self) -> None: candidate = self.root / "archive-candidate" (candidate / ".release-agent").mkdir(parents=True) (candidate / ".release-agent/candidate.env").write_text( "OPENAI_API_KEY=must-not-ship\n", encoding="utf-8", ) (candidate / "README.md").write_text("safe\n", encoding="utf-8") destination = self.root / "candidate.tar.gz" self.agent_module._archive_candidate(candidate, destination) with tarfile.open(destination, "r:gz") as archive: names = archive.getnames() self.assertIn("README.md", names) self.assertFalse(any(name == ".release-agent" or name.startswith(".release-agent/") for name in names)) def test_candidate_compose_project_is_unique_per_process(self) -> None: desired_sha = "a" * 64 first = self.agent_module.candidate_compose_project(desired_sha, process_id=101) second = self.agent_module.candidate_compose_project(desired_sha, process_id=202) self.assertNotEqual(first, second) self.assertEqual("vignette-release-gate-aaaaaaaaaaaa-p101", first) if __name__ == "__main__": unittest.main()