U-038: c_deploy_supervisor_health_check (process_alive/http_health/canary_task)

Adds supervisor/deploy_health_check.py: run_health_check() runs the three pre-cutover checks in order, stopping at the first failure (03-C: "任一检查项 fail 立即整体判定 health_check_failed，不等待其余检查项跑完") and persisting a deploy_health_signal row for every check that actually ran. - process_alive: polls deploy_supervisor.is_alive() up to HEALTH_CHECK_STARTUP_TIMEOUT. - http_health: hits the candidate's /api/health -- added to backend_server.py (U-037 deliberately left this for U-038). Not just "process exists" -- runs a real SELECT 1 to catch a wedged sqlite connection a live-but-stuck process would otherwise pass. - canary_task: a real (not mocked) project+task creation round-trip against the candidate over HTTP -- no real Org Graph/claude calls, since c_task_create's synchronous 'queued' landing is already a deterministic, always-should-pass result, which is what a canary needs. Timeouts are the MVP-fixed drafts from 05 §5.9, explicitly flagged there as needing real calibration later, not hardcoded as authoritative. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-038: c_deploy_supervisor_health_check (process_alive/http_health/canary_task)
Adds supervisor/deploy_health_check.py: run_health_check() runs the three pre-cutover checks in order, stopping at the first failure (03-C: "任一检查项 fail 立即整体判定 health_check_failed，不等待其余检查项跑完") and persisting a deploy_health_signal row for every check that actually ran. - process_alive: polls deploy_supervisor.is_alive() up to HEALTH_CHECK_STARTUP_TIMEOUT. - http_health: hits the candidate's /api/health -- added to backend_server.py (U-037 deliberately left this for U-038). Not just "process exists" -- runs a real SELECT 1 to catch a wedged sqlite connection a live-but-stuck process would otherwise pass. - canary_task: a real (not mocked) project+task creation round-trip against the candidate over HTTP -- no real Org Graph/claude calls, since c_task_create's synchronous 'queued' landing is already a deterministic, always-should-pass result, which is what a canary needs. Timeouts are the MVP-fixed drafts from 05 §5.9, explicitly flagged there as needing real calibration later, not hardcoded as authoritative. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 5a046a4f
Showing 6 changed files with 369 additions and 1 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/backend_server.py
src/test_backend_server.py
supervisor/deploy_health_check.py
supervisor/test_deploy_health_check.py
@@ -368,6 +368,8 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-038 [Risk Gate]** `c_deploy_supervisor_health_check`：`process_alive`/`http_health`/`canary_task` 三项 pre-cutover 检查
   - 依赖：U-037
   - Done：三项检查全部通过 → `status=health_check_passed`；任一失败/超时 → `status=health_check_failed`，Live 全程未受影响（对应回滚路径 A）
+  - **实现补充（2026-08-02）**：新增 `supervisor/deploy_health_check.py`。`process_alive` 复用 U-036 `deploy_supervisor.is_alive()`（`os.kill(pid,0)`），按 `HEALTH_CHECK_STARTUP_TIMEOUT` 轮询；`http_health` 请求候选的 `/api/health`——**这个端点 U-037 当时特意留空，这里补上**：`backend_server.py` 新增 `GET /api/health`，不只是"进程活着就 200"，还真的执行一次 `SELECT 1` 确认 sqlite 连接本身没卡死（连接卡死但进程还活着的场景，只测 `process_alive` 会漏检，这正是"health check"要抓的那类问题）；`canary_task` 是design doc §6"结果可预期的最小任务"字面实现——对候选发起一次真实的"建临时 project（临时目录 `git init` + 一个空提交）+ 建一个 goal 任务 + 读回确认 `status=queued`/`goal_text` 匹配"，不跑真实 Org Graph（不产生真实 `claude` 调用/花费），因为 `c_task_create` 本身落 `queued` 就是同步、确定性、"已知会通过"的最小验证，够用。三项检查按 03-C 原文"任一 fail 立即整体判定，不等其余项跑完"实现，`run_health_check()` 用 tuple-of-lambda 顺序短路，每项检查结果都真实写一条 `deploy_health_signal`（哪怕后续项因为短路没跑，已跑的那几条也留痕）。
+  - 覆盖：03-C `c_deploy_supervisor_health_check`；`backend_server.py` 新增 `GET /api/health`（U-037 留空的口子，这里补上）；01-M `deploy_health_signal`（首次被真正写入）
 - **U-039 [Risk Gate]** `c_deploy_supervisor_cutover`：蓝绿切换本体（Live 停止接受新触发 → 候选切换到生产端口/生产库 → 旧 Live checkout 保留为 `previous_instance_dir`）
   - 依赖：U-038
@@ -55,7 +55,7 @@
     {"id": "U-035", "phase": "PH5", "sub_phase": "PH5-S1", "name": "self-target worktree 无条件强制 + core_engine 分类审计（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-013", "U-016", "U-034"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-03"]}, "done": "self-target 时无条件走 worktree（work_graph_orchestrate.run_graph 内强制）；core_engine 分类（src/core_engine_classify.py，路径规则）在 self-target 任务 PASS 收尾时记一条审计用 bug_record；且改动落地后重跑 U-013/REQ-03（worktree 隔离）现有回归用例全部通过。**2026-08-02 用户决策修正**：Deploy Supervisor 触发条件按 supervisor-auto-rollback-design.md §1 字面设计（self-target+PASS 即触发，bug_record 是旁路产物不是前提），而非 03-C/04-P 字面文本要求的\"先有 core_engine bug_record\"——后者会让从不产生 bug_record 的 goal 型自我改进任务永远触发不了 Deploy Supervisor，已用 AskUserQuestion 与用户确认此订正。c_org_loop_route 未改动（fast_lane/full_replan 是 FAIL 路径重试路由，与此处 PASS 路径的蓝绿触发正交），deployment 行创建仍是 U-037 的范围，不在本 Unit 提前做"},
     {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架（见 development-plan.md 2026-08-02 实现补充）", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py（供 python3 -m unittest discover -s supervisor/isolation 运行，52 测试全绿）；Deploy Supervisor 骨架（supervisor/deploy_supervisor.py）独立进程可启停（真实子进程验证，非 mock），不受 Live/Candidate 崩溃影响；本 Unit 只做进程生命周期骨架，不含任何真实部署决策逻辑（U-037+ 的 poll_fn 挂载点）"},
     {"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", "classification": "Risk Gate", "depends_on": ["U-037"], "coverage": {"model": [], "view": [], "controller": ["c_deploy_supervisor_health_check"], "process": [], "requirement": []}, "done": "process_alive/http_health/canary_task 三项检查全过 -> health_check_passed；任一失败 -> health_check_failed 且 Live 未受影响"},
+    {"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", "classification": "Risk Gate", "depends_on": ["U-038"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_cutover"], "process": [], "requirement": []}, "done": "健康检查通过后完成真实切换；在途 work_node_execution 重启后可恢复；previous_instance_dir 保留至监控窗口结束"},
     {"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-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 冲突有明确处理"},
@@ -132,6 +132,9 @@ _SKILLS_PREFIX = &quot;/api/skills/&quot;
 class _Handler(BaseHTTPRequestHandler):
     def do_GET(self):
         parsed = urllib.parse.urlsplit(self.path)
+        if parsed.path == "/api/health":
+            self._handle_health()
+            return
         if parsed.path.startswith(_WS_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
             node_exec_id = parsed.path[len(_WS_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
             self._handle_ws_log_tail(node_exec_id)
@@ -235,6 +238,22 @@ class _Handler(BaseHTTPRequestHandler):
         except json.JSONDecodeError:
             return None, {"error": "INVALID_INPUT", "detail": "body is not valid JSON"}
+    def _handle_health(self):
+        """c_deploy_supervisor_health_check's `http_health` probe (U-038,
+        03-C §3.15/§6). Confirms both "the HTTP server thread answers" and
+        "the sqlite connection is actually usable" -- a process that's alive
+        but whose DB connection got wedged somehow would otherwise pass a
+        process_alive check and fail every real request afterward; catching
+        that here, not just accepting a 200 for existing, is the point of a
+        health check in the first place."""
+        try:
+            with self.server.conn_lock:
+                self.server.conn.execute("SELECT 1").fetchone()
+        except sqlite3.Error as e:
+            self._send_json(503, {"status": "error", "detail": str(e)})
+            return
+        self._send_json(200, {"status": "ok"})
+
     def _handle_project_create(self):
         payload, error = self._read_json_body()
         if error:
@@ -117,6 +117,40 @@ class StaticFileServingTest(unittest.TestCase):
         self.assertEqual(json.loads(body)["error"], "NOT_FOUND")
+class HealthRouteTest(unittest.TestCase):
+    """U-038: c_deploy_supervisor_health_check's http_health probe target."""
+
+    def setUp(self):
+        self.server = backend_server.create_server(db_path=":memory:")
+        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)
+
+    def _get(self, path):
+        host, port = self.server.server_address
+        conn = http.client.HTTPConnection(host, port, timeout=5)
+        conn.request("GET", path)
+        resp = conn.getresponse()
+        body = resp.read()
+        conn.close()
+        return resp.status, body
+
+    def test_returns_200_ok_when_db_connection_is_usable(self):
+        status, body = self._get("/api/health")
+        self.assertEqual(status, 200)
+        self.assertEqual(json.loads(body), {"status": "ok"})
+
+    def test_returns_503_when_db_connection_is_broken(self):
+        self.server.conn.close()  # simulate a wedged/closed connection
+        status, body = self._get("/api/health")
+        self.assertEqual(status, 503)
+        self.assertEqual(json.loads(body)["status"], "error")
+
+
 class PickFolderTest(unittest.TestCase):
     def test_success_returns_selected_path(self):
         with patch("src.backend_server.platform.system", return_value="Darwin"), \
+"""c_deploy_supervisor_health_check (03-C §3.15 / 04-P §4.6 / design doc §6):
+three pre-cutover checks against a staged candidate -- `process_alive`,
+`http_health`, `canary_task`. All confirmed fail-closed and non-retrying per
+03-C's own text: "任一检查项 fail 立即整体判定 health_check_failed，不等待
+其余检查项跑完" and "不重试——重试属于'给坏版本第二次机会'". Timeouts are the
+MVP-fixed draft values from 05 §5.9 (`HEALTH_CHECK_STARTUP_TIMEOUT`=30s,
+`HEALTH_CHECK_HTTP_TIMEOUT`=10s, `HEALTH_CHECK_CANARY_TIMEOUT`=120s) --
+05 §5.9's own text already flags these as engineering drafts needing
+real-world calibration, not authoritative constants to bake into assertions
+elsewhere.
+"""
+
+import json
+import shutil
+import subprocess
+import tempfile
+import time
+import urllib.error
+import urllib.request
+import uuid
+
+from supervisor import deploy_supervisor
+
+HEALTH_CHECK_STARTUP_TIMEOUT = 30
+HEALTH_CHECK_HTTP_TIMEOUT = 10
+HEALTH_CHECK_CANARY_TIMEOUT = 120
+
+_SIGNAL_TYPES = ("process_alive", "http_health", "canary_task")
+
+
+def check_process_alive(pid, timeout=HEALTH_CHECK_STARTUP_TIMEOUT, sleep=time.sleep, now=time.time):
+    deadline = now() + timeout
+    while now() < deadline:
+        if deploy_supervisor.is_alive(pid):
+            return {"signal_type": "process_alive", "status": "ok", "detail": None}
+        sleep(0.5)
+    return {
+        "signal_type": "process_alive", "status": "fail",
+        "detail": "pid {} not alive within {}s".format(pid, timeout),
+    }
+
+
+def check_http_health(port, timeout=HEALTH_CHECK_HTTP_TIMEOUT):
+    try:
+        with urllib.request.urlopen("http://127.0.0.1:{}/api/health".format(port), timeout=timeout) as resp:
+            if resp.status == 200:
+                return {"signal_type": "http_health", "status": "ok", "detail": None}
+            return {
+                "signal_type": "http_health", "status": "fail",
+                "detail": "unexpected HTTP status {}".format(resp.status),
+            }
+    except urllib.error.HTTPError as e:
+        # HTTPError IS the response object (a non-2xx status raises rather
+        # than returning) -- close it explicitly, `except` doesn't run our
+        # `with` block's __exit__ for us.
+        detail = "unexpected HTTP status {}".format(e.code)
+        e.close()
+        return {"signal_type": "http_health", "status": "fail", "detail": detail}
+    except (urllib.error.URLError, OSError) as e:
+        return {"signal_type": "http_health", "status": "fail", "detail": str(e)}
+
+
+def _post_json(port, path, payload, timeout):
+    body = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        "http://127.0.0.1:{}{}".format(port, path), data=body,
+        headers={"Content-Type": "application/json"}, method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return resp.status, json.loads(resp.read())
+
+
+def _get_json(port, path, timeout):
+    with urllib.request.urlopen("http://127.0.0.1:{}{}".format(port, path), timeout=timeout) as resp:
+        return resp.status, json.loads(resp.read())
+
+
+def check_canary_task(port, timeout=HEALTH_CHECK_CANARY_TIMEOUT):
+    """"结果可预期的最小任务" (design doc §6): a trivial project + task
+    creation round-trip. Doesn't run a real Org Graph (no real `claude`
+    calls, no real cost/latency) -- `c_task_create` itself lands
+    `status='queued'` synchronously, which is exactly the kind of
+    deterministic, always-should-pass result a canary needs; the point is
+    proving the candidate's HTTP+DB write path genuinely works, not
+    exercising the full pipeline."""
+    canary_dir = tempfile.mkdtemp(prefix="deploy-canary-")
+    try:
+        subprocess.run(["git", "init", "-q"], cwd=canary_dir, check=True, capture_output=True)
+        subprocess.run(
+            ["git", "-c", "user.name=graphlemon", "-c", "user.email=graphlemon@localhost",
+             "commit", "--allow-empty", "-q", "-m", "canary"],
+            cwd=canary_dir, check=True, capture_output=True,
+        )
+
+        status, project = _post_json(
+            port, "/api/projects", {"target_folder_path": canary_dir, "name": "deploy-canary"}, timeout
+        )
+        if status != 200:
+            return {"signal_type": "canary_task", "status": "fail", "detail": "project create: HTTP {}".format(status)}
+
+        goal_text = "deploy-supervisor canary " + uuid.uuid4().hex
+        status, task = _post_json(
+            port, "/api/tasks", {"project_id": project["project_id"], "goal_text": goal_text}, timeout
+        )
+        if status != 200:
+            return {"signal_type": "canary_task", "status": "fail", "detail": "task create: HTTP {}".format(status)}
+
+        status, fetched = _get_json(port, "/api/tasks/" + task["task_id"], timeout)
+        if status != 200 or fetched.get("status") != "queued" or fetched.get("goal_text") != goal_text:
+            return {
+                "signal_type": "canary_task", "status": "fail",
+                "detail": "unexpected task state after create: HTTP {} {!r}".format(status, fetched),
+            }
+        return {"signal_type": "canary_task", "status": "ok", "detail": None}
+    except urllib.error.HTTPError as e:
+        detail = "unexpected HTTP status {}".format(e.code)
+        e.close()
+        return {"signal_type": "canary_task", "status": "fail", "detail": detail}
+    except (urllib.error.URLError, OSError, subprocess.CalledProcessError, KeyError) as e:
+        return {"signal_type": "canary_task", "status": "fail", "detail": str(e)}
+    finally:
+        shutil.rmtree(canary_dir, ignore_errors=True)
+
+
+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 run_health_check(
+    conn, deployment_id, pid, port,
+    startup_timeout=HEALTH_CHECK_STARTUP_TIMEOUT, http_timeout=HEALTH_CHECK_HTTP_TIMEOUT,
+    canary_timeout=HEALTH_CHECK_CANARY_TIMEOUT, sleep=time.sleep, now=time.time,
+):
+    """c_deploy_supervisor_health_check. Returns {status, signals}. Marks
+    `deployment.status='health_check_running'` on entry, then
+    'health_check_passed'/'health_check_failed' once done -- fails closed
+    and stops at the FIRST failing check (no point running http_health/
+    canary_task against a process that isn't even alive yet)."""
+    conn.execute("UPDATE deployment SET status = 'health_check_running' WHERE deployment_id = ?", (deployment_id,))
+    conn.commit()
+
+    signals = []
+    checks = (
+        lambda: check_process_alive(pid, timeout=startup_timeout, sleep=sleep, now=now),
+        lambda: check_http_health(port, timeout=http_timeout),
+        lambda: check_canary_task(port, timeout=canary_timeout),
+    )
+    for check in checks:
+        signal = check()
+        _record_signal(conn, deployment_id, signal)
+        signals.append(signal)
+        if signal["status"] != "ok":
+            conn.execute(
+                "UPDATE deployment SET status = 'health_check_failed' WHERE deployment_id = ?", (deployment_id,)
+            )
+            conn.commit()
+            return {"status": "health_check_failed", "signals": signals}
+
+    conn.execute("UPDATE deployment SET status = 'health_check_passed' WHERE deployment_id = ?", (deployment_id,))
+    conn.commit()
+    return {"status": "health_check_passed", "signals": signals}
+"""U-038 tests: c_deploy_supervisor_health_check. Real HTTP server (in-process
+thread, same pattern src/test_backend_server.py already uses), real project/
+task creation round trips for the canary check -- nothing here is mocked
+except process liveness timing (injectable sleep/now, same convention as
+work_node_run's own quota-wait tests).
+
+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__))))
+
+import threading
+
+from supervisor import deploy_health_check as dhc
+from src import backend_server, schema
+
+
+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 CheckProcessAliveTest(unittest.TestCase):
+    def test_alive_process_returns_ok_immediately(self):
+        signal = dhc.check_process_alive(os.getpid())
+        self.assertEqual(signal, {"signal_type": "process_alive", "status": "ok", "detail": None})
+
+    def test_dead_pid_times_out_and_fails(self):
+        clock = {"t": 0.0}
+        sleeps = []
+        signal = dhc.check_process_alive(
+            2**30, timeout=5, sleep=lambda s: (sleeps.append(s), clock.__setitem__("t", clock["t"] + s)),
+            now=lambda: clock["t"],
+        )
+        self.assertEqual(signal["signal_type"], "process_alive")
+        self.assertEqual(signal["status"], "fail")
+        self.assertTrue(sleeps)
+
+    def test_process_that_becomes_alive_mid_poll_still_passes(self):
+        calls = {"n": 0}
+
+        def fake_is_alive(pid):
+            calls["n"] += 1
+            return calls["n"] >= 3
+
+        with patch("supervisor.deploy_supervisor.is_alive", side_effect=fake_is_alive):
+            signal = dhc.check_process_alive(99999, timeout=5, sleep=lambda s: None, now=lambda: 0.0)
+        self.assertEqual(signal["status"], "ok")
+        self.assertEqual(calls["n"], 3)
+
+
+class CheckHttpHealthTest(RealServerTestBase):
+    def test_healthy_server_returns_ok(self):
+        signal = dhc.check_http_health(self.port)
+        self.assertEqual(signal, {"signal_type": "http_health", "status": "ok", "detail": None})
+
+    def test_broken_connection_returns_fail(self):
+        self.server.conn.close()
+        signal = dhc.check_http_health(self.port)
+        self.assertEqual(signal["signal_type"], "http_health")
+        self.assertEqual(signal["status"], "fail")
+
+    def test_nothing_listening_on_port_returns_fail(self):
+        signal = dhc.check_http_health(1)  # privileged/unbound port, nothing listening
+        self.assertEqual(signal["status"], "fail")
+
+
+class CheckCanaryTaskTest(RealServerTestBase):
+    def test_real_project_and_task_round_trip_passes(self):
+        signal = dhc.check_canary_task(self.port)
+        self.assertEqual(signal, {"signal_type": "canary_task", "status": "ok", "detail": None})
+
+        conn = self.server.conn
+        count = conn.execute("SELECT COUNT(*) FROM task WHERE goal_text LIKE 'deploy-supervisor canary%'").fetchone()[0]
+        self.assertEqual(count, 1)
+
+    def test_nothing_listening_returns_fail(self):
+        signal = dhc.check_canary_task(1)
+        self.assertEqual(signal["signal_type"], "canary_task")
+        self.assertEqual(signal["status"], "fail")
+
+
+class RunHealthCheckTest(RealServerTestBase):
+    def setUp(self):
+        super().setUp()
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.conn.execute(
+            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
+            "previous_instance_dir) VALUES ('dep-1', 'c1', 'c0', 'candidate_starting', '/prev')"
+        )
+        self.conn.commit()
+        self.addCleanup(self.conn.close)
+
+    def test_all_checks_pass_marks_health_check_passed(self):
+        result = dhc.run_health_check(self.conn, "dep-1", os.getpid(), self.port)
+        self.assertEqual(result["status"], "health_check_passed")
+        self.assertEqual([s["signal_type"] for s in result["signals"]], ["process_alive", "http_health", "canary_task"])
+        self.assertTrue(all(s["status"] == "ok" for s in result["signals"]))
+
+        row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone()
+        self.assertEqual(row[0], "health_check_passed")
+
+        signal_count = self.conn.execute(
+            "SELECT COUNT(*) FROM deploy_health_signal WHERE deployment_id = 'dep-1'"
+        ).fetchone()[0]
+        self.assertEqual(signal_count, 3)
+
+    def test_process_alive_failure_stops_before_http_and_canary_checks(self):
+        result = dhc.run_health_check(
+            self.conn, "dep-1", 2**30, self.port,
+            startup_timeout=0, sleep=lambda s: None, now=lambda: 0.0,
+        )
+        self.assertEqual(result["status"], "health_check_failed")
+        self.assertEqual(len(result["signals"]), 1)
+        self.assertEqual(result["signals"][0]["signal_type"], "process_alive")
+
+        row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone()
+        self.assertEqual(row[0], "health_check_failed")
+
+        signal_count = self.conn.execute(
+            "SELECT COUNT(*) FROM deploy_health_signal WHERE deployment_id = 'dep-1'"
+        ).fetchone()[0]
+        self.assertEqual(signal_count, 1)  # http_health/canary_task never ran
+
+    def test_http_health_failure_stops_before_canary_check(self):
+        self.server.conn.close()
+        result = dhc.run_health_check(self.conn, "dep-1", os.getpid(), self.port)
+        self.assertEqual(result["status"], "health_check_failed")
+        self.assertEqual([s["signal_type"] for s in result["signals"]], ["process_alive", "http_health"])
+
+
+if __name__ == "__main__":
+    unittest.main()