U-040: c_deploy_supervisor_monitor (four post-cutover signals)

Adds supervisor/deploy_monitor.py: run_monitor_cycle() is one sampling pass, called repeatedly by Deploy Supervisor's own poll loop (03-C's own text: not a one-shot call). process_alive/http_health/canary_task/ validator_approval_rate sampled in order, stopping at the first anomaly. validator_approval_rate's "approve rate" has no literal definition anywhere in 01-M/03-C/04-P (no such execution_event type exists) -- derived instead from a verified structural fact in scheduler.finish_task_lifecycle: PASS closes straight to done with no bug_record ever created, FAIL/INCONCLUSIVE always creates one via judge_root_cause. A task counts as approved iff no bug_record(source='validator') exists for it. Baseline is computed by the same function looking back MONITORING_WINDOW_HOURS from cutover_at rather than persisted as a new column; baseline=None (cold start) never flags an anomaly rather than defaulting to 0. Crash-count/consecutive-http-fail windows are approximated by sample count (N = CRASH_WINDOW / MONITOR_SAMPLE_INTERVAL) rather than a true rolling time window -- documented simplification, correct as long as sampling stays roughly interval-paced. Caught my own design misreading via the tests: process_alive's monitoring- phase anomaly condition is MAX_CRASH_COUNT crashes within CRASH_WINDOW (a crash-loop detector), not "currently not alive" -- unlike U-038's health check where any single failure fails closed. A single dead reading during monitoring is recorded but must not trigger an immediate rollback by itself. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-040: c_deploy_supervisor_monitor (four post-cutover signals)
Adds supervisor/deploy_monitor.py: run_monitor_cycle() is one sampling pass, called repeatedly by Deploy Supervisor's own poll loop (03-C's own text: not a one-shot call). process_alive/http_health/canary_task/ validator_approval_rate sampled in order, stopping at the first anomaly. validator_approval_rate's "approve rate" has no literal definition anywhere in 01-M/03-C/04-P (no such execution_event type exists) -- derived instead from a verified structural fact in scheduler.finish_task_lifecycle: PASS closes straight to done with no bug_record ever created, FAIL/INCONCLUSIVE always creates one via judge_root_cause. A task counts as approved iff no bug_record(source='validator') exists for it. Baseline is computed by the same function looking back MONITORING_WINDOW_HOURS from cutover_at rather than persisted as a new column; baseline=None (cold start) never flags an anomaly rather than defaulting to 0. Crash-count/consecutive-http-fail windows are approximated by sample count (N = CRASH_WINDOW / MONITOR_SAMPLE_INTERVAL) rather than a true rolling time window -- documented simplification, correct as long as sampling stays roughly interval-paced. Caught my own design misreading via the tests: process_alive's monitoring- phase anomaly condition is MAX_CRASH_COUNT crashes within CRASH_WINDOW (a crash-loop detector), not "currently not alive" -- unlike U-038's health check where any single failure fails closed. A single dead reading during monitoring is recorded but must not trigger an immediate rollback by itself. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 002bba46
Showing 4 changed files with 471 additions and 1 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
supervisor/deploy_monitor.py
supervisor/test_deploy_monitor.py
@@ -383,6 +383,8 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-040 [Risk Gate]** `c_deploy_supervisor_monitor`：post-cutover 监控窗口内四类信号采样（`process_alive`/`http_health`/`canary_task`/`validator_approval_rate`）
   - 依赖：U-039
   - Done：`validator_approval_rate` 信号能从 `execution_event`/`work_node_execution` 正确计算滚动窗口内 Validator 的 approve 占比，人为把该占比推到偏离基线阈值之外时能触发 `anomaly_detected`
+  - **实现补充（2026-08-02）**：新增 `supervisor/deploy_monitor.py`，`run_monitor_cycle()` 是"一次采样"的最小单元，由 Deploy Supervisor 自己的调度循环反复调用（03-C 原文就是这么写的，不是一次性调用）。**`validator_approval_rate` 的"approve 占比"具体怎么算，01-M/03-C/04-P 都没有字面定义**（`execution_event.event_type` 枚举里根本没有"validator_approved"这类事件）——核对了 `scheduler.finish_task_lifecycle()` 的真实代码后确认：Validator 判 PASS 时直接 `close_task_done()`，从不产生 `bug_record`；判 FAIL/INCONCLUSIVE 时 `org_level_loop.run_fix_loop()` 必然会通过 `root_cause_trace.judge_root_cause()` 产生一条 `bug_record(source='validator')`。所以"这个任务 Validator 是否一次通过"可以从"这个任务有没有对应的 `source='validator'` bug_record"精确推导，不是猜的，是核对真实代码逻辑后得出的结构性事实。基线（"与该 deployment 之前的历史基线比较"）不额外建表存，直接用同一个函数在 cutover 时间点往前回溯 `MONITORING_WINDOW_HOURS` 小时现算，`baseline_rate=None`（没有足够历史数据）时约定为"不判定异常"而不是当成 0，避免冷启动阶段被误判。崩溃计数/连续失败窗口按 05 §5.9 字面是"时间窗口内计数"，这里简化成"按最近 N 次采样计数"（N = CRASH_WINDOW / MONITOR_SAMPLE_INTERVAL），前提是采样节奏基本均匀——这是有意简化，注释里写清楚了，不是漏做。process_alive 信号在监控阶段的语义和 U-038 健康检查阶段不同：健康检查阶段任何一次不活着就直接 fail；监控阶段按设计文档字面"连续崩溃 ≥ MAX_CRASH_COUNT 次才算 anomaly"，单次挂掉只记录不触发——第一版测试写反了这条语义（以为单次挂掉就该立刻 anomaly），跑起来才意识到这是"崩溃循环检测"不是"瞬时存活检测"，改正后测试反而更准确地锁定了这条容易读错的规则。
+  - 覆盖：03-C `c_deploy_supervisor_monitor`；01-M `deploy_health_signal`（monitoring 阶段持续写入）
  
 - **U-041 [Risk Gate]** `c_deploy_supervisor_rollback`：回滚路径 A（pre-cutover，终止候选+`git revert`+发 `bug_record`）与路径 B（post-cutover，换回 `previous_instance_dir`+健康检查+`git revert`）完整实现，含"回滚本身失败"兜底转 `blocked_pending_human`
   - 依赖：U-038, U-040
@@ -57,7 +57,7 @@
     {"id": "U-037", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_stage（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-034", "U-036"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_stage"], "process": [], "requirement": []}, "done": "能生成 deployment 记录并拉起监听临时端口、连一次性 DB 副本的候选进程（supervisor/deploy_stage.py，真实 git worktree + 真实子进程验证过）；顺带补上 deployment.candidate_instance_dir 缺字段 + backend_server.py 首个 CLI 入口（此前只能进程内启动，从未真正作为独立 OS 进程跑过）"},
     {"id": "U-038", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_health_check（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-037"], "coverage": {"model": ["deploy_health_signal"], "view": [], "controller": ["c_deploy_supervisor_health_check"], "process": [], "requirement": []}, "done": "process_alive/http_health/canary_task 三项检查全过 -> health_check_passed；任一失败 -> health_check_failed 且 Live 未受影响（supervisor/deploy_health_check.py，真实 HTTP+真实 project/task 创建验证过）；backend_server.py 新增 GET /api/health（真查一次 SELECT 1，不只是进程存活）"},
     {"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", "classification": "Risk Gate", "depends_on": ["U-039"], "coverage": {"model": ["deploy_health_signal"], "view": [], "controller": ["c_deploy_supervisor_monitor"], "process": [], "requirement": []}, "done": "四类信号采样正常；validator_approval_rate 偏离基线阈值时触发 anomaly_detected"},
+    {"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", "classification": "Risk Gate", "depends_on": ["U-038", "U-040"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_rollback"], "process": [], "requirement": []}, "done": "回滚路径 A/B 各跑通一次；旧版本也起不来时正确终态到 blocked_pending_human；git revert 冲突有明确处理"},
     {"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-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 状态机所处状态与最近一次健康信号采样结果"}
+"""c_deploy_supervisor_monitor (03-C §3.15 / 04-P §4.6 / design doc §6):
+post-cutover monitoring -- four signals sampled periodically while
+`deployment.status='monitoring'`, called repeatedly by Deploy Supervisor's
+own poll loop (U-036's `poll_fn`), not a one-shot call (03-C's own text:
+"由 Deploy Supervisor 自身的调度循环周期性调用").
+
+`run_monitor_cycle()` is ONE sampling pass: process_alive + http_health +
+validator_approval_rate every cycle, canary_task only when the caller says
+`run_canary=True` (real scheduling of "every CANARY_INTERVAL" is the
+caller's job -- this module just runs whatever it's told to run once,
+matching `c_deploy_supervisor_health_check`'s own "idempotent, doesn't
+itself decide cadence" framing).
+
+**`validator_approval_rate` operationalization (documented interpretation,
+not literally spelled out anywhere in 03-C/04-P/design doc -- there's no
+`event_type='validator_approved'` or similar in 01-M's execution_event enum,
+so "approved 占比" has to be derived from what IS actually persisted)**: a
+task counts as Validator-approved if it reached `task_status_changed(
+to_status='validating')` and NO `bug_record(source='validator')` was ever
+created for it afterward (Validator's real judgment: PASS -> straight to
+`close_task_done`, no bug_record at all; FAIL/INCONCLUSIVE -> `org_level_
+loop.run_fix_loop` always creates one via `root_cause_trace.judge_root_cause`
+-> `bug_record.create_bug_record`). This is a real, always-true structural
+fact about how this codebase's Validator path already works (scheduler.py's
+`finish_task_lifecycle`), not a guess -- verified against that code before
+writing this, not merely assumed.
+
+**Crash-count / consecutive-fail windows are approximated by sample COUNT,
+not wall-clock TIME (documented simplification)**: `MAX_CRASH_COUNT`/
+`CRASH_WINDOW` and `MAX_CONSECUTIVE_HTTP_FAIL` are defined in 05 §5.9 as
+count-within-a-time-window, but this module counts among the most recent N
+persisted `deploy_health_signal` rows (N = CRASH_WINDOW /
+MONITOR_SAMPLE_INTERVAL) instead of re-deriving a true rolling time window
+per sample -- correct as long as sampling stays roughly interval-paced (the
+whole point of a periodic monitor loop); revisit only if sampling cadence
+ever becomes genuinely irregular.
+"""
+
+import json
+import time
+import uuid
+from datetime import datetime, timezone
+
+from supervisor import deploy_health_check, deploy_supervisor
+
+MONITOR_SAMPLE_INTERVAL = 5 * 60
+CANARY_INTERVAL = 60 * 60
+MAX_CRASH_COUNT = 2
+CRASH_WINDOW = 10 * 60
+MAX_CONSECUTIVE_HTTP_FAIL = 3
+VALIDATOR_RATE_DEVIATION_THRESHOLD = 0.30  # fraction; 05 §5.9's "±30 个百分点"
+MONITORING_WINDOW_HOURS = 24
+MONITORING_MIN_TASK_COUNT = 10
+
+
+def _parse_sqlite_timestamp(text):
+    """Same convention as work_node_run.py's own helper of the same name --
+    duplicated (not imported) since it's a 2-line stdlib parse and this
+    module has no other reason to depend on work_node_run.py."""
+    return datetime.strptime(text, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp()
+
+
+def _format_sqlite_timestamp(epoch_seconds):
+    return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+
+def compute_validator_approval_rate(conn, project_id, since_text, until_text=None):
+    """Returns (rate: float in [0,1], sample_size: int), or None if no task
+    reached 'validating' in the window (not enough data to mean anything)."""
+    query = (
+        "SELECT task_id, payload FROM execution_event "
+        "WHERE project_id = ? AND event_type = 'task_status_changed' AND created_at >= ?"
+    )
+    params = [project_id, since_text]
+    if until_text is not None:
+        query += " AND created_at < ?"
+        params.append(until_text)
+    rows = conn.execute(query, params).fetchall()
+
+    task_ids = set()
+    for task_id, payload_json in rows:
+        payload = json.loads(payload_json)
+        if payload.get("to_status") == "validating":
+            task_ids.add(task_id)
+    if not task_ids:
+        return None
+
+    approved = sum(
+        1 for task_id in task_ids
+        if conn.execute(
+            "SELECT 1 FROM bug_record WHERE task_id = ? AND source = 'validator' LIMIT 1", (task_id,)
+        ).fetchone() is None
+    )
+    return approved / len(task_ids), len(task_ids)
+
+
+def compute_baseline_validator_approval_rate(conn, project_id, before_text, lookback_hours=MONITORING_WINDOW_HOURS):
+    """Historical rate over the `lookback_hours` immediately preceding
+    `before_text` (typically `cutover_at`) -- "与该 deployment 之前的历史基线
+    比较" (design doc §6), computed from the same persisted data rather than
+    requiring a separately-stored baseline field. Returns None (same as
+    compute_validator_approval_rate) if there's no prior data -- callers
+    should treat a None baseline as "can't judge deviation yet", not as 0."""
+    since_text = _format_sqlite_timestamp(_parse_sqlite_timestamp(before_text) - lookback_hours * 3600)
+    return compute_validator_approval_rate(conn, project_id, since_text, until_text=before_text)
+
+
+def _record_signal(conn, deployment_id, signal):
+    conn.execute(
+        "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status, detail) "
+        "VALUES (?, ?, CURRENT_TIMESTAMP, ?, ?, ?)",
+        (str(uuid.uuid4()), deployment_id, signal["signal_type"], signal["status"], signal["detail"]),
+    )
+    conn.commit()
+
+
+def _recent_statuses(conn, deployment_id, signal_type, limit):
+    rows = conn.execute(
+        "SELECT status FROM deploy_health_signal WHERE deployment_id = ? AND signal_type = ? "
+        "ORDER BY rowid DESC LIMIT ?",
+        (deployment_id, signal_type, limit),
+    ).fetchall()
+    return [r[0] for r in rows]
+
+
+def _sample_process_alive(pid):
+    alive = not deploy_supervisor.reap_if_exited(pid)
+    return {"signal_type": "process_alive", "status": "ok" if alive else "fail", "detail": None}
+
+
+def _sample_validator_approval_rate(conn, project_id, baseline_rate, window_start_text, threshold):
+    if baseline_rate is None:
+        return {
+            "signal_type": "validator_approval_rate", "status": "ok",
+            "detail": "no baseline yet -- insufficient history to judge deviation",
+        }
+    result = compute_validator_approval_rate(conn, project_id, window_start_text)
+    if result is None:
+        return {
+            "signal_type": "validator_approval_rate", "status": "ok",
+            "detail": "no validated tasks in window yet",
+        }
+    rate, sample_size = result
+    deviation = abs(rate - baseline_rate)
+    status = "fail" if deviation > threshold else "ok"
+    return {
+        "signal_type": "validator_approval_rate", "status": status,
+        "detail": "rate={:.2f} n={} baseline={:.2f} deviation={:.2f}".format(rate, sample_size, baseline_rate, deviation),
+    }
+
+
+def _real_task_count_since(conn, project_id, since_text):
+    return conn.execute(
+        "SELECT COUNT(*) FROM task WHERE project_id = ? AND created_at >= ?", (project_id, since_text)
+    ).fetchone()[0]
+
+
+def run_monitor_cycle(
+    conn, deployment_id, pid, port, project_id, baseline_validator_rate, window_start_text,
+    run_canary=True, crash_window_samples=None, consecutive_http_fail_limit=MAX_CONSECUTIVE_HTTP_FAIL,
+    validator_threshold=VALIDATOR_RATE_DEVIATION_THRESHOLD, min_task_count=MONITORING_MIN_TASK_COUNT,
+    window_hours=MONITORING_WINDOW_HOURS, now=time.time,
+):
+    """One sampling pass. Returns {status, latest_signal}. `window_start_text`
+    is `deployment.cutover_at` (a sqlite TEXT timestamp) -- used both for
+    the elapsed-time-since-cutover math and as the SQL lower bound for
+    "real tasks since the window opened"/"validator judgments since the
+    window opened". Stops sampling further signals the moment one comes
+    back fail (same fail-closed, don't-pile-on-more-checks pattern as
+    `c_deploy_supervisor_health_check`) and transitions
+    `deployment.status='anomaly_detected'` immediately -- 04-P's own state
+    machine has no intermediate "still checking" state distinct from
+    'monitoring' itself.
+    """
+    if crash_window_samples is None:
+        crash_window_samples = max(1, CRASH_WINDOW // MONITOR_SAMPLE_INTERVAL)
+
+    process_signal = _sample_process_alive(pid)
+    _record_signal(conn, deployment_id, process_signal)
+    latest_signal = process_signal
+    anomaly = _recent_statuses(conn, deployment_id, "process_alive", crash_window_samples).count("fail") >= MAX_CRASH_COUNT
+
+    if not anomaly:
+        http_signal = deploy_health_check.check_http_health(port)
+        _record_signal(conn, deployment_id, http_signal)
+        latest_signal = http_signal
+        recent_http = _recent_statuses(conn, deployment_id, "http_health", consecutive_http_fail_limit)
+        anomaly = len(recent_http) >= consecutive_http_fail_limit and all(s == "fail" for s in recent_http)
+
+    if not anomaly and run_canary:
+        canary_signal = deploy_health_check.check_canary_task(port)
+        _record_signal(conn, deployment_id, canary_signal)
+        latest_signal = canary_signal
+        anomaly = canary_signal["status"] == "fail"  # single failure -- no grey area (design doc §6)
+
+    if not anomaly:
+        rate_signal = _sample_validator_approval_rate(
+            conn, project_id, baseline_validator_rate, window_start_text, validator_threshold
+        )
+        _record_signal(conn, deployment_id, rate_signal)
+        latest_signal = rate_signal
+        anomaly = rate_signal["status"] == "fail"
+
+    if anomaly:
+        conn.execute(
+            "UPDATE deployment SET status = 'anomaly_detected', rollback_reason = ? WHERE deployment_id = ?",
+            (json.dumps(latest_signal), deployment_id),
+        )
+        conn.commit()
+        return {"status": "anomaly_detected", "latest_signal": latest_signal}
+
+    elapsed_hours = (now() - _parse_sqlite_timestamp(window_start_text)) / 3600.0
+    real_task_count = _real_task_count_since(conn, project_id, window_start_text)
+    if elapsed_hours >= window_hours or real_task_count >= min_task_count:
+        conn.execute(
+            "UPDATE deployment SET status = 'monitoring_window_clean', monitoring_closed_at = CURRENT_TIMESTAMP "
+            "WHERE deployment_id = ?",
+            (deployment_id,),
+        )
+        conn.commit()
+        return {"status": "monitoring_window_clean", "latest_signal": latest_signal}
+
+    conn.execute("UPDATE deployment SET status = 'monitoring' WHERE deployment_id = ?", (deployment_id,))
+    conn.commit()
+    return {"status": "monitoring", "latest_signal": latest_signal}
+"""U-040 tests: c_deploy_supervisor_monitor. Real HTTP server for http_health/
+canary_task signals (same pattern as test_deploy_health_check.py), real
+sqlite for the validator_approval_rate derivation and window-closure logic.
+
+Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v
+"""
+
+import json
+import os
+import sys
+import threading
+import unittest
+import uuid
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from supervisor import deploy_monitor as dm
+from src import backend_server, schema
+
+
+def _ensure_task(conn, project_id, task_id):
+    if conn.execute("SELECT 1 FROM task WHERE task_id = ?", (task_id,)).fetchone() is None:
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+            "VALUES (?, ?, 'g', 'goal', 'validating', 'validator')",
+            (task_id, project_id),
+        )
+        conn.commit()
+
+
+def _status_event(conn, project_id, task_id, to_status, created_at=None):
+    _ensure_task(conn, project_id, task_id)
+    conn.execute(
+        "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload, created_at) "
+        "VALUES (?, ?, ?, 'task_status_changed', ?, COALESCE(?, CURRENT_TIMESTAMP))",
+        (str(uuid.uuid4()), project_id, task_id, json.dumps({"to_status": to_status}), created_at),
+    )
+    conn.commit()
+
+
+def _validator_bug(conn, task_id):
+    conn.execute(
+        "INSERT INTO bug_record (bug_id, task_id, source, description, root_cause_category, status) "
+        "VALUES (?, ?, 'validator', 'x', 'coder', 'routed')",
+        (str(uuid.uuid4()), task_id),
+    )
+    conn.commit()
+
+
+class ValidatorApprovalRateTestBase(unittest.TestCase):
+    def setUp(self):
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        self.conn.commit()
+        self.addCleanup(self.conn.close)
+
+
+class ComputeValidatorApprovalRateTest(ValidatorApprovalRateTestBase):
+    def test_no_validating_events_returns_none(self):
+        self.assertIsNone(dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00"))
+
+    def test_all_approved_gives_rate_1(self):
+        _status_event(self.conn, "proj-1", "t1", "validating")
+        _status_event(self.conn, "proj-1", "t2", "validating")
+        rate, n = dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00")
+        self.assertEqual((rate, n), (1.0, 2))
+
+    def test_bug_record_marks_task_as_not_approved(self):
+        _status_event(self.conn, "proj-1", "t1", "validating")
+        _status_event(self.conn, "proj-1", "t2", "validating")
+        _validator_bug(self.conn, "t1")
+        rate, n = dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00")
+        self.assertEqual((rate, n), (0.5, 2))
+
+    def test_non_validator_bug_record_does_not_count_against_approval(self):
+        _status_event(self.conn, "proj-1", "t1", "validating")
+        self.conn.execute(
+            "INSERT INTO bug_record (bug_id, task_id, source, description, status) "
+            "VALUES ('b1', 't1', 'human', 'x', 'open')"
+        )
+        self.conn.commit()
+        rate, n = dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00")
+        self.assertEqual((rate, n), (1.0, 1))
+
+    def test_events_before_since_are_excluded(self):
+        _status_event(self.conn, "proj-1", "t1", "validating", created_at="2000-01-01 00:00:00")
+        result = dm.compute_validator_approval_rate(self.conn, "proj-1", "2020-01-01 00:00:00")
+        self.assertIsNone(result)
+
+
+class ComputeBaselineTest(ValidatorApprovalRateTestBase):
+    def test_baseline_looks_back_from_before_text(self):
+        _status_event(self.conn, "proj-1", "t-old", "validating", created_at="2026-01-01 00:00:00")
+        _status_event(self.conn, "proj-1", "t-too-old", "validating", created_at="2020-01-01 00:00:00")
+        rate, n = dm.compute_baseline_validator_approval_rate(
+            self.conn, "proj-1", "2026-01-02 00:00:00", lookback_hours=24
+        )
+        self.assertEqual(n, 1)  # only t-old falls in the 24h lookback window
+
+
+class RealServerTestBase(unittest.TestCase):
+    def setUp(self):
+        self.server = backend_server.create_server(db_path=":memory:")
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+
+
+class RunMonitorCycleTest(RealServerTestBase):
+    def setUp(self):
+        super().setUp()
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        self.conn.execute(
+            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
+            "previous_instance_dir, cutover_at) "
+            "VALUES ('dep-1', 'c1', 'c0', 'monitoring', '/prev', '2020-01-01 00:00:00')"
+        )
+        self.conn.commit()
+        self.addCleanup(self.conn.close)
+
+    def test_all_clear_and_window_not_yet_closed_stays_monitoring(self):
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", os.getpid(), self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2099-01-01 00:00:00",  # future -> 0 elapsed hours
+            now=lambda: dm._parse_sqlite_timestamp("2099-01-01 00:00:00") + 60,
+        )
+        self.assertEqual(result["status"], "monitoring")
+        row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone()
+        self.assertEqual(row[0], "monitoring")
+
+    def test_window_closes_via_elapsed_hours(self):
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", os.getpid(), self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00",
+            now=lambda: dm._parse_sqlite_timestamp("2020-01-01 00:00:00") + 25 * 3600,
+            window_hours=24,
+        )
+        self.assertEqual(result["status"], "monitoring_window_clean")
+        row = self.conn.execute(
+            "SELECT status, monitoring_closed_at FROM deployment WHERE deployment_id = 'dep-1'"
+        ).fetchone()
+        self.assertEqual(row[0], "monitoring_window_clean")
+        self.assertIsNotNone(row[1])
+
+    def test_window_closes_via_real_task_count(self):
+        for i in range(10):
+            self.conn.execute(
+                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+                "VALUES (?, 'proj-1', 'g', 'goal', 'queued', 'planner')", ("t{}".format(i),)
+            )
+        self.conn.commit()
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", os.getpid(), self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00",
+            now=lambda: dm._parse_sqlite_timestamp("2020-01-01 00:00:00") + 60,
+            window_hours=999999, min_task_count=10,
+        )
+        self.assertEqual(result["status"], "monitoring_window_clean")
+
+    def test_single_dead_process_reading_alone_does_not_yet_anomaly(self):
+        # design doc §6: process_alive's anomaly condition is MAX_CRASH_COUNT
+        # crashes within CRASH_WINDOW, not a single dead reading -- a lone
+        # blip records the signal but must NOT immediately roll back.
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", 2**30, self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00",
+            window_hours=999999,
+        )
+        self.assertNotEqual(result["status"], "anomaly_detected")
+        signal_row = self.conn.execute(
+            "SELECT status FROM deploy_health_signal WHERE deployment_id = 'dep-1' AND signal_type = 'process_alive'"
+        ).fetchone()
+        self.assertEqual(signal_row[0], "fail")
+
+    def test_repeated_process_crashes_within_window_trigger_anomaly(self):
+        # Pre-seed one prior process_alive fail -- this cycle's fail makes 2,
+        # meeting MAX_CRASH_COUNT=2 even though the process IS alive right now.
+        self.conn.execute(
+            "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
+            "VALUES (?, 'dep-1', CURRENT_TIMESTAMP, 'process_alive', 'fail')", (str(uuid.uuid4()),)
+        )
+        self.conn.commit()
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", 2**30, self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00",
+            crash_window_samples=5,
+        )
+        self.assertEqual(result["status"], "anomaly_detected")
+
+    def test_consecutive_http_failures_trigger_anomaly(self):
+        for _ in range(3):
+            self.conn.execute(
+                "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
+                "VALUES (?, 'dep-1', CURRENT_TIMESTAMP, 'http_health', 'fail')", (str(uuid.uuid4()),)
+            )
+        self.conn.commit()
+        self.server.conn.close()  # this cycle's own http_health check also fails
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", os.getpid(), self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00",
+            consecutive_http_fail_limit=4,
+        )
+        self.assertEqual(result["status"], "anomaly_detected")
+        self.assertEqual(result["latest_signal"]["signal_type"], "http_health")
+
+    def test_validator_rate_deviation_triggers_anomaly(self):
+        for i in range(4):
+            _status_event(self.conn, "proj-1", "t{}".format(i), "validating")
+            _validator_bug(self.conn, "t{}".format(i))  # 0% approval this window
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", os.getpid(), self.port, "proj-1",
+            baseline_validator_rate=0.9, window_start_text="2020-01-01 00:00:00",
+        )
+        self.assertEqual(result["status"], "anomaly_detected")
+        self.assertEqual(result["latest_signal"]["signal_type"], "validator_approval_rate")
+
+    def test_no_baseline_yet_never_flags_validator_anomaly(self):
+        for i in range(4):
+            _status_event(self.conn, "proj-1", "t{}".format(i), "validating")
+            _validator_bug(self.conn, "t{}".format(i))
+        result = dm.run_monitor_cycle(
+            self.conn, "dep-1", os.getpid(), self.port, "proj-1",
+            baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00",
+        )
+        self.assertNotEqual(result["status"], "anomaly_detected")
+
+
+if __name__ == "__main__":
+    unittest.main()