From aefdd35d9bd63c71b8d1d004baf770ecf5198ed7 Mon Sep 17 00:00:00 2001 From: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com> Date: Fri, 2 Oct 2026 23:23:40 +0100 Subject: [PATCH] fix(runtime): preserve resource denial diagnostics --- src/tool_execution.py | 4 +-- tests/test_wave3_diagnostics.py | 48 +++++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) create mode 100644 tests/test_wave3_diagnostics.py diff --git a/src/tool_execution.py b/src/tool_execution.py index f54eb9d47..4aa8c6be2 100644 --- a/src/tool_execution.py +++ b/src/tool_execution.py @@ -1425,7 +1425,7 @@ async def execute_tool_block( owner=owner, session_id=session_id, workspace=workspace, tool_name=getattr(block, "tool_type", None), content=getattr(block, "content", None))) admitted = valid and (authority.permits(operation) or exact_admission) - except (ValueError, TypeError, AttributeError) as error: + except (ValueError, TypeError) as error: return f"{getattr(block, 'tool_type', '')}: invalid arguments", { "error": (f"Tool arguments are not valid JSON: {error}" if isinstance(error, json.JSONDecodeError) else str(error)), @@ -1503,7 +1503,7 @@ async def execute_tool_block( authority, operation, document_id=active_document_id, approved=pending.owned_operation if pending is not None else None, exact_admission=exact_admission) - except (ValueError, TypeError, OSError, RuntimeError, AttributeError) as error: + except (ValueError, TypeError, OSError) as error: return f"{transport}: BLOCKED", { "error": str(error), "exit_code": 1, "blocked": True, "failure_kind": "resource_identity_denied", diff --git a/tests/test_wave3_diagnostics.py b/tests/test_wave3_diagnostics.py new file mode 100644 index 000000000..358a6a43e --- /dev/null +++ b/tests/test_wave3_diagnostics.py @@ -0,0 +1,48 @@ +"""Unexpected programming defects must not look like successful policy denial.""" +import pytest +from src import tool_execution +from src.agent_runtime.authority import ExactOperation, OperationGrant, RequestAuthority +from src.agent_runtime.resources import ResourceIdentityError +from src.tool_capabilities import ToolRunSecurityContext +from src.tool_types import ToolBlock + + +@pytest.mark.parametrize('seam,tool,content', [ + ('bind_backend_for_operation', 'bash', 'printf probe'), + ('resolve_process_operation', 'bash', 'printf probe'), + ('admit_owned_operation', 'edit_document', '{"document_id":"doc","content":"changed"}'), +]) +@pytest.mark.parametrize('error_type', [AttributeError, ResourceIdentityError, ValueError, TypeError]) +async def test_binding_errors_keep_diagnostic_identity(tmp_path, monkeypatch, seam, tool, content, error_type): + authority = RequestAuthority('request', 'alice', 'thread', str(tmp_path), (OperationGrant(tool),)) + monkeypatch.setattr(tool_execution, '_owner_is_admin', lambda owner: True) + def broken(*args, **kwargs): + raise error_type('injected defect') + monkeypatch.setattr(tool_execution, seam, broken) + args = dict(owner='alice', session_id='thread', workspace=str(tmp_path), request_authority=authority, + security_context=ToolRunSecurityContext()) + if error_type is AttributeError: + with pytest.raises(AttributeError, match='injected defect'): + await tool_execution.execute_tool_block(ToolBlock(tool, content), **args) + else: + _, result = await tool_execution.execute_tool_block(ToolBlock(tool, content), **args) + assert result['failure_kind'] == 'resource_identity_denied' and result['blocked'] + + +async def test_argument_normalization_defect_propagates(tmp_path, monkeypatch): + authority = RequestAuthority('request', 'alice', 'thread', str(tmp_path), (OperationGrant('bash'),)) + def broken(*args, **kwargs): raise AttributeError('internal-only diagnostic') + monkeypatch.setattr(ExactOperation, 'normalize', broken) + with pytest.raises(AttributeError): + await tool_execution.execute_tool_block(ToolBlock('bash', 'printf probe'), owner='alice', + session_id='thread', workspace=str(tmp_path), request_authority=authority, + security_context=ToolRunSecurityContext()) + + +@pytest.mark.parametrize('content', ['{invalid', None]) +async def test_expected_bad_input_still_has_authority_denial(tmp_path, content): + authority = RequestAuthority('request', 'alice', 'thread', str(tmp_path), (OperationGrant('api_call'),)) + _, result = await tool_execution.execute_tool_block(ToolBlock('api_call', content), owner='alice', + session_id='thread', workspace=str(tmp_path), request_authority=authority, + security_context=ToolRunSecurityContext()) + assert result['failure_kind'] == 'request_authority_denied'