主机：修正真实 E2E 报告和离线 Runner 夹具

E2E 报告区分 current_domain_error、last_recovered_error 与历史错误计数，纳入 Planner token 和工具统计；更新 scripted runner 以覆盖新 Direction、Compose、Element 和 Root 验收链路。

script_build_host/src/script_build_host/internal_e2e.py

@@ -591,6 +591,36 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
     task_seconds: dict[str, float] = {}
     preset_usage: dict[str, dict[str, int | float]] = {}
     domain_errors: list[dict[str, str]] = []
+    trace_presets = {binding.root_trace_id: "script_planner"}
+    trace_presets.update(
+        {item.worker_trace_id: item.worker_preset for item in attempts}
+    )
+    trace_presets.update(
+        {item.validator_trace_id: item.validator_preset for item in validations}
+    )
+    traces_with_usage: set[str] = set()
+    usage_reader = getattr(
+        host.composition.mission_service.runner.trace_store, "get_model_usage", None
+    )
+    if callable(usage_reader):
+        for trace_id, preset in trace_presets.items():
+            model_usage = await usage_reader(trace_id)
+            models = model_usage.get("models", []) if isinstance(model_usage, dict) else []
+            agent_models = [
+                item
+                for item in models
+                if isinstance(item, dict) and item.get("source") == "agent"
+            ]
+            if agent_models:
+                traces_with_usage.add(trace_id)
+            usage = preset_usage.setdefault(str(preset), _empty_preset_usage())
+            usage["agent_runs"] += 1
+            for model in agent_models:
+                usage["calls"] += int(model.get("call_count") or 0)
+                usage["prompt_tokens"] += int(model.get("prompt_tokens") or 0)
+                usage["completion_tokens"] += int(model.get("completion_tokens") or 0)
+                usage["cache_read_tokens"] += int(model.get("cache_read_tokens") or 0)
+                usage["tokens"] += int(model.get("total_tokens") or 0)
     for item in [*attempts, *validations]:
         task = ledger.tasks[item.task_id]
         kind = next(
@@ -608,19 +638,32 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
             key = getattr(item, "worker_preset", None) or getattr(
                 item, "validator_preset", "unknown"
             )
-            usage = preset_usage.setdefault(
-                str(key),
-                {"calls": 0, "tokens": 0, "timeouts": 0, "failures": 0, "seconds": 0.0},
+            usage = preset_usage.setdefault(str(key), _empty_preset_usage())
+            usage["stage_runs"] += 1
+            trace_id = getattr(item, "worker_trace_id", None) or getattr(
+                item, "validator_trace_id", ""
             )
-            usage["calls"] += 1
-            usage["tokens"] += int(stats.total_tokens or 0)
+            if trace_id not in traces_with_usage:
+                usage["calls"] += 1
+                usage["tokens"] += int(stats.total_tokens or 0)
             usage["seconds"] += (item.duration_ms or 0) / 1000
-            failure_code = str(stats.failure_code or "")
+            failure = getattr(item, "failure", None)
+            failure_code = str(
+                getattr(failure, "code", "")
+                or getattr(stats.failure_code, "value", stats.failure_code)
+                or ""
+            )
             if failure_code:
                 usage["failures"] += 1
                 if "timeout" in failure_code.lower():
                     usage["timeouts"] += 1
-                domain_errors.append({"code": failure_code, "message": str(item.error or "")})
+                domain_errors.append(
+                    {
+                        "code": failure_code,
+                        "message": str(getattr(failure, "message", "") or item.error or ""),
+                        "task_id": item.task_id,
+                    }
+                )
 
     trace_ids = {binding.root_trace_id}
     trace_ids.update(item.worker_trace_id for item in attempts)
@@ -650,7 +693,32 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
         "last_validation": (
             ledger_value["validations"][last_validation.validation_id] if last_validation else None
         ),
-        "last_domain_error": domain_errors[-1] if domain_errors else None,
+        "current_domain_error": (
+            None
+            if root.status.value == "completed"
+            else next(
+                (
+                    item
+                    for item in reversed(domain_errors)
+                    if ledger.tasks[item["task_id"]].status.value
+                    not in {"completed", "cancelled", "superseded"}
+                ),
+                None,
+            )
+        ),
+        "last_recovered_error": next(
+            (
+                item
+                for item in reversed(domain_errors)
+                if ledger.tasks[item["task_id"]].status.value
+                in {"completed", "cancelled", "superseded"}
+            ),
+            None,
+        ),
+        "error_history_by_code": {
+            code: sum(item["code"] == code for item in domain_errors)
+            for code in sorted({item["code"] for item in domain_errors})
+        },
         "phase_timing_seconds": task_seconds,
         "preset_usage": preset_usage,
         "tool_calls": tool_calls,
@@ -664,6 +732,21 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
     }
 
 
+def _empty_preset_usage() -> dict[str, int | float]:
+    return {
+        "agent_runs": 0,
+        "stage_runs": 0,
+        "calls": 0,
+        "prompt_tokens": 0,
+        "completion_tokens": 0,
+        "cache_read_tokens": 0,
+        "tokens": 0,
+        "timeouts": 0,
+        "failures": 0,
+        "seconds": 0.0,
+    }
+
+
 async def _attach_diagnostics(host: Any, report: RealE2EReport) -> RealE2EReport:
     try:
         diagnostics = await _collect_diagnostics(host, report.script_build_id)

script_build_host/tests/test_phase_two_real_runner_e2e.py

@@ -4,7 +4,9 @@ import asyncio
 import json
 from collections.abc import Mapping, Sequence
 from pathlib import Path
+from types import SimpleNamespace
 from typing import Any, cast
+from uuid import uuid4
 
 import httpx
 import pytest
@@ -43,6 +45,7 @@ from script_build_host.infrastructure.tables import (
     publication_table,
 )
 from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
+from script_build_host.internal_e2e import _collect_diagnostics
 from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
 from script_build_host.repositories.sqlalchemy import (
     SqlAlchemyInputSnapshotRepository,
@@ -55,7 +58,6 @@ from script_build_host.tools.gateway import RetrievalResult
 
 ROOT = "phase-two-real-runner-root"
 FULL_SCOPE = "script-build://scopes/full"
-DIRECTION_GOAL_ID = "direction-goal"
 
 
 def _tool(call_id: str, name: str, arguments: Mapping[str, Any]) -> dict[str, Any]:
@@ -138,18 +140,21 @@ def _contract(
     supersedes_decision_ids: Sequence[str] = (),
     goal_ids: Sequence[str] | None = None,
 ) -> dict[str, Any]:
-    output_schema = {
-        "direction": "script-direction/v1",
-        "decode-retrieval": "evidence-record/v1",
-        "structure": "structure-artifact/v1",
-        "paragraph": "paragraph-artifact/v1",
-        "element-set": "element-set-artifact/v1",
-        "compare": "comparison-artifact/v1",
-        "compose": "structured-script/v1",
-        "candidate-portfolio": "candidate-portfolio/v1",
-    }[task_kind]
+    del write_scope, closure, adopted, held, compose_order
+    input_decision_ids = [str(item["decision_id"]) for item in input_decision_refs]
+    base_decision_id = None
+    if base_artifact_ref is not None:
+        base_uri = str(base_artifact_ref["uri"])
+        base_decision_id = next(
+            (
+                str(item["decision_id"])
+                for item in input_decision_refs
+                if str(cast(Mapping[str, Any], item["artifact_ref"])["uri"]) == base_uri
+            ),
+            None,
+        )
+        assert base_decision_id is not None
     return {
-        "schema_version": "script-task-contract/v1",
         "task_kind": task_kind,
         "scope_ref": scope_ref,
         "intent_class": (
@@ -160,11 +165,9 @@ def _contract(
             else "explore"
         ),
         "objective": objective or f"produce one concrete and independently verifiable {task_kind}",
-        "input_decision_refs": list(input_decision_refs),
-        "base_artifact_ref": base_artifact_ref,
-        "write_scope": list(write_scope),
+        "input_decision_ids": input_decision_ids,
+        "base_decision_id": base_decision_id,
         "gap_ref": None,
-        "output_schema": output_schema,
         "criteria": [
             {
                 "criterion_id": f"{task_kind}-closed",
@@ -172,22 +175,11 @@ def _contract(
                 "hard": True,
             }
         ],
-        "budget": {
-            "max_attempts": 4,
-            "max_tokens": 32_000,
-            "max_seconds": 900,
-            "max_external_queries": 40,
-            "max_no_improvement": 3,
-        },
-        "goal_ids": list(goal_ids)
-        if goal_ids is not None
-        else ([] if task_kind in {"direction", "decode-retrieval"} else [DIRECTION_GOAL_ID]),
+        "goal_ids": list(goal_ids or ()),
         "supersedes_decision_ids": list(supersedes_decision_ids),
-        "candidate_closure_decision_refs": list(closure),
-        "adopted_decision_ids": list(adopted),
-        "held_or_rejected_decision_ids": list(held),
-        "compose_order": list(compose_order),
-        "comparison_decision_refs": list(comparison_decision_refs),
+        "comparison_decision_ids": [
+            str(item["decision_id"]) for item in comparison_decision_refs
+        ],
     }
 
 
@@ -214,19 +206,45 @@ class _PhaseTwoScriptedLLM:
     def __init__(self, store: FileSystemTaskStore) -> None:
         self.store = store
         self.calls = 0
+        self.call_namespace = uuid4().hex
+        self.tool_calls_by_name: dict[str, int] = {}
         self.submit_attempt_arguments: list[dict[str, Any]] = []
         self.worker_kinds: list[str] = []
         self.validator_kinds: list[str] = []
         self.phase_two_active = False
         self.compose_candidates_saved = 0
+        self.direction_goal_id: str | None = None
 
     def _call(self, name: str, arguments: Mapping[str, Any]) -> dict[str, Any]:
         self.calls += 1
+        self.tool_calls_by_name[name] = self.tool_calls_by_name.get(name, 0) + 1
         if name == "submit_attempt":
             self.submit_attempt_arguments.append(dict(arguments))
-        return _tool(f"phase2-call-{self.calls}", name, arguments)
+        return _tool(f"phase2-{self.call_namespace}-{self.calls}", name, arguments)
 
     async def __call__(self, messages, tools, **_kwargs):
+        if self.direction_goal_id is None:
+            for message in reversed(messages):
+                if not isinstance(message.get("content"), str):
+                    continue
+                try:
+                    value = json.loads(cast(str, message["content"]))
+                except ValueError:
+                    continue
+                if isinstance(value, dict) and isinstance(
+                    value.get("goal_ids_by_client_key"), dict
+                ):
+                    self.direction_goal_id = str(
+                        value["goal_ids_by_client_key"]["observable-opening"]
+                    )
+                    break
+                if isinstance(value, dict) and isinstance(
+                    value.get("active_direction_goal_ids"), list
+                ):
+                    goal_ids = cast(list[str], value["active_direction_goal_ids"])
+                    if goal_ids:
+                        self.direction_goal_id = goal_ids[0]
+                        break
         self.phase_two_active = self.phase_two_active or any(
             PHASE_TWO_CANDIDATE_PORTFOLIO_READY in str(item.get("content", "")) for item in messages
         )
@@ -308,6 +326,8 @@ class _PhaseTwoScriptedLLM:
             )
 
         direction_decision, direction_ref = _accepted_ref(ledger, direction)
+        assert self.direction_goal_id is not None
+        goal_ids = (self.direction_goal_id,)
         direction_pin = _decision_ref(
             direction_decision,
             direction_ref,
@@ -326,6 +346,7 @@ class _PhaseTwoScriptedLLM:
                             scope_ref=FULL_SCOPE,
                             write_scope=("script-build://writes",),
                             input_decision_refs=(direction_pin,),
+                            goal_ids=goal_ids,
                         )
                     ]
                 },
@@ -342,6 +363,7 @@ class _PhaseTwoScriptedLLM:
                             scope_ref=FULL_SCOPE,
                             write_scope=("script-build://writes",),
                             input_decision_refs=(direction_pin,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -358,6 +380,7 @@ class _PhaseTwoScriptedLLM:
                             scope_ref="script-build://scopes/full/opening",
                             write_scope=("script-build://writes/paragraphs/opening",),
                             input_decision_refs=(direction_pin,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -390,6 +413,7 @@ class _PhaseTwoScriptedLLM:
                             objective=first_paragraph_objective,
                             input_decision_refs=(direction_pin, structure_pin),
                             base_artifact_ref=_artifact_ref_payload(structure_ref),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -400,34 +424,49 @@ class _PhaseTwoScriptedLLM:
             return self._accept(paragraph, ledger)
 
         paragraph_decision, paragraph_ref = _accepted_ref(ledger, paragraph)
-        if compose.current_spec_version == 1:
-            structure_pin = _decision_ref(
-                structure_decision,
-                structure_ref,
-                scope_ref="script-build://scopes/full/opening",
-                task_kind="structure",
-            )
+        element_objective = "attach one independent element to the opening"
+        element = by_objective(element_objective)
+        if element is None:
             paragraph_pin = _decision_ref(
                 paragraph_decision,
                 paragraph_ref,
                 scope_ref="script-build://scopes/full/opening",
                 task_kind="paragraph",
             )
+            return self._call(
+                "plan_script_tasks",
+                {
+                    "parent_task_id": compose.task_id,
+                    "contracts": [
+                        _contract(
+                            "element-set",
+                            scope_ref="script-build://scopes/full/opening/elements",
+                            write_scope=("script-build://writes/elements",),
+                            objective=element_objective,
+                            input_decision_refs=(direction_pin, paragraph_pin),
+                            base_artifact_ref=_artifact_ref_payload(paragraph_ref),
+                            goal_ids=goal_ids,
+                        )
+                    ],
+                },
+            )
+        if element.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
+            return self._call("dispatch_script_tasks", {"task_ids": [element.task_id]})
+        if element.status is TaskStatus.AWAITING_DECISION:
+            return self._accept(element, ledger)
+        element_decision, _element_ref = _accepted_ref(ledger, element)
+        if compose.current_spec_version == 1:
             return self._call(
                 "decide_script_task",
                 {
                     "task_id": compose.task_id,
                     "action": "revise",
                     "reason": "pin the adopted paragraph and formal compose order",
-                    "replacement_contract": _contract(
-                        "compose",
-                        scope_ref=FULL_SCOPE,
-                        write_scope=("script-build://writes",),
-                        input_decision_refs=(direction_pin,),
-                        closure=(structure_pin, paragraph_pin),
-                        adopted=(structure_decision, paragraph_decision),
-                        compose_order=(structure_decision, paragraph_decision),
-                    ),
+                    "selected_decision_ids": [
+                        structure_decision,
+                        paragraph_decision,
+                        element_decision,
+                    ],
                 },
             )
         if compose.status is TaskStatus.PENDING:
@@ -466,6 +505,7 @@ class _PhaseTwoScriptedLLM:
                             ),
                             base_artifact_ref=_artifact_ref_payload(structure_ref),
                             supersedes_decision_ids=(paragraph_decision,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -493,7 +533,7 @@ class _PhaseTwoScriptedLLM:
                             scope_ref="script-build://scopes/full/opening/evidence",
                             write_scope=(),
                             objective="retrieve evidence for the replacement opening",
-                            goal_ids=(DIRECTION_GOAL_ID,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -535,6 +575,7 @@ class _PhaseTwoScriptedLLM:
                                     task_kind="paragraph",
                                 ),
                             ),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -544,50 +585,19 @@ class _PhaseTwoScriptedLLM:
         if comparison.status is TaskStatus.AWAITING_DECISION:
             return self._accept(comparison, ledger)
 
-        comparison_decision, comparison_ref = _accepted_ref(ledger, comparison)
+        _comparison_decision, _comparison_ref = _accepted_ref(ledger, comparison)
         if compose.status is TaskStatus.NEEDS_REPLAN:
-            structure_pin = _decision_ref(
-                structure_decision,
-                structure_ref,
-                scope_ref="script-build://scopes/full/opening",
-                task_kind="structure",
-            )
-            paragraph_pin = _decision_ref(
-                paragraph_decision,
-                paragraph_ref,
-                scope_ref="script-build://scopes/full/opening",
-                task_kind="paragraph",
-            )
-            replacement_pin = _decision_ref(
-                replacement_decision,
-                replacement_ref,
-                scope_ref="script-build://scopes/full/opening",
-                task_kind="paragraph",
-            )
             return self._call(
                 "decide_script_task",
                 {
                     "task_id": compose.task_id,
                     "action": "revise",
                     "reason": "pin the accepted replacement and the symmetric comparison",
-                    "replacement_contract": _contract(
-                        "compose",
-                        scope_ref=FULL_SCOPE,
-                        write_scope=("script-build://writes",),
-                        input_decision_refs=(direction_pin,),
-                        closure=(structure_pin, paragraph_pin, replacement_pin),
-                        adopted=(structure_decision, replacement_decision),
-                        held=(paragraph_decision,),
-                        compose_order=(structure_decision, replacement_decision),
-                        comparison_decision_refs=(
-                            _decision_ref(
-                                comparison_decision,
-                                comparison_ref,
-                                scope_ref="script-build://scopes/full/opening",
-                                task_kind="compare",
-                            ),
-                        ),
-                    ),
+                    "selected_decision_ids": [
+                        structure_decision,
+                        replacement_decision,
+                        element_decision,
+                    ],
                 },
             )
         if compose.status is TaskStatus.PENDING:
@@ -595,29 +605,15 @@ class _PhaseTwoScriptedLLM:
         if compose.status is TaskStatus.AWAITING_DECISION:
             return self._accept(compose, ledger)
 
-        compose_decision, compose_ref = _accepted_ref(ledger, compose)
+        compose_decision, _compose_ref = _accepted_ref(ledger, compose)
         if portfolio.current_spec_version == 1:
-            compose_pin = _decision_ref(
-                compose_decision,
-                compose_ref,
-                scope_ref=FULL_SCOPE,
-                task_kind="compose",
-            )
             return self._call(
                 "decide_script_task",
                 {
                     "task_id": portfolio.task_id,
                     "action": "revise",
                     "reason": "pin the uniquely adopted StructuredScript",
-                    "replacement_contract": _contract(
-                        "candidate-portfolio",
-                        scope_ref=FULL_SCOPE,
-                        write_scope=("script-build://writes",),
-                        input_decision_refs=(direction_pin,),
-                        closure=(compose_pin,),
-                        adopted=(compose_decision,),
-                        compose_order=(compose_decision,),
-                    ),
+                    "selected_decision_ids": [compose_decision],
                 },
             )
         if portfolio.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
@@ -675,7 +671,7 @@ class _PhaseTwoScriptedLLM:
                     {
                         "goals": [
                             {
-                                "goal_id": DIRECTION_GOAL_ID,
+                                "client_key": "observable-opening",
                                 "statement": "open with one observable reversal",
                                 "rationale": "the frozen topic benefits from a concrete hook",
                                 "success_criteria": [
@@ -694,37 +690,49 @@ class _PhaseTwoScriptedLLM:
                         "evidence_refs": [evidence_refs[0]["uri"]],
                     },
                 )
+            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):
                 return self._call(
-                    "create_script_paragraph",
+                    "create_script_paragraphs",
                     {
-                        "paragraph_index": 1,
-                        "name": "Opening structure",
-                        "content_range": {"scope": "opening", "beats": ["setup", "reversal"]},
-                        "theme_elements": [
-                            {
-                                "原子点": "assumption changes after evidence",
-                                "维度": "claim",
-                                "维度类型": "主维度",
-                            }
-                        ],
-                        "form_elements": [
-                            {
-                                "原子点": "setup then concrete reversal",
-                                "维度": "shape",
-                                "维度类型": "主维度",
-                            }
-                        ],
-                        "function_elements": [
+                        "paragraphs": [
                             {
-                                "原子点": "create curiosity without abstraction",
-                                "维度": "job",
-                                "维度类型": "主维度",
+                                "client_key": "opening",
+                                "paragraph_index": 1,
+                                "name": "Opening structure",
+                                "content_range": {
+                                    "scope": "opening",
+                                    "beats": ["setup", "reversal"],
+                                },
+                                "theme_elements": [
+                                    {
+                                        "原子点": "assumption changes after evidence",
+                                        "维度": "claim",
+                                        "维度类型": "主维度",
+                                    }
+                                ],
+                                "form_elements": [
+                                    {
+                                        "原子点": "setup then concrete reversal",
+                                        "维度": "shape",
+                                        "维度类型": "主维度",
+                                    }
+                                ],
+                                "function_elements": [
+                                    {
+                                        "原子点": "create curiosity without abstraction",
+                                        "维度": "job",
+                                        "维度类型": "主维度",
+                                    }
+                                ],
+                                "feeling_elements": [
+                                    {"原子点": "measured surprise", "维度": "tone"}
+                                ],
                             }
-                        ],
-                        "feeling_elements": [{"原子点": "measured surprise", "维度": "tone"}],
+                        ]
                     },
                 )
             return self._call("submit_attempt", {})
@@ -764,6 +772,45 @@ class _PhaseTwoScriptedLLM:
                     ]
                 },
             )
+        if kind == "element-set":
+            if isinstance(last, dict) 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",
+                        {
+                            "name": "Observable reversal",
+                            "dimension_primary": "实质",
+                            "dimension_secondary": "opening-evidence",
+                            "commonality_analysis": {
+                                "summary": "a visible detail changes the claim"
+                            },
+                            "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", {})
         if kind == "compare":
             if isinstance(last, dict) and "artifact_ref" in last:
                 return self._call("submit_attempt", {})
@@ -873,6 +920,12 @@ 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",
             {
@@ -882,6 +935,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])],
                     }
                     for item in criteria
                 ],
@@ -962,6 +1016,8 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
             return await super()._planner()
 
         root = ledger.tasks[ledger.root_task_id]
+        assert self.direction_goal_id is not None
+        goal_ids = (self.direction_goal_id,)
         direction_decision, direction_ref = _accepted_ref(ledger, direction)
         direction_pin = _decision_ref(
             direction_decision,
@@ -986,6 +1042,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                             scope_ref=FULL_SCOPE,
                             write_scope=("script-build://writes",),
                             input_decision_refs=(direction_pin,),
+                            goal_ids=goal_ids,
                         )
                     ]
                 },
@@ -1002,6 +1059,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                             scope_ref=FULL_SCOPE,
                             write_scope=("script-build://writes",),
                             input_decision_refs=(direction_pin,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -1038,6 +1096,20 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                         ),
                     )
                     base = _artifact_ref_payload(structure_ref)
+                elif kind == "element-set":
+                    structure_decision, structure_ref = _accepted_ref(
+                        ledger, exploratory["structure"]
+                    )
+                    inputs = (
+                        direction_pin,
+                        _decision_ref(
+                            structure_decision,
+                            structure_ref,
+                            scope_ref=scopes["structure"],
+                            task_kind="structure",
+                        ),
+                    )
+                    base = _artifact_ref_payload(structure_ref)
                 return self._call(
                     "plan_script_tasks",
                     {
@@ -1050,6 +1122,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                                 objective=objective,
                                 input_decision_refs=inputs,
                                 base_artifact_ref=base,
+                                goal_ids=goal_ids,
                             )
                         ],
                     },
@@ -1094,6 +1167,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                             ),
                             base_artifact_ref=_artifact_ref_payload(structure_ref),
                             supersedes_decision_ids=(paragraph_decision,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -1138,6 +1212,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                             ),
                             base_artifact_ref=_artifact_ref_payload(adopted_paragraph_ref),
                             supersedes_decision_ids=(element_decision,),
+                            goal_ids=goal_ids,
                         )
                     ],
                 },
@@ -1148,38 +1223,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
             return self._accept(element_replacement, ledger)
         adopted_element_decision, adopted_element_ref = _accepted_ref(ledger, element_replacement)
 
-        pins = (
-            _decision_ref(
-                structure_decision,
-                structure_ref,
-                scope_ref=scopes["structure"],
-                task_kind="structure",
-            ),
-            _decision_ref(
-                paragraph_decision,
-                paragraph_ref,
-                scope_ref="script-build://scopes/full/opening",
-                task_kind="paragraph",
-            ),
-            _decision_ref(
-                element_decision,
-                element_ref,
-                scope_ref=scopes["element-set"],
-                task_kind="element-set",
-            ),
-            _decision_ref(
-                adopted_paragraph_decision,
-                adopted_paragraph_ref,
-                scope_ref="script-build://scopes/full/opening",
-                task_kind="paragraph",
-            ),
-            _decision_ref(
-                adopted_element_decision,
-                adopted_element_ref,
-                scope_ref=scopes["element-set"],
-                task_kind="element-set",
-            ),
-        )
+        del adopted_paragraph_ref, adopted_element_ref
         if compose.current_spec_version == 1:
             return self._call(
                 "decide_script_task",
@@ -1187,24 +1231,11 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                     "task_id": compose.task_id,
                     "action": "revise",
                     "reason": "formal adoption follows immutable lineage, not exploration order",
-                    "replacement_contract": _contract(
-                        "compose",
-                        scope_ref=FULL_SCOPE,
-                        write_scope=("script-build://writes",),
-                        input_decision_refs=(direction_pin,),
-                        closure=pins,
-                        adopted=(
-                            structure_decision,
-                            adopted_paragraph_decision,
-                            adopted_element_decision,
-                        ),
-                        held=(paragraph_decision, element_decision),
-                        compose_order=(
-                            structure_decision,
-                            adopted_paragraph_decision,
-                            adopted_element_decision,
-                        ),
-                    ),
+                    "selected_decision_ids": [
+                        structure_decision,
+                        adopted_paragraph_decision,
+                        adopted_element_decision,
+                    ],
                 },
             )
         if compose.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
@@ -1212,7 +1243,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
         if compose.status is TaskStatus.AWAITING_DECISION:
             return self._accept(compose, ledger)
 
-        compose_decision, compose_ref = _accepted_ref(ledger, compose)
+        compose_decision, _compose_ref = _accepted_ref(ledger, compose)
         if portfolio.current_spec_version == 1:
             return self._call(
                 "decide_script_task",
@@ -1220,22 +1251,7 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                     "task_id": portfolio.task_id,
                     "action": "revise",
                     "reason": "pin the unique accepted StructuredScript",
-                    "replacement_contract": _contract(
-                        "candidate-portfolio",
-                        scope_ref=FULL_SCOPE,
-                        write_scope=("script-build://writes",),
-                        input_decision_refs=(direction_pin,),
-                        closure=(
-                            _decision_ref(
-                                compose_decision,
-                                compose_ref,
-                                scope_ref=FULL_SCOPE,
-                                task_kind="compose",
-                            ),
-                        ),
-                        adopted=(compose_decision,),
-                        compose_order=(compose_decision,),
-                    ),
+                    "selected_decision_ids": [compose_decision],
                 },
             )
         if portfolio.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
@@ -1258,7 +1274,8 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
         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" in names
+            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)
@@ -1269,11 +1286,31 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
             paragraphs = cast(Sequence[Mapping[str, Any]], last["paragraphs"])
             if not paragraphs:
                 return self._call(
-                    "create_script_paragraph",
+                    "create_script_paragraphs",
                     {
-                        "paragraph_index": 1,
-                        "name": "Exploratory opening",
-                        "content_range": {"scope": "opening", "beats": ["setup", "reversal"]},
+                        "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(
@@ -1338,7 +1375,8 @@ class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
                     },
                 )
         if kind == "structure":
-            assert "create_script_paragraph" in names
+            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)
 
@@ -1707,6 +1745,7 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         "direction",
         "structure",
         "paragraph",
+        "element-set",
         "compose",
         "compare",
         "candidate-portfolio",
@@ -1716,13 +1755,15 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         "direction",
         "structure",
         "paragraph",
-        "compose",
+        "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.tool_calls_by_name.get("create_script_paragraph", 0) == 0
 
     compose_task = next(
         task
@@ -1790,6 +1831,9 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
         ]
         assert len(validations) == 1
         validator_trace = await trace_store.get_trace(validations[0].validator_trace_id)
+        if validations[0].summary.startswith("Deterministic preflight rejected"):
+            assert validator_trace is None
+            continue
         assert validator_trace is not None
         assert (
             validator_trace.context["role_prompt_identity"]
@@ -1846,6 +1890,9 @@ async def test_mission_service_real_runner_auto_continues_phase_one_to_phase_two
             if ledger.validations[item].attempt_id == attempt.attempt_id
         )
         validator_trace = await reopened_traces.get_trace(validation.validator_trace_id)
+        if validation.summary.startswith("Deterministic preflight rejected"):
+            assert validator_trace is None
+            continue
         assert validator_trace is not None and validator_trace.status == "completed"
         assert await reopened_traces.get_trace_messages(validation.validator_trace_id)
 
@@ -2266,6 +2313,22 @@ async def test_real_runner_phase_one_to_three_final_publication_and_legacy_readb
     assert binding["accepted_root_artifact_version_id"] == publication["artifact_version_id"]
     assert candidate_ids_after == candidate_ids_before
 
+    await trace_store.record_model_usage(
+        ROOT,
+        sequence=10_000,
+        role="assistant",
+        model="scripted-fake-model",
+        prompt_tokens=123,
+        completion_tokens=7,
+    )
+    diagnostics = await _collect_diagnostics(
+        SimpleNamespace(composition=composition), script_build_id
+    )
+    assert diagnostics["current_domain_error"] is None
+    assert "last_domain_error" not in diagnostics
+    assert diagnostics["preset_usage"]["script_planner"]["calls"] >= 1
+    assert diagnostics["preset_usage"]["script_planner"]["tokens"] >= 130
+
 
 def _phase_two_policy_count(messages: Sequence[Any]) -> int:
     return len(