feat(application): 定义应用角色工具集 Provider 与规范哈希

新增实验性 cyber_agent.application 声明模型，固化 ApplicationRef、AgentApplication、AgentRole、ToolSet、ProviderRef 与 RunLimits 的严格序列化合同。

应用和角色哈希覆盖 Prompt、Skill 内容及顺序、工具声明与执行策略、Provider 公开配置、验证策略、上下文继承策略和资源限制；运行时对象与凭证不进入哈希。

同时为 ToolRegistry 增加定义导出、隔离克隆和冻结能力，使不同应用可以安全使用同名工具且不会共享统计或被注册后动态替换。

cyber_agent/application/__init__.py

@@ -0,0 +1,35 @@
+"""Experimental application framework API (no compatibility promise before M6)."""
+
+from cyber_agent.application.models import (
+    AgentApplication,
+    AgentRole,
+    ApplicationRef,
+    ProviderRef,
+    RoleRunLimits,
+    RunLimits,
+    SkillSpec,
+    ToolSet,
+    ToolSpec,
+)
+from cyber_agent.application.ports import (
+    ContextMaterial,
+    ContextRequest,
+    ContextResourceDescriptor,
+    TaskContextProvider,
+)
+
+__all__ = [
+    "AgentApplication",
+    "AgentRole",
+    "ApplicationRef",
+    "ContextMaterial",
+    "ContextRequest",
+    "ContextResourceDescriptor",
+    "ProviderRef",
+    "RoleRunLimits",
+    "RunLimits",
+    "SkillSpec",
+    "TaskContextProvider",
+    "ToolSet",
+    "ToolSpec",
+]

cyber_agent/application/models.py

@@ -0,0 +1,240 @@
+"""Serializable application declarations and their stable identity hashes."""
+
+from __future__ import annotations
+
+from hashlib import sha256
+import json
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+
+ValidationScopeName = Literal["evidence", "hypothesis", "output", "task"]
+
+
+class ApplicationModel(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        frozen=True,
+        str_strip_whitespace=True,
+    )
+
+
+def canonical_json(value: Any) -> str:
+    """Return the only JSON representation used by application hashes."""
+    if isinstance(value, BaseModel):
+        value = value.model_dump(mode="json")
+    try:
+        return json.dumps(
+            value,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"application declaration must be JSON serializable: {exc}") from exc
+
+
+def stable_hash(value: Any) -> str:
+    return sha256(canonical_json(value).encode("utf-8")).hexdigest()
+
+
+class ApplicationRef(ApplicationModel):
+    schema_version: Literal[1] = 1
+    application_id: str = Field(min_length=1, max_length=200)
+    application_version: str = Field(min_length=1, max_length=100)
+    config_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+
+
+class ProviderRef(ApplicationModel):
+    provider_id: str = Field(min_length=1, max_length=200)
+    provider_version: str = Field(min_length=1, max_length=100)
+    public_config: dict[str, Any] = Field(default_factory=dict)
+
+    @model_validator(mode="after")
+    def validate_public_config(self) -> "ProviderRef":
+        canonical_json(self.public_config)
+        return self
+
+
+class SkillSpec(ApplicationModel):
+    name: str = Field(min_length=1, max_length=200)
+    version: str = Field(min_length=1, max_length=100)
+    content: str = Field(min_length=1, max_length=100_000)
+
+    @property
+    def content_hash(self) -> str:
+        return stable_hash({
+            "name": self.name,
+            "version": self.version,
+            "content": self.content,
+        })
+
+
+class ToolSpec(ApplicationModel):
+    tool_name: str = Field(min_length=1, max_length=200)
+    implementation_version: str = Field(min_length=1, max_length=100)
+    side_effect_class: Literal["none", "read", "write", "external"] = "none"
+    idempotent: bool = True
+
+
+class ToolSet(ApplicationModel):
+    tool_set_id: str = Field(min_length=1, max_length=200)
+    tools: tuple[ToolSpec, ...] = ()
+
+    @model_validator(mode="after")
+    def validate_unique_tools(self) -> "ToolSet":
+        names = [item.tool_name for item in self.tools]
+        if len(names) != len(set(names)):
+            raise ValueError(f"ToolSet {self.tool_set_id} contains duplicate tools")
+        return self
+
+
+class RoleRunLimits(ApplicationModel):
+    max_iterations: int | None = Field(default=None, gt=0)
+    max_depth: int | None = Field(default=None, ge=0)
+    max_children_per_parent: int | None = Field(default=None, ge=0)
+    max_parallel_children: int | None = Field(default=None, gt=0)
+
+
+class RunLimits(ApplicationModel):
+    max_iterations: int = Field(default=200, gt=0)
+    max_depth: int = Field(default=10, ge=0)
+    max_children_per_parent: int = Field(default=20, ge=0)
+    max_parallel_children: int = Field(default=2, gt=0)
+    max_total_agents: int = Field(default=100, gt=0)
+    max_llm_calls: int = Field(default=1_000, gt=1)
+    max_total_tokens: int = Field(default=10_000_000, gt=0)
+    max_total_cost_usd: float = Field(default=100.0, gt=0)
+    max_duration_seconds: int = Field(default=86_400, gt=0)
+    max_validation_tool_calls: int = Field(default=100, gt=0)
+    max_validation_material_chars: int = Field(default=1_000_000, gt=0)
+
+    def tighten(self, role: RoleRunLimits | None) -> "RunLimits":
+        if role is None:
+            return self
+        updates = {
+            name: min(getattr(self, name), value)
+            for name, value in role.model_dump().items()
+            if value is not None
+        }
+        return self.model_copy(update=updates)
+
+
+class AgentRole(ApplicationModel):
+    role_id: str = Field(min_length=1, max_length=200)
+    description: str = Field(default="", max_length=2_000)
+    model: str = Field(min_length=1, max_length=200)
+    temperature: float = Field(default=0.3, ge=0, le=2)
+    model_parameters: dict[str, Any] = Field(default_factory=dict)
+    system_prompt: str = Field(min_length=1, max_length=200_000)
+    skills: tuple[SkillSpec, ...] = ()
+    tool_sets: tuple[str, ...] = ()
+    allowed_child_roles: tuple[str, ...] = ()
+    default_validation_scopes: tuple[ValidationScopeName, ...] = ()
+    run_limits: RoleRunLimits | None = None
+
+    @field_validator("tool_sets", "allowed_child_roles")
+    @classmethod
+    def validate_unique_refs(cls, values: tuple[str, ...]) -> tuple[str, ...]:
+        if len(values) != len(set(values)):
+            raise ValueError("role references must be unique")
+        return values
+
+    @model_validator(mode="after")
+    def validate_model_parameters(self) -> "AgentRole":
+        canonical_json(self.model_parameters)
+        return self
+
+
+class AgentApplication(ApplicationModel):
+    schema_version: Literal[1] = 1
+    application_id: str = Field(min_length=1, max_length=200)
+    application_version: str = Field(min_length=1, max_length=100)
+    root_role: str = Field(min_length=1, max_length=200)
+    roles: tuple[AgentRole, ...] = Field(min_length=1)
+    tool_sets: tuple[ToolSet, ...] = ()
+    context_provider_ref: ProviderRef | None = None
+    artifact_resolver_ref: ProviderRef | None = None
+    candidate_repository_ref: ProviderRef | None = None
+    quality_provider_ref: ProviderRef | None = None
+    event_projector_ref: ProviderRef | None = None
+    validation_policy: dict[str, Any] = Field(default_factory=dict)
+    run_limits: RunLimits = Field(default_factory=RunLimits)
+    context_inheritance_policy: Literal[
+        "explicit_only",
+        "allow_inheritable",
+    ] = "explicit_only"
+
+    @model_validator(mode="after")
+    def validate_graph(self) -> "AgentApplication":
+        role_ids = [item.role_id for item in self.roles]
+        if len(role_ids) != len(set(role_ids)):
+            raise ValueError("AgentApplication role_id values must be unique")
+        if self.root_role not in set(role_ids):
+            raise ValueError("root_role must reference an existing role")
+        tool_set_ids = [item.tool_set_id for item in self.tool_sets]
+        if len(tool_set_ids) != len(set(tool_set_ids)):
+            raise ValueError("AgentApplication tool_set_id values must be unique")
+        known_roles = set(role_ids)
+        known_tool_sets = set(tool_set_ids)
+        for role in self.roles:
+            missing_roles = set(role.allowed_child_roles) - known_roles
+            if missing_roles:
+                raise ValueError(
+                    f"role {role.role_id} references unknown child roles: "
+                    f"{sorted(missing_roles)}"
+                )
+            missing_sets = set(role.tool_sets) - known_tool_sets
+            if missing_sets:
+                raise ValueError(
+                    f"role {role.role_id} references unknown tool sets: "
+                    f"{sorted(missing_sets)}"
+                )
+        canonical_json(self.validation_policy)
+        return self
+
+    def role(self, role_id: str) -> AgentRole:
+        return next(item for item in self.roles if item.role_id == role_id)
+
+    def tool_set(self, tool_set_id: str) -> ToolSet:
+        return next(item for item in self.tool_sets if item.tool_set_id == tool_set_id)
+
+
+def role_manifest(
+    application: AgentApplication,
+    role: AgentRole,
+    tool_definitions: dict[str, dict[str, Any]],
+) -> dict[str, Any]:
+    selected_specs = [
+        spec
+        for tool_set_id in role.tool_sets
+        for spec in application.tool_set(tool_set_id).tools
+    ]
+    return {
+        "role": role.model_dump(mode="json"),
+        "prompt_hash": stable_hash(role.system_prompt),
+        "skill_hashes": [item.content_hash for item in role.skills],
+        "tools": [
+            {
+                "declaration": spec.model_dump(mode="json"),
+                "registry": tool_definitions[spec.tool_name],
+            }
+            for spec in selected_specs
+        ],
+    }
+
+
+def application_manifest(
+    application: AgentApplication,
+    tool_definitions: list[dict[str, Any]],
+) -> dict[str, Any]:
+    by_name = {item["tool_name"]: item for item in tool_definitions}
+    return {
+        "application": application.model_dump(mode="json"),
+        "roles": [
+            role_manifest(application, role, by_name)
+            for role in application.roles
+        ],
+    }

cyber_agent/application/ports.py

@@ -0,0 +1,57 @@
+"""Application-owned service ports used by the framework runtime."""
+
+from __future__ import annotations
+
+from typing import Any, Literal, Protocol, Sequence
+
+from pydantic import Field
+
+from cyber_agent.application.models import ApplicationModel, ApplicationRef
+from cyber_agent.core.task_protocol import ContextRef, TaskBrief
+
+
+class ContextRequest(ApplicationModel):
+    application_ref: ApplicationRef
+    root_trace_id: str = Field(min_length=1)
+    trace_id: str = Field(min_length=1)
+    parent_trace_id: str | None = None
+    uid: str | None = None
+    role_id: str = Field(min_length=1)
+    task_brief: TaskBrief | None = None
+    task_brief_revision: int = Field(default=0, ge=0)
+    authorized_context_refs: tuple[ContextRef, ...] = ()
+
+
+class ContextResourceDescriptor(ApplicationModel):
+    application_ref: ApplicationRef
+    root_trace_id: str = Field(min_length=1)
+    trace_id: str = Field(min_length=1)
+    uid: str | None = None
+    role_id: str = Field(min_length=1)
+    source_id: str = Field(min_length=1, max_length=500)
+    source_version: str = Field(min_length=1, max_length=200)
+    content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    kind: str = Field(min_length=1, max_length=200)
+    summary: str = Field(min_length=1, max_length=500)
+    inheritance: Literal["task_only", "inheritable"] = "task_only"
+    priority: int = 0
+    estimated_chars: int = Field(default=0, ge=0)
+    authorization: dict[str, Any] = Field(default_factory=dict)
+
+
+class ContextMaterial(ApplicationModel):
+    descriptor: ContextResourceDescriptor
+    content: dict[str, Any]
+
+
+class TaskContextProvider(Protocol):
+    async def list_context(
+        self,
+        request: ContextRequest,
+    ) -> Sequence[ContextResourceDescriptor]: ...
+
+    async def resolve_context(
+        self,
+        request: ContextRequest,
+        descriptor: ContextResourceDescriptor,
+    ) -> ContextMaterial: ...

cyber_agent/tools/registry.py

@@ -14,6 +14,7 @@ import json
 import inspect
 import logging
 import time
+from copy import deepcopy
 from typing import Any, Callable, Dict, List, Optional
 
 from cyber_agent.tools.approval import tool_argument_hash
@@ -59,6 +60,7 @@ class ToolRegistry:
 	def __init__(self):
 		self._tools: Dict[str, Dict[str, Any]] = {}
 		self._stats: Dict[str, ToolStats] = {}
+		self._frozen = False
 
 	def register(
 		self,
@@ -86,6 +88,8 @@ class ToolRegistry:
 			inject_params: 注入参数规则 {param_name: injector_func}
 			groups: 工具分组标签（如 ["core"]、["browser"]），用于 RunConfig.tool_groups 过滤
 		"""
+		if self._frozen:
+			raise RuntimeError("ToolRegistry is frozen")
 		func_name = func.__name__
 
 		# 如果没有提供 Schema，自动生成
@@ -148,6 +152,77 @@ class ToolRegistry:
 		"""检查工具是否已注册"""
 		return tool_name in self._tools
 
+	@property
+	def frozen(self) -> bool:
+		"""Whether registrations are permanently disabled for this instance."""
+		return self._frozen
+
+	def freeze(self) -> "ToolRegistry":
+		"""Freeze this registry in place and return it for fluent assembly."""
+		self._frozen = True
+		return self
+
+	def clone(
+		self,
+		tool_names: Optional[List[str]] = None,
+		*,
+		freeze: bool = False,
+	) -> "ToolRegistry":
+		"""Create an isolated registry with fresh statistics and shared callables."""
+		names = sorted(self._tools.keys() if tool_names is None else tool_names)
+		missing = [name for name in names if name not in self._tools]
+		if missing:
+			raise ValueError(f"Unknown tools for registry clone: {missing}")
+		cloned = ToolRegistry()
+		for name in names:
+			definition = self._tools[name]
+			cloned._tools[name] = {
+				"func": definition["func"],
+				"schema": deepcopy(definition["schema"]),
+				"url_patterns": deepcopy(definition.get("url_patterns")),
+				"hidden_params": list(definition.get("hidden_params", [])),
+				"inject_params": dict(definition.get("inject_params", {})),
+				"groups": list(definition.get("groups", [])),
+				"ui_metadata": deepcopy(definition.get("ui_metadata", {})),
+			}
+			cloned._stats[name] = ToolStats()
+		if freeze:
+			cloned.freeze()
+		return cloned
+
+	def export_definitions(
+		self,
+		tool_names: Optional[List[str]] = None,
+	) -> List[Dict[str, Any]]:
+		"""Export stable execution declarations without functions or statistics."""
+		names = sorted(self._tools.keys() if tool_names is None else tool_names)
+		missing = [name for name in names if name not in self._tools]
+		if missing:
+			raise ValueError(f"Unknown tools for definition export: {missing}")
+		definitions: List[Dict[str, Any]] = []
+		for name in names:
+			definition = self._tools[name]
+			inject_params = {}
+			for key, value in sorted(definition.get("inject_params", {}).items()):
+				if isinstance(value, dict):
+					inject_params[key] = {
+						item_key: item_value
+						for item_key, item_value in sorted(value.items())
+						if not callable(item_value)
+					}
+				else:
+					inject_params[key] = "runtime_value"
+			definitions.append({
+				"tool_name": name,
+				"schema": deepcopy(definition["schema"]),
+				"url_patterns": deepcopy(definition.get("url_patterns")),
+				"hidden_params": sorted(definition.get("hidden_params", [])),
+				"inject_params": inject_params,
+				"groups": sorted(definition.get("groups", [])),
+				"ui_metadata": deepcopy(definition.get("ui_metadata", {})),
+			})
+		return definitions
+
 	def get_schemas(self, tool_names: Optional[List[str]] = None) -> List[Dict]:
 		"""
 		获取工具 Schema