From 4bcf98a9e3b619357207dc5953f1aa3be8e1b52a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 30 Apr 2026 21:33:44 +0800 Subject: [PATCH] feat(cli): add `paseo import --provider ` to resume existing sessions (#632) Surfaces the daemon's existing resume-from-persistence capability through a new top-level CLI command so users can pull existing Claude Code, Codex, or OpenCode sessions into Paseo without losing history. Adds a new additive `import_agent_request` WebSocket message; no existing schemas change. Refs #611, #237, #492; partially addresses #268. --- packages/cli/src/cli.ts | 5 + packages/cli/src/commands/agent/import.ts | 167 ++++++++++++++++++ packages/cli/src/commands/agent/index.ts | 5 + packages/server/src/client/daemon-client.ts | 48 +++++ .../src/server/agent/agent-manager.test.ts | 67 +++++++ .../server/src/server/agent/agent-manager.ts | 19 ++ packages/server/src/server/session.ts | 107 +++++++++++ packages/server/src/shared/messages.ts | 10 ++ 8 files changed, 428 insertions(+) create mode 100644 packages/cli/src/commands/agent/import.ts diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index 67b1d4b91..11d9af154 100644 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -22,6 +22,7 @@ import { addInspectOptions, runInspectCommand } from "./commands/agent/inspect.j import { addWaitOptions, runWaitCommand } from "./commands/agent/wait.js"; import { addArchiveOptions, runArchiveCommand } from "./commands/agent/archive.js"; import { addAttachOptions, runAttachCommand } from "./commands/agent/attach.js"; +import { addImportOptions, runImportCommand } from "./commands/agent/import.js"; import { withOutput } from "./output/index.js"; import { onboardCommand } from "./commands/onboard.js"; import { @@ -60,6 +61,10 @@ export function createCli(): Command { withOutput(runRunCommand), ); + addJsonAndDaemonHostOptions(addImportOptions(program.command("import"))).action( + withOutput(runImportCommand), + ); + addDaemonHostOption(addAttachOptions(program.command("attach"))).action(runAttachCommand); addDaemonHostOption(addLogsOptions(program.command("logs"))).action(runLogsCommand); diff --git a/packages/cli/src/commands/agent/import.ts b/packages/cli/src/commands/agent/import.ts new file mode 100644 index 000000000..91ecc134e --- /dev/null +++ b/packages/cli/src/commands/agent/import.ts @@ -0,0 +1,167 @@ +import type { Command } from "commander"; +import { connectToDaemon, getDaemonHost } from "../../utils/client.js"; +import { collectMultiple } from "../../utils/command-options.js"; +import type { CommandError, CommandOptions, SingleResult } from "../../output/index.js"; +import { agentRunSchema, type AgentRunResult } from "./run.js"; +import type { AgentSnapshotPayload } from "@getpaseo/server"; + +const IMPORT_PROVIDERS = new Set(["claude", "codex", "opencode", "acp"]); + +export function addImportOptions(cmd: Command): Command { + return cmd + .description("Import an existing provider session as a Paseo agent") + .argument("", "Provider session/thread ID to import") + .requiredOption("--provider ", "Agent provider: claude, codex, opencode, or acp") + .option("--cwd ", "Working directory for providers that require it") + .option( + "--label ", + "Add label(s) to the agent (can be used multiple times)", + collectMultiple, + [], + ); +} + +export interface AgentImportOptions extends CommandOptions { + provider?: string; + cwd?: string; + label?: string[]; + host?: string; +} + +export type AgentImportCommandResult = SingleResult; + +function toImportResult(agent: AgentSnapshotPayload): AgentRunResult { + return { + agentId: agent.id, + status: agent.status === "running" ? "running" : "created", + provider: agent.provider, + cwd: agent.cwd, + title: agent.title, + }; +} + +function parseImportProvider(provider: string | undefined): string { + const normalizedProvider = provider?.trim(); + if (!normalizedProvider) { + throw { + code: "MISSING_PROVIDER", + message: "Provider is required", + details: "Usage: paseo import --provider ", + } satisfies CommandError; + } + + if (!IMPORT_PROVIDERS.has(normalizedProvider)) { + throw { + code: "INVALID_PROVIDER", + message: `Unsupported provider: ${normalizedProvider}`, + details: "Supported providers: claude, codex, opencode, acp", + } satisfies CommandError; + } + + return normalizedProvider; +} + +function parseImportLabels(labelFlags: string[] | undefined): Record { + const labels: Record = {}; + if (!labelFlags) { + return labels; + } + + for (const labelFlag of labelFlags) { + const eqIndex = labelFlag.indexOf("="); + if (eqIndex === -1) { + throw { + code: "INVALID_LABEL", + message: `Invalid label format: ${labelFlag}`, + details: "Labels must be in key=value format", + } satisfies CommandError; + } + + const key = labelFlag.slice(0, eqIndex).trim(); + if (!key) { + throw { + code: "INVALID_LABEL", + message: `Invalid label format: ${labelFlag}`, + details: "Labels must include a non-empty key in key=value format", + } satisfies CommandError; + } + + labels[key] = labelFlag.slice(eqIndex + 1); + } + + return labels; +} + +async function connectToDaemonOrThrow( + hostOption: string | undefined, + host: string, +): Promise>> { + try { + return await connectToDaemon({ host: hostOption }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + throw { + code: "DAEMON_NOT_RUNNING", + message: `Cannot connect to daemon at ${host}: ${message}`, + details: "Start the daemon with: paseo daemon start", + } satisfies CommandError; + } +} + +export async function runImportCommand( + sessionIdArg: string, + options: AgentImportOptions, + _command: Command, +): Promise { + const host = getDaemonHost({ host: options.host as string | undefined }); + const sessionId = sessionIdArg.trim(); + if (!sessionId) { + throw { + code: "MISSING_SESSION_ID", + message: "Session ID is required", + details: "Usage: paseo import --provider ", + } satisfies CommandError; + } + + const provider = parseImportProvider(options.provider); + const cwd = options.cwd?.trim(); + if (options.cwd !== undefined && !cwd) { + throw { + code: "INVALID_CWD", + message: "--cwd cannot be empty", + details: "Provide a working directory path or omit --cwd", + } satisfies CommandError; + } + + const labels = parseImportLabels(options.label); + const client = await connectToDaemonOrThrow(options.host as string | undefined, host); + + try { + const agent = await client.importAgent({ + provider, + sessionId, + ...(cwd ? { cwd } : {}), + ...(Object.keys(labels).length > 0 ? { labels } : {}), + }); + + await client.close(); + + return { + type: "single", + data: toImportResult(agent), + schema: agentRunSchema, + }; + } catch (err) { + await client.close().catch(() => {}); + + if (err && typeof err === "object" && "code" in err) { + throw err; + } + + const message = err instanceof Error ? err.message : String(err); + throw { + code: "AGENT_IMPORT_FAILED", + message: `Failed to import agent: ${message}`, + } satisfies CommandError; + } +} diff --git a/packages/cli/src/commands/agent/index.ts b/packages/cli/src/commands/agent/index.ts index 69afbaf7d..9c592d2ae 100644 --- a/packages/cli/src/commands/agent/index.ts +++ b/packages/cli/src/commands/agent/index.ts @@ -11,6 +11,7 @@ import { addInspectOptions, runInspectCommand } from "./inspect.js"; import { addWaitOptions, runWaitCommand } from "./wait.js"; import { addAttachOptions, runAttachCommand } from "./attach.js"; import { addReloadOptions, runReloadCommand } from "./reload.js"; +import { addImportOptions, runImportCommand } from "./import.js"; import { runUpdateCommand } from "./update.js"; import { withOutput } from "../../output/index.js"; import { @@ -29,6 +30,10 @@ export function createAgentCommand(): Command { withOutput(runRunCommand), ); + addJsonAndDaemonHostOptions(addImportOptions(agent.command("import"))).action( + withOutput(runImportCommand), + ); + addDaemonHostOption(addAttachOptions(agent.command("attach"))).action(runAttachCommand); addDaemonHostOption(addLogsOptions(agent.command("logs"))).action(runLogsCommand); diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index 5a6274fd2..8c748f1dc 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -122,6 +122,13 @@ const perfNow: () => number = ? () => performance.now() : () => Date.now(); +export interface ImportAgentInput { + provider: AgentProvider; + sessionId: string; + cwd?: string; + labels?: Record; +} + export type { DaemonTransport, DaemonTransportFactory, @@ -1763,6 +1770,47 @@ export class DaemonClient { return status.agent; } + async importAgent(input: ImportAgentInput): Promise { + const requestId = this.createRequestId(); + const message = SessionInboundMessageSchema.parse({ + type: "import_agent_request", + requestId, + provider: input.provider, + sessionId: input.sessionId, + ...(input.cwd ? { cwd: input.cwd } : {}), + ...(input.labels && Object.keys(input.labels).length > 0 ? { labels: input.labels } : {}), + }); + + const status = await this.sendRequest({ + requestId, + message, + timeout: 15000, + options: { skipQueue: true }, + select: (msg) => { + if (msg.type !== "status") { + return null; + } + const resumed = AgentResumedStatusPayloadSchema.safeParse(msg.payload); + if (resumed.success && resumed.data.requestId === requestId) { + return resumed.data; + } + + const failed = AgentCreateFailedStatusPayloadSchema.safeParse(msg.payload); + if (failed.success && failed.data.requestId === requestId) { + return failed.data; + } + + return null; + }, + }); + + if (status.status === "agent_create_failed") { + throw new Error(status.error); + } + + return status.agent; + } + async refreshAgent(agentId: string, requestId?: string): Promise { const resolvedRequestId = this.createRequestId(requestId); const message = SessionInboundMessageSchema.parse({ diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index a62c7fa54..33174a883 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -18,6 +18,7 @@ import type { AgentSessionConfig, AgentStreamEvent, AgentTimelineItem, + PersistedAgentDescriptor, } from "./agent-sdk-types.js"; import type { ProviderDefinition } from "./provider-registry.js"; @@ -55,6 +56,30 @@ function createFeature(args: { id: string; label: string; value: boolean }): Age }; } +function createPersistedDescriptor(args: { + cwd: string; + sessionId: string; + nativeHandle?: string; +}): PersistedAgentDescriptor { + return { + provider: "codex", + sessionId: args.sessionId, + cwd: args.cwd, + title: null, + lastActivityAt: new Date("2026-01-01T00:00:00Z"), + persistence: { + provider: "codex", + sessionId: args.sessionId, + nativeHandle: args.nativeHandle, + metadata: { + provider: "codex", + cwd: args.cwd, + }, + }, + timeline: [], + }; +} + class TestAgentClient implements AgentClient { readonly provider = "codex" as const; readonly capabilities = TEST_CAPABILITIES; @@ -1088,6 +1113,48 @@ test("resumeAgentFromPersistence keeps metadata config, applies overrides, and p }); }); +test("findPersistedAgent returns matching descriptors by session id or native handle", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-find-persisted-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + + const descriptors: PersistedAgentDescriptor[] = [ + createPersistedDescriptor({ + cwd: workdir, + sessionId: "session-direct", + nativeHandle: "native-direct", + }), + createPersistedDescriptor({ + cwd: workdir, + sessionId: "session-other", + nativeHandle: "native-match", + }), + ]; + + class PersistedAgentsClient extends TestAgentClient { + lastLimit: number | undefined; + + override async listPersistedAgents(options?: { limit?: number }) { + this.lastLimit = options?.limit; + return descriptors; + } + } + + const client = new PersistedAgentsClient(); + const manager = new AgentManager({ + clients: { + codex: client, + }, + registry: storage, + logger, + }); + + await expect(manager.findPersistedAgent("codex", "session-direct")).resolves.toBe(descriptors[0]); + await expect(manager.findPersistedAgent("codex", "native-match")).resolves.toBe(descriptors[1]); + await expect(manager.findPersistedAgent("codex", "missing")).resolves.toBeNull(); + expect(client.lastLimit).toBe(200); +}); + test("reloadAgentSession passes daemon launch env through the provider launch context", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-context-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index c09cf1791..37a3817d1 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -606,6 +606,25 @@ export class AgentManager { .slice(0, limit); } + async findPersistedAgent( + provider: AgentProvider, + sessionId: string, + ): Promise { + const client = this.requireClient(provider); + if (!client.listPersistedAgents) { + return null; + } + + const descriptors = await client.listPersistedAgents({ limit: 200 }); + return ( + descriptors.find((descriptor) => { + return ( + descriptor.sessionId === sessionId || descriptor.persistence.nativeHandle === sessionId + ); + }) ?? null + ); + } + async listProviderAvailability(): Promise { const checks = Array.from(this.clients.keys()).map(async (provider) => { const client = this.clients.get(provider); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index ac9949d54..2d0b93088 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -758,6 +758,45 @@ class VoiceFeatureUnavailableError extends Error { } } +interface BuildImportPersistenceHandleInput { + provider: AgentProvider; + sessionId: string; + cwd?: string; +} + +function buildImportPersistenceHandle( + input: BuildImportPersistenceHandleInput, +): AgentPersistenceHandle { + const cwd = input.cwd ?? process.cwd(); + return { + provider: input.provider, + sessionId: input.sessionId, + nativeHandle: input.sessionId, + metadata: { + provider: input.provider, + cwd, + }, + }; +} + +function applyImportCwdOverride( + handle: AgentPersistenceHandle, + cwd: string | undefined, +): AgentPersistenceHandle { + if (!cwd) { + return handle; + } + + return { + ...handle, + metadata: { + ...handle.metadata, + provider: handle.provider, + cwd, + }, + }; +} + function convertPCMToWavBuffer( pcmBuffer: Buffer, sampleRate: number, @@ -1846,6 +1885,8 @@ export class Session { return this.handleCreateAgentRequest(msg); case "resume_agent_request": return this.handleResumeAgentRequest(msg); + case "import_agent_request": + return this.handleImportAgentRequest(msg); case "refresh_agent_request": return this.handleRefreshAgentRequest(msg); case "cancel_agent_request": @@ -3251,6 +3292,72 @@ export class Session { } } + private async handleImportAgentRequest( + msg: Extract, + ): Promise { + const { provider, sessionId, cwd, labels, requestId } = msg; + this.sessionLogger.info({ sessionId, provider }, `Importing agent ${sessionId} (${provider})`); + + try { + const descriptor = await this.agentManager.findPersistedAgent(provider, sessionId); + if (!descriptor && provider === "opencode" && !cwd) { + throw new Error( + "OpenCode sessions require --cwd when the session cannot be found in persisted agents", + ); + } + + const handle = descriptor + ? applyImportCwdOverride(descriptor.persistence, cwd) + : buildImportPersistenceHandle({ provider, sessionId, cwd }); + const overrides = cwd ? ({ cwd } satisfies Partial) : undefined; + + await this.unarchiveAgentByHandle(handle); + const snapshot = await this.agentManager.resumeAgentFromPersistence( + handle, + overrides, + undefined, + { + labels, + }, + ); + await unarchiveAgentState(this.agentStorage, this.agentManager, snapshot.id); + await this.agentManager.hydrateTimelineFromProvider(snapshot.id); + await this.forwardAgentUpdate(snapshot); + const timelineSize = this.agentManager.getTimeline(snapshot.id).length; + const agentPayload = await this.buildAgentPayload(snapshot); + this.emit({ + type: "status", + payload: { + status: "agent_resumed", + agentId: snapshot.id, + requestId, + timelineSize, + agent: agentPayload, + }, + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.sessionLogger.error({ err: error }, "Failed to import agent"); + this.emit({ + type: "status", + payload: { + status: "agent_create_failed", + requestId, + error: message, + }, + }); + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "error", + content: `Failed to import agent: ${message}`, + }, + }); + } + } + private async handleRefreshAgentRequest( msg: Extract, ): Promise { diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index ee1f1533d..2995630a5 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -1010,6 +1010,15 @@ export const ResumeAgentRequestMessageSchema = z.object({ requestId: z.string(), }); +export const ImportAgentRequestMessageSchema = z.object({ + type: z.literal("import_agent_request"), + provider: AgentProviderSchema, + sessionId: z.string(), + cwd: z.string().optional(), + labels: z.record(z.string()).optional(), + requestId: z.string(), +}); + export const RefreshAgentRequestMessageSchema = z.object({ type: z.literal("refresh_agent_request"), agentId: z.string(), @@ -1652,6 +1661,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ RefreshProvidersSnapshotRequestMessageSchema, ProviderDiagnosticRequestMessageSchema, ResumeAgentRequestMessageSchema, + ImportAgentRequestMessageSchema, RefreshAgentRequestMessageSchema, CancelAgentRequestMessageSchema, ShutdownServerRequestMessageSchema,