Carlos Florêncio
1691 lines
58 KiB
TypeScript
1691 lines
58 KiB
TypeScript
/* eslint-disable max-lines -- WebSocketClient intentionally owns one connection's complete request, subscription, and reconnect lifecycle. */
|
|
|
|
import type { BackendMessageMap, BackendMessageType } from "@/lib/types/backend";
|
|
import type { ConversationChangedPayload } from "@/lib/types/session-events";
|
|
import type { ConnectionStatus } from "@/lib/types/connection";
|
|
import { generateUUID } from "@/lib/utils";
|
|
import { createDebugLogger, isDebug } from "@/lib/debug/log";
|
|
import { dispatchToPluginWsHandlers } from "@/lib/ws/plugin-bridge";
|
|
import {
|
|
isWebSocketRequestTimeoutError,
|
|
toWebSocketRequestError,
|
|
WebSocketRequestTimeoutError,
|
|
} from "./request-error";
|
|
import {
|
|
isRawSessionEvent,
|
|
orderedCoreDisposition,
|
|
orderedCoreAction,
|
|
projectCoreSessionPayload,
|
|
type CoreSessionStream,
|
|
type RawSessionEvent,
|
|
} from "./ordered-session-events";
|
|
export type { RawSessionEvent } from "./ordered-session-events";
|
|
export {
|
|
isWebSocketRequestTimeoutError,
|
|
WebSocketRequestError,
|
|
WebSocketRequestTimeoutError,
|
|
type WebSocketRequestErrorDetails,
|
|
} from "./request-error";
|
|
|
|
const debugDispatch = createDebugLogger("ws:dispatch");
|
|
|
|
// High-frequency notification types we skip in the dispatch log to avoid
|
|
// drowning the console during agent streams. Filter [ws:dispatch] to see
|
|
// everything else; if you need the streaming traffic, comment this out.
|
|
const DISPATCH_LOG_DENYLIST = new Set<string>([
|
|
"session.message.added",
|
|
"session.message.updated",
|
|
"session.message.deleted",
|
|
"session.shell.output",
|
|
"session.process.output",
|
|
]);
|
|
|
|
type MessageHandler<T extends BackendMessageType> = (message: BackendMessageMap[T]) => void;
|
|
|
|
// Internal alias for the status vocabulary shared with the UI.
|
|
type WebSocketStatus = ConnectionStatus;
|
|
|
|
export interface ReconnectOptions {
|
|
enabled?: boolean;
|
|
maxAttempts?: number;
|
|
initialDelay?: number;
|
|
maxDelay?: number;
|
|
backoffMultiplier?: number;
|
|
conversationProtocol?: "v1" | "v2";
|
|
}
|
|
|
|
export interface SessionSubscriptionHandle {
|
|
ready: Promise<void>;
|
|
unsubscribe: () => void;
|
|
}
|
|
|
|
export type CoreSessionRecoveryHandler = () => Promise<boolean>;
|
|
|
|
const CORE_CONVERSATION_ACTIONS = new Set<BackendMessageType>([
|
|
"session.message.added",
|
|
"session.message.updated",
|
|
"session.message.deleted",
|
|
"session.turn.started",
|
|
"session.turn.completed",
|
|
"session.turn.removed",
|
|
]);
|
|
const SESSION_SUBSCRIBE_ACTION = "session.subscribe";
|
|
const SESSION_UNSUBSCRIBE_ACTION = "session.unsubscribe";
|
|
const SESSION_CONVERSATION_CHANGED_ACTION = "session.conversation.changed" as const;
|
|
|
|
function isConversationNotificationAction(action: BackendMessageType): boolean {
|
|
return CORE_CONVERSATION_ACTIONS.has(action);
|
|
}
|
|
|
|
function isCoreConversationOperation(value: unknown): boolean {
|
|
if (!value || typeof value !== "object" || Array.isArray(value)) return false;
|
|
const operation = value as {
|
|
kind?: unknown;
|
|
entity?: unknown;
|
|
id?: unknown;
|
|
message?: unknown;
|
|
turn?: unknown;
|
|
};
|
|
if (
|
|
(operation.kind !== "upsert" && operation.kind !== "remove") ||
|
|
(operation.entity !== "message" && operation.entity !== "turn") ||
|
|
typeof operation.id !== "string" ||
|
|
operation.id === ""
|
|
) {
|
|
return false;
|
|
}
|
|
if (operation.kind === "remove") return true;
|
|
const payload = operation.entity === "message" ? operation.message : operation.turn;
|
|
return Boolean(payload && typeof payload === "object" && !Array.isArray(payload));
|
|
}
|
|
|
|
type ParsedCoreConversationChange = {
|
|
sessionId: string;
|
|
payload: Partial<ConversationChangedPayload> & { operations?: unknown[] };
|
|
valid: boolean;
|
|
};
|
|
|
|
function isDecimalRevision(value: unknown): value is string {
|
|
return typeof value === "string" && /^(0|[1-9][0-9]*)$/.test(value);
|
|
}
|
|
|
|
function parseCoreConversationChange(value: unknown): ParsedCoreConversationChange | null {
|
|
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
|
|
const payload = value as Partial<ConversationChangedPayload> & { operations?: unknown[] };
|
|
if (typeof payload.session_id !== "string") return null;
|
|
const valid =
|
|
payload.protocol_version === 2 &&
|
|
typeof payload.scope_id === "string" &&
|
|
typeof payload.epoch === "string" &&
|
|
isDecimalRevision(payload.base_revision) &&
|
|
isDecimalRevision(payload.revision) &&
|
|
Array.isArray(payload.operations) &&
|
|
payload.operations.length <= 256 &&
|
|
payload.operations.every(isCoreConversationOperation) &&
|
|
BigInt(payload.revision) > BigInt(payload.base_revision);
|
|
return { sessionId: payload.session_id, payload, valid };
|
|
}
|
|
|
|
function needsCoreConversationRecovery(
|
|
parsed: ParsedCoreConversationChange,
|
|
stream: CoreSessionStream,
|
|
): boolean {
|
|
return !parsed.valid || parsed.payload.scope_id !== stream.conversationScopeId;
|
|
}
|
|
|
|
function hasCoreConversationRevisionGap(
|
|
payload: ConversationChangedPayload,
|
|
stream: CoreSessionStream,
|
|
appliedRevision: string,
|
|
): boolean {
|
|
return (
|
|
payload.reset === true ||
|
|
payload.epoch !== stream.conversationEpoch ||
|
|
payload.base_revision !== appliedRevision
|
|
);
|
|
}
|
|
|
|
type CoreConversationOperation = {
|
|
kind: "upsert" | "remove";
|
|
entity: "message" | "turn";
|
|
id: string;
|
|
message?: Record<string, unknown>;
|
|
turn?: Record<string, unknown>;
|
|
};
|
|
|
|
function coreTurnAction(operation: CoreConversationOperation): BackendMessageType {
|
|
if (operation.kind === "remove") return "session.turn.removed";
|
|
return operation.turn?.completed_at ? "session.turn.completed" : "session.turn.started";
|
|
}
|
|
|
|
function projectCoreMessageOperation(
|
|
sessionId: string,
|
|
operation: CoreConversationOperation & { entity: "message" },
|
|
): { action: BackendMessageType; payload: Record<string, unknown> } {
|
|
const message = operation.message;
|
|
const payload: Record<string, unknown> = {
|
|
type: operation.kind === "remove" ? "message.deleted" : (message?.type ?? "message"),
|
|
session_id: sessionId,
|
|
task_id: message?.task_id ?? null,
|
|
message_id: operation.id,
|
|
};
|
|
if (operation.kind === "upsert" && message) {
|
|
Object.assign(payload, {
|
|
author_type: message.author_type,
|
|
author_id: message.author_id,
|
|
content: message.content,
|
|
raw_content: message.raw_content,
|
|
message_type: message.type,
|
|
turn_id: message.turn_id,
|
|
metadata: message.metadata,
|
|
requests_input: message.requests_input,
|
|
created_at: message.created_at,
|
|
updated_at: message.updated_at ?? message.created_at,
|
|
prompt_index: message.prompt_index,
|
|
});
|
|
}
|
|
return {
|
|
action: operation.kind === "remove" ? "session.message.deleted" : "session.message.added",
|
|
payload,
|
|
};
|
|
}
|
|
|
|
function projectCoreTurnOperation(
|
|
sessionId: string,
|
|
operation: CoreConversationOperation & { entity: "turn" },
|
|
): { action: BackendMessageType; payload: Record<string, unknown> } {
|
|
const turn = operation.turn;
|
|
const payload: Record<string, unknown> = {
|
|
type: operation.kind === "remove" ? "session.turn.removed" : "session.turn.started",
|
|
session_id: sessionId,
|
|
task_id: turn?.task_id ?? null,
|
|
id: operation.id,
|
|
};
|
|
if (operation.kind === "upsert" && turn) {
|
|
Object.assign(payload, {
|
|
started_at: turn.started_at,
|
|
completed_at: turn.completed_at,
|
|
updated_at: turn.updated_at ?? turn.started_at,
|
|
created_at: turn.created_at ?? turn.started_at,
|
|
execution_profile_id: turn.execution_profile_id,
|
|
route_generation: turn.route_generation,
|
|
metadata: turn.metadata,
|
|
had_output: turn.had_output,
|
|
});
|
|
}
|
|
return { action: coreTurnAction(operation), payload };
|
|
}
|
|
|
|
function projectCoreConversationOperation(
|
|
sessionId: string,
|
|
operation: CoreConversationOperation,
|
|
): { action: BackendMessageType; payload: Record<string, unknown> } {
|
|
if (operation.entity === "message") {
|
|
return projectCoreMessageOperation(
|
|
sessionId,
|
|
operation as CoreConversationOperation & { entity: "message" },
|
|
);
|
|
}
|
|
return projectCoreTurnOperation(
|
|
sessionId,
|
|
operation as CoreConversationOperation & { entity: "turn" },
|
|
);
|
|
}
|
|
|
|
type SessionSubscriptionReadiness = {
|
|
promise: Promise<void>;
|
|
resolve: () => void;
|
|
reject: (reason: unknown) => void;
|
|
requestStarted: boolean;
|
|
settled: boolean;
|
|
attempt: number;
|
|
retryTimer: ReturnType<typeof setTimeout> | null;
|
|
};
|
|
export const SESSION_ENTRY_REQUEST_TIMEOUT_MS = 10000;
|
|
export const SESSION_ENTRY_RETRY_DELAY_MS = 1000;
|
|
const MAX_SESSION_SUBSCRIPTION_ATTEMPTS = 2;
|
|
// i18n-exempt: transport/API diagnostic, never rendered as user-facing copy.
|
|
const SESSION_SUBSCRIPTION_RELEASED_ERROR = "Session subscription released";
|
|
|
|
type OrderedSessionResponse = Record<string, unknown>;
|
|
|
|
// i18n-exempt: transport protocol validation diagnostics are never rendered as user-facing copy.
|
|
function invalidOrderedSessionResponse(): Error {
|
|
return new Error("Invalid ordered session response");
|
|
}
|
|
|
|
function isNonNegativeSafeInteger(value: unknown): value is number {
|
|
return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
|
|
}
|
|
|
|
function requiredOrderedString(response: OrderedSessionResponse, key: string): string {
|
|
const value = response[key];
|
|
if (typeof value !== "string" || value === "") throw invalidOrderedSessionResponse();
|
|
return value;
|
|
}
|
|
|
|
// eslint-disable-next-line complexity -- The wire contract is validated field-by-field before state mutation.
|
|
function validateOrderedSubscribeResponse(
|
|
response: unknown,
|
|
sessionId: string,
|
|
wireId: string,
|
|
): { eventWatermark: number; resumeToken: string; result: "fresh" | "replay" | "invalid_resume" } {
|
|
if (!response || typeof response !== "object" || Array.isArray(response)) {
|
|
throw invalidOrderedSessionResponse();
|
|
}
|
|
const payload = response as OrderedSessionResponse;
|
|
if (payload.success !== true || payload.session_id !== sessionId || payload.wire_id !== wireId) {
|
|
throw invalidOrderedSessionResponse();
|
|
}
|
|
const eventWatermark = payload.event_watermark;
|
|
if (
|
|
!isNonNegativeSafeInteger(eventWatermark) ||
|
|
payload.snapshot_cutoff !== eventWatermark ||
|
|
!["fresh", "replay", "invalid_resume"].includes(payload.result as string)
|
|
) {
|
|
throw invalidOrderedSessionResponse();
|
|
}
|
|
requiredOrderedString(payload, "snapshot_token");
|
|
const resumeToken = requiredOrderedString(payload, "resume_token");
|
|
requiredOrderedString(payload, "expires_at");
|
|
if (payload.result === "replay") {
|
|
const replayFrom = payload.replay_from;
|
|
const replayTo = payload.replay_to;
|
|
if (
|
|
!isNonNegativeSafeInteger(replayFrom) ||
|
|
replayFrom === 0 ||
|
|
!isNonNegativeSafeInteger(replayTo) ||
|
|
replayTo < replayFrom ||
|
|
replayTo > eventWatermark
|
|
) {
|
|
throw invalidOrderedSessionResponse();
|
|
}
|
|
}
|
|
return {
|
|
eventWatermark,
|
|
resumeToken,
|
|
result: payload.result as "fresh" | "replay" | "invalid_resume",
|
|
};
|
|
}
|
|
|
|
function validateOrderedAckResponse(
|
|
response: unknown,
|
|
sessionId: string,
|
|
wireId: string,
|
|
sequence: number,
|
|
): string {
|
|
if (!response || typeof response !== "object" || Array.isArray(response)) {
|
|
throw invalidOrderedSessionResponse();
|
|
}
|
|
const payload = response as OrderedSessionResponse;
|
|
if (
|
|
payload.success !== true ||
|
|
payload.session_id !== sessionId ||
|
|
payload.wire_id !== wireId ||
|
|
payload.acknowledged_sequence !== sequence
|
|
) {
|
|
throw invalidOrderedSessionResponse();
|
|
}
|
|
return requiredOrderedString(payload, "resume_token");
|
|
}
|
|
type RawWebSocketMessage = {
|
|
type?: unknown;
|
|
id?: unknown;
|
|
payload?: unknown;
|
|
};
|
|
|
|
const DEFAULT_RECONNECT_OPTIONS: Required<ReconnectOptions> = {
|
|
enabled: true,
|
|
maxAttempts: 10,
|
|
initialDelay: 1000,
|
|
maxDelay: 30000,
|
|
backoffMultiplier: 1.5,
|
|
conversationProtocol: "v2",
|
|
};
|
|
// i18n-exempt: transport/API diagnostic. Callers branch on the error code and
|
|
// render translated copy; this text only ever appears in a console or as an
|
|
// interpolated English diagnostic (see docs/i18n.md on interpolated values).
|
|
const WEBSOCKET_CONNECTION_CLOSED_ERROR = "WebSocket connection closed";
|
|
|
|
export class WebSocketClient {
|
|
private socket: WebSocket | null = null;
|
|
private status: WebSocketStatus = "disconnected";
|
|
private handlers = new Map<BackendMessageType, Set<MessageHandler<BackendMessageType>>>();
|
|
private pendingRequests = new Map<
|
|
string,
|
|
{
|
|
resolve: (payload: unknown) => void;
|
|
reject: (error: Error) => void;
|
|
timeout: ReturnType<typeof setTimeout>;
|
|
}
|
|
>();
|
|
private rawSessionEventHandlers = new Set<(event: RawSessionEvent) => void>();
|
|
private statusHandlers = new Set<(status: WebSocketStatus) => void>();
|
|
private pendingQueue: string[] = [];
|
|
private reconnectOptions: Required<ReconnectOptions>;
|
|
private reconnectAttempts = 0;
|
|
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
|
private intentionalClose = false;
|
|
private subscriptions = new Map<string, number>();
|
|
private sessionSubscriptions = new Map<string, number>();
|
|
private sessionSubscriptionReadiness = new Map<string, SessionSubscriptionReadiness>();
|
|
private coreSessionStreams = new Map<string, CoreSessionStream>();
|
|
private coreSessionRecoveryHandlers = new Map<string, Set<CoreSessionRecoveryHandler>>();
|
|
private coreConversationSessions = new Set<string>();
|
|
// Ref-counted focus signals: a session can be focused by both the task panel
|
|
// and the task details page if both are mounted. Backend wakes its workspace
|
|
// tracker into fast-poll mode while any client has focus, falling back to
|
|
// slow when the count reaches 0 (debounced server-side).
|
|
private sessionFocusCounts = new Map<string, number>();
|
|
private userSubscriptionCount = 0;
|
|
private runSubscriptions = new Map<string, number>();
|
|
private systemMetricsSubscriptionCount = 0;
|
|
|
|
constructor(
|
|
private url: string,
|
|
private onStatusChange?: (status: WebSocketStatus) => void,
|
|
reconnectOptions?: ReconnectOptions,
|
|
) {
|
|
this.reconnectOptions = { ...DEFAULT_RECONNECT_OPTIONS, ...reconnectOptions };
|
|
}
|
|
|
|
getStatus() {
|
|
return this.status;
|
|
}
|
|
|
|
connect() {
|
|
if (this.socket) return;
|
|
this.intentionalClose = false;
|
|
this.clearReconnectTimer();
|
|
this.setStatus("connecting");
|
|
const socket = new WebSocket(this.url);
|
|
this.socket = socket;
|
|
|
|
socket.onopen = () => {
|
|
if (this.socket !== socket) return;
|
|
this.reconnectAttempts = 0;
|
|
this.setStatus("connected");
|
|
this.resubscribe();
|
|
this.flushQueue();
|
|
};
|
|
|
|
socket.onmessage = (event) => {
|
|
if (this.socket !== socket) return;
|
|
const parts = (event.data as string).split("\n");
|
|
for (const part of parts) {
|
|
const trimmed = part.trim();
|
|
if (!trimmed) continue;
|
|
try {
|
|
const message = JSON.parse(trimmed) as BackendMessageMap[BackendMessageType];
|
|
this.handleParsedMessage(message);
|
|
} catch {
|
|
// Ignore parse errors for individual messages
|
|
}
|
|
}
|
|
};
|
|
|
|
socket.onerror = () => (this.socket === socket ? this.setStatus("error") : undefined);
|
|
|
|
socket.onclose = (event) => {
|
|
if (this.socket !== socket) return;
|
|
this.socket = null;
|
|
this.handleDisconnect(event);
|
|
};
|
|
}
|
|
|
|
disconnect() {
|
|
this.intentionalClose = true;
|
|
this.clearReconnectTimer();
|
|
if (this.socket) {
|
|
this.socket.close();
|
|
this.socket = null;
|
|
}
|
|
this.setStatus("disconnected");
|
|
this.resetSessionSubscriptionReadiness(new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR));
|
|
this.cleanupPendingRequests();
|
|
}
|
|
|
|
send(payload: unknown) {
|
|
const data = JSON.stringify(payload);
|
|
if (isDebug()) {
|
|
const p = payload as { action?: string; id?: string; type?: string } | null;
|
|
const action = p?.action ?? "?";
|
|
if (!DISPATCH_LOG_DENYLIST.has(action)) {
|
|
const sessionId = (p as { payload?: { session_id?: string } } | null)?.payload?.session_id;
|
|
debugDispatch("send", {
|
|
action,
|
|
id: p?.id ?? null,
|
|
type: p?.type ?? null,
|
|
sessionId: sessionId ?? null,
|
|
queued: this.status !== "connected" || !this.socket,
|
|
});
|
|
}
|
|
}
|
|
if (this.status !== "connected" || !this.socket) {
|
|
this.pendingQueue.push(data);
|
|
return;
|
|
}
|
|
this.socket.send(data);
|
|
}
|
|
|
|
request<T>(action: string, payload: unknown, timeoutMs = 5000): Promise<T> {
|
|
const id = generateUUID();
|
|
return new Promise((resolve, reject) => {
|
|
const timeout = setTimeout(() => {
|
|
this.pendingRequests.delete(id);
|
|
reject(new WebSocketRequestTimeoutError(action));
|
|
}, timeoutMs);
|
|
this.pendingRequests.set(id, {
|
|
resolve: resolve as (payload: unknown) => void,
|
|
reject,
|
|
timeout,
|
|
});
|
|
this.send({ id, type: "request", action, payload });
|
|
});
|
|
}
|
|
|
|
subscribe(taskId: string) {
|
|
const currentCount = this.subscriptions.get(taskId) ?? 0;
|
|
const nextCount = currentCount + 1;
|
|
this.subscriptions.set(taskId, nextCount);
|
|
if (this.status === "connected" && nextCount === 1) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "task.subscribe",
|
|
payload: { task_id: taskId },
|
|
});
|
|
}
|
|
return () => this.unsubscribe(taskId);
|
|
}
|
|
|
|
subscribeSession(sessionId: string) {
|
|
return this.subscribeSessionWithReady(sessionId).unsubscribe;
|
|
}
|
|
|
|
subscribeSessionWithReady(sessionId: string): SessionSubscriptionHandle {
|
|
const currentCount = this.sessionSubscriptions.get(sessionId) ?? 0;
|
|
const nextCount = currentCount + 1;
|
|
this.sessionSubscriptions.set(sessionId, nextCount);
|
|
const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId);
|
|
if (this.status === "connected" && this.socket) {
|
|
this.startSessionSubscription(sessionId, readiness);
|
|
}
|
|
return {
|
|
ready: readiness.promise,
|
|
unsubscribe: () => this.unsubscribeSession(sessionId),
|
|
};
|
|
}
|
|
|
|
resubscribeSession(sessionId: string): Promise<void> {
|
|
if (!this.sessionSubscriptions.has(sessionId)) return Promise.resolve();
|
|
const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId, true);
|
|
if (this.status === "connected" && this.socket) {
|
|
this.startSessionSubscription(sessionId, readiness);
|
|
}
|
|
return readiness.promise;
|
|
}
|
|
|
|
getSessionSubscriptionReadiness(sessionId: string): Promise<void> {
|
|
if (!this.sessionSubscriptions.has(sessionId)) return Promise.resolve();
|
|
if (
|
|
!this.sessionSubscriptionReadiness.has(sessionId) &&
|
|
(this.status === "disconnected" || this.status === "error")
|
|
) {
|
|
const unavailable = Promise.reject<void>(new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR));
|
|
void unavailable.catch(() => undefined);
|
|
return unavailable;
|
|
}
|
|
const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId);
|
|
if (this.status === "connected" && this.socket) {
|
|
this.startSessionSubscription(sessionId, readiness);
|
|
}
|
|
return readiness.promise;
|
|
}
|
|
|
|
focusSession(sessionId: string) {
|
|
const currentCount = this.sessionFocusCounts.get(sessionId) ?? 0;
|
|
const nextCount = currentCount + 1;
|
|
this.sessionFocusCounts.set(sessionId, nextCount);
|
|
if (this.status === "connected" && nextCount === 1) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "session.focus",
|
|
payload: { session_id: sessionId },
|
|
});
|
|
}
|
|
return () => this.unfocusSession(sessionId);
|
|
}
|
|
|
|
/**
|
|
* Request a fresh git snapshot for an already-focused session without
|
|
* changing its focus ref-count. Focus itself is intentionally ACK-only;
|
|
* explicit git refresh keeps tab activation from replaying session data during
|
|
* ordinary task switching.
|
|
*/
|
|
refreshSessionData(sessionId: string) {
|
|
if (this.status !== "connected") return;
|
|
if (!this.sessionFocusCounts.get(sessionId)) return;
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "session.git.refresh",
|
|
payload: { session_id: sessionId },
|
|
});
|
|
}
|
|
|
|
unfocusSession(sessionId: string) {
|
|
const currentCount = this.sessionFocusCounts.get(sessionId);
|
|
if (!currentCount) return;
|
|
const nextCount = currentCount - 1;
|
|
if (nextCount <= 0) {
|
|
this.sessionFocusCounts.delete(sessionId);
|
|
if (this.status === "connected") {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "session.unfocus",
|
|
payload: { session_id: sessionId },
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
this.sessionFocusCounts.set(sessionId, nextCount);
|
|
}
|
|
|
|
subscribeUser() {
|
|
this.userSubscriptionCount += 1;
|
|
if (this.status === "connected" && this.userSubscriptionCount === 1) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "user.subscribe",
|
|
payload: {},
|
|
});
|
|
}
|
|
}
|
|
|
|
unsubscribe(taskId: string) {
|
|
const currentCount = this.subscriptions.get(taskId);
|
|
if (!currentCount) return;
|
|
const nextCount = currentCount - 1;
|
|
if (nextCount <= 0) {
|
|
this.subscriptions.delete(taskId);
|
|
if (this.status === "connected") {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "task.unsubscribe",
|
|
payload: { task_id: taskId },
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
this.subscriptions.set(taskId, nextCount);
|
|
}
|
|
|
|
unsubscribeSession(sessionId: string) {
|
|
const currentCount = this.sessionSubscriptions.get(sessionId);
|
|
if (!currentCount) return;
|
|
const nextCount = currentCount - 1;
|
|
|
|
if (nextCount <= 0) {
|
|
this.sessionSubscriptions.delete(sessionId);
|
|
this.cancelSessionSubscriptionReadiness(sessionId);
|
|
const stream = this.coreSessionStreams.get(sessionId);
|
|
if (
|
|
this.status === "connected" &&
|
|
stream &&
|
|
this.reconnectOptions.conversationProtocol === "v1"
|
|
) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: SESSION_UNSUBSCRIBE_ACTION,
|
|
payload: {
|
|
session_id: sessionId,
|
|
consumer_kind: "core",
|
|
wire_id: stream.wireId,
|
|
},
|
|
});
|
|
}
|
|
if (
|
|
this.status === "connected" &&
|
|
stream &&
|
|
this.reconnectOptions.conversationProtocol === "v2"
|
|
) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "session.conversation.unsubscribe",
|
|
payload: {
|
|
scope_id: stream.conversationScopeId,
|
|
session_id: sessionId,
|
|
consumer_kind: "core",
|
|
},
|
|
});
|
|
}
|
|
clearTimeout(stream?.conversationCheckTimer);
|
|
this.coreSessionStreams.delete(sessionId);
|
|
this.coreConversationSessions.delete(sessionId);
|
|
this.coreSessionRecoveryHandlers.delete(sessionId);
|
|
if (this.status === "connected") {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: SESSION_UNSUBSCRIBE_ACTION,
|
|
payload: { session_id: sessionId },
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
this.sessionSubscriptions.set(sessionId, nextCount);
|
|
}
|
|
|
|
/**
|
|
* Subscribe to office run-event notifications for the given run id.
|
|
* Ref-counted so multiple components can subscribe independently;
|
|
* only the first call sends `run.subscribe` over the wire, only the
|
|
* last unsubscribe sends `run.unsubscribe`. Returns an
|
|
* unsubscribe function.
|
|
*/
|
|
subscribeRun(runId: string) {
|
|
const currentCount = this.runSubscriptions.get(runId) ?? 0;
|
|
const nextCount = currentCount + 1;
|
|
this.runSubscriptions.set(runId, nextCount);
|
|
if (this.status === "connected" && nextCount === 1) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "run.subscribe",
|
|
payload: { run_id: runId },
|
|
});
|
|
}
|
|
return () => this.unsubscribeRun(runId);
|
|
}
|
|
|
|
subscribeSystemMetrics() {
|
|
this.systemMetricsSubscriptionCount += 1;
|
|
if (this.status === "connected" && this.systemMetricsSubscriptionCount === 1) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "system.metrics.subscribe",
|
|
payload: {},
|
|
});
|
|
}
|
|
return () => this.unsubscribeSystemMetrics();
|
|
}
|
|
|
|
unsubscribeSystemMetrics() {
|
|
this.systemMetricsSubscriptionCount = Math.max(0, this.systemMetricsSubscriptionCount - 1);
|
|
if (this.status === "connected" && this.systemMetricsSubscriptionCount === 0) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "system.metrics.unsubscribe",
|
|
payload: {},
|
|
});
|
|
}
|
|
}
|
|
|
|
unsubscribeRun(runId: string) {
|
|
const currentCount = this.runSubscriptions.get(runId);
|
|
if (!currentCount) return;
|
|
const nextCount = currentCount - 1;
|
|
if (nextCount <= 0) {
|
|
this.runSubscriptions.delete(runId);
|
|
if (this.status === "connected") {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "run.unsubscribe",
|
|
payload: { run_id: runId },
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
this.runSubscriptions.set(runId, nextCount);
|
|
}
|
|
|
|
unsubscribeUser() {
|
|
this.userSubscriptionCount = Math.max(0, this.userSubscriptionCount - 1);
|
|
if (this.status === "connected" && this.userSubscriptionCount === 0) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "user.unsubscribe",
|
|
payload: {},
|
|
});
|
|
}
|
|
}
|
|
|
|
on<T extends BackendMessageType>(type: T, handler: MessageHandler<T>) {
|
|
const handlers = this.handlers.get(type) ?? new Set();
|
|
handlers.add(handler as MessageHandler<BackendMessageType>);
|
|
this.handlers.set(type, handlers);
|
|
return () => this.off(type, handler);
|
|
}
|
|
|
|
off<T extends BackendMessageType>(type: T, handler: MessageHandler<T>) {
|
|
const handlers = this.handlers.get(type);
|
|
if (!handlers) return;
|
|
handlers.delete(handler as MessageHandler<BackendMessageType>);
|
|
if (!handlers.size) {
|
|
this.handlers.delete(type);
|
|
}
|
|
}
|
|
|
|
onConnectionStatus(handler: (status: WebSocketStatus) => void) {
|
|
this.statusHandlers.add(handler);
|
|
return () => this.statusHandlers.delete(handler);
|
|
}
|
|
|
|
onRawSessionEvent(handler: (event: RawSessionEvent) => void) {
|
|
this.rawSessionEventHandlers.add(handler);
|
|
return () => {
|
|
this.rawSessionEventHandlers.delete(handler);
|
|
};
|
|
}
|
|
|
|
registerCoreSessionRecovery(sessionId: string, handler: CoreSessionRecoveryHandler) {
|
|
const handlers = this.coreSessionRecoveryHandlers.get(sessionId) ?? new Set();
|
|
handlers.add(handler);
|
|
this.coreSessionRecoveryHandlers.set(sessionId, handlers);
|
|
const stream = this.coreSessionStreams.get(sessionId);
|
|
if (stream?.needsHydration) {
|
|
void this.continueCoreSessionRecovery(sessionId, stream);
|
|
}
|
|
return () => {
|
|
const current = this.coreSessionRecoveryHandlers.get(sessionId);
|
|
if (!current) return;
|
|
current.delete(handler);
|
|
if (current.size === 0) this.coreSessionRecoveryHandlers.delete(sessionId);
|
|
};
|
|
}
|
|
|
|
retryCoreSessionRecovery(sessionId: string): Promise<boolean> | undefined {
|
|
const stream = this.coreSessionStreams.get(sessionId);
|
|
if (!stream?.needsHydration) return undefined;
|
|
return this.continueCoreSessionRecovery(sessionId, stream);
|
|
}
|
|
|
|
private debugNotification(action: BackendMessageType, payload: unknown, handlerCount: number) {
|
|
if (!isDebug() || DISPATCH_LOG_DENYLIST.has(action)) return;
|
|
const payloadSessionId = (payload as { session_id?: string } | undefined)?.session_id;
|
|
debugDispatch("notification", {
|
|
action,
|
|
sessionId: payloadSessionId ?? null,
|
|
handlers: handlerCount,
|
|
});
|
|
}
|
|
|
|
private handleParsedMessage(value: unknown) {
|
|
if (isRawSessionEvent(value)) {
|
|
const disposition = orderedCoreDisposition(value);
|
|
this.handleCoreSessionEvent(value, disposition);
|
|
this.rawSessionEventHandlers.forEach((handler) => handler(value));
|
|
return;
|
|
}
|
|
if (this.handleControlFrame(value)) return;
|
|
const message = this.notificationMessage(value);
|
|
if (!message) return;
|
|
const action = message.action;
|
|
if (action === SESSION_CONVERSATION_CHANGED_ACTION) {
|
|
this.dispatchConversationChanged(message);
|
|
return;
|
|
}
|
|
const payloadSessionID =
|
|
message.payload && typeof message.payload === "object"
|
|
? (message.payload as { session_id?: unknown }).session_id
|
|
: undefined;
|
|
if (this.shouldSuppressCoreConversationNotification(action, payloadSessionID)) return;
|
|
const handlers = this.handlers.get(action);
|
|
this.debugNotification(action, message.payload, handlers?.size ?? 0);
|
|
if (handlers) {
|
|
handlers.forEach((handler) => handler(message));
|
|
}
|
|
dispatchToPluginWsHandlers(action, message.payload);
|
|
}
|
|
|
|
private handleControlFrame(value: unknown): boolean {
|
|
if (this.handleMalformedSessionEnvelope(value)) return true;
|
|
if (!value || typeof value !== "object" || !("type" in value)) return true;
|
|
return this.handleRequestResult(value as RawWebSocketMessage);
|
|
}
|
|
|
|
private notificationMessage(value: unknown): BackendMessageMap[BackendMessageType] | null {
|
|
if (!value || typeof value !== "object" || !("type" in value)) return null;
|
|
const message = value as BackendMessageMap[BackendMessageType];
|
|
if (message.type !== "notification" || !message.action) return null;
|
|
return message;
|
|
}
|
|
|
|
private dispatchConversationChanged(
|
|
message: BackendMessageMap[typeof SESSION_CONVERSATION_CHANGED_ACTION],
|
|
) {
|
|
this.handleCoreConversationChanged(message.payload);
|
|
const handlers = this.handlers.get(SESSION_CONVERSATION_CHANGED_ACTION);
|
|
this.debugNotification(
|
|
SESSION_CONVERSATION_CHANGED_ACTION,
|
|
message.payload,
|
|
handlers?.size ?? 0,
|
|
);
|
|
handlers?.forEach((handler) => handler(message));
|
|
dispatchToPluginWsHandlers(SESSION_CONVERSATION_CHANGED_ACTION, message.payload);
|
|
}
|
|
|
|
private shouldSuppressCoreConversationNotification(
|
|
action: BackendMessageType,
|
|
sessionId: unknown,
|
|
): boolean {
|
|
return (
|
|
this.reconnectOptions.conversationProtocol === "v2" &&
|
|
typeof sessionId === "string" &&
|
|
this.coreConversationSessions.has(sessionId) &&
|
|
isConversationNotificationAction(action)
|
|
);
|
|
}
|
|
|
|
private handleCoreConversationChanged(value: unknown) {
|
|
const parsed = parseCoreConversationChange(value);
|
|
if (!parsed) return;
|
|
const { sessionId } = parsed;
|
|
const stream = this.coreSessionStreams.get(sessionId);
|
|
if (!stream || this.reconnectOptions.conversationProtocol !== "v2") return;
|
|
if (this.handleCoreConversationControl(parsed, stream)) return;
|
|
if (needsCoreConversationRecovery(parsed, stream)) {
|
|
this.beginCoreConversationRecovery(
|
|
sessionId,
|
|
stream,
|
|
stream.conversationAppliedRevision ?? "0",
|
|
);
|
|
return;
|
|
}
|
|
if (!parsed.valid) return;
|
|
const validPayload = parsed.payload as ConversationChangedPayload;
|
|
if (!stream.ready || stream.needsHydration || stream.projectionPaused) {
|
|
stream.conversationPendingChanges.push(validPayload);
|
|
this.trimCoreConversationPendingChanges(sessionId, stream, validPayload.revision);
|
|
return;
|
|
}
|
|
const applied = stream.conversationAppliedRevision ?? "0";
|
|
const revision = validPayload.revision;
|
|
if (BigInt(revision) <= BigInt(applied)) return;
|
|
if (hasCoreConversationRevisionGap(validPayload, stream, applied)) {
|
|
stream.conversationPendingChanges.push(validPayload);
|
|
this.beginCoreConversationRecovery(sessionId, stream, revision);
|
|
return;
|
|
}
|
|
stream.conversationAppliedRevision = revision;
|
|
this.projectCoreConversationOperations(sessionId, validPayload.operations ?? []);
|
|
}
|
|
|
|
private handleCoreConversationControl(
|
|
parsed: ParsedCoreConversationChange,
|
|
stream: CoreSessionStream,
|
|
): boolean {
|
|
const { sessionId } = parsed;
|
|
if (parsed.payload.scope_id !== stream.conversationScopeId) return true;
|
|
if (parsed.payload.terminal) {
|
|
clearTimeout(stream.conversationCheckTimer);
|
|
stream.projectionPaused = true;
|
|
stream.conversationPendingChanges.length = 0;
|
|
this.handlers.get("session.removed")?.forEach((handler) =>
|
|
handler({
|
|
type: "notification",
|
|
action: "session.removed",
|
|
payload: { session_id: sessionId },
|
|
}),
|
|
);
|
|
return true;
|
|
}
|
|
if (parsed.payload.check) {
|
|
this.checkCoreConversationRevision(sessionId, stream, parsed.payload);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private checkCoreConversationRevision(
|
|
sessionId: string,
|
|
stream: CoreSessionStream,
|
|
payload: Partial<ConversationChangedPayload>,
|
|
) {
|
|
const applied = stream.conversationAppliedRevision ?? "0";
|
|
if (
|
|
!isDecimalRevision(payload.revision) ||
|
|
payload.epoch !== stream.conversationEpoch ||
|
|
BigInt(payload.revision) < BigInt(applied)
|
|
) {
|
|
this.beginCoreConversationRecovery(sessionId, stream, applied);
|
|
return;
|
|
}
|
|
if (payload.revision === applied || stream.conversationCheckTimer !== undefined) return;
|
|
const observed = payload.revision;
|
|
stream.conversationCheckTimer = setTimeout(() => {
|
|
stream.conversationCheckTimer = undefined;
|
|
if (this.coreSessionStreams.get(sessionId) !== stream || this.status !== "connected") return;
|
|
if (BigInt(observed) > BigInt(stream.conversationAppliedRevision ?? "0")) {
|
|
this.beginCoreConversationRecovery(sessionId, stream, observed);
|
|
}
|
|
}, 1000);
|
|
}
|
|
|
|
private trimCoreConversationPendingChanges(
|
|
sessionId: string,
|
|
stream: CoreSessionStream,
|
|
recoveryRevision: string,
|
|
) {
|
|
if (stream.conversationPendingChanges.length <= 256) return;
|
|
stream.conversationPendingChanges.splice(0, stream.conversationPendingChanges.length - 256);
|
|
this.beginCoreConversationRecovery(sessionId, stream, recoveryRevision);
|
|
}
|
|
|
|
private drainCoreConversationChanges(sessionId: string, stream: CoreSessionStream) {
|
|
if (!stream.ready || stream.needsHydration || stream.projectionPaused) return;
|
|
const pending = stream.conversationPendingChanges.splice(0);
|
|
for (const change of pending) {
|
|
this.handleCoreConversationChanged(change);
|
|
if (stream.needsHydration || stream.projectionPaused) return;
|
|
}
|
|
if (this.coreSessionStreams.get(sessionId) !== stream) return;
|
|
}
|
|
|
|
private projectCoreConversationOperations(sessionId: string, operations: unknown[]) {
|
|
for (const value of operations) {
|
|
if (!value || typeof value !== "object" || Array.isArray(value)) continue;
|
|
const operation = value as CoreConversationOperation;
|
|
if (!isCoreConversationOperation(operation)) continue;
|
|
const projected = projectCoreConversationOperation(sessionId, operation);
|
|
this.dispatchConversationNotification(projected.action, projected.payload);
|
|
}
|
|
}
|
|
|
|
private dispatchConversationNotification(action: BackendMessageType, payload: unknown) {
|
|
const handlers = this.handlers.get(action);
|
|
this.debugNotification(action, payload, handlers?.size ?? 0);
|
|
handlers?.forEach((handler) => handler({ type: "notification", action, payload } as never));
|
|
dispatchToPluginWsHandlers(action, payload);
|
|
}
|
|
|
|
private beginCoreConversationRecovery(
|
|
sessionId: string,
|
|
stream: CoreSessionStream,
|
|
revision: string,
|
|
) {
|
|
stream.projectionPaused = true;
|
|
stream.needsHydration = true;
|
|
if (
|
|
!stream.conversationRecoveryRevision ||
|
|
BigInt(revision) > BigInt(stream.conversationRecoveryRevision)
|
|
) {
|
|
stream.conversationRecoveryRevision = revision;
|
|
}
|
|
stream.recoveryGeneration += 1;
|
|
void this.continueCoreSessionRecovery(sessionId, stream);
|
|
}
|
|
|
|
/**
|
|
* A frame that is session.event-shaped but fails the strict envelope check
|
|
* (bad protocol version, missing identity fields, malformed payload) can
|
|
* never be a valid ordered row. If it targets a live core stream at the
|
|
* expected next sequence, run the same durable poison recovery as a poison
|
|
* disposition so the stream rebinds to the authoritative watermark instead
|
|
* of stalling silently on a hole; the malformed frame itself is never
|
|
* dispatched to consumers.
|
|
*/
|
|
private handleMalformedSessionEnvelope(value: unknown): boolean {
|
|
if (!value || typeof value !== "object") return false;
|
|
const candidate = value as {
|
|
type?: unknown;
|
|
session_id?: unknown;
|
|
task_id?: unknown;
|
|
sequence?: unknown;
|
|
};
|
|
if (candidate.type !== "session.event") return false;
|
|
const sessionId = candidate.session_id;
|
|
const sequence = candidate.sequence;
|
|
const taskId = candidate.task_id;
|
|
const stream =
|
|
typeof sessionId === "string" && typeof sequence === "number"
|
|
? this.coreSessionStreams.get(sessionId)
|
|
: undefined;
|
|
if (stream && sequence === stream.lastSeenSequence + 1) {
|
|
if (isDebug()) {
|
|
console.warn(
|
|
`[ws] malformed ordered session.event frame for "${sessionId}" at ${sequence}`,
|
|
);
|
|
}
|
|
this.recoverCoreSessionPoison(sessionId as string, stream);
|
|
}
|
|
if (
|
|
typeof sessionId === "string" &&
|
|
Number.isSafeInteger(sequence) &&
|
|
(sequence as number) > 0
|
|
) {
|
|
this.notifyMalformedSessionEvent(
|
|
sessionId,
|
|
sequence as number,
|
|
typeof taskId === "string" ? taskId : null,
|
|
);
|
|
}
|
|
return true;
|
|
}
|
|
private notifyMalformedSessionEvent(sessionId: string, sequence: number, taskId: string | null) {
|
|
const poison: RawSessionEvent = {
|
|
type: "session.event",
|
|
protocol_version: 1,
|
|
event_type: "session.event.poison",
|
|
session_id: sessionId,
|
|
task_id: taskId,
|
|
sequence,
|
|
event_id: `poison:${sessionId}:${sequence}`,
|
|
payload: {
|
|
type: "session.event.poison",
|
|
session_id: sessionId,
|
|
task_id: taskId,
|
|
},
|
|
};
|
|
this.rawSessionEventHandlers.forEach((handler) => handler(poison));
|
|
}
|
|
|
|
private handleRequestResult(message: RawWebSocketMessage): boolean {
|
|
if (message.type !== "response" && message.type !== "error") return false;
|
|
if (typeof message.id !== "string") return false;
|
|
if (message.type === "response") {
|
|
if (isDebug()) debugDispatch("response", { id: message.id });
|
|
this.resolvePendingRequest(message.id, message.payload);
|
|
} else {
|
|
if (isDebug()) debugDispatch("error-response", { id: message.id });
|
|
this.rejectPendingRequest(message.id, message.payload);
|
|
}
|
|
return true;
|
|
}
|
|
|
|
private resolvePendingRequest(msgId: string, payload: unknown) {
|
|
const pending = this.pendingRequests.get(msgId);
|
|
if (!pending) return;
|
|
clearTimeout(pending.timeout);
|
|
this.pendingRequests.delete(msgId);
|
|
pending.resolve(payload);
|
|
}
|
|
|
|
// i18n-exempt: transport/API diagnostic. Callers branch on the error code and
|
|
// render translated copy; this text only ever appears in a console or as an
|
|
// interpolated English diagnostic (see docs/i18n.md on interpolated values).
|
|
private rejectPendingRequest(msgId: string, payload: unknown) {
|
|
const pending = this.pendingRequests.get(msgId);
|
|
if (!pending) return;
|
|
clearTimeout(pending.timeout);
|
|
this.pendingRequests.delete(msgId);
|
|
pending.reject(toWebSocketRequestError(payload));
|
|
}
|
|
|
|
private handleDisconnect(event: CloseEvent) {
|
|
this.setStatus("disconnected");
|
|
const shouldRetainSessionReadiness =
|
|
!this.intentionalClose &&
|
|
this.reconnectOptions.enabled &&
|
|
this.reconnectAttempts < this.reconnectOptions.maxAttempts;
|
|
this.resetSessionSubscriptionReadiness(
|
|
new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR),
|
|
shouldRetainSessionReadiness,
|
|
);
|
|
|
|
// Don't reconnect if this was an intentional close
|
|
if (this.intentionalClose) {
|
|
return;
|
|
}
|
|
|
|
// Don't reconnect if reconnect is disabled
|
|
if (!this.reconnectOptions.enabled) {
|
|
return;
|
|
}
|
|
|
|
// Don't reconnect if we've exceeded max attempts
|
|
if (this.reconnectAttempts >= this.reconnectOptions.maxAttempts) {
|
|
console.warn(
|
|
`WebSocket max reconnect attempts (${this.reconnectOptions.maxAttempts}) reached`,
|
|
);
|
|
this.setStatus("error");
|
|
this.cleanupPendingRequests();
|
|
return;
|
|
}
|
|
|
|
// Calculate delay with exponential backoff
|
|
const delay = Math.min(
|
|
this.reconnectOptions.initialDelay *
|
|
Math.pow(this.reconnectOptions.backoffMultiplier, this.reconnectAttempts),
|
|
this.reconnectOptions.maxDelay,
|
|
);
|
|
|
|
this.reconnectAttempts++;
|
|
this.setStatus("reconnecting");
|
|
|
|
console.log(
|
|
`WebSocket disconnected (code: ${event.code}, reason: ${event.reason || "none"}). ` +
|
|
`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts}/${this.reconnectOptions.maxAttempts})...`,
|
|
);
|
|
|
|
this.reconnectTimer = setTimeout(() => {
|
|
this.connect();
|
|
}, delay);
|
|
}
|
|
|
|
private clearReconnectTimer() {
|
|
if (this.reconnectTimer) {
|
|
clearTimeout(this.reconnectTimer);
|
|
this.reconnectTimer = null;
|
|
}
|
|
}
|
|
|
|
private cleanupPendingRequests() {
|
|
// Reject all pending requests
|
|
this.pendingRequests.forEach(({ reject, timeout }) => {
|
|
clearTimeout(timeout);
|
|
reject(new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR));
|
|
});
|
|
this.pendingRequests.clear();
|
|
}
|
|
|
|
private getOrCreateSessionSubscriptionReadiness(
|
|
sessionId: string,
|
|
forceNewAfterSettled = false,
|
|
): SessionSubscriptionReadiness {
|
|
const existing = this.sessionSubscriptionReadiness.get(sessionId);
|
|
if (existing && !(forceNewAfterSettled && existing.settled)) return existing;
|
|
|
|
let resolve!: () => void;
|
|
let reject!: (reason: unknown) => void;
|
|
const promise = new Promise<void>((resolvePromise, rejectPromise) => {
|
|
resolve = resolvePromise;
|
|
reject = rejectPromise;
|
|
});
|
|
const readiness: SessionSubscriptionReadiness = {
|
|
promise,
|
|
resolve,
|
|
reject,
|
|
requestStarted: false,
|
|
settled: false,
|
|
attempt: 0,
|
|
retryTimer: null,
|
|
};
|
|
// subscribeSession() consumers do not await readiness, so handle failures
|
|
// while returning the original promise to readiness-aware consumers.
|
|
void promise.catch(() => undefined);
|
|
this.sessionSubscriptionReadiness.set(sessionId, readiness);
|
|
return readiness;
|
|
}
|
|
|
|
private startSessionSubscription(sessionId: string, readiness: SessionSubscriptionReadiness) {
|
|
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
|
|
if (readiness.requestStarted) return;
|
|
readiness.requestStarted = true;
|
|
readiness.attempt += 1;
|
|
const stream = this.getOrCreateCoreSessionStream(sessionId);
|
|
if (this.reconnectOptions.conversationProtocol === "v2") {
|
|
this.startSourceSessionSubscription(sessionId, readiness, stream);
|
|
return;
|
|
}
|
|
stream.ready = false;
|
|
stream.pendingEvents.length = 0;
|
|
const legacySubscription = this.request(
|
|
SESSION_SUBSCRIBE_ACTION,
|
|
{ session_id: sessionId },
|
|
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
|
|
);
|
|
const orderedSubscription = this.request<unknown>(
|
|
SESSION_SUBSCRIBE_ACTION,
|
|
{
|
|
session_id: sessionId,
|
|
consumer_kind: "core",
|
|
wire_id: stream.wireId,
|
|
...(stream.resumeToken || stream.lastSeenSequence > 0
|
|
? { last_seen_sequence: stream.lastSeenSequence }
|
|
: {}),
|
|
...(stream.resumeToken ? { resume_token: stream.resumeToken } : {}),
|
|
},
|
|
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
|
|
).then((response) => {
|
|
const validated = validateOrderedSubscribeResponse(response, sessionId, stream.wireId);
|
|
const recoveryRequired = validated.result === "invalid_resume" || stream.needsHydration;
|
|
if (recoveryRequired) {
|
|
stream.projectionPaused = true;
|
|
stream.needsHydration = true;
|
|
stream.recoveryGeneration += 1;
|
|
stream.recoveryWatermark = validated.eventWatermark;
|
|
} else if (validated.result !== "replay") {
|
|
stream.lastSeenSequence = Math.max(stream.lastSeenSequence, validated.eventWatermark);
|
|
}
|
|
stream.resumeToken = validated.resumeToken;
|
|
stream.ready = true;
|
|
this.drainCoreSessionEvents(sessionId, stream);
|
|
});
|
|
void Promise.all([legacySubscription, orderedSubscription])
|
|
.then(() => {
|
|
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
|
|
readiness.retryTimer = null;
|
|
readiness.settled = true;
|
|
readiness.resolve();
|
|
const stream = this.coreSessionStreams.get(sessionId);
|
|
if (stream?.needsHydration) {
|
|
void this.continueCoreSessionRecovery(sessionId, stream);
|
|
}
|
|
})
|
|
.catch((error: unknown) => {
|
|
stream.ready = false;
|
|
stream.pendingEvents.length = 0;
|
|
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
|
|
readiness.requestStarted = false;
|
|
if (
|
|
isWebSocketRequestTimeoutError(error) &&
|
|
readiness.attempt < MAX_SESSION_SUBSCRIPTION_ATTEMPTS &&
|
|
this.status === "connected" &&
|
|
this.socket
|
|
) {
|
|
readiness.retryTimer = setTimeout(() => {
|
|
readiness.retryTimer = null;
|
|
this.startSessionSubscription(sessionId, readiness);
|
|
}, SESSION_ENTRY_RETRY_DELAY_MS);
|
|
return;
|
|
}
|
|
this.sessionSubscriptionReadiness.delete(sessionId);
|
|
readiness.settled = true;
|
|
readiness.reject(error);
|
|
});
|
|
}
|
|
|
|
private startSourceSessionSubscription(
|
|
sessionId: string,
|
|
readiness: SessionSubscriptionReadiness,
|
|
stream: CoreSessionStream,
|
|
) {
|
|
stream.ready = false;
|
|
stream.conversationPendingChanges.length = 0;
|
|
const legacySubscription = this.request(
|
|
SESSION_SUBSCRIBE_ACTION,
|
|
{ session_id: sessionId },
|
|
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
|
|
);
|
|
const conversationSubscription = this.request<unknown>(
|
|
"session.conversation.subscribe",
|
|
{
|
|
scope_id: stream.conversationScopeId,
|
|
session_id: sessionId,
|
|
consumer_kind: "core",
|
|
},
|
|
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
|
|
).then((response) => {
|
|
const payload = response as Partial<ConversationChangedPayload> & {
|
|
success?: boolean;
|
|
error?: unknown;
|
|
};
|
|
if (
|
|
payload.success !== true ||
|
|
payload.protocol_version !== 2 ||
|
|
payload.scope_id !== stream.conversationScopeId ||
|
|
payload.session_id !== sessionId ||
|
|
typeof payload.epoch !== "string" ||
|
|
typeof payload.revision !== "string" ||
|
|
!/^(0|[1-9][0-9]*)$/.test(payload.revision)
|
|
) {
|
|
// i18n-exempt: transport/API diagnostic. The caller renders the
|
|
// structured error code through translated UI copy.
|
|
throw new Error("Invalid conversation subscription response");
|
|
}
|
|
const hadConversationState = stream.conversationEpoch !== undefined;
|
|
const missedConversationState =
|
|
hadConversationState &&
|
|
(stream.conversationEpoch !== payload.epoch ||
|
|
BigInt(payload.revision) > BigInt(stream.conversationAppliedRevision ?? "0"));
|
|
stream.conversationEpoch = payload.epoch;
|
|
if (!missedConversationState) stream.conversationAppliedRevision = payload.revision;
|
|
if (missedConversationState) {
|
|
stream.needsHydration = true;
|
|
stream.projectionPaused = true;
|
|
stream.conversationRecoveryRevision = payload.revision;
|
|
stream.recoveryGeneration += 1;
|
|
}
|
|
stream.ready = true;
|
|
this.coreConversationSessions.add(sessionId);
|
|
this.drainCoreConversationChanges(sessionId, stream);
|
|
});
|
|
void Promise.all([legacySubscription, conversationSubscription])
|
|
.then(() => {
|
|
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
|
|
readiness.retryTimer = null;
|
|
readiness.settled = true;
|
|
readiness.resolve();
|
|
if (stream.needsHydration) void this.continueCoreSessionRecovery(sessionId, stream);
|
|
})
|
|
.catch((error: unknown) => {
|
|
stream.ready = false;
|
|
stream.conversationPendingChanges.length = 0;
|
|
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
|
|
readiness.requestStarted = false;
|
|
if (
|
|
isWebSocketRequestTimeoutError(error) &&
|
|
readiness.attempt < MAX_SESSION_SUBSCRIPTION_ATTEMPTS &&
|
|
this.status === "connected" &&
|
|
this.socket
|
|
) {
|
|
readiness.retryTimer = setTimeout(() => {
|
|
readiness.retryTimer = null;
|
|
this.startSessionSubscription(sessionId, readiness);
|
|
}, SESSION_ENTRY_RETRY_DELAY_MS);
|
|
return;
|
|
}
|
|
this.sessionSubscriptionReadiness.delete(sessionId);
|
|
readiness.settled = true;
|
|
readiness.reject(error);
|
|
});
|
|
}
|
|
|
|
private getOrCreateCoreSessionStream(sessionId: string): CoreSessionStream {
|
|
const current = this.coreSessionStreams.get(sessionId);
|
|
if (current) return current;
|
|
const stream = {
|
|
wireId: `core:web:${generateUUID()}`,
|
|
conversationScopeId: `core:web:${generateUUID()}`,
|
|
lastSeenSequence: 0,
|
|
ready: false,
|
|
pendingEvents: [],
|
|
projectionPaused: false,
|
|
needsHydration: false,
|
|
recoveryGeneration: 0,
|
|
conversationPendingChanges: [],
|
|
};
|
|
this.coreSessionStreams.set(sessionId, stream);
|
|
return stream;
|
|
}
|
|
|
|
private handleCoreSessionEvent(
|
|
event: RawSessionEvent,
|
|
disposition = orderedCoreDisposition(event),
|
|
) {
|
|
const stream = this.coreSessionStreams.get(event.session_id);
|
|
if (!stream) return;
|
|
if (stream.projectionPaused || stream.needsHydration) {
|
|
stream.pendingEvents.push(event);
|
|
return;
|
|
}
|
|
if (!stream.ready) {
|
|
stream.pendingEvents.push(event);
|
|
return;
|
|
}
|
|
if (stream.pendingEvents.length > 0) {
|
|
stream.pendingEvents.push(event);
|
|
this.drainCoreSessionEvents(event.session_id, stream);
|
|
return;
|
|
}
|
|
this.processCoreSessionEvent(event, stream, disposition);
|
|
this.drainCoreSessionEvents(event.session_id, stream);
|
|
}
|
|
|
|
private drainCoreSessionEvents(sessionId: string, stream: CoreSessionStream) {
|
|
if (
|
|
!stream.ready ||
|
|
stream.projectionPaused ||
|
|
stream.needsHydration ||
|
|
stream.pendingEvents.length === 0
|
|
)
|
|
return;
|
|
stream.pendingEvents.sort((left, right) => left.sequence - right.sequence);
|
|
while (stream.pendingEvents.length > 0) {
|
|
const event = stream.pendingEvents[0];
|
|
if (!event) return;
|
|
if (event.sequence > stream.lastSeenSequence + 1) {
|
|
stream.pendingEvents.length = 0;
|
|
this.recoverCoreSessionPoison(sessionId, stream);
|
|
return;
|
|
}
|
|
stream.pendingEvents.shift();
|
|
this.processCoreSessionEvent(event, stream);
|
|
}
|
|
}
|
|
|
|
private processCoreSessionEvent(
|
|
event: RawSessionEvent,
|
|
stream: CoreSessionStream,
|
|
disposition = orderedCoreDisposition(event),
|
|
) {
|
|
if (event.sequence <= stream.lastSeenSequence) {
|
|
void this.acknowledgeCoreSessionEvent(
|
|
event.session_id,
|
|
stream,
|
|
stream.lastSeenSequence,
|
|
).catch(() => this.recoverCoreSessionPoison(event.session_id, stream));
|
|
return;
|
|
}
|
|
if (event.sequence !== stream.lastSeenSequence + 1) {
|
|
this.recoverCoreSessionPoison(event.session_id, stream);
|
|
return;
|
|
}
|
|
if (disposition === "poison") {
|
|
this.recoverCoreSessionPoison(event.session_id, stream);
|
|
return;
|
|
}
|
|
if (disposition === "project") {
|
|
const action = orderedCoreAction(event.event_type);
|
|
if (!action) return;
|
|
const payload = projectCoreSessionPayload(event);
|
|
const message = {
|
|
type: "notification",
|
|
action,
|
|
payload,
|
|
} as BackendMessageMap[BackendMessageType];
|
|
const handlers = this.handlers.get(action);
|
|
this.debugNotification(action, payload, handlers?.size ?? 0);
|
|
handlers?.forEach((handler) => handler(message));
|
|
dispatchToPluginWsHandlers(action, payload);
|
|
}
|
|
stream.lastSeenSequence = event.sequence;
|
|
void this.acknowledgeCoreSessionEvent(event.session_id, stream, event.sequence).catch(() =>
|
|
this.recoverCoreSessionPoison(event.session_id, stream),
|
|
);
|
|
}
|
|
|
|
private recoverCoreSessionPoison(sessionId: string, stream: CoreSessionStream) {
|
|
if (stream.poisonRecovery) return;
|
|
stream.projectionPaused = true;
|
|
stream.recoveryGeneration += 1;
|
|
stream.recoveryWatermark = undefined;
|
|
stream.poisonRecovery = this.request(SESSION_SUBSCRIBE_ACTION, {
|
|
session_id: sessionId,
|
|
consumer_kind: "core",
|
|
wire_id: stream.wireId,
|
|
last_seen_sequence: stream.lastSeenSequence,
|
|
...(stream.resumeToken ? { resume_token: stream.resumeToken } : {}),
|
|
replace_cursor: true,
|
|
})
|
|
.then((response) => {
|
|
const validated = validateOrderedSubscribeResponse(response, sessionId, stream.wireId);
|
|
if (this.coreSessionStreams.get(sessionId) !== stream) return;
|
|
stream.resumeToken = validated.resumeToken;
|
|
stream.needsHydration = true;
|
|
stream.recoveryWatermark = validated.eventWatermark;
|
|
return this.continueCoreSessionRecovery(sessionId, stream);
|
|
})
|
|
.catch(() => undefined)
|
|
.finally(() => {
|
|
stream.poisonRecovery = undefined;
|
|
});
|
|
}
|
|
|
|
private continueCoreSessionRecovery(
|
|
sessionId: string,
|
|
stream: CoreSessionStream,
|
|
): Promise<boolean> {
|
|
if (
|
|
!stream.needsHydration ||
|
|
(this.reconnectOptions.conversationProtocol === "v1" &&
|
|
stream.recoveryWatermark === undefined) ||
|
|
(this.reconnectOptions.conversationProtocol === "v2" &&
|
|
stream.conversationRecoveryRevision === undefined)
|
|
) {
|
|
return Promise.resolve(false);
|
|
}
|
|
const existing = stream.recoveryHydration;
|
|
if (existing) return existing;
|
|
const handlers = [...(this.coreSessionRecoveryHandlers.get(sessionId) ?? [])];
|
|
if (handlers.length === 0) return Promise.resolve(false);
|
|
const generation = stream.recoveryGeneration;
|
|
const pending = Promise.all(handlers.map((handler) => Promise.resolve().then(() => handler())))
|
|
.then((results) => {
|
|
if (
|
|
this.coreSessionStreams.get(sessionId) !== stream ||
|
|
stream.recoveryGeneration !== generation ||
|
|
!results.every(Boolean)
|
|
) {
|
|
return false;
|
|
}
|
|
if (this.reconnectOptions.conversationProtocol === "v2") {
|
|
const revision = stream.conversationRecoveryRevision;
|
|
if (revision === undefined) return false;
|
|
stream.conversationAppliedRevision = revision;
|
|
stream.conversationPendingChanges = stream.conversationPendingChanges.filter((change) => {
|
|
if (!change || typeof change !== "object") return false;
|
|
const candidate = change as { revision?: unknown };
|
|
return (
|
|
typeof candidate.revision === "string" &&
|
|
BigInt(candidate.revision) > BigInt(revision)
|
|
);
|
|
});
|
|
stream.conversationRecoveryRevision = undefined;
|
|
} else {
|
|
const watermark = stream.recoveryWatermark;
|
|
if (watermark === undefined) return false;
|
|
stream.lastSeenSequence = Math.max(stream.lastSeenSequence, watermark);
|
|
stream.pendingEvents = stream.pendingEvents.filter(
|
|
(event) => event.sequence > stream.lastSeenSequence,
|
|
);
|
|
}
|
|
stream.needsHydration = false;
|
|
stream.projectionPaused = false;
|
|
stream.recoveryWatermark = undefined;
|
|
this.drainCoreConversationChanges(sessionId, stream);
|
|
this.drainCoreSessionEvents(sessionId, stream);
|
|
return true;
|
|
})
|
|
.catch(() => false)
|
|
.finally(() => {
|
|
if (stream.recoveryHydration !== pending) return;
|
|
stream.recoveryHydration = undefined;
|
|
if (stream.needsHydration && stream.recoveryGeneration !== generation) {
|
|
void this.continueCoreSessionRecovery(sessionId, stream);
|
|
}
|
|
});
|
|
stream.recoveryHydration = pending;
|
|
return pending;
|
|
}
|
|
|
|
private async acknowledgeCoreSessionEvent(
|
|
sessionId: string,
|
|
stream: CoreSessionStream,
|
|
sequence: number,
|
|
) {
|
|
const response = await this.request<unknown>("session.ack", {
|
|
session_id: sessionId,
|
|
consumer_kind: "core",
|
|
wire_id: stream.wireId,
|
|
sequence,
|
|
resume_token: stream.resumeToken,
|
|
});
|
|
stream.resumeToken = validateOrderedAckResponse(response, sessionId, stream.wireId, sequence);
|
|
}
|
|
private cancelSessionSubscriptionReadiness(sessionId: string) {
|
|
const readiness = this.sessionSubscriptionReadiness.get(sessionId);
|
|
if (!readiness) return;
|
|
this.sessionSubscriptionReadiness.delete(sessionId);
|
|
if (readiness.retryTimer) {
|
|
clearTimeout(readiness.retryTimer);
|
|
readiness.retryTimer = null;
|
|
}
|
|
if (!readiness.settled) {
|
|
readiness.settled = true;
|
|
readiness.reject(new Error(SESSION_SUBSCRIPTION_RELEASED_ERROR));
|
|
}
|
|
}
|
|
|
|
private resetSessionSubscriptionReadiness(error: Error, retainActiveSubscriptions = false) {
|
|
const readinessEntries = [...this.sessionSubscriptionReadiness.entries()];
|
|
this.sessionSubscriptionReadiness.clear();
|
|
for (const [, readiness] of readinessEntries) {
|
|
if (readiness.retryTimer) {
|
|
clearTimeout(readiness.retryTimer);
|
|
readiness.retryTimer = null;
|
|
}
|
|
if (readiness.settled) continue;
|
|
readiness.settled = true;
|
|
readiness.reject(error);
|
|
}
|
|
if (retainActiveSubscriptions) {
|
|
this.sessionSubscriptions.forEach((_, id) =>
|
|
this.getOrCreateSessionSubscriptionReadiness(id),
|
|
);
|
|
}
|
|
}
|
|
|
|
private resubscribe() {
|
|
// Re-subscribe to all tasks after reconnection
|
|
this.subscriptions.forEach((_count, taskId) => {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "task.subscribe",
|
|
payload: { task_id: taskId },
|
|
});
|
|
});
|
|
this.sessionSubscriptions.forEach((_count, sessionId) => {
|
|
const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId);
|
|
this.startSessionSubscription(sessionId, readiness);
|
|
});
|
|
this.sessionFocusCounts.forEach((_count, sessionId) => {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "session.focus",
|
|
payload: { session_id: sessionId },
|
|
});
|
|
});
|
|
this.runSubscriptions.forEach((_count, runId) => {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "run.subscribe",
|
|
payload: { run_id: runId },
|
|
});
|
|
});
|
|
if (this.userSubscriptionCount > 0) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "user.subscribe",
|
|
payload: {},
|
|
});
|
|
}
|
|
if (this.systemMetricsSubscriptionCount > 0) {
|
|
this.send({
|
|
id: generateUUID(),
|
|
type: "request",
|
|
action: "system.metrics.subscribe",
|
|
payload: {},
|
|
});
|
|
}
|
|
}
|
|
|
|
private flushQueue() {
|
|
if (!this.socket || this.status !== "connected") return;
|
|
this.pendingQueue.forEach((data) => this.socket?.send(data));
|
|
this.pendingQueue = [];
|
|
}
|
|
|
|
private setStatus(status: WebSocketStatus) {
|
|
this.status = status;
|
|
this.onStatusChange?.(status);
|
|
this.statusHandlers.forEach((handler) => handler(status));
|
|
}
|
|
}
|