script-build-and-framework-tmp/script_build_host/tests/test_internal_e2e.py

import json
from pathlib import Path

import httpx
import pytest

from script_build_host.domain.records import BuildStatus
from script_build_host.internal_e2e import _context_broker_diagnostics, _wait_for_root


def test_recovery_checkpoint_fixtures_cover_all_segment_boundaries() -> None:
    path = Path(__file__).parent / "fixtures" / "recovery_checkpoints.json"
    payload = json.loads(path.read_text(encoding="utf-8"))
    assert set(payload["checkpoints"]) == {"direction", "structure", "portfolio", "root"}
    assert all(
        item["input_snapshot_digest"].startswith("sha256:")
        for item in payload["checkpoints"].values()
    )


def test_context_broker_diagnostics_count_repeated_pages_not_shared_revisions() -> None:
    diagnostics = _context_broker_diagnostics(
        [
            {
                "revision": "same-artifact",
                "estimated_tokens": 100,
                "detail_reads": 1,
                "detail_page_key": "page-1",
            },
            {
                "revision": "same-artifact",
                "estimated_tokens": 300,
                "detail_reads": 1,
                "detail_page_key": "page-2",
            },
            {
                "revision": "same-artifact",
                "estimated_tokens": 100,
                "detail_reads": 1,
                "detail_page_key": "page-1",
                "semantic_calls": 1,
                "fallback": True,
            },
        ]
    )

    assert diagnostics["requests"] == 3
    assert diagnostics["bundle_tokens"] == {"p50": 100, "p95": 300, "max": 300}
    assert diagnostics["detail_reads"] == 3
    assert diagnostics["duplicate_detail_reads"] == 1
    assert diagnostics["semantic_calls"] == 1
    assert diagnostics["semantic_fallbacks"] == 1


@pytest.mark.asyncio
async def test_mission_poll_allows_only_bounded_startup_failures() -> None:
    calls = 0

    def handler(_request: httpx.Request) -> httpx.Response:
        nonlocal calls
        calls += 1
        if calls <= 5:
            return httpx.Response(500, json={"error_code": "LEDGER_NOT_READY"})
        return httpx.Response(
            200,
            json={
                "root_task_id": "root",
                "tasks": [
                    {
                        "task_id": "root",
                        "status": "blocked",
                        "blocked_reason": "BOUNDARY",
                    }
                ],
            },
        )

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler), base_url="http://internal"
    ) as client:
        result = await _wait_for_root(
            client,
            7,
            expected_status="blocked",
            expected_reason="BOUNDARY",
            timeout_seconds=1,
            poll_seconds=0,
        )

    assert result["task_id"] == "root"
    assert calls == 6


@pytest.mark.asyncio
async def test_mission_poll_surfaces_persistent_server_failure() -> None:
    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, json={"error_code": "InternalServerError"})

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler), base_url="http://internal"
    ) as client:
        with pytest.raises(RuntimeError, match="mission poll failed with HTTP 500"):
            await _wait_for_root(
                client,
                7,
                expected_status="blocked",
                expected_reason="BOUNDARY",
                timeout_seconds=1,
                poll_seconds=0,
            )


@pytest.mark.asyncio
async def test_mission_poll_stops_when_build_is_terminal() -> None:
    class FailedState:
        async def get_status(self, _script_build_id):
            return BuildStatus.FAILED

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "root_task_id": "root",
                "tasks": [{"task_id": "root", "status": "needs_replan"}],
            },
        )

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler), base_url="http://internal"
    ) as client:
        with pytest.raises(RuntimeError, match="terminal status=failed"):
            await _wait_for_root(
                client,
                7,
                legacy_state=FailedState(),
                expected_status="blocked",
                expected_reason="BOUNDARY",
                timeout_seconds=1,
                poll_seconds=0,
            )