fix: handle CowApiUnavailableError at all CoW API entry points

api.cow.fi DNS hijacking (2026-04-14) caused triton reboot loop because
cowdao_cowpy.app_data.utils.build_all_app_codes() contacts the API at
import time and the unhandled NetworkError propagated all the way up.

- cow_utils.py: convert NetworkError → CowApiUnavailableError for DEFAULT_APP_DATA_HASH
- web/server.py: catch CowApiUnavailableError in _preload_cow_modules() (root crash site)
- swap.py/quotes.py: already inside try/except — verified by new tests
- 15 new tests covering the full failure cascade

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dvilelaf

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "iwa"
-version = "0.7.2"
+version = "0.7.3"
 description = "A secure, modular, and plugin-based framework for crypto agents and ops"
 readme = "README.md"
 requires-python = ">=3.12,<4.0"

src/iwa/plugins/gnosis/cow_utils.py

@@ -1,9 +1,20 @@
 """Utilities for CowSwap plugin."""
 
 import importlib
+import logging
 import sys
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
+
+class CowApiUnavailableError(RuntimeError):
+    """Raised when the CoW Protocol API is unreachable.
+
+    Callers should treat this as a transient failure and retry later,
+    rather than letting it propagate as an unhandled exception.
+    """
+
 # Lazy import cache for cowdao_cowpy modules to avoid asyncio.run() conflict
 _cowpy_cache: dict[str, Any] = {}
 
@@ -55,6 +66,21 @@ def get_cowpy_module(name: str) -> Any:
             module_path, attr_name = _COWPY_IMPORTS[name]
             module = importlib.import_module(module_path)
             _cowpy_cache[name] = getattr(module, attr_name)
+        except Exception as exc:
+            if name == "DEFAULT_APP_DATA_HASH":
+                # cowdao_cowpy.app_data.utils runs build_all_app_codes() at module
+                # level, which contacts api.cow.fi. If the API is unreachable (DNS
+                # hijack, outage, maintenance), the import raises. We convert this to
+                # a controlled error so callers can fail the swap gracefully instead
+                # of crashing the whole process.
+                logger.warning(
+                    "CoW Protocol API unreachable — could not fetch DEFAULT_APP_DATA_HASH "
+                    f"({exc}). CoW swaps will be unavailable until the API recovers."
+                )
+                raise CowApiUnavailableError(
+                    f"CoW Protocol API unreachable: {exc}"
+                ) from exc
+            raise
         finally:
             # Always restore real httpx, even if import fails
             if real_httpx is not None:

src/iwa/plugins/gnosis/tests/test_cow_utils.py

@@ -0,0 +1,71 @@
+"""Tests for cow_utils.get_cowpy_module — specifically CoW API unavailability."""
+
+from unittest.mock import patch
+
+import pytest
+
+from iwa.plugins.gnosis.cow_utils import CowApiUnavailableError
+
+
+class TestGetCowpyModuleApiDown:
+    """get_cowpy_module must not crash when api.cow.fi is unreachable."""
+
+    def _clear_cache(self):
+        from iwa.plugins.gnosis import cow_utils
+
+        cow_utils._cowpy_cache.clear()
+
+    def test_network_error_on_import_raises_cow_api_unavailable(self):
+        """When cowdao_cowpy.app_data.utils raises NetworkError at import time
+        (because api.cow.fi is down), get_cowpy_module('DEFAULT_APP_DATA_HASH')
+        must raise CowApiUnavailableError — a controlled error callers can catch."""
+        self._clear_cache()
+
+        network_error = Exception("Network error occurred: Connection refused")
+
+        with patch("importlib.import_module", side_effect=network_error):
+            from iwa.plugins.gnosis.cow_utils import get_cowpy_module
+
+            with pytest.raises(CowApiUnavailableError, match="CoW Protocol API unreachable"):
+                get_cowpy_module("DEFAULT_APP_DATA_HASH")
+
+    def test_other_modules_still_raise_original_error(self):
+        """Import failures for non-DEFAULT_APP_DATA_HASH modules propagate as-is."""
+        self._clear_cache()
+
+        network_error = Exception("Network error occurred: Connection refused")
+
+        with patch("importlib.import_module", side_effect=network_error):
+            from iwa.plugins.gnosis.cow_utils import get_cowpy_module
+
+            with pytest.raises(Exception, match="Network error"):
+                get_cowpy_module("OrderBookApi")
+
+    def test_cow_api_down_raises_specific_catchable_error(self):
+        """Regression: api.cow.fi being down must raise CowApiUnavailableError,
+        NOT a generic unhandled exception. This lets callers (swap, trader) catch
+        it explicitly without crashing the process."""
+        self._clear_cache()
+
+        network_error = Exception("Network error occurred: api.cow.fi unreachable")
+
+        with patch("importlib.import_module", side_effect=network_error):
+            from iwa.plugins.gnosis.cow_utils import get_cowpy_module
+
+            with pytest.raises(CowApiUnavailableError):
+                get_cowpy_module("DEFAULT_APP_DATA_HASH")
+
+    def test_cow_api_unavailable_error_wraps_original_cause(self):
+        """CowApiUnavailableError must chain the original exception as __cause__
+        so callers can inspect the root error if needed."""
+        self._clear_cache()
+
+        network_error = Exception("Connection refused to api.cow.fi")
+
+        with patch("importlib.import_module", side_effect=network_error):
+            from iwa.plugins.gnosis.cow_utils import get_cowpy_module
+
+            with pytest.raises(CowApiUnavailableError) as exc_info:
+                get_cowpy_module("DEFAULT_APP_DATA_HASH")
+
+        assert exc_info.value.__cause__ is network_error

src/iwa/web/server.py

@@ -22,12 +22,33 @@
 # Pre-load cowdao_cowpy modules BEFORE async loop starts
 # This is required because cowdao_cowpy uses asyncio.run() at import time
 # which fails if called from an already running event loop
-from iwa.plugins.gnosis.cow_utils import get_cowpy_module
+from iwa.plugins.gnosis.cow_utils import CowApiUnavailableError, get_cowpy_module
 
 # Import routers
 from iwa.web.routers import accounts, olas, rewards, state, subgraph, swap, transactions
 
-get_cowpy_module("DEFAULT_APP_DATA_HASH")  # Forces import now, not during async
+def _preload_cow_modules() -> None:
+    """Pre-load cowdao_cowpy before the async loop starts.
+
+    cowdao_cowpy uses asyncio.run() at import time, which fails if called from
+    an already-running event loop. We trigger the import here (synchronous
+    context) so it's cached for later async calls.
+
+    If api.cow.fi is unreachable at startup (DNS hijack, outage, etc.) we log a
+    warning and continue — CoW swaps will be unavailable but the server must NOT
+    crash. Individual swap calls will raise CowApiUnavailableError on demand.
+    """
+    try:
+        get_cowpy_module("DEFAULT_APP_DATA_HASH")
+    except CowApiUnavailableError as exc:
+        logger.warning(
+            "CoW Protocol API unreachable at startup (%s). "
+            "CoW swaps disabled until the API recovers.",
+            exc,
+        )
+
+
+_preload_cow_modules()
 
 # Configure logging (writes to iwa.log for frontend visibility)
 configure_logger()

src/tests/test_cow_protocol_gaps.py

@@ -5,6 +5,7 @@
 import pytest
 
 from iwa.plugins.gnosis.cow.types import OrderType
+from iwa.plugins.gnosis.cow_utils import CowApiUnavailableError
 
 # ---- CowSwap.swap() tests ----
 
@@ -249,3 +250,140 @@ async def test_get_max_buy_amount_wei(self):
 
         # 2000000 - 1.5% = 1970000
         assert result == 1970000
+
+
+# ---- CowApiUnavailableError resilience tests ----
+
+
+class TestCowApiUnavailableResilience:
+    """CoW API down must never crash triton. All entry points must degrade gracefully."""
+
+    @staticmethod
+    def _make_cowswap(mock_get_module):
+        """Build a CowSwap instance with fully mocked cowpy."""
+        from iwa.plugins.gnosis.cow.swap import CowSwap
+
+        chain = MagicMock()
+        chain.chain_id = 100
+        chain.name = "gnosis"
+        chain.get_token_address.return_value = "0xTokenAddr"
+
+        with patch.object(CowSwap, "get_chain", return_value=MagicMock()):
+            return CowSwap(private_key_or_signer=MagicMock(), chain=chain)
+
+    @pytest.mark.asyncio
+    @patch("iwa.plugins.gnosis.cow.swap.get_cowpy_module")
+    async def test_sell_swap_returns_none_when_cow_api_down(self, mock_get_module):
+        """Regression: CowSwap.swap(SELL) must return None when CoW API is down, not crash.
+
+        swap_tokens raises CowApiUnavailableError → caught by swap()'s except block.
+        """
+        mock_chain_cls = MagicMock()
+        mock_chain_cls.__iter__ = lambda self: iter([MagicMock(value=(MagicMock(),))])
+
+        def get_module(name):
+            return {
+                "SupportedChainId": MagicMock(),
+                "OrderBookApi": MagicMock(),
+                "OrderBookAPIConfigFactory": MagicMock(),
+                "Chain": mock_chain_cls,
+                "swap_tokens": AsyncMock(
+                    side_effect=CowApiUnavailableError("api.cow.fi unreachable")
+                ),
+            }.get(name, MagicMock())
+
+        mock_get_module.side_effect = get_module
+        cow = self._make_cowswap(mock_get_module)
+
+        result = await cow.swap(
+            amount_wei=10**18,
+            sell_token_name="olas",
+            buy_token_name="wxdai",
+            order_type=OrderType.SELL,
+            wait_for_execution=False,
+        )
+
+        assert result is None, "swap() must return None when CoW API is down, not crash"
+
+    @pytest.mark.asyncio
+    @patch("iwa.plugins.gnosis.cow.swap.get_cowpy_module")
+    async def test_buy_swap_returns_none_when_cow_api_down(self, mock_get_module):
+        """BUY path: CowApiUnavailableError from swap_tokens_to_exact_tokens → swap() returns None."""
+        mock_get_module.return_value = MagicMock()
+
+        cow = self._make_cowswap(mock_get_module)
+
+        with patch(
+            "iwa.plugins.gnosis.cow.swap.CowSwap.swap_tokens_to_exact_tokens",
+            new=AsyncMock(side_effect=CowApiUnavailableError("api.cow.fi unreachable")),
+        ):
+            result = await cow.swap(
+                amount_wei=10**18,
+                sell_token_name="olas",
+                buy_token_name="wxdai",
+                order_type=OrderType.BUY,
+                wait_for_execution=False,
+            )
+
+        assert result is None, "BUY swap() must return None when CoW API is down"
+
+    @pytest.mark.asyncio
+    async def test_get_max_sell_amount_raises_cow_api_unavailable(self):
+        """get_max_sell_amount_wei propagates CowApiUnavailableError when API is down.
+
+        Callers (e.g. get_swap_execution_cost_pct in trader.py) have their own
+        try/except and must handle this explicitly.
+        """
+        import iwa.plugins.gnosis.cow.quotes as quotes_mod
+
+        def raise_if_app_data(name):
+            if name == "DEFAULT_APP_DATA_HASH":
+                raise CowApiUnavailableError("api.cow.fi down")
+            return MagicMock()
+
+        with patch("iwa.plugins.gnosis.cow.quotes.get_cowpy_module", side_effect=raise_if_app_data):
+            quotes_mod.get_order_quote = None  # force re-fetch via get_cowpy_module
+
+            with pytest.raises(CowApiUnavailableError):
+                await quotes_mod.get_max_sell_amount_wei(
+                    amount_wei=10**18,
+                    sell_token="0xSell",
+                    buy_token="0xBuy",
+                    chain_id_val=100,
+                    account_address="0xAccount",
+                )
+
+    @pytest.mark.asyncio
+    async def test_get_max_buy_amount_raises_cow_api_unavailable(self):
+        """get_max_buy_amount_wei propagates CowApiUnavailableError when API is down."""
+        import iwa.plugins.gnosis.cow.quotes as quotes_mod
+
+        def raise_if_app_data(name):
+            if name == "DEFAULT_APP_DATA_HASH":
+                raise CowApiUnavailableError("api.cow.fi down")
+            return MagicMock()
+
+        with patch("iwa.plugins.gnosis.cow.quotes.get_cowpy_module", side_effect=raise_if_app_data):
+            quotes_mod.get_order_quote = None  # force re-fetch via get_cowpy_module
+
+            with pytest.raises(CowApiUnavailableError):
+                await quotes_mod.get_max_buy_amount_wei(
+                    sell_amount_wei=10**18,
+                    sell_token="0xSell",
+                    buy_token="0xBuy",
+                    chain_id_val=100,
+                    account_address="0xAccount",
+                )
+
+    def test_server_preload_does_not_crash_when_cow_api_down(self):
+        """Regression: api.cow.fi down at startup must NOT crash the web server.
+
+        _preload_cow_modules() catches CowApiUnavailableError and logs a warning.
+        This prevents the triton reboot loop caused by the CoW DNS hijacking incident.
+        """
+        from iwa.web import server as server_mod
+
+        with patch.object(server_mod, "get_cowpy_module") as mock_get:
+            mock_get.side_effect = CowApiUnavailableError("api.cow.fi unreachable at startup")
+            # Must not raise — must only log a warning
+            server_mod._preload_cow_modules()  # no exception = pass