主机：建立正式脚本内容完整性合同

集中检查叶子正文、段落原子、独立 Element、Paragraph–Element Link、重复名称和维度；区分 Structure 规划覆盖与 Paragraph、ElementSet 的真实实现，并补齐空心脚本和父子 Goal 测试。

script_build_host/src/script_build_host/domain/candidate_validation.py

@@ -3,16 +3,42 @@
 from __future__ import annotations
 
 import re
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
 from typing import Any
 
-from .phase_two_artifacts import ComparisonArtifactV1
+from .errors import ScriptBuildError
+from .phase_two_artifacts import (
+    ComparisonArtifactV1,
+    ElementSetArtifactV1,
+    ParagraphArtifactV1,
+    ScriptElementV1,
+    ScriptParagraphElementLinkV1,
+    ScriptParagraphV1,
+    StructureArtifactV1,
+    StructuredScriptArtifactV1,
+)
 
 _PLACEHOLDER = re.compile(
     r"(?:TODO|TBD|\[\s*待补|<placeholder>|后续(?:应|需|将)产出|"
     r"此处插入|此处(?:描述|说明|展示|呈现).{0,12}(?:存在|内容|信息)?)",
     re.IGNORECASE,
 )
+_META_PROSE = re.compile(
+    r"(?:本段|这一段|该段)(?:将|会|主要|旨在|用于)|"
+    r"(?:this|the)\s+(?:paragraph|section)\s+(?:will|should|aims?\s+to)",
+    re.IGNORECASE,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class ContentIssue:
+    path: str
+    message: str
+
+
+class ScriptContentError(ScriptBuildError):
+    pass
 
 
 def placeholder_paths(artifact: Any) -> tuple[str, ...]:
@@ -35,6 +61,266 @@ def placeholder_paths(artifact: Any) -> tuple[str, ...]:
     return tuple(findings)
 
 
+def candidate_content_issues(artifact: Any) -> tuple[ContentIssue, ...]:
+    """Validate the content actually produced by one creative candidate."""
+
+    if isinstance(artifact, StructureArtifactV1):
+        return _paragraph_shape_issues(artifact.paragraphs)
+    if isinstance(artifact, ParagraphArtifactV1):
+        target_ids = _changed_ids(artifact.change_manifest, "paragraph")
+        if not artifact.change_manifest:
+            target_ids = {item.paragraph_id for item in artifact.paragraphs if item.is_active}
+        targets = tuple(
+            item
+            for item in artifact.paragraphs
+            if item.is_active and item.paragraph_id in target_ids
+        )
+        if not targets:
+            return (ContentIssue("paragraphs", "Paragraph candidate changes no active paragraph"),)
+        leaf_ids = _leaf_ids(artifact.paragraphs)
+        leaves = tuple(item for item in targets if item.paragraph_id in leaf_ids)
+        if not leaves:
+            return (ContentIssue("paragraphs", "Paragraph candidate realizes no leaf paragraph"),)
+        return tuple(
+            issue
+            for paragraph in leaves
+            for issue in _leaf_prose_issues(paragraph, f"paragraphs[{paragraph.paragraph_id}]")
+        )
+    if isinstance(artifact, ElementSetArtifactV1):
+        element_ids = _changed_ids(artifact.change_manifest, "element")
+        if not artifact.change_manifest:
+            element_ids = {item.element_id for item in artifact.elements if item.is_active}
+        target_elements = tuple(
+            item
+            for item in artifact.elements
+            if item.is_active and item.element_id in element_ids
+        )
+        if not target_elements:
+            return (ContentIssue("elements", "ElementSet changes no active Element"),)
+        linked = {item.element_id for item in artifact.paragraph_element_links}
+        return tuple(
+            ContentIssue(
+                f"elements[{item.element_id}]",
+                "ElementSet Element is not linked to a Paragraph",
+            )
+            for item in target_elements
+            if item.element_id not in linked
+        )
+    if isinstance(artifact, StructuredScriptArtifactV1):
+        return complete_script_issues(
+            artifact.paragraphs,
+            artifact.elements,
+            artifact.paragraph_element_links,
+        )
+    return ()
+
+
+def artifact_realizes_goals(artifact: Any) -> bool:
+    """Only completed Paragraph and ElementSet candidates realize Direction Goals."""
+
+    return isinstance(artifact, (ParagraphArtifactV1, ElementSetArtifactV1)) and not (
+        candidate_content_issues(artifact) or placeholder_paths(artifact)
+    )
+
+
+def script_source_issues(artifacts: Sequence[Any]) -> tuple[ContentIssue, ...]:
+    """Validate the adopted creative source set behind one formal script."""
+
+    issues: list[ContentIssue] = []
+    for label, artifact_type in (
+        ("Structure", StructureArtifactV1),
+        ("Paragraph", ParagraphArtifactV1),
+        ("ElementSet", ElementSetArtifactV1),
+    ):
+        if not any(isinstance(item, artifact_type) for item in artifacts):
+            issues.append(ContentIssue("source_artifact_refs", f"missing {label} source"))
+    for index, artifact in enumerate(artifacts):
+        if not isinstance(
+            artifact,
+            (StructureArtifactV1, ParagraphArtifactV1, ElementSetArtifactV1),
+        ):
+            continue
+        issues.extend(
+            ContentIssue(f"sources[{index}].{item.path}", item.message)
+            for item in candidate_content_issues(artifact)
+        )
+        issues.extend(
+            ContentIssue(f"sources[{index}].{path}", "source contains a placeholder")
+            for path in placeholder_paths(artifact)
+        )
+    return tuple(issues)
+
+
+def require_complete_sources(artifacts: Sequence[Any]) -> None:
+    issues = script_source_issues(artifacts)
+    if issues:
+        summary = "; ".join(f"{item.path}: {item.message}" for item in issues[:8])
+        raise ScriptContentError("SCRIPT_CONTENT_INCOMPLETE", summary)
+
+
+def complete_script_issues(
+    paragraphs: tuple[ScriptParagraphV1, ...],
+    elements: tuple[ScriptElementV1, ...],
+    links: tuple[ScriptParagraphElementLinkV1, ...],
+) -> tuple[ContentIssue, ...]:
+    """Return every hard violation of the formal script delivery contract."""
+
+    issues = list(_paragraph_shape_issues(paragraphs))
+    active_paragraphs = tuple(item for item in paragraphs if item.is_active)
+    active_elements = tuple(item for item in elements if item.is_active)
+    if not active_paragraphs:
+        issues.append(ContentIssue("paragraphs", "Script has no active Paragraph"))
+    leaf_ids = _leaf_ids(paragraphs)
+    for paragraph in active_paragraphs:
+        if paragraph.paragraph_id in leaf_ids:
+            issues.extend(
+                _leaf_prose_issues(paragraph, f"paragraphs[{paragraph.paragraph_id}]")
+            )
+
+    if not active_elements:
+        issues.append(ContentIssue("elements", "Script has no independent Element"))
+    active_paragraph_ids = {item.paragraph_id for item in active_paragraphs}
+    active_element_ids = {item.element_id for item in active_elements}
+    valid_links = tuple(
+        item
+        for item in links
+        if item.paragraph_id in active_paragraph_ids
+        and item.element_id in active_element_ids
+    )
+    if not valid_links:
+        issues.append(ContentIssue("paragraph_element_links", "Script has no active Element links"))
+    for index, link in enumerate(links):
+        if link not in valid_links:
+            issues.append(
+                ContentIssue(
+                    f"paragraph_element_links[{index}]",
+                    "Link must connect an active Paragraph and active Element",
+                )
+            )
+
+    names: dict[str, int] = {}
+    dimensions: dict[tuple[str, str], int] = {}
+    for element in active_elements:
+        name = _normalized(element.name)
+        dimension = (
+            _normalized(element.dimension_primary),
+            _normalized(element.dimension_secondary),
+        )
+        if name in names:
+            issues.append(
+                ContentIssue(
+                    f"elements[{element.element_id}].name",
+                    f"Element name duplicates element {names[name]}",
+                )
+            )
+        else:
+            names[name] = element.element_id
+        if dimension in dimensions:
+            issues.append(
+                ContentIssue(
+                    f"elements[{element.element_id}].dimension",
+                    f"Element dimension duplicates element {dimensions[dimension]}",
+                )
+            )
+        else:
+            dimensions[dimension] = element.element_id
+
+    linked_paragraphs = {item.paragraph_id for item in valid_links}
+    linked_elements = {item.element_id for item in valid_links}
+    for paragraph_id in sorted(leaf_ids - linked_paragraphs):
+        issues.append(
+            ContentIssue(
+                f"paragraphs[{paragraph_id}]",
+                "Leaf Paragraph is not linked to an independent Element",
+            )
+        )
+    for element in active_elements:
+        if element.element_id not in linked_elements:
+            issues.append(
+                ContentIssue(
+                    f"elements[{element.element_id}]",
+                    "Active Element is not linked to a Paragraph",
+                )
+            )
+    return tuple(issues)
+
+
+def require_complete_script(
+    paragraphs: tuple[ScriptParagraphV1, ...],
+    elements: tuple[ScriptElementV1, ...],
+    links: tuple[ScriptParagraphElementLinkV1, ...],
+) -> None:
+    issues = complete_script_issues(paragraphs, elements, links)
+    if issues:
+        summary = "; ".join(f"{item.path}: {item.message}" for item in issues[:8])
+        raise ScriptContentError("SCRIPT_CONTENT_INCOMPLETE", summary)
+
+
+def _paragraph_shape_issues(
+    paragraphs: tuple[ScriptParagraphV1, ...],
+) -> tuple[ContentIssue, ...]:
+    issues: list[ContentIssue] = []
+    active_ids = {item.paragraph_id for item in paragraphs if item.is_active}
+    atom_fields = (
+        "theme_elements",
+        "form_elements",
+        "function_elements",
+        "feeling_elements",
+    )
+    for paragraph in paragraphs:
+        if not paragraph.is_active:
+            continue
+        path = f"paragraphs[{paragraph.paragraph_id}]"
+        if not paragraph.content_range:
+            issues.append(ContentIssue(f"{path}.content_range", "content_range is empty"))
+        if paragraph.parent_id is not None and paragraph.parent_id not in active_ids:
+            issues.append(ContentIssue(f"{path}.parent_id", "active parent is missing"))
+        for field in atom_fields:
+            values = getattr(paragraph, field)
+            if not values or any(not item for item in values):
+                issues.append(ContentIssue(f"{path}.{field}", "atom array is empty"))
+    return tuple(issues)
+
+
+def _leaf_prose_issues(paragraph: ScriptParagraphV1, path: str) -> tuple[ContentIssue, ...]:
+    issues: list[ContentIssue] = []
+    for field in ("theme", "form", "function", "feeling", "description", "full_description"):
+        value = getattr(paragraph, field)
+        if not isinstance(value, str) or not value.strip():
+            issues.append(ContentIssue(f"{path}.{field}", "required prose is empty"))
+        elif _PLACEHOLDER.search(value) or _META_PROSE.search(value):
+            issues.append(
+                ContentIssue(f"{path}.{field}", "required prose is a placeholder or metadata")
+            )
+    return tuple(issues)
+
+
+def _leaf_ids(paragraphs: tuple[ScriptParagraphV1, ...]) -> set[int]:
+    active = {item.paragraph_id for item in paragraphs if item.is_active}
+    parents = {
+        item.parent_id
+        for item in paragraphs
+        if item.is_active and item.parent_id is not None and item.parent_id in active
+    }
+    return active - parents
+
+
+def _changed_ids(manifest: Mapping[str, Any], entity: str) -> set[int]:
+    plural = f"{entity}_ids"
+    created = manifest.get("created")
+    updated = manifest.get("updated")
+    values: list[Any] = []
+    if isinstance(created, Mapping):
+        values.extend(created.get(plural) or ())
+    if isinstance(updated, Mapping):
+        values.extend(updated.get(plural) or ())
+    return {int(item) for item in values if isinstance(item, int) and not isinstance(item, bool)}
+
+
+def _normalized(value: str) -> str:
+    return " ".join(value.split()).casefold()
+
+
 def comparison_fairness_errors(
     artifact: ComparisonArtifactV1,
     expected_criterion_ids: set[str] | None = None,
@@ -68,4 +354,15 @@ def comparison_fairness_errors(
     return tuple(errors)
 
 
-__all__ = ["comparison_fairness_errors", "placeholder_paths"]
+__all__ = [
+    "ContentIssue",
+    "ScriptContentError",
+    "artifact_realizes_goals",
+    "candidate_content_issues",
+    "comparison_fairness_errors",
+    "complete_script_issues",
+    "placeholder_paths",
+    "require_complete_script",
+    "require_complete_sources",
+    "script_source_issues",
+]

script_build_host/tests/test_goal_coverage.py

@@ -46,8 +46,12 @@ from script_build_host.domain.phase_three_artifacts import (
 from script_build_host.domain.phase_two_artifacts import (
     CandidateLineageV1,
     CandidatePortfolioArtifactV1,
+    ElementSetArtifactV1,
     ParagraphArtifactV1,
+    ScriptElementV1,
+    ScriptParagraphElementLinkV1,
     ScriptParagraphV1,
+    StructureArtifactV1,
     StructuredScriptArtifactV1,
     hydrate_phase_two_artifact,
 )
@@ -64,6 +68,62 @@ from script_build_host.domain.task_contracts import (
 _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"),
@@ -233,13 +293,29 @@ async def test_root_preflight_checks_all_goal_nodes_and_frozen_direction_contrac
     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",),
+        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/3",)),
-            GoalCoverage("child", ("script-build://artifact-versions/3",)),
+            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=(),
@@ -286,7 +362,13 @@ async def test_root_preflight_checks_all_goal_nodes_and_frozen_direction_contrac
     service = RootDeliveryService(
         bindings=cast(Any, _Bindings()),
         task_store=cast(Any, _TaskStore(ledger)),
-        artifacts=cast(Any, _Artifacts(manifest_version)),
+        artifacts=cast(
+            Any,
+            _Artifacts(
+                manifest_version,
+                _realized_sources(("parent", "child")),
+            ),
+        ),
     )
 
     async def closure_override(_: Any) -> _RootClosure:
@@ -299,8 +381,9 @@ async def test_root_preflight_checks_all_goal_nodes_and_frozen_direction_contrac
     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())
+    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:
@@ -356,12 +439,20 @@ class _TaskStore:
 
 
 class _Artifacts:
-    def __init__(self, manifest: ArtifactVersion) -> None:
+    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]
+
 
 class _DirectionArtifacts:
     def __init__(self, direction: DirectionArtifact) -> None: