fix: make Hub lifecycle cleanup deterministic

This commit is contained in:
Mohamed Boudra
2026-07-13 14:32:14 +02:00
parent 0c4a4ace10
commit 6777bfb924
9 changed files with 311 additions and 36 deletions
@@ -766,6 +766,10 @@ export class AgentManager {
};
}
subscriptionCount(): number {
return this.subscribers.size;
}
listAgents(): ManagedAgent[] {
return Array.from(this.agents.values())
.filter((agent) => !agent.internal)
@@ -0,0 +1,47 @@
import { expect, test } from "vitest";
import type { AgentManagerEvent, AgentSubscriber } from "./agent-manager.js";
import { registerAgentAutoArchive } from "./create-agent-lifecycle-dispatch.js";
class AgentLifecycleEvents {
private readonly listeners = new Set<AgentSubscriber>();
subscribe(listener: AgentSubscriber): () => void {
this.listeners.add(listener);
return () => this.listeners.delete(listener);
}
completeTurn(agentId: string): void {
const event: AgentManagerEvent = {
type: "agent_stream",
agentId,
event: { type: "turn_completed", provider: "codex" },
};
for (const listener of this.listeners) listener(event);
}
listenerCount(): number {
return this.listeners.size;
}
}
test("auto-archive self-releases once and later cancellation waits harmlessly", async () => {
const agentId = "4a7e2521-286d-4ad5-af35-e091c55302e3";
const agents = new AgentLifecycleEvents();
let archiveCount = 0;
const registration = registerAgentAutoArchive({
agentManager: agents,
agentId,
archive: async () => {
archiveCount += 1;
},
});
agents.completeTurn(agentId);
await registration.cancel();
await registration.cancel();
agents.completeTurn(agentId);
expect(archiveCount).toBe(1);
expect(agents.listenerCount()).toBe(0);
});
@@ -18,7 +18,7 @@ import type {
FirstAgentContext,
SessionOutboundMessage,
} from "../messages.js";
import type { AgentManager } from "./agent-manager.js";
import type { AgentManager, AgentSubscriber, SubscribeOptions } from "./agent-manager.js";
import type { AgentStorage } from "./agent-storage.js";
interface CreateAgentLifecycleDispatchDependencies {
@@ -42,6 +42,16 @@ interface CreateAgentLifecycleDispatchDependencies {
logger: pino.Logger;
}
export interface LifecycleRegistration {
cancel(): Promise<void>;
}
interface AgentLifecycleEvents {
subscribe(callback: AgentSubscriber, options?: SubscribeOptions): () => void;
}
const inactiveRegistration: LifecycleRegistration = { cancel: async () => undefined };
export class CreateAgentLifecycleDispatch {
private readonly autoArchiveAgentIds = new Set<string>();
@@ -67,12 +77,12 @@ export class CreateAgentLifecycleDispatch {
autoArchive: boolean | undefined;
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}): void {
}): LifecycleRegistration {
if (input.autoArchive !== true) {
return;
return inactiveRegistration;
}
this.registerAutoArchiveOnTerminalState(input.agentId, {
return this.registerAutoArchiveOnTerminalState(input.agentId, {
worktreePath: input.createdWorktree?.worktree.worktreePath ?? null,
repoRoot: input.createdWorktree?.repoRoot ?? null,
});
@@ -146,8 +156,8 @@ export class CreateAgentLifecycleDispatch {
private registerAutoArchiveOnTerminalState(
agentId: string,
options: { worktreePath: string | null; repoRoot: string | null },
): void {
registerAgentAutoArchive({
): LifecycleRegistration {
return registerAgentAutoArchive({
agentManager: this.dependencies.agentManager,
agentId,
archive: () => this.autoArchiveAgentOnce(agentId, options),
@@ -240,11 +250,25 @@ export class CreateAgentLifecycleDispatch {
}
export function registerAgentAutoArchive(input: {
agentManager: AgentManager;
agentManager: AgentLifecycleEvents;
agentId: string;
archive: () => Promise<unknown>;
}): void {
const unsubscribe = input.agentManager.subscribe(
}): LifecycleRegistration {
let unsubscribe: (() => void) | null = null;
let archiveTask: Promise<unknown> | null = null;
const release = () => {
if (!unsubscribe) return;
const subscribed = unsubscribe;
unsubscribe = null;
subscribed();
};
const registration: LifecycleRegistration = {
async cancel() {
release();
await archiveTask;
},
};
unsubscribe = input.agentManager.subscribe(
(event) => {
if (event.type !== "agent_stream") return;
if (
@@ -254,9 +278,10 @@ export function registerAgentAutoArchive(input: {
) {
return;
}
unsubscribe();
void input.archive();
release();
archiveTask = input.archive();
},
{ agentId: input.agentId, replayState: false },
);
return registration;
}
@@ -107,11 +107,17 @@ test("Hub create forwards worktree and auto-archive through the existing create
hub.beginOwnedCreate("worktree-create", "execution-worktree", {
worktree: { mode: "branch-off", newBranch: "hub-created-worktree", base: "main" },
autoArchive: true,
prompt: "sleep 30",
modeId: "always-ask",
});
const worktreeCreated = await hub.ownedCreateResult("worktree-create");
const worktreeCwd = hub.latestCreatedCwd();
const permission = await hub.ownedPermissionRequest(worktreeCreated.payload.agentId!);
const duringRun = await hub.worktreeState(worktreeCwd!);
const archive = await hub.waitForOwnedArchiveCompletion(worktreeCreated.payload.agentId!);
const archiveCompletion = hub.waitForOwnedArchiveCompletion(worktreeCreated.payload.agentId!);
await hub.allowOwnedPermission(worktreeCreated.payload.agentId!, permission.id);
await hub.ownedTurnCompletion(worktreeCreated.payload.agentId!);
const archive = await archiveCompletion;
const afterArchive = await hub.worktreeState(worktreeCwd!);
expect(worktreeCreated).toMatchObject({
@@ -1,5 +1,6 @@
import { afterEach, describe, expect, test } from "vitest";
import { createHash } from "node:crypto";
import { platform } from "node:os";
import { afterEach, describe, expect, test } from "vitest";
import { HubRelationshipHarness } from "./test-utils/relationship-harness.js";
describe("Hub relationship", () => {
@@ -27,7 +28,7 @@ describe("Hub relationship", () => {
expect(relationship.loggableValues(status)).not.toContain(enrollment.token);
expect(relationship.loggableValues(status)).not.toContain(enrollment.idempotencyKey);
expect(disconnected.state).toBe("not_connected");
});
}, 30_000);
test("Hub URLs cannot persist embedded credentials", async () => {
relationship = await HubRelationshipHarness.start();
@@ -98,6 +99,7 @@ describe("Hub relationship", () => {
test("persists private generated authority before enrollment and active before dialing", async () => {
relationship = await HubRelationshipHarness.start();
const privateFileMode = platform() === "win32" ? 0o666 : 0o600;
relationship.holdEnrollment();
const connecting = relationship.beginConnect("one-time-token");
@@ -105,7 +107,6 @@ describe("Hub relationship", () => {
const pending = relationship.enrollmentInvocation();
expect(pending).toMatchObject({
mode: 0o600,
record: {
state: "pending",
relationship: { id: enrollment.relationshipId, idempotencyKey: enrollment.idempotencyKey },
@@ -114,6 +115,7 @@ describe("Hub relationship", () => {
identity: { serverId: expect.any(String), daemonPublicKey: expect.any(String) },
},
});
expect(pending.mode).toBe(privateFileMode);
const secret = pending.record.credential?.secret;
expect(secret).toEqual(expect.any(String));
expect(enrollment.credentialVerifier).toBe(
@@ -125,7 +127,7 @@ describe("Hub relationship", () => {
await connecting.result;
await relationship.socketDialed();
expect(relationship.socketInvocation()).toMatchObject({
mode: 0o600,
mode: privateFileMode,
record: { state: "active", relationship: { id: enrollment.relationshipId } },
});
});
@@ -146,6 +148,30 @@ describe("Hub relationship", () => {
expect(relationship.enrolledRelationships()).toBe(1);
});
test("a fresh token replaces the token for a pending enrollment ceremony", async () => {
relationship = await HubRelationshipHarness.start();
relationship.holdEnrollment();
const firstConnect = relationship.beginConnect("expired-token");
const firstResult = expect(firstConnect.result).resolves.toMatchObject({
state: "reconnecting",
});
await relationship.enrollmentBegins();
relationship.loseEnrollmentResponse();
await firstResult;
relationship.holdEnrollment();
const secondConnect = relationship.beginConnect("fresh-token");
const secondResult = expect(secondConnect.result).resolves.toMatchObject({
state: "connecting",
});
const retried = await relationship.enrollmentBegins();
expect(retried.token).toBe("fresh-token");
expect(relationship.relationshipFile()?.enrollment?.token).toBe("fresh-token");
relationship.completeEnrollment();
await secondResult;
});
test("a rejected pending enrollment is discarded without blocking daemon restart", async () => {
relationship = await HubRelationshipHarness.start();
relationship.holdEnrollment();
@@ -211,6 +211,8 @@ export class HubRelationshipController implements HubRelationshipManagement {
if (normalizeHubUrl(input.hubUrl) !== this.record.relationship.hubOrigin) {
throw new Error("A pending Hub enrollment already exists for a different Hub");
}
this.record = { ...this.record, enrollment: { token: input.token } };
this.persist(this.record);
await this.tryEnrollment(this.record);
return this.status();
}
@@ -54,16 +54,79 @@ test("a failed Hub create removes its auto-created worktree", async () => {
expect(await hub.durableOwnedAgentIds()).toEqual([]);
});
test("a Hub create fails when its initial prompt cannot start", async () => {
test("failed Hub auto-archive creates release their lifecycle subscriptions", async () => {
const hub = await launchRelationship();
hub.failNextProviderPromptStart();
hub.beginOwnedCreate("failed-prompt-create", "failed-prompt-execution");
const subscriptionBaseline = hub.agentSubscriptionCount();
const response = await hub.ownedCreateResult("failed-prompt-create");
hub.failNextProviderPromptStart();
hub.beginOwnedCreate("failed-prompt-create-1", "failed-prompt-execution-1", {
autoArchive: true,
worktree: { mode: "branch-off", newBranch: "failed-prompt-1" },
});
const first = await hub.ownedCreateResult("failed-prompt-create-1");
expect(first).toMatchObject({
type: "hub.agent.create.response",
payload: { success: false, executionId: "failed-prompt-execution-1" },
});
expect(hub.activeOwnedAgentIds()).toEqual([]);
expect(await hub.durableOwnedAgentIds()).toEqual([]);
expect(await hub.listedWorktrees()).toHaveLength(1);
expect(hub.agentSubscriptionCount()).toBe(subscriptionBaseline);
hub.failNextProviderPromptStart();
hub.beginOwnedCreate("failed-prompt-create-2", "failed-prompt-execution-2", {
autoArchive: true,
worktree: { mode: "branch-off", newBranch: "failed-prompt-2" },
});
const second = await hub.ownedCreateResult("failed-prompt-create-2");
expect(second).toMatchObject({
type: "hub.agent.create.response",
payload: { success: false, executionId: "failed-prompt-execution-2" },
});
expect(hub.activeOwnedAgentIds()).toEqual([]);
expect(await hub.durableOwnedAgentIds()).toEqual([]);
expect(await hub.listedWorktrees()).toHaveLength(1);
expect(hub.agentSubscriptionCount()).toBe(subscriptionBaseline);
});
test("Hub checkout uses the requested branch ref", async () => {
const hub = await launchRelationship();
await hub.createBranch("existing-hub-branch");
hub.beginOwnedCreate("checkout-create", "checkout-execution", {
worktree: { mode: "checkout-branch", branch: "existing-hub-branch" },
});
const response = await hub.ownedCreateResult("checkout-create");
expect(response).toMatchObject({
type: "hub.agent.create.response",
payload: { success: false, executionId: "failed-prompt-execution" },
payload: { success: true, executionId: "checkout-execution" },
});
expect(await hub.durableOwnedAgentIds()).toEqual([]);
expect(await hub.currentBranch(hub.latestCreatedCwd()!)).toBe("existing-hub-branch");
});
test("failed create never archives a reused worktree", async () => {
const hub = await launchRelationship();
hub.beginOwnedCreate("original-create", "original-execution", {
worktree: { mode: "branch-off", newBranch: "shared-hub-worktree" },
});
const original = await hub.ownedCreateResult("original-create");
const worktreeCwd = original.payload.agent?.cwd;
expect(worktreeCwd).toEqual(expect.any(String));
await hub.ownedTurnCompletion(original.payload.agentId!);
hub.failNextProviderPromptStart();
hub.beginOwnedCreate("reused-create", "reused-execution", {
autoArchive: true,
worktree: { mode: "branch-off", newBranch: "shared-hub-worktree" },
});
const reused = await hub.ownedCreateResult("reused-create");
expect(reused).toMatchObject({
type: "hub.agent.create.response",
payload: { success: false, executionId: "reused-execution" },
});
expect(await hub.worktreeState(worktreeCwd!)).toEqual({ exists: true, listed: true });
});
@@ -6,6 +6,7 @@ import type {
import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js";
import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js";
import type { LifecycleRegistration } from "../agent/create-agent-lifecycle-dispatch.js";
import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js";
import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js";
import { buildStoredAgentPayload } from "../agent/agent-projections.js";
@@ -49,7 +50,7 @@ interface RelationshipOwnedExecutionsOptions {
registerAutoArchive?: (input: {
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}) => void;
}) => LifecycleRegistration;
cleanupFailedCreate?: (input: {
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
createdAgentId: string | null;
@@ -71,7 +72,7 @@ export class RelationshipOwnedExecutions implements HubExecutions {
private readonly registerAutoArchive: (input: {
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}) => void;
}) => LifecycleRegistration;
private readonly cleanupFailedCreate: NonNullable<
RelationshipOwnedExecutionsOptions["cleanupFailedCreate"]
>;
@@ -81,7 +82,8 @@ export class RelationshipOwnedExecutions implements HubExecutions {
this.agentManager = options.agentManager;
this.agentStorage = options.agentStorage;
this.createAgentCommand = options.createAgent;
this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined);
this.registerAutoArchive =
options.registerAutoArchive ?? (() => ({ cancel: async () => undefined }));
this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined);
}
@@ -130,6 +132,7 @@ export class RelationshipOwnedExecutions implements HubExecutions {
let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null;
let createdAgentId: string | null = null;
let autoArchiveRegistration: LifecycleRegistration = { cancel: async () => undefined };
let result: Awaited<ReturnType<BoundCreateAgentCommand>>;
try {
result = await this.createAgentCommand({
@@ -153,15 +156,26 @@ export class RelationshipOwnedExecutions implements HubExecutions {
},
onCreated: (created) => {
createdAgentId = created.agentId;
if (input.autoArchive === true) this.registerAutoArchive(created);
if (input.autoArchive === true) {
autoArchiveRegistration = this.registerAutoArchive({
...created,
createdWorktree: ownedCreatedWorktree(created.createdWorktree),
});
}
},
});
} catch (error) {
if (createdAgentId) {
await autoArchiveRegistration.cancel();
if (createdAgentId && this.agentManager.getAgent(createdAgentId)) {
await this.agentManager.closeAgent(createdAgentId);
}
await this.cleanupFailedCreate({
createdWorktree: ownedCreatedWorktree(createdWorktree),
createdAgentId: null,
});
if (createdAgentId) {
await this.agentStorage.remove(createdAgentId);
}
await this.cleanupFailedCreate({ createdWorktree, createdAgentId: null });
throw error;
}
@@ -237,6 +251,12 @@ export class RelationshipOwnedExecutions implements HubExecutions {
}
}
function ownedCreatedWorktree(
worktree: CreatePaseoWorktreeWorkflowResult | null,
): CreatePaseoWorktreeWorkflowResult | null {
return worktree?.created === true ? worktree : null;
}
function toCreateAgentWorktree(target: CreateAgentWorktreeTarget | undefined) {
if (!target) return undefined;
if (target.mode === "branch-off") {
@@ -247,7 +267,7 @@ function toCreateAgentWorktree(target: CreateAgentWorktreeTarget | undefined) {
};
}
if (target.mode === "checkout-branch") {
return { branchName: target.branch, action: "checkout" as const };
return { refName: target.branch, action: "checkout" as const };
}
return { githubPrNumber: target.prNumber, action: "checkout" as const };
}
@@ -3,7 +3,7 @@ import { execFileSync } from "node:child_process";
import { EventEmitter } from "node:events";
import { existsSync, readFileSync, statSync, watch } from "node:fs";
import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
import { networkInterfaces, tmpdir } from "node:os";
import { networkInterfaces, platform, tmpdir } from "node:os";
import path from "node:path";
import { promisify } from "node:util";
import { Writable } from "node:stream";
@@ -424,6 +424,8 @@ export class HubRelationshipHarness {
private host = "";
private readonly logs: string[] = [];
private readonly providerPrompts: AgentPromptInput[] = [];
private readonly cliProcesses = new Set<Promise<unknown>>();
private readonly claimedCliSockets = new Set<WebSocket>();
private failNextPromptStart = false;
private observedEnrollments = 0;
private observedSockets = 0;
@@ -495,7 +497,7 @@ export class HubRelationshipHarness {
socket.send(JSON.stringify({ type: "session", message }));
responses.push((await this.nextSocketEnvelope(socket)).message);
}
socket.close();
await this.closeClaimedCliSocket(socket);
return responses;
}
@@ -517,7 +519,7 @@ export class HubRelationshipHarness {
}),
);
const response = (await this.nextSocketEnvelope(socket)).message;
socket.close();
await this.closeClaimedCliSocket(socket);
return response;
}
@@ -636,6 +638,16 @@ export class HubRelationshipHarness {
.map((record) => record.id);
}
activeOwnedAgentIds(): string[] {
return this.daemon!.agentManager.listAgents()
.filter((agent) => agent.owner?.kind === "hub")
.map((agent) => agent.id);
}
agentSubscriptionCount(): number {
return this.daemon!.agentManager.subscriptionCount();
}
async hubExecutionIntentFiles(): Promise<string[]> {
const directory = path.join(this.paseoHome, "hub-executions");
return existsSync(directory) ? readdir(directory) : [];
@@ -751,13 +763,39 @@ export class HubRelationshipHarness {
const listedPaths = stdout
.split("\n")
.filter((line) => line.startsWith("worktree "))
.map((line) => line.slice("worktree ".length));
.map((line) => this.comparablePath(line.slice("worktree ".length)));
return {
exists: existsSync(worktreePath),
listed: listedPaths.includes(worktreePath),
listed: listedPaths.includes(this.comparablePath(worktreePath)),
};
}
async createBranch(branch: string): Promise<void> {
await execFileAsync("git", ["-C", this.root, "branch", branch]);
}
async currentBranch(cwd: string): Promise<string> {
const { stdout } = await execFileAsync("git", ["-C", cwd, "branch", "--show-current"]);
return stdout.trim();
}
async ownedPermissionRequest(agentId: string) {
const message = await this.latestSocket().socket.messageMatching(
(candidate): candidate is HubAgentStream =>
candidate.type === "hub.agent.stream" &&
candidate.payload.agentId === agentId &&
candidate.payload.event.type === "permission_requested",
);
if (message.payload.event.type !== "permission_requested") {
throw new Error(`Owned agent ${agentId} did not request permission`);
}
return message.payload.event.request;
}
async allowOwnedPermission(agentId: string, requestId: string): Promise<void> {
await this.daemon!.agentManager.respondToPermission(agentId, requestId, { behavior: "allow" });
}
async listedWorktrees(): Promise<string[]> {
const { stdout } = await execFileAsync("git", [
"-C",
@@ -1074,7 +1112,11 @@ export class HubRelationshipHarness {
async close(): Promise<void> {
await this.stopDaemon();
await rm(this.root, { recursive: true, force: true });
await Promise.allSettled(this.cliProcesses);
await Promise.all(
[...this.claimedCliSockets].map((socket) => this.closeClaimedCliSocket(socket)),
);
await this.removeRoot();
}
private async createHome(): Promise<void> {
@@ -1130,7 +1172,11 @@ export class HubRelationshipHarness {
this.daemon = null;
}
private async runCli(args: string[]): Promise<Record<string, unknown>> {
private runCli(args: string[]): Promise<Record<string, unknown>> {
return this.trackCli(this.executeCli(args));
}
private async executeCli(args: string[]): Promise<Record<string, unknown>> {
const entrypoint = path.join(import.meta.dirname, "../../test-utils/hub-cli-entry.ts");
const { stdout } = await execFileAsync(
process.execPath,
@@ -1151,6 +1197,15 @@ export class HubRelationshipHarness {
return parsed as Record<string, unknown>;
}
private trackCli<T>(process: Promise<T>): Promise<T> {
this.cliProcesses.add(process);
void process.then(
() => this.cliProcesses.delete(process),
() => this.cliProcesses.delete(process),
);
return process;
}
private nextSocketEnvelope(socket: WebSocket): Promise<{ message: SessionOutboundMessage }> {
return new Promise((resolve, reject) => {
socket.once("message", (data) => {
@@ -1169,6 +1224,7 @@ export class HubRelationshipHarness {
options: { origin?: string } = {},
): Promise<WebSocket> {
const socket = new WebSocket(url, options);
this.claimedCliSockets.add(socket);
await new Promise<void>((resolve, reject) => {
socket.once("open", resolve);
socket.once("error", reject);
@@ -1185,6 +1241,32 @@ export class HubRelationshipHarness {
return socket;
}
private async closeClaimedCliSocket(socket: WebSocket): Promise<void> {
this.claimedCliSockets.delete(socket);
if (socket.readyState === WebSocket.CLOSED) return;
const closed = new Promise<void>((resolve) => socket.once("close", () => resolve()));
socket.terminate();
await closed;
}
private comparablePath(value: string): string {
const normalized = path.normalize(path.resolve(value));
return platform() === "win32" ? normalized.toLowerCase() : normalized;
}
private async removeRoot(): Promise<void> {
const attempts = platform() === "win32" ? 10 : 1;
for (let attempt = 1; attempt <= attempts; attempt++) {
try {
await rm(this.root, { recursive: true, force: true });
return;
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== "EBUSY" || attempt === attempts) throw error;
await new Promise((resolve) => setTimeout(resolve, 100 * attempt));
}
}
}
private latestSocket(): SocketAttempt {
const socket = this.remote.sockets.at(-1);
if (!socket) throw new Error("No Hub socket has been opened");