getDaemonSocketCloseReason,
isDaemonNotConnectedError,
isDaemonSocketTransientError,
} from "../daemon/daemon-client.js" ;
import { deserializeDaemonError, isSessionStartingOrRecoveringError } from "../daemon/daemon-errors.js" ;
import {
DaemonSessionStartingError,
deserializeDaemonError,
isSessionNotReadyError,
isSessionStartingOrRecoveringError,
} from "../daemon/daemon-errors.js" ;
import {
collectDaemonClientEnv,
collectDaemonLaunchEnv,
type DaemonAttachResult,
type DaemonCommand,
DaemonControlPlaneTransportError,
DaemonDirectTransportClosedError,
DaemonRoutedClient,
} from "../daemon/daemon-routed-client.js" ;
import { getDaemonRuntimeIdentity } from "../daemon/daemon-runtime-identity.js" ;
import type { SessionSummary } from "../daemon/daemon-session-list.js" ;
import { listDaemonHeartbeats } from "../daemon/heartbeat-catalog.js" ;
import {
deleteDaemonSavedSession,
}
return / [ .!? ] $ / . test(message) ? message : ` ${ message } .` ;
}
function reconnectDaemonTransportAfterUpdate ( client: DaemonTransportClient ): Promise < void > {
function reconnectDaemonTransportAfterUpdate (
client: DaemonTransportClient ,
timeoutMs = UPDATE_RECONNECT_TIMEOUT_MS ,
): Promise < void > {
const existing = updateTransportReconnects . get(client);
if (existing) {
return existing;
}
const reconnectPromise = Promise . resolve()
. then( async () => {
client . disconnectForReconnect( "update" );
const deadline = Date . now() + UPDATE_RECONNECT_TIMEOUT_MS ;
const deadline = Date . now() + timeoutMs ;
let lastError: unknown ;
while (Date . now() < deadline) {
try {
await client . reconnect( 1000 );
private daemonLogPath: string | undefined ;
private updateRestartPending = false ;
private updateReconnectFailed = false ;
private terminalCloseEmitted = false ;
private notifiedBuildIdMismatch = false ;
private emitConnectedStatus (): void {
const daemonBuildId = this . client . hello ?. runtime ?. buildId;
const clientBuildId = getDaemonRuntimeIdentity () . buildId;
const buildIdMismatch =
daemonBuildId && clientBuildId && daemonBuildId !== clientBuildId && ! this . notifiedBuildIdMismatch
? { client: clientBuildId, daemon: daemonBuildId }
: undefined ;
if (buildIdMismatch) this . notifiedBuildIdMismatch = true ;
void this . emit({ type: "connection_status" , status: "connected" , buildIdMismatch });
}
private updateReconnectPromise ? : Promise <void> ;
private readonly activeSideQuestionIds = new Set < string >();
private readonly snapshotAssemblies = new Map < string , DaemonSnapshotAssembly >();
private readonly completedSnapshots = new Map < string , DaemonSessionSnapshot >();
options ?. ownedSession === true || options ?. directTransport === false ,
);
const connection = new DaemonAgentConnection (transport, activeSessionId, options);
connection . initialAttachPending = true ;
try {
const deadline = Date . now() + (options ?. reconnectTimeoutMs ?? DAEMON_RECONNECT_TIMEOUT_MS );
let attempt = 0 ;
let lastError: unknown ;
while (Date . now() < deadline && ! connection . disposed && ! connection . terminalCloseEmitted) {
try {
await connection . attach();
} catch (error) {
if ( ! (transport instanceof DaemonRoutedClient)) throw error;
transport . fallbackToSupervisor();
try {
await connection . attach();
} catch (_error) {
if (transport instanceof DaemonRoutedClient) {
transport . fallbackToSupervisor();
}
try {
// This retry owns its failure; a parked request would pend the attach forever.
await connection . attach({ recoverable: false });
} catch (retryError) {
// A control-plane close saved during the window is the authoritative cause.
throw connection . initialControlPlaneClose ?? retryError;
// This retry owns its failure; a parked request would pend the attach forever.
await connection . attach({ recoverable: false });
} catch (retryError) {
// A control-plane close saved during the window is the authoritative cause.
throw connection . initialControlPlaneClose ?? retryError;
}
}
}
connection . initialAttachPending = false ;
const initialControlPlaneClose = connection . initialControlPlaneClose ;
connection . initialControlPlaneClose = undefined ;
if (initialControlPlaneClose) {
// No listeners exist yet: a terminal close rejects the attach; the rest replays through the one handler.
if ( getDaemonSocketCloseReason ( initialControlPlaneClose ) === "shutdown" ) {
throw initialControlPlaneClose;
connection . initialAttachPending = false ;
const initialControlPlaneClose = connection . initialControlPlaneClose ;
connection . initialControlPlaneClose = undefined ;
if ( initialControlPlaneClose ) {
// No listeners exist yet: a terminal close rejects the attach; the rest replays through the one handler.
if ( getDaemonSocketCloseReason (initialControlPlaneClose) === "shutdown" ) {
throw initialControlPlaneClose ;
}
connection . handleTransportClose(initialControlPlaneClose);
}
connection . handleTransportClose(initialControlPlaneClose) ;
return connection;
} catch (error) {
lastError = connection . initialControlPlaneClose ?? error;
if (connection . disposed || connection . terminalCloseEmitted || ! isSessionNotReadyError (error)) {
break ;
}
const delayMs = Math . min(deadline - Date . now(), 2000 , 250 * 2 ** Math . min(attempt ++ , 3 ));
await delay (delayMs);
}
return connection;
} catch (error) {
connection . initialAttachPending = false ;
await connection . dispose();
throw error;
}
connection . initialAttachPending = false ;
await connection . dispose();
throw lastError;
}
async attach (options ? : { recoverable? : boolean }): Promise < void > {
if ( this . disposed || this . terminalCloseEmitted) throw new Error ( "Daemon session closed during attach" );
1476 unmodified lines
Expand all
} catch (attachError) {
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
if ( isSessionStartingOrRecoveringError (attachError)) {
throw attachError ;
}
// If attach failed because session was not found (e.g. restored under a new activeSessionId),
// query list to find restored session by file or sessionId.
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) = >
s . activeSessionId !== undefined &&
(( this . attachedSessionFile !== undefined && s . sessionFile === this . attachedSessionFile) ||
( this . attachedSessionId !== undefined && s . sessionId === this . attachedSessionId)),
);
if (restored ?. activeSessionId && restored . activeSessionId !== this . activeSessionId) {
this . sessionRevision ++ ;
this . activeSessionId = restored . activeSessionId ;
this . lastEventSequence = undefined ;
this . lastEventCursor = undefined ;
this . retiredEventGenerations . clear();
await this . attach({ recoverable: false });
const isUnknownSession =
attachError instanceof Error &&
attachError . message . toLowerCase() . includes( "unknown active session" );
if (isUnknownSession) {
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),
) ;
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 });
} else {
throw new DaemonSessionStartingError (
this . activeSessionId,
"Session worker is queued for restore" ,
);
}
} else {
throw attachError;
}
} else {
if ( ! this . disposed && ! this . terminalCloseEmitted) {
const snapshot = await this . getInitialSnapshot({ recoverable: false });
if ( this . disposed || this . terminalCloseEmitted) return ;
void this . emit({ type: "session_resynced" , snapshot });
void this . emit({ type: "connection_status" , status: "connected" } );
this . emitConnectedStatus( );
}
return ;
} catch (error) {
lastError = error instanceof Error ? error : new Error ( String (error));
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
const isTransientWorkerNotReady = isSessionStartingOrRecoveringError (error);
const isTransientWorkerNotReady = isSessionNotReadyError (error);
// A direct-half failure must not tear down a control-plane socket with a completed handshake.
const shouldResetControlPlane =
! isTransientWorkerNotReady &&
( ! ( this . client instanceof DaemonRoutedClient) ||
this . client . close();
this . terminalCloseEmitted = true ;
const closeReason = getDaemonSocketCloseReason (lastError);
const errorMsg =
closeReason === " shutdown "
? this . formatDaemonSessionClosedError( "shutdown" )
: closeReason === "update"
? this . formatDaemonSessionClosedError( "update" )
: `Daemon reconnection failed: ${ formatErrorSentence (lastError) } The session transcript remains saved; restart Prime Agent or reopen the session from Agents View. ${ this . formatDaemonDiagnosticContext() } ` ;
this . updateRestartPending || closeReason === " update "
? this . formatUpdateReconnectError(lastError )
: `Daemon reconnection failed: ${ formatErrorSentence (lastError) } The session transcript remains saved; restart Prime Agent or reopen the session from Agents View. ${ this . formatDaemonDiagnosticContext() } ` ;
await this . emit({ type: "closed" , error: errorMsg });
}
})() . finally(() => {
this . reconnectPromise = undefined ;
200 unmodified lines
Expand all
this . updateRestartPending = true ;
void this . reconnectAfterUpdate();
return ;
}
if (message . reason === "shutdown" && this . options . recoverDaemon) {
this . captureDaemonLogPath();
void this . reconnect( new Error ( "Daemon session closed for shutdown" ));
return ;
}
this . terminalCloseEmitted = true ;
const error = this . formatDaemonSessionClosedError(message . reason);
this . rejectSnapshotAssemblies( new Error (error));
await this . emit({ type: "closed" , error });
type : "connection_status" ,
status: "reconnecting" ,
error: "session is restarting - waiting for the daemon" ,
});
const reconnectPromise = reconnectDaemonTransportAfterUpdate ( this . client)
const reconnectPromise = reconnectDaemonTransportAfterUpdate (
this . client,
this . options . reconnectTimeoutMs ?? UPDATE_RECONNECT_TIMEOUT_MS ,
)
. then(() => this . restoreConnectionAfterUpdate())
. then(() => {
if ( ! this . disposed && ! this . terminalCloseEmitted) {
void this . emit({ type: "connection_status" , status: "connected" } );
this . emitConnectedStatus( );
}
})
. catch( async (error: unknown ) => {
this . updateRestartPending = false ;
const sessionFile = this . attachedSessionFile;
if ( ! sessionId && ! sessionFile) {
throw new Error ( "the previous session identity is unavailable" );
}
const deadline = Date . now() + UPDATE_RECONNECT_TIMEOUT_MS ;
const deadline = Date . now() + ( this . options . reconnectTimeoutMs ?? UPDATE_RECONNECT_TIMEOUT_MS ) ;
let lastError: unknown ;
let attempt = 0 ;
while ( ! this . disposed && ! this . terminalCloseEmitted && Date . now() < deadline) {
try {
await this . client . reconnect( 1000 );
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
// This loop owns the retry: a socket close must reject these instead of parking them behind a hello it can never produce.
if ( this . disposed || this . terminalCloseEmitted) return ;
const response = await this . client . request({ type: "list" }, 30000 , { recoverable: false });
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
if ( this . disposed || this . terminalCloseEmitted) return ;
if ( ! response . success) {
throw deserializeDaemonError (response);
}
const sessions = readSessionSummaries (response . data);
const restored = sessions . find(
(summary) =>
summary . activeSessionId !== undefined &&
( ( sessionFile !== undefined && summary . sessionFile === sessionFile) ||
(sessionId !== undefined && summary . sessionId === sessionId) ) ,
(sessionFile !== undefined && summary . sessionFile === sessionFile) ||
(sessionId !== undefined && summary . sessionId === sessionId),
);
if (restored ?. activeSessionId) {
if ( this . disposed || this . terminalCloseEmitted) {
return ;
if ( this . disposed || this . terminalCloseEmitted) return ;
if (restored . activeSessionId !== this . activeSessionId) {
this . sessionRevision ++ ;
this . activeSessionId = restored . activeSessionId;
this . lastEventSequence = undefined ;
this . lastEventCursor = undefined ;
this . retiredEventGenerations . clear();
}
this . sessionRevision ++ ;
this . activeSessionId = restored . activeSessionId;
this . lastEventSequence = undefined ;
this . lastEventCursor = undefined ;
this . retiredEventGenerations . clear();
await this . attach({ recoverable: false });
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
if ( this . disposed || this . terminalCloseEmitted) return ;
const snapshot = await this . getInitialSnapshot({ recoverable: false });
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
if ( this . disposed || this . terminalCloseEmitted) return ;
this . updateRestartPending = false ;
void this . emit({ type: "session_resynced" , snapshot });
return ;
}
lastError = new DaemonSessionStartingError ( this . activeSessionId, "Session worker is queued for restore" );
} catch (error) {
lastError = error;
}
if ( this . disposed || this . terminalCloseEmitted) return ;
const delayMs = Math . min(deadline - Date . now(), 2000 , 250 * 2 ** Math . min(attempt ++ , 3 ));
await delay (delayMs);
}
if ( this . disposed || this . terminalCloseEmitted) {
return ;
}
if ( this . disposed || this . terminalCloseEmitted) return ;
throw lastError ?? new Error ( "the restored session did not become available" );
}
private getSnapshotAssembly (snapshotId: string ): DaemonSnapshotAssembly {
407 unmodified lines
Expand all