U-042: self-hosting closed-loop acceptance (orchestration + gated real test)

Adds supervisor/self_hosting_e2e.py: run_self_hosting_closed_loop() chains every PH5 module built this session (U-034~U-041) with the Graph Engine's own scheduler (U-032) into one real, callable path matching self-development- feasibility.md's 9-step scenario -- self-target Goal -> Planner/Coder/ Validator -> core_engine classification -> stage -> health check -> cutover (or rollback path A if health check fails) -> monitor until either monitoring_window_clean (finalized) or anomaly_detected (rollback path B). Both real outcome paths are branches of the same call, not two separate scripted scenarios. Explicitly documented as this acceptance scenario's own orchestration glue, not a claim about Deploy Supervisor's eventual production poll loop (U-036's discovery-loop hook is still unfilled; building that is future work). Mocked wiring tests (6, run in normal regression) stub each already independently-tested module boundary and only verify the branching logic. The real, literal acceptance test is built and ready (RealSelfHostingClosedLoopTest, reproduces the document's own "loop_round > 3 should be >= 3" example) but gated behind RUN_REAL_SELF_HOSTING_E2E=1 -- deliberately a separate, distinctly-named env var from this project's usual RUN_REAL_AGENT_E2E, since this is the highest-blast-radius action in the project (can replace the running graphlemon process and git-revert commits on this real repo). Per this Unit's own explicit build instruction, not executed as part of implementing it -- doing so requires checking with the user first. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-042: self-hosting closed-loop acceptance (orchestration + gated real test)
Adds supervisor/self_hosting_e2e.py: run_self_hosting_closed_loop() chains every PH5 module built this session (U-034~U-041) with the Graph Engine's own scheduler (U-032) into one real, callable path matching self-development- feasibility.md's 9-step scenario -- self-target Goal -> Planner/Coder/ Validator -> core_engine classification -> stage -> health check -> cutover (or rollback path A if health check fails) -> monitor until either monitoring_window_clean (finalized) or anomaly_detected (rollback path B). Both real outcome paths are branches of the same call, not two separate scripted scenarios. Explicitly documented as this acceptance scenario's own orchestration glue, not a claim about Deploy Supervisor's eventual production poll loop (U-036's discovery-loop hook is still unfilled; building that is future work). Mocked wiring tests (6, run in normal regression) stub each already independently-tested module boundary and only verify the branching logic. The real, literal acceptance test is built and ready (RealSelfHostingClosedLoopTest, reproduces the document's own "loop_round > 3 should be >= 3" example) but gated behind RUN_REAL_SELF_HOSTING_E2E=1 -- deliberately a separate, distinctly-named env var from this project's usual RUN_REAL_AGENT_E2E, since this is the highest-blast-radius action in the project (can replace the running graphlemon process and git-revert commits on this real repo). Per this Unit's own explicit build instruction, not executed as part of implementing it -- doing so requires checking with the user first. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 4f4b1853
Showing 4 changed files with 410 additions and 1 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
supervisor/self_hosting_e2e.py
supervisor/test_self_hosting_e2e.py
@@ -396,6 +396,10 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-042 [Risk Gate]** 自举闭环端到端验收：真实复现 `self-development-feasibility.md` 第 3 节的 9 步场景（修复 graphlemon 自己 `p_org_task_lifecycle` 状态机里的一个真实植入 bug），走完整"部署→健康检查→cutover→监控→finalize"或"检测异常→自动回滚"两条路径中至少一条
   - 依赖：U-035, U-041
   - Done：全程无需人工评审代码内容；参照 graphloop `RUN_REAL_AGENT_E2E=1` 的先例，真实拉起两个 graphlemon 实例的端到端用例默认跳过，需要显式环境变量开启，日常回归只跑 mock 版本
+  - **实现补充（2026-08-02）**：新增 `supervisor/self_hosting_e2e.py`：`run_self_hosting_closed_loop()` 把这次会话建好的全部 PH5 模块（U-034~U-041）+ Graph Engine 自己的 `scheduler.run_goal_to_completion`（U-032）串成一条真实可调用的链路——真实 self-target Goal → Planner/Coder/Validator（真实 `claude` 调用，`scheduler` 内部已经走 U-035 的 self-target 强制 worktree）→ PASS 后走 U-035 记的 `core_engine` 审计 `bug_record` → `deploy_stage.stage_deployment` → `deploy_health_check.run_health_check` → 通过则 `deploy_cutover.cutover`，不通过直接 `deploy_rollback.rollback_path_a` 停在这一步 → 循环 `deploy_monitor.run_monitor_cycle` 直到 `monitoring_window_clean`（`finalized`）或 `anomaly_detected`（`deploy_rollback.rollback_path_b`）——**这正是本 Unit 要求的"两条路径至少走一条"，实现上两条路径都是同一次真实调用可能走到的真实分支，不是分别写死两个脚本**。**明确不是**生产级 Deploy Supervisor 的最终 `poll_fn` 实现——真正独立轮询"有没有新的 self-target 合并等待部署"的发现机制是 U-036 留的钩子，本 Unit 只提供验收场景自己的编排入口，不代表以后正式的常驻发现循环长这样，避免把验收脚本和生产实现搞混。
+    - **Mock 版本（日常回归）**：`test_self_hosting_e2e.py` 前 6 个测试类在每个已经独立测试过真实机制的模块边界上打桩（`scheduler.run_goal_to_completion`/`deploy_stage.stage_deployment`/`deploy_health_check.run_health_check`/`deploy_cutover.cutover`/`deploy_monitor.run_monitor_cycle`/`deploy_rollback.rollback_path_a`/`rollback_path_b`），只验证编排本身的分支/顺序逻辑对不对（任务没到 done 提前返回、不是 core_engine 提前返回、健康检查失败走 A 不碰 cutover、监控干净走 finalized、监控异常走 B），不重新证明任何一个模块自己的真实机制——那些已经在各自的测试文件里用真实进程/真实 git 仓库验证过了。
+    - **真实版本，已建好但没有执行**：`RealSelfHostingClosedLoopTest`，字面复现 `self-development-feasibility.md` 例子本身（"loop_round > 3 应该是 loop_round >= 3"）。**用了一个和项目一贯的 `RUN_REAL_AGENT_E2E` 不同、单独命名的环境变量 `RUN_REAL_SELF_HOSTING_E2E`**——这是本项目目前为止影响面最大的一个动作，不只是花真钱调用 `claude`（那是 `RUN_REAL_AGENT_E2E` 覆盖的常规风险），而是真的可能替换掉正在跑的 graphlemon 进程本身、真的会在这个仓库自己的 git 历史上跑 `git revert`——用不同名字的环境变量，是为了不让"为了跑别的真实测试而设的 `RUN_REAL_AGENT_E2E=1`"意外带着触发这个。**没有替用户执行这个测试**：按本 Unit 开工前收到的明确指示，这类"影响范围超出本进程/本会话"的动作，写好、gate 好就停，真正执行前需要用 AskUserQuestion 问过用户——这次实现结束后立刻会问。
+  - 覆盖：`self_hosting_e2e.py`（编排层，串联 U-032/034~041 全部真实模块）；真实验收场景已建好、默认不跑
 ### PH5-S4 可观测性（非阻塞，可延后）
@@ -59,7 +59,7 @@
     {"id": "U-039", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_cutover（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-038"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_cutover"], "process": [], "requirement": []}, "done": "健康检查通过后完成真实切换（supervisor/deploy_cutover.py，真实进程停止+重启验证过）；在途 work_node_execution 重启后可恢复（复用既有 REQ-06 机制）；previous_instance_dir 保留至监控窗口结束；新增 LIVE_PORT=8420 常量（05 §5.9 缺的一个字段）；真实测试发现并修复 deploy_supervisor.is_alive() 的僵尸进程判活缺陷（新增 reap_if_exited()）"},
     {"id": "U-040", "phase": "PH5", "sub_phase": "PH5-S3", "name": "c_deploy_supervisor_monitor（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-039"], "coverage": {"model": ["deploy_health_signal"], "view": [], "controller": ["c_deploy_supervisor_monitor"], "process": [], "requirement": []}, "done": "四类信号采样正常（supervisor/deploy_monitor.py）；validator_approval_rate 偏离基线阈值时触发 anomaly_detected；approve 占比算法基于对 scheduler.finish_task_lifecycle 真实代码逻辑的核实（PASS 从不产生 bug_record，FAIL 必然产生），非字面 SDD 定义"},
     {"id": "U-041", "phase": "PH5", "sub_phase": "PH5-S3", "name": "c_deploy_supervisor_rollback（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-038", "U-040"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_rollback"], "process": [], "requirement": []}, "done": "回滚路径 A/B 各跑通一次（supervisor/deploy_rollback.py，真实进程+真实 git 仓库验证）；旧版本也起不来时正确终态到 blocked_pending_human（真实场景验证，非假设）；git revert 冲突有明确处理（真实构造冲突历史验证过，含 --abort 清理）"},
-    {"id": "U-042", "phase": "PH5", "sub_phase": "PH5-S3", "name": "自举闭环端到端验收", "classification": "Risk Gate", "depends_on": ["U-035", "U-041"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "真实复现 self-development-feasibility.md 9 步场景，全程无需人工评审代码内容；真实双实例 E2E 用例默认跳过，需显式环境变量开启"},
+    {"id": "U-042", "phase": "PH5", "sub_phase": "PH5-S3", "name": "自举闭环端到端验收（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-035", "U-041"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "真实复现 self-development-feasibility.md 9 步场景，全程无需人工评审代码内容（supervisor/self_hosting_e2e.py 编排层已建好）；真实双实例 E2E 用例（RUN_REAL_SELF_HOSTING_E2E=1，独立于常规 RUN_REAL_AGENT_E2E 的单独环境变量）默认跳过，需显式环境变量开启且需额外用户确认才真正执行；日常回归只跑 mock 版本（6 个编排逻辑测试，已跑通）"},
     {"id": "U-043", "phase": "PH5", "sub_phase": "PH5-S4", "name": "控制台展示 deployment 状态与健康信号", "classification": "Value Path", "depends_on": ["U-024", "U-040"], "coverage": {"model": [], "view": ["v_deployment_status"], "controller": [], "process": [], "requirement": []}, "done": "能看到当前 deployment 状态机所处状态与最近一次健康信号采样结果"}
   ],
   "classification_summary": {
+"""U-042: self-hosting closed-loop acceptance -- wires every PH5 module
+built so far (U-034~U-041) plus the Graph Engine's own scheduler into the
+literal 9-step scenario `self-development-feasibility.md` walks through:
+a real self-target Goal, Planner->Coder->Validator (real `claude` calls via
+`scheduler.run_goal_to_completion`, which already forces worktree isolation
+and records the `core_engine` audit trail for self-target tasks per U-035),
+merge, stage a candidate, health-check it, cut it over, monitor it, and
+land on EITHER `finalized` (clean monitoring window) OR a rollback (anomaly
+detected) -- "走完整...两条路径中至少一条", not a scripted single outcome.
+
+This module is the acceptance scenario's own orchestration glue, not a
+claim about what Deploy Supervisor's real production `poll_fn` (U-036's
+still-unfilled hook) should look like long-term -- a continuously-running
+daemon would discover self-target merges to stage on its own schedule
+rather than drive one linear call end to end. Building that discovery loop
+is future work, not this Unit's job; this function exists so the acceptance
+scenario has one real, callable entry point to run (or NOT run -- see
+test_self_hosting_e2e.py's gating) against.
+"""
+
+import time
+
+from supervisor import deploy_cutover, deploy_health_check, deploy_monitor, deploy_rollback, deploy_stage
+from src import paths, schema, scheduler
+
+
+class SelfHostingLoopError(Exception):
+    """Raised when a precondition for even attempting the scenario isn't
+    met (e.g. target_folder_path isn't genuinely self-target) -- distinct
+    from a real outcome like 'rolled_back', which is a valid, successful
+    exercise of the safety mechanism, not an error."""
+
+
+def run_self_hosting_closed_loop(
+    db_path, goal_text, target_folder_path=None, prod_db_path=None,
+    poll_interval=5.0, monitor_timeout_cycles=None, sleep=time.sleep, now=time.time,
+):
+    """Returns {"stage", "outcome", ...}. `stage` names how far the scenario
+    got (`task_lifecycle` / `classification` / `health_check` / `cutover` /
+    `monitor`); `outcome` is the concrete result at that stage. A task that
+    doesn't reach 'done', or doesn't get classified core_engine, is a valid
+    early-return (not every self-target task is a core-engine change -- see
+    U-035), not a scenario failure.
+    """
+    target_folder_path = target_folder_path or paths.repo_root()
+    prod_db_path = prod_db_path or paths.db_path()
+    if not paths.is_self_target(target_folder_path):
+        raise SelfHostingLoopError(
+            "target_folder_path {!r} is not this instance's own checkout ({!r}) -- "
+            "the self-hosting scenario requires self-target".format(target_folder_path, paths.repo_root())
+        )
+
+    task_result = scheduler.run_goal_to_completion(db_path, target_folder_path, goal_text)
+    if task_result.get("final_status") != "done":
+        return {"stage": "task_lifecycle", "outcome": task_result.get("final_status"), "task_id": task_result.get("task_id")}
+
+    conn = schema.connect(db_path)
+    try:
+        bug = conn.execute(
+            "SELECT bug_id FROM bug_record WHERE task_id = ? AND root_cause_category = 'core_engine' "
+            "ORDER BY rowid DESC LIMIT 1",
+            (task_result["task_id"],),
+        ).fetchone()
+        if bug is None:
+            return {"stage": "classification", "outcome": "not_core_engine", "task_id": task_result["task_id"]}
+
+        merge_commit = deploy_stage._current_commit(target_folder_path)
+
+        staged = deploy_stage.stage_deployment(
+            conn, merge_commit, repo_root=target_folder_path, prod_db_path=prod_db_path
+        )
+        health = deploy_health_check.run_health_check(
+            conn, staged["deployment_id"], staged["candidate_pid"], staged["candidate_port"]
+        )
+        if health["status"] != "health_check_passed":
+            candidate_db_path = conn.execute(
+                "SELECT candidate_db_path FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],)
+            ).fetchone()[0]
+            deploy_rollback.rollback_path_a(
+                conn, staged["deployment_id"], staged["candidate_pid"], candidate_db_path,
+                merge_commit, repo_root=target_folder_path,
+                description="health check failed: {}".format(health["signals"]),
+            )
+            return {"stage": "health_check", "outcome": "rolled_back_path_a", "deployment_id": staged["deployment_id"]}
+
+        cutover_result = deploy_cutover.cutover(
+            conn, staged["deployment_id"], live_pid=None, candidate_pid=staged["candidate_pid"],
+            candidate_instance_dir=staged["candidate_instance_dir"], prod_db_path=prod_db_path,
+        )
+
+        project_id = conn.execute(
+            "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
+        ).fetchone()[0]
+        cutover_at = conn.execute(
+            "SELECT cutover_at FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],)
+        ).fetchone()[0]
+        baseline = deploy_monitor.compute_baseline_validator_approval_rate(conn, project_id, cutover_at)
+        baseline_rate = baseline[0] if baseline else None
+
+        cycles = 0
+        while True:
+            cycles += 1
+            cycle_result = deploy_monitor.run_monitor_cycle(
+                conn, staged["deployment_id"], cutover_result["live_pid"], cutover_result["live_port"],
+                project_id, baseline_rate, cutover_at, now=now,
+            )
+            if cycle_result["status"] == "monitoring_window_clean":
+                conn.execute(
+                    "UPDATE deployment SET status = 'finalized' WHERE deployment_id = ?", (staged["deployment_id"],)
+                )
+                conn.commit()
+                return {"stage": "monitor", "outcome": "finalized", "deployment_id": staged["deployment_id"]}
+            if cycle_result["status"] == "anomaly_detected":
+                rb = deploy_rollback.rollback_path_b(
+                    conn, staged["deployment_id"], cutover_result["live_pid"],
+                    _previous_instance_dir(conn, staged["deployment_id"]),
+                    prod_db_path, merge_commit, repo_root=target_folder_path,
+                    description="post-cutover anomaly: {}".format(cycle_result["latest_signal"]),
+                )
+                outcome = "rolled_back_path_b" if rb["status"] == "rolled_back" else "rollback_failed"
+                return {"stage": "monitor", "outcome": outcome, "deployment_id": staged["deployment_id"]}
+            if monitor_timeout_cycles is not None and cycles >= monitor_timeout_cycles:
+                return {"stage": "monitor", "outcome": "timed_out_still_monitoring", "deployment_id": staged["deployment_id"]}
+            sleep(poll_interval)
+    finally:
+        conn.close()
+
+
+def _previous_instance_dir(conn, deployment_id):
+    return conn.execute(
+        "SELECT previous_instance_dir FROM deployment WHERE deployment_id = ?", (deployment_id,)
+    ).fetchone()[0]
+"""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())
+
+        self.assertEqual(result, {"stage": "monitor", "outcome": "finalized", "deployment_id": "dep-1"})
+        mock_monitor.assert_called_once()
+        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()