From 85d0ee8a13c212d62565c6e9535835497f390aef Mon Sep 17 00:00:00 2001 From: ciregenz Date: Tue, 23 Jun 2026 02:30:53 -0700 Subject: [PATCH] [eric] agents: harness test pins the loop's hook wiring (hook_ctx + 4 wrappers) --- backend/tests/test_streaming_harness.py | 38 +++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/backend/tests/test_streaming_harness.py b/backend/tests/test_streaming_harness.py index df5a0b0b..a1f304d7 100644 --- a/backend/tests/test_streaming_harness.py +++ b/backend/tests/test_streaming_harness.py @@ -53,6 +53,44 @@ def _assistant(blocks, **kw): return AssistantMessage(**base) +def test_loop_wires_all_four_hooks_to_a_live_hook_context(monkeypatch): + # Integration coverage the unit tests can't give: capture the ClaudeAgentOptions the real + # loop hands to query(), then invoke the WIRED hooks. This proves _run_agent_loop builds a + # HookContext (all required fields, incl. the live `sessions` registry) and the four thin + # wrappers delegate to the extracted hook modules. The SDK never fires these under a mocked + # query, so without this the wiring (not just the functions) would be untested. + captured = {} + + async def capturing_query(*args, **kwargs): + captured["options"] = kwargs.get("options") + yield _assistant([TextBlock(text="ok")]) + yield _result() + + async def fake_send(session_id, event, data): + pass + + monkeypatch.setattr(ws_mod.ws_manager, "send_to_session", fake_send, raising=True) + monkeypatch.setattr(claude_agent_sdk, "query", capturing_query, raising=True) + + mgr = AgentManager() + from backend.apps.agents.core.models import AgentSession + session = AgentSession(name="t", model="sonnet", dashboard_id="d") + mgr.sessions[session.id] = session + asyncio.run(mgr._run_agent_loop(session.id, "hi")) + + options = captured["options"] + assert options is not None + assert callable(options.can_use_tool) + pre = options.hooks["PreToolUse"][0].hooks + post = options.hooks["PostToolUse"][0].hooks + stop = options.hooks["Stop"][0].hooks + assert pre and post and stop + + # Invoke the wired Stop hook: a non-view-builder session short-circuits to {} by reading + # ctx.session.mode, so this drives the full wrapper -> hook_ctx -> stop_hook module path. + assert asyncio.run(stop[0]({}, None, None)) == {} + + def test_streamed_text_lands_as_assistant_message(monkeypatch): session, events = _drive(monkeypatch, [ _assistant([TextBlock(text="Hello there")]),