diff --git a/docs/hub.md b/docs/hub.md index 029083ff26..4a6d736ebd 100644 --- a/docs/hub.md +++ b/docs/hub.md @@ -15,29 +15,34 @@ The daemon persists a relationship ID and private connection credential before e 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. -Normal authenticated daemon sessions may run the `hub.relationship.connect`, -`hub.relationship.get_status`, and `hub.relationship.disconnect` RPCs regardless of transport. -The restricted Hub session cannot manage the relationship. +Normal authenticated daemon sessions may run the `hub.management.daemon.connect`, +`hub.management.daemon.get_status`, and `hub.management.daemon.disconnect` RPCs. Hub connections +receive only `hub.execution.*` authority, so execution credentials cannot manage the relationship. -## Hub session scope +## Session grants and execution ownership -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. +Trusted clients and the Hub use the same `Session` implementation. The connection boundary supplies +grants: trusted clients receive `*`, while an enrolled Hub connection receives its persisted +`hub.execution.*` grant. One matcher handles exact RPC names and trailing namespace wildcards for +both inbound requests and outbound messages. A denied request returns the ordinary `rpc_error` +shape. + +The Hub connection still has a narrow lifecycle boundary: it has no trusted-client hello/resume, +browser, binary, retained-session, or broadcast state. Its outbound execution events include only +agents owned by that daemon identity, so unrelated local agents remain 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. +owner before acknowledging creation. Duplicate or replayed creates for the same daemon and +execution resolve to the same durable agent. After a lost response, reconnect, or daemon restart, +the Hub retries `hub.execution.agent.create.request` with the same execution ID. The idempotent +response returns the existing agent and its current state; there is no separate reconciliation RPC. +Transient stream frames are not durably replayed. Daemon restart preserves the Hub relationship and owned execution identity, but interrupts any -active turn. The daemon persists that agent as `closed`; reconciliation returns the same relationship, -execution, and agent with that terminal state. Paseo never stores or automatically replays the -original prompt. A duplicate create remains idempotent and returns the existing agent without -starting another turn. +active turn. The daemon persists that agent as `closed`; an idempotent create retry returns the same +daemon, execution, and agent identity with that terminal state. Paseo never stores or automatically +replays the original prompt. A duplicate create returns the existing agent without starting another +turn. 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 @@ -59,9 +64,9 @@ opening a Hub socket. This also covers an enrollment whose request may have succ response was lost. `--force` removes local authority immediately and warns that remote revocation may still be pending. -## Cross-repository dependency +## Cross-repository compatibility -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. +The consumer implementation lives in Paseo Cloud. Cloud owns its copy of the Hub wire schemas and +has no Paseo runtime or build dependency. Cross-repository end-to-end verification separately builds +a Paseo source checkout and exercises the real daemon, CLI, direct WebSocket, Cloud service, and +Postgres. That compatibility fixture is not a package dependency or fallback implementation. diff --git a/packages/cli/src/commands/hub/index.ts b/packages/cli/src/commands/hub/index.ts index 81872bf5f8..fce9f1c65b 100644 --- a/packages/cli/src/commands/hub/index.ts +++ b/packages/cli/src/commands/hub/index.ts @@ -5,7 +5,7 @@ import { connectToDaemon } from "../../utils/client.js"; interface HubRow { state: string; - relationshipId: string | null; + daemonId: string | null; hub: string | null; scopes: string; connectedAt: string | null; @@ -18,7 +18,7 @@ const schema: OutputSchema = { columns: [ { header: "STATE", field: "state" }, { header: "HUB", field: "hub" }, - { header: "RELATIONSHIP", field: "relationshipId" }, + { header: "DAEMON", field: "daemonId" }, { header: "SCOPES", field: "scopes" }, { header: "CONNECTED", field: "connectedAt" }, { header: "ERROR", field: "error" }, @@ -29,7 +29,7 @@ const schema: OutputSchema = { function result( status: { state: string; - relationshipId: string | null; + daemonId: string | null; hubOrigin: string | null; scopes: string[]; connectedAt: string | null; @@ -42,7 +42,7 @@ function result( data: [ { state: status.state, - relationshipId: status.relationshipId, + daemonId: status.daemonId, hub: status.hubOrigin, scopes: status.scopes.join(", "), connectedAt: status.connectedAt, diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 08c85fd51c..da84eba4bd 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -3955,8 +3955,8 @@ export class DaemonClient { this.requireHubRelationshipSupport(); return this.sendCorrelatedSessionRequest({ requestId, - message: { type: "hub.relationship.connect.request", hubUrl, token }, - responseType: "hub.relationship.connect.response", + message: { type: "hub.management.daemon.connect.request", hubUrl, token }, + responseType: "hub.management.daemon.connect.response", }); } @@ -3964,8 +3964,8 @@ export class DaemonClient { this.requireHubRelationshipSupport(); return this.sendCorrelatedSessionRequest({ requestId, - message: { type: "hub.relationship.get_status.request" }, - responseType: "hub.relationship.get_status.response", + message: { type: "hub.management.daemon.get_status.request" }, + responseType: "hub.management.daemon.get_status.response", }); } @@ -3973,8 +3973,8 @@ export class DaemonClient { this.requireHubRelationshipSupport(); return this.sendCorrelatedSessionRequest({ requestId, - message: { type: "hub.relationship.disconnect.request", force }, - responseType: "hub.relationship.disconnect.response", + message: { type: "hub.management.daemon.disconnect.request", force }, + responseType: "hub.management.daemon.disconnect.response", }); } diff --git a/packages/protocol/src/messages.hub.test.ts b/packages/protocol/src/messages.hub.test.ts index 3f17123156..ae63b0dc60 100644 --- a/packages/protocol/src/messages.hub.test.ts +++ b/packages/protocol/src/messages.hub.test.ts @@ -5,7 +5,7 @@ import { HubMessageCorrelationError, SessionInboundMessageSchema, SessionOutboundMessageSchema, - parseHubSessionOutboundMessage, + parseHubExecutionOutboundMessage, } from "./messages.js"; const agent = { @@ -38,7 +38,7 @@ const agent = { // Frozen at the Hub create request shape shipped before worktree and autoArchive. const PreviousHubAgentCreateRequestSchema = z.object({ - type: z.literal("hub.agent.create.request"), + type: z.literal("hub.execution.agent.create.request"), requestId: z.string(), executionId: z.string(), provider: z.string(), @@ -53,22 +53,17 @@ const PreviousHubAgentCreateRequestSchema = z.object({ }); describe("Hub session protocol", () => { - test.each([ - { - type: "hub.agent.create.request", + test("accepts the Hub execution create request", () => { + const message = { + type: "hub.execution.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); }); @@ -79,7 +74,7 @@ describe("Hub session protocol", () => { { mode: "checkout-pr", prNumber: 42 }, ])("accepts Hub create worktree target %#", (worktree) => { const message = { - type: "hub.agent.create.request", + type: "hub.execution.agent.create.request", requestId: "hub-worktree", executionId: "execution-worktree", provider: "codex", @@ -93,7 +88,7 @@ describe("Hub session protocol", () => { test("the previous Hub create parser ignores additive worktree and auto-archive fields", () => { const newRequest = { - type: "hub.agent.create.request" as const, + type: "hub.execution.agent.create.request" as const, requestId: "hub-worktree", executionId: "execution-worktree", provider: "codex", @@ -104,7 +99,7 @@ describe("Hub session protocol", () => { }; expect(PreviousHubAgentCreateRequestSchema.parse(newRequest)).toEqual({ - type: "hub.agent.create.request", + type: "hub.execution.agent.create.request", requestId: "hub-worktree", executionId: "execution-worktree", provider: "codex", @@ -115,7 +110,7 @@ describe("Hub session protocol", () => { test.each([ { - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { requestId: "request-1", executionId: "execution-1", @@ -126,84 +121,67 @@ describe("Hub session protocol", () => { }, }, { - type: "hub.agent.update", + type: "hub.execution.agent.update", payload: { executionId: "execution-1", agentId: "agent-1", agent }, }, { - type: "hub.agent.stream", + type: "hub.execution.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); + expect(parseHubExecutionOutboundMessage(message)).toEqual(message); }); test("rejects a Hub update whose correlated agent ids disagree", () => { const malformed = { - type: "hub.agent.update", + type: "hub.execution.agent.update", payload: { executionId: "execution-1", agentId: "agent-2", agent }, }; expect(SessionOutboundMessageSchema.safeParse(malformed).success).toBe(true); - expect(() => parseHubSessionOutboundMessage(malformed)).toThrow(HubMessageCorrelationError); + expect(() => parseHubExecutionOutboundMessage(malformed)).toThrow(HubMessageCorrelationError); }); test.each([ { - type: "hub.relationship.connect.request", + type: "hub.management.daemon.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 }, + { type: "hub.management.daemon.get_status.request", requestId: "r2" }, + { type: "hub.management.daemon.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", + type: "hub.management.daemon.connect.response", payload: { requestId: "r1", status: { state: "connected", - relationshipId: "rel", + daemonId: "daemon-1", hubOrigin: "https://hub.example", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], connectedAt: "2026-07-13T00:00:00.000Z", lastError: null, }, }, }, { - type: "hub.relationship.get_status.response", + type: "hub.management.daemon.get_status.response", payload: { requestId: "r2", status: { state: "not_connected", - relationshipId: null, + daemonId: null, hubOrigin: null, scopes: [], connectedAt: null, @@ -212,14 +190,14 @@ describe("Hub session protocol", () => { }, }, { - type: "hub.relationship.disconnect.response", + type: "hub.management.daemon.disconnect.response", payload: { requestId: "r3", status: { state: "disconnecting", - relationshipId: "rel", + daemonId: "daemon-1", hubOrigin: "https://hub.example", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], connectedAt: null, lastError: "offline", }, diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 7ab505c38b..d84faddd14 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1088,18 +1088,18 @@ export const DaemonGetPairingOfferRequestSchema = z.object({ requestId: z.string(), }); -export const HubRelationshipConnectRequestSchema = z.object({ - type: z.literal("hub.relationship.connect.request"), +export const HubManagementDaemonConnectRequestSchema = z.object({ + type: z.literal("hub.management.daemon.connect.request"), requestId: z.string(), hubUrl: z.string(), token: z.string(), }); -export const HubRelationshipGetStatusRequestSchema = z.object({ - type: z.literal("hub.relationship.get_status.request"), +export const HubManagementDaemonGetStatusRequestSchema = z.object({ + type: z.literal("hub.management.daemon.get_status.request"), requestId: z.string(), }); -export const HubRelationshipDisconnectRequestSchema = z.object({ - type: z.literal("hub.relationship.disconnect.request"), +export const HubManagementDaemonDisconnectRequestSchema = z.object({ + type: z.literal("hub.management.daemon.disconnect.request"), requestId: z.string(), force: z.boolean().optional(), }); @@ -2115,8 +2115,8 @@ export const CaptureTerminalRequestSchema = z.object({ requestId: z.string(), }); -export const HubAgentCreateRequestSchema = z.object({ - type: z.literal("hub.agent.create.request"), +export const HubExecutionAgentCreateRequestSchema = z.object({ + type: z.literal("hub.execution.agent.create.request"), requestId: z.string(), executionId: z.string(), provider: z.string(), @@ -2132,18 +2132,10 @@ export const HubAgentCreateRequestSchema = z.object({ 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 type HubExecutionAgentCreateRequest = z.infer; export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ - HubAgentCreateRequestSchema, - HubExecutionReconcileRequestSchema, + HubExecutionAgentCreateRequestSchema, BrowserAutomationExecuteResponseSchema, VoiceAudioChunkMessageSchema, AbortRequestMessageSchema, @@ -2166,9 +2158,9 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ WaitForFinishRequestSchema, DaemonGetStatusRequestSchema, DaemonGetPairingOfferRequestSchema, - HubRelationshipConnectRequestSchema, - HubRelationshipGetStatusRequestSchema, - HubRelationshipDisconnectRequestSchema, + HubManagementDaemonConnectRequestSchema, + HubManagementDaemonGetStatusRequestSchema, + HubManagementDaemonDisconnectRequestSchema, DiagnosticsRequestSchema, GetDaemonConfigRequestMessageSchema, SetDaemonConfigRequestMessageSchema, @@ -3302,22 +3294,22 @@ export const HubRelationshipStatusSchema = z.object({ "disconnecting", "revoked", ]), - relationshipId: z.string().nullable(), + daemonId: 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"), +export const HubManagementDaemonConnectResponseSchema = z.object({ + type: z.literal("hub.management.daemon.connect.response"), payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema }), }); -export const HubRelationshipGetStatusResponseSchema = z.object({ - type: z.literal("hub.relationship.get_status.response"), +export const HubManagementDaemonGetStatusResponseSchema = z.object({ + type: z.literal("hub.management.daemon.get_status.response"), payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema }), }); -export const HubRelationshipDisconnectResponseSchema = z.object({ - type: z.literal("hub.relationship.disconnect.response"), +export const HubManagementDaemonDisconnectResponseSchema = z.object({ + type: z.literal("hub.management.daemon.disconnect.response"), payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema, @@ -4414,8 +4406,8 @@ export const DaemonUpdateProgressMessageSchema = z.object({ }), }); -export const HubAgentCreateResponseSchema = z.object({ - type: z.literal("hub.agent.create.response"), +export const HubExecutionAgentCreateResponseSchema = z.object({ + type: z.literal("hub.execution.agent.create.response"), payload: z.object({ requestId: z.string(), executionId: z.string(), @@ -4426,8 +4418,8 @@ export const HubAgentCreateResponseSchema = z.object({ }), }); -export const HubAgentUpdateSchema = z.object({ - type: z.literal("hub.agent.update"), +export const HubExecutionAgentUpdateSchema = z.object({ + type: z.literal("hub.execution.agent.update"), payload: z.object({ executionId: z.string(), agentId: z.string(), @@ -4435,8 +4427,8 @@ export const HubAgentUpdateSchema = z.object({ }), }); -export const HubAgentStreamSchema = z.object({ - type: z.literal("hub.agent.stream"), +export const HubExecutionAgentStreamSchema = z.object({ + type: z.literal("hub.execution.agent.stream"), payload: z.object({ executionId: z.string(), agentId: z.string(), @@ -4444,50 +4436,27 @@ export const HubAgentStreamSchema = z.object({ }), }); -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 type HubExecutionAgentCreateResponse = z.infer; +export type HubExecutionAgentUpdate = z.infer; +export type HubExecutionAgentStream = z.infer; -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 const HubExecutionOutboundMessageSchema = z.discriminatedUnion("type", [ + HubExecutionAgentCreateResponseSchema, + HubExecutionAgentUpdateSchema, + HubExecutionAgentStreamSchema, ]); -export type HubSessionOutboundMessage = z.infer; +export type HubExecutionOutboundMessage = z.infer; export class HubMessageCorrelationError extends Error { - constructor(messageType: HubSessionOutboundMessage["type"]) { + constructor(messageType: HubExecutionOutboundMessage["type"]) { super(`Hub message ${messageType} has mismatched agent correlation`); this.name = "HubMessageCorrelationError"; } } -export function parseHubSessionOutboundMessage(value: unknown): HubSessionOutboundMessage { - const message = HubSessionOutboundMessageSchema.parse(value); +export function parseHubExecutionOutboundMessage(value: unknown): HubExecutionOutboundMessage { + const message = HubExecutionOutboundMessageSchema.parse(value); const payload = message.payload; if ( "agent" in payload && @@ -4504,11 +4473,9 @@ export function parseHubSessionOutboundMessage(value: unknown): HubSessionOutbou export type DaemonUpdateProgressMessage = z.infer; export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ - HubAgentCreateResponseSchema, - HubAgentUpdateSchema, - HubAgentStreamSchema, - HubExecutionReconcileResponseSchema, - HubAuthorizationDeniedSchema, + HubExecutionAgentCreateResponseSchema, + HubExecutionAgentUpdateSchema, + HubExecutionAgentStreamSchema, BrowserAutomationExecuteRequestSchema, ActivityLogMessageSchema, AssistantChunkMessageSchema, @@ -4556,9 +4523,9 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ SetVoiceModeResponseMessageSchema, DaemonGetStatusResponseSchema, DaemonGetPairingOfferResponseSchema, - HubRelationshipConnectResponseSchema, - HubRelationshipGetStatusResponseSchema, - HubRelationshipDisconnectResponseSchema, + HubManagementDaemonConnectResponseSchema, + HubManagementDaemonGetStatusResponseSchema, + HubManagementDaemonDisconnectResponseSchema, DiagnosticsResponseSchema, GetDaemonConfigResponseMessageSchema, SetDaemonConfigResponseMessageSchema, diff --git a/packages/server/src/server/agent/agent-owner.ts b/packages/server/src/server/agent/agent-owner.ts index 24b5938ae9..7dd9e94420 100644 --- a/packages/server/src/server/agent/agent-owner.ts +++ b/packages/server/src/server/agent/agent-owner.ts @@ -2,15 +2,15 @@ import { z } from "zod"; export const AgentOwnerSchema = z.discriminatedUnion("kind", [ z.object({ - kind: z.literal("hub"), - relationshipId: z.string(), + kind: z.literal("daemon"), + daemonId: z.string(), executionId: z.string(), }), ]); export type AgentOwner = z.infer; -export type HubAgentOwner = Extract; +export type DaemonAgentOwner = Extract; -export function hubExecutionKey(owner: HubAgentOwner): string { - return `${owner.relationshipId}\0${owner.executionId}`; +export function daemonExecutionKey(owner: DaemonAgentOwner): string { + return `${owner.daemonId}\0${owner.executionId}`; } diff --git a/packages/server/src/server/agent/agent-storage.ts b/packages/server/src/server/agent/agent-storage.ts index 8d48454235..5e504cee64 100644 --- a/packages/server/src/server/agent/agent-storage.ts +++ b/packages/server/src/server/agent/agent-storage.ts @@ -1,4 +1,4 @@ -import { promises as fs } from "node:fs"; +import { promises as fs, type Dirent } from "node:fs"; import path from "node:path"; import { z } from "zod"; import type { Logger } from "pino"; @@ -8,7 +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"; +import { AgentOwnerSchema, daemonExecutionKey, type DaemonAgentOwner } from "./agent-owner.js"; const SERIALIZABLE_CONFIG_SCHEMA = z .object({ @@ -90,8 +90,8 @@ export class AgentStorage { private pathsById: Map> = new Map(); private pendingWrites: Map> = new Map(); private deleting: Set = new Set(); - private hubAgentIdsByExecution: Map = new Map(); - private hubExecutionKeysByAgentId: Map = new Map(); + private daemonAgentIdsByExecution: Map = new Map(); + private daemonExecutionKeysByAgentId: Map = new Map(); private loaded = false; private baseDir: string; private loadPromise: Promise | null = null; @@ -116,9 +116,9 @@ export class AgentStorage { return this.cache.get(agentId) ?? null; } - async findByHubExecution(owner: HubAgentOwner): Promise { + async findByDaemonExecution(owner: DaemonAgentOwner): Promise { await this.load(); - const agentId = this.hubAgentIdsByExecution.get(hubExecutionKey(owner)); + const agentId = this.daemonAgentIdsByExecution.get(daemonExecutionKey(owner)); return agentId ? (this.cache.get(agentId) ?? null) : null; } @@ -260,8 +260,8 @@ export class AgentStorage { this.cache.clear(); this.pathById.clear(); this.pathsById.clear(); - this.hubAgentIdsByExecution.clear(); - this.hubExecutionKeysByAgentId.clear(); + this.daemonAgentIdsByExecution.clear(); + this.daemonExecutionKeysByAgentId.clear(); try { const records = await this.scanDisk(); @@ -280,7 +280,7 @@ export class AgentStorage { private async scanDisk(): Promise { const records: StoredAgentRecord[] = []; - let entries: Array = []; + let entries: Dirent[] = []; try { entries = await fs.readdir(this.baseDir, { withFileTypes: true }); } catch (error) { @@ -367,24 +367,24 @@ export class AgentStorage { private indexOwner(record: StoredAgentRecord): void { this.removeOwnerIndex(record.id); - if (record.owner?.kind === "hub") { - const key = hubExecutionKey(record.owner); - const previousAgentId = this.hubAgentIdsByExecution.get(key); + if (record.owner?.kind === "daemon") { + const key = daemonExecutionKey(record.owner); + const previousAgentId = this.daemonAgentIdsByExecution.get(key); if (previousAgentId && previousAgentId !== record.id) { - this.hubExecutionKeysByAgentId.delete(previousAgentId); + this.daemonExecutionKeysByAgentId.delete(previousAgentId); } - this.hubAgentIdsByExecution.set(key, record.id); - this.hubExecutionKeysByAgentId.set(record.id, key); + this.daemonAgentIdsByExecution.set(key, record.id); + this.daemonExecutionKeysByAgentId.set(record.id, key); } } private removeOwnerIndex(agentId: string): void { - const key = this.hubExecutionKeysByAgentId.get(agentId); + const key = this.daemonExecutionKeysByAgentId.get(agentId); if (!key) return; - if (this.hubAgentIdsByExecution.get(key) === agentId) { - this.hubAgentIdsByExecution.delete(key); + if (this.daemonAgentIdsByExecution.get(key) === agentId) { + this.daemonAgentIdsByExecution.delete(key); } - this.hubExecutionKeysByAgentId.delete(agentId); + this.daemonExecutionKeysByAgentId.delete(agentId); } private async waitForPendingWrite(agentId: string): Promise { diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 4796f1ef8c..88c979469d 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -183,7 +183,7 @@ import { DirectHubRelationshipRemote, type HubRelationshipRemote, } from "./hub/relationship-remote.js"; -import { RelationshipOwnedExecutions } from "./hub/relationship-owned-executions.js"; +import { DaemonExecutions } from "./hub/daemon-executions.js"; const MAX_MCP_DEBUG_BATCH_ITEMS = 10; const REDACTED_LOG_VALUE = "[redacted]"; @@ -1041,9 +1041,9 @@ export async function createPaseoDaemon( if (!wsServer) throw new Error("WebSocket server is not running"); await wsServer.attachHubSocket(socket, options); }, - createExecutions: (relationshipId) => - new RelationshipOwnedExecutions({ - relationshipId, + createExecutionAgents: (daemonId) => + new DaemonExecutions({ + daemonId, agentManager, agentStorage, createAgent, diff --git a/packages/server/src/server/hub/relationship-owned-executions.test.ts b/packages/server/src/server/hub/daemon-executions.test.ts similarity index 90% rename from packages/server/src/server/hub/relationship-owned-executions.test.ts rename to packages/server/src/server/hub/daemon-executions.test.ts index 8c2cd0fa62..363f341d03 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.test.ts +++ b/packages/server/src/server/hub/daemon-executions.test.ts @@ -24,18 +24,15 @@ test("sequential replay after reconstruction keeps one durable owned agent", asy expect(reconstructed.replay.agent.id).toBe(created.first.agentId); expect(reconstructed.replay.agent.status).toBe("closed"); - expect(reconstructed.reconciliation?.agent.id).toBe(created.first.agentId); - expect(reconstructed.reconciliation?.agent.status).toBe("closed"); expect(reconstructed.durableAgentCount).toBe(1); }); -test("removing an owned agent removes its reconstructed execution association", async () => { +test("removing a daemon-owned agent removes its execution association", async () => { const hub = await launchRelationship(); const created = await hub.createOwnedConcurrently(); - const removed = await hub.removeAndReconcile(created.first.agentId); + const removed = await hub.removeOwnedAgent(created.first.agentId); - expect(removed.reconciliation).toBeNull(); expect(removed.durableAgentCount).toBe(0); }); @@ -49,7 +46,7 @@ test("a failed Hub create removes its auto-created worktree", async () => { const response = await hub.ownedCreateResult("failed-worktree-create"); expect(response).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: false, executionId: "failed-worktree-execution" }, }); expect(await hub.listedWorktrees()).toHaveLength(1); @@ -68,7 +65,7 @@ test("failed Hub auto-archive creates release their lifecycle subscriptions", as const first = await hub.ownedCreateResult("failed-prompt-create-1"); expect(first).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: false, executionId: "failed-prompt-execution-1" }, }); expect(hub.activeOwnedAgentIds()).toEqual([]); @@ -84,7 +81,7 @@ test("failed Hub auto-archive creates release their lifecycle subscriptions", as const second = await hub.ownedCreateResult("failed-prompt-create-2"); expect(second).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: false, executionId: "failed-prompt-execution-2" }, }); expect(hub.activeOwnedAgentIds()).toEqual([]); @@ -104,7 +101,7 @@ test("failed Hub create cleans durable state when provider close rejects", async const response = await hub.ownedCreateResult("failed-close-create"); expect(response).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: false, executionId: "failed-close-execution" }, }); expect(hub.activeOwnedAgentIds()).toEqual([]); @@ -122,7 +119,7 @@ test("Hub checkout uses the requested branch ref", async () => { const response = await hub.ownedCreateResult("checkout-create"); expect(response).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: true, executionId: "checkout-execution" }, }); expect(await hub.currentBranch(hub.latestCreatedCwd()!)).toBe("existing-hub-branch"); @@ -148,7 +145,7 @@ test("failed create never archives a reused worktree", async () => { const reused = await hub.ownedCreateResult("reused-create"); expect(reused).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: false, executionId: "reused-execution" }, }); expect(await hub.worktreeState(worktreeCwd!)).toEqual({ exists: true, listed: true }); diff --git a/packages/server/src/server/hub/relationship-owned-executions.ts b/packages/server/src/server/hub/daemon-executions.ts similarity index 82% rename from packages/server/src/server/hub/relationship-owned-executions.ts rename to packages/server/src/server/hub/daemon-executions.ts index f8f206fd5c..83b4d0693f 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.ts +++ b/packages/server/src/server/hub/daemon-executions.ts @@ -11,9 +11,9 @@ 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"; +import { daemonExecutionKey, type DaemonAgentOwner } from "../agent/agent-owner.js"; -export interface HubAgentCreateInput { +export interface HubExecutionAgentCreateInput { executionId: string; provider: string; cwd: string; @@ -42,8 +42,8 @@ export type OwnedAgentEvent = event: AgentStreamEventPayload; }; -interface RelationshipOwnedExecutionsOptions { - relationshipId: string; +interface DaemonExecutionsOptions { + daemonId: string; agentManager: AgentManager; agentStorage: AgentStorage; createAgent: BoundCreateAgentCommand; @@ -57,15 +57,14 @@ interface RelationshipOwnedExecutionsOptions { }) => Promise; } -export interface HubExecutions { - create(input: HubAgentCreateInput): Promise; - reconcile(executionId: string): Promise; +export interface HubExecutionAgents { + create(input: HubExecutionAgentCreateInput): Promise; subscribe(listener: (event: OwnedAgentEvent) => void): () => void; invalidateAuthority(): Promise; } -export class RelationshipOwnedExecutions implements HubExecutions { - private readonly relationshipId: string; +export class DaemonExecutions implements HubExecutionAgents { + private readonly daemonId: string; private readonly agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly createAgentCommand: BoundCreateAgentCommand; @@ -76,12 +75,10 @@ export class RelationshipOwnedExecutions implements HubExecutions { agentId: string; createdWorktree: CreatePaseoWorktreeWorkflowResult | null; }) => LifecycleRegistration; - private readonly cleanupFailedCreate: NonNullable< - RelationshipOwnedExecutionsOptions["cleanupFailedCreate"] - >; + private readonly cleanupFailedCreate: NonNullable; - constructor(options: RelationshipOwnedExecutionsOptions) { - this.relationshipId = options.relationshipId; + constructor(options: DaemonExecutionsOptions) { + this.daemonId = options.daemonId; this.agentManager = options.agentManager; this.agentStorage = options.agentStorage; this.createAgentCommand = options.createAgent; @@ -90,12 +87,12 @@ export class RelationshipOwnedExecutions implements HubExecutions { this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined); } - create(input: HubAgentCreateInput): Promise { + create(input: HubExecutionAgentCreateInput): Promise { if (!this.authorityActive) { return Promise.reject(new Error("Hub relationship authority is no longer active")); } const owner = this.owner(input.executionId); - const key = hubExecutionKey(owner); + const key = daemonExecutionKey(owner); const pending = this.pendingCreates.get(key); if (pending) { return pending; @@ -111,14 +108,6 @@ export class RelationshipOwnedExecutions implements HubExecutions { return create; } - async reconcile(executionId: string): Promise { - const owner = this.owner(executionId); - const pending = this.pendingCreates.get(hubExecutionKey(owner)); - if (pending) return pending; - const record = await this.agentStorage.findByHubExecution(owner); - return record ? this.resolveRecord(record) : null; - } - async invalidateAuthority(): Promise { this.authorityActive = false; this.authorityGeneration++; @@ -138,11 +127,11 @@ export class RelationshipOwnedExecutions implements HubExecutions { } private async createOrResolve( - owner: HubAgentOwner, - input: HubAgentCreateInput, + owner: DaemonAgentOwner, + input: HubExecutionAgentCreateInput, authorityGeneration: number, ): Promise { - const existing = await this.agentStorage.findByHubExecution(owner); + const existing = await this.agentStorage.findByDaemonExecution(owner); if (existing) { this.requireAuthority(authorityGeneration); return this.resolveRecord(existing); @@ -269,18 +258,18 @@ export class RelationshipOwnedExecutions implements HubExecutions { }; } - private isOwned(agent: ManagedAgent | null): agent is ManagedAgent & { owner: HubAgentOwner } { - return agent?.owner?.kind === "hub" && agent.owner.relationshipId === this.relationshipId; + private isOwned(agent: ManagedAgent | null): agent is ManagedAgent & { owner: DaemonAgentOwner } { + return agent?.owner?.kind === "daemon" && agent.owner.daemonId === this.daemonId; } - private owner(executionId: string): HubAgentOwner { - return { kind: "hub", relationshipId: this.relationshipId, executionId }; + private owner(executionId: string): DaemonAgentOwner { + return { kind: "daemon", daemonId: this.daemonId, executionId }; } - private requireOwner(record: StoredAgentRecord): HubAgentOwner { + private requireOwner(record: StoredAgentRecord): DaemonAgentOwner { 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}`); + if (owner?.kind !== "daemon" || owner.daemonId !== this.daemonId) { + throw new Error(`Agent ${record.id} is not owned by daemon ${this.daemonId}`); } return owner; } diff --git a/packages/server/src/server/hub/execution-controller.ts b/packages/server/src/server/hub/execution-controller.ts new file mode 100644 index 0000000000..a12853950a --- /dev/null +++ b/packages/server/src/server/hub/execution-controller.ts @@ -0,0 +1,105 @@ +import { isAbsolute } from "node:path"; +import type { + HubExecutionAgentCreateRequest, + SessionOutboundMessage, +} from "@getpaseo/protocol/messages"; + +import type { HubExecutionAgents, OwnedAgentEvent } from "./daemon-executions.js"; + +interface HubExecutionControllerOptions { + agents: HubExecutionAgents; + send: (message: SessionOutboundMessage) => void; +} + +export class HubExecutionController { + private readonly agents: HubExecutionAgents; + private readonly send: (message: SessionOutboundMessage) => void; + private readonly unsubscribe: () => void; + + constructor(options: HubExecutionControllerOptions) { + this.agents = options.agents; + this.send = options.send; + this.unsubscribe = this.agents.subscribe((event) => this.sendOwnedEvent(event)); + } + + cleanup(): void { + this.unsubscribe(); + } + + async createAgent(message: HubExecutionAgentCreateRequest): Promise { + try { + requireNonBlankHubAgentField("executionId", message.executionId); + requireNonBlankHubAgentField("prompt", message.prompt); + requireNonBlankHubAgentField("cwd", message.cwd); + if (!isAbsolute(message.cwd)) throw new Error("Hub agent cwd must be absolute"); + const result = await this.agents.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.execution.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.execution.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 sendOwnedEvent(event: OwnedAgentEvent): void { + if (event.type === "update") { + this.send({ + type: "hub.execution.agent.update", + payload: { + executionId: event.executionId, + agentId: event.agent.id, + agent: event.agent, + }, + }); + return; + } + this.send({ + type: "hub.execution.agent.stream", + payload: { + executionId: event.executionId, + agentId: event.agentId, + event: event.event, + }, + }); + } +} + +function requireNonBlankHubAgentField( + field: "executionId" | "prompt" | "cwd", + value: string, +): void { + if (value.trim().length === 0) { + throw new Error(`Hub agent ${field} cannot be blank`); + } +} diff --git a/packages/server/src/server/hub/hub-session.websocket.test.ts b/packages/server/src/server/hub/execution-session.websocket.test.ts similarity index 90% rename from packages/server/src/server/hub/hub-session.websocket.test.ts rename to packages/server/src/server/hub/execution-session.websocket.test.ts index 1a90b06256..caeccd6c60 100644 --- a/packages/server/src/server/hub/hub-session.websocket.test.ts +++ b/packages/server/src/server/hub/execution-session.websocket.test.ts @@ -19,13 +19,12 @@ async function launchRelationship(): Promise { return launched; } -test("Hub owns one durable execution across concurrency and reconstruction", async () => { +test("Hub retries one durable daemon 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); @@ -35,13 +34,7 @@ test("Hub owns one durable execution across concurrency and reconstruction", asy 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); }); @@ -55,12 +48,14 @@ test("Hub denies trusted steering and browser dispatch", async () => { expect(steeringDenial).toEqual({ requestId: "denied-steer", requestType: "send_agent_message_request", - code: "scope_denied", + error: "Session is not authorized for send_agent_message_request", + code: "access_denied", }); expect(browserDenial).toEqual({ requestId: "browser-1", requestType: "browser.automation.execute.response", - code: "scope_denied", + error: "Session is not authorized for browser.automation.execute.response", + code: "access_denied", }); expect(hub.observedAgentIds()).not.toContain(localAgentId); expect(hub.observedTrustedLifecycleMessages()).toEqual([]); @@ -93,7 +88,7 @@ test("Hub reconnects without retaining trusted session state", async () => { const hub = await launchRelationship(); const created = await hub.createOwnedConcurrently(); - const reconnected = await hub.reconnectAndReconcile(); + const reconnected = await hub.reconnectAndRetry(); expect(reconnected).toMatchObject({ executionId: "execution-1", @@ -121,7 +116,7 @@ test("Hub create forwards worktree and auto-archive through the existing create const afterArchive = await hub.worktreeState(worktreeCwd!); expect(worktreeCreated).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: true, agent: { cwd: worktreeCwd } }, }); expect(worktreeCwd).not.toBe(hub.repoRoot()); diff --git a/packages/server/src/server/hub/hub-session.test.ts b/packages/server/src/server/hub/hub-session.test.ts deleted file mode 100644 index 3ad2ac08a4..0000000000 --- a/packages/server/src/server/hub/hub-session.test.ts +++ /dev/null @@ -1,259 +0,0 @@ -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>(); - private reconcileError: Error | null = null; - private creates = 0; - private reconciles = 0; - - async create(_input: HubAgentCreateInput): Promise { - this.creates++; - return { executionId: "execution-1", agent: createAgentSnapshot() }; - } - - async reconcile(_executionId: string): Promise { - this.reconciles++; - if (this.reconcileError) throw this.reconcileError; - return { executionId: "execution-1", agent: createAgentSnapshot() }; - } - - subscribe(listener: (event: OwnedAgentEvent) => void): () => void { - this.listeners.add(listener); - return () => this.listeners.delete(listener); - } - - async invalidateAuthority(): Promise {} - - publishCurrentState(): void { - for (const listener of this.listeners) { - listener({ type: "update", executionId: "execution-1", agent: createAgentSnapshot() }); - } - } - - failReconcile(error: Error): void { - this.reconcileError = error; - } - - createCount(): number { - return this.creates; - } - - reconcileCount(): number { - return this.reconciles; - } -} - -interface HubCreateInput { - executionId?: string; - prompt: string; - cwd: string; -} - -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(); - } - - async failReconcile(): Promise { - this.executions.failReconcile(new Error("storage unavailable")); - await this.session.handleMessage({ - type: "hub.execution.reconcile.request", - requestId: "reconcile-1", - executionId: "execution-1", - }); - } - - async createWith(input: HubCreateInput): Promise { - await this.session.handleMessage({ - type: "hub.agent.create.request", - requestId: "create-1", - executionId: input.executionId ?? "execution-1", - provider: "codex", - workspaceId: "workspace-1", - ...input, - }); - } - - createCount(): number { - return this.executions.createCount(); - } - - async reconcileWith(executionId: string): Promise { - await this.session.handleMessage({ - type: "hub.execution.reconcile.request", - requestId: "reconcile-1", - executionId, - }); - } - - reconcileCount(): number { - return this.executions.reconcileCount(); - } - - deliveredMessages(): SessionOutboundMessage[] { - return this.messages.slice(); - } - - 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"]); -}); - -test("reconcile storage failures receive a terminal empty response", async () => { - const hub = new HubSessionBoundary(); - - await hub.failReconcile(); - - expect(hub.deliveredMessages()).toEqual([ - { - type: "hub.execution.reconcile.response", - payload: { - requestId: "reconcile-1", - executionId: "execution-1", - agentId: null, - agent: null, - }, - }, - ]); -}); - -test.each([ - { field: "prompt", prompt: " ", cwd: "/workspace" }, - { field: "cwd", prompt: "Do the work", cwd: "\t" }, -])("blank Hub $field values reject the create before execution", async ({ field, prompt, cwd }) => { - const hub = new HubSessionBoundary(); - - await hub.createWith({ prompt, cwd }); - - expect(hub.createCount()).toBe(0); - expect(hub.deliveredMessages()).toEqual([ - { - type: "hub.agent.create.response", - payload: { - requestId: "create-1", - executionId: "execution-1", - agentId: null, - agent: null, - success: false, - error: `Hub agent ${field} cannot be blank`, - }, - }, - ]); -}); - -test("blank Hub execution IDs reject create and reconcile before execution ownership", async () => { - const hub = new HubSessionBoundary(); - - await hub.createWith({ executionId: " ", prompt: "Do the work", cwd: "/workspace" }); - await hub.reconcileWith("\t"); - - expect(hub.createCount()).toBe(0); - expect(hub.reconcileCount()).toBe(0); - expect(hub.deliveredMessages()).toEqual([ - { - type: "hub.agent.create.response", - payload: { - requestId: "create-1", - executionId: " ", - agentId: null, - agent: null, - success: false, - error: "Hub agent executionId cannot be blank", - }, - }, - { - type: "hub.execution.reconcile.response", - payload: { - requestId: "reconcile-1", - executionId: "\t", - agentId: null, - agent: null, - }, - }, - ]); -}); - -test("relative Hub working directories reject create before execution ownership", async () => { - const hub = new HubSessionBoundary(); - - await hub.createWith({ prompt: "Do the work", cwd: "repo" }); - - expect(hub.createCount()).toBe(0); - expect(hub.deliveredMessages()).toEqual([ - { - type: "hub.agent.create.response", - payload: { - requestId: "create-1", - executionId: "execution-1", - agentId: null, - agent: null, - success: false, - error: "Hub agent cwd must be absolute", - }, - }, - ]); -}); - -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 deleted file mode 100644 index 197d12bc6b..0000000000 --- a/packages/server/src/server/hub/hub-session.ts +++ /dev/null @@ -1,163 +0,0 @@ -import { isAbsolute } from "node:path"; -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 { - requireNonBlankHubAgentField("executionId", message.executionId); - requireNonBlankHubAgentField("prompt", message.prompt); - requireNonBlankHubAgentField("cwd", message.cwd); - if (!isAbsolute(message.cwd)) throw new Error("Hub agent cwd must be absolute"); - 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 { - let result: Awaited> = null; - try { - requireNonBlankHubAgentField("executionId", message.executionId); - result = await this.executions.reconcile(message.executionId); - } catch { - // Reconcile has no error variant on the wire. A terminal empty response - // lets Hub stop waiting and decide whether to replay the execution. - } - this.send({ - type: "hub.execution.reconcile.response", - payload: { - 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 requireNonBlankHubAgentField( - field: "executionId" | "prompt" | "cwd", - value: string, -): void { - if (value.trim().length === 0) { - throw new Error(`Hub agent ${field} cannot be blank`); - } -} - -function requestIdForMessage(message: SessionInboundMessage): string | undefined { - if ("requestId" in message && typeof message.requestId === "string") { - return message.requestId; - } - 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/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index 6c499cb664..ca0df85709 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -60,9 +60,9 @@ describe("Hub relationship", () => { const responses = await relationship.manageRelationshipFromExternalSocket(); expect(responses.map((response) => response.type)).toEqual([ - "hub.relationship.connect.response", - "hub.relationship.get_status.response", - "hub.relationship.disconnect.response", + "hub.management.daemon.connect.response", + "hub.management.daemon.get_status.response", + "hub.management.daemon.disconnect.response", ]); expect(relationship.enrollmentAttempts()).toHaveLength(1); expect(relationship.relationshipFile()).toBeNull(); @@ -73,7 +73,7 @@ describe("Hub relationship", () => { const response = await relationship.connectFromBrowserSocket(); - expect(response.type).toBe("hub.relationship.connect.response"); + expect(response.type).toBe("hub.management.daemon.connect.response"); expect(relationship.enrollmentAttempts()).toHaveLength(1); expect(relationship.relationshipFile()?.state).toBe("active"); }); @@ -90,7 +90,10 @@ describe("Hub relationship", () => { expect(pending).toMatchObject({ record: { state: "pending", - relationship: { id: enrollment.relationshipId, idempotencyKey: enrollment.idempotencyKey }, + relationship: { + daemonId: enrollment.daemonId, + idempotencyKey: enrollment.idempotencyKey, + }, credential: { secret: expect.any(String) }, enrollment: { token: "one-time-token" }, identity: { serverId: expect.any(String), daemonPublicKey: expect.any(String) }, @@ -109,7 +112,7 @@ describe("Hub relationship", () => { await relationship.socketDialed(); expect(relationship.socketInvocation()).toMatchObject({ mode: privateFileMode, - record: { state: "active", relationship: { id: enrollment.relationshipId } }, + record: { state: "active", relationship: { daemonId: enrollment.daemonId } }, }); }); @@ -235,7 +238,7 @@ describe("Hub relationship", () => { idempotencyKey: "ceremony-1", hubOrigin: "https://hub.test", createdAt: "2026-07-13T00:00:00.000Z", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }, credential: { secret: "credential" }, transport: { kind: "direct_websocket", webSocketUrl: "ftp://hub.test/daemon" }, @@ -261,7 +264,7 @@ describe("Hub relationship", () => { idempotencyKey: "ceremony-1", hubOrigin: "https://hub.test", createdAt: "2026-07-13T00:00:00.000Z", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }, credential: { secret: "credential" }, transport: { @@ -299,7 +302,7 @@ describe("Hub relationship", () => { expect(relationship.revocationAttempts()).toBe(3); expect(relationship.latestRevocation()).toEqual( expect.objectContaining({ - relationshipId: enrollment.relationshipId, + daemonId: enrollment.daemonId, hubOrigin: enrollment.hubOrigin, credential, }), @@ -322,7 +325,7 @@ describe("Hub relationship", () => { const disconnected = await disconnecting.result; expect(disconnected.state).toBe("not_connected"); - expect(relationship.latestRevocation()?.relationshipId).toBe(enrollment.relationshipId); + expect(relationship.latestRevocation()?.daemonId).toBe(enrollment.daemonId); expect(relationship.revocationAttempts()).toBe(1); expect(relationship.socketAttempts()).toBe(0); expect(relationship.relationshipFile()).toBeNull(); @@ -331,14 +334,14 @@ describe("Hub relationship", () => { test("daemon restart reconnects the same durable relationship", async () => { relationship = await HubRelationshipHarness.start(); await relationship.beginConnect().result; - const id = relationship.relationshipFile()?.relationship.id; + const id = relationship.relationshipFile()?.relationship.daemonId; relationship.connectLatestSocket(); await relationship.socketDialed(); await relationship.restartDaemon(); await relationship.socketDialed(); - expect(relationship.relationshipFile()?.relationship.id).toBe(id); + expect(relationship.relationshipFile()?.relationship.daemonId).toBe(id); expect(relationship.socketAttempts()).toBe(2); }); @@ -347,7 +350,7 @@ describe("Hub relationship", () => { await relationship.beginConnect().result; await relationship.socketDialed(); relationship.connectLatestSocket(); - const relationshipId = relationship.relationshipFile()?.relationship.id; + const daemonId = relationship.relationshipFile()?.relationship.daemonId; const prompt = "sleep 30"; relationship.beginOwnedCreate("running-create", "execution-running", { prompt, @@ -359,14 +362,8 @@ describe("Hub relationship", () => { await relationship.restartDaemon(); await relationship.socketDialed(); relationship.connectLatestSocket(); - const reconciled = await relationship.reconcileOwned("execution-running"); relationship.beginOwnedCreate("running-duplicate", "execution-running", { prompt }); const duplicate = await relationship.ownedCreateResult("running-duplicate"); - expect(reconciled).toMatchObject({ - executionId: "execution-running", - agentId: created.payload.agentId, - agent: { id: created.payload.agentId, status: "closed" }, - }); expect(duplicate).toMatchObject({ payload: { success: true, @@ -382,7 +379,7 @@ describe("Hub relationship", () => { agentId: created.payload.agentId, agent: { id: created.payload.agentId, status: "running" }, }); - expect(relationship.relationshipFile()?.relationship.id).toBe(relationshipId); + expect(relationship.relationshipFile()?.relationship.daemonId).toBe(daemonId); expect(durableAgentIds).toEqual([created.payload.agentId]); expect(relationship.providerCreations()).toBe(1); expect(relationship.providerResumes()).toBe(0); @@ -413,7 +410,7 @@ describe("Hub relationship", () => { const duplicate = await relationship.ownedCreateResult("completed-duplicate"); expect(duplicate).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: true, executionId: "execution-completed", @@ -438,12 +435,15 @@ describe("Hub relationship", () => { await relationship.restartDaemon(); await relationship.socketDialed(); relationship.connectLatestSocket(); - const reconciled = await relationship.reconcileOwned("execution-idle"); + relationship.beginOwnedCreate("idle-retry", "execution-idle", { prompt }); + const retried = await relationship.ownedCreateResult("idle-retry"); - expect(reconciled).toMatchObject({ - executionId: "execution-idle", - agentId: created.payload.agentId, - agent: { id: created.payload.agentId, status: "closed" }, + expect(retried).toMatchObject({ + payload: { + executionId: "execution-idle", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "closed" }, + }, }); expect(relationship.providerPromptTexts()).toEqual([prompt]); expect(relationship.providerResumes()).toBe(0); @@ -462,17 +462,20 @@ describe("Hub relationship", () => { await relationship.restartDaemon(); await relationship.socketDialed(); relationship.connectLatestSocket(); - const reconciled = await relationship.reconcileOwned("execution-failed"); + relationship.beginOwnedCreate("failed-retry", "execution-failed", { prompt }); + const retried = await relationship.ownedCreateResult("failed-retry"); expect(failed).toMatchObject({ executionId: "execution-failed", agentId: created.payload.agentId, event: { type: "turn_failed" }, }); - expect(reconciled).toMatchObject({ - executionId: "execution-failed", - agentId: created.payload.agentId, - agent: { id: created.payload.agentId, status: "closed" }, + expect(retried).toMatchObject({ + payload: { + executionId: "execution-failed", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "closed" }, + }, }); expect(relationship.providerPromptTexts()).toEqual([prompt]); expect(relationship.providerResumes()).toBe(0); @@ -506,11 +509,12 @@ describe("Hub relationship", () => { }); expect(messages).toContainEqual({ - type: "hub.authorization.denied", + type: "rpc_error", payload: { requestId: "still-current", requestType: "daemon.get_status.request", - code: "scope_denied", + error: "Session is not authorized for daemon.get_status.request", + code: "access_denied", }, }); expect(relationship.socketAttempts()).toBe(2); @@ -535,7 +539,7 @@ describe("Hub relationship", () => { expect(relationship.socketDeliveredResponse(0, "first-create")).toBe(false); expect(replayed).toMatchObject({ - type: "hub.agent.create.response", + type: "hub.execution.agent.create.response", payload: { success: true, executionId: "execution-race", @@ -546,7 +550,7 @@ describe("Hub relationship", () => { expect(durableAgentIds).toHaveLength(1); }); - test("reconcile waits for a pending create across socket generations", async () => { + test("idempotent retry waits for a pending create across socket generations", async () => { relationship = await HubRelationshipHarness.start(); await relationship.beginConnect().result; relationship.connectLatestSocket(); @@ -557,13 +561,15 @@ describe("Hub relationship", () => { relationship.closeLatestSocket(1006); await relationship.retry(); relationship.connectLatestSocket(); - const reconciliation = relationship.reconcileOwned("pending-execution"); + relationship.beginOwnedCreate("pending-retry", "pending-execution"); relationship.finishAgentCreation(); - await expect(reconciliation).resolves.toMatchObject({ - executionId: "pending-execution", - agentId: expect.any(String), - agent: { id: expect.any(String) }, + await expect(relationship.ownedCreateResult("pending-retry")).resolves.toMatchObject({ + payload: { + executionId: "pending-execution", + agentId: expect.any(String), + agent: { id: expect.any(String) }, + }, }); expect(relationship.providerCreations()).toBe(1); expect(await relationship.durableOwnedAgentIds()).toHaveLength(1); @@ -606,16 +612,16 @@ describe("Hub relationship", () => { expect(status).toMatchObject({ state: "revoked", - relationshipId: enrollment.relationshipId, + daemonId: enrollment.daemonId, hub: "https://hub.test", - scopes: "hub.*", + scopes: "hub.execution.*", error: reason, }); expect(persisted?.state).toBe("revoked"); expect(persisted?.relationship).toMatchObject({ - id: enrollment.relationshipId, + daemonId: enrollment.daemonId, hubOrigin: "https://hub.test", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }); expect(persisted?.reason).toBe(reason); expect(persisted).not.toHaveProperty("credential"); diff --git a/packages/server/src/server/hub/relationship-controller.ts b/packages/server/src/server/hub/relationship-controller.ts index 23a8916f64..32275f03b2 100644 --- a/packages/server/src/server/hub/relationship-controller.ts +++ b/packages/server/src/server/hub/relationship-controller.ts @@ -5,7 +5,7 @@ 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 { HubExecutionAgents } from "./daemon-executions.js"; import type { HubRelationshipRemote, HubSocketConnection, @@ -15,10 +15,10 @@ import { HubEnrollmentRejectedError } from "./relationship-remote.js"; import { BoundedExponentialHubRetryPolicy } from "./relationship-retry.js"; const FILE_NAME = "hub-relationship.json"; -const SCOPES = ["hub.*"] as const; +const SCOPES = ["hub.execution.*"] as const; const RelationshipSchema = z.object({ - id: z.string().min(1), + daemonId: z.string().min(1), idempotencyKey: z.string().min(1), hubOrigin: z.string().url(), createdAt: z.string(), @@ -93,7 +93,7 @@ export type HubConnectionState = export interface HubRelationshipStatus { state: HubConnectionState; - relationshipId: string | null; + daemonId: string | null; hubOrigin: string | null; scopes: string[]; connectedAt: string | null; @@ -131,9 +131,9 @@ export interface HubRelationshipControllerOptions { retryPolicy?: HubRelationshipRetryPolicy; attachSocket: ( socket: WebSocketLike, - options: { relationshipId: string; executions: HubExecutions }, + options: { daemonId: string; grants: readonly string[]; agents: HubExecutionAgents }, ) => Promise; - createExecutions: (relationshipId: string) => HubExecutions; + createExecutionAgents: (daemonId: string) => HubExecutionAgents; } const systemClock: HubRelationshipClock = { @@ -173,7 +173,7 @@ export class HubRelationshipController implements HubRelationshipManagement { private enrollmentGeneration = 0; private retryAttempt = 0; private readonly inFlightEnrollments = new Set>(); - private executions: { relationshipId: string; value: HubExecutions } | null = null; + private executionAgents: { daemonId: string; value: HubExecutionAgents } | null = null; constructor(private readonly options: HubRelationshipControllerOptions) { this.filePath = path.join(options.paseoHome, FILE_NAME); @@ -213,7 +213,7 @@ export class HubRelationshipController implements HubRelationshipManagement { status(): HubRelationshipStatus { return { state: this.state, - relationshipId: this.record?.relationship.id ?? null, + daemonId: this.record?.relationship.daemonId ?? null, hubOrigin: this.record?.relationship.hubOrigin ?? null, scopes: this.record?.relationship.scopes.slice() ?? [], connectedAt: this.connectedAt, @@ -241,7 +241,7 @@ export class HubRelationshipController implements HubRelationshipManagement { version: 1, state: "pending", relationship: { - id: randomUUID(), + daemonId: randomUUID(), idempotencyKey: randomUUID(), hubOrigin: normalizeHubUrl(input.hubUrl), createdAt: this.clock.now().toISOString(), @@ -263,7 +263,8 @@ export class HubRelationshipController implements HubRelationshipManagement { force: boolean; }): Promise<{ status: HubRelationshipStatus; warning?: string }> { const waitForEnrollment = this.record?.state === "pending"; - const pendingCreateCleanup = this.executions?.value.invalidateAuthority() ?? Promise.resolve(); + const pendingCreateCleanup = + this.executionAgents?.value.invalidateAuthority() ?? Promise.resolve(); this.cancelLifecycle(); this.socket?.close(); this.socket = null; @@ -302,7 +303,7 @@ export class HubRelationshipController implements HubRelationshipManagement { if (enrollmentGeneration !== this.enrollmentGeneration) return; const verifier = createHash("sha256").update(pending.credential.secret).digest("base64url"); const request = this.options.remote.enroll({ - relationshipId: pending.relationship.id, + daemonId: pending.relationship.daemonId, idempotencyKey: pending.relationship.idempotencyKey, hubOrigin: pending.relationship.hubOrigin, token: pending.enrollment.token, @@ -320,8 +321,8 @@ export class HubRelationshipController implements HubRelationshipManagement { const enrollment = await request; if (enrollmentGeneration !== this.enrollmentGeneration) return; if ( - enrollment.relationshipId !== pending.relationship.id || - !enrollment.scopes.includes("hub.*") + enrollment.daemonId !== pending.relationship.daemonId || + !enrollment.scopes.includes("hub.execution.*") ) { throw new Error("Hub enrollment response did not match the pending relationship"); } @@ -361,7 +362,7 @@ export class HubRelationshipController implements HubRelationshipManagement { }; this.socket = this.options.remote.openSocket( { - relationshipId: record.relationship.id, + daemonId: record.relationship.daemonId, webSocketUrl: record.transport.webSocketUrl, credential: record.credential.secret, }, @@ -379,15 +380,16 @@ export class HubRelationshipController implements HubRelationshipManagement { this.connectedAt = this.clock.now().toISOString(); this.lastError = null; void this.options.attachSocket(socket, { - relationshipId: record.relationship.id, - executions: this.executionsFor(record.relationship.id), + daemonId: record.relationship.daemonId, + grants: record.relationship.scopes, + agents: this.executionAgentsFor(record.relationship.daemonId), }); } - private executionsFor(relationshipId: string): HubExecutions { - if (this.executions?.relationshipId === relationshipId) return this.executions.value; - const value = this.options.createExecutions(relationshipId); - this.executions = { relationshipId, value }; + private executionAgentsFor(daemonId: string): HubExecutionAgents { + if (this.executionAgents?.daemonId === daemonId) return this.executionAgents.value; + const value = this.options.createExecutionAgents(daemonId); + this.executionAgents = { daemonId, value }; return value; } @@ -434,7 +436,7 @@ export class HubRelationshipController implements HubRelationshipManagement { const generation = this.generation; try { await this.options.remote.revoke({ - relationshipId: record.relationship.id, + daemonId: record.relationship.daemonId, hubOrigin: record.relationship.hubOrigin, credential: record.credential.secret, }); @@ -458,14 +460,14 @@ export class HubRelationshipController implements HubRelationshipManagement { } private revoke(reason: string): void { - void this.executions?.value.invalidateAuthority(); + void this.executionAgents?.value.invalidateAuthority(); this.cancelLifecycle(); if (!this.record) return; const revoked: RevokedRecord = { version: 1, state: "revoked", relationship: { - id: this.record.relationship.id, + daemonId: this.record.relationship.daemonId, hubOrigin: this.record.relationship.hubOrigin, createdAt: this.record.relationship.createdAt, scopes: this.record.relationship.scopes, @@ -498,7 +500,7 @@ export class HubRelationshipController implements HubRelationshipManagement { } private remove(): void { - void this.executions?.value.invalidateAuthority(); + void this.executionAgents?.value.invalidateAuthority(); this.cancelLifecycle(); rmSync(this.filePath, { force: true }); this.record = null; diff --git a/packages/server/src/server/hub/relationship-remote.test.ts b/packages/server/src/server/hub/relationship-remote.test.ts index dc0e0cc1f6..e4950adaba 100644 --- a/packages/server/src/server/hub/relationship-remote.test.ts +++ b/packages/server/src/server/hub/relationship-remote.test.ts @@ -6,7 +6,7 @@ import path from "node:path"; import pino from "pino"; import { afterEach, expect, test } from "vitest"; import { WebSocket } from "ws"; -import type { HubExecutions } from "./relationship-owned-executions.js"; +import type { HubExecutionAgents } from "./daemon-executions.js"; import { HubRelationshipController, type HubRelationshipClock, @@ -37,7 +37,7 @@ test.each([401, 403, 404])( const remote = new DirectHubRelationshipRemote(); await expect( - remote.revoke({ relationshipId: "relationship-1", hubOrigin, credential: "invalid" }), + remote.revoke({ daemonId: "daemon-1", hubOrigin, credential: "invalid" }), ).resolves.toBeUndefined(); }, ); @@ -47,7 +47,7 @@ test("transient revocation failures remain retryable", async () => { const remote = new DirectHubRelationshipRemote(); await expect( - remote.revoke({ relationshipId: "relationship-1", hubOrigin, credential: "credential" }), + remote.revoke({ daemonId: "daemon-1", hubOrigin, credential: "credential" }), ).rejects.toThrow("Hub revocation failed (503)"); }); @@ -57,14 +57,14 @@ test.each([408, 429])("transient enrollment status %s remains retryable", async const error = await remote .enroll({ - relationshipId: "relationship-1", + daemonId: "daemon-1", idempotencyKey: "ceremony-1", hubOrigin, token: "token", serverId: "server-1", daemonPublicKey: "public-key", credentialVerifier: "verifier", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }) .catch((caught: unknown) => caught); @@ -79,14 +79,14 @@ test("enrollment rejects a transport URL that cannot open a WebSocket", async () await expect( remote.enroll({ - relationshipId: "relationship-1", + daemonId: "daemon-1", idempotencyKey: "ceremony-1", hubOrigin, token: "token", serverId: "server-1", daemonPublicKey: "public-key", credentialVerifier: "verifier", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }), ).rejects.toThrow("Hub WebSocket URL must use ws or wss"); }); @@ -97,14 +97,14 @@ test("enrollment rejects a WebSocket URL with a fragment", async () => { await expect( remote.enroll({ - relationshipId: "relationship-1", + daemonId: "daemon-1", idempotencyKey: "ceremony-1", hubOrigin, token: "token", serverId: "server-1", daemonPublicKey: "public-key", credentialVerifier: "verifier", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }), ).rejects.toThrow("Hub WebSocket URL cannot include a fragment"); }); @@ -115,14 +115,14 @@ test("enrollment rejects a WebSocket outside the enrolled Hub authority", async await expect( remote.enroll({ - relationshipId: "relationship-1", + daemonId: "daemon-1", idempotencyKey: "ceremony-1", hubOrigin, token: "token", serverId: "server-1", daemonPublicKey: "public-key", credentialVerifier: "verifier", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }), ).rejects.toThrow("Hub WebSocket URL must match the Hub origin"); }); @@ -134,17 +134,17 @@ test.each(["enrollment", "revocation"])("%s HTTP calls are bounded", async (oper const request = operation === "enrollment" ? remote.enroll({ - relationshipId: "relationship-1", + daemonId: "daemon-1", idempotencyKey: "ceremony-1", hubOrigin, token: "token", serverId: "server-1", daemonPublicKey: "public-key", credentialVerifier: "verifier", - scopes: ["hub.*"], + scopes: ["hub.execution.*"], }) : remote.revoke({ - relationshipId: "relationship-1", + daemonId: "daemon-1", hubOrigin, credential: "credential", }); @@ -161,7 +161,7 @@ test.each([401, 403] as const)( const socket = remote.openSocket( { - relationshipId: "relationship-1", + daemonId: "daemon-1", webSocketUrl: hub.webSocketUrl, credential: "invalid", }, @@ -183,7 +183,7 @@ test("a transient socket response releases the failed upgrade and reports one co const socket = remote.openSocket( { - relationshipId: "relationship-1", + daemonId: "daemon-1", webSocketUrl: hub.webSocketUrl, credential: "credential", }, @@ -204,7 +204,7 @@ test("a stalled socket opening handshake times out and releases the connection", const socket = remote.openSocket( { - relationshipId: "relationship-1", + daemonId: "daemon-1", webSocketUrl: hub.webSocketUrl, credential: "credential", }, @@ -294,11 +294,11 @@ async function startEnrollmentHub(webSocketUrl: string): Promise { const server = createServer(async (request, response) => { let body = ""; for await (const chunk of request) body += chunk; - const enrollment = JSON.parse(body) as { relationshipId: string }; + const enrollment = JSON.parse(body) as { daemonId: string }; response.writeHead(200, { "content-type": "application/json" }).end( JSON.stringify({ - relationshipId: enrollment.relationshipId, - scopes: ["hub.*"], + daemonId: enrollment.daemonId, + scopes: ["hub.execution.*"], webSocketUrl, }), ); @@ -385,11 +385,11 @@ class UpgradeRejectingHub { private readonly server = createServer(async (request, response) => { let body = ""; for await (const chunk of request) body += chunk; - const enrollment = JSON.parse(body) as { relationshipId: string }; + const enrollment = JSON.parse(body) as { daemonId: string }; response.writeHead(200, { "content-type": "application/json" }).end( JSON.stringify({ - relationshipId: enrollment.relationshipId, - scopes: ["hub.*"], + daemonId: enrollment.daemonId, + scopes: ["hub.execution.*"], webSocketUrl: this.webSocketUrl, }), ); @@ -507,11 +507,10 @@ class ManualRelationshipClock implements HubRelationshipClock, HubRelationshipRe } } -const unusedExecutions: HubExecutions = { +const unusedExecutionAgents: HubExecutionAgents = { create: async () => { throw new Error("Unexpected Hub agent create"); }, - reconcile: async () => null, subscribe: () => () => undefined, invalidateAuthority: async () => undefined, }; @@ -531,7 +530,7 @@ async function connectController( clock, retryPolicy: clock, attachSocket: async () => undefined, - createExecutions: () => unusedExecutions, + createExecutionAgents: () => unusedExecutionAgents, }); await controller.connect({ hubUrl: hub.origin, token: "enrollment-token" }); return controller; diff --git a/packages/server/src/server/hub/relationship-remote.ts b/packages/server/src/server/hub/relationship-remote.ts index 5c109c0fe4..82172a411a 100644 --- a/packages/server/src/server/hub/relationship-remote.ts +++ b/packages/server/src/server/hub/relationship-remote.ts @@ -3,7 +3,7 @@ import { z } from "zod"; import type { WebSocketLike } from "../websocket-server.js"; export interface HubEnrollment { - relationshipId: string; + daemonId: string; idempotencyKey: string; hubOrigin: string; token: string; @@ -14,19 +14,19 @@ export interface HubEnrollment { } export interface HubEnrollmentResult { - relationshipId: string; + daemonId: string; scopes: string[]; webSocketUrl: string; } export interface HubRevocation { - relationshipId: string; + daemonId: string; hubOrigin: string; credential: string; } export interface HubSocketCredentials { - relationshipId: string; + daemonId: string; webSocketUrl: string; credential: string; } @@ -56,7 +56,7 @@ export class HubEnrollmentRejectedError extends Error { } const EnrollmentResultSchema = z.object({ - relationshipId: z.string(), + daemonId: z.string(), scopes: z.array(z.string()), webSocketUrl: z .string() @@ -87,14 +87,14 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { async enroll(input: HubEnrollment): Promise { return this.withRequestTimeout(async (signal) => { - const response = await fetch(`${input.hubOrigin}/api/daemon-relationships/enroll`, { + const response = await fetch(`${input.hubOrigin}/api/daemons/enroll`, { method: "POST", headers: { "content-type": "application/json", authorization: `Bearer ${input.token}`, }, body: JSON.stringify({ - relationshipId: input.relationshipId, + daemonId: input.daemonId, idempotencyKey: input.idempotencyKey, serverId: input.serverId, daemonPublicKey: input.daemonPublicKey, @@ -118,7 +118,7 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { async revoke(input: HubRevocation): Promise { await this.withRequestTimeout(async (signal) => { const response = await fetch( - `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`, + `${input.hubOrigin}/api/daemons/${encodeURIComponent(input.daemonId)}`, { method: "DELETE", headers: { authorization: `Bearer ${input.credential}` }, @@ -136,7 +136,7 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote { handshakeTimeout: this.requestTimeoutMs, headers: { authorization: `Bearer ${input.credential}`, - "x-paseo-relationship-id": input.relationshipId, + "x-paseo-daemon-id": input.daemonId, }, }); let settled = false; diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index 6b439335d1..8b262be4c1 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -11,11 +11,10 @@ import pino from "pino"; import { WebSocket } from "ws"; import type { AgentSnapshotPayload, - HubAgentCreateResponse, - HubAgentStream, - HubAgentUpdate, - HubAuthorizationDenied, - HubExecutionReconcileResponse, + HubExecutionAgentCreateResponse, + HubExecutionAgentStream, + HubExecutionAgentUpdate, + RpcErrorMessage, CreateAgentWorktreeTarget, SessionOutboundMessage, } from "../../messages.js"; @@ -36,7 +35,7 @@ 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 { DaemonExecutions } from "../daemon-executions.js"; import { createAgentCommand, type CreateAgentCommandDependencies, @@ -114,7 +113,7 @@ interface PersistedRelationship { state: string; reason?: string; relationship: { - id: string; + daemonId: string; idempotencyKey?: string; hubOrigin: string; scopes: string[]; @@ -414,7 +413,11 @@ class InMemoryHubRelationships implements HubRelationshipRemote { } private enrollmentResult(input: HubEnrollment): HubEnrollmentResult { - return { relationshipId: input.relationshipId, scopes: ["hub.*"], webSocketUrl: SOCKET_URL }; + return { + daemonId: input.daemonId, + scopes: ["hub.execution.*"], + webSocketUrl: SOCKET_URL, + }; } } @@ -422,7 +425,10 @@ interface CliProcess { result: Promise>; } -type AcceptedCreate = Omit & { +type AcceptedCreate = Omit< + HubExecutionAgentCreateResponse["payload"], + "agentId" | "agent" | "success" +> & { agentId: string; agent: AgentSnapshotPayload; success: true; @@ -450,7 +456,6 @@ export class HubRelationshipHarness { private failNextSessionClose = false; private observedEnrollments = 0; private observedSockets = 0; - private reconcileRequests = 0; private readonly codex = new ControlledAgentClient( createTestAgentClients({ closeSession: async () => { @@ -523,14 +528,14 @@ export class HubRelationshipHarness { const socket = await this.openClaimedCliSocket(`ws://${address}:${target.port}/ws`); const messages = [ { - type: "hub.relationship.connect.request", + type: "hub.management.daemon.connect.request", requestId: "external-hub-connect", hubUrl: HUB_ORIGIN, token: "external-token", }, - { type: "hub.relationship.get_status.request", requestId: "external-hub-status" }, + { type: "hub.management.daemon.get_status.request", requestId: "external-hub-status" }, { - type: "hub.relationship.disconnect.request", + type: "hub.management.daemon.disconnect.request", requestId: "external-hub-disconnect", force: false, }, @@ -554,7 +559,7 @@ export class HubRelationshipHarness { JSON.stringify({ type: "session", message: { - type: "hub.relationship.connect.request", + type: "hub.management.daemon.connect.request", requestId: "browser-hub-connect", hubUrl: HUB_ORIGIN, token: "browser-token", @@ -664,7 +669,7 @@ export class HubRelationshipHarness { ): void { const { prompt = "Create through the Hub", ...requestOptions } = options; this.latestSocket().socket.receive({ - type: "hub.agent.create.request", + type: "hub.execution.agent.create.request", requestId, executionId, provider: "codex", @@ -689,13 +694,13 @@ export class HubRelationshipHarness { async durableOwnedAgentIds(): Promise { return (await this.daemon!.agentStorage.list()) - .filter((record) => record.owner?.kind === "hub") + .filter((record) => record.owner?.kind === "daemon") .map((record) => record.id); } activeOwnedAgentIds(): string[] { return this.daemon!.agentManager.listAgents() - .filter((agent) => agent.owner?.kind === "hub") + .filter((agent) => agent.owner?.kind === "daemon") .map((agent) => agent.id); } @@ -836,8 +841,8 @@ export class HubRelationshipHarness { async ownedPermissionRequest(agentId: string) { const message = await this.latestSocket().socket.messageMatching( - (candidate): candidate is HubAgentStream => - candidate.type === "hub.agent.stream" && + (candidate): candidate is HubExecutionAgentStream => + candidate.type === "hub.execution.agent.stream" && candidate.payload.agentId === agentId && candidate.payload.event.type === "permission_requested", ); @@ -876,39 +881,39 @@ export class HubRelationshipHarness { return { first, duplicate }; } - async ownedUpdate(agentId: string): Promise { + 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, + (candidate): candidate is HubExecutionAgentUpdate => + candidate.type === "hub.execution.agent.update" && candidate.payload.agentId === agentId, ); return message.payload; } - async ownedRunningUpdate(agentId: string): Promise { + async ownedRunningUpdate(agentId: string): Promise { const message = await this.latestSocket().socket.messageMatching( - (candidate): candidate is HubAgentUpdate => - candidate.type === "hub.agent.update" && + (candidate): candidate is HubExecutionAgentUpdate => + candidate.type === "hub.execution.agent.update" && candidate.payload.agentId === agentId && candidate.payload.agent.status === "running", ); return message.payload; } - async ownedTurnCompletion(agentId: string): Promise { + async ownedTurnCompletion(agentId: string): Promise { const socket = this.latestSocket().socket; const completed = await socket.messageMatching( - (candidate): candidate is HubAgentStream => - candidate.type === "hub.agent.stream" && + (candidate): candidate is HubExecutionAgentStream => + candidate.type === "hub.execution.agent.stream" && candidate.payload.agentId === agentId && candidate.payload.event.type === "turn_completed", ); return completed.payload; } - async ownedTurnFailure(agentId: string): Promise { + async ownedTurnFailure(agentId: string): Promise { const failed = await this.latestSocket().socket.messageMatching( - (candidate): candidate is HubAgentStream => - candidate.type === "hub.agent.stream" && + (candidate): candidate is HubExecutionAgentStream => + candidate.type === "hub.execution.agent.stream" && candidate.payload.agentId === agentId && candidate.payload.event.type === "turn_failed", ); @@ -918,34 +923,20 @@ export class HubRelationshipHarness { latestOwnedTurnCompletions(agentId: string): number { return this.latestSocket().socket.sent.filter( (message) => - message.type === "hub.agent.stream" && + message.type === "hub.execution.agent.stream" && message.payload.agentId === agentId && message.payload.event.type === "turn_completed", ).length; } - async ownedStream(agentId: string): Promise { + 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, + (candidate): candidate is HubExecutionAgentStream => + candidate.type === "hub.execution.agent.stream" && candidate.payload.agentId === agentId, ); return message.payload; } - async reconcileOwned( - executionId = "execution-1", - ): Promise { - const requestId = `reconcile-${executionId}-${++this.reconcileRequests}`; - 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 }, @@ -955,7 +946,7 @@ export class HubRelationshipHarness { return agent.id; } - async deniedSteering(agentId: string): Promise { + async deniedSteering(agentId: string): Promise { const requestId = "denied-steer"; this.latestSocket().socket.receive({ type: "send_agent_message_request", @@ -963,11 +954,10 @@ export class HubRelationshipHarness { agentId, text: "This must not run", }); - return ((await this.latestSocket().socket.messageFor(requestId)) as HubAuthorizationDenied) - .payload; + return ((await this.latestSocket().socket.messageFor(requestId)) as RpcErrorMessage).payload; } - async deniedBrowserDispatch(): Promise { + async deniedBrowserDispatch(): Promise { const requestId = "browser-1"; this.latestSocket().socket.receive({ type: "browser.automation.execute.response", @@ -977,15 +967,14 @@ export class HubRelationshipHarness { error: { code: "browser_denied", message: "denied", retryable: false }, }, }); - return ((await this.latestSocket().socket.messageFor(requestId)) as HubAuthorizationDenied) - .payload; + return ((await this.latestSocket().socket.messageFor(requestId)) as RpcErrorMessage).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]; + if (message.type === "hub.execution.agent.update") return [message.payload.agentId]; + if (message.type === "hub.execution.agent.stream") return [message.payload.agentId]; return []; }), ); @@ -1022,7 +1011,8 @@ export class HubRelationshipHarness { const client = await this.trustedClient(); try { await client.patchDaemonConfig({ appendSystemPrompt: "hub-broadcast-isolation" }); - await this.reconcileOwned("broadcast-barrier"); + this.beginOwnedCreate("broadcast-barrier", "broadcast-barrier"); + await this.acceptedCreate("broadcast-barrier"); return this.daemonConfigBroadcasts() - before; } finally { await client.close(); @@ -1038,11 +1028,12 @@ export class HubRelationshipHarness { } } - async reconnectAndReconcile(executionId = "execution-1") { + async reconnectAndRetry(executionId = "execution-1") { this.closeLatestSocket(1006); await this.retry(); this.connectLatestSocket(); - return this.reconcileOwned(executionId); + this.beginOwnedCreate("reconnect-retry", executionId); + return this.acceptedCreate("reconnect-retry"); } async reconstructAndReplay(executionId = "execution-1") { @@ -1057,30 +1048,20 @@ export class HubRelationshipHarness { }); 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", + (record) => record.owner?.kind === "daemon", ).length; - return { replay, reconciliation, durableAgentCount }; + return { replay, durableAgentCount }; } - async removeAndReconcile(agentId: string, executionId = "execution-1") { + async removeOwnedAgent(agentId: string) { 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") + durableAgentCount: (await storage.list()).filter((record) => record.owner?.kind === "daemon") .length, }; } @@ -1343,7 +1324,7 @@ export class HubRelationshipHarness { private async acceptedCreate(requestId: string): Promise { const message = (await this.latestSocket().socket.messageFor( requestId, - )) as HubAgentCreateResponse; + )) as HubExecutionAgentCreateResponse; if (!message.payload.success || !message.payload.agentId || !message.payload.agent) { throw new Error(message.payload.error ?? "Hub agent creation failed"); } @@ -1366,8 +1347,8 @@ export class HubRelationshipHarness { } private executionsForReconstruction(manager: AgentManager, storage: AgentStorage) { - return new RelationshipOwnedExecutions({ - relationshipId: this.relationshipFile()!.relationship.id, + return new DaemonExecutions({ + daemonId: this.relationshipFile()!.relationship.daemonId, agentManager: manager, agentStorage: storage, createAgent: (input) => diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index d85ce7f2b5..16687c6590 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -13,7 +13,7 @@ import { FileTransferOpcode, type FileTransferFrame, } from "@getpaseo/protocol/binary-frames/index"; -import { Session } from "./session.js"; +import { isSessionRpcGranted, Session } from "./session.js"; import { DownloadTokenStore } from "./file-download/token-store.js"; import { StructuredAgentFallbackError } from "./agent/agent-response-loop.js"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; @@ -188,6 +188,7 @@ vi.mock("./worktree-bootstrap.js", async (importOriginal) => { }); interface SessionForTestOptions { + grants?: readonly string[]; agentManager?: { [K in keyof SessionOptions["agentManager"]]?: unknown }; agentStorage?: { [K in keyof SessionOptions["agentStorage"]]?: unknown }; github?: Partial; @@ -248,7 +249,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { }; const messages = options.messages ?? []; - return new Session({ + const sessionOptions: SessionOptions & { grants?: readonly string[] } = { clientId: "test-client", onMessage: (message) => messages.push(message), onBinaryMessage: createBinaryMessageHandler(options.binaryMessages), @@ -304,9 +305,51 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { serverId: options.serverId, daemonVersion: options.daemonVersion, daemonRuntimeConfig: options.daemonRuntimeConfig, - }); + grants: options.grants, + }; + return new Session(sessionOptions); } +describe("session authorization grants", () => { + test("rejects an RPC outside an exact grant with the generic RPC error", async () => { + const messages: SessionOutboundMessage[] = []; + const session = createSessionForTest({ + grants: ["hub.execution.agent.create.request"], + messages, + }); + + await session.handleMessage({ type: "ping", requestId: "restricted-ping", clientSentAt: 42 }); + + expect(messages).toEqual([ + { + type: "rpc_error", + payload: { + requestId: "restricted-ping", + requestType: "ping", + error: "Session is not authorized for ping", + code: "access_denied", + }, + }, + ]); + }); + + test.each([ + ["*", "ping"], + ["hub.execution.*", "hub.execution.agent.create.request"], + ["hub.execution.agent.create.request", "hub.execution.agent.create.request"], + ])("grant %s authorizes %s", (grant, requestType) => { + expect(isSessionRpcGranted([grant], requestType)).toBe(true); + }); + + test.each([ + ["hub.execution.*", "hub.management.daemon.get_status.request"], + ["hub.execution.agent.create.request", "hub.execution.agent.update"], + ["hub.execution.*", "hub.executions.agent.create.request"], + ])("grant %s rejects %s", (grant, requestType) => { + expect(isSessionRpcGranted([grant], requestType)).toBe(false); + }); +}); + describe("file explorer binary responses", () => { const tempDirs: string[] = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index e504a163f4..5007befe40 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -151,6 +151,9 @@ import { AgentConfigSession } from "./session/agent-config/agent-config-session. import { ProjectConfigSession } from "./session/project-config/project-config-session.js"; import { DaemonSession, type DaemonRuntimeConfig } from "./session/daemon/daemon-session.js"; import type { DaemonWebSocketRuntimeDiagnosticSnapshot } from "./session/daemon/diagnostics.js"; +import type { HubRelationshipManagement } from "./hub/relationship-controller.js"; +import { HubExecutionController } from "./hub/execution-controller.js"; +import type { HubExecutionAgents } from "./hub/daemon-executions.js"; import { DownloadTokenStore } from "./file-download/token-store.js"; import { PushTokenStore } from "./push/token-store.js"; import { @@ -407,6 +410,7 @@ type AgentMcpTransportFactory = () => Promise; export interface SessionOptions { clientId: string; + grants?: readonly string[]; appVersion?: string | null; clientCapabilities?: Record | null; onMessage: (msg: SessionOutboundMessage) => void; @@ -442,7 +446,8 @@ export interface SessionOptions { terminalManager: TerminalManager | null; providerSnapshotManager: ProviderSnapshotManager; providerUsageService: ProviderUsageService; - hubRelationships?: import("./hub/relationship-controller.js").HubRelationshipManagement; + hubExecutionAgents?: HubExecutionAgents; + hubRelationships?: HubRelationshipManagement; serviceProxy?: ServiceProxySubsystem; scriptRuntimeStore?: WorkspaceScriptRuntimeStore; workspaceSetupSnapshots?: Map; @@ -506,6 +511,34 @@ function parseClientCapabilities( return new Set(result); } +export function isSessionRpcGranted(grants: readonly string[], rpcName: string): boolean { + return grants.some((grant) => { + if (grant === "*" || grant === rpcName) { + return true; + } + if (!grant.endsWith(".*")) { + return false; + } + return rpcName.startsWith(grant.slice(0, -1)); + }); +} + +function sessionRequestId(message: SessionInboundMessage): string | null { + 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 null; +} + interface AgentTimelineProjectionSelection { timeline: AgentTimelineFetchResult; entries: TimelineProjectionEntry[]; @@ -535,6 +568,7 @@ function describeRegistryTransition(record: ArchivedRecordSnapshot | null): Regi */ export class Session { private readonly clientId: string; + private readonly grants: readonly string[]; private appVersion: string | null; private clientCapabilities: ReadonlySet; private readonly sessionId: string; @@ -593,12 +627,14 @@ export class Session { private readonly agentConfigSession: AgentConfigSession; private readonly projectConfigSession: ProjectConfigSession; private readonly daemonSession: DaemonSession; + private readonly hubExecutionController: HubExecutionController | null; private readonly workspaceScripts: WorkspaceScriptsService; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; constructor(options: SessionOptions) { const { clientId, + grants, appVersion, clientCapabilities, onMessage, @@ -647,6 +683,7 @@ export class Session { getWebSocketRuntimeMetrics, } = options; this.clientId = clientId; + this.grants = grants ?? ["*"]; this.appVersion = appVersion ?? null; this.clientCapabilities = parseClientCapabilities(clientCapabilities); this.sessionId = uuidv4(); @@ -802,6 +839,12 @@ export class Session { logger: this.sessionLogger, hubRelationships: options.hubRelationships, }); + this.hubExecutionController = options.hubExecutionAgents + ? new HubExecutionController({ + agents: options.hubExecutionAgents, + send: (message) => this.emit(message), + }) + : null; this.daemonConfigStore = daemonConfigStore; this.terminalManager = terminalManager; this.terminalController = new TerminalSessionController({ @@ -1362,6 +1405,21 @@ export class Session { }, "agent.session.inbound", ); + if (!isSessionRpcGranted(this.grants, msg.type)) { + const requestId = sessionRequestId(msg); + if (requestId) { + this.emit({ + type: "rpc_error", + payload: { + requestId, + requestType: msg.type, + error: `Session is not authorized for ${msg.type}`, + code: "access_denied", + }, + }); + } + return; + } try { await this.dispatchInboundMessage(msg); } catch (error) { @@ -1407,6 +1465,7 @@ export class Session { this.dispatchAgentRewindMessage(msg) ?? this.dispatchAgentRelationshipMessage(msg) ?? this.dispatchAgentTimelineMessage(msg) ?? + this.dispatchHubExecutionMessage(msg) ?? this.dispatchAgentLifecycleMessage(msg) ?? this.dispatchAgentConfigMessage(msg) ?? this.dispatchCheckoutMessage(msg) ?? @@ -1502,6 +1561,12 @@ export class Session { } } + private dispatchHubExecutionMessage(msg: SessionInboundMessage): Promise | undefined { + return msg.type === "hub.execution.agent.create.request" + ? this.hubExecutionController?.createAgent(msg) + : undefined; + } + private dispatchAgentLifecycleMessage(msg: SessionInboundMessage): Promise | undefined { switch (msg.type) { case "fetch_agents_request": @@ -1565,9 +1630,9 @@ 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": + case "hub.management.daemon.connect.request": + case "hub.management.daemon.get_status.request": + case "hub.management.daemon.disconnect.request": return this.daemonSession.handleHubRelationshipRequest(msg); case "diagnostics.request": return this.daemonSession.handleDiagnosticsRequest(msg); @@ -5785,6 +5850,9 @@ export class Session { * Emit a message to the client */ private emit(msg: SessionOutboundMessage): void { + if (msg.type !== "rpc_error" && !isSessionRpcGranted(this.grants, msg.type)) { + return; + } // JSON.stringify(msg) is only computed when trace is enabled — it runs for // every outbound message otherwise, and trace is disabled by default. // Optional-chained because test logger stubs don't implement isLevelEnabled. @@ -5822,6 +5890,7 @@ export class Session { this.unsubscribeAgentEvents = null; } this.agentUpdates.dispose(); + this.hubExecutionController?.cleanup(); if (this.unsubscribeTerminalWorkspaceContributionEvents) { this.unsubscribeTerminalWorkspaceContributionEvents(); this.unsubscribeTerminalWorkspaceContributionEvents = null; diff --git a/packages/server/src/server/session/daemon/daemon-session.test.ts b/packages/server/src/server/session/daemon/daemon-session.test.ts index f025480035..661439bad4 100644 --- a/packages/server/src/server/session/daemon/daemon-session.test.ts +++ b/packages/server/src/server/session/daemon/daemon-session.test.ts @@ -77,7 +77,7 @@ describe("DaemonSession", () => { }, status: () => ({ state: "not_connected", - relationshipId: null, + daemonId: null, hubOrigin: null, scopes: [], connectedAt: null, @@ -90,13 +90,13 @@ describe("DaemonSession", () => { }); await subsystem.handleHubRelationshipRequest({ - type: "hub.relationship.connect.request", + type: "hub.management.daemon.connect.request", requestId: "connect-1", hubUrl: "https://hub.test", token: "token", }); await subsystem.handleHubRelationshipRequest({ - type: "hub.relationship.disconnect.request", + type: "hub.management.daemon.disconnect.request", requestId: "disconnect-1", force: false, }); @@ -106,7 +106,7 @@ describe("DaemonSession", () => { type: "rpc_error", payload: { requestId: "connect-1", - requestType: "hub.relationship.connect.request", + requestType: "hub.management.daemon.connect.request", error: "Hub rejected enrollment (401)", code: "handler_error", }, @@ -115,7 +115,7 @@ describe("DaemonSession", () => { type: "rpc_error", payload: { requestId: "disconnect-1", - requestType: "hub.relationship.disconnect.request", + requestType: "hub.management.daemon.disconnect.request", error: "Hub revocation failed (503)", code: "handler_error", }, diff --git a/packages/server/src/server/session/daemon/daemon-session.ts b/packages/server/src/server/session/daemon/daemon-session.ts index afd72b7f6e..ef7e627d29 100644 --- a/packages/server/src/server/session/daemon/daemon-session.ts +++ b/packages/server/src/server/session/daemon/daemon-session.ts @@ -101,35 +101,35 @@ export class DaemonSession { SessionInboundMessage, { type: - | "hub.relationship.connect.request" - | "hub.relationship.get_status.request" - | "hub.relationship.disconnect.request"; + | "hub.management.daemon.connect.request" + | "hub.management.daemon.get_status.request" + | "hub.management.daemon.disconnect.request"; } >, ): Promise { try { if (!this.hubRelationships) throw new Error("Hub relationship management is unavailable"); - if (msg.type === "hub.relationship.connect.request") { + if (msg.type === "hub.management.daemon.connect.request") { const status = await this.hubRelationships.connect({ hubUrl: msg.hubUrl, token: msg.token, }); this.host.emit({ - type: "hub.relationship.connect.response", + type: "hub.management.daemon.connect.response", payload: { requestId: msg.requestId, status }, }); return; } - if (msg.type === "hub.relationship.disconnect.request") { + if (msg.type === "hub.management.daemon.disconnect.request") { const result = await this.hubRelationships.disconnect({ force: msg.force ?? false }); this.host.emit({ - type: "hub.relationship.disconnect.response", + type: "hub.management.daemon.disconnect.response", payload: { requestId: msg.requestId, ...result }, }); return; } this.host.emit({ - type: "hub.relationship.get_status.response", + type: "hub.management.daemon.get_status.response", payload: { requestId: msg.requestId, status: this.hubRelationships.status() }, }); } catch (error) { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 2d6d01bb12..f2b112f001 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -17,6 +17,7 @@ import type { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-m import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js"; import { type ServerInfoStatusPayload, + type SessionOutboundMessage, type WorkspaceSetupSnapshot, type WSHelloMessage, type WSInboundMessage, @@ -31,9 +32,8 @@ 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 { HubSession } from "./hub/hub-session.js"; import type { HubRelationshipManagement } from "./hub/relationship-controller.js"; -import type { HubExecutions } from "./hub/relationship-owned-executions.js"; +import type { HubExecutionAgents } from "./hub/daemon-executions.js"; import type { AgentProvider } from "./agent/agent-sdk-types.js"; import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js"; @@ -347,15 +347,15 @@ interface TrustedSessionConnection { externalDisconnectCleanupTimeout: ReturnType | null; } -interface HubSessionConnection { +interface HubConnection { kind: "hub"; - session: HubSession; - relationshipId: string; + session: Session; + daemonId: string; connectionLogger: pino.Logger; socket: WebSocketLike; } -type SessionConnection = TrustedSessionConnection | HubSessionConnection; +type SessionConnection = TrustedSessionConnection | HubConnection; type TrustedLifecycleKey = | "clientId" @@ -363,9 +363,8 @@ type TrustedLifecycleKey = | "clientCapabilities" | "sockets" | "externalDisconnectCleanupTimeout"; -type HubTrustedLifecycleOverlap = Extract; -const HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE: HubTrustedLifecycleOverlap extends never ? true : never = - true; +type HubLifecycleOverlap = Extract; +const HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE: HubLifecycleOverlap extends never ? true : never = true; void HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE; interface BrowserToolsRegistration { @@ -373,6 +372,20 @@ interface BrowserToolsRegistration { unregister: () => void; } +interface SocketSessionOptions { + clientId: string; + appVersion: string | null; + clientCapabilities: Record | null; + grants: readonly string[]; + connectionLogger: pino.Logger; + onMessage: (message: SessionOutboundMessage) => void; + onBinaryMessage?: (frame: Uint8Array) => void; + getTransportBufferedAmount?: () => number | null; + onLifecycleIntent?: (intent: SessionLifecycleIntent) => void; + hubExecutionAgents?: HubExecutionAgents; + hubRelationships?: HubRelationshipManagement; +} + const SLOW_REQUEST_THRESHOLD_MS = 500; const EXTERNAL_SESSION_DISCONNECT_GRACE_MS = 90_000; const HELLO_TIMEOUT_MS = 15_000; @@ -845,8 +858,9 @@ export class VoiceAssistantWebSocketServer { public async attachHubSocket( ws: WebSocketLike, options: { - relationshipId: string; - executions: HubExecutions; + daemonId: string; + grants: readonly string[]; + agents: HubExecutionAgents; }, ): Promise { if (!this.acceptingConnections) { @@ -856,16 +870,21 @@ export class VoiceAssistantWebSocketServer { const connectionLogger = this.logger.child({ connectionKind: "hub", - relationshipId: options.relationshipId, + daemonId: options.daemonId, }); - const session = new HubSession({ - executions: options.executions, - send: (message) => this.sendToClient(ws, wrapSessionMessage(message)), + const session = this.createSocketSession({ + clientId: `hub:${options.daemonId}`, + appVersion: null, + clientCapabilities: null, + grants: options.grants, + connectionLogger, + onMessage: (message) => this.sendToClient(ws, wrapSessionMessage(message)), + hubExecutionAgents: options.agents, }); - const connection: HubSessionConnection = { + const connection: HubConnection = { kind: "hub", session, - relationshipId: options.relationshipId, + daemonId: options.daemonId, connectionLogger, socket: ws, }; @@ -1067,10 +1086,12 @@ export class VoiceAssistantWebSocketServer { const { ws, clientId, appVersion, clientCapabilities, connectionLogger } = params; let connection: TrustedSessionConnection | null = null; - const session = new Session({ + const session = this.createSocketSession({ clientId, appVersion, clientCapabilities, + grants: ["*"], + connectionLogger, onMessage: (msg) => { if (!connection) { return; @@ -1102,7 +1123,33 @@ export class VoiceAssistantWebSocketServer { onLifecycleIntent: (intent) => { this.onLifecycleIntent?.(intent); }, - logger: connectionLogger.child({ module: "session" }), + hubRelationships: this.hubRelationships ?? undefined, + }); + + connection = { + kind: "trusted", + session, + clientId, + appVersion, + clientCapabilities, + connectionLogger, + sockets: new Set([ws]), + externalDisconnectCleanupTimeout: null, + }; + return connection; + } + + private createSocketSession(options: SocketSessionOptions): Session { + return new Session({ + clientId: options.clientId, + appVersion: options.appVersion, + clientCapabilities: options.clientCapabilities, + grants: options.grants, + onMessage: options.onMessage, + onBinaryMessage: options.onBinaryMessage, + getTransportBufferedAmount: options.getTransportBufferedAmount, + onLifecycleIntent: options.onLifecycleIntent, + logger: options.connectionLogger.child({ module: "session" }), downloadTokenStore: this.downloadTokenStore, pushTokenStore: this.pushTokenStore, paseoHome: this.paseoHome, @@ -1126,6 +1173,8 @@ export class VoiceAssistantWebSocketServer { terminalManager: this.terminalManager, providerSnapshotManager: this.providerSnapshotManager, providerUsageService: this.providerUsageService, + hubExecutionAgents: options.hubExecutionAgents, + hubRelationships: options.hubRelationships, serviceProxy: this.serviceProxy ?? undefined, scriptRuntimeStore: this.scriptRuntimeStore ?? undefined, workspaceSetupSnapshots: this.workspaceSetupSnapshots, @@ -1164,20 +1213,7 @@ export class VoiceAssistantWebSocketServer { daemonVersion: this.daemonVersion, daemonRuntimeConfig: this.daemonRuntimeConfig, getWebSocketRuntimeMetrics: () => this.lastRuntimeMetricsSnapshot, - hubRelationships: this.hubRelationships ?? undefined, }); - - connection = { - kind: "trusted", - session, - clientId, - appVersion, - clientCapabilities, - connectionLogger, - sockets: new Set([ws]), - externalDisconnectCleanupTimeout: null, - }; - return connection; } private clearPendingConnection(ws: WebSocketLike): PendingConnection | null { @@ -1825,7 +1861,7 @@ export class VoiceAssistantWebSocketServer { } else if (activeConnection.kind === "trusted") { connectionFields = { clientId: activeConnection.clientId }; } else { - connectionFields = { relationshipId: activeConnection.relationshipId }; + connectionFields = { daemonId: activeConnection.daemonId }; } activeConnection.connectionLogger.warn( {