} from "../daemon/daemon-client.js";
import {
DaemonSessionStartingError,
deserializeDaemonError,
isSessionClosingError,
isSessionNotReadyError,
isSessionStartingOrRecoveringError,
} from "../daemon/daemon-errors.js";
import {
1951 unmodified lines
Expand all
private async requestOk(command: DaemonCommandBody): Promise<void> {
await this.requestData<unknown>(command);
}
private resolvingActiveSessionPromise?: Promise<string>;
private async resolveCurrentActiveSessionId(deadline: number): Promise<string> {
if (this.resolvingActiveSessionPromise) {
return this.resolvingActiveSessionPromise;
}
this.resolvingActiveSessionPromise = (async () => {
let attempt = 0;
let lastError: unknown;
while (Date.now() < deadline && !this.disposed && !this.terminalCloseEmitted) {
try {
await this.client.reconnect(1000);
if (this.disposed || this.terminalCloseEmitted) break;
if (!this.updateRestartPending) {
try {
await this.attach({ recoverable: false });
return this.activeSessionId;
} catch (attachErr) {
if (isSessionStartingOrRecoveringError(attachErr) || isSessionClosingError(attachErr)) {
throw attachErr;
}
}
}
const listResponse = await this.client
.request({ type: "list" }, 30000, { recoverable: false })
.catch(() => undefined);
if (listResponse && listResponse.success) {
const sessions = readSessionSummaries(listResponse.data);
const restored = sessions.find(
(s) =>
(this.attachedSessionFile !== undefined && s.sessionFile === this.attachedSessionFile) ||
(this.attachedSessionId !== undefined && s.sessionId === this.attachedSessionId) ||
s.activeSessionId === this.activeSessionId,
);
if (restored?.activeSessionId) {
if (restored.activeSessionId !== this.activeSessionId) {
this.sessionRevision++;
this.activeSessionId = restored.activeSessionId;
this.lastEventSequence = undefined;
this.lastEventCursor = undefined;
this.retiredEventGenerations.clear();
}
await this.attach({ recoverable: false });
return this.activeSessionId;
}
lastError = new DaemonSessionStartingError(
this.activeSessionId,
"Session worker is queued for restore",
);
}
} catch (error) {
lastError = error;
if (!isSessionNotReadyError(error)) {
throw error;
}
}
const delayMs = Math.min(deadline - Date.now(), 2000, 250 * 2 ** Math.min(attempt++, 3));
await delay(delayMs);
}
throw (
lastError ??
new Error(
`Could not resolve active session for ${this.attachedSessionId ?? this.attachedSessionFile ?? this.activeSessionId}`,
)
);
})().finally(() => {
this.resolvingActiveSessionPromise = undefined;
});
return this.resolvingActiveSessionPromise;
}
private async requestData<T>(
command: DaemonCommandBody,
timeoutMs?: number,
options?: Parameters<DaemonTransportClient["request"]>[2],
): Promise<T> {
const isInternalTeardownOrRecovery =
options?.recoverable === false ||
command.type === "attach" ||
command.type === "list" ||
command.type === "detach" ||
command.type === "release_session_input_pause" ||
command.type === "complete_owned_session";
if (isInternalTeardownOrRecovery) {
const response = await this.client.request(command, timeoutMs, options);
if (!response.success) {
const error = deserializeDaemonError(response);
this.definitiveRequestErrors.add(error);
throw error;
const response = await this.client.request(command, timeoutMs, options);
if (!response.success) {
const error = deserializeDaemonError(response);
this.definitiveRequestErrors.add(error);
throw error;
}
if (command.type === "get_model_catalog" || command.type === "get_available_models") {
this.stateFreshnessGeneration++;
this.latestSnapshotStateIsFresh = false;
} else if (invalidatesCachedSnapshot(command.type)) {
this.latestSnapshotIsFresh = false;
}
return response.data as T;
}
const deadline = Date.now() + (this.options.reconnectTimeoutMs ?? DAEMON_RECONNECT_TIMEOUT_MS);
let attempt = 0;
while (Date.now() < deadline && !this.disposed && !this.terminalCloseEmitted) {
await this.waitForReconnectionIfPending(deadline);
if ("activeSessionId" in command && typeof command.activeSessionId === "string") {
(command as { activeSessionId: string }).activeSessionId = this.activeSessionId;
}
try {
const response = await this.client.request(command, timeoutMs, options);
if (!response.success) {
const error = deserializeDaemonError(response);
if (isSessionNotReadyError(error)) {
throw error;
}
this.definitiveRequestErrors.add(error);
throw error;
}
if (command.type === "get_model_catalog" || command.type === "get_available_models") {
// Catalog refresh can change model/auth settings, but does not change the transcript.
this.stateFreshnessGeneration++;
this.latestSnapshotStateIsFresh = false;
} else if (invalidatesCachedSnapshot(command.type)) {
this.latestSnapshotIsFresh = false;
if (command.type === "get_model_catalog" || command.type === "get_available_models") {
// Catalog refresh can change model/auth settings, but does not change the transcript.
this.stateFreshnessGeneration++;
this.latestSnapshotStateIsFresh = false;
} else if (invalidatesCachedSnapshot(command.type)) {
this.latestSnapshotIsFresh = false;
}
return response.data as T;
} catch (error) {
if (!this.disposed && !this.terminalCloseEmitted && isSessionNotReadyError(error)) {
try {
await this.resolveCurrentActiveSessionId(deadline);
continue;
} catch (resolveError) {
if (!isSessionNotReadyError(resolveError)) {
throw resolveError;
}
}
const delayMs = Math.min(deadline - Date.now(), 2000, 250 * 2 ** Math.min(attempt++, 3));
await delay(delayMs);
continue;
}
throw error;
}
}
return response.data as T;
if (this.disposed || this.terminalCloseEmitted) {
throw new Error("Daemon connection is closed.");
}
throw new Error(`Timed out waiting for daemon session: ${this.activeSessionId}`);
}
private async handleDaemonMessage(message: DaemonOutbound): Promise<void> {
if (this.disposed || this.terminalCloseEmitted) return;
730 unmodified lines
Expand all