收拢检索降级与方向证据门禁

script_build_host/src/script_build_host/adapters/legacy_retrieval.py

@@ -16,11 +16,11 @@ from pathlib import Path
 from typing import Any, Protocol
 
 from script_build_host.adapters.retrieval import SafeHttpClient
+from script_build_host.application.retrieval_ports import RetrievalResult
 from script_build_host.domain.context_broker import ContextCard, rank_context_cards
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.infrastructure.canonical_json import canonical_sha256
 from script_build_host.infrastructure.redaction import redact
-from script_build_host.tools.gateway import RetrievalResult
 
 _ATOM_FIELD_KEYS = ("主题原子点", "形式原子点", "作用原子点", "感受原子点")
 _ATOM_KEY_TO_COLUMN = {

script_build_host/src/script_build_host/adapters/retrieval.py

@@ -15,11 +15,11 @@ from urllib.parse import urljoin
 
 import httpx
 
+from script_build_host.application.retrieval_ports import RetrievalResult
 from script_build_host.domain.errors import ProtocolViolation, UpstreamFailure
 from script_build_host.infrastructure.canonical_json import canonical_json_bytes, canonical_sha256
 from script_build_host.infrastructure.outbound import OutboundPolicy
 from script_build_host.infrastructure.redaction import redact, redact_text
-from script_build_host.tools.gateway import RetrievalResult
 
 
 class SafeHttpClient:

script_build_host/src/script_build_host/agents/prompts/__init__.py

@@ -5,13 +5,10 @@ from .contracts import (
     COMPOSE_WORKER_PROMPT,
     CONTEXT_ACCESS_PROTOCOL,
     CONTEXT_BROKER_PROMPT,
-    DECODE_RETRIEVAL_WORKER_PROMPT,
     DIRECTION_WORKER_PROMPT,
     ELEMENT_SET_WORKER_PROMPT,
-    EXTERNAL_RETRIEVAL_WORKER_PROMPT,
-    KNOWLEDGE_RETRIEVAL_WORKER_PROMPT,
+    EVIDENCE_RETRIEVAL_WORKER_PROMPT,
     PARAGRAPH_WORKER_PROMPT,
-    PATTERN_RETRIEVAL_WORKER_PROMPT,
     PLANNER_PROMPT,
     PORTFOLIO_WORKER_PROMPT,
     RETRIEVAL_VALIDATOR_PROMPT,
@@ -29,13 +26,10 @@ __all__ = [
     "COMPOSE_WORKER_PROMPT",
     "CONTEXT_ACCESS_PROTOCOL",
     "CONTEXT_BROKER_PROMPT",
-    "DECODE_RETRIEVAL_WORKER_PROMPT",
     "DIRECTION_WORKER_PROMPT",
     "ELEMENT_SET_WORKER_PROMPT",
-    "EXTERNAL_RETRIEVAL_WORKER_PROMPT",
-    "KNOWLEDGE_RETRIEVAL_WORKER_PROMPT",
+    "EVIDENCE_RETRIEVAL_WORKER_PROMPT",
     "PARAGRAPH_WORKER_PROMPT",
-    "PATTERN_RETRIEVAL_WORKER_PROMPT",
     "PLANNER_PROMPT",
     "PORTFOLIO_WORKER_PROMPT",
     "RETRIEVAL_VALIDATOR_PROMPT",

script_build_host/src/script_build_host/agents/prompts/decode_retrieval_worker.md

@@ -1,13 +0,0 @@
-你只执行一个 Decode Case Retrieval TaskSpec。先读冻结输入，再且仅再调用一次
-`search_script_decode_case`；一次查询必须覆盖本 Task 的全部验收标准。成功后 Host 会冻结
-EvidenceRecordV1、提交 Attempt 并立即终止 Worker，不要自行调用 `submit_attempt`。
-
-根据真实缺口选择 `return_field`：主脉络用于结构参考，元素用于人物、物件、动作和场景，
-主题/形式/作用/感受用于相应写法参考。keyword 只能来自冻结输入和 Task objective；只有
-账号专属任务才传 account_name。案例是可追溯参考，不是可照抄模板：保留来源定位与原意，
-不要把多个案例拼成虚构案例，也不要在检索 Worker 内起草正文。
-
-`search_script_decode_case` 是上游检索，整个 Attempt 只能调用一次，不得第二次查询、扩展或验证。
-`read_mission_context` 的 cursor 分页只读已冻结输入，可按 Context Access Protocol 使用，
-不计作第二次 Decode 检索。没有命中可以是诚实结果。不要创建 Task、文件、命令、
-数据库查询或任意网络请求。

script_build_host/src/script_build_host/agents/prompts/evidence_retrieval_worker.md

@@ -0,0 +1,6 @@
+这是 logical evidence-retrieval 的冻结执行身份。实际 provider 选择、重试、降级、幂等租约和
+EvidenceRecord 提交全部由 Host 的 RetrievalPipeline 完成，不由模型选择 Knowledge、Pattern
+或 External/XHS，也不允许模型补写检索结果。
+
+如果该 preset 到达模型执行边界，说明 deterministic worker 装配失效；应当失败关闭，不能自行
+调用工具或提交替代产物。

script_build_host/src/script_build_host/agents/prompts/external_retrieval_worker.md

@@ -1,12 +0,0 @@
-你只执行一个 External Retrieval TaskSpec。先读冻结输入，再且仅再调用一次
-`external_search_case`；一次 keyword 必须覆盖本 Task 的全部验收标准。成功后 Host 会冻结
-EvidenceRecordV1、提交 Attempt 并立即终止 Worker，不要自行调用 `submit_attempt`。
-
-优先检索能补足脚本的信息：可核验事实、真实案例、受众语言、平台表达和必要的视觉线索。
-保留原始来源、平台标识和上下文，明确区分事实、来源观点和你的归纳，不把搜索摘要当事实，
-不虚构数字、人物或出处。外部案例用于证据和启发，不在本 Worker 中直接生成脚本文案。
-
-本角色的成功检索会原子冻结结果并立即结束 Attempt。`external_search_case` 是上游检索，
-整个 Attempt 只能调用一次，不得二次查询、扩展或验证。`read_mission_context` 的 cursor
-分页只读已冻结输入，可按 Context Access Protocol 使用，不计作第二次外部检索。
-没有命中可以是诚实结果。不要创建 Task、文件、命令、数据库查询或任意网络请求。

script_build_host/src/script_build_host/agents/prompts/knowledge_retrieval_worker.md

@@ -1,12 +0,0 @@
-你只执行一个 Knowledge Retrieval TaskSpec。先读冻结输入，再且仅再调用一次
-`search_knowledge`；一次 keyword 必须覆盖本 Task 的全部验收标准。成功后 Host 会冻结
-EvidenceRecordV1、提交 Attempt 并立即终止 Worker，不要自行调用 `submit_attempt`。
-
-检索创作方法、领域规则、评估标准或可执行步骤，用来回答“应该怎样做、为什么这样做”。
-保留知识条目的适用条件、限制和来源；不要把通用方法冒充当前主题事实，不要虚构示例，
-也不要在检索 Worker 内起草脚本文案。
-
-`search_knowledge` 是上游检索，整个 Attempt 只能调用一次，不得第二次查询、扩展或验证。
-`read_mission_context` 的 cursor 分页只读已冻结输入，可按 Context Access Protocol 使用，
-不计作第二次 Knowledge 检索。没有命中可以是诚实结果。不要创建 Task、文件、命令、
-数据库查询或任意网络请求。

script_build_host/src/script_build_host/agents/prompts/pattern_retrieval_worker.md

@@ -1,12 +0,0 @@
-你只执行一个 Pattern Retrieval TaskSpec。先读冻结输入，再且仅再调用一次
-`query_pattern_qa`；一次 message 必须覆盖本 Task 的全部验收标准。成功后 Host 会冻结
-EvidenceRecordV1、提交 Attempt 并立即终止 Worker，不要自行调用 `submit_attempt`。
-
-查询重点是可复核的模式与关系：共同出现的创作特征、支持度、置信度、提升度、样本数和
-代表实例。按 Task scope 区分账号内模式与跨账号模式，不把相关性写成因果，不改写网关
-返回的统计含义。检索结果只为后续 Direction 或创作提供证据，不直接起草脚本。
-
-`query_pattern_qa` 是上游检索，整个 Attempt 只能调用一次，不得第二次查询、扩展或验证。
-`read_mission_context` 的 cursor 分页只是读取已冻结输入，可按 Context Access Protocol 使用，
-不计作第二次 Pattern 检索。不得伪造 Pattern；没有命中可以是诚实结果。不要创建
-Task、文件、命令、数据库查询或任意网络请求。

script_build_host/src/script_build_host/application/direction_evidence.py

@@ -0,0 +1,197 @@
+"""Host-owned execution of one frozen Direction evidence manifest."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from typing import Any
+from uuid import uuid4
+
+from agent.orchestration import ArtifactRef
+
+from script_build_host.application.retrieval_pipeline import (
+    EvidenceRetrievalProvider,
+    ProviderRetrievalResult,
+    RetrievalPipeline,
+)
+from script_build_host.domain.artifacts import EvidenceRecordV1
+from script_build_host.domain.evidence_requirements import (
+    EvidenceProvider,
+    EvidenceRequirementV1,
+)
+from script_build_host.domain.ports import (
+    InputSnapshotRepository,
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.domain.task_contracts import ScriptTaskContractV2
+from script_build_host.infrastructure.retrieval_journal import (
+    LedgerRetrievalLeaseRepository,
+)
+
+
+class _ProviderAdapter(EvidenceRetrievalProvider):
+    def __init__(self, adapter: Any, provider: EvidenceProvider) -> None:
+        self._adapter = adapter
+        self._provider = provider
+
+    async def retrieve(
+        self, *, query: Mapping[str, Any], snapshot: Any
+    ) -> ProviderRetrievalResult:
+        effective = dict(query)
+        if self._provider is EvidenceProvider.EXTERNAL_XHS:
+            effective["platforms"] = ["xhs"]
+        value = await self._adapter.retrieve(query=effective, snapshot=snapshot)
+        metadata = dict(value.metadata or {})
+        return ProviderRetrievalResult(
+            source_refs=tuple(value.source_refs),
+            summary=str(value.summary),
+            concepts=tuple(value.supports),
+            limitations=tuple(value.limitations),
+            partial=bool(metadata.get("partial_result")),
+            metadata=metadata,
+        )
+
+
+class DirectionEvidenceOrchestrator:
+    def __init__(
+        self,
+        *,
+        coordinator: Any,
+        bindings: MissionBindingRepository,
+        input_snapshots: InputSnapshotRepository,
+        artifacts: ScriptBusinessArtifactRepository,
+        retrieval_adapters: Mapping[str, Any],
+        owner: str = "script-build-host",
+    ) -> None:
+        self._coordinator = coordinator
+        self._bindings = bindings
+        self._snapshots = input_snapshots
+        self._artifacts = artifacts
+        self._adapters = dict(retrieval_adapters)
+        self._owner = owner
+
+    async def execute(
+        self,
+        *,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: str,
+        spec_version: int,
+        contract: ScriptTaskContractV2,
+    ) -> ArtifactRef:
+        binding = await self._bindings.get_by_root(root_trace_id)
+        snapshot = await self._snapshots.get(
+            str(binding.input_snapshot_id),
+            script_build_id=binding.script_build_id,
+        )
+        providers = self._providers()
+        pipeline = RetrievalPipeline(
+            providers,
+            LedgerRetrievalLeaseRepository(self._coordinator, root_trace_id),
+            owner=f"{self._owner}:{task_id}:{attempt_id}",
+        )
+        results = []
+        for requirement in contract.evidence_requirements.requirements:
+            if requirement.satisfied_by_input_ref is not None:
+                continue
+            results.append(
+                await pipeline.require(
+                    requirement,
+                    query=self._query(requirement, snapshot),
+                    snapshot=snapshot,
+                )
+            )
+        source_refs = tuple(
+            dict.fromkeys(
+                item.evidence_ref
+                for result in results
+                for item in result.evidence_items
+            )
+        )
+        metadata = {
+            "requirements": [
+                {
+                    "requirement_id": result.requirement_id,
+                    "status": result.status.value,
+                    "provider": result.provider.value if result.provider else None,
+                    "evidence_items": [
+                        {
+                            "evidence_ref": item.evidence_ref,
+                            "concepts": list(item.concepts),
+                        }
+                        for item in result.evidence_items
+                    ],
+                }
+                for result in results
+            ]
+        }
+        record = EvidenceRecordV1(
+            evidence_id=str(uuid4()),
+            source_type="logical_direction_evidence",
+            tool_name="retrieval_pipeline",
+            query={"_retrieval_metadata": metadata},
+            source_refs=source_refs,
+            raw_artifact_ref=None,
+            summary="; ".join(result.summary for result in results),
+            supports=tuple(
+                dict.fromkeys(
+                    concept
+                    for result in results
+                    for item in result.evidence_items
+                    for concept in item.concepts
+                )
+            ),
+            confidence=(
+                "partial"
+                if any(
+                    result.status.value == "partial_success"
+                    for result in results
+                )
+                else "verified"
+            ),
+            limitations=tuple(
+                limitation for result in results for limitation in result.limitations
+            ),
+            content_sha256="",
+            created_at=datetime.now(UTC),
+        )
+        _, ref = await self._artifacts.freeze(
+            script_build_id=binding.script_build_id,
+            task_id=task_id,
+            attempt_id=attempt_id,
+            spec_version=spec_version,
+            artifact=record,
+        )
+        return ref
+
+    def _providers(self) -> dict[EvidenceProvider, EvidenceRetrievalProvider]:
+        names = {
+            EvidenceProvider.KNOWLEDGE: "knowledge",
+            EvidenceProvider.EXTERNAL_XHS: "external",
+            EvidenceProvider.PATTERN: "pattern",
+            EvidenceProvider.EXTERNAL: "external",
+        }
+        return {
+            provider: _ProviderAdapter(self._adapters[name], provider)
+            for provider, name in names.items()
+            if name in self._adapters
+        }
+
+    @staticmethod
+    def _query(requirement: EvidenceRequirementV1, snapshot: Any) -> dict[str, Any]:
+        return {
+            "requirement_id": requirement.requirement_id,
+            "purpose": requirement.purpose.value,
+            "account_name": str(
+                snapshot.account.get("account_name")
+                or snapshot.account.get("name")
+                or snapshot.account.get("nickname")
+                or ""
+            ),
+            "topic": dict(snapshot.topic),
+            "required_concepts": list(requirement.required_concepts),
+        }
+
+
+__all__ = ["DirectionEvidenceOrchestrator"]

script_build_host/src/script_build_host/application/direction_evidence_gate.py

@@ -0,0 +1,102 @@
+"""The single evidence-closure adapter used at all Direction gates."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from agent.orchestration import TaskStatus
+
+from script_build_host.domain.artifacts import EvidenceRecordV1
+from script_build_host.domain.evidence_requirements import (
+    EvidenceClosure,
+    EvidenceClosureEvaluator,
+    EvidenceGate,
+    EvidenceItem,
+    EvidenceProvider,
+)
+from script_build_host.domain.task_contracts import (
+    ScriptTaskContractV2,
+    ScriptTaskKind,
+    task_kind_from_context_refs,
+)
+
+
+class DirectionEvidenceGate:
+    def __init__(self, artifacts: Any) -> None:
+        self._artifacts = artifacts
+        self._evaluator = EvidenceClosureEvaluator()
+
+    async def require(
+        self,
+        *,
+        script_build_id: int,
+        ledger: Any,
+        direction_task: Any,
+        contract: ScriptTaskContractV2,
+        gate: EvidenceGate,
+    ) -> EvidenceClosure:
+        items: list[EvidenceItem] = []
+        for child_id in direction_task.child_task_ids:
+            child = ledger.tasks.get(child_id)
+            if (
+                child is None
+                or child.status is not TaskStatus.COMPLETED
+                or task_kind_from_context_refs(child.current_spec.context_refs)
+                != ScriptTaskKind.EVIDENCE_RETRIEVAL.value
+            ):
+                continue
+            accepted = (
+                ledger.decisions.get(child.decision_ids[-1])
+                if child.decision_ids
+                else None
+            )
+            if accepted is None or accepted.action.value != "accept":
+                continue
+            attempt = (
+                ledger.attempts.get(accepted.attempt_id or "")
+                if accepted is not None
+                else None
+            )
+            if (
+                attempt is None
+                or attempt.spec_version != child.current_spec_version
+                or attempt.submission is None
+            ):
+                continue
+            for ref in attempt.submission.evidence_refs:
+                version = await self._artifacts.read_by_ref(
+                    ref,
+                    script_build_id=script_build_id,
+                    task_id=child.task_id,
+                    attempt_id=attempt.attempt_id,
+                )
+                if not isinstance(version.artifact, EvidenceRecordV1):
+                    continue
+                metadata = version.artifact.query.get("_retrieval_metadata", {})
+                for row in metadata.get("requirements", ()):
+                    provider = row.get("provider")
+                    if not provider:
+                        continue
+                    for evidence in row.get("evidence_items", ()):
+                        evidence_ref = str(evidence.get("evidence_ref", "")).strip()
+                        if not evidence_ref:
+                            continue
+                        items.append(
+                            EvidenceItem(
+                                str(row.get("requirement_id", "")),
+                                evidence_ref,
+                                EvidenceProvider(str(provider)),
+                                tuple(
+                                    str(item)
+                                    for item in evidence.get("concepts", ())
+                                ),
+                            )
+                        )
+        return self._evaluator.require_closed(
+            contract.evidence_requirements,
+            tuple(items),
+            gate=gate,
+        )
+
+
+__all__ = ["DirectionEvidenceGate"]

script_build_host/src/script_build_host/application/retrieval_pipeline.py

@@ -0,0 +1,460 @@
+"""Deterministic evidence fallback guarded by a durable lease."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any, Protocol
+
+from script_build_host.domain.canonical_json import canonical_sha256
+from script_build_host.domain.errors import (
+    ProtocolViolation,
+    RequiredEvidenceUnsatisfied,
+    ScriptBuildError,
+    UpstreamFailure,
+)
+from script_build_host.domain.evidence_requirements import (
+    EvidenceItem,
+    EvidenceProvider,
+    EvidenceRequirementV1,
+)
+
+
+class RetrievalStatus(StrEnum):
+    SATISFIED_BY_INPUT = "satisfied_by_input"
+    SUCCESS = "success"
+    PARTIAL_SUCCESS = "partial_success"
+    UNSATISFIED = "unsatisfied"
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderRetrievalResult:
+    source_refs: tuple[str, ...]
+    summary: str
+    concepts: tuple[str, ...] = ()
+    limitations: tuple[str, ...] = ()
+    partial: bool = False
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if any(
+            not isinstance(item, str) or not item.strip() for item in self.source_refs
+        ):
+            raise ProtocolViolation("provider returned an empty evidence reference")
+
+
+@dataclass(frozen=True, slots=True)
+class ProviderAttempt:
+    provider: EvidenceProvider
+    number: int
+    outcome: str
+    error_code: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class RetrievalPipelineResult:
+    requirement_id: str
+    status: RetrievalStatus
+    provider: EvidenceProvider | None
+    evidence_items: tuple[EvidenceItem, ...]
+    attempts: tuple[ProviderAttempt, ...]
+    summary: str = ""
+    limitations: tuple[str, ...] = ()
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "requirement_id": self.requirement_id,
+            "status": self.status.value,
+            "provider": self.provider.value if self.provider else None,
+            "evidence_items": [
+                {
+                    "requirement_id": item.requirement_id,
+                    "evidence_ref": item.evidence_ref,
+                    "provider": item.provider.value,
+                    "concepts": list(item.concepts),
+                }
+                for item in self.evidence_items
+            ],
+            "attempts": [
+                {
+                    "provider": item.provider.value,
+                    "number": item.number,
+                    "outcome": item.outcome,
+                    "error_code": item.error_code,
+                }
+                for item in self.attempts
+            ],
+            "summary": self.summary,
+            "limitations": list(self.limitations),
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> RetrievalPipelineResult:
+        return cls(
+            requirement_id=str(value["requirement_id"]),
+            status=RetrievalStatus(str(value["status"])),
+            provider=(
+                EvidenceProvider(str(value["provider"]))
+                if value.get("provider")
+                else None
+            ),
+            evidence_items=tuple(
+                EvidenceItem(
+                    str(item["requirement_id"]),
+                    _required_evidence_ref(item.get("evidence_ref")),
+                    EvidenceProvider(str(item["provider"])),
+                    tuple(str(concept) for concept in item.get("concepts", ())),
+                )
+                for item in value.get("evidence_items", ())
+            ),
+            attempts=tuple(
+                ProviderAttempt(
+                    EvidenceProvider(str(item["provider"])),
+                    int(item["number"]),
+                    str(item["outcome"]),
+                    str(item["error_code"]) if item.get("error_code") else None,
+                )
+                for item in value.get("attempts", ())
+            ),
+            summary=str(value.get("summary", "")),
+            limitations=tuple(str(item) for item in value.get("limitations", ())),
+        )
+
+
+def _required_evidence_ref(value: Any) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ProtocolViolation("persisted retrieval result has no evidence reference")
+    return value
+
+
+@dataclass(frozen=True, slots=True)
+class RetrievalLease:
+    retrieval_key: str
+    owner: str
+    epoch: int
+    acquired: bool
+    terminal: RetrievalPipelineResult | None = None
+    journal: tuple[Mapping[str, Any], ...] = ()
+
+
+class EvidenceRetrievalProvider(Protocol):
+    async def retrieve(
+        self, *, query: Mapping[str, Any], snapshot: Any
+    ) -> ProviderRetrievalResult: ...
+
+
+class RetrievalLeaseRepository(Protocol):
+    async def reserve(
+        self, retrieval_key: str, *, owner: str, lease_seconds: int
+    ) -> RetrievalLease: ...
+
+    async def append_attempt(
+        self, lease: RetrievalLease, attempt: ProviderAttempt
+    ) -> None: ...
+
+    async def append_provider_result(
+        self,
+        lease: RetrievalLease,
+        provider: EvidenceProvider,
+        number: int,
+        result: ProviderRetrievalResult,
+    ) -> None: ...
+
+    async def renew(self, lease: RetrievalLease, *, lease_seconds: int) -> None: ...
+
+    async def complete(
+        self, lease: RetrievalLease, result: RetrievalPipelineResult
+    ) -> RetrievalPipelineResult: ...
+
+
+class RetrievalPipeline:
+    def __init__(
+        self,
+        providers: Mapping[EvidenceProvider, EvidenceRetrievalProvider],
+        repository: RetrievalLeaseRepository,
+        *,
+        owner: str,
+        max_transient_attempts: int = 3,
+        lease_seconds: int = 60,
+        retry_backoff_seconds: float = 0.05,
+        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    ) -> None:
+        if not owner.strip() or max_transient_attempts < 1 or lease_seconds < 1:
+            raise ValueError("retrieval owner, attempts and lease must be valid")
+        self.providers = dict(providers)
+        self.repository = repository
+        self.owner = owner
+        self.max_transient_attempts = max_transient_attempts
+        self.lease_seconds = lease_seconds
+        self.retry_backoff_seconds = retry_backoff_seconds
+        self.sleep = sleep
+
+    async def execute(
+        self,
+        requirement: EvidenceRequirementV1,
+        *,
+        query: Mapping[str, Any],
+        snapshot: Any,
+    ) -> RetrievalPipelineResult:
+        if requirement.satisfied_by_input_ref is not None:
+            return RetrievalPipelineResult(
+                requirement.requirement_id,
+                RetrievalStatus.SATISFIED_BY_INPUT,
+                EvidenceProvider.INPUT,
+                (),
+                (),
+                requirement.satisfied_by_input_ref,
+            )
+        key = canonical_sha256(
+            {"requirement": requirement.to_payload(), "query": dict(query)}
+        ).wire
+        lease = await self.repository.reserve(
+            key, owner=self.owner, lease_seconds=self.lease_seconds
+        )
+        if lease.terminal is not None:
+            return lease.terminal
+        if not lease.acquired:
+            raise ProtocolViolation("RETRIEVAL_RESERVATION_BUSY")
+        result = await self._run(lease, requirement, query=query, snapshot=snapshot)
+        return await self.repository.complete(lease, result)
+
+    async def require(
+        self,
+        requirement: EvidenceRequirementV1,
+        *,
+        query: Mapping[str, Any],
+        snapshot: Any,
+    ) -> RetrievalPipelineResult:
+        result = await self.execute(requirement, query=query, snapshot=snapshot)
+        if requirement.hard and result.status is RetrievalStatus.UNSATISFIED:
+            raise RequiredEvidenceUnsatisfied(
+                f"provider chain exhausted for {requirement.requirement_id}"
+            )
+        return result
+
+    async def _run(
+        self,
+        lease: RetrievalLease,
+        requirement: EvidenceRequirementV1,
+        *,
+        query: Mapping[str, Any],
+        snapshot: Any,
+    ) -> RetrievalPipelineResult:
+        attempts = _journal_attempts(lease.journal)
+        limitations: list[str] = []
+        for provider_name in requirement.source_chain:
+            provider = self.providers.get(provider_name)
+            prior = [item for item in attempts if item.provider is provider_name]
+            cached = _journal_result(lease.journal, provider_name)
+            if cached is not None:
+                cached_response, number = cached
+                attempts.append(
+                    ProviderAttempt(
+                        provider_name,
+                        number,
+                        "result" if cached_response.source_refs else "empty",
+                    )
+                )
+            elif prior and prior[-1].outcome in {
+                "permanent_failure",
+                "empty",
+                "result",
+            }:
+                continue
+            if provider is None:
+                if prior:
+                    continue
+                attempt = ProviderAttempt(
+                    provider_name, 1, "permanent_failure", "PROVIDER_NOT_CONFIGURED"
+                )
+                attempts.append(attempt)
+                await self.repository.append_attempt(lease, attempt)
+                continue
+            response: ProviderRetrievalResult | None = (
+                cached[0] if cached is not None else None
+            )
+            start = 1 + sum(
+                item.outcome == "transient_failure" for item in prior
+            )
+            for number in (
+                ()
+                if cached is not None
+                else range(start, self.max_transient_attempts + 1)
+            ):
+                try:
+                    response = await self._retrieve_with_lease(
+                        lease,
+                        provider,
+                        query={
+                            **dict(query),
+                            "_retrieval_request_id": canonical_sha256(
+                                {
+                                    "retrieval_key": lease.retrieval_key,
+                                    "provider": provider_name.value,
+                                    "number": number,
+                                }
+                            ).wire,
+                        },
+                        snapshot=snapshot,
+                    )
+                    await self.repository.append_provider_result(
+                        lease, provider_name, number, response
+                    )
+                except UpstreamFailure as error:
+                    outcome = "transient_failure" if error.retryable else "permanent_failure"
+                    attempt = ProviderAttempt(provider_name, number, outcome, error.code)
+                    attempts.append(attempt)
+                    await self.repository.append_attempt(lease, attempt)
+                    limitations.append(f"{provider_name.value}:{error.code}")
+                    if not error.retryable or number == self.max_transient_attempts:
+                        break
+                    delay = error.retry_after_seconds
+                    if delay is None:
+                        delay = self.retry_backoff_seconds * (2 ** (number - 1))
+                    if delay:
+                        await self.sleep(delay)
+                    continue
+                except ScriptBuildError as error:
+                    attempt = ProviderAttempt(
+                        provider_name, number, "permanent_failure", error.code
+                    )
+                    attempts.append(attempt)
+                    await self.repository.append_attempt(lease, attempt)
+                    break
+                attempt = ProviderAttempt(
+                    provider_name, number, "result" if response.source_refs else "empty"
+                )
+                attempts.append(attempt)
+                await self.repository.append_attempt(lease, attempt)
+                break
+            if response is None or not response.source_refs:
+                continue
+            refs = tuple(dict.fromkeys(response.source_refs))
+            if len(refs) < requirement.minimum_items:
+                limitations.extend(response.limitations)
+                continue
+            if not set(requirement.required_concepts) <= set(response.concepts):
+                limitations.extend(response.limitations)
+                continue
+            items = tuple(
+                EvidenceItem(
+                    requirement.requirement_id,
+                    ref,
+                    provider_name,
+                    response.concepts,
+                )
+                for ref in refs
+            )
+            return RetrievalPipelineResult(
+                requirement.requirement_id,
+                (
+                    RetrievalStatus.PARTIAL_SUCCESS
+                    if response.partial or response.metadata.get("partial_result")
+                    else RetrievalStatus.SUCCESS
+                ),
+                provider_name,
+                items,
+                tuple(attempts),
+                response.summary,
+                tuple((*limitations, *response.limitations)),
+            )
+        return RetrievalPipelineResult(
+            requirement.requirement_id,
+            RetrievalStatus.UNSATISFIED,
+            None,
+            (),
+            tuple(attempts),
+            limitations=tuple(limitations),
+        )
+
+    async def _retrieve_with_lease(
+        self,
+        lease: RetrievalLease,
+        provider: EvidenceRetrievalProvider,
+        *,
+        query: Mapping[str, Any],
+        snapshot: Any,
+    ) -> ProviderRetrievalResult:
+        task = asyncio.create_task(provider.retrieve(query=query, snapshot=snapshot))
+        heartbeat_seconds = max(0.5, self.lease_seconds / 3)
+        try:
+            while True:
+                done, _ = await asyncio.wait({task}, timeout=heartbeat_seconds)
+                if done:
+                    return task.result()
+                await self.repository.renew(
+                    lease,
+                    lease_seconds=self.lease_seconds,
+                )
+        finally:
+            if not task.done():
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+
+
+def _journal_attempts(
+    journal: tuple[Mapping[str, Any], ...],
+) -> list[ProviderAttempt]:
+    attempts: list[ProviderAttempt] = []
+    for event in journal:
+        if event.get("event_type", "attempt") != "attempt":
+            continue
+        try:
+            attempts.append(
+                ProviderAttempt(
+                    EvidenceProvider(str(event["provider"])),
+                    int(event["number"]),
+                    str(event["outcome"]),
+                    (
+                        str(event["error_code"])
+                        if event.get("error_code")
+                        else None
+                    ),
+                )
+            )
+        except (KeyError, TypeError, ValueError):
+            continue
+    return attempts
+
+
+def _journal_result(
+    journal: tuple[Mapping[str, Any], ...],
+    provider: EvidenceProvider,
+) -> tuple[ProviderRetrievalResult, int] | None:
+    for event in reversed(journal):
+        if (
+            event.get("event_type") != "provider_result"
+            or event.get("provider") != provider.value
+            or not isinstance(event.get("result"), Mapping)
+        ):
+            continue
+        value = event["result"]
+        return (
+            ProviderRetrievalResult(
+                tuple(
+                    _required_evidence_ref(item)
+                    for item in value.get("source_refs", ())
+                ),
+                str(value.get("summary", "")),
+                tuple(str(item) for item in value.get("concepts", ())),
+                tuple(str(item) for item in value.get("limitations", ())),
+                bool(value.get("partial", False)),
+                dict(value.get("metadata") or {}),
+            ),
+            int(event.get("number", 1)),
+        )
+    return None
+
+
+__all__ = [
+    "EvidenceRetrievalProvider",
+    "ProviderAttempt",
+    "ProviderRetrievalResult",
+    "RetrievalLease",
+    "RetrievalLeaseRepository",
+    "RetrievalPipeline",
+    "RetrievalPipelineResult",
+    "RetrievalStatus",
+]

script_build_host/src/script_build_host/application/retrieval_ports.py

@@ -0,0 +1,30 @@
+"""Provider-facing retrieval DTOs; Agent tools never expose these adapters."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+
+@dataclass(frozen=True, slots=True)
+class RetrievalResult:
+    source_refs: tuple[str, ...]
+    summary: str
+    supports: tuple[str, ...] = ()
+    confidence: str = "unknown"
+    limitations: tuple[str, ...] = ()
+    raw_artifact_ref: str | None = None
+    metadata: Mapping[str, Any] | None = None
+
+
+class RetrievalAdapter(Protocol):
+    async def retrieve(
+        self,
+        *,
+        query: Mapping[str, Any],
+        snapshot: Any,
+    ) -> RetrievalResult: ...
+
+
+__all__ = ["RetrievalAdapter", "RetrievalResult"]

script_build_host/src/script_build_host/domain/evidence_requirements.py

@@ -0,0 +1,293 @@
+"""Frozen evidence requirements and the single closure evaluator."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+from .canonical_json import canonical_sha256
+from .errors import ProtocolViolation, RequiredEvidenceUnsatisfied
+
+
+class EvidencePurpose(StrEnum):
+    TOPIC = "topic"
+    PERSONA = "persona"
+    SECTION_PATTERN = "section_pattern"
+    FACTUAL_SUPPORT = "factual_support"
+
+
+class EvidenceProvider(StrEnum):
+    INPUT = "input"
+    KNOWLEDGE = "knowledge"
+    EXTERNAL_XHS = "external_xhs"
+    PATTERN = "pattern"
+    EXTERNAL = "external"
+
+
+class EvidenceGate(StrEnum):
+    DIRECTION_DISPATCH = "direction_dispatch"
+    DIRECTION_ACCEPT = "direction_accept"
+    PHASE_ONE_BOUNDARY = "phase_one_boundary"
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceRequirementV1:
+    requirement_id: str
+    purpose: EvidencePurpose
+    hard: bool
+    minimum_items: int
+    required_concepts: tuple[str, ...] = ()
+    source_chain: tuple[EvidenceProvider, ...] = ()
+    satisfied_by_input_ref: str | None = None
+
+    def __post_init__(self) -> None:
+        if not self.requirement_id.strip() or self.minimum_items < 1:
+            raise ProtocolViolation("evidence requirement identity and minimum must be valid")
+        if len(set(self.required_concepts)) != len(self.required_concepts):
+            raise ProtocolViolation("evidence concepts must be unique")
+        if len(set(self.source_chain)) != len(self.source_chain):
+            raise ProtocolViolation("evidence provider chain must be unique")
+        if self.satisfied_by_input_ref:
+            if self.source_chain:
+                raise ProtocolViolation("input-satisfied evidence cannot retain providers")
+        elif not self.source_chain:
+            raise ProtocolViolation("unsatisfied evidence requires a provider chain")
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "requirement_id": self.requirement_id,
+            "purpose": self.purpose.value,
+            "hard": self.hard,
+            "minimum_items": self.minimum_items,
+            "required_concepts": list(self.required_concepts),
+            "source_chain": [item.value for item in self.source_chain],
+            "satisfied_by_input_ref": self.satisfied_by_input_ref,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Any) -> EvidenceRequirementV1:
+        if not isinstance(value, dict):
+            raise ProtocolViolation("evidence requirement must be an object")
+        try:
+            return cls(
+                requirement_id=str(value["requirement_id"]),
+                purpose=EvidencePurpose(value["purpose"]),
+                hard=_boolean(value["hard"], "evidence hard"),
+                minimum_items=_integer(value["minimum_items"], "evidence minimum"),
+                required_concepts=_strings(value.get("required_concepts", ())),
+                source_chain=tuple(
+                    EvidenceProvider(item) for item in _strings(value.get("source_chain", ()))
+                ),
+                satisfied_by_input_ref=(
+                    str(value["satisfied_by_input_ref"])
+                    if value.get("satisfied_by_input_ref") is not None
+                    else None
+                ),
+            )
+        except (KeyError, ValueError) as exc:
+            raise ProtocolViolation("invalid evidence requirement") from exc
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceRequirementManifestV1:
+    requirements: tuple[EvidenceRequirementV1, ...]
+    schema_version: str = "evidence-requirements/v1"
+
+    def __post_init__(self) -> None:
+        if self.schema_version != "evidence-requirements/v1":
+            raise ProtocolViolation("unsupported evidence manifest schema")
+        identities = tuple(item.requirement_id for item in self.requirements)
+        if len(identities) != len(set(identities)):
+            raise ProtocolViolation("evidence requirement ids must be unique")
+
+    @property
+    def digest(self) -> str:
+        return canonical_sha256(self.to_payload()).wire
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": self.schema_version,
+            "requirements": [item.to_payload() for item in self.requirements],
+        }
+
+    @classmethod
+    def from_payload(cls, value: Any) -> EvidenceRequirementManifestV1:
+        if not isinstance(value, dict) or value.get("schema_version") != "evidence-requirements/v1":
+            raise ProtocolViolation("invalid evidence manifest")
+        rows = value.get("requirements")
+        if not isinstance(rows, list):
+            raise ProtocolViolation("evidence requirements must be an array")
+        return cls(tuple(EvidenceRequirementV1.from_payload(item) for item in rows))
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceItem:
+    requirement_id: str
+    evidence_ref: str
+    provider: EvidenceProvider
+    concepts: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.requirement_id.strip() or not self.evidence_ref.strip():
+            raise ProtocolViolation("evidence item identity and reference must be non-empty")
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceClosure:
+    gate: EvidenceGate
+    satisfied_requirement_ids: tuple[str, ...]
+    unsatisfied_hard_requirement_ids: tuple[str, ...]
+    item_counts: dict[str, int]
+
+    @property
+    def allowed(self) -> bool:
+        return not self.unsatisfied_hard_requirement_ids
+
+
+class EvidenceClosureEvaluator:
+    def evaluate(
+        self,
+        manifest: EvidenceRequirementManifestV1,
+        items: tuple[EvidenceItem, ...],
+        *,
+        gate: EvidenceGate,
+    ) -> EvidenceClosure:
+        known = {item.requirement_id: item for item in manifest.requirements}
+        refs = {key: set[str]() for key in known}
+        concepts = {key: set[str]() for key in known}
+        for item in items:
+            requirement = known.get(item.requirement_id)
+            if (
+                requirement is None
+                or not item.evidence_ref.strip()
+                or item.provider not in requirement.source_chain
+            ):
+                continue
+            refs[item.requirement_id].add(item.evidence_ref)
+            concepts[item.requirement_id].update(item.concepts)
+        satisfied: list[str] = []
+        blocked: list[str] = []
+        for requirement in manifest.requirements:
+            complete = bool(requirement.satisfied_by_input_ref) or (
+                len(refs[requirement.requirement_id]) >= requirement.minimum_items
+                and set(requirement.required_concepts)
+                <= concepts[requirement.requirement_id]
+            )
+            if complete:
+                satisfied.append(requirement.requirement_id)
+            elif requirement.hard:
+                blocked.append(requirement.requirement_id)
+        return EvidenceClosure(
+            gate,
+            tuple(satisfied),
+            tuple(blocked),
+            {key: len(value) for key, value in refs.items()},
+        )
+
+    def require_closed(
+        self,
+        manifest: EvidenceRequirementManifestV1,
+        items: tuple[EvidenceItem, ...],
+        *,
+        gate: EvidenceGate,
+    ) -> EvidenceClosure:
+        closure = self.evaluate(manifest, items, gate=gate)
+        if not closure.allowed:
+            raise RequiredEvidenceUnsatisfied(
+                f"hard evidence is unsatisfied at {gate.value}: "
+                + ",".join(closure.unsatisfied_hard_requirement_ids)
+            )
+        return closure
+
+
+class DirectionEvidenceRequirementFactory:
+    def compile(
+        self,
+        *,
+        topic_input_ref: str,
+        requested_account_name: str | None,
+        persona_points: tuple[object, ...],
+        persona_input_ref: str | None,
+        section_patterns: tuple[object, ...],
+        section_pattern_input_ref: str | None,
+        factual_support_requested: bool = False,
+        factual_required_concepts: tuple[str, ...] = (),
+    ) -> EvidenceRequirementManifestV1:
+        requirements = [
+            EvidenceRequirementV1(
+                "direction.topic",
+                EvidencePurpose.TOPIC,
+                True,
+                1,
+                satisfied_by_input_ref=topic_input_ref,
+            )
+        ]
+        if (requested_account_name or "").strip():
+            requirements.append(
+                EvidenceRequirementV1(
+                    "direction.persona",
+                    EvidencePurpose.PERSONA,
+                    True,
+                    1,
+                    source_chain=(
+                        ()
+                        if persona_points
+                        else (EvidenceProvider.KNOWLEDGE, EvidenceProvider.EXTERNAL_XHS)
+                    ),
+                    satisfied_by_input_ref=persona_input_ref if persona_points else None,
+                )
+            )
+        requirements.append(
+            EvidenceRequirementV1(
+                "direction.section_pattern",
+                EvidencePurpose.SECTION_PATTERN,
+                True,
+                1,
+                source_chain=() if section_patterns else (EvidenceProvider.PATTERN,),
+                satisfied_by_input_ref=section_pattern_input_ref if section_patterns else None,
+            )
+        )
+        if factual_support_requested:
+            requirements.append(
+                EvidenceRequirementV1(
+                    "direction.factual_support",
+                    EvidencePurpose.FACTUAL_SUPPORT,
+                    True,
+                    1,
+                    factual_required_concepts,
+                    (EvidenceProvider.KNOWLEDGE, EvidenceProvider.EXTERNAL),
+                )
+            )
+        return EvidenceRequirementManifestV1(tuple(requirements))
+
+
+def _strings(value: Any) -> tuple[str, ...]:
+    if not isinstance(value, (list, tuple)) or any(not isinstance(item, str) for item in value):
+        raise ProtocolViolation("evidence string collection is invalid")
+    return tuple(value)
+
+
+def _boolean(value: Any, label: str) -> bool:
+    if not isinstance(value, bool):
+        raise ProtocolViolation(f"{label} must be boolean")
+    return value
+
+
+def _integer(value: Any, label: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ProtocolViolation(f"{label} must be integer")
+    return value
+
+
+__all__ = [
+    "DirectionEvidenceRequirementFactory",
+    "EvidenceClosure",
+    "EvidenceClosureEvaluator",
+    "EvidenceGate",
+    "EvidenceItem",
+    "EvidenceProvider",
+    "EvidencePurpose",
+    "EvidenceRequirementManifestV1",
+    "EvidenceRequirementV1",
+]

script_build_host/src/script_build_host/domain/phase_protocol.py

@@ -6,8 +6,10 @@ from collections.abc import Mapping
 from dataclasses import asdict, dataclass, field
 from typing import Any
 
-from .task_capabilities import task_capability
-from .task_contracts import ScriptTaskKind
+from .task_capabilities import TaskCapabilityCatalog
+from .task_contracts import ScriptTaskKind, TaskContractError
+
+_CAPABILITIES = TaskCapabilityCatalog()
 
 
 @dataclass(frozen=True, slots=True)
@@ -41,21 +43,13 @@ class PhaseProtocolSnapshot:
 def allowed_child_kinds(
     *, phase: int, parent_kind: ScriptTaskKind | None
 ) -> frozenset[ScriptTaskKind]:
-    if phase == 1:
-        if parent_kind is None:
-            return frozenset({ScriptTaskKind.DIRECTION})
-        if parent_kind is ScriptTaskKind.DIRECTION:
-            return task_capability(parent_kind).child_kinds
+    if parent_kind is None:
+        return _CAPABILITIES.root_kinds(phase)
+    try:
+        parent = _CAPABILITIES.get(parent_kind)
+    except TaskContractError:
         return frozenset()
-    if phase == 2:
-        return (
-            frozenset({ScriptTaskKind.CANDIDATE_PORTFOLIO})
-            if parent_kind is None
-            else task_capability(parent_kind).child_kinds
-        )
-    if phase == 3:
-        return frozenset({ScriptTaskKind.ROOT_DELIVERY}) if parent_kind is None else frozenset()
-    return frozenset()
+    return parent.child_kinds if parent.phase == phase else frozenset()
 
 
 __all__ = [

script_build_host/src/script_build_host/infrastructure/retrieval_journal.py

@@ -0,0 +1,158 @@
+"""Durable retrieval reservation backed by the Agent Ledger CAS."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, Protocol
+
+from script_build_host.application.retrieval_pipeline import (
+    ProviderAttempt,
+    ProviderRetrievalResult,
+    RetrievalLease,
+    RetrievalPipelineResult,
+)
+from script_build_host.domain.evidence_requirements import EvidenceProvider
+
+
+class DurableLeaseCoordinator(Protocol):
+    async def reserve_durable_lease(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        lease_seconds: int,
+    ) -> Mapping[str, Any]: ...
+
+    async def append_durable_lease_event(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        epoch: int,
+        event: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+    async def renew_durable_lease(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        epoch: int,
+        lease_seconds: int,
+    ) -> Mapping[str, Any]: ...
+
+    async def complete_durable_lease(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        epoch: int,
+        result: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+
+class LedgerRetrievalLeaseRepository:
+    def __init__(
+        self, coordinator: DurableLeaseCoordinator, root_trace_id: str
+    ) -> None:
+        self._coordinator = coordinator
+        self._root_trace_id = root_trace_id
+
+    async def reserve(
+        self, retrieval_key: str, *, owner: str, lease_seconds: int
+    ) -> RetrievalLease:
+        value = await self._coordinator.reserve_durable_lease(
+            self._root_trace_id,
+            retrieval_key,
+            owner=owner,
+            lease_seconds=lease_seconds,
+        )
+        terminal = value.get("terminal_result")
+        return RetrievalLease(
+            retrieval_key,
+            str(value["owner"]),
+            int(value["epoch"]),
+            bool(value["acquired"]),
+            (
+                RetrievalPipelineResult.from_payload(terminal)
+                if isinstance(terminal, dict)
+                else None
+            ),
+            tuple(
+                dict(item)
+                for item in value.get("journal", ())
+                if isinstance(item, Mapping)
+            ),
+        )
+
+    async def append_attempt(
+        self, lease: RetrievalLease, attempt: ProviderAttempt
+    ) -> None:
+        await self._coordinator.append_durable_lease_event(
+            self._root_trace_id,
+            lease.retrieval_key,
+            owner=lease.owner,
+            epoch=lease.epoch,
+            event={
+                "event_type": "attempt",
+                "provider": attempt.provider.value,
+                "number": attempt.number,
+                "outcome": attempt.outcome,
+                "error_code": attempt.error_code,
+            },
+        )
+
+    async def append_provider_result(
+        self,
+        lease: RetrievalLease,
+        provider: EvidenceProvider,
+        number: int,
+        result: ProviderRetrievalResult,
+    ) -> None:
+        await self._coordinator.append_durable_lease_event(
+            self._root_trace_id,
+            lease.retrieval_key,
+            owner=lease.owner,
+            epoch=lease.epoch,
+            event={
+                "event_type": "provider_result",
+                "provider": provider.value,
+                "number": number,
+                "result": {
+                    "source_refs": list(result.source_refs),
+                    "summary": result.summary,
+                    "concepts": list(result.concepts),
+                    "limitations": list(result.limitations),
+                    "partial": result.partial,
+                    "metadata": dict(result.metadata),
+                },
+            },
+        )
+
+    async def renew(self, lease: RetrievalLease, *, lease_seconds: int) -> None:
+        await self._coordinator.renew_durable_lease(
+            self._root_trace_id,
+            lease.retrieval_key,
+            owner=lease.owner,
+            epoch=lease.epoch,
+            lease_seconds=lease_seconds,
+        )
+
+    async def complete(
+        self, lease: RetrievalLease, result: RetrievalPipelineResult
+    ) -> RetrievalPipelineResult:
+        value = await self._coordinator.complete_durable_lease(
+            self._root_trace_id,
+            lease.retrieval_key,
+            owner=lease.owner,
+            epoch=lease.epoch,
+            result=result.to_payload(),
+        )
+        return RetrievalPipelineResult.from_payload(value)
+
+
+__all__ = ["LedgerRetrievalLeaseRepository"]

script_build_host/src/script_build_host/tools/__init__.py

@@ -1,14 +1,12 @@
 """Ownership-aware tools exposed to script-build Agents."""
 
 from .contracts import AttemptManifest, ScriptCandidateToolPort, ScriptPlannerToolPort
-from .gateway import LegacyScriptToolGateway, RetrievalResult
-from .registry import TASK_PRESET_BY_KIND, register_script_tools
+from .gateway import LegacyScriptToolGateway
+from .registry import register_script_tools
 
 __all__ = [
-    "TASK_PRESET_BY_KIND",
     "AttemptManifest",
     "LegacyScriptToolGateway",
-    "RetrievalResult",
     "ScriptCandidateToolPort",
     "ScriptPlannerToolPort",
     "register_script_tools",

script_build_host/src/script_build_host/tools/gateway.py

@@ -7,31 +7,31 @@ framework's protected context and then checked against durable bindings.
 
 from __future__ import annotations
 
-import asyncio
 import re
 from collections.abc import Awaitable, Callable, Mapping, Sequence
-from dataclasses import asdict, dataclass
-from datetime import UTC, datetime
+from dataclasses import asdict
+from datetime import datetime
 from hashlib import sha256
 from typing import Any, Protocol, cast
 from unicodedata import normalize
-from uuid import uuid4
 
 from agent.orchestration import (
+    AgentRole,
     ArtifactRef,
     AttemptSubmission,
     CriterionResult,
     EvidenceQuery,
     ValidationVerdict,
 )
+from agent.orchestration.validation_evidence import evidence_handle
 
 from script_build_host.agents.validation import (
-    PHASE_TWO_KINDS,
     RETRIEVAL_KINDS,
     ROOT_DELIVERY_KIND,
     precheck_business_artifact,
     precheck_snapshot,
     task_kind,
+    validation_layer,
 )
 from script_build_host.domain.artifacts import (
     ArtifactKind,
@@ -39,9 +39,11 @@ from script_build_host.domain.artifacts import (
     DirectionConstraint,
     DirectionGoal,
     DirectionPreference,
-    EvidenceRecordV1,
 )
-from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation, ScriptBuildError
+from script_build_host.domain.errors import (
+    ProtocolViolation,
+    ValidationEvidenceUnauthorized,
+)
 from script_build_host.domain.ports import (
     InputSnapshotRepository,
     MissionBindingRepository,
@@ -54,7 +56,7 @@ from script_build_host.domain.workbench import (
     strategy_handle,
     strategy_ref,
 )
-from script_build_host.infrastructure.redaction import redact, redact_text
+from script_build_host.infrastructure.redaction import redact_text
 
 from .contracts import RootDeliveryToolPort, ScriptCandidateToolPort, ScriptPlannerToolPort
 
@@ -66,26 +68,6 @@ _BUSINESS_ARTIFACT_KINDS = frozenset(item.value for item in ArtifactKind)
 _DEFECT_CODE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")
 
 
-@dataclass(frozen=True, slots=True)
-class RetrievalResult:
-    source_refs: tuple[str, ...]
-    summary: str
-    supports: tuple[str, ...] = ()
-    confidence: str = "unknown"
-    limitations: tuple[str, ...] = ()
-    raw_artifact_ref: str | None = None
-    metadata: Mapping[str, Any] | None = None
-
-
-class RetrievalAdapter(Protocol):
-    async def retrieve(
-        self,
-        *,
-        query: Mapping[str, Any],
-        snapshot: Any,
-    ) -> RetrievalResult: ...
-
-
 class DispatchGuard(Protocol):
     async def ensure_dispatch_allowed(self, script_build_id: int) -> None: ...
 
@@ -108,7 +90,6 @@ class LegacyScriptToolGateway:
         snapshots: InputSnapshotRepository,
         artifacts: ScriptBusinessArtifactRepository,
         coordinator: Any,
-        retrieval_adapters: Mapping[str, RetrievalAdapter] | None = None,
         dispatch_guard: DispatchGuard | None = None,
         planner_tools: ScriptPlannerToolPort | None = None,
         candidate_tools: ScriptCandidateToolPort | None = None,
@@ -120,14 +101,12 @@ class LegacyScriptToolGateway:
         self.snapshots = snapshots
         self.artifacts = artifacts
         self.coordinator = coordinator
-        self.retrieval_adapters = dict(retrieval_adapters or {})
         self.dispatch_guard = dispatch_guard
         self.planner_tools = planner_tools
         self.candidate_tools = candidate_tools
         self.budget_guard = budget_guard
         self.root_tools = root_tools
         self.workbench = workbench
-        self._retrieval_locks: dict[tuple[str, str, str], asyncio.Lock] = {}
 
     async def ensure_dispatch_allowed(self, context: Mapping[str, Any]) -> None:
         binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
@@ -145,8 +124,7 @@ class LegacyScriptToolGateway:
     async def plan_script_tasks(
         self,
         *,
-        contracts: Sequence[Mapping[str, Any]],
-        parent_task_id: str | None,
+        intents: Sequence[Mapping[str, Any]],
         context: Mapping[str, Any],
     ) -> Mapping[str, Any]:
         return cast(
@@ -154,8 +132,7 @@ class LegacyScriptToolGateway:
             await self._execute_fenced(
                 context,
                 lambda: self._planner().plan_script_tasks(
-                    contract_payloads=contracts,
-                    parent_task_id=parent_task_id,
+                    intents=intents,
                     context=context,
                 ),
             ),
@@ -168,9 +145,6 @@ class LegacyScriptToolGateway:
         action: str,
         reason: str,
         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(
@@ -182,9 +156,6 @@ class LegacyScriptToolGateway:
                     action=action,
                     reason=reason,
                     validation_id=validation_id,
-                    replacement_contract=replacement_contract,
-                    child_contracts=child_contracts,
-                    selected_decision_ids=selected_decision_ids,
                     context=context,
                 ),
             ),
@@ -235,6 +206,67 @@ class LegacyScriptToolGateway:
     ) -> Mapping[str, Any]:
         if self.workbench is None:
             raise ProtocolViolation("Context Broker is not configured")
+        if str(context.get("role") or "") == AgentRole.VALIDATOR.value:
+            if section != "content":
+                raise ProtocolViolation(
+                    "Validator evidence reads must use the complete content view"
+                )
+            root_trace_id = _required(context, "root_trace_id")
+            validation_id = _required(context, "validation_id")
+            ledger = await self.coordinator.task_store.load(root_trace_id)
+            validation = ledger.validations.get(validation_id)
+            if validation is None or validation.read_session is None:
+                raise ProtocolViolation("Validation evidence session is not initialized")
+            grant = next(
+                (
+                    item
+                    for item in validation.evidence_snapshot.grants
+                    if item.handle == handle
+                ),
+                None,
+            )
+            if grant is None:
+                raise ProtocolViolation(
+                    "context handle is outside this Validation evidence snapshot"
+                )
+            reserved = await self.coordinator.reserve_validation_read(
+                root_trace_id,
+                validation_id,
+                evidence_revision=validation.evidence_snapshot.revision,
+                handle=handle,
+                cursor=cursor,
+            )
+            if reserved["replayed"]:
+                return cast(Mapping[str, Any], reserved["result"])
+            try:
+                payload = await self.workbench.read_validation_evidence(
+                    root_trace_id=root_trace_id,
+                    validation_id=validation_id,
+                    artifact_ref=grant.artifact_ref,
+                    handle=handle,
+                    section=section,
+                    cursor=cursor,
+                )
+            except Exception:
+                await self.coordinator.cancel_validation_read(
+                    root_trace_id,
+                    validation_id,
+                    evidence_revision=validation.evidence_snapshot.revision,
+                    handle=handle,
+                    cursor=cursor,
+                )
+                raise
+            await self.coordinator.complete_validation_read(
+                root_trace_id,
+                validation_id,
+                evidence_revision=validation.evidence_snapshot.revision,
+                handle=handle,
+                cursor=cursor,
+                next_cursor=cast(str | None, payload.get("next_cursor")),
+                exhausted=bool(payload.get("exhausted")),
+                result=payload,
+            )
+            return cast(Mapping[str, Any], payload)
         return cast(
             Mapping[str, Any],
             await self.workbench.read_mission_context(
@@ -352,121 +384,6 @@ class LegacyScriptToolGateway:
                 )
         return values
 
-    async def retrieve(
-        self,
-        source_type: str,
-        tool_name: str,
-        query: Mapping[str, Any],
-        context: Mapping[str, Any],
-    ) -> dict[str, Any]:
-        await self.ensure_dispatch_allowed(context)
-        if self.budget_guard is not None:
-            await self.budget_guard.ensure_tool_budget(
-                root_trace_id=_required(context, "root_trace_id"),
-                task_id=_required(context, "task_id"),
-                entry="retrieval",
-            )
-        root_trace_id = _required(context, "root_trace_id")
-        task_id = _required(context, "task_id")
-        attempt_id = _required(context, "attempt_id")
-        retrieval_key = (root_trace_id, task_id, attempt_id)
-        lock = self._retrieval_locks.setdefault(retrieval_key, asyncio.Lock())
-        async with lock:
-            binding = await self.bindings.get_by_root(root_trace_id)
-            try:
-                existing = await self.artifacts.get_by_attempt(
-                    script_build_id=binding.script_build_id,
-                    task_id=task_id,
-                    attempt_id=attempt_id,
-                )
-            except ArtifactNotFound:
-                pass
-            else:
-                artifact = existing.artifact
-                frozen_query = redact(dict(query))
-                if not isinstance(artifact, EvidenceRecordV1) or not isinstance(
-                    frozen_query, dict
-                ):
-                    raise ProtocolViolation(
-                        "Retrieval Attempt already owns a non-Evidence artifact"
-                    )
-                stored_query = dict(artifact.query)
-                stored_query.pop("_retrieval_metadata", None)
-                if (
-                    artifact.source_type != source_type
-                    or artifact.tool_name != tool_name
-                    or stored_query != frozen_query
-                ):
-                    raise ProtocolViolation(
-                        "Retrieval Attempt is already bound to a different external query"
-                    )
-                return _evidence_tool_payload(
-                    existing,
-                    root_trace_id=root_trace_id,
-                    task_id=task_id,
-                    attempt_id=attempt_id,
-                    resumed=True,
-                )
-            return await self._retrieve_once(source_type, tool_name, query, context)
-
-    async def _retrieve_once(
-        self,
-        source_type: str,
-        tool_name: str,
-        query: Mapping[str, Any],
-        context: Mapping[str, Any],
-    ) -> dict[str, Any]:
-        binding, snapshot = await self._scope(context)
-        if source_type == "decode" and query.get("account_name"):
-            frozen_accounts = {
-                str(snapshot.account.get(key) or "")
-                for key in ("account_name", "resolved_account_name")
-            }
-            if query["account_name"] not in frozen_accounts:
-                raise ProtocolViolation("decode account_name is outside the frozen account scope")
-        adapter = self.retrieval_adapters.get(source_type)
-        if adapter is None:
-            raise ProtocolViolation(f"retrieval adapter is not configured: {source_type}")
-        result = await adapter.retrieve(query=dict(query), snapshot=snapshot)
-        frozen_query = redact(dict(query))
-        if not isinstance(frozen_query, dict):
-            raise ProtocolViolation("redacted retrieval query must remain an object")
-        if result.metadata:
-            safe_metadata = redact(dict(result.metadata))
-            if not isinstance(safe_metadata, dict):
-                raise ProtocolViolation("redacted retrieval metadata must remain an object")
-            frozen_query["_retrieval_metadata"] = safe_metadata
-        record = EvidenceRecordV1(
-            evidence_id=str(uuid4()),
-            source_type=source_type,
-            tool_name=tool_name,
-            query=frozen_query,
-            source_refs=_safe_text_tuple(result.source_refs),
-            raw_artifact_ref=(
-                redact_text(result.raw_artifact_ref) if result.raw_artifact_ref else None
-            ),
-            summary=redact_text(result.summary),
-            supports=_safe_text_tuple(result.supports),
-            confidence=redact_text(result.confidence),
-            limitations=_safe_text_tuple(result.limitations),
-            content_sha256="",
-            created_at=datetime.now(UTC),
-        )
-        version, _ref = await self.artifacts.freeze(
-            script_build_id=binding.script_build_id,
-            task_id=_required(context, "task_id"),
-            attempt_id=_required(context, "attempt_id"),
-            spec_version=_required_int(context, "spec_version"),
-            artifact=record,
-        )
-        return _evidence_tool_payload(
-            version,
-            root_trace_id=_required(context, "root_trace_id"),
-            task_id=_required(context, "task_id"),
-            attempt_id=_required(context, "attempt_id"),
-            resumed=False,
-        )
-
     async def view_frozen_images(
         self, image_handles: Sequence[str], context: Mapping[str, Any]
     ) -> Sequence[Mapping[str, Any]]:
@@ -607,18 +524,18 @@ 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 allowed_values}
-        refs_by_handle = {_evidence_handle(ref): ref for ref in allowed_refs.values()}
+        validation_id = _required(context, "validation_id")
+        validation = ledger.validations.get(validation_id)
+        if validation is None:
+            raise ProtocolViolation("Validation is outside the protected mission")
+        refs_by_handle = {
+            item.handle: item.artifact_ref
+            for item in validation.evidence_snapshot.grants
+        }
+        allowed_refs = {
+            (ref.uri, ref.version, ref.digest, ref.kind): ref
+            for ref in refs_by_handle.values()
+        }
         normalized_defects = _normalize_defects(
             [_expand_validation_handles(item, refs_by_handle) for item in defects],
             allowed_refs,
@@ -662,17 +579,6 @@ class LegacyScriptToolGateway:
         if set(result_by_id) != expected_ids:
             raise ProtocolViolation("validation must report every Task criterion exactly once")
         overall = ValidationVerdict(verdict)
-        if overall is ValidationVerdict.PASSED and self.workbench is not None:
-            try:
-                await self.workbench.verify_context_exhausted(
-                    root_trace_id=root_trace_id,
-                    task_id=task_id,
-                    attempt_id=_required(context, "attempt_id"),
-                )
-            except Exception as exc:
-                if getattr(exc, "code", None) != "CONTEXT_NOT_EXHAUSTED":
-                    raise
-                raise ScriptBuildError("CONTEXT_NOT_EXHAUSTED", str(exc)) from exc
         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")
@@ -691,6 +597,7 @@ class LegacyScriptToolGateway:
             attempt=attempt,
             snapshot=snapshot,
             prior_validation_count=max(0, len(task.validation_ids) - 1),
+            context_documents=ledger.context_documents,
         )
         deterministic = precheck_snapshot(validation_context)
         if overall is ValidationVerdict.PASSED and any(
@@ -699,7 +606,10 @@ class LegacyScriptToolGateway:
             raise ProtocolViolation("passed validation conflicts with deterministic precheck")
         if (
             overall is ValidationVerdict.PASSED
-            and task_kind(task.current_spec.context_refs) in PHASE_TWO_KINDS
+            and validation_layer(
+                task_kind(task.current_spec.context_refs) or ""
+            )
+            in {"local", "compare", "global", "governance"}
         ):
             binding = await self.bindings.get_by_root(root_trace_id)
             for ref in snapshot.artifact_refs:
@@ -720,7 +630,10 @@ class LegacyScriptToolGateway:
                 )
         if (
             overall is ValidationVerdict.PASSED
-            and task_kind(task.current_spec.context_refs) == ROOT_DELIVERY_KIND
+            and validation_layer(
+                task_kind(task.current_spec.context_refs) or ""
+            )
+            == "root-delivery"
         ):
             root_rules = await self._root_tools().deterministic_precheck(context=context)
             if any(item.get("verdict") != "passed" for item in root_rules):
@@ -894,9 +807,9 @@ class LegacyScriptToolGateway:
         from agent.orchestration.validation_policy import ValidationContext
 
         from script_build_host.agents.validation import (
-            PHASE_TWO_KINDS,
             precheck_snapshot,
             task_kind,
+            validation_layer,
         )
 
         root_trace_id = _required(context, "root_trace_id")
@@ -917,9 +830,10 @@ class LegacyScriptToolGateway:
             item.to_dict() for item in precheck_snapshot(validation_context)
         ]
         kind = task_kind(task.current_spec.context_refs)
-        if kind in PHASE_TWO_KINDS:
+        layer = validation_layer(kind or "")
+        if layer in {"local", "compare", "global", "governance"}:
             rules.extend(await self._candidates().deterministic_precheck(context=context))
-        elif kind == "root-delivery":
+        elif layer == "root-delivery":
             rules.extend(await self._root_tools().deterministic_precheck(context=context))
         return {"rule_results": rules}
 
@@ -948,28 +862,6 @@ class LegacyScriptToolGateway:
         return self.root_tools
 
 
-def _evidence_tool_payload(
-    version: Any,
-    *,
-    root_trace_id: str,
-    task_id: str,
-    attempt_id: str,
-    resumed: bool,
-) -> dict[str, Any]:
-    artifact = version.artifact
-    if not isinstance(artifact, EvidenceRecordV1):
-        raise ProtocolViolation("Retrieval Attempt does not own Evidence")
-    digest = str(version.canonical_sha256 or version.artifact_version_id)
-    return {
-        "evidence_handle": semantic_handle("evidence", root_trace_id, task_id, digest),
-        "evidence": protect_model_value(
-            _json_values(asdict(artifact)), namespace=attempt_id
-        ),
-        "source_count": len(artifact.source_refs),
-        "resumed_from_frozen_evidence": resumed,
-    }
-
-
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
     if not isinstance(value, str) or not value.strip():
@@ -1110,7 +1002,7 @@ def _attempt_summary(ref: ArtifactRef, scope_ref: str) -> str:
 
 
 def _evidence_handle(ref: ArtifactRef) -> str:
-    return semantic_handle("src", ref.uri, ref.digest or ref.version or "")
+    return cast(str, evidence_handle(ref))
 
 
 def _required_evidence_handle(
@@ -1119,7 +1011,9 @@ def _required_evidence_handle(
     handle = str(value or "").strip()
     ref = refs_by_handle.get(handle)
     if ref is None:
-        raise ProtocolViolation("validation references an unknown evidence handle")
+        raise ValidationEvidenceUnauthorized(
+            "validation references an unknown evidence handle"
+        )
     return ref
 
 
@@ -1229,7 +1123,9 @@ def _normalize_evidence_refs(
         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")
+            raise ValidationEvidenceUnauthorized(
+                f"{label} must come from the protected validation closure"
+            )
         refs.append(frozen)
     return _dedupe_refs(refs)
 
@@ -1256,8 +1152,6 @@ def _dedupe_refs(values: Any) -> tuple[ArtifactRef, ...]:
 __all__ = [
     "DispatchGuard",
     "LegacyScriptToolGateway",
-    "RetrievalAdapter",
-    "RetrievalResult",
     "ScriptCandidateToolPort",
     "ScriptPlannerToolPort",
 ]

script_build_host/src/script_build_host/tools/registry.py

@@ -12,16 +12,11 @@ from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolReg
 from agent.tools.models import ToolResult
 
 from script_build_host.domain.errors import ScriptBuildError
-from script_build_host.domain.task_capabilities import task_capabilities
 from script_build_host.domain.workbench import protect_model_value
 
 from .failures import classify_script_tool_failure
 from .gateway import LegacyScriptToolGateway
 
-TASK_PRESET_BY_KIND = {
-    item.task_kind.value: item.worker_preset for item in task_capabilities()
-}
-
 _MAX_STRUCTURED_ARGUMENT_BYTES = 2 * 1024 * 1024
 _MAX_STRUCTURED_LIST_ITEMS = 1_000
 
@@ -33,42 +28,16 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     isolated tests deterministic while production creates only one composition.
     """
 
-    async def complete_retrieval(
-        source_type: str,
-        tool_name: str,
-        query: dict[str, Any],
-        context: dict[str, Any],
-    ) -> ToolResult:
-        evidence = await gateway.retrieve(source_type, tool_name, query, context)
-        await gateway.submit_current_attempt(context)
-        result_json = _json(
-            {
-                "evidence": protect_model_value(
-                    evidence, namespace=str(context.get("attempt_id") or tool_name)
-                ),
-                "submitted": True,
-            }
-        )
-        return ToolResult(
-            title="Evidence frozen and attempt submitted",
-            output=result_json,
-            long_term_memory="The unique retrieval Evidence was frozen and submitted",
-            terminate_run=True,
-            result_summary=result_json,
-        )
-
     async def plan_script_tasks(
-        contracts: list[dict[str, Any]],
-        parent_task_id: str | None = None,
+        intents: list[dict[str, Any]],
         context: dict[str, Any] | None = None,
     ) -> str | ToolResult:
         """Resolve Planner-owned semantics and atomically create Tasks."""
 
-        contracts = _structured_object_list(contracts, "contracts", max_items=50)
+        intents = _structured_object_list(intents, "intents", max_items=50)
         return _json(
             await gateway.plan_script_tasks(
-                contracts=contracts,
-                parent_task_id=parent_task_id,
+                intents=intents,
                 context=context or {},
             )
         )
@@ -85,9 +54,6 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         action: str,
         reason: str,
         validation_id: str | None = None,
-        replacement_contract: dict[str, Any] | None = None,
-        child_contracts: list[dict[str, Any]] | None = None,
-        selected_decision_ids: list[str] | None = None,
         context: dict[str, Any] | None = None,
     ) -> str | ToolResult:
         """Apply one phase-aware Planner decision after contract and boundary checks."""
@@ -96,26 +62,15 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             "accept",
             "repair",
             "retry",
-            "revise",
-            "split",
             "block",
-            "supersede",
             "cancel",
         }:
             raise ValueError("action is not supported by decide_script_task")
-        if replacement_contract is not None:
-            replacement_contract = _structured_object(replacement_contract, "replacement_contract")
-        child_contracts = _structured_object_list(
-            child_contracts or [], "child_contracts", max_items=50
-        )
         result = await gateway.decide_script_task(
             task_id=task_id,
             action=action,
             reason=reason,
             validation_id=validation_id,
-            replacement_contract=replacement_contract,
-            child_contracts=child_contracts,
-            selected_decision_ids=selected_decision_ids or [],
             context=context or {},
         )
         result_json = _json(result)
@@ -286,115 +241,6 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, dispatch_script_tasks, capabilities=["task_control"])
 
-    async def query_pattern_qa(message: str, context: dict[str, Any] | None = None) -> ToolResult:
-        """Query the configured Pattern service and freeze its evidence."""
-
-        return await complete_retrieval(
-            "pattern",
-            "query_pattern_qa",
-            {"message": message},
-            context or {},
-        )
-
-    _register(registry, query_pattern_qa, capabilities=["external_send", "write"])
-
-    async def search_script_decode_case(
-        return_field: str,
-        keyword: str = "",
-        account_name: str | None = None,
-        match_fields: dict[str, str] | None = None,
-        top_k: int = 3,
-        context: dict[str, Any] | None = None,
-    ) -> ToolResult:
-        """Search the frozen script-decode index and freeze ranked evidence."""
-
-        if return_field not in {"主脉络", "元素", "主题", "形式", "作用", "感受"}:
-            raise ValueError("return_field is not supported")
-        if not keyword.strip() and not account_name:
-            raise ValueError("keyword and account_name cannot both be empty")
-        if not 1 <= top_k <= 5:
-            raise ValueError("top_k must be between 1 and 5")
-        if match_fields is not None:
-            match_fields = _structured_object(match_fields, "match_fields")
-        valid_columns = {"主题", "形式", "作用", "感受"}
-        valid_subfields = {"维度名", "维度值"}
-        if match_fields and (
-            set(match_fields) - valid_columns or set(match_fields.values()) - valid_subfields
-        ):
-            raise ValueError("match_fields contains an unsupported column or subfield")
-        return await complete_retrieval(
-            "decode",
-            "search_script_decode_case",
-            {
-                "return_field": return_field,
-                "keyword": keyword,
-                "account_name": account_name,
-                "match_fields": match_fields or {},
-                "top_k": top_k,
-            },
-            context or {},
-        )
-
-    _register(
-        registry,
-        search_script_decode_case,
-        capabilities=["external_send", "read", "write"],
-        schema=_decode_schema(),
-    )
-
-    async def external_search_case(
-        keyword: str,
-        platform_channel: str = "xhs",
-        max_count: int = 5,
-        context: dict[str, Any] | None = None,
-    ) -> ToolResult:
-        """Search the allowlisted external service without legacy log writes."""
-
-        if platform_channel not in {"xhs", "zhihu"}:
-            raise ValueError("platform_channel must be xhs or zhihu")
-        if not 1 <= max_count <= 20:
-            raise ValueError("max_count must be between 1 and 20")
-        return await complete_retrieval(
-            "external",
-            "external_search_case",
-            {
-                "keyword": keyword,
-                "platform_channel": platform_channel,
-                "max_count": max_count,
-            },
-            context or {},
-        )
-
-    _register(
-        registry,
-        external_search_case,
-        capabilities=["external_send", "write"],
-        schema=_external_schema(),
-    )
-
-    async def search_knowledge(
-        keyword: str,
-        max_count: int = 3,
-        context: dict[str, Any] | None = None,
-    ) -> ToolResult:
-        """Search the frozen internal knowledge corpus and freeze evidence."""
-
-        if not 1 <= max_count <= 20:
-            raise ValueError("max_count must be between 1 and 20")
-        return await complete_retrieval(
-            "knowledge",
-            "search_knowledge",
-            {"keyword": keyword, "max_count": max_count},
-            context or {},
-        )
-
-    _register(
-        registry,
-        search_knowledge,
-        capabilities=["read", "write"],
-        schema=_knowledge_schema(),
-    )
-
     async def save_direction_candidate(
         goals: Any,
         constraints: Any = None,
@@ -745,43 +591,6 @@ def _strongest_cycle_failure(results: Any) -> FailureDetail | None:
     return strongest
 
 
-def _decode_schema() -> dict[str, Any]:
-    return {
-        "type": "function",
-        "function": {
-            "name": "search_script_decode_case",
-            "description": "Search frozen script-decode cases.",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "return_field": {
-                        "type": "string",
-                        "enum": ["主脉络", "元素", "主题", "形式", "作用", "感受"],
-                    },
-                    "keyword": {"type": "string", "default": ""},
-                    "account_name": {"type": ["string", "null"]},
-                    "match_fields": {
-                        "type": "object",
-                        "properties": {
-                            key: {"type": "string", "enum": ["维度名", "维度值"]}
-                            for key in ("主题", "形式", "作用", "感受")
-                        },
-                        "additionalProperties": False,
-                    },
-                    "top_k": {
-                        "type": "integer",
-                        "default": 3,
-                        "minimum": 1,
-                        "maximum": 5,
-                    },
-                },
-                "required": ["return_field"],
-                "additionalProperties": False,
-            },
-        },
-    }
-
-
 def _plan_script_tasks_schema() -> dict[str, Any]:
     return {
         "type": "function",
@@ -794,24 +603,14 @@ def _plan_script_tasks_schema() -> dict[str, Any]:
             "parameters": {
                 "type": "object",
                 "properties": {
-                    "contracts": {
+                    "intents": {
                         "type": "array",
                         "items": _planner_task_input_schema(),
                         "minItems": 1,
                         "maxItems": 50,
                     },
-                    "parent_task_id": {
-                        "type": ["string", "null"],
-                        "default": None,
-                        "description": (
-                            "Parent Task that owns every new semantic scope. Structure, Paragraph "
-                            "and ElementSet inputs stay in one creative branch and are normally "
-                            "siblings under the same Compose parent; completed producer Tasks "
-                            "cannot receive new children."
-                        ),
-                    },
                 },
-                "required": ["contracts"],
+                "required": ["intents"],
                 "additionalProperties": False,
             },
         },
@@ -823,10 +622,7 @@ def _decide_script_task_schema() -> dict[str, Any]:
         "type": "function",
         "function": {
             "name": "decide_script_task",
-            "description": (
-                "Apply one Planner decision. Use selected_decision_ids to close Compose or "
-                "Portfolio; the Host derives every immutable adoption field."
-            ),
+            "description": "Apply one Planner decision to durable Task state.",
             "parameters": {
                 "type": "object",
                 "properties": {
@@ -837,35 +633,12 @@ def _decide_script_task_schema() -> dict[str, Any]:
                             "accept",
                             "repair",
                             "retry",
-                            "revise",
-                            "split",
                             "block",
-                            "supersede",
                             "cancel",
                         ],
                     },
                     "reason": {"type": "string"},
                     "validation_id": {"type": ["string", "null"], "default": None},
-                    "replacement_contract": {
-                        "oneOf": [
-                            _planner_task_input_schema(),
-                            _root_replacement_schema(),
-                            {"type": "null"},
-                        ],
-                        "default": None,
-                    },
-                    "child_contracts": {
-                        "type": "array",
-                        "items": _planner_task_input_schema(),
-                        "maxItems": 50,
-                        "default": [],
-                    },
-                    "selected_decision_ids": {
-                        "type": "array",
-                        "items": {"type": "string"},
-                        "uniqueItems": True,
-                        "default": [],
-                    },
                 },
                 "required": ["task_id", "action", "reason"],
                 "additionalProperties": False,
@@ -877,10 +650,6 @@ def _decide_script_task_schema() -> dict[str, Any]:
 def _planner_task_input_schema() -> dict[str, Any]:
     task_kinds = [
         "direction",
-        "pattern-retrieval",
-        "decode-retrieval",
-        "external-retrieval",
-        "knowledge-retrieval",
         "structure",
         "paragraph",
         "element-set",
@@ -892,109 +661,38 @@ def _planner_task_input_schema() -> dict[str, Any]:
         "type": "object",
         "properties": {
             "task_kind": {"type": "string", "enum": task_kinds},
-            "scope_selector": {
-                "type": "object",
-                "properties": {
-                    "anchor": {"type": "string", "enum": ["mission", "parent"]},
-                    "path": {
-                        "type": "array",
-                        "maxItems": 4,
-                        "items": {"type": "string", "minLength": 1, "maxLength": 128},
-                    },
-                },
-                "required": ["anchor", "path"],
-                "additionalProperties": False,
-            },
-            "intent_class": {
-                "type": "string",
-                "enum": [
-                    "explore",
-                    "expand",
-                    "replace",
-                    "repair",
-                    "compare",
-                    "compose",
-                    "portfolio",
-                ],
-            },
             "objective": {"type": "string", "maxLength": 2000},
-            "input_decision_ids": {
+            "decision_ids": {
                 "type": "array",
                 "items": {"type": "string"},
                 "uniqueItems": True,
                 "default": [],
             },
-            "base_decision_id": {"type": ["string", "null"], "default": None},
-            "comparison_decision_ids": {
-                "type": "array",
-                "items": {"type": "string"},
-                "uniqueItems": True,
-                "default": [],
+            "change_mode": {
+                "type": "string",
+                "enum": ["new", "derive", "replace"],
+                "default": "new",
             },
-            "supersedes_decision_ids": {
+            "target_goal_ids": {
                 "type": "array",
                 "items": {"type": "string"},
                 "uniqueItems": True,
                 "default": [],
             },
-            "gap_key": {"type": ["string", "null"]},
-            "criteria": {
+            "quality_requirements": {
                 "type": "array",
-                "minItems": 0,
                 "maxItems": 16,
-                "items": {
-                    "type": "object",
-                    "properties": {
-                        "client_key": {"type": "string"},
-                        "criterion_type": {
-                            "type": "string",
-                            "enum": [
-                                "custom",
-                                "evidence_nonempty",
-                                "source_attributable",
-                                "semantic_relevance",
-                                "concept_coverage",
-                                "freshness",
-                            ],
-                        },
-                        "description": {"type": "string", "default": ""},
-                        "params": {
-                            "type": "object",
-                            "default": {},
-                            "description": (
-                                "Shapes by criterion_type: evidence_nonempty accepts optional "
-                                "integer minimum_count; source_attributable uses {}; "
-                                "semantic_relevance and concept_coverage require string array "
-                                "concepts and accept optional integer minimum_matches; freshness "
-                                "accepts optional integer max_age_days. Never use query."
-                            ),
-                        },
-                        "hard": {"type": "boolean"},
-                    },
-                    "required": ["client_key", "criterion_type", "hard"],
-                    "additionalProperties": False,
-                },
-            },
-            "goal_ids": {
-                "type": "array",
-                "maxItems": 30,
+                "items": {"type": "string", "minLength": 1, "maxLength": 500},
                 "uniqueItems": True,
-                "items": {"type": "string", "minLength": 1},
-                "description": (
-                    "Empty in Phase1. In Phase2, the active Direction Goal IDs implemented "
-                    "by this Task. Compose and Portfolio include every Goal."
-                ),
+                "default": [],
+            },
+            "evidence_request": {
+                "type": ["string", "null"],
+                "maxLength": 1000,
             },
+            "within_task_id": {"type": ["string", "null"]},
         },
-        "required": [
-            "task_kind",
-            "scope_selector",
-            "intent_class",
-            "objective",
-            "criteria",
-            "goal_ids",
-            "input_decision_ids",
-        ],
+        "required": ["task_kind", "objective"],
         "additionalProperties": False,
     }
 
@@ -1356,49 +1054,6 @@ def _submit_validation_schema() -> dict[str, Any]:
     }
 
 
-def _external_schema() -> dict[str, Any]:
-    return {
-        "type": "function",
-        "function": {
-            "name": "external_search_case",
-            "description": "Search allowlisted external content cases.",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "keyword": {"type": "string"},
-                    "platform_channel": {
-                        "type": "string",
-                        "enum": ["xhs", "zhihu"],
-                        "default": "xhs",
-                    },
-                    "max_count": {"type": "integer", "default": 5},
-                },
-                "required": ["keyword"],
-                "additionalProperties": False,
-            },
-        },
-    }
-
-
-def _knowledge_schema() -> dict[str, Any]:
-    return {
-        "type": "function",
-        "function": {
-            "name": "search_knowledge",
-            "description": "Search the frozen creation knowledge corpus.",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "keyword": {"type": "string"},
-                    "max_count": {"type": "integer", "default": 3},
-                },
-                "required": ["keyword"],
-                "additionalProperties": False,
-            },
-        },
-    }
-
-
 def _json(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
 
@@ -1495,4 +1150,4 @@ def _validated_images(
     return images, manifest
 
 
-__all__ = ["TASK_PRESET_BY_KIND", "register_script_tools"]
+__all__ = ["register_script_tools"]

script_build_host/tests/test_agent_surface.py

@@ -30,10 +30,7 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
     expected = {
         "script_planner",
         "script_direction_worker",
-        "script_pattern_retrieval_worker",
-        "script_decode_retrieval_worker",
-        "script_external_retrieval_worker",
-        "script_knowledge_retrieval_worker",
+        "script_evidence_retrieval_worker",
         "script_retrieval_validator",
         "script_candidate_validator",
         "script_structure_worker",
@@ -103,7 +100,7 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
 
 def test_new_build_prompt_manifest_freezes_every_phase_one_and_phase_two_role() -> None:
     manifest = script_build_prompt_manifest()
-    assert len(manifest) == 15
+    assert len(manifest) == 12
     assert len({str(item["preset"]) for item in manifest}) == len(manifest)
     assert {str(item["preset"]) for item in manifest} >= {
         "script_planner",
@@ -139,6 +136,7 @@ def test_every_model_role_freezes_one_shared_context_access_protocol() -> None:
         "script_context_broker",
         "script_compose_worker",
         "script_candidate_portfolio_worker",
+        "script_evidence_retrieval_worker",
     }
 
     for preset, content in manifest.items():
@@ -149,14 +147,7 @@ def test_every_model_role_freezes_one_shared_context_access_protocol() -> None:
             assert content.count("## Context Access Protocol") == 1
 
     assert "active_has_more" in manifest["script_planner"]
-    for preset in (
-        "script_pattern_retrieval_worker",
-        "script_decode_retrieval_worker",
-        "script_external_retrieval_worker",
-        "script_knowledge_retrieval_worker",
-    ):
-        assert "read_mission_context` 的 cursor" in manifest[preset]
-        assert "不计作第二次" in manifest[preset]
+    assert "不由模型选择" in manifest["script_evidence_retrieval_worker"]
     assert "Direction、CandidatePortfolio" in manifest["script_root_worker"]
     assert "required_handles" in manifest["script_candidate_validator"]
     assert "required_handles" in manifest["script_root_validator"]
@@ -166,7 +157,7 @@ class _Gateway:
     coordinator = SimpleNamespace(task_store=None)
 
 
-def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
+def test_physical_retrieval_workers_are_not_agent_tools() -> None:
     registry = ToolRegistry()
     register_script_tools(registry, _Gateway())  # type: ignore[arg-type]
     schemas = {
@@ -175,17 +166,12 @@ def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
     descriptions = {
         item["function"]["name"]: item["function"]["description"] for item in registry.get_schemas()
     }
-    decode = schemas["search_script_decode_case"]
-    assert decode["required"] == ["return_field"]
-    assert decode["properties"]["match_fields"]["type"] == "object"
-    assert decode["properties"]["top_k"]["default"] == 3
-    external = schemas["external_search_case"]
-    assert external["required"] == ["keyword"]
-    assert external["properties"]["platform_channel"]["default"] == "xhs"
-    knowledge = schemas["search_knowledge"]
-    assert knowledge["required"] == ["keyword"]
-    assert knowledge["properties"]["max_count"]["default"] == 3
-    assert schemas["query_pattern_qa"]["required"] == ["message"]
+    assert {
+        "search_script_decode_case",
+        "external_search_case",
+        "search_knowledge",
+        "query_pattern_qa",
+    }.isdisjoint(schemas)
     assert schemas["search_mission_context"]["required"] == ["collection"]
     assert schemas["read_mission_context"]["required"] == ["handle"]
     assert "sufficiency" in descriptions["search_mission_context"]
@@ -210,11 +196,6 @@ def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
     ]
     assert "summary" not in schemas["submit_attempt"].get("properties", {})
     assert "artifact_refs" not in schemas["submit_attempt"].get("properties", {})
-    assert {item.value for item in registry.get_capabilities("search_script_decode_case")} == {
-        "external_send",
-        "read",
-        "write",
-    }
 
 
 def test_agent_visible_write_schemas_never_expose_mechanical_identity_fields() -> None:

script_build_host/tests/test_architecture_retrieval.py

@@ -0,0 +1,260 @@
+from __future__ import annotations
+
+import pytest
+
+from script_build_host.application.retrieval_pipeline import (
+    ProviderAttempt,
+    ProviderRetrievalResult,
+    RetrievalLease,
+    RetrievalPipeline,
+    RetrievalPipelineResult,
+    RetrievalStatus,
+)
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.evidence_requirements import (
+    EvidenceProvider,
+    EvidencePurpose,
+    EvidenceRequirementV1,
+)
+
+
+class MemoryLeases:
+    def __init__(self) -> None:
+        self.epochs: dict[str, int] = {}
+        self.active: dict[str, RetrievalLease] = {}
+        self.terminals: dict[str, RetrievalPipelineResult] = {}
+        self.attempts: list[ProviderAttempt] = []
+        self.provider_results: list[
+            tuple[EvidenceProvider, int, ProviderRetrievalResult]
+        ] = []
+        self.renewals = 0
+
+    async def reserve(
+        self, retrieval_key: str, *, owner: str, lease_seconds: int
+    ) -> RetrievalLease:
+        if retrieval_key in self.terminals:
+            return RetrievalLease(
+                retrieval_key, owner, self.epochs[retrieval_key], False,
+                self.terminals[retrieval_key],
+            )
+        if retrieval_key in self.active:
+            return RetrievalLease(
+                retrieval_key, owner, self.active[retrieval_key].epoch, False
+            )
+        epoch = self.epochs.get(retrieval_key, 0) + 1
+        self.epochs[retrieval_key] = epoch
+        lease = RetrievalLease(retrieval_key, owner, epoch, True)
+        self.active[retrieval_key] = lease
+        return lease
+
+    async def append_attempt(
+        self, lease: RetrievalLease, attempt: ProviderAttempt
+    ) -> None:
+        assert self.active[lease.retrieval_key] == lease
+        self.attempts.append(attempt)
+
+    async def append_provider_result(
+        self,
+        lease: RetrievalLease,
+        provider: EvidenceProvider,
+        number: int,
+        result: ProviderRetrievalResult,
+    ) -> None:
+        assert self.active[lease.retrieval_key] == lease
+        self.provider_results.append((provider, number, result))
+
+    async def renew(self, lease: RetrievalLease, *, lease_seconds: int) -> None:
+        assert self.active[lease.retrieval_key] == lease
+        assert lease_seconds > 0
+        self.renewals += 1
+
+    async def complete(
+        self, lease: RetrievalLease, result: RetrievalPipelineResult
+    ) -> RetrievalPipelineResult:
+        assert self.active.pop(lease.retrieval_key) == lease
+        self.terminals[lease.retrieval_key] = result
+        return result
+
+
+class Empty:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def retrieve(self, **_: object) -> ProviderRetrievalResult:
+        self.calls += 1
+        return ProviderRetrievalResult((), "empty")
+
+
+class Success:
+    def __init__(self, *, partial: bool = False) -> None:
+        self.calls = 0
+        self.partial = partial
+
+    async def retrieve(self, **_: object) -> ProviderRetrievalResult:
+        self.calls += 1
+        return ProviderRetrievalResult(("note:1",), "ok", partial=self.partial)
+
+
+class ResumedLeaseRepository(MemoryLeases):
+    def __init__(self, journal: tuple[dict[str, object], ...]) -> None:
+        super().__init__()
+        self.journal = journal
+
+    async def reserve(
+        self, retrieval_key: str, *, owner: str, lease_seconds: int
+    ) -> RetrievalLease:
+        lease = RetrievalLease(
+            retrieval_key,
+            owner,
+            2,
+            True,
+            journal=self.journal,
+        )
+        self.active[retrieval_key] = lease
+        return lease
+
+
+@pytest.mark.asyncio
+async def test_empty_knowledge_falls_through_once_to_partial_xhs() -> None:
+    knowledge, xhs = Empty(), Success(partial=True)
+    repository = MemoryLeases()
+    pipeline = RetrievalPipeline(
+        {
+            EvidenceProvider.KNOWLEDGE: knowledge,
+            EvidenceProvider.EXTERNAL_XHS: xhs,
+        },
+        repository,
+        owner="host-a",
+    )
+    requirement = EvidenceRequirementV1(
+        "persona",
+        EvidencePurpose.PERSONA,
+        True,
+        1,
+        source_chain=(EvidenceProvider.KNOWLEDGE, EvidenceProvider.EXTERNAL_XHS),
+    )
+    result = await pipeline.require(requirement, query={"account": "a"}, snapshot={})
+    replay = await pipeline.require(requirement, query={"account": "a"}, snapshot={})
+    assert result.status is RetrievalStatus.PARTIAL_SUCCESS
+    assert replay == result
+    assert knowledge.calls == xhs.calls == 1
+
+
+@pytest.mark.asyncio
+async def test_input_satisfied_never_reserves_or_calls_provider() -> None:
+    provider = Success()
+    repository = MemoryLeases()
+    requirement = EvidenceRequirementV1(
+        "persona",
+        EvidencePurpose.PERSONA,
+        True,
+        1,
+        satisfied_by_input_ref="input:persona",
+    )
+    result = await RetrievalPipeline(
+        {EvidenceProvider.EXTERNAL_XHS: provider},
+        repository,
+        owner="host-a",
+    ).execute(requirement, query={}, snapshot={})
+    assert result.status is RetrievalStatus.SATISFIED_BY_INPUT
+    assert provider.calls == 0
+    assert repository.epochs == {}
+
+
+def test_provider_result_rejects_empty_evidence_identity() -> None:
+    with pytest.raises(ProtocolViolation, match="empty evidence"):
+        ProviderRetrievalResult(("",), "invalid")
+
+
+def test_persisted_null_evidence_identity_is_rejected() -> None:
+    payload = RetrievalPipelineResult(
+        "persona",
+        RetrievalStatus.SUCCESS,
+        EvidenceProvider.EXTERNAL_XHS,
+        (),
+        (),
+    ).to_payload()
+    payload["evidence_items"] = [
+        {
+            "requirement_id": "persona",
+            "evidence_ref": None,
+            "provider": "external_xhs",
+            "concepts": [],
+        }
+    ]
+    with pytest.raises(ProtocolViolation, match="no evidence reference"):
+        RetrievalPipelineResult.from_payload(payload)
+
+
+@pytest.mark.asyncio
+async def test_expired_lease_reuses_durable_provider_result_without_external_call() -> None:
+    provider = Success()
+    repository = ResumedLeaseRepository(
+        (
+            {
+                "event_type": "provider_result",
+                "provider": "external_xhs",
+                "number": 1,
+                "result": {
+                    "source_refs": ["note:durable"],
+                    "summary": "persisted",
+                    "concepts": [],
+                    "limitations": [],
+                    "partial": False,
+                    "metadata": {},
+                },
+            },
+        )
+    )
+    requirement = EvidenceRequirementV1(
+        "persona",
+        EvidencePurpose.PERSONA,
+        True,
+        1,
+        source_chain=(EvidenceProvider.EXTERNAL_XHS,),
+    )
+
+    result = await RetrievalPipeline(
+        {EvidenceProvider.EXTERNAL_XHS: provider},
+        repository,
+        owner="host-b",
+    ).require(requirement, query={"account": "a"}, snapshot={})
+
+    assert result.evidence_items[0].evidence_ref == "note:durable"
+    assert provider.calls == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid_ref", [None, 123])
+async def test_durable_provider_journal_rejects_invalid_evidence_identity(
+    invalid_ref: object,
+) -> None:
+    provider = Success()
+    repository = ResumedLeaseRepository(
+        (
+            {
+                "event_type": "provider_result",
+                "provider": "external_xhs",
+                "number": 1,
+                "result": {
+                    "source_refs": [invalid_ref],
+                    "summary": "invalid persisted result",
+                },
+            },
+        )
+    )
+    requirement = EvidenceRequirementV1(
+        "persona",
+        EvidencePurpose.PERSONA,
+        True,
+        1,
+        source_chain=(EvidenceProvider.EXTERNAL_XHS,),
+    )
+
+    with pytest.raises(ProtocolViolation, match="no evidence reference"):
+        await RetrievalPipeline(
+            {EvidenceProvider.EXTERNAL_XHS: provider},
+            repository,
+            owner="host-b",
+        ).require(requirement, query={"account": "a"}, snapshot={})
+    assert provider.calls == 0

script_build_host/tests/test_tool_argument_normalization.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from typing import Any
-
 import pytest
 from agent import ToolRegistry
 
@@ -41,13 +39,16 @@ def test_planner_schema_exposes_only_semantic_inputs_and_decision_ids() -> None:
     register_script_tools(registry, object())  # type: ignore[arg-type]
 
     schema = registry.get_schemas(["plan_script_tasks"])[0]
-    contracts = schema["function"]["parameters"]["properties"]["contracts"]
-    contract = contracts["items"]
+    intents = schema["function"]["parameters"]["properties"]["intents"]
+    intent = intents["items"]
 
-    assert contracts["type"] == "array"
-    assert "schema_version" not in contract["properties"]
-    assert "input_decision_ids" in contract["required"]
+    assert intents["type"] == "array"
+    assert "schema_version" not in intent["properties"]
+    assert "objective" in intent["required"]
     for derived in (
+        "parent_task_id",
+        "scope_ref",
+        "intent_class",
         "input_decision_refs",
         "base_artifact_ref",
         "write_scope",
@@ -56,13 +57,9 @@ def test_planner_schema_exposes_only_semantic_inputs_and_decision_ids() -> None:
         "candidate_closure_decision_refs",
         "compose_order",
     ):
-        assert derived not in contract["properties"]
-    assert contract["properties"]["task_kind"]["enum"] == [
+        assert derived not in intent["properties"]
+    assert intent["properties"]["task_kind"]["enum"] == [
         "direction",
-        "pattern-retrieval",
-        "decode-retrieval",
-        "external-retrieval",
-        "knowledge-retrieval",
         "structure",
         "paragraph",
         "element-set",
@@ -70,35 +67,17 @@ def test_planner_schema_exposes_only_semantic_inputs_and_decision_ids() -> None:
         "compose",
         "candidate-portfolio",
     ]
-    assert contract["additionalProperties"] is False
-
-
-@pytest.mark.asyncio
-async def test_retrieval_tool_freezes_and_submits_in_one_terminal_call() -> None:
-    class Gateway:
-        def __init__(self) -> None:
-            self.calls: list[tuple[str, Any]] = []
-
-        async def retrieve(self, source: str, tool: str, query: Any, context: Any) -> Any:
-            self.calls.append(("retrieve", (source, tool, query, context)))
-            return {"artifact_ref": {"uri": "script-build://artifact-versions/1"}}
+    assert intent["additionalProperties"] is False
 
-        async def submit_current_attempt(self, context: Any) -> Any:
-            self.calls.append(("submit", context))
-            return {"attempt_id": "attempt-1", "status": "awaiting_validation"}
 
+def test_physical_retrieval_tools_are_not_registered() -> None:
     from script_build_host.tools.registry import register_script_tools
 
-    gateway = Gateway()
     registry = ToolRegistry()
-    register_script_tools(registry, gateway)  # type: ignore[arg-type]
-    tool = registry._tools["search_knowledge"]["func"]
-
-    result = await tool(keyword="hard tech", max_count=3, context={"task_id": "task-1"})
-
-    assert result.terminate_run is True
-    assert [name for name, _ in gateway.calls] == ["retrieve", "submit"]
-    assert "attempt-1" not in result.output
-    assert "script-build://" not in result.output
-    assert "artifact_handle" in result.output
-    assert '"submitted": true' in result.output
+    register_script_tools(registry, object())  # type: ignore[arg-type]
+    assert {
+        "search_knowledge",
+        "external_search_case",
+        "query_pattern_qa",
+        "search_script_decode_case",
+    }.isdisjoint(registry._tools)

script_build_host/tests/test_tool_failure_bridge.py

@@ -154,13 +154,15 @@ def test_context_read_tool_accepts_only_broker_handles_and_opaque_cursor():
     assert properties["cursor"]["type"] == ["string", "null"]
 
 
-def test_validator_unread_required_context_is_retryable_without_replanning():
+def test_validator_unauthorized_evidence_is_retryable_without_replanning():
     detail = classify_script_tool_failure(
-        ScriptBuildError("CONTEXT_NOT_EXHAUSTED", "read remaining pages"),
+        ScriptBuildError(
+            "VALIDATION_EVIDENCE_UNAUTHORIZED", "read remaining pages"
+        ),
         source_tool="submit_validation",
     )
 
-    assert detail.code == "CONTEXT_NOT_EXHAUSTED"
+    assert detail.code == "VALIDATION_EVIDENCE_UNAUTHORIZED"
     assert detail.disposition is FailureDisposition.RETRY_CALL