from __future__ import annotations import json import os import socket import sqlite3 import stat import sys import threading import time from pathlib import Path from types import SimpleNamespace from unittest.mock import patch import pytest from adw_modules import agent_pi, quality from adw_modules.data_types import EventRecord, PiRequest, QualityCheckSpec from adw_modules.tracer import Tracer def _executable(path: Path, body: str) -> Path: path.write_text(f"#!{sys.executable}\n{body}") path.chmod(path.stat().st_mode | stat.S_IXUSR) return path def _request(tmp_path: Path) -> PiRequest: return PiRequest(prompt="prompt", system_prompt="system", model="fake", session_id="session-1", session_dir=str(tmp_path / "sessions"), raw_output_path=str(tmp_path / "raw.jsonl"), cwd=str(tmp_path)) def _traced_run(tmp_path: Path, program: Path, *, timeout: float = 5, idle_timeout: float = 600, agent_name: str = "builder"): tracer = Tracer(tmp_path / "trace.db", tmp_path / "events.jsonl") tracer.session_start("run-1", "test") events: list[dict] = [] tracker = agent_pi.ToolCallTracker() def start(info: dict) -> str: return tracer.agent_attempt_start("run-1", "phase-1", "builder", info["session_id"], info["command"], host=info["host"], account=info["account"], model=info["model"], parent_id="parent-1") def finish(attempt_id: object, info: dict) -> None: tracer.agent_attempt_finish(str(attempt_id), **info) def forward(event: dict) -> None: events.append(event) record = tracker.observe(event) if record is None: return tracer.event(EventRecord( adw_id="run-1", phase_id="phase-1", type=record.pop("event_type", "tool_call"), name=record.pop("label"), started_at=record.pop("started_at", None), ended_at=record.pop("ended_at", None), payload={**record, "agent": agent_name}, )) with patch.object(agent_pi, "PI_PATH", str(program)), \ patch.object(agent_pi, "resolve_model", return_value=("fake", "fake")), \ patch.object(agent_pi, "context_window", return_value=1000): result = agent_pi.run(_request(tmp_path), on_event=forward, on_attempt_start=start, on_attempt_end=finish, timeout_seconds=timeout, idle_timeout_seconds=idle_timeout, agent_name=agent_name) return tracer, result, events def test_nonzero_attempt_records_returncode_and_stderr_artifact(tmp_path: Path) -> None: program = _executable(tmp_path / "pi-fail", "import sys\nsys.stderr.write('provider exploded\\n')\nsys.exit(7)\n") with pytest.raises(RuntimeError, match="exited 7"): _traced_run(tmp_path, program) row = sqlite3.connect(tmp_path / "trace.db").execute( "SELECT returncode,signal,timed_out,stderr_path,parent_id,tokens FROM agent_attempts" ).fetchone() assert row[:3] == (7, None, 0) assert Path(row[3]).read_text() == "provider exploded\n" assert row[4] == "parent-1" assert row[5] is None def test_timeout_records_timeout_and_kills_process_group(tmp_path: Path) -> None: child_pid = tmp_path / "child.pid" program = _executable( tmp_path / "pi-hang", "import subprocess,sys,time\n" f"p=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)'])\n" f"f=open({str(child_pid)!r},'w')\nf.write(str(p.pid))\nf.flush()\nf.close()\ntime.sleep(60)\n", ) # Two interpreter startups must finish before the deadline fires. with pytest.raises(TimeoutError, match="timed out"): _traced_run(tmp_path, program, timeout=3.0) pid = int(child_pid.read_text()) deadline = time.monotonic() + 3 while Path(f"/proc/{pid}").exists() and time.monotonic() < deadline: time.sleep(0.02) assert not Path(f"/proc/{pid}").exists() row = sqlite3.connect(tmp_path / "trace.db").execute( "SELECT timed_out,returncode,signal,ended_at,timeout_kind FROM agent_attempts" ).fetchone() assert row[0] == 1 assert row[1] is not None assert row[3] is not None assert row[4] == "wall" def test_idle_timeout_preserves_completed_tool_calls(tmp_path: Path) -> None: events = [ {"type": "tool_execution_start", "toolCallId": "call-1", "toolName": "ls", "args": {"path": "specs"}}, {"type": "message_update", "assistantMessageEvent": { "type": "toolcall_end", "contentIndex": 4, "toolCall": {"type": "toolCall", "id": "call-1", "name": "ls", "arguments": {"path": "specs"}}}}, {"type": "tool_execution_end", "toolCallId": "call-1", "toolName": "ls", "args": {"path": "specs"}, "isError": False, "result": {"content": [{"type": "text", "text": "one\ntwo"}]}}, ] program = _executable( tmp_path / "pi-idle", "import json,time\n" f"events={events!r}\n" "time.sleep(.4)\n" "for event in events:\n print(json.dumps(event), flush=True)\n" "time.sleep(60)\n", ) with pytest.raises(TimeoutError, match="idle timeout"): _traced_run(tmp_path, program, timeout=5, idle_timeout=0.3) conn = sqlite3.connect(tmp_path / "trace.db") try: attempt = conn.execute( "SELECT timed_out,timeout_kind FROM agent_attempts" ).fetchone() tool = conn.execute( "SELECT tool_call_id,seq,tool_name,args_json,ended_at,ok,result_excerpt" " FROM tool_calls" ).fetchone() finally: conn.close() assert attempt == (1, "idle") assert tool[:4] == ("call-1", 0, "ls", '{"path": "specs"}') assert tool[4] is not None assert tool[5:] == (1, "one\ntwo") def test_role_timeout_resolution_precedence(monkeypatch) -> None: monkeypatch.delenv("FACTORY_AGENT_TIMEOUT_SECONDS", raising=False) monkeypatch.delenv("FACTORY_AGENT_TIMEOUT_SECONDS__PLANNER", raising=False) assert agent_pi._agent_timeout_seconds("planner") == 5400 assert agent_pi._agent_timeout_seconds("reviewer") == 3600 assert agent_pi._agent_timeout_seconds("builder") == 1800 monkeypatch.setenv("FACTORY_AGENT_TIMEOUT_SECONDS", "123") assert agent_pi._agent_timeout_seconds("planner") == 123 monkeypatch.setenv("FACTORY_AGENT_TIMEOUT_SECONDS__PLANNER", "456") assert agent_pi._agent_timeout_seconds("planner") == 456 def test_malformed_envelope_still_emits_completion_and_usage(tmp_path: Path) -> None: event = {"type": "message_end", "message": {"role": "assistant", "content": [{"type": "text", "text": "not an envelope"}], "usage": {"input": 3, "output": 2, "totalTokens": 5}, "stopReason": "stop"}} program = _executable(tmp_path / "pi-malformed", f"import json\nprint(json.dumps({event!r}), flush=True)\n") tracer, result, events = _traced_run(tmp_path, program) assert result.text == "not an envelope" completion = [item for item in events if item.get("type") == "agent_attempt_end"] assert len(completion) == 1 assert completion[0]["tokens"] == 5 row = tracer.conn.execute("SELECT tokens,usage_json,ended_at FROM agent_attempts").fetchone() assert row[0] == 5 assert json.loads(row[1])["total_tokens"] == 5 assert row[2] is not None def test_quality_process_is_live_then_cleared(tmp_path: Path) -> None: tracer = Tracer(tmp_path / "trace.db", tmp_path / "events.jsonl") tracer.session_start("run-1", "test") phase = SimpleNamespace(phase_id="phase-1", seq=1) run = SimpleNamespace(adw_id="run-1", repo_root=tmp_path, tracer=None, phases=[phase], context_handoff_dir=tmp_path / "artifacts", console=SimpleNamespace(note=lambda _message: None)) spec = QualityCheckSpec(name="test", area="backend", operation="build", argv=[sys.executable, "-c", "import time; time.sleep(.4)"], timeout_seconds=5) def execute() -> None: run.tracer = Tracer(tmp_path / "trace.db", tmp_path / "events.jsonl") quality._run(spec, run) thread = threading.Thread(target=execute) thread.start() conn = sqlite3.connect(tmp_path / "trace.db") deadline = time.monotonic() + 2 row = None while time.monotonic() < deadline: row = conn.execute( "SELECT pid,command,ended_at FROM processes WHERE kind='quality'" ).fetchone() if row: break time.sleep(0.01) assert row is not None and row[2] is None assert str(row[0]) and row[1] == quality.shlex.join(spec.argv) assert conn.execute("SELECT count(*) FROM events WHERE type='tool_call_start'").fetchone()[0] == 1 thread.join(timeout=3) assert not thread.is_alive() assert conn.execute("SELECT ended_at FROM processes WHERE kind='quality'").fetchone()[0] def test_unknown_usage_is_null_not_zero(tmp_path: Path) -> None: program = _executable(tmp_path / "pi-empty", "pass\n") tracer, _result, _events = _traced_run(tmp_path, program) assert tracer.conn.execute("SELECT tokens,usage_json FROM agent_attempts").fetchone() == (None, None) def test_attempt_round_trips_execution_identity_and_complete_argv(tmp_path: Path) -> None: program = _executable(tmp_path / "pi-ok", "pass\n") agent_dir = tmp_path / "agent" agent_dir.mkdir() auth = agent_dir / "auth.json" auth.write_text(json.dumps({"fake": {"accountId": "acct-real"}})) with patch.dict(os.environ, {"PI_CODING_AGENT_DIR": str(agent_dir)}): tracer, _result, _events = _traced_run(tmp_path, program) host, account, model, command = tracer.conn.execute( "SELECT host,account,model,command FROM agent_attempts" ).fetchone() assert host == socket.gethostname() assert account == "acct-real" assert model == "fake/fake" assert command == agent_pi.shlex.join([ str(program), "-p", "--mode", "json", "--provider", "fake", "--model", "fake", "--thinking", "medium", "--session-id", "session-1", "--session-dir", str(tmp_path / "sessions"), "--system-prompt", "system", "prompt", ])