tegwick
6c4f00e2c2
Claude Code executes its own tools internally in --print mode -- there is no way for a caller to externally dispatch individual tool calls without abandoning that self-contained agent model. What --output-format stream-json --include-hook-events does allow: observing each tool_use/tool_result/hook event in real time. - adapter.py: AgenticClaudeCodeAdapter gains an optional on_tool_event callback; streaming mode (Popen + background reader thread) is used only when set, blocking subprocess.run path is unchanged otherwise. - runner.py: run_task gains emit_tool_events/on_tool_event, collecting events onto RunResult.tool_events and posting a tool_call hub event per tool when reporting is enabled. - cli.py: --stream-tool-events flag on `run`, prints each event as a tagged JSON line ahead of the unchanged final result block. Live-verified against the real claude CLI: 5 real tool events streamed correctly (2x Bash, 1x Write) plus Stop hook lifecycle events, real commit landed, final result block unchanged. 13 new tests (test_adapter.py + 2 in test_runner.py), all passing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
import json
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from llm_connect.exceptions import LLMSubprocessError, LLMTimeoutError
|
|
from llm_connect.models import RunConfig
|
|
|
|
from rein_aharness.adapter import AgenticClaudeCodeAdapter, _is_tool_event
|
|
|
|
|
|
def _fake_proc(lines: list[str], returncode: int = 0, stderr: str = "") -> MagicMock:
|
|
proc = MagicMock()
|
|
proc.stdin = MagicMock()
|
|
proc.stdout = iter(lines)
|
|
proc.stderr = MagicMock()
|
|
proc.stderr.read.return_value = stderr
|
|
proc.wait.return_value = returncode
|
|
return proc
|
|
|
|
|
|
def test_is_tool_event_true_for_tool_use() -> None:
|
|
event = {"type": "assistant", "message": {"content": [{"type": "tool_use", "name": "Read"}]}}
|
|
assert _is_tool_event(event) is True
|
|
|
|
|
|
def test_is_tool_event_true_for_tool_result() -> None:
|
|
event = {"type": "user", "message": {"content": [{"type": "tool_result", "content": "x"}]}}
|
|
assert _is_tool_event(event) is True
|
|
|
|
|
|
def test_is_tool_event_true_for_hook_lifecycle() -> None:
|
|
assert _is_tool_event({"type": "system", "subtype": "hook_started"}) is True
|
|
assert _is_tool_event({"type": "system", "subtype": "hook_response"}) is True
|
|
|
|
|
|
def test_is_tool_event_false_for_plain_text() -> None:
|
|
event = {"type": "assistant", "message": {"content": [{"type": "text", "text": "hi"}]}}
|
|
assert _is_tool_event(event) is False
|
|
|
|
|
|
def test_is_tool_event_false_for_init() -> None:
|
|
assert _is_tool_event({"type": "system", "subtype": "init"}) is False
|
|
|
|
|
|
def test_build_command_adds_stream_json_when_callback_set(tmp_path) -> None:
|
|
adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, on_tool_event=lambda e: None)
|
|
cmd = adapter._build_command(RunConfig(timeout_seconds=60))
|
|
assert "--output-format" in cmd
|
|
assert "stream-json" in cmd
|
|
assert "--include-hook-events" in cmd
|
|
|
|
|
|
def test_build_command_omits_stream_json_without_callback(tmp_path) -> None:
|
|
adapter = AgenticClaudeCodeAdapter(workdir=tmp_path)
|
|
cmd = adapter._build_command(RunConfig(timeout_seconds=60))
|
|
assert "--output-format" not in cmd
|
|
assert "--include-hook-events" not in cmd
|
|
|
|
|
|
def test_execute_streaming_invokes_callback_only_for_tool_events(tmp_path) -> None:
|
|
events: list[dict] = []
|
|
lines = [
|
|
json.dumps({"type": "system", "subtype": "init"}) + "\n",
|
|
json.dumps(
|
|
{"type": "assistant", "message": {"content": [{"type": "tool_use", "id": "t1", "name": "Read"}]}}
|
|
)
|
|
+ "\n",
|
|
json.dumps(
|
|
{"type": "user", "message": {"content": [{"type": "tool_result", "tool_use_id": "t1", "content": "hi"}]}}
|
|
)
|
|
+ "\n",
|
|
json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "Done."}]}}) + "\n",
|
|
json.dumps({"type": "result", "subtype": "success"}) + "\n",
|
|
]
|
|
adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, on_tool_event=events.append)
|
|
config = RunConfig(timeout_seconds=30)
|
|
|
|
with patch("rein_aharness.adapter.subprocess.Popen", return_value=_fake_proc(lines)):
|
|
response = adapter.execute_prompt("do it", config)
|
|
|
|
assert response.content == "Done."
|
|
assert response.metadata["tool_event_count"] == 2
|
|
assert len(events) == 2
|
|
assert events[0]["message"]["content"][0]["type"] == "tool_use"
|
|
assert events[1]["message"]["content"][0]["type"] == "tool_result"
|
|
|
|
|
|
def test_execute_streaming_raises_on_nonzero_exit(tmp_path) -> None:
|
|
adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, on_tool_event=lambda e: None)
|
|
config = RunConfig(timeout_seconds=30)
|
|
proc = _fake_proc([json.dumps({"type": "result", "subtype": "error"}) + "\n"], returncode=1, stderr="boom")
|
|
|
|
with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc):
|
|
with pytest.raises(LLMSubprocessError, match="claude CLI exited"):
|
|
adapter.execute_prompt("do it", config)
|
|
|
|
|
|
def test_execute_streaming_raises_timeout(tmp_path) -> None:
|
|
adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, on_tool_event=lambda e: None)
|
|
config = RunConfig(timeout_seconds=1)
|
|
proc = _fake_proc([])
|
|
proc.wait.side_effect = [subprocess.TimeoutExpired(cmd="claude", timeout=1), 0]
|
|
|
|
with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc):
|
|
with pytest.raises(LLMTimeoutError):
|
|
adapter.execute_prompt("do it", config)
|
|
proc.kill.assert_called_once()
|
|
|
|
|
|
def test_execute_blocking_path_unchanged_without_callback(tmp_path) -> None:
|
|
adapter = AgenticClaudeCodeAdapter(workdir=tmp_path)
|
|
config = RunConfig(timeout_seconds=30)
|
|
fake_result = MagicMock(returncode=0, stdout="plain output", stderr="")
|
|
|
|
with patch("rein_aharness.adapter.subprocess.run", return_value=fake_result) as run:
|
|
response = adapter.execute_prompt("do it", config)
|
|
|
|
assert response.content == "plain output"
|
|
argv = run.call_args.args[0]
|
|
assert "--output-format" not in argv
|