Compare commits

..
Author SHA1 Message Date
Mohamed Boudra 573ae731f7 test(agent): update permission follow-up adapter 2026-10-26 15:18:05 +01:00
Mohamed Boudra 065893bd5b docs(plugins): show how to attach tools to agents 2026-10-26 14:23:31 +01:00
Mohamed Boudra 052004a97d feat(plugins): reopen idle agent sessions after hook changes
Defer automatic plugin changes until the next idle prompt so running turns keep steering. Expose session reload separately from snapshot refresh in the SDK.
2026-10-26 14:03:30 +01:00
Mohamed Boudra f5e311ffd8 feat(plugins): attach tools to individual agent sessions 2026-10-26 14:01:06 +01:00
65 changed files with 2261 additions and 1466 deletions
-28
View File
@@ -169,34 +169,6 @@ jobs:
NODE_OPTIONS: --max-old-space-size=${{ matrix.node_heap_mb }}
run: |
set -euo pipefail
export PASEO_REAL_XCRUN="$(command -v xcrun)"
stapler_bin="$(mktemp -d)"
trap 'rm -rf "$stapler_bin"' EXIT
cat > "$stapler_bin/xcrun" <<'STAPLER'
#!/bin/bash
if [[ "${1:-}" != stapler || "${2:-}" != staple ]]; then
exec "$PASEO_REAL_XCRUN" "$@"
fi
output="$(mktemp)"
trap 'rm -f "$output"' EXIT
for attempt in 1 2 3 4; do
if "$PASEO_REAL_XCRUN" "$@" >"$output" 2>&1; then
cat "$output"
exit 0
else
status=$?
fi
cat "$output"
if [[ "$status" != 68 || "$attempt" == 4 ]] ||
! grep -Fq 'NSURLErrorDomain Code=-1005' "$output"; then
exit "$status"
fi
echo "::warning::Apple staple connection lost; retrying the same signed app ($attempt/4)" >&2
sleep 30
done
STAPLER
chmod +x "$stapler_bin/xcrun"
export PATH="$stapler_bin:$PATH"
build_args=(-- --publish never --mac --${{ matrix.electron_arch }})
build_args+=("-c.publish.releaseType=$RELEASE_TYPE")
build_args+=("-c.publish.channel=$RELEASE_CHANNEL")
+3
View File
@@ -432,6 +432,9 @@ Server entries register lifecycle observers with `server.on()` and request trans
`server.before()`. The [public reference](../public-docs/plugins/reference.md#lifecycle-hooks)
owns callback shapes, ordering, and failure behavior. `plugin-examples/lifecycle-logger` registers all
eleven hooks; `plugin-examples/lifecycle-actions` demonstrates common automation callbacks.
Session-open hooks also [attach tools to individual agents](../public-docs/plugins/reference.md#attach-tools-to-agents),
as shown in `plugin-examples/agent-tools`. The agent manager stamps their metadata per session;
executable definitions stay in the plugin process.
Emit from the operation owner, not a client subscription. Provider history replay must not trigger
live hooks. Observers must not be awaited inside agent mutations: a callback can send a prompt or
+63
View File
@@ -1770,3 +1770,66 @@ test("qualified parents preserve foreign identity without resolving an equal loc
await client.close();
expect(localParent.address).toEqual({ serverId: "srv_sdk_test", agentId: "same-parent" });
});
test("agent reload reopens through the daemon RPC then returns the observed snapshot; refresh remains a read", async () => {
const { client, ws } = await connectClient();
try {
const agent = client.agents.ref("agent_sdk");
const snapshot = createAgent();
const reloaded = agent.reload({ requestId: "reload-sdk" });
expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({
type: "refresh_agent_request",
agentId: "agent_sdk",
requestId: "reload-sdk",
});
ws.message(
sessionMessage({
type: "status",
payload: {
status: "agent_refreshed",
requestId: "reload-sdk",
agentId: "agent_sdk",
timelineSize: 0,
},
}),
);
await expect
.poll(() => parseSentSessionMessage(ws.sent.at(-1)).type)
.toBe("fetch_agent_request");
const fetched = parseSentSessionMessage(ws.sent.at(-1));
ws.message(
sessionMessage({
type: "fetch_agent_response",
payload: {
requestId: fetched.requestId,
agentId: "agent_sdk",
agent: snapshot,
project: null,
error: null,
},
}),
);
expect(await reloaded).toEqual({ agent: snapshot, project: null });
expect(agent.current()).toEqual(snapshot);
const read = agent.refresh("snapshot-read");
expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({
type: "fetch_agent_request",
requestId: "snapshot-read",
});
ws.message(
sessionMessage({
type: "fetch_agent_response",
payload: {
requestId: "snapshot-read",
agentId: "agent_sdk",
agent: snapshot,
project: null,
error: null,
},
}),
);
expect(await read).toEqual({ agent: snapshot, project: null });
} finally {
await client.close();
}
});
+11
View File
@@ -369,6 +369,11 @@ export interface PaseoAgentHandle {
readonly timeline: PaseoAgentTimelineHandle;
current(): PaseoAgent | null;
refresh(requestId?: string): Promise<PaseoAgentRefetchResult | null>;
/**
* Reopens the provider session with current configuration, rerunning session-open hooks.
* Interrupts a running turn and returns the refreshed snapshot.
*/
reload(options?: { requestId?: string }): Promise<PaseoAgentRefetchResult | null>;
send(text: string, options?: PaseoAgentSendOptions): Promise<void>;
respondToPermission(options: PaseoAgentRespondToPermissionOptions): Promise<void>;
/** Sends a prompt and resolves when that turn finishes or needs attention. */
@@ -981,6 +986,12 @@ function createAgentHandleFactory(
current = result?.agent ?? null;
return result;
},
reload: async (options) => {
await daemonClient.refreshAgent(id, options?.requestId);
const result = await daemonClient.fetchAgent({ agentId: id });
current = result?.agent ?? null;
return result;
},
send: async (text, options) => {
await daemonClient.sendAgentMessage(id, text, options);
},
+57 -1
View File
@@ -92,7 +92,12 @@ function typeErrors(code: string): string[] {
}
describe("plugin SDK import boundaries", () => {
it("publishes process capabilities without their Windows implementation helpers", () => {
expect(Object.keys(server).sort()).toEqual(["execCommand", "spawnProcess", "terminateProcess"]);
expect(Object.keys(server).sort()).toEqual([
"defineAgentTool",
"execCommand",
"spawnProcess",
"terminateProcess",
]);
});
it("classifies every published entry", () => {
@@ -209,3 +214,54 @@ describe("plugin panel SDK boundary", () => {
).toEqual([]);
});
});
describe("plugin agent tools SDK boundary", () => {
it("lets a session hook attach typed tools without exposing their transport", () => {
expect(
typeErrors(`
import { z } from "zod";
import { defineAgentTool, type PluginServerContext, type AgentToolResult, type AgentToolContext, type AgentToolDefinition } from "@getpaseo/plugin/server";
const tool: AgentToolDefinition = defineAgentTool({
name: "create_issue",
description: "Create an issue",
input: z.object({ title: z.string(), count: z.string().transform(Number) }),
async execute(input, { agent, paseo, signal }) {
const count: number = input.count;
// @ts-expect-error input is inferred from the schema
input.missing;
return { text: agent.id + input.title, structuredContent: { count } };
},
});
export default function contribute(server: PluginServerContext) {
server.before("agent.session_open", ({ request }) => {
const title: string | null = request.title;
// @ts-expect-error labels are read-only
request.labels.x = "changed";
if (request.labels.x !== "y") return;
return { ...request, tools: [...request.tools, tool] };
});
return () => {};
}
`),
).toEqual([]);
});
});
describe("agent session reload SDK boundary", () => {
it("exposes reload separately from snapshot refresh", () => {
expect(
typeErrors(`
import { type PluginServerContext } from "@getpaseo/plugin/server";
import { type PaseoAgentRefetchResult } from "@getpaseo/client";
async function reload(server: PluginServerContext, paseo: import("@getpaseo/client").PaseoApi) {
const agent = paseo.agents.ref("chosen-agent");
const read: PaseoAgentRefetchResult | null = await agent.refresh("read-snapshot");
const reopened: PaseoAgentRefetchResult | null = await agent.reload({ requestId: "reload-session" });
await agent.reload();
// @ts-expect-error reload options can grow without a positional signature change
await agent.reload("reload-session");
}
`),
).toEqual([]);
});
});
+39
View File
@@ -0,0 +1,39 @@
import type { PaseoApi } from "@getpaseo/client";
import type { z } from "zod";
import type { PluginHookAgent } from "./lifecycle.js";
export interface AgentToolContext {
agent: PluginHookAgent;
paseo: PaseoApi;
signal: AbortSignal;
}
export interface AgentToolResult {
text?: string;
structuredContent?: unknown;
isError?: boolean;
}
export interface AgentToolDefinition {
name: string;
title?: string;
description: string;
input: z.ZodType;
execute(input: unknown, context: AgentToolContext): AgentToolResult | Promise<AgentToolResult>;
}
export function defineAgentTool<Input extends z.ZodType>(tool: {
name: string;
title?: string;
description: string;
input: Input;
execute(
input: z.output<Input>,
context: AgentToolContext,
): AgentToolResult | Promise<AgentToolResult>;
}): AgentToolDefinition {
if (!/^[a-z][a-z0-9_]*$/.test(tool.name) || tool.name.length > 64) {
throw new Error(`Invalid agent tool name: ${tool.name}`);
}
return tool;
}
+3
View File
@@ -24,3 +24,6 @@ export type {
} from "./lifecycle.js";
export { spawnProcess, execCommand, terminateProcess } from "./process.js";
export { defineAgentTool } from "./agent-tools.js";
export type { AgentToolDefinition, AgentToolResult, AgentToolContext } from "./agent-tools.js";
+4
View File
@@ -1,3 +1,4 @@
import type { AgentToolDefinition } from "./agent-tools.js";
import type {
AgentPermissionRequest,
AgentPermissionResponse,
@@ -36,6 +37,9 @@ export interface PluginSessionOpenRequest {
cwd: string;
reason: "create" | "resume" | "refresh" | "import";
purpose: "interactive" | "history";
readonly title: string | null;
readonly labels: Readonly<Record<string, string>>;
tools: AgentToolDefinition[];
env: Record<string, string>;
}
+116 -11
View File
@@ -1,3 +1,5 @@
import type { AttachedAgentTool, PluginAgentToolsRuntime } from "../plugins/agent-tools.js";
import type { AgentToolResult } from "@getpaseo/plugin/server";
import type { RuntimeCleanupCandidate } from "./idle-runtime-cleanup/index.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { projectTimelineRows } from "./timeline-projection.js";
@@ -333,6 +335,7 @@ export interface AgentManagerOptions {
terminalManager?: TerminalManager | null;
mcpBaseUrl?: string;
mcpAuthToken?: string;
pluginAgentTools?: PluginAgentToolsRuntime;
paseoToolsEnabled?: boolean;
paseoToolCatalogFactory?: PaseoToolCatalogFactory;
resolvePaseoToolPolicy?: (provider: AgentProvider) => ProviderPaseoToolsPolicy | undefined;
@@ -766,6 +769,10 @@ export class AgentManager {
private readonly agentStreamCoalescer: AgentStreamCoalescer;
private mcpBaseUrl: string | null;
private readonly mcpAuthToken: string | null;
private sessionOpenGeneration = 0;
private readonly sessionOpenGenerations = new Map<string, number>();
private readonly attachedPluginTools = new Map<string, AttachedAgentTool[]>();
private readonly pluginAgentTools: PluginAgentToolsRuntime | undefined;
private paseoToolsEnabled = true;
private paseoToolCatalogFactory: PaseoToolCatalogFactory | null = null;
private readonly paseoToolPolicies = new Map<string, ProviderPaseoToolsPolicy | undefined>();
@@ -783,6 +790,7 @@ export class AgentManager {
constructor(options: AgentManagerOptions) {
this.pluginLifecycle = options.pluginLifecycle;
this.pluginAgentTools = options.pluginAgentTools;
this.timelineStore = options.timelineStore ?? new InMemoryAgentTimelineStore();
this.idFactory = options.idFactory ?? (() => randomUUID());
this.registry = options?.registry;
@@ -887,6 +895,62 @@ export class AgentManager {
this.paseoToolCatalogFactory = factory;
}
invalidatePluginSessions(): void {
this.sessionOpenGeneration++;
}
async reloadStalePluginSession(agentId: string): Promise<void> {
if (!this.needsPluginSessionReload(agentId)) return;
await this.trackAgentRegistrationOperation(
this.runLifecycleMutation(agentId, async () => {
if (!this.needsPluginSessionReload(agentId)) return;
await this.reloadAgentSessionInternal(agentId);
}),
);
}
private needsPluginSessionReload(agentId: string): boolean {
const agent = this.getAgent(agentId);
return (
agent !== null &&
agent.lifecycle !== "closed" &&
this.sessionOpenGenerations.get(agentId) !== this.sessionOpenGeneration &&
!this.hasInFlightRun(agentId)
);
}
getPluginTools(agentId: string): readonly AttachedAgentTool[] {
return (this.attachedPluginTools.get(agentId) ?? []).filter((tool) =>
this.pluginAgentTools?.isToolPluginAvailable(tool.pluginId, tool.pluginInstanceId),
);
}
async executePluginTool(
tool: AttachedAgentTool,
agentId: string,
input: unknown,
signal?: AbortSignal,
): Promise<AgentToolResult> {
const agent = this.getAgent(agentId);
if (!agent || agent.lifecycle === "closed") throw new Error(`Agent is not live: ${agentId}`);
if (!this.pluginAgentTools) throw new Error("Plugin tools are unavailable");
const record = await this.registry?.get(agentId);
return this.pluginAgentTools.callTool(
tool.pluginId,
tool.pluginInstanceId,
describeHookAgent({ ...agent, title: record?.title ?? agent.config.title }),
tool.name,
input,
signal,
);
}
private detachPluginTools(agentId: string): void {
this.attachedPluginTools.delete(agentId);
this.sessionOpenGenerations.delete(agentId);
this.pluginAgentTools?.detachTools(agentId);
}
getPaseoToolPolicy(agentId: string): ProviderPaseoToolsPolicy | undefined {
return this.paseoToolPolicies.get(agentId);
}
@@ -1312,7 +1376,13 @@ export class AgentManager {
storedConfig.cwd,
paseoToolPolicy,
options?.env,
{ reason: "create", purpose: "interactive", workspaceId: options.workspaceId ?? null },
{
reason: "create",
purpose: "interactive",
workspaceId: options.workspaceId ?? null,
title: options.initialTitle ?? storedConfig.title ?? null,
labels: options.labels ?? {},
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
const createOptions = this.buildCreateSessionOptions(options);
@@ -1434,6 +1504,8 @@ export class AgentManager {
reason: "resume",
purpose,
workspaceId: options?.workspaceId ?? null,
title: record?.title ?? mergedConfig.title ?? null,
labels: record?.labels ?? options?.labels ?? {},
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
@@ -1491,7 +1563,12 @@ export class AgentManager {
storedConfig.cwd,
paseoToolPolicy,
undefined,
{ reason: "import", purpose: "interactive", workspaceId: input.workspaceId },
{
reason: "import",
purpose: "interactive",
workspaceId: input.workspaceId,
labels: input.labels ?? {},
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
const imported = await client.importSession(
@@ -1586,7 +1663,13 @@ export class AgentManager {
storedConfig.cwd,
paseoToolPolicy,
undefined,
{ reason: "refresh", purpose: "interactive", workspaceId: existing.workspaceId },
{
reason: "refresh",
purpose: "interactive",
workspaceId: existing.workspaceId,
title: await this.resolveInitialPersistedTitle(agentId, existing.config, null),
labels: existing.labels,
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
if (
@@ -1650,6 +1733,7 @@ export class AgentManager {
throw error;
} finally {
if (!handedToRegistration) {
this.detachPluginTools(agentId);
if (hadPreviousPaseoToolPolicy) {
this.paseoToolPolicies.set(agentId, previousPaseoToolPolicy);
} else {
@@ -1857,6 +1941,7 @@ export class AgentManager {
} catch (error) {
persistError = error;
}
this.detachPluginTools(agentId);
this.emitClosedAgent(closedAgent, { persist: false });
this.logger.trace(
{
@@ -3940,6 +4025,7 @@ export class AgentManager {
private discardRetainedAgentState(agentId: string): void {
this.timelineStore.delete(agentId);
this.paseoToolPolicies.delete(agentId);
this.detachPluginTools(agentId);
for (const event of this.providerSubagents.deleteParent(agentId)) {
this.dispatch({ type: "provider_subagent", event });
}
@@ -5426,10 +5512,15 @@ export class AgentManager {
reason: PluginSessionOpenRequest["reason"];
purpose: PluginSessionOpenRequest["purpose"];
workspaceId?: string | null;
title?: string | null;
labels?: Record<string, string>;
},
): Promise<AgentLaunchContext> {
// A plugin changing during this open must still make the resulting session stale.
const generation = this.sessionOpenGeneration;
this.detachPluginTools(agentId);
if (this.pluginLifecycle) {
const request: PluginSessionOpenRequest = {
const request = {
agentId,
provider: client.provider,
cwd,
@@ -5437,9 +5528,14 @@ export class AgentManager {
reason: opening?.reason ?? "resume",
purpose: opening?.purpose ?? "interactive",
env: { ...env },
title: opening?.title ?? null,
labels: opening?.labels ?? {},
tools: [],
};
const transformed = await this.pluginLifecycle.before("agent.session_open", request);
env = transformed.env;
if (request.purpose === "interactive")
this.attachedPluginTools.set(agentId, transformed.tools as AttachedAgentTool[]);
}
const context: AgentLaunchContext = {
agentId,
@@ -5449,17 +5545,26 @@ export class AgentManager {
PASEO_AGENT_CWD: cwd,
},
};
if (
const useNativeTools =
this.paseoToolsEnabled &&
isPaseoToolPolicyEnabled(paseoToolPolicy) &&
client.capabilities.supportsNativePaseoTools &&
this.paseoToolCatalogFactory
client.capabilities.supportsNativePaseoTools;
if (
this.paseoToolCatalogFactory &&
(useNativeTools || this.getPluginTools(agentId).length > 0)
) {
context.paseoTools = await this.paseoToolCatalogFactory({
callerAgentId: agentId,
paseoToolPolicy,
});
try {
const catalog = await this.paseoToolCatalogFactory({
callerAgentId: agentId,
paseoToolPolicy,
});
if (useNativeTools) context.paseoTools = catalog;
} catch (error) {
this.detachPluginTools(agentId);
throw error;
}
}
this.sessionOpenGenerations.set(agentId, generation);
return context;
}
@@ -15,6 +15,7 @@ export type AgentRunController = Pick<
| "getAgent"
| "tryRunOutOfBand"
| "hasInFlightRun"
| "reloadStalePluginSession"
| "replaceAgentRun"
| "steerOrReplaceActiveTurn"
| "streamAgent"
@@ -115,6 +116,7 @@ export async function startAgentRun(
return { disposition: "out_of_band" };
}
try {
await agentManager.reloadStalePluginSession(agentId);
return await startAgentRunInner(agentManager, agentId, prompt, logger, options);
} catch (error) {
if (!isStaleProviderSessionError(error)) throw error;
@@ -69,6 +69,7 @@ test("session create forwards clientMessageId to the initial prompt run options"
getAgent: vi.fn(() => snapshot),
tryRunOutOfBand: vi.fn(() => false),
hasInFlightRun: vi.fn(() => false),
reloadStalePluginSession: async () => {},
streamAgent,
waitForAgentRunStart: vi.fn(async () => undefined),
} as unknown as Parameters<typeof createAgentCommand>[0]["agentManager"],
@@ -270,6 +270,8 @@ function buildAgentManagerSpies() {
hydrateTimelineFromProvider: vi.fn().mockResolvedValue(undefined),
appendTimelineItem: vi.fn().mockResolvedValue(undefined),
emitLiveTimelineItem: vi.fn().mockResolvedValue(undefined),
getPluginTools: () => [],
reloadStalePluginSession: async () => {},
hasInFlightRun: vi.fn().mockReturnValue(false),
steerOrReplaceActiveTurn: async () => ({ status: "inactive" as const }),
tryRunOutOfBand: vi.fn().mockReturnValue(false),
@@ -617,6 +619,10 @@ class FakeBrowserToolsBroker {
}
class BoundaryAgentManagerFake {
public getPluginTools(): [] {
return [];
}
private readonly agent = createManagedAgent({
id: "agent-1",
cwd: REPO_CWD,
@@ -40,6 +40,8 @@ class FakePermissionAgentManager {
return undefined;
}
async reloadStalePluginSession(): Promise<void> {}
hasInFlightRun(): boolean {
return this.hasRunInFlight;
}
@@ -811,46 +811,6 @@ describe("ACPAgentSession terminal tools", () => {
truncated: false,
});
});
test("trims large terminal output without rescanning the retained suffix", async () => {
const session = createSession();
const terminal = await session.createTerminal({
sessionId: "session-1",
command: process.execPath,
args: ["-e", 'process.stdout.write("x".repeat(1024 * 1024))'],
cwd: process.cwd(),
outputByteLimit: 1024,
});
await session.waitForTerminalExit({ sessionId: "session-1", terminalId: terminal.terminalId });
await expect(
session.terminalOutput({ sessionId: "session-1", terminalId: terminal.terminalId }),
).resolves.toEqual({
output: "x".repeat(1024),
truncated: true,
exitStatus: { exitCode: 0, signal: null },
});
}, 1000);
test("trims terminal output at a UTF-8 character boundary", async () => {
const session = createSession();
const terminal = await session.createTerminal({
sessionId: "session-1",
command: process.execPath,
args: ["-e", 'process.stdout.write("😀ab")'],
cwd: process.cwd(),
outputByteLimit: 3,
});
await session.waitForTerminalExit({ sessionId: "session-1", terminalId: terminal.terminalId });
await expect(
session.terminalOutput({ sessionId: "session-1", terminalId: terminal.terminalId }),
).resolves.toEqual({
output: "ab",
truncated: true,
exitStatus: { exitCode: 0, signal: null },
});
});
});
describe("mapACPUsage", () => {
@@ -4093,16 +4093,10 @@ function appendTerminalOutput(entry: TerminalEntry, chunk: string): void {
if (!limit) {
return;
}
const output = Buffer.from(entry.output, "utf8");
if (output.length <= limit) {
return;
while (Buffer.byteLength(entry.output, "utf8") > limit && entry.output.length > 0) {
entry.output = entry.output.slice(1);
entry.truncated = true;
}
let start = output.length - limit;
while (start < output.length && (output[start] & 0xc0) === 0x80) {
start += 1;
}
entry.output = output.subarray(start).toString("utf8");
entry.truncated = true;
}
function readRecord(value: unknown): Record<string, unknown> | null {
@@ -3063,81 +3063,6 @@ describe("PiRpcAgentClient", () => {
]);
});
test.each([
{ outcome: "successful", error: null },
{ outcome: "failed", error: new Error("summarizer failed") },
])("waits for $outcome manual compaction before sending a new prompt", async ({ error }) => {
const { pi, session, events } = await createSession();
onTestFinished(() => session.close());
const fakeSession = pi.latestSession();
fakeSession.compactError = error;
const compactionFinished = Promise.withResolvers<void>();
fakeSession.compactCompletion = compactionFinished.promise;
const handler = session.tryHandleOutOfBand("/compact");
const compaction = handler!.run({ emit: () => {} });
const { turnId } = await session.startTurn("continue after compaction");
await flushTurnScheduling();
expect(fakeSession.compactRequests).toEqual([{}]);
expect(fakeSession.prompts).toEqual([]);
compactionFinished.resolve();
await compaction;
await flushTurnScheduling();
expect(fakeSession.prompts).toEqual([{ message: "continue after compaction", imageCount: 0 }]);
fakeSession.finishTurn();
await events.nextTurnCompletion();
expect(events.turnLifecycleEvents()).toContainEqual({ type: "turn_completed", turnId });
expect(events.eventTypes()).not.toContain("turn_failed");
});
test.each(["interrupt", "close"] as const)(
"does not send a prompt waiting on compaction after %s",
async (action) => {
const { pi, session } = await createSession();
onTestFinished(() => session.close());
const fakeSession = pi.latestSession();
const compactionFinished = Promise.withResolvers<void>();
fakeSession.compactCompletion = compactionFinished.promise;
const handler = session.tryHandleOutOfBand("/compact");
const compaction = handler!.run({ emit: () => {} });
await session.startTurn("do not send this prompt");
await session[action]();
compactionFinished.resolve();
await compaction;
await flushTurnScheduling();
expect(fakeSession.prompts).toEqual([]);
},
);
test("does not send a prompt when compaction finishes during interruption", async () => {
const { pi, session, events } = await createSession();
onTestFinished(() => session.close());
const fakeSession = pi.latestSession();
const compactionFinished = Promise.withResolvers<void>();
fakeSession.compactCompletion = compactionFinished.promise;
const clearQueueFinished = Promise.withResolvers<void>();
fakeSession.clearQueueCompletion = clearQueueFinished.promise;
const handler = session.tryHandleOutOfBand("/compact");
const compaction = handler!.run({ emit: () => {} });
const { turnId } = await session.startTurn("do not send this prompt");
const interruption = session.interrupt();
compactionFinished.resolve();
await compaction;
await flushTurnScheduling();
expect(fakeSession.controlRequests).toEqual(["clear_queue"]);
expect(fakeSession.prompts).toEqual([]);
clearQueueFinished.resolve();
await interruption;
expect(events.turnLifecycleEvents()).toEqual([{ type: "turn_canceled", turnId }]);
expect(fakeSession.prompts).toEqual([]);
});
test("closes Pi compact loading marker when RPC rejects after compaction starts", async () => {
const { pi, session } = await createSession();
const fakeSession = pi.latestSession();
@@ -1189,7 +1189,6 @@ export class PiRpcAgentSession implements AgentSession {
private readonly contextUserEntries: PiCapturedEntry[] = [];
private readonly treeUserEntriesById = new Map<string, PiCapturedEntry>();
private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
private compactionPromise: Promise<void> | null = null;
private outOfBandCompactionEmit: ((event: AgentStreamEvent) => void) | null = null;
private outOfBandCompactionStarted = false;
private outOfBandCompactionCompleted = false;
@@ -1297,15 +1296,6 @@ export class PiRpcAgentSession implements AgentSession {
void (async () => {
try {
if (this.compactionPromise) {
// Pi RPC rejects prompts until the manual compaction request settles.
await this.compactionPromise.catch(() => undefined);
const turnCanceled =
this.closed || this.activeTurnId !== turnId || this.interruptingTurn?.turnId === turnId;
if (turnCanceled) {
return;
}
}
const ack = await this.runtimeSession.prompt(payload.text, payload.images);
this.activePromptRequestId = ack.requestId ?? null;
const correlatedResult = ack.requestId
@@ -1814,8 +1804,7 @@ export class PiRpcAgentSession implements AgentSession {
this.outOfBandCompactionStarted = false;
this.outOfBandCompactionCompleted = false;
try {
this.compactionPromise = this.runtimeSession.compact(customInstructions);
await this.compactionPromise;
await this.runtimeSession.compact(customInstructions);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
if (
@@ -1841,7 +1830,6 @@ export class PiRpcAgentSession implements AgentSession {
},
});
} finally {
this.compactionPromise = null;
if (this.outOfBandCompactionEmit === emit && !this.outOfBandCompactionStarted) {
this.outOfBandCompactionEmit = null;
this.outOfBandCompactionStarted = false;
@@ -108,7 +108,6 @@ export class FakePiSession implements PiRuntimeSession {
steerError: Error | null = null;
readonly controlRequests: string[] = [];
clearQueueError: Error | null = null;
clearQueueCompletion: Promise<void> | null = null;
readonly compactRequests: Array<{ customInstructions?: string }> = [];
readonly setAutoCompactionRequests: boolean[] = [];
readonly subagentSubscriptionRequests: FakePiSubagentSubscriptionLevel[] = [];
@@ -141,7 +140,6 @@ export class FakePiSession implements PiRuntimeSession {
subagentSubscriptionError: Error | null = null;
setSessionNameError: Error | null = null;
compactError: Error | null = null;
compactCompletion: Promise<void> | null = null;
emitCompactEnd = true;
getStateError: Error | null = null;
getSessionStatsError: Error | null = null;
@@ -235,7 +233,6 @@ export class FakePiSession implements PiRuntimeSession {
async clearQueue(): Promise<void> {
this.controlRequests.push("clear_queue");
await this.clearQueueCompletion;
if (this.clearQueueError) {
throw this.clearQueueError;
}
@@ -244,7 +241,6 @@ export class FakePiSession implements PiRuntimeSession {
async compact(customInstructions?: string): Promise<void> {
this.compactRequests.push(customInstructions === undefined ? {} : { customInstructions });
this.emit({ type: "compaction_start", reason: "manual" });
await this.compactCompletion;
if (this.emitCompactEnd) {
this.emit({ type: "compaction_end", reason: "manual" });
}
@@ -1,3 +1,4 @@
import { registerPluginTools } from "./plugin-tools.js";
import type { FinishSubscriptions, FinishRegistration } from "../finish-subscriptions/index.js";
import type { AgentMessageDelivery } from "../agent-messages/index.js";
import {
@@ -704,6 +705,7 @@ function createToolCatalog(
}
const localHostOperations = new Map<string, HostOperation>();
const tools = new Map<string, PaseoToolDefinition>();
const builtinNames = new Set<string>();
const hostTools = new Map([
["create_agent", "createAgent"],
["create_workspace", "createWorkspace"],
@@ -738,6 +740,7 @@ function createToolCatalog(
// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Tool handlers are schema-validated at registration boundaries.
handler: (input: any, context: PaseoToolExecutionContext) => Promise<PaseoToolResult>,
) => {
builtinNames.add(name);
if (!isPaseoToolEnabled(options.paseoToolPolicy, name)) {
return;
}
@@ -3272,6 +3275,14 @@ function createToolCatalog(
},
);
registerPluginTools({
agentManager,
callerAgentId,
scope,
tools,
builtinNames,
policy: options.paseoToolPolicy,
});
return toCatalog();
}
@@ -0,0 +1,40 @@
import { z } from "zod";
import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-config";
import type { AgentManager } from "../agent-manager.js";
import { isPaseoToolEnabled } from "../paseo-tool-policy.js";
import type { PaseoToolDefinition } from "./types.js";
import { addModelVisibleStructuredContent } from "./paseo-tool-serialization.js";
export function registerPluginTools(options: {
agentManager: AgentManager;
callerAgentId?: string;
scope: "agent" | "top-level";
tools: Map<string, PaseoToolDefinition>;
builtinNames: Set<string>;
policy?: ProviderPaseoToolsPolicy;
}): void {
const { agentManager, callerAgentId, scope, tools, builtinNames, policy } = options;
if (scope !== "agent" || !callerAgentId) return;
for (const tool of agentManager.getPluginTools(callerAgentId)) {
if (builtinNames.has(tool.name)) {
throw new Error(
`Agent tool ${tool.name} from plugin ${tool.pluginId} collides with Paseo tool ${tool.name}`,
);
}
if (!isPaseoToolEnabled(policy, tool.name)) continue;
tools.set(tool.name, {
name: tool.name,
title: tool.title,
description: tool.description,
inputSchema: z.fromJSONSchema(tool.inputSchema),
async handler(input, { signal }) {
const result = await agentManager.executePluginTool(tool, callerAgentId, input, signal);
return addModelVisibleStructuredContent({
content: result.text === undefined ? [] : [{ type: "text", text: result.text }],
structuredContent: result.structuredContent,
isError: result.isError,
});
},
});
}
}
@@ -1,147 +0,0 @@
import { promises as fs } from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, expect, it } from "vitest";
import { writeFileAtomic, writeJsonFileAtomic } from "./atomic-file.js";
let directory: string;
beforeEach(async () => {
directory = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-atomic-file-"));
});
afterEach(async () => {
await fs.rm(directory, { recursive: true, force: true });
});
it.each(["EPERM", "EACCES", "EBUSY"])(
"commits a completed receipt after transient Windows %s replace failures",
async (code) => {
const file = path.join(directory, "receipt.json");
await writeJsonFileAtomic(file, { state: "pending" });
let failures = 3;
let elapsed = 0;
await writeJsonFileAtomic(
file,
{ state: "completed" },
{
platform: "win32",
now: () => elapsed,
sleep: async (milliseconds) => {
elapsed += milliseconds;
},
rename: async (source, destination) => {
if (failures-- > 0) {
expect(JSON.parse(await fs.readFile(destination, "utf8"))).toEqual({
state: "pending",
});
expect(JSON.parse(await fs.readFile(source, "utf8"))).toEqual({ state: "completed" });
throw Object.assign(new Error("destination held open"), { code });
}
await fs.rename(source, destination);
},
},
);
expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "completed" });
expect(await fs.readdir(directory)).toEqual(["receipt.json"]);
expect(elapsed).toBeGreaterThan(0);
},
);
it.each(["EPERM", "EACCES", "EBUSY"])(
"preserves the previous file and removes the temp file when Windows %s persists",
async (code) => {
const file = path.join(directory, "receipt.json");
await writeJsonFileAtomic(file, { state: "pending" });
const error = Object.assign(new Error("destination held open"), { code });
const waits: number[] = [];
let elapsed = 0;
const attempts: number[] = [];
await expect(
writeJsonFileAtomic(
file,
{ state: "completed" },
{
platform: "win32",
now: () => elapsed,
sleep: async (milliseconds) => {
waits.push(milliseconds);
elapsed += milliseconds;
},
rename: async () => {
attempts.push(elapsed);
throw error;
},
},
),
).rejects.toBe(error);
expect(elapsed).toBe(1000);
expect(waits.slice(0, 5)).toEqual([10, 20, 40, 80, 100]);
expect(Math.max(...waits)).toBe(100);
expect(attempts.every((time) => time < 1000)).toBe(true);
expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "pending" });
expect(await fs.readdir(directory)).toEqual(["receipt.json"]);
},
);
it.each([
["linux", "EPERM"],
["darwin", "EACCES"],
["linux", "EBUSY"],
["win32", "ENOENT"],
["win32", "ENOSPC"],
["win32", "EXDEV"],
] as const)("does not retry %s %s failures", async (platform, code) => {
const file = path.join(directory, "receipt.json");
await writeJsonFileAtomic(file, { state: "pending" });
const error = Object.assign(new Error("rename failed"), { code });
let attempts = 0;
const waits: number[] = [];
await expect(
writeFileAtomic(file, "replacement", {
platform,
now: () => 0,
sleep: async (milliseconds) => {
waits.push(milliseconds);
},
rename: async () => {
attempts++;
throw error;
},
}),
).rejects.toBe(error);
expect(attempts).toBe(1);
expect(waits).toEqual([]);
expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "pending" });
expect(await fs.readdir(directory)).toEqual(["receipt.json"]);
});
it("does not retry after a rename consumes the Windows retry budget", async () => {
const file = path.join(directory, "receipt.json");
const error = Object.assign(new Error("slow rename failed"), { code: "EPERM" });
let elapsed = 0;
const waits: number[] = [];
await expect(
writeFileAtomic(file, "replacement", {
platform: "win32",
now: () => elapsed,
sleep: async (milliseconds) => {
waits.push(milliseconds);
},
rename: async () => {
elapsed += 1000;
throw error;
},
}),
).rejects.toBe(error);
expect(waits).toEqual([]);
expect(await fs.readdir(directory)).toEqual([]);
});
it("creates parent directories and replaces binary and JSON files through the default runtime", async () => {
const file = path.join(directory, "nested", "file");
const bytes = Buffer.from([0, 255, 128, 1]);
await writeFileAtomic(file, bytes);
expect(await fs.readFile(file)).toEqual(bytes);
await writeJsonFileAtomic(file, { state: "completed" });
expect(await fs.readFile(file, "utf8")).toBe('{\n "state": "completed"\n}');
expect(await fs.readdir(path.dirname(file))).toEqual(["file"]);
});
+3 -54
View File
@@ -1,26 +1,10 @@
import { randomUUID } from "node:crypto";
import { promises as fs } from "node:fs";
import path from "node:path";
import { setTimeout as sleep } from "node:timers/promises";
interface AtomicFileRuntime {
platform: NodeJS.Platform;
rename: (source: string, destination: string) => Promise<void>;
now: () => number;
sleep: (milliseconds: number) => Promise<void>;
}
const runtime: AtomicFileRuntime = {
platform: process.platform,
rename: fs.rename,
now: () => performance.now(),
sleep,
};
export async function writeFileAtomic(
filePath: string,
data: string | NodeJS.ArrayBufferView,
operations: AtomicFileRuntime = runtime,
): Promise<void> {
await fs.mkdir(path.dirname(filePath), { recursive: true });
const tempPath = path.join(
@@ -29,48 +13,13 @@ export async function writeFileAtomic(
);
try {
await fs.writeFile(tempPath, data, "utf8");
await replaceFile(tempPath, filePath, operations);
await fs.rename(tempPath, filePath);
} catch (error) {
await fs.rm(tempPath, { force: true });
throw error;
}
}
async function replaceFile(
source: string,
destination: string,
operations: AtomicFileRuntime,
): Promise<void> {
// Windows readers/scanners can briefly deny replacement. Yield with backoff,
// but keep interactive persistence failures bounded to a one-second retry budget.
const deadline = operations.now() + 1000;
let delay = 10;
for (;;) {
try {
await operations.rename(source, destination);
return;
} catch (error) {
if (
operations.platform !== "win32" ||
!(error instanceof Error) ||
!("code" in error) ||
!["EPERM", "EACCES", "EBUSY"].includes(String(error.code))
) {
throw error;
}
const remaining = deadline - operations.now();
if (remaining <= 0) throw error;
await operations.sleep(Math.min(delay, remaining));
if (operations.now() >= deadline) throw error;
delay = Math.min(delay * 2, 100);
}
}
}
export async function writeJsonFileAtomic(
filePath: string,
value: unknown,
operations: AtomicFileRuntime = runtime,
): Promise<void> {
await writeFileAtomic(filePath, JSON.stringify(value, null, 2), operations);
export async function writeJsonFileAtomic(filePath: string, value: unknown): Promise<void> {
await writeFileAtomic(filePath, JSON.stringify(value, null, 2));
}
+7
View File
@@ -995,6 +995,7 @@ export async function createPaseoDaemon(
const initialAgentManagerState = providerSnapshotManager.getAgentManagerProviderState();
const agentManager = new AgentManager({
pluginLifecycle: pluginRuntime,
pluginAgentTools: pluginRuntime,
clients: initialAgentManagerState.clients,
providerDefinitions: initialAgentManagerState.providerDefinitions,
registry: agentStorage,
@@ -1015,6 +1016,10 @@ export async function createPaseoDaemon(
const unsubscribePluginProviders =
pluginRuntime.subscribeProviderRegistrations(syncPluginProviders);
const unsubscribePluginSessionChanges = pluginRuntime.subscribeSessionOpenChanges(() => {
agentManager.invalidatePluginSessions();
});
const detachAgentStoragePersistence = attachAgentStoragePersistence(
logger,
agentManager,
@@ -1914,6 +1919,7 @@ export async function createPaseoDaemon(
localCredential = null;
await deleteLocalCredential(config.paseoHome);
unsubscribePluginProviders();
unsubscribePluginSessionChanges();
await pluginRuntime.stopAllPlugins().catch(() => undefined);
await serviceProxy.stopStandalone().catch(() => undefined);
await agentProviderRuntime.shutdown().catch(() => undefined);
@@ -1933,6 +1939,7 @@ export async function createPaseoDaemon(
// that is still open. Plugins themselves are stopped once every session
// they serve has been closed, further down.
unsubscribePluginProviders();
unsubscribePluginSessionChanges();
await hubRelationships.stop();
workspaceReconciliation.dispose();
scriptHealthMonitor.stop();
@@ -1544,7 +1544,7 @@ test(
);
test(
"remote agent reports, local CLI sends and automatic completion reach the same recipient",
"remote agent reports, local CLI sends and automatic completion share the recipient batch",
{ timeout: 30000 },
async () => {
const parentPrompts: string[] = [];
@@ -1633,10 +1633,7 @@ test(
});
expect(report.isError).not.toBe(true);
expect(report.structuredContent).toMatchObject({ deliveryStatus: "queued", serverId: aId });
// A CLI process can take longer than the two-second quiet window to start.
// Exercise each delivery across windows; delivery.test.ts owns batching with a clock.
await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1);
expect(parentPrompts[0]).toContain("REMOTE_REPORT &lt;ready&gt;");
expect(parentPrompts).toEqual([]);
const cli = await promisify(execFile)(
process.execPath,
@@ -1659,8 +1656,7 @@ test(
},
);
expect(JSON.parse(cli.stdout)).toMatchObject({ status: "queued" });
await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(2);
expect(parentPrompts[1]).toContain("LOCAL_CLI_REPORT");
expect(parentPrompts).toEqual([]);
const receiptStates = async () => {
const directory = join(a.paseoHome, "agent-requests");
const receipts = await Promise.all(
@@ -1675,10 +1671,13 @@ test(
.map((receipt) => receipt.state)
.sort();
};
expect(await receiptStates()).toEqual(["pending", "pending"]);
finishChild.resolve();
await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(3);
expect(parentPrompts[2]).toContain('event="finished"');
expect(parentPrompts[2]).toContain(`source-agent-id="${child.id}"`);
await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1);
expect(parentPrompts[0]).toContain("REMOTE_REPORT &lt;ready&gt;");
expect(parentPrompts[0]).toContain("LOCAL_CLI_REPORT");
expect(parentPrompts[0]).toContain('event="finished"');
expect(parentPrompts[0]).toContain(`source-agent-id="${child.id}"`);
await expect
.poll(receiptStates, {
timeout: 10000,
@@ -1871,7 +1870,6 @@ test(
});
const a = await daemon({
mcpEnabled: true,
logger: completionLogger(),
agentClients: createTestAgentClients({
onStartTurn: (prompt) => {
if (String(prompt).includes('kind="agent-notification"'))
@@ -1933,14 +1931,8 @@ test(
await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(2);
expect(parentPrompts[1]).toContain('event="finished"');
await expect
// Provider start precedes dispatch acknowledgment and the durable removal commit.
.poll(
async () =>
JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")),
{
timeout: 10000,
message: "Parent accepted completion; waiting for durable intent removal",
},
.poll(async () =>
JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")),
)
.toEqual([]);
},
@@ -219,7 +219,7 @@ test(
);
test(
"real Pi daemon accepts a prompt after manual compact",
"real Pi daemon executes manual compact out-of-band instead of prompt text",
async () => {
const cwd = tmpCwd("pi-manual-compact-");
@@ -234,24 +234,20 @@ test(
cwd,
title: "pi-manual-compact",
provider: "pi",
model: PI_REAL_TEST_MODEL,
model: PI_COMPACTION_TEST_MODEL,
});
await client.sendMessage(agent.id, "Reply exactly: compact-ready");
const finish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS);
expect(finish.status, JSON.stringify(await fetchCanonicalTimeline(client, agent.id))).toBe(
"idle",
);
expect(finish.status).toBe("idle");
await client.sendMessage(agent.id, "/compact summarize this e2e run");
await waitForTimelineItem(
client,
agent.id,
(item) => item.type === "compaction" && item.status === "completed",
);
await client.sendMessage(agent.id, "Reply exactly: after-compaction");
const followUpFinish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS);
expect(followUpFinish.status).toBe("idle");
const items = await fetchCanonicalTimeline(client, agent.id);
expect(items).toEqual(
@@ -267,12 +263,9 @@ test(
}),
]),
);
expect(items).toContainEqual(
expect.objectContaining({
type: "assistant_message",
text: "after-compaction",
}),
);
expect(
items.some((item) => item.type === "user_message" && item.text.includes("/compact")),
).toBe(false);
expect(
items
.filter((item) => item.type === "assistant_message")
@@ -1,106 +0,0 @@
import { realpath } from "node:fs";
import { readlink } from "node:fs/promises";
import { promisify } from "node:util";
import pLimit from "p-limit";
import { dirname, resolve as resolvePath, sep } from "node:path";
import { looksLikeDefiniteWindowsPath, normalizePathPreservingCase } from "../utils/path.js";
const nativeRealpath = promisify(realpath.native);
const jsRealpath = promisify(realpath);
const identityIo = pLimit(2);
/** A subscription owns its lexical namespace, including the OS's real-root alias. */
export class GitObservationPaths {
private constructor(
private readonly root: string,
private readonly aliases: string[],
private readonly windows: boolean,
) {}
static async create(root: string): Promise<GitObservationPaths> {
const windows = looksLikeDefiniteWindowsPath(root);
const normalizePath = (value: string) =>
windows
? normalizePathPreservingCase(value, true).replaceAll("\\", "/").toLowerCase()
: normalizePathPreservingCase(value, false);
const aliases = new Set([normalizePath(root)]);
// Sequential: foreign paths must not fan out into the shared libuv pool.
for (const resolve of [nativeRealpath, jsRealpath]) {
try {
aliases.add(normalizePath(await identityIo(() => resolve(root))));
} catch {
/* Missing roots retain lexical identity. */
}
}
// A link may name /var/... while realpath returns /private/var/... on macOS.
// Retain literal targets too, including chains, without following links in callbacks.
let current = root;
const visited = new Set<string>();
for (let remaining = 40; remaining > 0; remaining--) {
const alias = normalizePath(current);
if (visited.has(alias)) break;
visited.add(alias);
aliases.add(alias);
try {
const target = await identityIo(() => readlink(current));
current = resolvePath(dirname(current), target);
} catch {
break;
}
}
return new GitObservationPaths(root, [...aliases], windows);
}
relative(candidate: string): string | null {
const normalized = this.normalizeEventPath(candidate);
const comparable = this.windows ? normalized.toLowerCase() : normalized;
for (const alias of this.aliases) {
if (comparable === alias) return "";
const prefix = alias.endsWith("/") ? alias : `${alias}/`;
if (comparable.startsWith(prefix)) {
// Slice the original spelling, preserving case in ref names on Windows.
return normalized.slice(prefix.length);
}
}
return null;
}
// Compile excluded subtrees in every established root namespace. The event callback
// only compares strings; neither an ignored root nor an event needs another realpath.
createEventFilter(excludedRoots: Iterable<string>): (candidate: string) => boolean {
const excluded: string[] = [];
for (const root of excludedRoots) {
const relative = this.relative(root);
if (relative === null) continue;
for (const alias of this.aliases) {
const subtree = relative ? `${alias}/${relative}` : alias;
excluded.push(this.windows ? subtree.toLowerCase() : subtree);
}
}
const prefixes = excluded.map((root) => `${root}/`);
return (candidate) => {
const normalized = this.normalizeEventPath(candidate);
const comparable = this.windows ? normalized.toLowerCase() : normalized;
if (this.aliases.includes(comparable)) return false;
for (let i = 0; i < excluded.length; i++) {
if (comparable === excluded[i] || comparable.startsWith(prefixes[i])) return false;
}
return true;
};
}
private normalizeEventPath(candidate: string): string {
if (this.windows) return normalizePathPreservingCase(candidate, true).replaceAll("\\", "/");
if (candidate.includes("/.") || candidate.includes("//") || candidate.endsWith("/"))
return normalizePathPreservingCase(candidate, false);
return candidate;
}
matches = (candidate: string): boolean => this.relative(candidate) === "";
toWatchPath(candidate: string): string {
const relative = this.relative(candidate);
if (relative === null) return candidate;
return relative ? `${this.root}${sep}${relative}` : this.root;
}
}
@@ -548,7 +548,7 @@ test("renames an eligible unnamed branch-off worktree once on first agent contex
);
expect(created.worktree.branchName).toBe("dazzling-yak");
expect(await readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
expect(readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
version: 2,
firstAgentBranchAutoName: {
status: "pending",
@@ -575,7 +575,7 @@ test("renames an eligible unnamed branch-off worktree once on first agent contex
branchName: "renamed-from-agent-context",
});
expect(branchAfterFirst).toBe("renamed-from-agent-context");
expect(await readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
expect(readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
version: 2,
firstAgentBranchAutoName: {
status: "attempted",
@@ -762,7 +762,7 @@ test("leaves the branch alone when generated branch text is invalid", async () =
.toString()
.trim(),
).toBe("dazzling-yak");
expect(await readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
expect(readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
version: 2,
firstAgentBranchAutoName: {
status: "attempted",
@@ -794,7 +794,7 @@ test("does not mark checkout branch worktrees as eligible for first-agent rename
// baseRefName must be the repo's default branch, not the checked-out branch
// itself — otherwise the diff panel compares the branch against itself and
// always shows no changes.
expect(await readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
expect(readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
version: 1,
baseRefName: "main",
});
@@ -831,7 +831,7 @@ test("does not mark GitHub PR checkout worktrees as eligible for first-agent ren
createDeps(),
);
expect(await readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
expect(readPaseoWorktreeMetadata(created.worktree.worktreePath)).toMatchObject({
version: 1,
baseRefName: "main",
});
@@ -77,7 +77,7 @@ async function createPaseoWorktreeWithPriority(
const workspaceCwdPlan = await planWorkspaceCwdForWorktree(input.cwd, deps.workspaceGitService);
const createdWorktree = await createWorktreeCore(input, deps);
try {
await maybeMarkFirstAgentBranchAutoNameEligible({ createdWorktree });
maybeMarkFirstAgentBranchAutoNameEligible({ createdWorktree });
const workspaceCwd = mapWorkspaceRelativeCwdToWorktree({
relativeWorkspaceCwd: workspaceCwdPlan.relativeWorkspaceCwd,
targetWorktreePath: createdWorktree.worktree.worktreePath,
@@ -182,9 +182,9 @@ export async function attemptFirstAgentBranchAutoName(options: {
return { attempted: false, renamed: false, branchName: null };
}
let metadata: Awaited<ReturnType<typeof readPaseoWorktreeMetadata>>;
let metadata: ReturnType<typeof readPaseoWorktreeMetadata>;
try {
metadata = await readPaseoWorktreeMetadata(options.cwd);
metadata = readPaseoWorktreeMetadata(options.cwd);
} catch {
return { attempted: false, renamed: false, branchName: null };
}
@@ -199,11 +199,11 @@ export async function attemptFirstAgentBranchAutoName(options: {
const getCurrentBranchImpl = options.getCurrentBranch ?? getCurrentBranch;
const placeholderBranchName = metadata.firstAgentBranchAutoName.placeholderBranchName;
if ((await getCurrentBranchImpl(options.cwd)) !== placeholderBranchName) {
await markPaseoWorktreeFirstAgentBranchAutoNameAttempted(options.cwd);
markPaseoWorktreeFirstAgentBranchAutoNameAttempted(options.cwd);
return { attempted: true, renamed: false, branchName: null };
}
await markPaseoWorktreeFirstAgentBranchAutoNameAttempted(options.cwd);
markPaseoWorktreeFirstAgentBranchAutoNameAttempted(options.cwd);
const branchName = await options.generateBranchNameFromContext({
cwd: options.cwd,
@@ -264,15 +264,15 @@ async function findAvailableBranchName(options: {
return null;
}
async function maybeMarkFirstAgentBranchAutoNameEligible(options: {
function maybeMarkFirstAgentBranchAutoNameEligible(options: {
createdWorktree: Awaited<ReturnType<typeof createWorktreeCore>>;
}): Promise<void> {
}): void {
const { createdWorktree } = options;
if (!createdWorktree.created || createdWorktree.intent.kind !== "branch-off") {
return;
}
await writePaseoWorktreeFirstAgentBranchAutoNameMetadata(createdWorktree.worktree.worktreePath, {
writePaseoWorktreeFirstAgentBranchAutoNameMetadata(createdWorktree.worktree.worktreePath, {
placeholderBranchName: createdWorktree.worktree.branchName,
});
}
@@ -0,0 +1,752 @@
import { fileURLToPath } from "node:url";
import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, test } from "vitest";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { DaemonClient } from "../test-utils/daemon-client.js";
import {
createTestAgentClient,
type TestAgentClientOptions,
} from "../test-utils/fake-agent-client.js";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { BuiltinPluginLoader } from "./builtin/index.js";
import { serializePaseoToolInputParameters } from "../agent/tools/paseo-tool-serialization.js";
import type {
AgentClient,
AgentLaunchContext,
AgentSessionConfig,
AgentPromptInput,
} from "../agent/agent-sdk-types.js";
function recordingProvider(native: boolean, callbacks: TestAgentClientOptions = {}) {
const launches: { config: AgentSessionConfig; context?: AgentLaunchContext }[] = [];
const turns: { prompt: AgentPromptInput; launchCount: number }[] = [];
const steers: AgentPromptInput[] = [];
const events = { interrupts: 0 };
const inner = createTestAgentClient("codex", {
...callbacks,
supportsMcpServers: true,
onStartTurn: (prompt, config) => {
turns.push({ prompt, launchCount: launches.length });
callbacks.onStartTurn?.(prompt, config);
},
steerActiveTurn: async (prompt) => {
steers.push(prompt);
return { status: "accepted" };
},
onInterrupt: () => {
events.interrupts++;
callbacks.onInterrupt?.();
},
});
const provider: AgentClient = {
provider: inner.provider,
capabilities: { ...inner.capabilities, supportsNativePaseoTools: native },
createSession: async (config, context, options) => {
launches.push({ config, context });
return inner.createSession(config, context, options);
},
resumeSession: async (handle, config, context, options) => {
launches.push({ config: config as AgentSessionConfig, context });
return inner.resumeSession(handle, config, context, options);
},
isAvailable: () => inner.isAvailable(),
fetchCatalog: (options) => inner.fetchCatalog(options),
};
return { provider, launches, turns, steers, events };
}
const pluginSource = `
import { z } from "zod";
import { defineAgentTool } from "@getpaseo/plugin/server";
let calls = 0;
const tool = defineAgentTool({
name: "plugin_echo", description: "Echo for the chosen agent",
input: z.object({ text: z.string().min(1).refine(value => value !== "forbidden", "Forbidden by plugin").describe("Message") }),
async execute(input, { agent, paseo, signal }) {
calls++;
if (input.text === "throw") throw new Error("handler failed");
return { text: JSON.stringify({ agent, input, calls, sdk: typeof paseo.agents.ref, aborted: signal.aborted }) };
},
});
export default function contribute(server) {
server.before("agent.session_open", ({ request }) => {
if (request.labels.x !== "y") return;
return { ...request, env: { ...request.env, OPEN_INFO: JSON.stringify({ title: request.title, labels: request.labels, reason: request.reason }) }, tools: [...request.tools, tool] };
});
return () => {};
}
`;
async function setup(
native = false,
source = pluginSource,
policy?: { enabled?: boolean; disabledTools?: string[] },
providerOptions?: TestAgentClientOptions,
) {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-agent-tools-"));
const recorder = recordingProvider(native, providerOptions);
const daemon = await createTestPaseoDaemon({
daemonVersion: "0.13.2",
agentClients: { codex: recorder.provider },
mcpEnabled: true,
providerOverrides: { codex: { paseoTools: policy } },
});
const client = new DaemonClient({
url: `ws://127.0.0.1:${daemon.port}/ws`,
appVersion: "0.13.2",
});
const mcps: Client[] = [];
async function install(id: string, code: string) {
const pluginDirectory = path.join(directory, id);
await mkdir(pluginDirectory);
await writeFile(
path.join(pluginDirectory, "paseo-plugin.json"),
JSON.stringify({ id, requirements: { paseo: ">=0.13.2" } }),
);
await writeFile(path.join(pluginDirectory, "index.server.ts"), code);
await client.installDirectoryPlugin(pluginDirectory);
}
await client.connect();
await client.fetchAgents({ subscribe: {} });
await client.patchDaemonConfig({ pluginsEnabled: true });
try {
await install("tools-test", source);
} catch (error) {
await client.close();
await daemon.close();
await rm(directory, { recursive: true, force: true });
throw error;
}
return {
...recorder,
daemon,
client,
directory,
install,
async create(labels: Record<string, string> = {}) {
return daemon.daemon.agentManager.createAgent(
{ provider: "codex", cwd: directory },
undefined,
{ labels, initialTitle: "Chosen title" },
);
},
async mcp(agentId?: string) {
const server = recorder.launches.at(-1)!.config.mcpServers?.paseo;
let url: string;
if (agentId) url = `http://127.0.0.1:${daemon.port}/mcp/agents?callerAgentId=${agentId}`;
else {
if (server?.type !== "http") throw new Error("Expected HTTP server");
url = server.url;
}
const mcp = new Client({ name: "plugin-tools-test", version: "1" });
mcps.push(mcp);
await mcp.connect(new StreamableHTTPClientTransport(new URL(url)));
return mcp;
},
async close() {
await Promise.all(mcps.map((mcp) => mcp.close()));
await client.close();
await daemon.close();
await rm(directory, { recursive: true, force: true });
},
};
}
test("session hooks attach tools only to matching agents over MCP, with validated input and caller context", async () => {
const h = await setup();
try {
const a = await h.create({ x: "y" });
expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO)).toEqual({
title: "Chosen title",
labels: { x: "y" },
reason: "create",
});
const mcpA = await h.mcp();
const listed = (await mcpA.listTools()).tools.find((tool) => tool.name === "plugin_echo");
expect(listed).toMatchObject({
name: "plugin_echo",
inputSchema: {
type: "object",
properties: { text: { type: "string", minLength: 1, description: "Message" } },
required: ["text"],
},
});
const b = await h.create();
const mcpB = await h.mcp();
expect((await mcpB.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
const invalid = await mcpA.callTool({ name: "plugin_echo", arguments: { text: "" } });
expect(invalid.isError).toBe(true);
expect(JSON.stringify(invalid)).toContain("text");
expect(
await mcpA.callTool({ name: "plugin_echo", arguments: { text: "forbidden" } }),
).toMatchObject({
isError: true,
content: [{ type: "text", text: expect.stringContaining("Forbidden by plugin") }],
});
const result = await mcpA.callTool({ name: "plugin_echo", arguments: { text: "hello" } });
expect(result.isError).not.toBe(true);
expect(JSON.parse((result.content as { text: string }[])[0].text)).toMatchObject({
agent: { id: a.id, cwd: h.directory, title: "Chosen title" },
input: { text: "hello" },
calls: 1,
sdk: "function",
aborted: false,
});
expect(a.id).not.toBe(b.id);
expect(
await mcpA.callTool({ name: "plugin_echo", arguments: { text: "throw" } }),
).toMatchObject({ isError: true, content: [{ type: "text", text: "handler failed" }] });
expect(
(await mcpA.callTool({ name: "plugin_echo", arguments: { text: "still alive" } })).isError,
).not.toBe(true);
} finally {
await h.close();
}
}, 60_000);
test("native providers get the same attached schema and execution", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const catalog = h.launches.at(-1)!.context!.paseoTools!;
expect(catalog.getTool("plugin_echo")?.description).toBe("Echo for the chosen agent");
const mcp = await h.mcp(a.id);
const advertised = (await mcp.listTools()).tools.find((tool) => tool.name === "plugin_echo")!;
expect(serializePaseoToolInputParameters(catalog.getTool("plugin_echo")!)).toEqual(
advertised.inputSchema,
);
const result = await catalog.executeTool("plugin_echo", { text: "native" });
expect(JSON.parse(result.content[0].text!)).toMatchObject({
agent: { id: a.id },
input: { text: "native" },
});
} finally {
await h.close();
}
}, 60_000);
test("resume uses the stored title and labels, and close removes the attachment", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const catalog = h.launches.at(-1)!.context!.paseoTools!;
const handle = a.persistence!;
await h.daemon.daemon.agentManager.setTitle(a.id, "Stored title");
await h.daemon.daemon.agentManager.closeAgent(a.id);
expect(h.daemon.daemon.agentManager.getPluginTools(a.id)).toEqual([]);
await expect(catalog.executeTool("plugin_echo", { text: "closed" })).rejects.toThrow(
"Agent is not live",
);
await h.daemon.daemon.agentManager.resumeAgentFromPersistence(handle, a.config, a.id);
expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO)).toEqual({
title: "Stored title",
labels: { x: "y" },
reason: "resume",
});
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")?.name).toBe(
"plugin_echo",
);
} finally {
await h.close();
}
}, 60_000);
test.each([
{
name: "Paseo",
source: pluginSource.replace('name: "plugin_echo"', 'name: "list_agents"'),
other: false,
message: /plugin tools-test.*Paseo tool list_agents/,
},
{
name: "another plugin",
source: pluginSource,
other: true,
message: /plugin tools-test.*plugin other-tools/,
},
{
name: "the same plugin",
source: pluginSource.replace("tools: [...request.tools, tool]", "tools: [tool, tool]"),
other: false,
message: /Duplicate agent tool: plugin_echo/,
},
])(
"collisions with $name fail only targeted session opens",
async ({ source, other, message }) => {
const h = await setup(false, source);
try {
if (other) await h.install("other-tools", pluginSource);
await expect(h.create({ x: "y" })).rejects.toThrow(message);
expect(h.launches).toHaveLength(0);
await h.create();
expect(h.launches).toHaveLength(1);
expect((await (await h.mcp()).listTools()).tools.map((tool) => tool.name)).not.toContain(
"plugin_echo",
);
} finally {
await h.close();
}
},
60_000,
);
test("each plugin owns its tool array and callbacks can remove earlier attachments", async () => {
const source = pluginSource.replace(
" return () => {};",
`
server.before("agent.session_open", ({ request }) => {
if (request.labels.x === "y" && request.tools[0]?.name !== "plugin_echo") throw new Error("Lost plugin-local definitions");
return { ...request, tools: [] };
});
return () => {};`,
);
const h = await setup(false, source);
try {
await h.install(
"other-tools",
pluginSource
.replace('name: "plugin_echo"', 'name: "other_echo"')
.replace(
' if (request.labels.x !== "y") return;',
' if (request.tools.length !== 0) throw new Error("Saw another plugin tools");\n if (request.labels.x !== "y") return;',
),
);
await h.create({ x: "y" });
const names = (await (await h.mcp()).listTools()).tools.map((tool) => tool.name);
expect(names).toContain("other_echo");
expect(names).not.toContain("plugin_echo");
} finally {
await h.close();
}
}, 60_000);
test("disabled plugins remove tools from existing and new MCP catalogs", async () => {
const h = await setup();
try {
await h.create({ x: "y" });
const mcp = await h.mcp();
expect((await mcp.listTools()).tools.map((tool) => tool.name)).toContain("plugin_echo");
await h.client.patchDaemonConfig({ pluginsEnabled: false });
expect((await mcp.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
await h.create({ x: "y" });
expect((await (await h.mcp()).listTools()).tools.map((tool) => tool.name)).not.toContain(
"plugin_echo",
);
} finally {
await h.close();
}
}, 60_000);
test.each([false, true])(
"provider tools disabled gates plugin tools (native=%s)",
async (native) => {
const h = await setup(native, pluginSource, { enabled: false });
try {
await h.create({ x: "y" });
expect(h.launches.at(-1)!.context!.paseoTools).toBeUndefined();
expect(h.launches.at(-1)!.config.mcpServers?.paseo).toBeUndefined();
} finally {
await h.close();
}
},
60_000,
);
test("disabledTools filters plugin tool names", async () => {
const h = await setup(true, pluginSource, { disabledTools: ["plugin_echo"] });
try {
await h.create({ x: "y" });
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")).toBeUndefined();
} finally {
await h.close();
}
}, 60_000);
const waitingSource = pluginSource
.replace("import { z }", 'import { writeFile } from "node:fs/promises";\nimport { z }')
.replace(
" calls++;",
`
calls++;
if (input.text === "wait") {
await writeFile(agent.cwd + "/started", agent.id);
await new Promise(resolve => signal.addEventListener("abort", resolve, { once: true }));
await writeFile(agent.cwd + "/aborted", agent.id);
return { text: "discard this result" };
}
`,
);
test("caller cancellation aborts the plugin handler and discards its result", async () => {
const h = await setup(true, waitingSource);
try {
const a = await h.create({ x: "y" });
const catalog = h.launches.at(-1)!.context!.paseoTools!;
const controller = new AbortController();
const call = catalog.executeTool(
"plugin_echo",
{ text: "wait" },
{ signal: controller.signal },
);
const rejected = expect(call).rejects.toThrow("Canceled by caller");
await expect.poll(() => readFile(path.join(h.directory, "started"), "utf8")).toBe(a.id);
controller.abort(new Error("Canceled by caller"));
await rejected;
await expect.poll(() => readFile(path.join(h.directory, "aborted"), "utf8")).toBe(a.id);
} finally {
await h.close();
}
}, 60_000);
test("a plugin stopping mid-call returns a tool error", async () => {
const h = await setup(true, waitingSource);
try {
const a = await h.create({ x: "y" });
const call = h.launches
.at(-1)!
.context!.paseoTools!.executeTool("plugin_echo", { text: "wait" });
await expect.poll(() => readFile(path.join(h.directory, "started"), "utf8")).toBe(a.id);
await h.client.patchDaemonConfig({ pluginsEnabled: false });
expect(await call).toMatchObject({
isError: true,
content: [{ type: "text", text: expect.stringContaining("Plugin stopped") }],
});
} finally {
await h.close();
}
}, 60_000);
test("history sessions ignore attached tools", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const handle = a.persistence!;
await h.client.archiveAgent(a.id);
await h.daemon.daemon.agentManager.resumeAgentFromPersistence(handle, a.config, a.id);
expect(h.daemon.daemon.agentManager.getPluginTools(a.id)).toEqual([]);
expect(h.launches.at(-1)!.context!.paseoTools?.getTool("plugin_echo")).toBeUndefined();
} finally {
await h.close();
}
}, 60_000);
test("disabled plugins attach no tools to a new native session", async () => {
const h = await setup(true);
try {
await h.client.patchDaemonConfig({ pluginsEnabled: false });
await h.create({ x: "y" });
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")).toBeUndefined();
} finally {
await h.close();
}
}, 60_000);
test("structured tool results get model-visible text when text is omitted", async () => {
const h = await setup(
true,
pluginSource.replace(
" calls++;",
' return { structuredContent: { status: "created" } };',
),
);
try {
await h.create({ x: "y" });
expect(
await h.launches
.at(-1)!
.context!.paseoTools!.executeTool("plugin_echo", { text: "structured" }),
).toEqual({
content: [{ type: "text", text: JSON.stringify({ status: "created" }, null, 2) }],
structuredContent: { status: "created" },
isError: undefined,
});
} finally {
await h.close();
}
}, 60_000);
test("reloading a plugin retires its old attachments before an agent reopens", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const mcp = await h.mcp(a.id);
const oldCatalog = h.launches.at(-1)!.context!.paseoTools!;
expect((await mcp.listTools()).tools.map((tool) => tool.name)).toContain("plugin_echo");
await h.client.reloadPlugin("tools-test");
expect((await mcp.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
expect(h.daemon.daemon.agentManager.getPluginTools(a.id)).toEqual([]);
expect(await oldCatalog.executeTool("plugin_echo", { text: "retired" })).toMatchObject({
isError: true,
content: [{ type: "text", text: expect.stringContaining("not available") }],
});
} finally {
await h.close();
}
}, 60_000);
test("built-in plugins attach and execute tools while configured plugins are disabled", async () => {
const root = await mkdtemp(path.join(tmpdir(), "paseo-builtin-tools-"));
const directory = path.join(root, "builtin-tools");
await mkdir(directory);
await writeFile(
path.join(directory, "paseo-plugin.json"),
JSON.stringify({ id: "builtin-tools", requirements: { paseo: ">=0.13.2" } }),
);
await writeFile(path.join(directory, "index.server.ts"), pluginSource);
const recorder = recordingProvider(true);
const daemon = await createTestPaseoDaemon({
daemonVersion: "0.13.2",
agentClients: { codex: recorder.provider },
pluginsEnabled: false,
builtinPlugins: new BuiltinPluginLoader(root, ["builtin-tools"]),
});
try {
const a = await daemon.daemon.agentManager.createAgent(
{ provider: "codex", cwd: root },
undefined,
{ labels: { x: "y" } },
);
const catalog = recorder.launches.at(-1)!.context!.paseoTools!;
expect(catalog.getTool("plugin_echo")?.name).toBe("plugin_echo");
const result = await catalog.executeTool("plugin_echo", { text: "builtin" });
expect(result.isError).not.toBe(true);
expect(JSON.parse(result.content[0].text!)).toMatchObject({
agent: { id: a.id },
input: { text: "builtin" },
});
} finally {
await daemon.close();
await rm(root, { recursive: true, force: true });
}
}, 60_000);
test("a session hook missing tools reports a schema issue", async () => {
const h = await setup(
false,
pluginSource.replace("tools: [...request.tools, tool]", "tools: undefined"),
);
try {
await expect(h.create({ x: "y" })).rejects.toThrow('"tools"');
} finally {
await h.close();
}
}, 60_000);
async function sendAndFinish(h: Awaited<ReturnType<typeof setup>>, agentId: string, text: string) {
const previous = h.turns.length;
await h.client.sendMessage(agentId, text, { activeTurnBehavior: "steer" });
await expect.poll(() => h.turns.length).toBe(previous + 1);
await expect.poll(() => h.daemon.daemon.agentManager.hasInFlightRun(agentId)).toBe(false);
}
test("an idle agent reopens once before its next prompt after a session-hook plugin reload", async () => {
const h = await setup();
try {
const b = await h.create();
const mcp = await h.mcp();
expect((await mcp.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
await writeFile(
path.join(h.directory, "tools-test", "index.server.ts"),
pluginSource.replace(' if (request.labels.x !== "y") return;', ""),
);
await h.client.reloadPlugin("tools-test");
expect(h.launches).toHaveLength(1);
await sendAndFinish(h, b.id, "first prompt");
expect(h.launches).toHaveLength(2);
expect(h.turns.map((turn) => turn.launchCount)).toEqual([2]);
expect((await mcp.listTools()).tools.map((tool) => tool.name)).toContain("plugin_echo");
await sendAndFinish(h, b.id, "second prompt");
expect(h.launches).toHaveLength(2);
expect(h.turns.map((turn) => turn.launchCount)).toEqual([2, 2]);
expect(h.events.interrupts).toBe(0);
} finally {
await h.close();
}
}, 60_000);
test("a busy agent is steered after plugin reload and reopens only at its next idle prompt", async () => {
let release!: () => void;
const blocked = new Promise<void>((resolve) => {
release = resolve;
});
const h = await setup(true, pluginSource, undefined, { beforeTurnComplete: async () => blocked });
try {
const a = await h.create({ x: "y" });
await h.client.sendMessage(a.id, "busy prompt", { activeTurnBehavior: "steer" });
await expect.poll(() => h.daemon.daemon.agentManager.hasInFlightRun(a.id)).toBe(true);
await expect
.poll(() => h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId)
.not.toBeNull();
const turnId = h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId;
await h.client.reloadPlugin("tools-test");
await h.client.sendMessage(a.id, "steering prompt", { activeTurnBehavior: "steer" });
expect(h.steers).toHaveLength(1);
expect(h.launches).toHaveLength(1);
expect(h.events.interrupts).toBe(0);
expect(h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId).toBe(turnId);
release();
await expect.poll(() => h.daemon.daemon.agentManager.hasInFlightRun(a.id)).toBe(false);
await sendAndFinish(h, a.id, "idle prompt");
expect(h.launches).toHaveLength(2);
expect(h.events.interrupts).toBe(0);
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")?.name).toBe(
"plugin_echo",
);
await sendAndFinish(h, a.id, "stable prompt");
expect(h.launches).toHaveLength(2);
} finally {
release();
await h.close();
}
}, 60_000);
const reloadRpcSource = pluginSource
.replace("import { z }", 'import { defineRpc } from "@getpaseo/plugin";\nimport { z }')
.replace(
" return () => {};",
`
server.handle(defineRpc({ name: "reload_agent", input: z.object({ id: z.string() }), output: z.object({ id: z.string() }) }), async ({ id }, { paseo }) => {
const agent = paseo.agents.ref(id);
await agent.refresh();
await agent.reload();
return { id: agent.current().id };
});
return () => {};
`,
);
test("a plugin can reload an agent through the SDK and rerun session hooks with reason refresh", async () => {
const h = await setup(true, reloadRpcSource);
try {
const a = await h.create({ x: "y" });
expect(h.launches).toHaveLength(1);
expect(await h.client.invokePluginRpc("tools-test", "reload_agent", { id: a.id })).toEqual({
id: a.id,
});
expect(h.launches).toHaveLength(2);
expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO).reason).toBe("refresh");
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")?.name).toBe(
"plugin_echo",
);
} finally {
await h.close();
}
}, 60_000);
test("starting and stopping a session-hook plugin each causes one idle reopen", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
await h.install(
"added-tools",
pluginSource.replace('name: "plugin_echo"', 'name: "added_echo"'),
);
await sendAndFinish(h, a.id, "after start");
expect(h.launches).toHaveLength(2);
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("added_echo")?.name).toBe("added_echo");
await h.client.disablePlugin("added-tools");
await sendAndFinish(h, a.id, "after stop");
expect(h.launches).toHaveLength(3);
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("added_echo")).toBeUndefined();
await sendAndFinish(h, a.id, "stable after stop");
expect(h.launches).toHaveLength(3);
expect(h.events.interrupts).toBe(0);
} finally {
await h.close();
}
}, 60_000);
test("plugins without a session-open hook do not make agent sessions stale", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
await h.install(
"no-session-hook",
"export default function contribute(server) { server.on('agent.created', () => {}); return () => {}; }",
);
await sendAndFinish(h, a.id, "after unrelated start");
expect(h.launches).toHaveLength(1);
await h.client.reloadPlugin("no-session-hook");
await sendAndFinish(h, a.id, "after unrelated reload");
expect(h.launches).toHaveLength(1);
await h.client.disablePlugin("no-session-hook");
await sendAndFinish(h, a.id, "after unrelated stop");
expect(h.launches).toHaveLength(1);
} finally {
await h.close();
}
}, 60_000);
test("explicit SDK reload interrupts a running turn before reopening its session", async () => {
let release!: () => void;
const blocked = new Promise<void>((resolve) => {
release = resolve;
});
const h = await setup(true, reloadRpcSource, undefined, {
beforeTurnComplete: async () => blocked,
onInterrupt: () => release(),
});
try {
const a = await h.create({ x: "y" });
await h.client.sendMessage(a.id, "running before explicit reload", {
activeTurnBehavior: "steer",
});
await expect
.poll(() => h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId)
.not.toBeNull();
expect(await h.client.invokePluginRpc("tools-test", "reload_agent", { id: a.id })).toEqual({
id: a.id,
});
expect(h.events.interrupts).toBe(1);
expect(h.launches).toHaveLength(2);
expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO).reason).toBe("refresh");
} finally {
release();
await h.close();
}
}, 60_000);
test("the agent-tools example appends notes only for labeled agents", async () => {
const h = await setup();
try {
await h.client.installDirectoryPlugin(
fileURLToPath(new URL("../../../../../plugin-examples/agent-tools", import.meta.url)),
);
const a = await h.create({ "notes/enabled": "true" });
const mcpA = await h.mcp(a.id);
expect((await mcpA.listTools()).tools.map((tool) => tool.name)).toContain(
"append_workspace_note",
);
expect(
await mcpA.callTool({
name: "append_workspace_note",
arguments: { note: "Run the targeted tests before committing." },
}),
).toMatchObject({
content: [
{ type: "text", text: `Saved note to ${path.join(h.directory, "AGENT_NOTES.md")}` },
],
structuredContent: { path: path.join(h.directory, "AGENT_NOTES.md") },
});
expect(await readFile(path.join(h.directory, "AGENT_NOTES.md"), "utf8")).toBe(
"Run the targeted tests before committing.\n",
);
await mcpA.callTool({
name: "append_workspace_note",
arguments: { note: "Keep existing notes." },
});
expect(await readFile(path.join(h.directory, "AGENT_NOTES.md"), "utf8")).toBe(
"Run the targeted tests before committing.\nKeep existing notes.\n",
);
await h.create();
const mcpB = await h.mcp();
expect((await mcpB.listTools()).tools.map((tool) => tool.name)).not.toContain(
"append_workspace_note",
);
} finally {
await h.close();
}
}, 60_000);
@@ -0,0 +1,46 @@
import { expect, test } from "vitest";
import { z } from "zod";
import { defineAgentTool } from "@getpaseo/plugin/server";
import { serializeAgentTool } from "./agent-tools.js";
import { serializePaseoToolInputParameters } from "../agent/tools/paseo-tool-serialization.js";
test("agent tool schemas survive the process boundary and catalog round trip", () => {
const input = z
.object({
title: z.string().min(1).describe("Issue title"),
state: z.enum(["open", "closed"]),
tags: z.array(z.string()).optional(),
nested: z.object({ enabled: z.boolean() }),
})
.describe("Create an issue");
const tool = defineAgentTool({
name: "create_issue",
description: "Create",
input,
execute: () => ({ text: "done" }),
});
const metadata = JSON.parse(JSON.stringify(serializeAgentTool(tool)));
const restored = z.fromJSONSchema(metadata.inputSchema);
expect(
serializePaseoToolInputParameters({
...metadata,
inputSchema: restored,
handler: async () => ({ content: [] }),
}),
).toEqual({
...metadata.inputSchema,
additionalProperties: {},
properties: {
...metadata.inputSchema.properties,
nested: { ...metadata.inputSchema.properties.nested, additionalProperties: {} },
},
});
expect(restored.parse({ title: "Bug", state: "open", nested: { enabled: true } })).toEqual({
title: "Bug",
state: "open",
nested: { enabled: true },
});
expect(
restored.safeParse({ title: "", state: "unknown", nested: { enabled: true } }).success,
).toBe(false);
});
@@ -0,0 +1,108 @@
import { z } from "zod";
import type {
AgentToolDefinition,
AgentToolResult,
PluginHookContext,
PluginSessionOpenRequest,
PluginHookAgent,
} from "@getpaseo/plugin/server";
import { toJsonSchemaCompat } from "@modelcontextprotocol/sdk/server/zod-json-schema-compat.js";
export const agentToolMetadataSchema = z
.object({
name: z
.string()
.regex(/^[a-z][a-z0-9_]*$/)
.max(64),
title: z.string().optional(),
description: z.string(),
inputSchema: z.record(z.string(), z.unknown()),
pluginId: z.string().optional(),
pluginInstanceId: z.string().optional(),
})
.strict();
export type AgentToolMetadata = z.infer<typeof agentToolMetadataSchema>;
export type AttachedAgentTool = AgentToolMetadata & { pluginId: string; pluginInstanceId: string };
export type SessionOpenWireRequest = Omit<PluginSessionOpenRequest, "tools"> & {
tools: AgentToolMetadata[];
};
export interface PluginAgentToolsRuntime {
isToolPluginAvailable(pluginId: string, pluginInstanceId: string): boolean;
detachTools(agentId: string): void;
callTool(
pluginId: string,
pluginInstanceId: string,
agent: PluginHookAgent,
name: string,
input: unknown,
signal?: AbortSignal,
): Promise<AgentToolResult>;
}
export function serializeAgentTool(tool: AgentToolDefinition): AgentToolMetadata {
return agentToolMetadataSchema.parse({
name: tool.name,
title: tool.title,
description: tool.description,
inputSchema: toJsonSchemaCompat(tool.input, { strictUnions: true, pipeStrategy: "input" }),
});
}
// This store lives only in the plugin process. Definitions and parsed values never cross IPC.
export class PluginAgentTools {
private readonly agents = new Map<string, AgentToolDefinition[]>();
private readonly calls = new Map<string, { agentId: string; controller: AbortController }>();
attach(agentId: string, tools: AgentToolDefinition[]): AgentToolMetadata[] {
const names = new Set<string>();
const metadata = tools.map((tool) => {
const serialized = serializeAgentTool(tool);
if (names.has(tool.name)) throw new Error(`Duplicate agent tool: ${tool.name}`);
names.add(tool.name);
return serialized;
});
this.agents.set(agentId, tools);
return metadata;
}
detach(agentId: string): void {
this.agents.delete(agentId);
for (const call of this.calls.values()) {
if (call.agentId === agentId) call.controller.abort();
}
}
cancel(requestId: string): void {
this.calls.get(requestId)?.controller.abort();
}
close(): void {
for (const call of this.calls.values()) call.controller.abort();
this.agents.clear();
}
async call(
requestId: string,
agent: PluginHookAgent,
name: string,
input: unknown,
paseo: PluginHookContext["paseo"],
): Promise<AgentToolResult> {
const controller = new AbortController();
this.calls.set(requestId, { agentId: agent.id, controller });
try {
const tool = this.agents.get(agent.id)?.find((definition) => definition.name === name);
if (!tool) throw new Error(`Agent tool not attached: ${name}`);
const parsed = await tool.input.parseAsync(input);
controller.signal.throwIfAborted();
const result = await tool.execute(parsed, { agent, paseo, signal: controller.signal });
controller.signal.throwIfAborted();
return result;
} catch (error) {
return { isError: true, text: error instanceof Error ? error.message : String(error) };
} finally {
this.calls.delete(requestId);
}
}
}
@@ -1,3 +1,4 @@
import type { PluginAgentToolsRuntime } from "./agent-tools.js";
import type { PluginLifecycle } from "./lifecycle/index.js";
import path from "node:path";
import { stat, rm } from "node:fs/promises";
@@ -37,6 +38,10 @@ import type { PluginUsageSourceMetadata } from "./plugin-process-protocol.js";
const BUILTIN_PROVIDER_ID_SET: ReadonlySet<string> = new Set(BUILTIN_PROVIDER_IDS);
interface PluginRuntimePort {
subscribeSessionOpenChanges?: PluginRuntime["subscribeSessionOpenChanges"];
isToolPluginAvailable?: PluginRuntime["isToolPluginAvailable"];
callTool?: PluginRuntime["callTool"];
detachTools?: PluginRuntime["detachTools"];
emit?: PluginLifecycle["emit"];
drainEvents?: PluginRuntime["drainEvents"];
before?: PluginLifecycle["before"];
@@ -131,6 +136,34 @@ export class PluginService {
this.runtime.emit?.(name, event);
};
subscribeSessionOpenChanges(listener: () => void): () => void {
return this.runtime.subscribeSessionOpenChanges?.(listener) ?? (() => {});
}
readonly isToolPluginAvailable: PluginAgentToolsRuntime["isToolPluginAvailable"] = (
pluginId,
pluginInstanceId,
) => {
return this.runtime.isToolPluginAvailable?.(pluginId, pluginInstanceId) ?? false;
};
readonly detachTools: PluginAgentToolsRuntime["detachTools"] = (agentId) => {
this.runtime.detachTools?.(agentId);
};
readonly callTool: PluginAgentToolsRuntime["callTool"] = async (
pluginId,
pluginInstanceId,
agent,
name,
input,
signal,
) => {
if (!this.isToolPluginAvailable(pluginId, pluginInstanceId) || !this.runtime.callTool)
return { isError: true, text: `Plugin is not available: ${pluginId}` };
return this.runtime.callTool(pluginId, pluginInstanceId, agent, name, input, signal);
};
async drainEvents(): Promise<void> {
await this.runtime.drainEvents?.();
}
@@ -93,8 +93,39 @@ test("session-open hooks reject changes to session identity instead of silently
reason: "resume",
purpose: "interactive",
env: {},
title: null,
labels: {},
tools: [],
},
paseo,
),
).rejects.toThrow("agent.session_open hooks can only change env");
});
test.each([{ title: "Changed" }, { labels: { changed: "true" } }])(
"session-open targeting metadata is read-only: %j",
async (patch) => {
const hooks = new PluginHookHandlers(() => {});
hooks.before("agent.session_open", ({ request }) => ({ ...request, ...patch }));
await expect(
hooks.invoke(
"operation",
"before",
"agent.session_open",
{
agentId: "agent",
workspaceId: null,
provider: "codex",
cwd: "/project",
title: "Original",
labels: {},
tools: [],
env: {},
reason: "create",
purpose: "interactive",
},
paseo,
),
).rejects.toThrow("agent.session_open hooks can only change env and tools");
},
);
@@ -1,3 +1,11 @@
import { isDeepStrictEqual } from "node:util";
import {
serializeAgentTool,
agentToolMetadataSchema,
type SessionOpenWireRequest,
type PluginAgentTools,
} from "../agent-tools.js";
import type { AgentToolDefinition } from "@getpaseo/plugin/server";
import type { AgentStreamEvent } from "../../agent/agent-sdk-types.js";
import { z } from "zod";
import { CreateAgentRequestMessageSchema } from "@getpaseo/protocol/messages";
@@ -15,6 +23,10 @@ import type {
import { WorkspaceCreateRequestSchema } from "@getpaseo/protocol/messages";
import type { PersistedWorkspaceRecord } from "../../workspace-registry.js";
export type PluginWireBeforeRequests = Omit<PluginBeforeRequests, "agent.session_open"> & {
"agent.session_open": SessionOpenWireRequest;
};
export const lifecycleEventNames = [
"agent.created",
"agent.closed",
@@ -38,28 +50,44 @@ const beforeSchemas = {
cwd: z.string(),
reason: z.enum(["create", "resume", "refresh", "import"]),
purpose: z.enum(["interactive", "history"]),
title: z.string().nullable(),
labels: z.record(z.string(), z.string()),
tools: z.array(agentToolMetadataSchema),
env: z.record(z.string(), z.string()),
})
.strict(),
"workspace.create": WorkspaceCreateRequestSchema.omit({ type: true, requestId: true }).strict(),
};
const sessionOpenLocalResultSchema = beforeSchemas["agent.session_open"].extend({
tools: z.array(
z.custom<AgentToolDefinition>(
(tool) =>
typeof tool === "object" &&
tool !== null &&
"input" in tool &&
"execute" in tool &&
typeof tool.execute === "function",
),
),
});
export interface PluginLifecycle {
emit<Name extends keyof PluginLifecycleEvents>(
name: Name,
event: PluginLifecycleEvents[Name],
): void;
before<Name extends keyof PluginBeforeRequests>(
before<Name extends keyof PluginWireBeforeRequests>(
name: Name,
request: PluginBeforeRequests[Name],
): Promise<PluginBeforeRequests[Name]>;
request: PluginWireBeforeRequests[Name],
): Promise<PluginWireBeforeRequests[Name]>;
}
export function validateBeforeRequest<Name extends keyof PluginBeforeRequests>(
export function validateBeforeRequest<Name extends keyof PluginWireBeforeRequests>(
name: Name,
value: unknown,
): PluginBeforeRequests[Name] {
return beforeSchemas[name].parse(value) as PluginBeforeRequests[Name];
): PluginWireBeforeRequests[Name] {
return beforeSchemas[name].parse(value) as PluginWireBeforeRequests[Name];
}
export function describeHookWorkspace(workspace: PersistedWorkspaceRecord): PluginHookWorkspace {
@@ -129,11 +157,11 @@ export function publishAgentStream(
}
}
export function validateBeforeResult<Name extends keyof PluginBeforeRequests>(
export function validateBeforeResult<Name extends keyof PluginWireBeforeRequests>(
name: Name,
input: PluginBeforeRequests[Name],
input: PluginWireBeforeRequests[Name],
output: unknown,
): PluginBeforeRequests[Name] {
): PluginWireBeforeRequests[Name] {
const result = validateBeforeRequest(name, output);
if (name === "agent.session_open") {
const previous = beforeSchemas["agent.session_open"].parse(input);
@@ -144,9 +172,11 @@ export function validateBeforeResult<Name extends keyof PluginBeforeRequests>(
previous.provider !== next.provider ||
previous.cwd !== next.cwd ||
previous.reason !== next.reason ||
previous.purpose !== next.purpose
previous.purpose !== next.purpose ||
previous.title !== next.title ||
!isDeepStrictEqual(previous.labels, next.labels)
) {
throw new Error("agent.session_open hooks can only change env");
throw new Error("agent.session_open hooks can only change env and tools");
}
}
if (name === "agent.create") {
@@ -168,7 +198,10 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
private readonly changed: () => void;
constructor(changed: () => void) {
private readonly agentTools: PluginAgentTools | undefined;
constructor(changed: () => void, agentTools?: PluginAgentTools) {
this.agentTools = agentTools;
this.changed = changed;
}
@@ -209,6 +242,9 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
throw new Error(`Unknown before hook: ${name}`);
}
const hookName = name as keyof import("@getpaseo/plugin/server").PluginBeforeRequests;
if (hookName === "agent.session_open") {
return await this.openSession(input, { paseo, signal: controller.signal });
}
let request = validateBeforeRequest(hookName, input);
for (const handler of this.transforms.get(name) ?? []) {
controller.signal.throwIfAborted();
@@ -216,9 +252,7 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
{ request: structuredClone(request) },
{ paseo, signal: controller.signal },
);
if (result !== undefined) {
request = validateBeforeResult(hookName, request, result);
}
if (result !== undefined) request = validateBeforeResult(hookName, request, result);
}
return request;
}
@@ -236,6 +270,31 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
}
}
private async openSession(
input: unknown,
context: PluginHookContext,
): Promise<SessionOpenWireRequest> {
let request = validateBeforeRequest("agent.session_open", input);
this.agentTools?.detach(request.agentId);
let tools: AgentToolDefinition[] = [];
for (const handler of this.transforms.get("agent.session_open") ?? []) {
context.signal.throwIfAborted();
const result = await handler(
{ request: { ...structuredClone(request), tools: [...tools] } },
context,
);
if (result === undefined) continue;
const local = sessionOpenLocalResultSchema.parse(result);
const metadata = local.tools.map(serializeAgentTool);
request = validateBeforeResult("agent.session_open", request, { ...local, tools: metadata });
tools = local.tools;
}
context.signal.throwIfAborted();
if (request.purpose === "history") request.tools = [];
else this.agentTools?.attach(request.agentId, tools);
return request;
}
cancel(id: string): void {
this.active.get(id)?.abort();
}
@@ -1,3 +1,4 @@
import type { PluginHookAgent } from "@getpaseo/plugin/server";
import { UsageScopeSchema, type UsageScope } from "@getpaseo/plugin/server/usage";
import type {
ProviderConnectRequest,
@@ -31,6 +32,9 @@ export interface PluginUsageSourceMetadata {
}
export type PluginProcessRequest =
| { type: "tool.call"; requestId: string; agent: PluginHookAgent; name: string; input: unknown }
| { type: "tool.cancel"; requestId: string }
| { type: "tools.detach"; agentId: string }
| {
type: "provider.status";
requestId: string;
@@ -141,6 +145,26 @@ const frameFields = {
export const PluginProcessRequestSchema: z.ZodType<PluginProcessRequest> = z.discriminatedUnion(
"type",
[
z
.object({
type: z.literal("tool.call"),
requestId: z.string(),
agent: z
.object({
id: z.string(),
workspaceId: z.string().nullable(),
parentAgentId: z.string().nullable(),
provider: z.string(),
cwd: z.string(),
title: z.string().nullable(),
})
.strict(),
name: z.string(),
input: z.unknown(),
})
.strict(),
z.object({ type: z.literal("tool.cancel"), requestId: z.string() }).strict(),
z.object({ type: z.literal("tools.detach"), agentId: z.string() }).strict(),
z
.object({
type: z.literal("provider.status"),
@@ -1,3 +1,4 @@
import { PluginAgentTools } from "./agent-tools.js";
import { PluginHookHandlers } from "./lifecycle/index.js";
import { evaluateBundle } from "./bundle-evaluator.js";
import {
@@ -59,9 +60,10 @@ export function createPluginWorker(options: {
handler: RpcHandler;
}
const agentTools = new PluginAgentTools();
const hooks = new PluginHookHandlers(() => {
send({ type: "hooks.changed", hooks: hooks.catalog() });
});
}, agentTools);
const handlers = new Map<string, RegisteredRpc>();
const providers = new Map<string, ProviderRegistration>();
const usageSources = new Map<string, UsageSourceRegistration>();
@@ -322,6 +324,7 @@ export function createPluginWorker(options: {
?.dispose()
.catch((error) => console.error("Plugin API cleanup failed", error));
hooks.close();
agentTools.close();
for (const pending of pendingProviderConnections.values()) pending.tombstoned = true;
const currentCleanup = cleanup;
cleanup = null;
@@ -381,6 +384,42 @@ export function createPluginWorker(options: {
}
}
function handleToolMessage(
message: PluginProcessRequest,
): message is Extract<
PluginProcessRequest,
{ type: "tool.call" | "tool.cancel" | "tools.detach" }
> {
if (message.type === "tools.detach") {
agentTools.detach(message.agentId);
return true;
}
if (message.type === "tool.cancel") {
agentTools.cancel(message.requestId);
return true;
}
if (message.type === "tool.call") {
if (!paseo) {
send({ type: "error", requestId: message.requestId, error: "Plugin API unavailable" });
return true;
}
void agentTools
.call(message.requestId, message.agent, message.name, message.input, paseo)
.then((output) =>
send({
type: "result",
requestId: message.requestId,
output: jsonTransportValue(output),
}),
)
.catch((error) =>
send({ type: "error", requestId: message.requestId, error: describeError(error) }),
);
return true;
}
return false;
}
function handleMessage(rawMessage: unknown): void {
const parsed = PluginProcessRequestSchema.safeParse(rawMessage);
if (!parsed.success) {
@@ -422,6 +461,7 @@ export function createPluginWorker(options: {
rejectWhileStopping(message);
return;
}
if (handleToolMessage(message)) return;
if (message.type === "provider.status") {
void (async () => {
const provider = providers.get(message.providerId);
+129 -12
View File
@@ -1,10 +1,16 @@
import type { PluginWireBeforeRequests } from "./lifecycle/index.js";
import type { SessionOpenWireRequest, AttachedAgentTool } from "./agent-tools.js";
import type { UsageScope } from "@getpaseo/plugin/server/usage";
import {
ProviderStatusSchema,
type ProviderStatus,
type ProviderStatusRequest,
} from "@getpaseo/plugin/server/provider";
import type { PluginBeforeRequests, PluginLifecycleEvents } from "@getpaseo/plugin/server";
import type {
PluginLifecycleEvents,
PluginHookAgent,
AgentToolResult,
} from "@getpaseo/plugin/server";
import { validateBeforeRequest, validateBeforeResult } from "./lifecycle/index.js";
import { fork } from "node:child_process";
import { stat } from "node:fs/promises";
@@ -40,6 +46,7 @@ import { evaluateBundle } from "./bundle-evaluator.js";
const CLIENT_ENTRY_FILENAMES = ["index.client.ts", "index.client.tsx"] as const;
const SERVER_ENTRY_FILENAMES = ["index.server.ts", "index.server.tsx"] as const;
const REQUEST_TIMEOUT_MS = 30_000;
const TOOL_CALL_TIMEOUT_MS = 10 * 60_000;
const MAX_LOG_ENTRIES = 500;
const MAX_LOG_BYTES = 256 * 1024;
const MAX_LOG_LINE_BYTES = 16 * 1024;
@@ -69,6 +76,7 @@ interface PendingInvocation {
interface LoadedPlugin {
id: string;
instanceId: string;
requirements: PluginRequirements | undefined;
clientBundle: string;
methods: ReadonlySet<string>;
@@ -302,6 +310,7 @@ export class PluginRuntime {
private readonly spawnChild: () => PluginChild;
private sessionHost: PluginPaseoSessionHost | null;
private readonly listeners = new Set<(pluginId: string, error?: string) => void>();
private readonly sessionOpenListeners = new Set<() => void>();
constructor(
logger: pino.Logger,
@@ -324,6 +333,16 @@ export class PluginRuntime {
return () => this.listeners.delete(listener);
}
subscribeSessionOpenChanges(listener: () => void): () => void {
this.sessionOpenListeners.add(listener);
return () => this.sessionOpenListeners.delete(listener);
}
private notifySessionOpenChange(loaded: LoadedPlugin): void {
if (!loaded.hooks.before.includes("agent.session_open")) return;
for (const listener of this.sessionOpenListeners) listener();
}
async startPlugin(
pluginId: string,
configuredPath: string,
@@ -340,6 +359,7 @@ export class PluginRuntime {
throw new Error(`Plugin start cancelled: ${pluginId}`);
}
this.plugins.set(pluginId, loaded);
this.notifySessionOpenChange(loaded);
this.appendLog(pluginId, "stdout", "[paseo] Plugin ready");
}
@@ -363,6 +383,7 @@ export class PluginRuntime {
clientBundle: bundles.clientBundle ?? "",
});
this.plugins.set(input.id, loaded);
this.notifySessionOpenChange(loaded);
this.appendLog(input.id, "stdout", "[paseo] Plugin ready");
}
@@ -378,6 +399,7 @@ export class PluginRuntime {
const loaded = this.plugins.get(pluginId);
if (!loaded) return false;
this.plugins.delete(pluginId);
this.notifySessionOpenChange(loaded);
this.rejectPending(loaded, `Plugin stopped: ${pluginId}`);
await this.stopPlugin(loaded);
return true;
@@ -546,14 +568,15 @@ export class PluginRuntime {
await Promise.all(this.pendingEvents);
}
async before<Name extends keyof PluginBeforeRequests>(
async before<Name extends keyof PluginWireBeforeRequests>(
name: Name,
input: PluginBeforeRequests[Name],
): Promise<PluginBeforeRequests[Name]> {
input: PluginWireBeforeRequests[Name],
): Promise<PluginWireBeforeRequests[Name]> {
let request = validateBeforeRequest(name, input);
const plugins = [...this.plugins.values()].sort((left, right) => {
return left.id.localeCompare(right.id);
});
const attached: AttachedAgentTool[] = [];
for (const loaded of plugins) {
if (!loaded.hooks.before.includes(name)) {
continue;
@@ -564,15 +587,21 @@ export class PluginRuntime {
requestId: randomUUID(),
kind: "before",
name,
input: request,
input: name === "agent.session_open" ? { ...request, tools: [] } : request,
});
request = validateBeforeResult(name, request, output);
if (name === "agent.session_open") {
this.collectAgentTools(attached, loaded, (request as SessionOpenWireRequest).tools);
}
} catch (error) {
if (name === "agent.session_open")
this.detachTools((request as SessionOpenWireRequest).agentId);
throw new Error(`Plugin ${loaded.id} before ${name} failed: ${describeError(error)}`, {
cause: error,
});
}
}
if (name === "agent.session_open") (request as SessionOpenWireRequest).tools = attached;
return request;
}
@@ -594,25 +623,102 @@ export class PluginRuntime {
return output;
}
private collectAgentTools(
attached: AttachedAgentTool[],
loaded: LoadedPlugin,
tools: SessionOpenWireRequest["tools"],
): void {
for (const tool of tools) {
const collision = attached.find((entry) => entry.name === tool.name);
if (collision)
throw new Error(
`Agent tool ${tool.name} from plugin ${loaded.id} collides with plugin ${collision.pluginId}`,
);
attached.push({ ...tool, pluginId: loaded.id, pluginInstanceId: loaded.instanceId });
}
}
isToolPluginAvailable(pluginId: string, pluginInstanceId: string): boolean {
return this.plugins.get(pluginId)?.instanceId === pluginInstanceId;
}
detachTools(agentId: string): void {
for (const loaded of this.plugins.values()) {
if (loaded.child) void send(loaded.child, { type: "tools.detach", agentId }).catch(() => {});
}
}
async callTool(
pluginId: string,
pluginInstanceId: string,
agent: PluginHookAgent,
name: string,
input: unknown,
signal?: AbortSignal,
): Promise<AgentToolResult> {
const loaded = this.plugins.get(pluginId);
if (!loaded || loaded.instanceId !== pluginInstanceId)
return { isError: true, text: `Plugin is not available: ${pluginId}` };
const requestId = randomUUID();
try {
return (await this.request(
loaded,
{ type: "tool.call", requestId, agent, name, input },
signal,
)) as AgentToolResult;
} catch (error) {
if (signal?.aborted) throw error;
return { isError: true, text: describeError(error) };
}
}
private request(
loaded: LoadedPlugin,
message: Extract<PluginProcessRequest, { requestId: string }>,
signal?: AbortSignal,
): Promise<unknown> {
const child = loaded.child;
const pluginId = loaded.id;
if (!child) throw new Error(`Plugin ${pluginId} has no server entry`);
const requestId = message.requestId;
signal?.throwIfAborted();
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
const cancel = () => {
clearTimeout(timeout);
loaded.pending.delete(requestId);
if (message.type === "hook") {
void send(child, { type: "hook.cancel", requestId }).catch(() => {});
}
reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
}, REQUEST_TIMEOUT_MS);
loaded.pending.set(requestId, { resolve, reject, timeout });
if (message.type === "tool.call")
void send(child, { type: "tool.cancel", requestId }).catch(() => {});
signal?.removeEventListener("abort", cancel);
reject(signal?.reason ?? new Error("Tool call canceled"));
};
const timeout = setTimeout(
() => {
loaded.pending.delete(requestId);
signal?.removeEventListener("abort", cancel);
if (message.type === "tool.call")
void send(child, { type: "tool.cancel", requestId }).catch(() => {});
if (message.type === "hook") {
void send(child, { type: "hook.cancel", requestId }).catch(() => {});
}
reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
},
message.type === "tool.call" ? TOOL_CALL_TIMEOUT_MS : REQUEST_TIMEOUT_MS,
);
signal?.addEventListener("abort", cancel, { once: true });
loaded.pending.set(requestId, {
resolve: (value) => {
signal?.removeEventListener("abort", cancel);
resolve(value);
},
reject: (error) => {
signal?.removeEventListener("abort", cancel);
reject(error);
},
timeout,
});
void send(child, message).catch((error) => {
clearTimeout(timeout);
signal?.removeEventListener("abort", cancel);
loaded.pending.delete(requestId);
reject(error);
});
@@ -623,6 +729,7 @@ export class PluginRuntime {
const loaded = [...this.plugins.values()];
this.plugins.clear();
for (const plugin of loaded) {
this.notifySessionOpenChange(plugin);
this.rejectPending(plugin, `Plugin stopped: ${plugin.id}`);
}
await Promise.all(loaded.map((plugin) => this.stopPlugin(plugin)));
@@ -641,6 +748,7 @@ export class PluginRuntime {
if (!serverBundle) {
return {
id: pluginId,
instanceId: randomUUID(),
clientBundle: bundles.clientBundle ?? "",
requirements: manifest.requirements,
methods: new Set(),
@@ -769,6 +877,7 @@ export class PluginRuntime {
}
loaded = {
id: pluginId,
instanceId: randomUUID(),
clientBundle,
requirements,
methods: new Set(ready.methods),
@@ -852,7 +961,14 @@ export class PluginRuntime {
private handleChildMessage(loaded: LoadedPlugin, message: PluginProcessMessage): void {
if (message.type === "hooks.changed") {
const hadSessionOpen = loaded.hooks.before.includes("agent.session_open");
loaded.hooks = message.hooks;
if (
this.plugins.get(loaded.id) === loaded &&
hadSessionOpen !== loaded.hooks.before.includes("agent.session_open")
) {
for (const listener of this.sessionOpenListeners) listener();
}
return;
}
if (message.type === "settings.changed") {
@@ -1166,6 +1282,7 @@ export class PluginRuntime {
const wasPublished = this.plugins.get(loaded.id) === loaded;
if (wasPublished) {
this.plugins.delete(loaded.id);
this.notifySessionOpenChange(loaded);
}
this.rejectPending(loaded, `Plugin process exited: ${loaded.id}`);
for (const state of loaded.providerConnections.values()) {
@@ -226,6 +226,7 @@ type ScheduleAgentManager = Pick<
AgentRunController,
| "getAgent"
| "reloadAgentSession"
| "reloadStalePluginSession"
| "tryRunOutOfBand"
| "hasInFlightRun"
| "replaceAgentRun"
@@ -9228,8 +9228,8 @@ test("workspace auto-name uses the backing root for a nested worktree", async ()
execFileSync("git", ["add", "README.md"], { cwd: repoDir, stdio: "pipe" });
execFileSync("git", ["commit", "-m", "initial"], { cwd: repoDir, stdio: "pipe" });
execFileSync("git", ["branch", "-M", "placeholder-branch"], { cwd: repoDir, stdio: "pipe" });
await writePaseoWorktreeMetadata(repoDir, { baseRefName: "main" });
await writePaseoWorktreeFirstAgentBranchAutoNameMetadata(repoDir, {
writePaseoWorktreeMetadata(repoDir, { baseRefName: "main" });
writePaseoWorktreeFirstAgentBranchAutoNameMetadata(repoDir, {
placeholderBranchName: "placeholder-branch",
});
const workspaceCwd = path.join(repoDir, "packages", "app");
@@ -9288,7 +9288,6 @@ test("workspace auto-name uses the backing root for a nested worktree", async ()
firstAgentContext: { prompt: "Fix checkout title" },
});
await vi.runAllTimersAsync();
await vi.waitFor(() => expect(emittedCwds).toEqual([workspaceCwd]));
expect(generateCalls).toBe(1);
expect(stored.get(workspace.workspaceId)).toMatchObject({
@@ -9300,7 +9299,7 @@ test("workspace auto-name uses the backing root for a nested worktree", async ()
.toString()
.trim(),
).toBe("placeholder-branch");
expect(await readPaseoWorktreeMetadata(repoDir)).toMatchObject({
expect(readPaseoWorktreeMetadata(repoDir)).toMatchObject({
version: 2,
firstAgentBranchAutoName: {
status: "attempted",
@@ -347,7 +347,7 @@ async function createBaseRecoveryFixture(baseBranch: string | null) {
test("preserves ordinary checkout behavior when no separate base was recorded", async () => {
const fixture = await createBaseRecoveryFixture(null);
await writePaseoWorktreeMetadata(fixture.workspace.cwd, { baseRefName: "feature" });
writePaseoWorktreeMetadata(fixture.workspace.cwd, { baseRefName: "feature" });
await fixture.archiveAndRemove();
await fixture.service.restore(fixture.workspace.workspaceId);
expect((await fixture.registry.get(fixture.workspace.workspaceId))?.baseBranch).toBeNull();
@@ -430,7 +430,7 @@ test("restores the branch HEAD with a base name when the exact base is missing",
});
expect(existsSync(fixture.workspace.cwd)).toBe(true);
expect((await fixture.registry.get(fixture.workspace.workspaceId))?.archivedAt).toBeNull();
const metadata = await readPaseoWorktreeMetadata(fixture.workspace.cwd);
const metadata = readPaseoWorktreeMetadata(fixture.workspace.cwd);
expect(metadata).toMatchObject({ baseRefName: "main" });
expect(metadata?.baseRef).toBeUndefined();
expect(
@@ -450,7 +450,7 @@ test("restore fetches a deleted local branch from origin and retains its base",
await fixture.archiveAndRemove();
execFileSync("git", ["branch", "-D", "feature"], { cwd: fixture.repoDir });
await fixture.service.restore(fixture.workspace.workspaceId);
expect((await readPaseoWorktreeMetadata(fixture.workspace.cwd))?.baseRef).toBe("refs/heads/main");
expect(readPaseoWorktreeMetadata(fixture.workspace.cwd)?.baseRef).toBe("refs/heads/main");
expect(
await getCheckoutStatus(fixture.workspace.cwd, { paseoHome: fixture.paseoHome }),
).toMatchObject({
@@ -60,6 +60,8 @@ interface FakeAgentSessionOptions {
closeSession?: () => Promise<void>;
onStartTurn?: (prompt: AgentPromptInput, config: AgentSessionConfig) => void;
beforeTurnComplete?: (prompt: AgentPromptInput) => Promise<void>;
steerActiveTurn?: AgentSession["steerActiveTurn"];
onInterrupt?: () => void;
historyEvents?: readonly AgentStreamEvent[];
}
@@ -68,6 +70,8 @@ export interface TestAgentClientOptions {
closeSession?: () => Promise<void>;
onStartTurn?: (prompt: AgentPromptInput, config: AgentSessionConfig) => void;
beforeTurnComplete?: (prompt: AgentPromptInput) => Promise<void>;
steerActiveTurn?: AgentSession["steerActiveTurn"];
onInterrupt?: () => void;
supportsMcpServers?: boolean;
historyEvents?: readonly AgentStreamEvent[];
}
@@ -340,6 +344,8 @@ class FakeAgentSession implements AgentSession {
private nextTurnOrdinal = 0;
private activeForegroundTurnId: string | null = null;
readonly steerActiveTurn: AgentSession["steerActiveTurn"];
private readonly onInterrupt: (() => void) | undefined;
private readonly closeSession: (() => Promise<void>) | undefined;
private readonly onStartTurn: TestAgentClientOptions["onStartTurn"];
private readonly beforeTurnComplete: ((prompt: AgentPromptInput) => Promise<void>) | undefined;
@@ -354,6 +360,8 @@ class FakeAgentSession implements AgentSession {
this.config = options.config;
this.id = options.sessionId ?? randomUUID();
this.memoryMarker = options.memoryMarker ?? null;
this.steerActiveTurn = options.steerActiveTurn;
this.onInterrupt = options.onInterrupt;
this.closeSession = options.closeSession;
this.onStartTurn = options.onStartTurn;
this.beforeTurnComplete = options.beforeTurnComplete;
@@ -896,6 +904,7 @@ class FakeAgentSession implements AgentSession {
}
async interrupt(): Promise<void> {
this.onInterrupt?.();
this.interruptSignal.resolve();
}
@@ -1234,6 +1243,8 @@ class FakeAgentClient implements AgentClient {
closeSession: this.options.closeSession,
onStartTurn: this.options.onStartTurn,
beforeTurnComplete: this.options.beforeTurnComplete,
steerActiveTurn: this.options.steerActiveTurn,
onInterrupt: this.options.onInterrupt,
historyEvents: this.options.historyEvents,
});
}
@@ -1261,6 +1272,8 @@ class FakeAgentClient implements AgentClient {
closeSession: this.options.closeSession,
onStartTurn: this.options.onStartTurn,
beforeTurnComplete: this.options.beforeTurnComplete,
steerActiveTurn: this.options.steerActiveTurn,
onInterrupt: this.options.onInterrupt,
historyEvents: this.options.historyEvents,
});
}
@@ -493,7 +493,6 @@ async function createRestorableWorkspace(baseRef: string) {
commitFile(cwd, "feature.txt", "feature change\n");
const head = runGit(cwd, "rev-parse", "HEAD");
await client.checkoutRefresh(cwd);
expect(await client.getCheckoutStatus(cwd)).toMatchObject({ isPaseoOwnedWorktree: true });
const before = await client.getCheckoutDiff(cwd, { mode: "base", baseRef: "main" });
expect(before.error).toBeNull();
expect(before.files.map((file) => file.path)).toEqual(["feature.txt"]);
@@ -526,183 +526,3 @@ test("a late Git-ignored tree is pruned while tracked changes still notify consu
rmSync(tempDir, { recursive: true, force: true });
}
}, 15_000);
test("Git refresh, command tracing and 10,000 ignored alias events do no synchronous IO", async () => {
const { execFile } = await import("node:child_process");
const { promisify } = await import("node:util");
const { fileURLToPath } = await import("node:url");
const repositoryRoot = fileURLToPath(new URL("../../../../", import.meta.url));
const source = `
import assert from 'node:assert/strict';
import { write } from 'node:fs';
import { mkdtemp, mkdir, writeFile, symlink, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { setImmediate as tick } from 'node:timers/promises';
import { createHook } from 'node:async_hooks';
import pino from 'pino';
import { WorkspaceGitServiceImpl } from './packages/server/src/server/workspace-git-service.ts';
import { runGitCommand } from './packages/server/src/utils/run-git-command.ts';
import { writePaseoWorktreeMetadata } from './packages/server/src/utils/worktree-metadata.ts';
const root = await mkdtemp(join(tmpdir(), 'git-async-contract-'));
const home = join(root, 'home');
const repo = join(root, 'repo');
const worktree = join(home, 'worktrees', 'project', 'topic');
const alias = join(root, 'alias');
const logger = pino({ level: 'trace' }, { write() {} });
// Async descriptor writes avoid Windows stderr's synchronous stream API.
const marker = (label) => new Promise((resolve, reject) => {
write(2, label + '\\n', (error) => error ? reject(error) : resolve());
});
let callback;
let notifications = 0;
const service = new WorkspaceGitServiceImpl({ logger, paseoHome: home, deps: {
subscribe: async (_root, onChange) => {
callback = onChange;
return { updateIgnore: async () => {}, unsubscribe: async () => {} };
}
} });
try {
await mkdir(repo);
await runGitCommand(['init', '-b', 'main'], { cwd: repo });
await runGitCommand(['-c', 'user.name=Test', '-c', 'user.email=test@example.com', 'commit', '--allow-empty', '-m', 'base'], { cwd: repo });
await mkdir(join(home, 'worktrees', 'project'), { recursive: true });
await runGitCommand(['worktree', 'add', '-b', 'topic', worktree], { cwd: repo });
await writePaseoWorktreeMetadata(worktree, { baseRefName: 'main' });
await writeFile(join(worktree, '.gitignore'), 'ignored/\\n');
await mkdir(join(worktree, 'ignored'));
await writeFile(join(worktree, 'ignored', 'noise'), 'noise');
await symlink(worktree, alias, process.platform === 'win32' ? 'junction' : 'dir');
const subscription = await service.requestWorkingTreeWatch(alias, () => { notifications++; });
const events = Array.from({ length: process.env.GIT_IO_TRACE ? 2 : 10000 }, (_, i) => ({ type: 'update', path: join(i % 2 ? alias : worktree, 'ignored', 'noise') }));
// A shared CI runner can deschedule the child during a wall-clock sample.
// Keep the 20ms gate, but take three independent batches and require one
// responsive callback/timer pair. All batches must remain filesystem-free.
const samples = [];
for (let trial = 0; trial < (process.env.GIT_IO_TRACE ? 1 : 3); trial++) {
await tick();
let fsCalls = 0;
const hook = createHook({ init(_id, type) { if (type.startsWith('FSREQ')) fsCalls++; } });
await marker('FILTER_START');
hook.enable();
const started = performance.now();
const pulse = new Promise((resolve) => setTimeout(() => resolve(performance.now() - started), 0));
callback(null, events);
const filterMs = performance.now() - started;
hook.disable();
const loopGapMs = await pulse;
await marker('FILTER_END');
assert.equal(fsCalls, 0, "filesystem calls");
assert.equal(notifications, 0, "ignored event notification");
samples.push({ filterMs, loopGapMs, fsCalls });
}
console.log(JSON.stringify({ ...samples[0], samples }));
assert.ok(process.env.GIT_IO_TRACE || samples.some(({ filterMs, loopGapMs }) => filterMs < 20 && loopGapMs < 20), 'no responsive filter sample: ' + JSON.stringify(samples));
// Alias identity remains useful after deletion: no per-event realpath can recover this file.
callback(null, [{ type: 'delete', path: join(worktree, 'tracked.txt') }]);
assert.equal(notifications, 1);
await marker('REFRESH_START');
const snapshot = await service.getSnapshot(worktree);
assert.equal(snapshot.git.isGit, true);
assert.equal(snapshot.git.isPaseoOwnedWorktree, true);
await runGitCommand(['status', '--porcelain'], { cwd: alias, logger });
await marker('REFRESH_END');
subscription.unsubscribe();
} finally {
await service.dispose();
await rm(root, { recursive: true, force: true });
}
`;
for (const tracing of [false, true]) {
const { stdout, stderr } = await promisify(execFile)(
process.execPath,
[
"--import",
"tsx",
...(tracing ? ["--trace-sync-io"] : []),
"--input-type=module",
"--eval",
source,
],
{
cwd: repositoryRoot,
env: { ...process.env, GIT_IO_TRACE: tracing ? "1" : "" },
timeout: 30_000,
maxBuffer: 4 * 1024 * 1024,
},
).catch((error) => {
throw new Error(String(error.stderr).replaceAll(/at .*/g, "[child stack omitted]"));
});
for (const phase of ["FILTER", "REFRESH"]) {
const output = stderr.split(`${phase}_START\n`)[1]?.split(`${phase}_END\n`)[0];
expect(output, `${phase}: synchronous filesystem IO`).toBe("");
}
expect(JSON.parse(stdout.trim().split("\n")[0])).toMatchObject({ fsCalls: 0 });
}
}, 60_000);
test("Git event identity preserves Windows namespace, case and subtree boundaries", async () => {
const { GitObservationPaths } = await import("./git-observation-paths.js");
const paths = await GitObservationPaths.create("C:\\paseo-observation-contract");
const accept = paths.createEventFilter(["C:\\paseo-observation-contract\\Build"]);
expect(paths.relative("\\\\?\\C:\\PASEO-OBSERVATION-CONTRACT\\refs\\heads\\MixedCase")).toBe(
"refs/heads/MixedCase",
);
expect(accept("\\\\?\\C:\\paseo-observation-contract\\build\\noise")).toBe(false);
expect(accept("C:\\paseo-observation-contract\\BuildSibling\\file")).toBe(true);
});
test("concurrent alias requests establish one working-tree subscription", async () => {
const { symlink } = await import("node:fs/promises");
const root = realpathSync(mkdtempSync(path.join(tmpdir(), "git-observation-concurrent-")));
const repo = path.join(root, "repo");
const alias = path.join(root, "alias");
execFileSync("git", ["init", "-b", "main", repo], { stdio: "pipe" });
await symlink(repo, alias, process.platform === "win32" ? "junction" : "dir");
let subscriptions = 0;
const service = new WorkspaceGitServiceImpl({
logger: pino({ enabled: false }),
paseoHome: path.join(root, "home"),
deps: {
subscribe: async () => {
subscriptions++;
return { updateIgnore: async () => {}, unsubscribe: async () => {} };
},
},
});
try {
const watches = await Promise.all([
service.requestWorkingTreeWatch(repo, () => {}),
service.requestWorkingTreeWatch(alias, () => {}),
]);
expect(subscriptions).toBe(1);
for (const watch of watches) watch.unsubscribe();
} finally {
await service.dispose();
rmSync(root, { recursive: true, force: true });
}
});
test("Git event identity retains a symlink's literal target before canonicalizing its parent", async () => {
const { mkdir, symlink } = await import("node:fs/promises");
const { GitObservationPaths } = await import("./git-observation-paths.js");
const root = realpathSync(mkdtempSync(path.join(tmpdir(), "git-observation-literal-target-")));
const repository = path.join(root, "repository");
const parentAlias = path.join(root, "parent-alias");
const workspaceAlias = path.join(root, "workspace-alias");
const literalTarget = path.join(parentAlias, "repository");
const linkType = process.platform === "win32" ? "junction" : "dir";
try {
await mkdir(repository);
await symlink(root, parentAlias, linkType);
await symlink(literalTarget, workspaceAlias, linkType);
const paths = await GitObservationPaths.create(workspaceAlias);
const accept = paths.createEventFilter([path.join(workspaceAlias, "ignored")]);
for (const spelling of [workspaceAlias, literalTarget, repository]) {
expect(paths.relative(path.join(spelling, "deleted.txt"))).toBe("deleted.txt");
expect(accept(path.join(spelling, "ignored", "deleted.txt"))).toBe(false);
}
} finally {
rmSync(root, { recursive: true, force: true });
}
});
@@ -1189,7 +1189,7 @@ describe("WorkspaceGitService checkout observation", () => {
]);
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => {
expect(getCalledCwds(getCheckoutStatus).sort()).toEqual([WORKTREE_A, WORKTREE_B].sort());
expect(getCalledCwds(getCheckoutStatus)).toEqual([WORKTREE_A, WORKTREE_B]);
});
first.unsubscribe();
@@ -1624,12 +1624,6 @@ describe("WorkspaceGitService checkout observation", () => {
const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
await vi.waitFor(() => {
expect(watcher.subscribe).toHaveBeenCalledWith(
GIT_DIR,
expect.any(Function),
expect.any(Object),
);
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
expect(service.getMetrics().workspaceObservationSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
@@ -1648,17 +1642,14 @@ describe("WorkspaceGitService checkout observation", () => {
[60_000, 6],
] as const) {
await vi.advanceTimersByTimeAsync(elapsedMs);
await vi.waitFor(() => {
expect(pollCount()).toBe(expectedPolls);
expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
});
expect(pollCount()).toBe(expectedPolls);
}
isDirty = true;
await vi.advanceTimersByTimeAsync(60_000);
await vi.waitFor(() => expect(pollCount()).toBe(7));
expect(pollCount()).toBe(7);
await vi.advanceTimersByTimeAsync(5_000);
await vi.waitFor(() => expect(pollCount()).toBe(8));
expect(pollCount()).toBe(8);
subscription.unsubscribe();
service.dispose();
@@ -1869,10 +1860,8 @@ describe("WorkspaceGitService checkout observation", () => {
expect(watcher.subscribe).toHaveBeenCalledTimes(1);
});
await vi.advanceTimersByTimeAsync(35_000);
await vi.waitFor(() => {
expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceObservationSetupAdmissionActiveCount).toBe(0);
});
expect(service.getMetrics().workingTreeWatchSetupInFlightCount).toBe(0);
expect(service.getMetrics().workspaceObservationSetupAdmissionActiveCount).toBe(0);
expect(getWatcherSubscribeCallCount(watcher, REPO_CWD)).toBe(1);
openedSubscription.resolve({
@@ -991,7 +991,7 @@ describe("WorkspaceGitServiceImpl primitive refresh entrypoint", () => {
await vi.advanceTimersByTimeAsync(60_000);
await flushPromises();
await vi.waitFor(() => expect(subscribe).toHaveBeenCalledTimes(2));
expect(subscribe).toHaveBeenCalledTimes(2);
expect(getCheckoutStatus).toHaveBeenCalledTimes(1);
subscription.unsubscribe();
@@ -2221,7 +2221,7 @@ describe("WorkspaceGitServiceImpl D2 read methods", () => {
expect(getCheckoutStatus).toHaveBeenCalledTimes(2);
expect(getPullRequestStatus).toHaveBeenCalledTimes(2);
await service.dispose();
service.dispose();
});
test("onWorkspaceStateMayHaveChanged preserves includeForge when a file watcher fires within the debounce window", async () => {
@@ -42,13 +42,18 @@ import {
type ForgeResolver,
} from "../services/forge-resolver.js";
import { parseGitRevParsePath } from "../utils/git-rev-parse-path.js";
import { isPathInsideRoot, looksLikeDefiniteWindowsPath } from "../utils/path.js";
import {
createRealpathAwarePathMatcher,
getRealpathAwareRelativePath,
isPathInsideRoot,
isRealpathInsideRoot,
looksLikeDefiniteWindowsPath,
} from "../utils/path.js";
import {
createRunGitCommand,
runGitCommand,
type RunGitCommand,
} from "../utils/run-git-command.js";
import { GitObservationPaths } from "./git-observation-paths.js";
import { branchNameFromRef } from "../utils/worktree-metadata.js";
import { listPaseoWorktrees, type PaseoWorktreeInfo } from "../utils/worktree.js";
import { READ_ONLY_GIT_ENV } from "./checkout-git-utils.js";
@@ -412,7 +417,6 @@ class WorkspaceGitWatcherSubscriptionTimeoutError extends Error {
}
interface WorkspaceGitTarget {
gitDirPaths: GitObservationPaths | null;
cwd: string;
listeners: Set<WorkspaceGitListener>;
workingTreeWatchTarget: WorkingTreeWatchTarget | null;
@@ -439,7 +443,6 @@ interface WorkspaceGitTarget {
}
interface RepoGitTarget {
paths: GitObservationPaths;
repoGitRoot: string;
cwd: string;
workspaceKeys: Set<string>;
@@ -467,8 +470,6 @@ interface RepoMetadataWorkspaceRefresh {
}
interface WorkingTreeWatchTarget {
paths: GitObservationPaths;
acceptsEvent: (candidate: string) => boolean;
cwd: string;
watchPath: string;
repoRoot: string | null;
@@ -1164,8 +1165,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private async resolveWorkingTreeWatchTarget(cwd: string): Promise<WorkingTreeWatchTarget> {
const repoRoot = await this.resolveCheckoutWatchRoot(cwd);
const cwdPaths = repoRoot ? await GitObservationPaths.create(cwd) : null;
this.assertNotDisposed();
const targetCwd = repoRoot ?? cwd;
const existingTarget = this.workingTreeWatchTargets.get(targetCwd);
if (existingTarget) {
@@ -1180,7 +1179,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return target;
}
const watchPath = repoRoot && cwdPaths?.matches(repoRoot) ? cwd : targetCwd;
const watchPath = repoRoot && createRealpathAwarePathMatcher(repoRoot)(cwd) ? cwd : targetCwd;
const setup = this.createWorkingTreeWatchTarget(targetCwd, watchPath, repoRoot).finally(() => {
if (this.workingTreeWatchSetups.get(targetCwd) === setup) {
this.workingTreeWatchSetups.delete(targetCwd);
@@ -1199,7 +1198,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private createWorkspaceTarget(cwd: string): WorkspaceGitTarget {
const target: WorkspaceGitTarget = {
gitDirPaths: null,
cwd,
listeners: new Set(),
workingTreeWatchTarget: null,
@@ -1301,11 +1299,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
await this.promoteWorkingTreeWatchTarget(workingTreeTarget, facts.worktreeRoot);
const gitDir = facts.absoluteGitDir;
const gitDirPaths = target.gitDirPaths?.matches(gitDir)
? target.gitDirPaths
: await GitObservationPaths.create(gitDir);
if (!this.isActiveObservedWorkspaceTarget(target)) return;
target.gitDirPaths = gitDirPaths;
const repoGitRoot = facts.gitCommonDir ?? (await this.resolveWorkspaceGitRefsRoot(gitDir));
if (!this.isActiveObservedWorkspaceTarget(target)) {
return;
@@ -1338,7 +1331,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const promise = this.deps
.getCheckoutSnapshotFacts(target.cwd, context)
.then((facts) => {
if (!target.closed) target.latestFacts = facts;
target.latestFacts = facts;
return facts;
})
.finally(() => {
@@ -1366,11 +1359,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const ignoredDirectories = repoRoot
? (await this.loadIgnoredDirs(watchPath)).ignored
: new Set<string>();
const paths = await GitObservationPaths.create(watchPath);
this.assertNotDisposed();
const target: WorkingTreeWatchTarget = {
paths,
acceptsEvent: paths.createEventFilter([join(watchPath, ".git"), ...ignoredDirectories]),
cwd,
watchPath,
repoRoot,
@@ -1534,14 +1523,15 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
if (watcherErrored) {
return;
}
events = this.relevantWorkingTreeEvents(target, events);
if (events.length === 0) return;
const discovered = this.noteWorkingTreeDirectories(target, events);
if (events.some((event) => basename(event.path) === ".gitignore")) {
void this.refreshWorkingTreeIgnoredDirectories(target);
} else if (discovered) {
this.scheduleWorkingTreeIgnoreRefresh(target);
}
if (!this.hasRelevantWorkingTreeEvent(target, events)) {
return;
}
this.notifyWorkingTreeChanged(target, "working-tree-watch");
},
{ ignore },
@@ -1773,15 +1763,20 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
this.removeDeletedKnownDirectories(target, events);
const gitDir = join(target.watchPath, ".git");
// Event paths have already been mapped into this subscription's established namespace.
const matchesWatchPath = target.paths.matches;
// `matchesWatchPath` is realpath-aware because it compares an event-reported
// directory against `target.watchPath` itself — a config-time root that can
// legitimately differ from the watcher's reported form (e.g. a symlinked
// temp dir). It is now gated behind the plain Set lookup below, so it only
// runs for a directory this target has never seen before, not on every
// event.
const matchesWatchPath = createRealpathAwarePathMatcher(target.watchPath);
// One new directory is enough: the scheduled Git query inventories every
// ignored directory that exists by then. Stop before a large event batch
// multiplies containment checks across all its new directories.
for (const event of events) {
const directory = dirname(event.path);
// Cheap Set lookup first — the steady-state case where the directory is
// already known do not need another root comparison.
// already known must never reach the realpath-aware matcher below.
if (target.knownDirectories.has(directory) || matchesWatchPath(directory)) {
continue;
}
@@ -1960,22 +1955,28 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
target.pendingKnownDirectories.clear();
}
private relevantWorkingTreeEvents(
private hasRelevantWorkingTreeEvent(
target: WorkingTreeWatchTarget,
events: FileChange[],
): FileChange[] {
const relevant: FileChange[] = [];
// Watch backends may report the same path more than once in a batch.
const acceptedPaths = new Map<string, boolean>();
for (const event of events) {
let accepted = acceptedPaths.get(event.path);
if (accepted === undefined) {
accepted = target.acceptsEvent(event.path);
acceptedPaths.set(event.path, accepted);
): boolean {
const gitDir = join(target.watchPath, ".git");
const matchesWatchPath = createRealpathAwarePathMatcher(target.watchPath);
return events.some((event) => {
// Directory metadata changes at the watch root do not change Git state.
// FSEvents may emit these when every changed descendant was ignored.
if (matchesWatchPath(event.path)) {
return false;
}
if (accepted) relevant.push({ ...event, path: target.paths.toWatchPath(event.path) });
}
return relevant;
if (isRealpathInsideRoot(gitDir, event.path)) {
return false;
}
for (const ignoredDirectory of target.ignoredDirectories) {
if (isRealpathInsideRoot(ignoredDirectory, event.path)) {
return false;
}
}
return true;
});
}
private refreshWorkingTreeIgnoredDirectories(target: WorkingTreeWatchTarget): Promise<void> {
@@ -2030,7 +2031,6 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
if (!this.haveSamePaths(target.ignoredDirectories, ignoredDirectories)) {
target.ignoredDirectories = ignoredDirectories;
target.acceptsEvent = target.paths.createEventFilter(this.workingTreeIgnoreList(target));
this.pruneKnownDirectories(target);
}
if (target.fallbackPolling) {
@@ -2152,15 +2152,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
return;
}
const paths = await GitObservationPaths.create(repoGitRoot);
if (!this.isActiveObservedWorkspaceTarget(workspaceTarget)) return;
const establishedTarget = this.repoTargets.get(repoGitRoot);
if (establishedTarget) {
establishedTarget.workspaceKeys.add(workspaceTarget.cwd);
return;
}
const repoTarget: RepoGitTarget = {
paths,
repoGitRoot,
cwd: workspaceTarget.cwd,
workspaceKeys: new Set([workspaceTarget.cwd]),
@@ -2213,11 +2205,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
target: RepoGitTarget,
options?: { replaceFallback?: boolean },
): Promise<boolean> {
const paths = target.paths;
const ignore = getPrunedGitMetadataPaths("common").map((path) =>
join(target.repoGitRoot, path),
);
const matchesRepoGitRoot = paths.matches;
const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot);
const canary = this.deps.createWatcherLivenessCanary(target.repoGitRoot);
let openedSubscription: FileObserverSubscription | null = null;
let watcherErrored = false;
@@ -2232,9 +2223,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const subscription = await this.subscribeWithDeadline(
target.repoGitRoot,
(error, events) => {
const liveEvents = canary.filterEvents(
events.map((event) => ({ ...event, path: paths.toWatchPath(event.path) })),
);
const liveEvents = canary.filterEvents(events);
if (error) {
if (watcherErrored) {
return;
@@ -2260,7 +2249,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
const relevantEvents = liveEvents.filter(
(event) =>
!matchesRepoGitRoot(event.path) &&
ignore.every((ignoredPath) => !isPathInsideRoot(ignoredPath, event.path)),
ignore.every((ignoredPath) => !isRealpathInsideRoot(ignoredPath, event.path)),
);
if (relevantEvents.length > 0) {
const immediateEvents = target.fetchInFlight
@@ -2373,7 +2362,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
events: FileChange[],
): Map<string, RepoMetadataWorkspaceRefresh> | null {
const refreshes = new Map<string, RepoMetadataWorkspaceRefresh>();
const matchesRepoGitRoot = target.paths.matches;
const matchesRepoGitRoot = createRealpathAwarePathMatcher(target.repoGitRoot);
for (const event of events) {
if (!this.routeRepoMetadataEvent(target, event, matchesRepoGitRoot, refreshes)) return null;
@@ -2388,7 +2377,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
): void {
const workspaceKeys = new Set<string>();
for (const event of events) {
const commonRelativePath = target.paths.relative(event.path)?.replaceAll("\\", "/");
const commonRelativePath = getRealpathAwareRelativePath(
target.repoGitRoot,
event.path,
)?.replaceAll("\\", "/");
if (commonRelativePath === "config" || commonRelativePath === "info/exclude") {
for (const workspaceKey of target.workspaceKeys) workspaceKeys.add(workspaceKey);
continue;
@@ -2398,10 +2390,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
if (
facts?.isGit &&
facts.absoluteGitDir &&
this.workspaceTargets
.get(workspaceKey)
?.gitDirPaths?.relative(event.path)
?.replaceAll("\\", "/") === "config.worktree"
getRealpathAwareRelativePath(facts.absoluteGitDir, event.path)?.replaceAll("\\", "/") ===
"config.worktree"
) {
workspaceKeys.add(workspaceKey);
}
@@ -2417,7 +2407,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
}
private isFetchRemoteMetadataEvent(target: RepoGitTarget, event: FileChange): boolean {
const relativePath = target.paths.relative(event.path);
const relativePath = getRealpathAwareRelativePath(target.repoGitRoot, event.path);
const effect = classifyGitMetadataPath("common", relativePath ?? "");
return (
(effect.kind === "ref" && effect.namespace === "remote") ||
@@ -2429,12 +2419,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private routeRepoMetadataEvent(
target: RepoGitTarget,
event: FileChange,
matchesRepoGitRoot: (candidate: string) => boolean,
matchesRepoGitRoot: ReturnType<typeof createRealpathAwarePathMatcher>,
refreshes: Map<string, RepoMetadataWorkspaceRefresh>,
): boolean {
if (this.routePrivateGitDirEvent(target, event, matchesRepoGitRoot, refreshes)) return true;
const commonRelativePath = target.paths.relative(event.path);
const commonRelativePath = getRealpathAwareRelativePath(target.repoGitRoot, event.path);
const effect = classifyGitMetadataPath("common", commonRelativePath ?? "");
switch (effect.kind) {
case "ignore":
@@ -2486,7 +2476,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private routePrivateGitDirEvent(
target: RepoGitTarget,
event: FileChange,
matchesRepoGitRoot: (candidate: string) => boolean,
matchesRepoGitRoot: ReturnType<typeof createRealpathAwarePathMatcher>,
refreshes: Map<string, RepoMetadataWorkspaceRefresh>,
): boolean {
let matched = false;
@@ -2495,10 +2485,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
if (!facts?.isGit || !facts.absoluteGitDir || matchesRepoGitRoot(facts.absoluteGitDir)) {
continue;
}
const relativePath = this.workspaceTargets
.get(workspaceKey)
?.gitDirPaths?.relative(event.path);
if (relativePath == null) continue;
const relativePath = getRealpathAwareRelativePath(facts.absoluteGitDir, event.path);
if (relativePath === null) continue;
matched = true;
const effect = classifyGitMetadataPath("worktree", relativePath);
if (effect.kind === "owner") {
@@ -2512,7 +2500,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
private routeMainCheckoutMetadata(
target: RepoGitTarget,
matchesRepoGitRoot: (candidate: string) => boolean,
matchesRepoGitRoot: ReturnType<typeof createRealpathAwarePathMatcher>,
refreshBase: boolean,
refreshes: Map<string, RepoMetadataWorkspaceRefresh>,
): void {
@@ -74,7 +74,7 @@ export async function bootstrapWorkspaceRegistries(options: {
continue;
}
try {
await pinPaseoWorktreeBranchIdentityIfMissing(workspace.worktreeRoot, workspace.branch);
pinPaseoWorktreeBranchIdentityIfMissing(workspace.worktreeRoot, workspace.branch);
} catch (error) {
options.logger.warn(
{ err: error, workspaceId: workspace.workspaceId },
@@ -343,8 +343,8 @@ describe("generateBranchNameFromFirstAgentContext", () => {
const repoRoot = createTempDir("paseo-branch-config-");
const worktreeRoot = createTempDir("paseo-branch-worktree-");
mkdirSync(path.join(worktreeRoot, ".git"));
await writePaseoWorktreeMetadata(worktreeRoot, { baseRefName: "main" });
await writePaseoWorktreeFirstAgentBranchAutoNameMetadata(worktreeRoot, {
writePaseoWorktreeMetadata(worktreeRoot, { baseRefName: "main" });
writePaseoWorktreeFirstAgentBranchAutoNameMetadata(worktreeRoot, {
placeholderBranchName: "dazzling-yak",
});
writeConfig(repoRoot, {
@@ -854,7 +854,7 @@ describe.skipIf(isPlatform("win32"))("worktree-core POSIX-only", () => {
second.worktree.worktreePath,
"remote.paseo-pr-14.url",
);
const metadata = await readPaseoWorktreeMetadata(second.worktree.worktreePath);
const metadata = readPaseoWorktreeMetadata(second.worktree.worktreePath);
const facts = await getCheckoutSnapshotFacts(second.worktree.worktreePath, { paseoHome });
expect(second.worktree.branchName).toBe("feature/gitlab-mr-1");
@@ -1113,7 +1113,7 @@ describe.skipIf(isPlatform("win32"))("worktree-core POSIX-only", () => {
createCoreDeps({ github }),
);
writeFileSync(
await getPaseoWorktreeMetadataPath(second.worktree.worktreePath),
getPaseoWorktreeMetadataPath(second.worktree.worktreePath),
`${JSON.stringify({ version: 1, baseRefName: "main" }, null, 2)}\n`,
"utf8",
);
@@ -172,7 +172,7 @@ describe("listCheckoutCommits", () => {
mkdirSync(join(worktreesRoot, "repo-hash"), { recursive: true });
git(["worktree", "add", "-b", "feature", worktreeDir], repoDir);
commitFile(worktreeDir, "feature.txt", "feature\n", "Feature work");
await writePaseoWorktreeMetadata(worktreeDir, { baseRefName: "deleted-base" });
writePaseoWorktreeMetadata(worktreeDir, { baseRefName: "deleted-base" });
const { baseRef, commits } = await listCheckoutCommits({
cwd: worktreeDir,
+12 -14
View File
@@ -2859,7 +2859,7 @@ const x = 1;
execFileSync("git", ["worktree", "add", workspaceDir, "contributor/old-change"], {
cwd: repoDir,
});
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "old-change",
@@ -2899,7 +2899,7 @@ const x = 1;
const workspaceDir = join(paseoHome, "worktrees", "repo", "renamed-by-agent");
mkdirSync(join(paseoHome, "worktrees", "repo"), { recursive: true });
execFileSync("git", ["worktree", "add", workspaceDir, "placeholder"], { cwd: repoDir });
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "placeholder",
@@ -2936,7 +2936,7 @@ const x = 1;
const workspaceDir = join(paseoHome, "worktrees", "repo", "pinned-worktree");
mkdirSync(join(paseoHome, "worktrees", "repo"), { recursive: true });
execFileSync("git", ["worktree", "add", workspaceDir, "feature/pinned"], { cwd: repoDir });
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "feature/pinned",
@@ -2975,7 +2975,7 @@ const x = 1;
execFileSync("git", ["worktree", "add", workspaceDir, "contributor/old-change-1"], {
cwd: repoDir,
});
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "old-change",
@@ -2997,7 +2997,7 @@ const x = 1;
execFileSync("git", ["worktree", "add", workspaceDir, "contributor/old-change"], {
cwd: repoDir,
});
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "old-change",
@@ -3062,7 +3062,7 @@ const x = 1;
execFileSync("git", ["worktree", "add", workspaceDir, "mixedowner/old-change"], {
cwd: repoDir,
});
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "old-change",
@@ -3114,7 +3114,7 @@ const x = 1;
execFileSync("git", ["worktree", "add", workspaceDir, "feature/gitlab-mr"], {
cwd: repoDir,
});
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "feature/gitlab-mr",
@@ -3128,9 +3128,7 @@ const x = 1;
const forge = createGitHubServiceRecordingPullRequestTargets({ requestedTargets });
await renameCurrentBranch(workspaceCwd, "feature/renamed");
expect(
(await readPaseoWorktreeMetadata(workspaceDir))?.changeRequestLookupTarget,
).toMatchObject({
expect(readPaseoWorktreeMetadata(workspaceDir)?.changeRequestLookupTarget).toMatchObject({
localBranchName: "feature/renamed",
});
const renamedFacts = await getCheckoutSnapshotFacts(workspaceDir, { paseoHome });
@@ -3166,7 +3164,7 @@ const x = 1;
execFileSync("git", ["worktree", "add", workspaceDir, "feature/placeholder"], {
cwd: repoDir,
});
await writePaseoWorktreeMetadata(workspaceDir, {
writePaseoWorktreeMetadata(workspaceDir, {
baseRefName: "main",
changeRequestLookupTarget: {
headRef: "feature/placeholder",
@@ -3176,7 +3174,7 @@ const x = 1;
await renameCurrentBranch(workspaceDir, "feature/generated");
expect((await readPaseoWorktreeMetadata(workspaceDir))?.changeRequestLookupTarget).toEqual({
expect(readPaseoWorktreeMetadata(workspaceDir)?.changeRequestLookupTarget).toEqual({
headRef: "feature/generated",
localBranchName: "feature/generated",
});
@@ -3953,7 +3951,7 @@ const x = 1;
cwd: worktree.worktreePath,
});
const metadataPath = await getPaseoWorktreeMetadataPath(worktree.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(worktree.worktreePath);
rmSync(metadataPath, { force: true });
const baseDiff = await getCheckoutDiff(worktree.worktreePath, { mode: "base" }, { paseoHome });
@@ -3972,7 +3970,7 @@ const x = 1;
paseoHome,
});
const metadataPath = await getPaseoWorktreeMetadataPath(worktree.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(worktree.worktreePath);
rmSync(metadataPath, { force: true });
const status = await getCheckoutStatus(worktree.worktreePath, { paseoHome });
+9 -31
View File
@@ -1,15 +1,7 @@
import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit";
import { resolve, dirname, basename } from "path";
import {
open as openFile,
readFile,
stat as statFile,
mkdtemp,
copyFile,
realpath,
access,
rm,
} from "fs/promises";
import { existsSync, realpathSync } from "fs";
import { open as openFile, readFile, stat as statFile, mkdtemp, copyFile, rm } from "fs/promises";
import { tmpdir } from "node:os";
import { setImmediate } from "node:timers/promises";
import { TTLCache } from "@isaacs/ttlcache";
@@ -1112,7 +1104,7 @@ async function getMainRepoRootFromCommonDir(
if (!commonDir) {
throw new Error("Not in a git repository");
}
const normalized = await realpath(commonDir);
const normalized = realpathSync(commonDir);
if (basename(normalized) === ".git") {
return dirname(normalized);
@@ -1237,7 +1229,7 @@ export async function renameCurrentBranch(
const currentBranch = await getCurrentBranch(cwd);
if (currentBranch) {
await rebindPaseoWorktreeChangeRequestHint(worktreeRoot, previousBranch, currentBranch);
rebindPaseoWorktreeChangeRequestHint(worktreeRoot, previousBranch, currentBranch);
}
return { previousBranch, currentBranch };
}
@@ -1282,8 +1274,8 @@ function storedBaseRefFromMetadata(metadata: PaseoWorktreeMetadata | null): stri
return metadata?.baseRef ?? metadata?.baseRefName ?? null;
}
async function readPaseoWorktreeBaseRef(worktreeRoot: string): Promise<string | null> {
return storedBaseRefFromMetadata(await readPaseoWorktreeMetadata(worktreeRoot));
function readPaseoWorktreeBaseRef(worktreeRoot: string): string | null {
return storedBaseRefFromMetadata(readPaseoWorktreeMetadata(worktreeRoot));
}
async function getStoredBaseRefForCwd(
@@ -1557,12 +1549,7 @@ async function abortGitPullConflictState(cwd: string): Promise<void> {
const rebaseMergePath = resolve(gitDir, "rebase-merge");
const rebaseApplyPath = resolve(gitDir, "rebase-apply");
if (
await access(mergeHeadPath).then(
() => true,
() => false,
)
) {
if (existsSync(mergeHeadPath)) {
try {
await runGitCommand(["merge", "--abort"], { cwd, timeout: 120_000 });
} catch {
@@ -1570,16 +1557,7 @@ async function abortGitPullConflictState(cwd: string): Promise<void> {
}
}
if (
(await access(rebaseMergePath).then(
() => true,
() => false,
)) ||
(await access(rebaseApplyPath).then(
() => true,
() => false,
))
) {
if (existsSync(rebaseMergePath) || existsSync(rebaseApplyPath)) {
try {
await runGitCommand(["rebase", "--abort"], { cwd, timeout: 120_000 });
} catch {
@@ -2100,7 +2078,7 @@ export async function getCheckoutSnapshotFacts(
}
const paseoWorktreeMetadata = inspected.paseoWorktree.isPaseoOwnedWorktree
? await readPaseoWorktreeMetadata(inspected.paseoWorktree.worktreeRoot)
? readPaseoWorktreeMetadata(inspected.paseoWorktree.worktreeRoot)
: null;
const storedBaseRef = storedBaseRefFromMetadata(paseoWorktreeMetadata);
const resolvedBaseRef = storedBaseRef ?? (await resolveBaseRef(cwd, context));
+2 -2
View File
@@ -107,7 +107,7 @@ export function isRealpathInsideRoot(root: string, candidate: string): boolean {
return getRealpathAwareRelativePath(root, candidate) !== null;
}
export function getRelativePathInsideRoot(root: string, candidate: string): string | null {
function getRelativePathInsideRoot(root: string, candidate: string): string | null {
const compareAsWindows = shouldCompareAsWindows(root, candidate);
const platformPath = compareAsWindows ? nodePath.win32 : nodePath.posix;
const normalizedRoot = normalizePathForComparison(root, compareAsWindows);
@@ -175,7 +175,7 @@ function normalizePathForComparison(value: string, compareAsWindows: boolean): s
return compareAsWindows ? normalized.toLowerCase() : normalized;
}
export function normalizePathPreservingCase(value: string, compareAsWindows: boolean): string {
function normalizePathPreservingCase(value: string, compareAsWindows: boolean): string {
const platformPath = compareAsWindows ? nodePath.win32 : nodePath.posix;
const comparableValue = compareAsWindows ? stripWindowsNamespacePrefix(value) : value;
const platformNormalized = platformPath.normalize(comparableValue);
@@ -507,6 +507,7 @@ describe("runGitCommand", () => {
command: "git",
args: ["status", "--short"],
cwd: process.cwd(),
cwdExists: true,
timeout: 30_000,
maxOutputBytes: 20 * 1024 * 1024,
acceptExitCodes: [0, 1],
@@ -519,6 +520,7 @@ describe("runGitCommand", () => {
command: "git",
args: ["status", "--short"],
cwd: process.cwd(),
cwdExists: true,
durationMs: expect.any(Number),
exitCode: 0,
signal: null,
@@ -528,9 +530,6 @@ describe("runGitCommand", () => {
}),
"Git command closed",
);
for (const [record] of trace.mock.calls) {
expect(record).not.toHaveProperty("cwdExists");
}
});
it("rejects non-zero exit codes that are not accepted and frees the slot", async () => {
@@ -1,4 +1,5 @@
import { AsyncLocalStorage } from "node:async_hooks";
import { existsSync } from "node:fs";
import type { Logger } from "pino";
import type { ProcessEnvRecord } from "../server/paseo-env.js";
import {
@@ -310,6 +311,7 @@ function executeGitCommand<Output>(
command: "git",
args,
cwd: options.cwd,
cwdExists: existsSync(options.cwd),
timeout,
maxOutputBytes,
acceptExitCodes,
+136 -238
View File
@@ -1,12 +1,7 @@
import { mkdir, open, realpath, rename, stat, writeFile } from "node:fs/promises";
import { randomUUID } from "node:crypto";
import { constants } from "node:fs";
import pLimit from "p-limit";
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "fs";
import { isAbsolute, join, resolve } from "path";
import { z } from "zod";
const MAX_WORKTREE_METADATA_BYTES = 64 * 1024;
const ChangeRequestLookupTargetSchema = z.object({
headRef: z.string().min(1),
headRepositoryOwner: z.string().min(1).optional(),
@@ -95,78 +90,75 @@ function normalizeLegacyGitHubOwnerForBranch(owner: string): string | null {
return /^[a-z0-9-]+$/.test(normalized) ? normalized : null;
}
export async function rebindPaseoWorktreeChangeRequestHint(
export function rebindPaseoWorktreeChangeRequestHint(
worktreeRoot: string,
previousBranch: string,
currentBranch: string,
): Promise<boolean> {
return mutateMetadata(worktreeRoot, async (root) => {
const metadata = await readMetadataFile(root);
const target = getPaseoWorktreeChangeRequestHintForBranch(metadata, previousBranch);
if (!metadata || !target) {
return false;
}
): boolean {
const metadata = readPaseoWorktreeMetadata(worktreeRoot);
const target = getPaseoWorktreeChangeRequestHintForBranch(metadata, previousBranch);
if (!metadata || !target) {
return false;
}
await writePaseoWorktreeMetadataFile(root, {
...metadata,
changeRequestLookupTarget: {
...target,
...(target.headRef === previousBranch &&
!target.headRepositoryOwner &&
target.changeRequestNumber === undefined
? { headRef: currentBranch }
: {}),
localBranchName: currentBranch,
},
});
return true;
writePaseoWorktreeMetadataFile(worktreeRoot, {
...metadata,
changeRequestLookupTarget: {
...target,
...(target.headRef === previousBranch &&
!target.headRepositoryOwner &&
target.changeRequestNumber === undefined
? { headRef: currentBranch }
: {}),
localBranchName: currentBranch,
},
});
return true;
}
export async function pinPaseoWorktreeBranchIdentityIfMissing(
export function pinPaseoWorktreeBranchIdentityIfMissing(
worktreeRoot: string,
branch: string,
): Promise<boolean> {
return mutateMetadata(worktreeRoot, async (root) => {
const metadata = await readMetadataFile(root);
if (!metadata || metadata.changeRequestLookupTarget) {
return false;
}
const target = createPaseoWorktreeChangeRequestHint({
headRef: branch,
localBranchName: branch,
});
await writePaseoWorktreeMetadataFile(root, {
...metadata,
changeRequestLookupTarget: target,
});
return true;
): boolean {
const metadata = readPaseoWorktreeMetadata(worktreeRoot);
if (!metadata || metadata.changeRequestLookupTarget) {
return false;
}
const target = createPaseoWorktreeChangeRequestHint({
headRef: branch,
localBranchName: branch,
});
writePaseoWorktreeMetadataFile(worktreeRoot, {
...metadata,
changeRequestLookupTarget: target,
});
return true;
}
async function getGitDirForWorktreeRoot(worktreeRoot: string): Promise<string> {
function getGitDirForWorktreeRoot(worktreeRoot: string): string {
const gitPath = join(worktreeRoot, ".git");
let info;
try {
info = await stat(gitPath);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT")
throw new Error(`Not a git repository: ${worktreeRoot}`, { cause: error });
throw error;
if (!existsSync(gitPath)) {
throw new Error(`Not a git repository: ${worktreeRoot}`);
}
if (info.isDirectory()) return gitPath;
const gitFileContent = await readBoundedFile(gitPath);
const match = gitFileContent.match(/gitdir:\s*(.+)/);
if (match?.[1]) {
const raw = match[1].trim();
return isAbsolute(raw) ? raw : resolve(worktreeRoot, raw);
// In a worktree checkout, `.git` is a file containing `gitdir: <path>`.
// In a normal checkout, `.git` is a directory.
try {
const gitFileContent = readFileSync(gitPath, "utf8");
const match = gitFileContent.match(/gitdir:\s*(.+)/);
if (match?.[1]) {
const raw = match[1].trim();
return isAbsolute(raw) ? raw : resolve(worktreeRoot, raw);
}
} catch {
// If `.git` is a directory, readFileSync will throw; fall through.
}
return gitPath;
}
export async function getPaseoWorktreeMetadataPath(worktreeRoot: string): Promise<string> {
const gitDir = await getGitDirForWorktreeRoot(worktreeRoot);
export function getPaseoWorktreeMetadataPath(worktreeRoot: string): string {
const gitDir = getGitDirForWorktreeRoot(worktreeRoot);
return join(gitDir, "paseo", "worktree.json");
}
@@ -221,166 +213,120 @@ function assertValidBaseRef(value: string): void {
}
}
export async function writePaseoWorktreeMetadata(
export function writePaseoWorktreeMetadata(
worktreeRoot: string,
options: {
baseRefName: string;
baseRef?: string;
changeRequestLookupTarget?: PaseoWorktreeChangeRequestHint;
},
): Promise<void> {
return mutateMetadata(worktreeRoot, async (root) => {
const baseRefName = normalizeBaseRefName(options.baseRefName);
assertValidBaseRef(baseRefName);
const baseRef = options.baseRef?.trim();
if (baseRef) {
assertValidBaseRef(baseRef);
}
): void {
const baseRefName = normalizeBaseRefName(options.baseRefName);
assertValidBaseRef(baseRefName);
const baseRef = options.baseRef?.trim();
if (baseRef) {
assertValidBaseRef(baseRef);
}
const metadata: PaseoWorktreeMetadata = {
version: 1,
baseRefName,
...(baseRef ? { baseRef } : {}),
...(options.changeRequestLookupTarget
? { changeRequestLookupTarget: options.changeRequestLookupTarget }
: {}),
};
await writePaseoWorktreeMetadataFile(root, metadata);
});
const metadata: PaseoWorktreeMetadata = {
version: 1,
baseRefName,
...(baseRef ? { baseRef } : {}),
...(options.changeRequestLookupTarget
? { changeRequestLookupTarget: options.changeRequestLookupTarget }
: {}),
};
writePaseoWorktreeMetadataFile(worktreeRoot, metadata);
}
export async function writePaseoWorktreeRuntimeMetadata(
export function writePaseoWorktreeRuntimeMetadata(
worktreeRoot: string,
options: { worktreePort: number },
): Promise<void> {
return mutateMetadata(worktreeRoot, async (root) => {
if (!Number.isInteger(options.worktreePort) || options.worktreePort <= 0) {
throw new Error(`Invalid worktree runtime port: ${options.worktreePort}`);
}
): void {
if (!Number.isInteger(options.worktreePort) || options.worktreePort <= 0) {
throw new Error(`Invalid worktree runtime port: ${options.worktreePort}`);
}
const current = await readMetadataFile(root);
if (!current) {
throw new Error("Cannot persist worktree runtime metadata: missing base metadata");
}
const current = readPaseoWorktreeMetadata(worktreeRoot);
if (!current) {
throw new Error("Cannot persist worktree runtime metadata: missing base metadata");
}
const next: PaseoWorktreeMetadata = {
...current,
version: 2,
runtime: {
worktreePort: options.worktreePort,
},
};
await writePaseoWorktreeMetadataFile(root, next);
});
const next: PaseoWorktreeMetadata = {
...current,
version: 2,
runtime: {
worktreePort: options.worktreePort,
},
};
writePaseoWorktreeMetadataFile(worktreeRoot, next);
}
export async function writePaseoWorktreeFirstAgentBranchAutoNameMetadata(
export function writePaseoWorktreeFirstAgentBranchAutoNameMetadata(
worktreeRoot: string,
options: { placeholderBranchName: string },
): Promise<void> {
return mutateMetadata(worktreeRoot, async (root) => {
const placeholderBranchName = options.placeholderBranchName.trim();
if (!placeholderBranchName) {
throw new Error("Placeholder branch name is required");
}
): void {
const placeholderBranchName = options.placeholderBranchName.trim();
if (!placeholderBranchName) {
throw new Error("Placeholder branch name is required");
}
const current = await readMetadataFile(root);
if (!current) {
throw new Error(
"Cannot persist first-agent branch auto-name metadata: missing base metadata",
);
}
const current = readPaseoWorktreeMetadata(worktreeRoot);
if (!current) {
throw new Error("Cannot persist first-agent branch auto-name metadata: missing base metadata");
}
await writePaseoWorktreeMetadataFile(root, {
...current,
version: 2,
firstAgentBranchAutoName: {
status: "pending",
placeholderBranchName,
},
});
writePaseoWorktreeMetadataFile(worktreeRoot, {
...current,
version: 2,
firstAgentBranchAutoName: {
status: "pending",
placeholderBranchName,
},
});
}
export async function markPaseoWorktreeFirstAgentBranchAutoNameAttempted(
export function markPaseoWorktreeFirstAgentBranchAutoNameAttempted(
worktreeRoot: string,
options: { attemptedAt?: string } = {},
): Promise<PaseoWorktreeMetadata | null> {
return mutateMetadata(worktreeRoot, async (root) => {
const current = await readMetadataFile(root);
if (
!current ||
current.version !== 2 ||
current.firstAgentBranchAutoName?.status !== "pending"
) {
return current;
}
const next: PaseoWorktreeMetadata = {
...current,
firstAgentBranchAutoName: {
status: "attempted",
placeholderBranchName: current.firstAgentBranchAutoName.placeholderBranchName,
attemptedAt: options.attemptedAt ?? new Date().toISOString(),
},
};
await writePaseoWorktreeMetadataFile(root, next);
return next;
});
}
export async function readPaseoWorktreeMetadata(
worktreeRoot: string,
): Promise<PaseoWorktreeMetadata | null> {
const key = await resolveMetadataIdentity(worktreeRoot);
await metadataMutations.get(key)?.catch(() => {});
return metadataIo(() => readMetadataFile(key));
}
async function readMetadataFile(worktreeRoot: string): Promise<PaseoWorktreeMetadata | null> {
const metadataPath = await getPaseoWorktreeMetadataPath(worktreeRoot);
try {
return PaseoWorktreeMetadataSchema.parse(JSON.parse(await readBoundedFile(metadataPath)));
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
throw error;
): PaseoWorktreeMetadata | null {
const current = readPaseoWorktreeMetadata(worktreeRoot);
if (!current || current.version !== 2 || current.firstAgentBranchAutoName?.status !== "pending") {
return current;
}
const next: PaseoWorktreeMetadata = {
...current,
firstAgentBranchAutoName: {
status: "attempted",
placeholderBranchName: current.firstAgentBranchAutoName.placeholderBranchName,
attemptedAt: options.attemptedAt ?? new Date().toISOString(),
},
};
writePaseoWorktreeMetadataFile(worktreeRoot, next);
return next;
}
// O_NONBLOCK lets us reject FIFOs by kind without hanging in open(). A fixed read also
// caps a regular file that grows between stat and read; readFile's whole-file loop does not.
async function readBoundedFile(path: string): Promise<string> {
const file = await open(path, constants.O_RDONLY | constants.O_NONBLOCK);
try {
const info = await file.stat();
const limit = MAX_WORKTREE_METADATA_BYTES;
if (!info.isFile() || info.size > limit)
throw new Error(`Invalid Paseo worktree metadata file: ${path}`);
const buffer = Buffer.alloc(limit + 1);
let length = 0;
while (length < buffer.length) {
const { bytesRead } = await file.read(buffer, length, buffer.length - length);
if (bytesRead === 0) break;
length += bytesRead;
}
if (length > limit) throw new Error(`Invalid Paseo worktree metadata file: ${path}`);
return buffer.toString("utf8", 0, length);
} finally {
await file.close();
export function readPaseoWorktreeMetadata(worktreeRoot: string): PaseoWorktreeMetadata | null {
const metadataPath = getPaseoWorktreeMetadataPath(worktreeRoot);
if (!existsSync(metadataPath)) {
return null;
}
const parsed = JSON.parse(readFileSync(metadataPath, "utf8"));
return PaseoWorktreeMetadataSchema.parse(parsed);
}
export async function requirePaseoWorktreeBaseRefName(worktreeRoot: string): Promise<string> {
const metadataPath = await getPaseoWorktreeMetadataPath(worktreeRoot);
const metadata = await readPaseoWorktreeMetadata(worktreeRoot);
export function requirePaseoWorktreeBaseRefName(worktreeRoot: string): string {
const metadataPath = getPaseoWorktreeMetadataPath(worktreeRoot);
const metadata = readPaseoWorktreeMetadata(worktreeRoot);
if (!metadata) {
throw new Error(`Missing Paseo worktree base metadata: ${metadataPath}`);
}
return metadata.baseRefName;
}
export async function readPaseoWorktreeRuntimePort(worktreeRoot: string): Promise<number | null> {
const metadata = await readPaseoWorktreeMetadata(worktreeRoot);
export function readPaseoWorktreeRuntimePort(worktreeRoot: string): number | null {
const metadata = readPaseoWorktreeMetadata(worktreeRoot);
if (!metadata) {
return null;
}
@@ -390,61 +336,13 @@ export async function readPaseoWorktreeRuntimePort(worktreeRoot: string): Promis
return null;
}
async function writePaseoWorktreeMetadataFile(
function writePaseoWorktreeMetadataFile(
worktreeRoot: string,
metadata: PaseoWorktreeMetadata,
): Promise<void> {
const metadataPath = await getPaseoWorktreeMetadataPath(worktreeRoot);
const contents = `${JSON.stringify(metadata, null, 2)}\n`;
if (Buffer.byteLength(contents) > MAX_WORKTREE_METADATA_BYTES) {
throw new Error(`Invalid Paseo worktree metadata file: ${metadataPath}`);
}
// Never recreate an archived Git directory while a queued mutation is pending.
try {
await mkdir(join(await getGitDirForWorktreeRoot(worktreeRoot), "paseo"));
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
}
const tempPath = `${metadataPath}.${randomUUID()}.tmp`;
await writeFile(tempPath, contents, "utf8");
await rename(tempPath, metadataPath);
}
const metadataIo = pLimit(2);
const metadataMutations = new Map<string, Promise<unknown>>();
// Only share in-flight identity resolution: a replaced symlink must not retain an old owner.
const metadataIdentities = new Map<string, Promise<string>>();
function resolveMetadataIdentity(worktreeRoot: string): Promise<string> {
const path = resolve(worktreeRoot);
const existing = metadataIdentities.get(path);
if (existing) return existing;
const identity = metadataIo(async () => {
try {
return await realpath(path);
} catch {
return path;
}
}).finally(() => {
if (metadataIdentities.get(path) === identity) metadataIdentities.delete(path);
});
metadataIdentities.set(path, identity);
return identity;
}
async function mutateMetadata<T>(
worktreeRoot: string,
mutation: (root: string) => Promise<T>,
): Promise<T> {
const key = await resolveMetadataIdentity(worktreeRoot);
const previous = metadataMutations.get(key) ?? Promise.resolve();
const pending = previous.catch(() => {}).then(() => metadataIo(() => mutation(key)));
metadataMutations.set(key, pending);
void pending
.finally(() => {
if (metadataMutations.get(key) === pending) metadataMutations.delete(key);
})
.catch(() => {});
return pending;
): void {
const metadataPath = getPaseoWorktreeMetadataPath(worktreeRoot);
mkdirSync(join(getGitDirForWorktreeRoot(worktreeRoot), "paseo"), { recursive: true });
const tempPath = `${metadataPath}.${process.pid}.${Date.now()}.tmp`;
writeFileSync(tempPath, `${JSON.stringify(metadata, null, 2)}\n`, "utf8");
renameSync(tempPath, metadataPath);
}
+10 -121
View File
@@ -21,13 +21,7 @@ import {
type WorktreeConfig,
} from "./worktree";
import type { PaseoConfig } from "@getpaseo/protocol/paseo-config-schema";
import {
getPaseoWorktreeMetadataPath,
readPaseoWorktreeMetadata,
writePaseoWorktreeMetadata,
writePaseoWorktreeRuntimeMetadata,
writePaseoWorktreeFirstAgentBranchAutoNameMetadata,
} from "./worktree-metadata.js";
import { getPaseoWorktreeMetadataPath, readPaseoWorktreeMetadata } from "./worktree-metadata.js";
import {
getCheckoutDiff,
getCheckoutStatus,
@@ -131,7 +125,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
expect(result.worktreePath).toBe(join(paseoHome, "worktrees", projectHash, "hello-world"));
expect(existsSync(result.worktreePath)).toBe(true);
expect(existsSync(join(result.worktreePath, "file.txt"))).toBe(true);
const metadataPath = await getPaseoWorktreeMetadataPath(result.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(result.worktreePath);
expect(existsSync(metadataPath)).toBe(true);
const metadata = JSON.parse(readFileSync(metadataPath, "utf8"));
expect(metadata).toMatchObject({
@@ -262,7 +256,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
cwd: result.worktreePath,
});
const metadataPath = await getPaseoWorktreeMetadataPath(result.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(result.worktreePath);
const metadata = JSON.parse(readFileSync(metadataPath, "utf8"));
expect(metadata).toMatchObject({
version: 1,
@@ -325,7 +319,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
// baseRefName must be the repo's default branch, not the checked-out branch
// itself — otherwise the diff panel compares the branch against itself and
// always shows no changes.
const metadataPath = await getPaseoWorktreeMetadataPath(result.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(result.worktreePath);
const metadata = JSON.parse(readFileSync(metadataPath, "utf8"));
expect(metadata).toMatchObject({
version: 1,
@@ -376,7 +370,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
});
expect(existsSync(result.worktreePath)).toBe(true);
const metadataPath = await getPaseoWorktreeMetadataPath(result.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(result.worktreePath);
const metadata = JSON.parse(readFileSync(metadataPath, "utf8"));
expect(metadata).toMatchObject({ version: 1, baseRefName: "develop" });
});
@@ -412,9 +406,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
expect(result.branchName).toBe("main-1");
expect(existsSync(result.worktreePath)).toBe(true);
expect(
(await readPaseoWorktreeMetadata(result.worktreePath))?.changeRequestLookupTarget,
).toEqual({
expect(readPaseoWorktreeMetadata(result.worktreePath)?.changeRequestLookupTarget).toEqual({
headRef: "main-1",
localBranchName: "main-1",
});
@@ -467,7 +459,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
.trim();
expect(currentBranch).toBe("user/feature");
const metadataPath = await getPaseoWorktreeMetadataPath(result.worktreePath);
const metadataPath = getPaseoWorktreeMetadataPath(result.worktreePath);
const metadata = JSON.parse(readFileSync(metadataPath, "utf8"));
expect(metadata).toMatchObject({ baseRefName: "main" });
});
@@ -574,9 +566,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
"from-origin\n",
);
expect(
JSON.parse(
readFileSync(await getPaseoWorktreeMetadataPath(originResult.worktreePath), "utf8"),
),
JSON.parse(readFileSync(getPaseoWorktreeMetadataPath(originResult.worktreePath), "utf8")),
).toMatchObject({ baseRefName: "main" });
});
@@ -617,7 +607,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
// to the commit the worktree was cut from: "main" resolves local-first, which here is
// a different commit than the fork's upstream.
expect(
JSON.parse(readFileSync(await getPaseoWorktreeMetadataPath(result.worktreePath), "utf8")),
JSON.parse(readFileSync(getPaseoWorktreeMetadataPath(result.worktreePath), "utf8")),
).toMatchObject({ baseRefName: "main", baseRef: "refs/remotes/upstream/main" });
// An untouched child must report no work of its own. Comparing against the wrong base
@@ -704,7 +694,7 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
});
expect(
JSON.parse(readFileSync(await getPaseoWorktreeMetadataPath(result.worktreePath), "utf8")),
JSON.parse(readFileSync(getPaseoWorktreeMetadataPath(result.worktreePath), "utf8")),
).toMatchObject({
baseRefName: "release+hotfix",
baseRef: "refs/remotes/upstream/release+hotfix",
@@ -1627,104 +1617,3 @@ describe.skipIf(isPlatform("win32"))("worktree POSIX-only", () => {
});
});
});
describe.skipIf(isPlatform("win32"))("async worktree metadata", () => {
it("serializes metadata mutations per worktree while another worktree progresses", async () => {
const root = mkdtempSync(join(tmpdir(), "worktree-metadata-order-"));
const first = join(root, "first");
const second = join(root, "second");
mkdirSync(join(first, ".git"), { recursive: true });
mkdirSync(join(second, ".git"), { recursive: true });
try {
await writePaseoWorktreeMetadata(first, { baseRefName: "main" });
await writePaseoWorktreeMetadata(second, { baseRefName: "release" });
const order: string[] = [];
const runtime = writePaseoWorktreeRuntimeMetadata(first, { worktreePort: 4101 }).then(() =>
order.push("first-runtime"),
);
const unrelated = writePaseoWorktreeRuntimeMetadata(second, { worktreePort: 4102 }).then(() =>
order.push("second-runtime"),
);
const naming = writePaseoWorktreeFirstAgentBranchAutoNameMetadata(first, {
placeholderBranchName: "topic",
}).then(() => order.push("first-naming"));
// A read issued while writes are pending observes their completed snapshot.
const snapshot = readPaseoWorktreeMetadata(first);
await Promise.all([runtime, unrelated, naming]);
expect(order.indexOf("first-runtime")).toBeLessThan(order.indexOf("first-naming"));
expect(await snapshot).toMatchObject({
baseRefName: "main",
runtime: { worktreePort: 4101 },
firstAgentBranchAutoName: { status: "pending", placeholderBranchName: "topic" },
});
expect(await readPaseoWorktreeMetadata(second)).toMatchObject({
baseRefName: "release",
runtime: { worktreePort: 4102 },
});
const alias = join(root, "first-alias");
symlinkSync(first, alias, "dir");
await Promise.all([
writePaseoWorktreeRuntimeMetadata(first, { worktreePort: 4103 }),
writePaseoWorktreeFirstAgentBranchAutoNameMetadata(alias, {
placeholderBranchName: "renamed",
}),
]);
expect(await readPaseoWorktreeMetadata(first)).toMatchObject({
runtime: { worktreePort: 4103 },
firstAgentBranchAutoName: { placeholderBranchName: "renamed" },
});
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it("a queued metadata mutation cannot recreate an archived Git directory", async () => {
const root = mkdtempSync(join(tmpdir(), "worktree-metadata-archive-"));
mkdirSync(join(root, ".git"));
try {
await writePaseoWorktreeMetadata(root, { baseRefName: "main" });
const pending = writePaseoWorktreeRuntimeMetadata(root, { worktreePort: 4101 });
// The mutation is queued in a microtask, so archive wins this ordering.
rmSync(join(root, ".git"), { recursive: true });
await expect(pending).rejects.toThrow("Not a git repository");
expect(existsSync(join(root, ".git"))).toBe(false);
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it("recognizes a removed worktree under a symlinked ownership root", async () => {
const root = mkdtempSync(join(tmpdir(), "worktree-ownership-alias-"));
const actual = join(root, "actual");
const alias = join(root, "worktrees");
mkdirSync(actual);
symlinkSync(actual, alias, "dir");
try {
const cwd = join(alias, "project", "removed");
expect(
await isPaseoOwnedWorktreeCwd(cwd, { worktreesRoot: alias, knownGitCommonDir: null }),
).toMatchObject({ allowed: true, worktreePath: cwd });
} finally {
rmSync(root, { recursive: true, force: true });
}
});
it("rejects oversized metadata and named pipes without reading them whole", async () => {
const root = mkdtempSync(join(tmpdir(), "worktree-metadata-admission-"));
mkdirSync(join(root, ".git", "paseo"), { recursive: true });
try {
const path = await getPaseoWorktreeMetadataPath(root);
writeFileSync(path, JSON.stringify({ version: 1, baseRefName: "x".repeat(64 * 1024) }));
await expect(readPaseoWorktreeMetadata(root)).rejects.toThrow(
"Invalid Paseo worktree metadata file",
);
rmSync(path);
execFileSync("mkfifo", [path]);
await expect(readPaseoWorktreeMetadata(root)).rejects.toThrow(
"Invalid Paseo worktree metadata file",
);
} finally {
rmSync(root, { recursive: true, force: true });
}
});
});
+1 -10
View File
@@ -5,7 +5,6 @@ import {
deletePaseoWorktree,
isPaseoOwnedWorktreeCwd,
mapWorkspaceCwdToWorktree,
runWorktreeTeardownCommands,
slugify,
type CreateWorktreeOptions,
type WorktreeConfig,
@@ -172,15 +171,7 @@ describe("paseo worktree manager", () => {
expect(existsSync(created.worktreePath)).toBe(true);
const ownership = await isPaseoOwnedWorktreeCwd(created.worktreePath, { paseoHome });
expect(ownership).toMatchObject({ allowed: true, worktreePath: created.worktreePath });
// Windows may report the same root through native 8.3 and JS long spellings.
const nativeWorktreePath = realpathSync.native(created.worktreePath);
await expect(
runWorktreeTeardownCommands({
worktreePath: created.worktreePath,
teardownCwd: nativeWorktreePath,
}),
).resolves.toEqual([]);
expect(ownership.allowed).toBe(true);
await expect(
isPaseoOwnedWorktreeCwd(join(created.worktreePath, "packages", "app"), { paseoHome }),
).resolves.toMatchObject({
+35 -74
View File
@@ -1,12 +1,11 @@
import { execFile } from "child_process";
import { promisify } from "util";
import { constants as fsConstants, realpath } from "fs";
import { copyFile, mkdir, rm, stat } from "fs/promises";
import { constants as fsConstants, existsSync, mkdirSync, realpathSync, statSync } from "fs";
import { copyFile, rm, stat } from "fs/promises";
import { join, basename, dirname, isAbsolute, resolve, sep } from "path";
import net from "node:net";
import { createHash } from "node:crypto";
import stripAnsi from "strip-ansi";
import pLimit from "p-limit";
import {
buildStringCommandShellInvocation,
createStringCommandShellEnv,
@@ -38,12 +37,7 @@ import { spawnProcess } from "./spawn.js";
import { resolvePaseoHome } from "../server/paseo-home.js";
import { createExternalProcessEnv } from "../server/paseo-env.js";
import { parseGitRevParsePath, resolveGitRevParsePath } from "./git-rev-parse-path.js";
import {
expandTilde,
getRealpathAwareRelativePath,
getRelativePathInsideRoot,
isPathInsideRoot,
} from "./path.js";
import { expandTilde, getRealpathAwareRelativePath, isPathInsideRoot } from "./path.js";
import { terminateWithTreeKill } from "./tree-kill.js";
export { slugify, validateBranchSlug } from "@getpaseo/protocol/branch-slug";
@@ -613,7 +607,7 @@ async function assertPortAvailable(port: number): Promise<void> {
async function inferRepoRootPathFromWorktreePath(worktreePath: string): Promise<string> {
try {
const commonDir = await getGitCommonDir(worktreePath);
const normalizedCommonDir = await normalizePathForOwnership(commonDir);
const normalizedCommonDir = normalizePathForOwnership(commonDir);
// Normal repo/worktree: common dir is <repoRoot>/.git
if (basename(normalizedCommonDir) === ".git") {
return dirname(normalizedCommonDir);
@@ -629,12 +623,12 @@ async function inferRepoRootPathFromWorktreePath(worktreePath: string): Promise<
});
const topLevel = parseGitRevParsePath(stdout);
if (topLevel) {
return await normalizePathForOwnership(topLevel);
return normalizePathForOwnership(topLevel);
}
} catch {
// ignore
}
return await normalizePathForOwnership(worktreePath);
return normalizePathForOwnership(worktreePath);
}
}
@@ -717,12 +711,12 @@ export async function resolveWorktreeRuntimeEnv(options: {
const branchName =
options.branchName ?? (await resolveBranchNameForWorktreePath(options.worktreePath));
let worktreePort = await readPaseoWorktreeRuntimePort(options.worktreePath);
let worktreePort = readPaseoWorktreeRuntimePort(options.worktreePath);
if (worktreePort === null) {
worktreePort = await getAvailablePort();
const metadata = await readPaseoWorktreeMetadata(options.worktreePath);
const metadata = readPaseoWorktreeMetadata(options.worktreePath);
if (metadata) {
await writePaseoWorktreeRuntimeMetadata(options.worktreePath, { worktreePort });
writePaseoWorktreeRuntimeMetadata(options.worktreePath, { worktreePort });
}
} else {
await assertPortAvailable(worktreePort);
@@ -748,7 +742,7 @@ export async function runWorktreeTeardownCommands(options: {
repoRootPath?: string;
}): Promise<WorktreeTeardownCommandResult[]> {
const teardownCwd = options.teardownCwd ?? options.worktreePath;
if ((await getOwnershipRelativePath(options.worktreePath, teardownCwd)) === null) {
if (getRealpathAwareRelativePath(options.worktreePath, teardownCwd) === null) {
throw new Error(`Worktree teardown cwd is outside the worktree: ${teardownCwd}`);
}
const teardownCommands = getWorktreeTeardownCommands(teardownCwd);
@@ -760,7 +754,7 @@ export async function runWorktreeTeardownCommands(options: {
options.repoRootPath ?? (await inferRepoRootPathFromWorktreePath(options.worktreePath));
const branchName =
options.branchName ?? (await resolveBranchNameForWorktreePath(options.worktreePath));
const worktreePort = await readPaseoWorktreeRuntimePort(options.worktreePath);
const worktreePort = readPaseoWorktreeRuntimePort(options.worktreePath);
const teardownEnv: NodeJS.ProcessEnv = createStringCommandShellEnv(
createExternalProcessEnv(process.env, {
@@ -837,12 +831,12 @@ function deriveShortAlphanumericHash(value: string): string {
export async function deriveWorktreeProjectHash(cwd: string): Promise<string> {
try {
const commonDir = await getGitCommonDir(cwd);
const normalizedCommonDir = await normalizePathForOwnership(commonDir);
const normalizedCommonDir = normalizePathForOwnership(commonDir);
const repoRoot =
basename(normalizedCommonDir) === ".git" ? dirname(normalizedCommonDir) : normalizedCommonDir;
return deriveShortAlphanumericHash(repoRoot);
} catch {
return deriveShortAlphanumericHash(await normalizePathForOwnership(cwd));
return deriveShortAlphanumericHash(normalizePathForOwnership(cwd));
}
}
@@ -910,46 +904,16 @@ export function mapWorkspaceRelativeCwdToWorktree(input: {
return mappedCwd;
}
// Keep the former realpathSync spelling on Windows: promises.realpath uses the
// native resolver and can return an 8.3 alias instead of the long path.
const resolveOwnershipPath = promisify(realpath);
const resolveNativeOwnershipPath = promisify(realpath.native);
const ownershipIdentityIo = pLimit(2);
async function normalizePathForOwnership(input: string): Promise<string> {
function normalizePathForOwnership(input: string): string {
try {
return await ownershipIdentityIo(() => resolveOwnershipPath(input));
return realpathSync(input);
} catch {
return resolve(input);
}
}
async function ownershipPathVariants(input: string): Promise<string[]> {
const variants = new Set([input, await normalizePathForOwnership(input)]);
try {
variants.add(await ownershipIdentityIo(() => resolveNativeOwnershipPath(input)));
} catch {
// Missing descendants still match a lexical or established root spelling.
}
return [...variants];
}
// Keep lexical variants for missing paths: archive must still recognize a removed
// worktree below a symlinked worktrees root, whose full realpath no longer exists.
async function getOwnershipRelativePath(root: string, candidate: string): Promise<string | null> {
const roots = await ownershipPathVariants(root);
const candidates = await ownershipPathVariants(candidate);
for (const rootPath of roots) {
for (const candidatePath of candidates) {
const relative = getRelativePathInsideRoot(rootPath, candidatePath);
if (relative !== null) return relative;
}
}
return null;
}
async function resolveRepoRootFromGitCommonDir(commonDir: string): Promise<string> {
const normalizedCommonDir = await normalizePathForOwnership(commonDir);
function resolveRepoRootFromGitCommonDir(commonDir: string): string {
const normalizedCommonDir = normalizePathForOwnership(commonDir);
return basename(normalizedCommonDir) === ".git"
? dirname(normalizedCommonDir)
: normalizedCommonDir;
@@ -959,25 +923,25 @@ export async function isPaseoOwnedWorktreeCwd(
cwd: string,
options?: PaseoWorktreeOwnershipOptions,
): Promise<PaseoWorktreeOwnership> {
const resolvedCwd = await normalizePathForOwnership(cwd);
const resolvedCwd = normalizePathForOwnership(cwd);
// repoRoot is best-effort: git may be unreachable from the worktree (e.g. a
// previous archive attempt removed the admin dir before the working tree
// could be fully cleaned up). We still want to allow archiving in that case.
let repoRoot: string | undefined;
if (options?.knownGitCommonDir) {
repoRoot = await resolveRepoRootFromGitCommonDir(options.knownGitCommonDir);
repoRoot = resolveRepoRootFromGitCommonDir(options.knownGitCommonDir);
} else if (options?.knownGitCommonDir === undefined) {
try {
const gitCommonDir = await getGitCommonDir(cwd);
repoRoot = await resolveRepoRootFromGitCommonDir(gitCommonDir);
repoRoot = resolveRepoRootFromGitCommonDir(gitCommonDir);
} catch {
// ignore
}
}
const worktreesBaseRoot = resolvePaseoWorktreesBaseRoot(options);
const relativePath = await getOwnershipRelativePath(worktreesBaseRoot, resolvedCwd);
const relativePath = getRealpathAwareRelativePath(worktreesBaseRoot, resolvedCwd);
// Ownership is defined by the path living under <worktrees-root>/<hash>/<slug>[/...].
// The <hash>/<slug> prefix is Paseo-private — nothing else writes there — so the
@@ -1048,9 +1012,9 @@ function parseWorktreeList(output: string): ParsedPaseoWorktreeInfo[] {
return entries;
}
async function resolveWorktreeCreatedAtIso(worktreePath: string): Promise<string> {
function resolveWorktreeCreatedAtIso(worktreePath: string): string {
try {
const stats = await stat(worktreePath);
const stats = statSync(worktreePath);
const birthtimeMs = stats.birthtimeMs;
const createdAtMs =
Number.isFinite(birthtimeMs) && birthtimeMs > 0 ? birthtimeMs : stats.ctimeMs;
@@ -1075,15 +1039,12 @@ export async function listPaseoWorktrees({
envOverlay: READ_ONLY_GIT_ENV,
});
const roots = await ownershipPathVariants(projectWorktreesRoot);
const worktrees: PaseoWorktreeInfo[] = [];
// Inspect foreign paths sequentially; a large worktree list must not consume the FS pool.
for (const entry of parseWorktreeList(stdout)) {
const path = await normalizePathForOwnership(entry.path);
if (!roots.some((root) => getRelativePathInsideRoot(root, path) !== null)) continue;
worktrees.push({ ...entry, path, createdAt: await resolveWorktreeCreatedAtIso(path) });
}
return worktrees;
return parseWorktreeList(stdout)
.map((entry) => Object.assign({}, entry, { path: normalizePathForOwnership(entry.path) }))
.filter((entry) => getRealpathAwareRelativePath(projectWorktreesRoot, entry.path) !== null)
.map((entry) =>
Object.assign({}, entry, { createdAt: resolveWorktreeCreatedAtIso(entry.path) }),
);
}
export interface DeletePaseoWorktreeOptions {
@@ -1122,7 +1083,7 @@ export async function deletePaseoWorktree({
}
const requestedPath = worktreePath ?? join(resolvedWorktreesRoot, worktreeSlug!);
const resolvedRequested = await normalizePathForOwnership(requestedPath);
const resolvedRequested = normalizePathForOwnership(requestedPath);
const ownership = await isPaseoOwnedWorktreeCwd(requestedPath, {
paseoHome,
worktreesRoot: worktreesBaseRoot,
@@ -1130,7 +1091,7 @@ export async function deletePaseoWorktree({
const resolvedWorktree =
ownership.allowed && ownership.worktreePath ? ownership.worktreePath : resolvedRequested;
const relativeWorktreePath = await getOwnershipRelativePath(
const relativeWorktreePath = getRealpathAwareRelativePath(
resolvedWorktreesRoot,
resolvedWorktree,
);
@@ -1246,12 +1207,12 @@ export const createWorktree = async ({
}: CreateWorktreeOptions): Promise<CreatedWorktree> => {
const sourcePlan = await resolveWorktreeSourcePlan({ cwd, source, desiredSlug: worktreeSlug });
let worktreePath = join(await getPaseoWorktreesRoot(cwd, paseoHome, worktreesRoot), worktreeSlug);
await mkdir(dirname(worktreePath), { recursive: true });
mkdirSync(dirname(worktreePath), { recursive: true });
// Also handle worktree path collision
let finalWorktreePath = worktreePath;
let pathSuffix = 1;
while (await pathExists(finalWorktreePath)) {
while (existsSync(finalWorktreePath)) {
finalWorktreePath = `${worktreePath}-${pathSuffix}`;
pathSuffix++;
}
@@ -1261,7 +1222,7 @@ export const createWorktree = async ({
cwd,
timeout: 120_000,
});
worktreePath = await normalizePathForOwnership(finalWorktreePath);
worktreePath = normalizePathForOwnership(finalWorktreePath);
if (sourcePlan.pushRemote) {
await configureWorktreePushRemote({
@@ -1278,7 +1239,7 @@ export const createWorktree = async ({
});
}
await writePaseoWorktreeMetadata(worktreePath, {
writePaseoWorktreeMetadata(worktreePath, {
baseRefName: sourcePlan.metadataBaseRefName,
...(sourcePlan.metadataBaseRef ? { baseRef: sourcePlan.metadataBaseRef } : {}),
...(sourcePlan.changeRequestLookupTarget
+33
View File
@@ -0,0 +1,33 @@
# Agent tools
Attaches `append_workspace_note` to agents whose `notes/enabled` label is `"true"`.
Calling the tool with `{ "note": "Run the targeted tests before committing." }` appends that
line to `AGENT_NOTES.md` in the calling agent's working directory and returns the file path.
It creates the file if it does not exist.
Install on a daemon with plugins enabled:
```bash
paseo plugin install /absolute/path/to/plugin-examples/agent-tools
```
Create a labeled agent through the SDK:
```ts
const agent = await paseo.agents.create({
config: { provider: "codex/gpt-5.4" },
cwd: "/projects/shop",
labels: { "notes/enabled": "true" },
});
```
The session-open callback also runs when an agent resumes or reloads. After changing an existing
agent's labels, call `paseo.agents.ref(id).reload()` to apply the selection immediately. This
interrupts a running turn. `refresh()` reads the agent snapshot without reopening its session.
From the repository root, `npm run typecheck` includes this example. Outside this repository,
use `paseo plugin init` to create the package and TypeScript setup, then copy `index.server.ts`
and the manifest into that project.
See [Attach tools to agents](../../public-docs/plugins/reference.md#attach-tools-to-agents)
for targeting, execution, provider gates, collisions, and plugin reload behavior.
@@ -0,0 +1,26 @@
import { writeFile } from "node:fs/promises";
import path from "node:path";
import { z } from "zod";
import { defineAgentTool, type PluginServerContext } from "@getpaseo/plugin/server";
const appendWorkspaceNote = defineAgentTool({
name: "append_workspace_note",
description: "Append a note to AGENT_NOTES.md in this agent's working directory.",
input: z.object({ note: z.string().min(1).describe("Note to append") }),
async execute({ note }, { agent, signal }) {
const notePath = path.join(agent.cwd, "AGENT_NOTES.md");
await writeFile(notePath, `${note}\n`, { encoding: "utf8", flag: "a", signal });
return {
text: `Saved note to ${notePath}`,
structuredContent: { path: notePath },
};
},
});
export default function contribute(server: PluginServerContext) {
server.before("agent.session_open", ({ request }) => {
if (request.labels["notes/enabled"] !== "true") return;
return { ...request, tools: [...request.tools, appendWorkspaceNote] };
});
return () => {};
}
@@ -0,0 +1,4 @@
{
"id": "agent-tools",
"requirements": { "paseo": ">=0.13.2" }
}
+127 -6
View File
@@ -711,6 +711,120 @@ server.before("agent.session_open", ({ request }) => {
This runs on create, resume, refresh, and import. To inject only on creation, set `env` in an
`agent.create` callback instead.
### Attach tools to agents
Define a tool in `index.server.ts` and attach it in `before("agent.session_open")`.
This example gives `append_workspace_note` only to agents with the label `notes/enabled: "true"`:
```ts
import { writeFile } from "node:fs/promises";
import path from "node:path";
import { z } from "zod";
import { defineAgentTool, type PluginServerContext } from "@getpaseo/plugin/server";
const appendWorkspaceNote = defineAgentTool({
name: "append_workspace_note",
description: "Append a note to AGENT_NOTES.md in this agent's working directory.",
input: z.object({ note: z.string().min(1).describe("Note to append") }),
async execute({ note }, { agent, signal }) {
const notePath = path.join(agent.cwd, "AGENT_NOTES.md");
await writeFile(notePath, `${note}\n`, { encoding: "utf8", flag: "a", signal });
return {
text: `Saved note to ${notePath}`,
structuredContent: { path: notePath },
};
},
});
export default function contribute(server: PluginServerContext) {
server.before("agent.session_open", ({ request }) => {
if (request.labels["notes/enabled"] !== "true") return;
return { ...request, tools: [...request.tools, appendWorkspaceNote] };
});
return () => {};
}
```
Set the label when creating an agent with the SDK:
```ts
const agent = await paseo.agents.create({
config: { provider: "codex/gpt-5.4" },
cwd: "/projects/shop",
labels: { "notes/enabled": "true" },
});
```
Select agents using any of the session-open request fields:
| Select by | Condition in the callback |
| ---------------- | -------------------------------------------- |
| Label | `request.labels["notes/enabled"] === "true"` |
| Provider | `request.provider === "codex"` |
| Workspace | `request.workspaceId === "workspace-456"` |
| Individual agent | `request.agentId === chosenAgentId` |
`request.tools` starts empty for each plugin. Each callback sees the tools returned by earlier
callbacks in that plugin; replace the array to add or remove tools. Returning `undefined` preserves
the request. Tools are ignored for `purpose: "history"`.
Import `defineAgentTool`, `AgentToolDefinition`, `AgentToolContext`, and `AgentToolResult` from
`@getpaseo/plugin/server`. The helper infers the validated input from your Zod schema:
```ts
declare function defineAgentTool<Input extends z.ZodType>(tool: {
name: string;
title?: string;
description: string;
input: Input;
execute(
input: z.output<Input>,
context: AgentToolContext,
): AgentToolResult | Promise<AgentToolResult>;
}): AgentToolDefinition;
interface AgentToolResult {
text?: string;
structuredContent?: unknown;
isError?: boolean;
}
```
| Execution context | Value |
| ----------------- | --------------------------------------------------------------------------------------------------- |
| `agent` | Calling agent as a `PluginHookAgent`, including `id`, `workspaceId`, `provider`, `cwd`, and `title` |
| `paseo` | SDK connected to this daemon |
| `signal` | Abort signal for this tool call; pass it to file operations or external requests |
`execute` runs in the plugin process. Closures and the original Zod schema stay there; only tool
metadata and call inputs/results cross the process boundary. Paseo validates input before calling
`execute`. Invalid input returns schema issues to the agent. A thrown error becomes
`{ isError: true, text: message }` and the plugin keeps running. Canceling the call or closing its
transport aborts `signal` and discards the result. Tool calls have a 10 minute ceiling, separate
from the 30 second hook timeout. Closing or reopening an agent session drops its old definitions.
Return model-visible text in `text`. `structuredContent` must be JSON-serializable; when `text` is
omitted, Paseo formats that value as text for the model. Set `isError: true` to report a tool error.
Paseo delivers attached tools through its agent tool transport, using MCP or native registration
as the provider supports. The same definition and callback work with either. Attached tools obey
the [same gates as Paseo tools](/docs/mcp#limit-paseo-tools-by-provider). Only that agent sees them.
Names must match `/^[a-z][a-z0-9_]*$/` and be at most 64 characters. A name colliding with a Paseo
tool or another plugin's tool on the same agent fails the session open and names both owners.
A duplicate name within one plugin also fails the session open. Agents the callback does not
target are unaffected.
| Change | When the agent receives it |
| -------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Start, stop, or reload a plugin declaring `agent.session_open` | Open sessions become stale. Each reopens before its next idle prompt; a running turn continues without interruption. The next prompt after that reopen reuses the session. |
| Change which agents the callback targets | Run `paseo plugin reload agent-tools` after editing this example, then send the agent a prompt when it is idle. |
| Apply immediately | Call `await paseo.agents.ref(id).reload({ requestId: "reload-notes" })`. `requestId` is optional. This reopens the provider session, reruns hooks with `reason: "refresh"`, and interrupts a running turn. |
| Read the current snapshot | Call `await paseo.agents.ref(id).refresh()`. This reads state without reopening the session. |
If an automatic reopen fails, the triggering prompt fails with that error. See the
[complete agent-tools example](https://github.com/getpaseo/paseo/tree/main/plugin-examples/agent-tools).
### Choose workspace isolation
```ts
@@ -878,11 +992,11 @@ type PluginTurnOutcome =
### Before hooks
| Name | Request fields | Editable |
| -------------------- | ----------------------------------------------------------------------- | --------------------------------------- |
| `agent.create` | `config`, optional `env` | Public agent config except `cwd`; `env` |
| `agent.session_open` | `agentId`, `workspaceId`, `provider`, `cwd`, `reason`, `purpose`, `env` | Only `env` |
| `workspace.create` | `source`, optional `title`, `firstAgentContext` | Entire explicit creation request |
| Name | Request fields | Editable |
| -------------------- | ---------------------------------------------------------------------------------------------------------------- | --------------------------------------- |
| `agent.create` | `config`, optional `env` | Public agent config except `cwd`; `env` |
| `agent.session_open` | `agentId`, `workspaceId`, `provider`, `cwd`, read-only `title` and `labels`, `reason`, `purpose`, `env`, `tools` | `env` and `tools` |
| `workspace.create` | `source`, optional `title`, `firstAgentContext` | Entire explicit creation request |
**`agent.create.config`** uses `AgentSessionConfig`:
@@ -903,6 +1017,9 @@ type PluginTurnOutcome =
"workspaceId": "workspace-456",
"provider": "codex",
"cwd": "/projects/shop",
"title": "Workspace notes",
"labels": { "notes/enabled": "true" },
"tools": [],
"reason": "resume",
"purpose": "interactive",
"env": { "COMPANY_ENV": "development" }
@@ -912,6 +1029,9 @@ type PluginTurnOutcome =
| Field | Values |
| ------------- | ----------------------------------------------------------------------------------------------------------------------- |
| `workspaceId` | String or `null` |
| `title` | Read-only string or `null`; creation title on create, stored title on resume. |
| `labels` | Read-only string map; creation labels on create, stored labels on resume. |
| `tools` | Editable `AgentToolDefinition[]`, local to this plugin. See [Attach tools to agents](#attach-tools-to-agents). |
| `reason` | `create`, `resume`, `refresh`, `import` |
| `purpose` | `interactive`, `history` |
| `env` | Launch override map; excludes the daemon's inherited environment. Replace the map to add, replace, or remove overrides. |
@@ -923,7 +1043,7 @@ Creation request
→ agent.create hooks (plugin-ID order; registration order within each plugin)
→ resolve defaults and validate provider configuration
→ derive launch configuration with Paseo runtime tools and daemon prompt
→ agent.session_open hooks (same ordering; env only)
→ agent.session_open hooks (same ordering; env and plugin-local tools)
→ set PASEO_AGENT_ID and PASEO_AGENT_CWD
→ open provider session and save agent configuration
```
@@ -960,6 +1080,7 @@ saved; environment overrides are not persisted with it.
| [lifecycle-logger](https://github.com/getpaseo/paseo/tree/main/plugin-examples/lifecycle-logger) | All eleven hooks; JSON logs with environment values redacted |
| [lifecycle-actions](https://github.com/getpaseo/paseo/tree/main/plugin-examples/lifecycle-actions) | Follow-ups, permissions, environment, provider switching, worktree selection |
| [agent-configuration](https://github.com/getpaseo/paseo/tree/main/plugin-examples/agent-configuration) | MCP injection and Codex sandbox/approval options |
| [agent-tools](https://github.com/getpaseo/paseo/tree/main/plugin-examples/agent-tools) | Label-targeted tools and file operations in the plugin process |
Read logger output with `paseo plugin logs lifecycle-logger` or the host's `daemon.log`.
+130 -12
View File
@@ -585,6 +585,117 @@ server.before("agent.session_open", ({ request }) => {
它会在创建、恢复、刷新和导入时运行。如果只想在创建时注入,请改为在 `agent.create` 回调中设置 `env`。
### 为 Agent 附加工具 {#attach-tools-to-agents}
在 `index.server.ts` 中定义工具,并在 `before("agent.session_open")` 中附加它。
此示例只向带有 `notes/enabled: "true"` 标签的 Agent 提供 `append_workspace_note`:
```ts
import { writeFile } from "node:fs/promises";
import path from "node:path";
import { z } from "zod";
import { defineAgentTool, type PluginServerContext } from "@getpaseo/plugin/server";
const appendWorkspaceNote = defineAgentTool({
name: "append_workspace_note",
description: "Append a note to AGENT_NOTES.md in this agent's working directory.",
input: z.object({ note: z.string().min(1).describe("Note to append") }),
async execute({ note }, { agent, signal }) {
const notePath = path.join(agent.cwd, "AGENT_NOTES.md");
await writeFile(notePath, `${note}\n`, { encoding: "utf8", flag: "a", signal });
return {
text: `Saved note to ${notePath}`,
structuredContent: { path: notePath },
};
},
});
export default function contribute(server: PluginServerContext) {
server.before("agent.session_open", ({ request }) => {
if (request.labels["notes/enabled"] !== "true") return;
return { ...request, tools: [...request.tools, appendWorkspaceNote] };
});
return () => {};
}
```
通过 SDK 创建 Agent 时设置标签:
```ts
const agent = await paseo.agents.create({
config: { provider: "codex/gpt-5.4" },
cwd: "/projects/shop",
labels: { "notes/enabled": "true" },
});
```
使用会话打开请求中的字段选择 Agent:
| 选择依据 | 回调中的条件 |
| ---------- | -------------------------------------------- |
| 标签 | `request.labels["notes/enabled"] === "true"` |
| Provider | `request.provider === "codex"` |
| 工作区 | `request.workspaceId === "workspace-456"` |
| 单个 Agent | `request.agentId === chosenAgentId` |
每个插件的 `request.tools` 都从空数组开始。同一插件中的后续回调会看到前面回调返回的工具;
替换数组即可添加或移除工具。返回 `undefined` 会保留请求。`purpose: "history"` 时会忽略工具。
从 `@getpaseo/plugin/server` 导入 `defineAgentTool`、`AgentToolDefinition`、`AgentToolContext`
和 `AgentToolResult`。辅助函数会根据 Zod schema 推断验证后的输入类型:
```ts
declare function defineAgentTool<Input extends z.ZodType>(tool: {
name: string;
title?: string;
description: string;
input: Input;
execute(
input: z.output<Input>,
context: AgentToolContext,
): AgentToolResult | Promise<AgentToolResult>;
}): AgentToolDefinition;
interface AgentToolResult {
text?: string;
structuredContent?: unknown;
isError?: boolean;
}
```
| 执行 context | 值 |
| ------------ | --------------------------------------------------------------------------------------------------- |
| `agent` | 发起调用的 Agent,以 `PluginHookAgent` 表示,包括 `id`、`workspaceId`、`provider`、`cwd` 和 `title` |
| `paseo` | 已连接到此 daemon 的 SDK |
| `signal` | 此次工具调用的中止信号;请把它传给文件操作或外部请求 |
`execute` 在插件进程中运行。闭包和原始 Zod schema 保留在该进程中;只有工具元数据和调用的
输入、结果会跨进程传递。Paseo 在调用 `execute` 前验证输入。无效输入会将 schema 问题返回给
Agent。抛出的错误会转换为 `{ isError: true, text: message }`,插件继续运行。取消调用或关闭
其传输连接会中止 `signal` 并丢弃结果。工具调用最长运行 10 分钟,与钩子的 30 秒超时分开。
关闭或重新打开 Agent 会话会移除旧定义。
在 `text` 中返回模型可见的文本。`structuredContent` 必须可序列化为 JSON;省略 `text` 时,
Paseo 会将该值格式化为供模型读取的文本。设置 `isError: true` 可报告工具错误。
Paseo 通过 Agent 工具传输提供附加的工具,按 Provider 的支持情况使用 MCP 或原生注册。
同一份定义和回调适用于这两种方式。附加的工具遵循[与 Paseo 工具相同的限制](/docs/mcp#limit-paseo-tools-by-provider)。
只有该 Agent 能看到这些工具。
名称必须匹配 `/^[a-z][a-z0-9_]*$/`,且不超过 64 个字符。如果名称与同一 Agent 的 Paseo
工具或另一插件的工具冲突,会话打开会失败,并指出双方。一个插件内的重复名称也会导致
会话打开失败。回调未选中的 Agent 不受影响。
| 变更 | Agent 何时收到变更 |
| ------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| 启动、停止或重新加载声明了 `agent.session_open` 的插件 | 已打开的会话会标记为过期。每个会话在下一次空闲提示词前重新打开;正在运行的轮次会继续,不会被中断。重新打开后的下一条提示词复用该会话。 |
| 修改回调选择的 Agent | 编辑此示例后运行 `paseo plugin reload agent-tools`,然后在 Agent 空闲时发送提示词。 |
| 立即应用 | 调用 `await paseo.agents.ref(id).reload({ requestId: "reload-notes" })`。`requestId` 可选。这会重新打开 Provider 会话,以 `reason: "refresh"` 再次运行钩子,并中断正在运行的轮次。 |
| 读取当前快照 | 调用 `await paseo.agents.ref(id).refresh()`。这只读取状态,不会重新打开会话。 |
如果自动重新打开失败,触发它的提示词会返回该错误。参阅
[完整 agent-tools 示例](https://github.com/getpaseo/paseo/tree/main/plugin-examples/agent-tools)。
### 选择工作区隔离方式 {#choose-workspace-isolation}
```ts
@@ -741,11 +852,11 @@ type PluginTurnOutcome =
### before 钩子 {#before-hooks}
| 名称 | 请求字段 | 可编辑 |
| -------------------- | ----------------------------------------------------------------------- | ----------------------------------- |
| `agent.create` | `config`,可选的 `env` | 除 `cwd` 外的公开 Agent 配置;`env` |
| `agent.session_open` | `agentId`, `workspaceId`, `provider`, `cwd`, `reason`, `purpose`, `env` | 仅 `env` |
| `workspace.create` | `source`,可选的 `title`,`firstAgentContext` | 整个显式创建请求 |
| 名称 | 请求字段 | 可编辑 |
| -------------------- | ------------------------------------------------------------------------------------------------------------ | ----------------------------------- |
| `agent.create` | `config`,可选的 `env` | 除 `cwd` 外的公开 Agent 配置;`env` |
| `agent.session_open` | `agentId`, `workspaceId`, `provider`, `cwd`,只读的 `title` 和 `labels`,`reason`, `purpose`, `env`, `tools` | `env` 和 `tools` |
| `workspace.create` | `source`,可选的 `title`,`firstAgentContext` | 整个显式创建请求 |
**`agent.create.config`** 使用 `AgentSessionConfig`:
@@ -766,18 +877,24 @@ type PluginTurnOutcome =
"workspaceId": "workspace-456",
"provider": "codex",
"cwd": "/projects/shop",
"title": "Workspace notes",
"labels": { "notes/enabled": "true" },
"tools": [],
"reason": "resume",
"purpose": "interactive",
"env": { "COMPANY_ENV": "development" }
}
```
| 字段 | 取值 |
| ------------- | ------------------------------------------------------------------------------------ |
| `workspaceId` | 字符串或 `null` |
| `reason` | `create`、`resume`、`refresh`、`import` |
| `purpose` | `interactive`、`history` |
| `env` | 启动覆盖映射;不包括 daemon 继承的环境变量。替换整个映射即可添加、替换或移除覆盖项。 |
| 字段 | 取值 |
| ------------- | -------------------------------------------------------------------------------------------------------- |
| `workspaceId` | 字符串或 `null` |
| `title` | 只读字符串或 `null`;创建时使用创建请求的标题,恢复时使用已保存的标题。 |
| `labels` | 只读的字符串映射;创建时使用创建请求的标签,恢复时使用已保存的标签。 |
| `tools` | 可编辑的 `AgentToolDefinition[]`,仅包含此插件的工具。参阅[为 Agent 附加工具](#attach-tools-to-agents)。 |
| `reason` | `create`、`resume`、`refresh`、`import` |
| `purpose` | `interactive`、`history` |
| `env` | 启动覆盖映射;不包括 daemon 继承的环境变量。替换整个映射即可添加、替换或移除覆盖项。 |
### 顺序与返回值 {#ordering-and-returned-values}
@@ -786,7 +903,7 @@ Creation request
→ agent.create hooks (plugin-ID order; registration order within each plugin)
→ resolve defaults and validate provider configuration
→ derive launch configuration with Paseo runtime tools and daemon prompt
→ agent.session_open hooks (same ordering; env only)
→ agent.session_open hooks (same ordering; env and plugin-local tools)
→ set PASEO_AGENT_ID and PASEO_AGENT_CWD
→ open provider session and save agent configuration
```
@@ -822,6 +939,7 @@ Creation request
| [lifecycle-logger](https://github.com/getpaseo/paseo/tree/main/plugin-examples/lifecycle-logger) | 全部十一个钩子;隐去环境变量值的 JSON 日志 |
| [lifecycle-actions](https://github.com/getpaseo/paseo/tree/main/plugin-examples/lifecycle-actions) | 后续消息、权限、环境变量、切换 Provider、选择 worktree |
| [agent-configuration](https://github.com/getpaseo/paseo/tree/main/plugin-examples/agent-configuration) | MCP 注入以及 Codex 沙盒/批准选项 |
| [agent-tools](https://github.com/getpaseo/paseo/tree/main/plugin-examples/agent-tools) | 按标签选择 Agent,并在插件进程中执行文件操作的工具 |
用 `paseo plugin logs lifecycle-logger` 或主机的 `daemon.log` 读取 logger 的输出。
+1 -112
View File
@@ -1,8 +1,5 @@
import assert from "node:assert/strict";
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";
import { readFileSync, readdirSync } from "node:fs";
import { relative as relativePath } from "node:path";
import test from "node:test";
import { runInNewContext } from "node:vm";
@@ -533,111 +530,3 @@ test("Windows units and integration run independently with complete routed cover
assert.equal(selected(integrationCondition, outputs, true), false);
}
});
test("macOS release retries only transient stapling failures without rebuilding the signed app", async (t) => {
const workflow = readFileSync(new URL(".github/workflows/desktop-release.yml", repoRoot), "utf8");
const wrapper = workflow
.match(/<<'STAPLER'\n([\s\S]*?)\n STAPLER/)[1]
.replace(/^ /gm, "");
const networkError =
'Error Domain=NSURLErrorDomain Code=-1005 "The network connection was lost."';
const cases = [
{
name: "success",
outcomes: [0],
output: "stapled",
command: "stapler",
expected: 0,
attempts: 1,
},
{
name: "transient then success",
outcomes: [68, 0],
output: networkError,
command: "stapler",
expected: 0,
attempts: 2,
},
{
name: "exhausted network retries",
outcomes: [68, 68, 68, 68, 0],
output: networkError,
command: "stapler",
expected: 68,
attempts: 4,
},
{
name: "non-network staple error",
outcomes: [68, 0],
output: "ticket not found",
command: "stapler",
expected: 68,
attempts: 1,
},
{
name: "different exit code",
outcomes: [1, 0],
output: networkError,
command: "stapler",
expected: 1,
attempts: 1,
},
{
name: "notary submission is not retried",
outcomes: [68, 0],
output: networkError,
command: "notarytool",
expected: 68,
attempts: 1,
},
];
for (const scenario of cases) {
await t.test(scenario.name, () => {
const directory = mkdtempSync(join(tmpdir(), "stapler-retry-"));
try {
const statePath = join(directory, "state.json");
const realXcrun = join(directory, "real-xcrun");
const wrapperPath = join(directory, "xcrun");
writeFileSync(wrapperPath, wrapper, { mode: 0o755 });
writeFileSync(join(directory, "sleep"), "#!/bin/sh\nexit 0\n", { mode: 0o755 });
writeFileSync(
statePath,
JSON.stringify({ outcomes: scenario.outcomes, output: scenario.output, calls: [] }),
);
writeFileSync(
realXcrun,
`#!/usr/bin/env node
const fs = require("node:fs");
const state = JSON.parse(fs.readFileSync(process.env.STAPLER_STATE, "utf8"));
state.calls.push(process.argv.slice(2));
const code = state.outcomes.shift();
fs.writeFileSync(process.env.STAPLER_STATE, JSON.stringify(state));
console.log(state.output);
process.exit(code);
`,
{ mode: 0o755 },
);
const args = [scenario.command, "staple", "-v", "Paseo signed.app"];
const result = spawnSync("bash", [wrapperPath, ...args], {
encoding: "utf8",
env: {
...process.env,
PASEO_REAL_XCRUN: realXcrun,
STAPLER_STATE: statePath,
PATH: `${directory}:${process.env.PATH}`,
},
timeout: 5000,
});
assert.equal(result.status, scenario.expected, result.stdout + result.stderr);
const state = JSON.parse(readFileSync(statePath, "utf8"));
assert.deepEqual(
state.calls,
Array.from({ length: scenario.attempts }, () => args),
);
assert.match(result.stdout, /stapled|network connection was lost|ticket not found/);
} finally {
rmSync(directory, { recursive: true, force: true });
}
});
}
});