候选工作区：合并段落元素原子语义命令

用 save_script_paragraphs 自动 upsert 新建与更新段落，用 save_script_elements 根据 paragraph_target_key 一次事务写入 Element 和 Link。命令采用 expected_state_revision、稳定 target key、精确重放 NotModified 与变更冲突拒绝，Agent 不再复制物理数据库 ID。

script_build_host/src/script_build_host/application/phase_two_candidates.py

@@ -9,6 +9,7 @@ business Artifact.
 from __future__ import annotations
 
 import base64
+import json
 from collections.abc import Mapping, Sequence
 from dataclasses import asdict, dataclass, replace
 from hashlib import sha256
@@ -64,6 +65,11 @@ from script_build_host.domain.task_contracts import (
     ScriptTaskContractV1,
     ScriptTaskKind,
 )
+from script_build_host.domain.workbench import (
+    CommandReceipt,
+    paragraph_target_key,
+    semantic_handle,
+)
 from script_build_host.domain.workspaces import (
     CandidateWorkspaceSnapshot,
     CandidateWriteContext,
@@ -122,6 +128,10 @@ class CandidateWorkspaceOperations(Protocol):
         self, context: CandidateWriteContext, updates: Sequence[Mapping[str, Any]]
     ) -> tuple[int, ...]: ...
 
+    async def save_paragraphs(
+        self, context: CandidateWriteContext, paragraphs: Sequence[Mapping[str, Any]]
+    ) -> tuple[Mapping[str, int], tuple[int, ...]]: ...
+
     async def append_paragraph_atoms(
         self, context: CandidateWriteContext, **values: Any
     ) -> int: ...
@@ -130,6 +140,13 @@ class CandidateWorkspaceOperations(Protocol):
 
     async def create_element(self, context: CandidateWriteContext, **values: Any) -> int: ...
 
+    async def create_elements(
+        self,
+        context: CandidateWriteContext,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+    ) -> tuple[Mapping[str, int], int]: ...
+
     async def update_element(
         self, context: CandidateWriteContext, *, element_id: int, values: Mapping[str, Any]
     ) -> None: ...
@@ -255,7 +272,7 @@ class PhaseTwoCandidateService:
     async def view_frozen_images(
         self,
         *,
-        raw_artifact_refs: Sequence[str],
+        image_handles: Sequence[str],
         context: Mapping[str, Any],
     ) -> Sequence[Mapping[str, Any]]:
         if self._raw_artifacts is None:
@@ -263,22 +280,24 @@ class PhaseTwoCandidateService:
                 "LEGACY_REFERENCE_INVALID", "raw Artifact storage is not configured"
             )
         if (
-            not raw_artifact_refs
-            or len(raw_artifact_refs) > self._max_images
-            or len(set(raw_artifact_refs)) != len(raw_artifact_refs)
+            not image_handles
+            or len(image_handles) > self._max_images
+            or len(set(image_handles)) != len(image_handles)
         ):
             raise PhaseTwoCandidateError(
                 "LEGACY_REFERENCE_INVALID", "image references exceed the frozen view limit"
             )
         scope, bundle = await self._resolve_bundle(context)
         allowed = await self._allowed_raw_refs(scope, bundle, context)
-        if any(ref not in allowed for ref in raw_artifact_refs):
+        refs_by_handle = {semantic_handle("image", ref): ref for ref in allowed}
+        if any(handle not in refs_by_handle for handle in image_handles):
             raise PhaseTwoCandidateError(
                 "LEGACY_REFERENCE_INVALID", "image is outside the accepted frozen inputs"
             )
         total = 0
         output: list[Mapping[str, Any]] = []
-        for ref in raw_artifact_refs:
+        for image_handle in image_handles:
+            ref = refs_by_handle[image_handle]
             content = await self._raw_artifacts.read_bytes(ref)
             expected = ref.removeprefix(_RAW_PREFIX)
             actual = sha256(content).hexdigest()
@@ -301,8 +320,8 @@ class PhaseTwoCandidateService:
                     "type": "base64",
                     "media_type": media_type,
                     "data": base64.b64encode(content).decode("ascii"),
-                    "raw_artifact_ref": ref,
-                    "digest": f"sha256:{actual}",
+                    "image_handle": image_handle,
+                    "sha256": f"sha256:{actual}",
                     "size_bytes": len(content),
                 }
             )
@@ -377,7 +396,8 @@ class PhaseTwoCandidateService:
                 and lineage.supersedes_decision_ids == scope.contract.supersedes_decision_ids
                 and lineage.base_artifact_ref == (base.uri if base else None)
                 and lineage.base_artifact_digest == (base.digest if base else None)
-                and lineage.base_revision == (int(base.version) if base else None)
+                and lineage.base_revision
+                == (int(base.version) if base is not None and base.version is not None else None)
             )
             rules.append(
                 {
@@ -458,9 +478,7 @@ class PhaseTwoCandidateService:
                     raise PhaseTwoCandidateError(
                         "GOAL_COVERAGE_INVALID", "Portfolio adoption is not a StructuredScript"
                     )
-                if structured.direction_ref != (
-                    f"script-build://artifact-versions/{direction_id}"
-                ):
+                if structured.direction_ref != (f"script-build://artifact-versions/{direction_id}"):
                     raise PhaseTwoCandidateError(
                         "GOAL_COVERAGE_INVALID", "StructuredScript uses a stale Direction"
                     )
@@ -562,6 +580,80 @@ class PhaseTwoCandidateService:
         ids = await self._workspace_repo().create_paragraphs(scope.write_context, ordered)
         return {"paragraph_ids_by_client_key": dict(ids), "created": len(ids)}
 
+    async def save_script_paragraphs(
+        self,
+        *,
+        paragraphs: Sequence[Mapping[str, Any]],
+        expected_state_revision: str,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        scope = await self._workspace_scope(context, paragraph=True)
+        before = await self._workspace_repo().snapshot(scope.write_context)
+        ledger = await self._task_store.load(scope.root_trace_id)
+        current_revision = _candidate_state_revision(scope, before, ledger.revision)
+        if expected_state_revision != current_revision:
+            applied = _applied_paragraph_targets(scope, before, paragraphs)
+            if applied is None:
+                _require_state_revision(scope, before, ledger.revision, expected_state_revision)
+                raise AssertionError("state revision check must reject a changed command")
+            return CommandReceipt(
+                state_revision=current_revision,
+                status="not_modified",
+                changed_target_keys=applied,
+                workspace_digest=_workspace_digest(before),
+            ).to_payload()
+        by_target = {
+            paragraph_target_key(
+                scope.root_trace_id,
+                scope.task.task_id,
+                scope.attempt.attempt_id,
+                item.paragraph_id,
+            ): item.paragraph_id
+            for item in before.paragraphs
+        }
+        normalized: list[dict[str, Any]] = []
+        for raw in paragraphs:
+            item = dict(raw)
+            target = item.pop("paragraph_target_key", None)
+            parent_target = item.pop("parent_target_key", None)
+            if target is not None:
+                paragraph_id = by_target.get(str(target))
+                if paragraph_id is None:
+                    raise PhaseTwoCandidateError(
+                        "LEGACY_REFERENCE_INVALID", "paragraph target is stale or unknown"
+                    )
+                item["paragraph_id"] = paragraph_id
+            if parent_target is not None:
+                parent_id = by_target.get(str(parent_target))
+                if parent_id is None:
+                    raise PhaseTwoCandidateError(
+                        "LEGACY_REFERENCE_INVALID", "paragraph parent target is stale or unknown"
+                    )
+                item["parent_paragraph_id"] = parent_id
+            normalized.append(item)
+        created, updated = await self._workspace_repo().save_paragraphs(
+            scope.write_context, normalized
+        )
+        after = await self._workspace_repo().snapshot(scope.write_context)
+        receipt = CommandReceipt(
+            state_revision=_candidate_state_revision(scope, after, ledger.revision),
+            status="saved",
+            created=len(created),
+            updated=len(updated),
+            changed_target_keys=tuple(
+                paragraph_target_key(
+                    scope.root_trace_id,
+                    scope.task.task_id,
+                    scope.attempt.attempt_id,
+                    item.paragraph_id,
+                )
+                for item in after.paragraphs
+                if item.paragraph_id in set(created.values()) | set(updated)
+            ),
+            workspace_digest=_workspace_digest(after),
+        )
+        return receipt.to_payload()
+
     async def append_paragraph_atoms(
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]:
@@ -615,6 +707,84 @@ class PhaseTwoCandidateService:
         )
         return {"element_id": element_id}
 
+    async def create_script_elements(
+        self,
+        *,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        scope = await self._workspace_scope(context, element=True)
+        ids, linked = await self._workspace_repo().create_elements(
+            scope.write_context,
+            elements,
+            links,
+        )
+        return {
+            "element_ids_by_client_key": dict(ids),
+            "created": len(ids),
+            "linked": linked,
+        }
+
+    async def save_script_elements(
+        self,
+        *,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+        expected_state_revision: str,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        scope = await self._workspace_scope(context, element=True)
+        before = await self._workspace_repo().snapshot(scope.write_context)
+        ledger = await self._task_store.load(scope.root_trace_id)
+        current_revision = _candidate_state_revision(scope, before, ledger.revision)
+        if expected_state_revision != current_revision:
+            if not _applied_element_command(scope, before, elements, links):
+                _require_state_revision(scope, before, ledger.revision, expected_state_revision)
+            return CommandReceipt(
+                state_revision=current_revision,
+                status="not_modified",
+                workspace_digest=_workspace_digest(before),
+            ).to_payload()
+        if before.elements or before.links:
+            raise PhaseTwoCandidateError(
+                "LEGACY_WRITE_INVALID", "ElementSet must be saved once per Attempt"
+            )
+        by_target = {
+            paragraph_target_key(
+                scope.root_trace_id,
+                scope.task.task_id,
+                scope.attempt.attempt_id,
+                item.paragraph_id,
+            ): item.paragraph_id
+            for item in before.paragraphs
+        }
+        normalized_links: list[dict[str, Any]] = []
+        for raw in links:
+            target = str(raw.get("paragraph_target_key") or "")
+            paragraph_id = by_target.get(target)
+            if paragraph_id is None:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_REFERENCE_INVALID", "Element link paragraph target is invalid"
+                )
+            normalized_links.append(
+                {
+                    "paragraph_id": paragraph_id,
+                    "element_client_keys": raw.get("element_client_keys"),
+                }
+            )
+        ids, linked = await self._workspace_repo().create_elements(
+            scope.write_context, elements, normalized_links
+        )
+        after = await self._workspace_repo().snapshot(scope.write_context)
+        return CommandReceipt(
+            state_revision=_candidate_state_revision(scope, after, ledger.revision),
+            status="saved",
+            created=len(ids),
+            linked=linked,
+            workspace_digest=_workspace_digest(after),
+        ).to_payload()
+
     async def update_script_element(
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]:
@@ -684,15 +854,25 @@ class PhaseTwoCandidateService:
                 "ACCEPTED_INPUT_CONFLICT", "comparison requires two frozen DecisionRefs"
             )
         await self._require_bundle_refs(bundle, scope.contract.comparison_decision_refs)
+        recommended_decision_id = _required_text(payload, "recommended_decision_id")
+        recommendation_by_decision = {
+            item.decision_id: item.artifact_ref.uri
+            for item in scope.contract.comparison_decision_refs
+        }
+        if recommended_decision_id not in recommendation_by_decision:
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "recommended Decision is outside frozen candidates"
+            )
+        criterion_results = tuple(
+            _comparison_criterion_result(item, recommendation_by_decision)
+            for item in _mapping_sequence(payload.get("criterion_results"), required=True)
+        )
         artifact = ComparisonArtifactV1(
             lineage=await self._lineage(scope, bundle),
             candidate_artifact_refs=expected,
-            criterion_results=tuple(
-                dict(item)
-                for item in _mapping_sequence(payload.get("criterion_results"), required=True)
-            ),
+            criterion_results=criterion_results,
             conflicts=_text_sequence(payload.get("conflicts")),
-            recommendation=_required_text(payload, "recommendation"),
+            recommendation=recommendation_by_decision[recommended_decision_id],
         )
         return await self._freeze_direct(scope, artifact)
 
@@ -1112,6 +1292,25 @@ class PhaseTwoCandidateService:
         )
         if workspace.branch_id <= 0:
             raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "workspace cannot use branch 0")
+        if scope.contract.task_kind is ScriptTaskKind.ELEMENT_SET and base is None:
+            snapshot = await self._workspace_repo().snapshot(scope.write_context)
+            if not snapshot.paragraphs and not snapshot.elements and not snapshot.links:
+                _, bundle = await self._resolve_bundle(context)
+                paragraph_versions = tuple(
+                    [
+                        await self._read_accepted_version(scope, item)
+                        for item in bundle.inputs
+                        if item.task_kind is ScriptTaskKind.PARAGRAPH
+                    ]
+                )
+                if paragraph_versions:
+                    await self._materialize(
+                        scope,
+                        paragraph_versions,
+                        (),
+                        require_structure_coverage=False,
+                        require_complete=False,
+                    )
         return scope
 
     def _workspace_repo(self) -> CandidateWorkspaceOperations:
@@ -1286,6 +1485,9 @@ class PhaseTwoCandidateService:
         scope: CandidateScope,
         versions: Sequence[ArtifactVersion],
         support_versions: Sequence[ArtifactVersion],
+        *,
+        require_structure_coverage: bool = True,
+        require_complete: bool = True,
     ) -> None:
         """Materialize the explicitly ordered logical projection once.
 
@@ -1416,7 +1618,7 @@ class PhaseTwoCandidateService:
                 link_projection.add((linked_paragraph_key, linked_element_key))
 
         uncovered = paragraph_candidate_keys - structure_paragraph_keys
-        if uncovered:
+        if require_structure_coverage and uncovered:
             raise PhaseTwoCandidateError(
                 "INPUT_SCOPE_MISMATCH",
                 "adopted Paragraphs must be covered or absorbed by an adopted Structure",
@@ -1436,11 +1638,12 @@ class PhaseTwoCandidateService:
                 "LEGACY_REFERENCE_INVALID",
                 "every adopted Element must be linked to an adopted Paragraph",
             )
-        _require_complete_projection(
-            paragraph_projection=paragraph_projection,
-            element_projection=element_projection,
-            links=active_links,
-        )
+        if require_complete:
+            _require_complete_projection(
+                paragraph_projection=paragraph_projection,
+                element_projection=element_projection,
+                links=active_links,
+            )
 
         paragraph_ids: dict[tuple[str, str, int], int] = {}
         next_index = 1
@@ -1609,6 +1812,210 @@ def _workspace_payload(value: CandidateWorkspaceSnapshot) -> dict[str, Any]:
     }
 
 
+def _workspace_digest(value: CandidateWorkspaceSnapshot) -> str:
+    encoded = json.dumps(
+        _workspace_payload(value),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    ).encode()
+    return "sha256:" + sha256(encoded).hexdigest()
+
+
+def _candidate_state_revision(
+    scope: CandidateScope, snapshot: CandidateWorkspaceSnapshot, ledger_revision: int
+) -> str:
+    encoded = json.dumps(
+        {
+            "ledger": ledger_revision,
+            "workspace": _workspace_digest(snapshot),
+            "contract": scope.contract.to_payload(),
+        },
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    ).encode()
+    return "sha256:" + sha256(encoded).hexdigest()
+
+
+def _require_state_revision(
+    scope: CandidateScope,
+    snapshot: CandidateWorkspaceSnapshot,
+    ledger_revision: int,
+    expected: str,
+) -> None:
+    current = _candidate_state_revision(scope, snapshot, ledger_revision)
+    if expected != current:
+        raise PhaseTwoCandidateError(
+            "STALE_WORKBENCH_STATE",
+            "Workbench state changed; refresh role context before saving",
+        )
+
+
+def _applied_paragraph_targets(
+    scope: CandidateScope,
+    snapshot: CandidateWorkspaceSnapshot,
+    paragraphs: Sequence[Mapping[str, Any]],
+) -> tuple[str, ...] | None:
+    """Recognize an exact replay after the first transaction committed."""
+
+    by_id = {item.paragraph_id: item for item in snapshot.paragraphs}
+    by_target = {
+        paragraph_target_key(
+            scope.root_trace_id,
+            scope.task.task_id,
+            scope.attempt.attempt_id,
+            item.paragraph_id,
+        ): item
+        for item in snapshot.paragraphs
+    }
+    by_identity: dict[tuple[int, str], list[Any]] = {}
+    for item in snapshot.paragraphs:
+        by_identity.setdefault((item.paragraph_index, item.name), []).append(item)
+
+    resolved: list[tuple[Mapping[str, Any], Any]] = []
+    client_ids: dict[str, int] = {}
+    for raw in paragraphs:
+        target = raw.get("paragraph_target_key")
+        client_key = str(raw.get("client_key") or "").strip()
+        if target is not None:
+            actual = by_target.get(str(target))
+        else:
+            index = raw.get("paragraph_index")
+            name = str(raw.get("name") or "").strip()
+            if isinstance(index, bool) or not isinstance(index, int) or not name:
+                return None
+            candidates = by_identity.get((index, name), [])
+            actual = candidates[0] if len(candidates) == 1 else None
+        if actual is None or (client_key and client_key in client_ids):
+            return None
+        if client_key:
+            client_ids[client_key] = actual.paragraph_id
+        resolved.append((raw, actual))
+
+    comparable = {
+        "paragraph_index",
+        "name",
+        "content_range",
+        "level",
+        "theme_elements",
+        "form_elements",
+        "function_elements",
+        "feeling_elements",
+        "theme",
+        "form",
+        "function",
+        "feeling",
+        "description",
+        "full_description",
+    }
+    for raw, actual in resolved:
+        payload = actual.to_payload()
+        for field in comparable & set(raw):
+            if _json_equivalent(payload.get(field), raw.get(field)) is False:
+                return None
+        if "parent_client_key" in raw or "parent_target_key" in raw:
+            parent_client = raw.get("parent_client_key")
+            parent_target = raw.get("parent_target_key")
+            if parent_client is not None:
+                expected_parent = client_ids.get(str(parent_client))
+            elif parent_target is not None:
+                parent = by_target.get(str(parent_target))
+                expected_parent = parent.paragraph_id if parent is not None else None
+            else:
+                expected_parent = None
+            if actual.parent_id != expected_parent:
+                return None
+        if actual.parent_id is not None and actual.parent_id not in by_id:
+            return None
+    return tuple(
+        paragraph_target_key(
+            scope.root_trace_id,
+            scope.task.task_id,
+            scope.attempt.attempt_id,
+            actual.paragraph_id,
+        )
+        for _, actual in resolved
+    )
+
+
+def _applied_element_command(
+    scope: CandidateScope,
+    snapshot: CandidateWorkspaceSnapshot,
+    elements: Sequence[Mapping[str, Any]],
+    links: Sequence[Mapping[str, Any]],
+) -> bool:
+    """Recognize an exact Element/Link command replay without storing a second state."""
+
+    if not snapshot.elements or not snapshot.links:
+        return False
+    actual_by_key = {
+        (item.name, item.dimension_primary, item.dimension_secondary): item
+        for item in snapshot.elements
+    }
+    client_ids: dict[str, int] = {}
+    comparable = {
+        "name",
+        "dimension_primary",
+        "dimension_secondary",
+        "commonality_analysis",
+        "topic_support",
+        "weight_score",
+        "support_elements",
+    }
+    for raw in elements:
+        client_key = str(raw.get("client_key") or "").strip()
+        identity = (
+            str(raw.get("name") or "").strip(),
+            str(raw.get("dimension_primary") or "").strip(),
+            str(raw.get("dimension_secondary") or "").strip(),
+        )
+        actual = actual_by_key.get(identity)
+        if not client_key or client_key in client_ids or actual is None:
+            return False
+        payload = actual.to_payload()
+        for field in comparable & set(raw):
+            if not _json_equivalent(payload.get(field), raw.get(field)):
+                return False
+        client_ids[client_key] = actual.element_id
+    if len(client_ids) != len(snapshot.elements):
+        return False
+    by_target = {
+        paragraph_target_key(
+            scope.root_trace_id,
+            scope.task.task_id,
+            scope.attempt.attempt_id,
+            item.paragraph_id,
+        ): item.paragraph_id
+        for item in snapshot.paragraphs
+    }
+    expected_links: set[tuple[int, int]] = set()
+    for raw in links:
+        paragraph_id = by_target.get(str(raw.get("paragraph_target_key") or ""))
+        element_keys = raw.get("element_client_keys")
+        if (
+            paragraph_id is None
+            or not isinstance(element_keys, Sequence)
+            or isinstance(element_keys, (str, bytes))
+        ):
+            return False
+        for key in element_keys:
+            element_id = client_ids.get(str(key))
+            if element_id is None:
+                return False
+            expected_links.add((paragraph_id, element_id))
+    actual_links = {(item.paragraph_id, item.element_id) for item in snapshot.links}
+    return expected_links == actual_links
+
+
+def _json_equivalent(left: Any, right: Any) -> bool:
+    return json.dumps(
+        left, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str
+    ) == json.dumps(right, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
+
+
 def _artifact_payload(version: ArtifactVersion) -> Mapping[str, Any]:
     content_payload = getattr(version.artifact, "content_payload", None)
     if callable(content_payload):
@@ -1664,9 +2071,7 @@ def _accepted_direction(versions: Sequence[ArtifactVersion]) -> DirectionArtifac
 
 def _require_paragraph_structures(versions: Sequence[ArtifactVersion]) -> None:
     structures = [
-        item.artifact
-        for item in versions
-        if isinstance(item.artifact, StructureArtifactV1)
+        item.artifact for item in versions if isinstance(item.artifact, StructureArtifactV1)
     ]
     for version in versions:
         paragraph = version.artifact
@@ -1696,13 +2101,8 @@ def _require_complete_projection(
     element_projection: Mapping[tuple[str, str, int], ScriptElementV1],
     links: set[tuple[tuple[str, str, int], tuple[str, str, int]]],
 ) -> None:
-    paragraph_keys = {
-        key: index
-        for index, key in enumerate(sorted(paragraph_projection), start=1)
-    }
-    element_keys = {
-        key: index for index, key in enumerate(sorted(element_projection), start=1)
-    }
+    paragraph_keys = {key: index for index, key in enumerate(sorted(paragraph_projection), start=1)}
+    element_keys = {key: index for index, key in enumerate(sorted(element_projection), start=1)}
     paragraphs = tuple(
         replace(
             paragraph,
@@ -1775,6 +2175,22 @@ def _ordered_paragraph_batch(
             "function_elements": _mapping_sequence(raw.get("function_elements")),
             "feeling_elements": _mapping_sequence(raw.get("feeling_elements")),
         }
+        for field in (
+            "theme",
+            "form",
+            "function",
+            "feeling",
+            "description",
+            "full_description",
+        ):
+            if field not in raw:
+                continue
+            value = raw[field]
+            if not isinstance(value, str) or not value.strip():
+                raise PhaseTwoCandidateError(
+                    "LEGACY_WRITE_INVALID", f"paragraph {field} must not be blank"
+                )
+            by_key[key][field] = value.strip()
     for item in by_key.values():
         parent = item["parent_client_key"]
         if parent is not None and parent not in by_key:
@@ -1812,6 +2228,41 @@ def _required_context(context: Mapping[str, Any], key: str) -> str:
     return value
 
 
+def _comparison_criterion_result(
+    value: Mapping[str, Any], candidates: Mapping[str, str]
+) -> dict[str, Any]:
+    unknown = set(value) - {"criterion_id", "candidate_results"}
+    results = value.get("candidate_results")
+    if unknown or not isinstance(results, Sequence) or isinstance(results, (str, bytes)):
+        raise PhaseTwoCandidateError(
+            "LEGACY_WRITE_INVALID", "comparison criterion result has an invalid shape"
+        )
+    normalized: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for raw in results:
+        if not isinstance(raw, Mapping) or set(raw) - {"decision_id", "reason"}:
+            raise PhaseTwoCandidateError(
+                "LEGACY_WRITE_INVALID", "comparison candidate result has an invalid shape"
+            )
+        decision_id = _required_text(raw, "decision_id")
+        if decision_id not in candidates or decision_id in seen:
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "comparison result references a non-candidate Decision"
+            )
+        seen.add(decision_id)
+        normalized.append(
+            {"artifact_ref": candidates[decision_id], "reason": _required_text(raw, "reason")}
+        )
+    if seen != set(candidates):
+        raise PhaseTwoCandidateError(
+            "CHILD_DECISION_INVALID", "comparison must evaluate every frozen candidate"
+        )
+    return {
+        "criterion_id": _required_text(value, "criterion_id"),
+        "candidate_results": normalized,
+    }
+
+
 def _required_text(payload: Mapping[str, Any], key: str) -> str:
     value = payload.get(key)
     if not isinstance(value, str) or not value.strip():

script_build_host/src/script_build_host/repositories/workspace.py

@@ -62,7 +62,12 @@ _ATOM_COLUMNS = frozenset(
     {"theme_elements", "form_elements", "function_elements", "feeling_elements"}
 )
 _PARAGRAPH_KINDS = frozenset(
-    {ArtifactKind.STRUCTURE, ArtifactKind.PARAGRAPH, ArtifactKind.STRUCTURED_SCRIPT}
+    {
+        ArtifactKind.STRUCTURE,
+        ArtifactKind.PARAGRAPH,
+        ArtifactKind.ELEMENT_SET,
+        ArtifactKind.STRUCTURED_SCRIPT,
+    }
 )
 _ELEMENT_KINDS = frozenset({ArtifactKind.ELEMENT_SET, ArtifactKind.STRUCTURED_SCRIPT})
 
@@ -292,6 +297,18 @@ class SqlAlchemyCandidateWorkspaceRepository:
                     "function_elements": _atoms(item.get("function_elements", ()), feeling=False),
                     "feeling_elements": _atoms(item.get("feeling_elements", ()), feeling=True),
                 }
+                prose = {
+                    field: item[field]
+                    for field in (
+                        "theme",
+                        "form",
+                        "function",
+                        "feeling",
+                        "description",
+                        "full_description",
+                    )
+                    if field in item
+                }
                 result = await session.execute(
                     insert(self._paragraphs).values(
                         script_build_id=workspace.script_build_id,
@@ -303,6 +320,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
                         name=name,
                         content_range=dict(content_range),
                         **atoms,
+                        **prose,
                         is_active=True,
                         created_at=_now(),
                         updated_at=_now(),
@@ -312,6 +330,195 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 ranges[key] = content_range
             return created
 
+    async def save_paragraphs(
+        self,
+        context: CandidateWriteContext,
+        paragraphs: Sequence[Mapping[str, Any]],
+    ) -> tuple[Mapping[str, int], tuple[int, ...]]:
+        """Atomically create or update Paragraphs using Host-resolved identities."""
+
+        if not paragraphs:
+            raise WorkspaceError("LEGACY_WRITE_INVALID", "paragraph save 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] = {}
+            updated: list[int] = []
+            ranges: dict[str, Mapping[str, Any]] = {}
+            seen_targets: set[int] = set()
+            for raw in paragraphs:
+                item = dict(raw)
+                client_key = str(item.pop("client_key", "") or "").strip()
+                paragraph_id_value = item.pop("paragraph_id", None)
+                if bool(client_key) == (paragraph_id_value is not None):
+                    raise WorkspaceError(
+                        "LEGACY_WRITE_INVALID",
+                        "each paragraph requires exactly one client_key or resolved target",
+                    )
+                parent_client_key = item.pop("parent_client_key", None)
+                parent_id = item.pop("parent_paragraph_id", None)
+                if parent_client_key is not None:
+                    if parent_id is not None or str(parent_client_key) not in created:
+                        raise WorkspaceError(
+                            "LEGACY_REFERENCE_INVALID", "paragraph parent client key is invalid"
+                        )
+                    parent_id = created[str(parent_client_key)]
+                row = None
+                if paragraph_id_value is not None:
+                    paragraph_id = _positive_int(paragraph_id_value, "paragraph_id")
+                    if paragraph_id in seen_targets:
+                        raise WorkspaceError(
+                            "LEGACY_WRITE_INVALID", "paragraph target is duplicated"
+                        )
+                    seen_targets.add(paragraph_id)
+                    row = await self._paragraph(session, workspace, paragraph_id)
+                    if row is None:
+                        raise WorkspaceError(
+                            "LEGACY_REFERENCE_INVALID", "paragraph target is outside workspace"
+                        )
+                required = ("paragraph_index", "name", "content_range", "level")
+                if row is None and any(field not in item for field in required):
+                    raise WorkspaceError(
+                        "LEGACY_WRITE_INVALID", "new paragraph is missing required fields"
+                    )
+                values: dict[str, Any] = {}
+                if "paragraph_index" in item:
+                    values["paragraph_index"] = _positive_int(
+                        item["paragraph_index"], "paragraph_index"
+                    )
+                if "name" in item:
+                    name = str(item["name"] or "").strip()
+                    if not name:
+                        raise WorkspaceError(
+                            "LEGACY_WRITE_INVALID", "paragraph name must not be blank"
+                        )
+                    values["name"] = name
+                if "content_range" in item:
+                    if not isinstance(item["content_range"], Mapping):
+                        raise WorkspaceError(
+                            "LEGACY_WRITE_INVALID", "paragraph content_range must be an object"
+                        )
+                    values["content_range"] = dict(item["content_range"])
+                if "level" in item:
+                    level = item["level"]
+                    if isinstance(level, bool) or level not in (1, 2):
+                        raise WorkspaceError(
+                            "LEGACY_WRITE_INVALID", "paragraph level must be 1 or 2"
+                        )
+                    values["level"] = int(level)
+                for field in _ATOM_COLUMNS:
+                    if field in item:
+                        values[field] = _atoms(item[field], feeling=field == "feeling_elements")
+                for field in (
+                    "theme",
+                    "form",
+                    "function",
+                    "feeling",
+                    "description",
+                    "full_description",
+                ):
+                    if field in item:
+                        value = item[field]
+                        if value is not None and not isinstance(value, str):
+                            raise WorkspaceError(
+                                "LEGACY_WRITE_INVALID", f"paragraph {field} must be text or null"
+                            )
+                        values[field] = value
+                allowed = {
+                    *required,
+                    *_ATOM_COLUMNS,
+                    "theme",
+                    "form",
+                    "function",
+                    "feeling",
+                    "description",
+                    "full_description",
+                }
+                if set(item) - allowed:
+                    raise WorkspaceError(
+                        "LEGACY_WRITE_INVALID", "paragraph save contains unsupported fields"
+                    )
+                level = int(values.get("level", row["level"] if row is not None else 1))
+                effective_parent = (
+                    parent_id
+                    if parent_id is not None
+                    else (row["parent_id"] if row is not None else None)
+                )
+                if level == 1:
+                    effective_parent = None
+                elif effective_parent is None:
+                    raise WorkspaceError(
+                        "LEGACY_REFERENCE_INVALID", "level-two paragraph requires a parent"
+                    )
+                if effective_parent is not None:
+                    parent = await self._paragraph(
+                        session, workspace, int(effective_parent), active=True
+                    )
+                    if parent is None or int(parent["level"]) != 1:
+                        raise WorkspaceError(
+                            "LEGACY_REFERENCE_INVALID", "paragraph parent is invalid"
+                        )
+                    content_range = cast(
+                        Mapping[str, Any],
+                        values.get(
+                            "content_range", row["content_range"] if row is not None else {}
+                        ),
+                    )
+                    if not _content_range_subset(content_range, parent["content_range"] or {}):
+                        raise WorkspaceError(
+                            "LEGACY_REFERENCE_INVALID", "child content range exceeds parent"
+                        )
+                values["parent_id"] = effective_parent
+                _authorize_write(
+                    context,
+                    entity="paragraphs",
+                    operation="update" if row is not None else "create",
+                    identities=(
+                        int(
+                            values.get(
+                                "paragraph_index",
+                                row["paragraph_index"] if row is not None else 0,
+                            )
+                        ),
+                        str(values.get("name", row["name"] if row is not None else "")),
+                    ),
+                    content_range=cast(
+                        Mapping[str, Any],
+                        values.get(
+                            "content_range", row["content_range"] if row is not None else {}
+                        ),
+                    ),
+                )
+                now = _now()
+                if row is None:
+                    result = await session.execute(
+                        insert(self._paragraphs).values(
+                            script_build_id=workspace.script_build_id,
+                            branch_id=workspace.branch_id,
+                            base_ref_id=None,
+                            **values,
+                            is_active=True,
+                            created_at=now,
+                            updated_at=now,
+                        )
+                    )
+                    created[client_key] = int(cast(Any, result).inserted_primary_key[0])
+                    ranges[client_key] = cast(Mapping[str, Any], values["content_range"])
+                else:
+                    values["updated_at"] = now
+                    await session.execute(
+                        update(self._paragraphs)
+                        .where(
+                            self._paragraphs.c.id == int(row["id"]),
+                            self._paragraphs.c.script_build_id == workspace.script_build_id,
+                            self._paragraphs.c.branch_id == workspace.branch_id,
+                        )
+                        .values(**values)
+                    )
+                    updated.append(int(row["id"]))
+            return created, tuple(updated)
+
     async def batch_update_paragraphs(
         self, context: CandidateWriteContext, updates: Sequence[Mapping[str, Any]]
     ) -> tuple[int, ...]:
@@ -602,6 +809,138 @@ class SqlAlchemyCandidateWorkspaceRepository:
             )
             return int(cast(Any, result).inserted_primary_key[0])
 
+    async def create_elements(
+        self,
+        context: CandidateWriteContext,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+    ) -> tuple[Mapping[str, int], int]:
+        """Create and link an Element batch in one workspace transaction."""
+
+        if not elements or not links:
+            raise WorkspaceError(
+                "LEGACY_WRITE_INVALID", "element and link batches must both be non-empty"
+            )
+        normalized: list[dict[str, Any]] = []
+        keys: set[str] = set()
+        names: set[str] = set()
+        dimensions: set[tuple[str, str]] = set()
+        for raw in elements:
+            client_key = str(raw.get("client_key") or "").strip()
+            name = str(raw.get("name") or "").strip()
+            primary = str(raw.get("dimension_primary") or "").strip()
+            secondary = str(raw.get("dimension_secondary") or "").strip()
+            if not client_key or len(client_key) > 128 or client_key in keys:
+                raise WorkspaceError(
+                    "LEGACY_WRITE_INVALID", "element client_key is empty or duplicated"
+                )
+            _element_fields(name, primary, secondary)
+            dimension_key = (primary, secondary)
+            if name in names or dimension_key in dimensions:
+                raise WorkspaceError(
+                    "LEGACY_WRITE_INVALID", "element name or dimension pair is duplicated"
+                )
+            support = raw.get("support_elements") or ()
+            if (
+                not isinstance(support, Sequence)
+                or isinstance(support, (str, bytes))
+                or any(not isinstance(item, Mapping) for item in support)
+            ):
+                raise WorkspaceError(
+                    "LEGACY_WRITE_INVALID", "element support_elements must be an object array"
+                )
+            optional_objects: dict[str, dict[str, Any] | None] = {}
+            for field in ("commonality_analysis", "topic_support", "weight_score"):
+                value = raw.get(field)
+                if value is not None and not isinstance(value, Mapping):
+                    raise WorkspaceError(
+                        "LEGACY_WRITE_INVALID", f"element {field} must be an object or null"
+                    )
+                optional_objects[field] = dict(value) if value is not None else None
+            normalized.append(
+                {
+                    "client_key": client_key,
+                    "name": name,
+                    "dimension_primary": primary,
+                    "dimension_secondary": secondary,
+                    **optional_objects,
+                    "support_elements": [dict(item) for item in support],
+                }
+            )
+            keys.add(client_key)
+            names.add(name)
+            dimensions.add(dimension_key)
+            _authorize_write(context, entity="elements", operation="create", identities=(name,))
+
+        normalized_links: list[tuple[int, tuple[str, ...]]] = []
+        linked_keys: set[str] = set()
+        for raw in links:
+            paragraph_id = raw.get("paragraph_id")
+            element_keys = raw.get("element_client_keys")
+            if (
+                isinstance(paragraph_id, bool)
+                or not isinstance(paragraph_id, int)
+                or paragraph_id < 1
+                or not isinstance(element_keys, Sequence)
+                or isinstance(element_keys, (str, bytes))
+            ):
+                raise WorkspaceError("LEGACY_WRITE_INVALID", "element link batch is invalid")
+            normalized_keys = tuple(str(item).strip() for item in element_keys)
+            if not normalized_keys or any(not item or item not in keys for item in normalized_keys):
+                raise WorkspaceError(
+                    "LEGACY_REFERENCE_INVALID", "element link references an unknown client_key"
+                )
+            normalized_links.append((paragraph_id, normalized_keys))
+            linked_keys.update(normalized_keys)
+        if linked_keys != keys:
+            raise WorkspaceError("LEGACY_REFERENCE_INVALID", "every batched Element must be linked")
+
+        async with self._sessions() as session, session.begin():
+            workspace = await self._locked_workspace(session, context)
+            _require_workspace_kind(workspace, _ELEMENT_KINDS)
+            await self._ensure_task_plan(session, context, workspace)
+            for paragraph_id, _ in normalized_links:
+                if await self._paragraph(session, workspace, paragraph_id, active=True) is None:
+                    raise WorkspaceError("LEGACY_REFERENCE_INVALID", "link paragraph is invalid")
+
+            element_ids: dict[str, int] = {}
+            for item in normalized:
+                result = await session.execute(
+                    insert(self._elements).values(
+                        script_build_id=workspace.script_build_id,
+                        branch_id=workspace.branch_id,
+                        base_ref_id=None,
+                        name=item["name"],
+                        dimension_primary=item["dimension_primary"],
+                        dimension_secondary=item["dimension_secondary"],
+                        commonality_analysis=item["commonality_analysis"],
+                        topic_support=item["topic_support"],
+                        weight_score=item["weight_score"],
+                        support_elements=item["support_elements"],
+                        is_active=True,
+                        created_at=_now(),
+                        updated_at=_now(),
+                    )
+                )
+                element_ids[item["client_key"]] = int(cast(Any, result).inserted_primary_key[0])
+
+            pairs = {
+                (paragraph_id, element_ids[client_key])
+                for paragraph_id, client_keys in normalized_links
+                for client_key in client_keys
+            }
+            for paragraph_id, element_id in sorted(pairs):
+                _authorize_link_write(context, paragraph_id, element_id, operation="create")
+                await session.execute(
+                    insert(self._links).values(
+                        script_build_id=workspace.script_build_id,
+                        branch_id=workspace.branch_id,
+                        paragraph_id=paragraph_id,
+                        element_id=element_id,
+                    )
+                )
+            return element_ids, len(pairs)
+
     async def update_element(
         self, context: CandidateWriteContext, *, element_id: int, values: Mapping[str, Any]
     ) -> None:

script_build_host/tests/test_phase_two_candidates.py

@@ -12,6 +12,7 @@ from agent.orchestration import ArtifactRef
 from script_build_host.application.phase_two_candidates import (
     PhaseTwoCandidateError,
     PhaseTwoCandidateService,
+    _candidate_state_revision,
 )
 from script_build_host.application.phase_two_inputs import (
     ActiveFrontierResolver,
@@ -49,6 +50,7 @@ from script_build_host.domain.task_contracts import (
     ScriptTaskContractV1,
     ScriptTaskKind,
 )
+from script_build_host.domain.workbench import paragraph_target_key, semantic_handle
 from script_build_host.domain.workspaces import CandidateWriteContext, WorkspaceError
 from script_build_host.infrastructure.canonical_json import canonical_sha256
 from script_build_host.repositories.sqlalchemy import (
@@ -330,6 +332,7 @@ def _scope_fixture(
         tasks={task_id: task},
         attempts={attempt_id: attempt},
         decisions={},
+        revision=1,
     )
     accepted = _AcceptedInputs(
         {task_id: contract},
@@ -431,6 +434,77 @@ async def test_paragraph_workspace_uses_positive_branch_and_freezes_on_manifest(
         )
 
 
+@pytest.mark.asyncio
+async def test_semantic_paragraph_save_is_atomic_and_exact_replay_is_idempotent(
+    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,
+    )
+    scope = await service._workspace_scope(context, paragraph=True)
+    before = await workspaces.snapshot(scope.write_context)
+    revision = _candidate_state_revision(scope, before, ledger.revision)
+    atom = [{"原子点": "observable reversal", "维度": "opening", "维度类型": "主维度"}]
+    payload = (
+        {
+            "client_key": "opening",
+            "paragraph_index": 1,
+            "name": "opening",
+            "content_range": {"scope": "opening"},
+            "level": 1,
+            "theme_elements": atom,
+            "form_elements": atom,
+            "function_elements": atom,
+            "feeling_elements": [{"原子点": "curiosity", "维度": "tone"}],
+            "theme": "A visible detail changes the original assumption.",
+            "form": "Concrete setup followed by a concise reversal.",
+            "function": "Give the audience a reason to continue.",
+            "feeling": "Measured surprise grounded in observation.",
+            "description": "The opening overturns one assumption with evidence.",
+            "full_description": (
+                "Start from a familiar expectation and show exactly how the observed "
+                "detail changes it."
+            ),
+        },
+    )
+    first = await service.save_script_paragraphs(
+        paragraphs=payload,
+        expected_state_revision=revision,
+        context=context,
+    )
+    replay = await service.save_script_paragraphs(
+        paragraphs=payload,
+        expected_state_revision=revision,
+        context=context,
+    )
+
+    assert first["status"] == "saved"
+    assert first["created"] == 1
+    assert replay["status"] == "not_modified"
+    assert replay["created"] == replay["updated"] == 0
+    assert replay["changed_target_keys"] == first["changed_target_keys"]
+    assert len((await workspaces.snapshot(scope.write_context)).paragraphs) == 1
+
+    changed_payload = (dict(payload[0], name="changed after stale projection"),)
+    with pytest.raises(PhaseTwoCandidateError, match="STALE_WORKBENCH_STATE"):
+        await service.save_script_paragraphs(
+            paragraphs=changed_payload,
+            expected_state_revision=revision,
+            context=context,
+        )
+
+
 @pytest.mark.asyncio
 async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(database: Any) -> None:
     _, sessions = database
@@ -454,6 +528,12 @@ async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(databa
                 "paragraph_index": 1,
                 "name": "opening",
                 "content_range": {"beats": ["setup", "turn"]},
+                "theme": "具体主题正文",
+                "form": "对话漫画",
+                "function": "建立冲突",
+                "feeling": "好奇",
+                "description": "角色在器械旁发生可见冲突。",
+                "full_description": "角色先指责别人占用器械,随后被发现自己也在刷手机。",
             },
             {
                 "client_key": "child",
@@ -468,6 +548,10 @@ async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(databa
     )
     assert created["created"] == 2
     assert set(created["paragraph_ids_by_client_key"]) == {"root", "child"}
+    workspace = await service.read_attempt_workspace(context=context)
+    root_row = next(item for item in workspace["paragraphs"] if item["name"] == "opening")
+    assert root_row["theme"] == "具体主题正文"
+    assert root_row["full_description"].endswith("自己也在刷手机。")
 
     contract2 = replace(contract, objective="second isolated workspace")
     (ledger2, accepted2), context2 = _scope_fixture(
@@ -514,6 +598,170 @@ async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(databa
     assert not (await workspaces.snapshot(write_context)).paragraphs
 
 
+@pytest.mark.asyncio
+async def test_element_set_materializes_all_accepted_paragraphs_before_linking(
+    database: Any,
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
+    paragraph = replace(
+        _complete_paragraph(),
+        theme_elements=(({"原子点": "内容", "维度": "主题", "维度类型": "主维度"}),),
+        form_elements=(({"原子点": "对话", "维度": "形式", "维度类型": "主维度"}),),
+        function_elements=(({"原子点": "冲突", "维度": "作用", "维度类型": "主维度"}),),
+        feeling_elements=(({"原子点": "好奇", "维度": "感受"}),),
+    )
+    _, paragraph_ref = await repository.freeze(
+        script_build_id=7,
+        task_id="paragraph-source",
+        attempt_id="paragraph-source-attempt",
+        spec_version=1,
+        artifact=ParagraphArtifactV1(
+            _lineage(),
+            (paragraph,),
+            (),
+            (),
+            {"created": {"paragraph_ids": [1]}},
+        ),
+    )
+    paragraph_ref = ArtifactRef(
+        paragraph_ref.uri,
+        paragraph_ref.kind,
+        paragraph_ref.version,
+        paragraph_ref.digest,
+    )
+    accepted_ref = AcceptedDecisionRef(
+        "paragraph-decision",
+        paragraph_ref,
+        SCOPE,
+        ScriptTaskKind.PARAGRAPH,
+    )
+    contract = ScriptTaskContractV1(
+        task_kind=ScriptTaskKind.ELEMENT_SET,
+        scope_ref=SCOPE,
+        intent_class=ScriptIntentClass.EXPLORE,
+        objective="attach concrete elements to every accepted paragraph",
+        input_decision_refs=(accepted_ref,),
+        base_artifact_ref=None,
+        write_scope=(
+            "script-build://writes/paragraphs",
+            "script-build://writes/elements",
+        ),
+        gap_ref=None,
+        output_schema="element-set-artifact/v1",
+        criteria=(ScriptCriterion("closed", "all elements are linked"),),
+        budget=ScriptTaskBudget(),
+        goal_ids=("goal-1",),
+    )
+    bundle = AcceptedInputBundleV1(
+        (
+            AcceptedInput(
+                "paragraph-decision",
+                paragraph_ref,
+                ScriptTaskKind.PARAGRAPH,
+                SCOPE,
+                "explicit",
+            ),
+        ),
+        DIGEST,
+    )
+    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=repository,
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=workspaces,
+    )
+
+    workspace = await service.read_attempt_workspace(context=context)
+    assert len(workspace["paragraphs"]) == 1
+    paragraph_id = workspace["paragraphs"][0]["paragraph_id"]
+    assert paragraph_id > 0
+    assert workspace["paragraphs"][0]["full_description"]
+
+    with pytest.raises(WorkspaceError, match="link paragraph"):
+        await service.create_script_elements(
+            elements=(
+                {
+                    "client_key": "invalid",
+                    "name": "不应落库的元素",
+                    "dimension_primary": "实质",
+                    "dimension_secondary": "回滚验证",
+                },
+            ),
+            links=(
+                {
+                    "paragraph_id": paragraph_id + 999,
+                    "element_client_keys": ["invalid"],
+                },
+            ),
+            context=context,
+        )
+    rolled_back = await service.read_attempt_workspace(context=context)
+    assert rolled_back["elements"] == []
+    assert rolled_back["paragraph_element_links"] == []
+
+    scope = await service._workspace_scope(context, element=True)
+    before = await workspaces.snapshot(scope.write_context)
+    revision = _candidate_state_revision(scope, before, ledger.revision)
+    target = paragraph_target_key("root", "task", "attempt", paragraph_id)
+    elements = (
+        {
+            "client_key": "machine",
+            "name": "被占用的高位下拉机",
+            "dimension_primary": "实质",
+            "dimension_secondary": "冲突道具",
+        },
+        {
+            "client_key": "bubble",
+            "name": "占着不练的对话气泡",
+            "dimension_primary": "形式",
+            "dimension_secondary": "视觉文字",
+        },
+    )
+    links = (
+        {
+            "paragraph_target_key": target,
+            "element_client_keys": ["machine", "bubble"],
+        },
+    )
+    created = await service.save_script_elements(
+        elements=elements,
+        links=links,
+        expected_state_revision=revision,
+        context=context,
+    )
+    replay = await service.save_script_elements(
+        elements=elements,
+        links=links,
+        expected_state_revision=revision,
+        context=context,
+    )
+    assert created["status"] == "saved"
+    assert created["created"] == 2
+    assert created["linked"] == 2
+    assert replay["status"] == "not_modified"
+    final = await workspaces.snapshot(scope.write_context)
+    assert len(final.elements) == 2
+    assert len(final.links) == 2
+
+    changed_elements = (dict(elements[0], name="changed after stale projection"), elements[1])
+    with pytest.raises(PhaseTwoCandidateError, match="STALE_WORKBENCH_STATE"):
+        await service.save_script_elements(
+            elements=changed_elements,
+            links=links,
+            expected_state_revision=revision,
+            context=context,
+        )
+
+    manifest = await service.resolve_attempt_manifest(context=context)
+    assert manifest.artifact_ref.kind == ArtifactKind.ELEMENT_SET.value
+
+
 @pytest.mark.asyncio
 async def test_candidate_service_discards_existing_draft_workspace_on_abandoned_attempt(
     database: Any,
@@ -605,11 +853,15 @@ async def test_frozen_image_must_be_accepted_and_digest_and_mime_are_rechecked()
         workspaces=None,
         raw_artifacts=_RawStore(content),
     )
-    images = await service.view_frozen_images(raw_artifact_refs=[raw_ref], context=context)
+    images = await service.view_frozen_images(
+        image_handles=[semantic_handle("image", raw_ref)], context=context
+    )
     assert images[0]["media_type"] == "image/png"
     assert "url" not in images[0]
     with pytest.raises(PhaseTwoCandidateError, match="outside the accepted"):
-        await service.view_frozen_images(raw_artifact_refs=[_RAW_OTHER], context=context)
+        await service.view_frozen_images(
+            image_handles=[semantic_handle("image", _RAW_OTHER)], context=context
+        )
 
 
 @pytest.mark.asyncio
@@ -700,8 +952,8 @@ async def test_comparison_candidates_are_contract_derived_and_pass_fairness_prec
         {
             "criterion_id": "closed",
             "candidate_results": [
-                {"artifact_ref": candidate_refs[0].uri, "reason": "uses one visible detail"},
-                {"artifact_ref": candidate_refs[1].uri, "reason": "remains more abstract"},
+                {"decision_id": "candidate-1", "reason": "uses one visible detail"},
+                {"decision_id": "candidate-2", "reason": "remains more abstract"},
             ],
         }
     ]
@@ -709,7 +961,7 @@ async def test_comparison_candidates_are_contract_derived_and_pass_fairness_prec
         payload={
             "criterion_results": criterion_results,
             "conflicts": ["both candidates write the opening scope"],
-            "recommendation": candidate_refs[0].uri,
+            "recommended_decision_id": "candidate-1",
         },
         context=context,
     )
@@ -717,7 +969,9 @@ async def test_comparison_candidates_are_contract_derived_and_pass_fairness_prec
     comparison = artifacts.frozen.artifact
     assert isinstance(comparison, ComparisonArtifactV1)
     assert comparison.candidate_artifact_refs == tuple(ref.uri for ref in candidate_refs)
-    assert comparison.criterion_results == tuple(criterion_results)
+    assert [
+        item["artifact_ref"] for item in comparison.criterion_results[0]["candidate_results"]
+    ] == [ref.uri for ref in candidate_refs]
 
     frozen_ref = ArtifactRef.from_dict(frozen["artifact_ref"])
     service._framework_artifact_store = _SnapshotArtifacts(frozen_ref)
@@ -736,7 +990,7 @@ async def test_comparison_candidates_are_contract_derived_and_pass_fairness_prec
             payload={
                 "candidate_artifact_refs": [candidate_refs[1].uri],
                 "criterion_results": criterion_results,
-                "recommendation": candidate_refs[1].uri,
+                "recommended_decision_id": "candidate-2",
             },
             context=context,
         )
@@ -810,9 +1064,7 @@ async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract
             elements=(element,),
             paragraph_element_links=(link,),
             source_artifact_refs=source_refs,
-            goal_coverage=(
-                GoalCoverage("goal-1", (source_refs[1], source_refs[2])),
-            ),
+            goal_coverage=(GoalCoverage("goal-1", (source_refs[1], source_refs[2])),),
             evidence_refs=(),
             acceptance_notes=("accepted",),
             canonical_sha256=DIGEST,
@@ -959,9 +1211,7 @@ async def test_candidate_portfolio_command_replay_reuses_one_frozen_version(data
                 elements=(element,),
                 paragraph_element_links=(link,),
                 source_artifact_refs=tuple(source_refs),
-                goal_coverage=(
-                    GoalCoverage("goal-1", (source_refs[1], source_refs[2])),
-                ),
+                goal_coverage=(GoalCoverage("goal-1", (source_refs[1], source_refs[2])),),
                 evidence_refs=(),
                 acceptance_notes=("accepted",),
             ),

script_build_host/tests/test_phase_two_workspace.py

@@ -354,7 +354,10 @@ async def test_wrong_adapter_cross_context_and_stale_base_are_rejected(database)
     _, sessions = database
     artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
     repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
-    context = _context("element-only")
+    context = replace(
+        _context("element-only"),
+        write_scope=("script-build://writes/elements",),
+    )
     await repository.get_or_create(context, artifact_kind=ArtifactKind.ELEMENT_SET)
     with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"):
         await repository.create_paragraph(