feat(harness): agent_loop foundation — non-interactive CLI driver (#30 PR1)

autoforge/harness/agent_loop.py

@@ -0,0 +1,140 @@
+"""Non-interactive driver for agent CLIs (Claude Code, Codex CLI).
+
+This module shells out to the user's installed agent CLI in non-interactive
+mode and returns a structured AgentResult. It is the deterministic execution
+layer for "let autoforge drive the agent itself" — distinct from harness mode
+where a human drives the agent and the agent calls autoforge-tools.
+
+Scope (PR1 / issue #30 task A):
+- Build argv for each backend (`claude -p` / `codex exec`)
+- Run the subprocess with timeout and capture stdout/stderr
+- Return AgentResult(success, exit_code, stdout, stderr, duration_seconds)
+
+Out of scope (later PRs in #30):
+- PATH auto-detection (PR2)
+- CLI integration (`autoforge loop --mode goals --backend ...`) (PR2)
+- goals dispatch from autoforge.yml (PR3)
+"""
+
+from __future__ import annotations
+
+import enum
+import subprocess
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class Backend(str, enum.Enum):
+    """Supported agent CLI backends.
+
+    Values are the executable names looked up on PATH at run time.
+    """
+
+    CLAUDE = "claude"
+    CODEX = "codex"
+
+
+@dataclass(frozen=True)
+class AgentResult:
+    """Outcome of one non-interactive agent invocation.
+
+    The caller decides what to do with stdout (parse / log / chain).
+    `success` is a convenience flag equivalent to `exit_code == 0`.
+    """
+
+    backend: Backend
+    success: bool
+    exit_code: int
+    stdout: str
+    stderr: str
+    duration_seconds: float
+
+
+def _argv_claude(prompt: str) -> list[str]:
+    """Argv for Claude Code non-interactive print mode (`-p`).
+
+    `claude -p "<prompt>"` runs the prompt without entering the chat REPL
+    and writes the assistant response to stdout.
+    """
+    return ["claude", "-p", prompt]
+
+
+def _argv_codex(prompt: str) -> list[str]:
+    """Argv for Codex CLI non-interactive mode (`exec`).
+
+    `codex exec "<prompt>"` runs the prompt and writes output to stdout
+    without opening the chat UI.
+    """
+    return ["codex", "exec", prompt]
+
+
+_BUILDERS = {
+    Backend.CLAUDE: _argv_claude,
+    Backend.CODEX: _argv_codex,
+}
+
+
+def build_argv(backend: Backend, prompt: str) -> list[str]:
+    """Return the full argv for invoking ``backend`` with ``prompt``."""
+    builder = _BUILDERS[backend]
+    return builder(prompt)
+
+
+# Sentinel exit codes used when subprocess never produced one of its own.
+_EXIT_TIMEOUT = -1
+_EXIT_NOT_FOUND = -2
+
+
+def run_agent(
+    backend: Backend,
+    prompt: str,
+    *,
+    cwd: Path | None = None,
+    timeout: int = 600,
+) -> AgentResult:
+    """Run the agent CLI non-interactively and return its outcome.
+
+    Captures stdout / stderr, normalises exceptions (timeout, missing CLI)
+    into structured AgentResult values rather than raising. This makes it
+    safe to call from a higher-level loop without try/except boilerplate.
+    """
+    argv = build_argv(backend, prompt)
+    started = time.monotonic()
+
+    try:
+        completed = subprocess.run(
+            argv,
+            cwd=str(cwd) if cwd else None,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        return AgentResult(
+            backend=backend,
+            success=False,
+            exit_code=_EXIT_TIMEOUT,
+            stdout="",
+            stderr=f"Timeout after {timeout}s",
+            duration_seconds=time.monotonic() - started,
+        )
+    except FileNotFoundError as exc:
+        return AgentResult(
+            backend=backend,
+            success=False,
+            exit_code=_EXIT_NOT_FOUND,
+            stdout="",
+            stderr=f"Backend CLI not found on PATH: {argv[0]} ({exc})",
+            duration_seconds=time.monotonic() - started,
+        )
+
+    return AgentResult(
+        backend=backend,
+        success=(completed.returncode == 0),
+        exit_code=completed.returncode,
+        stdout=completed.stdout,
+        stderr=completed.stderr,
+        duration_seconds=time.monotonic() - started,
+    )

tests/harness/test_agent_loop.py

@@ -0,0 +1,115 @@
+"""Tests for autoforge.harness.agent_loop — non-interactive agent driver."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from unittest.mock import patch
+
+from autoforge.harness.agent_loop import (
+    Backend,
+    _argv_claude,
+    _argv_codex,
+    build_argv,
+    run_agent,
+)
+
+
+class TestArgvBuilders:
+    """Snapshot tests for argv construction. Lock in the exact CLI form."""
+
+    def test_argv_claude_uses_print_mode(self) -> None:
+        assert _argv_claude("solve issue X") == ["claude", "-p", "solve issue X"]
+
+    def test_argv_codex_uses_exec_subcommand(self) -> None:
+        assert _argv_codex("solve issue X") == ["codex", "exec", "solve issue X"]
+
+    def test_build_argv_dispatches_claude(self) -> None:
+        assert build_argv(Backend.CLAUDE, "X") == ["claude", "-p", "X"]
+
+    def test_build_argv_dispatches_codex(self) -> None:
+        assert build_argv(Backend.CODEX, "X") == ["codex", "exec", "X"]
+
+
+class TestRunAgent:
+    """Exercise run_agent with mocked subprocess.run.
+
+    These tests verify run_agent's exception normalisation and result mapping
+    without depending on a real claude/codex binary on PATH.
+    """
+
+    def test_success_maps_to_agent_result(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=["claude", "-p", "hi"],
+            returncode=0,
+            stdout="agent reply",
+            stderr="",
+        )
+        with patch(
+            "autoforge.harness.agent_loop.subprocess.run",
+            return_value=completed,
+        ) as mock_run:
+            result = run_agent(Backend.CLAUDE, "hi")
+
+        assert result.backend is Backend.CLAUDE
+        assert result.success is True
+        assert result.exit_code == 0
+        assert result.stdout == "agent reply"
+        assert result.stderr == ""
+        assert result.duration_seconds >= 0
+
+        called_argv = mock_run.call_args[0][0]
+        assert called_argv == ["claude", "-p", "hi"]
+
+    def test_nonzero_exit_marks_failure(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=["codex", "exec", "X"],
+            returncode=2,
+            stdout="",
+            stderr="agent error",
+        )
+        with patch(
+            "autoforge.harness.agent_loop.subprocess.run",
+            return_value=completed,
+        ):
+            result = run_agent(Backend.CODEX, "X")
+
+        assert result.success is False
+        assert result.exit_code == 2
+        assert "agent error" in result.stderr
+
+    def test_timeout_returns_structured_result(self) -> None:
+        with patch(
+            "autoforge.harness.agent_loop.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(cmd="claude", timeout=5),
+        ):
+            result = run_agent(Backend.CLAUDE, "hi", timeout=5)
+
+        assert result.success is False
+        assert result.exit_code == -1
+        assert "Timeout after 5s" in result.stderr
+
+    def test_missing_cli_returns_structured_error(self) -> None:
+        with patch(
+            "autoforge.harness.agent_loop.subprocess.run",
+            side_effect=FileNotFoundError(2, "No such file or directory", "codex"),
+        ):
+            result = run_agent(Backend.CODEX, "X")
+
+        assert result.success is False
+        assert result.exit_code == -2
+        assert "not found on PATH" in result.stderr
+        assert "codex" in result.stderr
+
+    def test_cwd_is_passed_through_to_subprocess(self, tmp_path: Path) -> None:
+        completed = subprocess.CompletedProcess(
+            args=["claude", "-p", "x"], returncode=0, stdout="", stderr=""
+        )
+        with patch(
+            "autoforge.harness.agent_loop.subprocess.run",
+            return_value=completed,
+        ) as mock_run:
+            run_agent(Backend.CLAUDE, "x", cwd=tmp_path)
+
+        kwargs = mock_run.call_args.kwargs
+        assert kwargs["cwd"] == str(tmp_path)