From 51de9660f95dab86bf63c0f904f9f84cde3041e8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 13 Jul 2026 11:38:39 +0200 Subject: [PATCH] fix(hub): harden relationship lifecycle --- packages/client/src/daemon-client.test.ts | 22 ++++++- packages/client/src/daemon-client.ts | 10 ++++ packages/protocol/src/messages.ts | 2 + .../src/server/agent/create-agent/create.ts | 2 + packages/server/src/server/bootstrap.ts | 2 + .../src/server/daemon-client.e2e.test.ts | 1 + .../hub/relationship-controller.test.ts | 15 +++++ .../src/server/hub/relationship-controller.ts | 12 +++- .../hub/relationship-owned-executions.test.ts | 17 ++++++ .../hub/relationship-owned-executions.ts | 58 +++++++++++++------ .../server/hub/relationship-remote.test.ts | 56 ++++++++++++++++++ .../src/server/hub/relationship-remote.ts | 2 +- .../server/hub/test-utils/hub-cli-entry.ts | 4 ++ .../hub/test-utils/relationship-harness.ts | 31 +++++++++- .../server/src/server/websocket-server.ts | 2 + 15 files changed, 213 insertions(+), 23 deletions(-) create mode 100644 packages/server/src/server/hub/relationship-remote.test.ts create mode 100644 packages/server/src/server/hub/test-utils/hub-cli-entry.ts diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index a636fa4d5a..68e0002b6d 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -76,7 +76,7 @@ function createMockTransport() { return { transport, sent, - triggerOpen: (options?: { preserveSent?: boolean }) => { + triggerOpen: (options?: { preserveSent?: boolean; features?: Record }) => { 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: () => {}, diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index d448e9fe52..7ac4cbe47b 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -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; } diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index d88f9e3a2c..6da9061627 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -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(), }) diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index e8feef4d2f..a8d3687bb5 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -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, diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index c5f43739d6..c34f5b6b2f 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1053,6 +1053,8 @@ export async function createPaseoDaemon( agentId, createdWorktree, }), + cleanupFailedCreate: (input) => + hubAgentLifecycle.cleanupCreatedWorktreeAfterFailedAgentCreate(input), }), }); diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index c66c8068b6..c22354986f 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -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); diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index 5c07d983e1..918666e51a 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -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; diff --git a/packages/server/src/server/hub/relationship-controller.ts b/packages/server/src/server/hub/relationship-controller.ts index 8682c35733..c7a8201e2f 100644 --- a/packages/server/src/server/hub/relationship-controller.ts +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -159,7 +159,17 @@ export class HubRelationshipController implements HubRelationshipManagement { async start(): Promise { 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); } diff --git a/packages/server/src/server/hub/relationship-owned-executions.test.ts b/packages/server/src/server/hub/relationship-owned-executions.test.ts index e2320f61b2..099c8f5220 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.test.ts +++ b/packages/server/src/server/hub/relationship-owned-executions.test.ts @@ -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([]); +}); diff --git a/packages/server/src/server/hub/relationship-owned-executions.ts b/packages/server/src/server/hub/relationship-owned-executions.ts index 659c4f66ce..441637ebd8 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.ts +++ b/packages/server/src/server/hub/relationship-owned-executions.ts @@ -50,6 +50,10 @@ interface RelationshipOwnedExecutionsOptions { agentId: string; createdWorktree: CreatePaseoWorktreeWorkflowResult | null; }) => void; + cleanupFailedCreate?: (input: { + createdWorktree: CreatePaseoWorktreeWorkflowResult | null; + createdAgentId: string | null; + }) => Promise; } 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 { @@ -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>; + 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, diff --git a/packages/server/src/server/hub/relationship-remote.test.ts b/packages/server/src/server/hub/relationship-remote.test.ts new file mode 100644 index 0000000000..d6617a7abb --- /dev/null +++ b/packages/server/src/server/hub/relationship-remote.test.ts @@ -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[] = []; + +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 { + const server = createServer((_request, response) => { + response.writeHead(status).end(); + }); + openServers.push(server); + await new Promise((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): Promise { + return new Promise((resolve, reject) => { + server.close((error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); +} diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts index d2bbd24c6f..07aa6dacf1 100644 --- a/packages/server/src/server/hub/relationship-remote.ts +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -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})`); } } diff --git a/packages/server/src/server/hub/test-utils/hub-cli-entry.ts b/packages/server/src/server/hub/test-utils/hub-cli-entry.ts new file mode 100644 index 0000000000..5bb419d265 --- /dev/null +++ b/packages/server/src/server/hub/test-utils/hub-cli-entry.ts @@ -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" }); diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index 05666d67a1..1c20cb208d 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -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 | null = null; private enrollmentObserved = deferred(); private socketObserved = deferred(); + private enrollmentRejection: 401 | 403 | null = null; private revokeFailures = 0; private readonly relationships = new Set(); readonly enrollmentSnapshots: RelationshipInvocationSnapshot[] = []; @@ -317,11 +319,20 @@ class InMemoryHubRelationships implements HubRelationshipRemote { this.enrollmentGate = deferred(); } + rejectNextEnrollment(statusCode: 401 | 403): void { + this.enrollmentRejection = statusCode; + } + async enroll(input: HubEnrollment): Promise { 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 { + 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> { - 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"], diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index d220a08de0..d5b0015ef7 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -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, }, }; }