mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 21:23:04 +00:00
Mohamed Boudra
db4376d17a
* Support file uploads to agents * Harden prompt file upload handling * Fix duplicate upload retry handling * Update relay reconnect tests for queued messages * Make upload timeout track idle progress * Clean failed upload directories
1839 lines
60 KiB
TypeScript
1839 lines
60 KiB
TypeScript
import { WebSocket, WebSocketServer } from "ws";
|
|
import type { IncomingMessage, Server as HTTPServer } from "http";
|
|
import { basename, join } from "path";
|
|
import { hostname as getHostname } from "node:os";
|
|
import type { AgentManager } from "./agent/agent-manager.js";
|
|
import type { AgentStorage } from "./agent/agent-storage.js";
|
|
import type { DownloadTokenStore } from "./file-download/token-store.js";
|
|
import type { TerminalManager } from "../terminal/terminal-manager.js";
|
|
import type pino from "pino";
|
|
import type { ProjectRegistry, WorkspaceRegistry } from "./workspace-registry.js";
|
|
import type { FileBackedChatService } from "./chat/chat-service.js";
|
|
import type { LoopService } from "./loop-service.js";
|
|
import type { ScheduleService } from "./schedule/service.js";
|
|
import type { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-manager.js";
|
|
import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js";
|
|
import {
|
|
type ServerInfoStatusPayload,
|
|
type WorkspaceSetupSnapshot,
|
|
type WSHelloMessage,
|
|
type WSInboundMessage,
|
|
WSInboundMessageSchema,
|
|
type ServerCapabilityState,
|
|
type ServerCapabilities,
|
|
type WSOutboundMessage,
|
|
wrapSessionMessage,
|
|
} from "./messages.js";
|
|
import { asUint8Array, decodeBinaryFrame } from "@getpaseo/protocol/binary-frames/index";
|
|
import type { HostnamesConfig } from "./hostnames.js";
|
|
import { isHostnameAllowed } from "./hostnames.js";
|
|
import { Session, type SessionLifecycleIntent, type SessionRuntimeMetrics } from "./session.js";
|
|
import type { AgentProvider } from "./agent/agent-sdk-types.js";
|
|
import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
|
|
import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
|
|
import { buildWorkspaceGitMetadataFromSnapshot } from "./workspace-git-metadata.js";
|
|
import { PushTokenStore } from "./push/token-store.js";
|
|
import { createPushNotificationSender, type PushNotificationSender } from "./push/notifications.js";
|
|
import type { ScriptHealthState } from "./script-health-monitor.js";
|
|
import type { ServiceProxySubsystem } from "./service-proxy.js";
|
|
import type { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js";
|
|
import type { SpeechReadinessSnapshot, SpeechService } from "./speech/speech-runtime.js";
|
|
import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js";
|
|
import { computeNotificationPlan, type ClientPresenceState } from "./agent-attention-policy.js";
|
|
import {
|
|
buildAgentAttentionNotificationPayload,
|
|
findLatestPermissionRequest,
|
|
} from "@getpaseo/protocol/agent-attention-notification";
|
|
import { createGitHubService, type GitHubService } from "../services/github-service.js";
|
|
import {
|
|
extractWsBearerProtocol,
|
|
extractWsBearerToken,
|
|
isBearerTokenValid,
|
|
type DaemonAuthConfig,
|
|
} from "./auth.js";
|
|
import {
|
|
WebSocketRuntimeMetricsWindow,
|
|
type WebSocketRuntimeCounters,
|
|
} from "./websocket/runtime-metrics.js";
|
|
|
|
const WS_CLOSE_DAEMON_AUTH_FAILED = 4401;
|
|
|
|
export interface ExternalSocketMetadata {
|
|
transport: "relay";
|
|
externalSessionKey?: string;
|
|
}
|
|
|
|
interface PendingConnection {
|
|
connectionLogger: pino.Logger;
|
|
helloTimeout: ReturnType<typeof setTimeout> | null;
|
|
}
|
|
|
|
interface WebSocketServerConfig {
|
|
allowedOrigins: Set<string>;
|
|
hostnames?: HostnamesConfig;
|
|
}
|
|
|
|
type WebSocketRuntimeMetrics = SessionRuntimeMetrics & CheckoutDiffMetrics;
|
|
|
|
function createFallbackWorkspaceGitSnapshot(cwd: string): WorkspaceGitRuntimeSnapshot {
|
|
return {
|
|
cwd,
|
|
git: {
|
|
isGit: false,
|
|
repoRoot: null,
|
|
mainRepoRoot: null,
|
|
currentBranch: null,
|
|
remoteUrl: null,
|
|
isPaseoOwnedWorktree: false,
|
|
isDirty: null,
|
|
baseRef: null,
|
|
aheadBehind: null,
|
|
aheadOfOrigin: null,
|
|
behindOfOrigin: null,
|
|
hasRemote: false,
|
|
diffStat: null,
|
|
},
|
|
github: {
|
|
featuresEnabled: false,
|
|
pullRequest: null,
|
|
error: null,
|
|
},
|
|
};
|
|
}
|
|
|
|
function createFallbackWorkspaceGitService(): WorkspaceGitService {
|
|
return {
|
|
registerWorkspace: () => ({
|
|
unsubscribe: () => {},
|
|
}),
|
|
onSnapshotUpdated: () => ({
|
|
unsubscribe: () => {},
|
|
}),
|
|
peekSnapshot: () => null,
|
|
getCheckout: async (cwd: string) => ({
|
|
cwd,
|
|
isGit: false,
|
|
currentBranch: null,
|
|
remoteUrl: null,
|
|
worktreeRoot: null,
|
|
isPaseoOwnedWorktree: false,
|
|
mainRepoRoot: null,
|
|
}),
|
|
getSnapshot: async (cwd: string) => createFallbackWorkspaceGitSnapshot(cwd),
|
|
getCheckoutDiff: async () => ({ diff: "" }),
|
|
validateBranchRef: async () => ({ kind: "not-found" }),
|
|
hasLocalBranch: async () => false,
|
|
suggestBranchesForCwd: async () => [],
|
|
listStashes: async () => [],
|
|
listWorktrees: async () => [],
|
|
getWorkspaceGitMetadata: async (cwd: string, options) => {
|
|
const snapshot = createFallbackWorkspaceGitSnapshot(cwd);
|
|
return buildWorkspaceGitMetadataFromSnapshot({
|
|
cwd,
|
|
directoryName: options?.directoryName ?? basename(cwd),
|
|
isGit: snapshot.git.isGit,
|
|
repoRoot: snapshot.git.repoRoot,
|
|
mainRepoRoot: snapshot.git.mainRepoRoot,
|
|
currentBranch: snapshot.git.currentBranch,
|
|
remoteUrl: snapshot.git.remoteUrl,
|
|
});
|
|
},
|
|
resolveRepoRoot: async (cwd: string) => cwd,
|
|
resolveDefaultBranch: async () => "main",
|
|
resolveRepoRemoteUrl: async () => null,
|
|
refresh: async () => {},
|
|
requestWorkingTreeWatch: async () => ({
|
|
repoRoot: null,
|
|
unsubscribe: () => {},
|
|
}),
|
|
scheduleRefreshForCwd: () => {},
|
|
dispose: () => {},
|
|
};
|
|
}
|
|
|
|
function createNoopProjectRegistry(): ProjectRegistry {
|
|
return {
|
|
initialize: async () => {},
|
|
existsOnDisk: async () => true,
|
|
list: async () => [],
|
|
get: async () => null,
|
|
upsert: async () => {},
|
|
archive: async () => {},
|
|
remove: async () => {},
|
|
};
|
|
}
|
|
|
|
function createNoopWorkspaceRegistry(): WorkspaceRegistry {
|
|
return {
|
|
initialize: async () => {},
|
|
existsOnDisk: async () => true,
|
|
list: async () => [],
|
|
get: async () => null,
|
|
upsert: async () => {},
|
|
archive: async () => {},
|
|
remove: async () => {},
|
|
};
|
|
}
|
|
|
|
function toServerCapabilityState(params: {
|
|
state: SpeechReadinessSnapshot["dictation"];
|
|
reason: string;
|
|
}): ServerCapabilityState {
|
|
const { state, reason } = params;
|
|
return {
|
|
enabled: state.enabled,
|
|
reason,
|
|
};
|
|
}
|
|
|
|
function resolveCapabilityReason(params: {
|
|
state: SpeechReadinessSnapshot["dictation"];
|
|
readiness: SpeechReadinessSnapshot;
|
|
}): string {
|
|
const { state, readiness } = params;
|
|
if (state.available) {
|
|
return "";
|
|
}
|
|
|
|
if (readiness.voiceFeature.reasonCode === "model_download_in_progress") {
|
|
const baseMessage = readiness.voiceFeature.message.trim();
|
|
if (baseMessage.includes("Try again in a few minutes")) {
|
|
return baseMessage;
|
|
}
|
|
return `${baseMessage} Try again in a few minutes.`;
|
|
}
|
|
|
|
return state.message;
|
|
}
|
|
|
|
function buildServerCapabilities(params: {
|
|
readiness: SpeechReadinessSnapshot | null;
|
|
}): ServerCapabilities | undefined {
|
|
const readiness = params.readiness;
|
|
if (!readiness) {
|
|
return undefined;
|
|
}
|
|
return {
|
|
voice: {
|
|
dictation: toServerCapabilityState({
|
|
state: readiness.dictation,
|
|
reason: resolveCapabilityReason({
|
|
state: readiness.dictation,
|
|
readiness,
|
|
}),
|
|
}),
|
|
voice: toServerCapabilityState({
|
|
state: readiness.realtimeVoice,
|
|
reason: resolveCapabilityReason({
|
|
state: readiness.realtimeVoice,
|
|
readiness,
|
|
}),
|
|
}),
|
|
},
|
|
};
|
|
}
|
|
|
|
function areServerCapabilitiesEqual(
|
|
current: ServerCapabilities | undefined,
|
|
next: ServerCapabilities | undefined,
|
|
): boolean {
|
|
return JSON.stringify(current ?? null) === JSON.stringify(next ?? null);
|
|
}
|
|
|
|
function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffer {
|
|
if (typeof data === "string") return Buffer.from(data, "utf8");
|
|
if (Array.isArray(data)) {
|
|
return Buffer.concat(
|
|
data.map((item) => (Buffer.isBuffer(item) ? item : Buffer.from(item as ArrayBuffer))),
|
|
);
|
|
}
|
|
if (Buffer.isBuffer(data)) return data;
|
|
return Buffer.from(data);
|
|
}
|
|
|
|
interface WebSocketLike {
|
|
readyState: number;
|
|
bufferedAmount?: number;
|
|
send: (data: string | Uint8Array | ArrayBuffer) => void;
|
|
close: (code?: number, reason?: string) => void;
|
|
on: (event: "message" | "close" | "error", listener: (...args: unknown[]) => void) => void;
|
|
once: (event: "close" | "error", listener: (...args: unknown[]) => void) => void;
|
|
}
|
|
|
|
interface SessionConnection {
|
|
session: Session;
|
|
clientId: string;
|
|
appVersion: string | null;
|
|
clientCapabilities: Record<string, unknown> | null;
|
|
connectionLogger: pino.Logger;
|
|
sockets: Set<WebSocketLike>;
|
|
externalDisconnectCleanupTimeout: ReturnType<typeof setTimeout> | null;
|
|
}
|
|
|
|
const SLOW_REQUEST_THRESHOLD_MS = 500;
|
|
const EXTERNAL_SESSION_DISCONNECT_GRACE_MS = 90_000;
|
|
const HELLO_TIMEOUT_MS = 15_000;
|
|
const WS_CLOSE_HELLO_TIMEOUT = 4001;
|
|
const WS_CLOSE_INVALID_HELLO = 4002;
|
|
const WS_CLOSE_INCOMPATIBLE_PROTOCOL = 4003;
|
|
const WS_PROTOCOL_VERSION = 1;
|
|
const WS_RUNTIME_METRICS_FLUSH_MS = 30_000;
|
|
|
|
export class MissingDaemonVersionError extends Error {
|
|
constructor() {
|
|
super("VoiceAssistantWebSocketServer requires a non-empty daemonVersion.");
|
|
this.name = "MissingDaemonVersionError";
|
|
}
|
|
}
|
|
|
|
interface RequiredWebSocketServices {
|
|
chatService: FileBackedChatService;
|
|
loopService: LoopService;
|
|
scheduleService: ScheduleService;
|
|
checkoutDiffManager: CheckoutDiffManager;
|
|
}
|
|
|
|
function requireWebSocketServices(params: {
|
|
chatService?: FileBackedChatService;
|
|
loopService?: LoopService;
|
|
scheduleService?: ScheduleService;
|
|
checkoutDiffManager?: CheckoutDiffManager;
|
|
}): RequiredWebSocketServices {
|
|
const { chatService, loopService, scheduleService, checkoutDiffManager } = params;
|
|
if (!chatService) {
|
|
throw new Error("VoiceAssistantWebSocketServer requires a chat service.");
|
|
}
|
|
if (!loopService) {
|
|
throw new Error("VoiceAssistantWebSocketServer requires a loop service.");
|
|
}
|
|
if (!scheduleService) {
|
|
throw new Error("VoiceAssistantWebSocketServer requires a schedule service.");
|
|
}
|
|
if (!checkoutDiffManager) {
|
|
throw new Error("VoiceAssistantWebSocketServer requires a checkout diff manager.");
|
|
}
|
|
return { chatService, loopService, scheduleService, checkoutDiffManager };
|
|
}
|
|
|
|
/**
|
|
* WebSocket server that only accepts sockets + parses/forwards messages to the session layer.
|
|
*/
|
|
export class VoiceAssistantWebSocketServer {
|
|
private readonly logger: pino.Logger;
|
|
private readonly wss: WebSocketServer;
|
|
private readonly pendingConnections: Map<WebSocketLike, PendingConnection> = new Map();
|
|
private readonly sessions: Map<WebSocketLike, SessionConnection> = new Map();
|
|
private readonly externalSessionsByKey: Map<string, SessionConnection> = new Map();
|
|
private readonly socketMessageQueues: Map<WebSocketLike, Promise<void>> = new Map();
|
|
private readonly serverId: string;
|
|
private readonly daemonVersion: string;
|
|
private readonly daemonRuntimeConfig:
|
|
| {
|
|
listen: string | null;
|
|
relay: {
|
|
enabled: boolean;
|
|
endpoint: string;
|
|
publicEndpoint: string;
|
|
useTls: boolean;
|
|
publicUseTls: boolean;
|
|
};
|
|
}
|
|
| undefined;
|
|
private readonly agentManager: AgentManager;
|
|
private readonly agentStorage: AgentStorage;
|
|
private readonly projectRegistry: ProjectRegistry;
|
|
private readonly workspaceRegistry: WorkspaceRegistry;
|
|
private readonly chatService: FileBackedChatService;
|
|
private readonly loopService: LoopService;
|
|
private readonly scheduleService: ScheduleService;
|
|
private readonly checkoutDiffManager: CheckoutDiffManager;
|
|
private readonly github: GitHubService;
|
|
private readonly workspaceGitService: WorkspaceGitService;
|
|
private readonly downloadTokenStore: DownloadTokenStore;
|
|
private readonly paseoHome: string;
|
|
private readonly worktreesRoot: string | undefined;
|
|
private readonly daemonConfigStore: DaemonConfigStore;
|
|
private readonly pushTokenStore: PushTokenStore;
|
|
private readonly pushNotificationSender: PushNotificationSender;
|
|
private readonly mcpBaseUrl: string | null;
|
|
private speech!: SpeechService | null;
|
|
private terminalManager!: TerminalManager | null;
|
|
private serviceProxy!: ServiceProxySubsystem | null;
|
|
private scriptRuntimeStore!: WorkspaceScriptRuntimeStore | null;
|
|
private getDaemonTcpPort!: (() => number | null) | null;
|
|
private getDaemonTcpHost!: (() => string | null) | null;
|
|
private serviceProxyPublicBaseUrl!: string | null;
|
|
private resolveScriptHealth!: ((hostname: string) => ScriptHealthState | null) | null;
|
|
private dictation!: {
|
|
finalTimeoutMs?: number;
|
|
} | null;
|
|
private readonly voiceSpeakHandlers = new Map<string, VoiceSpeakHandler>();
|
|
private readonly voiceCallerContexts = new Map<string, VoiceCallerContext>();
|
|
private readonly workspaceSetupSnapshots = new Map<string, WorkspaceSetupSnapshot>();
|
|
private readonly providerSnapshotManager: ProviderSnapshotManager;
|
|
private onLifecycleIntent!: ((intent: SessionLifecycleIntent) => void) | null;
|
|
private onBranchChanged!:
|
|
| ((workspaceId: string, oldBranch: string | null, newBranch: string | null) => void)
|
|
| null;
|
|
private serverCapabilities: ServerCapabilities | undefined;
|
|
private readonly runtimeMetrics = new WebSocketRuntimeMetricsWindow();
|
|
private runtimeMetricsInterval: ReturnType<typeof setInterval> | null = null;
|
|
private unsubscribeSpeechReadiness: (() => void) | null = null;
|
|
private unsubscribeDaemonConfigChange: (() => void) | null = null;
|
|
|
|
constructor(
|
|
server: HTTPServer,
|
|
logger: pino.Logger,
|
|
serverId: string,
|
|
agentManager: AgentManager,
|
|
agentStorage: AgentStorage,
|
|
downloadTokenStore: DownloadTokenStore,
|
|
paseoHome: string,
|
|
daemonConfigStore: DaemonConfigStore,
|
|
mcpBaseUrl: string | null,
|
|
wsConfig: WebSocketServerConfig,
|
|
auth?: DaemonAuthConfig,
|
|
speech?: SpeechService | null,
|
|
terminalManager?: TerminalManager | null,
|
|
dictation?: {
|
|
finalTimeoutMs?: number;
|
|
},
|
|
daemonVersion?: string,
|
|
onLifecycleIntent?: (intent: SessionLifecycleIntent) => void,
|
|
projectRegistry?: ProjectRegistry,
|
|
workspaceRegistry?: WorkspaceRegistry,
|
|
chatService?: FileBackedChatService,
|
|
loopService?: LoopService,
|
|
scheduleService?: ScheduleService,
|
|
checkoutDiffManager?: CheckoutDiffManager,
|
|
serviceProxy?: ServiceProxySubsystem | null,
|
|
scriptRuntimeStore?: WorkspaceScriptRuntimeStore | null,
|
|
onBranchChanged?: (
|
|
workspaceId: string,
|
|
oldBranch: string | null,
|
|
newBranch: string | null,
|
|
) => void,
|
|
getDaemonTcpPort?: () => number | null,
|
|
getDaemonTcpHost?: () => string | null,
|
|
resolveScriptHealth?: (hostname: string) => ScriptHealthState | null,
|
|
workspaceGitService?: WorkspaceGitService,
|
|
github?: GitHubService,
|
|
pushNotificationSender?: PushNotificationSender,
|
|
providerSnapshotManager?: ProviderSnapshotManager,
|
|
daemonRuntimeConfig?: {
|
|
listen: string | null;
|
|
worktreesRoot?: string;
|
|
relay: {
|
|
enabled: boolean;
|
|
endpoint: string;
|
|
publicEndpoint: string;
|
|
useTls: boolean;
|
|
publicUseTls: boolean;
|
|
};
|
|
},
|
|
serviceProxyPublicBaseUrl?: string | null,
|
|
) {
|
|
this.logger = logger.child({ module: "websocket-server" });
|
|
this.serverId = serverId;
|
|
if (typeof daemonVersion !== "string" || daemonVersion.trim().length === 0) {
|
|
throw new MissingDaemonVersionError();
|
|
}
|
|
this.daemonVersion = daemonVersion.trim();
|
|
this.daemonRuntimeConfig = daemonRuntimeConfig;
|
|
this.agentManager = agentManager;
|
|
this.agentStorage = agentStorage;
|
|
this.projectRegistry = projectRegistry ?? createNoopProjectRegistry();
|
|
this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry();
|
|
const requiredServices = requireWebSocketServices({
|
|
chatService,
|
|
loopService,
|
|
scheduleService,
|
|
checkoutDiffManager,
|
|
});
|
|
this.chatService = requiredServices.chatService;
|
|
this.loopService = requiredServices.loopService;
|
|
this.scheduleService = requiredServices.scheduleService;
|
|
this.checkoutDiffManager = requiredServices.checkoutDiffManager;
|
|
this.github = github ?? createGitHubService();
|
|
this.workspaceGitService = workspaceGitService ?? createFallbackWorkspaceGitService();
|
|
this.downloadTokenStore = downloadTokenStore;
|
|
this.paseoHome = paseoHome;
|
|
this.worktreesRoot = daemonRuntimeConfig?.worktreesRoot;
|
|
this.daemonConfigStore = daemonConfigStore;
|
|
this.mcpBaseUrl = mcpBaseUrl;
|
|
this.assignOptionalServices({
|
|
speech,
|
|
terminalManager,
|
|
dictation,
|
|
onLifecycleIntent,
|
|
serviceProxy,
|
|
scriptRuntimeStore,
|
|
onBranchChanged,
|
|
getDaemonTcpPort,
|
|
getDaemonTcpHost,
|
|
serviceProxyPublicBaseUrl,
|
|
resolveScriptHealth,
|
|
});
|
|
if (!providerSnapshotManager) {
|
|
throw new Error("providerSnapshotManager is required");
|
|
}
|
|
this.providerSnapshotManager = providerSnapshotManager;
|
|
this.serverCapabilities = buildServerCapabilities({
|
|
readiness: this.speech?.getReadiness() ?? null,
|
|
});
|
|
this.unsubscribeSpeechReadiness =
|
|
this.speech?.onReadinessChange((snapshot) => {
|
|
this.publishSpeechReadiness(snapshot);
|
|
}) ?? null;
|
|
this.unsubscribeDaemonConfigChange = this.daemonConfigStore.onChange((config) => {
|
|
const nextAgentManagerState = this.providerSnapshotManager.applyMutableProviderConfig(
|
|
config.providers,
|
|
);
|
|
this.agentManager.updateProviderRegistry(nextAgentManagerState);
|
|
this.broadcastDaemonConfigChanged(config);
|
|
});
|
|
|
|
const pushLogger = this.logger.child({ module: "push" });
|
|
this.pushTokenStore = new PushTokenStore(pushLogger, join(paseoHome, "push-tokens.json"));
|
|
this.pushNotificationSender =
|
|
pushNotificationSender ?? createPushNotificationSender(pushLogger, this.pushTokenStore);
|
|
|
|
this.agentManager.setAgentAttentionCallback((params) => {
|
|
void this.broadcastAgentAttention(params).catch((err) => {
|
|
this.logger.warn({ err, agentId: params.agentId }, "Failed to broadcast agent attention");
|
|
});
|
|
});
|
|
|
|
this.wss = this.createWebSocketServer(server, wsConfig, auth);
|
|
this.startRuntimeMetricsInterval();
|
|
|
|
this.logger.info("WebSocket server initialized on /ws");
|
|
}
|
|
|
|
private assignOptionalServices(params: {
|
|
speech: SpeechService | null | undefined;
|
|
terminalManager: TerminalManager | null | undefined;
|
|
dictation: { finalTimeoutMs?: number } | undefined;
|
|
onLifecycleIntent: ((intent: SessionLifecycleIntent) => void) | undefined;
|
|
serviceProxy: ServiceProxySubsystem | null | undefined;
|
|
scriptRuntimeStore: WorkspaceScriptRuntimeStore | null | undefined;
|
|
onBranchChanged:
|
|
| ((workspaceId: string, oldBranch: string | null, newBranch: string | null) => void)
|
|
| undefined;
|
|
getDaemonTcpPort: (() => number | null) | undefined;
|
|
getDaemonTcpHost: (() => string | null) | undefined;
|
|
serviceProxyPublicBaseUrl: string | null | undefined;
|
|
resolveScriptHealth: ((hostname: string) => ScriptHealthState | null) | undefined;
|
|
}): void {
|
|
this.speech = params.speech ?? null;
|
|
this.terminalManager = params.terminalManager ?? null;
|
|
this.dictation = params.dictation ?? null;
|
|
this.onLifecycleIntent = params.onLifecycleIntent ?? null;
|
|
this.serviceProxy = params.serviceProxy ?? null;
|
|
this.scriptRuntimeStore = params.scriptRuntimeStore ?? null;
|
|
this.onBranchChanged = params.onBranchChanged ?? null;
|
|
this.getDaemonTcpPort = params.getDaemonTcpPort ?? null;
|
|
this.getDaemonTcpHost = params.getDaemonTcpHost ?? null;
|
|
this.serviceProxyPublicBaseUrl = params.serviceProxyPublicBaseUrl ?? null;
|
|
this.resolveScriptHealth = params.resolveScriptHealth ?? null;
|
|
}
|
|
|
|
private createWebSocketServer(
|
|
server: HTTPServer,
|
|
wsConfig: WebSocketServerConfig,
|
|
auth: DaemonAuthConfig | undefined,
|
|
): WebSocketServer {
|
|
const { allowedOrigins, hostnames } = wsConfig;
|
|
const password = auth?.password;
|
|
const wss = new WebSocketServer({
|
|
server,
|
|
path: "/ws",
|
|
handleProtocols: (protocols) => selectWebSocketProtocol(protocols, password),
|
|
verifyClient: ({ req }, callback) => {
|
|
this.verifyWsUpgrade(req, allowedOrigins, hostnames, callback);
|
|
},
|
|
});
|
|
wss.on("connection", (ws, request) => {
|
|
void this.attachAuthenticatedSocket(ws, request, password);
|
|
});
|
|
return wss;
|
|
}
|
|
|
|
private startRuntimeMetricsInterval(): void {
|
|
const runtimeMetricsInterval = setInterval(() => {
|
|
this.flushRuntimeMetrics();
|
|
}, WS_RUNTIME_METRICS_FLUSH_MS);
|
|
this.runtimeMetricsInterval = runtimeMetricsInterval;
|
|
(runtimeMetricsInterval as unknown as { unref?: () => void }).unref?.();
|
|
}
|
|
|
|
private verifyWsUpgrade(
|
|
req: IncomingMessage,
|
|
allowedOrigins: Set<string>,
|
|
hostnames: HostnamesConfig | undefined,
|
|
callback: (res: boolean, code?: number, message?: string) => void,
|
|
): void {
|
|
const requestMetadata = extractSocketRequestMetadata(req);
|
|
const origin = requestMetadata.origin;
|
|
const requestHost = requestMetadata.host ?? null;
|
|
if (requestHost && !isHostnameAllowed(requestHost, hostnames)) {
|
|
this.incrementRuntimeCounter("hostRejected");
|
|
this.logger.warn(
|
|
{ ...requestMetadata, host: requestHost },
|
|
"Rejected connection from disallowed host",
|
|
);
|
|
callback(false, 403, "Host not allowed");
|
|
return;
|
|
}
|
|
const sameOrigin =
|
|
!!origin &&
|
|
!!requestHost &&
|
|
(origin === `http://${requestHost}` || origin === `https://${requestHost}`);
|
|
|
|
if (!origin || allowedOrigins.has("*") || allowedOrigins.has(origin) || sameOrigin) {
|
|
callback(true);
|
|
} else {
|
|
this.incrementRuntimeCounter("originRejected");
|
|
this.logger.warn({ ...requestMetadata, origin }, "Rejected connection from origin");
|
|
callback(false, 403, "Origin not allowed");
|
|
}
|
|
}
|
|
|
|
private async attachAuthenticatedSocket(
|
|
ws: WebSocket,
|
|
request: IncomingMessage,
|
|
password: string | undefined,
|
|
): Promise<void> {
|
|
if (password) {
|
|
const requestMetadata = extractSocketRequestMetadata(request);
|
|
const protocol = extractWsBearerProtocol(request.headers["sec-websocket-protocol"]);
|
|
const token = extractWsBearerToken(protocol);
|
|
const isAuthorized = isBearerTokenValid({ password, token });
|
|
if (!isAuthorized) {
|
|
const reason = token === null ? "Password required" : "Incorrect password";
|
|
this.logger.warn(
|
|
{ ...requestMetadata, hasToken: token !== null },
|
|
"Rejected WebSocket connection with invalid daemon password",
|
|
);
|
|
ws.close(WS_CLOSE_DAEMON_AUTH_FAILED, reason);
|
|
return;
|
|
}
|
|
}
|
|
|
|
await this.attachSocket(ws, request);
|
|
}
|
|
|
|
public broadcast(message: WSOutboundMessage): void {
|
|
const payload = JSON.stringify(message);
|
|
for (const ws of this.sessions.keys()) {
|
|
// WebSocket.OPEN = 1
|
|
if (ws.readyState === 1) {
|
|
ws.send(payload);
|
|
this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
|
|
}
|
|
}
|
|
}
|
|
|
|
public listActiveSessions(): Session[] {
|
|
return Array.from(
|
|
new Set(
|
|
[...this.sessions.values(), ...this.externalSessionsByKey.values()].map(
|
|
(connection) => connection.session,
|
|
),
|
|
),
|
|
);
|
|
}
|
|
|
|
public publishSpeechReadiness(readiness: SpeechReadinessSnapshot | null): void {
|
|
this.updateServerCapabilities(buildServerCapabilities({ readiness }));
|
|
}
|
|
|
|
public updateServerCapabilities(capabilities: ServerCapabilities | null | undefined): void {
|
|
const next = capabilities ?? undefined;
|
|
if (areServerCapabilitiesEqual(this.serverCapabilities, next)) {
|
|
return;
|
|
}
|
|
this.serverCapabilities = next;
|
|
this.broadcastCapabilitiesUpdate();
|
|
}
|
|
|
|
public async attachExternalSocket(
|
|
ws: WebSocketLike,
|
|
metadata?: ExternalSocketMetadata,
|
|
): Promise<void> {
|
|
if (metadata?.transport === "relay") {
|
|
this.incrementRuntimeCounter("relayExternalSocketAttached");
|
|
}
|
|
await this.attachSocket(ws, undefined, metadata);
|
|
}
|
|
|
|
public async close(): Promise<void> {
|
|
this.unsubscribeSpeechReadiness?.();
|
|
this.unsubscribeSpeechReadiness = null;
|
|
this.unsubscribeDaemonConfigChange?.();
|
|
this.unsubscribeDaemonConfigChange = null;
|
|
if (this.runtimeMetricsInterval) {
|
|
clearInterval(this.runtimeMetricsInterval);
|
|
this.runtimeMetricsInterval = null;
|
|
}
|
|
this.flushRuntimeMetrics({ final: true });
|
|
|
|
const uniqueConnections = new Set<SessionConnection>([
|
|
...this.sessions.values(),
|
|
...this.externalSessionsByKey.values(),
|
|
]);
|
|
|
|
const pendingSockets = new Set<WebSocketLike>(this.pendingConnections.keys());
|
|
for (const pending of this.pendingConnections.values()) {
|
|
if (pending.helloTimeout) {
|
|
clearTimeout(pending.helloTimeout);
|
|
pending.helloTimeout = null;
|
|
}
|
|
}
|
|
|
|
const cleanupPromises: Promise<void>[] = [];
|
|
for (const connection of uniqueConnections) {
|
|
if (connection.externalDisconnectCleanupTimeout) {
|
|
clearTimeout(connection.externalDisconnectCleanupTimeout);
|
|
connection.externalDisconnectCleanupTimeout = null;
|
|
}
|
|
|
|
cleanupPromises.push(connection.session.cleanup());
|
|
for (const ws of connection.sockets) {
|
|
cleanupPromises.push(
|
|
new Promise<void>((resolve) => {
|
|
// WebSocket.CLOSED = 3
|
|
if (ws.readyState === 3) {
|
|
resolve();
|
|
return;
|
|
}
|
|
ws.once("close", () => resolve());
|
|
ws.close();
|
|
}),
|
|
);
|
|
}
|
|
}
|
|
|
|
for (const ws of pendingSockets) {
|
|
cleanupPromises.push(
|
|
new Promise<void>((resolve) => {
|
|
if (ws.readyState === 3) {
|
|
resolve();
|
|
return;
|
|
}
|
|
ws.once("close", () => resolve());
|
|
ws.close();
|
|
}),
|
|
);
|
|
}
|
|
|
|
await Promise.all(cleanupPromises);
|
|
this.providerSnapshotManager.destroy();
|
|
this.checkoutDiffManager.dispose();
|
|
this.workspaceGitService.dispose();
|
|
this.pendingConnections.clear();
|
|
this.sessions.clear();
|
|
this.externalSessionsByKey.clear();
|
|
this.wss.close();
|
|
}
|
|
|
|
private sendToClient(ws: WebSocketLike, message: WSOutboundMessage): void {
|
|
// WebSocket.OPEN = 1. The check is a fast path; the socket can still
|
|
// transition to closed between here and ws.send(), so guard the send too —
|
|
// a synchronous throw here would propagate as an uncaughtException.
|
|
if (ws.readyState !== 1) {
|
|
return;
|
|
}
|
|
try {
|
|
ws.send(JSON.stringify(message));
|
|
this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
|
|
} catch (err) {
|
|
this.logger.warn({ err }, "ws_send_failed");
|
|
}
|
|
}
|
|
|
|
private sendBinaryToClient(ws: WebSocketLike, frame: Uint8Array): void {
|
|
if (ws.readyState !== 1) {
|
|
return;
|
|
}
|
|
try {
|
|
ws.send(frame);
|
|
this.runtimeMetrics.recordOutboundBinaryFrame(ws.bufferedAmount);
|
|
} catch (err) {
|
|
this.logger.warn({ err }, "ws_send_binary_failed");
|
|
}
|
|
}
|
|
|
|
private sendToConnection(connection: SessionConnection, message: WSOutboundMessage): void {
|
|
for (const ws of connection.sockets) {
|
|
this.sendToClient(ws, message);
|
|
}
|
|
}
|
|
|
|
private sendBinaryToConnection(connection: SessionConnection, frame: Uint8Array): void {
|
|
for (const ws of connection.sockets) {
|
|
this.sendBinaryToClient(ws, frame);
|
|
}
|
|
}
|
|
|
|
private async attachSocket(
|
|
ws: WebSocketLike,
|
|
request?: unknown,
|
|
metadata?: ExternalSocketMetadata,
|
|
): Promise<void> {
|
|
const requestMetadata = extractSocketRequestMetadata(request);
|
|
const connectionLoggerFields: Record<string, string> = {
|
|
transport: metadata?.transport === "relay" ? "relay" : "direct",
|
|
};
|
|
if (requestMetadata.host) {
|
|
connectionLoggerFields.host = requestMetadata.host;
|
|
}
|
|
if (requestMetadata.origin) {
|
|
connectionLoggerFields.origin = requestMetadata.origin;
|
|
}
|
|
if (requestMetadata.userAgent) {
|
|
connectionLoggerFields.userAgent = requestMetadata.userAgent;
|
|
}
|
|
if (requestMetadata.remoteAddress) {
|
|
connectionLoggerFields.remoteAddress = requestMetadata.remoteAddress;
|
|
}
|
|
const connectionLogger = this.logger.child(connectionLoggerFields);
|
|
|
|
const pending: PendingConnection = {
|
|
connectionLogger,
|
|
helloTimeout: null,
|
|
};
|
|
const timeout = setTimeout(() => {
|
|
if (this.pendingConnections.get(ws) !== pending) {
|
|
return;
|
|
}
|
|
pending.helloTimeout = null;
|
|
this.pendingConnections.delete(ws);
|
|
pending.connectionLogger.warn(
|
|
{ timeoutMs: HELLO_TIMEOUT_MS },
|
|
"Closing connection due to missing hello",
|
|
);
|
|
try {
|
|
ws.close(WS_CLOSE_HELLO_TIMEOUT, "Hello timeout");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
}, HELLO_TIMEOUT_MS);
|
|
pending.helloTimeout = timeout;
|
|
(timeout as unknown as { unref?: () => void }).unref?.();
|
|
|
|
this.pendingConnections.set(ws, pending);
|
|
this.incrementRuntimeCounter("connectedAwaitingHello");
|
|
this.bindSocketHandlers(ws);
|
|
|
|
pending.connectionLogger.trace(
|
|
{
|
|
totalPendingConnections: this.pendingConnections.size,
|
|
},
|
|
"Client connected; awaiting hello",
|
|
);
|
|
}
|
|
|
|
private createSessionConnection(params: {
|
|
ws: WebSocketLike;
|
|
clientId: string;
|
|
appVersion: string | null;
|
|
clientCapabilities: Record<string, unknown> | null;
|
|
connectionLogger: pino.Logger;
|
|
}): SessionConnection {
|
|
const { ws, clientId, appVersion, clientCapabilities, connectionLogger } = params;
|
|
let connection: SessionConnection | null = null;
|
|
|
|
const session = new Session({
|
|
clientId,
|
|
appVersion,
|
|
clientCapabilities,
|
|
onMessage: (msg) => {
|
|
if (!connection) {
|
|
return;
|
|
}
|
|
this.sendToConnection(connection, wrapSessionMessage(msg));
|
|
},
|
|
onBinaryMessage: (frame) => {
|
|
if (!connection) {
|
|
return;
|
|
}
|
|
this.sendBinaryToConnection(connection, frame);
|
|
},
|
|
onLifecycleIntent: (intent) => {
|
|
this.onLifecycleIntent?.(intent);
|
|
},
|
|
logger: connectionLogger.child({ module: "session" }),
|
|
downloadTokenStore: this.downloadTokenStore,
|
|
pushTokenStore: this.pushTokenStore,
|
|
paseoHome: this.paseoHome,
|
|
worktreesRoot: this.worktreesRoot,
|
|
agentManager: this.agentManager,
|
|
agentStorage: this.agentStorage,
|
|
projectRegistry: this.projectRegistry,
|
|
workspaceRegistry: this.workspaceRegistry,
|
|
chatService: this.chatService,
|
|
loopService: this.loopService,
|
|
scheduleService: this.scheduleService,
|
|
checkoutDiffManager: this.checkoutDiffManager,
|
|
github: this.github,
|
|
workspaceGitService: this.workspaceGitService,
|
|
daemonConfigStore: this.daemonConfigStore,
|
|
mcpBaseUrl: this.mcpBaseUrl,
|
|
stt: () => this.speech?.resolveStt() ?? null,
|
|
sttLanguage: this.speech?.resolveSttLanguage() ?? "en",
|
|
tts: () => this.speech?.resolveTts() ?? null,
|
|
terminalManager: this.terminalManager,
|
|
providerSnapshotManager: this.providerSnapshotManager,
|
|
serviceProxy: this.serviceProxy ?? undefined,
|
|
scriptRuntimeStore: this.scriptRuntimeStore ?? undefined,
|
|
workspaceSetupSnapshots: this.workspaceSetupSnapshots,
|
|
onBranchChanged: this.onBranchChanged ?? undefined,
|
|
getDaemonTcpPort: this.getDaemonTcpPort ?? undefined,
|
|
getDaemonTcpHost: this.getDaemonTcpHost ?? undefined,
|
|
serviceProxyPublicBaseUrl: this.serviceProxyPublicBaseUrl,
|
|
resolveScriptHealth: this.resolveScriptHealth ?? undefined,
|
|
voice: {
|
|
turnDetection: () => this.speech?.resolveTurnDetection() ?? null,
|
|
},
|
|
voiceBridge: {
|
|
registerVoiceSpeakHandler: (agentId, handler) => {
|
|
this.voiceSpeakHandlers.set(agentId, handler);
|
|
},
|
|
unregisterVoiceSpeakHandler: (agentId) => {
|
|
this.voiceSpeakHandlers.delete(agentId);
|
|
},
|
|
registerVoiceCallerContext: (agentId, context) => {
|
|
this.voiceCallerContexts.set(agentId, context);
|
|
},
|
|
unregisterVoiceCallerContext: (agentId) => {
|
|
this.voiceCallerContexts.delete(agentId);
|
|
},
|
|
},
|
|
dictation:
|
|
this.dictation || this.speech
|
|
? {
|
|
finalTimeoutMs: this.dictation?.finalTimeoutMs,
|
|
stt: () => this.speech?.resolveDictationStt() ?? null,
|
|
sttLanguage: this.speech?.resolveDictationSttLanguage() ?? "en",
|
|
getSpeechReadiness: () => this.speech!.getReadiness(),
|
|
}
|
|
: undefined,
|
|
serverId: this.serverId,
|
|
daemonVersion: this.daemonVersion,
|
|
daemonRuntimeConfig: this.daemonRuntimeConfig,
|
|
});
|
|
|
|
connection = {
|
|
session,
|
|
clientId,
|
|
appVersion,
|
|
clientCapabilities,
|
|
connectionLogger,
|
|
sockets: new Set([ws]),
|
|
externalDisconnectCleanupTimeout: null,
|
|
};
|
|
return connection;
|
|
}
|
|
|
|
private clearPendingConnection(ws: WebSocketLike): PendingConnection | null {
|
|
const pending = this.pendingConnections.get(ws);
|
|
if (!pending) {
|
|
return null;
|
|
}
|
|
if (pending.helloTimeout) {
|
|
clearTimeout(pending.helloTimeout);
|
|
pending.helloTimeout = null;
|
|
}
|
|
this.pendingConnections.delete(ws);
|
|
return pending;
|
|
}
|
|
|
|
private handleHello(params: {
|
|
ws: WebSocketLike;
|
|
message: WSHelloMessage;
|
|
pending: PendingConnection;
|
|
}): void {
|
|
const { ws, message, pending } = params;
|
|
|
|
if (message.protocolVersion !== WS_PROTOCOL_VERSION) {
|
|
this.clearPendingConnection(ws);
|
|
pending.connectionLogger.warn(
|
|
{
|
|
receivedProtocolVersion: message.protocolVersion,
|
|
expectedProtocolVersion: WS_PROTOCOL_VERSION,
|
|
},
|
|
"Rejected hello due to protocol version mismatch",
|
|
);
|
|
try {
|
|
ws.close(WS_CLOSE_INCOMPATIBLE_PROTOCOL, "Incompatible protocol version");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
return;
|
|
}
|
|
|
|
const clientId = message.clientId.trim();
|
|
if (clientId.length === 0) {
|
|
this.clearPendingConnection(ws);
|
|
pending.connectionLogger.warn("Rejected hello with empty clientId");
|
|
try {
|
|
ws.close(WS_CLOSE_INVALID_HELLO, "Invalid hello");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
return;
|
|
}
|
|
|
|
this.clearPendingConnection(ws);
|
|
const existing = this.externalSessionsByKey.get(clientId);
|
|
if (existing) {
|
|
this.incrementRuntimeCounter("helloResumed");
|
|
if (existing.externalDisconnectCleanupTimeout) {
|
|
clearTimeout(existing.externalDisconnectCleanupTimeout);
|
|
existing.externalDisconnectCleanupTimeout = null;
|
|
}
|
|
const newAppVersion = message.appVersion ?? null;
|
|
if (newAppVersion && newAppVersion !== existing.appVersion) {
|
|
existing.appVersion = newAppVersion;
|
|
existing.session.updateAppVersion(newAppVersion);
|
|
}
|
|
const newClientCapabilities = message.capabilities ?? null;
|
|
if (
|
|
JSON.stringify(existing.clientCapabilities ?? null) !==
|
|
JSON.stringify(newClientCapabilities ?? null)
|
|
) {
|
|
existing.clientCapabilities = newClientCapabilities;
|
|
existing.session.updateClientCapabilities(newClientCapabilities);
|
|
}
|
|
existing.sockets.add(ws);
|
|
this.sessions.set(ws, existing);
|
|
this.sendToClient(ws, this.createServerInfoMessage());
|
|
existing.connectionLogger.trace(
|
|
{
|
|
clientId,
|
|
resumed: true,
|
|
totalSessions: this.sessions.size,
|
|
},
|
|
"Client connected via hello",
|
|
);
|
|
return;
|
|
}
|
|
|
|
const connectionLogger = pending.connectionLogger.child({ clientId });
|
|
this.incrementRuntimeCounter("helloNew");
|
|
const connection = this.createSessionConnection({
|
|
ws,
|
|
clientId,
|
|
appVersion: message.appVersion ?? null,
|
|
clientCapabilities: message.capabilities ?? null,
|
|
connectionLogger,
|
|
});
|
|
this.sessions.set(ws, connection);
|
|
this.externalSessionsByKey.set(clientId, connection);
|
|
this.sendToClient(ws, this.createServerInfoMessage());
|
|
connection.connectionLogger.trace(
|
|
{
|
|
clientId,
|
|
resumed: false,
|
|
totalSessions: this.sessions.size,
|
|
},
|
|
"Client connected via hello",
|
|
);
|
|
}
|
|
|
|
private buildServerInfoStatusPayload(): ServerInfoStatusPayload {
|
|
return {
|
|
status: "server_info",
|
|
serverId: this.serverId,
|
|
hostname: getHostname(),
|
|
version: this.daemonVersion,
|
|
...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
|
|
features: {
|
|
// COMPAT(providersSnapshot): keep optional until all clients rely on snapshot flow.
|
|
providersSnapshot: true,
|
|
// COMPAT(checkoutGithubSetAutoMerge): added in v0.1.75, remove gate after 2026-11-13.
|
|
checkoutGithubSetAutoMerge: true,
|
|
// COMPAT(daemonStatusRpc): added in v0.1.76, remove gate after 2026-11-18.
|
|
daemonStatusRpc: true,
|
|
// COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.
|
|
"terminal-restore-modes": true,
|
|
// COMPAT(rewind): added in v0.1.X, drop the gate when floor >= v0.1.X.
|
|
rewind: true,
|
|
// COMPAT(checkoutRefresh): added in v0.1.86, remove gate after 2026-11-29.
|
|
checkoutRefresh: true,
|
|
},
|
|
};
|
|
}
|
|
|
|
private createServerInfoMessage(): WSOutboundMessage {
|
|
return {
|
|
type: "session",
|
|
message: {
|
|
type: "status",
|
|
payload: this.buildServerInfoStatusPayload(),
|
|
},
|
|
};
|
|
}
|
|
|
|
private createDaemonConfigChangedMessage(config: MutableDaemonConfig): WSOutboundMessage {
|
|
return wrapSessionMessage({
|
|
type: "status",
|
|
payload: {
|
|
status: "daemon_config_changed",
|
|
config,
|
|
},
|
|
});
|
|
}
|
|
|
|
private broadcastCapabilitiesUpdate(): void {
|
|
this.broadcast(this.createServerInfoMessage());
|
|
}
|
|
|
|
private broadcastDaemonConfigChanged(config: MutableDaemonConfig): void {
|
|
this.broadcast(this.createDaemonConfigChangedMessage(config));
|
|
}
|
|
|
|
private bindSocketHandlers(ws: WebSocketLike): void {
|
|
ws.on("message", (...args: unknown[]) => {
|
|
const data = args[0] as Buffer | ArrayBuffer | Buffer[] | string;
|
|
this.enqueueRawMessage(ws, data);
|
|
});
|
|
|
|
ws.on("close", async (...args: unknown[]) => {
|
|
const code = args[0];
|
|
const reason = args[1];
|
|
await this.detachSocket(ws, {
|
|
code: typeof code === "number" ? code : undefined,
|
|
reason,
|
|
});
|
|
});
|
|
|
|
ws.on("error", async (...args: unknown[]) => {
|
|
const error = args[0];
|
|
const err = error instanceof Error ? error : new Error(String(error));
|
|
const active = this.sessions.get(ws);
|
|
const pending = this.pendingConnections.get(ws);
|
|
const log = active?.connectionLogger ?? pending?.connectionLogger ?? this.logger;
|
|
log.error({ err }, "Client error");
|
|
await this.detachSocket(ws, { error: err });
|
|
});
|
|
}
|
|
|
|
private enqueueRawMessage(
|
|
ws: WebSocketLike,
|
|
data: Buffer | ArrayBuffer | Buffer[] | string,
|
|
): void {
|
|
const previous = this.socketMessageQueues.get(ws) ?? Promise.resolve();
|
|
const next = previous.then(
|
|
() => this.handleRawMessage(ws, data),
|
|
() => this.handleRawMessage(ws, data),
|
|
);
|
|
this.socketMessageQueues.set(ws, next);
|
|
void next
|
|
.catch(() => undefined)
|
|
.finally(() => {
|
|
if (this.socketMessageQueues.get(ws) === next) {
|
|
this.socketMessageQueues.delete(ws);
|
|
}
|
|
});
|
|
}
|
|
|
|
public resolveVoiceSpeakHandler(callerAgentId: string): VoiceSpeakHandler | null {
|
|
return this.voiceSpeakHandlers.get(callerAgentId) ?? null;
|
|
}
|
|
|
|
public resolveVoiceCallerContext(callerAgentId: string): VoiceCallerContext | null {
|
|
return this.voiceCallerContexts.get(callerAgentId) ?? null;
|
|
}
|
|
|
|
private async detachSocket(
|
|
ws: WebSocketLike,
|
|
details: {
|
|
code?: number;
|
|
reason?: unknown;
|
|
error?: Error;
|
|
},
|
|
): Promise<void> {
|
|
const pending = this.clearPendingConnection(ws);
|
|
if (pending) {
|
|
this.incrementRuntimeCounter("pendingDisconnected");
|
|
pending.connectionLogger.trace(
|
|
{
|
|
code: details.code,
|
|
reason: stringifyCloseReason(details.reason),
|
|
},
|
|
"Pending client disconnected",
|
|
);
|
|
return;
|
|
}
|
|
|
|
const connection = this.sessions.get(ws);
|
|
if (!connection) {
|
|
return;
|
|
}
|
|
|
|
this.sessions.delete(ws);
|
|
connection.sockets.delete(ws);
|
|
|
|
if (connection.sockets.size === 0) {
|
|
this.incrementRuntimeCounter("sessionDisconnectedWaitingReconnect");
|
|
if (connection.externalDisconnectCleanupTimeout) {
|
|
clearTimeout(connection.externalDisconnectCleanupTimeout);
|
|
}
|
|
const timeout = setTimeout(() => {
|
|
if (connection.externalDisconnectCleanupTimeout !== timeout) {
|
|
return;
|
|
}
|
|
connection.externalDisconnectCleanupTimeout = null;
|
|
void this.cleanupConnection(connection, "Client disconnected (grace timeout)");
|
|
}, EXTERNAL_SESSION_DISCONNECT_GRACE_MS);
|
|
connection.externalDisconnectCleanupTimeout = timeout;
|
|
|
|
connection.connectionLogger.trace(
|
|
{
|
|
clientId: connection.clientId,
|
|
code: details.code,
|
|
reason: stringifyCloseReason(details.reason),
|
|
reconnectGraceMs: EXTERNAL_SESSION_DISCONNECT_GRACE_MS,
|
|
},
|
|
"Client disconnected; waiting for reconnect",
|
|
);
|
|
return;
|
|
}
|
|
|
|
if (connection.sockets.size > 0) {
|
|
this.incrementRuntimeCounter("sessionSocketDisconnectedAttached");
|
|
connection.connectionLogger.trace(
|
|
{
|
|
clientId: connection.clientId,
|
|
remainingSockets: connection.sockets.size,
|
|
code: details.code,
|
|
reason: stringifyCloseReason(details.reason),
|
|
},
|
|
"Client socket disconnected; session remains attached",
|
|
);
|
|
return;
|
|
}
|
|
|
|
await this.cleanupConnection(connection, "Client disconnected");
|
|
}
|
|
|
|
private async cleanupConnection(
|
|
connection: SessionConnection,
|
|
logMessage: string,
|
|
): Promise<void> {
|
|
this.incrementRuntimeCounter("sessionCleanup");
|
|
if (connection.externalDisconnectCleanupTimeout) {
|
|
clearTimeout(connection.externalDisconnectCleanupTimeout);
|
|
connection.externalDisconnectCleanupTimeout = null;
|
|
}
|
|
|
|
for (const socket of connection.sockets) {
|
|
this.sessions.delete(socket);
|
|
}
|
|
connection.sockets.clear();
|
|
const existing = this.externalSessionsByKey.get(connection.clientId);
|
|
if (existing === connection) {
|
|
this.externalSessionsByKey.delete(connection.clientId);
|
|
}
|
|
|
|
connection.connectionLogger.trace(
|
|
{ clientId: connection.clientId, totalSessions: this.sessions.size },
|
|
logMessage,
|
|
);
|
|
await connection.session.cleanup();
|
|
}
|
|
|
|
private handleInvalidInboundMessage(args: {
|
|
ws: WebSocketLike;
|
|
parsed: unknown;
|
|
parsedMessage: { success: false; error: { message: string } } & Record<string, unknown>;
|
|
pendingConnection: PendingConnection | undefined;
|
|
activeConnection: SessionConnection | undefined;
|
|
log: pino.Logger;
|
|
}): void {
|
|
const { ws, parsed, parsedMessage, pendingConnection, activeConnection, log } = args;
|
|
this.incrementRuntimeCounter("validationFailed");
|
|
if (pendingConnection) {
|
|
pendingConnection.connectionLogger.warn(
|
|
{ error: parsedMessage.error.message },
|
|
"Rejected pending message before hello",
|
|
);
|
|
this.clearPendingConnection(ws);
|
|
try {
|
|
ws.close(WS_CLOSE_INVALID_HELLO, "Invalid hello");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
return;
|
|
}
|
|
|
|
const requestInfo = extractRequestInfoFromUnknownWsInbound(parsed);
|
|
const isUnknownSchema =
|
|
requestInfo?.requestId != null &&
|
|
typeof parsed === "object" &&
|
|
parsed != null &&
|
|
"type" in parsed &&
|
|
(parsed as { type?: unknown }).type === "session";
|
|
|
|
log.warn(
|
|
{
|
|
clientId: activeConnection?.clientId,
|
|
requestId: requestInfo?.requestId,
|
|
requestType: requestInfo?.requestType,
|
|
error: parsedMessage.error.message,
|
|
},
|
|
"WS inbound message validation failed",
|
|
);
|
|
|
|
if (requestInfo) {
|
|
this.sendToClient(
|
|
ws,
|
|
wrapSessionMessage({
|
|
type: "rpc_error",
|
|
payload: {
|
|
requestId: requestInfo.requestId,
|
|
requestType: requestInfo.requestType,
|
|
error: isUnknownSchema
|
|
? `Unknown request, try upgrading the daemon (currently v${this.daemonVersion})`
|
|
: "Invalid message",
|
|
code: isUnknownSchema ? "unknown_schema" : "invalid_message",
|
|
},
|
|
}),
|
|
);
|
|
return;
|
|
}
|
|
|
|
const errorMessage = `Invalid message: ${parsedMessage.error.message}`;
|
|
this.sendToClient(
|
|
ws,
|
|
wrapSessionMessage({
|
|
type: "status",
|
|
payload: {
|
|
status: "error",
|
|
message: errorMessage,
|
|
},
|
|
}),
|
|
);
|
|
}
|
|
|
|
private async maybeHandleBinaryFrame(params: {
|
|
ws: WebSocketLike;
|
|
buffer: Buffer;
|
|
activeConnection: SessionConnection | undefined;
|
|
log: pino.Logger;
|
|
}): Promise<boolean> {
|
|
const { ws, buffer, activeConnection, log } = params;
|
|
const asBytes = asUint8Array(buffer);
|
|
if (!asBytes) {
|
|
return false;
|
|
}
|
|
const decodedFrame = decodeBinaryFrame(asBytes);
|
|
if (!decodedFrame) {
|
|
return false;
|
|
}
|
|
if (!activeConnection) {
|
|
this.incrementRuntimeCounter("binaryBeforeHelloRejected");
|
|
log.warn("Rejected binary frame before hello");
|
|
this.clearPendingConnection(ws);
|
|
try {
|
|
ws.close(WS_CLOSE_INVALID_HELLO, "Session message before hello");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
return true;
|
|
}
|
|
await activeConnection.session.handleBinaryFrame(decodedFrame);
|
|
return true;
|
|
}
|
|
|
|
private handlePendingConnectionMessage(params: {
|
|
ws: WebSocketLike;
|
|
message: WSInboundMessage;
|
|
pendingConnection: PendingConnection;
|
|
}): void {
|
|
const { ws, message, pendingConnection } = params;
|
|
if (message.type === "hello") {
|
|
this.handleHello({
|
|
ws,
|
|
message,
|
|
pending: pendingConnection,
|
|
});
|
|
return;
|
|
}
|
|
|
|
pendingConnection.connectionLogger.warn(
|
|
{
|
|
messageType: message.type,
|
|
},
|
|
"Rejected pending message before hello",
|
|
);
|
|
this.incrementRuntimeCounter("pendingMessageRejectedBeforeHello");
|
|
this.clearPendingConnection(ws);
|
|
try {
|
|
ws.close(WS_CLOSE_INVALID_HELLO, "Session message before hello");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
}
|
|
|
|
private async handleRawMessage(
|
|
ws: WebSocketLike,
|
|
data: Buffer | ArrayBuffer | Buffer[] | string,
|
|
): Promise<void> {
|
|
const activeConnection = this.sessions.get(ws);
|
|
const pendingConnection = this.pendingConnections.get(ws);
|
|
const log =
|
|
activeConnection?.connectionLogger ?? pendingConnection?.connectionLogger ?? this.logger;
|
|
|
|
try {
|
|
const buffer = bufferFromWsData(data);
|
|
const binaryHandled = await this.maybeHandleBinaryFrame({
|
|
ws,
|
|
buffer,
|
|
activeConnection,
|
|
log,
|
|
});
|
|
if (binaryHandled) {
|
|
return;
|
|
}
|
|
|
|
const parsed = JSON.parse(buffer.toString());
|
|
const parsedMessage = WSInboundMessageSchema.safeParse(parsed);
|
|
if (!parsedMessage.success) {
|
|
this.handleInvalidInboundMessage({
|
|
ws,
|
|
parsed,
|
|
parsedMessage,
|
|
pendingConnection,
|
|
activeConnection,
|
|
log,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const message = parsedMessage.data;
|
|
this.recordInboundMessageType(message.type);
|
|
|
|
if (message.type === "ping") {
|
|
this.sendToClient(ws, { type: "pong" });
|
|
return;
|
|
}
|
|
|
|
if (message.type === "recording_state") {
|
|
return;
|
|
}
|
|
|
|
if (pendingConnection) {
|
|
this.handlePendingConnectionMessage({
|
|
ws,
|
|
message,
|
|
pendingConnection,
|
|
});
|
|
return;
|
|
}
|
|
|
|
if (!activeConnection) {
|
|
this.incrementRuntimeCounter("missingConnectionForMessage");
|
|
this.logger.error("No connection found for websocket");
|
|
return;
|
|
}
|
|
|
|
if (message.type === "hello") {
|
|
this.incrementRuntimeCounter("unexpectedHelloOnActiveConnection");
|
|
activeConnection.connectionLogger.warn("Received hello on active connection");
|
|
try {
|
|
ws.close(WS_CLOSE_INVALID_HELLO, "Unexpected hello");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (message.type === "session") {
|
|
await this.dispatchSessionMessage(activeConnection, message);
|
|
}
|
|
} catch (error) {
|
|
this.handleRawMessageError({ ws, data, error, log });
|
|
}
|
|
}
|
|
|
|
private async dispatchSessionMessage(
|
|
activeConnection: SessionConnection,
|
|
message: Extract<WSInboundMessage, { type: "session" }>,
|
|
): Promise<void> {
|
|
this.recordInboundSessionRequestType(message.message.type);
|
|
const startMs = performance.now();
|
|
await activeConnection.session.handleMessage(message.message);
|
|
const durationMs = performance.now() - startMs;
|
|
this.recordRequestLatency(message.message.type, durationMs);
|
|
|
|
if (durationMs >= SLOW_REQUEST_THRESHOLD_MS) {
|
|
activeConnection.connectionLogger.warn(
|
|
{
|
|
requestType: message.message.type,
|
|
durationMs: Math.round(durationMs),
|
|
inflightRequests: activeConnection.session.getRuntimeMetrics().inflightRequests,
|
|
},
|
|
"ws_slow_request",
|
|
);
|
|
}
|
|
}
|
|
|
|
private handleRawMessageError(params: {
|
|
ws: WebSocketLike;
|
|
data: Buffer | ArrayBuffer | Buffer[] | string;
|
|
error: unknown;
|
|
log: pino.Logger;
|
|
}): void {
|
|
const { ws, data, error, log } = params;
|
|
const err = error instanceof Error ? error : new Error(String(error));
|
|
const { rawPayload, parsedPayload } = this.decodeRawMessagePayloadForError(data);
|
|
|
|
const trimmedRawPayload =
|
|
typeof rawPayload === "string" && rawPayload.length > 2000
|
|
? `${rawPayload.slice(0, 2000)}... (truncated)`
|
|
: rawPayload;
|
|
|
|
log.error(
|
|
{
|
|
err,
|
|
rawPayload: trimmedRawPayload,
|
|
parsedPayload,
|
|
},
|
|
"Failed to parse/handle message",
|
|
);
|
|
|
|
if (this.pendingConnections.has(ws)) {
|
|
this.clearPendingConnection(ws);
|
|
try {
|
|
ws.close(WS_CLOSE_INVALID_HELLO, "Invalid hello");
|
|
} catch {
|
|
// ignore close errors
|
|
}
|
|
return;
|
|
}
|
|
|
|
const requestInfo = extractRequestInfoFromUnknownWsInbound(parsedPayload);
|
|
if (requestInfo) {
|
|
this.sendToClient(
|
|
ws,
|
|
wrapSessionMessage({
|
|
type: "rpc_error",
|
|
payload: {
|
|
requestId: requestInfo.requestId,
|
|
requestType: requestInfo.requestType,
|
|
error: "Invalid message",
|
|
code: "invalid_message",
|
|
},
|
|
}),
|
|
);
|
|
return;
|
|
}
|
|
|
|
this.sendToClient(
|
|
ws,
|
|
wrapSessionMessage({
|
|
type: "status",
|
|
payload: {
|
|
status: "error",
|
|
message: `Invalid message: ${err.message}`,
|
|
},
|
|
}),
|
|
);
|
|
}
|
|
|
|
private decodeRawMessagePayloadForError(data: Buffer | ArrayBuffer | Buffer[] | string): {
|
|
rawPayload: string | null;
|
|
parsedPayload: unknown;
|
|
} {
|
|
let rawPayload: string | null = null;
|
|
let parsedPayload: unknown = null;
|
|
try {
|
|
const buffer = bufferFromWsData(data);
|
|
rawPayload = buffer.toString();
|
|
parsedPayload = JSON.parse(rawPayload);
|
|
} catch (payloadError) {
|
|
rawPayload = rawPayload ?? "<unreadable>";
|
|
parsedPayload = parsedPayload ?? rawPayload;
|
|
const payloadErr =
|
|
payloadError instanceof Error ? payloadError : new Error(String(payloadError));
|
|
this.logger.error({ err: payloadErr }, "Failed to decode raw payload");
|
|
}
|
|
return { rawPayload, parsedPayload };
|
|
}
|
|
|
|
private incrementRuntimeCounter(counter: keyof WebSocketRuntimeCounters): void {
|
|
this.runtimeMetrics.incrementCounter(counter);
|
|
}
|
|
|
|
private recordInboundMessageType(type: string): void {
|
|
this.runtimeMetrics.recordInboundMessage(type);
|
|
}
|
|
|
|
private recordInboundSessionRequestType(type: string): void {
|
|
this.runtimeMetrics.recordInboundSessionRequest(type);
|
|
}
|
|
|
|
private recordRequestLatency(type: string, durationMs: number): void {
|
|
this.runtimeMetrics.recordRequestLatency(type, durationMs);
|
|
}
|
|
|
|
private collectSessionRuntimeMetrics(): WebSocketRuntimeMetrics {
|
|
const uniqueConnections = new Set<SessionConnection>(this.externalSessionsByKey.values());
|
|
let terminalDirectorySubscriptionCount = 0;
|
|
let terminalSubscriptionCount = 0;
|
|
let inflightRequests = 0;
|
|
let peakInflightRequests = 0;
|
|
|
|
for (const connection of uniqueConnections) {
|
|
const sessionMetrics = connection.session.getRuntimeMetrics();
|
|
terminalDirectorySubscriptionCount += sessionMetrics.terminalDirectorySubscriptionCount;
|
|
terminalSubscriptionCount += sessionMetrics.terminalSubscriptionCount;
|
|
inflightRequests += sessionMetrics.inflightRequests;
|
|
peakInflightRequests = Math.max(peakInflightRequests, sessionMetrics.peakInflightRequests);
|
|
connection.session.resetPeakInflight();
|
|
}
|
|
|
|
return {
|
|
...this.checkoutDiffManager.getMetrics(),
|
|
terminalDirectorySubscriptionCount,
|
|
terminalSubscriptionCount,
|
|
inflightRequests,
|
|
peakInflightRequests,
|
|
};
|
|
}
|
|
|
|
private flushRuntimeMetrics(options?: { final?: boolean }): void {
|
|
const runtimeMetrics = this.runtimeMetrics.snapshotAndReset();
|
|
const activeConnections = new Set<SessionConnection>(this.sessions.values()).size;
|
|
const activeSockets = this.sessions.size;
|
|
const pendingConnections = this.pendingConnections.size;
|
|
const reconnectGraceSessions = [...this.externalSessionsByKey.values()].filter(
|
|
(connection) =>
|
|
connection.sockets.size === 0 && connection.externalDisconnectCleanupTimeout !== null,
|
|
).length;
|
|
const sessionMetrics = this.collectSessionRuntimeMetrics();
|
|
const agentSnapshot = this.agentManager.getMetricsSnapshot();
|
|
|
|
this.logger.info(
|
|
{
|
|
windowMs: runtimeMetrics.windowMs,
|
|
final: Boolean(options?.final),
|
|
sessions: {
|
|
activeConnections,
|
|
externalSessionKeys: this.externalSessionsByKey.size,
|
|
reconnectGraceSessions,
|
|
},
|
|
sockets: {
|
|
activeSockets,
|
|
pendingConnections,
|
|
},
|
|
counters: runtimeMetrics.counters,
|
|
inboundMessageTypesTop: runtimeMetrics.inboundMessageTypesTop,
|
|
inboundSessionRequestTypesTop: runtimeMetrics.inboundSessionRequestTypesTop,
|
|
outboundMessageTypesTop: runtimeMetrics.outboundMessageTypesTop,
|
|
outboundSessionMessageTypesTop: runtimeMetrics.outboundSessionMessageTypesTop,
|
|
outboundAgentStreamTypesTop: runtimeMetrics.outboundAgentStreamTypesTop,
|
|
outboundAgentStreamAgentsTop: runtimeMetrics.outboundAgentStreamAgentsTop,
|
|
outboundBinaryFrameTypesTop: runtimeMetrics.outboundBinaryFrameTypesTop,
|
|
bufferedAmount: runtimeMetrics.bufferedAmount,
|
|
runtime: sessionMetrics,
|
|
latency: runtimeMetrics.latency,
|
|
agents: agentSnapshot,
|
|
},
|
|
"ws_runtime_metrics",
|
|
);
|
|
}
|
|
|
|
private getClientActivityState(session: Session): ClientPresenceState {
|
|
const activity = session.getClientActivity();
|
|
if (!activity) {
|
|
return {
|
|
appVisible: false,
|
|
focusedAgentId: null,
|
|
lastActivityAtMs: null,
|
|
};
|
|
}
|
|
|
|
return {
|
|
appVisible: activity.appVisible,
|
|
focusedAgentId: activity.focusedAgentId,
|
|
lastActivityAtMs: activity.lastActivityAt.getTime(),
|
|
};
|
|
}
|
|
|
|
private async broadcastAgentAttention(params: {
|
|
agentId: string;
|
|
provider: AgentProvider;
|
|
reason: "finished" | "error" | "permission";
|
|
}): Promise<void> {
|
|
const clientEntries: Array<{
|
|
ws: WebSocketLike;
|
|
state: ClientPresenceState;
|
|
}> = [];
|
|
|
|
for (const [ws, connection] of this.sessions) {
|
|
clientEntries.push({
|
|
ws,
|
|
state: this.getClientActivityState(connection.session),
|
|
});
|
|
}
|
|
|
|
const allStates = clientEntries.map((e) => e.state);
|
|
const nowMs = Date.now();
|
|
const agent = this.agentManager.getAgent(params.agentId);
|
|
const assistantMessage = await this.agentManager.getLastAssistantMessage(params.agentId);
|
|
const notification = buildAgentAttentionNotificationPayload({
|
|
reason: params.reason,
|
|
serverId: this.serverId,
|
|
agentId: params.agentId,
|
|
assistantMessage,
|
|
permissionRequest: agent ? findLatestPermissionRequest(agent.pendingPermissions) : null,
|
|
});
|
|
|
|
const plan = computeNotificationPlan({
|
|
allStates,
|
|
agentId: params.agentId,
|
|
reason: params.reason,
|
|
nowMs,
|
|
});
|
|
|
|
if (plan.shouldPush) {
|
|
void this.pushNotificationSender.send(notification).catch((err) => {
|
|
this.logger.warn({ err, agentId: params.agentId }, "Failed to send push notification");
|
|
});
|
|
}
|
|
|
|
for (const [clientIndex, { ws }] of clientEntries.entries()) {
|
|
const shouldNotify = clientIndex === plan.inAppRecipientIndex;
|
|
const timestamp = new Date().toISOString();
|
|
const message = wrapSessionMessage({
|
|
type: "agent_stream",
|
|
payload: {
|
|
agentId: params.agentId,
|
|
event: {
|
|
type: "attention_required",
|
|
provider: params.provider,
|
|
reason: params.reason,
|
|
timestamp,
|
|
shouldNotify,
|
|
notification,
|
|
},
|
|
timestamp,
|
|
},
|
|
});
|
|
|
|
this.sendToClient(ws, message);
|
|
}
|
|
}
|
|
}
|
|
|
|
interface SocketRequestMetadata {
|
|
host?: string;
|
|
origin?: string;
|
|
userAgent?: string;
|
|
remoteAddress?: string;
|
|
}
|
|
|
|
function extractSocketRequestMetadata(request: unknown): SocketRequestMetadata {
|
|
if (!request || typeof request !== "object") {
|
|
return {};
|
|
}
|
|
|
|
const record = request as {
|
|
headers?: {
|
|
host?: unknown;
|
|
origin?: unknown;
|
|
"user-agent"?: unknown;
|
|
};
|
|
url?: unknown;
|
|
socket?: {
|
|
remoteAddress?: unknown;
|
|
};
|
|
};
|
|
|
|
const host = typeof record.headers?.host === "string" ? record.headers.host : undefined;
|
|
const origin = typeof record.headers?.origin === "string" ? record.headers.origin : undefined;
|
|
const userAgent =
|
|
typeof record.headers?.["user-agent"] === "string" ? record.headers["user-agent"] : undefined;
|
|
const remoteAddress =
|
|
typeof record.socket?.remoteAddress === "string" ? record.socket.remoteAddress : undefined;
|
|
|
|
return {
|
|
...(host ? { host } : {}),
|
|
...(origin ? { origin } : {}),
|
|
...(userAgent ? { userAgent } : {}),
|
|
...(remoteAddress ? { remoteAddress } : {}),
|
|
};
|
|
}
|
|
|
|
function selectWebSocketProtocol(
|
|
protocols: Set<string>,
|
|
password: string | undefined,
|
|
): string | false {
|
|
if (!password) {
|
|
return protocols.values().next().value ?? false;
|
|
}
|
|
|
|
for (const protocol of protocols) {
|
|
const token = extractWsBearerToken(protocol);
|
|
if (token !== null) {
|
|
return protocol;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
function stringifyCloseReason(reason: unknown): string | null {
|
|
if (typeof reason === "string") {
|
|
return reason.length > 0 ? reason : null;
|
|
}
|
|
if (Buffer.isBuffer(reason)) {
|
|
const text = reason.toString();
|
|
return text.length > 0 ? text : null;
|
|
}
|
|
if (reason == null) {
|
|
return null;
|
|
}
|
|
const text = String(reason);
|
|
return text.length > 0 ? text : null;
|
|
}
|
|
|
|
function extractRequestInfoFromUnknownWsInbound(
|
|
payload: unknown,
|
|
): { requestId: string; requestType?: string } | null {
|
|
if (!payload || typeof payload !== "object") {
|
|
return null;
|
|
}
|
|
|
|
const record = payload as {
|
|
type?: unknown;
|
|
requestId?: unknown;
|
|
message?: unknown;
|
|
};
|
|
|
|
// Session-wrapped messages
|
|
if (record.type === "session" && record.message && typeof record.message === "object") {
|
|
const msg = record.message as { requestId?: unknown; type?: unknown };
|
|
if (typeof msg.requestId === "string") {
|
|
return {
|
|
requestId: msg.requestId,
|
|
...(typeof msg.type === "string" ? { requestType: msg.type } : {}),
|
|
};
|
|
}
|
|
}
|
|
|
|
// Non-session messages (future-proof)
|
|
if (typeof record.requestId === "string") {
|
|
return {
|
|
requestId: record.requestId,
|
|
...(typeof record.type === "string" ? { requestType: record.type } : {}),
|
|
};
|
|
}
|
|
|
|
return null;
|
|
}
|