Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
118 kB · 3150 lines
TypeScript
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151import { randomUUID } from "node:crypto";import { existsSync } from "node:fs";import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core";import type { ImageContent, ServiceTier, Transport } from "@earendil-works/pi-ai";import { appendRotatingLog, getAgentLogPath, getDaemonLogPath } from "../../config.js";import type { AgentSessionMessageReceipt, AgentSessionMessageSafetyStatus } from "../../core/agent-messages.js";import type { AgentSessionEvent } from "../../core/agent-session.js";import type { AgentSessionRuntimeConfig } from "../../core/agent-session-config.js";import type { AgentAutonomousStatus } from "../../core/autonomous.js";import type { BashResult } from "../../core/bash-executor.js";import type { CompactionResult } from "../../core/compaction/index.js";import type { ContextTreeNode } from "../../core/context-tree.js";import type { AgentCronJob, AgentHeartbeatDeliveryMode, AgentHeartbeatManagementAction, AgentHeartbeatUpdateAction,} from "../../core/cron-jobs.js";import type { AcpMcpServerConfig } from "../../core/mcp/acp-mcp-types.js";import type { CustomMessage } from "../../core/messages.js";import type { RefinementResult } from "../../core/refinement/index.js";import { MissingSessionCwdError } from "../../core/session-cwd.js";import type { DeleteSessionFileResult } from "../../core/session-file-actions.js";import { SessionAlreadyActiveError } from "../../core/session-lease.js";import type { SessionTreeFlatNode } from "../../core/session-manager.js";import type { SessionStats } from "../../core/session-stats.js";import { AgentsViewRosterStore, STALE_ROSTER_DAEMON_MESSAGE } from "../agents-view/roster-store.js";import { DaemonCapabilityUnavailableError, type DaemonTransportClient, getDaemonSocketCloseReason, isDaemonNotConnectedError, isDaemonSocketTransientError,} from "../daemon/daemon-client.js";import { DaemonSessionStartingError, DaemonSessionUnrestorableError, deserializeDaemonError, isSessionClosingError, isSessionNotReadyError, isSessionStartingOrRecoveringError,} from "../daemon/daemon-errors.js";import { collectDaemonClientEnv, collectDaemonLaunchEnv, type DaemonAttachResult, type DaemonCommand, type DaemonEventCursor, type DaemonOutbound, type DaemonReplayInfo, type DaemonSessionClosedReason, type DaemonSessionSnapshot, isDaemonDialogExtensionUiRequest, isUnknownDaemonCommandError,} from "../daemon/daemon-protocol.js";import { createDaemonSessionTransport, 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, listDaemonSavedSessions, renameDaemonSavedSession,} from "../daemon/saved-session-catalog.js";import { buildSessionTreeFromFlatNodes, projectSessionTreeFlatNodes } from "./session-tree-projection.js";import type { AgentConnection, AgentConnectionBeforeSessionInvalidateListener, AgentConnectionEvent, AgentConnectionEventListener, AgentConnectionExecuteBashOptions, AgentConnectionExtensionUiResponse, AgentConnectionForkOptions, AgentConnectionHeadlessCompletionOptions, AgentConnectionHeartbeat, AgentConnectionModel, AgentConnectionModelCatalog, AgentConnectionModelCycleResult, AgentConnectionNavigateTreeOptions, AgentConnectionNavigateTreeResult, AgentConnectionNewSessionOptions, AgentConnectionPromptOptions, AgentConnectionQueuedMessageLane, AgentConnectionQueuedMessageMutation, AgentConnectionQueuedMessageMutationStatus, AgentConnectionQueueMode, AgentConnectionQueueState, AgentConnectionResourceSnapshot, AgentConnectionRlmChildAgentSnapshot, AgentConnectionSavedSessionInfo, AgentConnectionSavedSessionScope, AgentConnectionScopedModel, AgentConnectionSessionContext, AgentConnectionSessionHeader, AgentConnectionSessionInputPause, AgentConnectionSessionListCallbacks, AgentConnectionSessionTreeFlatNode, AgentConnectionSessionTreeNode, AgentConnectionSessionWatcher, AgentConnectionSideQuestionEvent, AgentConnectionSideQuestionTurn, AgentConnectionSlashCommand, AgentConnectionSnapshot, AgentConnectionState, AgentConnectionSubmitPromptOptions, AgentConnectionSwitchSessionOptions, AgentConnectionToolDefinition, AgentConnectionUserMessage,} from "./types.js";import { AgentConnectionPromptAdmissionError } from "./types.js";
type DistributiveOmit<T, K extends keyof T> = T extends unknown ? Omit<T, K> : never;type DaemonCommandBody = DistributiveOmit<DaemonCommand, "id">;type DaemonSnapshotBegin = Extract<DaemonOutbound, { type: "session_snapshot_begin" }>;
interface DaemonSnapshotAssembly { begin?: DaemonSnapshotBegin; apply?: (snapshot: DaemonSessionSnapshot) => void; chunks: Map<number, AgentMessage[]>; promise: Promise<DaemonSessionSnapshot>; resolve: (snapshot: DaemonSessionSnapshot) => void; reject: (error: Error) => void; timeout: ReturnType<typeof setTimeout>;}
export const DAEMON_REFINE_REQUEST_TIMEOUT_MS = 10 * 60 * 1000;const DAEMON_LONG_RUNNING_REQUEST_TIMEOUT_MS = 24 * 60 * 60 * 1000;export const DAEMON_RECONNECT_TIMEOUT_MS = 60_000;export const DAEMON_SNAPSHOT_TIMEOUT_MS = 30_000;const MAX_IGNORED_SNAPSHOT_IDS = 128;// Overflow replaces the initial render with a fresh attach snapshot.const MAX_DEFERRED_SESSION_EVENTS = 1000;const MAX_PENDING_EXTENSION_UI_REQUESTS = 128;const UPDATE_RECONNECT_TIMEOUT_MS = 120000;const UPDATE_RECONNECT_RETRY_MS = 100;/** * Bound on the search for the attached session in the restarted daemon's roster. * A session the daemon did restore is listed while its worker is still starting (a * row without activeSessionId), so a roster read that misses the identity for this * long means the daemon has nothing to restore: stop waiting instead of holding * commands behind the reconnect banner. */export const DAEMON_SESSION_SEARCH_TIMEOUT_MS = 20_000;const MAX_COMPLETED_SNAPSHOTS = 128;const OWNED_SESSION_DISPOSE_RECONNECT_WAIT_MS = 10_000;const updateTransportReconnects = new WeakMap<DaemonTransportClient, Promise<void>>();
function delay(ms: number): Promise<void> { return new Promise((resolve) => setTimeout(resolve, ms));}
function formatErrorSentence(error: unknown): string { const message = (error instanceof Error ? error.message : String(error)).trim(); if (!message) { return "Unknown daemon error."; } return /[.!?]$/.test(message) ? message : `${message}.`;}
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() + timeoutMs; let lastError: unknown; while (Date.now() < deadline) { try { await client.reconnect(1000); return; } catch (error) { lastError = error; } await delay(UPDATE_RECONNECT_RETRY_MS); } throw lastError ?? new Error("the updated daemon did not become available"); }) .finally(() => { if (updateTransportReconnects.get(client) === reconnectPromise) { updateTransportReconnects.delete(client); } }); updateTransportReconnects.set(client, reconnectPromise); return reconnectPromise;}
export interface DaemonAgentConnectionOptions { closeClientOnDispose?: boolean; /** * Defer session events between attach and the first * flushBufferedSessionEvents() call. The interactive connection opts in so * its initial render can use the attach snapshot as-is (no full * get_messages re-fetch) and apply buffered events on top afterwards. * Watchers and one-shot clients leave this off and keep live delivery. */ deferSessionEvents?: boolean; /** Secondary watchers pass false to stay on the shared control-plane socket. */ directTransport?: boolean; /** Restart/probe the detached supervisor after a transient socket loss. */ recoverDaemon?: () => Promise<void>; /** Bound supervisor recovery before surfacing a fatal connection error. */ reconnectTimeoutMs?: number; /** Bound an incomplete streamed snapshot before failing the attach or resync. */ snapshotTimeoutMs?: number; /** Bound the search for the attached session in the restarted daemon's roster. */ sessionSearchTimeoutMs?: number; /** * Send this client's allowlisted env (herdr pane identity) with attach so * an env-less session (e.g. cron-created) adopts it. Set only by the * primary interactive connection — the daemon adopts-if-absent, never * rebinds, so watchers must not send env at all. */ sendClientEnv?: boolean; /** Advertise support for interactive extension dialogs. */ supportsExtensionUi?: boolean; /** Dispose the connection by stopping its hidden worker instead of detaching. */ ownedSession?: boolean; /** Fresh runtime context used only if the owned worker must be relaunched. */ ownedSessionRecoveryConfig?: AgentSessionRuntimeConfig; /** Require the target worker to have been created with telemetry disabled. */ telemetryDisabled?: true;}
/** * AgentConnection adapter for the local daemon JSONL socket transport. * * InteractiveMode depends only on AgentConnection; local socket ownership and * daemon command details stay inside this adapter. */export class DaemonAgentConnection implements AgentConnection { private readonly listeners = new Set<AgentConnectionEventListener>(); private readonly pendingExtensionUiRequests: Extract<DaemonOutbound, { type: "extension_ui_request" }>[] = []; private readonly unsubscribeDaemonMessages: () => void; private readonly unsubscribeDaemonClose: () => void; private readonly clientId = `daemon-agent-connection:${randomUUID()}`; private readonly sessionInputPauses = new Map<string, Promise<AgentConnectionSessionInputPause>>(); private sessionInputPauseGeneration = 0; private ownedSessionPromotionTail = Promise.resolve(); private lastEventCursor: DaemonEventCursor | undefined; private readonly retiredEventGenerations = new Set<string>(); private lastEventSequence: number | undefined; private childRosterSequence: number | undefined; private latestSnapshot: AgentConnectionSnapshot | undefined; private latestSnapshotIsFresh = false; private latestSnapshotStateIsFresh = false; // Bumped whenever a catalog refresh invalidates state freshness so snapshot // reads already in flight cannot mark stale state fresh again. private stateFreshnessGeneration = 0; private deferredSessionEvents: { event: AgentSessionEvent; sequence: number | undefined; generation: string | undefined; }[] = []; private deferSessionEvents = false; private deferredSessionEventsOverflowed = false; private deferredSessionEventFlush: Promise<void> | undefined; private sessionRevision = 0; private attachedSessionId: string | undefined; private attachedSessionFile: string | undefined; private daemonLogPath: string | undefined; private sessionSearchDeadline: number | undefined; private sessionUnrestorable: Error | 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>(); private readonly pendingReattachActiveSessionIds = new Set<string>(); private readonly snapshotRecoveryPromises = new Map<string, Promise<void>>(); private readonly ignoredSnapshotIds = new Set<string>(); private rosterStore: AgentsViewRosterStore | undefined; private reconnectPromise?: Promise<void>; private initialAttachPending = false; private initialControlPlaneClose?: Error; private readonly definitiveRequestErrors = new WeakSet<Error>(); private disposing = false; private disposed = false;
constructor( private readonly client: DaemonTransportClient, private activeSessionId: string, private readonly options: DaemonAgentConnectionOptions = {}, ) { if (options.recoverDaemon) { this.client.enableRequestRecovery(); } this.deferSessionEvents = options.deferSessionEvents === true; this.unsubscribeDaemonMessages = this.client.onMessage((message) => { void this.handleDaemonMessage(message).catch((error: unknown) => { try { appendRotatingLog( getAgentLogPath(), `[${new Date().toISOString()}] daemon-message: ignored ${message.type} failure: ${String(error)}`, ); } catch { // Logging failure must not turn an isolated message error into a connection failure. } }); }); this.captureDaemonLogPath(); this.unsubscribeDaemonClose = this.client.onClose((error) => this.handleTransportClose(error)); }
private handleTransportClose(error: Error): void { const directSessionSurvives = this.client instanceof DaemonRoutedClient && this.client.hasDirectTransport && !(error instanceof DaemonDirectTransportClosedError); const invalidatedInputPause = !directSessionSurvives && this.sessionInputPauses.size > 0; if (!directSessionSurvives) { this.sessionInputPauses.clear(); this.sessionInputPauseGeneration++; this.rejectSnapshotAssemblies(error); } if (this.initialAttachPending) { // attach() owns failure handling until the initial attach settles. if (directSessionSurvives) this.initialControlPlaneClose = error; return; } if (this.disposed || this.terminalCloseEmitted) { return; } // A lost direct link invalidates the fence (holders learn via the generation bump) yet the session falls back. if (invalidatedInputPause && !(error instanceof DaemonDirectTransportClosedError)) { this.terminalCloseEmitted = true; void this.emit({ type: "closed", error: "Daemon connection closed while session input was paused; the fence was invalidated.", }); return; } // An authoritative shutdown/update reason outranks the surviving direct link. const closeReason = getDaemonSocketCloseReason(error); if (closeReason === "shutdown") { if (this.options.recoverDaemon) { void this.reconnect(error); return; } this.terminalCloseEmitted = true; void this.emit({ type: "closed", error: this.formatDaemonSessionClosedError("shutdown") }); return; } if ((this.updateRestartPending || closeReason === "update") && !this.updateReconnectFailed) { this.updateRestartPending = true; void this.reconnectAfterUpdate(); return; } // A direct-transport loss is never itself a session loss: fall back through a supervisor re-attach. if (directSessionSurvives || error instanceof DaemonDirectTransportClosedError || this.options.recoverDaemon) { void this.reconnect(error); return; } this.terminalCloseEmitted = true; void this.emit({ type: "closed", error: this.formatDaemonConnectionClosedError(error) }); }
static async attach( client: DaemonTransportClient, activeSessionId: string, options?: DaemonAgentConnectionOptions, ): Promise<DaemonAgentConnection> { const transport = await createDaemonSessionTransport( client, activeSessionId, options?.ownedSession === true || options?.directTransport === false, ); const connection = new DaemonAgentConnection(transport, activeSessionId, options); connection.initialAttachPending = true; 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 { 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; } } 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); } 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); } } 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"); const sessionRevision = this.sessionRevision; const supportsExtensionUi = this.options.supportsExtensionUi !== false; const result = await this.requestData<SessionSummary | DaemonAttachResult>( { type: "attach", activeSessionId: this.activeSessionId, supportsExtensionUi, clientId: this.clientId, capabilities: [ "attach_snapshot", "event_sequence", ...(supportsExtensionUi ? (["extension_ui"] as const) : []), "slim_attach", "chunked_snapshot", ...(this.options.ownedSession ? (["client_owned_sessions"] as const) : []), ], env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined, launchEnv: this.options.ownedSession ? collectDaemonLaunchEnv() : undefined, ...(this.options.ownedSession && this.options.ownedSessionRecoveryConfig && this.client.supportsServerCapability("owned_session_recovery_context") ? { recoveryConfig: this.options.ownedSessionRecoveryConfig } : {}), telemetryDisabled: this.options.telemetryDisabled, resumeCursor: this.lastEventCursor === undefined ? undefined : { activeSessionId: this.activeSessionId, ...this.lastEventCursor, }, }, undefined, { ...options, onResponse: (response) => { if ( !response.success || this.disposed || this.terminalCloseEmitted || sessionRevision !== this.sessionRevision ) return; const result = response.data as SessionSummary | DaemonAttachResult; this.activeSessionId = getAttachActiveSessionId(result); if ("snapshot" in result && result.snapshotStream) { this.getSnapshotAssembly(result.snapshotStream.id).apply = (snapshot) => { if (sessionRevision === this.sessionRevision) this.applySessionSnapshot(snapshot, result.replay); }; } }, }, ); if (this.disposed || this.terminalCloseEmitted) throw new Error("Daemon session closed during attach"); if (sessionRevision !== this.sessionRevision) { if ("snapshot" in result && result.snapshotStream) { const snapshotId = result.snapshotStream.id; const assembly = this.snapshotAssemblies.get(snapshotId); if (assembly) { this.rejectSnapshotAssembly(snapshotId, assembly, new Error("Snapshot attach was superseded")); this.snapshotAssemblies.delete(snapshotId); } this.ignoreSnapshotId(snapshotId); } return; } this.activeSessionId = getAttachActiveSessionId(result); // A successful attach is the authoritative "the miss resolved": a deadline armed // by an earlier roster miss must not survive into the next restart, where it would // turn the first miss into an immediate unrestorable verdict. this.sessionSearchDeadline = undefined; if ("snapshot" in result && result.snapshotStream) { await this.waitForSnapshot(result.snapshotStream.id); } else { const attachCursor = getAttachLastEventCursor(result); if (attachCursor) this.observeEventCursor(attachCursor); this.lastEventSequence = maxEventSequence(this.lastEventSequence, getAttachLastEventSequence(result)); if ("snapshot" in result) { this.applySessionSnapshot(result.snapshot, result.replay); } else { this.latestSnapshot = undefined; this.latestSnapshotIsFresh = false; } } if (this.disposed || this.terminalCloseEmitted) throw new Error("Daemon session closed during attach"); if (sessionRevision !== this.sessionRevision) return; const summary = "snapshot" in result ? result.snapshot.summary : result; this.attachedSessionId = summary.sessionId; this.attachedSessionFile = summary.sessionFile ?? ("snapshot" in result ? result.snapshot.state.sessionFile : undefined); this.captureDaemonLogPath(); this.updateReconnectFailed = false; this.terminalCloseEmitted = false; // The roster bar is an accessory: its subscribe failure must never fail an // otherwise-recovered session. The bar degrades; the next reconnect or rebind // re-attaches through this same seam. if (this.rosterStore) await this.rosterStore.attach(this.client).catch(() => undefined); }
async subscribeAgentRoster( listener: () => void, ): Promise<{ summaries(): SessionSummary[]; dispose(): Promise<void> }> { this.rosterStore ??= new AgentsViewRosterStore(); const store = this.rosterStore; if (!(await store.attach(this.client))) { throw new Error(STALE_ROSTER_DAEMON_MESSAGE); } const unsubscribe = store.onUpdate(listener); return { summaries: () => store.summaries(), dispose: async () => unsubscribe(), }; }
subscribe(listener: AgentConnectionEventListener): () => void { this.listeners.add(listener); for (const request of this.pendingExtensionUiRequests.splice(0)) void this.handleDaemonMessage(request); return () => { this.listeners.delete(listener); }; }
onBeforeSessionInvalidate(_listener: AgentConnectionBeforeSessionInvalidateListener): () => void { return () => {}; }
async getState(): Promise<AgentConnectionState> { if (this.latestSnapshotIsFresh && this.latestSnapshotStateIsFresh && this.latestSnapshot) { return this.latestSnapshot.state; } return this.requestData<AgentConnectionState>({ type: "get_connection_state", activeSessionId: this.activeSessionId, }); }
async getInitialSnapshot(options?: { recoverable?: boolean }): Promise<AgentConnectionSnapshot> { if (this.latestSnapshotIsFresh && this.latestSnapshot) { if (this.latestSnapshotStateIsFresh) return this.latestSnapshot; const snapshot = this.latestSnapshot; const stateFreshnessGeneration = this.stateFreshnessGeneration; const state = await this.requestData<AgentConnectionState>( { type: "get_connection_state", activeSessionId: this.activeSessionId }, undefined, options, ); if (this.latestSnapshotIsFresh && this.latestSnapshot?.messages === snapshot.messages) { const recap = this.latestSnapshot.state.recap; this.latestSnapshot = { ...this.latestSnapshot, state: recap !== snapshot.state.recap ? { ...state, recap } : state, ...(snapshot.sessionContext ? { sessionContext: { ...snapshot.sessionContext, thinkingLevel: state.thinkingLevel, serviceTier: state.serviceTier, model: state.model ? { provider: state.model.provider, modelId: state.model.id } : null, }, } : {}), }; this.latestSnapshotStateIsFresh = stateFreshnessGeneration === this.stateFreshnessGeneration; return this.latestSnapshot; } } // The session tree is intentionally not fetched here: it is large on long // sessions and only needed when the user opens the tree/branch selector. // getSessionTree() fetches it lazily via get_session_tree on first use. const snapshotCursor = this.lastEventCursor; const snapshotSequence = this.lastEventSequence; const stateFreshnessGeneration = this.stateFreshnessGeneration; const [state, messagesData, sessionContextData] = await Promise.all([ this.requestData<AgentConnectionState>( { type: "get_connection_state", activeSessionId: this.activeSessionId }, undefined, options, ), this.requestData<{ messages: AgentMessage[] }>( { type: "get_messages", activeSessionId: this.activeSessionId }, undefined, options, ), this.requestData<{ context: AgentConnectionSessionContext }>( { type: "get_session_context", activeSessionId: this.activeSessionId }, undefined, options, ), ]); const children = this.latestSnapshot?.children; const streamingMessage = this.latestSnapshot?.streamingMessage; this.latestSnapshot = { state, messages: messagesData.messages, sessionContext: sessionContextData.context, ...(children ? { children } : {}), ...(streamingMessage ? { streamingMessage } : {}), }; if (snapshotSequence !== undefined) { this.latestSnapshot.lastEventSequence = snapshotSequence; } if (snapshotCursor) { this.latestSnapshot.lastEventCursor = snapshotCursor; } this.latestSnapshotIsFresh = snapshotSequence === this.lastEventSequence && snapshotCursor?.generation === this.lastEventCursor?.generation && snapshotCursor?.sequence === this.lastEventCursor?.sequence; this.latestSnapshotStateIsFresh = this.latestSnapshotIsFresh && stateFreshnessGeneration === this.stateFreshnessGeneration; return this.latestSnapshot; }
/** * Stop deferring session events and replay everything buffered so far. The * interactive UI calls this once its initial transcript render is complete, * so deferred events apply on top of a fully rendered chat. */ flushBufferedSessionEvents(): Promise<void> { if (this.deferredSessionEventFlush) return this.deferredSessionEventFlush; if (!this.deferSessionEvents) { return Promise.resolve(); } this.deferredSessionEventFlush = Promise.resolve() .then(async () => { const deliveries: Promise<void>[] = []; while (!this.disposed && !this.terminalCloseEmitted) { if (this.deferredSessionEventsOverflowed) { this.deferredSessionEventsOverflowed = false; // Reattach provides one consistent snapshot, including the streaming message. const sessionRevision = this.sessionRevision; try { await this.attach(); } catch (error) { if (sessionRevision === this.sessionRevision) throw error; } if (this.disposed || this.terminalCloseEmitted) return; if (sessionRevision !== this.sessionRevision || this.deferredSessionEventsOverflowed) continue; const snapshot = await this.getInitialSnapshot(); if (sessionRevision === this.sessionRevision) deliveries.push(this.emit({ type: "session_resynced", snapshot })); continue; } const deferred = this.deferredSessionEvents.shift(); if (!deferred) { this.stopDeferringSessionEvents(); break; } const { event, sequence } = deferred; this.observeStreamingMessage(event); if (event.type === "rlm_child_update") { this.childRosterSequence = maxEventSequence(this.childRosterSequence, sequence); this.observeRlmChildUpdate(event.child); } this.latestSnapshotIsFresh = false; // Enqueue the whole replay before yielding; the UI serializes its rendering. deliveries.push(this.emit({ type: "session_event", event })); } await Promise.all(deliveries); }) .catch(async (error: unknown) => { if (this.disposed || this.terminalCloseEmitted) return; this.terminalCloseEmitted = true; await this.emit({ type: "closed", error: `Failed to recover deferred session events: ${String(error)}` }); }) .finally(() => { this.stopDeferringSessionEvents(); this.deferredSessionEventFlush = undefined; }); return this.deferredSessionEventFlush; }
private stopDeferringSessionEvents(): void { this.deferSessionEvents = false; this.deferredSessionEvents.length = 0; this.deferredSessionEventsOverflowed = false; }
/** * Keep only events newer than the snapshot in the same worker generation. */ private dropDeferredSessionEventsThrough( snapshotSequence: number | undefined, snapshotCursor: DaemonEventCursor | undefined, ): void { if ( snapshotCursor?.generation !== this.lastEventCursor?.generation || (snapshotSequence !== undefined && this.lastEventSequence !== undefined && snapshotSequence >= this.lastEventSequence) ) { this.deferredSessionEventsOverflowed = false; } this.deferredSessionEvents = this.deferredSessionEvents.filter( (entry) => entry.generation === snapshotCursor?.generation && (entry.sequence !== undefined && snapshotSequence !== undefined ? entry.sequence > snapshotSequence : entry.sequence === undefined && snapshotSequence === undefined), ); }
async getRlmChildSnapshots(): Promise<AgentConnectionRlmChildAgentSnapshot[]> { if (!this.client.supportsServerCapability("authoritative_child_roster")) { throw new DaemonCapabilityUnavailableError("get_rlm_children", "authoritative_child_roster"); } const data = await this.requestData<{ children: AgentConnectionRlmChildAgentSnapshot[]; eventSequence: number; }>({ type: "get_rlm_children", activeSessionId: this.activeSessionId }); if (!Array.isArray(data.children) || !Number.isInteger(data.eventSequence)) { throw new Error("Daemon returned an invalid child roster"); } if ((this.childRosterSequence ?? -1) > data.eventSequence) { return this.latestSnapshot?.children ?? data.children; } this.childRosterSequence = data.eventSequence; if (this.latestSnapshot) { this.latestSnapshot = { ...this.latestSnapshot, children: data.children }; } return data.children; }
async getMessages(): Promise<AgentMessage[]> { if (this.latestSnapshotIsFresh && this.latestSnapshot) { return this.latestSnapshot.messages; } const data = await this.requestData<{ messages: AgentMessage[] }>({ type: "get_messages", activeSessionId: this.activeSessionId, }); return data.messages; }
async getSessionHeader(): Promise<AgentConnectionSessionHeader | undefined> { const data = await this.requestData<{ header?: AgentConnectionSessionHeader | null }>({ type: "get_session_header", activeSessionId: this.activeSessionId, }); return data.header ?? undefined; }
async getCommands(): Promise<AgentConnectionSlashCommand[]> { const data = await this.requestData<{ commands: AgentConnectionSlashCommand[] }>({ type: "get_commands", activeSessionId: this.activeSessionId, }); if (this.client.supportsServerCapability("command_argument_completions")) return data.commands; return data.commands.map((command) => ({ ...command, argumentCompletions: undefined })); }
async getResourceSnapshot(): Promise<AgentConnectionResourceSnapshot> { return this.requestData<AgentConnectionResourceSnapshot>({ type: "get_resource_snapshot", activeSessionId: this.activeSessionId, }); }
supportsAcpMcpServers(): boolean { return this.client.supportsServerCapability("acp_mcp_servers"); }
async replaceAcpMcpServers(servers: readonly AcpMcpServerConfig[], ownerId: string): Promise<void> { if (!this.supportsAcpMcpServers()) { throw new DaemonCapabilityUnavailableError("replace_acp_mcp_servers", "acp_mcp_servers"); } await this.requestOk({ type: "replace_acp_mcp_servers", activeSessionId: this.activeSessionId, ownerId, servers: [...servers], }); }
async releaseAcpMcpServers(ownerId: string, _serverNames: readonly string[]): Promise<void> { await this.replaceAcpMcpServers([], ownerId); }
async getAvailableModels(): Promise<AgentConnectionModel[]> { const data = await this.requestData<{ models: AgentConnectionModel[] }>({ type: "get_available_models", activeSessionId: this.activeSessionId, }); return data.models; }
async getModelCatalog(): Promise<AgentConnectionModelCatalog> { if (!this.client.supportsServerCapability("model_catalog")) { const models = await this.getAvailableModels(); return { models, configuredProviders: [...new Set(models.map((model) => model.provider))], }; } return this.requestData<AgentConnectionModelCatalog>({ type: "get_model_catalog", activeSessionId: this.activeSessionId, }); }
async getSessionStats(): Promise<SessionStats> { return this.requestData<SessionStats>({ type: "get_session_stats", activeSessionId: this.activeSessionId, }); }
async getContextTree(): Promise<ContextTreeNode> { return this.requestData<ContextTreeNode>({ type: "get_context_tree", activeSessionId: this.activeSessionId, }); }
async getSessionContext(): Promise<AgentConnectionSessionContext> { if (this.latestSnapshotIsFresh && this.latestSnapshotStateIsFresh && this.latestSnapshot?.sessionContext) { return this.latestSnapshot.sessionContext; } const data = await this.requestData<{ context: AgentConnectionSessionContext }>({ type: "get_session_context", activeSessionId: this.activeSessionId, }); return data.context; }
async getSessionTree(): Promise<{ tree: AgentConnectionSessionTreeNode[]; leafId: string | null }> { if (this.latestSnapshotIsFresh && this.latestSnapshot?.sessionTree) { return this.latestSnapshot.sessionTree; } // Ask for the projected tree only when the daemon advertises it. An older // daemon answers with the full entries, which the client projects locally. const daemonProjects = this.client.supportsServerCapability("session_tree_projection"); const data = await this.requestData<{ entries?: AgentConnectionSessionTreeFlatNode[]; flatNodes?: SessionTreeFlatNode[]; leafId: string | null; }>({ type: "get_session_tree", activeSessionId: this.activeSessionId, ...(daemonProjects ? { projection: "tree" as const } : {}), }); const flatNodes = data.entries ?? projectSessionTreeFlatNodes(data.flatNodes ?? []); return { tree: buildSessionTreeFromFlatNodes(flatNodes), leafId: data.leafId }; }
async listSavedSessions( scope: AgentConnectionSavedSessionScope, callbacks?: AgentConnectionSessionListCallbacks, ): Promise<AgentConnectionSavedSessionInfo[]> { return listDaemonSavedSessions(this.client, { activeSessionId: this.activeSessionId }, scope, callbacks); }
async getQueue(): Promise<AgentConnectionQueueState> { return this.requestData<AgentConnectionQueueState>({ type: "get_queue", activeSessionId: this.activeSessionId, }); }
async mutateQueuedMessage( lane: AgentConnectionQueuedMessageLane, index: number, expectedText: string, mutation: AgentConnectionQueuedMessageMutation, ): Promise<AgentConnectionQueuedMessageMutationStatus> { if (!this.client.supportsServerCapability("queue_message_mutation")) return "unsupported"; const data = await this.requestData<{ status: AgentConnectionQueuedMessageMutationStatus }>({ type: "mutate_queued_message", activeSessionId: this.activeSessionId, lane, index, expectedText, mutation, }); return data.status; }
async clearQueue(): Promise<AgentConnectionQueueState> { return this.requestData<AgentConnectionQueueState>({ type: "clear_queue", activeSessionId: this.activeSessionId, }); }
async abortAndClearQueue(): Promise<AgentConnectionQueueState> { try { return await this.requestData<AgentConnectionQueueState>({ type: "abort_and_clear_queue", activeSessionId: this.activeSessionId, }); } catch (error) { if (isUnknownDaemonCommandError(error, "abort_and_clear_queue")) { throw new Error("the daemon is running an older build; restart the daemon and try again"); } throw error; } }
async acquireSessionInputPause(leaseKey: string): Promise<AgentConnectionSessionInputPause> { if (this.terminalCloseEmitted) throw new Error("Daemon connection is closed; cannot acquire an input pause."); const activeSessionId = this.activeSessionId; const generation = this.sessionInputPauseGeneration; const acquisitionKey = JSON.stringify([activeSessionId, leaseKey]); const existing = this.sessionInputPauses.get(acquisitionKey); if (existing) return existing; const acquisition = (async (): Promise<AgentConnectionSessionInputPause> => { const { pauseId } = await this.requestData<{ pauseId: string }>({ type: "acquire_session_input_pause", activeSessionId, leaseKey, }); if (generation !== this.sessionInputPauseGeneration || this.terminalCloseEmitted) { try { await this.requestData({ type: "release_session_input_pause", activeSessionId, pauseId, }); } catch { this.client.close(); } throw new Error("Session input pause acquisition was invalidated by a daemon reconnect."); } let released = false; return { release: async () => { if (released) return; if (generation !== this.sessionInputPauseGeneration) { throw new Error("Session input pause was invalidated by a daemon reconnect."); } await this.requestData({ type: "release_session_input_pause", activeSessionId, pauseId, }); released = true; if (this.sessionInputPauses.get(acquisitionKey) === acquisition) { this.sessionInputPauses.delete(acquisitionKey); } }, }; })(); this.sessionInputPauses.set(acquisitionKey, acquisition); try { return await acquisition; } catch (error) { if (this.sessionInputPauses.get(acquisitionKey) === acquisition) this.sessionInputPauses.delete(acquisitionKey); throw error; } }
async listCronJobs(options: { includeInactive?: boolean } = {}): Promise<AgentCronJob[]> { const data = await this.requestData<{ jobs: AgentCronJob[] }>({ type: "cron_list", activeSessionId: this.activeSessionId, includeInactive: options.includeInactive, }); return data.jobs; }
async listHeartbeats(): Promise<AgentConnectionHeartbeat[]> { return listDaemonHeartbeats(this.client, this.options.ownedSession ? this.activeSessionId : undefined); }
async manageHeartbeat( activeSessionId: string, jobId: string, action: AgentHeartbeatManagementAction, ): Promise<AgentCronJob> { if (!this.client.supportsServerCapability("heartbeat_management")) { throw new Error("Heartbeat management requires a newer Prime Agent daemon."); } try { const data = await this.requestData<{ heartbeat: AgentCronJob }>({ type: "heartbeat_manage", activeSessionId, jobId, action, }); return data.heartbeat; } catch (error) { if (isUnknownDaemonCommandError(error, "heartbeat_manage")) { throw new Error("Heartbeat management requires a newer Prime Agent daemon."); } throw error; } }
async addCronJob(schedule: string, prompt: string): Promise<AgentCronJob> { return this.withOwnedSessionPromotion(async (promoteOwnedSession) => { const data = await this.requestData<{ job: AgentCronJob }>({ type: "cron_add", activeSessionId: this.activeSessionId, schedule, prompt, promoteOwnedSession, }); return data.job; }); }
async cancelCronJob(jobId: string): Promise<AgentCronJob> { const data = await this.requestData<{ job: AgentCronJob }>({ type: "cron_cancel", activeSessionId: this.activeSessionId, jobId, }); return data.job; }
async getHeartbeat(): Promise<AgentCronJob | undefined> { const data = await this.requestData<{ heartbeat?: AgentCronJob | null }>({ type: "heartbeat_get", activeSessionId: this.activeSessionId, }); return data.heartbeat ?? undefined; }
async setHeartbeat( schedule: string, instruction: string, deliveryMode?: AgentHeartbeatDeliveryMode, ): Promise<AgentCronJob> { return this.withOwnedSessionPromotion(async (promoteOwnedSession) => { const data = await this.requestData<{ heartbeat: AgentCronJob }>({ type: "heartbeat_set", activeSessionId: this.activeSessionId, schedule, prompt: instruction, ...(deliveryMode ? { deliveryMode } : {}), promoteOwnedSession, }); return data.heartbeat; }); }
async updateHeartbeat(action: AgentHeartbeatUpdateAction): Promise<AgentCronJob | undefined> { const data = await this.requestData<{ heartbeat?: AgentCronJob | null }>({ type: "heartbeat_update", activeSessionId: this.activeSessionId, action, }); return data.heartbeat ?? undefined; }
async sendAgentMessage(targetActiveSessionId: string, message: string): Promise<AgentSessionMessageReceipt> { return this.requestData<AgentSessionMessageReceipt>({ type: "send_message", targetActiveSessionId, message, fromActiveSessionId: this.activeSessionId, }); }
async getAgentMessageStatus(): Promise<AgentSessionMessageSafetyStatus> { return this.requestData<AgentSessionMessageSafetyStatus>({ type: "agent_messages_status", activeSessionId: this.activeSessionId, }); }
async pauseAgentMessages(): Promise<AgentSessionMessageSafetyStatus> { return this.requestData<AgentSessionMessageSafetyStatus>({ type: "agent_messages_pause", activeSessionId: this.activeSessionId, }); }
async resumeAgentMessages(): Promise<AgentSessionMessageSafetyStatus> { return this.requestData<AgentSessionMessageSafetyStatus>({ type: "agent_messages_resume", activeSessionId: this.activeSessionId, }); }
async clearAgentMessages(): Promise<number> { return this.requestData<number>({ type: "agent_messages_clear", activeSessionId: this.activeSessionId, }); }
async getUserMessagesForForking(): Promise<AgentConnectionUserMessage[]> { const data = await this.requestData<{ messages: AgentConnectionUserMessage[] }>({ type: "get_user_messages_for_forking", activeSessionId: this.activeSessionId, }); return data.messages; }
async getLastAssistantText(): Promise<string | undefined> { const data = await this.requestData<{ text?: string | null }>({ type: "get_last_assistant_text", activeSessionId: this.activeSessionId, }); return data.text ?? undefined; }
async getSystemPrompt(): Promise<string> { const data = await this.requestData<{ systemPrompt: string }>({ type: "get_system_prompt", activeSessionId: this.activeSessionId, }); return data.systemPrompt; }
async getToolDefinition(name: string): Promise<AgentConnectionToolDefinition | undefined> { const data = await this.requestData<{ toolDefinition?: AgentConnectionToolDefinition }>({ type: "get_tool_definition", activeSessionId: this.activeSessionId, name, }); return data.toolDefinition; }
async setSessionEntryLabel(entryId: string, label: string | undefined): Promise<void> { await this.requestOk({ type: "set_session_entry_label", activeSessionId: this.activeSessionId, entryId, label, }); }
async respondToExtensionUiRequest(requestId: string, response: AgentConnectionExtensionUiResponse): Promise<void> { await this.requestOk({ type: "extension_ui_response", activeSessionId: this.activeSessionId, requestId, response, }); }
async prompt(message: string, options?: AgentConnectionSubmitPromptOptions): Promise<void> { await this.promptWithAdmissionCancellation("prompt", message, options); }
async promptAndWait(message: string, options?: AgentConnectionPromptOptions): Promise<void> { await this.promptWithAdmissionCancellation("prompt_and_wait", message, options); }
private async waitForReconnectionIfPending(deadline: number): Promise<void> { while (Date.now() < deadline && !this.disposed && !this.terminalCloseEmitted) { const pending = this.reconnectPromise ?? this.updateReconnectPromise; if (!pending) break; await pending.catch(() => undefined); } }
private async promptWithAdmissionCancellation( type: "prompt" | "prompt_and_wait", message: string, options?: AgentConnectionSubmitPromptOptions, ): Promise<void> { const signal = options?.signal; const customPromptOptions = type === "prompt" ? options : undefined; if (signal?.aborted) { throw new AgentConnectionPromptAdmissionError("Prompt admission was cancelled.", "cancelled"); } this.throwIfSessionUnrestorable(); if (!signal) { 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); this.throwIfSessionUnrestorable(); if (this.disposed || this.terminalCloseEmitted) { throw new Error("Daemon connection closed; cannot submit prompt."); } try { await this.requestData<unknown>( { type, activeSessionId: this.activeSessionId, message, images: options?.images, streamingBehavior: options?.streamingBehavior, queueIfBusy: options?.queueIfBusy, source: options?.source, ...(customPromptOptions ? { customMessage: customPromptOptions.customMessage, expandPromptTemplates: customPromptOptions.expandPromptTemplates, } : {}), }, DAEMON_LONG_RUNNING_REQUEST_TIMEOUT_MS, ); return; } catch (error) { if ( !this.disposed && !this.terminalCloseEmitted && (isDaemonNotConnectedError(error) || isSessionStartingOrRecoveringError(error) || isDaemonSocketTransientError(error)) ) { if (!this.reconnectPromise && !this.updateReconnectPromise && this.options.recoverDaemon) { void this.reconnect(error instanceof Error ? error : new Error(String(error))); } const delayMs = Math.min(deadline - Date.now(), 2000, 250 * 2 ** Math.min(attempt++, 3)); await delay(delayMs); continue; } throw error; } } this.throwIfSessionUnrestorable(); if (this.disposed || this.terminalCloseEmitted) { throw new Error("Daemon connection closed; cannot submit prompt."); } throw new Error("Timed out waiting for daemon to become ready to accept prompt."); } const admissionId = `prompt-admission:${randomUUID()}`; let resolveAbort = () => {}; const aborted = new Promise<"abort">((resolve) => { resolveAbort = () => resolve("abort"); }); const onAbort = () => resolveAbort(); signal.addEventListener("abort", onAbort, { once: true }); // Close the listener-registration race before issuing the first request. if (signal.aborted) { signal.removeEventListener("abort", onAbort); throw new AgentConnectionPromptAdmissionError("Prompt admission was cancelled.", "cancelled"); } const command = { type, activeSessionId: this.activeSessionId, message, images: options.images, streamingBehavior: options.streamingBehavior, queueIfBusy: options.queueIfBusy, source: options.source, ...(customPromptOptions ? { customMessage: customPromptOptions.customMessage, expandPromptTemplates: customPromptOptions.expandPromptTemplates, } : {}), admissionId, } as Extract<DaemonCommandBody, { type: typeof type }>; let promptError: unknown; const promptRequest = this.requestData<unknown>(command, DAEMON_LONG_RUNNING_REQUEST_TIMEOUT_MS).catch( (error: unknown) => { promptError = error instanceof DaemonCapabilityUnavailableError && !error.afterReconnect ? new AgentConnectionPromptAdmissionError(error.message, "unsupported", { cause: error }) : error; return "failed" as const; }, ); try { const first = await Promise.race([promptRequest.then(() => "settled" as const), aborted]); if (first === "settled" && promptError === undefined) return; if (first === "settled" && promptError instanceof AgentConnectionPromptAdmissionError) throw promptError; if ( first === "settled" && !signal.aborted && promptError instanceof Error && this.definitiveRequestErrors.has(promptError) ) { throw promptError; } let status: "cancelled" | "owned" | "unknown" = "unknown"; try { const result = await this.requestData<{ status: "cancelled" | "owned" | "unknown" }>({ type: "cancel_prompt_admission", activeSessionId: this.activeSessionId, admissionId, ...(this.client.supportsServerCapability("owned_prompt_cancellation") ? { cancelOwned: true } : {}), }); status = result.status; } catch { // Timeout/transport is indistinguishable from accepted ownership. } await promptRequest; if (promptError instanceof AgentConnectionPromptAdmissionError) throw promptError; const definitiveFailure = promptError instanceof Error && this.definitiveRequestErrors.has(promptError); if (promptError === undefined || (status === "owned" && type === "prompt" && !definitiveFailure)) return; throw new AgentConnectionPromptAdmissionError( promptError instanceof Error ? promptError.message : "Prompt admission did not complete.", status, promptError === undefined ? undefined : { cause: promptError }, ); } finally { signal.removeEventListener("abort", onAbort); } }
async startSideQuestion( id: string, question: string, previousTurns?: AgentConnectionSideQuestionTurn[], ): Promise<void> { if (previousTurns?.length && !this.client.supportsServerCapability("side_question_transcript")) { // An older daemon would silently ignore previousTurns and answer the // follow-up without the side-conversation context; fail loudly instead. throw new Error( "the daemon is running an older build without side-conversation follow-ups; restart the daemon and try again", ); } this.activeSideQuestionIds.add(id); try { await this.requestOk({ type: "start_side_question", activeSessionId: this.activeSessionId, sideQuestionId: id, question, previousTurns, }); } catch (error) { this.activeSideQuestionIds.delete(id); if (isUnknownDaemonCommandError(error, "start_side_question")) { throw new Error("the daemon is running an older build; restart the daemon and try again"); } throw error; } }
async abortSideQuestion(id: string): Promise<boolean> { const data = await this.requestData<{ aborted: boolean }>({ type: "abort_side_question", activeSessionId: this.activeSessionId, sideQuestionId: id, }); this.activeSideQuestionIds.delete(id); return data.aborted; }
async steer(message: string, images?: ImageContent[]): Promise<void> { await this.requestOk({ type: "steer", activeSessionId: this.activeSessionId, message, images }); }
async followUp(message: string, images?: ImageContent[]): Promise<void> { await this.requestOk({ type: "follow_up", activeSessionId: this.activeSessionId, message, images }); }
async abort(): Promise<void> { await this.requestOk({ type: "abort", activeSessionId: this.activeSessionId }); }
async cancelRlmChild(childId: string): Promise<boolean> { try { const result = await this.requestData<{ cancelled: boolean }>({ type: "cancel_rlm_child", activeSessionId: this.activeSessionId, childId, }); return result.cancelled; } catch (error) { if (isUnknownDaemonCommandError(error, "cancel_rlm_child")) { throw new Error("the daemon is running an older build; restart the daemon and try again"); } throw error; } }
async waitForIdle(): Promise<void> { await this.requestData<unknown>( { type: "wait_for_idle", activeSessionId: this.activeSessionId }, DAEMON_LONG_RUNNING_REQUEST_TIMEOUT_MS, ); }
async waitForHeadlessCompletion(options?: AgentConnectionHeadlessCompletionOptions): Promise<AgentAutonomousStatus> { if (options?.waitForRlmQuiescence && !this.client.supportsServerCapability("rlm_quiescence_barrier")) { throw new Error( "the daemon is running an older build without RLM quiescence barriers; restart the daemon and try again", ); } return this.requestData<AgentAutonomousStatus>( { type: "wait_for_headless_completion", activeSessionId: this.activeSessionId, ...(options?.waitForRlmQuiescence ? { waitForRlmQuiescence: true } : {}), }, DAEMON_LONG_RUNNING_REQUEST_TIMEOUT_MS, ); }
async appendCustomMessage( message: Pick<CustomMessage, "customType" | "content" | "display" | "details">, ): Promise<void> { await this.requestOk({ type: "append_custom_message", activeSessionId: this.activeSessionId, message, }); }
async executeBash(command: string, options?: AgentConnectionExecuteBashOptions): Promise<void> { if (options?.transient && !this.client.supportsServerCapability("transient_bash")) { // An older daemon would record the run into the session, leaking the // side conversation into the main transcript; fail loudly instead. throw new Error( "the daemon is running an older build without side-conversation bash; restart the daemon and try again", ); } try { await this.requestOk({ type: "execute_bash", activeSessionId: this.activeSessionId, command, excludeFromContext: options?.excludeFromContext, transient: options?.transient, runId: options?.runId, }); } catch (error) { if (isUnknownDaemonCommandError(error, "execute_bash")) { throw new Error("the daemon is running an older build; restart the daemon and try again"); } throw error; } }
async executeBashAndWait(command: string): Promise<BashResult> { return this.requestData<BashResult>( { type: "execute_bash_and_wait", activeSessionId: this.activeSessionId, command, }, DAEMON_LONG_RUNNING_REQUEST_TIMEOUT_MS, ); }
async abortBash(): Promise<void> { try { await this.requestOk({ type: "abort_bash", activeSessionId: this.activeSessionId }); } catch (error) { if (isUnknownDaemonCommandError(error, "abort_bash")) { throw new Error("the daemon is running an older build; restart the daemon and try again"); } throw error; } }
async setModel(provider: string, modelId: string): Promise<AgentConnectionModel> { return this.requestData<AgentConnectionModel>({ type: "set_model", activeSessionId: this.activeSessionId, provider, modelId, }); }
async cycleModel(direction?: "forward" | "backward"): Promise<AgentConnectionModelCycleResult | undefined> { const result = await this.requestData<AgentConnectionModelCycleResult | null>({ type: "cycle_model", activeSessionId: this.activeSessionId, direction, }); return result ?? undefined; }
async setScopedModels(scopedModels: AgentConnectionScopedModel[]): Promise<void> { await this.requestOk({ type: "set_scoped_models", activeSessionId: this.activeSessionId, scopedModels, }); }
async setThinkingLevel(level: ThinkingLevel): Promise<void> { await this.requestOk({ type: "set_thinking_level", activeSessionId: this.activeSessionId, level }); }
async setServiceTier(serviceTier: ServiceTier): Promise<void> { await this.requestOk({ type: "set_service_tier", activeSessionId: this.activeSessionId, serviceTier }); }
async cycleThinkingLevel(): Promise<ThinkingLevel | undefined> { const result = await this.requestData<{ level: ThinkingLevel } | null>({ type: "cycle_thinking_level", activeSessionId: this.activeSessionId, }); return result?.level; }
async setTransport(transport: Transport): Promise<void> { await this.requestOk({ type: "set_transport", activeSessionId: this.activeSessionId, transport }); }
async setSteeringMode(mode: AgentConnectionQueueMode): Promise<void> { await this.requestOk({ type: "set_steering_mode", activeSessionId: this.activeSessionId, mode }); }
async setFollowUpMode(mode: AgentConnectionQueueMode): Promise<void> { await this.requestOk({ type: "set_follow_up_mode", activeSessionId: this.activeSessionId, mode }); }
async setAutoCompactionEnabled(enabled: boolean): Promise<void> { await this.requestOk({ type: "set_auto_compaction", activeSessionId: this.activeSessionId, enabled }); }
async setAutoRetryEnabled(enabled: boolean): Promise<void> { await this.requestOk({ type: "set_auto_retry", activeSessionId: this.activeSessionId, enabled }); }
async compact(customInstructions?: string): Promise<CompactionResult> { return this.requestData<CompactionResult>({ type: "compact", activeSessionId: this.activeSessionId, customInstructions, }); }
async refine( options: { instructions?: string; rollbackId?: string; global?: boolean } = {}, ): Promise<RefinementResult> { const command: { type: "refine"; activeSessionId: string; instructions?: string; rollbackId?: string; global?: boolean; } = { type: "refine", activeSessionId: this.activeSessionId, instructions: options.instructions, rollbackId: options.rollbackId, }; if (options.global !== undefined) { command.global = options.global; } return this.requestData<RefinementResult>(command, DAEMON_REFINE_REQUEST_TIMEOUT_MS); }
async abortCompaction(): Promise<void> { await this.requestOk({ type: "abort_compaction", activeSessionId: this.activeSessionId }); }
async abortBranchSummary(): Promise<void> { await this.requestOk({ type: "abort_branch_summary", activeSessionId: this.activeSessionId }); }
async abortRetry(): Promise<void> { await this.requestOk({ type: "abort_retry", activeSessionId: this.activeSessionId }); }
async reload(): Promise<void> { await this.requestOk({ type: "reload", activeSessionId: this.activeSessionId }); }
async newSession(options?: AgentConnectionNewSessionOptions): Promise<{ cancelled: boolean }> { if (this.sessionUnrestorable || this.activeReplacementPromise) { if (options?.parentSession !== undefined) { // The daemon's create command cannot carry a parent edge, so retrying cannot add one. throw new DaemonSessionUnrestorableError( `The session this window was attached to is gone, and a parent-scoped session cannot be created through the daemon's create path. Relaunch this window. ${this.formatDaemonDiagnosticContext()}`, ); } return this.replaceLostSession({}); } return this.requestData<{ cancelled: boolean }>({ type: "new_session", activeSessionId: this.activeSessionId, parentSession: options?.parentSession, }); }
async switchSession( sessionPath: string, options?: AgentConnectionSwitchSessionOptions, ): Promise<{ cancelled: boolean }> { if (this.sessionUnrestorable || this.activeReplacementPromise) { return this.replaceLostSession({ sessionPath, cwdOverride: options?.cwdOverride }); } const sourceActiveSessionId = this.activeSessionId; try { return await this.requestData<{ cancelled: boolean }>({ type: "switch_session", activeSessionId: sourceActiveSessionId, sessionPath, cwdOverride: options?.cwdOverride, }); } catch (error) { if (!(error instanceof SessionAlreadyActiveError) || !error.activeSessionId) { throw error; } if (this.options.ownedSession) { throw error; } if (error.activeSessionId === sourceActiveSessionId) { return { cancelled: false }; } return this.reattachSession(sourceActiveSessionId, error.activeSessionId); } }
private async reattachSession( sourceActiveSessionId: string, targetActiveSessionId: string, ): Promise<{ cancelled: false }> { const previousState = { lastEventCursor: this.lastEventCursor, lastEventSequence: this.lastEventSequence, latestSnapshot: this.latestSnapshot, latestSnapshotIsFresh: this.latestSnapshotIsFresh, retiredEventGenerations: new Set(this.retiredEventGenerations), deferredSessionEvents: this.deferredSessionEvents, deferredSessionEventsOverflowed: this.deferredSessionEventsOverflowed, }; this.sessionRevision++; this.deferredSessionEvents = []; this.deferredSessionEventsOverflowed = false; this.activeSessionId = targetActiveSessionId; this.lastEventCursor = undefined; this.lastEventSequence = undefined; this.latestSnapshot = undefined; this.latestSnapshotIsFresh = false; this.retiredEventGenerations.clear(); this.pendingReattachActiveSessionIds.add(targetActiveSessionId); let reattached = false; try { const supportsExtensionUi = this.options.supportsExtensionUi !== false; const result = await this.requestData<DaemonAttachResult>({ type: "reattach", activeSessionId: sourceActiveSessionId, targetActiveSessionId, supportsExtensionUi, clientId: this.clientId, capabilities: [ "attach_snapshot", "event_sequence", ...(supportsExtensionUi ? (["extension_ui"] as const) : []), "slim_attach", "chunked_snapshot", ...(this.options.ownedSession ? (["client_owned_sessions"] as const) : []), ], env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined, launchEnv: this.options.ownedSession ? collectDaemonLaunchEnv() : undefined, telemetryDisabled: this.options.telemetryDisabled, }); // Reattach rebinds the connection (and drops any direct link); pauses on the old session are gone. this.sessionInputPauses.clear(); this.sessionInputPauseGeneration++; reattached = true; this.activeSessionId = result.activeSessionId; this.activeSideQuestionIds.clear(); if (result.snapshotStream) { try { await this.waitForSnapshot(result.snapshotStream.id); } catch (snapshotError) { await this.snapshotRecoveryPromises.get(result.snapshotStream.id); if (!this.latestSnapshotIsFresh) { throw snapshotError; } } } else { this.applySessionSnapshot(result.snapshot, result.replay); await this.emit({ type: "session_replaced", state: result.snapshot.state, messages: result.snapshot.messages, }); } return { cancelled: false }; } catch (error) { if (!reattached) { this.activeSessionId = sourceActiveSessionId; this.lastEventCursor = previousState.lastEventCursor; this.lastEventSequence = previousState.lastEventSequence; this.latestSnapshot = previousState.latestSnapshot; this.latestSnapshotIsFresh = previousState.latestSnapshotIsFresh; this.deferredSessionEvents = previousState.deferredSessionEvents; this.deferredSessionEventsOverflowed = previousState.deferredSessionEventsOverflowed; this.retiredEventGenerations.clear(); for (const generation of previousState.retiredEventGenerations) { this.retiredEventGenerations.add(generation); } } throw error; } finally { this.pendingReattachActiveSessionIds.delete(targetActiveSessionId); } }
async fork( entryId: string, options?: AgentConnectionForkOptions, ): Promise<{ cancelled: boolean; selectedText?: string }> { return this.requestData<{ cancelled: boolean; selectedText?: string }>({ type: "fork", activeSessionId: this.activeSessionId, entryId, position: options?.position, }); }
async navigateTree( targetId: string, options?: AgentConnectionNavigateTreeOptions, ): Promise<AgentConnectionNavigateTreeResult> { return this.requestData<AgentConnectionNavigateTreeResult>({ type: "navigate_tree", activeSessionId: this.activeSessionId, targetId, summarize: options?.summarize, customInstructions: options?.customInstructions, replaceInstructions: options?.replaceInstructions, label: options?.label, }); }
async importFromJsonl(inputPath: string, cwdOverride?: string): Promise<{ cancelled: boolean }> { return this.requestData<{ cancelled: boolean }>({ type: "import_jsonl", activeSessionId: this.activeSessionId, inputPath, cwdOverride, }); }
async exportToHtml(outputPath?: string): Promise<string> { const data = await this.requestData<{ path: string }>({ type: "export_html", activeSessionId: this.activeSessionId, outputPath, }); return data.path; }
async exportToJsonl(outputPath?: string): Promise<string> { const data = await this.requestData<{ path: string }>({ type: "export_jsonl", activeSessionId: this.activeSessionId, outputPath, }); return data.path; }
async setSessionName(name: string): Promise<void> { await this.requestOk({ type: "set_session_name", activeSessionId: this.activeSessionId, name }); }
async getRlmMaxDepthStatus() { return this.requestData<{ maxDepth: number; source: "default" | "env" | "global" | "inherited" | "chat" }>({ type: "get_rlm_max_depth_status", activeSessionId: this.activeSessionId, }); }
async setRlmMaxDepth(maxDepth: number, options?: { global?: boolean }) { return this.requestData<{ maxDepth: number; source: "default" | "env" | "global" | "inherited" | "chat"; globalSaved: boolean; globalError?: string; }>({ type: "set_rlm_max_depth", activeSessionId: this.activeSessionId, maxDepth, global: options?.global, }); }
async renameSavedSession(sessionPath: string, name: string): Promise<void> { await renameDaemonSavedSession(this.client, { activeSessionId: this.activeSessionId }, sessionPath, name); }
async deleteSavedSession(sessionPath: string): Promise<DeleteSessionFileResult> { return deleteDaemonSavedSession(this.client, { activeSessionId: this.activeSessionId }, sessionPath); }
async watchSession(activeSessionId: string): Promise<AgentConnectionSessionWatcher | undefined> { // A second connection on the shared client; each one filters to its own session id. // attach() rejects for an unknown/exited session — treat that as unreachable. let connection: DaemonAgentConnection; try { const watchClient = this.client instanceof DaemonRoutedClient ? this.client.controlPlaneTransport : this.client; connection = await DaemonAgentConnection.attach(watchClient, activeSessionId, { closeClientOnDispose: false, directTransport: false, }); } catch { return undefined; } return { getMessages: () => connection.getMessages(), getCommands: () => connection.getCommands(), subscribe: (listener) => connection.subscribe(listener), getToolDefinition: (name) => connection.getToolDefinition(name), close: () => connection.dispose(), }; }
async dispose(): Promise<void> { if (this.disposed || this.disposing) { return; } this.disposing = true; if (this.activeReplacementPromise) { await this.activeReplacementPromise.catch(() => undefined); } if (this.options.ownedSession && !this.client.isConnected && this.reconnectPromise) { await Promise.race([this.reconnectPromise, delay(OWNED_SESSION_DISPOSE_RECONNECT_WAIT_MS)]).catch( () => undefined, ); } this.disposed = true; this.pendingExtensionUiRequests.length = 0; this.updateRestartPending = false; await Promise.allSettled([...this.activeSideQuestionIds].map((id) => this.abortSideQuestion(id))); await this.rosterStore?.dispose().catch(() => undefined); this.rosterStore = undefined; this.unsubscribeDaemonMessages(); this.unsubscribeDaemonClose(); if (this.options.ownedSession) { await this.requestOk({ type: "complete_owned_session", activeSessionId: this.activeSessionId }).catch( () => undefined, ); } else { await this.requestOk({ type: "detach", activeSessionId: this.activeSessionId }).catch(() => undefined); } if (this.options.closeClientOnDispose) { this.client.close(); } this.rejectSnapshotAssemblies(new Error("Daemon connection disposed during snapshot transfer")); }
async promoteToResident(): Promise<void> { await this.withOwnedSessionPromotion(async (promoteOwnedSession) => { if (!promoteOwnedSession) return; await this.requestOk({ type: "promote_owned_session", activeSessionId: this.activeSessionId }); }); }
private withOwnedSessionPromotion<T>(operation: (promoteOwnedSession: boolean) => Promise<T>): Promise<T> { const run = this.ownedSessionPromotionTail.then(async () => { const promoteOwnedSession = this.options.ownedSession === true; const result = await operation(promoteOwnedSession); if (promoteOwnedSession) { this.options.ownedSession = false; } return result; }); this.ownedSessionPromotionTail = run.then( () => undefined, () => undefined, ); return run; }
private async reconnect(cause: Error): Promise<void> { if (this.reconnectPromise) { return this.reconnectPromise; } this.reconnectPromise = (async () => { void this.emit({ type: "connection_status", status: "reconnecting", error: "session is restarting - waiting for the daemon", }); const timeoutMs = this.options.reconnectTimeoutMs ?? DAEMON_RECONNECT_TIMEOUT_MS; let deadline: number | undefined; let attempt = 0; let lastError: Error = cause; while (!this.disposed && !this.terminalCloseEmitted) { // A held direct link owns session liveness: control-plane recovery retries unbounded, // and the bounded session-plane deadline arms only once the direct link is gone. const directSessionHeld = this.client instanceof DaemonRoutedClient && this.client.hasDirectTransport; if (directSessionHeld) { deadline = undefined; } else { deadline ??= Date.now() + timeoutMs; if (Date.now() >= deadline) break; } let controlPlaneHandshakeComplete = false; try { await this.options.recoverDaemon?.(); if (this.disposed || this.terminalCloseEmitted) { return; } await this.client.reconnect(1000); await this.client.waitForHello(3000); controlPlaneHandshakeComplete = true; if (directSessionHeld) { // The roster subscription is a control-plane accessory; its usual rebind seam (attach) is skipped while held. if (this.rosterStore) await this.rosterStore.attach(this.client).catch(() => undefined); // One check after the last await, against the close handler's own dispatch outputs: // terminal closes set terminalCloseEmitted, update closes set updateRestartPending // (restoration owns the client), and recoverable closes joined this loop. if (this.disposed || this.terminalCloseEmitted || this.updateRestartPending) { return; } if (this.client instanceof DaemonRoutedClient && this.client.hasDirectTransport) { void this.emit({ type: "connection_status", status: "connected" }); return; } // The direct link died mid-recovery: rerun as a bounded session-plane reconnect. continue; } // This loop owns the retry: a socket close must reject these instead of parking them behind a hello it can never produce. try { await this.attach({ recoverable: false }); } catch (attachError) { if (this.disposed || this.terminalCloseEmitted) { return; } 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((summary) => this.matchesAttachedSession(summary)); 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 { this.boundUnlistedSession(restored); throw new DaemonSessionStartingError( this.activeSessionId, "Session worker is queued for restore", ); } } else { throw attachError; } } else { throw attachError; } } 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 }); this.emitConnectedStatus(); } return; } catch (error) { lastError = error instanceof Error ? error : new Error(String(error)); if (this.disposed || this.terminalCloseEmitted) { return; } // Nothing left to wait for: the daemon does not have this session. if (lastError instanceof DaemonSessionUnrestorableError) break; 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) || !controlPlaneHandshakeComplete || error instanceof DaemonControlPlaneTransportError || !this.client.isControlPlaneReady); if (shouldResetControlPlane) this.client.resetTransportForReconnect(); if (deadline !== undefined && deadline - Date.now() <= 0) { break; } const delayMs = Math.min( ...(deadline !== undefined ? [deadline - Date.now()] : []), 2000, directSessionHeld ? 100 * 2 ** Math.min(attempt, 4) : 250 * 2 ** Math.min(attempt, 3), ); attempt++; await new Promise((resolveDelay) => setTimeout(resolveDelay, delayMs)); } } if (!this.disposed && !this.terminalCloseEmitted) { await this.emitTerminalClose(lastError); } })().finally(() => { this.reconnectPromise = undefined; }); return this.reconnectPromise; }
private async emitTerminalClose(lastError: Error): Promise<void> { if (this.disposed || this.terminalCloseEmitted) return; this.sessionInputPauses.clear(); this.sessionInputPauseGeneration++; if (lastError instanceof DaemonSessionUnrestorableError) { this.client.resetTransportForReconnect(); } else { this.client.close(); } this.terminalCloseEmitted = true; await this.emit({ type: "closed", error: this.formatTerminalReconnectError(lastError) }); }
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.matchesAttachedSession(s)); 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; } this.boundUnlistedSession(restored); lastError = new DaemonSessionStartingError( this.activeSessionId, "Session worker is queued for restore", ); } } catch (error) { lastError = error; if (lastError instanceof DaemonSessionUnrestorableError) break; if (!isSessionNotReadyError(error)) { throw error; } } const delayMs = Math.min(deadline - Date.now(), 2000, 250 * 2 ** Math.min(attempt++, 3)); await delay(delayMs); } if (lastError instanceof DaemonSessionUnrestorableError) { await this.emitTerminalClose(lastError); throw lastError; } this.throwIfSessionUnrestorable(); 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; } 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; }
// A session the restarted daemon does not have is never coming back: fail the caller // with that verdict instead of parking it behind the reconnect banner. this.throwIfSessionUnrestorable(); 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); this.throwIfSessionUnrestorable(); 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; } 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; } } if (this.disposed || this.terminalCloseEmitted) { this.throwIfSessionUnrestorable(); 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; if (message.type === "heartbeats_changed") { await this.emit({ type: "heartbeats_changed" }); return; } if (!this.isMessageForActiveSession(message)) { return; } // Requests bypass snapshot deferral and must not advance the replay cursor. if (message.type === "extension_ui_request") { const cursor = getDaemonMessageCursor(message); if (cursor && this.retiredEventGenerations.has(cursor.generation)) return; // Retain attach-time requests until the UI subscribes; snapshots cannot restore them. if (this.listeners.size === 0) { if (this.pendingExtensionUiRequests.length >= MAX_PENDING_EXTENSION_UI_REQUESTS) { // Discard older non-dialog updates before cancelling a dialog that cannot be queued. const disposable = this.pendingExtensionUiRequests.findIndex( (request) => !isDaemonDialogExtensionUiRequest(request.method), ); if (disposable === -1) { if (isDaemonDialogExtensionUiRequest(message.method)) { await this.respondToExtensionUiRequest(message.id, { cancelled: true }); } return; } this.pendingExtensionUiRequests.splice(disposable, 1); } this.pendingExtensionUiRequests.push(message); return; } await this.emit({ type: "extension_ui_request", request: { id: message.id, method: message.method, payload: message.payload }, }); return; } if ("snapshotId" in message && this.ignoredSnapshotIds.has(message.snapshotId)) { if (message.type === "session_snapshot_end" || message.type === "session_snapshot_failed") { this.ignoredSnapshotIds.delete(message.snapshotId); } return; } if (message.type === "session_snapshot_begin") { const assembly = this.getSnapshotAssembly(message.snapshotId); assembly.begin = message; return; } if (message.type === "session_snapshot_chunk") { this.getSnapshotAssembly(message.snapshotId).chunks.set(message.index, message.messages); return; } if (message.type === "session_snapshot_end") { await this.completeSnapshotAssembly(message); return; } if (message.type === "session_snapshot_failed") { const assembly = this.getSnapshotAssembly(message.snapshotId); const purpose = assembly.begin?.purpose ?? "attach"; const snapshotError = new Error(message.error); const recoveryPromise = purpose === "replacement" || purpose === "resync" ? this.recoverFailedSnapshot(purpose, snapshotError) : undefined; if (recoveryPromise) { this.snapshotRecoveryPromises.set(message.snapshotId, recoveryPromise); } this.rejectSnapshotAssembly(message.snapshotId, assembly, snapshotError); this.ignoreSnapshotId(message.snapshotId); if (recoveryPromise) { try { await recoveryPromise; } finally { this.snapshotRecoveryPromises.delete(message.snapshotId); } } return; } if (this.isStaleSequencedMessage(message)) { return; } this.observeDaemonEventSequence(message);
if (message.type === "session_event") { if (this.deferSessionEvents) { if (this.deferredSessionEventsOverflowed) return; if (this.deferredSessionEvents.length >= MAX_DEFERRED_SESSION_EVENTS) { this.deferredSessionEvents.length = 0; this.deferredSessionEventsOverflowed = true; } else { this.deferredSessionEvents.push({ event: message.event, sequence: getDaemonMessageSequence(message), generation: getDaemonMessageCursor(message)?.generation, }); } return; } if (message.event.type !== "refine_complete" && message.event.type !== "refine_failed") { this.observeStreamingMessage(message.event); } if (message.event.type === "rlm_child_update") { this.childRosterSequence = maxEventSequence(this.childRosterSequence, getDaemonMessageSequence(message)); this.observeRlmChildUpdate(message.event.child); } this.latestSnapshotIsFresh = false; await this.emit({ type: "session_event", event: message.event }); return; } if (message.type === "side_question_event") { this.observeSideQuestionEvent(message.event); await this.emit({ type: "side_question_event", event: message.event }); return; } if (message.type === "session_status") { // Keep a cached snapshot's recap current so a later re-attach seeds it. if (this.latestSnapshot) { this.latestSnapshot = { ...this.latestSnapshot, state: { ...this.latestSnapshot.state, recap: message.recap }, }; } await this.emit({ type: "session_status", recap: message.recap }); return; } if (message.type === "session_resynced") { this.applySessionSnapshot(message.snapshot); await this.emit({ type: "session_resynced", snapshot: this.latestSnapshot! }); return; } if (message.type === "session_replaced") { this.sessionRevision++; this.pendingExtensionUiRequests.length = 0; this.attachedSessionId = message.state.sessionId; this.attachedSessionFile = message.state.sessionFile; // Buffered events belong to the replaced session; they must never // replay onto the new one. The replacement snapshot rebuilds state. this.deferredSessionEvents.length = 0; this.deferredSessionEventsOverflowed = false; if (message.snapshotFollows) { this.latestSnapshotIsFresh = false; return; } const latestSnapshot: AgentConnectionSnapshot = { state: message.state, messages: message.messages, }; if (this.lastEventSequence !== undefined) { latestSnapshot.lastEventSequence = this.lastEventSequence; } if (this.lastEventCursor) { latestSnapshot.lastEventCursor = this.lastEventCursor; } this.latestSnapshot = latestSnapshot; this.childRosterSequence = undefined; this.latestSnapshotIsFresh = true; await this.emit({ type: "session_replaced", state: message.state, messages: message.messages }); return; } if (message.type === "extension_error") { await this.emit({ type: "extension_error", extensionPath: message.extensionPath, event: message.event, error: message.error, }); return; } if (message.type === "session_closed") { if (message.reason === "update") { this.captureDaemonLogPath(); 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 }); } }
private captureDaemonLogPath(): void { const socketPath = this.client.hello?.socketPath; if (socketPath) { this.daemonLogPath = getDaemonLogPath(socketPath); } }
private formatDaemonSessionClosedError(reason: DaemonSessionClosedReason): string { const explanation: Record<DaemonSessionClosedReason, string> = { killed: this.attachedSessionFile ? `The daemon stopped this agent session. Its transcript remains saved and can be reopened with /resume ${this.attachedSessionId ?? this.activeSessionId} or from Agents View.` : "The daemon stopped this agent session. Its transcript remains saved and can be reopened from Agents View.", shutdown: "The Prime Agent daemon shut down while this window was attached. The session transcript remains saved; restart Prime Agent and reopen it from Agents View.", completed: "The daemon closed this agent session after it completed. Its transcript remains available from Agents View.", replaced: "The daemon replaced this agent session with another session. Reopen the current session from Agents View.", update: "The Prime Agent daemon restarted for an update, but this window did not restore automatically. The session transcript remains saved; restart Prime Agent and reopen it from Agents View.", }; return `${explanation[reason]} ${this.formatDaemonDiagnosticContext()}`; }
private formatDaemonConnectionClosedError(error: Error): string { return `Lost connection to the Prime Agent daemon. Cause: ${formatErrorSentence(error)} The session transcript remains saved; restart Prime Agent or reopen the session from Agents View. ${this.formatDaemonDiagnosticContext()}`; }
private formatUpdateReconnectError(error: unknown): string { return `The Prime Agent daemon restarted for an update, but this window could not reconnect to its restored session before the recovery timeout expired. Last error: ${formatErrorSentence(error)} The session transcript remains saved; restart Prime Agent and reopen it from Agents View. ${this.formatDaemonDiagnosticContext()}`; }
private formatDaemonDiagnosticContext(): string { const details: string[] = []; if (this.attachedSessionId) { details.push(`Session ID: ${this.attachedSessionId}.`); } if (this.attachedSessionFile) { details.push(`Session file: ${this.attachedSessionFile}.`); } details.push(`Diagnostic log: ${this.daemonLogPath ?? getAgentLogPath()}.`); return details.join(" "); }
/** * The daemon is the authority on what it can restore: a roster read that misses the * attached session means it has nothing to bring back, so the wait is bounded from * that read instead of from the transport window. A session that is listed but not * yet routed keeps the full window. */ private boundUnlistedSession(restored: SessionSummary | undefined): void { if (restored !== undefined) { // A miss that resolves must clear the deadline: leaving it armed dooms the // first miss of the NEXT restart to an immediate verdict, which is how a // long-lived window can be told its session is unrestorable while a fresh // window re-attaches fine. this.sessionSearchDeadline = undefined; return; } this.sessionSearchDeadline ??= Date.now() + (this.options.sessionSearchTimeoutMs ?? DAEMON_SESSION_SEARCH_TIMEOUT_MS); if (Date.now() >= this.sessionSearchDeadline) throw this.abandonMissingSession(); }
private matchesAttachedSession(summary: SessionSummary): boolean { return ( (this.attachedSessionFile !== undefined && summary.sessionFile === this.attachedSessionFile) || (this.attachedSessionId !== undefined && summary.sessionId === this.attachedSessionId) || summary.activeSessionId === this.activeSessionId ); }
/** Terminal: no recovery loop may keep waiting for a session the daemon does not have. */ private abandonMissingSession(): Error { const error = this.formatMissingSessionError(); this.sessionUnrestorable = error; return error; }
private throwIfSessionUnrestorable(): void { if (this.sessionUnrestorable) throw this.sessionUnrestorable; }
private formatMissingSessionError(): Error { const sessionId = this.attachedSessionId ?? this.activeSessionId; const transcript = this.attachedSessionFile; const transcriptSaved = transcript !== undefined && existsSync(transcript); const recovery = transcriptSaved ? `Reopen the saved transcript with /resume ${sessionId}.` : "There is nothing to resume: start a fresh session with /new."; const details = [ `Session ID: ${sessionId}.`, transcript === undefined ? "No transcript path is known." : transcriptSaved ? `Transcript: ${transcript}.` : `No transcript exists at ${transcript}.`, `Diagnostic log: ${this.daemonLogPath ?? getAgentLogPath()}.`, ]; return new DaemonSessionUnrestorableError( `The Prime Agent daemon restarted, and the session this window was attached to is not in the restarted daemon's session list, so it cannot be restored. ${recovery} ${details.join(" ")}`, ); }
/** * A lost session cannot be rebound in the daemon, so the window's recovery is a session * the daemon creates now, through the same create route a relaunched window takes. Any * failure restores the verdict state, so the next session-creating command retries * instead of inheriting a half-rebuilt connection. */ private activeReplacementPromise?: Promise<{ cancelled: boolean }>;
private async replaceLostSession(options: { sessionPath?: string; cwdOverride?: string; }): Promise<{ cancelled: boolean }> { if (this.disposed || this.disposing) throw new Error("Daemon connection is disposed."); if (this.activeReplacementPromise) return this.activeReplacementPromise; const verdict = this.sessionUnrestorable; // Assembled from the pre-loss state: the snapshot is the window's own directory. const createCommand = this.replacementCreateCommand(options); const previous = { updateRestartPending: this.updateRestartPending, updateReconnectFailed: this.updateReconnectFailed, activeSessionId: this.activeSessionId, attachedSessionId: this.attachedSessionId, attachedSessionFile: this.attachedSessionFile, lastEventCursor: this.lastEventCursor, lastEventSequence: this.lastEventSequence, latestSnapshot: this.latestSnapshot, latestSnapshotIsFresh: this.latestSnapshotIsFresh, latestSnapshotStateIsFresh: this.latestSnapshotStateIsFresh, deferredSessionEvents: this.deferredSessionEvents, deferredSessionEventsOverflowed: this.deferredSessionEventsOverflowed, pendingExtensionUiRequests: [...this.pendingExtensionUiRequests], retiredEventGenerations: new Set(this.retiredEventGenerations), }; this.sessionSearchDeadline = undefined; this.updateRestartPending = false; this.updateReconnectFailed = false; this.terminalCloseEmitted = false; this.sessionRevision++; this.attachedSessionId = undefined; this.attachedSessionFile = undefined; this.lastEventCursor = undefined; this.lastEventSequence = undefined; this.latestSnapshot = undefined; this.latestSnapshotIsFresh = false; this.latestSnapshotStateIsFresh = false; this.retiredEventGenerations.clear(); this.deferredSessionEvents = []; this.deferredSessionEventsOverflowed = false; this.pendingExtensionUiRequests.length = 0; this.activeSideQuestionIds.clear(); this.activeReplacementPromise = (async () => { let createdActiveSessionId: string | undefined; try { createdActiveSessionId = await this.createReplacementSession(createCommand); this.activeSessionId = createdActiveSessionId; await this.attach({ recoverable: false }); this.sessionInputPauses.clear(); this.sessionInputPauseGeneration++; const snapshot = await this.getInitialSnapshot({ recoverable: false }); this.sessionUnrestorable = undefined; await this.emit({ type: "session_replaced", state: snapshot.state, messages: snapshot.messages }); this.emitConnectedStatus(); return { cancelled: false }; } catch (error) { if (createdActiveSessionId) { const release = this.options.ownedSession ? { type: "complete_owned_session" as const, activeSessionId: createdActiveSessionId } : { type: "detach" as const, activeSessionId: createdActiveSessionId }; await this.requestOk(release).catch(() => undefined); } this.updateRestartPending = previous.updateRestartPending; this.updateReconnectFailed = previous.updateReconnectFailed; this.activeSessionId = previous.activeSessionId; this.attachedSessionId = previous.attachedSessionId; this.attachedSessionFile = previous.attachedSessionFile; this.lastEventCursor = previous.lastEventCursor; this.lastEventSequence = previous.lastEventSequence; this.latestSnapshot = previous.latestSnapshot; this.latestSnapshotIsFresh = previous.latestSnapshotIsFresh; this.latestSnapshotStateIsFresh = previous.latestSnapshotStateIsFresh; this.deferredSessionEvents = previous.deferredSessionEvents; this.deferredSessionEventsOverflowed = previous.deferredSessionEventsOverflowed; this.pendingExtensionUiRequests.length = 0; this.pendingExtensionUiRequests.push(...previous.pendingExtensionUiRequests); this.retiredEventGenerations.clear(); for (const generation of previous.retiredEventGenerations) { this.retiredEventGenerations.add(generation); } this.sessionUnrestorable = verdict; this.terminalCloseEmitted = true; // The missing-cwd prompt belongs to the UI; every other failure is the verdict again. throw error instanceof MissingSessionCwdError ? error : this.formatReplacementFailure(error); } })().finally(() => { this.activeReplacementPromise = undefined; }); return this.activeReplacementPromise; }
/** * The create command a relaunched window would send for this window's replacement: a * fresh session in the directory the window was working in, or the transcript the UI * asked to switch to with the transcript's own cwd unless an override was chosen. */ private replacementCreateCommand(options: { sessionPath?: string; cwdOverride?: string; }): Extract<DaemonCommand, { type: "create" }> { const cwd = options.cwdOverride ?? (options.sessionPath === undefined ? (this.latestSnapshot?.state.cwd ?? process.cwd()) : undefined); const sessionDir = this.latestSnapshot?.state.sessionDir; const config: AgentSessionRuntimeConfig = { ...(cwd === undefined ? {} : { cwd }), ...(sessionDir === undefined ? {} : { sessionDir }), ...(this.options.telemetryDisabled ? { telemetryDisabled: true as const } : {}), }; return { type: "create", config, sessionPath: options.sessionPath, lifecycle: this.options.ownedSession ? "client_owned" : undefined, env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined, launchEnv: this.options.ownedSession ? collectDaemonLaunchEnv() : undefined, }; }
/** Create the replacement session on the daemon and return the identity this window binds to. */ private async createReplacementSession(command: Extract<DaemonCommand, { type: "create" }>): Promise<string> { if (this.options.ownedSession && !this.client.supportsServerCapability("client_owned_sessions")) { throw new DaemonCapabilityUnavailableError("create", "client_owned_sessions"); } // The old worker left with the daemon; a stale direct link would route the attach // to a session the daemon no longer has. if (this.client instanceof DaemonRoutedClient) this.client.fallbackToSupervisor(); await this.options.recoverDaemon?.(); await this.client.reconnect(1000); await this.client.waitForHello(3000); const response = await this.client.request(command); if (!response.success) throw deserializeDaemonError(response); const created = response.data as SessionSummary | undefined; const createdActiveSessionId = created ? getAttachActiveSessionId(created) : undefined; if (createdActiveSessionId === undefined) { throw new Error("Daemon returned an invalid create response"); } return createdActiveSessionId; }
/** The verdict stands and its replacement failed: name both, and the recovery that always works. */ private formatReplacementFailure(cause: unknown): Error { return new DaemonSessionUnrestorableError( `The Prime Agent daemon lost the session this window was attached to, and starting a replacement session failed: ${formatErrorSentence(cause)} Retry this command, or relaunch this window. ${this.formatDaemonDiagnosticContext()}`, ); }
private formatTerminalReconnectError(lastError: Error): string { if (lastError instanceof DaemonSessionUnrestorableError) { return lastError.message; } return this.updateRestartPending || getDaemonSocketCloseReason(lastError) === "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()}`; }
private reconnectAfterUpdate(): Promise<void> { if (this.updateReconnectPromise) { return this.updateReconnectPromise; } this.pendingExtensionUiRequests.length = 0; void this.emit({ type: "connection_status", status: "reconnecting", error: "session is restarting - waiting for the daemon", }); const reconnectPromise = reconnectDaemonTransportAfterUpdate( this.client, this.options.reconnectTimeoutMs ?? UPDATE_RECONNECT_TIMEOUT_MS, ) .then(() => this.restoreConnectionAfterUpdate()) .then(() => { if (!this.disposed && !this.terminalCloseEmitted) { this.emitConnectedStatus(); } }) .catch(async (error: unknown) => { this.updateRestartPending = false; this.updateReconnectFailed = true; if (!this.disposed && !this.terminalCloseEmitted) { this.terminalCloseEmitted = true; await this.emit({ type: "closed", error: error instanceof DaemonSessionUnrestorableError ? error.message : this.formatUpdateReconnectError(error), }); } }) .finally(() => { if (this.updateReconnectPromise === reconnectPromise) { this.updateReconnectPromise = undefined; } }); this.updateReconnectPromise = reconnectPromise; return reconnectPromise; }
private async restoreConnectionAfterUpdate(): Promise<void> { const sessionId = this.attachedSessionId; const sessionFile = this.attachedSessionFile; if (!sessionId && !sessionFile) { throw new Error("the previous session identity is unavailable"); } 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;
const response = await this.client.request({ type: "list" }, 30000, { recoverable: false }); if (this.disposed || this.terminalCloseEmitted) return; if (!response.success) { throw deserializeDaemonError(response); } const sessions = readSessionSummaries(response.data); const restored = sessions.find((summary) => this.matchesAttachedSession(summary)); if (restored?.activeSessionId) { 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(); } await this.attach({ recoverable: false }); if (this.disposed || this.terminalCloseEmitted) return; const snapshot = await this.getInitialSnapshot({ recoverable: false }); if (this.disposed || this.terminalCloseEmitted) return; this.updateRestartPending = false; void this.emit({ type: "session_resynced", snapshot }); return; } this.boundUnlistedSession(restored); lastError = new DaemonSessionStartingError(this.activeSessionId, "Session worker is queued for restore"); } catch (error) { if (error instanceof DaemonSessionUnrestorableError) throw 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; throw lastError ?? new Error("the restored session did not become available"); }
private getSnapshotAssembly(snapshotId: string): DaemonSnapshotAssembly { const existing = this.snapshotAssemblies.get(snapshotId); if (existing) { return existing; } let resolveSnapshot!: (snapshot: DaemonSessionSnapshot) => void; let rejectSnapshot!: (error: Error) => void; const promise = new Promise<DaemonSessionSnapshot>((resolve, reject) => { resolveSnapshot = resolve; rejectSnapshot = reject; }); void promise.catch(() => undefined); const timeout = setTimeout(() => { const current = this.snapshotAssemblies.get(snapshotId); if (current) { current.reject(new Error(`Timed out waiting for snapshot ${snapshotId}`)); this.snapshotAssemblies.delete(snapshotId); this.ignoreSnapshotId(snapshotId); } }, this.options.snapshotTimeoutMs ?? DAEMON_SNAPSHOT_TIMEOUT_MS); timeout.unref(); const assembly: DaemonSnapshotAssembly = { chunks: new Map(), promise, resolve: resolveSnapshot, reject: rejectSnapshot, timeout, }; this.snapshotAssemblies.set(snapshotId, assembly); return assembly; }
private rejectSnapshotAssemblies(error: Error): void { for (const assembly of this.snapshotAssemblies.values()) { clearTimeout(assembly.timeout); assembly.reject(error); } this.snapshotAssemblies.clear(); this.completedSnapshots.clear(); this.snapshotRecoveryPromises.clear(); this.ignoredSnapshotIds.clear(); }
private ignoreSnapshotId(snapshotId: string): void { this.ignoredSnapshotIds.add(snapshotId); while (this.ignoredSnapshotIds.size > MAX_IGNORED_SNAPSHOT_IDS) { const oldest = this.ignoredSnapshotIds.values().next().value; if (oldest === undefined) { break; } this.ignoredSnapshotIds.delete(oldest); } }
private rejectSnapshotAssembly(snapshotId: string, assembly: DaemonSnapshotAssembly, error: Error): void { assembly.reject(error); clearTimeout(assembly.timeout); if (assembly.begin?.purpose && assembly.begin.purpose !== "attach") { this.snapshotAssemblies.delete(snapshotId); } }
private async recoverFailedSnapshot(purpose: "replacement" | "resync", snapshotError: Error): Promise<void> { this.latestSnapshotIsFresh = false; if (purpose === "replacement") { this.latestSnapshot = undefined; } try { const snapshot = await this.getInitialSnapshot(); if (this.disposed || this.terminalCloseEmitted) { return; } this.attachedSessionId = snapshot.state.sessionId; this.attachedSessionFile = snapshot.state.sessionFile; if (purpose === "replacement") { await this.emit({ type: "session_replaced", state: snapshot.state, messages: snapshot.messages }); } else { await this.emit({ type: "session_resynced", snapshot }); } } catch (recoveryError) { if (this.disposed || this.terminalCloseEmitted) { return; } this.terminalCloseEmitted = true; await this.emit({ type: "closed", error: `Failed to recover from a ${purpose} snapshot transfer. Snapshot error: ${formatErrorSentence(snapshotError)} Recovery error: ${formatErrorSentence(recoveryError)} ${this.formatDaemonDiagnosticContext()}`, }); } }
private async waitForSnapshot(snapshotId: string): Promise<DaemonSessionSnapshot> { const completed = this.completedSnapshots.get(snapshotId); if (completed) { this.completedSnapshots.delete(snapshotId); return completed; } const assembly = this.getSnapshotAssembly(snapshotId); try { return await assembly.promise; } finally { clearTimeout(assembly.timeout); this.snapshotAssemblies.delete(snapshotId); this.completedSnapshots.delete(snapshotId); } }
private applySessionSnapshot(snapshot: DaemonSessionSnapshot, replay?: DaemonReplayInfo): void { this.dropDeferredSessionEventsThrough(snapshot.lastEventSequence, snapshot.lastEventCursor); if (snapshot.lastEventCursor) { this.observeEventCursor(snapshot.lastEventCursor); } this.lastEventSequence = maxEventSequence(this.lastEventSequence, snapshot.lastEventSequence); this.attachedSessionId = snapshot.state.sessionId; this.attachedSessionFile = snapshot.state.sessionFile; this.latestSnapshot = mapDaemonSessionSnapshot(snapshot, replay); this.latestSnapshot.lastEventSequence = this.lastEventSequence; this.latestSnapshot.lastEventCursor = this.lastEventCursor; this.childRosterSequence = Array.isArray(snapshot.children) ? snapshot.lastEventSequence : undefined; this.latestSnapshotIsFresh = true; this.latestSnapshotStateIsFresh = true; }
private async completeSnapshotAssembly( message: Extract<DaemonOutbound, { type: "session_snapshot_end" }>, ): Promise<void> { const assembly = this.getSnapshotAssembly(message.snapshotId); if (!assembly.begin) { this.rejectSnapshotAssembly( message.snapshotId, assembly, new Error(`Snapshot ${message.snapshotId} ended before it began`), ); return; } if (assembly.chunks.size !== message.chunkCount) { this.rejectSnapshotAssembly( message.snapshotId, assembly, new Error( `Snapshot ${message.snapshotId} ended with ${assembly.chunks.size} of ${message.chunkCount} chunks`, ), ); return; } const messages: AgentMessage[] = []; for (let index = 0; index < message.chunkCount; index++) { const chunk = assembly.chunks.get(index); if (!chunk) { this.rejectSnapshotAssembly( message.snapshotId, assembly, new Error(`Snapshot ${message.snapshotId} is missing chunk ${index}`), ); return; } messages.push(...chunk); } if (messages.length !== assembly.begin.messageCount) { this.rejectSnapshotAssembly( message.snapshotId, assembly, new Error( `Snapshot ${message.snapshotId} contained ${messages.length} of ${assembly.begin.messageCount} messages`, ), ); return; } const snapshot: DaemonSessionSnapshot = { ...assembly.begin.snapshot, messages, lastEventSequence: message.lastEventSequence, lastEventCursor: message.lastEventCursor, }; const purpose = assembly.begin.purpose ?? "attach"; // Attach owns the revision guard, but its snapshot must precede the next record in this socket read. if (purpose === "attach") assembly.apply?.(snapshot); assembly.resolve(snapshot); clearTimeout(assembly.timeout); if (purpose === "attach") return; this.applySessionSnapshot(snapshot); this.snapshotAssemblies.delete(message.snapshotId); if (this.pendingReattachActiveSessionIds.has(message.activeSessionId)) { this.completedSnapshots.set(message.snapshotId, snapshot); while (this.completedSnapshots.size > MAX_COMPLETED_SNAPSHOTS) { const oldest = this.completedSnapshots.keys().next().value; if (oldest === undefined) { break; } this.completedSnapshots.delete(oldest); } } if (purpose === "replacement") { await this.emit({ type: "session_replaced", state: snapshot.state, messages }); } else if (purpose === "resync") { await this.emit({ type: "session_resynced", snapshot: this.latestSnapshot! }); } }
private observeRlmChildUpdate(child: AgentConnectionRlmChildAgentSnapshot): void { if (!this.latestSnapshot) return; const children = this.latestSnapshot.children ?? []; const index = children.findIndex((candidate) => candidate.id === child.id); const updatedChildren = [...children]; if (index === -1) { updatedChildren.push(child); } else { updatedChildren[index] = child; } this.latestSnapshot = { ...this.latestSnapshot, children: updatedChildren }; }
private observeStreamingMessage(event: AgentSessionEvent): void { if (!this.latestSnapshot) { return; } if ((event.type === "message_start" || event.type === "message_update") && event.message.role === "assistant") { this.latestSnapshot = { ...this.latestSnapshot, streamingMessage: event.message }; return; } if ((event.type === "message_end" && event.message.role === "assistant") || event.type === "agent_end") { const { streamingMessage: _streamingMessage, ...snapshot } = this.latestSnapshot; this.latestSnapshot = snapshot; } }
private isMessageForActiveSession(message: DaemonOutbound): boolean { if (!("activeSessionId" in message)) { return false; } return message.activeSessionId === this.activeSessionId; }
private isStaleSequencedMessage(message: DaemonOutbound): boolean { const cursor = getDaemonMessageCursor(message); if (cursor) { if (this.retiredEventGenerations.has(cursor.generation)) { return true; } return ( this.lastEventCursor?.generation === cursor.generation && cursor.sequence <= this.lastEventCursor.sequence ); } const sequence = getDaemonMessageSequence(message); return sequence !== undefined && this.lastEventSequence !== undefined && sequence <= this.lastEventSequence; }
private observeDaemonEventSequence(message: DaemonOutbound): void { const cursor = getDaemonMessageCursor(message); if (cursor) { this.observeEventCursor(cursor); this.lastEventSequence = cursor.sequence; return; } const sequence = getDaemonMessageSequence(message); if (sequence === undefined) { return; } this.lastEventSequence = this.lastEventSequence === undefined ? sequence : Math.max(this.lastEventSequence, sequence); if (this.lastEventCursor) { this.lastEventCursor = { ...this.lastEventCursor, sequence: Math.max(this.lastEventCursor.sequence, sequence), }; } }
private observeEventCursor(cursor: DaemonEventCursor): void { const current = this.lastEventCursor; if (current?.generation !== cursor.generation) this.lastEventSequence = cursor.sequence; if (current && current.generation !== cursor.generation) { this.retiredEventGenerations.add(current.generation); } if (!current || current.generation !== cursor.generation || cursor.sequence > current.sequence) { this.lastEventCursor = cursor; } }
private async emit(event: AgentConnectionEvent): Promise<void> { const deliveries: Promise<void>[] = []; for (const listener of [...this.listeners]) { try { deliveries.push(Promise.resolve(listener(event))); } catch { // One attachment must not interrupt delivery or transport recovery for the others. } } await Promise.allSettled(deliveries); }
private observeSideQuestionEvent(event: AgentConnectionSideQuestionEvent): void { if (event.status !== "running") { this.activeSideQuestionIds.delete(event.id); } }}
function readSessionSummaries(value: unknown): SessionSummary[] { if (!value || typeof value !== "object" || !Array.isArray((value as { sessions?: unknown }).sessions)) { throw new Error("Daemon returned an invalid session list response"); } return (value as { sessions: SessionSummary[] }).sessions;}
function getAttachActiveSessionId(result: SessionSummary | DaemonAttachResult): string { if ("snapshot" in result) { return result.activeSessionId; } return result.activeSessionId ?? result.id;}
function getAttachLastEventSequence(result: SessionSummary | DaemonAttachResult): number | undefined { if ("lastEventSequence" in result) { return result.lastEventSequence; } return undefined;}
function getAttachLastEventCursor(result: SessionSummary | DaemonAttachResult): DaemonEventCursor | undefined { if ("lastEventCursor" in result) { return result.lastEventCursor; } return undefined;}
function maxEventSequence(current: number | undefined, observed: number | undefined): number | undefined { if (current === undefined) { return observed; } if (observed === undefined) { return current; } return Math.max(current, observed);}
function mapDaemonSessionSnapshot(snapshot: DaemonSessionSnapshot, replay?: DaemonReplayInfo): AgentConnectionSnapshot { const connectionSnapshot: AgentConnectionSnapshot = { state: snapshot.state, messages: snapshot.messages, ...(snapshot.summary.streamingMessage ? { streamingMessage: snapshot.summary.streamingMessage } : {}), lastEventSequence: snapshot.lastEventSequence, lastEventCursor: snapshot.lastEventCursor, }; if (snapshot.sessionContext) { connectionSnapshot.sessionContext = snapshot.sessionContext; } if (snapshot.sessionTree) { connectionSnapshot.sessionTree = snapshot.sessionTree; } if (snapshot.parent) { connectionSnapshot.parent = snapshot.parent; } if (snapshot.children) { connectionSnapshot.children = snapshot.children; } if (replay) { connectionSnapshot.replay = replay; } return connectionSnapshot;}
function getDaemonMessageSequence(message: DaemonOutbound): number | undefined { if (!("meta" in message)) { return undefined; } return message.meta?.sequence;}
function getDaemonMessageCursor(message: DaemonOutbound): DaemonEventCursor | undefined { if (!("meta" in message)) { return undefined; } return message.meta?.cursor;}
function invalidatesCachedSnapshot(commandType: DaemonCommandBody["type"]): boolean { switch (commandType) { case "attach": case "reattach": case "detach": case "list": case "list_saved_sessions": case "wait_for_idle": case "get_state": case "get_connection_state": case "get_messages": case "get_session_stats": case "get_commands": case "get_resource_snapshot": case "get_queue": case "cron_list": case "heartbeats_list": case "get_session_context": case "get_session_tree": case "get_user_messages_for_forking": case "get_last_assistant_text": case "get_system_prompt": case "get_tool_definition": case "start_side_question": case "abort_side_question": case "export_html": case "export_jsonl": return false; default: return true; }}