puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-23 17:52:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(agents): persist qualified subagent lifecycle and run evidence

Browse Source
This commit is contained in:
Mohamed Boudra
2026-10-20 10:57:13 +02:00
parent 888b54bb44
commit b1b5010bd2
49 changed files with 3580 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/connected-hosts.md
+8
View File
@@ -29,6 +29,10 @@ principal. An ordinary direct B consumer does not become that producer by supply
The optional `server_info.sourceId` reports this producer identity separately from the authenticated admission principal.
Trusted server plugins retain the authority of their runtime; they are not sandboxed by this feature.
A managed child has a qualified `{serverId, agentId}` address on every host. The parent's daemon owns membership and lifecycle intent; a destination's parent pointer is metadata. Workspace comparisons include the server ID, so matching workspace strings on different hosts do not share lifecycle ownership. Preserve the archive policy in [agent lifecycle](agent-lifecycle.md).
Durable relationships retain a stable authority reference and creation ceiling, never live credentials or execution capabilities. Recovery must resolve current authority through its owner. Derived runtime sessions cannot authorize their own recovery. Execution-scoped admissions and non-owner durable references without a supported authority resolver fail explicitly; do not recover them through the daemon owner's MCP runtime.
## Connections and recovery
The shared Node connector owns the target grammar and each connection attempt, including a fresh
@@ -52,6 +56,10 @@ Ordinary non-recoverable mutations are not automatically replayed. Existing ackn
creation recovery retains its request identity and resumes through the ordinary creation lifecycle. Multiplexing changes only the transport: ordinary permission
RPC codes pass through unchanged, and connection/recovery behavior matches direct clients.
Creation cancellation fences a reserved creation identity before admission. An admitted receipt can precede resource readiness and cannot be reported as canceled. Observe its current receipt and later changes through the ordinary SDK instead of replaying an uncertain prompt. Parent archive retains a pending obligation until cancellation or destination lifecycle confirmation; explicit forget removes local membership without claiming the destination changed.
Run journals retain submission and provider-turn identity. Restarted nonterminal evidence is interrupted, and subsequent evidence must match that turn. Idle and reconnect do not prove completion. A storage failure retains observed evidence with pending durability; a later read retries persistence. Notification recovery must preserve checkpoint identity and this durability distinction.
Text session envelopes carry a channel identifier. Binary routing adds an opcode, a bounded UTF-8
channel identifier, and the original terminal or file-transfer bytes. It never converts those bytes to
base64 JSON. The existing direct binary opcodes and 100 MiB message limit remain unchanged; routed
packages/client/src/agents/observation.ts
+50
View File
@@ -0,0 +1,50 @@
import type { DaemonClient } from "../daemon-client.js";
import type { OwnedSubscription } from "../connection/index.js";
type SubscriptionSnapshot<T> = Awaited<OwnedSubscription<T>["ready"]>;
import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages";
/** Exact-ID observation over the ordinary owned agent directory stream. */
export function observeAgent(
client: DaemonClient,
agentId: string,
): OwnedSubscription<AgentSnapshotPayload> {
client.requireFleetSubagents();
const source = client.observeAgents({
includeBackground: true,
filter: { agentIds: [agentId], includeArchived: true, includeBackground: true },
});
const project = (
snapshot: Awaited<typeof source.ready>,
): SubscriptionSnapshot<AgentSnapshotPayload> => {
const agent = snapshot.entries.find((entry) => entry.agent.id === agentId)?.agent;
if (!agent)
throw Object.assign(new Error(`Agent ${agentId} was not found`), { code: "agent_not_found" });
return { ...agent, subscriptionId: snapshot.subscriptionId };
};
const ready = source.ready.then(project).catch(async (error: unknown) => {
await source.release();
throw error;
});
void ready.catch(() => {});
return {
get subscriptionId() {
return source.subscriptionId;
},
ready,
subscribe(observer) {
return source.subscribe({
snapshot(snapshot) {
try {
observer.snapshot(project(snapshot));
} catch (error) {
observer.error?.(error);
void source.release();
}
},
update: observer.update,
error: observer.error,
});
},
release: () => source.release(),
};
}
packages/client/src/creation/index.ts
+4 -1
View File
@@ -168,7 +168,10 @@ export class CreationClient {
subscribe: true,
});
if (result.creation) this.receive(result.creation);
this.finish(operation, result);
this.finish(operation, {
...result,
errorCode: result.creation?.errorCode ?? result.errorCode,
});
} catch (error) {
if (
operation.settled ||
packages/client/src/daemon-client.ts
+100 -1
View File
@@ -460,6 +460,7 @@ export interface CreateAgentRequestOptions extends AgentConfigOverrides {
cwd?: string;
env?: CreateAgentRequestMessage["env"];
workspaceId?: string;
parent?: CreateAgentRequestMessage["parent"];
callerAgentId?: string;
initialPrompt?: string;
idempotencyKey?: string;
@@ -1678,6 +1679,96 @@ export class DaemonClient {
code: "unsupported_capability",
});
}
requireFleetSubagents(): void {
if (this.lastServerInfoMessage?.features?.fleetSubagents !== true)
throw Object.assign(new Error("Update the host to use fleet subagents"), {
code: "unsupported_capability",
});
}
observeAgentCreation(idempotencyKey: string) {
this.requireFleetSubagents();
return this.observe("creation.subscribe.response", {
type: "creation.subscribe.request",
kind: "agent",
idempotencyKey,
});
}
async cancelAgentCreation(input: { idempotencyKey: string; agentId: string }) {
this.requireFleetSubagents();
return (
await this.sendCorrelatedSessionRequest({
message: { type: "agent.create.cancel.request", ...input },
responseType: "agent.create.cancel.response",
})
).cancellation;
}
async getAgentRun(agentId: string, submissionId: string) {
this.requireFleetSubagents();
return (
await this.sendCorrelatedSessionRequest({
message: { type: "agent.runs.get.request", agentId, submissionId },
responseType: "agent.runs.get.response",
})
).run;
}
async createSubagent(
parentId: string,
input: import("@getpaseo/protocol/messages").SubagentCreateInput,
) {
this.requireFleetSubagents();
return (
await this.sendCorrelatedSessionRequest({
message: { type: "agent.subagents.create.request", parentId, ...input },
responseType: "agent.subagents.create.response",
})
).child;
}
async detachSubagent(
parentId: string,
child: import("@getpaseo/protocol/addresses").AgentAddress,
) {
this.requireFleetSubagents();
return (
await this.sendCorrelatedSessionRequest({
message: { type: "agent.subagents.detach.request", parentId, child },
responseType: "agent.subagents.detach.response",
})
).relationship;
}
async forgetSubagent(
parentId: string,
child: import("@getpaseo/protocol/addresses").AgentAddress,
): Promise<void> {
this.requireFleetSubagents();
await this.sendCorrelatedSessionRequest({
message: { type: "agent.subagents.forget.request", parentId, child },
responseType: "agent.subagents.forget.response",
});
}
async listSubagents(parentId: string) {
this.requireFleetSubagents();
return (
await this.sendCorrelatedSessionRequest({
message: { type: "agent.subagents.list.request", parentId },
responseType: "agent.subagents.list.response",
})
).directory;
}
observeSubagents(
parentId: string,
): OwnedSubscription<import("@getpaseo/protocol/messages").SubagentDirectory> {
this.requireFleetSubagents();
return this.owned.observe(async (accept) => {
const result = await this.sendCorrelatedSessionRequest({
message: { type: "agent.subagents.list.request", parentId, observe: true },
responseType: "agent.subagents.list.response",
});
if (!result.subscriptionId) throw new Error("Host omitted subagent subscription identity");
const snapshot = { ...result.directory, subscriptionId: result.subscriptionId };
accept(snapshot);
return snapshot;
});
}
async listHosts(): Promise<HostDirectory> {
this.requireConnectedHosts();
return (
@@ -2552,6 +2643,7 @@ export class DaemonClient {
observeAgents(
options: Omit<FetchAgentsOptions, "subscribe"> = {},
): OwnedSubscription<FetchAgentsPayload> {
if (options.filter?.agentIds) this.requireFleetSubagents();
if (options.includeBackground || options.filter?.includeBackground)
this.requireBackgroundWorkspaces();
const { signal, requestId, timeout, ...query } = options;
@@ -2585,6 +2677,7 @@ export class DaemonClient {
if (options?.includeBackground || options?.filter?.includeBackground)
this.requireBackgroundWorkspaces();
if (options?.subscribe) {
if (options.filter?.agentIds) this.requireFleetSubagents();
if (options.subscribe.subscriptionId !== undefined)
throw new Error("Subscription IDs are assigned by the host");
const subscription = this.observeAgents(options);
@@ -3094,12 +3187,16 @@ export class DaemonClient {
}
async createAgent(options: CreateAgentRequestOptions): Promise<AgentSnapshotPayload> {
if (options.parent) this.requireFleetSubagents();
if (options.background !== undefined) this.requireBackgroundWorkspaces();
const result = await this.creations.createAgent({
...options,
config: resolveAgentConfig(options),
});
if (result.error || !result.agent) throw new Error(result.error ?? "Agent creation failed");
if (result.error || !result.agent)
throw Object.assign(new Error(result.error ?? "Agent creation failed"), {
code: result.errorCode,
});
return result.agent;
}
@@ -3115,6 +3212,7 @@ export class DaemonClient {
config,
...(options.env ? { env: options.env } : {}),
...(options.workspaceId !== undefined ? { workspaceId: options.workspaceId } : {}),
...(options.parent !== undefined ? { parent: options.parent } : {}),
...(options.callerAgentId !== undefined ? { callerAgentId: options.callerAgentId } : {}),
...(options.initialPrompt ? { initialPrompt: options.initialPrompt } : {}),
idempotencyKey: options.idempotencyKey,
@@ -7270,6 +7368,7 @@ function resolveAgentConfig(options: CreateAgentRequestOptions): AgentSessionCon
onEvent: _onEvent,
idempotencyKey: _idempotencyKey,
clientMessageId: _clientMessageId,
parent: _parent,
callerAgentId: _callerAgentId,
outputSchema: _outputSchema,
attachments: _attachments,
packages/client/src/hosts/runtime.ts
+5 -1
View File
@@ -65,9 +65,13 @@ export class HostRuntime {
return api.agents.ref(agentId);
}
protected stopAcquisitions(): void {
this.disposed = true;
}
dispose(): Promise<void> {
if (this.disposal) return this.disposal;
this.disposed = true;
this.stopAcquisitions();
this.disposal = (async () => {
const results = await Promise.allSettled([...this.scopes].map((scope) => scope.dispose()));
// Acquisitions that finish after disposal release themselves before rejecting callers.
packages/client/src/index.ts
+167 -7
View File
@@ -1,3 +1,5 @@
import { observeAgent } from "./agents/observation.js";
import { safeRandomId } from "./daemon-client-transport-utils.js";
import { createHostActions, type PaseoHostActions } from "./hosts/index.js";
import type { AgentAddress } from "@getpaseo/protocol/addresses";
export type { AgentAddress, WorkspaceAddress } from "@getpaseo/protocol/addresses";
@@ -9,6 +11,8 @@ export type { OwnedSubscription, SubscriptionObserver } from "./connection/index
import type { DaemonClientConfig } from "./daemon-client.js";
import type { AgentPermissionResponse } from "@getpaseo/protocol/agent-types";
import type {
SubagentRelationship,
SubagentDirectory,
AgentSnapshotPayload,
CreationSnapshot,
CreateAgentRequestMessage,
@@ -252,7 +256,7 @@ export interface PaseoAgentCreateOptions {
onEvent?: (snapshot: CreationSnapshot) => void;
config: PaseoAgentConfig;
cwd: string;
parent?: string | PaseoAgentHandle;
parent?: string | PaseoAgentHandle | AgentAddress;
title?: PaseoAgentSessionConfig["title"];
env?: CreateAgentRequestMessage["env"];
prompt?: string;
@@ -344,7 +348,28 @@ export interface PaseoAgentTimelineHandle {
subscribe(handler: (event: PaseoAgentTimelineEvent) => void): PaseoAgentTimelineSubscription;
}
export type PaseoSubagentCreateOptions = Omit<PaseoAgentCreateOptions, "parent" | "cwd"> & {
serverId?: string;
workspaceId?: string;
cwd?: string;
};
export interface PaseoSubagentActions {
detach(child: AgentAddress): Promise<SubagentRelationship>;
forget(child: AgentAddress): Promise<void>;
create(input: PaseoSubagentCreateOptions): Promise<PaseoAgentHandle>;
list(): Promise<SubagentDirectory>;
observe(): OwnedSubscription<SubagentDirectory>;
}
export type { SubagentDirectory, SubagentRelationship } from "@getpaseo/protocol/messages";
export type { AgentRunRecord, AgentRunCheckpoint } from "@getpaseo/protocol/agent-runs";
export interface PaseoAgentHandle {
observe(): OwnedSubscription<PaseoAgent>;
readonly runs: {
get(
submissionId: string,
): Promise<import("@getpaseo/protocol/agent-runs").AgentRunRecord | null>;
};
readonly subagents: PaseoSubagentActions;
readonly id: string;
readonly serverId: string;
readonly address: AgentAddress;
@@ -491,7 +516,17 @@ export interface PaseoConfigActions {
): Promise<{ requestId: string; config: MutableDaemonConfig }>;
}
export interface PaseoCreationActions {
observeAgent(idempotencyKey: string): ReturnType<DaemonClient["observeAgentCreation"]>;
/** Cancel only pre-admission creation; admitted work retains its existing receipt. */
cancelAgent(input: {
idempotencyKey: string;
agentId: string;
}): Promise<import("@getpaseo/protocol/messages").AgentCreationCancellation>;
}
export interface PaseoApi {
readonly creations: PaseoCreationActions;
readonly hosts: PaseoHostActions;
dispose(): Promise<void>;
observeEvents: DaemonClient["observeEvents"];
@@ -534,9 +569,11 @@ export function createPaseoClient(config: PaseoClientConfig): PaseoClient {
function toDaemonAgentCreateOptions(
options: PaseoAgentCreateOptions,
serverId: string | null,
placement?: { workspaceId: string; cwd: string },
): CreateAgentRequestOptions {
const { config: agentConfig, cwd, parent, title, prompt, ...requestOptions } = options;
const parentAddress = normalizeParentAddress(parent, serverId);
const { provider: providerModel, options: providerOptions, ...runtimeConfig } = agentConfig;
const { provider, model } = parseProviderModel(providerModel);
return {
@@ -550,7 +587,11 @@ function toDaemonAgentCreateOptions(
...(providerOptions !== undefined ? { providerOptions } : {}),
},
...(placement ? { workspaceId: placement.workspaceId } : {}),
...(parent ? { callerAgentId: resolveAgentId(parent) } : {}),
...(parentAddress ? { parent: parentAddress } : {}),
...(parentAddress?.serverId === serverId &&
(typeof parent === "string" || (parent && "address" in parent))
? { callerAgentId: parentAddress.agentId }
: {}),
...(prompt !== undefined ? { initialPrompt: prompt } : {}),
};
}
@@ -588,17 +629,85 @@ export function createPaseoApi(
workspaceListeners.delete(handler);
};
};
const hosts = createHostActions(daemonClient, {
own,
signal: lifetime.signal,
api: createPaseoApi,
});
let subagentRuntime: import("./hosts/runtime.js").HostRuntime | undefined;
const createAgentHandle = createAgentHandleFactory(
daemonClient,
listenAgents,
(agentId, handler) => own(() => daemonClient.subscribeAgentTimeline(agentId, handler)),
lifetime.signal,
(agentId) => own(() => observeAgent(daemonClient, agentId)),
(parentId) => ({
detach: (child) => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
return daemonClient.detachSubagent(parentId, child);
},
forget: (child) => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
return daemonClient.forgetSubagent(parentId, child);
},
list: () => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
return daemonClient.listSubagents(parentId);
},
observe: () => own(() => daemonClient.observeSubagents(parentId)),
create: async ({ serverId, idempotencyKey, ...input }) => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
const normalized = toDaemonAgentCreateOptions(
{ ...input, cwd: input.cwd ?? "" },
daemonClient.getServerId(),
);
const child = await daemonClient.createSubagent(parentId, {
serverId: serverId ?? requireServerIdentity(daemonClient.getServerId()),
idempotencyKey: idempotencyKey ?? safeRandomId(),
input: { ...normalized, labels: input.labels ?? {}, config: normalized.config! },
});
subagentRuntime ??= hosts.runtime();
const handle = await subagentRuntime.agent(child);
await handle.refresh();
return handle;
},
}),
);
const createAgent = async (
options: PaseoAgentCreateOptions,
placement?: { workspaceId: string; cwd: string },
) => {
const agent = await daemonClient.createAgent(toDaemonAgentCreateOptions(options, placement));
const { parent, ...input } = options;
if (
parent &&
typeof parent !== "string" &&
"subagents" in parent &&
daemonClient.getLastServerInfoMessage()?.features?.fleetSubagents === true
) {
return parent.subagents.create({
...input,
serverId: requireServerIdentity(daemonClient.getServerId()),
workspaceId: placement?.workspaceId,
});
}
if (
typeof parent === "string" &&
daemonClient.getLastServerInfoMessage()?.features?.fleetSubagents === true
) {
return createAgentHandle(parent).subagents.create({
...input,
workspaceId: placement?.workspaceId,
});
}
const normalized = toDaemonAgentCreateOptions(options, daemonClient.getServerId(), placement);
// COMPAT(qualifiedSubagents): existing local parent creation on pre-fleet daemons.
// Added in v0.12; remove after 2027-04-20 once the host floor supports fleet subagents.
if (
normalized.callerAgentId &&
daemonClient.getLastServerInfoMessage()?.features?.fleetSubagents !== true
)
delete normalized.parent;
const agent = await daemonClient.createAgent(normalized);
return createAgentHandle(agent);
};
const terminals = createTerminalActions(daemonClient, async (workspaceId) => {
@@ -705,7 +814,14 @@ export function createPaseoApi(
return {
dispose,
hosts: createHostActions(daemonClient, { own, signal: lifetime.signal, api: createPaseoApi }),
creations: {
observeAgent: (key) => own(() => daemonClient.observeAgentCreation(key)),
cancelAgent: (input) => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
return daemonClient.cancelAgentCreation(input);
},
},
hosts,
observeEvents,
terminals,
projects: {
@@ -723,7 +839,12 @@ export function createPaseoApi(
openWorkspace(daemonClient, createWorkspaceHandle, input, requestId),
create: async ({ requestId, agent, ...options }) => {
const result = await daemonClient.createWorkspace(
{ ...options, ...(agent ? { agent: toDaemonAgentCreateOptions(agent) } : {}) },
{
...options,
...(agent
? { agent: toDaemonAgentCreateOptions(agent, daemonClient.getServerId()) }
: {}),
},
requestId,
);
if (result.error || !result.workspace) {
@@ -872,6 +993,8 @@ function createAgentHandleFactory(
listen: (handler: PaseoAgentUpdateHandler) => () => void,
subscribeTimeline: DaemonClient["subscribeAgentTimeline"],
signal: AbortSignal,
observeSnapshot: (agentId: string) => OwnedSubscription<PaseoAgent>,
subagents: (parentId: string) => PaseoSubagentActions,
): AgentHandleFactory {
const assertActive = () => {
if (signal.aborted) throw new Error("Paseo API is disposed");
@@ -881,6 +1004,31 @@ function createAgentHandleFactory(
let current = typeof agent === "string" ? null : agent;
const handle: PaseoAgentHandle = {
observe: () => {
assertActive();
const observation = observeSnapshot(id);
observation.subscribe({
snapshot: (snapshot) => {
current = snapshot;
},
update: (message) => {
if (
message.type === "agent_update" &&
message.payload.kind === "upsert" &&
message.payload.agent.id === id
)
current = message.payload.agent;
},
});
return observation;
},
runs: {
get: (submissionId) => {
assertActive();
return daemonClient.getAgentRun(id, submissionId);
},
},
subagents: subagents(id),
id,
get serverId() {
const serverId = daemonClient.getServerId();
@@ -1053,8 +1201,20 @@ function resolveWorkspaceId(workspace: string | PaseoWorkspaceHandle): string {
return typeof workspace === "string" ? workspace : workspace.id;
}
function resolveAgentId(agent: string | PaseoAgentHandle): string {
return typeof agent === "string" ? agent : agent.id;
function normalizeParentAddress(
parent: PaseoAgentCreateOptions["parent"],
serverId: string | null,
): AgentAddress | undefined {
if (!parent) return undefined;
if (typeof parent === "string")
return { serverId: requireServerIdentity(serverId), agentId: parent };
if ("address" in parent) return parent.address;
return parent;
}
function requireServerIdentity(serverId: string | null): string {
if (!serverId) throw new Error("Connect to the host before addressing a parent agent");
return serverId;
}
function parseProviderModel(selection: string): { provider: string; model: string } {
packages/protocol/src/agent-lifecycle.ts
+16
View File
@@ -1,3 +1,4 @@
import type { WorkspaceAddress } from "./addresses.js";
export const AGENT_LIFECYCLE_STATUSES = [
"initializing",
"idle",
@@ -7,3 +8,18 @@ export const AGENT_LIFECYCLE_STATUSES = [
] as const;
export type AgentLifecycleStatus = (typeof AGENT_LIFECYCLE_STATUSES)[number];
/** Archiving a parent preserves children in another workspace or an open tab. */
export function shouldDetachFromArchivedParent(input: {
parentWorkspace: WorkspaceAddress | undefined;
childWorkspace: WorkspaceAddress | undefined;
hasOpenTab: boolean;
}): boolean {
const { parentWorkspace, childWorkspace } = input;
const differentWorkspace =
parentWorkspace !== undefined &&
childWorkspace !== undefined &&
(parentWorkspace.serverId !== childWorkspace.serverId ||
parentWorkspace.workspaceId !== childWorkspace.workspaceId);
return differentWorkspace || input.hasOpenTab;
}
packages/protocol/src/agent-runs.ts
+48
View File
@@ -0,0 +1,48 @@
import { z } from "zod";
import { AgentAddressSchema } from "./addresses.js";
export const AgentRunCheckpointSchema = z.object({
sequence: z.number().int().positive(),
kind: z.enum([
"interrupted",
"submitted",
"running",
"permission",
"resumed",
"completed",
"failed",
"canceled",
]),
at: z.string(),
permissionRequestId: z.string().optional(),
result: z.string().optional(),
error: z.string().optional(),
});
export type AgentRunCheckpoint = z.infer<typeof AgentRunCheckpointSchema>;
export const AgentRunRecordSchema = z.object({
agent: AgentAddressSchema,
submissionId: z.string(),
turnId: z.string().nullable(),
durability: z.enum(["persisted", "pending"]).optional(),
status: z.enum([
"interrupted",
"submitted",
"running",
"permission",
"completed",
"failed",
"canceled",
]),
checkpoints: z.array(AgentRunCheckpointSchema),
});
export type AgentRunRecord = z.infer<typeof AgentRunRecordSchema>;
export const AgentRunGetRequestSchema = z.object({
type: z.literal("agent.runs.get.request"),
requestId: z.string(),
agentId: z.string(),
submissionId: z.string(),
});
export const AgentRunGetResponseSchema = z.object({
type: z.literal("agent.runs.get.response"),
payload: z.object({ requestId: z.string(), run: AgentRunRecordSchema.nullable() }),
});
packages/protocol/src/messages.ts
+127
View File
@@ -1,3 +1,5 @@
import { AgentRunGetRequestSchema, AgentRunGetResponseSchema } from "./agent-runs.js";
import { AgentAddressSchema, WorkspaceAddressSchema } from "./addresses.js";
import { AgentMessageSchema } from "./agent-message.js";
import {
HostsListRequestSchema,
@@ -830,6 +832,7 @@ const AgentActiveTurnPayloadSchema = z.object({
});
export const AgentSnapshotPayloadSchema = z.object({
parent: AgentAddressSchema.nullable().optional(),
id: z.string(),
provider: AgentProviderSchema,
cwd: z.string(),
@@ -865,6 +868,7 @@ export const AgentSnapshotPayloadSchema = z.object({
export type AgentSnapshotPayload = z.infer<typeof AgentSnapshotPayloadSchema>;
export const AgentListItemPayloadSchema = z.object({
parent: AgentAddressSchema.nullable().optional(),
id: z.string(),
shortId: z.string(),
title: z.string().nullable(),
@@ -925,6 +929,7 @@ export const AudioPlayedMessageSchema = z.object({
});
const AgentDirectoryFilterSchema = z.object({
agentIds: z.array(z.string()).optional(),
labels: z.record(z.string(), z.string()).optional(),
projectKeys: z.array(z.string()).optional(),
statuses: z.array(AgentStatusSchema).optional(),
@@ -1725,6 +1730,7 @@ export const CreateAgentWorktreeTargetSchema = z.discriminatedUnion("mode", [
export type CreateAgentWorktreeTarget = z.infer<typeof CreateAgentWorktreeTargetSchema>;
export const CreateAgentRequestMessageSchema = z.object({
parent: AgentAddressSchema.nullable().optional(),
type: z.literal("create_agent_request"),
// Legacy create_agent_request uses a separate initial-message receipt when keyed.
idempotencyKey: z.string().min(1).max(512).optional(),
@@ -3220,7 +3226,106 @@ export const SubscriptionReleaseResponseSchema = z.object({
payload: z.object({ requestId: z.string(), subscriptionId: z.string() }),
});
export const SubagentRelationshipSchema = z.object({
id: z.string(),
parent: AgentAddressSchema,
child: AgentAddressSchema,
workspace: WorkspaceAddressSchema.optional(),
phase: z.enum([
"creating",
"attached",
"detach_pending",
"lifecycle_pending",
"canceled",
"detached",
"archived",
"forgotten",
]),
snapshot: AgentSnapshotPayloadSchema.optional(),
observedAt: z.string().nullable(),
freshness: z.enum(["current", "stale", "unavailable"]),
error: z.object({ code: z.string(), message: z.string() }).nullable(),
});
export type SubagentRelationship = z.infer<typeof SubagentRelationshipSchema>;
export const SubagentDirectorySchema = z.object({
parent: AgentAddressSchema,
rows: z.array(SubagentRelationshipSchema),
});
export type SubagentDirectory = z.infer<typeof SubagentDirectorySchema>;
export const SubagentCreateInputSchema = z.object({
serverId: z.string().min(1),
idempotencyKey: z.string().min(1).max(512),
input: CreateAgentRequestMessageSchema.omit({
type: true,
requestId: true,
callerAgentId: true,
parent: true,
}).extend({ agentId: z.string().uuid().optional() }),
});
export type SubagentCreateInput = z.infer<typeof SubagentCreateInputSchema>;
export const SubagentsCreateRequestSchema = z.object({
type: z.literal("agent.subagents.create.request"),
requestId: z.string(),
parentId: z.string(),
...SubagentCreateInputSchema.shape,
});
export const SubagentsCreateResponseSchema = z.object({
type: z.literal("agent.subagents.create.response"),
payload: z.object({ requestId: z.string(), child: AgentAddressSchema }),
});
export const SubagentsListRequestSchema = z.object({
type: z.literal("agent.subagents.list.request"),
requestId: z.string(),
parentId: z.string(),
observe: z.boolean().optional(),
});
export const SubagentsListResponseSchema = z.object({
type: z.literal("agent.subagents.list.response"),
payload: z.object({
requestId: z.string(),
directory: SubagentDirectorySchema,
subscriptionId: z.string().optional(),
}),
});
export const SubagentsDetachRequestSchema = z.object({
type: z.literal("agent.subagents.detach.request"),
requestId: z.string(),
parentId: z.string(),
child: AgentAddressSchema,
});
export const SubagentsDetachResponseSchema = z.object({
type: z.literal("agent.subagents.detach.response"),
payload: z.object({ requestId: z.string(), relationship: SubagentRelationshipSchema }),
});
export const SubagentsForgetRequestSchema = z.object({
type: z.literal("agent.subagents.forget.request"),
requestId: z.string(),
parentId: z.string(),
child: AgentAddressSchema,
});
export const SubagentsForgetResponseSchema = z.object({
type: z.literal("agent.subagents.forget.response"),
payload: z.object({ requestId: z.string() }),
});
export const SubagentsChangedSchema = z.object({
type: z.literal("agent.subagents.changed"),
payload: z.object({ subscriptionId: z.string(), directory: SubagentDirectorySchema }),
});
export const AgentCreationCancelRequestSchema = z.object({
type: z.literal("agent.create.cancel.request"),
requestId: z.string(),
idempotencyKey: z.string().min(1),
agentId: z.string().uuid(),
});
export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
AgentCreationCancelRequestSchema,
AgentRunGetRequestSchema,
SubagentsDetachRequestSchema,
SubagentsForgetRequestSchema,
SubagentsCreateRequestSchema,
SubagentsListRequestSchema,
HostsListRequestSchema,
HostsAddRequestSchema,
HostsUpdateRequestSchema,
@@ -3623,6 +3728,7 @@ export const ServerInfoStatusPayloadSchema = z
backgroundWorkspaces: z.boolean().optional(),
providersSnapshot: z.boolean().optional(),
connectedHosts: z.boolean().optional(),
fleetSubagents: z.boolean().optional(),
hostMultiplexing: z.boolean().optional(),
usageSources: z.boolean().optional(),
// COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10.
@@ -4938,6 +5044,19 @@ export const CreationSnapshotSchema = z.object({
failedStage: z.enum(["workspace", "agent", "prompt"]).optional(),
outcomeUnknown: z.boolean().optional(),
});
export const AgentCreationCancellationSchema = z.discriminatedUnion("status", [
z.object({ status: z.literal("canceled"), reservedAgent: AgentAddressSchema }),
z.object({
status: z.literal("admitted"),
reservedAgent: AgentAddressSchema,
creation: CreationSnapshotSchema,
}),
]);
export type AgentCreationCancellation = z.infer<typeof AgentCreationCancellationSchema>;
export const AgentCreationCancelResponseSchema = z.object({
type: z.literal("agent.create.cancel.response"),
payload: z.object({ requestId: z.string(), cancellation: AgentCreationCancellationSchema }),
});
export type CreationSnapshot = z.infer<typeof CreationSnapshotSchema>;
export type AgentCreateRequest = z.infer<typeof AgentCreateRequestSchema>;
export type WorkspaceInitialAgent = z.infer<typeof WorkspaceInitialAgentSchema>;
@@ -4964,6 +5083,7 @@ export const AgentCreateResponseSchema = z.object({
requestId: z.string(),
agent: AgentSnapshotPayloadSchema.nullable(),
error: z.string().nullable(),
errorCode: z.string().optional(),
creation: CreationSnapshotSchema.optional(),
}),
});
@@ -6903,6 +7023,13 @@ export const AgentSkillsImportLegacySelectionResponseSchema = z.object({
});
export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
AgentCreationCancelResponseSchema,
AgentRunGetResponseSchema,
SubagentsDetachResponseSchema,
SubagentsForgetResponseSchema,
SubagentsCreateResponseSchema,
SubagentsListResponseSchema,
SubagentsChangedSchema,
HostsListResponseSchema,
HostsChangedSchema,
HostsAddResponseSchema,
packages/server/src/server/agent/agent-loading.test.ts
+3
View File
@@ -48,6 +48,7 @@ test("loads archived records for history and active records with the interactive
isAvailable: async () => await baseClient.isAvailable(),
};
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
registry: storage,
logger,
@@ -87,6 +88,7 @@ test("resuming a stored agent keeps its unread flag and its last-activity time",
const logger = createTestLogger();
const storage = new AgentStorage(path.join(root, "agents"), logger);
const manager = new AgentManager({
serverId: "srv_test",
clients: createTestAgentClients(),
registry: storage,
logger,
@@ -149,6 +151,7 @@ test("loads an archived agent's history after its working directory is removed",
const logger = createTestLogger();
const storage = new AgentStorage(path.join(root, "agents"), logger);
const manager = new AgentManager({
serverId: "srv_test",
clients: createTestAgentClients(),
registry: storage,
logger,
packages/server/src/server/agent/agent-loading.ts
+6 -1
View File
@@ -117,7 +117,11 @@ export async function ensureAgentLoaded(
handle,
buildConfigOverrides(record),
agentId,
{ ...extractTimestamps(record), attention: extractAttention(record) },
{
...extractTimestamps(record),
parent: record.parent,
attention: extractAttention(record),
},
record.archivedAt ? { purpose: "history" } : undefined,
);
deps.logger.info({ agentId, provider: record.provider }, "Agent resumed from persistence");
@@ -134,6 +138,7 @@ export async function ensureAgentLoaded(
snapshot = await deps.agentManager.createAgent(config, agentId, {
labels: record.labels,
workspaceId: record.workspaceId,
parent: record.parent,
owner: record.owner,
});
deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config");
packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts
+1
View File
@@ -247,6 +247,7 @@ function createHarness(options?: { provider?: AgentProvider }): Harness {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stream-coalescing-"));
const client = new TestAgentClient(options?.provider ?? "codex");
const manager = new AgentManager({
serverId: "srv_test",
clients: { [client.provider]: client },
idFactory: createIdFactory(),
logger: createTestLogger(),
packages/server/src/server/agent/agent-manager.test.ts
+310 -29
View File
File diff suppressed because it is too large Load Diff
packages/server/src/server/agent/agent-manager.ts
+113 -36
View File
@@ -1,3 +1,5 @@
import { AgentRuns } from "./runs/index.js";
import type { AgentAddress } from "@getpaseo/protocol/addresses";
import { projectTimelineRows } from "./timeline-projection.js";
import type { PluginLifecycle } from "../plugins/lifecycle/index.js";
import { describeHookAgent, publishAgentStream } from "../plugins/lifecycle/index.js";
@@ -7,6 +9,7 @@ import { basename, resolve } from "node:path";
import { stat } from "node:fs/promises";
import {
AGENT_LIFECYCLE_STATUSES,
shouldDetachFromArchivedParent,
type AgentLifecycleStatus,
} from "@getpaseo/protocol/agent-lifecycle";
import {
@@ -314,10 +317,13 @@ export interface CreateAgentOptions {
initialTitle?: string | null;
// undefined is an explicit decision: the agent never appears in the sidebar.
workspaceId: string | undefined;
parent?: AgentAddress | null;
owner?: AgentOwner;
}
export interface AgentManagerOptions {
runJournalDirectory?: string;
serverId: string;
pluginLifecycle?: PluginLifecycle;
/** An internal workspace makes every agent created inside it internal. */
clients?: ProviderClientMap;
@@ -408,6 +414,7 @@ interface ManagedAgentBase {
* Null/undefined for legacy agents created before ownership stamping.
*/
workspaceId?: string;
parent?: AgentAddress | null;
owner?: AgentOwner;
capabilities: AgentCapabilityFlags;
config: AgentSessionConfig;
@@ -547,6 +554,7 @@ interface WriteLabelsResult {
}
interface AgentMetadataPatch {
parent?: AgentAddress | null;
title?: string;
labels?: AgentLabelPatch;
}
@@ -697,17 +705,6 @@ function getFirstUserMessageTextFromRows(rows: readonly AgentTimelineRow[]): str
return null;
}
function shouldDetachFromArchivedParent(
parent: Pick<StoredAgentRecord, "workspaceId">,
child: StoredAgentRecord,
): boolean {
const isCrossWorkspace =
parent.workspaceId !== undefined &&
child.workspaceId !== undefined &&
parent.workspaceId !== child.workspaceId;
return isCrossWorkspace || hasOpenAgentTab(child.labels);
}
function detachedAgentLabelPatch(labels: Record<string, string>): AgentLabelPatch {
const patch: AgentLabelPatch = { [PARENT_AGENT_ID_LABEL]: null };
for (const label of Object.keys(labels)) {
@@ -747,6 +744,7 @@ export class AgentManager {
private readonly sessionEventTails = new Map<string, Promise<void>>();
private readonly steerEventBarriers = new Map<string, SteerEventBarrier>();
private readonly foregroundMutationTails = new Map<string, Promise<void>>();
private readonly runJournal: AgentRuns;
private readonly runs = new AgentRunState();
private readonly subscribers = new Set<SubscriptionRecord>();
private readonly idFactory: () => string;
@@ -776,7 +774,14 @@ export class AgentManager {
private readonly beforeSteerUnavailableFallback?: AgentManagerOptions["beforeSteerUnavailableFallback"];
private acceptingAgentRegistrations = true;
private readonly serverId: string;
constructor(options: AgentManagerOptions) {
this.serverId = options.serverId;
this.runJournal = new AgentRuns({
serverId: options.serverId,
directory: options.runJournalDirectory,
});
this.pluginLifecycle = options.pluginLifecycle;
this.idFactory = options?.idFactory ?? (() => randomUUID());
this.registry = options?.registry;
@@ -861,6 +866,10 @@ export class AgentManager {
this.onAgentAttention = callback;
}
getAgentRun(agentId: string, submissionId: string) {
return this.runJournal.get(agentId, submissionId);
}
setAgentArchivedCallback(callback: AgentArchivedCallback): void {
this.onAgentArchived = callback;
}
@@ -1316,6 +1325,7 @@ export class AgentManager {
labels: options.labels,
initialTitle: options.initialTitle,
workspaceId: options.workspaceId,
parent: options.parent,
owner: options.owner,
historyPrimed: true,
});
@@ -1347,6 +1357,7 @@ export class AgentManager {
lastUserMessageAt?: Date | null;
labels?: Record<string, string>;
workspaceId?: string;
parent?: AgentAddress | null;
owner?: AgentOwner;
attention?: AttentionState;
},
@@ -1379,6 +1390,7 @@ export class AgentManager {
lastUserMessageAt?: Date | null;
labels?: Record<string, string>;
workspaceId?: string;
parent?: AgentAddress | null;
owner?: AgentOwner;
attention?: AttentionState;
},
@@ -1623,6 +1635,7 @@ export class AgentManager {
return this.registerSession(session, storedConfig, agentId, {
labels: existing.labels,
workspaceId: existing.workspaceId,
parent: existing.parent,
owner: existing.owner,
createdAt: existing.createdAt,
updatedAt: existing.updatedAt,
@@ -1903,7 +1916,17 @@ export class AgentManager {
) {
return;
}
if (shouldDetachFromArchivedParent(parent, currentChild)) {
if (
shouldDetachFromArchivedParent({
parentWorkspace: parent.workspaceId
? { serverId: this.serverId, workspaceId: parent.workspaceId }
: undefined,
childWorkspace: currentChild.workspaceId
? { serverId: this.serverId, workspaceId: currentChild.workspaceId }
: undefined,
hasOpenTab: hasOpenAgentTab(currentChild.labels),
})
) {
await this.detachAgentUnlocked(currentChild.id);
} else if (this.agents.has(currentChild.id)) {
await this.archiveAgentUnlocked(currentChild.id);
@@ -1986,6 +2009,7 @@ export class AgentManager {
provider: record.provider,
cwd: record.cwd,
workspaceId: record.workspaceId,
parent: record.parent,
owner: record.owner,
session: null,
capabilities: STORED_AGENT_CAPABILITIES,
@@ -2154,6 +2178,7 @@ export class AgentManager {
const nextRecord = {
...record,
...(patch.parent !== undefined ? { parent: patch.parent } : {}),
...(patch.title ? { title: patch.title } : {}),
...(patch.labels ? { labels: applyLabelPatch(record.labels, patch.labels) } : {}),
updatedAt: this.nextStoredUpdatedAt(record),
@@ -2179,36 +2204,36 @@ export class AgentManager {
const liveAgent = this.agents.get(agentId);
if (liveAgent) {
const previousParentAgentId = getParentAgentIdFromLabels(liveAgent.labels);
if (!previousParentAgentId) {
const previous = {
parent: liveAgent.parent,
labels: liveAgent.labels,
updatedAt: liveAgent.updatedAt,
};
liveAgent.parent = null;
liveAgent.labels = applyLabelPatch(
liveAgent.labels,
detachedAgentLabelPatch(liveAgent.labels),
);
this.touchUpdatedAt(liveAgent);
try {
await this.persistSnapshot(liveAgent);
const record = await registry.get(agentId);
if (!record) {
throw new Error(`Agent not found in storage after detach: ${agentId}`);
}
return { record, live: true, previousParentAgentId: null };
}
const { record } = await this.writeLabels(agentId, detachedAgentLabelPatch(liveAgent.labels));
if (!record) {
throw new Error(`Agent not found in storage after detach: ${agentId}`);
} catch (error) {
Object.assign(liveAgent, previous);
throw error;
}
this.emitState(liveAgent, { persist: false });
const record = await registry.get(agentId);
if (!record) throw new Error(`Agent not found in storage after detach: ${agentId}`);
return { record, live: true, previousParentAgentId };
}
const record = await registry.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
}
if (!record) throw new Error(`Agent not found: ${agentId}`);
const previousParentAgentId = getParentAgentIdFromLabels(record.labels);
if (!previousParentAgentId) {
return { record, live: false, previousParentAgentId: null };
}
const result = await this.writeLabels(agentId, detachedAgentLabelPatch(record.labels));
if (!result.record) {
throw new Error(`Agent not found in storage after detach: ${agentId}`);
}
return { record: result.record, live: false, previousParentAgentId };
const detached = await this.writeStoredMetadata(agentId, {
parent: null,
labels: detachedAgentLabelPatch(record.labels),
});
return { record: detached, live: false, previousParentAgentId };
}
notifyAgentState(agentId: string): void {
@@ -2546,7 +2571,19 @@ export class AgentManager {
}): Promise<string> {
const { agent, agentId, pendingRun, prompt, options } = params;
try {
const submissionId = options?.clientMessageId ?? pendingRun.token;
await this.runJournal.prepare(agentId, submissionId);
const result = await agent.session.startTurn(prompt, options);
try {
await this.runJournal.start(agentId, submissionId, result.turnId);
} catch (error) {
// The provider already admitted this turn. A journal failure cannot turn
// accepted work into a rejection or justify submitting the prompt again.
this.logger.error(
{ err: error, agentId, submissionId, turnId: result.turnId },
"Failed to persist admitted run identity",
);
}
if (pendingRun.settled) {
throw new Error(`Agent ${agentId} run was canceled before its turn started`);
}
@@ -2564,6 +2601,12 @@ export class AgentManager {
}
agent.pendingReplacement = false;
const errorMsg = error instanceof Error ? error.message : "Failed to start turn";
if (!(error instanceof Error && "code" in error && error.code === "submission_exists"))
await this.runJournal
.failAdmission(agentId, options?.clientMessageId ?? pendingRun.token, errorMsg)
.catch((journalError) => {
this.logger.error({ err: journalError, agentId }, "Failed to persist rejected run");
});
pendingRun.start = { status: "failed", error: errorMsg };
// A terminal rejection belongs after the submitted prompt even though no provider turn exists.
if (options?.clientMessageId) {
@@ -3574,6 +3617,7 @@ export class AgentManager {
initialTitle?: string | null;
publishWhenReady?: boolean;
workspaceId?: string;
parent?: AgentAddress | null;
owner?: AgentOwner;
},
): Promise<ManagedAgent> {
@@ -3752,6 +3796,7 @@ export class AgentManager {
attention?: AttentionState;
persistence?: AgentPersistenceHandle;
workspaceId?: string;
parent?: AgentAddress | null;
owner?: AgentOwner;
}
| undefined;
@@ -3762,6 +3807,7 @@ export class AgentManager {
provider: config.provider,
cwd: config.cwd,
workspaceId: options?.workspaceId,
parent: options?.parent,
owner: options?.owner,
session,
capabilities: session.capabilities,
@@ -4325,6 +4371,13 @@ export class AgentManager {
const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true };
const runResult = this.getRunResult(
agent.id,
eventTurnId,
event,
options?.fromHistory === true,
terminalDisposition,
);
const dispatchPromise = this.dispatchStreamEventByType({
agent,
event,
@@ -4339,6 +4392,13 @@ export class AgentManager {
}
if (!options?.fromHistory) {
if (terminalDisposition !== "stale")
await this.runJournal.capture(agent.id, eventTurnId, event, runResult).catch((error) => {
this.logger.error(
{ err: error, agentId: agent.id, turnId: eventTurnId },
"Failed to persist provider checkpoint",
);
});
if (isTurnTerminalEvent(event)) {
this.runs.settleTerminalRun(agent.id, eventTurnId);
if (isForegroundEvent) {
@@ -4356,6 +4416,22 @@ export class AgentManager {
return flags.shouldNotifyWaiters;
}
private getRunResult(
agentId: string,
turnId: string | undefined,
event: AgentStreamEvent,
fromHistory: boolean,
disposition: ActiveTurnTerminalDisposition,
): string {
if (fromHistory || disposition === "stale" || !isTurnTerminalEvent(event)) return "";
// Read provider output before terminal handlers append presentation-only errors.
return this.timelineStore
.getRows(agentId)
.filter((row) => row.turnId === turnId)
.flatMap((row) => (row.item.type === "assistant_message" ? [row.item.text] : []))
.join("");
}
private traceHandleStreamEventStart(
agent: ActiveManagedAgent,
event: AgentStreamEvent,
@@ -5104,6 +5180,7 @@ export class AgentManager {
private async flushTasks(options: { includeAgentRegistrations: boolean }): Promise<void> {
this.agentStreamCoalescer.flushAll();
await this.runJournal.flush();
// Drain tasks, including tasks spawned while awaiting.
while (
this.backgroundTasks.size > 0 ||
packages/server/src/server/agent/agent-projections.ts
+4
View File
@@ -83,6 +83,7 @@ export function toStoredAgentRecord(
lastActivityAt: agent.updatedAt.toISOString(),
lastUserMessageAt: agent.lastUserMessageAt ? agent.lastUserMessageAt.toISOString() : null,
title: options?.title ?? null,
parent: agent.parent,
labels: agent.labels,
lastStatus: agent.lifecycle,
lastModeId: agent.currentModeId ?? config?.modeId ?? null,
@@ -138,6 +139,7 @@ export function toAgentPayload(
pendingPermissions: sanitizePendingPermissions(agent.pendingPermissions),
persistence: projectPersistenceHandleForWire(agent.persistence),
title: options?.title ?? null,
parent: agent.parent,
labels: agent.labels,
...(agent.internal ? { internal: true } : {}),
};
@@ -249,6 +251,7 @@ export function buildStoredAgentPayload(
attentionTimestamp: record.attentionTimestamp ?? null,
archivedAt: record.archivedAt ?? null,
...(record.internal ? { internal: true } : {}),
parent: record.parent,
labels: normalizeLabels(record.labels),
...(typeof record.lastError === "string" ? { lastError: record.lastError } : {}),
...(providerAvailable ? {} : { providerUnavailable: true }),
@@ -273,6 +276,7 @@ export function toAgentListItemPayload(agent: AgentSnapshotPayload): AgentListIt
requiresAttention: agent.requiresAttention ?? false,
attentionReason: agent.attentionReason ?? null,
attentionTimestamp: agent.attentionTimestamp ?? null,
parent: agent.parent,
labels: agent.labels,
...(agent.providerUnavailable ? { providerUnavailable: true } : {}),
};
packages/server/src/server/agent/agent-prompt.test.ts
+11 -2
View File
@@ -91,7 +91,11 @@ function createFinishNotificationScenario(
Reflect.set(callerAgent, "lifecycle", "idle");
Reflect.set(callerAgent, "config", { title: "Caller Agent" });
const agentManager = new AgentManager({ clients: {}, logger: createTestLogger() });
const agentManager = new AgentManager({
serverId: "srv_test",
clients: {},
logger: createTestLogger(),
});
Reflect.set(agentManager, "getAgent", (agentId: string) => {
if (agentId === "child-agent") {
return childAgent;
@@ -490,7 +494,11 @@ it("does not notify archived callers", async () => {
const streamAgentSpy = vi.fn(() => (async function* noop() {})());
const replaceAgentRunSpy = vi.fn(() => (async function* noop() {})());
const agentManager = new AgentManager({ clients: {}, logger: createTestLogger() });
const agentManager = new AgentManager({
serverId: "srv_test",
clients: {},
logger: createTestLogger(),
});
Reflect.set(
agentManager,
"getAgent",
@@ -701,6 +709,7 @@ async function createRunStartScenario(startDelayMs: number | null): Promise<{
const workdir = mkdtempSync(join(tmpdir(), "agent-run-start-budget-"));
const client = new SlowStartAgentClient(startDelayMs);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
logger: createTestLogger(),
});
packages/server/src/server/agent/agent-response-loop.real.e2e.test.ts
+2
View File
@@ -38,6 +38,7 @@ async function startAgentMcpServer(logger: pino.Logger): Promise<AgentMcpServerH
const storagePath = path.join(registryDir, "agents");
const agentStorage = new AgentStorage(storagePath, logger);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: {},
registry: agentStorage,
logger,
@@ -172,6 +173,7 @@ describe("getStructuredAgentResponse (e2e)", () => {
beforeEach(async () => {
cwd = mkdtempSync(path.join(tmpdir(), "agent-response-loop-"));
manager = new AgentManager({
serverId: "srv_test",
clients: createRealProviderClients(["codex", "claude"], logger),
logger,
});
packages/server/src/server/agent/agent-storage.ts
+2
View File
@@ -1,3 +1,4 @@
import { AgentAddressSchema } from "@getpaseo/protocol/addresses";
import { promises as fs, type Dirent } from "node:fs";
import path from "node:path";
import { z } from "zod";
@@ -43,6 +44,7 @@ const PERSISTENCE_HANDLE_SCHEMA = z
.optional();
const STORED_AGENT_SCHEMA = z.object({
parent: AgentAddressSchema.nullable().optional(),
id: z.string(),
provider: z.string(),
cwd: z.string(),
packages/server/src/server/agent/create-agent/create.test.ts
+1
View File
@@ -17,6 +17,7 @@ const logger = createTestLogger();
function createRealAgentManager(storage: AgentStorage): AgentManager {
return new AgentManager({
serverId: "srv_test",
clients: createTestAgentClients(),
registry: storage,
logger,
packages/server/src/server/agent/create-agent/create.ts
+3
View File
@@ -1,3 +1,4 @@
import type { AgentAddress } from "@getpaseo/protocol/addresses";
import type { AgentPromptSource } from "../agent-messages/index.js";
import type { Logger } from "pino";
@@ -56,6 +57,7 @@ export type EnsureWorkspaceForCreate = (
) => Promise<string>;
export interface CreateAgentFromSessionInput {
parent?: AgentAddress | null;
kind: "session";
onAgentReady?: (agent: ManagedAgent) => Promise<void>;
agentId?: string;
@@ -289,6 +291,7 @@ async function resolveSessionCreateAgent(
return {
config: sessionConfig,
createOptions: {
parent: input.parent,
labels: input.labels,
initialPrompt: trimmedPrompt,
env: input.env,
packages/server/src/server/agent/mcp-server.test.ts
+34 -4
View File
@@ -1598,6 +1598,7 @@ describe("create_agent MCP tool", () => {
const workdir = await mkdtemp(join(tmpdir(), "mcp-public-helper-removal-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: createTestAgentClients(),
registry: storage,
logger,
@@ -3547,6 +3548,7 @@ describe("create_agent MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", false, { finishTurnsDuringStart: true });
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -3761,6 +3763,7 @@ describe("create_agent MCP tool", () => {
const workdir = await mkdtemp(join(tmpdir(), "mcp-workspace-inherit-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: createTestAgentClients(),
registry: storage,
logger,
@@ -4277,7 +4280,12 @@ describe("send_agent_prompt MCP tool", () => {
const client = new HeldTurnAgentClient("codex", true, {
steerResult: available ? "accepted" : "unavailable",
});
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
registry: storage,
logger,
});
try {
const parent = await manager.createAgent(
{ provider: "codex", cwd: existingCwd },
@@ -4336,7 +4344,12 @@ describe("send_agent_prompt MCP tool", () => {
finishDuringSteer: true,
rejectSteer: rejected,
});
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
registry: storage,
logger,
});
let mcpClient: Client | undefined;
try {
const parent = await manager.createAgent(
@@ -4411,7 +4424,12 @@ describe("send_agent_prompt MCP tool", () => {
finishDuringSteer,
rejectSteer: true,
});
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
registry: storage,
logger,
});
let mcpClient: Client | undefined;
try {
const parent = await manager.createAgent(
@@ -4483,7 +4501,12 @@ describe("send_agent_prompt MCP tool", () => {
finishDuringSteer: false,
rejectSteer: true,
});
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
registry: storage,
logger,
});
let mcpClient: Client | undefined;
try {
const parent = await manager.createAgent(
@@ -4700,6 +4723,7 @@ describe("send_agent_prompt MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", true);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -4779,6 +4803,7 @@ describe("send_agent_prompt MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", true);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -4856,6 +4881,7 @@ describe("send_agent_prompt MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", true);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -4912,6 +4938,7 @@ describe("send_agent_prompt MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", true);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -4978,6 +5005,7 @@ describe("send_agent_prompt MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", true);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -5039,6 +5067,7 @@ describe("send_agent_prompt MCP tool", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const childClient = new HeldTurnAgentClient("codex", true);
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { claude: parentClient, codex: childClient },
registry: storage,
logger,
@@ -5113,6 +5142,7 @@ describe("send_agent_prompt MCP tool to an OMP child", () => {
const parentClient = new HeldTurnAgentClient("claude", false);
const omp = new FakeOmp();
const agentManager = new AgentManager({
serverId: "srv_test",
clients: {
claude: parentClient,
omp: new OmpAgentClient({
packages/server/src/server/agent/plugin-provider.test.ts
+1
View File
@@ -299,6 +299,7 @@ describe("PluginAgentClientRegistry", () => {
},
});
const manager = new AgentManager({
serverId: "srv_test",
logger,
clients: { "plugin-direct": registry["plugin-direct"].createClient(logger) },
providerDefinitions: { "plugin-direct": registry["plugin-direct"] },
packages/server/src/server/agent/provider-options.test.ts
+1
View File
@@ -255,6 +255,7 @@ describe("configured provider options", () => {
},
});
const manager = new AgentManager({
serverId: "srv_test",
logger,
clients: { claude: registry.claude.createClient(logger) },
providerDefinitions: { claude: registry.claude },
packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts
+1
View File
@@ -226,6 +226,7 @@ describe("Codex app-server provider (local e2e)", () => {
const logger = createTestLogger();
const storage = new AgentStorage(path.join(cwd, "agents"), logger);
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: new CodexAppServerAgentClient(logger) },
registry: storage,
logger,
packages/server/src/server/agent/providers/codex-app-server-agent.process-exit.test.ts
+8
View File
@@ -141,6 +141,7 @@ test("unexpected Codex app-server exit fails the active run and agent", async ()
const workdir = mkdtempSync(join(tmpdir(), "codex-process-exit-"));
const appServer = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: new ProcessExitCodexClient([appServer]) },
logger,
});
@@ -201,6 +202,7 @@ test("replacement self-heals when Codex reports that the tracked turn is already
}),
});
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: new ProcessExitCodexClient([appServer]) },
logger,
});
@@ -253,6 +255,7 @@ test("intentional agent close stays clean while a Codex turn is active", async (
const workdir = mkdtempSync(join(tmpdir(), "codex-process-close-"));
const appServer = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: new ProcessExitCodexClient([appServer]) },
logger,
});
@@ -302,6 +305,7 @@ test("idle Codex app-server exit reconnects without publishing a failed turn", a
const exitedAppServer = createFakeCodexAppServer();
const replacementAppServer = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]),
},
@@ -355,6 +359,7 @@ test("failed reconnect preserves manager events for a later successful run", asy
});
const successfulReconnect = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: new ProcessExitCodexClient([exitedAppServer, failedReconnect, successfulReconnect]),
},
@@ -395,6 +400,7 @@ test("unexpected exit fails an autonomous Codex turn", async () => {
const workdir = mkdtempSync(join(tmpdir(), "codex-process-autonomous-exit-"));
const appServer = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: new ProcessExitCodexClient([appServer]) },
logger,
});
@@ -436,6 +442,7 @@ test("provider permissions do not survive an unexpected exit and reconnect", asy
const exitedAppServer = createFakeCodexAppServer();
const replacementAppServer = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]),
},
@@ -512,6 +519,7 @@ test("idle provider exit preserves a synthetic plan approval across reconnect",
const exitedAppServer = createFakeCodexAppServer();
const replacementAppServer = createFakeCodexAppServer();
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: new ProcessExitCodexClient([exitedAppServer, replacementAppServer]),
},
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
+5 -1
View File
@@ -7149,7 +7149,11 @@ describe("Codex interactive resume startup", () => {
return session;
},
};
const manager = new AgentManager({ clients: { codex: client }, logger: createTestLogger() });
const manager = new AgentManager({
serverId: "srv_test",
clients: { codex: client },
logger: createTestLogger(),
});
let id: string | undefined;
try {
const resumed = await manager.resumeAgentFromPersistence(
packages/server/src/server/agent/providers/codex-async-questions.test.ts
+1
View File
@@ -186,6 +186,7 @@ test("failed rewind preserves question state", async () => {
async function manage(session: CodexAppServerAgentSession) {
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: {
provider: "codex",
packages/server/src/server/agent/providers/mock-load-test-agent.test.ts
+1
View File
@@ -500,6 +500,7 @@ describe("MockLoadTestAgentClient", () => {
try {
const client = new MockLoadTestAgentClient();
const manager = new AgentManager({
serverId: "srv_test",
clients: { mock: client },
idFactory: () => "00000000-0000-4000-8000-000000000001",
logger: createTestLogger(),
packages/server/src/server/agent/providers/opencode-bridge.local.e2e.test.ts
+1 -1
View File
@@ -33,7 +33,7 @@ test("real OpenCode server advertises agent-scoped notification descriptions", a
const logger = createTestLogger();
const providerSnapshotManager = new ProviderSnapshotManager({ logger });
const dependencies = {
agentManager: new AgentManager({ logger }),
agentManager: new AgentManager({ serverId: "srv_test", logger }),
agentStorage: new AgentStorage(path.join(root, "agents"), logger),
providerSnapshotManager,
logger,
packages/server/src/server/agent/providers/opencode/bridge-input-errors.test.ts
+1 -1
View File
@@ -22,7 +22,7 @@ test("bridge preserves tool failure messages and classifies invalid requests as
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-7045-"));
const logger = createTestLogger();
const providerSnapshotManager = new ProviderSnapshotManager({ logger });
const agentManager = new AgentManager({ logger });
const agentManager = new AgentManager({ serverId: "srv_test", logger });
const agentStorage = new AgentStorage(paseoHome, logger);
const unused = async (): Promise<never> => {
throw new Error("Unexpected workspace or agent creation");
packages/server/src/server/agent/providers/opencode/bridge.test.ts
+7 -2
View File
@@ -81,7 +81,12 @@ describe("OpenCodeBridge", () => {
const logger = createTestLogger();
const clients = createTestAgentClients();
const agentStorage = new AgentStorage(path.join(paseoHome, "agents"), logger);
const agentManager = new AgentManager({ clients, registry: agentStorage, logger });
const agentManager = new AgentManager({
serverId: "srv_test",
clients,
registry: agentStorage,
logger,
});
const providerSnapshotManager = new ProviderSnapshotManager({ logger, extraClients: clients });
const dependencies = { agentManager, agentStorage, providerSnapshotManager, logger };
const bridge = new OpenCodeBridge({ paseoHome, logger });
@@ -466,7 +471,7 @@ describe("OpenCodeBridge", () => {
const logger = createTestLogger();
const providerSnapshotManager = new ProviderSnapshotManager({ logger });
const catalog = createPaseoToolCatalog({
agentManager: new AgentManager({ logger }),
agentManager: new AgentManager({ serverId: "srv_test", logger }),
agentStorage: new AgentStorage(paseoHome, logger),
providerSnapshotManager,
logger,
packages/server/src/server/agent/providers/provider-availability.test.ts
+2
View File
@@ -122,6 +122,7 @@ describe("default provider availability", () => {
const workdir = makeTempDir("provider-availability-manager-work-");
const storage = new AgentStorage(join(workdir, "agents"), createTestLogger());
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: new CodexAppServerAgentClient(createTestLogger()),
},
@@ -144,6 +145,7 @@ describe("default provider availability", () => {
const workdir = makeTempDir("provider-availability-resume-work-");
const storage = new AgentStorage(join(workdir, "agents"), createTestLogger());
const manager = new AgentManager({
serverId: "srv_test",
clients: {
codex: new CodexAppServerAgentClient(createTestLogger()),
},
packages/server/src/server/agent/rewind/rewind.test.ts
+2
View File
@@ -52,6 +52,7 @@ class RewindHistoryGate {
async function createRewindHarness(options: { historyGate?: RewindHistoryGate } = {}) {
const session = new FakeRewindSession(options.historyGate?.wait.bind(options.historyGate));
const manager = new AgentManager({
serverId: "srv_test",
clients: { claude: new FakeRewindClient(session) },
logger: createTestLogger(),
idFactory: () => "00000000-0000-4000-8000-000000000901",
@@ -142,6 +143,7 @@ describe("AgentManager rewind", () => {
const session = new RejectingInterruptSession();
const manager = new AgentManager({
serverId: "srv_test",
clients: { claude: new FakeRewindClient(session) },
logger: createTestLogger(),
idFactory: () => "00000000-0000-4000-8000-000000000902",
packages/server/src/server/agent/runs/index.ts
+209
View File
@@ -0,0 +1,209 @@
import { createHash } from "node:crypto";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import {
AgentRunRecordSchema,
type AgentRunRecord,
type AgentRunCheckpoint,
} from "@getpaseo/protocol/agent-runs";
import type { AgentStreamEvent } from "../agent-sdk-types.js";
import { writeJsonFileAtomic } from "../../atomic-file.js";
/** Durable submission/turn identity and checkpoints; idle agent state is never a run result. */
export class AgentRuns {
private readonly records = new Map<string, AgentRunRecord>();
private readonly submissionsByTurn = new Map<string, string>();
private readonly loadedAgents = new Set<string>();
private queue: Promise<unknown> = Promise.resolve();
constructor(private readonly options: { serverId: string; directory?: string }) {}
private identity(agentId: string, submissionId: string): string {
return JSON.stringify([agentId, submissionId]);
}
private async load(agentId: string): Promise<void> {
if (this.loadedAgents.has(agentId)) return;
let records: AgentRunRecord[] = [];
if (this.options.directory) {
try {
const saved: unknown = JSON.parse(await readFile(this.path(agentId), "utf8"));
records = AgentRunRecordSchema.array().parse(saved);
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
}
}
for (const record of records) {
if (record.agent.serverId !== this.options.serverId || record.agent.agentId !== agentId)
throw new Error("Run journal identity does not match its owning agent");
if (!terminal(record) && record.status !== "interrupted")
this.append(record, {
kind: "interrupted",
error: "Destination restarted before terminal provider evidence was recorded",
});
this.records.set(this.identity(agentId, record.submissionId), record);
if (record.turnId && !terminal(record))
this.submissionsByTurn.set(this.identity(agentId, record.turnId), record.submissionId);
}
if (records.length && this.options.directory)
await writeJsonFileAtomic(this.path(agentId), records);
this.loadedAgents.add(agentId);
}
private async read(agentId: string, submissionId: string): Promise<AgentRunRecord | null> {
await this.load(agentId);
const record = this.records.get(this.identity(agentId, submissionId));
return record ? structuredClone(record) : null;
}
private path(agentId: string): string {
if (!this.options.directory) throw new Error("Run journal has no persistence directory");
return join(
this.options.directory,
`${createHash("sha256").update(agentId).digest("hex")}.json`,
);
}
private async persistAgent(agentId: string): Promise<void> {
const records = [...this.records.values()]
.filter((record) => record.agent.agentId === agentId)
.map((record) => Object.assign({}, record, { durability: "persisted" as const }));
if (this.options.directory) await writeJsonFileAtomic(this.path(agentId), records);
for (const record of records)
this.records.set(this.identity(agentId, record.submissionId), record);
}
private async write(record: AgentRunRecord, retainEvidence = false): Promise<void> {
const key = this.identity(record.agent.agentId, record.submissionId);
const previous = this.records.get(key);
this.records.set(key, { ...record, durability: "pending" });
try {
await this.persistAgent(record.agent.agentId);
} catch (error) {
if (!retainEvidence) {
if (previous) this.records.set(key, previous);
else this.records.delete(key);
}
throw error;
}
}
private mutate<T>(operation: () => Promise<T>): Promise<T> {
const result = this.queue.then(operation);
this.queue = result.catch(() => {});
return result;
}
async get(agentId: string, submissionId: string): Promise<AgentRunRecord | null> {
return this.mutate(async () => {
await this.load(agentId);
if (
[...this.records.values()].some(
(record) => record.agent.agentId === agentId && record.durability === "pending",
)
) {
// Keep observed evidence readable while storage is unavailable. A later
// read or checkpoint retries the same atomic collection, without replay.
await this.persistAgent(agentId).catch(() => {});
}
return this.read(agentId, submissionId);
});
}
prepare(agentId: string, submissionId: string): Promise<void> {
return this.mutate(async () => {
if (await this.read(agentId, submissionId))
throw Object.assign(
new Error(
"This run submission already exists; inspect it before submitting different work",
),
{ code: "submission_exists" },
);
await this.write({
agent: { serverId: this.options.serverId, agentId },
submissionId,
turnId: null,
status: "submitted",
checkpoints: [{ sequence: 1, kind: "submitted", at: new Date().toISOString() }],
});
});
}
start(agentId: string, submissionId: string, turnId: string): Promise<void> {
return this.mutate(async () => {
const record = await this.read(agentId, submissionId);
if (!record) throw new Error("Run was not prepared before provider admission");
if (record.turnId !== null && record.turnId !== turnId)
throw new Error("Run submission cannot bind to a different provider turn");
record.turnId = turnId;
this.append(record, { kind: "running" });
// Admission has happened. Retain its binding even if durable storage fails;
// subsequent provider evidence can retry persistence without replaying work.
this.records.set(this.identity(agentId, submissionId), record);
this.submissionsByTurn.set(this.identity(agentId, turnId), submissionId);
await this.write(record, true);
});
}
capture(
agentId: string,
turnId: string | undefined,
event: AgentStreamEvent,
result: string,
): Promise<void> {
if (!turnId) return Promise.resolve();
const checkpoint = checkpointFor(event, result);
if (!checkpoint) return Promise.resolve();
return this.mutate(async () => {
await this.load(agentId);
const submissionId = this.submissionsByTurn.get(this.identity(agentId, turnId));
if (!submissionId) return;
const record = await this.read(agentId, submissionId);
if (!record || terminal(record)) return;
if (
checkpoint.permissionRequestId &&
record.checkpoints.some(
(existing) =>
existing.kind === checkpoint.kind &&
existing.permissionRequestId === checkpoint.permissionRequestId,
)
)
return;
this.append(record, checkpoint);
if (terminal(record)) this.submissionsByTurn.delete(this.identity(agentId, turnId));
await this.write(record, true);
});
}
failAdmission(agentId: string, submissionId: string, error: string): Promise<void> {
return this.mutate(async () => {
const record = await this.read(agentId, submissionId);
if (!record || terminal(record)) return;
this.append(record, { kind: "failed", error });
await this.write(record, true);
});
}
private append(record: AgentRunRecord, input: Omit<AgentRunCheckpoint, "sequence" | "at">): void {
record.checkpoints.push({
...input,
sequence: record.checkpoints.length + 1,
at: new Date().toISOString(),
});
record.status = input.kind === "resumed" ? "running" : input.kind;
}
async flush(): Promise<void> {
await this.queue;
}
}
function terminal(record: AgentRunRecord): boolean {
return (
record.status === "completed" || record.status === "failed" || record.status === "canceled"
);
}
function checkpointFor(
event: AgentStreamEvent,
result: string,
): Omit<AgentRunCheckpoint, "sequence" | "at"> | null {
switch (event.type) {
case "permission_requested":
return { kind: "permission", permissionRequestId: event.request.id };
case "permission_resolved":
return { kind: "resumed", permissionRequestId: event.requestId };
case "turn_completed":
return { kind: "completed", result };
case "turn_failed":
return { kind: "failed", error: event.error, result };
case "turn_canceled":
return { kind: "canceled", error: event.reason, result };
default:
return null;
}
}
packages/server/src/server/authorization/operation-permissions.ts
+13
View File
@@ -6,6 +6,12 @@ type OutboundOperation = SessionOutboundMessage["type"];
export type PermissionRequirement = DaemonPermission | readonly DaemonPermission[] | null;
const INBOUND_PERMISSION = {
"agent.create.cancel.request": "workspace.write",
"agent.runs.get.request": "workspace.read",
"agent.subagents.detach.request": "workspace.write",
"agent.subagents.forget.request": "workspace.write",
"agent.subagents.create.request": "workspace.write",
"agent.subagents.list.request": "workspace.read",
"hosts.list.request": "daemon.read",
"hosts.add.request": "tunnel.manage",
"hosts.update.request": "tunnel.manage",
@@ -224,6 +230,13 @@ const INBOUND_PERMISSION = {
} as const satisfies Record<InboundOperation, PermissionRequirement>;
const OUTBOUND_PERMISSION = {
"agent.create.cancel.response": "workspace.write",
"agent.runs.get.response": "workspace.read",
"agent.subagents.detach.response": "workspace.write",
"agent.subagents.forget.response": "workspace.write",
"agent.subagents.create.response": "workspace.write",
"agent.subagents.list.response": "workspace.read",
"agent.subagents.changed": "workspace.read",
"hosts.list.response": "daemon.read",
"hosts.changed": "daemon.read",
"hosts.add.response": "tunnel.manage",
packages/server/src/server/bootstrap.ts
+2
View File
@@ -952,6 +952,8 @@ export async function createPaseoDaemon(
});
const initialAgentManagerState = providerSnapshotManager.getAgentManagerProviderState();
const agentManager = new AgentManager({
runJournalDirectory: path.join(config.paseoHome, "agent-runs"),
serverId,
pluginLifecycle: pluginRuntime,
clients: initialAgentManagerState.clients,
providerDefinitions: initialAgentManagerState.providerDefinitions,
packages/server/src/server/connected-hosts/hosts.e2e.test.ts
+982 -3
View File
File diff suppressed because it is too large Load Diff
packages/server/src/server/connected-hosts/index.ts
+2
View File
@@ -60,6 +60,7 @@ export class ConnectedHosts {
constructor(
private readonly home: string,
private readonly self: HostDirectory["self"],
private readonly assertRemovable: (serverId: string) => Promise<void> = async () => {},
) {
this.ready = this.load();
void this.ready.catch(() => {});
@@ -215,6 +216,7 @@ export class ConnectedHosts {
"permission_denied",
"Only the source or access manager can remove this host. Disable it to stop access.",
);
if (row.serverId) await this.assertRemovable(row.serverId);
const rows = this.rows.filter((entry) => entry !== row);
await this.persist(rows);
this.stop(id, "host_unknown");
packages/server/src/server/connected-hosts/session.ts
+2 -2
View File
@@ -74,8 +74,8 @@ export class ConnectedHostSession {
"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);
else if (error instanceof Error && "code" in error && typeof error.code === "string")
failure = new HostError(error.code, error.message);
if (message.type === "hosts.route.frame.request") {
const route = this.routes.get(message.routeId);
if (route && route.source === this.delivery.currentSource)
packages/server/src/server/creation/index.test.ts
+37
View File
@@ -322,3 +322,40 @@ test("observer failures are logged without failing creation", async () => {
}),
);
});
test("cancellation preserves an admitted outcome-unknown receipt after owner restart", async () => {
const f = await fixture();
const entered = deferred();
const input: CreationInput = {
kind: "agent",
key: "interrupted-admission",
request: { task: "start exactly once" },
agentId: agent.id,
hasAgent: true,
hasPrompt: true,
exists: async () => false,
readAgent: async () => null,
createAgent: async () => {
entered.resolve();
return new Promise(() => {});
},
};
void f.service.create(input);
await entered.promise;
const restarted = new CreationService(f.directory, silentLogger);
expect(await restarted.cancelAgent(input.key, agent.id)).toMatchObject({
status: "admitted",
creation: { phase: "accepted", agentId: agent.id },
});
const observation = await restarted.subscribe("agent", input.key, () => {});
expect(observation.snapshot).toMatchObject({
phase: "failed",
failedStage: "agent",
outcomeUnknown: true,
});
expect(await restarted.cancelAgent(input.key, agent.id)).toMatchObject({
status: "admitted",
creation: { phase: "failed", outcomeUnknown: true },
});
observation.unsubscribe();
});
packages/server/src/server/creation/index.ts
+83 -1
View File
@@ -65,6 +65,8 @@ export class CreationService {
const identity = identityFor(input.kind, input.key);
const admitted = this.admission.then(async () => {
await mkdir(this.directory, { recursive: true });
const canceled = await this.canceledSubmission(input, identity);
if (canceled) return { completion: Promise.resolve(canceled), snapshot: canceled };
const fingerprint = digest(input.request);
let record = await this.read(identity);
if (!record) record = await this.readLegacyAgent(input, fingerprint);
@@ -122,6 +124,62 @@ export class CreationService {
}
}
private async canceledSubmission(
input: CreationInput,
identity: string,
): Promise<CreationSnapshot | null> {
if (input.kind !== "agent") return null;
const agentId = await this.readCancellation(identity);
if (!agentId) return null;
if (input.agentId && input.agentId !== agentId) throw conflict();
return cancellationSnapshot(input.key, agentId);
}
/** Fences a creation identity only before admission. Running receipts remain owned by their runner. */
cancelAgent(
key: string,
agentId: string,
): Promise<{ status: "canceled" } | { status: "admitted"; creation: CreationSnapshot }> {
const identity = identityFor("agent", key);
const operation = this.admission.then(async () => {
await mkdir(this.directory, { recursive: true });
const record = await this.read(identity);
if (record) {
if (record.snapshot.agentId !== agentId) throw conflict();
return { status: "admitted" as const, creation: record.snapshot };
}
// COMPAT(agentRequestReceipts): added in v0.8.0, remove after 2027-03-11 once old creation receipts can expire.
// A legacy receipt is evidence of admission, never permission to cancel an uncertain runner.
if (
this.legacyDirectory &&
(await readOptional(join(this.legacyDirectory, `${digest(["create", key])}.json`))) !== null
)
throw Object.assign(new Error("Cannot cancel an existing legacy creation receipt"), {
code: "creation_already_admitted",
});
const canceledAgentId = await this.readCancellation(identity);
if (canceledAgentId && canceledAgentId !== agentId) throw conflict();
await writeFileAtomic(
join(this.directory, `${identity}.canceled.json`),
JSON.stringify({ agentId }),
);
const snapshot = cancellationSnapshot(key, agentId);
for (const observer of this.observers.get(identity) ?? []) this.notify(observer, snapshot);
return { status: "canceled" as const };
});
this.admission = operation.then(
() => undefined,
() => undefined,
);
return operation;
}
private async readCancellation(identity: string): Promise<string | null> {
const text = await readOptional(join(this.directory, `${identity}.canceled.json`));
return text === null
? null
: z.object({ agentId: z.string().uuid() }).parse(JSON.parse(text)).agentId;
}
async subscribe(
kind: CreationSnapshot["kind"],
key: string,
@@ -141,7 +199,12 @@ export class CreationService {
});
}
if (record?.snapshot.phase === "completed") await this.validateCompleted(record.snapshot);
return { snapshot: record?.snapshot ?? null, unsubscribe };
const canceledAgentId = kind === "agent" ? await this.readCancellation(identity) : null;
return {
snapshot:
record?.snapshot ?? (canceledAgentId ? cancellationSnapshot(key, canceledAgentId) : null),
unsubscribe,
};
} catch (error) {
unsubscribe();
throw error;
@@ -375,3 +438,22 @@ async function readOptional(file: string): Promise<string | null> {
throw error;
}
}
function conflict(): Error {
return Object.assign(new Error("The creation key already reserves a different agent identity"), {
code: "agent_request_key_conflict",
});
}
function cancellationSnapshot(key: string, agentId: string): CreationSnapshot {
return {
kind: "agent",
idempotencyKey: key,
revision: 0,
phase: "failed",
workspaceId: null,
agentId,
error: "Creation was canceled before admission",
errorCode: "creation_canceled",
failedStage: "agent",
};
}
packages/server/src/server/host-runtime/index.ts
+20 -15
View File
@@ -6,6 +6,7 @@ import type { DaemonPermission } from "@getpaseo/protocol/messages";
import type { SessionAdmission } from "../websocket-server.js";
export interface HostRuntimeAuthority {
sourceId?: string;
permissionCeiling: readonly DaemonPermission[];
resolveAdmission(): Promise<SessionAdmission>;
}
@@ -103,21 +104,25 @@ class OwnedHostRuntime extends HostRuntime {
override dispose(): Promise<void> {
if (this.ownedDisposal) return this.ownedDisposal;
this.ownedDisposal = (async () => {
try {
await super.dispose();
} finally {
try {
await this.originApi.dispose();
} finally {
try {
await this.originClient.close();
} finally {
this.onDisposed(this);
}
}
}
})();
this.stopAcquisitions();
this.ownedDisposal = Promise.resolve().then(() => this.disposeOwnedResources());
return this.ownedDisposal;
}
private async disposeOwnedResources(): Promise<void> {
try {
try {
// This adapter owns the transport. Stop it before dependent releases
// that cannot receive acknowledgments through a stopping daemon.
await this.originClient.close();
} finally {
await super.dispose();
}
} finally {
try {
await this.originApi.dispose();
} finally {
this.onDisposed(this);
}
}
}
}
packages/server/src/server/hub/test-utils/relationship-harness.ts
+1
View File
@@ -1266,6 +1266,7 @@ export class HubRelationshipHarness {
pino({ level: "silent" }),
);
const manager = new AgentManager({
serverId: "srv_test",
clients: createTestAgentClients(),
registry: storage,
logger: pino({ level: "silent" }),
packages/server/src/server/schedule/service.test.ts
+72 -56
View File
@@ -341,7 +341,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -383,7 +383,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -435,7 +435,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -462,7 +462,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -502,7 +502,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -537,7 +537,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -586,7 +586,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -615,7 +615,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -651,7 +651,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -684,6 +684,7 @@ describe("ScheduleService", () => {
test("executes new-agent schedules through AgentManager with real fake clients", async () => {
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -724,6 +725,7 @@ describe("ScheduleService", () => {
test("delivers agent-target schedules through the steer-or-interrupt path", async () => {
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -758,6 +760,7 @@ describe("ScheduleService", () => {
test("titles scheduled new agents from the schedule prompt", async () => {
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -801,7 +804,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
createDirectoryWorkspace: createScheduleDirectoryWorkspace,
@@ -838,6 +841,7 @@ describe("ScheduleService", () => {
createArchiveWorkspace,
} = await createRegistryBackedScheduleWorkspaceDeps(tempDir);
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -906,6 +910,7 @@ describe("ScheduleService", () => {
createArchiveWorkspace,
} = await createRegistryBackedScheduleWorkspaceDeps(tempDir);
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -970,6 +975,7 @@ describe("ScheduleService", () => {
createArchiveWorkspace,
} = await createRegistryBackedScheduleWorkspaceDeps(tempDir);
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -1024,7 +1030,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -1087,7 +1093,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -1129,6 +1135,7 @@ describe("ScheduleService", () => {
const createdInputs: Parameters<ScheduleServiceOptions["createAgent"]>[0][] = [];
const runPrompts: AgentPromptInput[] = [];
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -1198,6 +1205,7 @@ describe("ScheduleService", () => {
test("scheduled new-agent run output falls back to final text and curated timeline", async () => {
let runCount = 0;
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -1278,6 +1286,7 @@ describe("ScheduleService", () => {
test("scheduled new-agent cancellations fail the run", async () => {
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -1345,6 +1354,7 @@ describe("ScheduleService", () => {
logger.child = (() => logger) as typeof logger.child;
const archiveError = new Error("archive exploded");
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -1552,6 +1562,7 @@ describe("ScheduleService", () => {
}
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: { claude: new PromptEchoScheduleClient() },
registry: agentStorage,
@@ -1736,6 +1747,7 @@ describe("ScheduleService", () => {
const client = new CountingScheduleClient();
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: { claude: client },
registry: agentStorage,
@@ -1838,6 +1850,7 @@ describe("ScheduleService", () => {
}
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: {
claude: {
@@ -1893,6 +1906,7 @@ describe("ScheduleService", () => {
test("defaults new-agent modeId to provider's unattended mode", async () => {
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
@@ -1962,6 +1976,7 @@ describe("ScheduleService", () => {
isAvailable: () => opencodeClient.isAvailable(),
} satisfies AgentClient;
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients,
registry: agentStorage,
@@ -2033,6 +2048,7 @@ describe("ScheduleService", () => {
isAvailable: () => claudeClient.isAvailable(),
} satisfies AgentClient;
const manager = new AgentManager({
serverId: "srv_test",
logger: createTestLogger(),
clients,
providerDefinitions: { claude: TEST_CLAUDE_PROVIDER_DEFINITION },
@@ -2118,7 +2134,7 @@ describe("ScheduleService", () => {
const service1 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2143,7 +2159,7 @@ describe("ScheduleService", () => {
const service2 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2160,7 +2176,7 @@ describe("ScheduleService", () => {
const service1 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2189,7 +2205,7 @@ describe("ScheduleService", () => {
const service2 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2205,7 +2221,7 @@ describe("ScheduleService", () => {
const service1 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2249,7 +2265,7 @@ describe("ScheduleService", () => {
const service2 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2274,7 +2290,7 @@ describe("ScheduleService", () => {
const service1 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2317,7 +2333,7 @@ describe("ScheduleService", () => {
const service2 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2351,7 +2367,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2396,7 +2412,7 @@ describe("ScheduleService", () => {
});
test("rejects archived target agents before loading them", async () => {
const manager = new AgentManager({ logger: createTestLogger() });
const manager = new AgentManager({ serverId: "srv_test", logger: createTestLogger() });
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
@@ -2461,7 +2477,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2484,7 +2500,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2508,7 +2524,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2531,7 +2547,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2555,7 +2571,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2591,7 +2607,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2652,7 +2668,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2685,7 +2701,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2718,7 +2734,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2750,7 +2766,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2776,7 +2792,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2801,7 +2817,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2846,7 +2862,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2881,7 +2897,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2907,7 +2923,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -2974,7 +2990,7 @@ describe("ScheduleService", () => {
const service1 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3018,7 +3034,7 @@ describe("ScheduleService", () => {
const service2 = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3041,7 +3057,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3073,7 +3089,7 @@ describe("ScheduleService", () => {
service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3104,7 +3120,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3136,7 +3152,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3171,7 +3187,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3207,7 +3223,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3232,7 +3248,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3270,7 +3286,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3301,7 +3317,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3324,7 +3340,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3366,7 +3382,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3401,7 +3417,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3456,7 +3472,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3482,7 +3498,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3520,7 +3536,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3550,7 +3566,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
@@ -3596,7 +3612,7 @@ describe("ScheduleService", () => {
const service = createScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentManager: new AgentManager({ serverId: "srv_test", logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
packages/server/src/server/session.ts
+223 -2
View File
@@ -1,3 +1,5 @@
import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
import type { Subagents, SubagentAuthority } from "./subagents/index.js";
import { AgentDirectory } from "./agent/directory.js";
import { ConnectedHostSession } from "./connected-hosts/index.js";
import type { ConnectedHosts } from "./connected-hosts/index.js";
@@ -444,6 +446,8 @@ const nodeSessionFileSystem: SessionFileSystem = {
type AgentMcpTransportFactory = () => Promise<unknown>;
export interface SessionOptions {
subagents?: Subagents;
subagentAuthority?: () => SubagentAuthority;
connectedHosts?: ConnectedHosts;
sourceId?: string;
routed?: boolean;
@@ -469,7 +473,7 @@ export interface SessionOptions {
agentManager: AgentManager;
agentStorage: AgentStorage;
messageReceipts: Pick<MessageReceipts, "send">;
creationService: Pick<CreationService, "create" | "subscribe">;
creationService: Pick<CreationService, "create" | "subscribe" | "cancelAgent">;
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
directorySync?: DirectorySyncService;
@@ -681,6 +685,10 @@ interface ClientActivity {
}
export class Session {
private readonly subagentOptions: Pick<
SessionOptions,
"subagents" | "subagentAuthority" | "serverId"
>;
private readonly connectedHostSession: ConnectedHostSession | null;
private routed = false;
isRouted(): boolean {
@@ -823,7 +831,7 @@ export class Session {
private readonly workspaceScripts: WorkspaceScriptsService;
private readonly messageReceipts: Pick<MessageReceipts, "send">;
private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch;
private readonly creationService: Pick<CreationService, "create" | "subscribe">;
private readonly creationService: Pick<CreationService, "create" | "subscribe" | "cancelAgent">;
constructor(options: SessionOptions) {
const {
@@ -885,6 +893,7 @@ export class Session {
this.browserToolsBroker = options.browserToolsBroker;
this.clientId = clientId;
this.authorization = new SessionAuthorization(permissions);
this.subagentOptions = options;
this.appVersion = appVersion ?? null;
this.clientCapabilities = parseClientCapabilities(clientCapabilities);
@@ -2247,6 +2256,118 @@ export class Session {
}
}
private async handleSubagents(
message: Extract<
SessionInboundMessage,
{
type:
| "agent.subagents.create.request"
| "agent.subagents.list.request"
| "agent.subagents.detach.request"
| "agent.subagents.forget.request";
}
>,
): Promise<void> {
try {
await this.executeSubagents(message);
} catch (error) {
this.emit({
type: "rpc_error",
payload: {
requestId: message.requestId,
requestType: message.type,
code: error instanceof Error && "code" in error ? String(error.code) : "subagent_error",
error: error instanceof Error ? error.message : String(error),
},
});
}
}
private async executeSubagents(
message: Extract<
SessionInboundMessage,
{
type:
| "agent.subagents.create.request"
| "agent.subagents.list.request"
| "agent.subagents.detach.request"
| "agent.subagents.forget.request";
}
>,
): Promise<void> {
const { subagents, subagentAuthority, serverId } = this.subagentOptions;
if (!subagents || !subagentAuthority || !serverId)
throw new Error("Subagent relationships are not available");
const parent = { serverId, agentId: message.parentId };
if (message.type === "agent.subagents.create.request") {
const child = await subagents.create(parent, message, subagentAuthority());
this.emit({
type: "agent.subagents.create.response",
payload: { requestId: message.requestId, child },
});
return;
}
if (message.type === "agent.subagents.detach.request") {
const relationship = await subagents.detach(parent, message.child, subagentAuthority());
this.emit({
type: "agent.subagents.detach.response",
payload: { requestId: message.requestId, relationship },
});
return;
}
if (message.type === "agent.subagents.forget.request") {
await subagents.forget(parent, message.child, subagentAuthority());
this.emit({
type: "agent.subagents.forget.response",
payload: { requestId: message.requestId },
});
return;
}
// The session's workspace-read permission authorizes the origin parent directory.
if (
!(await this.agentStorage.get(parent.agentId)) &&
!this.agentManager.getAgent(parent.agentId)
)
throw new Error("Parent agent not found");
if (!message.observe) {
this.emit({
type: "agent.subagents.list.response",
payload: {
requestId: message.requestId,
directory: await subagents.list(parent, subagentAuthority()),
},
});
return;
}
let stop = () => {};
const owner = this.delivery.begin("subagents", undefined, () => stop(), parent.agentId);
try {
stop = subagents.observe(() => {
void subagents
.list(parent, subagentAuthority())
.then((directory) => {
owner.emit({
type: "agent.subagents.changed",
payload: { subscriptionId: owner.id, directory },
});
return undefined;
})
.catch(() => owner.release());
});
this.emit({
type: "agent.subagents.list.response",
payload: {
requestId: message.requestId,
subscriptionId: owner.id,
directory: await subagents.list(parent, subagentAuthority()),
},
});
} catch (error) {
await owner.release();
throw error;
}
}
public setPermissions(permissions: readonly DaemonPermission[]): void {
this.authorization.replacePermissions(permissions);
this.connectedHostSession?.revoke();
@@ -2828,6 +2949,8 @@ export class Session {
source?: object,
): Promise<void> | undefined {
switch (msg.type) {
case "agent.create.cancel.request":
return this.handleAgentCreationCancel(msg);
case "agent.create.request":
return this.handleAgentCreation(msg, source);
case "creation.subscribe.request":
@@ -2839,7 +2962,64 @@ export class Session {
}
}
private async handleAgentCreationCancel(
message: Extract<SessionInboundMessage, { type: "agent.create.cancel.request" }>,
): Promise<void> {
try {
// The originating execution-scoped capability cannot be reduced to ordinary grants.
this.subagentOptions.subagentAuthority?.();
const result = await this.creationService.cancelAgent(
message.idempotencyKey,
message.agentId,
);
const serverId = this.subagentOptions.serverId;
if (!serverId) throw new Error("Daemon identity is unavailable");
this.emit({
type: "agent.create.cancel.response",
payload: {
requestId: message.requestId,
cancellation: { ...result, reservedAgent: { serverId, agentId: message.agentId } },
},
});
} catch (error) {
this.emit({
type: "rpc_error",
payload: {
requestId: message.requestId,
requestType: message.type,
code:
error instanceof Error && "code" in error
? String(error.code)
: "creation_cancel_failed",
error: error instanceof Error ? error.message : String(error),
},
});
}
}
private async handleAgentRunGet(
message: Extract<SessionInboundMessage, { type: "agent.runs.get.request" }>,
): Promise<void> {
const run = await this.agentManager.getAgentRun(message.agentId, message.submissionId);
this.emit({ type: "agent.runs.get.response", payload: { requestId: message.requestId, run } });
}
private dispatchSubagentMessage(msg: SessionInboundMessage): Promise<void> | undefined {
if (msg.type === "agent.runs.get.request") return this.handleAgentRunGet(msg);
if (
msg.type === "agent.subagents.create.request" ||
msg.type === "agent.subagents.list.request" ||
msg.type === "agent.subagents.detach.request" ||
msg.type === "agent.subagents.forget.request"
)
return this.handleSubagents(msg);
}
private dispatchAgentLifecycleMessage(msg: SessionInboundMessage): Promise<void> | undefined {
return this.dispatchSubagentMessage(msg) ?? this.dispatchCoreAgentLifecycleMessage(msg);
}
private dispatchCoreAgentLifecycleMessage(msg: SessionInboundMessage): Promise<void> | undefined {
switch (msg.type) {
case "fetch_agents_request":
return this.handleFetchAgents(msg);
@@ -4155,6 +4335,7 @@ export class Session {
requestId,
agent: null,
error: error instanceof Error ? error.message : String(error),
errorCode: error instanceof Error && "code" in error ? String(error.code) : undefined,
},
},
source,
@@ -4355,11 +4536,50 @@ export class Session {
}
}
private async createLegacySubagent(
msg: CreateAgentRequestMessage,
agentId?: string,
onAgentReady?: (agent: AgentSnapshotPayload) => Promise<void>,
): Promise<CreatedAgent | null> {
// COMPAT(qualifiedSubagents): normalize pre-fleet caller/label creation at the session edge.
// Added in v0.12; remove after 2027-04-20 once all creation callers use parent capabilities.
const legacyParentId = msg.callerAgentId ?? msg.labels[PARENT_AGENT_ID_LABEL];
const { subagents, subagentAuthority, serverId } = this.subagentOptions;
if (legacyParentId && subagents && subagentAuthority && serverId) {
const {
type: _type,
requestId: _requestId,
callerAgentId: _caller,
parent: _parent,
...input
} = msg;
const labels = { ...input.labels };
delete labels[PARENT_AGENT_ID_LABEL];
const child = await subagents.create(
{ serverId, agentId: legacyParentId },
{
serverId,
idempotencyKey: msg.idempotencyKey ?? msg.requestId,
input: { ...input, labels, agentId },
},
subagentAuthority(),
);
const record = await this.agentStorage.get(child.agentId);
if (!record) throw new Error("Created subagent is not available");
const agent = this.buildStoredAgentPayload(record);
await onAgentReady?.(agent);
return { agent, initialPromptStarted: hasInitialCreationPrompt(msg) };
}
return null;
}
private async createSessionAgent(
msg: CreateAgentRequestMessage,
agentId?: string,
onAgentReady?: (agent: AgentSnapshotPayload) => Promise<void>,
): Promise<CreatedAgent> {
const delegated = await this.createLegacySubagent(msg, agentId, onAgentReady);
if (delegated) return delegated;
const {
config,
worktreeName,
@@ -4450,6 +4670,7 @@ export class Session {
images,
attachments,
git,
parent: msg.parent,
labels: resolvedIntent.intent.labels,
env,
provisionalTitle,
packages/server/src/server/session.workspaces.test.ts
+3
View File
@@ -958,6 +958,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist
};
const agentStorage = new AgentStorage(path.join(workdir, "agents"), asSessionLogger(logger));
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { codex: new CreateAgentTestClient() },
registry: agentStorage,
logger: asSessionLogger(logger),
@@ -1113,6 +1114,7 @@ test("create_agent_request launches from an exact subdirectory in a created work
};
const agentStorage = new AgentStorage(path.join(workdir, "agents"), asSessionLogger(logger));
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { codex: new CreateAgentTestClient() },
registry: agentStorage,
logger: asSessionLogger(logger),
@@ -1262,6 +1264,7 @@ test("create_agent_request does not title an existing workspace from the agent p
};
const agentStorage = new AgentStorage(path.join(workdir, "agents"), asSessionLogger(logger));
const agentManager = new AgentManager({
serverId: "srv_test",
clients: { codex: new CreateAgentTestClient() },
registry: agentStorage,
logger: asSessionLogger(logger),
packages/server/src/server/session/agent-updates/agent-updates-service.ts
+1
View File
@@ -123,6 +123,7 @@ export function matchesAgentUpdatesFilter(input: {
filter?: AgentUpdatesFilter;
}): boolean {
const { agent, project, filter } = input;
if (filter?.agentIds && !filter.agentIds.includes(agent.id)) return false;
if (filter?.labels) {
const matchesLabels = Object.entries(filter.labels).every(
packages/server/src/server/subagents/index.ts
+814
View File
@@ -0,0 +1,814 @@
import type { HostDirectory } from "@getpaseo/protocol/hosts";
import { shouldDetachFromArchivedParent } from "@getpaseo/protocol/agent-lifecycle";
import { hasOpenAgentTab } from "@getpaseo/protocol/agent-labels";
import { WorkspaceAddressSchema, type WorkspaceAddress } from "@getpaseo/protocol/addresses";
import type { AgentSnapshotPayload, CreationSnapshot } from "@getpaseo/protocol/messages";
import { createHash, randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import { z } from "zod";
import type { HostRuntime } from "@getpaseo/client";
import { AgentAddressSchema, type AgentAddress } from "@getpaseo/protocol/addresses";
import {
DaemonPermissionSchema,
SubagentCreateInputSchema,
SubagentRelationshipSchema,
type DaemonPermission,
type SubagentCreateInput,
type SubagentDirectory,
} from "@getpaseo/protocol/messages";
import { writeJsonFileAtomic } from "../atomic-file.js";
import type { HostRuntimeAuthority } from "../host-runtime/index.js";
/** Stored authority contains identity and a ceiling; live admission belongs to the host. */
export interface SubagentAuthority {
principalId: string;
sourceId: string;
permissionCeiling: readonly DaemonPermission[];
}
const AuthoritySchema = z.object({
principalId: z.string(),
sourceId: z.string(),
permissionCeiling: z.array(DaemonPermissionSchema),
});
const RecordSchema = z.object({
row: SubagentRelationshipSchema,
authority: AuthoritySchema,
creation: SubagentCreateInputSchema,
fingerprint: z.string(),
confirmed: z.enum(["attached", "detached", "archived", "canceled"]).nullable(),
intent: z.object({
revision: z.number().int().nonnegative(),
kind: z.enum(["attached", "detach", "archive-parent", "forgotten"]),
parentWorkspace: WorkspaceAddressSchema.optional(),
authority: AuthoritySchema,
}),
});
type RelationshipRecord = z.infer<typeof RecordSchema>;
type RelationshipIntent = RelationshipRecord["intent"];
/** Authoritative parent membership, independent of views and destination metadata. */
export class Subagents {
private readonly records = new Map<string, RelationshipRecord>();
private readonly listeners = new Set<() => void>();
private readonly settling = new Map<string, Promise<void>>();
private readonly startingObservers = new Map<string, Promise<void>>();
private readonly running = new Map<string, Promise<AgentAddress>>();
private readonly activeAuthorities = new Map<string, number>();
private readonly runtimes = new Map<string, Promise<HostRuntime>>();
private readonly observers = new Map<
string,
{
intent: RelationshipIntent;
release(): Promise<void>;
creation?: () => CreationSnapshot | null;
}
>();
private recovering: Promise<void> | null = null;
private recoveryRequested = false;
private mutation: Promise<unknown> = Promise.resolve();
private readonly ready: Promise<void>;
private closed = false;
constructor(
private readonly options: {
home: string;
serverId: string;
runtime(authority: HostRuntimeAuthority): Promise<HostRuntime>;
resolveAuthority(principalId: string): ReturnType<HostRuntimeAuthority["resolveAdmission"]>;
},
) {
this.ready = this.load();
void this.ready.catch(() => {});
}
private async load(): Promise<void> {
let records: RelationshipRecord[];
try {
records = z.array(RecordSchema).parse(JSON.parse(await readFile(this.path(), "utf8")));
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
throw error;
}
for (const record of records) {
record.row.freshness = "stale";
this.records.set(record.row.id, record);
}
}
private path(): string {
return join(this.options.home, "subagents.json");
}
private transact<T>(operation: () => Promise<T>): Promise<T> {
const result = this.mutation.then(async () => {
await this.ready;
if (this.closed) throw new Error("Subagent relationships are closed");
return operation();
});
this.mutation = result.catch(() => {});
return result;
}
private async persist(): Promise<void> {
for (const record of this.records.values()) record.row.phase = relationshipPhase(record);
await writeJsonFileAtomic(this.path(), [...this.records.values()]);
for (const changed of this.listeners) changed();
}
async assertHostRemovable(serverId: string): Promise<void> {
await this.ready;
if (
[...this.records.values()].some(
({ row }) => row.child.serverId === serverId && !terminalPhase(row.phase),
)
)
throw failure(
"relationships_pending",
"Detach children or explicitly forget their relationships before removing this host",
);
}
observe(changed: () => void): () => void {
this.listeners.add(changed);
return () => {
this.listeners.delete(changed);
};
}
async list(parent: AgentAddress, authority: SubagentAuthority): Promise<SubagentDirectory> {
await this.authorize(authority, "workspace.read");
await this.ready;
return {
parent: AgentAddressSchema.parse(parent),
rows: [...this.records.values()]
.filter(({ row }) => sameAddress(row.parent, parent) && row.phase !== "forgotten")
.map(({ row }) => structuredClone(row)),
};
}
private authorityKey(authority: SubagentAuthority): string {
return JSON.stringify([
authority.principalId,
authority.sourceId,
[...authority.permissionCeiling].sort(),
]);
}
private async usingAuthority<T>(
authority: SubagentAuthority,
operation: () => Promise<T>,
): Promise<T> {
const key = this.authorityKey(authority);
this.activeAuthorities.set(key, (this.activeAuthorities.get(key) ?? 0) + 1);
try {
return await operation();
} finally {
const remaining = this.activeAuthorities.get(key)! - 1;
if (remaining) this.activeAuthorities.set(key, remaining);
else this.activeAuthorities.delete(key);
await this.releaseUnusedRuntimes();
}
}
private async releaseUnusedRuntimes(): Promise<void> {
const used = new Set(this.activeAuthorities.keys());
for (const record of this.records.values()) {
if (terminalPhase(record.row.phase)) continue;
used.add(this.authorityKey(record.authority));
used.add(this.authorityKey(record.intent.authority));
}
const releases: Promise<unknown>[] = [];
for (const [key, runtime] of this.runtimes) {
if (used.has(key)) continue;
this.runtimes.delete(key);
releases.push(runtime.then((scope) => scope.dispose()));
}
await Promise.allSettled(releases);
}
private runtime(authority: SubagentAuthority): Promise<HostRuntime> {
if (this.closed) throw new Error("Subagent relationships are closed");
const key = this.authorityKey(authority);
let pending = this.runtimes.get(key);
if (!pending) {
pending = this.options.runtime({
sourceId: authority.sourceId,
permissionCeiling: authority.permissionCeiling,
resolveAdmission: () => this.options.resolveAuthority(authority.principalId),
});
this.runtimes.set(key, pending);
const acquiring = pending;
void acquiring.catch(() => {
if (this.runtimes.get(key) === acquiring) this.runtimes.delete(key);
});
}
return pending;
}
private async authorize(
authority: SubagentAuthority,
permission: DaemonPermission,
): Promise<void> {
const admission = await this.options.resolveAuthority(authority.principalId);
if (admission.hubExecutionAgents)
throw failure(
"unsupported_capability",
"Execution-scoped subagent relationships are not supported",
);
if (
!admission.permissions.includes(permission) ||
!authority.permissionCeiling.includes(permission)
)
throw failure("access_denied", `Subagent relationships require ${permission}`);
}
async create(
parent: AgentAddress,
creation: SubagentCreateInput,
authority: SubagentAuthority,
): Promise<AgentAddress> {
return this.usingAuthority(authority, () =>
this.createRelationship(parent, creation, authority),
);
}
private async createRelationship(
parent: AgentAddress,
creation: SubagentCreateInput,
authority: SubagentAuthority,
): Promise<AgentAddress> {
parent = AgentAddressSchema.parse(parent);
if (parent.serverId !== this.options.serverId)
throw failure("invalid_parent", "Create subagents through the parent's host");
await this.authorize(authority, "workspace.write");
const runtime = await this.runtime(authority);
const parentHandle = await runtime.agent(parent);
const refreshed = await parentHandle.refresh();
if (!refreshed) throw failure("invalid_parent", "Parent agent not found");
const snapshot = refreshed.agent;
if (snapshot.archivedAt)
throw failure("invalid_parent", "Cannot attach a child to an archived parent");
creation = SubagentCreateInputSchema.parse(creation);
let workspaceId = creation.input.workspaceId;
if (!workspaceId) {
if (creation.serverId !== parent.serverId)
throw failure("workspace_required", "Choose a workspace on the destination host");
workspaceId = snapshot.workspaceId;
if (!workspaceId) throw failure("workspace_required", "The parent has no workspace");
}
creation.input.workspaceId = workspaceId;
const fingerprint = JSON.stringify(creation);
const id = createHash("sha256")
.update(JSON.stringify([authority.sourceId, parent, creation.idempotencyKey]))
.digest("hex");
const record = await this.transact(async () => {
const existing = this.records.get(id);
if (existing) {
if (existing.fingerprint !== fingerprint)
throw failure("idempotency_conflict", "This subagent creation key has different inputs");
return existing;
}
const fresh: RelationshipRecord = {
row: {
id,
parent,
child: { serverId: creation.serverId, agentId: creation.input.agentId ?? randomUUID() },
workspace: { serverId: creation.serverId, workspaceId },
phase: "creating",
observedAt: null,
freshness: "stale",
error: null,
},
authority: { ...authority, permissionCeiling: [...authority.permissionCeiling] },
creation,
fingerprint,
confirmed: null,
intent: {
revision: 0,
kind: "attached",
authority: { ...authority, permissionCeiling: [...authority.permissionCeiling] },
},
};
this.records.set(id, fresh);
try {
await this.persist();
} catch (error) {
this.records.delete(id);
throw error;
}
return fresh;
});
if (record.intent.kind === "forgotten")
throw failure(
"relationship_forgotten",
"This relationship was explicitly forgotten; choose a new creation key to start different work",
);
if (record.confirmed !== null || record.intent.kind !== "attached") {
void this.observeChild(record).catch(() => {});
return record.row.child;
}
let running = this.running.get(id);
if (!running) {
running = this.createChild(record);
this.running.set(id, running);
void running.finally(() => this.running.delete(id)).catch(() => {});
}
return running;
}
private async createChild(record: RelationshipRecord): Promise<AgentAddress> {
const { row, creation } = record;
const creationIntent = record.intent;
try {
const runtime = await this.runtime(record.authority);
const api = await runtime.client(row.child.serverId);
try {
const workspace = api.workspaces.ref(creation.input.workspaceId!);
const { config, initialPrompt, ...input } = creation.input;
const agent = await workspace.agents.create({
...input,
agentId: row.child.agentId,
config: {
...config,
provider: config.model ? `${config.provider}/${config.model}` : config.provider,
},
parent: row.parent,
prompt: initialPrompt,
idempotencyKey: `subagent:${row.id}`,
clientMessageId: `subagent:${row.id}:initial`,
});
await this.transact(async () => {
// Creation confirms a resource; it never owns a newer lifecycle decision.
if (!terminalConfirmation(record)) record.confirmed = "attached";
if (record.intent.kind === "attached") this.captureSnapshot(record, agent.current());
await this.persist();
});
} finally {
await api.dispose();
}
if (record.intent.kind === "attached") void this.observeChild(record).catch(() => {});
else void this.settleObligation(record).catch(() => {});
return row.child;
} catch (error) {
await this.transact(async () => {
if (record.intent !== creationIntent || terminalPhase(record.row.phase)) return;
row.freshness = "unavailable";
row.error = {
code: error instanceof Error && "code" in error ? String(error.code) : "host_error",
message: error instanceof Error ? error.message : String(error),
};
await this.persist();
});
throw error;
}
}
async hostsChanged(directory: HostDirectory): Promise<void> {
await this.ready;
const unavailable = new Set(
directory.hosts.filter((host) => host.status !== "online").map((host) => host.serverId),
);
await this.transact(async () => {
let changed = false;
for (const record of this.records.values()) {
if (unavailable.has(record.row.child.serverId) && !terminalPhase(record.row.phase)) {
record.row.freshness = "unavailable";
changed = true;
}
}
if (changed) await this.persist();
});
await this.recover();
}
async detach(parent: AgentAddress, child: AgentAddress, authority: SubagentAuthority) {
await this.authorize(authority, "workspace.write");
const record = await this.find(parent, child);
if (record.intent.kind === "forgotten")
throw failure("relationship_forgotten", "This relationship was forgotten");
if (record.row.phase === "detached") return structuredClone(record.row);
await this.transact(async () => {
if (record.intent.kind === "forgotten")
throw failure("relationship_forgotten", "This relationship was forgotten");
if (record.intent.kind !== "detach")
record.intent = {
revision: record.intent.revision + 1,
kind: "detach",
authority: {
...authority,
permissionCeiling: authority.permissionCeiling.filter((permission) =>
record.authority.permissionCeiling.includes(permission),
),
},
};
await this.persist();
});
await this.settleObligation(record);
return structuredClone(record.row);
}
async forget(
parent: AgentAddress,
child: AgentAddress,
authority: SubagentAuthority,
): Promise<void> {
await this.authorize(authority, "workspace.write");
const record = await this.find(parent, child);
await this.transact(async () => {
record.intent = {
revision: record.intent.revision + 1,
kind: "forgotten",
authority: { ...authority, permissionCeiling: [...authority.permissionCeiling] },
};
await this.persist();
});
await this.releaseObservation(record);
await this.releaseUnusedRuntimes();
}
private async find(parent: AgentAddress, child: AgentAddress): Promise<RelationshipRecord> {
await this.ready;
const record = [...this.records.values()].find(
({ row }) => sameAddress(row.parent, parent) && sameAddress(row.child, child),
);
if (!record)
throw failure("relationship_not_found", "This child is not attached to the selected parent");
return record;
}
/** Called by the host lifecycle, never by a mounted view. */
recover(): Promise<void> {
this.recoveryRequested = true;
if (this.recovering) return this.recovering;
this.recovering = (async () => {
while (this.recoveryRequested && !this.closed) {
this.recoveryRequested = false;
await this.recoverAll();
}
})().finally(() => {
this.recovering = null;
});
return this.recovering;
}
private async recoverAll(): Promise<void> {
await this.ready;
if (this.closed) return;
await Promise.allSettled(
[...this.records.values()].map(async (record) => {
if (terminalPhase(record.row.phase)) return;
const runtime = await this.runtime(record.authority);
const parent = await runtime.agent(record.row.parent);
const refreshed = await parent.refresh();
if (refreshed?.agent.archivedAt) {
await this.parentArchived(
record.row.parent,
refreshed.agent.workspaceId
? { serverId: record.row.parent.serverId, workspaceId: refreshed.agent.workspaceId }
: undefined,
);
return;
}
if (record.intent.kind !== "attached") {
await this.settleObligation(record);
return;
}
if (record.confirmed === null) await this.createChild(record);
await this.observeChild(record);
}),
);
}
async parentArchived(
parent: AgentAddress,
workspace: WorkspaceAddress | undefined,
): Promise<void> {
await this.transact(async () => {
for (const record of this.records.values()) {
if (
!sameAddress(record.row.parent, parent) ||
record.intent.kind !== "attached" ||
terminalConfirmation(record)
)
continue;
record.intent = {
revision: record.intent.revision + 1,
kind: "archive-parent",
parentWorkspace: workspace,
authority: record.authority,
};
}
await this.persist();
});
await Promise.allSettled(
[...this.records.values()]
.filter(
(record) =>
sameAddress(record.row.parent, parent) &&
record.intent.kind !== "attached" &&
record.intent.kind !== "forgotten",
)
.map((record) => this.settleObligation(record)),
);
}
private observeChild(record: RelationshipRecord): Promise<void> {
let starting = this.startingObservers.get(record.row.id);
if (!starting) {
starting = this.startObservation(record);
this.startingObservers.set(record.row.id, starting);
void starting.finally(() => this.startingObservers.delete(record.row.id)).catch(() => {});
}
return starting;
}
private async startObservation(record: RelationshipRecord): Promise<void> {
const intent = record.intent;
if (this.closed || !observesChild(record) || this.observers.has(record.row.id)) return;
const runtime = await this.runtime(record.authority);
const child = await runtime.agent(record.row.child);
if (
this.closed ||
record.intent !== intent ||
!observesChild(record) ||
this.observers.has(record.row.id)
)
return;
const observation = child.observe();
let applying: Promise<void> = Promise.resolve();
const accept = (snapshot: AgentSnapshotPayload): void => {
applying = applying
.then(async () => {
await this.transact(async () => {
if (record.intent !== intent || !observesChild(record)) return;
this.captureSnapshot(record, snapshot);
await this.persist();
});
if (!observesChild(record)) {
this.observers.delete(record.row.id);
await observation.release();
await this.releaseUnusedRuntimes();
}
return undefined;
})
.catch(async (error) => {
if (record.intent === intent) await this.markUnavailable(record, error);
});
};
this.observers.set(record.row.id, {
intent,
release: async () => {
await observation.release();
await applying;
},
});
observation.subscribe({
snapshot: accept,
update: (message) => {
if (
message.type === "agent_update" &&
message.payload.kind === "upsert" &&
message.payload.agent.id === record.row.child.agentId
)
accept(message.payload.agent);
},
error: (error) => {
void this.markUnavailable(record, error);
},
});
try {
await observation.ready;
await applying;
} catch (error) {
this.observers.delete(record.row.id);
await observation.release();
if (record.intent === intent) await this.markUnavailable(record, error);
throw error;
}
}
private captureSnapshot(record: RelationshipRecord, snapshot: AgentSnapshotPayload | null): void {
if (!snapshot) return;
record.row.snapshot = snapshot;
record.row.observedAt = new Date().toISOString();
record.row.freshness = "current";
record.row.error = null;
if (snapshot.workspaceId)
record.row.workspace = {
serverId: record.row.child.serverId,
workspaceId: snapshot.workspaceId,
};
if (snapshot.archivedAt) record.confirmed = "archived";
else if (snapshot.parent === null) record.confirmed = "detached";
else record.confirmed = "attached";
}
private async markUnavailable(record: RelationshipRecord, error: unknown): Promise<void> {
if (this.closed) return;
await this.transact(async () => {
record.row.freshness = "unavailable";
record.row.error = {
code: error instanceof Error && "code" in error ? String(error.code) : "host_error",
message: error instanceof Error ? error.message : String(error),
};
await this.persist();
});
}
private settleObligation(record: RelationshipRecord): Promise<void> {
if (
record.intent.kind === "attached" ||
record.intent.kind === "forgotten" ||
terminalConfirmation(record)
)
return Promise.resolve();
let settling = this.settling.get(record.row.id);
if (!settling) {
settling = Promise.resolve().then(() => this.settleCurrentIntent(record));
this.settling.set(record.row.id, settling);
void settling.finally(() => this.settling.delete(record.row.id)).catch(() => {});
}
return settling;
}
private async settleCurrentIntent(record: RelationshipRecord): Promise<void> {
while (
!this.closed &&
record.intent.kind !== "attached" &&
record.intent.kind !== "forgotten"
) {
const intent = record.intent;
await this.performSettlement(record, intent);
if (record.intent === intent) break;
}
}
private async observeCreation(
record: RelationshipRecord,
intent: RelationshipIntent,
runtime: HostRuntime,
): Promise<CreationSnapshot | null> {
const existing = this.observers.get(record.row.id);
if (existing?.creation && existing.intent === intent) return existing.creation();
await existing?.release();
this.observers.delete(record.row.id);
const api = await runtime.client(record.row.child.serverId);
if (this.closed || record.intent !== intent) {
await api.dispose();
return null;
}
const observation = api.creations.observeAgent(`subagent:${record.row.id}`);
let latest: CreationSnapshot | null = null;
const accept = (snapshot: CreationSnapshot | null) => {
latest = snapshot;
if (record.intent === intent && (snapshot?.agent || snapshot?.phase === "failed"))
void this.recover().catch(() => {});
};
observation.subscribe({
snapshot: (snapshot) => accept(snapshot.snapshot),
update: (message) => {
if (message.type === "agent.create.update") accept(message.payload);
},
error: (error) => {
if (record.intent === intent) void this.markUnavailable(record, error);
},
});
const owner = { intent, creation: () => latest, release: () => api.dispose() };
this.observers.set(record.row.id, owner);
try {
const initial = await observation.ready;
if (initial.error) throw new Error(initial.error);
if (this.closed || record.intent !== intent) {
if (this.observers.get(record.row.id) === owner) this.observers.delete(record.row.id);
await owner.release();
}
return latest;
} catch (error) {
if (this.observers.get(record.row.id) === owner) this.observers.delete(record.row.id);
await owner.release();
throw error;
}
}
private async releaseObservation(record: RelationshipRecord): Promise<void> {
const observation = this.observers.get(record.row.id);
this.observers.delete(record.row.id);
await observation?.release();
}
private async performSettlement(
record: RelationshipRecord,
intent: RelationshipIntent,
): Promise<void> {
try {
await this.authorize(intent.authority, "workspace.write");
const runtime = await this.runtime(intent.authority);
const child = await runtime.agent(record.row.child);
let current = await child.refresh();
if (!current) {
const destination = await runtime.client(record.row.child.serverId);
try {
if (record.intent !== intent) return;
const cancellation = await destination.creations.cancelAgent({
idempotencyKey: `subagent:${record.row.id}`,
agentId: record.row.child.agentId,
});
if (cancellation.status === "canceled") {
await this.transact(async () => {
if (record.intent !== intent) return;
record.confirmed = "canceled";
record.row.error = null;
record.row.freshness = "current";
record.row.observedAt = new Date().toISOString();
await this.persist();
});
await this.releaseObservation(record);
await this.releaseUnusedRuntimes();
return;
}
// Admission reserves an identity before the resource is ready. Inspect
// its existing receipt; never resubmit an uncertain initial prompt.
const receipt =
(await this.observeCreation(record, intent, runtime)) ?? cancellation.creation;
current = await child.refresh();
if (!current) {
await this.transact(async () => {
if (record.intent !== intent) return;
record.row.freshness = "current";
record.row.error =
receipt.phase === "failed"
? {
code: receipt.outcomeUnknown ? "creation_outcome_unknown" : "creation_failed",
message: receipt.error ?? "Creation failed before its child became available",
}
: {
code: "creation_pending",
message:
"Creation was admitted but its child is not yet available; reconciliation remains pending",
};
await this.persist();
});
return;
}
} finally {
await destination.dispose();
}
}
const action = await this.transact(async () => {
// Completion ownership is tested inside the same serialization boundary as the write.
if (record.intent !== intent) return null;
this.captureSnapshot(record, current.agent);
if (terminalConfirmation(record)) {
await this.persist();
return null;
}
const detach =
intent.kind === "detach" ||
shouldDetachFromArchivedParent({
parentWorkspace: intent.parentWorkspace,
childWorkspace: record.row.workspace,
hasOpenTab: hasOpenAgentTab(current.agent.labels),
});
await this.persist();
return detach ? "detach" : "archive";
});
if (!action || record.intent !== intent) return;
// Forget can suppress future dispatch; an operation already submitted to the child cannot be undone.
if (action === "detach") await child.detach();
else await child.archive();
await this.transact(async () => {
if (record.intent !== intent) return;
record.confirmed = action === "detach" ? "detached" : "archived";
record.row.error = null;
record.row.freshness = "current";
record.row.observedAt = new Date().toISOString();
await this.persist();
});
await this.releaseObservation(record);
await this.releaseUnusedRuntimes();
} catch (error) {
if (record.intent === intent) await this.markUnavailable(record, error);
}
}
async close(): Promise<void> {
this.closed = true;
// Owned transports stop before subscription cleanup; a stopping daemon no
// longer forwards new requests, including graceful release acknowledgments.
await Promise.allSettled(
[...this.runtimes.values()].map(async (runtime) => (await runtime).dispose()),
);
this.runtimes.clear();
await Promise.allSettled(
[...this.observers.values()].map((observation) => observation.release()),
);
this.observers.clear();
this.listeners.clear();
await Promise.allSettled(this.running.values());
await this.mutation;
}
}
function sameAddress(a: AgentAddress, b: AgentAddress): boolean {
return a.serverId === b.serverId && a.agentId === b.agentId;
}
function failure(code: string, message: string): Error {
return Object.assign(new Error(message), { code });
}
function observesChild(record: RelationshipRecord): boolean {
return record.intent.kind === "attached" && !terminalConfirmation(record);
}
function relationshipPhase(record: RelationshipRecord): RelationshipRecord["row"]["phase"] {
if (record.intent.kind === "forgotten") return "forgotten";
if (record.confirmed !== null && record.confirmed !== "attached") return record.confirmed;
if (record.intent.kind === "detach") return "detach_pending";
if (record.intent.kind === "archive-parent")
return shouldDetachFromArchivedParent({
parentWorkspace: record.intent.parentWorkspace,
childWorkspace: record.row.workspace,
hasOpenTab: hasOpenAgentTab(record.row.snapshot?.labels),
})
? "detach_pending"
: "lifecycle_pending";
return record.confirmed === null ? "creating" : "attached";
}
function terminalConfirmation(record: RelationshipRecord): boolean {
return record.confirmed !== null && record.confirmed !== "attached";
}
function terminalPhase(phase: RelationshipRecord["row"]["phase"]): boolean {
return (
phase === "detached" || phase === "archived" || phase === "forgotten" || phase === "canceled"
);
}
packages/server/src/server/websocket-server.ts
+70 -1
View File
@@ -1,3 +1,4 @@
import { Subagents } from "./subagents/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";
@@ -465,6 +466,7 @@ interface OwnedSessionConnection extends SessionConnectionBase {
type SessionConnection = ReconnectableSessionConnection | OwnedSessionConnection;
interface SocketSessionOptions {
subagentAuthority: NonNullable<SessionOptions["subagentAuthority"]>;
sourceId?: string;
clientId: string;
appVersion: string | null;
@@ -531,7 +533,10 @@ function requireWebSocketServices(params: {
* WebSocket server that only accepts sockets + parses/forwards messages to the session layer.
*/
export class VoiceAssistantWebSocketServer {
private readonly subagents: Subagents;
private ownerAdmission: SessionAdmission = OWNER_SESSION_ADMISSION;
private readonly hostRuntimes = new Set<Awaited<ReturnType<typeof createDaemonHostRuntime>>>();
private readonly ownedSocketSources = new WeakMap<WebSocketLike, string>();
private readonly ownedSockets = new WeakSet<WebSocketLike>();
public async createHostRuntime(authority: HostRuntimeAuthority) {
@@ -551,6 +556,7 @@ export class VoiceAssistantWebSocketServer {
},
);
}
if (authority.sourceId) this.ownedSocketSources.set(socket, authority.sourceId);
this.ownedSockets.add(socket);
await this.attachExternalSocket(socket, undefined, admission);
},
@@ -740,7 +746,32 @@ export class VoiceAssistantWebSocketServer {
this.workspaceAutoName = workspaceAutoName;
this.downloadTokenStore = downloadTokenStore;
this.paseoHome = paseoHome;
this.connectedHosts = new ConnectedHosts(paseoHome, { serverId, label: getHostName() });
this.connectedHosts = new ConnectedHosts(
paseoHome,
{ serverId, label: getHostName() },
(destination) => this.subagents.assertHostRemovable(destination),
);
this.subagents = new Subagents({
home: paseoHome,
serverId,
runtime: (authority) => this.createHostRuntime(authority),
resolveAuthority: (principalId) => this.resolveRelationshipAdmission(principalId),
});
this.connectedHosts.observe(() => {
void this.connectedHosts
.list()
.then((directory) => this.subagents.hostsChanged(directory))
.catch((error) =>
this.logger.warn({ err: error }, "Failed to reconcile subagent host availability"),
);
});
this.agentManager.setAgentArchivedCallback(async (agentId) => {
const record = await this.agentStorage.get(agentId);
await this.subagents.parentArchived(
{ serverId, agentId },
record?.workspaceId ? { serverId, workspaceId: record.workspaceId } : undefined,
);
});
this.passwordHash = auth?.password;
this.worktreesRoot = daemonRuntimeConfig?.worktreesRoot;
this.daemonConfigStore = daemonConfigStore;
@@ -1084,10 +1115,19 @@ export class VoiceAssistantWebSocketServer {
return { closed };
}
private async resolveRelationshipAdmission(principalId: string): Promise<SessionAdmission> {
if (principalId === "owner") return this.ownerAdmission;
throw Object.assign(
new Error("This principal has no supported durable relationship authority"),
{ code: "unsupported_capability" },
);
}
public updatePrincipalPermissions(
principalId: string,
permissions: readonly DaemonPermission[],
): void {
if (principalId === "owner") this.ownerAdmission = { principalId, permissions };
for (const [socket, link] of this.hostLinks) {
if (this.sessions.get(socket)?.principalId === principalId) {
link.close("permission_denied");
@@ -1121,10 +1161,16 @@ export class VoiceAssistantWebSocketServer {
if (this.connectionLifecycle === "starting") {
this.connectionLifecycle = "accepting";
}
void this.subagents
.recover()
.catch((error) =>
this.logger.warn({ err: error }, "Failed to recover subagent relationships"),
);
}
public async close(): Promise<void> {
this.prepareForShutdown();
await this.subagents.close();
await Promise.allSettled([...this.hostRuntimes].map((runtime) => runtime.dispose()));
this.hostRuntimes.clear();
await this.connectedHosts.close();
@@ -1536,6 +1582,19 @@ export class VoiceAssistantWebSocketServer {
onLifecycleIntent: (intent) => {
this.onLifecycleIntent?.(intent);
},
subagentAuthority: () => {
if (!connection) throw new Error("Session admission is not ready");
if (connection.hubExecutionAgents)
throw Object.assign(
new Error("Execution-scoped subagent relationships are not supported"),
{ code: "unsupported_capability" },
);
return {
principalId: connection.principalId,
sourceId: connection.session.getHostSourceId(),
permissionCeiling: connection.session.getPermissions(),
};
},
hubExecutionAgents: admission.hubExecutionAgents,
hubRelationships: this.hubRelationships ?? undefined,
});
@@ -1561,6 +1620,8 @@ export class VoiceAssistantWebSocketServer {
private createSocketSession(options: SocketSessionOptions): Session {
return new Session({
subagents: this.subagents,
subagentAuthority: options.subagentAuthority,
connectedHosts: this.connectedHosts,
getConnectedHostBufferedAmount: () => {
let total = 0;
@@ -1935,6 +1996,13 @@ export class VoiceAssistantWebSocketServer {
pluginId?: string,
): void {
if (pluginId) return;
for (const socket of connection.sockets) {
const ownedSource = this.ownedSocketSources.get(socket);
if (ownedSource) {
connection.session.setHostSource(ownedSource);
return;
}
}
const producer = hello.producer;
const sourceId =
hello.routed && producer
@@ -1957,6 +2025,7 @@ export class VoiceAssistantWebSocketServer {
...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
features: {
connectedHosts: true,
fleetSubagents: true,
hostMultiplexing: true,
usageSources: true,
ownedSubscriptions: true,