glas-harness/tests/test_gateway.py

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

import pytest

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
from glas_harness.transport import transport_from_sandbox


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"


@pytest.mark.parametrize("request_id", ["exact-request", None])
def test_owner_commands_preserve_create_identity_routes_and_timeout(request_id):
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    manager.execute.return_value = MagicMock(
        timed_out=False, output_truncated=False, exit_code=0, stdout="", stderr=""
    )
    catalog = _catalog_with_readiness()
    profile, _ = catalog.resolve(PROFILE)
    refs = ["proof-route-a", "proof-route-b"]
    catalog.profiles()[(profile.id, profile.version)] = profile.model_copy(update={
        "credential_route_refs": refs,
        "limits": profile.limits.model_copy(update={"timeout_seconds": 7}),
    })

    class OwnerRein(_FakeRein):
        def start_session(self, profile, inputs, sandbox):
            assert "_owner_execute" not in sandbox.model_dump()
            transport = transport_from_sandbox(sandbox)
            transport.run(["printf", "%s", "non-secret"], timeout=30)
            path = transport.write_task_file({"title": "private prompt"})
            transport.remove_file(path)
            return super().start_session(profile, inputs, sandbox)

    request = _request().model_copy(update={
        "request_id": request_id, "actor": "atm", "project": "exact-project"
    })
    result = run_execution(request, catalog=catalog, rein=OwnerRein(), manager=manager)
    assert result.ok
    create_consumer = manager.create.call_args.args[0].consumer
    assert create_consumer.actor.value == "atm"
    assert create_consumer.project == "exact-project"
    assert create_consumer.run_id == result.evidence.request_id
    assert create_consumer.session_id is None
    assert manager.execute.call_count == 3
    for call in manager.execute.call_args_list:
        sandbox_id, command = call.args
        assert sandbox_id == "sbx1"
        assert command.consumer == create_consumer
        assert command.credential_route_refs == refs
        assert command.timeout_seconds == 7
    writer = manager.execute.call_args_list[1].args[1]
    assert writer.stdin_text == '{"title": "private prompt"}'
    assert "private prompt" not in str(writer.command)
    manager.destroy.assert_called_once_with("sbx1")


@pytest.mark.parametrize("failure", ["refused", "timeout", "truncated", "write"])
def test_owner_session_failure_still_destroys_sandbox(failure):
    from glas_harness.reins.rein_aharness import ReinAharness

    manager = MagicMock()
    manager.create.return_value = _fake_status()
    good = MagicMock(timed_out=False, output_truncated=False, exit_code=0,
                     stdout="head-before", stderr="")
    if failure == "refused":
        manager.execute.side_effect = PermissionError("owner denied execution")
    elif failure == "write":
        manager.execute.side_effect = [good, MagicMock(
            timed_out=False, output_truncated=False, exit_code=1,
            stdout="", stderr="task write refused",
        )]
    else:
        manager.execute.return_value = MagicMock(
            timed_out=failure == "timeout", output_truncated=failure == "truncated"
        )
    with patch("glas_harness.transport.subprocess.run") as host_run:
        result = run_execution(_request(), catalog=_catalog_with_readiness(),
                               rein=ReinAharness(), manager=manager)
    assert not result.ok
    assert result.evidence.failure_stage == "session_start"
    assert result.evidence.session_cleanup == "not_attempted"
    assert result.evidence.sandbox_destroy == "succeeded"
    manager.destroy.assert_called_once_with("sbx1")
    host_run.assert_not_called()


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"


def test_manager_initialization_failure_returns_redacted_evidence():
    with patch("glas_harness.gateway.SandboxManager", side_effect=RuntimeError(
        "private manager configuration"
    )), patch("glas_harness.gateway.hub.post_progress_event") as post:
        result = run_execution(_request(report_to_hub=True),
                               catalog=_catalog_with_readiness(), rein=_FakeRein())
    assert not result.ok
    assert result.evidence.failure_stage == "sandbox_create"
    assert result.evidence.sandbox_id is None
    assert result.tool_error == "private manager configuration"
    assert "private manager configuration" not in str(post.call_args)
    post.assert_called_once()


@pytest.mark.parametrize('cleanup_fails,destroy_fails', [(True, False), (False, True), (True, True)])
def test_teardown_outcomes_survive_original_execution_failure(cleanup_fails, destroy_fails):
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    if destroy_fails:
        manager.destroy.side_effect = RuntimeError('private destroy failure')

    class FailingRein(_FakeRein):
        def dispatch_tool(self, session, tool_call):
            raise RuntimeError('original execution failure')

        def cleanup_session(self, session):
            if cleanup_fails:
                raise RuntimeError('private cleanup failure')

    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)
    assert not result.ok
    assert result.evidence.failure_stage == 'execution'
    assert result.tool_error == 'original execution failure'
    assert result.evidence.session_cleanup == ('failed' if cleanup_fails else 'succeeded')
    assert result.evidence.sandbox_destroy == ('failed' if destroy_fails else 'succeeded')
    detail = post.call_args.kwargs['detail']
    assert detail['session_cleanup'] == result.evidence.session_cleanup
    assert detail['sandbox_destroy'] == result.evidence.sandbox_destroy
    assert 'private' not in str(detail)
    manager.destroy.assert_called_once_with('sbx1')


@pytest.mark.parametrize('report,accepted,expected', [
    (False, True, 'not_requested'), (True, True, 'accepted'), (True, False, 'failed')
])
def test_reporting_outcome_does_not_change_execution(report, accepted, expected):
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    rein = _FakeRein()
    with patch('glas_harness.gateway.hub.post_progress_event', return_value=accepted) as post:
        result = run_execution(_request(report_to_hub=report), catalog=_catalog_with_readiness(),
                               rein=rein, manager=manager)
    assert result.ok
    assert result.hub_report_status == expected
    assert result.evidence.session_cleanup == 'succeeded'
    assert result.evidence.sandbox_destroy == 'succeeded'
    assert post.call_count == int(report)
    assert rein.calls.count('dispatch_tool') == 1


def test_refusal_reports_no_teardown_attempt_and_reporting_failure():
    with patch('glas_harness.gateway.hub.post_progress_event', return_value=False):
        result = run_execution(_request(report_to_hub=True))
    assert result.evidence.outcome == 'refused'
    assert result.evidence.session_cleanup == 'not_attempted'
    assert result.evidence.sandbox_destroy == 'not_attempted'
    assert result.hub_report_status == 'failed'


@pytest.mark.parametrize('http_failure', ['timeout', 'status'])
def test_http_reporting_failure_preserves_success(http_failure):
    import httpx
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    with patch('glas_harness.hub.httpx.post') as post:
        if http_failure == 'timeout':
            post.side_effect = httpx.ConnectTimeout('private transport detail')
        else:
            post.return_value = httpx.Response(503, request=httpx.Request('POST', 'http://hub/progress/'))
        result = run_execution(_request(report_to_hub=True), catalog=_catalog_with_readiness(),
                               rein=_FakeRein(), manager=manager)
    assert result.ok
    assert result.hub_report_status == 'failed'
    assert 'private transport detail' not in result.model_dump_json()
    post.assert_called_once()
    manager.destroy.assert_called_once_with('sbx1')


def test_cleanup_failure_after_success_still_destroys_sandbox():
    manager = MagicMock()
    manager.create.return_value = _fake_status()
    rein = _FakeRein()
    rein.cleanup_session = MagicMock(side_effect=RuntimeError('cleanup failed'))
    result = run_execution(_request(), catalog=_catalog_with_readiness(),
                           rein=rein, manager=manager)
    assert not result.ok
    assert result.evidence.failure_stage == 'teardown'
    assert result.evidence.session_cleanup == 'failed'
    assert result.evidence.sandbox_destroy == 'succeeded'
    manager.destroy.assert_called_once_with('sbx1')


def test_old_result_records_have_unknown_outcomes():
    from glas_harness.contract import GatewayResult
    old_record = run_execution(_request()).model_dump()
    old_record.pop('hub_report_status')
    old_record['evidence'].pop('session_cleanup')
    old_record['evidence'].pop('sandbox_destroy')
    restored = GatewayResult.model_validate(old_record)
    assert restored.hub_report_status == 'unknown'
    assert restored.evidence.session_cleanup == 'unknown'
    assert restored.evidence.sandbox_destroy == 'unknown'


@pytest.mark.parametrize('state', [state for state in SandboxState if state != SandboxState.READY])
def test_nonready_owner_handoff_never_starts_rein(state):
    manager = MagicMock()
    manager.create.return_value = _fake_status().model_copy(update={'state': state})
    rein = _FakeRein()
    result = run_execution(_request(), catalog=_catalog_with_readiness(), rein=rein, manager=manager)
    assert not result.ok
    assert result.evidence.failure_stage == 'sandbox_create'
    assert result.evidence.session_cleanup == 'not_attempted'
    assert result.evidence.sandbox_destroy == 'succeeded'
    assert rein.calls == []
    manager.execute.assert_not_called()
    manager.destroy.assert_called_once_with('sbx1')


def test_reachability_preparation_failure_is_reported_and_destroyed():
    manager = MagicMock()
    reachability = MagicMock()
    reachability.model_dump.side_effect = RuntimeError('private owner detail')
    manager.create.return_value = _fake_status().model_copy(update={'reachability': reachability})
    rein = _FakeRein()
    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=rein, manager=manager)
    assert not result.ok
    assert result.evidence.failure_stage == 'session_start'
    assert result.tool_error == 'private owner detail'
    assert result.evidence.session_cleanup == 'not_attempted'
    assert result.evidence.sandbox_destroy == 'succeeded'
    assert 'private owner detail' not in str(post.call_args)
    assert rein.calls == []
    manager.destroy.assert_called_once_with('sbx1')