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.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,
    ElementSetArtifactV1,
    ParagraphArtifactV1,
    ScriptElementV1,
    ScriptParagraphElementLinkV1,
    ScriptParagraphV1,
    StructureArtifactV1,
    StructuredScriptArtifactV1,
    hydrate_phase_two_artifact,
)
from script_build_host.domain.task_contract_compiler import (
    PlanningContainer,
    PlanningSnapshot,
    TaskContractCompiler,
)
from script_build_host.domain.task_contracts import (
    AcceptedDecisionRef,
    PlannerTaskIntentV2,
    ScriptCriterion,
    ScriptIntentClass,
    ScriptTaskBudget,
    ScriptTaskContractV1,
    ScriptTaskKind,
    TaskContractError,
)

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


def _complete_paragraph() -> ScriptParagraphV1:
    atom = ({"原子点": "行动", "维度": "主题", "维度类型": "主维度"},)
    return ScriptParagraphV1(
        1,
        1,
        1,
        None,
        "body",
        {"scope": "body"},
        theme="主动行动",
        form="直接陈述",
        function="给出方法",
        feeling="坚定",
        theme_elements=atom,
        form_elements=atom,
        function_elements=atom,
        feeling_elements=atom,
        description="给出一个当下可执行的动作。",
        full_description="今天选择一件最小但具体的事。在十分钟内开始执行。",
    )


def _realized_sources(goal_ids: tuple[str, ...]) -> dict[int, ArtifactVersion]:
    now = datetime.now(UTC)
    paragraph = _complete_paragraph()
    element = ScriptElementV1(1, "立即行动", "实质", "行动建议")
    link = ScriptParagraphElementLinkV1(1, 1)
    lineage = CandidateLineageV1(
        scope_ref="script-build://scopes/full",
        input_snapshot_ref="script-build://inputs/11",
        input_closure_digest=_DIGEST,
        write_scope=("script-build://writes/paragraphs",),
        goal_ids=goal_ids,
    )
    structure = StructureArtifactV1(lineage, (paragraph,))
    paragraph_artifact = ParagraphArtifactV1(
        lineage,
        (paragraph,),
        (element,),
        (link,),
        {"created": {"paragraph_ids": [1]}},
    )
    element_artifact = ElementSetArtifactV1(
        replace(lineage, write_scope=("script-build://writes/elements",)),
        (element,),
        (link,),
        (paragraph,),
        {"created": {"element_ids": [1]}},
    )
    return {
        5: _version(5, ArtifactKind.STRUCTURE, structure, now),
        6: _version(6, ArtifactKind.PARAGRAPH, paragraph_artifact, now),
        7: _version(7, ArtifactKind.ELEMENT_SET, element_artifact, now),
    }


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_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)),)
        )


def test_contract_compiler_owns_goal_scope_before_task_creation() -> None:
    direction_ref = AcceptedDecisionRef(
        "direction-accept",
        ArtifactRef("script-build://artifact-versions/10", "direction", "10", _DIGEST),
        "script-build://scopes/full",
        ScriptTaskKind.DIRECTION,
    )
    snapshot = PlanningSnapshot(
        "root",
        2,
        (PlanningContainer("root-task", None, None, None),),
        (),
        active_direction_ref=direction_ref,
        direction_goal_ids=("goal-1", "goal-2"),
    )
    compiler = TaskContractCompiler()
    compiled = compiler.compile_batch(
        (
            PlannerTaskIntentV2(
                ScriptTaskKind.STRUCTURE,
                "cover the active direction",
            ),
        ),
        snapshot,
        command_id="valid",
        budget=ScriptTaskBudget(),
    )
    structure = next(
        item for item in compiled if item.contract.task_kind is ScriptTaskKind.STRUCTURE
    )
    assert structure.contract.goal_ids == ("goal-1", "goal-2")

    with pytest.raises(TaskContractError) as caught:
        compiler.compile_batch(
            (
                PlannerTaskIntentV2(
                    ScriptTaskKind.STRUCTURE,
                    "invalid goal",
                    target_goal_ids=("unknown",),
                ),
            ),
            snapshot,
            command_id="invalid",
            budget=ScriptTaskBudget(),
        )
    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=(_complete_paragraph(),),
        elements=(ScriptElementV1(1, "立即行动", "实质", "行动建议"),),
        paragraph_element_links=(ScriptParagraphElementLinkV1(1, 1),),
        source_artifact_refs=(
            "script-build://artifact-versions/5",
            "script-build://artifact-versions/6",
            "script-build://artifact-versions/7",
        ),
        goal_coverage=(
            GoalCoverage(
                "parent",
                (
                    "script-build://artifact-versions/6",
                    "script-build://artifact-versions/7",
                ),
            ),
            GoalCoverage(
                "child",
                (
                    "script-build://artifact-versions/6",
                    "script-build://artifact-versions/7",
                ),
            ),
        ),
        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,
                _realized_sources(("parent", "child")),
            ),
        ),
    )

    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"
    root_validator_prompt = " ".join(ROOT_VALIDATOR_PROMPT.split())
    assert "子 Goal 不能 自动证明父 Goal" in root_validator_prompt
    assert "不因可选偏好不足单独拒绝" in root_validator_prompt


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,
        versions: dict[int, ArtifactVersion] | None = None,
    ) -> None:
        self.manifest = manifest
        self.versions = versions or {}

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

    async def get_by_id(self, identifier: int, **__: Any) -> ArtifactVersion:
        return self.versions[identifier]