"""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): """Uses `reap_if_exited` (not plain `is_alive`) so a candidate that crashed immediately after being spawned by THIS process reads as dead rather than a false "alive" zombie reading -- see `deploy_supervisor.is_alive`'s own docstring for why the distinction matters here specifically (this module's caller is always the same process that Popen'd the candidate).""" deadline = now() + timeout while now() < deadline: if not deploy_supervisor.reap_if_exited(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}