test(hub): remove incidental observation and speculative shutdown changes

Hub control RPCs need a connected client without an agent observation. Preserve the failed-create subscription contract and remove product shutdown changes whose Windows cleanup cause was not reproduced.
This commit is contained in:
Mohamed Boudra
2026-09-15 01:21:36 +02:00
parent 7ec77c1b65
commit 94706b2d31
4 changed files with 21 additions and 79 deletions
+1 -1
View File
@@ -1779,7 +1779,7 @@ export async function createPaseoDaemon(
await pluginRuntime.stopAllPlugins();
unsubscribePluginProviders();
await hubRelationships.stop();
await workspaceReconciliation.dispose();
workspaceReconciliation.dispose();
scriptHealthMonitor.stop();
// Freeze both ingress and registration before taking the agent closure snapshot.
wsServer?.prepareForShutdown();
@@ -1602,7 +1602,6 @@ export class HubRelationshipHarness {
appVersion: "0.1.106",
});
await client.connect();
await client.observeAgents().ready;
return client;
}
@@ -1,4 +1,4 @@
import { mkdirSync, mkdtempSync, rmSync, watch as watchPath } from "node:fs";
import { mkdirSync, mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages";
@@ -30,52 +30,6 @@ afterEach(() => {
for (const target of cleanupPaths.splice(0)) rmSync(target, { recursive: true, force: true });
});
test("disposal releases a real project root watch before its directory is removed", async () => {
const root = mkdtempSync(path.join(tmpdir(), "project-watch-close-"));
cleanupPaths.push(root);
const logger = createTestLogger();
const projects = new FileBackedProjectRegistry(path.join(root, "projects.json"), logger);
const workspaces = new FileBackedWorkspaceRegistry(path.join(root, "workspaces.json"), logger);
await projects.initialize();
await workspaces.initialize();
await projects.upsert(
createPersistedProjectRecord({
projectId: "project-watch-close",
rootPath: root,
kind: "non_git",
displayName: "root watch",
createdAt: TIMESTAMP,
updatedAt: TIMESTAMP,
}),
);
let released = false;
const service = new WorkspaceReconciliationService({
projectRegistry: projects,
workspaceRegistry: workspaces,
logger,
watchProjectRoot: (rootPath, options, onChange, onError) => {
const watcher = watchPath(rootPath, options, onChange);
watcher.on("error", onError);
const closed = new Promise<void>((resolve) =>
watcher.once("close", () => {
released = true;
resolve();
}),
);
return {
close: () => {
watcher.close();
return closed;
},
};
},
});
await service.start();
await service.dispose();
expect(released).toBe(true);
rmSync(root, { recursive: true });
});
interface ProjectSpec {
id: string;
root: string;
@@ -234,11 +188,7 @@ class ObservedPlacements {
const watch = { rootPath, onChange, onError, closed: false };
this.watches.push(watch);
this.lifecycleEvents.push(`watch installed:${path.relative(this.home, rootPath)}`);
return {
close: async () => {
watch.closed = true;
},
};
return { close: () => (watch.closed = true) };
};
this.service = new WorkspaceReconciliationService({
projectRegistry: this.projects,
@@ -327,8 +277,8 @@ class ObservedPlacements {
await this.clock.advanceBy(elapsedMs);
}
async dispose(): Promise<void> {
await this.service.dispose();
dispose(): void {
this.service.dispose();
}
watchedRoots(): string[] {
@@ -457,7 +407,7 @@ describe("observed workspace placement", () => {
"registry mutation resolved:project-new",
]);
expect(observed.gitReads).toBe(0);
await observed.dispose();
observed.dispose();
});
test("deduplicates active root watches and tears them down on archive and remove", async () => {
@@ -484,7 +434,7 @@ describe("observed workspace placement", () => {
{ kind: "remove", projectId: "project-duplicate" },
{ kind: "remove", projectId: "project-remove" },
]);
await observed.dispose();
observed.dispose();
});
test("filters unrelated files and coalesces Git change bursts", async () => {
@@ -500,7 +450,7 @@ describe("observed workspace placement", () => {
observed.change("repo", null);
await observed.advanceBy(DEBOUNCE_MS);
expect(observed.gitReads).toBe(1);
await observed.dispose();
observed.dispose();
});
test("recovers errored and temporarily unavailable watchers on the periodic pass", async () => {
@@ -518,7 +468,7 @@ describe("observed workspace placement", () => {
expect(observed.watchedRoots()).toEqual(["errored", "unavailable"]);
expect(observed.closedRoots()).toEqual(["errored"]);
await observed.dispose();
observed.dispose();
});
test("archives missing workspace directories on the periodic pass", async () => {
@@ -532,7 +482,7 @@ describe("observed workspace placement", () => {
expect((await observed.placement("workspace-one"))?.archivedAt).toEqual(expect.any(String));
expect(observed.workspaceBatches).toEqual([["workspace-one"]]);
await observed.dispose();
observed.dispose();
});
test("preserves a periodic full pass queued behind metadata reconciliation", async () => {
@@ -554,7 +504,7 @@ describe("observed workspace placement", () => {
expect((await observed.placement("workspace-one"))?.archivedAt).toEqual(expect.any(String));
expect(observed.workspaceBatches).toEqual([["workspace-one"]]);
await observed.dispose();
observed.dispose();
});
test("contains a failed reconciliation and converges on the next change", async () => {
@@ -576,7 +526,7 @@ describe("observed workspace placement", () => {
branch: "main",
displayName: "Durable workspace-one",
});
await observed.dispose();
observed.dispose();
});
test("deduplicates direct placement and project-derived workspace fanout", async () => {
@@ -599,7 +549,7 @@ describe("observed workspace placement", () => {
expect(observed.projectUpdates).toHaveLength(1);
expect(observed.workspaceBatches).toEqual([["workspace-one", "workspace-two"]]);
expect(new Set(observed.workspaceBatches[0]).size).toBe(2);
await observed.dispose();
observed.dispose();
});
test("disposal suppresses in-flight mutations and reconciliation fanout", async () => {
@@ -608,7 +558,7 @@ describe("observed workspace placement", () => {
const registryRead = mutation.holdNextRegistryRead();
const adding = mutation.add({ id: "project-late", root: "late" });
await registryRead.started;
await mutation.dispose();
mutation.dispose();
registryRead.release();
await adding;
expect(mutation.projectUpdates).toEqual([]);
@@ -623,7 +573,7 @@ describe("observed workspace placement", () => {
reconciliation.change("repo", ".git");
const advancing = reconciliation.advanceBy(DEBOUNCE_MS);
await workspaceRead.started;
await reconciliation.dispose();
reconciliation.dispose();
workspaceRead.release();
await advancing;
@@ -25,7 +25,7 @@ export type ProjectUpdate =
| { kind: "remove"; projectId: string };
interface ProjectRootWatcher {
close(): Promise<void>;
close(): void;
}
export interface ProjectRootWatch {
@@ -58,13 +58,7 @@ const systemClock: ReconciliationClock = {
const watchProjectRoot: ProjectRootWatch = (rootPath, options, onChange, onError) => {
const watcher = watchPath(rootPath, options, onChange);
watcher.on("error", onError);
const closed = new Promise<void>((resolve) => watcher.once("close", resolve));
return {
close() {
watcher.close();
return closed;
},
};
return watcher;
};
export type ReconciliationChange =
@@ -184,15 +178,14 @@ export class WorkspaceReconciliationService {
this.rescanTimer.unref?.();
}
async dispose(): Promise<void> {
dispose(): void {
this.disposed = true;
this.unsubscribeRegistry?.();
this.unsubscribeRegistry = null;
if (this.rescanTimer) this.clock.clearInterval(this.rescanTimer);
if (this.debounceTimer) this.clock.clearTimeout(this.debounceTimer);
const closing = this.watchers.map(({ watcher }) => watcher.close());
for (const { watcher } of this.watchers) watcher.close();
this.watchers.length = 0;
await Promise.all(closing);
}
/** Reconciles mutable Git facts only; never archives missing records. */
@@ -420,7 +413,7 @@ export class WorkspaceReconciliationService {
areEquivalentPaths(project.rootPath, target.rootPath),
);
if (stillActive) continue;
void target.watcher.close();
target.watcher.close();
this.watchers.splice(index, 1);
}
@@ -440,7 +433,7 @@ export class WorkspaceReconciliationService {
}
},
(error) => {
void watcher.close();
watcher.close();
const index = this.watchers.findIndex((target) => target.watcher === watcher);
if (index >= 0) this.watchers.splice(index, 1);
this.logger.warn(