"""Primary graph assembly module for Deep Agents.

Provides [`create_deep_agent`][deepagents.graph.create_deep_agent], the main entry
point for constructing a fully configured deep agent with planning, filesystem,
subagent, and summarization middleware.
"""

import logging
from collections.abc import Callable, Sequence
from importlib import import_module
from typing import Annotated, Any, Required, cast

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware, InterruptOnConfig, TodoListMiddleware
from langchain.agents.middleware.types import AgentMiddleware, InputAgentState, OutputAgentState, ResponseT
from langchain.agents.structured_output import ResponseFormat
from langchain_anthropic import ChatAnthropic
from langchain_anthropic.middleware import AnthropicPromptCachingMiddleware
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AnyMessage, SystemMessage
from langchain_core.tools import BaseTool
from langgraph.cache.base import BaseCache
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.state import CompiledStateGraph
from langgraph.store.base import BaseStore
from langgraph.types import Checkpointer
from langgraph.typing import ContextT

from deepagents._api.deprecation import deprecated, warn_deprecated
from deepagents._excluded_middleware import (
    _apply_excluded_middleware,
    _validate_excluded_middleware_config,
    _verify_excluded_middleware_coverage,
)
from deepagents._messages_reducer import _messages_delta_reducer
from deepagents._models import resolve_model
from deepagents._tools import _apply_tool_description_overrides
from deepagents._version import __version__
from deepagents.backends import StateBackend
from deepagents.backends.protocol import BackendFactory, BackendProtocol
from deepagents.middleware._fs_interrupt import _build_interrupt_on_from_permissions
from deepagents.middleware._state import private_state_field_names
from deepagents.middleware._tool_exclusion import _ToolExclusionMiddleware
from deepagents.middleware.async_subagents import AsyncSubAgent, AsyncSubAgentMiddleware
from deepagents.middleware.filesystem import FilesystemMiddleware, FilesystemPermission
from deepagents.middleware.memory import MemoryMiddleware
from deepagents.middleware.patch_tool_calls import PatchToolCallsMiddleware
from deepagents.middleware.skills import SkillsMiddleware
from deepagents.middleware.subagents import (
    GENERAL_PURPOSE_SUBAGENT,
    CompiledSubAgent,
    SubAgent,
    SubAgentMiddleware,
)
from deepagents.middleware.summarization import create_summarization_middleware
from deepagents.profiles.harness.harness_profiles import (
    GeneralPurposeSubagentProfile,
    _apply_profile_prompt,
    _harness_profile_for_model,
)

logger = logging.getLogger(__name__)


class DeepAgentState(AgentState):
    """AgentState with `DeltaChannel` on messages to reduce checkpoint growth from O(N²) to O(N)."""

    messages: Required[Annotated[list[AnyMessage], DeltaChannel(_messages_delta_reducer, snapshot_frequency=50)]]  # ty: ignore[invalid-argument-type]


BASE_AGENT_PROMPT = """You are a deep agent, an AI assistant that helps users accomplish tasks using tools. You respond with text and tool calls. The user can see your responses and tool outputs in real time.

## Core Behavior

- Be concise and direct. Don't over-explain unless asked.
- NEVER add unnecessary preamble (\"Sure!\", \"Great question!\", \"I'll now...\").
- Don't say \"I'll now do X\" — just do it.
- If the request is underspecified, ask only the minimum followup needed to take the next useful action.
- If asked how to approach something, explain first, then act.

## Professional Objectivity

- Prioritize accuracy over validating the user's beliefs
- Disagree respectfully when the user is incorrect
- Avoid unnecessary superlatives, praise, or emotional validation

## Doing Tasks

When the user asks you to do something:

1. **Understand first** — read relevant files, check existing patterns. Quick but thorough — gather enough evidence to start, then iterate.
2. **Act** — implement the solution. Work quickly but accurately.
3. **Verify** — check your work against what was asked, not against your own output. Your first attempt is rarely correct — iterate.

Keep working until the task is fully complete. Don't stop partway and explain what you would do — just do it. Only yield back to the user when the task is done or you're genuinely blocked.

**When things go wrong:**

- If something fails repeatedly, stop and analyze *why* — don't keep retrying the same approach.
- If you're blocked, tell the user what's wrong and ask for guidance.

## Clarifying Requests

- Do not ask for details the user already supplied.
- Use reasonable defaults when the request clearly implies them.
- Prioritize missing semantics like content, delivery, detail level, or alert criteria.
- Avoid opening with a long explanation of tool, scheduling, or integration limitations when a concise blocking followup question would move the task forward.
- Ask domain-defining questions before implementation questions.
- For monitoring or alerting requests, ask what signals, thresholds, or conditions should trigger an alert.

## Progress Updates

For longer tasks, provide brief progress updates at reasonable intervals — a concise sentence recapping what you've done and what's next."""  # noqa: E501
"""Default base system prompt for every deep agent (`BASE`).

The final system prompt sent to the model is composed from up to four
named parts:

- `USER` — the `system_prompt=` argument to `create_deep_agent` (`str` or
    `SystemMessage`); when unset, no `USER` segment is included.
- `BASE` — this constant.
- `CUSTOM` — `HarnessProfile.base_system_prompt`. When set on a matching
    profile, replaces `BASE` outright; when unset, `BASE` is used.
- `SUFFIX` — `HarnessProfile.system_prompt_suffix`. When set on a
    matching profile, appended last; when unset, no `SUFFIX` segment is
    included.

The order is always `USER` -> (`BASE` or `CUSTOM`) -> `SUFFIX`, joined by
blank lines (`\\n\\n`). Two invariants follow:

1. `USER` is always at the front, so caller instructions take precedence
    over SDK and profile content regardless of which model is selected.
2. `SUFFIX` is always at the end, so model-tuning guidance sits closest
    to the conversation history (where the model attends most).

When `USER` is a `SystemMessage`, the right-hand assembly is appended as
an additional text content block onto the message's existing
`content_blocks` list, preserving any `cache_control` markers the caller
set.

See `create_deep_agent`'s `system_prompt` parameter or
[Prompt assembly](https://docs.langchain.com/oss/deepagents/customization#prompt-assembly)
for the full assembly order.
"""


def _build_default_model() -> ChatAnthropic:
    """Construct the default model without emitting a deprecation warning.

    Internal helper used by `create_deep_agent` so the parameter-level
    `model=None` warning isn't paired with a separate function-level warning
    from `get_default_model`. Direct user calls go through `get_default_model`,
    which keeps its decorator and warns once per process.
    """
    return ChatAnthropic(model_name="claude-sonnet-4-6")


@deprecated(
    since="0.5.3",
    removal="1.0.0",
    message=(
        "Relying on the default model is deprecated and will be removed in "
        "deepagents==1.0.0 alongside support for `model=None` in "
        "`create_deep_agent`. Construct your model explicitly "
        "(e.g., `ChatAnthropic(model_name=...)`). See "
        "https://docs.langchain.com/oss/python/deepagents/models"
    ),
    package="deepagents",
)
def get_default_model() -> ChatAnthropic:
    """Get the default model for Deep Agents.

    !!! deprecated

        Deprecated since `0.5.3`; will be removed in `deepagents==1.0.0`.
        Construct your model explicitly (e.g.,
        `ChatAnthropic(model_name="claude-sonnet-4-6")`).

    Used as a fallback when `model=None` is passed to `create_deep_agent`.

    Requires `ANTHROPIC_API_KEY` to be set in the environment.

    Returns:
        `ChatAnthropic` instance configured with `claude-sonnet-4-6`.
    """
    return _build_default_model()


def _create_bedrock_prompt_caching_middleware() -> AgentMiddleware[Any, Any, Any] | None:
    """Create Bedrock prompt caching middleware when `langchain-aws` is installed."""
    module_name = "langchain_aws.middleware.prompt_caching"
    try:
        module = import_module(module_name)
    except ImportError as exc:
        if exc.name not in {"langchain_aws", "langchain_aws.middleware", module_name}:
            raise
        logger.debug("Bedrock prompt caching middleware is unavailable.", exc_info=exc)
        return None
    middleware_cls = module.BedrockPromptCachingMiddleware
    return cast("AgentMiddleware[Any, Any, Any]", middleware_cls(unsupported_model_behavior="ignore"))


def _append_prompt_caching_middleware(middleware: list[AgentMiddleware[Any, Any, Any]]) -> None:
    """Append provider-specific prompt caching middleware."""
    middleware.append(AnthropicPromptCachingMiddleware(unsupported_model_behavior="ignore"))
    bedrock_middleware = _create_bedrock_prompt_caching_middleware()
    if bedrock_middleware is not None:
        middleware.append(bedrock_middleware)


def _merge_fs_interrupt_on(
    fs_interrupt_on: dict[str, InterruptOnConfig],
    user_interrupt_on: dict[str, bool | InterruptOnConfig] | None,
) -> dict[str, bool | InterruptOnConfig] | None:
    """Combine filesystem-permission configs with user-defined interrupts.

    User-defined `interrupt_on` entries take precedence over generated
    filesystem-permission entries with the same tool name. Returns `None` when
    there are no interrupts to configure, allowing `HumanInTheLoopMiddleware` to
    be omitted.
    """
    if not fs_interrupt_on and not user_interrupt_on:
        return None
    merged: dict[str, bool | InterruptOnConfig] = {**fs_interrupt_on}
    if user_interrupt_on:
        merged.update(user_interrupt_on)
    return merged


_REQUIRED_MIDDLEWARE: tuple[tuple[type[AgentMiddleware[Any, Any, Any]], tuple[str, ...]], ...] = (
    (FilesystemMiddleware, ()),
    (SubAgentMiddleware, ()),
)
"""Scaffolding middleware that core deep agent features depend on.

Each entry pairs a class with any extra string aliases its `.name` may take
beyond `__name__`. Removing any of these silently breaks core features:
`FilesystemMiddleware` backs every built-in file tool and now also enforces
`permissions` rules (a security guarantee), while `SubAgentMiddleware` backs
the `task` tool handler.

Tracked here so `HarnessProfile.excluded_middleware` cannot strip them:
`_apply_excluded_middleware` raises `ValueError` rather than proceeding with
a silently degraded agent.
"""

_REQUIRED_MIDDLEWARE_CLASSES: frozenset[type[AgentMiddleware[Any, Any, Any]]] = frozenset(cls for cls, _ in _REQUIRED_MIDDLEWARE)
"""Set of all class types that cannot be excluded from the middleware stack.

Derived from `_REQUIRED_MIDDLEWARE` and used for quick membership testing.
"""

_REQUIRED_MIDDLEWARE_NAMES: frozenset[str] = frozenset(name for cls, aliases in _REQUIRED_MIDDLEWARE for name in (cls.__name__, *aliases))
"""Set of all `.name` values that cannot be excluded from the middleware stack.

Derived from `_REQUIRED_MIDDLEWARE` and used for quick membership testing.
"""


def create_deep_agent(  # noqa: C901, PLR0912, PLR0915  # Complex graph assembly logic with many conditional branches
    model: str | BaseChatModel | None = None,
    tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
    *,
    system_prompt: str | SystemMessage | None = None,
    middleware: Sequence[AgentMiddleware] = (),
    subagents: Sequence[SubAgent | CompiledSubAgent | AsyncSubAgent] | None = None,
    skills: list[str] | None = None,
    memory: list[str] | None = None,
    permissions: list[FilesystemPermission] | None = None,
    backend: BackendProtocol | BackendFactory | None = None,
    interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
    response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None,
    state_schema: type[DeepAgentState] | None = None,
    context_schema: type[ContextT] | None = None,
    checkpointer: Checkpointer | None = None,
    store: BaseStore | None = None,
    debug: bool = False,
    name: str | None = None,
    cache: BaseCache | None = None,
) -> CompiledStateGraph[AgentState[ResponseT], ContextT, InputAgentState, OutputAgentState[ResponseT]]:  # ty: ignore[invalid-type-arguments]  # ty can't verify generic TypedDicts satisfy StateLike bound
    r"""Create a deep agent.

    By default, this agent has access to the following tools:

    - `write_todos`: manage a todo list
    - `ls`, `read_file`, `write_file`, `edit_file`, `glob`, `grep`: file operations
    - `execute`: run shell commands
    - `task`: call subagents

    The `execute` tool allows running shell commands if the backend implements
    [`SandboxBackendProtocol`][deepagents.backends.protocol.SandboxBackendProtocol].
    For non-sandbox backends, the `execute` tool will return an error message.

    Args:
        model: The model to use.

            !!! deprecated

                Specify a model explicitly.

                Passing `model=None` (relying on the default
                `claude-sonnet-4-6`) is deprecated since `0.5.3` and will
                be removed in `deepagents==1.0.0`. The parameter type
                will change from `BaseChatModel | str | None` to
                `BaseChatModel | str`. See
                [Models](https://docs.langchain.com/oss/python/deepagents/models).

            Accepts a `provider:model` string (e.g., `openai:gpt-5.5`); see
            [`init_chat_model`][langchain.chat_models.init_chat_model(model_provider)]
            for supported values. You can also pass a pre-initialized
            [`BaseChatModel`][langchain.chat_models.BaseChatModel] instance directly.

            !!! note "OpenAI Models and Data Retention"

                If an `openai:` model is used, the agent will use the OpenAI
                Responses API by default. To use OpenAI chat completions
                instead, initialize the model with
                `init_chat_model("openai:...", use_responses_api=False)` and
                pass the initialized model instance here.

                To disable data retention with the Responses API, use
                `init_chat_model("openai:...", use_responses_api=True, store=False, include=["reasoning.encrypted_content"])`
                and pass the initialized model instance here.
        tools: Additional tools the agent should have access to.

            These are merged with the built-in tool suite listed above
            (`write_todos`, filesystem tools, `execute`, and `task`).

            Passing tools here is additive — it never removes a built-in.
            To drop a built-in tool, register a
            [`HarnessProfile`][deepagents.HarnessProfile] with
            `excluded_tools`.
        system_prompt: Custom system instructions placed at the front of
            the system prompt sent to the model.

            Whatever you pass here always sits before the SDK's default
            deep-agent prompt and any model-tuning suffix from a
            registered `HarnessProfile`. With `system_prompt=None`, the
            SDK default is used on its own (plus the profile suffix
            when one applies). Sections are joined by a blank line.

            Passing a `SystemMessage` instead of a string preserves any
            `cache_control` markers on the message's content blocks —
            useful for placing explicit Anthropic prompt-cache
            breakpoints. The same ordering applies (caller's blocks
            first, SDK content appended as an additional text block).

            See [Prompt assembly](https://docs.langchain.com/oss/deepagents/customization#prompt-assembly)
            for the full case-by-case breakdown.
        middleware: Additional middleware to apply after the base stack
            but before the tail middleware. The full ordering is:

            Base stack:

            - [`TodoListMiddleware`][langchain.agents.middleware.TodoListMiddleware]
            - [`SkillsMiddleware`][deepagents.middleware.skills.SkillsMiddleware] (if `skills` is provided)
            - [`FilesystemMiddleware`][deepagents.middleware.filesystem.FilesystemMiddleware]
            - [`SubAgentMiddleware`][deepagents.middleware.subagents.SubAgentMiddleware]
                (if any inline subagents — declarative
                [`SubAgent`][deepagents.middleware.subagents.SubAgent] or
                [`CompiledSubAgent`][deepagents.middleware.subagents.CompiledSubAgent]
                — are available)
            - [`SummarizationMiddleware`][langchain.agents.middleware.SummarizationMiddleware]
            - [`PatchToolCallsMiddleware`][deepagents.middleware.patch_tool_calls.PatchToolCallsMiddleware]
            - [`AsyncSubAgentMiddleware`][deepagents.middleware.async_subagents.AsyncSubAgentMiddleware] (if async `subagents` are provided)

            *User middleware is inserted here.*

            Tail stack:

            - Harness profile `extra_middleware` (if any)
            - `_ToolExclusionMiddleware` (if profile has `excluded_tools`)
            - [`AnthropicPromptCachingMiddleware`][langchain_anthropic.middleware.AnthropicPromptCachingMiddleware] (unconditional; no-ops for
                non-Anthropic models)
            - [`BedrockPromptCachingMiddleware`](https://reference.langchain.com/python/langchain-aws/middleware/prompt_caching/BedrockPromptCachingMiddleware)
                when `langchain-aws` is installed (no-ops for non-Bedrock models)
            - [`MemoryMiddleware`][deepagents.middleware.memory.MemoryMiddleware] (if `memory` is provided)
            - [`HumanInTheLoopMiddleware`][langchain.agents.middleware.HumanInTheLoopMiddleware] (if `interrupt_on` is provided)

            After assembly, any entries in the profile's
            `excluded_middleware` are filtered from the final stack. Class
            entries match exact type; string entries match
            `AgentMiddleware.name` exactly (e.g. `"SummarizationMiddleware"`
            drops the summarization middleware via its public alias).
            Entries that match nothing in the assembled stack raise
            `ValueError`, as does excluding any class in the harness's
            protected scaffolding set (e.g.,
            [`FilesystemMiddleware`][deepagents.middleware.filesystem.FilesystemMiddleware]
            or [`SubAgentMiddleware`][deepagents.middleware.subagents.SubAgentMiddleware]).

            To run without the `task` tool, set
            `general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False)`
            on the active harness profile and pass no synchronous
            subagents via `subagents=`. Async subagents are unaffected.
        subagents: Subagent specs available to the main agent.

            This collection supports three forms:

            - [`SubAgent`][deepagents.middleware.subagents.SubAgent]: A declarative synchronous subagent spec.
            - [`CompiledSubAgent`][deepagents.middleware.subagents.CompiledSubAgent]: A pre-compiled runnable subagent.
            - [`AsyncSubAgent`][deepagents.middleware.async_subagents.AsyncSubAgent]: A remote/background subagent spec.

            `SubAgent` entries are invoked through the `task` tool. They should
            provide `name`, `description`, and `system_prompt`, and may also
            override `tools`, `model`, `middleware`, `interrupt_on`, `skills`,
            `permissions`, and `response_format`. See `interrupt_on` below for
            inheritance and override behavior.

            `CompiledSubAgent` entries are also exposed through the `task` tool,
            but provide a pre-built `runnable` instead of a declarative prompt
            and tool configuration.

            `AsyncSubAgent` entries are identified by their async-subagent
            fields (`graph_id`, and optionally `url`/`headers`) and are routed
            into `AsyncSubAgentMiddleware` instead of `SubAgentMiddleware`.
            They should provide `name`, `description`, and `graph_id`, and may
            optionally include `url` and `headers`. These subagents run as
            background tasks and expose the async subagent tools for launching,
            checking, updating, cancelling, and listing tasks.

            If no subagent named `general-purpose` is provided, a default
            general-purpose synchronous subagent is added automatically unless
            the active harness profile disables it. With no synchronous
            subagents in play — none passed and the default disabled via
            `general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False)`
            — the `task` tool is not exposed. Async subagents are independent.

        skills: List of skill source paths (e.g., `["/skills/user/", "/skills/project/"]`).

            Paths must be specified using POSIX conventions (forward slashes)
            and are relative to the backend's root. When using
            `StateBackend` (default), provide skill files via
            `invoke(files={...})`. With `FilesystemBackend`, skills are loaded
            from disk relative to the backend's `root_dir`. Later sources
            override earlier ones for skills with the same name (last one wins).
        memory: List of memory file paths (`AGENTS.md` files) to load
            (e.g., `["/memory/AGENTS.md"]`).

            Display names are automatically derived from paths.

            Memory is loaded at agent startup and added into the system prompt.
        permissions: List of `FilesystemPermission` rules for the main agent
            and its subagents.

            Rules are evaluated in declaration order; the first match wins.
            If no rule matches, the call is allowed.

            Each rule's `mode` can be:

            - `"allow"` (default): the call proceeds.
            - `"deny"`: the tool returns a permission-denied error.
            - `"interrupt"`: the call pauses for human approval via
                `HumanInTheLoopMiddleware`. A `HumanInTheLoopMiddleware` is
                auto-installed when any interrupt-mode rule is present, and the
                generated `interrupt_on` entries are merged with the
                `interrupt_on` argument below (user-supplied entries win per
                tool name). Requires a `langchain` version that supports the
                `when` predicate on `InterruptOnConfig`.

            Subagents inherit these rules unless they specify their own
            `permissions` field, which replaces the parent's rules entirely.

            `FilesystemMiddleware` applies these permissions at the tool
            level for its built-in filesystem tools, not at the backend
            level. Direct backend usage does not currently incorporate
            `permissions`.
        backend: Optional backend for file storage and execution.

            Pass a `Backend` instance (e.g. `StateBackend()`).

            For execution support, use a backend that
            implements [`SandboxBackendProtocol`][deepagents.backends.protocol.SandboxBackendProtocol].
        interrupt_on: Mapping of tool names to interrupt configs.

            Pass to pause agent execution at specified tool calls for human
            approval or modification.

            This config always applies to the main agent.

            For subagents:
            - Declarative `SubAgent` specs inherit the top-level `interrupt_on`
                config by default.
            - If a declarative `SubAgent` provides its own `interrupt_on`, that
                subagent-specific config overrides the inherited
                top-level config.
            - `CompiledSubAgent` runnables do not inherit top-level
                `interrupt_on`; configure human-in-the-loop behavior inside the
                compiled runnable itself.
            - Remote `AsyncSubAgent` specs do not inherit top-level
                `interrupt_on`; configure any approval behavior on the remote
                subagent itself.

            For example, `interrupt_on={"edit_file": True}` pauses before
            every edit.
        response_format: A structured output response format to use for the agent.
        state_schema: Custom state schema for the agent graph. Must be a
            `TypedDict` subclass of
            [`DeepAgentState`][deepagents.graph.DeepAgentState] so the
            built-in `DeltaChannel` reducer on `messages` is preserved.

            Generally, prefer defining state extensions with middleware so
            the extra fields stay scoped to the hooks and tools that use
            them.

            When provided, this schema is used as the base graph schema and
            is merged with state schemas contributed by middleware. It is
            also forwarded when compiling declarative
            [`SubAgent`][deepagents.middleware.subagents.SubAgent] specs for
            the `task` tool, so subagents see the same custom fields as the
            parent.

            [`CompiledSubAgent`][deepagents.middleware.subagents.CompiledSubAgent]
            runnables do not inherit this schema because they are already
            compiled — compile those runnables with a compatible state
            schema if they need access to the same custom state fields.
            Remote
            [`AsyncSubAgent`][deepagents.middleware.async_subagents.AsyncSubAgent]
            specs likewise use the schema configured on the remote graph.

            ```python
            from deepagents.graph import DeepAgentState


            class MyState(DeepAgentState):
                page_url: str
                file_urls: list[str]


            agent = create_deep_agent(model=..., state_schema=MyState)
            ```
        context_schema: Schema class that defines immutable run-scoped context.

            Passed through to [`create_agent`][langchain.agents.create_agent].
        checkpointer: Optional `Checkpointer` for persisting agent state
            between runs.

            Passed through to [`create_agent`][langchain.agents.create_agent].
        store: Optional store for persistent storage (required if backend
            uses `StoreBackend`).

            Passed through to [`create_agent`][langchain.agents.create_agent].
        debug: Whether to enable debug mode.

            Passed through to [`create_agent`][langchain.agents.create_agent].
        name: The name of the agent.

            Passed through to [`create_agent`][langchain.agents.create_agent].
        cache: The cache to use for the agent.

            Passed through to [`create_agent`][langchain.agents.create_agent].

    Returns:
        A configured deep agent.

    Raises:
        ImportError: If a required provider package is missing or below the
            minimum supported version (e.g., `langchain-openrouter`).
        ValueError: If the active `HarnessProfile.excluded_middleware`
            references a class in the harness's protected scaffolding set
            (e.g.,
            [`FilesystemMiddleware`][deepagents.middleware.filesystem.FilesystemMiddleware]
            or
            [`SubAgentMiddleware`][deepagents.middleware.subagents.SubAgentMiddleware]),
            uses a private (underscore-prefixed) name, collides with multiple
            distinct middleware classes, or matches no entry in the assembled
            stack.
    """
    # `DeepAgentState` is a `TypedDict`; TypedDicts disallow `issubclass`, so the
    # subclass constraint on `state_schema` is enforced by typing alone and not
    # validated at runtime.

    _model_spec: str | None = model if isinstance(model, str) else None

    if model is None:
        warn_deprecated(
            since="0.5.3",
            removal="1.0.0",
            message=(
                "Passing `model=None` to `create_deep_agent` is deprecated "
                "and will be removed in deepagents==1.0.0. The `model` "
                "parameter type will change from `BaseChatModel | str | None` "
                "to `BaseChatModel | str`. Specify a model explicitly "
                "(e.g., `ChatAnthropic(model_name=...)`). See "
                "https://docs.langchain.com/oss/python/deepagents/models"
            ),
            package="deepagents",
        )
        # Use the un-decorated builder so we don't burn the dedupe flag on
        # `get_default_model` — direct user callers still see one warning.
        model = _build_default_model()
    else:
        model = resolve_model(model)
    _profile = _harness_profile_for_model(model, _model_spec)
    # Validate profile-level invariants (required scaffolding, private names)
    _validate_excluded_middleware_config(
        _profile,
        required_classes=_REQUIRED_MIDDLEWARE_CLASSES,
        required_names=_REQUIRED_MIDDLEWARE_NAMES,
    )
    # Accumulate which entries matched across the main agent + general-purpose
    # subagent stacks (both use `_profile`). A profile-level entry only has to
    # match somewhere, not in every stack, so coverage is verified once after
    # all filters have run.
    _main_matched_classes: set[type[AgentMiddleware[Any, Any, Any]]] = set()
    _main_matched_names: set[str] = set()

    # Copy of `tools` with any harness-specific description rewrites.
    # (Tool exclusion is handled by _ToolExclusionMiddleware which filters
    # all tools (user-supplied and middleware-injected) in one place.)
    _tools = _apply_tool_description_overrides(
        tools,
        _profile.tool_description_overrides,
    )

    backend = backend if backend is not None else StateBackend()

    # Process caller-supplied subagents first so the decision of whether to
    # auto-add the default general-purpose subagent can factor in an explicit
    # override, and so its middleware stack (including any factory-based
    # `extra_middleware`) isn't built and then discarded.
    inline_subagents: list[SubAgent | CompiledSubAgent] = []
    async_subagents: list[AsyncSubAgent] = []
    for spec in subagents or []:
        if "graph_id" in spec:
            # Then spec is an AsyncSubAgent
            async_subagents.append(cast("AsyncSubAgent", spec))
            continue
        if "runnable" in spec:
            # CompiledSubAgent - use as-is
            inline_subagents.append(spec)
        else:
            # SubAgent - fill in defaults and prepend base middleware
            raw_subagent_model = spec.get("model", model)
            subagent_model = resolve_model(raw_subagent_model)

            _subagent_spec = raw_subagent_model if isinstance(raw_subagent_model, str) else None
            _subagent_profile = _harness_profile_for_model(subagent_model, _subagent_spec)

            # Resolve permissions: subagent's own rules take priority, else inherit parent's
            subagent_permissions = spec.get("permissions", permissions)

            # Build middleware: base stack + skills (if specified) + user's middleware
            subagent_middleware: list[AgentMiddleware[Any, Any, Any]] = [
                TodoListMiddleware(),
                FilesystemMiddleware(
                    backend=backend,
                    custom_tool_descriptions=_subagent_profile.tool_description_overrides,
                    _permissions=subagent_permissions,
                ),
                create_summarization_middleware(subagent_model, backend),
                PatchToolCallsMiddleware(),
            ]
            subagent_skills = spec.get("skills")
            if subagent_skills:
                subagent_middleware.append(SkillsMiddleware(backend=backend, sources=subagent_skills))
            subagent_middleware.extend(spec.get("middleware", []))

            # Harness-profile middleware for this subagent's model
            subagent_middleware.extend(_subagent_profile.materialize_extra_middleware())
            if _subagent_profile.excluded_tools:
                subagent_middleware.append(_ToolExclusionMiddleware(excluded=_subagent_profile.excluded_tools))

            _append_prompt_caching_middleware(subagent_middleware)

            _subagent_matched_classes: set[type[AgentMiddleware[Any, Any, Any]]] = set()
            _subagent_matched_names: set[str] = set()
            _validate_excluded_middleware_config(
                _subagent_profile,
                required_classes=_REQUIRED_MIDDLEWARE_CLASSES,
                required_names=_REQUIRED_MIDDLEWARE_NAMES,
            )
            subagent_middleware = _apply_excluded_middleware(
                subagent_middleware,
                _subagent_profile,
                matched_classes=_subagent_matched_classes,
                matched_names=_subagent_matched_names,
            )
            _verify_excluded_middleware_coverage(
                _subagent_profile,
                _subagent_matched_classes,
                _subagent_matched_names,
                required_classes=_REQUIRED_MIDDLEWARE_CLASSES,
                required_names=_REQUIRED_MIDDLEWARE_NAMES,
            )

            subagent_interrupt_on = spec.get("interrupt_on", interrupt_on)
            subagent_interrupt_on = _merge_fs_interrupt_on(
                _build_interrupt_on_from_permissions(subagent_permissions or []),
                subagent_interrupt_on,
            )

            # Inherit parent tools unless the subagent declares its own.
            # Descriptions are rewritten; exclusion is handled by middleware.
            raw_subagent_tools = spec.get("tools") if "tools" in spec else tools
            subagent_tools = _apply_tool_description_overrides(
                raw_subagent_tools,
                _subagent_profile.tool_description_overrides,
            )

            processed_spec: SubAgent = {
                **spec,
                "model": subagent_model,
                "tools": subagent_tools or [],
                "middleware": subagent_middleware,
            }
            processed_spec["system_prompt"] = _apply_profile_prompt(_subagent_profile, spec["system_prompt"])
            if subagent_interrupt_on is not None:
                processed_spec["interrupt_on"] = subagent_interrupt_on
            inline_subagents.append(processed_spec)

    # Auto-add the default general-purpose subagent unless the harness profile
    # disables it or the caller already supplied their own — an explicit spec
    # is how callers override the default. Skipping in those cases also avoids
    # invoking factory-based `extra_middleware` whose output would be thrown
    # away.
    gp_profile = _profile.general_purpose_subagent or GeneralPurposeSubagentProfile()
    if gp_profile.enabled is not False and not any(spec["name"] == GENERAL_PURPOSE_SUBAGENT["name"] for spec in inline_subagents):
        gp_middleware: list[AgentMiddleware[Any, Any, Any]] = [
            TodoListMiddleware(),
            FilesystemMiddleware(
                backend=backend,
                custom_tool_descriptions=_profile.tool_description_overrides,
                _permissions=permissions,
            ),
            create_summarization_middleware(model, backend),
            PatchToolCallsMiddleware(),
        ]
        if skills is not None:
            gp_middleware.append(SkillsMiddleware(backend=backend, sources=skills))

        # Add harness-profile middleware, if any
        gp_middleware.extend(_profile.materialize_extra_middleware())

        # Strip excluded tools after all tool-injecting middleware has run
        if _profile.excluded_tools:
            gp_middleware.append(_ToolExclusionMiddleware(excluded=_profile.excluded_tools))
        _append_prompt_caching_middleware(gp_middleware)

        gp_middleware = _apply_excluded_middleware(
            gp_middleware,
            _profile,
            matched_classes=_main_matched_classes,
            matched_names=_main_matched_names,
        )

        general_purpose_spec: SubAgent = {
            **GENERAL_PURPOSE_SUBAGENT,
            "model": model,
            "tools": _tools or [],
            "middleware": gp_middleware,
        }
        if gp_profile.description is not None:
            general_purpose_spec["description"] = gp_profile.description
        if gp_profile.system_prompt is not None:
            # GP-specific override beats `profile.base_system_prompt`; only the
            # profile suffix layers on top.
            gp_prompt = gp_profile.system_prompt
            if _profile.system_prompt_suffix is not None:
                gp_prompt = gp_prompt + "\n\n" + _profile.system_prompt_suffix
            general_purpose_spec["system_prompt"] = gp_prompt
        else:
            general_purpose_spec["system_prompt"] = _apply_profile_prompt(_profile, GENERAL_PURPOSE_SUBAGENT["system_prompt"])
        gp_interrupt_on = _merge_fs_interrupt_on(
            _build_interrupt_on_from_permissions(permissions or []),
            interrupt_on,
        )
        if gp_interrupt_on is not None:
            general_purpose_spec["interrupt_on"] = gp_interrupt_on

        inline_subagents.insert(0, general_purpose_spec)

    # Build main agent middleware stack
    deepagent_middleware: list[AgentMiddleware[Any, Any, Any]] = [
        TodoListMiddleware(),
    ]
    if skills is not None:
        deepagent_middleware.append(SkillsMiddleware(backend=backend, sources=skills))
    deepagent_middleware.append(
        FilesystemMiddleware(
            backend=backend,
            custom_tool_descriptions=_profile.tool_description_overrides,
            _permissions=permissions,
        )
    )
    sub_agent_middleware: SubAgentMiddleware | None = None
    if inline_subagents:
        sub_agent_middleware = SubAgentMiddleware(
            backend=backend,
            subagents=inline_subagents,
            # Overrides the task tool description. Value should include
            # {available_agents} — a format placeholder replaced with the
            # subagent name/description list. Without it the model can't
            # see which subagents exist. None (default) uses the built-in
            # template. Stale keys silently no-op if the tool is renamed.
            task_description=_profile.tool_description_overrides.get("task"),
            state_schema=state_schema,
        )
        deepagent_middleware.append(sub_agent_middleware)
    deepagent_middleware.extend(
        [
            create_summarization_middleware(model, backend),
            PatchToolCallsMiddleware(),
        ]
    )

    if async_subagents:
        # Async here means that we run these subagents in a non-blocking manner.
        # Currently this supports agents deployed via LangSmith deployments.
        deepagent_middleware.append(AsyncSubAgentMiddleware(async_subagents=async_subagents))

    if middleware:
        deepagent_middleware.extend(middleware)
    # Harness-profile middleware goes between user middleware and memory so
    # that memory updates (which change the system prompt) don't invalidate the
    # Anthropic prompt cache prefix.
    deepagent_middleware.extend(_profile.materialize_extra_middleware())
    if _profile.excluded_tools:
        deepagent_middleware.append(_ToolExclusionMiddleware(excluded=_profile.excluded_tools))
    _append_prompt_caching_middleware(deepagent_middleware)
    if memory is not None:
        # MemoryMiddleware applies the cache_control breakpoint only when the
        # request model is Anthropic, making it safe to enable unconditionally.
        deepagent_middleware.append(
            MemoryMiddleware(
                backend=backend,
                sources=memory,
                add_cache_control=True,
            )
        )
    main_interrupt_on = _merge_fs_interrupt_on(
        _build_interrupt_on_from_permissions(permissions or []),
        interrupt_on,
    )
    if main_interrupt_on is not None:
        deepagent_middleware.append(HumanInTheLoopMiddleware(interrupt_on=main_interrupt_on))
    deepagent_middleware = _apply_excluded_middleware(
        deepagent_middleware,
        _profile,
        matched_classes=_main_matched_classes,
        matched_names=_main_matched_names,
    )
    private_state_keys = private_state_field_names(*(mw.state_schema for mw in deepagent_middleware if getattr(mw, "state_schema", None) is not None))
    if sub_agent_middleware is not None:
        sub_agent_middleware.private_state_keys = private_state_keys
    # Verify every main-profile exclusion matched at least one middleware in
    # either the main agent stack or the GP subagent stack. An entry that
    # matched nothing across both is almost certainly a typo or a stale
    # profile.
    _verify_excluded_middleware_coverage(
        _profile,
        _main_matched_classes,
        _main_matched_names,
        required_classes=_REQUIRED_MIDDLEWARE_CLASSES,
        required_names=_REQUIRED_MIDDLEWARE_NAMES,
    )

    base_prompt = _apply_profile_prompt(_profile, BASE_AGENT_PROMPT)
    if system_prompt is None:
        final_system_prompt: str | SystemMessage = base_prompt
    elif isinstance(system_prompt, SystemMessage):
        final_system_prompt = SystemMessage(content_blocks=[*system_prompt.content_blocks, {"type": "text", "text": f"\n\n{base_prompt}"}])
    else:
        final_system_prompt = system_prompt + "\n\n" + base_prompt

    return create_agent(
        model,
        system_prompt=final_system_prompt,
        tools=_tools,
        middleware=deepagent_middleware,
        response_format=response_format,
        context_schema=context_schema,
        checkpointer=checkpointer,
        store=store,
        debug=debug,
        name=name,
        cache=cache,
        state_schema=state_schema if state_schema is not None else DeepAgentState,
    ).with_config(
        {
            "recursion_limit": 9_999,
            "metadata": {
                "ls_integration": "deepagents",
                "lc_versions": {"deepagents": __version__},
                "lc_agent_name": name,
            },
        }
    )
