领域：定义阶段二任务合同、候选制品与结构化缺陷

script_build_host/src/script_build_host/domain/artifacts.py

@@ -5,10 +5,25 @@ from datetime import datetime
 from enum import StrEnum
 from typing import Any
 
+from .phase_two_artifacts import (
+    CandidatePortfolioArtifactV1,
+    ComparisonArtifactV1,
+    ElementSetArtifactV1,
+    ParagraphArtifactV1,
+    StructureArtifactV1,
+    StructuredScriptArtifactV1,
+)
+
 
 class ArtifactKind(StrEnum):
     EVIDENCE = "evidence"
     DIRECTION = "direction"
+    STRUCTURE = "structure"
+    PARAGRAPH = "paragraph"
+    ELEMENT_SET = "element_set"
+    COMPARISON = "comparison"
+    STRUCTURED_SCRIPT = "structured_script"
+    CANDIDATE_PORTFOLIO = "candidate_portfolio"
 
 
 class ArtifactState(StrEnum):
@@ -155,7 +170,16 @@ class EvidenceRecordV1:
         }
 
 
-BusinessArtifact = EvidenceRecordV1 | ScriptDirectionArtifactV1
+BusinessArtifact = (
+    EvidenceRecordV1
+    | ScriptDirectionArtifactV1
+    | StructureArtifactV1
+    | ParagraphArtifactV1
+    | ElementSetArtifactV1
+    | ComparisonArtifactV1
+    | StructuredScriptArtifactV1
+    | CandidatePortfolioArtifactV1
+)
 
 
 @dataclass(frozen=True, slots=True)

script_build_host/src/script_build_host/domain/candidate_validation.py

@@ -0,0 +1,71 @@
+"""Framework-free deterministic checks for frozen creative artifacts."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from typing import Any
+
+from .phase_two_artifacts import ComparisonArtifactV1
+
+_PLACEHOLDER = re.compile(
+    r"(?:TODO|TBD|\[\s*待补|<placeholder>|后续(?:应|需|将)产出|"
+    r"此处插入|此处(?:描述|说明|展示|呈现).{0,12}(?:存在|内容|信息)?)",
+    re.IGNORECASE,
+)
+
+
+def placeholder_paths(artifact: Any) -> tuple[str, ...]:
+    """Return exact JSON paths that still contain unrealized prose."""
+
+    payload = artifact.content_payload()
+    findings: list[str] = []
+
+    def walk(value: Any, path: str) -> None:
+        if isinstance(value, str) and _PLACEHOLDER.search(value):
+            findings.append(path)
+        elif isinstance(value, Mapping):
+            for key, item in value.items():
+                walk(item, f"{path}.{key}")
+        elif isinstance(value, (list, tuple)):
+            for index, item in enumerate(value):
+                walk(item, f"{path}[{index}]")
+
+    walk(payload, "artifact")
+    return tuple(findings)
+
+
+def comparison_fairness_errors(
+    artifact: ComparisonArtifactV1,
+    expected_criterion_ids: set[str] | None = None,
+) -> tuple[str, ...]:
+    """Require every comparison criterion to assess every frozen candidate."""
+
+    expected = set(artifact.candidate_artifact_refs)
+    errors: list[str] = []
+    criterion_ids: set[str] = set()
+    for index, criterion in enumerate(artifact.criterion_results):
+        criterion_id = criterion.get("criterion_id")
+        if not isinstance(criterion_id, str) or not criterion_id.strip():
+            errors.append(f"criterion_results[{index}].criterion_id")
+        elif criterion_id in criterion_ids:
+            errors.append(f"criterion_results[{index}].criterion_id:duplicate")
+        else:
+            criterion_ids.add(criterion_id)
+        results = criterion.get("candidate_results")
+        if not isinstance(results, list) or any(not isinstance(item, Mapping) for item in results):
+            errors.append(f"criterion_results[{index}].candidate_results")
+            continue
+        refs = [item.get("artifact_ref") for item in results]
+        if len(refs) != len(set(refs)) or set(refs) != expected:
+            errors.append(f"criterion_results[{index}].candidate_results:asymmetric")
+        if any(not str(item.get("reason", "")).strip() for item in results):
+            errors.append(f"criterion_results[{index}].candidate_results:reason")
+    if expected_criterion_ids is not None and criterion_ids != expected_criterion_ids:
+        errors.append("criterion_results:criterion-set")
+    if artifact.recommendation not in expected:
+        errors.append("recommendation")
+    return tuple(errors)
+
+
+__all__ = ["comparison_fairness_errors", "placeholder_paths"]

script_build_host/src/script_build_host/domain/canonical_json.py

@@ -0,0 +1,79 @@
+"""Framework-free canonical JSON used by domain and application services."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from dataclasses import asdict, is_dataclass
+from datetime import UTC, date, datetime
+from decimal import Decimal
+from enum import Enum
+from typing import Any
+
+from .digests import Sha256Digest
+from .errors import ProtocolViolation
+
+
+def _stable_decimal(value: Decimal) -> str:
+    if not value.is_finite():
+        raise ProtocolViolation("canonical JSON does not permit non-finite numbers")
+    rendered = format(value, "f")
+    if "." in rendered:
+        rendered = rendered.rstrip("0").rstrip(".")
+    return "0" if rendered in {"-0", ""} else rendered
+
+
+def normalize_json(value: Any) -> Any:
+    """Return a JSON value with deterministic dates, floats, keys and containers."""
+
+    if value is None or isinstance(value, (str, bool, int)):
+        return value
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            raise ProtocolViolation("canonical JSON does not permit non-finite numbers")
+        return _stable_decimal(Decimal(str(value)))
+    if isinstance(value, Decimal):
+        return _stable_decimal(value)
+    if isinstance(value, datetime):
+        if value.tzinfo is None:
+            return value.isoformat(timespec="microseconds")
+        normalized = value.astimezone(UTC).isoformat(timespec="microseconds")
+        return normalized.removesuffix("+00:00") + "Z"
+    if isinstance(value, date):
+        return value.isoformat()
+    if isinstance(value, Enum):
+        return normalize_json(value.value)
+    if is_dataclass(value) and not isinstance(value, type):
+        return normalize_json(asdict(value))
+    if isinstance(value, dict):
+        output: dict[str, Any] = {}
+        for key in sorted(value, key=lambda item: str(item)):
+            if not isinstance(key, str):
+                raise ProtocolViolation("canonical JSON object keys must be strings")
+            output[key] = normalize_json(value[key])
+        return output
+    if isinstance(value, (list, tuple)):
+        return [normalize_json(item) for item in value]
+    raise ProtocolViolation(f"unsupported canonical JSON value: {type(value).__name__}")
+
+
+def canonical_json_bytes(value: Any) -> bytes:
+    return json.dumps(
+        normalize_json(value),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def canonical_json_text(value: Any) -> str:
+    return canonical_json_bytes(value).decode("utf-8")
+
+
+def canonical_sha256(value: Any) -> Sha256Digest:
+    return Sha256Digest(hashlib.sha256(canonical_json_bytes(value)).hexdigest())
+
+
+__all__ = ["canonical_json_bytes", "canonical_json_text", "canonical_sha256", "normalize_json"]

script_build_host/src/script_build_host/domain/defects.py

@@ -0,0 +1,141 @@
+"""Bounded, machine-readable defects produced by phase-two validators."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any, Self
+
+from .errors import ProtocolViolation
+
+MAX_DEFECTS_PER_VALIDATION = 50
+MAX_DEFECT_EXCERPT_CHARS = 500
+_DEFECT_CODE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")
+
+
+class DefectSeverity(StrEnum):
+    WARNING = "warning"
+    HARD = "hard"
+    CRITICAL = "critical"
+
+
+class RecommendedActionClass(StrEnum):
+    REPAIR = "repair"
+    RETRY = "retry"
+    REVISE = "revise"
+    SPLIT = "split"
+    RETRIEVE = "retrieve"
+    REPLACE = "replace"
+    COMPARE = "compare"
+    BLOCK = "block"
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptDefectV1:
+    """One concrete observation; it never decides the Planner's next action."""
+
+    defect_code: str
+    criterion_id: str
+    scope_ref: str
+    observed_excerpt: str
+    evidence_refs: tuple[str, ...]
+    severity: DefectSeverity
+    invalidated_inputs: tuple[str, ...]
+    recommended_action_class: RecommendedActionClass
+
+    def __post_init__(self) -> None:
+        for label, value, maximum in (
+            ("defect_code", self.defect_code, 64),
+            ("criterion_id", self.criterion_id, 128),
+            ("scope_ref", self.scope_ref, 500),
+        ):
+            if not isinstance(value, str) or not value.strip():
+                raise ProtocolViolation(f"{label} must not be blank")
+            if len(value) > maximum:
+                raise ProtocolViolation(f"{label} exceeds {maximum} characters")
+        if not _DEFECT_CODE.fullmatch(self.defect_code):
+            raise ProtocolViolation("defect_code must be an uppercase stable identifier")
+        if not self.scope_ref.startswith("script-build://"):
+            raise ProtocolViolation("defect scope_ref must use script-build://")
+        if not self.observed_excerpt.strip():
+            raise ProtocolViolation("defect observed_excerpt must not be blank")
+        if len(self.observed_excerpt) > MAX_DEFECT_EXCERPT_CHARS:
+            raise ProtocolViolation("defect excerpt exceeds 500 characters")
+        for label, values in (
+            ("evidence_refs", self.evidence_refs),
+            ("invalidated_inputs", self.invalidated_inputs),
+        ):
+            if any(not value.strip() for value in values) or len(set(values)) != len(values):
+                raise ProtocolViolation(f"{label} must contain unique non-blank references")
+
+    @property
+    def blocks_acceptance(self) -> bool:
+        return self.severity in {DefectSeverity.HARD, DefectSeverity.CRITICAL}
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "defect_code": self.defect_code,
+            "criterion_id": self.criterion_id,
+            "scope_ref": self.scope_ref,
+            "observed_excerpt": self.observed_excerpt,
+            "evidence_refs": list(self.evidence_refs),
+            "severity": self.severity.value,
+            "invalidated_inputs": list(self.invalidated_inputs),
+            "recommended_action_class": self.recommended_action_class.value,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        try:
+            return cls(
+                defect_code=str(value.get("defect_code", "")),
+                criterion_id=str(value.get("criterion_id", "")),
+                scope_ref=str(value.get("scope_ref", "")),
+                observed_excerpt=str(value.get("observed_excerpt", "")),
+                evidence_refs=_strings(value.get("evidence_refs"), "evidence_refs"),
+                severity=DefectSeverity(str(value.get("severity", ""))),
+                invalidated_inputs=_strings(value.get("invalidated_inputs"), "invalidated_inputs"),
+                recommended_action_class=RecommendedActionClass(
+                    str(value.get("recommended_action_class", ""))
+                ),
+            )
+        except ValueError as exc:
+            raise ProtocolViolation("defect contains an unsupported enum value") from exc
+
+
+def validate_defect_set(
+    values: tuple[ScriptDefectV1, ...],
+    *,
+    criterion_ids: set[str],
+    allowed_evidence_refs: set[str],
+    passed: bool,
+) -> None:
+    if len(values) > MAX_DEFECTS_PER_VALIDATION:
+        raise ProtocolViolation("one validation may report at most 50 defects")
+    for defect in values:
+        if defect.criterion_id not in criterion_ids:
+            raise ProtocolViolation("defect references a criterion outside the frozen contract")
+        if not set(defect.evidence_refs) <= allowed_evidence_refs:
+            raise ProtocolViolation("defect references evidence outside the validation snapshot")
+    if passed and any(item.blocks_acceptance for item in values):
+        raise ProtocolViolation("passed validation cannot contain hard or critical defects")
+
+
+def _strings(value: Any, label: str) -> tuple[str, ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
+        raise ProtocolViolation(f"{label} must be a string array")
+    return tuple(value)
+
+
+__all__ = [
+    "MAX_DEFECTS_PER_VALIDATION",
+    "MAX_DEFECT_EXCERPT_CHARS",
+    "DefectSeverity",
+    "RecommendedActionClass",
+    "ScriptDefectV1",
+    "validate_defect_set",
+]

script_build_host/src/script_build_host/domain/errors.py

@@ -28,6 +28,50 @@ class InputHashConflict(ScriptBuildError):
         )
 
 
+class InputDigestMismatch(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "INPUT_DIGEST_MISMATCH",
+            "the frozen input snapshot does not match its canonical digest",
+        )
+
+
+class DirectionProjectionConflict(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "DIRECTION_PROJECTION_CONFLICT",
+            "the accepted direction conflicts with an existing durable projection",
+        )
+
+
+class PhaseTwoBoundaryNotReady(ScriptBuildError):
+    def __init__(
+        self, summary: str = "the phase-two continuation preconditions are not met"
+    ) -> None:
+        super().__init__("PHASE_TWO_BOUNDARY_NOT_READY", summary)
+
+
+class PhaseTwoPromptsNotFrozen(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "PHASE_TWO_PROMPTS_NOT_FROZEN",
+            "the input snapshot does not freeze all required phase-two prompts",
+        )
+
+
+class MissionRecoveryRequired(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "MISSION_RECOVERY_REQUIRED",
+            "phase-two work already started and requires the phase-three recovery workflow",
+        )
+
+
+class PlannerPolicyMigrationRequired(ScriptBuildError):
+    def __init__(self, summary: str = "the phase-two Planner policy cannot be verified") -> None:
+        super().__init__("PLANNER_POLICY_MIGRATION_REQUIRED", summary)
+
+
 class BuildNotFound(ScriptBuildError):
     def __init__(self) -> None:
         super().__init__("BUILD_NOT_FOUND", "the requested script build resource does not exist")

script_build_host/src/script_build_host/domain/input_snapshot.py

@@ -19,10 +19,15 @@ class ScriptBuildInput:
     prompt_manifest: tuple[dict[str, Any], ...] = ()
     datasource_manifest: dict[str, Any] = field(default_factory=dict)
     model_manifest: dict[str, Any] = field(default_factory=dict)
+    parent_snapshot_id: str | None = None
+    parent_snapshot_sha256: str | None = None
+    business_input_sha256: str | None = None
 
     def canonical_payload(self) -> dict[str, Any]:
-        return {
-            "schema_version": "script-build-input/v1",
+        payload: dict[str, Any] = {
+            "schema_version": (
+                "script-build-input/v2" if self.parent_snapshot_id else "script-build-input/v1"
+            ),
             "script_build_id": self.script_build_id,
             "execution_id": self.execution_id,
             "topic_build_id": self.topic_build_id,
@@ -36,6 +41,30 @@ class ScriptBuildInput:
             "datasource_manifest": self.datasource_manifest,
             "model_manifest": self.model_manifest,
         }
+        if self.parent_snapshot_id:
+            payload.update(
+                {
+                    "parent_snapshot_id": self.parent_snapshot_id,
+                    "parent_snapshot_sha256": self.parent_snapshot_sha256,
+                    "business_input_sha256": self.business_input_sha256,
+                }
+            )
+        return payload
+
+    def business_payload(self) -> dict[str, Any]:
+        """Return the immutable source/business portion shared by snapshot versions."""
+
+        return {
+            "script_build_id": self.script_build_id,
+            "execution_id": self.execution_id,
+            "topic_build_id": self.topic_build_id,
+            "topic_id": self.topic_id,
+            "topic": self.topic,
+            "account": self.account,
+            "persona_points": list(self.persona_points),
+            "section_patterns": list(self.section_patterns),
+            "strategies": list(self.strategies),
+        }
 
 
 @dataclass(frozen=True, slots=True)
@@ -55,6 +84,9 @@ class ScriptBuildInputSnapshotV1:
     model_manifest: dict[str, Any]
     canonical_sha256: str
     created_at: datetime
+    parent_snapshot_id: str | None = None
+    parent_snapshot_sha256: str | None = None
+    business_input_sha256: str | None = None
 
     def to_input(self) -> ScriptBuildInput:
         return ScriptBuildInput(
@@ -70,4 +102,7 @@ class ScriptBuildInputSnapshotV1:
             prompt_manifest=self.prompt_manifest,
             datasource_manifest=self.datasource_manifest,
             model_manifest=self.model_manifest,
+            parent_snapshot_id=self.parent_snapshot_id,
+            parent_snapshot_sha256=self.parent_snapshot_sha256,
+            business_input_sha256=self.business_input_sha256,
         )

script_build_host/src/script_build_host/domain/phase_two_artifacts.py

@@ -0,0 +1,586 @@
+"""Canonical phase-two candidate artifacts, independent of ORM storage."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from typing import Any, Self
+
+from .defects import MAX_DEFECTS_PER_VALIDATION, DefectSeverity, ScriptDefectV1
+from .digests import Sha256Digest
+from .errors import ProtocolViolation
+
+_CONTROL_PREFIX = "script-build://"
+_ARTIFACT_PREFIX = "script-build://artifact-versions/"
+
+
+@dataclass(frozen=True, slots=True)
+class CandidateLineageV1:
+    scope_ref: str
+    input_snapshot_ref: str
+    input_closure_digest: str
+    write_scope: tuple[str, ...]
+    base_artifact_ref: str | None = None
+    base_artifact_digest: str | None = None
+    base_revision: int | None = None
+    supersedes_decision_ids: tuple[str, ...] = ()
+    source_lineage: tuple[dict[str, Any], ...] = ()
+
+    def __post_init__(self) -> None:
+        _control_uri(self.scope_ref, "scope_ref")
+        _control_uri(self.input_snapshot_ref, "input_snapshot_ref")
+        Sha256Digest.from_wire(self.input_closure_digest)
+        _unique_nonblank(self.write_scope, "write_scope")
+        for item in self.write_scope:
+            _control_uri(item, "write_scope")
+        _unique_nonblank(self.supersedes_decision_ids, "supersedes_decision_ids")
+        if (self.base_artifact_ref is None) != (self.base_artifact_digest is None):
+            raise ProtocolViolation("base artifact URI and digest must be supplied together")
+        if self.base_artifact_ref is not None:
+            _artifact_uri(self.base_artifact_ref, "base_artifact_ref")
+            Sha256Digest.from_wire(self.base_artifact_digest or "")
+            if not isinstance(self.base_revision, int) or self.base_revision < 1:
+                raise ProtocolViolation("base revision must pin one positive artifact version")
+        elif self.base_revision is not None:
+            raise ProtocolViolation("base revision requires a base artifact")
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "scope_ref": self.scope_ref,
+            "input_snapshot_ref": self.input_snapshot_ref,
+            "input_closure_digest": self.input_closure_digest,
+            "base_artifact_ref": self.base_artifact_ref,
+            "base_artifact_digest": self.base_artifact_digest,
+            "base_revision": self.base_revision,
+            "write_scope": list(self.write_scope),
+            "supersedes_decision_ids": list(self.supersedes_decision_ids),
+            "source_lineage": list(self.source_lineage),
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        return cls(
+            scope_ref=str(value.get("scope_ref", "")),
+            input_snapshot_ref=str(value.get("input_snapshot_ref", "")),
+            input_closure_digest=str(value.get("input_closure_digest", "")),
+            base_artifact_ref=_optional_text(value.get("base_artifact_ref")),
+            base_artifact_digest=_optional_text(value.get("base_artifact_digest")),
+            base_revision=(
+                int(value["base_revision"]) if value.get("base_revision") is not None else None
+            ),
+            write_scope=_text_tuple(value.get("write_scope")),
+            supersedes_decision_ids=_text_tuple(value.get("supersedes_decision_ids")),
+            source_lineage=_mapping_tuple(value.get("source_lineage")),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptParagraphV1:
+    paragraph_id: int
+    paragraph_index: int
+    level: int
+    parent_id: int | None
+    name: str
+    content_range: dict[str, Any]
+    theme: str | None = None
+    form: str | None = None
+    function: str | None = None
+    feeling: str | None = None
+    theme_elements: tuple[dict[str, Any], ...] = ()
+    form_elements: tuple[dict[str, Any], ...] = ()
+    function_elements: tuple[dict[str, Any], ...] = ()
+    feeling_elements: tuple[dict[str, Any], ...] = ()
+    description: str | None = None
+    full_description: str | None = None
+    is_active: bool = True
+
+    def __post_init__(self) -> None:
+        if self.paragraph_id < 1 or self.paragraph_index < 1:
+            raise ProtocolViolation("paragraph identities and indexes must be positive")
+        if self.level not in (1, 2):
+            raise ProtocolViolation("paragraph level must be 1 or 2")
+        if self.level == 1 and self.parent_id is not None:
+            raise ProtocolViolation("level-one paragraphs cannot have a parent")
+        if self.level == 2 and (self.parent_id is None or self.parent_id < 1):
+            raise ProtocolViolation("level-two paragraphs require a parent")
+        if not self.name.strip():
+            raise ProtocolViolation("paragraph name must not be blank")
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "paragraph_id": self.paragraph_id,
+            "paragraph_index": self.paragraph_index,
+            "level": self.level,
+            "parent_id": self.parent_id,
+            "name": self.name,
+            "content_range": self.content_range,
+            "theme": self.theme,
+            "form": self.form,
+            "function": self.function,
+            "feeling": self.feeling,
+            "theme_elements": list(self.theme_elements),
+            "form_elements": list(self.form_elements),
+            "function_elements": list(self.function_elements),
+            "feeling_elements": list(self.feeling_elements),
+            "description": self.description,
+            "full_description": self.full_description,
+            "is_active": self.is_active,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        return cls(
+            paragraph_id=int(value["paragraph_id"]),
+            paragraph_index=int(value["paragraph_index"]),
+            level=int(value.get("level", 1)),
+            parent_id=int(value["parent_id"]) if value.get("parent_id") is not None else None,
+            name=str(value.get("name", "")),
+            content_range=dict(value.get("content_range") or {}),
+            theme=_optional_text(value.get("theme")),
+            form=_optional_text(value.get("form")),
+            function=_optional_text(value.get("function")),
+            feeling=_optional_text(value.get("feeling")),
+            theme_elements=_mapping_tuple(value.get("theme_elements")),
+            form_elements=_mapping_tuple(value.get("form_elements")),
+            function_elements=_mapping_tuple(value.get("function_elements")),
+            feeling_elements=_mapping_tuple(value.get("feeling_elements")),
+            description=_optional_text(value.get("description")),
+            full_description=_optional_text(value.get("full_description")),
+            is_active=bool(value.get("is_active", True)),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptElementV1:
+    element_id: int
+    name: str
+    dimension_primary: str
+    dimension_secondary: str
+    commonality_analysis: dict[str, Any] | None = None
+    topic_support: dict[str, Any] | None = None
+    weight_score: dict[str, Any] | None = None
+    support_elements: tuple[dict[str, Any], ...] = ()
+    is_active: bool = True
+
+    def __post_init__(self) -> None:
+        if self.element_id < 1:
+            raise ProtocolViolation("element identity must be positive")
+        if self.dimension_primary not in {"实质", "形式"}:
+            raise ProtocolViolation("element primary dimension must be 实质 or 形式")
+        if not self.name.strip() or not self.dimension_secondary.strip():
+            raise ProtocolViolation("element name and secondary dimension must not be blank")
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "element_id": self.element_id,
+            "name": self.name,
+            "dimension_primary": self.dimension_primary,
+            "dimension_secondary": self.dimension_secondary,
+            "commonality_analysis": self.commonality_analysis,
+            "topic_support": self.topic_support,
+            "weight_score": self.weight_score,
+            "support_elements": list(self.support_elements),
+            "is_active": self.is_active,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        return cls(
+            element_id=int(value["element_id"]),
+            name=str(value.get("name", "")),
+            dimension_primary=str(value.get("dimension_primary", "")),
+            dimension_secondary=str(value.get("dimension_secondary", "")),
+            commonality_analysis=_optional_mapping(value.get("commonality_analysis")),
+            topic_support=_optional_mapping(value.get("topic_support")),
+            weight_score=_optional_mapping(value.get("weight_score")),
+            support_elements=_mapping_tuple(value.get("support_elements")),
+            is_active=bool(value.get("is_active", True)),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptParagraphElementLinkV1:
+    paragraph_id: int
+    element_id: int
+
+    def __post_init__(self) -> None:
+        if self.paragraph_id < 1 or self.element_id < 1:
+            raise ProtocolViolation("paragraph-element links require positive identities")
+
+    def to_payload(self) -> dict[str, int]:
+        return {"paragraph_id": self.paragraph_id, "element_id": self.element_id}
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        return cls(paragraph_id=int(value["paragraph_id"]), element_id=int(value["element_id"]))
+
+
+@dataclass(frozen=True, slots=True)
+class StructureArtifactV1:
+    lineage: CandidateLineageV1
+    paragraphs: tuple[ScriptParagraphV1, ...]
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        _validate_paragraphs(self.paragraphs)
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "structure-artifact/v1",
+            **self.lineage.to_payload(),
+            "paragraphs": [item.to_payload() for item in self.paragraphs],
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ParagraphArtifactV1:
+    lineage: CandidateLineageV1
+    paragraphs: tuple[ScriptParagraphV1, ...]
+    elements: tuple[ScriptElementV1, ...] = ()
+    paragraph_element_links: tuple[ScriptParagraphElementLinkV1, ...] = ()
+    change_manifest: dict[str, Any] = field(default_factory=dict)
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        _validate_paragraphs(self.paragraphs)
+        _validate_elements_and_links(
+            self.paragraphs,
+            self.elements,
+            self.paragraph_element_links,
+        )
+
+    def content_payload(self) -> dict[str, Any]:
+        schema = "paragraph-patch/v1" if self.lineage.base_artifact_ref else "paragraph-artifact/v1"
+        return {
+            "schema_version": schema,
+            **self.lineage.to_payload(),
+            "paragraphs": [item.to_payload() for item in self.paragraphs],
+            "elements": [item.to_payload() for item in self.elements],
+            "paragraph_element_links": [item.to_payload() for item in self.paragraph_element_links],
+            "change_manifest": self.change_manifest,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ElementSetArtifactV1:
+    lineage: CandidateLineageV1
+    elements: tuple[ScriptElementV1, ...]
+    paragraph_element_links: tuple[ScriptParagraphElementLinkV1, ...]
+    paragraphs: tuple[ScriptParagraphV1, ...] = ()
+    change_manifest: dict[str, Any] = field(default_factory=dict)
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        if self.paragraph_element_links and not self.paragraphs:
+            raise ProtocolViolation("element-set projection must include every linked paragraph")
+        if self.paragraphs:
+            _validate_paragraphs(self.paragraphs)
+        _validate_elements_and_links(
+            self.paragraphs,
+            self.elements,
+            self.paragraph_element_links,
+        )
+
+    def content_payload(self) -> dict[str, Any]:
+        schema = (
+            "element-set-patch/v1" if self.lineage.base_artifact_ref else "element-set-artifact/v1"
+        )
+        return {
+            "schema_version": schema,
+            **self.lineage.to_payload(),
+            "paragraphs": [item.to_payload() for item in self.paragraphs],
+            "elements": [item.to_payload() for item in self.elements],
+            "paragraph_element_links": [item.to_payload() for item in self.paragraph_element_links],
+            "change_manifest": self.change_manifest,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ComparisonArtifactV1:
+    lineage: CandidateLineageV1
+    candidate_artifact_refs: tuple[str, ...]
+    criterion_results: tuple[dict[str, Any], ...]
+    conflicts: tuple[str, ...]
+    recommendation: str
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        if len(self.candidate_artifact_refs) < 2:
+            raise ProtocolViolation("comparison requires at least two immutable candidates")
+        _unique_nonblank(self.candidate_artifact_refs, "candidate_artifact_refs")
+        for item in self.candidate_artifact_refs:
+            _artifact_uri(item, "candidate_artifact_ref")
+        if not self.criterion_results or not self.recommendation.strip():
+            raise ProtocolViolation("comparison requires criterion results and a recommendation")
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "comparison-artifact/v1",
+            **self.lineage.to_payload(),
+            "candidate_artifact_refs": list(self.candidate_artifact_refs),
+            "criterion_results": list(self.criterion_results),
+            "conflicts": list(self.conflicts),
+            "recommendation": self.recommendation,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class StructuredScriptArtifactV1:
+    direction_ref: str
+    input_closure_digest: str
+    paragraphs: tuple[ScriptParagraphV1, ...]
+    elements: tuple[ScriptElementV1, ...]
+    paragraph_element_links: tuple[ScriptParagraphElementLinkV1, ...]
+    source_artifact_refs: tuple[str, ...]
+    evidence_refs: tuple[str, ...]
+    acceptance_notes: tuple[str, ...]
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        _artifact_uri(self.direction_ref, "direction_ref")
+        Sha256Digest.from_wire(self.input_closure_digest)
+        _validate_paragraphs(self.paragraphs)
+        _validate_elements_and_links(self.paragraphs, self.elements, self.paragraph_element_links)
+        for label, refs in (
+            ("source_artifact_refs", self.source_artifact_refs),
+            ("evidence_refs", self.evidence_refs),
+        ):
+            _unique_nonblank(refs, label)
+            for item in refs:
+                _artifact_uri(item, label)
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "structured-script/v1",
+            "direction_ref": self.direction_ref,
+            "input_closure_digest": self.input_closure_digest,
+            "paragraphs": [item.to_payload() for item in self.paragraphs],
+            "elements": [item.to_payload() for item in self.elements],
+            "paragraph_element_links": [item.to_payload() for item in self.paragraph_element_links],
+            "source_artifact_refs": list(self.source_artifact_refs),
+            "evidence_refs": list(self.evidence_refs),
+            "acceptance_notes": list(self.acceptance_notes),
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class CandidatePortfolioArtifactV1:
+    adopted_structured_script_ref: str
+    candidate_structured_script_refs: tuple[str, ...]
+    accepted_decision_ids: tuple[str, ...]
+    superseded_decision_ids: tuple[str, ...]
+    rejected_or_held_decision_ids: tuple[str, ...]
+    input_closure_digest: str
+    unresolved_defects: tuple[ScriptDefectV1, ...]
+    compose_order: tuple[str, ...]
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        _artifact_uri(self.adopted_structured_script_ref, "adopted_structured_script_ref")
+        _unique_nonblank(self.candidate_structured_script_refs, "candidate_structured_script_refs")
+        for item in self.candidate_structured_script_refs:
+            _artifact_uri(item, "candidate_structured_script_ref")
+        if self.adopted_structured_script_ref not in self.candidate_structured_script_refs:
+            raise ProtocolViolation("adopted StructuredScript must be in the candidate set")
+        adopted = set(self.accepted_decision_ids)
+        disposed = set(self.superseded_decision_ids) | set(self.rejected_or_held_decision_ids)
+        if adopted & disposed:
+            raise ProtocolViolation("accepted and disposed decisions must be disjoint")
+        if len(self.unresolved_defects) > MAX_DEFECTS_PER_VALIDATION:
+            raise ProtocolViolation("portfolio may retain at most 50 warning defects")
+        for defect in self.unresolved_defects:
+            if not isinstance(defect, ScriptDefectV1):
+                raise ProtocolViolation("portfolio defects must use script-defect/v1")
+            if defect.severity is not DefectSeverity.WARNING:
+                raise ProtocolViolation("portfolio cannot retain hard or critical defects")
+        Sha256Digest.from_wire(self.input_closure_digest)
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "candidate-portfolio/v1",
+            "adopted_structured_script_ref": self.adopted_structured_script_ref,
+            "candidate_structured_script_refs": list(self.candidate_structured_script_refs),
+            "accepted_decision_ids": list(self.accepted_decision_ids),
+            "superseded_decision_ids": list(self.superseded_decision_ids),
+            "rejected_or_held_decision_ids": list(self.rejected_or_held_decision_ids),
+            "input_closure_digest": self.input_closure_digest,
+            "unresolved_defects": [item.to_payload() for item in self.unresolved_defects],
+            "compose_order": list(self.compose_order),
+        }
+
+
+PhaseTwoBusinessArtifact = (
+    StructureArtifactV1
+    | ParagraphArtifactV1
+    | ElementSetArtifactV1
+    | ComparisonArtifactV1
+    | StructuredScriptArtifactV1
+    | CandidatePortfolioArtifactV1
+)
+
+
+def hydrate_phase_two_artifact(payload: Mapping[str, Any]) -> PhaseTwoBusinessArtifact:
+    schema = payload.get("schema_version")
+    if schema == "structure-artifact/v1":
+        return StructureArtifactV1(
+            lineage=CandidateLineageV1.from_payload(payload),
+            paragraphs=_paragraphs(payload.get("paragraphs")),
+        )
+    if schema in {"paragraph-artifact/v1", "paragraph-patch/v1"}:
+        return ParagraphArtifactV1(
+            lineage=CandidateLineageV1.from_payload(payload),
+            paragraphs=_paragraphs(payload.get("paragraphs")),
+            elements=_elements(payload.get("elements")),
+            paragraph_element_links=_links(payload.get("paragraph_element_links")),
+            change_manifest=dict(payload.get("change_manifest") or {}),
+        )
+    if schema in {"element-set-artifact/v1", "element-set-patch/v1"}:
+        return ElementSetArtifactV1(
+            lineage=CandidateLineageV1.from_payload(payload),
+            elements=_elements(payload.get("elements")),
+            paragraph_element_links=_links(payload.get("paragraph_element_links")),
+            paragraphs=_paragraphs(payload.get("paragraphs")),
+            change_manifest=dict(payload.get("change_manifest") or {}),
+        )
+    if schema == "comparison-artifact/v1":
+        return ComparisonArtifactV1(
+            lineage=CandidateLineageV1.from_payload(payload),
+            candidate_artifact_refs=_text_tuple(payload.get("candidate_artifact_refs")),
+            criterion_results=_mapping_tuple(payload.get("criterion_results")),
+            conflicts=_text_tuple(payload.get("conflicts")),
+            recommendation=str(payload.get("recommendation", "")),
+        )
+    if schema == "structured-script/v1":
+        return StructuredScriptArtifactV1(
+            direction_ref=str(payload.get("direction_ref", "")),
+            input_closure_digest=str(payload.get("input_closure_digest", "")),
+            paragraphs=_paragraphs(payload.get("paragraphs")),
+            elements=_elements(payload.get("elements")),
+            paragraph_element_links=_links(payload.get("paragraph_element_links")),
+            source_artifact_refs=_text_tuple(payload.get("source_artifact_refs")),
+            evidence_refs=_text_tuple(payload.get("evidence_refs")),
+            acceptance_notes=_text_tuple(payload.get("acceptance_notes")),
+        )
+    if schema == "candidate-portfolio/v1":
+        return CandidatePortfolioArtifactV1(
+            adopted_structured_script_ref=str(payload.get("adopted_structured_script_ref", "")),
+            candidate_structured_script_refs=_text_tuple(
+                payload.get("candidate_structured_script_refs")
+            ),
+            accepted_decision_ids=_text_tuple(payload.get("accepted_decision_ids")),
+            superseded_decision_ids=_text_tuple(payload.get("superseded_decision_ids")),
+            rejected_or_held_decision_ids=_text_tuple(payload.get("rejected_or_held_decision_ids")),
+            input_closure_digest=str(payload.get("input_closure_digest", "")),
+            unresolved_defects=tuple(
+                ScriptDefectV1.from_payload(item)
+                for item in _mapping_tuple(payload.get("unresolved_defects"))
+            ),
+            compose_order=_text_tuple(payload.get("compose_order")),
+        )
+    raise ProtocolViolation("unsupported phase-two artifact schema")
+
+
+def _validate_paragraphs(values: tuple[ScriptParagraphV1, ...]) -> None:
+    if not values:
+        raise ProtocolViolation("candidate paragraph set must not be empty")
+    ids = {item.paragraph_id for item in values}
+    if len(ids) != len(values):
+        raise ProtocolViolation("paragraph identities must be unique")
+    active_indexes = [item.paragraph_index for item in values if item.is_active]
+    if len(set(active_indexes)) != len(active_indexes):
+        raise ProtocolViolation("active paragraph indexes must be unique")
+    for item in values:
+        if item.parent_id is not None and item.parent_id not in ids:
+            raise ProtocolViolation("paragraph parent reference is dangling")
+
+
+def _validate_elements_and_links(
+    paragraphs: tuple[ScriptParagraphV1, ...],
+    elements: tuple[ScriptElementV1, ...],
+    links: tuple[ScriptParagraphElementLinkV1, ...],
+    *,
+    allow_unknown_paragraph: bool = False,
+) -> None:
+    element_ids = {item.element_id for item in elements if item.is_active}
+    if len({item.element_id for item in elements}) != len(elements):
+        raise ProtocolViolation("element identities must be unique")
+    paragraph_ids = {item.paragraph_id for item in paragraphs if item.is_active}
+    pairs = {(item.paragraph_id, item.element_id) for item in links}
+    if len(pairs) != len(links):
+        raise ProtocolViolation("paragraph-element links must be unique")
+    for item in links:
+        if item.element_id not in element_ids:
+            raise ProtocolViolation("paragraph-element link points to an inactive element")
+        if not allow_unknown_paragraph and item.paragraph_id not in paragraph_ids:
+            raise ProtocolViolation("paragraph-element link points to an inactive paragraph")
+
+
+def _control_uri(value: str, label: str) -> None:
+    if not value.startswith(_CONTROL_PREFIX):
+        raise ProtocolViolation(f"{label} must use script-build://")
+
+
+def _artifact_uri(value: str, label: str) -> None:
+    if not value.startswith(_ARTIFACT_PREFIX) or not value.removeprefix(_ARTIFACT_PREFIX).isdigit():
+        raise ProtocolViolation(f"{label} must pin one immutable artifact version")
+
+
+def _unique_nonblank(values: tuple[str, ...], label: str) -> None:
+    if any(not item.strip() for item in values) or len(set(values)) != len(values):
+        raise ProtocolViolation(f"{label} must contain unique non-blank values")
+
+
+def _optional_text(value: Any) -> str | None:
+    return str(value) if value is not None else None
+
+
+def _optional_mapping(value: Any) -> dict[str, Any] | None:
+    if value is None:
+        return None
+    if not isinstance(value, Mapping):
+        raise ProtocolViolation("artifact JSON field must be an object")
+    return dict(value)
+
+
+def _mapping_tuple(value: Any) -> tuple[dict[str, Any], ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, list) or any(not isinstance(item, Mapping) for item in value):
+        raise ProtocolViolation("artifact field must be an object array")
+    return tuple(dict(item) for item in value)
+
+
+def _text_tuple(value: Any) -> tuple[str, ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
+        raise ProtocolViolation("artifact field must be a string array")
+    return tuple(value)
+
+
+def _paragraphs(value: Any) -> tuple[ScriptParagraphV1, ...]:
+    return tuple(ScriptParagraphV1.from_payload(item) for item in _mapping_tuple(value))
+
+
+def _elements(value: Any) -> tuple[ScriptElementV1, ...]:
+    return tuple(ScriptElementV1.from_payload(item) for item in _mapping_tuple(value))
+
+
+def _links(value: Any) -> tuple[ScriptParagraphElementLinkV1, ...]:
+    return tuple(ScriptParagraphElementLinkV1.from_payload(item) for item in _mapping_tuple(value))
+
+
+__all__ = [
+    "CandidateLineageV1",
+    "CandidatePortfolioArtifactV1",
+    "ComparisonArtifactV1",
+    "ElementSetArtifactV1",
+    "ParagraphArtifactV1",
+    "PhaseTwoBusinessArtifact",
+    "ScriptElementV1",
+    "ScriptParagraphElementLinkV1",
+    "ScriptParagraphV1",
+    "StructureArtifactV1",
+    "StructuredScriptArtifactV1",
+    "hydrate_phase_two_artifact",
+]

script_build_host/src/script_build_host/domain/phase_two_ports.py

@@ -0,0 +1,68 @@
+"""Ports owned by the script-build phase-two bounded context."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from agent.orchestration import ArtifactRef
+
+from .artifacts import ArtifactKind, ArtifactVersion
+from .phase_two_artifacts import CandidateLineageV1
+from .task_contracts import FrozenTaskContract, ScriptTaskContractV1
+from .workspaces import CandidateWorkspace, CandidateWorkspaceSnapshot, CandidateWriteContext
+
+
+@dataclass(frozen=True, slots=True)
+class AttemptManifest:
+    """The one immutable business output owned by a Worker Attempt."""
+
+    artifact_ref: ArtifactRef
+    scope_ref: str
+
+
+class ScriptTaskContractStore(Protocol):
+    async def freeze(
+        self, root_trace_id: str, contract: ScriptTaskContractV1
+    ) -> FrozenTaskContract: ...
+
+    async def read(self, root_trace_id: str, uri: str) -> FrozenTaskContract: ...
+
+    async def verify(self, root_trace_id: str, uri: str, digest: str) -> bool: ...
+
+    async def collect_unreferenced(
+        self, root_trace_id: str, referenced_digests: set[str]
+    ) -> tuple[str, ...]: ...
+
+    async def prune_unreferenced(
+        self, root_trace_id: str, referenced_digests: set[str]
+    ) -> tuple[str, ...]: ...
+
+
+class CandidateWorkspaceRepository(Protocol):
+    async def get_or_create(
+        self,
+        context: CandidateWriteContext,
+        *,
+        artifact_kind: ArtifactKind,
+        base_artifact_ref: ArtifactRef | None = None,
+    ) -> CandidateWorkspace: ...
+
+    async def require(self, context: CandidateWriteContext) -> CandidateWorkspace: ...
+
+    async def snapshot(self, context: CandidateWriteContext) -> CandidateWorkspaceSnapshot: ...
+
+    async def freeze(
+        self,
+        context: CandidateWriteContext,
+        *,
+        lineage: CandidateLineageV1,
+        structured_script: dict[str, Any] | None = None,
+    ) -> tuple[ArtifactVersion, ArtifactRef]: ...
+
+    async def discard(
+        self, context: CandidateWriteContext, *, reason: str
+    ) -> CandidateWorkspace: ...
+
+
+__all__ = ["AttemptManifest", "CandidateWorkspaceRepository", "ScriptTaskContractStore"]

script_build_host/src/script_build_host/domain/ports.py

@@ -82,6 +82,14 @@ class MissionBindingRepository(Protocol):
         self, *, script_build_id: int, artifact_version_id: int
     ) -> MissionBinding: ...
 
+    async def compare_and_set_input_snapshot(
+        self,
+        *,
+        script_build_id: int,
+        expected_snapshot_id: int,
+        new_snapshot_id: int,
+    ) -> MissionBinding: ...
+
 
 class ScriptBusinessArtifactRepository(Protocol):
     async def freeze(
@@ -109,6 +117,10 @@ class ScriptBusinessArtifactRepository(Protocol):
         self, artifact_version_id: int, *, script_build_id: int
     ) -> ArtifactVersion: ...
 
+    async def get_by_attempt(
+        self, *, script_build_id: int, task_id: str, attempt_id: str
+    ) -> ArtifactVersion: ...
+
 
 class PublicationRepository(Protocol):
     async def prepare(
@@ -153,6 +165,25 @@ class LegacyBuildStateRepository(Protocol):
 
     async def project_direction(self, script_build_id: int, legacy_markdown: str) -> None: ...
 
+    async def get_direction(self, script_build_id: int) -> str | None: ...
+
+    async def set_checkpoint(
+        self,
+        script_build_id: int,
+        *,
+        checkpoint_code: str,
+        summary: str,
+        root_trace_id: str,
+    ) -> None: ...
+
+    async def begin_phase(self, script_build_id: int) -> None: ...
+
+
+class DurableTraceStoreVerifier(Protocol):
+    """Deployment-owned proof that Trace state survives a process boundary."""
+
+    async def verify(self, trace_store: Any, *, require_attachments: bool = False) -> None: ...
+
 
 class PrincipalProvider(Protocol):
     async def current(self, request_context: object | None = None) -> Principal: ...

script_build_host/src/script_build_host/domain/redaction.py

@@ -0,0 +1,59 @@
+"""Framework-free secret redaction shared by inner Host layers."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping, Sequence
+from typing import Any
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+_SENSITIVE_KEY = re.compile(
+    r"(^|[_-])(authorization|cookie|password|passwd|secret|token|api[_-]?key|dsn|database[_-]?url)($|[_-])",
+    re.IGNORECASE,
+)
+_BEARER = re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]+")
+_CONNECTION_URL = re.compile(r"([a-z][a-z0-9+.-]*://)([^\s/@:]+):([^\s/@]+)@", re.IGNORECASE)
+
+
+def redact_text(value: str) -> str:
+    value = _BEARER.sub("Bearer [REDACTED]", value)
+    return _CONNECTION_URL.sub(r"\1[REDACTED]:[REDACTED]@", value)
+
+
+def redact_url(value: str) -> str:
+    try:
+        parts = urlsplit(value)
+    except ValueError:
+        return redact_text(value)
+    if not parts.scheme or not parts.netloc:
+        return redact_text(value)
+    host = parts.hostname or ""
+    try:
+        parsed_port = parts.port
+    except ValueError:
+        return redact_text(value)
+    port = f":{parsed_port}" if parsed_port is not None else ""
+    netloc = f"[REDACTED]@{host}{port}" if parts.username is not None else parts.netloc
+    query = urlencode(
+        [
+            (key, "[REDACTED]" if _SENSITIVE_KEY.search(key) else item)
+            for key, item in parse_qsl(parts.query, keep_blank_values=True)
+        ]
+    )
+    return urlunsplit((parts.scheme, netloc, parts.path, query, ""))
+
+
+def redact(value: Any) -> Any:
+    if isinstance(value, str):
+        return redact_url(value) if "://" in value else redact_text(value)
+    if isinstance(value, Mapping):
+        return {
+            str(key): "[REDACTED]" if _SENSITIVE_KEY.search(str(key)) else redact(item)
+            for key, item in value.items()
+        }
+    if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):
+        return [redact(item) for item in value]
+    return value
+
+
+__all__ = ["redact", "redact_text", "redact_url"]

script_build_host/src/script_build_host/domain/task_contracts.py

@@ -0,0 +1,635 @@
+"""Immutable business contracts for dynamically planned script-build tasks."""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
+
+from agent.orchestration import ArtifactRef
+
+from script_build_host.domain.digests import Sha256Digest
+from script_build_host.domain.errors import ScriptBuildError
+
+_CONTROL_URI_PREFIX = "script-build://"
+_ARTIFACT_URI_PREFIX = "script-build://artifact-versions/"
+_TASK_KIND_URI_PREFIX = "script-build://task-kinds/"
+_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+_LONG_BASE64 = re.compile(r"(?:[A-Za-z0-9+/]{80,}={0,2})")
+
+MAX_PHASE_TWO_TASKS = 64
+MAX_TASK_DEPTH = 8
+MAX_CANDIDATES_PER_SCOPE = 4
+MAX_ATTEMPTS_PER_TASK = 4
+MAX_TASK_TOKENS = 32_000
+MAX_PHASE_TWO_TOKENS = 250_000
+MAX_TASK_SECONDS = 900
+MAX_PHASE_TWO_SECONDS = 3_600
+MAX_EXTERNAL_QUERIES = 40
+MAX_NO_IMPROVEMENT = 3
+MAX_CONTRACT_BYTES = 64 * 1024
+MAX_OBJECTIVE_CHARS = 2_000
+MAX_CRITERIA = 16
+MAX_REFERENCES = 64
+
+
+class TaskContractError(ScriptBuildError):
+    def __init__(self, code: str, summary: str) -> None:
+        super().__init__(code, summary)
+
+
+class ScriptTaskKind(StrEnum):
+    DIRECTION = "direction"
+    PATTERN_RETRIEVAL = "pattern-retrieval"
+    DECODE_RETRIEVAL = "decode-retrieval"
+    EXTERNAL_RETRIEVAL = "external-retrieval"
+    KNOWLEDGE_RETRIEVAL = "knowledge-retrieval"
+    STRUCTURE = "structure"
+    PARAGRAPH = "paragraph"
+    ELEMENT_SET = "element-set"
+    COMPARE = "compare"
+    COMPOSE = "compose"
+    CANDIDATE_PORTFOLIO = "candidate-portfolio"
+
+
+class ScriptIntentClass(StrEnum):
+    EXPLORE = "explore"
+    EXPAND = "expand"
+    REPLACE = "replace"
+    REPAIR = "repair"
+    COMPARE = "compare"
+    COMPOSE = "compose"
+    PORTFOLIO = "portfolio"
+    DELIVER = "deliver"
+
+
+_OUTPUT_SCHEMAS: dict[ScriptTaskKind, frozenset[str]] = {
+    ScriptTaskKind.DIRECTION: frozenset({"script-direction/v1"}),
+    ScriptTaskKind.PATTERN_RETRIEVAL: frozenset({"evidence-record/v1"}),
+    ScriptTaskKind.DECODE_RETRIEVAL: frozenset({"evidence-record/v1"}),
+    ScriptTaskKind.EXTERNAL_RETRIEVAL: frozenset({"evidence-record/v1"}),
+    ScriptTaskKind.KNOWLEDGE_RETRIEVAL: frozenset({"evidence-record/v1"}),
+    ScriptTaskKind.STRUCTURE: frozenset({"structure-artifact/v1"}),
+    ScriptTaskKind.PARAGRAPH: frozenset({"paragraph-artifact/v1", "paragraph-patch/v1"}),
+    ScriptTaskKind.ELEMENT_SET: frozenset({"element-set-artifact/v1", "element-set-patch/v1"}),
+    ScriptTaskKind.COMPARE: frozenset({"comparison-artifact/v1"}),
+    ScriptTaskKind.COMPOSE: frozenset({"structured-script/v1"}),
+    ScriptTaskKind.CANDIDATE_PORTFOLIO: frozenset({"candidate-portfolio/v1"}),
+}
+
+_ADOPTION_FIELDS_KINDS = frozenset({ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO})
+_COMPARISON_FIELDS_KINDS = frozenset({ScriptTaskKind.COMPARE, ScriptTaskKind.COMPOSE})
+
+
+@dataclass(frozen=True, slots=True)
+class PhaseTwoLimits:
+    """Deployment limits. Values may be lowered, never raised above the product ceiling."""
+
+    max_tasks: int = MAX_PHASE_TWO_TASKS
+    max_depth: int = MAX_TASK_DEPTH
+    max_candidates_per_scope: int = MAX_CANDIDATES_PER_SCOPE
+    max_attempts_per_task: int = MAX_ATTEMPTS_PER_TASK
+    max_concurrent_operations: int = 4
+    max_task_tokens: int = MAX_TASK_TOKENS
+    max_total_tokens: int = MAX_PHASE_TWO_TOKENS
+    max_task_seconds: int = MAX_TASK_SECONDS
+    max_total_seconds: int = MAX_PHASE_TWO_SECONDS
+    max_external_queries: int = MAX_EXTERNAL_QUERIES
+    max_no_improvement: int = MAX_NO_IMPROVEMENT
+    max_repairs_per_trace: int = 1
+
+    def __post_init__(self) -> None:
+        ceilings = {
+            "max_tasks": MAX_PHASE_TWO_TASKS,
+            "max_depth": MAX_TASK_DEPTH,
+            "max_candidates_per_scope": MAX_CANDIDATES_PER_SCOPE,
+            "max_attempts_per_task": MAX_ATTEMPTS_PER_TASK,
+            "max_concurrent_operations": 4,
+            "max_task_tokens": MAX_TASK_TOKENS,
+            "max_total_tokens": MAX_PHASE_TWO_TOKENS,
+            "max_task_seconds": MAX_TASK_SECONDS,
+            "max_total_seconds": MAX_PHASE_TWO_SECONDS,
+            "max_external_queries": MAX_EXTERNAL_QUERIES,
+            "max_no_improvement": MAX_NO_IMPROVEMENT,
+            "max_repairs_per_trace": 1,
+        }
+        for name, ceiling in ceilings.items():
+            value = getattr(self, name)
+            if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= ceiling:
+                raise TaskContractError(
+                    "TASK_BUDGET_EXCEEDED", f"{name} must be between 1 and {ceiling}"
+                )
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptCriterion:
+    criterion_id: str
+    description: str
+    hard: bool = True
+
+    def __post_init__(self) -> None:
+        _require_identifier(self.criterion_id, "criterion_id")
+        _require_bounded_text(self.description, "criterion description", 1_000)
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "criterion_id": self.criterion_id,
+            "description": self.description,
+            "hard": self.hard,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> ScriptCriterion:
+        return cls(
+            criterion_id=str(value.get("criterion_id", "")),
+            description=str(value.get("description", "")),
+            hard=bool(value.get("hard", True)),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptTaskBudget:
+    max_attempts: int = MAX_ATTEMPTS_PER_TASK
+    max_tokens: int = MAX_TASK_TOKENS
+    max_seconds: int = MAX_TASK_SECONDS
+    max_external_queries: int = MAX_EXTERNAL_QUERIES
+    max_no_improvement: int = MAX_NO_IMPROVEMENT
+
+    def __post_init__(self) -> None:
+        for label, value, ceiling in (
+            ("max_attempts", self.max_attempts, MAX_ATTEMPTS_PER_TASK),
+            ("max_tokens", self.max_tokens, MAX_TASK_TOKENS),
+            ("max_seconds", self.max_seconds, MAX_TASK_SECONDS),
+            ("max_external_queries", self.max_external_queries, MAX_EXTERNAL_QUERIES),
+            ("max_no_improvement", self.max_no_improvement, MAX_NO_IMPROVEMENT),
+        ):
+            if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= ceiling:
+                raise TaskContractError(
+                    "TASK_BUDGET_EXCEEDED", f"{label} must be between 1 and {ceiling}"
+                )
+
+    def to_payload(self) -> dict[str, int]:
+        return {
+            "max_attempts": self.max_attempts,
+            "max_tokens": self.max_tokens,
+            "max_seconds": self.max_seconds,
+            "max_external_queries": self.max_external_queries,
+            "max_no_improvement": self.max_no_improvement,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> ScriptTaskBudget:
+        try:
+            return cls(
+                max_attempts=int(value.get("max_attempts", MAX_ATTEMPTS_PER_TASK)),
+                max_tokens=int(value.get("max_tokens", MAX_TASK_TOKENS)),
+                max_seconds=int(value.get("max_seconds", MAX_TASK_SECONDS)),
+                max_external_queries=int(value.get("max_external_queries", MAX_EXTERNAL_QUERIES)),
+                max_no_improvement=int(value.get("max_no_improvement", MAX_NO_IMPROVEMENT)),
+            )
+        except (TypeError, ValueError) as exc:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "task budget contains a non-integer value"
+            ) from exc
+
+
+@dataclass(frozen=True, slots=True)
+class AcceptedDecisionRef:
+    decision_id: str
+    artifact_ref: ArtifactRef
+    scope_ref: str
+    expected_task_kind: ScriptTaskKind
+
+    def __post_init__(self) -> None:
+        _require_identifier(self.decision_id, "decision_id")
+        _require_control_uri(self.scope_ref, "scope_ref")
+        _validate_artifact_ref(self.artifact_ref)
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "decision_id": self.decision_id,
+            "artifact_ref": _artifact_ref_payload(self.artifact_ref),
+            "scope_ref": self.scope_ref,
+            "expected_task_kind": self.expected_task_kind.value,
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> AcceptedDecisionRef:
+        raw_ref = value.get("artifact_ref")
+        if not isinstance(raw_ref, Mapping):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "accepted decision artifact_ref must be an object"
+            )
+        return cls(
+            decision_id=str(value.get("decision_id", "")),
+            artifact_ref=ArtifactRef.from_dict(dict(raw_ref)),
+            scope_ref=str(value.get("scope_ref", "")),
+            expected_task_kind=_enum_value(
+                ScriptTaskKind,
+                value.get("expected_task_kind"),
+                "expected_task_kind",
+            ),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptTaskContractV1:
+    task_kind: ScriptTaskKind
+    scope_ref: str
+    intent_class: ScriptIntentClass
+    objective: str
+    input_decision_refs: tuple[AcceptedDecisionRef, ...]
+    base_artifact_ref: ArtifactRef | None
+    write_scope: tuple[str, ...]
+    gap_ref: str | None
+    output_schema: str
+    criteria: tuple[ScriptCriterion, ...]
+    budget: ScriptTaskBudget
+    supersedes_decision_ids: tuple[str, ...] = ()
+    candidate_closure_decision_refs: tuple[AcceptedDecisionRef, ...] = ()
+    adopted_decision_ids: tuple[str, ...] = ()
+    held_or_rejected_decision_ids: tuple[str, ...] = ()
+    compose_order: tuple[str, ...] = ()
+    comparison_decision_refs: tuple[AcceptedDecisionRef, ...] = ()
+
+    def __post_init__(self) -> None:
+        _require_control_uri(self.scope_ref, "scope_ref")
+        _require_bounded_text(self.objective, "objective", MAX_OBJECTIVE_CHARS)
+        if self.output_schema not in _OUTPUT_SCHEMAS[self.task_kind]:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                "output_schema does not match the declared task kind",
+            )
+        if not self.criteria or len(self.criteria) > MAX_CRITERIA:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", f"criteria must contain between 1 and {MAX_CRITERIA} items"
+            )
+        _require_unique((item.criterion_id for item in self.criteria), "criterion IDs")
+        for item in self.write_scope:
+            _require_control_uri(item, "write_scope")
+        _require_unique(self.write_scope, "write_scope")
+        if self.gap_ref is not None:
+            _require_control_uri(self.gap_ref, "gap_ref")
+        if self.base_artifact_ref is not None:
+            _validate_artifact_ref(self.base_artifact_ref)
+        for label, values in (
+            ("input decision IDs", (item.decision_id for item in self.input_decision_refs)),
+            ("supersedes decision IDs", self.supersedes_decision_ids),
+            (
+                "candidate closure decision IDs",
+                (item.decision_id for item in self.candidate_closure_decision_refs),
+            ),
+            ("adopted decision IDs", self.adopted_decision_ids),
+            ("held or rejected decision IDs", self.held_or_rejected_decision_ids),
+            (
+                "comparison decision IDs",
+                (item.decision_id for item in self.comparison_decision_refs),
+            ),
+            ("compose order", self.compose_order),
+        ):
+            _require_unique(values, label)
+        for decision_id in (
+            *self.supersedes_decision_ids,
+            *self.adopted_decision_ids,
+            *self.held_or_rejected_decision_ids,
+        ):
+            _require_identifier(decision_id, "decision_id")
+        if set(self.adopted_decision_ids) & set(self.held_or_rejected_decision_ids):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "adopted and held/rejected decisions must be disjoint"
+            )
+        has_adoption_fields = any(
+            (
+                self.candidate_closure_decision_refs,
+                self.adopted_decision_ids,
+                self.held_or_rejected_decision_ids,
+                self.compose_order,
+            )
+        )
+        if has_adoption_fields and self.task_kind not in _ADOPTION_FIELDS_KINDS:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "candidate adoption fields require compose or portfolio"
+            )
+        if self.comparison_decision_refs and self.task_kind not in _COMPARISON_FIELDS_KINDS:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                "comparison_decision_refs require a compare or compose contract",
+            )
+        closure_ids = {item.decision_id for item in self.candidate_closure_decision_refs}
+        disposed_ids = set(self.adopted_decision_ids) | set(self.held_or_rejected_decision_ids)
+        if disposed_ids and not disposed_ids <= closure_ids:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "candidate disposition is outside the declared closure"
+            )
+        if self.compose_order and not set(self.compose_order) <= set(self.adopted_decision_ids):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "compose_order may contain adopted decisions only"
+            )
+        if self.task_kind is ScriptTaskKind.COMPARE:
+            if len(self.comparison_decision_refs) < 2:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "Compare requires at least two frozen candidates"
+                )
+            if any(
+                item.expected_task_kind
+                in {ScriptTaskKind.COMPARE, ScriptTaskKind.CANDIDATE_PORTFOLIO}
+                for item in self.comparison_decision_refs
+            ):
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "Compare inputs must be creative candidates"
+                )
+        elif self.task_kind is ScriptTaskKind.COMPOSE and any(
+            item.expected_task_kind is not ScriptTaskKind.COMPARE
+            for item in self.comparison_decision_refs
+        ):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                "Compose comparison_decision_refs must pin accepted Comparison decisions",
+            )
+        reference_count = (
+            len(self.input_decision_refs)
+            + len(self.supersedes_decision_ids)
+            + len(self.candidate_closure_decision_refs)
+            + len(self.adopted_decision_ids)
+            + len(self.held_or_rejected_decision_ids)
+            + len(self.comparison_decision_refs)
+        )
+        if reference_count > MAX_REFERENCES:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                f"a contract may contain at most {MAX_REFERENCES} references",
+            )
+        encoded = json.dumps(
+            self.to_payload(),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        if len(encoded) > MAX_CONTRACT_BYTES:
+            raise TaskContractError(
+                "TASK_CONTRACT_TOO_LARGE", "canonical task contract exceeds 64 KiB"
+            )
+
+    @property
+    def execution_ready(self) -> bool:
+        if self.task_kind not in _ADOPTION_FIELDS_KINDS:
+            return True
+        closure = {item.decision_id for item in self.candidate_closure_decision_refs}
+        disposed = set(self.adopted_decision_ids) | set(self.held_or_rejected_decision_ids)
+        adopted = set(self.adopted_decision_ids)
+        order = set(self.compose_order)
+        if not closure or closure != disposed or not adopted or order != adopted:
+            return False
+        if len(self.compose_order) != len(self.adopted_decision_ids):
+            return False
+        if self.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO:
+            return len(self.adopted_decision_ids) == 1
+        return True
+
+    def require_execution_ready(self) -> None:
+        if not self.execution_ready:
+            raise TaskContractError(
+                "TASK_CONTRACT_NOT_EXECUTABLE",
+                "compose or portfolio contract has not closed its candidate disposition",
+            )
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "script-task-contract/v1",
+            "task_kind": self.task_kind.value,
+            "scope_ref": self.scope_ref,
+            "intent_class": self.intent_class.value,
+            "objective": self.objective,
+            "input_decision_refs": [item.to_payload() for item in self.input_decision_refs],
+            "base_artifact_ref": (
+                _artifact_ref_payload(self.base_artifact_ref)
+                if self.base_artifact_ref is not None
+                else None
+            ),
+            "write_scope": list(self.write_scope),
+            "gap_ref": self.gap_ref,
+            "output_schema": self.output_schema,
+            "criteria": [item.to_payload() for item in self.criteria],
+            "budget": self.budget.to_payload(),
+            "supersedes_decision_ids": list(self.supersedes_decision_ids),
+            "candidate_closure_decision_refs": [
+                item.to_payload() for item in self.candidate_closure_decision_refs
+            ],
+            "adopted_decision_ids": list(self.adopted_decision_ids),
+            "held_or_rejected_decision_ids": list(self.held_or_rejected_decision_ids),
+            "compose_order": list(self.compose_order),
+            "comparison_decision_refs": [
+                item.to_payload() for item in self.comparison_decision_refs
+            ],
+        }
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> ScriptTaskContractV1:
+        if value.get("schema_version") != "script-task-contract/v1":
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "unsupported task contract schema version"
+            )
+        base = value.get("base_artifact_ref")
+        if base is not None and not isinstance(base, Mapping):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "base_artifact_ref must be an object or null"
+            )
+        budget = value.get("budget")
+        if not isinstance(budget, Mapping):
+            raise TaskContractError("TASK_CONTRACT_INVALID", "task budget must be an object")
+        return cls(
+            task_kind=_enum_value(ScriptTaskKind, value.get("task_kind"), "task_kind"),
+            scope_ref=str(value.get("scope_ref", "")),
+            intent_class=_enum_value(ScriptIntentClass, value.get("intent_class"), "intent_class"),
+            objective=str(value.get("objective", "")),
+            input_decision_refs=_decision_refs(value.get("input_decision_refs")),
+            base_artifact_ref=(ArtifactRef.from_dict(dict(base)) if base is not None else None),
+            write_scope=_strings(value.get("write_scope"), "write_scope"),
+            gap_ref=(str(value["gap_ref"]) if value.get("gap_ref") is not None else None),
+            output_schema=str(value.get("output_schema", "")),
+            criteria=_criteria(value.get("criteria")),
+            budget=ScriptTaskBudget.from_payload(budget),
+            supersedes_decision_ids=_strings(
+                value.get("supersedes_decision_ids"), "supersedes_decision_ids"
+            ),
+            candidate_closure_decision_refs=_decision_refs(
+                value.get("candidate_closure_decision_refs")
+            ),
+            adopted_decision_ids=_strings(
+                value.get("adopted_decision_ids"), "adopted_decision_ids"
+            ),
+            held_or_rejected_decision_ids=_strings(
+                value.get("held_or_rejected_decision_ids"),
+                "held_or_rejected_decision_ids",
+            ),
+            compose_order=_strings(value.get("compose_order"), "compose_order"),
+            comparison_decision_refs=_decision_refs(value.get("comparison_decision_refs")),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class FrozenTaskContract:
+    uri: str
+    digest: str
+    canonical_size: int
+    contract: ScriptTaskContractV1
+
+    def __post_init__(self) -> None:
+        Sha256Digest.from_wire(self.digest)
+
+
+@dataclass(frozen=True, slots=True)
+class AcceptedInput:
+    decision_id: str
+    artifact_ref: ArtifactRef
+    task_kind: ScriptTaskKind
+    scope_ref: str
+    source: str
+
+
+@dataclass(frozen=True, slots=True)
+class AcceptedInputBundleV1:
+    inputs: tuple[AcceptedInput, ...]
+    input_closure_digest: str
+    superseded_decision_ids: tuple[str, ...] = field(default_factory=tuple)
+
+    def __post_init__(self) -> None:
+        Sha256Digest.from_wire(self.input_closure_digest)
+
+
+def task_kind_from_context_refs(context_refs: tuple[str, ...] | list[str]) -> str | None:
+    values = [
+        item.removeprefix(_TASK_KIND_URI_PREFIX)
+        for item in context_refs
+        if item.startswith(_TASK_KIND_URI_PREFIX)
+    ]
+    if len(values) > 1:
+        raise TaskContractError(
+            "TASK_KIND_PRESET_MISMATCH", "TaskSpec contains multiple controlled Task kinds"
+        )
+    return values[0] if values else None
+
+
+def _artifact_ref_payload(value: ArtifactRef) -> dict[str, Any]:
+    return {
+        "uri": value.uri,
+        "kind": value.kind,
+        "version": value.version,
+        "digest": value.digest,
+        "summary": value.summary,
+        "metadata": value.metadata,
+    }
+
+
+def _validate_artifact_ref(value: ArtifactRef) -> None:
+    if not value.uri.startswith(_ARTIFACT_URI_PREFIX):
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", "artifact reference is outside the immutable namespace"
+        )
+    identifier = value.uri.removeprefix(_ARTIFACT_URI_PREFIX)
+    if not identifier.isdigit() or value.version != identifier:
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", "artifact reference must pin one numeric version"
+        )
+    try:
+        Sha256Digest.from_wire(value.digest or "")
+    except ScriptBuildError as exc:
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", "artifact reference must pin a valid sha256 digest"
+        ) from exc
+    if not value.kind.strip():
+        raise TaskContractError("TASK_CONTRACT_INVALID", "artifact kind must not be blank")
+    if value.summary is not None or value.metadata not in (None, {}):
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID",
+            "task contracts may pin Artifact identity only, not summary or metadata",
+        )
+
+
+def _require_control_uri(value: str, label: str) -> None:
+    if not isinstance(value, str) or not value.startswith(_CONTROL_URI_PREFIX):
+        raise TaskContractError("TASK_CONTRACT_INVALID", f"{label} must use script-build://")
+    _require_bounded_text(value, label, 1_000)
+
+
+def _require_bounded_text(value: str, label: str, maximum: int) -> None:
+    if not isinstance(value, str) or not value.strip() or len(value) > maximum:
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", f"{label} must contain between 1 and {maximum} characters"
+        )
+    lowered = value.casefold()
+    if "data:" in lowered or _LONG_BASE64.search(value):
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", f"{label} must not embed encoded candidate content"
+        )
+
+
+def _require_identifier(value: str, label: str) -> None:
+    if not _IDENTIFIER.fullmatch(value):
+        raise TaskContractError("TASK_CONTRACT_INVALID", f"{label} contains an invalid identifier")
+
+
+def _require_unique(values: Any, label: str) -> None:
+    materialized = tuple(values)
+    if len(set(materialized)) != len(materialized):
+        raise TaskContractError("TASK_CONTRACT_INVALID", f"{label} must be unique")
+
+
+def _strings(value: Any, label: str) -> tuple[str, ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
+        raise TaskContractError("TASK_CONTRACT_INVALID", f"{label} must be a string array")
+    return tuple(value)
+
+
+def _decision_refs(value: Any) -> tuple[AcceptedDecisionRef, ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, list) or any(not isinstance(item, Mapping) for item in value):
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", "accepted decision references must be an object array"
+        )
+    return tuple(AcceptedDecisionRef.from_payload(item) for item in value)
+
+
+def _criteria(value: Any) -> tuple[ScriptCriterion, ...]:
+    if not isinstance(value, list) or any(not isinstance(item, Mapping) for item in value):
+        raise TaskContractError("TASK_CONTRACT_INVALID", "criteria must be an object array")
+    return tuple(ScriptCriterion.from_payload(item) for item in value)
+
+
+def _enum_value(enum_type: type[Any], value: Any, label: str) -> Any:
+    try:
+        return enum_type(str(value))
+    except ValueError as exc:
+        raise TaskContractError("TASK_CONTRACT_INVALID", f"unsupported {label}") from exc
+
+
+__all__ = [
+    "MAX_ATTEMPTS_PER_TASK",
+    "MAX_CANDIDATES_PER_SCOPE",
+    "MAX_CONTRACT_BYTES",
+    "MAX_CRITERIA",
+    "MAX_OBJECTIVE_CHARS",
+    "MAX_PHASE_TWO_SECONDS",
+    "MAX_PHASE_TWO_TASKS",
+    "MAX_PHASE_TWO_TOKENS",
+    "MAX_REFERENCES",
+    "MAX_TASK_DEPTH",
+    "AcceptedDecisionRef",
+    "AcceptedInput",
+    "AcceptedInputBundleV1",
+    "FrozenTaskContract",
+    "PhaseTwoLimits",
+    "ScriptCriterion",
+    "ScriptIntentClass",
+    "ScriptTaskBudget",
+    "ScriptTaskContractV1",
+    "ScriptTaskKind",
+    "TaskContractError",
+    "task_kind_from_context_refs",
+]

script_build_host/src/script_build_host/domain/task_refs.py

@@ -0,0 +1,21 @@
+"""Immutable script-build Task reference parsing."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+TASK_KIND_PREFIX = "script-build://task-kinds/"
+
+
+def task_kind(context_refs: Sequence[str]) -> str | None:
+    values = [
+        item.removeprefix(TASK_KIND_PREFIX)
+        for item in context_refs
+        if item.startswith(TASK_KIND_PREFIX)
+    ]
+    if len(values) > 1:
+        raise ValueError("TaskSpec contains multiple script-build task kinds")
+    return values[0] if values else None
+
+
+__all__ = ["TASK_KIND_PREFIX", "task_kind"]

script_build_host/src/script_build_host/domain/workspaces.py

@@ -0,0 +1,70 @@
+"""Attempt-owned candidate workspace value objects."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .artifacts import ArtifactKind, ArtifactState
+from .errors import ScriptBuildError
+from .phase_two_artifacts import (
+    ScriptElementV1,
+    ScriptParagraphElementLinkV1,
+    ScriptParagraphV1,
+)
+
+
+class WorkspaceError(ScriptBuildError):
+    def __init__(self, code: str, summary: str) -> None:
+        super().__init__(code, summary)
+
+
+@dataclass(frozen=True, slots=True)
+class CandidateWorkspace:
+    artifact_version_id: int
+    script_build_id: int
+    task_id: str
+    attempt_id: str
+    spec_version: int
+    artifact_kind: ArtifactKind
+    branch_id: int
+    state: ArtifactState
+    base_revision: int | None = None
+    parent_artifact_version_id: int | None = None
+
+    def __post_init__(self) -> None:
+        if self.artifact_version_id < 1 or self.branch_id < 1:
+            raise WorkspaceError(
+                "LEGACY_WRITE_INVALID", "candidate workspace must use a positive branch identity"
+            )
+        if self.branch_id != self.artifact_version_id:
+            raise WorkspaceError(
+                "LEGACY_WRITE_INVALID", "candidate branch must equal its artifact version identity"
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class CandidateWorkspaceSnapshot:
+    workspace: CandidateWorkspace
+    paragraphs: tuple[ScriptParagraphV1, ...]
+    elements: tuple[ScriptElementV1, ...]
+    links: tuple[ScriptParagraphElementLinkV1, ...]
+    source_identity_map: tuple[dict[str, int | str], ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class CandidateWriteContext:
+    script_build_id: int
+    task_id: str
+    attempt_id: str
+    spec_version: int
+    objective: str
+    input_refs: tuple[str, ...]
+    write_scope: tuple[str, ...]
+
+
+__all__ = [
+    "CandidateWorkspace",
+    "CandidateWorkspaceSnapshot",
+    "CandidateWriteContext",
+    "WorkspaceError",
+]

script_build_host/src/script_build_host/infrastructure/canonical_json.py

@@ -1,74 +1,10 @@
-from __future__ import annotations
+"""Compatibility exports for adapters that historically imported this module."""
 
-import hashlib
-import json
-import math
-from dataclasses import asdict, is_dataclass
-from datetime import UTC, date, datetime
-from decimal import Decimal
-from enum import Enum
-from typing import Any
+from script_build_host.domain.canonical_json import (
+    canonical_json_bytes,
+    canonical_json_text,
+    canonical_sha256,
+    normalize_json,
+)
 
-from script_build_host.domain.digests import Sha256Digest
-from script_build_host.domain.errors import ProtocolViolation
-
-
-def _stable_decimal(value: Decimal) -> str:
-    if not value.is_finite():
-        raise ProtocolViolation("canonical JSON does not permit non-finite numbers")
-    rendered = format(value, "f")
-    if "." in rendered:
-        rendered = rendered.rstrip("0").rstrip(".")
-    return "0" if rendered in {"-0", ""} else rendered
-
-
-def normalize_json(value: Any) -> Any:
-    """Return a JSON value with deterministic dates, floats, keys and containers."""
-
-    if value is None or isinstance(value, (str, bool, int)):
-        return value
-    if isinstance(value, float):
-        if not math.isfinite(value):
-            raise ProtocolViolation("canonical JSON does not permit non-finite numbers")
-        return _stable_decimal(Decimal(str(value)))
-    if isinstance(value, Decimal):
-        return _stable_decimal(value)
-    if isinstance(value, datetime):
-        if value.tzinfo is None:
-            return value.isoformat(timespec="microseconds")
-        normalized = value.astimezone(UTC).isoformat(timespec="microseconds")
-        return normalized.removesuffix("+00:00") + "Z"
-    if isinstance(value, date):
-        return value.isoformat()
-    if isinstance(value, Enum):
-        return normalize_json(value.value)
-    if is_dataclass(value) and not isinstance(value, type):
-        return normalize_json(asdict(value))
-    if isinstance(value, dict):
-        output: dict[str, Any] = {}
-        for key in sorted(value, key=lambda item: str(item)):
-            if not isinstance(key, str):
-                raise ProtocolViolation("canonical JSON object keys must be strings")
-            output[key] = normalize_json(value[key])
-        return output
-    if isinstance(value, (list, tuple)):
-        return [normalize_json(item) for item in value]
-    raise ProtocolViolation(f"unsupported canonical JSON value: {type(value).__name__}")
-
-
-def canonical_json_bytes(value: Any) -> bytes:
-    return json.dumps(
-        normalize_json(value),
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-        allow_nan=False,
-    ).encode("utf-8")
-
-
-def canonical_json_text(value: Any) -> str:
-    return canonical_json_bytes(value).decode("utf-8")
-
-
-def canonical_sha256(value: Any) -> Sha256Digest:
-    return Sha256Digest(hashlib.sha256(canonical_json_bytes(value)).hexdigest())
+__all__ = ["canonical_json_bytes", "canonical_json_text", "canonical_sha256", "normalize_json"]

script_build_host/src/script_build_host/infrastructure/redaction.py

@@ -1,54 +1,5 @@
-from __future__ import annotations
+"""Compatibility exports for adapters that historically imported this module."""
 
-import re
-from collections.abc import Mapping, Sequence
-from typing import Any
-from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+from script_build_host.domain.redaction import redact, redact_text, redact_url
 
-_SENSITIVE_KEY = re.compile(
-    r"(^|[_-])(authorization|cookie|password|passwd|secret|token|api[_-]?key|dsn|database[_-]?url)($|[_-])",
-    re.IGNORECASE,
-)
-_BEARER = re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]+")
-_CONNECTION_URL = re.compile(r"([a-z][a-z0-9+.-]*://)([^\s/@:]+):([^\s/@]+)@", re.IGNORECASE)
-
-
-def redact_text(value: str) -> str:
-    value = _BEARER.sub("Bearer [REDACTED]", value)
-    return _CONNECTION_URL.sub(r"\1[REDACTED]:[REDACTED]@", value)
-
-
-def redact_url(value: str) -> str:
-    try:
-        parts = urlsplit(value)
-    except ValueError:
-        return redact_text(value)
-    if not parts.scheme or not parts.netloc:
-        return redact_text(value)
-    host = parts.hostname or ""
-    try:
-        parsed_port = parts.port
-    except ValueError:
-        return redact_text(value)
-    port = f":{parsed_port}" if parsed_port is not None else ""
-    netloc = f"[REDACTED]@{host}{port}" if parts.username is not None else parts.netloc
-    query = urlencode(
-        [
-            (key, "[REDACTED]" if _SENSITIVE_KEY.search(key) else item)
-            for key, item in parse_qsl(parts.query, keep_blank_values=True)
-        ]
-    )
-    return urlunsplit((parts.scheme, netloc, parts.path, query, ""))
-
-
-def redact(value: Any) -> Any:
-    if isinstance(value, str):
-        return redact_url(value) if "://" in value else redact_text(value)
-    if isinstance(value, Mapping):
-        return {
-            str(key): "[REDACTED]" if _SENSITIVE_KEY.search(str(key)) else redact(item)
-            for key, item in value.items()
-        }
-    if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):
-        return [redact(item) for item in value]
-    return value
+__all__ = ["redact", "redact_text", "redact_url"]