diff --git a/nemoguardrails/server/hooks/__init__.py b/nemoguardrails/server/hooks/__init__.py new file mode 100644 index 0000000000..2455b453d7 --- /dev/null +++ b/nemoguardrails/server/hooks/__init__.py @@ -0,0 +1,31 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Agent hook requests and responses for `nemoguardrails hooks-server`.""" + +from nemoguardrails.server.hooks.adapters import ADAPTERS, HookAdapter, HookAgent, UnsupportedHookEventError +from nemoguardrails.server.hooks.checking import HookOutcome, check_messages, outcome_from_result +from nemoguardrails.server.hooks.schemas import HookErrorCode + +__all__ = [ + "ADAPTERS", + "HookAdapter", + "HookAgent", + "HookErrorCode", + "HookOutcome", + "UnsupportedHookEventError", + "check_messages", + "outcome_from_result", +] diff --git a/nemoguardrails/server/hooks/adapters.py b/nemoguardrails/server/hooks/adapters.py new file mode 100644 index 0000000000..23899ed414 --- /dev/null +++ b/nemoguardrails/server/hooks/adapters.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Each agent's hook interface: the event its path handles, the tool call that event asks to check, and its answer.""" + +from dataclasses import dataclass +from typing import Any, Callable, Literal, Union + +from pydantic import BaseModel + +from nemoguardrails.guardrails.guardrails_types import quoted_identity +from nemoguardrails.server.hooks.checking import HookOutcome, ToolInvocation +from nemoguardrails.server.hooks.schemas import ( + ClaudeCodePreToolUseRequest, + CodexPreToolUseRequest, + HookBody, + NemoRelayToolConditionalExecutionGuardrailRequest, +) + +HookAgent = Literal["claude-code", "codex", "nemo-relay"] + +# Claude Code and Codex show the deny reason to the model, so it says who denied the call. +DENY_REASON_PREFIX = "NeMo Guardrails: " + + +class UnsupportedHookEventError(Exception): + """A `hook_event_name` that the agent's path doesn't handle.""" + + def __init__(self, agent: HookAgent, event: str, handled_event: str) -> None: + self.agent = agent + self.event = event + super().__init__( + f"hook_event_name {quoted_identity(event)!r} is not handled on /v1/hooks/{agent}; send {handled_event}" + ) + + +@dataclass(frozen=True, slots=True) +class HookAdapter: + """One agent's path: the event it handles, the tool call that event's body carries, and each outcome's response.""" + + agent: HookAgent + event: str + request_model: type[BaseModel] + invocation: Callable[[Any], ToolInvocation] + response_body: Callable[[HookOutcome], dict] + + def parse(self, body: object) -> ToolInvocation: + """The tool call *body* asks to check, as this path's event; error locations are relative to the body.""" + event = HookBody.model_validate(body).hook_event_name + if event != self.event: + raise UnsupportedHookEventError(self.agent, event, self.event) + return self.invocation(self.request_model.model_validate(body)) + + +def _pre_tool_use_invocation(request: Union[ClaudeCodePreToolUseRequest, CodexPreToolUseRequest]) -> ToolInvocation: + """The tool call a Claude Code or Codex `PreToolUse` hook asks to check.""" + return ToolInvocation(name=request.tool_name, arguments=request.tool_input, call_id=request.tool_use_id) + + +def _relay_invocation(request: NemoRelayToolConditionalExecutionGuardrailRequest) -> ToolInvocation: + """The tool call a NeMo Relay guardrail asks to check, with no id: Relay passes its guardrail callback none.""" + return ToolInvocation(name=request.tool_name, arguments=request.tool_input) + + +def _pre_tool_use_output(outcome: HookOutcome) -> dict: + """Claude Code's and Codex's `PreToolUse` output: no decision for an allow, the deny shape for a deny.""" + _require_decision(outcome) + if outcome.kind == "allow": + # No decision rather than "allow", which would skip Claude Code's permission prompt. + return {} + return { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": f"{DENY_REASON_PREFIX}{outcome.reason}", + } + } + + +def _relay_decision(outcome: HookOutcome) -> dict: + """The verdict the Relay plugin turns into its guardrail callback's `Ok(None)` or `Ok(Some(reason))`.""" + _require_decision(outcome) + if outcome.kind == "allow": + return {"decision": "allow"} + return {"decision": "deny", "reason": outcome.reason} + + +def _require_decision(outcome: HookOutcome) -> None: + """Raise for an error outcome, which is answered with an error status, not a `200` body.""" + if outcome.kind == "error": + raise ValueError("an error outcome is answered with an error status, not a 200 body") + + +ADAPTERS: dict[HookAgent, HookAdapter] = { + adapter.agent: adapter + for adapter in ( + HookAdapter( + agent="claude-code", + event="PreToolUse", + request_model=ClaudeCodePreToolUseRequest, + invocation=_pre_tool_use_invocation, + response_body=_pre_tool_use_output, + ), + HookAdapter( + agent="codex", + event="PreToolUse", + request_model=CodexPreToolUseRequest, + invocation=_pre_tool_use_invocation, + response_body=_pre_tool_use_output, + ), + HookAdapter( + agent="nemo-relay", + event="NemoRelayToolConditionalExecutionGuardrail", + request_model=NemoRelayToolConditionalExecutionGuardrailRequest, + invocation=_relay_invocation, + response_body=_relay_decision, + ), + ) +} diff --git a/nemoguardrails/server/hooks/checking.py b/nemoguardrails/server/hooks/checking.py new file mode 100644 index 0000000000..3010c38d5a --- /dev/null +++ b/nemoguardrails/server/hooks/checking.py @@ -0,0 +1,150 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The bridge to the engine: the check a hook's tool call asks for, and the verdict the check's result gives.""" + +import json +from dataclasses import dataclass +from typing import Any, Literal, Optional + +from nemoguardrails.rails.llm.options import RailsResult, RailStatus, ToolViolation, ToolViolationType +from nemoguardrails.server.hooks.schemas import HookErrorCode + +_UNSPECIFIED_DENY_REASON = "tool call blocked" + +# Violations that mean the check couldn't finish, rather than a decision on the call. A malformed call is a decision. +_CHECK_FAILURES = frozenset( + { + ToolViolationType.RAIL_FAILED, + ToolViolationType.EXTRACTION_FAILED, + ToolViolationType.INVALID_TOOLSET, + ToolViolationType.INVALID_TOOL_SCHEMA, + } +) + + +@dataclass(frozen=True, slots=True) +class ToolInvocation: + """The tool call a hook asks to check, whichever agent sent it; internal to the hooks server.""" + + name: str + arguments: Any + call_id: Optional[str] = None + + +def check_messages(call: ToolInvocation) -> list[dict]: + """The one assistant message carrying *call*, as `check_async()` takes it; with no id when the call has none.""" + tool_call: dict[str, Any] = { + "type": "function", + "function": {"name": call.name, "arguments": json.dumps(call.arguments)}, + } + if call.call_id is not None: + tool_call["id"] = call.call_id + return [{"role": "assistant", "content": None, "tool_calls": [tool_call]}] + + +@dataclass(frozen=True, slots=True) +class HookOutcome: + """A check's verdict on a tool call, before it becomes the agent's response.""" + + kind: Literal["allow", "deny", "error"] + reason: Optional[str] = None + code: Optional[HookErrorCode] = None + message: Optional[str] = None + + def __post_init__(self) -> None: + """Refuse any field the outcome's kind doesn't have, so only a valid allow, deny or error exists.""" + if self.code is not None: + object.__setattr__(self, "code", HookErrorCode(self.code)) + if self.kind == "allow": + self._require_unset("reason", "code", "message") + elif self.kind == "deny": + self._require_text("reason") + self._require_unset("code", "message") + elif self.kind == "error": + self._require_text("message") + self._require_unset("reason") + else: + raise ValueError(f"kind must be 'allow', 'deny' or 'error', got {self.kind!r}") + + def _require_text(self, name: str) -> None: + """Raise unless the field *name* holds a non-empty string.""" + value = getattr(self, name) + if value is None or value == "": + raise ValueError(f"a {self.kind} outcome needs a {name}") + if not isinstance(value, str): + raise TypeError(f"a {self.kind} outcome's {name} must be a string") + + def _require_unset(self, *names: str) -> None: + """Raise when any of the fields *names* is set, since this kind of outcome has none of them.""" + for name in names: + if getattr(self, name) is not None: + raise ValueError(f"a {self.kind} outcome has no {name}") + + @classmethod + def allow(cls) -> "HookOutcome": + return cls(kind="allow") + + @classmethod + def deny(cls, reason: str) -> "HookOutcome": + return cls(kind="deny", reason=reason) + + @classmethod + def error(cls, message: str, code: Optional[HookErrorCode] = None) -> "HookOutcome": + return cls(kind="error", code=code, message=message) + + +def outcome_from_result(result: RailsResult) -> HookOutcome: + """The verdict a tool-call check's result gives: allow, deny, or an error when no rail decided.""" + if result.status == RailStatus.PASSED: + return HookOutcome.allow() + if result.status == RailStatus.BLOCKED: + return _blocked_outcome(result) + # A tool-call check rewrites nothing, so any other status breaks the engine's contract. + return HookOutcome.error(f"unexpected check status {result.status.value!r}") + + +def _blocked_outcome(result: RailsResult) -> HookOutcome: + """A deny when any violation is a decision on the call, or an error when every violation is a check failure.""" + violations = result.tool_violations or [] + # A block that reports no violations still blocks, so it is a deny rather than an error. + if violations and all(violation.violation_type in _CHECK_FAILURES for violation in violations): + return _check_failed_outcome(violations) + return HookOutcome.deny(_deny_reason(result)) + + +def _check_failed_outcome(violations: list[ToolViolation]) -> HookOutcome: + """A `rail_failed` error when every failure is a rail's, otherwise an error with no code naming the first other.""" + other_failures = [ + violation for violation in violations if violation.violation_type != ToolViolationType.RAIL_FAILED + ] + if not other_failures: + return HookOutcome.error(_rail_failed_message(violations[0]), code=HookErrorCode.RAIL_FAILED) + return HookOutcome.error(f"the tool-call check failed: {other_failures[0].reason}") + + +def _deny_reason(result: RailsResult) -> str: + """The first non-empty reason among the decided violations, then the result's reason, then the rail that blocked.""" + for violation in result.tool_violations or []: + if violation.violation_type not in _CHECK_FAILURES and violation.reason: + return violation.reason + return result.reason or result.rail or _UNSPECIFIED_DENY_REASON + + +def _rail_failed_message(violation: ToolViolation) -> str: + """The error message for a failed rail, naming the rail when the violation does.""" + if violation.rail is None: + return f"a tool-call rail failed: {violation.reason}" + return f"rail '{violation.rail}' failed: {violation.reason}" diff --git a/nemoguardrails/server/hooks/schemas.py b/nemoguardrails/server/hooks/schemas.py new file mode 100644 index 0000000000..5e09492847 --- /dev/null +++ b/nemoguardrails/server/hooks/schemas.py @@ -0,0 +1,76 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Request models for the agent hooks the hooks server checks, one interface per agent path.""" + +from enum import Enum +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class HookErrorCode(str, Enum): + """The `error.code` of a hooks request that ends in an error rather than an allow or a deny.""" + + UNSUPPORTED_HOOK_EVENT = "unsupported_hook_event" + RAIL_FAILED = "rail_failed" + QUEUE_FULL = "queue_full" + + +class ClaudeCodePreToolUseRequest(BaseModel): + """Claude Code's `PreToolUse` hook input, requiring only the fields the server reads.""" + + # Undeclared fields are kept as extras, unchecked, so an agent release can add or change them. + model_config = ConfigDict(extra="allow") + + # Claude Code Common input fields (https://code.claude.com/docs/en/hooks#common-input-fields) + hook_event_name: Literal["PreToolUse"] + + # Claude Code PreToolUse input fields (https://code.claude.com/docs/en/hooks#pretooluse-input) + tool_name: str = Field(min_length=1) + tool_input: Any + tool_use_id: str + + +class CodexPreToolUseRequest(BaseModel): + """Codex's `PreToolUse` hook input, requiring only the fields the server reads.""" + + # Undeclared fields are kept as extras, unchecked, so an agent release can add or change them. + model_config = ConfigDict(extra="allow") + + # Codex common input fields: https://learn.chatgpt.com/docs/hooks#common-input-fields + hook_event_name: Literal["PreToolUse"] + + # Codex PreToolUse fields: https://learn.chatgpt.com/docs/hooks#pretooluse + tool_name: str = Field(min_length=1) + tool_use_id: str + tool_input: Any + + +class NemoRelayToolConditionalExecutionGuardrailRequest(BaseModel): + """NeMo Relay tool conditional-execution guardrail callback receives: the tool name and its arguments.""" + + model_config = ConfigDict(extra="forbid") + + hook_event_name: Literal["NemoRelayToolConditionalExecutionGuardrail"] + tool_name: str = Field(min_length=1) + tool_input: Any + + +# Not underscored: pydantic quotes the class name in the error for a body that isn't an object. +class HookBody(BaseModel): + """The field every hook body carries, read before the body is validated as its event.""" + + hook_event_name: str diff --git a/tests/server/test_hooks_models.py b/tests/server/test_hooks_models.py new file mode 100644 index 0000000000..ca199ff46a --- /dev/null +++ b/tests/server/test_hooks_models.py @@ -0,0 +1,832 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import dataclasses +import json +from typing import Any, Optional, get_args + +import jsonschema +import pytest +from pydantic import ValidationError +from pydantic_core import ErrorDetails + +from nemoguardrails.rails.llm.options import RailsResult, RailStatus, ToolViolation +from nemoguardrails.server.hooks import ( + ADAPTERS, + HookAgent, + HookErrorCode, + HookOutcome, + UnsupportedHookEventError, + check_messages, + outcome_from_result, +) +from nemoguardrails.server.hooks.checking import ToolInvocation +from nemoguardrails.server.hooks.schemas import ( + ClaudeCodePreToolUseRequest, + CodexPreToolUseRequest, + NemoRelayToolConditionalExecutionGuardrailRequest, +) +from tests.guardrails.tool_helpers import call_violation + +CLAUDE_CODE_BASH = { + "session_id": "abc123", + "prompt_id": "550e8400-e29b-41d4-a716-446655440000", + "transcript_path": "/home/user/.claude/projects/.../transcript.jsonl", + "cwd": "/home/user/my-project", + "scratchpad_dir": "/tmp/claude-1000/-home-user-my-project/abc123/scratchpad", + "permission_mode": "default", + "hook_event_name": "PreToolUse", + "tool_name": "Bash", + "tool_input": { + "command": "npm test", + "description": "Run test suite", + "timeout": 120000, + "run_in_background": False, + }, + "tool_use_id": "toolu_01ABC123...", +} + +CLAUDE_CODE_DESTRUCTIVE_BASH = { + "session_id": "abc123", + "transcript_path": "/home/user/.claude/projects/.../transcript.jsonl", + "cwd": "/home/user/my-project", + "permission_mode": "default", + "hook_event_name": "PreToolUse", + "tool_name": "Bash", + "tool_input": {"command": "rm -rf /tmp/build"}, + "tool_use_id": "toolu_01GHI789...", +} + +CLAUDE_CODE_WRITE_WINDOWS = { + "session_id": "abc123", + "transcript_path": "C:\\Users\\user\\.claude\\projects\\...\\transcript.jsonl", + "cwd": "C:\\project", + "permission_mode": "default", + "hook_event_name": "PreToolUse", + "tool_name": "Write", + "tool_input": {"file_path": "C:\\project\\src\\index.ts", "content": "..."}, + "tool_use_id": "toolu_01JKL012...", +} + +CLAUDE_CODE_MCP = { + "session_id": "abc123", + "transcript_path": "/home/user/.claude/projects/.../transcript.jsonl", + "cwd": "/home/user/my-project", + "permission_mode": "default", + "hook_event_name": "PreToolUse", + "tool_name": "mcp__github__search_repositories", + "tool_input": {"query": "nemo guardrails"}, + "tool_use_id": "toolu_01DEF456...", + "mcp_server": {"name": "github", "source": "user"}, +} + +CODEX_APPLY_PATCH = { + "session_id": "", + "transcript_path": None, + "cwd": "", + "hook_event_name": "PreToolUse", + "model": "", + "permission_mode": "default", + "turn_id": "", + "tool_name": "apply_patch", + "tool_input": { + "command": "*** Begin Patch\n" + "*** Add File: nested/new.txt\n" + "+created\n" + "*** Delete File: delete.txt\n" + "*** Update File: modify.txt\n" + "@@\n" + "-line2\n" + "+changed\n" + "*** End Patch\n" + }, + "tool_use_id": "", +} + +CODEX_MCP = { + "session_id": "", + "transcript_path": None, + "cwd": "", + "hook_event_name": "PreToolUse", + "model": "", + "permission_mode": "default", + "turn_id": "", + "tool_name": "mcp__fs__read", + "tool_input": {"file_path": "src/main.rs", "count": 3}, + "tool_use_id": "", +} + +NEMO_RELAY_SEARCH = { + "hook_event_name": "NemoRelayToolConditionalExecutionGuardrail", + "tool_name": "search", + "tool_input": {"query": "weather"}, +} + +NEMO_RELAY_EMPTY_QUERY = { + "hook_event_name": "NemoRelayToolConditionalExecutionGuardrail", + "tool_name": "search", + "tool_input": {"query": ""}, +} + +CLAUDE_CODE_EXAMPLES = [CLAUDE_CODE_BASH, CLAUDE_CODE_DESTRUCTIVE_BASH, CLAUDE_CODE_WRITE_WINDOWS, CLAUDE_CODE_MCP] +CODEX_EXAMPLES = [CODEX_APPLY_PATCH, CODEX_MCP] + +PRE_TOOL_USE_REQUIRED = ["hook_event_name", "tool_name", "tool_input", "tool_use_id"] +NEMO_RELAY_REQUIRED = ["hook_event_name", "tool_name", "tool_input"] + +# Codex's generated PreToolUse hook schemas, copied unchanged from openai/codex@e444aa99: +# https://github.com/openai/codex/tree/e444aa99d76beeb8b5b0610b793c0cdefe58da58/codex-rs/hooks/schema/generated +# OpenAI Codex, Copyright 2025 OpenAI, licensed under the Apache License, Version 2.0. +# Refresh both from one Codex commit, and update that commit here. + +# pre-tool-use.command.input.schema.json: what Codex sends to a PreToolUse hook. +CODEX_INPUT_SCHEMA = { + "$schema": "http://json-schema.org/draft-07/schema#", + "additionalProperties": False, + "definitions": {"NullableString": {"type": ["string", "null"]}}, + "properties": { + "agent_id": {"type": "string"}, + "agent_type": {"type": "string"}, + "cwd": {"type": "string"}, + "hook_event_name": {"const": "PreToolUse", "type": "string"}, + "model": {"type": "string"}, + "permission_mode": { + "enum": ["default", "acceptEdits", "plan", "dontAsk", "bypassPermissions"], + "type": "string", + }, + "session_id": {"type": "string"}, + "tool_input": True, + "tool_name": {"type": "string"}, + "tool_use_id": {"type": "string"}, + "transcript_path": {"$ref": "#/definitions/NullableString"}, + "turn_id": { + "description": "Codex extension: expose the active turn id to internal turn-scoped hooks.", + "type": "string", + }, + }, + "required": [ + "cwd", + "hook_event_name", + "model", + "permission_mode", + "session_id", + "tool_input", + "tool_name", + "tool_use_id", + "transcript_path", + "turn_id", + ], + "title": "pre-tool-use.command.input", + "type": "object", +} + +# pre-tool-use.command.output.schema.json: what Codex accepts back. +CODEX_OUTPUT_SCHEMA = { + "$schema": "http://json-schema.org/draft-07/schema#", + "additionalProperties": False, + "definitions": { + "PreToolUseDecisionWire": {"enum": ["approve", "block"], "type": "string"}, + "PreToolUseHookSpecificOutputWire": { + "additionalProperties": False, + "properties": { + "additionalContext": {"default": None, "type": "string"}, + "hookEventName": {"const": "PreToolUse", "type": "string"}, + "permissionDecision": { + "allOf": [{"$ref": "#/definitions/PreToolUsePermissionDecisionWire"}], + "default": None, + }, + "permissionDecisionReason": {"default": None, "type": "string"}, + "updatedInput": {"default": None}, + }, + "required": ["hookEventName"], + "type": "object", + }, + "PreToolUsePermissionDecisionWire": {"enum": ["allow", "deny", "ask"], "type": "string"}, + }, + "properties": { + "continue": {"default": True, "type": "boolean"}, + "decision": {"allOf": [{"$ref": "#/definitions/PreToolUseDecisionWire"}], "default": None}, + "hookSpecificOutput": {"allOf": [{"$ref": "#/definitions/PreToolUseHookSpecificOutputWire"}], "default": None}, + "reason": {"default": None, "type": "string"}, + "stopReason": {"default": None, "type": "string"}, + "suppressOutput": {"default": False, "type": "boolean"}, + "systemMessage": {"default": None, "type": "string"}, + }, + "title": "pre-tool-use.command.output", + "type": "object", +} + +REFUSAL = "I'm sorry, I can't respond to that." +INTERNAL_ERROR = "I'm sorry, an internal error has occurred." + +UNDECLARED_TOOL = call_violation( + "tool_not_allowed", + "tool call 'WebFetch' is not an allowed tool", + tool_call_id="toolu_01", + tool_name="WebFetch", + index=0, +) +PER_TOOL_BLOCK = call_violation( + "per_tool_rail", + "regex check tool call", + tool_call_id="toolu_01", + tool_name="Bash", + index=0, + rail="regex check tool call", +) +MALFORMED_ARGUMENTS = call_violation( + "malformed_arguments", + "tool call 'toolu_01' has malformed arguments", + tool_call_id="toolu_01", + tool_name="Bash", + index=0, +) +RAIL_FAILED = call_violation( + "rail_failed", + "regex check tool call error", + tool_call_id="toolu_01", + tool_name="Bash", + index=0, + rail="regex check tool call", +) +EXTRACTION_FAILED = call_violation("extraction_failed", "tool call extraction failed") +INVALID_TOOLSET = call_violation("invalid_toolset", "tool parsing failed") +INVALID_TOOL_SCHEMA = call_violation( + "invalid_tool_schema", + "declared schema for tool 'Bash' is not valid JSON Schema", + tool_call_id="toolu_01", + tool_name="Bash", + index=0, +) +EMPTY_REASON_BLOCK = call_violation( + "per_tool_rail", + "", + tool_call_id="toolu_01", + tool_name="Bash", + index=0, + rail="custom rail", +) + + +def _without(body: dict, field: str) -> dict: + """*body* with *field* removed.""" + return {key: value for key, value in body.items() if key != field} + + +def _validation_errors(agent: HookAgent, body: Any) -> list[ErrorDetails]: + """The pydantic errors *body* raises on *agent*'s path.""" + with pytest.raises(ValidationError) as exc_info: + ADAPTERS[agent].parse(body) + return exc_info.value.errors() + + +def _blocked( + *violations: ToolViolation, rail: Optional[str] = None, reason: Optional[str] = None, content: str = REFUSAL +) -> RailsResult: + """A blocked tool-call check carrying *violations*, as IORails reports one.""" + return RailsResult( + status=RailStatus.BLOCKED, + content=content, + rail=rail, + reason=reason, + tool_violations=list(violations) or None, + ) + + +class TestAdapters: + def test_registry_has_one_adapter_per_agent_naming_that_agent(self): + """The registry has an adapter for every agent path, and each adapter names the agent it is registered for.""" + assert {agent: adapter.agent for agent, adapter in ADAPTERS.items()} == { + agent: agent for agent in get_args(HookAgent) + } + + @pytest.mark.parametrize("agent", get_args(HookAgent)) + def test_adapter_event_is_the_event_its_request_model_takes(self, agent): + """Each adapter's event is the one `hook_event_name` its request model accepts.""" + adapter = ADAPTERS[agent] + + assert get_args(adapter.request_model.model_fields["hook_event_name"].annotation) == (adapter.event,) + + +class TestParse: + @pytest.mark.parametrize( + "agent, body, call_id", + [ + *[("claude-code", body, body["tool_use_id"]) for body in CLAUDE_CODE_EXAMPLES], + *[("codex", body, body["tool_use_id"]) for body in CODEX_EXAMPLES], + *[("nemo-relay", body, None) for body in [NEMO_RELAY_SEARCH, NEMO_RELAY_EMPTY_QUERY]], + ], + ) + def test_spec_request_examples_parse_on_their_own_path(self, agent, body, call_id): + """Every request example in the spec parses on its own path into the tool call it carries.""" + assert ADAPTERS[agent].parse(body) == ToolInvocation( + name=body["tool_name"], arguments=body["tool_input"], call_id=call_id + ) + + @pytest.mark.parametrize( + "agent, body", + [ + *[("codex", body) for body in CLAUDE_CODE_EXAMPLES], + *[("claude-code", body) for body in CODEX_EXAMPLES], + ], + ) + def test_payload_parses_on_the_other_agents_path(self, agent, body): + """A Claude Code payload parses on the Codex path and the reverse, since both paths read the same fields.""" + assert ADAPTERS[agent].parse(body) == ToolInvocation( + name=body["tool_name"], arguments=body["tool_input"], call_id=body["tool_use_id"] + ) + + @pytest.mark.parametrize("agent", ["claude-code", "codex"]) + def test_body_with_only_the_fields_the_server_reads_parses(self, agent): + """A body with only `hook_event_name`, `tool_name`, `tool_input` and `tool_use_id` parses on either agent path.""" + body = {field: CLAUDE_CODE_BASH[field] for field in PRE_TOOL_USE_REQUIRED} + + assert ADAPTERS[agent].parse(body).call_id == "toolu_01ABC123..." + + @pytest.mark.parametrize( + "agent, body, field", + [ + *[("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH, field) for field in PRE_TOOL_USE_REQUIRED], + *[("codex", CODEX_MCP, field) for field in PRE_TOOL_USE_REQUIRED], + *[("nemo-relay", NEMO_RELAY_SEARCH, field) for field in NEMO_RELAY_REQUIRED], + ], + ) + def test_missing_required_field_fails_at_that_field(self, agent, body, field): + """A body missing one required field fails with that field, and only that field, as the error.""" + errors = _validation_errors(agent, _without(body, field)) + + assert [(error["loc"], error["type"]) for error in errors] == [((field,), "missing")] + + @pytest.mark.parametrize( + "agent, body, field, value", + [ + ("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH, "model", ""), + ("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH, "turn_id", None), + ("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH, "transcript_path", None), + ("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH, "effort", "high"), + ("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH, "mcp_server", "github"), + ("codex", CODEX_MCP, "prompt_id", "550e8400-e29b-41d4-a716-446655440000"), + ("codex", CODEX_MCP, "mcp_server", {"name": "github", "source": "user"}), + ("codex", CODEX_MCP, "permission_mode", None), + ("codex", CODEX_MCP, "agent_id", 7), + ], + ) + def test_field_the_server_does_not_read_parses_with_any_value(self, agent, body, field, value): + """A field the server doesn't read parses with any value, even another agent's field, and leaves the call alone.""" + assert ADAPTERS[agent].parse({**body, field: value}) == ADAPTERS[agent].parse(body) + + def test_nemo_relay_rejects_unknown_field(self): + """The NeMo Relay path rejects a field its schema doesn't list.""" + errors = _validation_errors("nemo-relay", {**NEMO_RELAY_SEARCH, "session_id": "abc123"}) + + assert [(error["loc"], error["msg"]) for error in errors] == [ + (("session_id",), "Extra inputs are not permitted") + ] + + @pytest.mark.parametrize("agent, body", [("claude-code", CLAUDE_CODE_DESTRUCTIVE_BASH), ("codex", CODEX_MCP)]) + def test_agent_paths_accept_unknown_fields(self, agent, body): + """The Claude Code and Codex paths accept a field the spec doesn't list, so a new agent field is no error.""" + unknown_field = {"added_by_a_later_release": {"nested": 1}} + + assert ADAPTERS[agent].parse({**body, **unknown_field}) == ADAPTERS[agent].parse(body) + + @pytest.mark.parametrize("mode", ["python", "json"]) + @pytest.mark.parametrize( + "model, body", + [ + *[(ClaudeCodePreToolUseRequest, body) for body in CLAUDE_CODE_EXAMPLES], + *[(CodexPreToolUseRequest, body) for body in CODEX_EXAMPLES], + *[ + (NemoRelayToolConditionalExecutionGuardrailRequest, body) + for body in [NEMO_RELAY_SEARCH, NEMO_RELAY_EMPTY_QUERY] + ], + ], + ) + def test_dumped_request_validates_again_unchanged(self, model, body, mode): + """A request's `model_dump()` validates again to an equal request, so a saved request reloads.""" + request = model.model_validate(body) + + assert model.model_validate(request.model_dump(mode=mode)) == request + + @pytest.mark.parametrize( + "agent, body", [("claude-code", CLAUDE_CODE_BASH), ("codex", CODEX_MCP), ("nemo-relay", NEMO_RELAY_SEARCH)] + ) + def test_empty_tool_name_fails(self, agent, body): + """An empty `tool_name` fails at `tool_name` on every path.""" + errors = _validation_errors(agent, {**body, "tool_name": ""}) + + assert [(error["loc"], error["type"]) for error in errors] == [(("tool_name",), "string_too_short")] + + @pytest.mark.parametrize( + "agent, event", + [ + ("claude-code", "PostToolUse"), + ("codex", "PostToolUse"), + ("nemo-relay", "PostToolUse"), + ("claude-code", "NemoRelayToolConditionalExecutionGuardrail"), + ("nemo-relay", "PreToolUse"), + ], + ) + def test_event_the_path_does_not_handle_raises_unsupported_hook_event(self, agent, event): + """An event the path doesn't handle raises `UnsupportedHookEventError`, before the body is validated.""" + with pytest.raises(UnsupportedHookEventError) as exc_info: + ADAPTERS[agent].parse({"hook_event_name": event}) + + assert (exc_info.value.agent, exc_info.value.event) == (agent, event) + + def test_unsupported_hook_event_message_names_the_path_and_its_events(self): + """The message names the event, the path and the event to send instead, as in the spec's example.""" + with pytest.raises(UnsupportedHookEventError) as exc_info: + ADAPTERS["claude-code"].parse({**CLAUDE_CODE_BASH, "hook_event_name": "PostToolUse"}) + + assert str(exc_info.value) == ( + "hook_event_name 'PostToolUse' is not handled on /v1/hooks/claude-code; send PreToolUse" + ) + + def test_unsupported_hook_event_message_cuts_and_escapes_the_event(self): + """A long event name with line breaks reaches the message cut short and escaped.""" + with pytest.raises(UnsupportedHookEventError) as exc_info: + ADAPTERS["codex"].parse({"hook_event_name": "Evil\n" * 1000}) + + message = str(exc_info.value) + assert "\n" not in message + assert len(message) < 200 + + @pytest.mark.parametrize("body", [[CLAUDE_CODE_BASH], "PreToolUse", None, 1]) + def test_body_that_is_not_an_object_fails_at_the_body(self, body): + """A body that isn't a JSON object fails with an error at the body itself, naming no field.""" + errors = _validation_errors("claude-code", body) + + assert [error["loc"] for error in errors] == [()] + + @pytest.mark.parametrize("event", [5, None, ["PreToolUse"]]) + def test_non_string_event_fails_at_hook_event_name(self, event): + """A `hook_event_name` that isn't a string fails at `hook_event_name`.""" + errors = _validation_errors("claude-code", {**CLAUDE_CODE_BASH, "hook_event_name": event}) + + assert [(error["loc"], error["type"]) for error in errors] == [(("hook_event_name",), "string_type")] + + +class TestCheckMessages: + def test_claude_code_call_becomes_one_assistant_tool_call(self): + """A Claude Code hook becomes one assistant message whose one tool call carries the hook's id, name and input.""" + messages = check_messages(ADAPTERS["claude-code"].parse(CLAUDE_CODE_DESTRUCTIVE_BASH)) + + assert messages == [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "toolu_01GHI789...", + "type": "function", + "function": {"name": "Bash", "arguments": '{"command": "rm -rf /tmp/build"}'}, + } + ], + } + ] + + def test_codex_call_carries_its_tool_use_id_and_json_encoded_input(self): + """A Codex hook's call carries `tool_use_id` as its id, and `tool_input` encoded as JSON.""" + (tool_call,) = check_messages(ADAPTERS["codex"].parse(CODEX_APPLY_PATCH))[0]["tool_calls"] + + assert tool_call["id"] == "" + assert tool_call["function"]["name"] == "apply_patch" + assert json.loads(tool_call["function"]["arguments"]) == CODEX_APPLY_PATCH["tool_input"] + + def test_nemo_relay_call_has_no_id(self): + """A NeMo Relay call has no id, since its guardrail callback receives none.""" + (tool_call,) = check_messages(ADAPTERS["nemo-relay"].parse(NEMO_RELAY_SEARCH))[0]["tool_calls"] + + assert tool_call == { + "type": "function", + "function": {"name": "search", "arguments": '{"query": "weather"}'}, + } + + def test_non_object_tool_input_is_encoded_as_sent(self): + """A `tool_input` that isn't an object is encoded as sent, for the engine to deny as malformed arguments.""" + call = ADAPTERS["claude-code"].parse({**CLAUDE_CODE_DESTRUCTIVE_BASH, "tool_input": "rm -rf /"}) + + (tool_call,) = check_messages(call)[0]["tool_calls"] + + assert tool_call["function"]["arguments"] == '"rm -rf /"' + + +class TestHookOutcome: + @pytest.mark.parametrize( + "fields", + [ + {"kind": "allow"}, + {"kind": "deny", "reason": "regex check tool call"}, + {"kind": "error", "message": "rail 'x' failed: x error", "code": HookErrorCode.RAIL_FAILED}, + {"kind": "error", "message": "unexpected check status 'modified'"}, + ], + ids=["allow", "deny", "error-with-code", "error-without-code"], + ) + def test_valid_shape_of_each_kind_constructs(self, fields): + """Each kind's valid shape constructs and keeps the fields it was given.""" + outcome = HookOutcome(**fields) + + assert dataclasses.asdict(outcome) == {"reason": None, "code": None, "message": None, **fields} + + @pytest.mark.parametrize( + "fields", + [ + {"kind": "allow", "reason": "regex check tool call"}, + {"kind": "allow", "code": HookErrorCode.RAIL_FAILED}, + {"kind": "allow", "message": "rail 'x' failed: x error"}, + {"kind": "deny"}, + {"kind": "deny", "reason": ""}, + {"kind": "deny", "reason": "regex check tool call", "code": HookErrorCode.RAIL_FAILED}, + {"kind": "deny", "reason": "regex check tool call", "message": "rail 'x' failed: x error"}, + {"kind": "error"}, + {"kind": "error", "message": ""}, + {"kind": "error", "message": "rail 'x' failed: x error", "reason": "regex check tool call"}, + {"kind": "error", "message": "rail 'x' failed: x error", "code": "not_a_hook_error_code"}, + {"kind": "block"}, + ], + ids=[ + "allow-with-reason", + "allow-with-code", + "allow-with-message", + "deny-without-reason", + "deny-with-empty-reason", + "deny-with-code", + "deny-with-message", + "error-without-message", + "error-with-empty-message", + "error-with-reason", + "error-with-unknown-code", + "unknown-kind", + ], + ) + def test_invalid_shape_raises_value_error(self, fields): + """A field the kind doesn't have, a missing or empty reason or message, or an unknown code or kind is refused.""" + with pytest.raises(ValueError): + HookOutcome(**fields) + + @pytest.mark.parametrize( + "fields", + [{"kind": "deny", "reason": 5}, {"kind": "error", "message": ["rail 'x' failed: x error"]}], + ids=["deny-reason", "error-message"], + ) + def test_reason_or_message_that_is_not_a_string_raises_type_error(self, fields): + """A deny's reason or an error's message that isn't a string raises `TypeError`.""" + with pytest.raises(TypeError): + HookOutcome(**fields) + + def test_code_given_as_its_string_value_becomes_the_enum_member(self): + """A code given as its string value is stored as its `HookErrorCode` member.""" + outcome = HookOutcome(kind="error", message="rail 'x' failed: x error", code="rail_failed") + + assert outcome.code is HookErrorCode.RAIL_FAILED + + def test_outcome_cannot_be_changed_after_construction(self): + """An outcome is immutable, so a valid one can't be turned into an invalid one.""" + outcome = HookOutcome.deny("regex check tool call") + + with pytest.raises(dataclasses.FrozenInstanceError): + outcome.reason = "" + + +class TestOutcomeFromResult: + def test_passed_is_allow(self): + """A passed check is an allow.""" + result = RailsResult(status=RailStatus.PASSED, content="") + + assert outcome_from_result(result) == HookOutcome.allow() + + @pytest.mark.parametrize( + "result, reason", + [ + ( + _blocked(UNDECLARED_TOOL, rail="tool call validation", reason=UNDECLARED_TOOL.reason), + "tool call 'WebFetch' is not an allowed tool", + ), + ( + _blocked(PER_TOOL_BLOCK, rail="regex check tool call", reason="regex check tool call"), + "regex check tool call", + ), + ( + _blocked(MALFORMED_ARGUMENTS, reason=MALFORMED_ARGUMENTS.reason), + "tool call 'toolu_01' has malformed arguments", + ), + ], + ids=["tool-call-validation", "per-tool-rail", "malformed-arguments"], + ) + def test_block_by_a_rail_is_deny_with_the_violation_reason(self, result, reason): + """A rail's block is a deny carrying the violation's reason.""" + assert outcome_from_result(result) == HookOutcome.deny(reason) + + def test_every_violation_rail_failed_is_rail_failed_error(self): + """A block where every violation is a rail failure is a `rail_failed` error naming the rail.""" + result = _blocked(RAIL_FAILED, rail="regex check tool call", reason=RAIL_FAILED.reason, content=INTERNAL_ERROR) + + assert outcome_from_result(result) == HookOutcome.error( + "rail 'regex check tool call' failed: regex check tool call error", code=HookErrorCode.RAIL_FAILED + ) + + def test_rail_failure_without_a_rail_name_is_still_rail_failed_error(self): + """A rail failure whose violation names no rail is still a `rail_failed` error, with the failure's reason.""" + unnamed = call_violation("rail_failed", "tool call validation error") + + assert outcome_from_result(_blocked(unnamed, content=INTERNAL_ERROR)) == HookOutcome.error( + "a tool-call rail failed: tool call validation error", code=HookErrorCode.RAIL_FAILED + ) + + @pytest.mark.parametrize( + "decided, reason", + [ + (PER_TOOL_BLOCK, "regex check tool call"), + (MALFORMED_ARGUMENTS, "tool call 'toolu_01' has malformed arguments"), + ], + ids=["per-tool-rail", "malformed-arguments"], + ) + def test_rail_failure_beside_a_decided_violation_is_deny(self, decided, reason): + """A rail failure beside a decided violation, a rail's block or an unparsable call, is a deny with its reason.""" + result = _blocked(RAIL_FAILED, decided, reason=RAIL_FAILED.reason, content=INTERNAL_ERROR) + + assert outcome_from_result(result) == HookOutcome.deny(reason) + + @pytest.mark.parametrize( + "failure, message", + [ + (EXTRACTION_FAILED, "the tool-call check failed: tool call extraction failed"), + (INVALID_TOOLSET, "the tool-call check failed: tool parsing failed"), + ( + INVALID_TOOL_SCHEMA, + "the tool-call check failed: declared schema for tool 'Bash' is not valid JSON Schema", + ), + ], + ids=["extraction-failed", "invalid-toolset", "invalid-tool-schema"], + ) + def test_check_failure_other_than_a_rail_is_error_without_a_code(self, failure, message): + """A block whose one violation says the check couldn't finish, though no rail failed, is an error with no code.""" + result = _blocked(failure, reason=failure.reason) + + assert outcome_from_result(result) == HookOutcome.error(message) + + def test_rail_failure_beside_another_check_failure_is_error_without_a_code(self): + """A block whose violations are all check failures, not all of them a rail's, is an error with no code.""" + result = _blocked(RAIL_FAILED, EXTRACTION_FAILED, reason=RAIL_FAILED.reason, content=INTERNAL_ERROR) + + assert outcome_from_result(result) == HookOutcome.error( + "the tool-call check failed: tool call extraction failed" + ) + + @pytest.mark.parametrize( + "failure", + [EXTRACTION_FAILED, INVALID_TOOLSET, INVALID_TOOL_SCHEMA], + ids=["extraction-failed", "invalid-toolset", "invalid-tool-schema"], + ) + def test_check_failure_beside_a_decided_violation_is_deny(self, failure): + """A check failure beside a rail's block is a deny with the block's reason, since a policy denial wins.""" + result = _blocked(failure, PER_TOOL_BLOCK, reason=failure.reason) + + assert outcome_from_result(result) == HookOutcome.deny("regex check tool call") + + @pytest.mark.parametrize( + "rail, reason, expected", + [ + ("tool call validation", "the result's reason", "the result's reason"), + ("tool call validation", None, "tool call validation"), + (None, None, "tool call blocked"), + ], + ) + def test_block_without_violations_is_deny(self, rail, reason, expected): + """A block reporting no violations is a deny, with the result's reason, else its rail, else a fixed reason.""" + assert outcome_from_result(_blocked(rail=rail, reason=reason)) == HookOutcome.deny(expected) + + @pytest.mark.parametrize( + "result, expected", + [ + (_blocked(EMPTY_REASON_BLOCK, PER_TOOL_BLOCK, reason="custom rail"), "regex check tool call"), + (_blocked(EMPTY_REASON_BLOCK, rail="custom rail", reason=""), "custom rail"), + ], + ids=["next-violation", "rail-name"], + ) + def test_empty_violation_reason_falls_through_to_the_next_reason(self, result, expected): + """A violation with an empty reason is skipped, so a deny's reason is never empty.""" + assert outcome_from_result(result) == HookOutcome.deny(expected) + + def test_modified_is_error_without_a_code(self): + """A modified result, which a tool-call check never gives, is an error with no code.""" + result = RailsResult(status=RailStatus.MODIFIED, content="rewritten") + + assert outcome_from_result(result) == HookOutcome.error("unexpected check status 'modified'") + + +class TestResponseBody: + @pytest.mark.parametrize( + "agent, outcome, expected", + [ + ("claude-code", HookOutcome.allow(), {}), + ( + "claude-code", + HookOutcome.deny("Destructive command blocked by hook"), + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "NeMo Guardrails: Destructive command blocked by hook", + } + }, + ), + ( + "claude-code", + HookOutcome.deny("tool call 'WebFetch' is not an allowed tool"), + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "NeMo Guardrails: tool call 'WebFetch' is not an allowed tool", + } + }, + ), + ( + "claude-code", + HookOutcome.deny("tool safety check tool call"), + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "NeMo Guardrails: tool safety check tool call", + } + }, + ), + ("codex", HookOutcome.allow(), {}), + ( + "codex", + HookOutcome.deny("Destructive command blocked by hook."), + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "NeMo Guardrails: Destructive command blocked by hook.", + } + }, + ), + ("nemo-relay", HookOutcome.allow(), {"decision": "allow"}), + ( + "nemo-relay", + HookOutcome.deny("search.query is required"), + {"decision": "deny", "reason": "search.query is required"}, + ), + ], + ids=[ + "claude-code-allow", + "claude-code-deny", + "claude-code-deny-not-allowed", + "claude-code-deny-per-tool", + "codex-allow", + "codex-deny", + "nemo-relay-allow", + "nemo-relay-deny", + ], + ) + def test_response_equals_spec_example(self, agent, outcome, expected): + """Each allow's and deny's response body is exactly the spec's response example for that path.""" + assert ADAPTERS[agent].response_body(outcome) == expected + + @pytest.mark.parametrize("agent", ["claude-code", "codex", "nemo-relay"]) + def test_error_outcome_has_no_200_body(self, agent): + """An error outcome has no `200` body on any path.""" + with pytest.raises(ValueError, match="error outcome"): + ADAPTERS[agent].response_body(HookOutcome.error("rail 'x' failed: x error", code=HookErrorCode.RAIL_FAILED)) + + +class TestCodexSchemas: + @pytest.mark.parametrize( + "outcome", [HookOutcome.allow(), HookOutcome.deny("regex check tool call")], ids=["allow", "deny"] + ) + @pytest.mark.parametrize("agent", ["claude-code", "codex"]) + def test_agent_responses_validate_against_codex_output_schema(self, agent, outcome): + """Every Claude Code and Codex response validates against Codex's output schema, which rejects unknown fields.""" + jsonschema.validate(ADAPTERS[agent].response_body(outcome), CODEX_OUTPUT_SCHEMA) + + def test_codex_request_requires_only_fields_codex_always_sends(self): + """Every field the Codex request model requires is one Codex's input schema requires, so Codex always sends it.""" + required = {name for name, field in CodexPreToolUseRequest.model_fields.items() if field.is_required()} + + assert required <= set(CODEX_INPUT_SCHEMA["required"]) + + def test_codex_request_declares_only_codex_fields(self): + """The Codex request model declares no field outside Codex's input schema.""" + assert set(CodexPreToolUseRequest.model_fields) <= set(CODEX_INPUT_SCHEMA["properties"]) + + @pytest.mark.parametrize("body", CODEX_EXAMPLES) + def test_spec_codex_examples_validate_against_codex_input_schema(self, body): + """Every Codex request example in the spec is a body Codex's input schema accepts.""" + jsonschema.validate(body, CODEX_INPUT_SCHEMA)