tegwick
c6332916c2
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
129 lines
4 KiB
Python
129 lines
4 KiB
Python
"""Tests for the authoritative activity-core -> Glas consumer boundary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rein_aharness.glas_execution import GLAS_ACTOR, GlasExecutionError, execute_profiled_run
|
|
from rein_aharness.ops_run_client import OpsRun, OpsRunConfig
|
|
|
|
|
|
def _repo(tmp_path: Path) -> Path:
|
|
repo = tmp_path / "target"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
|
|
return repo
|
|
|
|
|
|
def _run(repo: Path) -> OpsRun:
|
|
return OpsRun(
|
|
id="run-1",
|
|
activity_definition_id="def-1",
|
|
idempotency_key="key-1",
|
|
target_repo="target",
|
|
title="Governed task",
|
|
description="Do the bounded work",
|
|
state="claimed",
|
|
harness_profile_ref="harness.agent-dev-local@1.0.0",
|
|
approach_hint="agent-session",
|
|
execution_refs={
|
|
"correlation_id": "corr-1",
|
|
"assignment_ref": "assignment:1",
|
|
"role_ref": "role:1",
|
|
"duty_ref": "duty:1",
|
|
"goal_refs": ["goal:1"],
|
|
"resource_envelope_refs": ["resource:1"],
|
|
"unknown": "must-not-cross",
|
|
},
|
|
)
|
|
|
|
|
|
def test_profiled_run_builds_complete_execution_request(tmp_path: Path) -> None:
|
|
repo = _repo(tmp_path)
|
|
captured = {}
|
|
|
|
def request_factory(**kwargs):
|
|
captured.update(kwargs)
|
|
return kwargs
|
|
|
|
expected = {
|
|
"ok": True,
|
|
"evidence": {"outcome": "succeeded", "profile_ref": "profile"},
|
|
"tool_output": "direct caller output",
|
|
"tool_error": None,
|
|
}
|
|
result = execute_profiled_run(
|
|
_run(repo),
|
|
OpsRunConfig(worker_id="rein-aharness@test", repo_roots=(str(tmp_path),)),
|
|
report_to_hub=False,
|
|
request_factory=request_factory,
|
|
gateway=lambda request: expected,
|
|
)
|
|
|
|
assert result is expected
|
|
assert captured == {
|
|
"harness_profile_ref": "harness.agent-dev-local@1.0.0",
|
|
"repo": str(repo.resolve()),
|
|
"title": "Governed task",
|
|
"description": "Do the bounded work",
|
|
"actor": GLAS_ACTOR,
|
|
"project": "rein-aharness",
|
|
"request_id": "run-1",
|
|
"correlation_id": "corr-1",
|
|
"assignment_ref": "assignment:1",
|
|
"role_ref": "role:1",
|
|
"duty_ref": "duty:1",
|
|
"goal_refs": ["goal:1"],
|
|
"resource_envelope_refs": ["resource:1"],
|
|
"report_to_hub": False,
|
|
}
|
|
|
|
|
|
def test_profiled_run_rejects_invalid_gateway_result(tmp_path: Path) -> None:
|
|
repo = _repo(tmp_path)
|
|
with pytest.raises(GlasExecutionError, match="invalid GatewayResult"):
|
|
execute_profiled_run(
|
|
_run(repo),
|
|
OpsRunConfig(repo_roots=(str(tmp_path),)),
|
|
request_factory=lambda **kwargs: kwargs,
|
|
gateway=lambda request: {"evidence": {}},
|
|
)
|
|
|
|
|
|
def test_profiled_actor_validates_against_real_glas_and_sandboxer(tmp_path: Path) -> None:
|
|
contract = pytest.importorskip("glas_harness.contract")
|
|
sandbox_models = pytest.importorskip("sandboxer.models")
|
|
repo = _repo(tmp_path)
|
|
captured = {}
|
|
|
|
def validating_gateway(request):
|
|
captured["request"] = request
|
|
captured["sandbox_request"] = sandbox_models.SandboxCreateRequest(
|
|
profile="profile.bwrap-local",
|
|
inputs={"repo": request.repo},
|
|
consumer=sandbox_models.Consumer(
|
|
actor=request.actor,
|
|
project=request.project,
|
|
run_id=request.request_id,
|
|
),
|
|
)
|
|
return {
|
|
"ok": True,
|
|
"evidence": {"outcome": "succeeded"},
|
|
"tool_output": "",
|
|
"tool_error": None,
|
|
}
|
|
|
|
execute_profiled_run(
|
|
_run(repo),
|
|
OpsRunConfig(worker_id="rein-aharness@railiance01", repo_roots=(str(tmp_path),)),
|
|
request_factory=contract.ExecutionRequest,
|
|
gateway=validating_gateway,
|
|
)
|
|
|
|
assert captured["request"].actor == "agt"
|
|
assert captured["sandbox_request"].consumer.actor == sandbox_models.ActorType.AGT
|
|
assert captured["sandbox_request"].consumer.run_id == "run-1"
|