feat(Phase2候选): 接入批量段落并前置 Compose 支撑校验

新增批量段落排序、重复 key、悬空父级和环检测。Compose 在创建 workspace 前先验证 Direction 与 evidence 支撑引用，避免留下半成品；放弃候选时同时清理 Compose 尝试工作区。

script_build_host/src/script_build_host/application/phase_two_candidates.py

@@ -104,6 +104,10 @@ class CandidateWorkspaceOperations(Protocol):
 
     async def create_paragraph(self, context: CandidateWriteContext, **values: Any) -> int: ...
 
+    async def create_paragraphs(
+        self, context: CandidateWriteContext, paragraphs: Sequence[Mapping[str, Any]]
+    ) -> Mapping[str, int]: ...
+
     async def batch_update_paragraphs(
         self, context: CandidateWriteContext, updates: Sequence[Mapping[str, Any]]
     ) -> tuple[int, ...]: ...
@@ -438,6 +442,14 @@ class PhaseTwoCandidateService:
         )
         return {"paragraph_id": paragraph_id}
 
+    async def create_script_paragraphs(
+        self, *, paragraphs: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        scope = await self._workspace_scope(context, paragraph=True)
+        ordered = _ordered_paragraph_batch(paragraphs)
+        ids = await self._workspace_repo().create_paragraphs(scope.write_context, ordered)
+        return {"paragraph_ids_by_client_key": dict(ids), "created": len(ids)}
+
     async def append_paragraph_atoms(
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]:
@@ -587,6 +599,10 @@ class PhaseTwoCandidateService:
                 "ACCEPTED_INPUT_CONFLICT",
                 "Compose order must enumerate every adopted active Decision exactly once",
             )
+        support_versions = tuple(
+            [await self._read_accepted_version(scope, item) for item in bundle.inputs]
+        )
+        direction_ref, evidence_refs = _compose_support_refs(support_versions)
         workspace = await self._workspace_repo().get_or_create(
             scope.write_context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT
         )
@@ -598,10 +614,6 @@ class PhaseTwoCandidateService:
                 "MISSION_RECOVERY_REQUIRED",
                 "partially materialized Compose workspace requires explicit recovery",
             )
-        support_versions = tuple(
-            [await self._read_accepted_version(scope, item) for item in bundle.inputs]
-        )
-        direction_ref, evidence_refs = _compose_support_refs(support_versions)
         await self._materialize(scope, versions, support_versions)
         version, ref = await self._workspace_repo().freeze(
             scope.write_context,
@@ -783,6 +795,7 @@ class PhaseTwoCandidateService:
             ScriptTaskKind.STRUCTURE,
             ScriptTaskKind.PARAGRAPH,
             ScriptTaskKind.ELEMENT_SET,
+            ScriptTaskKind.COMPOSE,
         }:
             return
         binding = await self._bindings.get_by_root(root_trace_id)
@@ -1465,6 +1478,64 @@ def _reject_identity(payload: Mapping[str, Any]) -> None:
         )
 
 
+def _ordered_paragraph_batch(
+    paragraphs: Sequence[Mapping[str, Any]],
+) -> tuple[Mapping[str, Any], ...]:
+    if not paragraphs:
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "paragraph batch is empty")
+    by_key: dict[str, dict[str, Any]] = {}
+    for raw in paragraphs:
+        _reject_identity(raw)
+        key = _required_text(raw, "client_key")
+        if key in by_key:
+            raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "client_key must be unique")
+        parent = raw.get("parent_client_key")
+        if parent is not None and (not isinstance(parent, str) or not parent.strip()):
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "parent_client_key must not be blank"
+            )
+        by_key[key] = {
+            "client_key": key,
+            "parent_client_key": parent.strip() if isinstance(parent, str) else None,
+            "paragraph_index": _positive_int(raw.get("paragraph_index"), "paragraph_index"),
+            "name": _required_text(raw, "name"),
+            "content_range": _mapping(raw.get("content_range"), "content_range"),
+            "level": _optional_int(raw.get("level"), default=1),
+            "theme_elements": _mapping_sequence(raw.get("theme_elements")),
+            "form_elements": _mapping_sequence(raw.get("form_elements")),
+            "function_elements": _mapping_sequence(raw.get("function_elements")),
+            "feeling_elements": _mapping_sequence(raw.get("feeling_elements")),
+        }
+    for item in by_key.values():
+        parent = item["parent_client_key"]
+        if parent is not None and parent not in by_key:
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "parent_client_key is outside the batch"
+            )
+        expected_level = 2 if parent is not None else 1
+        if item["level"] != expected_level:
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "paragraph level and parent_client_key disagree"
+            )
+    ordered: list[Mapping[str, Any]] = []
+    pending = dict(by_key)
+    emitted: set[str] = set()
+    while pending:
+        ready = [
+            key
+            for key, item in pending.items()
+            if item["parent_client_key"] is None or item["parent_client_key"] in emitted
+        ]
+        if not ready:
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "paragraph batch contains a parent cycle"
+            )
+        for key in ready:
+            ordered.append(pending.pop(key))
+            emitted.add(key)
+    return tuple(ordered)
+
+
 def _required_context(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
     if not isinstance(value, str) or not value.strip():