엔진 readiness 실패 캐시 TTL 분리와 프로브 타임아웃 보정

공개 런타임에서 reasoning_effort=high 콜드 프로브 1회가 20초를 넘기면
실패가 성공과 같은 TTL(운영 1800초)로 캐시되어 공개 API가 최장 30분간
engine=false로 오염되고, 워치독은 기본 /ready의 캐시된 200만 보고 API만
재시작하는 무한 루프에 빠졌다(2026-08-18 06:30 KST 실측).

- 실패 항목 전용 ENGINE_READY_FAILURE_TTL_SECONDS(기본 30초)를 분리해
  복구가 프로브 한 번 안에 감지되게 하고, 진짜 장애 중 프로브 폭풍은
  실패 TTL이 계속 막는다.
- ENGINE_READY_TIMEOUT_SECONDS 기본값을 20→45초로 올려 고효율 콜드
  스폰의 정상 지연을 장애로 오판하지 않는다.
- 회귀 4건 신설(실패 재프로브·성공 캐시 유지·실패 폭풍 억제·기본값 계약),
  게이트웨이 62/62·API 전체 986 passed·ruff clean.

apps/api/engine_gateway/gateway.py

@@ -51,7 +51,14 @@ DEFAULT_MODEL = os.environ.get("ENGINE_MODEL", "")  # 비우면 CLI 기본(Opus
 FALLBACK_MODEL = os.environ.get("ENGINE_FALLBACK_MODEL", "")
 DEFAULT_BUDGET = float(os.environ.get("SESSION_BUDGET_USD", "5.0"))
 READY_TTL_SECONDS = float(os.environ.get("ENGINE_READY_TTL_SECONDS", "30"))
-READY_TIMEOUT_SECONDS = float(os.environ.get("ENGINE_READY_TIMEOUT_SECONDS", "20"))
+# 실패 결과를 성공과 같은 TTL로 캐시하면 콜드 프로브 1회 타임아웃이 운영 TTL(1800초)
+# 내내 공개 API를 engine=false로 오염시킨다(2026-08-18 06:30 KST 30분 장애). 실패만
+# 짧게 캐시해 재프로브하되, 진짜 장애 중 프로브 폭풍은 이 TTL이 막는다.
+READY_FAILURE_TTL_SECONDS = float(
+    os.environ.get("ENGINE_READY_FAILURE_TTL_SECONDS", "30")
+)
+# reasoning_effort=high 콜드 스폰은 정상 상태에서도 20초를 넘길 수 있다(2026-08-18 실측).
+READY_TIMEOUT_SECONDS = float(os.environ.get("ENGINE_READY_TIMEOUT_SECONDS", "45"))
 READY_BUDGET_USD = float(os.environ.get("ENGINE_READY_BUDGET_USD", "0.5"))
 # 단발 생성(/v1/generate) 턴 타임아웃 — 페르소나 초안 생성 같은 대형 구조화 출력은
 # 120초를 넘길 수 있어 설정 가능하게 한다(2026-07-15). 호출부(app ENGINE_TIMEOUT)와 정합 필요.
@@ -459,6 +466,13 @@ async def health():
     }
 
 
+def _ready_entry_ttl(entry: dict[str, Any]) -> float:
+    """실패 항목은 짧은 TTL로만 캐시해 복구가 프로브 한 번 안에 감지되게 한다."""
+    if entry.get("ok"):
+        return READY_TTL_SECONDS
+    return min(READY_TTL_SECONDS, READY_FAILURE_TTL_SECONDS)
+
+
 def _ready_response(
     entry: dict[str, Any],
     *,
@@ -505,7 +519,7 @@ async def ready(
         cache_key, {"checked_at": 0.0, "ok": False, "detail": "not checked"}
     )
     age = time.monotonic() - float(entry.get("checked_at", 0.0) or 0.0)
-    if not force and age < READY_TTL_SECONDS:
+    if not force and age < _ready_entry_ttl(entry):
         return _ready_response(
             entry,
             provider=provider,
@@ -520,7 +534,7 @@ async def ready(
             cache_key, {"checked_at": 0.0, "ok": False, "detail": "not checked"}
         )
         age = time.monotonic() - float(entry.get("checked_at", 0.0) or 0.0)
-        if not force and age < READY_TTL_SECONDS:
+        if not force and age < _ready_entry_ttl(entry):
             return _ready_response(
                 entry,
                 provider=provider,

apps/api/engine_gateway/test_ready_cache.py

@@ -0,0 +1,122 @@
+"""/ready 캐시의 실패 TTL 계약.
+
+2026-08-18 공개 런타임 장애: reasoning_effort=high 콜드 프로브 1회가 20초 타임아웃을
+넘기자 실패 결과가 성공과 같은 TTL(운영 1800초)로 캐시돼, 공개 API가 30분간
+engine=false로 오염됐다. 워치독은 기본 /ready(캐시 200)만 보고 엔진을 healthy로
+오판해 API만 재시작하는 무한 루프에 빠졌다. 실패는 짧은 TTL 뒤 재프로브해야 한다.
+"""
+
+import asyncio
+import unittest
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+from engine_gateway import gateway
+
+
+class _FakeProbe:
+    """EngineSession 대역: 호출 순서별로 미리 정한 결과를 돌려준다."""
+
+    def __init__(self, outcomes, calls):
+        self._outcomes = outcomes
+        self._calls = calls
+
+    async def start(self):
+        return None
+
+    async def turn(self, content, timeout=None):
+        index = min(len(self._calls), len(self._outcomes) - 1)
+        outcome = self._outcomes[index]
+        self._calls.append(outcome)
+        if isinstance(outcome, Exception):
+            raise outcome
+        return {"text": outcome}
+
+    async def failure_detail(self, reason):
+        return reason
+
+    async def close(self):
+        return None
+
+
+def _probe_factory(outcomes, calls):
+    def factory(*args, **kwargs):
+        return _FakeProbe(outcomes, calls)
+
+    return factory
+
+
+class ReadyFailureCacheTest(unittest.TestCase):
+    def setUp(self):
+        gateway._READY_CACHE.clear()
+        self.addCleanup(gateway._READY_CACHE.clear)
+
+    def test_failed_probe_is_retried_after_failure_ttl(self):
+        calls = []
+        with (
+            patch.object(gateway, "ENGINE_GATEWAY_SHARED_SECRET", ""),
+            patch.object(gateway, "READY_TTL_SECONDS", 1800.0),
+            patch.object(gateway, "READY_FAILURE_TTL_SECONDS", 0.0),
+            patch.object(
+                gateway,
+                "EngineSession",
+                _probe_factory([asyncio.TimeoutError(), "OK"], calls),
+            ),
+        ):
+            client = TestClient(gateway.app)
+            first = client.get("/ready")
+            second = client.get("/ready")
+
+        self.assertEqual(first.status_code, 503)
+        self.assertEqual(second.status_code, 200)
+        self.assertEqual(len(calls), 2)
+        self.assertFalse(second.json()["cached"])
+
+    def test_successful_probe_stays_cached_for_full_ttl(self):
+        calls = []
+        with (
+            patch.object(gateway, "ENGINE_GATEWAY_SHARED_SECRET", ""),
+            patch.object(gateway, "READY_TTL_SECONDS", 1800.0),
+            patch.object(gateway, "READY_FAILURE_TTL_SECONDS", 0.0),
+            patch.object(
+                gateway,
+                "EngineSession",
+                _probe_factory(["OK", asyncio.TimeoutError()], calls),
+            ),
+        ):
+            client = TestClient(gateway.app)
+            first = client.get("/ready")
+            second = client.get("/ready")
+
+        self.assertEqual(first.status_code, 200)
+        self.assertEqual(second.status_code, 200)
+        self.assertEqual(len(calls), 1)
+        self.assertTrue(second.json()["cached"])
+
+    def test_failed_probe_is_not_hammered_within_failure_ttl(self):
+        calls = []
+        with (
+            patch.object(gateway, "ENGINE_GATEWAY_SHARED_SECRET", ""),
+            patch.object(gateway, "READY_TTL_SECONDS", 1800.0),
+            patch.object(gateway, "READY_FAILURE_TTL_SECONDS", 60.0),
+            patch.object(
+                gateway,
+                "EngineSession",
+                _probe_factory([asyncio.TimeoutError(), "OK"], calls),
+            ),
+        ):
+            client = TestClient(gateway.app)
+            first = client.get("/ready")
+            second = client.get("/ready")
+
+        self.assertEqual(first.status_code, 503)
+        self.assertEqual(second.status_code, 503)
+        self.assertEqual(len(calls), 1)
+        self.assertTrue(second.json()["cached"])
+
+    def test_defaults_give_cold_high_effort_spawn_headroom(self):
+        # reasoning_effort=high 콜드 스폰은 20초를 자주 넘긴다(2026-08-18 실측).
+        self.assertGreaterEqual(gateway.READY_TIMEOUT_SECONDS, 45.0)
+        # 실패 캐시는 1분 넘게 유지하면 안 된다 — 복구 지연이 곧 공개 장애 시간이 된다.
+        self.assertLessEqual(gateway.READY_FAILURE_TTL_SECONDS, 60.0)