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

from __future__ import annotations

from dataclasses import replace
from datetime import UTC, datetime
from inspect import signature
from types import SimpleNamespace
from typing import Any, cast

import pytest
from agent.orchestration import ArtifactRef, AttemptSubmission

from script_build_host.agents.prompts.contracts import ROOT_VALIDATOR_PROMPT
from script_build_host.application.phase_two_candidates import (
    PhaseTwoCandidateError,
    PhaseTwoCandidateService,
    _require_paragraph_structures,
)
from script_build_host.application.phase_two_planning import (
    PhasePolicyGuard,
    PhaseTwoPlanningService,
)
from script_build_host.application.root_delivery import (
    RootDeliveryService,
    _closure_digest,
    _RootClosure,
)
from script_build_host.domain.artifacts import (
    ArtifactKind,
    ArtifactState,
    ArtifactVersion,
    DirectionArtifact,
    DirectionConstraint,
    DirectionGoal,
    DirectionPreference,
)
from script_build_host.domain.goal_coverage import (
    GoalCoverage,
    GoalPolicyError,
    build_goal_coverage,
    validate_goal_coverage,
)
from script_build_host.domain.phase_three_artifacts import (
    RootDeliveryManifestV1,
    canonicalize_legacy_projection,
)
from script_build_host.domain.phase_two_artifacts import (
    CandidateLineageV1,
    CandidatePortfolioArtifactV1,
    ParagraphArtifactV1,
    ScriptParagraphV1,
    StructuredScriptArtifactV1,
    hydrate_phase_two_artifact,
)
from script_build_host.domain.task_contracts import (
    AcceptedDecisionRef,
    ScriptCriterion,
    ScriptIntentClass,
    ScriptTaskBudget,
    ScriptTaskContractV1,
    ScriptTaskKind,
    TaskContractError,
)

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


def test_goal_coverage_supports_parent_child_and_many_to_many_sources() -> None:
    coverage = build_goal_coverage(
        direction_goal_ids=("parent", "child", "tone"),
        creative_sources=(
            ("script-build://artifact-versions/11", ("parent", "tone")),
            ("script-build://artifact-versions/12", ("parent", "child")),
        ),
    )

    assert tuple(item.goal_id for item in coverage) == ("parent", "child", "tone")
    assert coverage[0].source_artifact_refs == (
        "script-build://artifact-versions/11",
        "script-build://artifact-versions/12",
    )
    validate_goal_coverage(
        direction_goal_ids=("parent", "child", "tone"),
        coverage=coverage,
        adopted_source_refs=(
            "script-build://artifact-versions/11",
            "script-build://artifact-versions/12",
        ),
    )


@pytest.mark.parametrize(
    ("sources", "code"),
    [
        ((("script-build://artifact-versions/11", ("parent",)),), "GOAL_COVERAGE_INCOMPLETE"),
        ((("script-build://artifact-versions/11", ("unknown",)),), "GOAL_SCOPE_INVALID"),
    ],
)
def test_goal_coverage_rejects_missing_and_unknown_goals(
    sources: tuple[tuple[str, tuple[str, ...]], ...], code: str
) -> None:
    with pytest.raises(GoalPolicyError) as caught:
        build_goal_coverage(
            direction_goal_ids=("parent", "child"),
            creative_sources=sources,
        )
    assert caught.value.code == code


def test_goal_coverage_rejects_dangling_source_and_old_payloads() -> None:
    with pytest.raises(GoalPolicyError, match="non-adopted"):
        validate_goal_coverage(
            direction_goal_ids=("goal-1",),
            coverage=(
                GoalCoverage("goal-1", ("script-build://artifact-versions/12",)),
            ),
            adopted_source_refs=("script-build://artifact-versions/11",),
        )
    with pytest.raises(GoalPolicyError) as missing_coverage:
        hydrate_phase_two_artifact({"schema_version": "structured-script/v1"})
    assert missing_coverage.value.code == "GOAL_COVERAGE_INCOMPLETE"

    payload = _contract(ScriptTaskKind.STRUCTURE).to_payload()
    payload.pop("goal_ids")
    with pytest.raises(TaskContractError) as missing_goals:
        ScriptTaskContractV1.from_payload(payload)
    assert missing_goals.value.code == "GOAL_SCOPE_INVALID"


def test_paragraph_first_is_allowed_but_structure_patch_remains_pinned() -> None:
    paragraph = _contract(ScriptTaskKind.PARAGRAPH)
    PhasePolicyGuard._phase_two_contracts((paragraph,))

    patched = replace(
        paragraph,
        base_artifact_ref=ArtifactRef(
            "script-build://artifact-versions/11", "structure", "11", _DIGEST
        ),
    )
    with pytest.raises(TaskContractError, match="Paragraph patch requires"):
        PhasePolicyGuard._phase_two_contracts((patched,))


def test_duplicate_goal_ids_are_rejected_by_the_frozen_contract() -> None:
    with pytest.raises(TaskContractError) as caught:
        replace(_contract(ScriptTaskKind.STRUCTURE), goal_ids=("goal-1", "goal-1"))
    assert caught.value.code == "GOAL_SCOPE_INVALID"


def test_compose_derives_coverage_and_requires_structure_for_adopted_paragraph() -> None:
    assert set(signature(PhaseTwoCandidateService.save_structured_script_candidate).parameters) == {
        "self",
        "acceptance_notes",
        "context",
    }
    lineage = CandidateLineageV1(
        scope_ref="script-build://scopes/full/opening",
        input_snapshot_ref="script-build://inputs/11",
        input_closure_digest=_DIGEST,
        write_scope=("script-build://writes/paragraphs/full/opening",),
        goal_ids=("goal-1",),
    )
    paragraph = ParagraphArtifactV1(
        lineage=lineage,
        paragraphs=(ScriptParagraphV1(1, 1, 1, None, "opening", {}),),
    )
    with pytest.raises(PhaseTwoCandidateError, match="covering Structure"):
        _require_paragraph_structures(
            (_version(11, ArtifactKind.PARAGRAPH, paragraph, datetime.now(UTC)),)
        )


@pytest.mark.asyncio
@pytest.mark.parametrize("goal_ids", [(), ("unknown",)])
async def test_phase_two_goal_guard_rejects_invalid_scope_before_dispatch(
    goal_ids: tuple[str, ...],
) -> None:
    direction = DirectionArtifact(
        goals=(
            DirectionGoal("goal-1", "Complete", "needed", None, ("complete",)),
            DirectionGoal("goal-2", "Specific", "needed", "goal-1", ("specific",)),
        ),
        evidence_refs=("script-build://artifact-versions/90",),
    )
    service = PhaseTwoPlanningService(
        coordinator=cast(Any, SimpleNamespace()),
        bindings=cast(Any, SimpleNamespace()),
        contracts=cast(Any, SimpleNamespace()),
        artifacts=cast(Any, _DirectionArtifacts(direction)),
    )
    direction_ref = AcceptedDecisionRef(
        "direction-accept",
        ArtifactRef("script-build://artifact-versions/10", "direction", "10", _DIGEST),
        "script-build://scopes/full",
        ScriptTaskKind.DIRECTION,
    )
    contract = replace(
        _contract(ScriptTaskKind.STRUCTURE),
        goal_ids=goal_ids,
        input_decision_refs=(direction_ref,),
    )

    with pytest.raises(GoalPolicyError) as caught:
        await service._guard_phase_two_goals(
            context={"phase": 2},
            binding=SimpleNamespace(
                active_direction_artifact_version_id=10,
                script_build_id=7,
            ),
            root_trace_id="root",
            ledger=SimpleNamespace(root_task_id="root"),
            contracts=(contract,),
            parent=SimpleNamespace(task_id="root"),
        )
    assert caught.value.code == "GOAL_SCOPE_INVALID"


@pytest.mark.asyncio
async def test_root_preflight_checks_all_goal_nodes_and_frozen_direction_contract() -> None:
    now = datetime.now(UTC)
    direction = DirectionArtifact(
        goals=(
            DirectionGoal(
                "parent", "Make a useful script", "business goal", None, ("clear outcome",)
            ),
            DirectionGoal(
                "child", "Give one concrete action", "supports parent", "parent", ("action exists",)
            ),
        ),
        constraints=(DirectionConstraint("constraint-1", "No fabricated facts"),),
        preferences=(DirectionPreference("preference-1", "Prefer a brisk tone"),),
        evidence_refs=("script-build://artifact-versions/90",),
    )
    structured = StructuredScriptArtifactV1(
        direction_ref="script-build://artifact-versions/1",
        input_closure_digest=_DIGEST,
        paragraphs=(ScriptParagraphV1(1, 1, 1, None, "body", {}, description="do this"),),
        elements=(),
        paragraph_element_links=(),
        source_artifact_refs=("script-build://artifact-versions/3",),
        goal_coverage=(
            GoalCoverage("parent", ("script-build://artifact-versions/3",)),
            GoalCoverage("child", ("script-build://artifact-versions/3",)),
        ),
        evidence_refs=(),
        acceptance_notes=(),
    )
    portfolio = CandidatePortfolioArtifactV1(
        adopted_structured_script_ref="script-build://artifact-versions/3",
        candidate_structured_script_refs=("script-build://artifact-versions/3",),
        accepted_decision_ids=("compose-accept",),
        superseded_decision_ids=(),
        rejected_or_held_decision_ids=(),
        input_closure_digest=_DIGEST,
        unresolved_defects=(),
        compose_order=("compose-accept",),
    )
    closure = _RootClosure(
        _version(1, ArtifactKind.DIRECTION, direction, now),
        _version(2, ArtifactKind.CANDIDATE_PORTFOLIO, portfolio, now),
        _version(3, ArtifactKind.STRUCTURED_SCRIPT, structured, now),
    )
    projection = canonicalize_legacy_projection(
        structured, direction=direction.legacy_markdown, summary="complete"
    )
    manifest = RootDeliveryManifestV1(
        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=_closure_digest(closure),
        legacy_projection_digest=projection.canonical_sha256,
        build_summary="complete",
    )
    manifest_version = _version(4, ArtifactKind.ROOT_DELIVERY_MANIFEST, manifest, now)
    manifest_ref = ArtifactRef(
        "script-build://artifact-versions/4", "root_delivery_manifest", "4", _DIGEST
    )
    ledger = SimpleNamespace(
        attempts={
            "attempt-root": SimpleNamespace(
                attempt_id="attempt-root",
                submission=AttemptSubmission("complete", artifact_refs=[manifest_ref]),
                task_id="root",
            )
        }
    )
    service = RootDeliveryService(
        bindings=cast(Any, _Bindings()),
        task_store=cast(Any, _TaskStore(ledger)),
        artifacts=cast(Any, _Artifacts(manifest_version)),
    )

    async def closure_override(_: Any) -> _RootClosure:
        return closure

    service._closure = closure_override  # type: ignore[method-assign]
    results = await service.deterministic_precheck(
        context={"root_trace_id": "root", "attempt_id": "attempt-root"}
    )
    by_id = {item["rule_id"]: item["verdict"] for item in results}
    assert by_id["root-goal-coverage"] == "passed"
    assert by_id["root-direction-contract"] == "passed"
    assert "child never proves its parent automatically" in ROOT_VALIDATOR_PROMPT
    assert "do not reject solely" in " ".join(ROOT_VALIDATOR_PROMPT.split())


def _contract(kind: ScriptTaskKind) -> ScriptTaskContractV1:
    output = {
        ScriptTaskKind.STRUCTURE: "structure-artifact/v1",
        ScriptTaskKind.PARAGRAPH: "paragraph-artifact/v1",
    }[kind]
    return ScriptTaskContractV1(
        task_kind=kind,
        scope_ref="script-build://scopes/full",
        intent_class=ScriptIntentClass.EXPLORE,
        objective="produce one concrete increment",
        input_decision_refs=(),
        base_artifact_ref=None,
        write_scope=("script-build://writes/paragraphs/full",),
        gap_ref=None,
        output_schema=output,
        criteria=(ScriptCriterion("closed", "complete"),),
        budget=ScriptTaskBudget(),
        goal_ids=("goal-1",),
    )


def _version(
    identifier: int, kind: ArtifactKind, artifact: Any, now: datetime
) -> ArtifactVersion:
    return ArtifactVersion(
        identifier,
        7,
        f"task-{identifier}",
        f"attempt-{identifier}",
        1,
        kind,
        _DIGEST,
        ArtifactState.FROZEN,
        artifact,
        now,
        now,
    )


class _Bindings:
    async def get_by_root(self, _: str) -> Any:
        return SimpleNamespace(script_build_id=7, root_trace_id="root")


class _TaskStore:
    def __init__(self, ledger: Any) -> None:
        self.ledger = ledger

    async def load(self, _: str) -> Any:
        return self.ledger


class _Artifacts:
    def __init__(self, manifest: ArtifactVersion) -> None:
        self.manifest = manifest

    async def read_by_ref(self, *_: Any, **__: Any) -> ArtifactVersion:
        return self.manifest


class _DirectionArtifacts:
    def __init__(self, direction: DirectionArtifact) -> None:
        self.direction = direction

    async def get_by_id(self, *_: Any, **__: Any) -> Any:
        return SimpleNamespace(artifact=self.direction)