框架：增加角色运行配置解析器并放宽派发工具合同

新增全部可选的角色运行配置覆盖接口，只允许调整模型、温度和最大轮数。

保持旧编排装配调用兼容，保护运行身份、工具与完成策略，并补充解析器异常和深层只读上下文回归测试。

agent/agent/__init__.py

@@ -15,7 +15,16 @@ from agent.core.runner import AgentRunner, CallResult, RunConfig
 from agent.core.presets import AgentPreset, AGENT_PRESETS, get_preset
 
 # 执行追踪
-from agent.trace.models import Trace, Message, Step, StepType, StepStatus, ChatMessage, Messages, MessageContent
+from agent.trace.models import (
+    Trace,
+    Message,
+    Step,
+    StepType,
+    StepStatus,
+    ChatMessage,
+    Messages,
+    MessageContent,
+)
 from agent.trace.goal_models import Goal, GoalTree, GoalStatus
 from agent.trace.protocols import TraceStore
 from agent.trace.store import FileSystemTraceStore
@@ -49,6 +58,8 @@ from agent.orchestration import (
     ArtifactStore,
     FileSystemArtifactStore,
     OrchestrationConfig,
+    RoleRunConfigOverrides,
+    RoleRunConfigResolver,
 )
 from agent.orchestration.wiring import wire_orchestration
 
@@ -109,6 +120,8 @@ __all__ = [
     "ArtifactStore",
     "FileSystemArtifactStore",
     "OrchestrationConfig",
+    "RoleRunConfigOverrides",
+    "RoleRunConfigResolver",
     "wire_orchestration",
     # SDK
     "invoke_agent",

agent/agent/core/prompts/orchestration.py

@@ -3,8 +3,9 @@
 PLANNER_ROLE_CONTRACT = """
 ## Framework role contract: Planner
 
-You are the only planner. Create and dispatch tasks with task_plan and
-dispatch_tasks. Worker completion is only an attempt submission. A task is
+You are the only planner. Create tasks with task_plan and dispatch them with
+the authorized task-dispatch tool exposed by your preset. Worker completion
+is only an attempt submission. A task is
 complete only after an independent passed validation and your task_decide
 action=accept. For failed, inconclusive, or validation errors, explicitly
 repair, retry, revise, split, revalidate, block, cancel, or supersede.
@@ -39,6 +40,8 @@ ROLE_CONTRACTS = {
 }
 
 __all__ = [
-    "PLANNER_ROLE_CONTRACT", "WORKER_ROLE_CONTRACT",
-    "VALIDATOR_ROLE_CONTRACT", "ROLE_CONTRACTS",
+    "PLANNER_ROLE_CONTRACT",
+    "WORKER_ROLE_CONTRACT",
+    "VALIDATOR_ROLE_CONTRACT",
+    "ROLE_CONTRACTS",
 ]

agent/agent/core/runner.py

@@ -148,6 +148,13 @@ class RunConfig:
     # --- 自定义元数据上下文 ---
     context: Dict[str, Any] = field(default_factory=dict)
 
+    # Trusted framework-only marker used by LocalAgentExecutor.  Ordinary
+    # presets keep their historical temperature/max-iteration precedence.
+    _role_run_config_override_fields: frozenset[str] = field(
+        default_factory=frozenset,
+        repr=False,
+    )
+
     # --- 研究流程控制 ---
     enable_research_flow: bool = True  # 是否启用自动研究流程（知识检索→经验检索→调研→计划）
     # --- 知识管理配置 ---

agent/agent/orchestration/__init__.py

@@ -48,6 +48,7 @@ from .models import (
     ValidationVerdict,
 )
 from .protocols import AgentExecutor, ArtifactStore, EventSink, TaskStore
+from .run_config import RoleRunConfigOverrides, RoleRunConfigResolver
 from .state_machine import InvalidTaskTransition
 from .store import (
     ArtifactConflict,
@@ -128,6 +129,8 @@ __all__ = [
     "PlannerDecisionView",
     "ProblemDetails",
     "RevisionConflict",
+    "RoleRunConfigOverrides",
+    "RoleRunConfigResolver",
     "RootCompletionView",
     "RuleBasedDeterministicValidator",
     "TaskAttempt",

agent/agent/orchestration/executor.py

@@ -3,21 +3,31 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Mapping
+from copy import deepcopy
+from dataclasses import replace
 from math import isfinite
+from types import MappingProxyType
 from typing import Any, Dict, Iterable, Optional, Tuple
 
 from agent.core.runner import RunConfig
 from agent.tools.builtin.knowledge import KnowledgeConfig
 
-from .models import CompletionPolicy, ExecutionStats, FailureCode
+from .models import AgentRole, CompletionPolicy, ExecutionStats, FailureCode
 from .protocols import ValidatorRunResult, WorkerRunResult
+from .run_config import RoleRunConfigOverrides, RoleRunConfigResolver
 
 
 class LocalAgentExecutor:
     """Run workers and validators with the same Runner in isolated traces."""
 
-    def __init__(self, runner: Any) -> None:
+    def __init__(
+        self,
+        runner: Any,
+        role_run_config_resolver: Optional[RoleRunConfigResolver] = None,
+    ) -> None:
         self.runner = runner
+        self.role_run_config_resolver = role_run_config_resolver
 
     async def stop(self, trace_id: str) -> bool:
         """Request cooperative cancellation of a local sub-trace."""
@@ -40,14 +50,17 @@ class LocalAgentExecutor:
         usage_baseline: Tuple[int, float] = (0, 0.0)
         if continue_trace:
             await self.runner.task_coordinator.validate_continue_from(
-                context["root_trace_id"], context["task_id"],
+                context["root_trace_id"],
+                context["task_id"],
                 self._prior_attempt_id(context),
             )
             trace = await self.runner.trace_store.get_trace(continue_trace)
             usage_baseline = _trace_usage(trace)
             trace_context = dict(trace.context or {})
             trace_context.update(protected)
-            await self.runner.trace_store.update_trace(continue_trace, context=trace_context)
+            await self.runner.trace_store.update_trace(
+                continue_trace, context=trace_context
+            )
 
         prompt = {
             "task_spec": context["task_spec"],
@@ -58,6 +71,7 @@ class LocalAgentExecutor:
         }
         result = await self._run_role(
             context,
+            role=AgentRole.WORKER,
             trace_key="worker_trace_id",
             preset=preset,
             protected=protected,
@@ -97,6 +111,7 @@ class LocalAgentExecutor:
         }
         result = await self._run_role(
             context,
+            role=AgentRole.VALIDATOR,
             trace_key="validator_trace_id",
             preset=preset,
             protected=protected,
@@ -109,6 +124,7 @@ class LocalAgentExecutor:
         self,
         context: Dict[str, Any],
         *,
+        role: AgentRole,
         trace_key: str,
         preset: str,
         protected: Dict[str, Any],
@@ -134,11 +150,19 @@ class LocalAgentExecutor:
             name=name,
         )
         try:
+            config = await self._apply_run_config_overrides(
+                config,
+                role=role,
+                preset=preset,
+                context=context,
+            )
             result = await self.runner.run_result(
-                messages=[{
-                    "role": "user",
-                    "content": json.dumps(prompt, ensure_ascii=False, indent=2),
-                }],
+                messages=[
+                    {
+                        "role": "user",
+                        "content": json.dumps(prompt, ensure_ascii=False, indent=2),
+                    }
+                ],
                 config=config,
             )
             result_trace_id = result.get("trace_id") or trace_id
@@ -171,6 +195,37 @@ class LocalAgentExecutor:
                 ),
             }
 
+    async def _apply_run_config_overrides(
+        self,
+        config: RunConfig,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Dict[str, Any],
+    ) -> RunConfig:
+        resolver = self.role_run_config_resolver
+        if resolver is None:
+            return config
+        overrides = await resolver.resolve(
+            role=role,
+            preset=preset,
+            context=_deep_readonly(deepcopy(context)),
+        )
+        if not isinstance(overrides, RoleRunConfigOverrides):
+            raise TypeError(
+                "RoleRunConfigResolver.resolve() must return RoleRunConfigOverrides"
+            )
+        values = {
+            field: value
+            for field in ("model", "temperature", "max_iterations")
+            if (value := getattr(overrides, field)) is not None
+        }
+        return replace(
+            config,
+            **values,
+            _role_run_config_override_fields=frozenset(values),
+        )
+
     async def _get_trace(self, trace_id: str) -> Any:
         trace_store = getattr(self.runner, "trace_store", None)
         if trace_store is None:
@@ -209,6 +264,20 @@ def _disabled_knowledge() -> KnowledgeConfig:
     )
 
 
+def _deep_readonly(value: Any) -> Any:
+    if isinstance(value, Mapping):
+        return MappingProxyType(
+            {key: _deep_readonly(item) for key, item in value.items()}
+        )
+    if isinstance(value, list):
+        return tuple(_deep_readonly(item) for item in value)
+    if isinstance(value, tuple):
+        return tuple(_deep_readonly(item) for item in value)
+    if isinstance(value, set):
+        return frozenset(_deep_readonly(item) for item in value)
+    return value
+
+
 def _trace_usage(trace: Any) -> Tuple[int, float]:
     if trace is None:
         return 0, 0.0
@@ -227,12 +296,18 @@ def _execution_stats(
     raw = result.get("stats")
     raw = raw if isinstance(raw, dict) else {}
     trace_tokens, trace_cost = _trace_usage(trace)
-    cumulative_tokens = raw.get("total_tokens", trace_tokens if trace is not None else None)
+    cumulative_tokens = raw.get(
+        "total_tokens", trace_tokens if trace is not None else None
+    )
     cumulative_cost = raw.get("total_cost", trace_cost if trace is not None else None)
     parsed_tokens = _nonnegative_int(cumulative_tokens)
     parsed_cost = _nonnegative_float(cumulative_cost)
-    total_tokens = max(0, parsed_tokens - baseline[0]) if parsed_tokens is not None else None
-    total_cost = max(0.0, parsed_cost - baseline[1]) if parsed_cost is not None else None
+    total_tokens = (
+        max(0, parsed_tokens - baseline[0]) if parsed_tokens is not None else None
+    )
+    total_cost = (
+        max(0.0, parsed_cost - baseline[1]) if parsed_cost is not None else None
+    )
     primary_model = (
         raw.get("primary_model")
         or raw.get("model")

agent/agent/orchestration/policy.py

@@ -80,11 +80,27 @@ class DefaultToolPolicy:
                 and declared.issubset(allowed_capabilities)
             }
 
+        trusted_overrides = frozenset(
+            getattr(config, "_role_run_config_override_fields", frozenset())
+        )
+        max_iterations = (
+            int(config.max_iterations)
+            if "max_iterations" in trusted_overrides
+            else min(int(config.max_iterations), int(preset.max_iterations))
+        )
+        temperature = (
+            float(config.temperature)
+            if "temperature" in trusted_overrides
+            else preset.temperature
+            if preset.temperature is not None
+            else float(config.temperature)
+        )
+
         return ResolvedAgentPolicy(
             role=role,
             effective_tools=frozenset(candidates),
-            max_iterations=min(int(config.max_iterations), int(preset.max_iterations)),
-            temperature=preset.temperature if preset.temperature is not None else float(config.temperature),
+            max_iterations=max_iterations,
+            temperature=temperature,
             skills=config.skills if config.skills is not None else preset.skills,
             system_prompt=config.system_prompt or preset.system_prompt,
             completion_policy=policy,

agent/agent/orchestration/run_config.py

@@ -0,0 +1,56 @@
+"""Trusted, role-aware model run configuration extension points."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from math import isfinite
+from typing import Any, Mapping, Optional, Protocol
+
+from .models import AgentRole
+
+
+@dataclass(frozen=True)
+class RoleRunConfigOverrides:
+    """The only ``RunConfig`` fields a role resolver may override."""
+
+    model: Optional[str] = None
+    temperature: Optional[float] = None
+    max_iterations: Optional[int] = None
+
+    def __post_init__(self) -> None:
+        if self.model is not None and (
+            not isinstance(self.model, str) or not self.model.strip()
+        ):
+            raise ValueError("model must be a non-empty string when provided")
+        if self.temperature is not None and (
+            isinstance(self.temperature, bool)
+            or not isinstance(self.temperature, (int, float))
+            or not isfinite(self.temperature)
+            or self.temperature < 0
+        ):
+            raise ValueError(
+                "temperature must be finite and non-negative when provided"
+            )
+        if self.max_iterations is not None and (
+            isinstance(self.max_iterations, bool)
+            or not isinstance(self.max_iterations, int)
+            or self.max_iterations < 1
+        ):
+            raise ValueError("max_iterations must be a positive integer when provided")
+
+
+class RoleRunConfigResolver(Protocol):
+    """Resolve trusted model settings for one isolated Agent role run."""
+
+    async def resolve(
+        self,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Mapping[str, Any],
+    ) -> RoleRunConfigOverrides:
+        """Return model-only overrides for the supplied role execution."""
+        ...
+
+
+__all__ = ["RoleRunConfigOverrides", "RoleRunConfigResolver"]

agent/agent/orchestration/wiring.py

@@ -9,6 +9,7 @@ from .coordinator import TaskCoordinator
 from .evidence import EvidenceProvider
 from .executor import LocalAgentExecutor
 from .protocols import ArtifactStore, EventSink, TaskStore
+from .run_config import RoleRunConfigResolver
 from .validation_policy import DeterministicValidator, ValidationPolicy
 
 
@@ -22,11 +23,13 @@ def wire_orchestration(
     validation_policy: Optional[ValidationPolicy] = None,
     deterministic_validator: Optional[DeterministicValidator] = None,
     evidence_provider: Optional[EvidenceProvider] = None,
+    role_run_config_resolver: Optional[RoleRunConfigResolver] = None,
 ) -> TaskCoordinator:
     """Wire generic orchestration ports without importing a business project.
 
-    Validation and evidence dependencies are optional keyword-only extension
-    points. Existing positional calls keep the default Coordinator behavior.
+    Validation, evidence, and role run configuration dependencies are optional
+    keyword-only extension points. Existing positional calls keep the default
+    Coordinator behavior.
     """
     coordinator = TaskCoordinator(
         task_store=task_store,
@@ -38,7 +41,10 @@ def wire_orchestration(
         deterministic_validator=deterministic_validator,
         evidence_provider=evidence_provider,
     )
-    executor = LocalAgentExecutor(runner)
+    executor = LocalAgentExecutor(
+        runner,
+        role_run_config_resolver=role_run_config_resolver,
+    )
     coordinator.set_executor(executor)
     runner.task_coordinator = coordinator
     return coordinator

agent/tests/test_orchestration_run_config_resolver.py

@@ -0,0 +1,282 @@
+from __future__ import annotations
+
+import json
+from math import nan
+from typing import Any, Mapping
+
+import pytest
+
+from agent import (
+    AgentRole,
+    FileSystemArtifactStore,
+    FileSystemTaskStore,
+    RunConfig,
+    RoleRunConfigOverrides,
+    RoleRunConfigResolver,
+    wire_orchestration,
+)
+from agent.core.prompts.orchestration import PLANNER_ROLE_CONTRACT
+from agent.orchestration.executor import LocalAgentExecutor
+from agent.orchestration.models import CompletionPolicy, FailureCode
+
+
+class CapturingRunner:
+    trace_store = None
+
+    def __init__(self) -> None:
+        self.configs = []
+        self.messages = []
+
+    async def run_result(self, *, messages, config):
+        self.messages.append(messages)
+        self.configs.append(config)
+        return {"status": "completed", "summary": "done"}
+
+
+class RecordingResolver:
+    def __init__(self, overrides: RoleRunConfigOverrides) -> None:
+        self.overrides = overrides
+        self.calls = []
+
+    async def resolve(
+        self,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Mapping[str, Any],
+    ) -> RoleRunConfigOverrides:
+        with pytest.raises(TypeError):
+            context["injected"] = True  # type: ignore[index]
+        self.calls.append((role, preset, dict(context)))
+        return self.overrides
+
+
+def worker_context(**updates):
+    context = {
+        "worker_preset": "custom_worker",
+        "worker_trace_id": "worker-trace",
+        "root_trace_id": "root",
+        "task_id": "task",
+        "spec_version": 2,
+        "attempt_id": "attempt",
+        "task_spec": {"objective": "test"},
+        "continue_trace_id": None,
+    }
+    context.update(updates)
+    return context
+
+
+def validator_context(**updates):
+    context = {
+        "validator_preset": "custom_validator",
+        "validator_trace_id": "validator-trace",
+        "root_trace_id": "root",
+        "task_id": "task",
+        "spec_version": 2,
+        "attempt_id": "attempt",
+        "snapshot_id": "snapshot",
+        "validation_id": "validation",
+        "task_spec": {"objective": "test"},
+        "artifact_snapshot": {"artifact_refs": []},
+    }
+    context.update(updates)
+    return context
+
+
+def test_role_run_config_types_are_publicly_exported():
+    assert RoleRunConfigResolver.__name__ == "RoleRunConfigResolver"
+    assert RoleRunConfigOverrides().__dict__ == {
+        "model": None,
+        "temperature": None,
+        "max_iterations": None,
+    }
+
+
+@pytest.mark.asyncio
+async def test_local_executor_without_resolver_preserves_run_config_defaults():
+    runner = CapturingRunner()
+
+    result = await LocalAgentExecutor(runner).run_worker(worker_context())
+
+    assert result.status == "completed"
+    config = runner.configs[0]
+    assert config.model == "gpt-4o"
+    assert config.temperature == 0.3
+    assert config.max_iterations == 200
+
+
+@pytest.mark.asyncio
+async def test_worker_resolver_overrides_only_model_run_fields():
+    runner = CapturingRunner()
+    resolver = RecordingResolver(
+        RoleRunConfigOverrides(
+            model="worker-model",
+            temperature=0.15,
+            max_iterations=17,
+        )
+    )
+
+    result = await LocalAgentExecutor(runner, resolver).run_worker(
+        worker_context(untrusted="resolver-can-see-but-runner-must-not-receive")
+    )
+
+    assert result.status == "completed"
+    assert resolver.calls[0][0:2] == (AgentRole.WORKER, "custom_worker")
+    assert resolver.calls[0][2]["untrusted"] == (
+        "resolver-can-see-but-runner-must-not-receive"
+    )
+    config = runner.configs[0]
+    assert (config.model, config.temperature, config.max_iterations) == (
+        "worker-model",
+        0.15,
+        17,
+    )
+    assert config.agent_type == "custom_worker"
+    assert config.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION
+    assert config.new_trace_id == "worker-trace"
+    assert config.parent_trace_id == "root"
+    assert config.tools is None and config.tool_groups is None
+    assert config.parallel_tool_execution is False
+    assert config.enable_memory is False
+    assert config.enable_research_flow is False
+    assert "untrusted" not in config.context
+
+
+@pytest.mark.asyncio
+async def test_resolver_context_is_recursively_read_only_and_cannot_mutate_prompt():
+    class NestedMutationResolver:
+        async def resolve(self, *, role, preset, context):
+            del role, preset
+            with pytest.raises(TypeError):
+                context["task_spec"]["objective"] = "mutated"
+            with pytest.raises(AttributeError):
+                context["task_spec"]["refs"].append("mutated")
+            return RoleRunConfigOverrides()
+
+    runner = CapturingRunner()
+    result = await LocalAgentExecutor(runner, NestedMutationResolver()).run_worker(
+        worker_context(task_spec={"objective": "original", "refs": ["safe"]})
+    )
+
+    assert result.status == "completed"
+    prompt = json.loads(runner.messages[0][0]["content"])
+    assert prompt["task_spec"] == {"objective": "original", "refs": ["safe"]}
+    assert runner.configs[0].context["task_id"] == "task"
+
+
+def test_trusted_resolver_values_override_preset_model_policy():
+    from agent import AgentPreset, ToolRegistry
+    from agent.orchestration.policy import DefaultToolPolicy
+
+    config = RunConfig(
+        temperature=0.7,
+        max_iterations=99,
+        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+        _role_run_config_override_fields=frozenset({"temperature", "max_iterations"}),
+    )
+    preset = AgentPreset(
+        role=AgentRole.VALIDATOR,
+        allowed_tools=[],
+        temperature=0.0,
+        max_iterations=30,
+    )
+
+    resolved = DefaultToolPolicy().resolve(config, preset, ToolRegistry())
+
+    assert resolved.temperature == 0.7
+    assert resolved.max_iterations == 99
+
+
+@pytest.mark.asyncio
+async def test_validator_resolver_receives_validator_role_and_preset():
+    runner = CapturingRunner()
+    resolver = RecordingResolver(
+        RoleRunConfigOverrides(model="validator-model", temperature=0.0)
+    )
+
+    result = await LocalAgentExecutor(runner, resolver).run_validator(
+        validator_context()
+    )
+
+    assert result.status == "completed"
+    assert resolver.calls[0][0:2] == (AgentRole.VALIDATOR, "custom_validator")
+    config = runner.configs[0]
+    assert config.model == "validator-model"
+    assert config.temperature == 0.0
+    assert config.max_iterations == 200
+    assert config.context["validation_id"] == "validation"
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"model": " "},
+        {"model": 1},
+        {"temperature": nan},
+        {"temperature": -0.1},
+        {"temperature": True},
+        {"max_iterations": 0},
+        {"max_iterations": True},
+    ],
+)
+def test_role_run_config_overrides_reject_invalid_values(kwargs):
+    with pytest.raises(ValueError):
+        RoleRunConfigOverrides(**kwargs)
+
+
+@pytest.mark.asyncio
+async def test_invalid_resolver_result_is_normalized_as_executor_failure():
+    class InvalidResolver:
+        async def resolve(self, **kwargs):
+            return {"model": "unsafe"}
+
+    runner = CapturingRunner()
+    result = await LocalAgentExecutor(runner, InvalidResolver()).run_worker(
+        worker_context()
+    )
+
+    assert result.status == "failed"
+    assert "must return RoleRunConfigOverrides" in result.error
+    assert result.execution_stats.failure_code == FailureCode.EXECUTOR_ERROR
+    assert runner.configs == []
+
+
+@pytest.mark.asyncio
+async def test_resolver_exception_is_normalized_as_executor_failure():
+    class BrokenResolver:
+        async def resolve(self, **kwargs):
+            raise RuntimeError("model manifest unavailable")
+
+    runner = CapturingRunner()
+    result = await LocalAgentExecutor(runner, BrokenResolver()).run_validator(
+        validator_context()
+    )
+
+    assert result.status == "failed"
+    assert result.error == "model manifest unavailable"
+    assert result.execution_stats.failure_code == FailureCode.EXECUTOR_ERROR
+    assert runner.configs == []
+
+
+@pytest.mark.asyncio
+async def test_wire_orchestration_injects_role_run_config_resolver(tmp_path):
+    runner = CapturingRunner()
+    runner.task_coordinator = None
+    resolver = RecordingResolver(RoleRunConfigOverrides(model="wired-model"))
+
+    coordinator = wire_orchestration(
+        runner,
+        FileSystemTaskStore(str(tmp_path)),
+        FileSystemArtifactStore(str(tmp_path)),
+        role_run_config_resolver=resolver,
+    )
+    result = await coordinator.executor.run_worker(worker_context())
+
+    assert runner.task_coordinator is coordinator
+    assert result.status == "completed"
+    assert runner.configs[0].model == "wired-model"
+
+
+def test_planner_contract_allows_host_specific_dispatch_tool_name():
+    assert "dispatch_tasks" not in PLANNER_ROLE_CONTRACT
+    assert "task-dispatch tool exposed by your preset" in PLANNER_ROLE_CONTRACT