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

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest

from script_build_host.application.phase_two_planning import (
    PHASE_ONE_CAPABILITY_BOUNDARY,
    PhasePolicyGuard,
    _execution_seconds,
)
from script_build_host.domain.task_capabilities import (
    LegacyTaskCapabilityCatalog,
    TaskCapabilityCatalog,
)
from script_build_host.domain.task_contracts import ScriptTaskKind, TaskContractError


def test_execution_budget_counts_runtime_intervals_not_planner_idle_time() -> None:
    now = datetime.now(UTC)
    stages = (
        SimpleNamespace(
            started_at=(now - timedelta(hours=2)).isoformat(),
            completed_at=(now - timedelta(hours=2) + timedelta(seconds=12)).isoformat(),
        ),
        SimpleNamespace(
            started_at=(now - timedelta(minutes=1)).isoformat(),
            completed_at=(now - timedelta(minutes=1) + timedelta(seconds=8)).isoformat(),
        ),
    )

    assert _execution_seconds(stages) == pytest.approx(20.0)


def test_phase_policy_dispatches_only_v2_logical_evidence() -> None:
    guard = PhasePolicyGuard()
    guard.dispatch(context={"phase": 1}, kind=ScriptTaskKind.DIRECTION)
    guard.dispatch(context={"phase": 1}, kind=ScriptTaskKind.EVIDENCE_RETRIEVAL)

    with pytest.raises(TaskContractError, match="PHASE_POLICY_VIOLATION"):
        guard.dispatch(
            context={"phase": 1},
            kind=ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
        )


def test_phase_boundary_reason_is_host_owned() -> None:
    guard = PhasePolicyGuard()
    guard.boundary(
        context={"phase": 1},
        reason=PHASE_ONE_CAPABILITY_BOUNDARY,
    )
    with pytest.raises(TaskContractError, match="PHASE_POLICY_VIOLATION"):
        guard.boundary(context={"phase": 1}, reason="model-invented-boundary")


def test_legacy_catalog_is_readable_but_physical_retrieval_is_not_v2_executable() -> None:
    assert (
        LegacyTaskCapabilityCatalog()
        .get(ScriptTaskKind.KNOWLEDGE_RETRIEVAL)
        .worker_preset
        == "script_knowledge_retrieval_worker"
    )
    with pytest.raises(TaskContractError, match="TASK_CAPABILITY_MISMATCH"):
        TaskCapabilityCatalog().get(ScriptTaskKind.KNOWLEDGE_RETRIEVAL)