boocode/apps/coder/src/services/paseo-client.ts

/**
 * v2.10 — PaseoClient: thin CLI-based client for the Paseo daemon.
 *
 * Paseo is a multi-agent hub daemon running at a configurable address
 * (default Unix socket / localhost:6767). This client wraps the `paseo` CLI
 * via child_process spawn for all operations (the daemon does not expose a
 * separate REST API for write operations). Read operations (listAgents,
 * getAgentStatus) use `paseo ls --json` / `paseo inspect --json`; write
 * operations (import, archive, send) use the corresponding subcommands.
 *
 * Spec: openspec/changes/v2-10-paseo-integration/design.md.
 */
import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { createInterface } from 'node:readline';

// ─── Types ───────────────────────────────────────────────────────────────────

/** Listing entry from `paseo ls --json`. Fields are lowercase. */
export interface PaseoAgentListItem {
  id: string;
  shortId: string;
  name: string;
  provider: string;
  status: string;
  cwd?: string;
  created?: string;
  thinking?: string;
}

/** Detailed agent info from `paseo inspect --json`. Fields are PascalCase. */
export interface PaseoAgentDetail {
  Id: string;
  Name: string;
  Provider: string;
  Model?: string;
  Status: string;
  Thinking?: string;
  Archived: boolean;
  ArchivedAt?: string | null;
  Cwd?: string;
  CreatedAt: string;
  UpdatedAt: string;
  Mode?: string;
  AvailableModes?: Array<{ id: string; label: string }>;
  Capabilities?: {
    Streaming?: boolean;
    Persistence?: boolean;
    DynamicModes?: boolean;
    McpServers?: boolean;
  };
  Labels?: Record<string, string>;
  Worktree?: string | null;
  ParentAgentId?: string | null;
}

/** Result of `paseo send --json`. */
export interface PaseoSendResult {
  /** The agent's textual response. */
  text?: string;
  /** Structured output if the agent produced any. */
  output?: unknown;
  /** Error message if the turn failed. */
  error?: string;
  /** True if the turn completed successfully. */
  ok?: boolean;
}

export interface PaseoClientConfig {
  /** Path to the paseo binary. Default: auto-resolved from PATH. */
  paseoBin: string;
  /**
   * Explicit `--host <host>` value for CLI calls.
   * Format: `host:port` or `tcp://host:port?ssl=true&password=secret`.
   * Omit to use the CLI default (Unix socket, fallback localhost:6767).
   */
  cliHost?: string;
}

const DEFAULT_PASEO_BIN = 'paseo';

// ─── Client ──────────────────────────────────────────────────────────────────

export class PaseoClientError extends Error {
  constructor(
    message: string,
    public readonly command: string,
    public readonly exitCode: number | null,
    public readonly stderr: string,
  ) {
    super(message);
    this.name = 'PaseoClientError';
  }
}

export class PaseoClient {
  /** @internal visible for testing */
  readonly bin: string;
  private readonly hostArgs: string[];

  constructor(config?: Partial<PaseoClientConfig>) {
    this.bin = config?.paseoBin ?? DEFAULT_PASEO_BIN;
    this.hostArgs = config?.cliHost ? ['--host', config.cliHost] : [];
  }

  // ─── Read operations (CLI `ls --json`, `inspect --json`) ──────────────────

  /** List all non-archived agents. */
  async listAgents(): Promise<PaseoAgentListItem[]> {
    const raw = await this.runJson(['ls', '--json', ...this.hostArgs]);
    return raw as PaseoAgentListItem[];
  }

  /** Get detailed status for a single agent by ID or prefix. */
  async getAgentStatus(agentId: string): Promise<PaseoAgentDetail> {
    const raw = await this.runJson(['inspect', '--json', agentId, ...this.hostArgs]);
    return raw as PaseoAgentDetail;
  }

  /**
   * Quick liveness check — runs `paseo ls --json --limit 1` and returns success.
   * The daemon is healthy if the CLI exits 0.
   */
  async health(): Promise<{ status: string }> {
    try {
      await this.runCli(['ls', '--json', '--limit', '1', ...this.hostArgs]);
      return { status: 'ok' };
    } catch {
      return { status: 'error' };
    }
  }

  // ─── Write operations (CLI subcommands) ───────────────────────────────────

  /**
   * Import a provider session as a Paseo agent.
   * Uses `paseo import <sessionId> --provider <provider> [--label k=v]`.
   */
  async importAgent(
    sessionId: string,
    provider: string,
    labels?: Record<string, string>,
  ): Promise<PaseoAgentDetail> {
    const args: string[] = ['import', '--json', ...this.hostArgs];

    if (provider) {
      args.push('--provider', provider);
    }
    if (labels) {
      for (const [k, v] of Object.entries(labels)) {
        args.push('--label', `${k}=${v}`);
      }
    }
    args.push(sessionId);

    const raw = await this.runJson(args);
    return raw as PaseoAgentDetail;
  }

  /** Archive (soft-delete) a Paseo agent by ID or prefix. */
  async archiveAgent(agentId: string): Promise<void> {
    await this.runCli(['archive', '--json', ...this.hostArgs, agentId]);
  }

  /**
   * Send a prompt to an existing agent.
   *
   * By default waits for the agent to complete the turn (streams text events
   * via the optional `onEvent` callback) and returns the structured result.
   * Pass `noWait: true` to fire-and-forget.
   */
  async sendPrompt(
    agentId: string,
    prompt: string,
    options?: {
      noWait?: boolean;
      onEvent?: (event: { type: 'text' | 'reasoning'; text: string }) => void;
      signal?: AbortSignal;
    },
  ): Promise<PaseoSendResult> {
    const args: string[] = ['send', '--json', ...this.hostArgs];

    if (options?.noWait) {
      args.push('--no-wait');
    }

    args.push(agentId, prompt);

    // With --json and no --no-wait, the output is JSON after completion.
    // For streaming, we read stderr without --json for real-time text.
    const raw = await this.runCli(args, options?.signal);
    try {
      return JSON.parse(raw) as PaseoSendResult;
    } catch {
      return { text: raw, ok: true };
    }
  }

  /**
   * Stream-send: runs `paseo send` WITHOUT `--json`, forward text/reasoning
   * lines to onEvent in real time. Use when the caller wants to stream agent
   * output as it arrives rather than wait for the full JSON result.
   */
  async streamSend(
    agentId: string,
    prompt: string,
    onEvent: (event: { type: 'text' | 'reasoning'; text: string }) => void,
    signal?: AbortSignal,
  ): Promise<PaseoSendResult> {
    return new Promise<PaseoSendResult>((resolve, reject) => {
      const args = ['send', ...this.hostArgs, agentId, prompt];

      const child = spawn(this.bin, args, {
        stdio: ['ignore', 'pipe', 'pipe'],
        signal,
      });

      let stdout = '';
      let stderr = '';

      if (child.stdout) {
        const rl = createInterface({ input: child.stdout });
        rl.on('line', (line: string) => {
          stdout += line + '\n';
          // Forward as text event for real-time display
          onEvent({ type: 'text', text: line + '\n' });
        });
      }

      if (child.stderr) {
        child.stderr.on('data', (chunk: Buffer) => {
          stderr += chunk.toString();
        });
      }

      once(child, 'close').then((raw) => {
        const exitCode = (raw[0] as number | null) ?? 0;
        if (exitCode !== 0) {
          reject(
            new PaseoClientError(
              `paseo send failed (exit ${exitCode}): ${stderr.trim()}`,
              'send',
              exitCode,
              stderr,
            ),
          );
          return;
        }
        resolve({ text: stdout, ok: true });
      });

      child.on('error', reject);
    });
  }

  /** Interrupt/stop a running agent. */
  async stopAgent(agentId: string): Promise<void> {
    await this.runCli(['stop', ...this.hostArgs, agentId]);
  }

  // ─── Private helpers ───────────────────────────────────────────────────────

  /**
   * Run a CLI command and return stdout as a string.
   * Throws PaseoClientError on non-zero exit.
   */
  private async runCli(
    args: string[],
    signal?: AbortSignal,
  ): Promise<string> {
    return new Promise<string>((resolve, reject) => {
      const child = spawn(this.bin, args, {
        stdio: ['ignore', 'pipe', 'pipe'],
        signal,
      });

      let stdout = '';
      let stderr = '';

      if (child.stdout) {
        child.stdout.on('data', (chunk: Buffer) => {
          stdout += chunk.toString();
        });
      }

      if (child.stderr) {
        child.stderr.on('data', (chunk: Buffer) => {
          stderr += chunk.toString();
        });
      }

      child.on('error', (err: Error) => {
        // If signal aborted, treat as cancellation not error
        if (signal?.aborted) {
          resolve('');
          return;
        }
        reject(err);
      });

      once(child, 'close').then((raw) => {
        const exitCode = (raw[0] as number | null) ?? 0;
        if (signal?.aborted) {
          resolve('');
          return;
        }
        if (exitCode !== 0) {
          const msg = stderr.trim() || `exit code ${exitCode}`;
          reject(
            new PaseoClientError(
              `paseo ${args[0] ?? '?'} failed: ${msg}`,
              args[0] ?? '?',
              exitCode,
              stderr,
            ),
          );
          return;
        }
        resolve(stdout);
      });
    });
  }

  /**
   * Run a CLI command and parse stdout as JSON.
   * Throws PaseoClientError on non-zero exit or parse failure.
   */
  private async runJson(args: string[]): Promise<unknown> {
    const stdout = await this.runCli(args);
    try {
      return JSON.parse(stdout);
    } catch (err) {
      throw new PaseoClientError(
        `paseo ${args[0] ?? '?'} returned invalid JSON: ${(stdout || '<empty>').slice(0, 200)}`,
        args[0] ?? '?',
        0,
        stdout,
      );
    }
  }
}