test(server): remove archive watcher race (#2868)
The archive response already waits for durable archival and owned-worktree cleanup. Assert those final states directly instead of racing a separate filesystem watcher with its own fixed timeout.
This commit is contained in:
@@ -1,8 +1,5 @@
|
||||
import { afterEach, expect, test } from "vitest";
|
||||
import {
|
||||
HubRelationshipHarness,
|
||||
SetupFailingArchiveWatchFiles,
|
||||
} from "./test-utils/relationship-harness.js";
|
||||
import { HubRelationshipHarness } from "./test-utils/relationship-harness.js";
|
||||
|
||||
let relationship: HubRelationshipHarness | null = null;
|
||||
|
||||
@@ -167,9 +164,7 @@ test("Hub archives a running execution's Paseo-created worktree", async () => {
|
||||
const worktreeCwd = hub.latestCreatedCwd();
|
||||
await hub.ownedRunningUpdate(worktreeCreated.payload.agentId!);
|
||||
const duringRun = await hub.worktreeState(worktreeCwd!);
|
||||
const archiveCompletion = hub.waitForOwnedArchiveCompletion(worktreeCreated.payload.agentId!);
|
||||
const response = await hub.archiveExecution("execution-worktree", "archive-worktree");
|
||||
const archive = await archiveCompletion;
|
||||
const afterArchive = await hub.worktreeState(worktreeCwd!);
|
||||
|
||||
expect(worktreeCreated).toMatchObject({
|
||||
@@ -180,11 +175,13 @@ test("Hub archives a running execution's Paseo-created worktree", async () => {
|
||||
expect(duringRun).toEqual({ exists: true, listed: true });
|
||||
expect(response).toMatchObject({ success: true, error: null, action: "archive" });
|
||||
expect(afterArchive).toEqual({ exists: false, listed: false });
|
||||
expect(archive).toEqual({
|
||||
agentArchivedAt: expect.any(String),
|
||||
workspaceArchivedAt: expect.any(String),
|
||||
});
|
||||
}, 30_000);
|
||||
expect(await hub.ownedAgentArchivedAt(worktreeCreated.payload.agentId!)).toEqual(
|
||||
expect.any(String),
|
||||
);
|
||||
expect(await hub.ownedWorkspaceArchivedAt(worktreeCreated.payload.agentId!)).toEqual(
|
||||
expect.any(String),
|
||||
);
|
||||
}, 20_000);
|
||||
|
||||
test("a sibling workspace keeps an archived execution's worktree directory alive", async () => {
|
||||
const hub = await launchRelationship();
|
||||
@@ -275,19 +272,3 @@ test("Hub treats missing and foreign executions as already controlled without ex
|
||||
]);
|
||||
expect(await hub.agentRemainsAvailable(foreignAgentId)).toBe(true);
|
||||
});
|
||||
|
||||
test("archive observation closes its first watcher when the second watcher cannot start", async () => {
|
||||
const watchFiles = new SetupFailingArchiveWatchFiles(2);
|
||||
const hub = await HubRelationshipHarness.start(watchFiles);
|
||||
relationship = hub;
|
||||
await hub.beginConnect().result;
|
||||
hub.connectLatestSocket();
|
||||
hub.beginOwnedCreate("watch-setup-create", "watch-setup-execution");
|
||||
const created = await hub.ownedCreateResult("watch-setup-create");
|
||||
|
||||
await expect(hub.waitForOwnedArchiveCompletion(created.payload.agentId!)).rejects.toThrow(
|
||||
"Cannot watch",
|
||||
);
|
||||
|
||||
expect(watchFiles.activeDirectories()).toEqual([]);
|
||||
});
|
||||
|
||||
@@ -63,53 +63,6 @@ const HUB_ORIGIN = "https://hub.test";
|
||||
const SOCKET_URL = "wss://hub.test/daemon";
|
||||
const REPOSITORY_ROOT = path.resolve(import.meta.dirname, "../../../../../..");
|
||||
|
||||
export interface ArchiveWatcher {
|
||||
close(): void;
|
||||
onError(listener: (error: Error) => void): void;
|
||||
}
|
||||
|
||||
export interface ArchiveWatchFiles {
|
||||
watchDirectory(path: string, onChange: () => void): ArchiveWatcher;
|
||||
}
|
||||
|
||||
const nodeArchiveWatchFiles: ArchiveWatchFiles = {
|
||||
watchDirectory(directory, onChange) {
|
||||
const watcher = watch(directory, onChange);
|
||||
return {
|
||||
close: () => watcher.close(),
|
||||
onError: (listener) => watcher.on("error", listener),
|
||||
};
|
||||
},
|
||||
};
|
||||
|
||||
export class SetupFailingArchiveWatchFiles implements ArchiveWatchFiles {
|
||||
private attempts = 0;
|
||||
private readonly openDirectories = new Set<string>();
|
||||
|
||||
constructor(private readonly failOnAttempt: number) {}
|
||||
|
||||
watchDirectory(directory: string): ArchiveWatcher {
|
||||
this.attempts++;
|
||||
if (this.attempts === this.failOnAttempt) {
|
||||
throw new Error(`Cannot watch ${directory}`);
|
||||
}
|
||||
this.openDirectories.add(directory);
|
||||
let open = true;
|
||||
return {
|
||||
close: () => {
|
||||
if (!open) return;
|
||||
open = false;
|
||||
this.openDirectories.delete(directory);
|
||||
},
|
||||
onError: () => {},
|
||||
};
|
||||
}
|
||||
|
||||
activeDirectories(): string[] {
|
||||
return [...this.openDirectories];
|
||||
}
|
||||
}
|
||||
|
||||
interface PersistedRelationship {
|
||||
version: number;
|
||||
state: string;
|
||||
@@ -465,10 +418,7 @@ export class HubRelationshipHarness {
|
||||
private observedSockets = 0;
|
||||
private readonly codex: ControlledAgentClient;
|
||||
|
||||
private constructor(
|
||||
private readonly archiveWatchFiles: ArchiveWatchFiles,
|
||||
private readonly mcpEnabled: boolean,
|
||||
) {
|
||||
private constructor(private readonly mcpEnabled: boolean) {
|
||||
this.codex = new ControlledAgentClient(
|
||||
createTestAgentClients({
|
||||
supportsMcpServers: mcpEnabled,
|
||||
@@ -488,23 +438,16 @@ export class HubRelationshipHarness {
|
||||
);
|
||||
}
|
||||
|
||||
static async start(
|
||||
archiveWatchFiles: ArchiveWatchFiles = nodeArchiveWatchFiles,
|
||||
): Promise<HubRelationshipHarness> {
|
||||
return this.launch(archiveWatchFiles, false);
|
||||
static async start(): Promise<HubRelationshipHarness> {
|
||||
return this.launch(false);
|
||||
}
|
||||
|
||||
static async startWithAgentMcp(
|
||||
archiveWatchFiles: ArchiveWatchFiles = nodeArchiveWatchFiles,
|
||||
): Promise<HubRelationshipHarness> {
|
||||
return this.launch(archiveWatchFiles, true);
|
||||
static async startWithAgentMcp(): Promise<HubRelationshipHarness> {
|
||||
return this.launch(true);
|
||||
}
|
||||
|
||||
private static async launch(
|
||||
archiveWatchFiles: ArchiveWatchFiles,
|
||||
mcpEnabled: boolean,
|
||||
): Promise<HubRelationshipHarness> {
|
||||
const harness = new HubRelationshipHarness(archiveWatchFiles, mcpEnabled);
|
||||
private static async launch(mcpEnabled: boolean): Promise<HubRelationshipHarness> {
|
||||
const harness = new HubRelationshipHarness(mcpEnabled);
|
||||
await harness.createHome();
|
||||
await harness.startDaemon();
|
||||
return harness;
|
||||
@@ -795,6 +738,13 @@ export class HubRelationshipHarness {
|
||||
return (await this.daemon!.agentStorage.get(agentId))?.archivedAt ?? null;
|
||||
}
|
||||
|
||||
async ownedWorkspaceArchivedAt(agentId: string): Promise<string | null> {
|
||||
const agent = await this.daemon!.agentStorage.get(agentId);
|
||||
if (!agent) throw new Error(`Owned agent ${agentId} does not exist`);
|
||||
if (!agent.workspaceId) throw new Error(`Owned agent ${agentId} has no workspace`);
|
||||
return this.workspaceArchivedAt(agent.workspaceId);
|
||||
}
|
||||
|
||||
async createForeignExecution(executionId: string): Promise<string> {
|
||||
const agent = await this.daemon!.agentManager.createAgent(
|
||||
{ provider: "codex", cwd: this.root },
|
||||
@@ -866,65 +816,6 @@ export class HubRelationshipHarness {
|
||||
return existsSync(this.root);
|
||||
}
|
||||
|
||||
async waitForOwnedArchiveCompletion(
|
||||
agentId: string,
|
||||
): Promise<{ agentArchivedAt: string; workspaceArchivedAt: string }> {
|
||||
const created = await this.daemon!.agentStorage.get(agentId);
|
||||
if (!created) throw new Error(`Owned agent ${agentId} does not exist`);
|
||||
if (!created.workspaceId) throw new Error(`Owned agent ${agentId} has no workspace`);
|
||||
return new Promise<{ agentArchivedAt: string; workspaceArchivedAt: string }>(
|
||||
(resolve, reject) => {
|
||||
const watchers: ArchiveWatcher[] = [];
|
||||
let settled = false;
|
||||
const timeout = setTimeout(
|
||||
() => finish(new Error(`Timed out waiting for owned agent ${agentId} to archive`)),
|
||||
15_000,
|
||||
);
|
||||
timeout.unref?.();
|
||||
const closeWatchers = () => {
|
||||
clearTimeout(timeout);
|
||||
for (const watcher of watchers) watcher.close();
|
||||
};
|
||||
const finish = (
|
||||
result: Error | { agentArchivedAt: string; workspaceArchivedAt: string },
|
||||
) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
closeWatchers();
|
||||
if (result instanceof Error) reject(result);
|
||||
else resolve(result);
|
||||
};
|
||||
const observeCompletion = async () => {
|
||||
try {
|
||||
const archived = await this.daemon!.agentStorage.get(agentId);
|
||||
const workspaceArchivedAt = this.workspaceArchivedAt(created.workspaceId!);
|
||||
if (!archived?.archivedAt || !workspaceArchivedAt || existsSync(created.cwd)) return;
|
||||
finish({ agentArchivedAt: archived.archivedAt, workspaceArchivedAt });
|
||||
} catch (error) {
|
||||
finish(error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
};
|
||||
try {
|
||||
const projectsWatcher = this.archiveWatchFiles.watchDirectory(
|
||||
path.join(this.paseoHome, "projects"),
|
||||
observeCompletion,
|
||||
);
|
||||
watchers.push(projectsWatcher);
|
||||
projectsWatcher.onError(finish);
|
||||
const worktreeWatcher = this.archiveWatchFiles.watchDirectory(
|
||||
path.dirname(created.cwd),
|
||||
observeCompletion,
|
||||
);
|
||||
watchers.push(worktreeWatcher);
|
||||
worktreeWatcher.onError(finish);
|
||||
void observeCompletion();
|
||||
} catch (error) {
|
||||
finish(error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private workspaceArchivedAt(workspaceId: string): string | null {
|
||||
const records = JSON.parse(
|
||||
readFileSync(path.join(this.paseoHome, "projects", "workspaces.json"), "utf8"),
|
||||
|
||||
Reference in New Issue
Block a user