docs: 网关摘除隔离期实施方案（供网关机器参考实现，暂不在本仓库改 gateway 代码）

复盘 9·19 磁盘满事故网关未切换的三层原因；给出 main 分支 gateway/pool.py
的隔离期改动说明（WorkerState.offline_until + 被动摘除冷却 60s + 探测复活
守卫 + 删死代码）、配置项、单测建议、部署/验证/回滚步骤。worker 侧深检
（方案 A）已在本分支 14562d4 上线。

app.py

@@ -2,15 +2,20 @@
 
 接口 1（四庭七眼测量）已为**真实算法实现**（见 face_analysis 包）；接口 2~5 仍为 Mock。
 拆分架构：worker 跑算法、返回 `annotated_image_base64`（不落盘、不拼 URL，由网关完成）。
-worker 对 `/api/*` 校验内网鉴权头 `X-Internal-Token`，`/health` 供网关探测不校验。
+worker 对 `/api/*` 校验内网鉴权头 `X-Internal-Token`，`/health` 供网关探测不校验
+（深检：磁盘水位 + 下游依赖 TCP 探测，异常返回 503 degraded，见 _load_health_cfg）。
 """
+import asyncio
 import base64
 import json
 import logging
 import os
+import shutil
+import time
 from contextlib import asynccontextmanager
 from io import BytesIO
 from typing import Any, List, Optional
+from urllib.parse import urlsplit
 
 import cv2
 import numpy as np
@@ -1911,12 +1916,127 @@ async def download_hairline_log(rid: Optional[str] = None, tail: int = 500):
 # 健康检查
 # ---------------------------------------------------------------------------
 
+def _load_health_cfg() -> dict:
+    """深健康检查配置：环境变量优先，回退 worker_config.json 的 "health" 段。
+
+    min_free_gb=0 关闭磁盘检查；check_deps=false 关闭依赖探测（均退化为旧行为，
+    即只看模型就绪标志）。
+    """
+    cfg_path = os.path.join(os.path.dirname(__file__), "worker_config.json")
+    file_cfg: dict = {}
+    if os.path.isfile(cfg_path):
+        try:
+            with open(cfg_path, encoding="utf-8") as f:
+                file_cfg = json.load(f).get("health") or {}
+        except Exception as e:  # noqa: BLE001
+            logger.warning("读取 worker_config.json health 段失败：%s", e)
+
+    def _get(env_key, file_key, default):
+        raw = os.getenv(env_key)
+        return raw if raw not in (None, "") else file_cfg.get(file_key, default)
+
+    def _as_float(value, default):
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return default
+
+    return {
+        "min_free_gb": _as_float(_get("HEALTH_MIN_FREE_GB", "min_free_gb", 5.0), 5.0),
+        "check_deps": str(_get("HEALTH_CHECK_DEPS", "check_deps", True)).strip().lower()
+                      not in ("0", "false", "no", "off"),
+        "dep_comfyui": str(_get("HEALTH_DEP_COMFYUI", "dep_comfyui", "127.0.0.1:8188")),
+        "deps_cache_seconds": max(0.0, _as_float(
+            _get("HEALTH_DEPS_CACHE_SECONDS", "deps_cache_seconds", 5.0), 5.0)),
+    }
+
+
+_HEALTH_CFG = _load_health_cfg()
+
+
+def _disk_free_gb() -> float:
+    """worker 所在文件系统的剩余空间（GB）。"""
+    st = shutil.disk_usage(os.path.dirname(os.path.abspath(__file__)))
+    return st.free / (1 << 30)
+
+
+def _parse_host_port(addr: str, default_port: int):
+    """解析 "host:port"（仅支持 IPv4/主机名形式）。"""
+    host, _, port = str(addr).partition(":")
+    try:
+        return host or "127.0.0.1", int(port) if port else default_port
+    except ValueError:
+        return host or "127.0.0.1", default_port
+
+
+def _change_hair_addr():
+    """从 _CHANGE_HAIR_BASE 解析 (host, port)，失败返回 None。"""
+    try:
+        u = urlsplit(_CHANGE_HAIR_BASE)
+        if u.hostname:
+            return u.hostname, u.port or 80
+    except ValueError:
+        pass
+    return None
+
+
+async def _tcp_probe(host: str, port: int, timeout: float = 0.8) -> bool:
+    """TCP 连通性探测：只判下游进程死活，不过 HTTP 认证、不受业务繁忙影响。"""
+    try:
+        _, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout)
+        writer.close()
+        return True
+    except Exception:  # noqa: BLE001
+        return False
+
+
+_DEP_CACHE = {"ts": -1.0, "detail": None}
+
+
+async def _check_deps():
+    """并发探测下游依赖（ComfyUI、change_hair），结果缓存 deps_cache_seconds 秒。"""
+    now = time.monotonic()
+    detail = _DEP_CACHE["detail"]
+    if detail is None or now - _DEP_CACHE["ts"] >= _HEALTH_CFG["deps_cache_seconds"]:
+        probes = {"comfyui": _tcp_probe(*_parse_host_port(_HEALTH_CFG["dep_comfyui"], 8188))}
+        ch = _change_hair_addr()
+        if ch:
+            probes["change_hair"] = _tcp_probe(*ch)
+        names = list(probes)
+        results = await asyncio.gather(*probes.values())
+        detail = dict(zip(names, results))
+        _DEP_CACHE["ts"], _DEP_CACHE["detail"] = now, detail
+    return all(detail.values()), detail
+
+
 @app.get("/health", include_in_schema=False)
 async def health():
-    # 模型就绪后才返回 200，供网关探测；未就绪返回 503。
+    """网关探测入口。
+
+    模型未就绪 → 503 starting；磁盘剩余低于阈值 / 下游依赖 TCP 不通 → 503 degraded；
+    正常 → 200（附 disk_free_gb 与 deps 明细，供人工排查）。深检自身异常时按就绪返回
+    （健康检查不能自己成为故障点）。
+    """
     if not _STATE["ready"]:
         return JSONResponse(status_code=503, content={"status": "starting"})
-    return {"status": "ok"}
+    try:
+        free_gb = _disk_free_gb()
+        body = {"status": "ok", "disk_free_gb": round(free_gb, 1)}
+        min_free = _HEALTH_CFG["min_free_gb"]
+        if min_free > 0 and free_gb < min_free:
+            body.update(status="degraded", reason=f"disk_free_below_{min_free}g")
+            return JSONResponse(status_code=503, content=body)
+        if _HEALTH_CFG["check_deps"]:
+            deps_ok, detail = await _check_deps()
+            body["deps"] = detail
+            if not deps_ok:
+                down = ",".join(k for k, v in detail.items() if not v)
+                body.update(status="degraded", reason=f"dep_down:{down}")
+                return JSONResponse(status_code=503, content=body)
+        return body
+    except Exception:  # noqa: BLE001
+        logger.exception("深健康检查异常，按就绪返回")
+        return {"status": "ok"}
 
 
 @app.get("/", include_in_schema=False)

docs/网关摘除隔离期实施方案.md

@@ -0,0 +1,234 @@
+# 网关摘除隔离期实施方案（方案 B）
+
+> 目标读者：网关机器的部署/维护者。本文档给出基于 **main 分支** `gateway/` 代码的
+> 完整改动说明，可直接照做。**不需要改 worker 侧任何代码。**
+
+## 1. 背景与事故复盘
+
+2026-09-19 晚，GPU worker 机器（ubuntu3090_debug_4090 所在主机）磁盘写满，业务接口
+全部失败，但网关**持续把请求打到该机器近 16 个小时**（21:30 故障发生 → 次日 13:36
+人工重启），没有切换到其他 worker。复盘结论是三层防线同时失效：
+
+| 层 | 问题 | 后果 |
+|---|---|---|
+| 1. worker `/health` 太浅 | 只返回内存里的"模型就绪"标志，磁盘满/下游 ComfyUI 挂掉不影响它 | 探测一直 200 |
+| 2. 业务错误不触发摘除 | 下游出错经包装后以 HTTP 200/400 + 错误码返回，`forward.py` 只对 连接异常/超时/≥500/401 摘除 | 请求失败但 worker 被视为健康 |
+| 3. 摘除后 8 秒即复活（**本方案要修的**） | `mark_worker_unhealthy()` 下线后，下一轮探测（8s 后）`/health` 返回 200，`healthy_threshold=1` 使其**一次成功立即重新上线** | 即使被摘除也会反复横跳，流量持续打入 |
+
+worker 侧的第 1 层已经修复（方案 A：`/health` 深检磁盘水位 + 下游依赖 TCP 探测，
+异常返回 503，代码在分支 `ubuntu3090_debug_4090`，提交 `14562d4`）。本方案修复
+第 3 层：**被动摘除后进入隔离期，隔离期内即使探测成功也不复活**。两层配合后：
+worker 磁盘满 → `/health` 503 → 网关 2 次探测失败摘除 → 隔离期内不再接流量。
+
+> 第 2 层（错误码语义化，涉及调用方兼容）暂不处理，单独立项。
+
+## 2. 现状代码行为（main 分支 gateway/pool.py）
+
+- `mark_worker_unhealthy(w)`：被 `forward.py` 在转发失败（连接异常/超时/5xx/401）时
+  调用，置 `w.online = False`。**没有任何冷却机制**。
+- `_health_check_loop(cfg)`：每 `interval_seconds`（默认 8s）探测所有 worker 的
+  `/health`；返回 200 时 `consecutive_successes += 1`，达到 `healthy_threshold`
+  （默认 **1**）且当前 offline → 立即 `online = True`。
+- 因此被动摘除最多 8 秒后就被一次探测成功拉回——如果 worker 是"浅健康"或故障
+  间歇性表现，就会陷入 摘除→复活→失败→摘除 的横跳。
+- 另外 58-96 行有一段**从未被调用的死代码** `_check_worker_health()` /
+  `_mark_failure()`（写了一半的旧实现），本次顺带删除，避免后人改错地方。
+
+## 3. 代码改动（共 2 个文件 + 1 个配置）
+
+### 3.1 `gateway/pool.py`
+
+**(1) `WorkerState` 增加一个字段：**
+
+```python
+@dataclass
+class WorkerState:
+    """单个 worker 的运行时状态。"""
+    url: str
+    online: bool = False  # 初始 offline，等健康检查通过后上线
+    busy: bool = False
+    consecutive_failures: int = 0
+    consecutive_successes: int = 0
+    offline_until: float = 0.0  # 被动摘除后的隔离截止时刻（time.monotonic() 基准）
+```
+
+**(2) 模块顶部（`logger = ...` 之后）加一个配置读取 helper：**
+
+```python
+def _cooldown_seconds() -> float:
+    """被动摘除后的隔离期时长（秒），配置 health_check.cooldown_seconds，默认 60。"""
+    try:
+        from gateway.config import get_config
+        return float(get_config()["health_check"].get("cooldown_seconds", 60))
+    except Exception:  # noqa: BLE001
+        return 60.0
+```
+
+（函数内 import 是为了不引入模块级循环依赖；`gateway/config.py` 不依赖 pool，顶层
+import 也可以。）
+
+**(3) `mark_worker_unhealthy()` 摘除时同时设置隔离期：**
+
+```python
+async def mark_worker_unhealthy(w: WorkerState) -> None:
+    """被动标记：转发失败时立即将该 worker 下线，并进入隔离期。
+
+    隔离期内即使 /health 探测成功也不复活（由 _health_check_loop 守卫），
+    避免"摘除→8 秒复活→再失败"的横跳。
+    """
+    async with _pool_condition:
+        if w.online:
+            w.online = False
+            w.consecutive_failures += 1
+            w.offline_until = time.monotonic() + _cooldown_seconds()
+            logger.warning("⚠ Worker 被动下线: %s（转发失败，隔离 %.0fs）",
+                           w.url, _cooldown_seconds())
+```
+
+**(4) `_health_check_loop()` 的探测成功分支加隔离期守卫。** 找到这段：
+
+```python
+                if resp.status_code == 200:
+                    w.consecutive_failures = 0
+                    w.consecutive_successes += 1
+                    if w.consecutive_successes >= healthy_threshold and not w.online:
+                        w.online = True
+                        logger.info("✅ Worker 上线: %s", w.url)
+```
+
+改为：
+
+```python
+                if resp.status_code == 200:
+                    w.consecutive_failures = 0
+                    w.consecutive_successes += 1
+                    if w.consecutive_successes >= healthy_threshold and not w.online:
+                        if time.monotonic() < w.offline_until:
+                            logger.debug("Worker %s 探测成功但处于隔离期，暂不复活", w.url)
+                        else:
+                            w.online = True
+                            logger.info("✅ Worker 上线: %s", w.url)
+```
+
+注意：`except`（连接异常）分支只累计失败、不涉及复活，无需改动；
+`init_pool()` 启动时的首轮探测同理无需改动（进程刚启动时 `offline_until` 为 0，
+不构成隔离）。
+
+**(5) 删除死代码**：`_check_worker_health()` 与
+`_mark_failure()` 两个函数整体删除（约 58-96 行，以函数名为准，别删到
+`_health_check_loop`）。删除前可确认无调用方：
+`grep -rn "_check_worker_health\|_mark_failure" .` 应只剩定义处。
+
+### 3.2 `gateway/config.py`
+
+`DEFAULTS["health_check"]` 增加一个 key：
+
+```python
+    "health_check": {
+        "path": "/health",
+        "interval_seconds": 8,
+        "timeout_seconds": 3,
+        "unhealthy_threshold": 2,
+        "healthy_threshold": 1,
+        "cooldown_seconds": 60,   # ← 新增：被动摘除后的隔离期（秒）
+    },
+```
+
+`config.example.json` 同步加这个 key。网关机器的实际 `config.json` **不加也不会报错**
+（代码里有 60 的默认兜底），但建议显式加上，便于运维感知。
+
+### 3.3 建议单测：`tests/test_gateway_pool.py`（新文件）
+
+gateway 子包目前没有任何测试。隔离期是纯逻辑，建议至少覆盖以下三个用例
+（不依赖真实网络）：
+
+```python
+"""gateway/pool.py 隔离期逻辑单测。"""
+import asyncio
+import time
+from unittest.mock import patch
+
+from gateway import pool
+from gateway.pool import WorkerState
+
+
+def _drain(coro):
+    return asyncio.run(coro)
+
+
+def test_passive_ejection_sets_cooldown():
+    w = WorkerState(url="http://127.0.0.1:9000", online=True)
+    async def run():
+        pool._pool_condition = asyncio.Condition()
+        with patch.object(pool, "_cooldown_seconds", return_value=60.0):
+            await pool.mark_worker_unhealthy(w)
+    _drain(run())
+    assert w.online is False
+    assert w.offline_until > time.monotonic()
+
+
+def test_probe_success_revives_after_cooldown():
+    w = WorkerState(url="http://127.0.0.1:9000")  # offline
+    w.offline_until = time.monotonic() - 1        # 隔离期已过
+    # 直接验证状态迁移逻辑（若按 3.1(4) 内联实现，可复制该分支逻辑做断言，
+    # 或将判断抽成 _try_revive(w, healthy_threshold) 小函数后单测）
+    ...
+
+
+def test_probe_success_blocked_during_cooldown():
+    w = WorkerState(url="http://127.0.0.1:9000")
+    w.offline_until = time.monotonic() + 60
+    ...  # 断言保持 offline
+```
+
+> 提示：如果按 3.1(4) 原样内联修改，后两个用例不好直接驱动，可以把"是否允许复活"
+> 抽成一个 6 行的纯函数 `_try_revive(w, healthy_threshold) -> bool` 再测，主循环
+> 调用它即可——实现方式自定，覆盖语义即可。
+
+## 4. 部署步骤（网关机器）
+
+1. `git pull` main 分支（含本方案代码改动）。
+2. （可选但建议）`config.json` 的 `health_check` 段加 `"cooldown_seconds": 60`。
+3. 重启网关服务（`hair-gateway.service` 或等价的 uvicorn 进程）。重启期间网关
+   本身短暂不可用，选低峰执行。
+4. 观察启动日志：`健康检查循环启动 | 间隔=8s | 下线阈值=2 | 上线阈值=1 | workers=N`
+   正常出现。
+
+## 5. 验证
+
+**功能验证（低峰演练）：**
+
+1. 人工把主 worker 摘除一次最直接的方法：临时在 worker 机器
+   `systemctl stop hair-worker`（或临时设 `HEALTH_MIN_FREE_GB=999` 重启）。
+2. 观察网关日志：
+   - 连续 2 次探测失败 → `⚠ Worker 下线`（原有行为）；
+   - **转发失败触发的** `⚠ Worker 被动下线: ...（转发失败，隔离 60s）`（新日志）；
+   - 恢复 worker 后，若刚被被动摘除过，即使 `/health` 恢复 200，也应看到 60 秒内
+     **不出现** `✅ Worker 上线`，隔离期满后下一次探测成功才上线。
+3. `GET /gateway-health` 观察 `workers_healthy` 的变化节奏。
+
+**回归确认：** 正常场景 worker 上下线不受影响（隔离期只对"被动摘除"生效；
+探测失败导致的下线，恢复后仍按 `healthy_threshold` 立即上线）。
+
+## 6. 回滚
+
+- 代码回滚：`git revert` 对应提交即可，无数据/配置迁移。
+- 紧急关闭隔离期：网关机器 `config.json` 设 `"cooldown_seconds": 0`
+  （等价于旧行为——摘除后下一次探测成功即复活），无需回代码。
+
+## 7. 附：worker 侧方案 A 的行为（已上线，供联调参考）
+
+worker（分支 `ubuntu3090_debug_4090`，提交 `14562d4`）的 `/health` 现在会在以下
+情况返回 **503**（网关 2 次探测失败即摘除）：
+
+- 模型加载中：`503 {"status":"starting"}`（原有行为不变）
+- 磁盘剩余低于阈值：`503 {"status":"degraded","reason":"disk_free_below_5.0g","disk_free_gb":...}`
+- 下游依赖（ComfyUI 8188 / change_hair 8801）TCP 不通：
+  `503 {"status":"degraded","reason":"dep_down:comfyui,...","deps":{...}}`
+
+正常时返回 `200 {"status":"ok","disk_free_gb":24.6,"deps":{"comfyui":true,"change_hair":true}}`。
+
+worker 侧配置（`worker_config.json` 的 `health` 段，环境变量可覆盖）：
+`min_free_gb`（默认 5，0=关闭）、`check_deps`（默认 true）、`dep_comfyui`（默认
+127.0.0.1:8188）、`deps_cache_seconds`（默认 5）。**其他 worker 机器如果也想启用
+深检，cherry-pick 该提交并补一份 worker_config.json 的 health 段即可。**

tests/test_api.py

@@ -36,12 +36,76 @@ def test_auth_missing_token_401(client):
     assert r.status_code == 401
 
 
-def test_health_no_token_200(client):
+def _patch_health_ok(monkeypatch):
+    """让 /health 深检确定性通过：不依赖本机磁盘水位与 8188/8801 服务存活。"""
+    monkeypatch.setattr(app_module, "_HEALTH_CFG", {
+        **app_module._HEALTH_CFG, "min_free_gb": 0.0, "check_deps": True})
+
+    async def _deps_ok():
+        return True, {"comfyui": True, "change_hair": True}
+    monkeypatch.setattr(app_module, "_check_deps", _deps_ok)
+
+
+def test_health_no_token_200(client, monkeypatch):
+    _patch_health_ok(monkeypatch)
     r = client.get("/health")
     assert r.status_code == 200
     assert r.json()["status"] == "ok"
 
 
+def test_health_disk_low_503(client, monkeypatch):
+    # 阈值调到不可能的 99999G → 必然触发磁盘水位 503
+    monkeypatch.setattr(app_module, "_HEALTH_CFG", {
+        **app_module._HEALTH_CFG, "min_free_gb": 99999.0, "check_deps": False})
+    r = client.get("/health")
+    assert r.status_code == 503
+    body = r.json()
+    assert body["status"] == "degraded"
+    assert body["reason"].startswith("disk_free_below")
+
+
+def test_health_dep_down_503(client, monkeypatch):
+    monkeypatch.setattr(app_module, "_HEALTH_CFG", {
+        **app_module._HEALTH_CFG, "min_free_gb": 0.0, "check_deps": True})
+
+    async def _deps_down():
+        return False, {"comfyui": False, "change_hair": True}
+    monkeypatch.setattr(app_module, "_check_deps", _deps_down)
+    r = client.get("/health")
+    assert r.status_code == 503
+    body = r.json()
+    assert body["status"] == "degraded"
+    assert "comfyui" in body["reason"]
+
+
+def test_health_ok_carries_detail(client, monkeypatch):
+    _patch_health_ok(monkeypatch)
+    r = client.get("/health")
+    assert r.status_code == 200
+    body = r.json()
+    assert "disk_free_gb" in body
+    assert body["deps"] == {"comfyui": True, "change_hair": True}
+
+
+def test_tcp_probe_real_socket():
+    # 用进程内真实 socket 验证探测本身（曾因 open_connection 返回元组误判全部不可达）
+    import asyncio
+
+    async def scenario():
+        server = await asyncio.start_server(lambda r, w: None, host="127.0.0.1", port=0)
+        port = server.sockets[0].getsockname()[1]
+        up = await app_module._tcp_probe("127.0.0.1", port)
+        server.close()
+        await server.wait_closed()
+        # 端口 1 无监听，本机 connect 立即被拒
+        down = await app_module._tcp_probe("127.0.0.1", 1)
+        return up, down
+
+    up, down = asyncio.run(scenario())
+    assert up is True
+    assert down is False
+
+
 def test_param_none_provided_1007(client):
     r = client.post(URL, headers=H)
     assert r.json()["code"] == 1007

worker_config.example.json

@@ -1,5 +1,11 @@
 {
-  "_comment": "复制为 worker_config.json（不入 git）。accept_passwords：worker 内网鉴权密码列表，网关带 X-Internal-Token: <其中之一>（也可用环境变量 WORKER_ACCEPT_PASSWORDS 覆盖）。comfyui_password：本机 ComfyUI(8188) Basic Auth 密码（user 默认 admin；也可放 password.txt 或环境变量 COMFYUI_PASSWORD）。注：接口4 火山方舟 ark_api_key 已迁到网关配置。",
+  "_comment": "复制为 worker_config.json（不入 git）。accept_passwords：worker 内网鉴权密码列表，网关带 X-Internal-Token: <其中之一>（也可用环境变量 WORKER_ACCEPT_PASSWORDS 覆盖）。comfyui_password：本机 ComfyUI(8188) Basic Auth 密码（user 默认 admin；也可放 password.txt 或环境变量 COMFYUI_PASSWORD）。health：深健康检查（/health 磁盘水位+下游依赖 TCP 探测，均可用环境变量 HEALTH_MIN_FREE_GB / HEALTH_CHECK_DEPS / HEALTH_DEP_COMFYUI / HEALTH_DEPS_CACHE_SECONDS 覆盖）。注：接口4 火山方舟 ark_api_key 已迁到网关配置。",
   "accept_passwords": ["change-me-to-a-strong-secret"],
-  "comfyui_password": "your-comfyui-basic-auth-password"
+  "comfyui_password": "your-comfyui-basic-auth-password",
+  "health": {
+    "min_free_gb": 5,
+    "check_deps": true,
+    "dep_comfyui": "127.0.0.1:8188",
+    "deps_cache_seconds": 5
+  }
 }