"""Agent management and creation."""

from __future__ import annotations

import functools
import logging
import os
import re
import shutil
import tempfile
import tomllib
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast

from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, LocalShellBackend
from deepagents.backends.filesystem import FilesystemBackend
from deepagents.middleware import MemoryMiddleware, SkillsMiddleware

# Backwards-compat flag: SDKs before 0.5.4 accept only `list[str]` for
# `SkillsMiddleware.sources`; newer SDKs expose the `SkillSource` alias
# that permits `(path, label)` tuples. The `skills` module is already
# loaded by the `SkillsMiddleware` import above, so the extra lookup
# here adds no startup cost.
try:
    from deepagents.middleware.skills import SkillSource as _SkillSource  # noqa: F401
except ImportError:
    _SUPPORTS_SKILL_SOURCE_TUPLES = False
else:
    _SUPPORTS_SKILL_SOURCE_TUPLES = True

if TYPE_CHECKING:
    from collections.abc import Awaitable, Callable, Sequence

    from deepagents.backends.sandbox import SandboxBackendProtocol
    from deepagents.middleware.async_subagents import AsyncSubAgent
    from deepagents.middleware.subagents import CompiledSubAgent, SubAgent
    from langchain.agents.middleware import InterruptOnConfig
    from langchain.agents.middleware.types import AgentState
    from langchain.messages import ToolCall
    from langchain.tools import BaseTool
    from langchain_core.language_models import BaseChatModel
    from langchain_core.messages import ToolMessage
    from langgraph.checkpoint.base import BaseCheckpointSaver
    from langgraph.prebuilt.tool_node import ToolCallRequest
    from langgraph.pregel import Pregel
    from langgraph.runtime import Runtime
    from langgraph.types import Command

    from deepagents_code.mcp_tools import MCPServerInfo
    from deepagents_code.output import OutputFormat

from langchain.agents.middleware.types import AgentMiddleware

from deepagents_code import theme
from deepagents_code._cli_context import CLIContextSchema
from deepagents_code._constants import DEFAULT_AGENT_NAME
from deepagents_code.config import (
    _INHERITED_PYTHONPATH_ENV,
    _ShellAllowAll,
    config,
    console,
    get_default_coding_instructions,
    get_glyphs,
    get_langsmith_project_name,
    restore_user_tracing_env,
    settings,
)
from deepagents_code.configurable_model import ConfigurableModelMiddleware
from deepagents_code.integrations.sandbox_factory import get_default_working_dir
from deepagents_code.local_context import (
    LocalContextMiddleware,
    _AsyncExecutableBackend,
    _ExecutableBackend,
)
from deepagents_code.project_utils import ProjectContext, get_server_project_context
from deepagents_code.subagents import list_subagents
from deepagents_code.unicode_security import (
    check_url_safety,
    detect_dangerous_unicode,
    format_warning_detail,
    render_with_unicode_markers,
    strip_dangerous_unicode,
    summarize_issues,
)

logger = logging.getLogger(__name__)

REQUIRE_COMPACT_TOOL_APPROVAL: bool = True
"""When `True`, `compact_conversation` requires HITL approval like other gated tools."""


def _sanitize_agent_message_name(agent_name: str) -> str:
    """Return a provider-safe message name for a user-facing agent name.

    Args:
        agent_name: Display/storage name for the selected agent.

    Returns:
        Name containing only alphanumerics, underscores, and hyphens.
    """
    sanitized = re.sub(r"[^a-zA-Z0-9_-]+", "_", agent_name).strip("_")
    return sanitized or DEFAULT_AGENT_NAME


class ShellAllowListMiddleware(AgentMiddleware):
    """Validate shell commands against an allow-list without HITL interrupts.

    When the agent invokes the `execute` shell tool, this middleware checks
    the command against the configured allow-list **before execution**.
    Rejected commands are returned as error `ToolMessage` objects — the
    graph never pauses, so LangSmith traces stay as a single continuous
    run.

    Use this middleware in non-interactive mode to avoid the
    interrupt/resume cycle that fragments traces.
    """

    def __init__(self, allow_list: list[str]) -> None:
        """Initialize with the shell allow-list to validate commands against.

        Args:
            allow_list: Allowed command names (e.g. `["ls", "cat", "grep"]`).
                Must be a non-empty restrictive list — not `SHELL_ALLOW_ALL`.

        Raises:
            ValueError: If `allow_list` is empty.
            TypeError: If `allow_list` is the `SHELL_ALLOW_ALL` sentinel.
        """
        from deepagents_code.config import SHELL_ALLOW_ALL

        super().__init__()
        if not allow_list:
            msg = "allow_list must not be empty; disable shell access instead"
            raise ValueError(msg)
        if isinstance(allow_list, type(SHELL_ALLOW_ALL)):
            msg = (
                "SHELL_ALLOW_ALL should not be used with "
                "ShellAllowListMiddleware; use auto_approve=True instead"
            )
            raise TypeError(msg)
        self._allow_list = list(allow_list)

    def _validate_tool_call(self, request: ToolCallRequest) -> ToolMessage | None:
        """Return an error tool message when a shell command is not allowed.

        Args:
            request: The tool call request being processed.

        Returns:
            An error `ToolMessage` when the shell command should be rejected,
            otherwise `None`.
        """
        from langchain_core.messages import ToolMessage as LCToolMessage

        from deepagents_code.config import is_shell_command_allowed

        if request.tool_call["name"] != "execute":
            return None

        args = request.tool_call.get("args") or {}
        command = args.get("command", "")
        if is_shell_command_allowed(command, self._allow_list):
            logger.debug("Shell command allowed: %r", command)
            return None

        logger.warning("Shell command rejected by allow-list: %r", command)
        allowed_str = ", ".join(self._allow_list)
        return LCToolMessage(
            content=(
                f"Shell command rejected: `{command}` is not in the allow-list. "
                f"Allowed commands: {allowed_str}. "
                f"Please use an allowed command or try another approach."
            ),
            name="execute",
            tool_call_id=request.tool_call["id"],
            status="error",
        )

    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
    ) -> ToolMessage | Command[Any]:
        """Reject disallowed shell commands; pass everything else through.

        Args:
            request: The tool call request being processed.
            handler: The next handler in the middleware chain.

        Returns:
            The tool execution result, or an error `ToolMessage` for rejected
            shell commands.
        """
        if (rejection := self._validate_tool_call(request)) is not None:
            return rejection
        return handler(request)

    async def awrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
    ) -> ToolMessage | Command[Any]:
        """Reject disallowed shell commands; pass everything else through.

        Args:
            request: The tool call request being processed.
            handler: The next handler in the middleware chain.

        Returns:
            The tool execution result, or an error `ToolMessage` for rejected
            shell commands.
        """
        if (rejection := self._validate_tool_call(request)) is not None:
            return rejection
        return await handler(request)


_INTERPRETER_WRITE_TOOLS: frozenset[str] = frozenset(
    {"execute", "write_file", "edit_file", "delete"}
)
"""Tools considered write/shell capable for PTC auditing.

When `interpreter_ptc="all"` resolves to this set, an INFO log names every
write tool that was included so the audit trail is searchable. The `"safe"`
preset already excludes them; this is the belt-and-braces check for `"all"`.
"""


def _resolve_ptc_option(
    ptc: str | bool | list[str],
    *,
    tools: Sequence[BaseTool | Callable | dict[str, Any]],
    acknowledge_unsafe: bool,
    auto_approve: bool,
) -> list[str] | None:
    """Resolve the configured PTC allowlist to a concrete list of tool names.

    Names are *not* validated against `tools`. The Deep Agents SDK injects the
    filesystem, `task`, `write_todos`, and `execute` tools via middleware in
    `create_deep_agent` — *after* this point — so they are absent from `tools`
    here, and the SDK exposes no importable list of them. `CodeInterpreterMiddleware`
    matches the resolved names against the live runtime registry and silently
    ignores any that are absent, so resolution passes names through and lets
    runtime decide. (Names that match nothing at runtime are dropped, so a typo
    silently exposes no tool rather than raising.)

    Args:
        ptc: Raw `interpreter_ptc` value from settings or CLI. Accepts
            `False`/`[]`, `"safe"`, `"all"`, or a list of names. A list may
            include `"safe"`, which expands to `INTERPRETER_PTC_SAFE_PRESET`;
            `"all"` is rejected inside a list.
        tools: Tools passed to `create_cli_agent`. Used only to enumerate
            `"all"`, which is therefore limited to these explicitly-passed
            tools (the SDK runtime built-ins cannot be enumerated here).
        acknowledge_unsafe: Mirrors `settings.interpreter_ptc_acknowledge_unsafe`;
            required when `ptc="all"` and `auto_approve` is `False`.
        auto_approve: Whether HITL approval is globally disabled. When `True`,
            `"all"` does not require `acknowledge_unsafe` because every host
            tool already runs without prompting.

    Returns:
        `None` when PTC should be disabled, otherwise a list of tool names
        suitable for `CodeInterpreterMiddleware(ptc=...)`.

    Raises:
        ValueError: For `"all"` inside a list, for `"all"` without
            `acknowledge_unsafe` outside of `auto_approve`, or for an invalid
            `ptc` type or string.
    """
    from langchain.tools import BaseTool as _BaseTool

    if ptc is False or ptc is None or ptc == []:
        return None

    live_names: list[str] = []
    for tool in tools:
        if isinstance(tool, _BaseTool):
            name = tool.name
            if isinstance(name, str):
                live_names.append(name)
        elif isinstance(tool, dict):
            raw_name = cast("dict[str, Any]", tool).get("name")
            if isinstance(raw_name, str):
                live_names.append(raw_name)
        else:
            attr = getattr(tool, "name", None)
            if isinstance(attr, str):
                live_names.append(attr)
    live_set: set[str] = set(live_names)

    if isinstance(ptc, str):
        normalized = ptc.strip().lower()
        if normalized == "safe":
            from deepagents_code.config import INTERPRETER_PTC_SAFE_PRESET

            # Return the preset as-is; the middleware exposes whichever members
            # exist in the live registry at runtime (they are SDK built-ins not
            # present in `tools` here).
            return sorted(INTERPRETER_PTC_SAFE_PRESET)
        if normalized == "all":
            if not auto_approve and not acknowledge_unsafe:
                msg = (
                    "interpreter_ptc='all' exposes every host tool to PTC "
                    "calls that bypass HITL approval. Set "
                    "interpreter_ptc_acknowledge_unsafe=True (or use "
                    "auto_approve=True) to opt in."
                )
                raise ValueError(msg)
            # `all` can only enumerate the tools passed to `create_cli_agent`;
            # SDK runtime built-ins (filesystem, `task`, …) are injected later
            # and are not enumerable here. Exposing them under `all` needs an
            # "expose everything" sentinel in `CodeInterpreterMiddleware`
            # (tracked in langchain-ai/deepagents#3847).
            included = sorted(live_set)
            write_included = sorted(_INTERPRETER_WRITE_TOOLS & live_set)
            if write_included:
                logger.info(
                    "interpreter_ptc='all' includes write/shell tools: %s",
                    write_included,
                )
            return included
        msg = (
            f"Invalid interpreter_ptc string {ptc!r}; expected 'safe', 'all', "
            "or a list of tool names."
        )
        raise ValueError(msg)

    if isinstance(ptc, list):
        from deepagents_code.config import INTERPRETER_PTC_SAFE_PRESET

        if any(name.strip().lower() == "all" for name in ptc):
            msg = (
                "interpreter_ptc list entries cannot include 'all'; use 'all' "
                "as a standalone value or list explicit tool names (optionally "
                "with the 'safe' preset)."
            )
            raise ValueError(msg)

        resolved: list[str] = []
        seen: set[str] = set()

        def _add(name: str) -> None:
            if name not in seen:
                seen.add(name)
                resolved.append(name)

        for name in ptc:
            if name.strip().lower() == "safe":
                for member in sorted(INTERPRETER_PTC_SAFE_PRESET):
                    _add(member)
                continue
            _add(name)

        # Explicit names are passed through unvalidated: the middleware resolves
        # them against the live runtime registry (which includes the SDK
        # built-ins absent from `tools`) and drops any that match nothing.
        absent = sorted(n for n in resolved if n not in live_set)
        if absent:
            logger.debug(
                "interpreter_ptc names not in the build-time toolset (resolved "
                "at runtime if present): %s",
                absent,
            )
        return resolved

    msg = (
        "interpreter_ptc must be False, 'safe', 'all', or a list of tool names; "
        f"got {type(ptc).__name__}."
    )
    raise ValueError(msg)


def load_async_subagents(config_path: Path | None = None) -> list[AsyncSubAgent]:
    """Load async subagent definitions from `config.toml`.

    Reads the `[async_subagents]` section where each sub-table defines a remote
    LangGraph deployment:

    ```toml
    [async_subagents.researcher]
    description = "Research agent"
    url = "https://my-deployment.langsmith.dev"
    graph_id = "agent"
    ```

    Args:
        config_path: Path to config file.

            Defaults to `~/.deepagents/config.toml`.

    Returns:
        List of `AsyncSubAgent` specs (empty if section is absent or invalid).
    """
    if config_path is None:
        config_path = Path.home() / ".deepagents" / "config.toml"

    if not config_path.exists():
        return []

    try:
        with config_path.open("rb") as f:
            data = tomllib.load(f)
    except (tomllib.TOMLDecodeError, PermissionError, OSError) as e:
        logger.warning("Could not read async subagents from %s: %s", config_path, e)
        console.print(
            f"[bold yellow]Warning:[/bold yellow] Could not read async subagents "
            f"from {config_path}: {e}",
        )
        return []

    section = data.get("async_subagents")
    if not isinstance(section, dict):
        return []

    required = {"description", "graph_id"}
    agents: list[AsyncSubAgent] = []
    for name, spec in section.items():
        if not isinstance(spec, dict):
            logger.warning("Skipping async subagent '%s': expected a table", name)
            continue
        missing = required - spec.keys()
        if missing:
            logger.warning(
                "Skipping async subagent '%s': missing fields %s", name, missing
            )
            continue
        agent: AsyncSubAgent = {
            "name": name,
            "description": spec["description"],
            "graph_id": spec["graph_id"],
        }
        if "url" in spec and isinstance(spec["url"], str):
            agent["url"] = spec["url"]
        if "headers" in spec and isinstance(spec["headers"], dict):
            agent["headers"] = spec["headers"]
        agents.append(agent)

    return agents


@functools.lru_cache(maxsize=1)
def _reserved_agent_dir_names() -> frozenset[str]:
    """Return non-agent directory names reserved by the app under `~/.deepagents/`.

    `bin/` holds the managed `rg` binary (`managed_tools.BIN_DIR`) and must
    never appear in the agent picker. The name is derived from `BIN_DIR` so it
    stays a single source of truth rather than being hardcoded here. The result
    is cached since the reserved set is constant for the process.
    """
    from deepagents_code.managed_tools import BIN_DIR

    return frozenset({BIN_DIR.name})


def _is_agent_dir_entry(entry: Path) -> bool:
    """Return whether a `~/.deepagents/` entry should be listed as an agent.

    Filters out symlinks (so dangling links don't masquerade as agents),
    dot-prefixed names — `.state/` (app internal state) plus any other
    hidden directory the user may have placed there — and reserved names
    the app owns (e.g. `bin/`, the managed-binary install dir).

    `OSError` from `is_dir`/`is_symlink` propagates so callers can log
    with the failing entry's name as context.
    """
    if entry.name.startswith(".") or entry.name in _reserved_agent_dir_names():
        return False
    return entry.is_dir() and not entry.is_symlink()


def get_available_agent_names() -> list[str]:
    """Return a sorted list of available agent names from `~/.deepagents/`.

    Scans the user's `.deepagents` directory and returns each real
    subdirectory found there. Symlinks excluded so a dangling link does not
    masquerade as an agent. Dot-prefixed entries (e.g., `.state/`) and
    reserved app-owned directories (e.g., `bin/`, the managed-binary install
    dir) are skipped so internal state never appears as an agent.

    Filesystem errors (missing parent, permission denied, broken entries) are
    logged and surfaced as an empty list rather than raised — the caller shows
    an empty modal instead of crashing mid-render.

    Returns:
        Sorted list of agent names. Empty when no agents exist yet or the
            directory is unreadable (see log for the underlying cause).
    """
    agents_dir = settings.user_deepagents_dir
    try:
        entries = list(agents_dir.iterdir())
    except FileNotFoundError:
        return []
    except OSError:
        logger.warning("Could not list agents in %s", agents_dir, exc_info=True)
        return []

    names: list[str] = []
    for entry in entries:
        try:
            if _is_agent_dir_entry(entry):
                names.append(entry.name)
        except OSError:
            logger.debug(
                "Skipping unreadable entry in %s: %s",
                agents_dir,
                entry.name,
                exc_info=True,
            )
    return sorted(names)


def list_agents(*, output_format: OutputFormat = "text") -> None:
    """List all available agents.

    Args:
        output_format: Output format — `'text'` (Rich) or `'json'`.
    """
    agents_dir = settings.user_deepagents_dir
    names = get_available_agent_names()

    if not names:
        if output_format == "json":
            from deepagents_code.output import write_json

            write_json("list", [])
            return
        console.print("[yellow]No agents found.[/yellow]")
        console.print(
            "[dim]Agents will be created in ~/.deepagents/ "
            "when you first use them.[/dim]",
            style=theme.MUTED,
        )
        return

    if output_format == "json":
        from deepagents_code.output import write_json

        agents = []
        for name in names:
            agent_path = agents_dir / name
            agents.append(
                {
                    "name": name,
                    "path": str(agent_path),
                    "has_agents_md": (agent_path / "AGENTS.md").exists(),
                    "is_default": name == DEFAULT_AGENT_NAME,
                }
            )
        write_json("list", agents)
        return

    from rich.markup import escape as escape_markup

    console.print("\n[bold]Available Agents:[/bold]\n", style=theme.PRIMARY)

    bullet = get_glyphs().bullet
    for name in names:
        agent_path = agents_dir / name
        agent_name = escape_markup(name)
        is_default = name == DEFAULT_AGENT_NAME
        default_label = " [dim](default)[/dim]" if is_default else ""

        if (agent_path / "AGENTS.md").exists():
            console.print(
                f"  {bullet} [bold]{agent_name}[/bold]{default_label}",
                style=theme.PRIMARY,
            )
        else:
            console.print(
                f"  {bullet} [bold]{agent_name}[/bold]{default_label}"
                " [dim](incomplete)[/dim]",
                style=theme.WARNING,
            )
        console.print(
            f"    {escape_markup(str(agent_path))}",
            style=theme.MUTED,
        )

    console.print()


def reset_agent(
    agent_name: str,
    source_agent: str | None = None,
    *,
    dry_run: bool = False,
    output_format: OutputFormat = "text",
) -> None:
    """Reset an agent to default or copy from another agent.

    Args:
        agent_name: Name of the agent to reset.
        source_agent: Copy AGENTS.md from this agent instead of default.
        dry_run: If `True`, print what would happen without making changes.
        output_format: Output format — `'text'` (Rich) or `'json'`.

    Raises:
        SystemExit: If the source agent is not found.
    """
    agents_dir = settings.user_deepagents_dir
    agent_dir = agents_dir / agent_name

    if source_agent:
        source_dir = agents_dir / source_agent
        source_md = source_dir / "AGENTS.md"

        if not source_md.exists():
            console.print(
                f"[bold red]Error:[/bold red] Source agent '{source_agent}' not found "
                "or has no AGENTS.md\n"
                "  Available agents: dcode agents list"
            )
            raise SystemExit(1)

        source_content = source_md.read_text()
        action_desc = f"contents of agent '{source_agent}'"
    else:
        source_content = get_default_coding_instructions()
        action_desc = "default"

    if dry_run:
        if output_format == "json":
            from deepagents_code.output import write_json

            write_json(
                "reset",
                {
                    "agent": agent_name,
                    "reset_to": source_agent or "default",
                    "path": str(agent_dir),
                    "dry_run": True,
                },
            )
            return
        exists = "remove and recreate" if agent_dir.exists() else "create"
        console.print(f"Would {exists} {agent_dir} with {action_desc} prompt.")
        console.print("No changes made.", style=theme.MUTED)
        return

    if agent_dir.exists():
        shutil.rmtree(agent_dir)
        if output_format != "json":
            console.print(
                f"Removed existing agent directory: {agent_dir}", style=theme.WARNING
            )

    agent_dir.mkdir(parents=True, exist_ok=True)
    agent_md = agent_dir / "AGENTS.md"
    agent_md.write_text(source_content)

    if output_format == "json":
        from deepagents_code.output import write_json

        write_json(
            "reset",
            {
                "agent": agent_name,
                "reset_to": source_agent or "default",
                "path": str(agent_dir),
            },
        )
        return

    console.print(
        f"{get_glyphs().checkmark} Agent '{agent_name}' reset to {action_desc}",
        style=theme.PRIMARY,
    )
    console.print(f"Location: {agent_dir}\n", style=theme.MUTED)


MODEL_IDENTITY_RE = re.compile(r"### Model Identity\n\n.*?(?=###|\Z)", re.DOTALL)
"""Matches the `### Model Identity` section in the system prompt, up to the
next heading or end of string."""


def build_model_identity_section(
    name: str | None,
    provider: str | None = None,
    context_limit: int | None = None,
    unsupported_modalities: frozenset[str] = frozenset(),
) -> str:
    """Build the `### Model Identity` section for the system prompt.

    Args:
        name: Model identifier (e.g. `claude-opus-4-6`).
        provider: Provider identifier (e.g. `anthropic`).
        context_limit: Max input tokens from the model profile.
        unsupported_modalities: Input modalities not indicated as supported by
            the model profile (e.g. `{"audio", "video"}`).

    Returns:
        The section text including the heading and trailing newline,
        or an empty string if `name` is falsy.
    """
    if not name:
        return ""
    section = f"### Model Identity\n\nYou are running as model `{name}`"
    if provider:
        section += f" (provider: {provider})"
    section += ".\n"
    if context_limit:
        section += f"Your context window is {context_limit:,} tokens.\n"
    if unsupported_modalities:
        items = sorted(unsupported_modalities)
        if len(items) == 1:
            joined = items[0]
        elif len(items) == 2:  # noqa: PLR2004
            joined = f"{items[0]} and {items[1]}"
        else:
            joined = ", ".join(items[:-1]) + f", and {items[-1]}"
        section += (
            f"{joined.capitalize()} input may not be available for this model. "
            "Do not attempt to read or process these content types.\n"
        )
    section += "\n"
    return section


def get_system_prompt(
    assistant_id: str,
    sandbox_type: str | None = None,
    *,
    interactive: bool = True,
    cwd: str | Path | None = None,
) -> str:
    """Get the base system prompt for the agent.

    Loads the base system prompt template from `system_prompt.md` and
    interpolates dynamic sections (model identity, working directory,
    skills path, execution mode, and todo-list guidance for
    interactive vs headless).

    Args:
        assistant_id: The agent identifier for path references
        sandbox_type: Type of sandbox provider
            (`'agentcore'`, `'daytona'`, `'langsmith'`, `'modal'`, `'runloop'`).

            If `None`, agent is operating in local mode.
        interactive: When `False`, the prompt is tailored for headless
            non-interactive execution (no human in the loop).
        cwd: Override the working directory shown in the prompt.

    Returns:
        The system prompt string

    Example:
        ```txt
        You are running as model {MODEL} (provider: {PROVIDER}).

        Your context window is {CONTEXT_WINDOW} tokens.

        ... {CONDITIONAL SECTIONS} ...
        ```
    """
    template = (Path(__file__).parent / "system_prompt.md").read_text()

    skills_path = f"~/.deepagents/{assistant_id}/skills"

    if interactive:
        mode_description = "an interactive TUI on the user's computer"
        interactive_preamble = (
            "The user sends you messages and you respond with text and tool "
            "calls. Your tools run on the user's machine. The user can see "
            "your responses and tool outputs in real time, so keep them "
            "informed — but don't over-explain."
        )
        ambiguity_guidance = (
            "- If the request is ambiguous, ask questions before acting.\n"
            "- If asked how to approach something, explain first, then act."
        )
        todo_guidance = (
            "6. When first creating a todo list for a task, ALWAYS ask the user if "
            "the plan looks good before starting work\n"
            '   - Create the todos, then ask: "Does this plan '
            'look good?" or similar\n'
            "   - Wait for the user's response before marking the first todo as "
            "in_progress\n"
            "7. Update todo status promptly as you complete each item"
        )
    else:
        mode_description = (
            "non-interactive (headless) mode — there is no human operator "
            "monitoring your output in real time"
        )
        interactive_preamble = (
            "You received a single task and must complete it fully and "
            "autonomously. There is no human available to answer follow-up "
            "questions, so do NOT ask for clarification — make reasonable "
            "assumptions and proceed."
        )
        ambiguity_guidance = (
            "- Do NOT ask clarifying questions — there is no human to answer "
            "them. Make reasonable assumptions and proceed.\n"
            "- If you encounter ambiguity, choose the most reasonable "
            "interpretation and note your assumption briefly.\n"
            "- Always use non-interactive command variants — no human is "
            "available to respond to prompts. Examples: `npm init -y` not "
            "`npm init`, `apt-get install -y` not `apt-get install`, "
            "`yes |` or `--no-input`/`--non-interactive` flags where "
            "available. Never run commands that block waiting for stdin."
        )
        todo_guidance = (
            "6. There is no human operator in this mode — do NOT ask the user to "
            "approve your plan or wait for a reply.\n"
            "   After you create todos for a multi-step task, mark the first item "
            "`in_progress` immediately and start work.\n"
            "   If the plan needs adjustment, revise the todo list yourself; do "
            "not block on human confirmation.\n"
            "7. Update todo status promptly as you complete each item"
        )

    model_identity_section = build_model_identity_section(
        settings.model_name,
        provider=settings.model_provider,
        context_limit=settings.model_context_limit,
        unsupported_modalities=settings.model_unsupported_modalities,
    )

    # Build working directory section (local vs sandbox)
    if sandbox_type:
        working_dir = get_default_working_dir(sandbox_type)
        working_dir_section = (
            f"### Current Working Directory\n\n"
            f"You are operating in a **remote Linux sandbox** at `{working_dir}`.\n\n"
            f"All code execution and file operations happen in this sandbox "
            f"environment.\n\n"
            f"**Important:**\n"
            f"- The application is running locally on the user's machine, but you "
            f"execute code remotely\n"
            f"- Use `{working_dir}` as your working directory for all operations\n"
            f"- **You do NOT have access to the user's local filesystem.** Paths "
            f"like `/Users/...`, `/home/<local-user>/...`, `C:\\...`, etc. do not "
            f"exist in this sandbox. Never reference or attempt to read/write local "
            f"paths — all files must be within the sandbox at `{working_dir}`\n"
            f"- When delegating to subagents, ensure they also use sandbox paths "
            f"(`{working_dir}/...`), not local paths\n\n"
        )
    else:
        if cwd is not None:
            resolved_cwd = Path(cwd)
        else:
            try:
                resolved_cwd = Path.cwd()
            except OSError:
                logger.warning(
                    "Could not determine working directory for system prompt",
                    exc_info=True,
                )
                resolved_cwd = Path()
        cwd = resolved_cwd
        working_dir_section = (
            f"### Current Working Directory\n\n"
            f"The filesystem backend is currently operating in: `{cwd}`\n\n"
            f"### File System and Paths\n\n"
            f"**IMPORTANT - Path Handling:**\n"
            f"- All file paths must be absolute paths (e.g., `{cwd}/file.txt`)\n"
            f"- Use the working directory to construct absolute paths\n"
            f"- Example: To create a file in your working directory, "
            f"use `{cwd}/research_project/file.md`\n"
            f"- Never use relative paths - always construct full absolute paths\n\n"
        )

    result = (
        template.replace("{mode_description}", mode_description)
        .replace("{interactive_preamble}", interactive_preamble)
        .replace("{ambiguity_guidance}", ambiguity_guidance)
        .replace("{todo_guidance}", todo_guidance)
        .replace("{model_identity_section}", model_identity_section)
        .replace("{working_dir_section}", working_dir_section)
        .replace("{skills_path}", skills_path)
    )

    # Detect unreplaced placeholders (defense-in-depth for template typos)
    unreplaced = re.findall(r"\{[a-z_]+\}", result)
    if unreplaced:
        logger.warning("System prompt contains unreplaced placeholders: %s", unreplaced)

    return result


def _format_write_file_description(
    tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format write_file tool call for approval prompt.

    Returns:
        Formatted description string for the write_file tool call.
    """
    args = tool_call["args"]
    file_path = args.get("file_path", "unknown")

    action = "Overwrite" if Path(file_path).exists() else "Create"

    return f"Action: {action} file"


def _format_edit_file_description(
    tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format edit_file tool call for approval prompt.

    Returns:
        Formatted description string for the edit_file tool call.
    """
    args = tool_call["args"]
    replace_all = bool(args.get("replace_all", False))

    scope = "all occurrences" if replace_all else "single occurrence"
    return f"Action: Replace text ({scope})"


def _format_delete_description(
    _tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format delete tool call for approval prompt.

    Returns:
        Formatted description string for the delete tool call.
    """
    return "Action: Delete file or directory"


def _format_web_search_description(
    tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format web_search tool call for approval prompt.

    Returns:
        Formatted description string for the web_search tool call.
    """
    args = tool_call["args"]
    query = args.get("query", "unknown")
    max_results = args.get("max_results", 5)

    return (
        f"Query: {query}\nMax results: {max_results}\n\n"
        f"{get_glyphs().warning}  This will use Tavily API credits"
    )


def _format_fetch_url_description(
    tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format fetch_url tool call for approval prompt.

    Returns:
        Formatted description string for the fetch_url tool call.
    """
    args = tool_call["args"]
    url = str(args.get("url", "unknown"))
    display_url = strip_dangerous_unicode(url)
    timeout = args.get("timeout", 30)
    safety = check_url_safety(url)

    warning_lines: list[str] = []
    if not safety.safe:
        detail = format_warning_detail(safety.warnings)
        warning_lines.append(f"{get_glyphs().warning}  URL warning: {detail}")
    if safety.decoded_domain:
        warning_lines.append(
            f"{get_glyphs().warning}  Decoded domain: {safety.decoded_domain}"
        )

    warning_block = "\n".join(warning_lines)
    if warning_block:
        warning_block = f"\n{warning_block}"

    return (
        f"URL: {display_url}\nTimeout: {timeout}s\n\n"
        f"{get_glyphs().warning}  Will fetch and convert web content to markdown"
        f"{warning_block}"
    )


def _format_task_description(
    tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format task (subagent) tool call for approval prompt.

    The task tool signature is: task(description: str, subagent_type: str)
    The description contains all instructions that will be sent to the subagent.

    Returns:
        Formatted description string for the task tool call.
    """
    args = tool_call["args"]
    description = args.get("description", "unknown")
    subagent_type = args.get("subagent_type", "unknown")

    # Truncate description if too long for display
    description_preview = description
    if len(description) > 500:  # noqa: PLR2004  # Subagent description length threshold
        description_preview = description[:500] + "..."

    glyphs = get_glyphs()
    separator = glyphs.box_horizontal * 40
    warning_msg = "Subagent will have access to file operations and shell commands"
    return (
        f"Subagent Type: {subagent_type}\n\n"
        f"{glyphs.warning} {warning_msg} {glyphs.warning}\n\n"
        f"Task Instructions:\n"
        f"{separator}\n"
        f"{description_preview}"
    )


def _format_execute_description(
    tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
) -> str:
    """Format execute tool call for approval prompt.

    Returns:
        Formatted description string for the execute tool call.
    """
    args = tool_call["args"]
    command_raw = str(args.get("command", "N/A"))
    command = strip_dangerous_unicode(command_raw)
    project_context = get_server_project_context()
    effective_cwd = (
        str(project_context.user_cwd)
        if project_context is not None
        else str(Path.cwd())
    )
    lines = [f"Execute Command: {command}", f"Working Directory: {effective_cwd}"]

    issues = detect_dangerous_unicode(command_raw)
    if issues:
        summary = summarize_issues(issues)
        lines.append(f"{get_glyphs().warning}  Hidden Unicode detected: {summary}")
        raw_marked = render_with_unicode_markers(command_raw)
        if len(raw_marked) > 220:  # noqa: PLR2004  # UI display truncation threshold
            raw_marked = raw_marked[:220] + "..."
        lines.append(f"Raw: {raw_marked}")

    return "\n".join(lines)


def _is_auto_approve_enabled(value: object) -> bool:
    """Return whether a context value explicitly enables auto-approve."""
    return isinstance(value, bool) and value


def _read_live_auto_approve(store: object, key: str | None) -> bool | None:
    """Return live approval mode from the LangGraph Store when configured.

    Args:
        store: `request.runtime.store` from the graph server.
        key: Live approval-mode store key, or `None` when this run has no live
            control record.

    Returns:
        `None` when no live key is configured for this run — the caller should
            fall back to the static `auto_approve` context snapshot.
        `True` or `False` when a live key is configured: these reflect
            the stored mode, and `False` is also returned when the key
            is configured but the store is unreadable (missing item,
            malformed value, read error), so an unreadable live mode fails
            closed and interrupts.
        `None` therefore means "feature not in play," the opposite of the store
            reader's `None` ("unreadable, be careful").
    """
    if not key:
        return None
    from deepagents_code.approval_mode import read_approval_mode_from_store

    value = read_approval_mode_from_store(store, key)
    if value is None:
        logger.warning(
            "Approval-mode store item is unavailable; interrupting for safety"
        )
        return False
    return value


def _should_interrupt_tool_call(request: ToolCallRequest) -> bool:
    """Decide whether a gated tool call should pause for human approval.

    Returns `False` once the run context carries `auto_approve=True` so
    `HumanInTheLoopMiddleware` skips the interrupt entirely. This avoids the
    interrupt-then-auto-resolve pattern that previously split each turn into a
    separate run after every tool call, producing noisy traces.

    Auto-approve is read from the run-scoped `CLIContext` (set by the client)
    rather than graph state. Sourcing it from state required seeding it with a
    first-turn `Command(update=...)`, which the LangGraph API server rebuilds
    with `goto=None` — crashing `_control_branch` on a fresh thread. Context
    is also safer: the model cannot self-approve by writing state.

    Args:
        request: The pending tool call under review.

    Returns:
        `True` to interrupt for approval, `False` to auto-approve.
    """
    runtime = getattr(request, "runtime", None)
    ctx = getattr(runtime, "context", None)
    store = getattr(runtime, "store", None)
    if isinstance(ctx, CLIContextSchema):
        if (live := _read_live_auto_approve(store, ctx.approval_mode_key)) is not None:
            return not live
        return not _is_auto_approve_enabled(ctx.auto_approve)
    if isinstance(ctx, dict):
        raw_key = ctx.get("approval_mode_key")
        key = raw_key if isinstance(raw_key, str) else None
        if (live := _read_live_auto_approve(store, key)) is not None:
            return not live
        # Type-checked (not truthiness) check: over the JSON/RemoteGraph boundary a
        # malformed payload (e.g. "yes", 1) must fail closed and interrupt, not
        # silently auto-approve. Only a genuine boolean `True` suppresses.
        return not _is_auto_approve_enabled(ctx.get("auto_approve"))
    if ctx is not None:
        # Context is present but neither expected shape. The registered
        # `context_schema=CLIContextSchema` guarantees in-process coercion to
        # that dataclass, and RemoteGraph delivers a dict — so this means the
        # context-plumbing contract broke (likely an SDK change). Fail closed
        # (interrupt), but surface it: otherwise auto-approve silently stops
        # working with no error, looking like a feature that just "broke".
        logger.warning(
            "auto-approve predicate received unexpected context type %s; "
            "interrupting for safety",
            type(ctx).__name__,
        )
    return True


def _add_interrupt_on() -> dict[str, InterruptOnConfig]:
    """Configure human-in-the-loop interrupt settings for all gated tools.

    Every tool that can have side effects or access external resources
    (shell execution, file writes/edits, web search, URL fetch, task
    delegation) is gated behind an approval prompt unless auto-approve
    is enabled.

    Each config carries a `when` predicate so that enabling "approve always"
    mid-session (carried in run-scoped context, not graph state) suppresses
    the interrupt itself instead of relying on the client to auto-resolve it.

    Returns:
        Dictionary mapping tool names to their interrupt configuration.
    """
    execute_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_execute_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    write_file_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_write_file_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    edit_file_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_edit_file_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    delete_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_delete_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    web_search_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_web_search_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    fetch_url_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_fetch_url_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    task_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": _format_task_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
        "when": _should_interrupt_tool_call,
    }

    async_subagent_interrupt_config: InterruptOnConfig = {
        "allowed_decisions": ["approve", "reject"],
        "description": "Launch, update, or cancel a remote async subagent.",
        "when": _should_interrupt_tool_call,
    }

    interrupt_map: dict[str, InterruptOnConfig] = {
        "execute": execute_interrupt_config,
        "write_file": write_file_interrupt_config,
        "edit_file": edit_file_interrupt_config,
        "delete": delete_interrupt_config,
        "web_search": web_search_interrupt_config,
        "fetch_url": fetch_url_interrupt_config,
        "task": task_interrupt_config,
        "start_async_task": async_subagent_interrupt_config,
        "update_async_task": async_subagent_interrupt_config,
        "cancel_async_task": async_subagent_interrupt_config,
    }

    if REQUIRE_COMPACT_TOOL_APPROVAL:
        interrupt_map["compact_conversation"] = {
            "allowed_decisions": ["approve", "reject"],
            "description": (
                "Offloads older messages to backend storage and "
                "replaces them with a summary, freeing context "
                "window space. Recent messages are kept as-is. "
                "Full history remains available for retrieval."
            ),
            "when": _should_interrupt_tool_call,
        }

    return interrupt_map


def _apply_inherited_pythonpath(env: dict[str, str]) -> None:
    """Re-apply a relayed launch-time `PYTHONPATH` to a shell-command env.

    `server._build_server_env` strips `PYTHONPATH` from the server interpreter
    and relays the launch value via `config._INHERITED_PYTHONPATH_ENV`. This
    restores it as `PYTHONPATH` for the approval-gated `execute` subprocesses,
    which run in the user's working directory and need the import path. Mutates
    `env` in place; a no-op when no value was relayed.

    Args:
        env: Environment mapping for the shell backend, modified in place.
    """
    inherited = env.pop(_INHERITED_PYTHONPATH_ENV, None)
    if inherited is not None:
        env["PYTHONPATH"] = inherited


def create_cli_agent(
    model: str | BaseChatModel,
    assistant_id: str,
    *,
    tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
    sandbox: SandboxBackendProtocol | None = None,
    sandbox_type: str | None = None,
    system_prompt: str | None = None,
    interactive: bool = True,
    auto_approve: bool = False,
    interrupt_shell_only: bool = False,
    shell_allow_list: list[str] | None = None,
    enable_ask_user: bool = True,
    enable_memory: bool = True,
    enable_skills: bool = True,
    enable_shell: bool = True,
    enable_interpreter: bool = False,
    checkpointer: BaseCheckpointSaver | None = None,
    mcp_server_info: list[MCPServerInfo] | None = None,
    cwd: str | Path | None = None,
    project_context: ProjectContext | None = None,
    async_subagents: list[AsyncSubAgent] | None = None,
) -> tuple[Pregel[Any, Any, Any, Any], CompositeBackend]:
    """Create a CLI-configured agent with flexible options.

    This is the main entry point for creating a Deep Agents Code agent, usable
    both internally and from external code (e.g., benchmarking frameworks).

    Args:
        model: LLM model to use (e.g., `'provider:model'`)
        assistant_id: Agent identifier for memory/state storage
        tools: Additional tools to provide to agent
        sandbox: Optional sandbox backend for remote execution
            (e.g., `ModalSandbox`).

            If `None`, uses local filesystem + shell.
        sandbox_type: Type of sandbox provider
            (`'agentcore'`, `'daytona'`, `'langsmith'`, `'modal'`, `'runloop'`).
            Used for system prompt generation.
        system_prompt: Override the default system prompt.

            If `None`, generates one based on `sandbox_type`, `assistant_id`,
            and `interactive`.
        interactive: When `False`, the auto-generated system prompt is
            tailored for headless non-interactive execution. Ignored when
            `system_prompt` is provided explicitly.
        auto_approve: If `True`, no tools trigger human-in-the-loop
            interrupts — all calls (shell execution, file writes/edits,
            web search, URL fetch) run automatically.

            If `False`, tools pause for user confirmation via the approval menu.
            See `_add_interrupt_on` for the full list of gated tools.
        interrupt_shell_only: If `True`, all HITL interrupts are disabled;
            shell commands are validated inline by `ShellAllowListMiddleware`
            against the configured allow-list instead.

            Used in non-interactive mode with a restrictive shell allow-list
            to avoid splitting traces into multiple LangSmith runs.

            Has no effect when `auto_approve` is `True` (interrupts are already
            disabled) or when `shell_allow_list` is `SHELL_ALLOW_ALL`.
        shell_allow_list: Explicit restrictive shell allow-list forwarded from
            the CLI process. When provided (and `interrupt_shell_only` is
            `True`), used directly instead of reading `settings.shell_allow_list`
            (which may not be set in the server subprocess environment).
        enable_ask_user: Enable `AskUserMiddleware` so the agent can ask
            clarifying questions.

            Disabled in non-interactive mode.
        enable_memory: Enable `MemoryMiddleware` for persistent memory
        enable_skills: Enable `SkillsMiddleware` for custom agent skills
        enable_shell: Enable shell execution via `LocalShellBackend`
            (only in local mode). When enabled, the `execute` tool is available.
        enable_interpreter: Wire `CodeInterpreterMiddleware` from
            `langchain-quickjs` into the main agent.

            Local-mode only — passing a non-`None` `sandbox` while
            `enable_interpreter=True` raises `ValueError`. Subagents do not
            receive the interpreter in v1.

            PTC (`tools.*` host bridge) calls bypass `interrupt_on`/HITL
            approval, so `settings.interpreter_ptc` is the only effective
            control over which host tools can be invoked from inside the
            REPL. `js_eval` itself is intentionally not gated by HITL —
            per-call approval would be unusably noisy and would not block
            PTC fan-out anyway. The `"safe"` preset is therefore restricted
            to tools that are already non-HITL outside the REPL (read-only
            file inspection); exposing HITL-gated tools — network fetch,
            subagent dispatch, shell, file writes — requires an explicit
            list or `interpreter_ptc="all"` with
            `interpreter_ptc_acknowledge_unsafe=True`.

            Requires the core `langchain-quickjs` dependency.
        checkpointer: Optional checkpointer for session persistence.
            When `None`, the graph is compiled without a checkpointer.
        mcp_server_info: MCP server metadata to surface in the system prompt.
        cwd: Override the working directory for the agent's filesystem backend
            and system prompt.
        project_context: Explicit project path context for project-sensitive
            behavior such as project `AGENTS.md` files, skills, subagents, and
            MCP trust.
        async_subagents: Remote LangGraph deployments to expose as async subagent tools.

            Loaded from `[async_subagents]` in `config.toml` or passed directly.

    Returns:
        2-tuple of `(agent_graph, backend)`

            - `agent_graph`: Configured LangGraph Pregel instance ready
                for execution
            - `composite_backend`: `CompositeBackend` for file operations

    Raises:
        ValueError: When `enable_interpreter=True` is paired with a
            non-`None` `sandbox`, when `settings.interpreter_ptc` contains
            unknown tool names, or when `interpreter_ptc="all"` is used
            without `auto_approve` or `interpreter_ptc_acknowledge_unsafe`.
    """
    tools = tools or []
    effective_cwd = (
        Path(cwd)
        if cwd is not None
        else (project_context.user_cwd if project_context is not None else None)
    )

    # Setup agent directory for persistent memory (if enabled)
    if enable_memory or enable_skills:
        agent_dir = settings.ensure_agent_dir(assistant_id)
        agent_md = agent_dir / "AGENTS.md"
        if not agent_md.exists():
            # Create empty file for user customizations
            # Base instructions are loaded fresh from get_system_prompt()
            agent_md.touch()

    # Skills directories (if enabled)
    skills_dir = None
    user_agent_skills_dir = None
    project_skills_dir = None
    project_agent_skills_dir = None
    if enable_skills:
        skills_dir = settings.ensure_user_skills_dir(assistant_id)
        user_agent_skills_dir = settings.get_user_agent_skills_dir()
        project_skills_dir = (
            project_context.project_skills_dir()
            if project_context is not None
            else settings.get_project_skills_dir()
        )
        project_agent_skills_dir = (
            project_context.project_agent_skills_dir()
            if project_context is not None
            else settings.get_project_agent_skills_dir()
        )

    # Load custom subagents from filesystem
    custom_subagents: list[SubAgent | CompiledSubAgent] = []
    restrictive_shell_allow_list: list[str] | None = None
    if interrupt_shell_only and not auto_approve:
        # Prefer the explicitly forwarded allow-list (set by the CLI process
        # and passed through ServerConfig).  Fall back to settings only for
        # direct callers (e.g. benchmarking frameworks) that don't go through
        # the server subprocess path.
        if shell_allow_list:
            restrictive_shell_allow_list = list(shell_allow_list)
        elif settings.shell_allow_list and not isinstance(
            settings.shell_allow_list, _ShellAllowAll
        ):
            restrictive_shell_allow_list = list(settings.shell_allow_list)
        else:
            logger.warning(
                "interrupt_shell_only=True but no restrictive shell allow-list "
                "available; falling back to standard HITL interrupts"
            )

    user_agents_dir = settings.get_user_agents_dir(assistant_id)
    project_agents_dir = (
        project_context.project_agents_dir()
        if project_context is not None
        else settings.get_project_agents_dir()
    )

    def _subagent_cli_middleware(*, has_explicit_model: bool) -> list[AgentMiddleware]:
        middleware: list[AgentMiddleware] = []
        if not has_explicit_model:
            middleware.append(ConfigurableModelMiddleware())
        if restrictive_shell_allow_list is not None:
            middleware.append(ShellAllowListMiddleware(restrictive_shell_allow_list))
        # Subagents share the on-disk filesystem backend and can edit the user
        # AGENTS.md, so they get the same managed onboarding-name block guard as
        # the main agent. Gated on memory because the block only exists when
        # memory is enabled.
        if enable_memory:
            from deepagents_code.memory_guard import ManagedMemoryGuardMiddleware

            middleware.append(
                ManagedMemoryGuardMiddleware(
                    [settings.get_user_agent_md_path(assistant_id)]
                )
            )
        return middleware

    for subagent_meta in list_subagents(
        user_agents_dir=user_agents_dir,
        project_agents_dir=project_agents_dir,
    ):
        # Treat a falsy spec (`None` or `""`) as "no explicit model" so an empty
        # `model:` in subagent frontmatter inherits the runtime model rather than
        # being forwarded verbatim to `resolve_model("")`.
        model_spec = subagent_meta["model"]
        has_explicit_model = bool(model_spec)
        subagent: SubAgent = {
            "name": subagent_meta["name"],
            "description": subagent_meta["description"],
            "system_prompt": subagent_meta["system_prompt"],
        }
        if model_spec:
            subagent["model"] = model_spec
        subagent_middleware = _subagent_cli_middleware(
            has_explicit_model=has_explicit_model
        )
        if subagent_middleware:
            subagent["middleware"] = subagent_middleware
        custom_subagents.append(subagent)

    from deepagents.middleware.subagents import (
        GENERAL_PURPOSE_SUBAGENT,
        SubAgent as RuntimeSubAgent,
    )

    if not any(
        subagent["name"] == GENERAL_PURPOSE_SUBAGENT["name"]
        for subagent in custom_subagents
    ):
        general_purpose_subagent: RuntimeSubAgent = {
            "name": GENERAL_PURPOSE_SUBAGENT["name"],
            "description": GENERAL_PURPOSE_SUBAGENT["description"],
            "system_prompt": GENERAL_PURPOSE_SUBAGENT["system_prompt"],
            "middleware": _subagent_cli_middleware(has_explicit_model=False),
        }
        custom_subagents.append(general_purpose_subagent)

    # Build middleware stack based on enabled features
    agent_middleware: list[AgentMiddleware[Any, Any]] = [
        ConfigurableModelMiddleware(),
    ]

    # Resume state: declares the `_context_tokens` and `_model_spec` channels
    # and writes them from `after_model` (token count from the latest
    # `AIMessage.usage_metadata`, model spec from `context["effective_model"]`).
    # The CLI reads them back from `state_values` on thread resume.
    from deepagents_code.resume_state import ResumeStateMiddleware

    agent_middleware.append(ResumeStateMiddleware())

    # Add ask_user middleware (must be early so its tool is available)
    if enable_ask_user:
        from deepagents_code.ask_user import AskUserMiddleware

        agent_middleware.append(AskUserMiddleware())

    # Add memory middleware
    if enable_memory:
        memory_sources = [str(settings.get_user_agent_md_path(assistant_id))]
        project_agent_md_paths = (
            project_context.project_agent_md_paths()
            if project_context is not None
            else settings.get_project_agent_md_path()
        )
        memory_sources.extend(str(p) for p in project_agent_md_paths)

        agent_middleware.append(
            MemoryMiddleware(
                backend=FilesystemBackend(virtual_mode=False),
                sources=memory_sources,
            )
        )

        # Protect the machine-managed onboarding-name block in the user
        # AGENTS.md from being rewritten by agent file edits. The block's
        # markers are HTML comments stripped before injection, so the model
        # can't see the boundary and would otherwise clobber it.
        from deepagents_code.memory_guard import ManagedMemoryGuardMiddleware

        agent_middleware.append(
            ManagedMemoryGuardMiddleware(
                [settings.get_user_agent_md_path(assistant_id)]
            )
        )

    # Add skills middleware
    if enable_skills:
        # Lowest to highest precedence:
        # built-in -> user .deepagents -> user .agents
        # -> project .deepagents -> project .agents
        # -> user .claude (experimental) -> project .claude (experimental)
        # Labels disambiguate user- vs project-scoped sources that share a
        # `.../skills` leaf; the middleware would otherwise derive identical
        # labels from the parent directory name.
        sources: list[tuple[str, str]] = [
            (str(settings.get_built_in_skills_dir()), "Built-in"),
            (str(skills_dir), "User Deepagents"),
            (str(user_agent_skills_dir), "User Agents"),
        ]
        if project_skills_dir:
            sources.append((str(project_skills_dir), "Project Deepagents"))
        if project_agent_skills_dir:
            sources.append((str(project_agent_skills_dir), "Project Agents"))

        # Experimental: Claude Code skill directories
        user_claude_skills_dir = settings.get_user_claude_skills_dir()
        if user_claude_skills_dir.exists():
            sources.append((str(user_claude_skills_dir), "User Claude"))
        project_claude_skills_dir = settings.get_project_claude_skills_dir()
        if project_claude_skills_dir:
            sources.append((str(project_claude_skills_dir), "Project Claude"))

        # Backwards-compat: strip labels when the installed SDK is too old
        # to accept `(path, label)` tuples. Label-based disambiguation
        # regresses to the pre-alias behavior (user- and project-scoped
        # `.claude/skills` collapse to the same label), but functionality
        # is preserved.
        middleware_sources: Sequence[str | tuple[str, str]] = (
            sources if _SUPPORTS_SKILL_SOURCE_TUPLES else [path for path, _ in sources]
        )

        agent_middleware.append(
            SkillsMiddleware(
                backend=FilesystemBackend(virtual_mode=False),
                sources=middleware_sources,
            )
        )

    # CONDITIONAL SETUP: Local vs Remote Sandbox
    if sandbox is None:
        # ========== LOCAL MODE ==========
        root_dir = effective_cwd if effective_cwd is not None else Path.cwd()
        if enable_shell:
            # Create environment for shell commands.
            # Restore the user's original LANGSMITH_PROJECT so their code traces
            # separately. When they had none, drop the agent's override (the
            # `deepagents-code` default applied at bootstrap) entirely so shell
            # commands don't inherit it.
            shell_env = os.environ.copy()
            if settings.user_langchain_project is not None:
                shell_env["LANGSMITH_PROJECT"] = settings.user_langchain_project
            else:
                shell_env.pop("LANGSMITH_PROJECT", None)
            restore_user_tracing_env(shell_env)
            # Re-apply a launch-time PYTHONPATH that was stripped from the server
            # interpreter but relayed for approval-gated `execute` commands.
            _apply_inherited_pythonpath(shell_env)

            # Use LocalShellBackend for filesystem + shell execution.
            # The SDK's FilesystemMiddleware exposes per-command timeout
            # on the execute tool natively.
            # `inherit_env=False`: `shell_env` is already a complete, curated
            # copy of `os.environ`. Inheriting again would re-copy `os.environ`
            # and resurrect the popped carrier var, leaking it into `execute`.
            backend = LocalShellBackend(
                root_dir=root_dir,
                virtual_mode=False,
                inherit_env=False,
                env=shell_env,
            )
        else:
            # No shell access - use plain FilesystemBackend
            backend = FilesystemBackend(root_dir=root_dir, virtual_mode=False)
    else:
        # ========== REMOTE SANDBOX MODE ==========
        backend = sandbox  # Remote sandbox (ModalSandbox, etc.)
        # Note: Shell middleware not used in sandbox mode
        # File operations and execute tool are provided by the sandbox backend

    if enable_interpreter:
        if sandbox is not None:
            msg = (
                "enable_interpreter=True is not supported with a remote "
                "sandbox in this release. Disable the sandbox or unset "
                "enable_interpreter."
            )
            raise ValueError(msg)
        # Lazy import keeps `dcode -v` fast — see AGENTS.md startup-perf rule.
        from langchain_quickjs import CodeInterpreterMiddleware, PTCOption

        ptc_names = _resolve_ptc_option(
            settings.interpreter_ptc,
            tools=tools,
            acknowledge_unsafe=settings.interpreter_ptc_acknowledge_unsafe,
            auto_approve=auto_approve,
        )
        ptc_option: PTCOption | None = (
            cast("PTCOption", list(ptc_names)) if ptc_names is not None else None
        )
        agent_middleware.append(
            CodeInterpreterMiddleware(
                tool_name="js_eval",
                timeout=settings.interpreter_timeout_seconds,
                memory_limit=settings.interpreter_memory_limit_mb * 1024 * 1024,
                max_ptc_calls=settings.interpreter_max_ptc_calls,
                max_result_chars=settings.interpreter_max_result_chars,
                ptc=ptc_option,
            )
        )

    # Local context middleware (git info, directory tree, etc.).
    if isinstance(backend, (_ExecutableBackend, _AsyncExecutableBackend)):
        agent_middleware.append(
            LocalContextMiddleware(
                backend=backend,
                mcp_server_info=mcp_server_info,
                tracing_project=get_langsmith_project_name(),
                user_tracing_project=settings.user_langchain_project,
            )
        )

    # Add shell allow-list middleware when interrupt_shell_only is active.
    shell_middleware_added = False
    if restrictive_shell_allow_list is not None:
        agent_middleware.append(ShellAllowListMiddleware(restrictive_shell_allow_list))
        shell_middleware_added = True

    # Get or use custom system prompt
    if system_prompt is None:
        system_prompt = get_system_prompt(
            assistant_id=assistant_id,
            sandbox_type=sandbox_type,
            interactive=interactive,
            cwd=effective_cwd,
        )

    # Configure interrupt_on based on auto_approve / shell_middleware_added
    interrupt_on: dict[str, bool | InterruptOnConfig] | None = None
    if auto_approve or shell_middleware_added:  # noqa: SIM108  # if-else clearer than ternary for dual-path config
        # No HITL interrupts — tools run automatically.
        # When shell_middleware_added is True, shell validation is handled by
        # ShellAllowListMiddleware (added above) which rejects disallowed
        # commands inline as error ToolMessages, keeping the entire run in
        # a single LangSmith trace.
        interrupt_on = {}
    else:
        # Full HITL for destructive operations
        interrupt_on = _add_interrupt_on()  # ty: ignore[invalid-assignment]  # InterruptOnConfig is compatible at runtime

    # Set up composite backend with routing
    # For local FilesystemBackend, route large tool results to /tmp to avoid polluting
    # the working directory. For sandbox backends, no special routing is needed.
    if sandbox is None:
        # Local mode: Route large results to a unique temp directory
        large_results_backend = FilesystemBackend(
            root_dir=tempfile.mkdtemp(prefix="deepagents_large_results_"),
            virtual_mode=True,
        )
        conversation_history_backend = FilesystemBackend(
            root_dir=tempfile.mkdtemp(prefix="deepagents_conversation_history_"),
            virtual_mode=True,
        )
        composite_backend = CompositeBackend(
            default=backend,
            routes={
                "/large_tool_results/": large_results_backend,
                "/conversation_history/": conversation_history_backend,
            },
        )
    else:
        # Sandbox mode: No special routing needed
        composite_backend = CompositeBackend(
            default=backend,
            routes={},
        )

    from deepagents.middleware.summarization import create_summarization_tool_middleware

    agent_middleware.append(
        create_summarization_tool_middleware(model, composite_backend)
    )

    # Create the agent
    all_subagents: list[SubAgent | CompiledSubAgent | AsyncSubAgent] = [
        *custom_subagents,
        *(async_subagents or []),
    ]
    agent = create_deep_agent(
        model=model,
        system_prompt=system_prompt,
        tools=tools,
        backend=composite_backend,
        middleware=agent_middleware,
        interrupt_on=interrupt_on,
        context_schema=CLIContextSchema,
        checkpointer=checkpointer,
        subagents=all_subagents or None,
        name=_sanitize_agent_message_name(assistant_id),
    ).with_config(config)
    return agent, composite_backend
