feat(pipeline): add multi-step agent pipeline execution

src/cli/commands/pipeline-command.test.ts

@@ -0,0 +1,212 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { Volume, createFsFromVolume } from "memfs";
+import path from "node:path";
+import type { FileSystem } from "../utils/file-system.js";
+
+const renderAcpStreamMock = vi.hoisted(
+  () =>
+    vi.fn(async (events: AsyncIterable<unknown>) => {
+      for await (const ignoredEvent of events) {
+        // noop
+      }
+    })
+);
+
+vi.mock("@poe-code/agent-spawn", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@poe-code/agent-spawn")>();
+  return {
+    ...actual,
+    renderAcpStream: renderAcpStreamMock
+  };
+});
+
+vi.mock("../../sdk/spawn.js", () => ({
+  spawn: vi.fn()
+}));
+
+import { createProgram } from "../program.js";
+import { spawn as sdkSpawn } from "../../sdk/spawn.js";
+
+const cwd = "/repo";
+const homeDir = "/home/test";
+
+function createMemFs(files?: Record<string, string>): FileSystem {
+  const vol = new Volume();
+  vol.mkdirSync(homeDir, { recursive: true });
+  vol.mkdirSync(cwd, { recursive: true });
+  if (files) {
+    for (const [filePath, content] of Object.entries(files)) {
+      const dir = path.dirname(filePath);
+      vol.mkdirSync(dir, { recursive: true });
+      vol.writeFileSync(filePath, content, "utf8");
+    }
+  }
+  return createFsFromVolume(vol).promises as unknown as FileSystem;
+}
+
+function mockSpawnSuccess(output: string) {
+  return {
+    events: (async function* () {
+      yield { event: "agent_message", text: output };
+    })(),
+    result: Promise.resolve({
+      stdout: output,
+      stderr: "",
+      exitCode: 0
+    })
+  };
+}
+
+const validPipeline = `
+name: test-pipeline
+steps:
+  - name: review
+    agent: claude-code
+    prompt: Review this code
+`;
+
+const invalidPipeline = `
+steps:
+  - name: review
+    agent: claude-code
+    prompt: Review
+`;
+
+const pipelinePath = "/repo/pipeline.yaml";
+
+describe("pipeline command", () => {
+  const originalEnv = { ...process.env };
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    process.env = { ...originalEnv, FORCE_COLOR: "1" };
+  });
+
+  afterEach(() => {
+    process.env = { ...originalEnv };
+  });
+
+  describe("pipeline validate", () => {
+    it("reports valid pipeline", async () => {
+      const fs = createMemFs({ [pipelinePath]: validPipeline });
+      const program = createProgram({
+        fs,
+        prompts: vi.fn().mockResolvedValue({}),
+        env: { cwd, homeDir },
+        logger: () => {},
+        suppressCommanderOutput: true
+      });
+
+      await program.parseAsync([
+        "node",
+        "cli",
+        "pipeline",
+        "validate",
+        pipelinePath
+      ]);
+    });
+
+    it("throws on invalid pipeline", async () => {
+      const fs = createMemFs({ [pipelinePath]: invalidPipeline });
+      const program = createProgram({
+        fs,
+        prompts: vi.fn().mockResolvedValue({}),
+        env: { cwd, homeDir },
+        logger: () => {},
+        suppressCommanderOutput: true
+      });
+
+      await expect(
+        program.parseAsync([
+          "node",
+          "cli",
+          "pipeline",
+          "validate",
+          pipelinePath
+        ])
+      ).rejects.toThrow("name");
+    });
+  });
+
+  describe("pipeline run", () => {
+    it("runs a pipeline and spawns agents", async () => {
+      vi.mocked(sdkSpawn).mockReturnValue(mockSpawnSuccess("review output"));
+
+      const fs = createMemFs({ [pipelinePath]: validPipeline });
+      const program = createProgram({
+        fs,
+        prompts: vi.fn().mockResolvedValue({}),
+        env: { cwd, homeDir },
+        logger: () => {},
+        suppressCommanderOutput: true
+      });
+
+      await program.parseAsync([
+        "node",
+        "cli",
+        "pipeline",
+        "run",
+        pipelinePath
+      ]);
+
+      expect(sdkSpawn).toHaveBeenCalledWith("claude-code", expect.objectContaining({
+        prompt: "Review this code",
+        mode: "yolo"
+      }));
+    });
+
+    it("shows dry run without spawning", async () => {
+      const fs = createMemFs({ [pipelinePath]: validPipeline });
+      const program = createProgram({
+        fs,
+        prompts: vi.fn().mockResolvedValue({}),
+        env: { cwd, homeDir },
+        logger: () => {},
+        suppressCommanderOutput: true
+      });
+
+      await program.parseAsync([
+        "node",
+        "cli",
+        "--dry-run",
+        "pipeline",
+        "run",
+        pipelinePath
+      ]);
+
+      expect(sdkSpawn).not.toHaveBeenCalled();
+    });
+
+    it("sets exit code on failure", async () => {
+      vi.mocked(sdkSpawn).mockReturnValue({
+        events: (async function* () {})(),
+        result: Promise.resolve({
+          stdout: "",
+          stderr: "error",
+          exitCode: 1
+        })
+      });
+
+      const fs = createMemFs({ [pipelinePath]: validPipeline });
+      const program = createProgram({
+        fs,
+        prompts: vi.fn().mockResolvedValue({}),
+        env: { cwd, homeDir },
+        logger: () => {},
+        suppressCommanderOutput: true
+      });
+
+      const originalExitCode = process.exitCode;
+      await program.parseAsync([
+        "node",
+        "cli",
+        "pipeline",
+        "run",
+        pipelinePath
+      ]);
+
+      expect(process.exitCode).toBe(1);
+      process.exitCode = originalExitCode;
+    });
+  });
+});

src/cli/commands/pipeline.ts

@@ -0,0 +1,127 @@
+import path from "node:path";
+import type { Command } from "commander";
+import type { CliContainer } from "../container.js";
+import { text } from "@poe-code/design-system";
+import {
+  resolveCommandFlags,
+  createExecutionResources
+} from "./shared.js";
+import { parsePipeline } from "../../sdk/pipeline/parse.js";
+import { validatePipeline } from "../../sdk/pipeline/validate.js";
+import { runPipeline } from "../../sdk/pipeline/run.js";
+import { isParallelGroup } from "../../sdk/pipeline/types.js";
+import type { PipelineDefinition } from "../../sdk/pipeline/types.js";
+
+export function registerPipelineCommand(
+  program: Command,
+  container: CliContainer
+): void {
+  const pipelineCmd = program
+    .command("pipeline")
+    .description("Run multi-step agent pipelines.");
+
+  pipelineCmd
+    .command("run")
+    .description("Run a pipeline from a YAML file.")
+    .argument("<file>", "Path to pipeline YAML file")
+    .option("-C, --cwd <path>", "Override working directory for all steps")
+    .action(async function (this: Command, file: string) {
+      const flags = resolveCommandFlags(program);
+      const commandOptions = this.opts<{ cwd?: string }>();
+
+      const filePath = path.resolve(file);
+      const yamlContent = await container.fs.readFile(filePath, "utf8");
+      const pipeline = parsePipeline(yamlContent);
+
+      const cwd = commandOptions.cwd
+        ? path.resolve(commandOptions.cwd)
+        : process.cwd();
+
+      if (flags.dryRun) {
+        renderDryRun(pipeline);
+        return;
+      }
+
+      const resources = createExecutionResources(
+        container,
+        flags,
+        "pipeline"
+      );
+      resources.logger.intro(`pipeline ${pipeline.name}`);
+
+      const result = await runPipeline(pipeline, { cwd });
+
+      if (result.summary.success) {
+        const duration = formatDuration(result.summary.totalDuration);
+        resources.logger.info(
+          `Pipeline completed: ${result.summary.completedSteps} steps (${duration})`
+        );
+      } else {
+        const duration = formatDuration(result.summary.totalDuration);
+        resources.logger.info(
+          `Pipeline aborted (${result.summary.completedSteps}/${result.summary.totalSteps} steps completed, ${duration})`
+        );
+        process.exitCode = 1;
+      }
+
+      resources.context.finalize();
+    });
+
+  pipelineCmd
+    .command("validate")
+    .description("Validate a pipeline YAML file without running it.")
+    .argument("<file>", "Path to pipeline YAML file")
+    .action(async function (this: Command, file: string) {
+      const filePath = path.resolve(file);
+      const yamlContent = await container.fs.readFile(filePath, "utf8");
+      const pipeline = parsePipeline(yamlContent);
+      validatePipeline(pipeline);
+
+      const resources = createExecutionResources(
+        container,
+        resolveCommandFlags(program),
+        "pipeline"
+      );
+      resources.logger.info(`Pipeline "${pipeline.name}" is valid.`);
+      resources.context.finalize();
+    });
+}
+
+function renderDryRun(pipeline: PipelineDefinition): void {
+  const lines: string[] = [
+    text.heading(`Pipeline: ${pipeline.name}`)
+  ];
+  if (pipeline.description) {
+    lines.push(`  ${pipeline.description}`);
+  }
+  lines.push("");
+
+  let stepIndex = 1;
+  for (const entry of pipeline.steps) {
+    if (isParallelGroup(entry)) {
+      const names = entry.parallel
+        .map((s) => {
+          const agent = s.agent ?? pipeline.defaults?.agent ?? "?";
+          const mode = s.mode ?? pipeline.defaults?.mode ?? "yolo";
+          return `${s.name} (${agent} · ${mode})`;
+        })
+        .join("  +  ");
+      lines.push(`  ${stepIndex}. ${text.muted("[parallel]")} ${names}`);
+      stepIndex += entry.parallel.length;
+    } else {
+      const agent = entry.agent ?? pipeline.defaults?.agent ?? "?";
+      const mode = entry.mode ?? pipeline.defaults?.mode ?? "yolo";
+      lines.push(`  ${stepIndex}. ${text.command(entry.name)} ${text.muted(`(${agent} · ${mode})`)}`);
+      stepIndex += 1;
+    }
+  }
+
+  process.stdout.write(lines.join("\n") + "\n");
+}
+
+function formatDuration(ms: number): string {
+  if (ms < 1000) {
+    return `${ms}ms`;
+  }
+  return `${(ms / 1000).toFixed(1)}s`;
+}

src/cli/program.ts

@@ -23,6 +23,7 @@ import { registerVersionOption } from "./commands/version.js";
 import { registerRalphCommand } from "./commands/ralph.js";
 import { registerUsageCommand } from "./commands/usage.js";
 import { registerModelsCommand } from "./commands/models.js";
+import { registerPipelineCommand } from "./commands/pipeline.js";
 import packageJson from "../../package.json" with { type: "json" };
 import { throwCommandNotFound } from "./command-not-found.js";
 import {
@@ -127,6 +128,16 @@ function formatHelpText(input: {
         name: "usage list",
         args: "",
         description: "Display usage history"
+      },
+      {
+        name: "pipeline run",
+        args: "<file>",
+        description: "Run a multi-step agent pipeline"
+      },
+      {
+        name: "pipeline validate",
+        args: "<file>",
+        description: "Validate a pipeline YAML file"
       }
     ];
   const nameWidth = Math.max(0, ...commandRows.map((row) => row.name.length));
@@ -320,6 +331,7 @@ function bootstrapProgram(container: CliContainer): Command {
   registerRalphCommand(program, container);
   registerUsageCommand(program, container);
   registerModelsCommand(program, container);
+  registerPipelineCommand(program, container);
 
   program.allowExcessArguments().action(function (this: Command) {
     const args = this.args;

src/index.ts

@@ -14,6 +14,16 @@ export type {
   GenerateResult,
   MediaGenerateResult
 } from "./sdk/types.js";
+export { runPipeline } from "./sdk/pipeline/index.js";
+export { parsePipeline } from "./sdk/pipeline/index.js";
+export type {
+  PipelineDefinition,
+  PipelineStep,
+  PipelineResult,
+  PipelineStepResult,
+  PipelineSummary,
+  RunPipelineOptions
+} from "./sdk/pipeline/index.js";
 
 async function main(): Promise<void> {
   const [{ createProgram }, { createCliMain }] = await Promise.all([