mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-22 17:22:47 +00:00
Fix agents showing running after daemon restart (#7573)
* fix(server): close stale agent status after daemon restart Repair persisted active statuses before serving clients. Let Claude process-tree cleanup run immediately after SDK close rather than waiting on an already closed control channel. * test(app): expect closed agent status after daemon restart
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<void>((resolve) => {
|
||||
ready = resolve;
|
||||
});
|
||||
let finishStream!: () => void;
|
||||
const streamFinished = new Promise<void>((resolve) => {
|
||||
finishStream = resolve;
|
||||
});
|
||||
const never = new Promise<never>(() => {});
|
||||
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<void>((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<unknown> | undefined;
|
||||
let peerRun: Promise<unknown> | 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,
|
||||
);
|
||||
|
||||
@@ -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<void> {
|
||||
// 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
|
||||
|
||||
Reference in New Issue
Block a user