deer-flow/backend/tests/test_agent_assembly_descriptor.py
kbsama 3e1349576e
fix(agents): filter assembly descriptor subagent policy by allowed_subagents (#5205) (#5262)
* fix(agents): filter assembly descriptor subagent policy by allowed_subagents (#5205)

* test(agents): ensure non-vacuous subagent catalog in assembly descriptor test

---------

Co-authored-by: 1747687484-collab <229902011+1747687484-collab@users.noreply.github.com>
2026-09-10 14:51:42 +08:00

667 lines
28 KiB
Python

"""What the agent was actually assembled from, captured at build time.
Everything here is knowable only inside the factory: the resolved model after
runtime overrides, the rendered prompt, the tool list after authorization
filtering, the composed middleware stack. None of it survives to any later
observation point.
"""
from pathlib import Path
from deerflow_extension_api import AgentAssemblyDescriptor, MiddlewareDescriptor, ToolDescriptor
def test_fingerprint_is_stable_for_identical_assemblies():
def make():
return AgentAssemblyDescriptor(
namespace="lead",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_parameters={"temperature": 0},
thinking_enabled=False,
reasoning_effort=None,
base_prompt_hash="abc",
tools=(ToolDescriptor(name="bash", description_hash="d", schema_hash="s", source="builtin"),),
middlewares=(MiddlewareDescriptor(name="M", module="m", policy_parameters={"limit": 1}),),
deferred_tool_names=(),
enabled_skills=(),
effective_policies={"recursion_limit": 100},
)
assert make().fingerprint == make().fingerprint
def test_fingerprint_changes_when_a_middleware_policy_changes():
from dataclasses import replace
base = AgentAssemblyDescriptor(
namespace="lead",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_parameters={},
thinking_enabled=False,
reasoning_effort=None,
base_prompt_hash="abc",
tools=(),
middlewares=(MiddlewareDescriptor(name="M", module="m", policy_parameters={"limit": 1}),),
deferred_tool_names=(),
enabled_skills=(),
effective_policies={},
)
changed = replace(base, middlewares=(MiddlewareDescriptor(name="M", module="m", policy_parameters={"limit": 2}),))
assert base.fingerprint != changed.fingerprint
def test_fingerprint_ignores_tool_ordering():
"""Tool order is an assembly detail, not a behavioural difference."""
from dataclasses import replace
a = ToolDescriptor(name="a", description_hash="1", schema_hash="1", source="builtin")
b = ToolDescriptor(name="b", description_hash="2", schema_hash="2", source="builtin")
base = AgentAssemblyDescriptor(
namespace="lead",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_parameters={},
thinking_enabled=False,
reasoning_effort=None,
base_prompt_hash="abc",
tools=(a, b),
middlewares=(),
deferred_tool_names=(),
enabled_skills=(),
effective_policies={},
)
assert base.fingerprint == replace(base, tools=(b, a)).fingerprint
def test_middleware_order_does_affect_the_fingerprint():
"""Stack order determines what wraps what, so it is behavioural."""
from dataclasses import replace
m1 = MiddlewareDescriptor(name="A", module="m", policy_parameters={})
m2 = MiddlewareDescriptor(name="B", module="m", policy_parameters={})
base = AgentAssemblyDescriptor(
namespace="lead",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_parameters={},
thinking_enabled=False,
reasoning_effort=None,
base_prompt_hash="abc",
tools=(),
middlewares=(m1, m2),
deferred_tool_names=(),
enabled_skills=(),
effective_policies={},
)
assert base.fingerprint != replace(base, middlewares=(m2, m1)).fingerprint
class TestLeadAgentAssembly:
def test_make_lead_agent_still_returns_a_bare_graph(self):
"""langgraph.json declares this factory; its ABI must not move."""
import inspect
from deerflow.agents.lead_agent.agent import make_lead_agent
signature = inspect.signature(make_lead_agent)
assert list(signature.parameters) == ["config"]
@staticmethod
def _isolate_from_the_ambient_config(monkeypatch):
"""Assemble against a config this test owns, not the machine's.
``assemble_lead_agent`` falls back to ``get_app_config()``, so without
this the test passes only where a developer happens to have a usable
``config.yaml``. CI checks out ``config.example.yaml``, whose ``models:``
entries are all commented out, and assembly raises "No chat models are
configured" before it can produce anything to assert on.
"""
from deerflow.agents.lead_agent import agent as lead_agent_module
from deerflow.config.app_config import AppConfig
from deerflow.config.model_config import ModelConfig
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.config.subagents_config import CustomSubagentConfig, SubagentsAppConfig
app_config = AppConfig(
models=[
ModelConfig(
name="assembly-test-model",
display_name="assembly-test-model",
description=None,
use="langchain_openai:ChatOpenAI",
model="assembly-test-model",
supports_thinking=False,
supports_vision=False,
)
],
subagents=SubagentsAppConfig(custom_agents={"researcher": CustomSubagentConfig(description="research", system_prompt="research")}),
sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
)
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(
lead_agent_module,
"create_chat_model",
lambda **kwargs: object(),
)
monkeypatch.setattr(lead_agent_module, "create_agent", lambda **kwargs: kwargs)
return app_config
@staticmethod
def _extensions_with_an_agent_assembly_observer(observer=None):
"""A minimal LoadedExtensions carrying one agent-assembly observer.
Building the descriptor is real work (hashing every tool's description
and schema, probing every middleware), so it only happens when an
observer is actually registered to receive it.
"""
from deerflow.extensions.registry import ExtensionRegistry
class _NoOpObserver:
def on_agent_assembled(self, app_store, descriptor):
return None
registry = ExtensionRegistry()
with registry.attributed_to("test"):
registry.agent_assembly_observer(observer or _NoOpObserver())
return registry.build()
def test_assemble_returns_both_the_graph_and_a_descriptor(self, monkeypatch):
from deerflow.agents.lead_agent.agent import LeadAgentAssembly, assemble_lead_agent
from deerflow.extensions import bind_agent_build_extensions
self._isolate_from_the_ambient_config(monkeypatch)
with bind_agent_build_extensions(self._extensions_with_an_agent_assembly_observer()):
assembly = assemble_lead_agent({"configurable": {"thread_id": "t-1"}})
assert isinstance(assembly, LeadAgentAssembly)
assert assembly.graph is not None
assert assembly.descriptor.effective_model
assert assembly.descriptor.fingerprint
def test_descriptor_hashes_the_same_scoped_prompt_passed_to_the_graph(self, monkeypatch):
from deerflow_extension_api import canonical_hash
from deerflow.agents.lead_agent import agent as lead_agent_module
from deerflow.agents.lead_agent.agent import assemble_lead_agent
from deerflow.config.agents_config import AgentConfig
from deerflow.extensions import bind_agent_build_extensions
self._isolate_from_the_ambient_config(monkeypatch)
agent_config = AgentConfig(name="custom", allowed_subagents=["general-purpose"])
monkeypatch.setattr(lead_agent_module, "load_agent_config", lambda name, *, user_id=None: agent_config)
prompt_calls = []
def render_prompt(**kwargs):
prompt_calls.append(kwargs)
return f"allowed_subagents={kwargs['allowed_subagents']}"
monkeypatch.setattr(lead_agent_module, "apply_prompt_template", render_prompt)
with bind_agent_build_extensions(self._extensions_with_an_agent_assembly_observer()):
assembly = assemble_lead_agent(
{
"configurable": {
"thread_id": "t-scoped-prompt",
"agent_name": "custom",
"subagent_enabled": True,
}
}
)
assert len(prompt_calls) == 1
assert prompt_calls[0]["allowed_subagents"] == ["general-purpose"]
assert assembly.graph["system_prompt"] == "allowed_subagents=['general-purpose']"
assert assembly.descriptor.base_prompt_hash == canonical_hash(assembly.graph["system_prompt"])
def test_descriptor_subagent_policy_respects_custom_agent_allowed_subagents(self, monkeypatch):
"""Fixes #5205: custom agent assembly descriptor must restrict its
subagents policy allowlist and runtime limits to allowed_subagents."""
from deerflow.agents.lead_agent import agent as lead_agent_module
from deerflow.agents.lead_agent.agent import assemble_lead_agent
from deerflow.config.agents_config import AgentConfig
from deerflow.extensions import bind_agent_build_extensions
self._isolate_from_the_ambient_config(monkeypatch)
agent_config = AgentConfig(name="custom", allowed_subagents=["general-purpose"])
monkeypatch.setattr(lead_agent_module, "load_agent_config", lambda name, *, user_id=None: agent_config)
with bind_agent_build_extensions(self._extensions_with_an_agent_assembly_observer()):
assembly = assemble_lead_agent(
{
"configurable": {
"thread_id": "t-scoped-subagents",
"agent_name": "custom",
"subagent_enabled": True,
}
}
)
subagent_policy = assembly.descriptor.effective_policies["subagents"]
assert subagent_policy["enabled"] is True
assert subagent_policy["type_allowlist"] == ["general-purpose"]
assert list(subagent_policy["runtime_limits"].keys()) == ["general-purpose"]
def test_observers_receive_the_descriptor(self, monkeypatch):
from deerflow.agents.lead_agent.agent import assemble_lead_agent
from deerflow.extensions import bind_agent_build_extensions
seen = []
class Observer:
def on_agent_assembled(self, app_store, descriptor):
seen.append(descriptor)
monkeypatch.setattr(
"deerflow.extensions.notify.notify_agent_assembled",
lambda descriptor, extensions=None: Observer().on_agent_assembled(None, descriptor),
)
self._isolate_from_the_ambient_config(monkeypatch)
with bind_agent_build_extensions(self._extensions_with_an_agent_assembly_observer()):
assemble_lead_agent({"configurable": {"thread_id": "t-2"}})
assert len(seen) == 1
def test_no_descriptor_is_built_without_a_registered_observer(self, monkeypatch):
"""The zero-observer fast path must skip the expensive build entirely,
not just skip notifying — mirroring notify_agent_assembled's own
zero-observer short-circuit."""
from deerflow.agents.lead_agent.agent import assemble_lead_agent
def _fail(*args, **kwargs):
raise AssertionError("build_assembly_descriptor must not run without an observer")
monkeypatch.setattr("deerflow.agents.assembly_descriptor.build_assembly_descriptor", _fail)
self._isolate_from_the_ambient_config(monkeypatch)
assembly = assemble_lead_agent({"configurable": {"thread_id": "t-3"}})
assert assembly.descriptor is None
class TestFactoryConsumersUnwrapTheGraph:
"""A missed unwrap fails at request time, not at import time."""
def test_worker_unwraps_the_assembly(self):
from deerflow.agents.lead_agent.agent import LeadAgentAssembly
from deerflow.runtime.runs.worker import _agent_graph
graph = object()
assert _agent_graph(LeadAgentAssembly(graph=graph, descriptor=object())) is graph
def test_worker_leaves_a_third_party_bare_graph_alone(self):
from deerflow.runtime.runs.worker import _agent_graph
graph = object()
assert _agent_graph(graph) is graph
class TestAssemblyObserverHost:
def test_registration_survives_rollback_of_a_later_install(self):
from deerflow.extensions.registry import ExtensionRegistry
class Observer:
def on_agent_assembled(self, app_store, descriptor):
return None
registry = ExtensionRegistry()
keeper = Observer()
with registry.attributed_to("keeper"):
registry.agent_assembly_observer(keeper)
mark = registry.mark()
with registry.attributed_to("doomed"):
registry.agent_assembly_observer(Observer())
registry.rollback_to(mark)
loaded = registry.build()
assert loaded.agent_assembly_observers == (("keeper", keeper),)
assert loaded.has_agent_assembly_observers is True
# The descriptor is app-scoped; it does not create a task store need.
assert loaded.needs_task_store is False
def test_a_broken_observer_does_not_stop_its_successors(self, caplog):
from deerflow.extensions.notify import notify_agent_assembled
from deerflow.extensions.registry import ExtensionRegistry
seen = []
class Broken:
def on_agent_assembled(self, app_store, descriptor):
raise RuntimeError("boom")
class Working:
def on_agent_assembled(self, app_store, descriptor):
seen.append(descriptor)
registry = ExtensionRegistry()
with registry.attributed_to("broken"):
registry.agent_assembly_observer(Broken())
with registry.attributed_to("working"):
registry.agent_assembly_observer(Working())
loaded = registry.build()
notify_agent_assembled("descriptor", loaded)
assert seen == ["descriptor"]
class TestBuildIdentityIsOutsideTheFingerprint:
"""A redeploy that changed nothing must not look like an assembly change."""
def _descriptor(self, build):
return AgentAssemblyDescriptor(
namespace="lead",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_parameters={},
thinking_enabled=False,
reasoning_effort=None,
base_prompt_hash="abc",
tools=(),
middlewares=(),
deferred_tool_names=(),
enabled_skills=(),
effective_policies={},
build=build,
)
def test_two_builds_of_the_same_assembly_share_a_fingerprint(self):
before = self._descriptor({"package_version": "1.0.0", "git_commit": "aaaa", "image_digest": "sha256:aaa"})
after = self._descriptor({"package_version": "1.0.1", "git_commit": "bbbb", "image_digest": "sha256:bbb"})
assert before.fingerprint == after.fingerprint
def test_the_build_itself_stays_comparable(self):
"""The coarser question must remain answerable, just separately."""
before = self._descriptor({"git_commit": "aaaa"})
after = self._descriptor({"git_commit": "bbbb"})
assert before.build != after.build
def test_the_builder_reports_a_build_without_hashing_it(self):
from deerflow.agents.assembly_descriptor import build_assembly_descriptor
def make():
return build_assembly_descriptor(
namespace="deerflow",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_config=None,
thinking_enabled=False,
reasoning_effort=None,
rendered_base_prompt="prompt",
tools=[],
middlewares=[],
deferred_names=frozenset(),
enabled_skills=[],
effective_policies={},
)
descriptor = make()
assert descriptor.build["package_version"]
assert "build" not in descriptor.effective_policies
assert descriptor.fingerprint == make().fingerprint
class TestWrappedExtensionMiddlewaresStayDistinguishable:
"""Contributed middlewares all share the isolation wrapper's class name."""
@staticmethod
def _wrap(inner, source):
from deerflow.extensions.isolation import IsolatedMiddleware
return IsolatedMiddleware(inner, source, lambda diagnostic: None)
@staticmethod
def _inner(name, *, policy=None):
from langchain.agents.middleware import AgentMiddleware
namespace = {}
if policy is not None:
namespace["release_policy_parameters"] = lambda self: dict(policy)
return type(name, (AgentMiddleware,), namespace)()
def test_two_extensions_middlewares_do_not_collapse_into_one_descriptor(self):
from deerflow.agents.assembly_descriptor import describe_middleware
first = describe_middleware(self._wrap(self._inner("AlphaMiddleware"), "ext-a"))
second = describe_middleware(self._wrap(self._inner("BetaMiddleware"), "ext-b"))
assert first.name == "AlphaMiddleware"
assert second.name == "BetaMiddleware"
assert first.extension == "ext-a"
assert second.extension == "ext-b"
assert first != second
def test_a_wrapped_declaration_reaches_the_descriptor(self):
from deerflow.agents.assembly_descriptor import describe_middleware
descriptor = describe_middleware(self._wrap(self._inner("DeclaringMiddleware", policy={"limit": 7}), "ext-a"))
assert descriptor.policy_parameters == {"limit": 7}
assert "probed" not in descriptor.policy_parameters
def test_a_policy_change_inside_a_wrapped_middleware_moves_the_fingerprint(self):
from dataclasses import replace
from deerflow.agents.assembly_descriptor import describe_middleware
def descriptor_for(limit):
return AgentAssemblyDescriptor(
namespace="lead",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_parameters={},
thinking_enabled=False,
reasoning_effort=None,
base_prompt_hash="abc",
tools=(),
middlewares=(describe_middleware(self._wrap(self._inner("DeclaringMiddleware", policy={"limit": limit}), "ext-a")),),
deferred_tool_names=(),
enabled_skills=(),
effective_policies={},
)
assert descriptor_for(1).fingerprint != descriptor_for(2).fingerprint
# And the same policy from a different extension is a different agent.
base = descriptor_for(1)
other = replace(base, middlewares=(replace(base.middlewares[0], extension="ext-b"),))
assert base.fingerprint != other.fingerprint
def test_an_unwrapped_host_middleware_reports_no_extension(self):
from deerflow.agents.assembly_descriptor import describe_middleware
descriptor = describe_middleware(self._inner("HostMiddleware", policy={"limit": 1}))
assert descriptor.extension is None
assert descriptor.name == "HostMiddleware"
class TestModelParametersProjectEffectiveSettings:
"""Provider kwargs a user actually sets must move the fingerprint.
``ModelConfig`` is ``extra="allow"``, so ``temperature``/``max_tokens``/
anything else a deployer sets live only as extra fields; a fixed allowlist
never saw them. And the *effective* per-agent override (issue #4336's
``model_settings``) must reach the descriptor too, not just the static
profile.
"""
@staticmethod
def _build(model_config, *, model_overrides=None):
from deerflow.agents.assembly_descriptor import build_assembly_descriptor
return build_assembly_descriptor(
namespace="deerflow",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_config=model_config,
model_overrides=model_overrides,
thinking_enabled=False,
reasoning_effort=None,
rendered_base_prompt="prompt",
tools=[],
middlewares=[],
deferred_names=frozenset(),
enabled_skills=[],
effective_policies={},
)
@staticmethod
def _model_config(**extra):
from deerflow.config.model_config import ModelConfig
return ModelConfig(
name="assembly-test-model",
display_name=None,
description=None,
use="langchain_openai:ChatOpenAI",
model="gpt-x",
**extra,
)
def test_changing_temperature_changes_the_fingerprint(self):
cold = self._build(self._model_config(temperature=0.1))
hot = self._build(self._model_config(temperature=0.9))
assert cold.fingerprint != hot.fingerprint
assert cold.model_parameters["temperature"] == 0.1
def test_changing_a_per_agent_model_setting_changes_the_fingerprint(self):
"""The *effective* override, not just the static profile, must count."""
model_config = self._model_config()
without_override = self._build(model_config)
with_override = self._build(model_config, model_overrides={"temperature": 0.7})
assert without_override.fingerprint != with_override.fingerprint
assert with_override.model_parameters["temperature"] == 0.7
def test_a_none_valued_override_does_not_clobber_the_profile(self):
model_config = self._model_config(temperature=0.3)
profile_only = self._build(model_config)
with_noop_override = self._build(model_config, model_overrides={"temperature": None})
assert profile_only.fingerprint == with_noop_override.fingerprint
def test_api_key_is_never_projected_and_never_moves_the_fingerprint(self):
quiet = self._build(self._model_config(api_key="sk-aaaaaaaaaaaa"))
loud = self._build(self._model_config(api_key="sk-bbbbbbbbbbbb"))
assert "api_key" not in quiet.model_parameters
assert quiet.fingerprint == loud.fingerprint
def test_an_override_named_like_a_credential_is_also_excluded(self):
model_config = self._model_config()
descriptor = self._build(model_config, model_overrides={"api_key": "sk-should-not-appear"})
assert "api_key" not in descriptor.model_parameters
class TestCustomAgentModelSettingsReachTheDescriptor:
"""End-to-end: a custom agent's ``model_settings`` must move the fingerprint.
``agent.py`` computes ``agent_model_overrides`` from ``agent_config.model_settings``
and passes it into ``create_chat_model``; this checks it also reaches
``_complete_assembly`` -> ``build_assembly_descriptor`` for the default
(non-bootstrap) assembly branch. Composes with (rather than subclasses)
``TestLeadAgentAssembly``'s isolation helpers so this class's own tests are
the only ones that run under it.
"""
def test_temperature_override_on_a_custom_agent_changes_the_fingerprint(self, monkeypatch):
from deerflow.agents.lead_agent import agent as lead_agent_module
from deerflow.agents.lead_agent.agent import assemble_lead_agent
from deerflow.config.agents_config import AgentConfig, AgentModelSettings
from deerflow.extensions import bind_agent_build_extensions
TestLeadAgentAssembly._isolate_from_the_ambient_config(monkeypatch)
def assemble(temperature):
agent_config = AgentConfig(name="custom", model_settings=AgentModelSettings(temperature=temperature))
monkeypatch.setattr(lead_agent_module, "load_agent_config", lambda name, user_id=None: agent_config)
with bind_agent_build_extensions(TestLeadAgentAssembly._extensions_with_an_agent_assembly_observer()):
return assemble_lead_agent({"configurable": {"thread_id": "t-model-settings", "agent_name": "custom"}})
low = assemble(0.1)
high = assemble(0.9)
assert low.descriptor.fingerprint != high.descriptor.fingerprint
assert high.descriptor.model_parameters["temperature"] == 0.9
def test_bootstrap_assembly_does_not_invent_model_overrides(self, monkeypatch):
"""The bootstrap branch has no ``agent_config``, so no overrides exist to project."""
from deerflow.agents.lead_agent.agent import assemble_lead_agent
from deerflow.extensions import bind_agent_build_extensions
TestLeadAgentAssembly._isolate_from_the_ambient_config(monkeypatch)
with bind_agent_build_extensions(TestLeadAgentAssembly._extensions_with_an_agent_assembly_observer()):
assembly = assemble_lead_agent({"configurable": {"thread_id": "t-bootstrap", "is_bootstrap": True}})
assert "temperature" not in assembly.descriptor.model_parameters
class TestSkillCatalogHashesContent:
"""Editing SKILL.md changes what ``SkillActivationMiddleware`` injects into
the turn, so it must change the fingerprint even though name/description/
allowed-tools are untouched."""
@staticmethod
def _skill(skill_dir: Path, *, required_secrets=(), secrets_autonomous=True):
from deerflow.skills.types import Skill, SkillCategory
skill_file = skill_dir / "SKILL.md"
return Skill(
name="my-skill",
description="A test skill",
license=None,
skill_dir=skill_dir,
skill_file=skill_file,
relative_path=Path(skill_dir.name),
category=SkillCategory.CUSTOM,
required_secrets=required_secrets,
secrets_autonomous=secrets_autonomous,
)
@staticmethod
def _build(enabled_skills):
from deerflow.agents.assembly_descriptor import build_assembly_descriptor
return build_assembly_descriptor(
namespace="deerflow",
agent_name="lead-agent",
requested_model=None,
effective_model="gpt-x",
model_config=None,
thinking_enabled=False,
reasoning_effort=None,
rendered_base_prompt="prompt",
tools=[],
middlewares=[],
deferred_names=frozenset(),
enabled_skills=enabled_skills,
effective_policies={},
)
def test_changing_only_skill_md_content_changes_the_fingerprint(self, tmp_path):
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
skill = self._skill(skill_dir)
(skill_dir / "SKILL.md").write_text("---\nname: my-skill\n---\nOriginal instructions.\n", encoding="utf-8")
before = self._build([skill])
(skill_dir / "SKILL.md").write_text("---\nname: my-skill\n---\nCompletely different instructions.\n", encoding="utf-8")
after = self._build([skill])
assert before.fingerprint != after.fingerprint
assert before.enabled_skills == after.enabled_skills # the catalog's visible identity is unchanged
def test_required_secrets_flag_changes_the_fingerprint(self):
from deerflow.skills.types import SecretRequirement
no_secrets = self._skill(Path("/nonexistent/skill-a"))
with_secret = self._skill(Path("/nonexistent/skill-b"), required_secrets=(SecretRequirement(name="API_KEY"),))
assert self._build([no_secrets]).fingerprint != self._build([with_secret]).fingerprint
def test_a_missing_skill_file_is_undescribable_not_fatal(self, tmp_path):
skill = self._skill(tmp_path / "missing-skill")
descriptor = self._build([skill])
assert descriptor.fingerprint