输入：按旧表关系组装并冻结业务快照

只读校验 execution、topic build、topic、point、composition、relation 和 source 的真实归属链。

解析 Persona、Section Pattern、策略和 Prompt 实际来源，保留文件版本与摘要，并对配置和输入中的凭据做冻结前脱敏。

script_build_host/src/script_build_host/adapters/persona.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.ports import PersonaInput
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+
+_ALIASES = {"每天心理学": "每天一点心理学"}
+_SECTION_FIELDS = ("分段规律摘要", "典型分段依据", "段落结构模式列表", "分析帖子数")
+
+
+def _safe_account_file(root: Path, account_name: str, filename: str) -> Path:
+    candidate = (root / account_name / filename).resolve()
+    resolved_root = root.resolve()
+    if not candidate.is_relative_to(resolved_root):
+        raise ProtocolViolation("account data path escapes its configured root")
+    return candidate
+
+
+def _read_json(path: Path) -> tuple[Any, str] | None:
+    if not path.is_file():
+        return None
+    raw = path.read_bytes()
+    try:
+        payload = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise ProtocolViolation("account data file is not valid UTF-8 JSON") from error
+    return payload, canonical_sha256(payload).wire
+
+
+def _trim_value_cluster_node(node: dict[str, Any], *, max_values: int = 10) -> dict[str, Any]:
+    values = node.get("值")
+    children = node.get("子聚类")
+    return {
+        "路径": node.get("路径"),
+        "值": list(values[:max_values]) if isinstance(values, list) else [],
+        "子聚类": [
+            _trim_value_cluster_node(item, max_values=max_values)
+            for item in children or []
+            if isinstance(item, dict)
+        ],
+    }
+
+
+def _trim_value_cluster(value: Any) -> dict[str, Any] | None:
+    if not isinstance(value, dict):
+        return None
+    clusters = value.get("聚类")
+    if isinstance(clusters, list):
+        return {
+            "聚类": [_trim_value_cluster_node(item) for item in clusters if isinstance(item, dict)]
+        }
+    return _trim_value_cluster_node(value) if "路径" in value else None
+
+
+def _script_persona_points(payload: Any) -> tuple[dict[str, Any], ...]:
+    rows = payload.get("records") if isinstance(payload, dict) else None
+    if not isinstance(rows, list):
+        return ()
+    selected: list[tuple[float, dict[str, Any]]] = []
+    for row in rows:
+        if not isinstance(row, dict) or row.get("阶段") != "创作":
+            continue
+        point_type = row.get("点类型")
+        raw_coverage = row.get("帖子覆盖率")
+        if not isinstance(raw_coverage, (int, float, str)):
+            continue
+        try:
+            coverage = float(raw_coverage)
+        except (TypeError, ValueError):
+            continue
+        if point_type == "维度名" and coverage < 0.3:
+            continue
+        if point_type == "维度值" and coverage < 0.2:
+            continue
+        if point_type not in {"维度名", "维度值"}:
+            continue
+        item = {
+            "点名称": row.get("名称"),
+            "点类型": point_type,
+            "分类路径": row.get("分类路径"),
+            "列": row.get("列"),
+            "帖子覆盖率": row.get("帖子覆盖率"),
+            "人设权重分": row.get("人设权重分"),
+        }
+        if point_type == "维度名":
+            item["值聚类"] = _trim_value_cluster(row.get("值聚类"))
+        selected.append((coverage, item))
+    selected.sort(key=lambda item: (item[1]["点类型"] == "维度名", -item[0]))
+    return tuple(item for _, item in selected)
+
+
+class FilePersonaSource:
+    def __init__(self, *, persona_root: Path, section_pattern_root: Path) -> None:
+        self._persona_root = persona_root
+        self._section_pattern_root = section_pattern_root
+
+    async def load(self, account_name: str) -> PersonaInput:
+        requested = account_name.strip()
+        candidates = tuple(dict.fromkeys((requested, _ALIASES.get(requested, requested))))
+        resolved = requested
+        points: tuple[dict[str, Any], ...] = ()
+        sections: tuple[dict[str, Any], ...] = ()
+        versions: dict[str, str] = {}
+        for candidate in candidates:
+            loaded = _read_json(
+                _safe_account_file(self._persona_root, candidate, "point_records.json")
+            )
+            if loaded is None:
+                continue
+            payload, digest = loaded
+            resolved = candidate
+            points = _script_persona_points(payload)
+            versions["persona_points"] = digest
+            break
+        for candidate in candidates:
+            loaded = _read_json(
+                _safe_account_file(
+                    self._section_pattern_root, candidate, "section_main_dimension_template.json"
+                )
+            )
+            if loaded is None:
+                continue
+            payload, digest = loaded
+            if isinstance(payload, dict):
+                sections = (dict({key: payload[key] for key in _SECTION_FIELDS if key in payload}),)
+                versions["section_patterns"] = digest
+                if not points:
+                    resolved = candidate
+            break
+        return PersonaInput(
+            requested_account_name=requested,
+            resolved_account_name=resolved,
+            persona_points=points,
+            section_patterns=sections,
+            source_versions=versions,
+        )

script_build_host/src/script_build_host/adapters/prompts.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.ports import PromptRequest
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.infrastructure.legacy_tables import prompt
+
+
+class DatabaseFirstPromptSource:
+    def __init__(
+        self,
+        sessions: async_sessionmaker[AsyncSession],
+        *,
+        fallback_root: Path,
+        file_only: bool = False,
+    ) -> None:
+        self._sessions = sessions
+        self._fallback_root = fallback_root
+        self._file_only = file_only
+
+    async def load(self, requests: tuple[PromptRequest, ...]) -> tuple[dict[str, Any], ...]:
+        output = []
+        for request in requests:
+            resolved = None if self._file_only else await self._load_database(request)
+            output.append(resolved or self._load_file(request))
+        return tuple(output)
+
+    async def _load_database(self, request: PromptRequest) -> dict[str, Any] | None:
+        async with self._sessions() as session:
+            row = (
+                (
+                    await session.execute(
+                        select(prompt)
+                        .where(prompt.c.biz_type == request.biz_type)
+                        .order_by(prompt.c.id)
+                        .limit(1)
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+        if row is None or not row["prompt_content"]:
+            return None
+        return self._manifest(
+            request,
+            content=str(row["prompt_content"]),
+            source="database",
+            version=int(row["current_version"]),
+            source_id=str(row["id"]),
+        )
+
+    def _load_file(self, request: PromptRequest) -> dict[str, Any]:
+        if Path(request.filename).name != request.filename:
+            raise ProtocolViolation("prompt fallback filename must not contain a path")
+        path = (self._fallback_root / request.filename).resolve()
+        if not path.is_relative_to(self._fallback_root.resolve()) or not path.is_file():
+            raise ProtocolViolation(f"prompt is unavailable for biz_type {request.biz_type}")
+        try:
+            content = path.read_text(encoding="utf-8")
+        except UnicodeDecodeError as error:
+            raise ProtocolViolation("prompt fallback must be UTF-8 text") from error
+        return self._manifest(
+            request,
+            content=content,
+            source="file",
+            version=None,
+            source_id=request.filename,
+        )
+
+    @staticmethod
+    def _manifest(
+        request: PromptRequest,
+        *,
+        content: str,
+        source: str,
+        version: int | None,
+        source_id: str,
+    ) -> dict[str, Any]:
+        return {
+            "biz_type": request.biz_type,
+            "preset": request.preset,
+            "role": request.role,
+            "source": source,
+            "source_id": source_id,
+            "version": version,
+            "content": content,
+            "content_sha256": canonical_sha256(content).wire,
+        }

script_build_host/src/script_build_host/adapters/strategy.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import or_, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.ports import StrategySelector
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.infrastructure.legacy_tables import build_strategy, build_strategy_version
+
+
+class SqlAlchemyStrategySource:
+    def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
+        self._sessions = sessions
+
+    async def load(
+        self, *, always_on: tuple[StrategySelector, ...], on_demand: tuple[StrategySelector, ...]
+    ) -> tuple[dict[str, Any], ...]:
+        requested = _deduplicate((*always_on, *on_demand))
+        if not requested:
+            return ()
+        requested_ids = [
+            value for value in (_selector_id(item) for item in requested) if value is not None
+        ]
+        requested_names = [
+            value for value in (_selector_name(item) for item in requested) if value is not None
+        ]
+        async with self._sessions() as session:
+            rows = (
+                (
+                    await session.execute(
+                        select(
+                            build_strategy.c.id,
+                            build_strategy.c.name,
+                            build_strategy.c.description,
+                            build_strategy.c.tags,
+                            build_strategy.c.current_version,
+                            build_strategy_version.c.content,
+                            build_strategy_version.c.created_at,
+                        )
+                        .join(
+                            build_strategy_version,
+                            (build_strategy_version.c.strategy_id == build_strategy.c.id)
+                            & (
+                                build_strategy_version.c.version == build_strategy.c.current_version
+                            ),
+                        )
+                        .where(
+                            build_strategy.c.strategy_type == "script",
+                            build_strategy.c.is_active.is_(True),
+                            or_(
+                                build_strategy.c.id.in_(requested_ids),
+                                build_strategy.c.name.in_(requested_names),
+                            ),
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+        by_id = {int(row["id"]): row for row in rows}
+        by_name = {str(row["name"]): row for row in rows}
+        resolved = []
+        for item in requested:
+            identifier = _selector_id(item)
+            row = by_id.get(identifier) if identifier is not None else None
+            resolved.append(row or by_name.get(_selector_name(item) or ""))
+        missing = [str(item) for item, row in zip(requested, resolved, strict=True) if row is None]
+        if missing:
+            raise ProtocolViolation(f"script strategy is missing or inactive: {', '.join(missing)}")
+        output = []
+        always_keys = {_selector_key(item) for item in always_on}
+        for selector, optional_row in zip(requested, resolved, strict=True):
+            if optional_row is None:
+                raise AssertionError("missing strategies were rejected above")
+            row = optional_row
+            name = str(row["name"])
+            content = str(row["content"])
+            mode = "always_on" if _selector_key(selector) in always_keys else "on_demand"
+            output.append(
+                {
+                    "strategy_id": int(row["id"]),
+                    "name": name,
+                    "mode": mode,
+                    "description": row["description"],
+                    "tags": row["tags"] or [],
+                    "version": int(row["current_version"]),
+                    # Both modes are frozen. On-demand controls when the Agent may
+                    # consume the strategy; it must not make mission input drift.
+                    "content": content,
+                    "content_sha256": canonical_sha256(content).wire,
+                    "created_at": row["created_at"],
+                }
+            )
+        return tuple(output)
+
+
+def _selector_id(value: StrategySelector) -> int | None:
+    if isinstance(value, int):
+        return value
+    if isinstance(value, dict):
+        candidate = value.get("id", value.get("strategy_id"))
+        return (
+            int(candidate)
+            if isinstance(candidate, (int, str)) and str(candidate).isdigit()
+            else None
+        )
+    return int(value) if value.isdigit() else None
+
+
+def _selector_name(value: StrategySelector) -> str | None:
+    if isinstance(value, str) and not value.isdigit():
+        return value
+    if isinstance(value, dict):
+        candidate = value.get("name")
+        return str(candidate) if candidate else None
+    return None
+
+
+def _selector_key(value: StrategySelector) -> str:
+    identifier = _selector_id(value)
+    return f"id:{identifier}" if identifier is not None else f"name:{_selector_name(value)}"
+
+
+def _deduplicate(values: tuple[StrategySelector, ...]) -> tuple[StrategySelector, ...]:
+    output: list[StrategySelector] = []
+    seen: set[str] = set()
+    for value in values:
+        key = _selector_key(value)
+        if key not in seen:
+            output.append(value)
+            seen.add(key)
+    return tuple(output)

script_build_host/src/script_build_host/application/__init__.py

@@ -0,0 +1,3 @@
+from .input_snapshot_service import AssembleInputRequest, ScriptInputSnapshotService
+
+__all__ = ["AssembleInputRequest", "ScriptInputSnapshotService"]

script_build_host/src/script_build_host/application/input_snapshot_service.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from script_build_host.domain.errors import InputRelationMismatch
+from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
+from script_build_host.domain.ports import (
+    InputSnapshotRepository,
+    LegacyInputReader,
+    PersonaSource,
+    PromptRequest,
+    PromptSource,
+    StrategySelector,
+    StrategySource,
+)
+from script_build_host.domain.records import Principal
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.infrastructure.redaction import redact, redact_text
+
+
+@dataclass(frozen=True, slots=True)
+class AssembleInputRequest:
+    script_build_id: int
+    execution_id: int
+    topic_build_id: int
+    topic_id: int
+    principal: Principal
+    strategies_always_on: tuple[StrategySelector, ...] = ()
+    strategies_on_demand: tuple[StrategySelector, ...] = ()
+    prompt_requests: tuple[PromptRequest, ...] = ()
+    runtime_prompt_manifest: tuple[dict[str, Any], ...] = ()
+    datasource_manifest: dict[str, Any] = field(default_factory=dict)
+    model_manifest: dict[str, Any] = field(default_factory=dict)
+
+
+class ScriptInputSnapshotService:
+    def __init__(
+        self,
+        *,
+        legacy_input: LegacyInputReader,
+        persona_source: PersonaSource,
+        strategy_source: StrategySource,
+        prompt_source: PromptSource,
+        snapshots: InputSnapshotRepository,
+    ) -> None:
+        self._legacy_input = legacy_input
+        self._persona_source = persona_source
+        self._strategy_source = strategy_source
+        self._prompt_source = prompt_source
+        self._snapshots = snapshots
+
+    async def validate_source(
+        self, *, execution_id: int, topic_build_id: int, topic_id: int
+    ) -> None:
+        await self._legacy_input.read_topic_graph(
+            execution_id=execution_id,
+            topic_build_id=topic_build_id,
+            topic_id=topic_id,
+        )
+
+    async def assemble(self, request: AssembleInputRequest) -> ScriptBuildInput:
+        topic = await self._legacy_input.read_topic_graph(
+            execution_id=request.execution_id,
+            topic_build_id=request.topic_build_id,
+            topic_id=request.topic_id,
+        )
+        build_record = topic.get("build_record")
+        if not isinstance(build_record, dict):
+            raise InputRelationMismatch()
+        personal_config = build_record.get("personal_config") or {}
+        account_name = (
+            personal_config.get("account_name") if isinstance(personal_config, dict) else None
+        )
+        persona = await self._persona_source.load(str(account_name or ""))
+        strategies_config = build_record.get("strategies_config") or {}
+        always_on = request.strategies_always_on or tuple(strategies_config.get("always_on", []))
+        on_demand = request.strategies_on_demand or tuple(strategies_config.get("on_demand", []))
+        strategies = await self._strategy_source.load(always_on=always_on, on_demand=on_demand)
+        loaded_prompts = await self._prompt_source.load(request.prompt_requests)
+        prompts = (*request.runtime_prompt_manifest, *loaded_prompts)
+        sanitized_topic = redact(topic)
+        sanitized_persona_points = _redacted_rows(persona.persona_points, "persona points")
+        sanitized_section_patterns = _redacted_rows(
+            persona.section_patterns,
+            "section patterns",
+        )
+        sanitized_strategies = _redacted_rows(strategies, "strategies")
+        sanitized_prompts = _redacted_rows(prompts, "prompt manifest")
+        persona_digest = canonical_sha256(
+            {
+                "persona_points": sanitized_persona_points,
+                "section_patterns": sanitized_section_patterns,
+                "source_versions": persona.source_versions,
+            }
+        ).wire
+        return ScriptBuildInput(
+            script_build_id=request.script_build_id,
+            execution_id=request.execution_id,
+            topic_build_id=request.topic_build_id,
+            topic_id=request.topic_id,
+            topic=sanitized_topic,
+            account={
+                "account_name": redact_text(persona.requested_account_name),
+                "resolved_account_name": redact_text(persona.resolved_account_name),
+                "source": "personal_config",
+                "source_versions": persona.source_versions,
+                "content_sha256": persona_digest,
+            },
+            persona_points=sanitized_persona_points,
+            section_patterns=sanitized_section_patterns,
+            strategies=sanitized_strategies,
+            prompt_manifest=sanitized_prompts,
+            datasource_manifest=redact(request.datasource_manifest),
+            model_manifest=redact(request.model_manifest),
+        )
+
+    async def freeze(self, assembled: ScriptBuildInput) -> ScriptBuildInputSnapshotV1:
+        return await self._snapshots.freeze(assembled)
+
+    async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
+        return await self._snapshots.get(snapshot_id, script_build_id=script_build_id)
+
+
+def _redacted_rows(
+    values: tuple[dict[str, Any], ...],
+    label: str,
+) -> tuple[dict[str, Any], ...]:
+    sanitized = redact(list(values))
+    if not isinstance(sanitized, list) or any(not isinstance(item, dict) for item in sanitized):
+        raise InputRelationMismatch()
+    return tuple(sanitized)

script_build_host/src/script_build_host/repositories/__init__.py

@@ -0,0 +1,17 @@
+from .legacy_input import LegacySqlAlchemyInputReader
+from .legacy_state import SqlAlchemyLegacyBuildStateRepository
+from .sqlalchemy import (
+    SqlAlchemyInputSnapshotRepository,
+    SqlAlchemyMissionBindingRepository,
+    SqlAlchemyPublicationRepository,
+    SqlAlchemyScriptBusinessArtifactRepository,
+)
+
+__all__ = [
+    "LegacySqlAlchemyInputReader",
+    "SqlAlchemyInputSnapshotRepository",
+    "SqlAlchemyLegacyBuildStateRepository",
+    "SqlAlchemyMissionBindingRepository",
+    "SqlAlchemyPublicationRepository",
+    "SqlAlchemyScriptBusinessArtifactRepository",
+]

script_build_host/src/script_build_host/repositories/legacy_input.py

@@ -0,0 +1,319 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from sqlalchemy import or_, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import InputRelationMismatch
+from script_build_host.infrastructure.legacy_tables import (
+    topic_build_composition_item,
+    topic_build_item_relation,
+    topic_build_item_source,
+    topic_build_point,
+    topic_build_point_item_relation,
+    topic_build_record,
+    topic_build_topic,
+    topic_pattern_execution,
+)
+
+
+def _mapping(value: Any) -> dict[str, Any]:
+    if value is None:
+        return {}
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except json.JSONDecodeError as exc:
+            raise InputRelationMismatch() from exc
+        if isinstance(parsed, dict):
+            return parsed
+    raise InputRelationMismatch()
+
+
+def _json_value(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return value
+
+
+def _row_dict(row: Any, fields: tuple[str, ...]) -> dict[str, Any]:
+    return {field: row[field] for field in fields}
+
+
+class LegacySqlAlchemyInputReader:
+    """Anti-corruption adapter over the old tables; it never imports the old runtime package."""
+
+    def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
+        self._sessions = sessions
+
+    async def read_topic_graph(
+        self, *, execution_id: int, topic_build_id: int, topic_id: int
+    ) -> dict[str, Any]:
+        async with self._sessions() as session:
+            execution = None
+            if execution_id != 0:
+                execution = (
+                    (
+                        await session.execute(
+                            select(topic_pattern_execution).where(
+                                topic_pattern_execution.c.id == execution_id,
+                                topic_pattern_execution.c.status == "success",
+                            )
+                        )
+                    )
+                    .mappings()
+                    .one_or_none()
+                )
+            build = (
+                (
+                    await session.execute(
+                        select(topic_build_record).where(
+                            topic_build_record.c.id == topic_build_id,
+                            topic_build_record.c.execution_id == execution_id,
+                            topic_build_record.c.is_deleted.is_(False),
+                        )
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            topic = (
+                (
+                    await session.execute(
+                        select(topic_build_topic).where(
+                            topic_build_topic.c.id == topic_id,
+                            topic_build_topic.c.build_id == topic_build_id,
+                            topic_build_topic.c.execution_id == execution_id,
+                        )
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            valid_upload = execution_id == 0 and build is not None and build["origin"] == "upload"
+            if (
+                (execution_id != 0 and execution is None)
+                or (execution_id == 0 and not valid_upload)
+                or build is None
+                or build["status"] != "success"
+                or topic is None
+                or topic["status"] not in {"mature", "success"}
+            ):
+                raise InputRelationMismatch()
+
+            point_rows = (
+                (
+                    await session.execute(
+                        select(topic_build_point)
+                        .where(
+                            topic_build_point.c.topic_id == topic_id,
+                            topic_build_point.c.build_id == topic_build_id,
+                            topic_build_point.c.is_active.is_(True),
+                        )
+                        .order_by(topic_build_point.c.id)
+                    )
+                )
+                .mappings()
+                .all()
+            )
+            item_rows = (
+                (
+                    await session.execute(
+                        select(topic_build_composition_item)
+                        .where(
+                            topic_build_composition_item.c.topic_id == topic_id,
+                            topic_build_composition_item.c.build_id == topic_build_id,
+                            topic_build_composition_item.c.is_active.is_(True),
+                        )
+                        .order_by(
+                            topic_build_composition_item.c.sort_order,
+                            topic_build_composition_item.c.id,
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+            point_ids = {int(row["id"]) for row in point_rows}
+            item_ids = {int(row["id"]) for row in item_rows}
+            point_relation_rows = (
+                (
+                    await session.execute(
+                        select(topic_build_point_item_relation).where(
+                            or_(
+                                topic_build_point_item_relation.c.point_id.in_(point_ids),
+                                topic_build_point_item_relation.c.item_id.in_(item_ids),
+                            ),
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+                if point_ids or item_ids
+                else []
+            )
+            item_relation_rows = (
+                (
+                    await session.execute(
+                        select(topic_build_item_relation)
+                        .where(
+                            topic_build_item_relation.c.topic_id == topic_id,
+                        )
+                        .order_by(topic_build_item_relation.c.id)
+                    )
+                )
+                .mappings()
+                .all()
+                if item_ids
+                else []
+            )
+            source_rows = (
+                (
+                    await session.execute(
+                        select(topic_build_item_source)
+                        .where(
+                            topic_build_item_source.c.topic_id == topic_id,
+                            topic_build_item_source.c.is_active.is_(True),
+                        )
+                        .order_by(topic_build_item_source.c.id)
+                    )
+                )
+                .mappings()
+                .all()
+                if item_ids
+                else []
+            )
+
+        if any(
+            int(row["point_id"]) not in point_ids or int(row["item_id"]) not in item_ids
+            for row in point_relation_rows
+        ):
+            raise InputRelationMismatch()
+        if any(
+            int(row["source_item_id"]) not in item_ids or int(row["target_item_id"]) not in item_ids
+            for row in item_relation_rows
+        ):
+            raise InputRelationMismatch()
+        if any(int(row["target_item_id"]) not in item_ids for row in source_rows):
+            raise InputRelationMismatch()
+
+        point_to_items: dict[int, list[int]] = {point_id: [] for point_id in point_ids}
+        for relation in point_relation_rows:
+            point_to_items[int(relation["point_id"])].append(int(relation["item_id"]))
+
+        build_fields = (
+            "id",
+            "execution_id",
+            "demand",
+            "demand_constraints",
+            "agent_type",
+            "agent_config",
+            "status",
+            "strategies_config",
+            "personal_config",
+            "origin",
+            "raw_upload_data",
+            "start_time",
+            "end_time",
+        )
+        topic_fields = (
+            "id",
+            "build_id",
+            "execution_id",
+            "sort_order",
+            "topic_direction",
+            "topic_understanding",
+            "inspiration_evaluation_config",
+            "result",
+            "status",
+            "failure_reason",
+        )
+        point_fields = (
+            "id",
+            "topic_id",
+            "build_id",
+            "point_type",
+            "point_result",
+            "reason",
+            "is_active",
+            "note",
+            "created_at",
+            "updated_at",
+        )
+        item_fields = (
+            "id",
+            "topic_id",
+            "build_id",
+            "item_level",
+            "dimension",
+            "point_type",
+            "element_name",
+            "category_path",
+            "category_id",
+            "derivation_type",
+            "step",
+            "reason",
+            "sort_order",
+            "is_active",
+            "note",
+            "created_at",
+            "updated_at",
+        )
+        source_fields = (
+            "id",
+            "topic_id",
+            "target_item_id",
+            "derivation_type",
+            "source_type",
+            "source_reference_id",
+            "source_detail",
+            "dataset_from",
+            "reason",
+            "is_active",
+            "created_at",
+        )
+        build_dict = _row_dict(build, build_fields)
+        build_dict["demand_constraints"] = _mapping(build_dict["demand_constraints"])
+        build_dict["agent_config"] = _mapping(build_dict["agent_config"])
+        build_dict["strategies_config"] = _mapping(build_dict["strategies_config"])
+        build_dict["personal_config"] = _mapping(build_dict["personal_config"])
+        build_dict["raw_upload_data"] = _json_value(build_dict["raw_upload_data"])
+        points = []
+        for row in point_rows:
+            item = _row_dict(row, point_fields)
+            item["item_ids"] = point_to_items[int(row["id"])]
+            points.append(item)
+        sources = []
+        for row in source_rows:
+            source = _row_dict(row, source_fields)
+            source["source_detail"] = _json_value(source["source_detail"])
+            sources.append(source)
+        return {
+            "execution": (
+                _row_dict(
+                    execution,
+                    ("id", "snapshot_id", "status", "start_time", "end_time"),
+                )
+                if execution is not None
+                else {"id": 0, "status": "upload"}
+            ),
+            "build_record": build_dict,
+            "topic": _row_dict(topic, topic_fields),
+            "points": points,
+            "composition_items": [_row_dict(row, item_fields) for row in item_rows],
+            "relations": [
+                _row_dict(
+                    row,
+                    ("id", "topic_id", "source_item_id", "target_item_id", "reason", "created_at"),
+                )
+                for row in item_relation_rows
+            ],
+            "sources": sources,
+        }

script_build_host/tests/conftest.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from script_build_host.infrastructure.legacy_tables import legacy_metadata
+from script_build_host.infrastructure.tables import metadata
+
+
+@pytest_asyncio.fixture
+async def database(
+    tmp_path: Path,
+) -> AsyncIterator[tuple[AsyncEngine, async_sessionmaker[AsyncSession]]]:
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
+    async with engine.begin() as connection:
+        await connection.run_sync(legacy_metadata.create_all)
+        await connection.run_sync(metadata.create_all)
+    yield engine, async_sessionmaker(engine, expire_on_commit=False)
+    await engine.dispose()

script_build_host/tests/test_input_snapshot_service.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from script_build_host.application.input_snapshot_service import (
+    AssembleInputRequest,
+    ScriptInputSnapshotService,
+)
+from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
+from script_build_host.domain.ports import PersonaInput
+from script_build_host.domain.records import Principal
+
+
+class FakeLegacyInput:
+    def __init__(self) -> None:
+        self.calls: list[tuple[int, int, int]] = []
+
+    async def read_topic_graph(
+        self, *, execution_id: int, topic_build_id: int, topic_id: int
+    ) -> dict[str, Any]:
+        self.calls.append((execution_id, topic_build_id, topic_id))
+        return {
+            "build_record": {
+                "personal_config": {"account_name": "account"},
+                "strategies_config": {"always_on": [7], "on_demand": [8]},
+                "agent_config": {"api_key": "hidden", "model": "fake"},
+            },
+            "topic": {"id": topic_id},
+            "points": [],
+            "composition_items": [],
+            "relations": [],
+            "sources": [],
+        }
+
+
+class FakePersona:
+    async def load(self, account_name: str) -> PersonaInput:
+        return PersonaInput(account_name, account_name, ({"point": "one"},), (), {"p": "sha256:x"})
+
+
+class FakeStrategies:
+    async def load(
+        self, *, always_on: tuple[Any, ...], on_demand: tuple[Any, ...]
+    ) -> tuple[dict[str, Any], ...]:
+        return ({"always_on": list(always_on), "on_demand": list(on_demand)},)
+
+
+class FakePrompts:
+    async def load(self, requests: tuple[Any, ...]) -> tuple[dict[str, Any], ...]:
+        return ({"source": "fixture", "count": len(requests)},)
+
+
+class FakeSnapshots:
+    def __init__(self) -> None:
+        self.frozen: ScriptBuildInput | None = None
+
+    async def freeze(
+        self, assembled: ScriptBuildInput, *, version: int = 1
+    ) -> ScriptBuildInputSnapshotV1:
+        self.frozen = assembled
+        raise NotImplementedError
+
+    async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
+        raise NotImplementedError
+
+
+@pytest.mark.asyncio
+async def test_validate_source_is_read_only_and_assemble_redacts_secrets() -> None:
+    legacy = FakeLegacyInput()
+    service = ScriptInputSnapshotService(
+        legacy_input=legacy,
+        persona_source=FakePersona(),
+        strategy_source=FakeStrategies(),
+        prompt_source=FakePrompts(),
+        snapshots=FakeSnapshots(),
+    )
+    await service.validate_source(execution_id=10, topic_build_id=20, topic_id=30)
+    assembled = await service.assemble(
+        AssembleInputRequest(
+            script_build_id=1,
+            execution_id=10,
+            topic_build_id=20,
+            topic_id=30,
+            principal=Principal("user-1"),
+            datasource_manifest={"endpoint": "https://api.example?q=ok&token=hidden"},
+            model_manifest={"planner": {"model": "fake", "api_key": "hidden"}},
+        )
+    )
+    assert legacy.calls == [(10, 20, 30), (10, 20, 30)]
+    assert "hidden" not in str(assembled.canonical_payload())
+    assert assembled.strategies[0] == {"always_on": [7], "on_demand": [8]}

script_build_host/tests/test_legacy_input_and_adapters.py

@@ -0,0 +1,277 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from sqlalchemy import insert
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from script_build_host.adapters.persona import FilePersonaSource
+from script_build_host.adapters.prompts import DatabaseFirstPromptSource
+from script_build_host.adapters.strategy import SqlAlchemyStrategySource
+from script_build_host.domain.errors import InputRelationMismatch
+from script_build_host.domain.ports import PromptRequest
+from script_build_host.infrastructure.legacy_tables import (
+    build_strategy,
+    build_strategy_version,
+    prompt,
+    topic_build_composition_item,
+    topic_build_item_relation,
+    topic_build_item_source,
+    topic_build_point,
+    topic_build_point_item_relation,
+    topic_build_record,
+    topic_build_topic,
+    topic_pattern_execution,
+)
+from script_build_host.repositories.legacy_input import LegacySqlAlchemyInputReader
+
+
+async def _seed_topic(sessions: async_sessionmaker[AsyncSession], raw_upload_data: object) -> None:
+    async with sessions() as session, session.begin():
+        await session.execute(insert(topic_pattern_execution).values(id=10, status="success"))
+        await session.execute(
+            insert(topic_build_record).values(
+                id=20,
+                execution_id=10,
+                demand="make a useful script",
+                demand_constraints={"audience": "reader"},
+                agent_type="AigcAgent",
+                agent_config={"model_name": "test-model", "api_key": "must-redact-later"},
+                status="success",
+                is_deleted=False,
+                strategies_config={"always_on": [7], "on_demand": [8]},
+                personal_config={"account_name": "每天心理学"},
+                origin="upload",
+                raw_upload_data=raw_upload_data,
+            )
+        )
+        await session.execute(
+            insert(topic_build_topic).values(
+                id=30,
+                build_id=20,
+                execution_id=10,
+                sort_order=0,
+                topic_direction='{"title":"direction"}',
+                result="topic result",
+                status="mature",
+            )
+        )
+        await session.execute(
+            insert(topic_build_point),
+            [
+                {
+                    "id": 40,
+                    "topic_id": 30,
+                    "build_id": 20,
+                    "point_type": "key",
+                    "point_result": "active",
+                    "is_active": True,
+                },
+                {
+                    "id": 41,
+                    "topic_id": 30,
+                    "build_id": 20,
+                    "point_type": "key",
+                    "point_result": "inactive",
+                    "is_active": False,
+                },
+            ],
+        )
+        await session.execute(
+            insert(topic_build_composition_item),
+            [
+                {
+                    "id": 50,
+                    "topic_id": 30,
+                    "build_id": 20,
+                    "item_level": "element",
+                    "element_name": "first",
+                    "step": 1,
+                    "sort_order": 2,
+                    "is_active": True,
+                    "derivation_type": "user_demand",
+                },
+                {
+                    "id": 51,
+                    "topic_id": 30,
+                    "build_id": 20,
+                    "item_level": "element",
+                    "element_name": "second",
+                    "step": 1,
+                    "sort_order": 1,
+                    "is_active": True,
+                    "derivation_type": "agent_reasoning",
+                },
+            ],
+        )
+        await session.execute(
+            insert(topic_build_point_item_relation).values(id=60, point_id=40, item_id=50)
+        )
+        await session.execute(
+            insert(topic_build_item_relation).values(
+                id=70, topic_id=30, source_item_id=51, target_item_id=50, reason="because"
+            )
+        )
+        await session.execute(
+            insert(topic_build_item_source).values(
+                id=80,
+                topic_id=30,
+                target_item_id=50,
+                derivation_type="post_extract",
+                source_type="post",
+                source_reference_id="post-1",
+                source_detail={"title": "source"},
+                dataset_from="account",
+                is_active=True,
+            )
+        )
+
+
+@pytest.mark.asyncio
+async def test_legacy_reader_freezes_relations_sources_and_arbitrary_raw_upload(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    await _seed_topic(sessions, ["uploaded", {"nested": True}])
+    graph = await LegacySqlAlchemyInputReader(sessions).read_topic_graph(
+        execution_id=10, topic_build_id=20, topic_id=30
+    )
+    assert graph["build_record"]["raw_upload_data"] == ["uploaded", {"nested": True}]
+    assert [point["id"] for point in graph["points"]] == [40]
+    assert [item["id"] for item in graph["composition_items"]] == [51, 50]
+    assert graph["points"][0]["item_ids"] == [50]
+    assert graph["relations"][0]["source_item_id"] == 51
+    assert graph["sources"][0]["source_reference_id"] == "post-1"
+    assert graph["sources"][0]["derivation_type"] == "post_extract"
+    assert graph["sources"][0]["dataset_from"] == "account"
+
+
+@pytest.mark.asyncio
+async def test_legacy_reader_fails_closed_on_relation_mismatch(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    await _seed_topic(sessions, {})
+    reader = LegacySqlAlchemyInputReader(sessions)
+    with pytest.raises(InputRelationMismatch):
+        await reader.read_topic_graph(execution_id=11, topic_build_id=20, topic_id=30)
+
+    async with sessions() as session, session.begin():
+        await session.execute(
+            insert(topic_build_point_item_relation).values(
+                id=61,
+                point_id=999,
+                item_id=50,
+            )
+        )
+    with pytest.raises(InputRelationMismatch):
+        await reader.read_topic_graph(execution_id=10, topic_build_id=20, topic_id=30)
+
+
+@pytest.mark.asyncio
+async def test_persona_adapter_applies_alias_filters_and_versions(tmp_path: Path) -> None:
+    persona_root = tmp_path / "persona"
+    section_root = tmp_path / "sections"
+    alias = "每天一点心理学"
+    (persona_root / alias).mkdir(parents=True)
+    (section_root / alias).mkdir(parents=True)
+    (persona_root / alias / "point_records.json").write_text(
+        json.dumps(
+            {
+                "records": [
+                    {"阶段": "创作", "点类型": "维度值", "名称": "keep", "帖子覆盖率": 0.2},
+                    {"阶段": "创作", "点类型": "维度值", "名称": "drop", "帖子覆盖率": 0.19},
+                    {"阶段": "选题", "点类型": "维度名", "名称": "wrong stage", "帖子覆盖率": 1},
+                ]
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (section_root / alias / "section_main_dimension_template.json").write_text(
+        json.dumps({"分段规律摘要": "summary", "ignored": "value"}, ensure_ascii=False),
+        encoding="utf-8",
+    )
+    loaded = await FilePersonaSource(
+        persona_root=persona_root, section_pattern_root=section_root
+    ).load("每天心理学")
+    assert loaded.resolved_account_name == alias
+    assert [item["点名称"] for item in loaded.persona_points] == ["keep"]
+    assert loaded.section_patterns == ({"分段规律摘要": "summary"},)
+    assert set(loaded.source_versions) == {"persona_points", "section_patterns"}
+
+
+@pytest.mark.asyncio
+async def test_strategy_adapter_uses_old_integer_ids_and_current_versions(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    async with sessions() as session, session.begin():
+        await session.execute(
+            insert(build_strategy),
+            [
+                {
+                    "id": 7,
+                    "strategy_type": "script",
+                    "name": "always",
+                    "description": "A",
+                    "is_active": True,
+                    "current_version": 2,
+                },
+                {
+                    "id": 8,
+                    "strategy_type": "script",
+                    "name": "demand",
+                    "description": "B",
+                    "is_active": True,
+                    "current_version": 1,
+                },
+            ],
+        )
+        await session.execute(
+            insert(build_strategy_version),
+            [
+                {"id": 70, "strategy_id": 7, "version": 1, "content": "old"},
+                {"id": 71, "strategy_id": 7, "version": 2, "content": "always content"},
+                {"id": 80, "strategy_id": 8, "version": 1, "content": "demand content"},
+            ],
+        )
+    loaded = await SqlAlchemyStrategySource(sessions).load(always_on=(7,), on_demand=(8,))
+    assert [(item["strategy_id"], item["mode"], item["version"]) for item in loaded] == [
+        (7, "always_on", 2),
+        (8, "on_demand", 1),
+    ]
+    assert loaded[0]["content"] == "always content"
+    assert loaded[1]["content"] == "demand content"
+    assert loaded[1]["content_sha256"].startswith("sha256:")
+
+
+@pytest.mark.asyncio
+async def test_prompt_adapter_records_actual_db_and_file_sources(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path
+) -> None:
+    _, sessions = database
+    async with sessions() as session, session.begin():
+        await session.execute(
+            insert(prompt).values(
+                id=1,
+                biz_type="script_planner",
+                name="planner",
+                prompt_content="from database",
+                current_version=3,
+            )
+        )
+    (tmp_path / "worker.md").write_text("from file", encoding="utf-8")
+    source = DatabaseFirstPromptSource(sessions, fallback_root=tmp_path)
+    manifests = await source.load(
+        (
+            PromptRequest("script_planner", "planner.md", "script_planner", "planner"),
+            PromptRequest("worker", "worker.md", "script_worker", "worker"),
+        )
+    )
+    assert [(item["source"], item["version"]) for item in manifests] == [
+        ("database", 3),
+        ("file", None),
+    ]