glas-harness/tests/test_rein_aharness.py

import json
import subprocess
from unittest.mock import MagicMock, patch

import pytest

from glas_harness.contract import Rein, SandboxHandle, ToolCall
from glas_harness.reins.rein_aharness import ReinAharness, ReinAharnessNotInstalled


def test_is_rein_subclass() -> None:
    assert issubclass(ReinAharness, Rein)


def test_bin_raises_if_not_installed() -> None:
    rein = ReinAharness(cli_bin="does-not-exist-binary")
    with pytest.raises(ReinAharnessNotInstalled):
        rein._bin()


def test_start_session_writes_task_file_and_captures_head(tmp_path) -> None:
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
    subprocess.run(["git", "commit", "-q", "--allow-empty", "-m", "init"], cwd=repo, check=True)

    rein = ReinAharness()
    sandbox = SandboxHandle(sandbox_id="abc", host="localhost", reachability={"workspace_dir": str(repo)})
    session = rein.start_session(
        profile={"id": "harness.agent-dev-local"},
        inputs={"title": "t", "description": "d"},
        sandbox=sandbox,
    )

    assert session["target_repo"] == str(repo)
    assert session["head_before"]
    task_spec = json.loads(open(session["task_file"]).read())
    assert task_spec["title"] == "t"
    assert task_spec["target_repo"] == str(repo)


def test_start_session_requires_resolvable_target_repo() -> None:
    rein = ReinAharness()
    sandbox = SandboxHandle(sandbox_id="abc", host="localhost", reachability={})
    with pytest.raises(ValueError, match="target_repo"):
        rein.start_session(profile={}, inputs={"title": "t", "description": "d"}, sandbox=sandbox)


def test_dispatch_tool_invokes_agent_harness_cli() -> None:
    rein = ReinAharness()
    session = {"task_file": "/tmp/task.json"}

    fake_result = MagicMock(returncode=0, stdout="ok", stderr="")
    with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
        with patch("glas_harness.reins.rein_aharness.subprocess.run", return_value=fake_result) as run:
            result = rein.dispatch_tool(session, ToolCall(name="run_task"))

    argv = run.call_args.args[0]
    assert argv[:3] == ["/usr/bin/agent-harness", "run", "--task-file"]
    assert result.ok is True
    assert result.output == "ok"


def test_dispatch_tool_reports_failure() -> None:
    rein = ReinAharness()
    session = {"task_file": "/tmp/task.json"}

    fake_result = MagicMock(returncode=1, stdout="", stderr="boom")
    with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
        with patch("glas_harness.reins.rein_aharness.subprocess.run", return_value=fake_result):
            result = rein.dispatch_tool(session, ToolCall(name="run_task"))

    assert result.ok is False
    assert result.error == "boom"


def test_end_session_detects_new_commit(tmp_path) -> None:
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
    subprocess.run(["git", "commit", "-q", "--allow-empty", "-m", "init"], cwd=repo, check=True)

    from glas_harness.reins._shared import git_head

    rein = ReinAharness()
    head_before = git_head(str(repo))

    subprocess.run(["git", "commit", "-q", "--allow-empty", "-m", "task"], cwd=repo, check=True)

    summary = rein.end_session({"target_repo": str(repo), "head_before": head_before})
    assert summary["committed"] == "True"
    assert summary["commit_sha"] != head_before


def test_end_session_no_new_commit(tmp_path) -> None:
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
    subprocess.run(["git", "commit", "-q", "--allow-empty", "-m", "init"], cwd=repo, check=True)

    from glas_harness.reins._shared import git_head

    rein = ReinAharness()
    head = git_head(str(repo))

    summary = rein.end_session({"target_repo": str(repo), "head_before": head})
    assert summary["committed"] == "False"


def test_dispatch_tool_streams_events_when_enabled() -> None:
    import json

    rein = ReinAharness(stream_tool_events=True)
    session = {"task_file": "/tmp/task.json"}

    stdout_lines = [
        json.dumps({"stream_event": {"type": "assistant", "message": {"content": [{"type": "tool_use", "name": "Read"}]}}}),
        json.dumps({"stream_event": {"type": "system", "subtype": "hook_started"}}),
        json.dumps({"ok": True, "committed": True}, indent=2),
    ]
    fake_result = MagicMock(returncode=0, stdout="\n".join(stdout_lines), stderr="")

    with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
        with patch("glas_harness.reins.rein_aharness.subprocess.run", return_value=fake_result) as run:
            result = rein.dispatch_tool(session, ToolCall(name="run_task"))

    argv = run.call_args.args[0]
    assert "--stream-tool-events" in argv
    assert len(result.events) == 2
    assert result.events[0]["message"]["content"][0]["name"] == "Read"
    assert result.events[1]["subtype"] == "hook_started"
    assert '"ok": true' in result.output.lower()
    assert "stream_event" not in result.output


def test_dispatch_tool_no_stream_flag_by_default() -> None:
    rein = ReinAharness()
    session = {"task_file": "/tmp/task.json"}
    fake_result = MagicMock(returncode=0, stdout="ok", stderr="")

    with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
        with patch("glas_harness.reins.rein_aharness.subprocess.run", return_value=fake_result) as run:
            result = rein.dispatch_tool(session, ToolCall(name="run_task"))

    argv = run.call_args.args[0]
    assert "--stream-tool-events" not in argv
    assert result.events == []