Files
paseo/packages/client/src/daemon-client.ts
Mohamed Boudra ea2f768d3f Keep connections alive through brief daemon slowdowns (#1581)
* Keep connections alive through brief daemon slowdowns

The active connection's liveness check was a side effect of the
candidate-probe loop: a single 5s latency measurement doubled as the
death timeout, so a daemon that was alive but briefly stalled (e.g. a
busy event loop) was misread as dead and the connection was torn down.

Give DaemonClient its own self-scheduling heartbeat that solely owns
liveness (15s timeout, 2 consecutive misses), and make the probe
read-only on the active connection — it reads the heartbeat's last RTT
instead of pinging. Split the shared ping into a pure measurement path
(measureLatency) and a private liveness path, so nothing outside the
client can drive teardown.

* Don't let a candidate latency timeout count as a liveness failure

The heartbeat (livenessPing) and candidate ranking (measureLatency) share
one in-flight ping slot. If a heartbeat tick lands while a candidate
measurement is still in flight, the heartbeat dedupes onto it — and a
measurement timeout was then recorded as a liveness failure, nudging the
connection one step closer to a spurious teardown.

Tag each ping with whether its timeout may drive a liveness failure
(only the heartbeat sets it) and gate recordLivenessFailure on that flag,
so an adopted measurement probe can never contribute to teardown.
2026-06-18 21:10:08 +08:00

5125 lines
155 KiB
TypeScript

import type { z } from "zod";
import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities";
import {
AgentCreateFailedStatusPayloadSchema,
AgentCreatedStatusPayloadSchema,
AgentRefreshedStatusPayloadSchema,
AgentResumedStatusPayloadSchema,
CheckoutRenameBranchResponseSchema,
parseServerInfoStatusPayload,
RenameTerminalResponseSchema,
RestartRequestedStatusPayloadSchema,
ShutdownRequestedStatusPayloadSchema,
SessionInboundMessageSchema,
type ServerInfoStatusPayload,
WSOutboundMessageSchema,
} from "@getpaseo/protocol/messages";
import type {
AgentStreamEventPayload,
AgentSnapshotPayload,
ProjectPlacementPayload,
AgentPermissionResolvedMessage,
CreateAgentRequestMessage,
CreatePaseoWorktreeRequest,
FileDownloadTokenResponse,
FileUploadResponse,
FileExplorerResponse,
FetchAgentTimelineResponseMessage,
GitSetupOptions,
CheckoutStatusResponse,
CheckoutCommitResponse,
CheckoutMergeResponse,
CheckoutMergeFromBaseResponse,
CheckoutPullResponse,
CheckoutPushResponse,
CheckoutRefreshResponse,
CheckoutPrCreateResponse,
CheckoutPrMergeResponse,
CheckoutPrMergeMethod,
CheckoutGithubSetAutoMergeResponse,
CheckoutGithubGetCheckDetailsResponse,
CheckoutPrStatusResponse,
PullRequestTimelineResponse,
CheckoutSwitchBranchResponse,
StashSaveResponse,
StashPopResponse,
StashListResponse,
ValidateBranchResponse,
BranchSuggestionsResponse,
GitHubSearchResponse,
GitHubSearchRequest,
DirectorySuggestionsResponse,
PaseoWorktreeListResponse,
PaseoWorktreeArchiveResponse,
ProjectIconResponse,
OpenProjectResponseMessage,
ArchiveWorkspaceResponseMessage,
WorkspaceSetupStatusResponseMessage,
ListCommandsResponse,
ListProviderFeaturesResponseMessage,
ListProviderModelsResponseMessage,
ListProviderModesResponseMessage,
ListAvailableProvidersResponse,
GetProvidersSnapshotResponseMessage,
RefreshProvidersSnapshotResponseMessage,
ProviderDiagnosticResponseMessage,
DaemonGetStatusResponse,
DaemonGetPairingOfferResponse,
AgentRewindResponseMessage,
ListTerminalsResponse,
CreateTerminalResponse,
SubscribeTerminalResponse,
SubscribeTerminalRequest,
CloseItemsResponse,
KillTerminalResponse,
CaptureTerminalResponse,
TerminalInput,
SessionInboundMessage,
SessionOutboundMessage,
SendAgentMessageRequest,
PaseoConfigRaw,
PaseoConfigRevision,
WorkspaceCreateRequest,
} from "@getpaseo/protocol/messages";
import type {
AgentPermissionRequest,
AgentPermissionResponse,
AgentPersistenceHandle,
AgentProvider,
AgentSessionConfig,
} from "@getpaseo/protocol/agent-types";
import type { MutableDaemonConfig, MutableDaemonConfigPatch } from "@getpaseo/protocol/messages";
import { isRelayClientWebSocketUrl } from "@getpaseo/protocol/daemon-endpoints";
import { terminalSubscriptionKey } from "@getpaseo/protocol/terminal-subscription-key";
import {
asUint8Array,
decodeFileTransferFrame,
encodeFileTransferFrame,
decodeTerminalStreamFrame,
FileTransferOpcode,
TerminalStreamOpcode,
type FileTransferFrame,
} from "@getpaseo/protocol/binary-frames/index";
import {
createRelayE2eeTransportFactory,
createWebSocketTransportFactory,
decodeMessageData,
defaultWebSocketFactory,
describeTransportClose,
describeTransportError,
type DaemonTransport,
type DaemonTransportFactory,
type WebSocketFactory,
} from "./daemon-client-transport.js";
import { DaemonClientRuntimeMetrics } from "./daemon-client-runtime-metrics.js";
import { TerminalStreamRouter, type TerminalStreamEvent } from "./terminal-stream-router.js";
export interface Logger {
debug(obj: object, msg?: string): void;
info(obj: object, msg?: string): void;
warn(obj: object, msg?: string): void;
error(obj: object, msg?: string): void;
}
const consoleLogger: Logger = {
debug: () => {},
info: (obj, msg) => console.log(msg, obj),
warn: (obj, msg) => console.warn(msg, obj),
error: (obj, msg) => console.error(msg, obj),
};
const perfNow: () => number =
typeof performance !== "undefined" && typeof performance.now === "function"
? () => performance.now()
: () => Date.now();
interface ImportAgentInputBase {
cwd?: string;
labels?: Record<string, string>;
}
export type ImportAgentInput =
| (ImportAgentInputBase & {
providerId: string;
providerHandleId: string;
})
| (ImportAgentInputBase & {
provider: AgentProvider;
sessionId: string;
});
function normalizePassword(value: string | undefined): string | null {
if (typeof value !== "string") {
return null;
}
return value.length > 0 ? value : null;
}
export type {
DaemonTransport,
DaemonTransportFactory,
WebSocketFactory,
WebSocketLike,
} from "./daemon-client-transport.js";
export type { TerminalStreamEvent };
export type ConnectionState =
| { status: "idle" }
| { status: "connecting"; attempt: number }
| { status: "connected" }
| { status: "disconnected"; reason?: string }
| { status: "disposed" };
export type DaemonEvent =
| {
type: "agent_update";
agentId: string;
payload: Extract<SessionOutboundMessage, { type: "agent_update" }>["payload"];
}
| {
type: "workspace_update";
workspaceId: string;
payload: Extract<SessionOutboundMessage, { type: "workspace_update" }>["payload"];
}
| {
type: "workspace_setup_progress";
workspaceId: string;
payload: Extract<SessionOutboundMessage, { type: "workspace_setup_progress" }>["payload"];
}
| {
type: "agent_stream";
agentId: string;
event: AgentStreamEventPayload;
timestamp: string;
seq?: number;
epoch?: string;
}
| { type: "status"; payload: { status: string } & Record<string, unknown> }
| { type: "agent_deleted"; agentId: string }
| {
type: "agent_permission_request";
agentId: string;
request: AgentPermissionRequest;
}
| {
type: "agent_permission_resolved";
agentId: string;
requestId: string;
resolution: AgentPermissionResponse;
}
| {
type: "providers_snapshot_update";
payload: Extract<SessionOutboundMessage, { type: "providers_snapshot_update" }>["payload"];
}
| { type: "error"; message: string };
export type DaemonEventHandler = (event: DaemonEvent) => void;
export interface DaemonClientConfig {
url: string;
clientId: string;
clientType?: "mobile" | "browser" | "cli" | "mcp";
appVersion?: string;
runtimeGeneration?: number | null;
password?: string;
authHeader?: string;
suppressSendErrors?: boolean;
transportFactory?: DaemonTransportFactory;
webSocketFactory?: WebSocketFactory;
logger?: Logger;
connectTimeoutMs?: number;
e2ee?: {
enabled?: boolean;
daemonPublicKeyB64?: string;
};
reconnect?: {
enabled?: boolean;
baseDelayMs?: number;
maxDelayMs?: number;
};
runtimeMetricsIntervalMs?: number;
runtimeMetricsWindowMs?: number;
}
export interface SendMessageOptions {
messageId?: string;
images?: Array<{ data: string; mimeType: string }>;
attachments?: SendAgentMessageRequest["attachments"];
}
type AgentConfigOverrides = Partial<Omit<AgentSessionConfig, "provider" | "cwd">>;
export interface CreateAgentRequestOptions extends AgentConfigOverrides {
config?: AgentSessionConfig;
provider?: AgentProvider;
cwd?: string;
env?: CreateAgentRequestMessage["env"];
workspaceId?: string;
initialPrompt?: string;
clientMessageId?: string;
outputSchema?: Record<string, unknown>;
images?: CreateAgentRequestMessage["images"];
attachments?: CreateAgentRequestMessage["attachments"];
git?: GitSetupOptions;
worktree?: CreateAgentRequestMessage["worktree"];
autoArchive?: CreateAgentRequestMessage["autoArchive"];
worktreeName?: string;
requestId?: string;
labels?: Record<string, string>;
}
export interface CreatePaseoWorktreeInput extends Pick<
CreatePaseoWorktreeRequest,
| "cwd"
| "projectId"
| "worktreeSlug"
| "firstAgentContext"
| "refName"
| "action"
| "githubPrNumber"
> {}
type CheckoutStatusPayload = CheckoutStatusResponse["payload"];
type SubscribeCheckoutDiffPayload = Extract<
SessionOutboundMessage,
{ type: "subscribe_checkout_diff_response" }
>["payload"];
type CheckoutDiffPayload = Omit<SubscribeCheckoutDiffPayload, "subscriptionId">;
type CheckoutCommitPayload = CheckoutCommitResponse["payload"];
type CheckoutMergePayload = CheckoutMergeResponse["payload"];
type CheckoutMergeFromBasePayload = CheckoutMergeFromBaseResponse["payload"];
type CheckoutPullPayload = CheckoutPullResponse["payload"];
type CheckoutPushPayload = CheckoutPushResponse["payload"];
type CheckoutRefreshPayload = CheckoutRefreshResponse["payload"];
type CheckoutPrCreatePayload = CheckoutPrCreateResponse["payload"];
type CheckoutPrMergePayload = CheckoutPrMergeResponse["payload"];
type CheckoutGithubSetAutoMergePayload = CheckoutGithubSetAutoMergeResponse["payload"];
type CheckoutGithubGetCheckDetailsPayload = CheckoutGithubGetCheckDetailsResponse["payload"];
type CheckoutPrStatusPayload = CheckoutPrStatusResponse["payload"];
type PullRequestTimelinePayload = PullRequestTimelineResponse["payload"];
type CheckoutSwitchBranchPayload = CheckoutSwitchBranchResponse["payload"];
export type RenameBranchResult = z.infer<typeof CheckoutRenameBranchResponseSchema>["payload"];
type StashSavePayload = StashSaveResponse["payload"];
type StashPopPayload = StashPopResponse["payload"];
type StashListPayload = StashListResponse["payload"];
type ValidateBranchPayload = ValidateBranchResponse["payload"];
type BranchSuggestionsPayload = BranchSuggestionsResponse["payload"];
type GitHubSearchPayload = GitHubSearchResponse["payload"];
type DirectorySuggestionsPayload = DirectorySuggestionsResponse["payload"];
type PaseoWorktreeListPayload = PaseoWorktreeListResponse["payload"];
type PaseoWorktreeArchivePayload = PaseoWorktreeArchiveResponse["payload"];
type CreatePaseoWorktreePayload = Extract<
SessionOutboundMessage,
{ type: "create_paseo_worktree_response" }
>["payload"];
type WorkspaceCreatePayload = Extract<
SessionOutboundMessage,
{ type: "workspace.create.response" }
>["payload"];
type FileExplorerPayload = FileExplorerResponse["payload"];
export type FileExplorerDirectoryPayload = NonNullable<FileExplorerPayload["directory"]>;
type LegacyFileExplorerFilePayload = NonNullable<FileExplorerPayload["file"]>;
export interface FileReadResult {
bytes: Uint8Array;
mime: string;
size: number;
path: string;
kind: LegacyFileExplorerFilePayload["kind"];
modifiedAt: string;
}
export interface FileUploadInput {
fileName: string;
mimeType: string;
bytes: Uint8Array | ArrayBuffer;
modifiedAt?: string;
requestId?: string;
chunkSize?: number;
}
export type FileUploadResult = FileUploadResponse["payload"];
type FileDownloadTokenPayload = FileDownloadTokenResponse["payload"];
type ListProviderFeaturesPayload = ListProviderFeaturesResponseMessage["payload"];
type ListProviderModelsPayload = ListProviderModelsResponseMessage["payload"];
type ListProviderModesPayload = ListProviderModesResponseMessage["payload"];
type ListAvailableProvidersPayload = ListAvailableProvidersResponse["payload"];
type GetProvidersSnapshotPayload = GetProvidersSnapshotResponseMessage["payload"];
type RefreshProvidersSnapshotPayload = RefreshProvidersSnapshotResponseMessage["payload"];
type ProviderDiagnosticPayload = ProviderDiagnosticResponseMessage["payload"];
type DaemonStatusPayload = DaemonGetStatusResponse["payload"];
type DaemonPairingOfferPayload = DaemonGetPairingOfferResponse["payload"];
type ReadProjectConfigPayload = Extract<
SessionOutboundMessage,
{ type: "read_project_config_response" }
>["payload"];
type WriteProjectConfigPayload = Extract<
SessionOutboundMessage,
{ type: "write_project_config_response" }
>["payload"];
type ListCommandsPayload = ListCommandsResponse["payload"];
type ListCommandsDraftConfig = Pick<
AgentSessionConfig,
"provider" | "cwd" | "modeId" | "model" | "thinkingOptionId" | "featureValues"
>;
export interface WriteProjectConfigInput {
repoRoot: string;
config: PaseoConfigRaw;
expectedRevision: PaseoConfigRevision | null;
requestId?: string;
}
interface ListCommandsOptions {
requestId?: string;
draftConfig?: ListCommandsDraftConfig;
}
type SetVoiceModePayload = Extract<
SessionOutboundMessage,
{ type: "set_voice_mode_response" }
>["payload"];
type DictationFinishAcceptedPayload = Extract<
SessionOutboundMessage,
{ type: "dictation_stream_finish_accepted" }
>["payload"];
type AgentPermissionResolvedPayload = AgentPermissionResolvedMessage["payload"];
type ListTerminalsPayload = ListTerminalsResponse["payload"];
type CreateTerminalPayload = CreateTerminalResponse["payload"];
export type RenameTerminalResult = z.infer<typeof RenameTerminalResponseSchema>["payload"];
type SubscribeTerminalPayload = SubscribeTerminalResponse["payload"];
type CloseItemsPayload = CloseItemsResponse["payload"];
type KillTerminalPayload = KillTerminalResponse["payload"];
type CaptureTerminalPayload = CaptureTerminalResponse["payload"];
type ChatCreatePayload = Extract<
SessionOutboundMessage,
{ type: "chat/create/response" }
>["payload"];
type ChatListPayload = Extract<SessionOutboundMessage, { type: "chat/list/response" }>["payload"];
type ChatInspectPayload = Extract<
SessionOutboundMessage,
{ type: "chat/inspect/response" }
>["payload"];
type ChatDeletePayload = Extract<
SessionOutboundMessage,
{ type: "chat/delete/response" }
>["payload"];
type ChatPostPayload = Extract<SessionOutboundMessage, { type: "chat/post/response" }>["payload"];
type ChatReadPayload = Extract<SessionOutboundMessage, { type: "chat/read/response" }>["payload"];
type ChatWaitPayload = Extract<SessionOutboundMessage, { type: "chat/wait/response" }>["payload"];
type LoopRunPayload = Extract<SessionOutboundMessage, { type: "loop/run/response" }>["payload"];
type LoopListPayload = Extract<SessionOutboundMessage, { type: "loop/list/response" }>["payload"];
type LoopInspectPayload = Extract<
SessionOutboundMessage,
{ type: "loop/inspect/response" }
>["payload"];
type LoopLogsPayload = Extract<SessionOutboundMessage, { type: "loop/logs/response" }>["payload"];
type LoopStopPayload = Extract<SessionOutboundMessage, { type: "loop/stop/response" }>["payload"];
type ScheduleCreatePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/create/response" }
>["payload"];
type ScheduleListPayload = Extract<
SessionOutboundMessage,
{ type: "schedule/list/response" }
>["payload"];
type ScheduleInspectPayload = Extract<
SessionOutboundMessage,
{ type: "schedule/inspect/response" }
>["payload"];
type ScheduleLogsPayload = Extract<
SessionOutboundMessage,
{ type: "schedule/logs/response" }
>["payload"];
type SchedulePausePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/pause/response" }
>["payload"];
type ScheduleResumePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/resume/response" }
>["payload"];
type ScheduleDeletePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/delete/response" }
>["payload"];
type ScheduleRunOncePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/run-once/response" }
>["payload"];
type ScheduleUpdatePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/update/response" }
>["payload"];
export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"];
export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"];
export type FetchAgentTimelineProjection = FetchAgentTimelinePayload["projection"];
export type FetchAgentTimelineCursor = NonNullable<FetchAgentTimelinePayload["startCursor"]>;
export interface FetchAgentTimelineOptions {
direction?: FetchAgentTimelineDirection;
cursor?: FetchAgentTimelineCursor;
limit?: number;
projection?: FetchAgentTimelineProjection;
requestId?: string;
}
type AgentRefreshedStatusPayload = z.infer<typeof AgentRefreshedStatusPayloadSchema>;
type RestartRequestedStatusPayload = z.infer<typeof RestartRequestedStatusPayloadSchema>;
type ShutdownRequestedStatusPayload = z.infer<typeof ShutdownRequestedStatusPayloadSchema>;
type FetchAgentsPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_agents_response" }
>["payload"];
type FetchAgentsRequest = Extract<SessionInboundMessage, { type: "fetch_agents_request" }>;
export type FetchAgentsOptions = Omit<FetchAgentsRequest, "type" | "requestId"> & {
requestId?: string;
};
export type FetchAgentsEntry = FetchAgentsPayload["entries"][number];
export type FetchAgentsPageInfo = FetchAgentsPayload["pageInfo"];
type FetchAgentHistoryPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_agent_history_response" }
>["payload"];
type FetchAgentHistoryRequest = Extract<
SessionInboundMessage,
{ type: "fetch_agent_history_request" }
>;
export type FetchAgentHistoryOptions = Omit<FetchAgentHistoryRequest, "type" | "requestId"> & {
requestId?: string;
};
export type FetchAgentHistoryEntry = FetchAgentHistoryPayload["entries"][number];
export type FetchAgentHistoryPageInfo = FetchAgentHistoryPayload["pageInfo"];
type FetchRecentProviderSessionsPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_recent_provider_sessions_response" }
>["payload"];
type FetchRecentProviderSessionsRequest = Extract<
SessionInboundMessage,
{ type: "fetch_recent_provider_sessions_request" }
>;
export type FetchRecentProviderSessionsOptions = Omit<
FetchRecentProviderSessionsRequest,
"type" | "requestId"
> & {
requestId?: string;
};
export type FetchRecentProviderSessionEntry = FetchRecentProviderSessionsPayload["entries"][number];
type FetchWorkspacesPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_workspaces_response" }
>["payload"];
type FetchWorkspacesRequest = Extract<SessionInboundMessage, { type: "fetch_workspaces_request" }>;
export type FetchWorkspacesOptions = Omit<FetchWorkspacesRequest, "type" | "requestId"> & {
requestId?: string;
};
export type FetchWorkspacesEntry = FetchWorkspacesPayload["entries"][number];
export type FetchWorkspacesPageInfo = FetchWorkspacesPayload["pageInfo"];
export interface CreateChatRoomOptions {
name: string;
purpose?: string | null;
requestId?: string;
}
export interface InspectChatRoomOptions {
room: string;
requestId?: string;
}
export interface DeleteChatRoomOptions {
room: string;
requestId?: string;
}
export interface PostChatMessageOptions {
room: string;
body: string;
authorAgentId?: string;
replyToMessageId?: string | null;
requestId?: string;
}
export interface ReadChatMessagesOptions {
room: string;
limit?: number;
since?: string;
authorAgentId?: string;
requestId?: string;
}
export interface WaitForChatMessagesOptions {
room: string;
afterMessageId?: string | null;
timeoutMs?: number;
requestId?: string;
}
export interface RunLoopOptions {
prompt: string;
cwd: string;
provider?: string;
model?: string;
modeId?: string;
verifierProvider?: string;
verifierModel?: string;
verifierModeId?: string;
verifyPrompt?: string | null;
verifyChecks?: string[];
name?: string | null;
sleepMs?: number;
maxIterations?: number;
maxTimeMs?: number;
requestId?: string;
}
export interface InspectLoopOptions {
id: string;
requestId?: string;
}
export interface LoopLogsOptions {
id: string;
afterSeq?: number;
requestId?: string;
}
export interface StopLoopOptions {
id: string;
requestId?: string;
}
export interface CreateScheduleOptions {
prompt: string;
name?: string | null;
cadence:
| {
type: "every";
everyMs: number;
}
| {
type: "cron";
expression: string;
timezone?: string;
};
target:
| {
type: "self";
agentId: string;
}
| {
type: "agent";
agentId: string;
}
| {
type: "new-agent";
config: {
provider: AgentProvider;
cwd: string;
modeId?: string;
model?: string;
thinkingOptionId?: string;
title?: string | null;
approvalPolicy?: string;
sandboxMode?: string;
networkAccess?: boolean;
webSearch?: boolean;
extra?: AgentSessionConfig["extra"];
systemPrompt?: string;
mcpServers?: AgentSessionConfig["mcpServers"];
};
};
maxRuns?: number;
expiresAt?: string;
runOnCreate?: boolean;
requestId?: string;
}
export interface InspectScheduleOptions {
id: string;
requestId?: string;
}
export interface UpdateScheduleNewAgentConfig {
provider?: string;
model?: string | null;
modeId?: string | null;
cwd?: string;
}
export interface UpdateScheduleOptions {
id: string;
name?: string | null;
prompt?: string;
cadence?:
| {
type: "every";
everyMs: number;
}
| {
type: "cron";
expression: string;
timezone?: string;
};
newAgentConfig?: UpdateScheduleNewAgentConfig;
maxRuns?: number | null;
expiresAt?: string | null;
requestId?: string;
}
export interface RenameBranchInput {
cwd: string;
branch: string;
requestId?: string;
}
export interface RenameTerminalInput {
terminalId: string;
title: string;
requestId?: string;
}
type OpenProjectPayload = OpenProjectResponseMessage["payload"];
type ArchiveWorkspacePayload = ArchiveWorkspaceResponseMessage["payload"];
type WorkspaceSetupStatusPayload = WorkspaceSetupStatusResponseMessage["payload"];
export interface FetchAgentResult {
agent: AgentSnapshotPayload;
project: ProjectPlacementPayload | null;
}
export interface WaitForFinishResult {
status: "idle" | "error" | "permission" | "timeout";
final: AgentSnapshotPayload | null;
error: string | null;
lastMessage: string | null;
}
interface Waiter<T> {
predicate: (msg: SessionOutboundMessage) => T | null;
resolve(value: T): void;
reject(error: Error): void;
timeoutHandle: ReturnType<typeof setTimeout> | null;
}
interface WaitHandle<T> {
promise: Promise<T>;
cancel: (error: Error) => void;
}
interface PendingBinaryFileRead {
cwd: string;
path: string;
}
interface BinaryFileTransferState extends PendingBinaryFileRead {
mime: string;
size: number;
encoding: Extract<
FileTransferFrame,
{ opcode: typeof FileTransferOpcode.FileBegin }
>["metadata"]["encoding"];
modifiedAt: string;
chunks: Uint8Array[];
}
type RpcWaitResult<T> = { kind: "ok"; value: T } | { kind: "error"; error: DaemonRpcError };
type GetDaemonConfigResponse = Extract<
SessionOutboundMessage,
{ type: "get_daemon_config_response" }
>;
type SetDaemonConfigResponse = Extract<
SessionOutboundMessage,
{ type: "set_daemon_config_response" }
>;
type CorrelatedResponseMessage =
| Extract<SessionOutboundMessage, { payload: { requestId: string } }>
| GetDaemonConfigResponse
| SetDaemonConfigResponse;
type CorrelatedResponseType = CorrelatedResponseMessage["type"];
type CorrelatedResponsePayload<TType extends CorrelatedResponseType> = Extract<
CorrelatedResponseMessage,
{ type: TType }
>["payload"];
class DaemonRpcError extends Error {
readonly requestId: string;
readonly requestType?: string;
readonly code?: string;
constructor(params: { requestId: string; error: string; requestType?: string; code?: string }) {
const parts = [params.error];
if (params.requestType) parts.push(`requestType=${params.requestType}`);
if (params.code) parts.push(`code=${params.code}`);
super(parts.join(" "));
this.name = "DaemonRpcError";
this.requestId = params.requestId;
this.requestType = params.requestType;
this.code = params.code;
}
}
class PingTimeoutError extends Error {
constructor(readonly timeoutMs: number) {
super(`Ping timed out (${timeoutMs}ms)`);
this.name = "PingTimeoutError";
}
}
function toTimeoutError(error: unknown, label: string, timeoutMs: number): Error {
if (error instanceof PingTimeoutError) {
return new Error(`${label} timed out (${timeoutMs}ms)`);
}
return error instanceof Error ? error : new Error(String(error));
}
const DEFAULT_RECONNECT_BASE_DELAY_MS = 1500;
const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000;
const DEFAULT_CONNECT_TIMEOUT_MS = 15000;
const DEFAULT_LIVENESS_TIMEOUT_MS = 5000;
const LIVENESS_HEARTBEAT_INTERVAL_MS = 10_000;
const LIVENESS_HEARTBEAT_TIMEOUT_MS = 15_000;
const LIVENESS_FAILURE_RECONNECT_THRESHOLD = 2;
/** Default timeout for waiting for connection before sending queued messages */
const DEFAULT_SEND_QUEUE_TIMEOUT_MS = 10000;
const DEFAULT_DICTATION_FINISH_ACCEPT_TIMEOUT_MS = 15000;
const DEFAULT_DICTATION_FINISH_FALLBACK_TIMEOUT_MS = 5 * 60 * 1000;
const DEFAULT_DICTATION_FINISH_TIMEOUT_GRACE_MS = 5000;
function isWaiterTimeoutError(error: unknown): boolean {
return error instanceof Error && error.message.startsWith("Timeout waiting for message");
}
function normalizeClientId(value: unknown): string | null {
if (typeof value !== "string") {
return null;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : null;
}
function decodeBase64ToBytes(base64: string): Uint8Array {
const binary = globalThis.atob(base64);
const bytes = new Uint8Array(binary.length);
for (let index = 0; index < binary.length; index += 1) {
bytes[index] = binary.charCodeAt(index);
}
return bytes;
}
function legacyExplorerFileToBytes(file: LegacyFileExplorerFilePayload): FileReadResult {
let bytes: Uint8Array;
if (file.encoding === "base64" && file.content) {
bytes = decodeBase64ToBytes(file.content);
} else if (file.encoding === "utf-8" && file.content) {
bytes = new TextEncoder().encode(file.content);
} else {
bytes = new Uint8Array();
}
return {
bytes,
mime: file.mimeType ?? "application/octet-stream",
size: file.size,
path: file.path,
kind: file.kind,
modifiedAt: file.modifiedAt,
};
}
function binaryFileKind(mime: string, encoding: string): FileReadResult["kind"] {
if (mime.startsWith("image/")) {
return "image";
}
if (encoding === "utf-8" || mime.startsWith("text/") || mime === "application/json") {
return "text";
}
return "binary";
}
function concatByteChunks(chunks: Uint8Array[], size: number): Uint8Array {
const bytes = new Uint8Array(size);
let offset = 0;
for (const chunk of chunks) {
bytes.set(chunk, offset);
offset += chunk.byteLength;
}
return bytes;
}
function hashForLog(value: string): string {
let hash = 0;
for (let index = 0; index < value.length; index += 1) {
hash = (hash * 31 + value.charCodeAt(index)) | 0;
}
return `h_${Math.abs(hash).toString(16)}`;
}
function toReasonCode(reason: string | null | undefined): string | null {
if (!reason) {
return null;
}
const normalized = reason.toLowerCase();
if (normalized.includes("timed out")) {
return "connect_timeout";
}
if (normalized.includes("disposed")) {
return "disposed";
}
if (normalized.includes("client closed")) {
return "client_closed";
}
if (normalized.includes("transport")) {
return "transport_error";
}
if (normalized.includes("failed to connect")) {
return "connect_failed";
}
return "unknown";
}
interface PendingSend {
message: SessionInboundMessage;
resolve: () => void;
reject: (error: Error) => void;
timeoutHandle: ReturnType<typeof setTimeout>;
}
interface PingProbe {
promise: Promise<number>;
resolve: (value: number) => void;
reject: (error: Error) => void;
timeoutHandle: ReturnType<typeof setTimeout>;
startedAt: number;
// Whether a timeout on this ping should be recorded as a liveness failure. Only the
// heartbeat sets this; a latency measurement never drives teardown, even when a
// heartbeat tick shares (dedupes onto) an in-flight measurement ping.
drivesLivenessFailure: boolean;
}
export class DaemonClient {
private transport: DaemonTransport | null = null;
private transportCleanup: Array<() => void> = [];
private rawMessageListeners: Set<(message: SessionOutboundMessage) => void> = new Set();
private messageHandlers: Map<
SessionOutboundMessage["type"],
Set<(message: SessionOutboundMessage) => void>
> = new Map();
private eventListeners: Set<DaemonEventHandler> = new Set();
private waiters: Set<Waiter<unknown>> = new Set();
private checkoutStatusInFlight: Map<string, Promise<CheckoutStatusPayload>> = new Map();
private connectionListeners: Set<(status: ConnectionState) => void> = new Set();
private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
private connectTimeout: ReturnType<typeof setTimeout> | null = null;
private pendingGenericTransportErrorTimeout: ReturnType<typeof setTimeout> | null = null;
private reconnectAttempt = 0;
private shouldReconnect = true;
private connectPromise: Promise<void> | null = null;
private connectResolve: (() => void) | null = null;
private connectReject: ((error: Error) => void) | null = null;
private lastErrorValue: string | null = null;
private connectionState: ConnectionState = { status: "idle" };
private checkoutDiffSubscriptions = new Map<
string,
{
cwd: string;
compare: { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean };
}
>();
private terminalDirectorySubscriptions = new Map<string, { cwd: string; workspaceId?: string }>();
private readonly terminalStreams = new TerminalStreamRouter();
private pendingBinaryFileReads = new Map<string, PendingBinaryFileRead>();
private activeBinaryFileTransfers = new Map<string, BinaryFileTransferState>();
private completedBinaryFileReads = new Map<string, FileReadResult>();
private logger: Logger;
private pendingSendQueue: PendingSend[] = [];
private readonly logConnectionPath: "direct" | "relay";
private readonly logServerId: string | null;
private readonly logClientIdHash: string;
private readonly logGeneration: number | null;
private lastServerInfoMessage: ServerInfoStatusPayload | null = null;
private runtimeMetricsInterval: ReturnType<typeof setInterval> | null = null;
private runtimeMetrics: DaemonClientRuntimeMetrics | null = null;
private pingProbe: PingProbe | null = null;
private livenessHeartbeatTimer: ReturnType<typeof setTimeout> | null = null;
private lastLivenessRttMs: number | null = null;
private consecutiveLivenessFailures = 0;
constructor(private config: DaemonClientConfig) {
this.logger = config.logger ?? consoleLogger;
this.logConnectionPath = isRelayClientWebSocketUrl(this.config.url) ? "relay" : "direct";
let parsedUrlForLog: URL | null = null;
try {
parsedUrlForLog = new URL(this.config.url);
} catch {
parsedUrlForLog = null;
}
const parsedServerIdForLog = normalizeClientId(parsedUrlForLog?.searchParams.get("serverId"));
this.logServerId = parsedServerIdForLog ?? parsedUrlForLog?.host ?? null;
const resolvedClientId = normalizeClientId(this.config.clientId);
if (!resolvedClientId) {
throw new Error("Daemon client requires a non-empty clientId");
}
this.config.clientId = resolvedClientId;
this.logClientIdHash = hashForLog(resolvedClientId);
this.logGeneration =
typeof this.config.runtimeGeneration === "number" &&
Number.isFinite(this.config.runtimeGeneration)
? this.config.runtimeGeneration
: null;
const runtimeMetricsIntervalMs =
typeof config.runtimeMetricsIntervalMs === "number" && config.runtimeMetricsIntervalMs > 0
? config.runtimeMetricsIntervalMs
: 0;
if (runtimeMetricsIntervalMs > 0) {
const runtimeMetricsWindowMs =
typeof config.runtimeMetricsWindowMs === "number" && config.runtimeMetricsWindowMs > 0
? Math.max(config.runtimeMetricsWindowMs, runtimeMetricsIntervalMs)
: undefined;
this.runtimeMetrics = new DaemonClientRuntimeMetrics(
this.logger,
{
connectionPath: this.logConnectionPath,
serverId: this.logServerId,
getConnectionStatus: () => this.connectionState.status,
},
runtimeMetricsWindowMs ? { windowMs: runtimeMetricsWindowMs } : undefined,
);
this.runtimeMetricsInterval = setInterval(() => {
this.runtimeMetrics?.flush();
}, runtimeMetricsIntervalMs);
}
}
// ============================================================================
// Connection
// ============================================================================
async connect(): Promise<void> {
if (this.connectionState.status === "disposed") {
throw new Error("Daemon client is disposed");
}
if (this.connectionState.status === "connected") {
return;
}
if (this.connectPromise) {
return this.connectPromise;
}
this.shouldReconnect = true;
this.connectPromise = new Promise((resolve, reject) => {
this.connectResolve = resolve;
this.connectReject = reject;
this.attemptConnect();
});
return this.connectPromise;
}
private attemptConnect(): void {
if (this.connectionState.status === "disposed") {
this.rejectConnect(new Error("Daemon client is disposed"));
return;
}
if (!this.shouldReconnect) {
this.rejectConnect(new Error("Daemon client is closed"));
return;
}
if (this.connectionState.status === "connecting") {
return;
}
const headers: Record<string, string> = {};
const password = normalizePassword(this.config.password);
if (password) {
headers.Authorization = `Bearer ${password}`;
} else if (this.config.authHeader) {
headers.Authorization = this.config.authHeader;
}
const protocols = password ? [`paseo.bearer.${password}`] : undefined;
try {
// Reconnect can overlap with browser close/error delivery ordering.
// Always dispose previous transport before constructing the next one.
this.disposeTransport();
const baseTransportFactory =
this.config.transportFactory ??
createWebSocketTransportFactory(this.config.webSocketFactory ?? defaultWebSocketFactory);
const shouldUseRelayE2ee =
this.config.e2ee?.enabled === true && isRelayClientWebSocketUrl(this.config.url);
let transportFactory = baseTransportFactory;
if (shouldUseRelayE2ee) {
const daemonPublicKeyB64 = this.config.e2ee?.daemonPublicKeyB64;
if (!daemonPublicKeyB64) {
throw new Error("daemonPublicKeyB64 is required for relay E2EE");
}
transportFactory = createRelayE2eeTransportFactory({
baseFactory: baseTransportFactory,
daemonPublicKeyB64,
logger: this.logger,
});
}
const transportUrl = this.resolveTransportUrlForAttempt();
const transport = transportFactory({
url: transportUrl,
headers,
...(protocols ? { protocols } : {}),
});
this.transport = transport;
this.lastServerInfoMessage = null;
this.updateConnectionState(
{
status: "connecting",
attempt: this.reconnectAttempt,
},
{ event: "CONNECT_REQUEST" },
);
this.resetConnectTimeout();
const timeoutMs = Math.max(1, this.config.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS);
this.connectTimeout = setTimeout(() => {
if (this.connectionState.status !== "connecting") {
return;
}
this.lastErrorValue = "Connection timed out";
this.disposeTransport(1001, "Connection timed out");
this.scheduleReconnect({
reason: "Connection timed out",
event: "CONNECT_TIMEOUT",
reasonCode: "connect_timeout",
});
}, timeoutMs);
this.transportCleanup = [
transport.onOpen(() => {
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout);
this.pendingGenericTransportErrorTimeout = null;
}
this.lastErrorValue = null;
this.sendHelloMessage();
}),
transport.onClose((event) => {
this.resetConnectTimeout();
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout);
this.pendingGenericTransportErrorTimeout = null;
}
const reason = describeTransportClose(event);
if (reason) {
this.lastErrorValue = reason;
}
this.scheduleReconnect({
reason,
event: "TRANSPORT_CLOSE",
reasonCode: "transport_closed",
});
}),
transport.onError((event) => {
this.resetConnectTimeout();
const reason = describeTransportError(event);
const isGeneric = reason === "Transport error";
// Browser WebSocket.onerror often provides no useful details and is followed
// by a close event (often with code 1006). Prefer surfacing the close details
// instead of immediately disconnecting with a generic "Transport error".
if (isGeneric) {
this.lastErrorValue ??= reason;
if (!this.pendingGenericTransportErrorTimeout) {
this.pendingGenericTransportErrorTimeout = setTimeout(() => {
this.pendingGenericTransportErrorTimeout = null;
if (
this.connectionState.status === "connected" ||
this.connectionState.status === "connecting"
) {
this.lastErrorValue = reason;
this.scheduleReconnect({
reason,
event: "TRANSPORT_ERROR",
reasonCode: "transport_error",
});
}
}, 250);
}
return;
}
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout);
this.pendingGenericTransportErrorTimeout = null;
}
this.lastErrorValue = reason;
this.scheduleReconnect({
reason,
event: "TRANSPORT_ERROR",
reasonCode: "transport_error",
});
}),
transport.onMessage((data) => this.handleTransportMessage(data)),
];
} catch (error) {
this.resetConnectTimeout();
const message = error instanceof Error ? error.message : "Failed to connect";
this.lastErrorValue = message;
this.scheduleReconnect({
reason: message,
event: "CONNECT_FAILED",
reasonCode: "connect_failed",
});
this.rejectConnect(error instanceof Error ? error : new Error(message));
}
}
private resolveConnect(): void {
if (this.connectResolve) {
this.connectResolve();
}
this.connectPromise = null;
this.connectResolve = null;
this.connectReject = null;
}
private rejectConnect(error: Error): void {
if (this.connectReject) {
this.connectReject(error);
}
this.connectPromise = null;
this.connectResolve = null;
this.connectReject = null;
}
async close(): Promise<void> {
if (this.connectionState.status === "disposed") {
return;
}
this.shouldReconnect = false;
this.connectPromise = null;
this.connectResolve = null;
this.connectReject = null;
if (this.reconnectTimeout) {
clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
}
this.resetConnectTimeout();
this.disposeTransport(1000, "Client closed");
this.clearWaiters(new Error("Daemon client closed"));
this.rejectPendingSendQueue(new Error("Daemon client closed"));
this.rejectPingProbe(new Error("Daemon client closed"));
this.terminalStreams.clearSlots();
this.lastServerInfoMessage = null;
if (this.runtimeMetricsInterval) {
clearInterval(this.runtimeMetricsInterval);
this.runtimeMetricsInterval = null;
this.runtimeMetrics?.flush({ final: true });
this.runtimeMetrics = null;
}
this.updateConnectionState(
{ status: "disposed" },
{ event: "DISPOSE", reason: "Client closed", reasonCode: "disposed" },
);
}
ensureConnected(): void {
if (this.connectionState.status === "disposed") {
return;
}
if (!this.shouldReconnect) {
this.shouldReconnect = true;
}
if (
this.connectionState.status === "connected" ||
this.connectionState.status === "connecting"
) {
return;
}
void this.connect();
}
getConnectionState(): ConnectionState {
return this.connectionState;
}
subscribeConnectionStatus(listener: (status: ConnectionState) => void): () => void {
this.connectionListeners.add(listener);
listener(this.connectionState);
return () => {
this.connectionListeners.delete(listener);
};
}
get isConnected(): boolean {
return this.connectionState.status === "connected";
}
get isConnecting(): boolean {
return this.connectionState.status === "connecting";
}
get lastError(): string | null {
return this.lastErrorValue;
}
getLastLivenessRttMs(): number | null {
return this.lastLivenessRttMs;
}
// ============================================================================
// Message Subscription
// ============================================================================
subscribe(handler: DaemonEventHandler): () => void {
this.eventListeners.add(handler);
return () => this.eventListeners.delete(handler);
}
subscribeRawMessages(handler: (message: SessionOutboundMessage) => void): () => void {
this.rawMessageListeners.add(handler);
return () => {
this.rawMessageListeners.delete(handler);
};
}
on<TType extends SessionOutboundMessage["type"]>(
type: TType,
handler: (message: Extract<SessionOutboundMessage, { type: TType }>) => void,
): () => void;
on(handler: DaemonEventHandler): () => void;
on(
arg1: SessionOutboundMessage["type"] | DaemonEventHandler,
arg2?: (message: SessionOutboundMessage) => void,
): () => void {
if (typeof arg1 === "function") {
return this.subscribe(arg1);
}
const type = arg1;
const handler = arg2!;
if (!this.messageHandlers.has(type)) {
this.messageHandlers.set(type, new Set());
}
this.messageHandlers.get(type)!.add(handler);
return () => {
const handlers = this.messageHandlers.get(type);
if (!handlers) {
return;
}
handlers.delete(handler);
if (handlers.size === 0) {
this.messageHandlers.delete(type);
}
};
}
// ============================================================================
// Core Send Helpers
// ============================================================================
/**
* Send a session message. For fire-and-forget messages (heartbeats, etc.),
* failures are suppressed if `suppressSendErrors` is configured.
* For RPC methods that wait for responses, use `sendSessionMessageOrThrow` instead.
*/
private sendSessionMessage(message: SessionInboundMessage): void {
if (!this.transport || this.connectionState.status !== "connected") {
if (this.config.suppressSendErrors) {
return;
}
throw new Error(`Transport not connected (status: ${this.connectionState.status})`);
}
const payload = SessionInboundMessageSchema.parse(message);
try {
this.transport.send(JSON.stringify({ type: "session", message: payload }));
} catch (error) {
if (this.config.suppressSendErrors) {
return;
}
throw error instanceof Error ? error : new Error(String(error));
}
}
private sendBinaryFrame(frame: Uint8Array): void {
if (!this.transport || this.connectionState.status !== "connected") {
if (this.config.suppressSendErrors) {
return;
}
throw new Error(`Transport not connected (status: ${this.connectionState.status})`);
}
try {
this.transport.send(frame);
} catch (error) {
if (this.config.suppressSendErrors) {
return;
}
throw error instanceof Error ? error : new Error(String(error));
}
}
/**
* Send a session message for RPC methods that create waiters.
* If the connection is still being established ("connecting"), the message
* is queued and will be sent once connected (or rejected after timeout).
* This prevents waiters from hanging forever when called during connection.
*/
private sendSessionMessageOrThrow(message: SessionInboundMessage): Promise<void> {
const status = this.connectionState.status;
// If connected, send immediately
if (this.transport && status === "connected") {
const payload = SessionInboundMessageSchema.parse(message);
this.transport.send(JSON.stringify({ type: "session", message: payload }));
return Promise.resolve();
}
// If connecting, queue the message to be sent once connected
if (status === "connecting") {
return new Promise((resolve, reject) => {
const timeoutHandle = setTimeout(() => {
// Remove from queue
const idx = this.pendingSendQueue.findIndex((p) => p.resolve === resolve);
if (idx !== -1) {
this.pendingSendQueue.splice(idx, 1);
}
reject(new Error(`Timed out waiting for connection to send message`));
}, DEFAULT_SEND_QUEUE_TIMEOUT_MS);
this.pendingSendQueue.push({ message, resolve, reject, timeoutHandle });
});
}
// Not connected and not connecting - fail immediately
return Promise.reject(new Error(`Transport not connected (status: ${status})`));
}
/**
* Flush pending send queue - called when connection is established.
*/
private flushPendingSendQueue(): void {
const queue = this.pendingSendQueue;
this.pendingSendQueue = [];
for (const pending of queue) {
clearTimeout(pending.timeoutHandle);
try {
if (this.transport && this.connectionState.status === "connected") {
const payload = SessionInboundMessageSchema.parse(pending.message);
this.transport.send(JSON.stringify({ type: "session", message: payload }));
pending.resolve();
} else {
pending.reject(new Error("Connection lost before message could be sent"));
}
} catch (error) {
pending.reject(error instanceof Error ? error : new Error(String(error)));
}
}
}
/**
* Reject all pending sends - called when connection fails or is closed.
*/
private rejectPendingSendQueue(error: Error): void {
const queue = this.pendingSendQueue;
this.pendingSendQueue = [];
for (const pending of queue) {
clearTimeout(pending.timeoutHandle);
pending.reject(error);
}
}
private async sendRequest<T>(params: {
requestId: string;
message: SessionInboundMessage;
timeout: number;
select: (msg: SessionOutboundMessage) => T | null;
options?: { skipQueue?: boolean };
}): Promise<T> {
const { promise, cancel } = this.waitForWithCancel<RpcWaitResult<T>>(
(msg) => {
if (msg.type === "rpc_error" && msg.payload.requestId === params.requestId) {
return {
kind: "error",
error: new DaemonRpcError({
requestId: msg.payload.requestId,
error: msg.payload.error,
requestType: msg.payload.requestType,
code: msg.payload.code,
}),
};
}
const value = params.select(msg);
if (value === null) {
return null;
}
return { kind: "ok", value };
},
params.timeout,
params.options,
);
try {
await this.sendSessionMessageOrThrow(params.message);
} catch (error) {
const err = error instanceof Error ? error : new Error(String(error));
cancel(err);
void promise.catch(() => undefined);
throw err;
}
const result = await promise;
if (result.kind === "error") {
throw result.error;
}
return result.value;
}
private async sendCorrelatedRequest<
TResponseType extends CorrelatedResponseType,
TResult = CorrelatedResponsePayload<TResponseType>,
>(params: {
requestId: string;
message: SessionInboundMessage;
timeout: number;
responseType: TResponseType;
options?: { skipQueue?: boolean };
selectPayload?: (payload: CorrelatedResponsePayload<TResponseType>) => TResult | null;
}): Promise<TResult> {
return this.sendRequest({
requestId: params.requestId,
message: params.message,
timeout: params.timeout,
options: params.options,
select: (msg) => {
const correlated = msg as CorrelatedResponseMessage;
if (correlated.type !== params.responseType) {
return null;
}
const payload = correlated.payload as unknown as CorrelatedResponsePayload<TResponseType>;
if (payload.requestId !== params.requestId) {
return null;
}
if (!params.selectPayload) {
return payload as TResult;
}
return params.selectPayload(payload);
},
});
}
private sendCorrelatedSessionRequest<
TResponseType extends CorrelatedResponseType,
TResult = CorrelatedResponsePayload<TResponseType>,
>(params: {
requestId?: string;
message: { type: SessionInboundMessage["type"] } & Record<string, unknown>;
responseType: TResponseType;
timeout: number;
selectPayload?: (payload: CorrelatedResponsePayload<TResponseType>) => TResult | null;
}): Promise<TResult> {
const resolvedRequestId = this.createRequestId(params.requestId);
const message = SessionInboundMessageSchema.parse({
...params.message,
requestId: resolvedRequestId,
});
return this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: params.responseType,
timeout: params.timeout,
options: { skipQueue: true },
...(params.selectPayload ? { selectPayload: params.selectPayload } : {}),
});
}
private sendNamespacedCorrelatedSessionRequest<
TResponseType extends CorrelatedResponseType,
TResult = CorrelatedResponsePayload<TResponseType>,
>(params: {
requestId?: string;
message: { type: Extract<SessionInboundMessage["type"], `${string}.request`> } & Record<
string,
unknown
>;
timeout: number;
selectPayload?: (payload: CorrelatedResponsePayload<TResponseType>) => TResult | null;
}): Promise<TResult> {
const responseType = params.message.type.replace(/\.request$/, ".response") as TResponseType;
return this.sendCorrelatedSessionRequest({
...params,
responseType,
});
}
private sendSessionMessageStrict(message: SessionInboundMessage): void {
if (!this.transport || this.connectionState.status !== "connected") {
throw new Error("Transport not connected");
}
const payload = SessionInboundMessageSchema.parse(message);
try {
this.transport.send(JSON.stringify({ type: "session", message: payload }));
} catch (error) {
throw error instanceof Error ? error : new Error(String(error));
}
}
async clearAgentAttention(agentId: string | string[]): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "clear_agent_attention",
agentId,
requestId,
});
await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "clear_agent_attention_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
}
async clearWorkspaceAttention(workspaceId: string | string[]): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "workspace.clear_attention.request",
workspaceId,
requestId,
});
const response = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "workspace.clear_attention.response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!response.success) {
throw new Error(response.error ?? "Failed to clear workspace attention");
}
}
sendHeartbeat(params: {
deviceType: "web" | "mobile";
focusedAgentId: string | null;
focusedTerminalId?: string | null;
lastActivityAt: string;
appVisible: boolean;
appVisibilityChangedAt?: string;
}): void {
this.sendSessionMessage({
type: "client_heartbeat",
deviceType: params.deviceType,
focusedAgentId: params.focusedAgentId,
focusedTerminalId: params.focusedTerminalId ?? null,
lastActivityAt: params.lastActivityAt,
appVisible: params.appVisible,
appVisibilityChangedAt: params.appVisibilityChangedAt,
});
}
registerPushToken(token: string): void {
this.sendSessionMessage({
type: "register_push_token",
token,
});
}
async ping(params?: { requestId?: string; timeoutMs?: number }): Promise<{
requestId: string;
clientSentAt: number;
serverReceivedAt: number;
serverSentAt: number;
rttMs: number;
}> {
const requestId =
params?.requestId ?? `ping-${Date.now()}-${Math.random().toString(36).slice(2)}`;
const clientSentAt = Date.now();
const payload = await this.sendRequest({
requestId,
message: { type: "ping", requestId, clientSentAt },
timeout: params?.timeoutMs ?? 5000,
select: (msg) => {
if (msg.type !== "pong") return null;
if (msg.payload.requestId !== requestId) return null;
if (typeof msg.payload.serverReceivedAt !== "number") return null;
if (typeof msg.payload.serverSentAt !== "number") return null;
return msg.payload;
},
});
return {
requestId,
clientSentAt,
serverReceivedAt: payload.serverReceivedAt,
serverSentAt: payload.serverSentAt,
rttMs: Date.now() - clientSentAt,
};
}
measureLatency(params?: { timeoutMs?: number }): Promise<number> {
const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS);
return this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: false }).catch((error) => {
throw toTimeoutError(error, "Latency measurement", timeoutMs);
});
}
private async livenessPing(params?: { timeoutMs?: number }): Promise<number> {
const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS);
try {
const rttMs = await this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: true });
this.lastLivenessRttMs = rttMs;
return rttMs;
} catch (error) {
throw toTimeoutError(error, "Liveness check", timeoutMs);
}
}
private sendPingAwaitRtt(params: {
timeoutMs: number;
drivesLivenessFailure: boolean;
}): Promise<number> {
if (this.connectionState.status !== "connected" || !this.transport) {
return Promise.reject(
new Error(`Transport not connected (status: ${this.connectionState.status})`),
);
}
if (this.pingProbe) {
return this.pingProbe.promise;
}
const startedAt = perfNow();
const timeoutMs = params.timeoutMs;
let resolveProbe: ((value: number) => void) | null = null;
let rejectProbe: ((error: Error) => void) | null = null;
const promise = new Promise<number>((resolve, reject) => {
resolveProbe = resolve;
rejectProbe = reject;
});
const probe: PingProbe = {
promise,
resolve: (value) => resolveProbe?.(value),
reject: (error) => rejectProbe?.(error),
timeoutHandle: setTimeout(() => {
if (this.pingProbe !== probe) {
return;
}
this.pingProbe = null;
const error = new PingTimeoutError(timeoutMs);
probe.reject(error);
if (probe.drivesLivenessFailure) {
this.recordLivenessFailure(toTimeoutError(error, "Liveness check", timeoutMs));
}
}, timeoutMs),
startedAt,
drivesLivenessFailure: params.drivesLivenessFailure,
};
this.pingProbe = probe;
try {
this.transport.send(JSON.stringify({ type: "ping" }));
} catch (error) {
this.clearPingProbe();
const sendError = error instanceof Error ? error : new Error(String(error));
if (probe.drivesLivenessFailure) {
this.recordLivenessFailure(sendError);
}
return Promise.reject(sendError);
}
return promise;
}
private startLivenessHeartbeat(): void {
this.stopLivenessHeartbeat();
this.lastLivenessRttMs = null;
this.scheduleNextLivenessHeartbeat();
}
private stopLivenessHeartbeat(): void {
if (!this.livenessHeartbeatTimer) {
return;
}
clearTimeout(this.livenessHeartbeatTimer);
this.livenessHeartbeatTimer = null;
}
private scheduleNextLivenessHeartbeat(): void {
if (this.connectionState.status !== "connected" || this.livenessHeartbeatTimer) {
return;
}
this.livenessHeartbeatTimer = setTimeout(() => {
this.livenessHeartbeatTimer = null;
this.livenessPing({ timeoutMs: LIVENESS_HEARTBEAT_TIMEOUT_MS })
.catch(() => {})
.finally(() => {
this.scheduleNextLivenessHeartbeat();
});
}, LIVENESS_HEARTBEAT_INTERVAL_MS);
}
// ============================================================================
// Agent RPCs (requestId-correlated)
// ============================================================================
async fetchAgents(options?: FetchAgentsOptions): Promise<FetchAgentsPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agents_request",
requestId: resolvedRequestId,
...(options?.scope ? { scope: options.scope } : {}),
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
...(options?.subscribe ? { subscribe: options.subscribe } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_agents_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
}
async fetchAgentHistory(options?: FetchAgentHistoryOptions): Promise<FetchAgentHistoryPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agent_history_request",
requestId: resolvedRequestId,
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_agent_history_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
}
async fetchRecentProviderSessions(
options?: FetchRecentProviderSessionsOptions,
): Promise<FetchRecentProviderSessionsPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_recent_provider_sessions_request",
requestId: resolvedRequestId,
...(options?.cwd ? { cwd: options.cwd } : {}),
...(options?.providers ? { providers: options.providers } : {}),
...(options?.since ? { since: options.since } : {}),
...(options?.limit ? { limit: options.limit } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_recent_provider_sessions_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
}
async fetchWorkspaces(options?: FetchWorkspacesOptions): Promise<FetchWorkspacesPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_workspaces_request",
requestId: resolvedRequestId,
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
...(options?.subscribe ? { subscribe: options.subscribe } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_workspaces_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
}
async openProject(cwd: string, requestId?: string): Promise<OpenProjectPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "open_project_request",
cwd,
},
responseType: "open_project_response",
timeout: 10000,
});
}
async startWorkspaceScript(
workspaceId: string,
scriptName: string,
requestId?: string,
): Promise<
Extract<SessionOutboundMessage, { type: "start_workspace_script_response" }>["payload"]
> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "start_workspace_script_request",
workspaceId,
scriptName,
},
responseType: "start_workspace_script_response",
timeout: 10000,
});
}
async archiveWorkspace(
workspaceId: string,
requestId?: string,
): Promise<ArchiveWorkspacePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "archive_workspace_request",
workspaceId,
},
responseType: "archive_workspace_response",
timeout: 10000,
});
}
async fetchWorkspaceSetupStatus(
workspaceId: string,
requestId?: string,
): Promise<WorkspaceSetupStatusPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "workspace_setup_status_request",
workspaceId,
},
responseType: "workspace_setup_status_response",
timeout: 10000,
});
}
async fetchAgent(agentId: string, requestId?: string): Promise<FetchAgentResult | null> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agent_request",
requestId: resolvedRequestId,
agentId,
});
const payload = await this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_agent_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
if (payload.error) {
throw new Error(payload.error);
}
if (!payload.agent) {
return null;
}
return { agent: payload.agent, project: payload.project ?? null };
}
private resubscribeCheckoutDiffSubscriptions(): void {
if (this.checkoutDiffSubscriptions.size === 0) {
return;
}
for (const [subscriptionId, subscription] of this.checkoutDiffSubscriptions) {
const message = SessionInboundMessageSchema.parse({
type: "subscribe_checkout_diff_request",
subscriptionId,
cwd: subscription.cwd,
compare: subscription.compare,
requestId: this.createRequestId(),
});
this.sendSessionMessage(message);
}
}
private resubscribeTerminalDirectorySubscriptions(): void {
if (this.terminalDirectorySubscriptions.size === 0) {
return;
}
for (const subscription of this.terminalDirectorySubscriptions.values()) {
this.sendSessionMessage({
type: "subscribe_terminals_request",
cwd: subscription.cwd,
...(subscription.workspaceId !== undefined
? { workspaceId: subscription.workspaceId }
: {}),
});
}
}
// ============================================================================
// Agent Lifecycle
// ============================================================================
async createAgent(options: CreateAgentRequestOptions): Promise<AgentSnapshotPayload> {
const requestId = this.createRequestId(options.requestId);
const config = resolveAgentConfig(options);
const message = SessionInboundMessageSchema.parse({
type: "create_agent_request",
requestId,
config,
...(options.env ? { env: options.env } : {}),
...(options.workspaceId !== undefined ? { workspaceId: options.workspaceId } : {}),
...(options.initialPrompt ? { initialPrompt: options.initialPrompt } : {}),
...(options.clientMessageId ? { clientMessageId: options.clientMessageId } : {}),
...(options.outputSchema ? { outputSchema: options.outputSchema } : {}),
...(options.images && options.images.length > 0 ? { images: options.images } : {}),
...(options.attachments && options.attachments.length > 0
? { attachments: options.attachments }
: {}),
...(options.git ? { git: options.git } : {}),
...(options.worktree ? { worktree: options.worktree } : {}),
...(options.autoArchive !== undefined ? { autoArchive: options.autoArchive } : {}),
...(options.worktreeName ? { worktreeName: options.worktreeName } : {}),
...(options.labels && Object.keys(options.labels).length > 0
? { labels: options.labels }
: {}),
});
const status = await this.sendRequest({
requestId,
message,
timeout: 60000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "status") {
return null;
}
const created = AgentCreatedStatusPayloadSchema.safeParse(msg.payload);
if (created.success && created.data.requestId === requestId) {
return created.data;
}
const failed = AgentCreateFailedStatusPayloadSchema.safeParse(msg.payload);
if (failed.success && failed.data.requestId === requestId) {
return failed.data;
}
return null;
},
});
if (status.status === "agent_create_failed") {
throw new Error(status.error);
}
return status.agent;
}
async deleteAgent(agentId: string): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "delete_agent_request",
agentId,
requestId,
});
await this.sendRequest({
requestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "agent_deleted") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
}
async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "archive_agent_request",
agentId,
requestId,
});
const result = await this.sendRequest({
requestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "agent_archived") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
return { archivedAt: result.archivedAt };
}
async updateAgent(
agentId: string,
updates: { name?: string; labels?: Record<string, string> },
): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "update_agent_request",
agentId,
...(updates.name !== undefined ? { name: updates.name } : {}),
...(updates.labels && Object.keys(updates.labels).length > 0
? { labels: updates.labels }
: {}),
requestId,
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "update_agent_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.accepted) {
throw new Error(payload.error ?? "updateAgent rejected");
}
}
async renameProject(
projectId: string,
customName: string | null,
requestId?: string,
): Promise<{ customName: string | null }> {
const payload = await this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "project.rename.request",
projectId,
customName,
},
responseType: "project.rename.response",
timeout: 10000,
});
if (!payload.accepted) {
throw new Error(payload.error ?? "renameProject rejected");
}
return { customName: payload.customName };
}
async setWorkspaceTitle(
workspaceId: string,
title: string | null,
requestId?: string,
): Promise<{ title: string | null }> {
const payload = await this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "workspace.title.set.request",
workspaceId,
title,
},
responseType: "workspace.title.set.response",
timeout: 10000,
});
if (!payload.accepted) {
throw new Error(payload.error ?? "setWorkspaceTitle rejected");
}
return { title: payload.title };
}
async resumeAgent(
handle: AgentPersistenceHandle,
overrides?: Partial<AgentSessionConfig>,
): Promise<AgentSnapshotPayload> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "resume_agent_request",
requestId,
handle,
...(overrides ? { overrides } : {}),
});
const status = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "status") {
return null;
}
const resumed = AgentResumedStatusPayloadSchema.safeParse(msg.payload);
if (resumed.success && resumed.data.requestId === requestId) {
return resumed.data;
}
return null;
},
});
return status.agent;
}
async importAgent(input: ImportAgentInput): Promise<AgentSnapshotPayload> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "import_agent_request",
requestId,
...("providerId" in input
? { providerId: input.providerId, providerHandleId: input.providerHandleId }
: { provider: input.provider, sessionId: input.sessionId }),
...(input.cwd ? { cwd: input.cwd } : {}),
...(input.labels && Object.keys(input.labels).length > 0 ? { labels: input.labels } : {}),
});
const status = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "status") {
return null;
}
const resumed = AgentResumedStatusPayloadSchema.safeParse(msg.payload);
if (resumed.success && resumed.data.requestId === requestId) {
return resumed.data;
}
const failed = AgentCreateFailedStatusPayloadSchema.safeParse(msg.payload);
if (failed.success && failed.data.requestId === requestId) {
return failed.data;
}
return null;
},
});
if (status.status === "agent_create_failed") {
throw new Error(status.error);
}
return status.agent;
}
async refreshAgent(agentId: string, requestId?: string): Promise<AgentRefreshedStatusPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "refresh_agent_request",
agentId,
requestId: resolvedRequestId,
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "status") {
return null;
}
const refreshed = AgentRefreshedStatusPayloadSchema.safeParse(msg.payload);
if (refreshed.success && refreshed.data.requestId === resolvedRequestId) {
return refreshed.data;
}
return null;
},
});
}
async fetchAgentTimeline(
agentId: string,
options: FetchAgentTimelineOptions = {},
): Promise<FetchAgentTimelinePayload> {
const resolvedRequestId = this.createRequestId(options.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agent_timeline_request",
agentId,
requestId: resolvedRequestId,
...(options.direction ? { direction: options.direction } : {}),
...(options.cursor ? { cursor: options.cursor } : {}),
...(typeof options.limit === "number" ? { limit: options.limit } : {}),
...(options.projection ? { projection: options.projection } : {}),
});
const payload = await this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_agent_timeline_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
if (payload.error) {
throw new Error(payload.error);
}
return payload;
}
// ============================================================================
// Agent Interaction
// ============================================================================
async sendAgentMessage(
agentId: string,
text: string,
options?: SendMessageOptions,
): Promise<void> {
const requestId = this.createRequestId();
const messageId = options?.messageId ?? crypto.randomUUID();
const message = SessionInboundMessageSchema.parse({
type: "send_agent_message_request",
requestId,
agentId,
text,
...(messageId ? { messageId } : {}),
...(options?.images ? { images: options.images } : {}),
...(options?.attachments ? { attachments: options.attachments } : {}),
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "send_agent_message_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.accepted) {
throw new Error(payload.error ?? "sendAgentMessage rejected");
}
}
async sendMessage(agentId: string, text: string, options?: SendMessageOptions): Promise<void> {
await this.sendAgentMessage(agentId, text, options);
}
async rewindAgent(
agentId: string,
messageId: string,
mode: "conversation" | "files" | "both",
): Promise<AgentRewindResponseMessage["payload"]> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "agent.rewind.request",
requestId,
agentId,
messageId,
mode,
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "agent.rewind.response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.ok) {
throw new Error(payload.error ?? "Agent rewind failed");
}
return payload;
}
async cancelAgent(agentId: string): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "cancel_agent_request",
agentId,
requestId,
});
await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "cancel_agent_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
}
async setAgentMode(agentId: string, modeId: string): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "set_agent_mode_request",
agentId,
modeId,
requestId,
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "set_agent_mode_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.accepted) {
throw new Error(payload.error ?? "setAgentMode rejected");
}
}
async setAgentModel(agentId: string, modelId: string | null): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "set_agent_model_request",
agentId,
modelId,
requestId,
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "set_agent_model_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.accepted) {
throw new Error(payload.error ?? "setAgentModel rejected");
}
}
async setAgentFeature(agentId: string, featureId: string, value: unknown): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "set_agent_feature_request",
agentId,
featureId,
value,
requestId,
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "set_agent_feature_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.accepted) {
throw new Error(payload.error ?? "setAgentFeature rejected");
}
}
async setAgentThinkingOption(agentId: string, thinkingOptionId: string | null): Promise<void> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "set_agent_thinking_request",
agentId,
thinkingOptionId,
requestId,
});
const payload = await this.sendRequest({
requestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "set_agent_thinking_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!payload.accepted) {
throw new Error(payload.error ?? "setAgentThinkingOption rejected");
}
}
async restartServer(reason?: string, requestId?: string): Promise<RestartRequestedStatusPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "restart_server_request",
...(reason && reason.trim().length > 0 ? { reason } : {}),
requestId: resolvedRequestId,
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "status") {
return null;
}
const restarted = RestartRequestedStatusPayloadSchema.safeParse(msg.payload);
if (!restarted.success) {
return null;
}
if (restarted.data.requestId !== resolvedRequestId) {
return null;
}
return restarted.data;
},
});
}
async shutdownServer(requestId?: string): Promise<ShutdownRequestedStatusPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "shutdown_server_request",
requestId: resolvedRequestId,
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "status") {
return null;
}
const shutdown = ShutdownRequestedStatusPayloadSchema.safeParse(msg.payload);
if (!shutdown.success) {
return null;
}
if (shutdown.data.requestId !== resolvedRequestId) {
return null;
}
return shutdown.data;
},
});
}
// ============================================================================
// Audio / Voice
// ============================================================================
async setVoiceMode(enabled: boolean, agentId?: string): Promise<SetVoiceModePayload> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "set_voice_mode",
enabled,
...(agentId ? { agentId } : {}),
requestId,
});
const response = await this.sendRequest({
requestId,
message,
timeout: 10000,
select: (msg) => {
if (msg.type !== "set_voice_mode_response") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
});
if (!response.accepted) {
const codeSuffix =
typeof response.reasonCode === "string" && response.reasonCode.trim().length > 0
? ` (${response.reasonCode})`
: "";
throw new Error((response.error ?? "Failed to set voice mode") + codeSuffix);
}
return response;
}
async sendVoiceAudioChunk(audio: string, format: string, isLast = false): Promise<void> {
this.sendSessionMessage({ type: "voice_audio_chunk", audio, format, isLast });
}
async startDictationStream(dictationId: string, format: string): Promise<void> {
const ack = this.waitForWithCancel(
(msg) => {
if (msg.type !== "dictation_stream_ack") {
return null;
}
if (msg.payload.dictationId !== dictationId) {
return null;
}
if (msg.payload.ackSeq !== -1) {
return null;
}
return msg.payload;
},
30000,
{ skipQueue: true },
);
const ackPromise = ack.promise.then(() => undefined);
const streamError = this.waitForWithCancel(
(msg) => {
if (msg.type !== "dictation_stream_error") {
return null;
}
if (msg.payload.dictationId !== dictationId) {
return null;
}
return msg.payload;
},
30000,
{ skipQueue: true },
);
const errorPromise = streamError.promise.then((payload) => {
throw new Error(payload.error);
});
const cleanupError = new Error("Cancelled dictation start waiter");
try {
this.sendSessionMessageStrict({ type: "dictation_stream_start", dictationId, format });
await Promise.race([ackPromise, errorPromise]);
} finally {
ack.cancel(cleanupError);
streamError.cancel(cleanupError);
void ackPromise.catch(() => undefined);
void errorPromise.catch(() => undefined);
}
}
sendDictationStreamChunk(dictationId: string, seq: number, audio: string, format: string): void {
this.sendSessionMessageStrict({
type: "dictation_stream_chunk",
dictationId,
seq,
audio,
format,
});
}
async finishDictationStream(
dictationId: string,
finalSeq: number,
): Promise<{ dictationId: string; text: string }> {
const final = this.waitForWithCancel(
(msg) => {
if (msg.type !== "dictation_stream_final") {
return null;
}
if (msg.payload.dictationId !== dictationId) {
return null;
}
return msg.payload;
},
0,
{ skipQueue: true },
);
const streamError = this.waitForWithCancel(
(msg) => {
if (msg.type !== "dictation_stream_error") {
return null;
}
if (msg.payload.dictationId !== dictationId) {
return null;
}
return msg.payload;
},
0,
{ skipQueue: true },
);
const finishAccepted = this.waitForWithCancel<DictationFinishAcceptedPayload>(
(msg) => {
if (msg.type !== "dictation_stream_finish_accepted") {
return null;
}
if (msg.payload.dictationId !== dictationId) {
return null;
}
return msg.payload;
},
DEFAULT_DICTATION_FINISH_ACCEPT_TIMEOUT_MS,
{ skipQueue: true },
);
const finalPromise = final.promise;
const errorPromise = streamError.promise.then((payload) => {
throw new Error(payload.error);
});
const finishAcceptedPromise = finishAccepted.promise;
const finalOutcomePromise = finalPromise.then((payload) => ({
kind: "final" as const,
payload,
}));
const errorOutcomePromise = errorPromise.then(
() => ({
kind: "error" as const,
error: new Error("Unexpected dictation stream error state"),
}),
(error) => ({
kind: "error" as const,
error: error instanceof Error ? error : new Error(String(error)),
}),
);
const finishAcceptedOutcomePromise = finishAcceptedPromise.then(
(payload) => ({ kind: "accepted" as const, payload }),
(error) => {
if (isWaiterTimeoutError(error)) {
return { kind: "accepted_timeout" as const };
}
return {
kind: "accepted_error" as const,
error: error instanceof Error ? error : new Error(String(error)),
};
},
);
const waitForFinalResult = async (
timeoutMs: number,
): Promise<{ dictationId: string; text: string }> => {
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
const outcome = await Promise.race([finalOutcomePromise, errorOutcomePromise]);
if (outcome.kind === "error") {
throw outcome.error;
}
return outcome.payload;
}
let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
const timeoutPromise = new Promise<{ kind: "timeout" }>((resolve) => {
timeoutHandle = setTimeout(() => resolve({ kind: "timeout" }), timeoutMs);
});
const outcome = await Promise.race([
finalOutcomePromise,
errorOutcomePromise,
timeoutPromise,
]);
if (timeoutHandle) {
clearTimeout(timeoutHandle);
}
if (outcome.kind === "timeout") {
throw new Error(`Timeout waiting for dictation finalization (${timeoutMs}ms)`);
}
if (outcome.kind === "error") {
throw outcome.error;
}
return outcome.payload;
};
const cleanupError = new Error("Cancelled dictation finish waiter");
try {
this.sendSessionMessageStrict({ type: "dictation_stream_finish", dictationId, finalSeq });
const firstOutcome = await Promise.race([
finalOutcomePromise,
errorOutcomePromise,
finishAcceptedOutcomePromise,
]);
if (firstOutcome.kind === "final") {
return firstOutcome.payload;
}
if (firstOutcome.kind === "error") {
throw firstOutcome.error;
}
if (firstOutcome.kind === "accepted") {
return await waitForFinalResult(
firstOutcome.payload.timeoutMs + DEFAULT_DICTATION_FINISH_TIMEOUT_GRACE_MS,
);
}
return await waitForFinalResult(DEFAULT_DICTATION_FINISH_FALLBACK_TIMEOUT_MS);
} finally {
final.cancel(cleanupError);
streamError.cancel(cleanupError);
finishAccepted.cancel(cleanupError);
void finalPromise.catch(() => undefined);
void errorPromise.catch(() => undefined);
void finishAcceptedPromise.catch(() => undefined);
}
}
cancelDictationStream(dictationId: string): void {
this.sendSessionMessageStrict({ type: "dictation_stream_cancel", dictationId });
}
async abortRequest(): Promise<void> {
this.sendSessionMessage({ type: "abort_request" });
}
async audioPlayed(id: string): Promise<void> {
this.sendSessionMessage({ type: "audio_played", id });
}
// ============================================================================
// Git Operations
// ============================================================================
async getCheckoutStatus(
cwd: string,
options?: { requestId?: string },
): Promise<CheckoutStatusPayload> {
const requestId = options?.requestId;
if (!requestId) {
const existing = this.checkoutStatusInFlight.get(cwd);
if (existing) {
return existing;
}
}
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "checkout_status_request",
cwd,
requestId: resolvedRequestId,
});
const responsePromise = this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 60000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "checkout_status_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
if (!requestId) {
this.checkoutStatusInFlight.set(cwd, responsePromise);
void responsePromise
.finally(() => {
if (this.checkoutStatusInFlight.get(cwd) === responsePromise) {
this.checkoutStatusInFlight.delete(cwd);
}
})
.catch(() => undefined);
}
return responsePromise;
}
private normalizeCheckoutDiffCompare(compare: {
mode: "uncommitted" | "base";
baseRef?: string;
ignoreWhitespace?: boolean;
}): { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean } {
if (compare.mode === "uncommitted") {
return compare.ignoreWhitespace === true
? { mode: "uncommitted", ignoreWhitespace: true }
: { mode: "uncommitted" };
}
const trimmedBaseRef = compare.baseRef?.trim();
if (!trimmedBaseRef) {
return compare.ignoreWhitespace === true
? { mode: "base", ignoreWhitespace: true }
: { mode: "base" };
}
return compare.ignoreWhitespace === true
? { mode: "base", baseRef: trimmedBaseRef, ignoreWhitespace: true }
: { mode: "base", baseRef: trimmedBaseRef };
}
async getCheckoutDiff(
cwd: string,
compare: { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean },
requestId?: string,
): Promise<CheckoutDiffPayload> {
const oneShotSubscriptionId = `oneshot-checkout-diff:${crypto.randomUUID()}`;
try {
const payload = await this.subscribeCheckoutDiff(cwd, compare, {
subscriptionId: oneShotSubscriptionId,
requestId,
});
return {
cwd: payload.cwd,
files: payload.files,
error: payload.error,
requestId: payload.requestId,
};
} finally {
try {
this.unsubscribeCheckoutDiff(oneShotSubscriptionId);
} catch {
// Ignore disconnect races during one-shot cleanup.
}
}
}
async subscribeCheckoutDiff(
cwd: string,
compare: { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean },
options?: { subscriptionId?: string; requestId?: string },
): Promise<SubscribeCheckoutDiffPayload> {
const subscriptionId = options?.subscriptionId ?? crypto.randomUUID();
const normalizedCompare = this.normalizeCheckoutDiffCompare(compare);
const previousSubscription = this.checkoutDiffSubscriptions.get(subscriptionId) ?? null;
this.checkoutDiffSubscriptions.set(subscriptionId, {
cwd,
compare: normalizedCompare,
});
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "subscribe_checkout_diff_request",
subscriptionId,
cwd,
compare: normalizedCompare,
requestId: resolvedRequestId,
});
try {
return await this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "subscribe_checkout_diff_response",
timeout: 60000,
options: { skipQueue: true },
selectPayload: (payload) => {
if (payload.subscriptionId !== subscriptionId) {
return null;
}
return payload;
},
});
} catch (error) {
if (previousSubscription) {
this.checkoutDiffSubscriptions.set(subscriptionId, previousSubscription);
} else {
this.checkoutDiffSubscriptions.delete(subscriptionId);
}
throw error;
}
}
unsubscribeCheckoutDiff(subscriptionId: string): void {
this.checkoutDiffSubscriptions.delete(subscriptionId);
this.sendSessionMessage({
type: "unsubscribe_checkout_diff_request",
subscriptionId,
});
}
async checkoutCommit(
cwd: string,
input: { message?: string; addAll?: boolean },
requestId?: string,
): Promise<CheckoutCommitPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_commit_request",
cwd,
message: input.message,
addAll: input.addAll,
},
responseType: "checkout_commit_response",
timeout: 60000,
});
}
async checkoutMerge(
cwd: string,
input: { baseRef?: string; strategy?: "merge" | "squash"; requireCleanTarget?: boolean },
requestId?: string,
): Promise<CheckoutMergePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_merge_request",
cwd,
baseRef: input.baseRef,
strategy: input.strategy,
requireCleanTarget: input.requireCleanTarget,
},
responseType: "checkout_merge_response",
timeout: 60000,
});
}
async checkoutMergeFromBase(
cwd: string,
input: { baseRef?: string; requireCleanTarget?: boolean },
requestId?: string,
): Promise<CheckoutMergeFromBasePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_merge_from_base_request",
cwd,
baseRef: input.baseRef,
requireCleanTarget: input.requireCleanTarget,
},
responseType: "checkout_merge_from_base_response",
timeout: 60000,
});
}
async checkoutPull(cwd: string, requestId?: string): Promise<CheckoutPullPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_pull_request",
cwd,
},
responseType: "checkout_pull_response",
timeout: 60000,
});
}
async checkoutPush(cwd: string, requestId?: string): Promise<CheckoutPushPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_push_request",
cwd,
},
responseType: "checkout_push_response",
timeout: 60000,
});
}
async checkoutRefresh(cwd: string, requestId?: string): Promise<CheckoutRefreshPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout.refresh.request",
cwd,
},
responseType: "checkout.refresh.response",
timeout: 60000,
});
}
async checkoutPrCreate(
cwd: string,
input: { title?: string; body?: string; baseRef?: string },
requestId?: string,
): Promise<CheckoutPrCreatePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_pr_create_request",
cwd,
title: input.title,
body: input.body,
baseRef: input.baseRef,
},
responseType: "checkout_pr_create_response",
timeout: 60000,
});
}
async checkoutPrMerge(
cwd: string,
input: { method: CheckoutPrMergeMethod },
requestId?: string,
): Promise<CheckoutPrMergePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_pr_merge_request",
cwd,
mergeMethod: input.method,
},
responseType: "checkout_pr_merge_response",
timeout: 60000,
});
}
async checkoutGithubSetAutoMerge(
cwd: string,
input: { enabled: true; method: CheckoutPrMergeMethod } | { enabled: false },
requestId?: string,
): Promise<CheckoutGithubSetAutoMergePayload> {
return this.sendNamespacedCorrelatedSessionRequest<"checkout.github.set_auto_merge.response">({
requestId,
message: {
type: "checkout.github.set_auto_merge.request",
cwd,
enabled: input.enabled,
...(input.enabled ? { mergeMethod: input.method } : {}),
},
timeout: 60000,
});
}
async checkoutGithubGetCheckDetails(
input: {
cwd: string;
repoOwner: string;
repoName: string;
checkRunId: number;
workflowRunId?: number;
},
requestId?: string,
): Promise<CheckoutGithubGetCheckDetailsPayload> {
return this.sendNamespacedCorrelatedSessionRequest<"checkout.github.get_check_details.response">(
{
requestId,
message: {
type: "checkout.github.get_check_details.request",
cwd: input.cwd,
repoOwner: input.repoOwner,
repoName: input.repoName,
checkRunId: input.checkRunId,
workflowRunId: input.workflowRunId,
},
timeout: 60000,
},
);
}
async checkoutPrStatus(cwd: string, requestId?: string): Promise<CheckoutPrStatusPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_pr_status_request",
cwd,
},
responseType: "checkout_pr_status_response",
timeout: 60000,
});
}
async pullRequestTimeline(
input: { cwd: string; prNumber: number; repoOwner: string; repoName: string },
requestId?: string,
): Promise<PullRequestTimelinePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "pull_request_timeline_request",
cwd: input.cwd,
prNumber: input.prNumber,
repoOwner: input.repoOwner,
repoName: input.repoName,
},
responseType: "pull_request_timeline_response",
timeout: 60000,
});
}
async checkoutSwitchBranch(
cwd: string,
branch: string,
requestId?: string,
): Promise<CheckoutSwitchBranchPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "checkout_switch_branch_request",
cwd,
branch,
},
responseType: "checkout_switch_branch_response",
timeout: 30000,
});
}
async renameBranch(input: RenameBranchInput): Promise<RenameBranchResult> {
return this.sendCorrelatedSessionRequest({
requestId: input.requestId,
message: {
type: "checkout.rename_branch.request",
cwd: input.cwd,
branch: input.branch,
},
responseType: "checkout.rename_branch.response",
timeout: 30000,
});
}
async stashSave(
cwd: string,
options?: { branch?: string },
requestId?: string,
): Promise<StashSavePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "stash_save_request",
cwd,
branch: options?.branch,
},
responseType: "stash_save_response",
timeout: 30000,
});
}
async stashPop(cwd: string, stashIndex: number, requestId?: string): Promise<StashPopPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "stash_pop_request",
cwd,
stashIndex,
},
responseType: "stash_pop_response",
timeout: 30000,
});
}
async stashList(
cwd: string,
options?: { paseoOnly?: boolean },
requestId?: string,
): Promise<StashListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "stash_list_request",
cwd,
paseoOnly: options?.paseoOnly,
},
responseType: "stash_list_response",
timeout: 10000,
});
}
async getPaseoWorktreeList(
input: { cwd?: string; repoRoot?: string },
requestId?: string,
): Promise<PaseoWorktreeListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "paseo_worktree_list_request",
cwd: input.cwd,
repoRoot: input.repoRoot,
},
responseType: "paseo_worktree_list_response",
timeout: 60000,
});
}
async archivePaseoWorktree(
input: {
worktreePath?: string;
repoRoot?: string;
branchName?: string;
workspaceId?: string;
scope?: "workspace" | "worktree";
},
requestId?: string,
): Promise<PaseoWorktreeArchivePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "paseo_worktree_archive_request",
worktreePath: input.worktreePath,
repoRoot: input.repoRoot,
branchName: input.branchName,
...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}),
...(input.scope !== undefined ? { scope: input.scope } : {}),
},
responseType: "paseo_worktree_archive_response",
timeout: 60000,
});
}
async createPaseoWorktree(
input: CreatePaseoWorktreeInput,
requestId?: string,
): Promise<CreatePaseoWorktreePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "create_paseo_worktree_request",
cwd: input.cwd,
...(input.projectId !== undefined ? { projectId: input.projectId } : {}),
worktreeSlug: input.worktreeSlug,
...(input.firstAgentContext !== undefined
? { firstAgentContext: input.firstAgentContext }
: {}),
...(input.refName !== undefined ? { refName: input.refName } : {}),
...(input.action !== undefined ? { action: input.action } : {}),
...(input.githubPrNumber !== undefined ? { githubPrNumber: input.githubPrNumber } : {}),
},
responseType: "create_paseo_worktree_response",
timeout: 60000,
});
}
async createWorkspace(
input: {
source: WorkspaceCreateRequest["source"];
title?: string;
firstAgentContext?: WorkspaceCreateRequest["firstAgentContext"];
},
requestId?: string,
): Promise<WorkspaceCreatePayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "workspace.create.request",
source: input.source,
...(input.title !== undefined ? { title: input.title } : {}),
...(input.firstAgentContext !== undefined
? { firstAgentContext: input.firstAgentContext }
: {}),
},
responseType: "workspace.create.response",
timeout: 60000,
});
}
async validateBranch(
options: { cwd: string; branchName: string },
requestId?: string,
): Promise<ValidateBranchPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "validate_branch_request",
cwd: options.cwd,
branchName: options.branchName,
},
responseType: "validate_branch_response",
timeout: 10000,
});
}
async getBranchSuggestions(
options: { cwd: string; query?: string; limit?: number },
requestId?: string,
): Promise<BranchSuggestionsPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "branch_suggestions_request",
cwd: options.cwd,
query: options.query,
limit: options.limit,
},
responseType: "branch_suggestions_response",
timeout: 10000,
});
}
async searchGitHub(
options: { cwd: string; query: string; limit?: number; kinds?: GitHubSearchRequest["kinds"] },
requestId?: string,
): Promise<GitHubSearchPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "github_search_request",
cwd: options.cwd,
query: options.query,
limit: options.limit,
kinds: options.kinds,
},
responseType: "github_search_response",
timeout: 15000,
});
}
async getDirectorySuggestions(
options: {
query: string;
limit?: number;
cwd?: string;
includeFiles?: boolean;
includeDirectories?: boolean;
matchMode?: "fuzzy" | "suffix";
},
requestId?: string,
): Promise<DirectorySuggestionsPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "directory_suggestions_request",
query: options.query,
cwd: options.cwd,
includeFiles: options.includeFiles,
includeDirectories: options.includeDirectories,
matchMode: options.matchMode,
limit: options.limit,
},
responseType: "directory_suggestions_response",
timeout: 10000,
});
}
// ============================================================================
// File Explorer
// ============================================================================
private async requestFileExplorer(
cwd: string,
path: string,
mode: "list" | "file",
requestId?: string,
acceptBinary = false,
): Promise<FileExplorerPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "file_explorer_request",
cwd,
path,
mode,
...(acceptBinary ? { acceptBinary: true } : {}),
},
responseType: "file_explorer_response",
timeout: 10000,
});
}
async listDirectory(
cwd: string,
path: string,
requestId?: string,
): Promise<FileExplorerDirectoryPayload> {
const payload = await this.requestFileExplorer(cwd, path, "list", requestId);
if (payload.error) {
throw new Error(payload.error);
}
if (!payload.directory) {
throw new Error("Directory listing unavailable.");
}
return payload.directory;
}
async readFile(cwd: string, path: string, requestId?: string): Promise<FileReadResult> {
const resolvedRequestId = this.createRequestId(requestId);
this.pendingBinaryFileReads.set(resolvedRequestId, { cwd, path });
try {
const payload = await this.requestFileExplorer(cwd, path, "file", resolvedRequestId, true);
if (payload.error) {
throw new Error(payload.error);
}
const binaryResult = this.completedBinaryFileReads.get(resolvedRequestId);
if (binaryResult) {
this.completedBinaryFileReads.delete(resolvedRequestId);
return binaryResult;
}
if (!payload.file) {
throw new Error("File unavailable.");
}
return legacyExplorerFileToBytes(payload.file);
} finally {
this.pendingBinaryFileReads.delete(resolvedRequestId);
this.activeBinaryFileTransfers.delete(resolvedRequestId);
}
}
async uploadFile(input: FileUploadInput): Promise<FileUploadResult> {
const bytes = asUint8Array(input.bytes);
if (!bytes) {
throw new Error("File bytes are required.");
}
const resolvedRequestId = this.createRequestId(input.requestId);
const modifiedAt = input.modifiedAt ?? new Date().toISOString();
const responsePromise = this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message: {
type: "file.upload.request",
fileName: input.fileName,
mimeType: input.mimeType,
size: bytes.byteLength,
modifiedAt,
requestId: resolvedRequestId,
},
responseType: "file.upload.response",
timeout: 60000,
options: { skipQueue: true },
});
this.sendBinaryFrame(
encodeFileTransferFrame({
opcode: FileTransferOpcode.FileBegin,
requestId: resolvedRequestId,
metadata: {
mime: input.mimeType,
size: bytes.byteLength,
encoding: "binary",
modifiedAt,
fileName: input.fileName,
},
}),
);
const chunkSize = input.chunkSize ?? 1024 * 1024;
for (let offset = 0; offset < bytes.byteLength; offset += chunkSize) {
this.sendBinaryFrame(
encodeFileTransferFrame({
opcode: FileTransferOpcode.FileChunk,
requestId: resolvedRequestId,
payload: bytes.subarray(offset, Math.min(offset + chunkSize, bytes.byteLength)),
}),
);
}
this.sendBinaryFrame(
encodeFileTransferFrame({
opcode: FileTransferOpcode.FileEnd,
requestId: resolvedRequestId,
}),
);
return responsePromise;
}
async requestDownloadToken(
cwd: string,
path: string,
requestId?: string,
): Promise<FileDownloadTokenPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "file_download_token_request",
cwd,
path,
},
responseType: "file_download_token_response",
timeout: 10000,
});
}
async requestProjectIcon(
cwd: string,
requestId?: string,
): Promise<ProjectIconResponse["payload"]> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "project_icon_request",
cwd,
},
responseType: "project_icon_response",
timeout: 10000,
});
}
// ============================================================================
// Provider Models / Commands
// ============================================================================
async listProviderModels(
provider: AgentProvider,
options?: { cwd?: string; requestId?: string },
): Promise<ListProviderModelsPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "list_provider_models_request",
provider,
cwd: options?.cwd,
},
responseType: "list_provider_models_response",
// Provider SDK cold starts (especially model discovery) can exceed 30s.
timeout: 45000,
});
}
async listProviderModes(
provider: AgentProvider,
options?: { cwd?: string; requestId?: string },
): Promise<ListProviderModesPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "list_provider_modes_request",
provider,
cwd: options?.cwd,
},
responseType: "list_provider_modes_response",
timeout: 45000,
});
}
async listProviderFeatures(
draftConfig: ListCommandsDraftConfig,
options?: { requestId?: string },
): Promise<ListProviderFeaturesPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "list_provider_features_request",
draftConfig,
},
responseType: "list_provider_features_response",
timeout: 45000,
});
}
async listAvailableProviders(options?: {
requestId?: string;
}): Promise<ListAvailableProvidersPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "list_available_providers_request",
},
responseType: "list_available_providers_response",
timeout: 30000,
});
}
async getProvidersSnapshot(options?: {
cwd?: string;
requestId?: string;
}): Promise<GetProvidersSnapshotPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "get_providers_snapshot_request",
cwd: options?.cwd,
},
responseType: "get_providers_snapshot_response",
timeout: 10000,
});
}
async getDaemonConfig(
requestId?: string,
): Promise<{ requestId: string; config: MutableDaemonConfig }> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "get_daemon_config_request",
},
responseType: "get_daemon_config_response",
timeout: 10000,
});
}
async getDaemonStatus(requestId?: string): Promise<DaemonStatusPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "daemon.get_status.request",
},
responseType: "daemon.get_status.response",
timeout: 10000,
});
}
async getDaemonPairingOffer(requestId?: string): Promise<DaemonPairingOfferPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "daemon.get_pairing_offer.request",
},
responseType: "daemon.get_pairing_offer.response",
timeout: 10000,
});
}
async patchDaemonConfig(
config: MutableDaemonConfigPatch,
requestId?: string,
): Promise<{ requestId: string; config: MutableDaemonConfig }> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "set_daemon_config_request",
config,
},
responseType: "set_daemon_config_response",
timeout: 10000,
});
}
async readProjectConfig(repoRoot: string, requestId?: string): Promise<ReadProjectConfigPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "read_project_config_request",
repoRoot,
},
responseType: "read_project_config_response",
timeout: 10000,
});
}
async writeProjectConfig(input: WriteProjectConfigInput): Promise<WriteProjectConfigPayload> {
return this.sendCorrelatedSessionRequest({
requestId: input.requestId,
message: {
type: "write_project_config_request",
repoRoot: input.repoRoot,
config: input.config,
expectedRevision: input.expectedRevision,
},
responseType: "write_project_config_response",
timeout: 10000,
});
}
async refreshProvidersSnapshot(options?: {
cwd?: string;
providers?: AgentProvider[];
requestId?: string;
}): Promise<RefreshProvidersSnapshotPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "refresh_providers_snapshot_request",
cwd: options?.cwd,
providers: options?.providers,
},
responseType: "refresh_providers_snapshot_response",
timeout: 60000,
});
}
async getProviderDiagnostic(
provider: AgentProvider,
options?: { requestId?: string },
): Promise<ProviderDiagnosticPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options?.requestId,
message: {
type: "provider_diagnostic_request",
provider,
},
responseType: "provider_diagnostic_response",
timeout: 30000,
});
}
async listCommands(agentId: string, requestId?: string): Promise<ListCommandsPayload>;
async listCommands(agentId: string, options?: ListCommandsOptions): Promise<ListCommandsPayload>;
async listCommands(
agentId: string,
requestIdOrOptions?: string | ListCommandsOptions,
): Promise<ListCommandsPayload> {
const requestId =
typeof requestIdOrOptions === "string" ? requestIdOrOptions : requestIdOrOptions?.requestId;
const draftConfig =
typeof requestIdOrOptions === "string" ? undefined : requestIdOrOptions?.draftConfig;
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "list_commands_request",
agentId,
...(draftConfig ? { draftConfig } : {}),
},
responseType: "list_commands_response",
timeout: 30000,
});
}
// ============================================================================
// Permissions
// ============================================================================
async respondToPermission(
agentId: string,
requestId: string,
response: AgentPermissionResponse,
): Promise<void> {
this.sendSessionMessage({
type: "agent_permission_response",
agentId,
requestId,
response,
});
}
async respondToPermissionAndWait(
agentId: string,
requestId: string,
response: AgentPermissionResponse,
timeout = 15000,
): Promise<AgentPermissionResolvedPayload> {
const message = SessionInboundMessageSchema.parse({
type: "agent_permission_response",
agentId,
requestId,
response,
});
return this.sendRequest({
requestId,
message,
timeout,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "agent_permission_resolved") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
if (msg.payload.agentId !== agentId) {
return null;
}
return msg.payload;
},
});
}
// ============================================================================
// Waiting / Streaming Helpers
// ============================================================================
async waitForAgentUpsert(
agentId: string,
predicate: (snapshot: AgentSnapshotPayload) => boolean,
timeout = 60000,
): Promise<AgentSnapshotPayload> {
const initialResult = await this.fetchAgent(agentId).catch(() => null);
if (initialResult && predicate(initialResult.agent)) {
return initialResult.agent;
}
const deadline = Date.now() + timeout;
return await new Promise<AgentSnapshotPayload>((resolve, reject) => {
let settled = false;
let pollInFlight = false;
let pollTimer: ReturnType<typeof setInterval> | null = null;
let timeoutTimer: ReturnType<typeof setTimeout> | null = null;
let unsubscribe: (() => void) | null = null;
const finish = (
result: { kind: "ok"; snapshot: AgentSnapshotPayload } | { kind: "error"; error: Error },
) => {
if (settled) {
return;
}
settled = true;
if (timeoutTimer) {
clearTimeout(timeoutTimer);
timeoutTimer = null;
}
if (pollTimer) {
clearInterval(pollTimer);
pollTimer = null;
}
if (unsubscribe) {
unsubscribe();
unsubscribe = null;
}
if (result.kind === "ok") {
resolve(result.snapshot);
return;
}
reject(result.error);
};
const maybeResolve = (snapshot: AgentSnapshotPayload | null) => {
if (!snapshot) {
return false;
}
if (!predicate(snapshot)) {
return false;
}
finish({ kind: "ok", snapshot });
return true;
};
const poll = async () => {
if (settled || pollInFlight) {
return;
}
pollInFlight = true;
try {
const result = await this.fetchAgent(agentId).catch(() => null);
maybeResolve(result?.agent ?? null);
} finally {
pollInFlight = false;
}
};
unsubscribe = this.on("agent_update", (message) => {
if (settled) {
return;
}
if (message.payload.kind !== "upsert") {
return;
}
const snapshot = message.payload.agent;
if (snapshot.id !== agentId) {
return;
}
maybeResolve(snapshot);
});
const remaining = Math.max(1, deadline - Date.now());
timeoutTimer = setTimeout(() => {
finish({
kind: "error",
error: new Error(`Timed out waiting for agent ${agentId}`),
});
}, remaining);
pollTimer = setInterval(() => {
void poll();
}, 250);
void poll();
});
}
async waitForFinish(agentId: string, timeout = 60000): Promise<WaitForFinishResult> {
const requestId = this.createRequestId();
const hasTimeout = Number.isFinite(timeout) && timeout > 0;
const message = SessionInboundMessageSchema.parse({
type: "wait_for_finish_request",
requestId,
agentId,
...(hasTimeout ? { timeoutMs: timeout } : {}),
});
const payload = await this.sendCorrelatedRequest({
requestId,
message,
responseType: "wait_for_finish_response",
timeout: hasTimeout ? timeout + 5000 : 0,
options: { skipQueue: true },
});
return {
status: payload.status,
final: payload.final,
error: payload.error,
lastMessage: payload.lastMessage,
};
}
// ============================================================================
// Terminals
// ============================================================================
subscribeTerminals(input: { cwd: string; workspaceId?: string }): void {
this.terminalDirectorySubscriptions.set(terminalSubscriptionKey(input.cwd, input.workspaceId), {
cwd: input.cwd,
workspaceId: input.workspaceId,
});
if (!this.transport || this.connectionState.status !== "connected") {
return;
}
this.sendSessionMessage({
type: "subscribe_terminals_request",
cwd: input.cwd,
...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}),
});
}
unsubscribeTerminals(input: { cwd: string; workspaceId?: string }): void {
this.terminalDirectorySubscriptions.delete(
terminalSubscriptionKey(input.cwd, input.workspaceId),
);
if (!this.transport || this.connectionState.status !== "connected") {
return;
}
this.sendSessionMessage({
type: "unsubscribe_terminals_request",
cwd: input.cwd,
...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}),
});
}
async listTerminals(
cwd?: string,
requestId?: string,
options?: { workspaceId?: string },
): Promise<ListTerminalsPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "list_terminals_request",
...(cwd === undefined ? {} : { cwd }),
...(options?.workspaceId !== undefined ? { workspaceId: options.workspaceId } : {}),
requestId: resolvedRequestId,
});
return this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "list_terminals_response",
timeout: 10000,
options: { skipQueue: true },
});
}
async createTerminal(
cwd: string,
name?: string,
requestId?: string,
options?: { agentId?: string; command?: string; args?: string[]; workspaceId?: string },
): Promise<CreateTerminalPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "create_terminal_request",
cwd,
name,
agentId: options?.agentId,
command: options?.command,
args: options?.args,
...(options?.workspaceId !== undefined ? { workspaceId: options.workspaceId } : {}),
requestId: resolvedRequestId,
});
return this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "create_terminal_response",
timeout: 10000,
options: { skipQueue: true },
});
}
async renameTerminal(input: RenameTerminalInput): Promise<RenameTerminalResult> {
return this.sendCorrelatedSessionRequest({
requestId: input.requestId,
message: {
type: "terminal.rename.request",
terminalId: input.terminalId,
title: input.title,
},
responseType: "terminal.rename.response",
timeout: 10000,
});
}
async subscribeTerminal(
terminalId: string,
optionsOrRequestId?:
| { restore?: SubscribeTerminalRequest["restore"]; requestId?: string }
| string,
): Promise<SubscribeTerminalPayload> {
const restore = typeof optionsOrRequestId === "object" ? optionsOrRequestId.restore : undefined;
const requestId =
typeof optionsOrRequestId === "object" ? optionsOrRequestId.requestId : optionsOrRequestId;
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "subscribe_terminal_request",
terminalId,
requestId: resolvedRequestId,
...(restore ? { restore } : {}),
});
const payload = await this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "subscribe_terminal_response",
timeout: 10000,
options: { skipQueue: true },
});
if (payload.error === null) {
this.terminalStreams.setSlot(terminalId, payload.slot);
}
return payload;
}
unsubscribeTerminal(terminalId: string): void {
this.terminalStreams.removeTerminal(terminalId);
this.sendSessionMessage({
type: "unsubscribe_terminal_request",
terminalId,
});
}
sendTerminalInput(terminalId: string, message: TerminalInput["message"]): void {
const frame = this.terminalStreams.encodeInput(terminalId, message);
if (frame) {
this.sendBinaryFrame(frame);
return;
}
this.sendSessionMessage({
type: "terminal_input",
terminalId,
message,
});
}
async killTerminal(terminalId: string, requestId?: string): Promise<KillTerminalPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "kill_terminal_request",
terminalId,
requestId: resolvedRequestId,
});
return this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "kill_terminal_response",
timeout: 10000,
options: { skipQueue: true },
});
}
async closeItems(
input: { agentIds?: string[]; terminalIds?: string[] },
requestId?: string,
): Promise<CloseItemsPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "close_items_request",
agentIds: input.agentIds ?? [],
terminalIds: input.terminalIds ?? [],
requestId: resolvedRequestId,
});
return this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "close_items_response",
timeout: 10000,
options: { skipQueue: true },
});
}
async captureTerminal(
terminalId: string,
options?: { start?: number; end?: number; stripAnsi?: boolean },
requestId?: string,
): Promise<CaptureTerminalPayload> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
type: "capture_terminal_request",
terminalId,
...(options?.start === undefined ? {} : { start: options.start }),
...(options?.end === undefined ? {} : { end: options.end }),
...(options?.stripAnsi === undefined ? {} : { stripAnsi: options.stripAnsi }),
requestId: resolvedRequestId,
});
return this.sendCorrelatedRequest({
requestId: resolvedRequestId,
message,
responseType: "capture_terminal_response",
timeout: 10000,
options: { skipQueue: true },
});
}
async createChatRoom(options: CreateChatRoomOptions): Promise<ChatCreatePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/create",
name: options.name,
...(options.purpose ? { purpose: options.purpose } : {}),
},
responseType: "chat/create/response",
timeout: 10000,
});
}
async listChatRooms(requestId?: string): Promise<ChatListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "chat/list",
},
responseType: "chat/list/response",
timeout: 10000,
});
}
async inspectChatRoom(options: InspectChatRoomOptions): Promise<ChatInspectPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/inspect",
room: options.room,
},
responseType: "chat/inspect/response",
timeout: 10000,
});
}
async deleteChatRoom(options: DeleteChatRoomOptions): Promise<ChatDeletePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/delete",
room: options.room,
},
responseType: "chat/delete/response",
timeout: 10000,
});
}
async postChatMessage(options: PostChatMessageOptions): Promise<ChatPostPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/post",
room: options.room,
body: options.body,
...(options.authorAgentId ? { authorAgentId: options.authorAgentId } : {}),
...(options.replyToMessageId ? { replyToMessageId: options.replyToMessageId } : {}),
},
responseType: "chat/post/response",
timeout: 10000,
});
}
async readChatMessages(options: ReadChatMessagesOptions): Promise<ChatReadPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/read",
room: options.room,
...(typeof options.limit === "number" ? { limit: options.limit } : {}),
...(options.since ? { since: options.since } : {}),
...(options.authorAgentId ? { authorAgentId: options.authorAgentId } : {}),
},
responseType: "chat/read/response",
timeout: 10000,
});
}
async waitForChatMessages(options: WaitForChatMessagesOptions): Promise<ChatWaitPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/wait",
room: options.room,
...(options.afterMessageId ? { afterMessageId: options.afterMessageId } : {}),
...(typeof options.timeoutMs === "number" ? { timeoutMs: options.timeoutMs } : {}),
},
responseType: "chat/wait/response",
timeout: (options.timeoutMs ?? 0) + 10000,
});
}
async scheduleCreate(options: CreateScheduleOptions): Promise<ScheduleCreatePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/create",
prompt: options.prompt,
cadence: options.cadence,
target: options.target,
...(options.name ? { name: options.name } : {}),
...(typeof options.maxRuns === "number" ? { maxRuns: options.maxRuns } : {}),
...(options.expiresAt ? { expiresAt: options.expiresAt } : {}),
...(typeof options.runOnCreate === "boolean" ? { runOnCreate: options.runOnCreate } : {}),
},
responseType: "schedule/create/response",
timeout: 10000,
});
}
async scheduleList(requestId?: string): Promise<ScheduleListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "schedule/list",
},
responseType: "schedule/list/response",
timeout: 10000,
});
}
async scheduleInspect(options: InspectScheduleOptions): Promise<ScheduleInspectPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/inspect",
scheduleId: options.id,
},
responseType: "schedule/inspect/response",
timeout: 10000,
});
}
async scheduleLogs(options: InspectScheduleOptions): Promise<ScheduleLogsPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/logs",
scheduleId: options.id,
},
responseType: "schedule/logs/response",
timeout: 10000,
});
}
async schedulePause(options: InspectScheduleOptions): Promise<SchedulePausePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/pause",
scheduleId: options.id,
},
responseType: "schedule/pause/response",
timeout: 10000,
});
}
async scheduleResume(options: InspectScheduleOptions): Promise<ScheduleResumePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/resume",
scheduleId: options.id,
},
responseType: "schedule/resume/response",
timeout: 10000,
});
}
async scheduleDelete(options: InspectScheduleOptions): Promise<ScheduleDeletePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/delete",
scheduleId: options.id,
},
responseType: "schedule/delete/response",
timeout: 10000,
});
}
async scheduleRunOnce(options: InspectScheduleOptions): Promise<ScheduleRunOncePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/run-once",
scheduleId: options.id,
},
responseType: "schedule/run-once/response",
timeout: 10000,
});
}
async scheduleUpdate(options: UpdateScheduleOptions): Promise<ScheduleUpdatePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/update",
scheduleId: options.id,
...(options.name !== undefined ? { name: options.name } : {}),
...(options.prompt !== undefined ? { prompt: options.prompt } : {}),
...(options.cadence !== undefined ? { cadence: options.cadence } : {}),
...(options.newAgentConfig !== undefined ? { newAgentConfig: options.newAgentConfig } : {}),
...(options.maxRuns !== undefined ? { maxRuns: options.maxRuns } : {}),
...(options.expiresAt !== undefined ? { expiresAt: options.expiresAt } : {}),
},
responseType: "schedule/update/response",
timeout: 10000,
});
}
async loopRun(options: RunLoopOptions): Promise<LoopRunPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "loop/run",
prompt: options.prompt,
cwd: options.cwd,
...(options.provider ? { provider: options.provider } : {}),
...(options.model ? { model: options.model } : {}),
...(options.modeId ? { modeId: options.modeId } : {}),
...(options.verifierProvider ? { verifierProvider: options.verifierProvider } : {}),
...(options.verifierModel ? { verifierModel: options.verifierModel } : {}),
...(options.verifierModeId ? { verifierModeId: options.verifierModeId } : {}),
...(options.verifyPrompt ? { verifyPrompt: options.verifyPrompt } : {}),
...(options.verifyChecks && options.verifyChecks.length > 0
? { verifyChecks: options.verifyChecks }
: {}),
...(options.name ? { name: options.name } : {}),
...(typeof options.sleepMs === "number" ? { sleepMs: options.sleepMs } : {}),
...(typeof options.maxIterations === "number"
? { maxIterations: options.maxIterations }
: {}),
...(typeof options.maxTimeMs === "number" ? { maxTimeMs: options.maxTimeMs } : {}),
},
responseType: "loop/run/response",
timeout: 15000,
});
}
async loopList(requestId?: string): Promise<LoopListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "loop/list",
},
responseType: "loop/list/response",
timeout: 10000,
});
}
async loopInspect(options: string | InspectLoopOptions): Promise<LoopInspectPayload> {
const normalized = typeof options === "string" ? { id: options } : options;
return this.sendCorrelatedSessionRequest({
requestId: normalized.requestId,
message: {
type: "loop/inspect",
id: normalized.id,
},
responseType: "loop/inspect/response",
timeout: 10000,
});
}
async loopLogs(options: string | LoopLogsOptions, afterSeq?: number): Promise<LoopLogsPayload> {
const normalized = typeof options === "string" ? { id: options, afterSeq } : options;
return this.sendCorrelatedSessionRequest({
requestId: normalized.requestId,
message: {
type: "loop/logs",
id: normalized.id,
...(typeof normalized.afterSeq === "number" ? { afterSeq: normalized.afterSeq } : {}),
},
responseType: "loop/logs/response",
timeout: 10000,
});
}
async loopStop(options: string | StopLoopOptions): Promise<LoopStopPayload> {
const normalized = typeof options === "string" ? { id: options } : options;
return this.sendCorrelatedSessionRequest({
requestId: normalized.requestId,
message: {
type: "loop/stop",
id: normalized.id,
},
responseType: "loop/stop/response",
timeout: 10000,
});
}
onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void {
return this.terminalStreams.onEvent(handler);
}
async waitForTerminalStreamEvent(
predicate: (event: TerminalStreamEvent) => boolean,
timeout = 5000,
): Promise<TerminalStreamEvent> {
return new Promise<TerminalStreamEvent>((resolve, reject) => {
const timeoutHandle = setTimeout(() => {
unsubscribe();
reject(new Error(`Timeout waiting for terminal stream event (${timeout}ms)`));
}, timeout);
const unsubscribe = this.onTerminalStreamEvent((event) => {
if (!predicate(event)) {
return;
}
clearTimeout(timeoutHandle);
unsubscribe();
resolve(event);
});
});
}
// ============================================================================
// Internals
// ============================================================================
private createRequestId(requestId?: string): string {
return requestId ?? crypto.randomUUID();
}
getLastServerInfoMessage(): ServerInfoStatusPayload | null {
return this.lastServerInfoMessage;
}
private resolveTransportUrlForAttempt(): string {
return this.config.url;
}
private sendHelloMessage(): void {
if (!this.transport) {
this.scheduleReconnect({
reason: "Transport unavailable before hello",
event: "HELLO_TRANSPORT_MISSING",
reasonCode: "transport_error",
});
return;
}
try {
this.transport.send(
JSON.stringify({
type: "hello",
clientId: this.config.clientId,
clientType: this.config.clientType ?? "cli",
protocolVersion: 1,
capabilities: {
[CLIENT_CAPS.customModeIcons]: true,
[CLIENT_CAPS.reasoningMergeEnum]: true,
[CLIENT_CAPS.terminalReflowableSnapshot]: true,
},
...(this.config.appVersion ? { appVersion: this.config.appVersion } : {}),
}),
);
} catch (error) {
const message = error instanceof Error ? error.message : "Failed to send hello message";
this.lastErrorValue = message;
this.scheduleReconnect({
reason: message,
event: "HELLO_SEND_FAILED",
reasonCode: "transport_error",
});
}
}
private disposeTransport(code = 1001, reason = "Reconnecting"): void {
this.stopLivenessHeartbeat();
this.cleanupTransport();
if (this.transport) {
try {
this.transport.close(code, reason);
} catch {
// no-op
}
this.transport = null;
}
}
private cleanupTransport(): void {
this.resetConnectTimeout();
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout);
this.pendingGenericTransportErrorTimeout = null;
}
for (const cleanup of this.transportCleanup) {
try {
cleanup();
} catch {
// no-op
}
}
this.transportCleanup = [];
}
private resetConnectTimeout(): void {
if (!this.connectTimeout) {
return;
}
clearTimeout(this.connectTimeout);
this.connectTimeout = null;
}
private handleTransportMessage(data: unknown): void {
const rawData =
data && typeof data === "object" && "data" in data ? (data as { data: unknown }).data : data;
if (
typeof Blob !== "undefined" &&
rawData instanceof Blob &&
typeof rawData.arrayBuffer === "function"
) {
void rawData
.arrayBuffer()
.then((buffer) => {
this.handleTransportMessage(buffer);
return;
})
.catch(() => {
// Ignore failed blob decoding and allow reconnect logic to recover.
});
return;
}
const rawBytes = asUint8Array(rawData);
if (rawBytes && this.tryHandleBinaryFrame(rawBytes)) {
return;
}
const payload = decodeMessageData(rawData);
if (!payload) {
return;
}
this.handleJsonPayload(payload, rawBytes?.byteLength);
}
private handleJsonPayload(payload: string, rawBytesLength: number | undefined): void {
const bytes = rawBytesLength ?? payload.length;
const startMs = perfNow();
let parsedJson: unknown;
try {
parsedJson = JSON.parse(payload);
} catch {
return;
}
const parsed = WSOutboundMessageSchema.safeParse(parsedJson);
if (!parsed.success) {
const msgType =
parsedJson != null &&
typeof parsedJson === "object" &&
"type" in parsedJson &&
typeof parsedJson.type === "string"
? parsedJson.type
: "unknown";
this.logger.warn({ msgType, error: parsed.error.message }, "Message validation failed");
return;
}
this.consecutiveLivenessFailures = 0;
if (parsed.data.type === "pong") {
this.resolvePingProbe();
this.runtimeMetrics?.recordMessage("pong", bytes, perfNow() - startMs);
return;
}
this.handleSessionMessage(parsed.data.message);
const msgType = parsed.data.message.type;
this.runtimeMetrics?.recordMessage(msgType, bytes, perfNow() - startMs);
if (parsed.data.message.type === "agent_stream") {
this.runtimeMetrics?.recordAgentStream(parsed.data.message.payload);
}
}
private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean {
const fileFrame = decodeFileTransferFrame(rawBytes);
if (fileFrame) {
this.consecutiveLivenessFailures = 0;
this.handleFileTransferFrame(fileFrame);
this.runtimeMetrics?.recordBinaryFrame("other", rawBytes.byteLength, 0);
return true;
}
const frame = decodeTerminalStreamFrame(rawBytes);
if (!frame) {
return false;
}
this.consecutiveLivenessFailures = 0;
const binaryStartMs = perfNow();
this.terminalStreams.handleFrame(frame);
let frameKind: "output" | "snapshot" | "other" = "other";
if (frame.opcode === TerminalStreamOpcode.Output) {
frameKind = "output";
} else if (frame.opcode === TerminalStreamOpcode.Snapshot) {
frameKind = "snapshot";
} else if (frame.opcode === TerminalStreamOpcode.Restore) {
frameKind = "output";
}
this.runtimeMetrics?.recordBinaryFrame(
frameKind,
rawBytes.byteLength,
perfNow() - binaryStartMs,
);
return true;
}
private handleFileTransferFrame(frame: FileTransferFrame): void {
if (frame.opcode === FileTransferOpcode.FileBegin) {
const pending = this.pendingBinaryFileReads.get(frame.requestId);
if (!pending) {
return;
}
this.activeBinaryFileTransfers.set(frame.requestId, {
...pending,
mime: frame.metadata.mime,
size: frame.metadata.size,
encoding: frame.metadata.encoding,
modifiedAt: frame.metadata.modifiedAt,
chunks: [],
});
return;
}
const transfer = this.activeBinaryFileTransfers.get(frame.requestId);
if (!transfer) {
return;
}
if (frame.opcode === FileTransferOpcode.FileChunk) {
transfer.chunks.push(frame.payload);
return;
}
const bytes = concatByteChunks(transfer.chunks, transfer.size);
this.activeBinaryFileTransfers.delete(frame.requestId);
this.completedBinaryFileReads.set(frame.requestId, {
bytes,
mime: transfer.mime,
size: transfer.size,
path: transfer.path,
kind: binaryFileKind(transfer.mime, transfer.encoding),
modifiedAt: transfer.modifiedAt,
});
this.handleSessionMessage({
type: "file_explorer_response",
payload: {
cwd: transfer.cwd,
path: transfer.path,
mode: "file",
directory: null,
file: null,
error: null,
requestId: frame.requestId,
},
});
}
private updateConnectionState(
next: ConnectionState,
metadata?: { event: string; reason?: string; reasonCode?: string },
): void {
const previous = this.connectionState;
this.connectionState = next;
const reasonFromNext =
next.status === "disconnected" && typeof next.reason === "string" ? next.reason : null;
const reason = metadata?.reason ?? reasonFromNext;
const reasonCode = metadata?.reasonCode ?? toReasonCode(reason);
this.logger.debug(
{
serverId: this.logServerId,
clientIdHash: this.logClientIdHash,
from: previous.status,
to: next.status,
event: metadata?.event ?? "STATE_UPDATE",
connectionPath: this.logConnectionPath,
generation: this.logGeneration,
reasonCode,
reason,
},
"DaemonClientTransition",
);
for (const listener of this.connectionListeners) {
try {
listener(next);
} catch {
// no-op
}
}
}
setReconnectEnabled(enabled: boolean): void {
this.config = { ...this.config, reconnect: { ...this.config.reconnect, enabled } };
}
private scheduleReconnect(input?: {
reason?: string;
event?: string;
reasonCode?: string;
}): void {
if (this.reconnectTimeout) {
clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
}
const wasDisposed = this.connectionState.status === "disposed";
const reason = input?.reason;
if (typeof reason === "string" && reason.trim().length > 0) {
this.lastErrorValue = reason.trim();
}
// Clear all pending waiters and queued sends since the connection was lost
// and responses from the previous connection will never arrive.
this.clearWaiters(new Error(reason ?? "Connection lost"));
this.rejectPendingSendQueue(new Error(reason ?? "Connection lost"));
this.rejectPingProbe(new Error(reason ?? "Connection lost"));
this.terminalStreams.clearSlots();
this.lastServerInfoMessage = null;
if (wasDisposed) {
this.rejectConnect(new Error(reason ?? "Daemon client is disposed"));
return;
}
this.emitDisconnectedStateForReconnect(reason, input);
if (!this.shouldReconnect || this.config.reconnect?.enabled === false) {
this.rejectConnect(new Error(reason ?? "Transport disconnected before connect"));
return;
}
this.armReconnectTimer();
}
private emitDisconnectedStateForReconnect(
reason: string | undefined,
input: { reason?: string; event?: string; reasonCode?: string } | undefined,
): void {
this.updateConnectionState(
{
status: "disconnected",
...(reason ? { reason } : {}),
},
{
event: input?.event ?? "TRANSPORT_CLOSE",
...(reason ? { reason } : {}),
...(input?.reasonCode ? { reasonCode: input.reasonCode } : {}),
},
);
}
private armReconnectTimer(): void {
const attempt = this.reconnectAttempt;
const baseDelay = this.config.reconnect?.baseDelayMs ?? DEFAULT_RECONNECT_BASE_DELAY_MS;
const maxDelay = this.config.reconnect?.maxDelayMs ?? DEFAULT_RECONNECT_MAX_DELAY_MS;
const delay = Math.min(baseDelay * 2 ** attempt, maxDelay);
this.reconnectAttempt = attempt + 1;
this.reconnectTimeout = setTimeout(() => {
this.reconnectTimeout = null;
if (!this.shouldReconnect) {
return;
}
this.attemptConnect();
}, delay);
}
private resolvePingProbe(): void {
const probe = this.pingProbe;
if (!probe) {
return;
}
this.pingProbe = null;
clearTimeout(probe.timeoutHandle);
probe.resolve(perfNow() - probe.startedAt);
}
private clearPingProbe(): void {
const probe = this.pingProbe;
if (!probe) {
return;
}
this.pingProbe = null;
clearTimeout(probe.timeoutHandle);
}
private rejectPingProbe(error: Error): void {
const probe = this.pingProbe;
if (!probe) {
return;
}
this.pingProbe = null;
clearTimeout(probe.timeoutHandle);
probe.reject(error);
}
private recordLivenessFailure(error: Error): void {
this.consecutiveLivenessFailures += 1;
if (this.consecutiveLivenessFailures < LIVENESS_FAILURE_RECONNECT_THRESHOLD) {
return;
}
this.consecutiveLivenessFailures = 0;
this.lastErrorValue = error.message;
this.disposeTransport(1001, "Liveness check timed out");
this.scheduleReconnect({
reason: error.message,
event: "LIVENESS_TIMEOUT",
reasonCode: "liveness_timeout",
});
}
private handleSessionMessage(msg: SessionOutboundMessage): void {
if (msg.type === "status") {
const serverInfo = parseServerInfoStatusPayload(msg.payload);
if (serverInfo) {
this.lastServerInfoMessage = serverInfo;
if (this.connectionState.status === "connecting") {
this.resetConnectTimeout();
this.reconnectAttempt = 0;
this.updateConnectionState({ status: "connected" }, { event: "HELLO_SERVER_INFO" });
this.startLivenessHeartbeat();
this.resubscribeCheckoutDiffSubscriptions();
this.resubscribeTerminalDirectorySubscriptions();
this.flushPendingSendQueue();
this.resolveConnect();
}
}
}
if (msg.type === "terminal_stream_exit") {
this.terminalStreams.removeTerminal(msg.payload.terminalId);
}
if (this.rawMessageListeners.size > 0) {
for (const handler of this.rawMessageListeners) {
try {
handler(msg);
} catch {
// no-op
}
}
}
const handlers = this.messageHandlers.get(msg.type);
if (handlers) {
for (const handler of handlers) {
try {
handler(msg);
} catch {
// no-op
}
}
}
const event = this.toEvent(msg);
if (event) {
for (const handler of this.eventListeners) {
handler(event);
}
}
this.resolveWaiters(msg);
}
private resolveWaiters(msg: SessionOutboundMessage): void {
for (const waiter of Array.from(this.waiters)) {
const result = waiter.predicate(msg);
if (result !== null) {
this.waiters.delete(waiter);
if (waiter.timeoutHandle) {
clearTimeout(waiter.timeoutHandle);
}
waiter.resolve(result);
}
}
}
private clearWaiters(error: Error): void {
for (const waiter of Array.from(this.waiters)) {
if (waiter.timeoutHandle) {
clearTimeout(waiter.timeoutHandle);
}
waiter.reject(error);
}
this.waiters.clear();
}
private toEvent(msg: SessionOutboundMessage): DaemonEvent | null {
switch (msg.type) {
case "agent_update":
return {
type: "agent_update",
agentId: msg.payload.kind === "upsert" ? msg.payload.agent.id : msg.payload.agentId,
payload: msg.payload,
};
case "workspace_update":
return {
type: "workspace_update",
workspaceId: msg.payload.kind === "upsert" ? msg.payload.workspace.id : msg.payload.id,
payload: msg.payload,
};
case "workspace_setup_progress":
return {
type: "workspace_setup_progress",
workspaceId: msg.payload.workspaceId,
payload: msg.payload,
};
case "agent_stream":
return {
type: "agent_stream",
agentId: msg.payload.agentId,
event: msg.payload.event,
timestamp: msg.payload.timestamp,
...(typeof msg.payload.seq === "number" ? { seq: msg.payload.seq } : {}),
...(typeof msg.payload.epoch === "string" ? { epoch: msg.payload.epoch } : {}),
};
case "status":
return { type: "status", payload: msg.payload };
case "agent_deleted":
return { type: "agent_deleted", agentId: msg.payload.agentId };
case "agent_permission_request":
return {
type: "agent_permission_request",
agentId: msg.payload.agentId,
request: msg.payload.request,
};
case "agent_permission_resolved":
return {
type: "agent_permission_resolved",
agentId: msg.payload.agentId,
requestId: msg.payload.requestId,
resolution: msg.payload.resolution,
};
case "providers_snapshot_update":
return {
type: "providers_snapshot_update",
payload: msg.payload,
};
default:
return null;
}
}
private waitForWithCancel<T>(
predicate: (msg: SessionOutboundMessage) => T | null,
timeout = 30000,
_options?: { skipQueue?: boolean },
): WaitHandle<T> {
// Capture stack trace at call site, not inside setTimeout
const timeoutError = new Error(`Timeout waiting for message (${timeout}ms)`);
let waiter: Waiter<T> | null = null;
let settled = false;
let rejectFn: ((error: Error) => void) | null = null;
const promise = new Promise<T>((resolve, reject) => {
const wrappedResolve = (value: T) => {
if (settled) return;
settled = true;
resolve(value);
};
const wrappedReject = (error: Error) => {
if (settled) return;
settled = true;
reject(error);
};
rejectFn = wrappedReject;
const timeoutHandle =
timeout > 0
? setTimeout(() => {
if (waiter) {
this.waiters.delete(waiter);
}
wrappedReject(timeoutError);
}, timeout)
: null;
waiter = {
predicate,
resolve: wrappedResolve,
reject: wrappedReject,
timeoutHandle,
};
this.waiters.add(waiter);
});
const cancel = (error: Error) => {
if (settled) {
return;
}
if (waiter) {
this.waiters.delete(waiter);
if (waiter.timeoutHandle) {
clearTimeout(waiter.timeoutHandle);
}
}
if (rejectFn) {
rejectFn(error);
return;
}
// Extremely unlikely: cancel called before the Promise executor ran.
queueMicrotask(() => {
if (!settled && rejectFn) {
rejectFn(error);
}
});
};
return { promise, cancel };
}
}
function resolveAgentConfig(options: CreateAgentRequestOptions): AgentSessionConfig {
const {
config,
provider,
cwd,
env: _env,
workspaceId: _workspaceId,
initialPrompt: _initialPrompt,
images: _images,
git: _git,
worktreeName: _worktreeName,
requestId: _requestId,
labels: _labels,
...overrides
} = options;
const baseConfig: Partial<AgentSessionConfig> = {
...(provider ? { provider } : {}),
...(cwd ? { cwd } : {}),
...overrides,
};
const merged = config ? { ...baseConfig, ...config } : baseConfig;
if (!merged.provider || !merged.cwd) {
throw new Error("createAgent requires provider and cwd");
}
return {
...merged,
provider: merged.provider,
cwd: merged.cwd,
};
}