diff --git a/packages/app/e2e/browser/workspace-model-restart.spec.ts b/packages/app/e2e/browser/workspace-model-restart.spec.ts index e1dd703508..976e745549 100644 --- a/packages/app/e2e/browser/workspace-model-restart.spec.ts +++ b/packages/app/e2e/browser/workspace-model-restart.spec.ts @@ -404,7 +404,7 @@ async function fetchWorkspaceStatuses( } test.describe("Workspace model restart regressions", () => { - test("browser-created same-cwd workspace preserves restarted agent status and migrated tab ownership", async ({ + test("browser-created same-cwd workspace preserves repaired restart status and migrated tab ownership", async ({ page, baseURL, }) => { @@ -423,9 +423,13 @@ test.describe("Workspace model restart regressions", () => { .toMatchObject({ id: LEGACY_AGENT_ID, workspaceId: seeded.workspaceA, - status: "running", + status: "closed", }); + await expect + .poll(() => fetchWorkspaceStatuses(client, [seeded.workspaceA, seeded.workspaceB])) + .toEqual({ [seeded.workspaceA]: "done", [seeded.workspaceB]: "done" }); + await page.goto(buildHostWorkspaceRoute(serverId, seeded.workspaceA)); await waitForSidebarHydration(page); await expectWorkspaceRowDoesNotShowIndicator(page, { @@ -472,15 +476,11 @@ test.describe("Workspace model restart regressions", () => { ]), ) .toMatchObject({ + [seeded.workspaceA]: "done", [seeded.workspaceB]: "done", [createdWorkspaceId]: "done", }); - // The restarted provider session may settle while the browser creates the sibling. Its - // initial running status is asserted above; this phase verifies that ownership never moves. - const workspaceStatuses = await fetchWorkspaceStatuses(client, [seeded.workspaceA]); - expect(["running", "done"]).toContain(workspaceStatuses[seeded.workspaceA]); - await expectWorkspaceRowDoesNotShowIndicator(page, { serverId, workspaceId: seeded.workspaceB, diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index c2f607665d..6a70d28d6a 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -2903,8 +2903,9 @@ class ClaudeAgentSession implements AgentSession { this.taskProtocolSource.reset(); this.input?.end(); this.query?.close?.(); - await this.awaitWithTimeout(this.query?.interrupt?.(), "close query interrupt"); - await this.awaitWithTimeout(this.query?.return?.(), "close query return"); + // close() already terminates the SDK query. Its control plane is no longer + // available; waiting for interrupt/return here delays our process-tree cleanup + // beyond the daemon's close deadline when the provider stops answering. this.query = null; this.input = null; // Terminate the entire process tree (claude + MCP children) to prevent diff --git a/packages/server/src/server/bootstrap.smoke.test.ts b/packages/server/src/server/bootstrap.smoke.test.ts index 697dfda8a7..f6304491ab 100644 --- a/packages/server/src/server/bootstrap.smoke.test.ts +++ b/packages/server/src/server/bootstrap.smoke.test.ts @@ -1,5 +1,10 @@ import { resolveDaemonVersion } from "./daemon-version.js"; import os from "node:os"; +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import type { ChildProcess } from "node:child_process"; +import type { Query } from "@anthropic-ai/claude-agent-sdk"; +import { ClaudeAgentClient } from "./agent/providers/claude/agent.js"; import http from "node:http"; import path from "node:path"; import { access, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; @@ -9,6 +14,14 @@ import { WebSocket } from "ws"; import { createPaseoDaemon, parseListenString, type PaseoDaemonConfig } from "./bootstrap.js"; import { loadConfig } from "./config.js"; +import { + FileBackedProjectRegistry, + FileBackedWorkspaceRegistry, + createPersistedProjectRecord, + createPersistedWorkspaceRecord, +} from "./workspace-registry.js"; +import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js"; +import { ensureAgentLoaded } from "./agent/agent-loading.js"; import { AgentManagerShuttingDownError } from "./agent/agent-manager.js"; import { hashDaemonPassword } from "./auth.js"; import { generateLocalPairingOffer } from "./pairing-offer.js"; @@ -1055,3 +1068,225 @@ function probeWebSocketConnection( }); }); } + +test("startup closes stale active records before serving status without resuming sessions", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "paseo-stale-status-")); + const logger = pino({ level: "silent" }); + const storagePath = path.join(root, ".paseo", "agents"); + const storage = new AgentStorage(storagePath, logger); + const timestamp = "2025-01-01T00:00:00.000Z"; + const registryDir = path.join(root, ".paseo", "projects"); + await new FileBackedProjectRegistry(path.join(registryDir, "projects.json"), logger).upsert( + createPersistedProjectRecord({ + projectId: "existing-project", + rootPath: root, + kind: "non_git", + displayName: "Existing", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + await new FileBackedWorkspaceRegistry(path.join(registryDir, "workspaces.json"), logger).upsert( + createPersistedWorkspaceRecord({ + workspaceId: "existing-workspace", + projectId: "existing-project", + cwd: root, + kind: "directory", + displayName: "Existing", + createdAt: timestamp, + updatedAt: timestamp, + }), + ); + const records: StoredAgentRecord[] = ["running", "initializing", "idle", "error", "closed"].map( + (status, index) => ({ + id: randomUUID(), + workspaceId: "existing-workspace", + provider: "codex", + cwd: root, + createdAt: "2025-01-01T00:00:00.000Z", + updatedAt: "2025-01-02T00:00:00.000Z", + lastActivityAt: "2025-01-02T00:00:00.000Z", + lastStatus: status as StoredAgentRecord["lastStatus"], + title: `Keep ${status}`, + labels: { "paseo.parent-agent-id": "parent", custom: "keep" }, + persistence: { provider: "codex", sessionId: `session-${index}` }, + config: { modeId: "full-access" }, + requiresAttention: true, + attentionReason: "permission", + attentionTimestamp: "2025-01-02T00:00:00.000Z", + archivedAt: index === 1 ? "2025-01-03T00:00:00.000Z" : null, + }), + ); + for (const record of records) await storage.upsert(record); + let sessionsCreated = 0; + const handle = await createTestPaseoDaemon({ + paseoHomeRoot: root, + agentClients: createTestAgentClients({ + beforeCreateSession: async () => { + sessionsCreated++; + }, + }), + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${handle.port}/ws` }); + try { + const expected = records.map((record) => ({ + ...record, + lastStatus: + record.lastStatus === "running" || record.lastStatus === "initializing" + ? "closed" + : record.lastStatus, + })); + expect( + (await handle.daemon.agentStorage.list()).sort((a, b) => a.id.localeCompare(b.id)), + ).toEqual(expected.sort((a, b) => a.id.localeCompare(b.id))); + // A fresh reader proves the repaired states were written, rather than projected only in memory. + expect( + (await new AgentStorage(storagePath, logger).list()).sort((a, b) => a.id.localeCompare(b.id)), + ).toEqual(expected); + expect(handle.daemon.agentManager.listAgents()).toEqual([]); + expect(sessionsCreated).toBe(0); + await client.connect(); + const snapshot = await client.fetchAgents({ filter: { includeArchived: true } }); + expect( + Object.fromEntries(snapshot.entries.map((entry) => [entry.agent.id, entry.agent.status])), + ).toEqual(Object.fromEntries(expected.map((record) => [record.id, record.lastStatus]))); + const opened = await ensureAgentLoaded(records[0]!.id, { + agentManager: handle.daemon.agentManager, + agentStorage: handle.daemon.agentStorage, + logger, + }); + expect(opened.id).toBe(records[0]!.id); + expect(opened.persistence?.sessionId).toBe("session-0"); + expect(opened.lifecycle).toBe("idle"); + await handle.daemon.agentManager.runAgent(opened.id, "hello"); + expect(handle.daemon.agentManager.getAgent(opened.id)?.lifecycle).toBe("idle"); + } finally { + await client.close(); + await handle.close(); + } +}, 30_000); + +test.each(["reject", "stall"] as const)( + "shutdown closes Claude and its process despite a %s peer and stalled SDK handshake", + async (failure) => { + let child: ChildProcess | undefined; + let ready!: () => void; + const processReady = new Promise((resolve) => { + ready = resolve; + }); + let finishStream!: () => void; + const streamFinished = new Promise((resolve) => { + finishStream = resolve; + }); + const never = new Promise(() => {}); + const claude = new ClaudeAgentClient({ + logger: pino({ level: "silent" }), + resolveBinary: async () => process.execPath, + queryFactory: ({ options }) => { + child = options.spawnClaudeCodeProcess!({ + command: process.execPath, + args: [ + "-e", + "process.on('SIGTERM', () => {}); console.log('ready'); setInterval(() => {}, 1000)", + ], + cwd: process.cwd(), + env: process.env, + signal: new AbortController().signal, + }) as ChildProcess; + child.stdout!.once("data", ready); + let initialized = false; + // An injected SDK adapter leaves an active turn and both close requests unanswered. + return { + async next() { + if (!initialized) { + initialized = true; + return { + done: false, + value: { + type: "system", + subtype: "init", + session_id: "shutdown-session", + permissionMode: "default", + }, + }; + } + await streamFinished; + return { done: true, value: undefined }; + }, + interrupt: () => never, + return: () => never, + close() {}, + async supportedModels() { + return []; + }, + async supportedCommands() { + return []; + }, + async setPermissionMode() {}, + async setModel() {}, + async applyFlagSettings() {}, + [Symbol.asyncIterator]() { + return this; + }, + } as Query; + }, + }); + let armed = false; + let peerRunning!: () => void; + const peerReady = new Promise((resolve) => { + peerRunning = resolve; + }); + const peers = createTestAgentClients({ + beforeTurnComplete: async () => { + peerRunning(); + await streamFinished; + }, + closeSession: async () => { + if (!armed) return; + if (failure === "reject") throw new Error("peer close rejected"); + await never; + }, + }); + const handle = await createTestPaseoDaemon({ agentClients: { claude, codex: peers.codex } }); + let stopped = false; + let run: Promise | undefined; + let peerRun: Promise | undefined; + try { + const agent = await handle.daemon.agentManager.createAgent( + { provider: "claude", cwd: handle.paseoHome }, + undefined, + { workspaceId: undefined }, + ); + run = handle.daemon.agentManager.runAgent(agent.id, "keep working").catch(() => undefined); + await processReady; + expect(handle.daemon.agentManager.getAgent(agent.id)?.lifecycle).toBe("running"); + const peer = await handle.daemon.agentManager.createAgent( + { provider: "codex", cwd: handle.paseoHome, modeId: "full-access" }, + undefined, + { workspaceId: undefined }, + ); + peerRun = handle.daemon.agentManager.runAgent(peer.id, "keep working").catch(() => undefined); + await peerReady; + armed = true; + await handle.daemon.stop(); + stopped = true; + // Windows taskkill is forceful on the first signal; POSIX escalates after SIGTERM. + expect(child!.exitCode !== null || child!.signalCode !== null).toBe(true); + expect((await handle.daemon.agentStorage.get(agent.id))?.lastStatus).toBe("closed"); + expect(handle.daemon.agentManager.getAgent(agent.id)).toBeNull(); + expect((await handle.daemon.agentStorage.get(peer.id))?.lastStatus).toBe("running"); + } finally { + finishStream(); + if (child && child.exitCode === null && child.signalCode === null) { + const exited = once(child, "exit"); + child.kill("SIGKILL"); + await exited; + } + if (!stopped) await handle.daemon.stop(); + await Promise.all([run, peerRun]); + await rm(path.dirname(handle.paseoHome), { recursive: true, force: true }); + await rm(handle.staticDir, { recursive: true, force: true }); + } + }, + 30_000, +); diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 09be15772e..22bf93b625 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -975,6 +975,7 @@ export async function createPaseoDaemon( agentStorage, ); await agentStorage.initialize(); + await closeStaleAgentRecords(agentStorage); logger.info({ elapsed: elapsed() }, "Agent storage initialized"); await bootstrapWorkspaceRegistries({ serverId, @@ -1887,6 +1888,16 @@ export async function createPaseoDaemon( }; } +async function closeStaleAgentRecords(agentStorage: AgentStorage): Promise { + // No provider runtimes are loaded at startup. Repair interrupted + // shutdowns before any status consumer sees the persisted active states. + for (const record of await agentStorage.list()) { + if (record.lastStatus === "running" || record.lastStatus === "initializing") { + await agentStorage.upsert({ ...record, lastStatus: "closed" }); + } + } +} + /** * Closing an agent asks its provider to close the session and waits for the * answer. A provider that never answers must not hold the daemon open, so a