tegwick
695438019c
Some checks failed
ci / validate (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0233b-178d-7162-b92f-31a31ea8ca9b
325 lines
10 KiB
Python
325 lines
10 KiB
Python
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from sandboxer.models import Reachability, SandboxState, SandboxStatus
|
|
|
|
from glas_harness.contract import (
|
|
ExecutionRequest,
|
|
ExecutionSummary,
|
|
OperationalReadiness,
|
|
Rein,
|
|
SandboxHandle,
|
|
ToolCall,
|
|
ToolResult,
|
|
)
|
|
from glas_harness.gateway import run_execution, run_task_through_rein
|
|
from glas_harness.profiles import ProfileCatalog
|
|
|
|
|
|
PROFILE = "harness.agent-dev-local@1.0.0"
|
|
|
|
|
|
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 str(profile.ref) == PROFILE
|
|
assert sandbox.sandbox_id == "sbx1"
|
|
assert sandbox.reachability.get("workspace_dir") == "/tmp/ws"
|
|
assert "target_repo" not in inputs
|
|
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="sensitive direct output",
|
|
events=[{"type": "tool_use"}],
|
|
events_completeness="complete",
|
|
tokens_spent=123,
|
|
resolved_model="claude-sonnet-4-6",
|
|
)
|
|
|
|
def end_session(self, session):
|
|
self.calls.append("end_session")
|
|
return ExecutionSummary(
|
|
committed=True,
|
|
commit_sha="deadbeef",
|
|
outcome="succeeded",
|
|
tokens_spent=123,
|
|
resolved_model="claude-sonnet-4-6",
|
|
)
|
|
|
|
def cleanup_session(self, session):
|
|
self.calls.append("cleanup_session")
|
|
|
|
|
|
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 _request(*, profile: str = PROFILE, report_to_hub: bool = False) -> ExecutionRequest:
|
|
return ExecutionRequest(
|
|
harness_profile_ref=profile,
|
|
repo="/tmp/repo",
|
|
title="t",
|
|
description="d",
|
|
request_id="req-1",
|
|
correlation_id="corr-1",
|
|
assignment_ref="assignment:42",
|
|
report_to_hub=report_to_hub,
|
|
)
|
|
|
|
|
|
def _catalog_with_readiness(status: str = "ready") -> ProfileCatalog:
|
|
catalog = ProfileCatalog()
|
|
profile, _ = catalog.resolve(PROFILE)
|
|
readiness = OperationalReadiness(
|
|
status=status,
|
|
reason="test-only readiness state",
|
|
owner="tests",
|
|
evidence_ref="test:gateway-readiness",
|
|
)
|
|
catalog.profiles()[(profile.id, profile.version)] = profile.model_copy(
|
|
update={"operational_readiness": readiness}
|
|
)
|
|
return catalog
|
|
|
|
|
|
def test_run_execution_creates_and_destroys_sandbox() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
rein = _FakeRein()
|
|
|
|
result = run_execution(
|
|
_request(), catalog=_catalog_with_readiness(), rein=rein, manager=manager
|
|
)
|
|
|
|
assert rein.calls == [
|
|
"start_session",
|
|
"dispatch_tool",
|
|
"end_session",
|
|
"cleanup_session",
|
|
]
|
|
manager.create.assert_called_once()
|
|
manager.destroy.assert_called_once_with("sbx1")
|
|
assert result.ok is True
|
|
assert result.evidence.profile_ref == PROFILE
|
|
assert result.evidence.profile_readiness.status == "ready"
|
|
assert result.evidence.rein_id == "rein-aharness"
|
|
assert result.evidence.commit_sha == "deadbeef"
|
|
assert result.evidence.tokens_spent == 123
|
|
assert result.evidence.refs["assignment_ref"] == "assignment:42"
|
|
assert result.tool_output == "sensitive direct output"
|
|
|
|
|
|
def test_run_execution_normalizes_execution_failure_and_tears_down() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
|
|
class _FailingRein(_FakeRein):
|
|
def dispatch_tool(self, session, tool_call):
|
|
raise RuntimeError("boom")
|
|
|
|
rein = _FailingRein()
|
|
result = run_execution(
|
|
_request(), catalog=_catalog_with_readiness(), rein=rein, manager=manager
|
|
)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.outcome == "failed"
|
|
assert result.evidence.failure_stage == "execution"
|
|
assert result.evidence.error == "execution failed; inspect direct caller error"
|
|
assert result.tool_error == "boom"
|
|
assert rein.calls == ["start_session", "cleanup_session"]
|
|
manager.destroy.assert_called_once_with("sbx1")
|
|
|
|
|
|
def test_run_execution_refuses_unknown_profile_before_sandbox() -> None:
|
|
manager = MagicMock()
|
|
|
|
result = run_execution(_request(profile="harness.unknown@1.0.0"), manager=manager)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.outcome == "refused"
|
|
assert result.evidence.failure_stage == "resolution"
|
|
assert "unknown harness profile" in (result.evidence.error or "")
|
|
manager.create.assert_not_called()
|
|
|
|
|
|
def test_run_execution_refuses_disabled_profile_before_sandbox() -> None:
|
|
manager = MagicMock()
|
|
catalog = ProfileCatalog()
|
|
profile, _ = catalog.resolve(PROFILE)
|
|
catalog.profiles()[(profile.id, profile.version)] = profile.model_copy(
|
|
update={"status": "disabled"}
|
|
)
|
|
|
|
result = run_execution(_request(), catalog=catalog, rein=_FakeRein(), manager=manager)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.outcome == "refused"
|
|
assert result.evidence.failure_stage == "resolution"
|
|
assert "profile disabled" in (result.evidence.error or "")
|
|
manager.create.assert_not_called()
|
|
|
|
|
|
def test_run_execution_refuses_ambiguous_profile_before_sandbox() -> None:
|
|
manager = MagicMock()
|
|
catalog = ProfileCatalog()
|
|
profile, _ = catalog.resolve(PROFILE)
|
|
catalog.profiles()[(profile.id, "2.0.0")] = profile.model_copy(
|
|
update={"version": "2.0.0"}
|
|
)
|
|
|
|
result = run_execution(
|
|
_request(profile=profile.id),
|
|
catalog=catalog,
|
|
rein=_FakeRein(),
|
|
manager=manager,
|
|
)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.outcome == "refused"
|
|
assert result.evidence.failure_stage == "resolution"
|
|
assert "pin one of" in (result.evidence.error or "")
|
|
manager.create.assert_not_called()
|
|
|
|
|
|
def test_run_execution_refuses_blocked_profile_before_sandbox() -> None:
|
|
manager = MagicMock()
|
|
|
|
result = run_execution(_request(), rein=_FakeRein(), manager=manager)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.outcome == "refused"
|
|
assert result.evidence.failure_stage == "resolution"
|
|
assert result.evidence.profile_readiness.status == "blocked"
|
|
assert "GLAS-IN-0002" in (result.evidence.error or "")
|
|
manager.create.assert_not_called()
|
|
|
|
|
|
def test_run_execution_allows_labeled_unverified_proof_attempt() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
|
|
result = run_execution(
|
|
_request(),
|
|
catalog=_catalog_with_readiness("unverified"),
|
|
rein=_FakeRein(),
|
|
manager=manager,
|
|
)
|
|
|
|
assert result.ok is True
|
|
assert result.evidence.profile_readiness.status == "unverified"
|
|
manager.create.assert_called_once()
|
|
|
|
|
|
def test_run_execution_refuses_worker_identifier_before_sandbox() -> None:
|
|
manager = MagicMock()
|
|
request = _request().model_copy(update={"actor": "rein-aharness@railiance01"})
|
|
|
|
result = run_execution(request, rein=_FakeRein(), manager=manager)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.outcome == "refused"
|
|
assert result.evidence.failure_stage == "resolution"
|
|
assert result.evidence.error == (
|
|
"execution actor must be a governed consumer type: adm, agt, or atm; "
|
|
"queue worker identifiers are not execution actors"
|
|
)
|
|
manager.create.assert_not_called()
|
|
|
|
|
|
def test_hub_receives_normalized_evidence_without_raw_output() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
|
|
with patch("glas_harness.gateway.hub.post_progress_event", return_value=True) as post:
|
|
result = run_execution(
|
|
_request(report_to_hub=True),
|
|
catalog=_catalog_with_readiness(),
|
|
rein=_FakeRein(),
|
|
manager=manager,
|
|
)
|
|
|
|
post.assert_called_once()
|
|
detail = post.call_args.kwargs["detail"]
|
|
assert detail["outcome"] == "succeeded"
|
|
assert detail["profile_ref"] == PROFILE
|
|
assert "tool_output" not in detail
|
|
assert "sensitive direct output" not in str(detail)
|
|
assert result.tool_output == "sensitive direct output"
|
|
|
|
|
|
def test_hub_failure_detail_excludes_raw_provider_error() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
|
|
class _FailingRein(_FakeRein):
|
|
def dispatch_tool(self, session, tool_call):
|
|
raise RuntimeError("provider body containing sensitive material")
|
|
|
|
with patch("glas_harness.gateway.hub.post_progress_event", return_value=True) as post:
|
|
result = run_execution(
|
|
_request(report_to_hub=True),
|
|
catalog=_catalog_with_readiness(),
|
|
rein=_FailingRein(),
|
|
manager=manager,
|
|
)
|
|
|
|
detail = post.call_args.kwargs["detail"]
|
|
assert detail["error"] == "execution failed; inspect direct caller error"
|
|
assert "sensitive material" not in str(detail)
|
|
assert "sensitive material" in (result.tool_error or "")
|
|
|
|
|
|
def test_wrapper_requires_and_reports_harness_profile() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
|
|
result = run_task_through_rein(
|
|
harness_profile=PROFILE,
|
|
repo="/tmp/repo",
|
|
title="t",
|
|
description="d",
|
|
rein=_FakeRein(),
|
|
manager=manager,
|
|
catalog=_catalog_with_readiness(),
|
|
report_to_hub=False,
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert result["evidence"]["profile_ref"] == PROFILE
|
|
|
|
|
|
def test_teardown_failure_is_visible_in_evidence() -> None:
|
|
manager = MagicMock()
|
|
manager.create.return_value = _fake_status()
|
|
manager.destroy.side_effect = RuntimeError("cannot teardown")
|
|
|
|
result = run_execution(
|
|
_request(),
|
|
catalog=_catalog_with_readiness(),
|
|
rein=_FakeRein(),
|
|
manager=manager,
|
|
)
|
|
|
|
assert result.ok is False
|
|
assert result.evidence.failure_stage == "teardown"
|
|
assert result.evidence.error == "teardown failed; inspect direct caller error"
|
|
assert result.tool_error == "cannot teardown"
|