test(批量段落): 覆盖父子映射 冻结 与事务回滚

验证正数 branch 下批量创建 Paragraph、client_key 父子映射和最终 manifest 冻结。补充非法父引用与失败回滚场景，确保不会留下半批候选数据。

script_build_host/tests/test_phase_two_candidates.py

@@ -314,6 +314,89 @@ async def test_paragraph_workspace_uses_positive_branch_and_freezes_on_manifest(
         )
 
 
+@pytest.mark.asyncio
+async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(database: Any) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
+    contract = _contract(ScriptTaskKind.PARAGRAPH)
+    (ledger, accepted), context = _scope_fixture(contract)
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=repository,
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=workspaces,
+    )
+    created = await service.create_script_paragraphs(
+        paragraphs=(
+            {
+                "client_key": "root",
+                "paragraph_index": 1,
+                "name": "opening",
+                "content_range": {"beats": ["setup", "turn"]},
+            },
+            {
+                "client_key": "child",
+                "parent_client_key": "root",
+                "paragraph_index": 2,
+                "level": 2,
+                "name": "turn",
+                "content_range": {"beats": ["turn"]},
+            },
+        ),
+        context=context,
+    )
+    assert created["created"] == 2
+    assert set(created["paragraph_ids_by_client_key"]) == {"root", "child"}
+
+    contract2 = replace(contract, objective="second isolated workspace")
+    (ledger2, accepted2), context2 = _scope_fixture(
+        contract2, task_id="task-2", attempt_id="attempt-2"
+    )
+    service2 = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger2),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=repository,
+        accepted_inputs=accepted2,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=workspaces,
+    )
+    with pytest.raises(WorkspaceError, match="child content range"):
+        await service2.create_script_paragraphs(
+            paragraphs=(
+                {
+                    "client_key": "root",
+                    "paragraph_index": 1,
+                    "name": "root",
+                    "content_range": {"beats": ["setup"]},
+                },
+                {
+                    "client_key": "child",
+                    "parent_client_key": "root",
+                    "paragraph_index": 2,
+                    "level": 2,
+                    "name": "child",
+                    "content_range": {"beats": ["missing"]},
+                },
+            ),
+            context=context2,
+        )
+    write_context = CandidateWriteContext(
+        script_build_id=7,
+        task_id="task-2",
+        attempt_id="attempt-2",
+        spec_version=1,
+        objective=contract2.objective,
+        input_refs=(),
+        write_scope=contract2.write_scope,
+    )
+    assert not (await workspaces.snapshot(write_context)).paragraphs
+
+
 @pytest.mark.asyncio
 async def test_candidate_service_discards_existing_draft_workspace_on_abandoned_attempt(
     database: Any,