Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions agent_assembly/adapters/openai_agents/patch.py
Original file line number Diff line number Diff line change
Expand Up @@ -435,8 +435,14 @@ async def _record_async_tool_result(


def _is_governance_error(error: Exception) -> bool:
del error
return True
"""Check if error is a governance-specific error that should be handled.

Non-governance errors (network, malformed response, etc.) should propagate
to maintain fail-closed security posture.
"""
from agent_assembly.exceptions import AssemblyError

return isinstance(error, AssemblyError)


def _apply_function_tool_patch(function_tool_cls: type[Any], callback_handler: Any) -> None:
Expand Down
41 changes: 37 additions & 4 deletions test/unit/adapters/openai_agents/test_patch.py
Original file line number Diff line number Diff line change
Expand Up @@ -280,9 +280,15 @@ async def record_result(self, **kwargs: object) -> None:


@pytest.mark.asyncio
async def test_gateway_error_uses_fail_open_and_executes_original_tool(
async def test_non_governance_error_propagates_fail_closed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Non-governance errors (RuntimeError, ValueError, etc.) must propagate.

This ensures fail-closed security posture: if an unexpected error occurs
during governance checks, the tool call is blocked rather than allowed
through. See AAASM-4252.
"""
function_tool_cls = _install_fake_openai_agents_module(monkeypatch)

class Interceptor:
Expand All @@ -293,25 +299,52 @@ async def check_tool_start(self, **kwargs: object) -> dict[str, str]:
patcher = openai_patch.OpenAIAgentsPatch(callback_handler=Interceptor())
assert patcher.apply() is True

tool = function_tool_cls(name="fail_closed_tool")
with pytest.raises(RuntimeError, match="gateway unavailable"):
await tool.on_invoke_tool(SimpleNamespace(agent_id="agent-fail-closed"), '{"x": 2}')


@pytest.mark.asyncio
async def test_governance_error_allows_fail_open(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""AssemblyError subclasses are handled gracefully (fail-open).

Governance-specific errors (e.g., transient gateway issues represented as
GatewayError) are caught and the tool executes. This is intentional: if
the governance system itself has a known error, we fail-open rather than
blocking all tool calls. See AAASM-4252.
"""
from agent_assembly.exceptions import GatewayError

function_tool_cls = _install_fake_openai_agents_module(monkeypatch)

class Interceptor:
async def check_tool_start(self, **kwargs: object) -> dict[str, str]:
del kwargs
raise GatewayError("gateway temporarily unavailable")

patcher = openai_patch.OpenAIAgentsPatch(callback_handler=Interceptor())
assert patcher.apply() is True

tool = function_tool_cls(name="fail_open_tool")
result = await tool.on_invoke_tool(SimpleNamespace(agent_id="agent-fail-open"), '{"x": 2}')

assert result["name"] == "fail_open_tool"


@pytest.mark.asyncio
async def test_non_governance_error_is_reraised(
async def test_value_error_propagates_fail_closed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""ValueError (non-governance) must propagate for fail-closed posture."""
function_tool_cls = _install_fake_openai_agents_module(monkeypatch)

class Interceptor:
async def check_tool_start(self, **kwargs: object) -> dict[str, str]:
del kwargs
raise ValueError("unexpected")

monkeypatch.setattr(openai_patch, "_is_governance_error", lambda _error: False)

patcher = openai_patch.OpenAIAgentsPatch(callback_handler=Interceptor())
assert patcher.apply() is True

Expand Down