fix: verify required vendor SDK options

CHANGELOG.md

@@ -16,6 +16,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   reports failure, so the typed result surface never exposes unvalidated data.
 - `AgentKit.aclose()` attempts to close every cached runtime before reraising
   the first close error, preventing later resources from leaking.
+- Permission- and budget-critical vendor options must now be explicit,
+  introspectable SDK parameters; opaque `**kwargs`, positional-only options,
+  and uninspectable callables fail closed.
 
 ## 0.4.0 - 2026-07-02
 

src/agent_runtime_kit/adapters/_common.py

@@ -174,7 +174,7 @@ def filter_supported_kwargs(
     factory: Any,
     kwargs: Mapping[str, Any],
     *,
-    required: Iterable[str] = (),
+    required: Iterable[str] | Mapping[str, str] = (),
     kind: AgentRuntimeKind | str | None = None,
 ) -> tuple[dict[str, Any], list[str]]:
     """Split kwargs into those the callable accepts and those it does not.
@@ -184,26 +184,58 @@ def filter_supported_kwargs(
     dropped, but drops must be observable, so the dropped key names are returned
     alongside the accepted kwargs and surfaced in ``AgentResult.metadata``.
 
-    ``required`` names the kwargs that carry the task's requested security posture
-    (sandbox, approval/permission mode, tool filters). Best-effort dropping is the
-    wrong failure mode there — the run would silently proceed with MORE access
-    than the caller asked for — so drift on a required key fails closed with
-    ``UnsupportedTaskInputError`` (``kind`` is needed for that error).
-
-    Two signature shapes limit detection, with different guarantees. A signature
-    that cannot be introspected passes everything through and still fails closed:
-    a truly unsupported kwarg raises ``TypeError`` inside the SDK call. A
-    signature with ``**kwargs`` also passes everything through, but that path is
-    NOT fail-closed — the callee accepts and may silently ignore unknown options,
-    so drift (required keys included) is undetectable here. Callers that need a
-    hard guarantee against a ``**kwargs``-style vendor API must verify the
-    behavior, not just the signature.
+    ``required`` names kwargs that carry mandatory task constraints (sandbox,
+    approval/permission mode, tool filters, spend caps). A mapping may associate
+    each kwarg with the public task field reported by ``UnsupportedTaskInputError``;
+    an iterable retains the historical ``permissions`` field. Best-effort
+    dropping is the wrong failure mode for either shape.
+
+    Required keys must be explicit parameters. If a callable cannot be
+    introspected, or accepts a required key only through ``**kwargs``, the adapter
+    cannot prove the option will be honored and fails closed. Non-required keys
+    remain best-effort under those opaque signatures.
     """
 
+    required_fields = (
+        dict(required)
+        if isinstance(required, Mapping)
+        else {key: "permissions" for key in required}
+    )
+    if required_fields and kind is None:
+        raise TypeError("filter_supported_kwargs(required=...) also requires kind")
+    required_keys = [key for key in required_fields if key in kwargs]
     try:
         signature = inspect.signature(factory)
     except (TypeError, ValueError):
+        if required_keys:
+            assert kind is not None
+            raise UnsupportedTaskInputError(
+                kind,
+                required_fields[required_keys[0]],
+                "the installed SDK callable cannot be inspected to verify "
+                + ", ".join(required_keys)
+                + "; refusing to run without verifiably honoring required task constraints",
+            ) from None
         return dict(kwargs), []
+
+    opaque_required = [
+        key
+        for key in required_keys
+        if key not in signature.parameters
+        or signature.parameters[key].kind is inspect.Parameter.POSITIONAL_ONLY
+    ]
+    if opaque_required:
+        assert kind is not None
+        raise UnsupportedTaskInputError(
+            kind,
+            required_fields[opaque_required[0]],
+            "the installed SDK does not expose "
+            + ", ".join(opaque_required)
+            + " as explicit keyword parameters; refusing to run without a verifiable "
+            "required task constraint (opaque **kwargs and positional-only parameters "
+            "are insufficient)",
+        )
+
     if any(param.kind is inspect.Parameter.VAR_KEYWORD for param in signature.parameters.values()):
         return dict(kwargs), []
     supported: dict[str, Any] = {}
@@ -213,16 +245,15 @@ def filter_supported_kwargs(
             supported[key] = value
         else:
             dropped.append(key)
-    required_dropped = sorted(set(required) & set(dropped))
+    required_dropped = [key for key in required_keys if key in dropped]
     if required_dropped:
-        if kind is None:
-            raise TypeError("filter_supported_kwargs(required=...) also requires kind")
+        assert kind is not None
         raise UnsupportedTaskInputError(
             kind,
-            "permissions",
+            required_fields[required_dropped[0]],
             "the installed SDK does not accept "
             + ", ".join(required_dropped)
-            + "; refusing to run with a weaker security posture than the task requested",
+            + "; refusing to run without required task constraints",
         )
     return supported, dropped
 

src/agent_runtime_kit/adapters/claude.py

@@ -9,7 +9,7 @@
 from collections.abc import AsyncIterator, Iterable, Mapping
 from typing import Any
 
-from agent_runtime_kit._errors import AgentRuntimeUnavailableError, UnsupportedTaskInputError
+from agent_runtime_kit._errors import AgentRuntimeUnavailableError
 from agent_runtime_kit._types import (
     AgentCapabilities,
     AgentResult,
@@ -360,24 +360,16 @@ def _build_options(
             kwargs["setting_sources"] = [str(item) for item in setting_sources]
         # Security posture must fail closed under vendor drift: the permission
         # mode always, and the tool filters whenever the task requested any.
-        required = ["permission_mode"]
+        required = {"permission_mode": "permissions"}
         if task.permissions.allowed_tools:
-            required.append("allowed_tools")
+            required["allowed_tools"] = "permissions"
         if task.permissions.disallowed_tools:
-            required.append("disallowed_tools")
+            required["disallowed_tools"] = "permissions"
+        if task.budget_usd is not None:
+            required["max_budget_usd"] = "budget_usd"
         supported, dropped = filter_supported_kwargs(
             options_cls, kwargs, required=required, kind=self.kind
         )
-        if task.budget_usd is not None and "max_budget_usd" in dropped:
-            # The spend cap is a limit, not a permission, so it is enforced here
-            # rather than via required= (whose typed error reports
-            # field="permissions"). Silently dropping it would run uncapped.
-            raise UnsupportedTaskInputError(
-                self.kind,
-                "budget_usd",
-                "the installed claude-agent-sdk does not accept max_budget_usd; "
-                "refusing to run without the requested spend cap",
-            )
         return options_cls(**supported), dropped
 
     def _model(self, task: AgentTask) -> str:

tests/test_adapter_common.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from agent_runtime_kit import AgentRuntimeKind, UnsupportedTaskInputError
+from agent_runtime_kit.adapters._common import filter_supported_kwargs
+
+
+def test_filter_supported_kwargs_rejects_required_key_hidden_by_var_kwargs() -> None:
+    def opaque(**_kwargs: object) -> None:
+        return None
+
+    with pytest.raises(UnsupportedTaskInputError, match="explicit keyword parameters"):
+        filter_supported_kwargs(
+            opaque,
+            {"sandbox": "workspace-write", "label": "kept"},
+            required=("sandbox",),
+            kind=AgentRuntimeKind.CODEX_AGENT_SDK,
+        )
+
+
+def test_filter_supported_kwargs_allows_explicit_required_key_with_var_kwargs() -> None:
+    def explicit(*, sandbox: str, **_kwargs: object) -> None:
+        del sandbox
+
+    supported, dropped = filter_supported_kwargs(
+        explicit,
+        {"sandbox": "workspace-write", "future_option": True},
+        required=("sandbox",),
+        kind=AgentRuntimeKind.CODEX_AGENT_SDK,
+    )
+
+    assert supported == {"sandbox": "workspace-write", "future_option": True}
+    assert dropped == []
+
+
+def test_filter_supported_kwargs_rejects_uninspectable_required_key(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def vendor_callable(**_kwargs: object) -> None:
+        return None
+
+    def fail_signature(_factory: object) -> inspect.Signature:
+        raise ValueError("opaque extension callable")
+
+    monkeypatch.setattr(inspect, "signature", fail_signature)
+
+    with pytest.raises(UnsupportedTaskInputError, match="cannot be inspected"):
+        filter_supported_kwargs(
+            vendor_callable,
+            {"permission_mode": "default"},
+            required=("permission_mode",),
+            kind=AgentRuntimeKind.CLAUDE_AGENT_SDK,
+        )
+
+
+def test_filter_supported_kwargs_keeps_non_security_options_best_effort_when_opaque(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def vendor_callable(**_kwargs: object) -> None:
+        return None
+
+    monkeypatch.setattr(
+        inspect,
+        "signature",
+        lambda _factory: (_ for _ in ()).throw(ValueError("opaque extension callable")),
+    )
+
+    supported, dropped = filter_supported_kwargs(vendor_callable, {"label": "kept"})
+
+    assert supported == {"label": "kept"}
+    assert dropped == []
+
+
+def test_filter_supported_kwargs_rejects_positional_only_required_key() -> None:
+    def positional_only(sandbox: str, /) -> None:
+        del sandbox
+
+    with pytest.raises(UnsupportedTaskInputError, match="explicit keyword parameters"):
+        filter_supported_kwargs(
+            positional_only,
+            {"sandbox": "workspace-write"},
+            required=("sandbox",),
+            kind=AgentRuntimeKind.CODEX_AGENT_SDK,
+        )
+
+
+def test_filter_supported_kwargs_requires_kind_for_any_required_contract() -> None:
+    with pytest.raises(TypeError, match="also requires kind"):
+        filter_supported_kwargs(lambda: None, {}, required=("sandbox",))
+
+
+def test_filter_supported_kwargs_maps_required_key_to_public_field() -> None:
+    def opaque(*, permission_mode: str, **_kwargs: object) -> None:
+        del permission_mode
+
+    with pytest.raises(UnsupportedTaskInputError) as exc_info:
+        filter_supported_kwargs(
+            opaque,
+            {"permission_mode": "default", "max_budget_usd": 1.0},
+            required={
+                "permission_mode": "permissions",
+                "max_budget_usd": "budget_usd",
+            },
+            kind=AgentRuntimeKind.CLAUDE_AGENT_SDK,
+        )
+
+    assert exc_info.value.field == "budget_usd"

tests/test_antigravity_adapter.py

@@ -94,7 +94,15 @@ def allow_all() -> str:
 
 
 class FakeConfig:
-    def __init__(self, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        *,
+        capabilities: Any = None,
+        policies: Any = None,
+        workspaces: Any = None,
+        **kwargs: Any,
+    ) -> None:
+        kwargs.update(capabilities=capabilities, policies=policies, workspaces=workspaces)
         self.kwargs = kwargs
 
 

tests/test_claude_adapter.py

@@ -609,6 +609,32 @@ class NoBudgetOptions:
     assert exc_info.value.field == "budget_usd"
 
 
+@pytest.mark.asyncio
+async def test_claude_budget_fails_closed_when_only_opaque_kwargs_accept_it() -> None:
+    class OpaqueBudgetOptions:
+        def __init__(
+            self,
+            *,
+            permission_mode: str | None = None,
+            allowed_tools: list[str] | None = None,
+            disallowed_tools: list[str] | None = None,
+            **_kwargs: Any,
+        ) -> None:
+            self.permission_mode = permission_mode
+            self.allowed_tools = allowed_tools or []
+            self.disallowed_tools = disallowed_tools or []
+
+    runtime = ClaudeAgentRuntime(
+        query_func=make_query([assistant("ok"), result_message()]),
+        options_cls=OpaqueBudgetOptions,
+    )
+
+    with pytest.raises(UnsupportedTaskInputError) as exc_info:
+        await runtime.run(AgentTask(goal="x", budget_usd=2.5))
+
+    assert exc_info.value.field == "budget_usd"
+
+
 @pytest.mark.asyncio
 async def test_claude_streams_events_before_completion() -> None:
     sink = RecordingEventSink()