diff --git a/docs/architecture.md b/docs/architecture.md index 331b0ecb79..e82c1568bc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -126,6 +126,11 @@ Enables remote access when the daemon is behind a firewall. See [SECURITY.md](../SECURITY.md) for the full threat model. +### Paseo Hub + +The optional Hub relationship is daemon-outbound and does not use the relay. Its connection, +authorization, ownership, persistence, and lifecycle contract is documented in [hub.md](hub.md). + ### `packages/desktop` — Desktop app (Electron) Electron wrapper for macOS, Linux, and Windows. diff --git a/docs/hub.md b/docs/hub.md new file mode 100644 index 0000000000..785c93e216 --- /dev/null +++ b/docs/hub.md @@ -0,0 +1,61 @@ +# Paseo Hub relationship + +Paseo Hub is an explicit opt-in connection from one Paseo daemon to one Hub. Running a daemon does +not register it with a Hub. The relationship begins only when a user runs +`paseo hub connect --token ` from the daemon machine. + +## Connection and authority + +The daemon enrolls over HTTP(S), then opens and maintains a direct outbound WebSocket to the Hub. +The Hub never discovers or acquires the daemon through Paseo's relay. The relay remains an optional +encrypted path for normal Paseo clients and has no role in Hub enrollment, authentication, dispatch, +or reconnects. + +The daemon persists a relationship ID and private connection credential before enrollment. The +relationship is independent of its current transport, so a future transport can replace the direct +WebSocket without pairing again. The current foundation supports one Hub relationship per daemon. + +Only a trusted session arriving through a server-observed local, non-browser daemon transport may +run the `hub.relationship.connect`, `hub.relationship.get_status`, and +`hub.relationship.disconnect` RPCs. Remote, relay, browser-origin, and Hub sessions cannot manage +the relationship. A hello message's `clientType` is descriptive and grants no authority. + +## Hub session scope + +An accepted Hub socket receives a dedicated `HubSession`, not the daemon's general client `Session`. +Its inbound allowlist contains only the `hub.*` operations required for Hub-owned execution, and its +outbound events include only agents owned by that relationship. Unrelated local agents, browser +control, retained client sessions, binary channels, and ordinary daemon broadcasts are outside the +Hub surface. + +Each Hub create carries an execution ID. The daemon stores that ID with the agent's relationship +owner before acknowledging creation. Duplicate or replayed creates for the same relationship and +execution resolve to the same durable agent. On reconnect or daemon restart, Hub reconciliation +reads that stored association and returns current state; transient stream frames are not durably +replayed. + +Hub creates use the same agent creation path as trusted clients. They may select any existing +worktree target shape and may request `autoArchive`. Worktree creation and terminal auto-archive use +the shared workspace-aware lifecycle policy; Hub does not have a second launch or cleanup path. + +## Disconnect and revocation + +Normal socket loss reconnects the active relationship with bounded exponential backoff and jitter. +Daemon restart loads the same relationship and credential and reconnects without another enrollment +ceremony. + +Hub authentication rejection or close code `4403` permanently revokes the local relationship. The +daemon deletes its credential, stops reconnecting, and retains only the relationship ID, Hub origin, +scopes, and a sanitized reason for status reporting. + +`paseo hub disconnect` disables socket reconnect before requesting remote revocation. If the Hub is +offline, the daemon persists `disconnecting` and retries revocation across daemon restarts without +opening a Hub socket. `--force` removes local authority immediately and warns that remote revocation +may still be pending. + +## Cross-repository dependency + +The consumer implementation lives in Paseo Cloud and depends on the Paseo protocol, client, server, +and CLI package versions that expose this Hub surface. Cross-repository end-to-end verification must +install those packages together and exercise the real daemon, CLI, direct WebSocket, Cloud service, +and Postgres. A Paseo Cloud change must not assume unpublished local package contents. diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index f89004dd51..e71a6ad452 100644 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -9,6 +9,7 @@ import { createScheduleCommand } from "./commands/schedule/index.js"; import { createSpeechCommand } from "./commands/speech/index.js"; import { createTerminalCommand } from "./commands/terminal/index.js"; import { createWorktreeCommand } from "./commands/worktree/index.js"; +import { createHubCommand } from "./commands/hub/index.js"; import { createHooksCommand } from "./commands/hooks.js"; import { startCommand as daemonStartCommand } from "./commands/daemon/start.js"; import { runStatusCommand as runDaemonStatusCommand } from "./commands/daemon/status.js"; @@ -145,6 +146,7 @@ export function createCli(): Command { // Daemon commands program.addCommand(createDaemonCommand()); + program.addCommand(createHubCommand()); // Chat commands program.addCommand(createChatCommand()); diff --git a/packages/cli/src/commands/hub/index.ts b/packages/cli/src/commands/hub/index.ts new file mode 100644 index 0000000000..81872bf5f8 --- /dev/null +++ b/packages/cli/src/commands/hub/index.ts @@ -0,0 +1,104 @@ +import { Command } from "commander"; +import { withOutput, type ListResult, type OutputSchema } from "../../output/index.js"; +import { addJsonAndDaemonHostOptions } from "../../utils/command-options.js"; +import { connectToDaemon } from "../../utils/client.js"; + +interface HubRow { + state: string; + relationshipId: string | null; + hub: string | null; + scopes: string; + connectedAt: string | null; + error: string | null; + warning?: string; +} + +const schema: OutputSchema = { + idField: "state", + columns: [ + { header: "STATE", field: "state" }, + { header: "HUB", field: "hub" }, + { header: "RELATIONSHIP", field: "relationshipId" }, + { header: "SCOPES", field: "scopes" }, + { header: "CONNECTED", field: "connectedAt" }, + { header: "ERROR", field: "error" }, + { header: "WARNING", field: "warning" }, + ], +}; + +function result( + status: { + state: string; + relationshipId: string | null; + hubOrigin: string | null; + scopes: string[]; + connectedAt: string | null; + lastError: string | null; + }, + warning?: string, +): ListResult { + return { + type: "list", + data: [ + { + state: status.state, + relationshipId: status.relationshipId, + hub: status.hubOrigin, + scopes: status.scopes.join(", "), + connectedAt: status.connectedAt, + error: status.lastError, + warning, + }, + ], + schema, + }; +} + +async function withClient( + host: string | undefined, + action: (client: Awaited>) => Promise, +): Promise { + const client = await connectToDaemon({ host }); + try { + return await action(client); + } finally { + await client.close().catch(() => undefined); + } +} + +export function createHubCommand(): Command { + const hub = new Command("hub").description("Manage this daemon's Paseo Hub relationship"); + addJsonAndDaemonHostOptions( + hub.command("connect").argument("").requiredOption("--token "), + ).action( + withOutput(async (...args) => { + const url = args[0] as string; + const options = args.at(-2) as { token: string; host?: string }; + return withClient(options.host, async (client) => + result((await client.connectHub(url, options.token)).status), + ); + }), + ); + addJsonAndDaemonHostOptions(hub.command("status")).action( + withOutput(async (...args) => { + const options = args.at(-2) as { host?: string }; + return withClient(options.host, async (client) => + result((await client.getHubStatus()).status), + ); + }), + ); + addJsonAndDaemonHostOptions( + hub + .command("disconnect") + .option("--force", "Remove local authority even if the Hub is offline"), + ).action( + withOutput(async (...args) => { + const options = args.at(-2) as { host?: string; force?: boolean }; + return withClient(options.host, async (client) => { + const response = await client.disconnectHub(options.force ?? false); + return result(response.status, response.warning); + }); + }), + ); + return hub; +} diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index a19be362f1..d448e9fe52 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -3929,6 +3929,30 @@ export class DaemonClient { }); } + async connectHub(hubUrl: string, token: string, requestId?: string) { + return this.sendCorrelatedSessionRequest({ + requestId, + message: { type: "hub.relationship.connect.request", hubUrl, token }, + responseType: "hub.relationship.connect.response", + }); + } + + async getHubStatus(requestId?: string) { + return this.sendCorrelatedSessionRequest({ + requestId, + message: { type: "hub.relationship.get_status.request" }, + responseType: "hub.relationship.get_status.response", + }); + } + + async disconnectHub(force = false, requestId?: string) { + return this.sendCorrelatedSessionRequest({ + requestId, + message: { type: "hub.relationship.disconnect.request", force }, + responseType: "hub.relationship.disconnect.response", + }); + } + async getDaemonPairingOffer( options?: DaemonPairingOfferOptions, ): Promise { diff --git a/packages/protocol/src/messages.hub.test.ts b/packages/protocol/src/messages.hub.test.ts new file mode 100644 index 0000000000..3f17123156 --- /dev/null +++ b/packages/protocol/src/messages.hub.test.ts @@ -0,0 +1,232 @@ +import { describe, expect, test } from "vitest"; +import { z } from "zod"; + +import { + HubMessageCorrelationError, + SessionInboundMessageSchema, + SessionOutboundMessageSchema, + parseHubSessionOutboundMessage, +} from "./messages.js"; + +const agent = { + id: "agent-1", + provider: "codex", + cwd: "/workspace", + model: null, + createdAt: "2026-07-13T00:00:00.000Z", + updatedAt: "2026-07-13T00:00:00.000Z", + lastUserMessageAt: null, + status: "idle", + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: true, + supportsMcpServers: false, + supportsReasoningStream: true, + supportsToolInvocations: true, + supportsRewindConversation: false, + supportsRewindFiles: false, + supportsRewindBoth: false, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: null, + labels: {}, +}; + +// Frozen at the Hub create request shape shipped before worktree and autoArchive. +const PreviousHubAgentCreateRequestSchema = z.object({ + type: z.literal("hub.agent.create.request"), + requestId: z.string(), + executionId: z.string(), + provider: z.string(), + cwd: z.string(), + workspaceId: z.string().optional(), + prompt: z.string(), + model: z.string().optional(), + modeId: z.string().optional(), + thinkingOptionId: z.string().optional(), + featureValues: z.record(z.string(), z.unknown()).optional(), + env: z.record(z.string(), z.string()).optional(), +}); + +describe("Hub session protocol", () => { + test.each([ + { + type: "hub.agent.create.request", + requestId: "request-1", + executionId: "execution-1", + provider: "codex", + cwd: "/workspace", + prompt: "Implement the requested change", + modeId: "code", + }, + { + type: "hub.execution.reconcile.request", + requestId: "request-2", + executionId: "execution-1", + }, + ])("accepts inbound variant $type", (message) => { + expect(SessionInboundMessageSchema.parse(message)).toEqual(message); + }); + + test.each([ + undefined, + { mode: "branch-off", newBranch: "hub-work", base: "main" }, + { mode: "checkout-branch", branch: "existing-work" }, + { mode: "checkout-pr", prNumber: 42 }, + ])("accepts Hub create worktree target %#", (worktree) => { + const message = { + type: "hub.agent.create.request", + requestId: "hub-worktree", + executionId: "execution-worktree", + provider: "codex", + cwd: "/repo", + prompt: "Work in the requested target", + ...(worktree ? { worktree, autoArchive: true } : {}), + }; + + expect(SessionInboundMessageSchema.parse(message)).toEqual(message); + }); + + test("the previous Hub create parser ignores additive worktree and auto-archive fields", () => { + const newRequest = { + type: "hub.agent.create.request" as const, + requestId: "hub-worktree", + executionId: "execution-worktree", + provider: "codex", + cwd: "/repo", + prompt: "Work in the requested target", + worktree: { mode: "branch-off", newBranch: "hub-work", base: "main" }, + autoArchive: true, + }; + + expect(PreviousHubAgentCreateRequestSchema.parse(newRequest)).toEqual({ + type: "hub.agent.create.request", + requestId: "hub-worktree", + executionId: "execution-worktree", + provider: "codex", + cwd: "/repo", + prompt: "Work in the requested target", + }); + }); + + test.each([ + { + type: "hub.agent.create.response", + payload: { + requestId: "request-1", + executionId: "execution-1", + agentId: "agent-1", + agent, + success: true, + error: null, + }, + }, + { + type: "hub.agent.update", + payload: { executionId: "execution-1", agentId: "agent-1", agent }, + }, + { + type: "hub.agent.stream", + payload: { + executionId: "execution-1", + agentId: "agent-1", + event: { type: "turn_started", provider: "codex" }, + }, + }, + { + type: "hub.execution.reconcile.response", + payload: { + requestId: "request-2", + executionId: "execution-1", + agentId: "agent-1", + agent, + }, + }, + { + type: "hub.authorization.denied", + payload: { + requestId: "request-3", + requestType: "daemon.get_status.request", + code: "scope_denied", + }, + }, + ])("accepts outbound variant $type", (message) => { + expect(SessionOutboundMessageSchema.parse(message)).toEqual(message); + expect(parseHubSessionOutboundMessage(message)).toEqual(message); + }); + + test("rejects a Hub update whose correlated agent ids disagree", () => { + const malformed = { + type: "hub.agent.update", + payload: { executionId: "execution-1", agentId: "agent-2", agent }, + }; + + expect(SessionOutboundMessageSchema.safeParse(malformed).success).toBe(true); + expect(() => parseHubSessionOutboundMessage(malformed)).toThrow(HubMessageCorrelationError); + }); + + test.each([ + { + type: "hub.relationship.connect.request", + requestId: "r1", + hubUrl: "https://hub.example", + token: "token", + }, + { type: "hub.relationship.get_status.request", requestId: "r2" }, + { type: "hub.relationship.disconnect.request", requestId: "r3", force: true }, + ])("accepts trusted management request $type", (message) => { + expect(SessionInboundMessageSchema.parse(message)).toEqual(message); + }); + + test.each([ + { + type: "hub.relationship.connect.response", + payload: { + requestId: "r1", + status: { + state: "connected", + relationshipId: "rel", + hubOrigin: "https://hub.example", + scopes: ["hub.*"], + connectedAt: "2026-07-13T00:00:00.000Z", + lastError: null, + }, + }, + }, + { + type: "hub.relationship.get_status.response", + payload: { + requestId: "r2", + status: { + state: "not_connected", + relationshipId: null, + hubOrigin: null, + scopes: [], + connectedAt: null, + lastError: null, + }, + }, + }, + { + type: "hub.relationship.disconnect.response", + payload: { + requestId: "r3", + status: { + state: "disconnecting", + relationshipId: "rel", + hubOrigin: "https://hub.example", + scopes: ["hub.*"], + connectedAt: null, + lastError: "offline", + }, + warning: "pending", + }, + }, + ])("accepts trusted management response $type", (message) => { + expect(SessionOutboundMessageSchema.parse(message)).toEqual(message); + }); +}); diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 74e7c5a3e5..d88f9e3a2c 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1088,6 +1088,22 @@ export const DaemonGetPairingOfferRequestSchema = z.object({ requestId: z.string(), }); +export const HubRelationshipConnectRequestSchema = z.object({ + type: z.literal("hub.relationship.connect.request"), + requestId: z.string(), + hubUrl: z.string(), + token: z.string(), +}); +export const HubRelationshipGetStatusRequestSchema = z.object({ + type: z.literal("hub.relationship.get_status.request"), + requestId: z.string(), +}); +export const HubRelationshipDisconnectRequestSchema = z.object({ + type: z.literal("hub.relationship.disconnect.request"), + requestId: z.string(), + force: z.boolean().optional(), +}); + export const DiagnosticsRequestSchema = z.object({ type: z.literal("diagnostics.request"), requestId: z.string(), @@ -2087,7 +2103,35 @@ export const CaptureTerminalRequestSchema = z.object({ requestId: z.string(), }); +export const HubAgentCreateRequestSchema = z.object({ + type: z.literal("hub.agent.create.request"), + requestId: z.string(), + executionId: z.string(), + provider: z.string(), + cwd: z.string(), + prompt: z.string(), + workspaceId: z.string().optional(), + model: z.string().optional(), + modeId: z.string().optional(), + thinkingOptionId: z.string().optional(), + featureValues: z.record(z.string(), z.unknown()).optional(), + env: z.record(z.string(), z.string()).optional(), + worktree: CreateAgentWorktreeTargetSchema.optional(), + autoArchive: z.boolean().optional(), +}); + +export const HubExecutionReconcileRequestSchema = z.object({ + type: z.literal("hub.execution.reconcile.request"), + requestId: z.string(), + executionId: z.string(), +}); + +export type HubAgentCreateRequest = z.infer; +export type HubExecutionReconcileRequest = z.infer; + export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ + HubAgentCreateRequestSchema, + HubExecutionReconcileRequestSchema, BrowserAutomationExecuteResponseSchema, VoiceAudioChunkMessageSchema, AbortRequestMessageSchema, @@ -2110,6 +2154,9 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ WaitForFinishRequestSchema, DaemonGetStatusRequestSchema, DaemonGetPairingOfferRequestSchema, + HubRelationshipConnectRequestSchema, + HubRelationshipGetStatusRequestSchema, + HubRelationshipDisconnectRequestSchema, DiagnosticsRequestSchema, GetDaemonConfigRequestMessageSchema, SetDaemonConfigRequestMessageSchema, @@ -3218,6 +3265,38 @@ export const DaemonGetStatusResponseSchema = z.object({ .passthrough(), }); +export const HubRelationshipStatusSchema = z.object({ + state: z.enum([ + "not_connected", + "connecting", + "connected", + "reconnecting", + "disconnecting", + "revoked", + ]), + relationshipId: z.string().nullable(), + hubOrigin: z.string().nullable(), + scopes: z.array(z.string()), + connectedAt: z.string().nullable(), + lastError: z.string().nullable(), +}); +export const HubRelationshipConnectResponseSchema = z.object({ + type: z.literal("hub.relationship.connect.response"), + payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema }), +}); +export const HubRelationshipGetStatusResponseSchema = z.object({ + type: z.literal("hub.relationship.get_status.response"), + payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema }), +}); +export const HubRelationshipDisconnectResponseSchema = z.object({ + type: z.literal("hub.relationship.disconnect.response"), + payload: z.object({ + requestId: z.string(), + status: HubRelationshipStatusSchema, + warning: z.string().optional(), + }), +}); + export const DaemonGetPairingOfferResponseSchema = z.object({ type: z.literal("daemon.get_pairing_offer.response"), payload: z @@ -4307,9 +4386,101 @@ export const DaemonUpdateProgressMessageSchema = z.object({ }), }); +export const HubAgentCreateResponseSchema = z.object({ + type: z.literal("hub.agent.create.response"), + payload: z.object({ + requestId: z.string(), + executionId: z.string(), + agentId: z.string().nullable(), + agent: AgentSnapshotPayloadSchema.nullable(), + success: z.boolean(), + error: z.string().nullable(), + }), +}); + +export const HubAgentUpdateSchema = z.object({ + type: z.literal("hub.agent.update"), + payload: z.object({ + executionId: z.string(), + agentId: z.string(), + agent: AgentSnapshotPayloadSchema, + }), +}); + +export const HubAgentStreamSchema = z.object({ + type: z.literal("hub.agent.stream"), + payload: z.object({ + executionId: z.string(), + agentId: z.string(), + event: AgentStreamEventPayloadSchema, + }), +}); + +export const HubExecutionReconcileResponseSchema = z.object({ + type: z.literal("hub.execution.reconcile.response"), + payload: z.object({ + requestId: z.string(), + executionId: z.string(), + agentId: z.string().nullable(), + agent: AgentSnapshotPayloadSchema.nullable(), + }), +}); + +export const HubAuthorizationDeniedSchema = z.object({ + type: z.literal("hub.authorization.denied"), + payload: z.object({ + requestId: z.string().optional(), + requestType: z.string(), + code: z.literal("scope_denied"), + }), +}); + +export type HubAgentCreateResponse = z.infer; +export type HubAgentUpdate = z.infer; +export type HubAgentStream = z.infer; +export type HubExecutionReconcileResponse = z.infer; +export type HubAuthorizationDenied = z.infer; + +export const HubSessionOutboundMessageSchema = z.discriminatedUnion("type", [ + HubAgentCreateResponseSchema, + HubAgentUpdateSchema, + HubAgentStreamSchema, + HubExecutionReconcileResponseSchema, + HubAuthorizationDeniedSchema, +]); + +export type HubSessionOutboundMessage = z.infer; + +export class HubMessageCorrelationError extends Error { + constructor(messageType: HubSessionOutboundMessage["type"]) { + super(`Hub message ${messageType} has mismatched agent correlation`); + this.name = "HubMessageCorrelationError"; + } +} + +export function parseHubSessionOutboundMessage(value: unknown): HubSessionOutboundMessage { + const message = HubSessionOutboundMessageSchema.parse(value); + const payload = message.payload; + if ( + "agent" in payload && + payload.agent !== null && + "agentId" in payload && + payload.agentId !== null && + payload.agent.id !== payload.agentId + ) { + throw new HubMessageCorrelationError(message.type); + } + return message; +} + export type DaemonUpdateProgressMessage = z.infer; export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ + HubAgentCreateResponseSchema, + HubAgentUpdateSchema, + HubAgentStreamSchema, + HubExecutionReconcileResponseSchema, + HubAuthorizationDeniedSchema, BrowserAutomationExecuteRequestSchema, ActivityLogMessageSchema, AssistantChunkMessageSchema, @@ -4356,6 +4527,9 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ SetVoiceModeResponseMessageSchema, DaemonGetStatusResponseSchema, DaemonGetPairingOfferResponseSchema, + HubRelationshipConnectResponseSchema, + HubRelationshipGetStatusResponseSchema, + HubRelationshipDisconnectResponseSchema, DiagnosticsResponseSchema, GetDaemonConfigResponseMessageSchema, SetDaemonConfigResponseMessageSchema, diff --git a/packages/server/src/server/agent/agent-loading.ts b/packages/server/src/server/agent/agent-loading.ts index 578502195d..cb79c324c7 100644 --- a/packages/server/src/server/agent/agent-loading.ts +++ b/packages/server/src/server/agent/agent-loading.ts @@ -75,6 +75,7 @@ export async function ensureAgentLoaded( snapshot = await deps.agentManager.createAgent(config, agentId, { labels: record.labels, workspaceId: record.workspaceId, + owner: record.owner, }); deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config"); } diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index f74bbdc995..17ec11c315 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -44,6 +44,7 @@ import { } from "./agent-sdk-types.js"; import { buildArchivedAgentRecord, type ArchivedStoredAgentRecord } from "./agent-archive.js"; import type { StoredAgentRecord, AgentStorage } from "./agent-storage.js"; +import type { AgentOwner } from "./agent-owner.js"; import { InMemoryAgentTimelineStore, type SeedAgentTimelineOptions, @@ -218,6 +219,7 @@ export interface CreateAgentOptions { initialTitle?: string | null; // undefined is an explicit decision: the agent never appears in the sidebar. workspaceId: string | undefined; + owner?: AgentOwner; } export interface AgentManagerOptions { @@ -292,6 +294,7 @@ interface ManagedAgentBase { * Null/undefined for legacy agents created before ownership stamping. */ workspaceId?: string; + owner?: AgentOwner; capabilities: AgentCapabilityFlags; config: AgentSessionConfig; runtimeInfo?: AgentRuntimeInfo; @@ -993,6 +996,7 @@ export class AgentManager { labels: options.labels, initialTitle: options.initialTitle, workspaceId: options.workspaceId, + owner: options.owner, }); } @@ -1016,6 +1020,7 @@ export class AgentManager { lastUserMessageAt?: Date | null; labels?: Record; workspaceId?: string; + owner?: AgentOwner; }, ): Promise { return this.trackAgentRegistrationOperation( @@ -1033,6 +1038,7 @@ export class AgentManager { lastUserMessageAt?: Date | null; labels?: Record; workspaceId?: string; + owner?: AgentOwner; }, ): Promise { this.assertAcceptingAgentRegistrations(); @@ -1212,6 +1218,7 @@ export class AgentManager { return this.registerSession(session, storedConfig, agentId, { labels: existing.labels, workspaceId: existing.workspaceId, + owner: existing.owner, createdAt: existing.createdAt, updatedAt: existing.updatedAt, lastUserMessageAt: existing.lastUserMessageAt, @@ -1281,7 +1288,7 @@ export class AgentManager { } } - async closeAgent(agentId: string): Promise { + async closeAgent(agentId: string, options: { persistClosedState?: boolean } = {}): Promise { const agent = this.requireAgent(agentId); this.logger.trace( { @@ -1301,7 +1308,9 @@ export class AgentManager { for (const event of this.providerSubagents.deleteParent(agentId)) { this.dispatch({ type: "provider_subagent", event }); } - await this.persistSnapshot(closedAgent); + if (options.persistClosedState !== false) { + await this.persistSnapshot(closedAgent); + } this.emitClosedAgent(closedAgent, { persist: false }); this.logger.trace( { @@ -1403,6 +1412,7 @@ export class AgentManager { provider: record.provider, cwd: record.cwd, workspaceId: record.workspaceId, + owner: record.owner, session: null, capabilities: STORED_AGENT_CAPABILITIES, config: buildStoredAgentConfig(record), @@ -2593,6 +2603,7 @@ export class AgentManager { initialTitle?: string | null; publishWhenReady?: boolean; workspaceId?: string; + owner?: AgentOwner; }, ): Promise { let registered = false; @@ -2728,6 +2739,7 @@ export class AgentManager { attention?: AttentionState; persistence?: AgentPersistenceHandle; workspaceId?: string; + owner?: AgentOwner; } | undefined; }): ActiveManagedAgent { @@ -2737,6 +2749,7 @@ export class AgentManager { provider: config.provider, cwd: config.cwd, workspaceId: options?.workspaceId, + owner: options?.owner, session, capabilities: session.capabilities, config, diff --git a/packages/server/src/server/agent/agent-owner.ts b/packages/server/src/server/agent/agent-owner.ts new file mode 100644 index 0000000000..24b5938ae9 --- /dev/null +++ b/packages/server/src/server/agent/agent-owner.ts @@ -0,0 +1,16 @@ +import { z } from "zod"; + +export const AgentOwnerSchema = z.discriminatedUnion("kind", [ + z.object({ + kind: z.literal("hub"), + relationshipId: z.string(), + executionId: z.string(), + }), +]); + +export type AgentOwner = z.infer; +export type HubAgentOwner = Extract; + +export function hubExecutionKey(owner: HubAgentOwner): string { + return `${owner.relationshipId}\0${owner.executionId}`; +} diff --git a/packages/server/src/server/agent/agent-projections.ts b/packages/server/src/server/agent/agent-projections.ts index 6adee4bd57..a01c67b29d 100644 --- a/packages/server/src/server/agent/agent-projections.ts +++ b/packages/server/src/server/agent/agent-projections.ts @@ -93,6 +93,7 @@ export function toStoredAgentRecord( ? agent.attention.attentionTimestamp.toISOString() : null, internal: options?.internal, + owner: agent.owner, } satisfies StoredAgentRecord; } diff --git a/packages/server/src/server/agent/agent-storage.ts b/packages/server/src/server/agent/agent-storage.ts index 069865f660..f78c0a4d66 100644 --- a/packages/server/src/server/agent/agent-storage.ts +++ b/packages/server/src/server/agent/agent-storage.ts @@ -8,6 +8,7 @@ import { AgentFeatureSchema, AgentStatusSchema } from "../messages.js"; import { toStoredAgentRecord } from "./agent-projections.js"; import type { ManagedAgent } from "./agent-manager.js"; import type { AgentSessionConfig } from "./agent-sdk-types.js"; +import { AgentOwnerSchema, hubExecutionKey, type HubAgentOwner } from "./agent-owner.js"; const SERIALIZABLE_CONFIG_SCHEMA = z .object({ @@ -64,6 +65,7 @@ const STORED_AGENT_SCHEMA = z.object({ attentionTimestamp: z.string().nullable().optional(), internal: z.boolean().optional(), archivedAt: z.string().nullable().optional(), + owner: AgentOwnerSchema.optional(), }); export type SerializableAgentConfig = Pick< @@ -88,6 +90,7 @@ export class AgentStorage { private pathsById: Map> = new Map(); private pendingWrites: Map> = new Map(); private deleting: Set = new Set(); + private hubAgentIdsByExecution: Map = new Map(); private loaded = false; private baseDir: string; private loadPromise: Promise | null = null; @@ -112,6 +115,12 @@ export class AgentStorage { return this.cache.get(agentId) ?? null; } + async findByHubExecution(owner: HubAgentOwner): Promise { + await this.load(); + const agentId = this.hubAgentIdsByExecution.get(hubExecutionKey(owner)); + return agentId ? (this.cache.get(agentId) ?? null) : null; + } + async upsert(record: StoredAgentRecord): Promise { await this.load(); await this.queueRecordWrite(record); @@ -157,6 +166,7 @@ export class AgentStorage { } this.cache.set(agentId, record); + this.indexOwner(record); this.pathById.set(agentId, nextPath); } @@ -186,6 +196,7 @@ export class AgentStorage { ); this.cache.delete(agentId); + this.removeOwnerIndex(agentId); this.pathById.delete(agentId); this.pathsById.delete(agentId); } @@ -248,6 +259,7 @@ export class AgentStorage { this.cache.clear(); this.pathById.clear(); this.pathsById.clear(); + this.hubAgentIdsByExecution.clear(); try { const records = await this.scanDisk(); @@ -310,6 +322,7 @@ export class AgentStorage { const { record, filePath } = item; records.push(record); this.cache.set(record.id, record); + this.indexOwner(record); this.pathById.set(record.id, filePath); this.addIndexedPath(record.id, filePath); } @@ -350,6 +363,21 @@ export class AgentStorage { } } + private indexOwner(record: StoredAgentRecord): void { + this.removeOwnerIndex(record.id); + if (record.owner?.kind === "hub") { + this.hubAgentIdsByExecution.set(hubExecutionKey(record.owner), record.id); + } + } + + private removeOwnerIndex(agentId: string): void { + for (const [key, indexedAgentId] of this.hubAgentIdsByExecution) { + if (indexedAgentId === agentId) { + this.hubAgentIdsByExecution.delete(key); + } + } + } + private async waitForPendingWrite(agentId: string): Promise { await (this.pendingWrites.get(agentId) ?? Promise.resolve()).catch(() => undefined); } diff --git a/packages/server/src/server/agent/create-agent-lifecycle-dispatch.ts b/packages/server/src/server/agent/create-agent-lifecycle-dispatch.ts index eeaced590a..b0a8a97912 100644 --- a/packages/server/src/server/agent/create-agent-lifecycle-dispatch.ts +++ b/packages/server/src/server/agent/create-agent-lifecycle-dispatch.ts @@ -147,23 +147,11 @@ export class CreateAgentLifecycleDispatch { agentId: string, options: { worktreePath: string | null; repoRoot: string | null }, ): void { - const unsubscribe = this.dependencies.agentManager.subscribe( - (event) => { - if (event.type !== "agent_stream") { - return; - } - if ( - event.event.type !== "turn_completed" && - event.event.type !== "turn_failed" && - event.event.type !== "turn_canceled" - ) { - return; - } - unsubscribe(); - void this.autoArchiveAgentOnce(agentId, options); - }, - { agentId, replayState: false }, - ); + registerAgentAutoArchive({ + agentManager: this.dependencies.agentManager, + agentId, + archive: () => this.autoArchiveAgentOnce(agentId, options), + }); } private async autoArchiveAgentOnce( @@ -250,3 +238,25 @@ export class CreateAgentLifecycleDispatch { } } } + +export function registerAgentAutoArchive(input: { + agentManager: AgentManager; + agentId: string; + archive: () => Promise; +}): void { + const unsubscribe = input.agentManager.subscribe( + (event) => { + if (event.type !== "agent_stream") return; + if ( + event.event.type !== "turn_completed" && + event.event.type !== "turn_failed" && + event.event.type !== "turn_canceled" + ) { + return; + } + unsubscribe(); + void input.archive(); + }, + { agentId: input.agentId, replayState: false }, + ); +} diff --git a/packages/server/src/server/agent/create-agent/create.test.ts b/packages/server/src/server/agent/create-agent/create.test.ts index 2c264501f6..73440926e0 100644 --- a/packages/server/src/server/agent/create-agent/create.test.ts +++ b/packages/server/src/server/agent/create-agent/create.test.ts @@ -242,6 +242,58 @@ test("mcp create stamps the new worktree's workspaceId, not the parent's", async } }); +test("mcp create exposes the created worktree before dispatching the initial prompt", async () => { + const workdir = mkdtempSync(join(tmpdir(), "create-agent-worktree-callback-test-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const agentManager = createRealAgentManager(storage); + const createdWorktree = await fakeWorktreeCreator({ + repoRoot: workdir, + createdWorkspaceId: "ws-created-worktree", + })(); + let observed: + | { + createdWorktree: CreatePaseoWorktreeWorkflowResult | null; + lifecycle: ManagedAgent["lifecycle"] | null; + } + | undefined; + + try { + await createAgentCommand( + { + agentManager, + agentStorage: storage, + logger, + providerSnapshotManager: { + async resolveCreateConfig() { + return {}; + }, + }, + createPaseoWorktree: async () => createdWorktree, + }, + { + kind: "mcp", + provider: "codex", + cwd: workdir, + title: "worktree callback", + initialPrompt: "Say done.", + background: true, + notifyOnFinish: false, + worktree: { worktreeName: "feature", baseBranch: "main" }, + onCreated: ({ agentId, createdWorktree: callbackWorktree }) => { + observed = { + createdWorktree: callbackWorktree, + lifecycle: agentManager.getAgent(agentId)?.lifecycle ?? null, + }; + }, + }, + ); + + expect(observed).toEqual({ createdWorktree, lifecycle: "idle" }); + } finally { + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("session create keeps the prompt title after the initial prompt settles", async () => { const workdir = mkdtempSync(join(tmpdir(), "create-agent-title-test-")); const storage = new AgentStorage(join(workdir, "agents"), logger); diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index 62b06bfcd9..e8feef4d2f 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -20,6 +20,7 @@ import type { AgentSessionConfig, } from "../agent-sdk-types.js"; import type { AgentStorage } from "../agent-storage.js"; +import type { AgentOwner } from "../agent-owner.js"; import type { ProviderSnapshotManager } from "../provider-snapshot-manager.js"; import { setupFinishNotification, startCreatedAgentInitialPrompt } from "../agent-prompt.js"; import { resolveCreateAgentTitles } from "../create-agent-title.js"; @@ -45,7 +46,7 @@ export interface CreateAgentCommandDependencies { paseoHome?: string; worktreesRoot?: string; terminalManager?: TerminalManager | null; - providerSnapshotManager: ProviderSnapshotManager; + providerSnapshotManager: Pick; createPaseoWorktree?: CreatePaseoWorktreeWorkflowFn; // Mints a fresh directory workspace for a cwd and returns its id. ensureWorkspaceForCreate?: EnsureWorkspaceForCreate; @@ -97,6 +98,12 @@ export interface CreateAgentFromMcpInput { notifyOnFinish: boolean; internal?: boolean; detached?: boolean; + owner?: AgentOwner; + env?: Record; + onCreated?: (created: { + agentId: string; + createdWorktree: CreatePaseoWorktreeWorkflowResult | null; + }) => void; callerAgentId?: string; callerContext?: { lockedCwd?: string; @@ -122,6 +129,7 @@ export interface CreateAgentCommandResult { background: boolean; initialPromptStarted: boolean; initialPromptError: unknown | null; + createdWorktree?: CreatePaseoWorktreeWorkflowResult; } export type BoundCreateAgentCommand = ( @@ -162,6 +170,7 @@ interface ResolvedCreateAgent { background: boolean; promptFailure: CreateAgentPromptFailureMode; promptLogger?: Logger; + createdWorktree?: CreatePaseoWorktreeWorkflowResult; } export async function createAgentCommand( @@ -186,6 +195,9 @@ export async function createAgentCommand( let liveSnapshot = snapshot; let initialPromptStarted = false; let initialPromptError: unknown | null = null; + if (input.kind === "mcp") { + input.onCreated?.({ agentId: snapshot.id, createdWorktree: resolved.createdWorktree ?? null }); + } if (resolved.prompt !== undefined) { const sendResult = await sendInitialPrompt(dependencies, resolved, snapshot); initialPromptStarted = sendResult.started; @@ -209,6 +221,7 @@ export async function createAgentCommand( background: resolved.background, initialPromptStarted, initialPromptError, + ...(resolved.createdWorktree ? { createdWorktree: resolved.createdWorktree } : {}), }; } @@ -268,12 +281,13 @@ async function resolveMcpCreateAgent( ? requireParentAgent(dependencies.agentManager, input.callerAgentId) : null; const cwd = resolveMcpInitialCwd(input, parentAgent); - const { resolvedCwd, setupContinuation, createdWorkspaceId } = await resolveMcpCwd({ - dependencies, - cwd, - worktree: input.worktree, - initialPrompt: input.initialPrompt ?? "", - }); + const { resolvedCwd, setupContinuation, createdWorkspaceId, createdWorktree } = + await resolveMcpCwd({ + dependencies, + cwd, + worktree: input.worktree, + initialPrompt: input.initialPrompt ?? "", + }); const workspaceId = await resolveMcpWorkspaceId({ dependencies, @@ -312,9 +326,12 @@ async function resolveMcpCreateAgent( createOptions: { ...(labels ? { labels } : {}), workspaceId: requireResolvedWorkspaceId(workspaceId), + owner: input.owner, + env: input.env, }, prompt: trimmedPrompt ? trimmedPrompt : undefined, setupContinuation, + createdWorktree, background: input.background, promptFailure: input.promptFailure ?? "log", }; @@ -516,6 +533,7 @@ async function resolveMcpCwd(params: { resolvedCwd: string; setupContinuation?: AgentWorktreeSetupContinuation; createdWorkspaceId?: string; + createdWorktree?: CreatePaseoWorktreeWorkflowResult; }> { const { dependencies, worktree } = params; if (!worktree) { @@ -574,6 +592,7 @@ async function resolveMcpCwd(params: { resolvedCwd: createdWorktree.worktree.worktreePath, setupContinuation: createdWorktree.setupContinuation, createdWorkspaceId: createdWorktree.workspace.workspaceId, + createdWorktree, }; } diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 2c0b2885fb..c5f43739d6 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -172,6 +172,18 @@ import { createAgentCommand, type CreateAgentCommandDependencies, } from "./agent/create-agent/create.js"; +import { archiveAgentCommand } from "./agent/lifecycle-command.js"; +import { CreateAgentLifecycleDispatch } from "./agent/create-agent-lifecycle-dispatch.js"; +import { + HubRelationshipController, + type HubRelationshipClock, + type HubRelationshipRetryPolicy, +} from "./hub/relationship-controller.js"; +import { + DirectHubRelationshipRemote, + type HubRelationshipRemote, +} from "./hub/relationship-remote.js"; +import { RelationshipOwnedExecutions } from "./hub/relationship-owned-executions.js"; const MAX_MCP_DEBUG_BATCH_ITEMS = 10; const REDACTED_LOG_VALUE = "[redacted]"; @@ -402,6 +414,12 @@ export interface PaseoDaemon { getListenTarget(): ListenTarget | null; } +export interface PaseoDaemonDependencies { + hubRelationshipRemote?: HubRelationshipRemote; + hubRelationshipClock?: HubRelationshipClock; + hubRelationshipRetryPolicy?: HubRelationshipRetryPolicy; +} + function createBootstrapManagedProcessRegistry( config: Pick, logger: Logger, @@ -479,6 +497,7 @@ function createInitialMutableDaemonConfig(config: PaseoDaemonConfig): MutableDae export async function createPaseoDaemon( config: PaseoDaemonConfig, rootLogger: Logger, + dependencies: PaseoDaemonDependencies = {}, ): Promise { const logger = rootLogger.child({ module: "bootstrap" }); const bootstrapStart = performance.now(); @@ -545,7 +564,7 @@ export async function createPaseoDaemon( serviceProxy, onChange: createScriptStatusEmitter({ sessions: () => - wsServer?.listActiveSessions().map((session) => ({ + wsServer?.listTrustedSessions().map((session) => ({ emit: (message) => session.emitServerMessage(message), })) ?? [], serviceProxy, @@ -809,7 +828,7 @@ export async function createPaseoDaemon( workspaceGitService, }); const archiveWorkspaceRecordExternal = async (workspaceId: string) => { - const sessions = wsServer?.listActiveSessions() ?? []; + const sessions = wsServer?.listTrustedSessions() ?? []; if (sessions.length > 0) { await Promise.all( sessions.map((session) => session.archiveWorkspaceRecordForExternalMutation(workspaceId)), @@ -856,20 +875,20 @@ export async function createPaseoDaemon( }; const markWorkspaceArchivingExternal = (workspaceIds: Iterable, archivingAt: string) => { const workspaceIdList = Array.from(workspaceIds); - for (const session of wsServer?.listActiveSessions() ?? []) { + for (const session of wsServer?.listTrustedSessions() ?? []) { session.markWorkspaceArchivingForExternalMutation(workspaceIdList, archivingAt); } }; const clearWorkspaceArchivingExternal = (workspaceIds: Iterable) => { const workspaceIdList = Array.from(workspaceIds); - for (const session of wsServer?.listActiveSessions() ?? []) { + for (const session of wsServer?.listTrustedSessions() ?? []) { session.clearWorkspaceArchivingForExternalMutation(workspaceIdList); } }; const emitWorkspaceUpdatesExternal = async (workspaceIds: Iterable) => { const workspaceIdList = Array.from(workspaceIds); await Promise.all( - (wsServer?.listActiveSessions() ?? []).map((session) => + (wsServer?.listTrustedSessions() ?? []).map((session) => session.emitWorkspaceUpdatesForExternalWorkspaceIds(workspaceIdList), ), ); @@ -949,7 +968,7 @@ export async function createPaseoDaemon( warmWorkspaceGitData: async (workspace) => { await Promise.all( wsServer - ?.listActiveSessions() + ?.listTrustedSessions() .map((session) => session.warmWorkspaceGitDataForWorkspace(workspace)) ?? [], ); }, @@ -988,6 +1007,54 @@ export async function createPaseoDaemon( }; const createAgent = (input: Parameters[1]) => createAgentCommand(createAgentCommandDependencies, input); + const hubAgentLifecycle = new CreateAgentLifecycleDispatch({ + paseoHome: config.paseoHome, + worktreesRoot: config.worktreesRoot, + agentManager, + agentStorage, + github, + workspaceGitService, + createPaseoWorktreeWorkflow: createPaseoWorktreeForTools, + archiveAgentForClose: (agentId) => + archiveAgentCommand({ agentManager, agentStorage, logger }, agentId), + findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal, + listActiveWorkspaces: listActiveWorkspacesExternal, + archiveWorkspaceRecord: archiveWorkspaceRecordExternal, + emit: emitExternalSessionMessage, + emitAgentRemove: () => undefined, + emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal, + markWorkspaceArchiving: markWorkspaceArchivingExternal, + clearWorkspaceArchiving: clearWorkspaceArchivingExternal, + killTerminalsForWorkspace: (workspaceId) => + killTerminalsForWorkspace({ terminalManager, sessionLogger: logger }, workspaceId), + logger, + }); + const hubRelationships = new HubRelationshipController({ + paseoHome: config.paseoHome, + serverId, + daemonPublicKey: daemonKeyPair.publicKeyB64, + logger, + remote: dependencies.hubRelationshipRemote ?? new DirectHubRelationshipRemote(), + clock: dependencies.hubRelationshipClock, + retryPolicy: dependencies.hubRelationshipRetryPolicy, + attachSocket: async (socket, options) => { + if (!wsServer) throw new Error("WebSocket server is not running"); + await wsServer.attachHubSocket(socket, options); + }, + createExecutions: (relationshipId) => + new RelationshipOwnedExecutions({ + relationshipId, + agentManager, + agentStorage, + createAgent, + registerAutoArchive: ({ agentId, createdWorktree }) => + hubAgentLifecycle.registerAutoArchiveIfRequested({ + autoArchive: true, + agentId, + createdWorktree, + }), + }), + }); const loopService = new LoopService({ paseoHome: config.paseoHome, @@ -1381,7 +1448,9 @@ export async function createPaseoDaemon( }, serviceProxyPublicBaseUrl, browserToolsBroker, + hubRelationships, ); + await hubRelationships.start(); if (relayEnabled) { const offer = await createConnectionOfferV2({ @@ -1442,6 +1511,7 @@ export async function createPaseoDaemon( }; const stop = async () => { + await hubRelationships.stop(); scriptHealthMonitor.stop(); // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); @@ -1495,7 +1565,9 @@ async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promi await Promise.all( agents.map(async (agent) => { try { - await agentManager.closeAgent(agent.id); + await agentManager.closeAgent(agent.id, { + persistClosedState: agent.owner?.kind !== "hub", + }); } catch (err) { logger.error({ err, agentId: agent.id }, "Failed to close agent"); } diff --git a/packages/server/src/server/hub/hub-session.test.ts b/packages/server/src/server/hub/hub-session.test.ts new file mode 100644 index 0000000000..e3763de874 --- /dev/null +++ b/packages/server/src/server/hub/hub-session.test.ts @@ -0,0 +1,98 @@ +import { expect, test } from "vitest"; + +import { + AgentSnapshotPayloadSchema, + type AgentSnapshotPayload, + type SessionOutboundMessage, +} from "@getpaseo/protocol/messages"; +import type { + HubAgentCreateInput, + HubExecutions, + OwnedAgentEvent, + OwnedAgentSnapshot, +} from "./relationship-owned-executions.js"; +import { HubSession } from "./hub-session.js"; + +class InMemoryHubExecutions implements HubExecutions { + private readonly listeners = new Set<(event: OwnedAgentEvent) => void>(); + + async create(_input: HubAgentCreateInput): Promise { + return { executionId: "execution-1", agent: createAgentSnapshot() }; + } + + async reconcile(_executionId: string): Promise { + return { executionId: "execution-1", agent: createAgentSnapshot() }; + } + + subscribe(listener: (event: OwnedAgentEvent) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + publishCurrentState(): void { + for (const listener of this.listeners) { + listener({ type: "update", executionId: "execution-1", agent: createAgentSnapshot() }); + } + } +} + +class HubSessionBoundary { + private readonly executions = new InMemoryHubExecutions(); + private readonly messages: SessionOutboundMessage[] = []; + private readonly session = new HubSession({ + executions: this.executions, + send: (message) => this.messages.push(message), + }); + + publishCurrentState(): void { + this.executions.publishCurrentState(); + } + + disconnect(): void { + this.session.cleanup(); + } + + deliveredTypes(): string[] { + return this.messages.map((message) => message.type); + } +} + +test("disconnect unsubscribes Hub from later owned-agent updates", () => { + const hub = new HubSessionBoundary(); + + hub.publishCurrentState(); + hub.disconnect(); + hub.publishCurrentState(); + + expect(hub.deliveredTypes()).toEqual(["hub.agent.update"]); +}); + +function createAgentSnapshot(): AgentSnapshotPayload { + return AgentSnapshotPayloadSchema.parse({ + id: "agent-1", + provider: "codex", + cwd: "/workspace", + model: null, + createdAt: "2026-07-13T00:00:00.000Z", + updatedAt: "2026-07-13T00:00:00.000Z", + lastUserMessageAt: null, + status: "idle", + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: true, + supportsMcpServers: false, + supportsReasoningStream: true, + supportsToolInvocations: true, + supportsRewindConversation: false, + supportsRewindFiles: false, + supportsRewindBoth: false, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: null, + labels: {}, + }); +} diff --git a/packages/server/src/server/hub/hub-session.ts b/packages/server/src/server/hub/hub-session.ts new file mode 100644 index 0000000000..8998193cb5 --- /dev/null +++ b/packages/server/src/server/hub/hub-session.ts @@ -0,0 +1,142 @@ +import type { + HubAgentCreateRequest, + HubExecutionReconcileRequest, + SessionInboundMessage, + SessionOutboundMessage, +} from "@getpaseo/protocol/messages"; + +import type { HubExecutions, OwnedAgentEvent } from "./relationship-owned-executions.js"; + +interface HubSessionOptions { + executions: HubExecutions; + send: (message: SessionOutboundMessage) => void; +} + +export class HubSession { + private readonly executions: HubExecutions; + private readonly send: (message: SessionOutboundMessage) => void; + private readonly unsubscribe: () => void; + + constructor(options: HubSessionOptions) { + this.executions = options.executions; + this.send = options.send; + this.unsubscribe = this.executions.subscribe((event) => this.sendOwnedEvent(event)); + } + + async handleMessage(message: SessionInboundMessage): Promise { + if (message.type === "hub.agent.create.request") { + await this.createAgent(message); + return; + } + if (message.type === "hub.execution.reconcile.request") { + await this.reconcile(message); + return; + } + const requestId = requestIdForMessage(message); + this.send({ + type: "hub.authorization.denied", + payload: { + ...(requestId ? { requestId } : {}), + requestType: message.type, + code: "scope_denied", + }, + }); + } + + cleanup(): void { + this.unsubscribe(); + } + + private async createAgent(message: HubAgentCreateRequest): Promise { + try { + const result = await this.executions.create({ + executionId: message.executionId, + provider: message.provider, + cwd: message.cwd, + workspaceId: message.workspaceId, + prompt: message.prompt, + model: message.model, + modeId: message.modeId, + thinkingOptionId: message.thinkingOptionId, + featureValues: message.featureValues, + env: message.env, + worktree: message.worktree, + autoArchive: message.autoArchive, + }); + this.send({ + type: "hub.agent.create.response", + payload: { + requestId: message.requestId, + executionId: message.executionId, + agentId: result.agent.id, + agent: result.agent, + success: true, + error: null, + }, + }); + } catch (error) { + this.send({ + type: "hub.agent.create.response", + payload: { + requestId: message.requestId, + executionId: message.executionId, + agentId: null, + agent: null, + success: false, + error: error instanceof Error ? error.message : String(error), + }, + }); + } + } + + private async reconcile(message: HubExecutionReconcileRequest): Promise { + const result = await this.executions.reconcile(message.executionId); + this.send({ + type: "hub.execution.reconcile.response", + payload: { + requestId: message.requestId, + executionId: message.executionId, + agentId: result?.agent.id ?? null, + agent: result?.agent ?? null, + }, + }); + } + + private sendOwnedEvent(event: OwnedAgentEvent): void { + if (event.type === "update") { + this.send({ + type: "hub.agent.update", + payload: { + executionId: event.executionId, + agentId: event.agent.id, + agent: event.agent, + }, + }); + return; + } + this.send({ + type: "hub.agent.stream", + payload: { + executionId: event.executionId, + agentId: event.agentId, + event: event.event, + }, + }); + } +} + +function requestIdForMessage(message: SessionInboundMessage): string | undefined { + if ("requestId" in message && typeof message.requestId === "string") { + return message.requestId; + } + if ( + "payload" in message && + typeof message.payload === "object" && + message.payload !== null && + "requestId" in message.payload && + typeof message.payload.requestId === "string" + ) { + return message.payload.requestId; + } + return undefined; +} diff --git a/packages/server/src/server/hub/hub-session.websocket.test.ts b/packages/server/src/server/hub/hub-session.websocket.test.ts new file mode 100644 index 0000000000..88b0cff34e --- /dev/null +++ b/packages/server/src/server/hub/hub-session.websocket.test.ts @@ -0,0 +1,144 @@ +import { afterEach, expect, test } from "vitest"; +import { + HubRelationshipHarness, + SetupFailingArchiveWatchFiles, +} from "./test-utils/relationship-harness.js"; + +let relationship: HubRelationshipHarness | null = null; + +afterEach(async () => { + await relationship?.close(); + relationship = null; +}); + +async function launchRelationship(): Promise { + const launched = await HubRelationshipHarness.start(); + await launched.beginConnect().result; + launched.connectLatestSocket(); + relationship = launched; + return launched; +} + +test("Hub owns one durable execution across concurrency and reconstruction", async () => { + const hub = await launchRelationship(); + + const created = await hub.createOwnedConcurrently(); + const update = await hub.ownedUpdate(created.first.agentId); + const stream = await hub.ownedStream(created.first.agentId); + const reconciled = await hub.reconcileOwned(); + const reconstructed = await hub.reconstructAndReplay(); + + expect(created.duplicate.agentId).toBe(created.first.agentId); + expect(update).toMatchObject({ + executionId: "execution-1", + agentId: created.first.agentId, + agent: { id: created.first.agentId }, + }); + expect(stream).toMatchObject({ executionId: "execution-1", agentId: created.first.agentId }); + expect(reconciled).toMatchObject({ + executionId: "execution-1", + agentId: created.first.agentId, + agent: { id: created.first.agentId }, + }); + expect(reconstructed.replay.agent.id).toBe(created.first.agentId); + expect(reconstructed.reconciliation?.agent.id).toBe(created.first.agentId); + expect(reconstructed.durableAgentCount).toBe(1); +}); + +test("Hub denies trusted steering and browser dispatch", async () => { + const hub = await launchRelationship(); + const localAgentId = await hub.createUnrelatedLocalAgent(); + + const steeringDenial = await hub.deniedSteering(localAgentId); + const browserDenial = await hub.deniedBrowserDispatch(); + + expect(steeringDenial).toEqual({ + requestId: "denied-steer", + requestType: "send_agent_message_request", + code: "scope_denied", + }); + expect(browserDenial).toEqual({ + requestId: "browser-1", + requestType: "browser.automation.execute.response", + code: "scope_denied", + }); + expect(hub.observedAgentIds()).not.toContain(localAgentId); + expect(hub.observedTrustedLifecycleMessages()).toEqual([]); +}); + +test("Hub sockets reject trusted hello and capabilities", async () => { + const hub = await launchRelationship(); + + expect(hub.probeTrustedHello()).toBe(4002); +}); + +test("Hub sockets reject trusted binary frames", async () => { + const hub = await launchRelationship(); + + expect(hub.probeBinaryFrame()).toBe(4002); +}); + +test("Hub does not receive trusted broadcasts", async () => { + const hub = await launchRelationship(); + + const trustedBroadcasts = await hub.trustedBroadcastCount(); + const trustedStatus = await hub.trustedDaemonStatus(); + + expect(trustedBroadcasts).toBe(0); + expect(trustedStatus).toMatchObject({ pid: process.pid, relay: { enabled: false } }); + expect(hub.observedTrustedLifecycleMessages()).toEqual([]); +}); + +test("Hub reconnects without retaining trusted session state", async () => { + const hub = await launchRelationship(); + const created = await hub.createOwnedConcurrently(); + + const reconnected = await hub.reconnectAndReconcile(); + + expect(reconnected).toMatchObject({ + executionId: "execution-1", + agentId: created.first.agentId, + }); + expect(hub.observedTrustedLifecycleMessages()).toEqual([]); +}); + +test("Hub create forwards worktree and auto-archive through the existing create path", async () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("worktree-create", "execution-worktree", { + worktree: { mode: "branch-off", newBranch: "hub-created-worktree", base: "main" }, + autoArchive: true, + }); + const worktreeCreated = await hub.ownedCreateResult("worktree-create"); + const worktreeCwd = hub.latestCreatedCwd(); + const duringRun = await hub.worktreeState(worktreeCwd!); + const archive = await hub.waitForOwnedArchiveCompletion(worktreeCreated.payload.agentId!); + const afterArchive = await hub.worktreeState(worktreeCwd!); + + expect(worktreeCreated).toMatchObject({ + type: "hub.agent.create.response", + payload: { success: true, agent: { cwd: worktreeCwd } }, + }); + expect(worktreeCwd).not.toBe(hub.repoRoot()); + expect(duringRun).toEqual({ exists: true, listed: true }); + expect(afterArchive).toEqual({ exists: false, listed: false }); + expect(archive).toEqual({ + agentArchivedAt: expect.any(String), + workspaceArchivedAt: expect.any(String), + }); +}, 20_000); + +test("archive observation closes its first watcher when the second watcher cannot start", async () => { + const watchFiles = new SetupFailingArchiveWatchFiles(2); + const hub = await HubRelationshipHarness.start(watchFiles); + relationship = hub; + await hub.beginConnect().result; + hub.connectLatestSocket(); + hub.beginOwnedCreate("watch-setup-create", "watch-setup-execution"); + const created = await hub.ownedCreateResult("watch-setup-create"); + + await expect(hub.waitForOwnedArchiveCompletion(created.payload.agentId!)).rejects.toThrow( + "Cannot watch", + ); + + expect(watchFiles.activeDirectories()).toEqual([]); +}); diff --git a/packages/server/src/server/hub/index.ts b/packages/server/src/server/hub/index.ts new file mode 100644 index 0000000000..2deb939266 --- /dev/null +++ b/packages/server/src/server/hub/index.ts @@ -0,0 +1,8 @@ +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 new file mode 100644 index 0000000000..5c07d983e1 --- /dev/null +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -0,0 +1,314 @@ +import { afterEach, describe, expect, test } from "vitest"; +import { createHash } from "node:crypto"; +import { HubRelationshipHarness } from "./test-utils/relationship-harness.js"; + +describe("Hub relationship", () => { + let relationship: HubRelationshipHarness | null = null; + + afterEach(async () => { + await relationship?.close(); + relationship = null; + }); + + test("the CLI connects, reports status, and disconnects through the daemon", async () => { + relationship = await HubRelationshipHarness.start(); + const connected = await relationship.beginConnect().result; + relationship.connectLatestSocket(); + + const status = await relationship.status(); + const enrollment = relationship.enrollmentAttempts()[0]; + const secret = relationship.relationshipFile()?.credential?.secret; + const disconnected = await relationship.disconnect(); + + expect(connected.state).toBe("connecting"); + expect(status.state).toBe("connected"); + expect(relationship.loggableValues(status)).not.toContain(secret); + expect(relationship.loggableValues(status)).not.toContain(enrollment.credentialVerifier); + expect(relationship.loggableValues(status)).not.toContain(enrollment.token); + expect(relationship.loggableValues(status)).not.toContain(enrollment.idempotencyKey); + expect(disconnected.state).toBe("not_connected"); + }); + + test("an external socket cannot manage Hub relationships even when it claims to be the CLI", async () => { + relationship = await HubRelationshipHarness.start(); + + const denials = await relationship.attemptExternalManagementAsCli(); + + expect(denials).toEqual([ + { + type: "rpc_error", + payload: { + requestId: "external-hub-connect", + requestType: "hub.relationship.connect.request", + error: "Hub relationship management requires a local daemon connection", + code: "local_management_required", + }, + }, + { + type: "rpc_error", + payload: { + requestId: "external-hub-status", + requestType: "hub.relationship.get_status.request", + error: "Hub relationship management requires a local daemon connection", + code: "local_management_required", + }, + }, + { + type: "rpc_error", + payload: { + requestId: "external-hub-disconnect", + requestType: "hub.relationship.disconnect.request", + error: "Hub relationship management requires a local daemon connection", + code: "local_management_required", + }, + }, + ]); + expect(relationship.enrollmentAttempts()).toEqual([]); + expect(relationship.relationshipFile()).toBeNull(); + }); + + test("a loopback browser socket cannot manage Hub relationships by claiming to be the CLI", async () => { + relationship = await HubRelationshipHarness.start(); + + const denial = await relationship.attemptLoopbackBrowserConnectAsCli(); + + expect(denial).toEqual({ + type: "rpc_error", + payload: { + requestId: "browser-hub-connect", + requestType: "hub.relationship.connect.request", + error: "Hub relationship management requires a local daemon connection", + code: "local_management_required", + }, + }); + expect(relationship.enrollmentAttempts()).toEqual([]); + expect(relationship.relationshipFile()).toBeNull(); + }); + + test("persists private generated authority before enrollment and active before dialing", async () => { + relationship = await HubRelationshipHarness.start(); + relationship.holdEnrollment(); + const connecting = relationship.beginConnect("one-time-token"); + + const enrollment = await relationship.enrollmentBegins(); + const pending = relationship.enrollmentInvocation(); + + expect(pending).toMatchObject({ + mode: 0o600, + record: { + state: "pending", + relationship: { id: enrollment.relationshipId, idempotencyKey: enrollment.idempotencyKey }, + credential: { secret: expect.any(String) }, + enrollment: { token: "one-time-token" }, + identity: { serverId: expect.any(String), daemonPublicKey: expect.any(String) }, + }, + }); + const secret = pending.record.credential?.secret; + expect(secret).toEqual(expect.any(String)); + expect(enrollment.credentialVerifier).toBe( + createHash("sha256") + .update(secret ?? "") + .digest("base64url"), + ); + relationship.completeEnrollment(); + await connecting.result; + await relationship.socketDialed(); + expect(relationship.socketInvocation()).toMatchObject({ + mode: 0o600, + record: { state: "active", relationship: { id: enrollment.relationshipId } }, + }); + }); + + test("a lost enrollment response reuses the exact ceremony", async () => { + relationship = await HubRelationshipHarness.start(); + relationship.holdEnrollment(); + const connecting = relationship.beginConnect("same-token"); + await relationship.enrollmentBegins(); + relationship.loseEnrollmentResponse(); + await connecting.result; + + await relationship.retry(); + const attempts = relationship.enrollmentAttempts(); + + expect(attempts).toHaveLength(2); + expect(attempts[1]).toEqual(attempts[0]); + expect(relationship.enrolledRelationships()).toBe(1); + }); + + test("daemon restart reconnects the same durable relationship", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + const id = relationship.relationshipFile()?.relationship.id; + relationship.connectLatestSocket(); + await relationship.socketDialed(); + + await relationship.restartDaemon(); + await relationship.socketDialed(); + + expect(relationship.relationshipFile()?.relationship.id).toBe(id); + expect(relationship.socketAttempts()).toBe(2); + }); + + test("daemon restart reconstructs an owned agent that was running without creating another", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + await relationship.socketDialed(); + relationship.connectLatestSocket(); + const relationshipId = relationship.relationshipFile()?.relationship.id; + relationship.beginOwnedCreate("running-create", "execution-running", { + prompt: "sleep 30", + modeId: "full-access", + }); + const created = await relationship.ownedCreateResult("running-create"); + const running = await relationship.ownedRunningUpdate(created.payload.agentId!); + + await relationship.restartDaemon(); + await relationship.socketDialed(); + relationship.connectLatestSocket(); + const reconciled = await relationship.reconcileOwned("execution-running"); + const durableAgentIds = await relationship.durableOwnedAgentIds(); + + expect(running).toMatchObject({ + executionId: "execution-running", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "running" }, + }); + expect(relationship.relationshipFile()?.relationship.id).toBe(relationshipId); + expect(reconciled).toMatchObject({ + executionId: "execution-running", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "running" }, + }); + expect(durableAgentIds).toEqual([created.payload.agentId]); + expect(relationship.providerCreations()).toBe(1); + }); + + test("stale socket generations cannot replace or unregister the current socket", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.closeSocket(0, 1006); + await relationship.retry(); + relationship.connectSocket(1); + + relationship.connectSocket(0); + relationship.closeSocket(0, 1000); + const messages = relationship.sendHubRequestOnLatest({ + type: "daemon.get_status.request", + requestId: "still-current", + }); + + expect(messages).toContainEqual({ + type: "hub.authorization.denied", + payload: { + requestId: "still-current", + requestType: "daemon.get_status.request", + code: "scope_denied", + }, + }); + expect(relationship.socketAttempts()).toBe(2); + }); + + test("replayed create across socket generations shares one pending durable execution", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + relationship.holdAgentCreation(); + relationship.beginOwnedCreate("first-create"); + await relationship.agentCreationAttempts(1); + + relationship.closeLatestSocket(1006); + await relationship.retry(); + relationship.connectLatestSocket(); + relationship.beginOwnedCreate("replayed-create"); + relationship.finishAgentCreation(); + + const replayed = await relationship.ownedCreateResult("replayed-create"); + const durableAgentIds = await relationship.durableOwnedAgentIds(); + + expect(relationship.socketDeliveredResponse(0, "first-create")).toBe(false); + expect(replayed).toMatchObject({ + type: "hub.agent.create.response", + payload: { + success: true, + executionId: "execution-race", + agentId: durableAgentIds[0], + }, + }); + expect(relationship.providerCreations()).toBe(1); + expect(durableAgentIds).toHaveLength(1); + }); + + test.each([ + [4403, "Hub revoked this relationship"], + [401, "Hub rejected socket authentication (401)"], + [403, "Hub rejected socket authentication (403)"], + ] as const)("authentication rejection %s revokes permanently", async (code, reason) => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect("private-token").result; + const enrollment = relationship.enrollmentAttempts()[0]; + const secret = relationship.relationshipFile()?.credential?.secret; + relationship.rejectRelationship(code); + + await relationship.restartDaemon(); + const status = await relationship.status(); + const persisted = relationship.relationshipFile(); + + expect(status).toMatchObject({ + state: "revoked", + relationshipId: enrollment.relationshipId, + hub: "https://hub.test", + scopes: "hub.*", + error: reason, + }); + expect(persisted?.state).toBe("revoked"); + expect(persisted?.relationship).toMatchObject({ + id: enrollment.relationshipId, + hubOrigin: "https://hub.test", + scopes: ["hub.*"], + }); + expect(persisted?.reason).toBe(reason); + expect(persisted).not.toHaveProperty("credential"); + expect(persisted).not.toHaveProperty("relationship.idempotencyKey"); + expect(relationship.socketAttempts()).toBe(1); + const loggable = relationship.loggableValues(status); + const reconstructed = JSON.stringify({ status, persisted }); + expect(reconstructed).not.toContain(secret); + expect(reconstructed).not.toContain(enrollment.credentialVerifier); + expect(reconstructed).not.toContain(enrollment.token); + expect(reconstructed).not.toContain(enrollment.idempotencyKey); + expect(loggable).not.toContain(secret); + expect(loggable).not.toContain(enrollment.credentialVerifier); + expect(loggable).not.toContain(enrollment.token); + expect(loggable).not.toContain(enrollment.idempotencyKey); + }); + + test("offline disconnect retries across runtime and restart without opening a socket", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + relationship.failRevocations(3); + + const disconnecting = await relationship.disconnect(); + await relationship.retry(); + await relationship.restartDaemon(); + await relationship.retry(); + + expect(disconnecting.state).toBe("disconnecting"); + expect(relationship.revocationAttempts()).toBe(4); + expect(relationship.socketAttempts()).toBe(1); + expect(relationship.relationshipFile()).toBeNull(); + }); + + test("force disconnect removes local authority and reports the remote warning", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.failRevocations(1); + await relationship.disconnect(); + + const forced = await relationship.disconnect(true); + + expect(forced.state).toBe("not_connected"); + expect(forced.warning).toContain("remote revocation"); + expect(relationship.relationshipFile()).toBeNull(); + }); +}); diff --git a/packages/server/src/server/hub/relationship-controller.ts b/packages/server/src/server/hub/relationship-controller.ts new file mode 100644 index 0000000000..8682c35733 --- /dev/null +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -0,0 +1,426 @@ +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { existsSync, readFileSync, rmSync } from "node:fs"; +import path from "node:path"; +import type pino from "pino"; +import { z } from "zod"; +import { ensurePrivateFile, writePrivateFileAtomicSync } from "../private-files.js"; +import type { WebSocketLike } from "../websocket-server.js"; +import type { HubExecutions } from "./relationship-owned-executions.js"; +import type { + HubRelationshipRemote, + HubSocketConnection, + HubSocketEvents, +} from "./relationship-remote.js"; +import { HubEnrollmentRejectedError } from "./relationship-remote.js"; +import { BoundedExponentialHubRetryPolicy } from "./relationship-retry.js"; + +const FILE_NAME = "hub-relationship.json"; +const SCOPES = ["hub.*"] as const; + +const RelationshipSchema = z.object({ + id: z.string().min(1), + idempotencyKey: z.string().min(1), + hubOrigin: z.string().url(), + createdAt: z.string(), + scopes: z.array(z.string()), +}); +const SanitizedRelationshipSchema = RelationshipSchema.omit({ idempotencyKey: true }); +const CredentialSchema = z.object({ secret: z.string().min(1) }); +const TransportSchema = z.object({ + kind: z.literal("direct_websocket"), + webSocketUrl: z.string().url(), +}); +const PendingSchema = z.object({ + version: z.literal(1), + state: z.literal("pending"), + relationship: RelationshipSchema, + credential: CredentialSchema, + enrollment: z.object({ token: z.string().min(1) }), + identity: z.object({ serverId: z.string().min(1), daemonPublicKey: z.string().min(1) }), +}); +const ActiveSchema = z.object({ + version: z.literal(1), + state: z.enum(["active", "disconnecting"]), + relationship: RelationshipSchema, + credential: CredentialSchema, + transport: TransportSchema, +}); +const RevokedSchema = z.object({ + version: z.literal(1), + state: z.literal("revoked"), + relationship: SanitizedRelationshipSchema, + transport: TransportSchema.optional(), + reason: z.string().optional(), +}); +const RecordSchema = z.discriminatedUnion("state", [PendingSchema, ActiveSchema, RevokedSchema]); +type PendingRecord = z.infer; +type ActiveRecord = z.infer; +type RevokedRecord = z.infer; +type HubRelationshipRecord = z.infer; + +export type HubConnectionState = + | "not_connected" + | "connecting" + | "connected" + | "reconnecting" + | "disconnecting" + | "revoked"; + +export interface HubRelationshipStatus { + state: HubConnectionState; + relationshipId: string | null; + hubOrigin: string | null; + scopes: string[]; + connectedAt: string | null; + lastError: string | null; +} + +export interface HubRelationshipManagement { + connect(input: { hubUrl: string; token: string }): Promise; + status(): HubRelationshipStatus; + disconnect(input: { + force: boolean; + }): Promise<{ status: HubRelationshipStatus; warning?: string }>; +} + +export interface ScheduledRelationshipTask { + cancel(): void; +} + +export interface HubRelationshipClock { + now(): Date; + schedule(delayMs: number, task: () => void): ScheduledRelationshipTask; +} + +export interface HubRelationshipRetryPolicy { + delay(attempt: number): number; +} + +export interface HubRelationshipControllerOptions { + paseoHome: string; + serverId: string; + daemonPublicKey: string; + logger: pino.Logger; + remote: HubRelationshipRemote; + clock?: HubRelationshipClock; + retryPolicy?: HubRelationshipRetryPolicy; + attachSocket: ( + socket: WebSocketLike, + options: { relationshipId: string; executions: HubExecutions }, + ) => Promise; + createExecutions: (relationshipId: string) => HubExecutions; +} + +const systemClock: HubRelationshipClock = { + now: () => new Date(), + schedule(delayMs, task) { + const timer = setTimeout(task, delayMs); + timer.unref?.(); + return { cancel: () => clearTimeout(timer) }; + }, +}; + +function normalizeHubUrl(value: string): string { + const url = new URL(value); + if (url.protocol !== "http:" && url.protocol !== "https:") { + throw new Error("Hub URL must use HTTP or HTTPS"); + } + url.pathname = url.pathname.replace(/\/$/, ""); + url.search = ""; + url.hash = ""; + return url.toString().replace(/\/$/, ""); +} + +export class HubRelationshipController implements HubRelationshipManagement { + private readonly filePath: string; + private readonly clock: HubRelationshipClock; + private readonly retryPolicy: HubRelationshipRetryPolicy; + private record: HubRelationshipRecord | null; + private state: HubConnectionState = "not_connected"; + private connectedAt: string | null = null; + private lastError: string | null = null; + private socket: HubSocketConnection | null = null; + private retry: ScheduledRelationshipTask | null = null; + private generation = 0; + private retryAttempt = 0; + private executions: { relationshipId: string; value: HubExecutions } | null = null; + + constructor(private readonly options: HubRelationshipControllerOptions) { + this.filePath = path.join(options.paseoHome, FILE_NAME); + this.clock = options.clock ?? systemClock; + this.retryPolicy = options.retryPolicy ?? new BoundedExponentialHubRetryPolicy(); + this.record = this.load(); + if (this.record?.state === "revoked") { + this.state = "revoked"; + this.lastError = this.record.reason ?? null; + } else if (this.record?.state === "disconnecting") this.state = "disconnecting"; + else if (this.record) this.state = "connecting"; + } + + 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 === "disconnecting") await this.tryRevocation(this.record); + } + + async stop(): Promise { + this.cancelLifecycle(); + this.socket?.close(); + this.socket = null; + } + + status(): HubRelationshipStatus { + return { + state: this.state, + relationshipId: this.record?.relationship.id ?? null, + hubOrigin: this.record?.relationship.hubOrigin ?? null, + scopes: this.record?.relationship.scopes.slice() ?? [], + connectedAt: this.connectedAt, + lastError: this.lastError, + }; + } + + async connect(input: { hubUrl: string; token: string }): Promise { + if (this.record?.state === "pending") { + if (normalizeHubUrl(input.hubUrl) !== this.record.relationship.hubOrigin) { + throw new Error("A pending Hub enrollment already exists for a different Hub"); + } + await this.tryEnrollment(this.record); + return this.status(); + } + if (this.record && this.record.state !== "revoked") { + throw new Error("This daemon already has a Hub relationship"); + } + const pending: PendingRecord = { + version: 1, + state: "pending", + relationship: { + id: randomUUID(), + idempotencyKey: randomUUID(), + hubOrigin: normalizeHubUrl(input.hubUrl), + createdAt: this.clock.now().toISOString(), + scopes: [...SCOPES], + }, + credential: { secret: randomBytes(32).toString("base64url") }, + enrollment: { token: input.token }, + identity: { serverId: this.options.serverId, daemonPublicKey: this.options.daemonPublicKey }, + }; + this.persist(pending); + this.record = pending; + this.state = "connecting"; + await this.tryEnrollment(pending); + return this.status(); + } + + async disconnect(input: { + force: boolean; + }): Promise<{ status: HubRelationshipStatus; warning?: string }> { + this.cancelLifecycle(); + this.socket?.close(); + this.socket = null; + if (!this.record || this.record.state === "revoked") { + this.remove(); + return { status: this.status() }; + } + if (input.force) { + this.remove(); + return { + status: this.status(), + 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" }; + this.persist(disconnecting); + this.record = disconnecting; + this.state = "disconnecting"; + await this.tryRevocation(disconnecting); + return { status: this.status() }; + } + + private async tryEnrollment(pending: PendingRecord): Promise { + const generation = this.generation; + const verifier = createHash("sha256").update(pending.credential.secret).digest("base64url"); + try { + const enrollment = await this.options.remote.enroll({ + relationshipId: pending.relationship.id, + idempotencyKey: pending.relationship.idempotencyKey, + hubOrigin: pending.relationship.hubOrigin, + token: pending.enrollment.token, + serverId: pending.identity.serverId, + daemonPublicKey: pending.identity.daemonPublicKey, + credentialVerifier: verifier, + scopes: pending.relationship.scopes, + }); + if (generation !== this.generation) return; + if ( + enrollment.relationshipId !== pending.relationship.id || + !enrollment.scopes.includes("hub.*") + ) { + throw new Error("Hub enrollment response did not match the pending relationship"); + } + const active: ActiveRecord = { + version: 1, + state: "active", + relationship: { ...pending.relationship, scopes: enrollment.scopes }, + credential: pending.credential, + transport: { kind: "direct_websocket", webSocketUrl: enrollment.webSocketUrl }, + }; + this.persist(active); + this.record = active; + this.retryAttempt = 0; + this.openSocket(active, false); + } catch (error) { + if (generation !== this.generation) return; + if (error instanceof HubEnrollmentRejectedError) { + this.remove(); + throw error; + } + this.lastError = error instanceof Error ? error.message : String(error); + this.scheduleEnrollment(pending); + } + } + + private openSocket(record: ActiveRecord, reconnecting: boolean): void { + const generation = ++this.generation; + this.state = reconnecting ? "reconnecting" : "connecting"; + const events: HubSocketEvents = { + 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; + }, + }; + this.socket = this.options.remote.openSocket( + { + relationshipId: record.relationship.id, + webSocketUrl: record.transport.webSocketUrl, + credential: record.credential.secret, + }, + events, + ); + } + + private socketConnected(generation: number, record: ActiveRecord, socket: WebSocketLike): void { + if (generation !== this.generation) { + socket.close(); + return; + } + this.retryAttempt = 0; + this.state = "connected"; + this.connectedAt = this.clock.now().toISOString(); + this.lastError = null; + void this.options.attachSocket(socket, { + relationshipId: record.relationship.id, + executions: this.executionsFor(record.relationship.id), + }); + } + + private executionsFor(relationshipId: string): HubExecutions { + if (this.executions?.relationshipId === relationshipId) return this.executions.value; + const value = this.options.createExecutions(relationshipId); + this.executions = { relationshipId, value }; + return value; + } + + private socketRejected(generation: number, statusCode: 401 | 403): void { + if (generation !== this.generation) return; + this.revoke(`Hub rejected socket authentication (${statusCode})`); + } + + private socketClosed(generation: number, record: ActiveRecord, code: number): void { + if (generation !== this.generation) return; + if (code === 4403) { + this.revoke("Hub revoked this relationship"); + return; + } + if (this.record?.state === "active") this.scheduleSocket(record); + } + + private scheduleSocket(record: ActiveRecord): void { + this.state = "reconnecting"; + this.schedule(() => this.openSocket(record, true)); + } + + private scheduleEnrollment(record: PendingRecord): void { + this.state = "reconnecting"; + this.schedule(() => void this.tryEnrollment(record)); + } + + private async tryRevocation(record: ActiveRecord): Promise { + const generation = this.generation; + try { + await this.options.remote.revoke({ + relationshipId: record.relationship.id, + hubOrigin: record.relationship.hubOrigin, + credential: record.credential.secret, + }); + if (generation !== this.generation) return; + this.remove(); + } catch (error) { + if (generation !== this.generation) return; + this.lastError = error instanceof Error ? error.message : String(error); + this.state = "disconnecting"; + this.schedule(() => void this.tryRevocation(record)); + } + } + + private schedule(task: () => void): void { + this.retry?.cancel(); + const generation = this.generation; + const delay = this.retryPolicy.delay(this.retryAttempt++); + this.retry = this.clock.schedule(delay, () => { + if (generation === this.generation) task(); + }); + } + + private revoke(reason: string): void { + this.cancelLifecycle(); + if (!this.record) return; + const revoked: RevokedRecord = { + version: 1, + state: "revoked", + relationship: { + id: this.record.relationship.id, + hubOrigin: this.record.relationship.hubOrigin, + createdAt: this.record.relationship.createdAt, + scopes: this.record.relationship.scopes, + }, + transport: "transport" in this.record ? this.record.transport : undefined, + reason, + }; + this.persist(revoked); + this.record = revoked; + this.state = "revoked"; + this.lastError = reason; + } + + private cancelLifecycle(): void { + ++this.generation; + this.retry?.cancel(); + this.retry = null; + } + + private persist(record: HubRelationshipRecord): void { + writePrivateFileAtomicSync(this.filePath, `${JSON.stringify(record, null, 2)}\n`); + } + + private remove(): void { + this.cancelLifecycle(); + rmSync(this.filePath, { force: true }); + this.record = null; + this.state = "not_connected"; + this.connectedAt = null; + } + + private load(): HubRelationshipRecord | null { + if (!existsSync(this.filePath)) return null; + const record = RecordSchema.parse(JSON.parse(readFileSync(this.filePath, "utf8"))); + ensurePrivateFile(this.filePath); + return 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 new file mode 100644 index 0000000000..e2320f61b2 --- /dev/null +++ b/packages/server/src/server/hub/relationship-owned-executions.test.ts @@ -0,0 +1,38 @@ +import { afterEach, expect, test } from "vitest"; +import { HubRelationshipHarness } from "./test-utils/relationship-harness.js"; + +let relationship: HubRelationshipHarness | null = null; + +afterEach(async () => { + await relationship?.close(); + relationship = null; +}); + +async function launchRelationship(): Promise { + const launched = await HubRelationshipHarness.start(); + await launched.beginConnect().result; + launched.connectLatestSocket(); + relationship = launched; + return launched; +} + +test("sequential replay after reconstruction keeps one durable owned agent", async () => { + const hub = await launchRelationship(); + const created = await hub.createOwnedConcurrently(); + + const reconstructed = await hub.reconstructAndReplay(); + + expect(reconstructed.replay.agent.id).toBe(created.first.agentId); + expect(reconstructed.reconciliation?.agent.id).toBe(created.first.agentId); + expect(reconstructed.durableAgentCount).toBe(1); +}); + +test("removing an owned agent removes its reconstructed execution association", async () => { + const hub = await launchRelationship(); + const created = await hub.createOwnedConcurrently(); + + const removed = await hub.removeAndReconcile(created.first.agentId); + + expect(removed.reconciliation).toBeNull(); + expect(removed.durableAgentCount).toBe(0); +}); diff --git a/packages/server/src/server/hub/relationship-owned-executions.ts b/packages/server/src/server/hub/relationship-owned-executions.ts new file mode 100644 index 0000000000..659c4f66ce --- /dev/null +++ b/packages/server/src/server/hub/relationship-owned-executions.ts @@ -0,0 +1,219 @@ +import type { + AgentSnapshotPayload, + AgentStreamEventPayload, + CreateAgentWorktreeTarget, +} from "@getpaseo/protocol/messages"; + +import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js"; +import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js"; +import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js"; +import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js"; +import { buildStoredAgentPayload } from "../agent/agent-projections.js"; +import { serializeAgentSnapshot, serializeAgentStreamEvent } from "../messages.js"; +import { hubExecutionKey, type HubAgentOwner } from "../agent/agent-owner.js"; + +export interface HubAgentCreateInput { + executionId: string; + provider: string; + cwd: string; + workspaceId?: string; + prompt: string; + model?: string; + modeId?: string; + thinkingOptionId?: string; + featureValues?: Record; + env?: Record; + worktree?: CreateAgentWorktreeTarget; + autoArchive?: boolean; +} + +export interface OwnedAgentSnapshot { + executionId: string; + agent: AgentSnapshotPayload; +} + +export type OwnedAgentEvent = + | { type: "update"; executionId: string; agent: AgentSnapshotPayload } + | { + type: "stream"; + executionId: string; + agentId: string; + event: AgentStreamEventPayload; + }; + +interface RelationshipOwnedExecutionsOptions { + relationshipId: string; + agentManager: AgentManager; + agentStorage: AgentStorage; + createAgent: BoundCreateAgentCommand; + registerAutoArchive?: (input: { + agentId: string; + createdWorktree: CreatePaseoWorktreeWorkflowResult | null; + }) => void; +} + +export interface HubExecutions { + create(input: HubAgentCreateInput): Promise; + reconcile(executionId: string): Promise; + subscribe(listener: (event: OwnedAgentEvent) => void): () => void; +} + +export class RelationshipOwnedExecutions implements HubExecutions { + private readonly relationshipId: string; + private readonly agentManager: AgentManager; + private readonly agentStorage: AgentStorage; + private readonly createAgentCommand: BoundCreateAgentCommand; + private readonly pendingCreates = new Map>(); + private readonly registerAutoArchive: (input: { + agentId: string; + createdWorktree: CreatePaseoWorktreeWorkflowResult | null; + }) => void; + + constructor(options: RelationshipOwnedExecutionsOptions) { + this.relationshipId = options.relationshipId; + this.agentManager = options.agentManager; + this.agentStorage = options.agentStorage; + this.createAgentCommand = options.createAgent; + this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined); + } + + create(input: HubAgentCreateInput): Promise { + const owner = this.owner(input.executionId); + const key = hubExecutionKey(owner); + const pending = this.pendingCreates.get(key); + if (pending) { + return pending; + } + + const create = this.createOrResolve(owner, input).finally(() => { + if (this.pendingCreates.get(key) === create) { + this.pendingCreates.delete(key); + } + }); + this.pendingCreates.set(key, create); + return create; + } + + async reconcile(executionId: string): Promise { + const record = await this.agentStorage.findByHubExecution(this.owner(executionId)); + return record ? this.projectRecord(record) : null; + } + + subscribe(listener: (event: OwnedAgentEvent) => void): () => void { + return this.agentManager.subscribe( + (event) => { + const owned = this.projectEvent(event); + if (owned) { + listener(owned); + } + }, + { replayState: true }, + ); + } + + private async createOrResolve( + owner: HubAgentOwner, + input: HubAgentCreateInput, + ): Promise { + const existing = await this.agentStorage.findByHubExecution(owner); + if (existing) { + 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) } + : {}), + }); + + return { + executionId: owner.executionId, + agent: serializeAgentSnapshot(result.liveSnapshot), + }; + } + + private projectRecord(record: StoredAgentRecord): OwnedAgentSnapshot { + const owner = record.owner; + if (owner?.kind !== "hub" || owner.relationshipId !== this.relationshipId) { + throw new Error(`Agent ${record.id} is not owned by Hub relationship ${this.relationshipId}`); + } + const live = this.agentManager.getAgent(record.id); + return { + executionId: owner.executionId, + agent: live + ? serializeAgentSnapshot(live) + : buildStoredAgentPayload(record, this.agentManager.getRegisteredProviderIds()), + }; + } + + private projectEvent(event: AgentManagerEvent): OwnedAgentEvent | null { + if (event.type === "agent_state") { + return this.projectAgentState(event.agent); + } + if (event.type !== "agent_stream") { + return null; + } + const agent = this.agentManager.getAgent(event.agentId); + if (!this.isOwned(agent)) { + return null; + } + const serialized = serializeAgentStreamEvent(event.event); + if (!serialized) { + return null; + } + return { + type: "stream", + executionId: agent.owner.executionId, + agentId: agent.id, + event: serialized, + }; + } + + private projectAgentState(agent: ManagedAgent): OwnedAgentEvent | null { + if (!this.isOwned(agent)) { + return null; + } + return { + type: "update", + executionId: agent.owner.executionId, + agent: serializeAgentSnapshot(agent), + }; + } + + private isOwned(agent: ManagedAgent | null): agent is ManagedAgent & { owner: HubAgentOwner } { + return agent?.owner?.kind === "hub" && agent.owner.relationshipId === this.relationshipId; + } + + private owner(executionId: string): HubAgentOwner { + return { kind: "hub", relationshipId: this.relationshipId, executionId }; + } +} + +function toCreateAgentWorktree(target: CreateAgentWorktreeTarget | undefined) { + if (!target) return undefined; + if (target.mode === "branch-off") { + return { + worktreeName: target.newBranch, + baseBranch: target.base, + action: "branch-off" as const, + }; + } + if (target.mode === "checkout-branch") { + return { branchName: target.branch, action: "checkout" as const }; + } + return { githubPrNumber: target.prNumber, action: "checkout" as const }; +} diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts new file mode 100644 index 0000000000..d2bbd24c6f --- /dev/null +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -0,0 +1,117 @@ +import { WebSocket } from "ws"; +import { z } from "zod"; +import type { WebSocketLike } from "../websocket-server.js"; + +export interface HubEnrollment { + relationshipId: string; + idempotencyKey: string; + hubOrigin: string; + token: string; + serverId: string; + daemonPublicKey: string; + credentialVerifier: string; + scopes: string[]; +} + +export interface HubEnrollmentResult { + relationshipId: string; + scopes: string[]; + webSocketUrl: string; +} + +export interface HubRevocation { + relationshipId: string; + hubOrigin: string; + credential: string; +} + +export interface HubSocketCredentials { + relationshipId: string; + webSocketUrl: string; + credential: string; +} + +export interface HubSocketEvents { + connected(socket: WebSocketLike): void; + rejected(statusCode: 401 | 403): void; + closed(code: number): void; + failed(error: Error): void; +} + +export interface HubSocketConnection { + close(): void; +} + +export interface HubRelationshipRemote { + enroll(input: HubEnrollment): Promise; + revoke(input: HubRevocation): Promise; + openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection; +} + +export class HubEnrollmentRejectedError extends Error { + constructor(readonly statusCode: number) { + super(`Hub enrollment failed (${statusCode})`); + this.name = "HubEnrollmentRejectedError"; + } +} + +const EnrollmentResultSchema = z.object({ + relationshipId: z.string(), + scopes: z.array(z.string()), + webSocketUrl: z.string().url(), +}); + +export class DirectHubRelationshipRemote implements HubRelationshipRemote { + async enroll(input: HubEnrollment): Promise { + 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); + } + throw new Error(`Hub enrollment failed (${response.status})`); + } + return EnrollmentResultSchema.parse(await response.json()); + } + + async revoke(input: HubRevocation): Promise { + const response = await fetch( + `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`, + { method: "DELETE", headers: { authorization: `Bearer ${input.credential}` } }, + ); + if (!response.ok && response.status !== 404) { + throw new Error(`Hub revocation failed (${response.status})`); + } + } + + openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection { + const socket = new WebSocket(input.webSocketUrl, { + headers: { + authorization: `Bearer ${input.credential}`, + "x-paseo-relationship-id": input.relationshipId, + }, + }); + socket.once("open", () => events.connected(socket as WebSocketLike)); + socket.once("unexpected-response", (_request, response) => { + if (response.statusCode === 401 || response.statusCode === 403) { + events.rejected(response.statusCode); + } + }); + socket.once("close", (code) => events.closed(code)); + socket.once("error", (error) => events.failed(error)); + return socket; + } +} diff --git a/packages/server/src/server/hub/relationship-retry.test.ts b/packages/server/src/server/hub/relationship-retry.test.ts new file mode 100644 index 0000000000..8331df1d55 --- /dev/null +++ b/packages/server/src/server/hub/relationship-retry.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, test } from "vitest"; +import { BoundedExponentialHubRetryPolicy } from "./relationship-retry.js"; + +describe("Hub relationship retry policy", () => { + test("keeps jitter within the lower and upper bounds", () => { + const lower = new BoundedExponentialHubRetryPolicy(() => 0); + const upper = new BoundedExponentialHubRetryPolicy(() => 1); + + expect(lower.delay(0)).toBe(375); + expect(upper.delay(0)).toBe(625); + }); + + test("grows exponentially and caps the base delay", () => { + const retry = new BoundedExponentialHubRetryPolicy(() => 0.5); + + expect([0, 1, 2, 3, 4, 5, 6, 20].map((attempt) => retry.delay(attempt))).toEqual([ + 500, 1_000, 2_000, 4_000, 8_000, 16_000, 30_000, 30_000, + ]); + }); + + test("never schedules a zero-delay tight loop", () => { + const retry = new BoundedExponentialHubRetryPolicy(() => 0); + + expect([0, 1, 2].map((attempt) => retry.delay(attempt))).toEqual([375, 750, 1_500]); + }); +}); diff --git a/packages/server/src/server/hub/relationship-retry.ts b/packages/server/src/server/hub/relationship-retry.ts new file mode 100644 index 0000000000..005aff89ff --- /dev/null +++ b/packages/server/src/server/hub/relationship-retry.ts @@ -0,0 +1,16 @@ +import type { HubRelationshipRetryPolicy } from "./relationship-controller.js"; + +const INITIAL_DELAY_MS = 500; +const MAX_BASE_DELAY_MS = 30_000; +const MIN_JITTER_FACTOR = 0.75; +const JITTER_RANGE = 0.5; + +export class BoundedExponentialHubRetryPolicy implements HubRelationshipRetryPolicy { + constructor(private readonly random: () => number = Math.random) {} + + delay(attempt: number): number { + const base = Math.min(MAX_BASE_DELAY_MS, INITIAL_DELAY_MS * 2 ** attempt); + const jitter = MIN_JITTER_FACTOR + this.random() * JITTER_RANGE; + return Math.max(1, Math.round(base * jitter)); + } +} diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts new file mode 100644 index 0000000000..482f7d039c --- /dev/null +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -0,0 +1,1128 @@ +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 { networkInterfaces, tmpdir } from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { Writable } from "node:stream"; +import pino from "pino"; +import { WebSocket } from "ws"; +import type { + AgentSnapshotPayload, + HubAgentCreateResponse, + HubAgentStream, + HubAgentUpdate, + HubAuthorizationDenied, + HubExecutionReconcileResponse, + CreateAgentWorktreeTarget, + SessionOutboundMessage, +} from "../../messages.js"; +import { createPaseoDaemon, type PaseoDaemon, type PaseoDaemonConfig } from "../../bootstrap.js"; +import type { WebSocketLike } from "../../websocket-server.js"; +import type { + AgentClient, + AgentCreateSessionOptions, + AgentLaunchContext, + AgentPersistenceHandle, + AgentSession, + AgentSessionConfig, + FetchCatalogOptions, + ProviderCatalog, +} from "../../agent/agent-sdk-types.js"; +import { createTestAgentClients } from "../../test-utils/fake-agent-client.js"; +import { DaemonClient } from "../../test-utils/daemon-client.js"; +import { AgentStorage } from "../../agent/agent-storage.js"; +import { AgentManager } from "../../agent/agent-manager.js"; +import { RelationshipOwnedExecutions } from "../relationship-owned-executions.js"; +import { + createAgentCommand, + type CreateAgentCommandDependencies, +} from "../../agent/create-agent/create.js"; +import type { + HubRelationshipClock, + HubRelationshipRetryPolicy, + ScheduledRelationshipTask, +} from "../relationship-controller.js"; +import type { + HubEnrollment, + HubEnrollmentResult, + HubRelationshipRemote, + HubRevocation, + HubSocketConnection, + HubSocketCredentials, + HubSocketEvents, +} from "../relationship-remote.js"; + +const execFileAsync = promisify(execFile); +const HUB_ORIGIN = "https://hub.test"; +const SOCKET_URL = "wss://hub.test/daemon"; + +export interface ArchiveWatcher { + close(): void; + onError(listener: (error: Error) => void): void; +} + +export interface ArchiveWatchFiles { + watchDirectory(path: string, onChange: () => void): ArchiveWatcher; +} + +const nodeArchiveWatchFiles: ArchiveWatchFiles = { + watchDirectory(directory, onChange) { + const watcher = watch(directory, onChange); + return { + close: () => watcher.close(), + onError: (listener) => watcher.on("error", listener), + }; + }, +}; + +export class SetupFailingArchiveWatchFiles implements ArchiveWatchFiles { + private attempts = 0; + private readonly openDirectories = new Set(); + + constructor(private readonly failOnAttempt: number) {} + + watchDirectory(directory: string): ArchiveWatcher { + this.attempts++; + if (this.attempts === this.failOnAttempt) { + throw new Error(`Cannot watch ${directory}`); + } + this.openDirectories.add(directory); + let open = true; + return { + close: () => { + if (!open) return; + open = false; + this.openDirectories.delete(directory); + }, + onError: () => {}, + }; + } + + activeDirectories(): string[] { + return [...this.openDirectories]; + } +} + +interface PersistedRelationship { + version: number; + state: string; + reason?: string; + relationship: { + id: string; + idempotencyKey?: string; + hubOrigin: string; + scopes: string[]; + }; + credential?: { secret: string }; + enrollment?: { token: string }; + identity?: { serverId: string; daemonPublicKey: string }; +} + +export interface RelationshipInvocationSnapshot { + mode: number; + record: PersistedRelationship; +} + +interface Deferred { + promise: Promise; + resolve(value: T): void; + reject(error: Error): void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((accept, decline) => { + resolve = accept; + reject = decline; + }); + return { promise, resolve, reject }; +} + +class TestRelationshipClock implements HubRelationshipClock, HubRelationshipRetryPolicy { + private current = new Date("2026-07-13T00:00:00.000Z"); + private tasks: Array<{ cancelled: boolean; task: () => void }> = []; + + now(): Date { + return this.current; + } + + delay(attempt: number): number { + return Math.min(8_000, 1_000 * 2 ** attempt); + } + + schedule(_delayMs: number, task: () => void): ScheduledRelationshipTask { + const scheduled = { cancelled: false, task }; + this.tasks.push(scheduled); + return { cancel: () => (scheduled.cancelled = true) }; + } + + async runNext(): Promise { + while (true) { + const scheduled = this.tasks.shift(); + if (!scheduled) throw new Error("No relationship retry is scheduled"); + if (!scheduled.cancelled) { + scheduled.task(); + await Promise.resolve(); + return; + } + } + } +} + +class MemoryHubSocket extends EventEmitter implements WebSocketLike, HubSocketConnection { + readyState = 1; + bufferedAmount = 0; + sent: SessionOutboundMessage[] = []; + closed = false; + closeCode: number | null = null; + private messageObserved = deferred(); + + send(data: string | Uint8Array | ArrayBuffer): void { + if (typeof data !== "string") return; + const frame = JSON.parse(data) as { type: "session"; message: SessionOutboundMessage }; + this.sent.push(frame.message); + this.messageObserved.resolve(); + } + + close(code = 1000): void { + this.closed = true; + this.readyState = 3; + this.closeCode = code; + this.emit("close", code); + } + + receive(message: unknown): void { + this.emit("message", JSON.stringify({ type: "session", message }), false); + } + + receiveEnvelope(message: unknown): void { + this.emit("message", JSON.stringify(message), false); + } + + receiveBinary(): void { + this.emit("message", new Uint8Array([1, 2, 3]), true); + } + + async messageFor(requestId: string): Promise { + while (true) { + const message = this.sent.find( + (candidate) => + "payload" in candidate && + "requestId" in candidate.payload && + candidate.payload.requestId === requestId, + ); + if (message) return message; + await this.messageObserved.promise; + this.messageObserved = deferred(); + } + } + + async messageMatching( + predicate: (message: SessionOutboundMessage) => message is TMessage, + ): Promise { + while (true) { + const message = this.sent.find(predicate); + if (message) return message; + await this.messageObserved.promise; + this.messageObserved = deferred(); + } + } +} + +class ControlledAgentClient implements AgentClient { + readonly provider; + readonly capabilities; + private gate: Deferred | null = null; + private creationObserved = deferred(); + creations = 0; + createdConfigs: AgentSessionConfig[] = []; + + constructor(private readonly client: AgentClient) { + this.provider = client.provider; + this.capabilities = client.capabilities; + } + + holdCreation(): void { + this.gate = deferred(); + } + + async creationAt(count: number): Promise { + while (this.creations < count) { + await this.creationObserved.promise; + this.creationObserved = deferred(); + } + } + + finishCreation(): void { + if (!this.gate) throw new Error("Agent creation is not held"); + this.gate.resolve(); + this.gate = null; + } + + async createSession( + config: AgentSessionConfig, + launchContext?: AgentLaunchContext, + options?: AgentCreateSessionOptions, + ): Promise { + this.creations++; + this.createdConfigs.push({ ...config }); + this.creationObserved.resolve(); + await this.gate?.promise; + return this.client.createSession(config, launchContext, options); + } + + resumeSession( + handle: AgentPersistenceHandle, + overrides?: Partial, + launchContext?: AgentLaunchContext, + ): Promise { + return this.client.resumeSession(handle, overrides, launchContext); + } + + fetchCatalog(options: FetchCatalogOptions): Promise { + return this.client.fetchCatalog(options); + } + + isAvailable(): Promise { + return this.client.isAvailable(); + } +} + +interface SocketAttempt { + input: HubSocketCredentials; + events: HubSocketEvents; + socket: MemoryHubSocket; +} + +class InMemoryHubRelationships implements HubRelationshipRemote { + enrollments: HubEnrollment[] = []; + revocations: HubRevocation[] = []; + sockets: SocketAttempt[] = []; + private enrollmentGate: Deferred | null = null; + private enrollmentObserved = deferred(); + private socketObserved = deferred(); + private revokeFailures = 0; + private readonly relationships = new Set(); + readonly enrollmentSnapshots: RelationshipInvocationSnapshot[] = []; + readonly socketSnapshots: RelationshipInvocationSnapshot[] = []; + + constructor(private readonly captureRelationship: () => RelationshipInvocationSnapshot) {} + + holdEnrollment(): void { + this.enrollmentGate = deferred(); + } + + 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.enrollmentGate) return this.enrollmentGate.promise; + return this.enrollmentResult(input); + } + + completeEnrollment(): void { + const input = this.enrollments.at(-1); + if (!input || !this.enrollmentGate) throw new Error("No enrollment is waiting"); + this.enrollmentGate.resolve(this.enrollmentResult(input)); + this.enrollmentGate = null; + } + + loseEnrollmentResponse(): void { + if (!this.enrollmentGate) throw new Error("No enrollment is waiting"); + this.enrollmentGate.reject(new Error("Enrollment response was lost")); + this.enrollmentGate = null; + } + + async enrollmentAt(index: number): Promise { + while (!this.enrollments[index]) { + await this.enrollmentObserved.promise; + this.enrollmentObserved = deferred(); + } + return this.enrollments[index]; + } + + failRevocations(count: number): void { + this.revokeFailures = count; + } + + async revoke(input: HubRevocation): Promise { + this.revocations.push({ ...input }); + if (this.revokeFailures > 0) { + this.revokeFailures--; + throw new Error("Hub is offline"); + } + } + + openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection { + this.socketSnapshots.push(this.captureRelationship()); + const attempt = { input: { ...input }, events, socket: new MemoryHubSocket() }; + this.sockets.push(attempt); + this.socketObserved.resolve(); + return attempt.socket; + } + + async socketAt(index: number): Promise { + while (!this.sockets[index]) { + await this.socketObserved.promise; + this.socketObserved = deferred(); + } + return this.sockets[index]; + } + + relationshipCount(): number { + return this.relationships.size; + } + + private enrollmentResult(input: HubEnrollment): HubEnrollmentResult { + return { relationshipId: input.relationshipId, scopes: ["hub.*"], webSocketUrl: SOCKET_URL }; + } +} + +interface CliProcess { + result: Promise>; +} + +type AcceptedCreate = Omit & { + agentId: string; + agent: AgentSnapshotPayload; + success: true; +}; + +const providerCatalog = { + async resolveCreateConfig(input) { + return { modeId: input.requestedMode, featureValues: input.featureValues }; + }, +} satisfies CreateAgentCommandDependencies["providerSnapshotManager"]; + +export class HubRelationshipHarness { + private readonly clock = new TestRelationshipClock(); + private readonly remote = new InMemoryHubRelationships(() => this.captureRelationship()); + private daemon: PaseoDaemon | null = null; + private config!: PaseoDaemonConfig; + private root = ""; + private paseoHome = ""; + private host = ""; + private readonly logs: string[] = []; + private observedEnrollments = 0; + private observedSockets = 0; + private readonly codex = new ControlledAgentClient(createTestAgentClients().codex); + + private constructor(private readonly archiveWatchFiles: ArchiveWatchFiles) {} + + static async start( + archiveWatchFiles: ArchiveWatchFiles = nodeArchiveWatchFiles, + ): Promise { + const harness = new HubRelationshipHarness(archiveWatchFiles); + await harness.createHome(); + await harness.startDaemon(); + return harness; + } + + holdEnrollment(): void { + this.remote.holdEnrollment(); + } + + beginConnect(token = "ceremony-token"): CliProcess { + return { result: this.runCli(["hub", "connect", HUB_ORIGIN, "--token", token]) }; + } + + async status(): Promise> { + return this.runCli(["hub", "status"]); + } + + async disconnect(force = false): Promise> { + return this.runCli(["hub", "disconnect", ...(force ? ["--force"] : [])]); + } + + async attemptExternalManagementAsCli(): Promise { + const address = Object.values(networkInterfaces()) + .flat() + .find((candidate) => candidate?.family === "IPv4" && !candidate.internal)?.address; + if (!address) + throw new Error("No non-loopback IPv4 address is available for the external test"); + const target = this.daemon?.getListenTarget(); + if (!target || target.type !== "tcp") throw new Error("Daemon did not bind TCP"); + const socket = await this.openClaimedCliSocket(`ws://${address}:${target.port}/ws`); + const messages = [ + { + type: "hub.relationship.connect.request", + requestId: "external-hub-connect", + hubUrl: HUB_ORIGIN, + token: "external-token", + }, + { type: "hub.relationship.get_status.request", requestId: "external-hub-status" }, + { + type: "hub.relationship.disconnect.request", + requestId: "external-hub-disconnect", + force: false, + }, + ]; + const responses: SessionOutboundMessage[] = []; + for (const message of messages) { + socket.send(JSON.stringify({ type: "session", message })); + responses.push((await this.nextSocketEnvelope(socket)).message); + } + socket.close(); + return responses; + } + + async attemptLoopbackBrowserConnectAsCli(): Promise { + const target = this.daemon?.getListenTarget(); + if (!target || target.type !== "tcp") throw new Error("Daemon did not bind TCP"); + const socket = await this.openClaimedCliSocket(`ws://127.0.0.1:${target.port}/ws`, { + origin: `http://127.0.0.1:${target.port}`, + }); + socket.send( + JSON.stringify({ + type: "session", + message: { + type: "hub.relationship.connect.request", + requestId: "browser-hub-connect", + hubUrl: HUB_ORIGIN, + token: "browser-token", + }, + }), + ); + const response = (await this.nextSocketEnvelope(socket)).message; + socket.close(); + return response; + } + + async enrollmentBegins(): Promise { + return this.remote.enrollmentAt(this.observedEnrollments++); + } + + completeEnrollment(): void { + this.remote.completeEnrollment(); + } + + loseEnrollmentResponse(): void { + this.remote.loseEnrollmentResponse(); + } + + failRevocations(count: number): void { + this.remote.failRevocations(count); + } + + async socketDialed(): Promise { + await this.remote.socketAt(this.observedSockets++); + } + + connectLatestSocket(): void { + const socket = this.latestSocket(); + socket.events.connected(socket.socket); + } + + rejectLatestSocket(statusCode: 401 | 403): void { + this.latestSocket().events.rejected(statusCode); + } + + rejectRelationship(code: 4403 | 401 | 403): void { + if (code === 4403) { + this.closeLatestSocket(code); + return; + } + this.rejectLatestSocket(code); + } + + closeLatestSocket(code: number): void { + const socket = this.latestSocket(); + socket.events.closed(code); + socket.socket.close(code); + } + + connectSocket(index: number): void { + const socket = this.remote.sockets[index]; + if (!socket) throw new Error(`Socket ${index} does not exist`); + socket.events.connected(socket.socket); + } + + closeSocket(index: number, code: number): void { + const socket = this.remote.sockets[index]; + if (!socket) throw new Error(`Socket ${index} does not exist`); + socket.events.closed(code); + socket.socket.close(code); + } + + sendHubRequestOnLatest(message: unknown): SessionOutboundMessage[] { + const socket = this.latestSocket().socket; + socket.receive(message); + return socket.sent.slice(); + } + + holdAgentCreation(): void { + this.codex.holdCreation(); + } + + beginOwnedCreate( + requestId: string, + executionId = "execution-race", + options: { + worktree?: CreateAgentWorktreeTarget; + autoArchive?: boolean; + prompt?: string; + modeId?: string; + } = {}, + ): void { + const { prompt = "Create through the Hub", ...requestOptions } = options; + this.latestSocket().socket.receive({ + type: "hub.agent.create.request", + requestId, + executionId, + provider: "codex", + cwd: this.root, + workspaceId: "hub-workspace", + prompt, + ...requestOptions, + }); + } + + async agentCreationAttempts(count: number): Promise { + await this.codex.creationAt(count); + } + + finishAgentCreation(): void { + this.codex.finishCreation(); + } + + async ownedCreateResult(requestId: string): Promise { + return this.latestSocket().socket.messageFor(requestId); + } + + async durableOwnedAgentIds(): Promise { + return (await this.daemon!.agentStorage.list()) + .filter((record) => record.owner?.kind === "hub") + .map((record) => record.id); + } + + socketDeliveredResponse(socketIndex: number, requestId: string): boolean { + const socket = this.remote.sockets[socketIndex]; + if (!socket) throw new Error(`Socket ${socketIndex} does not exist`); + return socket.socket.sent.some( + (message) => + "payload" in message && + "requestId" in message.payload && + message.payload.requestId === requestId, + ); + } + + providerCreations(): number { + return this.codex.creations; + } + + latestCreatedCwd(): string | null { + return this.codex.createdConfigs.at(-1)?.cwd ?? null; + } + + repoRoot(): string { + return this.root; + } + + async waitForOwnedArchiveCompletion( + agentId: string, + ): Promise<{ agentArchivedAt: string; workspaceArchivedAt: string }> { + const created = await this.daemon!.agentStorage.get(agentId); + if (!created) throw new Error(`Owned agent ${agentId} does not exist`); + if (!created.workspaceId) throw new Error(`Owned agent ${agentId} has no workspace`); + return new Promise<{ agentArchivedAt: string; workspaceArchivedAt: string }>( + (resolve, reject) => { + const watchers: ArchiveWatcher[] = []; + let settled = false; + const timeout = setTimeout( + () => finish(new Error(`Timed out waiting for owned agent ${agentId} to archive`)), + 15_000, + ); + timeout.unref?.(); + const closeWatchers = () => { + clearTimeout(timeout); + for (const watcher of watchers) watcher.close(); + }; + const finish = ( + result: Error | { agentArchivedAt: string; workspaceArchivedAt: string }, + ) => { + if (settled) return; + settled = true; + closeWatchers(); + if (result instanceof Error) reject(result); + else resolve(result); + }; + const observeCompletion = async () => { + try { + const archived = await this.daemon!.agentStorage.get(agentId); + const workspaceArchivedAt = this.workspaceArchivedAt(created.workspaceId!); + if (!archived?.archivedAt || !workspaceArchivedAt || existsSync(created.cwd)) return; + finish({ agentArchivedAt: archived.archivedAt, workspaceArchivedAt }); + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))); + } + }; + try { + const projectsWatcher = this.archiveWatchFiles.watchDirectory( + path.join(this.paseoHome, "projects"), + observeCompletion, + ); + watchers.push(projectsWatcher); + projectsWatcher.onError(finish); + const worktreeWatcher = this.archiveWatchFiles.watchDirectory( + path.dirname(created.cwd), + observeCompletion, + ); + watchers.push(worktreeWatcher); + worktreeWatcher.onError(finish); + void observeCompletion(); + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))); + } + }, + ); + } + + private workspaceArchivedAt(workspaceId: string): string | null { + const records = JSON.parse( + readFileSync(path.join(this.paseoHome, "projects", "workspaces.json"), "utf8"), + ) as Array<{ workspaceId: string; archivedAt?: string | null }>; + return records.find((workspace) => workspace.workspaceId === workspaceId)?.archivedAt ?? null; + } + + async worktreeState(worktreePath: string): Promise<{ exists: boolean; listed: boolean }> { + const { stdout } = await execFileAsync("git", [ + "-C", + this.root, + "worktree", + "list", + "--porcelain", + ]); + const listedPaths = stdout + .split("\n") + .filter((line) => line.startsWith("worktree ")) + .map((line) => line.slice("worktree ".length)); + return { + exists: existsSync(worktreePath), + listed: listedPaths.includes(worktreePath), + }; + } + + async createOwnedConcurrently(executionId = "execution-1"): Promise<{ + first: AcceptedCreate; + duplicate: AcceptedCreate; + }> { + this.beginOwnedCreate("create-1", executionId); + this.beginOwnedCreate("create-2", executionId); + const first = await this.acceptedCreate("create-1"); + const duplicate = await this.acceptedCreate("create-2"); + return { first, duplicate }; + } + + async ownedUpdate(agentId: string): Promise { + const message = await this.latestSocket().socket.messageMatching( + (candidate): candidate is HubAgentUpdate => + candidate.type === "hub.agent.update" && candidate.payload.agentId === agentId, + ); + return message.payload; + } + + async ownedRunningUpdate(agentId: string): Promise { + const message = await this.latestSocket().socket.messageMatching( + (candidate): candidate is HubAgentUpdate => + candidate.type === "hub.agent.update" && + candidate.payload.agentId === agentId && + candidate.payload.agent.status === "running", + ); + return message.payload; + } + + async ownedStream(agentId: string): Promise { + const message = await this.latestSocket().socket.messageMatching( + (candidate): candidate is HubAgentStream => + candidate.type === "hub.agent.stream" && candidate.payload.agentId === agentId, + ); + return message.payload; + } + + async reconcileOwned( + executionId = "execution-1", + ): Promise { + const requestId = `reconcile-${executionId}`; + this.latestSocket().socket.receive({ + type: "hub.execution.reconcile.request", + requestId, + executionId, + }); + return ( + (await this.latestSocket().socket.messageFor(requestId)) as HubExecutionReconcileResponse + ).payload; + } + + async createUnrelatedLocalAgent(): Promise { + const agent = await this.daemon!.agentManager.createAgent( + { provider: "codex", cwd: this.root }, + undefined, + { workspaceId: "local-workspace" }, + ); + return agent.id; + } + + async deniedSteering(agentId: string): Promise { + const requestId = "denied-steer"; + this.latestSocket().socket.receive({ + type: "send_agent_message_request", + requestId, + agentId, + text: "This must not run", + }); + return ((await this.latestSocket().socket.messageFor(requestId)) as HubAuthorizationDenied) + .payload; + } + + async deniedBrowserDispatch(): Promise { + const requestId = "browser-1"; + this.latestSocket().socket.receive({ + type: "browser.automation.execute.response", + payload: { + requestId, + ok: false, + error: { code: "browser_denied", message: "denied", retryable: false }, + }, + }); + return ((await this.latestSocket().socket.messageFor(requestId)) as HubAuthorizationDenied) + .payload; + } + + observedAgentIds(): string[] { + return this.remote.sockets.flatMap(({ socket }) => + socket.sent.flatMap((message) => { + if (message.type === "hub.agent.update") return [message.payload.agentId]; + if (message.type === "hub.agent.stream") return [message.payload.agentId]; + return []; + }), + ); + } + + observedTrustedLifecycleMessages(): string[] { + return this.remote.sockets.flatMap(({ socket }) => + socket.sent + .filter((message) => message.type === "status") + .map((message) => message.payload.status), + ); + } + + probeTrustedHello(): number | null { + const socket = this.latestSocket().socket; + socket.receiveEnvelope({ + type: "hello", + clientId: "hub-must-not-resume", + clientType: "browser", + protocolVersion: 1, + capabilities: { voice: true, pushNotifications: true }, + }); + return socket.closeCode; + } + + probeBinaryFrame(): number | null { + const socket = this.latestSocket().socket; + socket.receiveBinary(); + return socket.closeCode; + } + + async trustedBroadcastCount(): Promise { + const before = this.daemonConfigBroadcasts(); + const client = await this.trustedClient(); + try { + await client.patchDaemonConfig({ appendSystemPrompt: "hub-broadcast-isolation" }); + await this.reconcileOwned("broadcast-barrier"); + return this.daemonConfigBroadcasts() - before; + } finally { + await client.close(); + } + } + + async trustedDaemonStatus() { + const client = await this.trustedClient(); + try { + return await client.getDaemonStatus(); + } finally { + await client.close(); + } + } + + async reconnectAndReconcile(executionId = "execution-1") { + this.closeLatestSocket(1006); + await this.retry(); + this.connectLatestSocket(); + return this.reconcileOwned(executionId); + } + + async reconstructAndReplay(executionId = "execution-1") { + const storage = new AgentStorage( + path.join(this.paseoHome, "agents"), + pino({ level: "silent" }), + ); + const manager = new AgentManager({ + clients: createTestAgentClients(), + registry: storage, + logger: pino({ level: "silent" }), + }); + const executions = this.executionsForReconstruction(manager, storage); + const replay = await executions.create(this.ownedCreateInput(executionId)); + const reconciliation = await executions.reconcile(executionId); + const durableAgentCount = (await storage.list()).filter( + (record) => record.owner?.kind === "hub", + ).length; + return { replay, reconciliation, durableAgentCount }; + } + + async removeAndReconcile(agentId: string, executionId = "execution-1") { + await this.daemon!.agentStorage.remove(agentId); + const storage = new AgentStorage( + path.join(this.paseoHome, "agents"), + pino({ level: "silent" }), + ); + const manager = new AgentManager({ + clients: createTestAgentClients(), + registry: storage, + logger: pino({ level: "silent" }), + }); + const reconciliation = await this.executionsForReconstruction(manager, storage).reconcile( + executionId, + ); + return { + reconciliation, + durableAgentCount: (await storage.list()).filter((record) => record.owner?.kind === "hub") + .length, + }; + } + + socketAttempts(): number { + return this.remote.sockets.length; + } + + revocationAttempts(): number { + return this.remote.revocations.length; + } + + enrollmentAttempts(): HubEnrollment[] { + return this.remote.enrollments.map((input) => ({ ...input, scopes: input.scopes.slice() })); + } + + enrollmentInvocation(index = 0): RelationshipInvocationSnapshot { + const snapshot = this.remote.enrollmentSnapshots[index]; + if (!snapshot) throw new Error(`Enrollment invocation ${index} does not exist`); + return snapshot; + } + + socketInvocation(index = 0): RelationshipInvocationSnapshot { + const snapshot = this.remote.socketSnapshots[index]; + if (!snapshot) throw new Error(`Socket invocation ${index} does not exist`); + return snapshot; + } + + enrolledRelationships(): number { + return this.remote.relationshipCount(); + } + + loggableValues(status: Record): string { + return JSON.stringify({ status, logs: this.logs }); + } + + relationshipFile(): PersistedRelationship | null { + const file = path.join(this.paseoHome, "hub-relationship.json"); + if (!existsSync(file)) return null; + return JSON.parse(readFileSync(file, "utf8")) as PersistedRelationship; + } + + relationshipFileMode(): number { + return statSync(path.join(this.paseoHome, "hub-relationship.json")).mode & 0o777; + } + + private captureRelationship(): RelationshipInvocationSnapshot { + const record = this.relationshipFile(); + if (!record) throw new Error("Relationship authority was not persisted before invocation"); + return { mode: this.relationshipFileMode(), record }; + } + + async retry(): Promise { + await this.clock.runNext(); + } + + async restartDaemon(): Promise { + await this.stopDaemon(); + await this.startDaemon(); + } + + async close(): Promise { + await this.stopDaemon(); + await rm(this.root, { recursive: true, force: true }); + } + + private async createHome(): Promise { + this.root = await mkdtemp(path.join(tmpdir(), "paseo-hub-relationship-")); + this.paseoHome = path.join(this.root, ".paseo"); + const staticDir = path.join(this.root, "static"); + await Promise.all([mkdir(this.paseoHome, { recursive: true }), mkdir(staticDir)]); + execFileSync("git", ["init", "-b", "main", this.root], { stdio: "ignore" }); + execFileSync("git", ["-C", this.root, "config", "user.email", "hub@test.invalid"]); + execFileSync("git", ["-C", this.root, "config", "user.name", "Hub Test"]); + execFileSync("git", ["-C", this.root, "commit", "--allow-empty", "-m", "initial"], { + stdio: "ignore", + }); + this.config = { + listen: "0.0.0.0:0", + paseoHome: this.paseoHome, + corsAllowedOrigins: [], + hostnames: true, + mcpEnabled: false, + staticDir, + mcpDebug: false, + agentClients: { + ...createTestAgentClients(), + codex: this.codex, + }, + agentStoragePath: path.join(this.paseoHome, "agents"), + relayEnabled: false, + relayEndpoint: "relay.paseo.sh:443", + appBaseUrl: "https://app.paseo.sh", + }; + } + + private async startDaemon(): Promise { + const destination = new Writable({ + write: (chunk, _encoding, done) => { + this.logs.push(String(chunk)); + done(); + }, + }); + this.daemon = await createPaseoDaemon(this.config, pino({ level: "trace" }, destination), { + hubRelationshipRemote: this.remote, + hubRelationshipClock: this.clock, + hubRelationshipRetryPolicy: this.clock, + }); + await this.daemon.start(); + const target = this.daemon.getListenTarget(); + if (!target || target.type !== "tcp") throw new Error("Daemon did not bind TCP"); + this.host = `127.0.0.1:${target.port}`; + } + + private async stopDaemon(): Promise { + await this.daemon?.stop(); + this.daemon = null; + } + + private async runCli(args: string[]): Promise> { + const entrypoint = path.join(process.cwd(), "packages/cli/src/index.ts"); + const { stdout } = await execFileAsync( + process.execPath, + ["--import", "tsx", entrypoint, ...args, "--host", this.host, "--json"], + { cwd: process.cwd(), env: { ...process.env, NO_COLOR: "1" } }, + ); + const parsed = JSON.parse(stdout) as unknown; + if (Array.isArray(parsed)) return parsed[0] as Record; + return parsed as Record; + } + + private nextSocketEnvelope(socket: WebSocket): Promise<{ message: SessionOutboundMessage }> { + return new Promise((resolve, reject) => { + socket.once("message", (data) => { + try { + resolve(JSON.parse(data.toString()) as { message: SessionOutboundMessage }); + } catch (error) { + reject(error); + } + }); + socket.once("error", reject); + }); + } + + private async openClaimedCliSocket( + url: string, + options: { origin?: string } = {}, + ): Promise { + const socket = new WebSocket(url, options); + await new Promise((resolve, reject) => { + socket.once("open", resolve); + socket.once("error", reject); + }); + socket.send( + JSON.stringify({ + type: "hello", + clientId: "external-claimed-cli", + clientType: "cli", + protocolVersion: 1, + }), + ); + await this.nextSocketEnvelope(socket); + return socket; + } + + private latestSocket(): SocketAttempt { + const socket = this.remote.sockets.at(-1); + if (!socket) throw new Error("No Hub socket has been opened"); + return socket; + } + + private async acceptedCreate(requestId: string): Promise { + const message = (await this.latestSocket().socket.messageFor( + requestId, + )) as HubAgentCreateResponse; + if (!message.payload.success || !message.payload.agentId || !message.payload.agent) { + throw new Error(message.payload.error ?? "Hub agent creation failed"); + } + return { + ...message.payload, + success: true, + agentId: message.payload.agentId, + agent: message.payload.agent, + }; + } + + private ownedCreateInput(executionId: string) { + return { + executionId, + provider: "codex", + cwd: this.root, + workspaceId: "hub-workspace", + prompt: "Create through the Hub", + }; + } + + private executionsForReconstruction(manager: AgentManager, storage: AgentStorage) { + return new RelationshipOwnedExecutions({ + relationshipId: this.relationshipFile()!.relationship.id, + agentManager: manager, + agentStorage: storage, + createAgent: (input) => + createAgentCommand( + { + agentManager: manager, + agentStorage: storage, + logger: pino({ level: "silent" }), + providerSnapshotManager: providerCatalog, + }, + input, + ), + }); + } + + private async trustedClient(): Promise { + const client = new DaemonClient({ + url: `ws://${this.host}/ws`, + appVersion: "0.1.106", + }); + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "hub-relationship-trusted" } }); + return client; + } + + private daemonConfigBroadcasts(): number { + return this.remote.sockets + .flatMap(({ socket }) => socket.sent) + .filter( + (message) => + message.type === "status" && message.payload.status === "daemon_config_changed", + ).length; + } +} diff --git a/packages/server/src/server/persistence-hooks.ts b/packages/server/src/server/persistence-hooks.ts index 5fc5113729..ba220eefa5 100644 --- a/packages/server/src/server/persistence-hooks.ts +++ b/packages/server/src/server/persistence-hooks.ts @@ -110,6 +110,7 @@ export function extractTimestamps(record: StoredAgentRecord): { lastUserMessageAt: Date | null; labels?: Record; workspaceId?: string; + owner?: StoredAgentRecord["owner"]; } { return { createdAt: new Date(record.createdAt), @@ -117,6 +118,7 @@ export function extractTimestamps(record: StoredAgentRecord): { lastUserMessageAt: record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null, labels: record.labels, workspaceId: record.workspaceId, + owner: record.owner, }; } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 1783dcb706..190dcf43cb 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -394,6 +394,15 @@ export interface SessionFileSystem { isDirectory(path: string): Promise; } +export type SessionPeer = + | { + transport: "direct"; + peer: "loopback" | "local_ipc" | "external"; + browserOrigin: boolean; + } + | { transport: "relay"; peer: "external" } + | { transport: "internal"; peer: "internal" }; + const nodeSessionFileSystem: SessionFileSystem = { async isDirectory(path) { const stats = await stat(path).catch(() => null); @@ -441,6 +450,7 @@ export interface SessionOptions { terminalManager: TerminalManager | null; providerSnapshotManager: ProviderSnapshotManager; providerUsageService: ProviderUsageService; + hubRelationships?: import("./hub/relationship-controller.js").HubRelationshipManagement; serviceProxy?: ServiceProxySubsystem; scriptRuntimeStore?: WorkspaceScriptRuntimeStore; workspaceSetupSnapshots?: Map; @@ -504,6 +514,34 @@ function parseClientCapabilities( return new Set(result); } +type HubRelationshipManagementRequest = Extract< + SessionInboundMessage, + { + type: + | "hub.relationship.connect.request" + | "hub.relationship.get_status.request" + | "hub.relationship.disconnect.request"; + } +>; + +function isHubRelationshipManagementRequest( + message: SessionInboundMessage, +): message is HubRelationshipManagementRequest { + return ( + message.type === "hub.relationship.connect.request" || + message.type === "hub.relationship.get_status.request" || + message.type === "hub.relationship.disconnect.request" + ); +} + +function canManageHubRelationships(peer: SessionPeer): boolean { + return ( + peer.transport === "direct" && + !peer.browserOrigin && + (peer.peer === "loopback" || peer.peer === "local_ipc") + ); +} + interface AgentTimelineProjectionSelection { timeline: AgentTimelineFetchResult; entries: TimelineProjectionEntry[]; @@ -798,6 +836,7 @@ export class Session { listProjects: () => this.projectRegistry.list(), listWorkspaces: () => this.workspaceRegistry.list(), logger: this.sessionLogger, + hubRelationships: options.hubRelationships, }); this.daemonConfigStore = daemonConfigStore; this.terminalManager = terminalManager; @@ -1346,7 +1385,10 @@ export class Session { /** * Main entry point for processing session messages */ - public async handleMessage(msg: SessionInboundMessage): Promise { + public async handleMessage( + msg: SessionInboundMessage, + peer: SessionPeer = { transport: "internal", peer: "internal" }, + ): Promise { this.inflightRequests++; if (this.inflightRequests > this.peakInflightRequests) { this.peakInflightRequests = this.inflightRequests; @@ -1360,7 +1402,7 @@ export class Session { "agent.session.inbound", ); try { - await this.dispatchInboundMessage(msg); + await this.dispatchInboundMessage(msg, peer); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.sessionLogger.error({ err }, "Error handling message"); @@ -1398,7 +1440,22 @@ export class Session { } } - private async dispatchInboundMessage(msg: SessionInboundMessage): Promise { + private async dispatchInboundMessage( + msg: SessionInboundMessage, + peer: SessionPeer, + ): Promise { + if (isHubRelationshipManagementRequest(msg) && !canManageHubRelationships(peer)) { + this.emit({ + type: "rpc_error", + payload: { + requestId: msg.requestId, + requestType: msg.type, + error: "Hub relationship management requires a local daemon connection", + code: "local_management_required", + }, + }); + return; + } const promise = this.dispatchVoiceAndControlMessage(msg) ?? this.dispatchAgentRewindMessage(msg) ?? @@ -1561,6 +1618,10 @@ export class Session { return this.daemonSession.handleGetStatusRequest(msg); case "daemon.get_pairing_offer.request": return this.daemonSession.handleGetPairingOfferRequest(msg); + case "hub.relationship.connect.request": + case "hub.relationship.get_status.request": + case "hub.relationship.disconnect.request": + return this.daemonSession.handleHubRelationshipRequest(msg); case "diagnostics.request": return this.daemonSession.handleDiagnosticsRequest(msg); case "daemon.update.request": diff --git a/packages/server/src/server/session/daemon/daemon-session.ts b/packages/server/src/server/session/daemon/daemon-session.ts index bc9e89dec6..290e74e291 100644 --- a/packages/server/src/server/session/daemon/daemon-session.ts +++ b/packages/server/src/server/session/daemon/daemon-session.ts @@ -10,6 +10,7 @@ import { import { DaemonSelfUpdateSessionController } from "./daemon-self-update-session-controller.js"; import type { ManagedAgent } from "../../agent/agent-manager.js"; import type { PersistedProjectRecord, PersistedWorkspaceRecord } from "../../workspace-registry.js"; +import type { HubRelationshipManagement } from "../../hub/relationship-controller.js"; export interface DaemonRuntimeConfig { listen: string | null; @@ -46,6 +47,7 @@ export interface DaemonSessionOptions { listProviderAvailability: () => Promise; getWebSocketRuntimeMetrics?: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null; logger: pino.Logger; + hubRelationships?: HubRelationshipManagement; } /** @@ -69,6 +71,7 @@ export class DaemonSession { private readonly getWebSocketRuntimeMetrics: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null; private readonly logger: pino.Logger; private readonly selfUpdate: DaemonSelfUpdateSessionController; + private readonly hubRelationships: HubRelationshipManagement | null; constructor(options: DaemonSessionOptions) { this.host = options.host; @@ -83,6 +86,7 @@ export class DaemonSession { this.listProviderAvailability = options.listProviderAvailability; this.getWebSocketRuntimeMetrics = options.getWebSocketRuntimeMetrics ?? (() => null); this.logger = options.logger; + this.hubRelationships = options.hubRelationships ?? null; this.selfUpdate = new DaemonSelfUpdateSessionController({ clientId: this.clientId, daemonVersion: this.daemonVersion ?? null, @@ -92,6 +96,40 @@ export class DaemonSession { }); } + async handleHubRelationshipRequest( + msg: Extract< + SessionInboundMessage, + { + type: + | "hub.relationship.connect.request" + | "hub.relationship.get_status.request" + | "hub.relationship.disconnect.request"; + } + >, + ): Promise { + if (!this.hubRelationships) throw new Error("Hub relationship management is unavailable"); + if (msg.type === "hub.relationship.connect.request") { + const status = await this.hubRelationships.connect({ hubUrl: msg.hubUrl, token: msg.token }); + this.host.emit({ + type: "hub.relationship.connect.response", + payload: { requestId: msg.requestId, status }, + }); + return; + } + if (msg.type === "hub.relationship.disconnect.request") { + const result = await this.hubRelationships.disconnect({ force: msg.force ?? false }); + this.host.emit({ + type: "hub.relationship.disconnect.response", + payload: { requestId: msg.requestId, ...result }, + }); + return; + } + this.host.emit({ + type: "hub.relationship.get_status.response", + payload: { requestId: msg.requestId, status: this.hubRelationships.status() }, + }); + } + async handleGetStatusRequest( msg: Extract, ): Promise { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 4061fa464e..0cc52331d0 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -30,7 +30,14 @@ import { asUint8Array, decodeBinaryFrame } from "@getpaseo/protocol/binary-frame import type { TerminalActivity } from "@getpaseo/protocol/terminal-activity"; import type { HostnamesConfig } from "./hostnames.js"; import { isHostnameAllowed } from "./hostnames.js"; -import { Session, type SessionLifecycleIntent, type SessionRuntimeMetrics } from "./session.js"; +import { + Session, + type SessionLifecycleIntent, + type SessionPeer, + type SessionRuntimeMetrics, +} from "./session.js"; +import { HubSession, type HubExecutions } from "./hub/index.js"; +import type { HubRelationshipManagement } from "./hub/relationship-controller.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"; @@ -95,6 +102,8 @@ interface PendingConnection { interface WebSocketConnectionIdentity { connectionId: string; transport: "direct" | "relay"; + peer: "loopback" | "local_ipc" | "external"; + browserOrigin: boolean; host?: string; origin?: string; userAgent?: string; @@ -322,7 +331,7 @@ function getBrowserHostCapability( return parsed.success ? parsed.data : null; } -interface WebSocketLike { +export interface WebSocketLike { readyState: number; bufferedAmount?: number; send: (data: string | Uint8Array | ArrayBuffer) => void; @@ -331,7 +340,8 @@ interface WebSocketLike { once: (event: "close" | "error", listener: (...args: unknown[]) => void) => void; } -interface SessionConnection { +interface TrustedSessionConnection { + kind: "trusted"; session: Session; clientId: string; appVersion: string | null; @@ -341,6 +351,27 @@ interface SessionConnection { externalDisconnectCleanupTimeout: ReturnType | null; } +interface HubSessionConnection { + kind: "hub"; + session: HubSession; + relationshipId: string; + connectionLogger: pino.Logger; + socket: WebSocketLike; +} + +type SessionConnection = TrustedSessionConnection | HubSessionConnection; + +type TrustedLifecycleKey = + | "clientId" + | "appVersion" + | "clientCapabilities" + | "sockets" + | "externalDisconnectCleanupTimeout"; +type HubTrustedLifecycleOverlap = Extract; +const HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE: HubTrustedLifecycleOverlap extends never ? true : never = + true; +void HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE; + interface BrowserToolsRegistration { capabilitySignature: string; unregister: () => void; @@ -401,7 +432,7 @@ export class VoiceAssistantWebSocketServer { private readonly pendingConnections: Map = new Map(); private readonly sessions: Map = new Map(); private readonly socketIdentities: Map = new Map(); - private readonly externalSessionsByKey: Map = new Map(); + private readonly externalSessionsByKey: Map = new Map(); private readonly serverId: string; private readonly daemonVersion: string; private readonly daemonRuntimeConfig: @@ -464,6 +495,7 @@ export class VoiceAssistantWebSocketServer { private readonly providerUsageService: ProviderUsageService; private unsubscribeTerminalActivity: (() => void) | null = null; private readonly browserToolsBroker: BrowserToolsBroker | null; + private readonly hubRelationships: HubRelationshipManagement | null; private readonly browserToolsRegistrations = new Map(); private acceptingConnections = true; @@ -521,6 +553,7 @@ export class VoiceAssistantWebSocketServer { }, serviceProxyPublicBaseUrl?: string | null, browserToolsBroker?: BrowserToolsBroker | null, + hubRelationships?: HubRelationshipManagement | null, ) { this.logger = logger.child({ module: "websocket-server" }); this.serverId = serverId; @@ -530,6 +563,7 @@ export class VoiceAssistantWebSocketServer { this.daemonVersion = daemonVersion.trim(); this.daemonRuntimeConfig = daemonRuntimeConfig; this.browserToolsBroker = browserToolsBroker ?? null; + this.hubRelationships = hubRelationships ?? null; this.agentManager = agentManager; this.agentStorage = agentStorage; this.projectRegistry = projectRegistry ?? createNoopProjectRegistry(); @@ -765,7 +799,10 @@ export class VoiceAssistantWebSocketServer { public broadcast(message: WSOutboundMessage): void { const payload = JSON.stringify(message); - for (const ws of this.sessions.keys()) { + for (const [ws, connection] of this.sessions) { + if (connection.kind !== "trusted") { + continue; + } // WebSocket.OPEN = 1 if (ws.readyState === 1) { ws.send(payload); @@ -774,12 +811,14 @@ export class VoiceAssistantWebSocketServer { } } - public listActiveSessions(): Session[] { + public listTrustedSessions(): Session[] { return Array.from( new Set( - [...this.sessions.values(), ...this.externalSessionsByKey.values()].map( - (connection) => connection.session, - ), + [...this.sessions.values(), ...this.externalSessionsByKey.values()] + .filter( + (connection): connection is TrustedSessionConnection => connection.kind === "trusted", + ) + .map((connection) => connection.session), ), ); } @@ -807,6 +846,38 @@ export class VoiceAssistantWebSocketServer { await this.attachSocket(ws, undefined, metadata); } + public async attachHubSocket( + ws: WebSocketLike, + options: { + relationshipId: string; + executions: HubExecutions; + }, + ): Promise { + if (!this.acceptingConnections) { + ws.close(WS_CLOSE_SERVER_SHUTDOWN, "Server shutting down"); + return; + } + + const connectionLogger = this.logger.child({ + connectionKind: "hub", + relationshipId: options.relationshipId, + }); + const session = new HubSession({ + executions: options.executions, + send: (message) => this.sendToClient(ws, wrapSessionMessage(message)), + }); + const connection: HubSessionConnection = { + kind: "hub", + session, + relationshipId: options.relationshipId, + connectionLogger, + socket: ws, + }; + this.sessions.set(ws, connection); + this.bindSocketHandlers(ws); + connectionLogger.info("Hub session attached"); + } + public prepareForShutdown(): void { this.acceptingConnections = false; } @@ -842,13 +913,14 @@ export class VoiceAssistantWebSocketServer { const cleanupPromises: Promise[] = []; for (const connection of uniqueConnections) { - if (connection.externalDisconnectCleanupTimeout) { + if (connection.kind === "trusted" && connection.externalDisconnectCleanupTimeout) { clearTimeout(connection.externalDisconnectCleanupTimeout); connection.externalDisconnectCleanupTimeout = null; } - cleanupPromises.push(connection.session.cleanup()); - for (const ws of connection.sockets) { + cleanupPromises.push(Promise.resolve(connection.session.cleanup())); + const sockets = connection.kind === "trusted" ? connection.sockets : [connection.socket]; + for (const ws of sockets) { cleanupPromises.push( new Promise((resolve) => { // WebSocket.CLOSED = 3 @@ -918,12 +990,16 @@ export class VoiceAssistantWebSocketServer { } private sendToConnection(connection: SessionConnection, message: WSOutboundMessage): void { - for (const ws of connection.sockets) { + const sockets = connection.kind === "trusted" ? connection.sockets : [connection.socket]; + for (const ws of sockets) { this.sendToClient(ws, message); } } private sendBinaryToConnection(connection: SessionConnection, frame: Uint8Array): void { + if (connection.kind !== "trusted") { + return; + } for (const ws of connection.sockets) { this.sendBinaryToClient(ws, frame); } @@ -991,9 +1067,9 @@ export class VoiceAssistantWebSocketServer { appVersion: string | null; clientCapabilities: Record | null; connectionLogger: pino.Logger; - }): SessionConnection { + }): TrustedSessionConnection { const { ws, clientId, appVersion, clientCapabilities, connectionLogger } = params; - let connection: SessionConnection | null = null; + let connection: TrustedSessionConnection | null = null; const session = new Session({ clientId, @@ -1092,9 +1168,11 @@ export class VoiceAssistantWebSocketServer { daemonVersion: this.daemonVersion, daemonRuntimeConfig: this.daemonRuntimeConfig, getWebSocketRuntimeMetrics: () => this.lastRuntimeMetricsSnapshot, + hubRelationships: this.hubRelationships ?? undefined, }); connection = { + kind: "trusted", session, clientId, appVersion, @@ -1372,6 +1450,15 @@ export class VoiceAssistantWebSocketServer { } this.sessions.delete(ws); + if (connection.kind === "hub") { + this.socketIdentities.delete(ws); + connection.connectionLogger.info( + { code: details.code, reason: stringifyCloseReason(details.reason) }, + "Hub session disconnected", + ); + connection.session.cleanup(); + return; + } connection.sockets.delete(ws); this.socketIdentities.delete(ws); @@ -1420,7 +1507,7 @@ export class VoiceAssistantWebSocketServer { } private async cleanupConnection( - connection: SessionConnection, + connection: TrustedSessionConnection, logMessage: string, ): Promise { this.incrementRuntimeCounter("sessionCleanup"); @@ -1447,7 +1534,7 @@ export class VoiceAssistantWebSocketServer { await connection.session.cleanup(); } - private syncBrowserToolsClientRegistration(connection: SessionConnection): void { + private syncBrowserToolsClientRegistration(connection: TrustedSessionConnection): void { if (!this.browserToolsBroker) { return; } @@ -1523,7 +1610,7 @@ export class VoiceAssistantWebSocketServer { log.warn( { - clientId: activeConnection?.clientId, + clientId: activeConnection?.kind === "trusted" ? activeConnection.clientId : undefined, requestId: requestInfo?.requestId, requestType: requestInfo?.requestType, error: parsedMessage.error.message, @@ -1588,6 +1675,11 @@ export class VoiceAssistantWebSocketServer { } return true; } + if (activeConnection.kind === "hub") { + log.warn("Rejected binary frame on Hub session"); + ws.close(WS_CLOSE_INVALID_HELLO, "Binary frames are not supported on Hub sessions"); + return true; + } void Promise.resolve(activeConnection.session.handleBinaryFrame(decodedFrame)).catch( (error: unknown) => { this.handleRawMessageError({ @@ -1727,25 +1819,40 @@ export class VoiceAssistantWebSocketServer { const controlRpc = getControlRpcLogInfo(message.message); if (controlRpc) { const identity = this.socketIdentities.get(ws); + let connectionFields: Record; + if (identity) { + connectionFields = toConnectionLogFields(identity); + } else if (activeConnection.kind === "trusted") { + connectionFields = { clientId: activeConnection.clientId }; + } else { + connectionFields = { relationshipId: activeConnection.relationshipId }; + } activeConnection.connectionLogger.warn( { - ...(identity ? toConnectionLogFields(identity) : { clientId: activeConnection.clientId }), + ...connectionFields, ...controlRpc, }, "ws_control_rpc_received", ); } - if (message.message.type === "browser.automation.execute.response") { + if ( + activeConnection.kind === "trusted" && + message.message.type === "browser.automation.execute.response" + ) { this.browserToolsBroker?.receiveResponse(message.message as BrowserAutomationExecuteResponse); return; } const startMs = performance.now(); - await activeConnection.session.handleMessage(message.message); + const identity = this.socketIdentities.get(ws); + const peer = identity + ? toSessionPeer(identity) + : ({ transport: "internal", peer: "internal" } satisfies SessionPeer); + await activeConnection.session.handleMessage(message.message, peer); const durationMs = performance.now() - startMs; this.recordRequestLatency(message.message.type, durationMs); - if (durationMs >= SLOW_REQUEST_THRESHOLD_MS) { + if (durationMs >= SLOW_REQUEST_THRESHOLD_MS && activeConnection.kind === "trusted") { activeConnection.connectionLogger.warn( { requestType: message.message.type, @@ -1857,7 +1964,9 @@ export class VoiceAssistantWebSocketServer { } private collectSessionRuntimeMetrics(): WebSocketRuntimeMetrics { - const uniqueConnections = new Set(this.externalSessionsByKey.values()); + const uniqueConnections = new Set( + this.externalSessionsByKey.values(), + ); let terminalDirectorySubscriptionCount = 0; let terminalSubscriptionCount = 0; let inflightRequests = 0; @@ -1958,6 +2067,9 @@ export class VoiceAssistantWebSocketServer { }> = []; for (const [ws, connection] of this.sessions) { + if (connection.kind !== "trusted") { + continue; + } clientEntries.push({ ws, state: this.getClientActivityState(connection.session), @@ -2025,6 +2137,9 @@ export class VoiceAssistantWebSocketServer { }> = []; for (const [ws, connection] of this.sessions) { + if (connection.kind !== "trusted") { + continue; + } clientEntries.push({ ws, state: this.getClientActivityState(connection.session), @@ -2099,6 +2214,8 @@ function createWebSocketConnectionIdentity( return { connectionId: `conn_${randomUUID().replaceAll("-", "")}`, transport: metadata?.transport === "relay" ? "relay" : "direct", + peer: resolveConnectionPeer(requestMetadata, metadata), + browserOrigin: requestMetadata.origin !== undefined, ...(requestMetadata.host ? { host: requestMetadata.host } : {}), ...(requestMetadata.origin ? { origin: requestMetadata.origin } : {}), ...(requestMetadata.userAgent ? { userAgent: requestMetadata.userAgent } : {}), @@ -2111,6 +2228,7 @@ function toConnectionLogFields(identity: WebSocketConnectionIdentity): Record