框架：增加业务上下文与确定性执行扩展端口

新增 TaskContextProvider、RoleContextProvider 和 DeterministicWorker 三个通用端口，并接入编排、执行器与公开导出。补全真实失败反馈、无进展状态和 80% 触发压缩至 30% 的上下文预算行为；确定性 Worker 仍生成正式零令牌 Trace、Attempt 并进入独立 Validator。

agent/agent/__init__.py

@@ -64,6 +64,12 @@ from agent.orchestration import (
     RoleRunConfigResolver,
     RoleSystemPromptOverride,
     RoleSystemPromptResolver,
+    RoleContextProvider,
+    RoleContextRequest,
+    TaskContextProvider,
+    DeterministicWorker,
+    DeterministicWorkerContext,
+    DeterministicWorkerResult,
 )
 from agent.orchestration.wiring import wire_orchestration
 
@@ -134,6 +140,12 @@ __all__ = [
     "RoleRunConfigResolver",
     "RoleSystemPromptOverride",
     "RoleSystemPromptResolver",
+    "RoleContextProvider",
+    "RoleContextRequest",
+    "TaskContextProvider",
+    "DeterministicWorker",
+    "DeterministicWorkerContext",
+    "DeterministicWorkerResult",
     "wire_orchestration",
     # SDK
     "invoke_agent",

agent/agent/core/prompts/compression.py

@@ -57,11 +57,14 @@ SUMMARY_HEADER_TEMPLATE = """## 对话历史摘要（自动压缩）
 {summary_text}
 
 ---
-*以上为压缩摘要，原始对话历史已归档。*
+*以上为压缩摘要，原始对话历史已归档。把摘要中已经成功读取的不可变输入、Accepted Artifact
+和工具结果视为仍然可用；不要因为原文已压缩而用相同参数重复读取。直接执行摘要所列的下一个
+未完成动作。只有摘要明确记录读取失败、数据缺失或引用已变化时，才重新读取。*
 """
 
 # ===== 辅助函数 =====
 
+
 def build_compression_eval_prompt(
     execution_context: str,
     ex_reference_list: str = "",

agent/agent/core/runner.py

@@ -77,6 +77,7 @@ class NoProgressPolicy:
     same_failure_limit: int = 2
     planner_stall_limit: int = 3
     enabled_in_legacy: bool = False
+    progress_epoch: str | None = None
 
     def __post_init__(self) -> None:
         if self.same_failure_limit < 1:
@@ -682,7 +683,9 @@ class AgentRunner:
             status="running",
         )
 
-        goal_tree = (self.goal_tree or GoalTree(mission=task_name)) if is_legacy else None
+        goal_tree = (
+            (self.goal_tree or GoalTree(mission=task_name)) if is_legacy else None
+        )
 
         if self.trace_store:
             await self.trace_store.create_trace(trace_obj)
@@ -941,12 +944,9 @@ class AgentRunner:
                     f"Context 使用率达到 {threshold}%: {token_count:,} / {max_tokens:,} tokens ({msg_count} 条消息)"
                 )
 
-        needs_compression = (
-            token_count >= measurement.trigger_tokens
-            or (
-                config.compression.max_messages > 0
-                and msg_count >= config.compression.max_messages
-            )
+        needs_compression = token_count >= measurement.trigger_tokens or (
+            config.compression.max_messages > 0
+            and msg_count >= config.compression.max_messages
         )
 
         if not needs_compression:
@@ -959,7 +959,9 @@ class AgentRunner:
                 measurement,
                 message_count=msg_count,
                 actual_prompt_tokens=budget_state.get("last_actual_prompt_tokens"),
-                reason="trigger_ratio" if token_count >= measurement.trigger_tokens else "message_limit",
+                reason="trigger_ratio"
+                if token_count >= measurement.trigger_tokens
+                else "message_limit",
             )
             budget_state["compression_before"] = started_payload
             trace.runtime_state["context_budget"] = budget_state
@@ -1104,7 +1106,12 @@ class AgentRunner:
             try:
                 duration_ms = max(
                     0,
-                    int((datetime.now() - datetime.fromisoformat(started_at)).total_seconds() * 1000),
+                    int(
+                        (
+                            datetime.now() - datetime.fromisoformat(started_at)
+                        ).total_seconds()
+                        * 1000
+                    ),
                 )
             except ValueError:
                 duration_ms = None
@@ -2468,12 +2475,18 @@ class AgentRunner:
                                 explicit_role,
                                 None,
                             )
-                            if terminal_control and terminal_control.get("terminate_run"):
+                            if terminal_control and terminal_control.get(
+                                "terminate_run"
+                            ):
                                 terminal_summary = (
                                     terminal_control.get("result_summary") or tool_text
                                 )
                                 terminal_run = True
-                    if terminal_summary and terminal_failure is None and self.trace_store:
+                    if (
+                        terminal_summary
+                        and terminal_failure is None
+                        and self.trace_store
+                    ):
                         await self.trace_store.update_trace(
                             trace_id,
                             result_summary=terminal_summary,
@@ -2759,7 +2772,9 @@ class AgentRunner:
                                 explicit_role,
                                 None,
                             )
-                            if terminal_control and terminal_control.get("terminate_run"):
+                            if terminal_control and terminal_control.get(
+                                "terminate_run"
+                            ):
                                 terminal_run = True
                                 terminal_summary = (
                                     terminal_control.get("result_summary") or tool_text
@@ -2773,9 +2788,9 @@ class AgentRunner:
                                 break
 
                 if terminal_run:
-                    if (
-                        terminal_failure is None
-                        and explicit_role in (AgentRole.WORKER, AgentRole.VALIDATOR)
+                    if terminal_failure is None and explicit_role in (
+                        AgentRole.WORKER,
+                        AgentRole.VALIDATOR,
                     ):
                         explicit_terminal_submitted = True
                     break
@@ -2965,9 +2980,9 @@ class AgentRunner:
     ) -> bool:
         if failure.disposition == FailureDisposition.ABORT_RUN:
             return True
-        return (
-            failure.disposition == FailureDisposition.REPLAN_TASK
-            and role in (AgentRole.WORKER, AgentRole.VALIDATOR)
+        return failure.disposition == FailureDisposition.REPLAN_TASK and role in (
+            AgentRole.WORKER,
+            AgentRole.VALIDATOR,
         )
 
     @staticmethod
@@ -3006,7 +3021,7 @@ class AgentRunner:
 
         if not self._no_progress_enabled(config, role):
             return None
-        runtime = trace.runtime_state.setdefault("no_progress", {})
+        runtime = self._no_progress_runtime(trace, config)
         if failure is None:
             runtime["failure_fingerprint"] = None
             runtime["failure_count"] = 0
@@ -3044,15 +3059,19 @@ class AgentRunner:
             return None
         root_trace_id = (trace.context or {}).get("root_trace_id", trace.trace_id)
         snapshot = await method(root_trace_id)
+        runtime = self._no_progress_runtime(trace, config)
+        semantic_state = {
+            "ledger": snapshot,
+            "last_failure_fingerprint": runtime.get("failure_fingerprint"),
+        }
         fingerprint = hashlib.sha256(
             json.dumps(
-                snapshot,
+                semantic_state,
                 ensure_ascii=False,
                 sort_keys=True,
                 separators=(",", ":"),
             ).encode("utf-8")
         ).hexdigest()
-        runtime = trace.runtime_state.setdefault("no_progress", {})
         previous = runtime.get("planner_fingerprint")
         if previous is None or previous != fingerprint:
             count = 0
@@ -3072,6 +3091,20 @@ class AgentRunner:
             },
         )
 
+    @staticmethod
+    def _no_progress_runtime(
+        trace: Trace,
+        config: RunConfig,
+    ) -> Dict[str, Any]:
+        """Reset counters when the caller starts a new semantic planning epoch."""
+
+        runtime = trace.runtime_state.setdefault("no_progress", {})
+        epoch = config.no_progress.progress_epoch
+        if epoch is not None and runtime.get("progress_epoch") != epoch:
+            runtime.clear()
+            runtime["progress_epoch"] = epoch
+        return runtime
+
     async def _persist_runtime_state(self, trace: Trace) -> None:
         if self.trace_store:
             await self.trace_store.update_trace(

agent/agent/llm/openrouter.py

@@ -65,8 +65,9 @@ def _resolve_openrouter_model(model: str) -> str:
 def _sanitize_schema_name(title: str) -> str:
     """清理 schema title 为符合 API 要求的 name（只允许 [a-zA-Z0-9_-]）"""
     import re
-    sanitized = re.sub(r'[^a-zA-Z0-9_-]', '_', title)
-    return sanitized.strip('_') or "response"
+
+    sanitized = re.sub(r"[^a-zA-Z0-9_-]", "_", title)
+    return sanitized.strip("_") or "response"
 
 
 def _ensure_strict_schema(schema: Dict) -> Dict:
@@ -83,6 +84,7 @@ def _ensure_strict_schema(schema: Dict) -> Dict:
         转换后的 schema（深拷贝）
     """
     import copy
+
     result = copy.deepcopy(schema)
 
     def _process(obj):
@@ -100,7 +102,7 @@ def _ensure_strict_schema(schema: Dict) -> Dict:
             if len(non_null_types) == 1:
                 base_type = non_null_types[0]
                 if base_type == "object":
-                    # type: ["object", "null"] → anyOf
+                    # Convert nullable object schemas to anyOf.
                     props = obj.pop("properties", {})
                     obj.pop("type", None)
                     obj.pop("additionalProperties", None)
@@ -112,20 +114,17 @@ def _ensure_strict_schema(schema: Dict) -> Dict:
                             "required": list(props.keys()),
                             "additionalProperties": False,
                         },
-                        {"type": "null"}
+                        {"type": "null"},
                     ]
                     # 递归处理 props 内部
                     for v in props.values():
                         _process(v)
                     return
                 else:
-                    # type: ["string", "null"] 等简单类型 → anyOf
+                    # Convert nullable scalar schemas to anyOf.
                     remaining = {k: v for k, v in obj.items() if k != "type"}
                     obj.clear()
-                    obj["anyOf"] = [
-                        {"type": base_type, **remaining},
-                        {"type": "null"}
-                    ]
+                    obj["anyOf"] = [{"type": base_type, **remaining}, {"type": "null"}]
                     return
 
         # 处理纯 object 类型
@@ -158,12 +157,18 @@ def _to_anthropic_messages(
 
 # ── Provider detection / usage parsing ─────────────────────────────────────
 
+
 def _detect_provider_from_model(model: str) -> str:
     """根据模型名称检测提供商"""
     model_lower = model.lower()
     if model_lower.startswith("anthropic/") or "claude" in model_lower:
         return "anthropic"
-    elif model_lower.startswith("openai/") or model_lower.startswith("gpt") or model_lower.startswith("o1") or model_lower.startswith("o3"):
+    elif (
+        model_lower.startswith("openai/")
+        or model_lower.startswith("gpt")
+        or model_lower.startswith("o1")
+        or model_lower.startswith("o3")
+    ):
         return "openai"
     elif model_lower.startswith("deepseek/") or "deepseek" in model_lower:
         return "deepseek"
@@ -194,7 +199,8 @@ def _parse_openrouter_usage(usage: Dict[str, Any], model: str) -> TokenUsage:
 
         return TokenUsage(
             input_tokens=usage.get("prompt_tokens") or usage.get("input_tokens", 0),
-            output_tokens=usage.get("completion_tokens") or usage.get("output_tokens", 0),
+            output_tokens=usage.get("completion_tokens")
+            or usage.get("output_tokens", 0),
             # OpenRouter 格式：prompt_tokens_details.cached_tokens / cache_write_tokens
             cache_read_tokens=prompt_details.get("cached_tokens", 0),
             cache_creation_tokens=prompt_details.get("cache_write_tokens", 0),
@@ -252,7 +258,9 @@ async def _openrouter_anthropic_call(
                 if isinstance(_b, dict) and _b.get("type") == "image":
                     _img_count += 1
     if _img_count:
-        logger.info("[OpenRouter/Anthropic] payload contains %d image block(s)", _img_count)
+        logger.info(
+            "[OpenRouter/Anthropic] payload contains %d image block(s)", _img_count
+        )
 
     payload: Dict[str, Any] = {
         "model": resolved_model,
@@ -276,15 +284,20 @@ async def _openrouter_anthropic_call(
                 "name": schema_name,
                 "strict": True,
                 "schema": _ensure_strict_schema(schema),
-            }
+            },
         }
-        logger.info("[OpenRouter/Anthropic] Using structured output with schema: %s", schema_name)
+        logger.info(
+            "[OpenRouter/Anthropic] Using structured output with schema: %s",
+            schema_name,
+        )
 
     # Debug: 检查 cache_control 是否存在
     if logger.isEnabledFor(logging.DEBUG):
         cache_control_count = count_cache_controls(system_prompt, anthropic_messages)
         if cache_control_count > 0:
-            logger.debug(f"[OpenRouter/Anthropic] 发现 {cache_control_count} 个 cache_control 标记")
+            logger.debug(
+                f"[OpenRouter/Anthropic] 发现 {cache_control_count} 个 cache_control 标记"
+            )
 
     headers = {
         "Authorization": f"Bearer {api_key}",
@@ -308,24 +321,33 @@ async def _openrouter_anthropic_call(
                 status = e.response.status_code
                 error_body = e.response.text
                 if status in (429, 500, 502, 503, 504) and attempt < max_retries - 1:
-                    wait = 2 ** attempt * 2
+                    wait = 2**attempt * 2
                     logger.warning(
                         "[OpenRouter/Anthropic] HTTP %d (attempt %d/%d), retrying in %ds: %s",
-                        status, attempt + 1, max_retries, wait, error_body[:200],
+                        status,
+                        attempt + 1,
+                        max_retries,
+                        wait,
+                        error_body[:200],
                     )
                     await asyncio.sleep(wait)
                     last_exception = e
                     continue
-                logger.error("[OpenRouter/Anthropic] HTTP %d error body: %s", status, error_body)
+                logger.error(
+                    "[OpenRouter/Anthropic] HTTP %d error body: %s", status, error_body
+                )
                 raise
 
             except _RETRYABLE_EXCEPTIONS as e:
                 last_exception = e
                 if attempt < max_retries - 1:
-                    wait = 2 ** attempt * 2
+                    wait = 2**attempt * 2
                     logger.warning(
                         "[OpenRouter/Anthropic] %s (attempt %d/%d), retrying in %ds",
-                        type(e).__name__, attempt + 1, max_retries, wait,
+                        type(e).__name__,
+                        attempt + 1,
+                        max_retries,
+                        wait,
                     )
                     await asyncio.sleep(wait)
                     continue
@@ -336,7 +358,9 @@ async def _openrouter_anthropic_call(
     return build_anthropic_result(_parse_anthropic_response(result), model)
 
 
-def _resolve_local_images_in_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+def _resolve_local_images_in_messages(
+    messages: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
     """将消息中指向本地文件的 image_url 提前转为 data:image/...;base64,... URI，
     确保所有 Provider (包括 fallback 到 chat/completions 的 Gemini/GPT) 都能正确接收图片。
     """
@@ -358,7 +382,11 @@ def _resolve_local_images_in_messages(messages: List[Dict[str, Any]]) -> List[Di
                 continue
 
             image_url_obj = block.get("image_url", {})
-            url = image_url_obj.get("url", "") if isinstance(image_url_obj, dict) else str(image_url_obj)
+            url = (
+                image_url_obj.get("url", "")
+                if isinstance(image_url_obj, dict)
+                else str(image_url_obj)
+            )
 
             if not url.startswith("data:") and not url.startswith("http"):
                 local_path = Path(url)
@@ -368,7 +396,9 @@ def _resolve_local_images_in_messages(messages: List[Dict[str, Any]]) -> List[Di
                     raw = local_path.read_bytes()
                     b64_data = base64.b64encode(raw).decode("ascii")
                     new_url = f"data:{mime_type};base64,{b64_data}"
-                    logger.info(f"[OpenRouter] 本地图片统一转 base64: {url} ({len(raw)} bytes)")
+                    logger.info(
+                        f"[OpenRouter] 本地图片统一转 base64: {url} ({len(raw)} bytes)"
+                    )
 
                     new_block = copy.deepcopy(block)
                     if isinstance(new_block["image_url"], dict):
@@ -384,11 +414,12 @@ def _resolve_local_images_in_messages(messages: List[Dict[str, Any]]) -> List[Di
 
     return result
 
+
 async def openrouter_llm_call(
     messages: List[Dict[str, Any]],
     model: str = "anthropic/claude-sonnet-4.5",
     tools: Optional[List[Dict]] = None,
-    **kwargs
+    **kwargs,
 ) -> Dict[str, Any]:
     """
     OpenRouter LLM 调用函数
@@ -425,7 +456,9 @@ async def openrouter_llm_call(
         logger.debug("[OpenRouter] Routing Claude model to Anthropic native endpoint")
         if response_schema:
             kwargs["response_schema"] = response_schema
-        return await _openrouter_anthropic_call(messages, model, tools, api_key, **kwargs)
+        return await _openrouter_anthropic_call(
+            messages, model, tools, api_key, **kwargs
+        )
 
     base_url = "https://openrouter.ai/api/v1"
     endpoint = f"{base_url}/chat/completions"
@@ -457,7 +490,7 @@ async def openrouter_llm_call(
                 "name": schema_name,
                 "strict": True,
                 "schema": _ensure_strict_schema(response_schema),
-            }
+            },
         }
 
     # OpenRouter 特定参数
@@ -483,10 +516,14 @@ async def openrouter_llm_call(
                 status = e.response.status_code
                 # 429 (rate limit) 和 5xx 可重试
                 if status in (429, 500, 502, 503, 504) and attempt < max_retries - 1:
-                    wait = 2 ** attempt * 2  # 2s, 4s, 8s
+                    wait = 2**attempt * 2  # 2s, 4s, 8s
                     logger.warning(
                         "[OpenRouter] HTTP %d (attempt %d/%d), retrying in %ds: %s",
-                        status, attempt + 1, max_retries, wait, error_body[:200],
+                        status,
+                        attempt + 1,
+                        max_retries,
+                        wait,
+                        error_body[:200],
                     )
                     await asyncio.sleep(wait)
                     last_exception = e
@@ -497,14 +534,19 @@ async def openrouter_llm_call(
             except _RETRYABLE_EXCEPTIONS as e:
                 last_exception = e
                 if attempt < max_retries - 1:
-                    wait = 2 ** attempt * 2
+                    wait = 2**attempt * 2
                     logger.warning(
                         "[OpenRouter] %s (attempt %d/%d), retrying in %ds",
-                        type(e).__name__, attempt + 1, max_retries, wait,
+                        type(e).__name__,
+                        attempt + 1,
+                        max_retries,
+                        wait,
                     )
                     await asyncio.sleep(wait)
                     continue
-                logger.error("[OpenRouter] Request failed after %d attempts: %s", max_retries, e)
+                logger.error(
+                    "[OpenRouter] Request failed after %d attempts: %s", max_retries, e
+                )
                 raise
 
             except Exception as e:
@@ -520,7 +562,9 @@ async def openrouter_llm_call(
 
     content = message.get("content", "")
     tool_calls = message.get("tool_calls")
-    finish_reason = choice.get("finish_reason")  # stop, length, tool_calls, content_filter 等
+    finish_reason = choice.get(
+        "finish_reason"
+    )  # stop, length, tool_calls, content_filter 等
 
     # 提取 usage（完整版，根据模型类型解析）
     raw_usage = result.get("usage", {})
@@ -543,9 +587,7 @@ async def openrouter_llm_call(
     }
 
 
-def create_openrouter_llm_call(
-    model: str = "anthropic/claude-sonnet-4.5"
-):
+def create_openrouter_llm_call(model: str = "anthropic/claude-sonnet-4.5"):
     """
     创建 OpenRouter LLM 调用函数
 
@@ -559,11 +601,12 @@ def create_openrouter_llm_call(
     Returns:
         异步 LLM 调用函数
     """
+
     async def llm_call(
         messages: List[Dict[str, Any]],
         model: str = model,
         tools: Optional[List[Dict]] = None,
-        **kwargs
+        **kwargs,
     ) -> Dict[str, Any]:
         return await openrouter_llm_call(messages, model, tools, **kwargs)
 

agent/agent/orchestration/__init__.py

@@ -73,6 +73,16 @@ from .validation_policy import (
     ValidationContext,
     ValidationPolicy,
 )
+from .context_provider import (
+    RoleContextProvider,
+    RoleContextRequest,
+    TaskContextProvider,
+)
+from .deterministic_worker import (
+    DeterministicWorker,
+    DeterministicWorkerContext,
+    DeterministicWorkerResult,
+)
 from .wire import (
     ArtifactSnapshotView,
     CapabilitiesView,
@@ -105,6 +115,9 @@ __all__ = [
     "DeterministicRuleResult",
     "DeterministicValidationResult",
     "DeterministicValidator",
+    "DeterministicWorker",
+    "DeterministicWorkerContext",
+    "DeterministicWorkerResult",
     "EventPage",
     "EventSink",
     "EvidenceBudgetExceeded",
@@ -135,6 +148,8 @@ __all__ = [
     "ProblemDetails",
     "RevisionConflict",
     "RoleRunConfigOverrides",
+    "RoleContextProvider",
+    "RoleContextRequest",
     "RoleRunConfigResolver",
     "RoleSystemPromptOverride",
     "RoleSystemPromptResolver",
@@ -143,6 +158,7 @@ __all__ = [
     "TaskAttempt",
     "TaskConflict",
     "TaskCoordinator",
+    "TaskContextProvider",
     "TaskCycleResult",
     "TaskLedger",
     "TaskRecord",

agent/agent/orchestration/_task_graph.py

@@ -102,8 +102,8 @@ class TaskGraph:
                     "Attempt child decision bindings are not in display_path order"
                 )
             prior_path = child_path
-            bound_decision, child_attempt, validation = TaskGraph.accepted_result_binding(
-                ledger, child
+            bound_decision, child_attempt, validation = (
+                TaskGraph.accepted_result_binding(ledger, child)
             )
             if bound_decision.decision_id != decision_id:
                 raise OrchestrationError(
@@ -111,6 +111,7 @@ class TaskGraph:
                 )
             results.append(
                 {
+                    "decision_id": decision_id,
                     "task_id": child.task_id,
                     "task_spec": json_values(asdict(child.current_spec)),
                     "attempt_id": child_attempt.attempt_id,
@@ -243,7 +244,9 @@ class TaskGraph:
         parent_task_id: Optional[str],
         display_path: str,
     ) -> str:
-        task_id = new_id()
+        task_id = str(draft.get("_task_id") or new_id())
+        if task_id in ledger.tasks:
+            raise ValueError(f"Task already exists: {task_id}")
         spec = TaskGraph.task_spec_from_draft(draft, version=1)
         ledger.tasks[task_id] = TaskRecord(
             task_id=task_id,

agent/agent/orchestration/context_provider.py

@@ -0,0 +1,40 @@
+"""Optional business context projections for explicit orchestration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Mapping, Protocol
+
+from .models import AgentRole, TaskLedger
+
+
+class TaskContextProvider(Protocol):
+    """Render a compact Planner context from an authoritative ledger."""
+
+    async def render(self, root_trace_id: str, ledger: TaskLedger) -> str: ...
+
+
+@dataclass(frozen=True, slots=True)
+class RoleContextRequest:
+    """Framework-owned identity used to request a role-specific bootstrap."""
+
+    role: AgentRole
+    root_trace_id: str
+    task_id: str
+    spec_version: int
+    task_spec: Mapping[str, Any]
+    attempt_id: str
+    ledger_revision: int
+    accepted_child_results: tuple[Mapping[str, Any], ...] = ()
+    snapshot_id: str | None = None
+    artifact_snapshot: Mapping[str, Any] | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+
+class RoleContextProvider(Protocol):
+    """Build protected Worker or Validator bootstrap data."""
+
+    async def build(self, request: RoleContextRequest) -> Mapping[str, Any]: ...
+
+
+__all__ = ["RoleContextProvider", "RoleContextRequest", "TaskContextProvider"]

agent/agent/orchestration/coordinator.py

@@ -11,7 +11,8 @@ from dataclasses import asdict, is_dataclass
 from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
 from uuid import uuid4
 
-from agent.failures import FailureDetail
+from agent.failures import FailureDetail, ToolExecutionError
+from agent.trace.models import Trace
 
 from .config import OrchestrationConfig
 from .models import (
@@ -74,6 +75,12 @@ from .validation_policy import (
 )
 from ._decision_engine import DecisionEngine
 from ._task_context import render_task_context
+from .context_provider import (
+    RoleContextProvider,
+    RoleContextRequest,
+    TaskContextProvider,
+)
+from .deterministic_worker import DeterministicWorker, DeterministicWorkerContext
 from ._task_graph import (
     TERMINAL_TASK_STATUSES,
     TaskGraph,
@@ -102,6 +109,9 @@ class TaskCoordinator:
         validation_policy: Optional[ValidationPolicy] = None,
         deterministic_validator: Optional[DeterministicValidator] = None,
         evidence_provider: Optional[EvidenceProvider] = None,
+        task_context_provider: Optional[TaskContextProvider] = None,
+        role_context_provider: Optional[RoleContextProvider] = None,
+        deterministic_worker: Optional[DeterministicWorker] = None,
     ) -> None:
         self.task_store = task_store
         self.artifact_store = artifact_store
@@ -114,6 +124,9 @@ class TaskCoordinator:
         )
         self.deterministic_validator = deterministic_validator
         self.evidence_provider = evidence_provider
+        self.task_context_provider = task_context_provider
+        self.role_context_provider = role_context_provider
+        self.deterministic_worker = deterministic_worker
         self._locks: Dict[str, asyncio.Lock] = {}
         self._task_graph = TaskGraph()
         self._decision_engine = DecisionEngine(
@@ -157,7 +170,9 @@ class TaskCoordinator:
                 try:
                     ledger = await self.task_store.load(root_trace_id)
                     if ledger.root_trace_id != root_trace_id:
-                        raise OrchestrationError("Task ledger root trace identity is invalid")
+                        raise OrchestrationError(
+                            "Task ledger root trace identity is invalid"
+                        )
                     self._root_task(ledger)
                     return ledger
                 except TaskStoreNotFound:
@@ -173,7 +188,9 @@ class TaskCoordinator:
                     if not isinstance(root_task_spec, dict):
                         raise ValueError("root_task_spec must be an object")
                     unknown = set(root_task_spec) - {
-                        "objective", "acceptance_criteria", "context_refs",
+                        "objective",
+                        "acceptance_criteria",
+                        "context_refs",
                     }
                     if unknown:
                         raise ValueError(
@@ -222,9 +239,7 @@ class TaskCoordinator:
         root = self._root_task(ledger)
         nonterminal_descendants = self._nonterminal_descendants(ledger, root)
         if root.status == TaskStatus.COMPLETED and nonterminal_descendants:
-            raise OrchestrationError(
-                "Completed root task has non-terminal descendants"
-            )
+            raise OrchestrationError("Completed root task has non-terminal descendants")
         result_summary = None
         if root.status == TaskStatus.COMPLETED:
             _, attempt, _ = self._accepted_result_binding(ledger, root)
@@ -236,7 +251,9 @@ class TaskCoordinator:
                 "status": task.status.value,
                 "objective": task.current_spec.objective,
             }
-            for task in sorted(ledger.tasks.values(), key=lambda item: _path_key(item.display_path))
+            for task in sorted(
+                ledger.tasks.values(), key=lambda item: _path_key(item.display_path)
+            )
             if task.status not in TERMINAL_TASK_STATUSES
         ]
         return {
@@ -279,6 +296,8 @@ class TaskCoordinator:
 
         ledger = await self.task_store.load(root_trace_id)
         self._root_task(ledger)
+        if self.task_context_provider is not None:
+            return await self.task_context_provider.render(root_trace_id, ledger)
         return render_task_context(ledger)
 
     async def progress_snapshot(self, root_trace_id: str) -> Dict[str, Any]:
@@ -340,7 +359,11 @@ class TaskCoordinator:
                         for ref in attempt.submission.artifact_refs
                     ),
                     "snapshot_sha256": (
-                        (await self.artifact_store.get(root_trace_id, attempt.snapshot_id)).sha256
+                        (
+                            await self.artifact_store.get(
+                                root_trace_id, attempt.snapshot_id
+                            )
+                        ).sha256
                         if attempt.snapshot_id
                         else None
                     ),
@@ -395,7 +418,10 @@ class TaskCoordinator:
                 ledger = await self.task_store.load(root_trace_id)
                 if command_id and command_id in ledger.command_records:
                     command = ledger.command_records[command_id]
-                    if command.operation != event_type or command.fingerprint != fingerprint:
+                    if (
+                        command.operation != event_type
+                        or command.fingerprint != fingerprint
+                    ):
                         raise TaskConflict(
                             "Idempotency key is bound to a different operation or request"
                         )
@@ -483,8 +509,13 @@ class TaskCoordinator:
                 target = _task(ledger, after_task_id)
                 if target.task_id == ledger.root_task_id:
                     raise ValueError("The root task cannot have siblings")
-                if parent_task_id is not None and target.parent_task_id != parent_task_id:
-                    raise ValueError("after_task_id must be a sibling under parent_task_id")
+                if (
+                    parent_task_id is not None
+                    and target.parent_task_id != parent_task_id
+                ):
+                    raise ValueError(
+                        "after_task_id must be a sibling under parent_task_id"
+                    )
                 effective_parent_id = target.parent_task_id
             else:
                 effective_parent_id = parent_task_id or ledger.root_task_id
@@ -503,13 +534,23 @@ class TaskCoordinator:
                     f"{parent.status.value}"
                 )
             siblings = sorted(
-                (t for t in ledger.tasks.values() if t.parent_task_id == effective_parent_id),
+                (
+                    t
+                    for t in ledger.tasks.values()
+                    if t.parent_task_id == effective_parent_id
+                ),
                 key=lambda item: _path_key(item.display_path),
             )
             if after_task_id:
-                target_index = next(i for i, sibling in enumerate(siblings) if sibling.task_id == after_task_id)
+                target_index = next(
+                    i
+                    for i, sibling in enumerate(siblings)
+                    if sibling.task_id == after_task_id
+                )
                 start = target_index + 2
-                for index, sibling in enumerate(siblings[target_index + 1:], start=start + len(drafts)):
+                for index, sibling in enumerate(
+                    siblings[target_index + 1 :], start=start + len(drafts)
+                ):
                     self._rebase_task_path(
                         ledger,
                         sibling,
@@ -517,7 +558,11 @@ class TaskCoordinator:
                     )
             else:
                 start = len(siblings) + 1
-            parent_path = ledger.tasks[effective_parent_id].display_path if effective_parent_id else ""
+            parent_path = (
+                ledger.tasks[effective_parent_id].display_path
+                if effective_parent_id
+                else ""
+            )
             created: List[str] = []
             for offset, draft in enumerate(drafts):
                 index = start + offset
@@ -716,29 +761,35 @@ class TaskCoordinator:
                     reserved = await self._create_attempt(
                         operation.root_trace_id,
                         task_id,
-                        presets[index] if index < len(presets) else self.config.default_worker_preset,
+                        presets[index]
+                        if index < len(presets)
+                        else self.config.default_worker_preset,
                         None,
                         operation_id=operation.operation_id,
                         execution_epoch=operation.execution_epoch,
                     )
                     attempt_id = reserved["attempt_id"]
                 except Exception as exc:
-                    results.append(await self._cycle_error_result(
-                        operation.root_trace_id,
-                        task_id,
-                        None,
-                        f"Attempt reservation failed during resume: "
-                        f"{type(exc).__name__}: {exc}",
-                    ))
+                    results.append(
+                        await self._cycle_error_result(
+                            operation.root_trace_id,
+                            task_id,
+                            None,
+                            f"Attempt reservation failed during resume: "
+                            f"{type(exc).__name__}: {exc}",
+                        )
+                    )
                     continue
-            results.append(await self.advance_cycle(
-                operation.root_trace_id,
-                task_id,
-                attempt_id,
-                operation_id=operation.operation_id,
-                execution_epoch=operation.execution_epoch,
-                deadline=operation.deadline_at,
-            ))
+            results.append(
+                await self.advance_cycle(
+                    operation.root_trace_id,
+                    task_id,
+                    attempt_id,
+                    operation_id=operation.operation_id,
+                    execution_epoch=operation.execution_epoch,
+                    deadline=operation.deadline_at,
+                )
+            )
         return results
 
     async def _operation_cycle_results(
@@ -762,11 +813,15 @@ class TaskCoordinator:
                 task_id,
                 attempt_ids[index],
             )
-            if operation.status in {
-                OperationStatus.PENDING,
-                OperationStatus.RUNNING,
-                OperationStatus.STOP_REQUESTED,
-            } and not result.error:
+            if (
+                operation.status
+                in {
+                    OperationStatus.PENDING,
+                    OperationStatus.RUNNING,
+                    OperationStatus.STOP_REQUESTED,
+                }
+                and not result.error
+            ):
                 result = TaskCycleResult(
                     task_id=result.task_id,
                     task_status=result.task_status,
@@ -892,13 +947,17 @@ class TaskCoordinator:
                 # could corrupt a pre-existing run owned by another dispatch.
                 attempt_id = None
                 early_results[index] = await self._cycle_error_result(
-                    root_trace_id, task_id, attempt_id,
+                    root_trace_id,
+                    task_id,
+                    attempt_id,
                     f"Attempt reservation failed: {type(exc).__name__}: {exc}",
                 )
 
         semaphore = asyncio.Semaphore(self.config.max_parallel_tasks)
 
-        async def run_one(index: int, task_id: str, attempt_id: str) -> Tuple[int, TaskCycleResult]:
+        async def run_one(
+            index: int, task_id: str, attempt_id: str
+        ) -> Tuple[int, TaskCycleResult]:
             async with semaphore:
                 try:
                     result = await self.advance_cycle(
@@ -925,7 +984,10 @@ class TaskCoordinator:
                 return index, result
 
         completed = await asyncio.gather(
-            *(run_one(index, task_id, attempt_id) for index, task_id, attempt_id in reservations)
+            *(
+                run_one(index, task_id, attempt_id)
+                for index, task_id, attempt_id in reservations
+            )
         )
         indexed_results = dict(early_results)
         indexed_results.update(completed)
@@ -952,7 +1014,9 @@ class TaskCoordinator:
                     validation_id = candidate_id
                     break
         error = validation.error if validation else (attempt.error if attempt else None)
-        failure = validation.failure if validation else (attempt.failure if attempt else None)
+        failure = (
+            validation.failure if validation else (attempt.failure if attempt else None)
+        )
         if task.status in {
             TaskStatus.RUNNING,
             TaskStatus.AWAITING_VALIDATION,
@@ -1113,13 +1177,16 @@ class TaskCoordinator:
                     )
             task = _task(ledger, task_id)
             if task.status not in (TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN):
-                raise TaskConflict(f"Task {task_id} cannot dispatch from {task.status.value}")
+                raise TaskConflict(
+                    f"Task {task_id} cannot dispatch from {task.status.value}"
+                )
             if self._nonterminal_descendants(ledger, task):
                 raise TaskConflict(
                     f"Task {task_id} cannot dispatch while descendants are non-terminal"
                 )
             active = [
-                ledger.attempts[x] for x in task.attempt_ids
+                ledger.attempts[x]
+                for x in task.attempt_ids
                 if ledger.attempts[x].status == AttemptStatus.RUNNING
             ]
             if active:
@@ -1229,9 +1296,7 @@ class TaskCoordinator:
                 worker_result = WorkerRunResult(
                     trace_id=attempt.worker_trace_id,
                     status="failed",
-                    error=(
-                        "Legacy running attempt has no frozen child input binding"
-                    ),
+                    error=("Legacy running attempt has no frozen child input binding"),
                     execution_stats=ExecutionStats(
                         failure_code=FailureCode.PROTOCOL_VIOLATION
                     ),
@@ -1287,9 +1352,7 @@ class TaskCoordinator:
                     else None
                 ),
                 "repair_feedback": (
-                    prior_feedback
-                    if attempt.execution_mode == "repair"
-                    else None
+                    prior_feedback if attempt.execution_mode == "repair" else None
                 ),
                 "prior_feedback": prior_feedback,
                 "accepted_child_results": accepted_child_results,
@@ -1299,11 +1362,109 @@ class TaskCoordinator:
             }
             if worker_result is None:
                 try:
-                    worker_result = await asyncio.wait_for(
-                        self.executor.run_worker(worker_context),  # type: ignore[union-attr]
-                        timeout=stage_timeout(
-                            deadline,
-                            self.config.worker_timeout_seconds,
+                    role_context = await self._build_role_context(
+                        RoleContextRequest(
+                            role=AgentRole.WORKER,
+                            root_trace_id=root_trace_id,
+                            task_id=task_id,
+                            spec_version=task.current_spec_version,
+                            task_spec=worker_context["task_spec"],
+                            attempt_id=attempt_id,
+                            ledger_revision=ledger.revision,
+                            accepted_child_results=tuple(accepted_child_results),
+                            metadata={
+                                "operation_id": operation_id,
+                                "execution_epoch": execution_epoch,
+                            },
+                        )
+                    )
+                    worker_context["role_context"] = role_context
+                    deterministic_context = DeterministicWorkerContext(
+                        root_trace_id=root_trace_id,
+                        task=task,
+                        attempt=attempt,
+                        ledger_revision=ledger.revision,
+                        accepted_child_results=tuple(accepted_child_results),
+                        operation_id=operation_id,
+                        execution_epoch=execution_epoch,
+                        role_context=role_context,
+                    )
+                    timeout = stage_timeout(
+                        deadline, self.config.worker_timeout_seconds
+                    )
+                    if (
+                        self.deterministic_worker is not None
+                        and await self.deterministic_worker.supports(
+                            deterministic_context
+                        )
+                    ):
+                        deterministic_result = await asyncio.wait_for(
+                            self.deterministic_worker.execute(deterministic_context),
+                            timeout=timeout,
+                        )
+                        if self.trace_store is not None:
+                            existing_trace = await self.trace_store.get_trace(
+                                attempt.worker_trace_id
+                            )
+                            if existing_trace is None:
+                                await self.trace_store.create_trace(
+                                    Trace(
+                                        trace_id=attempt.worker_trace_id,
+                                        mode="agent",
+                                        task=task.current_spec.objective,
+                                        agent_type=attempt.worker_preset,
+                                        agent_role=AgentRole.WORKER.value,
+                                        parent_trace_id=root_trace_id,
+                                        status="completed",
+                                        total_messages=0,
+                                        total_tokens=0,
+                                        result_summary=deterministic_result.summary,
+                                        context={
+                                            "task_id": task_id,
+                                            "spec_version": task.current_spec_version,
+                                            "deterministic_worker": True,
+                                        },
+                                    )
+                                )
+                        await self.submit_attempt(
+                            {
+                                "role": AgentRole.WORKER.value,
+                                "root_trace_id": root_trace_id,
+                                "task_id": task_id,
+                                "attempt_id": attempt_id,
+                                "spec_version": task.current_spec_version,
+                                "trace_id": attempt.worker_trace_id,
+                                "tool_call_id": f"deterministic-worker:{attempt_id}",
+                                "operation_id": operation_id,
+                                "execution_epoch": execution_epoch,
+                            },
+                            AttemptSubmission(
+                                summary=deterministic_result.summary,
+                                artifact_refs=list(deterministic_result.artifact_refs),
+                                evidence_refs=list(deterministic_result.evidence_refs),
+                            ),
+                        )
+                        worker_result = WorkerRunResult(
+                            trace_id=attempt.worker_trace_id,
+                            status="completed",
+                            summary=deterministic_result.summary,
+                            execution_stats=ExecutionStats(
+                                total_tokens=0, total_cost=0.0
+                            ),
+                        )
+                    else:
+                        worker_result = await asyncio.wait_for(
+                            self.executor.run_worker(worker_context),  # type: ignore[union-attr]
+                            timeout=timeout,
+                        )
+                except ToolExecutionError as exc:
+                    worker_result = WorkerRunResult(
+                        trace_id=attempt.worker_trace_id,
+                        status="failed",
+                        error=exc.failure.message,
+                        failure=exc.failure,
+                        execution_stats=ExecutionStats(
+                            failure_code=FailureCode.TOOL_FAILURE
                         ),
                     )
                 except asyncio.TimeoutError:
@@ -1311,7 +1472,9 @@ class TaskCoordinator:
                         trace_id=attempt.worker_trace_id,
                         status="expired",
                         error="Worker execution timed out",
-                        execution_stats=ExecutionStats(failure_code=FailureCode.TIMEOUT),
+                        execution_stats=ExecutionStats(
+                            failure_code=FailureCode.TIMEOUT
+                        ),
                     )
                 except Exception as exc:
                     worker_result = WorkerRunResult(
@@ -1522,10 +1685,36 @@ class TaskCoordinator:
                     timeout=timeout,
                 )
             else:
+                validator_context["role_context"] = await self._build_role_context(
+                    RoleContextRequest(
+                        role=AgentRole.VALIDATOR,
+                        root_trace_id=root_trace_id,
+                        task_id=task_id,
+                        spec_version=task.current_spec_version,
+                        task_spec=validator_context["task_spec"],
+                        attempt_id=attempt_id,
+                        ledger_revision=ledger.revision,
+                        snapshot_id=snapshot.snapshot_id,
+                        artifact_snapshot=validator_context["artifact_snapshot"],
+                        metadata={
+                            "validation_id": validation_id,
+                            "operation_id": operation_id,
+                            "execution_epoch": execution_epoch,
+                        },
+                    )
+                )
                 validator_result = await asyncio.wait_for(
                     self.executor.run_validator(validator_context),  # type: ignore[union-attr]
                     timeout=timeout,
                 )
+        except ToolExecutionError as exc:
+            validator_result = ValidatorRunResult(
+                trace_id=validation.validator_trace_id,
+                status="error",
+                error=exc.failure.message,
+                failure=exc.failure,
+                execution_stats=ExecutionStats(failure_code=FailureCode.TOOL_FAILURE),
+            )
         except asyncio.TimeoutError:
             validator_result = ValidatorRunResult(
                 trace_id=validation.validator_trace_id,
@@ -1538,9 +1727,7 @@ class TaskCoordinator:
                 trace_id=validation.validator_trace_id,
                 status="error",
                 error=f"Validator executor error: {exc}",
-                execution_stats=ExecutionStats(
-                    failure_code=FailureCode.EXECUTOR_ERROR
-                ),
+                execution_stats=ExecutionStats(failure_code=FailureCode.EXECUTOR_ERROR),
             )
         ledger = await self.task_store.load(root_trace_id)
         validation = ledger.validations[validation_id]
@@ -1593,6 +1780,14 @@ class TaskCoordinator:
             failure=validation.failure,
         )
 
+    async def _build_role_context(self, request: RoleContextRequest) -> Dict[str, Any]:
+        if self.role_context_provider is None:
+            return {}
+        value = await self.role_context_provider.build(request)
+        if not isinstance(value, dict):
+            value = dict(value)
+        return _plain(value)
+
     async def _reuse_semantic_validation(
         self,
         context: ValidationContext,
@@ -1642,7 +1837,9 @@ class TaskCoordinator:
                 },
                 prior.verdict,
                 tuple(prior.criterion_results),
-                f"semantic_validation_reused_from={prior.validation_id};{prior.summary}"[:500],
+                f"semantic_validation_reused_from={prior.validation_id};{prior.summary}"[
+                    :500
+                ],
                 tuple(prior.evidence_refs),
                 tuple(prior.unverified_claims),
                 tuple(prior.risks),
@@ -1727,15 +1924,16 @@ class TaskCoordinator:
     ) -> ValidatorRunResult:
         if self.deterministic_validator is None:
             raise RuntimeError("No DeterministicValidator is configured")
-        result: DeterministicValidationResult = await self.deterministic_validator.validate(
-            context,
-            validation.validation_plan,
+        result: DeterministicValidationResult = (
+            await self.deterministic_validator.validate(
+                context,
+                validation.validation_plan,
+            )
         )
         returned_rule_ids = [item.rule_id for item in result.rule_results]
-        if (
-            len(set(returned_rule_ids)) != len(returned_rule_ids)
-            or set(returned_rule_ids) != set(validation.validation_plan.rule_ids)
-        ):
+        if len(set(returned_rule_ids)) != len(returned_rule_ids) or set(
+            returned_rule_ids
+        ) != set(validation.validation_plan.rule_ids):
             raise ValueError(
                 "DeterministicValidator results must match the frozen rule_ids exactly"
             )
@@ -1761,9 +1959,7 @@ class TaskCoordinator:
                     )
                 )
         evidence_refs = [
-            ref
-            for item in result.rule_results
-            for ref in item.evidence_refs
+            ref for item in result.rule_results for ref in item.evidence_refs
         ]
         errors = result.errors
         summary = (
@@ -1789,7 +1985,9 @@ class TaskCoordinator:
             evidence_refs,
             errors,
             errors,
-            "Planner review required" if result.verdict != ValidationVerdict.PASSED else "Accept",
+            "Planner review required"
+            if result.verdict != ValidationVerdict.PASSED
+            else "Accept",
         )
         return ValidatorRunResult(
             trace_id=validation.validator_trace_id,
@@ -1814,7 +2012,9 @@ class TaskCoordinator:
                     operation.status != OperationStatus.RUNNING
                     or operation.execution_epoch != execution_epoch
                 ):
-                    raise TaskConflict("Execution epoch is stale or operation is stopping")
+                    raise TaskConflict(
+                        "Execution epoch is stale or operation is stopping"
+                    )
             if attempt_id:
                 stage = ledger.attempts[attempt_id]
             elif validation_id:
@@ -1890,13 +2090,22 @@ class TaskCoordinator:
             raise OrchestrationError("Attempt does not belong to task")
         if attempt.worker_trace_id != actor_context.get("trace_id"):
             raise OrchestrationError("Worker trace does not own this attempt")
-        if attempt.spec_version != int(actor_context.get("spec_version")):
-            raise OrchestrationError("Worker submitted against the wrong TaskSpec version")
+        actor_spec_version = actor_context.get("spec_version")
+        if isinstance(actor_spec_version, bool) or not isinstance(
+            actor_spec_version, int
+        ):
+            raise OrchestrationError("Worker context has no valid TaskSpec version")
+        if attempt.spec_version != actor_spec_version:
+            raise OrchestrationError(
+                "Worker submitted against the wrong TaskSpec version"
+            )
         self._assert_execution_owner(ledger, attempt, actor_context)
         if attempt.status != AttemptStatus.RUNNING or task.status != TaskStatus.RUNNING:
             raise TaskConflict("Attempt can only be submitted once while running")
 
-        snapshot = await self.artifact_store.freeze(root_trace_id, attempt_id, submission)
+        snapshot = await self.artifact_store.freeze(
+            root_trace_id, attempt_id, submission
+        )
 
         def mutate(current: TaskLedger) -> Dict[str, Any]:
             current_task = _task(current, task_id)
@@ -1953,7 +2162,10 @@ class TaskCoordinator:
                     )
             task = _task(ledger, task_id)
             attempt = ledger.attempts[attempt_id]
-            if task.status != TaskStatus.AWAITING_VALIDATION or attempt.status != AttemptStatus.SUBMITTED:
+            if (
+                task.status != TaskStatus.AWAITING_VALIDATION
+                or attempt.status != AttemptStatus.SUBMITTED
+            ):
                 raise TaskConflict("Validation requires a submitted attempt")
             validation = ValidationReport(
                 validation_id=new_id(),
@@ -2020,7 +2232,9 @@ class TaskCoordinator:
             if plan.validator_preset is not None and not plan.validator_preset.strip():
                 raise ValueError("Agent validator_preset cannot be blank")
             if not plan.validator_preset:
-                plan = replace(plan, validator_preset=self.config.default_validator_preset)
+                plan = replace(
+                    plan, validator_preset=self.config.default_validator_preset
+                )
         if plan.mode == ValidationMode.DETERMINISTIC:
             criteria = {
                 item.criterion_id: item
@@ -2082,14 +2296,21 @@ class TaskCoordinator:
             return replay
         task = _task(ledger, task_id)
         validation = ledger.validations.get(validation_id)
-        if not validation or validation.task_id != task_id or validation.attempt_id != attempt_id:
+        if (
+            not validation
+            or validation.task_id != task_id
+            or validation.attempt_id != attempt_id
+        ):
             raise OrchestrationError("Validation identity does not match task/attempt")
         if validation.validator_trace_id != actor_context.get("trace_id"):
             raise OrchestrationError("Validator trace does not own this validation")
         if validation.snapshot_id != snapshot_id:
             raise OrchestrationError("Validator must evaluate the fixed snapshot")
         self._assert_execution_owner(ledger, validation, actor_context)
-        if validation.status != ValidationRunStatus.RUNNING or task.status != TaskStatus.VALIDATING:
+        if (
+            validation.status != ValidationRunStatus.RUNNING
+            or task.status != TaskStatus.VALIDATING
+        ):
             raise TaskConflict("Validation can only be submitted once while running")
         self._validate_report(task, verdict, criterion_results)
 
@@ -2196,7 +2417,9 @@ class TaskCoordinator:
         validation = self._owned_evidence_validation(ledger, actor_context, request)
         self._assert_execution_owner(ledger, validation, actor_context)
         if validation.status != ValidationRunStatus.RUNNING:
-            raise TaskConflict("Evidence can only be queried during a running validation")
+            raise TaskConflict(
+                "Evidence can only be queried during a running validation"
+            )
         await self._validate_evidence_snapshot(request)
         return tool_call_id, validation
 
@@ -2213,7 +2436,9 @@ class TaskCoordinator:
             "validation_id": request.validation_id,
         }
         if any(actor_context.get(key) != value for key, value in expected.items()):
-            raise EvidenceOwnershipError("Evidence request does not match Validator scope")
+            raise EvidenceOwnershipError(
+                "Evidence request does not match Validator scope"
+            )
 
     @staticmethod
     def _owned_evidence_validation(
@@ -2240,7 +2465,9 @@ class TaskCoordinator:
         ):
             raise EvidenceOwnershipError("Evidence snapshot ownership mismatch")
         if validation.validator_trace_id != actor_context.get("trace_id"):
-            raise EvidenceOwnershipError("Validator trace does not own this evidence scope")
+            raise EvidenceOwnershipError(
+                "Validator trace does not own this evidence scope"
+            )
         return validation
 
     async def _validate_evidence_snapshot(self, request: EvidenceQuery) -> None:
@@ -2311,7 +2538,9 @@ class TaskCoordinator:
                 timeout=plan.evidence_timeout_seconds,
             )
         except asyncio.TimeoutError as exc:
-            await self._fail_evidence_query(request, query_key, "Evidence provider timed out")
+            await self._fail_evidence_query(
+                request, query_key, "Evidence provider timed out"
+            )
             raise EvidenceProviderError("Evidence provider timed out") from exc
         except asyncio.CancelledError:
             await self._fail_evidence_query(
@@ -2324,10 +2553,10 @@ class TaskCoordinator:
             error = f"Evidence provider failed: {type(exc).__name__}: {exc}"
             await self._fail_evidence_query(request, query_key, error)
             raise EvidenceProviderError(error) from exc
-        error = self._evidence_result_error(result, request.limit)
-        if error:
-            await self._fail_evidence_query(request, query_key, error)
-            raise EvidenceProviderError(error)
+        result_error = self._evidence_result_error(result, request.limit)
+        if result_error:
+            await self._fail_evidence_query(request, query_key, result_error)
+            raise EvidenceProviderError(result_error)
         return result
 
     async def _fail_evidence_query(
@@ -2414,7 +2643,9 @@ class TaskCoordinator:
                     queries_remaining=int(record["queries_remaining"]),
                 )
             if status == "error":
-                raise EvidenceProviderError(str(record.get("error", "Evidence query failed")))
+                raise EvidenceProviderError(
+                    str(record.get("error", "Evidence query failed"))
+                )
             if asyncio.get_running_loop().time() >= deadline:
                 raise EvidenceProviderError("Evidence query is already in progress")
             await asyncio.sleep(0.01)
@@ -2432,7 +2663,9 @@ class TaskCoordinator:
         actual = {c.criterion_id: c for c in criterion_results}
         missing = set(expected) - set(actual)
         if missing:
-            raise ValueError(f"Validation report is missing criteria: {sorted(missing)}")
+            raise ValueError(
+                f"Validation report is missing criteria: {sorted(missing)}"
+            )
         unknown = set(actual) - set(expected)
         if unknown:
             raise ValueError(
@@ -2440,11 +2673,14 @@ class TaskCoordinator:
             )
         if verdict == ValidationVerdict.PASSED:
             non_passed = [
-                cid for cid, criterion in expected.items()
+                cid
+                for cid, criterion in expected.items()
                 if criterion.hard and actual[cid].verdict != ValidationVerdict.PASSED
             ]
             if non_passed:
-                raise ValueError(f"Overall passed conflicts with hard criteria: {non_passed}")
+                raise ValueError(
+                    f"Overall passed conflicts with hard criteria: {non_passed}"
+                )
 
     @staticmethod
     def _assert_execution_owner(
@@ -2467,7 +2703,9 @@ class TaskCoordinator:
             or actor_epoch != stage.execution_epoch
             or actor_epoch != operation.execution_epoch
         ):
-            raise TaskConflict("Agent execution epoch is stale or operation is stopping")
+            raise TaskConflict(
+                "Agent execution epoch is stale or operation is stopping"
+            )
 
     async def decide_task(
         self,
@@ -2567,16 +2805,27 @@ class TaskCoordinator:
             attempt = ledger.attempts.get(attempt_id)
             if not attempt or attempt.task_id != task_id or not attempt.snapshot_id:
                 raise OrchestrationError("Attempt has no immutable snapshot")
-            prior = [ledger.validations[x] for x in task.validation_ids if ledger.validations[x].attempt_id == attempt_id]
+            prior = [
+                ledger.validations[x]
+                for x in task.validation_ids
+                if ledger.validations[x].attempt_id == attempt_id
+            ]
             if not prior or prior[-1].status not in {
-                ValidationRunStatus.ERROR, ValidationRunStatus.STOPPED, ValidationRunStatus.EXPIRED
+                ValidationRunStatus.ERROR,
+                ValidationRunStatus.STOPPED,
+                ValidationRunStatus.EXPIRED,
             }:
-                raise TaskConflict("Revalidation is only allowed after validator error/stopped/expired")
+                raise TaskConflict(
+                    "Revalidation is only allowed after validator error/stopped/expired"
+                )
             if task.status != TaskStatus.NEEDS_REPLAN:
                 raise TaskConflict("Task must be in needs_replan before revalidation")
             report = ValidationReport(
-                validation_id=new_id(), task_id=task_id, attempt_id=attempt_id,
-                spec_version=attempt.spec_version, snapshot_id=attempt.snapshot_id,
+                validation_id=new_id(),
+                task_id=task_id,
+                attempt_id=attempt_id,
+                spec_version=attempt.spec_version,
+                snapshot_id=attempt.snapshot_id,
                 validator_trace_id=str(uuid4()),
                 validator_preset=plan.validator_preset or "deterministic",
                 validation_plan=plan,
@@ -2629,17 +2878,34 @@ class TaskCoordinator:
             raise OrchestrationError("Prior attempt does not belong to task")
         if attempt.spec_version != task.current_spec_version:
             raise TaskConflict("Cannot continue a worker across TaskSpec revisions")
-        if not task.decision_ids or ledger.decisions[task.decision_ids[-1]].action != DecisionAction.REPAIR:
-            raise TaskConflict("Worker continuation requires the latest decision to be repair")
-        if task.repair_count_by_version.get(str(task.current_spec_version), 0) > self.config.max_repair_continuations:
+        if (
+            not task.decision_ids
+            or ledger.decisions[task.decision_ids[-1]].action != DecisionAction.REPAIR
+        ):
+            raise TaskConflict(
+                "Worker continuation requires the latest decision to be repair"
+            )
+        if (
+            task.repair_count_by_version.get(str(task.current_spec_version), 0)
+            > self.config.max_repair_continuations
+        ):
             raise TaskConflict("Repair continuation limit exceeded")
-        trace = await self.trace_store.get_trace(attempt.worker_trace_id) if self.trace_store else None
+        trace = (
+            await self.trace_store.get_trace(attempt.worker_trace_id)
+            if self.trace_store
+            else None
+        )
         if not trace or trace.agent_role != AgentRole.WORKER.value:
             raise TaskConflict("Prior worker trace is missing or has the wrong role")
         if trace.status not in ("completed", "stopped"):
-            raise TaskConflict("Prior worker trace is not in a recoverable terminal state")
+            raise TaskConflict(
+                "Prior worker trace is not in a recoverable terminal state"
+            )
         context = trace.context or {}
-        if context.get("root_trace_id") != root_trace_id or context.get("task_id") != task_id:
+        if (
+            context.get("root_trace_id") != root_trace_id
+            or context.get("task_id") != task_id
+        ):
             raise TaskConflict("Prior worker trace identity is invalid")
         if int(context.get("spec_version", -1)) != task.current_spec_version:
             raise TaskConflict("Prior worker trace TaskSpec version is invalid")
@@ -2658,7 +2924,10 @@ class TaskCoordinator:
         operation_id: Optional[str],
         execution_epoch: int,
     ) -> bool:
-        if stage.operation_id != operation_id or stage.execution_epoch != execution_epoch:
+        if (
+            stage.operation_id != operation_id
+            or stage.execution_epoch != execution_epoch
+        ):
             return False
         if operation_id is None:
             return True
@@ -2693,11 +2962,8 @@ class TaskCoordinator:
             if isinstance(stage, TaskAttempt)
             else stage.status == ValidationRunStatus.COMPLETED
         )
-        if (
-            not terminal
-            or not self._stage_execution_is_current(
-                ledger, stage, operation_id, execution_epoch
-            )
+        if not terminal or not self._stage_execution_is_current(
+            ledger, stage, operation_id, execution_epoch
         ):
             return False
         if stage.execution_stats is not None:
@@ -2716,11 +2982,8 @@ class TaskCoordinator:
                 if isinstance(current_stage, TaskAttempt)
                 else current_stage.status == ValidationRunStatus.COMPLETED
             )
-            if (
-                not current_terminal
-                or not self._stage_execution_is_current(
-                    current, current_stage, operation_id, execution_epoch
-                )
+            if not current_terminal or not self._stage_execution_is_current(
+                current, current_stage, operation_id, execution_epoch
             ):
                 return {
                     "recorded": False,
@@ -2775,7 +3038,10 @@ class TaskCoordinator:
                 )
             ):
                 return {"task_id": task_id, "attempt_id": attempt_id, "ignored": True}
-            status_map = {"stopped": AttemptStatus.STOPPED, "expired": AttemptStatus.EXPIRED}
+            status_map = {
+                "stopped": AttemptStatus.STOPPED,
+                "expired": AttemptStatus.EXPIRED,
+            }
             attempt.status = status_map.get(run_status, AttemptStatus.FAILED)
             attempt.error = error
             attempt.failure = failure
@@ -2973,7 +3239,9 @@ class TaskCoordinator:
     def _update_parent_after_child(ledger: TaskLedger, child: TaskRecord) -> None:
         TaskGraph.update_parent_after_child(ledger, child)
 
-    async def _tasks_result(self, root_trace_id: str, task_ids: Iterable[str]) -> Dict[str, Any]:
+    async def _tasks_result(
+        self, root_trace_id: str, task_ids: Iterable[str]
+    ) -> Dict[str, Any]:
         ledger = await self.task_store.load(root_trace_id)
         return {
             "tasks": [
@@ -2990,7 +3258,9 @@ class TaskCoordinator:
         }
 
     @staticmethod
-    def _display_path(ledger: TaskLedger, parent_task_id: Optional[str], index: int) -> str:
+    def _display_path(
+        ledger: TaskLedger, parent_task_id: Optional[str], index: int
+    ) -> str:
         return TaskGraph.display_path(ledger, parent_task_id, index)
 
     @staticmethod
@@ -3019,7 +3289,9 @@ def _required(context: Dict[str, Any], key: str) -> str:
 def _require_actor(context: Dict[str, Any], expected: AgentRole) -> None:
     actual = context.get("role")
     if actual != expected.value:
-        raise OrchestrationError(f"Expected actor role {expected.value}, got {actual!r}")
+        raise OrchestrationError(
+            f"Expected actor role {expected.value}, got {actual!r}"
+        )
 
 
 def _plain(value: Any) -> Any:

agent/agent/orchestration/deterministic_worker.py

@@ -0,0 +1,42 @@
+"""Generic deterministic Worker extension point."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping, Protocol
+
+from .models import ArtifactRef, TaskAttempt, TaskRecord
+
+
+@dataclass(frozen=True, slots=True)
+class DeterministicWorkerContext:
+    root_trace_id: str
+    task: TaskRecord
+    attempt: TaskAttempt
+    ledger_revision: int
+    accepted_child_results: tuple[Mapping[str, Any], ...]
+    operation_id: str | None
+    execution_epoch: int
+    role_context: Mapping[str, Any]
+
+
+@dataclass(frozen=True, slots=True)
+class DeterministicWorkerResult:
+    summary: str
+    artifact_refs: tuple[ArtifactRef, ...]
+    evidence_refs: tuple[ArtifactRef, ...] = ()
+
+
+class DeterministicWorker(Protocol):
+    async def supports(self, context: DeterministicWorkerContext) -> bool: ...
+
+    async def execute(
+        self, context: DeterministicWorkerContext
+    ) -> DeterministicWorkerResult: ...
+
+
+__all__ = [
+    "DeterministicWorker",
+    "DeterministicWorkerContext",
+    "DeterministicWorkerResult",
+]

agent/agent/orchestration/executor.py

@@ -75,7 +75,12 @@ class LocalAgentExecutor:
             "attempt_id": context["attempt_id"],
             "repair_feedback": context.get("repair_feedback"),
             "prior_feedback": context.get("prior_feedback"),
-            "accepted_child_results": context.get("accepted_child_results", []),
+            "accepted_child_results": (
+                []
+                if context.get("role_context")
+                else context.get("accepted_child_results", [])
+            ),
+            "role_context": context.get("role_context", {}),
             "instruction": "Execute this TaskSpec and finish with submit_attempt.",
         }
         result = await self._run_role(
@@ -116,6 +121,7 @@ class LocalAgentExecutor:
             "attempt_id": context["attempt_id"],
             "artifact_snapshot": context["artifact_snapshot"],
             "validation_id": context["validation_id"],
+            "role_context": context.get("role_context", {}),
             "instruction": "Independently verify every criterion and finish with submit_validation.",
         }
         result = await self._run_role(

agent/agent/orchestration/wiring.py

@@ -11,6 +11,8 @@ from .executor import LocalAgentExecutor
 from .protocols import ArtifactStore, EventSink, TaskStore
 from .run_config import RoleRunConfigResolver, RoleSystemPromptResolver
 from .validation_policy import DeterministicValidator, ValidationPolicy
+from .context_provider import RoleContextProvider, TaskContextProvider
+from .deterministic_worker import DeterministicWorker
 
 
 def wire_orchestration(
@@ -25,6 +27,9 @@ def wire_orchestration(
     evidence_provider: Optional[EvidenceProvider] = None,
     role_run_config_resolver: Optional[RoleRunConfigResolver] = None,
     role_system_prompt_resolver: Optional[RoleSystemPromptResolver] = None,
+    task_context_provider: Optional[TaskContextProvider] = None,
+    role_context_provider: Optional[RoleContextProvider] = None,
+    deterministic_worker: Optional[DeterministicWorker] = None,
 ) -> TaskCoordinator:
     """Wire generic orchestration ports without importing a business project.
 
@@ -41,6 +46,9 @@ def wire_orchestration(
         validation_policy=validation_policy,
         deterministic_validator=deterministic_validator,
         evidence_provider=evidence_provider,
+        task_context_provider=task_context_provider,
+        role_context_provider=role_context_provider,
+        deterministic_worker=deterministic_worker,
     )
     executor = LocalAgentExecutor(
         runner,

agent/tests/test_context_budget.py

@@ -2,6 +2,7 @@ import math
 
 import pytest
 
+from agent.core.prompts import build_summary_header
 from agent.core.runner import AgentRunner, RunConfig
 from agent.trace.compaction import (
     CompressionConfig,
@@ -25,14 +26,16 @@ def _knowledge_off():
 def test_prompt_measurement_includes_tools_and_provider_calibration():
     config = CompressionConfig(max_tokens=100_000)
     messages = [{"role": "user", "content": "x" * 4_000}]
-    schemas = [{
-        "type": "function",
-        "function": {
-            "name": "large_tool",
-            "description": "y" * 4_000,
-            "parameters": {"type": "object"},
-        },
-    }]
+    schemas = [
+        {
+            "type": "function",
+            "function": {
+                "name": "large_tool",
+                "description": "y" * 4_000,
+                "parameters": {"type": "object"},
+            },
+        }
+    ]
 
     raw = measure_prompt_tokens(messages, [], config, "unknown-model", 1.0)
     measured = measure_prompt_tokens(messages, schemas, config, "unknown-model", 1.0)
@@ -68,14 +71,16 @@ async def test_build_900036_growth_replay_compresses_before_hard_limit(tmp_path)
         compression=CompressionConfig(max_tokens=100_000),
         knowledge=_knowledge_off(),
     )
-    schemas = [{
-        "type": "function",
-        "function": {
-            "name": "offline_step",
-            "description": "append one deterministic fixture step",
-            "parameters": {"type": "object", "properties": {}},
-        },
-    }]
+    schemas = [
+        {
+            "type": "function",
+            "function": {
+                "name": "offline_step",
+                "description": "append one deterministic fixture step",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        }
+    ]
     history = [
         {"role": "system", "content": "keep system policy"},
         {"role": "user", "content": "offline 97-round context fixture"},
@@ -84,10 +89,12 @@ async def test_build_900036_growth_replay_compresses_before_hard_limit(tmp_path)
     max_prompt_tokens = 0
 
     for turn in range(97):
-        history.extend([
-            {"role": "assistant", "content": f"turn {turn}", "tool_calls": []},
-            {"role": "tool", "content": "x" * 4_000},
-        ])
+        history.extend(
+            [
+                {"role": "assistant", "content": f"turn {turn}", "tool_calls": []},
+                {"role": "tool", "content": "x" * 4_000},
+            ]
+        )
         history, _, _, needs_compression = await runner._manage_context_usage(
             trace.trace_id,
             history,
@@ -193,3 +200,12 @@ async def test_oversized_minimum_context_fails_before_model_request(tmp_path):
 def test_compression_config_rejects_unsafe_budgets(kwargs):
     with pytest.raises(ValueError):
         CompressionConfig(**kwargs)
+
+
+def test_compression_summary_forbids_repeating_successful_immutable_reads():
+    summary = build_summary_header(
+        "The accepted artifact was read; next save the candidate."
+    )
+
+    assert "不要因为原文已压缩而用相同参数重复读取" in summary
+    assert "直接执行摘要所列的下一个" in summary

agent/tests/test_governance_public_contract.py

@@ -52,6 +52,9 @@ def test_runner_and_coordinator_public_call_shapes_are_frozen() -> None:
         "validation_policy",
         "deterministic_validator",
         "evidence_provider",
+        "task_context_provider",
+        "role_context_provider",
+        "deterministic_worker",
     )
     assert _parameter_names(TaskCoordinator.decide_task) == (
         "self",

agent/tests/test_no_progress_policy.py

@@ -58,19 +58,32 @@ def _explicit_config(preset, *, max_iterations=8, tools=None):
     )
 
 
+def _epoch_config(preset, epoch, *, max_iterations):
+    config = _explicit_config(preset, max_iterations=max_iterations)
+    config.no_progress = NoProgressPolicy(progress_epoch=epoch)
+    config._role_run_config_override_fields = frozenset({"max_iterations"})
+    return config
+
+
 @pytest.mark.asyncio
 async def test_planner_stops_after_three_unchanged_semantic_cycles(tmp_path):
     preset = "test_no_progress_planner_stall"
     register_preset(
         preset,
-        AgentPreset(role=AgentRole.PLANNER, allowed_tools=[], max_iterations=8, skills=[]),
+        AgentPreset(
+            role=AgentRole.PLANNER, allowed_tools=[], max_iterations=8, skills=[]
+        ),
     )
     calls = 0
 
     async def llm_call(**_kwargs):
         nonlocal calls
         calls += 1
-        return {"content": "still thinking", "tool_calls": None, "finish_reason": "stop"}
+        return {
+            "content": "still thinking",
+            "tool_calls": None,
+            "finish_reason": "stop",
+        }
 
     store = FileSystemTraceStore(str(tmp_path))
     result = await AgentRunner(
@@ -121,11 +134,13 @@ async def test_second_identical_tool_failure_stops_worker(tmp_path):
         calls += 1
         return {
             "content": "",
-            "tool_calls": [{
-                "id": f"call-{calls}",
-                "type": "function",
-                "function": {"name": "flaky_write", "arguments": "{}"},
-            }],
+            "tool_calls": [
+                {
+                    "id": f"call-{calls}",
+                    "type": "function",
+                    "function": {"name": "flaky_write", "arguments": "{}"},
+                }
+            ],
             "finish_reason": "tool_calls",
         }
 
@@ -145,12 +160,74 @@ async def test_second_identical_tool_failure_stops_worker(tmp_path):
     assert result["failure"]["details"]["last_failure"]["code"] == "WORKSPACE_NOT_READY"
 
 
+@pytest.mark.asyncio
+async def test_distinct_planner_failures_count_as_semantic_feedback_progress(tmp_path):
+    preset = "test_no_progress_distinct_planner_failures"
+    register_preset(
+        preset,
+        AgentPreset(
+            role=AgentRole.PLANNER,
+            allowed_tools=["reject_plan"],
+            max_iterations=4,
+            skills=[],
+        ),
+    )
+    registry = ToolRegistry()
+
+    async def reject_plan(attempted_kind: str):
+        raise ToolExecutionError(
+            FailureDetail(
+                code="PHASE_POLICY_VIOLATION",
+                message=f"cannot plan {attempted_kind}",
+                disposition=FailureDisposition.REPLAN_TASK,
+                source_tool="reject_plan",
+                details={"attempted_task_kinds": [attempted_kind]},
+            )
+        )
+
+    registry.register(reject_plan, capabilities=[ToolCapability.TASK_CONTROL])
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        attempted = ["structure", "paragraph", "element-set", "compare"][calls - 1]
+        return {
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": f"call-{calls}",
+                    "type": "function",
+                    "function": {
+                        "name": "reject_plan",
+                        "arguments": f'{{"attempted_kind":"{attempted}"}}',
+                    },
+                }
+            ],
+            "finish_reason": "tool_calls",
+        }
+
+    await AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        tool_registry=registry,
+        llm_call=llm_call,
+        task_coordinator=StaticProgressCoordinator(),
+    ).run_result(
+        [{"role": "user", "content": "plan"}],
+        _explicit_config(preset, max_iterations=4, tools=["reject_plan"]),
+    )
+
+    assert calls == 4
+
+
 @pytest.mark.asyncio
 async def test_planner_stall_counter_survives_resume(tmp_path):
     preset = "test_no_progress_resume"
     register_preset(
         preset,
-        AgentPreset(role=AgentRole.PLANNER, allowed_tools=[], max_iterations=4, skills=[]),
+        AgentPreset(
+            role=AgentRole.PLANNER, allowed_tools=[], max_iterations=4, skills=[]
+        ),
     )
     calls = 0
 
@@ -181,6 +258,49 @@ async def test_planner_stall_counter_survives_resume(tmp_path):
     assert resumed["failure"]["code"] == "NO_PROGRESS"
 
 
+@pytest.mark.asyncio
+async def test_planner_stall_counter_resets_for_new_progress_epoch(tmp_path):
+    preset = "test_no_progress_new_epoch"
+    register_preset(
+        preset,
+        AgentPreset(
+            role=AgentRole.PLANNER, allowed_tools=[], max_iterations=4, skills=[]
+        ),
+    )
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {"content": "", "tool_calls": None, "finish_reason": "stop"}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    runner = AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+        task_coordinator=StaticProgressCoordinator(),
+    )
+    first = await runner.run_result(
+        [{"role": "user", "content": "phase one"}],
+        _epoch_config(preset, "phase-one", max_iterations=2),
+    )
+    assert first["status"] == "incomplete"
+    assert calls == 2
+
+    resumed = await runner.run_result(
+        [{"role": "user", "content": "phase two"}],
+        RunConfig(
+            trace_id=first["trace_id"],
+            no_progress=NoProgressPolicy(progress_epoch="phase-two"),
+            knowledge=_knowledge_off(),
+        ),
+    )
+
+    assert calls == 5
+    assert resumed["status"] == "incomplete"
+    assert resumed["failure"]["code"] == "NO_PROGRESS"
+
+
 @pytest.mark.asyncio
 async def test_semantically_equal_task_ledgers_ignore_generated_ids(tmp_path):
     first, _, _ = await make_coordinator(tmp_path / "first", FakeExecutor([]))
@@ -189,7 +309,9 @@ async def test_semantically_equal_task_ledgers_ignore_generated_ids(tmp_path):
     second_task_id = await create_task(second, objective="equivalent child")
 
     assert first_task_id != second_task_id
-    assert await first.progress_snapshot("root") == await second.progress_snapshot("root")
+    assert await first.progress_snapshot("root") == await second.progress_snapshot(
+        "root"
+    )
 
 
 def test_no_progress_policy_rejects_invalid_thresholds():

agent/tests/test_orchestration_workbench_ports.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+from agent.orchestration import (
+    AgentRole,
+    ArtifactRef,
+    CriterionResult,
+    DeterministicWorkerContext,
+    DeterministicWorkerResult,
+    OrchestrationConfig,
+    RoleContextRequest,
+    TaskCoordinator,
+    ValidationVerdict,
+)
+from agent.orchestration.protocols import ValidatorRunResult
+from agent.orchestration.store import FileSystemArtifactStore, FileSystemTaskStore
+from agent.trace.models import Trace
+from agent.trace.store import FileSystemTraceStore
+import pytest
+
+
+class _TaskContext:
+    async def render(self, root_trace_id: str, ledger: object) -> str:
+        assert root_trace_id == "root"
+        return '{"state_revision":"ledger:compact"}'
+
+
+class _RoleContext:
+    def __init__(self) -> None:
+        self.requests: list[RoleContextRequest] = []
+
+    async def build(self, request: RoleContextRequest) -> dict[str, object]:
+        self.requests.append(request)
+        return {
+            "state_revision": f"ledger:{request.ledger_revision}",
+            "role": request.role.value,
+        }
+
+
+class _DeterministicWorker:
+    def __init__(self) -> None:
+        self.calls: list[DeterministicWorkerContext] = []
+
+    async def supports(self, context: DeterministicWorkerContext) -> bool:
+        return True
+
+    async def execute(
+        self, context: DeterministicWorkerContext
+    ) -> DeterministicWorkerResult:
+        self.calls.append(context)
+        return DeterministicWorkerResult(
+            summary="Host assembled the immutable artifact",
+            artifact_refs=(ArtifactRef(uri="memory://deterministic", version="1"),),
+        )
+
+
+class _Executor:
+    def __init__(self) -> None:
+        self.coordinator: TaskCoordinator | None = None
+        self.worker_calls = 0
+        self.validator_calls = 0
+
+    async def run_worker(self, context: dict[str, object]) -> object:
+        self.worker_calls += 1
+        raise AssertionError(
+            "LLM Worker must not run when deterministic execution supports Task"
+        )
+
+    async def run_validator(self, context: dict[str, object]) -> ValidatorRunResult:
+        self.validator_calls += 1
+        assert context["role_context"]
+        coordinator = self.coordinator
+        assert coordinator is not None
+        task_spec = context["task_spec"]
+        assert isinstance(task_spec, dict)
+        criteria = [
+            CriterionResult(
+                criterion_id=str(item["criterion_id"]),
+                verdict=ValidationVerdict.PASSED,
+                reason="independently checked",
+            )
+            for item in task_spec["acceptance_criteria"]
+        ]
+        await coordinator.submit_validation(
+            {
+                **context,
+                "role": AgentRole.VALIDATOR.value,
+                "trace_id": context["validator_trace_id"],
+                "tool_call_id": "validator-submit",
+            },
+            ValidationVerdict.PASSED,
+            criteria,
+            "passed",
+            [],
+            [],
+            [],
+            "accept",
+        )
+        return ValidatorRunResult(str(context["validator_trace_id"]), "completed")
+
+
+@pytest.mark.asyncio
+async def test_context_ports_and_deterministic_worker_preserve_full_audit_chain(
+    tmp_path,
+) -> None:
+    trace_store = FileSystemTraceStore(str(tmp_path / "traces"))
+    await trace_store.create_trace(
+        Trace(trace_id="root", mode="agent", task="mission", agent_role="planner")
+    )
+    role_context = _RoleContext()
+    deterministic = _DeterministicWorker()
+    executor = _Executor()
+    task_store = FileSystemTaskStore(str(tmp_path / "ledger"))
+    coordinator = TaskCoordinator(
+        task_store,
+        FileSystemArtifactStore(str(tmp_path / "artifacts")),
+        trace_store,
+        OrchestrationConfig(),
+        executor=executor,
+        task_context_provider=_TaskContext(),
+        role_context_provider=role_context,
+        deterministic_worker=deterministic,
+    )
+    executor.coordinator = coordinator
+    await coordinator.ensure_ledger(
+        "root",
+        {
+            "objective": "mission",
+            "acceptance_criteria": [
+                {"criterion_id": "root", "description": "complete", "hard": True}
+            ],
+        },
+    )
+    created = await coordinator.create_tasks(
+        "root",
+        [
+            {
+                "objective": "mechanical assembly",
+                "acceptance_criteria": [
+                    {"criterion_id": "closed", "description": "closed", "hard": True}
+                ],
+            }
+        ],
+    )
+    task_id = created["tasks"][0]["task_id"]
+
+    cycle = (await coordinator.dispatch_tasks("root", [task_id]))[0]
+
+    assert cycle.validation is not None
+    assert cycle.validation.verdict is ValidationVerdict.PASSED
+    assert executor.worker_calls == 0
+    assert executor.validator_calls == 1
+    assert len(deterministic.calls) == 1
+    assert [item.role for item in role_context.requests] == [
+        AgentRole.WORKER,
+        AgentRole.VALIDATOR,
+    ]
+    assert (
+        await coordinator.task_context("root") == '{"state_revision":"ledger:compact"}'
+    )
+    ledger = await task_store.load("root")
+    attempt = ledger.attempts[cycle.attempt_id]
+    assert attempt.submission is not None
+    worker_trace = await trace_store.get_trace(attempt.worker_trace_id)
+    assert worker_trace is not None
+    assert worker_trace.status == "completed"
+    assert worker_trace.total_tokens == 0
+    assert worker_trace.context["deterministic_worker"] is True
+    assert await trace_store.get_trace_messages(attempt.worker_trace_id) == []