#!/usr/bin/env python3
"""Terminal tool: run shell commands in the configured backend.

Backends (``TERMINAL_ENV``): local (default), docker, modal (direct or
managed gateway), daytona, vercel_sandbox, ssh, plus plugin-registered
backends (the ``singularity`` catalog plugin, ...). Handles background
processes, sandbox lifecycle (per-task cache, idle reaper, atexit teardown)
and sudo password plumbing.
Cloud-sandbox persistent filesystems preserve working state across sandbox
recreation but do NOT guarantee the same live sandbox or long-running
processes survive cleanup, idle reaping, or Hermes exit.

Companion modules (re-exported here, so ``tools.terminal_tool.<name>`` stays the
import/patch target): ``terminal_tool_config`` (TERMINAL_* reads, ``_quiet``),
``terminal_tool_backends`` (env builders + requirement checkers),
``terminal_tool_lifecycle`` (reaper/teardown/ensure_task_env),
``terminal_tool_sudo`` (sudo password + shell rewrites), ``terminal_tool_guards``
(pre-exec blocks), ``terminal_tool_background`` (background spawn),
``terminal_tool_result`` (foreground result post-processing).
"""

import json
import logging
import os
import sys
import time
import threading
import atexit
from dataclasses import dataclass
from typing import Optional, Dict, Any, List

logger = logging.getLogger(__name__)


def _redact_terminal_error_text(value: Any) -> str:
    """Force-redact text before serializing a terminal error envelope."""
    from agent.redact import redact_sensitive_text

    return redact_sensitive_text("" if value is None else str(value), force=True)


from tools.registry import tool_error
from tools.terminal_tool_lifecycle import (
    _cleanup_inactive_envs, _create_configured_env,
    _evict_environment_for_task, cleanup_all_environments, ensure_task_env,
)
from tools.terminal_tool_config import (
    _is_container_backend, _is_host_cwd, _is_mounted_host_cwd, _is_unusable_container_cwd,
    _is_windows_drive_path, _parse_env_var, _plugin_env_flag, _quiet, _safe_getcwd, _tenv, _tenv_bool,
    _in_sandbox_home, coerce_ssh_remote_cwd, translate_mounted_host_path,
)
from tools.terminal_tool_backends import (
    _REQUIREMENT_CHECKERS, _VERCEL_SANDBOX_DEFAULT_CWD, _check_plugin_requirements,
    _record_unavailable_reason, terminal_backend_unavailable_reason,
)
# display_hermes_home imported lazily at call site (stale-module safety during hermes update)
from tools.tool_backend_helpers import coerce_modal_mode, managed_nous_tools_enabled


def _safe_parse_import_env(name: str, default: Any, converter, type_label: str):
    """Parse a module-level numeric env var; a malformed value must never make
    the module unloadable at import time (CLI, ACP, tests, tool discovery)."""
    raw = os.getenv(name)
    if raw is None or raw == "":
        return default
    try:
        return converter(raw)
    except (TypeError, ValueError):
        logger.warning(
            "Invalid value for %s: %r (expected %s). Falling back to %r.",
            name, raw, type_label, default,
        )
        return default


# Hard cap on foreground timeout; override via TERMINAL_MAX_FOREGROUND_TIMEOUT env var.
FOREGROUND_MAX_TIMEOUT = _safe_parse_import_env("TERMINAL_MAX_FOREGROUND_TIMEOUT", 600, int, "integer")


# Approval / sudo-prompt UI callbacks (CLI registers prompt_toolkit-aware
# ones). Thread-local so overlapping ACP sessions, each on its own executor
# thread, can't stomp on each other (GHSA-qg5c-hvr5-hjgr). Gateway mode
# resolves approvals via the per-session queue in tools.approval instead.
_callback_tls = threading.local()


def _get_sudo_password_callback():
    return getattr(_callback_tls, "sudo_password", None)


def _get_approval_callback():
    return getattr(_callback_tls, "approval", None)


def set_sudo_password_callback(cb):
    """Register the CLI's sudo password prompt callback (per-thread slot)."""
    _callback_tls.sudo_password = cb


def set_approval_callback(cb):
    """Register the dangerous-command approval prompt callback (per-thread slot)."""
    _callback_tls.approval = cb


def _current_session_key() -> str:
    """Active gateway/WebUI session key, or "" outside sessions (ContextVar with
    the ``get_session_env`` os.environ fallback for CLI/cron/tests)."""
    from gateway.session_context import get_session_env

    return get_session_env("HERMES_SESSION_KEY", "")


def _current_session_profile() -> str:
    """Active session's Hermes profile name, or "" (same lookup discipline as
    :func:`_current_session_key`)."""
    from gateway.session_context import get_session_env

    return get_session_env("HERMES_SESSION_PROFILE", "")


def _container_profile() -> str:
    """Profile whose persistent container a session-bound command uses: the bound home's profile when it is not
    the session's own (a profile-routed subagent runs in its target's home while the session, and the delivery
    identity HERMES_SESSION_PROFILE carries, stay the parent's; that target may be the launch profile, reached
    from a routed profile's chat), else the session's. A session with no profile name is the launch profile's,
    so the launch home's own work keeps the session's key whatever the launch profile is named."""
    from hermes_constants import get_hermes_home_override, profile_name_for_home
    from tools.environments.local import _is_routed_home

    override = get_hermes_home_override()
    session = _current_session_profile()
    bound = profile_name_for_home(override) if override else None
    if bound and (bound != session if session else _is_routed_home(override)):
        return bound
    return session or "default"


from tools.approval import (
    check_all_command_guards as _check_all_guards_impl,
)


def _docker_volume_uses_host_path(volume_spec: str) -> bool:
    """Return True when a docker volume spec bind-mounts a host path."""
    if not isinstance(volume_spec, str):
        return False
    vol = volume_spec.strip()
    return bool(vol) and (
        vol.startswith(("/", "~", "./", "../")) or
        (len(vol) >= 3 and vol[1] == ":" and vol[2] in ("/", "\\"))
    )


def _docker_has_host_access(config: dict[str, Any]) -> bool:
    """Return True when a Docker sandbox exposes host paths through bind mounts."""
    if config.get("env_type") != "docker":
        return False
    if config.get("host_cwd") and config.get("docker_mount_cwd_to_workspace"):
        return True
    return any(_docker_volume_uses_host_path(vol) for vol in config.get("docker_volumes", []))


def _check_all_guards(command: str, env_type: str,
                      has_host_access: bool = False) -> dict:
    """Delegate to the consolidated command guard with the CLI callback."""
    return _check_all_guards_impl(command, env_type,
                                  approval_callback=_get_approval_callback(),
                                  has_host_access=has_host_access)


from tools.environments.base import EnvironmentConnectionError


# Tool description for LLM
TERMINAL_TOOL_DESCRIPTION = """Execute shell commands on the OS, shell and backend named in your environment section. The working directory, files and exported variables persist between calls: activate a virtualenv or export once, not before every command.

Use read_file, search_files, patch and write_file instead of cat/head/tail, grep/find/ls, sed/awk and echo/heredoc. Output is auto-truncated with the full text saved to a file; never pipe through head/tail to shorten it.

Foreground (default) returns as soon as the command finishes. To wait a fixed time, run the wait in the foreground; never background a sleep or polling loop. Background is only for commands that must keep running after this call returns; health-check a server in a separate call and manage jobs with process_manage. When a command changes directory, trust the result's "cwd" instead of prefixing later commands with cd.
"""

# Environment lifecycle state.
_active_environments: dict[str, Any] = {}
_last_activity: dict[str, float] = {}
_env_lock = threading.Lock()
_creation_locks: dict[str, threading.Lock] = {}  # Per-task locks for sandbox creation
_creation_locks_lock = threading.Lock()  # Protects _creation_locks dict itself
_cleanup_thread = None
_cleanup_running = False

# Once-per-process guard for the docker orphan reaper.
_docker_orphan_reaper_ran = False
_docker_orphan_reaper_lock = threading.Lock()


def _maybe_reap_docker_orphans(container_config: dict[str, Any]) -> None:
    """Run the docker orphan reaper once per process, if enabled.

    Sweeps Exited containers labeled ``hermes-agent=1`` for the current
    profile — leftovers of Hermes processes that died without firing
    ``atexit`` (SIGKILL, OOM, closed terminal). Conservative: only containers
    older than ``2 × lifetime_seconds``, profile-scoped. Gates:
    ``terminal.docker_orphan_reaper: false`` (operator opt-out, e.g. several
    Hermes processes sharing a profile) and the once-per-interpreter flag so
    parallel subagent / RL-rollout calls don't re-sweep.
    """
    global _docker_orphan_reaper_ran
    if not container_config.get("docker_orphan_reaper", True):
        return
    if _docker_orphan_reaper_ran:  # double-checked locking
        return
    with _docker_orphan_reaper_lock:
        if _docker_orphan_reaper_ran:
            return
        _docker_orphan_reaper_ran = True

    # 2 × lifetime gives sibling processes a grace window; floor at 60s so
    # TERMINAL_LIFETIME_SECONDS=0 can't instant-reap a sibling's own setup.
    # container_config only carries container_* keys, so read the env var.
    try:
        lifetime = int(_tenv("TERMINAL_LIFETIME_SECONDS", "300"))
    except (TypeError, ValueError):
        lifetime = 300
    max_age = max(60, lifetime) * 2

    try:
        from tools.environments.docker import reap_orphan_containers, _container_identity
    except ImportError:
        return
    # Never fail the env-creation path because of a janitor problem.
    with _quiet("Docker orphan reaper raised"):
        profile = _container_identity(container_config.get("docker_shared_container_key", ""))
        removed = reap_orphan_containers(max_age_seconds=max_age, profile_filter=profile)
        if removed:
            logger.info(
                "Docker orphan reaper removed %d stale container(s) for profile %s",
                removed, profile,
            )


# Per-task environment overrides (never exposed to the model). RL/benchmark
# envs and ACP register a custom image / cwd for a task_id BEFORE the agent
# loop; sandbox creation consults this first, then the TERMINAL_* env vars.
_task_env_overrides: dict[str, dict[str, Any]] = {}

# Per-session cwd records: the durable source of truth for "which directory
# is THIS session in". Keyed by the raw session/task key, NOT the collapsed
# container id — the env is shared across sessions, so cwd state stored on
# it is a global mutable timeshared between sessions (the wrong-worktree bug
# class). Written after every completed command and on cwd-override
# registration; readers resolve against it before any env-side cwd.
_session_cwd: dict[str, str] = {}
_session_cwd_lock = threading.Lock()

# Subagent → parent container aliasing. delegate_task children have their own
# task_id but must share the PARENT's container; under per-session isolation
# the collapse-to-"default" shortcut no longer provides that, so the spawn
# site registers an explicit alias.
_container_aliases: dict[str, str] = {}
_container_alias_lock = threading.Lock()


def record_session_cwd(session_key: Optional[str], cwd: Optional[str]) -> None:
    """Record *cwd* as *session_key*'s working directory (after a completed
    command, or on workspace-override registration). None/empty keys collapse
    to ``"default"``; non-string / empty cwds are ignored. Keys are routed-profile
    qualified (:func:`_qualify_task_key`); the read/clear side qualifies identically."""
    if not isinstance(cwd, str) or not cwd.strip():
        return
    key = _qualify_task_key(str(session_key or "default"))
    with _session_cwd_lock:
        if _session_cwd.get(key) != cwd:
            _session_cwd[key] = cwd


def get_session_cwd(session_key: Optional[str]) -> Optional[str]:
    """Recorded cwd for *session_key*, or None. No fallback chain on purpose:
    callers decide what an absent record means. None/empty keys read ``"default"``."""
    with _session_cwd_lock:
        return _session_cwd.get(_qualify_task_key(str(session_key or "default")))


def clear_session_cwd(session_key: str) -> None:
    """Drop a session's cwd record (session teardown)."""
    with _session_cwd_lock:
        _session_cwd.pop(_qualify_task_key(session_key), None)


def _sanitize_cwd_for_live_env(env: Any, new_cwd: str) -> Optional[str]:
    """Cwd to write into a LIVE cached env, or None to leave it untouched.

    On container backends a raw host path (a desktop/TUI session registering its
    workspace, e.g. ``C:\\Users\\me`` or ``/Users/me/workspace``) cannot be the
    in-sandbox workdir: every file-tools ``_exec`` wrapper does
    ``builtin cd -- <env.cwd> || exit 126``, so a host cwd poisons all later
    file operations with an unrelated ``cd:`` error. Prefix-shaped host paths
    are already rejected on the creation paths. This write classifies the
    directory mounted at ``/workspace`` as unusable before that prefix
    heuristic, then remaps the match (or a child of it) to its container mount
    instead of storing the host path. SSH maps the Hermes subprocess home onto
    the peer's home, as environment creation already does. Other backends apply
    the override verbatim (ACP project-root switching must keep working).
    """
    env_type = getattr(env, "env_type", None)
    new_cwd = coerce_ssh_remote_cwd(new_cwd, env_type)
    if not env_type or not _is_container_backend(env_type):
        return new_cwd
    host_mount = getattr(env, "host_cwd", None)
    mounted = host_mount if isinstance(host_mount, str) and host_mount else None
    # Mount equality before the prefix heuristic. /mnt and /srv are absolute,
    # so the heuristic alone would write the host path through as env.cwd and
    # every later file-tools exec would `cd` to it (exit 126).
    if not _is_unusable_container_cwd(new_cwd, mounted_host=mounted):
        return new_cwd
    if mounted:
        # The bind may sit at a fallback mount when /workspace is claimed.
        container_mount = getattr(env, "host_cwd_mount", None) or "/workspace"
        if _is_mounted_host_cwd(new_cwd, mounted):
            return container_mount
        translated = translate_mounted_host_path(new_cwd, mounted, container_mount)
        if translated:
            return translated
    return None


def register_task_env_overrides(task_id: str, overrides: dict[str, Any]):
    """Register per-task sandbox overrides (``docker_image``/``modal_image``/
    ``daytona_image`` or a plugin backend's ``<backend>_image``, ``env_type``,
    ``cwd``) before the agent loop runs.

    A ``cwd`` override takes effect immediately: it becomes the session's
    recorded cwd (until a ``cd`` changes it) and any live env's cwd is updated
    too, so env-side seeding stays consistent (ACP switching project root
    mid-session via ``session/load``). The session record keeps the RAW path
    (host workspaces are tracked there on purpose); only the live-env write is
    sanitized, since a host cwd can never be a container workdir.

    Keyed like ``_session_cwd`` — routed-profile qualified (:func:`_qualify_task_key`) — so two
    profiles registering the same task id never read each other's image/cwd (#123989).
    """
    _task_env_overrides[_qualify_task_key(task_id)] = overrides

    new_cwd = overrides.get("cwd")
    if isinstance(new_cwd, str) and new_cwd.strip():
        record_session_cwd(task_id, new_cwd)
        # Live env may be cached under the raw task_id (per-session surfaces)
        # or the collapsed container id (isolation-keyed rollouts); try both so
        # a CWD-only override (which collapses to "default") still finds it.
        container_id = _resolve_container_task_id(task_id)
        with _env_lock:
            env = _active_environments.get(task_id) or _active_environments.get(container_id)
        if env is not None and getattr(env, "cwd", None) is not None:
            sanitized = _sanitize_cwd_for_live_env(env, new_cwd)
            if sanitized is not None:
                env.cwd = sanitized


def clear_task_env_overrides(task_id: str):
    """Drop a task's overrides, cwd record and container alias (rollout cleanup)."""
    _task_env_overrides.pop(_qualify_task_key(task_id), None)
    clear_session_cwd(task_id)
    with _container_alias_lock:
        _container_aliases.pop(task_id, None)


def register_container_alias(child_task_id: str, parent_task_id: Optional[str]) -> None:
    """Make *child_task_id* resolve to *parent_task_id*'s container (called at
    delegate_task spawn). A missing parent id aliases to ``"default"``."""
    if not child_task_id:
        return
    with _container_alias_lock:
        _container_aliases[child_task_id] = str(parent_task_id or "default")


def _resolve_container_alias(task_id: str) -> str:
    """Follow the child→parent alias chain (cycle-safe) for *task_id*."""
    seen = set()
    key = task_id
    with _container_alias_lock:
        while key in _container_aliases and key not in seen:
            seen.add(key)
            key = _container_aliases[key]
    return key


def _is_isolation_override_key(key: str) -> bool:
    """An image (``docker_image``, a plugin backend's ``<backend>_image``) or ``env_type`` override."""
    return key == "env_type" or key.endswith("_image")


def _has_isolation_overrides(task_id: Optional[str]) -> bool:
    """True when *task_id* registered image/env_type overrides — the single
    "isolated RL/benchmark rollout" predicate shared by key resolution and
    container creation so the two can't drift."""
    overrides = _task_env_overrides.get(_qualify_task_key(task_id)) if task_id else None
    return bool(overrides and any(_is_isolation_override_key(str(key)) for key in overrides))


@dataclass(frozen=True)
class _SessionScope:
    """Backend identity + scoping predicates for one call, read once.

    ``env_type`` is the scope-aware TERMINAL_ENV; ``persistent`` is
    ``TERMINAL_CONTAINER_PERSISTENT``. Derived predicates:

    * ``session_isolated`` — non-persistent sandboxes get per-session identities:
      ``container_persistent: false`` means state must not survive or be shared
      across sessions, so one shared sandbox contradicts it. Docker, plus plugin
      backends declaring ``session_isolated_when_nonpersistent`` (sandboxes resumed
      by name, where a shared deterministic name would let two ephemeral runs
      attach one VM and delete it under each other).
    * ``docker_session_isolated`` — docker-only view: the workspace mount and
      session-scoped teardown paths must not fire for other backends.
    * ``docker_profile_scoped`` — docker + ``container_persistent: true``: ONE
      long-lived container per profile shared by every session (CLI, gateway,
      WebUI). The session-key fallback in :func:`_resolve_container_task_id` stops
      cross-profile SSH reuse; ungated it fragmented persistent Docker into one
      container per gateway session, so this restores profile scoping for exactly
      this backend/mode.
    """
    env_type: str
    persistent: bool

    @property
    def session_isolated(self) -> bool:
        if self.env_type != "docker" and not _plugin_env_flag(
            self.env_type, "session_isolated_when_nonpersistent"
        ):
            return False
        return not self.persistent

    @property
    def docker_session_isolated(self) -> bool:
        return self.env_type == "docker" and self.session_isolated

    @property
    def docker_profile_scoped(self) -> bool:
        return self.env_type == "docker" and self.persistent


def _session_scope() -> _SessionScope:
    """Bridge config → env once, then snapshot the backend scope for this call."""
    _ensure_terminal_env_bridged()
    return _SessionScope(
        env_type=_tenv("TERMINAL_ENV", "local"),
        persistent=_tenv_bool("TERMINAL_CONTAINER_PERSISTENT", "true"),
    )


def _docker_session_isolation_enabled() -> bool:
    """See :attr:`_SessionScope.docker_session_isolated` (used by the docker builder)."""
    return _session_scope().docker_session_isolated


def _routed_home_task_key(profile_scoped: bool) -> Optional[str]:
    """Key for a session-less task serving a routed (non-launch) profile home, else None.

    A multiplexed host runs every profile's cron jobs without a session key; collapsing them all onto
    ``"default"`` made profile B's cron tool calls reuse the environment the launch profile's job
    created (its ``.env`` residue, its bridged ``TERMINAL_*``, its shell), so B ran with A's settings.
    Persistent Docker keys the profile name exactly like B's session-bound work, so B keeps ONE
    container instead of a second one per home path.
    """
    from hermes_constants import get_hermes_home_override, profile_name_for_home
    from tools.environments.local import _is_routed_home

    override = get_hermes_home_override()
    if not override or not _is_routed_home(override):
        return None
    profile = profile_name_for_home(override) if profile_scoped else None
    if profile:
        return "default" if profile == "default" else f"profile:{profile}"
    try:
        return f"home:{os.path.realpath(override)}"
    except OSError:
        return f"home:{override}"


def _qualify_task_key(key: str) -> str:
    """Prefix a session-derived key with the routed profile/home, or return it unchanged.

    A multiplexed host serves every profile in one process, and session ids are not
    profile-unique (a header-less API client's ``api-<digest>`` fingerprint, a DM chat id
    shared by two bots), so raw session keys in ``_active_environments`` / ``_session_cwd``
    would hand profile B the sandbox and cwd profile A created (#123989). Persistent Docker
    already keys the profile (branches 3/4); this covers the per-session keys. CLI and
    single-profile gateways (no routed home) keep the historical raw key.
    """
    scope = _routed_home_task_key(profile_scoped=True)
    return f"{scope}:{key}" if scope else key


def _resolve_container_task_id(task_id: Optional[str]) -> str:
    """Map a tool-call ``task_id`` to the ``_active_environments`` key. Order matters —
    earlier branches are authoritative where they apply:

    1. Image/``env_type`` overrides (RL/benchmark rollouts) key their own sandbox;
       CWD-only overrides (ACP workspace tracking) are NOT isolation signals.
    2. Per-session isolation (docker + ``container_persistent: false``): each
       session's task_id is its own key (a fresh chat gets a fresh sandbox with only
       ITS mounts); delegate_task children follow the alias registry to the parent.
       Routed profiles qualify the key (:func:`_qualify_task_key`, #123989).
    3. Session key present (WebUI per-session, gateway per-message): persistent
       Docker is PROFILE-scoped — ``shared:<key>`` opt-in, else ``profile:<name>``,
       with the default profile staying literally ``"default"`` so CLI and
       default-profile gateway sessions share ONE container; other backends key
       ``session:<key>`` so switching profiles can't reuse another profile's
       SSHEnvironment on the wrong host.
    4. No session key (CLI, cron): ``shared:<key>`` when opted in (else a CLI run of a
       keyed profile would split from its gateway sessions); a routed multiplexed profile
       keys its own home (``profile:<name>`` under persistent Docker, matching branch 3);
       else ``"default"``, which subagent ids collapse onto to share the parent's container.
    """
    if task_id and _has_isolation_overrides(task_id):
        return _qualify_task_key(task_id)
    scope = _session_scope()
    if task_id and scope.session_isolated:
        return _qualify_task_key(_resolve_container_alias(task_id))
    # Per-session isolation: when a session key is present (the WebUI streaming layer sets it per-session,
    # the gateway per-message via contextvars), scope the container to it so switching profiles can't reuse
    # a previous profile's SSHEnvironment and silently run commands on the wrong remote host. Subagents
    # inherit the same session key, so they still collapse onto the parent's container (the #16177
    # shared-container intent). CLI mode has no session key and falls through to "default", behaviour
    # unchanged. See commit e00f940a9. This runs *after* the isolation-override and
    # docker/container_persistent branches above: those paths already key containers per task_id, so they
    # stay authoritative where they apply and this only covers the cases that would otherwise collapse to
    # the shared "default" key (notably SSH).
    session_key = _current_session_key()
    shared = _tenv("TERMINAL_DOCKER_SHARED_CONTAINER_KEY", "").strip() if scope.docker_profile_scoped else ""
    if shared:
        # Explicit opt-in: trusted profiles configuring the same terminal.docker_shared_container_key share
        # ONE container/cache slot (and sandbox dir) regardless of profile name (#84671).
        return f"shared:{shared}"
    if not session_key:
        return _routed_home_task_key(scope.docker_profile_scoped) or "default"
    if not scope.docker_profile_scoped:
        return _qualify_task_key(f"session:{session_key}")
    profile = _container_profile()
    base = "default" if profile == "default" else f"profile:{profile}"
    return _workspace_container_key(base, task_id)


def workspace_key_for(cwd: str) -> str:
    """The ``:ws:`` digest of a host workspace: computed ONCE when the session binds it
    (``gateway.session_policy.bind_turn_workspace``), so container keying never depends on live
    filesystem state at lookup time."""
    import hashlib
    resolved = os.path.realpath(os.path.expanduser(cwd))
    return hashlib.sha256(resolved.encode("utf-8")).hexdigest()[:16]


def _workspace_container_key(base: str, task_id: Optional[str]) -> str:
    """Persistent Docker key for a session that registered a workspace: one container per
    (profile, workspace), not per profile (#140665).

    A gateway-owned CLI/Desktop session registers its client cwd as a ``cwd_source: "session"``
    override (``gateway.session_policy.bind_turn_workspace``). Keyed by profile alone, every
    concurrent session shared the container bound to whichever workspace created it first; sessions
    in the same folder still share one. Sessions with no registered workspace keep the profile key.

    Idempotent: an already-resolved ``<base>:ws:<digest>`` key maps to itself (``ensure_task_env``
    hands the resolved key to ``get_active_env``, which resolves again). A ``delegate_task`` child has
    only a container alias to its parent (``register_container_alias``), so the alias chain is
    followed: the child runs in its parent's workspace container, never the profile's.
    Reads the raw table (not :func:`resolve_task_overrides`, which calls back into key resolution).
    """
    if not task_id:
        return base
    if task_id.startswith(f"{base}:ws:"):
        return task_id
    for key in dict.fromkeys((task_id, _resolve_container_alias(task_id))):
        overrides = _task_env_overrides.get(_qualify_task_key(key))
        if not overrides or overrides.get("cwd_source") != "session":
            continue
        digest = overrides.get("workspace_key")
        cwd = overrides.get("cwd")
        if not digest and isinstance(cwd, str) and cwd.strip() and "\0" not in cwd:
            digest = workspace_key_for(cwd)
        if digest:
            return f"{base}:ws:{digest}"
    return base


def resolve_task_overrides(task_id: Optional[str]) -> dict[str, Any]:
    """Return the env overrides for *task_id*, raw key first then collapsed.

    ``register_task_env_overrides`` writes under the *raw* task/session id, but
    a CWD-only override collapses (:func:`_resolve_container_task_id`) to the
    shared ``"default"`` container. Callers must therefore read the raw id
    FIRST and only fall back to the collapsed container id, or the originating
    session's override is silently dropped. Single source of that lookup so
    the terminal and file layers can't drift apart.
    """
    raw = task_id or "default"
    return (
        _task_env_overrides.get(_qualify_task_key(raw))
        or _task_env_overrides.get(_resolve_container_task_id(raw))
        or {}
    )


# Backends that take an image, keyed to the override/config key carrying it.
_IMAGE_KEY_BY_BACKEND = {
    "docker": "docker_image",
    "modal": "modal_image",
    "daytona": "daytona_image",
}


def _select_image(env_type: str, overrides: dict[str, Any], config: dict[str, Any]) -> str:
    """Image for *env_type*: per-task override first, then config; "" for imageless backends. A plugin
    backend gets only its per-task ``<backend>_image`` override (batch runs register one); otherwise it
    reads its own image setting."""
    key = _IMAGE_KEY_BY_BACKEND.get(env_type)
    if key is None:
        return overrides.get(f"{env_type}_image") or ""
    return overrides.get(key) or config[key]


def _lookup_active_env(effective_task_id: str, task_id: Optional[str]):
    """Return the cached env for the collapsed id, else for the raw task_id, else None.

    Caller holds ``_env_lock``. Per-session surfaces (ACP/gateway/dashboard)
    with a CWD-only override collapse to ``"default"`` for container sharing,
    yet an env may already be cached under the originating task_id; honor it
    instead of spawning a duplicate. Refreshes ``_last_activity`` on a hit.
    """
    for key in (effective_task_id, task_id):
        if key and key in _active_environments:
            _last_activity[key] = time.time()
            return _active_environments[key]
    return None


def _resolve_task_host_cwd(config: dict[str, Any], task_id: Optional[str]) -> Optional[str]:
    """Host directory to bind into *task_id*'s container.

    Single owner of the cwd-mount policy for every creation site. Shared-
    container mode: the ``TERMINAL_CWD``-derived ``config["host_cwd"]``.
    Per-session isolation (docker + ``container_persistent: false``): only
    the SESSION's own registered workspace may mount — the process env var is
    a launch artifact that outlives the session that set it, so deriving a
    fresh session's mount from it would leak the previous session's directory.
    Overrides tagged ``cwd_source: "process"`` are refused for the same reason;
    ``cwd_source: "session"`` or untagged (ACP/RL) overrides mount.
    A Windows drive path is not a mount source while the cwd-to-/workspace flag
    is off. A raw host override must stay out of ``docker run -w`` and fall
    back to the sanitized config cwd. The Windows bind, including when
    ``/workspace`` is already claimed, is the volume mount, not this override.
    """
    if config.get("env_type") != "docker" or not config.get("docker_mount_cwd_to_workspace"):
        return None
    # Top-level CLI parent ("default") is a single-session process — legacy behavior.
    if not _docker_session_isolation_enabled() or _resolve_container_task_id(task_id) == "default":
        return config.get("host_cwd")
    overrides = resolve_task_overrides(task_id)
    candidate = overrides.get("cwd")
    if overrides.get("cwd_source") == "process" or not isinstance(candidate, str) or not candidate.strip():
        return None
    candidate = os.path.abspath(os.path.expanduser(candidate))
    # Must exist on the host and not already be an in-container path.
    if not os.path.isdir(candidate) or candidate.startswith(("/workspace", "/root")):
        return None
    return candidate


# One-shot guard for the config-fallback bridge: after the first attempt
# either TERMINAL_ENV is set or the import failed, so retrying is wasted work.
_terminal_config_bridge_attempted = False


def _ensure_terminal_env_bridged() -> None:
    """Backfill TERMINAL_* env vars from config.yaml when no launcher did.

    CLI, gateway and TUI/dashboard PTY launches bridge ``terminal.*`` into env vars
    at startup; processes that skip those paths (``hermes serve``, Desktop
    in-process agents, desktop cron ticker, ACP) would otherwise fall back to the
    local backend even when config selects docker — running on the host the user
    meant to sandbox. Explicit keys in the ``terminal`` section override matching
    env values (possibly stale from ``hermes setup``); env values for omitted keys
    are preserved. Without a terminal section an existing TERMINAL_ENV is kept and
    defaults are backfilled only when none is set. A per-turn terminal scope
    suppresses the bridge entirely: writing scope values into the process-global
    env would re-create the first-writer-wins cross-profile leak the scope fixes.

    Ambient ``os.environ`` is the *launch* profile's authority only. Under a
    context-local ``HERMES_HOME`` override (multiplexed dashboard / gateway
    secondary profile), this bridge is a no-op — otherwise the first unscoped
    call under that override would latch the secondary profile's ``terminal.*``
    into process-global env and poison later unscoped launch-profile turns
    (#107422 residual of #68559). Routed profiles must bind a terminal scope
    instead (same rule as ``env_loader._reapply_terminal_config_bridge``).

    terminal_tool reads ALL terminal settings from os.environ (TERMINAL_*). See #61115, #65696.
    """
    from tools.terminal_scope import get_terminal_scope

    if get_terminal_scope() is not None:
        return
    # Never write a secondary profile's terminal.* into process-global env.
    from hermes_constants import get_hermes_home_override

    if get_hermes_home_override() is not None:
        return
    global _terminal_config_bridge_attempted
    if _terminal_config_bridge_attempted:
        return
    _terminal_config_bridge_attempted = True
    # Never let a config problem take the terminal tool down.
    with _quiet("terminal config → env fallback bridge failed"):
        from hermes_cli.config import apply_terminal_config_to_env, read_raw_config

        raw_config = read_raw_config()
        if isinstance(raw_config.get("terminal"), dict):
            apply_terminal_config_to_env(env=None, override=True)
        elif "TERMINAL_ENV" not in os.environ:
            apply_terminal_config_to_env(env=None, override=False)


# Default cwd per backend; plugins may declare ``default_cwd``; anything else (container backends) is "/root".
_DEFAULT_CWD_BY_BACKEND = {"ssh": "~", "vercel_sandbox": _VERCEL_SANDBOX_DEFAULT_CWD}


def _backend_default_cwd(env_type: str) -> str:
    if env_type == "local":
        return _safe_getcwd()
    if env_type in _DEFAULT_CWD_BY_BACKEND:
        return _DEFAULT_CWD_BY_BACKEND[env_type]
    declared = _plugin_env_flag(env_type, "default_cwd", None)
    # "~" is each sandbox's own home, expanded inside it; otherwise only an absolute in-sandbox path
    # (a relative value would be anchored on the host).
    if isinstance(declared, str) and (declared == "~" or declared.startswith("/")):
        return declared
    return "/root"


def _resolve_config_cwd(env_type: str, mount_docker_cwd: bool) -> tuple:
    """``(cwd, host_cwd)`` from TERMINAL_CWD for *env_type*.

    Container backends are sanity-checked: with Docker cwd passthrough the host
    path is remapped to /workspace and tracked as host_cwd; otherwise host paths
    are discarded in favor of the backend default.
    """
    default_cwd = _backend_default_cwd(env_type)
    cwd = _tenv("TERMINAL_CWD", default_cwd)
    from hermes_cli.config import _is_ssh_remote_tilde_cwd
    # The backend's own default is an in-sandbox path: its ``~`` is the sandbox home, not ours.
    if cwd and cwd != default_cwd and not _is_ssh_remote_tilde_cwd(env_type, cwd):
        cwd = os.path.expanduser(cwd)
    host_cwd = None
    if env_type == "docker" and mount_docker_cwd:
        candidate = os.path.abspath(os.path.expanduser(_tenv("TERMINAL_CWD") or _safe_getcwd()))
        if (
            _is_host_cwd(candidate)
            or (os.path.isabs(candidate) and os.path.isdir(candidate) and not candidate.startswith(("/workspace", "/root")))
        ):
            host_cwd = candidate
            cwd = "/workspace"
    elif env_type == "docker" and _is_windows_drive_path(cwd):
        # A Windows workspace cannot exist inside the Linux container. Bind it
        # even when docker_mount_cwd_to_workspace is off; the env retargets cwd
        # to the mount (which may not be /workspace).
        candidate = os.path.expanduser(cwd)
        if os.name == "nt":
            candidate = os.path.abspath(candidate)
        if os.path.isdir(candidate):
            host_cwd = candidate
            cwd = candidate
        else:
            logger.info("Ignoring TERMINAL_CWD=%r for %s backend "
                        "(host/relative path won't work in sandbox). Using %r instead.",
                        cwd, env_type, default_cwd)
            cwd = default_cwd
    elif _is_container_backend(env_type) and cwd and _is_unusable_container_cwd(cwd) and cwd != default_cwd:
        logger.info("Ignoring TERMINAL_CWD=%r for %s backend "
                    "(host/relative path won't work in sandbox). Using %r instead.",
                    cwd, env_type, default_cwd)
        cwd = default_cwd
    return cwd, host_cwd


def _get_env_config() -> dict[str, Any]:
    """Resolve the terminal configuration dict from TERMINAL_* env vars."""
    from hermes_cli.config_defaults import DEFAULT_SANDBOX_IMAGE as default_image
    _ensure_terminal_env_bridged()
    env_type = _tenv("TERMINAL_ENV", "local")
    mount_docker_cwd = _tenv_bool("TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false")

    # Container/docker-only payloads are parsed only when such a backend is
    # selected: a stale or invalid Docker value bridged from config.yaml must
    # not make local terminal/execute_code unusable.
    if _is_container_backend(env_type):
        container_cpu = _parse_env_var("TERMINAL_CONTAINER_CPU", "1", float, "number")
        container_memory = _parse_env_var("TERMINAL_CONTAINER_MEMORY", "5120")
        container_disk = _parse_env_var("TERMINAL_CONTAINER_DISK", "51200")
    else:
        container_cpu, container_memory, container_disk = 1.0, 5120, 51200

    if env_type == "docker":
        docker_forward_env = _parse_env_var("TERMINAL_DOCKER_FORWARD_ENV", "[]", json.loads, "valid JSON")
        docker_volumes = _parse_env_var("TERMINAL_DOCKER_VOLUMES", "[]", json.loads, "valid JSON")
        docker_env = _parse_env_var("TERMINAL_DOCKER_ENV", "{}", json.loads, "valid JSON")
        docker_extra_args = _parse_env_var("TERMINAL_DOCKER_EXTRA_ARGS", "[]", json.loads, "valid JSON")
        docker_shm_size = _tenv("TERMINAL_DOCKER_SHM_SIZE", "1g")
    else:
        docker_forward_env, docker_volumes, docker_env, docker_extra_args, docker_shm_size = [], [], {}, [], "1g"

    cwd, host_cwd = _resolve_config_cwd(env_type, mount_docker_cwd)

    return {
        "env_type": env_type,
        "modal_mode": coerce_modal_mode(_tenv("TERMINAL_MODAL_MODE", "auto")),
        "docker_image": _tenv("TERMINAL_DOCKER_IMAGE", default_image),
        "docker_image_pinned": _tenv("TERMINAL_DOCKER_IMAGE_PINNED", "0") == "1",
        "docker_forward_env": docker_forward_env,
        "modal_image": _tenv("TERMINAL_MODAL_IMAGE", default_image),
        "daytona_image": _tenv("TERMINAL_DAYTONA_IMAGE", default_image),
        "vercel_runtime": _tenv("TERMINAL_VERCEL_RUNTIME", "").strip(),
        "vercel_image": _tenv("TERMINAL_VERCEL_IMAGE", "").strip(),
        "cwd": cwd,
        "host_cwd": host_cwd,
        "docker_mount_cwd_to_workspace": mount_docker_cwd,
        "timeout": _parse_env_var("TERMINAL_TIMEOUT", "180"),
        "lifetime_seconds": _parse_env_var("TERMINAL_LIFETIME_SECONDS", "300"),
        # SSH-specific config
        "ssh_host": _tenv("TERMINAL_SSH_HOST", ""),
        "ssh_user": _tenv("TERMINAL_SSH_USER", ""),
        "ssh_port": _parse_env_var("TERMINAL_SSH_PORT", "22"),
        "ssh_key": _tenv("TERMINAL_SSH_KEY", ""),
        # Exported env carries across commands (cwd always does). SSH follows
        # terminal.persistent_shell; local has its own knob. Both default on.
        "ssh_persistent": _tenv_bool(
            "TERMINAL_SSH_PERSISTENT", _tenv("TERMINAL_PERSISTENT_SHELL", "true"),
        ),
        "local_persistent": _tenv_bool("TERMINAL_LOCAL_PERSISTENT", "true"),
        # Container resources (MB); ignored for local/ssh.
        "container_cpu": container_cpu,
        "container_memory": container_memory,
        "container_disk": container_disk,
        "container_persistent": _tenv_bool("TERMINAL_CONTAINER_PERSISTENT", "true"),
        "docker_volumes": docker_volumes,
        "docker_env": docker_env,
        "docker_run_as_host_user": _tenv_bool("TERMINAL_DOCKER_RUN_AS_HOST_USER", "false"),
        "docker_snap_compat": _tenv_bool("TERMINAL_DOCKER_SNAP_COMPAT", "false"),
        "docker_network": _tenv_bool("TERMINAL_DOCKER_NETWORK", "true"),
        "docker_extra_args": docker_extra_args,
        "docker_shm_size": docker_shm_size,
        # Cross-process reuse: attach to a labeled container at startup
        # instead of starting fresh; false = per-process isolation.
        "docker_persist_across_processes": _tenv_bool("TERMINAL_DOCKER_PERSIST_ACROSS_PROCESSES", "true"),
        "docker_shared_container_key": _tenv("TERMINAL_DOCKER_SHARED_CONTAINER_KEY", "").strip(),
        "docker_orphan_reaper": _tenv_bool("TERMINAL_DOCKER_ORPHAN_REAPER", "true"),
    }


def _cleanup_thread_worker():
    """Background thread worker that periodically cleans up inactive environments."""
    while _cleanup_running:
        with _quiet("Error in cleanup thread", level=logging.WARNING):
            _cleanup_inactive_envs(_get_env_config()["lifetime_seconds"])
        for _ in range(60):
            if not _cleanup_running:
                break
            time.sleep(1)


def _start_cleanup_thread():
    """Start the background cleanup thread if not already running."""
    global _cleanup_thread, _cleanup_running

    with _env_lock:
        if _cleanup_thread is None or not _cleanup_thread.is_alive():
            _cleanup_running = True
            _cleanup_thread = threading.Thread(target=_cleanup_thread_worker, daemon=True)
            _cleanup_thread.start()


def _stop_cleanup_thread():
    """Stop the background cleanup thread."""
    global _cleanup_running
    _cleanup_running = False
    if _cleanup_thread is not None:
        try:
            _cleanup_thread.join(timeout=5)
        except (SystemExit, KeyboardInterrupt):
            pass


def _atexit_cleanup():
    """Stop the cleanup thread and shut down all remaining sandboxes on exit."""
    _stop_cleanup_thread()
    if _active_environments:
        logger.info("Shutting down %d remaining sandbox(es)...", len(_active_environments))
        # Snapshot BEFORE cleanup_all_environments empties the dict, then
        # block briefly so docker stop/rm completes before the interpreter
        # exits — otherwise daemon cleanup threads die mid-`docker stop` and
        # Exited containers pile up on the host.
        envs_to_wait = list(_active_environments.values())
        cleanup_all_environments()
        for env in envs_to_wait:
            wait_fn = getattr(env, "wait_for_cleanup", None)
            if wait_fn is not None:
                with _quiet("wait_for_cleanup raised on exit"):  # never block shutdown on a bad backend
                    wait_fn(timeout=15.0)
    # Workers of envs the idle reaper already detached are not in the registry (#86317).
    if "tools.environments.docker" in sys.modules:
        with _quiet("teardown drain raised on exit"):
            sys.modules["tools.environments.docker"].DockerEnvironment.wait_for_all_teardowns(timeout=15.0)

atexit.register(_atexit_cleanup)


def _command_requires_pipe_stdin(command: str) -> bool:
    """True when PTY mode would break a stdin-driven command: `gh auth login
    --with-token` waits for EOF on piped stdin, and under a PTY
    `process.submit()` only sends a newline, so it hangs forever."""
    normalized = " ".join(command.lower().split())
    return normalized.startswith("gh auth login") and "--with-token" in normalized


from tools.terminal_tool_guards import (
    _foreground_background_guidance, _safe_command_preview, _validate_workdir,
    gateway_lifecycle_block, self_repo_block, trailing_amp_background_command,
)
from tools.terminal_tool_background import _YIELDED_NOTE, spawn_background_process, yield_to_background_handler
from tools.terminal_tool_result import finalize_foreground_result


def _resolve_notification_flag_conflict(*, notify_on_complete: bool, watch_patterns, background: bool) -> tuple:
    """Resolve notify_on_complete + watch_patterns both set: drop watch_patterns
    (combined they produce duplicate async notifications — one per match plus
    one on exit — that can spam the user long after the process ends).
    Returns ``(watch_patterns_to_use, conflict_note)``; note is "" without conflict."""
    if background and notify_on_complete and watch_patterns:
        return None, (
            "watch_patterns ignored because notify_on_complete=True; "
            "these two flags produce duplicate notifications when combined"
        )
    return watch_patterns, ""


def _rewrite_via_env_mount(path: str, env) -> str | None:
    """Container path for *path* when *env* bind-mounted that host directory."""
    if env is None or not path:
        return None
    host = getattr(env, "host_cwd", None)
    mount = getattr(env, "host_cwd_mount", None)
    if not isinstance(host, str) or not host or not mount:
        return None
    return translate_mounted_host_path(path, host, mount)


def _mount_envs(env):
    if env is not None:
        return [env]
    return list(_active_environments.values())


def _container_visible_cwd(path: str, env_type: str | None, env=None) -> str:
    if not path or not _is_container_backend(env_type or ""):
        return path
    for item in _mount_envs(env):
        translated = _rewrite_via_env_mount(path, item)
        if translated:
            return translated
    return path


def _container_visible_default(default_cwd: str, env_type: str | None, env=None) -> str:
    """Point a planner ``/workspace`` assumption at the mount that actually holds the host cwd."""
    if not _is_container_backend(env_type or ""):
        return default_cwd
    for item in _mount_envs(env):
        mount = getattr(item, "host_cwd_mount", None)
        host = getattr(item, "host_cwd", None)
        if not mount or not host or mount == default_cwd:
            continue
        if default_cwd == "/workspace" or _is_unusable_container_cwd(default_cwd):
            return mount
    return default_cwd


def _resolve_command_cwd(
    *,
    workdir: Optional[str],
    default_cwd: str,
    session_key: Optional[str] = None,
    env_type: Optional[str] = None,
    mounted_host: Optional[str] = None,
    env=None,
) -> str:
    """cwd for a command: explicit ``workdir`` > the session's own cwd record >
    ``default_cwd``.

    The record is written after every completed command of THIS session, so
    it is the session's ``cd`` state with no shared-env ambiguity. On
    container backends a recorded HOST path (a desktop/TUI surface registering
    its workspace) is unusable in the sandbox — ``cd <host path>`` fails with
    exit 126 — so it is discarded in favor of ``default_cwd``.

    A recorded cwd that IS the host directory mounted at ``/workspace`` is
    classified unusable before the ``/Users`` / ``/home`` / drive-letter
    heuristic (``/mnt/...``, ``/srv/...`` are absolute and miss that heuristic)
    and remapped to ``/workspace`` so the session wrapper does not ``cd`` to it.

    Same guard class as the env-creation sanitizers (#50636, #54447); this is the per-command sibling site.
    """
    if workdir:
        return coerce_ssh_remote_cwd(_container_visible_cwd(workdir, env_type, env), env_type)
    recorded = get_session_cwd(session_key)
    if recorded and _is_container_backend(env_type) and _is_unusable_container_cwd(
        recorded, mounted_host=mounted_host
    ) and not _in_sandbox_home(recorded, env):
        visible = _container_visible_cwd(recorded, env_type, env)
        if visible != recorded:
            return visible
        if _is_mounted_host_cwd(recorded, mounted_host):
            logger.info(
                "Remapping recorded session cwd %r for %s backend "
                "(mounted host directory). Using '/workspace' instead.",
                recorded, env_type,
            )
            return "/workspace"
        logger.info(
            "Ignoring recorded session cwd %r for %s backend "
            "(host/relative path won't work in sandbox). Using %r instead.",
            recorded, env_type, default_cwd,
        )
        return _container_visible_default(default_cwd, env_type, env)
    return coerce_ssh_remote_cwd(recorded or _container_visible_default(default_cwd, env_type, env), env_type)


def _error_json(error: str, *, exit_code: int = -1, status: Optional[str] = None, **extra) -> str:
    """The terminal error envelope: ``output``/``exit_code``/``error`` (+ ``status``, extras)."""
    body: dict[str, Any] = {"output": "", "exit_code": exit_code, "error": error}
    if status is not None:
        body["status"] = status
    body.update(extra)
    return json.dumps(body, ensure_ascii=False)


def _fatal_error_json(e: BaseException) -> str:
    """Log the traceback and return the redacted error+traceback envelope.

    Exception text can embed the failing command line (and any secrets inline
    in it), so both fields are force-redacted before reaching the model.
    """
    import traceback
    tb_str = traceback.format_exc()
    logger.error("terminal_tool exception:\n%s", tb_str)
    return json.dumps({
        "output": "",
        "exit_code": -1,
        "error": _redact_terminal_error_text(f"Failed to execute command: {e}"),
        "traceback": _redact_terminal_error_text(tb_str),
        "status": "error"
    }, ensure_ascii=False)


class _Rejected(Exception):
    """Carries a finished tool-result JSON out of the planning/guard helpers, so
    each early-return site is one ``raise`` instead of an isinstance-checked
    ``str | plan`` union at the caller."""

    def __init__(self, result_json: str):
        super().__init__(result_json)
        self.result_json = result_json


@dataclass
class _ApprovalVerdict:
    """Outcome of the pre-exec guard pass.

    ``note`` is the audit note attached to the result. ``approved_run`` is True
    when the user explicitly approved (or pre-confirmed via ``force``); it drives
    the clean-interrupt-slate clear before ``env.execute`` so an approved command
    can't be SIGINT-killed by a bit that landed during the approval-wait.
    """
    note: Optional[str] = None
    approved_run: bool = False


def _run_approval_guards(command: str, env_type: str, config: dict[str, Any], *, force: bool) -> _ApprovalVerdict:
    """Run the command guards; ``force`` skips them entirely.
    Raises :class:`_Rejected` when the command may not run (denied, or pending
    gateway approval)."""
    if force:
        return _ApprovalVerdict(approved_run=True)
    approval = _check_all_guards(command, env_type, has_host_access=_docker_has_host_access(config))
    if not approval["approved"]:
        if approval.get("status") == "pending_approval":  # gateway ask mode
            raise _Rejected(_error_json(
                "", status="pending_approval",
                approval_pending=True,
                command=approval.get("command", command),
                description=approval.get("description", "command flagged"),
                pattern_key=approval.get("pattern_key", ""),
                smart_denied=approval.get("smart_denied", False),
                allow_permanent=approval.get("allow_permanent", True),
            ))
        desc = approval.get("description", "command flagged")
        fallback_msg = (
            f"Command denied: {desc}. "
            "Use the approval prompt to allow it, or rephrase the command."
        )
        raise _Rejected(_error_json(approval.get("message", fallback_msg), status="blocked",
                                    **({"user_summary": approval["user_summary"]} if approval.get("user_summary") else {})))
    desc = approval.get("description", "flagged as dangerous")
    if approval.get("user_approved"):
        return _ApprovalVerdict(
            note=f"Command required approval ({desc}) and was approved by the user.",
            approved_run=True,
        )
    if approval.get("smart_approved"):
        return _ApprovalVerdict(note=f"Command was flagged ({desc}) and auto-approved by smart approval.")
    return _ApprovalVerdict()


@dataclass
class _ExecPlan:
    """Per-call execution parameters resolved before any environment is touched."""
    config: dict[str, Any]
    env_type: str
    effective_task_id: str
    image: str
    cwd: str
    host_cwd: Optional[str]
    effective_timeout: int
    # Set when a foreground call asked for more than FOREGROUND_MAX_TIMEOUT and was promoted to a
    # tracked background process instead of being refused (the requested seconds, for the note).
    promoted_from_foreground_timeout: Optional[int] = None
    # Set when a foreground `cmd &` (the trailing `&` its only backgrounding) is run as
    # terminal(command="cmd", background=true) instead of being refused: the command to run.
    auto_background_command: Optional[str] = None


_PROMOTED_NOTE = (
    "Requested foreground timeout {requested}s exceeds the {cap}s cap, so this command was started as a "
    "tracked background process with notify_on_complete=true instead of being refused. Do NOT re-run it. "
    "Its completion (exit code + output tail) arrives as a notification; poll with "
    "process(action=\"poll\", session_id=...) if you need it sooner."
)


def _plan_execution(
    command: Any, *, task_id: Optional[str], timeout: Optional[int],
    background: bool, _host_local: bool,
) -> _ExecPlan:
    """Resolve backend, env-cache key, image, cwd and timeout for one call.

    Raises :class:`_Rejected` when the call is rejected up front (non-string
    command, non-positive or over-cap timeout, a foreground command that must
    run in the background).
    """
    if not isinstance(command, str):
        logger.warning("Rejected invalid terminal command value: %s", type(command).__name__)
        raise _Rejected(_error_json(
            f"Invalid command: expected string, got {type(command).__name__}", status="error",
        ))

    config = _get_env_config()
    env_type = "local" if _host_local else config["env_type"]

    # Fail closed under a refusal scope: the routed profile's terminal
    # policy could not be resolved, so running with the launch process's
    # ambient policy is forbidden.
    # See #68559.
    if not _host_local:
        from tools.terminal_scope import enforce_no_refusal

        enforce_no_refusal()

    effective_task_id = _resolve_container_task_id(task_id)
    if _host_local:
        # Control-plane children run beside this interpreter, never inside
        # the configured Docker/SSH backend; keep their env cache separate.
        effective_task_id = f"host-local-{effective_task_id}"

    # Per-task overrides (RL/benchmark envs, ACP workspace cwd) win over
    # the global env-var config; ``resolve_task_overrides`` reads the raw
    # task id first, then the collapsed container id.
    overrides = resolve_task_overrides(task_id)
    image = _select_image(env_type, overrides, config)

    cwd = coerce_ssh_remote_cwd(
        overrides.get("cwd") or get_session_cwd(task_id) or config["cwd"], env_type)
    host_cwd = _resolve_task_host_cwd(config, task_id)
    # config["cwd"] was sanitized for container backends in _get_env_config
    # but an override / session record is raw: a host path would reach
    # `docker run -w` and fail with exit 125. Re-apply the guard to the
    # resolved cwd; when the host path IS this session's mounted workspace,
    # remap to /workspace instead of discarding it. Mount equality is part of
    # the unusable check so /mnt and /srv are not left as the container cwd.
    if _is_container_backend(env_type) and _is_unusable_container_cwd(cwd, mounted_host=host_cwd):
        remapped = "/workspace" if host_cwd else config["cwd"]
        if cwd != remapped:
            logger.info(
                "Remapping host/relative cwd override %r for %s backend "
                "(won't exist in sandbox). Using %r instead.",
                cwd, env_type, remapped,
            )
        cwd = remapped
    # Reject non-positive timeouts before deadline math: ``timeout or
    # default`` would silently turn 0 into the default, and a negative
    # value is truthy and would fire an immediate "-Ns" timeout.
    if timeout is not None and timeout <= 0:
        raise _Rejected(tool_error(f"timeout must be a positive number of seconds (got {timeout})."))
    promoted = None
    if not background:
        # An over-cap foreground timeout is a bounded job the caller wants to wait for (test suites,
        # builds). Refusing it only bought a mechanical retry: 454 refusals in one run, every one
        # re-sent lower/split/background. Promote to a tracked background process instead; the
        # caller is told in the result. The `&`/nohup/server guidance below stays a refusal: those
        # need the command itself rewritten, which the tool cannot do safely.
        # The detachment guidance applies whether or not the call is promoted: a promoted `cmd &`
        # would start a tracked shell that exits at once while its payload runs untracked.
        # Exception: a lone TRAILING `&` is unambiguous (the caller asked for exactly what
        # background=true does), so run it that way instead of spending a model turn on a resend.
        auto_bg = trailing_amp_background_command(command)
        if auto_bg is None:
            guidance = _foreground_background_guidance(command)
            if guidance:
                raise _Rejected(_error_json(guidance, status="error"))
            if timeout and timeout > FOREGROUND_MAX_TIMEOUT:
                promoted = timeout
    else:
        auto_bg = None

    return _ExecPlan(
        config=config, env_type=env_type, effective_task_id=effective_task_id,
        image=image, cwd=cwd, host_cwd=host_cwd, effective_timeout=timeout or config["timeout"],
        promoted_from_foreground_timeout=promoted, auto_background_command=auto_bg,
    )


_PROMOTED_NOTE_POLL_ONLY = (
    "Requested foreground timeout {requested}s exceeds the {cap}s cap, so this command was started as a "
    "tracked background process instead of being refused. Do NOT re-run it. This session cannot receive "
    "completion notifications, so poll it with process(action=\"poll\", session_id=...) until it exits."
)


def _with_promoted_note(result_json: str, requested_timeout: int) -> str:
    """Attach the foreground->background promotion note to a spawn result (unchanged on error). The
    note only promises a notification when the spawn actually kept notify_on_complete (finite sessions
    such as one-shot runners cannot route one back; the spawn already said so and cleared the flag)."""
    try:
        data = json.loads(result_json)
    except (TypeError, ValueError):
        return result_json
    if not isinstance(data, dict) or data.get("error"):
        return result_json
    template = _PROMOTED_NOTE if data.get("notify_on_complete") else _PROMOTED_NOTE_POLL_ONLY
    data["promoted_from_foreground"] = template.format(requested=requested_timeout, cap=FOREGROUND_MAX_TIMEOUT)
    return json.dumps(data, ensure_ascii=False)


def _acquire_env(plan: _ExecPlan, task_id: Optional[str]) -> Any:
    """Cached env for the task, else create it under the per-task creation lock.

    Concurrent calls for the same task_id wait for the first sandbox instead
    of each creating their own; the cache is re-checked under that lock.
    Raises :class:`_Rejected` with the ``"disabled"`` envelope when creation
    raises ImportError.
    """
    _start_cleanup_thread()
    env_type, eff = plan.env_type, plan.effective_task_id

    with _env_lock:
        env: Any = _lookup_active_env(eff, task_id)
    if env is not None:
        return env

    with _creation_locks_lock:
        task_lock = _creation_locks.setdefault(eff, threading.Lock())

    with task_lock:
        with _env_lock:
            env = _lookup_active_env(eff, task_id)
        if env is not None:
            return env

        logger.info("Creating new %s environment for task %s...", env_type, eff[:8])
        try:
            new_env = _create_configured_env(
                plan.config, env_type, image=plan.image, cwd=plan.cwd,
                timeout=plan.effective_timeout, task_id=eff, host_cwd=plan.host_cwd,
            )
        except ImportError as e:
            raise _Rejected(_error_json(
                _redact_terminal_error_text(f"Terminal tool disabled: environment creation failed ({e})"),
                status="disabled",
            ))

        with _env_lock:
            _active_environments[eff] = new_env
            _last_activity[eff] = time.time()
        logger.info("%s environment ready for task %s", env_type, eff[:8])
        return new_env


def _yield_kwargs(command: str, **ctx) -> dict:
    """``env.execute`` kwargs enabling yield-to-background (local backend only)."""
    handler = yield_to_background_handler(command=command, **ctx)
    return {"yield_handler": handler} if handler is not None else {}


def _run_foreground(
    command: str, env: Any, plan: _ExecPlan, *,
    task_id: Optional[str], session_id: Optional[str], session_key: str,
    workdir: Optional[str], approval_note: Optional[str], clear_interrupt: bool,
    metered: bool = True,
) -> str:
    """Execute in the foreground with retry on transient errors, then finalize. ``metered``
    is False for Hermes' own control-plane commands (``_host_local``)."""
    from hermes_cli.observability.shared_metrics_harness import record_terminal_outcome
    max_retries = 3
    env_type, eff, effective_timeout = plan.env_type, plan.effective_task_id, plan.effective_timeout

    # Clean interrupt slate for an approved command, ONCE before the retry
    # loop: drop a stale bit that landed during the approval-wait so it
    # can't SIGINT the just-approved run. Do NOT re-clear inside the loop —
    # a genuine interrupt during the backoff sleep must survive and abort
    # the next attempt (rc 130).
    if clear_interrupt:
        from tools.interrupt import clear_current_thread_interrupt
        clear_current_thread_interrupt()

    for retry_count in range(max_retries + 1):
        try:
            command_cwd = _resolve_command_cwd(
                workdir=workdir, default_cwd=plan.cwd, session_key=session_key, env_type=env_type,
                mounted_host=getattr(env, "host_cwd", None) or plan.host_cwd,
                env=env,
            )
            # bounded_capture: model-facing output keeps a head/tail window
            # while streaming so a verbose command can't OOM the gateway;
            # internal env.execute() consumers stay unbounded.
            result = env.execute(
                command, timeout=effective_timeout, cwd=command_cwd, bounded_capture=True,
                **_yield_kwargs(command, env_type=env_type, cwd=command_cwd, effective_task_id=eff,
                                task_id=task_id, session_key=session_key),
            )
            break
        except Exception as e:
            # A backend exception (e.g. an SSH connect timeout) never reached an exit status, so it
            # is not a terminal outcome; Hermes' own deadline arrives as ``hermes_timed_out``.
            if "timeout" in str(e).lower():
                return _error_json(f"Command timed out after {effective_timeout} seconds", exit_code=124)
            # Retry on transient errors
            if retry_count < max_retries:
                wait_time = 2 ** (retry_count + 1)
                logger.warning("Execution error, retrying in %ds (attempt %d/%d) - Command: %s - Error: %s: %s - Task: %s, Backend: %s",
                               wait_time, retry_count + 1, max_retries, _safe_command_preview(command), type(e).__name__, e, eff, env_type)
                time.sleep(wait_time)
                continue
            logger.error("Execution failed after %d retries - Command: %s - Error: %s: %s - Task: %s, Backend: %s",
                         max_retries, _safe_command_preview(command), type(e).__name__, e, eff, env_type)
            return _error_json(_redact_terminal_error_text(f"Command execution failed: {type(e).__name__}: {e}"))

    if result.get("yielded_session_id"):  # handed to the background: no exit status yet
        return json.dumps({
            "output": result.get("output", ""), "exit_code": None, "error": None,
            "status": "yielded_to_background", "session_id": result["yielded_session_id"],
            "pid": result.get("pid"), "notify_on_complete": True, "note": _YIELDED_NOTE,
        }, ensure_ascii=False)
    if metered:
        record_terminal_outcome(command, env_type, result)
    return finalize_foreground_result(
        command=command, result=result, env=env, env_type=env_type, effective_task_id=eff,
        task_id=task_id, session_id=session_id, session_key=session_key, workdir=workdir,
        command_cwd=command_cwd, approval_note=approval_note,
    )


# Floor for the pre-exec guard's share of the command deadline: a short command timeout
# (1s in tests, a few seconds in practice) must not turn the guard's own cold-start cost
# (module imports, git probes under load) into a refusal; the wedge it bounds lasted an hour.
_PRE_EXEC_GUARD_MIN_TIMEOUT_S = 30


def _pre_exec_block(
    command: str, *, env: Any, env_type: str, cwd: str,
    workdir: Optional[str], session_key: str,
) -> None:
    """Raise :class:`_Rejected` with the blocked-result JSON when the command must not run.

    Order matters: gateway lifecycle first (protects the running gateway),
    then the dangerous-workdir check, then the self-repo guard (local only).
    """
    blocked = gateway_lifecycle_block(
        command=command, env=env, env_type=env_type, cwd=cwd, workdir=workdir, session_key=session_key,
    )
    if blocked:
        raise _Rejected(blocked)
    if workdir:
        workdir_error = _validate_workdir(workdir)
        if workdir_error:
            logger.warning("Blocked dangerous workdir: %s (command: %s)",
                           workdir[:200], _safe_command_preview(command))
            raise _Rejected(_error_json(workdir_error, status="blocked"))
    if env_type == "local":
        blocked = self_repo_block(command=command, cwd=cwd, workdir=workdir, session_key=session_key)
        if blocked:
            raise _Rejected(blocked)


_PTY_DISABLED_REASON = (
    "PTY disabled for this command because it expects piped stdin/EOF "
    "(for example gh auth login --with-token). For local background "
    "processes, call process(action='close') after writing so it receives "
    "EOF."
)


def _degraded_result(e: EnvironmentConnectionError, task_id: Optional[str]) -> str:
    """Infrastructure failure (SSH host down, Docker daemon unreachable), distinct
    from a nonzero exit. ``terminal.degraded_mode``: warn (default) returns a
    structured degraded result with a retry hint; fail preserves the historical
    error+traceback result."""
    if _tenv("TERMINAL_DEGRADED_MODE", "warn").strip().lower() == "fail":
        return _fatal_error_json(e)
    logger.warning("terminal backend degraded: %s", e.reason)
    # Evict the possibly-broken backend so the next call re-creates it.
    with _quiet("degraded-env eviction failed"):
        _evict_environment_for_task(task_id)
    return json.dumps({
        "output": "",
        "exit_code": -1,
        "status": "degraded",
        "reason": e.reason,
        "retry_hint": e.retry_hint,
        "error": f"Terminal backend degraded: {e.reason}",
    }, ensure_ascii=False)


def terminal_tool(
    command: str,
    background: bool = False,
    timeout: Optional[int] = None,
    task_id: Optional[str] = None,
    session_id: Optional[str] = None,
    force: bool = False,
    workdir: Optional[str] = None,
    pty: bool = False,
    notify_on_complete: bool = False,
    watch_patterns: Optional[list[str]] = None,
    _host_local: bool = False,
    _completion_output_chars: int = 0,
    heartbeat: int = 0,
    persist_on_release: bool = False,
) -> str:
    """Execute *command* in the configured terminal environment; returns a JSON string.

    ``force`` (internal, not in the model schema) skips the dangerous-command
    check after the user confirmed. ``workdir`` is per-command and never
    recorded as the session cwd. ``pty`` applies to the local backend only.
    ``notify_on_complete`` and ``watch_patterns`` are mutually exclusive
    background-only flags: on conflict watch_patterns is dropped. watch_patterns
    is hard rate-limited (1 notification / 15s / process) and auto-disabled
    after repeated strikes or a lifetime cap, promoting to notify_on_complete —
    use it only for rare one-shot signals on long-lived processes. ``heartbeat`` (seconds,
    background-only, implies notify_on_complete) emits a "still running + output since last
    time" event every N seconds so the agent stays current on a long job without polling.
    ``persist_on_release`` (background-only) keeps the process alive across agent-lifecycle
    cleanup — session end, context compression, error recovery, max-iteration stop — all of
    which kill the task's background processes; the user can still stop it on purpose via
    process_manage kill (#41225).
    ``_completion_output_chars`` (internal) sizes the completion notification's output for a
    spawner whose output is the payload (a bot DM's reply); 0 keeps the usual tail.
    ``_host_local`` forces the local backend for Hermes-owned control-plane
    children (kept in a separate env cache from the configured backend).
    """
    from hermes_cli.observability.shared_metrics_loop import record_terminal_backend as _metered
    plan = None
    try:
        plan = _plan_execution(
            command, task_id=task_id, timeout=timeout, background=background, _host_local=_host_local,
        )
        auto_bg = getattr(plan, "auto_background_command", None)
        if auto_bg is not None:
            # From here on this IS a background=true call for the stripped command: every guard
            # below (lifecycle, workdir, self-repo, dangerous-command approval) sees what runs.
            command, background = auto_bg, True
        env = _acquire_env(plan, task_id)
        env_type, cwd, effective_task_id = plan.env_type, plan.cwd, plan.effective_task_id

        # Session key for cwd records: the contextvar doesn't cross tool-worker
        # threads, so fall back to the raw task_id (the top-level agent's
        # session_key) as a stable anchor.
        from tools.approval import get_current_session_key

        session_key = get_current_session_key(default="") or (task_id or "")

        # The supervised-gateway identity probe ends in a kernel process query
        # (psutil create_time) that has wedged for the better part of an hour on
        # macOS; ``env.execute`` is already behind ``run_bounded_sync`` but this
        # chain ran ahead of it, so the tool call never returned and the cron
        # slot stayed occupied (#111922). Share the command's own deadline. A
        # guard that never rendered a verdict fails CLOSED: these checks apply
        # unconditionally (``force`` cannot bypass them), so the command is
        # refused with a retryable error instead of running unguarded.
        from agent.deadline import run_bounded_sync
        from tools.interrupt import acting_for_tid

        # The guard chain runs on the deadline worker; keep it answerable to /stop
        # aimed at this tool thread (a remote-backend script read polls is_interrupted()).
        guard_timeout = max(plan.effective_timeout, _PRE_EXEC_GUARD_MIN_TIMEOUT_S)
        _acting_token = acting_for_tid.set(threading.current_thread().ident)
        try:
            bounded_guard = run_bounded_sync(
                lambda: _pre_exec_block(
                    command, env=env, env_type=env_type, cwd=cwd, workdir=workdir, session_key=session_key,
                ),
                guard_timeout,
                label="terminal.pre-exec-guard",
            )
        finally:
            acting_for_tid.reset(_acting_token)
        if bounded_guard.timed_out:
            raise _Rejected(_error_json(
                f"Terminal pre-execution guard did not finish within {guard_timeout}s "
                "(process-identity probe wedged); the command was not run. Retry the call.",
                status="error",
            ))
        # Pre-exec security checks (floors + dangerous command detection);
        # force=True means the user already confirmed.
        verdict = _run_approval_guards(command, env_type, plan.config, force=force)

        pty_disabled = pty and _command_requires_pipe_stdin(command)
        if plan.promoted_from_foreground_timeout is not None:
            # Promotion implies notify_on_complete; watch_patterns is a background-only flag the
            # caller could not have meant for a foreground call, and the two are exclusive anyway.
            background, notify_on_complete, watch_patterns = True, True, None
        if background:
            result = spawn_background_process(
                command=command, env=env, env_type=env_type, effective_task_id=effective_task_id,
                task_id=task_id, session_key=session_key, workdir=workdir, cwd=cwd,
                mounted_host=getattr(env, "host_cwd", None) or plan.host_cwd,
                effective_pty=pty and not pty_disabled, notify_on_complete=notify_on_complete,
                watch_patterns=watch_patterns, approval_note=verdict.note,
                pty_disabled_reason=_PTY_DISABLED_REASON if pty_disabled else None,
                completion_output_chars=_completion_output_chars,
                heartbeat_seconds=heartbeat,
                persist_on_release=persist_on_release,
                auto_backgrounded=auto_bg is not None and plan.promoted_from_foreground_timeout is None,
            )
            if plan.promoted_from_foreground_timeout is not None:
                result = _with_promoted_note(result, plan.promoted_from_foreground_timeout)
            return _metered(None if _host_local else plan, result)
        return _metered(None if _host_local else plan, _run_foreground(
            command, env, plan,
            task_id=task_id, session_id=session_id, session_key=session_key,
            workdir=workdir, approval_note=verdict.note, clear_interrupt=verdict.approved_run,
            metered=not _host_local,
        ))
    except _Rejected as r:
        return r.result_json
    except EnvironmentConnectionError as e:
        return _metered(None if _host_local else plan, _degraded_result(e, task_id), error_class="tool_error")
    except Exception as e:
        return _metered(None if _host_local else plan, _fatal_error_json(e), error_class="exception")


def check_terminal_requirements() -> bool:
    """Check if all requirements for the terminal tool are met. The reason for a failure is kept for
    :func:`terminal_backend_unavailable_reason` (CLI startup notice / doctor)."""
    try:
        config = _get_env_config()
        checker = _REQUIREMENT_CHECKERS.get(config["env_type"], _check_plugin_requirements)
        return checker(config)
    except Exception as e:
        logger.error("Terminal requirements check failed: %s", e, exc_info=True)
        _record_unavailable_reason(f"the requirements check failed: {e}")
        return False


from tools.registry import registry

TERMINAL_SCHEMA = {
    "name": "terminal",
    "description": TERMINAL_TOOL_DESCRIPTION,
    "parameters": {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "The shell command to execute"
            },
            "background": {
                "type": "boolean",
                "description": "Keep running after this call returns; returns a session_id.",
                "default": False
            },
            "timeout": {
                "type": "integer",
                "description": f"Max seconds (default 180, foreground max {FOREGROUND_MAX_TIMEOUT}). Finishing early returns at once, so set it generously. Above {FOREGROUND_MAX_TIMEOUT}s the command runs as a tracked background process with notify_on_complete=true (the result says so; do not re-run it).",
                "minimum": 1
            },
            "workdir": {
                "type": "string",
                "description": "Absolute working directory for this command only."
            },
            "pty": {
                "type": "boolean",
                "description": "With background=true: pseudo-terminal for interactive CLIs (REPLs, coding agents); drive it with process_manage write/submit. Local backend only.",
                "default": False
            },
            "notify": {
                "description": "With background=true. true: one notification on exit; use it for anything that ends (builds, tests, deploys), or the job runs silently. ['pattern', ...]: notify when a line matches, only for one-shot readiness on processes that never exit; rate-limited. Omit for servers/daemons.",
                "anyOf": [
                    {"type": "boolean"},
                    {"type": "array", "items": {"type": "string"}}
                ]
            },
            "heartbeat": {
                "type": "integer",
                "minimum": 0,
                "default": 0,
                "description": "With background=true: also notify every N seconds (min 60) with new output; quiet ticks are skipped. For bounded jobs you must react to mid-run, never servers. Implies notify=true; 0 = off.",
            },
            "persist_on_release": {
                "type": "boolean",
                "default": False,
                "description": "With background=true: survive session end, /new, compression and other cleanup (not host exit). Only for jobs the user explicitly wants to outlive the conversation."
            }
            # Legacy aliases (unadvertised, still accepted): notify_on_complete
            # (bool) and watch_patterns (list). notify=true|[...] maps onto
            # them in the dispatch wrapper; explicit notify wins on conflict.
        },
        "required": ["command"]
    }
}


def _handle_terminal(args, **kw):
    from agent.terminal_approval_batch import validate_prepared_terminal
    validate_prepared_terminal(args)
    # Models sometimes send execute_code's ``code`` here; name the stray
    # argument and the right tool instead of failing on command=None.
    if "command" not in args and "code" in args:
        return tool_error(
            "terminal received a 'code' parameter, but it requires a shell "
            "command in 'command'. Use execute_code(code=...) for Python; "
            "for shell, retry as terminal(command=...)."
        )
    # `notify` is the advertised interface (true → notify_on_complete,
    # [...] → watch_patterns); the legacy args stay accepted, explicit
    # `notify` wins. Notification INTENT on a foreground call is refused with
    # the corrected call. A foreground `heartbeat` carries no intent: it has
    # no execution meaning without a tracked process, and providers that
    # materialize every schema property (or models copying a background call
    # shape) send `heartbeat=60` on ordinary commands — refusing it produced
    # identical-call retry loops (46 subagent sessions, 4 refusals each, 38 of
    # which never ran a single command), so it is normalized away instead.
    notify = args.get("notify")
    notify_on_complete = args.get("notify_on_complete", False)
    watch_patterns = args.get("watch_patterns")
    heartbeat = args.get("heartbeat") or 0
    persist_on_release = bool(args.get("persist_on_release", False))
    if not isinstance(heartbeat, int) or isinstance(heartbeat, bool) or heartbeat < 0:
        return tool_error("heartbeat must be a whole number of seconds (0 disables; positive values are clamped to min 60).")
    if not args.get("background", False):
        if notify or watch_patterns or notify_on_complete:
            return tool_error(
                "notify only applies to background commands (foreground "
                "results return directly). Either drop it, or run as "
                "terminal(command=..., background=true, notify=...)."
            )
        heartbeat = 0
        if args.get("pty", False):
            return tool_error(
                "pty requires background=true (a PTY session is interacted "
                "with via process(action='write'/'submit'), which needs a "
                "tracked background process). Retry as terminal(command=..., "
                "background=true, pty=true)."
            )
        if persist_on_release:
            return tool_error(
                "persist_on_release only applies to background commands (a foreground "
                "process is awaited inline and has nothing to persist). Retry as "
                "terminal(command=..., background=true, persist_on_release=true)."
            )
    if notify is not None:
        if isinstance(notify, bool):
            notify_on_complete = notify
            watch_patterns = None
        elif isinstance(notify, list):
            watch_patterns = notify
            notify_on_complete = False
        else:
            return tool_error(
                "notify must be true/false (notify on exit) or a list of "
                "strings (notify on output pattern match)."
            )
    if heartbeat:
        notify_on_complete = True  # the heartbeat rides the completion delivery path
    return terminal_tool(
        command=args.get("command"),
        background=args.get("background", False),
        timeout=args.get("timeout"),
        task_id=kw.get("task_id"),
        session_id=kw.get("session_id"),
        workdir=args.get("workdir"),
        pty=args.get("pty", False),
        notify_on_complete=notify_on_complete,
        watch_patterns=watch_patterns,
        heartbeat=heartbeat,
        persist_on_release=persist_on_release,
    )


registry.register(
    name="terminal",
    toolset="terminal",
    schema=TERMINAL_SCHEMA,
    handler=_handle_terminal,
    check_fn=check_terminal_requirements,
    emoji="💻",
    max_result_size_chars=100_000,
)