fix(creation): prevent duplicate workspaces and agents on retries

This commit is contained in:
Mohamed Boudra
2026-09-14 10:38:49 +02:00
parent d1b705a0cd
commit 815b83c96d
31 changed files with 800 additions and 243 deletions
@@ -31,6 +31,7 @@ import type { FirstAgentContext } from "@getpaseo/protocol/messages";
import { runWithGitCommandPriority } from "../utils/run-git-command.js";
export interface CreatePaseoWorktreeInput extends CreateWorktreeCoreInput {
workspaceId?: string;
projectId?: string;
title?: string;
}
@@ -93,6 +94,7 @@ async function createPaseoWorktreeWithPriority(
const workspace = await deps.workspaceProvisioning.createWorkspaceForWorktree({
sourceCwd: workspaceCwdPlan.inputCwd,
projectId: input.projectId,
workspaceId: input.workspaceId,
repoRoot: createdWorktree.repoRoot,
cwd: workspaceCwd,
worktreeRoot: createdWorktree.worktree.worktreePath,
@@ -2,7 +2,7 @@ import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { afterEach, expect, test } from "vitest";
import { AgentRequests } from "./index.js";
import { RequestReceipts } from "./index.js";
const directories: string[] = [];
afterEach(async () => {
@@ -13,7 +13,7 @@ afterEach(async () => {
async function fixture() {
const directory = await mkdtemp(path.join(tmpdir(), "agent-requests-"));
directories.push(directory);
return { directory, requests: new AgentRequests(directory) };
return { directory, requests: new RequestReceipts(directory) };
}
test("concurrent and reconstructed creates return the same durable agent", async () => {
@@ -29,9 +29,9 @@ test("concurrent and reconstructed creates return the same durable agent", async
agents.add(id);
},
};
const ids = await Promise.all([requests.create(input), requests.create(input)]);
const ids = await Promise.all([requests.createAgent(input), requests.createAgent(input)]);
expect(ids[0]).toBe(ids[1]);
expect(await new AgentRequests(directory).create(input)).toBe(ids[0]);
expect(await new RequestReceipts(directory).createAgent(input)).toBe(ids[0]);
expect(creations).toBe(1);
});
@@ -47,8 +47,8 @@ test("recovers creation when the agent was persisted before acknowledgement fail
throw new Error("acknowledgement lost");
},
};
await expect(requests.create(input)).rejects.toThrow("acknowledgement lost");
expect(await new AgentRequests(directory).create(input)).toBe([...agents][0]);
await expect(requests.createAgent(input)).rejects.toThrow("acknowledgement lost");
expect(await new RequestReceipts(directory).createAgent(input)).toBe([...agents][0]);
expect(agents.size).toBe(1);
});
@@ -60,8 +60,8 @@ test("reusing a create key with different configuration is a conflict", async ()
findAgent: async () => true,
create: async () => {},
};
await requests.create(input);
await expect(requests.create({ ...input, request: { model: "b" } })).rejects.toThrow(
await requests.createAgent(input);
await expect(requests.createAgent({ ...input, request: { model: "b" } })).rejects.toThrow(
"agent_request_key_conflict",
);
});
@@ -77,10 +77,10 @@ test("message retries survive reconstruction without submitting twice", async ()
deliveries++;
},
};
await Promise.all([requests.send(input), requests.send(input)]);
await new AgentRequests(directory).send(input);
await Promise.all([requests.sendMessage(input), requests.sendMessage(input)]);
await new RequestReceipts(directory).sendMessage(input);
expect(deliveries).toBe(1);
await requests.send({ ...input, agentId: "another" });
await requests.sendMessage({ ...input, agentId: "another" });
expect(deliveries).toBe(2);
});
@@ -96,8 +96,8 @@ test("ambiguous provider delivery is never blindly replayed after restart", asyn
throw new Error("connection lost");
},
};
await expect(requests.send(input)).rejects.toThrow("connection lost");
await expect(new AgentRequests(directory).send(input)).rejects.toThrow(
await expect(requests.sendMessage(input)).rejects.toThrow("connection lost");
await expect(new RequestReceipts(directory).sendMessage(input)).rejects.toThrow(
"agent_request_outcome_unknown",
);
expect(deliveries).toBe(1);
@@ -114,9 +114,9 @@ test("a creation failure with no stored agent can be retried", async () => {
if (!available) throw new Error("provider unavailable");
},
};
await expect(requests.create(input)).rejects.toThrow("provider unavailable");
await expect(requests.createAgent(input)).rejects.toThrow("provider unavailable");
available = true;
await expect(requests.create(input)).resolves.toEqual(expect.any(String));
await expect(requests.createAgent(input)).resolves.toEqual(expect.any(String));
});
test("failed local message preparation does not leave an ambiguous receipt", async () => {
@@ -134,10 +134,66 @@ test("failed local message preparation does not leave an ambiguous receipt", asy
sends++;
},
};
await expect(requests.send(input)).rejects.toThrow("load failed");
await expect(requests.sendMessage(input)).rejects.toThrow("load failed");
available = true;
await new AgentRequests(directory).send(input);
await new RequestReceipts(directory).sendMessage(input);
available = false;
await requests.send(input);
await requests.sendMessage(input);
expect(sends).toBe(1);
});
test("workspace receipts serialize concurrent requests and survive reconstruction", async () => {
const { requests, directory } = await fixture();
const workspaces = new Set<string>();
let creations = 0;
const input = {
key: "workspace",
workspaceId: "wks_first",
request: { source: { kind: "directory", path: "/project" } },
findWorkspace: async (id: string) => workspaces.has(id),
create: async (id: string) => {
creations++;
workspaces.add(id);
},
};
expect(
await Promise.all([
requests.createWorkspace(input),
requests.createWorkspace({ ...input, workspaceId: "wks_second" }),
]),
).toEqual(["wks_first", "wks_first"]);
expect(await new RequestReceipts(directory).createWorkspace(input)).toBe("wks_first");
expect(creations).toBe(1);
await expect(
requests.createWorkspace({
...input,
request: { source: { kind: "directory", path: "/other" } },
}),
).rejects.toThrow("workspace_request_key_conflict");
expect(
await requests.createAgent({
key: input.key,
request: {},
findAgent: async () => false,
create: async () => {},
}),
).not.toBe("wks_first");
});
test("a workspace persisted before a lost acknowledgement is recovered", async () => {
const { requests, directory } = await fixture();
const workspaces = new Set<string>();
const input = {
key: "lost-workspace",
workspaceId: "wks_created",
request: {},
findWorkspace: async (id: string) => workspaces.has(id),
create: async (id: string) => {
workspaces.add(id);
throw new Error("acknowledgement lost");
},
};
await expect(requests.createWorkspace(input)).rejects.toThrow("acknowledgement lost");
expect(await new RequestReceipts(directory).createWorkspace(input)).toBe("wks_created");
expect([...workspaces]).toEqual(["wks_created"]);
});
@@ -2,36 +2,57 @@ import { createHash, randomUUID } from "node:crypto";
import { readFile, rm } from "node:fs/promises";
import path from "node:path";
import { z } from "zod";
import { writeJsonFileAtomic } from "../../atomic-file.js";
import { writeJsonFileAtomic } from "../atomic-file.js";
const ReceiptSchema = z.object({
const ReceiptStateSchema = z.object({
fingerprint: z.string(),
agentId: z.string(),
state: z.enum(["pending", "completed"]),
});
type Receipt = z.infer<typeof ReceiptSchema>;
// Preserve existing agent receipt files while giving workspaces their own identity.
const ReceiptSchema = z.union([
ReceiptStateSchema.extend({ agentId: z.string() }),
ReceiptStateSchema.extend({ workspaceId: z.string() }),
]);
type Receipt = z.infer<typeof ReceiptStateSchema> & {
resource: { kind: "agent" | "workspace"; id: string };
};
/** One daemon-owned request journal, shared by all of its socket sessions. */
export class AgentRequests {
export class RequestReceipts {
private readonly pending = new Map<string, Promise<string>>();
constructor(private readonly directory: string) {}
create(input: {
createAgent(input: {
key: string;
request: unknown;
findAgent: (agentId: string) => Promise<boolean>;
create: (agentId: string) => Promise<void>;
}): Promise<string> {
return this.execute(["create", input.key], input.request, {
agentId: randomUUID(),
resource: { kind: "agent", id: randomUUID() },
recover: input.findAgent,
run: input.create,
retrySafe: async (agentId) => !(await input.findAgent(agentId)),
});
}
async send(input: {
createWorkspace(input: {
key: string;
request: unknown;
workspaceId: string;
findWorkspace: (workspaceId: string) => Promise<boolean>;
create: (workspaceId: string) => Promise<void>;
}): Promise<string> {
return this.execute(["create-workspace", input.key], input.request, {
resource: { kind: "workspace", id: input.workspaceId },
recover: input.findWorkspace,
run: input.create,
retrySafe: async (workspaceId) => !(await input.findWorkspace(workspaceId)),
});
}
async sendMessage(input: {
agentId: string;
messageId: string;
request: unknown;
@@ -39,7 +60,7 @@ export class AgentRequests {
prepare?: () => Promise<void>;
}): Promise<void> {
await this.execute(["send", input.agentId, input.messageId], input.request, {
agentId: input.agentId,
resource: { kind: "agent", id: input.agentId },
// A provider call can take effect before the daemon records its outcome.
// Never repeat that call merely because a process died in this window.
recover: async () => false,
@@ -52,7 +73,7 @@ export class AgentRequests {
identity: string[],
request: unknown,
operation: {
agentId: string;
resource: Receipt["resource"];
recover: (agentId: string) => Promise<boolean>;
run: (agentId: string) => Promise<void>;
prepare?: (() => Promise<void>) | undefined;
@@ -79,7 +100,7 @@ export class AgentRequests {
key: string,
fingerprint: string,
operation: {
agentId: string;
resource: Receipt["resource"];
recover: (agentId: string) => Promise<boolean>;
run: (agentId: string) => Promise<void>;
prepare?: (() => Promise<void>) | undefined;
@@ -89,39 +110,58 @@ export class AgentRequests {
const file = path.join(this.directory, `${key}.json`);
const existing = await readReceipt(file);
if (existing) {
if (existing.fingerprint !== fingerprint) throw new Error("agent_request_key_conflict");
if (existing.state === "completed") return existing.agentId;
if (!(await operation.recover(existing.agentId))) {
throw new Error("agent_request_outcome_unknown");
if (existing.fingerprint !== fingerprint)
throw new Error(`${operation.resource.kind}_request_key_conflict`);
if (existing.state === "completed") return existing.resource.id;
if (!(await operation.recover(existing.resource.id))) {
throw new Error(`${operation.resource.kind}_request_outcome_unknown`);
}
await writeJsonFileAtomic(file, { ...existing, state: "completed" });
return existing.agentId;
await writeReceipt(file, { ...existing, state: "completed" });
return existing.resource.id;
}
await operation.prepare?.();
const receipt: Receipt = { fingerprint, agentId: operation.agentId, state: "pending" };
await writeJsonFileAtomic(file, receipt);
const receipt: Receipt = { fingerprint, resource: operation.resource, state: "pending" };
await writeReceipt(file, receipt);
try {
await operation.run(receipt.agentId);
await operation.run(receipt.resource.id);
} catch (error) {
// Keyed creation has no initial prompt. Once its normal cleanup finished,
// absence of an agent confirms that retrying cannot duplicate one.
if (await operation.retrySafe?.(receipt.agentId)) await rm(file, { force: true });
// Once creation cleanup finished, absence of its persisted resource
// confirms that retrying cannot duplicate it.
if (await operation.retrySafe?.(receipt.resource.id)) await rm(file, { force: true });
throw error;
}
await writeJsonFileAtomic(file, { ...receipt, state: "completed" });
return receipt.agentId;
await writeReceipt(file, { ...receipt, state: "completed" });
return receipt.resource.id;
}
}
async function readReceipt(file: string): Promise<Receipt | null> {
try {
return ReceiptSchema.parse(JSON.parse(await readFile(file, "utf8")));
const stored = ReceiptSchema.parse(JSON.parse(await readFile(file, "utf8")));
return {
fingerprint: stored.fingerprint,
state: stored.state,
resource:
"agentId" in stored
? { kind: "agent", id: stored.agentId }
: { kind: "workspace", id: stored.workspaceId },
};
} catch (error) {
if (error instanceof Error && "code" in error && error.code === "ENOENT") return null;
throw error;
}
}
async function writeReceipt(file: string, receipt: Receipt): Promise<void> {
await writeJsonFileAtomic(file, {
fingerprint: receipt.fingerprint,
state: receipt.state,
...(receipt.resource.kind === "agent"
? { agentId: receipt.resource.id }
: { workspaceId: receipt.resource.id }),
});
}
function digest(value: unknown): string {
return createHash("sha256")
.update(
+2 -2
View File
@@ -1,4 +1,4 @@
import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
import { createRequestReceiptsStub } from "./test-utils/session-stubs.js";
import { execSync } from "child_process";
import { existsSync, mkdtempSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "fs";
import { tmpdir } from "os";
@@ -364,7 +364,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
const messages = options.messages ?? [];
const sessionOptions: SessionOptions = {
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: options.clientId ?? "test-client",
onMessage: (message) => messages.push(message),
...(options.targetedMessages
+93 -73
View File
@@ -1,7 +1,7 @@
import type { BrowserToolsBroker } from "./browser-tools/broker.js";
import { BrowserAutomationHostCapabilitySchema } from "@getpaseo/protocol/browser-automation/capabilities";
import type { SessionEventSubscription } from "@getpaseo/protocol/messages";
import type { AgentRequests } from "./agent/requests/index.js";
import type { RequestReceipts } from "./request-receipts/index.js";
import equal from "fast-deep-equal";
import { SessionDelivery, type OwnedSubscription } from "./session/owned-subscriptions/index.js";
import { v4 as uuidv4 } from "uuid";
@@ -147,6 +147,7 @@ import {
checkoutLiteFromGitSnapshot,
checkoutFromPersistedWorkspacePlacement,
deriveWorkspaceDisplayName,
generateWorkspaceId,
} from "./workspace-registry-model.js";
import { resolveWorkspaceIdForPath } from "./resolve-workspace-id-for-path.js";
import {
@@ -458,7 +459,7 @@ export interface SessionOptions {
worktreesRoot?: string;
agentManager: AgentManager;
agentStorage: AgentStorage;
agentRequests: Pick<AgentRequests, "create" | "send">;
requestReceipts: Pick<RequestReceipts, "createAgent" | "createWorkspace" | "sendMessage">;
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
directorySync?: DirectorySyncService;
@@ -780,7 +781,10 @@ export class Session {
private readonly daemonSession: DaemonSession;
private readonly hubExecutionController: HubExecutionController | null;
private readonly workspaceScripts: WorkspaceScriptsService;
private readonly agentRequests: Pick<AgentRequests, "create" | "send">;
private readonly requestReceipts: Pick<
RequestReceipts,
"createAgent" | "createWorkspace" | "sendMessage"
>;
private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch;
constructor(options: SessionOptions) {
@@ -855,7 +859,7 @@ export class Session {
this.onWorkspaceRecovered = onWorkspaceRecovered ?? null;
this.pushNotifications = pushNotifications;
this.paseoHome = paseoHome;
this.agentRequests = options.agentRequests;
this.requestReceipts = options.requestReceipts;
this.projectIcons = new ProjectIconReader(paseoHome);
this.worktreesRoot = worktreesRoot;
this.pluginRuntime = pluginRuntime;
@@ -3826,7 +3830,7 @@ export class Session {
throw new Error("Idempotent creation requires sending the initial prompt separately");
}
const { requestId: _requestId, idempotencyKey, ...request } = msg;
const id = await this.agentRequests.create({
const id = await this.requestReceipts.createAgent({
key: idempotencyKey,
request,
findAgent: async (agentId) =>
@@ -6346,23 +6350,67 @@ export class Session {
request: Extract<SessionInboundMessage, { type: "workspace.create.request" }>,
): Promise<void> {
try {
if (this.pluginRuntime) {
const { type, requestId, ...input } = request;
const transformed = await this.pluginRuntime.before("workspace.create", input);
request = { ...transformed, type, requestId };
const create = async (workspaceId?: string) => {
let creationRequest = request;
// Hooks belong to the operation: retries fingerprint the caller's input
// and must not rerun hooks or compare their potentially changing output.
if (this.pluginRuntime) {
const { type, requestId, ...input } = request;
const transformed = await this.pluginRuntime.before("workspace.create", input);
creationRequest = { ...transformed, type, requestId };
}
return creationRequest.source.kind === "directory"
? this.handleWorkspaceCreateLocal(creationRequest, workspaceId)
: this.handleWorkspaceCreateWorktree(creationRequest, workspaceId);
};
let descriptor: WorkspaceDescriptorPayload | undefined;
if (request.idempotencyKey !== undefined) {
const { requestId: _requestId, idempotencyKey, ...payload } = request;
const workspaceId = await this.requestReceipts.createWorkspace({
key: idempotencyKey,
request: payload,
workspaceId: generateWorkspaceId(),
findWorkspace: async (id) => (await this.workspaceRegistry.get(id)) !== null,
create: async (id) => {
descriptor = await create(id);
},
});
if (!descriptor) {
const workspace = await this.workspaceRegistry.get(workspaceId);
if (!workspace) throw new Error("Previously created workspace no longer exists");
descriptor = await this.describeWorkspaceRecordWithGitData(workspace);
}
} else {
descriptor = await create();
}
if (request.source.kind === "directory") {
await this.handleWorkspaceCreateLocal(request);
return;
}
await this.handleWorkspaceCreateWorktree(request);
const workspace = await this.workspaceRegistry.get(descriptor.id);
this.emit({
type: "workspace.create.response",
payload: {
requestId: request.requestId,
workspace: descriptor,
setupTerminalId: null,
...(workspace?.untrustedSource
? {
setupSkippedReason: formatWorkspaceAutomationBlockedMessage(
workspace.untrustedSource,
),
}
: {}),
error: null,
},
});
} catch (error) {
const message = error instanceof Error ? error.message : "Failed to create workspace";
this.sessionLogger.error(
{ err: error, sourceKind: request.source.kind, requestId: request.requestId },
"Failed to create workspace",
);
const errorCode = error instanceof WorkspaceProvisioningError ? error.code : undefined;
const errorCode =
error instanceof WorkspaceProvisioningError || error instanceof SessionRequestError
? error.code
: undefined;
this.emit({
type: "workspace.create.response",
payload: {
@@ -6378,25 +6426,16 @@ export class Session {
private async handleWorkspaceCreateLocal(
request: Extract<SessionInboundMessage, { type: "workspace.create.request" }>,
): Promise<void> {
workspaceId?: string,
): Promise<WorkspaceDescriptorPayload> {
if (request.source.kind !== "directory") {
return;
throw new Error("Unexpected workspace source");
}
const cwd = expandTilde(request.source.path);
const directoryExists = await this.filesystem.isDirectory(cwd).catch(() => false);
if (!directoryExists) {
this.emit({
type: "workspace.create.response",
payload: {
requestId: request.requestId,
workspace: null,
setupTerminalId: null,
error: `Directory not found: ${cwd}`,
errorCode: "directory_not_found",
},
});
return;
throw new SessionRequestError("directory_not_found", `Directory not found: ${cwd}`);
}
const explicitTitle = request.title?.trim() || null;
@@ -6405,19 +6444,10 @@ export class Session {
cwd,
explicitTitle ?? promptTitle,
request.source.projectId,
{ expectsInitialAgent: Boolean(request.firstAgentContext) },
{ expectsInitialAgent: Boolean(request.firstAgentContext), workspaceId },
);
await this.syncWorkspaceGitObserverForWorkspace(workspace);
const descriptor = await this.describeWorkspaceRecord(workspace);
this.emit({
type: "workspace.create.response",
payload: {
requestId: request.requestId,
workspace: descriptor,
setupTerminalId: null,
error: null,
},
});
await this.emitCreatedWorkspaceUpdate(
descriptor,
request.firstAgentContext ? "running" : undefined,
@@ -6441,29 +6471,24 @@ export class Session {
{ currentSelection: this.getFocusedAgentSelectionForCwd(workspace.cwd) },
);
}
return descriptor;
}
private async handleWorkspaceCreateWorktree(
request: Extract<SessionInboundMessage, { type: "workspace.create.request" }>,
): Promise<void> {
workspaceId?: string,
): Promise<WorkspaceDescriptorPayload> {
if (request.source.kind !== "worktree") {
return;
throw new Error("Unexpected workspace source");
}
const source = request.source;
if (!source.cwd && !source.projectId) {
this.emit({
type: "workspace.create.response",
payload: {
requestId: request.requestId,
workspace: null,
setupTerminalId: null,
error: "cwd or projectId is required for a worktree-backed workspace",
errorCode: "source_required",
},
});
return;
throw new SessionRequestError(
"source_required",
"cwd or projectId is required for a worktree-backed workspace",
);
}
const sourceCwd = await resolveWorktreeSourceCwd(source, this.projectRegistry);
@@ -6471,6 +6496,7 @@ export class Session {
const result = await this.createPaseoWorktreeWorkflow(
{
cwd: sourceCwd,
workspaceId,
projectId: source.projectId,
worktreeSlug: source.worktreeSlug,
action: source.action,
@@ -6487,26 +6513,11 @@ export class Session {
);
const descriptor = await this.describeCreatedWorktreeWorkspace(result);
this.emit({
type: "workspace.create.response",
payload: {
requestId: request.requestId,
workspace: descriptor,
setupTerminalId: null,
...(result.workspace.untrustedSource
? {
setupSkippedReason: formatWorkspaceAutomationBlockedMessage(
result.workspace.untrustedSource,
),
}
: {}),
error: null,
},
});
await this.emitCreatedWorkspaceUpdate(
descriptor,
request.firstAgentContext ? "running" : undefined,
);
return descriptor;
}
private async handleOpenProjectRequest(
@@ -7837,6 +7848,19 @@ export class Session {
}
}
private async prepareAgentMessage(agentId: string, text: string): Promise<void> {
await ensureAgentLoaded(agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
});
const stored = await this.agentStorage.get(agentId);
if (stored && !stored.title && !stored.lastUserMessageAt) {
const { provisionalTitle } = resolveCreateAgentTitles({ initialPrompt: text });
if (provisionalTitle) await this.agentManager.setTitle(agentId, provisionalTitle);
}
}
private async handleSendAgentMessageRequest(
msg: Extract<SessionInboundMessage, { type: "send_agent_message_request" }>,
): Promise<void> {
@@ -7887,16 +7911,12 @@ export class Session {
}
};
if (msg.messageId) {
await this.agentRequests.send({
await this.requestReceipts.sendMessage({
agentId,
messageId: msg.messageId,
request: { prompt, activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt" },
prepare: async () => {
await ensureAgentLoaded(agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
});
await this.prepareAgentMessage(agentId, msg.text);
},
send,
});
@@ -1,4 +1,4 @@
import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
import { createRequestReceiptsStub } from "./test-utils/session-stubs.js";
import { describe, expect, test, vi } from "vitest";
import path from "node:path";
import type pino from "pino";
@@ -185,7 +185,7 @@ function createSessionForWorkspaceGitWatchTests(options?: {
};
const session = new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage: (message) => emitted.push(message as { type: string; payload: unknown }),
@@ -1,4 +1,4 @@
import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
import { createRequestReceiptsStub } from "./test-utils/session-stubs.js";
// Invariant tests for cwd → workspace resolution on the open_project_request path.
// Each test encodes a default behavior we want from `findOrCreateWorkspaceForDirectory`.
// Run to see which invariants the current code already satisfies (green) and which
@@ -89,7 +89,7 @@ function createHarness(input: {
};
const session = new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test",
permissions: OWNER_PERMISSIONS,
appVersion: null,
@@ -1,4 +1,4 @@
import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
import { createRequestReceiptsStub } from "./test-utils/session-stubs.js";
import { execFileSync } from "node:child_process";
import {
existsSync,
@@ -642,7 +642,7 @@ function createSessionForWorkspaceTests(
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
appVersion: options.appVersion ?? null,
@@ -1003,7 +1003,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist
const emitted: SessionOutboundMessage[] = [];
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
serverId: "test-server",
permissions: OWNER_PERMISSIONS,
@@ -1157,7 +1157,7 @@ test("create_agent_request launches from an exact subdirectory in a created work
const emitted: SessionOutboundMessage[] = [];
const session = new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
appVersion: null,
@@ -1295,7 +1295,7 @@ test("create_agent_request does not title an existing workspace from the agent p
let generateCalls = 0;
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
appVersion: null,
@@ -1565,7 +1565,7 @@ test("archive emits an authoritative agent_update upsert for subscribed clients"
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage: (message) => emitted.push(message),
@@ -2048,7 +2048,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy
const cancelAgentRun = vi.fn(async () => ({ status: "settled" as const }));
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage: (message) => emitted.push(message),
@@ -2217,7 +2217,7 @@ test("close_items_request archives stored agents that are not currently loaded",
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage: (message) => emitted.push(message),
@@ -2377,7 +2377,7 @@ test("close_items_request continues after an archive failure", async () => {
const killTerminalBestEffort = vi.fn();
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage: (message) => emitted.push(message),
@@ -3649,7 +3649,7 @@ test("workspace update stream keeps persisted workspace visible after agents sto
const session = asTestSession(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage: (message) => emitted.push(message),
@@ -40,6 +40,7 @@ export interface ImportWorkspaceResult<T> {
export interface CreateWorktreeWorkspaceInput {
sourceCwd: string;
projectId?: string;
workspaceId?: string;
repoRoot: string;
cwd: string;
worktreeRoot: string;
@@ -61,7 +62,7 @@ export interface WorkspaceProvisioningService {
cwd: string,
title?: string | null,
projectId?: string,
context?: { expectsInitialAgent?: boolean },
context?: { expectsInitialAgent?: boolean; workspaceId?: string },
): Promise<PersistedWorkspaceRecord>;
createWorkspaceForWorktree(
input: CreateWorktreeWorkspaceInput,
@@ -201,7 +202,7 @@ export function createWorkspaceProvisioningService(deps: {
cwd: string,
title?: string | null,
projectId?: string,
context?: { expectsInitialAgent?: boolean },
context?: { expectsInitialAgent?: boolean; workspaceId?: string },
): Promise<PersistedWorkspaceRecord> {
const normalizedCwd = resolve(cwd);
const checkout = await workspaceGitService.getCheckout(normalizedCwd);
@@ -211,7 +212,7 @@ export function createWorkspaceProvisioningService(deps: {
await findOrCreateProjectForDirectory(normalizedCwd);
const timestamp = new Date().toISOString();
const workspace = createPersistedWorkspaceRecord({
workspaceId: generateWorkspaceId(),
workspaceId: context?.workspaceId ?? generateWorkspaceId(),
projectId: project.projectId,
...initialWorkspacePlacement({ source: "checkout", cwd: normalizedCwd, checkout }),
title: title?.trim() || null,
@@ -237,7 +238,7 @@ export function createWorkspaceProvisioningService(deps: {
});
const timestamp = new Date().toISOString();
const workspace = createPersistedWorkspaceRecord({
workspaceId: generateWorkspaceId(),
workspaceId: input.workspaceId ?? generateWorkspaceId(),
projectId: project.projectId,
...initialWorkspacePlacement({
source: "created_worktree",
@@ -1,4 +1,4 @@
import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
import { createRequestReceiptsStub } from "./test-utils/session-stubs.js";
import os from "node:os";
import path from "node:path";
import { mkdtempSync, rmSync } from "node:fs";
@@ -100,7 +100,7 @@ describe("snapshot mutation ownership boundary", () => {
const session = asInternals<SessionInternals>(
new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "test-client",
permissions: OWNER_PERMISSIONS,
onMessage,
@@ -281,14 +281,18 @@ export function createProviderSnapshotManagerStub(): {
};
}
export function createAgentRequestsStub(): SessionOptions["agentRequests"] {
export function createRequestReceiptsStub(): SessionOptions["requestReceipts"] {
return {
async create(input) {
async createAgent(input) {
const agentId = randomUUID();
await input.create(agentId);
return agentId;
},
send: (input) => input.send(),
async createWorkspace(input) {
await input.create(input.workspaceId);
return input.workspaceId;
},
sendMessage: (input) => input.send(),
};
}
@@ -1,4 +1,4 @@
import { AgentRequests } from "./agent/requests/index.js";
import { RequestReceipts } from "./request-receipts/index.js";
import { WebSocket, WebSocketServer } from "ws";
import type { IncomingMessage, Server as HTTPServer } from "http";
import { join } from "path";
@@ -529,7 +529,7 @@ export class VoiceAssistantWebSocketServer {
private readonly daemonRuntimeConfig: DaemonRuntimeConfig | undefined;
private readonly agentManager: AgentManager;
private readonly agentStorage: AgentStorage;
private readonly agentRequests: AgentRequests;
private readonly requestReceipts: RequestReceipts;
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly workspaceLabelService: WorkspaceLabelService | null;
@@ -649,7 +649,7 @@ export class VoiceAssistantWebSocketServer {
this.orchestrationSkills = orchestrationSkills;
this.agentManager = agentManager;
this.agentStorage = agentStorage;
this.agentRequests = new AgentRequests(join(paseoHome, "agent-requests"));
this.requestReceipts = new RequestReceipts(join(paseoHome, "agent-requests"));
this.projectRegistry = projectRegistry ?? createNoopProjectRegistry();
this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry();
this.workspaceLabelService = workspaceLabelService ?? null;
@@ -1425,7 +1425,7 @@ export class VoiceAssistantWebSocketServer {
worktreesRoot: this.worktreesRoot,
agentManager: this.agentManager,
agentStorage: this.agentStorage,
agentRequests: this.agentRequests,
requestReceipts: this.requestReceipts,
projectRegistry: this.projectRegistry,
workspaceRegistry: this.workspaceRegistry,
workspaceLabelService: this.workspaceLabelService ?? undefined,
@@ -1646,6 +1646,7 @@ export class VoiceAssistantWebSocketServer {
features: {
ownedSubscriptions: true,
agentRequestReceipts: true,
workspaceRequestReceipts: true,
hubAgentRpc: true,
// COMPAT(directorySync): added in v0.3.x, remove gate after 2027-02-12.
directorySync: true,
@@ -1,4 +1,4 @@
import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
import { createRequestReceiptsStub } from "./test-utils/session-stubs.js";
import pino from "pino";
import { z } from "zod";
import { describe, expect, test } from "vitest";
@@ -208,7 +208,7 @@ function createSessionForWireCompatTest(options?: {
];
const session = new Session({
agentRequests: createAgentRequestsStub(),
requestReceipts: createRequestReceiptsStub(),
clientId: "wire-compat-client",
permissions: OWNER_PERMISSIONS,
clientCapabilities: options?.clientCapabilities ?? null,