script-build-and-framework-tmp/script_build_host/tests/test_context_broker_integration.py

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"}
    )