diff --git a/docs/architecture.md b/docs/architecture.md index 7febb9006a..6d21103cb6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -327,8 +327,11 @@ the app keeps its existing disk-ready navigation and observes the remaining crea Callbacks do not advance the workflow. Resource reservations in acknowledgement are identities, not ready workspace records. -The creation journal (`server/creation/`) keeps cumulative milestones and permanent IDs -under an operation kind and idempotency key. Reconnect subscribes to that key; retries +The creation journal (`server/creation/`) owns identity and execution for both legacy and +modern creation RPCs. Requesting progress never selects a different journal. Existing +agent receipts are imported at this boundary; message delivery receipts remain separate. +It keeps cumulative milestones and permanent IDs under an operation kind and idempotency +key. Reconnect subscribes to that key; retries join active work or reuse committed stages. A persisted resource alone cannot prove that a provider accepted its initial prompt. Interrupted side effects with no conclusive receipt return an unknown outcome instead of being repeated. diff --git a/packages/app/src/screens/new-workspace-screen.tsx b/packages/app/src/screens/new-workspace-screen.tsx index 86315e8208..6f05370927 100644 --- a/packages/app/src/screens/new-workspace-screen.tsx +++ b/packages/app/src/screens/new-workspace-screen.tsx @@ -797,6 +797,11 @@ interface WorkspaceDraftSubmissionConfig { target: WorkspaceTabTarget; } +interface WorkspaceCreationResult { + workspace: ReturnType; + agent?: AgentSnapshotPayload; +} + async function createMultiplicityWorkspace(input: { idempotencyKey: string; worktreeSlug: string; @@ -808,7 +813,6 @@ async function createMultiplicityWorkspace(input: { withInitialAgent: boolean; agent?: CreateWorkspaceRequestOptions["agent"]; onEvent?: (snapshot: CreationSnapshot) => void; - onAgentCreated?: (agent: AgentSnapshotPayload) => void; prompt: string; attachments: AgentAttachment[]; mergeWorkspaces: ( @@ -817,7 +821,7 @@ async function createMultiplicityWorkspace(input: { ) => void; serverId: string; createFailedMessage: string; -}): Promise> { +}): Promise { const projectId = getHostProjectId(input.project, input.serverId); if (!projectId) throw new Error("Project is not available on the selected host"); const isWorktree = input.isolation === "worktree"; @@ -847,13 +851,12 @@ async function createMultiplicityWorkspace(input: { if (payload.error || !payload.workspace) { throw new Error(payload.error ?? input.createFailedMessage); } - if (payload.agent) input.onAgentCreated?.(payload.agent); const normalizedWorkspace = normalizeWorkspaceDescriptor(payload.workspace); const workspaceForInitialMerge = input.withInitialAgent ? { ...normalizedWorkspace, status: "running" as const, statusEnteredAt: new Date() } : normalizedWorkspace; input.mergeWorkspaces(input.serverId, [workspaceForInitialMerge]); - return normalizedWorkspace; + return { workspace: normalizedWorkspace, agent: payload.agent }; } interface CreateChatAgentInput { @@ -867,8 +870,7 @@ interface CreateChatAgentInput { withInitialAgent: boolean; agent?: CreateWorkspaceRequestOptions["agent"]; onEvent?: (snapshot: CreationSnapshot) => void; - onAgentCreated?: (agent: AgentSnapshotPayload) => void; - }) => Promise>; + }) => Promise; serverId: string; draftKey: string; clearDraft: (lifecycle: "sent" | "abandoned") => void; @@ -965,14 +967,6 @@ async function createWorkspaceChatAgent(input: CreateChatAgentInput): Promise void; - let rejectAgent!: (error: unknown) => void; - const result = new Promise((resolve, reject) => { - resolveAgent = resolve; - rejectAgent = reject; - }); - // The destination may mount after creation finishes or fails. - void result.catch(() => undefined); let navigated = false; let outcome: SubmitOutcome = "background"; const initialAgent: NonNullable = { @@ -990,16 +984,12 @@ async function createWorkspaceChatAgent(input: CreateChatAgentInput): Promise => { - let createdAgent: AgentSnapshotPayload | undefined; - await ensureWorkspace({ + const { agent } = await ensureWorkspace({ cwd, prompt: text, attachments: workspaceNamingAttachments, withInitialAgent: true, agent: requestedAgent, - onAgentCreated: (agent) => { - createdAgent = agent; - }, onEvent: (snapshot) => { if (!snapshot.workspace || navigated) return; navigated = true; @@ -1035,11 +1025,11 @@ async function createWorkspaceChatAgent(input: CreateChatAgentInput): Promise execute()), retry: (request: CreateAgentRequestOptions) => execute({ ...initialAgent, @@ -1050,14 +1040,9 @@ async function createWorkspaceChatAgent(input: CreateChatAgentInput): Promise(null); - const [createdWorkspace, setCreatedWorkspace] = useState | null>(null); + const [creationResult, setCreationResult] = useState< + WorkspaceCreationResult | { workspace: null } + >({ workspace: null }); const [pendingAction, setPendingAction] = useState<"chat" | "empty" | "terminal" | null>(null); const [pickerOpen, setPickerOpen] = useState(false); const [projectPickerOpen, setProjectPickerOpen] = useState(false); @@ -1740,7 +1725,7 @@ export function NewWorkspaceScreen({ return () => clearTimeout(timer); }, [pickerSearchQuery]); - const workspace = createdWorkspace; + const { workspace } = creationResult; const client = useHostRuntimeClient(selectedServerId); const isConnected = useHostRuntimeIsConnected(selectedServerId); const { @@ -2049,10 +2034,9 @@ export function NewWorkspaceScreen({ withInitialAgent: boolean; agent?: CreateWorkspaceRequestOptions["agent"]; onEvent?: (snapshot: CreationSnapshot) => void; - onAgentCreated?: (agent: AgentSnapshotPayload) => void; }) => { - if (createdWorkspace) { - return createdWorkspace; + if (creationResult.workspace) { + return creationResult; } if (!selectedProject) { throw new Error("Choose a project"); @@ -2088,17 +2072,16 @@ export function NewWorkspaceScreen({ attachments: input.attachments, agent: input.agent, onEvent: input.onEvent, - onAgentCreated: input.onAgentCreated, mergeWorkspaces, serverId: selectedServerId, createFailedMessage: t("newWorkspace.errors.createWorktreeFailed"), }); - setCreatedWorkspace(normalizedWorkspace); + setCreationResult(normalizedWorkspace); return normalizedWorkspace; }, [ creationIdentity, - createdWorkspace, + creationResult, effectiveIsolation, mergeWorkspaces, queryClient, @@ -2123,7 +2106,7 @@ export function NewWorkspaceScreen({ let outcome: SubmitOutcome = "background"; await runCreateEmptyWorkspace({ payload, - ensureWorkspace, + ensureWorkspace: async (request) => (await ensureWorkspace(request)).workspace, serverId: selectedServerId, navigate: (targetServerId, workspaceId) => { if (!isStillOnCreateScreen()) { @@ -2200,7 +2183,7 @@ export function NewWorkspaceScreen({ prompt: terminalPromptText, profile: selectedTerminalProfile, profileName: selectedTerminalProfile?.name, - ensureWorkspace, + ensureWorkspace: async (request) => (await ensureWorkspace(request)).workspace, createTerminal: async (input) => { const connectedClient = withConnectedClient(); const createdTerminal = await connectedClient.createTerminal( diff --git a/packages/server/src/server/creation/creation.e2e.test.ts b/packages/server/src/server/creation/creation.e2e.test.ts index 0d6e55547c..f0d21be18f 100644 --- a/packages/server/src/server/creation/creation.e2e.test.ts +++ b/packages/server/src/server/creation/creation.e2e.test.ts @@ -1,3 +1,7 @@ +import { execFileSync } from "node:child_process"; +import type { z } from "zod"; +import { WebSocket } from "ws"; +import { SessionInboundMessageSchema, WSOutboundMessageSchema } from "@getpaseo/protocol/messages"; import { randomUUID } from "node:crypto"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; @@ -121,3 +125,162 @@ test("creation progresses before agent readiness and continues after the disconn await rm(directory, { recursive: true, force: true }); } }, 60000); + +async function connectCreationPeer(port: number) { + const socket = new WebSocket(`ws://127.0.0.1:${port}/ws`); + const frames: SessionOutboundMessage[] = []; + socket.on("message", (data) => { + const frame = WSOutboundMessageSchema.parse(JSON.parse(data.toString())); + if (frame.type === "session") frames.push(frame.message); + }); + await new Promise((resolve, reject) => { + socket.once("open", resolve); + socket.once("error", reject); + }); + socket.send( + JSON.stringify({ + type: "hello", + clientType: "browser", + clientId: randomUUID(), + protocolVersion: 1, + appVersion: "0.8.0", + }), + ); + await expect + .poll(() => frames.some((m) => m.type === "status" && m.payload.status === "server_info")) + .toBe(true); + return { + close: () => socket.close(), + request: async ( + message: z.input & { requestId: string }, + ) => { + socket.send(JSON.stringify({ type: "session", message })); + const response = () => + frames.find( + (m) => + "payload" in m && "requestId" in m.payload && m.payload.requestId === message.requestId, + ); + await expect.poll(response).toBeDefined(); + return response()!; + }, + }; +} + +test.each([false, true])( + "workspace identity does not depend on subscribing (first subscribe=%s)", + async (subscribe) => { + const directory = await mkdtemp(join(tmpdir(), "creation-identity-")); + const daemon = await createTestPaseoDaemon(); + const peer = await connectCreationPeer(daemon.port); + try { + const request = { + type: "workspace.create.request" as const, + source: { kind: "directory" as const, path: directory }, + idempotencyKey: "same-workspace", + }; + const first = await peer.request({ ...request, requestId: "first", subscribe }); + const replay = await peer.request({ ...request, requestId: "replay", subscribe: !subscribe }); + if (first.type !== "workspace.create.response" || replay.type !== "workspace.create.response") + throw new Error("Expected workspace responses"); + expect(first.payload.error).toBeNull(); + expect(replay.payload.error).toBeNull(); + expect(replay.payload.workspace?.id).toBe(first.payload.workspace?.id); + } finally { + peer.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true }); + } + }, + 60000, +); + +test.each(["create_agent_request", "agent.create.request"] as const)( + "legacy and modern agent RPCs share one creation identity (first %s)", + async (type) => { + const directory = await mkdtemp(join(tmpdir(), "creation-agent-identity-")); + let creations = 0; + const daemon = await createTestPaseoDaemon({ + agentClients: createTestAgentClients({ + beforeCreateSession: async () => { + creations++; + }, + }), + }); + const peer = await connectCreationPeer(daemon.port); + try { + const request = { + config: { provider: "codex", cwd: directory }, + idempotencyKey: "same-agent", + }; + const first = await peer.request({ ...request, type, requestId: "first" }); + const replay = await peer.request({ + ...request, + type: type === "agent.create.request" ? "create_agent_request" : "agent.create.request", + requestId: "replay", + }); + const createdAgent = (message: SessionOutboundMessage) => { + if (message.type === "agent.create.response") { + expect(message.payload.error).toBeNull(); + return message.payload.agent; + } + if (message.type === "status" && message.payload.status === "agent_created") + return message.payload.agent; + throw new Error(`Unexpected creation result ${JSON.stringify(message)}`); + }; + expect(createdAgent(replay)?.id).toBe(createdAgent(first)?.id); + const agentId = createdAgent(first)!.id; + await daemon.daemon.agentManager.setTitle(agentId, "Updated after creation"); + await daemon.daemon.agentManager.flush(); + const legacyReplay = await peer.request({ + ...request, + type: "create_agent_request", + requestId: "legacy-after-update", + }); + expect(createdAgent(legacyReplay)?.title).toBe("Updated after creation"); + expect(creations).toBe(1); + } finally { + peer.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true }); + } + }, + 60000, +); + +test("legacy keyed creation preserves checkout error codes", async () => { + const directory = await mkdtemp(join(tmpdir(), "creation-checkout-error-")); + execFileSync("git", ["init", "-b", "main", directory], { stdio: "pipe" }); + execFileSync( + "git", + [ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.com", + "commit", + "--allow-empty", + "-m", + "initial", + ], + { cwd: directory, stdio: "pipe" }, + ); + const daemon = await createTestPaseoDaemon({ agentClients: createTestAgentClients() }); + const peer = await connectCreationPeer(daemon.port); + try { + const result = await peer.request({ + type: "create_agent_request", + requestId: "missing-branch", + idempotencyKey: "missing-branch", + config: { provider: "codex", cwd: directory }, + worktree: { mode: "checkout-branch", branch: "does-not-exist" }, + }); + expect(result).toMatchObject({ + type: "status", + payload: { status: "agent_create_failed", errorCode: "unknown_branch" }, + }); + } finally { + peer.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true }); + } +}, 60000); diff --git a/packages/server/src/server/creation/index.test.ts b/packages/server/src/server/creation/index.test.ts index 3762e38923..16a151a289 100644 --- a/packages/server/src/server/creation/index.test.ts +++ b/packages/server/src/server/creation/index.test.ts @@ -1,4 +1,5 @@ -import { cp, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { createHash } from "node:crypto"; +import { cp, mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, expect, test } from "vitest"; @@ -249,3 +250,44 @@ test("partial provisioning without a registered workspace is not repeated", asyn expect(await new CreationService(f.directory).create(f.input)).toEqual(result); expect(provisions).toBe(1); }); + +test.each(["pending", "completed"])( + "imports a %s legacy agent receipt without creating another agent", + async (state) => { + const f = await fixture(); + const legacyDirectory = join(f.directory, "legacy"); + await mkdir(legacyDirectory); + const request = { config: { cwd: "/project", provider: "codex" } }; + const hash = (value: unknown) => + createHash("sha256").update(JSON.stringify(value)).digest("hex"); + await writeFile( + join(legacyDirectory, `${hash(["create", "old-key"])}.json`), + JSON.stringify({ + fingerprint: hash({ ...request, type: "create_agent_request" }), + state, + agentId: agent.id, + }), + ); + const input: CreationInput = { + kind: "agent", + key: "old-key", + request, + hasAgent: true, + hasPrompt: false, + exists: async () => true, + readAgent: async () => agent, + createAgent: async () => { + throw new Error("Must not create again"); + }, + }; + const service = new CreationService(f.directory, undefined, legacyDirectory); + await expect( + service.create({ ...input, request: { config: { ...request.config, cwd: "/changed" } } }), + ).rejects.toThrow("agent_request_key_conflict"); + expect(await service.create(input)).toMatchObject({ phase: "completed", agent }); + expect(await new CreationService(f.directory).create(input)).toMatchObject({ + phase: "completed", + agent, + }); + }, +); diff --git a/packages/server/src/server/creation/index.ts b/packages/server/src/server/creation/index.ts index 73e348af62..00bb664291 100644 --- a/packages/server/src/server/creation/index.ts +++ b/packages/server/src/server/creation/index.ts @@ -12,10 +12,9 @@ import { writeJsonFileAtomic } from "../atomic-file.js"; import { generateWorkspaceId } from "../workspace-registry-model.js"; type Observer = (snapshot: CreationSnapshot) => void; -export interface CreationInput { - kind: "workspace" | "agent"; +interface CreationRequest { key: string; - request: unknown; + request: { [key: string]: unknown }; workspaceId?: string; agentId?: string; hasAgent: boolean; @@ -30,6 +29,11 @@ export interface CreationInput { onReady: (agent: AgentSnapshotPayload) => Promise, ) => Promise; } +export type CreationInput = CreationRequest & + ( + | { kind: "workspace" } + | { kind: "agent"; readAgent: (id: string) => Promise } + ); const RecordSchema = z.object({ fingerprint: z.string(), snapshot: CreationSnapshotSchema, @@ -47,14 +51,16 @@ export class CreationService { private readonly validateCompleted: ( snapshot: CreationSnapshot, ) => Promise = async () => {}, + private readonly legacyDirectory?: string, ) {} async create(input: CreationInput, observer?: Observer): Promise { const identity = identityFor(input.kind, input.key); const admitted = this.admission.then(async () => { await mkdir(this.directory, { recursive: true }); - let record = await this.read(identity); const fingerprint = digest(input.request); + let record = await this.read(identity); + if (!record) record = await this.readLegacyAgent(input, fingerprint); if (record && record.fingerprint !== fingerprint) throw new Error(`${input.kind}_request_key_conflict`); const running = this.active.get(identity); @@ -204,6 +210,39 @@ export class CreationService { return record.snapshot; } + // COMPAT(agentRequestReceipts): added in v0.8.0, remove after 2027-03-11 once old creation receipts can expire. + // Import the previous receipt format; it is never a second execution path. + private async readLegacyAgent(input: CreationInput, fingerprint: string): Promise { + if (input.kind !== "agent" || !this.legacyDirectory) return null; + const text = await readOptional( + join(this.legacyDirectory, `${digest(["create", input.key])}.json`), + ); + if (text === null) return null; + const receipt = z + .object({ + fingerprint: z.string(), + state: z.enum(["pending", "completed"]), + agentId: z.string(), + }) + .parse(JSON.parse(text)); + if (receipt.fingerprint !== digest({ type: "create_agent_request", ...input.request })) + throw new Error("agent_request_key_conflict"); + const agent = await input.readAgent(receipt.agentId); + if (!agent && receipt.state === "completed") + throw new Error("Previously created agent no longer exists"); + const record = initialRecord({ ...input, agentId: receipt.agentId }, fingerprint); + record.snapshot = { + ...record.snapshot, + phase: agent ? "completed" : "failed", + agent: agent ?? undefined, + workspaceId: agent?.workspaceId ?? record.snapshot.workspaceId, + error: agent ? null : "agent_request_outcome_unknown", + ...(agent ? {} : { failedStage: "agent", outcomeUnknown: true }), + }; + await this.write(identityFor(input.kind, input.key), record); + return record; + } + private async claimResources( identity: string, snapshot: CreationSnapshot, diff --git a/packages/server/src/server/message-receipts/index.test.ts b/packages/server/src/server/message-receipts/index.test.ts new file mode 100644 index 0000000000..10c1615a6a --- /dev/null +++ b/packages/server/src/server/message-receipts/index.test.ts @@ -0,0 +1,77 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, expect, test } from "vitest"; +import { MessageReceipts } from "./index.js"; + +const directories: string[] = []; +afterEach(async () => { + await Promise.all( + directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })), + ); +}); +async function fixture() { + const directory = await mkdtemp(path.join(tmpdir(), "agent-requests-")); + directories.push(directory); + return { directory, requests: new MessageReceipts(directory) }; +} + +test("message retries survive reconstruction without submitting twice", async () => { + const { requests, directory } = await fixture(); + let deliveries = 0; + const input = { + agentId: "agent", + messageId: "arrival", + request: { text: "hello" }, + send: async () => { + deliveries++; + }, + }; + await Promise.all([requests.send(input), requests.send(input)]); + await new MessageReceipts(directory).send(input); + expect(deliveries).toBe(1); + await requests.send({ ...input, agentId: "another" }); + expect(deliveries).toBe(2); +}); + +test("ambiguous provider delivery is never blindly replayed after restart", async () => { + const { requests, directory } = await fixture(); + let deliveries = 0; + const input = { + agentId: "agent", + messageId: "arrival", + request: {}, + send: async () => { + deliveries++; + throw new Error("connection lost"); + }, + }; + await expect(requests.send(input)).rejects.toThrow("connection lost"); + await expect(new MessageReceipts(directory).send(input)).rejects.toThrow( + "agent_request_outcome_unknown", + ); + expect(deliveries).toBe(1); +}); + +test("failed local message preparation does not leave an ambiguous receipt", async () => { + const { requests, directory } = await fixture(); + let available = false; + let sends = 0; + const input = { + agentId: "agent", + messageId: "message", + request: {}, + prepare: async () => { + if (!available) throw new Error("load failed"); + }, + send: async () => { + sends++; + }, + }; + await expect(requests.send(input)).rejects.toThrow("load failed"); + available = true; + await new MessageReceipts(directory).send(input); + available = false; + await requests.send(input); + expect(sends).toBe(1); +}); diff --git a/packages/server/src/server/message-receipts/index.ts b/packages/server/src/server/message-receipts/index.ts new file mode 100644 index 0000000000..6103160b12 --- /dev/null +++ b/packages/server/src/server/message-receipts/index.ts @@ -0,0 +1,81 @@ +import { createHash } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { z } from "zod"; +import { writeJsonFileAtomic } from "../atomic-file.js"; + +const ReceiptSchema = z.object({ + fingerprint: z.string(), + state: z.enum(["pending", "completed"]), + agentId: z.string(), +}); +interface SendMessageInput { + agentId: string; + messageId: string; + request: unknown; + send: () => Promise; + prepare?: () => Promise; +} + +/** Owns message delivery receipts; creation is owned by CreationService. */ +export class MessageReceipts { + private readonly pending = new Map>(); + constructor(private readonly directory: string) {} + + send(input: SendMessageInput): Promise { + // Preserve the existing on-disk identity and shape across daemon upgrades. + const key = digest(["send", input.agentId, input.messageId]); + const previous = this.pending.get(key); + const result = (previous ? previous.catch(() => undefined) : Promise.resolve()).then(() => + this.sendOnce(key, input), + ); + this.pending.set(key, result); + void result + .finally(() => { + if (this.pending.get(key) === result) this.pending.delete(key); + }) + .catch(() => undefined); + return result; + } + + private async sendOnce(key: string, input: SendMessageInput): Promise { + const file = path.join(this.directory, `${key}.json`); + const fingerprint = digest(input.request); + const existing = await readReceipt(file); + if (existing) { + if (existing.fingerprint !== fingerprint) throw new Error("agent_request_key_conflict"); + if (existing.state === "completed") return; + // A provider may have accepted the message before its receipt was committed. + throw new Error("agent_request_outcome_unknown"); + } + await input.prepare?.(); + const receipt = { fingerprint, agentId: input.agentId }; + await writeJsonFileAtomic(file, { ...receipt, state: "pending" }); + await input.send(); + await writeJsonFileAtomic(file, { ...receipt, state: "completed" }); + } +} + +async function readReceipt(file: string): Promise | null> { + try { + return ReceiptSchema.parse(JSON.parse(await readFile(file, "utf8"))); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") return null; + throw error; + } +} + +function digest(value: unknown): string { + return createHash("sha256") + .update( + JSON.stringify(value, (_key, candidate: unknown) => { + if (candidate !== null && typeof candidate === "object" && !Array.isArray(candidate)) { + return Object.fromEntries( + Object.entries(candidate).sort(([a], [b]) => a.localeCompare(b)), + ); + } + return candidate; + }), + ) + .digest("hex"); +} diff --git a/packages/server/src/server/request-receipts/index.test.ts b/packages/server/src/server/request-receipts/index.test.ts deleted file mode 100644 index 4a540e2bdc..0000000000 --- a/packages/server/src/server/request-receipts/index.test.ts +++ /dev/null @@ -1,201 +0,0 @@ -import { mkdtemp, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import path from "node:path"; -import { afterEach, expect, test } from "vitest"; -import { RequestReceipts } from "./index.js"; - -const directories: string[] = []; -afterEach(async () => { - await Promise.all( - directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })), - ); -}); -async function fixture() { - const directory = await mkdtemp(path.join(tmpdir(), "agent-requests-")); - directories.push(directory); - return { directory, requests: new RequestReceipts(directory) }; -} - -test("concurrent and reconstructed creates return the same durable agent", async () => { - const { requests, directory } = await fixture(); - const agents = new Set(); - let creations = 0; - const input = { - key: "create-1", - request: { provider: "test", cwd: "/project" }, - findAgent: async (id: string) => agents.has(id), - create: async (id: string) => { - creations++; - agents.add(id); - }, - }; - const ids = await Promise.all([requests.createAgent(input), requests.createAgent(input)]); - expect(ids[0]).toBe(ids[1]); - expect(await new RequestReceipts(directory).createAgent(input)).toBe(ids[0]); - expect(creations).toBe(1); -}); - -test("recovers creation when the agent was persisted before acknowledgement failed", async () => { - const { requests, directory } = await fixture(); - const agents = new Set(); - const input = { - key: "lost-create", - request: {}, - findAgent: async (id: string) => agents.has(id), - create: async (id: string) => { - agents.add(id); - throw new Error("acknowledgement lost"); - }, - }; - await expect(requests.createAgent(input)).rejects.toThrow("acknowledgement lost"); - expect(await new RequestReceipts(directory).createAgent(input)).toBe([...agents][0]); - expect(agents.size).toBe(1); -}); - -test("reusing a create key with different configuration is a conflict", async () => { - const { requests } = await fixture(); - const input = { - key: "key", - request: { model: "a" }, - findAgent: async () => true, - create: async () => {}, - }; - await requests.createAgent(input); - await expect(requests.createAgent({ ...input, request: { model: "b" } })).rejects.toThrow( - "agent_request_key_conflict", - ); -}); - -test("message retries survive reconstruction without submitting twice", async () => { - const { requests, directory } = await fixture(); - let deliveries = 0; - const input = { - agentId: "agent", - messageId: "arrival", - request: { text: "hello" }, - send: async () => { - deliveries++; - }, - }; - await Promise.all([requests.sendMessage(input), requests.sendMessage(input)]); - await new RequestReceipts(directory).sendMessage(input); - expect(deliveries).toBe(1); - await requests.sendMessage({ ...input, agentId: "another" }); - expect(deliveries).toBe(2); -}); - -test("ambiguous provider delivery is never blindly replayed after restart", async () => { - const { requests, directory } = await fixture(); - let deliveries = 0; - const input = { - agentId: "agent", - messageId: "arrival", - request: {}, - send: async () => { - deliveries++; - throw new Error("connection lost"); - }, - }; - 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); -}); - -test("a creation failure with no stored agent can be retried", async () => { - const { requests } = await fixture(); - let available = false; - const input = { - key: "unavailable", - request: {}, - findAgent: async () => false, - create: async () => { - if (!available) throw new Error("provider unavailable"); - }, - }; - await expect(requests.createAgent(input)).rejects.toThrow("provider unavailable"); - available = true; - await expect(requests.createAgent(input)).resolves.toEqual(expect.any(String)); -}); - -test("failed local message preparation does not leave an ambiguous receipt", async () => { - const { requests, directory } = await fixture(); - let available = false; - let sends = 0; - const input = { - agentId: "agent", - messageId: "message", - request: {}, - prepare: async () => { - if (!available) throw new Error("load failed"); - }, - send: async () => { - sends++; - }, - }; - await expect(requests.sendMessage(input)).rejects.toThrow("load failed"); - available = true; - await new RequestReceipts(directory).sendMessage(input); - available = false; - 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(); - 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 record left by failed provisioning is not mistaken for completion", async () => { - const { requests, directory } = await fixture(); - const workspaces = new Set(); - const input = { - key: "partial-workspace", - workspaceId: "wks_created", - request: {}, - findWorkspace: async (id: string) => workspaces.has(id), - create: async (id: string) => { - workspaces.add(id); - throw new Error("checkout rolled back"); - }, - }; - await expect(requests.createWorkspace(input)).rejects.toThrow("checkout rolled back"); - await expect(new RequestReceipts(directory).createWorkspace(input)).rejects.toThrow( - "workspace_request_outcome_unknown", - ); - expect([...workspaces]).toEqual(["wks_created"]); -}); diff --git a/packages/server/src/server/request-receipts/index.ts b/packages/server/src/server/request-receipts/index.ts deleted file mode 100644 index 4f457710c9..0000000000 --- a/packages/server/src/server/request-receipts/index.ts +++ /dev/null @@ -1,186 +0,0 @@ -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"; - -const ReceiptStateSchema = z.object({ - fingerprint: z.string(), - state: z.enum(["pending", "completed"]), -}); -// 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 & { - resource: { kind: "agent" | "workspace"; id: string }; -}; - -interface CreateAgentInput { - key: string; - request: unknown; - findAgent: (agentId: string) => Promise; - create: (agentId: string) => Promise; -} - -interface CreateWorkspaceInput { - key: string; - request: unknown; - workspaceId: string; - findWorkspace: (workspaceId: string) => Promise; - create: (workspaceId: string) => Promise; -} - -interface SendMessageInput { - agentId: string; - messageId: string; - request: unknown; - send: () => Promise; - prepare?: () => Promise; -} - -/** One daemon-owned request journal, shared by all of its socket sessions. */ -export class RequestReceipts { - private readonly pending = new Map>(); - - constructor(private readonly directory: string) {} - - createAgent(input: CreateAgentInput): Promise { - return this.execute(["create", input.key], input.request, { - resource: { kind: "agent", id: randomUUID() }, - recover: input.findAgent, - run: input.create, - retrySafe: async (agentId) => !(await input.findAgent(agentId)), - }); - } - - createWorkspace(input: CreateWorkspaceInput): Promise { - return this.execute(["create-workspace", input.key], input.request, { - resource: { kind: "workspace", id: input.workspaceId }, - // A registry entry can survive failed provisioning and checkout rollback. - // Only a completed receipt proves that the entire operation succeeded. - recover: async () => false, - run: input.create, - retrySafe: async (workspaceId) => !(await input.findWorkspace(workspaceId)), - }); - } - - async sendMessage(input: SendMessageInput): Promise { - await this.execute(["send", input.agentId, input.messageId], input.request, { - 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, - run: input.send, - prepare: input.prepare, - }); - } - - private execute( - identity: string[], - request: unknown, - operation: { - resource: Receipt["resource"]; - recover: (agentId: string) => Promise; - run: (agentId: string) => Promise; - prepare?: (() => Promise) | undefined; - retrySafe?: (agentId: string) => Promise; - }, - ): Promise { - const key = digest(identity); - const fingerprint = digest(request); - const previous = this.pending.get(key); - // Serialize even conflicting requests: each caller validates its own fingerprint. - const result = (previous ? previous.catch(() => undefined) : Promise.resolve()).then(() => - this.executeOnce(key, fingerprint, operation), - ); - this.pending.set(key, result); - void result - .finally(() => { - if (this.pending.get(key) === result) this.pending.delete(key); - }) - .catch(() => undefined); - return result; - } - - private async executeOnce( - key: string, - fingerprint: string, - operation: { - resource: Receipt["resource"]; - recover: (agentId: string) => Promise; - run: (agentId: string) => Promise; - prepare?: (() => Promise) | undefined; - retrySafe?: (agentId: string) => Promise; - }, - ): Promise { - const file = path.join(this.directory, `${key}.json`); - const existing = await readReceipt(file); - if (existing) { - 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 writeReceipt(file, { ...existing, state: "completed" }); - return existing.resource.id; - } - await operation.prepare?.(); - const receipt: Receipt = { fingerprint, resource: operation.resource, state: "pending" }; - await writeReceipt(file, receipt); - try { - await operation.run(receipt.resource.id); - } catch (error) { - // 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 writeReceipt(file, { ...receipt, state: "completed" }); - return receipt.resource.id; - } -} - -async function readReceipt(file: string): Promise { - try { - 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 { - 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( - JSON.stringify(value, (_key, candidate: unknown) => { - if (candidate !== null && typeof candidate === "object" && !Array.isArray(candidate)) { - return Object.fromEntries( - Object.entries(candidate).sort(([a], [b]) => a.localeCompare(b)), - ); - } - return candidate; - }), - ) - .digest("hex"); -} diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 9c77dceebf..7a6384f9a6 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -1,6 +1,6 @@ import { - createRequestReceiptsStub, - createCreationServiceStub, + createMessageReceiptsStub, + createTestCreationService, } from "./test-utils/session-stubs.js"; import { execSync } from "child_process"; import { existsSync, mkdtempSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "fs"; @@ -367,8 +367,8 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { const messages = options.messages ?? []; const sessionOptions: SessionOptions = { - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: options.clientId ?? "test-client", onMessage: (message) => messages.push(message), ...(options.targetedMessages diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 6a41036e65..3e63d3c9f3 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -5,7 +5,7 @@ import { relative } from "node:path"; import { isAbsolute } from "node:path"; import { CreationService } from "./creation/index.js"; import type { CreationSnapshot, AgentCreateRequest } from "@getpaseo/protocol/messages"; -import type { RequestReceipts } from "./request-receipts/index.js"; +import type { MessageReceipts } from "./message-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"; @@ -151,7 +151,6 @@ import { checkoutLiteFromGitSnapshot, checkoutFromPersistedWorkspacePlacement, deriveWorkspaceDisplayName, - generateWorkspaceId, } from "./workspace-registry-model.js"; import { resolveWorkspaceIdForPath } from "./resolve-workspace-id-for-path.js"; import { @@ -206,7 +205,6 @@ import { } from "./session/git-mutation/git-mutation-service.js"; import { createWorkspaceProvisioningService, - WorkspaceProvisioningError, type WorkspaceProvisioningService, } from "./session/workspace-provisioning/workspace-provisioning-service.js"; import { @@ -463,7 +461,7 @@ export interface SessionOptions { worktreesRoot?: string; agentManager: AgentManager; agentStorage: AgentStorage; - requestReceipts: Pick; + messageReceipts: Pick; creationService: Pick; projectRegistry: ProjectRegistry; workspaceRegistry: WorkspaceRegistry; @@ -796,10 +794,7 @@ export class Session { private readonly daemonSession: DaemonSession; private readonly hubExecutionController: HubExecutionController | null; private readonly workspaceScripts: WorkspaceScriptsService; - private readonly requestReceipts: Pick< - RequestReceipts, - "createAgent" | "createWorkspace" | "sendMessage" - >; + private readonly messageReceipts: Pick; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; private readonly creationService: Pick; @@ -875,7 +870,7 @@ export class Session { this.onWorkspaceRecovered = onWorkspaceRecovered ?? null; this.pushNotifications = pushNotifications; this.paseoHome = paseoHome; - this.requestReceipts = options.requestReceipts; + this.messageReceipts = options.messageReceipts; this.creationService = options.creationService; this.projectIcons = new ProjectIconReader(paseoHome); this.worktreesRoot = worktreesRoot; @@ -2667,9 +2662,7 @@ export class Session { case "creation.subscribe.request": return this.handleCreationSubscription(msg, source); case "workspace.create.request": - if (msg.subscribe || msg.agent || msg.workspaceId) - return this.handleWorkspaceCreation(msg, source); - return undefined; + return this.handleWorkspaceCreation(msg, source); default: return undefined; } @@ -2872,8 +2865,6 @@ export class Session { return this.handleArchiveWorkspaceRequest(msg); case "project.remove.request": return this.handleProjectRemoveRequest(msg); - case "workspace.create.request": - return this.handleWorkspaceCreateRequest(msg); case "workspace.title.set.request": return this.handleWorkspaceTitleSetRequest(msg.workspaceId, msg.title, msg.requestId); case "workspace.pin.set.request": @@ -3923,8 +3914,43 @@ export class Session { ? (await this.workspaceRegistry.get(id)) !== null : this.agentManager.getAgent(id) != null || (await this.agentStorage.get(id)) !== null; + private createRequestedAgent( + request: AgentCreateRequest, + observer?: (snapshot: CreationSnapshot) => void, + ): Promise { + const { requestId, type: _type, subscribe: _subscribe, idempotencyKey, ...intent } = request; + return this.creationService.create( + { + kind: "agent", + key: idempotencyKey ?? requestId, + request: intent, + workspaceId: request.workspaceId, + agentId: request.agentId, + hasAgent: true, + hasPrompt: hasInitialCreationPrompt(request), + exists: this.creationResourceExists, + readAgent: async (id) => { + const record = await this.agentStorage.get(id); + return record ? this.buildStoredAgentPayload(record) : null; + }, + createAgent: async (id, _workspace, onReady) => { + try { + return await this.createSessionAgent( + { ...request, type: "create_agent_request" }, + id, + onReady, + ); + } catch (error) { + throw new WorktreeRequestError(toWorktreeWireError(error)); + } + }, + }, + observer, + ); + } + private async handleAgentCreation(request: AgentCreateRequest, source?: object): Promise { - const { requestId, subscribe, type: _type, idempotencyKey, ...intent } = request; + const { requestId, subscribe } = request; const progress = subscribe ? this.delivery.operation( (message) => @@ -3933,19 +3959,8 @@ export class Session { ) : undefined; try { - const creation = await this.creationService.create( - { - kind: "agent", - key: idempotencyKey ?? requestId, - request: intent, - workspaceId: request.workspaceId, - agentId: request.agentId, - hasAgent: true, - hasPrompt: hasInitialCreationPrompt(request), - exists: this.creationResourceExists, - createAgent: (id, _workspace, onReady) => - this.createSessionAgent({ ...request, type: "create_agent_request" }, id, onReady), - }, + const creation = await this.createRequestedAgent( + request, progress ? (snapshot) => progress.emit(this.creationUpdate(snapshot)) : undefined, ); this.emitForSource( @@ -4100,18 +4115,13 @@ export class Session { if (msg.initialPrompt !== undefined) { throw new Error("Idempotent creation requires sending the initial prompt separately"); } - const { requestId: _requestId, idempotencyKey, ...request } = msg; - const id = await this.requestReceipts.createAgent({ - key: idempotencyKey, - request, - findAgent: async (agentId) => - this.agentManager.getAgent(agentId) != null || - (await this.agentStorage.get(agentId)) !== null, - create: async (agentId) => { - await this.createSessionAgent(msg, agentId); - }, - }); - const record = await this.agentStorage.get(id); + const creation = await this.createRequestedAgent({ ...msg, type: "agent.create.request" }); + if (creation.error || !creation.agent) + throw new SessionRequestError( + creation.errorCode ?? "unknown", + creation.error ?? "Agent creation failed", + ); + const record = await this.agentStorage.get(creation.agent.id); if (!record) throw new Error("Previously created agent no longer exists"); agent = this.buildStoredAgentPayload(record); } else { @@ -4127,7 +4137,7 @@ export class Session { }, }); } catch (error) { - const wireError = toWorktreeWireError(error); + const wireError = error instanceof SessionRequestError ? error : toWorktreeWireError(error); this.sessionLogger.error({ err: error }, "Failed to create agent"); this.emit({ type: "status", @@ -6622,78 +6632,6 @@ export class Session { } } - private async handleWorkspaceCreateRequest( - request: Extract, - ): Promise { - try { - 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 this.createRequestedWorkspace(request, id); - }, - }); - if (!descriptor) { - const workspace = await this.workspaceRegistry.get(workspaceId); - if (!workspace) throw new Error("Previously created workspace no longer exists"); - if (!(await this.filesystem.isDirectory(workspace.cwd))) { - throw new SessionRequestError( - "directory_not_found", - `Directory not found: ${workspace.cwd}`, - ); - } - await this.syncWorkspaceGitObserverForWorkspace(workspace); - descriptor = await this.describeWorkspaceRecordWithGitData(workspace); - await this.emitCreatedWorkspaceUpdate(descriptor); - } - } else { - descriptor = await this.createRequestedWorkspace(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 instanceof SessionRequestError - ? error.code - : undefined; - this.emit({ - type: "workspace.create.response", - payload: { - requestId: request.requestId, - workspace: null, - setupTerminalId: null, - error: message, - errorCode, - }, - }); - } - } - private async createRequestedWorkspace( request: Extract, workspaceId?: string, @@ -8198,7 +8136,7 @@ export class Session { } }; if (msg.messageId) { - await this.requestReceipts.sendMessage({ + await this.messageReceipts.send({ agentId, messageId: msg.messageId, request: { prompt, activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt" }, diff --git a/packages/server/src/server/session.workspace-git-watch.test.ts b/packages/server/src/server/session.workspace-git-watch.test.ts index 3d33904374..836703fd30 100644 --- a/packages/server/src/server/session.workspace-git-watch.test.ts +++ b/packages/server/src/server/session.workspace-git-watch.test.ts @@ -1,6 +1,6 @@ import { - createRequestReceiptsStub, - createCreationServiceStub, + createMessageReceiptsStub, + createTestCreationService, } from "./test-utils/session-stubs.js"; import { describe, expect, test, vi } from "vitest"; import path from "node:path"; @@ -188,8 +188,8 @@ function createSessionForWorkspaceGitWatchTests(options?: { }; const session = new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message as { type: string; payload: unknown }), diff --git a/packages/server/src/server/session.workspace-resolution-invariants.test.ts b/packages/server/src/server/session.workspace-resolution-invariants.test.ts index 207e07956a..e848e46f01 100644 --- a/packages/server/src/server/session.workspace-resolution-invariants.test.ts +++ b/packages/server/src/server/session.workspace-resolution-invariants.test.ts @@ -1,6 +1,6 @@ import { - createRequestReceiptsStub, - createCreationServiceStub, + createMessageReceiptsStub, + createTestCreationService, } 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`. @@ -92,8 +92,8 @@ function createHarness(input: { }; const session = new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test", permissions: OWNER_PERMISSIONS, appVersion: null, diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index f2c63ccca3..2866732abe 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -1,6 +1,6 @@ import { - createRequestReceiptsStub, - createCreationServiceStub, + createMessageReceiptsStub, + createTestCreationService, } from "./test-utils/session-stubs.js"; import { execFileSync } from "node:child_process"; import { @@ -645,8 +645,8 @@ function createSessionForWorkspaceTests( const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, appVersion: options.appVersion ?? null, @@ -1007,8 +1007,8 @@ test("create_agent_request keeps requested child cwd when grouped under an exist const emitted: SessionOutboundMessage[] = []; const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", serverId: "test-server", permissions: OWNER_PERMISSIONS, @@ -1162,8 +1162,8 @@ test("create_agent_request launches from an exact subdirectory in a created work const emitted: SessionOutboundMessage[] = []; const session = new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, appVersion: null, @@ -1301,8 +1301,8 @@ test("create_agent_request does not title an existing workspace from the agent p let generateCalls = 0; const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, appVersion: null, @@ -1572,8 +1572,8 @@ test("archive emits an authoritative agent_update upsert for subscribed clients" const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -2056,8 +2056,8 @@ 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({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -2226,8 +2226,8 @@ test("close_items_request archives stored agents that are not currently loaded", const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -2387,8 +2387,8 @@ test("close_items_request continues after an archive failure", async () => { const killTerminalBestEffort = vi.fn(); const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), @@ -3660,8 +3660,8 @@ test("workspace update stream keeps persisted workspace visible after agents sto const session = asTestSession( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage: (message) => emitted.push(message), diff --git a/packages/server/src/server/snapshot-mutation-ownership.test.ts b/packages/server/src/server/snapshot-mutation-ownership.test.ts index 63aa871d5e..2cfe5d3e18 100644 --- a/packages/server/src/server/snapshot-mutation-ownership.test.ts +++ b/packages/server/src/server/snapshot-mutation-ownership.test.ts @@ -1,6 +1,6 @@ import { - createRequestReceiptsStub, - createCreationServiceStub, + createMessageReceiptsStub, + createTestCreationService, } from "./test-utils/session-stubs.js"; import os from "node:os"; import path from "node:path"; @@ -103,8 +103,8 @@ describe("snapshot mutation ownership boundary", () => { const session = asInternals( new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "test-client", permissions: OWNER_PERMISSIONS, onMessage, diff --git a/packages/server/src/server/test-utils/session-stubs.ts b/packages/server/src/server/test-utils/session-stubs.ts index 03acd72fee..6b46cfcbef 100644 --- a/packages/server/src/server/test-utils/session-stubs.ts +++ b/packages/server/src/server/test-utils/session-stubs.ts @@ -1,5 +1,9 @@ import { randomUUID } from "node:crypto"; -import { vi } from "vitest"; +import { afterEach, vi } from "vitest"; +import { rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { CreationService } from "../creation/index.js"; import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest"; @@ -281,18 +285,9 @@ export function createProviderSnapshotManagerStub(): { }; } -export function createRequestReceiptsStub(): SessionOptions["requestReceipts"] { +export function createMessageReceiptsStub(): SessionOptions["messageReceipts"] { return { - async createAgent(input) { - const agentId = randomUUID(); - await input.create(agentId); - return agentId; - }, - async createWorkspace(input) { - await input.create(input.workspaceId); - return input.workspaceId; - }, - sendMessage: (input) => input.send(), + send: (input) => input.send(), }; } @@ -309,13 +304,16 @@ export function createProviderSnapshot( }; } -export function createCreationServiceStub(): SessionOptions["creationService"] { - return { - create: async () => { - throw new Error("Unexpected creation in legacy session fixture"); - }, - subscribe: async () => { - throw new Error("Unexpected creation subscription in legacy session fixture"); - }, - }; +const creationDirectories: string[] = []; +afterEach(async () => { + await Promise.all( + creationDirectories + .splice(0) + .map((directory) => rm(directory, { recursive: true, force: true })), + ); +}); +export function createTestCreationService(): SessionOptions["creationService"] { + const directory = join(tmpdir(), `session-creation-${randomUUID()}`); + creationDirectories.push(directory); + return new CreationService(directory); } diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 8f40f5791c..14711e240c 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,7 +1,7 @@ import { stat } from "node:fs/promises"; import type { CreationSnapshot } from "@getpaseo/protocol/messages"; import { CreationService } from "./creation/index.js"; -import { RequestReceipts } from "./request-receipts/index.js"; +import { MessageReceipts } from "./message-receipts/index.js"; import { WebSocket, WebSocketServer } from "ws"; import type { IncomingMessage, Server as HTTPServer } from "http"; import { join } from "path"; @@ -532,7 +532,7 @@ export class VoiceAssistantWebSocketServer { private readonly daemonRuntimeConfig: DaemonRuntimeConfig | undefined; private readonly agentManager: AgentManager; private readonly agentStorage: AgentStorage; - private readonly requestReceipts: RequestReceipts; + private readonly messageReceipts: MessageReceipts; private readonly creationService: CreationService; private readonly projectRegistry: ProjectRegistry; private readonly workspaceRegistry: WorkspaceRegistry; @@ -672,9 +672,11 @@ export class VoiceAssistantWebSocketServer { this.orchestrationSkills = orchestrationSkills; this.agentManager = agentManager; this.agentStorage = agentStorage; - this.requestReceipts = new RequestReceipts(join(paseoHome, "agent-requests")); - this.creationService = new CreationService(join(paseoHome, "creations"), (snapshot) => - this.validateCompletedCreation(snapshot), + this.messageReceipts = new MessageReceipts(join(paseoHome, "agent-requests")); + this.creationService = new CreationService( + join(paseoHome, "creations"), + (snapshot) => this.validateCompletedCreation(snapshot), + join(paseoHome, "agent-requests"), ); this.projectRegistry = projectRegistry ?? createNoopProjectRegistry(); this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry(); @@ -1451,7 +1453,7 @@ export class VoiceAssistantWebSocketServer { worktreesRoot: this.worktreesRoot, agentManager: this.agentManager, agentStorage: this.agentStorage, - requestReceipts: this.requestReceipts, + messageReceipts: this.messageReceipts, creationService: this.creationService, projectRegistry: this.projectRegistry, workspaceRegistry: this.workspaceRegistry, diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index 0b2d899326..34abdebc3f 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -1,6 +1,6 @@ import { - createRequestReceiptsStub, - createCreationServiceStub, + createMessageReceiptsStub, + createTestCreationService, } from "./test-utils/session-stubs.js"; import pino from "pino"; import { z } from "zod"; @@ -211,8 +211,8 @@ function createSessionForWireCompatTest(options?: { ]; const session = new Session({ - requestReceipts: createRequestReceiptsStub(), - creationService: createCreationServiceStub(), + messageReceipts: createMessageReceiptsStub(), + creationService: createTestCreationService(), clientId: "wire-compat-client", permissions: OWNER_PERMISSIONS, clientCapabilities: options?.clientCapabilities ?? null,