puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-07 08:54:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(hub): authorize execution through sessions

Browse Source
This commit is contained in:
Mohamed Boudra
2026-07-13 18:41:35 +02:00
parent dfcaa9bf89
commit 16c7fe9709
24 changed files with 647 additions and 897 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/hub.md
+27 -22
View File
@@ -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.
packages/cli/src/commands/hub/index.ts
+4 -4
View File
@@ -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<HubRow> = {
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<HubRow> = {
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,
packages/client/src/daemon-client.ts
+6 -6
View File
@@ -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",
});
}
packages/protocol/src/messages.hub.test.ts
+27 -49
View File
@@ -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",
},
packages/protocol/src/messages.ts
+43 -76
View File
@@ -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<typeof HubAgentCreateRequestSchema>;
export type HubExecutionReconcileRequest = z.infer<typeof HubExecutionReconcileRequestSchema>;
export type HubExecutionAgentCreateRequest = z.infer<typeof HubExecutionAgentCreateRequestSchema>;
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<typeof HubExecutionAgentCreateResponseSchema>;
export type HubExecutionAgentUpdate = z.infer<typeof HubExecutionAgentUpdateSchema>;
export type HubExecutionAgentStream = z.infer<typeof HubExecutionAgentStreamSchema>;
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<typeof HubAgentCreateResponseSchema>;
export type HubAgentUpdate = z.infer<typeof HubAgentUpdateSchema>;
export type HubAgentStream = z.infer<typeof HubAgentStreamSchema>;
export type HubExecutionReconcileResponse = z.infer<typeof HubExecutionReconcileResponseSchema>;
export type HubAuthorizationDenied = z.infer<typeof HubAuthorizationDeniedSchema>;
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<typeof HubSessionOutboundMessageSchema>;
export type HubExecutionOutboundMessage = z.infer<typeof HubExecutionOutboundMessageSchema>;
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<typeof DaemonUpdateProgressMessageSchema>;
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,
packages/server/src/server/agent/agent-owner.ts
+5 -5
View File
@@ -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<typeof AgentOwnerSchema>;
export type HubAgentOwner = Extract<AgentOwner, { kind: "hub" }>;
export type DaemonAgentOwner = Extract<AgentOwner, { kind: "daemon" }>;
export function hubExecutionKey(owner: HubAgentOwner): string {
return `${owner.relationshipId}\0${owner.executionId}`;
export function daemonExecutionKey(owner: DaemonAgentOwner): string {
return `${owner.daemonId}\0${owner.executionId}`;
}
packages/server/src/server/agent/agent-storage.ts
+19 -19
View File
@@ -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<string, Set<string>> = new Map();
private pendingWrites: Map<string, Promise<void>> = new Map();
private deleting: Set<string> = new Set();
private hubAgentIdsByExecution: Map<string, string> = new Map();
private hubExecutionKeysByAgentId: Map<string, string> = new Map();
private daemonAgentIdsByExecution: Map<string, string> = new Map();
private daemonExecutionKeysByAgentId: Map<string, string> = new Map();
private loaded = false;
private baseDir: string;
private loadPromise: Promise<StoredAgentRecord[]> | null = null;
@@ -116,9 +116,9 @@ export class AgentStorage {
return this.cache.get(agentId) ?? null;
}
async findByHubExecution(owner: HubAgentOwner): Promise<StoredAgentRecord | null> {
async findByDaemonExecution(owner: DaemonAgentOwner): Promise<StoredAgentRecord | null> {
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<StoredAgentRecord[]> {
const records: StoredAgentRecord[] = [];
let entries: Array<import("node:fs").Dirent> = [];
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<void> {
packages/server/src/server/bootstrap.ts
+4 -4
View File
@@ -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,
packages/server/src/server/hub/relationship-owned-executions.test.ts → packages/server/src/server/hub/daemon-executions.test.ts
+8 -11
View File
@@ -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 });
packages/server/src/server/hub/relationship-owned-executions.ts → packages/server/src/server/hub/daemon-executions.ts
+23 -34
View File
@@ -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<void>;
}
export interface HubExecutions {
create(input: HubAgentCreateInput): Promise<OwnedAgentSnapshot>;
reconcile(executionId: string): Promise<OwnedAgentSnapshot | null>;
export interface HubExecutionAgents {
create(input: HubExecutionAgentCreateInput): Promise<OwnedAgentSnapshot>;
subscribe(listener: (event: OwnedAgentEvent) => void): () => void;
invalidateAuthority(): Promise<void>;
}
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<DaemonExecutionsOptions["cleanupFailedCreate"]>;
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<OwnedAgentSnapshot> {
create(input: HubExecutionAgentCreateInput): Promise<OwnedAgentSnapshot> {
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<OwnedAgentSnapshot | null> {
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<void> {
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<OwnedAgentSnapshot> {
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;
}
packages/server/src/server/hub/execution-controller.ts
+105
View File
@@ -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<void> {
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`);
}
}
packages/server/src/server/hub/hub-session.websocket.test.ts → packages/server/src/server/hub/execution-session.websocket.test.ts
+7 -12
View File
@@ -19,13 +19,12 @@ async function launchRelationship(): Promise<HubRelationshipHarness> {
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());
packages/server/src/server/hub/hub-session.test.ts
-259
View File
@@ -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<OwnedAgentSnapshot> {
this.creates++;
return { executionId: "execution-1", agent: createAgentSnapshot() };
}
async reconcile(_executionId: string): Promise<OwnedAgentSnapshot | null> {
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<void> {}
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<void> {
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<void> {
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<void> {
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: {},
});
}
packages/server/src/server/hub/hub-session.ts
-163
View File
@@ -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<void> {
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<void> {
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<void> {
let result: Awaited<ReturnType<HubExecutions["reconcile"]>> = 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;
}
packages/server/src/server/hub/relationship-controller.test.ts
+50 -44
View File
@@ -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");
packages/server/src/server/hub/relationship-controller.ts
+26 -24
View File
@@ -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<void>;
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<Promise<void>>();
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;
packages/server/src/server/hub/relationship-remote.test.ts
+25 -26
View File
@@ -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<string> {
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;
packages/server/src/server/hub/relationship-remote.ts
+9 -9
View File
@@ -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<HubEnrollmentResult> {
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<void> {
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;
packages/server/src/server/hub/test-utils/relationship-harness.ts
+58 -77
View File
@@ -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<Record<string, unknown>>;
}
type AcceptedCreate = Omit<HubAgentCreateResponse["payload"], "agentId" | "agent" | "success"> & {
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<string[]> {
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<HubAgentUpdate["payload"]> {
async ownedUpdate(agentId: string): Promise<HubExecutionAgentUpdate["payload"]> {
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<HubAgentUpdate["payload"]> {
async ownedRunningUpdate(agentId: string): Promise<HubExecutionAgentUpdate["payload"]> {
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<HubAgentStream["payload"]> {
async ownedTurnCompletion(agentId: string): Promise<HubExecutionAgentStream["payload"]> {
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<HubAgentStream["payload"]> {
async ownedTurnFailure(agentId: string): Promise<HubExecutionAgentStream["payload"]> {
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<HubAgentStream["payload"]> {
async ownedStream(agentId: string): Promise<HubExecutionAgentStream["payload"]> {
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<HubExecutionReconcileResponse["payload"]> {
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<string> {
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<HubAuthorizationDenied["payload"]> {
async deniedSteering(agentId: string): Promise<RpcErrorMessage["payload"]> {
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<HubAuthorizationDenied["payload"]> {
async deniedBrowserDispatch(): Promise<RpcErrorMessage["payload"]> {
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<AcceptedCreate> {
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) =>
packages/server/src/server/session.test.ts
+46 -3
View File
@@ -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<GitHubService>;
@@ -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[] = [];
packages/server/src/server/session.ts
+73 -4
View File
@@ -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<unknown>;
export interface SessionOptions {
clientId: string;
grants?: readonly string[];
appVersion?: string | null;
clientCapabilities?: Record<string, unknown> | 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<string, WorkspaceSetupSnapshot>;
@@ -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<ClientCapability>;
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<void> | undefined {
return msg.type === "hub.execution.agent.create.request"
? this.hubExecutionController?.createAgent(msg)
: undefined;
}
private dispatchAgentLifecycleMessage(msg: SessionInboundMessage): Promise<void> | 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;
packages/server/src/server/session/daemon/daemon-session.test.ts
+5 -5
View File
@@ -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",
},
packages/server/src/server/session/daemon/daemon-session.ts
+8 -8
View File
@@ -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<void> {
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) {
packages/server/src/server/websocket-server.ts
+69 -33
View File
@@ -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<typeof setTimeout> | 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<keyof HubSessionConnection, TrustedLifecycleKey>;
const HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE: HubTrustedLifecycleOverlap extends never ? true : never =
true;
type HubLifecycleOverlap = Extract<keyof HubConnection, TrustedLifecycleKey>;
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<string, unknown> | 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<void> {
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(
{