Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 68 additions & 1 deletion src/claude_agent_sdk/_internal/transport/subprocess_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import re
import shutil
import signal
import tempfile
from collections.abc import AsyncIterable, AsyncIterator
from contextlib import suppress
from pathlib import Path
Expand All @@ -30,6 +31,15 @@
_DEFAULT_MAX_BUFFER_SIZE = 1024 * 1024 # 1MB buffer limit
MINIMUM_CLAUDE_CODE_VERSION = "2.0.0"

# Linux caps a single argv element (MAX_ARG_STRLEN) at 32 * PAGE_SIZE, which is
# 131072 bytes (including the trailing NUL) on the common 4 KiB page size. A
# ``--system-prompt`` value at or above this makes execve() fail with E2BIG
# ("Argument list too long") before the CLI ever starts, so oversized prompts
# are written to a temp file and passed via ``--system-prompt-file`` instead
# (see issue #1096). The threshold is a fixed conservative value: systems with
# larger pages allow more, and spilling to a file early is harmless.
_MAX_SYSTEM_PROMPT_ARG_BYTES = 131072

# Track live CLI subprocesses so we can terminate them when the parent Python
# process exits. This mirrors the TypeScript SDK's parent-exit cleanup and
# prevents orphaned `claude` processes from leaking when callers crash or exit
Expand Down Expand Up @@ -138,6 +148,8 @@ def __init__(
else _DEFAULT_MAX_BUFFER_SIZE
)
self._write_lock: anyio.Lock = anyio.Lock()
# Temp file backing an oversized --system-prompt-file, removed in close().
self._system_prompt_tmp: Path | None = None

def _find_cli(self) -> str:
"""Find Claude Code CLI binary."""
Expand Down Expand Up @@ -279,6 +291,43 @@ def _apply_skills_defaults(

return allowed_tools, setting_sources

def _remove_system_prompt_tmp(self) -> None:
"""Remove the temp file backing ``--system-prompt-file``, if any."""
if self._system_prompt_tmp is not None:
with suppress(OSError):
self._system_prompt_tmp.unlink()
self._system_prompt_tmp = None

def _write_system_prompt_file(self, system_prompt: str) -> str:
"""Write an oversized system prompt to a temp file and return its path.

Passing a very large ``--system-prompt`` value directly overflows the
per-argument OS limit and aborts the CLI before startup (issue #1096).
The file is removed in :meth:`close`.
"""
# _build_command() may run more than once; drop the previous temp file
# so it cannot leak.
self._remove_system_prompt_tmp()
# NamedTemporaryFile owns the fd, so it cannot leak even if opening or
# writing fails; delete=False because the CLI reads it after we return.
tmp = None
try:
with tempfile.NamedTemporaryFile(
"w",
encoding="utf-8",
prefix="claude-system-prompt-",
suffix=".txt",
delete=False,
) as tmp:
tmp.write(system_prompt)
except BaseException:
if tmp is not None:
with suppress(OSError):
Path(tmp.name).unlink()
raise
self._system_prompt_tmp = Path(tmp.name)
return tmp.name

def _build_command(self) -> list[str]:
"""Build CLI command with arguments."""
if self._cli_path is None:
Expand All @@ -288,7 +337,20 @@ def _build_command(self) -> list[str]:
if self._options.system_prompt is None:
cmd.extend(["--system-prompt", ""])
elif isinstance(self._options.system_prompt, str):
cmd.extend(["--system-prompt", self._options.system_prompt])
system_prompt = self._options.system_prompt
# Measure the same bytes subprocess hands to execve(): argv elements
# are encoded with the filesystem encoding, not necessarily UTF-8.
if len(os.fsencode(system_prompt)) >= _MAX_SYSTEM_PROMPT_ARG_BYTES:
# Too large to pass as a single argv element (see #1096); hand it
# to the CLI as a file via the already-supported flag instead.
cmd.extend(
[
"--system-prompt-file",
self._write_system_prompt_file(system_prompt),
]
)
else:
cmd.extend(["--system-prompt", system_prompt])
else:
sp = self._options.system_prompt
if sp.get("type") == "file":
Expand Down Expand Up @@ -649,6 +711,11 @@ async def close(self) -> None:
atexit reaper instead of dropping it. Making the escalation robust to a
foreign `CancelledError` is a follow-up.
"""
# Remove the temp file backing an oversized --system-prompt-file, if we
# created one. Runs regardless of process state so a failed connect()
# does not leak it (see #1096).
self._remove_system_prompt_tmp()

if not self._process:
self._ready = False
return
Expand Down
84 changes: 83 additions & 1 deletion tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,16 @@
import uuid
from collections.abc import AsyncIterator
from contextlib import nullcontext
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import anyio
import pytest

from claude_agent_sdk._internal.transport.subprocess_cli import SubprocessCLITransport
from claude_agent_sdk._internal.transport.subprocess_cli import (
_MAX_SYSTEM_PROMPT_ARG_BYTES,
SubprocessCLITransport,
)
from claude_agent_sdk.types import ClaudeAgentOptions

DEFAULT_CLI_PATH = "/usr/bin/claude"
Expand Down Expand Up @@ -183,6 +187,84 @@ def test_build_command_with_system_prompt_file(self):
assert "--system-prompt-file" in cmd
assert "/path/to/prompt.md" in cmd

def test_build_command_with_oversized_system_prompt_uses_file(self):
"""A system prompt at the argv limit goes via file, not inline (#1096)."""
big = "x" * _MAX_SYSTEM_PROMPT_ARG_BYTES # smallest size that must spill
transport = SubprocessCLITransport(
prompt="test",
options=make_options(system_prompt=big),
)

cmd = transport._build_command()
try:
# Must not be inlined as a single oversized argv element.
assert big not in cmd
assert "--system-prompt" not in cmd
assert "--system-prompt-file" in cmd

path = cmd[cmd.index("--system-prompt-file") + 1]
assert transport._system_prompt_tmp is not None
assert Path(path).read_text(encoding="utf-8") == big
finally:
# close() would normally remove this; don't leave it behind if an
# assertion above fails.
if transport._system_prompt_tmp is not None:
transport._system_prompt_tmp.unlink(missing_ok=True)

def test_build_command_system_prompt_under_limit_stays_inline(self):
"""The largest prompt that fits in one argv element stays inline (#1096)."""
prompt = "x" * (_MAX_SYSTEM_PROMPT_ARG_BYTES - 1)
transport = SubprocessCLITransport(
prompt="test",
options=make_options(system_prompt=prompt),
)

cmd = transport._build_command()
assert "--system-prompt" in cmd
assert "--system-prompt-file" not in cmd
assert transport._system_prompt_tmp is None

def test_build_command_twice_replaces_oversized_prompt_file(self):
"""Rebuilding the command must not leak the previous temp file (#1096)."""
big = "x" * _MAX_SYSTEM_PROMPT_ARG_BYTES
transport = SubprocessCLITransport(
prompt="test",
options=make_options(system_prompt=big),
)

first = second = None
try:
transport._build_command()
first = transport._system_prompt_tmp
transport._build_command()
second = transport._system_prompt_tmp

assert first is not None
assert second is not None
assert not first.exists() # replaced file was removed
assert second.exists()
finally:
for path in (first, second):
if path is not None:
path.unlink(missing_ok=True)

@pytest.mark.anyio
async def test_close_removes_oversized_system_prompt_file(self):
"""close() deletes the temp file created for an oversized prompt (#1096)."""
transport = SubprocessCLITransport(
prompt="test",
options=make_options(system_prompt="x" * _MAX_SYSTEM_PROMPT_ARG_BYTES),
)
transport._build_command()

tmp = transport._system_prompt_tmp
assert tmp is not None and tmp.exists()

await transport.close()

assert not tmp.exists()
assert transport._system_prompt_tmp is None

def test_build_command_with_options(self):
"""Test building CLI command with options."""
transport = SubprocessCLITransport(
Expand Down