feat(gateway): 被动摘除后进入 60s 隔离期，防探测复活横跳

落地 docs/网关摘除隔离期实施方案.md（9·19 worker 磁盘满事故第 3 层防线）：
- pool.py: WorkerState.offline_until + _cooldown_seconds()；mark_worker_unhealthy
  摘除时设隔离截止时刻，隔离期内探测成功也不复活（_try_revive 守卫）
- pool.py: 删除从未被调用的死代码 _check_worker_health/_mark_failure
- config.py + config.example.json: health_check.cooldown_seconds（默认 60，0=关闭）
- tests/test_gateway_pool.py: 9 用例（摘除设隔离/隔离内不复活/期满复活/整链路等）

网关机已重启验证：启动日志含"隔离期=60s"，/gateway-health 正常。

gateway/config.example.json

@@ -11,7 +11,8 @@
     "interval_seconds": 8,
     "timeout_seconds": 3,
     "unhealthy_threshold": 2,
-    "healthy_threshold": 1
+    "healthy_threshold": 1,
+    "cooldown_seconds": 60
   },
   "ark_api_key": "",
   "dispatch": {

gateway/config.py

@@ -21,6 +21,7 @@ DEFAULTS = {
         "timeout_seconds": 3,
         "unhealthy_threshold": 2,
         "healthy_threshold": 1,
+        "cooldown_seconds": 60,  # 被动摘除后的隔离期（秒），0=关闭（等价旧行为）
     },
     "dispatch": {
         "per_worker_concurrency": 1,

gateway/pool.py

@@ -18,6 +18,15 @@ import httpx
 logger = logging.getLogger("gateway.pool")
 
 
+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
+
+
 # ---------------------------------------------------------------------------
 # 异常
 # ---------------------------------------------------------------------------
@@ -39,6 +48,7 @@ class WorkerState:
     busy: bool = False
     consecutive_failures: int = 0
     consecutive_successes: int = 0
+    offline_until: float = 0.0  # 被动摘除后的隔离截止时刻（time.monotonic() 基准）
 
 
 # ---------------------------------------------------------------------------
@@ -63,45 +73,18 @@ _global_waiting: int = 0
 # 健康检查后台任务
 # ---------------------------------------------------------------------------
 
-async def _check_worker_health(
-    client: httpx.AsyncClient,
-    w: WorkerState,
-    cfg: dict,
-) -> None:
-    """探测单个 worker 的 /health，更新上下线状态。"""
-    hc_cfg = cfg["health_check"]
-    url = f"{w.url}{hc_cfg['path']}"
-    token = cfg["shared_password"]
+def _try_revive(w: WorkerState, healthy_threshold: int) -> bool:
+    """探测成功后是否允许将该 worker 复活上线。
 
-    try:
-        resp = await client.get(
-            url,
-            headers={"X-Internal-Token": token},
-            timeout=hc_cfg["timeout_seconds"],
-        )
-        if resp.status_code == 200:
-            w.consecutive_failures = 0
-            w.consecutive_successes += 1
-            if w.consecutive_successes >= hc_cfg["healthy_threshold"]:
-                if not w.online:
-                    w.online = True
-                    logger.info("Worker 上线: %s（连续成功 %d 次）", w.url, w.consecutive_successes)
-        else:
-            _mark_failure(w, f"HTTP {resp.status_code}")
-    except Exception as exc:
-        _mark_failure(w, str(exc))
-
-
-def _mark_failure(w: WorkerState, reason: str) -> None:
-    """记录一次失败，达到阈值后下线。"""
-    w.consecutive_successes = 0
-    w.consecutive_failures += 1
-    threshold = w.consecutive_failures  # used in log
-    hc_threshold = 2  # default, will be overridden
-    if w.consecutive_failures >= hc_threshold:
-        # 实际 threshold 从配置读取，这里先做基本判断
-        pass
-    logger.debug("Worker %s 健康检查失败 (%d/%d): %s", w.url, w.consecutive_failures, 99, reason)
+    条件：未在线 + 连续成功达到阈值 + **不在被动摘除隔离期内**
+    （offline_until 之前即使 /health 探测成功也压着不上线，防横跳）。
+    """
+    if w.online or w.consecutive_successes < healthy_threshold:
+        return False
+    if time.monotonic() < w.offline_until:
+        logger.debug("Worker %s 探测成功但处于隔离期，暂不复活", w.url)
+        return False
+    return True
 
 
 async def _health_check_loop(cfg: dict) -> None:
@@ -113,8 +96,8 @@ async def _health_check_loop(cfg: dict) -> None:
     token = cfg["shared_password"]
 
     logger.info(
-        "健康检查循环启动 | 间隔=%ds | 下线阈值=%d | 上线阈值=%d | workers=%d",
-        interval, unhealthy_threshold, healthy_threshold, len(_workers),
+        "健康检查循环启动 | 间隔=%ds | 下线阈值=%d | 上线阈值=%d | 隔离期=%.0fs | workers=%d",
+        interval, unhealthy_threshold, healthy_threshold, _cooldown_seconds(), len(_workers),
     )
 
     async with httpx.AsyncClient() as client:
@@ -130,7 +113,7 @@ async def _health_check_loop(cfg: dict) -> None:
                     if resp.status_code == 200:
                         w.consecutive_failures = 0
                         w.consecutive_successes += 1
-                        if w.consecutive_successes >= healthy_threshold and not w.online:
+                        if _try_revive(w, healthy_threshold):
                             w.online = True
                             logger.info("✅ Worker 上线: %s", w.url)
                     else:
@@ -272,12 +255,19 @@ def release_global_slot() -> None:
 # ---------------------------------------------------------------------------
 
 async def mark_worker_unhealthy(w: WorkerState) -> None:
-    """被动标记：转发失败时立即将该 worker 下线。"""
+    """被动标记：转发失败时立即将该 worker 下线，并进入隔离期。
+
+    隔离期内即使 /health 探测成功也不复活（由 _try_revive 守卫），
+    避免"摘除→8 秒复活→再失败"的横跳。
+    """
     async with _pool_condition:
         if w.online:
             w.online = False
             w.consecutive_failures += 1
-            logger.warning("⚠ Worker 被动下线: %s（转发失败）", w.url)
+            cooldown = _cooldown_seconds()
+            w.offline_until = time.monotonic() + cooldown
+            logger.warning("⚠ Worker 被动下线: %s（转发失败，隔离 %.0fs）",
+                           w.url, cooldown)
 
 
 async def acquire_worker(cfg: dict) -> WorkerState:

tests/test_gateway_pool.py

@@ -0,0 +1,154 @@
+"""gateway/pool.py 隔离期逻辑单测（不依赖真实网络）。
+
+覆盖《docs/网关摘除隔离期实施方案.md》§3.3 的三个用例：
+- 被动摘除设置隔离期（offline_until > now）
+- 隔离期内探测成功不复活
+- 隔离期满后探测成功恢复上线
+外加：_try_revive 的阈值/在线短路、mark_worker_unhealthy 对已下线 worker 的幂等、
+_health_check_loop 整链路验证、_cooldown_seconds 配置读取。
+
+与 test_gateway_serialization.py 同款：asyncio.run 套同步测试，不依赖 pytest-asyncio。
+"""
+
+import asyncio
+import time
+from unittest.mock import patch
+
+from gateway import pool
+from gateway.pool import WorkerState
+
+_HC_CFG = {
+    "health_check": {
+        "path": "/health",
+        "interval_seconds": 0.05,  # 测试用：快速轮询
+        "timeout_seconds": 3,
+        "unhealthy_threshold": 2,
+        "healthy_threshold": 1,
+    },
+    "shared_password": "test-token",
+}
+
+
+# ---------------------------------------------------------------------------
+# 被动摘除 → 隔离期
+# ---------------------------------------------------------------------------
+
+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)
+
+    asyncio.run(run())
+    assert w.online is False
+    assert w.offline_until > time.monotonic(), "被动摘除必须设置未来的隔离截止时刻"
+    assert w.consecutive_failures == 1
+
+
+def test_passive_ejection_idempotent_when_offline():
+    """已下线的 worker 再次被标记：不重复计失败、不续期隔离。"""
+    w = WorkerState(url="http://127.0.0.1:9000", online=False)
+    w.offline_until = time.monotonic() + 30
+    before = w.offline_until
+
+    async def run():
+        pool._pool_condition = asyncio.Condition()
+        with patch.object(pool, "_cooldown_seconds", return_value=60.0):
+            await pool.mark_worker_unhealthy(w)
+
+    asyncio.run(run())
+    assert w.online is False
+    assert w.consecutive_failures == 0
+    assert w.offline_until == before
+
+
+# ---------------------------------------------------------------------------
+# _try_revive：探测成功后是否允许复活
+# ---------------------------------------------------------------------------
+
+def test_probe_success_blocked_during_cooldown():
+    w = WorkerState(url="http://127.0.0.1:9000")  # offline
+    w.consecutive_successes = 1
+    w.offline_until = time.monotonic() + 60
+    assert pool._try_revive(w, healthy_threshold=1) is False
+    assert w.online is False
+
+
+def test_probe_success_revives_after_cooldown():
+    w = WorkerState(url="http://127.0.0.1:9000")
+    w.consecutive_successes = 1
+    w.offline_until = time.monotonic() - 1  # 隔离期已过
+    assert pool._try_revive(w, healthy_threshold=1) is True
+
+
+def test_try_revive_requires_success_threshold():
+    w = WorkerState(url="http://127.0.0.1:9000")
+    w.consecutive_successes = 1
+    w.offline_until = 0.0
+    assert pool._try_revive(w, healthy_threshold=2) is False
+
+
+def test_try_revive_noop_when_already_online():
+    w = WorkerState(url="http://127.0.0.1:9000", online=True)
+    w.consecutive_successes = 1
+    assert pool._try_revive(w, healthy_threshold=1) is False
+
+
+# ---------------------------------------------------------------------------
+# _health_check_loop 整链路：隔离期内探测 200 不上线，期满后上线
+# ---------------------------------------------------------------------------
+
+def test_health_check_loop_respects_cooldown(monkeypatch):
+    w = WorkerState(url="http://127.0.0.1:9000", online=False)
+    w.offline_until = time.monotonic() + 60  # 处于隔离期
+
+    class _FakeResp:
+        status_code = 200
+
+    class _FakeClient:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            return False
+
+        async def get(self, url, headers=None, timeout=None):
+            return _FakeResp()
+
+    async def run():
+        monkeypatch.setattr(pool, "_workers", {"w1": w})
+        monkeypatch.setattr(pool, "_shutdown_event", asyncio.Event())
+        monkeypatch.setattr(pool.httpx, "AsyncClient", _FakeClient)
+        task = asyncio.create_task(pool._health_check_loop(_HC_CFG))
+        try:
+            await asyncio.sleep(0.2)  # 跑若干轮探测
+            assert w.consecutive_successes >= 1, "探测应已成功过"
+            assert w.online is False, "隔离期内即使探测成功也不得复活"
+
+            w.offline_until = time.monotonic() - 1  # 隔离期结束
+            await asyncio.sleep(0.2)
+            assert w.online is True, "隔离期满后下一次探测成功应上线"
+        finally:
+            pool._shutdown_event.set()
+            await asyncio.wait_for(task, timeout=2)
+
+    asyncio.run(run())
+
+
+# ---------------------------------------------------------------------------
+# _cooldown_seconds 配置读取
+# ---------------------------------------------------------------------------
+
+def test_cooldown_seconds_reads_config(monkeypatch):
+    from gateway import config as gw_config
+    monkeypatch.setattr(gw_config, "_config_cache",
+                        {"health_check": {"cooldown_seconds": 120}})
+    assert pool._cooldown_seconds() == 120.0
+
+
+def test_cooldown_seconds_falls_back_to_60(monkeypatch):
+    from gateway import config as gw_config
+    monkeypatch.setattr(gw_config, "_config_cache", None)  # get_config 会抛 RuntimeError
+    assert pool._cooldown_seconds() == 60.0