chore(chat): merge main while preserving stacked find features

This commit is contained in:
Mohamed Boudra
2026-09-15 16:19:18 +02:00
794 changed files with 36260 additions and 17173 deletions
@@ -10,13 +10,12 @@ export async function subscribeToEvents(
url: string,
agentId: string,
workspaceId: string,
): Promise<() => void> {
): Promise<() => Promise<void>> {
const client = createClient(url);
await client.connect();
await client.workspaces.list({
subscribe: { subscriptionId: `workspace-${workspaceId}` },
});
const agents = await client.agents.list({ subscribe: {} });
const workspaces = await client.workspaces.list({ subscribe: {} });
const unsubscribeAgentUpdates = client.agents.subscribe((update) => {
if (update.kind === "upsert" && update.agent.id === agentId) {
@@ -31,14 +30,24 @@ export async function subscribeToEvents(
});
const unsubscribeTimeline = client.agents.ref(agentId).timeline.subscribe((event) => {
void event.event;
if (event.event.type === "subscription_restored") {
console.log("Live subscription restored; request any missed history explicitly.");
} else if (event.event.type === "error") {
console.error(event.event.error);
}
});
return () => {
await unsubscribeTimeline.ready;
return async () => {
unsubscribeTimeline();
unsubscribeWorkspaceUpdates();
unsubscribeAgentUpdates();
void client.close();
try {
await Promise.all([agents.subscription.release(), workspaces.subscription.release()]);
} finally {
await client.close();
}
};
}
+3 -3
View File
@@ -1,6 +1,6 @@
{
"name": "@getpaseo/client",
"version": "0.8.0-beta.1",
"version": "0.8.0",
"description": "Paseo client SDK package",
"files": [
"dist",
@@ -39,8 +39,8 @@
"test": "vitest run"
},
"dependencies": {
"@getpaseo/protocol": "0.8.0-beta.1",
"@getpaseo/relay": "0.8.0-beta.1",
"@getpaseo/protocol": "0.8.0",
"@getpaseo/relay": "0.8.0",
"zod": "^4.4.3"
},
"devDependencies": {
File diff suppressed because it is too large Load Diff
+130 -98
View File
@@ -1,15 +1,138 @@
import type { SessionEventSubscription } from "@getpaseo/protocol/messages";
export { OwnedSubscriptions, type OwnedSubscription, type SubscriptionObserver } from "./owned.js";
import {
OwnedSubscriptions,
type OwnedSubscription,
type OwnedSubscriptionsHost,
} from "./owned.js";
import { LegacySubscriptions, type ObservationRequest } from "./legacy.js";
import {
type SessionInboundMessage,
type SessionOutboundMessage,
type ServerInfoStatusPayload,
} from "@getpaseo/protocol/messages";
import { CLIENT_CAPS, type ClientCapability } from "@getpaseo/protocol/client-capabilities";
export class ConnectionSubscriptions extends OwnedSubscriptions {
private legacy: LegacySubscriptions | null = null;
constructor(
private readonly connection: OwnedSubscriptionsHost & {
send(message: ObservationRequest): Promise<void>;
},
) {
super({
...connection,
release: async (id) => {
if (!this.legacy) return connection.release(id);
const message = this.legacy.release(id);
if (message) await connection.send(message);
},
});
}
override restore(
info?: ServerInfoStatusPayload,
capabilities?: Partial<Record<ClientCapability, unknown>>,
): void {
if (info) {
const clientCapabilities = capabilities ?? DEFAULT_CLIENT_CAPABILITIES;
const owned =
info.features?.ownedSubscriptions === true &&
clientCapabilities[CLIENT_CAPS.ownedSubscriptions] === true;
this.legacy = null;
if (!owned) {
this.legacy = new LegacySubscriptions(
info,
this.connection.send,
capabilities?.[CLIENT_CAPS.browserHost],
);
}
}
super.restore();
}
override disconnected(): void {
super.disconnected();
this.legacy = null;
}
observeRequest<T>(
query: ObservationRequest,
request: (query: ObservationRequest, accept: (snapshot: T) => void) => Promise<T>,
signal?: AbortSignal,
failed?: () => void,
): OwnedSubscription<T> {
return super.observe(
async (accept) => {
const legacy = this.legacy;
if (!legacy) return request(query, accept);
const interest = legacy.start(query);
const snapshot = (payload: T): T & { subscriptionId: string } => ({
...payload,
subscriptionId: interest.id,
});
try {
const message = legacy.request(interest);
if (!message) {
// Broadcast-era hosts have no acknowledgement. Readiness means the
// local listener is attached; no server membership is being claimed.
const value = snapshot({
requestId: interest.id,
...(query.type === "agent.timeline.set_subscription.request"
? { agentIds: query.agentIds }
: {}),
} as T);
accept(value);
return value;
}
return snapshot(await request(message, (payload) => accept(snapshot(payload))));
} catch (error) {
const release = legacy.release(interest.id);
if (release && this.legacy === legacy)
await this.connection.send(release).catch(this.connection.failed);
throw error;
}
},
signal,
failed,
);
}
override owns(message: SessionOutboundMessage): boolean {
return this.legacy ? this.legacy.owns(message) : super.owns(message);
}
override receive(message: SessionOutboundMessage): void {
if (this.legacy) this.legacy.receive(message, (update) => super.receive(update));
else super.receive(message);
}
normalize(message: SessionOutboundMessage): SessionOutboundMessage {
return this.legacy?.normalize(message) ?? message;
}
prepareRequest(message: SessionInboundMessage) {
return (
this.legacy?.prepareRequest(message) ?? {
message,
receive: (value: SessionOutboundMessage) => value,
finish: async () => {},
}
);
}
}
// Protocol support belongs to the installed client. Only browser hosting needs
// a resource supplied by the caller. Keep this exhaustive as the protocol evolves.
export const DEFAULT_CLIENT_CAPABILITIES = {
[CLIENT_CAPS.ownedSubscriptions]: true,
[CLIENT_CAPS.allProviders]: true,
[CLIENT_CAPS.selectiveAgentTimeline]: true,
[CLIENT_CAPS.reasoningMergeEnum]: true,
[CLIENT_CAPS.customModeIcons]: true,
[CLIENT_CAPS.terminalReflowableSnapshot]: true,
[CLIENT_CAPS.providerSubagents]: true,
[CLIENT_CAPS.projectedSubagentTimeline]: true,
[CLIENT_CAPS.projectUpdates]: true,
[CLIENT_CAPS.compactProviderSnapshots]: true,
[CLIENT_CAPS.providerSnapshotReferences]: true,
@@ -20,100 +143,9 @@ export const DEFAULT_CLIENT_CAPABILITIES = {
[CLIENT_CAPS.explicitEventSubscriptions]: true,
} satisfies Record<Exclude<ClientCapability, typeof CLIENT_CAPS.browserHost>, true>;
/** Calling releases demand; ready acknowledges the initial daemon membership. */
export type TimelineSubscription = (() => void) & { readonly ready: Promise<void> };
class TimelineInterest {
resolve!: () => void;
reject!: (error: unknown) => void;
readonly ready = new Promise<void>((resolve, reject) => {
this.resolve = resolve;
this.reject = reject;
});
constructor() {
// Readiness is optional for fire-and-forget listeners.
void this.ready.catch(() => {});
}
}
/** Owns connection demand, independently of individual facades and React lifetimes. */
export class ConnectionSubscriptions {
private viewed = new Set<string>();
private timelines = new Map<string, Set<TimelineInterest>>();
private events: SessionEventSubscription[] = [];
constructor(
private readonly send: {
timelines(agentIds: string[]): Promise<void> | null;
events(events: SessionEventSubscription[]): Promise<void>;
failed(error: unknown): void;
},
) {}
private agentIds(): string[] {
return [...new Set([...this.viewed, ...this.timelines.keys()])].sort();
}
setViewed(agentIds: string[]): Promise<void> {
this.viewed = new Set(agentIds);
return this.syncTimelines();
}
observeTimeline(agentId: string): TimelineSubscription {
const interests = this.timelines.get(agentId) ?? new Set<TimelineInterest>();
const interest = new TimelineInterest();
interests.add(interest);
this.timelines.set(agentId, interests);
// Each caller gets acknowledgement for its own registration, even when an
// existing listener first subscribed on an earlier connection.
void this.syncTimelines().catch(this.send.failed);
let active = true;
return Object.assign(
() => {
if (!active) return;
active = false;
interest.reject(new Error("Timeline subscription released before it was ready"));
interests.delete(interest);
if (interests.size === 0) {
this.timelines.delete(agentId);
void this.syncTimelines().catch(this.send.failed);
}
},
{ ready: interest.ready },
);
}
private async syncTimelines(): Promise<void> {
const interests: TimelineInterest[] = [];
for (const listeners of this.timelines.values()) {
for (const interest of listeners) interests.push(interest);
}
try {
const sent = this.send.timelines(this.agentIds());
if (!sent) return;
await sent;
for (const interest of interests) interest.resolve();
} catch (error) {
for (const interest of interests) interest.reject(error);
throw error;
}
}
setEvents(events: SessionEventSubscription[]): void {
if (JSON.stringify(this.events) === JSON.stringify(events)) return;
this.events = events;
void this.send.events(events).catch(this.send.failed);
}
restore(): void {
if (this.agentIds().length) void this.syncTimelines().catch(this.send.failed);
if (this.events.length) void this.send.events(this.events).catch(this.send.failed);
}
close(): void {
for (const interests of this.timelines.values()) {
for (const interest of interests) interest.reject(new Error("Daemon client closed"));
}
}
}
/** Calling releases demand; ready waits for membership, or local attachment on broadcast hosts. */
export type TimelineSubscription = (() => void) & {
readonly ready: Promise<void>;
readonly subscriptionId: string | null;
release(): Promise<void>;
};
@@ -0,0 +1,176 @@
import { AgentSnapshotPayloadSchema } from "@getpaseo/protocol/messages";
import type { AgentSnapshotPayload, SessionOutboundMessage } from "@getpaseo/protocol/messages";
import {
deriveAgentStateBucket,
getWorkspaceStateBucketPriority,
} from "@getpaseo/protocol/agent-state-bucket";
type AgentEntry = Extract<
SessionOutboundMessage,
{ type: "fetch_agents_response" }
>["payload"]["entries"][number];
type Workspace = Extract<
SessionOutboundMessage,
{ type: "fetch_workspaces_response" }
>["payload"]["entries"][number];
type WorkspaceUpdate = Extract<SessionOutboundMessage, { type: "workspace_update" }>;
// Preserve the pre-registry app's identity format. This is an opaque ID, never
// a filesystem path: converting C:\ to C: changes a drive root into a relative path.
function legacyWorkspaceId(value: string): string {
return value.trim().replace(/\\/g, "/").replace(/\/+$/, "") || "/";
}
// COMPAT(legacyWorkspaceDaemon): restored in v0.8.0; remove after 2027-03-11 once daemon floor >= v0.1.97.
// The pre-registry app grouped agents by checkout path. Keep that representation
// at the client edge, including live updates, so app workflows need no version branch.
export class LegacyWorkspaces {
private readonly agents = new Map<string, AgentEntry>();
normalize(message: SessionOutboundMessage): SessionOutboundMessage {
const stamp = (agent: AgentSnapshotPayload): AgentSnapshotPayload => ({
...agent,
workspaceId: agent.workspaceId ?? legacyWorkspaceId(agent.cwd),
});
if (message.type === "fetch_agents_response")
return {
...message,
payload: {
...message.payload,
entries: message.payload.entries.map((entry) => ({
...entry,
agent: stamp(entry.agent),
})),
},
};
if (message.type === "agent_update" && message.payload.kind === "upsert")
return { ...message, payload: { ...message.payload, agent: stamp(message.payload.agent) } };
if (
(message.type === "fetch_agent_response" ||
message.type === "fetch_agent_timeline_response" ||
message.type === "cancel_agent_response") &&
message.payload.agent
) {
return {
...message,
payload: { ...message.payload, agent: stamp(message.payload.agent) },
} as SessionOutboundMessage;
}
if (
message.type === "status" &&
(message.payload.status === "agent_created" || message.payload.status === "agent_resumed")
) {
return {
...message,
payload: {
...message.payload,
agent: stamp(AgentSnapshotPayloadSchema.parse(message.payload.agent)),
},
};
}
return message;
}
read(entries: AgentEntry[], reset: boolean): Workspace[] {
if (reset) this.agents.clear();
for (const entry of entries) this.agents.set(entry.agent.id, entry);
const pageIds = new Set(entries.map(workspaceId));
return [...this.workspaces().values()].filter((workspace) => pageIds.has(workspace.id));
}
update(message: SessionOutboundMessage): WorkspaceUpdate[] {
if (message.type !== "agent_update") return [];
const before = this.workspaces();
const update = message.payload;
if (update.kind === "remove") this.agents.delete(update.agentId);
else {
const project = update.project ?? this.agents.get(update.agent.id)?.project;
if (update.agent.archivedAt) this.agents.delete(update.agent.id);
else if (project) this.agents.set(update.agent.id, { agent: update.agent, project });
}
const after = this.workspaces();
const changes: WorkspaceUpdate[] = [];
for (const [id, workspace] of after) {
if (JSON.stringify(before.get(id)) !== JSON.stringify(workspace))
changes.push({ type: "workspace_update", payload: { kind: "upsert", workspace } });
}
for (const id of before.keys())
if (!after.has(id))
changes.push({ type: "workspace_update", payload: { kind: "remove", id } });
return changes;
}
private workspaces(): Map<string, Workspace> {
const workspaces = new Map<string, Workspace>();
for (const entry of this.agents.values()) {
const { agent, project } = entry;
const { checkout } = project;
const id = workspaceId(entry);
const status = deriveAgentStateBucket({
status: agent.status,
pendingPermissionCount: agent.pendingPermissions.length,
requiresAttention: agent.requiresAttention,
attentionReason: agent.attentionReason,
});
const existing = workspaces.get(id);
if (
existing &&
getWorkspaceStateBucketPriority(existing.status) <= getWorkspaceStateBucketPriority(status)
)
continue;
workspaces.set(id, {
id,
projectId: project.projectKey,
projectDisplayName: project.projectName,
projectCustomName: null,
projectRootPath: checkout.mainRepoRoot ?? checkout.worktreeRoot ?? checkout.cwd,
workspaceDirectory: checkout.cwd,
projectKind: checkout.isGit ? "git" : "non_git",
workspaceKind: workspaceKind(checkout),
name: workspaceName(entry, id),
title: null,
status,
statusEnteredAt: agent.attentionTimestamp ?? agent.updatedAt,
activityAt: agent.updatedAt,
archivingAt: null,
diffStat: null,
scripts: [],
gitRuntime: gitRuntime(checkout),
githubRuntime: null,
project,
});
}
return workspaces;
}
}
function workspaceId(entry: AgentEntry): string {
return entry.agent.workspaceId ?? legacyWorkspaceId(entry.project.checkout.cwd);
}
function workspaceKind(checkout: AgentEntry["project"]["checkout"]): Workspace["workspaceKind"] {
if (!checkout.isGit) return "directory";
if (checkout.isPaseoOwnedWorktree) return "worktree";
return "checkout";
}
function workspaceName(entry: AgentEntry, id: string): string {
const name = entry.project.workspaceName?.trim();
if (name) return name;
const branch = entry.project.checkout.currentBranch?.trim();
if (branch && branch !== "HEAD") return branch;
return id.slice(id.lastIndexOf("/") + 1);
}
function gitRuntime(checkout: AgentEntry["project"]["checkout"]): Workspace["gitRuntime"] {
if (!checkout.isGit) return null;
return {
currentBranch: checkout.currentBranch,
remoteUrl: checkout.remoteUrl,
isPaseoOwnedWorktree: checkout.isPaseoOwnedWorktree,
isDirty: null,
aheadBehind: null,
aheadOfOrigin: null,
behindOfOrigin: null,
};
}
+282
View File
@@ -0,0 +1,282 @@
import { SessionInboundMessageSchema } from "@getpaseo/protocol/messages";
import { LegacyWorkspaces } from "./legacy-workspaces.js";
import { BrowserAutomationHostCapabilitySchema } from "@getpaseo/protocol/browser-automation/capabilities";
import type {
ServerInfoStatusPayload,
SessionInboundMessage,
SessionOutboundMessage,
} from "@getpaseo/protocol/messages";
export type ObservationRequest = { type: SessionInboundMessage["type"] } & Record<string, unknown>;
interface Interest {
id: string;
query: ObservationRequest;
}
// COMPAT(ownedSubscriptions): added in v0.8.0; remove after 2027-03-11 once daemon floor >= v0.8.0.
// Legacy directories retain their connection-wide, last-query-wins semantics.
// IDs below identify local listeners, not independent server subscriptions.
export class LegacySubscriptions {
private readonly interests = new Map<string, Interest>();
private readonly workspaces: LegacyWorkspaces | null;
constructor(
private readonly info: ServerInfoStatusPayload,
private readonly send: (message: ObservationRequest) => Promise<void>,
private readonly browserHost: unknown,
) {
this.workspaces = info.features?.workspaceMultiplicity === true ? null : new LegacyWorkspaces();
}
normalize(message: SessionOutboundMessage): SessionOutboundMessage {
return this.workspaces?.normalize(message) ?? message;
}
prepareRequest(message: SessionInboundMessage): {
message: SessionInboundMessage;
receive(message: SessionOutboundMessage): SessionOutboundMessage;
finish(): Promise<void>;
} {
const identity = {
message,
receive: (value: SessionOutboundMessage) => value,
finish: async () => {},
};
const workspaces = this.workspaces;
if (workspaces && message.type === "fetch_workspaces_request") {
return {
...identity,
message: SessionInboundMessageSchema.parse({
type: "fetch_agents_request",
requestId: message.requestId,
scope: "active",
sort: [{ key: "updated_at", direction: "desc" }],
page: message.page,
subscribe: message.subscribe,
}),
receive: (value) => {
if (value.type !== "fetch_agents_response") return value;
if (value.payload.requestId !== message.requestId) return value;
return {
type: "fetch_workspaces_response",
payload: {
...value.payload,
entries: workspaces.read(value.payload.entries, !message.page?.cursor),
emptyProjects: [],
},
};
},
};
}
if (message.type === "subscribe_terminals_request") {
return {
...identity,
receive: (value) => {
if (value.type !== "terminals_changed" || value.payload.cwd !== message.cwd) return value;
return { ...value, payload: { ...value.payload, requestId: message.requestId } };
},
};
}
if (message.type === "workspace.label.list.request" && !message.subscribe) {
return {
...identity,
message: { ...message, subscribe: { subscriptionId: `legacy:${crypto.randomUUID()}` } },
};
}
if (message.type === "checkout.diff.get.request") {
const subscriptionId = `legacy:${crypto.randomUUID()}`;
return {
message: SessionInboundMessageSchema.parse({
...message,
type: "subscribe_checkout_diff_request",
subscriptionId,
}),
receive: (value) => {
if (value.type !== "subscribe_checkout_diff_response") return value;
if (value.payload.requestId !== message.requestId) return value;
return { ...value, type: "checkout.diff.get.response" };
},
finish: () => this.send({ type: "unsubscribe_checkout_diff_request", subscriptionId }),
};
}
return identity;
}
start(query: ObservationRequest): Interest {
const interest = { id: `legacy:${crypto.randomUUID()}`, query };
this.interests.set(interest.id, interest);
return interest;
}
request({ id, query }: Interest): ObservationRequest | null {
switch (query.type) {
case "fetch_agents_request":
case "fetch_workspaces_request":
case "workspace.label.list.request":
return { ...query, subscribe: { subscriptionId: id } };
case "fs.file.subscribe.request":
case "subscribe_checkout_diff_request":
return { ...query, subscriptionId: id };
case "agent.timeline.set_subscription.request":
if (!this.info.features?.selectiveAgentTimeline) return null;
return { ...query, agentIds: this.members(query.type, "agentIds") };
case "session.events.set_subscription.request":
if (!this.info.features?.explicitEventSubscriptions) return null;
return {
...query,
events: this.members(query.type, "events").filter(isLegacyEvent),
};
case "browser.host.register.request":
if (!matchesBrowserRegistration(this.browserHost, query)) {
throw new Error(
"This daemon requires browser_host registration in the client's hello capabilities",
);
}
return null; // Older hosts register in hello.
default:
return query;
}
}
private members(type: ObservationRequest["type"], field: string): string[] {
const members = new Set<string>();
for (const interest of this.interests.values()) {
if (interest.query.type !== type) continue;
for (const value of interest.query[field] as string[]) members.add(value);
}
return [...members].sort();
}
release(id: string): ObservationRequest | null {
const interest = this.interests.get(id);
if (!interest) return null;
this.interests.delete(id);
const { query } = interest;
switch (query.type) {
case "agent.timeline.set_subscription.request":
case "session.events.set_subscription.request":
return this.request(interest);
case "fs.file.subscribe.request":
return { type: "fs.file.unsubscribe.request", subscriptionId: id };
case "subscribe_checkout_diff_request":
return { type: "unsubscribe_checkout_diff_request", subscriptionId: id };
case "subscribe_terminal_request":
if (
[...this.interests.values()].some(
(item) => item.query.type === query.type && item.query.terminalId === query.terminalId,
)
)
return null;
return { type: "unsubscribe_terminal_request", terminalId: query.terminalId };
case "subscribe_terminals_request":
if (
[...this.interests.values()].some(
(item) =>
item.query.type === query.type &&
item.query.cwd === query.cwd &&
item.query.workspaceId === query.workspaceId,
)
)
return null;
return {
type: "unsubscribe_terminals_request",
cwd: query.cwd,
workspaceId: query.workspaceId,
};
default:
// Directory slots have no release RPC. Detach the local listener.
return null;
}
}
owns(message: SessionOutboundMessage): boolean {
return [...this.interests.values()].some((interest) => this.matches(interest, message));
}
receive(
message: SessionOutboundMessage,
deliver: (message: SessionOutboundMessage) => void,
): void {
const updates: SessionOutboundMessage[] = [
message,
...(this.workspaces?.update(message) ?? []),
];
if (message.type === "agent_stream" && message.payload.event.type === "attention_required") {
const { agentId, event } = message.payload;
updates.push({
type: "agent_attention_required",
payload: {
agentId,
reason: event.reason,
timestamp: event.timestamp,
shouldNotify: event.shouldNotify,
...(event.notification ? { notification: event.notification } : {}),
},
});
}
for (const update of updates)
for (const interest of this.interests.values()) {
if (!this.matches(interest, update)) continue;
// Normalize at the connection edge; all consumers receive the same handle shape.
deliver(
("payload" in update
? { ...update, payload: { ...update.payload, subscriptionId: interest.id } }
: { ...update, subscriptionId: interest.id }) as SessionOutboundMessage,
);
}
}
private matches({ id, query }: Interest, message: SessionOutboundMessage): boolean {
switch (query.type) {
case "fetch_agents_request":
return message.type === "agent_update";
case "fetch_workspaces_request":
return message.type === "workspace_update";
case "workspace.label.list.request":
return message.type === "workspace.label.update";
case "fs.file.subscribe.request":
return message.type === "fs.file.update" && message.payload.subscriptionId === id;
case "subscribe_checkout_diff_request":
return message.type === "checkout_diff_update" && message.payload.subscriptionId === id;
case "subscribe_terminals_request":
return message.type === "terminals_changed" && message.payload.cwd === query.cwd;
case "subscribe_terminal_request":
return (
message.type === "terminal_stream_exit" && message.payload.terminalId === query.terminalId
);
case "agent.timeline.set_subscription.request":
return (
(message.type === "agent_stream" || message.type === "agent.timeline.replacement") &&
(query.agentIds as string[]).includes(message.payload.agentId)
);
case "session.events.set_subscription.request":
return (query.events as string[]).includes(
message.type === "status" ? `status.${message.payload.status}` : message.type,
);
case "browser.host.register.request":
return message.type === "browser.automation.execute.request";
default:
return false;
}
}
}
function isLegacyEvent(event: string): boolean {
return [
"project.update",
"providers_snapshot_update",
"agent_attention_required",
"agent_permission_request",
"agent_permission_resolved",
].includes(event);
}
function matchesBrowserRegistration(advertised: unknown, request: ObservationRequest): boolean {
const host = BrowserAutomationHostCapabilitySchema.safeParse(advertised);
if (!host.success) return false;
if (host.data.hostKind !== request.hostKind) return false;
for (const command of request.supportedCommands as string[]) {
if (!host.data.supportedCommands.some((supported) => supported === command)) return false;
}
return true;
}
@@ -0,0 +1,129 @@
import { expect, test } from "vitest";
import { OwnedSubscriptions } from "./owned.js";
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
interface Snapshot {
subscriptionId: string;
requestId: string;
}
function connection() {
const released: string[] = [];
const failures: unknown[] = [];
const requests: Array<(id: string) => void> = [];
const subscriptions = new OwnedSubscriptions({
release: async (id) => {
released.push(id);
},
failed: (error) => {
failures.push(error);
},
});
subscriptions.restore();
return {
subscriptions,
released,
failures,
requests,
observe: (signal?: AbortSignal) =>
subscriptions.observe<Snapshot>(
(accept) =>
new Promise((resolve) => {
requests.push((subscriptionId) => {
const snapshot = { subscriptionId, requestId: `request-${requests.length}` };
accept(snapshot);
resolve(snapshot);
});
}),
signal,
),
};
}
function update(subscriptionId: string, seq: number): SessionOutboundMessage {
return {
type: "agent_update",
payload: { kind: "remove", agentId: `agent-${seq}`, subscriptionId, seq },
};
}
test("a caller consuming the snapshot later receives its buffered updates in order", async () => {
const c = connection();
const a = c.observe();
const b = c.observe();
c.requests[0]!("a");
c.requests[1]!("b");
c.subscriptions.receive(update("a", 1));
c.subscriptions.receive(update("b", 2));
await a.ready;
const seen: string[] = [];
a.subscribe({
snapshot: (s) => seen.push(s.subscriptionId),
update: (m) => seen.push(JSON.stringify(m)),
});
expect(seen).toEqual(["a", JSON.stringify(update("a", 1))]);
await a.release();
c.subscriptions.receive(update("a", 3));
expect(seen).toHaveLength(2);
expect(c.released).toEqual(["a"]);
expect(b.subscriptionId).toBe("b");
await b.release();
});
test("cancellation before the ID waits for bootstrap and releases exactly that ID", async () => {
const c = connection();
const abort = new AbortController();
const handle = c.observe(abort.signal);
abort.abort();
const released = handle.release();
expect(c.released).toEqual([]);
c.requests[0]!("cancelled");
await released;
await expect(handle.ready).rejects.toThrow("released");
expect(c.released).toEqual(["cancelled"]);
});
test("surviving handles use new remote IDs after reconnect and released handles stay gone", async () => {
const c = connection();
const a = c.observe();
const b = c.observe();
c.requests[0]!("old-a");
c.requests[1]!("old-b");
await Promise.all([a.ready, b.ready]);
await b.release();
const seen: string[] = [];
a.subscribe({ snapshot: (s) => seen.push(s.subscriptionId), update: () => seen.push("update") });
c.subscriptions.disconnected();
c.subscriptions.restore();
expect(c.requests).toHaveLength(3);
c.requests[2]!("new-a");
c.subscriptions.receive(update("old-a", 1));
c.subscriptions.receive(update("new-a", 2));
expect(seen).toEqual(["old-a", "new-a", "update"]);
expect(a.subscriptionId).toBe("new-a");
await a.release();
});
test("a late listener after bounded replay overflow gets a fresh snapshot", async () => {
const c = connection();
const handle = c.observe();
c.requests[0]!("old");
await handle.ready;
const first: string[] = [];
handle.subscribe({ snapshot: (s) => first.push(s.subscriptionId), update: () => {} });
for (let seq = 0; seq < 129; seq++) c.subscriptions.receive(update("old", seq));
const late: string[] = [];
handle.subscribe({
snapshot: (s) => late.push(s.subscriptionId),
update: () => late.push("update"),
});
await expect.poll(() => c.requests.length).toBe(2);
expect(c.released).toEqual(["old"]);
expect(late).toEqual([]);
c.requests[1]!("fresh");
c.subscriptions.receive(update("old", 130));
c.subscriptions.receive(update("fresh", 131));
expect(first).toEqual(["old", "fresh"]);
expect(late).toEqual(["fresh", "update"]);
await handle.release();
});
+235
View File
@@ -0,0 +1,235 @@
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
export type SubscriptionSnapshot<T> = T & { subscriptionId: string };
export interface SubscriptionObserver<T> {
snapshot(snapshot: T): void;
update(message: SessionOutboundMessage): void;
error?(error: unknown): void;
}
/** A stable local lifetime. IDs change on reconnect; legacy IDs identify local listeners only. */
export interface OwnedSubscription<T> {
readonly subscriptionId: string | null;
readonly ready: Promise<SubscriptionSnapshot<T>>;
subscribe(observer: SubscriptionObserver<SubscriptionSnapshot<T>>): () => void;
release(): Promise<void>;
}
interface Registration {
start(): void;
disconnected(): void;
receive(message: SessionOutboundMessage): void;
release(): Promise<void>;
}
export interface OwnedSubscriptionsHost {
release(subscriptionId: string): Promise<void>;
failed(error: unknown): void;
}
const MAX_BOOTSTRAP_UPDATES = 128;
/** Transport routing and lifetime only; query semantics remain with each domain. */
export class OwnedSubscriptions {
private readonly registrations = new Set<Registration>();
private readonly routes = new Map<string, Registration>();
private generation = 0;
private connected = false;
constructor(private readonly host: OwnedSubscriptionsHost) {}
observe<T>(
request: (accept: (snapshot: T) => void) => Promise<T>,
signal?: AbortSignal,
requestFailed?: () => void,
): OwnedSubscription<T> {
let id: string | null = null;
let snapshot: SubscriptionSnapshot<T> | undefined;
let pending: Promise<void> | null = null;
let teardown: Promise<void> | null = null;
let disposed = false;
let overflow = false;
const buffered: SessionOutboundMessage[] = [];
const observers = new Set<SubscriptionObserver<SubscriptionSnapshot<T>>>();
let resolve!: (snapshot: SubscriptionSnapshot<T>) => void;
let reject!: (error: unknown) => void;
const ready = new Promise<SubscriptionSnapshot<T>>((accept, fail) => {
resolve = accept;
reject = fail;
});
void ready.catch(() => {});
const deliver = (
observer: SubscriptionObserver<SubscriptionSnapshot<T>>,
message?: SessionOutboundMessage,
) => {
try {
if (message) observer.update(message);
else if (snapshot) observer.snapshot(snapshot);
} catch (error) {
this.host.failed(error);
}
};
const releaseId = async () => {
const released = id;
id = null;
if (!released) return;
this.routes.delete(released);
await this.host.release(released);
};
const registration: Registration = {
start: () => {
if (disposed || pending || id || !this.connected) return;
const generation = this.generation;
pending = request((value) => {
if (generation !== this.generation) return;
assertSubscriptionSnapshot(value);
id = value.subscriptionId;
snapshot = value;
buffered.length = 0;
overflow = false;
if (disposed) return;
this.routes.set(id, registration);
for (const observer of observers) deliver(observer);
})
.then(() => {
if (generation === this.generation && !disposed) {
if (!snapshot) throw new Error("The host omitted its subscription snapshot");
resolve(snapshot);
}
return undefined;
})
.catch((error) => {
if (generation !== this.generation) return;
reject(error);
for (const observer of observers) {
try {
observer.error?.(error);
} catch (failure) {
this.host.failed(failure);
}
}
void handle.release().catch(this.host.failed);
this.host.failed(error);
requestFailed?.();
return undefined;
})
.finally(() => {
pending = null;
if (generation !== this.generation && !disposed) registration.start();
});
},
disconnected: () => {
id = null;
snapshot = undefined;
buffered.length = 0;
overflow = false;
},
receive: (message) => {
if (disposed || ("payload" in message && Object.is(message.payload, snapshot))) return;
if (buffered.length < MAX_BOOTSTRAP_UPDATES && !overflow) buffered.push(message);
else {
buffered.length = 0;
overflow = true;
}
for (const observer of observers) deliver(observer, message);
},
release: () => {
if (teardown) return teardown;
disposed = true;
signal?.removeEventListener("abort", abort);
observers.clear();
buffered.length = 0;
if (id) this.routes.delete(id);
reject(new Error("Subscription released"));
teardown = (async () => {
await pending;
await releaseId();
this.registrations.delete(registration);
})();
return teardown;
},
};
const abort = () => {
void handle.release().catch(this.host.failed);
};
const handle: OwnedSubscription<T> = {
get subscriptionId() {
return id;
},
ready,
release: registration.release,
subscribe: (observer) => {
if (disposed) throw new Error("Subscription released");
observers.add(observer);
if (overflow) {
overflow = false;
snapshot = undefined;
void releaseId()
.then(() => {
registration.start();
return undefined;
})
.catch(this.host.failed);
} else {
deliver(observer);
for (const message of buffered) deliver(observer, message);
}
return () => {
observers.delete(observer);
};
},
};
this.registrations.add(registration);
if (signal?.aborted) abort();
else {
signal?.addEventListener("abort", abort, { once: true });
registration.start();
}
return handle;
}
owns(message: SessionOutboundMessage): boolean {
const payload = "payload" in message ? message.payload : message;
return (
"subscriptionId" in payload &&
typeof payload.subscriptionId === "string" &&
this.routes.has(payload.subscriptionId)
);
}
receive(message: SessionOutboundMessage): void {
const payload = "payload" in message ? message.payload : message;
if ("subscriptionId" in payload && typeof payload.subscriptionId === "string")
this.routes.get(payload.subscriptionId)?.receive(message);
}
restore(): void {
this.connected = true;
for (const registration of this.registrations) registration.start();
}
disconnected(): void {
this.connected = false;
this.generation++;
this.routes.clear();
for (const registration of this.registrations) registration.disconnected();
}
async close(): Promise<void> {
await Promise.all([...this.registrations].map((registration) => registration.release()));
}
}
function assertSubscriptionSnapshot(value: unknown): asserts value is { subscriptionId: string } {
if (
value &&
typeof value === "object" &&
"subscriptionId" in value &&
typeof value.subscriptionId === "string" &&
value.subscriptionId
)
return;
throw new Error("The host omitted its subscription ID");
}
+234
View File
@@ -0,0 +1,234 @@
import { expect, test } from "vitest";
import type {
AgentSnapshotPayload,
CreationSnapshot,
WorkspaceDescriptorPayload,
} from "@getpaseo/protocol/messages";
import { CreationClient, type CreationResult } from "./index.js";
const workspace: WorkspaceDescriptorPayload = {
id: "wks_0123456789abcdef",
projectId: "project",
projectDisplayName: "Project",
projectRootPath: "/project",
workspaceDirectory: "/project/worktree",
projectKind: "git",
workspaceKind: "worktree",
name: "Worktree",
status: "running",
statusEnteredAt: null,
activityAt: null,
archivingAt: null,
scripts: [],
};
const agent: AgentSnapshotPayload = {
id: "00000000-0000-4000-8000-000000000001",
workspaceId: workspace.id,
provider: "codex",
cwd: "/project/worktree",
model: null,
createdAt: "2026-09-11T00:00:00Z",
updatedAt: "2026-09-11T00:00:00Z",
lastUserMessageAt: null,
status: "idle",
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: true,
supportsRewindConversation: false,
supportsRewindFiles: false,
supportsRewindBoth: false,
},
currentModeId: null,
availableModes: [],
pendingPermissions: [],
persistence: null,
title: "Start once",
labels: {},
};
function fixture(modern: boolean) {
const requests: Array<{ kind: string; input: Record<string, unknown> }> = [];
const legacy: Array<{ kind: string; input: unknown }> = [];
let resolve!: (result: CreationResult) => void;
const result = new Promise<CreationResult>((done) => {
resolve = done;
});
const client = new CreationClient({
supports: () => modern,
requestId: () => "generated-key",
request: async (kind, input) => {
requests.push({ kind, input });
return result;
},
observe: () => () => {},
legacyWorkspace: async (input) => {
legacy.push({ kind: "workspace", input });
return { requestId: "legacy-workspace", workspace, error: null, setupTerminalId: null };
},
legacyAgent: async (input) => {
legacy.push({ kind: "agent", input });
return agent;
},
sendMessage: async (id, text, options) => {
legacy.push({ kind: "message", input: { id, text, ...options } });
},
});
const input = {
idempotencyKey: "intent-one",
source: { kind: "directory" as const, path: "/project" },
agent: {
config: { provider: "codex", cwd: "/project" },
initialPrompt: "Start once",
clientMessageId: "message-one",
},
};
return { client, requests, legacy, resolve, input };
}
test("duplicate client submissions join one complete intent and cumulative updates never regress", async () => {
const f = fixture(true);
const phases: string[] = [];
const first = f.client.createWorkspace({
...f.input,
onEvent: (snapshot) => phases.push(snapshot.phase),
});
const duplicate = f.client.createWorkspace(f.input);
expect(duplicate).toBe(first);
expect(f.requests).toEqual([{ kind: "workspace", input: { ...f.input, subscribe: true } }]);
await expect(
f.client.createWorkspace({
...f.input,
agent: { ...f.input.agent, initialPrompt: "Different intent" },
}),
).rejects.toThrow("workspace_request_key_conflict");
const snapshot: CreationSnapshot = {
kind: "workspace",
idempotencyKey: "intent-one",
revision: 2,
phase: "workspace_ready",
workspaceId: workspace.id,
agentId: agent.id,
workspace,
error: null,
};
f.client.receive(snapshot);
f.client.receive({ ...snapshot, revision: 1, phase: "accepted" });
expect(phases).toEqual(["workspace_ready"]);
const late: string[] = [];
f.client.createWorkspace({ ...f.input, onEvent: (update) => late.push(update.phase) });
expect(late).toEqual(["workspace_ready"]);
f.resolve({
workspace,
agent,
error: null,
setupSkippedReason: "Untrusted source",
requestId: "response-one",
});
expect(await first).toMatchObject({
workspace,
agent,
setupSkippedReason: "Untrusted source",
requestId: "response-one",
});
expect(await duplicate).toEqual(await first);
expect(f.legacy).toEqual([]);
f.client.close();
});
test("legacy adaptation keeps workspace, agent and initial prompt sequencing inside the client", async () => {
const f = fixture(false);
const phases: string[] = [];
const result = await f.client.createWorkspace({
...f.input,
onEvent: (snapshot) => phases.push(snapshot.phase),
});
expect(result).toMatchObject({ workspace, agent, error: null });
expect(phases).toEqual(["workspace_ready"]);
expect(f.legacy).toEqual([
{ kind: "workspace", input: { idempotencyKey: "intent-one", source: f.input.source } },
{
kind: "agent",
input: {
config: { provider: "codex", cwd: workspace.workspaceDirectory },
workspaceId: workspace.id,
idempotencyKey: "intent-one:agent",
clientMessageId: "message-one",
},
},
{
kind: "message",
input: {
id: agent.id,
text: "Start once",
messageId: "message-one",
images: undefined,
attachments: undefined,
},
},
]);
expect(f.requests).toEqual([]);
f.client.close();
});
test("legacy daemons cannot silently replace caller-selected IDs", async () => {
const f = fixture(false);
await expect(f.client.createWorkspace({ ...f.input, workspaceId: workspace.id })).rejects.toThrow(
"caller-selected",
);
await expect(
f.client.createAgent({ agentId: agent.id, config: f.input.agent.config }),
).rejects.toThrow("caller-selected");
expect(f.legacy).toEqual([]);
f.client.close();
});
test("keyed creation with an empty prompt works on legacy daemons without sending an empty message", async () => {
const f = fixture(false);
await f.client.createAgent({
idempotencyKey: "empty-agent",
config: f.input.agent.config,
initialPrompt: "",
});
expect(f.legacy).toEqual([
{ kind: "agent", input: { idempotencyKey: "empty-agent", config: f.input.agent.config } },
]);
f.client.close();
});
test.each([
["/repo/project", "/repo/project/../outside"],
["C:\\repo\\project", "C:\\repo\\project\\..\\outside"],
])("legacy creation rejects escaping %s before creating a workspace", async (source, cwd) => {
const f = fixture(false);
await expect(
f.client.createWorkspace({
...f.input,
source: { kind: "directory", path: source },
agent: { ...f.input.agent, config: { provider: "codex", cwd } },
}),
).rejects.toThrow("inside the workspace source");
expect(f.legacy).toEqual([]);
f.client.close();
});
test.each([
{ source: { kind: "directory" as const, path: "/project" }, cwd: "/project/src" },
{ source: { kind: "directory" as const, path: "C:\\project" }, cwd: "C:\\project\\src" },
{ source: { kind: "worktree" as const, projectId: "project" }, cwd: "/project/src" },
])("legacy creation preserves the agent subdirectory for $source", async ({ source, cwd }) => {
const f = fixture(false);
await f.client.createWorkspace({
...f.input,
source,
agent: { config: { provider: "codex", cwd } },
});
expect(f.legacy[1]).toMatchObject({
kind: "agent",
input: { workspaceId: workspace.id, config: { cwd: "/project/worktree/src" } },
});
f.client.close();
});
+276
View File
@@ -0,0 +1,276 @@
import type {
AgentSnapshotPayload,
CreationSnapshot,
WorkspaceCreateResponse,
} from "@getpaseo/protocol/messages";
import type { CreateAgentRequestOptions, CreateWorkspaceRequestOptions } from "../daemon-client.js";
type Kind = CreationSnapshot["kind"];
type Observer = (snapshot: CreationSnapshot) => void;
export interface CreationResult {
agent?: AgentSnapshotPayload | null;
workspace?: WorkspaceCreateResponse["payload"]["workspace"];
creation?: CreationSnapshot;
error: string | null;
errorCode?: string;
setupTerminalId?: string | null;
setupSkippedReason?: string;
requestId?: string;
}
interface Dependencies {
supports: () => boolean;
requestId: () => string;
request: (kind: Kind, input: Record<string, unknown>) => Promise<CreationResult>;
observe: (
kind: Kind,
key: string,
next: (snapshot: CreationSnapshot | null) => void,
error: (error: unknown) => void,
) => () => void;
legacyAgent: (input: CreateAgentRequestOptions) => Promise<AgentSnapshotPayload>;
legacyWorkspace: (
input: CreateWorkspaceRequestOptions,
) => Promise<WorkspaceCreateResponse["payload"]>;
sendMessage: (
id: string,
text: string,
options: {
messageId: string;
images?: CreateAgentRequestOptions["images"];
attachments?: CreateAgentRequestOptions["attachments"];
},
) => Promise<unknown>;
}
interface Operation {
kind: Kind;
key: string;
fingerprint: string;
input: Record<string, unknown>;
observers: Set<Observer>;
snapshot?: CreationSnapshot;
result: Promise<CreationResult>;
resolve: (result: CreationResult) => void;
reject: (error: Error) => void;
settled: boolean;
modern: boolean;
recovering: boolean;
stopObserving?: () => void;
}
/** Client-owned observation and compatibility. UI callbacks never advance creation. */
export class CreationClient {
private readonly operations = new Map<string, Operation>();
constructor(private readonly deps: Dependencies) {}
createAgent(input: CreateAgentRequestOptions): Promise<CreationResult> {
const { onEvent, ...request } = input;
return this.start("agent", request, onEvent, () => this.legacyAgent(request));
}
createWorkspace(input: CreateWorkspaceRequestOptions): Promise<CreationResult> {
const { onEvent, ...request } = input;
return this.start("workspace", request, onEvent, async (operation) => {
if (request.workspaceId || request.agent?.agentId)
throw new Error("Update the host to use caller-selected creation IDs.");
const { agent, ...workspaceInput } = request;
const sourceCwd =
request.source.kind === "directory" ? request.source.path : request.source.cwd;
const relativeCwd =
agent && sourceCwd ? relativeDirectory(agent.config!.cwd, sourceCwd) : undefined;
const workspace = await this.deps.legacyWorkspace(workspaceInput);
if (workspace.error || !workspace.workspace) return workspace;
this.receive({
kind: "workspace",
idempotencyKey: operation.key,
revision: 1,
phase: "workspace_ready",
workspaceId: workspace.workspace.id,
agentId: null,
workspace: workspace.workspace,
setupSkippedReason: workspace.setupSkippedReason,
error: null,
});
if (!agent) return workspace;
const cwd = `${workspace.workspace.workspaceDirectory!.replace(/[\\/]+$/, "")}${
relativeCwd ?? relativeDirectory(agent.config!.cwd, workspace.workspace.projectRootPath)
}`;
const created = await this.legacyAgent({
...agent,
config: { ...agent.config!, cwd },
workspaceId: workspace.workspace.id,
idempotencyKey: `${operation.key}:agent`,
});
return { ...workspace, agent: created.agent };
});
}
private async legacyAgent(input: CreateAgentRequestOptions): Promise<CreationResult> {
if (input.agentId) throw new Error("Update the host to use caller-selected creation IDs.");
// COMPAT(creationLifecycle): added in v0.8.0, remove after 2027-03-11 once the daemon floor supports creationLifecycle.
// Old keyed creates reject initialPrompt. Keep this adaptation private to the SDK.
if (input.idempotencyKey) {
const { initialPrompt = "", images, attachments, ...creation } = input;
const hasPrompt = Boolean(initialPrompt.trim() || images?.length || attachments?.length);
if (hasPrompt && input.outputSchema)
throw new Error("Update the host to create a keyed agent with structured initial output.");
const agent = await this.deps.legacyAgent(creation);
if (hasPrompt)
await this.deps.sendMessage(agent.id, initialPrompt, {
messageId: input.clientMessageId ?? `${input.idempotencyKey}:initial-message`,
images,
attachments,
});
return { agent, error: null };
}
return { agent: await this.deps.legacyAgent(input), error: null };
}
private start(
kind: Kind,
input: Record<string, unknown>,
observer: Observer | undefined,
legacy: (operation: Operation) => Promise<CreationResult>,
): Promise<CreationResult> {
const key =
typeof input.idempotencyKey === "string" ? input.idempotencyKey : this.deps.requestId();
const identity = JSON.stringify([kind, key]);
const { requestId: _requestId, ...intent } = input;
const fingerprint = JSON.stringify(intent, (_key, value: unknown) =>
value && typeof value === "object" && !Array.isArray(value)
? Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b)))
: value,
);
const previous = this.operations.get(identity);
if (previous) {
if (previous.fingerprint !== fingerprint)
return Promise.reject(new Error(`${kind}_request_key_conflict`));
if (observer) {
previous.observers.add(observer);
if (previous.snapshot) this.notify(observer, previous.snapshot);
}
return previous.result;
}
let resolve!: Operation["resolve"];
let reject!: Operation["reject"];
const result = new Promise<CreationResult>((done, fail) => {
resolve = done;
reject = fail;
});
const operation: Operation = {
kind,
key,
input: { ...input, idempotencyKey: key },
fingerprint,
observers: new Set(observer ? [observer] : []),
result,
resolve,
reject,
settled: false,
modern: this.deps.supports(),
recovering: false,
};
this.operations.set(identity, operation);
if (operation.modern) void this.submit(operation);
else
void legacy(operation).then(
(value) => this.finish(operation, value),
(error) => this.fail(operation, error),
);
return result;
}
private async submit(operation: Operation): Promise<void> {
try {
const result = await this.deps.request(operation.kind, {
...operation.input,
subscribe: true,
});
if (result.creation) this.receive(result.creation);
this.finish(operation, result);
} catch (error) {
if (
operation.settled ||
(error instanceof Error && "code" in error && error.code === "DAEMON_CONNECTION_LOST")
)
return;
this.fail(operation, error);
}
}
receive(snapshot: CreationSnapshot): void {
const operation = this.operations.get(JSON.stringify([snapshot.kind, snapshot.idempotencyKey]));
if (!operation || (operation.snapshot && operation.snapshot.revision > snapshot.revision))
return;
if (operation.snapshot?.revision !== snapshot.revision) {
operation.snapshot = snapshot;
for (const observer of operation.observers) this.notify(observer, snapshot);
}
if (operation.recovering && (snapshot.phase === "completed" || snapshot.phase === "failed")) {
this.finish(operation, {
agent: snapshot.agent,
workspace: snapshot.workspace,
creation: snapshot,
setupSkippedReason: snapshot.setupSkippedReason,
error: snapshot.error,
errorCode: snapshot.errorCode,
});
}
}
reconnect(): void {
for (const operation of this.operations.values()) {
if (operation.settled || !operation.modern || operation.recovering) continue;
operation.recovering = true;
operation.stopObserving = this.deps.observe(
operation.kind,
operation.key,
(snapshot) => {
if (!snapshot) void this.submit(operation);
else {
// Re-subscription is authoritative even if its revision matches the last live event.
this.receive(snapshot);
}
},
(error) => this.fail(operation, error),
);
}
}
close(): void {
for (const operation of this.operations.values())
if (!operation.settled) this.fail(operation, new Error("Daemon client closed"));
this.operations.clear();
}
private finish(operation: Operation, result: CreationResult): void {
if (operation.settled) return;
operation.settled = true;
operation.stopObserving?.();
operation.resolve(result);
operation.observers.clear();
this.operations.delete(JSON.stringify([operation.kind, operation.key]));
}
private fail(operation: Operation, error: unknown): void {
if (operation.settled) return;
operation.settled = true;
operation.stopObserving?.();
operation.reject(error instanceof Error ? error : new Error(String(error)));
operation.observers.clear();
this.operations.delete(JSON.stringify([operation.kind, operation.key]));
}
private notify(observer: Observer, snapshot: CreationSnapshot): void {
try {
observer(snapshot);
} catch {
/* Observation does not control execution. */
}
}
}
function relativeDirectory(cwd: string, source: string): string {
const normalize = (value: string) => value.replace(/\\/g, "/").replace(/\/+$/, "");
const normalized = normalize(cwd),
root = normalize(source);
// Parent traversal must never survive mapping into the newly created checkout.
if ([normalized, root].some((path) => path.split("/").includes("..")))
throw new Error("Agent directory must be inside the workspace source");
if (normalized === root) return "";
if (!normalized.startsWith(`${root}/`))
throw new Error("Agent directory must be inside the workspace source");
return normalized.slice(root.length);
}
@@ -21,7 +21,7 @@ describe("daemon-client transport helpers", () => {
createClientChannelMock.mockReset();
createClientChannelMock.mockRejectedValueOnce(new Error("handshake failed"));
let openHandler: (() => void) | null = null;
const connection: { open: (() => void) | null } = { open: null };
const close = vi.fn();
createEncryptedTransport(
@@ -29,10 +29,10 @@ describe("daemon-client transport helpers", () => {
send: vi.fn(),
close,
onOpen: (handler) => {
openHandler = handler;
connection.open = handler;
return () => {
if (openHandler === handler) {
openHandler = null;
if (connection.open === handler) {
connection.open = null;
}
};
},
@@ -44,8 +44,8 @@ describe("daemon-client transport helpers", () => {
{ warn: vi.fn() },
);
expect(openHandler).not.toBeNull();
openHandler?.();
expect(connection.open).not.toBeNull();
connection.open?.();
await vi.waitFor(() => {
expect(close).toHaveBeenCalledWith(4001, "E2EE handshake failed");
+487 -136
View File
@@ -14,9 +14,6 @@ import {
FileTransferOpcode,
} from "@getpaseo/protocol/binary-frames/index";
import {
asUint8Array,
decodeTerminalResizePayload,
decodeTerminalStreamFrame,
encodeTerminalSnapshotPayload,
encodeTerminalStreamFrame,
TerminalStreamOpcode,
@@ -79,7 +76,7 @@ function createMockTransport() {
},
close: () => {},
onMessage: (handler) => {
onMessage = handler;
onMessage = (data) => handler(data, typeof data !== "string");
return () => {};
},
onOpen: (handler) => {
@@ -115,7 +112,7 @@ function createMockTransport() {
serverId: `srv_test_${serverInfoOrdinal++}`,
hostname: null,
version: null,
...(options?.features ? { features: options.features } : {}),
features: options?.features ?? { ownedSubscriptions: true },
},
},
}),
@@ -255,7 +252,7 @@ test("does not infer browser automation capabilities from Electron runtime", asy
const hello = z
.object({
type: z.literal("hello"),
capabilities: z.record(z.unknown()),
capabilities: z.record(z.string(), z.unknown()),
})
.parse(JSON.parse(assertStr(mock.sent[0])));
expect(hello.capabilities[CLIENT_CAPS.browserHost]).toBeUndefined();
@@ -285,7 +282,7 @@ test("advertises consumer-provided browser automation capabilities", async () =>
const hello = z
.object({
type: z.literal("hello"),
capabilities: z.record(z.unknown()),
capabilities: z.record(z.string(), z.unknown()),
})
.parse(JSON.parse(assertStr(mock.sent[0])));
expect(hello.capabilities[CLIENT_CAPS.browserHost]).toEqual({
@@ -331,57 +328,35 @@ test("Hub management requires daemon support before dispatching requests", async
expect(mock.sent).toEqual([]);
});
test("sets the complete viewed timeline subscription only when the daemon supports it", async () => {
const supportedTransport = createMockTransport();
const supportedClient = new DaemonClient({
url: "ws://test",
clientId: "timeline_supported",
transportFactory: () => supportedTransport.transport,
reconnect: { enabled: false },
});
const legacyTransport = createMockTransport();
const legacyClient = new DaemonClient({
test("timeline observation consumes broadcasts from a host without selective delivery", async () => {
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "timeline_legacy",
transportFactory: () => legacyTransport.transport,
transportFactory: () => mock.transport,
reconnect: { enabled: false },
});
clients.push(supportedClient, legacyClient);
const supportedConnect = supportedClient.connect();
supportedTransport.triggerOpen({ features: { selectiveAgentTimeline: true } });
await supportedConnect;
const legacyConnect = legacyClient.connect();
legacyTransport.triggerOpen();
await legacyConnect;
expect(supportedClient.getLastServerInfoMessage()?.features).toEqual({
selectiveAgentTimeline: true,
});
const setPromise = supportedClient.setAgentTimelineSubscription(["agent-b", "agent-a"]);
await Promise.resolve();
const request = parseSentFrame(supportedTransport.sent[0]);
supportedTransport.triggerMessage(
clients.push(client);
const connecting = client.connect();
mock.triggerOpen({ features: {} });
await connecting;
const observation = client.observeTimeline(["agent"]);
await expect(observation.ready).resolves.toMatchObject({ agentIds: ["agent"] });
const updates: unknown[] = [];
observation.subscribe({ snapshot: () => {}, update: (message) => updates.push(message) });
mock.triggerMessage(
wrapSessionMessage({
type: "agent.timeline.set_subscription.response",
type: "agent_stream",
payload: {
requestId: request.requestId,
agentIds: ["agent-a", "agent-b"],
agentId: "agent",
timestamp: "2026-09-11T00:00:00Z",
event: { type: "turn_completed", provider: "codex" },
},
}),
);
await setPromise;
await legacyClient.setAgentTimelineSubscription(["agent-a"]);
expect({ request, legacyFrames: legacyTransport.sent }).toEqual({
request: {
type: "agent.timeline.set_subscription.request",
requestId: expect.any(String),
agentIds: ["agent-a", "agent-b"],
},
legacyFrames: [],
});
expect(updates).toEqual([expect.objectContaining({ type: "agent_stream" })]);
await observation.release();
expect(mock.sent).toEqual([]);
});
test("normalizes legacy and dedicated agent attention notifications", async () => {
@@ -444,8 +419,13 @@ class FakeDaemon {
if (typeof data !== "string") {
return;
}
const frame = JSON.parse(data) as { type?: string };
if (frame.type !== "ping") {
const frame = JSON.parse(data) as {
type?: string;
message?: { type?: string; requestId: string; clientSentAt: number };
};
const sessionPing =
frame.type === "session" && frame.message?.type === "ping" ? frame.message : null;
if (frame.type !== "ping" && !sessionPing) {
return;
}
this.pingsSentAt.push(performance.now());
@@ -456,19 +436,25 @@ class FakeDaemon {
if (this.pongMode.kind === "silent") {
return;
}
const pong = sessionPing
? wrapSessionMessage({
type: "pong",
payload: { ...sessionPing, serverReceivedAt: Date.now(), serverSentAt: Date.now() },
})
: JSON.stringify({ type: "pong" });
if (this.pongMode.delayMs === 0) {
this.onMessage(JSON.stringify({ type: "pong" }));
this.onMessage(pong);
return;
}
setTimeout(() => {
this.onMessage(JSON.stringify({ type: "pong" }));
this.onMessage(pong);
}, this.pongMode.delayMs);
},
close: (code?: number, reason?: string) => {
this.closeEvents.push({ code, reason });
},
onMessage: (handler) => {
this.onMessage = handler;
this.onMessage = (data) => handler(data, typeof data !== "string");
return () => {};
},
onOpen: (handler) => {
@@ -760,10 +746,12 @@ test("advertises client capabilities in hello", async () => {
timeline_replacement_invalidation: true,
provider_snapshot_references: true,
explicit_event_subscriptions: true,
owned_subscriptions: true,
compact_provider_snapshots: true,
custom_mode_icons: true,
project_updates: true,
provider_subagents: true,
projected_subagent_timeline: true,
reasoning_merge_enum: true,
terminal_reflowable_snapshot: true,
timeline_notifications: true,
@@ -797,7 +785,7 @@ test("allows callers to disable default client capabilities", async () => {
const hello = z
.object({
type: z.literal("hello"),
capabilities: z.record(z.unknown()),
capabilities: z.record(z.string(), z.unknown()),
})
.parse(JSON.parse(assertStr(mock.sent[0])));
expect(hello.capabilities[CLIENT_CAPS.projectUpdates]).toBe(false);
@@ -1012,6 +1000,49 @@ test("ensureConnected reconnects immediately without leaving the scheduled retry
}
});
test("connect during a scheduled retry reconnects immediately without leaving the retry armed", async () => {
useHeartbeatClock();
try {
const first = createMockTransport();
const second = createMockTransport();
const third = createMockTransport();
const transports = [first, second, third];
let transportIndex = 0;
const client = new DaemonClient({
url: "ws://test",
clientId: "clsk_explicit_reconnect",
reconnect: { enabled: true, baseDelayMs: 1_500, maxDelayMs: 1_500 },
transportFactory: () => {
const transport = transports[transportIndex];
if (!transport) throw new Error("unexpected extra reconnect");
transportIndex += 1;
return transport.transport;
},
});
clients.push(client);
const initialConnect = client.connect();
first.triggerOpen();
await initialConnect;
first.triggerClose({ code: 1001, reason: "daemon restarted" });
expect(client.getConnectionState().status).toBe("disconnected");
const reconnect = client.connect();
expect(client.getConnectionState().status).toBe("connecting");
expect(transportIndex).toBe(2);
second.triggerOpen();
await reconnect;
expect(client.getConnectionState().status).toBe("connected");
await vi.advanceTimersByTimeAsync(1_500);
expect(client.getConnectionState().status).toBe("connected");
expect(transportIndex).toBe(2);
} finally {
vi.useRealTimers();
}
});
test("disabling reconnect cancels a pending retry until explicitly resumed", async () => {
useHeartbeatClock();
try {
@@ -1469,7 +1500,7 @@ test("honors explicit getDaemonStatus timeout below the session RPC default", as
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen();
mock.triggerOpen({ features: { daemonStatusRpc: true } });
await connectPromise;
const responsePromise = client.getDaemonStatus({
@@ -1740,6 +1771,78 @@ test("keeps default connect timeout shorter than session RPC waiters", async ()
}
});
test("foreground verification replaces a silently broken socket without waiting for heartbeats", async () => {
useHeartbeatClock();
const first = new FakeDaemon();
const second = new FakeDaemon();
first.daemonGoesSilent();
let attempts = 0;
const client = new DaemonClient({
url: "ws://test",
clientId: "foreground-recovery",
logger: noopLogger,
transportFactory: () => (++attempts === 1 ? first : second).transport,
});
clients.push(client);
const connection = client.connect();
first.openConnection();
await connection;
client.ensureConnected({ verify: true });
await vi.advanceTimersByTimeAsync(3_000);
expect(attempts).toBe(2);
second.openConnection();
expect(client.getConnectionState()).toEqual({ status: "connected" });
});
test("foreground verification preserves a healthy socket and deduplicates simultaneous checks", async () => {
useHeartbeatClock();
const daemon = new FakeDaemon();
daemon.daemonAnswersPingsAfter("0.1s");
const client = new DaemonClient({
url: "ws://test",
clientId: "healthy-resume",
logger: noopLogger,
transportFactory: () => daemon.transport,
});
clients.push(client);
const connection = client.connect();
daemon.openConnection();
await connection;
client.ensureConnected({ verify: true });
client.ensureConnected({ verify: true });
await vi.advanceTimersByTimeAsync(3_000);
expect(client.getConnectionState()).toEqual({ status: "connected" });
expect(daemon.pingTimestamps()).toEqual(["0s"]);
expect(daemon.closesFromClient()).toEqual([]);
});
test("an obsolete foreground probe cannot close a replacement connection", async () => {
useHeartbeatClock();
const first = new FakeDaemon();
const second = new FakeDaemon();
first.daemonGoesSilent();
let attempts = 0;
const client = new DaemonClient({
url: "ws://test",
clientId: "resume-race",
logger: noopLogger,
transportFactory: () => (++attempts === 1 ? first : second).transport,
});
clients.push(client);
const connection = client.connect();
first.openConnection();
await connection;
client.ensureConnected({ verify: true });
first.daemonClosesWith("network changed");
client.ensureConnected();
second.openConnection();
await vi.advanceTimersByTimeAsync(3_000);
expect(attempts).toBe(2);
expect(client.getConnectionState()).toEqual({ status: "connected" });
expect(second.closesFromClient()).toEqual([]);
});
test("stays online through ten minutes of pongs that arrive five seconds late", async () => {
useHeartbeatClock();
const session = new DaemonClientSession();
@@ -3642,16 +3745,14 @@ test("subscribes to checkout diff updates via RPC handshake", async () => {
mock.triggerOpen();
await connectPromise;
const promise = client.subscribeCheckoutDiff(
"/tmp/project",
{ mode: "uncommitted" },
{ subscriptionId: "checkout-sub-1" },
);
const subscription = client.observeCheckoutDiff("/tmp/project", { mode: "uncommitted" });
const promise = subscription.ready;
await vi.waitFor(() => expect(mock.sent).toHaveLength(1));
expect(mock.sent).toHaveLength(1);
const request = parseSentFrame(mock.sent[0]);
expect(request.type).toBe("subscribe_checkout_diff_request");
expect(request.subscriptionId).toBe("checkout-sub-1");
expect(request.subscriptionId).toBeUndefined();
expect(request.cwd).toBe("/tmp/project");
expect(request.compare).toEqual({ mode: "uncommitted" });
@@ -3680,7 +3781,7 @@ test("subscribes to checkout diff updates via RPC handshake", async () => {
});
});
test("getCheckoutDiff uses one-shot subscription protocol", async () => {
test("getCheckoutDiff reads a snapshot without creating a subscription", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
@@ -3701,7 +3802,7 @@ test("getCheckoutDiff uses one-shot subscription protocol", async () => {
expect(mock.sent).toHaveLength(1);
const subscribeRequest = parseSentFrame(mock.sent[0]);
expect(subscribeRequest.type).toBe("subscribe_checkout_diff_request");
expect(subscribeRequest.type).toBe("checkout.diff.get.request");
expect(subscribeRequest.cwd).toBe("/tmp/project");
expect(subscribeRequest.compare).toEqual({
mode: "base",
@@ -3712,9 +3813,8 @@ test("getCheckoutDiff uses one-shot subscription protocol", async () => {
JSON.stringify({
type: "session",
message: {
type: "subscribe_checkout_diff_response",
type: "checkout.diff.get.response",
payload: {
subscriptionId: subscribeRequest.subscriptionId,
cwd: "/tmp/project",
files: [],
error: null,
@@ -3731,10 +3831,7 @@ test("getCheckoutDiff uses one-shot subscription protocol", async () => {
requestId: subscribeRequest.requestId,
});
expect(mock.sent).toHaveLength(2);
const unsubscribeRequest = parseSentFrame(mock.sent[1]);
expect(unsubscribeRequest.type).toBe("unsubscribe_checkout_diff_request");
expect(unsubscribeRequest.subscriptionId).toBe(subscribeRequest.subscriptionId);
expect(mock.sent).toHaveLength(1);
});
test("requests branch suggestions via RPC", async () => {
@@ -4279,7 +4376,7 @@ test("renames a branch via RPC", async () => {
});
expect(mock.sent).toHaveLength(1);
const request = JSON.parse(mock.sent[0]) as {
const request = JSON.parse(assertStr(mock.sent[0])) as {
type: "session";
message: {
type: "checkout.rename_branch.request";
@@ -4381,29 +4478,52 @@ test("resubscribes checkout diff streams after reconnect", async () => {
});
clients.push(client);
const internal = client as unknown as {
checkoutDiffSubscriptions: Map<
string,
{ cwd: string; compare: { mode: "uncommitted" | "base"; baseRef?: string } }
>;
};
internal.checkoutDiffSubscriptions.set("checkout-sub-1", {
cwd: "/tmp/project",
compare: { mode: "base", baseRef: "main" },
});
const connectPromise = client.connect();
mock.triggerOpen();
await connectPromise;
expect(mock.sent).toHaveLength(1);
const observation = client.observeCheckoutDiff("/tmp/project", { mode: "base", baseRef: "main" });
await vi.waitFor(() => expect(mock.sent).toHaveLength(1));
const first = parseSentFrame(mock.sent[0]);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_checkout_diff_response",
payload: {
requestId: first.requestId,
subscriptionId: "server-first",
cwd: "/tmp/project",
files: [],
error: null,
},
}),
);
await observation.ready;
mock.triggerClose();
const reconnect = client.connect();
mock.triggerOpen();
await reconnect;
await vi.waitFor(() => expect(mock.sent).toHaveLength(1));
const request = parseSentFrame(mock.sent[0]);
expect(request.type).toBe("subscribe_checkout_diff_request");
expect(request.subscriptionId).toBe("checkout-sub-1");
expect(request.cwd).toBe("/tmp/project");
expect(request.compare).toEqual({ mode: "base", baseRef: "main" });
expect(typeof request.requestId).toBe("string");
expect(z.string().parse(request.requestId).length).toBeGreaterThan(0);
expect(request).toMatchObject({
type: "subscribe_checkout_diff_request",
cwd: "/tmp/project",
compare: { mode: "base", baseRef: "main" },
});
expect(request.subscriptionId).toBeUndefined();
expect(request.requestId).not.toBe(first.requestId);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_checkout_diff_response",
payload: {
requestId: request.requestId,
subscriptionId: "server-second",
cwd: "/tmp/project",
files: [],
error: null,
},
}),
);
await vi.waitFor(() => expect(observation.subscriptionId).toBe("server-second"));
});
test("fetches agents via RPC with filters, sort, and pagination", async () => {
@@ -4430,10 +4550,10 @@ test("fetches agents via RPC with filters, sort, and pagination", async () => {
{ key: "created_at", direction: "desc" },
],
page: { limit: 25, cursor: "cursor-1" },
subscribe: { subscriptionId: "sub-1" },
subscribe: {},
});
expect(mock.sent).toHaveLength(1);
await vi.waitFor(() => expect(mock.sent).toHaveLength(1));
const request = parseSentFrame(mock.sent[0]);
expect(request.type).toBe("fetch_agents_request");
expect(request.sort).toEqual([
@@ -4441,7 +4561,7 @@ test("fetches agents via RPC with filters, sort, and pagination", async () => {
{ key: "created_at", direction: "desc" },
]);
expect(request.page).toEqual({ limit: 25, cursor: "cursor-1" });
expect(request.subscribe).toEqual({ subscriptionId: "sub-1" });
expect(request.subscribe).toEqual({});
mock.triggerMessage(
JSON.stringify({
@@ -4465,6 +4585,10 @@ test("fetches agents via RPC with filters, sort, and pagination", async () => {
await expect(promise).resolves.toEqual({
requestId: request.requestId,
subscriptionId: "sub-1",
subscription: expect.objectContaining({
subscriptionId: "sub-1",
release: expect.any(Function),
}),
entries: [],
pageInfo: {
nextCursor: null,
@@ -5310,11 +5434,15 @@ test("emits output events for the active terminal stream", async () => {
seen.push(new TextDecoder().decode(event.data));
});
const subscribePromise = client.subscribeTerminal("term-1", "sub-1");
const subscribePromise = client.subscribeTerminal("term-1", { requestId: "sub-1" });
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscribe_terminal_request"),
);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_terminal_response",
payload: {
subscriptionId: "server-" + String(parseSentFrame(mock.sent.at(-1)).requestId),
terminalId: "term-1",
slot: 11,
error: null,
@@ -5362,11 +5490,15 @@ test("emits snapshot events for the subscribed terminal stream", async () => {
snapshots.push(event.state);
});
const subscribePromise = client.subscribeTerminal("term-1", "sub-2");
const subscribePromise = client.subscribeTerminal("term-1", { requestId: "sub-2" });
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscribe_terminal_request"),
);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_terminal_response",
payload: {
subscriptionId: "server-" + String(parseSentFrame(mock.sent.at(-1)).requestId),
terminalId: "term-1",
slot: 12,
error: null,
@@ -5394,7 +5526,7 @@ test("emits snapshot events for the subscribed terminal stream", async () => {
expect(snapshots).toEqual([state]);
});
test("sends input and resize frames for the subscribed terminal slot", async () => {
test("sends explicit terminal input and resize commands", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
@@ -5411,11 +5543,15 @@ test("sends input and resize frames for the subscribed terminal slot", async ()
mock.triggerOpen();
await connectPromise;
const subscribePromise = client.subscribeTerminal("term-1", "sub-3");
const subscribePromise = client.subscribeTerminal("term-1", { requestId: "sub-3" });
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscribe_terminal_request"),
);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_terminal_response",
payload: {
subscriptionId: "server-" + String(parseSentFrame(mock.sent.at(-1)).requestId),
terminalId: "term-1",
slot: 13,
error: null,
@@ -5437,19 +5573,18 @@ test("sends input and resize frames for the subscribed terminal slot", async ()
intent: "update",
});
const inputFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[0])!);
const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!);
expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input);
expect(inputFrame?.slot).toBe(13);
expect(new TextDecoder().decode(inputFrame?.payload ?? new Uint8Array())).toBe("echo hello\r");
expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize);
expect(resizeFrame?.slot).toBe(13);
expect(decodeTerminalResizePayload(resizeFrame?.payload ?? new Uint8Array())).toEqual({
rows: 24,
cols: 80,
intent: "update",
});
expect(mock.sent.map(parseSentFrame)).toEqual([
{
type: "terminal_input",
terminalId: "term-1",
message: { type: "input", data: "echo hello\r" },
},
{
type: "terminal_input",
terminalId: "term-1",
message: { type: "resize", rows: 24, cols: 80, intent: "update" },
},
]);
});
test("routes concurrent terminal stream frames by slot", async () => {
@@ -5477,11 +5612,15 @@ test("routes concurrent terminal stream frames by slot", async () => {
seen.push(`${event.terminalId}:${new TextDecoder().decode(event.data)}`);
});
const subscribeFirstPromise = client.subscribeTerminal("term-1", "sub-multi-1");
const subscribeFirstPromise = client.subscribeTerminal("term-1", { requestId: "sub-multi-1" });
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscribe_terminal_request"),
);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_terminal_response",
payload: {
subscriptionId: "server-" + String(parseSentFrame(mock.sent.at(-1)).requestId),
terminalId: "term-1",
slot: 21,
error: null,
@@ -5491,11 +5630,15 @@ test("routes concurrent terminal stream frames by slot", async () => {
);
await subscribeFirstPromise;
const subscribeSecondPromise = client.subscribeTerminal("term-2", "sub-multi-2");
const subscribeSecondPromise = client.subscribeTerminal("term-2", { requestId: "sub-multi-2" });
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscribe_terminal_request"),
);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_terminal_response",
payload: {
subscriptionId: "server-" + String(parseSentFrame(mock.sent.at(-1)).requestId),
terminalId: "term-2",
slot: 22,
error: null,
@@ -5531,14 +5674,19 @@ test("routes concurrent terminal stream frames by slot", async () => {
cols: 20,
});
const inputFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[0])!);
const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!);
expect(seen).toEqual(["term-2:beta", "term-1:alpha"]);
expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input);
expect(inputFrame?.slot).toBe(22);
expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize);
expect(resizeFrame?.slot).toBe(21);
expect(mock.sent.map(parseSentFrame)).toEqual([
{
type: "terminal_input",
terminalId: "term-2",
message: { type: "input", data: "echo beta\r" },
},
{
type: "terminal_input",
terminalId: "term-1",
message: { type: "resize", rows: 10, cols: 20 },
},
]);
});
test("ignores terminal stream frames after terminal_stream_exit", async () => {
@@ -5566,11 +5714,15 @@ test("ignores terminal stream frames after terminal_stream_exit", async () => {
seen.push(new TextDecoder().decode(event.data));
});
const subscribePromise = client.subscribeTerminal("term-1", "sub-4");
const subscribePromise = client.subscribeTerminal("term-1", { requestId: "sub-4" });
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscribe_terminal_request"),
);
mock.triggerMessage(
wrapSessionMessage({
type: "subscribe_terminal_response",
payload: {
subscriptionId: "server-" + String(parseSentFrame(mock.sent.at(-1)).requestId),
terminalId: "term-1",
slot: 14,
error: null,
@@ -5593,6 +5745,7 @@ test("ignores terminal stream frames after terminal_stream_exit", async () => {
wrapSessionMessage({
type: "terminal_stream_exit",
payload: {
subscriptionId: "server-sub-4",
terminalId: "term-1",
},
}),
@@ -5607,6 +5760,21 @@ test("ignores terminal stream frames after terminal_stream_exit", async () => {
);
expect(seen).toEqual(["before-exit"]);
await vi.waitFor(() =>
expect(parseSentFrame(mock.sent.at(-1)).type).toBe("subscription.release.request"),
);
const release = parseSentFrame(mock.sent.at(-1));
mock.triggerMessage(
wrapSessionMessage({
type: "subscription.release.response",
payload: {
requestId: release.requestId,
subscriptionId: "server-sub-4",
released: true,
},
}),
);
await Promise.resolve();
unsubscribe();
});
@@ -5952,24 +6120,41 @@ test("sends subscribe/unsubscribe terminals messages", async () => {
mock.triggerOpen();
await connectPromise;
client.subscribeTerminals({ cwd: "/tmp/project" });
client.unsubscribeTerminals({ cwd: "/tmp/project" });
expect(mock.sent).toHaveLength(2);
expect(JSON.parse(assertStr(mock.sent[0]))).toEqual({
type: "session",
message: {
type: "subscribe_terminals_request",
cwd: "/tmp/project",
},
});
expect(JSON.parse(assertStr(mock.sent[1]))).toEqual({
type: "session",
message: {
type: "unsubscribe_terminals_request",
cwd: "/tmp/project",
},
const observation = client.observeTerminals({ cwd: "/tmp/project" });
await vi.waitFor(() => expect(mock.sent).toHaveLength(1));
const request = parseSentFrame(mock.sent[0]);
expect(request).toMatchObject({ type: "subscribe_terminals_request", cwd: "/tmp/project" });
expect(request.subscriptionId).toBeUndefined();
mock.triggerMessage(
wrapSessionMessage({
type: "terminals_changed",
payload: {
requestId: request.requestId,
subscriptionId: "server-terminals",
cwd: "/tmp/project",
terminals: [],
},
}),
);
await observation.ready;
const release = observation.release();
await vi.waitFor(() => expect(mock.sent).toHaveLength(2));
const requestRelease = parseSentFrame(mock.sent[1]);
expect(requestRelease).toMatchObject({
type: "subscription.release.request",
subscriptionId: "server-terminals",
});
mock.triggerMessage(
wrapSessionMessage({
type: "subscription.release.response",
payload: {
requestId: requestRelease.requestId,
subscriptionId: "server-terminals",
released: true,
},
}),
);
await release;
});
test("dispatches terminals_changed events to typed listeners", async () => {
@@ -6264,3 +6449,169 @@ test("wire snapshot callers own expansion and receive hash references unchanged"
);
expect(await request).toEqual(body);
});
test("creation lifecycle sends the keyed agent and initial prompt as one intent", async () => {
const transport = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "creation-contract",
transportFactory: () => transport.transport,
reconnect: { enabled: false },
});
clients.push(client);
const connected = client.connect();
transport.triggerOpen({ features: { creationLifecycle: true, agentRequestReceipts: true } });
await connected;
const creation = client.createAgent({
idempotencyKey: "draft-one",
provider: "codex",
cwd: "/project",
workspaceId: "wks_0123456789abcdef",
initialPrompt: "Start once",
clientMessageId: "first-message",
});
void creation.catch(() => undefined);
const request = parseSentFrame(transport.sent[0]);
expect(request).toMatchObject({
type: "agent.create.request",
idempotencyKey: "draft-one",
initialPrompt: "Start once",
clientMessageId: "first-message",
});
transport.triggerMessage(
wrapSessionMessage({
type: "agent.create.response",
payload: { requestId: request.requestId, agent: null, error: "provider unavailable" },
}),
);
await expect(creation).rejects.toThrow("provider unavailable");
expect(transport.sent).toHaveLength(1);
});
test("creation lifecycle acknowledgement reaches the observer before the final response", async () => {
const transport = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "creation-progress",
transportFactory: () => transport.transport,
reconnect: { enabled: false },
});
clients.push(client);
const connected = client.connect();
transport.triggerOpen({ features: { creationLifecycle: true } });
await connected;
const phases: string[] = [];
const creation = client.createAgent({
provider: "codex",
cwd: "/project",
idempotencyKey: "observe-one",
onEvent: (snapshot) => phases.push(snapshot.phase),
});
void creation.catch(() => undefined);
const request = parseSentFrame(transport.sent[0]);
transport.triggerMessage(
wrapSessionMessage({
type: "agent.create.update",
payload: {
kind: "agent",
idempotencyKey: "observe-one",
revision: 0,
phase: "accepted",
workspaceId: null,
agentId: "00000000-0000-4000-8000-000000000001",
error: null,
},
}),
);
expect(phases).toEqual(["accepted"]);
transport.triggerMessage(
wrapSessionMessage({
type: "agent.create.response",
payload: { requestId: request.requestId, agent: null, error: "provider unavailable" },
}),
);
await expect(creation).rejects.toThrow("provider unavailable");
});
test("creation reconnect observation uses connection-owned subscriptions and releases on completion", async () => {
const transport = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "creation-reconnect",
transportFactory: () => transport.transport,
reconnect: { enabled: false },
});
clients.push(client);
const features = { creationLifecycle: true, ownedSubscriptions: true };
const connect = client.connect();
transport.triggerOpen({ features });
await connect;
const phases: string[] = [];
const creation = client.createWorkspace({
idempotencyKey: "reconnect-one",
source: { kind: "directory", path: "/project" },
onEvent: (snapshot) => phases.push(snapshot.phase),
});
void creation.catch(() => undefined);
const snapshot = {
kind: "workspace",
idempotencyKey: "reconnect-one",
phase: "accepted",
revision: 0,
workspaceId: "wks_0123456789abcdef",
agentId: null,
error: null,
};
for (const subscriptionId of ["first-connection", "second-connection"]) {
transport.triggerClose();
const reconnected = client.connect();
transport.triggerOpen({ features });
await reconnected;
await expect.poll(() => transport.sent.length).toBe(1);
const request = parseSentFrame(transport.sent[0]);
expect(request).toMatchObject({
type: "creation.subscribe.request",
idempotencyKey: "reconnect-one",
});
transport.triggerMessage(
wrapSessionMessage({
type: "creation.subscribe.response",
payload: {
requestId: request.requestId,
subscriptionId,
snapshot,
error: null,
},
}),
);
}
transport.triggerMessage(
wrapSessionMessage({
type: "workspace.create.update",
payload: {
...snapshot,
subscriptionId: "second-connection",
revision: 1,
phase: "failed",
error: "Provisioning failed",
},
}),
);
expect(await creation).toMatchObject({ error: "Provisioning failed" });
await expect.poll(() => transport.sent.length).toBe(2);
const release = parseSentFrame(transport.sent.at(-1));
expect(release).toMatchObject({
type: "subscription.release.request",
subscriptionId: "second-connection",
});
transport.triggerMessage(
wrapSessionMessage({
type: "subscription.release.response",
payload: {
requestId: release.requestId,
subscriptionId: "second-connection",
},
}),
);
expect(phases).toEqual(["accepted", "failed"]);
});
File diff suppressed because it is too large Load Diff
+144 -17
View File
@@ -1,3 +1,5 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { afterEach, expect, test, vi } from "vitest";
import { createPaseoApi, createPaseoClient } from "./index.js";
import { DaemonClient } from "./daemon-client.js";
@@ -24,6 +26,17 @@ class FakeWebSocket {
send(data: string | ArrayBuffer | Uint8Array): void {
this.sent.push(data);
const frame = JSON.parse(String(data));
if (frame.message?.type === "subscription.release.request")
this.message(
sessionMessage({
type: "subscription.release.response",
payload: {
requestId: frame.message.requestId,
subscriptionId: frame.message.subscriptionId,
},
}),
);
}
close(): void {
@@ -86,11 +99,12 @@ async function connectClient(
features: Record<string, boolean> = {
providerUsageList: true,
providersSnapshotCwd: true,
ownedSubscriptions: true,
},
): Promise<{ client: PaseoClient; ws: FakeWebSocket }> {
vi.stubGlobal("WebSocket", FakeWebSocket);
const client = createPaseoClient({
url: "ws://daemon.test",
webSocketFactory: (url) => new FakeWebSocket(url),
reconnect: { enabled: false },
});
@@ -121,6 +135,27 @@ async function connectClient(
return { client, ws };
}
function acknowledgeObservation(ws: FakeWebSocket, subscriptionId: string): void {
const request = parseSentSessionMessage(ws.sent.at(-1));
ws.message(
sessionMessage({
type: request.type.replace(/([._])request$/, "$1response"),
payload: {
requestId: request.requestId,
subscriptionId,
entries: [],
pageInfo: { nextCursor: null, prevCursor: null, hasMore: false },
},
}),
);
}
async function observeAgents(client: PaseoClient, ws: FakeWebSocket): Promise<string> {
const ready = client.agents.list({ subscribe: {} });
acknowledgeObservation(ws, "agents-sdk");
return (await ready).subscriptionId!;
}
function createWorkspace(input: Partial<PaseoWorkspace> = {}): PaseoWorkspace {
return {
id: "workspace_sdk",
@@ -231,6 +266,8 @@ test("createPaseoApi borrows daemon capabilities without exposing connection own
expect(Object.keys(paseo).sort()).toEqual([
"agents",
"config",
"dispose",
"observeEvents",
"projects",
"providers",
"terminals",
@@ -344,11 +381,17 @@ test("project actions subscribe to existing project updates", async () => {
const unsubscribe = client.projects.subscribe((update) => {
updates.push(update.kind === "upsert" ? update.project.projectDisplayName : update.projectId);
});
expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({
type: "session.events.set_subscription.request",
events: ["project.update"],
});
acknowledgeObservation(ws, "projects-sdk");
ws.message(
sessionMessage({
type: "project.update",
payload: {
subscriptionId: "projects-sdk",
kind: "upsert",
project: {
projectId: "project_sdk",
@@ -370,6 +413,7 @@ test("project actions subscribe to existing project updates", async () => {
sessionMessage({
type: "project.update",
payload: {
subscriptionId: "projects-sdk",
kind: "remove",
projectId: "project_removed",
generation: "daemon-generation",
@@ -385,6 +429,7 @@ test("project actions subscribe to existing project updates", async () => {
sessionMessage({
type: "project.update",
payload: {
subscriptionId: "projects-sdk",
kind: "remove",
projectId: "project_after_unsubscribe",
},
@@ -522,6 +567,9 @@ test("workspace handles keep identity and refresh snapshots through existing dri
await expect(refetchPromise).resolves.toEqual(refreshedWorkspace);
expect(workspace.current()).toEqual(refreshedWorkspace);
const directoryReady = client.workspaces.list({ subscribe: {} });
acknowledgeObservation(ws, "workspaces-sdk");
await directoryReady;
const updates: string[] = [];
const unsubscribe = workspace.subscribe((update) => {
if (update.kind === "upsert") {
@@ -532,10 +580,7 @@ test("workspace handles keep identity and refresh snapshots through existing dri
ws.message(
sessionMessage({
type: "workspace_update",
payload: {
kind: "upsert",
workspace: pushedWorkspace,
},
payload: { subscriptionId: "workspaces-sdk", kind: "upsert", workspace: pushedWorkspace },
}),
);
expect(updates).toEqual(["sdk pushed"]);
@@ -567,6 +612,7 @@ test("workspace handles keep identity and refresh snapshots through existing dri
sessionMessage({
type: "workspace_update",
payload: {
subscriptionId: "workspaces-sdk",
kind: "upsert",
workspace: createWorkspace({ name: "sdk after unsubscribe" }),
},
@@ -681,6 +727,7 @@ test("agent handles delegate create, send, timeline refetch, archive, and local
expect(agent.id).toBe("agent_sdk");
expect(agent.current()).toEqual(createdAgent);
await observeAgents(client, ws);
const updatedAgents: string[] = [];
const unsubscribe = agent.subscribe((update) => {
if (update.kind === "upsert") {
@@ -691,11 +738,7 @@ test("agent handles delegate create, send, timeline refetch, archive, and local
ws.message(
sessionMessage({
type: "agent_update",
payload: {
kind: "upsert",
agent: updatedAgent,
project: null,
},
payload: { subscriptionId: "agents-sdk", kind: "upsert", agent: updatedAgent, project: null },
}),
);
expect(updatedAgents).toEqual(["Updated"]);
@@ -959,11 +1002,13 @@ test("agent handles expose the observed snapshot through readonly properties", a
runtimeInfo: { provider: "codex", sessionId: "session-sdk", model: "gpt-5.4" },
archivedAt: "2026-05-16T02:00:00.000Z",
});
await observeAgents(client, ws);
const unsubscribe = agent.subscribe(() => {});
ws.message(
sessionMessage({
type: "agent_update",
payload: {
subscriptionId: "agents-sdk",
kind: "upsert",
agent: observedAgent,
project: null,
@@ -1138,6 +1183,8 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
});
const readyPromise = client.providers.waitForReady({ cwd: "/repo/sdk", timeoutMs: 5_000 });
acknowledgeObservation(ws, "provider-wait");
await new Promise((resolve) => setTimeout(resolve, 0));
const readyRequest = parseSentSessionMessage(ws.sent.at(-1));
expect(readyRequest).toMatchObject({
type: "get_providers_snapshot_request",
@@ -1158,6 +1205,7 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
sessionMessage({
type: "providers_snapshot_update",
payload: {
subscriptionId: "provider-wait",
cwd: "/repo/other",
entries: [{ provider: "codex", status: "ready", enabled: true }],
generatedAt: "2026-05-16T00:00:30.000Z",
@@ -1168,6 +1216,7 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
sessionMessage({
type: "providers_snapshot_update",
payload: {
subscriptionId: "provider-wait",
cwd: "/repo/sdk",
entries: [{ provider: "codex", status: "ready", enabled: true }],
generatedAt: "2026-05-16T00:00:01.000Z",
@@ -1184,11 +1233,14 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
cwd: "/repo/./sdk",
timeoutMs: 5_000,
});
acknowledgeObservation(ws, "provider-canonical-wait");
await new Promise((resolve) => setTimeout(resolve, 0));
const canonicalReadyRequest = parseSentSessionMessage(ws.sent.at(-1));
ws.message(
sessionMessage({
type: "providers_snapshot_update",
payload: {
subscriptionId: "provider-canonical-wait",
cwd: "/repo/sdk",
entries: [{ provider: "codex", status: "ready", enabled: true }],
generatedAt: "2026-05-16T00:00:02.000Z",
@@ -1323,10 +1375,16 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
snapshotUpdates.push(update.generatedAt);
snapshotModelDefaults.push(update.entries[0]?.models?.[0]?.defaultThinkingOptionId);
});
expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({
type: "session.events.set_subscription.request",
events: ["providers_snapshot_update"],
});
acknowledgeObservation(ws, "provider-updates");
ws.message(
sessionMessage({
type: "providers_snapshot_update",
payload: {
subscriptionId: "provider-updates",
cwd: "/repo/sdk",
entries: [
{
@@ -1354,15 +1412,25 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
await client.close();
});
test("waitForReady requires canonical provider snapshot identity from the host", async () => {
test("waitForReady reads an older host on the existing connection", async () => {
const { client, ws } = await connectClient({});
const sentBeforeWait = ws.sent.length;
await expect(client.providers.waitForReady({ cwd: "/repo/./sdk" })).rejects.toThrow(
"Update the host to wait for provider discovery.",
const waiting = client.providers.waitForReady({ cwd: "/repo/./sdk" });
await expect
.poll(() => parseSentSessionMessage(ws.sent.at(-1)).type)
.toBe("get_providers_snapshot_request");
const request = parseSentSessionMessage(ws.sent.at(-1));
ws.message(
sessionMessage({
type: "get_providers_snapshot_response",
payload: {
requestId: request.requestId,
entries: [],
generatedAt: "2026-09-11T00:00:00.000Z",
},
}),
);
expect(ws.sent).toHaveLength(sentBeforeWait);
await expect(waiting).resolves.toMatchObject({ entries: [] });
expect(FakeWebSocket.instances).toHaveLength(1);
await client.close();
});
@@ -1546,3 +1614,62 @@ test("agent config requires provider/model syntax", async () => {
await client.close();
});
test("canceled timeline handles and captured state are collectible while their API stays alive", async () => {
const source = `
import assert from "node:assert/strict";
import { createPaseoApi } from ${JSON.stringify(new URL("./index.ts", import.meta.url).href)};
import { DaemonClient } from ${JSON.stringify(new URL("./daemon-client.ts", import.meta.url).href)};
const drivers = [], apis = [];
let errors = 0;
async function batch(mode) {
const driver = new DaemonClient({ url: "ws://127.0.0.1:1/ws", clientId: mode, reconnect: { enabled: false } });
const scope = new AbortController();
const api = createPaseoApi(driver, { signal: scope.signal });
drivers.push(driver); apis.push(api);
const refs = [], ready = [];
for (let i = 0; i < 100; i++) {
const state = { bytes: new Uint8Array(64 * 1024) };
const handle = api.agents.ref("unconnected-agent").timeline.subscribe(update => {
if (update.event.type === "error") errors++;
return state.bytes[0];
});
refs.push({ handle: new WeakRef(handle), state: new WeakRef(state) });
ready.push(handle.ready.catch(() => {}));
if (mode === "callable") handle();
if (mode === "release") await handle.release();
}
if (mode === "scope-abort") scope.abort();
if (mode === "client-close") await driver.close();
if (mode !== "active") await Promise.all(ready);
return refs;
}
const groups = {};
for (const mode of ["callable", "release", "scope-abort", "client-close", "active"]) groups[mode] = await batch(mode);
async function collect() {
for (let i = 0; i < 12; i++) { await new Promise(r => setImmediate(r)); global.gc(); }
}
const count = refs => ({ handles: refs.filter(r => r.handle.deref()).length, states: refs.filter(r => r.state.deref()).length });
await collect();
const live = Object.fromEntries(Object.entries(groups).map(([mode, refs]) => [mode, count(refs)]));
console.log(JSON.stringify({ phase: "API alive", live, errors }));
for (const api of apis) await api.dispose();
await collect();
const disposed = Object.fromEntries(Object.entries(groups).map(([mode, refs]) => [mode, count(refs)]));
console.log(JSON.stringify({ phase: "API disposed", disposed }));
for (const driver of drivers) await driver.close();
for (const mode of ["callable", "release", "scope-abort", "client-close"]) assert.deepEqual(live[mode], { handles: 0, states: 0 }, mode);
assert.deepEqual(live.active, { handles: 100, states: 100 }, "active ownership is retained");
assert.equal(errors, 100, "client shutdown reports failure; explicit cancellation does not");
for (const result of Object.values(disposed)) assert.deepEqual(result, { handles: 0, states: 0 });
`;
const env = Object.fromEntries(
Object.entries(process.env).filter(([key]) => !/^(PASEO_|EXPO_|E2E_|AGENT_BROWSER_)/.test(key)),
);
const result = await promisify(execFile)(
process.execPath,
["--expose-gc", "--import", "tsx", "--input-type=module", "-e", source],
{ env, timeout: 15000 },
);
expect(result.stdout).toContain('"phase":"API alive"');
}, 20000);
+347 -137
View File
@@ -1,7 +1,10 @@
import type { OwnedSubscription } from "./connection/index.js";
export type { OwnedSubscription, SubscriptionObserver } from "./connection/index.js";
import type { DaemonClientConfig } from "./daemon-client.js";
import type { AgentPermissionResponse } from "@getpaseo/protocol/agent-types";
import type {
AgentSnapshotPayload,
CreationSnapshot,
CreateAgentRequestMessage,
FetchWorkspacesRequestMessage,
FetchWorkspacesResponseMessage,
@@ -26,7 +29,7 @@ import type {
WorkspaceDescriptorPayload,
WorkspaceCreateRequest,
} from "@getpaseo/protocol/messages";
import { DaemonClient } from "./daemon-client.js";
import { DaemonClient, type CreateAgentRequestOptions } from "./daemon-client.js";
import {
createTerminalActions,
type PaseoTerminalActions,
@@ -114,6 +117,7 @@ export type PaseoProjectUpdate = Extract<
export type PaseoProjectUpdateHandler = (update: PaseoProjectUpdate) => void;
export interface PaseoAgentListResult {
subscription?: OwnedSubscription<PaseoAgentListResult>;
requestId: string;
subscriptionId?: string | null;
entries: FetchAgentsEntry[];
@@ -127,6 +131,7 @@ export type PaseoWorkspaceListOptions = Omit<
};
export interface PaseoWorkspaceListResult {
subscription?: OwnedSubscription<PaseoWorkspaceListResult>;
requestId: string;
subscriptionId?: string | null;
entries: PaseoWorkspace[];
@@ -138,8 +143,16 @@ export interface PaseoWorkspaceOpenOptions {
requestId?: string;
}
export type PaseoWorkspaceCreateOptions = Omit<WorkspaceCreateRequest, "type" | "requestId"> & {
export type PaseoWorkspaceCreateOptions = Omit<
WorkspaceCreateRequest,
"type" | "requestId" | "agent" | "subscribe"
> & {
requestId?: string;
agent?: Omit<
PaseoAgentCreateOptions,
"worktree" | "git" | "onEvent" | "idempotencyKey" | "requestId"
>;
onEvent?: (snapshot: CreationSnapshot) => void;
};
export interface PaseoWorkspaceArchiveResult {
@@ -185,6 +198,12 @@ export interface PaseoProjectActions {
}
export interface PaseoWorkspaceActions {
list(options: PaseoWorkspaceListOptions & { subscribe: {} }): Promise<
PaseoWorkspaceListResult & {
subscriptionId: string;
subscription: OwnedSubscription<PaseoWorkspaceListResult>;
}
>;
list(options?: PaseoWorkspaceListOptions): Promise<PaseoWorkspaceListResult>;
ref(workspace: string | PaseoWorkspace): PaseoWorkspaceHandle;
open(
@@ -222,6 +241,9 @@ export interface PaseoAgentConfig {
}
export interface PaseoAgentCreateOptions {
idempotencyKey?: string;
agentId?: string;
onEvent?: (snapshot: CreationSnapshot) => void;
config: PaseoAgentConfig;
cwd: string;
parent?: string | PaseoAgentHandle;
@@ -289,7 +311,13 @@ export type PaseoAgentTimelineEvent =
| {
agentId: string;
event: { type: "replacement"; epoch: string };
};
}
| {
agentId: string;
subscriptionId: string;
event: { type: "subscription_restored" };
}
| { agentId: string; event: { type: "error"; error: string } };
export type PaseoAgentTimelineSubscription = ReturnType<DaemonClient["subscribeAgentTimeline"]>;
@@ -302,8 +330,10 @@ export interface PaseoAgentTimelineHandle {
*/
refetch(options?: PaseoAgentTimelineRefetchOptions): Promise<FetchAgentTimelinePayload>;
/**
* Subscribe to this agent and restore demand after reconnect. A replacement
* event invalidates previously fetched history; refetch the page you need.
* Delivers live events only. After reconnect, subscription_restored precedes
* subsequent updates. History may have been missed; use refetch() to request
* the range you need. No history is fetched automatically. A replacement event
* invalidates the previous epoch. Subscription errors release this observation.
* Await the returned unsubscribe function's `ready` promise before starting
* work that must be observed. It rejects if establishment fails.
*/
@@ -354,6 +384,12 @@ export interface PaseoAgentHandle {
}
export interface PaseoAgentActions {
list(options: PaseoAgentListOptions & { subscribe: {} }): Promise<
PaseoAgentListResult & {
subscriptionId: string;
subscription: OwnedSubscription<PaseoAgentListResult>;
}
>;
list(options?: PaseoAgentListOptions): Promise<PaseoAgentListResult>;
ref(agent: string | PaseoAgent): PaseoAgentHandle;
create(options: PaseoAgentCreateOptions): Promise<PaseoAgentHandle>;
@@ -450,6 +486,8 @@ export interface PaseoConfigActions {
}
export interface PaseoApi {
dispose(): Promise<void>;
observeEvents: DaemonClient["observeEvents"];
readonly terminals: PaseoTerminalActions;
readonly workspaces: PaseoWorkspaceActions;
readonly projects: PaseoProjectActions;
@@ -471,39 +509,88 @@ export function createPaseoClient(config: PaseoClientConfig): PaseoClient {
clientId: config.clientId ?? createGeneratedClientId(),
clientType: "cli",
});
const api = createPaseoApi(daemonClient);
return {
...createPaseoApi(daemonClient),
...api,
connect: () => daemonClient.connect(),
close: () => daemonClient.close(),
close: async () => {
try {
await api.dispose();
} finally {
await daemonClient.close();
}
},
ensureConnected: () => daemonClient.ensureConnected(),
getConnectionState: () => daemonClient.getConnectionState(),
};
}
export function createPaseoApi(daemonClient: DaemonClient): PaseoApi {
const createAgentHandle = createAgentHandleFactory(daemonClient);
function toDaemonAgentCreateOptions(
options: PaseoAgentCreateOptions,
placement?: { workspaceId: string; cwd: string },
): CreateAgentRequestOptions {
const { config: agentConfig, cwd, parent, title, prompt, ...requestOptions } = options;
const { provider: providerModel, options: providerOptions, ...runtimeConfig } = agentConfig;
const { provider, model } = parseProviderModel(providerModel);
return {
...requestOptions,
config: {
...runtimeConfig,
provider,
model,
cwd: placement?.cwd ?? cwd,
...(title !== undefined ? { title } : {}),
...(providerOptions !== undefined ? { providerOptions } : {}),
},
...(placement ? { workspaceId: placement.workspaceId } : {}),
...(parent ? { callerAgentId: resolveAgentId(parent) } : {}),
...(prompt !== undefined ? { initialPrompt: prompt } : {}),
};
}
export function createPaseoApi(
daemonClient: DaemonClient,
scopeOptions?: { signal?: AbortSignal },
): PaseoApi {
const handles = new Set<{ release(): Promise<void> }>();
const agentListeners = new Set<PaseoAgentUpdateHandler>();
const workspaceListeners = new Set<PaseoWorkspaceUpdateHandler>();
const lifetime = new AbortController();
const own = <T extends { release(): Promise<void> }>(create: () => T): T => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
const handle = create();
handles.add(handle);
const release = handle.release.bind(handle);
handle.release = async () => {
await release();
handles.delete(handle);
};
return handle;
};
const listenAgents = (handler: PaseoAgentUpdateHandler) => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
agentListeners.add(handler);
return () => {
agentListeners.delete(handler);
};
};
const listenWorkspaces = (handler: PaseoWorkspaceUpdateHandler) => {
if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
workspaceListeners.add(handler);
return () => {
workspaceListeners.delete(handler);
};
};
const createAgentHandle = createAgentHandleFactory(
daemonClient,
listenAgents,
(agentId, handler) => own(() => daemonClient.subscribeAgentTimeline(agentId, handler)),
);
const createAgent = async (
options: PaseoAgentCreateOptions,
placement?: { workspaceId: string; cwd: string },
) => {
const { config: agentConfig, cwd, parent, title, prompt, ...requestOptions } = options;
const { provider: providerModel, options: providerOptions, ...runtimeConfig } = agentConfig;
const { provider, model } = parseProviderModel(providerModel);
const effectiveCwd = placement?.cwd ?? cwd;
const agent = await daemonClient.createAgent({
...requestOptions,
config: {
...runtimeConfig,
provider,
model,
cwd: effectiveCwd,
...(title !== undefined ? { title } : {}),
...(providerOptions !== undefined ? { providerOptions } : {}),
},
...(placement ? { workspaceId: placement.workspaceId } : {}),
...(parent ? { callerAgentId: resolveAgentId(parent) } : {}),
...(prompt !== undefined ? { initialPrompt: prompt } : {}),
});
const agent = await daemonClient.createAgent(toDaemonAgentCreateOptions(options, placement));
return createAgentHandle(agent);
};
const terminals = createTerminalActions(daemonClient, async (workspaceId) => {
@@ -513,24 +600,123 @@ export function createPaseoApi(daemonClient: DaemonClient): PaseoApi {
}
return workspace.workspaceDirectory;
});
const createWorkspaceHandle = createWorkspaceHandleFactory(daemonClient, createAgent, terminals);
const createWorkspaceHandle = createWorkspaceHandleFactory(
daemonClient,
createAgent,
terminals,
listenWorkspaces,
);
let disposal: Promise<void> | null = null;
const dispose = (): Promise<void> => {
if (disposal) return disposal;
lifetime.abort();
scopeOptions?.signal?.removeEventListener("abort", abort);
agentListeners.clear();
workspaceListeners.clear();
disposal = Promise.allSettled([...handles].map((handle) => handle.release())).then(
(results) => {
handles.clear();
const failures = results.flatMap((result) =>
result.status === "rejected" ? [result.reason] : [],
);
if (failures.length)
throw new AggregateError(failures, "Failed to release API subscriptions");
return undefined;
},
);
return disposal;
};
const abort = () => {
void dispose().catch((error) => console.error("API subscription cleanup failed", error));
};
if (scopeOptions?.signal?.aborted) abort();
else scopeOptions?.signal?.addEventListener("abort", abort, { once: true });
const observeEvents: DaemonClient["observeEvents"] = (events, options) =>
own(() => daemonClient.observeEvents(events, options));
const subscribeEvent = (
event: "project.update" | "providers_snapshot_update",
update: (message: SessionOutboundMessage) => void,
): (() => void) => {
const observation = observeEvents([event]);
observation.subscribe({ snapshot: () => {}, update });
return () => {
void observation
.release()
.catch((error) => console.error("Event subscription cleanup failed", error));
};
};
function listWorkspaces(options: PaseoWorkspaceListOptions & { subscribe: {} }): Promise<
PaseoWorkspaceListResult & {
subscriptionId: string;
subscription: OwnedSubscription<PaseoWorkspaceListResult>;
}
>;
function listWorkspaces(options?: PaseoWorkspaceListOptions): Promise<PaseoWorkspaceListResult>;
async function listWorkspaces(
options?: PaseoWorkspaceListOptions,
): Promise<PaseoWorkspaceListResult> {
if (!options?.subscribe) return daemonClient.fetchWorkspaces(options);
if (options.subscribe.subscriptionId !== undefined)
throw new Error("Subscription IDs are assigned by the host");
const subscription = own(() => daemonClient.observeWorkspaces(options));
subscription.subscribe({
snapshot: () => {},
update: (message) => {
if (message.type === "workspace_update")
for (const listener of workspaceListeners) listener(message.payload);
},
});
return { ...(await subscription.ready), subscription };
}
function listAgents(options: PaseoAgentListOptions & { subscribe: {} }): Promise<
PaseoAgentListResult & {
subscriptionId: string;
subscription: OwnedSubscription<PaseoAgentListResult>;
}
>;
function listAgents(options?: PaseoAgentListOptions): Promise<PaseoAgentListResult>;
async function listAgents(options?: PaseoAgentListOptions): Promise<PaseoAgentListResult> {
if (!options?.subscribe) return daemonClient.fetchAgents(options);
if (options.subscribe.subscriptionId !== undefined)
throw new Error("Subscription IDs are assigned by the host");
const subscription = own(() => daemonClient.observeAgents(options));
subscription.subscribe({
snapshot: () => {},
update: (message) => {
if (message.type === "agent_update")
for (const listener of agentListeners) listener(message.payload);
},
});
return { ...(await subscription.ready), subscription };
}
return {
dispose,
observeEvents,
terminals,
projects: {
list: (options) => daemonClient.listProjects(options),
subscribe: (handler) =>
daemonClient.on("project.update", (message) => {
handler(message.payload);
}),
subscribe: (handler) => {
return subscribeEvent("project.update", (message) => {
if (message.type === "project.update") handler(message.payload);
});
},
},
workspaces: {
list: (options) => daemonClient.fetchWorkspaces(options),
list: listWorkspaces,
ref: (workspace) => createWorkspaceHandle(workspace),
open: (input, requestId) =>
openWorkspace(daemonClient, createWorkspaceHandle, input, requestId),
create: async ({ requestId, ...options }) => {
const result = await daemonClient.createWorkspace(options, requestId);
create: async ({ requestId, agent, ...options }) => {
const result = await daemonClient.createWorkspace(
{ ...options, ...(agent ? { agent: toDaemonAgentCreateOptions(agent) } : {}) },
requestId,
);
if (result.error || !result.workspace) {
throw new Error(result.error ?? "The daemon did not create a workspace");
}
@@ -538,19 +724,13 @@ export function createPaseoApi(daemonClient: DaemonClient): PaseoApi {
},
archive: (workspace, requestId) =>
daemonClient.archiveWorkspace(resolveWorkspaceId(workspace), requestId),
subscribe: (handler) =>
daemonClient.on("workspace_update", (message) => {
handler(message.payload);
}),
subscribe: listenWorkspaces,
},
agents: {
list: (options) => daemonClient.fetchAgents(options),
list: listAgents,
ref: (agent) => createAgentHandle(agent),
create: (options) => createAgent(options),
subscribe: (handler) =>
daemonClient.on("agent_update", (message) => {
handler(message.payload);
}),
subscribe: listenAgents,
},
providers: {
listModels: (provider, options) => daemonClient.listProviderModels(provider, options),
@@ -561,14 +741,21 @@ export function createPaseoApi(daemonClient: DaemonClient): PaseoApi {
},
listAvailable: (options) => daemonClient.listAvailableProviders(options),
snapshot: (options) => daemonClient.getProvidersSnapshot(options),
waitForReady: (options) => waitForProvidersReady(daemonClient, options),
waitForReady: (options) =>
waitForProvidersReady(
daemonClient,
observeEvents(["providers_snapshot_update"]),
lifetime.signal,
options,
),
refresh: (options) => daemonClient.refreshProvidersSnapshot(options),
diagnostic: (provider, options) => daemonClient.getProviderDiagnostic(provider, options),
listUsage: (options) => listProviderUsage(daemonClient, options),
subscribe: (handler) =>
daemonClient.on("providers_snapshot_update", (message) => {
handler(message.payload);
}),
subscribe: (handler) => {
return subscribeEvent("providers_snapshot_update", (message) => {
if (message.type === "providers_snapshot_update") handler(message.payload);
});
},
},
config: {
get: (requestId) => daemonClient.getDaemonConfig(requestId),
@@ -588,6 +775,7 @@ function createWorkspaceHandleFactory(
daemonClient: DaemonClient,
createAgent: CreateAgent,
terminals: PaseoTerminalActions,
listen: (handler: PaseoWorkspaceUpdateHandler) => () => void,
): WorkspaceHandleFactory {
return (workspace) => {
const id = typeof workspace === "string" ? workspace : workspace.id;
@@ -654,14 +842,12 @@ function createWorkspaceHandleFactory(
return result;
},
subscribe: (handler) =>
daemonClient.on("workspace_update", (message) => {
const update = message.payload;
listen((update) => {
if (update.kind === "upsert" && update.workspace.id === id) {
current = update.workspace;
handler(update);
}
if (update.kind === "remove" && update.id === id) {
current = null;
handler(update);
}
}),
@@ -669,7 +855,11 @@ function createWorkspaceHandleFactory(
};
}
function createAgentHandleFactory(daemonClient: DaemonClient): AgentHandleFactory {
function createAgentHandleFactory(
daemonClient: DaemonClient,
listen: (handler: PaseoAgentUpdateHandler) => () => void,
subscribeTimeline: DaemonClient["subscribeAgentTimeline"],
): AgentHandleFactory {
return (agent) => {
const id = typeof agent === "string" ? agent : agent.id;
let current = typeof agent === "string" ? null : agent;
@@ -686,15 +876,27 @@ function createAgentHandleFactory(daemonClient: DaemonClient): AgentHandleFactor
return result;
},
subscribe: (handler) =>
daemonClient.subscribeAgentTimeline(id, (message) => {
handler(
message.type === "agent_stream"
? message.payload
: {
agentId: message.payload.agentId,
event: { type: "replacement", epoch: message.payload.epoch },
},
);
subscribeTimeline(id, (message) => {
switch (message.type) {
case "agent_stream":
return handler(message.payload);
case "agent.timeline.subscription_restored":
return handler({
agentId: id,
subscriptionId: message.payload.subscriptionId,
event: { type: "subscription_restored" },
});
case "agent.timeline.error":
return handler({
agentId: id,
event: { type: "error", error: message.payload.error },
});
case "agent.timeline.replacement":
return handler({
agentId: id,
event: { type: "replacement", epoch: message.payload.epoch },
});
}
}),
},
get workspaceId() {
@@ -779,14 +981,12 @@ function createAgentHandleFactory(daemonClient: DaemonClient): AgentHandleFactor
await daemonClient.detachAgent(id);
},
subscribe: (handler) =>
daemonClient.on("agent_update", (message) => {
const update = message.payload;
listen((update) => {
if (update.kind === "upsert" && update.agent.id === id) {
current = update.agent;
handler(update);
}
if (update.kind === "remove" && update.agentId === id) {
current = null;
handler(update);
}
}),
@@ -840,86 +1040,96 @@ function listProviderUsage(
return daemonClient.listProviderUsage(options);
}
function waitForProvidersReady(
async function waitForProvidersReady(
daemonClient: DaemonClient,
observation: ReturnType<DaemonClient["observeEvents"]>,
signal: AbortSignal,
options: PaseoProviderWaitOptions = {},
): Promise<PaseoProviderSnapshotResult> {
// COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10.
if (daemonClient.getLastServerInfoMessage()?.features?.providersSnapshotCwd !== true) {
return Promise.reject(new Error("Update the host to wait for provider discovery."));
}
const { timeoutMs = 60_000, ...snapshotOptions } = options;
return new Promise((resolve, reject) => {
let settled = false;
let requestId: string | null = null;
let snapshotCwd: string | undefined;
const pendingUpdates = new Map<string | undefined, PaseoProviderSnapshotUpdate>();
let latestEntries: PaseoProviderSnapshotResult["entries"] = [];
try {
await observation.ready;
signal.throwIfAborted();
return await new Promise<PaseoProviderSnapshotResult>((resolve, reject) => {
let settled = false;
let requestId: string | null = null;
let snapshotCwd: string | undefined;
const pendingUpdates = new Map<string | undefined, PaseoProviderSnapshotUpdate>();
let latestEntries: PaseoProviderSnapshotResult["entries"] = [];
const cleanup = () => {
clearTimeout(timeout);
unsubscribe();
};
const finish = (snapshot: PaseoProviderSnapshotResult) => {
if (settled) return;
settled = true;
cleanup();
resolve(snapshot);
};
const fail = (error: unknown) => {
if (settled) return;
settled = true;
cleanup();
reject(error instanceof Error ? error : new Error(String(error)));
};
const updateMatches = (update: PaseoProviderSnapshotUpdate) => update.cwd === snapshotCwd;
const cleanup = () => {
clearTimeout(timeout);
unsubscribe();
signal.removeEventListener("abort", abort);
};
const finish = (snapshot: PaseoProviderSnapshotResult) => {
if (settled) return;
settled = true;
cleanup();
resolve(snapshot);
};
const fail = (error: unknown) => {
if (settled) return;
settled = true;
cleanup();
reject(error instanceof Error ? error : new Error(String(error)));
};
const updateMatches = (update: PaseoProviderSnapshotUpdate) => update.cwd === snapshotCwd;
const unsubscribe = daemonClient.on("providers_snapshot_update", (message) => {
const update = message.payload;
if (!requestId) {
pendingUpdates.set(update.cwd, update);
return;
}
if (!updateMatches(update)) return;
latestEntries = update.entries;
if (update.entries.some((entry) => entry.status === "loading")) return;
finish({ ...update, requestId });
const unsubscribe = observation.subscribe({
snapshot: () => {},
update: (message) => {
if (message.type !== "providers_snapshot_update") return;
const update = message.payload;
if (!requestId) {
pendingUpdates.set(update.cwd, update);
return;
}
if (!updateMatches(update)) return;
latestEntries = update.entries;
if (update.entries.some((entry) => entry.status === "loading")) return;
finish({ ...update, requestId });
},
});
const abort = () => fail(new Error("Paseo API is disposed"));
signal.addEventListener("abort", abort, { once: true });
const timeout = setTimeout(() => {
const loading = latestEntries
.filter((entry) => entry.status === "loading")
.map((entry) => entry.provider)
.join(", ");
fail(
new Error(
loading
? `Timed out waiting for providers: ${loading}`
: "Timed out waiting for provider discovery",
),
);
}, timeoutMs);
void daemonClient
.getProvidersSnapshot(snapshotOptions)
.then((snapshot) => {
requestId = snapshot.requestId;
snapshotCwd = snapshot.cwd;
latestEntries = snapshot.entries;
if (!snapshot.entries.some((entry) => entry.status === "loading")) {
finish(snapshot);
return;
}
const pendingUpdate = pendingUpdates.get(snapshotCwd);
if (pendingUpdate && !pendingUpdate.entries.some((entry) => entry.status === "loading")) {
finish({ ...pendingUpdate, requestId });
}
return undefined;
})
.catch(fail);
});
const timeout = setTimeout(() => {
const loading = latestEntries
.filter((entry) => entry.status === "loading")
.map((entry) => entry.provider)
.join(", ");
fail(
new Error(
loading
? `Timed out waiting for providers: ${loading}`
: "Timed out waiting for provider discovery",
),
);
}, timeoutMs);
void daemonClient
.getProvidersSnapshot(snapshotOptions)
.then((snapshot) => {
requestId = snapshot.requestId;
snapshotCwd = snapshot.cwd;
latestEntries = snapshot.entries;
if (!snapshot.entries.some((entry) => entry.status === "loading")) {
finish(snapshot);
return;
}
const pendingUpdate = pendingUpdates.get(snapshotCwd);
if (pendingUpdate && !pendingUpdate.entries.some((entry) => entry.status === "loading")) {
finish({ ...pendingUpdate, requestId });
}
return undefined;
})
.catch(fail);
});
} finally {
await observation.release();
}
}
function createGeneratedClientId(): string {
+21 -28
View File
@@ -13,11 +13,11 @@ interface PendingSnapshot {
/** Resolve announcements only while observed, coalescing changes during a fetch. */
export class ProviderSnapshotUpdates {
private pending = new Map<string | undefined, PendingSnapshot>();
private paused = false;
private pending = new Map<string, PendingSnapshot>();
constructor(
private readonly host: {
active(update: Update): boolean;
fetch(cwd: string | undefined): Promise<Snapshot>;
emit(update: Update): void;
failed(error: unknown): void;
@@ -28,35 +28,25 @@ export class ProviderSnapshotUpdates {
this.pending.clear();
}
pause(): void {
this.paused = true;
// Invalidate old responses while retaining the announcements they would resolve.
for (const [cwd, pending] of this.pending) {
this.pending.set(cwd, { latest: pending.latest, running: false });
}
}
resume(): void {
this.paused = false;
for (const pending of this.pending.values()) this.receive(pending.latest);
}
receive(update: Update): void {
const cwd = update.payload.cwd;
const pending = this.pending.get(cwd) ?? { latest: update, running: false };
if (!this.host.active(update)) return;
const key = JSON.stringify([update.payload.subscriptionId, update.payload.cwd]);
const pending = this.pending.get(key) ?? { latest: update, running: false };
pending.latest = update;
this.pending.set(cwd, pending);
if (this.paused || pending.running) return;
this.pending.set(key, pending);
if (pending.running) return;
pending.running = true;
void this.resolve(cwd, pending);
void this.resolve(key, pending);
}
private async resolve(cwd: string | undefined, pending: PendingSnapshot): Promise<void> {
private async resolve(key: string, pending: PendingSnapshot): Promise<void> {
try {
while (this.pending.get(cwd) === pending) {
while (this.pending.get(key) === pending) {
const announced = pending.latest;
const snapshot = await this.host.fetch(cwd);
if (this.pending.get(cwd) !== pending) return;
if (!this.host.active(announced)) return;
const snapshot = await this.host.fetch(announced.payload.cwd);
if (!this.host.active(announced)) return;
if (this.pending.get(key) !== pending) return;
// A change arriving while the body was in flight needs the current body.
// If that response already covers it, no second fetch is necessary.
if (
@@ -64,14 +54,17 @@ export class ProviderSnapshotUpdates {
pending.latest.payload.snapshotHash !== snapshot.snapshotHash
)
continue;
this.pending.delete(cwd);
this.host.emit({ type: "providers_snapshot_update", payload: snapshot });
this.pending.delete(key);
this.host.emit({
type: "providers_snapshot_update",
payload: { ...snapshot, subscriptionId: announced.payload.subscriptionId },
});
return;
}
} catch (error) {
if (this.pending.get(cwd) === pending) this.host.failed(error);
if (this.pending.get(key) === pending) this.host.failed(error);
} finally {
if (this.pending.get(cwd) === pending) this.pending.delete(cwd);
if (this.pending.get(key) === pending) this.pending.delete(key);
}
}
}
@@ -9,8 +9,7 @@ describe("terminal-stream-router", () => {
const events: TerminalStreamEvent[] = [];
const payload = new TextEncoder().encode("restored screen");
router.setSlot("term-1", 7);
router.onEvent((event) => events.push(event));
router.attach("subscription-1", "term-1", 7, (event) => events.push(event));
router.handleFrame({
opcode: TerminalStreamOpcode.Restore,
slot: 7,
@@ -20,9 +19,45 @@ describe("terminal-stream-router", () => {
expect(events).toEqual([
{
terminalId: "term-1",
subscriptionId: "subscription-1",
type: "restore",
data: payload,
},
]);
});
});
test("two registrations for one terminal retain independent slots and release", () => {
const router = new TerminalStreamRouter();
const a: TerminalStreamEvent[] = [];
const b: TerminalStreamEvent[] = [];
const releaseA = router.attach("subscription-a", "same-terminal", 1, (event) => a.push(event));
router.attach("subscription-b", "same-terminal", 2, (event) => b.push(event));
const payload = new TextEncoder().encode("output");
router.handleFrame({ opcode: TerminalStreamOpcode.Output, slot: 1, payload });
router.handleFrame({ opcode: TerminalStreamOpcode.Output, slot: 2, payload });
expect(a).toEqual([expect.objectContaining({ subscriptionId: "subscription-a" })]);
expect(b).toEqual([expect.objectContaining({ subscriptionId: "subscription-b" })]);
releaseA();
router.handleFrame({ opcode: TerminalStreamOpcode.Output, slot: 1, payload });
router.handleFrame({ opcode: TerminalStreamOpcode.Output, slot: 2, payload });
expect(a).toHaveLength(1);
expect(b).toHaveLength(2);
});
test("legacy listeners share a terminal slot without releasing each other", () => {
const router = new TerminalStreamRouter();
const received: string[] = [];
const release = router.attach("first", "terminal", 1, () => received.push("first"));
router.attach("second", "terminal", 1, () => received.push("second"));
const frame = {
opcode: TerminalStreamOpcode.Output,
slot: 1,
payload: new TextEncoder().encode("bytes"),
};
router.handleFrame(frame);
expect(received).toEqual(["first", "second"]);
release();
router.handleFrame(frame);
expect(received).toEqual(["first", "second", "second"]);
});
+54 -106
View File
@@ -1,22 +1,28 @@
import {
decodeTerminalSnapshotPayload,
encodeTerminalResizePayload,
encodeTerminalStreamFrame,
TerminalStreamOpcode,
type TerminalStreamFrame,
} from "@getpaseo/protocol/binary-frames/index";
import type { TerminalInput, TerminalState } from "@getpaseo/protocol/messages";
import type { TerminalState } from "@getpaseo/protocol/messages";
export type TerminalStreamEvent =
| { terminalId: string; type: "output"; data: Uint8Array }
| { terminalId: string; type: "snapshot"; state: TerminalState }
| { terminalId: string; type: "restore"; data: Uint8Array };
export type TerminalStreamEvent = { terminalId: string; subscriptionId: string } & (
| { type: "output"; data: Uint8Array }
| { type: "snapshot"; state: TerminalState }
| { type: "restore"; data: Uint8Array }
);
export class TerminalStreamRouter {
private readonly terminalSlots = new Map<string, number>();
private readonly slotTerminals = new Map<number, string>();
private readonly slots = new Map<
number,
Set<{
subscriptionId: string;
terminalId: string;
receive: (event: TerminalStreamEvent) => void;
}>
>();
private readonly listeners = new Set<(event: TerminalStreamEvent) => void>();
/** Passive wire diagnostics. App state consumes the registration's callback. */
onEvent(handler: (event: TerminalStreamEvent) => void): () => void {
this.listeners.add(handler);
return () => {
@@ -24,109 +30,51 @@ export class TerminalStreamRouter {
};
}
setSlot(terminalId: string, slot: number): void {
const existingTerminalId = this.slotTerminals.get(slot);
if (existingTerminalId && existingTerminalId !== terminalId) {
this.terminalSlots.delete(existingTerminalId);
}
const existingSlot = this.terminalSlots.get(terminalId);
if (typeof existingSlot === "number" && existingSlot !== slot) {
this.slotTerminals.delete(existingSlot);
}
this.terminalSlots.set(terminalId, slot);
this.slotTerminals.set(slot, terminalId);
}
removeTerminal(terminalId: string): void {
const slot = this.terminalSlots.get(terminalId);
if (typeof slot !== "number") {
return;
}
this.terminalSlots.delete(terminalId);
if (this.slotTerminals.get(slot) === terminalId) {
this.slotTerminals.delete(slot);
}
attach(
subscriptionId: string,
terminalId: string,
slot: number,
receive: (event: TerminalStreamEvent) => void,
): () => void {
const registration = { subscriptionId, terminalId, receive };
const registrations = this.slots.get(slot) ?? new Set();
registrations.add(registration);
this.slots.set(slot, registrations);
return () => {
registrations.delete(registration);
if (registrations.size === 0) this.slots.delete(slot);
};
}
clearSlots(): void {
this.terminalSlots.clear();
this.slotTerminals.clear();
}
encodeInput(terminalId: string, message: TerminalInput["message"]): Uint8Array | null {
const slot = this.terminalSlots.get(terminalId);
if (typeof slot !== "number") {
return null;
}
if (message.type === "input") {
return encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Input,
slot,
payload: message.data,
});
}
if (message.type === "resize") {
return encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Resize,
slot,
payload: encodeTerminalResizePayload({
rows: message.rows,
cols: message.cols,
...(message.intent ? { intent: message.intent } : {}),
}),
});
}
return null;
this.slots.clear();
}
handleFrame(frame: TerminalStreamFrame): void {
const terminalId = this.slotTerminals.get(frame.slot);
if (!terminalId) {
return;
}
if (frame.opcode === TerminalStreamOpcode.Output) {
this.emit({
terminalId,
type: "output",
data: frame.payload,
});
return;
}
if (frame.opcode === TerminalStreamOpcode.Restore) {
this.emit({
terminalId,
type: "restore",
data: frame.payload,
});
return;
}
if (frame.opcode === TerminalStreamOpcode.Snapshot) {
const state = decodeTerminalSnapshotPayload(frame.payload);
if (!state) {
return;
}
this.emit({
terminalId,
type: "snapshot",
state,
});
}
}
private emit(event: TerminalStreamEvent): void {
for (const listener of this.listeners) {
try {
listener(event);
} catch {
// no-op
const registrations = this.slots.get(frame.slot);
if (!registrations) return;
for (const registration of registrations) {
const identity = {
subscriptionId: registration.subscriptionId,
terminalId: registration.terminalId,
};
let event: TerminalStreamEvent;
if (frame.opcode === TerminalStreamOpcode.Output)
event = { ...identity, type: "output", data: frame.payload };
else if (frame.opcode === TerminalStreamOpcode.Restore)
event = { ...identity, type: "restore", data: frame.payload };
else if (frame.opcode === TerminalStreamOpcode.Snapshot) {
const state = decodeTerminalSnapshotPayload(frame.payload);
if (!state) return;
event = { ...identity, type: "snapshot", state };
} else return;
registration.receive(event);
for (const listener of this.listeners) {
try {
listener(event);
} catch {
/* Passive diagnostics cannot interrupt delivery. */
}
}
}
}
@@ -11,6 +11,8 @@ test("an older host rejects terminal operations before sending requests", async
getLastServerInfoMessage: () => ({
status: "server_info",
serverId: "older-host",
hostname: null,
version: null,
features: {},
}),
createTerminal: unexpectedRequest,
@@ -0,0 +1,75 @@
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
import type { OwnedSubscription, TimelineSubscription } from "../connection/index.js";
type TimelineUpdate = Extract<
SessionOutboundMessage,
{ type: "agent_stream" | "agent.timeline.replacement" }
>;
/** Local SDK lifecycle messages; these are not additional wire RPCs. */
export type TimelineMessage =
| TimelineUpdate
| {
type: "agent.timeline.subscription_restored";
payload: { agentId: string; subscriptionId: string };
}
| { type: "agent.timeline.error"; payload: { agentId: string; error: string } };
/** Restores live delivery; the consumer owns history reads and gap recovery. */
export function subscribeTimeline(
agentId: string,
observation: OwnedSubscription<{ agentIds: string[] }>,
handler: (message: TimelineMessage) => void,
reportError: (error: unknown) => void,
): TimelineSubscription {
let established = false;
let released = false;
const notify = (message: TimelineMessage) => {
if (released) return;
try {
handler(message);
} catch (error) {
reportError(error);
}
};
const release = () => {
released = true;
return observation.release();
};
const stop = () => {
// Use the public release entry point so the owning API also forgets this handle.
void subscription.release().catch(reportError);
};
const fail = (error: unknown) => {
notify({
type: "agent.timeline.error",
payload: { agentId, error: error instanceof Error ? error.message : String(error) },
});
stop();
};
observation.subscribe({
snapshot: ({ subscriptionId }) => {
if (!established) {
established = true;
return;
}
notify({
type: "agent.timeline.subscription_restored",
payload: { agentId, subscriptionId },
});
},
update: (message) => {
if (message.type === "agent_stream" || message.type === "agent.timeline.replacement")
notify(message);
},
error: fail,
});
const subscription: TimelineSubscription = Object.assign(stop, {
ready: observation.ready.then(() => undefined),
release,
subscriptionId: null,
});
Object.defineProperty(subscription, "subscriptionId", { get: () => observation.subscriptionId });
void subscription.ready.catch(fail);
return subscription;
}