feat(候选工作区): 单事务批量落库 Paragraph

按父优先顺序在同一 workspace 事务中创建段落，校验正数 index、层级、父引用、content_range 子集、写权限和四类原子。任一段失败即整体回滚，并返回稳定的 client_key 映射。

script_build_host/src/script_build_host/repositories/workspace.py

@@ -243,6 +243,74 @@ class SqlAlchemyCandidateWorkspaceRepository:
             )
             return int(cast(Any, result).inserted_primary_key[0])
 
+    async def create_paragraphs(
+        self,
+        context: CandidateWriteContext,
+        paragraphs: Sequence[Mapping[str, Any]],
+    ) -> Mapping[str, int]:
+        """Create a validated parent-first batch in one workspace transaction."""
+
+        if not paragraphs:
+            raise WorkspaceError("LEGACY_WRITE_INVALID", "paragraph batch is empty")
+        async with self._sessions() as session, session.begin():
+            workspace = await self._locked_workspace(session, context)
+            _require_workspace_kind(workspace, _PARAGRAPH_KINDS)
+            await self._ensure_task_plan(session, context, workspace)
+            created: dict[str, int] = {}
+            ranges: dict[str, Mapping[str, Any]] = {}
+            for item in paragraphs:
+                key = str(item["client_key"])
+                parent_key = item.get("parent_client_key")
+                paragraph_index = int(item["paragraph_index"])
+                level = int(item["level"])
+                name = str(item["name"]).strip()
+                content_range = cast(Mapping[str, Any], item["content_range"])
+                if key in created or paragraph_index < 1 or level not in (1, 2) or not name:
+                    raise WorkspaceError("LEGACY_WRITE_INVALID", "paragraph fields are invalid")
+                parent_id = created.get(str(parent_key)) if parent_key is not None else None
+                if (level == 1) != (parent_id is None):
+                    raise WorkspaceError(
+                        "LEGACY_REFERENCE_INVALID", "paragraph parent is outside this batch"
+                    )
+                if parent_key is not None and not _content_range_subset(
+                    dict(content_range), dict(ranges[str(parent_key)])
+                ):
+                    raise WorkspaceError(
+                        "LEGACY_REFERENCE_INVALID", "child content range exceeds its parent"
+                    )
+                _authorize_write(
+                    context,
+                    entity="paragraphs",
+                    operation="create",
+                    identities=(paragraph_index, name),
+                    content_range=content_range,
+                )
+                atoms = {
+                    "theme_elements": _atoms(item.get("theme_elements", ()), feeling=False),
+                    "form_elements": _atoms(item.get("form_elements", ()), feeling=False),
+                    "function_elements": _atoms(item.get("function_elements", ()), feeling=False),
+                    "feeling_elements": _atoms(item.get("feeling_elements", ()), feeling=True),
+                }
+                result = await session.execute(
+                    insert(self._paragraphs).values(
+                        script_build_id=workspace.script_build_id,
+                        branch_id=workspace.branch_id,
+                        base_ref_id=None,
+                        paragraph_index=paragraph_index,
+                        level=level,
+                        parent_id=parent_id,
+                        name=name,
+                        content_range=dict(content_range),
+                        **atoms,
+                        is_active=True,
+                        created_at=_now(),
+                        updated_at=_now(),
+                    )
+                )
+                created[key] = int(cast(Any, result).inserted_primary_key[0])
+                ranges[key] = content_range
+            return created
+
     async def batch_update_paragraphs(
         self, context: CandidateWriteContext, updates: Sequence[Mapping[str, Any]]
     ) -> tuple[int, ...]: