"""U-042 tests: self_hosting_e2e.py's orchestration wiring, mocked at each already-independently-tested module's own boundary (deploy_stage/ deploy_health_check/deploy_cutover/deploy_monitor/deploy_rollback/scheduler each have their own real-mechanics test files -- this file's job is proving run_self_hosting_closed_loop() sequences and branches between them correctly, not re-proving any of their own internals). The genuinely real, un-mocked acceptance run (self-development-feasibility.md's literal 9-step scenario, actual `claude` calls, an actual candidate cutover of THIS repo) lives in RealSelfHostingClosedLoopTest below, gated behind RUN_REAL_SELF_HOSTING_E2E=1 -- a SEPARATE, stricter env var from this project's usual RUN_REAL_AGENT_E2E, since this is the single highest-blast- radius action in the whole project (it can actually replace the running graphlemon process and git-revert commits on this real repo). Not run as part of this Unit's own implementation -- see development-plan.md's U-042 note for why. Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v """ import os import sys import unittest from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from supervisor import ( deploy_cutover, deploy_health_check, deploy_monitor, deploy_rollback, deploy_stage, self_hosting_e2e, ) from src import paths, schema, scheduler class SelfTargetGuardTest(unittest.TestCase): def test_non_self_target_raises_before_doing_anything(self): with self.assertRaises(self_hosting_e2e.SelfHostingLoopError): self_hosting_e2e.run_self_hosting_closed_loop("db.sqlite", "goal", target_folder_path="/tmp/not-graphlemon") class TaskLifecycleNotDoneTest(unittest.TestCase): def test_returns_early_without_touching_deploy_supervisor(self): with patch.object(scheduler, "run_goal_to_completion", return_value={"task_id": "t1", "final_status": "blocked_pending_human"}), \ patch.object(deploy_stage, "stage_deployment") as mock_stage: result = self_hosting_e2e.run_self_hosting_closed_loop("unused.db", "goal", target_folder_path=paths.repo_root()) self.assertEqual(result, {"stage": "task_lifecycle", "outcome": "blocked_pending_human", "task_id": "t1"}) mock_stage.assert_not_called() class NotCoreEngineTest(unittest.TestCase): def test_returns_early_when_no_core_engine_audit_bug_exists(self): import tempfile with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: db_path = tmp.name conn = schema.connect(db_path) schema.create_schema(conn) conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', ?, 1, 'active')", (paths.repo_root(),) ) conn.commit() conn.close() try: with patch.object(scheduler, "run_goal_to_completion", return_value={"task_id": "t1", "final_status": "done"}), \ patch.object(deploy_stage, "stage_deployment") as mock_stage: result = self_hosting_e2e.run_self_hosting_closed_loop(db_path, "goal", target_folder_path=paths.repo_root()) self.assertEqual(result, {"stage": "classification", "outcome": "not_core_engine", "task_id": "t1"}) mock_stage.assert_not_called() finally: os.remove(db_path) class _RealDbTestBase(unittest.TestCase): def setUp(self): import tempfile self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) self.tmp.close() self.db_path = self.tmp.name self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path)) conn = schema.connect(self.db_path) schema.create_schema(conn) conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', ?, 1, 'active')", (paths.repo_root(),) ) conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'done', 'validator')" ) conn.execute( "INSERT INTO bug_record (bug_id, task_id, source, description, root_cause_category, status) " "VALUES ('bug-1', 'task-1', 'validator', 'self-target change', 'core_engine', 'resolved')" ) conn.commit() conn.close() def _mock_scheduler(self): return patch.object( scheduler, "run_goal_to_completion", return_value={"task_id": "task-1", "final_status": "done"}, ) class HealthCheckFailedTest(_RealDbTestBase): def test_routes_to_rollback_path_a_and_stops(self): conn = schema.connect(self.db_path) conn.execute( "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, " "previous_instance_dir, candidate_db_path) " "VALUES ('dep-1', 'c1', 'c0', 'candidate_starting', '/prev', '/tmp/candidate.db')" ) conn.commit() conn.close() with self._mock_scheduler(), \ patch.object(deploy_stage, "_current_commit", return_value="deadbeef"), \ patch.object(deploy_stage, "stage_deployment", return_value={ "deployment_id": "dep-1", "status": "candidate_starting", "candidate_port": 1, "candidate_pid": 2, "candidate_instance_dir": "/x", }), \ patch.object(deploy_health_check, "run_health_check", return_value={ "status": "health_check_failed", "signals": [{"signal_type": "http_health", "status": "fail", "detail": "x"}], }), \ patch.object(deploy_rollback, "rollback_path_a", return_value={"status": "rolled_back"}) as mock_rb_a, \ patch.object(deploy_cutover, "cutover") as mock_cutover: result = self_hosting_e2e.run_self_hosting_closed_loop(self.db_path, "goal", target_folder_path=paths.repo_root()) self.assertEqual(result, {"stage": "health_check", "outcome": "rolled_back_path_a", "deployment_id": "dep-1"}) mock_rb_a.assert_called_once() mock_cutover.assert_not_called() class MonitorReachesFinalizedTest(_RealDbTestBase): def test_clean_monitoring_window_finalizes_deployment(self): conn = schema.connect(self.db_path) conn.execute( "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, " "previous_instance_dir, cutover_at) " "VALUES ('dep-1', 'c1', 'c0', 'live', '/prev', '2020-01-01 00:00:00')" ) conn.commit() conn.close() with self._mock_scheduler(), \ patch.object(deploy_stage, "_current_commit", return_value="deadbeef"), \ patch.object(deploy_stage, "stage_deployment", return_value={ "deployment_id": "dep-1", "status": "candidate_starting", "candidate_port": 1, "candidate_pid": 2, "candidate_instance_dir": "/x", }), \ patch.object(deploy_health_check, "run_health_check", return_value={"status": "health_check_passed", "signals": []}), \ patch.object(deploy_cutover, "cutover", return_value={"status": "live", "live_pid": 3, "live_port": 9999}), \ patch.object(deploy_monitor, "compute_baseline_validator_approval_rate", return_value=None), \ patch.object(deploy_monitor, "run_monitor_cycle", return_value={ "status": "monitoring_window_clean", "latest_signal": {"signal_type": "process_alive", "status": "ok", "detail": None}, }) as mock_monitor, \ patch.object(deploy_rollback, "rollback_path_b") as mock_rb_b: result = self_hosting_e2e.run_self_hosting_closed_loop( self.db_path, "goal", target_folder_path=paths.repo_root(), monitor_window_hours=0.001, monitor_min_task_count=1, ) self.assertEqual(result, {"stage": "monitor", "outcome": "finalized", "deployment_id": "dep-1"}) mock_monitor.assert_called_once() # 2026-08-03: a real run reached 'cutover' and then couldn't # practically finish observing the monitor stage -- the real # production window is 24h/10 real tasks (deploy_monitor's own # constants), impractical to wait out interactively. These must be # overridable and actually reach run_monitor_cycle, not silently # stuck at the real production defaults. self.assertEqual(mock_monitor.call_args.kwargs["window_hours"], 0.001) self.assertEqual(mock_monitor.call_args.kwargs["min_task_count"], 1) mock_rb_b.assert_not_called() conn = schema.connect(self.db_path) status = conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone()[0] conn.close() self.assertEqual(status, "finalized") class MonitorDetectsAnomalyTest(_RealDbTestBase): def test_anomaly_routes_to_rollback_path_b(self): conn = schema.connect(self.db_path) conn.execute( "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, " "previous_instance_dir, cutover_at) " "VALUES ('dep-1', 'c1', 'c0', 'live', '/prev', '2020-01-01 00:00:00')" ) conn.commit() conn.close() with self._mock_scheduler(), \ patch.object(deploy_stage, "_current_commit", return_value="deadbeef"), \ patch.object(deploy_stage, "stage_deployment", return_value={ "deployment_id": "dep-1", "status": "candidate_starting", "candidate_port": 1, "candidate_pid": 2, "candidate_instance_dir": "/x", }), \ patch.object(deploy_health_check, "run_health_check", return_value={"status": "health_check_passed", "signals": []}), \ patch.object(deploy_cutover, "cutover", return_value={"status": "live", "live_pid": 3, "live_port": 9999}), \ patch.object(deploy_monitor, "compute_baseline_validator_approval_rate", return_value=None), \ patch.object(deploy_monitor, "run_monitor_cycle", return_value={ "status": "anomaly_detected", "latest_signal": {"signal_type": "http_health", "status": "fail", "detail": "x"}, }), \ patch.object(deploy_rollback, "rollback_path_b", return_value={"status": "rolled_back"}) as mock_rb_b: result = self_hosting_e2e.run_self_hosting_closed_loop(self.db_path, "goal", target_folder_path=paths.repo_root()) self.assertEqual(result, {"stage": "monitor", "outcome": "rolled_back_path_b", "deployment_id": "dep-1"}) mock_rb_b.assert_called_once() args, kwargs = mock_rb_b.call_args self.assertEqual(args[2], 3) # live_pid from cutover's own return value self.assertEqual(args[3], "/prev") # previous_instance_dir threaded through correctly class RealSelfHostingClosedLoopTest(unittest.TestCase): """The literal acceptance scenario: self-development-feasibility.md's 9 steps, for real -- a real self-target Goal ("fix the loop_round off-by-one boundary bug in p_org_task_lifecycle", the exact example that document uses), real Planner/Coder/Validator `claude` calls, and (if the Coder's fix genuinely lands on a core-engine file and Validator PASSes) a REAL candidate cutover of a worktree checkout of THIS repo -- meaning a real subprocess bound to a real port, a real one-time DB copy, and on success a real `git revert`-reversible commit or a real rollback exercised against this actual repository's git history. Gated behind RUN_REAL_SELF_HOSTING_E2E=1 -- deliberately a DIFFERENT, stricter env var than this project's usual RUN_REAL_AGENT_E2E (used by every other real-`claude`-call test in this codebase), because this is the single highest-blast-radius action in the whole project: unlike a normal RUN_REAL_AGENT_E2E test (which costs real API usage but only touches ephemeral temp directories/in-memory state), this one can actually replace the currently-running graphlemon process and mutate this real repository's commit history via `git revert`. Requiring a SEPARATE, unambiguously-named env var means accidentally setting RUN_REAL_AGENT_E2E=1 to run some other, cheaper real test in this codebase can never accidentally trigger this one too. Not executed as part of implementing this Unit -- per the explicit instruction this Unit was built under, flipping this switch requires checking with the user first (AskUserQuestion), same as any other action whose blast radius extends beyond this process/session. See development-plan.md's U-042 note for that decision's own record. """ @unittest.skipUnless( os.environ.get("RUN_REAL_SELF_HOSTING_E2E") == "1", "set RUN_REAL_SELF_HOSTING_E2E=1 to run -- shells out to real `claude` calls AND performs a " "real candidate cutover of this repository; costs real API usage and is genuinely destructive " "if anything here has a bug. Requires explicit user sign-off, not just this env var, before " "ever actually being run -- see development-plan.md's U-042 note.", ) def test_real_self_development_feasibility_9_step_scenario(self): import tempfile with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: db_path = tmp.name try: conn = schema.connect(db_path) schema.create_schema(conn) conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'graphlemon-self', ?, 1, 'active')", (paths.repo_root(),) ) conn.commit() conn.close() result = self_hosting_e2e.run_self_hosting_closed_loop( db_path, "Fix the loop_round off-by-one boundary bug in p_org_task_lifecycle: " "loop_round > 3 should be loop_round >= 3 for the blocked_pending_human transition.", target_folder_path=paths.repo_root(), monitor_timeout_cycles=5, ) self.assertIn(result["stage"], ("health_check", "cutover", "monitor")) self.assertIn( result["outcome"], ("finalized", "rolled_back_path_a", "rolled_back_path_b", "timed_out_still_monitoring"), ) finally: if os.path.exists(db_path): os.remove(db_path) if __name__ == "__main__": unittest.main()