subprocess_cli.py

"""Subprocess transport implementation using Claude Code CLI."""

import json
import logging
import os
import platform
import re
import shutil
import tempfile
from collections.abc import AsyncIterable, AsyncIterator
from contextlib import suppress
from pathlib import Path
from subprocess import PIPE
from typing import Any, cast

import anyio
from anyio.abc import Process
from anyio.streams.text import TextReceiveStream, TextSendStream

from ..._errors import CLIConnectionError, CLINotFoundError, ProcessError
from ..._errors import CLIJSONDecodeError as SDKJSONDecodeError
from ..._version import __version__
from ...types import ClaudeAgentOptions, SystemPromptFile, SystemPromptPreset
from .._task_compat import TaskHandle, spawn_detached
from . import Transport

logger = logging.getLogger(__name__)

_DEFAULT_MAX_BUFFER_SIZE = 1024 * 1024  # 1MB buffer limit
MINIMUM_CLAUDE_CODE_VERSION = "2.0.0"


class SubprocessCLITransport(Transport):
    """Subprocess transport using Claude Code CLI."""

    def __init__(
        self,
        prompt: str | AsyncIterable[dict[str, Any]],
        options: ClaudeAgentOptions,
    ):
        self._prompt = prompt
        # Always use streaming mode internally (matching TypeScript SDK)
        # This allows agents and other large configs to be sent via initialize request
        self._is_streaming = True
        self._options = options
        self._cli_path: str | None = (
            str(options.cli_path) if options.cli_path is not None else None
        )
        self._cwd = str(options.cwd) if options.cwd else None
        self._process: Process | None = None
        self._stdout_stream: TextReceiveStream | None = None
        self._stdin_stream: TextSendStream | None = None
        self._stderr_stream: TextReceiveStream | None = None
        self._stderr_task: TaskHandle | None = None
        self._ready = False
        self._exit_error: Exception | None = None  # Track process exit errors
        self._max_buffer_size = (
            options.max_buffer_size
            if options.max_buffer_size is not None
            else _DEFAULT_MAX_BUFFER_SIZE
        )
        self._generated_system_prompt_file: Path | None = None
        self._write_lock: anyio.Lock = anyio.Lock()

    def _find_cli(self) -> str:
        """Find Claude Code CLI binary."""
        # First, check for bundled CLI
        bundled_cli = self._find_bundled_cli()
        if bundled_cli:
            return bundled_cli

        # Fall back to system-wide search
        if cli := shutil.which("claude"):
            return cli

        locations = [
            Path.home() / ".npm-global/bin/claude",
            Path("/usr/local/bin/claude"),
            Path.home() / ".local/bin/claude",
            Path.home() / "node_modules/.bin/claude",
            Path.home() / ".yarn/bin/claude",
            Path.home() / ".claude/local/claude",
        ]

        for path in locations:
            if path.exists() and path.is_file():
                return str(path)

        raise CLINotFoundError(
            "Claude Code not found. Install with:\n"
            "  npm install -g @anthropic-ai/claude-code\n"
            "\nIf already installed locally, try:\n"
            '  export PATH="$HOME/node_modules/.bin:$PATH"\n'
            "\nOr provide the path via ClaudeAgentOptions:\n"
            "  ClaudeAgentOptions(cli_path='/path/to/claude')"
        )

    def _find_bundled_cli(self) -> str | None:
        """Find bundled CLI binary if it exists."""
        # Determine the CLI binary name based on platform
        cli_name = "claude.exe" if platform.system() == "Windows" else "claude"

        # Get the path to the bundled CLI
        # The _bundled directory is in the same package as this module
        bundled_path = Path(__file__).parent.parent.parent / "_bundled" / cli_name

        if bundled_path.exists() and bundled_path.is_file():
            logger.info(f"Using bundled Claude Code CLI: {bundled_path}")
            return str(bundled_path)

        return None

    def _build_settings_value(self) -> str | None:
        """Build settings value, merging sandbox settings if provided.

        Returns the settings value as either:
        - A JSON string (if sandbox is provided or settings is JSON)
        - A file path (if only settings path is provided without sandbox)
        - None if neither settings nor sandbox is provided
        """
        has_settings = self._options.settings is not None
        has_sandbox = self._options.sandbox is not None

        if not has_settings and not has_sandbox:
            return None

        # If only settings path and no sandbox, pass through as-is
        if has_settings and not has_sandbox:
            return self._options.settings

        # If we have sandbox settings, we need to merge into a JSON object
        settings_obj: dict[str, Any] = {}

        if has_settings:
            assert self._options.settings is not None
            settings_str = self._options.settings.strip()
            # Check if settings is a JSON string or a file path
            if settings_str.startswith("{") and settings_str.endswith("}"):
                # Parse JSON string
                try:
                    settings_obj = json.loads(settings_str)
                except json.JSONDecodeError:
                    # If parsing fails, treat as file path
                    logger.warning(
                        f"Failed to parse settings as JSON, treating as file path: {settings_str}"
                    )
                    # Read the file
                    settings_path = Path(settings_str)
                    if settings_path.exists():
                        with settings_path.open(encoding="utf-8") as f:
                            settings_obj = json.load(f)
            else:
                # It's a file path - read and parse
                settings_path = Path(settings_str)
                if settings_path.exists():
                    with settings_path.open(encoding="utf-8") as f:
                        settings_obj = json.load(f)
                else:
                    logger.warning(f"Settings file not found: {settings_path}")

        # Merge sandbox settings
        if has_sandbox:
            settings_obj["sandbox"] = self._options.sandbox

        return json.dumps(settings_obj)

    def _apply_skills_defaults(
        self,
    ) -> tuple[list[str], list[str] | None]:
        """Compute effective allowed_tools and setting_sources for skills.

        When ``options.skills`` is ``"all"``, injects the bare ``Skill`` tool;
        when it is a list, injects ``Skill(name)`` for each entry. In either
        case ``setting_sources`` defaults to ``["user", "project"]`` when
        unset so the CLI discovers installed skills without the caller having
        to wire up both options manually. ``None`` is a no-op.

        Does not mutate the original options object.
        """
        allowed_tools: list[str] = list(self._options.allowed_tools)
        setting_sources: list[str] | None = (
            list(self._options.setting_sources)
            if self._options.setting_sources is not None
            else None
        )

        skills = self._options.skills
        if skills is None:
            return allowed_tools, setting_sources

        if skills == "all":
            if "Skill" not in allowed_tools:
                allowed_tools.append("Skill")
        else:
            for name in skills:
                pattern = f"Skill({name})"
                if pattern not in allowed_tools:
                    allowed_tools.append(pattern)

        if setting_sources is None:
            setting_sources = ["user", "project"]

        return allowed_tools, setting_sources

    def _materialize_system_prompt_blocks(self, blocks: list[dict[str, Any]]) -> str:
        """Persist structured system prompt blocks for CLI handoff."""
        if self._generated_system_prompt_file is None:
            with tempfile.NamedTemporaryFile(
                mode="w",
                encoding="utf-8",
                prefix="claude-agent-sdk-system-prompt-",
                suffix=".json",
                delete=False,
            ) as temp_file:
                json.dump(blocks, temp_file)
                self._generated_system_prompt_file = Path(temp_file.name)

        return str(self._generated_system_prompt_file)

    def _cleanup_generated_system_prompt_file(self) -> None:
        """Remove any temporary system prompt file created by the SDK."""
        path = self._generated_system_prompt_file
        self._generated_system_prompt_file = None
        if path is not None:
            with suppress(FileNotFoundError):
                path.unlink()

    def _build_command(self) -> list[str]:
        """Build CLI command with arguments."""
        if self._cli_path is None:
            raise CLINotFoundError("CLI path not resolved. Call connect() first.")
        cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"]

        sp = self._options.system_prompt
        if sp is None:
            cmd.extend(["--system-prompt", ""])
        elif isinstance(sp, str):
            cmd.extend(["--system-prompt", sp])
        elif isinstance(sp, list):
            cmd.extend(
                [
                    "--system-prompt-file",
                    self._materialize_system_prompt_blocks(sp),
                ]
            )
        elif isinstance(sp, dict):
            if sp.get("type") == "file":
                cmd.extend(["--system-prompt-file", cast(SystemPromptFile, sp)["path"]])
            elif sp.get("type") == "preset" and "append" in sp:
                cmd.extend(
                    ["--append-system-prompt", cast(SystemPromptPreset, sp)["append"]]
                )
        else:
            raise TypeError(
                "system_prompt must be None, a string, a system prompt preset/file "
                "mapping, or a list of Anthropic content blocks"
            )

        # Handle tools option (base set of tools)
        if self._options.tools is not None:
            tools = self._options.tools
            if isinstance(tools, list):
                if len(tools) == 0:
                    cmd.extend(["--tools", ""])
                else:
                    cmd.extend(["--tools", ",".join(tools)])
            else:
                # Preset object - 'claude_code' preset maps to 'default'
                cmd.extend(["--tools", "default"])

        effective_allowed_tools, effective_setting_sources = (
            self._apply_skills_defaults()
        )

        if effective_allowed_tools:
            cmd.extend(["--allowedTools", ",".join(effective_allowed_tools)])

        if self._options.max_turns:
            cmd.extend(["--max-turns", str(self._options.max_turns)])

        if self._options.max_budget_usd is not None:
            cmd.extend(["--max-budget-usd", str(self._options.max_budget_usd)])

        if self._options.disallowed_tools:
            cmd.extend(["--disallowedTools", ",".join(self._options.disallowed_tools)])

        if self._options.task_budget is not None:
            cmd.extend(["--task-budget", str(self._options.task_budget["total"])])

        if self._options.model:
            cmd.extend(["--model", self._options.model])

        if self._options.fallback_model:
            cmd.extend(["--fallback-model", self._options.fallback_model])

        if self._options.betas:
            cmd.extend(["--betas", ",".join(self._options.betas)])

        if self._options.permission_prompt_tool_name:
            cmd.extend(
                ["--permission-prompt-tool", self._options.permission_prompt_tool_name]
            )

        if self._options.permission_mode:
            cmd.extend(["--permission-mode", self._options.permission_mode])

        if self._options.continue_conversation:
            cmd.append("--continue")

        if self._options.resume:
            cmd.extend(["--resume", self._options.resume])

        if self._options.session_id:
            cmd.extend(["--session-id", self._options.session_id])

        # Handle settings and sandbox: merge sandbox into settings if both are provided
        settings_value = self._build_settings_value()
        if settings_value:
            cmd.extend(["--settings", settings_value])

        if self._options.add_dirs:
            # Convert all paths to strings and add each directory
            for directory in self._options.add_dirs:
                cmd.extend(["--add-dir", str(directory)])

        if self._options.mcp_servers:
            if isinstance(self._options.mcp_servers, dict):
                # Process all servers, stripping instance field from SDK servers
                servers_for_cli: dict[str, Any] = {}
                for name, config in self._options.mcp_servers.items():
                    if isinstance(config, dict) and config.get("type") == "sdk":
                        # For SDK servers, pass everything except the instance field
                        sdk_config: dict[str, object] = {
                            k: v for k, v in config.items() if k != "instance"
                        }
                        servers_for_cli[name] = sdk_config
                    else:
                        # For external servers, pass as-is
                        servers_for_cli[name] = config

                # Pass all servers to CLI
                if servers_for_cli:
                    cmd.extend(
                        [
                            "--mcp-config",
                            json.dumps({"mcpServers": servers_for_cli}),
                        ]
                    )
            else:
                # String or Path format: pass directly as file path or JSON string
                cmd.extend(["--mcp-config", str(self._options.mcp_servers)])

        if self._options.include_partial_messages:
            cmd.append("--include-partial-messages")

        if self._options.fork_session:
            cmd.append("--fork-session")

        if self._options.session_store is not None:
            cmd.append("--session-mirror")

        # Agents are always sent via initialize request (matching TypeScript SDK)
        # No --agents CLI flag needed

        if effective_setting_sources is not None:
            cmd.append(f"--setting-sources={','.join(effective_setting_sources)}")

        # Add plugin directories
        if self._options.plugins:
            for plugin in self._options.plugins:
                if plugin["type"] == "local":
                    cmd.extend(["--plugin-dir", plugin["path"]])
                else:
                    raise ValueError(f"Unsupported plugin type: {plugin['type']}")

        # Add extra args for future CLI flags
        for flag, value in self._options.extra_args.items():
            if value is None:
                # Boolean flag without value
                cmd.append(f"--{flag}")
            else:
                # Flag with value
                cmd.extend([f"--{flag}", str(value)])

        # Resolve thinking config -> --thinking / --max-thinking-tokens
        # `thinking` takes precedence over the deprecated `max_thinking_tokens`
        if self._options.thinking is not None:
            t = self._options.thinking
            if t["type"] == "adaptive":
                cmd.extend(["--thinking", "adaptive"])
            elif t["type"] == "enabled":
                cmd.extend(["--max-thinking-tokens", str(t["budget_tokens"])])
            elif t["type"] == "disabled":
                cmd.extend(["--thinking", "disabled"])

            # Narrow off the Disabled variant first so mypy knows `t["display"]` is a str
            # rather than widening to `object` across the union.
            if t["type"] != "disabled" and "display" in t:
                cmd.extend(["--thinking-display", t["display"]])
        elif self._options.max_thinking_tokens is not None:
            cmd.extend(
                ["--max-thinking-tokens", str(self._options.max_thinking_tokens)]
            )

        if self._options.effort is not None:
            cmd.extend(["--effort", self._options.effort])

        # Extract schema from output_format structure if provided
        # Expected: {"type": "json_schema", "schema": {...}}
        if (
            self._options.output_format is not None
            and isinstance(self._options.output_format, dict)
            and self._options.output_format.get("type") == "json_schema"
        ):
            schema = self._options.output_format.get("schema")
            if schema is not None:
                cmd.extend(["--json-schema", json.dumps(schema)])

        # Always use streaming mode with stdin (matching TypeScript SDK)
        # This allows agents and other large configs to be sent via initialize request
        cmd.extend(["--input-format", "stream-json"])

        return cmd

    async def connect(self) -> None:
        """Start subprocess."""
        if self._process:
            return

        if self._cli_path is None:
            self._cli_path = await anyio.to_thread.run_sync(self._find_cli)

        if not os.environ.get("CLAUDE_AGENT_SDK_SKIP_VERSION_CHECK"):
            await self._check_claude_version()

        cmd = self._build_command()
        try:
            # Merge environment variables. CLAUDE_CODE_ENTRYPOINT defaults to
            # sdk-py regardless of inherited process env; options.env can override
            # it. CLAUDE_AGENT_SDK_VERSION is always set by the SDK.
            # Filter out CLAUDECODE so SDK-spawned subprocesses don't think
            # they're running inside a Claude Code parent (see #573).
            inherited_env = {k: v for k, v in os.environ.items() if k != "CLAUDECODE"}
            process_env = {
                **inherited_env,
                "CLAUDE_CODE_ENTRYPOINT": "sdk-py",
                **self._options.env,
                "CLAUDE_AGENT_SDK_VERSION": __version__,
            }

            # Propagate active OTEL trace context to the CLI so its spans
            # parent under the caller's distributed trace. No-op if
            # opentelemetry-api is not installed or there's no active span.
            try:
                from opentelemetry import propagate

                carrier: dict[str, str] = {}
                propagate.inject(carrier)
                if "traceparent" in carrier:
                    # Active span present: scrub stale inherited W3C context
                    # (CI/k8s ambient env) before writing the fresh values, so
                    # an inherited TRACESTATE isn't paired with a new
                    # TRACEPARENT. Explicit ClaudeAgentOptions.env always wins.
                    # Gate on the traceparent key (not carrier truthiness) so a
                    # baggage-only / non-W3C carrier doesn't scrub a valid
                    # inherited TRACEPARENT.
                    for key in ("TRACEPARENT", "TRACESTATE"):
                        if key not in self._options.env:
                            process_env.pop(key, None)
                    for k, v in carrier.items():
                        key = k.upper()
                        if key not in self._options.env:
                            process_env[key] = v
            except Exception:  # noqa: BLE001 - best-effort tracing must never break connect()
                logger.debug("OTEL trace context injection failed", exc_info=True)

            # Enable file checkpointing if requested
            if self._options.enable_file_checkpointing:
                process_env["CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING"] = "true"

            if self._cwd:
                process_env["PWD"] = self._cwd

            # Pipe stderr only when the caller registered a callback.
            stderr_dest = PIPE if self._options.stderr is not None else None

            self._process = await anyio.open_process(
                cmd,
                stdin=PIPE,
                stdout=PIPE,
                stderr=stderr_dest,
                cwd=self._cwd,
                env=process_env,
                user=self._options.user,
            )

            if self._process.stdout:
                self._stdout_stream = TextReceiveStream(self._process.stdout)

            # Setup stderr stream if piped
            if stderr_dest is PIPE and self._process.stderr:
                self._stderr_stream = TextReceiveStream(self._process.stderr)
                # Spawn the stderr reader via spawn_detached (not a manually-
                # entered TaskGroup) so cleanup has no trio task-affinity —
                # same pattern as Query._read_task.
                self._stderr_task = spawn_detached(self._handle_stderr())

            # Setup stdin for streaming (always used now)
            if self._process.stdin:
                self._stdin_stream = TextSendStream(self._process.stdin)

            self._ready = True

        except FileNotFoundError as e:
            self._cleanup_generated_system_prompt_file()
            # Check if the error comes from the working directory or the CLI
            if self._cwd and not Path(self._cwd).exists():
                error = CLIConnectionError(
                    f"Working directory does not exist: {self._cwd}"
                )
                self._exit_error = error
                raise error from e
            error = CLINotFoundError(f"Claude Code not found at: {self._cli_path}")
            self._exit_error = error
            raise error from e
        except Exception as e:
            self._cleanup_generated_system_prompt_file()
            error = CLIConnectionError(f"Failed to start Claude Code: {e}")
            self._exit_error = error
            raise error from e

    async def _handle_stderr(self) -> None:
        """Handle stderr stream - read and invoke callbacks."""
        if not self._stderr_stream:
            return

        try:
            async for line in self._stderr_stream:
                line_str = line.rstrip()
                if not line_str:
                    continue

                # Call the stderr callback if provided
                if self._options.stderr:
                    self._options.stderr(line_str)
        except anyio.ClosedResourceError:
            pass  # Stream closed, exit normally
        except Exception:
            pass  # Ignore other errors during stderr reading

    async def close(self) -> None:
        """Close the transport and clean up resources."""
        if not self._process:
            self._ready = False
            self._cleanup_generated_system_prompt_file()
            return

        # Cancel stderr reader if active
        if self._stderr_task is not None and not self._stderr_task.done():
            self._stderr_task.cancel()
            with suppress(Exception):
                await self._stderr_task.wait()
        self._stderr_task = None

        # Close stdin stream (acquire lock to prevent race with concurrent writes)
        async with self._write_lock:
            self._ready = False  # Set inside lock to prevent TOCTOU with write()
            if self._stdin_stream:
                with suppress(Exception):
                    await self._stdin_stream.aclose()
                self._stdin_stream = None

        if self._stderr_stream:
            with suppress(Exception):
                await self._stderr_stream.aclose()
            self._stderr_stream = None

        # Wait for graceful shutdown after stdin EOF, then terminate if needed.
        # The subprocess needs time to flush its session file after receiving
        # EOF on stdin. Without this grace period, SIGTERM can interrupt the
        # write and cause the last assistant message to be lost (see #625).
        if self._process.returncode is None:
            try:
                with anyio.fail_after(5):
                    await self._process.wait()
            except TimeoutError:
                # Graceful shutdown timed out — force terminate
                with suppress(ProcessLookupError):
                    self._process.terminate()
                try:
                    with anyio.fail_after(5):
                        await self._process.wait()
                except TimeoutError:
                    # SIGTERM handler blocked — force kill (SIGKILL)
                    with suppress(ProcessLookupError):
                        self._process.kill()
                    with suppress(Exception):
                        await self._process.wait()

        self._process = None
        self._stdout_stream = None
        self._stdin_stream = None
        self._stderr_stream = None
        self._exit_error = None
        self._cleanup_generated_system_prompt_file()

    async def write(self, data: str) -> None:
        """Write raw data to the transport."""
        async with self._write_lock:
            # All checks inside lock to prevent TOCTOU races with close()/end_input()
            if not self._ready or not self._stdin_stream:
                raise CLIConnectionError("ProcessTransport is not ready for writing")

            if self._process and self._process.returncode is not None:
                raise CLIConnectionError(
                    f"Cannot write to terminated process (exit code: {self._process.returncode})"
                )

            if self._exit_error:
                raise CLIConnectionError(
                    f"Cannot write to process that exited with error: {self._exit_error}"
                ) from self._exit_error

            try:
                await self._stdin_stream.send(data)
            except Exception as e:
                self._ready = False
                self._exit_error = CLIConnectionError(
                    f"Failed to write to process stdin: {e}"
                )
                raise self._exit_error from e

    async def end_input(self) -> None:
        """End the input stream (close stdin)."""
        async with self._write_lock:
            if self._stdin_stream:
                with suppress(Exception):
                    await self._stdin_stream.aclose()
                self._stdin_stream = None

    def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        """Read and parse messages from the transport."""
        return self._read_messages_impl()

    async def _read_messages_impl(self) -> AsyncIterator[dict[str, Any]]:
        """Internal implementation of read_messages."""
        if not self._process or not self._stdout_stream:
            raise CLIConnectionError("Not connected")

        json_buffer = ""

        # Process stdout messages
        try:
            async for line in self._stdout_stream:
                line_str = line.strip()
                if not line_str:
                    continue

                # Accumulate partial JSON until we can parse it
                # Note: TextReceiveStream can truncate long lines, so we need to buffer
                # and speculatively parse until we get a complete JSON object
                json_lines = line_str.split("\n")

                for json_line in json_lines:
                    json_line = json_line.strip()
                    if not json_line:
                        continue

                    # Skip non-JSON lines (e.g. [SandboxDebug]) when not
                    # mid-parse — they corrupt the buffer otherwise (#347).
                    if not json_buffer and not json_line.startswith("{"):
                        logger.debug(
                            "Skipping non-JSON line from CLI stdout: %s",
                            json_line[:200],
                        )
                        continue

                    # Keep accumulating partial JSON until we can parse it
                    json_buffer += json_line

                    if len(json_buffer) > self._max_buffer_size:
                        buffer_length = len(json_buffer)
                        json_buffer = ""
                        raise SDKJSONDecodeError(
                            f"JSON message exceeded maximum buffer size of {self._max_buffer_size} bytes",
                            ValueError(
                                f"Buffer size {buffer_length} exceeds limit {self._max_buffer_size}"
                            ),
                        )

                    try:
                        data = json.loads(json_buffer)
                        json_buffer = ""
                        yield data
                    except json.JSONDecodeError:
                        # We are speculatively decoding the buffer until we get
                        # a full JSON object. If there is an actual issue, we
                        # raise an error after exceeding the configured limit.
                        continue

        except anyio.ClosedResourceError:
            pass
        except GeneratorExit:
            # Client disconnected
            pass

        # Check process completion and handle errors
        try:
            returncode = await self._process.wait()
        except Exception:
            returncode = -1

        # Use exit code for error detection
        if returncode is not None and returncode != 0:
            self._exit_error = ProcessError(
                f"Command failed with exit code {returncode}",
                exit_code=returncode,
                stderr="Check stderr output for details",
            )
            raise self._exit_error

    async def _check_claude_version(self) -> None:
        """Check Claude Code version and warn if below minimum."""
        if self._cli_path is None:
            raise CLINotFoundError("CLI path not resolved. Call connect() first.")
        version_process = None
        try:
            with anyio.fail_after(2):  # 2 second timeout
                version_process = await anyio.open_process(
                    [self._cli_path, "-v"],
                    stdout=PIPE,
                    stderr=PIPE,
                )

                if version_process.stdout:
                    stdout_bytes = await version_process.stdout.receive()
                    version_output = stdout_bytes.decode().strip()

                    match = re.match(r"([0-9]+\.[0-9]+\.[0-9]+)", version_output)
                    if match:
                        version = match.group(1)
                        version_parts = [int(x) for x in version.split(".")]
                        min_parts = [
                            int(x) for x in MINIMUM_CLAUDE_CODE_VERSION.split(".")
                        ]

                        if version_parts < min_parts:
                            logger.warning(
                                "Claude Code version %s at %s is unsupported in the Agent SDK. "
                                "Minimum required version is %s. "
                                "Some features may not work correctly.",
                                version,
                                self._cli_path,
                                MINIMUM_CLAUDE_CODE_VERSION,
                            )
        except Exception:
            pass
        finally:
            if version_process:
                with suppress(Exception):
                    version_process.terminate()
                with suppress(Exception):
                    await version_process.wait()

    def is_ready(self) -> bool:
        """Check if transport is ready for communication."""
        return self._ready