服务：接入鉴权 API 和生产运行组装

挂载旧创建、上传、停止接口以及 Mission、Task、Artifact、Trace、V2 和 WebSocket 只读观测接口。

所有查询从已授权 build 反查 Root；生产启动要求显式 Principal、读写数据库分离、真实 Decode 服务和单 worker 运行。

script_build_host/src/script_build_host/__init__.py

@@ -0,0 +1,17 @@
+"""Agentic script-build Host, phase one."""
+
+from .composition import HostComposition, HostDependencies, compose_host
+from .production import (
+    ProductionHost,
+    ProductionRuntimeDependencies,
+    compose_production_host,
+)
+
+__all__ = [
+    "HostComposition",
+    "HostDependencies",
+    "ProductionHost",
+    "ProductionRuntimeDependencies",
+    "compose_host",
+    "compose_production_host",
+]

script_build_host/src/script_build_host/api/__init__.py

@@ -0,0 +1,7 @@
+"""Authenticated HTTP boundary for script-build phase one."""
+
+from .app import create_app
+from .dependencies import ApiSecurity
+from .routes import UploadedTopicGateway, create_script_build_router
+
+__all__ = ["ApiSecurity", "UploadedTopicGateway", "create_app", "create_script_build_router"]

script_build_host/src/script_build_host/api/app.py

@@ -0,0 +1,43 @@
+"""FastAPI application factory; no anonymous production fallback exists."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from script_build_host.domain.errors import ScriptBuildError
+
+from .routes import create_script_build_router
+
+
+def create_app(**router_dependencies: Any) -> FastAPI:
+    security = router_dependencies.get("security")
+    if security is None:
+        raise RuntimeError("ApiSecurity is required; anonymous Host mode is not supported")
+    app = FastAPI(title="Script Build Host", version="0.1.0")
+
+    @app.exception_handler(ScriptBuildError)
+    async def domain_error(_: Request, exc: ScriptBuildError) -> JSONResponse:
+        status = 404 if exc.code.endswith("NOT_FOUND") else 409
+        if exc.code in {
+            "INPUT_RELATION_MISMATCH",
+            "PROTOCOL_VIOLATION",
+            "UNSAFE_OUTBOUND_TARGET",
+        }:
+            status = 400
+        return JSONResponse(
+            status_code=status,
+            content={
+                "success": False,
+                "error_code": exc.code,
+                "message": exc.summary,
+            },
+        )
+
+    app.include_router(create_script_build_router(**router_dependencies))
+    return app
+
+
+__all__ = ["create_app"]

script_build_host/src/script_build_host/api/dependencies.py

@@ -0,0 +1,82 @@
+"""Shared authentication and build-level authorization dependencies."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from fastapi import HTTPException, WebSocket
+
+from script_build_host.domain.ports import BuildAuthorizer, PrincipalProvider
+from script_build_host.domain.records import Principal
+
+
+@dataclass(frozen=True)
+class ApiSecurity:
+    principals: PrincipalProvider
+    authorizer: BuildAuthorizer
+    hide_forbidden: bool = True
+    websocket_allowed_origins: tuple[str, ...] = ()
+
+    async def principal(self, request_context: object | None = None) -> Principal:
+        try:
+            principal = await self.principals.current(request_context)
+        except Exception as exc:
+            raise HTTPException(
+                status_code=401,
+                detail={"error_code": "AUTHENTICATION_REQUIRED"},
+            ) from exc
+        if not principal.subject.strip():
+            raise HTTPException(
+                status_code=401,
+                detail={"error_code": "AUTHENTICATION_REQUIRED"},
+            )
+        return principal
+
+    def require_websocket_origin(self, websocket: WebSocket) -> None:
+        origin = websocket.headers.get("origin")
+        allowed = {item.rstrip("/") for item in self.websocket_allowed_origins}
+        if not origin or origin.rstrip("/") not in allowed:
+            raise HTTPException(
+                status_code=403,
+                detail={"error_code": "WEBSOCKET_ORIGIN_FORBIDDEN"},
+            )
+
+    async def authorize(self, script_build_id: int, principal: Principal) -> None:
+        try:
+            await self.authorizer.require_access(principal, script_build_id)
+        except Exception as exc:
+            raise HTTPException(
+                status_code=404 if self.hide_forbidden else 403,
+                detail={"error_code": "BUILD_NOT_FOUND" if self.hide_forbidden else "FORBIDDEN"},
+            ) from exc
+
+    async def authorize_source(
+        self,
+        principal: Principal,
+        *,
+        execution_id: int,
+        topic_build_id: int,
+        topic_id: int,
+    ) -> None:
+        try:
+            await self.authorizer.require_source_access(
+                principal,
+                execution_id=execution_id,
+                topic_build_id=topic_build_id,
+                topic_id=topic_id,
+            )
+        except Exception as exc:
+            raise HTTPException(
+                status_code=404 if self.hide_forbidden else 403,
+                detail={"error_code": "SOURCE_NOT_FOUND" if self.hide_forbidden else "FORBIDDEN"},
+            ) from exc
+
+
+def problem(status: int, code: str, message: str) -> HTTPException:
+    return HTTPException(
+        status_code=status,
+        detail={"success": False, "error_code": code, "message": message},
+    )
+
+
+__all__ = ["ApiSecurity", "problem"]

script_build_host/src/script_build_host/api/routes.py

@@ -0,0 +1,517 @@
+"""Legacy start/stop endpoints and authenticated read-only observation API."""
+
+import asyncio
+import json
+import re
+from collections.abc import Mapping
+from dataclasses import asdict
+from typing import Annotated, Any, Protocol, cast
+from urllib.parse import urlsplit, urlunsplit
+
+from fastapi import APIRouter, Depends, Query, Request, WebSocket, WebSocketDisconnect
+
+from script_build_host.agents.prompts import phase_one_prompt_manifest
+from script_build_host.application.mission_service import (
+    ScriptMissionService,
+    StartScriptBuildCommand,
+)
+from script_build_host.domain.ports import (
+    MissionBindingRepository,
+    PublicationRepository,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.domain.records import Principal, PublicationType
+from script_build_host.infrastructure.outbound import OutboundPolicy
+from script_build_host.infrastructure.redaction import redact
+
+from .dependencies import ApiSecurity, problem
+from .schemas import (
+    ParseTopicJsonRequest,
+    ScriptBuildFromTopicJsonRequest,
+    ScriptBuildRequest,
+    ScriptBuildStartResponse,
+)
+
+
+class UploadedTopicGateway(Protocol):
+    async def parse(self, value: dict[str, Any] | str) -> dict[str, Any]: ...
+
+    async def create(
+        self, parsed: dict[str, Any], *, account_name: str | None
+    ) -> dict[str, Any]: ...
+
+
+def create_script_build_router(
+    *,
+    mission_service: ScriptMissionService,
+    security: ApiSecurity,
+    bindings: MissionBindingRepository,
+    business_artifacts: ScriptBusinessArtifactRepository,
+    publications: PublicationRepository,
+    coordinator: Any,
+    trace_store: Any,
+    uploaded_topics: UploadedTopicGateway | None = None,
+    outbound_policy: OutboundPolicy | None = None,
+    default_model_manifest: Mapping[str, object] | None = None,
+    default_datasource_manifest: Mapping[str, object] | None = None,
+) -> APIRouter:
+    router = APIRouter(tags=["script-build"])
+
+    async def principal(request: Request) -> Principal:
+        return await security.principal(request)
+
+    PrincipalDependency = Annotated[Principal, Depends(principal)]
+
+    async def scope(script_build_id: int, actor: Principal) -> Any:
+        await security.authorize(script_build_id, actor)
+        return await bindings.get_by_build(script_build_id)
+
+    @router.post("/api/pattern/script_builds", response_model=ScriptBuildStartResponse)
+    async def start_script_build(
+        body: ScriptBuildRequest,
+        actor: PrincipalDependency,
+    ) -> ScriptBuildStartResponse:
+        if body.agent_type != "AigcAgent":
+            raise problem(400, "UNSUPPORTED_AGENT_TYPE", "script build supports AigcAgent only")
+        await security.authorize_source(
+            actor,
+            execution_id=body.execution_id,
+            topic_build_id=body.topic_build_id,
+            topic_id=body.topic_id,
+        )
+        command = await _start_command(
+            body,
+            actor,
+            outbound_policy,
+            default_model_manifest=default_model_manifest,
+            default_datasource_manifest=default_datasource_manifest,
+        )
+        result = await mission_service.start(command)
+        return ScriptBuildStartResponse(
+            script_build_id=result.script_build_id,
+            status=result.status.value,
+            root_trace_id=result.root_trace_id,
+        )
+
+    @router.post("/api/pattern/script_builds/parse_topic_json")
+    async def parse_topic_json(
+        body: ParseTopicJsonRequest,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        if uploaded_topics is None:
+            raise problem(
+                503, "UPLOAD_TOPIC_NOT_CONFIGURED", "uploaded topic gateway is unavailable"
+            )
+        await security.authorize_source(
+            actor,
+            execution_id=0,
+            topic_build_id=0,
+            topic_id=0,
+        )
+        parsed = await uploaded_topics.parse(body.json_data)
+        points = list(parsed.get("points", []))
+        stats: dict[str, int] = {}
+        for point in points:
+            key = str(point.get("point_type", ""))
+            stats[key] = stats.get(key, 0) + 1
+        return {
+            "success": True,
+            "account_name": parsed.get("account_name"),
+            "topic_fusion": parsed.get("topic_fusion"),
+            "point_count": len(points),
+            "item_count": parsed.get("item_count", 0),
+            "point_type_stats": stats,
+            "points": points,
+        }
+
+    @router.post("/api/pattern/script_builds/from_topic_json")
+    async def start_from_topic_json(
+        body: ScriptBuildFromTopicJsonRequest,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        if uploaded_topics is None:
+            raise problem(
+                503, "UPLOAD_TOPIC_NOT_CONFIGURED", "uploaded topic gateway is unavailable"
+            )
+        await security.authorize_source(
+            actor,
+            execution_id=0,
+            topic_build_id=0,
+            topic_id=0,
+        )
+        parsed = await uploaded_topics.parse(body.json_data)
+        created = await uploaded_topics.create(parsed, account_name=body.account_name)
+        request = ScriptBuildRequest(
+            execution_id=0,
+            topic_build_id=int(created["topic_build_id"]),
+            topic_id=int(created["topic_id"]),
+            agent_config=body.agent_config,
+            data_source_url=body.data_source_url,
+            strategies_always_on=body.strategies_always_on,
+            strategies_on_demand=body.strategies_on_demand,
+        )
+        await security.authorize_source(
+            actor,
+            execution_id=request.execution_id,
+            topic_build_id=request.topic_build_id,
+            topic_id=request.topic_id,
+        )
+        result = await mission_service.start(
+            await _start_command(
+                request,
+                actor,
+                outbound_policy,
+                default_model_manifest=default_model_manifest,
+                default_datasource_manifest=default_datasource_manifest,
+            )
+        )
+        return {
+            "success": True,
+            "message": "脚本构建任务已提交",
+            "script_build_id": result.script_build_id,
+            "topic_id": created["topic_id"],
+            "topic_build_id": created["topic_build_id"],
+            "account_name": created.get("account_name"),
+            "item_count": created.get("item_count", 0),
+            "point_count": created.get("point_count", 0),
+            "status": result.status.value,
+            "root_trace_id": result.root_trace_id,
+        }
+
+    @router.post("/api/pattern/script_builds/{script_build_id}/stop")
+    async def stop_script_build(
+        script_build_id: int,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        result = await mission_service.stop(script_build_id, actor)
+        return {"success": True, **asdict(result), "status": result.status.value}
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/mission")
+    async def mission_snapshot(
+        script_build_id: int,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        ledger = await coordinator.task_store.load(binding.root_trace_id)
+        return cast(dict[str, Any], ledger.to_dict())
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/mission/events")
+    async def mission_events(
+        script_build_id: int,
+        actor: PrincipalDependency,
+        after: str | None = None,
+        limit: int = Query(100, ge=1, le=500),
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        page = await coordinator.task_store.list_events(
+            binding.root_trace_id, cursor=after, limit=limit
+        )
+        return {
+            "events": [item.to_dict() for item in page.events],
+            "next_cursor": page.next_cursor,
+        }
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/tasks/{task_id}")
+    async def task_detail(
+        script_build_id: int,
+        task_id: str,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        ledger = await coordinator.task_store.load(binding.root_trace_id)
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise problem(404, "TASK_NOT_FOUND", "task does not exist in this build")
+        return task.to_dict() if hasattr(task, "to_dict") else asdict(task)
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/artifacts/{artifact_version_id}")
+    async def artifact_detail(
+        script_build_id: int,
+        artifact_version_id: int,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        await scope(script_build_id, actor)
+        value = await business_artifacts.get_by_id(
+            artifact_version_id, script_build_id=script_build_id
+        )
+        return _json_values(asdict(value))
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/publication")
+    async def publication_detail(
+        script_build_id: int,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        await scope(script_build_id, actor)
+        value = await publications.get_by_build(
+            script_build_id, publication_type=PublicationType.DIRECTION
+        )
+        if value is None:
+            raise problem(404, "PUBLICATION_NOT_FOUND", "publication does not exist")
+        return _json_values(asdict(value))
+
+    # Authenticated, build-scoped subset of the framework V2 observation API.
+    @router.get("/api/v2/script-builds/{script_build_id}/roots/{root_trace_id}/snapshot")
+    async def v2_snapshot(
+        script_build_id: int,
+        root_trace_id: str,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        _same_root(binding.root_trace_id, root_trace_id)
+        return cast(dict[str, Any], (await coordinator.task_store.load(root_trace_id)).to_dict())
+
+    @router.get("/api/v2/script-builds/{script_build_id}/roots/{root_trace_id}/completion")
+    async def v2_completion(
+        script_build_id: int,
+        root_trace_id: str,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        _same_root(binding.root_trace_id, root_trace_id)
+        return cast(dict[str, Any], await coordinator.root_completion(root_trace_id))
+
+    @router.get("/api/v2/script-builds/{script_build_id}/roots/{root_trace_id}/events")
+    async def v2_events(
+        script_build_id: int,
+        root_trace_id: str,
+        actor: PrincipalDependency,
+        after: str | None = None,
+        limit: int = Query(100, ge=1, le=500),
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        _same_root(binding.root_trace_id, root_trace_id)
+        page = await coordinator.task_store.list_events(
+            root_trace_id,
+            cursor=after,
+            limit=limit,
+        )
+        return {
+            "events": [item.to_dict() for item in page.events],
+            "next_cursor": page.next_cursor,
+        }
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/traces")
+    async def list_traces(
+        script_build_id: int,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        traces = await trace_store.list_traces(root_trace_id=binding.root_trace_id, limit=100)
+        root = await trace_store.get_trace(binding.root_trace_id)
+        values = ([root] if root is not None else []) + list(traces)
+        deduplicated = {item.trace_id: item.to_dict() for item in values}
+        return {"traces": list(deduplicated.values())}
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/traces/{trace_id}")
+    async def trace_detail(
+        script_build_id: int,
+        trace_id: str,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        trace = await _owned_trace(trace_store, binding.root_trace_id, trace_id)
+        tree = await trace_store.get_goal_tree(trace_id)
+        return {
+            "trace": trace.to_dict(),
+            "goal_tree": tree.to_dict() if tree else None,
+        }
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/traces/{trace_id}/messages")
+    async def trace_messages(
+        script_build_id: int,
+        trace_id: str,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        binding = await scope(script_build_id, actor)
+        await _owned_trace(trace_store, binding.root_trace_id, trace_id)
+        messages = await trace_store.get_trace_messages(trace_id)
+        return {"messages": [item.to_dict() for item in messages]}
+
+    @router.websocket("/api/pattern/script_builds/{script_build_id}/traces/{trace_id}/watch")
+    async def watch_trace(
+        websocket: WebSocket,
+        script_build_id: int,
+        trace_id: str,
+    ) -> None:
+        try:
+            security.require_websocket_origin(websocket)
+            actor = await security.principal(websocket)
+            binding = await scope(script_build_id, actor)
+            await _owned_trace(trace_store, binding.root_trace_id, trace_id)
+        except Exception:
+            await websocket.close(code=4404)
+            return
+        await websocket.accept()
+        since = 0
+        try:
+            while True:
+                events = await trace_store.get_events(trace_id, since)
+                for event in events:
+                    await websocket.send_json(event)
+                    if isinstance(event.get("event_id"), int):
+                        since = max(since, event["event_id"])
+                try:
+                    value = await asyncio.wait_for(
+                        websocket.receive_text(),
+                        timeout=0.5,
+                    )
+                    if value == "ping":
+                        await websocket.send_json({"event": "pong"})
+                except TimeoutError:
+                    continue
+                except WebSocketDisconnect:
+                    return
+        except WebSocketDisconnect:
+            return
+
+    return router
+
+
+async def _start_command(
+    body: ScriptBuildRequest,
+    actor: Principal,
+    outbound_policy: OutboundPolicy | None,
+    *,
+    default_model_manifest: Mapping[str, object] | None,
+    default_datasource_manifest: Mapping[str, object] | None,
+) -> StartScriptBuildCommand:
+    endpoint = body.data_source_url
+    sanitized_datasources = redact(dict(default_datasource_manifest or {}))
+    if not isinstance(sanitized_datasources, dict):
+        raise problem(500, "INVALID_HOST_CONFIG", "datasource manifest must be an object")
+    datasource_manifest: dict[str, Any] = sanitized_datasources
+    safe_endpoint: str | None = None
+    if endpoint:
+        parsed = urlsplit(endpoint)
+        if (
+            parsed.scheme != "https"
+            or not parsed.hostname
+            or parsed.username
+            or parsed.password
+            or parsed.query
+            or parsed.fragment
+        ):
+            raise problem(
+                400, "UNSAFE_DATA_SOURCE_URL", "data_source_url must be credential-free HTTPS"
+            )
+        if outbound_policy is None:
+            raise problem(
+                503,
+                "OUTBOUND_POLICY_NOT_CONFIGURED",
+                "data_source_url requires an outbound policy",
+            )
+        validated = await outbound_policy.validate_url(endpoint)
+        safe_endpoint = urlunsplit(urlsplit(validated)._replace(query="", fragment=""))
+        datasource_manifest = {
+            **datasource_manifest,
+            "data_source_host": parsed.hostname.rstrip(".").lower(),
+            "scheme": "https",
+        }
+    raw_config = dict(body.agent_config or {})
+    sanitized_config = redact(raw_config)
+    if not isinstance(sanitized_config, dict):
+        raise problem(400, "INVALID_AGENT_CONFIG", "agent_config must be an object")
+    if isinstance(raw_config.get("temperature"), bool) or isinstance(
+        raw_config.get("max_iterations"), bool
+    ):
+        raise problem(400, "INVALID_AGENT_CONFIG", "model run configuration is invalid")
+    default_presets = (default_model_manifest or {}).get(
+        "presets",
+        default_model_manifest or {},
+    )
+    planner_default = (
+        default_presets.get("script_planner", {}) if isinstance(default_presets, Mapping) else {}
+    )
+    if not isinstance(planner_default, Mapping):
+        planner_default = {}
+    try:
+        model = str(raw_config.get("model_name", planner_default.get("model", "gpt-4o")))
+        temperature = float(raw_config.get("temperature", 0.3))
+        max_iterations = int(raw_config.get("max_iterations", 120))
+        presets: dict[str, dict[str, object]] = {}
+        preset_keys = {
+            "script_planner": "model_name",
+            "script_direction_worker": "model_name",
+            "script_pattern_retrieval_worker": "model_name_means_pattern_relation",
+            "script_decode_retrieval_worker": "model_name_means_case",
+            "script_external_retrieval_worker": "model_name_means_data",
+            "script_knowledge_retrieval_worker": "model_name_means_knowledge",
+            "script_retrieval_validator": "model_name_script_evaluator",
+            "script_candidate_validator": "model_name_script_evaluator",
+        }
+        for name, key in preset_keys.items():
+            preset_default = (
+                default_presets.get(name, {}) if isinstance(default_presets, Mapping) else {}
+            )
+            if not isinstance(preset_default, Mapping):
+                preset_default = {}
+            presets[name] = {
+                "model": str(raw_config.get(key, preset_default.get("model", model))),
+                "temperature": float(
+                    cast(
+                        Any,
+                        raw_config.get(
+                            "temperature",
+                            preset_default.get(
+                                "temperature",
+                                0.0 if "validator" in name else temperature,
+                            ),
+                        ),
+                    )
+                ),
+                "max_iterations": int(
+                    cast(
+                        Any,
+                        raw_config.get(
+                            "max_iterations",
+                            preset_default.get("max_iterations", max_iterations),
+                        ),
+                    )
+                ),
+            }
+    except (TypeError, ValueError, OverflowError) as exc:
+        raise problem(400, "INVALID_AGENT_CONFIG", "model run configuration is invalid") from exc
+    model_pattern = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,199}$")
+    if any(
+        not model_pattern.fullmatch(str(value["model"]))
+        or not 0 <= float(cast(Any, value["temperature"])) <= 2
+        or not 1 <= int(cast(Any, value["max_iterations"])) <= 1000
+        for value in presets.values()
+    ):
+        raise problem(400, "INVALID_AGENT_CONFIG", "model names contain unsafe characters")
+    return StartScriptBuildCommand(
+        execution_id=body.execution_id,
+        topic_build_id=body.topic_build_id,
+        topic_id=body.topic_id,
+        principal=actor,
+        agent_type=body.agent_type,
+        agent_config=sanitized_config,
+        data_source_url=safe_endpoint,
+        strategies_always_on=tuple(body.strategies_always_on or ()),
+        strategies_on_demand=tuple(body.strategies_on_demand or ()),
+        runtime_prompt_manifest=phase_one_prompt_manifest(),
+        datasource_manifest=datasource_manifest,
+        model_manifest={"presets": presets},
+    )
+
+
+def _same_root(expected: str, supplied: str) -> None:
+    if expected != supplied:
+        raise problem(404, "ROOT_NOT_FOUND", "root does not exist in this build")
+
+
+async def _owned_trace(trace_store: Any, root_trace_id: str, trace_id: str) -> Any:
+    trace = await trace_store.get_trace(trace_id)
+    if trace is None:
+        raise problem(404, "TRACE_NOT_FOUND", "trace does not exist in this build")
+    trace_root = (trace.context or {}).get("root_trace_id")
+    if trace_id != root_trace_id and trace_root != root_trace_id:
+        raise problem(404, "TRACE_NOT_FOUND", "trace does not exist in this build")
+    return trace
+
+
+def _json_values(value: dict[str, Any]) -> dict[str, Any]:
+    return cast(dict[str, Any], json.loads(json.dumps(value, ensure_ascii=False, default=str)))
+
+
+__all__ = ["UploadedTopicGateway", "create_script_build_router"]

script_build_host/src/script_build_host/api/schemas.py

@@ -0,0 +1,62 @@
+"""Legacy-compatible request shapes and stable Host response models."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+StrategySelector = int | str | dict[str, Any]
+
+
+class ScriptBuildRequest(BaseModel):
+    execution_id: int
+    topic_build_id: int
+    topic_id: int
+    agent_type: str = "AigcAgent"
+    agent_config: dict[str, Any] | None = None
+    data_source_url: str | None = None
+    strategies_always_on: list[StrategySelector] | None = None
+    strategies_on_demand: list[StrategySelector] | None = None
+
+
+class ParseTopicJsonRequest(BaseModel):
+    json_data: dict[str, Any] | str
+
+
+class ScriptBuildFromTopicJsonRequest(BaseModel):
+    json_data: dict[str, Any] | str
+    account_name: str | None = None
+    agent_config: dict[str, Any] | None = None
+    data_source_url: str | None = None
+    strategies_always_on: list[StrategySelector] | None = None
+    strategies_on_demand: list[StrategySelector] | None = None
+
+
+class ScriptBuildStartResponse(BaseModel):
+    success: bool = True
+    message: str = "脚本构建任务已提交"
+    script_build_id: int
+    status: str
+    root_trace_id: str
+
+
+class ProblemResponse(BaseModel):
+    success: bool = False
+    error_code: str
+    message: str
+
+
+class EventQuery(BaseModel):
+    after: str | None = None
+    limit: int = Field(default=100, ge=1, le=500)
+
+
+__all__ = [
+    "ParseTopicJsonRequest",
+    "ProblemResponse",
+    "ScriptBuildFromTopicJsonRequest",
+    "ScriptBuildRequest",
+    "ScriptBuildStartResponse",
+    "StrategySelector",
+]

script_build_host/src/script_build_host/cli.py

@@ -0,0 +1,53 @@
+"""Single-worker production entry point."""
+
+from __future__ import annotations
+
+import importlib
+from collections.abc import Callable
+from typing import Any, cast
+
+from fastapi import FastAPI
+
+from script_build_host.infrastructure.config import ScriptBuildSettings
+from script_build_host.production import (
+    ProductionRuntimeDependencies,
+    compose_production_host,
+)
+
+
+def create_cli_app() -> FastAPI:
+    settings = ScriptBuildSettings()  # type: ignore[call-arg]
+    if not settings.runtime_factory:
+        raise RuntimeError("SCRIPT_BUILD_RUNTIME_FACTORY=module:callable is required")
+    factory = _load_factory(settings.runtime_factory)
+    runtime = factory(settings)
+    if not isinstance(runtime, ProductionRuntimeDependencies):
+        raise TypeError("runtime factory must return ProductionRuntimeDependencies")
+    return cast(FastAPI, compose_production_host(settings, runtime).composition.app)
+
+
+def _load_factory(value: str) -> Callable[[ScriptBuildSettings], Any]:
+    module_name, separator, attribute = value.partition(":")
+    if not separator or not module_name or not attribute:
+        raise RuntimeError("SCRIPT_BUILD_RUNTIME_FACTORY must use module:callable syntax")
+    module = importlib.import_module(module_name)
+    factory = getattr(module, attribute, None)
+    if not callable(factory):
+        raise RuntimeError("SCRIPT_BUILD_RUNTIME_FACTORY does not resolve to a callable")
+    return cast(Callable[[ScriptBuildSettings], Any], factory)
+
+
+def main() -> None:
+    import uvicorn
+
+    settings = ScriptBuildSettings()  # type: ignore[call-arg]
+    uvicorn.run(
+        "script_build_host.cli:create_cli_app",
+        factory=True,
+        host=settings.bind_host,
+        port=settings.bind_port,
+        workers=1,
+    )
+
+
+__all__ = ["create_cli_app", "main"]

script_build_host/src/script_build_host/composition.py

@@ -0,0 +1,163 @@
+"""Explicit phase-one composition over generic Agent framework ports."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from typing import Any
+
+from agent import wire_orchestration
+
+from script_build_host.agents import (
+    ScriptBuildArtifactEvidenceReader,
+    ScriptBuildDeterministicValidator,
+    ScriptBuildEvidenceProvider,
+    ScriptBuildValidationPolicy,
+    ScriptRoleRunConfigResolver,
+    register_script_presets,
+)
+from script_build_host.agents.model_resolver import SnapshotModelManifestSource
+from script_build_host.api import ApiSecurity, UploadedTopicGateway, create_app
+from script_build_host.application.input_snapshot_service import ScriptInputSnapshotService
+from script_build_host.application.mission_factory import ScriptMissionFactory
+from script_build_host.application.mission_service import (
+    BuildTransitionGate,
+    DirectionReconciler,
+    ScriptMissionService,
+)
+from script_build_host.domain.ports import (
+    BuildAuthorizer,
+    InputSnapshotRepository,
+    LegacyBuildStateRepository,
+    MissionBindingRepository,
+    PrincipalProvider,
+    PublicationRepository,
+    RuntimeManifestProvider,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.infrastructure.outbound import OutboundPolicy
+from script_build_host.tools import LegacyScriptToolGateway, register_script_tools
+from script_build_host.tools.gateway import ImageAdapter, RetrievalAdapter
+
+
+@dataclass(frozen=True)
+class HostDependencies:
+    runner: Any
+    task_store: Any
+    framework_artifact_store: Any
+    input_snapshot_service: ScriptInputSnapshotService
+    input_snapshots: InputSnapshotRepository
+    bindings: MissionBindingRepository
+    business_artifacts: ScriptBusinessArtifactRepository
+    publications: PublicationRepository
+    legacy_state: LegacyBuildStateRepository
+    principal_provider: PrincipalProvider
+    build_authorizer: BuildAuthorizer
+    retrieval_adapters: Mapping[str, RetrievalAdapter] = field(default_factory=dict)
+    image_adapter: ImageAdapter | None = None
+    uploaded_topics: UploadedTopicGateway | None = None
+    orchestration_config: Any = None
+    event_sink: Any = None
+    hide_forbidden: bool = True
+    runtime_manifest_provider: RuntimeManifestProvider | None = None
+    outbound_policy: OutboundPolicy | None = None
+    websocket_allowed_origins: tuple[str, ...] = ()
+    default_model_manifest: Mapping[str, object] = field(default_factory=dict)
+    default_datasource_manifest: Mapping[str, object] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class HostComposition:
+    app: Any
+    coordinator: Any
+    mission_service: ScriptMissionService
+    tool_gateway: LegacyScriptToolGateway
+
+
+def compose_host(dependencies: HostDependencies) -> HostComposition:
+    """Wire one Host without reconstructing Coordinator or LocalAgentExecutor."""
+
+    register_script_presets()
+    model_resolver = ScriptRoleRunConfigResolver(
+        {},
+        manifest_source=SnapshotModelManifestSource(
+            dependencies.bindings, dependencies.input_snapshots
+        ),
+    )
+    evidence_provider = ScriptBuildEvidenceProvider(
+        dependencies.framework_artifact_store,
+        reader=ScriptBuildArtifactEvidenceReader(
+            dependencies.business_artifacts,
+            dependencies.bindings,
+        ),
+    )
+    coordinator = wire_orchestration(
+        dependencies.runner,
+        dependencies.task_store,
+        dependencies.framework_artifact_store,
+        dependencies.orchestration_config,
+        dependencies.event_sink,
+        validation_policy=ScriptBuildValidationPolicy(),
+        deterministic_validator=ScriptBuildDeterministicValidator(),
+        evidence_provider=evidence_provider,
+        role_run_config_resolver=model_resolver,
+    )
+    transition_gate = BuildTransitionGate()
+    direction_reconciler = DirectionReconciler(
+        coordinator=coordinator,
+        bindings=dependencies.bindings,
+        artifacts=dependencies.business_artifacts,
+        publications=dependencies.publications,
+        legacy_state=dependencies.legacy_state,
+        transition_gate=transition_gate,
+    )
+    mission_service = ScriptMissionService(
+        runner=dependencies.runner,
+        coordinator=coordinator,
+        factory=ScriptMissionFactory(),
+        input_snapshots=dependencies.input_snapshot_service,
+        bindings=dependencies.bindings,
+        legacy_state=dependencies.legacy_state,
+        authorizer=dependencies.build_authorizer,
+        direction_reconciler=direction_reconciler,
+        transition_gate=transition_gate,
+        runtime_manifest_provider=dependencies.runtime_manifest_provider,
+    )
+    gateway = LegacyScriptToolGateway(
+        bindings=dependencies.bindings,
+        snapshots=dependencies.input_snapshots,
+        artifacts=dependencies.business_artifacts,
+        coordinator=coordinator,
+        retrieval_adapters=dependencies.retrieval_adapters,
+        image_adapter=dependencies.image_adapter,
+        dispatch_guard=mission_service,
+    )
+    register_script_tools(dependencies.runner.tools, gateway)
+    security = ApiSecurity(
+        dependencies.principal_provider,
+        dependencies.build_authorizer,
+        hide_forbidden=dependencies.hide_forbidden,
+        websocket_allowed_origins=dependencies.websocket_allowed_origins,
+    )
+    app = create_app(
+        mission_service=mission_service,
+        security=security,
+        bindings=dependencies.bindings,
+        business_artifacts=dependencies.business_artifacts,
+        publications=dependencies.publications,
+        coordinator=coordinator,
+        trace_store=dependencies.runner.trace_store,
+        uploaded_topics=dependencies.uploaded_topics,
+        outbound_policy=dependencies.outbound_policy,
+        default_model_manifest=dependencies.default_model_manifest,
+        default_datasource_manifest=dependencies.default_datasource_manifest,
+    )
+    return HostComposition(
+        app=app,
+        coordinator=coordinator,
+        mission_service=mission_service,
+        tool_gateway=gateway,
+    )
+
+
+__all__ = ["HostComposition", "HostDependencies", "compose_host"]

script_build_host/src/script_build_host/production.py

@@ -0,0 +1,220 @@
+"""Production composition root for the phase-one Script Build Host."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import httpx
+from agent import FileSystemArtifactStore, FileSystemTaskStore
+
+from script_build_host.adapters import (
+    DatabaseFirstPromptSource,
+    ExternalRetrievalAdapter,
+    FileDecodeRetrievalAdapter,
+    FileKnowledgeRetrievalAdapter,
+    FilePersonaSource,
+    HttpDecodeRetrievalAdapter,
+    PatternRetrievalAdapter,
+    SafeHttpClient,
+    SafeImageAdapter,
+    SqlAlchemyStrategySource,
+    SqlUploadedTopicGateway,
+)
+from script_build_host.application import ScriptInputSnapshotService
+from script_build_host.composition import HostComposition, HostDependencies, compose_host
+from script_build_host.domain.ports import BuildAuthorizer, PrincipalProvider
+from script_build_host.infrastructure.config import ScriptBuildSettings
+from script_build_host.infrastructure.db import DatabaseSessions, create_database_sessions
+from script_build_host.infrastructure.manifests import SettingsRuntimeManifestProvider
+from script_build_host.infrastructure.outbound import OutboundPolicy
+from script_build_host.infrastructure.raw_artifacts import FileRawArtifactStore
+from script_build_host.repositories import (
+    LegacySqlAlchemyInputReader,
+    SqlAlchemyInputSnapshotRepository,
+    SqlAlchemyLegacyBuildStateRepository,
+    SqlAlchemyMissionBindingRepository,
+    SqlAlchemyPublicationRepository,
+    SqlAlchemyScriptBusinessArtifactRepository,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class ProductionRuntimeDependencies:
+    """Deployment-owned capabilities that the Host must never fake."""
+
+    runner: Any
+    principal_provider: PrincipalProvider
+    build_authorizer: BuildAuthorizer
+    orchestration_config: Any = None
+    event_sink: Any = None
+    http_transport: httpx.AsyncBaseTransport | None = None
+
+
+@dataclass(slots=True)
+class ProductionHost:
+    composition: HostComposition
+    database: DatabaseSessions
+    http_client: httpx.AsyncClient
+    runtime_manifest: SettingsRuntimeManifestProvider
+    _closed: bool = False
+
+    async def validate_startup(self) -> None:
+        """Resolve configured endpoints and sources before serving traffic."""
+
+        await self.runtime_manifest.load()
+
+    async def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        await self.http_client.aclose()
+        await self.database.dispose()
+
+
+def compose_production_host(
+    settings: ScriptBuildSettings,
+    runtime: ProductionRuntimeDependencies,
+) -> ProductionHost:
+    """Build the production Host from explicit framework/auth dependencies."""
+
+    if getattr(runtime.runner, "trace_store", None) is None:
+        raise RuntimeError("production AgentRunner must provide a durable trace store")
+
+    decode_path: Path | None = None
+    if not settings.decode_endpoint:
+        if settings.environment.lower() != "test":
+            raise RuntimeError("SCRIPT_BUILD_DECODE_ENDPOINT is required in production")
+        decode_path = _one_json_source(settings.decode_index_root, "decode")
+    knowledge_path = _one_json_source(settings.knowledge_root, "knowledge")
+
+    database = create_database_sessions(settings)
+    outbound_policy = OutboundPolicy(
+        frozenset(settings.outbound_allowed_hosts),
+        frozenset(settings.outbound_allowed_ports),
+    )
+    http_client = httpx.AsyncClient(
+        transport=runtime.http_transport,
+        trust_env=False,
+    )
+    safe_http = SafeHttpClient(
+        http_client,
+        outbound_policy,
+        timeout_seconds=settings.request_timeout_seconds,
+        max_response_bytes=settings.max_response_bytes,
+    )
+
+    retrieval_adapters: dict[str, Any] = {}
+    if settings.pattern_endpoint:
+        retrieval_adapters["pattern"] = PatternRetrievalAdapter(
+            settings.pattern_endpoint,
+            safe_http,
+            poll_interval_seconds=settings.poll_interval_seconds,
+            poll_timeout_seconds=settings.request_timeout_seconds,
+        )
+    if settings.decode_endpoint:
+        retrieval_adapters["decode"] = HttpDecodeRetrievalAdapter(
+            settings.decode_endpoint,
+            safe_http,
+        )
+    elif decode_path is not None:
+        retrieval_adapters["decode"] = FileDecodeRetrievalAdapter(decode_path)
+    else:
+        raise AssertionError("decode configuration was validated before resource creation")
+    if settings.external_endpoint:
+        retrieval_adapters["external"] = ExternalRetrievalAdapter(
+            settings.external_endpoint,
+            safe_http,
+        )
+    retrieval_adapters["knowledge"] = FileKnowledgeRetrievalAdapter(knowledge_path)
+
+    snapshots = SqlAlchemyInputSnapshotRepository(database.write)
+    bindings = SqlAlchemyMissionBindingRepository(database.write)
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(database.write)
+    publications = SqlAlchemyPublicationRepository(database.write)
+    legacy_state = SqlAlchemyLegacyBuildStateRepository(database.write)
+    input_service = ScriptInputSnapshotService(
+        legacy_input=LegacySqlAlchemyInputReader(database.read),
+        persona_source=FilePersonaSource(
+            persona_root=settings.persona_root,
+            section_pattern_root=settings.section_pattern_root,
+        ),
+        strategy_source=SqlAlchemyStrategySource(database.read),
+        prompt_source=DatabaseFirstPromptSource(
+            database.read,
+            fallback_root=settings.prompt_fallback_root,
+            file_only=settings.prompts_from_file,
+        ),
+        snapshots=snapshots,
+    )
+    runtime_manifest = SettingsRuntimeManifestProvider(
+        settings,
+        outbound_policy,
+        decode_path=decode_path,
+        knowledge_path=knowledge_path,
+    )
+    data_root = settings.agent_data_root.resolve()
+    composition = compose_host(
+        HostDependencies(
+            runner=runtime.runner,
+            task_store=FileSystemTaskStore(str(data_root / "task-ledger")),
+            framework_artifact_store=FileSystemArtifactStore(
+                str(data_root / "framework-artifacts")
+            ),
+            input_snapshot_service=input_service,
+            input_snapshots=snapshots,
+            bindings=bindings,
+            business_artifacts=artifacts,
+            publications=publications,
+            legacy_state=legacy_state,
+            principal_provider=runtime.principal_provider,
+            build_authorizer=runtime.build_authorizer,
+            retrieval_adapters=retrieval_adapters,
+            image_adapter=SafeImageAdapter(
+                safe_http,
+                FileRawArtifactStore(data_root),
+                max_images=settings.max_images,
+                max_image_bytes=settings.max_image_bytes,
+                max_total_image_bytes=settings.max_total_image_bytes,
+            ),
+            uploaded_topics=SqlUploadedTopicGateway(
+                database.write,
+                max_payload_bytes=settings.max_upload_bytes,
+                max_points=settings.max_upload_points,
+                max_items=settings.max_upload_items,
+            ),
+            orchestration_config=runtime.orchestration_config,
+            event_sink=runtime.event_sink,
+            runtime_manifest_provider=runtime_manifest,
+            outbound_policy=outbound_policy,
+            websocket_allowed_origins=settings.websocket_allowed_origins,
+            default_model_manifest=settings.model_manifest,
+            default_datasource_manifest=settings.datasource_manifest,
+        )
+    )
+    host = ProductionHost(composition, database, http_client, runtime_manifest)
+    composition.app.state.production_host = host
+    composition.app.router.add_event_handler("startup", host.validate_startup)
+    composition.app.router.add_event_handler("shutdown", host.close)
+    return host
+
+
+def _one_json_source(path: Path, label: str) -> Path:
+    resolved = path.resolve()
+    if resolved.is_file() and resolved.suffix.lower() == ".json":
+        return resolved
+    candidates = sorted(resolved.glob("*.json")) if resolved.is_dir() else []
+    if len(candidates) != 1:
+        raise RuntimeError(
+            f"SCRIPT_BUILD_{label.upper()} source must be one JSON file or a directory "
+            "containing exactly one JSON file"
+        )
+    return candidates[0]
+
+
+__all__ = [
+    "ProductionHost",
+    "ProductionRuntimeDependencies",
+    "compose_production_host",
+]

script_build_host/tests/fixtures/legacy_script_build_request.json

@@ -0,0 +1,15 @@
+{
+  "execution_id": 1,
+  "topic_build_id": 2,
+  "topic_id": 3,
+  "agent_type": "AigcAgent",
+  "agent_config": {
+    "model_name": "fake-planner",
+    "model_name_means_case": "fake-decode",
+    "model_name_script_evaluator": "fake-validator",
+    "temperature": 0.2,
+    "max_iterations": 50
+  },
+  "strategies_always_on": [13],
+  "strategies_on_demand": []
+}

script_build_host/tests/test_api_security.py

@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from types import SimpleNamespace
+from typing import ClassVar
+
+import httpx
+import pytest
+from fastapi.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+from script_build_host.api import ApiSecurity, create_app
+from script_build_host.application.mission_service import ScriptMissionStartResult
+from script_build_host.domain.records import BuildStatus, Principal
+from script_build_host.infrastructure.outbound import OutboundPolicy
+
+
+async def _public_resolver(_host: str, _port: int) -> tuple[str, ...]:
+    return ("93.184.216.34",)
+
+
+class _PrincipalProvider:
+    def __init__(self, principal: Principal | None) -> None:
+        self.value = principal
+
+    async def current(self, request_context: object | None = None) -> Principal:
+        assert request_context is not None
+        if self.value is None:
+            raise PermissionError("missing")
+        return self.value
+
+
+class _Authorizer:
+    async def require_source_access(self, principal: Principal, **_source: int) -> None:
+        if principal.subject != "owner":
+            raise PermissionError("forbidden")
+
+    async def require_access(self, principal: Principal, script_build_id: int) -> None:
+        if principal.subject != "owner" or script_build_id != 7:
+            raise PermissionError("forbidden")
+
+
+class _MissionService:
+    def __init__(self) -> None:
+        self.command = None
+
+    async def start(self, command: object) -> ScriptMissionStartResult:
+        self.command = command
+        return ScriptMissionStartResult(7, BuildStatus.RUNNING, "root-7", "11")
+
+    async def stop(self, script_build_id: int, principal: Principal) -> object:
+        del principal
+        return SimpleNamespace(
+            script_build_id=script_build_id,
+            status=BuildStatus.STOPPED,
+            stopped_operation_ids=(),
+        )
+
+
+class _Bindings:
+    async def get_by_build(self, script_build_id: int) -> object:
+        return SimpleNamespace(script_build_id=script_build_id, root_trace_id="root-7")
+
+
+class _TaskStore:
+    async def load(self, root_trace_id: str) -> object:
+        del root_trace_id
+        return SimpleNamespace(to_dict=lambda: {"root_trace_id": "root-7"})
+
+
+def _app(
+    principal: Principal | None,
+    *,
+    trace_store: object | None = None,
+    uploaded_topics: object | None = None,
+) -> tuple[object, _MissionService]:
+    mission = _MissionService()
+    app = create_app(
+        mission_service=mission,
+        security=ApiSecurity(
+            _PrincipalProvider(principal),
+            _Authorizer(),
+            websocket_allowed_origins=("https://ui.example",),
+        ),
+        bindings=_Bindings(),
+        business_artifacts=SimpleNamespace(),
+        publications=SimpleNamespace(),
+        coordinator=SimpleNamespace(task_store=_TaskStore()),
+        trace_store=trace_store or SimpleNamespace(),
+        uploaded_topics=uploaded_topics,
+        outbound_policy=OutboundPolicy(
+            frozenset({"data.example"}),
+            resolver=_public_resolver,
+        ),
+    )
+    return app, mission
+
+
+@pytest.mark.asyncio
+async def test_start_requires_auth_and_preserves_old_response_fields() -> None:
+    unauthenticated, _ = _app(None)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=unauthenticated), base_url="http://test"
+    ) as client:
+        response = await client.post(
+            "/api/pattern/script_builds",
+            json={"execution_id": 1, "topic_build_id": 2, "topic_id": 3},
+        )
+        assert response.status_code == 401
+
+    forbidden, _ = _app(Principal("intruder"))
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=forbidden), base_url="http://test"
+    ) as client:
+        response = await client.post(
+            "/api/pattern/script_builds",
+            json={"execution_id": 1, "topic_build_id": 2, "topic_id": 3},
+        )
+        assert response.status_code == 404
+
+    authenticated, mission = _app(Principal("owner"))
+    legacy_request = json.loads(
+        (Path(__file__).parent / "fixtures" / "legacy_script_build_request.json").read_text(
+            encoding="utf-8"
+        )
+    )
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=authenticated), base_url="http://test"
+    ) as client:
+        response = await client.post(
+            "/api/pattern/script_builds",
+            json=legacy_request,
+        )
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert body["success"] is True
+        assert body["script_build_id"] == 7
+        assert body["status"] == "running"
+        assert mission.command.strategies_always_on == (13,)
+        assert len(mission.command.runtime_prompt_manifest) == 8
+
+        secret_config = dict(legacy_request)
+        secret_config["agent_config"] = {
+            **legacy_request["agent_config"],
+            "api_key": "must-not-persist",
+        }
+        secret_response = await client.post(
+            "/api/pattern/script_builds",
+            json=secret_config,
+        )
+        assert secret_response.status_code == 200
+        assert "must-not-persist" not in str(mission.command.agent_config)
+
+        safe = await client.post(
+            "/api/pattern/script_builds",
+            json={
+                "execution_id": 1,
+                "topic_build_id": 2,
+                "topic_id": 3,
+                "data_source_url": "https://data.example/input",
+            },
+        )
+        assert safe.status_code == 200
+        assert mission.command.data_source_url == "https://data.example/input"
+        assert mission.command.datasource_manifest == {
+            "data_source_host": "data.example",
+            "scheme": "https",
+        }
+
+
+@pytest.mark.asyncio
+async def test_observation_is_build_scoped_and_control_routes_are_not_mounted() -> None:
+    app, _ = _app(Principal("owner"))
+    paths = set(app.openapi()["paths"])
+    assert not any(path.endswith("/operations") for path in paths)
+    assert not any(path.startswith("/api/traces") for path in paths)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        allowed = await client.get("/api/pattern/script_builds/7/mission")
+        assert allowed.status_code == 200
+        denied = await client.get("/api/pattern/script_builds/8/mission")
+        assert denied.status_code == 404
+        unsafe = await client.post(
+            "/api/pattern/script_builds",
+            json={
+                "execution_id": 1,
+                "topic_build_id": 2,
+                "topic_id": 3,
+                "data_source_url": "http://127.0.0.1/private",
+            },
+        )
+        assert unsafe.status_code == 400
+        secret_query = await client.post(
+            "/api/pattern/script_builds",
+            json={
+                "execution_id": 1,
+                "topic_build_id": 2,
+                "topic_id": 3,
+                "data_source_url": "https://data.example/private?token=secret",
+            },
+        )
+        assert secret_query.status_code == 400
+
+
+def test_websocket_rejects_unauthenticated_subscription_before_accept() -> None:
+    app, _ = _app(None)
+    with TestClient(app) as client:
+        with pytest.raises(WebSocketDisconnect) as caught:
+            with client.websocket_connect("/api/pattern/script_builds/7/traces/root-7/watch"):
+                pass
+    assert caught.value.code == 4404
+
+
+class _Trace:
+    trace_id = "root-7"
+    context: ClassVar[dict[str, str]] = {}
+
+    def to_dict(self) -> dict[str, str]:
+        return {"trace_id": self.trace_id}
+
+
+class _TraceStore:
+    async def get_trace(self, trace_id: str) -> _Trace | None:
+        return _Trace() if trace_id == "root-7" else None
+
+    async def get_events(self, trace_id: str, since: int) -> list[dict[str, object]]:
+        assert trace_id == "root-7"
+        del since
+        return []
+
+
+def test_websocket_requires_origin_and_accepts_authorized_build_trace() -> None:
+    app, _ = _app(Principal("owner"), trace_store=_TraceStore())
+    with TestClient(app) as client:
+        with pytest.raises(WebSocketDisconnect) as caught:
+            with client.websocket_connect(
+                "/api/pattern/script_builds/7/traces/root-7/watch",
+                headers={"origin": "https://evil.example"},
+            ):
+                pass
+        assert caught.value.code == 4404
+
+        with client.websocket_connect(
+            "/api/pattern/script_builds/7/traces/root-7/watch",
+            headers={"origin": "https://ui.example"},
+        ) as websocket:
+            websocket.send_text("ping")
+            assert websocket.receive_json() == {"event": "pong"}
+
+
+class _UploadedTopics:
+    async def parse(self, _value: object) -> dict[str, object]:
+        return {
+            "account_name": "acct",
+            "topic_fusion": "topic",
+            "points": [{"point_type": "关键点"}],
+            "item_count": 1,
+        }
+
+    async def create(
+        self, _parsed: dict[str, object], *, account_name: str | None
+    ) -> dict[str, object]:
+        return {
+            "topic_build_id": 22,
+            "topic_id": 33,
+            "account_name": account_name or "acct",
+            "item_count": 1,
+            "point_count": 1,
+        }
+
+
+@pytest.mark.asyncio
+async def test_upload_parse_and_start_routes_keep_legacy_response_contract() -> None:
+    app, mission = _app(Principal("owner"), uploaded_topics=_UploadedTopics())
+    payload = {"选题融合": "topic"}
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        parsed = await client.post(
+            "/api/pattern/script_builds/parse_topic_json",
+            json={"json_data": payload},
+        )
+        assert parsed.status_code == 200
+        assert parsed.json()["point_type_stats"] == {"关键点": 1}
+        started = await client.post(
+            "/api/pattern/script_builds/from_topic_json",
+            json={"json_data": payload, "account_name": "acct"},
+        )
+        assert started.status_code == 200
+        assert started.json()["topic_build_id"] == 22
+        assert started.json()["topic_id"] == 33
+        assert mission.command.execution_id == 0

script_build_host/tests/test_production_composition.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from agent import AgentRunner, FileSystemTraceStore
+
+from script_build_host.domain.records import Principal
+from script_build_host.infrastructure.config import ScriptBuildSettings
+from script_build_host.production import (
+    ProductionRuntimeDependencies,
+    compose_production_host,
+)
+
+
+class _Principals:
+    async def current(self, request_context: object | None = None) -> Principal:
+        if request_context is None:
+            raise PermissionError("request context is required")
+        return Principal("owner")
+
+
+class _Authorizer:
+    async def require_source_access(self, _principal: Principal, **_source: int) -> None:
+        return None
+
+    async def require_access(self, _principal: Principal, _script_build_id: int) -> None:
+        return None
+
+
+def _runtime(tmp_path: Path) -> ProductionRuntimeDependencies:
+    async def llm_call(*_args: object, **_kwargs: object) -> dict[str, str]:
+        return {"content": "unused", "finish_reason": "stop"}
+
+    return ProductionRuntimeDependencies(
+        runner=AgentRunner(
+            trace_store=FileSystemTraceStore(str(tmp_path / "traces")),
+            llm_call=llm_call,
+        ),
+        principal_provider=_Principals(),
+        build_authorizer=_Authorizer(),
+    )
+
+
+def _settings(tmp_path: Path, **overrides: object) -> ScriptBuildSettings:
+    decode = tmp_path / "decode.json"
+    knowledge = tmp_path / "knowledge.json"
+    decode.write_text(json.dumps({"items": []}), encoding="utf-8")
+    knowledge.write_text(json.dumps({"items": []}), encoding="utf-8")
+    values: dict[str, object] = {
+        "environment": "test",
+        "read_database_url": f"sqlite+aiosqlite:///{tmp_path / 'read.db'}",
+        "write_database_url": f"sqlite+aiosqlite:///{tmp_path / 'write.db'}",
+        "agent_data_root": tmp_path / "agent-data",
+        "persona_root": tmp_path / "persona",
+        "section_pattern_root": tmp_path / "sections",
+        "decode_index_root": decode,
+        "knowledge_root": knowledge,
+        "prompt_fallback_root": tmp_path / "prompts",
+        "websocket_allowed_origins": ("https://ui.example",),
+    }
+    values.update(overrides)
+    return ScriptBuildSettings(**values)  # type: ignore[arg-type]
+
+
+@pytest.mark.asyncio
+async def test_production_composition_wires_real_stores_and_closes_resources(
+    tmp_path: Path,
+) -> None:
+    host = compose_production_host(_settings(tmp_path), _runtime(tmp_path))
+    await host.validate_startup()
+    manifest = await host.runtime_manifest.load()
+    assert manifest["decode_index"]["file_count"] == 1
+    assert manifest["knowledge"]["file_count"] == 1
+    paths = set(host.composition.app.openapi()["paths"])
+    assert "/api/pattern/script_builds" in paths
+    assert any(path.startswith("/api/v2/script-builds/") for path in paths)
+    assert (
+        host.composition.coordinator.task_store.base_path
+        == (tmp_path / "agent-data" / "task-ledger").resolve()
+    )
+    await host.close()
+    await host.close()
+    assert host.http_client.is_closed
+
+
+def test_production_requires_decode_service_and_durable_trace_store(tmp_path: Path) -> None:
+    production = _settings(
+        tmp_path,
+        environment="production",
+        read_database_url="mysql+asyncmy://reader:read@db.example/app",
+        write_database_url="mysql+asyncmy://writer:write@db.example/app",
+    )
+    with pytest.raises(RuntimeError, match="DECODE_ENDPOINT"):
+        compose_production_host(production, _runtime(tmp_path))
+
+    runtime = _runtime(tmp_path)
+    runtime.runner.trace_store = None
+    with pytest.raises(RuntimeError, match="durable trace store"):
+        compose_production_host(_settings(tmp_path), runtime)

script_build_host/tests/test_start_integration.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+from sqlalchemy import event, insert, select
+
+from script_build_host.application.input_snapshot_service import ScriptInputSnapshotService
+from script_build_host.application.mission_factory import ScriptMissionFactory
+from script_build_host.application.mission_service import (
+    BuildTransitionGate,
+    ScriptMissionService,
+    StartScriptBuildCommand,
+)
+from script_build_host.domain.ports import PersonaInput
+from script_build_host.domain.records import BuildStatus, Principal
+from script_build_host.infrastructure.legacy_tables import (
+    script_build_record,
+    topic_build_record,
+    topic_build_topic,
+    topic_pattern_execution,
+)
+from script_build_host.infrastructure.tables import (
+    input_snapshot_table,
+    mission_binding_table,
+)
+from script_build_host.repositories import (
+    LegacySqlAlchemyInputReader,
+    SqlAlchemyInputSnapshotRepository,
+    SqlAlchemyLegacyBuildStateRepository,
+    SqlAlchemyMissionBindingRepository,
+)
+
+
+class _Persona:
+    async def load(self, account_name: str) -> PersonaInput:
+        return PersonaInput(account_name, account_name, (), (), {})
+
+
+class _Strategies:
+    async def load(self, **_selectors: object) -> tuple[dict[str, object], ...]:
+        return ()
+
+
+class _Prompts:
+    async def load(self, _requests: object) -> tuple[dict[str, object], ...]:
+        return ()
+
+
+class _Authorizer:
+    def __init__(self) -> None:
+        self.source_checked = False
+
+    async def require_source_access(self, _principal: Principal, **source: int) -> None:
+        assert source == {"execution_id": 10, "topic_build_id": 20, "topic_id": 30}
+        self.source_checked = True
+
+    async def require_access(self, _principal: Principal, _build: int) -> None:
+        return None
+
+
+class _StartOnlyMissionService(ScriptMissionService):
+    def __init__(self, **kwargs: object) -> None:
+        super().__init__(**kwargs)  # type: ignore[arg-type]
+        self.started_builds: list[int] = []
+
+    async def run(self, script_build_id: int) -> None:
+        self.started_builds.append(script_build_id)
+
+
+@pytest.mark.asyncio
+async def test_start_uses_real_input_snapshot_binding_and_phase_one_write_allowlist(
+    database,
+) -> None:
+    engine, sessions = database
+    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="topic demand",
+                status="success",
+                is_deleted=False,
+                personal_config={"account_name": "acct"},
+                origin="generated",
+            )
+        )
+        await session.execute(
+            insert(topic_build_topic).values(
+                id=30,
+                build_id=20,
+                execution_id=10,
+                sort_order=0,
+                result="topic",
+                status="mature",
+            )
+        )
+
+    statements: list[str] = []
+
+    def record_sql(
+        _connection: object,
+        _cursor: object,
+        statement: str,
+        _parameters: object,
+        _context: object,
+        _executemany: object,
+    ) -> None:
+        if statement.lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE")):
+            statements.append(statement.lower())
+
+    event.listen(engine.sync_engine, "before_cursor_execute", record_sql)
+    try:
+        snapshots = SqlAlchemyInputSnapshotRepository(sessions)
+        bindings = SqlAlchemyMissionBindingRepository(sessions)
+        legacy = SqlAlchemyLegacyBuildStateRepository(sessions)
+        input_service = ScriptInputSnapshotService(
+            legacy_input=LegacySqlAlchemyInputReader(sessions),
+            persona_source=_Persona(),
+            strategy_source=_Strategies(),
+            prompt_source=_Prompts(),
+            snapshots=snapshots,
+        )
+        authorizer = _Authorizer()
+        service = _StartOnlyMissionService(
+            runner=SimpleNamespace(),
+            coordinator=SimpleNamespace(),
+            factory=ScriptMissionFactory(),
+            input_snapshots=input_service,
+            bindings=bindings,
+            legacy_state=legacy,
+            authorizer=authorizer,
+            direction_reconciler=SimpleNamespace(),
+            transition_gate=BuildTransitionGate(),
+        )
+        result = await service.start(
+            StartScriptBuildCommand(
+                execution_id=10,
+                topic_build_id=20,
+                topic_id=30,
+                principal=Principal("owner"),
+                runtime_prompt_manifest=(
+                    {
+                        "preset": "script_planner",
+                        "source": "fixture",
+                        "content_sha256": "sha256:" + "1" * 64,
+                    },
+                ),
+                model_manifest={"presets": {"script_planner": {"model": "fake-model"}}},
+            )
+        )
+        await asyncio.sleep(0)
+    finally:
+        event.remove(engine.sync_engine, "before_cursor_execute", record_sql)
+
+    assert authorizer.source_checked
+    assert result.status is BuildStatus.RUNNING
+    assert service.started_builds == [result.script_build_id]
+    binding = await bindings.get_by_build(result.script_build_id)
+    snapshot = await snapshots.get(
+        str(binding.input_snapshot_id),
+        script_build_id=result.script_build_id,
+    )
+    assert snapshot.topic["execution"]["id"] == 10
+    assert snapshot.account["account_name"] == "acct"
+    async with sessions() as session:
+        assert await session.scalar(select(script_build_record.c.status)) == "running"
+        assert await session.scalar(select(input_snapshot_table.c.id)) is not None
+        assert await session.scalar(select(mission_binding_table.c.id)) is not None
+
+    allowed = {
+        "script_build_record",
+        "script_build_input_snapshot",
+        "script_build_mission_binding",
+    }
+    assert statements
+    assert all(any(table in statement for table in allowed) for statement in statements)
+    forbidden = ("paragraph", "element", "link", "round", "branch", "plan_step", "external_log")
+    assert not any(name in statement for statement in statements for name in forbidden)