"""02 · Agent Configuration v1.1 demo. 确定性配置合成与安全不变量实验。标准库 only； 不调用真实模型 / 工具 / 密钥服务。 运行：python3 02-Agent-Configuration-Development-and-Practice-v1.1-demo.py """ from __future__ import annotations import copy import hashlib import json from dataclasses import dataclass, field from typing import Any, Dict, FrozenSet, Mapping, Optional, Set class ConfigError(Exception): def __init__(self, code: str, message: str = ""): self.code = code super().__init__(message or code) PROTECTED_POLICY_KEYS = frozenset({"policy_ref", "security_policy"}) @dataclass(frozen=True) class BaseConfig: tools: FrozenSet[str] budget: int approvals: FrozenSet[str] policy_ref: str skills: Mapping[str, FrozenSet[str]] = field(default_factory=dict) # skill_name -> required tools @dataclass(frozen=True) class EffectiveConfig: tools: FrozenSet[str] budget: int approvals: FrozenSet[str] policy_ref: str skills: FrozenSet[str] digest: str _snapshot: Dict[str, Any] = field(repr=False, compare=False) def as_dict(self) -> Dict[str, Any]: # 返回深拷贝，防止调用方篡改内部快照 return copy.deepcopy(self._snapshot) def _digest(payload: Mapping[str, Any]) -> str: raw = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def compose(base: BaseConfig, overlay: Optional[Mapping[str, Any]] = None) -> EffectiveConfig: """合成有效配置：工具取交集、预算取最小值、审批取并集；保护策略引用。""" overlay = dict(overlay or {}) unknown = set(overlay) - { "tools", "budget", "approvals", "skills", "policy_ref", "security_policy", } if unknown: raise ConfigError("UNKNOWN_FIELD", ",".join(sorted(unknown))) # 3+4) 受保护策略引用不能被任务级覆盖删除或改写 for key in PROTECTED_POLICY_KEYS: if key in overlay: raise ConfigError("PROTECTED_FIELD", key) # 1) 工具只能缩小 if "tools" in overlay: requested = overlay["tools"] if not isinstance(requested, (list, set, frozenset)): raise ConfigError("INVALID_TOOLS") requested_set = frozenset(requested) if not requested_set.issubset(base.tools): # 不允许通过覆盖引入 base 之外的工具 raise ConfigError("TOOLS_EXPANSION", ",".join(sorted(requested_set - base.tools))) effective_tools = base.tools & requested_set else: effective_tools = base.tools # 2) 预算只能收紧 if "budget" in overlay: budget = overlay["budget"] if not isinstance(budget, int) or isinstance(budget, bool) or budget < 1: raise ConfigError("INVALID_BUDGET") if budget > base.budget: raise ConfigError("BUDGET_EXPANSION") effective_budget = min(base.budget, budget) else: effective_budget = base.budget # 3) 审批只能增加 if "approvals" in overlay: requested_appr = overlay["approvals"] if not isinstance(requested_appr, (list, set, frozenset)): raise ConfigError("INVALID_APPROVALS") effective_approvals = base.approvals | frozenset(requested_appr) else: effective_approvals = base.approvals # skills: 默认启用 base 中全部；可缩小集合 available_skills = frozenset(base.skills) if "skills" in overlay: requested_skills = overlay["skills"] if not isinstance(requested_skills, (list, set, frozenset)): raise ConfigError("INVALID_SKILLS") requested_skills_set = frozenset(requested_skills) unknown_skills = requested_skills_set - available_skills if unknown_skills: raise ConfigError("UNKNOWN_SKILL", ",".join(sorted(unknown_skills))) effective_skills = requested_skills_set else: effective_skills = available_skills # 7) Skill 工具依赖 for skill in sorted(effective_skills): required = base.skills[skill] missing = required - effective_tools if missing: raise ConfigError( "SKILL_DEPENDENCY", f"{skill} missing tools: {','.join(sorted(missing))}", ) snapshot = { "tools": sorted(effective_tools), "budget": effective_budget, "approvals": sorted(effective_approvals), "policy_ref": base.policy_ref, "skills": sorted(effective_skills), } digest = _digest(snapshot) return EffectiveConfig( tools=effective_tools, budget=effective_budget, approvals=effective_approvals, policy_ref=base.policy_ref, skills=effective_skills, digest=digest, _snapshot=snapshot, ) def make_base() -> BaseConfig: return BaseConfig( tools=frozenset({"search", "read_source", "save_draft", "publish"}), budget=24, approvals=frozenset({"publish"}), policy_ref="policies/research-policy@1.0.0", skills={ "research_brief": frozenset({"search", "read_source", "save_draft"}), "publish_brief": frozenset({"publish"}), }, ) if __name__ == "__main__": base = make_base() safe = { "tools": ["search", "read_source", "save_draft"], "skills": ["research_brief"], } # 1) 运行覆盖只能缩小工具集合 narrowed = compose(base, safe) assert narrowed.tools == frozenset({"search", "read_source", "save_draft"}) try: compose(base, {"tools": ["search", "shell"], "skills": ["research_brief"]}) raise AssertionError("expected TOOLS_EXPANSION") except ConfigError as err: assert err.code == "TOOLS_EXPANSION" # 2) 预算只能收紧 tight = compose(base, {**safe, "budget": 10}) assert tight.budget == 10 try: compose(base, {**safe, "budget": 100}) raise AssertionError("expected BUDGET_EXPANSION") except ConfigError as err: assert err.code == "BUDGET_EXPANSION" # 3) 审批只能增加，不能删除 more = compose(base, {**safe, "approvals": ["external_send"]}) assert "publish" in more.approvals and "external_send" in more.approvals # 4) 受保护策略引用不能被任务级覆盖修改 try: compose(base, {**safe, "policy_ref": "policies/evil@9.9.9"}) raise AssertionError("expected PROTECTED_FIELD") except ConfigError as err: assert err.code == "PROTECTED_FIELD" # 5) 未知覆盖字段被拒绝 try: compose(base, {**safe, "evil_flag": True}) raise AssertionError("expected UNKNOWN_FIELD") except ConfigError as err: assert err.code == "UNKNOWN_FIELD" # 6) 有效配置包含稳定摘要，且不受后续输入对象修改影响 overlay = {**safe, "budget": 8} eff = compose(base, overlay) d1 = eff.digest overlay["budget"] = 999 # 修改输入不应影响已合成结果 assert eff.budget == 8 view = eff.as_dict() view["budget"] = 1 assert eff.as_dict()["budget"] == 8 assert eff.digest == d1 assert compose(base, {**safe, "budget": 8}).digest == d1 # 7) 不满足 Skill 工具依赖时拒绝配置 try: compose(base, {"tools": ["search"], "skills": ["research_brief"]}) raise AssertionError("expected SKILL_DEPENDENCY") except ConfigError as err: assert err.code == "SKILL_DEPENDENCY" # 8) 非法预算值被拒绝 try: compose(base, {"budget": 0}) raise AssertionError("expected INVALID_BUDGET") except ConfigError as err: assert err.code == "INVALID_BUDGET" try: compose(base, {"budget": 3.5}) # type: ignore[arg-type] raise AssertionError("expected INVALID_BUDGET") except ConfigError as err: assert err.code == "INVALID_BUDGET" print("8 configuration-contract checks passed")