Compare commits

..
Author SHA1 Message Date
Mohamed Boudra 8bd38fb236 chore(release): cut 0.15.0-beta.1 2026-11-01 06:19:33 +01:00
Mohamed Boudra 7e5ad53491 docs(changelog): prepare 0.15.0-beta.1 2026-11-01 06:11:09 +01:00
paseo-ai[bot] a529578cd1 fix: update lockfile signatures and Nix hash [skip ci]
(cherry picked from commit b2666a0aec)
2026-11-01 06:11:09 +01:00
36 changed files with 1477 additions and 2679 deletions
+1 -4
View File
@@ -74,10 +74,7 @@ registrations remain inspectable; a changed destination requires explicit identi
Both daemons advertise `connectedHosts`; the shared link additionally requires `hostMultiplexing`.
A completes capability and pinned-identity checks on each physical recovery before admitting channels. There is no local fallback. An older daemon continues to serve existing local
operations; requesting this feature gives update guidance. The registry retries an older destination
with capped backoff, so it comes online once updated. Self, duplicate, changed-identity, and
authentication failures stay in `error` until an update changes the connection, `enabled`, or
`replaceIdentity`, or A restarts.
operations; requesting this feature gives update guidance.
Disposing a routed API releases its observations and session. It does not remove the saved host or
affect sibling API scopes. Disposing the origin cascades to its derived scopes. Observations restore
+1 -1
View File
@@ -127,7 +127,7 @@ buildNpmPackage {
# The Nix output is not a distributable DMG, so leave it unsigned and
# disable the hardened runtime that requires a matching signature.
CSC_IDENTITY_AUTO_DISCOVERY=false \
npm exec --offline -- electron-builder \
../../node_modules/.bin/electron-builder \
--config electron-builder.yml \
--dir \
--mac \
+1 -1
View File
@@ -1 +1 @@
sha256-KN/AVZV/o0TleLLpmggSrdJqjFDJ/NJYVHiC54xb4jQ=
sha256-CeT2PtYIAFwvBC9LIku8VkTl0pvKxYzSCn0QW2ManfU=
+1295 -1018
View File
File diff suppressed because it is too large Load Diff
@@ -717,13 +717,11 @@ describe("terminal emulator runtime in a real browser", () => {
{ name: "DA1-zero", bytes: "\x1b[0c" },
{ name: "DA2", bytes: "\x1b[>c" },
{ name: "DA3", bytes: "\x1b[=c" },
{ name: "XTVERSION", bytes: "\x1b[>q" },
{ name: "DSR-5", bytes: "\x1b[5n" },
{ name: "DSR-6", bytes: "\x1b[6n" },
{ name: "DSR-?6", bytes: "\x1b[?6n" },
{ name: "DECRQM", bytes: "\x1b[1$p" },
{ name: "DECRQM-?", bytes: "\x1b[?1$p" },
{ name: "DECRQSS", bytes: "\x1bP$qm\x1b\\" },
{ name: "OSC-10-foreground-color", bytes: "\x1b]10;?\x07" },
{ name: "OSC-11-background-color", bytes: "\x1b]11;?\x07" },
{ name: "OSC-12-cursor-color", bytes: "\x1b]12;?\x07" },
@@ -647,7 +647,6 @@ export class TerminalEmulatorRuntime {
terminal.parser.registerCsiHandler({ final: "c" }, () => true);
terminal.parser.registerCsiHandler({ prefix: ">", final: "c" }, () => true);
terminal.parser.registerCsiHandler({ prefix: "=", final: "c" }, () => true);
terminal.parser.registerCsiHandler({ prefix: ">", final: "q" }, () => true);
terminal.parser.registerCsiHandler({ final: "n" }, () => true);
terminal.parser.registerCsiHandler({ prefix: "?", final: "n" }, () => true);
terminal.parser.registerCsiHandler({ final: "R" }, () => true);
@@ -656,7 +655,6 @@ export class TerminalEmulatorRuntime {
{ prefix: "?", intermediates: "$", final: "p" },
() => true,
);
terminal.parser.registerDcsHandler({ intermediates: "$", final: "q" }, () => true);
for (const code of [10, 11, 12]) {
terminal.parser.registerOscHandler(code, (data) => data.trim() === "?");
}
+3 -4
View File
@@ -204,16 +204,15 @@ export function createHostTransportFactory(
}
export function createHostClient(
connection: string,
config: Omit<DaemonClientConfig, "url"> & HostTransportOptions,
config: Omit<DaemonClientConfig, "url" | "transportFactory"> & HostTransportOptions,
budget?: HostTransportBudget,
): DaemonClient {
const target = parseHostTarget(connection);
const { webSocketAgent, transportFactory, ...clientConfig } = config;
const { webSocketAgent, ...clientConfig } = config;
return new DaemonClient({
...clientConfig,
url: "ws://host-connector/ws",
...(target.password ? { password: target.password } : {}),
transportFactory:
transportFactory ?? createHostTransportFactory(connection, budget, { webSocketAgent }),
transportFactory: createHostTransportFactory(connection, budget, { webSocketAgent }),
});
}
+1 -1
View File
@@ -41,7 +41,7 @@
"@types/node": "24.6.0",
"@types/ws": "^8.5.14",
"electron": "44.2.0",
"electron-builder": "26.18.0",
"electron-builder": "26.8.1",
"typescript": "5.9.3",
"unzip-crx-3": "^0.2.0",
"vitest": "^4.1.6",
@@ -120,7 +120,6 @@ 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,
@@ -166,6 +165,18 @@ function rejectOnProcessFailure(child: ChildProcess, stderrChunks: string[]): Pr
});
}
async function isDirectory(target: string): Promise<boolean> {
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<string, unknown> {
return value != null && typeof value === "object" && !Array.isArray(value);
}
@@ -2872,8 +2883,11 @@ 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 resolveProcessCwd(): Promise<string> {
return resolveResumeProcessCwd(this.config.cwd, this.resumePurpose);
private async resolveProcessCwd(): Promise<string> {
if (this.resumePurpose !== "history" || (await isDirectory(this.config.cwd))) {
return this.config.cwd;
}
return homedir();
}
private async spawnProcess(): Promise<SpawnedACPProcess> {
@@ -5,13 +5,11 @@ import {
mkdirSync,
mkdtempSync,
openSync,
readFileSync,
readSync,
realpathSync,
rmSync,
writeFileSync,
} from "node:fs";
import { homedir, tmpdir } from "node:os";
import { tmpdir } from "node:os";
import path from "node:path";
import pino from "pino";
import { setImmediate as waitForImmediate } from "node:timers/promises";
@@ -25,7 +23,6 @@ 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";
@@ -2722,7 +2719,6 @@ 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(
@@ -3536,320 +3532,6 @@ 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<string, unknown>;
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<AgentSessionConfig> = {},
): Promise<AgentSession> {
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(
@@ -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 { basename, join, resolve as resolvePath } from "node:path";
import { join, resolve as resolvePath } from "node:path";
import type { Logger } from "pino";
import stripAnsi from "strip-ansi";
import { z } from "zod";
@@ -22,7 +22,6 @@ import {
type AgentPersistenceHandle,
type AgentPromptInput,
type AgentProvider,
type AgentResumeSessionOptions,
type AgentRunOptions,
type AgentRunResult,
type AgentRuntimeInfo,
@@ -76,13 +75,8 @@ import {
piExtensionRuntimeBridge,
type PiExtensionEventOutput,
} from "./extensions/index.js";
import { resolveResumeProcessCwd } from "../resume-process-cwd.js";
import { revertPiConversation } from "./rewind.js";
import {
copyPiSessionFileWithCwd,
listPiImportableSessions,
readPiImportSessionConfig,
} from "./session-descriptor.js";
import { listPiImportableSessions, readPiImportSessionConfig } from "./session-descriptor.js";
import type { PiRuntime, PiRuntimeSession, PiStartSessionInput } from "./runtime.js";
import type {
PiAgentSessionEvent,
@@ -251,7 +245,6 @@ interface PiRpcAgentSessionOptions {
initialState: PiSessionState;
capabilities: AgentCapabilityFlags;
currentModeId?: string | null;
persistedSessionFile?: string;
cleanup?: () => void;
extensionTimeoutMs?: number;
logger: Logger;
@@ -499,14 +492,13 @@ function buildResumeConfig(
function buildResumeStartInput(input: {
resumeConfig: PiResumeConfig;
processCwd: string;
sessionFile: string;
launchContext: AgentLaunchContext | undefined;
mcpConfigFile: PiTempFile | null;
paseoExtension: PiTempFile | null;
}): PiStartSessionInput {
return {
cwd: input.processCwd,
cwd: input.resumeConfig.cwd,
env: input.launchContext?.env,
session: input.sessionFile,
thinkingOptionId: normalizePiThinkingOption(input.resumeConfig.thinkingOptionId) ?? undefined,
@@ -626,19 +618,6 @@ function createPiMcpConfigFile(
};
}
async function createPiSessionCopy(sessionFile: string, cwd: string): Promise<PiTempFile> {
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 = {},
@@ -1253,7 +1232,6 @@ 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;
@@ -1284,7 +1262,6 @@ 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;
@@ -1519,7 +1496,7 @@ export class PiRpcAgentSession implements AgentSession {
return {
provider: this.provider,
sessionId: this.state.sessionId,
nativeHandle: this.persistedSessionFile ?? this.state.sessionFile,
nativeHandle: this.state.sessionFile,
metadata: {
cwd: this.config.cwd,
...(this.config.model ? { model: this.config.model } : {}),
@@ -2668,7 +2645,6 @@ export class PiRpcAgentClient implements AgentClient {
handle: AgentPersistenceHandle,
overrides?: Partial<AgentSessionConfig>,
launchContext?: AgentLaunchContext,
options?: AgentResumeSessionOptions,
): Promise<AgentSession> {
const sessionFile = handle.nativeHandle;
if (!sessionFile) {
@@ -2677,7 +2653,6 @@ 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 ?? {},
@@ -2688,7 +2663,7 @@ export class PiRpcAgentClient implements AgentClient {
...launchContext?.env,
};
const mcp = await this.prepareMcpInjection(
processCwd,
resumeConfig.cwd,
resumeConfig.config.mcpServers,
mcpEnv,
runtime,
@@ -2701,29 +2676,20 @@ 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,
processCwd,
sessionFile: sessionCopy?.path ?? sessionFile,
sessionFile,
launchContext,
mcpConfigFile,
paseoExtension,
}),
);
} catch (error) {
cleanupTempFiles();
mcpConfigFile?.cleanup();
paseoExtension?.cleanup();
throw error;
}
try {
@@ -2736,15 +2702,15 @@ export class PiRpcAgentClient implements AgentClient {
},
initialState,
capabilities: capabilitiesForSession(mcp !== null),
persistedSessionFile: sessionCopy ? sessionFile : undefined,
cleanup: cleanupTempFiles,
cleanup: combineCleanup([mcpConfigFile?.cleanup, paseoExtension?.cleanup]),
extensionTimeoutMs: providerOptions.extensionTimeoutMs,
logger: this.logger,
usagePollScheduler: this.usagePollScheduler,
});
} catch (error) {
await runtimeSession.close().catch(() => undefined);
cleanupTempFiles();
mcpConfigFile?.cleanup();
paseoExtension?.cleanup();
throw error;
}
}
@@ -1,13 +1,9 @@
import { mkdtemp, mkdir, readFile, rm, utimes, writeFile } from "node:fs/promises";
import { mkdtemp, mkdir, utimes, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, onTestFinished, test } from "vitest";
import { expect, test } from "vitest";
import {
copyPiSessionFileWithCwd,
listPiImportableSessions,
readPiImportSessionConfig,
} from "./session-descriptor.js";
import { listPiImportableSessions, readPiImportSessionConfig } from "./session-descriptor.js";
async function writeSession(root: string, lines: unknown[]): Promise<string> {
const sessionsDir = path.join(root, "sessions", "project");
@@ -220,97 +216,3 @@ 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<Buffer> {
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}`);
});
@@ -1,8 +1,7 @@
import { createReadStream, createWriteStream, type Dirent } from "node:fs";
import { open, readdir, readFile, stat, writeFile } from "node:fs/promises";
import type { Dirent } from "node:fs";
import { open, readdir, readFile, stat } from "node:fs/promises";
import { homedir } from "node:os";
import path from "node:path";
import { pipeline } from "node:stream/promises";
import type {
ImportableProviderSession,
@@ -19,7 +18,6 @@ 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;
@@ -102,59 +100,6 @@ export async function listPiImportableSessions(
);
}
export async function copyPiSessionFileWithCwd(input: {
sourceFile: string;
targetFile: string;
cwd: string;
}): Promise<void> {
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<string, unknown>; 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<Buffer> {
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<PiImportSessionConfig> {
const descriptor = await readPiSessionDescriptor(filePath);
if (!descriptor) return {};
@@ -1,5 +1,3 @@
import { existsSync, readFileSync } from "node:fs";
import type {
PiRuntime,
PiRuntimeLaunch,
@@ -77,12 +75,6 @@ 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;
}
@@ -109,27 +101,6 @@ 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<string, string>;
readonly prompts: Array<{ message: string; imageCount: number }> = [];
@@ -1,26 +0,0 @@
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<string> {
if (purpose !== "history" || (await isDirectory(cwd))) {
return cwd;
}
return homedir();
}
async function isDirectory(target: string): Promise<boolean> {
try {
return (await stat(target)).isDirectory();
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code === "ENOENT" || code === "ENOTDIR") {
return false;
}
throw error;
}
}
@@ -670,11 +670,11 @@ describe("paseo daemon bootstrap", () => {
test("rolls back standalone listener without starting plugins when main listen fails", async () => {
const mainPort = await findFreePort();
const standalonePort = await findFreePort();
const occupiedMain = http.createServer((_req, res) => {
res.end("occupied-main");
});
await new Promise<void>((resolve) => occupiedMain.listen(mainPort, "127.0.0.1", resolve));
const standalonePort = await findFreePort();
const paseoHomeRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-main-rollback-"));
const paseoHome = path.join(paseoHomeRoot, ".paseo");
@@ -784,211 +784,6 @@ test("old capabilities fail before execution and authentication errors recover t
expect((await online(paseo)).hosts[0]!.serverId).toBe(b.daemon.getServerId());
});
async function updatableDestination() {
let features: { connectedHosts: boolean; hostMultiplexing: boolean } | undefined;
let holding = false;
const held: (() => void)[] = [];
let connections = 0;
const http = createServer();
const ws = new WebSocketServer({ server: http });
ws.on("connection", (socket) => {
connections++;
socket.on("message", (data) => {
const frame = JSON.parse(data.toString());
if (frame.type === "ping") socket.send(JSON.stringify({ type: "pong" }));
if (frame.type !== "hello") return;
const reply = () =>
socket.send(
JSON.stringify({
type: "session",
message: {
type: "status",
payload: {
status: "server_info",
serverId: "srv_updated",
hostname: "updated",
version: "0.12.0",
...(features ? { features } : {}),
},
},
}),
);
if (holding) held.push(reply);
else reply();
});
});
http.listen(0, "127.0.0.1");
await once(http, "listening");
cleanup.push(async () => {
for (const socket of ws.clients) socket.terminate();
await new Promise<void>((resolve) => ws.close(() => resolve()));
await new Promise<void>((resolve) => http.close(() => resolve()));
});
const address = http.address();
if (!address || typeof address === "string") throw new Error("Missing destination port");
return {
port: address.port,
connections: () => connections,
hold: () => {
holding = true;
},
update: () => {
features = { connectedHosts: true, hostMultiplexing: true };
holding = false;
for (const reply of held.splice(0)) reply();
},
};
}
async function capabilityRetry(paseo: ReturnType<typeof createPaseoApi>): Promise<number> {
let retryAt = "";
await expect
.poll(
async () => {
const row = (await paseo.hosts.list()).hosts[0];
retryAt = row?.nextRetryAt ?? "";
return row;
},
{ timeout: 5000 },
)
.toMatchObject({
status: "offline",
nextRetryAt: expect.any(String),
error: { code: "unsupported_capability" },
});
return Date.parse(retryAt);
}
test("a destination updated past unsupported_capability keeps its guidance while retrying and comes online, while duplicate_host stays terminal", async () => {
const destination = await updatableDestination();
const a = await daemon();
const { paseo } = await api(a);
const host = await paseo.hosts.add({ connection: `127.0.0.1:${destination.port}` });
await capabilityRetry(paseo);
destination.hold();
await expect
.poll(async () => (await paseo.hosts.list()).hosts[0], { timeout: 5000 })
.toMatchObject({ status: "connecting", error: { code: "unsupported_capability" } });
await expect(paseo.hosts.client("srv_updated")).rejects.toMatchObject({
code: "unsupported_capability",
});
destination.update();
expect((await online(paseo)).hosts[0]).toMatchObject({
registrationId: host.registrationId,
serverId: "srv_updated",
nextRetryAt: null,
error: null,
});
await paseo.hosts.add({ connection: `ws://127.0.0.1:${destination.port}/ws` });
await expect
.poll(async () => (await paseo.hosts.list()).hosts[1], { timeout: 5000 })
.toMatchObject({ status: "error", nextRetryAt: null, error: { code: "duplicate_host" } });
});
test("disabling or removing a host cancels its pending capability retry", async () => {
const destination = await updatableDestination();
const a = await daemon();
const { paseo } = await api(a);
const host = await paseo.hosts.add({ connection: `127.0.0.1:${destination.port}` });
let retryAt = await capabilityRetry(paseo);
let connections = destination.connections();
await paseo.hosts.update(host.registrationId, { enabled: false });
await new Promise((resolve) => setTimeout(resolve, retryAt - Date.now() + 500));
expect(destination.connections()).toBe(connections);
await paseo.hosts.update(host.registrationId, { enabled: true });
retryAt = await capabilityRetry(paseo);
connections = destination.connections();
await paseo.hosts.remove(host.registrationId);
await new Promise((resolve) => setTimeout(resolve, retryAt - Date.now() + 500));
expect(destination.connections()).toBe(connections);
});
import {
createDaemonChannel,
exportPublicKey,
generateKeyPair,
openDaemonSecureChannel,
type Transport,
} from "@getpaseo/relay/e2ee";
test("a capability retry over a relay keeps the protected endpoint security it negotiated", async () => {
const keys = generateKeyPair();
const legacyReceived: unknown[] = [];
let connections = 0;
let retryClosed = false;
const http = createServer();
const ws = new WebSocketServer({ server: http });
ws.on("connection", (socket) => {
connections++;
const wire: Transport = {
send: (data) => socket.send(data),
close: (code, reason) => socket.close(code, reason),
onmessage: null,
onclose: null,
onerror: null,
};
socket.on("message", (data, isBinary) => {
const bytes = Buffer.from(data as Buffer);
wire.onmessage?.({
data: isBinary
? bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength)
: bytes.toString(),
isBinary,
});
});
socket.on("close", (code, reason) => wire.onclose?.(code, reason.toString()));
if (connections > 1) {
socket.on("close", () => {
retryClosed = true;
});
void createDaemonChannel(wire, keys, {
onmessage: (data) => legacyReceived.push(data),
}).catch(() => {});
return;
}
const channel = openDaemonSecureChannel(wire, {
daemonKeyPair: keys,
events: {
onmessage: (data) => {
if (typeof data !== "string" || JSON.parse(data).type !== "hello") return;
void channel.then((secure) =>
secure.send(
JSON.stringify({
type: "session",
message: {
type: "status",
payload: {
status: "server_info",
serverId: "srv_relayed",
hostname: "relayed",
version: "0.12.0",
},
},
}),
),
);
},
},
});
void channel.catch(() => {});
});
http.listen(0, "127.0.0.1");
await once(http, "listening");
cleanup.push(async () => {
for (const socket of ws.clients) socket.terminate();
await new Promise<void>((resolve) => ws.close(() => resolve()));
await new Promise<void>((resolve) => http.close(() => resolve()));
});
const address = http.address();
if (!address || typeof address === "string") throw new Error("Missing relay port");
const a = await daemon();
const { paseo } = await api(a);
await paseo.hosts.add({
connection: `relay://127.0.0.1:${address.port}/srv_relayed?key=${encodeURIComponent(exportPublicKey(keys.publicKey))}`,
});
await capabilityRetry(paseo);
await expect.poll(() => retryClosed, { timeout: 15000 }).toBe(true);
expect(legacyReceived).toEqual([]);
});
test("an unreachable saved target becomes online after its daemon starts without being added again", async () => {
const reservation = createServer();
reservation.listen(0, "127.0.0.1");
@@ -7,12 +7,10 @@ import { z } from "zod";
import {
createHostClient,
createHostChannelTransport,
createHostTransportFactory,
parseHostTarget,
HostTransportBudget,
} from "@getpaseo/client/node";
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types";
import {
HostSummarySchema,
HostAddSchema,
@@ -237,30 +235,23 @@ export class ConnectedHosts {
if (route.registrationId === id)
route.close(code, "Host access was disabled, replaced or removed");
}
private start(
row: Saved,
resume?: { failures: number; transports: DaemonTransportFactory },
): void {
private start(row: Saved): void {
if (!row.enabled || this.closed) return;
const transports =
resume?.transports ??
createHostTransportFactory(row.connection, this.budget, {
webSocketAgent: webSocketProxyAgent,
});
let stopped = false;
let attempts = resume?.failures ?? 0;
let retry: ReturnType<typeof setTimeout> | undefined;
const client = createHostClient(row.connection, {
clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
connectTimeoutMs: 10000,
transportFactory: transports,
});
const client = createHostClient(
row.connection,
{
clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
connectTimeoutMs: 10000,
webSocketAgent: webSocketProxyAgent,
},
this.budget,
);
const live: Live = {
client,
accepted: false,
stop: () => {
stopped = true;
clearTimeout(retry);
void client.close();
},
};
@@ -294,7 +285,7 @@ export class ConnectedHosts {
if (info.features?.connectedHosts !== true || info.features?.hostMultiplexing !== true)
throw new HostError(
"unsupported_capability",
"Update the destination daemon to use connected hosts; retrying automatically.",
"Update the destination daemon to use connected hosts.",
);
// Pin and persist under the same writer as management to prevent discovery races.
await this.mutate(async () => {
@@ -313,22 +304,16 @@ export class ConnectedHosts {
if (stopped || client.getLastServerInfoMessage() !== info || !client.isConnected) return;
Object.assign(row, next);
live.accepted = true;
attempts = 0;
this.changed();
});
} catch (error) {
if (stopped) return;
row.status = "error";
row.error = connectionFailure(error, client.authFailureReason !== null) ?? {
code: "host_offline",
message: "Host connection failed",
};
if (terminalFailures.has(row.error.code)) {
row.status = "error";
row.nextRetryAt = null;
} else {
stopped = true;
retry = this.scheduleAcceptanceRetry(row, attempts, transports);
}
row.nextRetryAt = null;
this.changed();
await client.close();
}
@@ -344,17 +329,18 @@ export class ConnectedHosts {
if (route.registrationId === row.registrationId) route.close();
row.status = state.status === "connecting" ? "connecting" : "offline";
row.nextRetryAt = state.status === "disconnected" ? (state.nextRetryAt ?? null) : null;
if (state.status === "disconnected")
row.error = connectionFailure(
Object.assign(new Error("Check the SSH configuration and daemon credentials"), {
code: state.failureCode,
}),
client.authFailureReason !== null,
) ?? {
code: "host_offline",
message: "Cannot reach host. Check the connection; retrying automatically.",
};
else if (row.error?.code !== "unsupported_capability") row.error = null;
row.error =
state.status === "disconnected"
? (connectionFailure(
Object.assign(new Error("Check the SSH configuration and daemon credentials"), {
code: state.failureCode,
}),
client.authFailureReason !== null,
) ?? {
code: "host_offline",
message: "Cannot reach host. Check the connection; retrying automatically.",
})
: null;
if (row.error?.code === "authentication_failed") {
row.status = "error";
row.nextRetryAt = null;
@@ -365,18 +351,6 @@ export class ConnectedHosts {
});
void client.connect().catch(() => {});
}
private scheduleAcceptanceRetry(
row: Saved,
failures: number,
transports: DaemonTransportFactory,
): ReturnType<typeof setTimeout> {
const delay = acceptanceRetryDelayMs(failures);
row.status = "offline";
row.nextRetryAt = new Date(Date.now() + delay).toISOString();
const retry = setTimeout(() => this.start(row, { failures: failures + 1, transports }), delay);
retry.unref();
return retry;
}
async openRoute(input: {
serverId: string;
scopeId: string;
@@ -557,20 +531,6 @@ export class ConnectedHosts {
}
}
const ACCEPTANCE_RETRY_BASE_MS = 1500;
const ACCEPTANCE_RETRY_MAX_MS = 30_000;
function acceptanceRetryDelayMs(failures: number): number {
return Math.min(ACCEPTANCE_RETRY_BASE_MS * 2 ** failures, ACCEPTANCE_RETRY_MAX_MS);
}
const terminalFailures = new Set([
"server_id_mismatch",
"self_host",
"duplicate_host",
"authentication_failed",
]);
function connectionFailure(error: unknown, authenticationFailed: boolean): HostSummary["error"] {
if (error instanceof HostError) return { code: error.code, message: error.message };
if (error instanceof Error && "code" in error && error.code === "authentication_failed")
@@ -88,9 +88,13 @@ test.each(["uncommitted", "base", "commit"] as const)(
60_000,
);
test.each(["single", "batch", "rename"] as const)(
"%s diff includes hunks from a repository subdirectory before and after commit",
async (shape) => {
test.each(
(["uncommitted", "base"] as const).flatMap((mode) =>
(["single", "batch", "rename"] as const).map((shape) => ({ mode, shape })),
),
)(
"$mode $shape diff includes hunks from a repository subdirectory",
async ({ mode, shape }) => {
const ctx = await createDaemonTestContext();
const cwd = mkdtempSync(join(tmpdir(), "checkout-diff-subdirectory-"));
@@ -108,7 +112,9 @@ test.each(["single", "batch", "rename"] as const)(
}
git(cwd, "add", ".");
git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "initial");
git(cwd, "checkout", "-b", "feature");
if (mode === "base") {
git(cwd, "checkout", "-b", "feature");
}
const filename = shape === "rename" ? "renamed.txt" : "normal.txt";
if (shape === "rename") {
git(cwd, "mv", "sub/normal.txt", `sub/${filename}`);
@@ -117,38 +123,36 @@ test.each(["single", "batch", "rename"] as const)(
if (shape === "batch") {
writeFileSync(join(cwd, "sibling.txt"), "new sibling\n");
}
async function expectSubdirectoryDiff(mode: "uncommitted" | "base") {
const compare = { mode, baseRef: "main" };
const rootDiff = await ctx.client.getCheckoutDiff(cwd, compare);
const subdirectoryDiff = await ctx.client.getCheckoutDiff(subdirectory, compare);
const subscription = ctx.client.observeCheckoutDiff(subdirectory, compare);
const liveDiff = await subscription.ready;
await subscription.release();
expect(rootDiff.error).toBeNull();
expect(subdirectoryDiff.error).toBeNull();
expect(liveDiff.error).toBeNull();
expect(liveDiff.files).toEqual(rootDiff.files);
expect(rootDiff.files).toHaveLength(shape === "batch" ? 2 : 1);
const file = rootDiff.files.find((candidate) => candidate.path === `sub/${filename}`);
expect(file).toMatchObject({
additions: 1,
deletions: 1,
...(shape === "rename" ? { oldPath: "sub/normal.txt" } : {}),
});
expect(file?.hunks.flatMap((hunk) => hunk.lines)).toEqual(
expect.arrayContaining([
expect.objectContaining({ type: "remove", content: "before" }),
expect.objectContaining({ type: "add", content: "after" }),
]),
);
expect(subdirectoryDiff.files).toEqual(rootDiff.files);
if (mode === "base") {
git(cwd, "add", ".");
git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "edit");
}
await expectSubdirectoryDiff("uncommitted");
git(cwd, "add", ".");
git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "edit");
await expectSubdirectoryDiff("base");
const compare = { mode, baseRef: "main" };
const rootDiff = await ctx.client.getCheckoutDiff(cwd, compare);
const subdirectoryDiff = await ctx.client.getCheckoutDiff(subdirectory, compare);
const subscription = ctx.client.observeCheckoutDiff(subdirectory, compare);
const liveDiff = await subscription.ready;
await subscription.release();
expect(rootDiff.error).toBeNull();
expect(subdirectoryDiff.error).toBeNull();
expect(liveDiff.error).toBeNull();
expect(liveDiff.files).toEqual(rootDiff.files);
expect(rootDiff.files).toHaveLength(shape === "batch" ? 2 : 1);
const file = rootDiff.files.find((candidate) => candidate.path === `sub/${filename}`);
expect(file).toMatchObject({
additions: 1,
deletions: 1,
...(shape === "rename" ? { oldPath: "sub/normal.txt" } : {}),
});
expect(file?.hunks.flatMap((hunk) => hunk.lines)).toEqual(
expect.arrayContaining([
expect.objectContaining({ type: "remove", content: "before" }),
expect.objectContaining({ type: "add", content: "after" }),
]),
);
expect(subdirectoryDiff.files).toEqual(rootDiff.files);
} finally {
await ctx.cleanup();
rmSync(cwd, { recursive: true, force: true });
@@ -20,7 +20,6 @@ import { DaemonConfigStore } from "../daemon-config-store.js";
import { PluginService } from "./index.js";
import { BuiltinPluginLoader } from "./builtin/index.js";
import { ManagedPluginSources } from "./managed-source.js";
import type { PluginProviderMetadata } from "./plugin-process-protocol.js";
import {
startNpmRegistry,
npmPluginPackages,
@@ -180,55 +179,6 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) {
return { runtime, started, releaseStart, starts };
}
function createGatedRuntime(providers: Record<string, readonly PluginProviderMetadata[]> = {}) {
const gates = new Map<string, PromiseWithResolvers<void>>();
const gate = (pluginId: string) => {
const existing = gates.get(pluginId);
if (existing) return existing;
const created = Promise.withResolvers<void>();
gates.set(pluginId, created);
return created;
};
const begun: string[] = [];
const running = new Set<string>();
const runtime: TestPluginRuntime = {
...emptyUsageRuntime,
catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })),
invoke: async () => undefined,
getLogs: () => [],
clearLogs: () => undefined,
getProviderRegistrations: (pluginId) =>
running.has(pluginId) ? (providers[pluginId] ?? []) : [],
startPlugin: async (pluginId) => {
begun.push(pluginId);
await gate(pluginId).promise;
running.add(pluginId);
},
stopPluginById: async (pluginId) => running.delete(pluginId),
stopAll: async () => {
running.clear();
},
subscribe: () => () => undefined,
bindPaseoSessionHost: () => undefined,
};
return {
runtime,
begun,
finish: (pluginId: string) => gate(pluginId).resolve(),
fail: (pluginId: string, error: Error) => gate(pluginId).reject(error),
};
}
function nextNotification(service: PluginService, pluginId: string): Promise<void> {
return new Promise((resolve) => {
const unsubscribe = service.subscribe((notified) => {
if (notified !== pluginId) return;
unsubscribe();
resolve();
});
});
}
describe("PluginService", () => {
it.each([
{ shadow: false, runtime: "subprocess" },
@@ -422,127 +372,15 @@ describe("PluginService", () => {
"startup-first": { source: "directory", path: first },
"startup-second": { source: "directory", path: second },
});
const notifications: { pluginId: string; running: boolean }[] = [];
service.subscribe((pluginId) =>
notifications.push({ pluginId, running: catalogIds(service).includes(pluginId) }),
);
const snapshots: string[][] = [];
service.subscribe(() => snapshots.push(catalogIds(service)));
await service.start();
expect(notifications).toEqual([
{ pluginId: "startup-first", running: true },
{ pluginId: "startup-second", running: true },
]);
expect(catalogIds(service)).toEqual(["startup-first", "startup-second"]);
expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]);
await service.stopAllPlugins();
}, 20_000);
it("starts configured plugins concurrently and publishes them in config order", async () => {
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
roots.push(home);
const gated = createGatedRuntime();
const service = createService(
home,
{
alpha: { source: "directory", path: "/plugins/alpha" },
beta: { source: "directory", path: "/plugins/beta" },
gamma: { source: "directory", path: "/plugins/gamma" },
},
{ runtime: gated.runtime },
);
const notified: string[] = [];
service.subscribe((pluginId) => notified.push(pluginId));
let started = false;
const starting = (async () => {
await service.start();
started = true;
})();
await expect.poll(() => gated.begun).toEqual(["alpha", "beta", "gamma"]);
gated.finish("gamma");
const alphaPublished = nextNotification(service, "alpha");
gated.finish("alpha");
await alphaPublished;
expect(notified).toEqual(["alpha"]);
expect(started).toBe(false);
gated.finish("beta");
await starting;
expect(notified).toEqual(["alpha", "beta", "gamma"]);
expect(catalogIds(service)).toEqual(["alpha", "beta", "gamma"]);
await service.stopAllPlugins();
});
it("starts the other configured plugins when one fails while they are still starting", async () => {
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
roots.push(home);
const gated = createGatedRuntime();
const service = createService(
home,
{
broken: { source: "directory", path: "/plugins/broken" },
healthy: { source: "directory", path: "/plugins/healthy" },
},
{ runtime: gated.runtime },
);
const notified: string[] = [];
service.subscribe((pluginId) => notified.push(pluginId));
const starting = service.start();
await expect.poll(() => gated.begun).toEqual(["broken", "healthy"]);
const brokenPublished = nextNotification(service, "broken");
gated.fail("broken", new Error("startup exploded"));
await brokenPublished;
gated.finish("healthy");
await starting;
expect(notified).toEqual(["broken", "healthy"]);
expect(catalogIds(service)).toEqual(["healthy"]);
expect(await service.listPlugins()).toMatchObject([
{ id: "broken", status: "failed", error: "startup exploded" },
{ id: "healthy", status: "running" },
]);
await service.stopAllPlugins();
});
it("keeps a provider ID with the plugin listed first when both start at once", async () => {
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
roots.push(home);
const gated = createGatedRuntime({
first: [{ id: "shared-agent", label: "First agent" }],
second: [{ id: "shared-agent", label: "Second agent" }],
});
const service = createService(
home,
{
first: { source: "directory", path: "/plugins/first" },
second: { source: "directory", path: "/plugins/second" },
},
{ runtime: gated.runtime },
);
const starting = service.start();
await expect.poll(() => gated.begun).toEqual(["first", "second"]);
gated.finish("second");
gated.finish("first");
await starting;
expect(service.getProviderRegistrations()).toMatchObject([
{ id: "shared-agent", label: "First agent" },
]);
expect(catalogIds(service)).toEqual(["first"]);
expect(await service.listPlugins()).toMatchObject([
{ id: "first", status: "running" },
{
id: "second",
status: "failed",
error: 'Plugin second cannot register provider ID "shared-agent" twice',
},
]);
await service.stopAllPlugins();
});
it("uses an explicit config key, exposes reload failure, and retries from disk", async () => {
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
roots.push(home);
+8 -20
View File
@@ -230,18 +230,11 @@ export class PluginService {
}
});
if (config.pluginsEnabled === true) {
let earlierPublished: Promise<unknown> = Promise.resolve();
await Promise.all(
Object.entries(config.plugins ?? {})
.filter(([, source]) => source.enabled !== false)
.map(async ([pluginId]) => {
const publishTurn = earlierPublished;
const starting = this.startConfigured(pluginId, publishTurn);
earlierPublished = Promise.all([publishTurn, starting]);
await starting;
this.notify(pluginId);
}),
);
for (const [pluginId, source] of Object.entries(config.plugins ?? {})) {
if (source.enabled === false) continue;
await this.startConfigured(pluginId);
this.notify(pluginId);
}
}
this.configStore.onFieldChange("pluginsEnabled", (value) => {
this.handleGlobalSwitch(value === true);
@@ -567,12 +560,12 @@ export class PluginService {
}).catch((error) => this.logger.error({ err: error }, "Failed to enable plugins"));
}
private async startConfigured(pluginId: string, publishTurn?: Promise<unknown>): Promise<void> {
private async startConfigured(pluginId: string): Promise<void> {
const source = this.configStore.get().plugins?.[pluginId];
if (!source || source.enabled === false || !this.canPublish(pluginId)) return;
this.errors.delete(pluginId);
try {
await this.startPlugin(pluginId, source.path, publishTurn);
await this.startPlugin(pluginId, source.path);
} catch (error) {
if (this.canPublish(pluginId)) this.recordFailure(pluginId, error);
}
@@ -601,13 +594,8 @@ export class PluginService {
);
}
private async startPlugin(
pluginId: string,
sourcePath: string,
publishTurn?: Promise<unknown>,
): Promise<void> {
private async startPlugin(pluginId: string, sourcePath: string): Promise<void> {
await this.runtime.startPlugin(pluginId, sourcePath, () => this.canPublish(pluginId));
await publishTurn;
try {
await this.publishProviderRegistrations(pluginId, sourcePath);
this.publishUsageSources(pluginId);
@@ -61,24 +61,10 @@ describe("managed Git plugin sources", () => {
roots.push(home);
const sources = new ManagedPluginSources(home);
// Own the failure response instead of relying on a closed port and OS connect timing.
const server = createServer((_request, response) => {
response.writeHead(500, { "Content-Type": "text/plain" });
response.end("clone rejected for oauth2:super-secret\n");
const failure = sources.prepareInstall({
source: "git:https://oauth2:super-secret@127.0.0.1:1/missing.git",
});
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
const address = server.address() as { port: number };
try {
const failure = sources.prepareInstall({
source: `git:http://oauth2:super-secret@127.0.0.1:${address.port}/missing.git`,
});
await expect(failure).rejects.toThrow("500");
await expect(failure).rejects.toThrow("clone rejected for [redacted]:[redacted]");
await expect(failure).rejects.not.toThrow(/oauth2|super-secret/);
} finally {
server.closeAllConnections();
await new Promise<void>((resolve) => server.close(() => resolve()));
}
await expect(failure).rejects.not.toThrow(/oauth2|super-secret/);
});
it("normalizes explicit GitHub and Git identifiers to the same Git source", async () => {
+31 -246
View File
@@ -1,9 +1,7 @@
import { type EventEmitter, once } from "node:events";
import { readdirSync, readFileSync, statSync } from "node:fs";
import path from "node:path";
import http from "node:http";
import net from "node:net";
import type { Duplex } from "node:stream";
import express from "express";
import { describe, expect, it } from "vitest";
import pino from "pino";
@@ -334,39 +332,48 @@ describe("service proxy subsystem shape", () => {
});
});
interface ProxiedServiceFixture {
interface ForwardedFixture {
upstreamPort: number;
daemonPort: number;
daemon: http.Server;
hostname: string;
close(): Promise<void>;
}
/**
* Runs a real workspace service behind a real daemon listener that serves the
* service proxy for both HTTP and upgrades.
* Runs a real workspace service behind a real daemon listener. The upstream
* echoes the headers it received so tests can assert what actually crossed the
* proxy, not what a helper returned.
*/
async function startProxiedService(
upstream: http.Server,
options: { upstreamHost?: string; logger?: pino.Logger } = {},
): Promise<ProxiedServiceFixture> {
const { upstreamHost = "127.0.0.1", logger: proxyLogger = logger } = options;
// Upgraded sockets keep the server alive; close() alone would hang.
const sockets = new Set<net.Socket>();
function trackConnections(server: http.Server) {
server.on("connection", (socket: net.Socket) => {
sockets.add(socket);
socket.once("close", () => sockets.delete(socket));
async function startForwardedHeadersFixture(upstreamHost = "127.0.0.1"): Promise<ForwardedFixture> {
const upstream = http.createServer((req, res) => {
let body = "";
req.on("data", (chunk: Buffer) => {
body += chunk.toString();
});
}
trackConnections(upstream);
req.on("end", () => {
res.writeHead(200, { "content-type": "application/json" });
res.end(JSON.stringify({ ...req.headers, method: req.method, url: req.url, body }));
});
});
const upgradeSockets: net.Socket[] = [];
upstream.on("upgrade", (req, socket) => {
upgradeSockets.push(socket);
// The client resets this connection once it has the echo, so the reset
// reaches us as an 'error'. A raw upgrade socket has no default handler —
// without this the event goes unhandled and takes down the test process.
socket.on("error", () => socket.destroy());
const payload = JSON.stringify(req.headers);
socket.write(
`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nX-Echo-Length: ${payload.length}\r\n\r\n${payload}`,
);
});
await new Promise<void>((resolve, reject) => {
upstream.once("error", reject);
upstream.listen(0, upstreamHost, resolve);
});
const upstreamPort = (upstream.address() as net.AddressInfo).port;
const serviceProxy = createServiceProxySubsystem({ logger: proxyLogger });
const serviceProxy = createServiceProxySubsystem({ logger });
const route = serviceProxy.registerWorkspaceService({
workspaceId: "workspace-a",
projectSlug: "repo",
@@ -383,51 +390,24 @@ async function startProxiedService(
res.status(404).send("404 Not Found");
});
const daemon = http.createServer(app);
trackConnections(daemon);
daemon.on("upgrade", serviceProxy.upgradeHandler({ passthroughUnknown: false }));
await new Promise<void>((resolve) => daemon.listen(daemonPort, "127.0.0.1", resolve));
return {
upstreamPort,
daemonPort,
daemon,
hostname: route.hostname,
async close() {
for (const socket of sockets) socket.destroy();
// Upgraded sockets keep the server alive; close() alone would hang.
for (const socket of upgradeSockets) socket.destroy();
daemon.closeAllConnections();
await new Promise<void>((resolve) => daemon.close(() => resolve()));
upstream.closeAllConnections();
await new Promise<void>((resolve) => upstream.close(() => resolve()));
},
};
}
/**
* The upstream echoes the headers it received so tests can assert what
* actually crossed the proxy, not what a helper returned.
*/
function startForwardedHeadersFixture(upstreamHost = "127.0.0.1"): Promise<ProxiedServiceFixture> {
const upstream = http.createServer((req, res) => {
let body = "";
req.on("data", (chunk: Buffer) => {
body += chunk.toString();
});
req.on("end", () => {
res.writeHead(200, { "content-type": "application/json" });
res.end(JSON.stringify({ ...req.headers, method: req.method, url: req.url, body }));
});
});
upstream.on("upgrade", (req, socket) => {
// The client resets this connection once it has the echo, so the reset
// reaches us as an 'error'. A raw upgrade socket has no default handler —
// without this the event goes unhandled and takes down the test process.
socket.on("error", () => socket.destroy());
const payload = JSON.stringify(req.headers);
socket.write(
`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nX-Echo-Length: ${payload.length}\r\n\r\n${payload}`,
);
});
return startProxiedService(upstream, { upstreamHost });
}
function upgradeThroughProxy(
port: number,
host: string,
@@ -671,198 +651,3 @@ describe("service proxy forwarded headers", () => {
}
});
});
function closed(emitter: EventEmitter): Promise<void> {
return new Promise((resolve) => {
emitter.once("close", () => resolve());
});
}
async function within<T>(promise: Promise<T>, ms: number, failure: string): Promise<T> {
let timer: NodeJS.Timeout | undefined;
const timeout = new Promise<never>((_resolve, reject) => {
timer = setTimeout(() => reject(new Error(`${failure} after ${ms}ms`)), ms);
});
try {
return await Promise.race([promise, timeout]);
} finally {
clearTimeout(timer);
}
}
function requestThroughProxy(
fixture: ProxiedServiceFixture,
requestPath: string,
): http.ClientRequest {
return http.get({
hostname: "127.0.0.1",
port: fixture.daemonPort,
path: requestPath,
headers: { host: fixture.hostname },
});
}
const STRAY_WARNING_GRACE_MS = 100;
function createUpgradeUpstream(onUpgrade: (socket: Duplex) => void): http.Server {
const upstream = http.createServer();
upstream.on("upgrade", (_req, socket) => {
socket.on("error", () => socket.destroy());
socket.on("end", () => socket.end());
socket.write(
"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n",
);
onUpgrade(socket);
});
return upstream;
}
function openUpgradeThroughProxy(fixture: ProxiedServiceFixture): Promise<net.Socket> {
return new Promise((resolve, reject) => {
const socket = net.connect({ host: "127.0.0.1", port: fixture.daemonPort }, () => {
const lines = [
"GET /ws HTTP/1.1",
`Host: ${fixture.hostname}`,
"Upgrade: websocket",
"Connection: Upgrade",
"",
"",
];
socket.write(lines.join("\r\n"));
});
let raw = "";
socket.on("data", (chunk: Buffer) => {
raw += chunk.toString();
if (raw.includes("\r\n\r\n")) resolve(socket);
});
socket.on("close", () => reject(new Error("upgrade closed before the 101 arrived")));
socket.on("error", reject);
});
}
describe("service proxy connection teardown", () => {
it("closes the upstream connection when a client leaves an event stream", async () => {
const upstreamResponse = Promise.withResolvers<http.ServerResponse>();
const upstream = http.createServer((_req, res) => {
res.writeHead(200, { "content-type": "text/event-stream" });
res.write("data: open\n\n");
upstreamResponse.resolve(res);
});
const fixture = await startProxiedService(upstream);
try {
const request = requestThroughProxy(fixture, "/events");
const [response] = (await once(request, "response")) as [http.IncomingMessage];
await once(response, "data");
const upstreamClosed = closed(await upstreamResponse.promise);
request.destroy();
await within(upstreamClosed, 1000, "the service still holds the event stream");
} finally {
await fixture.close();
}
}, 5000);
it("aborts the upstream request when a client leaves before the response starts", async () => {
const warnings: unknown[] = [];
const proxyLogger = pino(
{ level: "warn" },
{ write: (line: string) => warnings.push(JSON.parse(line)) },
);
const upstreamResponse = Promise.withResolvers<http.ServerResponse>();
const upstream = http.createServer((_req, res) => {
upstreamResponse.resolve(res);
});
const fixture = await startProxiedService(upstream, { logger: proxyLogger });
try {
const request = requestThroughProxy(fixture, "/slow");
request.on("error", () => undefined);
const upstreamClosed = closed(await upstreamResponse.promise);
request.destroy();
await within(upstreamClosed, 1000, "the service still holds the abandoned request");
await new Promise((resolve) => setTimeout(resolve, STRAY_WARNING_GRACE_MS));
expect(warnings).toEqual([]);
} finally {
await fixture.close();
}
}, 5000);
it("lets a complete response finish without resetting the service", async () => {
const upstreamResponse = Promise.withResolvers<http.ServerResponse>();
const upstream = http.createServer((_req, res) => {
res.writeHead(200, { "content-type": "text/plain" });
res.write("first;");
upstreamResponse.resolve(res);
});
const fixture = await startProxiedService(upstream);
try {
const response = httpGet(fixture.daemonPort, fixture.hostname, { path: "/" });
const res = await upstreamResponse.promise;
const upstreamClosed = closed(res);
res.end("second");
await expect(response).resolves.toEqual({ status: 200, body: "first;second" });
await within(upstreamClosed, 1000, "the service response never closed");
expect(res.writableFinished).toBe(true);
} finally {
await fixture.close();
}
}, 5000);
it("closes the upstream WebSocket when the daemon drops the client socket", async () => {
const upstreamSocket = Promise.withResolvers<Duplex>();
const fixture = await startProxiedService(
createUpgradeUpstream((socket) => upstreamSocket.resolve(socket)),
);
const daemonSide = Promise.withResolvers<Duplex>();
fixture.daemon.on("upgrade", (_req, socket) => daemonSide.resolve(socket));
try {
const client = await openUpgradeThroughProxy(fixture);
const upstreamClosed = closed(await upstreamSocket.promise);
(await daemonSide.promise).destroy();
await within(upstreamClosed, 1000, "the service still holds the WebSocket");
client.destroy();
} finally {
await fixture.close();
}
}, 5000);
it("closes the upstream WebSocket when the client disconnects", async () => {
const upstreamSocket = Promise.withResolvers<Duplex>();
const fixture = await startProxiedService(
createUpgradeUpstream((socket) => upstreamSocket.resolve(socket)),
);
try {
const client = await openUpgradeThroughProxy(fixture);
const upstreamClosed = closed(await upstreamSocket.promise);
client.destroy();
await within(upstreamClosed, 1000, "the service still holds the WebSocket");
} finally {
await fixture.close();
}
}, 5000);
it("closes the client WebSocket when the service closes it", async () => {
const upstreamSocket = Promise.withResolvers<Duplex>();
const fixture = await startProxiedService(
createUpgradeUpstream((socket) => upstreamSocket.resolve(socket)),
);
try {
const client = await openUpgradeThroughProxy(fixture);
const clientClosed = closed(client);
(await upstreamSocket.promise).destroy();
await within(clientClosed, 1000, "the client WebSocket is still open");
} finally {
await fixture.close();
}
}, 5000);
});
@@ -342,14 +342,7 @@ function proxyHttpRequest({
proxyRes.pipe(res, { end: true });
},
);
let clientClosed = false;
res.on("close", () => {
if (res.writableFinished) return;
clientClosed = true;
proxyReq.destroy();
});
proxyReq.on("error", (err) => {
if (clientClosed) return;
logger.warn(
{ err, hostname: route.hostname, port: route.port },
"Service proxy: upstream unreachable",
@@ -406,9 +399,6 @@ function proxyUpgradeRequest({
socket.on("error", () => {
targetSocket.destroy();
});
socket.on("close", () => {
targetSocket.destroy();
});
}
function sameRouteOwner(left: ServiceProxyRouteEntry, right: ServiceProxyRouteEntry): boolean {
@@ -95,13 +95,7 @@ it("accepts terminalId and token reports through the route into the tracker", as
command: process.execPath,
args: [
"-e",
`
const fs = require("node:fs");
const envPath = ${JSON.stringify(envPath)};
fs.writeFileSync(envPath + ".tmp", JSON.stringify({ terminalId: process.env.PASEO_TERMINAL_ID, token: process.env.PASEO_ACTIVITY_TOKEN, url: process.env.PASEO_TERMINAL_ACTIVITY_URL }));
fs.renameSync(envPath + ".tmp", envPath);
setInterval(() => {}, 1000);
`,
`require("node:fs").writeFileSync(${JSON.stringify(envPath)}, JSON.stringify({ terminalId: process.env.PASEO_TERMINAL_ID, token: process.env.PASEO_ACTIVITY_TOKEN, url: process.env.PASEO_TERMINAL_ACTIVITY_URL })); setInterval(() => {}, 1000);`,
],
});
await waitForCondition(() => existsSync(envPath), 10000);
@@ -410,12 +410,7 @@ describe("createTerminal", () => {
cwd: realpathSync(tmpdir()),
cols: 500,
command: process.execPath,
// Keep the PTY alive until the screen assertion; exit disposes its parser.
args: [
"-e",
"process.stdout.write(JSON.stringify(process.argv.slice(1))); setInterval(() => {}, 100000)",
nastyPrompt,
],
args: ["-e", "process.stdout.write(JSON.stringify(process.argv.slice(1)))", nastyPrompt],
}),
);
+1 -7
View File
@@ -186,13 +186,7 @@ function insertDocByPath(nodes: DocsNavNode[], doc: Doc, stripTopDirectory = fal
// A section's top-level directory and its category name the same thing, so
// only directories below it become collapsible groups.
if (
stripTopDirectory &&
directories[0] &&
doc.frontmatter.category?.toLowerCase().endsWith(directories[0].toLowerCase())
) {
directories.shift();
}
if (stripTopDirectory) directories.shift();
let current = nodes;
let parent: Extract<DocsNavNode, { type: "group" }> | undefined;
@@ -2,25 +2,6 @@ import { describe, expect, it } from "vitest";
import { buildDocsNavTree, getDoc, getDocs, getLegacyDocsRedirect, parseHeadings } from "./docs";
describe("plugin documentation", () => {
it.each(["en", "zh-CN"] as const)(
"nests local file copying under Git worktrees in %s",
(locale) => {
const workspaces = buildDocsNavTree(getDocs(locale)).find(
(node) => node.type === "category" && node.label === "Workspaces",
);
expect(workspaces).toMatchObject({
children: expect.arrayContaining([
expect.objectContaining({
type: "group",
href: "/docs/worktrees",
children: [expect.objectContaining({ href: "/docs/worktrees/worktreeinclude" })],
}),
]),
});
expect(getDoc("worktrees/worktreeinclude", locale)).toBeDefined();
},
);
it("groups current guides by task without version navigation", () => {
const plugins = buildDocsNavTree(getDocs("en")).find(
(node) => node.type === "category" && node.label === "Plugins",
+1 -1
View File
@@ -4,7 +4,7 @@ import { expect, it } from "vitest";
const docsRoot = path.resolve(import.meta.dirname, "../../../public-docs");
it.each(["schedules.md", "schedules-cli.md", "mcp.md", "worktrees/index.md"])(
it.each(["schedules.md", "schedules-cli.md", "mcp.md", "worktrees.md"])(
"%s makes schedule run retention discoverable",
async (file) => {
const markdown = await readFile(path.join(docsRoot, file), "utf8");
@@ -250,84 +250,6 @@ describe("cursor usage source", () => {
expect(cursor.status).toBe("available");
});
describe.each([
["modern state.vscdb", "v1"],
["modern state.vscdb", "v99"],
["legacy cursorAuthStatus", "v1"],
["auth.json", "v1"],
["CURSOR_ACCESS_TOKEN", "v1"],
["CURSOR_TOKEN", "v1"],
])("encrypted %s (%s)", (store, version) => {
beforeEach(() => {
const encrypted = `cursor-safe-storage:${version}:${Buffer.from("ciphertext").toString("base64")}`;
if (store === "modern state.vscdb") {
writeCursorStateDb(homeDir, { "cursorAuth/accessToken": encrypted });
} else if (store === "legacy cursorAuthStatus") {
writeCursorStateDb(homeDir, {
cursorAuthStatus: Buffer.from(JSON.stringify({ accessToken: encrypted }), "utf8"),
});
} else if (store === "auth.json") {
writeCursorAuthJson(homeDir, encrypted);
} else {
process.env[store] = encrypted;
}
});
it("discovers no account when the only credential is encrypted", async () => {
expect(await discover()).toEqual([]);
});
it("never sends an encrypted bearer token when fetching from a stored locator", async () => {
let input: Parameters<typeof fetchUsage>[0];
if (store === "modern state.vscdb" || store === "legacy cursorAuthStatus") {
input = {
store: "sqlite",
locator: join(homeDir, ".config", "Cursor", "User", "globalStorage", "state.vscdb"),
};
} else if (store === "auth.json") {
input = { store: "file", locator: join(homeDir, ".config", "cursor", "auth.json") };
} else {
input = { store: "env", locator: store };
}
const requests: RequestInit[] = [];
fetchApi = async (_url, init) => {
requests.push(init ?? {});
return new Response(null, { status: 401 });
};
await expect(fetchUsage(input, fetchApi)).rejects.toThrow(
"Cursor login store no longer exists",
);
expect(requests).toEqual([]);
});
});
it("discovers and uses auth.json after skipping an encrypted desktop token", async () => {
writeCursorStateDb(homeDir, {
"cursorAuth/accessToken": `cursor-safe-storage:v1:${Buffer.from("ciphertext").toString("base64")}`,
});
writeCursorAuthJson(homeDir, "cursor_cli_jwt");
expect(await discover()).toEqual([
{
key: expect.stringMatching(/^store-[a-f0-9]{64}$/),
input: { store: "file", locator: join(homeDir, ".config", "cursor", "auth.json") },
},
]);
const authorizations: string[] = [];
fetchApi = async (_url, init) => {
authorizations.push(new Headers(init?.headers).get("Authorization") ?? "");
return jsonResponse({
planUsage: { totalSpend: "0", remaining: "100", limit: "100" },
billingCycleStart: null,
billingCycleEnd: null,
});
};
const cursor = findProvider(await service().listUsage(), "cursor");
expect(authorizations).toEqual(["Bearer cursor_cli_jwt"]);
expect(cursor.status).toBe("available");
});
it("omits unreadable state.vscdb", async () => {
const directory = join(homeDir, ".config", "Cursor", "User", "globalStorage");
mkdirSync(directory, { recursive: true });
+7 -15
View File
@@ -27,11 +27,9 @@ function toIsoStringOrNull(timestampMs: number): string | null {
return Number.isFinite(date.getTime()) ? date.toISOString() : null;
}
// Cursor desktop stores auth in VS Code's ItemTable (state.vscdb). The modern
// `cursorAuth/accessToken` value is a plain JWT or encrypted at rest. Any
// `cursor-safe-storage:` value (any version) is Electron safeStorage ciphertext
// and is skipped. Older builds kept a JSON blob under `cursorAuthStatus`.
// Read it with node:sqlite so we don't
// Cursor desktop stores auth in VS Code's ItemTable (state.vscdb). Modern builds keep
// the access token as a plain JWT string under `cursorAuth/accessToken`; older builds
// kept a JSON blob under `cursorAuthStatus`. Read it with node:sqlite so we don't
// depend on a `sqlite3` CLI, which isn't installed by default on Windows (or on many
// Linux hosts) — a missing binary silently rendered Cursor usage unavailable.
// Headless hosts (VPS, cursor-agent only) have no desktop db; their session lives in
@@ -108,21 +106,15 @@ function readItemTableValue(db: CursorStateDatabase, key: string): string | null
return null;
}
function usableCursorToken(value: string | null | undefined): string | null {
const token = value?.trim();
// Electron safeStorage ciphertext cannot authenticate an API request.
return token && !token.startsWith("cursor-safe-storage:") ? token : null;
}
function cursorTokenFromDb(db: CursorStateDatabase): string | null {
const modern = usableCursorToken(readItemTableValue(db, CURSOR_ACCESS_TOKEN_KEY));
const modern = readItemTableValue(db, CURSOR_ACCESS_TOKEN_KEY)?.trim();
if (modern) return modern;
const legacy = readItemTableValue(db, CURSOR_LEGACY_AUTH_KEY);
if (legacy) {
try {
const parsed = CursorAuthStatusSchema.parse(JSON.parse(legacy));
return usableCursorToken(parsed.accessToken);
if (parsed.accessToken) return parsed.accessToken;
} catch {
// ignore a malformed legacy blob
}
@@ -157,7 +149,7 @@ async function readCursorTokenFromAuthJson(path: string): Promise<string | null>
if (!existsSync(path)) return null;
try {
const parsed = CursorAuthStatusSchema.parse(JSON.parse(await readFile(path, "utf8")));
return usableCursorToken(parsed.accessToken);
return parsed.accessToken?.trim() || null;
} catch {
return null;
}
@@ -238,7 +230,7 @@ export async function fetchUsage(
}
async function readToken(input: UsageInput): Promise<string | undefined> {
if (input.store === "env") return usableCursorToken(process.env[input.locator]) ?? undefined;
if (input.store === "env") return process.env[input.locator];
const token =
input.store === "sqlite"
? await readCursorTokenFromSqlite(input.locator)
@@ -84,7 +84,16 @@ Add `--forge <name>` when Paseo cannot infer the forge from the source checkout.
## Copy local files with .worktreeinclude
Use [`.worktreeinclude`](/docs/worktrees/worktreeinclude) to copy selected gitignored files, such as `.env`, from your source checkout before setup runs.
Create `.worktreeinclude` at the root of your source checkout to copy selected gitignored files into new worktrees:
```gitignore
.env
config/*.local.json
```
Patterns use gitignore syntax. A file must match `.worktreeinclude` and be gitignored to be copied. Tracked files are never copied, and existing destination files are never overwritten. Without `.worktreeinclude`, no local files are copied. Symlinked directories and nested repositories inside an included path cause the copy to fail and abort worktree creation.
The built-in `worktree-include` plugin copies from the checkout you create the worktree from, including when that checkout is another worktree. Copying finishes before setup, terminals, and the first agent. A copy failure aborts creation and removes the new worktree. Copies use copy-on-write clones on APFS and btrfs when supported, with a regular copy elsewhere.
## paseo.json
@@ -82,7 +82,16 @@ paseo workspace create \
## 使用 .worktreeinclude 复制本地文件 {#copy-local-files-with-worktreeinclude}
使用 [`.worktreeinclude`](/docs/worktrees/worktreeinclude),在 setup 运行前将 `.env` 等选定的 Git 忽略文件从源检出复制到新 worktree。
在源检出的根目录创建 `.worktreeinclude`,将选定的 Git 忽略文件复制到新的 worktree:
```gitignore
.env
config/*.local.json
```
模式使用 gitignore 语法。只有同时匹配 `.worktreeinclude` 且被 Git 忽略的文件才会复制。已跟踪的文件不会被复制,目标中已存在的文件不会被覆盖。没有 `.worktreeinclude` 时,不会复制任何本地文件。包含路径中的符号链接目录和嵌套仓库会导致复制失败,并中止 worktree 创建。
内置的 `worktree-include` 插件从你创建 worktree 时所选的检出复制文件,即使源检出本身也是 worktree。复制会在 setup、终端和第一个 Agent 运行之前完成。复制失败会中止创建并删除新的 worktree。支持时会在 APFS 和 btrfs 上使用写时复制克隆,否则执行普通复制。
## paseo.json {#paseojson}
-61
View File
@@ -1,61 +0,0 @@
---
title: .worktreeinclude
description: Copy gitignored local files into new Paseo worktrees.
nav: .worktreeinclude
order: 12
category: Workspaces
---
# .worktreeinclude
Add `.worktreeinclude` at the root of your source checkout to copy local configuration into new [Git worktrees](/docs/worktrees). Paseo copies matching gitignored files before setup, terminals, and the first Agent start.
## Copy your local configuration
For a project whose `.gitignore` includes `.env` and `config/*.local.json`, create this `.worktreeinclude`:
```gitignore
.env
config/*.local.json
```
When you create a worktree through Paseo, an existing `.env` and matching files such as `config/dev.local.json` are copied to the same relative paths in the new worktree. Their contents stay out of Git.
Commit `.worktreeinclude` to share the patterns with your team. Each teammate supplies their own ignored files in the checkout they create worktrees from.
## Pattern syntax
Use gitignore syntax, with one pattern per line. Comments start with `#`. Prefix a pattern with `!` to exclude a match:
```gitignore
# Local environment and configuration
.env
config/*.local.json
!config/production.local.json
```
A pattern only selects files that Git already ignores. Adding a path here does not ignore it, so also configure `.gitignore` when needed. Tracked files and untracked files that Git does not ignore are excluded.
## When files are copied
The built-in `worktree-include` plugin reads the file from the checkout you create the worktree from. That source can itself be another worktree. Keep the plugin enabled for copying to run.
- Without `.worktreeinclude`, the plugin copies nothing.
- Matching files must exist in the source checkout.
- Existing destination files are skipped, preserving their contents.
- Copies happen during creation, with no ongoing synchronization between checkouts.
- A copy error aborts creation and removes the new worktree. Avoid including symlinked directories or nested repositories, which cannot be copied as files.
## Run commands during setup
Use `.worktreeinclude` for files that already exist locally. Use a [setup hook](/docs/worktrees#setup-and-teardown) to install dependencies, generate files, or fetch secrets:
```json
{
"worktree": {
"setup": "npm ci"
}
}
```
Put this configuration in `paseo.json` on the base branch used for the new worktree. Copied files are available when setup runs.
@@ -1,61 +0,0 @@
---
title: .worktreeinclude
description: 将 Git 忽略的本地文件复制到新的 Paseo worktree。
nav: .worktreeinclude
order: 12
category: Workspaces
---
# .worktreeinclude
在源检出的根目录添加 `.worktreeinclude`,将本地配置复制到新的 [Git worktree](/docs/worktrees)。Paseo 会在 setup、终端和首个 Agent 启动之前复制匹配的 Git 忽略文件。
## 复制本地配置 {#copy-your-local-configuration}
如果项目的 `.gitignore` 包含 `.env` 和 `config/*.local.json`,创建以下 `.worktreeinclude`:
```gitignore
.env
config/*.local.json
```
通过 Paseo 创建 worktree 时,已有的 `.env` 和 `config/dev.local.json` 等匹配文件会被复制到新 worktree 中的相同相对路径。文件内容仍不会纳入 Git。
提交 `.worktreeinclude`,与团队共享这些模式。每位团队成员在用于创建 worktree 的检出中提供自己的忽略文件。
## 模式语法 {#pattern-syntax}
使用 gitignore 语法,每行一个模式。注释以 `#` 开头。模式前加 `!` 可排除匹配项:
```gitignore
# Local environment and configuration
.env
config/*.local.json
!config/production.local.json
```
模式只会选择 Git 已忽略的文件。在此添加路径不会使 Git 忽略该文件,需要时还应配置 `.gitignore`。已跟踪的文件和未被 Git 忽略的未跟踪文件都不会被复制。
## 文件复制时机 {#when-files-are-copied}
内置的 `worktree-include` 插件从用于创建 worktree 的检出读取该文件。源检出也可以是另一个 worktree。保持该插件启用才能执行复制。
- 没有 `.worktreeinclude` 时,插件不会复制任何文件。
- 匹配的文件必须存在于源检出中。
- 目标中已存在的文件会被跳过,保留原有内容。
- 复制发生在创建期间,之后不会在检出之间持续同步。
- 复制错误会中止创建并删除新 worktree。避免包含符号链接目录或嵌套仓库,因为它们无法作为文件复制。
## 在 setup 期间运行命令 {#run-commands-during-setup}
使用 `.worktreeinclude` 复制本地已有文件。使用 [setup 钩子](/docs/worktrees#setup-and-teardown) 安装依赖、生成文件或获取密钥:
```json
{
"worktree": {
"setup": "npm ci"
}
}
```
将此配置放入新 worktree 所用基准分支的 `paseo.json` 中。setup 运行时,复制的文件已可使用。
+1 -46
View File
@@ -1,5 +1,5 @@
import assert from "node:assert/strict";
import { readFileSync, readdirSync, mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs";
import { readFileSync, readdirSync, mkdtempSync, writeFileSync, rmSync } from "node:fs";
import { spawnSync } from "node:child_process";
import { tmpdir } from "node:os";
import { join } from "node:path";
@@ -679,48 +679,3 @@ test("macOS jobs use latest runners while preserving release architectures", ()
// GitHub has no standard macos-latest-intel alias; use the newest Intel image.
assert.match(release, /runner: macos-26-intel\n\s+electron_arch: x64/);
});
test("Nix desktop packaging resolves the workspace-local builder", () => {
const source = readFileSync(new URL("nix/desktop-package.nix", repoRoot), "utf8");
const invocation = source.match(/^\s+([^\n]*electron-builder) \\\n/m)?.[1];
assert.ok(invocation, "Nix invokes the desktop packaging tool");
const directory = mkdtempSync(join(tmpdir(), "nix-builder-resolution-"));
try {
const desktop = join(directory, "packages/desktop");
mkdirSync(join(desktop, "node_modules/.bin"), { recursive: true });
mkdirSync(join(desktop, "node_modules/electron-builder"));
writeFileSync(
join(desktop, "node_modules/electron-builder/package.json"),
JSON.stringify({
name: "electron-builder",
version: "1.0.0",
bin: { "electron-builder": "../.bin/electron-builder" },
}),
);
writeFileSync(
join(directory, "package.json"),
JSON.stringify({ private: true, workspaces: ["packages/desktop"] }),
);
writeFileSync(
join(desktop, "package.json"),
JSON.stringify({
name: "desktop-fixture",
private: true,
devDependencies: { "electron-builder": "1.0.0" },
}),
);
writeFileSync(
join(desktop, "node_modules/.bin/electron-builder"),
'#!/bin/sh\nprintf "workspace-builder\\n"\n',
{ mode: 0o755 },
);
const result = spawnSync("sh", ["-c", `${invocation} --version`], {
cwd: desktop,
encoding: "utf8",
});
assert.equal(result.status, 0, result.stderr);
assert.equal(result.stdout.trim(), "workspace-builder");
} finally {
rmSync(directory, { recursive: true, force: true });
}
});