orchestrata_ultimate.drivers.claude_code_driver

Reference › orchestrata_ultimate › drivers › claude_code_driver

claude_code_driver runs the Claude Code command-line tool as an asynchronous subprocess and returns its result (SPEC-025, REQ-025-021 and related). driver_factory creates ClaudeCodeDriver for a volk whose provider is claude_code, passing the volk alias along.

execute locates the binary with _find_claude and returns a failure message if it is missing. It then runs the command built by _build_command under asyncio.wait_for (default 900 seconds, resolved by _get_timeout_seconds) and hands the output to _parse_envelope, which reads the CLI's JSON output envelope. _build_command assembles the argument list: JSON output format, the volk's configured model if any, an appended system prompt with the shared volk team preamble from the persona module, a resume of the previous session when the task is a fix and a session id is known, and a --settings path. An explicit ORCHESTRATA_CLAUDE_SETTINGS always wins over a file generated at construction by claude_settings; ORCHESTRATA_CLAUDE_GOVERNANCE is the switch for that generation and _resolve_repo_root finds the repository root for the governance gate. Timeouts and a missing binary return a failure tuple and do not raise. Tests are in tests/drivers/test_claude_code_driver.py and tests/drivers/test_claude_code_driver_live.py.

At a glance

Class diagram

Dependency diagram

API

Classes

ClaudeCodeDriver

Decorators: traces(spec='025', reqs=['REQ-025-021']) · Line 114

Async CLI driver for Claude Code.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/drivers/claude_code_driver.py

# mypy: ignore-errors
"""Async Claude Code CLI driver.

Runs ``claude --print --dangerously-skip-permissions`` as an async
subprocess. Timeout enforced via asyncio.wait_for instead of a
watchdog thread.
"""

import asyncio
import json
import logging
import os
import shutil
from collections.abc import Mapping

from orchestrata_ultimate.core.paths import ROOT_ENV_VAR, resolve_project_root
from orchestrata_ultimate.core.traceability import traces
from orchestrata_ultimate.drivers.claude_settings import write_governance_settings

_GOVERNANCE_ENV_KEY = "ORCHESTRATA_CLAUDE_GOVERNANCE"
_SETTINGS_ENV_KEY = "ORCHESTRATA_CLAUDE_SETTINGS"
_ROOT_ENV_KEY = ROOT_ENV_VAR


@traces(spec="025", story="US-025-001", reqs=["REQ-025-096"])
def _resolve_repo_root(env: Mapping[str, str]) -> str:
    """Resolve the orchestrata-ultimate repo root for the governance-gate
    hook command (the ``uv run --project <repo_root> python
    <repo_root>/scripts/claude_governance_gate.py`` invocation built by
    ``claude_settings.write_governance_settings``).

    Resolved through ``core.paths.resolve_project_root()``, which walks up
    to the nearest ``pyproject.toml``. ``get_project_root`` once stopped at
    the first ancestor holding a ``prolog/`` directory, which this package
    also ships, so it returned ``src/orchestrata_ultimate`` (no ``scripts/``
    there) and the PreToolUse hook command failed outright; this function
    therefore counted four parents up from its own file instead. Since the
    proper install (dd919d55c) the file sits in site-packages, where four
    parents up is ``.venv/lib/python3.14t``. ``ORCHESTRATA_ROOT`` in
    *env*, when set, still overrides; the walk is handed an empty mapping
    so *env* stays the only environment this reads.

    Args:
        env: Environment mapping to check for an ``ORCHESTRATA_ROOT``
            override.

    Returns:
        Absolute path to the repo root.
    """
    override = env.get(_ROOT_ENV_KEY)
    if override:
        return override
    return str(resolve_project_root({}))


_logger = logging.getLogger(__name__)

TIMEOUT_SECONDS = 900


@traces(spec="025", story="US-025-001", reqs=["REQ-025-023"])
def _get_timeout_seconds(env: Mapping[str, str] | None = None) -> int:
    """Resolve the subprocess timeout, in seconds.

    Defaults to the spec-mandated ``TIMEOUT_SECONDS`` (900, REQ-025-023).
    Overridable via ``ORCHESTRATA_CLAUDE_TIMEOUT_SECONDS`` so the timeout
    branch can be exercised in tests without waiting out the real 900s.

    Args:
        env: Optional environment mapping. Defaults to ``os.environ``.

    Returns:
        Timeout in seconds.
    """
    env = env if env is not None else os.environ
    return int(env.get("ORCHESTRATA_CLAUDE_TIMEOUT_SECONDS", str(TIMEOUT_SECONDS)))


@traces(spec="025", story="US-025-001", reqs=["REQ-025-019", "REQ-025-020"])
def _find_claude(env: Mapping[str, str] | None = None) -> str:
    """Find the claude CLI binary.

    Args:
        env: Optional environment mapping. Defaults to ``os.environ``.

    Returns:
        Path to claude binary, or empty string if not found.
    """
    if env is None:
        found = shutil.which("claude")
        home = os.environ.get("HOME", "")
    else:
        found = shutil.which("claude", path=env.get("PATH", ""))
        home = env.get("HOME", "")

    if found:
        return found

    candidates = [
        os.path.join(home, ".local", "bin", "claude") if home else "",
        "/usr/bin/claude",
        "/usr/local/bin/claude",
    ]
    for candidate in candidates:
        if candidate and os.path.isfile(candidate):
            return candidate
    return ""


@traces(
    spec="025",
    reqs=["REQ-025-021"],
)
class ClaudeCodeDriver:
    """Async CLI driver for Claude Code.

    Runs ``claude --print`` in non-interactive mode. Output is
    streamed line-by-line and collected.

    Command (REQ-025-021): claude --print --dangerously-skip-permissions
    --output-format json [--model M] [--append-system-prompt P]
    [--resume SID] [--settings PATH] <task>

    Attributes:
        workspace: Directory where code is generated.
        volk: Optional volk alias driving --model / persona resolution.
        last_session_id: Claude Code session id from the most recent JSON
            envelope, used to --resume on the next is_fix round.
        _generated_settings_path: Path to a governance-gate settings file
            this driver auto-generated at construction time (see
            ``ORCHESTRATA_CLAUDE_GOVERNANCE``), or ``None`` when governance
            wasn't requested or generation failed.
    """

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-096"],
    )
    def __init__(
        self,
        workspace: str | None = None,
        env: Mapping[str, str] | None = None,
        volk: str | None = None,
    ) -> None:
        """Initialize with target workspace directory.

        Args:
            workspace: Working directory for the subprocess.
            env: Optional environment mapping. Defaults to ``os.environ``.
            volk: Optional volk alias (e.g. "J:Junior") used to resolve
                ``--model`` and to attach the shared team persona via
                ``--append-system-prompt``. ``None`` preserves today's
                behaviour (no model flag, no persona flag).
        """
        self.env = env if env is not None else os.environ
        self.workspace = workspace or self.env.get(
            "ORCHESTRATA_WORKSPACE", "/home/mohamed/workspace"
        )
        self.volk = volk
        self.last_session_id: str | None = None
        os.makedirs(self.workspace, exist_ok=True)

        # Auto-generate the SPEC-025 Phase 2 governance-gate settings file
        # when governance is requested but no explicit settings file was
        # given: ORCHESTRATA_CLAUDE_GOVERNANCE="1" opts in, and an explicit
        # ORCHESTRATA_CLAUDE_SETTINGS always wins over the generated one
        # (checked again in _build_command). Never let a broken filesystem
        # seam (or a repo root we can't locate — FileNotFoundError is an
        # OSError subclass) crash driver construction: log and move on
        # without settings.
        self._generated_settings_path: str | None = None
        if self.env.get(_GOVERNANCE_ENV_KEY) == "1" and not self.env.get(
            _SETTINGS_ENV_KEY
        ):
            try:
                repo_root = _resolve_repo_root(self.env)
                self._generated_settings_path = write_governance_settings(
                    self.workspace, repo_root
                )
            except OSError as exc:
                _logger.warning(
                    "Could not auto-generate governance settings for %s: %s",
                    self.workspace,
                    exc,
                )

    @traces(
        spec="025",
        story="US-025-001",
        reqs=[
            "REQ-025-091",
            "REQ-025-092",
            "REQ-025-093",
            "REQ-025-094",
            "REQ-025-095",
        ],
    )
    def _build_command(self, claude_bin: str, task: str, is_fix: bool) -> list[str]:
        """Assemble the claude CLI invocation for one task.

        Args:
            claude_bin: Resolved path to the claude binary.
            task: The prompt to send to Claude Code.
            is_fix: True if this is a repair prompt (enables --resume).

        Returns:
            Full argv for the subprocess.
        """
        # Local imports avoid layer-3 peer-import violations: the driver sits in
        # the drivers layer and only needs these values when actually building a
        # command for a configured volk.
        from orchestrata_ultimate.agents.volks.persona import VOLK_TEAM_PREAMBLE
        from orchestrata_ultimate.llm.config import LLMConfigError, get_volk_model

        cmd = [
            claude_bin,
            "--print",
            "--dangerously-skip-permissions",
            "--output-format",
            "json",
        ]

        if self.volk is not None:
            try:
                model = get_volk_model(self.volk)
            except LLMConfigError as exc:
                _logger.debug("No model configured for volk %s: %s", self.volk, exc)
            else:
                if model:
                    cmd += ["--model", model]
            cmd += ["--append-system-prompt", VOLK_TEAM_PREAMBLE]

        if is_fix and self.last_session_id:
            cmd += ["--resume", self.last_session_id]

        # An explicit ORCHESTRATA_CLAUDE_SETTINGS always wins over the
        # settings file this driver may have auto-generated at construction
        # time (see __init__): explicit beats generated.
        settings_path = self.env.get(_SETTINGS_ENV_KEY) or self._generated_settings_path
        if settings_path:
            cmd += ["--settings", settings_path]

        cmd.append(task)
        return cmd

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-093", "REQ-025-094"],
    )
    def _parse_envelope(self, proc_ok: bool, raw_output: str) -> tuple[bool, str]:
        """Parse a ``--output-format json`` envelope from collected stdout.

        Falls back to the raw collected output exactly as before this
        driver spoke JSON, for older CLI versions / plain-text runs that
        don't emit an envelope.

        Claude Code CLI 2.1.207 (confirmed live) prints a JSON ARRAY of
        every message/tool event in the turn (system/init, assistant,
        rate_limit_event, user/tool_use_result, ...), terminated by exactly
        one ``{"type": "result", ...}`` object carrying ``result``/
        ``session_id``/``total_cost_usd``/``is_error`` -- NOT a single flat
        result dict as earlier CLI versions did. Both shapes are handled:
        a bare dict is used as-is; a list is reduced to its last
        ``type == "result"`` element.

        Args:
            proc_ok: Whether the subprocess exited zero.
            raw_output: Raw collected stdout/stderr text.

        Returns:
            (success, output_text) tuple.
        """
        try:
            envelope = json.loads(raw_output.strip())
        except json.JSONDecodeError, ValueError:
            return proc_ok, raw_output

        if isinstance(envelope, list):
            result_events = [
                event
                for event in envelope
                if isinstance(event, dict) and event.get("type") == "result"
            ]
            envelope = result_events[-1] if result_events else None

        if not isinstance(envelope, dict):
            return proc_ok, raw_output

        session_id = envelope.get("session_id")
        if session_id:
            self.last_session_id = session_id

        total_cost_usd = envelope.get("total_cost_usd")
        if total_cost_usd is not None:
            _logger.info("Claude Code session cost: $%s", total_cost_usd)

        is_error = bool(envelope.get("is_error", False))
        result_text = envelope.get("result", raw_output)
        return not is_error, result_text

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-021", "REQ-025-023", "REQ-025-025", "REQ-025-026"],
    )
    async def execute(
        self,
        task: str,
        files: list[str] | None = None,
        is_fix: bool = False,
    ) -> tuple[bool, str]:
        """Send a task to Claude Code CLI and return (success, output).

        Args:
            task: The prompt to send to Claude Code.
            files: Optional file paths (unused by this driver).
            is_fix: True if this is a repair prompt.

        Returns:
            Tuple of (success, output_text).
        """
        claude_bin = _find_claude(self.env)
        if not claude_bin:
            msg = "Claude Code CLI not found"
            _logger.error(msg)
            return False, msg

        _logger.info("Executing task in %s", self.workspace)

        cmd = self._build_command(claude_bin, task, is_fix)

        timeout_seconds = _get_timeout_seconds(self.env)
        try:
            proc_ok, raw_output = await asyncio.wait_for(
                self._run_process(cmd),
                timeout=timeout_seconds,
            )
        except TimeoutError:
            msg = f"Timed out after {timeout_seconds}s"
            _logger.warning(msg)
            return False, msg
        except FileNotFoundError:
            msg = f"Claude Code CLI not found at {claude_bin}"
            _logger.error(msg)
            return False, msg

        return self._parse_envelope(proc_ok, raw_output)

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-022", "REQ-025-024"],
    )
    async def _run_process(self, cmd: list[str]) -> tuple[bool, str]:
        """Run subprocess and collect output.

        Args:
            cmd: Command and arguments.

        Returns:
            Tuple of (success, output_text).
        """
        # Forward self.env (not just os.environ) to the subprocess: env vars
        # this driver itself reads through the injection seam --
        # ORCHESTRATA_CLAUDE_GATE_LOG in particular -- must also reach the
        # `claude` CLI process and, transitively, every PreToolUse hook
        # command IT spawns per tool call. Without this, a caller-supplied
        # env mapping was silently discarded at the actual subprocess
        # boundary and the governance-gate audit log never received a
        # single entry (confirmed live: gate.jsonl stayed empty even with
        # ORCHESTRATA_CLAUDE_GOVERNANCE=1 correctly wiring --settings).
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT,
            cwd=self.workspace,
            env=dict(self.env),
        )
        assert proc.stdout is not None  # guaranteed by stdout=PIPE above

        # Chunk reads, NOT readline(): --output-format json emits the whole
        # session's event array as ONE line, and readline() raises
        # LimitOverrunError past asyncio's 64KB stream limit — hit live the
        # first time a session ran long enough (the gen-skeleton leaf-fill
        # convergence run, 2026-07-13; the short fizzbuzz smoke fit under
        # the limit and never saw it).
        output_chunks: list[bytes] = []
        while True:
            chunk = await proc.stdout.read(65536)
            if not chunk:
                break
            _logger.debug("[Claude] read %d bytes", len(chunk))
            output_chunks.append(chunk)

        await proc.wait()
        output = b"".join(output_chunks).decode("utf-8", errors="replace")

        if proc.returncode != 0:
            _logger.warning("Exited with code %d", proc.returncode)
            return False, output

        _logger.info("Task completed successfully")
        return True, output