Python · 1971 bytes Raw Blame History
1 """Runtime coverage for Sprint 04 workflow tools."""
2
3 from __future__ import annotations
4
5 import pytest
6
7 from loader.agent.loop import AgentConfig
8 from loader.llm.base import CompletionResponse, ToolCall
9 from tests.helpers.runtime_harness import ScriptedBackend, run_scenario
10
11
12 def non_streaming_config() -> AgentConfig:
13 """Shared deterministic config for runtime tool tests."""
14
15 return AgentConfig(
16 auto_context=False,
17 stream=False,
18 max_iterations=4,
19 workflow_mode_override="execute",
20 )
21
22
23 async def _answer(question: str, options: list[str] | None) -> str:
24 assert "Which path" in question
25 assert options == ["Plan first", "Execute now"]
26 return "1"
27
28
29 @pytest.mark.asyncio
30 async def test_ask_user_question_round_trips_through_runtime() -> None:
31 backend = ScriptedBackend(
32 completions=[
33 CompletionResponse(
34 content="I need one clarification.",
35 tool_calls=[
36 ToolCall(
37 id="ask-1",
38 name="AskUserQuestion",
39 arguments={
40 "question": "Which path should we take?",
41 "options": ["Plan first", "Execute now"],
42 },
43 )
44 ],
45 ),
46 CompletionResponse(content="We'll plan first."),
47 ]
48 )
49
50 run = await run_scenario(
51 "Implement the task, but ask me which path to take first.",
52 backend,
53 config=non_streaming_config(),
54 on_user_question=_answer,
55 )
56
57 tool_events = [event for event in run.events if event.type == "tool_call"]
58 tool_results = [event for event in run.events if event.type == "tool_result"]
59
60 assert "We'll plan first." in run.response
61 assert [event.tool_name for event in tool_events] == ["AskUserQuestion"]
62 assert any("Plan first" in event.content for event in tool_results)