feat: Wave 1 complete — state machine, Paseo hub, collision detection, PTY search
- Task state machine: TIMED_OUT state, retriable steps, timeout detection - Paseo hub: paseo-client.ts (HTTP+CLI), PaseoBackend (AgentBackend), 14 tests - Collision detection: collision-detector.ts, conflict-index.ts, ws-frames type - PTY search: ring buffer, search route, capture-pane fallback
This commit is contained in:
341
apps/coder/src/services/paseo-client.ts
Normal file
341
apps/coder/src/services/paseo-client.ts
Normal file
@@ -0,0 +1,341 @@
|
||||
/**
|
||||
* 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,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user