Software Design Document

Project: orchcore | Author: Abdelaziz Abdelrasol | Date: 2026-03-25 | Status: DRAFT

TL;DR

orchcore extracts the common orchestration infrastructure from four production AI agent orchestration systems (Planora, Articles, Finvault, Raven/Ralph) into a standalone Python package. It provides 12 components: agent registry, subprocess runner, stream processing pipeline, phase engine, rate-limit recovery, workspace management, configuration, prompt templating, logging, UI protocol, signal handling, and observability. The engine reports progress through the UICallback protocol, which keeps presentation out of the package. The extraction targets the estimated 60-70% of orchestration code that the four systems duplicate.

Problem Statement

Current Situation

• Four production orchestration systems, Planora (Python), Articles (Bash), Finvault (Bash), and Raven/Ralph (Go), each implement the same infrastructure: subprocess launching, JSONL stream parsing, rate-limit detection, workspace management, and configuration handling.
• An estimated 60-70% of the orchestration code across these four systems is functionally identical infrastructure.
• Bug fixes in one system must be manually ported to the other three.
• Each new agent CLI (Gemini, Copilot) requires N implementations across N systems.

Opportunity

• With a shared package, a fix or improvement reaches every consumer through a version bump.
• A new orchestration system can start from the shared package instead of reimplementing subprocess management, stream parsing, and recovery.
• The Python ecosystem lacks a purpose-built library for orchestrating CLI-based AI coding agents (existing tools like LangChain operate at the API level, not the subprocess level).

Goals

• Provide a pip-installable Python package with 12 documented, tested components
• Agent configurations are data: consumers register agents programmatically or load complete TOML entries; new CLIs using a supported stream format need no orchcore code changes
• Sequential and parallel phase execution with configurable partial failure semantics
• Process JSONL from 5 agent CLI formats into a unified StreamEvent model
• Automatic rate-limit recovery with timezone-aware reset parsing and exponential backoff
• mypy strict mode with zero errors
• Each component usable independently (no all-or-nothing dependency)

Non-Goals

• orchcore is not an AI agent. It does not make API calls or generate code.
• orchcore does not include a TUI framework. It provides the UICallback hooks that a TUI implements.
• orchcore does not handle API keys. Agent CLIs manage their own credentials, and which environment variables a subprocess inherits stays explicit and configurable.

Success Metrics

Requirements

Functional Requirements

Non-Functional Requirements

Configuration Resolution

Settings sources are resolved from highest priority to lowest:

1. Constructor kwargs / CLI overrides
2. Environment variables (ORCHCORE_*)
3. .env values
4. Profile overlay, if active
5. Project orchcore.toml
6. User ~/.config/orchcore/config.toml
7. pyproject.toml [tool.orchcore]
8. Built-in field defaults on OrchcoreSettings

Key Interfaces

UICallback Protocol

class UICallback(Protocol):
    def on_pipeline_start(self, phases: Sequence[Phase]) -> None: ...
    def on_pipeline_complete(self, result: PipelineResult) -> None: ...
    def on_phase_start(self, phase: Phase) -> None: ...
    def on_phase_end(self, phase: Phase, result: PhaseResult) -> None: ...
    def on_phase_skip(self, phase: Phase, reason: str) -> None: ...
    def on_agent_start(self, agent_name: str, phase: str) -> None: ...
    def on_agent_event(self, event: StreamEvent) -> None: ...
    def on_agent_complete(self, agent_name: str, result: AgentResult) -> None: ...
    def on_agent_error(self, agent_name: str, error: str) -> None: ...
    def on_stall_detected(self, agent_name: str, duration: float) -> None: ...
    def on_rate_limit(self, agent_name: str, message: str) -> None: ...
    def on_rate_limit_wait(self, agent_name: str, wait_seconds: float) -> None: ...
    def on_retry(self, agent_name: str, attempt: int, max_attempts: int) -> None: ...
    def on_git_recovery(self, action: str, detail: str) -> None: ...
    def on_shutdown(self, reason: str) -> None: ...

AgentRunner

class AgentRunner:
    async def run(
        self,
        agent: AgentConfig,
        prompt: str,
        output_path: Path,
        *,
        flag_profile: str | None = None,
        dry_run: bool = False,
        on_event: Callable[[StreamEvent], None] | None = None,
        on_snapshot: Callable[[AgentMonitorSnapshot], None] | None = None,
        snapshot_interval: float | None = None,
        stall_check_interval: float = 5.0,
        on_process_start: Callable[[asyncio.subprocess.Process], None] | None = None,
        on_process_end: Callable[[asyncio.subprocess.Process], None] | None = None,
        toolset: ToolSet | None = None,
        on_stall: Callable[[str, float], None] | None = None,
        cwd: Path | None = None,
    ) -> AgentResult: ...

PipelineRunner

class PipelineRunner:
    async def run_pipeline(
        self,
        phases: list[Phase],
        prompts: dict[str, str],
        ui_callback: UICallback,
        flag_profile: str | None = None,
        resume_from: str | None = None,
        skip_phases: list[str] | None = None,
        only_phase: str | None = None,
        allow_empty_prompts: bool = False,
    ) -> PipelineResult: ...

Alternatives Considered

Related

• Architecture Overview: package layout and component diagrams
• Architecture Decision Records: the list of all 11 ADRs
• Configuration Reference
• Stream Events Reference