test: assert request construction, streaming order, status mapping, SDK contracts

- Fakes updated to real SDK contracts: McpStdioServer requires name; CapabilitiesConfig gains
  disabled_tools (enabled_tools defaults None, mutually exclusive); FakeApprovalMode values
  deny_all/auto_review; enum-based FakeBuiltinTools; FakeClaudeOptions gains cwd/mcp_servers/
  resume/max_budget_usd/setting_sources/max_turns; ToolResult gains error/exception.
- claude: assert permission_mode per mode, full request shape, resume, dropped_options,
  streaming order (deltas/tool events before completed; error status via tool_use_id),
  result-only single delta, max_turns and errors-list finish_reason, network rejection.
- codex: assert new approval mapping (4 modes), sandbox mapping (3), thread_start/run kwargs,
  failed/interrupted status, tool audits from items, RootModel unwrap, usage fallback,
  allowed/disallowed/budget/network rejections.
- antigravity: assert McpStdioServer name, disallowed->disabled, DEFAULT nondestructive,
  PERMISSIVE all_tools, STRICT read_only+empty policies, invalid tool + allow/deny-both +
  budget/network rejections, private data dir, missing-key unavailable error.
- events: access_token redacted, input_tokens not.
- new tests/test_sdk_contract.py: importorskip-gated pure introspection of the real surfaces
  this package depends on across all three SDKs.

Author: ebarti

tests/test_antigravity_adapter.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+import enum
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
@@ -11,14 +12,45 @@
     AgentTask,
     FilesystemAccess,
     McpServerConfig,
+    PermissionMode,
     PermissionProfile,
 )
 from agent_runtime_kit._errors import UnsupportedTaskInputError
 from agent_runtime_kit.adapters import AntigravityAgentRuntime
 from agent_runtime_kit.testing import RecordingEventSink
 
 
+class FakeBuiltinTools(str, enum.Enum):
+    """Mirror the real ``BuiltinTools`` enum values so validation behaves the same."""
+
+    LIST_DIR = "list_directory"
+    VIEW_FILE = "view_file"
+    EDIT_FILE = "edit_file"
+    RUN_COMMAND = "run_command"
+    START_SUBAGENT = "start_subagent"
+    FINISH = "finish"
+
+    @staticmethod
+    def read_only() -> list[FakeBuiltinTools]:
+        return [FakeBuiltinTools.LIST_DIR, FakeBuiltinTools.VIEW_FILE, FakeBuiltinTools.FINISH]
+
+    @staticmethod
+    def nondestructive() -> list[FakeBuiltinTools]:
+        return [
+            FakeBuiltinTools.LIST_DIR,
+            FakeBuiltinTools.VIEW_FILE,
+            FakeBuiltinTools.EDIT_FILE,
+            FakeBuiltinTools.FINISH,
+        ]
+
+    @staticmethod
+    def all_tools() -> list[FakeBuiltinTools]:
+        return list(FakeBuiltinTools)
+
+
 class FakeTypes:
+    BuiltinTools = FakeBuiltinTools
+
     @dataclass
     class Text:
         text: str
@@ -37,31 +69,20 @@ class ToolResult:
         name: str
         result: str
         args: dict[str, Any] | None = None
+        error: str | None = None
+        exception: Exception | None = None
 
     @dataclass
     class CapabilitiesConfig:
-        enabled_tools: list[Any]
+        enabled_tools: list[Any] | None = None
+        disabled_tools: list[Any] | None = None
         enable_subagents: bool = False
 
     @dataclass
     class McpStdioServer:
+        name: str
         command: str
-        args: list[str]
-
-    class BuiltinTools:
-        START_SUBAGENT = "start_subagent"
-
-        @staticmethod
-        def read_only() -> list[str]:
-            return ["read"]
-
-        @staticmethod
-        def nondestructive() -> list[str]:
-            return ["read", "search"]
-
-        @staticmethod
-        def all_tools() -> list[str]:
-            return ["read", "write", "start_subagent"]
+        args: list[str] = field(default_factory=list)
 
 
 class FakePolicy:
@@ -75,8 +96,26 @@ def __init__(self, **kwargs: Any) -> None:
         self.kwargs = kwargs
 
 
+class FakeUsage:
+    prompt_token_count = 5
+    candidates_token_count = 7
+    thoughts_token_count = 2
+    cached_content_token_count = 1
+    total_token_count = 14
+
+
+class FakeResponse:
+    def __init__(self, prompt: str, chunks_factory: Any) -> None:
+        self.chunks = chunks_factory(prompt)
+        self.usage_metadata = FakeUsage()
+
+    async def structured_output(self) -> dict[str, bool]:
+        return {"ok": True}
+
+
 class FakeAgent:
     last_config: FakeConfig | None = None
+    chunks_factory: Any = None
 
     def __init__(self, config: FakeConfig) -> None:
         FakeAgent.last_config = config
@@ -89,24 +128,7 @@ async def __aexit__(self, *args: object) -> None:
         return None
 
     async def chat(self, prompt: str) -> FakeResponse:
-        return FakeResponse(prompt)
-
-
-class FakeUsage:
-    prompt_token_count = 5
-    candidates_token_count = 7
-    thoughts_token_count = 2
-    cached_content_token_count = 1
-    total_token_count = 14
-
-
-class FakeResponse:
-    def __init__(self, prompt: str) -> None:
-        self.chunks = _chunks(prompt)
-        self.usage_metadata = FakeUsage()
-
-    async def structured_output(self) -> dict[str, bool]:
-        return {"ok": True}
+        return FakeResponse(prompt, FakeAgent.chunks_factory or _chunks)
 
 
 async def _chunks(prompt: str):
@@ -116,17 +138,28 @@ async def _chunks(prompt: str):
     yield FakeTypes.ToolResult("Read", "ok", {"path": "README.md"})
 
 
-@pytest.mark.asyncio
-async def test_antigravity_runtime_runs_with_injected_sdk(tmp_path: Path) -> None:
-    sink = RecordingEventSink()
-    runtime = AntigravityAgentRuntime(
-        api_key="key",
+def make_runtime(*, data_dir: Path, api_key: str = "key") -> AntigravityAgentRuntime:
+    return AntigravityAgentRuntime(
+        api_key=api_key,
+        data_dir=data_dir,
         agent_cls=FakeAgent,
         config_cls=FakeConfig,
         types_module=FakeTypes,
         policy_module=FakePolicy,
     )
 
+
+@pytest.fixture(autouse=True)
+def _reset_agent() -> None:
+    FakeAgent.last_config = None
+    FakeAgent.chunks_factory = None
+
+
+@pytest.mark.asyncio
+async def test_antigravity_runtime_runs_with_injected_sdk(tmp_path: Path) -> None:
+    sink = RecordingEventSink()
+    runtime = make_runtime(data_dir=tmp_path)
+
     result = await runtime.run(
         AgentTask(
             goal="task",
@@ -149,15 +182,141 @@ async def test_antigravity_runtime_runs_with_injected_sdk(tmp_path: Path) -> Non
 
 
 @pytest.mark.asyncio
-async def test_antigravity_rejects_mcp_env() -> None:
-    runtime = AntigravityAgentRuntime(
-        api_key="key",
-        agent_cls=FakeAgent,
-        config_cls=FakeConfig,
-        types_module=FakeTypes,
-        policy_module=FakePolicy,
+async def test_antigravity_mcp_server_gets_name(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    await runtime.run(
+        AgentTask(
+            goal="task",
+            mcp_servers=(McpServerConfig(name="fs", command="mcp", args=("--root",)),),
+        )
     )
 
+    assert FakeAgent.last_config is not None
+    servers = FakeAgent.last_config.kwargs["mcp_servers"]
+    assert servers[0].name == "fs"
+    assert servers[0].command == "mcp"
+
+
+@pytest.mark.asyncio
+async def test_antigravity_default_mode_uses_nondestructive(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    await runtime.run(AgentTask(goal="task"))
+
+    capabilities = FakeAgent.last_config.kwargs["capabilities"]  # type: ignore[union-attr]
+    assert capabilities.enabled_tools == FakeBuiltinTools.nondestructive()
+
+
+@pytest.mark.asyncio
+async def test_antigravity_permissive_uses_all_tools(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    await runtime.run(
+        AgentTask(goal="task", permissions=PermissionProfile(mode=PermissionMode.PERMISSIVE))
+    )
+
+    config = FakeAgent.last_config
+    assert config is not None
+    assert config.kwargs["capabilities"].enabled_tools == FakeBuiltinTools.all_tools()
+    assert config.kwargs["capabilities"].enable_subagents is True
+    assert config.kwargs["policies"] == ["allow_all"]
+
+
+@pytest.mark.asyncio
+async def test_antigravity_strict_uses_read_only_and_no_policies(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    await runtime.run(
+        AgentTask(goal="task", permissions=PermissionProfile(mode=PermissionMode.STRICT))
+    )
+
+    config = FakeAgent.last_config
+    assert config is not None
+    assert config.kwargs["capabilities"].enabled_tools == FakeBuiltinTools.read_only()
+    assert config.kwargs["policies"] == []
+
+
+@pytest.mark.asyncio
+async def test_antigravity_disallowed_tools_map_to_disabled(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    await runtime.run(
+        AgentTask(
+            goal="task",
+            permissions=PermissionProfile(disallowed_tools=("run_command",)),
+        )
+    )
+
+    config = FakeAgent.last_config
+    assert config is not None
+    capabilities = config.kwargs["capabilities"]
+    assert capabilities.disabled_tools == ["run_command"]
+    # enabled_tools and disabled_tools are mutually exclusive in the real SDK.
+    assert capabilities.enabled_tools is None
+
+
+@pytest.mark.asyncio
+async def test_antigravity_rejects_allow_and_deny_list_together(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    with pytest.raises(UnsupportedTaskInputError):
+        await runtime.run(
+            AgentTask(
+                goal="task",
+                permissions=PermissionProfile(
+                    allowed_tools=("view_file",), disallowed_tools=("run_command",)
+                ),
+            )
+        )
+
+
+@pytest.mark.asyncio
+async def test_antigravity_invalid_allowed_tool_raises(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    with pytest.raises(UnsupportedTaskInputError):
+        await runtime.run(
+            AgentTask(goal="task", permissions=PermissionProfile(allowed_tools=("Read",)))
+        )
+
+
+@pytest.mark.asyncio
+async def test_antigravity_tool_result_error_status(tmp_path: Path) -> None:
+    async def error_chunks(prompt: str):
+        yield FakeTypes.Text("partial")
+        yield FakeTypes.ToolResult("Edit", "", {"p": "x"}, error="disk full")
+
+    FakeAgent.chunks_factory = error_chunks
+    runtime = make_runtime(data_dir=tmp_path)
+
+    result = await runtime.run(AgentTask(goal="task"))
+
+    assert result.tool_calls[0].status == "error"
+
+
+@pytest.mark.asyncio
+async def test_antigravity_rejects_budget(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    with pytest.raises(UnsupportedTaskInputError):
+        await runtime.run(AgentTask(goal="task", budget_usd=1.0))
+
+
+@pytest.mark.asyncio
+async def test_antigravity_rejects_network(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
+    with pytest.raises(UnsupportedTaskInputError):
+        await runtime.run(
+            AgentTask(goal="task", permissions=PermissionProfile(network=True))
+        )
+
+
+@pytest.mark.asyncio
+async def test_antigravity_rejects_mcp_env(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
     with pytest.raises(UnsupportedTaskInputError):
         await runtime.run(
             AgentTask(
@@ -167,15 +326,43 @@ async def test_antigravity_rejects_mcp_env() -> None:
         )
 
 
-def test_antigravity_availability_uses_injected_sdk() -> None:
+@pytest.mark.asyncio
+async def test_antigravity_missing_api_key_raises_unavailable(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from agent_runtime_kit._errors import AgentRuntimeUnavailableError
+
+    monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+    monkeypatch.delenv("GOOGLE_API_KEY", raising=False)
     runtime = AntigravityAgentRuntime(
-        api_key="key",
+        api_key=None,
+        data_dir=tmp_path,
         agent_cls=FakeAgent,
         config_cls=FakeConfig,
         types_module=FakeTypes,
         policy_module=FakePolicy,
     )
 
+    with pytest.raises(AgentRuntimeUnavailableError):
+        await runtime.run(AgentTask(goal="task"))
+
+
+def test_antigravity_data_dir_is_private(tmp_path: Path) -> None:
+    import os
+    import stat
+
+    runtime = make_runtime(data_dir=tmp_path)
+
+    path = runtime._runtime_dir("antigravity-sessions")
+
+    assert path.parent == tmp_path
+    mode = stat.S_IMODE(os.stat(path).st_mode)
+    assert mode == 0o700
+
+
+def test_antigravity_availability_uses_injected_sdk(tmp_path: Path) -> None:
+    runtime = make_runtime(data_dir=tmp_path)
+
     diagnostic = runtime.availability()
 
     assert diagnostic.kind is AgentRuntimeKind.ANTIGRAVITY_AGENT_SDK