Orchestrata Ultimate › Plans › pipeline-automation-plan

Pipeline Automation — Detailed Implementation Plan

Source: plans/pipeline-automation-plan.md at 8a2cb42ad · last changed 2026-08-19 · first committed 2026-06-18 · 746 lines.

Prose pending.

Plan text

Pipeline Automation — Detailed Implementation Plan

Date: 2026-06-25 Specs: SPEC-003 (Intent), SPEC-005 (Multi-Agent), SPEC-016 (VizInt), SPEC-032 (Concierge) Standards: SPEC-097 (Testing), SPEC-100 (Logging) Architecture: Layer 3 (domain services) + Layer 4 (orchestration) Session: Deferred task #6 — critical priority

1. Architecture: Layer Placement

New code in Layer 4 (Orchestration) — coordinates Layer 3 agents. May import concrete classes from Layer 3. Never imports from Layer 5.

Layer 4: Orchestration (8 NEW files)
  orchestrator/pipeline_runner.py     PipelineRunner class
  orchestrator/pipeline_stage.py      PipelineStage, StageResult, StageStatus
  orchestrator/pipeline_result.py     PipelineResult, PipelineMetrics
  orchestrator/artifact_store.py      ArtifactStore, PipelineArtifact, ArtifactType
  orchestrator/pipeline_errors.py     PipelineError hierarchy (5 types)
  orchestrator/pipeline_logger.py     PipelineLogger (structlog)
  orchestrator/pipeline_checkpoint.py  PipelineCheckpoint (Redis-backed)
  orchestrator/pipeline_cli.py        CLI entry

Layer 3: Domain Services (7 MODIFIED)
  agents/volks/concierge/volk.py      Add structured logging
  agents/volks/analyst/agent.py       Add structured logging
  agents/volks/pm/volk.py             Add structured logging
  agents/volks/architect/volk.py      Add structured logging
  agents/volks/codedirector/volk.py   Add structured logging
  agents/volks/qa/volk.py             Add structured logging
  agents/volks/compliance/volk.py     Add structured logging

Layer 2: Communication (1 MODIFIED)
  gateway/handlers/orchestrator.py    Route intent → Concierge via CXP

Layer 4: Session (1 MODIFIED)
  session/session.py                  Add log_pipeline_stage(), log_pipeline_error()

Protocol Boundaries

PipelineRunner communicates with agents via CXP packets — no direct method calls:

# CORRECT — through CXP protocol
packet = CXPPacket(from_agent="U:Ox", to_agent="U:Cx", ...)
response = await agent.process(packet)

# INCORRECT — bypassing CXP
result = concierge._handle_intent(raw_text)

Decision: Phase 1 uses direct instantiation (agents are in-process). Phase 1b will switch to Postman/CXP routing when Postman is verified. This technical debt is documented.

Dependency Injection

All new classes receive dependencies via constructor:

class PipelineRunner:
    def __init__(self, artifact_store, session_controller, pipeline_logger, ...):
        self.artifact_store = artifact_store
        self.session_controller = session_controller
        ...

No global singletons. No module-level state. Everything injectable for testability.

2. Data Models

2.1 PipelineStage & StageResult

# File: orchestrator/pipeline_stage.py
# @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001-003"])

@dataclass
class PipelineStage:
    alias: str                    # "U:Cx", "U:BA", etc.
    agent_class: type             # ConciergeVolk, AnalystAgent, ...
    input_subject: str            # "intent", "analyze_requirements", ...
    output_subject: str           # "analyze_requirements", "scope_stories", ...
    timeout_seconds: int          # Max time for this stage
    required: bool = True         # If False, skip on failure
    ui_conditional: bool = False  # Only run if UI keywords detected
    artifact_types: list[str]     # ["requirements", "prolog_facts", ...]

@dataclass
class StageResult:
    stage_alias: str
    status: StageStatus           # PENDING|IN_PROGRESS|RESOLVED|FAILED|TIMED_OUT|SKIPPED
    output_packet: CXPPacket | None
    artifacts: list
    duration_ms: float
    error: PipelineError | None
    llm_calls: int = 0

class StageStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"
    FAILED = "failed"
    TIMED_OUT = "timed_out"
    SKIPPED = "skipped"
    GOVERNANCE_DENIED = "governance_denied"

2.2 PipelineResult

# File: orchestrator/pipeline_result.py
# @traces(spec="003", story="US-005", reqs=["REQ-003-US-005-002-007"])

@dataclass
class PipelineResult:
    thread_id: str
    intent: str
    stages: dict[str, StageResult]
    artifacts: list
    metrics: PipelineMetrics
    started_at: datetime
    completed_at: datetime | None
    overall_status: StageStatus

    def successful(self) -> bool: ...
    def failed_stages(self) -> list[str]: ...
    def to_dict(self) -> dict: ...

2.3 PipelineMetrics

# File: orchestrator/pipeline_metrics.py
# @traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001-002"])

@dataclass
class PipelineMetrics:
    stages_completed: int = 0
    stages_failed: int = 0
    stages_skipped: int = 0
    total_duration_ms: float = 0.0
    per_stage_duration: dict[str, float] = {}
    llm_calls_total: int = 0
    prolog_checks_total: int = 0
    governance_violations: int = 0
    artifacts_generated: int = 0
    retries_total: int = 0

2.4 PipelineArtifact & ArtifactType

# File: orchestrator/artifact_store.py
# @traces(spec="003", story="US-006", reqs=["REQ-003-US-006-001-006"])

@dataclass
class PipelineArtifact:
    stage_alias: str
    artifact_type: ArtifactType
    content: str
    content_format: ArtifactFormat
    file_path: str | None = None
    timestamp: datetime = _utc_now()
    thread_id: str = ""
    session_id: str = ""

class ArtifactType(Enum):
    REQUIREMENTS = "requirements"
    PROLOG_FACTS = "prolog_facts"
    TASK_SUMMARY = "task_summary"
    USER_STORIES = "user_stories"
    REQUIREMENTS_ANALYSIS = "requirements_analysis"
    PROJECT_META = "project_meta"
    SCOPE_ESTIMATE = "scope_estimate"
    WARNINGS = "warnings"
    ARCHITECTURE_DESIGN = "architecture_design"
    FILE_TREE = "file_tree"
    API_SPEC = "api_spec"
    CLASS_DIAGRAM = "class_diagram"
    SOURCE_CODE = "source_code"
    TEST_CODE = "test_code"
    IMPLEMENTATION_PLAN = "implementation_plan"
    QA_REPORT = "qa_report"
    RTM = "rtm"
    COMPLIANCE_REPORT = "compliance_report"
    ERROR_LOG = "error_log"

class ArtifactFormat(Enum):
    TEXT = "text"
    JSON = "json"
    MERMAID = "mermaid"
    PYTHON = "python"
    PROLOG = "prolog"
    MARKDOWN = "markdown"

2.5 PipelineError Hierarchy

# File: orchestrator/pipeline_errors.py
# @traces(spec="003", story="US-005", reqs=["REQ-003-US-005-010-015"])

class PipelineError(Exception):
    def __init__(self, stage_alias, thread_id, message,
                 original_error=None, partial_artifacts=None, conversation_history=None):
        self.stage_alias = stage_alias
        self.thread_id = thread_id
        self.original_error = original_error
        self.partial_artifacts = partial_artifacts or []
        ...

class StageTimeoutError(PipelineError): ...
class GovernanceRejectionError(PipelineError): ...
class LLMCallFailedError(PipelineError): ...
class AgentReturnedNoneError(PipelineError): ...
class InfrastructureError(PipelineError): ...

3. Phase 1: PipelineRunner (~700 LOC)

File: src/orchestrata_ultimate/orchestrator/pipeline_runner.py

Stage Definitions (Fixed Pipeline)

STAGES = [
    PipelineStage("U:Cx", ConciergeVolk, "intent", "analyze_requirements", 300,
                  artifact_types=["requirements", "prolog_facts", "task_summary"]),
    PipelineStage("U:BA", AnalystAgent, "analyze_requirements", "scope_stories", 300,
                  artifact_types=["user_stories", "requirements_analysis", "project_meta"]),
    PipelineStage("U:PM", PMVolk, "scope_stories", "design_architecture", 120,
                  artifact_types=["scope_estimate", "warnings"]),
    PipelineStage("U:Ar", ArchitectVolk, "design_architecture", "code_implementation", 300,
                  artifact_types=["architecture_design", "file_tree", "api_spec", "class_diagram"]),
    PipelineStage("U:Cd", CodeDirectorVolk, "code_implementation", "review_code", 600,
                  artifact_types=["source_code", "implementation_plan"]),
    PipelineStage("U:Qa", QaVolk, "review_code", "compliance_check", 600,
                  artifact_types=["test_code", "qa_report"]),
    PipelineStage("U:Cp", ComplianceVolk, "compliance_check", None, 120,
                  artifact_types=["rtm", "compliance_report"]),
    PipelineStage("U:Ux", UXVolk, "design_ux", None, 180,
                  required=False, ui_conditional=True,
                  artifact_types=["ux_design"]),
]

Core Method

async def run(self, intent: str, thread_id: str = None,
              has_ui: bool = False) -> PipelineResult:
    """Execute full pipeline: Cx→BA→PM→Ar→Cd→QA→Cp."""
    thread_id = thread_id or f"pipeline_{uuid.uuid4().hex[:8]}"
    started_at = datetime.now(UTC)
    self.pipeline_logger.pipeline_start(thread_id, intent, len(self.STAGES))

    metrics = PipelineMetrics()
    stages_result = {}
    all_artifacts = []
    current_packet = None

    for stage in self.STAGES:
        if stage.ui_conditional and not has_ui:
            stages_result[stage.alias] = StageResult(status=StageStatus.SKIPPED, ...)
            metrics.stages_skipped += 1
            continue

        try:
            result = await self._run_stage(stage, current_packet, thread_id, intent)
            stages_result[stage.alias] = result
            all_artifacts.extend(result.artifacts)
            metrics.stages_completed += 1
            current_packet = result.output_packet
            self.session_controller.log_work(
                desc=f"Pipeline: {stage.alias} completed ({result.duration_ms:.0f}ms)",
                details=[f"Status: {result.status.value}"]
            )
        except PipelineError as e:
            stages_result[stage.alias] = StageResult(status=StageStatus.FAILED, error=e, ...)
            metrics.stages_failed += 1
            if stage.required:
                break

    completed_at = datetime.now(UTC)
    metrics.total_duration_ms = (completed_at - started_at).total_seconds() * 1000
    metrics.artifacts_generated = len(all_artifacts)

    result = PipelineResult(thread_id=thread_id, intent=intent, stages=stages_result,
                            artifacts=all_artifacts, metrics=metrics, ...)
    self.pipeline_logger.pipeline_complete(thread_id, result)
    return result

Artifact Extraction

def _extract_artifacts(self, stage, packet, thread_id) -> list[PipelineArtifact]:
    """Map payload fields to artifact types."""
    mapping = {
        "requirements": packet.payload.get("requirements"),
        "prolog_facts": "\n".join(packet.payload.get("prolog_facts", [])),
        "task_summary": packet.payload.get("task_summary"),
        "user_stories": packet.payload.get("user_stories"),
        "scope_estimate": str(packet.payload.get("estimate", "")),
        "architecture_design": packet.payload.get("architecture"),
        "file_tree": packet.payload.get("file_tree"),
        "api_spec": packet.payload.get("api_spec"),
        "class_diagram": packet.payload.get("class_diagram"),
        "source_code": str(packet.payload.get("files_created", [])),
        "qa_report": packet.payload.get("review_report"),
        "rtm": str(packet.payload.get("rtm", "")),
        "compliance_report": packet.payload.get("compliance_report"),
    }
    artifacts = []
    for at_str in stage.artifact_types:
        content = mapping.get(at_str)
        if content:
            artifact = PipelineArtifact(stage_alias=stage.alias,
                                        artifact_type=ArtifactType(at_str),
                                        content=content, thread_id=thread_id)
            artifacts.append(artifact)
            self.artifact_store.add(artifact)
    return artifacts

Verification Gate

python -c "from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineRunner"
python scripts/architecture_checker.py --strict
grep -c "@traces" src/orchestrata_ultimate/orchestrator/pipeline_runner.py  # ≥ 8
grep -rn 'MagicMock\|Mock\|patch(' src/orchestrata_ultimate/orchestrator/pipeline_runner.py  # 0

4. Phase 2: ArtifactStore (~350 LOC)

File: src/orchestrata_ultimate/orchestrator/artifact_store.py

class ArtifactStore:
    """Thread-safe artifact collection with disk persistence + Neo4j sync."""

    def __init__(self, output_dir: Path = None):
        self._lock = threading.Lock()
        self._artifacts: list[PipelineArtifact] = []
        self._output_dir = output_dir or Path("pipeline_output")

    def add(self, artifact: PipelineArtifact) -> None:
        with self._lock:
            self._artifacts.append(artifact)
        self._write_to_disk(artifact)

    def by_stage(self, stage_alias: str) -> list[PipelineArtifact]:
        with self._lock:
            return [a for a in self._artifacts if a.stage_alias == stage_alias]

    def by_type(self, artifact_type: ArtifactType) -> list[PipelineArtifact]:
        with self._lock:
            return [a for a in self._artifacts if a.artifact_type == artifact_type]

    def all(self) -> list[PipelineArtifact]:
        with self._lock:
            return list(self._artifacts)

    def by_thread(self, thread_id: str) -> list[PipelineArtifact]:
        with self._lock:
            return [a for a in self._artifacts if a.thread_id == thread_id]

    def _write_to_disk(self, artifact: PipelineArtifact) -> None:
        dir_path = self._output_dir / artifact.thread_id / artifact.stage_alias
        dir_path.mkdir(parents=True, exist_ok=True)
        ext = _format_to_extension(artifact.content_format)
        file_path = dir_path / f"{artifact.artifact_type.value}{ext}"
        file_path.write_text(artifact.content, encoding="utf-8")
        artifact.file_path = str(file_path)

    def to_json(self) -> str:
        with self._lock:
            return json.dumps([...], indent=2, default=str)

Expected Artifacts Per Stage

5. Phase 3: Gateway Bridge (~100 LOC modified)

File: src/orchestrata_ultimate/gateway/handlers/orchestrator.py (MODIFY)

Current behavior: handle_intent() parses text → creates Neo4j task → returns. Never dispatches to Concierge.

New behavior:

async def handle_intent(self, connection_id, params):
    text = params.get("text", "")
    intent_id = f"intent_{uuid.uuid4().hex[:8]}"

    # 1. Create task (existing behavior — keep)
    task = await self._create_task(connection_id, text)

    # 2. NEW: Dispatch to Concierge via CXP
    cxp_result = await self._cxp_handler.handle_send(connection_id, {
        "target": "U:Cx",
        "payload": {"text": text, "thread_id": intent_id, "subject": "intent"},
    })

    # 3. NEW: Subscribe UI to pipeline progress
    subscription_id = await self._pubsub.subscribe(
        connection_id, f"pipeline.{intent_id}", {})

    return {
        "intent_id": intent_id, "task": task,
        "pipeline_thread_id": intent_id, "pipeline_started": True,
        "subscription_id": subscription_id,
    }

6. Phase 4: Observability (~250 LOC + 7 agents modified)

PipelineLogger

# File: orchestrator/pipeline_logger.py
# @traces(spec="100", story="US-100-001", reqs=["REQ-100-001-007"])

class PipelineLogger:
    def __init__(self, logger: logging.Logger):
        self._log = logger

    def stage_start(self, stage, thread_id, intent_summary):
        self._log.info("[PIPELINE] Stage starting", extra={
            "pipeline_stage": stage, "thread_id": thread_id,
            "intent_summary": intent_summary[:200],
            "event": "pipeline.stage.start"})

    def stage_complete(self, stage, thread_id, duration_ms, artifact_count):
        self._log.info("[PIPELINE] Stage complete", extra={
            "pipeline_stage": stage, "thread_id": thread_id,
            "duration_ms": duration_ms, "artifact_count": artifact_count,
            "event": "pipeline.stage.complete"})

    def stage_failed(self, stage, thread_id, error):
        self._log.error("[PIPELINE] Stage failed", extra={
            "pipeline_stage": stage, "thread_id": thread_id,
            "error_type": type(error).__name__,
            "error_message": str(error)[:500],
            "event": "pipeline.stage.failed"})

    def pipeline_start(self, thread_id, intent, stage_count): ...
    def pipeline_complete(self, thread_id, result): ...

Agent Logging Pattern

Add to each agent's process():

logger.info("[PIPELINE] %s received packet", self.alias, extra={
    "stage": self.alias, "thread_id": packet.thread_id,
    "subject": packet.subject, "event": "pipeline.agent.received"})

# ... existing process logic ...

logger.info("[PIPELINE] %s completed — escalating", self.alias, extra={
    "stage": self.alias, "thread_id": packet.thread_id,
    "status": self.status.value, "event": "pipeline.agent.completed"})

Gateway PubSub for UI Updates

# After each stage completes:
if self._pubsub:
    await self._pubsub.publish(f"pipeline.{thread_id}", {
        "event": "stage.complete", "stage": stage.alias,
        "status": "resolved", "duration_ms": duration_ms,
        "artifact_count": len(artifacts),
        "next_stage": self._next_stage_alias(stage),
    })

7. Phase 5: Error Handling & Recovery (~400 LOC)

Retry Strategy

MAX_RETRIES = 3
RETRY_BACKOFF = [5, 10, 20]  # seconds

for attempt in range(MAX_RETRIES + 1):
    try:
        output_packet = await asyncio.wait_for(agent.process(packet), stage.timeout_seconds)
        break
    except asyncio.TimeoutError:
        if attempt < MAX_RETRIES:
            await asyncio.sleep(RETRY_BACKOFF[attempt])
        else:
            raise StageTimeoutError(stage_alias=stage.alias, ...)
    except Exception as e:
        if attempt < MAX_RETRIES and _is_retryable(e):
            await asyncio.sleep(RETRY_BACKOFF[attempt])
        else:
            raise

Checkpointing

# File: orchestrator/pipeline_checkpoint.py

@dataclass
class PipelineCheckpoint:
    thread_id: str
    last_completed_stage: str
    completed_stages: list[str]
    artifacts_so_far: list
    conversation_history: list[str]
    timestamp: datetime

    def save(self, redis_client) -> None:
        key = f"pipeline:checkpoint:{self.thread_id}"
        redis_client.setex(key, 86400, json.dumps(self.to_dict()))

    @staticmethod
    def load(redis_client, thread_id: str) -> "PipelineCheckpoint | None":
        data = redis_client.get(f"pipeline:checkpoint:{thread_id}")
        return PipelineCheckpoint.from_dict(json.loads(data)) if data else None

Error Handling Per Type

8. Phase 6: Session Integration (~100 LOC modified)

File: src/orchestrata_ultimate/session/session.py (MODIFY)

def log_pipeline_stage(self, stage_alias, status, artifact_count, duration_ms):
    self.log_work(
        desc=f"Pipeline: {stage_alias} — {status} "
             f"({artifact_count} artifacts, {duration_ms:.0f}ms)",
        details=[f"Stage: {stage_alias}", f"Status: {status}",
                 f"Artifacts: {artifact_count}", f"Duration: {duration_ms:.0f}ms"])

def log_pipeline_error(self, stage_alias, error_type, error_message):
    self.log_work(
        desc=f"Pipeline: {stage_alias} — FAILED ({error_type})",
        details=[f"Error: {error_message}"])

9. Phase 7: E2E Testing (~600 LOC)

File: tests/orchestrator/test_pipeline_e2e.py

Test Standards Compliance

All tests follow SPEC-097 + AGENTS.md:

• T1: No mocking — real agents, real LLM calls
• T2: No os.environ mutations
• T3: Function-scope fixtures only
• T4: No global mutable state
• T5: No hardcoded ports
• T7: test_id for all DB ops
• T9: Per-test Prolog engine
• No @pytest.mark.skip, no FakeX/StubX/DummyX

Test Cases

@pytest.mark.asyncio
async def test_full_pipeline_simple_intent(runner, test_id):
    intent = "Create a Python CLI tool that reads CSV and outputs JSON"
    result = await runner.run(intent, thread_id=test_id)

    assert result.overall_status == StageStatus.RESOLVED
    assert result.metrics.stages_completed == 7
    assert result.metrics.stages_failed == 0
    assert result.metrics.artifacts_generated >= 5
    assert os.path.isdir(f"/tmp/pipeline_test_{test_id}/{test_id}")

Verification Gate

uv run pytest tests/orchestrator/test_pipeline_e2e.py -x -v --timeout=600
# Purity checks
grep -rn 'MagicMock\|Mock\|Fake\|Stub\|Dummy\|skip' tests/orchestrator/test_pipeline_e2e.py  # Expect: 0
python scripts/test_isolation_audit.py tests/orchestrator/
uv run pytest tests/orchestrator/ -n auto -x --timeout=120

10. Phase 8: CLI Command (~200 LOC)

File: src/orchestrata_ultimate/session/cli.py (MODIFY)

session pipeline run "Create a Redis task queue with REST API"
session pipeline run "Build an async event bus" --has-ui
session pipeline status <thread_id>
session pipeline resume <thread_id>
session pipeline list

11. Spec Traceability Matrix

Requirements Referenced Per Spec

@traces Decorator Pattern

Every public class, method, and function must have:

@traces(
    spec="003",          # Spec ID (number only)
    story="US-004",      # User story
    reqs=[               # Requirement IDs
        "REQ-003-US-004-001",
        "REQ-003-US-004-002",
    ],
)

RTM Update

After implementation, regenerate the requirements traceability matrix:

python scripts/generate_comprehensive_audit.py \
    --specs specs/ \
    --source src/orchestrata_ultimate/orchestrator/

12. Implementation Order

Phase 1: PipelineRunner (4 new files)      ← Must be first
Phase 2: ArtifactStore (1 new file)        ← Needed by Phase 1
Phase 5: Error Handling (2 new files)      ← Recovery depends on Phase 1
Phase 4: Observability (1 new + 7 mod)     ← Logging during Phase 1
Phase 6: Session Integration (1 mod)       ← Wraps Phase 1
Phase 3: Gateway Bridge (1 mod)            ← Connects UI to Phase 1
Phase 8: CLI Command (1 mod)               ← User-facing entry
Phase 7: E2E Testing (1 new + 6 support)   ← Validates all phases

Effort Estimate

Success Criteria

1. session pipeline run "intent" → 7 stages execute automatically
2. Every stage appears in CXPFeed gadget in real-time
3. 17 artifact types collected to pipeline_output/<thread_id>/
4. Errors surface with full context (stage, type, partial output)
5. Failed pipelines resume from Redis checkpoint
6. Session work log shows one entry per completed stage
7. E2E test passes: intent → 7 stages → 5+ artifacts → session entries
8. Architecture checker: 0 violations
9. Test purity: 0 mocks, 0 stubs, 0 fakes, 0 skips
10. All 78 SPEC requirements traced via @traces decorators

References: SPEC-003, SPEC-005, SPEC-016, SPEC-032, SPEC-006, SPEC-097, SPEC-100, AGENTS.md § Test Standards, architecture/layer-dependency-rules.md ENDOFPLAN

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks