refactor(providers): declare Responses capabilities

nanobot/channels/whatsapp/runtime.py

@@ -12,9 +12,7 @@ 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
@@ -22,7 +20,6 @@ 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):
@@ -42,8 +39,6 @@ class _NeonizeAPI(NamedTuple):
     MessageEv: Any
     PairStatusEv: Any
     build_jid: Any
-    detect_mime: Any
-    detect_buffer: Any
 
 
 class _MediaInfo(NamedTuple):
@@ -57,15 +52,6 @@ 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:
@@ -82,15 +68,9 @@ 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"
@@ -103,8 +83,6 @@ def _load_neonize() -> _NeonizeAPI:
         MessageEv=MessageEv,
         PairStatusEv=PairStatusEv,
         build_jid=build_jid,
-        detect_mime=detect_mime,
-        detect_buffer=detect_buffer,
     )
     return _NEONIZE_API
 
@@ -439,84 +417,23 @@ class WhatsAppChannel(BaseChannel):
         return api.build_jid(user, server)
 
     async def _send_media(self, client: Any, to: Any, media_path: str) -> None:
-        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)
+        path = str(Path(media_path).expanduser())
+        mime, _ = mimetypes.guess_type(path)
+        mimetype = mime or "application/octet-stream"
         if mimetype.startswith("image/"):
-            await client.send_image(to, source)
+            await client.send_image(to, path)
         elif mimetype.startswith("video/"):
-            await client.send_video(to, source)
-        elif mimetype in _DIRECT_AUDIO_MIMETYPES:
-            await client.send_audio(to, source)
+            await client.send_video(to, path)
+        elif mimetype.startswith("audio/"):
+            await client.send_audio(to, path)
         else:
             await client.send_document(
                 to,
-                source,
-                filename=filename,
+                path,
+                filename=Path(path).name,
                 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,13 +1,11 @@
 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
@@ -80,21 +78,7 @@ def _make_channel(config: dict | None = None) -> WhatsAppChannel:
     return ch
 
 
-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")
+def _patch_neonize_api(monkeypatch) -> None:
     monkeypatch.setattr(
         whatsapp_module,
         "_NEONIZE_API",
@@ -105,8 +89,6 @@ def _patch_neonize_api(monkeypatch, detect_mime=None, detect_buffer=None) -> Non
             MessageEv=object(),
             PairStatusEv=object(),
             build_jid=lambda user, server="s.whatsapp.net": (user, server),
-            detect_mime=detect_mime,
-            detect_buffer=detect_buffer,
         ),
     )
 
@@ -196,7 +178,13 @@ 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 = _make_send_client()
+    client = SimpleNamespace(
+        send_message=AsyncMock(),
+        send_image=AsyncMock(),
+        send_video=AsyncMock(),
+        send_audio=AsyncMock(),
+        send_document=AsyncMock(),
+    )
     ch = _make_channel()
     ch._client = client
     ch._connected = True
@@ -209,7 +197,13 @@ 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 = _make_send_client()
+    client = SimpleNamespace(
+        send_message=AsyncMock(),
+        send_image=AsyncMock(),
+        send_video=AsyncMock(),
+        send_audio=AsyncMock(),
+        send_document=AsyncMock(),
+    )
     ch = _make_channel()
     ch._client = client
     ch._connected = True
@@ -219,14 +213,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.mp3", "report.pdf"],
+            media=["photo.jpg", "clip.mp4", "voice.ogg", "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.mp3")
+    client.send_audio.assert_awaited_once_with(jid, "voice.ogg")
     client.send_document.assert_awaited_once_with(
         jid,
         "report.pdf",
@@ -235,191 +229,6 @@ 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,7 +24,6 @@ class ProviderSnapshot:
 @dataclass(frozen=True)
 class _ProviderSetup:
     model: str
-    provider_name: str
     provider_config: ProviderConfig | None
     spec: ProviderSpec | None
     backend: str
@@ -100,7 +99,6 @@ def _resolve_provider_setup(
 
     return _ProviderSetup(
         model=model,
-        provider_name=provider_name,
         provider_config=p,
         spec=spec,
         backend=backend,
@@ -136,7 +134,6 @@ def _make_provider_core(
         model=model,
     )
     model = setup.model
-    provider_name = setup.provider_name
     p = setup.provider_config
     spec = setup.spec
     backend = setup.backend
@@ -201,7 +198,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 and provider_name == "openai" else "auto",
+            api_type=p.api_type if p 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
+    from nanobot.providers.registry import ProviderSpec, ResponsesCapabilities
 
 # 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,7 +470,12 @@ class OpenAICompatProvider(LLMProvider):
         self.extra_headers = extra_headers or {}
         self._spec = spec
         self._extra_body = extra_body or {}
-        self._api_type = api_type if spec and spec.name == "openai" else "auto"
+        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._extra_query = extra_query or {}
         self._proxy = proxy or None
         self._native_compaction_available = True
@@ -961,39 +966,33 @@ class OpenAICompatProvider(LLMProvider):
         """Choose Responses for providers/models that explicitly support it."""
         if self._api_type == "chat_completions":
             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:
+        capabilities = self._responses_capabilities()
+        if capabilities is None:
             return False
+        model_name = self._request_model_name(model or self.default_model).lower()
         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 provider_responses and (self._spec is None or self._spec.name != "github_copilot"):
-            if not _is_direct_openai_base(self._effective_base):
-                return False
+        if (
+            capabilities.requires_direct_openai_base
+            and not _is_direct_openai_base(self._effective_base)
+        ):
+            return False
 
-        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:
+        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:
             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"
@@ -1016,14 +1015,20 @@ 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 self._spec is not None and self._spec.name != "openai":
+        if (
+            capabilities.requires_direct_openai_base
+            and not _is_direct_openai_base(self._effective_base)
+        ):
             return False
-        return _is_direct_openai_base(self._effective_base)
+        return True
 
     def _responses_circuit_allows_probe(
         self,
@@ -1111,7 +1116,10 @@ class OpenAICompatProvider(LLMProvider):
                     self._sanitize_empty_content(sanitized_state.pending_messages)
                 )
             )
-        preserve_reasoning = bool(self._spec and self._spec.name == "deepseek")
+        capabilities = self._responses_capabilities()
+        preserve_reasoning = (
+            capabilities is not None and capabilities.reasoning_replay == "plaintext"
+        )
         instructions, input_items, replayed = prepare_responses_input(
             sanitized_messages,
             state=sanitized_state,
@@ -1142,10 +1150,15 @@ class OpenAICompatProvider(LLMProvider):
                 "compact_threshold": compact_threshold,
             }]
 
-        if self._supports_temperature(model_name, reasoning_effort):
+        supports_temperature = self._supports_temperature(model_name, reasoning_effort)
+        if supports_temperature:
             body["temperature"] = temperature
 
-        if not self._supports_temperature(model_name, reasoning_effort) and not preserve_reasoning:
+        if (
+            not supports_temperature
+            and capabilities is not None
+            and capabilities.reasoning_replay == "encrypted"
+        ):
             body["include"] = ["reasoning.encrypted_content"]
         if reasoning_effort and reasoning_effort.lower() != "none":
             body["reasoning"] = {"effort": reasoning_effort}
@@ -1766,10 +1779,8 @@ class OpenAICompatProvider(LLMProvider):
                     self._record_responses_success(model, reasoning_effort)
                     return result
                 except Exception as responses_error:
-                    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.
+                    capabilities = self._responses_capabilities()
+                    if capabilities is not None and not capabilities.allows_chat_fallback:
                         raise
                     if self._api_type == "responses":
                         raise
@@ -1862,10 +1873,8 @@ class OpenAICompatProvider(LLMProvider):
                         )
                     return result
                 except Exception as responses_error:
-                    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.
+                    capabilities = self._responses_capabilities()
+                    if capabilities is not None and not capabilities.allows_chat_fallback:
                         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
+from typing import Any, Literal
 
 from pydantic.alias_generators import to_snake
 
@@ -28,6 +28,32 @@ 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.
@@ -111,10 +137,8 @@ class ProviderSpec:
     # Substring match against the wire model name (lowercased).
     implicit_reasoning_models: tuple[str, ...] = ()
 
-    # 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, ...] = ()
+    # Capabilities for providers/models served through the shared Responses path.
+    responses: ResponsesCapabilities | None = None
 
     # When the model returns content as a list of {"type":"thinking",...} +
     # {"type":"text",...} blocks, extract the thinking text into
@@ -373,6 +397,13 @@ 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(
@@ -456,6 +487,11 @@ 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(
@@ -466,7 +502,10 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         backend="openai_compat",
         default_api_base="https://api.deepseek.com",
         thinking_style="thinking_type",
-        responses_models=("deepseek-v4-flash",),
+        responses=ResponsesCapabilities(
+            models=("deepseek-v4-flash",),
+            reasoning_replay="plaintext",
+        ),
     ),
     # Gemini: Google's OpenAI-compatible endpoint
     ProviderSpec(

tests/providers/test_github_copilot_routing.py

@@ -48,6 +48,7 @@ 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,6 +10,11 @@ from nanobot.providers.openai_compat_provider import (
     _RESPONSES_PROBE_INTERVAL_S,
     OpenAICompatProvider,
 )
+from nanobot.providers.registry import (
+    ProviderSpec,
+    ResponsesCapabilities,
+    find_by_name,
+)
 
 
 @pytest.fixture()
@@ -17,7 +22,7 @@ def provider():
     """A direct-OpenAI provider with Responses API support."""
     p = OpenAICompatProvider.__new__(OpenAICompatProvider)
     p.default_model = "gpt-5"
-    p._spec = type("Spec", (), {"name": "openai"})()
+    p._spec = find_by_name("openai")
     p._effective_base = "https://api.openai.com/v1"
     p._api_type = "auto"
     p._responses_failures = {}
@@ -30,12 +35,7 @@ def test_responses_api_available_by_default(provider):
 
 
 def test_deepseek_v4_flash_uses_responses_by_model(provider):
-    provider._spec = type("Spec", (), {
-        "name": "deepseek",
-        "responses_models": ("deepseek-v4-flash",),
-        "strip_model_prefix": False,
-        "strip_model_prefixes": (),
-    })()
+    provider._spec = find_by_name("deepseek")
     provider._effective_base = "https://api.deepseek.com"
     provider.default_model = "deepseek-v4-flash"
 
@@ -44,17 +44,48 @@ def test_deepseek_v4_flash_uses_responses_by_model(provider):
 
 
 def test_deepseek_v4_flash_matches_provider_prefixed_model(provider):
-    provider._spec = type("Spec", (), {
-        "name": "deepseek",
-        "responses_models": ("deepseek-v4-flash",),
-        "strip_model_prefix": False,
-        "strip_model_prefixes": (),
-    })()
+    provider._spec = find_by_name("deepseek")
     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 = {}
 
@@ -73,6 +104,7 @@ 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):
@@ -96,7 +128,7 @@ def test_api_type_responses_ignores_circuit_breaker(provider):
 
 
 def test_api_type_responses_does_not_force_non_openai(provider):
-    provider._spec = type("Spec", (), {"name": "custom"})()
+    provider._spec = find_by_name("custom")
     provider._api_type = "responses"
 
     assert provider._should_use_responses_api("gpt-4o", None) is False