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

from __future__ import annotations

from types import SimpleNamespace

import pytest
from agent.orchestration import ArtifactRef, ResourceClaim, TaskStatus

from script_build_host.application.direction_evidence_gate import DirectionEvidenceGate
from script_build_host.application.phase_two_planning import (
    _claim_serialized_waves,
)
from script_build_host.domain.errors import RequiredEvidenceUnsatisfied
from script_build_host.domain.evidence_requirements import (
    DirectionEvidenceRequirementFactory,
    EvidenceClosureEvaluator,
    EvidenceGate,
    EvidenceItem,
    EvidenceProvider,
)
from script_build_host.domain.task_contract_compiler import (
    PlanningContainer,
    PlanningDecision,
    PlanningSnapshot,
    PlanningTarget,
    TaskContractCompiler,
    TaskContractMutation,
)
from script_build_host.domain.task_contracts import (
    AcceptedDecisionRef,
    PlannerTaskIntentV2,
    ScriptCriterion,
    ScriptTaskBudget,
    ScriptTaskKind,
    TaskContractError,
)
from script_build_host.infrastructure.task_contract_store import (
    FileScriptTaskContractStore,
)


def _decision(
    identity: str, kind: ScriptTaskKind, branch: str
) -> PlanningDecision:
    return PlanningDecision(
        AcceptedDecisionRef(
            identity,
            ArtifactRef(
                uri="script-build://artifact-versions/1",
                kind="artifact",
                version="1",
                digest="sha256:" + "1" * 64,
            ),
            f"script-build://scopes/{branch}/{kind.value}",
            kind,
        ),
        f"producer-{identity}",
        f"script-build://branches/{branch}",
        ("goal-a",),
    )


def test_same_workspace_lane_is_serialized_while_distinct_lanes_share_a_wave() -> None:
    lane_a = (ResourceClaim("script-build://resources/workspace/a"),)
    lane_b = (ResourceClaim("script-build://resources/workspace/b"),)
    waves = _claim_serialized_waves(
        (
            ("paragraph-a", "worker", lane_a),
            ("element-a", "worker", lane_a),
            ("paragraph-b", "worker", lane_b),
        )
    )
    assert tuple(item[0] for item in waves[0]) == (
        "paragraph-a",
        "paragraph-b",
    )
    assert tuple(item[0] for item in waves[1]) == ("element-a",)


def test_host_owned_root_delivery_uses_the_v2_catalog_contract() -> None:
    direction = _decision(
        "direction-1", ScriptTaskKind.DIRECTION, "phase-one"
    ).ref
    portfolio = _decision(
        "portfolio-1", ScriptTaskKind.CANDIDATE_PORTFOLIO, "portfolio"
    ).ref
    contract = TaskContractCompiler().compile_host_contract(
        task_id="root-task",
        parent_task_id="root-task",
        task_kind=ScriptTaskKind.ROOT_DELIVERY,
        objective="publish",
        scope_ref=direction.scope_ref,
        branch_ref="script-build://branches/root-delivery",
        input_decision_refs=(direction, portfolio),
        criteria=(ScriptCriterion("published", "publish", True),),
        budget=ScriptTaskBudget(),
        goal_ids=("goal-a",),
    )
    assert contract.to_payload()["schema_version"] == "script-task-contract/v2"
    assert contract.execution_profile.worker_preset == "script_root_worker"
    assert contract.catalog_digest == TaskContractCompiler().capabilities.digest


@pytest.mark.asyncio
async def test_file_contract_reader_preserves_v2_frozen_execution_profile(
    tmp_path,
) -> None:
    direction = _decision(
        "direction-1", ScriptTaskKind.DIRECTION, "phase-one"
    ).ref
    portfolio = _decision(
        "portfolio-1", ScriptTaskKind.CANDIDATE_PORTFOLIO, "portfolio"
    ).ref
    contract = TaskContractCompiler().compile_host_contract(
        task_id="root-task",
        parent_task_id="root-task",
        task_kind=ScriptTaskKind.ROOT_DELIVERY,
        objective="publish",
        scope_ref=direction.scope_ref,
        branch_ref="script-build://branches/root-delivery",
        input_decision_refs=(direction, portfolio),
        criteria=(ScriptCriterion("published", "publish", True),),
        budget=ScriptTaskBudget(),
        goal_ids=("goal-a",),
    )
    store = FileScriptTaskContractStore(tmp_path)
    frozen = await store.freeze("root", contract)

    replay = await store.read("root", frozen.uri)

    assert replay.contract == contract


def test_intent_order_does_not_change_compiled_contracts() -> None:
    snapshot = PlanningSnapshot(
        root_trace_id="root",
        phase=2,
        containers=(
            PlanningContainer(
                "root-task",
                None,
                None,
                None,
            ),
        ),
        decisions=(),
        active_direction_ref=_decision(
            "direction-1", ScriptTaskKind.DIRECTION, "phase-one"
        ).ref,
        direction_goal_ids=("goal-a",),
    )
    intents = (
        PlannerTaskIntentV2(
            ScriptTaskKind.STRUCTURE, "approach a", target_goal_ids=("goal-a",)
        ),
        PlannerTaskIntentV2(
            ScriptTaskKind.STRUCTURE, "approach b", target_goal_ids=("goal-a",)
        ),
    )
    compiler = TaskContractCompiler()
    first = compiler.compile_batch(
        intents, snapshot, command_id="command", budget=ScriptTaskBudget()
    )
    second = compiler.compile_batch(
        tuple(reversed(intents)),
        snapshot,
        command_id="command",
        budget=ScriptTaskBudget(),
    )
    assert [item.contract.to_payload() for item in first] == [
        item.contract.to_payload() for item in second
    ]
    assert [item.contract.task_kind for item in first].count(
        ScriptTaskKind.CANDIDATE_PORTFOLIO
    ) == 1
    assert [item.contract.task_kind for item in first].count(
        ScriptTaskKind.COMPOSE
    ) == 2


def test_phase_two_scaffolds_then_closes_without_planner_contract_fields() -> None:
    compiler = TaskContractCompiler()
    direction = _decision("direction-1", ScriptTaskKind.DIRECTION, "phase-one").ref
    initial = PlanningSnapshot(
        "root",
        2,
        (PlanningContainer("root-task", None, None, None),),
        (),
        active_direction_ref=direction,
        direction_goal_ids=("goal-a",),
    )
    opened = compiler.compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.STRUCTURE,
                "open one creative approach",
                target_goal_ids=("goal-a",),
            ),
        ),
        initial,
        command_id="open",
        budget=ScriptTaskBudget(),
    )
    portfolio = next(
        item
        for item in opened
        if item.contract.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO
    )
    compose = next(
        item for item in opened if item.contract.task_kind is ScriptTaskKind.COMPOSE
    )
    structure = next(
        item for item in opened if item.contract.task_kind is ScriptTaskKind.STRUCTURE
    )
    assert not portfolio.contract.execution_ready
    assert not compose.contract.execution_ready
    assert structure.parent_task_id == compose.task_id

    structure_decision = PlanningDecision(
        AcceptedDecisionRef(
            "structure-1",
            ArtifactRef(
                uri="script-build://artifact-versions/10",
                kind="artifact",
                version="10",
                digest="sha256:" + "a" * 64,
            ),
            structure.contract.scope_ref,
            ScriptTaskKind.STRUCTURE,
        ),
        structure.task_id,
        structure.contract.branch_ref,
        structure.contract.goal_ids,
    )
    containers = (
        PlanningContainer("root-task", None, None, None),
        PlanningContainer(
            portfolio.task_id,
            portfolio.contract.task_kind,
            portfolio.contract.scope_ref,
            portfolio.contract.branch_ref,
            portfolio.contract.goal_ids,
            portfolio.parent_task_id,
            True,
        ),
        PlanningContainer(
            compose.task_id,
            compose.contract.task_kind,
            compose.contract.scope_ref,
            compose.contract.branch_ref,
            compose.contract.goal_ids,
            compose.parent_task_id,
            True,
        ),
    )
    paragraph = compiler.compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.PARAGRAPH,
                "write paragraph",
                decision_ids=("structure-1",),
                target_goal_ids=("goal-a",),
            ),
        ),
        PlanningSnapshot(
            "root",
            2,
            containers,
            (structure_decision,),
            (
                PlanningTarget(
                    "script-build://targets/paragraph-1",
                    "structure-1",
                    ScriptTaskKind.STRUCTURE,
                    structure.contract.branch_ref,
                    ("goal-a",),
                ),
            ),
            direction,
            ("goal-a",),
        ),
        command_id="paragraph",
        budget=ScriptTaskBudget(),
    )[0]
    paragraph_decision = PlanningDecision(
        AcceptedDecisionRef(
            "paragraph-1",
            ArtifactRef(
                uri="script-build://artifact-versions/11",
                kind="artifact",
                version="11",
                digest="sha256:" + "b" * 64,
            ),
            paragraph.contract.scope_ref,
            ScriptTaskKind.PARAGRAPH,
        ),
        paragraph.task_id,
        paragraph.contract.branch_ref,
        paragraph.contract.goal_ids,
    )
    element = compiler.compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.ELEMENT_SET,
                "add elements",
                decision_ids=("paragraph-1",),
                target_goal_ids=("goal-a",),
            ),
        ),
        PlanningSnapshot(
            "root",
            2,
            containers,
            (paragraph_decision,),
            (
                PlanningTarget(
                    "script-build://targets/paragraph-1",
                    "paragraph-1",
                    ScriptTaskKind.PARAGRAPH,
                    paragraph.contract.branch_ref,
                    ("goal-a",),
                ),
            ),
            direction,
            ("goal-a",),
        ),
        command_id="element",
        budget=ScriptTaskBudget(),
    )[0]
    element_decision = PlanningDecision(
        AcceptedDecisionRef(
            "element-1",
            ArtifactRef(
                uri="script-build://artifact-versions/12",
                kind="artifact",
                version="12",
                digest="sha256:" + "c" * 64,
            ),
            element.contract.scope_ref,
            ScriptTaskKind.ELEMENT_SET,
        ),
        element.task_id,
        element.contract.branch_ref,
        element.contract.goal_ids,
    )
    closed_compose = compiler.compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.COMPOSE,
                "compose accepted branch",
                decision_ids=("structure-1", "paragraph-1", "element-1"),
            ),
        ),
        PlanningSnapshot(
            "root",
            2,
            containers,
            (structure_decision, paragraph_decision, element_decision),
            active_direction_ref=direction,
            direction_goal_ids=("goal-a",),
        ),
        command_id="compose",
        budget=ScriptTaskBudget(),
    )[0]
    assert closed_compose.task_id == compose.task_id
    assert closed_compose.mutation is TaskContractMutation.REVISE
    assert closed_compose.contract.execution_ready

    compose_decision = PlanningDecision(
        AcceptedDecisionRef(
            "compose-1",
            ArtifactRef(
                uri="script-build://artifact-versions/13",
                kind="artifact",
                version="13",
                digest="sha256:" + "d" * 64,
            ),
            closed_compose.contract.scope_ref,
            ScriptTaskKind.COMPOSE,
        ),
        compose.task_id,
        compose.contract.branch_ref,
        closed_compose.contract.goal_ids,
    )
    closed_portfolio = compiler.compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.CANDIDATE_PORTFOLIO,
                "adopt accepted script",
                decision_ids=("compose-1",),
            ),
        ),
        PlanningSnapshot(
            "root",
            2,
            containers,
            (compose_decision,),
            active_direction_ref=direction,
            direction_goal_ids=("goal-a",),
        ),
        command_id="portfolio",
        budget=ScriptTaskBudget(),
    )[0]
    assert closed_portfolio.task_id == portfolio.task_id
    assert closed_portfolio.mutation is TaskContractMutation.REVISE
    assert closed_portfolio.contract.execution_ready


def test_compose_scaffold_cannot_close_before_child_frontier_is_terminal() -> None:
    decisions = tuple(
        _decision(f"{kind.value}-1", kind, "branch-a")
        for kind in (
            ScriptTaskKind.STRUCTURE,
            ScriptTaskKind.PARAGRAPH,
            ScriptTaskKind.ELEMENT_SET,
        )
    )
    snapshot = PlanningSnapshot(
        "root",
        2,
        (
            PlanningContainer("root-task", None, None, None),
            PlanningContainer(
                "compose-parent",
                ScriptTaskKind.COMPOSE,
                "script-build://scopes/compose",
                "script-build://branches/branch-a",
                ("goal-a",),
                "portfolio-parent",
                False,
            ),
        ),
        decisions,
        active_direction_ref=_decision(
            "direction-1", ScriptTaskKind.DIRECTION, "phase-one"
        ).ref,
        direction_goal_ids=("goal-a",),
    )
    with pytest.raises(TaskContractError, match="nonterminal child work"):
        TaskContractCompiler().compile_batch(
            (
                PlannerTaskIntentV2(
                    ScriptTaskKind.COMPOSE,
                    "close too early",
                    decision_ids=tuple(
                        item.ref.decision_id for item in decisions
                    ),
                ),
            ),
            snapshot,
            command_id="premature-compose",
            budget=ScriptTaskBudget(),
        )


def test_paragraph_target_is_host_semantic_target_not_objective_text() -> None:
    structure = _decision("structure-1", ScriptTaskKind.STRUCTURE, "branch-a")
    snapshot = PlanningSnapshot(
        root_trace_id="root",
        phase=2,
        containers=(
            PlanningContainer(
                "compose-parent",
                ScriptTaskKind.COMPOSE,
                None,
                "script-build://branches/branch-a",
                ("goal-a",),
            ),
        ),
        decisions=(structure,),
        targets=(
            PlanningTarget(
                "script-build://targets/paragraph-a",
                structure.ref.decision_id,
                ScriptTaskKind.STRUCTURE,
                "script-build://branches/branch-a",
                ("goal-a",),
            ),
        ),
        active_direction_ref=_decision(
            "direction-1", ScriptTaskKind.DIRECTION, "phase-one"
        ).ref,
        direction_goal_ids=("goal-a",),
    )
    compiled = TaskContractCompiler().compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.PARAGRAPH,
                "wording can change freely",
                decision_ids=("structure-1",),
                target_goal_ids=("goal-a",),
            ),
        ),
        snapshot,
        command_id="command",
        budget=ScriptTaskBudget(),
    )[0]
    assert compiled.contract.workspace_seed is not None
    assert (
        compiled.contract.workspace_seed.semantic_target_ref
        == "script-build://targets/paragraph-a"
    )


def test_ambiguous_semantic_target_fails_before_task_creation() -> None:
    structure = _decision("structure-1", ScriptTaskKind.STRUCTURE, "branch-a")
    targets = tuple(
        PlanningTarget(
            f"script-build://targets/{suffix}",
            structure.ref.decision_id,
            ScriptTaskKind.STRUCTURE,
            "script-build://branches/branch-a",
            ("goal-a",),
        )
        for suffix in ("a", "b")
    )
    snapshot = PlanningSnapshot(
        "root",
        2,
        (
            PlanningContainer(
                "compose-parent",
                ScriptTaskKind.COMPOSE,
                None,
                "script-build://branches/branch-a",
                ("goal-a",),
            ),
        ),
        (structure,),
        targets,
        _decision("direction-1", ScriptTaskKind.DIRECTION, "phase-one").ref,
        ("goal-a",),
    )
    with pytest.raises(TaskContractError, match="semantic targets"):
        TaskContractCompiler().compile_batch(
            (
                PlannerTaskIntentV2(
                    ScriptTaskKind.PARAGRAPH,
                    "paragraph",
                    decision_ids=("structure-1",),
                    target_goal_ids=("goal-a",),
                ),
            ),
            snapshot,
            command_id="command",
            budget=ScriptTaskBudget(),
        )


def test_pattern_evidence_cannot_satisfy_missing_persona() -> None:
    manifest = DirectionEvidenceRequirementFactory().compile(
        topic_input_ref="input:topic",
        requested_account_name="account",
        persona_points=(),
        persona_input_ref=None,
        section_patterns=(),
        section_pattern_input_ref=None,
    )
    closure = EvidenceClosureEvaluator().evaluate(
        manifest,
        (
            EvidenceItem(
                "direction.section_pattern",
                "pattern:1",
                EvidenceProvider.PATTERN,
            ),
        ),
        gate=EvidenceGate.DIRECTION_ACCEPT,
    )
    assert "direction.persona" in closure.unsatisfied_hard_requirement_ids


@pytest.mark.asyncio
async def test_direction_gate_rejects_accept_from_stale_child_spec() -> None:
    manifest = DirectionEvidenceRequirementFactory().compile(
        topic_input_ref="input:topic",
        requested_account_name="account",
        persona_points=(),
        persona_input_ref=None,
        section_patterns=({"pattern": "hook"},),
        section_pattern_input_ref="input:patterns",
    )
    direction = _decision(
        "direction-1", ScriptTaskKind.DIRECTION, "phase-one"
    ).ref
    portfolio = _decision(
        "portfolio-1", ScriptTaskKind.CANDIDATE_PORTFOLIO, "portfolio"
    ).ref
    contract = TaskContractCompiler().compile_host_contract(
        task_id="root-task",
        parent_task_id="root-task",
        task_kind=ScriptTaskKind.ROOT_DELIVERY,
        objective="publish",
        scope_ref=direction.scope_ref,
        branch_ref="script-build://branches/root-delivery",
        input_decision_refs=(direction, portfolio),
        criteria=(ScriptCriterion("published", "publish", True),),
        budget=ScriptTaskBudget(),
        goal_ids=("goal-a",),
        evidence_requirements=manifest,
    )
    child = SimpleNamespace(
        task_id="evidence-child",
        status=TaskStatus.COMPLETED,
        current_spec_version=2,
        current_spec=SimpleNamespace(
            context_refs=("script-build://task-kinds/evidence-retrieval",)
        ),
        decision_ids=["accepted-old-spec"],
    )
    ledger = SimpleNamespace(
        tasks={child.task_id: child},
        decisions={
            "accepted-old-spec": SimpleNamespace(
                action=SimpleNamespace(value="accept"),
                attempt_id="attempt-v1",
            )
        },
        attempts={
            "attempt-v1": SimpleNamespace(
                attempt_id="attempt-v1",
                spec_version=1,
                submission=SimpleNamespace(evidence_refs=("evidence://persona",)),
            )
        },
    )

    class UnreadArtifacts:
        async def read_by_ref(self, *_: object, **__: object) -> object:
            raise AssertionError("stale accepted evidence must not be read")

    with pytest.raises(RequiredEvidenceUnsatisfied):
        await DirectionEvidenceGate(UnreadArtifacts()).require(
            script_build_id=1,
            ledger=ledger,
            direction_task=SimpleNamespace(child_task_ids=(child.task_id,)),
            contract=contract,
            gate=EvidenceGate.DIRECTION_ACCEPT,
        )