mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-27 00:17:53 +00:00
65afc9b1d2
* fix(skills): scope allowed-tools to active skills * fix(skills): tolerate stale active skill paths * chore: retrigger CI * fix(skills): document policy activation limits * perf(skills): reuse per-step tool policy decisions * fix(skills): harden runtime tool policy contracts * fix(skills): redact cached policy decisions * fix(skills): make slash tool policy authoritative * fix(skills): preserve policy-safe discovery tools * test(skills): cover explicit task delegation policy
776 lines
27 KiB
Python
776 lines
27 KiB
Python
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from langchain.agents.middleware.types import ModelRequest
|
|
from langchain.tools import ToolRuntime
|
|
from langchain_core.messages import HumanMessage
|
|
from langgraph.prebuilt.tool_node import ToolCallRequest
|
|
from langgraph.runtime import Runtime
|
|
|
|
from deerflow.runtime.secret_context import SKILL_TOOL_POLICY_DECISION_CONTEXT_KEY, write_slash_skill_source_path
|
|
from deerflow.skills.types import Skill, SkillCategory
|
|
|
|
_SLASH_SOURCE_OWNER_TOKEN = "test-slash-source-owner"
|
|
|
|
|
|
class NamedTool:
|
|
def __init__(self, name: str):
|
|
self.name = name
|
|
|
|
|
|
class ModelRequestStub:
|
|
def __init__(self, tools, *, state=None, context=None, messages=None):
|
|
self.tools = tools
|
|
self.state = state or {}
|
|
self.runtime = SimpleNamespace(context={} if context is None else context)
|
|
self.messages = messages or []
|
|
|
|
def override(self, **updates):
|
|
return ModelRequestStub(
|
|
updates.get("tools", self.tools),
|
|
state=updates.get("state", self.state),
|
|
context=self.runtime.context,
|
|
messages=updates.get("messages", self.messages),
|
|
)
|
|
|
|
|
|
class ToolRequestStub:
|
|
def __init__(self, name: str, *, state=None, context=None):
|
|
self.tool_call = {"name": name, "id": "call-1", "args": {}}
|
|
self.state = state or {}
|
|
self.runtime = SimpleNamespace(context={} if context is None else context)
|
|
|
|
|
|
class StorageStub:
|
|
def __init__(self, skills):
|
|
self._skills = skills
|
|
self.load_calls = 0
|
|
|
|
def load_skills(self, *, enabled_only=False):
|
|
self.load_calls += 1
|
|
return [skill for skill in self._skills if skill.enabled or not enabled_only]
|
|
|
|
def get_container_root(self):
|
|
return "/mnt/skills"
|
|
|
|
|
|
def _skill(name: str, allowed_tools, *, enabled=True):
|
|
skill_dir = Path(f"/tmp/skills/public/{name}")
|
|
return Skill(
|
|
name=name,
|
|
description=f"Description for {name}",
|
|
license="MIT",
|
|
skill_dir=skill_dir,
|
|
skill_file=skill_dir / "SKILL.md",
|
|
relative_path=Path(name),
|
|
category=SkillCategory.PUBLIC,
|
|
allowed_tools=None if allowed_tools is None else tuple(allowed_tools),
|
|
enabled=enabled,
|
|
)
|
|
|
|
|
|
def _middleware(skills, *, available_skills=None):
|
|
from deerflow.agents.middlewares.skill_tool_policy_middleware import SkillToolPolicyMiddleware
|
|
|
|
middleware = SkillToolPolicyMiddleware(
|
|
available_skills=available_skills,
|
|
slash_source_owner_token=_SLASH_SOURCE_OWNER_TOKEN,
|
|
)
|
|
middleware._storage = lambda: StorageStub(skills)
|
|
return middleware
|
|
|
|
|
|
def _tool_names(request):
|
|
return [tool.name for tool in request.tools]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"middleware_class_path",
|
|
[
|
|
"deerflow.agents.middlewares.skill_activation_middleware.SkillActivationMiddleware",
|
|
"deerflow.agents.middlewares.skill_tool_policy_middleware.SkillToolPolicyMiddleware",
|
|
],
|
|
)
|
|
def test_skill_policy_middlewares_require_shared_slash_source_token(middleware_class_path):
|
|
module_name, class_name = middleware_class_path.rsplit(".", 1)
|
|
module = __import__(module_name, fromlist=[class_name])
|
|
middleware_class = getattr(module, class_name)
|
|
|
|
with pytest.raises(TypeError, match="slash_source_owner_token"):
|
|
middleware_class()
|
|
|
|
|
|
@pytest.mark.parametrize("invalid_token", [None, "", 7])
|
|
def test_skill_policy_middlewares_reject_invalid_slash_source_tokens(invalid_token):
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
from deerflow.agents.middlewares.skill_tool_policy_middleware import SkillToolPolicyMiddleware
|
|
|
|
for middleware_class in (SkillActivationMiddleware, SkillToolPolicyMiddleware):
|
|
with pytest.raises(ValueError, match="non-empty string"):
|
|
middleware_class(slash_source_owner_token=invalid_token)
|
|
|
|
|
|
def test_passive_enabled_skill_does_not_filter_lead_tools():
|
|
middleware = _middleware([_skill("reviewer", ["review_skill_package"])])
|
|
request = ModelRequestStub([NamedTool("task"), NamedTool("web_search"), NamedTool("review_skill_package")])
|
|
|
|
filtered = middleware._filter_model_request(request)
|
|
|
|
assert _tool_names(filtered) == ["task", "web_search", "review_skill_package"]
|
|
|
|
|
|
def test_sync_passive_model_call_skips_storage():
|
|
middleware = _middleware([])
|
|
|
|
def fail_storage():
|
|
raise AssertionError("passive model calls must not load skill storage")
|
|
|
|
middleware._storage = fail_storage
|
|
request = ModelRequestStub([NamedTool("task")])
|
|
|
|
assert middleware.wrap_model_call(request, lambda model_request: model_request) is request
|
|
|
|
|
|
def test_async_passive_model_call_skips_storage_and_thread_offload():
|
|
middleware = _middleware([])
|
|
|
|
def fail_storage():
|
|
raise AssertionError("passive model calls must not load skill storage")
|
|
|
|
middleware._storage = fail_storage
|
|
request = ModelRequestStub([NamedTool("task")])
|
|
|
|
async def handler(model_request):
|
|
return model_request
|
|
|
|
assert asyncio.run(middleware.awrap_model_call(request, handler)) is request
|
|
|
|
|
|
def test_slash_activated_skill_filters_first_model_call_and_task():
|
|
skill = _skill("reviewer", ["review_skill_package"])
|
|
context = {}
|
|
write_slash_skill_source_path(
|
|
context,
|
|
skill.get_container_file_path(),
|
|
owner_token=_SLASH_SOURCE_OWNER_TOKEN,
|
|
)
|
|
middleware = _middleware([skill])
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("read_file"), NamedTool("review_skill_package")],
|
|
context=context,
|
|
)
|
|
|
|
filtered = middleware._filter_model_request(request)
|
|
|
|
assert _tool_names(filtered) == ["read_file", "review_skill_package"]
|
|
|
|
|
|
@pytest.mark.parametrize("active_source", ["slash", "skill_context"])
|
|
def test_restrictive_skill_explicitly_allows_task_schema_and_execution(active_source):
|
|
skill = _skill("delegating", ["task"])
|
|
context = {}
|
|
state = {}
|
|
if active_source == "slash":
|
|
write_slash_skill_source_path(
|
|
context,
|
|
skill.get_container_file_path(),
|
|
owner_token=_SLASH_SOURCE_OWNER_TOKEN,
|
|
)
|
|
else:
|
|
state = {
|
|
"skill_context": [
|
|
{
|
|
"name": skill.name,
|
|
"path": skill.get_container_file_path(),
|
|
}
|
|
]
|
|
}
|
|
|
|
middleware = _middleware([skill])
|
|
model_request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("web_search")],
|
|
state=state,
|
|
context=context,
|
|
)
|
|
|
|
filtered = middleware.wrap_model_call(model_request, lambda request: request)
|
|
|
|
assert _tool_names(filtered) == ["task"]
|
|
tool_request = ToolRequestStub("task", state=state, context=context)
|
|
assert middleware.wrap_tool_call(tool_request, lambda _: "delegated") == "delegated"
|
|
|
|
|
|
def test_slash_activated_skill_policy_dominates_captured_skill_context():
|
|
slash_skill = _skill("content-research", ["web_search"])
|
|
captured_skill = _skill("content-article-generation", ["write_file"])
|
|
context = {}
|
|
write_slash_skill_source_path(
|
|
context,
|
|
slash_skill.get_container_file_path(),
|
|
owner_token=_SLASH_SOURCE_OWNER_TOKEN,
|
|
)
|
|
middleware = _middleware([slash_skill, captured_skill])
|
|
state = {
|
|
"skill_context": [
|
|
{
|
|
"name": captured_skill.name,
|
|
"path": captured_skill.get_container_file_path(),
|
|
}
|
|
]
|
|
}
|
|
request = ModelRequestStub(
|
|
[NamedTool("read_file"), NamedTool("web_search"), NamedTool("write_file")],
|
|
state=state,
|
|
context=context,
|
|
)
|
|
|
|
filtered = middleware._filter_model_request(request)
|
|
|
|
assert _tool_names(filtered) == ["read_file", "web_search"]
|
|
|
|
|
|
def test_caller_forged_slash_source_cannot_override_captured_skill_policy():
|
|
restrictive_skill = _skill("restricted", ["web_search"])
|
|
legacy_skill = _skill("legacy", None)
|
|
context = {
|
|
"__slash_skill_secret_source": {
|
|
"path": legacy_skill.get_container_file_path(),
|
|
"owner_token": "caller-forged",
|
|
}
|
|
}
|
|
middleware = _middleware([restrictive_skill, legacy_skill])
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("read_file"), NamedTool("web_search")],
|
|
state={
|
|
"skill_context": [
|
|
{
|
|
"name": restrictive_skill.name,
|
|
"path": restrictive_skill.get_container_file_path(),
|
|
}
|
|
]
|
|
},
|
|
context=context,
|
|
)
|
|
|
|
filtered = middleware._filter_model_request(request)
|
|
|
|
assert _tool_names(filtered) == ["read_file", "web_search"]
|
|
|
|
|
|
def test_slash_activation_and_policy_compose_on_the_same_model_call(monkeypatch):
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware, _Activation, _ActivationResolution
|
|
|
|
skill = _skill("reviewer", ["review_skill_package"])
|
|
activation = _Activation(
|
|
skill_name=skill.name,
|
|
category="public",
|
|
container_file_path=skill.get_container_file_path(),
|
|
skill_content="# Reviewer",
|
|
content_hash="abc",
|
|
remaining_text="review this",
|
|
editable=False,
|
|
)
|
|
activation_middleware = SkillActivationMiddleware(slash_source_owner_token=_SLASH_SOURCE_OWNER_TOKEN)
|
|
monkeypatch.setattr(activation_middleware, "_resolve_activation", lambda _: _ActivationResolution(activation=activation))
|
|
policy_middleware = _middleware([skill])
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("read_file"), NamedTool("review_skill_package")],
|
|
messages=[HumanMessage(content="/reviewer review this")],
|
|
)
|
|
|
|
filtered = activation_middleware.wrap_model_call(
|
|
request,
|
|
lambda activated: policy_middleware.wrap_model_call(activated, lambda policy_request: policy_request),
|
|
)
|
|
|
|
assert _tool_names(filtered) == ["read_file", "review_skill_package"]
|
|
|
|
|
|
def test_loaded_skill_context_filters_follow_up_model_calls():
|
|
skill = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([skill])
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("read_file"), NamedTool("web_search")],
|
|
state={"skill_context": [{"name": skill.name, "path": skill.get_container_file_path()}]},
|
|
)
|
|
|
|
filtered = middleware._filter_model_request(request)
|
|
|
|
assert _tool_names(filtered) == ["read_file", "web_search"]
|
|
|
|
|
|
def test_active_skill_union_and_legacy_semantics_are_preserved():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
second = _skill("second", ["bash"])
|
|
legacy = _skill("legacy", None)
|
|
middleware = _middleware([restricted, second, legacy])
|
|
state = {
|
|
"skill_context": [
|
|
{"path": restricted.get_container_file_path()},
|
|
{"path": second.get_container_file_path()},
|
|
{"path": legacy.get_container_file_path()},
|
|
]
|
|
}
|
|
request = ModelRequestStub([NamedTool("task"), NamedTool("bash"), NamedTool("web_search")], state=state)
|
|
|
|
filtered = middleware._filter_model_request(request)
|
|
|
|
assert _tool_names(filtered) == ["bash", "web_search"]
|
|
|
|
|
|
def test_only_legacy_active_skill_preserves_all_tools():
|
|
legacy = _skill("legacy", None)
|
|
middleware = _middleware([legacy])
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("bash")],
|
|
state={"skill_context": [{"path": legacy.get_container_file_path()}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == ["task", "bash"]
|
|
|
|
|
|
def test_explicit_empty_allowed_tools_keeps_only_framework_tools():
|
|
restricted = _skill("restricted", [])
|
|
middleware = _middleware([restricted])
|
|
request = ModelRequestStub(
|
|
[
|
|
NamedTool("task"),
|
|
NamedTool("read_file"),
|
|
NamedTool("review_skill_package"),
|
|
NamedTool("tool_search"),
|
|
NamedTool("describe_skill"),
|
|
],
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == [
|
|
"read_file",
|
|
"review_skill_package",
|
|
"tool_search",
|
|
"describe_skill",
|
|
]
|
|
|
|
|
|
def test_active_skill_keeps_framework_discovery_tools():
|
|
restricted = _skill("restricted", ["calc"])
|
|
middleware = _middleware([restricted])
|
|
request = ModelRequestStub(
|
|
[NamedTool("calc"), NamedTool("tool_search"), NamedTool("describe_skill")],
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == ["calc", "tool_search", "describe_skill"]
|
|
|
|
|
|
def test_custom_agent_allowlist_rejects_all_out_of_scope_active_skills():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([restricted], available_skills={"other"})
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("web_search")],
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == []
|
|
|
|
|
|
def test_unauthorized_tool_execution_is_blocked():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([restricted])
|
|
request = ToolRequestStub(
|
|
"task",
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
)
|
|
|
|
result = middleware.wrap_tool_call(request, lambda _: "executed")
|
|
|
|
assert result.status == "error"
|
|
assert result.name == "task"
|
|
assert "not allowed" in result.content
|
|
|
|
|
|
def test_allowed_tool_execution_reaches_handler():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([restricted])
|
|
request = ToolRequestStub(
|
|
"web_search",
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
)
|
|
|
|
assert middleware.wrap_tool_call(request, lambda _: "executed") == "executed"
|
|
|
|
|
|
def test_async_unauthorized_tool_execution_is_blocked():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([restricted])
|
|
request = ToolRequestStub(
|
|
"task",
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
)
|
|
|
|
async def handler(_):
|
|
return "executed"
|
|
|
|
result = asyncio.run(middleware.awrap_tool_call(request, handler))
|
|
|
|
assert result.status == "error"
|
|
assert result.name == "task"
|
|
|
|
|
|
def test_unknown_skill_context_path_is_skipped_while_resolvable_skills_apply():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([restricted])
|
|
request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("read_file"), NamedTool("web_search")],
|
|
state={
|
|
"skill_context": [
|
|
{"path": "/mnt/skills/public/missing/SKILL.md"},
|
|
{"path": restricted.get_container_file_path()},
|
|
]
|
|
},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == ["read_file", "web_search"]
|
|
|
|
|
|
def test_all_unknown_active_paths_fail_closed_to_framework_tools():
|
|
middleware = _middleware([])
|
|
request = ModelRequestStub(
|
|
[
|
|
NamedTool("task"),
|
|
NamedTool("read_file"),
|
|
NamedTool("review_skill_package"),
|
|
NamedTool("tool_search"),
|
|
NamedTool("describe_skill"),
|
|
],
|
|
state={"skill_context": [{"path": "/mnt/skills/public/missing/SKILL.md"}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == [
|
|
"read_file",
|
|
"review_skill_package",
|
|
"tool_search",
|
|
"describe_skill",
|
|
]
|
|
|
|
|
|
def test_all_disabled_active_paths_fail_closed_to_framework_tools():
|
|
disabled = _skill("disabled", ["task"], enabled=False)
|
|
middleware = _middleware([disabled])
|
|
request = ModelRequestStub(
|
|
[
|
|
NamedTool("task"),
|
|
NamedTool("read_file"),
|
|
NamedTool("review_skill_package"),
|
|
NamedTool("tool_search"),
|
|
NamedTool("describe_skill"),
|
|
],
|
|
state={"skill_context": [{"path": disabled.get_container_file_path()}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == [
|
|
"read_file",
|
|
"review_skill_package",
|
|
"tool_search",
|
|
"describe_skill",
|
|
]
|
|
|
|
|
|
def test_async_passive_tool_call_skips_storage_and_thread_offload():
|
|
middleware = _middleware([])
|
|
|
|
def fail_storage():
|
|
raise AssertionError("passive tool calls must not load skill storage")
|
|
|
|
middleware._storage = fail_storage
|
|
request = ToolRequestStub("task")
|
|
|
|
async def handler(_):
|
|
return "executed"
|
|
|
|
assert asyncio.run(middleware.awrap_tool_call(request, handler)) == "executed"
|
|
|
|
|
|
def test_sync_passive_tool_call_skips_policy_resolution():
|
|
middleware = _middleware([])
|
|
request = ToolRequestStub("task")
|
|
middleware._blocked_tool_message = MagicMock(side_effect=AssertionError("passive tool calls must bypass policy resolution"))
|
|
|
|
assert middleware.wrap_tool_call(request, lambda _: "executed") == "executed"
|
|
middleware._blocked_tool_message.assert_not_called()
|
|
|
|
|
|
def test_tool_calls_reuse_the_current_model_step_policy_decision():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
context = {}
|
|
state = {"skill_context": [{"path": restricted.get_container_file_path()}]}
|
|
model_request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("web_search")],
|
|
state=state,
|
|
context=context,
|
|
)
|
|
|
|
filtered = middleware.wrap_model_call(model_request, lambda request: request)
|
|
assert _tool_names(filtered) == ["web_search"]
|
|
|
|
for _ in range(3):
|
|
tool_request = ToolRequestStub("web_search", state=state, context=context)
|
|
assert middleware.wrap_tool_call(tool_request, lambda _: "executed") == "executed"
|
|
|
|
assert storage.load_calls == 1
|
|
|
|
|
|
def test_async_tool_calls_reuse_the_current_model_step_policy_decision():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
context = {}
|
|
state = {"skill_context": [{"path": restricted.get_container_file_path()}]}
|
|
model_request = ModelRequestStub(
|
|
[NamedTool("task"), NamedTool("web_search")],
|
|
state=state,
|
|
context=context,
|
|
)
|
|
|
|
async def go():
|
|
filtered = await middleware.awrap_model_call(model_request, lambda request: asyncio.sleep(0, result=request))
|
|
assert _tool_names(filtered) == ["web_search"]
|
|
|
|
async def execute(_):
|
|
return "executed"
|
|
|
|
results = await asyncio.gather(*(middleware.awrap_tool_call(ToolRequestStub("web_search", state=state, context=context), execute) for _ in range(3)))
|
|
assert results == ["executed", "executed", "executed"]
|
|
|
|
asyncio.run(go())
|
|
assert storage.load_calls == 1
|
|
|
|
|
|
def test_real_model_and_tool_requests_share_the_model_step_policy_decision():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
context = {}
|
|
state = {
|
|
"messages": [],
|
|
"skill_context": [{"path": restricted.get_container_file_path()}],
|
|
}
|
|
model_request = ModelRequest(
|
|
model=MagicMock(),
|
|
messages=[],
|
|
tools=[NamedTool("task"), NamedTool("web_search")],
|
|
state=state,
|
|
runtime=Runtime(context=context),
|
|
)
|
|
|
|
filtered = middleware.wrap_model_call(model_request, lambda request: request)
|
|
assert _tool_names(filtered) == ["web_search"]
|
|
|
|
tool_runtime = ToolRuntime(
|
|
state=state,
|
|
context=context,
|
|
config={},
|
|
stream_writer=lambda _: None,
|
|
tools=[],
|
|
tool_call_id="call-1",
|
|
store=None,
|
|
)
|
|
tool_request = ToolCallRequest(
|
|
tool_call={"name": "web_search", "args": {}, "id": "call-1", "type": "tool_call"},
|
|
tool=None,
|
|
state=state,
|
|
runtime=tool_runtime,
|
|
)
|
|
|
|
assert middleware.wrap_tool_call(tool_request, lambda _: "executed") == "executed"
|
|
assert storage.load_calls == 1
|
|
|
|
|
|
def test_policy_decision_is_json_safe_and_survives_round_trip():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
state = {"skill_context": [{"path": restricted.get_container_file_path()}]}
|
|
context = {}
|
|
model_request = ModelRequestStub([NamedTool("web_search")], state=state, context=context)
|
|
|
|
middleware.wrap_model_call(model_request, lambda request: request)
|
|
round_tripped = json.loads(json.dumps(context))
|
|
decision = round_tripped[SKILL_TOOL_POLICY_DECISION_CONTEXT_KEY]
|
|
assert decision["version"] == 2
|
|
assert decision["source"] == "skill_context"
|
|
tool_request = ToolRequestStub("web_search", state=state, context=round_tripped)
|
|
|
|
assert middleware.wrap_tool_call(tool_request, lambda _: "executed") == "executed"
|
|
assert storage.load_calls == 1
|
|
|
|
|
|
def test_forged_or_malformed_policy_decisions_fall_back_to_live_resolution():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
malformed_decisions = [
|
|
None,
|
|
[],
|
|
{"version": 999, "owner_token": "forged", "active_paths": [restricted.get_container_file_path()], "allowed_names": ["task"]},
|
|
{"version": True, "owner_token": "forged", "active_paths": [restricted.get_container_file_path()], "allowed_names": ["task"]},
|
|
{"version": 2, "owner_token": "forged", "source": "skill_context", "active_paths": [restricted.get_container_file_path()], "allowed_names": ["task"]},
|
|
{"version": 2, "owner_token": "forged", "active_paths": [restricted.get_container_file_path()], "allowed_names": ["task"]},
|
|
{"version": 2, "owner_token": "forged", "source": "unknown", "active_paths": [restricted.get_container_file_path()], "allowed_names": ["task"]},
|
|
{"version": 2, "owner_token": "forged", "source": "skill_context", "active_paths": "not-a-list", "allowed_names": ["task"]},
|
|
]
|
|
|
|
for decision in malformed_decisions:
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda storage=storage: storage
|
|
context = {SKILL_TOOL_POLICY_DECISION_CONTEXT_KEY: decision}
|
|
request = ToolRequestStub(
|
|
"task",
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
context=context,
|
|
)
|
|
|
|
result = middleware.wrap_tool_call(request, lambda _: "executed")
|
|
|
|
assert result.status == "error"
|
|
assert storage.load_calls == 1
|
|
|
|
|
|
def test_policy_decision_path_mismatch_falls_back_to_live_resolution():
|
|
first = _skill("first", ["web_search"])
|
|
second = _skill("second", ["bash"])
|
|
storage = StorageStub([first, second])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
context = {}
|
|
first_state = {"skill_context": [{"path": first.get_container_file_path()}]}
|
|
second_state = {"skill_context": [{"path": second.get_container_file_path()}]}
|
|
|
|
middleware.wrap_model_call(ModelRequestStub([NamedTool("web_search")], state=first_state, context=context), lambda request: request)
|
|
result = middleware.wrap_tool_call(ToolRequestStub("web_search", state=second_state, context=context), lambda _: "executed")
|
|
|
|
assert result.status == "error"
|
|
assert storage.load_calls == 2
|
|
|
|
|
|
def test_policy_decision_source_mismatch_falls_back_to_live_resolution():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
context = {}
|
|
state = {"skill_context": [{"path": restricted.get_container_file_path()}]}
|
|
|
|
middleware.wrap_model_call(ModelRequestStub([NamedTool("web_search")], state=state, context=context), lambda request: request)
|
|
write_slash_skill_source_path(
|
|
context,
|
|
restricted.get_container_file_path(),
|
|
owner_token=_SLASH_SOURCE_OWNER_TOKEN,
|
|
)
|
|
result = middleware.wrap_tool_call(ToolRequestStub("web_search", state=state, context=context), lambda _: "executed")
|
|
|
|
assert result == "executed"
|
|
assert storage.load_calls == 2
|
|
|
|
|
|
def test_active_paths_support_attribute_based_state():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
middleware = _middleware([restricted])
|
|
state = SimpleNamespace(skill_context=[{"path": restricted.get_container_file_path()}])
|
|
request = ModelRequestStub([NamedTool("task"), NamedTool("web_search")], state=state)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == ["web_search"]
|
|
|
|
|
|
def test_active_paths_support_falsey_attribute_based_state():
|
|
class FalseyState:
|
|
skill_context = []
|
|
|
|
def __bool__(self):
|
|
return False
|
|
|
|
restricted = _skill("restricted", ["web_search"])
|
|
state = FalseyState()
|
|
state.skill_context = [{"path": restricted.get_container_file_path()}]
|
|
middleware = _middleware([restricted])
|
|
request = ModelRequestStub([NamedTool("task"), NamedTool("web_search")])
|
|
request.state = state
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == ["web_search"]
|
|
|
|
|
|
def test_unknown_state_shape_is_logged_instead_of_silently_ignored(caplog):
|
|
middleware = _middleware([])
|
|
request = ModelRequestStub([NamedTool("task")], state=object())
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == ["task"]
|
|
assert "Unsupported agent state shape" in caplog.text
|
|
|
|
|
|
def test_next_model_call_refreshes_the_policy_decision():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
context = {}
|
|
state = {"skill_context": [{"path": restricted.get_container_file_path()}]}
|
|
|
|
first = ModelRequestStub([NamedTool("bash"), NamedTool("web_search")], state=state, context=context)
|
|
assert _tool_names(middleware.wrap_model_call(first, lambda request: request)) == ["web_search"]
|
|
|
|
storage._skills = [_skill("restricted", ["bash"])]
|
|
second = ModelRequestStub([NamedTool("bash"), NamedTool("web_search")], state=state, context=context)
|
|
assert _tool_names(middleware.wrap_model_call(second, lambda request: request)) == ["bash"]
|
|
assert storage.load_calls == 2
|
|
|
|
|
|
def test_tool_call_without_matching_model_decision_revalidates_registry():
|
|
restricted = _skill("restricted", ["web_search"])
|
|
storage = StorageStub([restricted])
|
|
middleware = _middleware([])
|
|
middleware._storage = lambda: storage
|
|
request = ToolRequestStub(
|
|
"task",
|
|
state={"skill_context": [{"path": restricted.get_container_file_path()}]},
|
|
context={},
|
|
)
|
|
|
|
result = middleware.wrap_tool_call(request, lambda _: "executed")
|
|
|
|
assert result.status == "error"
|
|
assert storage.load_calls == 1
|
|
|
|
|
|
def test_active_policy_load_failure_fails_closed_to_framework_tools():
|
|
middleware = _middleware([])
|
|
|
|
def fail_storage():
|
|
raise RuntimeError("storage unavailable")
|
|
|
|
middleware._storage = fail_storage
|
|
request = ModelRequestStub(
|
|
[
|
|
NamedTool("task"),
|
|
NamedTool("read_file"),
|
|
NamedTool("review_skill_package"),
|
|
NamedTool("tool_search"),
|
|
NamedTool("describe_skill"),
|
|
],
|
|
state={"skill_context": [{"path": "/mnt/skills/public/restricted/SKILL.md"}]},
|
|
)
|
|
|
|
assert _tool_names(middleware._filter_model_request(request)) == [
|
|
"read_file",
|
|
"review_skill_package",
|
|
"tool_search",
|
|
"describe_skill",
|
|
]
|