fix(r131): harden provider adapter error contract and retry-after handling

README.md

@@ -15,9 +15,9 @@ ComfyUI-OpenClaw is a **security-first orchestration layer** for ComfyUI that co
 
 This project is designed to make **ComfyUI a reliable automation target** with an explicit admin boundary and hardened defaults.
 
-<center>
-<img src="assets/adminMobileConsole.png" width="70%" />
-</center>
+<div align="center">
+  <img src="assets/adminMobileConsole.png" width="70%" />
+</div>
 
 ## Security stance (how this project differs from convenience-first automation packs):
 

services/providers/anthropic.py

@@ -4,12 +4,22 @@ from typing import Any, Dict, List, Optional
 
 try:
     from ..provider_errors import ProviderHTTPError
-    from ..retry_after import get_retry_after_seconds
-    from ..safe_io import STANDARD_OUTBOUND_POLICY, SSRFError, safe_request_json
+    from ..retry_after import parse_retry_after_body, parse_retry_after_header
+    from ..safe_io import (
+        STANDARD_OUTBOUND_POLICY,
+        SafeIOHTTPError,
+        SSRFError,
+        safe_request_json,
+    )
 except ImportError:
     from services.provider_errors import ProviderHTTPError
-    from services.retry_after import get_retry_after_seconds
-    from services.safe_io import STANDARD_OUTBOUND_POLICY, SSRFError, safe_request_json
+    from services.retry_after import parse_retry_after_body, parse_retry_after_header
+    from services.safe_io import (
+        STANDARD_OUTBOUND_POLICY,
+        SafeIOHTTPError,
+        SSRFError,
+        safe_request_json,
+    )
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.providers.anthropic")
 
@@ -17,6 +27,16 @@ logger = logging.getLogger("ComfyUI-OpenClaw.services.providers.anthropic")
 ANTHROPIC_API_VERSION = "2023-06-01"
 
 
+def _parse_error_body_dict(body: Optional[str]) -> Optional[Dict[str, Any]]:
+    if not body:
+        return None
+    try:
+        parsed = json.loads(body)
+    except (TypeError, json.JSONDecodeError):
+        return None
+    return parsed if isinstance(parsed, dict) else None
+
+
 def build_chat_request(
     messages: List[Dict[str, Any]],
     model: str,
@@ -97,22 +117,24 @@ def make_request(
 
         return {"text": text, "raw": raw}
 
+    except SafeIOHTTPError as e:
+        error_body = _parse_error_body_dict(e.body)
+        retry_after = parse_retry_after_header(e.headers)
+        if retry_after is None:
+            retry_after = parse_retry_after_body(error_body)
+
+        logger.error(f"Anthropic API error: {e}")
+        raise ProviderHTTPError(
+            status_code=e.status_code,
+            message=str(e),
+            provider="anthropic",
+            model=model,
+            retry_after=retry_after,
+            headers=e.headers,
+            body=error_body if error_body is not None else e.body,
+        )
+
     except RuntimeError as e:
-        # S65: safe_io wraps HTTP errors in RuntimeError with status code in message?
-        # No, safe_io implementation:
-        # raise RuntimeError(f"HTTP error {e.code}: {e.reason}")
-        # raise RuntimeError(f"Request failed: {e}")
-
-        # We need to parse the error message to extract status/body if possible,
-        # OR update safe_io to raise structured errors.
-        # Given existing safe_io implementation raises RuntimeError string,
-        # we try to parse it best-effort or treat as generic 500.
-
-        # However, for ProviderHTTPError compliance, we need status code and headers.
-        # safe_io currently DOES NOT return headers on error.
-        # This is a limitation of safe_io replacement.
-
-        # Let's try to parse status code from string "HTTP error 400: ..."
         params = str(e)
         status_code = 500
         import re
@@ -122,14 +144,12 @@ def make_request(
             status_code = int(m.group(1))
 
         logger.error(f"Anthropic API error: {e}")
-
-        # Re-raise as ProviderHTTPError if possible
         raise ProviderHTTPError(
             status_code=status_code,
             message=str(e),
             provider="anthropic",
             model=model,
-            retry_after=0,  # Header access lost in safe_io exception
+            retry_after=None,
         )
 
     except SSRFError as e:

services/providers/openai_compat.py

@@ -4,18 +4,20 @@ from typing import Any, Callable, Dict, List, Optional
 
 try:
     from ..provider_errors import ProviderHTTPError
-    from ..retry_after import get_retry_after_seconds
+    from ..retry_after import parse_retry_after_body, parse_retry_after_header
     from ..safe_io import (
         STANDARD_OUTBOUND_POLICY,
+        SafeIOHTTPError,
         SSRFError,
         safe_request_json,
         safe_request_text_stream,
     )
 except ImportError:
     from services.provider_errors import ProviderHTTPError
-    from services.retry_after import get_retry_after_seconds
+    from services.retry_after import parse_retry_after_body, parse_retry_after_header
     from services.safe_io import (
         STANDARD_OUTBOUND_POLICY,
+        SafeIOHTTPError,
         SSRFError,
         safe_request_json,
         safe_request_text_stream,
@@ -24,6 +26,16 @@ except ImportError:
 logger = logging.getLogger("ComfyUI-OpenClaw.services.providers.openai_compat")
 
 
+def _parse_error_body_dict(body: Optional[str]) -> Optional[Dict[str, Any]]:
+    if not body:
+        return None
+    try:
+        parsed = json.loads(body)
+    except (TypeError, json.JSONDecodeError):
+        return None
+    return parsed if isinstance(parsed, dict) else None
+
+
 def build_chat_request(
     messages: List[Dict[str, Any]],
     model: str,
@@ -43,10 +55,15 @@ def build_chat_request(
     # R39: Sanitize and include tools if provided
     if tools:
         try:
-            from services.schema_sanitizer import (
-                get_sanitization_summary,
-                sanitize_tools,
-            )
+            # CRITICAL: package-relative import must be tried first for ComfyUI
+            # custom_nodes package loading; fallback absolute import is for local tools.
+            try:
+                from ..schema_sanitizer import get_sanitization_summary, sanitize_tools
+            except ImportError:
+                from services.schema_sanitizer import (
+                    get_sanitization_summary,
+                    sanitize_tools,
+                )
 
             sanitized = sanitize_tools(tools, profile="openai_compat")
             if sanitized:
@@ -128,6 +145,23 @@ def make_request(
 
         return {"text": text, "raw": raw}
 
+    except SafeIOHTTPError as e:
+        error_body = _parse_error_body_dict(e.body)
+        retry_after = parse_retry_after_header(e.headers)
+        if retry_after is None:
+            retry_after = parse_retry_after_body(error_body)
+
+        logger.error(f"OpenAI-compat API error: {e}")
+        raise ProviderHTTPError(
+            status_code=e.status_code,
+            message=str(e),
+            provider="openai_compat",
+            model=model,
+            retry_after=retry_after,
+            headers=e.headers,
+            body=error_body if error_body is not None else e.body,
+        )
+
     except RuntimeError as e:
         # S65/R14: Attempt to reconstruct ProviderHTTPError from safe_io exception
 
@@ -147,7 +181,7 @@ def make_request(
             message=str(e),
             provider="openai_compat",
             model=model,
-            retry_after=0,
+            retry_after=None,
         )
 
     except SSRFError as e:
@@ -269,6 +303,23 @@ def make_request_stream(
             },
         }
 
+    except SafeIOHTTPError as e:
+        error_body = _parse_error_body_dict(e.body)
+        retry_after = parse_retry_after_header(e.headers)
+        if retry_after is None:
+            retry_after = parse_retry_after_body(error_body)
+
+        logger.error(f"OpenAI-compat streaming API error: {e}")
+        raise ProviderHTTPError(
+            status_code=e.status_code,
+            message=str(e),
+            provider="openai_compat",
+            model=model,
+            retry_after=retry_after,
+            headers=e.headers,
+            body=error_body if error_body is not None else e.body,
+        )
+
     except RuntimeError as e:
         params = str(e)
         status_code = 500
@@ -284,7 +335,7 @@ def make_request_stream(
             message=str(e),
             provider="openai_compat",
             model=model,
-            retry_after=0,
+            retry_after=None,
         )
     except SSRFError as e:
         logger.error(f"OpenAI-compat streaming SSRF blocked: {e}")

services/safe_io.py

@@ -14,11 +14,20 @@ import socket
 import tempfile
 import urllib.request
 from dataclasses import dataclass, field
-from typing import Any, FrozenSet, Optional, Set, Tuple
+from typing import Any, Dict, FrozenSet, Optional, Set, Tuple
 from urllib.parse import urlparse
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.safe_io")
 
+# IMPORTANT: Keep outbound header forwarding parity across JSON and stream
+# callers. Drift here previously caused provider behavior mismatches.
+ALLOWED_OUTBOUND_HEADER_PREFIXES = (
+    "x-",
+    "content-type",
+    "authorization",
+    "accept",
+)
+
 # ============================================================================
 # FILESYSTEM SAFETY
 # ============================================================================
@@ -193,6 +202,51 @@ class SSRFError(ValueError):
     pass
 
 
+class SafeIOHTTPError(RuntimeError):
+    """Structured HTTP failure raised by safe_request_* helpers."""
+
+    def __init__(
+        self,
+        *,
+        status_code: int,
+        reason: str,
+        method: str,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        body: Optional[str] = None,
+    ) -> None:
+        self.status_code = int(status_code)
+        self.reason = reason
+        self.method = method
+        self.url = url
+        self.headers = headers or {}
+        self.body = body
+        super().__init__(f"HTTP error {self.status_code}: {self.reason}")
+
+
+def _headers_to_dict(headers: Any) -> Dict[str, str]:
+    """Convert HTTP header container to plain dict[str, str]."""
+    if not headers:
+        return {}
+    try:
+        return {str(k): str(v) for k, v in headers.items()}
+    except Exception:
+        return {}
+
+
+def _http_error_body_preview(error: Exception, max_bytes: int = 4096) -> Optional[str]:
+    """Best-effort decode of HTTP error response body for retry-hint parsing."""
+    try:
+        raw = error.read(max_bytes)
+    except Exception:
+        return None
+    if not raw:
+        return None
+    if isinstance(raw, bytes):
+        return raw.decode("utf-8", errors="replace")
+    return str(raw)
+
+
 # ---------------------------------------------------------------------------
 # S51: Outbound Endpoint Policy v2
 # ---------------------------------------------------------------------------
@@ -593,11 +647,12 @@ def safe_request_json(
         # Add safe headers
         # R106: external control-plane adapter requires Authorization header support.
         # Keep this allowlist narrow to avoid leaking arbitrary caller headers.
-        ALLOWED_HEADER_PREFIXES = ("x-", "content-type", "authorization")
         if headers:
             for key, value in headers.items():
                 key_lower = key.lower()
-                if any(key_lower.startswith(p) for p in ALLOWED_HEADER_PREFIXES):
+                if any(
+                    key_lower.startswith(p) for p in ALLOWED_OUTBOUND_HEADER_PREFIXES
+                ):
                     request.add_header(key, value)
                 else:
                     logger.debug(f"Skipping disallowed header: {key}")
@@ -635,7 +690,14 @@ def safe_request_json(
                     }
 
         except urllib.error.HTTPError as e:
-            raise RuntimeError(f"HTTP error {e.code}: {e.reason}")
+            raise SafeIOHTTPError(
+                status_code=e.code,
+                reason=str(getattr(e, "reason", "HTTPError")),
+                method=current_method,
+                url=current_url,
+                headers=_headers_to_dict(getattr(e, "headers", None)),
+                body=_http_error_body_preview(e),
+            )
 
         except urllib.error.URLError as e:
             if isinstance(e.reason, SSRFError):
@@ -694,11 +756,12 @@ def safe_request_text_stream(
         request.add_header("User-Agent", f"ComfyUI-OpenClaw/{PACK_VERSION}")
         request.add_header("Content-Type", "application/json")
 
-        ALLOWED_HEADER_PREFIXES = ("x-", "content-type", "authorization", "accept")
         if headers:
             for key, value in headers.items():
                 key_lower = key.lower()
-                if any(key_lower.startswith(p) for p in ALLOWED_HEADER_PREFIXES):
+                if any(
+                    key_lower.startswith(p) for p in ALLOWED_OUTBOUND_HEADER_PREFIXES
+                ):
                     request.add_header(key, value)
                 else:
                     logger.debug(f"Skipping disallowed header: {key}")
@@ -744,7 +807,14 @@ def safe_request_text_stream(
             return
 
         except urllib.error.HTTPError as e:
-            raise RuntimeError(f"HTTP error {e.code}: {e.reason}")
+            raise SafeIOHTTPError(
+                status_code=e.code,
+                reason=str(getattr(e, "reason", "HTTPError")),
+                method=current_method,
+                url=current_url,
+                headers=_headers_to_dict(getattr(e, "headers", None)),
+                body=_http_error_body_preview(e),
+            )
         except urllib.error.URLError as e:
             if isinstance(e.reason, SSRFError):
                 raise e.reason

tests/test_r131_provider_adapter_contract.py

@@ -0,0 +1,128 @@
+"""
+R131: Provider adapter contract hardening tests.
+"""
+
+import unittest
+import warnings
+from unittest.mock import patch
+
+from services.provider_errors import ProviderHTTPError
+from services.providers import anthropic, openai_compat
+from services.safe_io import SafeIOHTTPError
+
+
+class TestR131ProviderRetryAfterPropagation(unittest.TestCase):
+    def test_openai_compat_maps_retry_after_from_structured_http_error(self):
+        with patch("services.providers.openai_compat.safe_request_json") as mock_safe:
+            mock_safe.side_effect = SafeIOHTTPError(
+                status_code=429,
+                reason="Too Many Requests",
+                method="POST",
+                url="https://api.example.com/v1/chat/completions",
+                headers={"Retry-After": "19"},
+                body='{"error":{"retry_after":5}}',
+            )
+
+            with self.assertRaises(ProviderHTTPError) as ctx:
+                openai_compat.make_request(
+                    base_url="https://api.example.com/v1",
+                    api_key="sk-test",
+                    messages=[{"role": "user", "content": "hi"}],
+                    model="gpt-4",
+                    allow_any_public_host=True,
+                )
+
+        self.assertEqual(ctx.exception.status_code, 429)
+        self.assertEqual(ctx.exception.retry_after, 19)
+
+    def test_openai_compat_stream_maps_retry_after_from_error_body(self):
+        with patch(
+            "services.providers.openai_compat.safe_request_text_stream"
+        ) as mock_stream:
+            mock_stream.side_effect = SafeIOHTTPError(
+                status_code=429,
+                reason="Too Many Requests",
+                method="POST",
+                url="https://api.example.com/v1/chat/completions",
+                headers={},
+                body='{"retry_after":27}',
+            )
+
+            with self.assertRaises(ProviderHTTPError) as ctx:
+                openai_compat.make_request_stream(
+                    base_url="https://api.example.com/v1",
+                    api_key="sk-test",
+                    messages=[{"role": "user", "content": "hi"}],
+                    model="gpt-4",
+                    allow_any_public_host=True,
+                )
+
+        self.assertEqual(ctx.exception.status_code, 429)
+        self.assertEqual(ctx.exception.retry_after, 27)
+
+    def test_anthropic_maps_retry_after_from_headers(self):
+        with patch("services.providers.anthropic.safe_request_json") as mock_safe:
+            mock_safe.side_effect = SafeIOHTTPError(
+                status_code=429,
+                reason="Too Many Requests",
+                method="POST",
+                url="https://api.anthropic.com/v1/messages",
+                headers={"x-ratelimit-reset-after": "13"},
+                body='{"retry_after":99}',
+            )
+
+            with self.assertRaises(ProviderHTTPError) as ctx:
+                anthropic.make_request(
+                    base_url="https://api.anthropic.com",
+                    api_key="sk-ant-test",
+                    messages=[{"role": "user", "content": "hi"}],
+                    model="claude-sonnet-4",
+                    allow_any_public_host=True,
+                )
+
+        self.assertEqual(ctx.exception.status_code, 429)
+        self.assertEqual(ctx.exception.retry_after, 13)
+
+
+class TestR131SchemaSanitizerImportFallback(unittest.TestCase):
+    def test_build_chat_request_uses_fallback_import_when_package_context_missing(self):
+        tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "tool_ping",
+                    "description": "Ping test tool",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "q": {"type": "string"},
+                        },
+                    },
+                },
+            }
+        ]
+        original_package = openai_compat.__package__
+        try:
+            openai_compat.__package__ = ""
+            with warnings.catch_warnings():
+                warnings.filterwarnings(
+                    "ignore",
+                    category=DeprecationWarning,
+                    message="__package__ != __spec__.parent",
+                )
+                payload = openai_compat.build_chat_request(
+                    messages=[{"role": "user", "content": "hi"}],
+                    model="gpt-4",
+                    tools=tools,
+                    tool_choice="auto",
+                )
+        finally:
+            openai_compat.__package__ = original_package
+
+        self.assertIn("tools", payload)
+        self.assertEqual(payload.get("tool_choice"), "auto")
+        self.assertEqual(payload["tools"][0]["function"]["name"], "tool_ping")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_safe_io.py

@@ -3,12 +3,15 @@ import shutil
 import sys
 import tempfile
 import unittest
+import urllib.error
+from io import BytesIO
 from unittest.mock import MagicMock, patch
 
 sys.path.append(os.getcwd())
 
 from services.safe_io import (
     PathTraversalError,
+    SafeIOHTTPError,
     SSRFError,
     _normalize_host,
     is_private_ip,
@@ -16,6 +19,7 @@ from services.safe_io import (
     safe_read_bytes,
     safe_read_text,
     safe_request_json,
+    safe_request_text_stream,
     safe_write_text,
     validate_outbound_url,
 )
@@ -227,6 +231,119 @@ class TestURLSafety(unittest.TestCase):
         )
         self.assertEqual(out["ok"], True)
 
+    @patch("services.safe_io._build_pinned_opener")
+    @patch("services.safe_io.validate_outbound_url")
+    def test_safe_request_json_http_error_preserves_retry_headers_and_body(
+        self, mock_validate, mock_build
+    ):
+        """HTTP errors should surface structured metadata for provider retry logic."""
+        mock_validate.return_value = ("https", "example.com", 443, ["93.184.216.34"])
+
+        mock_opener = MagicMock()
+        mock_opener.open.side_effect = urllib.error.HTTPError(
+            url="https://example.com/fail",
+            code=429,
+            msg="Too Many Requests",
+            hdrs={"Retry-After": "17", "Content-Type": "application/json"},
+            fp=BytesIO(b'{"error":{"retry_after":11}}'),
+        )
+        mock_build.return_value = mock_opener
+
+        with self.assertRaises(SafeIOHTTPError) as ctx:
+            safe_request_json(
+                method="POST",
+                url="https://example.com/fail",
+                json_body={"x": 1},
+                allow_hosts={"example.com"},
+            )
+
+        self.assertEqual(ctx.exception.status_code, 429)
+        self.assertEqual(ctx.exception.headers.get("Retry-After"), "17")
+        self.assertIn("retry_after", ctx.exception.body or "")
+
+    @patch("services.safe_io._build_pinned_opener")
+    @patch("services.safe_io.validate_outbound_url")
+    def test_safe_request_json_accept_header_is_allowed(
+        self, mock_validate, mock_build
+    ):
+        """JSON request path should allow Accept header (parity with stream path)."""
+        mock_validate.return_value = ("https", "example.com", 443, ["93.184.216.34"])
+
+        mock_response = MagicMock()
+        mock_response.getcode.return_value = 200
+        mock_response.read.return_value = b'{"ok": true}'
+
+        mock_opener = MagicMock()
+        mock_opener.open.return_value.__enter__.return_value = mock_response
+        mock_build.return_value = mock_opener
+
+        out = safe_request_json(
+            method="POST",
+            url="https://example.com/accept",
+            json_body={"x": 1},
+            headers={
+                "Accept": "application/json",
+                "X-Test": "ok",
+                "Bad-Header": "blocked",
+            },
+            allow_hosts={"example.com"},
+        )
+
+        self.assertEqual(out["ok"], True)
+        request_arg = mock_opener.open.call_args.args[0]
+        header_map = {k.lower(): v for k, v in request_arg.header_items()}
+        self.assertEqual(header_map.get("accept"), "application/json")
+        self.assertEqual(header_map.get("x-test"), "ok")
+        self.assertNotIn("bad-header", header_map)
+
+    @patch("services.safe_io._build_pinned_opener")
+    @patch("services.safe_io.validate_outbound_url")
+    def test_safe_request_text_stream_accept_header_is_allowed(
+        self, mock_validate, mock_build
+    ):
+        """Stream request path should share same allowed-header contract."""
+        mock_validate.return_value = ("https", "example.com", 443, ["93.184.216.34"])
+
+        class _FakeStreamResponse:
+            def __init__(self):
+                self.headers = {}
+                self._lines = [b"data: one\n", b""]
+
+            def getcode(self):
+                return 200
+
+            def readline(self, _max_bytes):
+                return self._lines.pop(0)
+
+            def close(self):
+                return None
+
+        fake_response = _FakeStreamResponse()
+        mock_opener = MagicMock()
+        mock_opener.open.return_value = fake_response
+        mock_build.return_value = mock_opener
+
+        lines = list(
+            safe_request_text_stream(
+                method="POST",
+                url="https://example.com/stream",
+                json_body={"x": 1},
+                headers={
+                    "Accept": "text/event-stream",
+                    "X-Test": "ok",
+                    "Bad-Header": "blocked",
+                },
+                allow_hosts={"example.com"},
+            )
+        )
+
+        self.assertEqual(lines, ["data: one\n"])
+        request_arg = mock_opener.open.call_args.args[0]
+        header_map = {k.lower(): v for k, v in request_arg.header_items()}
+        self.assertEqual(header_map.get("accept"), "text/event-stream")
+        self.assertEqual(header_map.get("x-test"), "ok")
+        self.assertNotIn("bad-header", header_map)
+
     def test_host_normalization_case(self):
         """Test host normalization is case-insensitive."""
         self.assertEqual(_normalize_host("Example.COM"), "example.com")