diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 6831274d2d..339c491e46 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -120,6 +120,7 @@ import { } from "../provider-launch-config.js"; import { renderPromptAttachmentAsText } from "../prompt-attachments.js"; import { appendOrReplaceGrowingAssistantMessage, runProviderTurn } from "./provider-runner.js"; +import { resolveResumeProcessCwd } from "./resume-process-cwd.js"; import { buildStringCommandShellInvocation, createStringCommandShellEnvOverlay, @@ -165,18 +166,6 @@ function rejectOnProcessFailure(child: ChildProcess, stderrChunks: string[]): Pr }); } -async function isDirectory(target: string): Promise { - try { - return (await fs.stat(target)).isDirectory(); - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code === "ENOENT" || code === "ENOTDIR") { - return false; - } - throw error; - } -} - function isRecord(value: unknown): value is Record { return value != null && typeof value === "object" && !Array.isArray(value); } @@ -2883,11 +2872,8 @@ export class ACPAgentSession implements AgentSession, ACPClient { // Loading an archived agent's history runs nothing in its working directory, which // may have been removed with its worktree. session/load still names the original // directory; only the process starts from the home directory. - private async resolveProcessCwd(): Promise { - if (this.resumePurpose !== "history" || (await isDirectory(this.config.cwd))) { - return this.config.cwd; - } - return homedir(); + private resolveProcessCwd(): Promise { + return resolveResumeProcessCwd(this.config.cwd, this.resumePurpose); } private async spawnProcess(): Promise { diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index 936bdbc973..7dc3fd184c 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -5,11 +5,13 @@ import { mkdirSync, mkdtempSync, openSync, + readFileSync, readSync, + realpathSync, rmSync, writeFileSync, } from "node:fs"; -import { tmpdir } from "node:os"; +import { homedir, tmpdir } from "node:os"; import path from "node:path"; import pino from "pino"; import { setImmediate as waitForImmediate } from "node:timers/promises"; @@ -23,6 +25,7 @@ import { PiRpcAgentSession, transformPiModels, } from "./agent.js"; +import { PiCliRuntime } from "./cli-runtime.js"; import { FakePi } from "./test-utils/fake-pi.js"; import { createPiExtensionHost } from "./extensions/index.js"; import { PiExtensionHost } from "./extensions/host.js"; @@ -2719,6 +2722,7 @@ describe("PiRpcAgentClient", () => { const root = mkdtempSync(path.join(tmpdir(), "paseo-pi-import-config-")); const cwd = path.join(root, "workspace"); const sessionsDir = path.join(root, "sessions"); + mkdirSync(cwd, { recursive: true }); mkdirSync(sessionsDir, { recursive: true }); const sessionFile = path.join(sessionsDir, "20260103_session.jsonl"); writeFileSync( @@ -3532,6 +3536,320 @@ describe("PiRpcAgentClient", () => { }); }); +describe("PiRpcAgentClient history resume", () => { + const BUILTIN_MCP_COMMAND = { + name: "mcp", + source: "extension", + sourceInfo: { path: "builtin:mcp" }, + } as const; + const PASEO_MCP_SERVERS = { + paseo: { type: "http", url: "http://127.0.0.1:6767/mcp/agents" }, + } as const; + + interface ArchivedPiSession { + handle: { + provider: string; + sessionId: string; + nativeHandle: string; + metadata: { cwd: string; model?: string }; + }; + sessionFile: string; + header: Record; + body: string; + originalBytes: Buffer; + } + + function createDeletedFolder(): string { + const folder = mkdtempSync(path.join(tmpdir(), "paseo-pi-deleted-cwd-")); + rmSync(folder, { recursive: true, force: true }); + return folder; + } + + function createFileWhereFolderWas(): string { + const folder = createDeletedFolder(); + writeFileSync(folder, "not a directory"); + onTestFinished(() => rmSync(folder, { force: true })); + return folder; + } + + function createExistingFolder(): string { + const folder = mkdtempSync(path.join(tmpdir(), "paseo-pi-existing-cwd-")); + onTestFinished(() => rmSync(folder, { recursive: true, force: true })); + return folder; + } + + function writeArchivedSession(cwd: string, metadata: { model?: string } = {}): ArchivedPiSession { + const dir = mkdtempSync(path.join(tmpdir(), "paseo-pi-archived-session-")); + onTestFinished(() => rmSync(dir, { recursive: true, force: true })); + const sessionFile = path.join(dir, "2026-01-01_pi-session-1.jsonl"); + const header = { + type: "session", + version: 3, + id: "pi-session-1", + timestamp: "2026-01-01T00:00:00.000Z", + cwd, + }; + const body = [ + JSON.stringify({ + type: "message", + id: "entry-1", + parentId: null, + message: { role: "user", content: "Où est le café ☕ ?" }, + }), + JSON.stringify({ + type: "message", + id: "entry-2", + parentId: "entry-1", + message: { role: "assistant", content: [{ type: "text", text: "Dans la cuisine." }] }, + }), + ] + .map((line) => `${line}\n`) + .join(""); + writeFileSync(sessionFile, `${JSON.stringify(header)}\n${body}`); + return { + handle: { + provider: "pi", + sessionId: "pi-session-1", + nativeHandle: sessionFile, + metadata: { cwd, ...metadata }, + }, + sessionFile, + header, + body, + originalBytes: readFileSync(sessionFile), + }; + } + + function readSessionCopy(sessionFile: string): { header: unknown; body: string } { + const content = readUtf8File(sessionFile); + const headerEnd = content.indexOf("\n"); + return { + header: JSON.parse(content.slice(0, headerEnd)), + body: content.slice(headerEnd + 1), + }; + } + + function resumeArchived( + client: PiRpcAgentClient, + archived: ArchivedPiSession, + purpose: "history" | "interactive", + overrides: Partial = {}, + ): Promise { + return client.resumeSession(archived.handle, overrides, undefined, { purpose }); + } + + function writeFakePiProcess(): { + command: [string, ...string[]]; + readLaunches: () => Array<{ cwd: string; session: string | null }>; + } { + const dir = mkdtempSync(path.join(tmpdir(), "paseo-pi-process-")); + onTestFinished(() => rmSync(dir, { recursive: true, force: true })); + const scriptPath = path.join(dir, "fake-pi.cjs"); + const launchesPath = path.join(dir, "launches.jsonl"); + writeFileSync( + scriptPath, + ` +const fs = require("node:fs"); +const readline = require("node:readline"); +const [launchesPath, ...piArgs] = process.argv.slice(2); +const sessionFlag = piArgs.indexOf("--session"); +const sessionFile = sessionFlag === -1 ? null : piArgs[sessionFlag + 1]; +fs.appendFileSync(launchesPath, JSON.stringify({ cwd: process.cwd(), session: sessionFile }) + "\\n"); +if (sessionFile) { + const header = JSON.parse(fs.readFileSync(sessionFile, "utf8").split("\\n", 1)[0]); + if (header.cwd && !fs.existsSync(header.cwd)) { + fs.writeSync( + 2, + "Stored session working directory does not exist: " + header.cwd + "\\nSession file: " + sessionFile + "\\nCurrent working directory: " + process.cwd() + "\\n", + ); + process.exit(1); + } +} +const models = [ + { provider: "openrouter", id: "a" }, + { provider: "openrouter", id: "b" }, +]; +const state = { + sessionId: "pi-session-1", + sessionFile, + model: models[0], + thinkingLevel: "medium", + isStreaming: false, + isCompacting: false, + messageCount: 0, + pendingMessageCount: 0, +}; +const commands = [${JSON.stringify(BUILTIN_MCP_COMMAND)}]; +function handle(request) { + switch (request.type) { + case "get_state": + return state; + case "get_commands": + return { commands }; + case "get_available_models": + return { models }; + case "set_model": + fs.appendFileSync( + sessionFile, + JSON.stringify({ type: "model_change", id: "model-1", provider: request.provider, modelId: request.modelId }) + "\\n", + ); + state.model = models.find((model) => model.id === request.modelId); + return state.model; + default: + return {}; + } +} +readline.createInterface({ input: process.stdin }).on("line", (line) => { + const request = JSON.parse(line); + process.stdout.write( + JSON.stringify({ id: request.id, type: "response", command: request.type, success: true, data: handle(request) }) + "\\n", + ); +}); +`, + ); + return { + command: [process.execPath, scriptPath, launchesPath], + readLaunches: () => + readUtf8File(launchesPath) + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { cwd: string; session: string | null }), + }; + } + + test.each([ + ["is gone", createDeletedFolder], + ["is now a file", createFileWhereFolderWas], + ])( + "a history resume of an agent whose folder %s starts Pi from the home directory on a copy of its session", + async (_folderState, createFolder) => { + const archived = writeArchivedSession(createFolder()); + const pi = new FakePi(); + + const session = await resumeArchived(createClient(pi), archived, "history"); + onTestFinished(() => session.close()); + + expect(pi.recordedLaunches).toHaveLength(1); + const launch = pi.recordedLaunches[0]!; + expect(launch.cwd).toBe(homedir()); + expect(launch.session).not.toBe(archived.sessionFile); + expect(readSessionCopy(launch.session!)).toEqual({ + header: { ...archived.header, cwd: homedir() }, + body: archived.body, + }); + }, + ); + + test("a history resume of an agent whose folder is gone removes its session copy on close and leaves the original unchanged", async () => { + const archived = writeArchivedSession(createDeletedFolder()); + const pi = new FakePi(); + + const session = await resumeArchived(createClient(pi), archived, "history"); + const sessionCopy = pi.recordedLaunches[0]!.session!; + expect(existsSync(sessionCopy)).toBe(true); + await session.close(); + + expect(existsSync(path.dirname(sessionCopy))).toBe(false); + expect(readFileSync(archived.sessionFile)).toEqual(archived.originalBytes); + }); + + test("a history resume of an agent whose folder is gone persists the original session file while Pi reports the copy", async () => { + const storedCwd = createDeletedFolder(); + const archived = writeArchivedSession(storedCwd); + const pi = new FakePi(); + + const session = await resumeArchived(createClient(pi), archived, "history"); + onTestFinished(() => session.close()); + await session.getRuntimeInfo(); + + expect(pi.latestSession().state.sessionFile).toBe(pi.recordedLaunches[0]!.session); + expect(session.describePersistence()).toMatchObject({ + nativeHandle: archived.sessionFile, + metadata: { cwd: storedCwd }, + }); + }); + + test("a history resume of an agent whose folder is gone removes its session copy when Pi fails to start", async () => { + const archived = writeArchivedSession(createDeletedFolder()); + const pi = new FakePi(); + pi.queueSessionSetup((session) => { + session.getStateError = new Error("Pi RPC process exited with code 1 and signal null"); + }); + + await expect(resumeArchived(createClient(pi), archived, "history")).rejects.toThrow( + "Pi RPC process exited with code 1", + ); + + expect(pi.recordedLaunches).toHaveLength(1); + expect(pi.recordedLaunches[0]!.session).not.toBe(archived.sessionFile); + expect(existsSync(path.dirname(pi.recordedLaunches[0]!.session!))).toBe(false); + }); + + test("a history resume of an agent whose folder is gone checks MCP support from the home directory", async () => { + const archived = writeArchivedSession(createDeletedFolder()); + const pi = new FakePi(); + pi.queueCommands([BUILTIN_MCP_COMMAND]); + + const session = await resumeArchived(createClient(pi), archived, "history", { + mcpServers: PASEO_MCP_SERVERS, + }); + onTestFinished(() => session.close()); + + expect(pi.recordedLaunches.map((launch) => launch.cwd)).toEqual([homedir(), homedir()]); + expect(session.capabilities.supportsMcpServers).toBe(true); + }); + + test("an interactive resume of an agent whose folder is gone still fails in its stored folder", async () => { + const storedCwd = createDeletedFolder(); + const archived = writeArchivedSession(storedCwd); + const pi = new FakePi(); + + await expect(resumeArchived(createClient(pi), archived, "interactive")).rejects.toThrow( + `Stored session working directory does not exist: ${storedCwd}`, + ); + + expect(pi.recordedLaunches).toMatchObject([{ cwd: storedCwd, session: archived.sessionFile }]); + }); + + test("a history resume of an agent whose folder exists keeps the stored folder and session file", async () => { + const storedCwd = createExistingFolder(); + const archived = writeArchivedSession(storedCwd); + const pi = new FakePi(); + + const session = await resumeArchived(createClient(pi), archived, "history"); + onTestFinished(() => session.close()); + + expect(pi.recordedLaunches).toMatchObject([{ cwd: storedCwd, session: archived.sessionFile }]); + }); + + test("a history resume of an agent whose folder is gone opens its session in a real Pi process started from the home directory", async () => { + const archived = writeArchivedSession(createDeletedFolder(), { model: "openrouter/b" }); + const fakePiProcess = writeFakePiProcess(); + const logger = pino({ level: "silent" }); + const client = new PiRpcAgentClient({ + logger, + runtime: new PiCliRuntime({ logger, command: fakePiProcess.command }), + }); + + const session = await resumeArchived(client, archived, "history", { + mcpServers: PASEO_MCP_SERVERS, + }); + onTestFinished(() => session.close()); + + const launches = fakePiProcess.readLaunches(); + const home = realpathSync.native(homedir()); + expect(launches.map((launch) => realpathSync.native(launch.cwd))).toEqual([home, home]); + expect(session.capabilities.supportsMcpServers).toBe(true); + const sessionCopy = launches[1]!.session!; + expect(readUtf8File(sessionCopy)).toContain('"type":"model_change"'); + expect(session.describePersistence()?.nativeHandle).toBe(archived.sessionFile); + + await session.close(); + expect(existsSync(path.dirname(sessionCopy))).toBe(false); + expect(readFileSync(archived.sessionFile)).toEqual(archived.originalBytes); + }); +}); + describe("transformPiModels", () => { test("normalizes labels that include the upstream provider prefix", () => { expect( diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 9474e98cc8..8051c57fa9 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -2,7 +2,7 @@ import { mapCustomMessageToToolCall } from "../custom-message.js"; import { randomUUID } from "node:crypto"; import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { homedir, tmpdir } from "node:os"; -import { join, resolve as resolvePath } from "node:path"; +import { basename, join, resolve as resolvePath } from "node:path"; import type { Logger } from "pino"; import stripAnsi from "strip-ansi"; import { z } from "zod"; @@ -22,6 +22,7 @@ import { type AgentPersistenceHandle, type AgentPromptInput, type AgentProvider, + type AgentResumeSessionOptions, type AgentRunOptions, type AgentRunResult, type AgentRuntimeInfo, @@ -75,8 +76,13 @@ import { piExtensionRuntimeBridge, type PiExtensionEventOutput, } from "./extensions/index.js"; +import { resolveResumeProcessCwd } from "../resume-process-cwd.js"; import { revertPiConversation } from "./rewind.js"; -import { listPiImportableSessions, readPiImportSessionConfig } from "./session-descriptor.js"; +import { + copyPiSessionFileWithCwd, + listPiImportableSessions, + readPiImportSessionConfig, +} from "./session-descriptor.js"; import type { PiRuntime, PiRuntimeSession, PiStartSessionInput } from "./runtime.js"; import type { PiAgentSessionEvent, @@ -245,6 +251,7 @@ interface PiRpcAgentSessionOptions { initialState: PiSessionState; capabilities: AgentCapabilityFlags; currentModeId?: string | null; + persistedSessionFile?: string; cleanup?: () => void; extensionTimeoutMs?: number; logger: Logger; @@ -492,13 +499,14 @@ function buildResumeConfig( function buildResumeStartInput(input: { resumeConfig: PiResumeConfig; + processCwd: string; sessionFile: string; launchContext: AgentLaunchContext | undefined; mcpConfigFile: PiTempFile | null; paseoExtension: PiTempFile | null; }): PiStartSessionInput { return { - cwd: input.resumeConfig.cwd, + cwd: input.processCwd, env: input.launchContext?.env, session: input.sessionFile, thinkingOptionId: normalizePiThinkingOption(input.resumeConfig.thinkingOptionId) ?? undefined, @@ -618,6 +626,19 @@ function createPiMcpConfigFile( }; } +async function createPiSessionCopy(sessionFile: string, cwd: string): Promise { + const dir = mkdtempSync(join(tmpdir(), "paseo-pi-session-")); + const cleanup = () => rmSync(dir, { recursive: true, force: true }); + const filePath = join(dir, basename(sessionFile)); + try { + await copyPiSessionFileWithCwd({ sourceFile: sessionFile, targetFile: filePath, cwd }); + } catch (error) { + cleanup(); + throw error; + } + return { path: filePath, cleanup }; +} + function createPiPaseoExtensionFile({ systemPrompt, mcpServers = {}, @@ -1232,6 +1253,7 @@ export class PiRpcAgentSession implements AgentSession { this.capabilities = options.capabilities; this.provider = PI_PROVIDER; this.currentModeId = options.currentModeId ?? null; + this.persistedSessionFile = options.persistedSessionFile; this.cleanup = options.cleanup; this.extensionTimeoutMs = options.extensionTimeoutMs ?? DEFAULT_PI_EXTENSION_RESULT_TIMEOUT_MS; this.logger = options.logger; @@ -1262,6 +1284,7 @@ export class PiRpcAgentSession implements AgentSession { private readonly runtimeSession: PiRuntimeSession; private readonly config: AgentSessionConfig; + private readonly persistedSessionFile?: string; private readonly cleanup?: () => void; private readonly extensionTimeoutMs: number; @@ -1496,7 +1519,7 @@ export class PiRpcAgentSession implements AgentSession { return { provider: this.provider, sessionId: this.state.sessionId, - nativeHandle: this.state.sessionFile, + nativeHandle: this.persistedSessionFile ?? this.state.sessionFile, metadata: { cwd: this.config.cwd, ...(this.config.model ? { model: this.config.model } : {}), @@ -2645,6 +2668,7 @@ export class PiRpcAgentClient implements AgentClient { handle: AgentPersistenceHandle, overrides?: Partial, launchContext?: AgentLaunchContext, + options?: AgentResumeSessionOptions, ): Promise { const sessionFile = handle.nativeHandle; if (!sessionFile) { @@ -2653,6 +2677,7 @@ export class PiRpcAgentClient implements AgentClient { const persistenceMetadata = parsePersistenceMetadata(handle.metadata); const resumeConfig = buildResumeConfig(persistenceMetadata, overrides, this.provider); + const processCwd = await resolveResumeProcessCwd(resumeConfig.cwd, options?.purpose); const providerOptions = PiProviderOptionsSchema.parse( resumeConfig.config.providerOptions ?? {}, @@ -2663,7 +2688,7 @@ export class PiRpcAgentClient implements AgentClient { ...launchContext?.env, }; const mcp = await this.prepareMcpInjection( - resumeConfig.cwd, + processCwd, resumeConfig.config.mcpServers, mcpEnv, runtime, @@ -2676,20 +2701,29 @@ export class PiRpcAgentClient implements AgentClient { ), mcpServers: mcp?.kind === "builtin" ? mcp.servers : undefined, }); + const tempFiles = mcpConfigFile ? [mcpConfigFile, paseoExtension] : [paseoExtension]; + const cleanupTempFiles = () => { + for (const tempFile of tempFiles) tempFile.cleanup(); + }; + let sessionCopy: PiTempFile | null = null; let runtimeSession: PiRuntimeSession; try { + if (processCwd !== resumeConfig.cwd) { + sessionCopy = await createPiSessionCopy(sessionFile, processCwd); + tempFiles.push(sessionCopy); + } runtimeSession = await runtime.startSession( buildResumeStartInput({ resumeConfig, - sessionFile, + processCwd, + sessionFile: sessionCopy?.path ?? sessionFile, launchContext, mcpConfigFile, paseoExtension, }), ); } catch (error) { - mcpConfigFile?.cleanup(); - paseoExtension?.cleanup(); + cleanupTempFiles(); throw error; } try { @@ -2702,15 +2736,15 @@ export class PiRpcAgentClient implements AgentClient { }, initialState, capabilities: capabilitiesForSession(mcp !== null), - cleanup: combineCleanup([mcpConfigFile?.cleanup, paseoExtension?.cleanup]), + persistedSessionFile: sessionCopy ? sessionFile : undefined, + cleanup: cleanupTempFiles, extensionTimeoutMs: providerOptions.extensionTimeoutMs, logger: this.logger, usagePollScheduler: this.usagePollScheduler, }); } catch (error) { await runtimeSession.close().catch(() => undefined); - mcpConfigFile?.cleanup(); - paseoExtension?.cleanup(); + cleanupTempFiles(); throw error; } } diff --git a/packages/server/src/server/agent/providers/pi/session-descriptor.test.ts b/packages/server/src/server/agent/providers/pi/session-descriptor.test.ts index 280b558739..c6b0df5dbf 100644 --- a/packages/server/src/server/agent/providers/pi/session-descriptor.test.ts +++ b/packages/server/src/server/agent/providers/pi/session-descriptor.test.ts @@ -1,9 +1,13 @@ -import { mkdtemp, mkdir, utimes, writeFile } from "node:fs/promises"; +import { mkdtemp, mkdir, readFile, rm, utimes, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; -import { expect, test } from "vitest"; +import { expect, onTestFinished, test } from "vitest"; -import { listPiImportableSessions, readPiImportSessionConfig } from "./session-descriptor.js"; +import { + copyPiSessionFileWithCwd, + listPiImportableSessions, + readPiImportSessionConfig, +} from "./session-descriptor.js"; async function writeSession(root: string, lines: unknown[]): Promise { const sessionsDir = path.join(root, "sessions", "project"); @@ -216,3 +220,97 @@ test("Pi import config preserves thinking before a later model in large sessions thinkingOptionId: "low", }); }); + +test("a Pi session copy names the new cwd in its header and keeps every later byte", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-pi-session-copy-")); + onTestFinished(() => rm(root, { recursive: true, force: true })); + const header = { + type: "session", + version: 3, + id: "pi-session-1", + timestamp: "2026-01-01T00:00:00.000Z", + cwd: path.join(root, "removed-worktree"), + }; + const body = Buffer.from( + Array.from( + { length: 2_000 }, + (_, index) => + `${JSON.stringify({ type: "message", id: `entry-${index}`, message: { role: "user", content: `café ☕ ${index}` } })}\n`, + ).join(""), + ); + expect(body.length).toBeGreaterThan(64 * 1024); + const sourceFile = path.join(root, "source.jsonl"); + const targetFile = path.join(root, "copy.jsonl"); + await writeFile(sourceFile, Buffer.concat([Buffer.from(`${JSON.stringify(header)}\n`), body])); + + await copyPiSessionFileWithCwd({ sourceFile, targetFile, cwd: "/home/user" }); + + const copy = await readFile(targetFile); + const headerEnd = copy.indexOf(0x0a); + expect(JSON.parse(copy.subarray(0, headerEnd).toString("utf8"))).toEqual({ + ...header, + cwd: "/home/user", + }); + expect(copy.subarray(headerEnd + 1).equals(body)).toBe(true); +}); + +test("a Pi session copy refuses a file without a session header", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-pi-session-copy-")); + onTestFinished(() => rm(root, { recursive: true, force: true })); + const sourceFile = path.join(root, "source.jsonl"); + await writeFile(sourceFile, `${JSON.stringify({ type: "message", id: "entry-1" })}\n`); + + await expect( + copyPiSessionFileWithCwd({ sourceFile, targetFile: path.join(root, "copy.jsonl"), cwd: "/" }), + ).rejects.toThrow(`Pi session file has no session header: ${sourceFile}`); +}); + +const COPY_BODY = [ + { type: "message", id: "entry-1", message: { role: "user", content: "café ☕" } }, + { type: "message", id: "entry-2", message: { role: "assistant", content: "Dans la cuisine." } }, +] + .map((entry) => `${JSON.stringify(entry)}\n`) + .join(""); + +const COPY_HEADER = { + type: "session", + version: 3, + id: "pi-session-1", + timestamp: "2026-01-01T00:00:00.000Z", + cwd: "/removed-worktree", +}; + +async function copySessionContent(content: string): Promise { + const root = await mkdtemp(path.join(tmpdir(), "paseo-pi-session-copy-")); + onTestFinished(() => rm(root, { recursive: true, force: true })); + const sourceFile = path.join(root, "source.jsonl"); + const targetFile = path.join(root, "copy.jsonl"); + await writeFile(sourceFile, content); + await copyPiSessionFileWithCwd({ sourceFile, targetFile, cwd: "/home/user" }); + return readFile(targetFile); +} + +test.each([ + ["a header line longer than 64 KiB", "", { ...COPY_HEADER, padding: "x".repeat(70 * 1024) }], + ["a header with an empty cwd", "", { ...COPY_HEADER, cwd: "" }], + ["blank lines before the header", "\n\r\n", COPY_HEADER], +])("a Pi session copy accepts %s", async (_headerShape, prefix, header) => { + const copy = await copySessionContent(`${prefix}${JSON.stringify(header)}\n${COPY_BODY}`); + + const headerEnd = copy.indexOf(0x0a); + expect(JSON.parse(copy.subarray(0, headerEnd).toString("utf8"))).toEqual({ + ...header, + cwd: "/home/user", + }); + expect(copy.subarray(headerEnd + 1).equals(Buffer.from(COPY_BODY))).toBe(true); +}); + +test("a Pi session copy reports a missing session file", async () => { + const root = await mkdtemp(path.join(tmpdir(), "paseo-pi-session-copy-")); + onTestFinished(() => rm(root, { recursive: true, force: true })); + const sourceFile = path.join(root, "missing.jsonl"); + + await expect( + copyPiSessionFileWithCwd({ sourceFile, targetFile: path.join(root, "copy.jsonl"), cwd: "/" }), + ).rejects.toThrow(`Pi session file is missing: ${sourceFile}`); +}); diff --git a/packages/server/src/server/agent/providers/pi/session-descriptor.ts b/packages/server/src/server/agent/providers/pi/session-descriptor.ts index 01f55bc0ce..6c0cc300f2 100644 --- a/packages/server/src/server/agent/providers/pi/session-descriptor.ts +++ b/packages/server/src/server/agent/providers/pi/session-descriptor.ts @@ -1,7 +1,8 @@ -import type { Dirent } from "node:fs"; -import { open, readdir, readFile, stat } from "node:fs/promises"; +import { createReadStream, createWriteStream, type Dirent } from "node:fs"; +import { open, readdir, readFile, stat, writeFile } from "node:fs/promises"; import { homedir } from "node:os"; import path from "node:path"; +import { pipeline } from "node:stream/promises"; import type { ImportableProviderSession, @@ -18,6 +19,7 @@ const PI_SESSION_DIR_ENV = "PI_CODING_AGENT_SESSION_DIR"; const HEAD_BYTES = 64 * 1024; const TAIL_BYTES = 256 * 1024; const FULL_SCAN_LINE_LIMIT = 2_000; +const PI_SESSION_HEADER_SCAN_BYTES = 1024 * 1024; // Rank all discovered files cheaply, then parse only a bounded recent window. const IMPORT_CANDIDATE_OVERSCAN = 40; const IMPORT_CANDIDATE_MIN = 400; @@ -100,6 +102,59 @@ export async function listPiImportableSessions( ); } +export async function copyPiSessionFileWithCwd(input: { + sourceFile: string; + targetFile: string; + cwd: string; +}): Promise { + const found = findSessionHeaderLine( + await readSessionFileStart(input.sourceFile, PI_SESSION_HEADER_SCAN_BYTES), + ); + if (!found) { + throw new Error(`Pi session file has no session header: ${input.sourceFile}`); + } + await writeFile(input.targetFile, `${JSON.stringify({ ...found.header, cwd: input.cwd })}\n`, { + mode: 0o600, + }); + await pipeline( + createReadStream(input.sourceFile, { start: found.bodyStart }), + createWriteStream(input.targetFile, { flags: "a" }), + ); +} + +function findSessionHeaderLine( + scanned: Buffer, +): { header: Record; bodyStart: number } | null { + let lineStart = 0; + while (lineStart < scanned.length) { + const newline = scanned.indexOf(0x0a, lineStart); + const bodyStart = newline === -1 ? scanned.length : newline + 1; + const line = scanned.subarray(lineStart, bodyStart).toString("utf8").trim(); + if (line) { + const header = parseJsonRecord(line); + return header?.type === "session" ? { header, bodyStart } : null; + } + lineStart = bodyStart; + } + return null; +} + +async function readSessionFileStart(filePath: string, maxBytes: number): Promise { + const handle = await open(filePath, "r").catch((error: unknown) => { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + throw new Error(`Pi session file is missing: ${filePath}`); + } + throw error; + }); + try { + const buffer = Buffer.alloc(maxBytes); + const { bytesRead } = await handle.read(buffer, 0, maxBytes, 0); + return buffer.subarray(0, bytesRead); + } finally { + await handle.close().catch(() => undefined); + } +} + export async function readPiImportSessionConfig(filePath: string): Promise { const descriptor = await readPiSessionDescriptor(filePath); if (!descriptor) return {}; diff --git a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts index 983b86c19d..90a80e4c66 100644 --- a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts +++ b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts @@ -1,3 +1,5 @@ +import { existsSync, readFileSync } from "node:fs"; + import type { PiRuntime, PiRuntimeLaunch, @@ -75,6 +77,12 @@ export class FakePi implements PiRuntime { const session = new FakePiSession(launch, this.removedModels); session.commands = this.queuedCommands.shift() ?? []; this.queuedSessionSetups.shift()?.(session); + const missingSessionCwd = launch.noSession ? null : readMissingSessionCwd(launch.session); + if (missingSessionCwd !== null) { + session.getStateError = new Error( + `Pi RPC process exited with code 1 and signal null\nStored session working directory does not exist: ${missingSessionCwd}\nSession file: ${launch.session}\nCurrent working directory: ${launch.cwd}`, + ); + } this.sessions.push(session); return session; } @@ -101,6 +109,27 @@ export class FakePi implements PiRuntime { } } +function readMissingSessionCwd(sessionFile: string | undefined): string | null { + if (!sessionFile || !existsSync(sessionFile)) { + return null; + } + const [headerLine = ""] = readFileSync(sessionFile, "utf8").split("\n", 1); + let header: unknown; + try { + header = JSON.parse(headerLine); + } catch { + return null; + } + if (typeof header !== "object" || header === null || !("cwd" in header)) { + return null; + } + const { type, cwd } = header as { type?: unknown; cwd?: unknown }; + if (type !== "session" || typeof cwd !== "string" || !cwd || existsSync(cwd)) { + return null; + } + return cwd; +} + export class FakePiSession implements PiRuntimeSession { readonly environment: Record; readonly prompts: Array<{ message: string; imageCount: number }> = []; diff --git a/packages/server/src/server/agent/providers/resume-process-cwd.ts b/packages/server/src/server/agent/providers/resume-process-cwd.ts new file mode 100644 index 0000000000..d32637a9a6 --- /dev/null +++ b/packages/server/src/server/agent/providers/resume-process-cwd.ts @@ -0,0 +1,26 @@ +import { stat } from "node:fs/promises"; +import { homedir } from "node:os"; + +import type { AgentResumePurpose } from "../agent-sdk-types.js"; + +export async function resolveResumeProcessCwd( + cwd: string, + purpose: AgentResumePurpose | undefined, +): Promise { + if (purpose !== "history" || (await isDirectory(cwd))) { + return cwd; + } + return homedir(); +} + +async function isDirectory(target: string): Promise { + try { + return (await stat(target)).isDirectory(); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === "ENOENT" || code === "ENOTDIR") { + return false; + } + throw error; + } +}