claude_cli

claude_cli

Claude CLI backend using subprocess.

Wraps the claude CLI command for running prompts. Based on patterns from run-sheet-review.sh.

Security Note: This module uses asyncio.create_subprocess_exec() which is the safe subprocess method in Python - it does NOT use shell=True, so there is no shell injection risk. Arguments are passed as a list, not interpolated into a shell command string.

Progress Tracking: When a progress_callback is provided, this backend streams output in real-time and reports bytes/lines received periodically. This enables the CLI to show "Still running... 5.2KB received, 3m elapsed" during long executions.

Attributes

Classes

ClaudeCliBackend

ClaudeCliBackend(skip_permissions=True, disable_mcp=True, output_format='text', cli_model=None, allowed_tools=None, system_prompt_file=None, working_directory=None, timeout_seconds=1800.0, progress_callback=None, progress_interval_seconds=5.0, cli_extra_args=None)

Bases: Backend

Run prompts via the Claude CLI.

Uses asyncio.create_subprocess_exec to invoke claude -p <prompt>. This is shell-injection safe as arguments are passed as a list.

Initialize CLI backend.

Parameters:

Name	Type	Description	Default
skip_permissions	bool	Pass --dangerously-skip-permissions	True
disable_mcp	bool	Disable MCP servers for faster execution (--strict-mcp-config)	True
output_format	str	Output format (json, text, stream-json)	'text'
cli_model	str | None	Model to use (--model flag), None uses default	None
allowed_tools	list[str] | None	Restrict to specific tools (--allowedTools)	None
system_prompt_file	Path | None	Custom system prompt file (--system-prompt)	None
working_directory	Path | None	Working directory for running commands	None
timeout_seconds	float	Maximum time allowed per prompt	1800.0
progress_callback	ProgressCallback | None	Optional callback for progress updates during execution. Called with dict containing: bytes_received, lines_received, elapsed_seconds, phase.	None
progress_interval_seconds	float	How often to call progress callback (default 5s).	5.0
cli_extra_args	list[str] | None	Extra arguments to pass to claude CLI (escape hatch).	None

Source code in src/marianne/backends/claude_cli.py

def __init__(
    self,
    skip_permissions: bool = True,
    disable_mcp: bool = True,
    output_format: str = "text",
    cli_model: str | None = None,
    allowed_tools: list[str] | None = None,
    system_prompt_file: Path | None = None,
    working_directory: Path | None = None,
    timeout_seconds: float = 1800.0,  # 30 minute default
    progress_callback: ProgressCallback | None = None,
    progress_interval_seconds: float = 5.0,
    cli_extra_args: list[str] | None = None,
):
    """Initialize CLI backend.

    Args:
        skip_permissions: Pass --dangerously-skip-permissions
        disable_mcp: Disable MCP servers for faster execution (--strict-mcp-config)
        output_format: Output format (json, text, stream-json)
        cli_model: Model to use (--model flag), None uses default
        allowed_tools: Restrict to specific tools (--allowedTools)
        system_prompt_file: Custom system prompt file (--system-prompt)
        working_directory: Working directory for running commands
        timeout_seconds: Maximum time allowed per prompt
        progress_callback: Optional callback for progress updates during execution.
            Called with dict containing: bytes_received, lines_received,
            elapsed_seconds, phase.
        progress_interval_seconds: How often to call progress callback (default 5s).
        cli_extra_args: Extra arguments to pass to claude CLI (escape hatch).
    """
    self.skip_permissions = skip_permissions
    self.disable_mcp = disable_mcp
    self.output_format = output_format
    self.cli_model = cli_model
    self.allowed_tools = allowed_tools
    self.system_prompt_file = system_prompt_file
    self._working_directory = working_directory
    self.timeout_seconds = timeout_seconds
    self.progress_callback = progress_callback
    self.progress_interval_seconds = progress_interval_seconds
    self.cli_extra_args = cli_extra_args or []

    # PID tracking callbacks for orphan detection.
    # Set by the daemon's ProcessGroupManager when running under the conductor.
    # Standalone CLI mode leaves these as None — no orphan tracking needed.
    self._on_process_spawned: Callable[[int], None] | None = None
    self._on_process_exited: Callable[[int], None] | None = None

    # Real-time output logging paths (set per-sheet by runner)
    # Industry standard: separate files for stdout and stderr
    self._stdout_log_path: Path | None = None
    self._stderr_log_path: Path | None = None

    # Partial output accumulator — populated during execution so that
    # _handle_execution_timeout() can capture partial output on timeout
    # instead of returning empty strings.
    self._partial_stdout_chunks: list[bytes] = []
    self._partial_stderr_chunks: list[bytes] = []

    # Verify claude CLI is available
    self._claude_path = shutil.which("claude")

    # Track log write failures for filesystem flakiness visibility
    self.log_write_failures: int = 0

    # Use shared ErrorClassifier for rate limit detection
    # This ensures consistent classification with the runner
    self._error_classifier = ErrorClassifier()

    # Dynamic preamble — context-aware identity/position/retry info built
    # per-sheet by the runner via set_preamble().
    self._preamble: str | None = None

    # Prompt extensions (GH#76) — additional directives injected after the
    # preamble and before the user prompt. Set per-sheet by runner
    # via set_prompt_extensions().
    self._prompt_extensions: list[str] = []

    # Per-sheet overrides (GH#78) — saved originals for clear_overrides()
    self._saved_cli_model: str | None = None
    self._has_overrides: bool = False

Functions

from_config classmethod

from_config(config)

Create backend from configuration.

Source code in src/marianne/backends/claude_cli.py

@classmethod
def from_config(cls, config: BackendConfig) -> "ClaudeCliBackend":
    """Create backend from configuration."""
    return cls(
        skip_permissions=config.skip_permissions,
        disable_mcp=config.disable_mcp,
        output_format=config.output_format,
        cli_model=config.cli_model,
        allowed_tools=config.allowed_tools,
        system_prompt_file=config.system_prompt_file,
        working_directory=config.working_directory,
        timeout_seconds=config.timeout_seconds,
        cli_extra_args=config.cli_extra_args,
    )

apply_overrides

apply_overrides(overrides)

Apply per-sheet overrides for the next execution.

Source code in src/marianne/backends/claude_cli.py

def apply_overrides(self, overrides: dict[str, object]) -> None:
    """Apply per-sheet overrides for the next execution."""
    if not overrides:
        return
    self._saved_cli_model = self.cli_model
    self._has_overrides = True
    if "cli_model" in overrides:
        self.cli_model = str(overrides["cli_model"])

clear_overrides

clear_overrides()

Restore original backend parameters after per-sheet execution.

Source code in src/marianne/backends/claude_cli.py

def clear_overrides(self) -> None:
    """Restore original backend parameters after per-sheet execution."""
    if not self._has_overrides:
        return
    self.cli_model = self._saved_cli_model
    self._saved_cli_model = None
    self._has_overrides = False

set_output_log_path

set_output_log_path(path)

Set base path for real-time output logging.

Called per-sheet by runner to enable streaming output to log files. This provides visibility into Claude's output during long executions.

Uses industry-standard separate files for stdout and stderr: - {path}.stdout.log - standard output - {path}.stderr.log - standard error

This enables clean tail -f monitoring without stream interleaving.

Parameters:

Name	Type	Description	Default
path	Path | None	Base path for log files (without extension), or None to disable. Example: workspace/logs/sheet-01 creates sheet-01.stdout.log	required

Source code in src/marianne/backends/claude_cli.py

def set_output_log_path(self, path: Path | None) -> None:
    """Set base path for real-time output logging.

    Called per-sheet by runner to enable streaming output to log files.
    This provides visibility into Claude's output during long executions.

    Uses industry-standard separate files for stdout and stderr:
    - {path}.stdout.log - standard output
    - {path}.stderr.log - standard error

    This enables clean `tail -f` monitoring without stream interleaving.

    Args:
        path: Base path for log files (without extension), or None to disable.
              Example: workspace/logs/sheet-01 creates sheet-01.stdout.log
    """
    if path is None:
        self._stdout_log_path = None
        self._stderr_log_path = None
    else:
        self._stdout_log_path = path.with_suffix(".stdout.log")
        self._stderr_log_path = path.with_suffix(".stderr.log")

set_preamble

set_preamble(preamble)

Set the dynamic preamble for the next execution.

Parameters:

Name	Type	Description	Default
preamble	str | None	Preamble text to prepend, or None to clear.	required

Source code in src/marianne/backends/claude_cli.py

def set_preamble(self, preamble: str | None) -> None:
    """Set the dynamic preamble for the next execution.

    Args:
        preamble: Preamble text to prepend, or None to clear.
    """
    self._preamble = preamble

set_prompt_extensions

set_prompt_extensions(extensions)

Set prompt extensions for the next execution.

Extensions are additional directive blocks injected after the preamble. Called per-sheet by the runner to apply score-level and sheet-level extensions.

Parameters:

Name	Type	Description	Default
extensions	list[str]	List of extension text blocks. Empty strings are ignored.	required

Source code in src/marianne/backends/claude_cli.py

def set_prompt_extensions(self, extensions: list[str]) -> None:
    """Set prompt extensions for the next execution.

    Extensions are additional directive blocks injected after the
    preamble. Called per-sheet by the runner to apply score-level and
    sheet-level extensions.

    Args:
        extensions: List of extension text blocks. Empty strings are ignored.
    """
    self._prompt_extensions = [e for e in extensions if e.strip()]

execute async

execute(prompt, *, timeout_seconds=None)

Execute a prompt (Backend protocol implementation).

Source code in src/marianne/backends/claude_cli.py

async def execute(
    self, prompt: str, *, timeout_seconds: float | None = None,
) -> ExecutionResult:
    """Execute a prompt (Backend protocol implementation)."""
    return await self._execute_impl(prompt, timeout_seconds=timeout_seconds)

health_check async

health_check()

Check if claude CLI is available and responsive.

Source code in src/marianne/backends/claude_cli.py

async def health_check(self) -> bool:
    """Check if claude CLI is available and responsive."""
    if not self._claude_path:
        return False

    try:
        # Simple test prompt
        result = await self._execute_impl("Say 'ready' and nothing else.")
        return result.success and "ready" in result.stdout.lower()
    except (TimeoutError, OSError, RuntimeError) as e:
        _logger.warning("health_check_failed", error_type=type(e).__name__, error=str(e))
        return False

availability_check async

availability_check()

Check if the claude CLI binary exists and is executable.

Unlike health_check(), this does NOT send a prompt or consume API quota. Used after quota exhaustion waits.

Source code in src/marianne/backends/claude_cli.py

async def availability_check(self) -> bool:
    """Check if the claude CLI binary exists and is executable.

    Unlike health_check(), this does NOT send a prompt or consume
    API quota. Used after quota exhaustion waits.
    """
    if not self._claude_path:
        return False
    return os.path.isfile(self._claude_path) and os.access(
        self._claude_path, os.X_OK,
    )

Functions