script-build-and-framework-tmp/script_build_host/src/script_build_host/tools/gateway.py

"""Application gateway used by all script-build Agent tools.

Agent-visible functions never receive a build, root, task, attempt, or
validation identity from model arguments.  Those values are resolved from the
framework's protected context and then checked against durable bindings.
"""

from __future__ import annotations

import re
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from hashlib import sha256
from typing import Any, Protocol, cast
from unicodedata import normalize
from uuid import uuid4

from agent.orchestration import (
    ArtifactRef,
    AttemptSubmission,
    CriterionResult,
    EvidenceQuery,
    ValidationVerdict,
)

from script_build_host.agents.validation import (
    PHASE_TWO_KINDS,
    RETRIEVAL_KINDS,
    ROOT_DELIVERY_KIND,
    precheck_business_artifact,
    precheck_snapshot,
    task_kind,
)
from script_build_host.domain.artifacts import (
    ArtifactKind,
    DirectionArtifact,
    DirectionConstraint,
    DirectionGoal,
    DirectionPreference,
    EvidenceRecordV1,
)
from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation, ScriptBuildError
from script_build_host.domain.ports import (
    InputSnapshotRepository,
    MissionBindingRepository,
    ScriptBusinessArtifactRepository,
)
from script_build_host.domain.task_contracts import stable_semantic_id
from script_build_host.domain.workbench import (
    protect_model_value,
    semantic_handle,
    strategy_handle,
    strategy_ref,
)
from script_build_host.infrastructure.redaction import redact, redact_text

from .contracts import RootDeliveryToolPort, ScriptCandidateToolPort, ScriptPlannerToolPort

_DEFECT_SEVERITIES = frozenset({"warning", "hard", "critical"})
_DEFECT_ACTIONS = frozenset(
    {"repair", "retry", "revise", "split", "retrieve", "replace", "compare", "block"}
)
_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: ...

    async def execute_fenced(
        self, script_build_id: int, mutation: Callable[[], Awaitable[Any]]
    ) -> Any: ...


class PhaseTwoBudgetGuard(Protocol):
    async def ensure_tool_budget(self, *, root_trace_id: str, task_id: str, entry: str) -> None: ...


class LegacyScriptToolGateway:
    """Translate protected Agent tool calls to typed domain operations."""

    def __init__(
        self,
        *,
        bindings: MissionBindingRepository,
        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,
        budget_guard: PhaseTwoBudgetGuard | None = None,
        root_tools: RootDeliveryToolPort | None = None,
        workbench: Any | None = None,
    ) -> None:
        self.bindings = bindings
        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._active_retrieval_attempts: set[tuple[str, str, str]] = set()

    async def ensure_dispatch_allowed(self, context: Mapping[str, Any]) -> None:
        binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
        if self.dispatch_guard is not None:
            await self.dispatch_guard.ensure_dispatch_allowed(binding.script_build_id)

    async def _execute_fenced(
        self, context: Mapping[str, Any], mutation: Callable[[], Awaitable[Any]]
    ) -> Any:
        binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
        if self.dispatch_guard is None:
            return await mutation()
        return await self.dispatch_guard.execute_fenced(binding.script_build_id, mutation)

    async def plan_script_tasks(
        self,
        *,
        contracts: Sequence[Mapping[str, Any]],
        parent_task_id: str | None,
        context: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        return cast(
            Mapping[str, Any],
            await self._execute_fenced(
                context,
                lambda: self._planner().plan_script_tasks(
                    contract_payloads=contracts,
                    parent_task_id=parent_task_id,
                    context=context,
                ),
            ),
        )

    async def decide_script_task(
        self,
        *,
        task_id: str,
        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(
            Mapping[str, Any],
            await self._execute_fenced(
                context,
                lambda: self._planner().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,
                    context=context,
                ),
            ),
        )

    async def inspect_script_plan(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
        return await self._planner().inspect_script_plan(context=context)

    async def search_mission_context(
        self,
        *,
        collection: str,
        query: str,
        goal_ids: Sequence[str],
        task_kinds: Sequence[str],
        source_types: Sequence[str],
        known_revision: str | None,
        cursor: str | None,
        page_size: int,
        context: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        if self.workbench is None:
            raise ProtocolViolation("Context Broker is not configured")
        return cast(
            Mapping[str, Any],
            await self.workbench.search_mission_context(
                root_trace_id=_required(context, "root_trace_id"),
                role=str(context.get("role") or "agent"),
                task_id=cast(str | None, context.get("task_id")),
                collection=collection,
                query=query,
                goal_ids=goal_ids,
                task_kinds=task_kinds,
                source_types=source_types,
                known_revision=known_revision,
                cursor=cursor,
                page_size=page_size,
            ),
        )

    async def read_mission_context(
        self,
        *,
        handle: str,
        section: str,
        cursor: str | None,
        context: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        if self.workbench is None:
            raise ProtocolViolation("Context Broker is not configured")
        return cast(
            Mapping[str, Any],
            await self.workbench.read_mission_context(
                root_trace_id=_required(context, "root_trace_id"),
                role=str(context.get("role") or "agent"),
                task_id=cast(str | None, context.get("task_id")),
                attempt_id=cast(str | None, context.get("attempt_id")),
                handle=handle,
                section=section,
                cursor=cursor,
            ),
        )

    async def dispatch_script_tasks(
        self, *, task_ids: Sequence[str], context: Mapping[str, Any]
    ) -> Sequence[Mapping[str, Any]]:
        # Dispatch waits for the durable Operation (Worker + Validator) to
        # finish. Holding the binding row lock across that wait deadlocks the
        # child submissions, which must independently pass the same fence.
        # The planning service performs a short owner/fence verification
        # immediately before it reserves the Operation.
        return await self._planner().dispatch_script_tasks(task_ids=task_ids, context=context)

    async def read_accepted_portfolio(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
        return await self._root_tools().read_accepted_portfolio(context=context)

    async def legacy_projection_dry_run(
        self, build_summary: str, context: Mapping[str, Any]
    ) -> Mapping[str, Any]:
        return await self._root_tools().legacy_projection_dry_run(
            build_summary=build_summary, context=context
        )

    async def save_root_delivery_manifest(
        self,
        *,
        build_summary: str,
        blocking_defects: Sequence[str],
        context: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        return cast(
            Mapping[str, Any],
            await self._execute_fenced(
                context,
                lambda: self._root_tools().save_root_delivery_manifest(
                    build_summary=build_summary,
                    blocking_defects=blocking_defects,
                    context=context,
                ),
            ),
        )

    async def read_accepted_artifacts(self, context: Mapping[str, Any]) -> list[dict[str, Any]]:
        binding, _ = await self._scope(context)
        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
        attempt_id = _required(context, "attempt_id")
        attempt = ledger.attempts.get(attempt_id)
        if attempt is None or attempt.task_id != _required(context, "task_id"):
            raise ProtocolViolation("attempt is outside the protected task context")
        decision_ids = attempt.accepted_child_decision_ids
        if decision_ids is None:
            raise ProtocolViolation("attempt has no frozen child decision binding")
        values: list[dict[str, Any]] = []
        for decision_id in decision_ids:
            decision = ledger.decisions.get(decision_id)
            if (
                decision is None
                or decision.action.value != "accept"
                or not decision.attempt_id
                or not decision.validation_id
            ):
                raise ProtocolViolation("accepted child decision binding is invalid")
            child_attempt = ledger.attempts[decision.attempt_id]
            child_task = ledger.tasks[child_attempt.task_id]
            validation = ledger.validations.get(decision.validation_id)
            child_kind = task_kind(child_task.current_spec.context_refs)
            if (
                child_task.parent_task_id != attempt.task_id
                or child_kind not in RETRIEVAL_KINDS
                or decision.task_id != child_task.task_id
                or validation is None
                or validation.attempt_id != child_attempt.attempt_id
                or validation.verdict is None
                or validation.verdict.value != "passed"
                or child_attempt.snapshot_id != validation.snapshot_id
            ):
                raise ProtocolViolation(
                    "accepted decision is not closed over a passed direct-child snapshot"
                )
            if child_attempt.submission is None:
                raise ProtocolViolation("accepted child attempt has no submission")
            refs = [
                *child_attempt.submission.artifact_refs,
                *child_attempt.submission.evidence_refs,
            ]
            for ref in refs:
                if ref.kind != ArtifactKind.EVIDENCE.value:
                    raise ProtocolViolation(
                        "direction child decisions may expose only evidence artifacts"
                    )
                version = await self.artifacts.read_by_ref(
                    ref,
                    script_build_id=binding.script_build_id,
                    task_id=child_attempt.task_id,
                    attempt_id=child_attempt.attempt_id,
                )
                values.append(
                    {
                        "decision_id": decision_id,
                        "task_id": child_attempt.task_id,
                        "attempt_id": child_attempt.attempt_id,
                        "artifact_ref": _ref_dict(ref),
                        "artifact": _json_values(asdict(version.artifact)),
                    }
                )
        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)
        if retrieval_key in self._active_retrieval_attempts:
            raise ProtocolViolation("one Retrieval Attempt may execute only one external query")
        binding = await self.bindings.get_by_root(root_trace_id)
        try:
            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:
            raise ProtocolViolation("one Retrieval Attempt may freeze only one Evidence artifact")
        self._active_retrieval_attempts.add(retrieval_key)
        try:
            return await self._retrieve_once(source_type, tool_name, query, context)
        finally:
            self._active_retrieval_attempts.discard(retrieval_key)

    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_handle": semantic_handle(
                "evidence",
                _required(context, "root_trace_id"),
                _required(context, "task_id"),
                ref.digest or ref.version or "",
            ),
            "evidence": protect_model_value(
                _json_values(asdict(version.artifact)),
                namespace=_required(context, "attempt_id"),
            ),
            "source_count": len(result.source_refs),
        }

    async def view_frozen_images(
        self, image_handles: Sequence[str], context: Mapping[str, Any]
    ) -> Sequence[Mapping[str, Any]]:
        return await self._candidates().view_frozen_images(
            image_handles=image_handles,
            context=context,
        )

    async def candidate_command(
        self,
        operation: str,
        payload: Mapping[str, Any] | Sequence[Mapping[str, Any]],
        context: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        await self.ensure_dispatch_allowed(context)
        tools = self._candidates()
        if operation == "save_script_paragraphs":
            batch = _mapping_payload(payload)
            return await tools.save_script_paragraphs(
                paragraphs=_sequence_payload(batch.get("paragraphs", ())),
                expected_state_revision=str(batch.get("expected_state_revision", "")),
                context=context,
            )
        if operation == "save_script_elements":
            batch = _mapping_payload(payload)
            return await tools.save_script_elements(
                elements=_sequence_payload(batch.get("elements", ())),
                links=_sequence_payload(batch.get("links", ())),
                expected_state_revision=str(batch.get("expected_state_revision", "")),
                context=context,
            )
        if operation == "save_comparison_candidate":
            return await tools.save_comparison_candidate(
                payload=_mapping_payload(payload), context=context
            )
        if operation == "save_candidate_portfolio":
            return await tools.save_candidate_portfolio(
                payload=_mapping_payload(payload), context=context
            )
        raise ProtocolViolation(f"unsupported candidate operation: {operation}")

    async def save_structured_script_candidate(
        self, acceptance_notes: Sequence[str], context: Mapping[str, Any]
    ) -> Mapping[str, Any]:
        await self.ensure_dispatch_allowed(context)
        notes = tuple(_bounded_text(item, "acceptance note", 500) for item in acceptance_notes)
        if len(notes) > 16:
            raise ProtocolViolation("a StructuredScript may contain at most 16 acceptance notes")
        return await self._candidates().save_structured_script_candidate(
            acceptance_notes=notes,
            context=context,
        )

    async def submit_current_attempt(self, context: Mapping[str, Any]) -> Mapping[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="submit",
            )
        root_trace_id = _required(context, "root_trace_id")
        task_id = _required(context, "task_id")
        attempt_id = _required(context, "attempt_id")
        ledger = await self.coordinator.task_store.load(root_trace_id)
        task = ledger.tasks.get(task_id)
        if task is None:
            raise ProtocolViolation("attempt task is outside the protected mission")
        if task_kind(task.current_spec.context_refs) == ROOT_DELIVERY_KIND:
            binding = await self.bindings.get_by_root(root_trace_id)
            root_version = await self.artifacts.get_by_attempt(
                script_build_id=binding.script_build_id,
                task_id=task_id,
                attempt_id=attempt_id,
            )
            if root_version.artifact_type is not ArtifactKind.ROOT_DELIVERY_MANIFEST:
                raise ProtocolViolation("Root Attempt does not own a delivery manifest")
            ref = ArtifactRef(
                uri=(f"script-build://artifact-versions/{root_version.artifact_version_id}"),
                kind=root_version.artifact_type.value,
                version=str(root_version.artifact_version_id),
                digest=root_version.canonical_sha256,
            )
            scope_ref = "script-build://scope/root"
        else:
            manifest = await self._candidates().resolve_attempt_manifest(context=context)
            ref = manifest.artifact_ref
            scope_ref = manifest.scope_ref
        if ref.kind not in _BUSINESS_ARTIFACT_KINDS or not ref.digest:
            raise ProtocolViolation("attempt manifest must contain one frozen business artifact")
        binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
        version = await self.artifacts.read_by_ref(
            ref,
            script_build_id=binding.script_build_id,
            task_id=_required(context, "task_id"),
            attempt_id=_required(context, "attempt_id"),
        )
        if (
            version.artifact_type.value != ref.kind
            or version.canonical_sha256 != ref.digest
            or version.state.value != "frozen"
        ):
            raise ProtocolViolation("attempt artifact ownership or frozen digest is invalid")
        safe_ref = ArtifactRef(
            uri=ref.uri,
            kind=ref.kind,
            version=ref.version,
            digest=ref.digest,
        )
        summary = _attempt_summary(safe_ref, scope_ref)
        submission = AttemptSubmission(
            summary=summary,
            artifact_refs=([] if safe_ref.kind == ArtifactKind.EVIDENCE.value else [safe_ref]),
            evidence_refs=([safe_ref] if safe_ref.kind == ArtifactKind.EVIDENCE.value else []),
        )
        return cast(
            dict[str, Any],
            await self._execute_fenced(
                context,
                lambda: self.coordinator.submit_attempt(dict(context), submission),
            ),
        )

    async def submit_structured_validation(
        self,
        *,
        verdict: str,
        criterion_results: Sequence[Mapping[str, Any]],
        defects: Sequence[Mapping[str, Any]],
        recommendation: str,
        context: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        root_trace_id = _required(context, "root_trace_id")
        task_id = _required(context, "task_id")
        snapshot_id = _required(context, "snapshot_id")
        ledger = await self.coordinator.task_store.load(root_trace_id)
        task = ledger.tasks.get(task_id)
        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()}
        normalized_defects = _normalize_defects(
            [_expand_validation_handles(item, refs_by_handle) for item in defects],
            allowed_refs,
        )
        expected_ids = {item.criterion_id for item in task.current_spec.acceptance_criteria}
        if any(item["criterion_id"] not in expected_ids for item in normalized_defects):
            raise ProtocolViolation("validation defect references an unknown Task criterion")
        result_by_id: dict[str, CriterionResult] = {}
        for raw in criterion_results:
            criterion_id = _bounded_text(raw.get("criterion_id"), "criterion_id", 128)
            if criterion_id not in expected_ids or criterion_id in result_by_id:
                raise ProtocolViolation("validation criterion identity is unknown or duplicated")
            criterion_verdict = ValidationVerdict(str(raw.get("verdict", "")))
            reason = _bounded_text(raw.get("reason"), "criterion reason", 300)
            defect_codes = tuple(
                item["defect_code"]
                for item in normalized_defects
                if item["criterion_id"] == criterion_id
            )
            if defect_codes:
                reason = f"{reason}; defect_codes={','.join(defect_codes)}"[:500]
            evidence_handles = raw.get("evidence_handle_ids", ())
            if not isinstance(evidence_handles, list) or len(evidence_handles) > 64:
                raise ProtocolViolation("criterion evidence handles are invalid")
            evidence = _normalize_evidence_refs(
                [
                    _ref_dict(_required_evidence_handle(value, refs_by_handle))
                    for value in evidence_handles
                ],
                allowed_refs,
                label="criterion evidence_handle_ids",
            )
            if criterion_verdict is ValidationVerdict.PASSED and not evidence:
                raise ProtocolViolation("passed criterion requires frozen evidence_refs")
            result_by_id[criterion_id] = CriterionResult(
                criterion_id=criterion_id,
                verdict=criterion_verdict,
                reason=reason,
                evidence_refs=list(evidence),
            )
        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")
        criterion_verdicts = {item.verdict for item in result_by_id.values()}
        every_criterion_passed = criterion_verdicts == {ValidationVerdict.PASSED}
        if (overall is ValidationVerdict.PASSED) != every_criterion_passed:
            raise ProtocolViolation("overall validation verdict must match all criterion verdicts")
        from agent.orchestration.validation_policy import ValidationContext

        attempt = ledger.attempts.get(_required(context, "attempt_id"))
        if attempt is None or attempt.task_id != task_id:
            raise ProtocolViolation("validation Attempt is outside the protected Task")
        validation_context = ValidationContext(
            root_trace_id=root_trace_id,
            task=task,
            attempt=attempt,
            snapshot=snapshot,
            prior_validation_count=max(0, len(task.validation_ids) - 1),
        )
        deterministic = precheck_snapshot(validation_context)
        if overall is ValidationVerdict.PASSED and any(
            item.verdict is not ValidationVerdict.PASSED for item in deterministic
        ):
            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
        ):
            binding = await self.bindings.get_by_root(root_trace_id)
            for ref in snapshot.artifact_refs:
                version = await self.artifacts.read_by_ref(
                    ref,
                    script_build_id=binding.script_build_id,
                    task_id=task_id,
                    attempt_id=attempt.attempt_id,
                )
                if precheck_business_artifact(version.artifact):
                    raise ProtocolViolation(
                        "passed validation conflicts with unrealized placeholder content"
                    )
            business_rules = await self._candidates().deterministic_precheck(context=context)
            if any(item.get("verdict") != "passed" for item in business_rules):
                raise ProtocolViolation(
                    "passed validation conflicts with phase-two deterministic precheck"
                )
        if (
            overall is ValidationVerdict.PASSED
            and task_kind(task.current_spec.context_refs) == ROOT_DELIVERY_KIND
        ):
            root_rules = await self._root_tools().deterministic_precheck(context=context)
            if any(item.get("verdict") != "passed" for item in root_rules):
                raise ProtocolViolation(
                    "passed validation conflicts with Root deterministic precheck"
                )
        if overall is not ValidationVerdict.PASSED and not normalized_defects:
            raise ProtocolViolation("failed or inconclusive validation must report a defect")
        if any(
            result_by_id[item["criterion_id"]].verdict is ValidationVerdict.PASSED
            for item in blocking
        ):
            raise ProtocolViolation("blocking defect conflicts with a passed criterion")
        codes = tuple(dict.fromkeys(item["defect_code"] for item in normalized_defects))
        scopes = tuple(dict.fromkeys(item["scope_ref"] for item in normalized_defects))
        action_classes = tuple(
            dict.fromkeys(item["recommended_action_class"] for item in normalized_defects)
        )
        summary = (
            f"validation={overall.value};criteria={len(result_by_id)};"
            f"defects={len(normalized_defects)};codes={','.join(codes) or 'none'};"
            f"scopes={','.join(scopes) or 'none'};"
            f"actions={','.join(action_classes) or 'none'}"
        )[:500]
        evidence_refs = _dedupe_refs(
            ref for item in normalized_defects for ref in item["evidence_refs"]
        )
        bounded_recommendation = _bounded_text(
            recommendation or (action_classes[0] if action_classes else "accept"),
            "recommendation",
            200,
        )
        return cast(
            dict[str, Any],
            await self._execute_fenced(
                context,
                lambda: self.coordinator.submit_validation(
                    dict(context),
                    overall,
                    tuple(result_by_id.values()),
                    summary,
                    evidence_refs,
                    (),
                    codes[:50],
                    bounded_recommendation,
                ),
            ),
        )

    async def save_direction_candidate(
        self,
        *,
        goals: Sequence[Mapping[str, Any]],
        constraints: Sequence[Mapping[str, Any]],
        preferences: Sequence[Mapping[str, Any]],
        strategy_handles: Sequence[str],
        evidence_decision_ids: Sequence[str],
        context: Mapping[str, Any],
    ) -> dict[str, Any]:
        await self.ensure_dispatch_allowed(context)
        binding, snapshot = await self._scope(context)
        accepted = await self.read_accepted_artifacts(context)
        evidence_by_decision = {
            str(item["decision_id"]): str(item["artifact_ref"]["uri"]) for item in accepted
        }
        if len(set(evidence_decision_ids)) != len(evidence_decision_ids) or any(
            item not in evidence_by_decision for item in evidence_decision_ids
        ):
            raise ProtocolViolation(
                "direction references evidence outside frozen child ACCEPT decisions"
            )
        task_id = _required(context, "task_id")
        direction_goals, goal_ids_by_client_key = _direction_goals(task_id, goals)
        strategy_by_handle = {
            strategy_handle(snapshot.snapshot_id, index, item): strategy_ref(
                snapshot.snapshot_id, index, item
            )
            for index, item in enumerate(snapshot.strategies)
        }
        if len(set(strategy_handles)) != len(strategy_handles) or any(
            item not in strategy_by_handle for item in strategy_handles
        ):
            raise ProtocolViolation("direction contains an unknown strategy handle")
        artifact = DirectionArtifact(
            goals=direction_goals,
            constraints=tuple(
                DirectionConstraint(
                    constraint_id=stable_semantic_id(
                        task_id,
                        "constraint",
                        _normalized_client_key(item.get("client_key"), "constraint client_key"),
                    ),
                    statement=redact_text(str(item["statement"])),
                    rationale=redact_text(str(item.get("rationale", ""))),
                )
                for item in constraints
            ),
            preferences=tuple(
                DirectionPreference(
                    preference_id=stable_semantic_id(
                        task_id,
                        "preference",
                        _normalized_client_key(item.get("client_key"), "preference client_key"),
                    ),
                    statement=redact_text(str(item["statement"])),
                    rationale=redact_text(str(item.get("rationale", ""))),
                    priority=(int(item["priority"]) if item.get("priority") is not None else None),
                )
                for item in preferences
            ),
            topic_refs=(f"script-build://inputs/{snapshot.snapshot_id}/topic",),
            persona_refs=(f"script-build://inputs/{snapshot.snapshot_id}/persona",),
            strategy_refs=tuple(strategy_by_handle[item] for item in strategy_handles),
            evidence_refs=tuple(evidence_by_decision[item] for item in evidence_decision_ids),
        )
        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=artifact,
        )
        return {
            "direction_handle": semantic_handle(
                "direction",
                _required(context, "root_trace_id"),
                _required(context, "task_id"),
                ref.digest or ref.version or "",
            ),
            "direction": protect_model_value(
                _json_values(version.artifact.content_payload()),
                namespace=_required(context, "attempt_id"),
            ),
            "goal_ids_by_client_key": goal_ids_by_client_key,
        }

    async def query_validation_evidence(
        self,
        query: str,
        limit: int,
        context: Mapping[str, Any],
    ) -> dict[str, Any]:
        request = EvidenceQuery(
            root_trace_id=_required(context, "root_trace_id"),
            task_id=_required(context, "task_id"),
            attempt_id=_required(context, "attempt_id"),
            snapshot_id=_required(context, "snapshot_id"),
            validation_id=_required(context, "validation_id"),
            query=query,
            limit=limit,
        )
        response = await self.coordinator.query_evidence(dict(context), request)
        payload = cast(dict[str, Any], _json_values(asdict(response)))
        refs = tuple(response.evidence_refs)
        payload.pop("evidence_refs", None)
        payload["evidence_handles"] = [
            {
                "evidence_handle": _evidence_handle(ref),
                "kind": ref.kind,
                "summary": ref.summary,
            }
            for ref in refs
        ]
        payload["items"] = protect_model_value(
            payload.get("items", []),
            namespace=f"validation:{request.validation_id}",
        )
        return payload

    async def deterministic_precheck(self, context: Mapping[str, Any]) -> dict[str, Any]:
        from agent.orchestration.validation_policy import ValidationContext

        from script_build_host.agents.validation import (
            PHASE_TWO_KINDS,
            precheck_snapshot,
            task_kind,
        )

        root_trace_id = _required(context, "root_trace_id")
        ledger = await self.coordinator.task_store.load(root_trace_id)
        task = ledger.tasks[_required(context, "task_id")]
        attempt = ledger.attempts[_required(context, "attempt_id")]
        snapshot = await self.coordinator.artifact_store.get(
            root_trace_id, _required(context, "snapshot_id")
        )
        validation_context = ValidationContext(
            root_trace_id=root_trace_id,
            task=task,
            attempt=attempt,
            snapshot=snapshot,
            prior_validation_count=max(0, len(task.validation_ids) - 1),
        )
        rules: list[Mapping[str, Any]] = [
            item.to_dict() for item in precheck_snapshot(validation_context)
        ]
        kind = task_kind(task.current_spec.context_refs)
        if kind in PHASE_TWO_KINDS:
            rules.extend(await self._candidates().deterministic_precheck(context=context))
        elif kind == "root-delivery":
            rules.extend(await self._root_tools().deterministic_precheck(context=context))
        return {"rule_results": rules}

    async def _scope(self, context: Mapping[str, Any]) -> tuple[Any, Any]:
        root_trace_id = _required(context, "root_trace_id")
        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,
        )
        return binding, snapshot

    def _planner(self) -> ScriptPlannerToolPort:
        if self.planner_tools is None:
            raise ProtocolViolation("script planner tools are not configured")
        return self.planner_tools

    def _candidates(self) -> ScriptCandidateToolPort:
        if self.candidate_tools is None:
            raise ProtocolViolation("script candidate tools are not configured")
        return self.candidate_tools

    def _root_tools(self) -> RootDeliveryToolPort:
        if self.root_tools is None:
            raise ProtocolViolation("script Root delivery tools are not configured")
        return self.root_tools


def _required(context: Mapping[str, Any], key: str) -> str:
    value = context.get(key)
    if not isinstance(value, str) or not value.strip():
        raise ProtocolViolation(f"protected context is missing {key}")
    return value


def _required_int(context: Mapping[str, Any], key: str) -> int:
    value = context.get(key)
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ProtocolViolation(f"protected context is missing {key}")
    return value


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


def _json_values(value: Any) -> Any:
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, Mapping):
        return {str(key): _json_values(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_json_values(item) for item in value]
    if hasattr(value, "value"):
        return value.value
    return value


def _safe_text_tuple(values: Sequence[str]) -> tuple[str, ...]:
    normalized: list[str] = []
    seen: set[str] = set()
    for value in values:
        safe_value = redact_text(str(value))
        if safe_value in seen:
            continue
        seen.add(safe_value)
        normalized.append(safe_value)
    return tuple(normalized)


def _mapping_payload(
    value: Mapping[str, Any] | Sequence[Mapping[str, Any]],
) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise ProtocolViolation("candidate command expects an object payload")
    return value


def _sequence_payload(
    value: Mapping[str, Any] | Sequence[Mapping[str, Any]],
) -> Sequence[Mapping[str, Any]]:
    if isinstance(value, Mapping) or isinstance(value, (str, bytes)):
        raise ProtocolViolation("candidate batch command expects an array payload")
    if any(not isinstance(item, Mapping) for item in value):
        raise ProtocolViolation("candidate batch entries must be objects")
    return value


def _bounded_text(value: Any, label: str, maximum: int) -> str:
    if not isinstance(value, str) or not value.strip() or len(value) > maximum:
        raise ProtocolViolation(f"{label} must contain between 1 and {maximum} characters")
    return redact_text(value.strip())


def _normalized_client_key(value: Any, label: str) -> str:
    text = _bounded_text(value, label, 128)
    normalized = " ".join(normalize("NFKC", text).split()).casefold()
    if not normalized or len(normalized) > 128:
        raise ProtocolViolation(f"{label} must contain between 1 and 128 characters")
    return normalized


def _direction_goals(
    task_id: str, goals: Sequence[Mapping[str, Any]]
) -> tuple[tuple[DirectionGoal, ...], dict[str, str]]:
    if not 1 <= len(goals) <= 30:
        raise ValueError("direction requires between one and thirty Goal nodes")
    normalized: list[tuple[str, str | None, Mapping[str, Any]]] = []
    for item in goals:
        client_key = _normalized_client_key(item.get("client_key"), "goal client_key")
        parent_key = (
            _normalized_client_key(item.get("parent_client_key"), "goal parent_client_key")
            if item.get("parent_client_key") is not None
            else None
        )
        normalized.append((client_key, parent_key, item))
    keys = [item[0] for item in normalized]
    if len(set(keys)) != len(keys):
        raise ValueError("goal client_key values must be unique")
    key_set = set(keys)
    for client_key, parent_key, _ in normalized:
        if parent_key == client_key:
            raise ValueError("a goal cannot parent itself")
        if parent_key is not None and parent_key not in key_set:
            raise ValueError(f"goal parent_client_key does not exist: {parent_key}")
    parent_by_key = {key: parent for key, parent, _ in normalized}
    if any(
        parent is not None and parent_by_key.get(parent) is not None for _, parent, _ in normalized
    ):
        raise ValueError("direction goal hierarchy may contain only two levels")
    goal_ids = {
        key: "goal-" + sha256(f"{task_id}\0{key}".encode()).hexdigest()[:16] for key in keys
    }
    if len(set(goal_ids.values())) != len(goal_ids):
        raise ValueError("generated goal identity collision")
    return (
        tuple(
            DirectionGoal(
                goal_id=goal_ids[client_key],
                statement=redact_text(str(item["statement"])),
                rationale=redact_text(str(item["rationale"])),
                parent_goal_id=goal_ids[parent_key] if parent_key is not None else None,
                success_criteria=tuple(
                    redact_text(str(value)) for value in item["success_criteria"]
                ),
            )
            for client_key, parent_key, item in normalized
        ),
        goal_ids,
    )


def _attempt_summary(ref: ArtifactRef, scope_ref: str) -> str:
    scope = _bounded_text(scope_ref, "scope_ref", 300)
    if not scope.startswith("script-build://"):
        raise ProtocolViolation("scope_ref must use the script-build namespace")
    digest = _bounded_text(ref.digest, "artifact digest", 71)
    return f"kind={ref.kind};scope={scope};digest={digest};status=frozen"[:500]


def _evidence_handle(ref: ArtifactRef) -> str:
    return semantic_handle("src", ref.uri, ref.digest or ref.version or "")


def _required_evidence_handle(
    value: object, refs_by_handle: Mapping[str, ArtifactRef]
) -> ArtifactRef:
    handle = str(value or "").strip()
    ref = refs_by_handle.get(handle)
    if ref is None:
        raise ProtocolViolation("validation references an unknown evidence handle")
    return ref


def _expand_validation_handles(
    value: Mapping[str, Any], refs_by_handle: Mapping[str, ArtifactRef]
) -> dict[str, Any]:
    unknown = set(value) - {
        "defect_code",
        "criterion_id",
        "scope_selector",
        "observed_excerpt",
        "evidence_handle_ids",
        "severity",
        "invalidated_inputs",
        "recommended_action_class",
    }
    if unknown:
        raise ProtocolViolation("validation defect contains unsupported fields")
    selector = value.get("scope_selector")
    if not isinstance(selector, Mapping) or selector.get("anchor") != "mission":
        raise ProtocolViolation("validation defect scope_selector is invalid")
    path = selector.get("path")
    if (
        not isinstance(path, list)
        or not 1 <= len(path) <= 4
        or any(not isinstance(item, str) or not item.strip() for item in path)
    ):
        raise ProtocolViolation("validation defect scope_selector path is invalid")
    handles = value.get("evidence_handle_ids", ())
    if not isinstance(handles, list) or len(handles) > 64:
        raise ProtocolViolation("validation defect evidence handles are invalid")
    return {
        **{
            key: item
            for key, item in value.items()
            if key not in {"scope_selector", "evidence_handle_ids"}
        },
        "scope_ref": "script-build://scopes/" + "/".join(path),
        "evidence_refs": [
            _ref_dict(_required_evidence_handle(item, refs_by_handle)) for item in handles
        ],
    }


def _normalize_defects(
    defects: Sequence[Mapping[str, Any]],
    allowed_refs: Mapping[tuple[Any, Any, Any, Any], ArtifactRef],
) -> tuple[dict[str, Any], ...]:
    if len(defects) > 50:
        raise ProtocolViolation("a validation may report at most 50 defects")
    values: list[dict[str, Any]] = []
    for raw in defects:
        if not isinstance(raw, Mapping):
            raise ProtocolViolation("each validation defect must be an object")
        code = _bounded_text(raw.get("defect_code"), "defect_code", 64)
        if not _DEFECT_CODE.fullmatch(code):
            raise ProtocolViolation("defect_code must be an uppercase stable identifier")
        criterion_id = _bounded_text(raw.get("criterion_id"), "criterion_id", 128)
        scope_ref = _bounded_text(raw.get("scope_ref"), "scope_ref", 500)
        if not scope_ref.startswith("script-build://"):
            raise ProtocolViolation("defect scope_ref must use script-build://")
        excerpt = _bounded_text(raw.get("observed_excerpt"), "observed_excerpt", 500)
        severity = str(raw.get("severity", ""))
        if severity not in _DEFECT_SEVERITIES:
            raise ProtocolViolation("defect severity must be warning, hard, or critical")
        action = str(raw.get("recommended_action_class", ""))
        if action not in _DEFECT_ACTIONS:
            raise ProtocolViolation("defect recommended action class is unsupported")
        invalidated = raw.get("invalidated_inputs", [])
        if (
            not isinstance(invalidated, list)
            or len(invalidated) > 64
            or any(not isinstance(item, str) or not item.strip() for item in invalidated)
        ):
            raise ProtocolViolation("invalidated_inputs must be a bounded string array")
        refs = _normalize_evidence_refs(
            raw.get("evidence_refs"), allowed_refs, label="defect evidence_refs"
        )
        values.append(
            {
                "defect_code": code,
                "criterion_id": criterion_id,
                "scope_ref": scope_ref,
                "observed_excerpt": excerpt,
                "evidence_refs": refs,
                "severity": severity,
                "invalidated_inputs": tuple(str(item) for item in invalidated),
                "recommended_action_class": action,
            }
        )
    return tuple(values)


def _normalize_evidence_refs(
    value: Any,
    allowed_refs: Mapping[tuple[Any, Any, Any, Any], ArtifactRef],
    *,
    label: str,
) -> tuple[ArtifactRef, ...]:
    if not isinstance(value, list) or len(value) > 64:
        raise ProtocolViolation(f"{label} must be a bounded array")
    refs: list[ArtifactRef] = []
    for item in value:
        if not isinstance(item, Mapping):
            raise ProtocolViolation(f"{label} entries must be objects")
        candidate = ArtifactRef.from_dict(dict(item))
        key = (candidate.uri, candidate.version, candidate.digest, candidate.kind)
        frozen = allowed_refs.get(key)
        if frozen is None:
            raise ProtocolViolation(f"{label} must come from the protected validation closure")
        refs.append(frozen)
    return _dedupe_refs(refs)


def _dedupe_refs(values: Any) -> tuple[ArtifactRef, ...]:
    result: list[ArtifactRef] = []
    seen: set[tuple[Any, Any, Any, Any]] = set()
    for ref in values:
        key = (ref.uri, ref.version, ref.digest, ref.kind)
        if key in seen:
            continue
        seen.add(key)
        result.append(
            ArtifactRef(
                uri=ref.uri,
                kind=ref.kind,
                version=ref.version,
                digest=ref.digest,
            )
        )
    return tuple(result)


__all__ = [
    "DispatchGuard",
    "LegacyScriptToolGateway",
    "RetrievalAdapter",
    "RetrievalResult",
    "ScriptCandidateToolPort",
    "ScriptPlannerToolPort",
]