主机：由工具网关生成 Goal ID 与创作闭包

Direction 使用 client_key 批量校验并生成稳定 Goal ID；Paragraph/Element 写入采用批量接口；Compose 由 Host 派生 Artifact ref、digest、scope、Goal Coverage 和完整 source closure。

script_build_host/src/script_build_host/tools/gateway.py

@@ -11,7 +11,9 @@ import re
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime
+from hashlib import sha256
 from typing import Any, Protocol, cast
+from unicodedata import normalize
 from uuid import uuid4
 
 from agent.orchestration import (
@@ -165,6 +167,7 @@ class LegacyScriptToolGateway:
         validation_id: str | None,
         replacement_contract: Mapping[str, Any] | None,
         child_contracts: Sequence[Mapping[str, Any]],
+        selected_decision_ids: Sequence[str],
         context: Mapping[str, Any],
     ) -> Mapping[str, Any]:
         return cast(
@@ -178,6 +181,7 @@ class LegacyScriptToolGateway:
                     validation_id=validation_id,
                     replacement_contract=replacement_contract,
                     child_contracts=child_contracts,
+                    selected_decision_ids=selected_decision_ids,
                     context=context,
                 ),
             ),
@@ -617,9 +621,23 @@ class LegacyScriptToolGateway:
         if task is None:
             raise ProtocolViolation("validation task is outside the protected mission")
         snapshot = await self.coordinator.artifact_store.get(root_trace_id, snapshot_id)
+        allowed_values = [*snapshot.artifact_refs, *snapshot.evidence_refs]
+        kind = task_kind(task.current_spec.context_refs)
+        if kind in PHASE_TWO_KINDS:
+            evidence_reader = getattr(
+                self._candidates(), "validation_evidence_refs", None
+            )
+            if callable(evidence_reader):
+                allowed_values.extend(await evidence_reader(context=context))
+        elif kind == ROOT_DELIVERY_KIND:
+            evidence_reader = getattr(
+                self._root_tools(), "validation_evidence_refs", None
+            )
+            if callable(evidence_reader):
+                allowed_values.extend(await evidence_reader(context=context))
         allowed_refs = {
             (ref.uri, ref.version, ref.digest, ref.kind): ref
-            for ref in (*snapshot.artifact_refs, *snapshot.evidence_refs)
+            for ref in allowed_values
         }
         normalized_defects = _normalize_defects(defects, allowed_refs)
         expected_ids = {item.criterion_id for item in task.current_spec.acceptance_criteria}
@@ -639,12 +657,13 @@ class LegacyScriptToolGateway:
             )
             if defect_codes:
                 reason = f"{reason}; defect_codes={','.join(defect_codes)}"[:500]
-            evidence = _dedupe_refs(
-                ref
-                for item in normalized_defects
-                if item["criterion_id"] == criterion_id
-                for ref in item["evidence_refs"]
+            evidence = _normalize_evidence_refs(
+                raw.get("evidence_refs"),
+                allowed_refs,
+                label="criterion evidence_refs",
             )
+            if criterion_verdict is ValidationVerdict.PASSED and not evidence:
+                raise ProtocolViolation("passed criterion requires frozen evidence_refs")
             result_by_id[criterion_id] = CriterionResult(
                 criterion_id=criterion_id,
                 verdict=criterion_verdict,
@@ -657,6 +676,12 @@ class LegacyScriptToolGateway:
         blocking = [item for item in normalized_defects if item["severity"] in {"hard", "critical"}]
         if overall is ValidationVerdict.PASSED and blocking:
             raise ProtocolViolation("passed validation cannot contain hard or critical defects")
+        criterion_verdicts = {item.verdict for item in result_by_id.values()}
+        every_criterion_passed = criterion_verdicts == {ValidationVerdict.PASSED}
+        if (overall is ValidationVerdict.PASSED) != every_criterion_passed:
+            raise ProtocolViolation(
+                "overall validation verdict must match all criterion verdicts"
+            )
         from agent.orchestration.validation_policy import ValidationContext
 
         attempt = ledger.attempts.get(_required(context, "attempt_id"))
@@ -767,23 +792,10 @@ class LegacyScriptToolGateway:
             raise ProtocolViolation(
                 "direction references evidence outside frozen child ACCEPT decisions"
             )
+        task_id = _required(context, "task_id")
+        direction_goals, goal_ids_by_client_key = _direction_goals(task_id, goals)
         artifact = DirectionArtifact(
-            goals=tuple(
-                DirectionGoal(
-                    goal_id=redact_text(str(item["goal_id"])),
-                    statement=redact_text(str(item["statement"])),
-                    rationale=redact_text(str(item["rationale"])),
-                    parent_goal_id=(
-                        redact_text(str(item["parent_goal_id"]))
-                        if item.get("parent_goal_id") is not None
-                        else None
-                    ),
-                    success_criteria=tuple(
-                        redact_text(str(value)) for value in item["success_criteria"]
-                    ),
-                )
-                for item in goals
-            ),
+            goals=direction_goals,
             constraints=tuple(
                 DirectionConstraint(
                     constraint_id=redact_text(str(item["constraint_id"])),
@@ -817,6 +829,7 @@ class LegacyScriptToolGateway:
             "artifact_ref": _ref_dict(ref),
             "artifact_version_id": version.artifact_version_id,
             "direction": _json_values(version.artifact.content_payload()),
+            "goal_ids_by_client_key": goal_ids_by_client_key,
         }
 
     async def query_validation_evidence(
@@ -1001,6 +1014,68 @@ def _bounded_text(value: Any, label: str, maximum: int) -> str:
     return redact_text(value.strip())
 
 
+def _normalized_client_key(value: Any, label: str) -> str:
+    text = _bounded_text(value, label, 128)
+    normalized = " ".join(normalize("NFKC", text).split()).casefold()
+    if not normalized or len(normalized) > 128:
+        raise ProtocolViolation(f"{label} must contain between 1 and 128 characters")
+    return normalized
+
+
+def _direction_goals(
+    task_id: str, goals: Sequence[Mapping[str, Any]]
+) -> tuple[tuple[DirectionGoal, ...], dict[str, str]]:
+    if not 1 <= len(goals) <= 30:
+        raise ValueError("direction requires between one and thirty Goal nodes")
+    normalized: list[tuple[str, str | None, Mapping[str, Any]]] = []
+    for item in goals:
+        client_key = _normalized_client_key(item.get("client_key"), "goal client_key")
+        parent_key = (
+            _normalized_client_key(
+                item.get("parent_client_key"), "goal parent_client_key"
+            )
+            if item.get("parent_client_key") is not None
+            else None
+        )
+        normalized.append((client_key, parent_key, item))
+    keys = [item[0] for item in normalized]
+    if len(set(keys)) != len(keys):
+        raise ValueError("goal client_key values must be unique")
+    key_set = set(keys)
+    for client_key, parent_key, _ in normalized:
+        if parent_key == client_key:
+            raise ValueError("a goal cannot parent itself")
+        if parent_key is not None and parent_key not in key_set:
+            raise ValueError(f"goal parent_client_key does not exist: {parent_key}")
+    parent_by_key = {key: parent for key, parent, _ in normalized}
+    if any(
+        parent is not None and parent_by_key.get(parent) is not None
+        for _, parent, _ in normalized
+    ):
+        raise ValueError("direction goal hierarchy may contain only two levels")
+    goal_ids = {
+        key: "goal-" + sha256(f"{task_id}\0{key}".encode()).hexdigest()[:16]
+        for key in keys
+    }
+    if len(set(goal_ids.values())) != len(goal_ids):
+        raise ValueError("generated goal identity collision")
+    return (
+        tuple(
+            DirectionGoal(
+                goal_id=goal_ids[client_key],
+                statement=redact_text(str(item["statement"])),
+                rationale=redact_text(str(item["rationale"])),
+                parent_goal_id=goal_ids[parent_key] if parent_key is not None else None,
+                success_criteria=tuple(
+                    redact_text(str(value)) for value in item["success_criteria"]
+                ),
+            )
+            for client_key, parent_key, item in normalized
+        ),
+        goal_ids,
+    )
+
+
 def _attempt_summary(ref: ArtifactRef, scope_ref: str) -> str:
     scope = _bounded_text(scope_ref, "scope_ref", 300)
     if not scope.startswith("script-build://"):
@@ -1040,28 +1115,16 @@ def _normalize_defects(
             or any(not isinstance(item, str) or not item.strip() for item in invalidated)
         ):
             raise ProtocolViolation("invalidated_inputs must be a bounded string array")
-        raw_refs = raw.get("evidence_refs", [])
-        if not isinstance(raw_refs, list) or len(raw_refs) > 64:
-            raise ProtocolViolation("defect evidence_refs must be a bounded array")
-        refs: list[ArtifactRef] = []
-        for item in raw_refs:
-            if not isinstance(item, Mapping):
-                raise ProtocolViolation("defect evidence reference must be an object")
-            candidate = ArtifactRef.from_dict(dict(item))
-            key = (candidate.uri, candidate.version, candidate.digest, candidate.kind)
-            frozen = allowed_refs.get(key)
-            if frozen is None:
-                raise ProtocolViolation(
-                    "defect evidence must come from the fixed validation snapshot"
-                )
-            refs.append(frozen)
+        refs = _normalize_evidence_refs(
+            raw.get("evidence_refs"), allowed_refs, label="defect evidence_refs"
+        )
         values.append(
             {
                 "defect_code": code,
                 "criterion_id": criterion_id,
                 "scope_ref": scope_ref,
                 "observed_excerpt": excerpt,
-                "evidence_refs": tuple(refs),
+                "evidence_refs": refs,
                 "severity": severity,
                 "invalidated_inputs": tuple(str(item) for item in invalidated),
                 "recommended_action_class": action,
@@ -1070,6 +1133,27 @@ def _normalize_defects(
     return tuple(values)
 
 
+def _normalize_evidence_refs(
+    value: Any,
+    allowed_refs: Mapping[tuple[Any, Any, Any, Any], ArtifactRef],
+    *,
+    label: str,
+) -> tuple[ArtifactRef, ...]:
+    if not isinstance(value, list) or len(value) > 64:
+        raise ProtocolViolation(f"{label} must be a bounded array")
+    refs: list[ArtifactRef] = []
+    for item in value:
+        if not isinstance(item, Mapping):
+            raise ProtocolViolation(f"{label} entries must be objects")
+        candidate = ArtifactRef.from_dict(dict(item))
+        key = (candidate.uri, candidate.version, candidate.digest, candidate.kind)
+        frozen = allowed_refs.get(key)
+        if frozen is None:
+            raise ProtocolViolation(f"{label} must come from the protected validation closure")
+        refs.append(frozen)
+    return _dedupe_refs(refs)
+
+
 def _dedupe_refs(values: Any) -> tuple[ArtifactRef, ...]:
     result: list[ArtifactRef] = []
     seen: set[tuple[Any, Any, Any, Any]] = set()

script_build_host/tests/test_direction_artifact.py

@@ -11,6 +11,7 @@ from script_build_host.domain.artifacts import (
 )
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.repositories.sqlalchemy import _hydrate_artifact
+from script_build_host.tools.gateway import _direction_goals
 from script_build_host.tools.registry import _save_direction_candidate_schema
 
 EVIDENCE_REF = "script-build://artifact-versions/1"
@@ -100,6 +101,54 @@ def test_direction_tool_contract_keeps_schema_name_and_removes_model_markdown()
     assert "criteria" not in parameters["properties"]
 
 
+def _client_goal(key: str, parent: str | None = None) -> dict[str, object]:
+    return {
+        "client_key": key,
+        "parent_client_key": parent,
+        "statement": f"statement for {key}",
+        "rationale": f"rationale for {key}",
+        "success_criteria": [f"criterion for {key}"],
+    }
+
+
+def test_direction_client_keys_generate_stable_order_independent_goal_ids() -> None:
+    first, first_map = _direction_goals(
+        "task-a", [_client_goal("Parent"), _client_goal("Child", "Parent")]
+    )
+    second, second_map = _direction_goals(
+        "task-a", [_client_goal(" child ", " parent "), _client_goal("parent")]
+    )
+
+    assert first_map == second_map
+    assert {item.goal_id for item in first} == {item.goal_id for item in second}
+    child = next(item for item in second if item.parent_goal_id)
+    assert child.parent_goal_id == first_map["parent"]
+
+
+@pytest.mark.parametrize(
+    ("goals", "message"),
+    [
+        ([_client_goal("same"), _client_goal("SAME")], "must be unique"),
+        ([_client_goal("child", "missing")], "does not exist"),
+        ([_client_goal("self", "self")], "cannot parent itself"),
+        (
+            [
+                _client_goal("root"),
+                _client_goal("child", "root"),
+                _client_goal("grandchild", "child"),
+            ],
+            "only two levels",
+        ),
+        ([_client_goal(str(index)) for index in range(31)], "between one and thirty"),
+    ],
+)
+def test_direction_client_keys_reject_invalid_batches(
+    goals: list[dict[str, object]], message: str
+) -> None:
+    with pytest.raises(ValueError, match=message):
+        _direction_goals("task-a", goals)
+
+
 def test_direction_hydration_rejects_historical_structure_without_compatibility() -> None:
     historical_payload = {
         "schema_version": "script-direction/v1",

script_build_host/tests/test_phase_two_agent_tools.py

@@ -2,20 +2,21 @@ from __future__ import annotations
 
 import base64
 from collections.abc import Mapping
+from dataclasses import asdict
 from types import SimpleNamespace
 from typing import Any
 
 import pytest
+from agent import ToolRegistry
 from agent.orchestration import ArtifactRef, ValidationVerdict
 from agent.orchestration.evidence import EvidenceResponse
+
 from script_build_host.domain.artifacts import ArtifactState
 from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
 from script_build_host.tools.contracts import AttemptManifest
 from script_build_host.tools.gateway import LegacyScriptToolGateway, RetrievalResult
 from script_build_host.tools.registry import register_script_tools
 
-from agent import ToolRegistry
-
 
 class _CandidateTools:
     def __init__(self, artifact_ref: ArtifactRef) -> None:
@@ -387,7 +388,12 @@ async def test_structured_validation_keeps_excerpt_out_of_framework_summary() ->
     await gateway.submit_structured_validation(
         verdict="failed",
         criterion_results=[
-            {"criterion_id": "criterion-a", "verdict": "failed", "reason": "not realized"}
+            {
+                "criterion_id": "criterion-a",
+                "verdict": "failed",
+                "reason": "not realized",
+                "evidence_refs": [],
+            }
         ],
         defects=[
             {
@@ -455,7 +461,12 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
         await gateway.submit_structured_validation(
             verdict="passed",
             criterion_results=[
-                {"criterion_id": "criterion-a", "verdict": "passed", "reason": "ok"}
+                {
+                    "criterion_id": "criterion-a",
+                    "verdict": "passed",
+                    "reason": "ok",
+                    "evidence_refs": [asdict(ref)],
+                }
             ],
             defects=[defect],
             recommendation="replace",
@@ -471,11 +482,16 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
             "digest": "sha256:" + "b" * 64,
         }
     ]
-    with pytest.raises(ProtocolViolation, match="fixed validation snapshot"):
+    with pytest.raises(ProtocolViolation, match="protected validation closure"):
         await gateway.submit_structured_validation(
             verdict="passed",
             criterion_results=[
-                {"criterion_id": "criterion-a", "verdict": "passed", "reason": "ok"}
+                {
+                    "criterion_id": "criterion-a",
+                    "verdict": "passed",
+                    "reason": "ok",
+                    "evidence_refs": [asdict(ref)],
+                }
             ],
             defects=[defect],
             recommendation="accept",

script_build_host/tests/test_phase_two_workspace.py

@@ -622,6 +622,12 @@ async def test_run_454_shape_roundtrips_without_branch_zero(database) -> None:
         structured_script={
             "direction_ref": "script-build://artifact-versions/999",
             "source_artifact_refs": ["script-build://artifact-versions/888"],
+            "goal_coverage": [
+                {
+                    "goal_id": "goal-1",
+                    "source_artifact_refs": ["script-build://artifact-versions/888"],
+                }
+            ],
             "evidence_refs": ["script-build://artifact-versions/777"],
             "acceptance_notes": ["legacy detail shape preserved"],
         },