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

from __future__ import annotations

import hashlib
import json
from dataclasses import replace
from datetime import UTC, datetime

import pytest
from agent.orchestration import (
    AcceptanceCriterion,
    ArtifactRef,
    ArtifactSnapshot,
    AttemptStatus,
    AttemptSubmission,
    DecisionAction,
    PlannerDecision,
    TaskAttempt,
    TaskLedger,
    TaskRecord,
    TaskSpec,
    TaskStatus,
    ValidationReport,
    ValidationRunStatus,
    ValidationVerdict,
)

from script_build_host.application.phase_two_boundary import ScriptPhaseTwoBoundaryVerifier
from script_build_host.application.phase_two_inputs import AcceptedInputResolver
from script_build_host.domain.artifacts import (
    ArtifactKind,
    ArtifactState,
    ArtifactVersion,
    DirectionArtifact,
    DirectionGoal,
)
from script_build_host.domain.errors import PhaseTwoBoundaryNotReady
from script_build_host.domain.phase_two_artifacts import (
    CandidateLineageV1,
    CandidatePortfolioArtifactV1,
    ElementSetArtifactV1,
    GoalCoverage,
    ParagraphArtifactV1,
    ScriptElementV1,
    ScriptParagraphElementLinkV1,
    ScriptParagraphV1,
    StructureArtifactV1,
    StructuredScriptArtifactV1,
)
from script_build_host.domain.records import MissionBinding
from script_build_host.domain.task_contracts import (
    AcceptedDecisionRef,
    ScriptCriterion,
    ScriptIntentClass,
    ScriptTaskBudget,
    ScriptTaskContractV1,
    ScriptTaskKind,
)

SNAPSHOT_ID = "11"
INPUT_DIGEST = "sha256:" + "c" * 64
SCRIPT_DIGEST = "sha256:" + "a" * 64
PORTFOLIO_DIGEST = "sha256:" + "b" * 64
SCOPE = "script-build://scopes/full"
WRITE = "script-build://writes/full"


def _complete_paragraph() -> ScriptParagraphV1:
    atom = ({"原子点": "反转", "维度": "开场", "维度类型": "主维度"},)
    return ScriptParagraphV1(
        1,
        1,
        1,
        None,
        "opening",
        {"scope": "opening"},
        theme="具体反转",
        form="对比",
        function="吸引注意",
        feeling="好奇",
        theme_elements=atom,
        form_elements=atom,
        function_elements=atom,
        feeling_elements=atom,
        description="用可观察的细节推翻初始判断。",
        full_description="大家以为问题来自天赋。但一个真实行动已经改变了结果。",
    )


def _task(task_id: str, parent: str | None, kind: str, status: TaskStatus) -> TaskRecord:
    return TaskRecord(
        task_id=task_id,
        parent_task_id=parent,
        display_path=task_id,
        specs=[
            TaskSpec(
                version=1,
                objective="bounded",
                acceptance_criteria=(AcceptanceCriterion("closed", "closed"),),
                context_refs=(
                    f"script-build://task-kinds/{kind}",
                    f"script-build://inputs/{SNAPSHOT_ID}",
                ),
            )
        ],
        status=status,
    )


def _compose_contract() -> ScriptTaskContractV1:
    return ScriptTaskContractV1(
        ScriptTaskKind.COMPOSE,
        SCOPE,
        ScriptIntentClass.COMPOSE,
        "compose one complete candidate",
        (),
        None,
        (WRITE,),
        None,
        "structured-script/v1",
        (ScriptCriterion("closed", "closed"),),
        ScriptTaskBudget(),
        ("goal-1",),
    )


def _portfolio_contract(ref: ArtifactRef) -> ScriptTaskContractV1:
    accepted = AcceptedDecisionRef("decision-compose", ref, SCOPE, ScriptTaskKind.COMPOSE)
    return ScriptTaskContractV1(
        ScriptTaskKind.CANDIDATE_PORTFOLIO,
        SCOPE,
        ScriptIntentClass.PORTFOLIO,
        "govern the closed candidate set",
        (accepted,),
        None,
        (WRITE,),
        None,
        "candidate-portfolio/v1",
        (ScriptCriterion("closed", "closed"),),
        ScriptTaskBudget(),
        ("goal-1",),
        candidate_closure_decision_refs=(accepted,),
        adopted_decision_ids=("decision-compose",),
        compose_order=("decision-compose",),
    )


class _Store:
    def __init__(self, value: object) -> None:
        self.value = value

    async def load(self, root: str) -> TaskLedger:
        assert root == "root"
        assert isinstance(self.value, TaskLedger)
        return self.value

    async def get(self, root: str, snapshot: str) -> ArtifactSnapshot:
        assert root == "root"
        return self.value[snapshot]  # type: ignore[index]


class _Contracts:
    def __init__(self, values: dict[str, ScriptTaskContractV1]) -> None:
        self.values = values

    async def read_for_task(
        self, *, root_trace_id: str, task: TaskRecord, spec_version: int | None = None
    ) -> ScriptTaskContractV1:
        assert root_trace_id == "root" and spec_version == 1
        return self.values[task.task_id]


class _Artifacts:
    def __init__(self, versions: dict[int, ArtifactVersion]) -> None:
        self.versions = versions

    async def read_by_ref(
        self,
        ref: ArtifactRef,
        *,
        script_build_id: int,
        task_id: str | None = None,
        attempt_id: str | None = None,
    ) -> ArtifactVersion:
        assert script_build_id == 7 and ref.version is not None
        value = self.versions[int(ref.version)]
        assert value.task_id == task_id and value.attempt_id == attempt_id
        assert value.canonical_sha256 == ref.digest
        return value

    async def get_by_id(self, identifier: int, *, script_build_id: int) -> ArtifactVersion:
        assert script_build_id == 7
        return self.versions[identifier]


class _Bindings:
    async def get_by_build(self, script_build_id: int) -> MissionBinding:
        assert script_build_id == 7
        now = datetime.now(UTC)
        return MissionBinding(1, 7, "root", 11, 9, None, "test", "v1", now, now)


def _version(
    identifier: int,
    task_id: str,
    attempt_id: str,
    kind: ArtifactKind,
    digest: str,
    artifact: object,
) -> ArtifactVersion:
    now = datetime.now(UTC)
    return ArtifactVersion(
        identifier,
        7,
        task_id,
        attempt_id,
        1,
        kind,
        digest,
        ArtifactState.FROZEN,
        artifact,  # type: ignore[arg-type]
        now,
        now,
    )


async def _fixture() -> tuple[ScriptPhaseTwoBoundaryVerifier, TaskLedger]:
    script_ref = ArtifactRef(
        "script-build://artifact-versions/1",
        ArtifactKind.STRUCTURED_SCRIPT.value,
        "1",
        SCRIPT_DIGEST,
    )
    root = _task("root-task", None, "root", TaskStatus.BLOCKED)
    portfolio_task = _task("portfolio", "root-task", "candidate-portfolio", TaskStatus.RUNNING)
    compose_task = _task("compose", "portfolio", "compose", TaskStatus.COMPLETED)
    root.child_task_ids.append("portfolio")
    portfolio_task.child_task_ids.append("compose")

    compose_attempt = TaskAttempt(
        "attempt-compose",
        "compose",
        1,
        "worker-compose",
        "script_compose_worker",
        "new",
        (),
        AttemptStatus.SUBMITTED,
        snapshot_id="snapshot-compose",
        submission=AttemptSubmission("closed", artifact_refs=[script_ref]),
    )
    compose_validation = ValidationReport(
        "validation-compose",
        "compose",
        "attempt-compose",
        1,
        "snapshot-compose",
        "validator-compose",
        status=ValidationRunStatus.COMPLETED,
        verdict=ValidationVerdict.PASSED,
    )
    compose_decision = PlannerDecision(
        "decision-compose",
        "compose",
        DecisionAction.ACCEPT,
        "passed",
        TaskStatus.AWAITING_DECISION,
        TaskStatus.COMPLETED,
        "attempt-compose",
        "validation-compose",
    )
    compose_task.attempt_ids.append("attempt-compose")
    compose_task.validation_ids.append("validation-compose")
    compose_task.decision_ids.append("decision-compose")

    portfolio_attempt = TaskAttempt(
        "attempt-portfolio",
        "portfolio",
        1,
        "worker-portfolio",
        "script_candidate_portfolio_worker",
        "new",
        ("decision-compose",),
    )
    ledger = TaskLedger(
        "root",
        "mission",
        "root-task",
        tasks={item.task_id: item for item in (root, portfolio_task, compose_task)},
        attempts={
            compose_attempt.attempt_id: compose_attempt,
            portfolio_attempt.attempt_id: portfolio_attempt,
        },
        validations={compose_validation.validation_id: compose_validation},
        decisions={compose_decision.decision_id: compose_decision},
    )
    compose_normalized = _normalized("closed", script_ref)
    snapshots = {
        "snapshot-compose": ArtifactSnapshot(
            "snapshot-compose",
            "attempt-compose",
            compose_normalized,
            _snapshot_sha(compose_normalized),
            [script_ref],
            [],
        )
    }
    paragraph = _complete_paragraph()
    element = ScriptElementV1(1, "可观察反转", "实质", "叙事证据")
    link = ScriptParagraphElementLinkV1(1, 1)
    lineage = CandidateLineageV1(
        scope_ref=SCOPE,
        input_snapshot_ref=f"script-build://inputs/{SNAPSHOT_ID}",
        input_closure_digest=INPUT_DIGEST,
        write_scope=("script-build://writes/paragraphs",),
        goal_ids=("goal-1",),
    )
    structured = StructuredScriptArtifactV1(
        "script-build://artifact-versions/9",
        INPUT_DIGEST,
        (paragraph,),
        (element,),
        (link,),
        (
            "script-build://artifact-versions/6",
            "script-build://artifact-versions/7",
            "script-build://artifact-versions/8",
        ),
        (
            GoalCoverage(
                "goal-1",
                (
                    "script-build://artifact-versions/7",
                    "script-build://artifact-versions/8",
                ),
            ),
        ),
        (),
        (),
        SCRIPT_DIGEST,
    )
    artifacts = _Artifacts(
        {
            1: _version(
                1,
                "compose",
                "attempt-compose",
                ArtifactKind.STRUCTURED_SCRIPT,
                SCRIPT_DIGEST,
                structured,
            ),
            9: _version(
                9,
                "direction",
                "attempt-direction",
                ArtifactKind.DIRECTION,
                INPUT_DIGEST,
                DirectionArtifact(
                    goals=(
                        DirectionGoal(
                            "goal-1",
                            "complete the script",
                            "the delivery needs one complete script",
                            success_criteria=("the script is complete",),
                        ),
                    ),
                    evidence_refs=("script-build://artifact-versions/99",),
                ),
            ),
            6: _version(
                6,
                "structure",
                "attempt-structure",
                ArtifactKind.STRUCTURE,
                INPUT_DIGEST,
                StructureArtifactV1(lineage, (paragraph,)),
            ),
            7: _version(
                7,
                "paragraph",
                "attempt-paragraph",
                ArtifactKind.PARAGRAPH,
                INPUT_DIGEST,
                ParagraphArtifactV1(
                    lineage,
                    (paragraph,),
                    (element,),
                    (link,),
                    {"created": {"paragraph_ids": [1]}},
                ),
            ),
            8: _version(
                8,
                "elements",
                "attempt-elements",
                ArtifactKind.ELEMENT_SET,
                INPUT_DIGEST,
                ElementSetArtifactV1(
                    replace(lineage, write_scope=("script-build://writes/elements",)),
                    (element,),
                    (link,),
                    (paragraph,),
                    {"created": {"element_ids": [1]}},
                ),
            ),
        }
    )
    contracts = _Contracts(
        {"compose": _compose_contract(), "portfolio": _portfolio_contract(script_ref)}
    )
    accepted = AcceptedInputResolver(
        task_store=_Store(ledger),
        artifact_store=_Store(snapshots),
        artifacts=artifacts,
        contracts=contracts,
    )
    bundle = await accepted.resolve(
        root_trace_id="root",
        script_build_id=7,
        task_id="portfolio",
        attempt_id="attempt-portfolio",
        contract=contracts.values["portfolio"],
        input_snapshot_id=SNAPSHOT_ID,
    )
    portfolio = CandidatePortfolioArtifactV1(
        script_ref.uri,
        (script_ref.uri,),
        ("decision-compose",),
        (),
        (),
        bundle.input_closure_digest,
        (),
        ("decision-compose",),
        PORTFOLIO_DIGEST,
    )
    portfolio_ref = ArtifactRef(
        "script-build://artifact-versions/2",
        ArtifactKind.CANDIDATE_PORTFOLIO.value,
        "2",
        PORTFOLIO_DIGEST,
    )
    portfolio_attempt.status = AttemptStatus.SUBMITTED
    portfolio_attempt.snapshot_id = "snapshot-portfolio"
    portfolio_attempt.submission = AttemptSubmission("closed", artifact_refs=[portfolio_ref])
    portfolio_validation = ValidationReport(
        "validation-portfolio",
        "portfolio",
        "attempt-portfolio",
        1,
        "snapshot-portfolio",
        "validator-portfolio",
        status=ValidationRunStatus.COMPLETED,
        verdict=ValidationVerdict.PASSED,
    )
    portfolio_decision = PlannerDecision(
        "decision-portfolio",
        "portfolio",
        DecisionAction.ACCEPT,
        "passed",
        TaskStatus.AWAITING_DECISION,
        TaskStatus.COMPLETED,
        "attempt-portfolio",
        "validation-portfolio",
    )
    portfolio_task.status = TaskStatus.COMPLETED
    portfolio_task.attempt_ids.append("attempt-portfolio")
    portfolio_task.validation_ids.append("validation-portfolio")
    portfolio_task.decision_ids.append("decision-portfolio")
    ledger.validations[portfolio_validation.validation_id] = portfolio_validation
    ledger.decisions[portfolio_decision.decision_id] = portfolio_decision
    portfolio_normalized = _normalized("closed", portfolio_ref)
    snapshots["snapshot-portfolio"] = ArtifactSnapshot(
        "snapshot-portfolio",
        "attempt-portfolio",
        portfolio_normalized,
        _snapshot_sha(portfolio_normalized),
        [portfolio_ref],
        [],
    )
    artifacts.versions[2] = _version(
        2,
        "portfolio",
        "attempt-portfolio",
        ArtifactKind.CANDIDATE_PORTFOLIO,
        PORTFOLIO_DIGEST,
        portfolio,
    )
    verifier = ScriptPhaseTwoBoundaryVerifier(
        task_store=_Store(ledger),
        artifact_store=_Store(snapshots),
        artifacts=artifacts,
        bindings=_Bindings(),
        accepted_inputs=accepted,
    )
    return verifier, ledger


@pytest.mark.asyncio
async def test_phase_two_boundary_requires_one_closed_adopted_portfolio() -> None:
    verifier, _ = await _fixture()
    await verifier.verify_checkpoint(
        script_build_id=7, root_trace_id="root", input_snapshot_id=SNAPSHOT_ID
    )


@pytest.mark.asyncio
async def test_phase_two_boundary_rejects_nonterminal_subtree_and_early_advance() -> None:
    verifier, ledger = await _fixture()
    ledger.tasks["compose"].status = TaskStatus.RUNNING
    with pytest.raises(PhaseTwoBoundaryNotReady, match="subtree is not terminal"):
        await verifier.verify_checkpoint(
            script_build_id=7, root_trace_id="root", input_snapshot_id=SNAPSHOT_ID
        )
    with pytest.raises(PhaseTwoBoundaryNotReady, match="portfolio already exists"):
        await verifier.verify_before_advance(
            script_build_id=7, root_trace_id="root", input_snapshot_id=SNAPSHOT_ID
        )


@pytest.mark.asyncio
async def test_phase_two_boundary_rejects_closure_digest_drift() -> None:
    verifier, _ = await _fixture()
    version = verifier._artifacts.versions[2]
    artifact = replace(version.artifact, input_closure_digest="sha256:" + "f" * 64)
    verifier._artifacts.versions[2] = replace(version, artifact=artifact)
    with pytest.raises(PhaseTwoBoundaryNotReady, match="closure digest"):
        await verifier.verify_checkpoint(
            script_build_id=7, root_trace_id="root", input_snapshot_id=SNAPSHOT_ID
        )


def _normalized(summary: str, ref: ArtifactRef) -> dict[str, object]:
    return {
        "summary": summary,
        "artifact_refs": [
            {
                "uri": ref.uri,
                "kind": ref.kind,
                "version": ref.version,
                "digest": ref.digest,
                "summary": ref.summary,
                "metadata": ref.metadata,
            }
        ],
        "evidence_refs": [],
    }


def _snapshot_sha(value: dict[str, object]) -> str:
    canonical = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode()).hexdigest()