deer-flow/backend/tests/test_agent_assembly_descriptor.py

"""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 TestShelfIndexReleasePolicy:
    """The shelf rendering caps change the model-visible ``<documents>`` block,
    so DynamicContextMiddleware declares both effective limits and each change
    moves the assembly fingerprint."""

    @staticmethod
    def _params(**projects_kwargs):
        from deerflow.agents.middlewares.dynamic_context_middleware import DynamicContextMiddleware
        from deerflow.config.app_config import AppConfig
        from deerflow.config.model_config import ModelConfig
        from deerflow.config.projects_config import ProjectsConfig
        from deerflow.config.sandbox_config import SandboxConfig

        config = AppConfig(
            models=[ModelConfig(name="m", display_name="m", description=None, use="langchain_openai:ChatOpenAI", model="m", supports_thinking=False, supports_vision=False)],
            projects=ProjectsConfig(**projects_kwargs),
            sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
        )
        return DynamicContextMiddleware(app_config=config).release_policy_parameters()

    @staticmethod
    def _descriptor_with(policy):
        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=(MiddlewareDescriptor(name="DynamicContextMiddleware", module="m", policy_parameters=dict(policy)),),
            deferred_tool_names=(),
            enabled_skills=(),
            effective_policies={},
        )

    def test_both_effective_shelf_limits_are_declared(self):
        from deerflow.config.projects_config import ProjectsConfig

        params = self._params()
        assert params["shelf_index_max_entries"] == ProjectsConfig().shelf_index_max_entries
        assert params["shelf_index_max_bytes"] == ProjectsConfig().shelf_index_max_bytes

    def test_changing_either_shelf_limit_moves_the_fingerprint(self):
        base = self._params()
        by_entries = self._params(shelf_index_max_entries=99)
        by_bytes = self._params(shelf_index_max_bytes=8192)
        # Each knob moves independently.
        assert by_entries["shelf_index_max_entries"] == 99
        assert by_entries["shelf_index_max_bytes"] == base["shelf_index_max_bytes"]
        assert by_bytes["shelf_index_max_bytes"] == 8192
        assert by_bytes["shelf_index_max_entries"] == base["shelf_index_max_entries"]
        base_descriptor = self._descriptor_with(base)
        assert base_descriptor.fingerprint != self._descriptor_with(by_entries).fingerprint
        assert base_descriptor.fingerprint != self._descriptor_with(by_bytes).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.graph["context_schema"] is dict
        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_native_reasoning_setting_is_projected_and_moves_the_fingerprint(self):
        """``reasoning`` is a real provider kwarg for ChatOllama (bool or level string), so
        flipping it changes requests and must change the fingerprint (PR #5780 review)."""
        on = self._build(self._model_config(reasoning=True))
        off = self._build(self._model_config(reasoning=False))
        high = self._build(self._model_config(reasoning="high"))
        assert on.fingerprint != off.fingerprint
        assert on.fingerprint != high.fingerprint
        assert on.model_parameters["reasoning"] is True
        assert high.model_parameters["reasoning"] == "high"

    def test_declared_reasoning_contract_is_projected_and_moves_the_fingerprint(self):
        """A contract's dialect/history change the request payload, so they belong in the fingerprint."""

        def contract(history):
            return {"thinking": "optional", "dialect": "openai_extra_body", "history": history}

        clear = self._build(self._model_config(reasoning=contract("clear")))
        preserve = self._build(self._model_config(reasoning=contract("preserve")))
        assert clear.fingerprint != preserve.fingerprint
        assert clear.model_parameters["reasoning"]["history"] == "clear"
        assert clear.model_parameters["reasoning"]["dialect"] == "openai_extra_body"

    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 assembly.graph["context_schema"] is dict
        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, *, allowed_tools=None, 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,
            allowed_tools=allowed_tools,
            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_allowed_tools_declaration_states_have_distinct_fingerprints(self, tmp_path):
        descriptors = [self._build([self._skill(tmp_path, allowed_tools=allowed_tools)]) for allowed_tools in (None, (), ("bash",))]

        assert len({descriptor.fingerprint for descriptor in descriptors}) == 3

    def test_allowed_tools_order_does_not_change_the_fingerprint(self, tmp_path):
        before = self._build([self._skill(tmp_path, allowed_tools=("bash", "read_file"))])
        reordered = self._build([self._skill(tmp_path, allowed_tools=("read_file", "bash"))])

        assert before.fingerprint == reordered.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


class TestProjectDocumentToolRegistration:
    """The two shelf tools exist only in project runs (spec §7.3, §10.11).

    Registration follows the admission-pinned ``PROJECT_CONTEXT_KEY`` and
    nothing else: a run assembled without the key never carries the tools'
    schemas, a run with it always does — even when instructions and shelf
    are both empty. Since the tool list feeds ``build_assembly_descriptor``,
    both variants are pinned here (descriptor and fingerprint).
    """

    @staticmethod
    def _assemble(monkeypatch, config):
        from deerflow.agents.lead_agent.agent import assemble_lead_agent
        from deerflow.extensions import bind_agent_build_extensions

        helpers = TestLeadAgentAssembly
        helpers._isolate_from_the_ambient_config(monkeypatch)
        with bind_agent_build_extensions(helpers._extensions_with_an_agent_assembly_observer()):
            return assemble_lead_agent(config)

    def test_project_run_registers_both_tools(self, monkeypatch):
        from deerflow.runtime.context_keys import PROJECT_CONTEXT_KEY

        assembly = self._assemble(
            monkeypatch,
            {
                "configurable": {"thread_id": "t-proj"},
                "context": {PROJECT_CONTEXT_KEY: {"project_id": "p-1", "name": "P", "instructions": "ctx"}},
            },
        )
        tool_names = {getattr(tool, "name", None) for tool in assembly.graph["tools"]}
        assert {"list_project_documents", "read_project_document"} <= tool_names
        descriptor_names = {tool.name for tool in assembly.descriptor.tools}
        assert {"list_project_documents", "read_project_document"} <= descriptor_names

    def test_non_project_run_registers_neither_tool(self, monkeypatch):
        assembly = self._assemble(monkeypatch, {"configurable": {"thread_id": "t-plain"}})
        tool_names = {getattr(tool, "name", None) for tool in assembly.graph["tools"]}
        assert "list_project_documents" not in tool_names
        assert "read_project_document" not in tool_names
        descriptor_names = {tool.name for tool in assembly.descriptor.tools}
        assert "list_project_documents" not in descriptor_names
        assert "read_project_document" not in descriptor_names

    def test_both_descriptor_variants_are_distinct_and_stable(self, monkeypatch):
        from deerflow.runtime.context_keys import PROJECT_CONTEXT_KEY

        with_project = self._assemble(
            monkeypatch,
            {
                "configurable": {"thread_id": "t-proj"},
                "context": {PROJECT_CONTEXT_KEY: {"project_id": "p-1", "name": "P", "instructions": ""}},
            },
        )
        without_project = self._assemble(monkeypatch, {"configurable": {"thread_id": "t-plain"}})
        assert with_project.descriptor.fingerprint != without_project.descriptor.fingerprint

        again = self._assemble(
            monkeypatch,
            {
                "configurable": {"thread_id": "t-proj"},
                "context": {PROJECT_CONTEXT_KEY: {"project_id": "p-1", "name": "P", "instructions": ""}},
            },
        )
        assert again.descriptor.fingerprint == with_project.descriptor.fingerprint

    def test_registration_follows_the_key_when_blocks_are_empty(self, monkeypatch):
        """Empty instructions and an empty shelf still mean a project run."""
        from deerflow.runtime.context_keys import PROJECT_CONTEXT_KEY

        assembly = self._assemble(
            monkeypatch,
            {
                "configurable": {"thread_id": "t-empty"},
                "context": {PROJECT_CONTEXT_KEY: {"project_id": "p-1", "name": "P", "instructions": "", "shelf": {"total": 0, "entries": []}}},
            },
        )
        tool_names = {getattr(tool, "name", None) for tool in assembly.graph["tools"]}
        assert {"list_project_documents", "read_project_document"} <= tool_names