主机：收紧候选物化与 Compose 闭包

统一候选 lineage、Direction 绑定、Paragraph-first、Compose source closure 和 Portfolio adoption 验证；在 legacy 写入前执行完整脚本与 Goal Coverage 检查。

script_build_host/src/script_build_host/application/phase_two_candidates.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 
 import base64
 from collections.abc import Mapping, Sequence
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, replace
 from hashlib import sha256
 from typing import Any, Protocol, cast
 
@@ -28,8 +28,12 @@ from script_build_host.domain.artifacts import (
     EvidenceRecordV1,
 )
 from script_build_host.domain.candidate_validation import (
+    artifact_realizes_goals,
+    candidate_content_issues,
     comparison_fairness_errors,
     placeholder_paths,
+    require_complete_script,
+    require_complete_sources,
 )
 from script_build_host.domain.defects import ScriptDefectV1
 from script_build_host.domain.errors import ProtocolViolation, ScriptBuildError
@@ -44,6 +48,7 @@ from script_build_host.domain.phase_two_artifacts import (
     ElementSetArtifactV1,
     ParagraphArtifactV1,
     ScriptElementV1,
+    ScriptParagraphElementLinkV1,
     ScriptParagraphV1,
     StructureArtifactV1,
     StructuredScriptArtifactV1,
@@ -344,15 +349,19 @@ class PhaseTwoCandidateService:
                 ),
             }
         ]
+        content_issues = candidate_content_issues(version.artifact)
         placeholders = placeholder_paths(version.artifact)
+        content_errors = tuple(
+            [*(f"{item.path}: {item.message}" for item in content_issues), *placeholders]
+        )
         rules.append(
             {
                 "rule_id": "realized-content",
-                "verdict": "failed" if placeholders else "passed",
+                "verdict": "failed" if content_errors else "passed",
                 "reason": (
-                    "Unrealized content at " + ", ".join(placeholders[:8])
-                    if placeholders
-                    else "Artifact contains no placeholder or deferred-production prose"
+                    "Unrealized content at " + ", ".join(content_errors[:8])
+                    if content_errors
+                    else "Artifact satisfies its deterministic content contract"
                 ),
             }
         )
@@ -455,12 +464,17 @@ class PhaseTwoCandidateService:
                     raise PhaseTwoCandidateError(
                         "GOAL_COVERAGE_INVALID", "StructuredScript uses a stale Direction"
                     )
+                realized_refs = await self._realized_source_refs(
+                    script_build_id=scope.script_build_id,
+                    source_refs=structured.source_artifact_refs,
+                )
                 validate_goal_coverage(
                     direction_goal_ids=tuple(
                         item.goal_id for item in direction_version.artifact.goals
                     ),
                     coverage=structured.goal_coverage,
                     adopted_source_refs=structured.source_artifact_refs,
+                    realized_source_refs=realized_refs,
                 )
             except ScriptBuildError as exc:
                 coverage_error = f"{exc.code}: {exc.summary}"
@@ -473,6 +487,39 @@ class PhaseTwoCandidateService:
             )
         return tuple(rules)
 
+    async def validation_evidence_refs(
+        self, *, context: Mapping[str, Any]
+    ) -> tuple[ArtifactRef, ...]:
+        scope, bundle = await self._resolve_bundle(context)
+        snapshot = await self._framework_artifact_store.get(
+            scope.root_trace_id, _required_context(context, "snapshot_id")
+        )
+        values = [
+            *snapshot.artifact_refs,
+            *snapshot.evidence_refs,
+            *(item.artifact_ref for item in bundle.inputs),
+        ]
+        output: list[ArtifactRef] = []
+        seen: set[tuple[str, str | None, str | None, str]] = set()
+        for ref in values:
+            key = (ref.uri, ref.version, ref.digest, ref.kind)
+            if key not in seen:
+                seen.add(key)
+                output.append(ref)
+        return tuple(output)
+
+    async def _realized_source_refs(
+        self, *, script_build_id: int, source_refs: Sequence[str]
+    ) -> tuple[str, ...]:
+        realized: list[str] = []
+        for ref in source_refs:
+            version = await self._artifacts.get_by_id(
+                _artifact_version_id(ref), script_build_id=script_build_id
+            )
+            if artifact_realizes_goals(version.artifact):
+                realized.append(ref)
+        return tuple(realized)
+
     async def read_attempt_workspace(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
         scope = await self._scope(context)
         if scope.contract.task_kind in {ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH}:
@@ -670,25 +717,30 @@ class PhaseTwoCandidateService:
         direction_ref, evidence_refs = _compose_support_refs(support_versions)
         direction = _accepted_direction(support_versions)
         source_refs = tuple(item.artifact_ref.uri for item in frontier)
+        _require_complete_frontier(versions)
+        creative_sources: list[tuple[str, tuple[str, ...]]] = []
+        for version in versions:
+            artifact = version.artifact
+            if isinstance(artifact, (ParagraphArtifactV1, ElementSetArtifactV1)) and (
+                artifact_realizes_goals(artifact)
+            ):
+                creative_sources.append(
+                    (
+                        f"script-build://artifact-versions/{version.artifact_version_id}",
+                        artifact.lineage.goal_ids,
+                    )
+                )
+        realized_refs = tuple(ref for ref, _ in creative_sources)
         goal_coverage = build_goal_coverage(
             direction_goal_ids=tuple(item.goal_id for item in direction.goals),
-            creative_sources=(
-                (
-                    f"script-build://artifact-versions/{version.artifact_version_id}",
-                    version.artifact.lineage.goal_ids,
-                )
-                for version in versions
-                if isinstance(
-                    version.artifact,
-                    (StructureArtifactV1, ParagraphArtifactV1, ElementSetArtifactV1),
-                )
-            ),
+            creative_sources=creative_sources,
         )
         _require_paragraph_structures(versions)
         validate_goal_coverage(
             direction_goal_ids=tuple(item.goal_id for item in direction.goals),
             coverage=goal_coverage,
             adopted_source_refs=source_refs,
+            realized_source_refs=realized_refs,
         )
         workspace = await self._workspace_repo().get_or_create(
             scope.write_context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT
@@ -791,10 +843,19 @@ class PhaseTwoCandidateService:
             raise PhaseTwoCandidateError(
                 "GOAL_COVERAGE_INVALID", "StructuredScript uses a stale Direction"
             )
+        require_complete_script(
+            adopted_script.paragraphs,
+            adopted_script.elements,
+            adopted_script.paragraph_element_links,
+        )
         validate_goal_coverage(
             direction_goal_ids=tuple(item.goal_id for item in direction_version.artifact.goals),
             coverage=adopted_script.goal_coverage,
             adopted_source_refs=adopted_script.source_artifact_refs,
+            realized_source_refs=await self._realized_source_refs(
+                script_build_id=scope.script_build_id,
+                source_refs=adopted_script.source_artifact_refs,
+            ),
         )
         superseded = _text_sequence(payload.get("superseded_decision_ids"))
         if not set(superseded) <= set(scope.contract.held_or_rejected_decision_ids):
@@ -1375,6 +1436,11 @@ class PhaseTwoCandidateService:
                 "LEGACY_REFERENCE_INVALID",
                 "every adopted Element must be linked to an adopted Paragraph",
             )
+        _require_complete_projection(
+            paragraph_projection=paragraph_projection,
+            element_projection=element_projection,
+            links=active_links,
+        )
 
         paragraph_ids: dict[tuple[str, str, int], int] = {}
         next_index = 1
@@ -1616,6 +1682,52 @@ def _require_paragraph_structures(versions: Sequence[ArtifactVersion]) -> None:
             )
 
 
+def _require_complete_frontier(versions: Sequence[ArtifactVersion]) -> None:
+    artifacts = tuple(item.artifact for item in versions)
+    require_complete_sources(artifacts)
+
+
+def _require_complete_projection(
+    *,
+    paragraph_projection: Mapping[
+        tuple[str, str, int],
+        tuple[ScriptParagraphV1, tuple[str, str, int] | None],
+    ],
+    element_projection: Mapping[tuple[str, str, int], ScriptElementV1],
+    links: set[tuple[tuple[str, str, int], tuple[str, str, int]]],
+) -> None:
+    paragraph_keys = {
+        key: index
+        for index, key in enumerate(sorted(paragraph_projection), start=1)
+    }
+    element_keys = {
+        key: index for index, key in enumerate(sorted(element_projection), start=1)
+    }
+    paragraphs = tuple(
+        replace(
+            paragraph,
+            paragraph_id=paragraph_keys[key],
+            paragraph_index=index,
+            parent_id=paragraph_keys[parent_key] if parent_key is not None else None,
+        )
+        for index, (key, (paragraph, parent_key)) in enumerate(
+            sorted(paragraph_projection.items()), start=1
+        )
+    )
+    elements = tuple(
+        replace(element, element_id=element_keys[key])
+        for key, element in sorted(element_projection.items())
+    )
+    projected_links = tuple(
+        ScriptParagraphElementLinkV1(
+            paragraph_id=paragraph_keys[paragraph_key],
+            element_id=element_keys[element_key],
+        )
+        for paragraph_key, element_key in sorted(links)
+    )
+    require_complete_script(paragraphs, elements, projected_links)
+
+
 def _image_media_type(content: bytes) -> str:
     if content.startswith(b"\x89PNG\r\n\x1a\n"):
         return "image/png"

script_build_host/tests/test_phase_two_candidates.py

@@ -27,10 +27,16 @@ from script_build_host.domain.artifacts import (
 )
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.domain.phase_two_artifacts import (
+    CandidateLineageV1,
     CandidatePortfolioArtifactV1,
     ComparisonArtifactV1,
+    ElementSetArtifactV1,
     GoalCoverage,
+    ParagraphArtifactV1,
+    ScriptElementV1,
+    ScriptParagraphElementLinkV1,
     ScriptParagraphV1,
+    StructureArtifactV1,
     StructuredScriptArtifactV1,
 )
 from script_build_host.domain.task_contracts import (
@@ -57,6 +63,74 @@ SCOPE = "script-build://scopes/opening"
 WRITE = "script-build://writes/paragraphs/opening"
 
 
+def _complete_paragraph() -> ScriptParagraphV1:
+    atom = ({"原子点": "内容", "维度": "开场"},)
+    return ScriptParagraphV1(
+        1,
+        1,
+        1,
+        None,
+        "opening",
+        {"beats": ["抛出问题", "给出答案"]},
+        theme="为什么要主动改变",
+        form="问题与回答",
+        function="建立紧迫感",
+        feeling="坚定且真诚",
+        theme_elements=atom,
+        form_elements=atom,
+        function_elements=atom,
+        feeling_elements=atom,
+        description="先指出停滞的代价。再给出可执行的转折。",
+        full_description="你可能一直在等待更好的时机。但真正的改变来自今天的第一步。",
+    )
+
+
+def _complete_element() -> ScriptElementV1:
+    return ScriptElementV1(1, "停滞代价", "实质", "核心论点")
+
+
+def _complete_link() -> ScriptParagraphElementLinkV1:
+    return ScriptParagraphElementLinkV1(1, 1)
+
+
+def _lineage() -> CandidateLineageV1:
+    return CandidateLineageV1(
+        scope_ref=SCOPE,
+        input_snapshot_ref="script-build://input-snapshots/11",
+        input_closure_digest=DIGEST,
+        write_scope=(WRITE,),
+        goal_ids=("goal-1",),
+    )
+
+
+def _source_artifacts() -> tuple[
+    StructureArtifactV1,
+    ParagraphArtifactV1,
+    ElementSetArtifactV1,
+]:
+    paragraph = _complete_paragraph()
+    element = _complete_element()
+    link = _complete_link()
+    lineage = _lineage()
+    return (
+        StructureArtifactV1(lineage, (paragraph,)),
+        ParagraphArtifactV1(
+            lineage,
+            (paragraph,),
+            (element,),
+            (link,),
+            {"created": {"paragraph_ids": [1]}},
+        ),
+        ElementSetArtifactV1(
+            lineage,
+            (element,),
+            (link,),
+            (paragraph,),
+            {"created": {"element_ids": [1]}},
+        ),
+    )
+
+
 def _contract(
     kind: ScriptTaskKind,
     *,
@@ -103,12 +177,15 @@ def _contract(
 
 
 class _Bindings:
+    def __init__(self, active_direction_artifact_version_id: int = 10) -> None:
+        self.active_direction_artifact_version_id = active_direction_artifact_version_id
+
     async def get_by_root(self, root_trace_id: str) -> Any:
         return SimpleNamespace(
             root_trace_id=root_trace_id,
             script_build_id=7,
             input_snapshot_id=11,
-            active_direction_artifact_version_id=10,
+            active_direction_artifact_version_id=self.active_direction_artifact_version_id,
         )
 
 
@@ -688,20 +765,53 @@ async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract
         DIGEST,
     )
     (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
-    paragraph = ScriptParagraphV1(1, 1, 1, None, "opening", {})
+    paragraph = _complete_paragraph()
+    element = _complete_element()
+    link = _complete_link()
+
+    def source_versions(identifier: int) -> dict[str, ArtifactVersion]:
+        base = 20 + identifier * 10
+        versions: dict[str, ArtifactVersion] = {}
+        for offset, (kind, artifact) in enumerate(
+            zip(
+                (
+                    ArtifactKind.STRUCTURE,
+                    ArtifactKind.PARAGRAPH,
+                    ArtifactKind.ELEMENT_SET,
+                ),
+                _source_artifacts(),
+                strict=True,
+            ),
+            start=1,
+        ):
+            version_id = base + offset
+            uri = f"script-build://artifact-versions/{version_id}"
+            versions[uri] = ArtifactVersion(
+                version_id,
+                7,
+                f"source-{version_id}",
+                f"source-attempt-{version_id}",
+                1,
+                kind,
+                DIGEST,
+                ArtifactState.FROZEN,
+                artifact,
+                datetime.now(UTC),
+                datetime.now(UTC),
+            )
+        return versions
 
     def structured(identifier: int) -> ArtifactVersion:
+        source_refs = tuple(source_versions(identifier))
         artifact = StructuredScriptArtifactV1(
             direction_ref="script-build://artifact-versions/10",
             input_closure_digest=DIGEST,
             paragraphs=(paragraph,),
-            elements=(),
-            paragraph_element_links=(),
-            source_artifact_refs=(f"script-build://artifact-versions/{20 + identifier}",),
+            elements=(element,),
+            paragraph_element_links=(link,),
+            source_artifact_refs=source_refs,
             goal_coverage=(
-                GoalCoverage(
-                    "goal-1", (f"script-build://artifact-versions/{20 + identifier}",)
-                ),
+                GoalCoverage("goal-1", (source_refs[1], source_refs[2])),
             ),
             evidence_refs=(),
             acceptance_notes=("accepted",),
@@ -721,7 +831,13 @@ async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract
             datetime.now(UTC),
         )
 
-    artifacts = _Artifacts({ref1.uri: structured(1), ref2.uri: structured(2)})
+    versions = {
+        **source_versions(1),
+        **source_versions(2),
+        ref1.uri: structured(1),
+        ref2.uri: structured(2),
+    }
+    artifacts = _Artifacts(versions)
     service = PhaseTwoCandidateService(
         bindings=cast(Any, _Bindings()),
         task_store=_TaskStore(ledger),
@@ -794,25 +910,57 @@ async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract
 async def test_candidate_portfolio_command_replay_reuses_one_frozen_version(database: Any) -> None:
     _, sessions = database
     repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
-    paragraph = ScriptParagraphV1(1, 1, 1, None, "opening", {})
+    paragraph = _complete_paragraph()
+    element = _complete_element()
+    link = _complete_link()
+    direction_version, direction_ref = await repository.freeze(
+        script_build_id=7,
+        task_id="direction",
+        attempt_id="direction-attempt",
+        spec_version=1,
+        artifact=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/90",),
+        ),
+    )
 
     async def freeze_structured(identifier: int) -> tuple[ArtifactVersion, ArtifactRef]:
+        structure, paragraph_source, element_source = _source_artifacts()
+        source_refs: list[str] = []
+        for suffix, source in (
+            ("structure", structure),
+            ("paragraph", paragraph_source),
+            ("elements", element_source),
+        ):
+            _, source_ref = await repository.freeze(
+                script_build_id=7,
+                task_id=f"{suffix}-{identifier}",
+                attempt_id=f"{suffix}-attempt-{identifier}",
+                spec_version=1,
+                artifact=source,
+            )
+            source_refs.append(source_ref.uri)
         return await repository.freeze(
             script_build_id=7,
             task_id=f"compose-{identifier}",
             attempt_id=f"compose-attempt-{identifier}",
             spec_version=1,
             artifact=StructuredScriptArtifactV1(
-                direction_ref="script-build://artifact-versions/10",
+                direction_ref=direction_ref.uri,
                 input_closure_digest=DIGEST,
                 paragraphs=(paragraph,),
-                elements=(),
-                paragraph_element_links=(),
-                source_artifact_refs=(f"script-build://artifact-versions/{20 + identifier}",),
+                elements=(element,),
+                paragraph_element_links=(link,),
+                source_artifact_refs=tuple(source_refs),
                 goal_coverage=(
-                    GoalCoverage(
-                        "goal-1", (f"script-build://artifact-versions/{20 + identifier}",)
-                    ),
+                    GoalCoverage("goal-1", (source_refs[1], source_refs[2])),
                 ),
                 evidence_refs=(),
                 acceptance_notes=("accepted",),
@@ -857,7 +1005,7 @@ async def test_candidate_portfolio_command_replay_reuses_one_frozen_version(data
         attempt_id="portfolio-attempt",
     )
     service = PhaseTwoCandidateService(
-        bindings=cast(Any, _Bindings()),
+        bindings=cast(Any, _Bindings(direction_version.artifact_version_id)),
         task_store=_TaskStore(ledger),
         framework_artifact_store=_FrameworkArtifacts(),
         artifacts=repository,