deer-flow/backend/packages/harness/deerflow/agents/assembly_descriptor.py

"""Projection of an assembled agent into a comparable descriptor.

The factory knows things nothing downstream can recover: which model survived
the runtime overrides, what the rendered prompt actually said, which tools
authorization left in place, and the order the middleware stack ended up in.
This module turns that transient knowledge into
:class:`~deerflow_extension_api.assembly.AgentAssemblyDescriptor`.

Two rules shape the projection:

* **Declared beats probed.** A middleware that implements
  ``release_policy_parameters()`` owns its own behaviour identity; probing
  private attributes is the fallback for the ones that do not, and is marked as
  such so a reader can tell a contract from a guess.
* **Hash, do not copy.** Prompts, tool descriptions, and argument schemas are
  reduced to hashes. A descriptor is an identity, not a second copy of the
  agent's payload.
"""

from __future__ import annotations

import logging
import os
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path

from deerflow_extension_api import (
    AgentAssemblyDescriptor,
    MiddlewareDescriptor,
    ToolDescriptor,
    canonical_hash,
    collect_release_policies,
)

from deerflow.sandbox.env_policy import is_blocked_env_name
from deerflow.tools.mcp_metadata import get_mcp_source, is_mcp_tool

logger = logging.getLogger(__name__)

# Fields on a model profile that are pure identity/presentation metadata: they
# never reach the provider constructor (see ``create_chat_model``'s own
# exclude set) and renaming/re-describing a model must not look like a
# behaviour change. ``use`` is surfaced separately as ``provider``.
_MODEL_METADATA_FIELDS = frozenset(
    {
        "name",
        "display_name",
        "description",
        "use",
        "context_window",
        "pricing",
    }
)
_MIDDLEWARE_PUBLIC_FIELDS = (
    "max_concurrent",
    "max_total",
    "warn_threshold",
    "hard_limit",
    "window_size",
    "max_tracked_threads",
    "tool_freq_warn",
    "tool_freq_hard_limit",
    "trigger",
    "keep",
    "trim_tokens_to_summarize",
    "fail_closed",
    "passport",
    "_tool_freq_overrides",
    "_top_k",
    "_deferred",
    "_catalog_hash",
)
_MIDDLEWARE_HASHED_TEXT_FIELDS = (
    "summary_prompt",
    "system_prompt",
    "tool_description",
)
_MODEL_IDENTITY_FIELDS = (
    "model",
    "model_name",
    "deployment_name",
)
_PROVIDER_PARAMETER_FIELDS = (
    "_allowed",
    "_denied",
    "_default_role",
    "_resource_type",
    "_action",
)
_DETECTOR_PARAMETER_FIELDS = (
    "_finish_reasons",
    "_stop_reasons",
)


def _plain_value(value: object) -> object | None:
    """Reduce ``value`` to JSON-shaped data, or ``None`` when it cannot be.

    ``None`` means "not describable", which is deliberately indistinguishable
    from a real ``None``: both are equally uninformative for identity, and
    inventing a marker would make two undescribable values look different.
    """
    if value is None or isinstance(value, bool | int | float | str):
        return value
    if isinstance(value, (list, tuple, set, frozenset)):
        children = [_plain_value(child) for child in value]
        if any(child is None and original is not None for child, original in zip(children, value, strict=True)):
            return None
        return sorted(children, key=str) if isinstance(value, (set, frozenset)) else children
    if isinstance(value, dict):
        result: dict[str, object] = {}
        for key, child in value.items():
            if not isinstance(key, str):
                continue
            plain = _plain_value(child)
            if plain is not None or child is None:
                result[key] = plain
        return result
    model_dump = getattr(value, "model_dump", None)
    if callable(model_dump):
        try:
            return _plain_value(model_dump(mode="python"))
        except Exception:
            return None
    return None


def _stable_type_name(value: object) -> str:
    value_type = type(value)
    return f"{value_type.__module__}.{value_type.__qualname__}"


def describe_model_identity(value: object) -> dict[str, str]:
    """Name a chat model without serialising it.

    Chat-model objects carry credentials and clients, so they are never plain
    data. What identifies them for comparison is the class plus the configured
    model name, unwrapped through any ``bound`` runnable wrapper the middleware
    stack layered on top.
    """
    original = value
    if isinstance(value, str):
        return {"class": "builtins.str", "name": value}
    resolved = value
    seen: set[int] = set()
    for _ in range(8):
        if not hasattr(resolved, "bound") or id(resolved) in seen:
            break
        seen.add(id(resolved))
        bound = getattr(resolved, "bound")
        if bound is None or bound is resolved:
            break
        resolved = bound
    identity = {"class": _stable_type_name(resolved)}
    for candidate in (resolved, original):
        for field_name in _MODEL_IDENTITY_FIELDS:
            field_value = getattr(candidate, field_name, None)
            if isinstance(field_value, str) and field_value:
                identity["name"] = field_value
                break
        if "name" in identity:
            break
    return identity


def _provider_identity(value: object) -> dict[str, object]:
    identity: dict[str, object] = {
        "class": _stable_type_name(value),
        "name": str(getattr(value, "name", type(value).__name__)),
    }
    parameters: dict[str, object] = {}
    for field_name in _PROVIDER_PARAMETER_FIELDS:
        if not hasattr(value, field_name):
            continue
        raw_value = getattr(value, field_name)
        plain = _plain_value(raw_value)
        if plain is not None or raw_value is None:
            parameters[field_name.removeprefix("_")] = plain
    if parameters:
        identity["parameters"] = parameters
    nested = getattr(value, "_provider", None)
    if nested is not None and nested is not value:
        identity["provider"] = _provider_identity(nested)
    return identity


def _tool_schema(tool: object) -> dict[str, object]:
    get_input_schema = getattr(tool, "get_input_schema", None)
    if callable(get_input_schema):
        try:
            schema_model = get_input_schema()
            schema = schema_model.model_json_schema()
            if isinstance(schema, dict):
                return schema
        except Exception:
            pass
    args_schema = getattr(tool, "args_schema", None)
    model_json_schema = getattr(args_schema, "model_json_schema", None)
    if callable(model_json_schema):
        try:
            schema = model_json_schema()
            if isinstance(schema, dict):
                return schema
        except Exception:
            pass
    return {}


def _tool_source(tool: object) -> str:
    if is_mcp_tool(tool):
        source = get_mcp_source(tool)
        return f"mcp:{source['server_name']}" if source is not None else "mcp:unknown"
    metadata = getattr(tool, "metadata", None)
    if isinstance(metadata, dict):
        declared = metadata.get("deerflow_tool_source")
        if isinstance(declared, str) and declared:
            return declared
    callable_object = getattr(tool, "func", None) or getattr(tool, "coroutine", None)
    module = getattr(callable_object, "__module__", "") or ""
    if module.startswith("deerflow.tools.builtins") or module.startswith("deerflow.agents.memory"):
        return "builtin"
    if "skill" in module:
        return "skill"
    return "community" if module else "builtin"


def describe_tool(tool: object) -> ToolDescriptor:
    """Project one bound tool into its identity."""
    source = get_mcp_source(tool) if is_mcp_tool(tool) else None
    return ToolDescriptor(
        name=str(getattr(tool, "name", type(tool).__name__)),
        description_hash=canonical_hash(str(getattr(tool, "description", "") or "")),
        schema_hash=canonical_hash(_plain_value(_tool_schema(tool))),
        source=_tool_source(tool),
        mcp_server=source["server_name"] if source is not None else None,
        mcp_transport=source["transport"] if source is not None else None,
    )


def _probe_middleware_parameters(middleware: object) -> dict[str, object]:
    """Best-effort identity for a middleware that declares none of its own."""
    parameters: dict[str, object] = {}
    for field_name in _MIDDLEWARE_PUBLIC_FIELDS:
        if not hasattr(middleware, field_name):
            continue
        raw_value = getattr(middleware, field_name)
        plain = _plain_value(raw_value)
        if plain is not None or raw_value is None:
            parameters[field_name.removeprefix("_")] = plain
    for field_name in _MIDDLEWARE_HASHED_TEXT_FIELDS:
        raw_value = getattr(middleware, field_name, None)
        if isinstance(raw_value, str):
            parameters[f"{field_name}_hash"] = canonical_hash(raw_value)
    model = getattr(middleware, "model", None)
    if model is not None:
        parameters["model"] = describe_model_identity(model)
    provider = getattr(middleware, "provider", None)
    if provider is not None:
        parameters["provider"] = _provider_identity(provider)
    routing_index = getattr(middleware, "_routing_index", None)
    plain_routing_index = _plain_value(routing_index)
    if plain_routing_index is not None:
        parameters["routing_index_hash"] = canonical_hash(plain_routing_index)
    detectors = getattr(middleware, "_detectors", None)
    if isinstance(detectors, (list, tuple)):
        detector_descriptors: list[dict[str, object]] = []
        for detector in detectors:
            descriptor: dict[str, object] = {
                "class": _stable_type_name(detector),
                "name": str(getattr(detector, "name", type(detector).__name__)),
            }
            detector_parameters: dict[str, object] = {}
            for field_name in _DETECTOR_PARAMETER_FIELDS:
                if not hasattr(detector, field_name):
                    continue
                raw_value = getattr(detector, field_name)
                plain = _plain_value(raw_value)
                if plain is not None or raw_value is None:
                    detector_parameters[field_name.removeprefix("_")] = plain
            if detector_parameters:
                descriptor["parameters"] = detector_parameters
            detector_descriptors.append(descriptor)
        parameters["detectors"] = detector_descriptors
    config = getattr(middleware, "_config", None)
    plain_config = _plain_value(config)
    if isinstance(plain_config, dict):
        parameters["config"] = plain_config
    return parameters


def _unwrap_middleware(middleware: object) -> tuple[object, str | None]:
    """Return the middleware that owns the behaviour, plus its extension.

    Extension contributions reach the stack inside an isolation wrapper whose
    dynamically generated subclass is named after the wrapper, not the
    contribution — so every contributed middleware in the process shares one
    class name and one (empty) probe result. Describing the wrapper would
    collapse them all into a single indistinguishable descriptor and hide any
    policy change inside them.

    Duck-typed on ``inner``/``source`` rather than importing the wrapper type:
    ``deerflow.extensions`` sits below this layer, so importing it here would
    point the dependency backwards, and any future wrapper of the same shape
    is handled for free.
    """
    described = middleware
    extension: str | None = None
    for _ in range(4):
        inner = getattr(described, "inner", None)
        if inner is None or inner is described:
            break
        source = getattr(described, "source", None)
        if not isinstance(source, str) or not source:
            source = getattr(described, "name", None)
        if isinstance(source, str) and source and extension is None:
            extension = source
        described = inner
    return described, extension


def describe_middleware(middleware: object) -> MiddlewareDescriptor:
    """Project one middleware, preferring its own declaration over probing."""
    described, extension = _unwrap_middleware(middleware)
    name = type(described).__name__
    declared = collect_release_policies([described])
    if name in declared:
        parameters = _plain_value(declared[name])
        if not isinstance(parameters, dict):
            logger.warning("%s declared a release policy that is not plain data; recording it as unserialisable", name)
            parameters = {"error": "UnserialisableDeclaration"}
    else:
        parameters = {"probed": True, **_probe_middleware_parameters(described)}
    return MiddlewareDescriptor(
        name=name,
        module=type(described).__module__,
        policy_parameters=parameters,
        extension=extension,
    )


def _build_identity() -> dict[str, str]:
    """Which build produced this assembly, when the deployment says so.

    Reported on its own descriptor field rather than inside
    ``effective_policies`` because the latter is hashed into the fingerprint:
    a redeploy that changes nothing about an agent must not change that
    agent's fingerprint.
    """
    try:
        package_version = version("deerflow-harness")
    except PackageNotFoundError:
        package_version = "unknown"
    return {
        "package_version": package_version,
        "image_digest": os.environ.get("DEER_FLOW_IMAGE_DIGEST", "unknown"),
        "git_commit": os.environ.get("DEER_FLOW_GIT_COMMIT", "unknown"),
    }


def _effective_model_fields(model_config: object) -> dict[str, object]:
    """All fields the model profile actually carries, declared or extra.

    ``ModelConfig`` is ``extra="allow"``, so a provider kwarg a user sets
    (``temperature``, ``max_tokens``, anything else) lives only as an extra
    field — a fixed allowlist would never see it. ``model_dump`` is the
    profile's own account of its fields, extras included, so it is preferred
    over probing named attributes. Falls back to plain instance attributes
    for a non-pydantic profile (e.g. the ``SimpleNamespace`` a subagent
    builds when its model name has no entry in the config table).
    """
    model_dump = getattr(model_config, "model_dump", None)
    if callable(model_dump):
        try:
            dumped = model_dump(mode="python")
        except Exception:
            dumped = None
        if isinstance(dumped, dict):
            return dumped
    namespace = getattr(model_config, "__dict__", None)
    return dict(namespace) if isinstance(namespace, dict) else {}


def _model_parameters(model_config: object, model_overrides: dict[str, object] | None = None) -> dict[str, object]:
    """The behaviour-affecting half of a model profile, as actually constructed.

    Built from the profile's own effective fields plus any per-caller
    overrides actually applied on top (e.g. a custom agent's ``model_settings``
    sampling overrides, or a request's runtime overrides) — mirroring what
    ``create_chat_model`` layers onto the constructor — rather than a fixed
    allowlist, so a changed ``temperature``/``max_tokens``/arbitrary provider
    kwarg is always visible here.

    Credential-shaped field names (``api_key`` and anything else
    ``is_blocked_env_name`` flags) are never projected, and values that
    cannot be reduced to plain JSON-shaped data are silently dropped rather
    than raising (see ``_plain_value``).

    ``thinking_enabled`` and ``reasoning_effort`` are descriptor fields in their
    own right, so they are deliberately not duplicated here.
    """
    use = getattr(model_config, "use", None)
    result: dict[str, object] = {"provider": str(use) if use else None}
    effective = _effective_model_fields(model_config)
    if model_overrides:
        effective = {**effective, **{key: value for key, value in model_overrides.items() if value is not None}}
    for field_name, value in effective.items():
        if not isinstance(field_name, str) or field_name in _MODEL_METADATA_FIELDS:
            continue
        if is_blocked_env_name(field_name):
            continue
        plain = _plain_value(value)
        if plain is not None or value is None:
            result[field_name] = plain
    return result


def _skill_content_hash(skill: object) -> str | None:
    """Digest of a skill's ``SKILL.md`` body.

    ``SkillActivationMiddleware`` injects this body as current-turn context
    (see ``skill_activation_middleware._read_skill_content``), so it is what
    actually drives the agent's behaviour when the skill is used — not just
    the name/description/allowed-tools already projected above. ``Skill``
    does not cache its content as a field (the middleware re-reads it from
    disk at activation time), so this does the same rather than inventing a
    cached-content field the rest of the system does not have. A skill whose
    file has gone missing or become unreadable is recorded as undescribable
    (``None``) rather than raising — assembly must not fail because a skill
    file disappeared out from under it.
    """
    skill_file = getattr(skill, "skill_file", None)
    if not isinstance(skill_file, Path):
        return None
    try:
        content = skill_file.read_text(encoding="utf-8")
    except OSError:
        return None
    return canonical_hash(content)


def build_assembly_descriptor(
    *,
    namespace: str,
    agent_name: str,
    requested_model: str | None,
    effective_model: str,
    model_config: object,
    model_overrides: dict[str, object] | None = None,
    thinking_enabled: bool,
    reasoning_effort: object,
    rendered_base_prompt: str,
    prompt_template_id: str = "deerflow-lead-agent-v1",
    tools: list[object],
    middlewares: list[object],
    deferred_names: frozenset[str],
    enabled_skills: list[object],
    effective_policies: dict[str, object],
) -> AgentAssemblyDescriptor:
    """Describe one finished assembly.

    ``tools`` is the list bound to the graph; middleware-owned tools are folded
    in because the model sees them exactly the same way. The skill catalog is
    hashed rather than listed field-by-field so editing a skill's body changes
    the fingerprint while the descriptor stays small.
    """
    skill_catalog = [
        {
            "name": str(getattr(skill, "name", "")),
            "description": str(getattr(skill, "description", "")),
            "allowed_tools": sorted(str(item) for item in (getattr(skill, "allowed_tools", None) or ())),
            "content_hash": _skill_content_hash(skill),
            "secrets_autonomous": bool(getattr(skill, "secrets_autonomous", True)),
            "required_secrets": sorted(f"{getattr(requirement, 'name', '')}:{bool(getattr(requirement, 'optional', False))}" for requirement in (getattr(skill, "required_secrets", None) or ())),
        }
        for skill in enabled_skills
    ]
    assembled_tools = list(tools)
    for middleware in middlewares:
        middleware_tools = getattr(middleware, "tools", None)
        if isinstance(middleware_tools, (list, tuple)):
            assembled_tools.extend(middleware_tools)

    resolved_policies = dict(effective_policies)
    resolved_policies["prompt_template_id"] = prompt_template_id
    resolved_policies["skill_catalog_hash"] = canonical_hash(sorted(skill_catalog, key=lambda item: item["name"]))

    return AgentAssemblyDescriptor(
        namespace=namespace,
        agent_name=agent_name,
        requested_model=requested_model,
        effective_model=effective_model,
        model_parameters=_model_parameters(model_config, model_overrides),
        thinking_enabled=thinking_enabled,
        reasoning_effort=_plain_value(reasoning_effort),
        base_prompt_hash=canonical_hash(rendered_base_prompt),
        tools=tuple(describe_tool(tool) for tool in assembled_tools),
        middlewares=tuple(describe_middleware(middleware) for middleware in middlewares),
        deferred_tool_names=tuple(sorted(str(name) for name in deferred_names)),
        enabled_skills=tuple(entry["name"] for entry in skill_catalog),
        effective_policies=resolved_policies,
        build=_build_identity(),
    )


__all__ = [
    "build_assembly_descriptor",
    "describe_middleware",
    "describe_model_identity",
    "describe_tool",
]