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

from __future__ import annotations

from dataclasses import replace
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from agent.orchestration import TaskStatus

from script_build_host.application.phase_three import (
    PHASE_THREE_EXECUTION_STARTED,
    PHASE_THREE_POLICY,
    PhaseThreeContinuationService,
)
from script_build_host.domain.errors import (
    LegacyCanonicalIdentityConflict,
    MissionRecoveryRequired,
    PhaseThreeBoundaryNotReady,
    ProtocolViolation,
)
from script_build_host.domain.phase_three_artifacts import (
    RootDeliveryManifestV1,
    canonicalize_legacy_projection,
)
from script_build_host.domain.phase_two_artifacts import (
    ScriptElementV1,
    ScriptParagraphElementLinkV1,
    ScriptParagraphV1,
    StructuredScriptArtifactV1,
)
from script_build_host.domain.records import BuildStatus

_DIGEST = "sha256:" + "a" * 64


def _script(
    *,
    root_id: int = 10,
    child_id: int = 20,
    element_id: int = 30,
    child_description: str = "child body",
) -> StructuredScriptArtifactV1:
    return StructuredScriptArtifactV1(
        direction_ref="script-build://artifact-versions/1",
        input_closure_digest=_DIGEST,
        paragraphs=(
            ScriptParagraphV1(
                paragraph_id=root_id,
                paragraph_index=1,
                level=1,
                parent_id=None,
                name="root",
                content_range={"start": 0, "end": 10},
                description="root body",
            ),
            ScriptParagraphV1(
                paragraph_id=child_id,
                paragraph_index=2,
                level=2,
                parent_id=root_id,
                name="child",
                content_range={"start": 10, "end": 20},
                description=child_description,
            ),
        ),
        elements=(
            ScriptElementV1(
                element_id=element_id,
                name="hook",
                dimension_primary="形式",
                dimension_secondary="opening",
                topic_support={"score": 1},
            ),
        ),
        paragraph_element_links=(
            ScriptParagraphElementLinkV1(
                paragraph_id=child_id,
                element_id=element_id,
            ),
        ),
        source_artifact_refs=(),
        evidence_refs=(),
        acceptance_notes=(),
    )


def test_legacy_canonical_digest_ignores_every_physical_identity() -> None:
    first = canonicalize_legacy_projection(_script(), direction="# D", summary="summary")
    second = canonicalize_legacy_projection(
        _script(root_id=101, child_id=202, element_id=303),
        direction="# D",
        summary="summary",
    )

    assert first.content_payload() == second.content_payload()
    assert first.canonical_sha256 == second.canonical_sha256


def test_legacy_canonical_digest_changes_with_business_content() -> None:
    first = canonicalize_legacy_projection(_script(), direction="# D", summary="summary")
    changed_body = canonicalize_legacy_projection(
        _script(child_description="changed"), direction="# D", summary="summary"
    )
    changed_direction = canonicalize_legacy_projection(
        _script(), direction="# other", summary="summary"
    )

    assert first.canonical_sha256 != changed_body.canonical_sha256
    assert first.canonical_sha256 != changed_direction.canonical_sha256


def test_legacy_canonical_rejects_parent_cycles_and_duplicate_natural_elements() -> None:
    cycle = replace(
        _script(),
        paragraphs=(
            ScriptParagraphV1(10, 1, 2, 20, "a", {}),
            ScriptParagraphV1(20, 2, 2, 10, "b", {}),
        ),
    )
    with pytest.raises(LegacyCanonicalIdentityConflict, match="cycle"):
        canonicalize_legacy_projection(cycle)

    duplicate_elements = replace(
        _script(),
        elements=(
            ScriptElementV1(30, "hook", "形式", "opening"),
            ScriptElementV1(31, "hook", "形式", "opening"),
        ),
        paragraph_element_links=(),
    )
    with pytest.raises(LegacyCanonicalIdentityConflict, match="duplicated"):
        canonicalize_legacy_projection(duplicate_elements)


def test_root_delivery_manifest_is_strict_bounded_and_publishable() -> None:
    manifest = RootDeliveryManifestV1.from_payload(
        {
            "schema_version": "root-delivery-manifest/v1",
            "direction_ref": "script-build://artifact-versions/1",
            "candidate_portfolio_ref": "script-build://artifact-versions/2",
            "structured_script_ref": "script-build://artifact-versions/3",
            "input_closure_digest": _DIGEST,
            "legacy_projection_digest": _DIGEST,
            "build_summary": "ok",
            "blocking_defects": [],
        }
    ).with_digest()
    manifest.require_publishable()
    assert manifest.canonical_sha256.startswith("sha256:")

    with pytest.raises(ProtocolViolation, match="unknown fields"):
        RootDeliveryManifestV1.from_payload({**manifest.content_payload(), "unknown": True})
    with pytest.raises(ProtocolViolation, match="blocking defects"):
        replace(
            manifest, blocking_defects=("not ready",), canonical_sha256=""
        ).require_publishable()
    with pytest.raises(ProtocolViolation, match="2,000"):
        replace(manifest, build_summary="x" * 2_001, canonical_sha256="")


@pytest.mark.asyncio
async def test_phase_three_model_execution_marker_is_single_use() -> None:
    class TraceStore:
        def __init__(self) -> None:
            self.events: list[dict[str, str]] = []

        async def get_events(self, _trace_id: str, _since: int):
            return list(self.events)

        async def append_event(self, _trace_id: str, event: str, payload: dict[str, str]):
            self.events.append({"event": event, **payload})
            return len(self.events)

    service = object.__new__(PhaseThreeContinuationService)
    trace_store = TraceStore()
    service.runner = SimpleNamespace(trace_store=trace_store)

    await service._claim_policy_execution("root", "snapshot-3")
    assert trace_store.events == [
        {
            "event": PHASE_THREE_EXECUTION_STARTED,
            "policy_version": PHASE_THREE_POLICY,
            "input_snapshot_id": "snapshot-3",
        }
    ]
    with pytest.raises(MissionRecoveryRequired):
        await service._claim_policy_execution("root", "snapshot-3")


@pytest.mark.asyncio
async def test_phase_three_reentry_is_allowed_only_before_model_execution() -> None:
    class TraceStore:
        def __init__(self) -> None:
            self.events = [
                {
                    "event": "planner_policy_migrated",
                    "policy_version": PHASE_THREE_POLICY,
                    "input_snapshot_id": "snapshot-3",
                }
            ]

        async def get_events(self, _trace_id: str, _since: int):
            return list(self.events)

        async def append_event(self, _trace_id: str, event: str, payload: dict[str, str]):
            self.events.append({"event": event, **payload})
            return len(self.events)

    service = object.__new__(PhaseThreeContinuationService)
    trace_store = TraceStore()
    service.runner = SimpleNamespace(trace_store=trace_store)

    assert await service._phase_three_reentry_ready("root") is True
    await service._claim_policy_execution("root", "snapshot-3")
    with pytest.raises(MissionRecoveryRequired):
        await service._phase_three_reentry_ready("root")


@pytest.mark.asyncio
async def test_completed_root_does_not_restore_a_failed_build() -> None:
    service = object.__new__(PhaseThreeContinuationService)
    service._verify_owner = AsyncMock()
    service._set_root_accepted_checkpoint = AsyncMock()
    service.bindings = SimpleNamespace(
        get_by_build=AsyncMock(
            return_value=SimpleNamespace(root_trace_id="root", input_snapshot_id=3)
        )
    )
    service.legacy_state = SimpleNamespace(get_status=AsyncMock(return_value=BuildStatus.FAILED))
    service.input_snapshots = SimpleNamespace(get=AsyncMock(return_value=SimpleNamespace()))
    root = SimpleNamespace(status=TaskStatus.COMPLETED)
    service.coordinator = SimpleNamespace(
        task_store=SimpleNamespace(
            load=AsyncMock(
                return_value=SimpleNamespace(root_task_id="root-task", tasks={"root-task": root})
            )
        )
    )
    owner_token = SimpleNamespace(root_trace_id="root")

    with pytest.raises(PhaseThreeBoundaryNotReady, match="failed"):
        await service.run(7, owner_token=owner_token)
    service._set_root_accepted_checkpoint.assert_not_awaited()