取数：实现受控检索、图片和上传适配器

实现 Pattern 异步轮询、Decode 精确协议、External、Knowledge 和图片取数，并统一经过出站安全策略与响应大小限制。

原始图片按内容寻址冻结，上传 Topic 受数量和字节护栏保护，检索结果只输出脱敏摘要、来源引用和限制。

script_build_host/src/script_build_host/adapters/__init__.py

@@ -0,0 +1,31 @@
+"""Concrete Host adapters."""
+
+from .persona import FilePersonaSource
+from .prompts import DatabaseFirstPromptSource
+from .retrieval import (
+    ExternalRetrievalAdapter,
+    FileDecodeRetrievalAdapter,
+    FileKnowledgeRetrievalAdapter,
+    HttpDecodeRetrievalAdapter,
+    PatternRetrievalAdapter,
+    RawArtifactStore,
+    SafeHttpClient,
+    SafeImageAdapter,
+)
+from .strategy import SqlAlchemyStrategySource
+from .uploaded_topic import SqlUploadedTopicGateway
+
+__all__ = [
+    "DatabaseFirstPromptSource",
+    "ExternalRetrievalAdapter",
+    "FileDecodeRetrievalAdapter",
+    "FileKnowledgeRetrievalAdapter",
+    "FilePersonaSource",
+    "HttpDecodeRetrievalAdapter",
+    "PatternRetrievalAdapter",
+    "RawArtifactStore",
+    "SafeHttpClient",
+    "SafeImageAdapter",
+    "SqlAlchemyStrategySource",
+    "SqlUploadedTopicGateway",
+]

script_build_host/src/script_build_host/adapters/retrieval.py

@@ -0,0 +1,468 @@
+"""Concrete, side-effect-contained retrieval and image adapters."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+import time
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from hashlib import sha256
+from pathlib import Path
+from typing import Any, Protocol
+from urllib.parse import urljoin
+
+import httpx
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.infrastructure.outbound import OutboundPolicy
+from script_build_host.infrastructure.redaction import redact, redact_text
+from script_build_host.tools.gateway import RetrievalResult
+
+
+class SafeHttpClient:
+    """HTTP JSON/bytes client that revalidates every redirect target."""
+
+    def __init__(
+        self,
+        client: httpx.AsyncClient,
+        policy: OutboundPolicy,
+        *,
+        timeout_seconds: float = 30.0,
+        max_response_bytes: int = 5_000_000,
+        max_redirects: int = 3,
+    ) -> None:
+        self.client = client
+        self.policy = policy
+        self.timeout_seconds = timeout_seconds
+        self.max_response_bytes = max_response_bytes
+        self.max_redirects = max_redirects
+
+    async def request(
+        self,
+        method: str,
+        url: str,
+        *,
+        json_body: Mapping[str, Any] | None = None,
+    ) -> httpx.Response:
+        current = await self.policy.validate_url(url)
+        for _ in range(self.max_redirects + 1):
+            request = self.client.build_request(
+                method,
+                current,
+                json=dict(json_body) if json_body is not None else None,
+                timeout=self.timeout_seconds,
+            )
+            response = await self.client.send(request, stream=True, follow_redirects=False)
+            if 300 <= response.status_code < 400:
+                location = response.headers.get("location")
+                await response.aclose()
+                if not location:
+                    raise ProtocolViolation("redirect response has no location")
+                current = await self.policy.validate_redirect(current, location)
+                method = "GET" if response.status_code in {301, 302, 303} else method
+                json_body = None if method == "GET" else json_body
+                continue
+            content = bytearray()
+            async for chunk in response.aiter_bytes():
+                content.extend(chunk)
+                if len(content) > self.max_response_bytes:
+                    await response.aclose()
+                    raise ProtocolViolation("upstream response exceeds the configured byte limit")
+            await response.aclose()
+            if response.is_error:
+                raise ProtocolViolation(f"upstream returned HTTP {response.status_code}")
+            return httpx.Response(
+                status_code=response.status_code,
+                headers=response.headers,
+                content=bytes(content),
+                request=request,
+            )
+        raise ProtocolViolation("upstream exceeded the redirect limit")
+
+    async def json(
+        self,
+        method: str,
+        url: str,
+        *,
+        json_body: Mapping[str, Any] | None = None,
+    ) -> Any:
+        response = await self.request(method, url, json_body=json_body)
+        try:
+            return response.json()
+        except ValueError as exc:
+            raise ProtocolViolation("upstream response is not valid JSON") from exc
+
+
+class PatternRetrievalAdapter:
+    def __init__(
+        self,
+        endpoint: str,
+        http: SafeHttpClient,
+        *,
+        poll_interval_seconds: float = 1.0,
+        poll_timeout_seconds: float = 600.0,
+    ) -> None:
+        self.endpoint = endpoint
+        self.http = http
+        self.poll_interval_seconds = poll_interval_seconds
+        self.poll_timeout_seconds = poll_timeout_seconds
+
+    async def retrieve(self, *, query: Mapping[str, Any], snapshot: Any) -> RetrievalResult:
+        del snapshot
+        payload = await self.http.json("POST", self.endpoint, json_body=query)
+        task_id = (
+            str(payload.get("session_id") or payload.get("task_id") or "")
+            if isinstance(payload, dict)
+            else ""
+        )
+        deadline = time.monotonic() + self.poll_timeout_seconds
+        while isinstance(payload, dict) and payload.get("status") in {
+            "pending",
+            "running",
+            "queued",
+        }:
+            if time.monotonic() >= deadline:
+                return _limited_result(
+                    "pattern",
+                    "pattern retrieval timed out",
+                    "timeout",
+                    task_id=task_id,
+                    payload=payload,
+                )
+            poll_url = payload.get("poll_url")
+            if not isinstance(poll_url, str) or not poll_url:
+                if not task_id:
+                    raise ProtocolViolation("pattern pending response has no task identity")
+                poll_url = urljoin(self.endpoint.rstrip("/") + "/", task_id)
+            await asyncio.sleep(self.poll_interval_seconds)
+            payload = await self.http.json("GET", poll_url)
+        if isinstance(payload, dict) and payload.get("status") in {"failed", "error"}:
+            return _limited_result(
+                "pattern",
+                "pattern retrieval failed",
+                "upstream failure",
+                task_id=task_id,
+                payload=payload,
+            )
+        return _result(
+            "pattern",
+            payload,
+            metadata={
+                "task_id": _safe_source_id(task_id, rank=0) if task_id else None,
+                "response_sha256": canonical_sha256(payload).wire,
+            },
+        )
+
+
+class ExternalRetrievalAdapter:
+    """Pure HTTP client; it contains no legacy-log repository dependency."""
+
+    def __init__(self, endpoint: str, http: SafeHttpClient) -> None:
+        self.endpoint = endpoint
+        self.http = http
+
+    async def retrieve(self, *, query: Mapping[str, Any], snapshot: Any) -> RetrievalResult:
+        del snapshot
+        payload = await self.http.json("POST", self.endpoint, json_body=query)
+        return _result(
+            "external",
+            payload,
+            metadata={"response_sha256": canonical_sha256(payload).wire},
+        )
+
+
+class HttpDecodeRetrievalAdapter:
+    """Call the configured Decode service with the legacy query contract."""
+
+    def __init__(self, endpoint: str, http: SafeHttpClient) -> None:
+        self.endpoint = endpoint
+        self.http = http
+
+    async def retrieve(self, *, query: Mapping[str, Any], snapshot: Any) -> RetrievalResult:
+        payload = await self.http.json("POST", self.endpoint, json_body=query)
+        rows = (
+            payload.get("data", payload.get("results", payload.get("items", [])))
+            if isinstance(payload, dict)
+            else payload
+        )
+        if not isinstance(rows, list):
+            raise ProtocolViolation("decode response items must be an array")
+        safe_rows = redact(rows)
+        if not isinstance(safe_rows, list):
+            raise ProtocolViolation("redacted decode rows must remain an array")
+        refs: list[str] = []
+        scores: list[float] = []
+        for row in safe_rows:
+            if not isinstance(row, dict):
+                raise ProtocolViolation("decode hit must be an object")
+            identifier = row.get("post_id", row.get("id"))
+            raw_score = row.get("score", row.get("similarity_score"))
+            if (
+                identifier is None
+                or isinstance(raw_score, bool)
+                or not isinstance(raw_score, (int, float))
+            ):
+                raise ProtocolViolation("decode hit must contain an ID and numeric score")
+            refs.append(f"decode:{_safe_source_id(identifier, rank=len(refs))}")
+            scores.append(float(raw_score))
+        datasource_manifest = getattr(snapshot, "datasource_manifest", {})
+        model_manifest = getattr(snapshot, "model_manifest", {})
+        return RetrievalResult(
+            source_refs=tuple(refs),
+            summary=json.dumps(safe_rows, ensure_ascii=False, sort_keys=True),
+            supports=tuple(str(query["return_field"]) for _ in refs),
+            confidence="ranked",
+            limitations=(),
+            metadata={
+                "response_sha256": canonical_sha256(payload).wire,
+                "decode_index": datasource_manifest.get("decode_index"),
+                "embedding_model": model_manifest.get("embedding_model"),
+                "hit_ids": refs,
+                "ranks": list(range(1, len(refs) + 1)),
+                "scores": scores,
+            },
+        )
+
+
+class FileDecodeRetrievalAdapter:
+    def __init__(self, index_path: Path) -> None:
+        self.index_path = index_path
+        self._payload, self._digest = _read_json(index_path)
+
+    async def retrieve(self, *, query: Mapping[str, Any], snapshot: Any) -> RetrievalResult:
+        payload, digest = self._payload, self._digest
+        _assert_frozen_source(snapshot, "decode_index", digest)
+        rows = payload if isinstance(payload, list) else payload.get("items", [])
+        if not isinstance(rows, list):
+            raise ProtocolViolation("decode index items must be an array")
+        account = query.get("account_name")
+        keyword = str(query.get("keyword", "")).casefold()
+        scored: list[tuple[int, int, dict[str, Any]]] = []
+        for position, item in enumerate(rows):
+            if not isinstance(item, dict):
+                continue
+            if account and item.get("account") != account:
+                continue
+            text = json.dumps(item, ensure_ascii=False, sort_keys=True).casefold()
+            score = text.count(keyword) if keyword else 1
+            if score:
+                scored.append((score, -position, item))
+        scored.sort(reverse=True, key=lambda item: (item[0], item[1]))
+        top_k = int(query.get("top_k", 3))
+        ranked = scored[:top_k]
+        hits = [item for _, _, item in ranked]
+        refs = tuple(
+            f"decode:{_safe_source_id(item.get('post_id', item.get('id')), rank=index)}"
+            for index, item in enumerate(hits)
+        )
+        safe_hits = redact(hits)
+        if not isinstance(safe_hits, list):
+            raise ProtocolViolation("redacted decode hits must remain an array")
+        embedding_model = str(snapshot.model_manifest.get("embedding_model", "lexical-fixture-v1"))
+        return RetrievalResult(
+            source_refs=refs,
+            summary=json.dumps(safe_hits, ensure_ascii=False, sort_keys=True),
+            supports=tuple(str(query["return_field"]) for _ in hits),
+            confidence="ranked",
+            limitations=("offline frozen index",),
+            metadata={
+                "index_sha256": digest,
+                "embedding_model": embedding_model,
+                "hit_ids": list(refs),
+                "ranks": list(range(1, len(refs) + 1)),
+                "scores": [float(score) for score, _, _ in ranked],
+            },
+        )
+
+
+class FileKnowledgeRetrievalAdapter:
+    def __init__(self, knowledge_path: Path) -> None:
+        self.knowledge_path = knowledge_path
+        self._payload, self._digest = _read_json(knowledge_path)
+
+    async def retrieve(self, *, query: Mapping[str, Any], snapshot: Any) -> RetrievalResult:
+        payload, digest = self._payload, self._digest
+        _assert_frozen_source(snapshot, "knowledge", digest)
+        rows = payload if isinstance(payload, list) else payload.get("items", [])
+        if not isinstance(rows, list):
+            raise ProtocolViolation("knowledge items must be an array")
+        keyword = str(query.get("keyword", "")).casefold()
+        selected = [
+            item
+            for item in rows
+            if isinstance(item, dict) and keyword in json.dumps(item, ensure_ascii=False).casefold()
+        ][: int(query.get("max_count", 3))]
+        refs = tuple(
+            f"knowledge:{_safe_source_id(item.get('id'), rank=index)}"
+            for index, item in enumerate(selected)
+        )
+        safe_selected = redact(selected)
+        if not isinstance(safe_selected, list):
+            raise ProtocolViolation("redacted knowledge hits must remain an array")
+        return RetrievalResult(
+            source_refs=refs,
+            summary=json.dumps(safe_selected, ensure_ascii=False, sort_keys=True),
+            supports=tuple(
+                redact_text(str(item.get("title", ref)))
+                for item, ref in zip(selected, refs, strict=True)
+            ),
+            confidence="lexical",
+            limitations=("frozen local knowledge snapshot",),
+            metadata={
+                "file_sha256": digest,
+                "hit_ids": list(refs),
+                "ranks": list(range(1, len(refs) + 1)),
+            },
+        )
+
+
+class RawArtifactStore(Protocol):
+    async def freeze_bytes(self, content: bytes, *, media_type: str) -> str: ...
+
+
+@dataclass(frozen=True)
+class SafeImageAdapter:
+    http: SafeHttpClient
+    raw_store: RawArtifactStore
+    max_images: int = 12
+    max_image_bytes: int = 10_000_000
+    max_total_image_bytes: int = 50_000_000
+    allowed_mime_types: frozenset[str] = frozenset(
+        {"image/jpeg", "image/png", "image/webp", "image/gif"}
+    )
+
+    async def load(self, *, urls: Sequence[str], snapshot: Any) -> tuple[dict[str, Any], ...]:
+        del snapshot
+        if len(urls) > self.max_images:
+            raise ProtocolViolation("image count exceeds the configured limit")
+        total = 0
+        items: list[dict[str, Any]] = []
+        for url in urls:
+            response = await self.http.request("GET", url)
+            declared_type = response.headers.get("content-type", "").split(";", 1)[0]
+            content_type = _sniff_image_type(response.content)
+            if content_type is None or content_type not in self.allowed_mime_types:
+                raise ProtocolViolation("upstream image MIME type is not permitted")
+            if declared_type and declared_type != content_type:
+                raise ProtocolViolation("upstream image MIME type does not match its bytes")
+            size = len(response.content)
+            if size > self.max_image_bytes:
+                raise ProtocolViolation("image exceeds the configured byte limit")
+            total += size
+            if total > self.max_total_image_bytes:
+                raise ProtocolViolation("images exceed the configured total byte limit")
+            raw_artifact_ref = await self.raw_store.freeze_bytes(
+                response.content,
+                media_type=content_type,
+            )
+            items.append(
+                {
+                    "kind": "image",
+                    "digest": "sha256:" + sha256(response.content).hexdigest(),
+                    "raw_artifact_ref": raw_artifact_ref,
+                    "content_type": content_type,
+                    "size_bytes": size,
+                }
+            )
+        return tuple(items)
+
+
+def _read_json(path: Path) -> tuple[Any, str]:
+    if not path.is_file():
+        raise ProtocolViolation("configured retrieval file does not exist")
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ProtocolViolation("configured retrieval file is not valid UTF-8 JSON") from exc
+    return payload, canonical_sha256(payload).wire
+
+
+def _assert_frozen_source(snapshot: Any, key: str, actual_digest: str) -> None:
+    manifest = getattr(snapshot, "datasource_manifest", {})
+    expected = manifest.get(key) if isinstance(manifest, Mapping) else None
+    expected_digest = expected.get("sha256") if isinstance(expected, Mapping) else None
+    if expected_digest is not None and expected_digest != actual_digest:
+        raise ProtocolViolation(f"{key} changed after the input snapshot was frozen")
+
+
+def _sniff_image_type(content: bytes) -> str | None:
+    if content.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image/png"
+    if content.startswith(b"\xff\xd8\xff"):
+        return "image/jpeg"
+    if content.startswith((b"GIF87a", b"GIF89a")):
+        return "image/gif"
+    if len(content) >= 12 and content[:4] == b"RIFF" and content[8:12] == b"WEBP":
+        return "image/webp"
+    return None
+
+
+def _result(
+    source: str,
+    payload: Any,
+    *,
+    metadata: Mapping[str, Any],
+) -> RetrievalResult:
+    rows = payload.get("data", payload.get("results", [])) if isinstance(payload, dict) else payload
+    if not isinstance(rows, list):
+        rows = [payload]
+    safe_rows = redact(rows)
+    if not isinstance(safe_rows, list):
+        raise ProtocolViolation("redacted upstream rows must remain an array")
+    refs = tuple(
+        f"{source}:{_safe_source_id(item.get('channel_content_id', item.get('id')), rank=index)}"
+        if isinstance(item, dict)
+        else f"{source}:{index}"
+        for index, item in enumerate(safe_rows)
+    )
+    return RetrievalResult(
+        source_refs=refs,
+        summary=json.dumps(safe_rows, ensure_ascii=False, sort_keys=True),
+        supports=refs,
+        confidence="upstream",
+        limitations=(),
+        metadata=metadata,
+    )
+
+
+def _safe_source_id(value: Any, *, rank: int) -> str:
+    text = str(value if value is not None else rank)
+    if re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", text):
+        return text
+    return "sha256-" + sha256(text.encode("utf-8")).hexdigest()[:20]
+
+
+def _limited_result(
+    source: str,
+    summary: str,
+    limitation: str,
+    *,
+    task_id: str,
+    payload: Any,
+) -> RetrievalResult:
+    return RetrievalResult(
+        source_refs=(),
+        summary=summary,
+        supports=(),
+        confidence="unavailable",
+        limitations=(limitation,),
+        metadata={
+            "task_id": _safe_source_id(task_id, rank=0) if task_id else None,
+            "response_sha256": canonical_sha256(payload).wire,
+        },
+    )
+
+
+__all__ = [
+    "ExternalRetrievalAdapter",
+    "FileDecodeRetrievalAdapter",
+    "FileKnowledgeRetrievalAdapter",
+    "HttpDecodeRetrievalAdapter",
+    "PatternRetrievalAdapter",
+    "RawArtifactStore",
+    "SafeHttpClient",
+    "SafeImageAdapter",
+]

script_build_host/src/script_build_host/adapters/uploaded_topic.py

@@ -0,0 +1,240 @@
+"""Safe uploaded-topic parser and transactional legacy input writer."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any, cast
+
+from sqlalchemy import insert
+from sqlalchemy.engine import CursorResult
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.infrastructure.legacy_tables import (
+    topic_build_composition_item,
+    topic_build_point,
+    topic_build_point_item_relation,
+    topic_build_record,
+    topic_build_topic,
+)
+from script_build_host.infrastructure.redaction import redact
+
+
+class SqlUploadedTopicGateway:
+    def __init__(
+        self,
+        sessions: async_sessionmaker[AsyncSession],
+        *,
+        max_payload_bytes: int = 5_000_000,
+        max_points: int = 500,
+        max_items: int = 5_000,
+    ) -> None:
+        self._sessions = sessions
+        self._max_payload_bytes = max_payload_bytes
+        self._max_points = max_points
+        self._max_items = max_items
+
+    async def parse(self, value: dict[str, Any] | str) -> dict[str, Any]:
+        if isinstance(value, str):
+            if len(value.encode("utf-8")) > self._max_payload_bytes:
+                raise ProtocolViolation("uploaded topic exceeds the configured byte limit")
+            try:
+                data = json.loads(value)
+            except json.JSONDecodeError as exc:
+                raise ProtocolViolation("uploaded topic is not valid JSON") from exc
+        else:
+            data = dict(value)
+            encoded = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+            if len(encoded) > self._max_payload_bytes:
+                raise ProtocolViolation("uploaded topic exceeds the configured byte limit")
+        fusion = data.get("选题融合")
+        if not isinstance(fusion, str) or not fusion.strip():
+            raise ProtocolViolation("uploaded topic is missing 选题融合")
+        target = data.get("target_post")
+        account_name = (
+            target.get("channel_account_name")
+            if isinstance(target, dict) and isinstance(target.get("channel_account_name"), str)
+            else None
+        )
+        points: list[dict[str, Any]] = []
+        item_count = 0
+        for point_type in ("灵感点", "目的点", "关键点"):
+            rows = data.get(point_type)
+            if not isinstance(rows, list):
+                continue
+            for row in rows:
+                if not isinstance(row, dict):
+                    continue
+                items = _point_items(row)
+                point_result = row.get("点")
+                if not isinstance(point_result, str) or not point_result.strip():
+                    point_result = items[0]["element_name"] if items else None
+                if point_result is None:
+                    continue
+                description = row.get("点描述")
+                points.append(
+                    {
+                        "point_type": point_type,
+                        "point_result": point_result.strip()[:255],
+                        "point_description": (
+                            description.strip()
+                            if isinstance(description, str) and description.strip()
+                            else None
+                        ),
+                        "items": items,
+                    }
+                )
+                if len(points) > self._max_points:
+                    raise ProtocolViolation("uploaded topic contains too many points")
+                item_count += len(items)
+                if item_count > self._max_items:
+                    raise ProtocolViolation("uploaded topic contains too many elements")
+        if item_count == 0:
+            raise ProtocolViolation("uploaded topic contains no point elements")
+        return {
+            "account_name": account_name.strip() if account_name else None,
+            "topic_fusion": fusion.strip(),
+            "points": points,
+            "item_count": item_count,
+            "raw": redact(data),
+        }
+
+    async def create(self, parsed: dict[str, Any], *, account_name: str | None) -> dict[str, Any]:
+        resolved_account = (account_name or parsed.get("account_name") or "").strip() or None
+        now = datetime.now(UTC).replace(tzinfo=None)
+        async with self._sessions() as session, session.begin():
+            build_result = cast(
+                CursorResult[Any],
+                await session.execute(
+                    insert(topic_build_record).values(
+                        execution_id=0,
+                        demand=(
+                            f"[上传选题解构] {resolved_account}"
+                            if resolved_account
+                            else "[上传选题解构]"
+                        ),
+                        agent_type="UploadedTopic",
+                        status="success",
+                        is_deleted=False,
+                        personal_config=(
+                            {"account_name": resolved_account} if resolved_account else None
+                        ),
+                        origin="upload",
+                        raw_upload_data=parsed["raw"],
+                        start_time=now,
+                        end_time=now,
+                    )
+                ),
+            )
+            topic_build_id = _inserted_id(build_result)
+            topic_result = cast(
+                CursorResult[Any],
+                await session.execute(
+                    insert(topic_build_topic).values(
+                        build_id=topic_build_id,
+                        execution_id=0,
+                        sort_order=0,
+                        result=parsed["topic_fusion"],
+                        status="mature",
+                    )
+                ),
+            )
+            topic_id = _inserted_id(topic_result)
+            sort_order = 0
+            for point_data in parsed["points"]:
+                point_result = cast(
+                    CursorResult[Any],
+                    await session.execute(
+                        insert(topic_build_point).values(
+                            topic_id=topic_id,
+                            build_id=topic_build_id,
+                            point_type=point_data["point_type"],
+                            point_result=point_data["point_result"],
+                            is_active=True,
+                            note=point_data.get("point_description"),
+                            created_at=now,
+                            updated_at=now,
+                        )
+                    ),
+                )
+                point_id = _inserted_id(point_result)
+                for item in point_data["items"]:
+                    item_result = cast(
+                        CursorResult[Any],
+                        await session.execute(
+                            insert(topic_build_composition_item).values(
+                                topic_id=topic_id,
+                                build_id=topic_build_id,
+                                item_level="element",
+                                dimension=item["dimension"],
+                                point_type=point_data["point_type"],
+                                element_name=item["element_name"],
+                                step=0,
+                                sort_order=sort_order,
+                                is_active=True,
+                                note=item.get("note"),
+                                created_at=now,
+                                updated_at=now,
+                            )
+                        ),
+                    )
+                    item_id = _inserted_id(item_result)
+                    await session.execute(
+                        insert(topic_build_point_item_relation).values(
+                            point_id=point_id, item_id=item_id
+                        )
+                    )
+                    sort_order += 1
+        return {
+            "topic_build_id": topic_build_id,
+            "topic_id": topic_id,
+            "item_count": sort_order,
+            "point_count": len(parsed["points"]),
+            "account_name": resolved_account,
+        }
+
+
+def _point_items(point: dict[str, Any]) -> list[dict[str, Any]]:
+    items: list[dict[str, Any]] = []
+    seen: set[tuple[str, str]] = set()
+    for dimension in ("实质", "形式", "意图"):
+        block = point.get(dimension)
+        lists = (
+            [value for value in block.values() if isinstance(value, list)]
+            if isinstance(block, dict)
+            else [block]
+            if isinstance(block, list)
+            else []
+        )
+        for values in lists:
+            for value in values:
+                if not isinstance(value, dict):
+                    continue
+                name = value.get("名称")
+                if not isinstance(name, str) or not name.strip():
+                    continue
+                key = (dimension, name.strip())
+                if key in seen:
+                    continue
+                seen.add(key)
+                note = value.get("说明")
+                items.append(
+                    {
+                        "element_name": name.strip()[:500],
+                        "dimension": dimension,
+                        "note": note.strip() if isinstance(note, str) and note.strip() else None,
+                    }
+                )
+    return items
+
+
+def _inserted_id(result: CursorResult[Any]) -> int:
+    keys = result.inserted_primary_key
+    value = keys[0] if keys else None
+    if value is None:
+        raise ProtocolViolation("legacy insert did not return a primary key")
+    return int(value)
+
+
+__all__ = ["SqlUploadedTopicGateway"]

script_build_host/src/script_build_host/infrastructure/__init__.py

@@ -0,0 +1,13 @@
+from .canonical_json import canonical_json_bytes, canonical_json_text, canonical_sha256
+from .config import ScriptBuildSettings
+from .outbound import OutboundPolicy
+from .raw_artifacts import FileRawArtifactStore
+
+__all__ = [
+    "FileRawArtifactStore",
+    "OutboundPolicy",
+    "ScriptBuildSettings",
+    "canonical_json_bytes",
+    "canonical_json_text",
+    "canonical_sha256",
+]

script_build_host/src/script_build_host/infrastructure/raw_artifacts.py

@@ -0,0 +1,45 @@
+"""Content-addressed storage for bounded raw retrieval payloads."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from dataclasses import dataclass
+from hashlib import sha256
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+
+
+@dataclass(frozen=True, slots=True)
+class FileRawArtifactStore:
+    root: Path
+
+    async def freeze_bytes(self, content: bytes, *, media_type: str) -> str:
+        del media_type
+        digest = sha256(content).hexdigest()
+        await asyncio.to_thread(self._write_once, digest, content)
+        return f"script-build://raw-artifacts/sha256/{digest}"
+
+    def _write_once(self, digest: str, content: bytes) -> None:
+        directory = (self.root / "raw-artifacts" / "sha256").resolve()
+        resolved_root = self.root.resolve()
+        if not directory.is_relative_to(resolved_root):
+            raise ValueError("raw artifact directory escapes its configured root")
+        directory.mkdir(parents=True, exist_ok=True)
+        target = directory / digest
+        if target.exists():
+            if sha256(target.read_bytes()).hexdigest() != digest:
+                raise ValueError("existing raw artifact failed digest verification")
+            return
+        with NamedTemporaryFile(dir=directory, delete=False) as temporary:
+            temporary.write(content)
+            temporary.flush()
+            os.fsync(temporary.fileno())
+            temporary_path = Path(temporary.name)
+        try:
+            os.replace(temporary_path, target)
+        finally:
+            temporary_path.unlink(missing_ok=True)
+
+
+__all__ = ["FileRawArtifactStore"]

script_build_host/tests/test_retrieval_adapters.py

@@ -0,0 +1,317 @@
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from script_build_host.adapters.retrieval import (
+    ExternalRetrievalAdapter,
+    FileDecodeRetrievalAdapter,
+    FileKnowledgeRetrievalAdapter,
+    HttpDecodeRetrievalAdapter,
+    PatternRetrievalAdapter,
+    SafeHttpClient,
+    SafeImageAdapter,
+)
+from script_build_host.adapters.uploaded_topic import SqlUploadedTopicGateway
+from script_build_host.domain.errors import ProtocolViolation, UnsafeOutboundTarget
+from script_build_host.infrastructure.outbound import OutboundPolicy
+from script_build_host.repositories.legacy_input import LegacySqlAlchemyInputReader
+
+
+async def _public_resolver(_: str, __: int) -> tuple[str, ...]:
+    return ("93.184.216.34",)
+
+
+class _RawArtifacts:
+    def __init__(self) -> None:
+        self.values: list[bytes] = []
+
+    async def freeze_bytes(self, content: bytes, *, media_type: str) -> str:
+        assert media_type == "image/png"
+        self.values.append(content)
+        return "script-build://raw-artifacts/sha256/" + "a" * 64
+
+
+@pytest.mark.asyncio
+async def test_http_retrieval_revalidates_redirect_and_freezes_response_metadata() -> None:
+    calls: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(str(request.url))
+        return httpx.Response(
+            200,
+            json={"data": [{"channel_content_id": "post-1", "title": "case"}]},
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        result = await ExternalRetrievalAdapter("https://api.example.com/search", safe).retrieve(
+            query={"keyword": "case"}, snapshot=SimpleNamespace()
+        )
+    assert calls == ["https://api.example.com/search"]
+    assert result.source_refs == ("external:post-1",)
+    assert str(result.metadata["response_sha256"]).startswith("sha256:")
+
+    def redirect(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(302, headers={"location": "https://127.0.0.1/private"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(redirect)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        with pytest.raises(UnsafeOutboundTarget):
+            await safe.request("GET", "https://api.example.com/start")
+
+    def oversized(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, content=b"123456")
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(oversized)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+            max_response_bytes=5,
+        )
+        with pytest.raises(ProtocolViolation, match="byte limit"):
+            await safe.request("GET", "https://api.example.com/large")
+
+
+@pytest.mark.asyncio
+async def test_external_response_secrets_are_redacted_before_evidence_summary() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "data": [
+                    {
+                        "id": "p1",
+                        "authorization": "Bearer raw-token",
+                        "url": "https://source.example/item?token=secret",
+                    }
+                ]
+            },
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        result = await ExternalRetrievalAdapter("https://api.example.com/search", safe).retrieve(
+            query={"keyword": "case"}, snapshot=SimpleNamespace()
+        )
+    assert "raw-token" not in result.summary
+    assert "secret" not in result.summary
+
+
+@pytest.mark.asyncio
+async def test_pattern_polling_and_image_limits_use_mock_transport() -> None:
+    poll_count = 0
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal poll_count
+        if request.url.path == "/pattern":
+            return httpx.Response(
+                200,
+                json={
+                    "status": "pending",
+                    "session_id": "task-1",
+                    "poll_url": "https://api.example.com/poll/task-1",
+                },
+            )
+        if request.url.path.startswith("/poll"):
+            poll_count += 1
+            return httpx.Response(200, json={"status": "done", "results": [{"id": 2}]})
+        return httpx.Response(
+            200,
+            content=b"\x89PNG\r\n\x1a\nimage",
+            headers={"content-type": "image/png"},
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        result = await PatternRetrievalAdapter(
+            "https://api.example.com/pattern",
+            safe,
+            poll_interval_seconds=0.001,
+        ).retrieve(query={"message": "shape"}, snapshot=SimpleNamespace())
+        raw_store = _RawArtifacts()
+        images = await SafeImageAdapter(
+            safe,
+            raw_store,
+            max_images=1,
+            max_image_bytes=20,
+            max_total_image_bytes=20,
+        ).load(urls=["https://api.example.com/image"], snapshot=SimpleNamespace())
+    assert poll_count == 1
+    assert result.source_refs == ("pattern:2",)
+    assert images[0]["digest"].startswith("sha256:")
+    assert "url" not in images[0]
+    assert images[0]["raw_artifact_ref"].startswith("script-build://raw-artifacts/")
+    assert len(raw_store.values) == 1
+
+
+@pytest.mark.asyncio
+async def test_pattern_timeout_is_frozen_as_a_limitation() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"status": "pending", "session_id": "task-1"})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        result = await PatternRetrievalAdapter(
+            "https://api.example.com/pattern",
+            safe,
+            poll_interval_seconds=0.001,
+            poll_timeout_seconds=0,
+        ).retrieve(query={"message": "shape"}, snapshot=SimpleNamespace())
+    assert result.source_refs == ()
+    assert result.limitations == ("timeout",)
+    assert result.metadata["task_id"] == "task-1"
+
+
+@pytest.mark.asyncio
+async def test_file_decode_and_knowledge_freeze_file_hash_and_rank(tmp_path) -> None:
+    decode_path = tmp_path / "decode.json"
+    decode_path.write_text(
+        json.dumps({"items": [{"post_id": 4, "account": "acct", "text": "focus"}]}),
+        encoding="utf-8",
+    )
+    knowledge_path = tmp_path / "knowledge.json"
+    knowledge_path.write_text(json.dumps([{"id": "k1", "title": "focus method"}]), encoding="utf-8")
+    decode_adapter = FileDecodeRetrievalAdapter(decode_path)
+    snapshot = SimpleNamespace(
+        model_manifest={"embedding_model": "frozen-embed"},
+        datasource_manifest={},
+    )
+    decode = await decode_adapter.retrieve(
+        query={
+            "keyword": "focus",
+            "account_name": "acct",
+            "top_k": 3,
+            "return_field": "主脉络",
+        },
+        snapshot=snapshot,
+    )
+    knowledge = await FileKnowledgeRetrievalAdapter(knowledge_path).retrieve(
+        query={"keyword": "focus", "max_count": 3}, snapshot=snapshot
+    )
+    assert decode.metadata["embedding_model"] == "frozen-embed"
+    assert decode.metadata["ranks"] == [1]
+    assert decode.metadata["scores"] == [1.0]
+    assert str(knowledge.metadata["file_sha256"]).startswith("sha256:")
+
+    frozen_digest = decode.metadata["index_sha256"]
+    decode_path.write_text(json.dumps({"items": []}), encoding="utf-8")
+    replay = await decode_adapter.retrieve(
+        query={"keyword": "focus", "top_k": 3, "return_field": "主脉络"},
+        snapshot=SimpleNamespace(
+            model_manifest={},
+            datasource_manifest={"decode_index": {"sha256": frozen_digest}},
+        ),
+    )
+    assert replay.source_refs == ("decode:4",)
+    with pytest.raises(ProtocolViolation, match="changed"):
+        await FileDecodeRetrievalAdapter(decode_path).retrieve(
+            query={"keyword": "focus", "top_k": 3, "return_field": "主脉络"},
+            snapshot=SimpleNamespace(
+                model_manifest={},
+                datasource_manifest={"decode_index": {"sha256": frozen_digest}},
+            ),
+        )
+
+
+@pytest.mark.asyncio
+async def test_http_decode_preserves_legacy_query_and_ranked_scores() -> None:
+    requests: list[dict[str, object]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(json.loads(request.content))
+        return httpx.Response(
+            200,
+            json={"items": [{"post_id": "p1", "score": 0.91, "text": "safe"}]},
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"decode.example"}), resolver=_public_resolver),
+        )
+        result = await HttpDecodeRetrievalAdapter(
+            "https://decode.example/search",
+            safe,
+        ).retrieve(
+            query={
+                "return_field": "主脉络",
+                "keyword": "focus",
+                "account_name": "acct",
+                "match_fields": {},
+                "top_k": 3,
+            },
+            snapshot=SimpleNamespace(
+                datasource_manifest={"decode_index": {"sha256": "sha256:" + "1" * 64}},
+                model_manifest={"embedding_model": "embed-v1"},
+            ),
+        )
+    assert requests[0]["top_k"] == 3
+    assert result.source_refs == ("decode:p1",)
+    assert result.metadata["scores"] == [0.91]
+    assert result.metadata["ranks"] == [1]
+
+
+@pytest.mark.asyncio
+async def test_uploaded_topic_parser_keeps_old_shape_without_writing() -> None:
+    gateway = SqlUploadedTopicGateway(SimpleNamespace())  # type: ignore[arg-type]
+    parsed = await gateway.parse(
+        {
+            "选题融合": "topic",
+            "target_post": {"channel_account_name": "acct"},
+            "灵感点": [
+                {
+                    "点": "point",
+                    "实质": {"具体元素": [{"名称": "element", "说明": "note"}]},
+                }
+            ],
+        }
+    )
+    assert parsed["account_name"] == "acct"
+    assert parsed["item_count"] == 1
+    assert parsed["points"][0]["items"][0]["dimension"] == "实质"
+
+
+@pytest.mark.asyncio
+async def test_uploaded_topic_create_writes_a_readable_legacy_graph(database) -> None:
+    _, sessions = database
+    gateway = SqlUploadedTopicGateway(sessions)
+    parsed = await gateway.parse(
+        {
+            "选题融合": "uploaded topic",
+            "target_post": {"channel_account_name": "acct"},
+            "关键点": [
+                {
+                    "点": "point",
+                    "形式": [{"名称": "element", "说明": "note"}],
+                }
+            ],
+        }
+    )
+    created = await gateway.create(parsed, account_name=None)
+    graph = await LegacySqlAlchemyInputReader(sessions).read_topic_graph(
+        execution_id=0,
+        topic_build_id=created["topic_build_id"],
+        topic_id=created["topic_id"],
+    )
+    assert graph["execution"] == {"id": 0, "status": "upload"}
+    assert graph["topic"]["result"] == "uploaded topic"
+    assert graph["points"][0]["item_ids"] == [graph["composition_items"][0]["id"]]