glas-harness/tests/test_gateway.py

from datetime import UTC, datetime
from unittest.mock import MagicMock, patch

from sandboxer.models import Reachability, SandboxState, SandboxStatus

from glas_harness.contract import Rein, SandboxHandle, ToolCall, ToolResult
from glas_harness.gateway import run_task_through_rein


class _FakeRein(Rein):
    def __init__(self) -> None:
        self.calls: list[str] = []

    def start_session(self, profile, inputs, sandbox: SandboxHandle):
        self.calls.append("start_session")
        assert sandbox.sandbox_id == "sbx1"
        assert sandbox.reachability.get("workspace_dir") == "/tmp/ws"
        return {"session": "s1"}

    def dispatch_tool(self, session, tool_call: ToolCall) -> ToolResult:
        self.calls.append("dispatch_tool")
        assert tool_call.name == "run_task"
        return ToolResult(ok=True, output="done")

    def end_session(self, session):
        self.calls.append("end_session")
        return {"commit_sha": "deadbeef", "committed": "True"}


def _fake_status(sandbox_id: str = "sbx1") -> SandboxStatus:
    now = datetime.now(UTC)
    return SandboxStatus(
        sandbox_id=sandbox_id,
        profile_id="profile.bwrap-local",
        extension_id="ext.bwrap",
        state=SandboxState.READY,
        consumer={"actor": "agt", "project": "glas-harness"},
        host="localhost",
        reachability=Reachability(host="localhost", pid="123", workspace_dir="/tmp/ws"),
        created_at=now,
        updated_at=now,
    )


def test_run_task_through_rein_creates_and_destroys_sandbox() -> None:
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    rein = _FakeRein()

    result = run_task_through_rein(
        sandbox_profile="profile.bwrap-local",
        repo="/tmp/repo",
        title="t",
        description="d",
        rein=rein,
        manager=manager,
        report_to_hub=False,
    )

    assert rein.calls == ["start_session", "dispatch_tool", "end_session"]
    manager.create.assert_called_once()
    manager.destroy.assert_called_once_with("sbx1")
    assert result["tool_ok"] is True
    assert result["summary"]["committed"] == "True"


def test_run_task_through_rein_destroys_sandbox_even_on_failure() -> None:
    manager = MagicMock()
    manager.create.return_value = _fake_status()

    class _FailingRein(_FakeRein):
        def dispatch_tool(self, session, tool_call):
            raise RuntimeError("boom")

    rein = _FailingRein()

    try:
        run_task_through_rein(
            sandbox_profile="profile.bwrap-local",
            repo="/tmp/repo",
            title="t",
            description="d",
            rein=rein,
            manager=manager,
            report_to_hub=False,
        )
    except RuntimeError:
        pass

    manager.destroy.assert_called_once_with("sbx1")


def test_run_task_through_rein_reports_success_event() -> None:
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    rein = _FakeRein()

    with patch("glas_harness.gateway.hub.post_progress_event", return_value=True) as post:
        run_task_through_rein(
            sandbox_profile="profile.bwrap-local",
            repo="/tmp/repo",
            title="t",
            description="d",
            rein=rein,
            manager=manager,
        )

    post.assert_called_once()
    kwargs = post.call_args.kwargs
    assert kwargs["event_type"] == "gateway_run"
    assert "ok)" in kwargs["summary"]
    assert kwargs["detail"]["ok"] is True
    assert kwargs["detail"]["sandbox_id"] == "sbx1"
    assert kwargs["detail"]["rein"] == "_FakeRein"


def test_run_task_through_rein_reports_failure_event_and_still_raises() -> None:
    manager = MagicMock()
    manager.create.return_value = _fake_status()

    class _FailingRein(_FakeRein):
        def dispatch_tool(self, session, tool_call):
            raise RuntimeError("boom")

    rein = _FailingRein()

    with patch("glas_harness.gateway.hub.post_progress_event", return_value=True) as post:
        try:
            run_task_through_rein(
                sandbox_profile="profile.bwrap-local",
                repo="/tmp/repo",
                title="t",
                description="d",
                rein=rein,
                manager=manager,
            )
            assert False, "expected RuntimeError to propagate"
        except RuntimeError:
            pass

    post.assert_called_once()
    kwargs = post.call_args.kwargs
    assert "failed)" in kwargs["summary"]
    assert kwargs["detail"]["ok"] is False
    assert kwargs["detail"]["error"] == "boom"
    assert kwargs["detail"]["result"] is None


def test_run_task_through_rein_skips_hub_when_disabled() -> None:
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    rein = _FakeRein()

    with patch("glas_harness.gateway.hub.post_progress_event") as post:
        run_task_through_rein(
            sandbox_profile="profile.bwrap-local",
            repo="/tmp/repo",
            title="t",
            description="d",
            rein=rein,
            manager=manager,
            report_to_hub=False,
        )

    post.assert_not_called()