fix(hub): harden relationship lifecycle

Keep optional Hub authority from blocking daemon startup, revoke ambiguous enrollments durably, and close owned agents when their relationship no longer exists. Reject remote CLI connect targets before transmitting enrollment authority.

packages/server/src/server/bootstrap.ts

@@ -1520,7 +1520,9 @@ export async function createPaseoDaemon(
     // Freeze both ingress and registration before taking the agent closure snapshot.
     wsServer?.prepareForShutdown();
     agentManager.prepareForShutdown();
-    await closeAllAgents(logger, agentManager);
+    await closeAllAgents(logger, agentManager, (relationshipId) =>
+      hubRelationships.hasReconnectableRelationship(relationshipId),
+    );
     await agentManager.flushForShutdown().catch(() => undefined);
     detachAgentStoragePersistence();
     await agentStorage.flush().catch(() => undefined);
@@ -1564,13 +1566,19 @@ export async function createPaseoDaemon(
   };
 }
 
-async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promise<void> {
+async function closeAllAgents(
+  logger: Logger,
+  agentManager: AgentManager,
+  hasReconnectableHubRelationship: (relationshipId: string) => boolean,
+): Promise<void> {
   const agents = agentManager.listAgents();
   await Promise.all(
     agents.map(async (agent) => {
       try {
         await agentManager.closeAgent(agent.id, {
-          persistClosedState: agent.owner?.kind !== "hub",
+          persistClosedState:
+            agent.owner?.kind !== "hub" ||
+            !hasReconnectableHubRelationship(agent.owner.relationshipId),
         });
       } catch (err) {
         logger.error({ err, agentId: agent.id }, "Failed to close agent");

packages/server/src/server/hub/relationship-controller.test.ts

@@ -150,6 +150,48 @@ describe("Hub relationship", () => {
     expect(relationship.relationshipFile()).toBeNull();
   });
 
+  test.each(["{not-json", JSON.stringify({ version: 1, state: "unknown" })])(
+    "an invalid relationship file is quarantined without blocking daemon startup",
+    async (contents) => {
+      relationship = await HubRelationshipHarness.start();
+      await relationship.corruptRelationshipFile(contents);
+
+      await relationship.startStoppedDaemon();
+
+      expect(await relationship.status()).toMatchObject({ state: "not_connected" });
+      expect(relationship.relationshipFile()).toBeNull();
+      expect(await relationship.quarantinedRelationshipFiles()).toHaveLength(1);
+    },
+  );
+
+  test("disconnect revokes an ambiguous pending enrollment before removing local authority", async () => {
+    relationship = await HubRelationshipHarness.start();
+    relationship.holdEnrollment();
+    const connecting = relationship.beginConnect("one-time-token");
+    const enrollment = await relationship.enrollmentBegins();
+    const credential = relationship.relationshipFile()?.credential?.secret;
+    relationship.loseEnrollmentResponse();
+    await connecting.result;
+    relationship.failRevocations(2);
+
+    const disconnected = await relationship.disconnect();
+    expect(disconnected.state).toBe("disconnecting");
+    expect(relationship.relationshipFile()?.state).toBe("disconnecting");
+
+    await relationship.restartDaemon();
+    await relationship.retry();
+
+    expect(relationship.revocationAttempts()).toBe(3);
+    expect(relationship.latestRevocation()).toEqual(
+      expect.objectContaining({
+        relationshipId: enrollment.relationshipId,
+        hubOrigin: enrollment.hubOrigin,
+        credential,
+      }),
+    );
+    expect(relationship.relationshipFile()).toBeNull();
+  });
+
   test("daemon restart reconnects the same durable relationship", async () => {
     relationship = await HubRelationshipHarness.start();
     await relationship.beginConnect().result;
@@ -427,4 +469,18 @@ describe("Hub relationship", () => {
     expect(forced.warning).toContain("remote revocation");
     expect(relationship.relationshipFile()).toBeNull();
   });
+
+  test("daemon shutdown closes owned agents after their Hub relationship is removed", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    relationship.connectLatestSocket();
+    relationship.beginOwnedCreate("orphan-create", "execution-orphan", { prompt: "sleep 30" });
+    const created = await relationship.ownedCreateResult("orphan-create");
+    expect(created).toMatchObject({ payload: { agent: { status: "running" } } });
+
+    await relationship.disconnect(true);
+    await relationship.restartDaemon();
+
+    expect(await relationship.storedOwnedStatus(created.payload.agentId!)).toBe("closed");
+  });
 });

packages/server/src/server/hub/relationship-controller.ts

@@ -1,5 +1,5 @@
 import { createHash, randomBytes, randomUUID } from "node:crypto";
-import { existsSync, readFileSync, rmSync } from "node:fs";
+import { existsSync, readFileSync, renameSync, rmSync } from "node:fs";
 import path from "node:path";
 import type pino from "pino";
 import { z } from "zod";
@@ -40,11 +40,18 @@ const PendingSchema = z.object({
 });
 const ActiveSchema = z.object({
   version: z.literal(1),
-  state: z.enum(["active", "disconnecting"]),
+  state: z.literal("active"),
   relationship: RelationshipSchema,
   credential: CredentialSchema,
   transport: TransportSchema,
 });
+const DisconnectingSchema = z.object({
+  version: z.literal(1),
+  state: z.literal("disconnecting"),
+  relationship: RelationshipSchema,
+  credential: CredentialSchema,
+  transport: TransportSchema.optional(),
+});
 const RevokedSchema = z.object({
   version: z.literal(1),
   state: z.literal("revoked"),
@@ -52,9 +59,15 @@ const RevokedSchema = z.object({
   transport: TransportSchema.optional(),
   reason: z.string().optional(),
 });
-const RecordSchema = z.discriminatedUnion("state", [PendingSchema, ActiveSchema, RevokedSchema]);
+const RecordSchema = z.discriminatedUnion("state", [
+  PendingSchema,
+  ActiveSchema,
+  DisconnectingSchema,
+  RevokedSchema,
+]);
 type PendingRecord = z.infer<typeof PendingSchema>;
 type ActiveRecord = z.infer<typeof ActiveSchema>;
+type DisconnectingRecord = z.infer<typeof DisconnectingSchema>;
 type RevokedRecord = z.infer<typeof RevokedSchema>;
 type HubRelationshipRecord = z.infer<typeof RecordSchema>;
 
@@ -190,6 +203,10 @@ export class HubRelationshipController implements HubRelationshipManagement {
     };
   }
 
+  hasReconnectableRelationship(relationshipId: string): boolean {
+    return this.record?.state === "active" && this.record.relationship.id === relationshipId;
+  }
+
   async connect(input: { hubUrl: string; token: string }): Promise<HubRelationshipStatus> {
     if (this.record?.state === "pending") {
       if (normalizeHubUrl(input.hubUrl) !== this.record.relationship.hubOrigin) {
@@ -239,11 +256,13 @@ export class HubRelationshipController implements HubRelationshipManagement {
         warning: "Local Hub credential removed; remote revocation may remain pending.",
       };
     }
-    if (this.record.state === "pending") {
-      this.remove();
-      return { status: this.status() };
-    }
-    const disconnecting: ActiveRecord = { ...this.record, state: "disconnecting" };
+    const disconnecting: DisconnectingRecord = {
+      version: 1,
+      state: "disconnecting",
+      relationship: this.record.relationship,
+      credential: this.record.credential,
+      ...(this.record.state === "active" ? { transport: this.record.transport } : {}),
+    };
     this.persist(disconnecting);
     this.record = disconnecting;
     this.state = "disconnecting";
@@ -365,7 +384,7 @@ export class HubRelationshipController implements HubRelationshipManagement {
     this.schedule(() => void this.tryEnrollment(record));
   }
 
-  private async tryRevocation(record: ActiveRecord): Promise<void> {
+  private async tryRevocation(record: DisconnectingRecord): Promise<void> {
     const generation = this.generation;
     try {
       await this.options.remote.revoke({
@@ -433,7 +452,22 @@ export class HubRelationshipController implements HubRelationshipManagement {
 
   private load(): HubRelationshipRecord | null {
     if (!existsSync(this.filePath)) return null;
-    const record = RecordSchema.parse(JSON.parse(readFileSync(this.filePath, "utf8")));
+    let record: HubRelationshipRecord;
+    try {
+      record = RecordSchema.parse(JSON.parse(readFileSync(this.filePath, "utf8")));
+    } catch (error) {
+      const quarantinePath = path.join(
+        path.dirname(this.filePath),
+        `hub-relationship.invalid-${this.clock.now().getTime()}-${randomUUID()}.json`,
+      );
+      renameSync(this.filePath, quarantinePath);
+      ensurePrivateFile(quarantinePath);
+      this.options.logger.error(
+        { err: error, quarantinePath },
+        "Quarantined invalid Hub relationship authority",
+      );
+      return null;
+    }
     ensurePrivateFile(this.filePath);
     return record;
   }

packages/server/src/server/hub/relationship-remote.test.ts

@@ -50,6 +50,31 @@ test("transient revocation failures remain retryable", async () => {
   ).rejects.toThrow("Hub revocation failed (503)");
 });
 
+test.each(["enrollment", "revocation"])("%s HTTP calls are bounded", async (operation) => {
+  const hubOrigin = await startStalledHub();
+  const remote = new DirectHubRelationshipRemote({ requestTimeoutMs: 25 });
+
+  const request =
+    operation === "enrollment"
+      ? remote.enroll({
+          relationshipId: "relationship-1",
+          idempotencyKey: "ceremony-1",
+          hubOrigin,
+          token: "token",
+          serverId: "server-1",
+          daemonPublicKey: "public-key",
+          credentialVerifier: "verifier",
+          scopes: ["hub.*"],
+        })
+      : remote.revoke({
+          relationshipId: "relationship-1",
+          hubOrigin,
+          credential: "credential",
+        });
+
+  await expect(request).rejects.toThrow("Hub request timed out");
+});
+
 test.each([401, 403] as const)(
   "socket authentication status %s is terminal and releases the failed upgrade",
   async (status) => {
@@ -164,6 +189,17 @@ async function startHubReturning(status: number): Promise<string> {
   return `http://127.0.0.1:${address.port}`;
 }
 
+async function startStalledHub(): Promise<string> {
+  const server = createServer(() => undefined);
+  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) => {

packages/server/src/server/hub/relationship-remote.ts

@@ -62,39 +62,54 @@ const EnrollmentResultSchema = z.object({
 });
 
 export class DirectHubRelationshipRemote implements HubRelationshipRemote {
+  private readonly requestTimeoutMs: number;
+
+  constructor(options: { requestTimeoutMs?: number } = {}) {
+    this.requestTimeoutMs = options.requestTimeoutMs ?? 15_000;
+  }
+
   async enroll(input: HubEnrollment): Promise<HubEnrollmentResult> {
-    const response = await fetch(`${input.hubOrigin}/api/daemon-relationships/enroll`, {
-      method: "POST",
-      headers: {
-        "content-type": "application/json",
-        authorization: `Bearer ${input.token}`,
-      },
-      body: JSON.stringify({
-        relationshipId: input.relationshipId,
-        idempotencyKey: input.idempotencyKey,
-        serverId: input.serverId,
-        daemonPublicKey: input.daemonPublicKey,
-        credentialVerifier: input.credentialVerifier,
-        scopes: input.scopes,
-      }),
-    });
-    if (!response.ok) {
-      if (response.status >= 400 && response.status < 500) {
-        throw new HubEnrollmentRejectedError(response.status);
+    return this.withRequestTimeout(async (signal) => {
+      const response = await fetch(`${input.hubOrigin}/api/daemon-relationships/enroll`, {
+        method: "POST",
+        headers: {
+          "content-type": "application/json",
+          authorization: `Bearer ${input.token}`,
+        },
+        body: JSON.stringify({
+          relationshipId: input.relationshipId,
+          idempotencyKey: input.idempotencyKey,
+          serverId: input.serverId,
+          daemonPublicKey: input.daemonPublicKey,
+          credentialVerifier: input.credentialVerifier,
+          scopes: input.scopes,
+        }),
+        signal,
+      });
+      if (!response.ok) {
+        if (response.status >= 400 && response.status < 500) {
+          throw new HubEnrollmentRejectedError(response.status);
+        }
+        throw new Error(`Hub enrollment failed (${response.status})`);
       }
-      throw new Error(`Hub enrollment failed (${response.status})`);
-    }
-    return EnrollmentResultSchema.parse(await response.json());
+      return EnrollmentResultSchema.parse(await response.json());
+    });
   }
 
   async revoke(input: HubRevocation): Promise<void> {
-    const response = await fetch(
-      `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`,
-      { method: "DELETE", headers: { authorization: `Bearer ${input.credential}` } },
-    );
-    if (!response.ok && ![401, 403, 404].includes(response.status)) {
-      throw new Error(`Hub revocation failed (${response.status})`);
-    }
+    await this.withRequestTimeout(async (signal) => {
+      const response = await fetch(
+        `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`,
+        {
+          method: "DELETE",
+          headers: { authorization: `Bearer ${input.credential}` },
+          signal,
+        },
+      );
+      if (!response.ok && ![401, 403, 404].includes(response.status)) {
+        throw new Error(`Hub revocation failed (${response.status})`);
+      }
+    });
   }
 
   openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection {
@@ -134,4 +149,18 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote {
     });
     return socket;
   }
+
+  private async withRequestTimeout<T>(operation: (signal: AbortSignal) => Promise<T>): Promise<T> {
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(), this.requestTimeoutMs);
+    timeout.unref?.();
+    try {
+      return await operation(controller.signal);
+    } catch (error) {
+      if (controller.signal.aborted) throw new Error("Hub request timed out", { cause: error });
+      throw error;
+    } finally {
+      clearTimeout(timeout);
+    }
+  }
 }

packages/server/src/server/hub/test-utils/relationship-harness.ts

@@ -2,7 +2,7 @@ import { execFile } from "node:child_process";
 import { execFileSync } from "node:child_process";
 import { EventEmitter } from "node:events";
 import { existsSync, readFileSync, statSync, watch } from "node:fs";
-import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
 import { networkInterfaces, tmpdir } from "node:os";
 import path from "node:path";
 import { promisify } from "node:util";
@@ -993,6 +993,10 @@ export class HubRelationshipHarness {
     return this.remote.revocations.length;
   }
 
+  latestRevocation(): HubRevocation | null {
+    return this.remote.revocations.at(-1) ?? null;
+  }
+
   enrollmentAttempts(): HubEnrollment[] {
     return this.remote.enrollments.map((input) => ({ ...input, scopes: input.scopes.slice() }));
   }
@@ -1027,6 +1031,25 @@ export class HubRelationshipHarness {
     return statSync(path.join(this.paseoHome, "hub-relationship.json")).mode & 0o777;
   }
 
+  async corruptRelationshipFile(contents = "{not-json"): Promise<void> {
+    await this.stopDaemon();
+    await writeFile(path.join(this.paseoHome, "hub-relationship.json"), contents, "utf8");
+  }
+
+  async quarantinedRelationshipFiles(): Promise<string[]> {
+    return (await readdir(this.paseoHome)).filter((file) =>
+      file.startsWith("hub-relationship.invalid-"),
+    );
+  }
+
+  async startStoppedDaemon(): Promise<void> {
+    await this.startDaemon();
+  }
+
+  async storedOwnedStatus(agentId: string): Promise<string | null> {
+    return (await this.daemon!.agentStorage.get(agentId))?.lastStatus ?? null;
+  }
+
   private captureRelationship(): RelationshipInvocationSnapshot {
     const record = this.relationshipFile();
     if (!record) throw new Error("Relationship authority was not persisted before invocation");