fix(hub): harden relationship lifecycle

This commit is contained in:
Mohamed Boudra
2026-07-13 11:40:42 +02:00
parent 056b2a47d9
commit 51de9660f9
15 changed files with 213 additions and 23 deletions
+21 -1
View File
@@ -76,7 +76,7 @@ function createMockTransport() {
return {
transport,
sent,
triggerOpen: (options?: { preserveSent?: boolean }) => {
triggerOpen: (options?: { preserveSent?: boolean; features?: Record<string, boolean> }) => {
onOpen();
if (!options?.preserveSent) {
// Ignore HELLO handshake payloads in assertions.
@@ -92,6 +92,7 @@ function createMockTransport() {
serverId: `srv_test_${serverInfoOrdinal++}`,
hostname: null,
version: null,
features: options?.features,
},
},
}),
@@ -219,6 +220,25 @@ test("advertises consumer-provided browser automation capabilities", async () =>
});
});
test("Hub management requires daemon support before dispatching requests", async () => {
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "hub_feature_gate_unit_test",
transportFactory: () => mock.transport,
reconnect: { enabled: false },
});
clients.push(client);
const connecting = client.connect();
mock.triggerOpen();
await connecting;
await expect(client.getHubStatus()).rejects.toThrow(
"Update the host to use Hub relationship management.",
);
expect(mock.sent).toEqual([]);
});
const noopLogger: Logger = {
debug: () => {},
info: () => {},
+10
View File
@@ -3930,6 +3930,7 @@ export class DaemonClient {
}
async connectHub(hubUrl: string, token: string, requestId?: string) {
this.requireHubRelationshipSupport();
return this.sendCorrelatedSessionRequest({
requestId,
message: { type: "hub.relationship.connect.request", hubUrl, token },
@@ -3938,6 +3939,7 @@ export class DaemonClient {
}
async getHubStatus(requestId?: string) {
this.requireHubRelationshipSupport();
return this.sendCorrelatedSessionRequest({
requestId,
message: { type: "hub.relationship.get_status.request" },
@@ -3946,6 +3948,7 @@ export class DaemonClient {
}
async disconnectHub(force = false, requestId?: string) {
this.requireHubRelationshipSupport();
return this.sendCorrelatedSessionRequest({
requestId,
message: { type: "hub.relationship.disconnect.request", force },
@@ -4773,6 +4776,13 @@ export class DaemonClient {
return this.lastServerInfoMessage;
}
private requireHubRelationshipSupport(): void {
// COMPAT(hubRelationship): added in v0.1.X, drop the gate when floor >= v0.1.X.
if (this.lastServerInfoMessage?.features?.hubRelationship !== true) {
throw new Error("Update the host to use Hub relationship management.");
}
}
private resolveTransportUrlForAttempt(): string {
return this.config.url;
}
+2
View File
@@ -2467,6 +2467,8 @@ export const ServerInfoStatusPayloadSchema = z
providerSubagents: z.boolean().optional(),
// COMPAT(workspacePinning): added in v0.1.107, remove gate after 2027-01-12.
workspacePinning: z.boolean().optional(),
// COMPAT(hubRelationship): added in v0.1.X, drop the gate when floor >= v0.1.X.
hubRelationship: z.boolean().optional(),
})
.optional(),
})
@@ -104,6 +104,7 @@ export interface CreateAgentFromMcpInput {
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}) => void;
onWorktreeCreated?: (createdWorktree: CreatePaseoWorktreeWorkflowResult) => void;
callerAgentId?: string;
callerContext?: {
lockedCwd?: string;
@@ -288,6 +289,7 @@ async function resolveMcpCreateAgent(
worktree: input.worktree,
initialPrompt: input.initialPrompt ?? "",
});
if (createdWorktree) input.onWorktreeCreated?.(createdWorktree);
const workspaceId = await resolveMcpWorkspaceId({
dependencies,
+2
View File
@@ -1053,6 +1053,8 @@ export async function createPaseoDaemon(
agentId,
createdWorktree,
}),
cleanupFailedCreate: (input) =>
hubAgentLifecycle.cleanupCreatedWorktreeAfterFailedAgentCreate(input),
}),
});
@@ -903,6 +903,7 @@ test("receives server_info on websocket connect", async () => {
expect(serverInfo).not.toBeNull();
expect(serverInfo?.serverId.length).toBeGreaterThan(0);
expect(serverInfo?.features?.["terminal-restore-modes"]).toBe(true);
expect(serverInfo?.features?.hubRelationship).toBe(true);
await client.close();
}, 15000);
@@ -135,6 +135,21 @@ describe("Hub relationship", () => {
expect(relationship.enrolledRelationships()).toBe(1);
});
test("a rejected pending enrollment is discarded without blocking daemon restart", async () => {
relationship = await HubRelationshipHarness.start();
relationship.holdEnrollment();
const connecting = relationship.beginConnect("expired-token");
await relationship.enrollmentBegins();
relationship.loseEnrollmentResponse();
await connecting.result;
relationship.rejectNextEnrollment(401);
await relationship.restartDaemon();
expect(await relationship.status()).toMatchObject({ state: "not_connected" });
expect(relationship.relationshipFile()).toBeNull();
});
test("daemon restart reconnects the same durable relationship", async () => {
relationship = await HubRelationshipHarness.start();
await relationship.beginConnect().result;
@@ -159,7 +159,17 @@ export class HubRelationshipController implements HubRelationshipManagement {
async start(): Promise<void> {
if (this.record?.state === "active") this.openSocket(this.record, false);
if (this.record?.state === "pending") await this.tryEnrollment(this.record);
if (this.record?.state === "pending") {
try {
await this.tryEnrollment(this.record);
} catch (error) {
if (!(error instanceof HubEnrollmentRejectedError)) throw error;
this.options.logger.warn(
{ statusCode: error.statusCode },
"Discarded rejected pending Hub enrollment during startup",
);
}
}
if (this.record?.state === "disconnecting") await this.tryRevocation(this.record);
}
@@ -36,3 +36,20 @@ test("removing an owned agent removes its reconstructed execution association",
expect(removed.reconciliation).toBeNull();
expect(removed.durableAgentCount).toBe(0);
});
test("a failed Hub create removes its auto-created worktree", async () => {
const hub = await launchRelationship();
hub.beginOwnedCreate("failed-worktree-create", "failed-worktree-execution", {
modeId: "missing-mode",
worktree: { mode: "branch-off", newBranch: "failed-hub-create" },
});
const response = await hub.ownedCreateResult("failed-worktree-create");
expect(response).toMatchObject({
type: "hub.agent.create.response",
payload: { success: false, executionId: "failed-worktree-execution" },
});
expect(await hub.listedWorktrees()).toHaveLength(1);
expect(await hub.durableOwnedAgentIds()).toEqual([]);
});
@@ -50,6 +50,10 @@ interface RelationshipOwnedExecutionsOptions {
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}) => void;
cleanupFailedCreate?: (input: {
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
createdAgentId: string | null;
}) => Promise<void>;
}
export interface HubExecutions {
@@ -68,6 +72,9 @@ export class RelationshipOwnedExecutions implements HubExecutions {
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}) => void;
private readonly cleanupFailedCreate: NonNullable<
RelationshipOwnedExecutionsOptions["cleanupFailedCreate"]
>;
constructor(options: RelationshipOwnedExecutionsOptions) {
this.relationshipId = options.relationshipId;
@@ -75,6 +82,7 @@ export class RelationshipOwnedExecutions implements HubExecutions {
this.agentStorage = options.agentStorage;
this.createAgentCommand = options.createAgent;
this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined);
this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined);
}
create(input: HubAgentCreateInput): Promise<OwnedAgentSnapshot> {
@@ -120,25 +128,37 @@ export class RelationshipOwnedExecutions implements HubExecutions {
return this.projectRecord(existing);
}
const result = await this.createAgentCommand({
kind: "mcp",
provider: input.model ? `${input.provider}/${input.model}` : input.provider,
title: input.prompt,
initialPrompt: input.prompt,
cwd: input.cwd,
workspaceId: input.workspaceId,
mode: input.modeId,
thinking: input.thinkingOptionId,
features: input.featureValues,
env: input.env,
worktree: toCreateAgentWorktree(input.worktree),
background: true,
notifyOnFinish: false,
owner,
...(input.autoArchive === true
? { onCreated: (created) => this.registerAutoArchive(created) }
: {}),
});
let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null;
let createdAgentId: string | null = null;
let result: Awaited<ReturnType<BoundCreateAgentCommand>>;
try {
result = await this.createAgentCommand({
kind: "mcp",
provider: input.model ? `${input.provider}/${input.model}` : input.provider,
title: input.prompt,
initialPrompt: input.prompt,
cwd: input.cwd,
workspaceId: input.workspaceId,
mode: input.modeId,
thinking: input.thinkingOptionId,
features: input.featureValues,
env: input.env,
worktree: toCreateAgentWorktree(input.worktree),
background: true,
notifyOnFinish: false,
owner,
onWorktreeCreated: (worktree) => {
createdWorktree = worktree;
},
onCreated: (created) => {
createdAgentId = created.agentId;
if (input.autoArchive === true) this.registerAutoArchive(created);
},
});
} catch (error) {
await this.cleanupFailedCreate({ createdWorktree, createdAgentId });
throw error;
}
return {
executionId: owner.executionId,
@@ -0,0 +1,56 @@
import { createServer } from "node:http";
import type { AddressInfo } from "node:net";
import { afterEach, expect, test } from "vitest";
import { DirectHubRelationshipRemote } from "./relationship-remote.js";
const openServers: ReturnType<typeof createServer>[] = [];
afterEach(async () => {
await Promise.all(openServers.splice(0).map((server) => closeServer(server)));
});
test.each([401, 403, 404])(
"revocation status %s clears already-invalid local authority",
async (status) => {
const hubOrigin = await startHubReturning(status);
const remote = new DirectHubRelationshipRemote();
await expect(
remote.revoke({ relationshipId: "relationship-1", hubOrigin, credential: "invalid" }),
).resolves.toBeUndefined();
},
);
test("transient revocation failures remain retryable", async () => {
const hubOrigin = await startHubReturning(503);
const remote = new DirectHubRelationshipRemote();
await expect(
remote.revoke({ relationshipId: "relationship-1", hubOrigin, credential: "credential" }),
).rejects.toThrow("Hub revocation failed (503)");
});
async function startHubReturning(status: number): Promise<string> {
const server = createServer((_request, response) => {
response.writeHead(status).end();
});
openServers.push(server);
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", resolve);
});
const address = server.address() as AddressInfo;
return `http://127.0.0.1:${address.port}`;
}
function closeServer(server: ReturnType<typeof createServer>): Promise<void> {
return new Promise((resolve, reject) => {
server.close((error) => {
if (error) {
reject(error);
return;
}
resolve();
});
});
}
@@ -92,7 +92,7 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote {
`${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`,
{ method: "DELETE", headers: { authorization: `Bearer ${input.credential}` } },
);
if (!response.ok && response.status !== 404) {
if (!response.ok && ![401, 403, 404].includes(response.status)) {
throw new Error(`Hub revocation failed (${response.status})`);
}
}
@@ -0,0 +1,4 @@
import { createHubCommand } from "../../../../../cli/src/commands/hub/index.js";
const argv = [process.argv[0] ?? "node", process.argv[1] ?? "paseo", ...process.argv.slice(3)];
await createHubCommand().parseAsync(argv, { from: "node" });
@@ -54,6 +54,7 @@ import type {
HubSocketCredentials,
HubSocketEvents,
} from "../relationship-remote.js";
import { HubEnrollmentRejectedError } from "../relationship-remote.js";
const execFileAsync = promisify(execFile);
const HUB_ORIGIN = "https://hub.test";
@@ -306,6 +307,7 @@ class InMemoryHubRelationships implements HubRelationshipRemote {
private enrollmentGate: Deferred<HubEnrollmentResult> | null = null;
private enrollmentObserved = deferred<void>();
private socketObserved = deferred<void>();
private enrollmentRejection: 401 | 403 | null = null;
private revokeFailures = 0;
private readonly relationships = new Set<string>();
readonly enrollmentSnapshots: RelationshipInvocationSnapshot[] = [];
@@ -317,11 +319,20 @@ class InMemoryHubRelationships implements HubRelationshipRemote {
this.enrollmentGate = deferred<HubEnrollmentResult>();
}
rejectNextEnrollment(statusCode: 401 | 403): void {
this.enrollmentRejection = statusCode;
}
async enroll(input: HubEnrollment): Promise<HubEnrollmentResult> {
this.enrollmentSnapshots.push(this.captureRelationship());
this.enrollments.push({ ...input, scopes: input.scopes.slice() });
this.relationships.add(input.idempotencyKey);
this.enrollmentObserved.resolve();
if (this.enrollmentRejection) {
const statusCode = this.enrollmentRejection;
this.enrollmentRejection = null;
throw new HubEnrollmentRejectedError(statusCode);
}
if (this.enrollmentGate) return this.enrollmentGate.promise;
return this.enrollmentResult(input);
}
@@ -506,6 +517,10 @@ export class HubRelationshipHarness {
this.remote.loseEnrollmentResponse();
}
rejectNextEnrollment(statusCode: 401 | 403): void {
this.remote.rejectNextEnrollment(statusCode);
}
failRevocations(count: number): void {
this.remote.failRevocations(count);
}
@@ -708,6 +723,20 @@ export class HubRelationshipHarness {
};
}
async listedWorktrees(): Promise<string[]> {
const { stdout } = await execFileAsync("git", [
"-C",
this.root,
"worktree",
"list",
"--porcelain",
]);
return stdout
.split("\n")
.filter((line) => line.startsWith("worktree "))
.map((line) => line.slice("worktree ".length));
}
async createOwnedConcurrently(executionId = "execution-1"): Promise<{
first: AcceptedCreate;
duplicate: AcceptedCreate;
@@ -1014,7 +1043,7 @@ export class HubRelationshipHarness {
}
private async runCli(args: string[]): Promise<Record<string, unknown>> {
const entrypoint = path.join(REPOSITORY_ROOT, "packages/cli/src/index.ts");
const entrypoint = path.join(import.meta.dirname, "hub-cli-entry.ts");
const { stdout } = await execFileAsync(
process.execPath,
["--import", "tsx", entrypoint, ...args, "--host", this.host, "--json"],
@@ -1342,6 +1342,8 @@ export class VoiceAssistantWebSocketServer {
providerSubagents: true,
// COMPAT(workspacePinning): added in v0.1.107, remove gate after 2027-01-12.
workspacePinning: true,
// COMPAT(hubRelationship): added in v0.1.X, drop the gate when floor >= v0.1.X.
hubRelationship: true,
},
};
}