GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-15 19:14:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(agent): align policy contracts with runtime (#6306)

Browse Source
This commit is contained in:
InfinityPacer
2026-08-13 22:14:43 +08:00
committed by GitHub
parent 57e7f86f38
commit 402543f636
9 changed files with 42 additions and 573 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/agent/middleware/policy.py
View File

@@ -114,8 +114,7 @@ class AgentPolicyMiddleware(AgentMiddleware):
handler=lambda: handler(request),
enforce_decision=False,
)
# 普通 ToolNode 在严格策略接管前保持 shadow 观测语义。
# 已确认的受保护调用会使用默认的强制决策语义。
# 普通 ToolNode 保持 shadow 观测；已确认调用使用默认的强制决策语义。
return result
async def execute_tool_call(

16
app/agent/policy/__init__.py
View File

@@ -4,14 +4,9 @@ from app.agent.policy.contracts import (
ActionEffect,
ActionPolicy,
AuthSource,
CanonicalInvocation,
ConversationKind,
DeliveryTarget,
ConfirmationMode,
ExecutionOutcome,
ExecutionReceipt,
InboundEnvelope,
InboundProvenance,
MigrationState,
PolicyDecision,
PolicyObservation,
@@ -19,10 +14,7 @@ from app.agent.policy.contracts import (
PrincipalRole,
PrincipalType,
RecoveryMode,
RECEIPT_STATE_TRANSITIONS,
ReceiptState,
ResultSensitivity,
TERMINAL_RECEIPT_STATES,
ToolInvocation,
ToolOrigin,
ToolPolicyContext,
@@ -48,16 +40,11 @@ __all__ = [
"ActionPolicy",
"AgentToolPolicyOrchestrator",
"AuthSource",
"CanonicalInvocation",
"ConversationKind",
"DeliveryTarget",
"ConfirmationMode",
"DEFAULT_TOOL_POLICY_ORCHESTRATOR",
"DEFAULT_TOOL_POLICY_REGISTRY",
"ExecutionOutcome",
"ExecutionReceipt",
"InboundEnvelope",
"InboundProvenance",
"MigrationState",
"PolicyDecision",
"PolicyObservation",
@@ -66,10 +53,7 @@ __all__ = [
"PrincipalType",
"REDACTED_VALUE",
"RecoveryMode",
"RECEIPT_STATE_TRANSITIONS",
"ReceiptState",
"ResultSensitivity",
"TERMINAL_RECEIPT_STATES",
"ToolInvocation",
"ToolOrigin",
"ToolPolicyContext",

130
app/agent/policy/canonical.py
View File

@@ -1,130 +0,0 @@
"""严格工具调用规范化，不读取设置值或执行工具实现。"""
import hashlib
import json
from typing import Any, Mapping
from pydantic import BaseModel, ValidationError
from app.agent.policy.contracts import (
ActionPolicy,
CanonicalInvocation,
ToolRevision,
)
from app.agent.tools.impl._system_setting_utils import (
list_setting_specs,
resolve_setting_spec,
)
class CanonicalizationError(ValueError):
"""当前工具 schema 或静态前提无法产生严格调用时的稳定失败。"""
def _stable_json(value: Any) -> str:
"""生成拒绝 NaN 且保留 Unicode 的稳定紧凑 JSON。"""
try:
return json.dumps(
value,
ensure_ascii=False,
allow_nan=False,
sort_keys=True,
separators=(",", ":"),
)
except (TypeError, ValueError, ValidationError) as error:
raise CanonicalizationError("调用参数无法规范化") from error
def _accepted_argument_names(args_schema: type[BaseModel]) -> set[str]:
"""返回严格入口可接受的字段名与字符串别名。"""
accepted = set(args_schema.model_fields)
for field in args_schema.model_fields.values():
for alias in (field.alias, field.validation_alias):
if isinstance(alias, str):
accepted.add(alias)
return accepted
def canonicalize_invocation(
*,
tool: Any,
arguments: Mapping[str, Any],
policy: ActionPolicy,
tool_revision: ToolRevision,
) -> CanonicalInvocation:
"""使用当前 Pydantic schema 与静态设置定义生成不可变调用摘要。"""
tool_name = str(getattr(tool, "name", "") or "")
args_schema = getattr(tool, "args_schema", None)
if not tool_name or not isinstance(args_schema, type) or not issubclass(
args_schema, BaseModel
):
raise CanonicalizationError("工具缺少严格 Pydantic 参数契约")
raw_arguments = dict(arguments or {})
unknown_arguments = set(raw_arguments) - _accepted_argument_names(args_schema)
if unknown_arguments:
raise CanonicalizationError("工具参数校验失败")
try:
validated = args_schema.model_validate(raw_arguments)
normalized = validated.model_dump(
mode="json",
exclude_unset=False,
exclude_none=False,
)
schema_json = _stable_json(args_schema.model_json_schema())
except (TypeError, ValueError) as error:
raise CanonicalizationError("工具参数校验失败") from error
preconditions: tuple[tuple[str, str], ...] = ()
setting_key = normalized.get("setting_key")
if tool_name == "query_system_settings":
if setting_key:
spec = resolve_setting_spec(setting_key)
if spec is None:
raise CanonicalizationError("系统设置项不存在")
specs = [spec]
else:
try:
specs = list_setting_specs(
group=normalized.get("group"),
keyword=normalized.get("keyword"),
)
except ValueError as error:
raise CanonicalizationError("系统设置选择器无效") from error
if not specs:
raise CanonicalizationError("系统设置选择器没有匹配项")
preconditions = tuple(
(
"setting",
f"{spec.source}:{spec.key}:{spec.group}",
)
for spec in specs
)
payload = {
"canonical_version": "p1-g2a2-canonical-v1",
"action_subtype": policy.effect.value,
"arguments": normalized,
"policy_version": policy.policy_version,
"preconditions": preconditions,
"schema_digest": hashlib.sha256(schema_json.encode("utf-8")).hexdigest(),
"tool_name": tool_name,
"tool_revision": {
"factory": tool_revision.factory,
"implementation": tool_revision.implementation,
"plugin": tool_revision.plugin,
},
}
canonical_json = _stable_json(payload)
return CanonicalInvocation(
tool_name=tool_name,
arguments=normalized,
canonical_json=canonical_json,
digest=hashlib.sha256(canonical_json.encode("utf-8")).hexdigest(),
policy_version=policy.policy_version,
tool_revision=tool_revision,
schema_digest=payload["schema_digest"],
preconditions=preconditions,
)
__all__ = ["CanonicalizationError", "canonicalize_invocation"]

168
app/agent/policy/contracts.py
View File

@@ -2,23 +2,9 @@
from dataclasses import dataclass, field
from enum import Enum
from types import MappingProxyType
from typing import Any, Mapping, MutableMapping, Optional
def _freeze_contract_value(value: Any) -> Any:
"""递归冻结确认边界中的容器，避免等待期间被调用方修改。"""
if isinstance(value, Mapping):
return MappingProxyType(
{str(key): _freeze_contract_value(item) for key, item in value.items()}
)
if isinstance(value, (list, tuple)):
return tuple(_freeze_contract_value(item) for item in value)
if isinstance(value, set):
return frozenset(_freeze_contract_value(item) for item in value)
return value
class ToolOrigin(str, Enum):
"""工具调用的宿主可信入口。"""
@@ -49,24 +35,6 @@ class AuthSource(str, Enum):
AGENT_TOKEN = "agent_token"
class InboundProvenance(str, Enum):
"""入站事实经宿主验证后的可信等级。"""
WEB_SESSION = "web_session"
VERIFIED_ADAPTER = "verified_adapter"
ADMIN_INTEGRATION = "admin_integration"
UNTRUSTED = "untrusted"
class ConversationKind(str, Enum):
"""目标会话的隐私边界。"""
WEB = "web"
PRIVATE = "private"
GROUP = "group"
UNKNOWN = "unknown"
class PrincipalRole(str, Enum):
"""策略授权使用的角色层级。"""
@@ -118,92 +86,19 @@ class ResultSensitivity(str, Enum):
class MigrationState(str, Enum):
"""工具策略从兼容观测迁移到宿主执行的状态。"""
"""工具策略当前采用宿主强制还是兼容观测。"""
ENFORCED = "enforced"
LEGACY_SHADOW = "legacy_shadow"
class ExecutionOutcome(str, Enum):
"""P1-G1 handler 生命周期终态；成功不代表业务授权或副作用已完成。"""
"""工具 handler 观测终态；成功不代表业务授权或副作用已完成。"""
SUCCEEDED = "succeeded"
FAILED = "failed"
class ReceiptState(str, Enum):
"""严格执行回执的单向生命周期状态。"""
WAITING_CONFIRMATION = "waiting_confirmation"
VALIDATING = "validating"
EXECUTING = "executing"
DELIVERING = "delivering"
SUCCEEDED = "succeeded"
CANCELLED = "cancelled"
EXPIRED = "expired"
EXPIRED_RESTART = "expired_restart"
EXPIRED_ORPHANED = "expired_orphaned"
PREPARATION_FAILED = "preparation_failed"
PROMPT_DELIVERY_FAILED = "prompt_delivery_failed"
VALIDATION_FAILED = "validation_failed"
VALIDATION_RECORD_FAILED = "validation_record_failed"
EXECUTION_FAILED = "execution_failed"
DELIVERY_FAILED = "delivery_failed"
UNKNOWN_AFTER_RESTART = "unknown_after_restart"
UNKNOWN_ORPHANED = "unknown_orphaned"
TERMINAL_RECEIPT_STATES = frozenset({
ReceiptState.SUCCEEDED,
ReceiptState.CANCELLED,
ReceiptState.EXPIRED,
ReceiptState.EXPIRED_RESTART,
ReceiptState.EXPIRED_ORPHANED,
ReceiptState.PREPARATION_FAILED,
ReceiptState.PROMPT_DELIVERY_FAILED,
ReceiptState.VALIDATION_FAILED,
ReceiptState.VALIDATION_RECORD_FAILED,
ReceiptState.EXECUTION_FAILED,
ReceiptState.DELIVERY_FAILED,
ReceiptState.UNKNOWN_AFTER_RESTART,
ReceiptState.UNKNOWN_ORPHANED,
})
RECEIPT_STATE_TRANSITIONS = {
ReceiptState.WAITING_CONFIRMATION: frozenset({
ReceiptState.VALIDATING,
ReceiptState.CANCELLED,
ReceiptState.EXPIRED,
ReceiptState.EXPIRED_RESTART,
ReceiptState.EXPIRED_ORPHANED,
ReceiptState.PREPARATION_FAILED,
ReceiptState.PROMPT_DELIVERY_FAILED,
}),
ReceiptState.VALIDATING: frozenset({
ReceiptState.EXECUTING,
ReceiptState.CANCELLED,
ReceiptState.EXPIRED,
ReceiptState.EXPIRED_RESTART,
ReceiptState.EXPIRED_ORPHANED,
ReceiptState.VALIDATION_FAILED,
ReceiptState.VALIDATION_RECORD_FAILED,
}),
ReceiptState.EXECUTING: frozenset({
ReceiptState.DELIVERING,
ReceiptState.EXECUTION_FAILED,
ReceiptState.UNKNOWN_AFTER_RESTART,
ReceiptState.UNKNOWN_ORPHANED,
}),
ReceiptState.DELIVERING: frozenset({
ReceiptState.SUCCEEDED,
ReceiptState.DELIVERY_FAILED,
ReceiptState.UNKNOWN_AFTER_RESTART,
ReceiptState.UNKNOWN_ORPHANED,
}),
}
@dataclass(frozen=True)
class PolicyPrincipal:
"""由可信入口建立、不可由工具参数覆盖的调用主体。"""
@@ -229,62 +124,15 @@ class ToolInvocation:
source: Optional[str] = None
@dataclass(frozen=True)
class DeliveryTarget:
"""受保护结果的精确宿主路由，actor 与 recipient 不得互相替代。"""
channel: str
source_instance_id: str
tenant_or_account_id: str
conversation_kind: ConversationKind
conversation_id: str
recipient_id: str
actor_id: str
server_session_id: str
@dataclass(frozen=True)
class InboundEnvelope:
"""由可信宿主入口创建的不可变入站事实。"""
provenance: InboundProvenance
target: DeliveryTarget
inbound_event_id: str
raw_text: str = field(repr=False)
normalized_text: str = field(repr=False)
has_images: bool = False
has_audio: bool = False
has_files: bool = False
is_callback: bool = False
@dataclass(frozen=True)
class ToolRevision:
"""严格调用绑定的工具实现、工厂和插件目录版本。"""
"""记录目录项绑定的工具实现、工厂和插件目录版本。"""
implementation: str
factory: str
plugin: str
@dataclass(frozen=True)
class CanonicalInvocation:
"""经当前工具 schema 校验、可绑定确认与版本前提的调用。"""
tool_name: str
arguments: Mapping[str, Any] = field(repr=False)
canonical_json: str = field(repr=False)
digest: str
policy_version: str
tool_revision: ToolRevision
schema_digest: str
preconditions: tuple[tuple[str, str], ...] = ()
def __post_init__(self) -> None:
"""阻止调用方在确认等待期间修改规范化参数。"""
object.__setattr__(self, "arguments", _freeze_contract_value(self.arguments))
@dataclass(frozen=True)
class ActionPolicy:
"""参数级动作策略及其兼容迁移状态。"""
@@ -325,7 +173,7 @@ class PolicyObservation:
@dataclass(frozen=True)
class ExecutionReceipt:
"""P1-G1 的非持久化脱敏回执 envelope。"""
"""工具策略生命周期生成的非持久化脱敏回执。"""
invocation_id: str
tool_name: str
@@ -391,14 +239,9 @@ __all__ = [
"ActionEffect",
"ActionPolicy",
"AuthSource",
"CanonicalInvocation",
"ConfirmationMode",
"ConversationKind",
"DeliveryTarget",
"ExecutionOutcome",
"ExecutionReceipt",
"InboundEnvelope",
"InboundProvenance",
"MigrationState",
"PolicyDecision",
"PolicyObservation",
@@ -406,10 +249,7 @@ __all__ = [
"PrincipalRole",
"PrincipalType",
"RecoveryMode",
"RECEIPT_STATE_TRANSITIONS",
"ReceiptState",
"ResultSensitivity",
"TERMINAL_RECEIPT_STATES",
"ToolInvocation",
"ToolOrigin",
"ToolPolicyContext",

8
app/agent/policy/orchestrator.py
View File

@@ -37,7 +37,7 @@ def call_policy_hook(
*args: Any,
**kwargs: Any,
) -> Optional[_HookResult]:
"""以 fail-open 方式调用 P1-G1 观测 hook，故障只记录稳定类型。"""
"""以 fail-open 方式调用兼容观测 hook，故障只记录稳定类型。"""
try:
return hook(*args, **kwargs)
except Exception as error:
@@ -76,7 +76,7 @@ class AgentToolPolicyOrchestrator:
"""让 Agent middleware 与 direct manager 复用同一策略生命周期。"""
def __init__(self, registry: ToolPolicyRegistry = DEFAULT_TOOL_POLICY_REGISTRY) -> None:
"""绑定固定工具迁移注册表。"""
"""绑定工具策略解析表。"""
self.registry = registry
def start(
@@ -103,12 +103,12 @@ class AgentToolPolicyOrchestrator:
reason_code="legacy_shadow_allow",
)
elif policy.confirmation is ConfirmationMode.REQUIRED:
# 严格运行时接管前保持既有调用能力，但不得把敏感动作记为安全读取。
# 通用编排器保持 shadow；支持的 Agent 入口会在 ToolNode 前独立完成确认。
decision = PolicyDecision(
allowed=True,
confirmation_required=False,
shadow=True,
reason_code="strict_runtime_pending",
reason_code="confirmation_policy_shadow_allow",
)
else:
decision = PolicyDecision(

38
app/agent/policy/registry.py
View File

@@ -1,4 +1,4 @@
"""固定工具迁移注册表与参数级策略解析。"""
"""工具策略例外与参数级策略解析。"""
from typing import Any, Mapping
@@ -13,7 +13,7 @@ from app.agent.policy.contracts import (
)
# 这些读取已具备清晰的无副作用语义，用于证明新宿主边界不会改变正常结果。
# 这些非管理员读取在运行时解析为强制 SAFE_READ；管理员门禁仍沿用原有授权事实源。
SAFE_READ_TOOL_NAMES = frozenset(
{
"list_slash_commands",
@@ -25,8 +25,8 @@ SAFE_READ_TOOL_NAMES = frozenset(
)
# 其余固定工具先显式处于兼容观测状态，待领域叶子 Goal 逐个迁移。
LEGACY_SHADOW_TOOL_NAMES = frozenset(
# 该清单只校验固定工具 inventory；未命中的固定或动态工具同样默认 LEGACY_SHADOW。
BUILTIN_LEGACY_SHADOW_INVENTORY = frozenset(
{
"add_custom_filter_rule",
"add_download_tasks",
@@ -112,25 +112,29 @@ LEGACY_SHADOW_TOOL_NAMES = frozenset(
class ToolPolicyRegistry:
"""解析固定和动态工具的 P1-G1 迁移策略。"""
"""解析固定和动态工具的宿主策略。"""
def __init__(
self,
*,
safe_read_tool_names: frozenset[str] = SAFE_READ_TOOL_NAMES,
legacy_shadow_tool_names: frozenset[str] = LEGACY_SHADOW_TOOL_NAMES,
builtin_legacy_shadow_inventory: frozenset[str] = (
BUILTIN_LEGACY_SHADOW_INVENTORY
),
) -> None:
"""建立互斥的固定工具迁移表。"""
overlap = safe_read_tool_names & legacy_shadow_tool_names
"""建立 SAFE_READ 例外与固定工具 inventory。"""
overlap = safe_read_tool_names & builtin_legacy_shadow_inventory
if overlap:
raise ValueError(f"工具策略迁移表存在重复项: {sorted(overlap)}")
raise ValueError(f"工具策略 inventory 存在重复项: {sorted(overlap)}")
self._safe_read_tool_names = safe_read_tool_names
self._legacy_shadow_tool_names = legacy_shadow_tool_names
self._builtin_legacy_shadow_inventory = builtin_legacy_shadow_inventory
@property
def builtin_tool_names(self) -> set[str]:
"""返回注册表覆盖的全部固定工具名。"""
return set(self._safe_read_tool_names | self._legacy_shadow_tool_names)
def builtin_tool_inventory(self) -> set[str]:
"""返回用于测试校验的固定工具 inventory。"""
return set(
self._safe_read_tool_names | self._builtin_legacy_shadow_inventory
)
def resolve(
self,
@@ -139,7 +143,7 @@ class ToolPolicyRegistry:
arguments: Mapping[str, Any],
requires_admin: bool,
) -> ActionPolicy:
"""根据工具名和宿主权限声明解析当前迁移策略。"""
"""根据工具名和宿主权限声明解析当前参数级策略。"""
required_role = (
PrincipalRole.SYSTEM_ADMIN if requires_admin else PrincipalRole.USER
)
@@ -166,7 +170,7 @@ class ToolPolicyRegistry:
confirmation=ConfirmationMode.NONE,
recovery=RecoveryMode.NONE,
result_sensitivity=ResultSensitivity.NORMAL,
# 角色门禁仍可能异步识别渠道管理员；G1 不复制旧授权事实源。
# 角色门禁仍由既有授权事实源判断，管理员读取保持兼容观测。
migration_state=(
MigrationState.LEGACY_SHADOW
if requires_admin
@@ -174,7 +178,7 @@ class ToolPolicyRegistry:
),
)
# 固定未迁移工具和动态工具都保持现有执行能力，但不得被视为安全读取。
# 除明确例外外，固定和动态工具都保持现有能力，但不得被视为安全读取。
return ActionPolicy(
effect=ActionEffect.UNKNOWN,
required_role=required_role,
@@ -189,8 +193,8 @@ DEFAULT_TOOL_POLICY_REGISTRY = ToolPolicyRegistry()
__all__ = [
"BUILTIN_LEGACY_SHADOW_INVENTORY",
"DEFAULT_TOOL_POLICY_REGISTRY",
"LEGACY_SHADOW_TOOL_NAMES",
"SAFE_READ_TOOL_NAMES",
"ToolPolicyRegistry",
]

2
app/agent/tools/catalog.py
View File

@@ -65,7 +65,7 @@ class ToolCatalogEntry:
@dataclass(frozen=True)
class ToolCatalogSnapshot:
"""同一图构造与严格执行共享的本地工具事实源。"""
"""图构造、缓存签名和身份碰撞审计共享的本地工具事实源。"""
entries: tuple[ToolCatalogEntry, ...]
plugin_revision: int

228
tests/test_agent_policy_canonical.py
View File

@@ -1,228 +0,0 @@
import math
from dataclasses import FrozenInstanceError
import pytest
from pydantic import BaseModel, ConfigDict, Field
from app.agent.policy.canonical import (
CanonicalizationError,
canonicalize_invocation,
)
from app.agent.policy.contracts import (
ConversationKind,
DeliveryTarget,
InboundEnvelope,
InboundProvenance,
ToolRevision,
)
from app.agent.policy.registry import DEFAULT_TOOL_POLICY_REGISTRY
from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool
def _secret_policy():
return DEFAULT_TOOL_POLICY_REGISTRY.resolve(
tool_name="query_system_settings",
arguments={"show_secrets": True},
requires_admin=True,
)
def _tool_revision() -> ToolRevision:
return ToolRevision(
implementation="query-system-settings:1",
factory="builtin-factory:1",
plugin="plugin-catalog:1",
)
def test_canonical_invocation_includes_defaults_and_is_stable() -> None:
"""省略的默认值必须进入相同规范化参数与摘要。"""
tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin")
omitted = canonicalize_invocation(
tool=tool,
arguments={"setting_key": "COOKIECLOUD_KEY", "show_secrets": True},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
explicit = canonicalize_invocation(
tool=tool,
arguments={
"setting_key": "COOKIECLOUD_KEY",
"group": "all",
"keyword": None,
"include_values": None,
"show_secrets": True,
},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
assert omitted.digest == explicit.digest
assert omitted.arguments == explicit.arguments
assert omitted.preconditions == (
("setting", "settings:COOKIECLOUD_KEY:settings"),
)
def test_canonical_invocation_repr_hides_arguments_and_json() -> None:
"""调用对象的 repr 不得泄露确认参数或完整规范化 JSON。"""
marker = "canonical-private-marker"
class _Input(BaseModel):
model_config = ConfigDict(extra="forbid")
value: str
class _Tool:
name = "private_tool"
args_schema = _Input
invocation = canonicalize_invocation(
tool=_Tool(),
arguments={"value": marker},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
assert marker not in repr(invocation)
assert "canonical_json" not in repr(invocation)
def test_canonical_arguments_are_recursively_immutable() -> None:
"""确认等待期间不能修改嵌套参数后复用旧 digest。"""
class _Input(BaseModel):
payload: dict[str, list[str]]
class _Tool:
name = "nested_tool"
args_schema = _Input
invocation = canonicalize_invocation(
tool=_Tool(),
arguments={"payload": {"items": ["one"]}},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
with pytest.raises(TypeError):
invocation.arguments["payload"]["items"] += ("two",)
def test_inbound_contract_hides_raw_text_and_keeps_target_identities_separate() -> None:
"""入站原文不可进入 repr，actor、recipient 和 conversation 独立绑定。"""
target = DeliveryTarget(
channel="telegram",
source_instance_id="source-1",
tenant_or_account_id="account-1",
conversation_kind=ConversationKind.PRIVATE,
conversation_id="chat-1",
recipient_id="recipient-1",
actor_id="actor-1",
server_session_id="session-1",
)
envelope = InboundEnvelope(
provenance=InboundProvenance.VERIFIED_ADAPTER,
target=target,
inbound_event_id="event-1",
raw_text=" confirm K7P4-M2Q8 ",
normalized_text="confirm K7P4-M2Q8",
)
rendered = repr(envelope)
assert "K7P4-M2Q8" not in rendered
assert target.actor_id != target.recipient_id
assert target.recipient_id != target.conversation_id
with pytest.raises(FrozenInstanceError):
envelope.inbound_event_id = "forged"
def test_canonical_invocation_rejects_unknown_arguments() -> None:
"""严格确认不能把 schema 外参数降级为原始字典。"""
tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin")
with pytest.raises(CanonicalizationError, match="参数校验失败"):
canonicalize_invocation(
tool=tool,
arguments={"show_secrets": True, "forged": "value"},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
def test_canonical_invocation_rejects_missing_pydantic_schema() -> None:
"""动态 dict schema 不能进入严格确认路径。"""
class _Tool:
name = "dynamic_tool"
args_schema = {"type": "object"}
with pytest.raises(CanonicalizationError, match="Pydantic"):
canonicalize_invocation(
tool=_Tool(),
arguments={},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
def test_canonical_invocation_preserves_unicode_and_rejects_nan() -> None:
"""稳定 JSON 保留 Unicode，同时禁止非标准 NaN。"""
class _Input(BaseModel):
label: str
value: float = Field(allow_inf_nan=True)
class _Tool:
name = "unicode_tool"
args_schema = _Input
with pytest.raises(CanonicalizationError, match="无法规范化"):
canonicalize_invocation(
tool=_Tool(),
arguments={"label": "中文", "value": math.nan},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
def test_canonicalization_reads_no_setting_value(monkeypatch) -> None:
"""确认前只能读取静态 SettingSpec，不能访问实际设置值。"""
tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin")
load_value = monkeypatch.setattr(
QuerySystemSettingsTool,
"_load_setting_value",
lambda *_: pytest.fail("canonicalization must not read setting value"),
)
invocation = canonicalize_invocation(
tool=tool,
arguments={"setting_key": "COOKIECLOUD_KEY", "show_secrets": True},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
assert invocation.digest
assert load_value is None
def test_group_selector_binds_static_setting_set_without_loading_values(
monkeypatch,
) -> None:
"""列表型密钥读取必须绑定匹配的静态设置集合。"""
tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin")
monkeypatch.setattr(
QuerySystemSettingsTool,
"_load_setting_value",
lambda *_: pytest.fail("canonicalization must not read setting value"),
)
invocation = canonicalize_invocation(
tool=tool,
arguments={"group": "ai_agent", "show_secrets": True},
policy=_secret_policy(),
tool_revision=_tool_revision(),
)
assert len(invocation.preconditions) > 1
assert all(kind == "setting" for kind, _identity in invocation.preconditions)

22
tests/test_agent_tool_policy.py
View File

@@ -126,8 +126,8 @@ def _interactive_context(*, is_admin: bool = True) -> ToolPolicyContext:
)
def test_builtin_policy_registry_covers_every_fixed_tool() -> None:
"""固定内置工具必须全部具有显式 migration registry 条目。"""
def test_builtin_policy_inventory_covers_every_fixed_tool() -> None:
"""固定内置工具 inventory 必须随工厂入口同步。"""
fixed_tool_names = {
_tool_class_name(tool_class)
for tool_class in MoviePilotToolFactory.BUILTIN_TOOL_CLASSES
@@ -140,11 +140,11 @@ def test_builtin_policy_registry_covers_every_fixed_tool() -> None:
}
)
assert DEFAULT_TOOL_POLICY_REGISTRY.builtin_tool_names == fixed_tool_names
assert DEFAULT_TOOL_POLICY_REGISTRY.builtin_tool_inventory == fixed_tool_names
def test_registry_separates_safe_read_from_legacy_shadow() -> None:
"""少量安全读取可直接迁移，其余高影响工具保持 shadow allow。"""
def test_registry_applies_safe_read_exceptions_and_defaults_to_shadow() -> None:
"""SAFE_READ 仅用于明确例外，其他固定和动态工具默认 shadow。"""
safe_policy = DEFAULT_TOOL_POLICY_REGISTRY.resolve(
tool_name="query_personas",
arguments={},
@@ -216,7 +216,7 @@ def test_system_settings_secret_read_has_enforced_sensitive_policy() -> None:
def test_legacy_shadow_decision_allows_without_claiming_enforcement() -> None:
"""G1 的 shadow 决策只能观测，不能拒绝或要求确认。"""
"""shadow 决策只能观测，不能拒绝或要求确认。"""
context = _interactive_context()
tool = _EchoTool(session_id="session-1", user_id="user-1")
@@ -231,8 +231,8 @@ def test_legacy_shadow_decision_allows_without_claiming_enforcement() -> None:
assert observation.decision.reason_code == "legacy_shadow_allow"
def test_sensitive_policy_does_not_claim_safe_read_before_strict_runtime() -> None:
"""严格运行时接管前，敏感读取只能以明确的兼容 shadow 语义通过。"""
def test_sensitive_policy_stays_shadow_in_generic_orchestrator() -> None:
"""敏感读取在通用编排器中保持明确的兼容 shadow 语义。"""
tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin")
tool.set_agent_context({"is_admin": True})
@@ -246,7 +246,7 @@ def test_sensitive_policy_does_not_claim_safe_read_before_strict_runtime() -> No
assert observation.decision.allowed is True
assert observation.decision.shadow is True
assert observation.decision.confirmation_required is False
assert observation.decision.reason_code == "strict_runtime_pending"
assert observation.decision.reason_code == "confirmation_policy_shadow_allow"
def test_policy_context_reads_mutable_admin_state_without_model_fields() -> None:
@@ -439,8 +439,8 @@ def test_middleware_fail_observation_does_not_mask_tool_error() -> None:
assert error_info.value is tool_error
def test_middleware_keeps_shadow_observation_until_strict_runtime_takeover() -> None:
"""普通 ToolNode 在严格策略接管前不得因观测决策改变既有行为。"""
def test_middleware_keeps_shadow_observation_without_enforcing_decision() -> None:
"""普通 ToolNode 不得因 shadow 观测决策改变既有行为。"""
orchestrator = MagicMock()
orchestrator.start.return_value = SimpleNamespace(
decision=SimpleNamespace(allowed=False)