From 422cc6797ebc049cc69720c9e4084798e2ada36f Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 15:11:38 +0800 Subject: [PATCH 1/7] =?UTF-8?q?=F0=9F=9A=A8=20(test):=20Mark=20patched=20R?= =?UTF-8?q?unner.run=20stubs=20positional-only=20for=20mypy=202.2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mypy 2.2.0 checks classmethod()'s argument against `def (type[Never], /, ...)`, so a named first parameter is rejected. The pre-commit mypy hook has been failing on remote/main since the bump, blocking every commit in this repo; the marker is runtime-inert (the argument was already positional). Unblocks AAASM-5529 --- test/unit/adapters/openai_agents/test_runner_spawn_patch.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/unit/adapters/openai_agents/test_runner_spawn_patch.py b/test/unit/adapters/openai_agents/test_runner_spawn_patch.py index e166b526..4d45b946 100644 --- a/test/unit/adapters/openai_agents/test_runner_spawn_patch.py +++ b/test/unit/adapters/openai_agents/test_runner_spawn_patch.py @@ -61,7 +61,7 @@ def teardown_method(self) -> None: async def test_patched_run_sets_spawn_ctx(self, monkeypatch: pytest.MonkeyPatch) -> None: captured: list[SpawnContext | None] = [] - async def capturing_run(agent: object, *, input: object, **kwargs: object) -> str: + async def capturing_run(agent: object, /, *, input: object, **kwargs: object) -> str: captured.append(_SPAWN_CTX.get()) return "done" @@ -79,7 +79,7 @@ async def capturing_run(agent: object, *, input: object, **kwargs: object) -> st @pytest.mark.asyncio async def test_spawn_ctx_is_reset_after_run(self, monkeypatch: pytest.MonkeyPatch) -> None: - async def passthrough_run(agent: object, *, input: object, **kwargs: object) -> str: + async def passthrough_run(agent: object, /, *, input: object, **kwargs: object) -> str: return "ok" monkeypatch.setattr(FakeRunner, "run", classmethod(passthrough_run)) @@ -90,7 +90,7 @@ async def passthrough_run(agent: object, *, input: object, **kwargs: object) -> @pytest.mark.asyncio async def test_spawn_ctx_reset_on_exception(self, monkeypatch: pytest.MonkeyPatch) -> None: - async def failing_run(agent: object, *, input: object, **kwargs: object) -> str: + async def failing_run(agent: object, /, *, input: object, **kwargs: object) -> str: raise RuntimeError("runner failed") monkeypatch.setattr(FakeRunner, "run", classmethod(failing_run)) From a7a9053254a0c4651fd1f77aaa63d19894c62d39 Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 15:11:52 +0800 Subject: [PATCH 2/7] =?UTF-8?q?=E2=9C=85=20(test):=20Add=20reusable=20enfo?= =?UTF-8?q?rcement-truth=20negative-control=20fixture?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real, externally-observable side effects (a file on disk, a live loopback HTTP listener) plus an audit-recording interceptor that delegates every verdict to the real one. Existing deny tests assert over a closure-captured executed list, which proves the SDK did not call a function it holds — not that the effect the tool exists to produce was prevented. Refs AAASM-5529, Epic AAASM-5526 --- test/unit/negative_control.py | 176 ++++++++++++++++++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 test/unit/negative_control.py diff --git a/test/unit/negative_control.py b/test/unit/negative_control.py new file mode 100644 index 00000000..d2f72f88 --- /dev/null +++ b/test/unit/negative_control.py @@ -0,0 +1,176 @@ +"""Reusable enforcement-truth negative-control fixture (AAASM-5529). + +A test that only asserts ``pytest.raises(PolicyViolationError)`` proves the SDK +produced a refusal, not that the refusal *prevented* anything: a tool whose body +has no observable effect would satisfy the same assertion. These helpers give a +denied tool a real, externally-observable effect — a file on disk, an HTTP +request delivered to a live loopback listener — so a deny can be asserted as the +*absence* of that effect and the matching allow as its *presence*. + +Every control built on this fixture is used as a pair: + +* **positive control** — policy allows, the effect is observed. Without it, + "no file on disk" is equally well explained by "the tool never ran at all", + and the negative control proves nothing. +* **negative control** — policy denies, the same effect is absent. + +The effects are deliberately real (``pathlib``, ``http.server``) rather than +``Mock`` call counters: a recorded call is evidence of intent, and Epic +AAASM-5526 exists because intent-level evidence is what over-claimed enforcement +looks like. +""" + +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from typing import Any +from urllib import request as urllib_request + + +@dataclass +class FileSideEffect: + """A filesystem-backed side effect rooted at ``path``. + + ``write`` really creates the file and ``occurred`` really stats it, so an + assertion over ``occurred()`` is an assertion about the world rather than + about the SDK's own bookkeeping. + """ + + path: Path + + def write(self, content: str) -> str: + self.path.write_text(content, encoding="utf-8") + return str(self.path) + + def occurred(self) -> bool: + return self.path.exists() + + def content(self) -> str | None: + if not self.path.exists(): + return None + return self.path.read_text(encoding="utf-8") + + +@dataclass +class ReceivedRequest: + method: str + path: str + body: str + + +class _RecordingHandler(BaseHTTPRequestHandler): + """Records every request instead of serving anything.""" + + received: list[ReceivedRequest] + + def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler contract + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length).decode("utf-8") if length else "" + type(self).received.append(ReceivedRequest(method="POST", path=self.path, body=body)) + self.send_response(204) + self.end_headers() + + def log_message(self, _format: str, *_args: Any) -> None: + """Silence the default stderr access log so test output stays readable.""" + return None + + +@dataclass +class NetworkSideEffect: + """A loopback HTTP listener that records every request it receives. + + A denied tool must leave ``requests`` empty. Because the positive control + exercises the same live listener, an empty log is evidence the egress did + not happen rather than evidence it could not have. + """ + + url: str + _server: HTTPServer + _thread: threading.Thread + requests: list[ReceivedRequest] = field(default_factory=list) + + def call(self, body: str) -> int: + req = urllib_request.Request(self.url, data=body.encode("utf-8"), method="POST") + with urllib_request.urlopen(req, timeout=5) as response: # noqa: S310 - fixed loopback URL + return int(response.status) + + def occurred(self) -> bool: + return len(self.requests) > 0 + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5) + + +def start_network_side_effect() -> NetworkSideEffect: + """Start a loopback listener on an ephemeral port and return its fixture.""" + received: list[ReceivedRequest] = [] + handler = type("_BoundRecordingHandler", (_RecordingHandler,), {"received": received}) + server = HTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return NetworkSideEffect( + url=f"http://127.0.0.1:{server.server_port}/exfiltrate", + _server=server, + _thread=thread, + requests=received, + ) + + +@dataclass +class RecordedResult: + """One post-execution audit record the governed path emitted.""" + + tool_name: str + agent_id: str | None + run_id: str | None + result: str + + +class AuditRecordingInterceptor: + """Delegates every governance call to ``inner`` and records the audit hook. + + Only the post-execution ``record_result`` hook is added — the authoritative + verdict still comes from the wrapped interceptor, so the deny under test is + the real one. This exists because the SDK's ``GatewayClient`` implements no + audit sink of its own (the interceptor is the only one), and AAASM-5529 + requires deny/allow evidence to carry agent and tool identity. + """ + + def __init__(self, inner: Any) -> None: + self._inner = inner + self.records: list[RecordedResult] = [] + + def check_tool_start(self, **kwargs: Any) -> Any: + return self._inner.check_tool_start(**kwargs) + + def record_result( + self, + *, + tool_name: str, + result: str, + agent_id: str | None = None, + run_id: str | None = None, + ) -> None: + self.records.append(RecordedResult(tool_name=tool_name, agent_id=agent_id, run_id=run_id, result=result)) + + def __getattr__(self, name: str) -> Any: + return getattr(self._inner, name) + + +def tool_args_of(query_call: tuple[Any, ...]) -> dict[str, Any]: + """Decode the ``tool_args_json`` the SDK presented to the native runtime. + + ``FakeRuntimeClient.query_calls`` entries are + ``(agent_id, action_type, tool_name, tool_args_json)``. + """ + raw = query_call[3] + if not raw: + return {} + decoded = json.loads(raw) + return decoded if isinstance(decoded, dict) else {} From 638d0164cc28262f185e7df6394f8ea43e0b49a2 Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 15:12:34 +0800 Subject: [PATCH 3/7] =?UTF-8?q?=E2=9C=85=20(test):=20Prove=20a=20denied=20?= =?UTF-8?q?quick-start=20tool=20writes=20no=20file=20to=20disk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Runs the real init_assembly so the interceptor under test is the RuntimeQueryInterceptor the SDK actually builds, then drives the shared governed-tool chain so the SDK — not the test — decides whether the body runs. The side-effect assertion precedes the exception assertion so removing the deny fails the suite on the absence check. Refs AAASM-5529, Epic AAASM-5526 --- test/unit/test_quickstart_negative_control.py | 180 ++++++++++++++++++ 1 file changed, 180 insertions(+) create mode 100644 test/unit/test_quickstart_negative_control.py diff --git a/test/unit/test_quickstart_negative_control.py b/test/unit/test_quickstart_negative_control.py new file mode 100644 index 00000000..4fb8ea96 --- /dev/null +++ b/test/unit/test_quickstart_negative_control.py @@ -0,0 +1,180 @@ +"""Enforcement-truth negative controls for the documented Python quick-start. + +AAASM-5529, Epic AAASM-5526. + +``docs/quick-start.md`` tells a reader that after ``init_assembly(...)`` "every +tool call from now on goes through the policy gate", and that a denied tool +surfaces as a ``ToolExecutionBlockedError``. Existing tests of that claim assert +either the returned verdict dict or an empty ``executed`` list captured by a +closure. Neither shows that the *effect the tool exists to produce* was +prevented. + +Each control here therefore: + +1. runs the real ``init_assembly`` so the interceptor under test is the one the + SDK actually builds (a genuine ``RuntimeQueryInterceptor`` over a fake native + runtime, not a hand-rolled stand-in); +2. drives the SDK's own governed-call chain, ``run_governed_async_tool`` — the + shared pre-execution gate behind the Google ADK and Pydantic AI quick-start + tabs — so the SDK, not the test, decides whether the tool body runs; +3. asserts a real side effect (a file on disk, an HTTP request delivered to a + live loopback listener) is present under allow and absent under deny. + +The side-effect assertion is made *before* the exception assertion on purpose. +Asserting the exception first would short-circuit the test when enforcement is +removed, leaving the absence check unexercised — the falsification run would +then only ever prove "no error was raised", which is precisely the weak evidence +this suite replaces. + +The ``FALSIFICATION`` cases run the identical function with governance removed. +They must observe the side effect; if they stop doing so, every deny assertion +here has become vacuous. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any + +import pytest + +from agent_assembly import init_assembly +from agent_assembly.adapters._shared.tool_governance import run_governed_async_tool +from agent_assembly.core import assembly as core_assembly +from agent_assembly.core.runtime_interceptor import build_governance_interceptor +from agent_assembly.exceptions import ToolExecutionBlockedError + +from .core._fake_core import FakeRuntimeClient, install_fake_core +from .negative_control import AuditRecordingInterceptor, FileSideEffect + +# A non-loopback https gateway: the register-endpoint TLS guard (AAASM-4655) +# fail-closes a plaintext http:// register channel to a non-loopback host. +_GW_URL = "https://gateway.test" +_API_KEY = "test-key" +_AGENT_ID = "quickstart-negative-control-agent" + + +@pytest.fixture(autouse=True) +def _cleanup_active_context() -> None: + """Release the process-singleton context so each control inits cleanly.""" + active = core_assembly._ACTIVE_CONTEXT + if active is not None and not active.is_shutdown: + active.shutdown() + core_assembly._ACTIVE_CONTEXT = None + + +@pytest.fixture +def file_effect(tmp_path: Path) -> FileSideEffect: + return FileSideEffect(path=tmp_path / "denied-write.txt") + + +class _GovernedQuickStart: + """The real ``init_assembly`` context plus the interceptor it produced.""" + + def __init__(self, context: Any, interceptor: AuditRecordingInterceptor, runtime: FakeRuntimeClient) -> None: + self.context = context + self.interceptor = interceptor + self.runtime = runtime + + def call(self, tool_name: str, tool_args: dict[str, Any], body: Any) -> Any: + """Run ``body`` through the SDK's governed-tool chain.""" + return asyncio.run( + run_governed_async_tool( + self.interceptor, + enforce=True, + tool_name=tool_name, + tool_args=tool_args, + agent_id=_AGENT_ID, + run_id="run-1", + invoke_original=body, + ) + ) + + +def _init_quickstart(monkeypatch: pytest.MonkeyPatch, *, decision: str, reason: str = "") -> _GovernedQuickStart: + """Run the documented ``init_assembly`` and capture the interceptor it built. + + ``build_governance_interceptor`` is wrapped rather than replaced, so the real + ``_register_adapters`` runs and the interceptor handed back is the SDK's own. + """ + runtime = FakeRuntimeClient(decision=decision, reason=reason) + install_fake_core(monkeypatch, runtime) + monkeypatch.setattr( + core_assembly, + "_start_network_layer", + lambda **_kwargs: ("sdk-only", core_assembly._noop_shutdown), + ) + + captured: list[AuditRecordingInterceptor] = [] + + def _spy(*args: Any, **kwargs: Any) -> Any: + interceptor = AuditRecordingInterceptor(build_governance_interceptor(*args, **kwargs)) + captured.append(interceptor) + return interceptor + + monkeypatch.setattr(core_assembly, "build_governance_interceptor", _spy) + + context = init_assembly( + gateway_url=_GW_URL, + api_key=_API_KEY, + agent_id=_AGENT_ID, + mode="sdk-only", + enforcement_mode="enforce", + ) + assert captured, "init_assembly did not build a governance interceptor" + return _GovernedQuickStart(context, captured[0], runtime) + + +def _settle(call: Any) -> Any: + """Run ``call`` and return its result or its exception, never raising. + + Keeps the side-effect assertion reachable even when the governed call + unexpectedly succeeds — see the module docstring. + """ + try: + return call() + except Exception as error: # noqa: BLE001 - the exception is the value under test + return error + + +class TestFilesystemSideEffect: + def test_positive_control_allowed_write_creates_the_file( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="allow") + try: + quickstart.call("write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("allowed")) + finally: + quickstart.context.shutdown() + + assert file_effect.occurred() is True + assert file_effect.content() == "allowed" + + def test_negative_control_denied_write_leaves_no_file( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="deny", reason="policy forbids disk writes") + try: + outcome = _settle( + lambda: quickstart.call( + "write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("denied") + ) + ) + finally: + quickstart.context.shutdown() + + # The load-bearing assertion: the effect the tool exists to produce is + # absent from the filesystem, not merely that an error was raised. + assert file_effect.occurred() is False + assert file_effect.content() is None + assert isinstance(outcome, ToolExecutionBlockedError) + assert "policy forbids disk writes" in str(outcome) + + def test_falsification_the_same_write_ungoverned_creates_the_file(self, file_effect: FileSideEffect) -> None: + # No init_assembly, no interceptor — enforcement removed. If this does + # not write, the negative control above is vacuous. + file_effect.write("ungoverned") + + assert file_effect.occurred() is True + assert file_effect.content() == "ungoverned" From e60c4ad56a89d3a966b8792e4b9257e4a09d4fc7 Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 15:13:02 +0800 Subject: [PATCH 4/7] =?UTF-8?q?=E2=9C=85=20(test):=20Prove=20a=20denied=20?= =?UTF-8?q?quick-start=20tool=20sends=20no=20HTTP=20request?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A real loopback listener records every request it receives, so the deny is asserted as zero deliveries rather than as a raised exception. The positive control on the same live fixture establishes reachability, which is what makes the empty request log evidence of prevention. Refs AAASM-5529, Epic AAASM-5526 --- test/unit/test_quickstart_negative_control.py | 65 ++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) diff --git a/test/unit/test_quickstart_negative_control.py b/test/unit/test_quickstart_negative_control.py index 4fb8ea96..aee2c6a3 100644 --- a/test/unit/test_quickstart_negative_control.py +++ b/test/unit/test_quickstart_negative_control.py @@ -46,7 +46,12 @@ from agent_assembly.exceptions import ToolExecutionBlockedError from .core._fake_core import FakeRuntimeClient, install_fake_core -from .negative_control import AuditRecordingInterceptor, FileSideEffect +from .negative_control import ( + AuditRecordingInterceptor, + FileSideEffect, + NetworkSideEffect, + start_network_side_effect, +) # A non-loopback https gateway: the register-endpoint TLS guard (AAASM-4655) # fail-closes a plaintext http:// register channel to a non-loopback host. @@ -69,6 +74,15 @@ def file_effect(tmp_path: Path) -> FileSideEffect: return FileSideEffect(path=tmp_path / "denied-write.txt") +@pytest.fixture +def network_effect() -> Any: + effect = start_network_side_effect() + try: + yield effect + finally: + effect.close() + + class _GovernedQuickStart: """The real ``init_assembly`` context plus the interceptor it produced.""" @@ -178,3 +192,52 @@ def test_falsification_the_same_write_ungoverned_creates_the_file(self, file_eff assert file_effect.occurred() is True assert file_effect.content() == "ungoverned" + + +class TestNetworkSideEffect: + def test_positive_control_allowed_egress_reaches_the_listener( + self, monkeypatch: pytest.MonkeyPatch, network_effect: NetworkSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="allow") + try: + status = quickstart.call( + "send_http_request", + {"url": network_effect.url}, + lambda: network_effect.call("allowed-payload"), + ) + finally: + quickstart.context.shutdown() + + assert status == 204 + assert len(network_effect.requests) == 1 + assert network_effect.requests[0].body == "allowed-payload" + + def test_negative_control_denied_egress_never_reaches_the_listener( + self, monkeypatch: pytest.MonkeyPatch, network_effect: NetworkSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="deny", reason="egress denied") + try: + outcome = _settle( + lambda: quickstart.call( + "send_http_request", + {"url": network_effect.url}, + lambda: network_effect.call("denied-payload"), + ) + ) + finally: + quickstart.context.shutdown() + + # The listener is live and was reachable throughout — the positive + # control proves that on the same fixture — so zero received requests is + # evidence the egress did not happen, not that it could not have. + assert network_effect.occurred() is False + assert network_effect.requests == [] + assert isinstance(outcome, ToolExecutionBlockedError) + + def test_falsification_the_same_egress_ungoverned_reaches_the_listener( + self, network_effect: NetworkSideEffect + ) -> None: + assert network_effect.call("ungoverned-payload") == 204 + + assert network_effect.occurred() is True + assert network_effect.requests[0].body == "ungoverned-payload" From 44a48df35a4a5bb9a22a0246c58e874421ed2ae4 Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 15:13:26 +0800 Subject: [PATCH 5/7] =?UTF-8?q?=E2=9C=85=20(test):=20Assert=20a=20deny=20c?= =?UTF-8?q?arries=20the=20agent=20and=20tool=20identity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Identity is read from the query the SDK presented to the authoritative runtime, not reconstructed by the test, and the allow control checks the same triple reaches the post-execution audit hook. An anonymous refusal is not usable evidence. Refs AAASM-5529, Epic AAASM-5526 --- test/unit/test_quickstart_negative_control.py | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/test/unit/test_quickstart_negative_control.py b/test/unit/test_quickstart_negative_control.py index aee2c6a3..93ac8bd8 100644 --- a/test/unit/test_quickstart_negative_control.py +++ b/test/unit/test_quickstart_negative_control.py @@ -51,6 +51,7 @@ FileSideEffect, NetworkSideEffect, start_network_side_effect, + tool_args_of, ) # A non-loopback https gateway: the register-endpoint TLS guard (AAASM-4655) @@ -241,3 +242,46 @@ def test_falsification_the_same_egress_ungoverned_reaches_the_listener( assert network_effect.occurred() is True assert network_effect.requests[0].body == "ungoverned-payload" + + +class TestDenyIsAttributable: + def test_the_runtime_saw_the_agent_and_tool_the_deny_was_decided_against( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="deny", reason="policy forbids disk writes") + try: + outcome = _settle( + lambda: quickstart.call( + "write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("denied") + ) + ) + finally: + quickstart.context.shutdown() + + assert isinstance(outcome, ToolExecutionBlockedError) + assert file_effect.occurred() is False + + # Identity as presented to the authoritative policy query, not as the + # test reconstructed it: an anonymous deny is not usable evidence. + assert len(quickstart.runtime.query_calls) == 1 + agent_id, action_type, tool_name, _args_json = quickstart.runtime.query_calls[0] + assert agent_id == _AGENT_ID + assert action_type == "tool_call" + assert tool_name == "write_to_disk" + assert tool_args_of(quickstart.runtime.query_calls[0]) == {"path": str(file_effect.path)} + + def test_an_allowed_call_is_recorded_with_the_same_identity( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="allow") + try: + quickstart.call("write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("allowed")) + finally: + quickstart.context.shutdown() + + assert file_effect.occurred() is True + assert len(quickstart.interceptor.records) == 1 + record = quickstart.interceptor.records[0] + assert record.tool_name == "write_to_disk" + assert record.agent_id == _AGENT_ID + assert record.run_id == "run-1" From 9de955901eebfe11538e0c32dae7a48b5cc58379 Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 15:13:48 +0800 Subject: [PATCH 6/7] =?UTF-8?q?=E2=9C=85=20(test):=20Prove=20a=20missing?= =?UTF-8?q?=20native=20runtime=20denies=20rather=20than=20passes=20through?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under enforce with no agent_assembly._core installed the SDK has no authoritative verdict source. AAASM-5526 forbids that degraded path presenting as protected, so the control holds it to the same standard as every other one here: the file the tool would have written is absent. Refs AAASM-5529, Epic AAASM-5526 --- test/unit/test_quickstart_negative_control.py | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/test/unit/test_quickstart_negative_control.py b/test/unit/test_quickstart_negative_control.py index 93ac8bd8..76d7e6e4 100644 --- a/test/unit/test_quickstart_negative_control.py +++ b/test/unit/test_quickstart_negative_control.py @@ -285,3 +285,57 @@ def test_an_allowed_call_is_recorded_with_the_same_identity( assert record.tool_name == "write_to_disk" assert record.agent_id == _AGENT_ID assert record.run_id == "run-1" + + +class TestDegradedRuntimeCannotLookProtected: + def test_an_unavailable_native_runtime_denies_rather_than_silently_allowing( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + """Under enforce, no native extension must block — not pass through. + + AAASM-5526 forbids a degraded path presenting as protected. The control + proves the posture is real by the same standard as every other one here: + the side effect is absent. + """ + monkeypatch.setattr( + core_assembly, + "_start_network_layer", + lambda **_kwargs: ("sdk-only", core_assembly._noop_shutdown), + ) + captured: list[Any] = [] + + def _spy(*args: Any, **kwargs: Any) -> Any: + interceptor = build_governance_interceptor(*args, **kwargs) + captured.append(interceptor) + return interceptor + + monkeypatch.setattr(core_assembly, "build_governance_interceptor", _spy) + + # No install_fake_core: agent_assembly._core is absent, so the SDK has no + # authoritative verdict source at all. + context = init_assembly( + gateway_url=_GW_URL, + api_key=_API_KEY, + agent_id=_AGENT_ID, + mode="sdk-only", + enforcement_mode="enforce", + ) + try: + outcome = _settle( + lambda: asyncio.run( + run_governed_async_tool( + captured[0], + enforce=True, + tool_name="write_to_disk", + tool_args={"path": str(file_effect.path)}, + agent_id=_AGENT_ID, + run_id="run-1", + invoke_original=lambda: file_effect.write("degraded"), + ) + ) + ) + finally: + context.shutdown() + + assert file_effect.occurred() is False + assert isinstance(outcome, ToolExecutionBlockedError) From a4027f2effe660980cfeaa677a40caef7c4fea1d Mon Sep 17 00:00:00 2001 From: Bryant Date: Thu, 6 Aug 2026 19:58:00 +0800 Subject: [PATCH 7/7] =?UTF-8?q?=E2=9C=85=20(test):=20Assert=20the=20absent?= =?UTF-8?q?=20side=20effect=20before=20the=20blocked-tool=20error?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestDenyIsAttributable asserted isinstance(outcome, ToolExecutionBlockedError) ahead of the absence check, so the failed assert aborted the test before the side effect was ever inspected. Under the falsification mutation that control failed on "no error was raised" — the weak evidence this suite exists to replace — while its three siblings correctly failed on the side effect. Swapped to match the shape the other controls already use. --- test/unit/test_quickstart_negative_control.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/unit/test_quickstart_negative_control.py b/test/unit/test_quickstart_negative_control.py index 76d7e6e4..493a212f 100644 --- a/test/unit/test_quickstart_negative_control.py +++ b/test/unit/test_quickstart_negative_control.py @@ -258,8 +258,11 @@ def test_the_runtime_saw_the_agent_and_tool_the_deny_was_decided_against( finally: quickstart.context.shutdown() - assert isinstance(outcome, ToolExecutionBlockedError) + # Absence first, as in every other control here: an exception assertion + # placed ahead of it aborts the test before the side effect is checked, + # so the absence would never be exercised by the falsification run. assert file_effect.occurred() is False + assert isinstance(outcome, ToolExecutionBlockedError) # Identity as presented to the authoritative policy query, not as the # test reconstructed it: an anonymous deny is not usable evidence.