统一编排证据授权与上下文预算

agent/agent/core/runner.py

@@ -1089,6 +1089,30 @@ class AgentRunner:
         budget["last_actual_prompt_tokens"] = actual_prompt_tokens
         budget["last_estimated_prompt_tokens"] = raw.estimated_tokens
         await self._persist_runtime_state(trace)
+        if self.trace_store:
+            packing = (
+                dict(trace.context.get("prompt_packing", {}))
+                if isinstance(trace.context, dict)
+                and isinstance(trace.context.get("prompt_packing"), dict)
+                else {}
+            )
+            await self.trace_store.append_event(
+                trace.trace_id,
+                "prompt_envelope_measured",
+                {
+                    "token_limit": int(
+                        packing.get("token_limit", raw.hard_limit)
+                    ),
+                    "output_reserve": int(packing.get("output_reserve", 0)),
+                    "initial_estimated_tokens": packing.get("estimated_tokens"),
+                    "estimated_prompt_tokens": raw.estimated_tokens,
+                    "actual_prompt_tokens": actual_prompt_tokens,
+                    "packing_mode": packing.get("mode", "runner"),
+                    "deduplicated_segments": int(
+                        packing.get("deduplicated_segments", 0)
+                    ),
+                },
+            )
 
     async def _finish_context_compression(
         self,
@@ -1712,6 +1736,49 @@ class AgentRunner:
                 config.model,
                 budget_state.get("calibration_factor"),
             )
+            from agent.orchestration.prompt_budget import (
+                PromptBudgetExceeded,
+                RolePromptEnvelopePacker,
+            )
+
+            try:
+                packed_envelope = RolePromptEnvelopePacker(
+                    token_limit=request_measurement.hard_limit,
+                    output_reserve=0,
+                ).pack_messages(
+                    llm_messages,
+                    tool_schemas=tool_schemas,
+                )
+            except PromptBudgetExceeded:
+                terminal_failure = await self._context_budget_exceeded(
+                    trace,
+                    request_measurement,
+                    reason="minimum_envelope_would_exceed_hard_limit",
+                    extra={"after_message_count": len(llm_messages)},
+                )
+                break
+            llm_messages = list(packed_envelope.messages)
+            trace.context["prompt_packing"] = {
+                **(
+                    dict(trace.context.get("prompt_packing", {}))
+                    if isinstance(trace.context.get("prompt_packing"), dict)
+                    else {}
+                ),
+                "mode": packed_envelope.packing_mode,
+                "estimated_tokens": packed_envelope.estimated_tokens,
+                "token_limit": request_measurement.hard_limit,
+                "output_reserve": 0,
+                "deduplicated_segments": (
+                    packed_envelope.deduplicated_segments
+                ),
+            }
+            request_measurement = measure_prompt_tokens(
+                llm_messages,
+                tool_schemas,
+                config.compression,
+                config.model,
+                budget_state.get("calibration_factor"),
+            )
             if request_measurement.calibrated_tokens > request_measurement.hard_limit:
                 terminal_failure = await self._context_budget_exceeded(
                     trace,
@@ -3849,6 +3916,16 @@ class AgentRunner:
             get_tool_schemas=self._get_tool_schemas,
         )
 
+    def get_run_tool_schemas(self, config: RunConfig) -> List[Dict]:
+        """Expose the exact final tool schemas for pre-call envelope packing."""
+
+        return self._get_run_tool_schemas(config)
+
+    async def get_run_system_prompt(self, config: RunConfig) -> str:
+        """Expose the exact system prompt used for a new model run."""
+
+        return str(await self._build_system_prompt(config) or "")
+
     def _build_protected_tool_context(
         self,
         config: RunConfig,

agent/agent/orchestration/__init__.py

@@ -35,6 +35,8 @@ from .models import (
     OperationStatus,
     OrchestrationEvent,
     PlannerDecision,
+    ResourceClaim,
+    FrozenContextDocument,
     TaskAttempt,
     TaskCycleResult,
     TaskLedger,
@@ -44,6 +46,9 @@ from .models import (
     ValidationMode,
     ValidationPlan,
     ValidationReport,
+    ValidationEvidenceGrant,
+    ValidationEvidenceSnapshot,
+    ValidationReadSession,
     ValidationRunStatus,
     ValidationVerdict,
 )
@@ -77,6 +82,8 @@ from .context_provider import (
     RoleContextProvider,
     RoleContextRequest,
     TaskContextProvider,
+    ValidationContextBootstrap,
+    ValidationEvidenceProvider,
 )
 from .deterministic_worker import (
     DeterministicWorker,
@@ -144,6 +151,8 @@ __all__ = [
     "OrchestrationError",
     "OrchestrationEvent",
     "PlannerDecision",
+    "ResourceClaim",
+    "FrozenContextDocument",
     "PlannerDecisionView",
     "ProblemDetails",
     "RevisionConflict",
@@ -168,10 +177,15 @@ __all__ = [
     "TaskStoreError",
     "TaskStoreNotFound",
     "ValidationContext",
+    "ValidationContextBootstrap",
+    "ValidationEvidenceProvider",
     "ValidationMode",
     "ValidationPlan",
     "ValidationPolicy",
     "ValidationReport",
+    "ValidationEvidenceGrant",
+    "ValidationEvidenceSnapshot",
+    "ValidationReadSession",
     "ValidationRunStatus",
     "ValidationVerdict",
     "create_orchestration_router",

agent/agent/orchestration/api_v2.py

@@ -239,6 +239,14 @@ def create_orchestration_router(coordinator: TaskCoordinator) -> Any:
         async def action() -> OperationView:
             if isinstance(body, DispatchOperationRequest):
                 request = body
+                dispatch_options: dict[str, Any] = {}
+                if request.resource_claims:
+                    from .models import ResourceClaim
+
+                    dispatch_options["resource_claims"] = [
+                        ResourceClaim(item.uri, item.exclusive)
+                        for item in request.resource_claims
+                    ]
                 operation = await coordinator.start_operation(
                     root_trace_id,
                     request.kind,
@@ -246,6 +254,7 @@ def create_orchestration_router(coordinator: TaskCoordinator) -> Any:
                     worker_presets=request.worker_presets,
                     deadline_at=request.deadline_at,
                     idempotency_key=idempotency_key,
+                    **dispatch_options,
                 )
             else:
                 request = body

agent/agent/orchestration/context_provider.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import Any, Mapping, Protocol
 
-from .models import AgentRole, TaskLedger
+from .models import AgentRole, TaskLedger, ValidationEvidenceGrant
 
 
 class TaskContextProvider(Protocol):
@@ -37,4 +37,27 @@ class RoleContextProvider(Protocol):
     async def build(self, request: RoleContextRequest) -> Mapping[str, Any]: ...
 
 
-__all__ = ["RoleContextProvider", "RoleContextRequest", "TaskContextProvider"]
+@dataclass(frozen=True, slots=True)
+class ValidationContextBootstrap:
+    """Host projection that Coordinator freezes before a Validator starts."""
+
+    role_context: Mapping[str, Any]
+    evidence_grants: tuple[ValidationEvidenceGrant, ...]
+    required_handles: tuple[str, ...] = ()
+
+
+class ValidationEvidenceProvider(Protocol):
+    """Build one Validation's initial immutable evidence projection."""
+
+    async def build_validation_context(
+        self, request: RoleContextRequest
+    ) -> ValidationContextBootstrap: ...
+
+
+__all__ = [
+    "RoleContextProvider",
+    "RoleContextRequest",
+    "TaskContextProvider",
+    "ValidationContextBootstrap",
+    "ValidationEvidenceProvider",
+]

agent/agent/orchestration/coordinator.py

@@ -6,9 +6,10 @@ import asyncio
 import hashlib
 import json
 import logging
+from datetime import datetime, timedelta, timezone
 from dataclasses import replace
 from dataclasses import asdict, is_dataclass
-from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
+from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
 from uuid import uuid4
 
 from agent.failures import FailureDetail, ToolExecutionError
@@ -24,12 +25,15 @@ from .models import (
     CommandRecord,
     CriterionResult,
     DecisionAction,
+    DurableLeaseRecord,
     EventDraft,
     ExecutionStats,
     FailureCode,
+    FrozenContextDocument,
     OperationKind,
     OperationStatus,
     PlannerDecision,
+    ResourceClaim,
     TaskAttempt,
     TaskCycleResult,
     TaskLedger,
@@ -37,6 +41,7 @@ from .models import (
     TaskSpec,
     TaskStatus,
     ValidationReport,
+    ValidationEvidenceGrant,
     ValidationMode,
     ValidationPlan,
     ValidationRunStatus,
@@ -73,12 +78,23 @@ from .validation_policy import (
     ValidationContext,
     ValidationPolicy,
 )
+from .validation_evidence import (
+    cancel_read,
+    complete_read,
+    evidence_handle,
+    extend_snapshot,
+    require_authorized_refs,
+    require_complete,
+    reserve_read,
+    start_read_session,
+)
 from ._decision_engine import DecisionEngine
 from ._task_context import render_task_context
 from .context_provider import (
     RoleContextProvider,
     RoleContextRequest,
     TaskContextProvider,
+    ValidationContextBootstrap,
 )
 from .deterministic_worker import DeterministicWorker, DeterministicWorkerContext
 from ._task_graph import (
@@ -256,6 +272,7 @@ class TaskCoordinator:
             )
             if task.status not in TERMINAL_TASK_STATUSES
         ]
+        blocking_failures = self._blocking_failures(ledger)
         return {
             "root_task_id": root.task_id,
             "root_objective": ledger.root_objective,
@@ -263,7 +280,68 @@ class TaskCoordinator:
             "blocked_reason": root.blocked_reason,
             "result_summary": result_summary,
             "pending_tasks": pending,
+            "blocking_failures": blocking_failures,
+        }
+
+    @staticmethod
+    def _blocking_failures(ledger: TaskLedger) -> List[Dict[str, Any]]:
+        """Expose only the latest unresolved runtime failure per active Task."""
+
+        blocking: List[Dict[str, Any]] = []
+        resolved_task_statuses = {
+            TaskStatus.COMPLETED,
+            TaskStatus.CANCELLED,
+            TaskStatus.SUPERSEDED,
         }
+        for task in ledger.tasks.values():
+            if task.status in resolved_task_statuses:
+                continue
+            attempts = [
+                ledger.attempts[item]
+                for item in task.attempt_ids
+                if item in ledger.attempts
+            ]
+            validations = [
+                ledger.validations[item]
+                for item in task.validation_ids
+                if item in ledger.validations
+            ]
+            latest_validation = next(
+                (item for item in reversed(validations) if item.failure is not None),
+                None,
+            )
+            if latest_validation is not None and not any(
+                item.created_at > latest_validation.created_at
+                and item.status is ValidationRunStatus.COMPLETED
+                for item in validations
+            ):
+                blocking.append(
+                    {
+                        "task_id": task.task_id,
+                        "attempt_id": latest_validation.attempt_id,
+                        "validation_id": latest_validation.validation_id,
+                        **latest_validation.failure.to_dict(),
+                    }
+                )
+                continue
+            latest_attempt = next(
+                (item for item in reversed(attempts) if item.failure is not None),
+                None,
+            )
+            if latest_attempt is not None and not any(
+                item.created_at > latest_attempt.created_at
+                and item.status is AttemptStatus.SUBMITTED
+                for item in attempts
+            ):
+                blocking.append(
+                    {
+                        "task_id": task.task_id,
+                        "attempt_id": latest_attempt.attempt_id,
+                        "validation_id": None,
+                        **latest_attempt.failure.to_dict(),
+                    }
+                )
+        return blocking
 
     async def mission_completion(self, root_trace_id: str) -> Dict[str, Any]:
         """Compatibility alias for callers using the pre-Root terminology."""
@@ -502,12 +580,21 @@ class TaskCoordinator:
         parent_task_id: Optional[str] = None,
         placement: Optional[Dict[str, Any]] = None,
         idempotency_key: Optional[str] = None,
+        context_documents: Optional[Sequence["FrozenContextDocument"]] = None,
     ) -> Dict[str, Any]:
         if not drafts:
             raise ValueError("At least one task draft is required")
         placement = dict(placement or {})
+        documents = tuple(context_documents or ())
 
         def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            for document in documents:
+                existing = ledger.context_documents.get(document.uri)
+                if existing is not None and existing != document:
+                    raise TaskConflict(
+                        f"frozen context document conflict: {document.uri}"
+                    )
+                ledger.context_documents[document.uri] = document
             after_task_id = placement.get("after_task_id")
             if after_task_id:
                 target = _task(ledger, after_task_id)
@@ -603,10 +690,279 @@ class TaskCoordinator:
                 "drafts": list(drafts),
                 "parent_task_id": parent_task_id,
                 "placement": placement,
+                "context_documents": [_plain(asdict(item)) for item in documents],
             },
         )
         return await self._tasks_result(root_trace_id, result["task_ids"])
 
+    async def get_context_document(
+        self, root_trace_id: str, uri: str
+    ) -> "FrozenContextDocument":
+        ledger = await self.task_store.load(root_trace_id)
+        try:
+            return ledger.context_documents[uri]
+        except KeyError as exc:
+            raise ValueError(f"Context document not found: {uri}") from exc
+
+    async def replay_command(
+        self,
+        root_trace_id: str,
+        *,
+        operation: str,
+        idempotency_key: str,
+        payload: Any,
+    ) -> Dict[str, Any] | None:
+        """Read a durable command result before rebuilding mutable snapshots."""
+
+        ledger = await self.task_store.load(root_trace_id)
+        return _command_replay(
+            ledger,
+            self._idem(root_trace_id, idempotency_key),
+            operation,
+            payload,
+        )
+
+    async def get_tasks(
+        self, root_trace_id: str, task_ids: Sequence[str]
+    ) -> Dict[str, Any]:
+        return await self._tasks_result(root_trace_id, task_ids)
+
+    async def reserve_durable_lease(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        lease_seconds: int,
+    ) -> Dict[str, Any]:
+        now = datetime.now(timezone.utc)
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            existing = ledger.durable_leases.get(lease_key)
+            if existing is not None and existing.status == "completed":
+                return {
+                    "lease_key": lease_key,
+                    "owner": existing.owner,
+                    "epoch": existing.epoch,
+                    "acquired": False,
+                    "terminal_result": existing.terminal_result,
+                    "journal": list(existing.journal),
+                }
+            expired = (
+                existing is None
+                or datetime.fromisoformat(existing.expires_at) <= now
+            )
+            if existing is not None and not expired:
+                return {
+                    "lease_key": lease_key,
+                    "owner": existing.owner,
+                    "epoch": existing.epoch,
+                    "acquired": False,
+                    "terminal_result": None,
+                    "journal": list(existing.journal),
+                }
+            epoch = 1 if existing is None else existing.epoch + int(expired)
+            record = DurableLeaseRecord(
+                lease_key=lease_key,
+                owner=owner,
+                epoch=epoch,
+                status="reserved",
+                expires_at=(now + timedelta(seconds=lease_seconds)).isoformat(),
+                journal=[] if existing is None else list(existing.journal),
+            )
+            ledger.durable_leases[lease_key] = record
+            return {
+                "lease_key": lease_key,
+                "owner": owner,
+                "epoch": epoch,
+                "acquired": True,
+                "terminal_result": None,
+                "journal": list(record.journal),
+            }
+
+        return await self._mutate(
+            root_trace_id,
+            "durable_lease_reserved",
+            mutate,
+        )
+
+    async def append_durable_lease_event(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        epoch: int,
+        event: Mapping[str, Any],
+    ) -> None:
+        now = datetime.now(timezone.utc)
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            lease = ledger.durable_leases[lease_key]
+            if (
+                lease.owner != owner
+                or lease.epoch != epoch
+                or lease.status != "reserved"
+                or datetime.fromisoformat(lease.expires_at) <= now
+            ):
+                raise TaskConflict("durable lease ownership is stale")
+            lease.journal.append(dict(event))
+            return {"lease_key": lease_key, "journal_size": len(lease.journal)}
+
+        await self._mutate(root_trace_id, "durable_lease_event_appended", mutate)
+
+    async def renew_durable_lease(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        epoch: int,
+        lease_seconds: int,
+    ) -> None:
+        now = datetime.now(timezone.utc)
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            lease = ledger.durable_leases[lease_key]
+            if (
+                lease.owner != owner
+                or lease.epoch != epoch
+                or lease.status != "reserved"
+                or datetime.fromisoformat(lease.expires_at) <= now
+            ):
+                raise TaskConflict("durable lease ownership is stale")
+            lease.expires_at = (
+                now + timedelta(seconds=lease_seconds)
+            ).isoformat()
+            return {"lease_key": lease_key, "expires_at": lease.expires_at}
+
+        await self._mutate(root_trace_id, "durable_lease_renewed", mutate)
+
+    async def complete_durable_lease(
+        self,
+        root_trace_id: str,
+        lease_key: str,
+        *,
+        owner: str,
+        epoch: int,
+        result: Mapping[str, Any],
+    ) -> Dict[str, Any]:
+        now = datetime.now(timezone.utc)
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            lease = ledger.durable_leases[lease_key]
+            if lease.status == "completed":
+                return dict(lease.terminal_result or {})
+            if (
+                lease.owner != owner
+                or lease.epoch != epoch
+                or lease.status != "reserved"
+                or datetime.fromisoformat(lease.expires_at) <= now
+            ):
+                raise TaskConflict("durable lease ownership is stale")
+            lease.status = "completed"
+            lease.terminal_result = dict(result)
+            return dict(result)
+
+        return await self._mutate(root_trace_id, "durable_lease_completed", mutate)
+
+    async def create_task_graph(
+        self,
+        root_trace_id: str,
+        drafts: Sequence[Dict[str, Any]],
+        *,
+        idempotency_key: str,
+        context_documents: Sequence["FrozenContextDocument"] = (),
+        command_payload: Any | None = None,
+    ) -> Dict[str, Any]:
+        """Atomically freeze documents and create or revise a typed task plan."""
+
+        if not drafts:
+            raise ValueError("At least one task draft is required")
+        documents = tuple(context_documents)
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            for document in documents:
+                existing = ledger.context_documents.get(document.uri)
+                if existing is not None and existing != document:
+                    raise TaskConflict(
+                        f"frozen context document conflict: {document.uri}"
+                    )
+                ledger.context_documents[document.uri] = document
+            created: List[str] = []
+            for draft in drafts:
+                revision_task_id = draft.get("_revision_task_id")
+                if revision_task_id is not None:
+                    task_id = str(revision_task_id)
+                    task = ledger.tasks.get(task_id)
+                    if task is None:
+                        raise ValueError(f"Revision task not found: {task_id}")
+                    if task.status is not TaskStatus.NEEDS_REPLAN:
+                        raise TaskConflict(
+                            f"Task {task_id} cannot be revised from "
+                            f"{task.status.value}"
+                        )
+                    if (
+                        draft.get("_parent_task_id") is not None
+                        and str(draft["_parent_task_id"]) != task.parent_task_id
+                    ):
+                        raise TaskConflict(
+                            f"Task {task_id} cannot change its frozen parent"
+                        )
+                    version = task.current_spec_version + 1
+                    task.specs.append(
+                        self._task_spec_from_draft(draft, version=version)
+                    )
+                    task.current_spec_version = version
+                    transition(task.status, TaskStatus.PENDING)
+                    task.status = TaskStatus.PENDING
+                    task.updated_at = utc_now()
+                    created.append(task_id)
+                    continue
+                raw_parent = draft.get("_parent_task_id") or ledger.root_task_id
+                parent_id = str(raw_parent)
+                parent = ledger.tasks.get(parent_id)
+                if parent is None:
+                    raise ValueError(f"Parent task not found: {parent_id}")
+                if parent.status not in {
+                    TaskStatus.PENDING,
+                    TaskStatus.NEEDS_REPLAN,
+                    TaskStatus.WAITING_CHILDREN,
+                }:
+                    raise TaskConflict(
+                        f"Task {parent_id} cannot receive children from "
+                        f"{parent.status.value}"
+                    )
+                sibling_count = sum(
+                    item.parent_task_id == parent_id for item in ledger.tasks.values()
+                )
+                task_id = self._add_task_record(
+                    ledger,
+                    {key: value for key, value in draft.items() if key != "_parent_task_id"},
+                    parent_id,
+                    self._display_path(ledger, parent_id, sibling_count + 1),
+                )
+                parent.child_task_ids.append(task_id)
+                self._mark_parent_waiting(parent)
+                created.append(task_id)
+            return {"task_ids": created}
+
+        result = await self._mutate(
+            root_trace_id,
+            "task_graph_created",
+            mutate,
+            idempotency_key,
+            idempotency_payload=(
+                command_payload
+                if command_payload is not None
+                else {
+                    "drafts": list(drafts),
+                    "context_documents": [_plain(asdict(item)) for item in documents],
+                }
+            ),
+        )
+        return await self._tasks_result(root_trace_id, result["task_ids"])
+
     async def focus_task(self, root_trace_id: str, task_id: str) -> Dict[str, Any]:
         def mutate(ledger: TaskLedger) -> Dict[str, Any]:
             task = _task(ledger, task_id)
@@ -621,6 +977,7 @@ class TaskCoordinator:
         task_ids: Sequence[str],
         worker_presets: Optional[Sequence[str]] = None,
         idempotency_key: Optional[str] = None,
+        resource_claims: Optional[Sequence["ResourceClaim"]] = None,
     ) -> List[TaskCycleResult]:
         if not task_ids:
             return []
@@ -630,6 +987,7 @@ class TaskCoordinator:
             task_ids=task_ids,
             worker_presets=worker_presets,
             idempotency_key=idempotency_key,
+            resource_claims=resource_claims,
         )
         operation = await self.await_operation(root_trace_id, operation.operation_id)
         return await self._operation_cycle_results(operation)
@@ -645,6 +1003,7 @@ class TaskCoordinator:
         attempt_id: Optional[str] = None,
         idempotency_key: Optional[str] = None,
         deadline_at: Optional[str] = None,
+        resource_claims: Optional[Sequence["ResourceClaim"]] = None,
     ) -> BackgroundOperation:
         return await self._operations.start(
             root_trace_id,
@@ -655,6 +1014,7 @@ class TaskCoordinator:
             attempt_id=attempt_id,
             idempotency_key=idempotency_key,
             deadline_at=deadline_at,
+            resource_claims=resource_claims,
         )
 
     async def get_operation(
@@ -1666,6 +2026,7 @@ class TaskCoordinator:
                 attempt=attempt,
                 snapshot=snapshot,
                 prior_validation_count=max(len(task.validation_ids) - 1, 0),
+                context_documents=ledger.context_documents,
             )
             preflight_result = await self._run_validation_preflight(
                 validation_context,
@@ -1689,7 +2050,7 @@ class TaskCoordinator:
                     timeout=timeout,
                 )
             else:
-                validator_context["role_context"] = await self._build_role_context(
+                bootstrap = await self._build_validation_context(
                     RoleContextRequest(
                         role=AgentRole.VALIDATOR,
                         root_trace_id=root_trace_id,
@@ -1707,6 +2068,13 @@ class TaskCoordinator:
                         },
                     )
                 )
+                await self.initialize_validation_evidence(
+                    root_trace_id,
+                    validation_id,
+                    grants=bootstrap.evidence_grants,
+                    required_handles=bootstrap.required_handles,
+                )
+                validator_context["role_context"] = _plain(bootstrap.role_context)
                 validator_result = await asyncio.wait_for(
                     self.executor.run_validator(validator_context),  # type: ignore[union-attr]
                     timeout=timeout,
@@ -1792,6 +2160,24 @@ class TaskCoordinator:
             value = dict(value)
         return _plain(value)
 
+    async def _build_validation_context(
+        self, request: RoleContextRequest
+    ) -> ValidationContextBootstrap:
+        if self.role_context_provider is None:
+            return ValidationContextBootstrap({}, ())
+        builder = getattr(self.role_context_provider, "build_validation_context", None)
+        if callable(builder):
+            value = await builder(request)
+            if not isinstance(value, ValidationContextBootstrap):
+                raise TypeError(
+                    "build_validation_context must return ValidationContextBootstrap"
+                )
+            return value
+        return ValidationContextBootstrap(
+            await self._build_role_context(request),
+            (),
+        )
+
     async def _reuse_semantic_validation(
         self,
         context: ValidationContext,
@@ -1826,6 +2212,19 @@ class TaskCoordinator:
                 or prior_snapshot.evidence_refs != context.snapshot.evidence_refs
             ):
                 continue
+            await self.initialize_validation_evidence(
+                context.root_trace_id,
+                validation.validation_id,
+                grants=tuple(
+                    ValidationEvidenceGrant(
+                        artifact_ref=ref,
+                        handle=evidence_handle(ref),
+                        source="reused_validation",
+                    )
+                    for ref in prior.evidence_refs
+                ),
+                required_handles=(),
+            )
             await self.submit_validation(
                 {
                     "role": AgentRole.VALIDATOR.value,
@@ -1886,6 +2285,12 @@ class TaskCoordinator:
             if item.verdict == ValidationVerdict.FAILED
         ]
         reason = "; ".join(f"{item.rule_id}: {item.reason}" for item in failures)
+        await self.initialize_validation_evidence(
+            context.root_trace_id,
+            validation.validation_id,
+            grants=(),
+            required_handles=(),
+        )
         await self.submit_validation(
             {
                 "role": AgentRole.VALIDATOR.value,
@@ -1965,6 +2370,19 @@ class TaskCoordinator:
         evidence_refs = [
             ref for item in result.rule_results for ref in item.evidence_refs
         ]
+        await self.initialize_validation_evidence(
+            context.root_trace_id,
+            validation.validation_id,
+            grants=tuple(
+                ValidationEvidenceGrant(
+                    artifact_ref=ref,
+                    handle=evidence_handle(ref),
+                    source="deterministic_validation",
+                )
+                for ref in evidence_refs
+            ),
+            required_handles=(),
+        )
         errors = result.errors
         summary = (
             f"Deterministic validation {result.verdict.value}; "
@@ -2228,6 +2646,7 @@ class TaskCoordinator:
                 ledger.validations[item].attempt_id == attempt_id
                 for item in task.validation_ids
             ),
+            context_documents=ledger.context_documents,
         )
         plan = self.validation_policy.plan(context)
         if not isinstance(plan, ValidationPlan):
@@ -2316,14 +2735,39 @@ class TaskCoordinator:
             or task.status != TaskStatus.VALIDATING
         ):
             raise TaskConflict("Validation can only be submitted once while running")
-        self._validate_report(task, verdict, criterion_results)
-
         def mutate(current: TaskLedger) -> Dict[str, Any]:
             current_task = _task(current, task_id)
             report = current.validations[validation_id]
+            if (
+                report.task_id != task_id
+                or report.attempt_id != attempt_id
+                or report.snapshot_id != snapshot_id
+                or report.validator_trace_id != actor_context.get("trace_id")
+            ):
+                raise OrchestrationError(
+                    "Validation identity changed before submission"
+                )
             self._assert_execution_owner(current, report, actor_context)
-            if report.status != ValidationRunStatus.RUNNING:
+            if (
+                report.status != ValidationRunStatus.RUNNING
+                or current_task.status != TaskStatus.VALIDATING
+            ):
                 raise TaskConflict("Validation was already submitted")
+            if report.read_session is None:
+                raise TaskConflict("Validation evidence was not initialized")
+            require_complete(report.read_session)
+            require_authorized_refs(
+                report.evidence_snapshot,
+                (
+                    *evidence_refs,
+                    *(
+                        ref
+                        for result in criterion_results
+                        for ref in result.evidence_refs
+                    ),
+                ),
+            )
+            self._validate_report(current_task, verdict, criterion_results)
             report.status = ValidationRunStatus.COMPLETED
             report.verdict = verdict
             report.criterion_results = list(criterion_results)
@@ -2354,6 +2798,197 @@ class TaskCoordinator:
         )
         return result
 
+    async def grant_validation_evidence(
+        self,
+        root_trace_id: str,
+        validation_id: str,
+        grants: Sequence[ValidationEvidenceGrant],
+        *,
+        require_read: bool = False,
+    ) -> Dict[str, Any]:
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            validation = ledger.validations[validation_id]
+            if (
+                validation.read_session is not None
+                and validation.read_session.active_reads
+            ):
+                raise TaskConflict(
+                    "Validation evidence cannot change during an active page read"
+                )
+            validation.evidence_snapshot = extend_snapshot(
+                validation.evidence_snapshot, grants
+            )
+            if validation.read_session is not None:
+                required = validation.read_session.required_handles
+                if require_read:
+                    required = tuple(
+                        dict.fromkeys((*required, *(item.handle for item in grants)))
+                    )
+                validation.read_session = replace(
+                    validation.read_session,
+                    evidence_revision=validation.evidence_snapshot.revision,
+                    required_handles=required,
+                    authorized_handles=tuple(
+                        item.handle for item in validation.evidence_snapshot.grants
+                    ),
+                )
+            return {
+                "validation_id": validation_id,
+                "evidence_revision": validation.evidence_snapshot.revision,
+            }
+
+        return await self._mutate(
+            root_trace_id, "validation_evidence_granted", mutate
+        )
+
+    async def initialize_validation_evidence(
+        self,
+        root_trace_id: str,
+        validation_id: str,
+        *,
+        grants: Sequence[ValidationEvidenceGrant],
+        required_handles: Sequence[str],
+    ) -> Dict[str, Any]:
+        """Freeze initial grants and read requirements in one Ledger mutation."""
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            validation = ledger.validations[validation_id]
+            if validation.read_session is not None:
+                raise TaskConflict("Validation evidence was already initialized")
+            validation.evidence_snapshot = extend_snapshot(
+                validation.evidence_snapshot, grants
+            )
+            validation.read_session = start_read_session(
+                validation.evidence_snapshot, required_handles
+            )
+            return {
+                "validation_id": validation_id,
+                "evidence_revision": validation.evidence_snapshot.revision,
+                "required_handles": list(validation.read_session.required_handles),
+            }
+
+        return await self._mutate(
+            root_trace_id, "validation_evidence_initialized", mutate
+        )
+
+    async def apply_packed_role_context(
+        self,
+        role: AgentRole,
+        context: Mapping[str, Any],
+        required_handles: Sequence[str],
+    ) -> None:
+        """Persist final-packer read requirements before a Validator call."""
+
+        if role is not AgentRole.VALIDATOR:
+            return
+        root_trace_id = str(context.get("root_trace_id") or "")
+        validation_id = str(context.get("validation_id") or "")
+        if not root_trace_id or not validation_id:
+            raise TaskConflict("packed Validator context has no Validation identity")
+
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            validation = ledger.validations[validation_id]
+            session = validation.read_session
+            if session is None:
+                raise TaskConflict("Validation has no evidence read session")
+            if set(required_handles) - set(session.authorized_handles):
+                raise TaskConflict("packed context requires unauthorized evidence")
+            validation.read_session = replace(
+                session,
+                required_handles=tuple(
+                    dict.fromkeys((*session.required_handles, *required_handles))
+                ),
+            )
+            return {
+                "validation_id": validation_id,
+                "required_handles": list(validation.read_session.required_handles),
+            }
+
+        await self._mutate(root_trace_id, "validation_prompt_packed", mutate)
+
+    async def reserve_validation_read(
+        self,
+        root_trace_id: str,
+        validation_id: str,
+        *,
+        evidence_revision: int,
+        handle: str,
+        cursor: str | None,
+    ) -> Dict[str, Any]:
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            validation = ledger.validations[validation_id]
+            if validation.read_session is None:
+                raise TaskConflict("Validation has no evidence read session")
+            session, replay_result = reserve_read(
+                validation.read_session,
+                evidence_revision=evidence_revision,
+                handle=handle,
+                cursor=cursor,
+            )
+            validation.read_session = session
+            return {
+                "validation_id": validation_id,
+                "replayed": replay_result is not None,
+                "result": dict(replay_result) if replay_result is not None else None,
+            }
+
+        return await self._mutate(root_trace_id, "validation_read_reserved", mutate)
+
+    async def complete_validation_read(
+        self,
+        root_trace_id: str,
+        validation_id: str,
+        *,
+        evidence_revision: int,
+        handle: str,
+        cursor: str | None,
+        next_cursor: str | None,
+        exhausted: bool,
+        result: Mapping[str, Any],
+    ) -> Dict[str, Any]:
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            validation = ledger.validations[validation_id]
+            if validation.read_session is None:
+                raise TaskConflict("Validation has no evidence read session")
+            validation.read_session = complete_read(
+                validation.read_session,
+                evidence_revision=evidence_revision,
+                handle=handle,
+                cursor=cursor,
+                next_cursor=next_cursor,
+                exhausted=exhausted,
+                result=result,
+            )
+            return {
+                "validation_id": validation_id,
+                "completed": handle in validation.read_session.completed_handles,
+            }
+
+        return await self._mutate(root_trace_id, "validation_read_completed", mutate)
+
+    async def cancel_validation_read(
+        self,
+        root_trace_id: str,
+        validation_id: str,
+        *,
+        evidence_revision: int,
+        handle: str,
+        cursor: str | None,
+    ) -> Dict[str, Any]:
+        def mutate(ledger: TaskLedger) -> Dict[str, Any]:
+            validation = ledger.validations[validation_id]
+            if validation.read_session is None:
+                raise TaskConflict("Validation has no evidence read session")
+            validation.read_session = cancel_read(
+                validation.read_session,
+                evidence_revision=evidence_revision,
+                handle=handle,
+                cursor=cursor,
+            )
+            return {"validation_id": validation_id, "cancelled": True}
+
+        return await self._mutate(root_trace_id, "validation_read_cancelled", mutate)
+
     async def query_evidence(
         self,
         actor_context: Dict[str, Any],
@@ -2596,7 +3231,8 @@ class TaskCoordinator:
         response: EvidenceResponse,
     ) -> None:
         def complete(ledger: TaskLedger) -> Dict[str, Any]:
-            record = ledger.validations[validation_id].evidence_query_results[query_key]
+            validation = ledger.validations[validation_id]
+            record = validation.evidence_query_results[query_key]
             record.update(
                 {
                     "status": "completed",
@@ -2606,7 +3242,39 @@ class TaskCoordinator:
                     "queries_remaining": response.queries_remaining,
                 }
             )
-            return {"validation_id": validation_id, "status": "completed"}
+            grants = tuple(
+                ValidationEvidenceGrant(
+                    artifact_ref=ref,
+                    handle=evidence_handle(ref),
+                    source="validation_evidence_query",
+                )
+                for ref in response.evidence_refs
+            )
+            validation.evidence_snapshot = extend_snapshot(
+                validation.evidence_snapshot, grants
+            )
+            if validation.read_session is None:
+                raise TaskConflict("Validation has no evidence read session")
+            validation.read_session = replace(
+                validation.read_session,
+                evidence_revision=validation.evidence_snapshot.revision,
+                required_handles=tuple(
+                    dict.fromkeys(
+                        (
+                            *validation.read_session.required_handles,
+                            *(item.handle for item in grants),
+                        )
+                    )
+                ),
+                authorized_handles=tuple(
+                    item.handle for item in validation.evidence_snapshot.grants
+                ),
+            )
+            return {
+                "validation_id": validation_id,
+                "status": "completed",
+                "evidence_revision": validation.evidence_snapshot.revision,
+            }
 
         await self._mutate(root_trace_id, "evidence_query_completed", complete)
 
@@ -2824,6 +3492,10 @@ class TaskCoordinator:
                 )
             if task.status != TaskStatus.NEEDS_REPLAN:
                 raise TaskConflict("Task must be in needs_replan before revalidation")
+            if attempt.spec_version != task.current_spec_version:
+                raise TaskConflict(
+                    "Revalidation cannot use an Attempt from a stale Task spec"
+                )
             report = ValidationReport(
                 validation_id=new_id(),
                 task_id=task_id,

agent/agent/orchestration/executor.py

@@ -6,7 +6,7 @@ import json
 import logging
 from collections.abc import Mapping
 from copy import deepcopy
-from dataclasses import replace
+from dataclasses import dataclass, replace
 from math import isfinite
 from types import MappingProxyType
 from typing import Any, Dict, Iterable, Optional, Tuple
@@ -17,6 +17,7 @@ from agent.core.runner import RunConfig
 from agent.core.knowledge_config import KnowledgeConfig
 
 from .models import AgentRole, CompletionPolicy, ExecutionStats, FailureCode
+from .prompt_budget import PromptBudgetExceeded, RolePromptEnvelopePacker
 from .protocols import ValidatorRunResult, WorkerRunResult
 from .run_config import (
     RoleRunConfigOverrides,
@@ -29,6 +30,15 @@ from .run_config import (
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True, slots=True)
+class ModelRequest:
+    messages: tuple[dict[str, str], ...]
+    config: RunConfig
+    required_handles: tuple[str, ...]
+    packing_mode: str
+    estimated_tokens: int
+
+
 class LocalAgentExecutor:
     """Run workers and validators with the same Runner in isolated traces."""
 
@@ -37,10 +47,13 @@ class LocalAgentExecutor:
         runner: Any,
         role_run_config_resolver: Optional[RoleRunConfigResolver] = None,
         role_system_prompt_resolver: Optional[RoleSystemPromptResolver] = None,
+        prompt_pack_observer: Any | None = None,
     ) -> None:
         self.runner = runner
         self.role_run_config_resolver = role_run_config_resolver
         self.role_system_prompt_resolver = role_system_prompt_resolver
+        self.prompt_pack_observer = prompt_pack_observer
+        self.prompt_packer = RolePromptEnvelopePacker()
 
     async def stop(self, trace_id: str) -> bool:
         """Request cooperative cancellation of a local sub-trace."""
@@ -198,14 +211,35 @@ class LocalAgentExecutor:
                 preset=preset,
                 context=context,
             )
-            result = await self.runner.run_result(
-                messages=[
-                    {
-                        "role": "user",
-                        "content": json.dumps(prompt, ensure_ascii=False, indent=2),
-                    }
-                ],
+            history: tuple[Mapping[str, Any], ...] = ()
+            if continue_trace_id:
+                read_history = getattr(
+                    self.runner.trace_store, "get_main_path_messages", None
+                )
+                if callable(read_history):
+                    stored = await read_history(continue_trace_id)
+                    history = tuple(item.to_llm_dict() for item in stored)
+            packed_system = (
+                prompt_override.content
+                if prompt_override is not None
+                else await self._run_system_prompt(config)
+            )
+            if any(item.get("role") == "system" for item in history):
+                packed_system = ""
+            request = self.build_model_request(
+                prompt=prompt,
                 config=config,
+                system_prompt=packed_system,
+                history=history,
+                context=context,
+            )
+            if self.prompt_pack_observer is not None:
+                await self.prompt_pack_observer(
+                    role, context, request.required_handles
+                )
+            result = await self.runner.run_result(
+                messages=list(request.messages),
+                config=request.config,
             )
             result_trace_id = result.get("trace_id") or trace_id
             trace = await self._get_trace(result_trace_id)
@@ -228,7 +262,11 @@ class LocalAgentExecutor:
         except Exception as exc:
             trace = await self._get_trace(trace_id)
             failure = FailureDetail(
-                code="EXECUTOR_ERROR",
+                code=(
+                    PromptBudgetExceeded.code
+                    if isinstance(exc, PromptBudgetExceeded)
+                    else "EXECUTOR_ERROR"
+                ),
                 message=str(exc),
                 disposition=FailureDisposition.ABORT_RUN,
             )
@@ -247,6 +285,71 @@ class LocalAgentExecutor:
                 ),
             }
 
+    def build_model_request(
+        self,
+        *,
+        prompt: Mapping[str, Any],
+        config: RunConfig,
+        system_prompt: str,
+        history: tuple[Mapping[str, Any], ...],
+        context: Dict[str, Any],
+    ) -> ModelRequest:
+        """Return the final immutable request; callers must not mutate it."""
+
+        packed = self.prompt_packer.pack(
+            prompt,
+            system_prompt=system_prompt,
+            tool_schemas=self._run_tool_schemas(config),
+            history=history,
+        )
+        role_context = packed.payload.get("role_context")
+        if isinstance(role_context, dict):
+            context["role_context"] = role_context
+            refresh = dict(config.protected_context_refresh or {})
+            refresh["role_context"] = role_context
+            config = replace(config, protected_context_refresh=refresh)
+        context["prompt_packing"] = {
+            "mode": packed.packing_mode,
+            "estimated_tokens": packed.estimated_tokens,
+            "token_limit": self.prompt_packer.token_limit,
+            "output_reserve": self.prompt_packer.output_reserve,
+            "deduplicated_segments": packed.deduplicated_segments,
+        }
+        config = replace(
+            config,
+            context={
+                **dict(config.context or {}),
+                "prompt_packing": dict(context["prompt_packing"]),
+            },
+        )
+        message = {
+            "role": "user",
+            "content": json.dumps(
+                packed.payload,
+                ensure_ascii=False,
+                indent=2,
+            ),
+        }
+        return ModelRequest(
+            (message,),
+            config,
+            packed.required_handles,
+            packed.packing_mode,
+            packed.estimated_tokens,
+        )
+
+    async def _run_system_prompt(self, config: RunConfig) -> str:
+        getter = getattr(self.runner, "get_run_system_prompt", None)
+        if callable(getter):
+            return str(await getter(config))
+        return str(config.system_prompt or "")
+
+    def _run_tool_schemas(self, config: RunConfig) -> tuple[Mapping[str, Any], ...]:
+        getter = getattr(self.runner, "get_run_tool_schemas", None)
+        if not callable(getter):
+            return ()
+        return tuple(getter(config))
+
     async def _apply_run_config_overrides(
         self,
         config: RunConfig,

agent/agent/orchestration/models.py

@@ -445,6 +445,90 @@ class ArtifactSnapshot:
         )
 
 
+@dataclass(frozen=True)
+class ValidationEvidenceGrant:
+    artifact_ref: ArtifactRef
+    handle: str
+    source: str = "attempt_snapshot"
+    granted_at: str = field(default_factory=utc_now)
+
+    def __post_init__(self) -> None:
+        if not self.handle.strip():
+            raise ValueError("ValidationEvidenceGrant.handle is required")
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ValidationEvidenceGrant":
+        return cls(
+            ArtifactRef.from_dict(dict(data.get("artifact_ref", {}))),
+            str(data.get("handle", "")),
+            str(data.get("source", "attempt_snapshot")),
+            str(data.get("granted_at", utc_now())),
+        )
+
+
+@dataclass(frozen=True)
+class ValidationEvidenceSnapshot:
+    revision: int = 0
+    grants: Tuple[ValidationEvidenceGrant, ...] = ()
+    updated_at: str = field(default_factory=utc_now)
+
+    def __post_init__(self) -> None:
+        if self.revision < 0:
+            raise ValueError("Validation evidence revision cannot be negative")
+        object.__setattr__(self, "grants", tuple(self.grants))
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ValidationEvidenceSnapshot":
+        return cls(
+            int(data.get("revision", 0)),
+            tuple(
+                ValidationEvidenceGrant.from_dict(item)
+                for item in data.get("grants", [])
+            ),
+            str(data.get("updated_at", utc_now())),
+        )
+
+
+@dataclass
+class ValidationReadSession:
+    evidence_revision: int
+    required_handles: Tuple[str, ...] = ()
+    authorized_handles: Tuple[str, ...] = ()
+    completed_handles: List[str] = field(default_factory=list)
+    next_cursors: Dict[str, Optional[str]] = field(default_factory=dict)
+    active_reads: Dict[str, Optional[str]] = field(default_factory=dict)
+    read_results: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        self.required_handles = tuple(dict.fromkeys(self.required_handles))
+        self.authorized_handles = tuple(dict.fromkeys(self.authorized_handles))
+        self.completed_handles = list(dict.fromkeys(self.completed_handles))
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ValidationReadSession":
+        required = tuple(str(item) for item in data.get("required_handles", []))
+        return cls(
+            evidence_revision=int(data.get("evidence_revision", 0)),
+            required_handles=required,
+            authorized_handles=tuple(
+                str(item) for item in data.get("authorized_handles", required)
+            ),
+            completed_handles=[str(item) for item in data.get("completed_handles", [])],
+            next_cursors={
+                str(key): (str(value) if value is not None else None)
+                for key, value in data.get("next_cursors", {}).items()
+            },
+            active_reads={
+                str(key): (str(value) if value is not None else None)
+                for key, value in data.get("active_reads", {}).items()
+            },
+            read_results={
+                str(key): dict(value)
+                for key, value in data.get("read_results", {}).items()
+            },
+        )
+
+
 @dataclass
 class TaskAttempt:
     attempt_id: str
@@ -549,6 +633,10 @@ class ValidationReport:
     validation_plan: ValidationPlan = field(default_factory=ValidationPlan)
     evidence_queries_used: int = 0
     evidence_query_results: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    evidence_snapshot: ValidationEvidenceSnapshot = field(
+        default_factory=ValidationEvidenceSnapshot
+    )
+    read_session: Optional[ValidationReadSession] = None
     status: ValidationRunStatus = ValidationRunStatus.PENDING
     operation_id: Optional[str] = None
     execution_epoch: int = 0
@@ -587,6 +675,14 @@ class ValidationReport:
             evidence_query_results={
                 str(key): dict(value) for key, value in data.get("evidence_query_results", {}).items()
             },
+            evidence_snapshot=ValidationEvidenceSnapshot.from_dict(
+                dict(data.get("evidence_snapshot", {}))
+            ),
+            read_session=(
+                ValidationReadSession.from_dict(data["read_session"])
+                if data.get("read_session") is not None
+                else None
+            ),
             status=ValidationRunStatus(data.get("status", ValidationRunStatus.PENDING.value)),
             operation_id=data.get("operation_id"),
             execution_epoch=int(data.get("execution_epoch", 0)),
@@ -642,6 +738,66 @@ class PlannerDecision:
         )
 
 
+@dataclass(frozen=True)
+class ResourceClaim:
+    uri: str
+    exclusive: bool = True
+
+    def __post_init__(self) -> None:
+        if not self.uri.strip():
+            raise ValueError("ResourceClaim.uri is required")
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ResourceClaim":
+        return cls(str(data.get("uri", "")), bool(data.get("exclusive", True)))
+
+
+@dataclass(frozen=True)
+class FrozenContextDocument:
+    uri: str
+    digest: str
+    payload: Dict[str, Any]
+
+    def __post_init__(self) -> None:
+        if not self.uri.strip() or not self.digest.strip():
+            raise ValueError("FrozenContextDocument identity is required")
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "FrozenContextDocument":
+        return cls(
+            str(data.get("uri", "")),
+            str(data.get("digest", "")),
+            dict(data.get("payload", {})),
+        )
+
+
+@dataclass
+class DurableLeaseRecord:
+    lease_key: str
+    owner: str
+    epoch: int
+    status: str
+    expires_at: str
+    journal: List[Dict[str, Any]] = field(default_factory=list)
+    terminal_result: Optional[Dict[str, Any]] = None
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "DurableLeaseRecord":
+        return cls(
+            lease_key=str(data["lease_key"]),
+            owner=str(data["owner"]),
+            epoch=int(data["epoch"]),
+            status=str(data["status"]),
+            expires_at=str(data["expires_at"]),
+            journal=[dict(item) for item in data.get("journal", [])],
+            terminal_result=(
+                dict(data["terminal_result"])
+                if data.get("terminal_result") is not None
+                else None
+            ),
+        )
+
+
 @dataclass
 class BackgroundOperation:
     operation_id: str
@@ -653,6 +809,7 @@ class BackgroundOperation:
     task_ids: List[str] = field(default_factory=list)
     attempt_ids: List[str] = field(default_factory=list)
     validation_ids: List[str] = field(default_factory=list)
+    resource_claims: List[ResourceClaim] = field(default_factory=list)
     result_ref: Optional[Dict[str, Any]] = None
     execution_epoch: int = 0
     deadline_at: Optional[str] = None
@@ -674,6 +831,9 @@ class BackgroundOperation:
             task_ids=list(data.get("task_ids", [])),
             attempt_ids=list(data.get("attempt_ids", [])),
             validation_ids=list(data.get("validation_ids", [])),
+            resource_claims=[
+                ResourceClaim.from_dict(item) for item in data.get("resource_claims", [])
+            ],
             result_ref=dict(data["result_ref"]) if data.get("result_ref") is not None else None,
             execution_epoch=int(data.get("execution_epoch", 0)),
             deadline_at=data.get("deadline_at"),
@@ -764,6 +924,8 @@ class TaskLedger:
     decisions: Dict[str, PlannerDecision] = field(default_factory=dict)
     operations: Dict[str, BackgroundOperation] = field(default_factory=dict)
     command_records: Dict[str, CommandRecord] = field(default_factory=dict)
+    context_documents: Dict[str, FrozenContextDocument] = field(default_factory=dict)
+    durable_leases: Dict[str, DurableLeaseRecord] = field(default_factory=dict)
     focused_task_id: Optional[str] = None
     idempotency_results: Dict[str, Dict[str, Any]] = field(default_factory=dict)
     created_at: str = field(default_factory=utc_now)
@@ -797,6 +959,14 @@ class TaskLedger:
             decisions={k: PlannerDecision.from_dict(v) for k, v in data.get("decisions", {}).items()},
             operations={k: BackgroundOperation.from_dict(v) for k, v in data.get("operations", {}).items()},
             command_records={k: CommandRecord.from_dict(v) for k, v in data.get("command_records", {}).items()},
+            context_documents={
+                key: FrozenContextDocument.from_dict(value)
+                for key, value in data.get("context_documents", {}).items()
+            },
+            durable_leases={
+                str(key): DurableLeaseRecord.from_dict(value)
+                for key, value in data.get("durable_leases", {}).items()
+            },
             focused_task_id=data.get("focused_task_id"),
             idempotency_results=dict(data.get("idempotency_results", {})),
             created_at=data.get("created_at", utc_now()),
@@ -881,9 +1051,14 @@ __all__ = [
     "ArtifactRef",
     "ArtifactSnapshot",
     "AttemptSubmission",
+    "ValidationEvidenceGrant",
+    "ValidationEvidenceSnapshot",
+    "ValidationReadSession",
     "CriterionResult",
     "ValidationReport",
     "PlannerDecision",
+    "ResourceClaim",
+    "FrozenContextDocument",
     "BackgroundOperation",
     "CommandRecord",
     "EventDraft",

agent/agent/orchestration/operations.py

@@ -22,6 +22,7 @@ from .models import (
     FailureCode,
     OperationKind,
     OperationStatus,
+    ResourceClaim,
     TaskCycleResult,
     TaskLedger,
     TaskStatus,
@@ -86,6 +87,7 @@ class OperationController:
         attempt_id: Optional[str] = None,
         idempotency_key: Optional[str] = None,
         deadline_at: Optional[str] = None,
+        resource_claims: Optional[Sequence[ResourceClaim]] = None,
     ) -> BackgroundOperation:
         if not self._get_executor():
             raise RuntimeError("TaskCoordinator has no AgentExecutor")
@@ -97,6 +99,7 @@ class OperationController:
             attempt_id,
             deadline_at,
             self._get_config().default_worker_preset,
+            resource_claims,
         )
         operation_id = new_id()
         fingerprint = self._fingerprint("start_operation", request)
@@ -356,6 +359,11 @@ class OperationController:
                     unsafe.append(validation_id)
             if unsafe:
                 raise TaskConflict(f"resume_not_safe: stages already started: {unsafe}")
+            _require_claims_available(
+                ledger,
+                operation.resource_claims,
+                excluding_operation_id=operation.operation_id,
+            )
             operation.status = OperationStatus.PENDING
             operation.completed_at = None
             operation.error = None
@@ -668,6 +676,7 @@ def _normalize_start_request(
     attempt_id: Optional[str],
     deadline_at: Optional[str],
     default_worker_preset: str,
+    resource_claims: Optional[Sequence[ResourceClaim]],
 ) -> tuple[OperationKind, List[str], List[str], Dict[str, Any]]:
     operation_kind = OperationKind(kind)
     requested_task_ids = list(task_ids or ([] if task_id is None else [task_id]))
@@ -680,12 +689,20 @@ def _normalize_start_request(
         default_worker_preset,
     )
     normalized_deadline = parse_deadline(deadline_at)
+    claims = list(resource_claims or ())
+    if any(not isinstance(item, ResourceClaim) for item in claims):
+        raise ValueError("resource_claims must contain ResourceClaim values")
+    if len({item.uri for item in claims}) != len(claims):
+        raise TaskConflict("resource_claims must contain unique URIs")
     request = {
         "kind": operation_kind.value,
         "task_ids": requested_task_ids,
         "worker_presets": presets,
         "attempt_id": attempt_id,
         "deadline_at": normalized_deadline,
+        "resource_claims": [
+            {"uri": item.uri, "exclusive": item.exclusive} for item in claims
+        ],
     }
     return operation_kind, requested_task_ids, presets, request
 
@@ -735,6 +752,10 @@ def _create_operation(
     for task_id in task_ids:
         if task_id not in ledger.tasks:
             raise ValueError(f"Task not found: {task_id}")
+    claims = [
+        ResourceClaim.from_dict(item) for item in request.get("resource_claims", [])
+    ]
+    _require_claims_available(ledger, claims)
     ledger.operations[operation_id] = BackgroundOperation(
         operation_id=operation_id,
         root_trace_id=root_trace_id,
@@ -743,10 +764,33 @@ def _create_operation(
         request_fingerprint=fingerprint,
         task_ids=list(task_ids),
         deadline_at=request["deadline_at"],
+        resource_claims=claims,
     )
     return {"operation_id": operation_id}
 
 
+def _require_claims_available(
+    ledger: TaskLedger,
+    claims: Sequence[ResourceClaim],
+    *,
+    excluding_operation_id: str | None = None,
+) -> None:
+    active = {
+        OperationStatus.PENDING,
+        OperationStatus.RUNNING,
+        OperationStatus.STOP_REQUESTED,
+    }
+    for operation in ledger.operations.values():
+        if operation.operation_id == excluding_operation_id or operation.status not in active:
+            continue
+        for requested in claims:
+            for held in operation.resource_claims:
+                if requested.uri == held.uri and (requested.exclusive or held.exclusive):
+                    raise TaskConflict(
+                        f"resource_claim_conflict:{requested.uri}:{operation.operation_id}"
+                    )
+
+
 def parse_deadline(value: Optional[str]) -> Optional[str]:
     if value is None:
         return None

agent/agent/orchestration/prompt_budget.py

@@ -0,0 +1,188 @@
+"""Business-neutral whole-envelope prompt packing."""
+
+from __future__ import annotations
+
+import json
+import math
+from collections.abc import Mapping, Sequence
+from copy import deepcopy
+from dataclasses import dataclass
+from typing import Any
+
+
+class PromptBudgetExceeded(RuntimeError):
+    code = "CONTEXT_BUDGET_EXCEEDED"
+
+
+@dataclass(frozen=True, slots=True)
+class PackedPrompt:
+    payload: dict[str, Any]
+    estimated_tokens: int
+    packing_mode: str
+    required_handles: tuple[str, ...] = ()
+    deduplicated_segments: int = 0
+
+
+@dataclass(frozen=True, slots=True)
+class PackedModelEnvelope:
+    messages: tuple[dict[str, Any], ...]
+    estimated_tokens: int
+    packing_mode: str
+    deduplicated_segments: int = 0
+
+
+@dataclass(frozen=True, slots=True)
+class RolePromptEnvelopePacker:
+    token_limit: int = 32_000
+    output_reserve: int = 8_192
+
+    def pack(
+        self,
+        payload: Mapping[str, Any],
+        *,
+        system_prompt: str,
+        tool_schemas: Sequence[Mapping[str, Any]],
+        history: Sequence[Mapping[str, Any]] = (),
+    ) -> PackedPrompt:
+        packed = deepcopy(dict(payload))
+        available = self.token_limit - self.output_reserve
+        if available <= 0:
+            raise PromptBudgetExceeded("output reserve exhausts the context budget")
+
+        def measure() -> int:
+            return estimate_prompt_tokens(
+                {
+                    "system": system_prompt,
+                    "history": history,
+                    "tools": tool_schemas,
+                    "user": packed,
+                }
+            )
+
+        mode = "full"
+        required_handles: tuple[str, ...] = ()
+        deduplicated_segments = 0
+        context = packed.get("role_context")
+        if isinstance(context, dict):
+            artifact = context.get("artifact")
+            bundle = context.get("context_bundle")
+            snapshot = packed.get("artifact_snapshot")
+            if (
+                isinstance(artifact, dict)
+                and artifact.get("items")
+                and isinstance(snapshot, dict)
+                and snapshot.get("normalized_content") is not None
+            ):
+                snapshot.pop("normalized_content", None)
+                deduplicated_segments += 1
+            if isinstance(artifact, dict) and isinstance(bundle, dict):
+                fallback_cards = deepcopy(bundle.get("cards", []))
+                if artifact.get("items") and fallback_cards:
+                    bundle["cards"] = []
+                    deduplicated_segments += len(fallback_cards)
+                if measure() > available and artifact.get("items"):
+                    artifact["items"] = []
+                    artifact["full_content_in_bootstrap"] = False
+                    bundle["cards"] = fallback_cards
+                    required_handles = _evidence_handles(context)
+                    mode = "cards"
+                if measure() > available and bundle.get("cards"):
+                    bundle["cards"] = []
+                    mode = "handles"
+        used = measure()
+        if used > available:
+            raise PromptBudgetExceeded(
+                f"minimum prompt uses about {used} tokens; available={available}"
+            )
+        return PackedPrompt(
+            packed,
+            used,
+            mode,
+            required_handles,
+            deduplicated_segments,
+        )
+
+    def pack_messages(
+        self,
+        messages: Sequence[Mapping[str, Any]],
+        *,
+        tool_schemas: Sequence[Mapping[str, Any]],
+    ) -> PackedModelEnvelope:
+        """Pack the final request for every role, including side branches."""
+
+        packed = [deepcopy(dict(item)) for item in messages]
+        mode = "full"
+        deduplicated = 0
+        for index in range(len(packed) - 1, -1, -1):
+            message = packed[index]
+            if message.get("role") != "user" or not isinstance(
+                message.get("content"), str
+            ):
+                continue
+            try:
+                payload = json.loads(message["content"])
+            except (TypeError, ValueError):
+                continue
+            if not isinstance(payload, Mapping):
+                continue
+            structured = self.pack(
+                payload,
+                system_prompt="",
+                tool_schemas=tool_schemas,
+                history=packed[:index],
+            )
+            message["content"] = json.dumps(
+                structured.payload,
+                ensure_ascii=False,
+                indent=2,
+            )
+            mode = structured.packing_mode
+            deduplicated = structured.deduplicated_segments
+            break
+        used = estimate_prompt_tokens(
+            {"messages": packed, "tools": tool_schemas}
+        )
+        available = self.token_limit - self.output_reserve
+        if used > available:
+            raise PromptBudgetExceeded(
+                f"minimum prompt uses about {used} tokens; available={available}"
+            )
+        return PackedModelEnvelope(
+            tuple(packed),
+            used,
+            mode,
+            deduplicated,
+        )
+
+
+def estimate_prompt_tokens(value: Any, *, safety_factor: float = 1.25) -> int:
+    text = json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    ascii_count = sum(ord(char) < 128 for char in text)
+    base = math.ceil(ascii_count / 3) + len(text) - ascii_count
+    return math.ceil(base * safety_factor)
+
+
+def _evidence_handles(context: Mapping[str, Any]) -> tuple[str, ...]:
+    values = context.get("evidence_handles", ())
+    if not isinstance(values, (list, tuple)):
+        return ()
+    return tuple(
+        str(item["evidence_handle"])
+        for item in values
+        if isinstance(item, Mapping) and item.get("evidence_handle")
+    )
+
+
+__all__ = [
+    "PackedPrompt",
+    "PackedModelEnvelope",
+    "PromptBudgetExceeded",
+    "RolePromptEnvelopePacker",
+    "estimate_prompt_tokens",
+]

agent/agent/orchestration/validation_evidence.py

@@ -0,0 +1,220 @@
+"""Pure rules for one Validation's evidence authorization and read progress."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from dataclasses import replace
+from hashlib import sha256
+from typing import Any
+
+from .errors import TaskConflict
+from .models import (
+    ArtifactRef,
+    ValidationEvidenceGrant,
+    ValidationEvidenceSnapshot,
+    ValidationReadSession,
+    utc_now,
+)
+
+
+class ValidationEvidenceConflict(TaskConflict):
+    code = "VALIDATION_EVIDENCE_CONFLICT"
+
+
+class ValidationEvidenceUnauthorized(TaskConflict):
+    code = "VALIDATION_EVIDENCE_UNAUTHORIZED"
+
+
+def evidence_handle(ref: ArtifactRef) -> str:
+    immutable = ref.digest or ref.version or ""
+    encoded = f"{ref.uri}\0{immutable}".encode()
+    return "src_" + sha256(encoded).hexdigest()[:20]
+
+
+def extend_snapshot(
+    snapshot: ValidationEvidenceSnapshot,
+    grants: Iterable[ValidationEvidenceGrant],
+) -> ValidationEvidenceSnapshot:
+    by_uri = {item.artifact_ref.uri: item for item in snapshot.grants}
+    by_handle = {item.handle: item for item in snapshot.grants}
+    changed = False
+    for grant in grants:
+        prior = by_uri.get(grant.artifact_ref.uri)
+        if prior is not None:
+            if _immutable_identity(prior.artifact_ref) != _immutable_identity(
+                grant.artifact_ref
+            ):
+                raise ValidationEvidenceConflict(
+                    "the same evidence URI was granted with different immutable content"
+                )
+            continue
+        prior_handle = by_handle.get(grant.handle)
+        if prior_handle is not None and prior_handle.artifact_ref != grant.artifact_ref:
+            raise ValidationEvidenceConflict(
+                "the same evidence handle identifies different immutable content"
+            )
+        by_uri[grant.artifact_ref.uri] = grant
+        by_handle[grant.handle] = grant
+        changed = True
+    if not changed:
+        return snapshot
+    return ValidationEvidenceSnapshot(
+        snapshot.revision + 1,
+        tuple(sorted(by_uri.values(), key=lambda item: item.handle)),
+        utc_now(),
+    )
+
+
+def start_read_session(
+    snapshot: ValidationEvidenceSnapshot,
+    required_handles: Iterable[str],
+) -> ValidationReadSession:
+    authorized = {item.handle for item in snapshot.grants}
+    required = tuple(dict.fromkeys(required_handles))
+    if not set(required) <= authorized:
+        raise ValidationEvidenceUnauthorized("read session contains an unauthorized handle")
+    return ValidationReadSession(
+        evidence_revision=snapshot.revision,
+        required_handles=required,
+        authorized_handles=tuple(item.handle for item in snapshot.grants),
+    )
+
+
+def reserve_read(
+    session: ValidationReadSession,
+    *,
+    evidence_revision: int,
+    handle: str,
+    cursor: str | None,
+) -> tuple[ValidationReadSession, Mapping[str, Any] | None]:
+    _authorize(session, evidence_revision, handle)
+    page_key = _page_key(handle, cursor)
+    if page_key in session.read_results:
+        return session, session.read_results[page_key]
+    if handle in session.completed_handles:
+        raise ValidationEvidenceUnauthorized("evidence handle was already exhausted")
+    expected = session.next_cursors.get(handle)
+    if expected != cursor:
+        raise ValidationEvidenceUnauthorized("evidence cursor is stale or out of sequence")
+    if handle in session.active_reads:
+        raise ValidationEvidenceUnauthorized("evidence handle already has an active read")
+    active = dict(session.active_reads)
+    active[handle] = cursor
+    return replace(session, active_reads=active), None
+
+
+def complete_read(
+    session: ValidationReadSession,
+    *,
+    evidence_revision: int,
+    handle: str,
+    cursor: str | None,
+    next_cursor: str | None,
+    exhausted: bool,
+    result: Mapping[str, Any],
+) -> ValidationReadSession:
+    _authorize(session, evidence_revision, handle)
+    if session.active_reads.get(handle) != cursor:
+        raise ValidationEvidenceUnauthorized("read completion has no matching reservation")
+    active = dict(session.active_reads)
+    active.pop(handle)
+    next_cursors = dict(session.next_cursors)
+    completed = list(session.completed_handles)
+    read_results = dict(session.read_results)
+    read_results[_page_key(handle, cursor)] = dict(result)
+    if exhausted:
+        next_cursors.pop(handle, None)
+        if handle not in completed:
+            completed.append(handle)
+    else:
+        if not next_cursor:
+            raise ValidationEvidenceUnauthorized("non-terminal evidence page needs a cursor")
+        next_cursors[handle] = next_cursor
+    return replace(
+        session,
+        active_reads=active,
+        next_cursors=next_cursors,
+        completed_handles=completed,
+        read_results=read_results,
+    )
+
+
+def cancel_read(
+    session: ValidationReadSession,
+    *,
+    evidence_revision: int,
+    handle: str,
+    cursor: str | None,
+) -> ValidationReadSession:
+    """Release a failed page reservation without advancing durable progress."""
+
+    _authorize(session, evidence_revision, handle)
+    if session.active_reads.get(handle) != cursor:
+        raise ValidationEvidenceUnauthorized("read cancellation has no matching reservation")
+    active = dict(session.active_reads)
+    active.pop(handle)
+    return replace(session, active_reads=active)
+
+
+def require_complete(session: ValidationReadSession) -> None:
+    missing = sorted(set(session.required_handles) - set(session.completed_handles))
+    if missing or session.active_reads:
+        raise ValidationEvidenceUnauthorized(
+            f"validation evidence reads are incomplete: {missing}"
+        )
+
+
+def require_authorized_refs(
+    snapshot: ValidationEvidenceSnapshot,
+    refs: Iterable[ArtifactRef],
+) -> None:
+    authorized = {
+        (
+            item.artifact_ref.uri,
+            item.artifact_ref.version,
+            item.artifact_ref.digest,
+            item.artifact_ref.kind,
+        )
+        for item in snapshot.grants
+    }
+    submitted = {
+        (ref.uri, ref.version, ref.digest, ref.kind)
+        for ref in refs
+    }
+    if submitted - authorized:
+        raise ValidationEvidenceUnauthorized(
+            "validation references evidence outside its frozen snapshot"
+        )
+
+
+def _authorize(
+    session: ValidationReadSession,
+    evidence_revision: int,
+    handle: str,
+) -> None:
+    if session.evidence_revision != evidence_revision:
+        raise ValidationEvidenceUnauthorized("validation evidence revision is stale")
+    if handle not in session.authorized_handles:
+        raise ValidationEvidenceUnauthorized("evidence handle is outside this Validation")
+
+
+def _immutable_identity(ref: ArtifactRef) -> tuple[str | None, str | None, str]:
+    return ref.version, ref.digest, ref.kind
+
+
+def _page_key(handle: str, cursor: str | None) -> str:
+    return f"{handle}\0{cursor or ''}"
+
+
+__all__ = [
+    "ValidationEvidenceConflict",
+    "ValidationEvidenceUnauthorized",
+    "cancel_read",
+    "complete_read",
+    "evidence_handle",
+    "extend_snapshot",
+    "require_authorized_refs",
+    "require_complete",
+    "reserve_read",
+    "start_read_session",
+]

agent/agent/orchestration/validation_policy.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import inspect
 from dataclasses import asdict, dataclass, field
-from typing import Any, Awaitable, Dict, List, Optional, Protocol, Sequence, Union
+from typing import Any, Awaitable, Dict, List, Mapping, Optional, Protocol, Sequence, Union
 
 from .models import (
     ArtifactRef,
@@ -25,6 +25,7 @@ class ValidationContext:
     attempt: TaskAttempt
     snapshot: ArtifactSnapshot
     prior_validation_count: int = 0
+    context_documents: Mapping[str, Any] = field(default_factory=dict)
 
 
 class ValidationPolicy(Protocol):

agent/agent/orchestration/wire.py

@@ -34,10 +34,16 @@ class WireModel(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
 
+class ResourceClaimView(WireModel):
+    uri: str
+    exclusive: bool = True
+
+
 class DispatchOperationRequest(WireModel):
     kind: Literal["dispatch"] = "dispatch"
     task_ids: List[Annotated[str, Field(min_length=1)]] = Field(min_length=1)
     worker_presets: Optional[List[str]] = None
+    resource_claims: List["ResourceClaimView"] = Field(default_factory=list)
     deadline_at: Optional[str] = None
 
 
@@ -62,6 +68,7 @@ class OperationView(WireModel):
     task_ids: List[str]
     attempt_ids: List[str]
     validation_ids: List[str]
+    resource_claims: List[ResourceClaimView] = Field(default_factory=list)
     result_ref: Optional[Dict[str, Any]] = None
     execution_epoch: int
     deadline_at: Optional[str] = None
@@ -125,6 +132,29 @@ class ValidationPlanView(WireModel):
     evidence_timeout_seconds: float
 
 
+class ValidationEvidenceGrantView(WireModel):
+    artifact_ref: ArtifactRefView
+    handle: str
+    source: str
+    granted_at: str
+
+
+class ValidationEvidenceSnapshotView(WireModel):
+    revision: int
+    grants: List[ValidationEvidenceGrantView]
+    updated_at: str
+
+
+class ValidationReadSessionView(WireModel):
+    evidence_revision: int
+    required_handles: List[str]
+    authorized_handles: List[str]
+    completed_handles: List[str]
+    next_cursors: Dict[str, Optional[str]]
+    active_reads: Dict[str, Optional[str]]
+    read_results: Dict[str, Dict[str, Any]]
+
+
 class ExecutionStatsView(WireModel):
     primary_model: Optional[str] = None
     total_tokens: Optional[int] = None
@@ -210,6 +240,8 @@ class ValidationView(WireModel):
     validator_preset: str
     validation_plan: ValidationPlanView
     evidence_queries_used: int
+    evidence_snapshot: Optional[ValidationEvidenceSnapshotView] = None
+    read_session: Optional[ValidationReadSessionView] = None
     status: ValidationRunStatus
     operation_id: Optional[str] = None
     execution_epoch: int

agent/agent/orchestration/wiring.py

@@ -54,6 +54,7 @@ def wire_orchestration(
         runner,
         role_run_config_resolver=role_run_config_resolver,
         role_system_prompt_resolver=role_system_prompt_resolver,
+        prompt_pack_observer=coordinator.apply_packed_role_context,
     )
     coordinator.set_executor(executor)
     runner.task_coordinator = coordinator

agent/tests/test_architecture_boundaries.py

@@ -0,0 +1,305 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from agent.orchestration import (
+    AgentRole,
+    ArtifactRef,
+    AttemptSubmission,
+    CriterionResult,
+    FrozenContextDocument,
+    ResourceClaim,
+    TaskStatus,
+    ValidationVerdict,
+)
+from agent.orchestration.coordinator import TaskConflict
+from agent.orchestration.models import ValidationEvidenceGrant, ValidationEvidenceSnapshot
+from agent.orchestration.prompt_budget import PromptBudgetExceeded, RolePromptEnvelopePacker
+from agent.orchestration.validation_evidence import (
+    ValidationEvidenceConflict,
+    ValidationEvidenceUnauthorized,
+    complete_read,
+    extend_snapshot,
+    require_authorized_refs,
+    require_complete,
+    reserve_read,
+    start_read_session,
+)
+
+
+def _grant(handle: str, uri: str, digest: str) -> ValidationEvidenceGrant:
+    return ValidationEvidenceGrant(
+        ArtifactRef(uri, "evidence", "1", digest),
+        handle,
+    )
+
+
+def test_validation_snapshot_deduplicates_and_rejects_digest_conflict() -> None:
+    original = _grant("handle-a", "evidence://a", "sha256:" + "1" * 64)
+    snapshot = extend_snapshot(ValidationEvidenceSnapshot(), (original, original))
+    assert snapshot.revision == 1
+    assert snapshot.grants == (original,)
+    with pytest.raises(ValidationEvidenceConflict):
+        extend_snapshot(
+            snapshot,
+            (_grant("handle-b", "evidence://a", "sha256:" + "2" * 64),),
+        )
+
+
+def test_validation_snapshot_rejects_version_only_identity_conflict() -> None:
+    first = ValidationEvidenceGrant(
+        ArtifactRef("evidence://versioned", "evidence", "1"),
+        "handle-v1",
+    )
+    second = ValidationEvidenceGrant(
+        ArtifactRef("evidence://versioned", "evidence", "2"),
+        "handle-v2",
+    )
+    snapshot = extend_snapshot(ValidationEvidenceSnapshot(), (first,))
+    with pytest.raises(ValidationEvidenceConflict):
+        extend_snapshot(snapshot, (second,))
+
+
+def test_validation_submission_accepts_only_snapshot_refs() -> None:
+    grant = _grant("handle-a", "evidence://a", "sha256:" + "1" * 64)
+    snapshot = extend_snapshot(ValidationEvidenceSnapshot(), (grant,))
+    require_authorized_refs(snapshot, (grant.artifact_ref,))
+    with pytest.raises(ValidationEvidenceUnauthorized):
+        require_authorized_refs(
+            snapshot,
+            (
+                ArtifactRef(
+                    "evidence://forged",
+                    "evidence",
+                    "1",
+                    "sha256:" + "2" * 64,
+                ),
+            ),
+        )
+
+
+def test_validation_read_replay_is_core_session_semantics() -> None:
+    snapshot = extend_snapshot(
+        ValidationEvidenceSnapshot(),
+        (_grant("handle-a", "evidence://a", "sha256:" + "1" * 64),),
+    )
+    session = start_read_session(snapshot, ("handle-a",))
+    session, replay_result = reserve_read(
+        session,
+        evidence_revision=snapshot.revision,
+        handle="handle-a",
+        cursor=None,
+    )
+    assert replay_result is None
+    session = complete_read(
+        session,
+        evidence_revision=snapshot.revision,
+        handle="handle-a",
+        cursor=None,
+        next_cursor=None,
+        exhausted=True,
+        result={"content_fragment": "frozen", "exhausted": True},
+    )
+    require_complete(session)
+    same, replay_result = reserve_read(
+        session,
+        evidence_revision=snapshot.revision,
+        handle="handle-a",
+        cursor=None,
+    )
+    assert replay_result == {"content_fragment": "frozen", "exhausted": True}
+    assert same is session
+
+
+def test_prompt_packer_counts_system_tools_history_and_payload() -> None:
+    packer = RolePromptEnvelopePacker(token_limit=300, output_reserve=100)
+    with pytest.raises(PromptBudgetExceeded):
+        packer.pack(
+            {"role_context": {"artifact": {"items": []}}},
+            system_prompt="s" * 500,
+            tool_schemas=({"name": "tool", "description": "x" * 500},),
+            history=({"role": "user", "content": "h" * 500},),
+        )
+
+
+def test_prompt_packer_uses_the_same_final_envelope_for_planner_messages() -> None:
+    packer = RolePromptEnvelopePacker(token_limit=2_000, output_reserve=100)
+    packed = packer.pack_messages(
+        (
+            {"role": "system", "content": "planner policy"},
+            {
+                "role": "user",
+                "content": (
+                    '{"role_context":{"artifact":{"items":["body"]},'
+                    '"context_bundle":{"cards":[{"id":"duplicate"}]}}}'
+                ),
+            },
+        ),
+        tool_schemas=({"name": "plan_script_tasks"},),
+    )
+    assert packed.packing_mode == "full"
+    assert packed.deduplicated_segments == 1
+    assert '"cards": []' in packed.messages[-1]["content"]
+
+
+def test_resource_claim_requires_a_nonblank_canonical_identity() -> None:
+    assert ResourceClaim("script-build://resources/workspace/a").exclusive
+    with pytest.raises(ValueError):
+        ResourceClaim(" ")
+
+
+@pytest.mark.asyncio
+async def test_agent_validation_cannot_submit_before_evidence_initialization(
+    tmp_path,
+) -> None:
+    from test_coordinator_integration import (
+        FakeExecutor,
+        create_task,
+        make_coordinator,
+    )
+
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    task_id = await create_task(coordinator, "evidence must be initialized")
+    created = await coordinator._create_attempt("root", task_id, "worker", None)
+    ledger = await store.load("root")
+    attempt = ledger.attempts[created["attempt_id"]]
+    await coordinator.submit_attempt(
+        {
+            "role": AgentRole.WORKER.value,
+            "root_trace_id": "root",
+            "task_id": task_id,
+            "attempt_id": attempt.attempt_id,
+            "spec_version": attempt.spec_version,
+            "trace_id": attempt.worker_trace_id,
+            "tool_call_id": "submit-attempt",
+        },
+        AttemptSubmission("done", [ArtifactRef("memory://artifact", version="1")]),
+    )
+    ledger = await store.load("root")
+    attempt = ledger.attempts[attempt.attempt_id]
+    validation_id = await coordinator._start_validation(
+        "root", task_id, attempt.attempt_id
+    )
+    ledger = await store.load("root")
+    report = ledger.validations[validation_id]
+
+    with pytest.raises(TaskConflict, match="evidence was not initialized"):
+        await coordinator.submit_validation(
+            {
+                "role": AgentRole.VALIDATOR.value,
+                "root_trace_id": "root",
+                "task_id": task_id,
+                "attempt_id": attempt.attempt_id,
+                "validation_id": validation_id,
+                "snapshot_id": report.snapshot_id,
+                "trace_id": report.validator_trace_id,
+                "tool_call_id": "premature-validation",
+            },
+            ValidationVerdict.PASSED,
+            [CriterionResult("c1", ValidationVerdict.PASSED, "checked")],
+            "passed",
+            [],
+            [],
+            [],
+            "accept",
+        )
+
+
+@pytest.mark.asyncio
+async def test_durable_lease_renewal_preserves_epoch_and_rejects_stale_owner(
+    tmp_path,
+) -> None:
+    from test_coordinator_integration import FakeExecutor, make_coordinator
+
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    reserved = await coordinator.reserve_durable_lease(
+        "root", "retrieval:test", owner="worker-a", lease_seconds=30
+    )
+    await coordinator.renew_durable_lease(
+        "root",
+        "retrieval:test",
+        owner="worker-a",
+        epoch=reserved["epoch"],
+        lease_seconds=60,
+    )
+    ledger = await store.load("root")
+    renewed = ledger.durable_leases["retrieval:test"]
+    assert renewed.epoch == reserved["epoch"]
+    assert datetime.fromisoformat(renewed.expires_at) > datetime.now(
+        timezone.utc
+    ) + timedelta(seconds=50)
+    with pytest.raises(TaskConflict, match="ownership is stale"):
+        await coordinator.renew_durable_lease(
+            "root",
+            "retrieval:test",
+            owner="worker-b",
+            epoch=reserved["epoch"],
+            lease_seconds=60,
+        )
+
+
+@pytest.mark.asyncio
+async def test_host_contract_revision_and_document_freeze_are_one_mutation(
+    tmp_path,
+) -> None:
+    from test_coordinator_integration import FakeExecutor, make_coordinator
+
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    root_task_id = (await store.load("root")).root_task_id
+    first = await coordinator.create_task_graph(
+        "root",
+        (
+            {
+                "_task_id": "container",
+                "_parent_task_id": root_task_id,
+                "objective": "open",
+                "acceptance_criteria": (
+                    {"criterion_id": "closed", "description": "closed", "hard": True},
+                ),
+                "context_refs": ("contract://v1",),
+            },
+        ),
+        idempotency_key="open",
+        context_documents=(
+            FrozenContextDocument(
+                "contract://v1",
+                "sha256:" + "1" * 64,
+                {"schema_version": "test/v1"},
+            ),
+        ),
+    )
+    assert first["tasks"][0]["task_id"] == "container"
+    ledger = await store.load("root")
+    ledger.tasks["container"].status = TaskStatus.NEEDS_REPLAN
+    await store.commit(ledger, ledger.revision)
+
+    await coordinator.create_task_graph(
+        "root",
+        (
+            {
+                "_task_id": "container",
+                "_revision_task_id": "container",
+                "_parent_task_id": root_task_id,
+                "objective": "closed",
+                "acceptance_criteria": (
+                    {"criterion_id": "closed", "description": "closed", "hard": True},
+                ),
+                "context_refs": ("contract://v2",),
+            },
+        ),
+        idempotency_key="close",
+        context_documents=(
+            FrozenContextDocument(
+                "contract://v2",
+                "sha256:" + "2" * 64,
+                {"schema_version": "test/v2"},
+            ),
+        ),
+    )
+    revised = await store.load("root")
+    task = revised.tasks["container"]
+    assert task.status is TaskStatus.PENDING
+    assert task.current_spec_version == 2
+    assert task.current_spec.objective == "closed"
+    assert "contract://v2" in revised.context_documents

agent/tests/test_context_budget.py

@@ -151,6 +151,12 @@ async def test_build_900036_growth_replay_compresses_before_hard_limit(tmp_path)
     event_names = [event["event"] for event in events]
     assert "context_compression_started" in event_names
     assert "context_compression_completed" in event_names
+    prompt_events = [
+        event for event in events if event["event"] == "prompt_envelope_measured"
+    ]
+    assert len(prompt_events) == 97
+    assert prompt_events[-1]["actual_prompt_tokens"] > 0
+    assert prompt_events[-1]["packing_mode"] == "runner"
     assert max_prompt_tokens < 100_000
     assert total_prompt_tokens < 6_327_689 * 0.50
     usage = runner.get_context_usage(trace.trace_id)

agent/tests/test_orchestration_v2_control.py

@@ -186,7 +186,9 @@ async def test_worker_protects_operation_epoch_and_deadline():
     assert result.status == "completed"
     assert runner.config.trace_id is None
     assert runner.config.new_trace_id == "worker-trace"
-    assert runner.config.context == {
+    protected = dict(runner.config.context)
+    packing = protected.pop("prompt_packing")
+    assert protected == {
         "root_trace_id": "root",
         "task_id": "task",
         "spec_version": 2,
@@ -196,6 +198,13 @@ async def test_worker_protects_operation_epoch_and_deadline():
         "execution_epoch": 7,
         "deadline": "2030-01-01T00:00:00+00:00",
     }
+    assert packing == {
+        "mode": "full",
+        "estimated_tokens": 110,
+        "token_limit": 32_000,
+        "output_reserve": 8_192,
+        "deduplicated_segments": 0,
+    }
 
 
 @pytest.mark.asyncio