mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-11 10:57:07 +00:00
Connect saved hosts and delegate agents across daemons (#7370)
* feat(hosts): connect to saved hosts through a daemon * fix(hosts): validate safe targets and verify encrypted relay routes * test(hosts): include relay coverage in existing CI contracts * fix(cli): preserve validated addresses in daemon command hints * fix(hosts): share daemon links across isolated routed sessions Multiplex logical sessions over the monitored daemon connection and wrap original binary bytes in routing headers. Admit channels under their authenticated parent and current grants; preserve scoped delivery and physical send completion. * fix(hosts): bound encrypted overload and initial acquisition Check encrypted wire capacity before the async channel converts a resource limit into a physical connection error. Bound initial routed acquisition with the existing client timeout, then enable reconnect after admission. * fix(hosts): keep compact row actions clear of host labels Give compact connected-host rows separate text and action widths so long names and producer labels cannot compete with primary controls. * refactor(hosts): remove management settings UI from phase one Keep connected-host management in the CLI and shared SDK while preserving app saved connections and plugin access. * test(hosts): control physical write backpressure across platforms Hold real TCP writes until assertions finish instead of assuming a paused receiver exhausts every platform receive buffer. Preserve physical send settlement, byte identity, capacity and sibling checks. * refactor(hosts): use the ordinary client contract over channel transports Keep queuing, permission codes and reconnection independent of transport selection. Track submitted request outcomes generically while retaining connection error codes and structured causes. * test(e2e): keep transient developer toasts from stalling checks Concurrent locator handlers can wait for a dismiss button another handler removed. Dismiss existing toast DOM at document lifetime, including its open shadow root, without intercepting app actions. * Create, prompt, and monitor agents across connected hosts (#7507) * feat(hosts): delegate agents across connected hosts Resolve MCP host operations through in-process commands for self and the existing routed client transport for other destinations. Keep ordinary parent labels, app sessions and idle finish notifications. * test(app): normalize subagent fixtures through snapshots * docs(protocol): clarify same-host parent labels * fix(hosts): resolve subagents through destination sessions A child address identifies its destination regardless of the connection used. Reuse the existing host candidate selection for discovered routes and direct connections, and keep navigation and row actions independent of the parent transport. * fix(hosts): keep routed destinations distinct during discovery A routed candidate names only the origin daemon. Match it within the named destination profile so two destinations discovered through the same origin retain separate identities and connections, including existing saved routes. * feat(hosts): prompt and query agents on selected hosts Extend the existing MCP host operation boundary with prompt, status, activity and list operations. Local calls retain their handlers and notification owner; selected destinations use ordinary client RPCs and the existing idle finish wait. * test(e2e): preserve toast coverage with CSS hiding * docs(hosts): translate connected host references
This commit is contained in:
Mohamed Boudra
@@ -5,7 +5,7 @@ import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
|
||||
import { isLikelyExternalToolName } from "@getpaseo/protocol/tool-name-normalization";
|
||||
import { buildAgentMessageDisplay } from "@getpaseo/protocol/paseo-tool-presentation/index";
|
||||
import { buildToolCallDisplayModel } from "@getpaseo/protocol/tool-call-display";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import { projectTimelineRows, selectItemsByProjectedLimit } from "./timeline-projection.js";
|
||||
|
||||
const DEFAULT_MAX_ITEMS = 0;
|
||||
const MAX_TOOL_INPUT_CHARS = 400;
|
||||
@@ -210,6 +210,24 @@ export function curateAgentActivity(
|
||||
: "No activity to display.";
|
||||
}
|
||||
|
||||
export function summarizeAgentActivity(timeline: AgentTimelineItem[], limit?: number) {
|
||||
const selection = selectItemsByProjectedLimit({
|
||||
items: timeline,
|
||||
direction: "tail",
|
||||
limit: limit ?? 0,
|
||||
});
|
||||
const { totalProjected, shownProjected } = selection;
|
||||
const noun = totalProjected === 1 ? "activity" : "activities";
|
||||
const countHeader =
|
||||
limit && shownProjected < totalProjected
|
||||
? `Showing ${shownProjected} of ${totalProjected} ${noun} (limited to ${limit})`
|
||||
: `Showing all ${totalProjected} ${noun}`;
|
||||
return {
|
||||
updateCount: timeline.length,
|
||||
content: `${countHeader}\n\n${curateAgentActivity(selection.items)}`,
|
||||
};
|
||||
}
|
||||
|
||||
interface ForkCursorBoundary {
|
||||
timelineEpoch: string;
|
||||
cursor: { epoch: string; seq: number };
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { isSameOrDescendantPath } from "../path-utils.js";
|
||||
import type {
|
||||
AgentListItemPayload,
|
||||
AgentSnapshotPayload,
|
||||
@@ -286,6 +287,66 @@ export function toAgentListItemPayload(agent: AgentSnapshotPayload): AgentListIt
|
||||
};
|
||||
}
|
||||
|
||||
function parseTimestamp(value: string | null | undefined): number {
|
||||
if (!value) {
|
||||
return 0;
|
||||
}
|
||||
const parsed = Date.parse(value);
|
||||
return Number.isNaN(parsed) ? 0 : parsed;
|
||||
}
|
||||
|
||||
function resolveAgentListActivityTime(agent: AgentListItemPayload): number {
|
||||
return Math.max(
|
||||
parseTimestamp(agent.updatedAt),
|
||||
parseTimestamp(agent.lastUserMessageAt),
|
||||
parseTimestamp(agent.attentionTimestamp),
|
||||
parseTimestamp(agent.archivedAt),
|
||||
parseTimestamp(agent.createdAt),
|
||||
);
|
||||
}
|
||||
|
||||
function compareAgentListItems(a: AgentListItemPayload, b: AgentListItemPayload): number {
|
||||
const attentionDelta =
|
||||
Number(b.requiresAttention ?? false) - Number(a.requiresAttention ?? false);
|
||||
if (attentionDelta !== 0) {
|
||||
return attentionDelta;
|
||||
}
|
||||
|
||||
const statusOrder = {
|
||||
running: 0,
|
||||
initializing: 1,
|
||||
idle: 2,
|
||||
error: 3,
|
||||
closed: 4,
|
||||
} as Record<string, number>;
|
||||
const statusDelta = (statusOrder[a.status] ?? 999) - (statusOrder[b.status] ?? 999);
|
||||
if (statusDelta !== 0) {
|
||||
return statusDelta;
|
||||
}
|
||||
|
||||
return resolveAgentListActivityTime(b) - resolveAgentListActivityTime(a);
|
||||
}
|
||||
|
||||
export function selectAgentListItems(
|
||||
agents: readonly AgentSnapshotPayload[],
|
||||
options: {
|
||||
cwd?: string;
|
||||
statuses?: AgentSnapshotPayload["status"][];
|
||||
sinceHours?: number;
|
||||
limit?: number;
|
||||
} = {},
|
||||
): AgentListItemPayload[] {
|
||||
const statusFilter = options.statuses?.length ? new Set(options.statuses) : null;
|
||||
const sinceMs = Date.now() - (options.sinceHours ?? 48) * 60 * 60 * 1000;
|
||||
return agents
|
||||
.map(toAgentListItemPayload)
|
||||
.filter((agent) => !options.cwd || isSameOrDescendantPath(options.cwd, agent.cwd))
|
||||
.filter((agent) => !statusFilter || statusFilter.has(agent.status))
|
||||
.filter((agent) => !agent.archivedAt || resolveAgentListActivityTime(agent) >= sinceMs)
|
||||
.sort(compareAgentListItems)
|
||||
.slice(0, options.limit ?? 50);
|
||||
}
|
||||
|
||||
export function toRecentProviderSessionDescriptorPayload(
|
||||
session: ImportableProviderSession & { provider: string },
|
||||
options: RecentProviderSessionProjectionOptions,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client";
|
||||
import { prepareAgentMessage, type AgentPromptSource } from "./agent-messages/index.js";
|
||||
import type { Logger } from "pino";
|
||||
|
||||
@@ -10,7 +11,10 @@ import type { AgentManager, ManagedAgent } from "./agent-manager.js";
|
||||
import type { AgentStorage } from "./agent-storage.js";
|
||||
import { ensureAgentLoaded } from "./agent-loading.js";
|
||||
import { isStaleProviderSessionError } from "./stale-provider-session-error.js";
|
||||
import { getParentAgentIdFromLabels } from "@getpaseo/protocol/agent-labels";
|
||||
import {
|
||||
getParentAgentIdFromLabels,
|
||||
getParentAddressFromLabels,
|
||||
} from "@getpaseo/protocol/agent-labels";
|
||||
import type { ActiveTurnBehavior } from "@getpaseo/protocol/messages";
|
||||
|
||||
export type AgentUnarchiveController = Pick<AgentManager, "notifyAgentState" | "unarchiveSnapshot">;
|
||||
@@ -292,7 +296,7 @@ async function resolvePromptSource(
|
||||
manager: Pick<AgentManager, "getAgent">,
|
||||
storage: AgentStorage,
|
||||
): Promise<AgentPromptSource | undefined> {
|
||||
if (!source) return undefined;
|
||||
if (!source || source.title) return source;
|
||||
const title = (
|
||||
manager.getAgent(source.agentId)?.config.title ?? (await storage.get(source.agentId))?.title
|
||||
)?.trim();
|
||||
@@ -401,6 +405,7 @@ export interface SetupFinishNotificationParams {
|
||||
agentStorage: AgentStorage;
|
||||
childAgentId: string;
|
||||
callerAgentId: string;
|
||||
childServerId?: string;
|
||||
requireParentOwnership?: boolean;
|
||||
/** Retain pre-dispatch outcomes only when dispatch keeps the existing turn. */
|
||||
dispatchResult?: Promise<PromptDispatchDisposition | false>;
|
||||
@@ -429,6 +434,7 @@ const FINISH_NOTIFICATION_MESSAGE_LIMIT = 4000;
|
||||
|
||||
interface FinishNotificationBodyInput {
|
||||
childAgentId: string;
|
||||
serverId?: string;
|
||||
title: string;
|
||||
reason: FinishNotificationReason;
|
||||
lastAssistantMessage: string | null;
|
||||
@@ -438,12 +444,14 @@ interface FinishNotificationBodyInput {
|
||||
function formatFinishNotificationBody(params: FinishNotificationBodyInput): string {
|
||||
const statusLine = `Agent ${params.childAgentId} (${params.title}) ${params.reason}.`;
|
||||
const sections = [statusLine];
|
||||
if (params.serverId) sections.push(`Server: ${params.serverId}`);
|
||||
if (params.reason === "needs permission" && params.permissionRequest) {
|
||||
sections.push(
|
||||
"Respond with `respond_to_permission` using the `agentId` and `requestId` below.",
|
||||
`<permission-request>\n${JSON.stringify(
|
||||
{
|
||||
agentId: params.childAgentId,
|
||||
...(params.serverId ? { serverId: params.serverId } : {}),
|
||||
requestId: params.permissionRequest.id,
|
||||
request: params.permissionRequest,
|
||||
},
|
||||
@@ -480,6 +488,101 @@ const armedFinishNotifications = new WeakMap<
|
||||
Map<string, Set<ArmedFinishNotification>>
|
||||
>();
|
||||
|
||||
/** Creation waits and accepted follow-ups share notification ownership. */
|
||||
export function setupFinishWait(params: {
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
originServerId?: string;
|
||||
childServerId?: string;
|
||||
childAgentId: string;
|
||||
callerAgentId: string;
|
||||
wait(signal: AbortSignal): Promise<WaitForFinishResult>;
|
||||
afterPermission(requestId: string, signal: AbortSignal): Promise<void>;
|
||||
logger: Logger;
|
||||
}): () => void {
|
||||
const controller = new AbortController();
|
||||
const byManager =
|
||||
armedFinishNotifications.get(params.agentManager) ??
|
||||
new Map<string, Set<ArmedFinishNotification>>();
|
||||
armedFinishNotifications.set(params.agentManager, byManager);
|
||||
const key = JSON.stringify([
|
||||
params.childServerId ?? null,
|
||||
params.childAgentId,
|
||||
params.callerAgentId,
|
||||
]);
|
||||
const registrations = byManager.get(key) ?? new Set<ArmedFinishNotification>();
|
||||
byManager.set(key, registrations);
|
||||
for (const previous of registrations) previous.stop();
|
||||
byManager.set(key, registrations);
|
||||
let superseded = Promise.resolve(false);
|
||||
const stop = () => {
|
||||
controller.abort();
|
||||
registrations.delete(registration);
|
||||
if (registrations.size === 0 && byManager.get(key) === registrations) byManager.delete(key);
|
||||
};
|
||||
const registration: ArmedFinishNotification = {
|
||||
stop,
|
||||
supersedeWhen(accepted) {
|
||||
superseded = superseded.then(async (prior) => prior || (await accepted));
|
||||
},
|
||||
};
|
||||
registrations.add(registration);
|
||||
async function watch(): Promise<void> {
|
||||
while (!controller.signal.aborted) {
|
||||
const result = await params.wait(controller.signal);
|
||||
if (controller.signal.aborted || (await superseded) || result.status === "timeout") return;
|
||||
const parent = getParentAddressFromLabels(
|
||||
result.final?.labels,
|
||||
params.childServerId ?? "self",
|
||||
);
|
||||
if (
|
||||
parent?.agentId !== params.callerAgentId ||
|
||||
parent.serverId !== (params.originServerId ?? "self")
|
||||
)
|
||||
return;
|
||||
if ((await params.agentStorage.get(params.callerAgentId))?.archivedAt) return;
|
||||
const reason = { idle: "finished", error: "errored", permission: "needs permission" }[
|
||||
result.status
|
||||
] as FinishNotificationReason;
|
||||
const permissionRequest = result.final?.pendingPermissions[0];
|
||||
await sendPromptToAgent({
|
||||
agentManager: params.agentManager,
|
||||
agentStorage: params.agentStorage,
|
||||
agentId: params.callerAgentId,
|
||||
prompt: formatFinishNotificationBody({
|
||||
childAgentId: params.childAgentId,
|
||||
serverId: params.childServerId,
|
||||
title: result.final?.title ?? params.childAgentId,
|
||||
reason,
|
||||
lastAssistantMessage: result.lastMessage,
|
||||
permissionRequest,
|
||||
}),
|
||||
source: {
|
||||
kind: "agent-notification",
|
||||
agentId: params.childAgentId,
|
||||
title: result.final?.title ?? params.childAgentId,
|
||||
event: finishNotificationEvents[reason],
|
||||
},
|
||||
activeTurnBehavior: "steer",
|
||||
unarchive: false,
|
||||
logger: params.logger,
|
||||
});
|
||||
if (result.status !== "permission" || !permissionRequest) return;
|
||||
await params.afterPermission(permissionRequest.id, controller.signal);
|
||||
}
|
||||
}
|
||||
void watch()
|
||||
.catch((error) => {
|
||||
if (!controller.signal.aborted)
|
||||
params.logger.warn(
|
||||
{ err: error, agentId: params.childAgentId },
|
||||
"Child finish wait stopped",
|
||||
);
|
||||
})
|
||||
.finally(stop);
|
||||
return stop;
|
||||
}
|
||||
|
||||
export function setupFinishNotification(params: SetupFinishNotificationParams): () => void {
|
||||
const {
|
||||
agentManager,
|
||||
@@ -499,7 +602,7 @@ export function setupFinishNotification(params: SetupFinishNotificationParams):
|
||||
|
||||
const armedByManager = armedFinishNotifications.get(agentManager) ?? new Map();
|
||||
armedFinishNotifications.set(agentManager, armedByManager);
|
||||
const armedKey = JSON.stringify([childAgentId, callerAgentId]);
|
||||
const armedKey = JSON.stringify([params.childServerId ?? null, childAgentId, callerAgentId]);
|
||||
const registrations = armedByManager.get(armedKey) ?? new Set<ArmedFinishNotification>();
|
||||
armedByManager.set(armedKey, registrations);
|
||||
const previous = [...registrations];
|
||||
|
||||
@@ -0,0 +1,393 @@
|
||||
import { parentAddressLabels } from "@getpaseo/protocol/agent-labels";
|
||||
import type { ActiveTurnBehavior, AgentSnapshotPayload } from "@getpaseo/protocol/messages";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { z } from "zod";
|
||||
import {
|
||||
DaemonClient,
|
||||
type FetchWorkspacesEntry,
|
||||
type WaitForFinishResult,
|
||||
} from "@getpaseo/client/internal/daemon-client";
|
||||
import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types";
|
||||
import type { HostDirectory } from "@getpaseo/protocol/hosts";
|
||||
import { OWNER_PERMISSIONS } from "../../authorization/index.js";
|
||||
import type { ConnectedHosts } from "../../connected-hosts/index.js";
|
||||
import { resolveRequiredProviderModel } from "../mcp-shared.js";
|
||||
import { selectAgentListItems } from "../agent-projections.js";
|
||||
import { summarizeAgentActivity } from "../activity-curator.js";
|
||||
import { expandUserPath } from "../../path-utils.js";
|
||||
import type { PaseoToolExecutionContext, PaseoToolResult } from "../tools/types.js";
|
||||
|
||||
export type HostOperation = (
|
||||
input: Record<string, unknown>,
|
||||
context: PaseoToolExecutionContext,
|
||||
) => Promise<PaseoToolResult>;
|
||||
export interface HostOperations {
|
||||
createAgent: HostOperation;
|
||||
createWorkspace: HostOperation;
|
||||
listWorkspaces: HostOperation;
|
||||
sendPrompt: HostOperation;
|
||||
getStatus: HostOperation;
|
||||
getActivity: HostOperation;
|
||||
listAgents: HostOperation;
|
||||
wait(agentId: string, signal: AbortSignal): Promise<WaitForFinishResult>;
|
||||
observe(
|
||||
agentId: string,
|
||||
handler: (snapshot: AgentSnapshotPayload | null) => void,
|
||||
signal: AbortSignal,
|
||||
): Promise<() => Promise<void>>;
|
||||
}
|
||||
export interface ToolHosts {
|
||||
directory(): Promise<HostDirectory>;
|
||||
use<T>(serverId: string, action: (client: DaemonClient) => Promise<T>): Promise<T>;
|
||||
}
|
||||
|
||||
/** Daemon MCP has the daemon's authority, bounded by the destination's admission of its link. */
|
||||
export function createToolHosts(hosts: ConnectedHosts): ToolHosts {
|
||||
return {
|
||||
directory: () => hosts.list(),
|
||||
async use(serverId, action) {
|
||||
const lifetime = new AbortController();
|
||||
const transportFactory: DaemonTransportFactory = () => {
|
||||
const opens = new Set<() => void>();
|
||||
const closes = new Set<(event?: unknown) => void>();
|
||||
const errors = new Set<(event: unknown) => void>();
|
||||
const messages = new Set<(data: unknown, binary: boolean) => void>();
|
||||
let closed = false;
|
||||
let route: Awaited<ReturnType<ConnectedHosts["openRoute"]>> | undefined;
|
||||
void hosts
|
||||
.openRoute({
|
||||
serverId,
|
||||
scopeId: randomUUID(),
|
||||
sourceId: "mcp",
|
||||
originSessionId: "mcp",
|
||||
permissions: OWNER_PERMISSIONS,
|
||||
signal: lifetime.signal,
|
||||
frame: (data) => {
|
||||
for (const listener of messages) listener(data, typeof data !== "string");
|
||||
},
|
||||
ended: (code, message) => {
|
||||
for (const listener of errors)
|
||||
listener(Object.assign(new Error(message), { code, serverId }));
|
||||
for (const listener of closes) listener();
|
||||
},
|
||||
})
|
||||
.then((opened) => {
|
||||
route = opened;
|
||||
if (closed) opened.close();
|
||||
else for (const listener of opens) listener();
|
||||
return undefined;
|
||||
})
|
||||
.catch((error) => {
|
||||
for (const listener of errors) listener(error);
|
||||
for (const listener of closes) listener(error);
|
||||
});
|
||||
return {
|
||||
send: (data) => {
|
||||
if (!route) throw new Error("Host route is not open");
|
||||
route.send(data instanceof ArrayBuffer ? new Uint8Array(data) : data);
|
||||
},
|
||||
close: () => {
|
||||
closed = true;
|
||||
route?.close();
|
||||
},
|
||||
onOpen: (listener) => {
|
||||
opens.add(listener);
|
||||
return () => opens.delete(listener);
|
||||
},
|
||||
onClose: (listener) => {
|
||||
closes.add(listener);
|
||||
return () => closes.delete(listener);
|
||||
},
|
||||
onError: (listener) => {
|
||||
errors.add(listener);
|
||||
return () => errors.delete(listener);
|
||||
},
|
||||
onMessage: (listener) => {
|
||||
messages.add(listener);
|
||||
return () => messages.delete(listener);
|
||||
},
|
||||
};
|
||||
};
|
||||
const client = new DaemonClient({
|
||||
url: "ws://host-route/ws",
|
||||
clientId: randomUUID(),
|
||||
transportFactory,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
try {
|
||||
await client.connect();
|
||||
return await action(client);
|
||||
} finally {
|
||||
lifetime.abort();
|
||||
await client.close();
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const remoteCreate = z
|
||||
.object({
|
||||
provider: z.string().min(1),
|
||||
title: z.string().trim().min(1).max(60),
|
||||
initialPrompt: z.string().min(1),
|
||||
workspaceId: z.string().min(1),
|
||||
labels: z.record(z.string(), z.string()).optional(),
|
||||
notifyOnFinish: z.boolean().optional(),
|
||||
idempotencyKey: z.string().optional(),
|
||||
settings: z
|
||||
.object({
|
||||
modeId: z.string().optional(),
|
||||
thinkingOptionId: z.string().optional(),
|
||||
features: z.record(z.string(), z.unknown()).optional(),
|
||||
})
|
||||
.optional(),
|
||||
})
|
||||
.strict();
|
||||
const result = (structuredContent: unknown): PaseoToolResult => ({
|
||||
content: [],
|
||||
structuredContent,
|
||||
});
|
||||
|
||||
export function createHostResolver(input: {
|
||||
serverId: string;
|
||||
local: HostOperations;
|
||||
hosts: () => ToolHosts;
|
||||
callerAgentId?: string;
|
||||
lockedCwd?: string;
|
||||
}) {
|
||||
return {
|
||||
directory: () => input.hosts().directory(),
|
||||
async use<T>(
|
||||
serverId: string | undefined,
|
||||
action: (host: HostOperations) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const destination = serverId ?? input.serverId;
|
||||
if (destination === input.serverId) return action(input.local);
|
||||
if (input.lockedCwd)
|
||||
throw new Error("This caller is restricted to its local working directory");
|
||||
return input.hosts().use(destination, async (client) =>
|
||||
action({
|
||||
async createAgent(args) {
|
||||
const parsed = remoteCreate.parse(args);
|
||||
const { provider, model } = resolveRequiredProviderModel(parsed.provider);
|
||||
const workspace = (await client.fetchWorkspaces({})).entries.find(
|
||||
(row) => row.id === parsed.workspaceId,
|
||||
);
|
||||
if (!workspace) throw new Error(`Workspace ${parsed.workspaceId} not found`);
|
||||
const labels = {
|
||||
...parsed.labels,
|
||||
...(input.callerAgentId
|
||||
? parentAddressLabels({ agentId: input.callerAgentId, serverId: input.serverId })
|
||||
: {}),
|
||||
};
|
||||
const snapshot = await client.createAgent({
|
||||
config: {
|
||||
provider,
|
||||
model,
|
||||
cwd: workspace.workspaceDirectory,
|
||||
title: parsed.title,
|
||||
modeId: parsed.settings?.modeId,
|
||||
thinkingOptionId: parsed.settings?.thinkingOptionId,
|
||||
featureValues: parsed.settings?.features,
|
||||
},
|
||||
workspaceId: parsed.workspaceId,
|
||||
labels,
|
||||
initialPrompt: parsed.initialPrompt,
|
||||
idempotencyKey: parsed.idempotencyKey,
|
||||
});
|
||||
return result({
|
||||
agentId: snapshot.id,
|
||||
type: snapshot.provider,
|
||||
status: snapshot.status,
|
||||
cwd: snapshot.cwd,
|
||||
workspaceId: snapshot.workspaceId,
|
||||
currentModeId: snapshot.currentModeId,
|
||||
availableModes: snapshot.availableModes,
|
||||
});
|
||||
},
|
||||
async sendPrompt(args) {
|
||||
const agentId = args.agentId as string;
|
||||
if (args.sessionMode !== undefined)
|
||||
await client.setAgentMode(agentId, args.sessionMode as string);
|
||||
await client.sendMessage(agentId, args.prompt as string, {
|
||||
sourceAgentId: input.callerAgentId,
|
||||
activeTurnBehavior:
|
||||
(args.activeTurnBehavior as ActiveTurnBehavior | undefined) ??
|
||||
(input.callerAgentId ? "steer" : "interrupt"),
|
||||
});
|
||||
const agent = (await client.fetchAgent(agentId))?.agent;
|
||||
if (!agent) throw new Error(`Agent ${agentId} not found`);
|
||||
return result({
|
||||
success: true,
|
||||
status: agent.status,
|
||||
guidance:
|
||||
input.callerAgentId && args.notifyOnFinish !== false
|
||||
? "You will get notified when the prompted agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives."
|
||||
: "Dispatch acknowledged. Observe completion using get_agent_status/get_agent_activity or SDK/RPC subscriptions and wait APIs.",
|
||||
});
|
||||
},
|
||||
async getStatus(args) {
|
||||
const agent = (await client.fetchAgent(args.agentId as string))?.agent;
|
||||
if (!agent) throw new Error(`Agent ${args.agentId} not found`);
|
||||
return result({ status: agent.status, snapshot: agent });
|
||||
},
|
||||
async getActivity(args) {
|
||||
const agentId = args.agentId as string;
|
||||
const limit = args.limit as number | undefined;
|
||||
const timeline = await client.fetchAgentTimeline(agentId, {
|
||||
direction: "tail",
|
||||
limit: limit ?? 0,
|
||||
});
|
||||
const agent = (await client.fetchAgent(agentId))?.agent;
|
||||
if (!agent) throw new Error(`Agent ${agentId} not found`);
|
||||
return result({
|
||||
agentId,
|
||||
...summarizeAgentActivity(
|
||||
timeline.entries.map((entry) => entry.item),
|
||||
limit,
|
||||
),
|
||||
currentModeId: agent.currentModeId,
|
||||
});
|
||||
},
|
||||
async listAgents(args) {
|
||||
const { entries } = await client.fetchAgents({
|
||||
includeBackground: args.includeBackground === true,
|
||||
filter: { includeArchived: args.includeArchived === true },
|
||||
});
|
||||
return result({
|
||||
agents: selectAgentListItems(
|
||||
entries.map((entry) => entry.agent),
|
||||
{
|
||||
cwd: (args.cwd as string | undefined)?.trim()
|
||||
? expandUserPath(args.cwd as string)
|
||||
: undefined,
|
||||
statuses: args.statuses as AgentSnapshotPayload["status"][] | undefined,
|
||||
sinceHours: args.sinceHours as number | undefined,
|
||||
limit: args.limit as number | undefined,
|
||||
},
|
||||
),
|
||||
});
|
||||
},
|
||||
async listWorkspaces(args) {
|
||||
const { entries } = await client.fetchWorkspaces({
|
||||
includeBackground: args.includeBackground === true,
|
||||
});
|
||||
return result({ workspaces: entries.map(workspaceSummary) });
|
||||
},
|
||||
async createWorkspace(args) {
|
||||
const isolation = z.enum(["local", "worktree"]).parse(args.isolation);
|
||||
const path = z.string().min(1).parse(args.path);
|
||||
const response = await client.createWorkspace({
|
||||
source:
|
||||
isolation === "local"
|
||||
? { kind: "directory", path, projectId: args.projectId as string | undefined }
|
||||
: {
|
||||
kind: "worktree",
|
||||
cwd: path,
|
||||
projectId: args.projectId as string | undefined,
|
||||
action:
|
||||
args.mode === "checkout-branch" || args.mode === "checkout-pr"
|
||||
? "checkout"
|
||||
: "branch-off",
|
||||
baseBranch: args.baseBranch as string | undefined,
|
||||
branchName: args.branchName as string | undefined,
|
||||
refName: args.branch as string | undefined,
|
||||
githubPrNumber: args.prNumber as number | undefined,
|
||||
worktreeSlug: args.worktreeSlug as string | undefined,
|
||||
},
|
||||
title: args.title as string | undefined,
|
||||
background: args.background as boolean | undefined,
|
||||
});
|
||||
if (response.error || !response.workspace)
|
||||
throw new Error(response.error ?? "Workspace creation failed");
|
||||
return result(workspaceSummary(response.workspace));
|
||||
},
|
||||
async observe(agentId, handler, signal) {
|
||||
const observation = client.observeTimeline([agentId], { signal });
|
||||
const refresh = async () => handler((await client.fetchAgent(agentId))?.agent ?? null);
|
||||
const unsubscribe = observation.subscribe({
|
||||
snapshot: () => {},
|
||||
update: (message) => {
|
||||
if (message.type !== "agent_stream" || message.payload.agentId !== agentId) return;
|
||||
const event = message.payload.event;
|
||||
if (
|
||||
event.type === "permission_resolved" ||
|
||||
event.type === "turn_started" ||
|
||||
event.type === "turn_completed" ||
|
||||
event.type === "turn_canceled" ||
|
||||
event.type === "turn_failed"
|
||||
)
|
||||
void refresh().catch(() => handler(null));
|
||||
},
|
||||
error: () => handler(null),
|
||||
});
|
||||
try {
|
||||
await observation.ready;
|
||||
await refresh();
|
||||
} catch (error) {
|
||||
unsubscribe();
|
||||
await observation.release();
|
||||
throw error;
|
||||
}
|
||||
return async () => {
|
||||
unsubscribe();
|
||||
await observation.release();
|
||||
};
|
||||
},
|
||||
wait: (agentId, signal) => {
|
||||
const close = () => void client.close();
|
||||
signal.addEventListener("abort", close, { once: true });
|
||||
return client
|
||||
.waitForFinish(agentId, 0)
|
||||
.finally(() => signal.removeEventListener("abort", close));
|
||||
},
|
||||
}),
|
||||
);
|
||||
},
|
||||
};
|
||||
}
|
||||
function workspaceSummary(workspace: FetchWorkspacesEntry) {
|
||||
return {
|
||||
workspaceId: workspace.id,
|
||||
projectId: workspace.projectId,
|
||||
cwd: workspace.workspaceDirectory,
|
||||
isolation: workspace.workspaceKind === "worktree" ? "worktree" : "local",
|
||||
kind: workspace.workspaceKind === "checkout" ? "local_checkout" : workspace.workspaceKind,
|
||||
title: workspace.title,
|
||||
background: workspace.background,
|
||||
};
|
||||
}
|
||||
|
||||
/** Wait pauses at permissions; observe only until that request clears before re-arming it. */
|
||||
export async function waitForPermissionChange(
|
||||
host: HostOperations,
|
||||
agentId: string,
|
||||
requestId: string,
|
||||
signal: AbortSignal,
|
||||
): Promise<void> {
|
||||
signal.throwIfAborted();
|
||||
let changed!: () => void;
|
||||
let failed!: (error: unknown) => void;
|
||||
const ready = new Promise<void>((resolve, reject) => {
|
||||
changed = resolve;
|
||||
failed = reject;
|
||||
});
|
||||
const abort = () => failed(signal.reason ?? new Error("Finish wait stopped"));
|
||||
signal.addEventListener("abort", abort, { once: true });
|
||||
let release: (() => Promise<void>) | undefined;
|
||||
try {
|
||||
release = await host.observe(
|
||||
agentId,
|
||||
(snapshot) => {
|
||||
if (!snapshot) failed(new Error(`Agent ${agentId} is unavailable`));
|
||||
else if (!snapshot.pendingPermissions.some((request) => request.id === requestId))
|
||||
changed();
|
||||
},
|
||||
signal,
|
||||
);
|
||||
await ready;
|
||||
} finally {
|
||||
signal.removeEventListener("abort", abort);
|
||||
await release?.();
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,10 @@
|
||||
import {
|
||||
createHostResolver,
|
||||
waitForPermissionChange,
|
||||
type HostOperation,
|
||||
type HostOperations,
|
||||
type ToolHosts,
|
||||
} from "../host-operations/index.js";
|
||||
import { AgentDirectory } from "../directory.js";
|
||||
import { stat } from "node:fs/promises";
|
||||
import { z } from "zod";
|
||||
@@ -16,14 +23,12 @@ import {
|
||||
AgentSnapshotPayloadSchema,
|
||||
WorkspaceScriptPayloadSchema,
|
||||
} from "../../messages.js";
|
||||
import type { AgentListItemPayload } from "../../messages.js";
|
||||
import {
|
||||
buildStoredAgentPayload,
|
||||
toAgentListItemPayload,
|
||||
selectAgentListItems,
|
||||
toAgentPayload,
|
||||
} from "../agent-projections.js";
|
||||
import { curateAgentActivity } from "../activity-curator.js";
|
||||
import { selectItemsByProjectedLimit } from "../timeline-projection.js";
|
||||
import { summarizeAgentActivity } from "../activity-curator.js";
|
||||
import type { AgentStorage } from "../agent-storage.js";
|
||||
import { AgentNotFoundError, ensureAgentLoaded } from "../agent-loading.js";
|
||||
import { isStoredAgentProviderAvailable } from "../../persistence-hooks.js";
|
||||
@@ -72,6 +77,7 @@ import {
|
||||
type PromptDispatchDisposition,
|
||||
sendPromptToAgent,
|
||||
setupFinishNotification,
|
||||
setupFinishWait,
|
||||
waitForAgentRunStartWithTimeout,
|
||||
} from "../agent-prompt.js";
|
||||
import { respondToAgentPermission } from "../permission-response.js";
|
||||
@@ -110,6 +116,8 @@ import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-confi
|
||||
import { isPaseoToolEnabled } from "../paseo-tool-policy.js";
|
||||
|
||||
export interface PaseoToolHostDependencies {
|
||||
serverId?: string;
|
||||
hosts?: () => ToolHosts;
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
terminalManager?: TerminalManager | null;
|
||||
@@ -165,22 +173,8 @@ export interface PaseoToolHostDependencies {
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
function parseTimestamp(value: string | null | undefined): number {
|
||||
if (!value) {
|
||||
return 0;
|
||||
}
|
||||
const parsed = Date.parse(value);
|
||||
return Number.isNaN(parsed) ? 0 : parsed;
|
||||
}
|
||||
|
||||
function resolveAgentListActivityTime(agent: AgentListItemPayload): number {
|
||||
return Math.max(
|
||||
parseTimestamp(agent.updatedAt),
|
||||
parseTimestamp(agent.lastUserMessageAt),
|
||||
parseTimestamp(agent.attentionTimestamp),
|
||||
parseTimestamp(agent.archivedAt),
|
||||
parseTimestamp(agent.createdAt),
|
||||
);
|
||||
function settledStatus(status: string): "error" | "idle" {
|
||||
return status === "error" ? "error" : "idle";
|
||||
}
|
||||
|
||||
interface ProviderSummary {
|
||||
@@ -328,28 +322,6 @@ function toProviderSummary(entry: {
|
||||
};
|
||||
}
|
||||
|
||||
function compareAgentListItems(a: AgentListItemPayload, b: AgentListItemPayload): number {
|
||||
const attentionDelta =
|
||||
Number(b.requiresAttention ?? false) - Number(a.requiresAttention ?? false);
|
||||
if (attentionDelta !== 0) {
|
||||
return attentionDelta;
|
||||
}
|
||||
|
||||
const statusOrder = {
|
||||
running: 0,
|
||||
initializing: 1,
|
||||
idle: 2,
|
||||
error: 3,
|
||||
closed: 4,
|
||||
} as Record<string, number>;
|
||||
const statusDelta = (statusOrder[a.status] ?? 999) - (statusOrder[b.status] ?? 999);
|
||||
if (statusDelta !== 0) {
|
||||
return statusDelta;
|
||||
}
|
||||
|
||||
return resolveAgentListActivityTime(b) - resolveAgentListActivityTime(a);
|
||||
}
|
||||
|
||||
function resolveScheduleProviderAndModel(params: {
|
||||
provider?: string;
|
||||
defaultProvider: AgentProvider;
|
||||
@@ -685,7 +657,66 @@ function createToolCatalog(
|
||||
return result.data;
|
||||
};
|
||||
|
||||
const localHostOperations = new Map<string, HostOperation>();
|
||||
const tools = new Map<string, PaseoToolDefinition>();
|
||||
const hostTools = new Map([
|
||||
["create_agent", "createAgent"],
|
||||
["create_workspace", "createWorkspace"],
|
||||
["list_workspaces", "listWorkspaces"],
|
||||
["send_agent_prompt", "sendPrompt"],
|
||||
["get_agent_status", "getStatus"],
|
||||
["get_agent_activity", "getActivity"],
|
||||
["list_agents", "listAgents"],
|
||||
] as const satisfies ReadonlyArray<readonly [string, keyof HostOperations]>);
|
||||
const hostResolver = createHostResolver({
|
||||
serverId: options.serverId ?? "self",
|
||||
callerAgentId,
|
||||
lockedCwd: callerContext?.lockedCwd,
|
||||
hosts: () => {
|
||||
if (!options.hosts) throw new Error("Connected hosts are not configured");
|
||||
return options.hosts();
|
||||
},
|
||||
local: {
|
||||
createAgent: (args, context) => localHostOperations.get("create_agent")!(args, context),
|
||||
createWorkspace: (args, context) =>
|
||||
localHostOperations.get("create_workspace")!(args, context),
|
||||
listWorkspaces: (args, context) => localHostOperations.get("list_workspaces")!(args, context),
|
||||
sendPrompt: (args, context) => localHostOperations.get("send_agent_prompt")!(args, context),
|
||||
getStatus: (args, context) => localHostOperations.get("get_agent_status")!(args, context),
|
||||
getActivity: (args, context) => localHostOperations.get("get_agent_activity")!(args, context),
|
||||
listAgents: (args, context) => localHostOperations.get("list_agents")!(args, context),
|
||||
async observe(agentId, handler, signal) {
|
||||
const unsubscribe = agentManager.subscribe(
|
||||
(event) => {
|
||||
if (event.type === "agent_state") handler(toAgentPayload(event.agent));
|
||||
},
|
||||
{ agentId, replayState: true },
|
||||
);
|
||||
const stop = () => unsubscribe();
|
||||
signal.addEventListener("abort", stop, { once: true });
|
||||
return async () => {
|
||||
signal.removeEventListener("abort", stop);
|
||||
unsubscribe();
|
||||
};
|
||||
},
|
||||
async wait(agentId, signal) {
|
||||
const settled = await agentManager.waitForAgentEvent(agentId, {
|
||||
signal,
|
||||
waitForActive: true,
|
||||
});
|
||||
const snapshot = agentManager.getAgent(agentId);
|
||||
const final = snapshot
|
||||
? await serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger)
|
||||
: null;
|
||||
return {
|
||||
status: settled.permission ? "permission" : settledStatus(settled.status),
|
||||
final,
|
||||
lastMessage: settled.lastMessage,
|
||||
error: final?.lastError ?? null,
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
const registerTool = (
|
||||
name: string,
|
||||
config: PaseoToolConfig,
|
||||
@@ -695,6 +726,76 @@ function createToolCatalog(
|
||||
if (!isPaseoToolEnabled(options.paseoToolPolicy, name)) {
|
||||
return;
|
||||
}
|
||||
const operation = hostTools.get(name as Parameters<typeof hostTools.get>[0]);
|
||||
if (operation) {
|
||||
localHostOperations.set(name, handler);
|
||||
const original = config.inputSchema;
|
||||
const selector = {
|
||||
serverId: z
|
||||
.string()
|
||||
.min(1)
|
||||
.optional()
|
||||
.describe("Destination server ID from list_hosts; omit for this host."),
|
||||
};
|
||||
config = {
|
||||
...config,
|
||||
inputSchema:
|
||||
original instanceof z.ZodObject
|
||||
? original.extend(selector)
|
||||
: { ...(original as z.ZodRawShape), ...selector },
|
||||
};
|
||||
handler = async ({ serverId, ...input }, context) => {
|
||||
const destination = serverId ?? options.serverId;
|
||||
try {
|
||||
const response = await hostResolver.use(serverId, (host) =>
|
||||
host[operation](input, context),
|
||||
);
|
||||
let childAgentId: string | undefined;
|
||||
if (name === "create_agent") {
|
||||
childAgentId = (response.structuredContent as { agentId: string }).agentId;
|
||||
} else if (name === "send_agent_prompt" && destination !== options.serverId) {
|
||||
childAgentId = input.agentId as string;
|
||||
}
|
||||
if (childAgentId && callerAgentId && input.notifyOnFinish !== false) {
|
||||
setupFinishWait({
|
||||
agentManager,
|
||||
agentStorage,
|
||||
originServerId: options.serverId,
|
||||
childServerId: destination,
|
||||
childAgentId,
|
||||
callerAgentId,
|
||||
logger: childLogger,
|
||||
wait: (signal) =>
|
||||
hostResolver.use(destination, (host) => host.wait(childAgentId, signal)),
|
||||
afterPermission: (requestId, signal) =>
|
||||
hostResolver.use(destination, (host) =>
|
||||
waitForPermissionChange(host, childAgentId, requestId, signal),
|
||||
),
|
||||
});
|
||||
}
|
||||
return {
|
||||
...response,
|
||||
structuredContent: {
|
||||
...(response.structuredContent as object),
|
||||
...(destination ? { serverId: destination } : {}),
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
if (!options.serverId) throw error;
|
||||
const failure = {
|
||||
serverId: destination,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
...(error instanceof Error && "code" in error ? { code: error.code } : {}),
|
||||
...(error instanceof Error && "outcome" in error ? { outcome: error.outcome } : {}),
|
||||
};
|
||||
return {
|
||||
isError: true,
|
||||
content: [{ type: "text", text: JSON.stringify(failure) }],
|
||||
structuredContent: failure,
|
||||
};
|
||||
}
|
||||
};
|
||||
}
|
||||
tools.set(name, {
|
||||
name,
|
||||
title: config.title,
|
||||
@@ -1259,6 +1360,19 @@ function createToolCatalog(
|
||||
});
|
||||
}
|
||||
|
||||
registerTool(
|
||||
"list_hosts",
|
||||
{
|
||||
title: "List hosts",
|
||||
description: "Discover this host and registered destination hosts for fleet tools.",
|
||||
inputSchema: {},
|
||||
},
|
||||
async () => ({
|
||||
content: [],
|
||||
structuredContent: ensureValidJson(await hostResolver.directory()),
|
||||
}),
|
||||
);
|
||||
|
||||
registerTool(
|
||||
"create_workspace",
|
||||
{
|
||||
@@ -1520,10 +1634,15 @@ function createToolCatalog(
|
||||
workspaceId: resolvedArgs.workspaceId,
|
||||
thinking: parsedArgs.settings?.thinkingOptionId,
|
||||
features: parsedArgs.settings?.features,
|
||||
labels: parsedArgs.labels,
|
||||
labels: {
|
||||
...parsedArgs.labels,
|
||||
...(callerAgentId && !resolvedArgs.detached && options.serverId
|
||||
? { "paseo.parent-server-id": options.serverId }
|
||||
: {}),
|
||||
},
|
||||
mode: parsedArgs.settings?.modeId,
|
||||
promptFailure: "throw",
|
||||
notifyOnFinish,
|
||||
notifyOnFinish: false,
|
||||
detached: resolvedArgs.detached,
|
||||
callerAgentId,
|
||||
callerContext,
|
||||
@@ -1849,6 +1968,7 @@ function createToolCatalog(
|
||||
agentManager,
|
||||
agentStorage,
|
||||
childAgentId: agentId,
|
||||
childServerId: options.serverId,
|
||||
callerAgentId,
|
||||
waitForTurnStart: false,
|
||||
dispatchResult,
|
||||
@@ -1869,6 +1989,7 @@ function createToolCatalog(
|
||||
agentManager,
|
||||
agentStorage,
|
||||
childAgentId: agentId,
|
||||
childServerId: options.serverId,
|
||||
callerAgentId,
|
||||
waitForTurnStart: true,
|
||||
logger: childLogger,
|
||||
@@ -2004,8 +2125,6 @@ function createToolCatalog(
|
||||
}) => {
|
||||
const callerCwd = callerAgentId ? resolveCallerAgent()?.cwd : undefined;
|
||||
const requestedCwd = cwd?.trim() ? expandUserPath(cwd) : callerCwd;
|
||||
const statusFilter = statuses && statuses.length > 0 ? new Set(statuses) : null;
|
||||
const sinceMs = Date.now() - sinceHours * 60 * 60 * 1000;
|
||||
if (!options.workspaceRegistry) throw new Error("Workspace registry is not configured");
|
||||
const directory = new AgentDirectory({
|
||||
manager: agentManager,
|
||||
@@ -2025,13 +2144,10 @@ function createToolCatalog(
|
||||
new Set(providerSnapshotManager.listRegisteredProviderIds()),
|
||||
),
|
||||
});
|
||||
const agents = (await directory.list({ includeBackground, includeArchived }))
|
||||
.map(toAgentListItemPayload)
|
||||
.filter((agent) => !requestedCwd || isSameOrDescendantPath(requestedCwd, agent.cwd))
|
||||
.filter((agent) => !statusFilter || statusFilter.has(agent.status))
|
||||
.filter((agent) => !agent.archivedAt || resolveAgentListActivityTime(agent) >= sinceMs)
|
||||
.sort(compareAgentListItems)
|
||||
.slice(0, limit);
|
||||
const agents = selectAgentListItems(
|
||||
await directory.list({ includeBackground, includeArchived }),
|
||||
{ cwd: requestedCwd, statuses, sinceHours, limit },
|
||||
);
|
||||
|
||||
return {
|
||||
content: [],
|
||||
@@ -3028,29 +3144,12 @@ function createToolCatalog(
|
||||
const timeline = agentManager.getTimeline(agentId);
|
||||
const snapshot = agentManager.getAgent(agentId);
|
||||
|
||||
const selection = selectItemsByProjectedLimit({
|
||||
items: timeline,
|
||||
direction: "tail",
|
||||
limit: limit ?? 0,
|
||||
});
|
||||
const curatedContent = curateAgentActivity(selection.items);
|
||||
const { totalProjected, shownProjected } = selection;
|
||||
|
||||
const noun = totalProjected === 1 ? "activity" : "activities";
|
||||
const countHeader =
|
||||
limit && shownProjected < totalProjected
|
||||
? `Showing ${shownProjected} of ${totalProjected} ${noun} (limited to ${limit})`
|
||||
: `Showing all ${totalProjected} ${noun}`;
|
||||
|
||||
const contentWithCount = `${countHeader}\n\n${curatedContent}`;
|
||||
|
||||
return {
|
||||
content: [],
|
||||
structuredContent: ensureValidJson({
|
||||
agentId,
|
||||
updateCount: timeline.length,
|
||||
...summarizeAgentActivity(timeline, limit),
|
||||
currentModeId: snapshot?.currentModeId ?? null,
|
||||
content: contentWithCount,
|
||||
}),
|
||||
};
|
||||
},
|
||||
|
||||
@@ -6,6 +6,14 @@ type OutboundOperation = SessionOutboundMessage["type"];
|
||||
export type PermissionRequirement = DaemonPermission | readonly DaemonPermission[] | null;
|
||||
|
||||
const INBOUND_PERMISSION = {
|
||||
"hosts.list.request": "daemon.read",
|
||||
"hosts.add.request": "tunnel.manage",
|
||||
"hosts.update.request": "tunnel.manage",
|
||||
"hosts.remove.request": "tunnel.manage",
|
||||
"hosts.link.open.request": null,
|
||||
"hosts.route.open.request": null,
|
||||
"hosts.route.frame.request": null,
|
||||
"hosts.route.close.request": null,
|
||||
abort_request: "workspace.write",
|
||||
"agent.config.apply.request": ["workspace.write", "hub.execute"],
|
||||
"agent.detach.request": "workspace.write",
|
||||
@@ -216,6 +224,15 @@ const INBOUND_PERMISSION = {
|
||||
} as const satisfies Record<InboundOperation, PermissionRequirement>;
|
||||
|
||||
const OUTBOUND_PERMISSION = {
|
||||
"hosts.list.response": "daemon.read",
|
||||
"hosts.changed": "daemon.read",
|
||||
"hosts.add.response": "tunnel.manage",
|
||||
"hosts.update.response": "tunnel.manage",
|
||||
"hosts.remove.response": "tunnel.manage",
|
||||
"hosts.link.open.response": null,
|
||||
"hosts.route.open.response": null,
|
||||
"hosts.route.frame": null,
|
||||
"hosts.route.closed": null,
|
||||
"agent.create.response": ["workspace.write", "hub.execute"],
|
||||
"agent.create.update": ["workspace.write", "hub.execute"],
|
||||
"workspace.create.update": "workspace.manage",
|
||||
|
||||
@@ -1393,6 +1393,11 @@ export async function createPaseoDaemon(
|
||||
const createAgentToolHostDependencies = (
|
||||
runtime: PaseoToolRuntimeContext,
|
||||
): PaseoToolHostDependencies => ({
|
||||
serverId,
|
||||
hosts: () => {
|
||||
if (!wsServer) throw new Error("Daemon is not accepting host operations");
|
||||
return wsServer.toolHosts();
|
||||
},
|
||||
agentManager,
|
||||
agentStorage,
|
||||
terminalManager,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,539 @@
|
||||
export { DaemonLinkEndpoint, VirtualHostSocket } from "./link.js";
|
||||
export { ConnectedHostSession } from "./session.js";
|
||||
import { randomUUID, createHash } from "node:crypto";
|
||||
import { mkdir, readFile, rename, writeFile, rm } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { z } from "zod";
|
||||
import {
|
||||
createHostClient,
|
||||
createHostChannelTransport,
|
||||
parseHostTarget,
|
||||
HostTransportBudget,
|
||||
} from "@getpaseo/client/node";
|
||||
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
||||
import {
|
||||
HostSummarySchema,
|
||||
HostAddSchema,
|
||||
HostUpdateSchema,
|
||||
type HostDirectory,
|
||||
type HostSummary,
|
||||
type HostAdd,
|
||||
type HostUpdate,
|
||||
} from "@getpaseo/protocol/hosts";
|
||||
import type { DaemonPermission, WSHelloMessage } from "@getpaseo/protocol/messages";
|
||||
|
||||
export class HostError extends Error {
|
||||
constructor(
|
||||
readonly code: string,
|
||||
message: string,
|
||||
readonly serverId?: string,
|
||||
) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
const SavedSchema = HostSummarySchema.extend({ connection: z.string() });
|
||||
type Saved = z.infer<typeof SavedSchema>;
|
||||
export interface HostActor {
|
||||
sourceId: string;
|
||||
manageSources: boolean;
|
||||
}
|
||||
interface Live {
|
||||
client: DaemonClient;
|
||||
accepted: boolean;
|
||||
stop(): void;
|
||||
}
|
||||
interface Route {
|
||||
registrationId: string;
|
||||
close(code?: string, message?: string): void;
|
||||
}
|
||||
|
||||
/** Daemon-owned configuration, identity pinning, retry and routed session lifetime. */
|
||||
export class ConnectedHosts {
|
||||
private readonly budget = new HostTransportBudget();
|
||||
private rows: Saved[] = [];
|
||||
private readonly live = new Map<string, Live>();
|
||||
private readonly routes = new Set<Route>();
|
||||
private readonly listeners = new Set<() => void>();
|
||||
private queue: Promise<unknown> = Promise.resolve();
|
||||
private readonly ready: Promise<void>;
|
||||
private closed = false;
|
||||
constructor(
|
||||
private readonly home: string,
|
||||
private readonly self: HostDirectory["self"],
|
||||
) {
|
||||
this.ready = this.load();
|
||||
void this.ready.catch(() => {});
|
||||
}
|
||||
private async load(): Promise<void> {
|
||||
try {
|
||||
this.rows = z
|
||||
.array(SavedSchema)
|
||||
.parse(JSON.parse(await readFile(join(this.home, "hosts.json"), "utf8")));
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT")
|
||||
throw new Error("Cannot read saved connected hosts", { cause: error });
|
||||
}
|
||||
if (this.closed) return;
|
||||
for (const row of this.rows) {
|
||||
row.status = row.enabled ? "connecting" : "disabled";
|
||||
row.nextRetryAt = null;
|
||||
this.start(row);
|
||||
}
|
||||
}
|
||||
async list(): Promise<HostDirectory> {
|
||||
await this.ready;
|
||||
return {
|
||||
self: { ...this.self },
|
||||
hosts: this.rows.map(({ connection: _secret, ...row }) => structuredClone(row)),
|
||||
};
|
||||
}
|
||||
observe(changed: () => void): () => void {
|
||||
this.listeners.add(changed);
|
||||
return () => this.listeners.delete(changed);
|
||||
}
|
||||
private changed(): void {
|
||||
for (const listener of this.listeners) listener();
|
||||
}
|
||||
private async persist(rows: Saved[]): Promise<void> {
|
||||
await mkdir(this.home, { recursive: true });
|
||||
const temp = join(this.home, `.hosts-${randomUUID()}.tmp`);
|
||||
try {
|
||||
await writeFile(temp, JSON.stringify(rows), { mode: 0o600 });
|
||||
await rename(temp, join(this.home, "hosts.json"));
|
||||
} finally {
|
||||
await rm(temp, { force: true });
|
||||
}
|
||||
}
|
||||
private mutate<T>(run: () => Promise<T>): Promise<T> {
|
||||
const result = this.queue.then(async () => {
|
||||
await this.ready;
|
||||
if (this.closed) throw new HostError("host_offline", "Daemon is stopping");
|
||||
return run();
|
||||
});
|
||||
this.queue = result.catch(() => {});
|
||||
return result;
|
||||
}
|
||||
private summary(row: Saved): HostSummary {
|
||||
const { connection: _secret, ...summary } = row;
|
||||
return structuredClone(summary);
|
||||
}
|
||||
add(input: HostAdd, actor: HostActor): Promise<HostSummary> {
|
||||
return this.mutate(async () => {
|
||||
input = HostAddSchema.parse(input);
|
||||
const target = parseHostTarget(input.connection);
|
||||
const source = { id: actor.sourceId, key: input.sourceKey ?? randomUUID() };
|
||||
const existing = this.rows.find(
|
||||
(row) => row.source.id === source.id && row.source.key === source.key,
|
||||
);
|
||||
if (existing) {
|
||||
if (
|
||||
existing.connection === input.connection &&
|
||||
existing.label === (input.label ?? null) &&
|
||||
JSON.stringify(Object.entries(existing.metadata).sort()) ===
|
||||
JSON.stringify(Object.entries(input.metadata ?? {}).sort())
|
||||
)
|
||||
return this.summary(existing);
|
||||
throw new HostError(
|
||||
"source_conflict",
|
||||
"This source key already exists. Update its registration explicitly.",
|
||||
);
|
||||
}
|
||||
if (this.rows.length >= 128)
|
||||
throw new HostError("resource_limit", "Remove a host before adding another (limit 128)");
|
||||
const row: Saved = {
|
||||
registrationId: `hst_${randomUUID()}`,
|
||||
serverId: null,
|
||||
connection: input.connection,
|
||||
label: input.label ?? null,
|
||||
metadata: input.metadata ?? {},
|
||||
source,
|
||||
enabled: true,
|
||||
status: "connecting",
|
||||
targetDescription: target.description,
|
||||
lastOnlineAt: null,
|
||||
nextRetryAt: null,
|
||||
error: null,
|
||||
};
|
||||
await this.persist([...this.rows, row]);
|
||||
this.rows.push(row);
|
||||
this.start(row);
|
||||
this.changed();
|
||||
return this.summary(row);
|
||||
});
|
||||
}
|
||||
update(id: string, patch: HostUpdate, actor: HostActor): Promise<HostSummary> {
|
||||
return this.mutate(async () => {
|
||||
patch = HostUpdateSchema.parse(patch);
|
||||
const previous = this.require(id);
|
||||
if (
|
||||
(patch.connection !== undefined || patch.metadata !== undefined || patch.replaceIdentity) &&
|
||||
actor.sourceId !== previous.source.id &&
|
||||
!actor.manageSources
|
||||
)
|
||||
throw new HostError(
|
||||
"permission_denied",
|
||||
"Only the source or access manager can replace its connection and metadata",
|
||||
);
|
||||
if (patch.replaceIdentity && patch.connection === undefined)
|
||||
throw new HostError(
|
||||
"invalid_connection",
|
||||
"Identity replacement requires a replacement connection",
|
||||
);
|
||||
const row: Saved = {
|
||||
...previous,
|
||||
...(patch.label !== undefined ? { label: patch.label } : {}),
|
||||
...(patch.metadata !== undefined ? { metadata: patch.metadata } : {}),
|
||||
...(patch.enabled !== undefined ? { enabled: patch.enabled } : {}),
|
||||
};
|
||||
if (patch.connection !== undefined) {
|
||||
row.connection = patch.connection;
|
||||
row.targetDescription = parseHostTarget(patch.connection).description;
|
||||
}
|
||||
if (patch.replaceIdentity) row.serverId = null;
|
||||
const restart =
|
||||
patch.connection !== undefined || patch.enabled !== undefined || patch.replaceIdentity;
|
||||
if (restart) {
|
||||
row.status = row.enabled ? "connecting" : "disabled";
|
||||
row.error = null;
|
||||
row.nextRetryAt = null;
|
||||
}
|
||||
const rows = this.rows.map((entry) => (entry === previous ? row : entry));
|
||||
await this.persist(rows);
|
||||
if (restart) this.stop(id, row.enabled ? "host_offline" : "host_disabled");
|
||||
Object.assign(previous, row);
|
||||
this.rows = rows.map((entry) => (entry === row ? previous : entry));
|
||||
if (restart) this.start(previous);
|
||||
this.changed();
|
||||
return this.summary(previous);
|
||||
});
|
||||
}
|
||||
remove(id: string, actor: HostActor): Promise<void> {
|
||||
return this.mutate(async () => {
|
||||
const row = this.require(id);
|
||||
if (actor.sourceId !== row.source.id && !actor.manageSources)
|
||||
throw new HostError(
|
||||
"permission_denied",
|
||||
"Only the source or access manager can remove this host. Disable it to stop access.",
|
||||
);
|
||||
const rows = this.rows.filter((entry) => entry !== row);
|
||||
await this.persist(rows);
|
||||
this.stop(id, "host_unknown");
|
||||
this.rows = rows;
|
||||
this.changed();
|
||||
});
|
||||
}
|
||||
private require(id: string): Saved {
|
||||
const row = this.rows.find((entry) => entry.registrationId === id);
|
||||
if (!row) throw new HostError("host_unknown", "Unknown host registration");
|
||||
return row;
|
||||
}
|
||||
private stop(id: string, code = "host_offline"): void {
|
||||
this.live.get(id)?.stop();
|
||||
this.live.delete(id);
|
||||
for (const route of this.routes)
|
||||
if (route.registrationId === id)
|
||||
route.close(code, "Host access was disabled, replaced or removed");
|
||||
}
|
||||
private start(row: Saved): void {
|
||||
if (!row.enabled || this.closed) return;
|
||||
let stopped = false;
|
||||
const client = createHostClient(
|
||||
row.connection,
|
||||
{
|
||||
clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
|
||||
connectTimeoutMs: 10000,
|
||||
},
|
||||
this.budget,
|
||||
);
|
||||
const live: Live = {
|
||||
client,
|
||||
accepted: false,
|
||||
stop: () => {
|
||||
stopped = true;
|
||||
void client.close();
|
||||
},
|
||||
};
|
||||
this.live.set(row.registrationId, live);
|
||||
const accept = async () => {
|
||||
try {
|
||||
const info = client.getLastServerInfoMessage();
|
||||
if (!info) throw new HostError("host_offline", "No daemon identity received");
|
||||
if (row.serverId && info.serverId !== row.serverId)
|
||||
throw new HostError(
|
||||
"server_id_mismatch",
|
||||
"Daemon identity changed. Verify the destination and explicitly replace its identity.",
|
||||
);
|
||||
if (!row.serverId)
|
||||
await this.mutate(async () => {
|
||||
if (stopped) return;
|
||||
const discovered = { ...row, serverId: info.serverId };
|
||||
await this.persist(this.rows.map((entry) => (entry === row ? discovered : entry)));
|
||||
row.serverId = info.serverId;
|
||||
});
|
||||
if (info.serverId === this.self.serverId)
|
||||
throw new HostError(
|
||||
"self_host",
|
||||
"This connection reaches this daemon. Remove it or replace the connection.",
|
||||
);
|
||||
if (this.rows.find((entry) => entry.serverId === info.serverId) !== row)
|
||||
throw new HostError(
|
||||
"duplicate_host",
|
||||
"Another registration already owns this daemon. Remove the duplicate registration.",
|
||||
);
|
||||
if (info.features?.connectedHosts !== true || info.features?.hostMultiplexing !== true)
|
||||
throw new HostError(
|
||||
"unsupported_capability",
|
||||
"Update the destination daemon to use connected hosts.",
|
||||
);
|
||||
// Pin and persist under the same writer as management to prevent discovery races.
|
||||
await this.mutate(async () => {
|
||||
if (stopped) return;
|
||||
if (this.rows.find((entry) => entry.serverId === info.serverId) !== row)
|
||||
throw new HostError("duplicate_host", "Another registration already owns this daemon");
|
||||
const next = {
|
||||
...row,
|
||||
serverId: info.serverId,
|
||||
status: "online" as const,
|
||||
lastOnlineAt: new Date().toISOString(),
|
||||
error: null,
|
||||
nextRetryAt: null,
|
||||
};
|
||||
await this.persist(this.rows.map((entry) => (entry === row ? next : entry)));
|
||||
if (stopped || client.getLastServerInfoMessage() !== info || !client.isConnected) return;
|
||||
Object.assign(row, next);
|
||||
live.accepted = true;
|
||||
this.changed();
|
||||
});
|
||||
} catch (error) {
|
||||
if (stopped) return;
|
||||
row.status = "error";
|
||||
row.error = connectionFailure(error, client.authFailureReason !== null) ?? {
|
||||
code: "host_offline",
|
||||
message: "Host connection failed",
|
||||
};
|
||||
row.nextRetryAt = null;
|
||||
this.changed();
|
||||
await client.close();
|
||||
}
|
||||
};
|
||||
client.subscribeConnectionStatus((state) => {
|
||||
if (stopped || row.status === "error") return;
|
||||
live.accepted = false;
|
||||
if (state.status === "connected") {
|
||||
void accept();
|
||||
return;
|
||||
}
|
||||
for (const route of this.routes)
|
||||
if (route.registrationId === row.registrationId) route.close();
|
||||
row.status = state.status === "connecting" ? "connecting" : "offline";
|
||||
row.nextRetryAt = state.status === "disconnected" ? (state.nextRetryAt ?? null) : null;
|
||||
row.error =
|
||||
state.status === "disconnected"
|
||||
? (connectionFailure(
|
||||
Object.assign(new Error("Check the SSH configuration and daemon credentials"), {
|
||||
code: state.failureCode,
|
||||
}),
|
||||
client.authFailureReason !== null,
|
||||
) ?? {
|
||||
code: "host_offline",
|
||||
message: "Cannot reach host. Check the connection; retrying automatically.",
|
||||
})
|
||||
: null;
|
||||
if (row.error?.code === "authentication_failed") {
|
||||
row.status = "error";
|
||||
row.nextRetryAt = null;
|
||||
client.setReconnectEnabled(false);
|
||||
void client.close();
|
||||
}
|
||||
this.changed();
|
||||
});
|
||||
void client.connect().catch(() => {});
|
||||
}
|
||||
async openRoute(input: {
|
||||
serverId: string;
|
||||
scopeId: string;
|
||||
sourceId: string;
|
||||
originSessionId: string;
|
||||
permissions: readonly DaemonPermission[];
|
||||
signal: AbortSignal;
|
||||
frame(data: string | Uint8Array): void;
|
||||
ended(code: string, message: string): void;
|
||||
}): Promise<{ send(data: string | Uint8Array): void; close(): void }> {
|
||||
await this.ready;
|
||||
const row = this.rows.find((entry) => entry.serverId === input.serverId);
|
||||
if (!row) throw new HostError("host_unknown", `Unknown host ${input.serverId}`, input.serverId);
|
||||
if (!row.enabled)
|
||||
throw new HostError(
|
||||
"host_disabled",
|
||||
`Enable host ${input.serverId} before using it`,
|
||||
input.serverId,
|
||||
);
|
||||
if (row.status !== "online")
|
||||
throw new HostError(
|
||||
row.error?.code ?? "host_offline",
|
||||
row.error?.message ?? `Host ${input.serverId} is offline`,
|
||||
input.serverId,
|
||||
);
|
||||
if (this.routes.size >= 128) throw new HostError("resource_limit", "Too many routed sessions");
|
||||
const live = this.live.get(row.registrationId);
|
||||
const client = live?.client;
|
||||
if (!live?.accepted || !client?.isConnected)
|
||||
throw new HostError("host_offline", "Host is offline", input.serverId);
|
||||
const transport = createHostChannelTransport(
|
||||
client,
|
||||
input.serverId,
|
||||
randomUUID(),
|
||||
null,
|
||||
true,
|
||||
)({ url: "ws://host-link/ws" });
|
||||
let closed = false;
|
||||
let verified = false;
|
||||
let helloSent = false;
|
||||
const timer = setTimeout(() => route.close("host_offline", "Host handshake timed out"), 15000);
|
||||
timer.unref();
|
||||
const route: Route = {
|
||||
registrationId: row.registrationId,
|
||||
close: (code = "host_offline", message = `Host ${input.serverId} disconnected`) => {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
input.signal.removeEventListener("abort", abort);
|
||||
clearTimeout(timer);
|
||||
this.routes.delete(route);
|
||||
try {
|
||||
transport.close();
|
||||
} finally {
|
||||
failed(new HostError(code, message, input.serverId));
|
||||
input.ended(code, message);
|
||||
}
|
||||
},
|
||||
};
|
||||
this.routes.add(route);
|
||||
const abort = () =>
|
||||
route.close("host_offline", "Origin session closed or its authority changed");
|
||||
input.signal.addEventListener("abort", abort, { once: true });
|
||||
let opened!: () => void;
|
||||
let failed!: (error: unknown) => void;
|
||||
const ready = new Promise<void>((resolve, reject) => {
|
||||
opened = resolve;
|
||||
failed = reject;
|
||||
});
|
||||
if (input.signal.aborted) {
|
||||
route.close();
|
||||
}
|
||||
transport.onOpen(opened);
|
||||
transport.onError((error) => {
|
||||
const authenticationFailed =
|
||||
error instanceof Error && "code" in error && error.code === "authentication_failed";
|
||||
const code = authenticationFailed ? "authentication_failed" : "host_offline";
|
||||
const message = authenticationFailed
|
||||
? "Host authentication failed. Check the SSH configuration and daemon credentials."
|
||||
: "Host transport failed";
|
||||
failed(new HostError(code, message, input.serverId));
|
||||
route.close(code, message);
|
||||
});
|
||||
transport.onClose(() => {
|
||||
failed(new HostError("host_offline", "Host transport closed"));
|
||||
route.close();
|
||||
});
|
||||
transport.onMessage((data, binary) => {
|
||||
if (closed) return;
|
||||
if (!verified) {
|
||||
if (binary) {
|
||||
route.close();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const envelope = JSON.parse(String(data));
|
||||
if (envelope.type === "hello.rejected") {
|
||||
route.close("authentication_failed", "Replace the host credentials");
|
||||
return;
|
||||
}
|
||||
const info =
|
||||
envelope.type === "session" && envelope.message.type === "status"
|
||||
? envelope.message.payload
|
||||
: null;
|
||||
if (info?.status !== "server_info") return;
|
||||
if (info.serverId !== row.serverId) {
|
||||
route.close(
|
||||
"server_id_mismatch",
|
||||
"Host identity changed; explicit replacement required",
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (info.features?.connectedHosts !== true) {
|
||||
route.close("unsupported_capability", "Update the destination daemon");
|
||||
return;
|
||||
}
|
||||
verified = true;
|
||||
clearTimeout(timer);
|
||||
} catch {
|
||||
route.close();
|
||||
return;
|
||||
}
|
||||
}
|
||||
input.frame(binary ? (data as Uint8Array) : String(data));
|
||||
});
|
||||
await ready;
|
||||
if (closed) throw new HostError("host_offline", "Host closed during connection");
|
||||
return {
|
||||
close: () => route.close(),
|
||||
send: (data) => {
|
||||
const binary = typeof data !== "string";
|
||||
if (closed) throw new HostError("host_offline", "Host is offline");
|
||||
if (!helloSent) {
|
||||
if (binary) throw new HostError("invalid_message", "Expected hello");
|
||||
const hello = JSON.parse(data as string) as WSHelloMessage;
|
||||
if (hello.type !== "hello") throw new HostError("invalid_message", "Expected hello");
|
||||
const clientId = `route-${createHash("sha256")
|
||||
.update(
|
||||
JSON.stringify([
|
||||
this.self.serverId,
|
||||
input.sourceId,
|
||||
input.originSessionId,
|
||||
input.scopeId,
|
||||
]),
|
||||
)
|
||||
.digest("hex")}`;
|
||||
transport.send(
|
||||
JSON.stringify({
|
||||
...hello,
|
||||
clientId,
|
||||
routed: true,
|
||||
producer: { serverId: this.self.serverId, sourceId: input.sourceId },
|
||||
sessionPermissions: input.permissions,
|
||||
auth: undefined,
|
||||
}),
|
||||
);
|
||||
helloSent = true;
|
||||
return;
|
||||
}
|
||||
if (!verified) throw new HostError("host_offline", "Host identity has not been verified");
|
||||
try {
|
||||
transport.send(data);
|
||||
} catch {
|
||||
route.close("resource_limit", "Host buffer limit reached or transport unavailable");
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
async close(): Promise<void> {
|
||||
this.closed = true;
|
||||
await this.ready;
|
||||
await this.queue;
|
||||
for (const id of this.live.keys()) this.stop(id);
|
||||
for (const route of this.routes) route.close();
|
||||
this.listeners.clear();
|
||||
}
|
||||
}
|
||||
|
||||
function connectionFailure(error: unknown, authenticationFailed: boolean): HostSummary["error"] {
|
||||
if (error instanceof HostError) return { code: error.code, message: error.message };
|
||||
if (error instanceof Error && "code" in error && error.code === "authentication_failed")
|
||||
return { code: "authentication_failed", message: error.message };
|
||||
if (authenticationFailed)
|
||||
return {
|
||||
code: "authentication_failed",
|
||||
message: "Host rejected the credentials. Replace the connection credentials.",
|
||||
};
|
||||
return null;
|
||||
}
|
||||
@@ -0,0 +1,210 @@
|
||||
import type { SessionDelivery } from "../session/owned-subscriptions/index.js";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { createHash } from "node:crypto";
|
||||
import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
|
||||
import type {
|
||||
DaemonPermission,
|
||||
SessionInboundMessage,
|
||||
SessionOutboundMessage,
|
||||
} from "@getpaseo/protocol/messages";
|
||||
|
||||
/** A logical socket has no network resources or authority of its own. */
|
||||
export class VirtualHostSocket extends EventEmitter {
|
||||
readyState = 1;
|
||||
constructor(
|
||||
private readonly output: (data: string | Uint8Array) => Promise<void>,
|
||||
private readonly ended: (code: number, reason: string) => void,
|
||||
private readonly buffered: () => number,
|
||||
private readonly transformHello: (data: string) => string,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
get bufferedAmount(): number {
|
||||
return this.buffered();
|
||||
}
|
||||
private helloPending = true;
|
||||
feed(data: string | Uint8Array): void {
|
||||
if (this.readyState !== 1) return;
|
||||
if (this.helloPending) {
|
||||
if (typeof data !== "string") {
|
||||
this.close(1008, "Expected hello");
|
||||
return;
|
||||
}
|
||||
try {
|
||||
data = this.transformHello(data);
|
||||
} catch {
|
||||
this.close(1008, "Invalid hello");
|
||||
return;
|
||||
}
|
||||
this.helloPending = false;
|
||||
}
|
||||
this.emit("message", typeof data === "string" ? data : Buffer.from(data));
|
||||
}
|
||||
async send(data: string | Uint8Array | ArrayBuffer): Promise<void> {
|
||||
if (this.readyState !== 1) throw new Error("Host channel closed");
|
||||
await this.output(
|
||||
typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data),
|
||||
);
|
||||
}
|
||||
close(code = 1000, reason = "Host channel closed"): void {
|
||||
if (this.readyState !== 1) return;
|
||||
this.readyState = 3;
|
||||
this.ended(code, reason);
|
||||
this.emit("close", 1000, "Host channel closed");
|
||||
}
|
||||
}
|
||||
|
||||
/** One authenticated physical source owns all its channels and their admission ceiling. */
|
||||
export class DaemonLinkEndpoint {
|
||||
private readonly channels = new Map<string, { scopeId: string; socket: VirtualHostSocket }>();
|
||||
private closed = false;
|
||||
constructor(
|
||||
private readonly parent: {
|
||||
identity: string;
|
||||
permissions(): readonly DaemonPermission[];
|
||||
routed(): boolean;
|
||||
bufferedAmount(): number | null;
|
||||
delivery: SessionDelivery;
|
||||
admit(socket: VirtualHostSocket, permissions: readonly DaemonPermission[]): Promise<void>;
|
||||
},
|
||||
) {}
|
||||
handles(message: SessionInboundMessage): boolean {
|
||||
return (
|
||||
message.type === "hosts.link.open.request" ||
|
||||
(message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) ||
|
||||
(message.type === "hosts.route.close.request" &&
|
||||
((!!message.routeId && this.channels.has(message.routeId)) ||
|
||||
[...this.channels.values()].some((channel) => channel.scopeId === message.scopeId)))
|
||||
);
|
||||
}
|
||||
async dispatch(message: SessionInboundMessage): Promise<void> {
|
||||
if (message.type === "hosts.link.open.request") {
|
||||
const error = (code: string, text: string) =>
|
||||
this.parent.delivery.reply({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: message.requestId,
|
||||
requestType: message.type,
|
||||
code,
|
||||
error: text,
|
||||
},
|
||||
});
|
||||
if (this.closed || this.parent.routed()) {
|
||||
error("unsupported_capability", "Connected hosts support one hop");
|
||||
return;
|
||||
}
|
||||
if (this.channels.size >= 128) {
|
||||
error("resource_limit", "Too many host channels");
|
||||
return;
|
||||
}
|
||||
let socket: VirtualHostSocket;
|
||||
const owner = this.parent.delivery.begin("host-link", undefined, () => {
|
||||
this.channels.delete(owner.id);
|
||||
socket?.close();
|
||||
});
|
||||
const id = owner.id;
|
||||
// Capture actual authenticated authority. The caller hello can only reduce it.
|
||||
const ceiling = [...this.parent.permissions()];
|
||||
socket = new VirtualHostSocket(
|
||||
async (data) => {
|
||||
const output: SessionOutboundMessage | null =
|
||||
typeof data === "string"
|
||||
? {
|
||||
type: "hosts.route.frame",
|
||||
payload: { subscriptionId: id, data, binary: false },
|
||||
}
|
||||
: null;
|
||||
const frame = typeof data === "string" ? null : encodeHostRouteFrame(id, data);
|
||||
const bytes = output
|
||||
? Buffer.byteLength(JSON.stringify({ type: "session", message: output }))
|
||||
: frame!.byteLength;
|
||||
const buffered = this.parent.bufferedAmount();
|
||||
// Account for the exact escaped outer frame, retaining control-frame headroom.
|
||||
if (buffered === null || buffered + bytes > 64 * 1024 * 1024 - 65536) {
|
||||
socket.close(1009, "Host channel output limit reached");
|
||||
throw new Error("Host channel output limit reached");
|
||||
}
|
||||
try {
|
||||
if (output) await owner.emitAndWait(output);
|
||||
else await owner.emitBinaryAndWait(frame!);
|
||||
} catch (failure) {
|
||||
socket.close();
|
||||
throw failure;
|
||||
}
|
||||
},
|
||||
(code, reason) => {
|
||||
this.channels.delete(id);
|
||||
let failure = "host_offline";
|
||||
if (code === 1009) failure = "resource_limit";
|
||||
if (code === 1008) failure = "permission_denied";
|
||||
owner.emit({
|
||||
type: "hosts.route.closed",
|
||||
payload: {
|
||||
subscriptionId: id,
|
||||
code: failure,
|
||||
message: reason,
|
||||
},
|
||||
});
|
||||
void owner.release();
|
||||
},
|
||||
() => this.parent.bufferedAmount() ?? 0,
|
||||
(data) => {
|
||||
const hello = JSON.parse(data);
|
||||
if (hello.type !== "hello" || typeof hello.clientId !== "string")
|
||||
throw new Error("Expected hello");
|
||||
return JSON.stringify({
|
||||
...hello,
|
||||
auth: undefined,
|
||||
routed: true,
|
||||
clientId: `link-${createHash("sha256")
|
||||
.update(JSON.stringify([this.parent.identity, hello.clientId]))
|
||||
.digest("hex")}`,
|
||||
});
|
||||
},
|
||||
);
|
||||
this.channels.set(id, { scopeId: message.channelId, socket });
|
||||
try {
|
||||
await this.parent.admit(socket, ceiling);
|
||||
if (this.closed || socket.readyState !== 1) {
|
||||
socket.close();
|
||||
return;
|
||||
}
|
||||
this.parent.delivery.reply({
|
||||
type: "hosts.link.open.response",
|
||||
payload: { requestId: message.requestId, routeId: id },
|
||||
});
|
||||
} catch {
|
||||
socket.close();
|
||||
error("host_offline", "Could not admit host channel");
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) {
|
||||
this.channels.get(message.routeId)!.socket.feed(message.data);
|
||||
return;
|
||||
}
|
||||
if (message.type === "hosts.route.close.request") {
|
||||
for (const [id, channel] of this.channels) {
|
||||
if (message.routeId === id || (message.scopeId && message.scopeId === channel.scopeId))
|
||||
channel.socket.close();
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
binary(frame: HostRouteFrame): boolean {
|
||||
const socket = this.channels.get(frame.routeId)?.socket;
|
||||
if (!socket) return false;
|
||||
socket.feed(frame.payload);
|
||||
return true;
|
||||
}
|
||||
close(reason: "host_offline" | "permission_denied" = "host_offline"): void {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
for (const { socket } of this.channels.values())
|
||||
socket.close(
|
||||
reason === "permission_denied" ? 1008 : 1000,
|
||||
"Physical link authority or connection changed",
|
||||
);
|
||||
this.channels.clear();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,252 @@
|
||||
import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
|
||||
import type {
|
||||
SessionInboundMessage,
|
||||
SessionOutboundMessage,
|
||||
DaemonPermission,
|
||||
} from "@getpaseo/protocol/messages";
|
||||
import { ConnectedHosts, HostError, type HostActor } from "./index.js";
|
||||
import type { SessionDelivery } from "../session/owned-subscriptions/index.js";
|
||||
|
||||
type HostRequest<T extends SessionInboundMessage["type"]> = Extract<
|
||||
SessionInboundMessage,
|
||||
{ type: T }
|
||||
>;
|
||||
interface SessionRoute {
|
||||
scopeId: string;
|
||||
source: object;
|
||||
send(data: string | Uint8Array): void;
|
||||
release(): Promise<void>;
|
||||
fail(code: string, message: string): void;
|
||||
}
|
||||
|
||||
/** Adapts host capabilities to source-owned session delivery. */
|
||||
export class ConnectedHostSession {
|
||||
private readonly routes = new Map<string, SessionRoute>();
|
||||
constructor(
|
||||
private readonly hosts: ConnectedHosts,
|
||||
private readonly delivery: SessionDelivery,
|
||||
private readonly actor: HostActor,
|
||||
private readonly permissions: () => DaemonPermission[],
|
||||
private readonly emit: (message: SessionOutboundMessage) => void,
|
||||
private readonly routed: () => boolean,
|
||||
private readonly originSessionId: string,
|
||||
private readonly bufferedAmount: (source: object) => number | null,
|
||||
) {}
|
||||
binary(frame: HostRouteFrame, source: object): void {
|
||||
const route = this.routes.get(frame.routeId);
|
||||
if (!route || route.source !== source) return;
|
||||
try {
|
||||
route.send(frame.payload);
|
||||
} catch {
|
||||
route.fail("resource_limit", "Host input buffer limit reached");
|
||||
}
|
||||
}
|
||||
setSource(sourceId: string): void {
|
||||
this.actor.sourceId = sourceId;
|
||||
}
|
||||
dispatch(message: SessionInboundMessage): Promise<void> | undefined {
|
||||
if (!message.type.startsWith("hosts.")) return;
|
||||
return this.handle(message);
|
||||
}
|
||||
revoke(): void {
|
||||
for (const route of this.routes.values())
|
||||
route.fail(
|
||||
"permission_denied",
|
||||
"Caller authority changed. Reconnecting with current grants.",
|
||||
);
|
||||
void this.delivery.releaseFamily("host-route");
|
||||
}
|
||||
private async handle(message: SessionInboundMessage): Promise<void> {
|
||||
try {
|
||||
if (this.routed() && message.type.startsWith("hosts.route."))
|
||||
throw new HostError(
|
||||
"unsupported_capability",
|
||||
"Connected hosts support one hop. Connect to the managing daemon to select a host.",
|
||||
);
|
||||
await this.execute(message);
|
||||
} catch (error) {
|
||||
this.fail(message, error);
|
||||
}
|
||||
}
|
||||
private fail(message: SessionInboundMessage, error: unknown): void {
|
||||
let failure = new HostError(
|
||||
"host_error",
|
||||
"Host operation failed. Check the connection and retry.",
|
||||
);
|
||||
if (error instanceof HostError) failure = error;
|
||||
else if (error instanceof Error && "code" in error && error.code === "invalid_connection")
|
||||
failure = new HostError("invalid_connection", error.message);
|
||||
if (message.type === "hosts.route.frame.request") {
|
||||
const route = this.routes.get(message.routeId);
|
||||
if (route && route.source === this.delivery.currentSource)
|
||||
route.fail(failure.code, failure.message);
|
||||
return;
|
||||
}
|
||||
if (!("requestId" in message) || !message.requestId) return;
|
||||
this.emit({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: message.requestId,
|
||||
requestType: message.type,
|
||||
code: failure.code,
|
||||
error: failure.message,
|
||||
},
|
||||
});
|
||||
}
|
||||
private async execute(message: SessionInboundMessage): Promise<void> {
|
||||
const actor = { ...this.actor, manageSources: this.permissions().includes("access.manage") };
|
||||
switch (message.type) {
|
||||
case "hosts.list.request":
|
||||
return this.list(message);
|
||||
case "hosts.add.request":
|
||||
this.emit({
|
||||
type: "hosts.add.response",
|
||||
payload: {
|
||||
requestId: message.requestId,
|
||||
host: await this.hosts.add(message.input, actor),
|
||||
},
|
||||
});
|
||||
return;
|
||||
case "hosts.update.request":
|
||||
this.emit({
|
||||
type: "hosts.update.response",
|
||||
payload: {
|
||||
requestId: message.requestId,
|
||||
host: await this.hosts.update(message.registrationId, message.patch, actor),
|
||||
},
|
||||
});
|
||||
return;
|
||||
case "hosts.remove.request":
|
||||
await this.hosts.remove(message.registrationId, actor);
|
||||
this.emit({ type: "hosts.remove.response", payload: { requestId: message.requestId } });
|
||||
return;
|
||||
case "hosts.route.open.request":
|
||||
return this.open(message);
|
||||
case "hosts.route.frame.request": {
|
||||
const route = this.routes.get(message.routeId);
|
||||
if (!route || route.source !== this.delivery.currentSource)
|
||||
throw new HostError("host_unknown", "Unknown routed session");
|
||||
route.send(message.data);
|
||||
return;
|
||||
}
|
||||
case "hosts.route.close.request": {
|
||||
const route = message.routeId
|
||||
? this.routes.get(message.routeId)
|
||||
: [...this.routes.values()].find(
|
||||
(entry) =>
|
||||
entry.scopeId === message.scopeId && entry.source === this.delivery.currentSource,
|
||||
);
|
||||
if (route && route.source === this.delivery.currentSource) await route.release();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
private async list(message: HostRequest<"hosts.list.request">): Promise<void> {
|
||||
if (!message.observe) {
|
||||
this.emit({
|
||||
type: "hosts.list.response",
|
||||
payload: { requestId: message.requestId, directory: await this.hosts.list() },
|
||||
});
|
||||
return;
|
||||
}
|
||||
let stop = () => {};
|
||||
const owner = this.delivery.begin("hosts", undefined, () => stop());
|
||||
try {
|
||||
stop = this.hosts.observe(() => {
|
||||
void this.hosts
|
||||
.list()
|
||||
.then((directory) => {
|
||||
owner.emit({ type: "hosts.changed", payload: { subscriptionId: owner.id, directory } });
|
||||
return undefined;
|
||||
})
|
||||
.catch(() => owner.release());
|
||||
});
|
||||
this.emit({
|
||||
type: "hosts.list.response",
|
||||
payload: {
|
||||
requestId: message.requestId,
|
||||
subscriptionId: owner.id,
|
||||
directory: await this.hosts.list(),
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
await owner.release();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
private async open(message: HostRequest<"hosts.route.open.request">): Promise<void> {
|
||||
if (this.delivery.subscriptionIds(this.delivery.currentSource!, "host-route").length >= 32)
|
||||
throw new HostError("resource_limit", "Too many routed sessions for this caller");
|
||||
let close = () => {};
|
||||
const owner = this.delivery.begin("host-route", undefined, () => {
|
||||
this.routes.delete(owner.id);
|
||||
close();
|
||||
});
|
||||
const fail = (code: string, text: string) => {
|
||||
owner.emit({
|
||||
type: "hosts.route.closed",
|
||||
payload: { subscriptionId: owner.id, code, message: text },
|
||||
});
|
||||
void owner.release();
|
||||
};
|
||||
this.routes.set(owner.id, {
|
||||
scopeId: message.scopeId,
|
||||
source: owner.source,
|
||||
send: () => {
|
||||
throw new HostError("host_offline", "Host is connecting");
|
||||
},
|
||||
release: owner.release,
|
||||
fail,
|
||||
});
|
||||
try {
|
||||
const route = await this.hosts.openRoute({
|
||||
serverId: message.serverId,
|
||||
scopeId: message.scopeId,
|
||||
sourceId: this.actor.sourceId,
|
||||
originSessionId: this.originSessionId,
|
||||
permissions: this.permissions(),
|
||||
signal: owner.signal,
|
||||
frame: (data) => {
|
||||
const output: SessionOutboundMessage | null =
|
||||
typeof data === "string"
|
||||
? {
|
||||
type: "hosts.route.frame",
|
||||
payload: { subscriptionId: owner.id, data, binary: false },
|
||||
}
|
||||
: null;
|
||||
const binary = typeof data === "string" ? null : encodeHostRouteFrame(owner.id, data);
|
||||
const size = output
|
||||
? Buffer.byteLength(JSON.stringify({ type: "session", message: output }))
|
||||
: binary!.byteLength;
|
||||
const buffered = this.bufferedAmount(owner.source);
|
||||
if (buffered === null || buffered + size > 32 * 1024 * 1024) {
|
||||
fail("resource_limit", "The receiving client cannot keep up with host output");
|
||||
return;
|
||||
}
|
||||
if (output) owner.emit(output);
|
||||
else owner.emitBinary(binary!);
|
||||
},
|
||||
ended: fail,
|
||||
});
|
||||
close = route.close;
|
||||
if (owner.signal.aborted) {
|
||||
route.close();
|
||||
return;
|
||||
}
|
||||
this.routes.set(owner.id, {
|
||||
scopeId: message.scopeId,
|
||||
source: owner.source,
|
||||
send: route.send,
|
||||
release: owner.release,
|
||||
fail,
|
||||
});
|
||||
this.emit({
|
||||
type: "hosts.route.open.response",
|
||||
payload: { requestId: message.requestId, routeId: owner.id },
|
||||
});
|
||||
} catch (error) {
|
||||
await owner.release();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,14 @@
|
||||
import { createPaseoApi, type PaseoApi } from "@getpaseo/client";
|
||||
import { createHostClient } from "@getpaseo/client/node";
|
||||
import { hashSync } from "bcryptjs";
|
||||
import { once } from "node:events";
|
||||
import {
|
||||
decodeTerminalStreamFrame,
|
||||
encodeTerminalStreamFrame,
|
||||
TerminalStreamOpcode,
|
||||
} from "@getpaseo/protocol/binary-frames/index";
|
||||
import type { DaemonClient as HostClient } from "@getpaseo/client/internal/daemon-client";
|
||||
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { withTimeout } from "../../utils/promise-timeout.js";
|
||||
import { DaemonClient } from "../test-utils/daemon-client.js";
|
||||
@@ -194,9 +205,9 @@ async function waitForCapturedLog(
|
||||
let relayProcess: ChildProcess | null = null;
|
||||
let relayStdoutLines: string[] = [];
|
||||
|
||||
const startRelay = async (options: { useLocalRelay?: boolean } = {}) => {
|
||||
const startRelay = async (options: { useLocalRelay?: boolean; port?: number } = {}) => {
|
||||
relayStdoutLines = [];
|
||||
relayPort = await getAvailablePort();
|
||||
relayPort = options.port ?? (await getAvailablePort());
|
||||
const relayDir = path.resolve(process.cwd(), "../relay");
|
||||
const relayArgs = [
|
||||
"wrangler",
|
||||
@@ -255,10 +266,152 @@ async function waitForCapturedLog(
|
||||
|
||||
const stopRelay = async () => {
|
||||
if (!relayProcess) return;
|
||||
relayProcess.kill("SIGTERM");
|
||||
const process = relayProcess;
|
||||
relayProcess = null;
|
||||
const exited = once(process, "exit");
|
||||
process.kill("SIGTERM");
|
||||
await withTimeout(exited, 10000, "Local relay did not stop");
|
||||
};
|
||||
|
||||
test("connected hosts route RPC and binary traffic through local E2EE relay and recover scopes", async () => {
|
||||
let a: Awaited<ReturnType<typeof createTestPaseoDaemon>> | undefined;
|
||||
let b: Awaited<ReturnType<typeof createTestPaseoDaemon>> | undefined;
|
||||
let origin: ReturnType<typeof createHostClient> | undefined;
|
||||
let api: PaseoApi | undefined;
|
||||
let remote: PaseoApi | undefined;
|
||||
let sibling: PaseoApi | undefined;
|
||||
let terminalRoute: Awaited<ReturnType<typeof openConnectedHostTerminal>> | undefined;
|
||||
let wire: Awaited<ReturnType<typeof relayWire>> | undefined;
|
||||
try {
|
||||
await startRelay({ useLocalRelay: true });
|
||||
const port = relayPort;
|
||||
b = await createTestPaseoDaemon({
|
||||
relayEnabled: true,
|
||||
relayEndpoint: `127.0.0.1:${port}`,
|
||||
relayUseTls: false,
|
||||
relayPublicUseTls: false,
|
||||
auth: { password: hashSync("relay-password", 4) },
|
||||
});
|
||||
a = await createTestPaseoDaemon();
|
||||
origin = createHostClient(`127.0.0.1:${a.port}`, { clientId: crypto.randomUUID() });
|
||||
await origin.connect();
|
||||
api = createPaseoApi(origin);
|
||||
const offerUrl = await getPairingOfferUrl({
|
||||
paseoHome: b.paseoHome,
|
||||
relayEnabled: true,
|
||||
relayEndpoint: b.config.relayEndpoint,
|
||||
});
|
||||
const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
|
||||
wire = await relayWire(port);
|
||||
const connection = new URL(`relay://127.0.0.1:${wire.port}/${serverId}`);
|
||||
connection.searchParams.set("key", daemonPublicKeyB64);
|
||||
connection.searchParams.set("password", "wrong-password");
|
||||
const saved = await api.hosts.add({ connection: connection.toString() });
|
||||
await expect
|
||||
.poll(async () => (await api!.hosts.list()).hosts[0]?.error?.code, { timeout: 20000 })
|
||||
.toBe("authentication_failed");
|
||||
expect(JSON.stringify(await api.hosts.list())).not.toMatch(/wrong-password|relay-password/);
|
||||
connection.searchParams.set("password", "relay-password");
|
||||
await api.hosts.update(saved.registrationId, { connection: connection.toString() });
|
||||
await expect
|
||||
.poll(async () => (await api!.hosts.list()).hosts[0]?.status, { timeout: 20000 })
|
||||
.toBe("online");
|
||||
const admittedConnections = wire.connections();
|
||||
expect(wire.active()).toBe(1);
|
||||
remote = await api.hosts.client(serverId);
|
||||
sibling = await api.hosts.client(serverId);
|
||||
expect(wire.connections()).toBe(admittedConnections);
|
||||
expect(wire.active()).toBe(1);
|
||||
expect((await remote.workspaces.list()).entries).toEqual([]);
|
||||
const observed = await remote.workspaces.list({ subscribe: {} });
|
||||
let snapshots = 0;
|
||||
observed.subscription.subscribe({
|
||||
snapshot: () => {
|
||||
snapshots++;
|
||||
},
|
||||
update() {},
|
||||
});
|
||||
const workspace = await remote.workspaces.create({
|
||||
source: { kind: "directory", path: b.staticDir },
|
||||
title: "Relayed workspace",
|
||||
});
|
||||
const terminal = await workspace.terminals.create({
|
||||
command: process.execPath,
|
||||
args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
|
||||
});
|
||||
terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id);
|
||||
expect(wire.connections()).toBe(admittedConnections);
|
||||
terminalRoute.write("ENCRYPTED_BINARY_BEFORE\n");
|
||||
await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_BEFORE");
|
||||
terminalRoute.close();
|
||||
terminalRoute = undefined;
|
||||
const before = snapshots;
|
||||
await stopRelay();
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
try {
|
||||
await remote!.workspaces.list();
|
||||
return "online";
|
||||
} catch (error) {
|
||||
return (error as { code: string }).code;
|
||||
}
|
||||
},
|
||||
{ timeout: 15000 },
|
||||
)
|
||||
.toBe("DAEMON_CONNECTION_LOST");
|
||||
await startRelay({ useLocalRelay: true, port });
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
try {
|
||||
return (await remote!.workspaces.list()).entries.length;
|
||||
} catch {
|
||||
return -1;
|
||||
}
|
||||
},
|
||||
{ timeout: 30000 },
|
||||
)
|
||||
.toBe(1);
|
||||
await expect.poll(() => snapshots, { timeout: 15000 }).toBeGreaterThan(before);
|
||||
expect(wire.active()).toBe(1);
|
||||
terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id);
|
||||
terminalRoute.write("ENCRYPTED_BINARY_AFTER\n");
|
||||
await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_AFTER");
|
||||
terminalRoute.close();
|
||||
terminalRoute = undefined;
|
||||
await remote.dispose();
|
||||
remote = undefined;
|
||||
expect((await sibling.workspaces.list()).entries).toHaveLength(1);
|
||||
await sibling.terminals.ref(terminal.id).kill();
|
||||
await api.dispose();
|
||||
await expect(sibling.workspaces.list()).rejects.toThrow();
|
||||
const direct = createHostClient(`127.0.0.1:${b.port}`, {
|
||||
clientId: crypto.randomUUID(),
|
||||
password: "relay-password",
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
try {
|
||||
await direct.connect();
|
||||
const directApi = createPaseoApi(direct);
|
||||
expect((await directApi.workspaces.list()).entries).toHaveLength(1);
|
||||
await directApi.dispose();
|
||||
} finally {
|
||||
await direct.close();
|
||||
}
|
||||
} finally {
|
||||
terminalRoute?.close();
|
||||
await remote?.dispose();
|
||||
await sibling?.dispose();
|
||||
await api?.dispose();
|
||||
await origin?.close();
|
||||
await a?.close();
|
||||
await b?.close();
|
||||
await wire?.close();
|
||||
await stopRelay();
|
||||
}
|
||||
}, 120000);
|
||||
|
||||
test("oversized attachments fail without disconnecting or resubmitting over the relay", async () => {
|
||||
await startRelay({ useLocalRelay: true });
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
@@ -932,3 +1085,112 @@ async function waitForCapturedLog(
|
||||
}
|
||||
}, 90000);
|
||||
});
|
||||
|
||||
async function openConnectedHostTerminal(origin: HostClient, serverId: string, terminalId: string) {
|
||||
const routeId = await origin.openHostRoute(serverId, crypto.randomUUID());
|
||||
const messages: SessionOutboundMessage[] = [];
|
||||
let output = "";
|
||||
const detachBinary = origin.subscribeHostFrames((routed) => {
|
||||
if (routed.routeId !== routeId) return;
|
||||
const frame = decodeTerminalStreamFrame(routed.payload);
|
||||
if (frame?.opcode === TerminalStreamOpcode.Output)
|
||||
output += new TextDecoder().decode(frame.payload);
|
||||
});
|
||||
const detach = origin.subscribeRawMessages((message) => {
|
||||
if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId) return;
|
||||
const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data));
|
||||
if (frame.type === "session") messages.push(frame.message);
|
||||
});
|
||||
const close = () => {
|
||||
detach();
|
||||
detachBinary();
|
||||
origin.closeHostRoute(routeId);
|
||||
};
|
||||
try {
|
||||
origin.sendHostFrame(
|
||||
routeId,
|
||||
JSON.stringify({
|
||||
type: "hello",
|
||||
clientId: "relay-binary",
|
||||
clientType: "cli",
|
||||
protocolVersion: 1,
|
||||
}),
|
||||
);
|
||||
await expect
|
||||
.poll(() =>
|
||||
messages.some(
|
||||
(message) => message.type === "status" && message.payload.status === "server_info",
|
||||
),
|
||||
)
|
||||
.toBe(true);
|
||||
origin.sendHostFrame(
|
||||
routeId,
|
||||
JSON.stringify({
|
||||
type: "session",
|
||||
message: { type: "subscribe_terminal_request", terminalId, requestId: "binary" },
|
||||
}),
|
||||
);
|
||||
await expect
|
||||
.poll(() => messages.some((message) => message.type === "subscribe_terminal_response"))
|
||||
.toBe(true);
|
||||
const response = messages.find((message) => message.type === "subscribe_terminal_response");
|
||||
if (response?.type !== "subscribe_terminal_response" || response.payload.slot == null)
|
||||
throw new Error("Missing binary stream slot");
|
||||
const slot = response.payload.slot;
|
||||
return {
|
||||
close,
|
||||
output: () => output,
|
||||
write: (payload: string) =>
|
||||
origin.sendHostFrame(
|
||||
routeId,
|
||||
encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }),
|
||||
),
|
||||
};
|
||||
} catch (error) {
|
||||
close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function relayWire(relayPort: number) {
|
||||
const sockets = new Set<net.Socket>();
|
||||
let connections = 0;
|
||||
let active = 0;
|
||||
const server = net.createServer((source) => {
|
||||
const target = net.connect(relayPort, "127.0.0.1");
|
||||
sockets.add(source);
|
||||
sockets.add(target);
|
||||
let counted = false;
|
||||
target.on("connect", () => {
|
||||
counted = true;
|
||||
connections++;
|
||||
active++;
|
||||
});
|
||||
target.on("error", () => source.destroy());
|
||||
source.on("error", () => target.destroy());
|
||||
target.on("close", () => {
|
||||
if (counted) active--;
|
||||
sockets.delete(target);
|
||||
source.destroy();
|
||||
});
|
||||
source.on("close", () => {
|
||||
sockets.delete(source);
|
||||
target.destroy();
|
||||
});
|
||||
source.pipe(target);
|
||||
target.pipe(source);
|
||||
});
|
||||
server.listen(0, "127.0.0.1");
|
||||
await once(server, "listening");
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing relay wire port");
|
||||
return {
|
||||
port: address.port,
|
||||
connections: () => connections,
|
||||
active: () => active,
|
||||
close: async () => {
|
||||
for (const socket of sockets) socket.destroy();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -288,6 +288,7 @@ vi.mock("./worktree-bootstrap.js", async (importOriginal) => {
|
||||
});
|
||||
|
||||
interface SessionForTestOptions {
|
||||
sendToSource?: SessionOptions["onMessageToSource"];
|
||||
clientId?: string;
|
||||
permissions?: readonly DaemonPermission[];
|
||||
agentManager?: { [K in keyof SessionOptions["agentManager"]]?: unknown };
|
||||
@@ -441,6 +442,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
|
||||
daemonRuntimeConfig: options.daemonRuntimeConfig,
|
||||
permissions: options.permissions ?? OWNER_PERMISSIONS,
|
||||
};
|
||||
if (options.sendToSource) sessionOptions.onMessageToSource = options.sendToSource;
|
||||
return new Session(sessionOptions);
|
||||
}
|
||||
|
||||
@@ -5813,3 +5815,40 @@ test("provider snapshots preserve versionless visibility while capabilities upda
|
||||
]);
|
||||
expect(references.compactSnapshot!.entries[0]!.modes![0]!.icon).toBe("ShieldCheck");
|
||||
});
|
||||
|
||||
test("a failing direct source observes fire-and-forget file replies without unhandled rejections", async () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "session-send-failure-"));
|
||||
writeFileSync(join(root, "readme.txt"), "source reply");
|
||||
let attempts = 0;
|
||||
const unhandled: unknown[] = [];
|
||||
const collect = (error: unknown) => unhandled.push(error);
|
||||
process.on("unhandledRejection", collect);
|
||||
const session = createSessionForTest({
|
||||
sendToSource: async () => {
|
||||
attempts++;
|
||||
throw new Error("Physical source closed during send");
|
||||
},
|
||||
});
|
||||
try {
|
||||
const source = {};
|
||||
session.updateClientCapabilities(null, source);
|
||||
await session.handleMessage(
|
||||
{
|
||||
type: "file_explorer_request",
|
||||
requestId: "direct-file",
|
||||
cwd: root,
|
||||
path: "readme.txt",
|
||||
mode: "file",
|
||||
acceptBinary: false,
|
||||
},
|
||||
source,
|
||||
);
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 30));
|
||||
expect(attempts).toBeGreaterThan(0);
|
||||
expect(unhandled).toEqual([]);
|
||||
} finally {
|
||||
process.off("unhandledRejection", collect);
|
||||
await session.cleanup();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import { AgentDirectory } from "./agent/directory.js";
|
||||
import { ConnectedHostSession } from "./connected-hosts/index.js";
|
||||
import type { ConnectedHosts } from "./connected-hosts/index.js";
|
||||
import { searchTimeline } from "./agent/chat-search/index.js";
|
||||
import type { BrowserToolsBroker } from "./browser-tools/broker.js";
|
||||
import { BrowserAutomationHostCapabilitySchema } from "@getpaseo/protocol/browser-automation/capabilities";
|
||||
@@ -434,16 +436,20 @@ const nodeSessionFileSystem: SessionFileSystem = {
|
||||
type AgentMcpTransportFactory = () => Promise<unknown>;
|
||||
|
||||
export interface SessionOptions {
|
||||
connectedHosts?: ConnectedHosts;
|
||||
sourceId?: string;
|
||||
routed?: boolean;
|
||||
browserToolsBroker?: BrowserToolsBroker | null;
|
||||
clientId: string;
|
||||
permissions: readonly DaemonPermission[];
|
||||
appVersion?: string | null;
|
||||
clientCapabilities?: Record<string, unknown> | null;
|
||||
onMessage: (msg: SessionOutboundMessage) => void;
|
||||
onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void;
|
||||
onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void | Promise<void>;
|
||||
onBinaryMessage?: (frame: Uint8Array) => void;
|
||||
onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise<void>;
|
||||
getTransportBufferedAmount?: (source?: object) => number | null;
|
||||
getConnectedHostBufferedAmount?: () => number | null;
|
||||
onLifecycleIntent?: (intent: SessionLifecycleIntent) => void;
|
||||
onWorkspaceRecovered?: (workspace: PersistedWorkspaceRecord) => Promise<void>;
|
||||
publishScriptStatusUpdate?: (message: ScriptStatusUpdateMessage) => void;
|
||||
@@ -666,17 +672,31 @@ interface ClientActivity {
|
||||
}
|
||||
|
||||
export class Session {
|
||||
private readonly connectedHostSession: ConnectedHostSession | null;
|
||||
private routed = false;
|
||||
isRouted(): boolean {
|
||||
return this.routed;
|
||||
}
|
||||
getHostSourceId(): string {
|
||||
return this.hostSourceId;
|
||||
}
|
||||
private hostSourceId: string;
|
||||
setHostSource(sourceId: string): void {
|
||||
this.hostSourceId = sourceId;
|
||||
this.connectedHostSession?.setSource(sourceId);
|
||||
}
|
||||
setRouted(routed: boolean): void {
|
||||
this.routed = routed;
|
||||
}
|
||||
readonly delivery = new SessionDelivery(
|
||||
(source, message) => {
|
||||
if (!this.authorization.allowsOutbound(message)) return;
|
||||
if (this.onMessageToSource) this.onMessageToSource(source, message);
|
||||
else this.onMessage(message);
|
||||
if (this.onMessageToSource) return this.onMessageToSource(source, message);
|
||||
this.onMessage(message);
|
||||
},
|
||||
(source, frame) => {
|
||||
if (this.onBinaryMessageToSource) {
|
||||
void this.onBinaryMessageToSource(source, frame).catch((error) =>
|
||||
this.sessionLogger.warn({ err: error }, "Failed to emit binary frame"),
|
||||
);
|
||||
return this.onBinaryMessageToSource(source, frame);
|
||||
} else this.emitBinary(frame);
|
||||
},
|
||||
(source, message) => this.workspaceSetupMessageForClient(message, source),
|
||||
@@ -698,7 +718,7 @@ export class Session {
|
||||
private readonly sessionId: string;
|
||||
private readonly onMessage: (msg: SessionOutboundMessage) => void;
|
||||
private readonly onMessageToSource:
|
||||
| ((source: object, msg: SessionOutboundMessage) => void)
|
||||
| ((source: object, msg: SessionOutboundMessage) => void | Promise<void>)
|
||||
| null;
|
||||
private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null;
|
||||
private readonly onBinaryMessageToSource:
|
||||
@@ -856,13 +876,20 @@ export class Session {
|
||||
this.browserToolsBroker = options.browserToolsBroker;
|
||||
this.clientId = clientId;
|
||||
this.authorization = new SessionAuthorization(permissions);
|
||||
|
||||
this.appVersion = appVersion ?? null;
|
||||
this.clientCapabilities = parseClientCapabilities(clientCapabilities);
|
||||
this.sessionId = uuidv4();
|
||||
this.hostSourceId = hostSource(options);
|
||||
this.connectedHostSession = createHostSession(options, this);
|
||||
this.onMessage = onMessage;
|
||||
this.onMessageToSource = onMessageToSource ?? null;
|
||||
this.onMessageToSource = onMessageToSource
|
||||
? (source, message) => this.observeSourceSend(() => onMessageToSource(source, message))
|
||||
: null;
|
||||
this.onBinaryMessage = onBinaryMessage ?? null;
|
||||
this.onBinaryMessageToSource = onBinaryMessageToSource ?? null;
|
||||
this.onBinaryMessageToSource = onBinaryMessageToSource
|
||||
? (source, frame) => this.observeSourceSend(() => onBinaryMessageToSource(source, frame))
|
||||
: null;
|
||||
this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0);
|
||||
this.onLifecycleIntent = onLifecycleIntent ?? null;
|
||||
this.onWorkspaceRecovered = onWorkspaceRecovered ?? null;
|
||||
@@ -2165,7 +2192,7 @@ export class Session {
|
||||
requestId,
|
||||
requestType: msg.type,
|
||||
error: `Session is not authorized for ${msg.type}`,
|
||||
code: "access_denied",
|
||||
code: msg.type.startsWith("hosts.") ? "permission_denied" : "access_denied",
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -2212,6 +2239,7 @@ export class Session {
|
||||
|
||||
public setPermissions(permissions: readonly DaemonPermission[]): void {
|
||||
this.authorization.replacePermissions(permissions);
|
||||
this.connectedHostSession?.revoke();
|
||||
if (!this.authorization.allowsPermission("workspace.write")) {
|
||||
void this.delivery
|
||||
.releaseFamily("browser-host")
|
||||
@@ -2366,11 +2394,10 @@ export class Session {
|
||||
|
||||
private async dispatchInboundMessage(msg: SessionInboundMessage, source?: object): Promise<void> {
|
||||
const promise =
|
||||
this.connectedHostSession?.dispatch(msg) ??
|
||||
this.dispatchSubscriptionMessage(msg, source) ??
|
||||
this.dispatchVoiceAndControlMessage(msg) ??
|
||||
this.dispatchAgentRewindMessage(msg, source) ??
|
||||
this.dispatchAgentRelationshipMessage(msg) ??
|
||||
this.dispatchAgentTimelineMessage(msg, source) ??
|
||||
this.dispatchAgentObservationMessage(msg, source) ??
|
||||
this.dispatchHubExecutionMessage(msg) ??
|
||||
this.dispatchCreationMessage(msg, source) ??
|
||||
this.dispatchAgentLifecycleMessage(msg) ??
|
||||
@@ -2388,6 +2415,17 @@ export class Session {
|
||||
if (promise) await promise;
|
||||
}
|
||||
|
||||
private dispatchAgentObservationMessage(
|
||||
msg: SessionInboundMessage,
|
||||
source?: object,
|
||||
): Promise<void> | undefined {
|
||||
return (
|
||||
this.dispatchAgentRewindMessage(msg, source) ??
|
||||
this.dispatchAgentRelationshipMessage(msg) ??
|
||||
this.dispatchAgentTimelineMessage(msg, source)
|
||||
);
|
||||
}
|
||||
|
||||
private dispatchWorkspaceLifecycleMessage(msg: SessionInboundMessage): Promise<void> | undefined {
|
||||
return (
|
||||
this.dispatchWorkspaceStateMessage(msg) ??
|
||||
@@ -3169,6 +3207,10 @@ export class Session {
|
||||
}
|
||||
|
||||
public async handleBinaryFrame(binaryFrame: BinaryFrame, source: object): Promise<void> {
|
||||
if (binaryFrame.kind === "host_route") {
|
||||
this.connectedHostSession?.binary(binaryFrame.frame, source);
|
||||
return;
|
||||
}
|
||||
if (!this.authorization.allowsPermission("workspace.write")) {
|
||||
return;
|
||||
}
|
||||
@@ -8543,6 +8585,20 @@ export class Session {
|
||||
this.emitBinary(frame);
|
||||
}
|
||||
|
||||
private observeSourceSend(send: () => void | Promise<void>): Promise<void> {
|
||||
let pending: Promise<void>;
|
||||
try {
|
||||
pending = Promise.resolve(send());
|
||||
} catch (error) {
|
||||
pending = Promise.reject(error);
|
||||
}
|
||||
// Legacy producers fire and forget; owned streams await the same failure.
|
||||
void pending.catch((error) =>
|
||||
this.sessionLogger.warn({ err: error }, "Failed to emit source frame"),
|
||||
);
|
||||
return pending;
|
||||
}
|
||||
|
||||
private emitForSource(msg: SessionOutboundMessage, source?: object): void {
|
||||
if (source && this.onMessageToSource) {
|
||||
msg = this.delivery.authorizeReply(msg, source);
|
||||
@@ -8691,3 +8747,21 @@ function legacyWantsEvent(
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
function createHostSession(options: SessionOptions, session: Session): ConnectedHostSession | null {
|
||||
if (!options.connectedHosts) return null;
|
||||
return new ConnectedHostSession(
|
||||
options.connectedHosts,
|
||||
session.delivery,
|
||||
{ sourceId: options.sourceId ?? "owner", manageSources: false },
|
||||
() => session.getPermissions(),
|
||||
(message) => session.publish(message),
|
||||
() => session.isRouted(),
|
||||
session.getSessionId(),
|
||||
() => options.getConnectedHostBufferedAmount?.() ?? 0,
|
||||
);
|
||||
}
|
||||
|
||||
function hostSource(options: SessionOptions): string {
|
||||
return options.sourceId ?? "owner";
|
||||
}
|
||||
|
||||
@@ -169,3 +169,40 @@ test("failed teardown still closes the delivery and rejects late work", async ()
|
||||
delivery.request(source, { type: "ping", requestId: "late" }, async () => {}),
|
||||
).rejects.toThrow("Session delivery is closed");
|
||||
});
|
||||
|
||||
test("owned send completion retains source proofs and reports physical failures to awaiting producers", async () => {
|
||||
const source = {};
|
||||
const failure = new Error("Physical send failed");
|
||||
const delivery = new SessionDelivery(
|
||||
async (socket, message) => {
|
||||
expect(socket).toBe(source);
|
||||
expect(delivery.permits(socket, message)).toBe(true);
|
||||
throw failure;
|
||||
},
|
||||
async (socket, frame) => {
|
||||
expect(socket).toBe(source);
|
||||
expect(delivery.permitsBinary(socket, frame)).toBe(true);
|
||||
throw failure;
|
||||
},
|
||||
);
|
||||
delivery.attach(source, true);
|
||||
await delivery.request(
|
||||
source,
|
||||
{ type: "hosts.link.open.request", requestId: "open", channelId: "scope" },
|
||||
async () => {
|
||||
const owner = delivery.begin("host-link", undefined, () => {});
|
||||
await expect(
|
||||
owner.emitAndWait({
|
||||
type: "hosts.route.frame",
|
||||
payload: { subscriptionId: owner.id, data: "text", binary: false },
|
||||
}),
|
||||
).rejects.toBe(failure);
|
||||
await expect(owner.emitBinaryAndWait(new Uint8Array([0, 255]))).rejects.toBe(failure);
|
||||
await owner.release();
|
||||
await expect(owner.emitBinaryAndWait(new Uint8Array([0]))).rejects.toThrow(
|
||||
"Subscription source is closed",
|
||||
);
|
||||
},
|
||||
);
|
||||
await delivery.close();
|
||||
});
|
||||
|
||||
@@ -51,6 +51,8 @@ export interface OwnedSubscription {
|
||||
readonly signal: AbortSignal;
|
||||
emit(message: SessionOutboundMessage): void;
|
||||
emitBinary(frame: Uint8Array): void;
|
||||
emitAndWait(message: SessionOutboundMessage): Promise<void>;
|
||||
emitBinaryAndWait(frame: Uint8Array): Promise<void>;
|
||||
release(): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -67,8 +69,14 @@ export class SessionDelivery {
|
||||
private closing?: Promise<void>;
|
||||
|
||||
constructor(
|
||||
private readonly send: (source: object, message: SessionOutboundMessage) => void,
|
||||
private readonly sendBinary: (source: object, frame: Uint8Array) => void = () => {},
|
||||
private readonly send: (
|
||||
source: object,
|
||||
message: SessionOutboundMessage,
|
||||
) => void | Promise<void>,
|
||||
private readonly sendBinary: (
|
||||
source: object,
|
||||
frame: Uint8Array,
|
||||
) => void | Promise<void> = () => {},
|
||||
private readonly project: (
|
||||
source: object,
|
||||
message: SessionOutboundMessage,
|
||||
@@ -319,7 +327,18 @@ export class SessionDelivery {
|
||||
emitBinary: (frame) => {
|
||||
if (!owner.active || !source.active) return;
|
||||
this.proofs.set(frame, owner);
|
||||
this.sendBinary(source.socket, frame);
|
||||
void Promise.resolve(this.sendBinary(source.socket, frame)).catch(() => {});
|
||||
},
|
||||
emitAndWait: async (message) => {
|
||||
if (!owner.active || !source.active) throw new Error("Subscription source is closed");
|
||||
const projected = this.project(source.socket, withSubscriptionId(message, responseId));
|
||||
this.proofs.set(projected, owner);
|
||||
await this.send(source.socket, projected);
|
||||
},
|
||||
emitBinaryAndWait: async (frame) => {
|
||||
if (!owner.active || !source.active) throw new Error("Subscription source is closed");
|
||||
this.proofs.set(frame, owner);
|
||||
await this.sendBinary(source.socket, frame);
|
||||
},
|
||||
release: () => this.releaseOwner(owner),
|
||||
};
|
||||
@@ -414,7 +433,7 @@ export class SessionDelivery {
|
||||
private sendOwned(owner: DeliveryOwner, message: SessionOutboundMessage): void {
|
||||
const projected = this.project(owner.source.socket, message);
|
||||
this.proofs.set(projected, owner);
|
||||
this.send(owner.source.socket, projected);
|
||||
void Promise.resolve(this.send(owner.source.socket, projected)).catch(() => {});
|
||||
}
|
||||
|
||||
private releaseOwner(owner: RetainedOwner): Promise<void> {
|
||||
|
||||
@@ -71,6 +71,18 @@ const sessionMock = vi.hoisted(() => {
|
||||
wantsSourceEvent = (source: object) => !this.delivery.isModern(source);
|
||||
getSessionId = vi.fn(() => "mock-session-id");
|
||||
getPermissions = vi.fn(() => this.args.permissions as string[]);
|
||||
setPermissions = (permissions: string[]) => {
|
||||
this.args.permissions = permissions;
|
||||
};
|
||||
private routed = false;
|
||||
setRouted = (routed: boolean) => {
|
||||
this.routed = routed;
|
||||
};
|
||||
isRouted = () => this.routed;
|
||||
setHostSource = (sourceId: string) => {
|
||||
this.args.sourceId = sourceId;
|
||||
};
|
||||
getHostSourceId = () => this.args.sourceId as string;
|
||||
allowsInbound = vi.fn(() => true);
|
||||
allowsPermission = vi.fn(() => true);
|
||||
publish = vi.fn((message: unknown) => {
|
||||
@@ -473,16 +485,12 @@ function holdNextSessionMessage(session: (typeof sessionMock.instances)[number])
|
||||
function holdSessionCleanup(session: (typeof sessionMock.instances)[number]): {
|
||||
finish: () => void;
|
||||
} {
|
||||
let finish = () => {};
|
||||
session.cleanup.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
finish = resolve;
|
||||
}),
|
||||
);
|
||||
return {
|
||||
finish: () => finish(),
|
||||
};
|
||||
let finish!: () => void;
|
||||
const pending = new Promise<void>((resolve) => {
|
||||
finish = resolve;
|
||||
});
|
||||
session.cleanup.mockImplementationOnce(() => pending);
|
||||
return { finish };
|
||||
}
|
||||
|
||||
describe("relay external socket reconnect behavior", () => {
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
import { createToolHosts } from "./agent/host-operations/index.js";
|
||||
import { HOST_ROUTE_OPCODE } from "@getpaseo/protocol/binary-frames/index";
|
||||
import { DaemonLinkEndpoint, VirtualHostSocket } from "./connected-hosts/index.js";
|
||||
import { ConnectedHosts } from "./connected-hosts/index.js";
|
||||
import { stat } from "node:fs/promises";
|
||||
import type { CreationSnapshot } from "@getpaseo/protocol/messages";
|
||||
import { CreationService } from "./creation/index.js";
|
||||
@@ -437,6 +441,8 @@ export interface WebSocketLike {
|
||||
}
|
||||
|
||||
interface SessionConnectionBase {
|
||||
hubExecutionAgents?: HubExecutionAgents;
|
||||
permissionCeiling?: readonly DaemonPermission[];
|
||||
session: Session;
|
||||
principalId: string;
|
||||
sessionKey: string;
|
||||
@@ -460,13 +466,14 @@ interface PluginSessionConnection extends SessionConnectionBase {
|
||||
type SessionConnection = ReconnectableSessionConnection | PluginSessionConnection;
|
||||
|
||||
interface SocketSessionOptions {
|
||||
sourceId?: string;
|
||||
clientId: string;
|
||||
appVersion: string | null;
|
||||
clientCapabilities: Record<string, unknown> | null;
|
||||
permissions: readonly DaemonPermission[];
|
||||
connectionLogger: pino.Logger;
|
||||
onMessage: (message: SessionOutboundMessage) => void;
|
||||
onMessageToSource?: (source: object, message: SessionOutboundMessage) => void;
|
||||
onMessageToSource?: (source: object, message: SessionOutboundMessage) => void | Promise<void>;
|
||||
onBinaryMessage?: (frame: Uint8Array) => void;
|
||||
onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise<void>;
|
||||
getTransportBufferedAmount?: (source?: object) => number | null;
|
||||
@@ -527,6 +534,7 @@ function requireWebSocketServices(params: {
|
||||
export class VoiceAssistantWebSocketServer {
|
||||
private readonly logger: pino.Logger;
|
||||
private readonly wss: WebSocketServer;
|
||||
private readonly hostLinks = new Map<WebSocketLike, DaemonLinkEndpoint>();
|
||||
private readonly pendingConnections: Map<WebSocketLike, PendingConnection> = new Map();
|
||||
private readonly sessions: Map<WebSocketLike, SessionConnection> = new Map();
|
||||
private readonly socketIdentities: Map<WebSocketLike, WebSocketConnectionIdentity> = new Map();
|
||||
@@ -550,6 +558,11 @@ export class VoiceAssistantWebSocketServer {
|
||||
private readonly workspaceAutoName: WorkspaceAutoName;
|
||||
private readonly downloadTokenStore: DownloadTokenStore;
|
||||
private readonly paseoHome: string;
|
||||
private readonly connectedHosts: ConnectedHosts;
|
||||
|
||||
toolHosts() {
|
||||
return createToolHosts(this.connectedHosts);
|
||||
}
|
||||
private readonly passwordHash: string | undefined;
|
||||
private readonly credentialSource: DaemonAuthConfig | undefined;
|
||||
private readonly worktreesRoot: string | undefined;
|
||||
@@ -701,6 +714,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.workspaceAutoName = workspaceAutoName;
|
||||
this.downloadTokenStore = downloadTokenStore;
|
||||
this.paseoHome = paseoHome;
|
||||
this.connectedHosts = new ConnectedHosts(paseoHome, { serverId, label: getHostName() });
|
||||
this.passwordHash = auth?.password;
|
||||
this.worktreesRoot = daemonRuntimeConfig?.worktreesRoot;
|
||||
this.daemonConfigStore = daemonConfigStore;
|
||||
@@ -1047,6 +1061,12 @@ export class VoiceAssistantWebSocketServer {
|
||||
principalId: string,
|
||||
permissions: readonly DaemonPermission[],
|
||||
): void {
|
||||
for (const [socket, link] of this.hostLinks) {
|
||||
if (this.sessions.get(socket)?.principalId === principalId) {
|
||||
link.close("permission_denied");
|
||||
this.hostLinks.delete(socket);
|
||||
}
|
||||
}
|
||||
for (const pending of this.pendingConnections.values()) {
|
||||
if (pending.admission?.principalId === principalId) {
|
||||
pending.admission = { ...pending.admission, permissions };
|
||||
@@ -1054,7 +1074,11 @@ export class VoiceAssistantWebSocketServer {
|
||||
}
|
||||
for (const connection of new Set(this.externalSessionsByKey.values())) {
|
||||
if (connection.principalId === principalId) {
|
||||
connection.session.setPermissions(permissions);
|
||||
connection.session.setPermissions(
|
||||
connection.permissionCeiling
|
||||
? permissions.filter((permission) => connection.permissionCeiling!.includes(permission))
|
||||
: permissions,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1071,6 +1095,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
|
||||
public async close(): Promise<void> {
|
||||
this.prepareForShutdown();
|
||||
await this.connectedHosts.close();
|
||||
this.unsubscribeSpeechReadiness?.();
|
||||
this.unsubscribeSpeechReadiness = null;
|
||||
this.unsubscribeDaemonConfigChange?.();
|
||||
@@ -1159,6 +1184,20 @@ export class VoiceAssistantWebSocketServer {
|
||||
message: WSOutboundMessage,
|
||||
reportRejection = false,
|
||||
): void {
|
||||
const prepared = this.prepareJsonDelivery(sockets, message, reportRejection);
|
||||
if (!prepared) return;
|
||||
for (const ws of prepared.sockets) {
|
||||
this.sendFrameToClient(ws, prepared.payload, prepared.bytes, () => {
|
||||
this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private prepareJsonDelivery(
|
||||
sockets: Iterable<WebSocketLike>,
|
||||
message: WSOutboundMessage,
|
||||
reportRejection: boolean,
|
||||
): { sockets: WebSocketLike[]; payload: string; bytes: number } | null {
|
||||
const writableSockets = [...sockets].filter((ws) => {
|
||||
if (
|
||||
message.type === "session" &&
|
||||
@@ -1180,7 +1219,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
return this.ensureOutboundCapacity(ws, 0);
|
||||
});
|
||||
if (writableSockets.length === 0) {
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
|
||||
let payload: string;
|
||||
@@ -1188,14 +1227,30 @@ export class VoiceAssistantWebSocketServer {
|
||||
payload = JSON.stringify(message);
|
||||
} catch (err) {
|
||||
this.logger.warn({ err }, "ws_serialize_failed");
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
|
||||
const payloadBytes = outboundFrameByteLength(payload);
|
||||
for (const ws of writableSockets) {
|
||||
this.sendFrameToClient(ws, payload, payloadBytes, () => {
|
||||
this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
|
||||
return { sockets: writableSockets, payload, bytes: outboundFrameByteLength(payload) };
|
||||
}
|
||||
|
||||
private async sendMessageToClientAndWait(
|
||||
ws: WebSocketLike,
|
||||
message: WSOutboundMessage,
|
||||
): Promise<void> {
|
||||
const prepared = this.prepareJsonDelivery([ws], message, true);
|
||||
if (!prepared) throw new Error("Source message delivery unavailable");
|
||||
try {
|
||||
const sent = await sendBoundedPhysicalFrameAndWait({
|
||||
socket: ws,
|
||||
frame: prepared.payload,
|
||||
frameBytes: prepared.bytes,
|
||||
onHighWater: () => this.closeAtOutboundHighWater(ws),
|
||||
});
|
||||
if (!sent) throw new Error("Source is closed");
|
||||
this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
|
||||
} catch (err) {
|
||||
this.logger.warn({ err }, "ws_send_failed");
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1394,6 +1449,10 @@ export class VoiceAssistantWebSocketServer {
|
||||
appVersion,
|
||||
clientCapabilities,
|
||||
permissions: admission.permissions,
|
||||
sourceId:
|
||||
lifecycle.kind === "ephemeral-plugin"
|
||||
? `plugin:${lifecycle.pluginId}`
|
||||
: admission.principalId,
|
||||
connectionLogger,
|
||||
onMessage: (msg) => {
|
||||
if (!connection) {
|
||||
@@ -1401,11 +1460,11 @@ export class VoiceAssistantWebSocketServer {
|
||||
}
|
||||
this.sendToConnection(connection, wrapSessionMessage(msg));
|
||||
},
|
||||
onMessageToSource: (source, msg) => {
|
||||
onMessageToSource: async (source, msg) => {
|
||||
if (!connection || !connection.sockets.has(source as WebSocketLike)) {
|
||||
return;
|
||||
}
|
||||
this.sendToClient(source as WebSocketLike, wrapSessionMessage(msg));
|
||||
await this.sendMessageToClientAndWait(source as WebSocketLike, wrapSessionMessage(msg));
|
||||
},
|
||||
onBinaryMessage: (frame) => {
|
||||
if (!connection) {
|
||||
@@ -1446,6 +1505,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
const base: SessionConnectionBase = {
|
||||
session,
|
||||
principalId: admission.principalId,
|
||||
hubExecutionAgents: admission.hubExecutionAgents,
|
||||
sessionKey: sessionConnectionKey(admission.principalId, clientId),
|
||||
clientId,
|
||||
appVersion,
|
||||
@@ -1463,6 +1523,17 @@ export class VoiceAssistantWebSocketServer {
|
||||
|
||||
private createSocketSession(options: SocketSessionOptions): Session {
|
||||
return new Session({
|
||||
connectedHosts: this.connectedHosts,
|
||||
getConnectedHostBufferedAmount: () => {
|
||||
let total = 0;
|
||||
for (const socket of this.sessions.keys()) {
|
||||
if (socket instanceof VirtualHostSocket) continue;
|
||||
if (typeof socket.bufferedAmount !== "number") return null;
|
||||
total += socket.bufferedAmount;
|
||||
}
|
||||
return total;
|
||||
},
|
||||
sourceId: options.sourceId,
|
||||
browserToolsBroker: this.browserToolsBroker,
|
||||
clientId: options.clientId,
|
||||
appVersion: options.appVersion,
|
||||
@@ -1616,8 +1687,16 @@ export class VoiceAssistantWebSocketServer {
|
||||
}
|
||||
|
||||
this.clearPendingConnection(ws);
|
||||
const admitted = pending.admission;
|
||||
let admitted = pending.admission;
|
||||
if (!admitted) throw new Error("Admitted hello has no session admission");
|
||||
if (message.sessionPermissions)
|
||||
admitted = {
|
||||
...admitted,
|
||||
permissions: admitted.permissions.filter((permission) =>
|
||||
message.sessionPermissions!.includes(permission),
|
||||
),
|
||||
};
|
||||
pending.admission = admitted;
|
||||
pending.identity.clientId = clientId;
|
||||
if (message.appVersion) {
|
||||
pending.identity.appVersion = message.appVersion;
|
||||
@@ -1640,6 +1719,9 @@ export class VoiceAssistantWebSocketServer {
|
||||
lifecycle: pluginId ? { kind: "ephemeral-plugin", pluginId } : { kind: "reconnectable" },
|
||||
admission: admitted,
|
||||
});
|
||||
connection.permissionCeiling = message.sessionPermissions;
|
||||
connection.session.setRouted(message.routed === true);
|
||||
this.assignHostSource(connection, message, admitted.principalId, pluginId);
|
||||
this.sessions.set(ws, connection);
|
||||
if (connection.lifecycle === "reconnectable") {
|
||||
this.externalSessionsByKey.set(sessionKey, connection);
|
||||
@@ -1755,6 +1837,14 @@ export class VoiceAssistantWebSocketServer {
|
||||
clearTimeout(existing.externalDisconnectCleanupTimeout);
|
||||
existing.externalDisconnectCleanupTimeout = null;
|
||||
}
|
||||
for (const socket of existing.sockets) {
|
||||
this.hostLinks.get(socket)?.close("permission_denied");
|
||||
this.hostLinks.delete(socket);
|
||||
}
|
||||
existing.permissionCeiling = message.sessionPermissions;
|
||||
existing.session.setPermissions(pending.admission!.permissions);
|
||||
existing.session.setRouted(message.routed === true);
|
||||
this.assignHostSource(existing, message, pending.admission!.principalId);
|
||||
const newAppVersion = message.appVersion ?? null;
|
||||
if (newAppVersion && newAppVersion !== existing.appVersion) {
|
||||
existing.appVersion = newAppVersion;
|
||||
@@ -1785,6 +1875,21 @@ export class VoiceAssistantWebSocketServer {
|
||||
);
|
||||
}
|
||||
|
||||
private assignHostSource(
|
||||
connection: SessionConnection,
|
||||
hello: WSHelloMessage,
|
||||
principalId: string,
|
||||
pluginId?: string,
|
||||
): void {
|
||||
if (pluginId) return;
|
||||
const producer = hello.producer;
|
||||
const sourceId =
|
||||
hello.routed && producer
|
||||
? `route:${[principalId, producer.serverId, producer.sourceId].map(encodeURIComponent).join(":")}`
|
||||
: principalId;
|
||||
connection.session.setHostSource(sourceId);
|
||||
}
|
||||
|
||||
private buildServerInfoStatusPayload(session: Session): ServerInfoStatusPayload {
|
||||
return {
|
||||
status: "server_info",
|
||||
@@ -1792,11 +1897,14 @@ export class VoiceAssistantWebSocketServer {
|
||||
serverId: this.serverId,
|
||||
hostname: getHostName(),
|
||||
version: this.daemonVersion,
|
||||
sourceId: session.getHostSourceId(),
|
||||
permissions: session.getPermissions(),
|
||||
// COMPAT(desktopManaged): added in v0.1.X, remove optional parsing after 2027-01-16.
|
||||
desktopManaged: this.daemonRuntimeConfig?.desktopManaged === true,
|
||||
...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
|
||||
features: {
|
||||
connectedHosts: true,
|
||||
hostMultiplexing: true,
|
||||
usageSources: true,
|
||||
ownedSubscriptions: true,
|
||||
agentRequestReceipts: true,
|
||||
@@ -2040,6 +2148,9 @@ export class VoiceAssistantWebSocketServer {
|
||||
error?: Error;
|
||||
},
|
||||
): Promise<void> {
|
||||
this.hostLinks.get(ws)?.close();
|
||||
this.hostLinks.delete(ws);
|
||||
|
||||
this.applicationSocketLease.release(ws);
|
||||
const identity = this.socketIdentities.get(ws);
|
||||
const identityFields = identity ? toConnectionLogFields(identity) : {};
|
||||
@@ -2088,6 +2199,10 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.finishPluginSocketCleanup(ws);
|
||||
return;
|
||||
}
|
||||
if (connection.session.isRouted()) {
|
||||
await this.cleanupConnection(connection, "Routed session disconnected");
|
||||
return;
|
||||
}
|
||||
this.incrementRuntimeCounter("sessionDisconnectedWaitingReconnect");
|
||||
if (connection.externalDisconnectCleanupTimeout) {
|
||||
clearTimeout(connection.externalDisconnectCleanupTimeout);
|
||||
@@ -2230,6 +2345,12 @@ export class VoiceAssistantWebSocketServer {
|
||||
return false;
|
||||
}
|
||||
const decodedFrame = decodeBinaryFrame(asBytes);
|
||||
if (asBytes[0] === HOST_ROUTE_OPCODE && !decodedFrame) {
|
||||
ws.close(1008, "Invalid host routing frame");
|
||||
return true;
|
||||
}
|
||||
if (decodedFrame?.kind === "host_route" && this.hostLinks.get(ws)?.binary(decodedFrame.frame))
|
||||
return true;
|
||||
if (!decodedFrame) {
|
||||
return false;
|
||||
}
|
||||
@@ -2384,11 +2505,38 @@ export class VoiceAssistantWebSocketServer {
|
||||
}
|
||||
}
|
||||
|
||||
private hostLink(ws: WebSocketLike, connection: SessionConnection): DaemonLinkEndpoint {
|
||||
let link = this.hostLinks.get(ws);
|
||||
if (!link) {
|
||||
link = new DaemonLinkEndpoint({
|
||||
identity: JSON.stringify([connection.principalId, connection.clientId]),
|
||||
permissions: () => connection.session.getPermissions(),
|
||||
routed: () => connection.session.isRouted(),
|
||||
bufferedAmount: () => ws.bufferedAmount ?? null,
|
||||
delivery: connection.session.delivery,
|
||||
admit: (socket, permissions) =>
|
||||
this.attachSocket(socket, undefined, undefined, false, {
|
||||
principalId: connection.principalId,
|
||||
permissions,
|
||||
}),
|
||||
});
|
||||
this.hostLinks.set(ws, link);
|
||||
}
|
||||
return link;
|
||||
}
|
||||
|
||||
private async dispatchSessionMessage(
|
||||
ws: WebSocketLike,
|
||||
activeConnection: SessionConnection,
|
||||
message: Extract<WSInboundMessage, { type: "session" }>,
|
||||
): Promise<void> {
|
||||
const link = this.hostLink(ws, activeConnection);
|
||||
if (link.handles(message.message)) {
|
||||
await activeConnection.session.delivery.request(ws, message.message, () =>
|
||||
link.dispatch(message.message),
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.recordInboundSessionRequestType(message.message.type);
|
||||
const controlRpc = getControlRpcLogInfo(message.message);
|
||||
if (controlRpc) {
|
||||
|
||||
Reference in New Issue
Block a user