orchestrata_ultimate.session.cli.pipeline

Reference › orchestrata_ultimate › session › cli › pipeline

The pipeline commands drive the agent pipeline of the orchestrator from the session CLI. _get_runner builds a PipelineRunner for the project and _get_checkpoint_store opens the pipeline checkpoint store, if one is available. pipeline_run runs the full pipeline from an intent, with a work type, a flag for UI work and an optional thread id. pipeline_status shows status and stage progress for a thread. pipeline_resume resumes a checkpointed pipeline from its first incomplete stage. pipeline_respond sends a human response to the volk that is waiting on input, pipeline_approve approves the pending escalation, and pipeline_reject rejects it with feedback; _echo_escalation_result prints the outcome. pipeline_list lists recent runs from the output directory. The runner coroutines are executed from the synchronous Click commands. See the orchestrator pipeline runner.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/cli/pipeline.py

"""Pipeline CLI — run, status, resume, respond/approve/reject, and list
agent pipelines.

Usage:
    session pipeline run "Create a Redis task queue"
    session pipeline run "Refactor the auth module" --work-type refactor
    session pipeline run "Review the agent code" --work-type review
    session pipeline status <thread_id>
    session pipeline resume <thread_id>
    session pipeline respond --thread <thread_id> "<text>"
    session pipeline approve --thread <thread_id>
    session pipeline reject --thread <thread_id> "<reason>"
    session pipeline list

respond/approve/reject resolve a run PipelineRunner.run()/resume() suspended
with overall_status=escalated (D1-addendum-2, W3.1 escalation absorption —
plans/design-escalation-absorption-2026-07-06.md): a stage addressed HUMAN
while still working (blocked / needs review) and the run checkpointed the
escalation and stopped. respond delivers text to the same stage mid-
conversation; reject re-enters the same stage with feedback; approve resumes
the fixed sequence from the next stage. See
``orchestrator.pipeline_escalation.EscalationInterruptMixin``.
"""

from __future__ import annotations

import asyncio
import logging
import os
import sys
from collections.abc import Mapping
from pathlib import Path
from typing import TYPE_CHECKING

import click

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.kernel.pipeline_checkpoint import PipelineCheckpoint
    from orchestrata_ultimate.orchestrator.pipeline_runner import (
        PipelineResult,
        PipelineRunner,
    )

logger = logging.getLogger(__name__)


@traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-002", "REQ-003-FR24-003"])
def _get_checkpoint_store(
    env: Mapping[str, str] | None = None,
) -> PipelineCheckpoint | None:
    """Best-effort real checkpoint store construction.

    Returns ``None`` on any failure (misconfigured Redis env vars, a
    transient connection issue) so ``pipeline status``/``pipeline resume``
    can degrade to "no checkpoint found" instead of crashing with a
    traceback. Real ``kernel.pipeline_checkpoint.PipelineCheckpoint`` — not
    the nonexistent ``orchestrator.pipeline_checkpoint`` module the previous
    implementation imported (see the bugfix notes on ``pipeline_status``/
    ``pipeline_resume``).

    Args:
        env: Optional environment mapping (defaults to ``os.environ``).
            Exposed so tests can exercise misconfiguration without mutating
            the process environment.
    """
    try:
        from orchestrata_ultimate.core.redis_factory import create_sync_redis
        from orchestrata_ultimate.kernel.pipeline_checkpoint import (
            PipelineCheckpoint,
        )

        return PipelineCheckpoint(redis_client=create_sync_redis(env=env))
    except Exception:
        return None


@traces(spec="000", reqs=["REQ-000-053"])
def _get_runner(
    root: Path,
    env: Mapping[str, str] | None = None,
) -> PipelineRunner:
    """Create PipelineRunner with session integration."""
    from orchestrata_ultimate.orchestrator.artifact_store import ArtifactStore
    from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineRunner
    from orchestrata_ultimate.session.controller._core import SessionController

    ctrl = SessionController(str(root), ai_model="deepseek-v4")
    # Durable store: the JSONL layer under ``<root>/pipeline_output/artifacts``
    # is what makes a CLI run's work product retrievable after the process
    # exits (the memory-only construction here used to discard every artifact
    # on exit).
    store = ArtifactStore(output_dir=root / "pipeline_output" / "artifacts")
    return PipelineRunner(
        artifact_store=store,
        session_controller=ctrl,
        env=env,
        project_dir=str(root),
    )


@traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-001"])
@click.command("run")
@click.argument("intent")
@click.option(
    "--work-type",
    "-w",
    default="feature",
    type=click.Choice(
        [
            "feature",
            "refactor",
            "test",
            "documentation",
            "analysis",
            "fix",
            "review",
            "spec",
            "dependency",
            "performance",
        ]
    ),
    help="Work type (default: feature)",
)
@click.option(
    "--has-ui", is_flag=True, help="Project includes a UI (triggers UX agent)"
)
@click.option("--thread-id", default=None, help="Thread ID (auto-generated if omitted)")
@click.pass_context
def pipeline_run(
    ctx: click.Context,
    intent: str,
    work_type: str,
    has_ui: bool,
    thread_id: str | None,
) -> None:
    """Run the full agent pipeline from an intent.

    \b
    Examples:
        session pipeline run "Create a Redis task queue with REST API"
        session pipeline run "Review the auth module" --work-type review
        session pipeline run "Write docs for the API" --work-type documentation
        session pipeline run "Add dark mode to dashboard" --has-ui
    """
    root = ctx.obj.get("root", Path.cwd())
    env = ctx.obj.get("env", os.environ) if ctx.obj else os.environ
    runner = _get_runner(root, env=env)

    click.echo(
        f'Pipeline: {work_type} — "{intent[:80]}{"..." if len(intent) > 80 else ""}"'
    )
    click.echo(f"Thread:  {thread_id or 'auto-generated'}")
    click.echo()

    async def _run() -> PipelineResult:
        return await runner.run(
            intent=intent,
            work_type=work_type,
            thread_id=thread_id,
            has_ui=has_ui,
        )

    result = asyncio.run(_run())

    click.echo(f"\nStatus: {result.overall_status.value}")
    click.echo(f"Thread: {result.thread_id}")
    click.echo("Stages:")
    for alias, sr in result.stages.items():
        icon = (
            "✅"
            if sr.status.value == "resolved"
            else (
                "❌"
                if sr.status.value == "failed"
                else "⏭️" if sr.status.value == "skipped" else "⏳"
            )
        )
        retry_info = f" (retries: {sr.retries})" if sr.retries > 0 else ""
        click.echo(
            f"  {icon} {alias}: {sr.status.value} ({sr.duration_ms:.0f}ms, {len(sr.artifacts)} artifacts){retry_info}"
        )
        if sr.error:
            click.echo(f"      Error: {sr.error[:200]}")

    click.echo(f"\nArtifacts: {len(result.artifacts)}")
    click.echo(f"Messages:  {len(runner.get_message_log())}")

    if runner._session_ctrl:
        entries = runner._session_log
        click.echo(f"Session:   {len(entries)} work log entries")

    if not result.successful():
        click.echo(
            f"\n⚠ Pipeline failed. Resume with: session pipeline resume {result.thread_id}"
        )
        sys.exit(1)


@traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-002"])
@click.command("status")
@click.argument("thread_id")
@click.pass_context
def pipeline_status(ctx: click.Context, thread_id: str) -> None:
    """Show pipeline status and stage progress.

    \b
    Example:
        session pipeline status pipeline_abc123
    """
    # BUGFIX: this previously imported a module that has never existed
    # (``orchestrata_ultimate.orchestrator.pipeline_checkpoint``) and called a
    # ``PipelineCheckpoint.load()`` classmethod that doesn't exist on the real
    # checkpoint store either (kernel/pipeline_checkpoint.py has no ``.load()``,
    # ``.thread_id``, ``.last_completed_stage``, ``.artifacts_so_far``, or
    # ``.timestamp`` — only per-stage ``completed_stages()``/``resume_state()``
    # lookups). Every invocation therefore hit the bare except unconditionally
    # and reported "no checkpoint found", even for a thread that really was
    # checkpointed. Wired to the real store below (``_get_checkpoint_store``).
    # The real store has no notion of stage completion *order* within a
    # thread (no wall-clock ordering is tracked across stages), so "last
    # completed stage" is intentionally not reported — only the set of
    # checkpointed stages and how many have a resumable output.
    env = ctx.obj.get("env", os.environ) if ctx.obj else os.environ
    store = _get_checkpoint_store(env=env)
    stages = store.completed_stages(thread_id) if store is not None else []

    if not stages or store is None:
        click.echo(f"No checkpoint found for thread: {thread_id}")
        click.echo('Run a pipeline first: session pipeline run "<intent>"')
        sys.exit(1)

    resumable = store.resume_state(thread_id)
    click.echo(f"Thread:     {thread_id}")
    click.echo(f"Completed:  {', '.join(stages)}")
    click.echo(f"Artifacts:  {len(resumable)}")


@traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-003"])
@click.command("resume")
@click.argument("thread_id")
@click.option(
    "--work-type",
    "-w",
    default="feature",
    type=click.Choice(
        [
            "feature",
            "refactor",
            "test",
            "documentation",
            "analysis",
            "fix",
            "review",
            "spec",
            "dependency",
            "performance",
        ]
    ),
    help="Work type used to derive the remaining stage order — should "
    "match the original run's (default: feature)",
)
@click.option(
    "--has-ui",
    is_flag=True,
    help="Include the UX stage in the derived order (matches the "
    "original run's --has-ui, if any)",
)
@click.pass_context
def pipeline_resume(
    ctx: click.Context, thread_id: str, work_type: str, has_ui: bool
) -> None:
    """Resume a checkpointed pipeline from its first incomplete stage.

    \b
    Example:
        session pipeline resume pipeline_abc123
    """
    root = ctx.obj.get("root", Path.cwd())

    # W3.1 absorption (plans/decisions-2026-07-06.md D1-addendum):
    # PipelineRunner.resume() is now a REAL resume (backlog #38) instead of
    # the "re-run the whole feature pipeline" fallback this comment used to
    # document (see the matching bugfix note in pipeline_status for the
    # earlier nonexistent-module/fictional-.load() history). It reconstructs
    # completed stages from the real checkpoint store
    # (kernel/pipeline_checkpoint.py — the same store read below) and derives
    # the remaining stage order from PipelineRouter.route(work_type,
    # has_ui=...), re-entering PipelineRunner._run_stage from the first stage
    # with no checkpointed output — chaining the last checkpointed stage's
    # output in as input, exactly like a live run chains stage-to-stage. LLM
    # stages are not replayable, so already-checkpointed stages are reused
    # verbatim rather than re-invoked.
    env = ctx.obj.get("env", os.environ) if ctx.obj else os.environ
    store = _get_checkpoint_store(env=env)
    stages = store.completed_stages(thread_id) if store is not None else []

    if not stages or store is None:
        click.echo(f"No checkpoint found for thread: {thread_id}", err=True)
        sys.exit(1)

    resumable = store.resume_state(thread_id)
    runner = _get_runner(root)
    click.echo(f"Resuming pipeline: {thread_id}")
    click.echo(f"Completed stages:  {', '.join(stages)}")
    click.echo(f"Partial artifacts: {len(resumable)}")

    async def _resume() -> PipelineResult:
        return await runner.resume(thread_id, work_type=work_type, has_ui=has_ui)

    result = asyncio.run(_resume())
    click.echo(f"\nStatus: {result.overall_status.value}")


@traces(spec="000", reqs=["REQ-000-097"])
def _echo_escalation_result(thread_id: str, result: PipelineResult) -> None:
    """Shared output + exit-code handling for respond/approve/reject.

    ``EscalationInterruptMixin`` reports "no pending escalation for this
    thread" the same way run()/resume() report "nothing checkpointed": a
    FAILED result with no stages (``_no_pending_escalation_result``) — so
    the three commands below share this rendering.
    """
    click.echo(f"Thread: {thread_id}")
    click.echo(f"Status: {result.overall_status.value}")
    for alias, sr in result.stages.items():
        click.echo(f"  {alias}: {sr.status.value} ({sr.duration_ms:.0f}ms)")
    if result.overall_status.value == "failed" and not result.stages:
        click.echo("No pending escalation found for this thread.", err=True)
        sys.exit(1)


@traces(spec="000", story="US-004", reqs=["REQ-000-039", "REQ-000-040", "REQ-000-041"])
@click.command("respond")
@click.option("--thread", "thread_id", required=True, help="Thread ID to respond to.")
@click.argument("text")
@click.pass_context
def pipeline_respond(ctx: click.Context, thread_id: str, text: str) -> None:
    """Send a human response to the volk awaiting input on a thread.

    Re-enters the escalated stage mid-conversation (e.g. answering
    Concierge's clarification questions) — if the stage isn't done yet it
    escalates again for the next respond().

    \b
    Example:
        session pipeline respond --thread pipeline_abc123 "Use Postgres for storage"
    """
    root = ctx.obj.get("root", Path.cwd())
    env = ctx.obj.get("env", os.environ) if ctx.obj else os.environ
    runner = _get_runner(root, env=env)

    async def _respond() -> PipelineResult:
        return await runner.respond(thread_id, text)

    result = asyncio.run(_respond())
    _echo_escalation_result(thread_id, result)


@traces(
    spec="000",
    story="US-006",
    reqs=["REQ-000-053", "REQ-000-054", "REQ-000-057", "REQ-000-058"],
)
@click.command("approve")
@click.option("--thread", "thread_id", required=True, help="Thread ID to approve.")
@click.pass_context
def pipeline_approve(ctx: click.Context, thread_id: str) -> None:
    """Approve the pending escalation on a thread.

    The escalated stage's own output is already checkpointed — approving
    resumes the fixed sequence from the NEXT stage (``PipelineRunner.
    resume()``). Human must explicitly approve — nothing auto-approves.

    \b
    Example:
        session pipeline approve --thread pipeline_abc123
    """
    root = ctx.obj.get("root", Path.cwd())
    env = ctx.obj.get("env", os.environ) if ctx.obj else os.environ
    runner = _get_runner(root, env=env)

    async def _approve() -> PipelineResult:
        return await runner.approve(thread_id)

    result = asyncio.run(_approve())
    _echo_escalation_result(thread_id, result)


@traces(spec="000", story="US-006", reqs=["REQ-000-055", "REQ-000-056", "REQ-000-057"])
@click.command("reject")
@click.option("--thread", "thread_id", required=True, help="Thread ID to reject.")
@click.argument("reason")
@click.pass_context
def pipeline_reject(ctx: click.Context, thread_id: str, reason: str) -> None:
    """Reject the pending escalation on a thread with feedback.

    Re-enters the SAME stage with ``reason`` folded into its payload instead
    of advancing — the volk keeps working with feedback rather than the run
    continuing to the next stage.

    \b
    Example:
        session pipeline reject --thread pipeline_abc123 "Missing error handling"
    """
    root = ctx.obj.get("root", Path.cwd())
    env = ctx.obj.get("env", os.environ) if ctx.obj else os.environ
    runner = _get_runner(root, env=env)

    async def _reject() -> PipelineResult:
        return await runner.reject(thread_id, reason)

    result = asyncio.run(_reject())
    _echo_escalation_result(thread_id, result)


@traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-004"])
@click.command("list")
@click.pass_context
def pipeline_list(ctx: click.Context) -> None:
    """List recent pipeline runs from output directory.

    \b
    Example:
        session pipeline list
    """
    root = ctx.obj.get("root", Path.cwd())
    output_dir = Path(root) / "pipeline_output"

    if not output_dir.exists():
        click.echo("No pipeline runs found.")
        click.echo(f"Output directory does not exist: {output_dir}")
        return

    # ``artifacts/`` is the ArtifactStore JSONL layer (wired by _get_runner),
    # a reserved sibling of the per-thread run directories — never a run.
    runs = sorted(
        (p for p in output_dir.iterdir() if p.name != "artifacts"),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    if not runs:
        click.echo("No pipeline runs found.")
        return

    click.echo(f"Recent pipeline runs ({len(runs)}):")
    for run_dir in runs[:20]:
        if run_dir.is_dir():
            stage_dirs = [d for d in run_dir.iterdir() if d.is_dir()]
            artifact_count = sum(1 for d in stage_dirs for _ in d.iterdir())
            click.echo(
                f"  {run_dir.name:30s}  {len(stage_dirs)} stages  {artifact_count} artifacts"
            )