"""Terminal API routes — interactive PTY sessions with whitelist security.

Security model:
  - Commands matching ShellTool's _SAFE_COMMAND_PREFIXES are executed directly.
  - Non-whitelist commands require user confirmation via WebSocket.
  - User can choose to add confirmed command to session whitelist.
  - Session whitelist is stored in memory only (not persisted).
  - All endpoints require TERMINAL_LOCAL_USE permission + is_terminal_authorized flag.
"""

from __future__ import annotations

import asyncio
import hmac
import json
import logging
import os
import shlex
import time
import uuid
from dataclasses import dataclass, field
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect
from pydantic import BaseModel

from agentkit.server.auth.dependencies import require_terminal_authorized
from agentkit.server.auth.permissions import Permission, has_permission
from agentkit.server.auth.terminal_security import (
    DECISION_BLOCKED,
    DECISION_CONFIRMED,
    DECISION_EXECUTED,
    DECISION_REJECTED,
    SafetyDecision,
    check_command_safety_v2,
    compute_whitelist_entry,
    write_audit_log,
)
from agentkit.tools.pty_session import PTYSession
from agentkit.tools.shell import (
    _SAFE_COMMAND_PREFIXES,
    _DANGEROUS_BINARIES,
    _DANGEROUS_BINARY_FLAGS,
    _DANGEROUS_ARG_PATTERNS,
    _SHELL_PIPE_OPERATORS,
    _SHELL_CHAIN_OPERATORS,
)

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/terminal", tags=["terminal"])


# ── In-memory session store ───────────────────────────────────────────


@dataclass
class TerminalSessionState:
    """Per-session state for a terminal connection."""

    session_id: str
    pty: PTYSession | None = None
    cwd: str = ""
    created_at: float = field(default_factory=time.time)
    whitelist: set[str] = field(default_factory=set)
    history: list[dict[str, Any]] = field(default_factory=list)


_sessions: dict[str, TerminalSessionState] = {}


def _get_or_create_session(session_id: str | None = None) -> TerminalSessionState:
    """Get an existing session or create a new one."""
    if session_id and session_id in _sessions:
        return _sessions[session_id]

    sid = session_id or str(uuid.uuid4())
    state = TerminalSessionState(session_id=sid, cwd=os.getcwd())
    _sessions[sid] = state
    return state


def _cleanup_session(session_id: str) -> None:
    """Remove a session and close its PTY."""
    state = _sessions.pop(session_id, None)
    if state and state.pty and state.pty.is_running:
        # Best-effort close; cannot await here
        try:
            asyncio.get_running_loop().create_task(state.pty.close())
        except RuntimeError:
            pass


# ── Security helpers ──────────────────────────────────────────────────


def _is_single_command_dangerous(command: str) -> bool:
    """Check if a single command (no pipes/chains) is dangerous.

    Mirrors ShellTool._is_single_command_dangerous logic.
    """
    command_stripped = command.strip()
    if not command_stripped:
        return True

    try:
        tokens = shlex.split(command_stripped)
        if not tokens:
            return True
        binary = os.path.basename(tokens[0])
    except ValueError:
        return True

    # Whitelist check
    for prefix in _SAFE_COMMAND_PREFIXES:
        prefix_stripped = prefix.lower().strip()
        if " " in prefix_stripped:
            if command_stripped.lower().startswith(prefix_stripped):
                return False
        else:
            if binary.lower() == prefix_stripped:
                return False

    binary_lower = binary.lower()

    # Always dangerous binaries
    if binary_lower in _DANGEROUS_BINARIES:
        return True

    # Dangerous with specific flags
    if binary_lower in _DANGEROUS_BINARY_FLAGS:
        cmd_str = " ".join(tokens).lower()
        for flag_pattern in _DANGEROUS_BINARY_FLAGS[binary_lower]:
            if flag_pattern in cmd_str:
                return True
        return False

    # Cross-token dangerous patterns
    command_lower = command_stripped.lower()
    for pattern in _DANGEROUS_ARG_PATTERNS:
        if pattern.search(command_lower):
            return True

    return True


def _is_dangerous(command: str) -> bool:
    """Check if command is dangerous (mirrors ShellTool._is_dangerous)."""
    command_stripped = command.strip()

    if _SHELL_CHAIN_OPERATORS.search(command_stripped):
        return True

    if _SHELL_PIPE_OPERATORS.search(command_stripped):
        parts = command_stripped.split("|")
        for part in parts:
            part = part.strip()
            if not part:
                continue
            if _is_single_command_dangerous(part):
                return True
        return False

    return _is_single_command_dangerous(command_stripped)


def _check_command_safety(command: str, session_id: str) -> dict[str, Any]:
    """Check if command is safe to execute.

    Returns dict with:
      - safe: bool
      - reason: str (only when not safe)

    .. deprecated::
        This function only checks the session whitelist + built-in danger
        detection. It does NOT consult the DB-backed blocklist or user
        whitelist. New code should use :func:`check_command_safety_v2`
        via :func:`_check_safety_with_db`.
    """
    # Check session whitelist first
    state = _sessions.get(session_id)
    if state:
        command_stripped = command.strip()
        try:
            tokens = shlex.split(command_stripped)
            if tokens:
                binary = os.path.basename(tokens[0]).lower()
                # Check if binary or full command prefix is in session whitelist
                for wl_entry in state.whitelist:
                    if binary == wl_entry.lower() or command_stripped.lower().startswith(wl_entry.lower()):
                        return {"safe": True}
        except ValueError:
            pass

    # Check global whitelist / danger detection
    if not _is_dangerous(command):
        return {"safe": True}

    return {"safe": False, "reason": "此命令被识别为潜在危险操作，需要用户确认"}


async def _resolve_db_path_from_app(app_state: Any) -> Any:
    """Resolve the auth DB path from app.state (or None in dev mode)."""
    path = getattr(app_state, "auth_db_path", None)
    if path:
        from pathlib import Path

        return Path(path)
    from agentkit.server.auth.models import DEFAULT_AUTH_DB_PATH

    return DEFAULT_AUTH_DB_PATH


async def _check_safety_with_db(
    command: str,
    state: TerminalSessionState,
    user: dict[str, Any] | None,
    app_state: Any,
) -> SafetyDecision:
    """Run the three-layer whitelist check with DB access.

    Falls back to the legacy in-memory check if the DB path cannot be
    resolved (e.g. tests without an auth DB).
    """
    db_path = await _resolve_db_path_from_app(app_state)
    user_id = user.get("user_id") if user and not user.get("dev_mode") else None

    return await check_command_safety_v2(
        command,
        session_id=state.session_id,
        session_whitelist=state.whitelist,
        user_id=user_id,
        db_path=db_path,
    )


async def _audit(
    app_state: Any,
    *,
    user: dict[str, Any] | None,
    session_id: str,
    command: str,
    decision: str,
    reason: str | None = None,
    cwd: str | None = None,
    exit_code: int | None = None,
    terminal_mode: str = "local",
) -> None:
    """Write an audit log entry (best-effort, never raises)."""
    db_path = await _resolve_db_path_from_app(app_state)
    user_id = user.get("user_id") if user and not user.get("dev_mode") else None
    username = user.get("username") if user else None
    await write_audit_log(
        db_path,
        user_id=user_id,
        username=username,
        session_id=session_id,
        command=command,
        decision=decision,
        reason=reason,
        cwd=cwd,
        exit_code=exit_code,
        terminal_mode=terminal_mode,
    )


# ── Request/Response schemas ──────────────────────────────────────────


class ExecuteRequest(BaseModel):
    command: str
    session_id: str | None = None
    timeout: float = 30.0


class ExecuteResponse(BaseModel):
    session_id: str
    command: str
    exit_code: int
    output: str
    cwd: str
    duration_ms: int
    confirmation_required: bool = False
    reason: str | None = None


class SessionInfo(BaseModel):
    session_id: str
    cwd: str
    created_at: float


class HistoryResponse(BaseModel):
    history: list[dict[str, Any]]


# ── REST endpoints ────────────────────────────────────────────────────


@router.post("/execute", response_model=ExecuteResponse)
async def execute_command(
    request: ExecuteRequest,
    http_request: Request,
    user: dict[str, Any] = Depends(require_terminal_authorized),
):
    """Execute a command. Returns output for safe commands, or confirmation_required for unsafe ones."""
    state = _get_or_create_session(request.session_id)

    # Three-layer whitelist safety check (blocklist → builtin → user → session → danger)
    decision = await _check_safety_with_db(
        request.command, state, user, http_request.app.state
    )

    if not decision.safe:
        # Audit: denied / blocked
        await _audit(
            http_request.app.state,
            user=user,
            session_id=state.session_id,
            command=request.command,
            decision=decision.decision,
            reason=decision.reason,
            cwd=state.cwd,
            terminal_mode="local",
        )
        return ExecuteResponse(
            session_id=state.session_id,
            command=request.command,
            exit_code=126,
            output="",
            cwd=state.cwd,
            duration_ms=0,
            confirmation_required=True,
            reason=decision.reason or "需要用户确认",
        )

    # Execute the command
    start_time = time.monotonic()
    try:
        pty = PTYSession()
        await pty.start()
        result = await pty.run_command(
            request.command,
            timeout=request.timeout,
            cwd=state.cwd or None,
        )
        await pty.close()

        output = result.output
        exit_code = result.exit_code
        duration_ms = int((time.monotonic() - start_time) * 1000)

        # Update cwd if command was cd
        if request.command.strip().startswith("cd "):
            try:
                check_pty = PTYSession()
                await check_pty.start()
                cwd_result = await check_pty.run_command("pwd", timeout=5.0, cwd=state.cwd or None)
                await check_pty.close()
                if cwd_result.exit_code == 0:
                    state.cwd = cwd_result.output.strip()
            except (OSError, RuntimeError, asyncio.TimeoutError):
                pass

    except asyncio.TimeoutError:
        output = f"命令执行超时（{request.timeout}s）"
        exit_code = -1
        duration_ms = int((time.monotonic() - start_time) * 1000)
    except (OSError, RuntimeError, ValueError) as e:
        output = str(e)
        exit_code = -1
        duration_ms = int((time.monotonic() - start_time) * 1000)

    # Record history
    state.history.append({
        "command": request.command,
        "exit_code": exit_code,
        "output": output[:5000],
        "cwd": state.cwd,
        "timestamp": time.time(),
        "duration_ms": duration_ms,
    })

    # Audit: executed (whitelisted) — confirmation flow is WebSocket-only
    await _audit(
        http_request.app.state,
        user=user,
        session_id=state.session_id,
        command=request.command,
        decision=DECISION_EXECUTED,
        reason=f"matched_layer={decision.matched_layer}",
        cwd=state.cwd,
        exit_code=exit_code,
        terminal_mode="local",
    )

    return ExecuteResponse(
        session_id=state.session_id,
        command=request.command,
        exit_code=exit_code,
        output=output,
        cwd=state.cwd,
        duration_ms=duration_ms,
    )


@router.get("/sessions", response_model=dict)
async def list_sessions(
    user: dict[str, Any] = Depends(require_terminal_authorized),
):
    """List active terminal sessions."""
    sessions = [
        SessionInfo(
            session_id=s.session_id,
            cwd=s.cwd,
            created_at=s.created_at,
        ).model_dump()
        for s in _sessions.values()
    ]
    return {"sessions": sessions}


@router.get("/sessions/{session_id}/history", response_model=HistoryResponse)
async def get_session_history(
    session_id: str,
    limit: int = 100,
    offset: int = 0,
    user: dict[str, Any] = Depends(require_terminal_authorized),
):
    """Get command history for a session."""
    state = _sessions.get(session_id)
    if state is None:
        raise HTTPException(status_code=404, detail=f"Session '{session_id}' not found")

    history = state.history[offset: offset + limit]
    return HistoryResponse(history=history)


@router.delete("/sessions/{session_id}")
async def close_session(
    session_id: str,
    user: dict[str, Any] = Depends(require_terminal_authorized),
):
    """Close and remove a terminal session."""
    if session_id not in _sessions:
        raise HTTPException(status_code=404, detail=f"Session '{session_id}' not found")
    _cleanup_session(session_id)
    return {"status": "closed", "session_id": session_id}


# ── WebSocket endpoint ────────────────────────────────────────────────


@router.websocket("/ws")
async def terminal_websocket(websocket: WebSocket) -> None:
    """WebSocket endpoint for interactive terminal.

    Client → Server messages:
        {"type": "execute", "command": "..."}                          — Execute a command
        {"type": "confirm", "command": "...", "add_to_whitelist": bool} — Confirm a blocked command
        {"type": "input", "data": "..."}                               — Send input to running PTY
        {"type": "resize", "rows": N, "cols": N}                       — Resize terminal
        {"type": "ping"}                                               — Heartbeat

    Server → Client messages:
        {"type": "connected", "session_id": "..."}                     — Connection confirmed
        {"type": "output", "data": "..."}                              — Terminal output chunk
        {"type": "result", "command": "...", "exit_code": N, ...}      — Command completed
        {"type": "confirmation_required", "command": "...", "reason"}  — Need user confirmation
        {"type": "error", "message": "..."}                            — Error occurred
        {"type": "pong"}                                               — Heartbeat response
    """
    # Authentication (accept first, then check — FastAPI requires accept before close)
    await websocket.accept()

    # --- Auth + permission check ---
    # The terminal WebSocket requires TERMINAL_LOCAL_USE permission.
    # Auth is checked via (in order):
    #   1. JWT via ?token= query param (preferred for browser clients)
    #   2. API key via ?api_key= query param (legacy / programmatic clients)
    #   3. Dev mode (no auth configured) — still requires no permission check,
    #      but terminal is a high-risk endpoint so dev mode is NOT allowed
    #      unless explicitly enabled via app.state.allow_dev_terminal.

    jwt_secret: str | None = getattr(websocket.app.state, "jwt_secret", None)
    configured_api_key: str | None = None
    if hasattr(websocket.app.state, "server_config") and websocket.app.state.server_config:
        configured_api_key = websocket.app.state.server_config.api_key
    if configured_api_key is None and hasattr(websocket.app.state, "api_key"):
        configured_api_key = websocket.app.state.api_key

    user_payload: dict[str, Any] | None = None
    auth_ok = False

    # 1. JWT via ?token=
    jwt_token = websocket.query_params.get("token")
    if jwt_token and jwt_secret:
        try:
            from agentkit.server.auth.jwt_utils import verify_token

            payload = verify_token(jwt_token, jwt_secret)
            if payload.get("type") == "access":
                user_payload = {
                    "user_id": payload.get("sub"),
                    "username": payload.get("username"),
                    "role": payload.get("role"),
                }
                # Check role permission
                if has_permission(user_payload, Permission.TERMINAL_LOCAL_USE):
                    auth_ok = True
                else:
                    await websocket.send_json({
                        "type": "error",
                        "message": "Permission denied: TERMINAL_LOCAL_USE required",
                    })
                    await websocket.close(code=4003, reason="Permission denied")
                    return
        except (ValueError, KeyError, RuntimeError, OSError):
            pass  # Fall through to API key / dev mode

    # 2. API key via ?api_key=
    if not auth_ok and configured_api_key:
        provided = websocket.query_params.get("api_key")
        if hmac.compare_digest((provided or "").encode(), configured_api_key.encode()):
            auth_ok = True
        else:
            await websocket.send_json({"type": "error", "message": "Invalid api_key"})
            await websocket.close(code=4001, reason="Invalid api_key")
            return

    # 3. Dev mode — only allowed if explicitly enabled
    if not auth_ok:
        allow_dev = getattr(websocket.app.state, "allow_dev_terminal", False)
        if not jwt_secret and not configured_api_key and allow_dev:
            auth_ok = True
        else:
            await websocket.send_json({
                "type": "error",
                "message": "Authentication required",
            })
            await websocket.close(code=4001, reason="Authentication required")
            return

    # Resolve or create session
    session_id = websocket.query_params.get("session_id")
    state = _get_or_create_session(session_id)

    # Track pending confirmations: confirmation_id -> Future[bool]
    pending_confirmations: dict[str, asyncio.Future] = {}

    # Currently running PTY for streaming output
    active_pty: PTYSession | None = None

    try:
        await websocket.send_json({
            "type": "connected",
            "session_id": state.session_id,
        })

        while True:
            try:
                raw = await asyncio.wait_for(websocket.receive_text(), timeout=300.0)
            except asyncio.TimeoutError:
                await websocket.send_json({"type": "pong"})
                continue

            try:
                msg = json.loads(raw)
            except json.JSONDecodeError:
                continue

            msg_type = msg.get("type")

            if msg_type == "execute":
                command = msg.get("command", "").strip()
                if not command:
                    continue

                # Three-layer whitelist safety check
                # (blocklist → builtin → user → session → danger)
                decision = await _check_safety_with_db(
                    command, state, user_payload, websocket.app.state
                )

                # Blocklist match: reject outright, no confirmation prompt
                if decision.decision == DECISION_BLOCKED:
                    await _audit(
                        websocket.app.state,
                        user=user_payload,
                        session_id=state.session_id,
                        command=command,
                        decision=DECISION_BLOCKED,
                        reason=decision.reason,
                        cwd=state.cwd,
                        terminal_mode="local",
                    )
                    await websocket.send_json({
                        "type": "result",
                        "command": command,
                        "exit_code": 126,
                        "output": f"命令被黑名单拒绝: {decision.reason or ''}",
                        "cwd": state.cwd,
                        "duration_ms": 0,
                    })
                    continue

                # Dangerous command: require user confirmation
                if not decision.safe:
                    confirmation_id = str(uuid.uuid4())
                    await websocket.send_json({
                        "type": "confirmation_required",
                        "confirmation_id": confirmation_id,
                        "command": command,
                        "reason": decision.reason or "需要用户确认",
                    })
                    # Wait for confirmation by reading messages inline
                    # (cannot await a future because the main loop is blocked)
                    loop = asyncio.get_running_loop()
                    future: asyncio.Future[bool] = loop.create_future()
                    pending_confirmations[confirmation_id] = future
                    approved = False
                    add_to_whitelist_flag = False
                    try:
                        # Read messages until we get the confirmation or timeout
                        deadline = time.monotonic() + 300.0
                        while time.monotonic() < deadline:
                            try:
                                raw = await asyncio.wait_for(
                                    websocket.receive_text(),
                                    timeout=min(deadline - time.monotonic(), 30.0),
                                )
                            except asyncio.TimeoutError:
                                continue

                            try:
                                confirm_msg = json.loads(raw)
                            except json.JSONDecodeError:
                                continue

                            if (confirm_msg.get("type") == "confirm"
                                    and confirm_msg.get("confirmation_id") == confirmation_id):
                                approved = confirm_msg.get("approved", False)
                                add_to_whitelist_flag = confirm_msg.get("add_to_whitelist", False)
                                break

                            # Handle other message types while waiting
                            if confirm_msg.get("type") == "ping":
                                await websocket.send_json({"type": "pong"})

                    except asyncio.TimeoutError:
                        pass
                    finally:
                        pending_confirmations.pop(confirmation_id, None)

                    if not approved:
                        # Audit: user rejected the confirmation
                        await _audit(
                            websocket.app.state,
                            user=user_payload,
                            session_id=state.session_id,
                            command=command,
                            decision=DECISION_REJECTED,
                            reason="用户拒绝确认",
                            cwd=state.cwd,
                            terminal_mode="local",
                        )
                        await websocket.send_json({
                            "type": "result",
                            "command": command,
                            "exit_code": 126,
                            "output": "命令被用户拒绝",
                            "cwd": state.cwd,
                            "duration_ms": 0,
                        })
                        continue

                    # Add to session whitelist if requested
                    if add_to_whitelist_flag:
                        entry = compute_whitelist_entry(command)
                        if entry:
                            state.whitelist.add(entry)

                    # Audit: user confirmed — will execute
                    await _audit(
                        websocket.app.state,
                        user=user_payload,
                        session_id=state.session_id,
                        command=command,
                        decision=DECISION_CONFIRMED,
                        reason=f"matched_layer={decision.matched_layer}",
                        cwd=state.cwd,
                        terminal_mode="local",
                    )

                # Execute command with streaming output
                start_time = time.monotonic()
                try:
                    pty = PTYSession()
                    await pty.start()
                    active_pty = pty

                    # Start command execution in background
                    async def _run_and_stream():
                        result = await pty.run_command(
                            command,
                            timeout=msg.get("timeout", 30.0),
                            cwd=state.cwd or None,
                        )
                        return result

                    run_task = asyncio.create_task(_run_and_stream())

                    # Stream output while command runs
                    while not run_task.done():
                        try:
                            chunk = await asyncio.wait_for(pty.read_output(timeout=0.5), timeout=1.0)
                            if chunk:
                                await websocket.send_json({
                                    "type": "output",
                                    "data": chunk,
                                })
                        except asyncio.TimeoutError:
                            continue
                        except (OSError, RuntimeError):
                            break

                    # Get final result
                    result = await run_task
                    await pty.close()
                    active_pty = None

                    # Drain any remaining output
                    duration_ms = int((time.monotonic() - start_time) * 1000)

                    # Update cwd if command was cd
                    if command.strip().startswith("cd "):
                        try:
                            check_pty = PTYSession()
                            await check_pty.start()
                            cwd_result = await check_pty.run_command("pwd", timeout=5.0, cwd=state.cwd or None)
                            await check_pty.close()
                            if cwd_result.exit_code == 0:
                                state.cwd = cwd_result.output.strip()
                        except (OSError, RuntimeError, asyncio.TimeoutError):
                            pass

                    # Record history
                    state.history.append({
                        "command": command,
                        "exit_code": result.exit_code,
                        "output": result.output[:5000],
                        "cwd": state.cwd,
                        "timestamp": time.time(),
                        "duration_ms": duration_ms,
                    })

                    # Audit: command executed (either whitelisted or confirmed)
                    # If the command was confirmed above, we already wrote a
                    # DECISION_CONFIRMED audit entry. For whitelisted commands
                    # (no confirmation needed), write DECISION_EXECUTED here.
                    if decision.safe:
                        await _audit(
                            websocket.app.state,
                            user=user_payload,
                            session_id=state.session_id,
                            command=command,
                            decision=DECISION_EXECUTED,
                            reason=f"matched_layer={decision.matched_layer}",
                            cwd=state.cwd,
                            exit_code=result.exit_code,
                            terminal_mode="local",
                        )

                    await websocket.send_json({
                        "type": "result",
                        "command": command,
                        "exit_code": result.exit_code,
                        "output": result.output,
                        "cwd": state.cwd,
                        "duration_ms": duration_ms,
                    })

                except asyncio.TimeoutError:
                    duration_ms = int((time.monotonic() - start_time) * 1000)
                    await websocket.send_json({
                        "type": "result",
                        "command": command,
                        "exit_code": -1,
                        "output": "命令执行超时",
                        "cwd": state.cwd,
                        "duration_ms": duration_ms,
                    })
                except asyncio.CancelledError:
                    raise
                except Exception as e:
                    duration_ms = int((time.monotonic() - start_time) * 1000)
                    await websocket.send_json({
                        "type": "error",
                        "message": str(e)[:200],
                    })
                finally:
                    if active_pty and active_pty.is_running:
                        try:
                            await active_pty.close()
                        except (OSError, RuntimeError):
                            pass
                        active_pty = None

            elif msg_type == "confirm":
                # Handle confirmation reply from client
                confirmation_id = msg.get("confirmation_id")
                approved = msg.get("approved", False)
                add_to_whitelist = msg.get("add_to_whitelist", False)

                if confirmation_id and confirmation_id in pending_confirmations:
                    future = pending_confirmations[confirmation_id]
                    if not future.done():
                        future.set_result(approved)

                    # Add to whitelist if approved and requested
                    if approved and add_to_whitelist:
                        command = msg.get("command", "")
                        entry = compute_whitelist_entry(command)
                        if entry:
                            state.whitelist.add(entry)

            elif msg_type == "input":
                # Send input to active PTY
                if active_pty and active_pty.is_running:
                    await active_pty.send(msg.get("data", ""))

            elif msg_type == "resize":
                # Terminal resize (future support)
                pass

            elif msg_type == "ping":
                await websocket.send_json({"type": "pong"})

    except WebSocketDisconnect:
        logger.debug(f"Terminal WebSocket disconnected for session {state.session_id}")
    except asyncio.CancelledError:
        raise
    except Exception as e:
        logger.error(f"Terminal WebSocket error for session {state.session_id}: {e}")
        try:
            await websocket.send_json({"type": "error", "message": str(e)[:200]})
        except (ConnectionError, RuntimeError, asyncio.TimeoutError):
            pass
    finally:
        # Clean up pending confirmations
        for fut in pending_confirmations.values():
            if not fut.done():
                fut.cancel()
        # Close active PTY
        if active_pty and active_pty.is_running:
            try:
                await active_pty.close()
            except (OSError, RuntimeError):
                pass
