| 1 | """Tests for explicit runtime turn-phase tracking.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from pathlib import Path |
| 6 | |
| 7 | import pytest |
| 8 | |
| 9 | from loader.agent.loop import AgentConfig |
| 10 | from loader.llm.base import CompletionResponse, ToolCall |
| 11 | from tests.helpers.runtime_harness import ScriptedBackend, run_scenario |
| 12 | |
| 13 | |
| 14 | def _config(*, completion_check: bool = False) -> AgentConfig: |
| 15 | config = AgentConfig(auto_context=False, stream=False, max_iterations=8) |
| 16 | config.reasoning.completion_check = completion_check |
| 17 | return config |
| 18 | |
| 19 | |
| 20 | def _turn_phases(run) -> list[str]: |
| 21 | return [ |
| 22 | event.turn_phase |
| 23 | for event in run.events |
| 24 | if event.type == "turn_phase" and event.turn_phase |
| 25 | ] |
| 26 | |
| 27 | |
| 28 | def _turn_phase_events(run) -> list: |
| 29 | return [event for event in run.events if event.type == "turn_phase"] |
| 30 | |
| 31 | |
| 32 | @pytest.mark.asyncio |
| 33 | async def test_empty_output_enters_repair_phase(temp_dir: Path) -> None: |
| 34 | backend = ScriptedBackend( |
| 35 | completions=[ |
| 36 | CompletionResponse(content=""), |
| 37 | CompletionResponse(content="I can help with that now."), |
| 38 | ] |
| 39 | ) |
| 40 | |
| 41 | run = await run_scenario( |
| 42 | "Inspect the workspace and summarize it briefly.", |
| 43 | backend, |
| 44 | config=_config(completion_check=False), |
| 45 | project_root=temp_dir, |
| 46 | ) |
| 47 | |
| 48 | phases = _turn_phases(run) |
| 49 | repair_event = next( |
| 50 | event |
| 51 | for event in _turn_phase_events(run) |
| 52 | if event.turn_phase == "repair" |
| 53 | ) |
| 54 | assert "repair" in phases |
| 55 | assert phases[:3] == ["prepare", "assistant", "repair"] |
| 56 | assert phases[-2:] == ["completion", "finalize"] |
| 57 | assert repair_event.transition_kind == "retry" |
| 58 | assert repair_event.transition_reason_code == "repair_empty_response" |
| 59 | assert run.agent.last_turn_summary is not None |
| 60 | assert run.agent.last_turn_summary.last_turn_transition_summary == ( |
| 61 | "completion -> finalize [terminal] Finalizing completed turn" |
| 62 | ) |
| 63 | assert run.agent.session.active_turn_phase is None |
| 64 | assert run.agent.session.last_turn_transition_reason_code == "turn_complete" |
| 65 | |
| 66 | |
| 67 | @pytest.mark.asyncio |
| 68 | async def test_completion_nudge_and_tool_batch_emit_named_phases( |
| 69 | temp_dir: Path, |
| 70 | monkeypatch: pytest.MonkeyPatch, |
| 71 | ) -> None: |
| 72 | monkeypatch.chdir(temp_dir) |
| 73 | target = temp_dir / "hello.py" |
| 74 | backend = ScriptedBackend( |
| 75 | completions=[ |
| 76 | CompletionResponse(content="I looked into it."), |
| 77 | CompletionResponse( |
| 78 | content="You're right, I'll create the file first.", |
| 79 | tool_calls=[ |
| 80 | ToolCall( |
| 81 | id="write-1", |
| 82 | name="write", |
| 83 | arguments={ |
| 84 | "file_path": str(target), |
| 85 | "content": "print('hello from loader')\n", |
| 86 | }, |
| 87 | ) |
| 88 | ], |
| 89 | ), |
| 90 | CompletionResponse( |
| 91 | content="Now I'll run the script.", |
| 92 | tool_calls=[ |
| 93 | ToolCall( |
| 94 | id="bash-1", |
| 95 | name="bash", |
| 96 | arguments={"command": f"python {target.name}"}, |
| 97 | ) |
| 98 | ], |
| 99 | ), |
| 100 | CompletionResponse(content="Successfully created and ran hello.py."), |
| 101 | ] |
| 102 | ) |
| 103 | |
| 104 | run = await run_scenario( |
| 105 | "Create a hello.py file and run it.", |
| 106 | backend, |
| 107 | config=_config(completion_check=True), |
| 108 | project_root=temp_dir, |
| 109 | ) |
| 110 | |
| 111 | phases = _turn_phases(run) |
| 112 | assert "completion" in phases |
| 113 | assert "tools" in phases |
| 114 | assert phases[0] == "prepare" |
| 115 | assert phases[-1] == "finalize" |
| 116 | assert run.agent.last_turn_summary is not None |
| 117 | assert run.agent.last_turn_summary.last_turn_transition_summary == ( |
| 118 | "completion -> finalize [terminal] Finalizing completed turn" |
| 119 | ) |
| 120 | assert run.agent.session.last_turn_transition_reason_code == "turn_complete" |
| 121 | assert any(event.type == "completion_check" for event in run.events) |