diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index c34f5b6b2f..70927ad0c8 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1047,6 +1047,7 @@ export async function createPaseoDaemon( agentManager, agentStorage, createAgent, + logger, registerAutoArchive: ({ agentId, createdWorktree }) => hubAgentLifecycle.registerAutoArchiveIfRequested({ autoArchive: true, diff --git a/packages/server/src/server/hub/hub-session.test.ts b/packages/server/src/server/hub/hub-session.test.ts index e3763de874..3518d2505a 100644 --- a/packages/server/src/server/hub/hub-session.test.ts +++ b/packages/server/src/server/hub/hub-session.test.ts @@ -15,12 +15,14 @@ import { HubSession } from "./hub-session.js"; class InMemoryHubExecutions implements HubExecutions { private readonly listeners = new Set<(event: OwnedAgentEvent) => void>(); + private reconcileError: Error | null = null; async create(_input: HubAgentCreateInput): Promise { return { executionId: "execution-1", agent: createAgentSnapshot() }; } async reconcile(_executionId: string): Promise { + if (this.reconcileError) throw this.reconcileError; return { executionId: "execution-1", agent: createAgentSnapshot() }; } @@ -34,6 +36,10 @@ class InMemoryHubExecutions implements HubExecutions { listener({ type: "update", executionId: "execution-1", agent: createAgentSnapshot() }); } } + + failReconcile(error: Error): void { + this.reconcileError = error; + } } class HubSessionBoundary { @@ -52,6 +58,19 @@ class HubSessionBoundary { this.session.cleanup(); } + async failReconcile(): Promise { + this.executions.failReconcile(new Error("storage unavailable")); + await this.session.handleMessage({ + type: "hub.execution.reconcile.request", + requestId: "reconcile-1", + executionId: "execution-1", + }); + } + + deliveredMessages(): SessionOutboundMessage[] { + return this.messages.slice(); + } + deliveredTypes(): string[] { return this.messages.map((message) => message.type); } @@ -67,6 +86,24 @@ test("disconnect unsubscribes Hub from later owned-agent updates", () => { expect(hub.deliveredTypes()).toEqual(["hub.agent.update"]); }); +test("reconcile storage failures receive a terminal empty response", async () => { + const hub = new HubSessionBoundary(); + + await hub.failReconcile(); + + expect(hub.deliveredMessages()).toEqual([ + { + type: "hub.execution.reconcile.response", + payload: { + requestId: "reconcile-1", + executionId: "execution-1", + agentId: null, + agent: null, + }, + }, + ]); +}); + 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 8998193cb5..6e80bdc8de 100644 --- a/packages/server/src/server/hub/hub-session.ts +++ b/packages/server/src/server/hub/hub-session.ts @@ -90,7 +90,13 @@ export class HubSession { } private async reconcile(message: HubExecutionReconcileRequest): Promise { - const result = await this.executions.reconcile(message.executionId); + let result: Awaited> = null; + try { + result = await this.executions.reconcile(message.executionId); + } catch { + // Reconcile has no error variant on the wire. A terminal empty response + // lets Hub stop waiting and decide whether to replay the execution. + } this.send({ type: "hub.execution.reconcile.response", payload: { diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index 918666e51a..d83c68bf6a 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -192,10 +192,11 @@ describe("Hub relationship", () => { expect(reconciled).toMatchObject({ executionId: "execution-running", agentId: created.payload.agentId, - agent: { id: created.payload.agentId, status: "running" }, + agent: { id: created.payload.agentId, status: "idle" }, }); expect(durableAgentIds).toEqual([created.payload.agentId]); expect(relationship.providerCreations()).toBe(1); + expect(relationship.providerResumes()).toBe(1); }); test("stale socket generations cannot replace or unregister the current socket", async () => { diff --git a/packages/server/src/server/hub/relationship-controller.ts b/packages/server/src/server/hub/relationship-controller.ts index c7a8201e2f..e060f84fad 100644 --- a/packages/server/src/server/hub/relationship-controller.ts +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -301,9 +301,7 @@ export class HubRelationshipController implements HubRelationshipManagement { connected: (socket) => this.socketConnected(generation, record, socket), rejected: (statusCode) => this.socketRejected(generation, statusCode), closed: (code) => this.socketClosed(generation, record, code), - failed: (error) => { - if (generation === this.generation) this.lastError = error.message; - }, + failed: (error) => this.socketFailed(generation, record, error), }; this.socket = this.options.remote.openSocket( { @@ -351,6 +349,12 @@ export class HubRelationshipController implements HubRelationshipManagement { if (this.record?.state === "active") this.scheduleSocket(record); } + private socketFailed(generation: number, record: ActiveRecord, error: Error): void { + if (generation !== this.generation) return; + this.lastError = error.message; + if (this.record?.state === "active") this.scheduleSocket(record); + } + private scheduleSocket(record: ActiveRecord): void { this.state = "reconnecting"; this.schedule(() => this.openSocket(record, true)); diff --git a/packages/server/src/server/hub/relationship-owned-executions.ts b/packages/server/src/server/hub/relationship-owned-executions.ts index 441637ebd8..804536da43 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.ts +++ b/packages/server/src/server/hub/relationship-owned-executions.ts @@ -3,9 +3,11 @@ import type { AgentStreamEventPayload, CreateAgentWorktreeTarget, } from "@getpaseo/protocol/messages"; +import type { Logger } from "pino"; import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js"; import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js"; +import { ensureAgentLoaded } from "../agent/agent-loading.js"; import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js"; import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js"; import { buildStoredAgentPayload } from "../agent/agent-projections.js"; @@ -46,6 +48,7 @@ interface RelationshipOwnedExecutionsOptions { agentManager: AgentManager; agentStorage: AgentStorage; createAgent: BoundCreateAgentCommand; + logger: Logger; registerAutoArchive?: (input: { agentId: string; createdWorktree: CreatePaseoWorktreeWorkflowResult | null; @@ -67,6 +70,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { private readonly agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly createAgentCommand: BoundCreateAgentCommand; + private readonly logger: Logger; private readonly pendingCreates = new Map>(); private readonly registerAutoArchive: (input: { agentId: string; @@ -81,6 +85,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { this.agentManager = options.agentManager; this.agentStorage = options.agentStorage; this.createAgentCommand = options.createAgent; + this.logger = options.logger; this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined); this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined); } @@ -104,7 +109,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { async reconcile(executionId: string): Promise { const record = await this.agentStorage.findByHubExecution(this.owner(executionId)); - return record ? this.projectRecord(record) : null; + return record ? this.resolveRecord(record) : null; } subscribe(listener: (event: OwnedAgentEvent) => void): () => void { @@ -125,7 +130,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { ): Promise { const existing = await this.agentStorage.findByHubExecution(owner); if (existing) { - return this.projectRecord(existing); + return this.resolveRecord(existing); } let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null; @@ -166,6 +171,21 @@ export class RelationshipOwnedExecutions implements HubExecutions { }; } + private async resolveRecord(record: StoredAgentRecord): Promise { + if ( + !this.agentManager.getAgent(record.id) && + !record.archivedAt && + record.lastStatus !== "closed" + ) { + await ensureAgentLoaded(record.id, { + agentManager: this.agentManager, + agentStorage: this.agentStorage, + logger: this.logger, + }); + } + return this.projectRecord(record); + } + private projectRecord(record: StoredAgentRecord): OwnedAgentSnapshot { const owner = record.owner; if (owner?.kind !== "hub" || owner.relationshipId !== this.relationshipId) { diff --git a/packages/server/src/server/hub/relationship-remote.test.ts b/packages/server/src/server/hub/relationship-remote.test.ts index d6617a7abb..a2be94dc2e 100644 --- a/packages/server/src/server/hub/relationship-remote.test.ts +++ b/packages/server/src/server/hub/relationship-remote.test.ts @@ -1,12 +1,32 @@ import { createServer } from "node:http"; -import type { AddressInfo } from "node:net"; +import { mkdtemp, rm } from "node:fs/promises"; +import type { AddressInfo, Socket } from "node:net"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; import { afterEach, expect, test } from "vitest"; -import { DirectHubRelationshipRemote } from "./relationship-remote.js"; +import { WebSocket } from "ws"; +import type { HubExecutions } from "./relationship-owned-executions.js"; +import { + HubRelationshipController, + type HubRelationshipClock, + type HubRelationshipRetryPolicy, + type ScheduledRelationshipTask, +} from "./relationship-controller.js"; +import { + DirectHubRelationshipRemote, + type HubSocketConnection, + type HubSocketEvents, +} from "./relationship-remote.js"; const openServers: ReturnType[] = []; +const openUpgradeHubs: UpgradeRejectingHub[] = []; +const openPaseoHomes: string[] = []; afterEach(async () => { + for (const hub of openUpgradeHubs.splice(0)) hub.destroyConnections(); await Promise.all(openServers.splice(0).map((server) => closeServer(server))); + await Promise.all(openPaseoHomes.splice(0).map((home) => rm(home, { recursive: true }))); }); test.each([401, 403, 404])( @@ -30,6 +50,107 @@ test("transient revocation failures remain retryable", async () => { ).rejects.toThrow("Hub revocation failed (503)"); }); +test.each([401, 403] as const)( + "socket authentication status %s is terminal and releases the failed upgrade", + async (status) => { + const hub = await UpgradeRejectingHub.start([status]); + const outcomes = new SocketOutcomes(); + const remote = new DirectHubRelationshipRemote(); + + const socket = remote.openSocket( + { + relationshipId: "relationship-1", + webSocketUrl: hub.webSocketUrl, + credential: "invalid", + }, + outcomes.events, + ) as HubSocketConnection & WebSocket; + + expect(await outcomes.next()).toEqual({ type: "rejected", status }); + await hub.expectAttemptReleased(1); + expect(await outcomes.afterCleanup()).toEqual([{ type: "rejected", status }]); + expect(socket.readyState).toBe(WebSocket.CLOSED); + expect(hub.openConnections()).toBe(0); + }, +); + +test("a transient socket response releases the failed upgrade and reports one connection loss", async () => { + const hub = await UpgradeRejectingHub.start([503]); + const outcomes = new SocketOutcomes(); + const remote = new DirectHubRelationshipRemote(); + + const socket = remote.openSocket( + { + relationshipId: "relationship-1", + webSocketUrl: hub.webSocketUrl, + credential: "credential", + }, + outcomes.events, + ) as HubSocketConnection & WebSocket; + + expect(await outcomes.next()).toEqual({ type: "closed", code: 1006 }); + await hub.expectAttemptReleased(1); + expect(await outcomes.afterCleanup()).toEqual([{ type: "closed", code: 1006 }]); + expect(socket.readyState).toBe(WebSocket.CLOSED); + expect(hub.openConnections()).toBe(0); +}); + +test.each([401, 403] as const)( + "controller treats socket authentication status %s as terminal without redialing", + async (status) => { + const hub = await UpgradeRejectingHub.start([status]); + const clock = new ManualRelationshipClock(); + const controller = await connectController(hub, clock); + + await hub.expectAttemptReleased(1); + + expect(controller.status()).toMatchObject({ + state: "revoked", + lastError: `Hub rejected socket authentication (${status})`, + }); + expect(hub.attemptCount()).toBe(1); + expect(clock.pendingTasks()).toBe(0); + expect(hub.openConnections()).toBe(0); + }, +); + +test("controller redials after a transient socket response and releases both failed upgrades", async () => { + const hub = await UpgradeRejectingHub.start([503, 401]); + const clock = new ManualRelationshipClock(); + const controller = await connectController(hub, clock); + + await hub.expectAttemptReleased(1); + expect(controller.status().state).toBe("reconnecting"); + expect(clock.pendingTasks()).toBe(1); + + clock.runNext(); + await hub.expectAttemptReleased(2); + + expect(hub.attemptCount()).toBe(2); + expect(controller.status().state).toBe("revoked"); + expect(clock.pendingTasks()).toBe(0); + expect(hub.openConnections()).toBe(0); +}); + +test("controller redials once after a failed upgrade without also handling its close", async () => { + const hub = await UpgradeRejectingHub.start([0, 401]); + const clock = new ManualRelationshipClock(); + const controller = await connectController(hub, clock); + + await hub.expectAttemptReleased(1); + + expect(controller.status().state).toBe("reconnecting"); + expect(clock.scheduledAttempts()).toEqual([0]); + expect(clock.pendingTasks()).toBe(1); + + clock.runNext(); + await hub.expectAttemptReleased(2); + + expect(hub.attemptCount()).toBe(2); + expect(controller.status().state).toBe("revoked"); + expect(clock.pendingTasks()).toBe(0); +}); + async function startHubReturning(status: number): Promise { const server = createServer((_request, response) => { response.writeHead(status).end(); @@ -54,3 +175,214 @@ function closeServer(server: ReturnType): Promise { }); }); } + +type SocketOutcome = + | { type: "connected" } + | { type: "rejected"; status: 401 | 403 } + | { type: "closed"; code: number } + | { type: "failed"; message: string }; + +interface Deferred { + promise: Promise; + resolve(value: T): void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + const promise = new Promise((accept) => { + resolve = accept; + }); + return { promise, resolve }; +} + +class SocketOutcomes { + private readonly outcomes: SocketOutcome[] = []; + private nextOutcome = deferred(); + + readonly events: HubSocketEvents = { + connected: () => this.record({ type: "connected" }), + rejected: (status) => this.record({ type: "rejected", status }), + closed: (code) => this.record({ type: "closed", code }), + failed: (error) => this.record({ type: "failed", message: error.message }), + }; + + next(): Promise { + return withDeadline(this.nextOutcome.promise, "Hub socket produced no outcome"); + } + + async afterCleanup(): Promise { + await new Promise((resolve) => setImmediate(resolve)); + return this.outcomes.slice(); + } + + private record(outcome: SocketOutcome): void { + this.outcomes.push(outcome); + this.nextOutcome.resolve(outcome); + this.nextOutcome = deferred(); + } +} + +class UpgradeRejectingHub { + private readonly server = createServer(async (request, response) => { + let body = ""; + for await (const chunk of request) body += chunk; + const enrollment = JSON.parse(body) as { relationshipId: string }; + response.writeHead(200, { "content-type": "application/json" }).end( + JSON.stringify({ + relationshipId: enrollment.relationshipId, + scopes: ["hub.*"], + webSocketUrl: this.webSocketUrl, + }), + ); + }); + private readonly sockets = new Set(); + private readonly releasedAttempts = new Map>(); + private attemptObserved = deferred(); + private attempts = 0; + + private constructor(private readonly statuses: number[]) { + this.server.on("upgrade", (_request, socket) => { + const attempt = ++this.attempts; + this.attemptObserved.resolve(); + this.attemptObserved = deferred(); + const released = deferred(); + this.releasedAttempts.set(attempt, released); + this.sockets.add(socket); + socket.once("close", () => { + this.sockets.delete(socket); + released.resolve(); + }); + const status = this.statuses[attempt - 1] ?? this.statuses.at(-1) ?? 503; + if (status === 0) { + socket.destroy(); + return; + } + socket.end( + `HTTP/1.1 ${status} Rejected\r\nContent-Length: 4\r\nConnection: close\r\n\r\ndeny`, + ); + }); + } + + static async start(statuses: number[]): Promise { + const hub = new UpgradeRejectingHub(statuses); + openUpgradeHubs.push(hub); + openServers.push(hub.server); + await new Promise((resolve, reject) => { + hub.server.once("error", reject); + hub.server.listen(0, "127.0.0.1", resolve); + }); + return hub; + } + + get webSocketUrl(): string { + const address = this.server.address() as AddressInfo; + return `ws://127.0.0.1:${address.port}/daemon`; + } + + get origin(): string { + const address = this.server.address() as AddressInfo; + return `http://127.0.0.1:${address.port}`; + } + + attemptCount(): number { + return this.attempts; + } + + openConnections(): number { + return this.sockets.size; + } + + destroyConnections(): void { + for (const socket of this.sockets) socket.destroy(); + } + + async expectAttemptReleased(attempt: number): Promise { + while (this.attempts < attempt) { + await withDeadline( + this.attemptObserved.promise, + `Hub did not receive socket attempt ${attempt}`, + ); + } + const released = this.releasedAttempts.get(attempt); + if (!released) throw new Error(`Hub did not receive socket attempt ${attempt}`); + await withDeadline(released.promise, `Hub socket attempt ${attempt} remained open`); + await new Promise((resolve) => setImmediate(resolve)); + } +} + +class ManualRelationshipClock implements HubRelationshipClock, HubRelationshipRetryPolicy { + private readonly tasks: Array<{ cancelled: boolean; task: () => void }> = []; + private readonly attempts: number[] = []; + + now(): Date { + return new Date("2026-07-13T00:00:00.000Z"); + } + + delay(attempt: number): number { + this.attempts.push(attempt); + return 0; + } + + schedule(_delayMs: number, task: () => void): ScheduledRelationshipTask { + const scheduled = { cancelled: false, task }; + this.tasks.push(scheduled); + return { cancel: () => (scheduled.cancelled = true) }; + } + + pendingTasks(): number { + return this.tasks.filter((task) => !task.cancelled).length; + } + + scheduledAttempts(): number[] { + return this.attempts.slice(); + } + + runNext(): void { + const scheduled = this.tasks.shift(); + if (!scheduled || scheduled.cancelled) throw new Error("No Hub reconnect is scheduled"); + scheduled.task(); + } +} + +const unusedExecutions: HubExecutions = { + create: async () => { + throw new Error("Unexpected Hub agent create"); + }, + reconcile: async () => null, + subscribe: () => () => undefined, +}; + +async function connectController( + hub: UpgradeRejectingHub, + clock: ManualRelationshipClock, +): Promise { + const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-hub-socket-")); + openPaseoHomes.push(paseoHome); + const controller = new HubRelationshipController({ + paseoHome, + serverId: "server-1", + daemonPublicKey: "daemon-public-key", + logger: pino({ level: "silent" }), + remote: new DirectHubRelationshipRemote(), + clock, + retryPolicy: clock, + attachSocket: async () => undefined, + createExecutions: () => unusedExecutions, + }); + await controller.connect({ hubUrl: hub.origin, token: "enrollment-token" }); + return controller; +} + +async function withDeadline(promise: Promise, message: string): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(message)), 2_000); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts index 07aa6dacf1..38d430897c 100644 --- a/packages/server/src/server/hub/relationship-remote.ts +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -104,14 +104,34 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { "x-paseo-relationship-id": input.relationshipId, }, }); - socket.once("open", () => events.connected(socket as WebSocketLike)); + let settled = false; + socket.once("open", () => { + if (!settled) events.connected(socket as WebSocketLike); + }); socket.once("unexpected-response", (_request, response) => { + if (settled) { + response.destroy(); + return; + } + settled = true; + response.destroy(); + socket.terminate(); if (response.statusCode === 401 || response.statusCode === 403) { events.rejected(response.statusCode); + return; } + events.closed(1006); + }); + socket.once("close", (code) => { + if (settled) return; + settled = true; + events.closed(code); + }); + socket.once("error", (error) => { + if (settled) return; + settled = true; + events.failed(error); }); - socket.once("close", (code) => events.closed(code)); - socket.once("error", (error) => events.failed(error)); return socket; } } 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 3a48da799d..e0c9fd358e 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -241,6 +241,7 @@ class ControlledAgentClient implements AgentClient { private gate: Deferred | null = null; private creationObserved = deferred(); creations = 0; + resumes = 0; createdConfigs: AgentSessionConfig[] = []; constructor(private readonly client: AgentClient) { @@ -282,6 +283,7 @@ class ControlledAgentClient implements AgentClient { overrides?: Partial, launchContext?: AgentLaunchContext, ): Promise { + this.resumes++; return this.client.resumeSession(handle, overrides, launchContext); } @@ -631,6 +633,10 @@ export class HubRelationshipHarness { return this.codex.creations; } + providerResumes(): number { + return this.codex.resumes; + } + latestCreatedCwd(): string | null { return this.codex.createdConfigs.at(-1)?.cwd ?? null; } @@ -1133,6 +1139,7 @@ export class HubRelationshipHarness { relationshipId: this.relationshipFile()!.relationship.id, agentManager: manager, agentStorage: storage, + logger: pino({ level: "silent" }), createAgent: (input) => createAgentCommand( {