diff --git a/docs/development-plan.md b/docs/development-plan.md index 0c67d7b..aa79dd2 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -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 diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 98ede73..5ee7eb0 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -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 状态机所处状态与最近一次健康信号采样结果"} diff --git a/supervisor/deploy_monitor.py b/supervisor/deploy_monitor.py new file mode 100644 index 0000000..a1999ea --- /dev/null +++ b/supervisor/deploy_monitor.py @@ -0,0 +1,225 @@ +"""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} diff --git a/supervisor/test_deploy_monitor.py b/supervisor/test_deploy_monitor.py new file mode 100644 index 0000000..26ce31c --- /dev/null +++ b/supervisor/test_deploy_monitor.py @@ -0,0 +1,243 @@ +"""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()