orchestrata_ultimate.drivers.base

Reference › orchestrata_ultimate › drivers › base

base defines AsyncCodeDriver, the runtime-checkable protocol that every code-generation driver implements (SPEC-025, REQ-025-001 to REQ-025-006). It has one async method, execute(task, files, is_fix), which returns a tuple of a success flag and output text. The concrete drivers (Claude Code, Claude CLI with stdin, Aider, Qwen, DeepSeek and browser) satisfy it, and driver_factory returns it as the common type. The code director and junior volks and the swarm coordinator depend on the protocol and not on a concrete driver.

At a glance


Source	`src/orchestrata_ultimate/drivers/base.py`
Lines of code	48
Classes	1
Functions	0
Imports (internal)	`orchestrata_ultimate.core.traceability`
Imported by	`orchestrata_ultimate.agents.volks.codedirector.review`, `orchestrata_ultimate.agents.volks.codedirector.volk`, `orchestrata_ultimate.agents.volks.junior.volk`, `orchestrata_ultimate.drivers`, `orchestrata_ultimate.drivers.driver_factory`
Tests that import it	`tests/drivers/test_base.py`, `tests/generated_recovery/test_drivers_base.py`, `tests/test_drivers.py`

Class diagram

Dependency diagram

API

Classes

`AsyncCodeDriver` (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='025', reqs=['REQ-025-001', 'REQ-025-002', 'REQ-025-003', 'REQ-025-004', 'REQ-025-005', 'REQ-025-006']) · Line 20

Protocol for async code generation drivers.

Method	Signature	Summary
`execute`	`async execute(task: str, files: list[str] \| None = None, is_fix: bool = False) -> tuple[bool, str]`	Execute a code generation task.

Source

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

"""Base protocol for async code generation drivers."""

from typing import Protocol, runtime_checkable

from orchestrata_ultimate.core.traceability import traces


@runtime_checkable
@traces(
    spec="025",
    reqs=[
        "REQ-025-001",
        "REQ-025-002",
        "REQ-025-003",
        "REQ-025-004",
        "REQ-025-005",
        "REQ-025-006",
    ],
)
class AsyncCodeDriver(Protocol):
    """Protocol for async code generation drivers.

    All drivers must implement execute() which sends a coding task
    to an external tool and returns (success, output).
    """

    @traces(
        spec="025",
        story="US-025-001",
        reqs=["REQ-025-003", "REQ-025-004"],
    )
    async def execute(
        self,
        task: str,
        files: list[str] | None = None,
        is_fix: bool = False,
    ) -> tuple[bool, str]:
        """Execute a code generation task.

        Args:
            task: The coding task / prompt.
            files: Optional list of file paths to operate on.
            is_fix: True if this is a repair/fix prompt.

        Returns:
            Tuple of (success, output_text).
        """
        ...