fix(whatsapp): inspect remote media before dispatch

nanobot/channels/whatsapp/runtime.py

@@ -12,7 +12,9 @@ from collections import OrderedDict
 from contextlib import suppress
 from pathlib import Path
 from typing import Any, Literal, NamedTuple, cast
+from urllib.parse import urlparse
 
+import httpx
 from pydantic import Field
 
 from nanobot.bus.events import OutboundMessage
@@ -20,6 +22,7 @@ from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.config.paths import get_media_dir, get_runtime_subdir
 from nanobot.config.schema import Base
+from nanobot.security.network import PinnedDNSAsyncTransport
 
 
 class WhatsAppConfig(Base):
@@ -39,6 +42,8 @@ class _NeonizeAPI(NamedTuple):
     MessageEv: Any
     PairStatusEv: Any
     build_jid: Any
+    detect_mime: Any
+    detect_buffer: Any
 
 
 class _MediaInfo(NamedTuple):
@@ -52,6 +57,15 @@ class _MediaInfo(NamedTuple):
 _NEONIZE_API: _NeonizeAPI | None = None
 _JID_RE = re.compile(r"^(?P<user>[^@]+)@(?P<server>[^@]+)$")
 _LEGACY_BRIDGE_CONFIG_FIELDS = ("bridgeUrl", "bridgeToken", "bridge_url", "bridge_token")
+_REMOTE_MEDIA_MAX_BYTES = 32 * 1024 * 1024
+_REMOTE_MEDIA_MAX_REDIRECTS = 5
+_REMOTE_MEDIA_TIMEOUT_SECONDS = 120.0
+# OGG is intentionally excluded: WhatsApp accepts only mono Opus, which MIME sniffing cannot prove.
+_DIRECT_AUDIO_MIMETYPES = {"audio/aac", "audio/amr", "audio/mp4", "audio/mpeg"}
+_MIMETYPE_ALIASES = {
+    "audio/x-hx-aac-adts": "audio/aac",
+    "audio/x-m4a": "audio/mp4",
+}
 
 
 def _default_database_path() -> Path:
@@ -68,9 +82,15 @@ def _load_neonize() -> _NeonizeAPI:
         return _NEONIZE_API
 
     try:
+        import magic
         from neonize.aioze.client import NewAClient
         from neonize.aioze.events import ConnectedEv, DisconnectedEv, MessageEv, PairStatusEv
         from neonize.utils.jid import build_jid
+
+        detect_mime = getattr(magic, "from_file", None)
+        detect_buffer = getattr(magic, "from_buffer", None)
+        if not callable(detect_mime) or not callable(detect_buffer):
+            raise ImportError("python-magic does not expose from_file/from_buffer")
     except ImportError as exc:
         raise RuntimeError(
             "WhatsApp dependencies not installed. Run: nanobot plugins enable whatsapp"
@@ -83,6 +103,8 @@ def _load_neonize() -> _NeonizeAPI:
         MessageEv=MessageEv,
         PairStatusEv=PairStatusEv,
         build_jid=build_jid,
+        detect_mime=detect_mime,
+        detect_buffer=detect_buffer,
     )
     return _NEONIZE_API
 
@@ -417,23 +439,84 @@ class WhatsAppChannel(BaseChannel):
         return api.build_jid(user, server)
 
     async def _send_media(self, client: Any, to: Any, media_path: str) -> None:
-        path = str(Path(media_path).expanduser())
-        mime, _ = mimetypes.guess_type(path)
-        mimetype = mime or "application/octet-stream"
+        source: str | bytes
+        if media_path.startswith(("http://", "https://")):
+            source = await self._fetch_remote_media(media_path)
+            filename = Path(urlparse(media_path).path).name or "attachment"
+        else:
+            source = str(Path(media_path).expanduser())
+            filename = Path(source).name
+
+        mimetype = self._detect_mimetype(source)
         if mimetype.startswith("image/"):
-            await client.send_image(to, path)
+            await client.send_image(to, source)
         elif mimetype.startswith("video/"):
-            await client.send_video(to, path)
-        elif mimetype.startswith("audio/"):
-            await client.send_audio(to, path)
+            await client.send_video(to, source)
+        elif mimetype in _DIRECT_AUDIO_MIMETYPES:
+            await client.send_audio(to, source)
         else:
             await client.send_document(
                 to,
-                path,
-                filename=Path(path).name,
+                source,
+                filename=filename,
                 mimetype=mimetype,
             )
 
+    async def _fetch_remote_media(self, url: str) -> bytes:
+        timeout = httpx.Timeout(_REMOTE_MEDIA_TIMEOUT_SECONDS, connect=10.0)
+        async with httpx.AsyncClient(
+            transport=PinnedDNSAsyncTransport(),
+            follow_redirects=True,
+            max_redirects=_REMOTE_MEDIA_MAX_REDIRECTS,
+            timeout=timeout,
+            trust_env=False,
+        ) as http:
+            async with http.stream("GET", url) as response:
+                response.raise_for_status()
+                declared_size = response.headers.get("content-length")
+                if (
+                    declared_size
+                    and declared_size.isdigit()
+                    and int(declared_size) > _REMOTE_MEDIA_MAX_BYTES
+                ):
+                    raise ValueError(
+                        f"Remote WhatsApp media exceeds the {_REMOTE_MEDIA_MAX_BYTES}-byte limit"
+                    )
+
+                chunks: list[bytes] = []
+                total = 0
+                async for chunk in response.aiter_bytes():
+                    total += len(chunk)
+                    if total > _REMOTE_MEDIA_MAX_BYTES:
+                        raise ValueError(
+                            f"Remote WhatsApp media exceeds the {_REMOTE_MEDIA_MAX_BYTES}-byte limit"
+                        )
+                    chunks.append(chunk)
+        return b"".join(chunks)
+
+    def _detect_mimetype(self, source: str | bytes) -> str:
+        try:
+            api = _load_neonize()
+            detected = (
+                api.detect_buffer(source, mime=True)
+                if isinstance(source, bytes)
+                else api.detect_mime(source, mime=True)
+            )
+        except Exception as exc:
+            label = f"{len(source)} downloaded bytes" if isinstance(source, bytes) else source
+            self.logger.debug("Failed to inspect WhatsApp media {}: {}", label, exc)
+            detected = None
+
+        if isinstance(detected, str) and "/" in detected:
+            mimetype = detected.partition(";")[0].strip().lower()
+            return _MIMETYPE_ALIASES.get(mimetype, mimetype)
+
+        if isinstance(source, bytes):
+            return "application/octet-stream"
+
+        guessed, _ = mimetypes.guess_type(source)
+        return guessed or "application/octet-stream"
+
     def _register_handlers(
         self,
         client: Any,

nanobot/channels/whatsapp/tests/test_whatsapp_channel.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+import mimetypes
 import sys
 import types
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
+import httpx
 import pytest
 
 import nanobot.channels.whatsapp.runtime as whatsapp_module
@@ -78,7 +80,21 @@ def _make_channel(config: dict | None = None) -> WhatsAppChannel:
     return ch
 
 
-def _patch_neonize_api(monkeypatch) -> None:
+def _make_send_client() -> SimpleNamespace:
+    return SimpleNamespace(
+        send_message=AsyncMock(),
+        send_image=AsyncMock(),
+        send_video=AsyncMock(),
+        send_audio=AsyncMock(),
+        send_document=AsyncMock(),
+    )
+
+
+def _patch_neonize_api(monkeypatch, detect_mime=None, detect_buffer=None) -> None:
+    detect_mime = detect_mime or (
+        lambda path, *, mime: mimetypes.guess_type(path)[0] or "application/octet-stream"
+    )
+    detect_buffer = detect_buffer or (lambda data, *, mime: "application/octet-stream")
     monkeypatch.setattr(
         whatsapp_module,
         "_NEONIZE_API",
@@ -89,6 +105,8 @@ def _patch_neonize_api(monkeypatch) -> None:
             MessageEv=object(),
             PairStatusEv=object(),
             build_jid=lambda user, server="s.whatsapp.net": (user, server),
+            detect_mime=detect_mime,
+            detect_buffer=detect_buffer,
         ),
     )
 
@@ -178,13 +196,7 @@ async def test_login_fails_when_connect_task_fails(monkeypatch) -> None:
 @pytest.mark.asyncio
 async def test_send_text_uses_neonize_send_message(monkeypatch) -> None:
     _patch_neonize_api(monkeypatch)
-    client = SimpleNamespace(
-        send_message=AsyncMock(),
-        send_image=AsyncMock(),
-        send_video=AsyncMock(),
-        send_audio=AsyncMock(),
-        send_document=AsyncMock(),
-    )
+    client = _make_send_client()
     ch = _make_channel()
     ch._client = client
     ch._connected = True
@@ -197,13 +209,7 @@ async def test_send_text_uses_neonize_send_message(monkeypatch) -> None:
 @pytest.mark.asyncio
 async def test_send_media_dispatches_by_mimetype(monkeypatch) -> None:
     _patch_neonize_api(monkeypatch)
-    client = SimpleNamespace(
-        send_message=AsyncMock(),
-        send_image=AsyncMock(),
-        send_video=AsyncMock(),
-        send_audio=AsyncMock(),
-        send_document=AsyncMock(),
-    )
+    client = _make_send_client()
     ch = _make_channel()
     ch._client = client
     ch._connected = True
@@ -213,14 +219,14 @@ async def test_send_media_dispatches_by_mimetype(monkeypatch) -> None:
             channel="whatsapp",
             chat_id="12345@s.whatsapp.net",
             content="",
-            media=["photo.jpg", "clip.mp4", "voice.ogg", "report.pdf"],
+            media=["photo.jpg", "clip.mp4", "voice.mp3", "report.pdf"],
         )
     )
 
     jid = ("12345", "s.whatsapp.net")
     client.send_image.assert_awaited_once_with(jid, "photo.jpg")
     client.send_video.assert_awaited_once_with(jid, "clip.mp4")
-    client.send_audio.assert_awaited_once_with(jid, "voice.ogg")
+    client.send_audio.assert_awaited_once_with(jid, "voice.mp3")
     client.send_document.assert_awaited_once_with(
         jid,
         "report.pdf",
@@ -229,6 +235,191 @@ async def test_send_media_dispatches_by_mimetype(monkeypatch) -> None:
     )
 
 
+@pytest.mark.asyncio
+async def test_send_mislabeled_audio_as_document(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch, detect_mime=lambda path, *, mime: "audio/x-wav")
+    client = _make_send_client()
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id="12345@s.whatsapp.net",
+            content="",
+            media=["recording.mpeg"],
+        )
+    )
+
+    jid = ("12345", "s.whatsapp.net")
+    client.send_document.assert_awaited_once_with(
+        jid,
+        "recording.mpeg",
+        filename="recording.mpeg",
+        mimetype="audio/x-wav",
+    )
+    client.send_video.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_send_remote_mislabeled_audio_as_document(monkeypatch) -> None:
+    payload = b"remote wav payload"
+    media_url = "https://cdn.example/recording.mpeg?token=secret"
+
+    def handle_request(request: httpx.Request) -> httpx.Response:
+        assert str(request.url) == media_url
+        return httpx.Response(200, content=payload)
+
+    monkeypatch.setattr(
+        whatsapp_module,
+        "PinnedDNSAsyncTransport",
+        lambda: httpx.MockTransport(handle_request),
+    )
+
+    def detect_buffer(data: bytes, *, mime: bool) -> str:
+        assert data == payload
+        assert mime is True
+        return "audio/x-wav"
+
+    _patch_neonize_api(
+        monkeypatch,
+        detect_buffer=detect_buffer,
+    )
+    client = _make_send_client()
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id="12345@s.whatsapp.net",
+            content="",
+            media=[media_url],
+        )
+    )
+
+    jid = ("12345", "s.whatsapp.net")
+    client.send_document.assert_awaited_once_with(
+        jid,
+        payload,
+        filename="recording.mpeg",
+        mimetype="audio/x-wav",
+    )
+    client.send_video.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_send_remote_media_blocks_private_url(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch)
+    client = _make_send_client()
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    with pytest.raises(httpx.RequestError, match="private/internal"):
+        await ch.send(
+            OutboundMessage(
+                channel="whatsapp",
+                chat_id="12345@s.whatsapp.net",
+                content="",
+                media=["http://127.0.0.1/recording.mpeg"],
+            )
+        )
+
+    client.send_video.assert_not_awaited()
+    client.send_document.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_send_remote_media_enforces_download_limit(monkeypatch) -> None:
+    monkeypatch.setattr(whatsapp_module, "_REMOTE_MEDIA_MAX_BYTES", 3)
+    monkeypatch.setattr(
+        whatsapp_module,
+        "PinnedDNSAsyncTransport",
+        lambda: httpx.MockTransport(lambda request: httpx.Response(200, content=b"1234")),
+    )
+    _patch_neonize_api(monkeypatch)
+    client = _make_send_client()
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    with pytest.raises(ValueError, match="exceeds the 3-byte limit"):
+        await ch.send(
+            OutboundMessage(
+                channel="whatsapp",
+                chat_id="12345@s.whatsapp.net",
+                content="",
+                media=["https://cdn.example/recording.mpeg"],
+            )
+        )
+
+    client.send_video.assert_not_awaited()
+    client.send_document.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_send_unsupported_ogg_audio_as_document(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch, detect_mime=lambda path, *, mime: "audio/ogg")
+    client = _make_send_client()
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id="12345@s.whatsapp.net",
+            content="",
+            media=["voice.ogg"],
+        )
+    )
+
+    jid = ("12345", "s.whatsapp.net")
+    client.send_document.assert_awaited_once_with(
+        jid,
+        "voice.ogg",
+        filename="voice.ogg",
+        mimetype="audio/ogg",
+    )
+    client.send_audio.assert_not_awaited()
+
+
+@pytest.mark.parametrize(
+    ("detected_mimetype", "filename"),
+    [
+        ("audio/x-m4a", "recording.m4a"),
+        ("audio/x-hx-aac-adts", "recording.aac"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_send_supported_audio_magic_aliases_inline(
+    monkeypatch, detected_mimetype: str, filename: str
+) -> None:
+    _patch_neonize_api(
+        monkeypatch,
+        detect_mime=lambda path, *, mime: detected_mimetype,
+    )
+    client = _make_send_client()
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id="12345@s.whatsapp.net",
+            content="",
+            media=[filename],
+        )
+    )
+
+    client.send_audio.assert_awaited_once_with(("12345", "s.whatsapp.net"), filename)
+    client.send_document.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_send_when_disconnected_raises() -> None:
     ch = _make_channel()

nanobot/providers/factory.py

@@ -24,6 +24,7 @@ class ProviderSnapshot:
 @dataclass(frozen=True)
 class _ProviderSetup:
     model: str
+    provider_name: str
     provider_config: ProviderConfig | None
     spec: ProviderSpec | None
     backend: str
@@ -99,6 +100,7 @@ def _resolve_provider_setup(
 
     return _ProviderSetup(
         model=model,
+        provider_name=provider_name,
         provider_config=p,
         spec=spec,
         backend=backend,
@@ -134,6 +136,7 @@ def _make_provider_core(
         model=model,
     )
     model = setup.model
+    provider_name = setup.provider_name
     p = setup.provider_config
     spec = setup.spec
     backend = setup.backend
@@ -198,7 +201,7 @@ def _make_provider_core(
             extra_headers=_provider_extra_headers(spec, p),
             spec=spec,
             extra_body=p.extra_body if p else None,
-            api_type=p.api_type if p else "auto",
+            api_type=p.api_type if p and provider_name == "openai" else "auto",
             extra_query=p.extra_query if p else None,
             proxy=p.proxy if p else None,
         )

nanobot/providers/openai_compat_provider.py

@@ -49,7 +49,7 @@ from nanobot.providers.openai_responses import (
 if TYPE_CHECKING:
     from openai import AsyncOpenAI as AsyncOpenAIType
 
-    from nanobot.providers.registry import ProviderSpec, ResponsesCapabilities
+    from nanobot.providers.registry import ProviderSpec
 
 # Module-level placeholder — set lazily by _ensure_client on first real
 # use, or replaced by tests via ``patch(...)``.  Kept as a plain name so
@@ -470,12 +470,7 @@ class OpenAICompatProvider(LLMProvider):
         self.extra_headers = extra_headers or {}
         self._spec = spec
         self._extra_body = extra_body or {}
-        responses = spec.responses if spec is not None else None
-        self._api_type = (
-            api_type
-            if responses is not None and responses.allows_api_type_override
-            else "auto"
-        )
+        self._api_type = api_type if spec and spec.name == "openai" else "auto"
         self._extra_query = extra_query or {}
         self._proxy = proxy or None
         self._native_compaction_available = True
@@ -966,33 +961,39 @@ class OpenAICompatProvider(LLMProvider):
         """Choose Responses for providers/models that explicitly support it."""
         if self._api_type == "chat_completions":
             return False
-        capabilities = self._responses_capabilities()
-        if capabilities is None:
-            return False
+        spec_name = self._spec.name if self._spec is not None else None
         model_name = self._request_model_name(model or self.default_model).lower()
+        supported_models = {
+            supported.lower()
+            for supported in getattr(self._spec, "responses_models", ())
+        }
+        model_responses = any(
+            model_name == supported or model_name.endswith(f"/{supported}")
+            for supported in supported_models
+        )
+        provider_responses = spec_name in ("openai", "github_copilot")
+        if not provider_responses and not model_responses:
+            return False
         if self._api_type == "responses":
             # Explicit configuration means Responses is mandatory; do not
             # consult the circuit breaker or fall back to Chat Completions.
             return True
-        if (
-            capabilities.requires_direct_openai_base
-            and not _is_direct_openai_base(self._effective_base)
-        ):
-            return False
+        if provider_responses and (self._spec is None or self._spec.name != "github_copilot"):
+            if not _is_direct_openai_base(self._effective_base):
+                return False
 
-        explicitly_supported = capabilities.matches_model(model_name)
-        wants_auto_route = capabilities.auto_route and (
-            (reasoning_effort is not None and reasoning_effort.lower() != "none")
-            or any(token in model_name for token in ("gpt-5", "o1", "o3", "o4"))
-        )
-        if not explicitly_supported and not wants_auto_route:
+        wants = False
+        if model_responses:
+            wants = True
+        elif reasoning_effort and reasoning_effort.lower() != "none":
+            wants = True
+        elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")):
+            wants = True
+        if not wants:
             return False
 
         return self._responses_circuit_allows_probe(model, reasoning_effort)
 
-    def _responses_capabilities(self) -> ResponsesCapabilities | None:
-        return self._spec.responses if self._spec is not None else None
-
     def _responses_state_provider(self) -> str:
         spec_name = self._spec.name if self._spec is not None else "custom"
         effective_base = self._effective_base or "https://api.openai.com/v1"
@@ -1015,20 +1016,14 @@ class OpenAICompatProvider(LLMProvider):
     def supports_native_compaction(self, model: str | None = None) -> bool:
         """Enable server compaction only on direct OpenAI Responses endpoints."""
         _ = model
-        capabilities = self._responses_capabilities()
         if (
             not self._native_compaction_available
             or self._api_type == "chat_completions"
-            or capabilities is None
-            or not capabilities.supports_native_compaction
         ):
             return False
-        if (
-            capabilities.requires_direct_openai_base
-            and not _is_direct_openai_base(self._effective_base)
-        ):
+        if self._spec is not None and self._spec.name != "openai":
             return False
-        return True
+        return _is_direct_openai_base(self._effective_base)
 
     def _responses_circuit_allows_probe(
         self,
@@ -1116,10 +1111,7 @@ class OpenAICompatProvider(LLMProvider):
                     self._sanitize_empty_content(sanitized_state.pending_messages)
                 )
             )
-        capabilities = self._responses_capabilities()
-        preserve_reasoning = (
-            capabilities is not None and capabilities.reasoning_replay == "plaintext"
-        )
+        preserve_reasoning = bool(self._spec and self._spec.name == "deepseek")
         instructions, input_items, replayed = prepare_responses_input(
             sanitized_messages,
             state=sanitized_state,
@@ -1150,15 +1142,10 @@ class OpenAICompatProvider(LLMProvider):
                 "compact_threshold": compact_threshold,
             }]
 
-        supports_temperature = self._supports_temperature(model_name, reasoning_effort)
-        if supports_temperature:
+        if self._supports_temperature(model_name, reasoning_effort):
             body["temperature"] = temperature
 
-        if (
-            not supports_temperature
-            and capabilities is not None
-            and capabilities.reasoning_replay == "encrypted"
-        ):
+        if not self._supports_temperature(model_name, reasoning_effort) and not preserve_reasoning:
             body["include"] = ["reasoning.encrypted_content"]
         if reasoning_effort and reasoning_effort.lower() != "none":
             body["reasoning"] = {"effort": reasoning_effort}
@@ -1779,8 +1766,10 @@ class OpenAICompatProvider(LLMProvider):
                     self._record_responses_success(model, reasoning_effort)
                     return result
                 except Exception as responses_error:
-                    capabilities = self._responses_capabilities()
-                    if capabilities is not None and not capabilities.allows_chat_fallback:
+                    if self._spec and self._spec.name == "github_copilot":
+                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                        # falling back to /chat/completions cannot succeed and would
+                        # hide the real error.
                         raise
                     if self._api_type == "responses":
                         raise
@@ -1873,8 +1862,10 @@ class OpenAICompatProvider(LLMProvider):
                         )
                     return result
                 except Exception as responses_error:
-                    capabilities = self._responses_capabilities()
-                    if capabilities is not None and not capabilities.allows_chat_fallback:
+                    if self._spec and self._spec.name == "github_copilot":
+                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                        # falling back to /chat/completions cannot succeed and would
+                        # hide the real error.
                         raise
                     if self._api_type == "responses":
                         raise

nanobot/providers/registry.py

@@ -13,7 +13,7 @@ Every entry writes out all fields so you can copy-paste as a template.
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Literal
+from typing import Any
 
 from pydantic.alias_generators import to_snake
 
@@ -28,32 +28,6 @@ class ProviderModelSpec:
     context_window: int | None = None
 
 
-@dataclass(frozen=True)
-class ResponsesCapabilities:
-    """Provider capabilities for the shared OpenAI Responses execution path.
-
-    ``reasoning_replay`` selects whether multi-turn reasoning is retained as
-    encrypted server content, plaintext local history, or not requested.
-    """
-
-    models: tuple[str, ...] = ()
-    auto_route: bool = False
-    requires_direct_openai_base: bool = False
-    allows_api_type_override: bool = False
-    reasoning_replay: Literal["none", "encrypted", "plaintext"] = "none"
-    supports_native_compaction: bool = False
-    allows_chat_fallback: bool = True
-
-    def matches_model(self, model: str) -> bool:
-        """Return whether *model* is explicitly routed through Responses."""
-        model_name = model.lower()
-        return any(
-            model_name == supported.lower()
-            or model_name.endswith(f"/{supported.lower()}")
-            for supported in self.models
-        )
-
-
 @dataclass(frozen=True)
 class ProviderSpec:
     """One LLM provider's metadata. See PROVIDERS below for real examples.
@@ -137,8 +111,10 @@ class ProviderSpec:
     # Substring match against the wire model name (lowercased).
     implicit_reasoning_models: tuple[str, ...] = ()
 
-    # Capabilities for providers/models served through the shared Responses path.
-    responses: ResponsesCapabilities | None = None
+    # Models that expose the OpenAI Responses wire format.  This is model-level
+    # because providers may add Responses support incrementally (DeepSeek V4
+    # Flash is supported before V4 Pro).
+    responses_models: tuple[str, ...] = ()
 
     # When the model returns content as a list of {"type":"thinking",...} +
     # {"type":"text",...} blocks, extract the thinking text into
@@ -397,13 +373,6 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         display_name="OpenAI",
         backend="openai_compat",
         supports_max_completion_tokens=True,
-        responses=ResponsesCapabilities(
-            auto_route=True,
-            requires_direct_openai_base=True,
-            allows_api_type_override=True,
-            reasoning_replay="encrypted",
-            supports_native_compaction=True,
-        ),
     ),
     # OpenAI Codex: OAuth-based, dedicated provider
     ProviderSpec(
@@ -487,11 +456,6 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         strip_model_prefix=True,
         is_oauth=True,
         supports_max_completion_tokens=True,
-        responses=ResponsesCapabilities(
-            auto_route=True,
-            reasoning_replay="encrypted",
-            allows_chat_fallback=False,
-        ),
     ),
     # DeepSeek: OpenAI-compatible at api.deepseek.com
     ProviderSpec(
@@ -502,10 +466,7 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         backend="openai_compat",
         default_api_base="https://api.deepseek.com",
         thinking_style="thinking_type",
-        responses=ResponsesCapabilities(
-            models=("deepseek-v4-flash",),
-            reasoning_replay="plaintext",
-        ),
+        responses_models=("deepseek-v4-flash",),
     ),
     # Gemini: Google's OpenAI-compatible endpoint
     ProviderSpec(

tests/providers/test_github_copilot_routing.py

@@ -48,7 +48,6 @@ def test_build_responses_body_strips_github_copilot_prefix():
         provider_context=ProviderCallContext(context_window_tokens=128_000),
     )
     assert body["model"] == "gpt-5.4-mini"
-    assert body["include"] == ["reasoning.encrypted_content"]
     assert "context_management" not in body
 
 

tests/providers/test_responses_circuit_breaker.py

@@ -10,11 +10,6 @@ from nanobot.providers.openai_compat_provider import (
     _RESPONSES_PROBE_INTERVAL_S,
     OpenAICompatProvider,
 )
-from nanobot.providers.registry import (
-    ProviderSpec,
-    ResponsesCapabilities,
-    find_by_name,
-)
 
 
 @pytest.fixture()
@@ -22,7 +17,7 @@ def provider():
     """A direct-OpenAI provider with Responses API support."""
     p = OpenAICompatProvider.__new__(OpenAICompatProvider)
     p.default_model = "gpt-5"
-    p._spec = find_by_name("openai")
+    p._spec = type("Spec", (), {"name": "openai"})()
     p._effective_base = "https://api.openai.com/v1"
     p._api_type = "auto"
     p._responses_failures = {}
@@ -35,7 +30,12 @@ def test_responses_api_available_by_default(provider):
 
 
 def test_deepseek_v4_flash_uses_responses_by_model(provider):
-    provider._spec = find_by_name("deepseek")
+    provider._spec = type("Spec", (), {
+        "name": "deepseek",
+        "responses_models": ("deepseek-v4-flash",),
+        "strip_model_prefix": False,
+        "strip_model_prefixes": (),
+    })()
     provider._effective_base = "https://api.deepseek.com"
     provider.default_model = "deepseek-v4-flash"
 
@@ -44,48 +44,17 @@ def test_deepseek_v4_flash_uses_responses_by_model(provider):
 
 
 def test_deepseek_v4_flash_matches_provider_prefixed_model(provider):
-    provider._spec = find_by_name("deepseek")
+    provider._spec = type("Spec", (), {
+        "name": "deepseek",
+        "responses_models": ("deepseek-v4-flash",),
+        "strip_model_prefix": False,
+        "strip_model_prefixes": (),
+    })()
     provider._effective_base = "https://api.deepseek.com"
 
     assert provider._should_use_responses_api("deepseek/deepseek-v4-flash", None) is True
 
 
-def test_responses_behavior_is_declared_by_capabilities(provider):
-    provider._spec = ProviderSpec(
-        name="example",
-        keywords=("example",),
-        env_key="EXAMPLE_API_KEY",
-        responses=ResponsesCapabilities(
-            models=("example-o3",),
-            reasoning_replay="plaintext",
-        ),
-    )
-    provider._effective_base = "https://example.test"
-
-    assert provider._should_use_responses_api("example-o3", None) is True
-
-    body = provider._build_responses_body(
-        messages=[
-            {"role": "user", "content": "question"},
-            {
-                "role": "assistant",
-                "reasoning_content": "think first",
-                "content": "answer",
-            },
-            {"role": "user", "content": "follow-up"},
-        ],
-        tools=None,
-        model="example-o3",
-        max_tokens=100,
-        temperature=0.1,
-        reasoning_effort="high",
-        tool_choice=None,
-    )
-
-    assert {"type": "reasoning", "content": "think first"} in body["input"]
-    assert "include" not in body
-
-
 def test_direct_openai_enables_server_compaction(provider):
     provider._extra_body = {}
 
@@ -104,7 +73,6 @@ def test_direct_openai_enables_server_compaction(provider):
         "type": "compaction",
         "compact_threshold": 70_000,
     }]
-    assert body["include"] == ["reasoning.encrypted_content"]
 
 
 def test_api_type_chat_completions_disables_responses(provider):
@@ -128,7 +96,7 @@ def test_api_type_responses_ignores_circuit_breaker(provider):
 
 
 def test_api_type_responses_does_not_force_non_openai(provider):
-    provider._spec = find_by_name("custom")
+    provider._spec = type("Spec", (), {"name": "custom"})()
     provider._api_type = "responses"
 
     assert provider._should_use_responses_api("gpt-4o", None) is False