tegwick
5681fce787
- src/glas_harness/contract.py: Rein ABC (start_session/dispatch_tool/
end_session), SandboxHandle/ToolCall/ToolResult, per docs/harness-contract.md.
- reins/rein_aharness.py: adapter around the rein-aharness CLI. Collapses
a whole `agent-harness run` into one dispatch_tool call for now (no
per-tool hooks yet — tracked in rein-aharness HARNESS-WP-0002-T03);
verifies success via new-commit detection, mirroring rein-aharness's
own success signal.
- gateway.py + cli.py: resolves a sand-boxer sandbox, runs one task
through a rein, tears the sandbox down.
- registry/reins/*.yaml (rein-aharness implemented, rein-openweights
planned) and profiles/harness.agent-dev{,-local}.yaml, pairing with
sand-boxer's profile.agent-dev and the new profile.bwrap-local.
Tested against a real local git repo + mocked SandboxManager/CLI
subprocess (10 tests, all passing). A real live run against the actual
Claude Code CLI is deliberately left for a human-triggered follow-up —
not executed autonomously since it spends real API credits/credentials.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock
|
|
|
|
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,
|
|
)
|
|
|
|
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,
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
manager.destroy.assert_called_once_with("sbx1")
|