diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 2bdd0b998f..2663735fe2 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1499,7 +1499,8 @@ test("canceling provider startup settles creation and closes a late session with await client.waitForCreationToStart(); try { controller.abort(new Error("startup expired")); - await expect.poll(() => outcome, { timeout: 500 }).toBe("canceled"); + await creation; + expect(outcome).toBe("canceled"); client.finishCreating(); await expect.poll(() => client.createdSessionClosed).toBe(true); expect(manager.listAgents()).toEqual([]); diff --git a/packages/server/src/server/agent/requests/index.test.ts b/packages/server/src/server/agent/requests/index.test.ts index d066cda8a4..2920f2e45d 100644 --- a/packages/server/src/server/agent/requests/index.test.ts +++ b/packages/server/src/server/agent/requests/index.test.ts @@ -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 { AgentRequests, AgentRequestError } from "./index.js"; const directories: string[] = []; afterEach(async () => { @@ -233,8 +233,8 @@ test("a reconstructed journal reports interrupted startup as unknown and never l restarted.run(context, async () => "must not launch"), ]); expect(replays).toEqual([ - { status: "rejected", reason: new Error("agent_request_outcome_unknown") }, - { status: "rejected", reason: new Error("agent_request_outcome_unknown") }, + { status: "rejected", reason: new AgentRequestError("agent_request_outcome_unknown") }, + { status: "rejected", reason: new AgentRequestError("agent_request_outcome_unknown") }, ]); expect(await restarted.cancelOperation(context.key, "conversation")).toEqual({ agentId: null, @@ -248,21 +248,36 @@ test("a reconstructed journal reports interrupted startup as unknown and never l }); test("a live deadline settles a hung waiter without a cancel frame and preserves late cleanup", async () => { - const { requests } = await fixture(); + const { directory } = await fixture(); + let expire!: () => void; + const now = Date.parse("2026-09-09T00:00:00Z"); + const requests = new AgentRequests(directory, { + now: () => now, + schedule(callback, delayMs) { + expect(delayMs).toBe(120_000); + expire = callback; + return () => {}; + }, + }); let release!: () => void; const gate = new Promise((resolve) => { release = resolve; }); - let entered = false; + let enter!: () => void; + const entered = new Promise((resolve) => { + enter = resolve; + }); const pending = requests.run( - { key: "deadline", deadlineAt: new Date(Date.now() + 200).toISOString() }, + { key: "deadline", deadlineAt: new Date(now + 120_000).toISOString() }, async () => { - entered = true; + enter(); await gate; }, ); - await expect(pending).rejects.toThrow("agent_request_canceled"); - expect(entered).toBe(true); + const rejected = expect(pending).rejects.toThrow("agent_request_canceled"); + await entered; + expire(); + await rejected; expect(await requests.inspectOperation("deadline", "conversation")).toMatchObject({ outcome: "pending", }); @@ -270,9 +285,10 @@ test("a live deadline settles a hung waiter without a cancel frame and preserves "agent_request_canceled", ); release(); - await expect - .poll(() => requests.inspectOperation("deadline", "conversation")) - .toMatchObject({ outcome: "settled" }); + await requests.cancelAndSettle("deadline"); + expect(await requests.inspectOperation("deadline", "conversation")).toMatchObject({ + outcome: "settled", + }); expect(await requests.run({ key: "next" }, async () => "ready")).toBe("ready"); }); @@ -325,7 +341,7 @@ test("refused provider cancellation remains unknown after restart", async () => const { requests, directory } = await fixture(); await expect( requests.run({ key: "refused" }, async () => { - throw new Error("agent_request_outcome_unknown"); + throw new AgentRequestError("agent_request_outcome_unknown"); }), ).rejects.toThrow("agent_request_outcome_unknown"); expect( diff --git a/packages/server/src/server/agent/requests/index.ts b/packages/server/src/server/agent/requests/index.ts index dfb0a61184..8c0a703755 100644 --- a/packages/server/src/server/agent/requests/index.ts +++ b/packages/server/src/server/agent/requests/index.ts @@ -11,6 +11,30 @@ const ReceiptSchema = z.object({ }); type Receipt = z.infer; +export class AgentRequestError extends Error { + constructor( + public readonly code: + | "agent_request_canceled" + | "agent_request_outcome_unknown" + | "agent_request_key_conflict", + ) { + super(code); + this.name = "AgentRequestError"; + } +} + +interface RequestClock { + now(): number; + schedule(callback: () => void, delayMs: number): () => void; +} +const systemClock: RequestClock = { + now: () => Date.now(), + schedule(callback, delayMs) { + const timer = setTimeout(callback, delayMs); + return () => clearTimeout(timer); + }, +}; + /** One daemon-owned request journal, shared by all of its socket sessions. */ export class AgentRequests { private readonly pending = new Map>(); @@ -20,7 +44,10 @@ export class AgentRequests { >(); private readonly operationWrites = new Map>(); - constructor(private readonly directory: string) {} + constructor( + private readonly directory: string, + private readonly clock: RequestClock = systemClock, + ) {} /** Runtime cancellation belongs to the same journal as idempotent create/send receipts. */ async run( @@ -37,7 +64,7 @@ export class AgentRequests { initialized: (async () => { await previousWrite; if (await this.wasInterrupted(context.key)) - throw new Error("agent_request_outcome_unknown"); + throw new AgentRequestError("agent_request_outcome_unknown"); })(), }; this.operations.set(context.key, active); @@ -46,23 +73,28 @@ export class AgentRequests { const controller = new AbortController(); const controllers = active.controllers; controllers.add(controller); - let timer: ReturnType | undefined; + let clearTimer: (() => void) | undefined; let abort: (() => void) | undefined; let started = false; let recorded = false; try { await active.initialized; const remaining = - context.deadlineAt === undefined ? undefined : Date.parse(context.deadlineAt) - Date.now(); + context.deadlineAt === undefined + ? undefined + : Date.parse(context.deadlineAt) - this.clock.now(); if ( (remaining !== undefined && (!Number.isFinite(remaining) || remaining <= 0)) || (await this.wasCanceled(context.key)) ) { - throw new Error("agent_request_canceled"); + throw new AgentRequestError("agent_request_canceled"); } controller.signal.throwIfAborted(); if (remaining !== undefined) - timer = setTimeout(() => controller.abort(new Error("agent_request_canceled")), remaining); + clearTimer = this.clock.schedule( + () => controller.abort(new AgentRequestError("agent_request_canceled")), + remaining, + ); await this.recordOperationState(context.key); recorded = true; controller.signal.throwIfAborted(); @@ -76,7 +108,7 @@ export class AgentRequests { return operation(controller.signal); }) .catch((error: unknown) => { - if (error instanceof Error && error.message === "agent_request_outcome_unknown") + if (error instanceof AgentRequestError && error.code === "agent_request_outcome_unknown") operationState.unknown = true; throw error; }) @@ -96,7 +128,7 @@ export class AgentRequests { }), ]); } finally { - clearTimeout(timer); + clearTimer?.(); if (abort) controller.signal.removeEventListener("abort", abort); if (controller.signal.aborted) await this.recordCancellation(context.key); if (!started) { @@ -111,7 +143,7 @@ export class AgentRequests { // Persist before acknowledging: a delayed request or a new socket must also be fenced. await this.recordCancellation(key); for (const controller of this.operations.get(key)?.controllers ?? []) { - controller.abort(new Error("agent_request_canceled")); + controller.abort(new AgentRequestError("agent_request_canceled")); } } @@ -131,7 +163,8 @@ export class AgentRequests { const receipt = z .object({ fingerprint: z.string() }) .parse(JSON.parse(await readFile(file, "utf8"))); - if (receipt.fingerprint !== fingerprint) throw new Error("agent_request_key_conflict"); + if (receipt.fingerprint !== fingerprint) + throw new AgentRequestError("agent_request_key_conflict"); return; } catch (error) { if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) throw error; @@ -145,7 +178,8 @@ export class AgentRequests { await this.cancel(key); await this.pending.get(digest(["operation", key]))?.catch(() => undefined); await this.operationWrites.get(key); - if (await this.wasInterrupted(key)) throw new Error("agent_request_outcome_unknown"); + if (await this.wasInterrupted(key)) + throw new AgentRequestError("agent_request_outcome_unknown"); } async cancelOperation( @@ -297,10 +331,11 @@ 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.fingerprint !== fingerprint) + throw new AgentRequestError("agent_request_key_conflict"); if (existing.state === "completed") return existing.agentId; if (!(await operation.recover(existing.agentId))) { - throw new Error("agent_request_outcome_unknown"); + throw new AgentRequestError("agent_request_outcome_unknown"); } await writeJsonFileAtomic(file, { ...existing, state: "completed" }); return existing.agentId; diff --git a/packages/server/src/server/hub/daemon-executions.test.ts b/packages/server/src/server/hub/daemon-executions.test.ts index 33fcb94d2b..abbea2ecf4 100644 --- a/packages/server/src/server/hub/daemon-executions.test.ts +++ b/packages/server/src/server/hub/daemon-executions.test.ts @@ -333,13 +333,8 @@ test("control cancels a held legacy provider create and fences a delayed replay" hub.beginOwnedCreate("held-create", "expired-create"); await hub.agentCreationAttempts(1); const creation = hub.ownedCreateResult("held-create"); - let settled = false; - const control = hub.controlExecution("expired-create", "archive").then((result) => { - settled = true; - return result; - }); + const control = hub.controlExecution("expired-create", "archive"); try { - await expect.poll(() => settled, { timeout: 500 }).toBe(true); expect(await control).toMatchObject({ success: true }); expect(await creation).toMatchObject({ payload: { success: false } }); } finally { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 2de6556c7b..ea235f1d69 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1,5 +1,5 @@ import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; -import type { AgentRequests } from "./agent/requests/index.js"; +import { AgentRequestError, type AgentRequests } from "./agent/requests/index.js"; import equal from "fast-deep-equal"; import { v4 as uuidv4 } from "uuid"; import { lstat, mkdir, mkdtemp, rename, rm, stat } from "node:fs/promises"; @@ -3671,7 +3671,10 @@ export class Session { signal?: AbortSignal, ): Promise { if (!signal?.aborted || !agentId) return agentId; + beginAgentDeleteIfSupported(this.agentStorage, agentId); await this.agentManager.closeAgent(agentId); + await this.agentManager.flush(); + await this.agentStorage.remove(agentId); await this.agentManager.deleteAgentState(agentId); return null; } @@ -7656,7 +7659,8 @@ export class Session { const abort = () => { if (this.agentManager.getAgent(resolved.agentId)) { cancellation ??= this.agentManager.cancelAgentRun(resolved.agentId).then((result) => { - if (result.status === "refused") throw new Error("agent_request_outcome_unknown"); + if (result.status === "refused") + throw new AgentRequestError("agent_request_outcome_unknown"); return result; }); void cancellation.catch(() => undefined); diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 2914e58215..83fa9bade2 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -1,3 +1,4 @@ +import { AgentRequests } from "./agent/requests/index.js"; import { createAgentRequestsStub } from "./test-utils/session-stubs.js"; import { execFileSync } from "node:child_process"; import { @@ -932,6 +933,126 @@ test("client heartbeat clears attention for the focused terminal", async () => { }); }); +test("expiry after agent registration removes its durable record before creation can retry", async () => { + const workdir = mkdtempSync(path.join(tmpdir(), "paseo-canceled-create-")); + const logger = createTestLogger(); + const agentStorage = new AgentStorage(path.join(workdir, "agents"), logger); + const agentManager = new AgentManager({ + clients: { codex: new CreateAgentTestClient() }, + registry: agentStorage, + logger, + }); + const projectRegistry = new FileBackedProjectRegistry( + path.join(workdir, "projects.json"), + logger, + ); + const workspaceRegistry = new FileBackedWorkspaceRegistry( + path.join(workdir, "workspaces.json"), + logger, + ); + let expire!: () => void; + const requests = new AgentRequests(path.join(workdir, "requests"), { + now: () => 0, + schedule(callback) { + expire = callback; + return () => {}; + }, + }); + let registeredId: string | undefined; + const unsubscribe = agentManager.subscribe( + (event) => { + if (event.type === "agent_state" && registeredId === undefined) { + registeredId = event.agent.id; + expire(); + } + }, + { replayState: false }, + ); + try { + const emitted: SessionOutboundMessage[] = []; + const session = asTestSession( + new Session({ + agentRequests: requests, + clientId: "test-client", + serverId: "test-server", + permissions: OWNER_PERMISSIONS, + appVersion: null, + onMessage: (message) => emitted.push(message), + logger: asSessionLogger(logger), + downloadTokenStore: asDownloadTokenStore(), + pushNotifications: asPushNotifications(), + paseoHome: path.join(workdir, "paseo-home"), + agentManager, + agentStorage, + projectRegistry, + workspaceRegistry, + scheduleService: asScheduleService(), + checkoutDiffManager: asCheckoutDiffManager({ + subscribe: async () => ({ + initial: { cwd: workdir, files: [], error: null }, + unsubscribe: () => {}, + }), + scheduleRefreshForCwd: () => {}, + onWorkspaceStateMayHaveChanged: () => {}, + invalidateForge: () => {}, + getMetrics: () => ({ + checkoutDiffTargetCount: 0, + checkoutDiffSubscriptionCount: 0, + checkoutDiffWatcherCount: 0, + checkoutDiffFallbackRefreshTargetCount: 0, + }), + dispose: () => {}, + }), + workspaceGitService: createNoopWorkspaceGitService(), + daemonConfigStore: asDaemonConfigStore({ + get: () => ({ mcp: { injectIntoAgents: false }, providers: {} }), + onChange: () => () => {}, + }), + mcpBaseUrl: null, + stt: null, + tts: null, + providerSnapshotManager: createProviderSnapshotManagerStub().manager, + terminalManager: null, + }), + ); + + await session.handleMessage({ + type: "create_agent_request", + requestId: "canceled", + idempotencyKey: "conversation", + operation: { key: "arrival", deadlineAt: "2026-09-09T00:00:00Z" }, + config: { provider: "codex", cwd: workdir }, + }); + await requests.cancelAndSettle("arrival"); + expect(registeredId).toEqual(expect.any(String)); + expect(agentManager.listAgents()).toEqual([]); + expect(await agentStorage.list()).toEqual([]); + expect(await requests.inspectOperation("arrival", "conversation")).toEqual({ + outcome: "settled", + agentId: null, + }); + unsubscribe(); + await session.handleMessage({ + type: "create_agent_request", + requestId: "retry", + idempotencyKey: "conversation", + operation: { key: "next", deadlineAt: "2026-09-09T00:00:00Z" }, + config: { provider: "codex", cwd: workdir }, + }); + expect(agentManager.listAgents()).toHaveLength(1); + expect( + emitted.filter( + (message) => message.type === "status" && message.payload.status === "agent_created", + ), + ).toHaveLength(1); + } finally { + unsubscribe(); + await Promise.all(agentManager.listAgents().map((agent) => agentManager.closeAgent(agent.id))); + await agentManager.flush(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("create_agent_request keeps requested child cwd when grouped under an existing parent workspace", async () => { const workdir = mkdtempSync(path.join(tmpdir(), "paseo-create-agent-cwd-")); try { diff --git a/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.test.ts b/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.test.ts index 8061eba69f..8307eb9a24 100644 --- a/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.test.ts +++ b/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.test.ts @@ -119,7 +119,7 @@ describe("workspace recovery", () => { test("a repeated restore of an already active workspace succeeds without another unarchive", async () => { const workspace = createWorkspace({ archivedAt: null }); - const { service, unarchived } = createHarness({ workspace }); + const { service, unarchived } = createHarness({ workspace, directories: [workspace.cwd] }); await expect(service.restore(workspace.workspaceId)).resolves.toEqual({ workspaceId: workspace.workspaceId, action: "unarchive", @@ -127,6 +127,64 @@ describe("workspace recovery", () => { expect(unarchived).toEqual([]); }); + test("an active workspace with a missing directory cannot report a successful restore", async () => { + const workspace = createWorkspace({ archivedAt: null }); + const { service, unarchived } = createHarness({ workspace }); + await expect(service.restore(workspace.workspaceId)).rejects.toThrow(); + expect(unarchived).toEqual([]); + }); + + test("expiry after filesystem recreation rolls back the new checkout", async () => { + const { tempDir, repoDir } = createGitRepository(); + const branch = "feature/canceled-restore"; + execFileSync("git", ["branch", branch], { cwd: repoDir, stdio: "pipe" }); + const paseoHome = join(tempDir, "paseo-home"); + const worktreesRoot = join(tempDir, "worktrees"); + const created = await createWorktree({ + cwd: repoDir, + worktreeSlug: "canceled-restore", + source: { kind: "checkout-branch", branchName: branch }, + runSetup: false, + paseoHome, + worktreesRoot, + }); + const worktreeRoot = realpathSync(created.worktreePath); + rmSync(worktreeRoot, { recursive: true, force: true }); + const workspace = createWorkspace({ + cwd: worktreeRoot, + branch, + worktreeRoot, + mainRepoRoot: repoDir, + }); + const controller = new AbortController(); + let unarchived = false; + const service = createWorkspaceRecoveryService({ + paseoHome, + worktreesRoot, + getWorkspace: async () => workspace, + getProject: async () => createProject({ rootPath: repoDir }), + isDirectory: async (target) => { + const exists = existsSync(target) && statSync(target).isDirectory(); + if (target === worktreeRoot && exists) controller.abort(new Error("expired")); + return exists; + }, + unarchiveWorkspace: async () => { + unarchived = true; + }, + }); + await expect(service.restore(workspace.workspaceId, controller.signal)).rejects.toThrow( + "expired", + ); + expect(unarchived).toBe(false); + expect(existsSync(worktreeRoot)).toBe(false); + expect( + execFileSync("git", ["worktree", "list", "--porcelain"], { + cwd: repoDir, + stdio: "pipe", + }).toString(), + ).not.toContain(worktreeRoot); + }); + test("expiry during restore preparation cannot unarchive the workspace", async () => { const workspace = createWorkspace({ kind: "directory", branch: null }); let release!: () => void; diff --git a/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.ts b/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.ts index 63801f8625..84227c9483 100644 --- a/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.ts +++ b/packages/server/src/server/session/workspace-recovery/workspace-recovery-service.ts @@ -81,7 +81,7 @@ export function createWorkspaceRecoveryService(deps: { message: "This workspace is no longer known to the host.", }; } - if (!workspace.archivedAt) { + if (!workspace.archivedAt && (await deps.isDirectory(workspace.cwd))) { return { kind: "unavailable", workspaceId, @@ -90,6 +90,14 @@ export function createWorkspaceRecoveryService(deps: { }; } + if (!workspace.archivedAt) { + return { + kind: "unavailable", + workspaceId, + reason: "workspace_directory_missing", + message: "The workspace directory no longer exists.", + }; + } const project = await deps.getProject(workspace.projectId); if (!project) { return { @@ -154,7 +162,7 @@ export function createWorkspaceRecoveryService(deps: { } if (resolved.kind === "restore") { - await recreateArchivedWorktree(resolved.workspace, resolved.sourceRepoRoot); + await recreateArchivedWorktree(resolved.workspace, resolved.sourceRepoRoot, signal); } signal?.throwIfAborted(); await deps.unarchiveWorkspace(resolved.workspace); @@ -164,6 +172,7 @@ export function createWorkspaceRecoveryService(deps: { async function recreateArchivedWorktree( workspace: PersistedWorkspaceRecord, sourceRepoRoot: string, + signal?: AbortSignal, ): Promise { const branch = workspace.branch; if (!branch) { @@ -208,6 +217,7 @@ export function createWorkspaceRecoveryService(deps: { } try { + signal?.throwIfAborted(); const recreatedWorkspacePath = mapWorkspaceCwdToWorktree({ sourceWorktreePath: previousWorktreePath, workspaceCwd: workspace.cwd, @@ -225,6 +235,7 @@ export function createWorkspaceRecoveryService(deps: { message: `Selected project directory is missing from the restored worktree: ${recreatedWorkspacePath}`, }); } + signal?.throwIfAborted(); } catch (error) { return rollbackCreatedPaseoWorktree( {