diff --git a/packages/server/src/server/hub/hub-session.test.ts b/packages/server/src/server/hub/hub-session.test.ts index 3518d2505a..6a0981f09b 100644 --- a/packages/server/src/server/hub/hub-session.test.ts +++ b/packages/server/src/server/hub/hub-session.test.ts @@ -16,8 +16,10 @@ import { HubSession } from "./hub-session.js"; class InMemoryHubExecutions implements HubExecutions { private readonly listeners = new Set<(event: OwnedAgentEvent) => void>(); private reconcileError: Error | null = null; + private creates = 0; async create(_input: HubAgentCreateInput): Promise { + this.creates++; return { executionId: "execution-1", agent: createAgentSnapshot() }; } @@ -31,6 +33,8 @@ class InMemoryHubExecutions implements HubExecutions { return () => this.listeners.delete(listener); } + async invalidateAuthority(): Promise {} + publishCurrentState(): void { for (const listener of this.listeners) { listener({ type: "update", executionId: "execution-1", agent: createAgentSnapshot() }); @@ -40,6 +44,10 @@ class InMemoryHubExecutions implements HubExecutions { failReconcile(error: Error): void { this.reconcileError = error; } + + createCount(): number { + return this.creates; + } } class HubSessionBoundary { @@ -67,6 +75,21 @@ class HubSessionBoundary { }); } + async createWith(input: { prompt: string; cwd: string }): Promise { + await this.session.handleMessage({ + type: "hub.agent.create.request", + requestId: "create-1", + executionId: "execution-1", + provider: "codex", + workspaceId: "workspace-1", + ...input, + }); + } + + createCount(): number { + return this.executions.createCount(); + } + deliveredMessages(): SessionOutboundMessage[] { return this.messages.slice(); } @@ -104,6 +127,30 @@ test("reconcile storage failures receive a terminal empty response", async () => ]); }); +test.each([ + { field: "prompt", prompt: " ", cwd: "/workspace" }, + { field: "cwd", prompt: "Do the work", cwd: "\t" }, +])("blank Hub $field values reject the create before execution", async ({ field, prompt, cwd }) => { + const hub = new HubSessionBoundary(); + + await hub.createWith({ prompt, cwd }); + + expect(hub.createCount()).toBe(0); + expect(hub.deliveredMessages()).toEqual([ + { + type: "hub.agent.create.response", + payload: { + requestId: "create-1", + executionId: "execution-1", + agentId: null, + agent: null, + success: false, + error: `Hub agent ${field} cannot be blank`, + }, + }, + ]); +}); + function createAgentSnapshot(): AgentSnapshotPayload { return AgentSnapshotPayloadSchema.parse({ id: "agent-1", diff --git a/packages/server/src/server/hub/hub-session.ts b/packages/server/src/server/hub/hub-session.ts index 6e80bdc8de..e9b6b3c633 100644 --- a/packages/server/src/server/hub/hub-session.ts +++ b/packages/server/src/server/hub/hub-session.ts @@ -49,6 +49,8 @@ export class HubSession { private async createAgent(message: HubAgentCreateRequest): Promise { try { + requireNonBlankHubAgentField("prompt", message.prompt); + requireNonBlankHubAgentField("cwd", message.cwd); const result = await this.executions.create({ executionId: message.executionId, provider: message.provider, @@ -131,6 +133,12 @@ export class HubSession { } } +function requireNonBlankHubAgentField(field: "prompt" | "cwd", value: string): void { + if (value.trim().length === 0) { + throw new Error(`Hub agent ${field} cannot be blank`); + } +} + function requestIdForMessage(message: SessionInboundMessage): string | undefined { if ("requestId" in message && typeof message.requestId === "string") { return message.requestId; diff --git a/packages/server/src/server/hub/index.ts b/packages/server/src/server/hub/index.ts deleted file mode 100644 index 2deb939266..0000000000 --- a/packages/server/src/server/hub/index.ts +++ /dev/null @@ -1,8 +0,0 @@ -export { HubSession } from "./hub-session.js"; -export { RelationshipOwnedExecutions } from "./relationship-owned-executions.js"; -export type { - HubAgentCreateInput, - HubExecutions, - OwnedAgentEvent, - OwnedAgentSnapshot, -} from "./relationship-owned-executions.js"; diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index 48ff6b1797..d8a8d9e13b 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -282,6 +282,35 @@ describe("Hub relationship", () => { expect(relationship.socketAttempts()).toBe(0); }); + test("a persisted WebSocket transport with a fragment is quarantined before startup", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.corruptRelationshipFile( + JSON.stringify({ + version: 1, + state: "active", + relationship: { + id: "relationship-1", + idempotencyKey: "ceremony-1", + hubOrigin: "https://hub.test", + createdAt: "2026-07-13T00:00:00.000Z", + scopes: ["hub.*"], + }, + credential: { secret: "credential" }, + transport: { + kind: "direct_websocket", + webSocketUrl: "wss://hub.test/daemon#fragment", + }, + }), + ); + + await relationship.startStoppedDaemon(); + + expect(await relationship.status()).toMatchObject({ state: "not_connected" }); + expect(relationship.relationshipFile()).toBeNull(); + expect(await relationship.quarantinedRelationshipFiles()).toHaveLength(1); + expect(relationship.socketAttempts()).toBe(0); + }); + test("disconnect revokes an ambiguous pending enrollment before removing local authority", async () => { relationship = await HubRelationshipHarness.start(); relationship.holdEnrollment(); @@ -549,6 +578,49 @@ describe("Hub relationship", () => { expect(durableAgentIds).toHaveLength(1); }); + test("reconcile waits for a pending create across socket generations", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + relationship.holdAgentCreation(); + relationship.beginOwnedCreate("pending-create", "pending-execution"); + await relationship.agentCreationAttempts(1); + + relationship.closeLatestSocket(1006); + await relationship.retry(); + relationship.connectLatestSocket(); + const reconciliation = relationship.reconcileOwned("pending-execution"); + relationship.finishAgentCreation(); + + await expect(reconciliation).resolves.toMatchObject({ + executionId: "pending-execution", + agentId: expect.any(String), + agent: { id: expect.any(String) }, + }); + expect(relationship.providerCreations()).toBe(1); + expect(await relationship.durableOwnedAgentIds()).toHaveLength(1); + }); + + test("force disconnect fences a pending create before removing authority", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + relationship.holdAgentCreation(); + relationship.beginOwnedCreate("cancelled-create", "cancelled-execution", { + worktree: { mode: "branch-off", newBranch: "cancelled-hub-create" }, + }); + await relationship.agentCreationAttempts(1); + + const disconnecting = relationship.beginDisconnect(true); + await relationship.relationshipStateBecomes(null); + relationship.finishAgentCreation(); + await disconnecting.result; + + expect(relationship.activeOwnedAgentIds()).toEqual([]); + expect(await relationship.durableOwnedAgentIds()).toEqual([]); + expect(await relationship.listedWorktrees()).toHaveLength(1); + }); + test.each([ [4403, "Hub revoked this relationship"], [401, "Hub rejected socket authentication (401)"], @@ -610,6 +682,18 @@ describe("Hub relationship", () => { expect(relationship.relationshipFile()).toBeNull(); }); + test("successful disconnect clears a transient revocation error", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.failRevocations(1); + + const disconnecting = await relationship.disconnect(); + await relationship.retry(); + + expect(disconnecting).toMatchObject({ state: "disconnecting", error: expect.any(String) }); + expect(await relationship.status()).toMatchObject({ state: "not_connected", error: null }); + }); + test("force disconnect removes local authority and reports the remote warning", 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 55b90932a8..23a8916f64 100644 --- a/packages/server/src/server/hub/relationship-controller.ts +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -31,7 +31,8 @@ const TransportSchema = z.object({ webSocketUrl: z .string() .url() - .refine((value) => ["ws:", "wss:"].includes(new URL(value).protocol)), + .refine((value) => ["ws:", "wss:"].includes(new URL(value).protocol)) + .refine((value) => new URL(value).hash === ""), }); const PendingSchema = z.object({ version: z.literal(1), @@ -262,15 +263,18 @@ export class HubRelationshipController implements HubRelationshipManagement { force: boolean; }): Promise<{ status: HubRelationshipStatus; warning?: string }> { const waitForEnrollment = this.record?.state === "pending"; + const pendingCreateCleanup = this.executions?.value.invalidateAuthority() ?? Promise.resolve(); this.cancelLifecycle(); this.socket?.close(); this.socket = null; if (!this.record || this.record.state === "revoked") { this.remove(); + await pendingCreateCleanup; return { status: this.status() }; } if (input.force) { this.remove(); + await pendingCreateCleanup; return { status: this.status(), warning: "Local Hub credential removed; remote revocation may remain pending.", @@ -290,6 +294,7 @@ export class HubRelationshipController implements HubRelationshipManagement { await Promise.all(this.inFlightEnrollments); } await this.tryRevocation(disconnecting); + await pendingCreateCleanup; return { status: this.status() }; } @@ -453,6 +458,7 @@ export class HubRelationshipController implements HubRelationshipManagement { } private revoke(reason: string): void { + void this.executions?.value.invalidateAuthority(); this.cancelLifecycle(); if (!this.record) return; const revoked: RevokedRecord = { @@ -492,11 +498,13 @@ export class HubRelationshipController implements HubRelationshipManagement { } private remove(): void { + void this.executions?.value.invalidateAuthority(); this.cancelLifecycle(); rmSync(this.filePath, { force: true }); this.record = null; this.state = "not_connected"; this.connectedAt = null; + this.lastError = null; } private load(): HubRelationshipRecord | null { diff --git a/packages/server/src/server/hub/relationship-owned-executions.ts b/packages/server/src/server/hub/relationship-owned-executions.ts index e27c905eb9..f8f206fd5c 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.ts +++ b/packages/server/src/server/hub/relationship-owned-executions.ts @@ -61,6 +61,7 @@ export interface HubExecutions { create(input: HubAgentCreateInput): Promise; reconcile(executionId: string): Promise; subscribe(listener: (event: OwnedAgentEvent) => void): () => void; + invalidateAuthority(): Promise; } export class RelationshipOwnedExecutions implements HubExecutions { @@ -69,6 +70,8 @@ export class RelationshipOwnedExecutions implements HubExecutions { private readonly agentStorage: AgentStorage; private readonly createAgentCommand: BoundCreateAgentCommand; private readonly pendingCreates = new Map>(); + private authorityGeneration = 0; + private authorityActive = true; private readonly registerAutoArchive: (input: { agentId: string; createdWorktree: CreatePaseoWorktreeWorkflowResult | null; @@ -88,6 +91,9 @@ export class RelationshipOwnedExecutions implements HubExecutions { } create(input: HubAgentCreateInput): Promise { + if (!this.authorityActive) { + return Promise.reject(new Error("Hub relationship authority is no longer active")); + } const owner = this.owner(input.executionId); const key = hubExecutionKey(owner); const pending = this.pendingCreates.get(key); @@ -95,7 +101,8 @@ export class RelationshipOwnedExecutions implements HubExecutions { return pending; } - const create = this.createOrResolve(owner, input).finally(() => { + const authorityGeneration = this.authorityGeneration; + const create = this.createOrResolve(owner, input, authorityGeneration).finally(() => { if (this.pendingCreates.get(key) === create) { this.pendingCreates.delete(key); } @@ -105,10 +112,19 @@ export class RelationshipOwnedExecutions implements HubExecutions { } async reconcile(executionId: string): Promise { - const record = await this.agentStorage.findByHubExecution(this.owner(executionId)); + const owner = this.owner(executionId); + const pending = this.pendingCreates.get(hubExecutionKey(owner)); + if (pending) return pending; + const record = await this.agentStorage.findByHubExecution(owner); return record ? this.resolveRecord(record) : null; } + async invalidateAuthority(): Promise { + this.authorityActive = false; + this.authorityGeneration++; + await Promise.allSettled(this.pendingCreates.values()); + } + subscribe(listener: (event: OwnedAgentEvent) => void): () => void { return this.agentManager.subscribe( (event) => { @@ -124,11 +140,14 @@ export class RelationshipOwnedExecutions implements HubExecutions { private async createOrResolve( owner: HubAgentOwner, input: HubAgentCreateInput, + authorityGeneration: number, ): Promise { const existing = await this.agentStorage.findByHubExecution(owner); if (existing) { + this.requireAuthority(authorityGeneration); return this.resolveRecord(existing); } + this.requireAuthority(authorityGeneration); let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null; let createdAgentId: string | null = null; @@ -164,6 +183,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { } }, }); + this.requireAuthority(authorityGeneration); } catch (error) { try { await autoArchiveRegistration.cancel(); @@ -195,6 +215,12 @@ export class RelationshipOwnedExecutions implements HubExecutions { return this.projectRecord(record); } + private requireAuthority(authorityGeneration: number): void { + if (!this.authorityActive || authorityGeneration !== this.authorityGeneration) { + throw new Error("Hub relationship authority ended during agent creation"); + } + } + private projectRecord(record: StoredAgentRecord): OwnedAgentSnapshot { const owner = this.requireOwner(record); const live = this.agentManager.getAgent(record.id); diff --git a/packages/server/src/server/hub/relationship-remote.test.ts b/packages/server/src/server/hub/relationship-remote.test.ts index 280f6a7841..af1c03639e 100644 --- a/packages/server/src/server/hub/relationship-remote.test.ts +++ b/packages/server/src/server/hub/relationship-remote.test.ts @@ -91,6 +91,24 @@ test("enrollment rejects a transport URL that cannot open a WebSocket", async () ).rejects.toThrow("Hub WebSocket URL must use ws or wss"); }); +test("enrollment rejects a WebSocket URL with a fragment", async () => { + const hubOrigin = await startEnrollmentHub("ws://hub.test/daemon#fragment"); + const remote = new DirectHubRelationshipRemote(); + + await expect( + remote.enroll({ + relationshipId: "relationship-1", + idempotencyKey: "ceremony-1", + hubOrigin, + token: "token", + serverId: "server-1", + daemonPublicKey: "public-key", + credentialVerifier: "verifier", + scopes: ["hub.*"], + }), + ).rejects.toThrow("Hub WebSocket URL cannot include a fragment"); +}); + test("enrollment rejects a WebSocket outside the enrolled Hub authority", async () => { const hubOrigin = await startEnrollmentHub("ws://other-hub.test/daemon"); const remote = new DirectHubRelationshipRemote(); @@ -467,6 +485,7 @@ const unusedExecutions: HubExecutions = { }, reconcile: async () => null, subscribe: () => () => undefined, + invalidateAuthority: async () => undefined, }; async function connectController( diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts index 37e1fbb95e..fcc22f6266 100644 --- a/packages/server/src/server/hub/relationship-remote.ts +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -63,6 +63,9 @@ const EnrollmentResultSchema = z.object({ .url() .refine((value) => ["ws:", "wss:"].includes(new URL(value).protocol), { message: "Hub WebSocket URL must use ws or wss", + }) + .refine((value) => new URL(value).hash === "", { + message: "Hub WebSocket URL cannot include a fragment", }), }); 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 2d98cde428..b7d3b3f8d1 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -499,12 +499,12 @@ export class HubRelationshipHarness { return { result: this.runCli(["hub", "disconnect", ...(force ? ["--force"] : [])]) }; } - async relationshipStateBecomes(expected: string): Promise { + async relationshipStateBecomes(expected: string | null): Promise { const observed = deferred(); const watcher = watch(this.paseoHome, () => { - if (this.relationshipFile()?.state === expected) observed.resolve(); + if ((this.relationshipFile()?.state ?? null) === expected) observed.resolve(); }); - if (this.relationshipFile()?.state === expected) observed.resolve(); + if ((this.relationshipFile()?.state ?? null) === expected) observed.resolve(); try { await observed.promise; } finally { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index dee65a001a..5bcceb22a8 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -36,8 +36,9 @@ import { type SessionPeer, type SessionRuntimeMetrics, } from "./session.js"; -import { HubSession, type HubExecutions } from "./hub/index.js"; +import { HubSession } from "./hub/hub-session.js"; import type { HubRelationshipManagement } from "./hub/relationship-controller.js"; +import type { HubExecutions } from "./hub/relationship-owned-executions.js"; import type { AgentProvider } from "./agent/agent-sdk-types.js"; import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";