执行：确定性生成 Compose 与 Portfolio 审计产物

Mission Workbench 接管 Compose 和 CandidatePortfolio 的机械 closure、GoalCoverage 与 Artifact 生成，Coordinator 仍创建正式 Trace、Attempt 并调用一次独立 Validator。普通 Structure、Paragraph、Element Worker 保持模型创作职责，Compose 和 Portfolio 的 Worker provider 调用归零。

script_build_host/src/script_build_host/agents/presets.py

@@ -77,10 +77,9 @@ def register_script_presets() -> None:
             allowed_tools=[
                 "plan_script_tasks",
                 "decide_script_task",
-                "inspect_script_plan",
+                "read_workbench_detail",
                 "dispatch_script_tasks",
                 "validate_attempt",
-                "read_input_snapshot",
             ],
             denied_tools=[*_FORBIDDEN, "submit_attempt", "submit_validation"],
             max_iterations=100,
@@ -91,58 +90,44 @@ def register_script_presets() -> None:
     register_preset(
         "script_pattern_retrieval_worker",
         _worker(
-            ["read_input_snapshot", "query_pattern_qa"],
+            ["query_pattern_qa"],
             prompt=PATTERN_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_decode_retrieval_worker",
         _worker(
-            ["read_input_snapshot", "search_script_decode_case"],
+            ["search_script_decode_case"],
             prompt=DECODE_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_external_retrieval_worker",
         _worker(
-            ["read_input_snapshot", "external_search_case", "load_images"],
+            ["external_search_case", "load_images"],
             prompt=EXTERNAL_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_knowledge_retrieval_worker",
         _worker(
-            ["read_input_snapshot", "search_knowledge"],
+            ["search_knowledge"],
             prompt=KNOWLEDGE_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_direction_worker",
         _worker(
-            ["read_input_snapshot", "read_accepted_artifact", "save_direction_candidate"],
+            ["read_workbench_detail", "save_direction_candidate"],
             prompt=DIRECTION_WORKER_PROMPT,
         ),
     )
-    structure_writes = [
-        "read_attempt_workspace",
-        "create_script_paragraphs",
-    ]
-    paragraph_writes = [
-        "read_attempt_workspace",
-        "create_script_paragraphs",
-        "batch_update_script_paragraphs",
-    ]
-    element_writes = [
-        "read_attempt_workspace",
-        "create_script_element",
-        "update_script_element",
-        "batch_link_paragraph_elements",
-        "delete_paragraph_element_links",
-    ]
+    structure_writes = ["save_script_paragraphs"]
+    paragraph_writes = ["save_script_paragraphs"]
+    element_writes = ["save_script_elements"]
     common_candidate_reads = [
-        "read_input_snapshot",
+        "read_workbench_detail",
         "load_frozen_strategy",
-        "read_accepted_input_bundle",
         "view_frozen_images",
     ]
     register_preset(
@@ -169,28 +154,21 @@ def register_script_presets() -> None:
     register_preset(
         "script_candidate_compare_worker",
         _worker(
-            ["read_input_snapshot", "read_pinned_candidates", "save_comparison_candidate"],
+            ["read_workbench_detail", "save_comparison_candidate"],
             prompt=COMPARISON_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_compose_worker",
         _worker(
-            [
-                "read_active_frontier",
-                "save_structured_script_candidate",
-            ],
+            [],
             prompt=COMPOSE_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_candidate_portfolio_worker",
         _worker(
-            [
-                "read_input_snapshot",
-                "read_pinned_candidates",
-                "save_candidate_portfolio",
-            ],
+            [],
             prompt=PORTFOLIO_WORKER_PROMPT,
         ),
     )
@@ -198,8 +176,7 @@ def register_script_presets() -> None:
         "script_root_worker",
         _worker(
             [
-                "read_input_snapshot",
-                "read_accepted_portfolio",
+                "read_workbench_detail",
                 "legacy_projection_dry_run",
                 "save_root_delivery_manifest",
             ],
@@ -207,7 +184,6 @@ def register_script_presets() -> None:
         ),
     )
     validator_tools = [
-        "read_input_snapshot",
         "query_validation_evidence",
         "deterministic_precheck",
         "submit_validation",
@@ -251,8 +227,6 @@ def register_script_presets() -> None:
         AgentPreset(
             role=AgentRole.VALIDATOR,
             allowed_tools=[
-                "read_input_snapshot",
-                "read_accepted_portfolio",
                 "query_validation_evidence",
                 "deterministic_precheck",
                 "submit_validation",

script_build_host/src/script_build_host/application/mission_factory.py

@@ -7,7 +7,7 @@ from collections.abc import Sequence
 from hashlib import sha256
 from math import isfinite
 
-from agent import CompletionPolicy, RunConfig
+from agent import CompletionPolicy, NoProgressPolicy, RunConfig
 from agent.tools.builtin.knowledge import KnowledgeConfig
 
 from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
@@ -122,6 +122,7 @@ class ScriptMissionFactory:
             max_iterations=planner_config["max_iterations"],
             system_prompt=_frozen_prompt(snapshot, "script_planner"),
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            no_progress=NoProgressPolicy(progress_epoch="script-build-phase-one"),
             new_trace_id=binding.root_trace_id,
             root_task_spec=self.build_root_task_spec(snapshot),
             tool_groups=None,
@@ -241,6 +242,7 @@ class ScriptMissionFactory:
             temperature=planner_config["temperature"],
             max_iterations=planner_config["max_iterations"],
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            no_progress=NoProgressPolicy(progress_epoch="script-build-phase-two"),
             trace_id=binding.root_trace_id,
             root_task_spec=None,
             tool_groups=None,
@@ -315,6 +317,7 @@ class ScriptMissionFactory:
             temperature=planner_config["temperature"],
             max_iterations=planner_config["max_iterations"],
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            no_progress=NoProgressPolicy(progress_epoch="script-build-phase-three"),
             trace_id=binding.root_trace_id,
             root_task_spec=None,
             tool_groups=None,

script_build_host/src/script_build_host/application/workbench_workers.py

@@ -0,0 +1,106 @@
+"""Host-owned deterministic Workers for mechanical candidate assembly."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from agent.failures import ToolExecutionError
+from agent.orchestration import (
+    ArtifactRef,
+    DeterministicWorkerContext,
+    DeterministicWorkerResult,
+)
+
+from script_build_host.domain.errors import ScriptBuildError
+from script_build_host.domain.task_contracts import ScriptTaskKind, task_kind_from_context_refs
+from script_build_host.tools.failures import classify_script_tool_failure
+
+
+class ScriptBuildDeterministicWorker:
+    def __init__(self, candidates: Any, *, bindings: Any | None = None) -> None:
+        self._candidates = candidates
+        self._bindings = bindings
+        self.dispatch_guard: Any | None = None
+
+    async def supports(self, context: DeterministicWorkerContext) -> bool:
+        return task_kind_from_context_refs(context.task.current_spec.context_refs) in {
+            ScriptTaskKind.COMPOSE.value,
+            ScriptTaskKind.CANDIDATE_PORTFOLIO.value,
+        }
+
+    async def execute(self, context: DeterministicWorkerContext) -> DeterministicWorkerResult:
+        kind = task_kind_from_context_refs(context.task.current_spec.context_refs)
+        protected = {
+            "root_trace_id": context.root_trace_id,
+            "task_id": context.task.task_id,
+            "attempt_id": context.attempt.attempt_id,
+            "spec_version": context.task.current_spec_version,
+            "operation_id": context.operation_id,
+            "execution_epoch": context.execution_epoch,
+        }
+        try:
+            result = await self._execute_fenced(kind, protected)
+            if isinstance(result, ArtifactRef):
+                return DeterministicWorkerResult(
+                    summary=f"Host resumed deterministic {kind} Artifact",
+                    artifact_refs=(result,),
+                )
+            if not isinstance(result, dict):
+                raise RuntimeError("deterministic candidate result has an invalid shape")
+            return DeterministicWorkerResult(
+                summary=f"Host deterministically assembled {kind}",
+                artifact_refs=(ArtifactRef.from_dict(dict(result["artifact_ref"])),),
+            )
+        except ToolExecutionError:
+            raise
+        except ScriptBuildError as exc:
+            raise ToolExecutionError(
+                classify_script_tool_failure(
+                    exc,
+                    source_tool=f"deterministic_{str(kind).replace('-', '_')}",
+                    context=protected,
+                )
+            ) from exc
+
+    async def _execute_fenced(self, kind: str | None, context: dict[str, object]) -> object:
+        async def mutation() -> object:
+            existing = await self._existing(context)
+            if existing is not None:
+                return existing
+            if kind == ScriptTaskKind.COMPOSE.value:
+                return await self._candidates.save_structured_script_candidate(
+                    acceptance_notes=(
+                        "Host assembled the frozen creative frontier; "
+                        "semantic acceptance is independent."
+                    ),
+                    context=context,
+                )
+            if kind == ScriptTaskKind.CANDIDATE_PORTFOLIO.value:
+                return await self._candidates.save_candidate_portfolio(
+                    payload={"unresolved_defects": [], "superseded_decision_ids": []},
+                    context=context,
+                )
+            raise RuntimeError("unsupported deterministic script worker")
+
+        guard = self.dispatch_guard
+        if guard is None or self._bindings is None:
+            return await mutation()
+        binding = await self._bindings.get_by_root(context["root_trace_id"])
+        await guard.ensure_dispatch_allowed(binding.script_build_id)
+        return await guard.execute_fenced(binding.script_build_id, mutation)
+
+    async def _existing(self, context: dict[str, object]) -> ArtifactRef | None:
+        try:
+            manifest = await self._candidates.resolve_attempt_manifest(context=context)
+        except ScriptBuildError as exc:
+            if getattr(exc, "code", "") in {
+                "ARTIFACT_NOT_FOUND",
+                "LEGACY_ARTIFACT_NOT_FOUND",
+                "MISSION_RECOVERY_REQUIRED",
+            }:
+                return None
+            raise
+        return manifest.artifact_ref
+
+
+__all__ = ["ScriptBuildDeterministicWorker"]

script_build_host/src/script_build_host/composition.py

@@ -32,6 +32,7 @@ from script_build_host.application.mission_service import (
     DirectionReconciler,
     ScriptMissionService,
 )
+from script_build_host.application.mission_workbench import MissionWorkbench
 from script_build_host.application.phase_three import PhaseThreeContinuationService
 from script_build_host.application.phase_two_boundary import ScriptPhaseTwoBoundaryVerifier
 from script_build_host.application.phase_two_candidates import PhaseTwoCandidateService
@@ -47,6 +48,7 @@ from script_build_host.application.recovery import (
     PhaseThreeRecoveryService,
 )
 from script_build_host.application.root_delivery import RootDeliveryService
+from script_build_host.application.workbench_workers import ScriptBuildDeterministicWorker
 from script_build_host.domain.ports import (
     BuildAuthorizer,
     InputSnapshotRepository,
@@ -114,6 +116,7 @@ class HostComposition:
     tool_gateway: LegacyScriptToolGateway
     phase_two_planning: PhaseTwoPlanningService
     phase_two_candidates: PhaseTwoCandidateService
+    workbench: MissionWorkbench
     phase_three: PhaseThreeContinuationService | None = None
     finalization: PhaseThreeFinalizationService | None = None
     recovery: PhaseThreeRecoveryService | None = None
@@ -183,6 +186,17 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         task_store=dependencies.task_store,
         artifacts=dependencies.business_artifacts,
     )
+    workbench = MissionWorkbench(
+        task_store=dependencies.task_store,
+        bindings=dependencies.bindings,
+        snapshots=dependencies.input_snapshots,
+        artifacts=dependencies.business_artifacts,
+        contracts=contract_store,
+        candidates=candidate_service,
+    )
+    deterministic_worker = ScriptBuildDeterministicWorker(
+        candidate_service, bindings=dependencies.bindings
+    )
     coordinator = wire_orchestration(
         dependencies.runner,
         dependencies.task_store,
@@ -197,6 +211,9 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         evidence_provider=evidence_provider,
         role_run_config_resolver=model_resolver,
         role_system_prompt_resolver=prompt_resolver,
+        task_context_provider=workbench,
+        role_context_provider=workbench,
+        deterministic_worker=deterministic_worker,
     )
     planning_service = PhaseTwoPlanningService(
         coordinator=coordinator,
@@ -238,6 +255,7 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         fenced_command_gate=dependencies.fenced_command_gate,
     )
     planning_service.dispatch_guard = mission_service
+    deterministic_worker.dispatch_guard = mission_service
     gateway = LegacyScriptToolGateway(
         bindings=dependencies.bindings,
         snapshots=dependencies.input_snapshots,
@@ -250,6 +268,7 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         candidate_tools=candidate_service,
         budget_guard=planning_service,
         root_tools=root_delivery,
+        workbench=workbench,
     )
     register_script_tools(dependencies.runner.tools, gateway)
     security = ApiSecurity(
@@ -308,6 +327,7 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
             artifacts=dependencies.business_artifacts,
             legacy_projection=dependencies.legacy_projection,
             mission_service=mission_service,
+            contract_reader=planning_service,
         )
     app = create_app(
         mission_service=mission_service,
@@ -337,6 +357,7 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         tool_gateway=gateway,
         phase_two_planning=planning_service,
         phase_two_candidates=candidate_service,
+        workbench=workbench,
         phase_three=phase_three,
         finalization=finalization,
         recovery=recovery,

script_build_host/tests/test_phase_two_real_runner_e2e.py

@@ -156,7 +156,10 @@ def _contract(
         assert base_decision_id is not None
     return {
         "task_kind": task_kind,
-        "scope_ref": scope_ref,
+        "scope_selector": {
+            "anchor": "mission",
+            "path": scope_ref.removeprefix("script-build://scopes/").split("/")[:4],
+        },
         "intent_class": (
             "compose"
             if task_kind == "compose"
@@ -167,19 +170,17 @@ def _contract(
         "objective": objective or f"produce one concrete and independently verifiable {task_kind}",
         "input_decision_ids": input_decision_ids,
         "base_decision_id": base_decision_id,
-        "gap_ref": None,
+        "gap_key": None,
         "criteria": [
             {
-                "criterion_id": f"{task_kind}-closed",
+                "client_key": f"{task_kind}-closed",
                 "description": "the immutable output is concrete and causally closed",
                 "hard": True,
             }
         ],
         "goal_ids": list(goal_ids or ()),
         "supersedes_decision_ids": list(supersedes_decision_ids),
-        "comparison_decision_ids": [
-            str(item["decision_id"]) for item in comparison_decision_refs
-        ],
+        "comparison_decision_ids": [str(item["decision_id"]) for item in comparison_decision_refs],
     }
 
 
@@ -511,6 +512,35 @@ class _PhaseTwoScriptedLLM:
                 },
             )
 
+        # Compose is now a deterministic Workbench Worker. A complete creative
+        # frontier produces one valid artifact instead of asking an LLM Worker
+        # to manufacture an invalid first candidate for this fixture.
+        if compose.status is TaskStatus.COMPLETED:
+            compose_decision, _compose_ref = _accepted_ref(ledger, compose)
+            if portfolio.current_spec_version == 1:
+                return self._call(
+                    "decide_script_task",
+                    {
+                        "task_id": portfolio.task_id,
+                        "action": "revise",
+                        "reason": "pin the uniquely adopted StructuredScript",
+                        "selected_decision_ids": [compose_decision],
+                    },
+                )
+            if portfolio.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
+                return self._call("dispatch_script_tasks", {"task_ids": [portfolio.task_id]})
+            if portfolio.status is TaskStatus.AWAITING_DECISION:
+                return self._accept(portfolio, ledger)
+            assert portfolio.status is TaskStatus.COMPLETED
+            return self._call(
+                "decide_script_task",
+                {
+                    "task_id": root.task_id,
+                    "action": "block",
+                    "reason": PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
+                },
+            )
+
         replacement = by_objective(replacement_objective)
         if replacement is None:
             raise AssertionError("failed Compose must SPLIT one bounded replacement Task")
@@ -661,11 +691,8 @@ class _PhaseTwoScriptedLLM:
             return self._call("submit_attempt", {})
         if kind == "direction":
             if not isinstance(last, dict) or "artifact_ref" not in last:
-                accepted = cast(Sequence[Mapping[str, Any]], prompt["accepted_child_results"])
-                evidence_refs = cast(
-                    Sequence[Mapping[str, Any]],
-                    cast(Mapping[str, Any], accepted[0]["submission"])["evidence_refs"],
-                )
+                role_context = cast(Mapping[str, Any], prompt["role_context"])
+                accepted = cast(Sequence[Mapping[str, Any]], role_context["accepted_inputs"])
                 return self._call(
                     "save_direction_candidate",
                     {
@@ -681,22 +708,24 @@ class _PhaseTwoScriptedLLM:
                         ],
                         "constraints": [
                             {
-                                "constraint_id": "grounded-direction",
+                                "client_key": "grounded-direction",
                                 "statement": "the direction is concrete and testable",
                                 "rationale": "accepted evidence must remain traceable",
                             }
                         ],
                         "preferences": [],
-                        "evidence_refs": [evidence_refs[0]["uri"]],
+                        "strategy_handles": [],
+                        "evidence_decision_ids": [accepted[0]["decision_id"]],
                     },
                 )
             goal_map = cast(Mapping[str, str], last["goal_ids_by_client_key"])
             self.direction_goal_id = goal_map["observable-opening"]
             return self._call("submit_attempt", {})
         if kind == "structure":
-            if not isinstance(last, dict):
+            if not isinstance(last, dict) or last.get("status") != "saved":
+                role_context = cast(Mapping[str, Any], prompt["role_context"])
                 return self._call(
-                    "create_script_paragraphs",
+                    "save_script_paragraphs",
                     {
                         "paragraphs": [
                             {
@@ -707,6 +736,7 @@ class _PhaseTwoScriptedLLM:
                                     "scope": "opening",
                                     "beats": ["setup", "reversal"],
                                 },
+                                "level": 1,
                                 "theme_elements": [
                                     {
                                         "原子点": "assumption changes after evidence",
@@ -732,58 +762,76 @@ class _PhaseTwoScriptedLLM:
                                     {"原子点": "measured surprise", "维度": "tone"}
                                 ],
                             }
-                        ]
+                        ],
+                        "expected_state_revision": role_context["state_revision"],
                     },
                 )
             return self._call("submit_attempt", {})
         if kind == "paragraph":
-            if isinstance(last, dict) and last.get("updated") == 1:
+            if isinstance(last, dict) and last.get("status") == "saved":
                 return self._call("submit_attempt", {})
-            if not isinstance(last, dict) or "paragraphs" not in last:
-                return self._call("read_attempt_workspace", {})
-            paragraphs = cast(Sequence[Mapping[str, Any]], last["paragraphs"])
-            assert len(paragraphs) == 1
+            role_context = cast(Mapping[str, Any], prompt["role_context"])
+            paragraphs = cast(Sequence[Mapping[str, Any]], role_context["paragraph_targets"])
             replacement = "replace the placeholder" in str(spec["objective"])
+            paragraph: dict[str, Any] = {
+                "theme": (
+                    "The retrieved observable reversal changes the opening claim."
+                    if replacement
+                    else "A familiar expectation changes after a visible detail."
+                ),
+                "form": "A concrete setup followed by one concise reversal.",
+                "function": "Give the audience an immediate reason to continue.",
+                "feeling": "Measured surprise grounded in observation.",
+                "description": "The opening overturns one assumption with evidence.",
+                "full_description": (
+                    "Use the accepted retrieval evidence to replace the unrealized opening and "
+                    "show exactly how the observed detail changes the interpretation."
+                    if replacement
+                    else "Start from the familiar expectation, name the visible detail, and show "
+                    "exactly how that detail changes the interpretation."
+                ),
+            }
+            if paragraphs:
+                paragraph["paragraph_target_key"] = paragraphs[0]["paragraph_target_key"]
+            else:
+                paragraph.update(
+                    {
+                        "client_key": "opening",
+                        "paragraph_index": 1,
+                        "name": "Opening",
+                        "content_range": {"scope": "opening", "beats": ["setup", "reversal"]},
+                        "level": 1,
+                        "theme_elements": [
+                            {"原子点": "visible reversal", "维度": "claim", "维度类型": "主维度"}
+                        ],
+                        "form_elements": [
+                            {"原子点": "setup and reversal", "维度": "shape", "维度类型": "主维度"}
+                        ],
+                        "function_elements": [
+                            {"原子点": "create curiosity", "维度": "job", "维度类型": "主维度"}
+                        ],
+                        "feeling_elements": [{"原子点": "measured surprise", "维度": "tone"}],
+                    }
+                )
             return self._call(
-                "batch_update_script_paragraphs",
+                "save_script_paragraphs",
                 {
-                    "updates": [
-                        {
-                            "paragraph_id": paragraphs[0]["paragraph_id"],
-                            "theme": (
-                                "The retrieved observable reversal changes the opening claim."
-                                if replacement
-                                else "A familiar expectation changes after a visible detail."
-                            ),
-                            "form": "A concrete setup followed by one concise reversal.",
-                            "function": "Give the audience an immediate reason to continue.",
-                            "feeling": "Measured surprise grounded in observation.",
-                            "description": "The opening overturns one assumption with evidence.",
-                            "full_description": (
-                                "Use the accepted retrieval evidence to replace the unrealized "
-                                "opening and show exactly how the observed detail changes the "
-                                "interpretation."
-                                if replacement
-                                else "Start from the familiar expectation, name the visible "
-                                "detail, and show exactly how that detail changes the "
-                                "interpretation."
-                            ),
-                        }
-                    ]
+                    "paragraphs": [paragraph],
+                    "expected_state_revision": role_context["state_revision"],
                 },
             )
         if kind == "element-set":
-            if isinstance(last, dict) and last.get("linked") == 1:
+            if isinstance(last, dict) and last.get("created") == 1 and last.get("linked") == 1:
                 return self._call("submit_attempt", {})
-            if not isinstance(last, dict):
-                return self._call("read_attempt_workspace", {})
-            if "paragraphs" in last and "elements" in last:
-                paragraphs = cast(Sequence[Mapping[str, Any]], last["paragraphs"])
-                elements = cast(Sequence[Mapping[str, Any]], last["elements"])
-                if not elements:
-                    return self._call(
-                        "create_script_element",
+            role_context = cast(Mapping[str, Any], prompt["role_context"])
+            paragraphs = cast(Sequence[Mapping[str, Any]], role_context["paragraph_targets"])
+            assert len(paragraphs) == 1
+            return self._call(
+                "save_script_elements",
+                {
+                    "elements": [
                         {
+                            "client_key": "reversal",
                             "name": "Observable reversal",
                             "dimension_primary": "实质",
                             "dimension_secondary": "opening-evidence",
@@ -792,35 +840,23 @@ class _PhaseTwoScriptedLLM:
                             },
                             "topic_support": {"strength": "high"},
                             "weight_score": {"value": 0.9},
-                            "support_elements": [
-                                {"ref": "decode-index://fixture/opening-reversal"}
-                            ],
-                        },
-                    )
-                assert len(paragraphs) == 1 and len(elements) == 1
-                return self._call(
-                    "batch_link_paragraph_elements",
-                    {
-                        "links": [
-                            {
-                                "paragraph_id": paragraphs[0]["paragraph_id"],
-                                "element_ids": [elements[0]["element_id"]],
-                            }
-                        ]
-                    },
-                )
-            if "element_id" in last:
-                return self._call("read_attempt_workspace", {})
+                            "support_elements": [{"source_handle": "accepted-opening"}],
+                        }
+                    ],
+                    "links": [
+                        {
+                            "paragraph_target_key": paragraphs[0]["paragraph_target_key"],
+                            "element_client_keys": ["reversal"],
+                        }
+                    ],
+                    "expected_state_revision": role_context["state_revision"],
+                },
+            )
         if kind == "compare":
             if isinstance(last, dict) and "artifact_ref" in last:
                 return self._call("submit_attempt", {})
-            if not isinstance(last, list):
-                return self._call("read_pinned_candidates", {})
-            candidates = [
-                str(item["artifact_ref"]["uri"])
-                for item in last
-                if item["task_kind"] == "paragraph"
-            ]
+            role_context = cast(Mapping[str, Any], prompt["role_context"])
+            candidates = list(role_context["task"]["comparison_decision_ids"])
             assert len(candidates) == 2
             return self._call(
                 "save_comparison_candidate",
@@ -830,7 +866,7 @@ class _PhaseTwoScriptedLLM:
                             "criterion_id": "compare-closed",
                             "candidate_results": [
                                 {
-                                    "artifact_ref": candidate,
+                                    "decision_id": candidate,
                                     "reason": "reviewed against the same frozen criterion",
                                 }
                                 for candidate in candidates
@@ -840,38 +876,11 @@ class _PhaseTwoScriptedLLM:
                     "conflicts": [
                         "candidate B realizes the retrieved evidence while candidate A does not"
                     ],
-                    "recommendation": candidates[-1],
-                },
-            )
-        if kind == "compose":
-            if isinstance(last, dict) and "artifact_ref" in last:
-                return self._call("submit_attempt", {})
-            if not isinstance(last, dict) or last.get("schema_version") != "active-frontier/v1":
-                return self._call("read_active_frontier", {})
-            self.compose_candidates_saved += 1
-            return self._call(
-                "save_structured_script_candidate",
-                {
-                    "acceptance_notes": [
-                        "TODO replace this placeholder opening with retrieved evidence."
-                        if self.compose_candidates_saved == 1
-                        else "The adopted replacement is concrete and fully rendered."
-                    ]
-                },
-            )
-        if kind == "candidate-portfolio":
-            if isinstance(last, dict) and "artifact_ref" in last:
-                return self._call("submit_attempt", {})
-            if not isinstance(last, list):
-                return self._call("read_pinned_candidates", {})
-            assert sorted(item["task_kind"] for item in last) == ["compose", "direction"]
-            return self._call(
-                "save_candidate_portfolio",
-                {
-                    "superseded_decision_ids": [],
-                    "unresolved_defects": [],
+                    "recommended_decision_id": candidates[-1],
                 },
             )
+        if kind in {"compose", "candidate-portfolio"}:
+            raise AssertionError(f"{kind} must be executed by DeterministicWorker")
         raise AssertionError(f"unexpected Worker kind {kind}; tools={sorted(names)}")
 
     def _validator(self, messages) -> dict[str, Any]:
@@ -885,11 +894,10 @@ class _PhaseTwoScriptedLLM:
         if not isinstance(last, dict) or "rule_results" not in last:
             return self._call("deterministic_precheck", {})
         criteria = cast(Sequence[Mapping[str, Any]], spec["acceptance_criteria"])
+        role_context = cast(Mapping[str, Any], prompt["role_context"])
+        evidence_handle = role_context["evidence_handles"][0]["evidence_handle"]
         failed_rules = [item for item in last["rule_results"] if item["verdict"] != "passed"]
         if failed_rules:
-            artifact_snapshot = cast(Mapping[str, Any], prompt["artifact_snapshot"])
-            refs = cast(Sequence[Mapping[str, Any]], artifact_snapshot["artifact_refs"])
-            assert len(refs) == 1
             return self._call(
                 "submit_validation",
                 {
@@ -906,11 +914,14 @@ class _PhaseTwoScriptedLLM:
                         {
                             "defect_code": "REALIZATION_PLACEHOLDER",
                             "criterion_id": criteria[0]["criterion_id"],
-                            "scope_ref": "script-build://scopes/full/opening",
+                            "scope_selector": {
+                                "anchor": "mission",
+                                "path": ["full", "opening"],
+                            },
                             "observed_excerpt": (
                                 "TODO replace this placeholder opening with retrieved evidence."
                             ),
-                            "evidence_refs": [dict(refs[0])],
+                            "evidence_handle_ids": [evidence_handle],
                             "severity": "hard",
                             "invalidated_inputs": [],
                             "recommended_action_class": "split",
@@ -920,12 +931,6 @@ class _PhaseTwoScriptedLLM:
                 },
             )
         assert all(item["verdict"] == "passed" for item in last["rule_results"])
-        artifact_snapshot = cast(Mapping[str, Any], prompt["artifact_snapshot"])
-        evidence_refs = [
-            *cast(Sequence[Mapping[str, Any]], artifact_snapshot["artifact_refs"]),
-            *cast(Sequence[Mapping[str, Any]], artifact_snapshot["evidence_refs"]),
-        ]
-        assert evidence_refs
         return self._call(
             "submit_validation",
             {
@@ -935,7 +940,7 @@ class _PhaseTwoScriptedLLM:
                         "criterion_id": item["criterion_id"],
                         "verdict": "passed",
                         "reason": "the immutable artifact passed deterministic and semantic review",
-                        "evidence_refs": [dict(evidence_refs[0])],
+                        "evidence_handle_ids": [evidence_handle],
                     }
                     for item in criteria
                 ],
@@ -989,9 +994,7 @@ class _PhaseThreeScriptedLLM(_PhaseTwoScriptedLLM):
                 "save_root_delivery_manifest",
                 {"build_summary": "final summary", "blocking_defects": []},
             )
-        if isinstance(last, dict) and "direction_ref" in last:
-            return self._call("legacy_projection_dry_run", {"build_summary": "final summary"})
-        return self._call("read_accepted_portfolio", {})
+        return self._call("legacy_projection_dry_run", {"build_summary": "final summary"})
 
 
 class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
@@ -1267,119 +1270,6 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
             },
         )
 
-    def _worker(self, messages, names: set[str]) -> dict[str, Any]:
-        prompt = _role_prompt(messages)
-        spec = cast(Mapping[str, Any], prompt["task_spec"])
-        refs = cast(Sequence[str], spec["context_refs"])
-        kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
-        last = _last_tool_value(messages)
-        if kind == "paragraph":
-            assert "create_script_paragraph" not in names
-            assert "create_script_paragraphs" in names
-            assert "create_script_element" not in names
-            if kind not in self.worker_kinds:
-                self.worker_kinds.append(kind)
-            if isinstance(last, dict) and last.get("updated") == 1:
-                return self._call("submit_attempt", {})
-            if not isinstance(last, dict) or "paragraphs" not in last:
-                return self._call("read_attempt_workspace", {})
-            paragraphs = cast(Sequence[Mapping[str, Any]], last["paragraphs"])
-            if not paragraphs:
-                return self._call(
-                    "create_script_paragraphs",
-                    {
-                        "paragraphs": [
-                            {
-                                "client_key": "opening",
-                                "paragraph_index": 1,
-                                "name": "Exploratory opening",
-                                "content_range": {
-                                    "scope": "opening",
-                                    "beats": ["setup", "reversal"],
-                                },
-                                "theme_elements": [
-                                    {"原子点": "visible reversal", "维度": "claim"}
-                                ],
-                                "form_elements": [
-                                    {"原子点": "setup and reversal", "维度": "shape"}
-                                ],
-                                "function_elements": [
-                                    {"原子点": "create curiosity", "维度": "job"}
-                                ],
-                                "feeling_elements": [
-                                    {"原子点": "measured surprise", "维度": "tone"}
-                                ],
-                            }
-                        ]
-                    },
-                )
-            return self._call(
-                "batch_update_script_paragraphs",
-                {
-                    "updates": [
-                        {
-                            "paragraph_id": paragraphs[0]["paragraph_id"],
-                            "theme": "One visible detail overturns the familiar assumption.",
-                            "form": "Concrete setup followed by a concise reversal.",
-                            "function": "Give the audience a grounded reason to continue.",
-                            "feeling": "Measured surprise.",
-                            "description": "A complete opening increment.",
-                            "full_description": (
-                                "The opening names the visible detail and explains exactly how "
-                                "it changes the audience's initial interpretation."
-                            ),
-                        }
-                    ]
-                },
-            )
-        if kind == "element-set":
-            assert "create_script_element" in names
-            assert "batch_link_paragraph_elements" in names
-            assert "create_script_paragraph" not in names
-            if kind not in self.worker_kinds:
-                self.worker_kinds.append(kind)
-            if isinstance(last, dict) and "artifact_ref" in last:
-                return self._call("submit_attempt", {})
-            if isinstance(last, dict) and last.get("linked") == 1:
-                return self._call("submit_attempt", {})
-            if not isinstance(last, dict):
-                return self._call(
-                    "create_script_element",
-                    {
-                        "name": "Observable reversal",
-                        "dimension_primary": "实质",
-                        "dimension_secondary": "opening evidence",
-                        "commonality_analysis": {"summary": "visible detail changes belief"},
-                        "topic_support": {"strength": "high"},
-                        "weight_score": {"value": 0.9},
-                        "support_elements": [{"ref": "decode-index://fixture/opening-reversal"}],
-                    },
-                )
-            if "element_id" in last:
-                return self._call("read_attempt_workspace", {})
-            if "paragraphs" in last and "elements" in last:
-                paragraphs = cast(Sequence[Mapping[str, Any]], last["paragraphs"])
-                elements = cast(Sequence[Mapping[str, Any]], last["elements"])
-                assert len(elements) == 1
-                if not paragraphs:
-                    return self._call("submit_attempt", {})
-                return self._call(
-                    "batch_link_paragraph_elements",
-                    {
-                        "links": [
-                            {
-                                "paragraph_id": paragraphs[0]["paragraph_id"],
-                                "element_ids": [elements[0]["element_id"]],
-                            }
-                        ]
-                    },
-                )
-        if kind == "structure":
-            assert "create_script_paragraph" not in names
-            assert "create_script_paragraphs" in names
-            assert "create_script_element" not in names
-        return super()._worker(messages, names)
-
 
 class _PrincipalProvider:
     async def current(self, request_context: Any = None) -> Principal:
@@ -1593,7 +1483,11 @@ async def test_real_runner_creative_exploration_order_does_not_choose_adoption_o
         assert attempt.worker_preset == expected_presets[kind]
         worker_trace = await reopened_traces.get_trace(attempt.worker_trace_id)
         assert worker_trace is not None and worker_trace.status == "completed"
-        assert await reopened_traces.get_trace_messages(attempt.worker_trace_id)
+        messages = await reopened_traces.get_trace_messages(attempt.worker_trace_id)
+        if worker_trace.context.get("deterministic_worker"):
+            assert messages == []
+        else:
+            assert messages
         validation = next(
             ledger.validations[item]
             for item in task.validation_ids
@@ -1746,9 +1640,6 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         "structure",
         "paragraph",
         "element-set",
-        "compose",
-        "compare",
-        "candidate-portfolio",
     ]
     assert llm.validator_kinds == [
         "decode-retrieval",
@@ -1756,13 +1647,12 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         "structure",
         "paragraph",
         "element-set",
-        "compare",
         "compose",
         "candidate-portfolio",
     ]
     assert llm.submit_attempt_arguments
     assert all(not item for item in llm.submit_attempt_arguments)
-    assert llm.compose_candidates_saved == 2
+    assert llm.compose_candidates_saved == 0
     assert llm.tool_calls_by_name.get("create_script_paragraph", 0) == 0
 
     compose_task = next(
@@ -1770,15 +1660,12 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         for task in ledger.tasks.values()
         if task.task_id != ledger.root_task_id and _task_kind(task) == "compose"
     )
-    assert compose_task.current_spec_version == 3
-    assert len(compose_task.attempt_ids) == 2
+    assert compose_task.current_spec_version == 2
+    assert len(compose_task.attempt_ids) == 1
     assert [ledger.validations[item].verdict for item in compose_task.validation_ids] == [
-        ValidationVerdict.FAILED,
         ValidationVerdict.PASSED,
     ]
     assert [ledger.decisions[item].action for item in compose_task.decision_ids] == [
-        DecisionAction.REVISE,
-        DecisionAction.SPLIT,
         DecisionAction.REVISE,
         DecisionAction.ACCEPT,
     ]
@@ -1788,21 +1675,7 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         for task in ledger.tasks.values()
         if task.task_id != ledger.root_task_id and _task_kind(task) == "paragraph"
     ]
-    assert len(paragraph_tasks) == 2
-    replacement = next(
-        task for task in paragraph_tasks if "replace the placeholder" in task.current_spec.objective
-    )
-    phase_two_retrieval = next(
-        task
-        for task in ledger.tasks.values()
-        if task.task_id != ledger.root_task_id
-        and _task_kind(task) == "decode-retrieval"
-        and task.parent_task_id == replacement.task_id
-    )
-    replacement_attempt = ledger.attempts[replacement.attempt_ids[-1]]
-    assert replacement_attempt.accepted_child_decision_ids == (
-        phase_two_retrieval.decision_ids[-1],
-    )
+    assert len(paragraph_tasks) == 1
     assert (
         len(
             [
@@ -1811,7 +1684,7 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
                 if task.task_id != ledger.root_task_id and _task_kind(task) == "compare"
             ]
         )
-        == 1
+        == 0
     )
 
     expected_identities = {
@@ -1820,6 +1693,13 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
     for attempt in ledger.attempts.values():
         worker_trace = await trace_store.get_trace(attempt.worker_trace_id)
         assert worker_trace is not None
+        if worker_trace.context.get("deterministic_worker"):
+            assert worker_trace.total_tokens == 0
+            assert attempt.worker_preset in {
+                "script_compose_worker",
+                "script_candidate_portfolio_worker",
+            }
+            continue
         assert (
             worker_trace.context["role_prompt_identity"]
             == expected_identities[attempt.worker_preset]
@@ -1883,7 +1763,11 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
     for attempt in phase_two_attempts:
         worker_trace = await reopened_traces.get_trace(attempt.worker_trace_id)
         assert worker_trace is not None and worker_trace.status == "completed"
-        assert await reopened_traces.get_trace_messages(attempt.worker_trace_id)
+        messages = await reopened_traces.get_trace_messages(attempt.worker_trace_id)
+        if worker_trace.context.get("deterministic_worker"):
+            assert messages == []
+        else:
+            assert messages
         validation = next(
             ledger.validations[item]
             for item in ledger.tasks[attempt.task_id].validation_ids
@@ -1928,7 +1812,6 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
     assert {
         ArtifactKind.DIRECTION.value,
         ArtifactKind.PARAGRAPH.value,
-        ArtifactKind.COMPARISON.value,
         ArtifactKind.STRUCTURED_SCRIPT.value,
         ArtifactKind.CANDIDATE_PORTFOLIO.value,
     } <= kinds