Broker回归：覆盖全角色闭包、重启恢复与历史压力形态

新增跨 Planner、Phase1、Phase2、Phase3 的集成矩阵，覆盖 193 个 Pattern候选、Root Validator必读页、Compare all-and-only、Element精确缺口、Persona稳定handle、语义缓存和跨重启配额。以 Build 900049 的54轮与49轮形态验证上下文不线性增长。

script_build_host/tests/test_context_broker_integration.py

@@ -0,0 +1,1147 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from dataclasses import asdict, dataclass
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from agent.orchestration import (
+    AgentRole,
+    ArtifactRef,
+    DecisionAction,
+    RoleContextRequest,
+    TaskStatus,
+)
+
+from script_build_host.application.context_semantics import AdaptiveSemanticSelector
+from script_build_host.application.mission_workbench import (
+    MissionWorkbench,
+    WorkbenchError,
+    required_context_source_types,
+)
+from script_build_host.domain.context_broker import ContextCard, estimate_context_tokens
+from script_build_host.domain.task_contracts import (
+    AcceptedDecisionRef,
+    ScriptCriterion,
+    ScriptIntentClass,
+    ScriptTaskBudget,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+)
+
+
+class _TaskStore:
+    def __init__(self, ledger: Any) -> None:
+        self.ledger = ledger
+
+    async def load(self, _root_trace_id: str) -> Any:
+        return self.ledger
+
+
+@pytest.mark.parametrize(
+    ("kind", "required"),
+    [
+        (ScriptTaskKind.PATTERN_RETRIEVAL, {"topic"}),
+        (ScriptTaskKind.DECODE_RETRIEVAL, {"topic"}),
+        (ScriptTaskKind.EXTERNAL_RETRIEVAL, {"topic"}),
+        (ScriptTaskKind.KNOWLEDGE_RETRIEVAL, {"topic"}),
+        (
+            ScriptTaskKind.DIRECTION,
+            {"topic", "persona", "section_pattern", "strategy"},
+        ),
+        (ScriptTaskKind.STRUCTURE, {"direction", "section_pattern"}),
+        (ScriptTaskKind.PARAGRAPH, {"direction", "structure", "persona"}),
+        (ScriptTaskKind.ELEMENT_SET, {"paragraph"}),
+        (ScriptTaskKind.COMPARE, set()),
+        (
+            ScriptTaskKind.ROOT_DELIVERY,
+            {"direction", "candidate-portfolio", "structured-script"},
+        ),
+    ],
+)
+def test_every_e2e_role_has_one_authoritative_context_source_policy(
+    kind: ScriptTaskKind, required: set[str]
+) -> None:
+    assert set(required_context_source_types(kind)) == required
+
+
+class _Bindings:
+    async def get_by_root(self, root_trace_id: str) -> Any:
+        return SimpleNamespace(
+            root_trace_id=root_trace_id,
+            script_build_id=900049,
+            input_snapshot_id=3,
+            active_direction_artifact_version_id=None,
+        )
+
+
+class _Snapshots:
+    def __init__(self) -> None:
+        self.value = SimpleNamespace(
+            snapshot_id="3",
+            topic_id=49,
+            topic={"topic": {"result": "健身房中的职场反转故事"}},
+            account={"account_name": "fixture"},
+            persona_points=tuple(
+                {
+                    "列": ("实质", "形式", "作用", "感受")[index % 4],
+                    "点名称": f"persona-{index}",
+                    "人设权重分": 1 - index / 100,
+                    "帖子覆盖率": 0.8,
+                }
+                for index in range(80)
+            ),
+            section_patterns=tuple(
+                {"模式名称": f"section-{index}", "说明": "转折节奏"} for index in range(40)
+            ),
+            strategies=(
+                {"name": "must", "mode": "always_on", "content": "始终使用可视化叙事"},
+                {"name": "optional", "mode": "on_demand", "content": "按需策略" * 2_000},
+            ),
+            model_manifest={"presets": {"script_context_broker": {"model": "fake"}}},
+        )
+
+    async def get(self, _snapshot_id: str, *, script_build_id: int) -> Any:
+        assert script_build_id == 900049
+        return self.value
+
+
+class _Contracts:
+    def __init__(self, values: dict[str, ScriptTaskContractV1]) -> None:
+        self.values = values
+
+    async def read(self, _root: str, ref: str) -> Any:
+        return SimpleNamespace(contract=self.values[ref])
+
+
+@dataclass
+class _Artifact:
+    text: str
+    raw_artifact_ref: str | None = None
+
+    def content_payload(self) -> dict[str, Any]:
+        return {"schema_version": "fixture", "full_description": self.text}
+
+
+class _Artifacts:
+    def __init__(self, values: dict[str, _Artifact]) -> None:
+        self.values = values
+
+    async def read_by_ref(self, ref: ArtifactRef, *, script_build_id: int, **_: Any) -> Any:
+        assert script_build_id == 900049
+        return SimpleNamespace(artifact=self.values[ref.uri])
+
+
+class _RawStore:
+    def __init__(self, raw_ref: str, payload: dict[str, Any]) -> None:
+        self.raw_ref = raw_ref
+        self.content = json.dumps(payload, ensure_ascii=False).encode()
+        self.reads = 0
+
+    async def read_bytes(self, ref: str) -> bytes:
+        assert ref == self.raw_ref
+        self.reads += 1
+        return self.content
+
+
+class _ReceiptStore:
+    def __init__(self) -> None:
+        self.events: list[dict[str, Any]] = []
+
+    async def append_event(self, _trace_id: str, event: str, payload: dict[str, Any]) -> int:
+        event_id = len(self.events) + 1
+        self.events.append({"event_id": event_id, "event": event, **payload})
+        return event_id
+
+    async def get_events(self, _trace_id: str) -> list[dict[str, Any]]:
+        return list(self.events)
+
+
+def _contract(kind: ScriptTaskKind, index: int) -> ScriptTaskContractV1:
+    return ScriptTaskContractV1(
+        task_kind=kind,
+        scope_ref=f"script-build://scopes/full/part-{index}",
+        intent_class=(
+            ScriptIntentClass.EXPLORE
+            if kind is not ScriptTaskKind.PATTERN_RETRIEVAL
+            else ScriptIntentClass.EXPAND
+        ),
+        objective=f"task objective {index}",
+        input_decision_refs=(),
+        base_artifact_ref=None,
+        write_scope=(f"script-build://writes/part-{index}",),
+        gap_ref=None,
+        output_schema=(
+            "evidence-record/v1"
+            if kind is ScriptTaskKind.PATTERN_RETRIEVAL
+            else "paragraph-artifact/v1"
+        ),
+        criteria=(ScriptCriterion(f"c-{index}", "quality"),),
+        budget=ScriptTaskBudget(),
+        goal_ids=(f"g-{index % 6}",) if kind is not ScriptTaskKind.PATTERN_RETRIEVAL else (),
+        compiler_manifest_digest="sha256:" + "a" * 64,
+    )
+
+
+def _fixture() -> tuple[MissionWorkbench, Any, _Snapshots, str]:
+    root = SimpleNamespace(
+        task_id="root-task",
+        parent_task_id=None,
+        child_task_ids=[],
+        display_path="1",
+        status=TaskStatus.RUNNING,
+        blocked_reason=None,
+        current_spec=SimpleNamespace(objective="build complete script", context_refs=()),
+        current_spec_version=1,
+        attempt_ids=[],
+        decision_ids=[],
+    )
+    tasks = {"root-task": root}
+    attempts: dict[str, Any] = {}
+    decisions: dict[str, Any] = {}
+    contracts: dict[str, ScriptTaskContractV1] = {}
+    artifacts: dict[str, _Artifact] = {}
+    for index in range(58):
+        task_id = f"task-{index}"
+        contract_ref = f"script-build://task-contracts/sha256/{index:064x}"
+        accepted = index < 33
+        attempt_id = f"attempt-{index}"
+        decision_id = f"decision-{index}"
+        artifact_uri = f"script-build://artifact-versions/{index + 1}"
+        task = SimpleNamespace(
+            task_id=task_id,
+            parent_task_id="root-task",
+            child_task_ids=[],
+            display_path=f"1.{index + 1}",
+            status=TaskStatus.COMPLETED if accepted else TaskStatus.PENDING,
+            blocked_reason=None,
+            current_spec=SimpleNamespace(
+                objective=f"objective {index}",
+                context_refs=("script-build://task-kinds/paragraph", contract_ref),
+            ),
+            current_spec_version=1,
+            attempt_ids=[attempt_id] if accepted else [],
+            decision_ids=[decision_id] if accepted else [],
+        )
+        tasks[task_id] = task
+        contracts[contract_ref] = _contract(ScriptTaskKind.PARAGRAPH, index)
+        if accepted:
+            ref = ArtifactRef(
+                uri=artifact_uri,
+                kind="paragraph",
+                version=str(index + 1),
+                digest="sha256:" + f"{index + 1:064x}",
+            )
+            attempts[attempt_id] = SimpleNamespace(
+                task_id=task_id,
+                submission=SimpleNamespace(artifact_refs=(ref,), evidence_refs=()),
+                failure=None,
+                status=SimpleNamespace(value="succeeded"),
+            )
+            decisions[decision_id] = SimpleNamespace(
+                decision_id=decision_id,
+                task_id=task_id,
+                action=DecisionAction.ACCEPT,
+                attempt_id=attempt_id,
+                reason=f"FORGED REASON {index}",
+            )
+            artifacts[artifact_uri] = _Artifact(f"REAL ARTIFACT CONTENT {index}")
+
+    raw_ref = "script-build://raw-artifacts/sha256/" + "b" * 64
+    pattern_index = 58
+    task_id = "pattern-task"
+    contract_ref = "script-build://task-contracts/sha256/" + "f" * 64
+    attempt_id = "pattern-attempt"
+    decision_id = "pattern-decision"
+    evidence_ref = ArtifactRef(
+        uri="script-build://artifact-versions/100",
+        kind="evidence",
+        version="100",
+        digest="sha256:" + "c" * 64,
+    )
+    tasks[task_id] = SimpleNamespace(
+        task_id=task_id,
+        parent_task_id="root-task",
+        child_task_ids=[],
+        display_path="1.59",
+        status=TaskStatus.COMPLETED,
+        blocked_reason=None,
+        current_spec=SimpleNamespace(
+            objective="retrieve patterns",
+            context_refs=("script-build://task-kinds/pattern-retrieval", contract_ref),
+        ),
+        current_spec_version=1,
+        attempt_ids=[attempt_id],
+        decision_ids=[decision_id],
+    )
+    contracts[contract_ref] = _contract(ScriptTaskKind.PATTERN_RETRIEVAL, pattern_index)
+    attempts[attempt_id] = SimpleNamespace(
+        task_id=task_id,
+        submission=SimpleNamespace(artifact_refs=(), evidence_refs=(evidence_ref,)),
+        failure=None,
+        status=SimpleNamespace(value="succeeded"),
+    )
+    decisions[decision_id] = SimpleNamespace(
+        decision_id=decision_id,
+        task_id=task_id,
+        action=DecisionAction.ACCEPT,
+        attempt_id=attempt_id,
+        reason="pattern accepted",
+    )
+    artifacts[evidence_ref.uri] = _Artifact("pattern summary", raw_artifact_ref=raw_ref)
+    ledger = SimpleNamespace(
+        revision=49,
+        root_task_id="root-task",
+        focused_task_id="task-57",
+        tasks=tasks,
+        attempts=attempts,
+        decisions=decisions,
+        operations={},
+    )
+    patterns = {
+        "candidates": [
+            {"name": f"pattern-{index}", "content": "模式详情" * 350} for index in range(193)
+        ]
+    }
+    snapshots = _Snapshots()
+    workbench = MissionWorkbench(
+        task_store=_TaskStore(ledger),
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=_Artifacts(artifacts),
+        contracts=_Contracts(contracts),
+        raw_artifacts=_RawStore(raw_ref, patterns),
+    )
+    return workbench, ledger, snapshots, raw_ref
+
+
+@pytest.mark.asyncio
+async def test_planner_hot_context_is_bounded_and_every_item_is_pageable() -> None:
+    workbench, _ledger, _snapshots, _ = _fixture()
+    state = await workbench.planner_state("root")
+    assert state.active_total == 26
+    assert state.active_has_more is True
+    assert state.active_subgraph[0]["task_id"] == "task-57"
+    assert state.accepted_total == 34
+    encoded = json.dumps(state.to_payload(), ensure_ascii=False)
+    assert len(encoded) < 24_000
+    assert "FORGED REASON" not in " ".join(str(item["summary"]) for item in state.accepted_frontier)
+
+    assert any("REAL ARTIFACT CONTENT" in str(item["summary"]) for item in state.accepted_frontier)
+
+    cursor = None
+    found: list[str] = []
+    while True:
+        page = await workbench.search_mission_context(
+            root_trace_id="root",
+            role="planner",
+            collection="accepted_decisions",
+            cursor=cursor,
+        )
+        found.extend(item["handle"] for item in page["page"]["items"])
+        cursor = page["page"]["next_cursor"]
+        if cursor is None:
+            break
+    assert len(found) == 34
+    assert len(set(found)) == 34
+
+
+@pytest.mark.asyncio
+async def test_duplicate_long_inputs_have_unique_handles_and_budgeted_finite_pagination() -> None:
+    workbench, _ledger, snapshots, _ = _fixture()
+    snapshots.value.persona_points = tuple(
+        {"列": "实质", "点名称": "相同人设" * 350, "人设权重分": 1, "帖子覆盖率": 1}
+        for _ in range(30)
+    )
+    cursor = None
+    handles: list[str] = []
+
+    while True:
+        page = await workbench.search_mission_context(
+            root_trace_id="root",
+            role="planner",
+            collection="inputs",
+            source_types=("persona",),
+            page_size=20,
+            cursor=cursor,
+        )
+        assert page["receipt"]["estimated_tokens"] <= 2_000
+        handles.extend(item["handle"] for item in page["page"]["items"])
+        cursor = page["page"]["next_cursor"]
+        if cursor is None:
+            break
+
+    assert len(handles) == len(set(handles)) == 30
+
+
+@pytest.mark.asyncio
+async def test_role_selected_persona_handles_keep_snapshot_identity_and_are_readable() -> None:
+    workbench, ledger, _snapshots, _ = _fixture()
+    state = await workbench.worker_state(
+        RoleContextRequest(
+            role=AgentRole.WORKER,
+            root_trace_id="root",
+            task_id="task-57",
+            spec_version=1,
+            task_spec={},
+            attempt_id="persona-attempt",
+            ledger_revision=ledger.revision,
+        )
+    )
+    persona_handles = [
+        item["handle"] for item in state.context_bundle["cards"] if item["source_type"] == "persona"
+    ]
+
+    assert len(persona_handles) == 12
+    for handle in persona_handles:
+        page = await workbench.read_mission_context(
+            root_trace_id="root",
+            role="worker",
+            task_id="task-57",
+            attempt_id="persona-attempt",
+            handle=handle,
+        )
+        assert page["source_handle"] == handle
+        assert page["content_fragment"]
+
+
+@pytest.mark.asyncio
+async def test_cursor_revision_tamper_and_cross_root_are_rejected() -> None:
+    workbench, ledger, _snapshots, _ = _fixture()
+    first = await workbench.search_mission_context(
+        root_trace_id="root", role="planner", collection="tasks", page_size=5
+    )
+    cursor = first["page"]["next_cursor"]
+    assert cursor
+    with pytest.raises(WorkbenchError, match="invalid context cursor"):
+        await workbench.search_mission_context(
+            root_trace_id="root",
+            role="planner",
+            collection="tasks",
+            cursor=cursor[:-1] + ("A" if cursor[-1] != "A" else "B"),
+            page_size=5,
+        )
+    with pytest.raises(WorkbenchError) as cross_root:
+        await workbench.search_mission_context(
+            root_trace_id="other-root",
+            role="planner",
+            collection="tasks",
+            cursor=cursor,
+            page_size=5,
+        )
+    assert cross_root.value.code == "CONTEXT_CURSOR_SCOPE_MISMATCH"
+    ledger.revision += 1
+    with pytest.raises(WorkbenchError) as stale:
+        await workbench.search_mission_context(
+            root_trace_id="root",
+            role="planner",
+            collection="tasks",
+            cursor=cursor,
+            page_size=5,
+        )
+    assert stale.value.code == "STALE_WORKBENCH_STATE"
+
+
+@pytest.mark.asyncio
+async def test_inputs_and_193_patterns_are_bounded_but_fully_recoverable() -> None:
+    workbench, ledger, _snapshots, _ = _fixture()
+    inputs = await workbench.search_mission_context(
+        root_trace_id="root", role="planner", collection="inputs", query="optional"
+    )
+    strategy = next(item for item in inputs["page"]["items"] if item["source_type"] == "strategy")
+    detail = await workbench.read_mission_context(
+        root_trace_id="root", role="planner", handle=strategy["handle"]
+    )
+    assert estimate_context_tokens(detail) <= 2_000
+    assert isinstance(detail["next_cursor"], str)
+    with pytest.raises(WorkbenchError) as tampered_detail:
+        await workbench.read_mission_context(
+            root_trace_id="root",
+            role="planner",
+            handle=strategy["handle"],
+            cursor=detail["next_cursor"][:-1] + "x",
+        )
+    assert tampered_detail.value.code == "CONTEXT_CURSOR_INVALID"
+    fragments = [detail["content_fragment"]]
+    while detail["next_cursor"] is not None:
+        detail = await workbench.read_mission_context(
+            root_trace_id="root",
+            role="planner",
+            handle=strategy["handle"],
+            cursor=detail["next_cursor"],
+        )
+        fragments.append(detail["content_fragment"])
+    assert "按需策略" in "".join(fragments)
+
+    cursor = None
+    immutable_cursor = None
+    found: list[dict[str, Any]] = []
+    while True:
+        page = await workbench.search_mission_context(
+            root_trace_id="root",
+            role="planner",
+            collection="patterns",
+            cursor=cursor,
+            page_size=20,
+        )
+        found.extend(page["page"]["items"])
+        cursor = page["page"]["next_cursor"]
+        immutable_cursor = immutable_cursor or cursor
+        if cursor is None:
+            break
+    assert len(found) == 193
+    assert len({item["handle"] for item in found}) == 193
+    raw_prompt = json.dumps(
+        (await workbench.planner_state("root")).to_payload(), ensure_ascii=False
+    )
+    assert len(raw_prompt) < 24_000
+    assert "模式详情" * 20 not in raw_prompt
+    ledger.revision += 1
+    immutable_page = await workbench.search_mission_context(
+        root_trace_id="root",
+        role="planner",
+        collection="patterns",
+        cursor=immutable_cursor,
+        page_size=20,
+    )
+    assert immutable_page["page"]["total"] == 193
+    assert workbench.broker._raw_artifacts.reads == 1
+
+
+@pytest.mark.asyncio
+async def test_semantic_selector_cannot_escape_allowlist(tmp_path) -> None:
+    calls = 0
+
+    async def llm_call(**_: Any) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        return {"content": json.dumps({"handles": ["bad_handle"]})}
+
+    selector = AdaptiveSemanticSelector(llm_call=llm_call, cache_root=tmp_path)
+    workbench, ledger, snapshots, _ = _fixture()
+    workbench = MissionWorkbench(
+        task_store=_TaskStore(ledger),
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=workbench.broker._artifacts,
+        contracts=workbench.broker._contracts,
+        raw_artifacts=workbench.broker._raw_artifacts,
+        semantic_selector=selector,
+    )
+    result = await workbench.search_mission_context(
+        root_trace_id="root", role="planner", collection="inputs", query="not present"
+    )
+    handles = {item["handle"] for item in result["page"]["items"]}
+    assert "bad_handle" not in handles
+    assert result["receipt"]["semantic_calls"] == 1
+    assert result["receipt"]["fallback_reason"]
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_semantic_selector_caches_valid_allowlisted_order(tmp_path) -> None:
+    calls = 0
+
+    async def llm_call(**kwargs: Any) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        request = json.loads(kwargs["messages"][1]["content"])
+        handles = [item["handle"] for item in request["candidates"]]
+        return {"content": json.dumps({"handles": list(reversed(handles))})}
+
+    selector = AdaptiveSemanticSelector(llm_call=llm_call, cache_root=tmp_path)
+    cards = (
+        ContextCard(handle="one", source_type="input", summary="alpha"),
+        ContextCard(handle="two", source_type="input", summary="beta"),
+    )
+    first = await selector.select(
+        root_trace_id="root", query="similar", cards=cards, model_config={"model": "fake"}
+    )
+    second = await selector.select(
+        root_trace_id="root", query="similar", cards=cards, model_config={"model": "fake"}
+    )
+
+    assert first.handles == ("two", "one")
+    assert second.handles == first.handles
+    assert second.cache_hit is True
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_semantic_cache_survives_call_limit_and_timeout_falls_back(tmp_path) -> None:
+    calls = 0
+
+    async def llm_call(**_: Any) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        return {"content": json.dumps({"handles": ["one", "two"]})}
+
+    cards = (
+        ContextCard(handle="one", source_type="input", summary="alpha"),
+        ContextCard(handle="two", source_type="input", summary="beta"),
+    )
+    selector = AdaptiveSemanticSelector(
+        llm_call=llm_call, cache_root=tmp_path, mission_call_limit=1
+    )
+    first = await selector.select(
+        root_trace_id="root", query="same", cards=cards, model_config={"model": "fake"}
+    )
+    restarted_selector = AdaptiveSemanticSelector(
+        llm_call=llm_call, cache_root=tmp_path, mission_call_limit=1
+    )
+    cached = await restarted_selector.select(
+        root_trace_id="root", query="same", cards=cards, model_config={"model": "fake"}
+    )
+    limited = await restarted_selector.select(
+        root_trace_id="root", query="different", cards=cards, model_config={"model": "fake"}
+    )
+
+    assert first.calls == 1
+    assert cached.cache_hit is True
+    assert limited.fallback_reason == "semantic_call_limit"
+    assert calls == 1
+
+    async def slow_call(**_: Any) -> dict[str, Any]:
+        await asyncio.sleep(0.05)
+        return {"content": "{}"}
+
+    timeout_selector = AdaptiveSemanticSelector(
+        llm_call=slow_call,
+        cache_root=tmp_path / "timeout",
+        timeout_seconds=0.001,
+    )
+    timed_out = await timeout_selector.select(
+        root_trace_id="timeout-root",
+        query="slow",
+        cards=cards,
+        model_config={"model": "fake"},
+    )
+    assert timed_out.calls == 1
+    assert timed_out.fallback_reason == "TimeoutError:semantic_fallback"
+
+
+@pytest.mark.asyncio
+async def test_semantic_rerank_never_removes_candidates_beyond_its_window(tmp_path) -> None:
+    semantic_first: list[str] = []
+
+    async def llm_call(**kwargs: Any) -> dict[str, Any]:
+        request = json.loads(kwargs["messages"][1]["content"])
+        handles = [item["handle"] for item in request["candidates"]]
+        semantic_first[:] = [handles[-1]]
+        return {"content": json.dumps({"handles": list(reversed(handles))})}
+
+    selector = AdaptiveSemanticSelector(llm_call=llm_call, cache_root=tmp_path)
+    workbench, ledger, snapshots, _ = _fixture()
+    workbench = MissionWorkbench(
+        task_store=_TaskStore(ledger),
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=workbench.broker._artifacts,
+        contracts=workbench.broker._contracts,
+        raw_artifacts=workbench.broker._raw_artifacts,
+        semantic_selector=selector,
+    )
+
+    cursor = None
+    handles: list[str] = []
+    while True:
+        result = await workbench.search_mission_context(
+            root_trace_id="root",
+            role="planner",
+            collection="inputs",
+            query="unmatched semantic query",
+            page_size=20,
+            cursor=cursor,
+        )
+        handles.extend(item["handle"] for item in result["page"]["items"])
+        cursor = result["page"]["next_cursor"]
+        if cursor is None:
+            break
+
+    assert len(handles) == 123
+    assert len(set(handles)) == 123
+    assert handles[0] == semantic_first[0]
+
+
+@pytest.mark.asyncio
+async def test_build_900049_shape_automatic_refresh_stays_bounded_and_reduces_by_half() -> None:
+    workbench, ledger, _snapshots, _ = _fixture()
+    refreshes = [await workbench.render("root", ledger) for _ in range(11)]
+    # The framework refreshes every five turns and marks each result output-only-once,
+    # so only the latest bounded projection remains in the provider prompt.
+    assert len(set(refreshes)) == 1
+    bounded = len(refreshes[-1])
+
+    legacy_round = json.dumps(
+        {
+            "task_tree": [task.current_spec.objective for task in ledger.tasks.values()],
+            "accepted_artifacts": ["x" * 4_000 for _ in range(33)],
+            "pattern_payload": ["模式详情" * 350 for _ in range(193)],
+        },
+        ensure_ascii=False,
+    )
+    legacy = len(legacy_round) * 54
+
+    assert bounded < legacy * 0.50
+    assert bounded < 24_000
+
+
+@pytest.mark.asyncio
+async def test_root_validator_gets_full_closure_as_required_pages_when_large() -> None:
+    workbench, ledger, snapshots, _ = _fixture()
+    contract_ref = "script-build://task-contracts/sha256/" + "d" * 64
+    contract = SimpleNamespace(
+        task_kind=ScriptTaskKind.ROOT_DELIVERY,
+        scope_ref="script-build://scopes/full",
+        intent_class=ScriptIntentClass.DELIVER,
+        objective="deliver",
+        goal_ids=("g-1",),
+        criteria=(ScriptCriterion("root", "complete"),),
+        input_decision_refs=(),
+        candidate_closure_decision_refs=(),
+        comparison_decision_refs=(),
+        base_artifact_ref=None,
+        to_payload=lambda: {"task_kind": "root-delivery"},
+    )
+    ledger.tasks["root-task"].current_spec = SimpleNamespace(
+        objective="deliver",
+        context_refs=("script-build://task-kinds/root-delivery", contract_ref),
+    )
+    workbench.broker._contracts.values[contract_ref] = contract
+    refs = tuple(
+        ArtifactRef(
+            uri=f"script-build://artifact-versions/{200 + index}",
+            kind=kind,
+            version=str(200 + index),
+            digest="sha256:" + f"{200 + index:064x}",
+        )
+        for index, kind in enumerate(("direction", "candidate-portfolio", "structured-script"))
+    )
+    for index, ref in enumerate(refs):
+        workbench.broker._artifacts.values[ref.uri] = _Artifact(
+            f"ROOT CLOSURE {index} " + "正文" * 12_000
+        )
+
+    class RootTools:
+        async def validation_evidence_refs(self, *, context: dict[str, Any]):
+            assert context["task_id"] == "root-task"
+            return refs
+
+    root_workbench = MissionWorkbench(
+        task_store=workbench.broker._task_store,
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=workbench.broker._artifacts,
+        contracts=workbench.broker._contracts,
+        root_tools=RootTools(),
+    )
+    request = RoleContextRequest(
+        role=AgentRole.VALIDATOR,
+        root_trace_id="root",
+        task_id="root-task",
+        spec_version=1,
+        task_spec={},
+        attempt_id="root-attempt",
+        ledger_revision=ledger.revision,
+        snapshot_id="root-snapshot",
+        artifact_snapshot={
+            "snapshot_id": "root-snapshot",
+            "artifact_refs": [asdict(refs[-1])],
+            "evidence_refs": [],
+        },
+    )
+    state = await root_workbench.validator_state(request)
+
+    assert state.artifact["full_content_in_bootstrap"] is False
+    assert len(state.required_handles) == 3
+    assert state.context_bundle["sufficiency"]["status"] == "ambiguous"
+    assert (
+        root_workbench.pending_required_context(
+            root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
+        )
+        == 3
+    )
+    for handle in state.required_handles:
+        cursor = None
+        while True:
+            page = await root_workbench.read_mission_context(
+                root_trace_id="root",
+                role="validator",
+                task_id="root-task",
+                attempt_id="root-attempt",
+                handle=handle,
+                cursor=cursor,
+            )
+            cursor = page["next_cursor"]
+            if cursor is None:
+                break
+    assert (
+        root_workbench.pending_required_context(
+            root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
+        )
+        == 0
+    )
+
+
+@pytest.mark.asyncio
+async def test_root_worker_can_expand_structured_script_and_validator_reads_survive_restart() -> (
+    None
+):
+    workbench, ledger, snapshots, _ = _fixture()
+    contract_ref = "script-build://task-contracts/sha256/" + "e" * 64
+    contract = SimpleNamespace(
+        task_kind=ScriptTaskKind.ROOT_DELIVERY,
+        scope_ref="script-build://scopes/full",
+        intent_class=ScriptIntentClass.DELIVER,
+        objective="deliver",
+        goal_ids=("g-1",),
+        criteria=(ScriptCriterion("root", "complete"),),
+        input_decision_refs=(),
+        candidate_closure_decision_refs=(),
+        comparison_decision_refs=(),
+        base_artifact_ref=None,
+        to_payload=lambda: {"task_kind": "root-delivery"},
+    )
+    ledger.tasks["root-task"].current_spec = SimpleNamespace(
+        objective="deliver",
+        context_refs=("script-build://task-kinds/root-delivery", contract_ref),
+    )
+    workbench.broker._contracts.values[contract_ref] = contract
+    refs = tuple(
+        ArtifactRef(
+            uri=f"script-build://artifact-versions/{500 + index}",
+            kind=kind,
+            version=str(500 + index),
+            digest="sha256:" + f"{500 + index:064x}",
+        )
+        for index, kind in enumerate(("direction", "candidate-portfolio", "structured-script"))
+    )
+    for index, ref in enumerate(refs):
+        workbench.broker._artifacts.values[ref.uri] = _Artifact(
+            f"ROOT SOURCE {index} " + "正文" * 12_000
+        )
+
+    class RootTools:
+        async def validation_evidence_refs(self, *, context: dict[str, Any]):
+            assert context["task_id"] == "root-task"
+            return refs
+
+    receipt_store = _ReceiptStore()
+    root_workbench = MissionWorkbench(
+        task_store=workbench.broker._task_store,
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=workbench.broker._artifacts,
+        contracts=workbench.broker._contracts,
+        root_tools=RootTools(),
+        receipt_store=receipt_store,
+    )
+    worker_request = RoleContextRequest(
+        role=AgentRole.WORKER,
+        root_trace_id="root",
+        task_id="root-task",
+        spec_version=1,
+        task_spec={},
+        attempt_id="root-attempt",
+        ledger_revision=ledger.revision,
+    )
+    worker = await root_workbench.worker_state(worker_request)
+    source_types = {item["source_type"] for item in worker.context_bundle["cards"]}
+    assert {"direction", "candidate-portfolio", "structured-script"}.issubset(source_types)
+    structured_handle = next(
+        item["handle"]
+        for item in worker.context_bundle["cards"]
+        if item["source_type"] == "structured-script"
+    )
+    detail = await root_workbench.read_mission_context(
+        root_trace_id="root",
+        role="worker",
+        task_id="root-task",
+        attempt_id="root-attempt",
+        handle=structured_handle,
+    )
+    assert "ROOT SOURCE 2" in detail["content_fragment"]
+
+    validator_request = RoleContextRequest(
+        role=AgentRole.VALIDATOR,
+        root_trace_id="root",
+        task_id="root-task",
+        spec_version=1,
+        task_spec={},
+        attempt_id="root-attempt",
+        ledger_revision=ledger.revision,
+        snapshot_id="root-snapshot",
+        artifact_snapshot={
+            "snapshot_id": "root-snapshot",
+            "artifact_refs": [asdict(refs[-1])],
+            "evidence_refs": [],
+        },
+    )
+    validator_state = await root_workbench.validator_state(validator_request)
+    restarted = MissionWorkbench(
+        task_store=workbench.broker._task_store,
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=workbench.broker._artifacts,
+        contracts=workbench.broker._contracts,
+        root_tools=RootTools(),
+        receipt_store=receipt_store,
+    )
+    with pytest.raises(WorkbenchError) as incomplete:
+        await restarted.verify_context_exhausted(
+            root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
+        )
+    assert incomplete.value.code == "CONTEXT_NOT_EXHAUSTED"
+    for handle in validator_state.required_handles:
+        cursor = None
+        while True:
+            page = await root_workbench.read_mission_context(
+                root_trace_id="root",
+                role="validator",
+                task_id="root-task",
+                attempt_id="root-attempt",
+                handle=handle,
+                cursor=cursor,
+            )
+            cursor = page["next_cursor"]
+            if cursor is None:
+                break
+    await restarted.verify_context_exhausted(
+        root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
+    )
+
+
+@pytest.mark.asyncio
+async def test_compare_worker_receives_all_and_only_frozen_candidates() -> None:
+    workbench, ledger, _snapshots, _ = _fixture()
+    task = ledger.tasks["task-57"]
+    contract_ref = "script-build://task-contracts/sha256/" + "9" * 64
+    refs = tuple(
+        AcceptedDecisionRef(
+            decision_id=f"decision-{index}",
+            artifact_ref=ledger.attempts[f"attempt-{index}"].submission.artifact_refs[0],
+            scope_ref=f"script-build://scopes/full/part-{index}",
+            expected_task_kind=ScriptTaskKind.PARAGRAPH,
+        )
+        for index in range(25)
+    )
+    direction_ref = AcceptedDecisionRef(
+        decision_id="decision-32",
+        artifact_ref=ledger.attempts["attempt-32"].submission.artifact_refs[0],
+        scope_ref="script-build://scopes/direction",
+        expected_task_kind=ScriptTaskKind.DIRECTION,
+    )
+    contract = ScriptTaskContractV1(
+        task_kind=ScriptTaskKind.COMPARE,
+        scope_ref="script-build://scopes/full/compare",
+        intent_class=ScriptIntentClass.COMPARE,
+        objective="compare every frozen candidate",
+        # Production contracts add Direction as a normal input. It is protected
+        # policy context, not one of the candidates being compared.
+        input_decision_refs=(direction_ref,),
+        base_artifact_ref=None,
+        write_scope=("script-build://writes/compare",),
+        gap_ref=None,
+        output_schema="comparison-artifact/v1",
+        criteria=(ScriptCriterion("compare", "compare all candidates"),),
+        budget=ScriptTaskBudget(),
+        goal_ids=("g-1",),
+        comparison_decision_refs=refs,
+        compiler_manifest_digest="sha256:" + "a" * 64,
+    )
+    workbench.broker._contracts.values[contract_ref] = contract
+    task.current_spec = SimpleNamespace(
+        objective=contract.objective,
+        context_refs=("script-build://task-kinds/compare", contract_ref),
+    )
+    state = await workbench.worker_state(
+        RoleContextRequest(
+            role=AgentRole.WORKER,
+            root_trace_id="root",
+            task_id=task.task_id,
+            spec_version=1,
+            task_spec={},
+            attempt_id="compare-attempt",
+            ledger_revision=ledger.revision,
+        )
+    )
+
+    cards = state.context_bundle["cards"]
+    assert len(cards) == 25
+    assert {item["source_type"] for item in cards} == {"paragraph"}
+    assert set(state.context_bundle["required_handles"]) == {item["handle"] for item in cards}
+
+
+@pytest.mark.asyncio
+async def test_element_worker_gets_target_text_existing_dimensions_and_link_gaps() -> None:
+    workbench, ledger, snapshots, _ = _fixture()
+    task = ledger.tasks["task-57"]
+    contract_ref = "script-build://task-contracts/sha256/" + "8" * 64
+    paragraph_ref = AcceptedDecisionRef(
+        decision_id="decision-0",
+        artifact_ref=ledger.attempts["attempt-0"].submission.artifact_refs[0],
+        scope_ref="script-build://scopes/full/part-0",
+        expected_task_kind=ScriptTaskKind.PARAGRAPH,
+    )
+    contract = ScriptTaskContractV1(
+        task_kind=ScriptTaskKind.ELEMENT_SET,
+        scope_ref="script-build://scopes/full/part-0/elements",
+        intent_class=ScriptIntentClass.EXPAND,
+        objective="extract linked elements",
+        input_decision_refs=(paragraph_ref,),
+        base_artifact_ref=None,
+        write_scope=("script-build://writes/part-0/elements",),
+        gap_ref=None,
+        output_schema="element-set-artifact/v1",
+        criteria=(ScriptCriterion("elements", "cover paragraph"),),
+        budget=ScriptTaskBudget(),
+        goal_ids=("g-1",),
+        compiler_manifest_digest="sha256:" + "a" * 64,
+    )
+    workbench.broker._contracts.values[contract_ref] = contract
+    task.current_spec = SimpleNamespace(
+        objective=contract.objective,
+        context_refs=("script-build://task-kinds/element-set", contract_ref),
+    )
+
+    class Candidates:
+        calls = 0
+
+        async def read_attempt_workspace(self, *, context: dict[str, Any]):
+            self.calls += 1
+            assert context["task_id"] == task.task_id
+            base = {
+                "level": 1,
+                "theme": "theme",
+                "form": "form",
+                "function": "function",
+                "feeling": "feeling",
+                "description": "description",
+                "is_active": True,
+            }
+            return {
+                "paragraphs": [
+                    {
+                        **base,
+                        "paragraph_id": 1,
+                        "name": "opening",
+                        "paragraph_index": 1,
+                        "content_range": {"from": 0, "to": 20},
+                        "full_description": "real paragraph text",
+                    },
+                    {
+                        **base,
+                        "paragraph_id": 2,
+                        "name": "ending",
+                        "paragraph_index": 2,
+                        "content_range": {"from": 21, "to": 40},
+                        "full_description": "second paragraph text",
+                    },
+                ],
+                "elements": [
+                    {
+                        "element_id": 1,
+                        "name": "conflict",
+                        "dimension_primary": "实质",
+                        "dimension_secondary": "叙事冲突",
+                        "is_active": True,
+                    }
+                ],
+                "paragraph_element_links": [{"paragraph_id": 1, "element_id": 1}],
+            }
+
+    element_workbench = MissionWorkbench(
+        task_store=workbench.broker._task_store,
+        bindings=_Bindings(),
+        snapshots=snapshots,
+        artifacts=workbench.broker._artifacts,
+        contracts=workbench.broker._contracts,
+        candidates=Candidates(),
+    )
+    state = await element_workbench.worker_state(
+        RoleContextRequest(
+            role=AgentRole.WORKER,
+            root_trace_id="root",
+            task_id=task.task_id,
+            spec_version=1,
+            task_spec={},
+            attempt_id="element-attempt",
+            ledger_revision=ledger.revision,
+        )
+    )
+
+    assert len(state.paragraph_targets) == 2
+    assert all(
+        "paragraph_id" not in item and "content_excerpt" not in item
+        for item in state.paragraph_targets
+    )
+    assert state.workspace_summary["unlinked_paragraph_count"] == 1
+    assert state.workspace_summary["unlinked_element_count"] == 0
+    assert state.workspace_summary["unlinked_paragraph_target_keys"] == [
+        state.paragraph_targets[1]["paragraph_target_key"]
+    ]
+    assert state.workspace_summary["unlinked_element_handles"] == []
+    cards = state.context_bundle["cards"]
+    assert {item["source_type"] for item in cards} == {"paragraph", "existing_element"}
+    assert any("实质/叙事冲突" in item["summary"] for item in cards)
+    element_card = next(item for item in cards if item["source_type"] == "existing_element")
+    assert element_card["metadata"]["linked_paragraph_target_keys"] == [
+        state.paragraph_targets[0]["paragraph_target_key"]
+    ]
+    for _ in range(49):
+        await element_workbench.read_mission_context(
+            root_trace_id="root",
+            role="worker",
+            task_id=task.task_id,
+            attempt_id="element-attempt",
+            handle=state.paragraph_targets[0]["paragraph_target_key"],
+            section="outline",
+        )
+    assert element_workbench.broker._candidates.calls == 1
+
+
+@pytest.mark.asyncio
+async def test_direction_worker_always_receives_always_on_strategy_before_optional_inputs() -> None:
+    workbench, ledger, _snapshots, _ = _fixture()
+    task = ledger.tasks["task-57"]
+    contract_ref = "script-build://task-contracts/sha256/" + "7" * 64
+    contract = ScriptTaskContractV1(
+        task_kind=ScriptTaskKind.DIRECTION,
+        scope_ref="script-build://scopes/direction",
+        intent_class=ScriptIntentClass.EXPLORE,
+        objective="choose a direction",
+        input_decision_refs=(),
+        base_artifact_ref=None,
+        write_scope=("script-build://writes/direction",),
+        gap_ref=None,
+        output_schema="script-direction/v1",
+        criteria=(ScriptCriterion("direction", "coherent direction"),),
+        budget=ScriptTaskBudget(),
+        goal_ids=(),
+        compiler_manifest_digest="sha256:" + "a" * 64,
+    )
+    workbench.broker._contracts.values[contract_ref] = contract
+    task.current_spec = SimpleNamespace(
+        objective=contract.objective,
+        context_refs=("script-build://task-kinds/direction", contract_ref),
+    )
+    state = await workbench.worker_state(
+        RoleContextRequest(
+            role=AgentRole.WORKER,
+            root_trace_id="root",
+            task_id=task.task_id,
+            spec_version=1,
+            task_spec={},
+            attempt_id="direction-attempt",
+            ledger_revision=ledger.revision,
+        )
+    )
+
+    cards = state.context_bundle["cards"]
+    assert cards[0]["source_type"] == "strategy"
+    assert cards[0]["metadata"]["mode"] == "always_on"
+    assert {item["source_type"] for item in cards}.issubset(
+        {"topic", "persona", "section_pattern", "strategy"}
+    )