mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 14:12:51 +00:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
919e44b319 | ||
|
|
4fe44038e3 | ||
|
|
9b0542d127 | ||
|
|
8130147426 | ||
|
|
fa89738e3a | ||
|
|
14519bdb5b | ||
|
|
891e62309c |
@@ -74,7 +74,10 @@ 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.
|
||||
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.
|
||||
|
||||
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
|
||||
|
||||
@@ -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 \
|
||||
../../node_modules/.bin/electron-builder \
|
||||
npm exec --offline -- electron-builder \
|
||||
--config electron-builder.yml \
|
||||
--dir \
|
||||
--mac \
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
sha256-CeT2PtYIAFwvBC9LIku8VkTl0pvKxYzSCn0QW2ManfU=
|
||||
sha256-KN/AVZV/o0TleLLpmggSrdJqjFDJ/NJYVHiC54xb4jQ=
|
||||
|
||||
Generated
+1018
-1295
File diff suppressed because it is too large
Load Diff
@@ -717,11 +717,13 @@ 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,6 +647,7 @@ 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);
|
||||
@@ -655,6 +656,7 @@ 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() === "?");
|
||||
}
|
||||
|
||||
@@ -204,15 +204,16 @@ export function createHostTransportFactory(
|
||||
}
|
||||
export function createHostClient(
|
||||
connection: string,
|
||||
config: Omit<DaemonClientConfig, "url" | "transportFactory"> & HostTransportOptions,
|
||||
config: Omit<DaemonClientConfig, "url"> & HostTransportOptions,
|
||||
budget?: HostTransportBudget,
|
||||
): DaemonClient {
|
||||
const target = parseHostTarget(connection);
|
||||
const { webSocketAgent, ...clientConfig } = config;
|
||||
const { webSocketAgent, transportFactory, ...clientConfig } = config;
|
||||
return new DaemonClient({
|
||||
...clientConfig,
|
||||
url: "ws://host-connector/ws",
|
||||
...(target.password ? { password: target.password } : {}),
|
||||
transportFactory: createHostTransportFactory(connection, budget, { webSocketAgent }),
|
||||
transportFactory:
|
||||
transportFactory ?? createHostTransportFactory(connection, budget, { webSocketAgent }),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -41,7 +41,7 @@
|
||||
"@types/node": "24.6.0",
|
||||
"@types/ws": "^8.5.14",
|
||||
"electron": "44.2.0",
|
||||
"electron-builder": "26.8.1",
|
||||
"electron-builder": "26.18.0",
|
||||
"typescript": "5.9.3",
|
||||
"unzip-crx-3": "^0.2.0",
|
||||
"vitest": "^4.1.6",
|
||||
|
||||
@@ -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<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);
|
||||
}
|
||||
@@ -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<string> {
|
||||
if (this.resumePurpose !== "history" || (await isDirectory(this.config.cwd))) {
|
||||
return this.config.cwd;
|
||||
}
|
||||
return homedir();
|
||||
private resolveProcessCwd(): Promise<string> {
|
||||
return resolveResumeProcessCwd(this.config.cwd, this.resumePurpose);
|
||||
}
|
||||
|
||||
private async spawnProcess(): Promise<SpawnedACPProcess> {
|
||||
|
||||
@@ -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<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 { 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<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 = {},
|
||||
@@ -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<AgentSessionConfig>,
|
||||
launchContext?: AgentLaunchContext,
|
||||
options?: AgentResumeSessionOptions,
|
||||
): Promise<AgentSession> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string> {
|
||||
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<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,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<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,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<string, string>;
|
||||
readonly prompts: Array<{ message: string; imageCount: number }> = [];
|
||||
|
||||
@@ -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<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;
|
||||
}
|
||||
}
|
||||
@@ -784,6 +784,211 @@ 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,10 +7,12 @@ 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,
|
||||
@@ -235,23 +237,30 @@ export class ConnectedHosts {
|
||||
if (route.registrationId === id)
|
||||
route.close(code, "Host access was disabled, replaced or removed");
|
||||
}
|
||||
private start(row: Saved): void {
|
||||
private start(
|
||||
row: Saved,
|
||||
resume?: { failures: number; transports: DaemonTransportFactory },
|
||||
): void {
|
||||
if (!row.enabled || this.closed) return;
|
||||
let stopped = false;
|
||||
const client = createHostClient(
|
||||
row.connection,
|
||||
{
|
||||
clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
|
||||
connectTimeoutMs: 10000,
|
||||
const transports =
|
||||
resume?.transports ??
|
||||
createHostTransportFactory(row.connection, this.budget, {
|
||||
webSocketAgent: webSocketProxyAgent,
|
||||
},
|
||||
this.budget,
|
||||
);
|
||||
});
|
||||
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 live: Live = {
|
||||
client,
|
||||
accepted: false,
|
||||
stop: () => {
|
||||
stopped = true;
|
||||
clearTimeout(retry);
|
||||
void client.close();
|
||||
},
|
||||
};
|
||||
@@ -285,7 +294,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.",
|
||||
"Update the destination daemon to use connected hosts; retrying automatically.",
|
||||
);
|
||||
// Pin and persist under the same writer as management to prevent discovery races.
|
||||
await this.mutate(async () => {
|
||||
@@ -304,16 +313,22 @@ 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",
|
||||
};
|
||||
row.nextRetryAt = null;
|
||||
if (terminalFailures.has(row.error.code)) {
|
||||
row.status = "error";
|
||||
row.nextRetryAt = null;
|
||||
} else {
|
||||
stopped = true;
|
||||
retry = this.scheduleAcceptanceRetry(row, attempts, transports);
|
||||
}
|
||||
this.changed();
|
||||
await client.close();
|
||||
}
|
||||
@@ -329,18 +344,17 @@ 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;
|
||||
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 (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;
|
||||
if (row.error?.code === "authentication_failed") {
|
||||
row.status = "error";
|
||||
row.nextRetryAt = null;
|
||||
@@ -351,6 +365,18 @@ 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;
|
||||
@@ -531,6 +557,20 @@ 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,13 +88,9 @@ test.each(["uncommitted", "base", "commit"] as const)(
|
||||
60_000,
|
||||
);
|
||||
|
||||
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 }) => {
|
||||
test.each(["single", "batch", "rename"] as const)(
|
||||
"%s diff includes hunks from a repository subdirectory before and after commit",
|
||||
async (shape) => {
|
||||
const ctx = await createDaemonTestContext();
|
||||
const cwd = mkdtempSync(join(tmpdir(), "checkout-diff-subdirectory-"));
|
||||
|
||||
@@ -112,9 +108,7 @@ test.each(
|
||||
}
|
||||
git(cwd, "add", ".");
|
||||
git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "initial");
|
||||
if (mode === "base") {
|
||||
git(cwd, "checkout", "-b", "feature");
|
||||
}
|
||||
git(cwd, "checkout", "-b", "feature");
|
||||
const filename = shape === "rename" ? "renamed.txt" : "normal.txt";
|
||||
if (shape === "rename") {
|
||||
git(cwd, "mv", "sub/normal.txt", `sub/${filename}`);
|
||||
@@ -123,36 +117,38 @@ test.each(
|
||||
if (shape === "batch") {
|
||||
writeFileSync(join(cwd, "sibling.txt"), "new sibling\n");
|
||||
}
|
||||
if (mode === "base") {
|
||||
git(cwd, "add", ".");
|
||||
git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "edit");
|
||||
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);
|
||||
}
|
||||
|
||||
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);
|
||||
await expectSubdirectoryDiff("uncommitted");
|
||||
git(cwd, "add", ".");
|
||||
git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "edit");
|
||||
await expectSubdirectoryDiff("base");
|
||||
} finally {
|
||||
await ctx.cleanup();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
|
||||
@@ -20,6 +20,7 @@ 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,
|
||||
@@ -179,6 +180,55 @@ 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" },
|
||||
@@ -372,15 +422,127 @@ describe("PluginService", () => {
|
||||
"startup-first": { source: "directory", path: first },
|
||||
"startup-second": { source: "directory", path: second },
|
||||
});
|
||||
const snapshots: string[][] = [];
|
||||
service.subscribe(() => snapshots.push(catalogIds(service)));
|
||||
const notifications: { pluginId: string; running: boolean }[] = [];
|
||||
service.subscribe((pluginId) =>
|
||||
notifications.push({ pluginId, running: catalogIds(service).includes(pluginId) }),
|
||||
);
|
||||
|
||||
await service.start();
|
||||
|
||||
expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]);
|
||||
expect(notifications).toEqual([
|
||||
{ pluginId: "startup-first", running: true },
|
||||
{ pluginId: "startup-second", running: true },
|
||||
]);
|
||||
expect(catalogIds(service)).toEqual(["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);
|
||||
|
||||
@@ -230,11 +230,18 @@ export class PluginService {
|
||||
}
|
||||
});
|
||||
if (config.pluginsEnabled === true) {
|
||||
for (const [pluginId, source] of Object.entries(config.plugins ?? {})) {
|
||||
if (source.enabled === false) continue;
|
||||
await this.startConfigured(pluginId);
|
||||
this.notify(pluginId);
|
||||
}
|
||||
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);
|
||||
}),
|
||||
);
|
||||
}
|
||||
this.configStore.onFieldChange("pluginsEnabled", (value) => {
|
||||
this.handleGlobalSwitch(value === true);
|
||||
@@ -560,12 +567,12 @@ export class PluginService {
|
||||
}).catch((error) => this.logger.error({ err: error }, "Failed to enable plugins"));
|
||||
}
|
||||
|
||||
private async startConfigured(pluginId: string): Promise<void> {
|
||||
private async startConfigured(pluginId: string, publishTurn?: Promise<unknown>): 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);
|
||||
await this.startPlugin(pluginId, source.path, publishTurn);
|
||||
} catch (error) {
|
||||
if (this.canPublish(pluginId)) this.recordFailure(pluginId, error);
|
||||
}
|
||||
@@ -594,8 +601,13 @@ export class PluginService {
|
||||
);
|
||||
}
|
||||
|
||||
private async startPlugin(pluginId: string, sourcePath: string): Promise<void> {
|
||||
private async startPlugin(
|
||||
pluginId: string,
|
||||
sourcePath: string,
|
||||
publishTurn?: Promise<unknown>,
|
||||
): Promise<void> {
|
||||
await this.runtime.startPlugin(pluginId, sourcePath, () => this.canPublish(pluginId));
|
||||
await publishTurn;
|
||||
try {
|
||||
await this.publishProviderRegistrations(pluginId, sourcePath);
|
||||
this.publishUsageSources(pluginId);
|
||||
|
||||
@@ -410,7 +410,12 @@ describe("createTerminal", () => {
|
||||
cwd: realpathSync(tmpdir()),
|
||||
cols: 500,
|
||||
command: process.execPath,
|
||||
args: ["-e", "process.stdout.write(JSON.stringify(process.argv.slice(1)))", nastyPrompt],
|
||||
// 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,
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { readFileSync, readdirSync, mkdtempSync, writeFileSync, rmSync } from "node:fs";
|
||||
import { readFileSync, readdirSync, mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
@@ -679,3 +679,48 @@ 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 });
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user