工作台：建立有界任务状态与角色投影

新增 PlannerState、WorkerState、ValidatorState、语义 handle 与增量 revision。Planner 只接收活动子图和 ACCEPT frontier，Worker 自动获得角色化 Bootstrap，Validator 保留完整正文；详情分页、NotModified 与大小上限共同阻断历史任务和大素材反复进入上下文。

script_build_host/src/script_build_host/application/mission_workbench.py

@@ -0,0 +1,660 @@
+"""Bounded read and command surface for one script-build Mission."""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from collections.abc import Mapping
+from dataclasses import asdict, is_dataclass
+from hashlib import sha256
+from typing import Any, cast
+
+from agent.failures import FailureDetail, FailureDisposition, ToolExecutionError
+from agent.orchestration import AgentRole, DecisionAction, RoleContextRequest, TaskStatus
+
+from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
+from script_build_host.domain.ports import (
+    InputSnapshotRepository,
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.domain.task_contracts import (
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+    task_kind_from_context_refs,
+)
+from script_build_host.domain.workbench import (
+    NotModified,
+    PlannerState,
+    ValidatorState,
+    WorkerState,
+    paragraph_target_key,
+    protect_model_value,
+    semantic_handle,
+    strategy_handle,
+)
+
+TASK_CONTRACT_REF_PREFIX = "script-build://task-contracts/sha256/"
+_ACTIVE = {
+    TaskStatus.PENDING,
+    TaskStatus.RUNNING,
+    TaskStatus.AWAITING_VALIDATION,
+    TaskStatus.VALIDATING,
+    TaskStatus.AWAITING_DECISION,
+    TaskStatus.NEEDS_REPLAN,
+    TaskStatus.WAITING_CHILDREN,
+    TaskStatus.BLOCKED,
+}
+
+
+class WorkbenchError(RuntimeError):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+class MissionWorkbench:
+    """Projects durable Mission state without becoming a second source of truth."""
+
+    def __init__(
+        self,
+        *,
+        task_store: Any,
+        bindings: MissionBindingRepository,
+        snapshots: InputSnapshotRepository,
+        artifacts: ScriptBusinessArtifactRepository,
+        contracts: ScriptTaskContractStore,
+        candidates: Any | None = None,
+    ) -> None:
+        self._task_store = task_store
+        self._bindings = bindings
+        self._snapshots = snapshots
+        self._artifacts = artifacts
+        self._contracts = contracts
+        self._candidates = candidates
+
+    async def render(self, root_trace_id: str, ledger: Any) -> str:
+        state = await self.planner_state(root_trace_id, ledger=ledger)
+        return _bounded_json(state.to_payload(), 24_000, "PlannerState")
+
+    async def build(self, request: RoleContextRequest) -> Mapping[str, Any]:
+        try:
+            if request.role is AgentRole.WORKER:
+                return _json_safe((await self.worker_state(request)).to_payload())
+            if request.role is AgentRole.VALIDATOR:
+                return _json_safe((await self.validator_state(request)).to_payload())
+            return {}
+        except WorkbenchError as exc:
+            raise ToolExecutionError(
+                FailureDetail(
+                    code=exc.code,
+                    message=exc.message,
+                    disposition=FailureDisposition.ABORT_RUN,
+                    source_tool="mission_workbench_role_context",
+                    details={"task_id": request.task_id, "role": request.role.value},
+                )
+            ) from exc
+
+    async def planner_state(
+        self,
+        root_trace_id: str,
+        *,
+        known_revision: str | None = None,
+        ledger: Any | None = None,
+    ) -> PlannerState | NotModified:
+        ledger = ledger or await self._task_store.load(root_trace_id)
+        revision = f"ledger:{ledger.revision}"
+        if known_revision == revision:
+            return NotModified(revision)
+        binding = await self._bindings.get_by_root(root_trace_id)
+        counts = Counter(task.status.value for task in ledger.tasks.values())
+        included = self._active_subgraph_ids(ledger)
+        active_tasks: list[dict[str, Any]] = []
+        accepted: list[dict[str, Any]] = []
+        covered_goals: set[str] = set()
+        all_goals: tuple[str, ...] = ()
+        for task in sorted(ledger.tasks.values(), key=lambda item: item.display_path):
+            contract = (
+                None
+                if task.task_id == ledger.root_task_id
+                else await self._contract(root_trace_id, task)
+            )
+            if contract is not None:
+                covered_goals.update(
+                    contract.goal_ids if self._accepted_decision(ledger, task) is not None else ()
+                )
+            if task.task_id in included:
+                active_tasks.append(self._task_summary(task, contract))
+            decision = self._accepted_decision(ledger, task)
+            if decision is not None and contract is not None:
+                accepted.append(
+                    {
+                        "decision_id": decision.decision_id,
+                        "task_id": task.task_id,
+                        "task_kind": contract.task_kind.value,
+                        "scope_selector": _scope_selector(contract.scope_ref),
+                        "goal_ids": list(contract.goal_ids),
+                        "summary": decision.reason[:300],
+                    }
+                )
+        if binding.active_direction_artifact_version_id is not None:
+            version = await self._artifacts.get_by_id(
+                binding.active_direction_artifact_version_id,
+                script_build_id=binding.script_build_id,
+            )
+            goals = getattr(version.artifact, "goals", ())
+            all_goals = tuple(str(item.goal_id) for item in goals)
+        root = ledger.tasks[ledger.root_task_id]
+        failures = self._latest_failures(ledger)
+        operations = tuple(
+            {
+                "operation_id": item.operation_id,
+                "kind": item.kind.value,
+                "status": item.status.value,
+                "task_count": len(item.task_ids),
+            }
+            for item in ledger.operations.values()
+            if item.status.value in {"pending", "running", "stop_requested"}
+        )
+        state = PlannerState(
+            state_revision=revision,
+            phase=_phase(ledger, root, binding.active_direction_artifact_version_id),
+            root={
+                "task_id": root.task_id,
+                "status": root.status.value,
+                "blocked_reason": root.blocked_reason,
+                "objective": root.current_spec.objective,
+            },
+            focused_task_id=ledger.focused_task_id,
+            active_subgraph=tuple(active_tasks[:24]),
+            accepted_frontier=tuple(accepted[-32:]),
+            goal_progress={
+                "total": len(all_goals),
+                "covered": len(set(all_goals) & covered_goals),
+                "uncovered_goal_ids": [item for item in all_goals if item not in covered_goals],
+            },
+            counts_by_status=dict(sorted(counts.items())),
+            active_operations=operations,
+            latest_failures=failures,
+            retired_task_count=sum(
+                1 for item in ledger.tasks.values() if item.task_id not in included
+            ),
+            changes=self._recent_changes(ledger),
+        )
+        _bounded_json(state.to_payload(), 24_000, "PlannerState")
+        return state
+
+    async def worker_state(self, request: RoleContextRequest) -> WorkerState:
+        ledger = await self._task_store.load(request.root_trace_id)
+        task = ledger.tasks[request.task_id]
+        contract = await self._contract(request.root_trace_id, task)
+        if contract is None:
+            raise WorkbenchError("TASK_CONTRACT_INVALID", "Worker Task has no contract")
+        binding = await self._bindings.get_by_root(request.root_trace_id)
+        snapshot = await self._snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=binding.script_build_id
+        )
+        semantic_ref_values: list[Any] = []
+        seen_decisions: set[str] = set()
+        for item in (*contract.input_decision_refs, *contract.comparison_decision_refs):
+            if item.decision_id in seen_decisions:
+                continue
+            seen_decisions.add(item.decision_id)
+            semantic_ref_values.append(item)
+        semantic_refs = tuple(semantic_ref_values)
+        accepted_values = [self._accepted_input_handle(item) for item in semantic_refs]
+        for index, item in enumerate(semantic_refs):
+            version = await self._artifacts.read_by_ref(
+                item.artifact_ref, script_build_id=binding.script_build_id
+            )
+            raw_ref = getattr(version.artifact, "raw_artifact_ref", None)
+            if isinstance(raw_ref, str) and raw_ref:
+                accepted_values[index]["image_handle"] = semantic_handle("image", raw_ref)
+        for child in request.accepted_child_results:
+            submission = child.get("submission")
+            if not isinstance(submission, Mapping):
+                continue
+            for raw in [
+                *submission.get("artifact_refs", ()),
+                *submission.get("evidence_refs", ()),
+            ]:
+                if not isinstance(raw, Mapping):
+                    continue
+                accepted_values.append(
+                    {
+                        "decision_id": child.get("decision_id"),
+                        "source_handle": _source_handle(
+                            str(raw.get("uri", "")),
+                            str(raw.get("digest") or raw.get("version") or ""),
+                        ),
+                        "task_kind": task_kind_from_context_refs(
+                            tuple(child.get("task_spec", {}).get("context_refs", ()))
+                        ),
+                        "summary": raw.get("summary"),
+                    }
+                )
+        accepted = tuple(accepted_values)
+        workspace = await self._workspace(request, contract)
+        workspace_digest = _digest(workspace)
+        paragraphs = tuple(
+            {
+                "paragraph_target_key": paragraph_target_key(
+                    request.root_trace_id,
+                    request.task_id,
+                    request.attempt_id,
+                    int(item["paragraph_id"]),
+                ),
+                "name": item.get("name"),
+                "paragraph_index": item.get("paragraph_index"),
+                "level": item.get("level"),
+                "content_range": item.get("content_range"),
+                "content_complete": all(
+                    bool(str(item.get(field) or "").strip())
+                    for field in (
+                        "theme",
+                        "form",
+                        "function",
+                        "feeling",
+                        "description",
+                        "full_description",
+                    )
+                ),
+            }
+            for item in workspace.get("paragraphs", ())
+        )
+        revision = _digest(
+            {
+                "ledger": ledger.revision,
+                "workspace": workspace_digest,
+                "contract": contract.to_payload(),
+            }
+        )
+        latest = ledger.attempts.get(task.attempt_ids[-2]) if len(task.attempt_ids) > 1 else None
+        state = WorkerState(
+            state_revision=revision,
+            task=_semantic_contract(contract),
+            input_summary=_input_summary(snapshot),
+            accepted_inputs=accepted,
+            source_handles=accepted,
+            paragraph_targets=paragraphs,
+            workspace_summary={
+                "paragraph_count": len(workspace.get("paragraphs", ())),
+                "element_count": len(workspace.get("elements", ())),
+                "link_count": len(workspace.get("paragraph_element_links", ())),
+                "workspace_digest": workspace_digest,
+            },
+            latest_failure=(
+                latest.failure.to_dict()
+                if latest is not None and latest.failure is not None
+                else None
+            ),
+        )
+        _bounded_json(state.to_payload(), 32_000, "WorkerState")
+        return state
+
+    async def validator_state(self, request: RoleContextRequest) -> ValidatorState:
+        ledger = await self._task_store.load(request.root_trace_id)
+        task = ledger.tasks[request.task_id]
+        contract = await self._contract(request.root_trace_id, task)
+        if contract is None or request.artifact_snapshot is None:
+            raise WorkbenchError("TASK_CONTRACT_INVALID", "Validator context is incomplete")
+        binding = await self._bindings.get_by_root(request.root_trace_id)
+        refs = [
+            *request.artifact_snapshot.get("artifact_refs", ()),
+            *request.artifact_snapshot.get("evidence_refs", ()),
+            *(asdict(item.artifact_ref) for item in contract.input_decision_refs),
+        ]
+        handles: list[dict[str, Any]] = []
+        artifacts: list[dict[str, Any]] = []
+        seen_uris: set[str] = set()
+        for raw in refs:
+            uri = str(raw.get("uri", ""))
+            if not uri or uri in seen_uris:
+                continue
+            seen_uris.add(uri)
+            handle = _source_handle(uri, str(raw.get("digest") or raw.get("version") or ""))
+            handles.append(
+                {"evidence_handle": handle, "kind": raw.get("kind"), "summary": raw.get("summary")}
+            )
+            ref = _artifact_ref(raw)
+            version = await self._artifacts.read_by_ref(
+                ref, script_build_id=binding.script_build_id
+            )
+            artifacts.append(
+                {
+                    "source_handle": handle,
+                    "content": _protected_artifact_payload(version.artifact, handle),
+                }
+            )
+        artifact_payload = {"items": artifacts}
+        if len(_json(artifact_payload)) > 80_000:
+            raise WorkbenchError(
+                "CONTEXT_BUDGET_EXCEEDED", "Validator Artifact exceeds the protected role budget"
+            )
+        state = ValidatorState(
+            state_revision=_digest(
+                {
+                    "ledger": ledger.revision,
+                    "snapshot": request.artifact_snapshot,
+                    "contract": contract.to_payload(),
+                }
+            ),
+            task=_semantic_contract(contract),
+            artifact=artifact_payload,
+            evidence_handles=tuple(handles),
+            accepted_closure=tuple(
+                self._accepted_input_handle(item) for item in contract.input_decision_refs
+            ),
+        )
+        metadata = state.to_payload() | {"artifact": {"item_count": len(artifacts)}}
+        _bounded_json(metadata, 16_000, "ValidatorState metadata")
+        return state
+
+    async def detail(
+        self,
+        *,
+        root_trace_id: str,
+        view: str,
+        handle: str | None = None,
+        known_revision: str | None = None,
+        cursor: int = 0,
+    ) -> Mapping[str, Any]:
+        ledger = await self._task_store.load(root_trace_id)
+        revision = f"ledger:{ledger.revision}"
+        if known_revision == revision:
+            return NotModified(revision).to_payload()
+        if view == "source":
+            if not handle:
+                raise WorkbenchError("INVALID_TOOL_ARGUMENT", "source view requires a handle")
+            binding = await self._bindings.get_by_root(root_trace_id)
+            accepted_attempts = {
+                item.attempt_id
+                for item in ledger.decisions.values()
+                if item.action is DecisionAction.ACCEPT and item.attempt_id is not None
+            }
+            selected = None
+            for attempt_id in accepted_attempts:
+                attempt = ledger.attempts.get(attempt_id)
+                if attempt is None or attempt.submission is None:
+                    continue
+                for ref in [*attempt.submission.artifact_refs, *attempt.submission.evidence_refs]:
+                    if _source_handle(ref.uri, ref.digest or ref.version or "") == handle:
+                        selected = ref
+                        break
+                if selected is not None:
+                    break
+            if selected is None:
+                raise WorkbenchError("CHILD_DECISION_INVALID", "source handle is stale or unknown")
+            version = await self._artifacts.read_by_ref(
+                selected, script_build_id=binding.script_build_id
+            )
+            content = _json(_protected_artifact_payload(version.artifact, handle))
+            start = max(cursor, 0)
+            fragment = content[start : start + 7_000]
+            payload = {
+                "state_revision": revision,
+                "view": view,
+                "source_handle": handle,
+                "content_fragment": fragment,
+                "next_cursor": start + len(fragment)
+                if start + len(fragment) < len(content)
+                else None,
+            }
+            _bounded_json(payload, 8_000, "Workbench detail")
+            return payload
+        if view != "task_tree":
+            raise WorkbenchError("INVALID_TOOL_ARGUMENT", "unsupported Workbench detail view")
+        tasks = sorted(ledger.tasks.values(), key=lambda item: item.display_path)
+        page = tasks[max(cursor, 0) : max(cursor, 0) + 20]
+        tree_payload: dict[str, Any] = {
+            "state_revision": revision,
+            "view": view,
+            "items": [
+                {
+                    "task_id": item.task_id,
+                    "parent_task_id": item.parent_task_id,
+                    "display_path": item.display_path,
+                    "status": item.status.value,
+                    "objective": item.current_spec.objective[:250],
+                }
+                for item in page
+            ],
+            "next_cursor": cursor + len(page) if cursor + len(page) < len(tasks) else None,
+        }
+        _bounded_json(tree_payload, 8_000, "Workbench detail")
+        return tree_payload
+
+    async def _contract(self, root_trace_id: str, task: Any) -> ScriptTaskContractV1 | None:
+        refs = [
+            item
+            for item in task.current_spec.context_refs
+            if item.startswith(TASK_CONTRACT_REF_PREFIX)
+        ]
+        if len(refs) != 1:
+            return None
+        return (await self._contracts.read(root_trace_id, refs[0])).contract
+
+    async def _workspace(
+        self, request: RoleContextRequest, contract: ScriptTaskContractV1
+    ) -> Mapping[str, Any]:
+        if self._candidates is None or contract.task_kind not in {
+            ScriptTaskKind.STRUCTURE,
+            ScriptTaskKind.PARAGRAPH,
+            ScriptTaskKind.ELEMENT_SET,
+        }:
+            return {}
+        try:
+            return cast(
+                Mapping[str, Any],
+                await self._candidates.read_attempt_workspace(
+                    context={
+                        "root_trace_id": request.root_trace_id,
+                        "task_id": request.task_id,
+                        "attempt_id": request.attempt_id,
+                        "spec_version": request.spec_version,
+                    }
+                ),
+            )
+        except Exception as exc:
+            if getattr(exc, "code", None) in {
+                "LEGACY_ARTIFACT_NOT_FOUND",
+                "TASK_KIND_PRESET_MISMATCH",
+            }:
+                return {}
+            raise
+
+    @staticmethod
+    def _accepted_decision(ledger: Any, task: Any) -> Any | None:
+        if not task.decision_ids:
+            return None
+        decision = ledger.decisions.get(task.decision_ids[-1])
+        if decision is not None and decision.action is DecisionAction.ACCEPT:
+            return decision
+        return None
+
+    @staticmethod
+    def _active_subgraph_ids(ledger: Any) -> set[str]:
+        included = {ledger.root_task_id}
+        for task in ledger.tasks.values():
+            if task.status not in _ACTIVE:
+                continue
+            current = task
+            while current is not None and current.task_id not in included:
+                included.add(current.task_id)
+                current = ledger.tasks.get(current.parent_task_id)
+        return {item for item in included if item is not None}
+
+    @staticmethod
+    def _task_summary(task: Any, contract: ScriptTaskContractV1 | None) -> dict[str, Any]:
+        return {
+            "task_id": task.task_id,
+            "parent_task_id": task.parent_task_id,
+            "display_path": task.display_path,
+            "status": task.status.value,
+            "task_kind": contract.task_kind.value if contract is not None else "root",
+            "scope_selector": _scope_selector(contract.scope_ref) if contract is not None else None,
+            "objective": task.current_spec.objective,
+            "attempt_count": len(task.attempt_ids),
+            "blocked_reason": task.blocked_reason,
+        }
+
+    @staticmethod
+    def _latest_failures(ledger: Any) -> tuple[Mapping[str, Any], ...]:
+        values: list[dict[str, Any]] = []
+        for attempt in reversed(list(ledger.attempts.values())):
+            if attempt.failure is None:
+                continue
+            values.append({"task_id": attempt.task_id, **attempt.failure.to_dict()})
+            if len(values) == 5:
+                break
+        return tuple(values)
+
+    @staticmethod
+    def _recent_changes(ledger: Any) -> tuple[Mapping[str, Any], ...]:
+        changes: list[dict[str, Any]] = []
+        for decision in list(ledger.decisions.values())[-5:]:
+            changes.append(
+                {
+                    "kind": "decision",
+                    "task_id": decision.task_id,
+                    "action": decision.action.value,
+                }
+            )
+        for attempt in list(ledger.attempts.values())[-5:]:
+            changes.append(
+                {
+                    "kind": "attempt",
+                    "task_id": attempt.task_id,
+                    "status": attempt.status.value,
+                    "has_submission": attempt.submission is not None,
+                }
+            )
+        return tuple(changes[-8:])
+
+    @staticmethod
+    def _accepted_input_handle(item: Any) -> dict[str, Any]:
+        return {
+            "decision_id": item.decision_id,
+            "source_handle": _source_handle(
+                item.artifact_ref.uri, item.artifact_ref.digest or item.artifact_ref.version or ""
+            ),
+            "task_kind": item.expected_task_kind.value,
+            "scope_selector": _scope_selector(item.scope_ref),
+            "summary": item.artifact_ref.summary,
+        }
+
+
+def _semantic_contract(contract: ScriptTaskContractV1) -> dict[str, Any]:
+    return {
+        "task_kind": contract.task_kind.value,
+        "scope_selector": _scope_selector(contract.scope_ref),
+        "intent_class": contract.intent_class.value,
+        "objective": contract.objective,
+        "goal_ids": list(contract.goal_ids),
+        "criteria": [item.to_payload() for item in contract.criteria],
+        "input_decision_ids": [item.decision_id for item in contract.input_decision_refs],
+        "comparison_decision_ids": [item.decision_id for item in contract.comparison_decision_refs],
+        "base_decision_id": next(
+            (
+                item.decision_id
+                for item in contract.input_decision_refs
+                if contract.base_artifact_ref is not None
+                and item.artifact_ref.uri == contract.base_artifact_ref.uri
+            ),
+            None,
+        ),
+    }
+
+
+def _input_summary(snapshot: Any) -> dict[str, Any]:
+    topic = snapshot.topic.get("topic") if isinstance(snapshot.topic, Mapping) else None
+    return {
+        "snapshot_id": snapshot.snapshot_id,
+        "topic_id": snapshot.topic_id,
+        "topic": protect_model_value(
+            topic if isinstance(topic, Mapping) else snapshot.topic,
+            namespace=f"input:{snapshot.snapshot_id}",
+        ),
+        "account_name": snapshot.account.get("account_name"),
+        "persona_point_count": len(snapshot.persona_points),
+        "section_pattern_count": len(snapshot.section_patterns),
+        "strategies": [
+            {
+                "strategy_handle": strategy_handle(snapshot.snapshot_id, index, item),
+                "name": item.get("name") or item.get("strategy_name"),
+                "mode": item.get("mode"),
+            }
+            for index, item in enumerate(snapshot.strategies)
+        ],
+    }
+
+
+def _phase(ledger: Any, root: Any, active_direction_id: int | None) -> int:
+    if root.blocked_reason == "PHASE_ONE_CAPABILITY_BOUNDARY":
+        return 1
+    if root.blocked_reason == "PHASE_TWO_CANDIDATE_PORTFOLIO_READY":
+        return 2
+    if any(
+        task_kind_from_context_refs(item.current_spec.context_refs)
+        == ScriptTaskKind.ROOT_DELIVERY.value
+        for item in ledger.tasks.values()
+    ):
+        return 3
+    return 2 if active_direction_id is not None else 1
+
+
+def _scope_selector(scope_ref: str) -> dict[str, Any]:
+    prefix = "script-build://scopes/"
+    path = scope_ref.removeprefix(prefix).split("/") if scope_ref.startswith(prefix) else []
+    return {"anchor": "mission", "path": [item for item in path if item]}
+
+
+def _source_handle(uri: str, immutable: str) -> str:
+    return semantic_handle("src", uri, immutable)
+
+
+def _artifact_ref(raw: Mapping[str, Any]) -> Any:
+    from agent.orchestration import ArtifactRef
+
+    return ArtifactRef.from_dict(dict(raw))
+
+
+def _artifact_payload(artifact: Any) -> Mapping[str, Any]:
+    content = getattr(artifact, "content_payload", None)
+    if callable(content):
+        return cast(Mapping[str, Any], content())
+    if is_dataclass(artifact):
+        return cast(Mapping[str, Any], asdict(cast(Any, artifact)))
+    return {"value": str(artifact)}
+
+
+def _protected_artifact_payload(artifact: Any, source_handle: str) -> Mapping[str, Any]:
+    """Expose semantics while replacing persisted identities and references with handles."""
+
+    return cast(
+        Mapping[str, Any],
+        protect_model_value(_artifact_payload(artifact), namespace=source_handle),
+    )
+
+
+def _digest(value: Any) -> str:
+    return "sha256:" + sha256(_json(value).encode()).hexdigest()
+
+
+def _json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
+
+
+def _json_safe(value: Any) -> Mapping[str, Any]:
+    return cast(Mapping[str, Any], json.loads(_json(value)))
+
+
+def _bounded_json(value: Any, limit: int, label: str) -> str:
+    encoded = _json(value)
+    if len(encoded) > limit:
+        raise WorkbenchError("CONTEXT_BUDGET_EXCEEDED", f"{label} exceeds {limit} characters")
+    return encoded
+
+
+__all__ = ["MissionWorkbench", "WorkbenchError"]

script_build_host/src/script_build_host/domain/workbench.py

@@ -0,0 +1,201 @@
+"""Mission Workbench read models and semantic handles."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass, field
+from hashlib import sha256
+from typing import Any
+
+
+@dataclass(frozen=True, slots=True)
+class NotModified:
+    state_revision: str
+
+    def to_payload(self) -> dict[str, Any]:
+        return {"state_revision": self.state_revision, "not_modified": True}
+
+
+@dataclass(frozen=True, slots=True)
+class PlannerState:
+    state_revision: str
+    phase: int
+    root: Mapping[str, Any]
+    focused_task_id: str | None
+    active_subgraph: tuple[Mapping[str, Any], ...]
+    accepted_frontier: tuple[Mapping[str, Any], ...]
+    goal_progress: Mapping[str, Any]
+    counts_by_status: Mapping[str, int]
+    active_operations: tuple[Mapping[str, Any], ...]
+    latest_failures: tuple[Mapping[str, Any], ...]
+    retired_task_count: int
+    changes: tuple[Mapping[str, Any], ...] = ()
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class WorkerState:
+    state_revision: str
+    task: Mapping[str, Any]
+    input_summary: Mapping[str, Any]
+    accepted_inputs: tuple[Mapping[str, Any], ...]
+    source_handles: tuple[Mapping[str, Any], ...]
+    paragraph_targets: tuple[Mapping[str, Any], ...]
+    workspace_summary: Mapping[str, Any]
+    latest_failure: Mapping[str, Any] | None = None
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class ValidatorState:
+    state_revision: str
+    task: Mapping[str, Any]
+    artifact: Mapping[str, Any]
+    evidence_handles: tuple[Mapping[str, Any], ...]
+    accepted_closure: tuple[Mapping[str, Any], ...]
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class CommandReceipt:
+    state_revision: str
+    status: str
+    created: int = 0
+    updated: int = 0
+    linked: int = 0
+    changed_target_keys: tuple[str, ...] = ()
+    workspace_digest: str | None = None
+    details: Mapping[str, Any] = field(default_factory=dict)
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def semantic_handle(prefix: str, *parts: object) -> str:
+    encoded = "\0".join(str(item) for item in parts).encode()
+    return f"{prefix}_" + sha256(encoded).hexdigest()[:20]
+
+
+def strategy_handle(snapshot_id: str, index: int, value: Mapping[str, Any]) -> str:
+    identity = value.get("source_uri") or value.get("strategy_id") or index
+    immutable = (
+        value.get("content_sha256")
+        or value.get("sha256")
+        or json.dumps(value, sort_keys=True, default=str)
+    )
+    return semantic_handle("strategy", snapshot_id, identity, immutable)
+
+
+def strategy_ref(snapshot_id: str, index: int, value: Mapping[str, Any]) -> str:
+    handle = strategy_handle(snapshot_id, index, value)
+    return f"script-build://inputs/{snapshot_id}/strategies/{handle}"
+
+
+def paragraph_target_key(root_trace_id: str, task_id: str, attempt_id: str, local_id: int) -> str:
+    return semantic_handle("pt", root_trace_id, task_id, attempt_id, local_id)
+
+
+def protect_model_value(value: Any, *, namespace: str) -> Any:
+    """Replace persisted identities and mechanical references with semantic handles."""
+
+    paragraph_handles: dict[str, str] = {}
+    element_handles: dict[str, str] = {}
+
+    def collect(current: Any) -> None:
+        if isinstance(current, Mapping):
+            if current.get("paragraph_id") is not None:
+                identity = str(current["paragraph_id"])
+                paragraph_handles.setdefault(
+                    identity, semantic_handle("paragraph", namespace, identity)
+                )
+            if current.get("element_id") is not None:
+                identity = str(current["element_id"])
+                element_handles.setdefault(
+                    identity, semantic_handle("element", namespace, identity)
+                )
+            for item in current.values():
+                collect(item)
+        elif isinstance(current, Sequence) and not isinstance(current, (str, bytes)):
+            for item in current:
+                collect(item)
+
+    collect(value)
+
+    def reference_handle(item: Any) -> str:
+        identity = (
+            item.get("uri") or item.get("url") or json.dumps(item, sort_keys=True, default=str)
+            if isinstance(item, Mapping)
+            else item
+        )
+        return semantic_handle("source", namespace, identity)
+
+    def scope_selector(item: Any) -> dict[str, Any]:
+        value = str(item or "")
+        prefix = "script-build://scopes/"
+        path = value.removeprefix(prefix).split("/") if value.startswith(prefix) else []
+        return {"anchor": "mission", "path": [part for part in path if part]}
+
+    def protect(current: Any) -> Any:
+        if isinstance(current, Mapping):
+            output: dict[str, Any] = {}
+            for raw_key, item in current.items():
+                name = str(raw_key)
+                lowered = name.lower()
+                if "digest" in lowered or lowered in {"canonical_sha256", "content_sha256"}:
+                    continue
+                if name == "paragraph_id":
+                    output["paragraph_handle"] = paragraph_handles[str(item)]
+                elif name in {"parent_id", "parent_paragraph_id"}:
+                    output["parent_paragraph_handle"] = (
+                        paragraph_handles.get(str(item)) if item is not None else None
+                    )
+                elif name == "element_id":
+                    output["element_handle"] = element_handles[str(item)]
+                elif name == "element_ids" and isinstance(item, Sequence):
+                    output["element_handles"] = [
+                        element_handles.get(
+                            str(identity), semantic_handle("element", namespace, identity)
+                        )
+                        for identity in item
+                    ]
+                elif name in {"scope_ref", "write_scope"}:
+                    output["scope_selector"] = scope_selector(item)
+                elif lowered.endswith("_ref"):
+                    output[f"{name[:-4]}_handle"] = reference_handle(item)
+                elif (
+                    lowered.endswith("_refs")
+                    and isinstance(item, Sequence)
+                    and not isinstance(item, (str, bytes))
+                ):
+                    output[f"{name[:-5]}_handles"] = [
+                        reference_handle(identity) for identity in item
+                    ]
+                else:
+                    output[name] = protect(item)
+            return output
+        if isinstance(current, Sequence) and not isinstance(current, (str, bytes)):
+            return [protect(item) for item in current]
+        return current
+
+    return protect(value)
+
+
+__all__ = [
+    "CommandReceipt",
+    "NotModified",
+    "PlannerState",
+    "ValidatorState",
+    "WorkerState",
+    "paragraph_target_key",
+    "protect_model_value",
+    "semantic_handle",
+    "strategy_handle",
+    "strategy_ref",
+]

script_build_host/tests/test_mission_workbench.py

@@ -0,0 +1,286 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from agent.orchestration import AgentRole, ArtifactRef, RoleContextRequest, TaskStatus
+
+from script_build_host.application.mission_workbench import MissionWorkbench
+from script_build_host.domain.task_contracts import (
+    ScriptCriterion,
+    ScriptIntentClass,
+    ScriptTaskBudget,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+)
+from script_build_host.domain.workbench import NotModified
+
+
+class _TaskStore:
+    def __init__(self, ledger: Any) -> None:
+        self.ledger = ledger
+
+    async def load(self, root_trace_id: str) -> Any:
+        assert root_trace_id == "root"
+        return self.ledger
+
+
+class _Bindings:
+    async def get_by_root(self, root_trace_id: str) -> Any:
+        assert root_trace_id == "root"
+        return SimpleNamespace(
+            root_trace_id="root",
+            script_build_id=900049,
+            input_snapshot_id=3,
+            active_direction_artifact_version_id=None,
+        )
+
+
+class _Snapshots:
+    async def get(self, snapshot_id: str, *, script_build_id: int) -> Any:
+        assert (snapshot_id, script_build_id) == ("3", 900049)
+        return SimpleNamespace(
+            snapshot_id="3",
+            topic_id=49,
+            topic={"topic": {"result": "one bounded topic"}},
+            account={"account_name": "fixture"},
+            persona_points=tuple({"name": f"point-{index}"} for index in range(80)),
+            section_patterns=tuple({"name": f"pattern-{index}"} for index in range(80)),
+            strategies=(),
+        )
+
+
+class _Contracts:
+    def __init__(self, contract: ScriptTaskContractV1 | None = None) -> None:
+        self.contract = contract
+
+    async def read(self, root_trace_id: str, contract_ref: str) -> Any:
+        assert root_trace_id == "root"
+        assert contract_ref.startswith("script-build://task-contracts/sha256/")
+        assert self.contract is not None
+        return SimpleNamespace(contract=self.contract)
+
+
+class _Artifact:
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "paragraphs": [
+                {
+                    "paragraph_id": 77,
+                    "name": "opening",
+                    "full_description": "complete candidate prose",
+                }
+            ],
+            "artifact_ref": "script-build://artifact-versions/77",
+            "canonical_sha256": "sha256:" + "a" * 64,
+        }
+
+
+class _Artifacts:
+    async def read_by_ref(self, ref: ArtifactRef, *, script_build_id: int) -> Any:
+        assert script_build_id == 900049
+        assert ref.uri == "script-build://artifact-versions/77"
+        return SimpleNamespace(artifact=_Artifact())
+
+
+class _Candidates:
+    async def read_attempt_workspace(self, *, context: dict[str, Any]) -> dict[str, Any]:
+        assert context["attempt_id"] == "attempt"
+        return {
+            "paragraphs": [
+                {
+                    "paragraph_id": 77,
+                    "paragraph_index": 1,
+                    "name": "opening",
+                    "content_range": {"scope": "opening"},
+                    "level": 1,
+                    "theme": "theme",
+                    "form": "form",
+                    "function": "function",
+                    "feeling": "feeling",
+                    "description": "description",
+                    "full_description": "complete prose",
+                }
+            ],
+            "elements": [],
+            "paragraph_element_links": [],
+        }
+
+
+def _root_task(*, children: list[str] | None = None) -> Any:
+    return SimpleNamespace(
+        task_id="root-task",
+        parent_task_id=None,
+        child_task_ids=list(children or []),
+        display_path="1",
+        status=TaskStatus.RUNNING,
+        blocked_reason=None,
+        current_spec=SimpleNamespace(objective="build script", context_refs=()),
+        current_spec_version=1,
+        attempt_ids=[],
+        decision_ids=[],
+    )
+
+
+def _contract() -> ScriptTaskContractV1:
+    return ScriptTaskContractV1(
+        task_kind=ScriptTaskKind.PARAGRAPH,
+        scope_ref="script-build://scopes/full/opening",
+        intent_class=ScriptIntentClass.EXPLORE,
+        objective="write the opening",
+        input_decision_refs=(),
+        base_artifact_ref=None,
+        write_scope=("script-build://writes/paragraphs/opening",),
+        gap_ref=None,
+        output_schema="paragraph-artifact/v1",
+        criteria=(ScriptCriterion("criterion", "complete prose"),),
+        budget=ScriptTaskBudget(),
+        goal_ids=("goal-1",),
+        compiler_manifest_digest="sha256:" + "b" * 64,
+    )
+
+
+def _worker_task() -> Any:
+    return SimpleNamespace(
+        task_id="worker",
+        parent_task_id="root-task",
+        child_task_ids=[],
+        display_path="1.1",
+        status=TaskStatus.RUNNING,
+        blocked_reason=None,
+        current_spec=SimpleNamespace(
+            objective="write the opening",
+            context_refs=(
+                "script-build://task-kinds/paragraph",
+                "script-build://task-contracts/sha256/contract",
+            ),
+        ),
+        current_spec_version=1,
+        attempt_ids=["attempt"],
+        decision_ids=[],
+    )
+
+
+@pytest.mark.asyncio
+async def test_planner_projection_is_bounded_incremental_and_does_not_copy_retired_tree() -> None:
+    tasks = {"root-task": _root_task(children=[f"done-{index}" for index in range(120)])}
+    for index in range(120):
+        tasks[f"done-{index}"] = SimpleNamespace(
+            task_id=f"done-{index}",
+            parent_task_id="root-task",
+            child_task_ids=[],
+            display_path=f"1.{index + 1}",
+            status=TaskStatus.COMPLETED,
+            blocked_reason=None,
+            current_spec=SimpleNamespace(
+                objective="retired detail " + "x" * 1000,
+                context_refs=(),
+            ),
+            current_spec_version=1,
+            attempt_ids=[],
+            decision_ids=[],
+        )
+    ledger = SimpleNamespace(
+        revision=49,
+        root_task_id="root-task",
+        focused_task_id="root-task",
+        tasks=tasks,
+        attempts={},
+        decisions={},
+        operations={},
+    )
+    workbench = MissionWorkbench(
+        task_store=_TaskStore(ledger),
+        bindings=_Bindings(),
+        snapshots=_Snapshots(),
+        artifacts=_Artifacts(),
+        contracts=_Contracts(),
+    )
+
+    state = await workbench.planner_state("root")
+    assert not isinstance(state, NotModified)
+    assert state.retired_task_count == 120
+    assert len(state.active_subgraph) == 1
+    encoded = await workbench.render("root", ledger)
+    assert len(encoded) < 24_000
+    assert "retired detail" not in encoded
+    legacy_full_tree = json.dumps(
+        [item.current_spec.objective for item in tasks.values()], ensure_ascii=False
+    )
+    assert len(encoded) * 2 < len(legacy_full_tree)
+    for _ in range(15):
+        assert isinstance(
+            await workbench.planner_state("root", known_revision="ledger:49"),
+            NotModified,
+        )
+    first_page = await workbench.detail(root_trace_id="root", view="task_tree")
+    assert len(first_page["items"]) == 20
+    assert len(json.dumps(first_page, ensure_ascii=False)) < 8_000
+
+
+@pytest.mark.asyncio
+async def test_role_bootstrap_uses_semantic_targets_and_protects_artifact_identities() -> None:
+    root = _root_task(children=["worker"])
+    task = _worker_task()
+    attempt = SimpleNamespace(attempt_id="attempt", failure=None)
+    ledger = SimpleNamespace(
+        revision=7,
+        root_task_id="root-task",
+        focused_task_id="worker",
+        tasks={"root-task": root, "worker": task},
+        attempts={"attempt": attempt},
+        decisions={},
+        operations={},
+    )
+    workbench = MissionWorkbench(
+        task_store=_TaskStore(ledger),
+        bindings=_Bindings(),
+        snapshots=_Snapshots(),
+        artifacts=_Artifacts(),
+        contracts=_Contracts(_contract()),
+        candidates=_Candidates(),
+    )
+    worker_request = RoleContextRequest(
+        role=AgentRole.WORKER,
+        root_trace_id="root",
+        task_id="worker",
+        spec_version=1,
+        task_spec={},
+        attempt_id="attempt",
+        ledger_revision=7,
+    )
+    worker = await workbench.build(worker_request)
+    encoded_worker = json.dumps(worker, ensure_ascii=False)
+    assert worker["paragraph_targets"][0]["paragraph_target_key"].startswith("pt_")
+    assert "paragraph_id" not in encoded_worker
+    assert len(encoded_worker) < 32_000
+
+    artifact_ref = ArtifactRef(
+        uri="script-build://artifact-versions/77",
+        kind="paragraph",
+        version="77",
+        digest="sha256:" + "a" * 64,
+    )
+    validator_request = RoleContextRequest(
+        role=AgentRole.VALIDATOR,
+        root_trace_id="root",
+        task_id="worker",
+        spec_version=1,
+        task_spec={},
+        attempt_id="attempt",
+        ledger_revision=7,
+        snapshot_id="snapshot",
+        artifact_snapshot={
+            "artifact_refs": [asdict(artifact_ref)],
+            "evidence_refs": [],
+        },
+    )
+    validator = await workbench.build(validator_request)
+    encoded_validator = json.dumps(validator, ensure_ascii=False)
+    assert "complete candidate prose" in encoded_validator
+    assert "paragraph_id" not in encoded_validator
+    assert "script-build://artifact-versions/77" not in encoded_validator
+    assert "canonical_sha256" not in encoded_validator