fix: harden IBKR runtime against historical failures

.github/workflows/ci.yml

@@ -100,6 +100,9 @@ jobs:
           assert resolve_hk_canonical_profile("hk_global_etf_tactical_rotation") == "hk_global_etf_tactical_rotation"
           PY
 
+      - name: Validate production Cloud Run startup
+        run: uv run --no-sync python scripts/validate_cloud_run_startup.py
+
       - name: Install editable shared repositories
         run: |
           set -euo pipefail

Dockerfile

@@ -16,6 +16,7 @@ RUN apt-get update \
 COPY . .
 RUN python -m pip install --upgrade pip uv \
     && uv sync --frozen --no-dev \
+    && python scripts/validate_cloud_run_startup.py \
     && apt-get purge -y git \
     && apt-get autoremove -y --purge \
     && rm -rf /var/lib/apt/lists/* \

application/runtime_broker_adapters.py

@@ -10,6 +10,10 @@
 from quant_platform_kit.common.models import OrderIntent, PortfolioSnapshot, Position
 
 
+class IBKRGatewayUnavailableError(ConnectionError):
+    """Raised after retryable IBKR gateway connection attempts are exhausted."""
+
+
 @dataclass(frozen=True)
 class IBKRRuntimeBrokerAdapters:
     host_resolver: Any
@@ -51,6 +55,7 @@ class IBKRRuntimeBrokerAdapters:
     execution_mode: str = "paper"
     limit_buy_premium_by_symbol: dict[str, float] | None = None
     printer: Any = print
+    refresh_host_fn: Any = None
 
     def validate_configured_accounts(self, ib):
         if not self.account_ids:
@@ -119,9 +124,23 @@ def connect_ib(self):
                     f"error={exc})",
                     flush=True,
                 )
-                if attempt < self.connect_attempts and self.connect_retry_delay_seconds > 0:
-                    self.sleep_fn(self.connect_retry_delay_seconds)
-        raise last_error
+                if attempt < self.connect_attempts:
+                    if callable(self.refresh_host_fn):
+                        try:
+                            host = self.refresh_host_fn()
+                        except Exception as refresh_exc:
+                            self.printer(
+                                "IB gateway host refresh failed; retrying last-known-good host "
+                                f"(host={host}, error_type={type(refresh_exc).__name__}, "
+                                f"error={refresh_exc})",
+                                flush=True,
+                            )
+                    if self.connect_retry_delay_seconds > 0:
+                        self.sleep_fn(self.connect_retry_delay_seconds)
+        raise IBKRGatewayUnavailableError(
+            "IB gateway unavailable after "
+            f"{self.connect_attempts} attempt(s) to {host}:{self.ib_port}: {last_error}"
+        ) from last_error
 
     def get_current_portfolio(self, ib):
         snapshot = self.fetch_account_portfolio_snapshot(ib)
@@ -300,6 +319,7 @@ def build_runtime_broker_adapters(
     execution_mode: str = "paper",
     limit_buy_premium_by_symbol: dict[str, float] | None = None,
     printer=print,
+    refresh_host_fn=None,
 ) -> IBKRRuntimeBrokerAdapters:
     return IBKRRuntimeBrokerAdapters(
         host_resolver=host_resolver,
@@ -341,4 +361,5 @@ def build_runtime_broker_adapters(
         market_currency=str(market_currency or "USD").upper(),
         execution_mode=str(execution_mode or "paper").strip().lower().replace("-", "_"),
         printer=printer,
+        refresh_host_fn=refresh_host_fn,
     )

main.py

@@ -19,7 +19,10 @@
     get_compute_discovery = None
 
 from application.cycle_result import coerce_strategy_cycle_result
-from application.runtime_broker_adapters import build_runtime_broker_adapters
+from application.runtime_broker_adapters import (
+    IBKRGatewayUnavailableError,
+    build_runtime_broker_adapters,
+)
 from application.runtime_composer import build_runtime_composer
 from application.runtime_deadline import (
     RuntimeDeadlineExceeded,
@@ -141,6 +144,26 @@ def get_ib_host():
     return host
 
 
+def refresh_ib_host():
+    """Refresh a zoned gateway address while retaining last-known-good state."""
+    global IB_HOST
+    instance_name = RUNTIME_SETTINGS.ib_gateway_instance_name
+    zone = RUNTIME_SETTINGS.ib_gateway_zone
+    if not zone:
+        return get_ib_host()
+
+    cached_host = IB_HOST
+    try:
+        resolved_host = resolve_gce_instance_ip(instance_name, zone)
+    except Exception as exc:
+        print(f"GCE host refresh failed for {instance_name}: {exc}", flush=True)
+        resolved_host = cached_host or instance_name
+    if cached_host and resolved_host == instance_name:
+        resolved_host = cached_host
+    IB_HOST = resolved_host
+    return resolved_host
+
+
 def get_ib_gateway_mode():
     return RUNTIME_SETTINGS.ib_gateway_mode
 
@@ -514,6 +537,7 @@ def build_broker_adapters(*, dry_run_only_override: bool | None = None):
     effective_dry_run_only = RUNTIME_SETTINGS.dry_run_only if dry_run_only_override is None else bool(dry_run_only_override)
     return build_runtime_broker_adapters(
         host_resolver=get_ib_host,
+        refresh_host_fn=refresh_ib_host,
         ib_port=IB_PORT,
         ib_client_id=IB_CLIENT_ID,
         connect_timeout_seconds=IB_CONNECT_TIMEOUT_SECONDS,
@@ -1309,29 +1333,35 @@ def _handle_request(
             },
         )
         raise
-    except TimeoutError as exc:
+    except IBKRGatewayUnavailableError as exc:
         append_runtime_report_error(
             report,
             stage="ibkr_connect",
             message=str(exc),
             error_type=type(exc).__name__,
+            failure_category="ibkr_gateway_unavailable",
+        )
+        finalize_runtime_report(
+            report,
+            status="error",
+            diagnostics={"failure_category": "ibkr_gateway_unavailable"},
         )
-        finalize_runtime_report(report, status="error")
         log_runtime_event(
             log_context,
-            "ibkr_gateway_connect_timeout",
-            message="IBKR gateway handshake timed out",
+            "ibkr_gateway_connect_failed",
+            message="IBKR gateway connection attempts exhausted",
             severity="ERROR",
             error_type=type(exc).__name__,
             error_message=str(exc),
+            failure_category="ibkr_gateway_unavailable",
         )
         error_msg = f"🚨 【IBKR 连接异常】\n{str(exc)}"
         _publish_runtime_failure_notification(
             detailed_text=error_msg,
             compact_text=error_msg,
             exc=exc,
         )
-        return "Error", 500
+        return "Error", 503
     except Exception as exc:
         append_runtime_report_error(
             report,

scripts/validate_cloud_run_startup.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Import the production WSGI app with an inert IBKR configuration.
+
+This check performs no network or broker operations. It catches dependency API
+drift, module import failures, and Flask route registration conflicts before a
+container can be deployed.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+def _install_smoke_environment() -> None:
+    account_group = "startup-smoke"
+    os.environ["ACCOUNT_GROUP"] = account_group
+    os.environ["IB_ACCOUNT_GROUP_CONFIG_JSON"] = json.dumps(
+        {
+            "groups": {
+                account_group: {
+                    "ib_gateway_instance_name": "127.0.0.1",
+                    "ib_gateway_mode": "paper",
+                    "ib_client_id": 1,
+                    "account_ids": ["DU000000"],
+                }
+            }
+        },
+        separators=(",", ":"),
+    )
+    os.environ["IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME"] = ""
+    os.environ["RUNTIME_TARGET_JSON"] = json.dumps(
+        {
+            "platform_id": "ibkr",
+            "strategy_profile": "global_etf_rotation",
+            "dry_run_only": True,
+            "execution_mode": "paper",
+            "account_scope": account_group,
+        },
+        separators=(",", ":"),
+    )
+
+
+def validate_startup() -> None:
+    _install_smoke_environment()
+
+    import main
+
+    routes = {
+        rule.rule: set(rule.methods) - {"HEAD", "OPTIONS"}
+        for rule in main.app.url_map.iter_rules()
+    }
+    required_routes = {
+        "/health": {"GET"},
+        "/run": {"GET", "POST"},
+        "/dry-run": {"GET", "POST"},
+        "/probe": {"GET", "POST"},
+        "/monitor-dispatch": {"GET", "POST"},
+    }
+    missing_or_invalid = {
+        path: {"expected": sorted(methods), "actual": sorted(routes.get(path, set()))}
+        for path, methods in required_routes.items()
+        if routes.get(path) != methods
+    }
+    if missing_or_invalid:
+        raise RuntimeError(f"Cloud Run route contract is invalid: {missing_or_invalid}")
+
+    print(f"Cloud Run startup validation passed: routes={len(routes)}")
+
+
+if __name__ == "__main__":
+    validate_startup()

tests/test_cloud_run_startup_validation.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_production_startup_validation_is_a_ci_and_image_build_gate():
+    command = "python scripts/validate_cloud_run_startup.py"
+    dockerfile = (ROOT / "Dockerfile").read_text()
+    ci_workflow = (ROOT / ".github" / "workflows" / "ci.yml").read_text()
+
+    assert command in dockerfile
+    assert "Validate production Cloud Run startup" in ci_workflow
+    assert f"uv run --no-sync {command}" in ci_workflow

tests/test_event_loop.py

@@ -138,6 +138,42 @@ def get(self, project, zone, instance):
     assert module.IB_HOST == "10.0.0.8"
 
 
+def test_get_ib_host_refreshes_zoned_gateway_ip(strategy_module_factory, monkeypatch):
+    module = strategy_module_factory(
+        IB_GATEWAY_ZONE="us-central1-a",
+        IB_ACCOUNT_GROUP_CONFIG_JSON=(
+            '{"groups":{"default":{"ib_gateway_instance_name":"ib-gateway",'
+            '"ib_gateway_mode":"live","ib_client_id":1}}}'
+        ),
+    )
+    resolved = iter(("10.0.0.8", "10.0.0.9"))
+    monkeypatch.setattr(module, "resolve_gce_instance_ip", lambda *_args: next(resolved))
+
+    assert module.get_ib_host() == "10.0.0.8"
+    assert module.get_ib_host() == "10.0.0.8"
+    assert module.refresh_ib_host() == "10.0.0.9"
+    assert module.IB_HOST == "10.0.0.9"
+
+
+def test_refresh_ib_host_keeps_cached_ip_when_gce_lookup_fails(strategy_module_factory, monkeypatch):
+    module = strategy_module_factory(
+        IB_GATEWAY_ZONE="us-central1-a",
+        IB_ACCOUNT_GROUP_CONFIG_JSON=(
+            '{"groups":{"default":{"ib_gateway_instance_name":"ib-gateway",'
+            '"ib_gateway_mode":"live","ib_client_id":1}}}'
+        ),
+    )
+    module.IB_HOST = "10.0.0.8"
+    monkeypatch.setattr(
+        module,
+        "resolve_gce_instance_ip",
+        lambda *_args: (_ for _ in ()).throw(RuntimeError("compute API unavailable")),
+    )
+
+    assert module.refresh_ib_host() == "10.0.0.8"
+    assert module.IB_HOST == "10.0.0.8"
+
+
 def test_ib_gateway_mode_derives_paper_port(strategy_module_factory):
     module = strategy_module_factory(
         IB_ACCOUNT_GROUP_CONFIG_JSON=(

tests/test_request_handling.py

@@ -903,6 +903,79 @@ def test_handle_request_error_persists_machine_readable_report(strategy_module,
     assert len(observed["messages"]) == 1
 
 
+def test_handle_request_classifies_gateway_connection_failure_as_unavailable(strategy_module, monkeypatch):
+    observed = {"events": [], "notifications": []}
+
+    monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True)
+    monkeypatch.setattr(
+        strategy_module,
+        "run_strategy_core",
+        lambda **_kwargs: (_ for _ in ()).throw(
+            strategy_module.IBKRGatewayUnavailableError(
+                "TCP preflight failed: connection refused"
+            )
+        ),
+    )
+    monkeypatch.setattr(
+        strategy_module,
+        "persist_execution_report",
+        lambda report: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json",
+    )
+    monkeypatch.setattr(
+        strategy_module,
+        "log_runtime_event",
+        lambda _context, event, **fields: observed["events"].append((event, fields)),
+    )
+    monkeypatch.setattr(
+        strategy_module,
+        "publish_notification",
+        lambda **kwargs: observed["notifications"].append(kwargs),
+    )
+
+    with strategy_module.app.test_request_context("/run", method="POST"):
+        body, status = strategy_module.handle_request()
+
+    assert status == 503
+    assert body == "Error"
+    assert observed["report"]["status"] == "error"
+    assert observed["report"]["errors"][0]["stage"] == "ibkr_connect"
+    assert observed["report"]["errors"][0]["failure_category"] == "ibkr_gateway_unavailable"
+    assert observed["report"]["diagnostics"]["failure_category"] == "ibkr_gateway_unavailable"
+    assert observed["events"][-1][0] == "ibkr_gateway_connect_failed"
+    assert observed["events"][-1][1]["failure_category"] == "ibkr_gateway_unavailable"
+    assert len(observed["notifications"]) == 1
+
+
+def test_handle_request_does_not_misclassify_unrelated_timeout_as_gateway_failure(strategy_module, monkeypatch):
+    observed = {"events": []}
+
+    monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True)
+    monkeypatch.setattr(
+        strategy_module,
+        "run_strategy_core",
+        lambda **_kwargs: (_ for _ in ()).throw(TimeoutError("snapshot storage timed out")),
+    )
+    monkeypatch.setattr(
+        strategy_module,
+        "persist_execution_report",
+        lambda report: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json",
+    )
+    monkeypatch.setattr(
+        strategy_module,
+        "log_runtime_event",
+        lambda _context, event, **fields: observed["events"].append((event, fields)),
+    )
+    monkeypatch.setattr(strategy_module, "publish_notification", lambda **_kwargs: None)
+
+    with strategy_module.app.test_request_context("/run", method="POST"):
+        body, status = strategy_module.handle_request()
+
+    assert status == 500
+    assert body == "Error"
+    assert observed["report"]["errors"][0]["stage"] == "strategy_cycle"
+    assert observed["events"][-1][0] == "strategy_cycle_failed"
+
+
 def test_run_strategy_core_allows_multiple_runs_in_same_process(strategy_module, monkeypatch):
     observed = {"connect_calls": 0, "disconnect_calls": 0, "messages": []}