mirror of
https://github.com/shareAI-lab/learn-claude-code.git
synced 2026-08-26 08:42:12 +00:00
331 lines
11 KiB
Python
331 lines
11 KiB
Python
import importlib.util
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
LESSONS = tuple(
|
|
ROOT / chapter / "code.py"
|
|
for chapter in (
|
|
"s01_agent_loop",
|
|
"s02_tool_use",
|
|
"s03_permission",
|
|
"s04_hooks",
|
|
"s05_todo_write",
|
|
"s06_subagent",
|
|
"s07_skill_loading",
|
|
"s08_context_compact",
|
|
"s09_memory",
|
|
"s10_task_system",
|
|
"s11_background_tasks",
|
|
"s12_cron_scheduler",
|
|
"s13_agent_teams",
|
|
"s14_mcp_plugin",
|
|
)
|
|
)
|
|
INTEGRATED_LESSON = ROOT / "s15_integrated_harness" / "code.py"
|
|
|
|
|
|
class FakeMessagesApi:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.calls = 0
|
|
|
|
def create(self, **_kwargs):
|
|
self.calls += 1
|
|
if not self.responses:
|
|
raise AssertionError("agent loop requested another model turn")
|
|
return self.responses.pop(0)
|
|
|
|
|
|
def load_lesson(workdir: Path, lesson_path: Path):
|
|
fake_anthropic = types.ModuleType("anthropic")
|
|
fake_dotenv = types.ModuleType("dotenv")
|
|
|
|
class FakeAnthropic:
|
|
def __init__(self, *args, **kwargs):
|
|
self.messages = FakeMessagesApi([])
|
|
|
|
fake_anthropic.Anthropic = FakeAnthropic
|
|
fake_dotenv.load_dotenv = lambda override=True: None
|
|
|
|
previous_modules = {
|
|
"anthropic": sys.modules.get("anthropic"),
|
|
"dotenv": sys.modules.get("dotenv"),
|
|
}
|
|
previous_cwd = Path.cwd()
|
|
previous_model = os.environ.get("MODEL_ID")
|
|
module_name = f"agent_loop_boundary_{lesson_path.parent.name}_{time.time_ns()}"
|
|
spec = importlib.util.spec_from_file_location(module_name, lesson_path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
|
|
sys.modules["anthropic"] = fake_anthropic
|
|
sys.modules["dotenv"] = fake_dotenv
|
|
sys.modules[module_name] = module
|
|
try:
|
|
os.chdir(workdir)
|
|
os.environ["MODEL_ID"] = "test-model"
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
finally:
|
|
os.chdir(previous_cwd)
|
|
if previous_model is None:
|
|
os.environ.pop("MODEL_ID", None)
|
|
else:
|
|
os.environ["MODEL_ID"] = previous_model
|
|
for name, previous in previous_modules.items():
|
|
if previous is None:
|
|
sys.modules.pop(name, None)
|
|
else:
|
|
sys.modules[name] = previous
|
|
sys.modules.pop(module_name, None)
|
|
|
|
|
|
def empty_tool_use_response(content=None):
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use",
|
|
content=(
|
|
[types.SimpleNamespace(type="text", text="")]
|
|
if content is None else content
|
|
),
|
|
)
|
|
|
|
|
|
def disable_lesson_side_effects(lesson):
|
|
if hasattr(lesson, "trigger_hooks"):
|
|
lesson.trigger_hooks = lambda *_args, **_kwargs: None
|
|
if hasattr(lesson, "inject_background_results"):
|
|
lesson.inject_background_results = lambda _messages: None
|
|
if hasattr(lesson, "consume_cron_queue"):
|
|
lesson.consume_cron_queue = lambda: []
|
|
if hasattr(lesson, "extract_memories"):
|
|
lesson.extract_memories = lambda _messages: False
|
|
if hasattr(lesson, "release_completed_assignment"):
|
|
lesson.release_completed_assignment = lambda _owner: None
|
|
if hasattr(lesson, "assemble_tool_pool"):
|
|
lesson.assemble_tool_pool = lambda: ([], {})
|
|
if hasattr(lesson, "assemble_system_prompt"):
|
|
lesson.assemble_system_prompt = lambda: "test system"
|
|
if hasattr(lesson, "COMPACTOR"):
|
|
lesson.COMPACTOR.prepare = lambda messages, _request: messages
|
|
|
|
|
|
def use_successful_bash_handler(lesson):
|
|
if hasattr(lesson, "run_bash"):
|
|
lesson.run_bash = lambda *_args, **_kwargs: "tool output"
|
|
if hasattr(lesson, "check_permission"):
|
|
lesson.check_permission = lambda _block: True
|
|
if hasattr(lesson, "execute_tool"):
|
|
lesson.execute_tool = lambda *_args, **_kwargs: "tool output"
|
|
if hasattr(lesson, "TOOL_HANDLERS"):
|
|
lesson.TOOL_HANDLERS["bash"] = lambda **_kwargs: "tool output"
|
|
|
|
|
|
def bash_tool_call():
|
|
return types.SimpleNamespace(
|
|
type="tool_use",
|
|
id="tool_1",
|
|
name="bash",
|
|
input={"command": "true"},
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("lesson_path", LESSONS, ids=lambda path: path.parent.name)
|
|
@pytest.mark.parametrize("content", ([], None), ids=("empty-content", "empty-text"))
|
|
def test_parent_loop_does_not_append_an_empty_tool_result_turn(
|
|
lesson_path: Path, content):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
disable_lesson_side_effects(lesson)
|
|
api = FakeMessagesApi([empty_tool_use_response(content)])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
messages = [{"role": "user", "content": "hello"}]
|
|
|
|
if lesson_path.parent.name == "s08_context_compact":
|
|
lesson.agent_loop(messages, "hello")
|
|
else:
|
|
lesson.agent_loop(messages)
|
|
|
|
assert api.calls == 1
|
|
assert messages[-1]["role"] == "assistant"
|
|
assert not any(
|
|
message.get("role") == "user" and message.get("content") == []
|
|
for message in messages
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("lesson_path", LESSONS, ids=lambda path: path.parent.name)
|
|
def test_parent_loop_executes_a_real_tool_call_even_if_stop_reason_disagrees(
|
|
lesson_path: Path):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
disable_lesson_side_effects(lesson)
|
|
use_successful_bash_handler(lesson)
|
|
api = FakeMessagesApi([
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[bash_tool_call()],
|
|
),
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(type="text", text="done")],
|
|
),
|
|
])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
messages = [{"role": "user", "content": "hello"}]
|
|
|
|
if lesson_path.parent.name == "s08_context_compact":
|
|
lesson.agent_loop(messages, "hello")
|
|
else:
|
|
lesson.agent_loop(messages)
|
|
|
|
assert api.calls == 2
|
|
tool_result_turns = [
|
|
message for message in messages
|
|
if message.get("role") == "user"
|
|
and isinstance(message.get("content"), list)
|
|
]
|
|
assert len(tool_result_turns) == 1
|
|
assert tool_result_turns[0]["content"][0]["tool_use_id"] == "tool_1"
|
|
|
|
|
|
def test_subagent_stops_without_an_empty_tool_result_turn():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), ROOT / "s06_subagent" / "code.py")
|
|
disable_lesson_side_effects(lesson)
|
|
api = FakeMessagesApi([empty_tool_use_response()])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
|
|
assert lesson.run_subagent("inspect the repository") == "(no summary)"
|
|
assert api.calls == 1
|
|
|
|
|
|
def test_subagent_still_executes_a_real_tool_call_with_text_present():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), ROOT / "s06_subagent" / "code.py")
|
|
disable_lesson_side_effects(lesson)
|
|
tool_call = types.SimpleNamespace(
|
|
type="tool_use",
|
|
id="tool_1",
|
|
name="read_file",
|
|
input={"path": "README.md"},
|
|
)
|
|
api = FakeMessagesApi([
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(type="text", text=""), tool_call],
|
|
),
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(type="text", text="done")],
|
|
),
|
|
])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
lesson.execute_tool = lambda _block, _handlers: "tool output"
|
|
|
|
assert lesson.run_subagent("inspect the repository") == "done"
|
|
assert api.calls == 2
|
|
|
|
|
|
def test_s13_teammate_does_not_continue_with_an_empty_tool_result_turn():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), ROOT / "s13_agent_teams" / "code.py")
|
|
api = FakeMessagesApi([empty_tool_use_response()])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
runtime = lesson.TeammateRuntime(
|
|
"alice", "reviewer", "inspect the repository", None, False
|
|
)
|
|
|
|
assert runtime.work() == "idle"
|
|
assert api.calls == 1
|
|
assert not any(
|
|
message.get("role") == "user" and message.get("content") == []
|
|
for message in runtime.messages
|
|
)
|
|
|
|
|
|
def test_s13_teammate_executes_a_real_tool_call_when_stop_reason_disagrees():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), ROOT / "s13_agent_teams" / "code.py")
|
|
api = FakeMessagesApi([
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[bash_tool_call()],
|
|
)
|
|
])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
lesson._run_teammate_tool = lambda *_args: "tool output"
|
|
runtime = lesson.TeammateRuntime(
|
|
"alice", "reviewer", "inspect the repository", None, False
|
|
)
|
|
|
|
assert runtime.work() == "continue"
|
|
assert api.calls == 1
|
|
assert runtime.messages[-1]["content"][0]["tool_use_id"] == "tool_1"
|
|
|
|
|
|
def stop_s15_teammate_when_idle(lesson, name: str):
|
|
deadline = time.monotonic() + 2
|
|
while time.monotonic() < deadline:
|
|
with lesson.team_lock:
|
|
state = lesson.active_teammates.get(name)
|
|
if state == "idle":
|
|
lesson.run_request_shutdown(name)
|
|
break
|
|
if state is None:
|
|
break
|
|
time.sleep(0.01)
|
|
|
|
deadline = time.monotonic() + 2
|
|
while time.monotonic() < deadline:
|
|
with lesson.team_lock:
|
|
if name not in lesson.active_teammates:
|
|
return
|
|
time.sleep(0.01)
|
|
|
|
|
|
def test_s15_teammate_does_not_request_another_turn_for_empty_tool_use():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), INTEGRATED_LESSON)
|
|
api = FakeMessagesApi([empty_tool_use_response()])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
|
|
lesson.spawn_teammate_thread("alice", "reviewer", "inspect the repository")
|
|
stop_s15_teammate_when_idle(lesson, "alice")
|
|
|
|
assert api.calls == 1
|
|
with lesson.team_lock:
|
|
assert "alice" not in lesson.active_teammates
|
|
|
|
|
|
def test_s15_teammate_executes_a_real_tool_call_when_stop_reason_disagrees():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), INTEGRATED_LESSON)
|
|
api = FakeMessagesApi([
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[bash_tool_call()],
|
|
),
|
|
types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(type="text", text="done")],
|
|
),
|
|
])
|
|
lesson.client = types.SimpleNamespace(messages=api)
|
|
calls = []
|
|
lesson._run_teammate_tool = lambda *_args: calls.append("bash") or "ok"
|
|
|
|
lesson.spawn_teammate_thread("alice", "reviewer", "inspect the repository")
|
|
stop_s15_teammate_when_idle(lesson, "alice")
|
|
|
|
assert api.calls == 2
|
|
assert calls == ["bash"]
|
|
with lesson.team_lock:
|
|
assert "alice" not in lesson.active_teammates
|