领域：定义阶段一输入、制品和运行安全合同

建立输入快照、Evidence、Direction、Mission binding、publication、Principal 和状态领域模型。

实现规范摘要、配置校验、数据源清单、敏感信息脱敏和 HTTPS 出站策略，并覆盖序列化与安全边界测试。

script_build_host/src/script_build_host/domain/__init__.py

@@ -0,0 +1,29 @@
+from .artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ArtifactVersion,
+    Criterion,
+    DirectionGoal,
+    EvidenceRecordV1,
+    ScriptDirectionArtifactV1,
+)
+from .errors import ScriptBuildError
+from .input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
+from .records import BuildStatus, MissionBinding, Principal, Publication
+
+__all__ = [
+    "ArtifactKind",
+    "ArtifactState",
+    "ArtifactVersion",
+    "BuildStatus",
+    "Criterion",
+    "DirectionGoal",
+    "EvidenceRecordV1",
+    "MissionBinding",
+    "Principal",
+    "Publication",
+    "ScriptBuildError",
+    "ScriptBuildInput",
+    "ScriptBuildInputSnapshotV1",
+    "ScriptDirectionArtifactV1",
+]

script_build_host/src/script_build_host/domain/artifacts.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from enum import StrEnum
+from typing import Any
+
+
+class ArtifactKind(StrEnum):
+    EVIDENCE = "evidence"
+    DIRECTION = "direction"
+
+
+class ArtifactState(StrEnum):
+    DRAFT = "draft"
+    FROZEN = "frozen"
+    PUBLISHED = "published"
+    DISCARDED = "discarded"
+
+
+@dataclass(frozen=True, slots=True)
+class DirectionGoal:
+    goal_id: str
+    statement: str
+    rationale: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class Criterion:
+    criterion_id: str
+    description: str
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptDirectionArtifactV1:
+    goals: tuple[DirectionGoal, ...]
+    criteria: tuple[Criterion, ...] = ()
+    domain_criteria: tuple[Criterion, ...] = ()
+    topic_refs: tuple[str, ...] = ()
+    persona_refs: tuple[str, ...] = ()
+    strategy_refs: tuple[str, ...] = ()
+    evidence_refs: tuple[str, ...] = ()
+    legacy_markdown: str = ""
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        if not 1 <= len(self.goals) <= 3:
+            raise ValueError("a direction artifact must contain between one and three goals")
+        _require_unique_text(
+            ((item.goal_id, item.statement) for item in self.goals),
+            label="direction goal",
+        )
+        _require_unique_text(
+            ((item.criterion_id, item.description) for item in self.criteria),
+            label="direction criterion",
+        )
+        _require_unique_text(
+            ((item.criterion_id, item.description) for item in self.domain_criteria),
+            label="domain criterion",
+        )
+        for label, refs in (
+            ("topic", self.topic_refs),
+            ("persona", self.persona_refs),
+            ("strategy", self.strategy_refs),
+            ("evidence", self.evidence_refs),
+        ):
+            if any(not isinstance(ref, str) or not ref.strip() for ref in refs):
+                raise ValueError(f"{label} references must be non-empty strings")
+            if len(set(refs)) != len(refs):
+                raise ValueError(f"{label} references must be unique")
+        if any(
+            not ref.startswith("script-build://artifact-versions/") for ref in self.evidence_refs
+        ):
+            raise ValueError("evidence references must use immutable business Artifact URIs")
+        if not self.evidence_refs:
+            raise ValueError("a direction artifact must cite accepted retrieval evidence")
+        if not self.legacy_markdown.strip():
+            raise ValueError("legacy direction markdown must not be blank")
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "script-direction/v1",
+            "goals": [
+                {"goal_id": item.goal_id, "statement": item.statement, "rationale": item.rationale}
+                for item in self.goals
+            ],
+            "criteria": [
+                {"criterion_id": item.criterion_id, "description": item.description}
+                for item in self.criteria
+            ],
+            "domain_criteria": [
+                {"criterion_id": item.criterion_id, "description": item.description}
+                for item in self.domain_criteria
+            ],
+            "topic_refs": list(self.topic_refs),
+            "persona_refs": list(self.persona_refs),
+            "strategy_refs": list(self.strategy_refs),
+            "evidence_refs": list(self.evidence_refs),
+            "legacy_markdown": self.legacy_markdown,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceRecordV1:
+    evidence_id: str
+    source_type: str
+    tool_name: str
+    query: dict[str, Any]
+    source_refs: tuple[str, ...]
+    raw_artifact_ref: str | None
+    summary: str
+    supports: tuple[str, ...]
+    confidence: str
+    limitations: tuple[str, ...]
+    content_sha256: str
+    created_at: datetime
+
+    def __post_init__(self) -> None:
+        for label, value in (
+            ("evidence_id", self.evidence_id),
+            ("source_type", self.source_type),
+            ("tool_name", self.tool_name),
+        ):
+            if not value.strip():
+                raise ValueError(f"{label} must not be blank")
+        if any(not ref.strip() for ref in self.source_refs):
+            raise ValueError("evidence source references must not be blank")
+        if len(set(self.source_refs)) != len(self.source_refs):
+            raise ValueError("evidence source references must be unique")
+        if not self.summary.strip() or not self.confidence.strip():
+            raise ValueError("evidence summary and confidence must not be blank")
+        if any(not item.strip() for item in (*self.supports, *self.limitations)):
+            raise ValueError("evidence supports and limitations must not contain blank values")
+        if len(set(self.supports)) != len(self.supports):
+            raise ValueError("evidence supports must be unique")
+        if self.raw_artifact_ref is not None and not self.raw_artifact_ref.startswith(
+            "script-build://raw-artifacts/sha256/"
+        ):
+            raise ValueError("raw artifact references must use the immutable raw namespace")
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "evidence-record/v1",
+            "evidence_id": self.evidence_id,
+            "source_type": self.source_type,
+            "tool_name": self.tool_name,
+            "query": self.query,
+            "source_refs": list(self.source_refs),
+            "raw_artifact_ref": self.raw_artifact_ref,
+            "summary": self.summary,
+            "supports": list(self.supports),
+            "confidence": self.confidence,
+            "limitations": list(self.limitations),
+            "created_at": self.created_at,
+        }
+
+
+BusinessArtifact = EvidenceRecordV1 | ScriptDirectionArtifactV1
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactVersion:
+    artifact_version_id: int
+    script_build_id: int
+    task_id: str
+    attempt_id: str
+    spec_version: int
+    artifact_type: ArtifactKind
+    canonical_sha256: str
+    state: ArtifactState
+    artifact: BusinessArtifact
+    created_at: datetime
+    frozen_at: datetime
+
+
+def _require_unique_text(
+    values: Any,
+    *,
+    label: str,
+) -> None:
+    identifiers: list[str] = []
+    for identifier, text in values:
+        if not identifier.strip() or not text.strip():
+            raise ValueError(f"{label} identity and content must not be blank")
+        identifiers.append(identifier)
+    if len(set(identifiers)) != len(identifiers):
+        raise ValueError(f"{label} IDs must be unique")

script_build_host/src/script_build_host/domain/digests.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+from .errors import ProtocolViolation
+
+_BARE_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+
+
+@dataclass(frozen=True, slots=True)
+class Sha256Digest:
+    """One digest semantics with explicit wire/DB conversion boundaries."""
+
+    hex_value: str
+
+    def __post_init__(self) -> None:
+        if not _BARE_SHA256.fullmatch(self.hex_value):
+            raise ProtocolViolation(
+                "sha256 digest must be exactly 64 lowercase hexadecimal characters"
+            )
+
+    @property
+    def wire(self) -> str:
+        return f"sha256:{self.hex_value}"
+
+    @classmethod
+    def from_wire(cls, value: str) -> Sha256Digest:
+        algorithm, separator, digest = value.partition(":")
+        if separator != ":" or algorithm != "sha256":
+            raise ProtocolViolation("artifact digest must use the sha256 algorithm")
+        return cls(digest)
+
+    @classmethod
+    def from_db(cls, value: str) -> Sha256Digest:
+        return cls(value)

script_build_host/src/script_build_host/domain/errors.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(eq=False)
+class ScriptBuildError(Exception):
+    """Stable domain error safe to map at an HTTP or Agent tool boundary."""
+
+    code: str
+    summary: str
+
+    def __str__(self) -> str:
+        return f"{self.code}: {self.summary}"
+
+
+class InputRelationMismatch(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "INPUT_RELATION_MISMATCH", "the requested execution, build and topic do not match"
+        )
+
+
+class InputHashConflict(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "INPUT_HASH_CONFLICT", "the input version is already frozen with different content"
+        )
+
+
+class BuildNotFound(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__("BUILD_NOT_FOUND", "the requested script build resource does not exist")
+
+
+class ArtifactNotFound(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__("ARTIFACT_NOT_FOUND", "the requested immutable artifact does not exist")
+
+
+class ArtifactDigestMismatch(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "ARTIFACT_DIGEST_MISMATCH", "the artifact digest does not match its frozen content"
+        )
+
+
+class ArtifactOwnershipMismatch(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "ARTIFACT_OWNERSHIP_MISMATCH", "the artifact is outside the authorized context"
+        )
+
+
+class ArtifactAlreadyFrozen(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "ATTEMPT_WORKSPACE_FROZEN", "the attempt already owns different frozen content"
+        )
+
+
+class UnsafeOutboundTarget(ScriptBuildError):
+    def __init__(self, summary: str = "the outbound target is not permitted") -> None:
+        super().__init__("UNSAFE_OUTBOUND_TARGET", summary)
+
+
+class InvalidConfiguration(ScriptBuildError):
+    def __init__(self, summary: str) -> None:
+        super().__init__("INVALID_CONFIGURATION", summary)
+
+
+class ProtocolViolation(ScriptBuildError):
+    def __init__(self, summary: str) -> None:
+        super().__init__("PROTOCOL_VIOLATION", summary)

script_build_host/src/script_build_host/domain/input_snapshot.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptBuildInput:
+    script_build_id: int
+    execution_id: int
+    topic_build_id: int
+    topic_id: int
+    topic: dict[str, Any]
+    account: dict[str, Any]
+    persona_points: tuple[dict[str, Any], ...] = ()
+    section_patterns: tuple[dict[str, Any], ...] = ()
+    strategies: tuple[dict[str, Any], ...] = ()
+    prompt_manifest: tuple[dict[str, Any], ...] = ()
+    datasource_manifest: dict[str, Any] = field(default_factory=dict)
+    model_manifest: dict[str, Any] = field(default_factory=dict)
+
+    def canonical_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "script-build-input/v1",
+            "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),
+            "prompt_manifest": list(self.prompt_manifest),
+            "datasource_manifest": self.datasource_manifest,
+            "model_manifest": self.model_manifest,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ScriptBuildInputSnapshotV1:
+    snapshot_id: str
+    script_build_id: int
+    execution_id: int
+    topic_build_id: int
+    topic_id: int
+    topic: dict[str, Any]
+    account: dict[str, Any]
+    persona_points: tuple[dict[str, Any], ...]
+    section_patterns: tuple[dict[str, Any], ...]
+    strategies: tuple[dict[str, Any], ...]
+    prompt_manifest: tuple[dict[str, Any], ...]
+    datasource_manifest: dict[str, Any]
+    model_manifest: dict[str, Any]
+    canonical_sha256: str
+    created_at: datetime
+
+    def to_input(self) -> ScriptBuildInput:
+        return ScriptBuildInput(
+            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=self.persona_points,
+            section_patterns=self.section_patterns,
+            strategies=self.strategies,
+            prompt_manifest=self.prompt_manifest,
+            datasource_manifest=self.datasource_manifest,
+            model_manifest=self.model_manifest,
+        )

script_build_host/src/script_build_host/domain/ports.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from agent.orchestration import ArtifactRef
+
+from .artifacts import ArtifactVersion, BusinessArtifact
+from .input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
+from .records import BuildStatus, MissionBinding, Principal, Publication, PublicationType
+
+StrategySelector = int | str | dict[str, Any]
+
+
+@dataclass(frozen=True, slots=True)
+class PersonaInput:
+    requested_account_name: str
+    resolved_account_name: str
+    persona_points: tuple[dict[str, Any], ...]
+    section_patterns: tuple[dict[str, Any], ...]
+    source_versions: dict[str, str]
+
+
+@dataclass(frozen=True, slots=True)
+class PromptRequest:
+    biz_type: str
+    filename: str
+    preset: str
+    role: str
+
+
+class LegacyInputReader(Protocol):
+    async def read_topic_graph(
+        self, *, execution_id: int, topic_build_id: int, topic_id: int
+    ) -> dict[str, Any]: ...
+
+
+class PersonaSource(Protocol):
+    async def load(self, account_name: str) -> PersonaInput: ...
+
+
+class StrategySource(Protocol):
+    async def load(
+        self, *, always_on: tuple[StrategySelector, ...], on_demand: tuple[StrategySelector, ...]
+    ) -> tuple[dict[str, Any], ...]: ...
+
+
+class PromptSource(Protocol):
+    async def load(self, requests: tuple[PromptRequest, ...]) -> tuple[dict[str, Any], ...]: ...
+
+
+class RuntimeManifestProvider(Protocol):
+    async def load(self) -> dict[str, object]: ...
+
+
+class InputSnapshotRepository(Protocol):
+    async def freeze(
+        self, assembled: ScriptBuildInput, *, version: int = 1
+    ) -> ScriptBuildInputSnapshotV1: ...
+
+    async def get(
+        self, snapshot_id: str, *, script_build_id: int
+    ) -> ScriptBuildInputSnapshotV1: ...
+
+
+class MissionBindingRepository(Protocol):
+    async def create(
+        self,
+        *,
+        script_build_id: int,
+        root_trace_id: str,
+        input_snapshot_id: int,
+        engine_version: str,
+        schema_version: str,
+    ) -> MissionBinding: ...
+
+    async def get_by_build(self, script_build_id: int) -> MissionBinding: ...
+
+    async def get_by_root(self, root_trace_id: str) -> MissionBinding: ...
+
+    async def set_active_direction(
+        self, *, script_build_id: int, artifact_version_id: int
+    ) -> MissionBinding: ...
+
+
+class ScriptBusinessArtifactRepository(Protocol):
+    async def freeze(
+        self,
+        *,
+        script_build_id: int,
+        task_id: str,
+        attempt_id: str,
+        spec_version: int,
+        artifact: BusinessArtifact,
+    ) -> tuple[ArtifactVersion, ArtifactRef]: ...
+
+    async def read_by_ref(
+        self,
+        artifact_ref: ArtifactRef,
+        *,
+        script_build_id: int,
+        task_id: str | None = None,
+        attempt_id: str | None = None,
+    ) -> ArtifactVersion: ...
+
+    async def verify_digest(self, artifact_ref: ArtifactRef, *, script_build_id: int) -> bool: ...
+
+    async def get_by_id(
+        self, artifact_version_id: int, *, script_build_id: int
+    ) -> ArtifactVersion: ...
+
+
+class PublicationRepository(Protocol):
+    async def prepare(
+        self,
+        *,
+        script_build_id: int,
+        publication_type: PublicationType,
+        accept_decision_id: str,
+        artifact_version_id: int,
+        expected_sha256: str,
+    ) -> Publication: ...
+
+    async def mark_published(self, publication_id: int) -> Publication: ...
+
+    async def mark_failed(
+        self, publication_id: int, *, error_code: str, error_summary: str
+    ) -> Publication: ...
+
+    async def get_by_build(
+        self, script_build_id: int, *, publication_type: PublicationType
+    ) -> Publication | None: ...
+
+
+class LegacyBuildStateRepository(Protocol):
+    async def create(
+        self,
+        *,
+        execution_id: int,
+        topic_build_id: int,
+        topic_id: int,
+        agent_type: str | None,
+        agent_config: dict[str, Any] | None,
+        data_source_url: str | None,
+        strategies_config: dict[str, Any],
+    ) -> int: ...
+
+    async def set_status(
+        self, script_build_id: int, status: BuildStatus, *, error_summary: str | None = None
+    ) -> None: ...
+
+    async def get_status(self, script_build_id: int) -> BuildStatus: ...
+
+    async def project_direction(self, script_build_id: int, legacy_markdown: str) -> None: ...
+
+
+class PrincipalProvider(Protocol):
+    async def current(self, request_context: object | None = None) -> Principal: ...
+
+
+class BuildAuthorizer(Protocol):
+    async def require_source_access(
+        self,
+        principal: Principal,
+        *,
+        execution_id: int,
+        topic_build_id: int,
+        topic_id: int,
+    ) -> None: ...
+
+    async def require_access(self, principal: Principal, script_build_id: int) -> None: ...

script_build_host/src/script_build_host/domain/records.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from enum import StrEnum
+
+
+class BuildStatus(StrEnum):
+    RUNNING = "running"
+    STOPPING = "stopping"
+    SUCCESS = "success"
+    FAILED = "failed"
+    PARTIAL = "partial"
+    STOPPED = "stopped"
+
+
+class PublicationType(StrEnum):
+    DIRECTION = "direction"
+    FINAL = "final"
+
+
+class PublicationState(StrEnum):
+    PENDING = "pending"
+    PUBLISHING = "publishing"
+    PUBLISHED = "published"
+    FAILED = "failed"
+
+
+@dataclass(frozen=True, slots=True)
+class MissionBinding:
+    binding_id: int
+    script_build_id: int
+    root_trace_id: str
+    input_snapshot_id: int
+    active_direction_artifact_version_id: int | None
+    accepted_root_artifact_version_id: int | None
+    engine_version: str
+    schema_version: str
+    created_at: datetime
+    updated_at: datetime
+
+
+@dataclass(frozen=True, slots=True)
+class Publication:
+    publication_id: int
+    script_build_id: int
+    publication_type: PublicationType
+    accept_decision_id: str
+    artifact_version_id: int
+    expected_sha256: str
+    state: PublicationState
+    attempt_count: int
+    publication_revision: int
+    last_error_code: str | None
+    last_error_summary: str | None
+    created_at: datetime
+    updated_at: datetime
+    published_at: datetime | None
+
+
+@dataclass(frozen=True, slots=True)
+class Principal:
+    subject: str
+    tenant_id: str | None = None

script_build_host/src/script_build_host/infrastructure/canonical_json.py

@@ -0,0 +1,74 @@
+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 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())

script_build_host/src/script_build_host/infrastructure/config.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Self
+
+from pydantic import Field, SecretStr, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from sqlalchemy.engine import make_url
+
+
+class ScriptBuildSettings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_prefix="SCRIPT_BUILD_", env_file=".env", extra="ignore", case_sensitive=False
+    )
+
+    environment: str = "production"
+    read_database_url: SecretStr = Field(repr=False)
+    write_database_url: SecretStr = Field(repr=False)
+    agent_data_root: Path
+    persona_root: Path
+    section_pattern_root: Path
+    decode_index_root: Path
+    knowledge_root: Path
+    prompt_fallback_root: Path
+    prompts_from_file: bool = False
+    model_manifest: dict[str, object] = Field(default_factory=dict)
+    datasource_manifest: dict[str, object] = Field(default_factory=dict)
+    pattern_endpoint: str | None = None
+    decode_endpoint: str | None = None
+    embedding_endpoint: str | None = None
+    external_endpoint: str | None = None
+    image_endpoint: str | None = None
+    outbound_allowed_hosts: tuple[str, ...] = ()
+    outbound_allowed_ports: tuple[int, ...] = (443,)
+    websocket_allowed_origins: tuple[str, ...] = ()
+    runtime_factory: str | None = None
+    bind_host: str = "127.0.0.1"
+    bind_port: int = Field(default=8080, ge=1, le=65535)
+    request_timeout_seconds: float = Field(default=30.0, gt=0, le=600)
+    poll_interval_seconds: float = Field(default=1.0, gt=0, le=30)
+    max_response_bytes: int = Field(default=5_000_000, gt=0)
+    max_images: int = Field(default=12, ge=0, le=100)
+    max_image_bytes: int = Field(default=10_000_000, gt=0)
+    max_total_image_bytes: int = Field(default=50_000_000, gt=0)
+    max_upload_bytes: int = Field(default=5_000_000, gt=0)
+    max_upload_points: int = Field(default=500, gt=0, le=10_000)
+    max_upload_items: int = Field(default=5_000, gt=0, le=100_000)
+
+    @model_validator(mode="after")
+    def validate_database_separation(self) -> Self:
+        read_url = make_url(self.read_database_url.get_secret_value())
+        write_url = make_url(self.write_database_url.get_secret_value())
+        if self.environment.lower() != "test":
+            if read_url.get_backend_name() == "sqlite" or write_url.get_backend_name() == "sqlite":
+                raise ValueError("SQLite database URLs are permitted only in the test environment")
+            if read_url.username == write_url.username:
+                raise ValueError("production read and write database users must be different")
+        if self.max_total_image_bytes < self.max_image_bytes:
+            raise ValueError("max_total_image_bytes must not be smaller than max_image_bytes")
+        return self
+
+    def read_dsn(self) -> str:
+        return self.read_database_url.get_secret_value()
+
+    def write_dsn(self) -> str:
+        return self.write_database_url.get_secret_value()

script_build_host/src/script_build_host/infrastructure/manifests.py

@@ -0,0 +1,90 @@
+"""Resolve the concrete retrieval sources used by a newly started mission."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from dataclasses import dataclass
+from hashlib import sha256
+from pathlib import Path
+from urllib.parse import urlsplit, urlunsplit
+
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.infrastructure.config import ScriptBuildSettings
+from script_build_host.infrastructure.outbound import OutboundPolicy
+
+
+@dataclass(frozen=True, slots=True)
+class SettingsRuntimeManifestProvider:
+    settings: ScriptBuildSettings
+    outbound_policy: OutboundPolicy
+    decode_path: Path | None = None
+    knowledge_path: Path | None = None
+
+    async def load(self) -> dict[str, object]:
+        values: dict[str, object] = {}
+        for name, endpoint in (
+            ("pattern", self.settings.pattern_endpoint),
+            ("decode", self.settings.decode_endpoint),
+            ("embedding", self.settings.embedding_endpoint),
+            ("external", self.settings.external_endpoint),
+            ("image", self.settings.image_endpoint),
+        ):
+            if endpoint:
+                validated = await self.outbound_policy.validate_url(endpoint)
+                values[name] = {"endpoint": _credential_free_endpoint(validated)}
+        values["decode_index"] = await asyncio.to_thread(
+            _path_manifest,
+            self.decode_path or self.settings.decode_index_root,
+        )
+        values["knowledge"] = await asyncio.to_thread(
+            _path_manifest,
+            self.knowledge_path or self.settings.knowledge_root,
+        )
+        return values
+
+
+def _credential_free_endpoint(value: str) -> str:
+    parsed = urlsplit(value)
+    host = parsed.hostname or ""
+    port = f":{parsed.port}" if parsed.port and parsed.port != 443 else ""
+    return urlunsplit(("https", f"{host}{port}", parsed.path, "", ""))
+
+
+def _path_manifest(path: Path) -> dict[str, object]:
+    resolved = path.resolve()
+    if resolved.is_file() and resolved.suffix.lower() == ".json":
+        try:
+            payload = json.loads(resolved.read_text(encoding="utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            payload = None
+        if payload is not None:
+            return {
+                "source_id": resolved.name,
+                "file_count": 1,
+                "sha256": canonical_sha256(payload).wire,
+            }
+    files = (
+        [resolved]
+        if resolved.is_file()
+        else sorted(item for item in resolved.rglob("*") if item.is_file())
+        if resolved.is_dir()
+        else []
+    )
+    digest = sha256()
+    for item in files:
+        relative = item.name if resolved.is_file() else item.relative_to(resolved).as_posix()
+        digest.update(relative.encode("utf-8"))
+        digest.update(b"\0")
+        with item.open("rb") as handle:
+            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+                digest.update(chunk)
+        digest.update(b"\0")
+    return {
+        "source_id": resolved.name,
+        "file_count": len(files),
+        "sha256": "sha256:" + digest.hexdigest(),
+    }
+
+
+__all__ = ["SettingsRuntimeManifestProvider"]

script_build_host/src/script_build_host/infrastructure/outbound.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import socket
+from collections.abc import Awaitable, Callable, Iterable
+from dataclasses import dataclass
+from urllib.parse import urljoin, urlsplit
+
+from script_build_host.domain.errors import UnsafeOutboundTarget
+
+AddressResolver = Callable[[str, int], Awaitable[Iterable[str]]]
+
+
+async def _system_resolver(host: str, port: int) -> Iterable[str]:
+    loop = asyncio.get_running_loop()
+    records = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    return {str(record[4][0]) for record in records}
+
+
+def _unsafe_address(value: str) -> bool:
+    try:
+        address = ipaddress.ip_address(value.split("%", 1)[0])
+    except ValueError as error:
+        raise UnsafeOutboundTarget("DNS returned an invalid address") from error
+    return not address.is_global
+
+
+@dataclass(frozen=True, slots=True)
+class OutboundPolicy:
+    allowed_hosts: frozenset[str]
+    allowed_ports: frozenset[int] = frozenset({443})
+    resolver: AddressResolver = _system_resolver
+
+    async def validate_url(self, url: str) -> str:
+        try:
+            parsed = urlsplit(url)
+            port = parsed.port or 443
+        except ValueError as error:
+            raise UnsafeOutboundTarget("the outbound URL is malformed") from error
+        if parsed.scheme.lower() != "https":
+            raise UnsafeOutboundTarget("only HTTPS outbound requests are permitted")
+        if parsed.username is not None or parsed.password is not None:
+            raise UnsafeOutboundTarget("outbound URLs must not contain credentials")
+        if parsed.fragment:
+            raise UnsafeOutboundTarget("outbound URLs must not contain fragments")
+        host = (parsed.hostname or "").rstrip(".").lower()
+        if not host or host not in {item.rstrip(".").lower() for item in self.allowed_hosts}:
+            raise UnsafeOutboundTarget("the outbound host is not allowlisted")
+        if port not in self.allowed_ports:
+            raise UnsafeOutboundTarget("the outbound port is not allowlisted")
+        addresses = tuple(await self.resolver(host, port))
+        if not addresses or any(_unsafe_address(address) for address in addresses):
+            raise UnsafeOutboundTarget("the outbound host resolves to a non-public address")
+        return url
+
+    async def validate_redirect(self, current_url: str, location: str) -> str:
+        redirected = urljoin(current_url, location)
+        return await self.validate_url(redirected)

script_build_host/src/script_build_host/infrastructure/redaction.py

@@ -0,0 +1,54 @@
+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

script_build_host/src/script_build_host/py.typed

@@ -0,0 +1 @@
+

script_build_host/tests/test_canonical_security.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from decimal import Decimal
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from script_build_host.domain.digests import Sha256Digest
+from script_build_host.domain.errors import ProtocolViolation, UnsafeOutboundTarget
+from script_build_host.infrastructure.canonical_json import canonical_json_text, canonical_sha256
+from script_build_host.infrastructure.config import ScriptBuildSettings
+from script_build_host.infrastructure.outbound import OutboundPolicy
+from script_build_host.infrastructure.redaction import redact, redact_text
+
+
+def test_canonical_json_is_stable_and_preserves_array_order() -> None:
+    first = {
+        "z": 1.2300,
+        "when": datetime(2026, 7, 19, 4, 5, 6, 7, tzinfo=UTC),
+        "items": ["b", "a"],
+        "amount": Decimal("2.500"),
+        "empty": [],
+        "missing": None,
+    }
+    second = dict(reversed(tuple(first.items())))
+    text = canonical_json_text(first)
+    assert text == canonical_json_text(second)
+    assert '"items":["b","a"]' in text
+    assert '"z":"1.23"' in text
+    assert '"amount":"2.5"' in text
+    assert canonical_sha256(first) == canonical_sha256(second)
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["SHA256:" + "0" * 64, "sha256:" + "A" * 64, "md5:" + "0" * 64, "0" * 64],
+)
+def test_digest_rejects_noncanonical_wire_forms(value: str) -> None:
+    with pytest.raises(ProtocolViolation):
+        Sha256Digest.from_wire(value)
+
+
+def test_redaction_removes_secrets_from_nested_values() -> None:
+    value = {
+        "authorization": "Bearer raw-token",
+        "database_url": "mysql://reader:secret@db.example/app",
+        "nested": {"url": "https://api.example/search?token=abc&q=safe"},
+    }
+    result = redact(value)
+    assert "raw-token" not in str(result)
+    assert "secret" not in str(result)
+    assert "abc" not in str(result)
+    assert redact_text("Bearer abc") == "Bearer [REDACTED]"
+
+
+def _settings(tmp_path: Path, **overrides: object) -> ScriptBuildSettings:
+    values: dict[str, object] = {
+        "environment": "test",
+        "read_database_url": "sqlite+aiosqlite:///read.db",
+        "write_database_url": "sqlite+aiosqlite:///write.db",
+        "agent_data_root": tmp_path,
+        "persona_root": tmp_path,
+        "section_pattern_root": tmp_path,
+        "decode_index_root": tmp_path,
+        "knowledge_root": tmp_path,
+        "prompt_fallback_root": tmp_path,
+    }
+    values.update(overrides)
+    return ScriptBuildSettings(**values)  # type: ignore[arg-type]
+
+
+def test_settings_hide_dsns_and_enforce_production_user_separation(tmp_path: Path) -> None:
+    settings = _settings(tmp_path)
+    assert "sqlite" not in repr(settings)
+    with pytest.raises(ValidationError, match="must be different"):
+        _settings(
+            tmp_path,
+            environment="production",
+            read_database_url="mysql+asyncmy://same:read@db.example/app",
+            write_database_url="mysql+asyncmy://same:write@db.example/app",
+        )
+
+
+@pytest.mark.asyncio
+async def test_outbound_policy_revalidates_dns_and_redirects() -> None:
+    calls: list[tuple[str, int]] = []
+
+    async def resolver(host: str, port: int) -> tuple[str, ...]:
+        calls.append((host, port))
+        return ("93.184.216.34",)
+
+    policy = OutboundPolicy(frozenset({"api.example"}), resolver=resolver)
+    assert await policy.validate_url("https://api.example/path") == "https://api.example/path"
+    assert (
+        await policy.validate_redirect("https://api.example/path", "/next")
+        == "https://api.example/next"
+    )
+    assert calls == [("api.example", 443), ("api.example", 443)]
+
+
+@pytest.mark.asyncio
+async def test_outbound_policy_rejects_private_dns_and_non_allowlisted_redirect() -> None:
+    async def private_resolver(_host: str, _port: int) -> tuple[str, ...]:
+        return ("169.254.169.254",)
+
+    policy = OutboundPolicy(frozenset({"api.example"}), resolver=private_resolver)
+    with pytest.raises(UnsafeOutboundTarget):
+        await policy.validate_url("https://api.example/latest/meta-data")
+
+    async def public_resolver(_host: str, _port: int) -> tuple[str, ...]:
+        return ("93.184.216.34",)
+
+    public_policy = OutboundPolicy(frozenset({"api.example"}), resolver=public_resolver)
+    with pytest.raises(UnsafeOutboundTarget):
+        await public_policy.validate_redirect("https://api.example/path", "https://evil.example/")