Agent 装配

def assemble_lead_agent(
    config: RunnableConfig,
    *,
    app_config: AppConfig | None = None,
) -> LeadAgentAssembly:
    """Return the compiled lead graph together with its assembly descriptor.

    Gateway workers use this explicit assembly result so what the agent was
    built from does not have to be recovered from LangGraph private runtime
    keys or mutable graph attributes. ``make_lead_agent`` remains the
    graph-only LangGraph Server ABI declared in ``langgraph.json``.
    """
    runtime_config = _get_runtime_config(config)
    runtime_app_config = app_config or runtime_config.get("app_config")
    if not isinstance(runtime_app_config, AppConfig):
        runtime_app_config = get_app_config()
    # Mode selection precedence, pinned by test_checkpoint_mode.py:
    # - First freeze: the app config owns the process mode; a client-supplied
    #   configurable key is ignored so a direct LangGraph request cannot
    #   reconfigure (or crash) a fresh process.
    # - Once frozen: an internally injected key (run worker / gateway) or the
    #   app config must match the frozen mode; ``freeze_checkpoint_channel_mode``
    #   fails closed on any mismatch, so neither a forged key nor a config.yaml
    #   change can silently reconfigure the process.
    frozen_mode = frozen_checkpoint_channel_mode()
    if frozen_mode is None:
        requested_mode = runtime_app_config.database.checkpoint_channel_mode
    else:
        requested_mode = (config.get("configurable", {}) or {}).get(
            INTERNAL_CHECKPOINT_MODE_KEY,
            runtime_app_config.database.checkpoint_channel_mode,
        )
    mode = freeze_checkpoint_channel_mode(requested_mode)
    # The snapshot cadence travels with the mode: restart-required, frozen
    # from the app config, and deliberately not client-injectable (a forged
    # configurable key must not recompile the channel table either).
    freeze_checkpoint_snapshot_frequency(runtime_app_config.database.checkpoint_delta.snapshot_frequency)
    inject_checkpoint_mode(config, mode)
    return _assemble_lead_agent(config, app_config=runtime_app_config)

final_tools, setup = assemble_deferred_tools(configured_tools, enabled=resolved_app_config.tool_search.enabled)
final_tools.extend(late_tools)
mcp_routing_middleware = build_mcp_routing_middleware(
    final_tools,
    setup,
    top_k=resolved_app_config.tool_search.auto_promote_top_k,
)
mcp_routing_hints_section = get_mcp_routing_hints_prompt_section(authorized_tools, deferred_names=setup.deferred_names)
middlewares = build_middlewares(
    config,
    model_name=model_name,
    agent_name=agent_name,
    available_skills=available_skills,
    app_config=resolved_app_config,
    deferred_setup=setup,
    mcp_routing_middleware=mcp_routing_middleware,
    user_id=resolved_user_id,
    authorization_provider=_authz_provider,
    subagent_execution_capacity=subagent_execution_capacity,
)
system_prompt = apply_prompt_template(
    subagent_enabled=subagent_enabled,
    max_concurrent_subagents=max_concurrent_subagents,
    max_total_subagents=max_total_subagents,
    agent_name=agent_name,
    available_skills=available_skills,
    app_config=resolved_app_config,
    deferred_names=setup.deferred_names,
    mcp_routing_hints_section=mcp_routing_hints_section,
    user_id=resolved_user_id,
    skill_names=skill_setup.skill_names or None,
    allowed_subagents=allowed_subagents,
    subagent_execution_capacity=subagent_execution_capacity,
)
graph = create_agent(
    model=create_chat_model(name=model_name, thinking_enabled=thinking_enabled, reasoning_effort=reasoning_effort, app_config=resolved_app_config, attach_tracing=False, model_overrides=agent_model_overrides),
    tools=final_tools,
    middleware=normalize_middleware_state_schemas(middlewares, mode),
    system_prompt=system_prompt,
    state_schema=get_thread_state_schema(mode),
)

# Layer 1 — outermost wrap_model_call wrappers (listed outer→inner).
# InputSanitizationMiddleware is first so it becomes the outermost
# wrapper — sanitised messages are what every inner middleware sees.
# ToolResultSanitizationMiddleware mirrors that guardrail for the other
# untrusted-content entry point: remote tool results (web_fetch /
# web_search) get the same framework/injection-tag neutralization. It sits
# inner of ToolOutputBudgetMiddleware (listed after it) so it neutralizes
# the raw tool output first; the budget wrapper then truncates the already
# neutralized text.
outer_wrappers: list[AgentMiddleware] = [
    InputSanitizationMiddleware(),
    ToolOutputBudgetMiddleware.from_app_config(app_config),
    ToolResultSanitizationMiddleware(),
]

# Layer 2 — before_agent hooks that read/annotate thread-scoped data.
thread_hooks: list[AgentMiddleware] = [
    ThreadDataMiddleware(lazy_init=lazy_init),
]
if include_uploads:
    from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware

    thread_hooks.append(UploadsMiddleware())
thread_hooks.append(SandboxMiddleware(lazy_init=lazy_init))

# Layer 3 — post-processing append-only middlewares.
tail: list[AgentMiddleware] = []
if include_dangling_tool_call_patch:
    from deerflow.agents.middlewares.dangling_tool_call_middleware import DanglingToolCallMiddleware

    tail.append(DanglingToolCallMiddleware())
tail.append(LLMErrorHandlingMiddleware(app_config=app_config))

# ToolReceiptMiddleware is the outermost wrap_tool_call layer: Guardrail,
# SandboxAudit, ReadBeforeWrite, and ToolProgress can all short-circuit a
# call with their own ToolMessage, and SandboxAudit rebuilds medium-risk
# results — an inner receipt layer would miss those results and silently
# gap the ledger. Stamping out here still sees deerflow_tool_meta on
# normal results (ToolErrorHandling stamps it on the inner return path)
# and on self-stamped short-circuit messages; the remainder fall back to
# message.status (see make_tool_receipt).
verification_config = app_config.verification
if verification_config.receipts_enabled:
    from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware

    tail.append(ToolReceiptMiddleware(render_mode=receipts_render_mode))

# Authorization uses the existing GuardrailMiddleware so execution-time
# deny, audit, and fail-closed handling stay in one proven implementation.
# It is appended before an explicit guardrail provider, making authorization
# the outer guard and avoiding an unnecessary external policy call for an
# already-denied tool.
authorization_config = app_config.authorization
if authorization_config.enabled is True:
    if authorization_provider is None:
        from deerflow.authz.runtime import resolve_authorization_provider

        authorization_provider = resolve_authorization_provider(authorization_config)
    if authorization_provider is not None:
        from deerflow.authz.adapter import GuardrailAuthorizationAdapter
        from deerflow.guardrails.middleware import GuardrailMiddleware

        tail.append(
            GuardrailMiddleware(
                GuardrailAuthorizationAdapter(
                    authorization_provider,
                    default_role=authorization_config.default_role,
                    infrastructure_tool_names=authorization_infrastructure_tool_names,
                ),
                fail_closed=authorization_config.fail_closed,
            )
        )

# Explicit guardrail middleware remains independently active when configured.
guardrails_config = app_config.guardrails
if guardrails_config.enabled and guardrails_config.provider:
    import inspect

    from deerflow.guardrails.middleware import GuardrailMiddleware
    from deerflow.reflection import resolve_variable

    provider_cls = resolve_variable(guardrails_config.provider.use)
    provider_kwargs = dict(guardrails_config.provider.config) if guardrails_config.provider.config else {}
    # Pass framework hint if the provider accepts it (e.g. for config discovery).
    # Built-in providers like AllowlistProvider don't need it, so only inject
    # when the constructor accepts 'framework' or '**kwargs'.
    if "framework" not in provider_kwargs:
        try:
            sig = inspect.signature(provider_cls.__init__)
            if "framework" in sig.parameters or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()):
                provider_kwargs["framework"] = "deerflow"
        except (ValueError, TypeError):
            pass
    provider = provider_cls(**provider_kwargs)
    tail.append(GuardrailMiddleware(provider, fail_closed=guardrails_config.fail_closed, passport=guardrails_config.passport))

from deerflow.agents.middlewares.sandbox_audit_middleware import SandboxAuditMiddleware

tail.append(SandboxAuditMiddleware())

# ReadBeforeWriteMiddleware is the outermost write gate: it blocks writes to files
# the model hasn't read in their current version.  It must sit outside ToolProgress
# and ToolErrorHandling so that a blocked write returns immediately without consuming
# a ToolProgress slot.  The middleware stamps deerflow_tool_meta on the blocked
# ToolMessage itself so downstream callers receive a well-formed result.
if app_config.read_before_write.enabled:
    from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware

    tail.append(ReadBeforeWriteMiddleware())

# ToolProgressMiddleware must be outer (lower index) so its wrap_tool_call handler
# chain includes ToolErrorHandlingMiddleware (inner), which stamps deerflow_tool_meta
# on every result before ToolProgressMiddleware reads it in _update_state_from_result.
# Framework rule: first in list = outermost (types.py: "compose with first in list as outermost layer").
tool_progress_config = app_config.tool_progress
if tool_progress_config.enabled:
    from deerflow.agents.middlewares.tool_progress_middleware import ToolProgressMiddleware

    tail.append(ToolProgressMiddleware.from_config(tool_progress_config))

tail.append(ToolErrorHandlingMiddleware(app_config=app_config))

middlewares = [*outer_wrappers, *thread_hooks, *tail]

# Ordering invariants are declared in deerflow.extensions.ordering and
# validated once at the end of the composing builder, after extension
# contributions are merged in — otherwise a contribution could silently
# reverse an invariant this builder had already checked.
return middlewares

return (
    OrderingConstraint(
        outer=ToolProgressMiddleware,
        inner=ToolErrorHandlingMiddleware,
        reason=("ToolProgressMiddleware reads deerflow_tool_meta in _update_state_from_result, so its wrap_tool_call chain must enclose the ToolErrorHandlingMiddleware step that stamps it"),
    ),
    OrderingConstraint(
        outer=ToolReceiptMiddleware,
        inner=ToolErrorHandlingMiddleware,
        reason=("ToolReceiptMiddleware reads the deerflow_tool_meta status stamped by ToolErrorHandlingMiddleware when building each receipt, so its wrap_tool_call chain must enclose the stamping step"),
    ),
    *(
        OrderingConstraint(
            outer=ToolReceiptMiddleware,
            inner=short_circuiter,
            reason=(f"{short_circuiter.__name__} can return or rebuild a ToolMessage without invoking its handler; ToolReceiptMiddleware must wrap it or those results never get a receipt and the ledger silently gaps"),
        )
        for short_circuiter in (
            GuardrailMiddleware,
            SandboxAuditMiddleware,
            ReadBeforeWriteMiddleware,
            ToolProgressMiddleware,
        )
    ),
)

middlewares = build_lead_runtime_middlewares(**runtime_middleware_kwargs)

# Always inject current date (and optionally memory) as <system-reminder> into the
# first HumanMessage to keep the system prompt fully static for prefix-cache reuse.
from deerflow.agents.middlewares.dynamic_context_middleware import DynamicContextMiddleware

middlewares.append(DynamicContextMiddleware(agent_name=agent_name, app_config=resolved_app_config))

# Deterministically load a full SKILL.md when the user starts the turn with
# /skill-name. This keeps the base system prompt metadata-only while giving
# explicit user activation priority over model-side relevance guessing.
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware

slash_source_owner_token = secrets.token_urlsafe(24)
middlewares.append(
    SkillActivationMiddleware(
        available_skills=available_skills,
        app_config=resolved_app_config,
        user_id=user_id,
        slash_source_owner_token=slash_source_owner_token,
    )
)

# Enabled skills are only discoverable metadata. Apply allowed-tools at
# runtime after explicit slash activation or an actual skill-file load.
from deerflow.agents.middlewares.skill_tool_policy_middleware import SkillToolPolicyMiddleware

middlewares.append(
    SkillToolPolicyMiddleware(
        available_skills=available_skills,
        app_config=resolved_app_config,
        user_id=user_id,
        slash_source_owner_token=slash_source_owner_token,
    )
)

# Capture completed task delegations and loaded skill files before
# summarization can compact them, then inject durable context channels
# (summary + ledger + skills) into model calls.
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware

middlewares.append(
    DurableContextMiddleware(
        skills_container_path=resolved_app_config.skills.container_path,
        skill_file_read_tool_names=resolved_app_config.summarization.skill_file_read_tool_names,
    )
)

# Add summarization middleware if enabled
summarization_middleware = _create_summarization_middleware(
    app_config=resolved_app_config,
    run_model_name=model_name,
    extensions=resolved_extensions,
)
if summarization_middleware is not None:
    middlewares.append(summarization_middleware)

# Add TodoList middleware if plan mode is enabled
cfg = _get_runtime_config(config)
is_plan_mode = cfg.get("is_plan_mode", False)
todo_list_middleware = _create_todo_list_middleware(is_plan_mode)
if todo_list_middleware is not None:
    middlewares.append(todo_list_middleware)

# Add TokenUsageMiddleware when token_usage tracking is enabled
if resolved_app_config.token_usage.enabled:
    middlewares.append(TokenUsageMiddleware())

# Add TitleMiddleware
middlewares.append(
    TitleMiddleware(
        app_config=resolved_app_config,
        extensions=resolved_extensions,
    )
)

# Add MemoryMiddleware after TitleMiddleware. Tool mode normally skips it;
# conversation-extraction backends may explicitly retain passive writes.
if should_use_memory_tools(resolved_app_config.memory):
    from deerflow.agents.memory.manager import backend_requires_passive_writes_in_tool_mode

    if backend_requires_passive_writes_in_tool_mode(resolved_app_config.memory.manager_class):
        middlewares.append(MemoryMiddleware(agent_name=agent_name, memory_config=resolved_app_config.memory))
else:
    if resolved_app_config.memory.mode == "tool" and not resolved_app_config.memory.enabled:
        logger.warning("memory.mode is 'tool' but memory.enabled is false; memory tools will not be registered.")
    middlewares.append(MemoryMiddleware(agent_name=agent_name, memory_config=resolved_app_config.memory))

# Add ViewImageMiddleware only if the current model supports vision.
# Use the resolved runtime model_name from make_lead_agent to avoid stale config values.
model_config = resolved_app_config.get_model_config(model_name) if model_name else None
if model_config is not None and model_config.supports_vision:
    middlewares.append(ViewImageMiddleware())

# Auto-promote deferred MCP schemas from PR1 routing metadata before the
# deferred filter decides which schemas to hide for this model call.
if mcp_routing_middleware is not None:
    middlewares.append(mcp_routing_middleware)

# Hide deferred tool schemas from model binding until tool_search promotes them.
# The lead deferred set + catalog hash come from the full build-time MCP
# catalog; SkillToolPolicyMiddleware separately filters model visibility,
# tool_search results, and execution for the active skill at runtime.
if deferred_setup is not None and deferred_setup.deferred_names:
    from deerflow.agents.middlewares.deferred_tool_filter_middleware import DeferredToolFilterMiddleware

    middlewares.append(DeferredToolFilterMiddleware(deferred_setup.deferred_names, deferred_setup.catalog_hash))
    from deerflow.agents.middlewares.mcp_routing_middleware import assert_mcp_routing_before_deferred_filter

    assert_mcp_routing_before_deferred_filter(middlewares)

# Coalesce every SystemMessage into a single leading one before the request
# reaches the provider. Strict backends (vLLM, SGLang, Qwen, Anthropic)
# reject non-leading SystemMessages. See system_message_coalescing_middleware.py.
from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware

middlewares.append(SystemMessageCoalescingMiddleware())