From 228a0a2566cdbb72c0f27f7885124ab30b513330 Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam <51058514+Sanjays2402@users.noreply.github.com> Date: Sat, 2 May 2026 13:47:24 -0700 Subject: [PATCH] test(run_agent): align concurrent-interrupt _Stub with current AIAgent interface `_execute_tool_calls_concurrent` has gained two attribute / call-site dependencies that the lightweight `_Stub` test fixture didn't carry: 1. `self._tool_guardrails.before_call(...)` is now consulted before each tool runs, and the result's `.allows_execution` gate decides whether to invoke the tool. The stub had no `_tool_guardrails` attribute, so both interrupt-fanout tests crashed with `AttributeError: '_Stub' object has no attribute '_tool_guardrails'`. 2. `self._append_guardrail_observation(name, args, result, failed=...)` is called for non-blocked results to fold guardrail observations into the tool output. The stub had no such method. Additionally, `_invoke_tool` is now invoked with `messages=` and `pre_tool_block_checked=` kwargs by the concurrent path, so the test fixtures' `slow_tool` / `polling_tool` callables raised `TypeError: ... got an unexpected keyword argument 'pre_tool_block_checked'`. Fix all three drifts: - `_tool_guardrails` is set to a permissive `MagicMock` whose `before_call`/`after_call` return objects with `allows_execution=True` (these tests aren't exercising guardrails). - `_append_guardrail_observation` is stubbed as a no-op pass-through. - `slow_tool` / `polling_tool` accept `**kwargs` so any future `_invoke_tool` plumbing is absorbed without yet another rewrite. The interrupt-fanout invariants the tests *do* exercise (worker-thread tid registration, `is_interrupted()` propagation across ThreadPoolExecutor workers, post-clear cleanup) are unchanged. No production code change. Fixes the two assertion failures observed on `main` (run 25250051126): `tests/run_agent/test_concurrent_interrupt.py::test_concurrent_interrupt_cancels_pending` `tests/run_agent/test_concurrent_interrupt.py::test_running_concurrent_worker_sees_is_interrupted` --- tests/run_agent/test_concurrent_interrupt.py | 25 ++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/tests/run_agent/test_concurrent_interrupt.py b/tests/run_agent/test_concurrent_interrupt.py index 9a6ba73e7e4d..5714b1c75b5a 100644 --- a/tests/run_agent/test_concurrent_interrupt.py +++ b/tests/run_agent/test_concurrent_interrupt.py @@ -48,6 +48,17 @@ class _Stub: # real interrupt() method can fan out to concurrent-tool workers. _active_children: list = [] + # PR #18108 added guardrail gating to ``_execute_tool_calls_concurrent``; + # the stub doesn't exercise that path so a permissive mock is enough, + # but the attribute must exist or AttributeError aborts the test + # before the interrupt logic under test runs. + _tool_guardrails = MagicMock( + **{ + "before_call.return_value": MagicMock(allows_execution=True), + "after_call.return_value": MagicMock(allows_execution=True), + } + ) + def __init__(self): # Instance-level (not class-level) so each test gets a fresh set. self._tool_worker_threads: set = set() @@ -81,6 +92,10 @@ def _has_stream_consumers(self): # tool batch. Stub it as a no-op — this test exercises interrupt # fanout, not steer injection. stub._apply_pending_steer_to_tool_results = lambda *a, **kw: None + # Guardrail post-processing was added to the concurrent path; the + # interrupt tests don't exercise guardrails so just pass results + # through unchanged. + stub._append_guardrail_observation = lambda name, args, result, failed=False: result stub._invoke_tool = MagicMock(side_effect=lambda *a, **kw: '{"ok": true}') return stub @@ -107,7 +122,10 @@ def test_concurrent_interrupt_cancels_pending(monkeypatch): original_invoke = agent._invoke_tool - def slow_tool(name, args, task_id, call_id=None): + def slow_tool(name, args, task_id, call_id=None, **kwargs): + # ``_execute_tool_calls_concurrent`` now forwards ``messages=`` and + # ``pre_tool_block_checked=`` into ``_invoke_tool``; absorb any future + # plumbing kwargs without forcing a test rewrite each time. if name == "slow_one": # Block until the test sets the interrupt barrier.wait(timeout=10) @@ -184,7 +202,10 @@ def test_running_concurrent_worker_sees_is_interrupted(monkeypatch): observed = {"saw_true": False, "poll_count": 0, "worker_tid": None} worker_started = threading.Event() - def polling_tool(name, args, task_id, call_id=None, messages=None): + def polling_tool(name, args, task_id, call_id=None, messages=None, **kwargs): + # Absorb any future ``_invoke_tool`` plumbing kwargs (e.g. + # ``pre_tool_block_checked``) without forcing this fixture to + # track signature drift. observed["worker_tid"] = threading.current_thread().ident worker_started.set() deadline = time.monotonic() + 5.0