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.
313 kB · 8394 lines
TypeScript
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036403740384039404040414042404340444045404640474048404940504051405240534054405540564057405840594060406140624063406440654066406740684069407040714072407340744075407640774078407940804081408240834084408540864087408840894090409140924093409440954096409740984099410041014102410341044105410641074108410941104111411241134114411541164117411841194120412141224123412441254126412741284129413041314132413341344135413641374138413941404141414241434144414541464147414841494150415141524153415441554156415741584159416041614162416341644165416641674168416941704171417241734174417541764177417841794180418141824183418441854186418741884189419041914192419341944195419641974198419942004201420242034204420542064207420842094210421142124213421442154216421742184219422042214222422342244225422642274228422942304231423242334234423542364237423842394240424142424243424442454246424742484249425042514252425342544255425642574258425942604261426242634264426542664267426842694270427142724273427442754276427742784279428042814282428342844285428642874288428942904291429242934294429542964297429842994300430143024303430443054306430743084309431043114312431343144315431643174318431943204321432243234324432543264327432843294330433143324333433443354336433743384339434043414342434343444345434643474348434943504351435243534354435543564357435843594360436143624363436443654366436743684369437043714372437343744375437643774378437943804381438243834384438543864387438843894390439143924393439443954396439743984399440044014402440344044405440644074408440944104411441244134414441544164417441844194420442144224423442444254426442744284429443044314432443344344435443644374438443944404441444244434444444544464447444844494450445144524453445444554456445744584459446044614462446344644465446644674468446944704471447244734474447544764477447844794480448144824483448444854486448744884489449044914492449344944495449644974498449945004501450245034504450545064507450845094510451145124513451445154516451745184519452045214522452345244525452645274528452945304531453245334534453545364537453845394540454145424543454445454546454745484549455045514552455345544555455645574558455945604561456245634564456545664567456845694570457145724573457445754576457745784579458045814582458345844585458645874588458945904591459245934594459545964597459845994600460146024603460446054606460746084609461046114612461346144615461646174618461946204621462246234624462546264627462846294630463146324633463446354636463746384639464046414642464346444645464646474648464946504651465246534654465546564657465846594660466146624663466446654666466746684669467046714672467346744675467646774678467946804681468246834684468546864687468846894690469146924693469446954696469746984699470047014702470347044705470647074708470947104711471247134714471547164717471847194720472147224723472447254726472747284729473047314732473347344735473647374738473947404741474247434744474547464747474847494750475147524753475447554756475747584759476047614762476347644765476647674768476947704771477247734774477547764777477847794780478147824783478447854786478747884789479047914792479347944795479647974798479948004801480248034804480548064807480848094810481148124813481448154816481748184819482048214822482348244825482648274828482948304831483248334834483548364837483848394840484148424843484448454846484748484849485048514852485348544855485648574858485948604861486248634864486548664867486848694870487148724873487448754876487748784879488048814882488348844885488648874888488948904891489248934894489548964897489848994900490149024903490449054906490749084909491049114912491349144915491649174918491949204921492249234924492549264927492849294930493149324933493449354936493749384939494049414942494349444945494649474948494949504951495249534954495549564957495849594960496149624963496449654966496749684969497049714972497349744975497649774978497949804981498249834984498549864987498849894990499149924993499449954996499749984999500050015002500350045005500650075008500950105011501250135014501550165017501850195020502150225023502450255026502750285029503050315032503350345035503650375038503950405041504250435044504550465047504850495050505150525053505450555056505750585059506050615062506350645065506650675068506950705071507250735074507550765077507850795080508150825083508450855086508750885089509050915092509350945095509650975098509951005101510251035104510551065107510851095110511151125113511451155116511751185119512051215122512351245125512651275128512951305131513251335134513551365137513851395140514151425143514451455146514751485149515051515152515351545155515651575158515951605161516251635164516551665167516851695170517151725173517451755176517751785179518051815182518351845185518651875188518951905191519251935194519551965197519851995200520152025203520452055206520752085209521052115212521352145215521652175218521952205221522252235224522552265227522852295230523152325233523452355236523752385239524052415242524352445245524652475248524952505251525252535254525552565257525852595260526152625263526452655266526752685269527052715272527352745275527652775278527952805281528252835284528552865287528852895290529152925293529452955296529752985299530053015302530353045305530653075308530953105311531253135314531553165317531853195320532153225323532453255326532753285329533053315332533353345335533653375338533953405341534253435344534553465347534853495350535153525353535453555356535753585359536053615362536353645365536653675368536953705371537253735374537553765377537853795380538153825383538453855386538753885389539053915392539353945395539653975398539954005401540254035404540554065407540854095410541154125413541454155416541754185419542054215422542354245425542654275428542954305431543254335434543554365437543854395440544154425443544454455446544754485449545054515452545354545455545654575458545954605461546254635464546554665467546854695470547154725473547454755476547754785479548054815482548354845485548654875488548954905491549254935494549554965497549854995500550155025503550455055506550755085509551055115512551355145515551655175518551955205521552255235524552555265527552855295530553155325533553455355536553755385539554055415542554355445545554655475548554955505551555255535554555555565557555855595560556155625563556455655566556755685569557055715572557355745575557655775578557955805581558255835584558555865587558855895590559155925593559455955596559755985599560056015602560356045605560656075608560956105611561256135614561556165617561856195620562156225623562456255626562756285629563056315632563356345635563656375638563956405641564256435644564556465647564856495650565156525653565456555656565756585659566056615662566356645665566656675668566956705671567256735674567556765677567856795680568156825683568456855686568756885689569056915692569356945695569656975698569957005701570257035704570557065707570857095710571157125713571457155716571757185719572057215722572357245725572657275728572957305731573257335734573557365737573857395740574157425743574457455746574757485749575057515752575357545755575657575758575957605761576257635764576557665767576857695770577157725773577457755776577757785779578057815782578357845785578657875788578957905791579257935794579557965797579857995800580158025803580458055806580758085809581058115812581358145815581658175818581958205821582258235824582558265827582858295830583158325833583458355836583758385839584058415842584358445845584658475848584958505851585258535854585558565857585858595860586158625863586458655866586758685869587058715872587358745875587658775878587958805881588258835884588558865887588858895890589158925893589458955896589758985899590059015902590359045905590659075908590959105911591259135914591559165917591859195920592159225923592459255926592759285929593059315932593359345935593659375938593959405941594259435944594559465947594859495950595159525953595459555956595759585959596059615962596359645965596659675968596959705971597259735974597559765977597859795980598159825983598459855986598759885989599059915992599359945995599659975998599960006001600260036004600560066007600860096010601160126013601460156016601760186019602060216022602360246025602660276028602960306031603260336034603560366037603860396040604160426043604460456046604760486049605060516052605360546055605660576058605960606061606260636064606560666067606860696070607160726073607460756076607760786079608060816082608360846085608660876088608960906091609260936094609560966097609860996100610161026103610461056106610761086109611061116112611361146115611661176118611961206121612261236124612561266127612861296130613161326133613461356136613761386139614061416142614361446145614661476148614961506151615261536154615561566157615861596160616161626163616461656166616761686169617061716172617361746175617661776178617961806181618261836184618561866187618861896190619161926193619461956196619761986199620062016202620362046205620662076208620962106211621262136214621562166217621862196220622162226223622462256226622762286229623062316232623362346235623662376238623962406241624262436244624562466247624862496250625162526253625462556256625762586259626062616262626362646265626662676268626962706271627262736274627562766277627862796280628162826283628462856286628762886289629062916292629362946295629662976298629963006301630263036304630563066307630863096310631163126313631463156316631763186319632063216322632363246325632663276328632963306331633263336334633563366337633863396340634163426343634463456346634763486349635063516352635363546355635663576358635963606361636263636364636563666367636863696370637163726373637463756376637763786379638063816382638363846385638663876388638963906391639263936394639563966397639863996400640164026403640464056406640764086409641064116412641364146415641664176418641964206421642264236424642564266427642864296430643164326433643464356436643764386439644064416442644364446445644664476448644964506451645264536454645564566457645864596460646164626463646464656466646764686469647064716472647364746475647664776478647964806481648264836484648564866487648864896490649164926493649464956496649764986499650065016502650365046505650665076508650965106511651265136514651565166517651865196520652165226523652465256526652765286529653065316532653365346535653665376538653965406541654265436544654565466547654865496550655165526553655465556556655765586559656065616562656365646565656665676568656965706571657265736574657565766577657865796580658165826583658465856586658765886589659065916592659365946595659665976598659966006601660266036604660566066607660866096610661166126613661466156616661766186619662066216622662366246625662666276628662966306631663266336634663566366637663866396640664166426643664466456646664766486649665066516652665366546655665666576658665966606661666266636664666566666667666866696670667166726673667466756676667766786679668066816682668366846685668666876688668966906691669266936694669566966697669866996700670167026703670467056706670767086709671067116712671367146715671667176718671967206721672267236724672567266727672867296730673167326733673467356736673767386739674067416742674367446745674667476748674967506751675267536754675567566757675867596760676167626763676467656766676767686769677067716772677367746775677667776778677967806781678267836784678567866787678867896790679167926793679467956796679767986799680068016802680368046805680668076808680968106811681268136814681568166817681868196820682168226823682468256826682768286829683068316832683368346835683668376838683968406841684268436844684568466847684868496850685168526853685468556856685768586859686068616862686368646865686668676868686968706871687268736874687568766877687868796880688168826883688468856886688768886889689068916892689368946895689668976898689969006901690269036904690569066907690869096910691169126913691469156916691769186919692069216922692369246925692669276928692969306931693269336934693569366937693869396940694169426943694469456946694769486949695069516952695369546955695669576958695969606961696269636964696569666967696869696970697169726973697469756976697769786979698069816982698369846985698669876988698969906991699269936994699569966997699869997000700170027003700470057006700770087009701070117012701370147015701670177018701970207021702270237024702570267027702870297030703170327033703470357036703770387039704070417042704370447045704670477048704970507051705270537054705570567057705870597060706170627063706470657066706770687069707070717072707370747075707670777078707970807081708270837084708570867087708870897090709170927093709470957096709770987099710071017102710371047105710671077108710971107111711271137114711571167117711871197120712171227123712471257126712771287129713071317132713371347135713671377138713971407141714271437144714571467147714871497150715171527153715471557156715771587159716071617162716371647165716671677168716971707171717271737174717571767177717871797180718171827183718471857186718771887189719071917192719371947195719671977198719972007201720272037204720572067207720872097210721172127213721472157216721772187219722072217222722372247225722672277228722972307231723272337234723572367237723872397240724172427243724472457246724772487249725072517252725372547255725672577258725972607261726272637264726572667267726872697270727172727273727472757276727772787279728072817282728372847285728672877288728972907291729272937294729572967297729872997300730173027303730473057306730773087309731073117312731373147315731673177318731973207321732273237324732573267327732873297330733173327333733473357336733773387339734073417342734373447345734673477348734973507351735273537354735573567357735873597360736173627363736473657366736773687369737073717372737373747375737673777378737973807381738273837384738573867387738873897390739173927393739473957396739773987399740074017402740374047405740674077408740974107411741274137414741574167417741874197420742174227423742474257426742774287429743074317432743374347435743674377438743974407441744274437444744574467447744874497450745174527453745474557456745774587459746074617462746374647465746674677468746974707471747274737474747574767477747874797480748174827483748474857486748774887489749074917492749374947495749674977498749975007501750275037504750575067507750875097510751175127513751475157516751775187519752075217522752375247525752675277528752975307531753275337534753575367537753875397540754175427543754475457546754775487549755075517552755375547555755675577558755975607561756275637564756575667567756875697570757175727573757475757576757775787579758075817582758375847585758675877588758975907591759275937594759575967597759875997600760176027603760476057606760776087609761076117612761376147615761676177618761976207621762276237624762576267627762876297630763176327633763476357636763776387639764076417642764376447645764676477648764976507651765276537654765576567657765876597660766176627663766476657666766776687669767076717672767376747675767676777678767976807681768276837684768576867687768876897690769176927693769476957696769776987699770077017702770377047705770677077708770977107711771277137714771577167717771877197720772177227723772477257726772777287729773077317732773377347735773677377738773977407741774277437744774577467747774877497750775177527753775477557756775777587759776077617762776377647765776677677768776977707771777277737774777577767777777877797780778177827783778477857786778777887789779077917792779377947795779677977798779978007801780278037804780578067807780878097810781178127813781478157816781778187819782078217822782378247825782678277828782978307831783278337834783578367837783878397840784178427843784478457846784778487849785078517852785378547855785678577858785978607861786278637864786578667867786878697870787178727873787478757876787778787879788078817882788378847885788678877888788978907891789278937894789578967897789878997900790179027903790479057906790779087909791079117912791379147915791679177918791979207921792279237924792579267927792879297930793179327933793479357936793779387939794079417942794379447945794679477948794979507951795279537954795579567957795879597960796179627963796479657966796779687969797079717972797379747975797679777978797979807981798279837984798579867987798879897990799179927993799479957996799779987999800080018002800380048005800680078008800980108011801280138014801580168017801880198020802180228023802480258026802780288029803080318032803380348035803680378038803980408041804280438044804580468047804880498050805180528053805480558056805780588059806080618062806380648065806680678068806980708071807280738074807580768077807880798080808180828083808480858086808780888089809080918092809380948095809680978098809981008101810281038104810581068107810881098110811181128113811481158116811781188119812081218122812381248125812681278128812981308131813281338134813581368137813881398140814181428143814481458146814781488149815081518152815381548155815681578158815981608161816281638164816581668167816881698170817181728173817481758176817781788179818081818182818381848185818681878188818981908191819281938194819581968197819881998200820182028203820482058206820782088209821082118212821382148215821682178218821982208221822282238224822582268227822882298230823182328233823482358236823782388239824082418242824382448245824682478248824982508251825282538254825582568257825882598260826182628263826482658266826782688269827082718272827382748275827682778278827982808281828282838284828582868287828882898290829182928293829482958296829782988299830083018302830383048305830683078308830983108311831283138314831583168317831883198320832183228323832483258326832783288329833083318332833383348335833683378338833983408341834283438344834583468347834883498350835183528353835483558356835783588359836083618362836383648365836683678368836983708371837283738374837583768377837883798380838183828383838483858386838783888389839083918392839383948395/** * Background daemon mode. * * The daemon owns live AgentSessionRuntime instances and exposes a small JSONL * protocol over a local socket. Clients can attach/detach from sessions without * disposing the underlying agent loop. */
import { createHash, randomUUID, timingSafeEqual } from "node:crypto";import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs";import { stat } from "node:fs/promises";import { createConnection, createServer, type Server, type Socket } from "node:net";import { basename, dirname, isAbsolute, join, resolve } from "node:path";import type { AgentMessage } from "@earendil-works/pi-agent-core";import { type Api, findEnvKeys, getLogger, type Model } from "@earendil-works/pi-ai";import { createCliSubprocessEnv, createCliSubprocessLaunchSpec } from "../../cli/subprocess-launch.js";import { appendRotatingLog, getCronJobsPath, getDaemonLogPath, getDaemonUpdateRestartManifestPath, getSessionsDir, VERSION,} from "../../config.js";import { AGENT_FAMILY_REACH_ERROR, AGENT_MESSAGE_BACKPRESSURE_WAIT_MS, AGENT_MESSAGE_SOURCE, type AgentFamilyCatalogEntry, type AgentFamilyMember, type AgentFamilyRelationship, type AgentSessionMessageAgentSummary, type AgentSessionMessageController, type AgentSessionMessageDeliveryStatus, type AgentSessionMessageEndpoint, type AgentSessionMessageListResult, type AgentSessionMessagePayload, AgentSessionMessageRateLimiter, type AgentSessionMessageReceipt, type AgentSessionMessageSender, agentFamilyRelationship, assertAgentFamilyReach, assertAgentSessionNameAvailable, assertDirectAgentMessageTarget, createAgentSessionMessage, createAgentSessionMessageId, createAgentSessionMessageReceipt, DEFAULT_AGENT_MESSAGE_MAX_CHARS, DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY, DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS, formatAgentSessionNameUnavailable, normalizeAgentSessionMessage, selectAgentFamily, sessionNameReservationKey,} from "../../core/agent-messages.js";import { AGENT_OBSERVE_PREVIEW_MAX_CHARS, type AgentObserveAgentSnapshot, type AgentObserveAgentSummary, type AgentObserveController, type AgentObserveListResult, type AgentObserveRecentMessagesInput, type AgentObserveRecentMessagesResult, createAgentObserveMessagePreview, normalizeObserveLimit, normalizeObserveMaxChars,} from "../../core/agent-observe.js";import { type PromptOptions, rlmChildLabel } from "../../core/agent-session.js";import { type AgentSessionRuntimeConfig, mergeAgentSessionRuntimeConfig } from "../../core/agent-session-config.js";import { type AgentSessionRuntime, type AgentSessionRuntimeDisposeOptions, type AgentSessionRuntimeMetadata, type CreateAgentSessionRuntimeFactory, createAgentSessionRuntime,} from "../../core/agent-session-runtime.js";import { type AgentSessionCapability, type AgentSessionController, type AgentSessionOperationResult, type AgentSessionRecentTurnsInput, type AgentSessionRecentTurnsResult, type AgentSessionSendInput, type AgentSessionSendResult, type AgentSessionSummary, createAgentSessionTurnSummaries, normalizeAgentSessionMaxChars, normalizeAgentSessionTurnCount,} from "../../core/agent-sessions.js";import { type AgentCronJob, AgentCronJobStore, AgentCronScheduler, type AgentHeartbeatDeliveryMode, type AgentHeartbeatManagementAction, type AgentHeartbeatUpdateAction, DEFAULT_HEARTBEAT_SCHEDULE, isHeartbeatCronJob, normalizeHeartbeatDeliveryMode, normalizeHeartbeatSchedule, resolveHeartbeatStreamingBehavior, shouldDeferHeartbeatCronJob,} from "../../core/cron-jobs.js";import { ORPHAN_PROCESS_JOURNAL_ENV } from "../../core/orphan-process-journal.js";import { PromptAdmissionCancelledError, waitForPromptAdmission } from "../../core/prompt-admission.js";import { providerRetryPolicy } from "../../core/provider-retry.js";import type { CreateRlmRootSessionOptions, CreateRlmSubagentRuntimeOptions, RlmCreateSessionResult, SubagentRuntimeHost,} from "../../core/rlm-runtime.js";import { canPassivateSession, type IdleEvictionMinutes, type SessionPassivationSnapshot,} from "../../core/session-action-store.js";import { deleteSessionArtifacts, deleteSessionFile } from "../../core/session-file-actions.js";import { acquireSessionLease, canonicalSessionPath, type SessionLease } from "../../core/session-lease.js";import { getSessionArtifactPathForFile, readSessionInfo, resolveSessionRlmDepth, type SessionInfo, SessionManager,} from "../../core/session-manager.js";import { resolveSessionPath } from "../../core/session-resolver.js";import type { SessionStats } from "../../core/session-stats.js";import { SettingsManager } from "../../core/settings-manager.js";import { type SideQuestionRun, startSideQuestion } from "../../core/side-question.js";import { isProcessAlive, spawnHidden, waitForChildProcess } from "../../utils/child-process.js";import { tryAcquireDirLock } from "../../utils/dir-lock.js";import { killTrackedDetachedChildren } from "../../utils/shell.js";import { projectSessionTreeFlatNodes } from "../agent-connection/session-tree-projection.js";import { createAgentConnectionCommands, createAgentConnectionResourceSnapshot, createAgentConnectionState,} from "../agent-connection/snapshot.js";import { createAgentConnectionToolDefinition } from "../agent-connection/tool-definition.js";import type { AgentConnectionHeartbeat, AgentConnectionRlmChildAgentSnapshot } from "../agent-connection/types.js";import { waitForHeadlessCompletion } from "../headless-completion.js";import { initTheme } from "../interactive/theme/theme.js";import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js";import { encodePrivateFrame, PrivateFrameDecoder } from "../session-worker/private-framing.js";import { type ActiveSessionState, AmbiguousActiveSessionError, createActiveSessionId, type DaemonSocketClient, resolveActiveSessionState,} from "./active-session-state.js";import { passivatedWorkerRosterEntry, type RosterSessionSummary, rosterAgentIdForSummary, type WorkerRosterEntry, workerRosterEntryFromSummary,} from "./agent-roster.js";import { createCompactAssistantDelta } from "./compact-session-stream.js";import { createInFlightTracker, installDaemonCrashHandlers, installWorkerExitHandlers } from "./crash-handlers.js";import { DaemonClient } from "./daemon-client.js";import { filterClientEnv, withClientEnv } from "./daemon-client-env.js";import { deserializeDaemonError, serializeDaemonError } from "./daemon-errors.js";import { bindActiveSessionState } from "./daemon-extension-binding.js";import { collectDaemonLaunchEnv, createDaemonEventMeta, createDaemonReplayInfo, DAEMON_DEFAULT_CLIENT_CAPABILITIES, DAEMON_DEFAULT_SERVER_CAPABILITIES, DAEMON_PROTOCOL_INFO, DAEMON_SCHEMA_ID, DAEMON_SCHEMA_REVISION, DAEMON_SUPPORTED_CLIENT_CAPABILITIES, DAEMON_UPDATE_RESTART_FORMAT_VERSION, type DaemonAttachResult, type DaemonClientCapability, type DaemonClosingReason, type DaemonCommand, type DaemonOutbound, type DaemonResponse, type DaemonSessionClosedReason, type DaemonSessionSnapshot, type DaemonUpdateRestartManifest, type DaemonUpdateRestartSession, failure, isDaemonCommandEnvelope, isDaemonDialogExtensionUiRequest, isDaemonMutatingCommand, isSessionPlaneDaemonCommand, salvageDaemonCommandId, success, UPDATE_RESTART_DRAIN_COMMANDS,} from "./daemon-protocol.js";import { getDaemonRuntimeIdentity } from "./daemon-runtime-identity.js";import { matchesSessionIdSuffix } from "./daemon-session-id.js";import { buildRlmChildSnapshots, buildSessionList, classifySessionRosterStatus, hasLiveSessionWork, inactiveLifecycleForSession, type SessionSummary, scheduledJobRegistrations, summaryForActiveSession,} from "./daemon-session-list.js";import { DaemonSessionSummarizer } from "./daemon-session-summarizer.js";import { cleanupDaemonSocketPath, type DaemonSocketIdentity, defaultDaemonSocketPath, getDaemonSocketIdentity, normalizeSocketPath, prepareDaemonSocketPath, restrictDaemonSocketPath,} from "./daemon-socket.js";import { assertDaemonSupervisorOwnerCurrent, isDaemonShutdownAdmissionActive } from "./daemon-supervisor-ownership.js";import { DAEMON_WORKER_ACTIVE_SESSION_ID_ENV, DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY, DAEMON_WORKER_RECOVERY_JOURNAL_ENV, DAEMON_WORKER_ROLE_ENV, DAEMON_WORKER_ROSTER_CAPABILITY, DAEMON_WORKER_SUPERVISOR_SOCKET_ENV, DAEMON_WORKER_TOKEN_ENV, type DaemonWorkerCommand, type DaemonWorkerFrameHeader, type DaemonWorkerPeerGrant, type DaemonWorkerRosterOutbound, isDaemonWorkerFrameHeader, ROSTER_HEARTBEAT_INTERVAL_MS, SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV,} from "./daemon-worker-protocol.js";import { MutationDrainLatch } from "./mutation-drain-latch.js";import { createRlmLedgerRegistrySeedSource, type LegacyRlmSubagentRegistryEntry, type RlmLedgerDeleteReason, type RlmLedgerEdge, RlmSpawnLedger, readLegacyRlmSubagentRegistry as readLegacyRlmSubagentRegistryFile, tombstoneSavedSessionDelete, withPassiveRlmDescendantInfos,} from "./rlm-ledger.js";import { readRlmSubagentDisplayEntry, rlmSubagentDisplayPath, writeRlmSubagentDisplayEntry,} from "./rlm-subagent-display.js";import { serializeSavedSessionInfo } from "./saved-session-info.js";import { createSnapshotTranscriptChunks, SNAPSHOT_TARGET_CHUNK_BYTES, type SnapshotTranscriptChunkSource,} from "./snapshot-transcript-cache.js";import { SupervisorLink } from "./supervisor-link.js";import { WorkerRecoveryJournal } from "./worker-recovery-journal.js";
export interface DaemonModeOptions { socketPath?: string; defaultSessionConfig: AgentSessionRuntimeConfig; createRuntime: CreateAgentSessionRuntimeFactory; worker?: { authenticationToken: string; workerInstanceId?: string; restoreActiveSessionId?: string; };}
export type { DaemonCommand, DaemonOutbound, DaemonResponse,} from "./daemon-protocol.js";export type { SessionActivity, SessionLifecycle, SessionSummary,} from "./daemon-session-list.js";export { defaultDaemonSocketPath } from "./daemon-socket.js";
const structuredLog = getLogger("coding-agent.daemon");const WORKER_SNAPSHOT_TERMINAL_DRAIN_TIMEOUT_MS = 1_000;const UPDATE_RESTART_PREPARE_TIMEOUT_MS = 90_000;const MAX_SESSION_SNAPSHOT_STABILIZATION_RETRIES = 3;// Orphaned-worker garbage collection: a session worker whose supervisor stays// unreachable (no supervisor claim, no successful replacement launch) exits// after this window instead of retrying the resurrection loop forever.// Sessions persist on disk, and a later supervisor spawns fresh workers on// demand, so an unreachable-supervisor worker serves nothing by lingering.const WORKER_SUPERVISOR_LOST_EXIT_MS_ENV = "PRIME_AGENT_INTERNAL_WORKER_SUPERVISOR_LOST_EXIT_MS";const DEFAULT_WORKER_SUPERVISOR_LOST_EXIT_MS = 5 * 60_000;
function workerSupervisorLostExitMs(): number { const raw = Number(process.env[WORKER_SUPERVISOR_LOST_EXIT_MS_ENV]); return Number.isFinite(raw) && raw >= 0 ? raw : DEFAULT_WORKER_SUPERVISOR_LOST_EXIT_MS;}
const DAEMON_COMMAND_TYPES: ReadonlySet<string> = new Set([ "ack_result", "list", "list_saved_sessions", "create", "attach", "detach", "kill", "rename", "prompt", "cancel_prompt_admission", "prompt_and_wait", "steer", "follow_up", "restore_next_turn", "restore_actions", "append_custom_message", "resume_queue", "send_message", "agent_messages_status", "agent_messages_pause", "agent_messages_resume", "agent_messages_clear", "abort", "start_side_question", "abort_side_question", "execute_bash", "execute_bash_and_wait", "abort_bash", "cancel_rlm_child", "delete_rlm_subagent", "wait_for_idle", "wait_for_headless_completion", "get_session_header", "get_state", "get_connection_state", "get_messages", "get_rlm_children", "get_session_stats", "get_context_tree", "get_commands", "get_resource_snapshot", "replace_acp_mcp_servers", "get_model_catalog", "get_available_models", "get_queue", "mutate_queued_message", "clear_queue", "abort_and_clear_queue", "acquire_session_input_pause", "release_session_input_pause", "cron_list", "heartbeats_list", "heartbeat_manage", "cron_add", "cron_cancel", "heartbeat_get", "heartbeat_set", "heartbeat_update", "set_model", "cycle_model", "set_scoped_models", "set_thinking_level", "set_service_tier", "cycle_thinking_level", "set_transport", "set_steering_mode", "set_follow_up_mode", "set_auto_compaction", "set_auto_retry", "compact", "refine", "abort_compaction", "abort_branch_summary", "abort_retry", "reload", "new_session", "switch_session", "fork", "navigate_tree", "import_jsonl", "export_html", "export_jsonl", "set_session_name", "get_rlm_max_depth_status", "set_rlm_max_depth", "rename_saved_session", "delete_saved_session", "get_session_context", "get_session_tree", "get_user_messages_for_forking", "get_last_assistant_text", "get_system_prompt", "get_tool_definition", "set_session_entry_label", "extension_ui_response", "prepare_update_restart", "retry_worker", "restart", "shutdown",]);
const DAEMON_CLIENT_CAPABILITY_SET: ReadonlySet<string> = new Set(DAEMON_SUPPORTED_CLIENT_CAPABILITIES);const CLIENT_CATCHUP_RETRY_MS = 250;/** * Upper bound on frames deferred per client and session while a snapshot * stream is active. Deferral spans one stream (seconds), so overflow means a * pathologically slow client; the sender then falls back to a catch-up * snapshot instead of buffering without limit. */const MAX_DEFERRED_SESSION_FRAMES = 256;const MAX_DEFERRED_SESSION_BYTES = 8 * 1024 * 1024;const UPDATE_RESTART_ABORT_BASH_TIMEOUT_MS = 5000;const SUPERVISOR_FENCE_POLL_MS = 250;const UPDATE_RESTART_MARKER = "<prime_agent_update_interrupted>\n" + "Prime Agent was updated and intentionally interrupted this session. Continue from the saved transcript and restored tool/kernel state. Any running model, tool, bash, or child-agent work may have been partially completed.\n" + "</prime_agent_update_interrupted>";const RECOVERY_CHECKPOINT_EVENTS: ReadonlySet<string> = new Set([ "agent_start", "agent_end", "turn_start", "turn_end", "message_start", "message_end", "tool_execution_start", "tool_execution_end", "compaction_start", "compaction_end", "auto_retry_start", "auto_retry_end", "bash_start", "bash_end", "session_action_update", "rlm_child_update",]);
function delay(ms: number): Promise<void> { return new Promise((resolveDelay) => setTimeout(resolveDelay, ms));}
type RuntimeOpenGuard = () => boolean | Promise<boolean>;type SupervisorGenerationClaim = Omit<Extract<DaemonWorkerCommand, { type: "worker_auth" }>, "id" | "type" | "token">;
interface BoundSupervisorGenerationClaim { claim: SupervisorGenerationClaim; ownerFingerprint: string;}
const PEER_GRANT_TTL_LIMIT_MS = 30_000;// Only the supervisor registers grants, so the cap is a tripwire, never an eviction policy.const PEER_GRANT_LIMIT = 1024;
const RLM_SUBAGENT_REGISTRY_FILE = "rlm-subagents.jsonl";
/** * One passive child as the daemon presents it: topology (sessionFile, parent, * depth, name) from the spawn ledger; hydration metadata (prompt, spawnCode, * model, rlmMaxDepth, status, createdAt) from the per-child display file, or * the legacy registry for pre-ledger children without one. */interface PassiveRlmSubagentEntry { childId: string; sessionName: string; sessionDir: string; sessionFile: string; parentSessionId: string; parentSessionFile?: string; rlmDepth?: number; rlmMaxDepth?: number; rlmParentNodeId?: string; prompt?: string; spawnCode?: string; model?: { provider: string; modelId: string }; status: "running" | "completed" | "deleted"; createdAt: number;}
/** Spread-ready optional metadata fields shared by display files and legacy registry entries. */function rlmSubagentMetadataFields(source: { rlmMaxDepth?: number; rlmParentNodeId?: string; prompt?: string; spawnCode?: string; model?: { provider: string; modelId: string };}): Pick<PassiveRlmSubagentEntry, "rlmMaxDepth" | "rlmParentNodeId" | "prompt" | "spawnCode" | "model"> { return { ...(source.rlmMaxDepth !== undefined ? { rlmMaxDepth: source.rlmMaxDepth } : {}), ...(source.rlmParentNodeId ? { rlmParentNodeId: source.rlmParentNodeId } : {}), ...(source.prompt ? { prompt: source.prompt } : {}), ...(source.spawnCode ? { spawnCode: source.spawnCode } : {}), ...(source.model ? { model: source.model } : {}), };}
type PassiveRlmRoot = | { rootParentState: ActiveSessionState; rootInfo?: never } | { rootParentState?: never; rootInfo: SessionInfo };
type PassiveRlmSubagent = PassiveRlmRoot & { entry: PassiveRlmSubagentEntry; info: SessionInfo; chain: PassiveRlmSubagentEntry[];};
class RuntimeOpenCancelledError extends Error {}class BoundSessionUnavailableError extends Error {}
export async function runDaemonMode(options: DaemonModeOptions): Promise<never> { const socketPath = normalizeSocketPath(options.socketPath ?? defaultDaemonSocketPath()); const daemon = new AgentDaemon(socketPath, options); await daemon.start(); return new Promise(() => {});}
export class AgentDaemon { private server?: Server; private shuttingDown = false; private readonly updateRestartQueuePauses = new Map<string, { release(): void }>(); private readonly sessionInputPauses = new Map< string, { activeSessionId: string; owner: DaemonSocketClient; leaseKey: string; pause: { release(): void } } >(); private readonly acpMcpOwners = new Map< string, { client: DaemonSocketClient; ownerId: string; serverNames: string[]; release?: Promise<void> } >(); private readonly mutationDrain = new MutationDrainLatch(); private updateRestart?: { id: symbol; owner?: DaemonSocketClient; abort: AbortController; deadline?: ReturnType<typeof setTimeout>; phase: "preparing" | "fencing" | "prepared" | "publishing"; manifest?: DaemonUpdateRestartManifest; deferredClientEnv: Array<{ client: DaemonSocketClient; state: ActiveSessionState; env: Record<string, string>; }>; }; private ownsSocketPath = false; private socketIdentity?: DaemonSocketIdentity; private readonly inFlightTracker = createInFlightTracker(); private readonly activeToolFinishes = new Map<string, () => void>(); private readonly clients = new Set<DaemonSocketClient>(); private readonly sessions = new Map<string, ActiveSessionState>(); private readonly openingSessions = new Map<string, Promise<ActiveSessionState>>(); /** Covers path resolution through publication in openingSessions, before the runtime promise exists. */ private readonly reservingSessionOpens = new Map<string, Promise<void>>(); private readonly bindingCompletions = new Map<string, Promise<void>>(); /** * Resolved-session-file keyed passivations let wake paths join after closeSessionOnce * removes the live session. This intentionally complements closingSessions: that map * coalesces every close by transient activeSessionId, while this one identifies the * passivation-only close reason by the durable identity needed by hydration/opening. */ private readonly passivatingSessions = new Map<string, Promise<void>>(); private readonly closingSessions = new Map< string, { promise: Promise<void>; reason: DaemonSessionClosedReason; descendants: Set<ActiveSessionState>; reasonUpgrade?: Promise<void>; } >(); private readonly sideQuestionRuns = new Map< string, { run: SideQuestionRun; client: DaemonSocketClient; activeSessionId: string; } >(); /** Live prompt admissions, keyed by session and caller-generated admission id. */ private readonly promptAdmissions = new Map< string, { activeSessionId: string; admissionId: string; controller?: AbortController; status: "waiting" | "owned" | "cancelled"; } >(); private readonly signalCleanupHandlers: Array<() => void> = []; private readonly cronStore: AgentCronJobStore; private readonly agentDir: string; private readonly cronScheduler: AgentCronScheduler; private readonly agentMessageRateLimiter = new AgentSessionMessageRateLimiter(); // Sessions inserted into `sessions` but still awaiting extension binding; // visible to host controllers during bind, excluded from targeting. private readonly bindingSessions = new Set<string>(); private readonly pendingSessionNames = new Set<string>(); private restoreActiveSessionId: string | undefined; private supervisorMonitorTimer?: ReturnType<typeof setTimeout>; /** * Settles when the armed supervisor availability check has run to completion * (including the reschedule it performs) or has been cancelled. The check is * a real async chain started from a timer callback, so this is the only way * an observer can tell that the monitor reached a steady state; the daemon * supervisor monitor tests await it instead of polling a timer loop. */ supervisorAvailabilityCheckSettled?: Promise<void>; private settleArmedSupervisorAvailabilityCheck?: () => void; private supervisorFenceTimer?: ReturnType<typeof setTimeout>; private supervisorLaunchInProgress = false; private supervisorAbsentSince?: number; private readonly supervisorClaims = new Map<DaemonSocketClient, BoundSupervisorGenerationClaim>(); private readonly peerGrants = new Map<string, DaemonWorkerPeerGrant>(); private readonly peerClaims = new Map<DaemonSocketClient, DaemonWorkerPeerGrant>(); private peerAdmissionsFenced = false; private agentMessagesPaused = false; private readonly summarizer = new DaemonSessionSummarizer( () => [...this.sessions.values()], (state) => { // Subagents share their recap with the parent so it shows in the parent's // subagent tree; their own session channel usually has no attached client. if (state.runtime.metadata.kind === "subagent") { state.runtime.session.setCurrentRecap(state.summaryState?.summary); } this.broadcastToSession(state, { type: "session_status", activeSessionId: state.activeSessionId, recap: this.sessionRecapEnabled() ? state.summaryState?.summary : undefined, }); }, undefined, () => this.sessionRecapEnabled(), ); private readonly recoveryJournal?: WorkerRecoveryJournal; private readonly rosterReporter: WorkerRosterReporterState = { lastComposed: new Map(), lastComposedJson: new Map(), queuedChildren: new Map(), removedAgentIds: new Map(), snapshotPending: false, }; private rosterFlushScheduled = false; private rosterHeartbeatTimer?: ReturnType<typeof setInterval>; private rlmSpawnLedgerInstance?: RlmSpawnLedger; /** In-flight admission spawn appends, awaited (and consumed) by createRlmSubagentRuntime. */ private readonly pendingRlmSpawnAppends = new Map<string, Promise<void>>(); private sessionRecapEnabledValue?: boolean;
constructor( private readonly socketPath: string, private readonly options: DaemonModeOptions, ) { if (!options.defaultSessionConfig.agentDir) { throw new Error("Daemon config is missing agentDir"); } this.agentDir = options.defaultSessionConfig.agentDir; // Hosted extensions get ctx.ui.theme; init it headlessly (no TTY, watcher off) or their first access kills the worker. initTheme( SettingsManager.create(options.defaultSessionConfig.cwd ?? process.cwd(), this.agentDir).getTheme(), false, ); this.cronStore = options.worker ? AgentCronJobStore.forSessionArtifacts() : new AgentCronJobStore(getCronJobsPath(this.agentDir)); this.restoreActiveSessionId = options.worker?.restoreActiveSessionId; const recoveryJournalPath = process.env[DAEMON_WORKER_RECOVERY_JOURNAL_ENV]; if (options.worker && recoveryJournalPath) { this.recoveryJournal = new WorkerRecoveryJournal(recoveryJournalPath); } this.cronScheduler = new AgentCronScheduler(this.cronStore, { runJob: (job) => this.runCronJob(job), beginDispatch: () => { this.mutationDrain.begin(); return () => this.mutationDrain.end(); }, onError: (job, error) => { this.log(`Cron job ${job.id} failed: ${error instanceof Error ? error.message : String(error)}`); }, }); this.cronStore.onHeartbeatChange(() => { this.broadcastGlobal({ type: "heartbeats_changed" }); this.scheduleRosterFlush(); }); }
/** * The daemon's session recap ("Recap: ..." above the prompt) is a per-user * preference: `sessionRecap.enabled: false` in global settings, or * `PRIME_AGENT_DISABLE_SESSION_RECAP=1` for a launcher. Resolved once. */ private sessionRecapEnabled(): boolean { if (this.sessionRecapEnabledValue === undefined) { const off = process.env.PRIME_AGENT_DISABLE_SESSION_RECAP; const disabledByEnv = typeof off === "string" && ["1", "true", "yes", "on"].includes(off.toLowerCase()); this.sessionRecapEnabledValue = disabledByEnv ? false : SettingsManager.create( this.options.defaultSessionConfig.cwd ?? process.cwd(), this.agentDir, ).getSessionRecapEnabled(); } return this.sessionRecapEnabledValue; }
// The daemon runs detached with no terminal, so route its diagnostics to its // rotating log file and the shared structured log (and stderr too, for when // it's run in the foreground). private log(message: string): void { console.error(message); structuredLog.warn(message, { socketPath: this.socketPath }); appendRotatingLog(getDaemonLogPath(this.socketPath), `[${new Date().toISOString()}] ${message}`); }
async start(): Promise<void> { if (this.options.worker) { process.stderr.on("error", () => { // A detached worker must survive the supervisor closing its diagnostic pipe. }); installWorkerExitHandlers({ sink: { log: (message) => this.log(message), exit: (code) => process.exit(code) }, tracker: this.inFlightTracker, onSignal: (signal, code) => { this.log(`received ${signal}; shutting down`); killTrackedDetachedChildren(); void this.shutdown(code); }, }); } else { installDaemonCrashHandlers({ log: (message) => this.log(message), exit: (code) => process.exit(code) }); } await prepareDaemonSocketPath(this.socketPath);
this.server = createServer((socket) => this.handleConnection(socket));
try { await new Promise<void>((resolveListen, rejectListen) => { const onError = (error: Error) => { this.server?.off("listening", onListening); rejectListen(error); }; const onListening = () => { this.server?.off("error", onError); try { this.socketIdentity = getDaemonSocketIdentity(this.socketPath); this.ownsSocketPath = true; if (process.platform !== "win32") { restrictDaemonSocketPath(this.socketPath); } } catch (error) { this.server?.close(); rejectListen(error); return; } resolveListen(); }; this.server?.once("error", onError); this.server?.once("listening", onListening); this.server?.listen(this.socketPath); }); } catch (error) { this.cleanupSocketPath(); throw error; }
this.registerSignalHandlers(); this.summarizer.start(); this.log(`Prime Agent daemon listening on ${this.socketPath}`); // No startup restore: on-disk sessions return only via --resume or the agents view. if (!this.shuttingDown) { this.cronScheduler.start(); } if (this.options.worker) { this.rosterHeartbeatTimer = setInterval( () => this.broadcastRosterFrame({ type: "roster_heartbeat" }), ROSTER_HEARTBEAT_INTERVAL_MS, ); this.rosterHeartbeatTimer.unref(); } this.startSupervisorMonitor(); }
private supervisorSocketPathFromEnv(): string | undefined { const raw = process.env[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV]; return raw ? normalizeSocketPath(raw) : undefined; }
/** Persistent supervisor connection for cross-worker requests; one link per daemon process. */ private supervisorLinkInstance?: SupervisorLink; private supervisorLink(): SupervisorLink | undefined { const supervisorSocketPath = this.supervisorSocketPathFromEnv(); if (!supervisorSocketPath) return undefined; this.supervisorLinkInstance ??= new SupervisorLink({ socketPath: supervisorSocketPath }); return this.supervisorLinkInstance; }
private startSupervisorMonitor(): void { const supervisorSocketPath = this.supervisorSocketPathFromEnv(); if (!this.options.worker || !supervisorSocketPath) { return; } this.scheduleSupervisorAvailabilityCheck(supervisorSocketPath, 1500); }
private scheduleSupervisorAvailabilityCheck(supervisorSocketPath: string, delayMs: number): void { if (this.shuttingDown || this.hasAuthenticatedSupervisorConnection()) { return; } this.cancelArmedSupervisorAvailabilityCheck(); let settle: () => void = () => undefined; this.supervisorAvailabilityCheckSettled = new Promise<void>((resolveSettled) => { settle = resolveSettled; }); this.settleArmedSupervisorAvailabilityCheck = settle; this.supervisorMonitorTimer = setTimeout(() => { this.supervisorMonitorTimer = undefined; // The check owns its settle from here on, so a reschedule started // inside it installs a fresh one instead of resolving this one early. this.settleArmedSupervisorAvailabilityCheck = undefined; void this.checkSupervisorAvailability(supervisorSocketPath) .catch(() => { if (!this.shuttingDown && !this.hasAuthenticatedSupervisorConnection()) { this.scheduleSupervisorAvailabilityCheck(supervisorSocketPath, 5000); } }) .finally(settle); }, delayMs); }
/** Disarms a pending check and settles it, since it will never run. */ private cancelArmedSupervisorAvailabilityCheck(): void { if (this.supervisorMonitorTimer) { clearTimeout(this.supervisorMonitorTimer); this.supervisorMonitorTimer = undefined; } const settle = this.settleArmedSupervisorAvailabilityCheck; this.settleArmedSupervisorAvailabilityCheck = undefined; settle?.(); }
private async checkSupervisorAvailability(supervisorSocketPath: string): Promise<void> { if (this.shuttingDown || this.hasAuthenticatedSupervisorConnection()) { this.supervisorAbsentSince = undefined; return; } if (await isDaemonShutdownAdmissionActive()) { this.supervisorAbsentSince = undefined; this.scheduleSupervisorAvailabilityCheck(supervisorSocketPath, 5000); return; } if (await this.canConnectToSupervisor(supervisorSocketPath)) { this.supervisorAbsentSince = undefined; return; } // The supervisor socket is unreachable; remember when the worker last // saw it so the orphan window below stays bounded. this.supervisorAbsentSince ??= Date.now(); await this.launchReplacementSupervisor(supervisorSocketPath); if (await this.canConnectToSupervisor(supervisorSocketPath)) { // A replacement came up during the launch: the orphan window must // restart instead of exiting the worker. The monitor must stay // armed, though — a replacement can bind and then exit before it // ever claims the worker, and only an authenticated claim (or its // later close) re-arms the monitor. Falling through reschedules // the next availability check below. this.supervisorAbsentSince = undefined; } await this.exitIfSupervisorOrphanedForTooLong(supervisorSocketPath); if (!this.shuttingDown && !this.hasAuthenticatedSupervisorConnection()) { this.scheduleSupervisorAvailabilityCheck(supervisorSocketPath, 5000); } }
/** * Garbage-collect orphaned workers. If the supervisor has been unreachable * for the whole bounded window — no supervisor claim, no successful * replacement launch — the worker exits: retrying the resurrection loop * forever only accumulates garbage. Sessions persist on disk, so the next * supervisor to claim the socket spawns fresh workers on demand. */ private async exitIfSupervisorOrphanedForTooLong(supervisorSocketPath: string): Promise<void> { const absentSince = this.supervisorAbsentSince; if (absentSince === undefined || Date.now() - absentSince < workerSupervisorLostExitMs()) { return; } if (this.hasAuthenticatedSupervisorConnection()) { this.supervisorAbsentSince = undefined; return; } if (this.hasOngoingSessionWork()) { // An active run owns the worker a little longer; its turn end // lets the next availability check reconsider. return; } this.log( `supervisor ${supervisorSocketPath} unreachable for ${Math.round((Date.now() - absentSince) / 1000)}s; exiting orphaned worker`, ); await this.shutdown(0); }
private hasOngoingSessionWork(): boolean { for (const state of this.sessions.values()) { // Same predicate the daemon reports as isSessionActive: retrying, // bash/kernel background work, refinement, compaction settlement, // and queued actions must all keep holding the worker. if (state.runtime.session.isSessionActive) { return true; } } return false; }
private hasAuthenticatedSupervisorConnection(): boolean { return this.supervisorClaims.size > 0; }
private revokeSupervisorClaim(client: DaemonSocketClient, expected?: BoundSupervisorGenerationClaim): boolean { if (expected && this.supervisorClaims.get(client) !== expected) return false; if (!this.supervisorClaims.delete(client)) return false; if (this.options.worker && this.updateRestart?.owner === client) { this.cancelPreparedUpdateRestart(this.updateRestart.id); } return true; }
/** Burn every outstanding grant and end direct peers; admissions stay fenced until an update cancel. */ private fencePeerTransports(closingReason?: DaemonClosingReason): void { this.peerAdmissionsFenced = true; this.peerGrants.clear(); for (const client of [...this.peerClaims.keys()]) { this.peerClaims.delete(client); // The reason reaches the client before FIN so its close maps to shutdown/update, not a transport loss. if (closingReason) this.write(client, { type: "daemon_closing", reason: closingReason }); client.socket.end(); } }
private clearSupervisorAvailabilityCheck(): void { this.cancelArmedSupervisorAvailabilityCheck(); if (this.supervisorFenceTimer) { clearTimeout(this.supervisorFenceTimer); this.supervisorFenceTimer = undefined; } }
private scheduleSupervisorFenceCheck(): void { if (this.shuttingDown || this.supervisorFenceTimer || this.supervisorClaims.size === 0) { return; } this.supervisorFenceTimer = setTimeout(() => { this.supervisorFenceTimer = undefined; void this.checkSupervisorFences(); }, SUPERVISOR_FENCE_POLL_MS); }
private async checkSupervisorFences(): Promise<void> { for (const [client, boundClaim] of this.supervisorClaims) { try { boundClaim.ownerFingerprint = await this.assertSupervisorClaimCurrent( boundClaim.claim, boundClaim.ownerFingerprint, ); } catch { if (this.revokeSupervisorClaim(client, boundClaim)) client.socket.end(); } } this.scheduleSupervisorFenceCheck(); }
private assertSupervisorClaimCurrent( claim: SupervisorGenerationClaim, validatedFingerprint?: string, ): Promise<string> { return assertDaemonSupervisorOwnerCurrent( { generation: claim.supervisorGeneration, pid: claim.supervisorPid, ...(claim.supervisorProcessStartId ? { processStartId: claim.supervisorProcessStartId } : {}), socketPath: claim.supervisorSocketPath, }, validatedFingerprint, ); }
private canConnectToSupervisor(socketPath: string): Promise<boolean> { return new Promise((resolveConnect) => { const socket = createConnection(socketPath); let settled = false; const finish = (connected: boolean) => { if (settled) { return; } settled = true; clearTimeout(timeout); socket.removeAllListeners(); socket.destroy(); resolveConnect(connected); }; const timeout = setTimeout(() => finish(false), 250); socket.once("connect", () => finish(true)); socket.once("error", () => finish(false)); }); }
private async launchReplacementSupervisor(supervisorSocketPath: string): Promise<void> { if (this.supervisorLaunchInProgress || this.shuttingDown) { return; } this.supervisorLaunchInProgress = true; const key = createHash("sha256").update(supervisorSocketPath).digest("hex").slice(0, 12); const lockDirectory = join(dirname(supervisorSocketPath), `.supervisor-launch-${key}.lock`); let ownsLock = false; try { for (let attempt = 0; attempt < 3 && !ownsLock; attempt++) { const result = await tryAcquireDirLock(lockDirectory, (ownerPid) => ownerPid !== undefined ? isProcessAlive(ownerPid) : false, ); if (result === "held") { return; } ownsLock = result === "acquired"; } if (!ownsLock) { return; } if (await this.canConnectToSupervisor(supervisorSocketPath)) { return; } if (await isDaemonShutdownAdmissionActive()) { return; } const launch = createCliSubprocessLaunchSpec(["--mode", "daemon", "--daemon-socket", supervisorSocketPath]); const environment = createCliSubprocessEnv(); delete environment[DAEMON_WORKER_ROLE_ENV]; delete environment[DAEMON_WORKER_TOKEN_ENV]; delete environment[DAEMON_WORKER_ACTIVE_SESSION_ID_ENV]; delete environment[DAEMON_WORKER_RECOVERY_JOURNAL_ENV]; delete environment[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV]; delete environment[ORPHAN_PROCESS_JOURNAL_ENV]; delete environment[SESSION_LEASES_ENABLED_ENV]; delete environment[SESSION_LEASE_OWNER_ID_ENV]; const child = spawnHidden(launch.command, launch.args, { cwd: this.options.defaultSessionConfig.cwd ?? process.cwd(), detached: true, env: environment, stdio: "ignore", }); const childExited = waitForChildProcess(child); void childExited.catch(() => undefined); child.unref(); const deadline = Date.now() + 10_000; while (!this.shuttingDown && Date.now() < deadline) { if (child.pid === undefined || child.exitCode !== null || child.signalCode !== null) { await childExited; return; } for (const [client, boundClaim] of this.supervisorClaims) { const { claim } = boundClaim; if ( claim.supervisorSocketPath !== supervisorSocketPath || !client.authenticated || client.socket.destroyed ) continue; try { await this.assertSupervisorClaimCurrent(claim); } catch { continue; } if ( this.shuttingDown || Date.now() >= deadline || this.supervisorClaims.get(client) !== boundClaim || !client.authenticated || client.socket.destroyed ) continue; if (claim.supervisorPid === child.pid) { this.log(`launched replacement supervisor on ${supervisorSocketPath}`); return; } if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); await waitForPromptAdmission(childExited, AbortSignal.timeout(Math.max(1, deadline - Date.now()))); this.log(`stopped losing replacement supervisor ${child.pid} on ${supervisorSocketPath}`); return; } await delay(50); } this.log(`replacement supervisor ${child.pid} left running without a current authenticated supervisor`); } catch (error) { this.log(`failed to launch replacement supervisor: ${String(error)}`); } finally { if (ownsLock) { rmSync(lockDirectory, { recursive: true, force: true }); } this.supervisorLaunchInProgress = false; } }
private cleanupSocketPath(): void { if (!this.ownsSocketPath) { return; } this.ownsSocketPath = false; const socketIdentity = this.socketIdentity; this.socketIdentity = undefined; cleanupDaemonSocketPath(this.socketPath, socketIdentity); }
/** * Adopt env for a session that has none, propagating to subagents spawned * before adoption (their exec-env providers read state.clientEnv live). * Never overwrites an existing identity. */ private adoptClientEnv(state: ActiveSessionState, env?: Record<string, string>): void { if (!env || state.clientEnv) { return; } state.clientEnv = env; for (const child of this.sessions.values()) { const metadata = child.runtime.metadata; if (metadata.kind === "subagent" && metadata.parentActiveSessionId === state.activeSessionId) { this.adoptClientEnv(child, env); } } }
/** Root sessions dir that keys this daemon's spawn ledger. */ private rlmLedgerSessionsDir(): string { return this.options.defaultSessionConfig.sessionDir ?? getSessionsDir(this.agentDir); }
/** * Supervisor-owned spawn ledger for this daemon's sessions dir. Seeded * lazily from the existing per-parent registries via the same tolerant * reader the daemon already uses for passive hydration. */ private rlmSpawnLedger(): RlmSpawnLedger { this.rlmSpawnLedgerInstance ??= new RlmSpawnLedger( this.agentDir, this.rlmLedgerSessionsDir(), createRlmLedgerRegistrySeedSource(), (message) => this.log(message), ); return this.rlmSpawnLedgerInstance; }
// Ledgers are per sessions-dir family: a catalog request for another dir must read that dir's ledger. private rlmSpawnLedgerFor(sessionDir: string | undefined): RlmSpawnLedger { if (sessionDir === undefined || resolve(sessionDir) === resolve(this.rlmLedgerSessionsDir())) { return this.rlmSpawnLedger(); } return new RlmSpawnLedger(this.agentDir, sessionDir, createRlmLedgerRegistrySeedSource(), (message) => this.log(message), ); }
private async appendRlmLedgerRenameForState(state: ActiveSessionState, name: string): Promise<void> { const childId = state.runtime.metadata.rlmChildId; const child = state.runtime.session.sessionFile; if (!childId || !child) return; // Awaited: the supervisor answers sibling-name checks from the ledger, // so the rename must be durable before the reservation is released. await this.rlmSpawnLedger() .appendRename({ childId, child, name }) .catch((error) => { this.log(`failed to append RLM ledger rename: ${error instanceof Error ? error.message : String(error)}`); }); }
/** * Legacy per-parent registry path. Read-only: consumed solely as fallback * hydration metadata for pre-ledger children without a display file (the * ledger seed source has its own equivalent reader). */ private legacyRlmSubagentRegistryPath(parentSessionFile: string, parentSessionId: string): string { return join(getSessionArtifactPathForFile(parentSessionFile, parentSessionId), RLM_SUBAGENT_REGISTRY_FILE); }
private readLegacyRlmSubagentRegistry( path: string, throwOnReadError = false, onReadError?: () => void, ): Promise<LegacyRlmSubagentRegistryEntry[]> { return readLegacyRlmSubagentRegistryFile(path, { throwOnReadError, onReadError, log: (message) => this.log(message), }); }
/** * Record a spawned or completed child: topology (the spawn edge) goes to * the daemon-owned ledger at admission; hydration/display metadata goes to * the child's per-child display file at both moments. */ private recordRlmSubagentState( parentState: ActiveSessionState, input: { childId: string; sessionName: string; sessionDir: string; sessionFile: string; rlmDepth: number; rlmMaxDepth: number; rlmParentNodeId?: string; prompt?: string; spawnCode?: string; model?: { provider: string; modelId: string }; status: "running" | "completed"; createdAt?: number; }, ): boolean { const parentSession = parentState.runtime.session; // Spawn admission is the moment the daemon knows the edge firsthand. // The ledger is the only topology store, so the append's outcome is // load-bearing: the promise is stashed per childId for the admission // path to await — admission fails if the spawn record cannot be made // durable (a swallowed failure would admit a child that listing, // hydration, and a2a wake can never find after passivation). if (input.status === "running" && parentSession.sessionFile) { const spawnAppend = this.rlmSpawnLedger().appendSpawn({ childId: input.childId, parent: parentSession.sessionFile, child: input.sessionFile, depth: input.rlmDepth, name: input.sessionName, }); // Mark handled so an early rejection cannot surface as an // unhandled-rejection crash before the admission path awaits it. spawnAppend.catch(() => undefined); // Child ids are only unique per parent; the parent scopes the pending-append key. this.pendingRlmSpawnAppends.set(`${parentState.activeSessionId}#${input.childId}`, spawnAppend); } try { const written = writeRlmSubagentDisplayEntry({ type: "rlm_subagent", childId: input.childId, sessionName: input.sessionName, sessionDir: input.sessionDir, sessionFile: input.sessionFile, ...rlmSubagentMetadataFields(input), status: input.status, createdAt: input.createdAt ?? Date.now(), updatedAt: new Date().toISOString(), }); if (!written) { this.log(`skipped RLM subagent display entry for ${input.childId}: deleted tombstone exists`); } return written; } catch (error) { this.log( `failed to persist RLM subagent display entry: ${error instanceof Error ? error.message : String(error)}`, ); return false; } }
private async recordRlmSubagentDeletion( parentState: ActiveSessionState, childId: string, reason: RlmLedgerDeleteReason = "user", ): Promise<void> { const parentFile = parentState.runtime.session.sessionFile; if (!parentFile) { return; } const parentPath = canonicalSessionPath(parentFile); const edges = (await this.rlmSpawnLedger().edges(true)).filter( (candidate) => candidate.childId === childId && canonicalSessionPath(candidate.parent) === parentPath, ); const edge = edges.find((candidate) => !candidate.deleted); let entry: PassiveRlmSubagentEntry | undefined; if (edge) { entry = await this.passiveRlmSubagentEntryForEdge(edge, { sessionId: parentState.runtime.session.sessionId, sessionFile: parentFile, }); } else if (edges.length > 0) { // Only tombstoned edges: the tombstones are already durable, nothing // to re-append. A prior deletion may have crashed before its artifact // sweep. Restore the display tombstone before sweeping artifacts. for (const tombstoned of edges) { try { const currentDisplay = await readRlmSubagentDisplayEntry(dirname(tombstoned.child)); if (!currentDisplay || currentDisplay.status !== "deleted") { writeRlmSubagentDisplayEntry({ type: "rlm_subagent", childId, sessionName: currentDisplay?.sessionName ?? tombstoned.name, sessionDir: dirname(tombstoned.child), sessionFile: currentDisplay?.sessionFile ?? tombstoned.child, ...rlmSubagentMetadataFields(currentDisplay ?? {}), status: "deleted", createdAt: currentDisplay?.createdAt ?? 0, updatedAt: new Date().toISOString(), }); } } catch { // Best-effort: the ledger tombstone is the authority; the display // file is display-grade and the sweep below will remove artifacts. this.log(`failed to reconcile display entry for tombstoned RLM subagent ${childId}`); } await this.deleteRlmSubagentArtifacts(childId, tombstoned.child); } return; } else { // No edge at all. A pre-ledger child the seed missed may still exist // in the legacy registry; an unreadable registry means the durable // deletion boundary cannot be established, so the deletion fails. const legacy = ( await this.readLegacyRlmSubagentRegistry( this.legacyRlmSubagentRegistryPath(parentFile, parentState.runtime.session.sessionId), true, ) ).find((candidate) => candidate.childId === childId); if (!legacy || legacy.status === "deleted") { // The child never existed under this parent, or its tombstone is // already durable. return; } entry = { childId: legacy.childId, sessionName: legacy.sessionName, sessionDir: legacy.sessionDir, sessionFile: legacy.sessionFile, parentSessionId: parentState.runtime.session.sessionId, parentSessionFile: parentFile, ...(legacy.rlmDepth !== undefined ? { rlmDepth: legacy.rlmDepth } : {}), ...rlmSubagentMetadataFields(legacy), status: legacy.status, createdAt: legacy.createdAt, }; } // Display tombstone first ("deleted deliberately, transcript retained"): // a crash in between leaves a live ledger edge over a deleted display // entry, healed by retrying the deletion; the reverse order could // tombstone the ledger while the display file still claims the child // exists. try { writeRlmSubagentDisplayEntry({ type: "rlm_subagent", childId: entry.childId, sessionName: entry.sessionName, sessionDir: entry.sessionDir, sessionFile: entry.sessionFile, ...rlmSubagentMetadataFields(entry), status: "deleted", createdAt: entry.createdAt, updatedAt: new Date().toISOString(), }); } catch (error) { throw new Error( `Failed to persist deletion for RLM subagent ${childId}: ${error instanceof Error ? error.message : String(error)}`, ); } // The ledger delete record is the topology tombstone; unlike the // dual-write era it has no other writer to fall back on, so a failed // append is a failed deletion. await this.rlmSpawnLedger().appendDelete({ childId, child: entry.sessionFile, reason }); if (this.options.worker) { this.rosterReporter.removedAgentIds.set( this.rosterAgentIdForRlmChild(childId, entry.parentSessionFile), basename(entry.sessionFile, ".jsonl"), ); this.scheduleRosterFlush(); } // Deletion boundary: transcript + display tombstone are the durable // record and stay; the nested artifact dir is a runtime cache and goes. await this.deleteRlmSubagentArtifacts(childId, entry.sessionFile); }
/** Best-effort artifact-dir removal: cache cleanup must never fail a deletion. */ private async deleteRlmSubagentArtifacts(childId: string, childSessionFile: string): Promise<void> { try { await deleteSessionArtifacts(childSessionFile); } catch (error) { this.log( `failed to remove artifact dir for deleted RLM subagent ${childId}: ${error instanceof Error ? error.message : String(error)}`, ); } }
/** * Hydration/display metadata for one live ledger edge: per-child display * file first, then the legacy registry for pre-ledger children, then * edge-only defaults. Topology (childId, paths, depth, name) always comes * from the ledger edge. */ private async passiveRlmSubagentEntryForEdge( edge: RlmLedgerEdge, parent: { sessionId: string; sessionFile: string }, legacyRegistryCache?: Map<string, Promise<LegacyRlmSubagentRegistryEntry[]>>, onReadError?: (path: string) => void, ): Promise<PassiveRlmSubagentEntry> { const edgeChild = canonicalSessionPath(edge.child); const base = { childId: edge.childId, sessionName: edge.name, sessionDir: dirname(edge.child), sessionFile: edge.child, parentSessionId: parent.sessionId, parentSessionFile: parent.sessionFile, }; // The ledger stores realpath-canonical paths, the rest of the daemon // keys maps by resolve(): present the paths the writer recorded (the // metadata file is validated to describe this same child) so passive // rows keep matching residency, opens, and passivation bookkeeping. const metadataFields = (source: { sessionDir: string; sessionFile: string; rlmMaxDepth?: number; rlmParentNodeId?: string; prompt?: string; spawnCode?: string; model?: { provider: string; modelId: string }; status: "running" | "completed" | "deleted"; createdAt: number; }) => ({ ...base, ...(canonicalSessionPath(source.sessionFile) === edgeChild ? { sessionDir: source.sessionDir, sessionFile: source.sessionFile } : {}), ...rlmSubagentMetadataFields(source), status: source.status, createdAt: source.createdAt, }); const display = await readRlmSubagentDisplayEntry(dirname(edge.child), () => onReadError?.(rlmSubagentDisplayPath(dirname(edge.child))), ); if (display && display.childId === edge.childId) { // A display-file child was ledger-spawned: the edge depth is real. return { ...metadataFields(display), rlmDepth: edge.depth }; } const registryPath = this.legacyRlmSubagentRegistryPath(parent.sessionFile, parent.sessionId); let registryRead = legacyRegistryCache?.get(registryPath); if (!registryRead) { registryRead = this.readLegacyRlmSubagentRegistry(registryPath, false, () => onReadError?.(registryPath)); legacyRegistryCache?.set(registryPath, registryRead); } const legacy = (await registryRead).find((entry) => entry.childId === edge.childId); if (legacy) { // A seeded edge's depth may be a parent+1 guess for legacy entries // without one: leave it absent so hydration falls back to the // persisted header depth, exactly as the registry reader did. return { ...metadataFields(legacy), ...(legacy.rlmDepth !== undefined ? { rlmDepth: legacy.rlmDepth } : {}) }; } // Ledger-only child (metadata lost): hydratable with defaults. let createdAt = 0; try { createdAt = (await stat(edge.child)).birthtimeMs || 0; } catch { // The stat is display-grade; a failed read keeps the epoch default. } return { ...base, rlmDepth: edge.depth, status: "completed", createdAt }; }
// One memoized passive-topology derivation per argument shape; all daemon // consumers read the cached walk. A hit requires the ledger stat, roster, // live roots, root-state identities, and every visited file input's stat // (captured before its read, so mid-walk writes cost one extra re-walk, // never a stale memo) to be unchanged. Same-shape walks run one at a time; // a caller never joins an earlier walk. private readonly passiveRlmSubagentWalks = new Map<string, Promise<PassiveRlmSubagent[]>>(); private readonly passiveRlmSubagentMemo = new Map< string, { fingerprint: string; inputStats: Map<string, string>; result: PassiveRlmSubagent[] } >(); private static readonly PASSIVE_RLM_MEMO_MAX_KEYS = 4;
private hasPersistedResidentSession(): boolean { for (const state of this.sessions.values()) { if (state.runtime.session.sessionFile) return true; } return false; }
private async passiveRlmTopologyFingerprint(savedRootInfos: SessionInfo[]): Promise<string> { const ledgerStat = await this.passiveRlmStatString(this.rlmSpawnLedger().ledgerPath); const resident = [...this.sessions.values()] .map((state) => `${state.activeSessionId}:${state.runtime.session.sessionFile ?? ""}`) .sort(); const roots = savedRootInfos.map((info) => resolve(info.path)).sort(); return `${ledgerStat}|${resident.join(",")}|${roots.join(",")}`; }
private async passiveRlmStatString(path: string): Promise<string> { try { const stats = await stat(path); return `${stats.size}:${stats.mtimeMs}:${stats.ino}`; } catch { return "absent"; } }
private async passiveRlmInputStatsUnchanged(inputStats: Map<string, string>): Promise<boolean> { const checks = await Promise.all( [...inputStats].map(async ([path, statString]) => (await this.passiveRlmStatString(path)) === statString), ); return checks.every(Boolean); }
private passiveRlmRootsStillResident(result: PassiveRlmSubagent[]): boolean { return result.every( (passive) => !passive.rootParentState || this.sessions.get(passive.rootParentState.activeSessionId) === passive.rootParentState, ); }
private listPassiveRlmSubagents( savedRoots: SessionInfo[] = [], includeResident = false, ): Promise<PassiveRlmSubagent[]> { const savedRootInfos = savedRoots.filter((rootInfo) => inactiveLifecycleForSession(rootInfo) === "live"); if (savedRootInfos.length === 0 && !this.hasPersistedResidentSession()) { // Keep the empty topology IO-free: no roots means no walk, no ledger stat. return Promise.resolve([]); } const key = `${includeResident}|${savedRootInfos .map((info) => resolve(info.path)) .sort() .join(",")}`; const run = async (): Promise<PassiveRlmSubagent[]> => { const before = await this.passiveRlmTopologyFingerprint(savedRootInfos); const memo = this.passiveRlmSubagentMemo.get(key); if ( memo && memo.fingerprint === before && this.passiveRlmRootsStillResident(memo.result) && (await this.passiveRlmInputStatsUnchanged(memo.inputStats)) ) { return memo.result; } const walked = await this.walkPassiveRlmSubagents(savedRootInfos, includeResident); // Only a walk whose inputs held still qualifies as a memo: not the // ledger-seeding first walk, not a degraded one. const after = await this.passiveRlmTopologyFingerprint(savedRootInfos); if (after === before && !walked.degraded) { this.passiveRlmSubagentMemo.delete(key); this.passiveRlmSubagentMemo.set(key, { fingerprint: after, inputStats: walked.inputStats, result: walked.result, }); for (const staleKey of this.passiveRlmSubagentMemo.keys()) { if (this.passiveRlmSubagentMemo.size <= AgentDaemon.PASSIVE_RLM_MEMO_MAX_KEYS) break; this.passiveRlmSubagentMemo.delete(staleKey); } } else { this.passiveRlmSubagentMemo.delete(key); } return walked.result; }; const previous = this.passiveRlmSubagentWalks.get(key); const walk = previous ? previous.then(run, run) : run(); this.passiveRlmSubagentWalks.set(key, walk); // Not finally(): its discarded promise would turn a rejecting walk into a // daemon-crashing unhandled rejection. The caller still sees the rejection. const cleanup = () => { if (this.passiveRlmSubagentWalks.get(key) === walk) this.passiveRlmSubagentWalks.delete(key); }; walk.then(cleanup, cleanup); return walk; }
/** List each root's passive (non-resident) descendants from the ledger, without creating runtimes. */ private async walkPassiveRlmSubagents( savedRootInfos: SessionInfo[], includeResident: boolean, ): Promise<{ result: PassiveRlmSubagent[]; inputStats: Map<string, string>; degraded: boolean }> { // Captured before each read: the identity can only be older than the content. const inputStats = new Map<string, string>(); let degraded = false; const recordInputStat = async (path: string): Promise<string> => { const resolved = resolve(path); const existing = inputStats.get(resolved); if (existing !== undefined) return existing; const statString = await this.passiveRlmStatString(path); inputStats.set(resolved, statString); return statString; }; const residentRoots: Array<{ parentState: ActiveSessionState; sessionFile: string }> = []; for (const parentState of this.sessions.values()) { const parentFile = parentState.runtime.session.sessionFile; // An in-memory session cannot own persisted children. if (parentFile) residentRoots.push({ parentState, sessionFile: parentFile }); } if (residentRoots.length === 0 && savedRootInfos.length === 0) { return { result: [], inputStats, degraded }; } const edges = await this.rlmSpawnLedger().edges(); const childrenByParent = new Map<string, RlmLedgerEdge[]>(); for (const edge of edges) { const parentPath = canonicalSessionPath(edge.parent); const siblings = childrenByParent.get(parentPath) ?? []; siblings.push(edge); childrenByParent.set(parentPath, siblings); } const legacyRegistryCache = new Map<string, Promise<LegacyRlmSubagentRegistryEntry[]>>(); const passive: PassiveRlmSubagent[] = []; const visit = async ( root: PassiveRlmRoot, parent: { sessionId: string; sessionFile: string }, parentChain: PassiveRlmSubagentEntry[], visited: Set<string>, ): Promise<void> => { for (const edge of childrenByParent.get(canonicalSessionPath(parent.sessionFile)) ?? []) { await recordInputStat(rlmSubagentDisplayPath(dirname(edge.child))); await recordInputStat(this.legacyRlmSubagentRegistryPath(parent.sessionFile, parent.sessionId)); // The ledger stores realpath-canonical paths while the rest of the // daemon keys by resolve(): work with the writer-recorded path from // the metadata entry so passive rows keep matching residency, // opens, and passivation bookkeeping. const entry = await this.passiveRlmSubagentEntryForEdge(edge, parent, legacyRegistryCache, (path) => { // A present metadata file may recover without a stat change. if (inputStats.get(resolve(path)) !== "absent") degraded = true; }); const sessionKey = resolve(entry.sessionFile); if (entry.status === "deleted" || visited.has(sessionKey)) continue; visited.add(sessionKey); // A resident child walks as an outer root below; skipping before the // stat capture keeps its streamed transcript out of the input set. if (!includeResident && this.findSessionBySessionFile(entry.sessionFile)) continue; const childStat = await recordInputStat(entry.sessionFile); const info = await readSessionInfo(entry.sessionFile); if (!info) { // A present file that fails to list may recover without a stat change. if (childStat !== "absent") degraded = true; continue; } const chain = [...parentChain, entry]; passive.push({ ...root, entry, info, chain }); await visit(root, { sessionId: info.id, sessionFile: entry.sessionFile }, chain, visited); } }; const residentRootPaths = new Set<string>(); for (const { parentState, sessionFile } of residentRoots) { const parentPath = resolve(sessionFile); residentRootPaths.add(parentPath); await visit( { rootParentState: parentState }, { sessionId: parentState.runtime.session.sessionId, sessionFile }, [], new Set([parentPath]), ); } for (const rootInfo of savedRootInfos) { const rootPath = resolve(rootInfo.path); if (residentRootPaths.has(rootPath)) continue; await visit({ rootInfo }, { sessionId: rootInfo.id, sessionFile: rootInfo.path }, [], new Set([rootPath])); } return { result: passive, inputStats, degraded }; }
private async passiveRlmSubagentsByPath( savedRoots: SessionInfo[] = [], includeResident = false, ): Promise<Map<string, PassiveRlmSubagent>> { return new Map( (await this.listPassiveRlmSubagents(savedRoots, includeResident)).map((passive) => [ resolve(passive.entry.sessionFile), passive, ]), ); }
/** Add saved-only descendants discovered by the shared passive-registry walk. */ private async buildRlmChildSnapshotsWithPassiveRlmSubagents( rootState: ActiveSessionState, ): Promise<AgentConnectionRlmChildAgentSnapshot[]> { const snapshots = buildRlmChildSnapshots(rootState.activeSessionId, [...this.sessions.values()]); const residentParentIds = new Set([ rootState.activeSessionId, ...snapshots.flatMap((snapshot) => (snapshot.activeSessionId ? [snapshot.activeSessionId] : [])), ]); const seenChildIds = new Set(snapshots.map((snapshot) => snapshot.id)); for (const passive of await this.listPassiveRlmSubagents()) { if ( !passive.rootParentState || !residentParentIds.has(passive.rootParentState.activeSessionId) || seenChildIds.has(passive.entry.childId) ) { continue; } const parentEntry = passive.chain.at(-2); const parentId = parentEntry?.childId ?? passive.rootParentState.runtime.metadata.rlmChildId; snapshots.push({ id: passive.entry.childId, ...(parentId ? { parentId } : {}), sessionName: passive.info.name ?? passive.entry.sessionName, model: passive.entry.model ? `${passive.entry.model.provider}/${passive.entry.model.modelId}` : undefined, label: rlmChildLabel(passive.entry.prompt ?? ""), status: passive.entry.status === "completed" ? "done" : "error", sessionDir: passive.entry.sessionDir, }); seenChildIds.add(passive.entry.childId); } return snapshots; }
private async buildSessionListWithPassiveRlmSubagents( activeSessions: ActiveSessionState[], savedSessions: SessionInfo[], scheduledJobs: AgentCronJob[], ): Promise<SessionSummary[]> { const passiveByPath = await this.passiveRlmSubagentsByPath(savedSessions); const savedByPath = new Map(savedSessions.map((session) => [resolve(session.path), session])); for (const [path, passive] of passiveByPath) { savedByPath.set(path, passive.info); } return buildSessionList(activeSessions, [...savedByPath.values()], scheduledJobs).map((summary) => { const passive = summary.sessionFile ? passiveByPath.get(resolve(summary.sessionFile)) : undefined; if (!passive || summary.activeSessionId) return summary; const parentEntry = passive.chain.at(-2); return { ...summary, runtimeKind: "subagent", ...(passive.chain.length === 1 && passive.rootParentState ? { parentActiveSessionId: passive.rootParentState.activeSessionId } : {}), parentSessionId: passive.entry.parentSessionId, parentSessionPath: passive.entry.parentSessionFile ?? parentEntry?.sessionFile ?? passive.rootParentState?.runtime.session.sessionFile ?? passive.rootInfo?.path, rlmDepth: passive.entry.rlmDepth ?? passive.info.rlmDepth, rlmChildId: passive.entry.childId, rlmParentNodeId: passive.entry.rlmParentNodeId ?? passive.entry.childId, spawnCode: passive.entry.spawnCode, }; }); }
private async findPassiveRlmSubagent( target: string, includeResident = false, ): Promise<PassiveRlmSubagent | undefined> { const matches = [...(await this.passiveRlmSubagentsByPath([], includeResident)).values()].filter( ({ entry, info }) => entry.childId === target || resolve(entry.sessionFile) === resolve(target) || info.id === target || (info.name ?? entry.sessionName) === target, ); if (matches.length > 1) { throw new Error(`Session selector "${target}" is ambiguous`); } return matches[0]; }
private async addRuntime( runtime: AgentSessionRuntime, name?: string, clientEnv?: Record<string, string>, onStateCreated?: (state: ActiveSessionState) => void, runtimeOpenGuard?: RuntimeOpenGuard, onStateBound?: (state: ActiveSessionState) => void, restoreActiveSessionId?: string, ): Promise<ActiveSessionState> { const desiredActiveSessionId = runtime.metadata.kind === "top-level" ? this.restoreActiveSessionId : restoreActiveSessionId; if (runtime.metadata.kind === "top-level" && desiredActiveSessionId) { this.restoreActiveSessionId = undefined; } const state: ActiveSessionState = { activeSessionId: desiredActiveSessionId && !this.sessions.has(desiredActiveSessionId) ? desiredActiveSessionId : createActiveSessionId(this.sessions), runtime, clients: new Set(), pendingAttaches: 0, extensionUiRequests: new Map(), eventGeneration: createActiveSessionId(), lastEventSequence: 0, clientEnv, }; this.sessions.set(state.activeSessionId, state); this.bindingSessions.add(state.activeSessionId); let completeBinding!: () => void; const bindingCompletion = new Promise<void>((resolveBinding) => { completeBinding = resolveBinding; }); this.bindingCompletions.set(state.activeSessionId, bindingCompletion); onStateCreated?.(state); try { if (name) { await this.setStateSessionName(state, name); } await bindActiveSessionState(state, { broadcast: (targetSessionState, message) => this.broadcastToSession(targetSessionState, message), createConnectionState: (targetSessionState) => this.createConnectionState(targetSessionState), sessionReplaced: (targetSessionState) => this.refreshReplacedSessionState(targetSessionState), shutdown: () => { void this.shutdown(0); }, subagentRuntimeHost: this.createSubagentRuntimeHost(state), }); if (runtimeOpenGuard) { const guardResult = runtimeOpenGuard(); if (!(typeof guardResult === "boolean" ? guardResult : await guardResult)) { throw new RuntimeOpenCancelledError(); } } onStateBound?.(state); this.scheduleRosterFlush(); } catch (error) { state.unsubscribe?.(); this.sessions.delete(state.activeSessionId); await runtime.dispose().catch(() => undefined); throw error; } finally { this.bindingSessions.delete(state.activeSessionId); this.bindingCompletions.delete(state.activeSessionId); completeBinding(); } this.registerCronStoreForState(state); this.rebindCronJobsToState(state); if (runtime.metadata.kind !== "subagent") { // Mark the session as daemon-resident so a restarted daemon can // restore it. Closes for kill/completed/replaced flip this back to // sleep; clean shutdowns leave it in place on purpose. try { runtime.session.sessionManager.appendSessionState({ status: "active" }); } catch { // Marking is best-effort; the session still works unrestored. } } // Restore the last persisted status so it shows before the first sweep. this.summarizer.seed(state); this.recordWorkerRecoveryState(state, "ready"); return state; }
private refreshReplacedSessionState(state: ActiveSessionState): void { this.acpMcpOwners?.delete(state.activeSessionId); for (const client of state.clients) { this.abortSideQuestionsFor(client, state.activeSessionId); } this.summarizer.forget(state.activeSessionId); state.summaryState = undefined; state.runtime.session.setCurrentRecap(undefined); this.summarizer.seed(state); if (state.runtime.metadata.kind === "subagent") { const summaryState = state.summaryState as ActiveSessionState["summaryState"]; state.runtime.session.setCurrentRecap(summaryState?.summary); } this.registerCronStoreForState(state); this.rebindCronJobsToState(state); }
private registerCronStoreForState(state: ActiveSessionState): void { if (!this.options.worker) { return; } const session = state.runtime.session; const artifactDir = session.sessionManager.getSessionArtifactDir(); if (!artifactDir) { return; } if (this.cronStore.registerSessionArtifact(session.sessionId, artifactDir)) { this.cronStore.recoverSessionArtifact(session.sessionId); this.cronScheduler.wake(); } // A fresh worker only knows resident sessions' jobs; passive descendants' schedules must fire without hydration. if (state.runtime.metadata.kind !== "subagent") { void this.registerPassiveDescendantCronArtifacts().catch((error) => { this.log(`Could not register passive descendant scheduled jobs: ${String(error)}`); }); } }
private async registerPassiveDescendantCronArtifacts(): Promise<void> { let registered = false; for (const passive of await this.listPassiveRlmSubagents()) { try { const artifactDir = getSessionArtifactPathForFile(resolve(passive.entry.sessionFile), passive.info.id); if (this.cronStore.registerSessionArtifact(passive.info.id, artifactDir)) { registered = true; this.cronStore.recoverSessionArtifact(passive.info.id); } } catch (error) { this.log( `Could not register scheduled jobs for passive subagent ${passive.entry.childId}: ${String(error)}`, ); } } if (registered) { this.cronScheduler.wake(); } }
private async createRuntime( command: Extract<DaemonCommand, { type: "create" }>, runtimeOpenGuard?: RuntimeOpenGuard, ): Promise<ActiveSessionState> { const config = mergeAgentSessionRuntimeConfig(this.options.defaultSessionConfig, command.config); if (!config.cwd) { throw new Error("Active session config is missing cwd"); } if (!config.agentDir) { throw new Error("Active session config is missing agentDir"); }
const cwd = resolve(config.cwd); const agentDir = config.agentDir; const clientEnv = filterClientEnv(command.env); const cwdOverride = command.config?.cwd ? resolve(command.config.cwd) : undefined; const sessionPath = command.sessionPath ? await resolveDaemonSessionPath(command.sessionPath, cwd, config.sessionDir) : undefined; const sessionKey = sessionPath ? resolve(sessionPath) : undefined; if (sessionKey && this.findPassivationBySessionFile(sessionKey)) { await this.waitForPassivation(sessionKey); return this.createRuntime(command, runtimeOpenGuard); } const pending = sessionKey ? this.openingSessions.get(sessionKey) : undefined; if (pending && sessionKey) { // Join the in-process open before attempting the filesystem lease. The // creator owns that lease until its runtime is ready. let state: ActiveSessionState; try { state = await pending; } catch (error) { if (!runtimeOpenGuard && error instanceof RuntimeOpenCancelledError) { if (this.openingSessions.get(sessionKey) === pending) { this.openingSessions.delete(sessionKey); } return this.createRuntime(command); } throw error; } if (runtimeOpenGuard && !(await runtimeOpenGuard())) { throw new RuntimeOpenCancelledError(); } if (command.name) { await this.setStateSessionName(state, command.name); } this.adoptClientEnv(state, clientEnv); this.rebindCronJobsToState(state); return state; }
const passiveSubagent = sessionPath ? await this.findPassiveRlmSubagent(sessionPath) : undefined; if (passiveSubagent) { if (runtimeOpenGuard && !(await runtimeOpenGuard())) { throw new RuntimeOpenCancelledError(); } if (command.name) { const normalizedName = command.name.trim(); if (!normalizedName) { throw new Error("Session name cannot be empty"); } await this.assertFamilySessionNameAvailable({ name: normalizedName, depth: passiveSubagent.info.rlmDepth ?? passiveSubagent.entry.rlmDepth ?? 1, parentSessionId: passiveSubagent.entry.parentSessionId, parentSessionPath: passiveSubagent.entry.parentSessionFile ?? passiveSubagent.chain.at(-2)?.sessionFile ?? passiveSubagent.rootParentState?.runtime.session.sessionFile ?? passiveSubagent.rootInfo?.path, ignoreSessionId: passiveSubagent.info.id, }); } const state = await this.hydratePassiveRlmSubagent(passiveSubagent, clientEnv); if (runtimeOpenGuard && !(await runtimeOpenGuard())) { throw new RuntimeOpenCancelledError(); } if (command.name) { await this.setStateSessionName(state, command.name); } if (passiveSubagent.rootParentState) this.adoptClientEnv(passiveSubagent.rootParentState, clientEnv); this.adoptClientEnv(state, clientEnv); return state; }
// Hydration may have started while the async registry walk above was in // progress. Re-enter so the explicit opener joins its published promise. if (sessionKey && this.openingSessions.has(sessionKey)) { return this.createRuntime(command, runtimeOpenGuard); }
let releaseOpenReservation = () => {}; if (sessionKey) { const reservation = this.reservingSessionOpens.get(sessionKey); if (reservation) { await reservation; return this.createRuntime(command, runtimeOpenGuard); } let release!: () => void; const reserved = new Promise<void>((resolveReservation) => { release = resolveReservation; }); this.reservingSessionOpens.set(sessionKey, reserved); let released = false; releaseOpenReservation = () => { if (released) return; released = true; if (this.reservingSessionOpens.get(sessionKey) === reserved) { this.reservingSessionOpens.delete(sessionKey); } release(); }; }
const existing = sessionPath ? this.findSessionBySessionFile(sessionPath) : undefined; if (existing) { releaseOpenReservation(); await this.waitForBoundSession(existing); if (runtimeOpenGuard && !(await runtimeOpenGuard())) { throw new RuntimeOpenCancelledError(); } if (command.name) { await this.setStateSessionName(existing, command.name); } this.adoptClientEnv(existing, clientEnv); this.rebindCronJobsToState(existing); return existing; }
let sessionLease: SessionLease | undefined; let sessionManager: SessionManager; try { sessionLease = acquireSessionLease(sessionPath, agentDir); sessionManager = sessionPath ? await SessionManager.openAsync(sessionPath, config.sessionDir, cwdOverride) : command.noSession ? SessionManager.inMemory(cwd) : command.continueRecent ? SessionManager.continueRecent(cwd, config.sessionDir) : SessionManager.create(cwd, config.sessionDir); } catch (error) { sessionLease?.release(); releaseOpenReservation(); throw error; } const createState = async (): Promise<ActiveSessionState> => { if (runtimeOpenGuard && !(await runtimeOpenGuard())) { sessionLease?.release(); throw new RuntimeOpenCancelledError(); } const existing = this.findSessionBySessionFile(sessionManager.getSessionFile()); if (existing) { sessionLease?.release(); await this.waitForBoundSession(existing); // A live runtime already owns this session file; reuse it instead of // starting a second runtime that would interleave writes to one file. // clientEnv adopts the first offered identity (e.g. a pane opening a // cron-created session) but never overwrites one: extensions captured // the creator's identity at load, and swapping it would only make // pi.exec disagree with those captures. if (command.name) { await this.setStateSessionName(existing, command.name); } this.adoptClientEnv(existing, clientEnv); this.rebindCronJobsToState(existing); return existing; } let stateRef: ActiveSessionState | undefined; // Extensions capture client env (e.g. herdr pane identity) synchronously // while the runtime loads them, so it must be in process.env for the // duration; withClientEnv restores it after. const runtime = await withClientEnv(clientEnv, () => createAgentSessionRuntime(this.options.createRuntime, { cwd: sessionManager.getCwd(), agentDir, sessionManager, sessionConfig: config, runtimeMetadata: command.runtimeMetadata, sessionLease, sessionOptions: { rlmHeartbeatController: { listRlmHeartbeats: (listOptions) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.cronStore.listRlmHeartbeats(stateRef.activeSessionId, listOptions); }, createRlmHeartbeat: (input) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.createRlmHeartbeatForState(stateRef, input); }, updateRlmHeartbeat: (input) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.updateRlmHeartbeatForState(stateRef, input); }, deleteRlmHeartbeat: (id) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.deleteRlmHeartbeatForState(stateRef, id); }, }, agentMessageController: this.createAgentMessageController(() => stateRef), agentObserveController: this.createAgentObserveController(() => stateRef), agentSessionController: this.createAgentSessionController(() => stateRef), }, }), ); if (runtimeOpenGuard && !(await runtimeOpenGuard())) { await runtime.dispose().catch(() => undefined); throw new RuntimeOpenCancelledError(); } const state = await this.addRuntime( runtime, command.name, clientEnv, (state) => { stateRef = state; }, runtimeOpenGuard, ); return state; };
const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { releaseOpenReservation(); return createState(); } const openedSessionKey = resolve(sessionFile); if (this.openingSessions.has(openedSessionKey)) { sessionLease?.release(); releaseOpenReservation(); return this.createRuntime({ ...command, sessionPath: sessionFile }, runtimeOpenGuard); } const opening = Promise.resolve().then(createState); this.openingSessions.set(openedSessionKey, opening); releaseOpenReservation(); try { return await opening; } finally { releaseOpenReservation(); if (this.openingSessions.get(openedSessionKey) === opening) { this.openingSessions.delete(openedSessionKey); } } }
private async runCronJob(job: AgentCronJob): Promise<"skipped" | undefined> { const requirePersistedJob = this.cronStore.list().some((candidate) => candidate.id === job.id); const dueJob = requirePersistedJob ? this.getRunnableCronJob(job.id) : job; if (!dueJob) { return "skipped"; } const state = await this.getOrCreateCronJobSession(dueJob, requirePersistedJob); const runnableJob = requirePersistedJob ? this.getRunnableCronJob(job.id) : dueJob; if (!state || !runnableJob || !this.isCronJobRunnableForState(runnableJob, state, requirePersistedJob)) { return "skipped"; } const session = state.runtime.session; if (shouldDeferHeartbeatCronJob(runnableJob, session)) { return "skipped"; } const shouldQueueCronPrompt = session.isStreaming || session.isCompacting || session.isRetrying || session.isBashRunning || session.unfinishedActionCount > 0; if (!isHeartbeatCronJob(runnableJob) && shouldQueueCronPrompt) { if (!this.isCronJobRunnableForState(runnableJob, state, requirePersistedJob)) { return "skipped"; } await session.followUp(runnableJob.prompt, undefined, { resumeIfIdle: true, // A scheduled prompt is machine-triggered, so it must not outrank live human input. priority: "background", }); return; } const getRunnableJob = (): AgentCronJob | undefined => { const current = requirePersistedJob ? this.getRunnableCronJob(job.id) : runnableJob; return current && this.isCronJobRunnableForState(current, state, requirePersistedJob) ? current : undefined; }; const current = getRunnableJob(); if (!current) { return "skipped"; } // Re-check after the session admission fence wait: the job may have been cancelled, completed, or updated meanwhile. const unrunnableAtAdmission = new Error("Cron job became unrunnable before admission"); const admissionCommitted = () => { const refreshed = getRunnableJob(); if (!refreshed || refreshed.prompt !== current.prompt || refreshed.deliveryMode !== current.deliveryMode) { throw unrunnableAtAdmission; } }; try { if (isHeartbeatCronJob(current)) { await session.promptHeartbeat(current, { streamingBehavior: resolveHeartbeatStreamingBehavior(current.deliveryMode), followUpQueueKey: `heartbeat:${current.id}`, source: "rpc", admissionCommitted, }); return; } await session.promptUntilAccepted(current.prompt, { streamingBehavior: "followUp", source: "rpc", // A scheduled prompt is machine-triggered, so it must not outrank live human input. priority: "background", admissionCommitted, }); } catch (error) { if (error === unrunnableAtAdmission) { return "skipped"; } throw error; } }
private getRunnableCronJob(jobId: string): AgentCronJob | undefined { return this.cronStore.getClaimedJob(jobId) ?? this.cronStore.getDueJob(jobId); }
private createCronJobForState(state: ActiveSessionState, schedule: string, prompt: string): AgentCronJob { const session = state.runtime.session; const sessionFile = session.sessionFile; if (!sessionFile) { throw new Error("Cron jobs require a persisted session file"); } const job = this.cronStore.create({ activeSessionId: state.activeSessionId, sessionId: session.sessionId, sessionFile, cwd: state.runtime.cwd, runtimeKind: state.runtime.metadata.kind, scheduleText: schedule, prompt, }); this.cronScheduler.wake(); return job; }
private createHeartbeatForState( state: ActiveSessionState, schedule: string, instruction: string, deliveryMode?: AgentHeartbeatDeliveryMode, ): AgentCronJob { const session = state.runtime.session; const sessionFile = session.sessionFile; if (!sessionFile) { throw new Error("Heartbeats require a persisted session file"); } const previousHeartbeat = this.cronStore.getHeartbeat(state.activeSessionId); const job = this.cronStore.createHeartbeat({ activeSessionId: state.activeSessionId, sessionId: session.sessionId, sessionFile, cwd: state.runtime.cwd, runtimeKind: state.runtime.metadata.kind, scheduleText: normalizeHeartbeatSchedule(schedule), prompt: instruction, deliveryMode: deliveryMode ?? previousHeartbeat?.deliveryMode, }); if (previousHeartbeat) { this.removeQueuedHeartbeatFollowUp(state, previousHeartbeat); } this.cronScheduler.wake(); return job; }
private updateHeartbeatForState( state: ActiveSessionState, action: AgentHeartbeatUpdateAction, ): AgentCronJob | undefined { const job = action === "pause" ? this.cronStore.pauseHeartbeat(state.activeSessionId) : action === "resume" ? this.cronStore.resumeHeartbeat(state.activeSessionId) : this.cronStore.clearHeartbeat(state.activeSessionId); if (job && action !== "resume") { this.removeQueuedHeartbeatFollowUp(state, job); } this.cronScheduler.wake(); return job; }
private createRlmHeartbeatForState( state: ActiveSessionState, input: { instruction: string; interval?: string; label?: string; deliveryMode?: AgentHeartbeatDeliveryMode; }, ): AgentCronJob { const session = state.runtime.session; const sessionFile = session.sessionFile; if (!sessionFile) { throw new Error("RLM heartbeats require a persisted session file"); } const job = this.cronStore.createRlmHeartbeat({ activeSessionId: state.activeSessionId, sessionId: session.sessionId, sessionFile, cwd: state.runtime.cwd, runtimeKind: state.runtime.metadata.kind, label: input.label, scheduleText: normalizeHeartbeatSchedule(input.interval ?? DEFAULT_HEARTBEAT_SCHEDULE), prompt: input.instruction, deliveryMode: input.deliveryMode, }); this.cronScheduler.wake(); return job; }
private updateRlmHeartbeatForState( state: ActiveSessionState, input: { id: string; instruction?: string; interval?: string; label?: string; status?: "pause" | "resume"; deliveryMode?: AgentHeartbeatDeliveryMode; }, ): AgentCronJob | undefined { const job = this.cronStore.updateRlmHeartbeat(state.activeSessionId, input.id, { label: input.label, prompt: input.instruction, scheduleText: input.interval ? normalizeHeartbeatSchedule(input.interval) : undefined, status: input.status, deliveryMode: input.deliveryMode, }); if (job) { if ( input.instruction !== undefined || input.interval !== undefined || input.status === "pause" || input.deliveryMode !== undefined ) { this.removeQueuedHeartbeatFollowUp(state, job); } this.cronScheduler.wake(); } return job; }
private deleteRlmHeartbeatForState(state: ActiveSessionState, id: string): AgentCronJob | undefined { const job = this.cronStore.deleteRlmHeartbeat(state.activeSessionId, id); if (job) { this.removeQueuedHeartbeatFollowUp(state, job); this.cronScheduler.wake(); } return job; }
private listHeartbeats(): AgentConnectionHeartbeat[] { return this.cronStore .list() .filter((job) => isHeartbeatCronJob(job) && (job.status === "active" || job.status === "paused")) .map((job) => { const state = this.sessions.get(job.activeSessionId); const summary = state ? summaryForActiveSession(state) : undefined; return { job, ...(summary?.sessionName ? { sessionName: summary.sessionName } : {}), ...(summary?.firstMessage ? { firstMessage: summary.firstMessage } : {}), }; }); }
private manageHeartbeat( activeSessionId: string, jobId: string, action: AgentHeartbeatManagementAction, ): AgentCronJob | undefined { const job = this.cronStore.manageHeartbeat(activeSessionId, jobId, action); const state = this.sessions.get(activeSessionId); if (job && action !== "resume" && state) { this.removeQueuedHeartbeatFollowUp(state, job); } if (job) { this.cronScheduler.wake(); } return job; }
private rebindCronJobsToState(state: ActiveSessionState): void { const sessionFile = state.runtime.session.sessionFile; if (!sessionFile) { return; } const reboundJobs = this.cronStore.rebindSessionJobs({ activeSessionId: state.activeSessionId, sessionId: state.runtime.session.sessionId, sessionFile, cwd: state.runtime.cwd, }); if (reboundJobs.some((job) => job.status === "active")) { this.cronScheduler.wake(); } }
private cancelSubagentRlmHeartbeats(state: ActiveSessionState): void { if (state.runtime.metadata.kind !== "subagent") { return; } const cancelled = this.cronStore.cancelRlmHeartbeatsForSession(state.activeSessionId); for (const job of cancelled) { this.removeQueuedHeartbeatFollowUp(state, job); } if (cancelled.length > 0) { this.cronScheduler.wake(); } }
private cancelScheduledJobsForSession(state: ActiveSessionState): void { const session = state.runtime.session; const target: { activeSessionId: string; sessionId?: string; sessionFile?: string; } = { activeSessionId: state.activeSessionId, }; if (session?.sessionId) { target.sessionId = session.sessionId; } if (session?.sessionFile) { target.sessionFile = session.sessionFile; } const cancelled = this.cronStore.cancelJobsForSession(target); for (const job of cancelled) { this.removeQueuedHeartbeatFollowUp(state, job); } if (cancelled.length > 0) { this.cronScheduler.wake(); } }
private cancelScheduledJobsForSessionFile(sessionFile: string): void { const cancelled = this.cronStore.cancelJobsForSession({ sessionFile }); if (cancelled.length > 0) { this.cronScheduler.wake(); } }
private deleteSavedSessionFile( sessionPath: string, options?: Parameters<typeof deleteSessionFile>[1], ): ReturnType<typeof deleteSessionFile> { return deleteSessionFile(sessionPath, options); }
private removeQueuedHeartbeatFollowUp(state: ActiveSessionState, job: AgentCronJob): void { if (!isHeartbeatCronJob(job)) { return; } state.runtime.session.removeQueuedFollowUp(`heartbeat:${job.id}`); }
private async getOrCreateCronJobSession( job: AgentCronJob, requirePersistedJob: boolean, ): Promise<ActiveSessionState | undefined> { const dueJob = requirePersistedJob ? this.getRunnableCronJob(job.id) : job; if (!dueJob) { return undefined; } const activeIdMatch = this.sessions.get(dueJob.activeSessionId); const current = this.findSessionBySessionFile(dueJob.sessionFile) ?? (!requirePersistedJob || activeIdMatch?.runtime.session.sessionId === dueJob.sessionId ? activeIdMatch : undefined); const requiresRlmSubagentRestore = dueJob.source === "rlm_heartbeat" && dueJob.runtimeKind === "subagent" && current?.runtime.metadata.kind !== "subagent"; // A half-bound match falls through to createRuntime, which awaits the // pending create for the same session file instead of prompting mid-bind. if (current && !this.bindingSessions.has(current.activeSessionId) && !requiresRlmSubagentRestore) { this.rebindCronJobsToState(current); const reboundJob = requirePersistedJob ? this.getRunnableCronJob(job.id) : dueJob; return reboundJob && this.isCronJobRunnableForState(reboundJob, current, requirePersistedJob) ? current : undefined; } if (!requirePersistedJob) { return undefined; } if (dueJob.source === "rlm_heartbeat" && dueJob.runtimeKind === "subagent") { if (current && this.bindingSessions.has(current.activeSessionId)) { return undefined; } return this.restoreRlmHeartbeatSession(dueJob); } if (!(await this.isPersistedCronJobRunnable(dueJob.id))) { return undefined; } try { return await this.createRuntime({ type: "create", sessionPath: dueJob.sessionFile }, () => this.isPersistedCronJobRunnable(dueJob.id), ); } catch (error) { if (error instanceof RuntimeOpenCancelledError) { return undefined; } throw error; } }
private async restoreRlmHeartbeatSession(job: AgentCronJob): Promise<ActiveSessionState | undefined> { const childInfo = await readSessionInfo(job.sessionFile); const parentSessionPath = childInfo?.parentSessionPath; const parentInfo = parentSessionPath ? await readSessionInfo(parentSessionPath) : undefined; if ( !childInfo || childInfo.id !== job.sessionId || !parentSessionPath || !parentInfo || parentInfo.state?.status !== "active" ) { this.cancelRlmHeartbeat(job.id); return undefined; }
try { const residentChild = this.findSessionBySessionFile(job.sessionFile); await this.createRuntime( { type: "create", sessionPath: parentSessionPath }, () => this.getRunnableCronJob(job.id) !== undefined, ); const passiveSubagent = await this.findPassiveRlmSubagent( job.sessionFile, residentChild !== undefined && residentChild.runtime.metadata.kind !== "subagent", ); const resident = this.findSessionBySessionFile(job.sessionFile); const childState = passiveSubagent ? await this.hydratePassiveRlmSubagent(passiveSubagent) : resident ? await this.waitForBoundSession(resident) : undefined; if (passiveSubagent && childState && this.getRunnableCronJob(job.id) === undefined) { throw new RuntimeOpenCancelledError(); } if (!childState || childState.runtime.metadata.kind !== "subagent") { this.cancelRlmHeartbeat(job.id); return undefined; } this.rebindCronJobsToState(childState); const reboundJob = this.getRunnableCronJob(job.id); return reboundJob && this.isCronJobRunnableForState(reboundJob, childState, true) ? childState : undefined; } catch (error) { if (error instanceof RuntimeOpenCancelledError || error instanceof BoundSessionUnavailableError) { return undefined; } throw error; } }
private cancelRlmHeartbeat(jobId: string): void { if (this.cronStore.cancel(jobId)) { this.cronScheduler.wake(); } }
private async isPersistedCronJobRunnable(jobId: string): Promise<boolean> { for (let attempt = 0; attempt < 2; attempt++) { const job = this.getRunnableCronJob(jobId); if (!job) { return false; } const sessionFile = resolve(job.sessionFile); const sessionInfo = await readSessionInfo(sessionFile); const current = this.getRunnableCronJob(jobId); if (!current) { return false; } if (resolve(current.sessionFile) !== sessionFile || current.sessionId !== job.sessionId) { continue; } if (!sessionInfo || sessionInfo.id !== current.sessionId || sessionInfo.state?.status !== "active") { this.cancelScheduledJobsForSessionFile(current.sessionFile); return false; } return true; } return false; }
private isCronJobRunnableForState( job: AgentCronJob, state: ActiveSessionState, requirePersistedJob: boolean, ): boolean { if (this.sessions.get(state.activeSessionId) !== state || this.closingSessions.has(state.activeSessionId)) { return false; } if (!requirePersistedJob) { return job.status === "active" && job.activeSessionId === state.activeSessionId; } const current = this.getRunnableCronJob(job.id); const sessionFile = state.runtime.session.sessionFile; return ( current !== undefined && current.activeSessionId === state.activeSessionId && current.sessionId === state.runtime.session.sessionId && sessionFile !== undefined && resolve(current.sessionFile) === resolve(sessionFile) ); }
private findSessionBySessionFile(sessionFile: string | undefined): ActiveSessionState | undefined { if (!sessionFile) { return undefined; } const target = resolve(sessionFile); for (const state of this.sessions.values()) { const file = state.runtime.session.sessionFile; if (file && resolve(file) === target) { return state; } } return undefined; }
private getSessionState(id: string): ActiveSessionState { return resolveActiveSessionState(this.sessions, id); }
// A bind failure disposes the runtime, so half-bound sessions must not be // targetable by attach, agent messages, or observe. private getBoundSessionState(id: string): ActiveSessionState { const state = this.getSessionState(id); if (this.bindingSessions.has(state.activeSessionId)) { throw new BoundSessionUnavailableError(`Active session ${state.activeSessionId} is still initializing`); } if (this.closingSessions.has(state.activeSessionId)) { throw new BoundSessionUnavailableError(`Active session ${state.activeSessionId} is closing`); } return state; }
private async getOrHydrateBoundSessionState(id: string): Promise<ActiveSessionState> { let lookupError: unknown; try { return this.getBoundSessionState(id); } catch (error) { if (error instanceof BoundSessionUnavailableError) { return this.waitForHydratingChild(this.getSessionState(id), id); } if (error instanceof AmbiguousActiveSessionError) { throw error; } lookupError = error; } const passiveSubagent = await this.findPassiveRlmSubagent(id); if (passiveSubagent) { return this.hydratePassiveRlmSubagent(passiveSubagent); } const hydratingChild = [...this.sessions.values()].find( (state) => state.runtime.metadata.kind === "subagent" && state.runtime.metadata.rlmChildId === id, ); if (hydratingChild) { return this.waitForHydratingChild(hydratingChild, id); } try { return this.getBoundSessionState(id); } catch (error) { if (error instanceof BoundSessionUnavailableError) { return this.waitForHydratingChild(this.getSessionState(id), id); } if (error instanceof AmbiguousActiveSessionError) throw error; throw lookupError; } }
private async waitForHydratingChild(state: ActiveSessionState, selector: string): Promise<ActiveSessionState> { const sessionFile = state.runtime.session.sessionFile; if (!sessionFile || !this.findPassivationBySessionFile(sessionFile)) { return this.waitForBoundSession(state); } await this.waitForPassivation(sessionFile); const passive = await this.findPassiveRlmSubagent(sessionFile); return passive ? this.hydratePassiveRlmSubagent(passive) : this.getOrHydrateBoundSessionState(selector); }
private createSubagentRuntimeHost(parentState: ActiveSessionState): SubagentRuntimeHost { return { createRlmSubagentRuntime: async (options) => this.createRlmSubagentRuntime(parentState, options), createRlmRootSession: async (options) => this.createRlmRootSession(parentState, options), completeRlmSubagentRuntime: (childId, session) => { const state = [...this.sessions.values()].find( (candidate) => candidate.runtime.metadata.kind === "subagent" && candidate.runtime.metadata.parentActiveSessionId === parentState.activeSessionId && candidate.runtime.metadata.rlmChildId === childId && candidate.runtime.session === session, ); if (!state?.runtime.session.sessionFile) return false; if (state.runtime.metadata.rehydratedCompleted) return true; const metadata = state.runtime.metadata; const model = session.model; return this.recordRlmSubagentState(parentState, { childId, sessionName: session.sessionName ?? childId, sessionDir: metadata.sessionDir ?? dirname(state.runtime.session.sessionFile), sessionFile: state.runtime.session.sessionFile, rlmDepth: session.rlmDepth, rlmMaxDepth: session.rlmMaxDepth, rlmParentNodeId: metadata.rlmParentNodeId, prompt: metadata.prompt && metadata.prompt.length <= 4096 ? metadata.prompt : undefined, spawnCode: metadata.spawnCode, ...(model ? { model: { provider: model.provider, modelId: model.id } } : {}), status: "completed", createdAt: metadata.createdAt, }); }, releaseRlmSubagentRuntime: async (runtime, options, status) => { // Persist the deletion boundary first, but never let a registry failure // strand the cancelled child as a stale resident session. let deletionError: unknown; if (status === "cancelled") { try { await this.recordRlmSubagentDeletion(parentState, options.id, "revoked"); } catch (error) { deletionError = error; } } const state = [...this.sessions.values()].find( (candidate) => candidate.runtime.metadata.kind === "subagent" && candidate.runtime.metadata.parentActiveSessionId === parentState.activeSessionId && candidate.runtime.metadata.rlmChildId === options.id && candidate.runtime.session === runtime.session, ); const disposal = status === "cancelled" ? { kernelSnapshot: false } : undefined; try { if (state) { await this.closeSession( state, status === "cancelled" ? "killed" : "completed", true, true, undefined, disposal, ); } else { await runtime.session.disposeAsync(disposal); } } finally { // Sweep even when teardown throws (see deleteRlmSubagentRuntime); // never throws, so it cannot mask a teardown error. if (status === "cancelled" && deletionError === undefined) { const childSessionFile = runtime.session?.sessionFile; if (childSessionFile) { await this.deleteRlmSubagentArtifacts(options.id, childSessionFile); } } } if (deletionError !== undefined) throw deletionError; }, deleteRlmSubagentRuntime: async (childId, session) => { const state = [...this.sessions.values()].find( (candidate) => candidate.runtime.metadata.kind === "subagent" && candidate.runtime.metadata.parentActiveSessionId === parentState.activeSessionId && candidate.runtime.metadata.rlmChildId === childId, ); const parentFile = parentState.runtime.session.sessionFile; const parentPath = parentFile ? canonicalSessionPath(parentFile) : undefined; // Tombstoned edges included: a retried delete must still resolve the // child's path so its scheduled jobs get cancelled. const persistedEdge = parentPath ? (await this.rlmSpawnLedger().edges(true)).find( (edge) => edge.childId === childId && canonicalSessionPath(edge.parent) === parentPath, ) : undefined; // The writer-recorded path (not the ledger's realpath-canonical one) // keys the cron store and residency maps. const persisted = persistedEdge && parentFile ? await this.passiveRlmSubagentEntryForEdge(persistedEdge, { sessionId: parentState.runtime.session.sessionId, sessionFile: parentFile, }) : undefined; // Pre-ledger children have no edge at all (tombstoned or live): the // legacy registry (including its tombstones) is the last path source. const legacyFallback = !persisted && !state && parentFile ? ( await this.readLegacyRlmSubagentRegistry( this.legacyRlmSubagentRegistryPath(parentFile, parentState.runtime.session.sessionId), ) ).find((entry) => entry.childId === childId) : undefined; const childSessionFile = persisted?.sessionFile ?? state?.runtime.session.sessionFile ?? legacyFallback?.sessionFile; // Persist the deletion boundary before tearing down the runtime. await this.recordRlmSubagentDeletion(parentState, childId); const staleSession = state && session && state.runtime.session !== session ? session : undefined; try { try { if (state) { await this.closeSession(state, "killed", false, true, undefined, { kernelSnapshot: false }); } else { await session?.disposeAsync({ kernelSnapshot: false }); } } finally { await staleSession?.disposeAsync({ kernelSnapshot: false }); } } finally { // Runs even when teardown throws: the jobs-cancel rewrite and the // kernel dispose's final snapshot flush may have already happened, // resurrecting the artifact dir swept in recordRlmSubagentDeletion. // A killed close can join a passivation close that already skipped // killed cleanup. Neither step may throw here: a jobs-store error // would mask the teardown error and skip the sweep. if (childSessionFile) { try { this.cancelScheduledJobsForSessionFile(childSessionFile); } catch (error) { this.log( `failed to cancel scheduled jobs for deleted RLM subagent ${childId}: ${error instanceof Error ? error.message : String(error)}`, ); } await this.deleteRlmSubagentArtifacts(childId, childSessionFile); } } }, disposeRlmSubagentRuntimes: async () => { const cascadeError = await this.closeChildSessions(parentState, "replaced"); if (cascadeError) { throw cascadeError; } }, }; }
private async createRlmRootSession( parentState: ActiveSessionState, options: CreateRlmRootSessionOptions, ): Promise<RlmCreateSessionResult> { const link = this.supervisorLink(); if (!this.options.worker || !link) { throw new Error("rlm.create_session requires a daemon worker connected to its supervisor"); }
let activeSessionId: string | undefined; try { const client = link; const runtimeConfig = parentState.runtime.runtimeConfig; const inheritsProvider = options.model.provider === parentState.runtime.session.model?.provider; const authSource = parentState.runtime.services.authStorage.getAuthStatus(options.model.provider).source; const apiKey = inheritsProvider && authSource === "runtime" && (runtimeConfig?.provider ?? parentState.runtime.session.model?.provider) === options.model.provider ? runtimeConfig?.apiKey : undefined; // Deliberately narrow: PATH comes from the worker, which the supervisor seeded with // the launching client's environment, plus the one provider key this session needs. // Forwarding the whole environment here would leak unrelated provider keys into a // session that was never granted them. const launchEnv = collectDaemonLaunchEnv({ PATH: process.env.PATH }); const envKey = authSource === "environment" ? findEnvKeys(options.model.provider)?.[0] : undefined; if (envKey && process.env[envKey]) launchEnv[envKey] = process.env[envKey]; if (options.model.provider === "prime-inference" && process.env.PRIME_TEAM_ID !== undefined) { launchEnv.PRIME_TEAM_ID = process.env.PRIME_TEAM_ID; } const createResponse = await client.request( { type: "create", lifecycle: "resident", launchEnv, ...(options.sessionName ? { name: options.sessionName } : {}), config: { cwd: options.cwd, agentDir: parentState.runtime.services.agentDir, ...(runtimeConfig?.sessionDir ? { sessionDir: runtimeConfig.sessionDir } : {}), provider: options.model.provider, model: options.model.id, ...(apiKey ? { apiKey } : {}), thinking: options.thinkingLevel, ...(runtimeConfig?.telemetryDisabled ? { telemetryDisabled: true as const } : {}), }, }, 120_000, ); if (!createResponse.success) throw deserializeDaemonError(createResponse); const summary = createResponse.data as Partial<SessionSummary> | undefined; activeSessionId = summary?.activeSessionId ?? summary?.id; if ( !activeSessionId || typeof summary?.sessionId !== "string" || !summary.sessionId || typeof summary.sessionFile !== "string" || !summary.sessionFile || (summary.rlmDepth !== undefined && summary.rlmDepth !== 0) ) { throw new Error("Daemon supervisor returned an invalid depth-0 session summary"); }
const promptResponse = await client.request( { type: "prompt", activeSessionId, message: options.prompt, source: "rpc", }, 30_000, ); if (!promptResponse.success) throw deserializeDaemonError(promptResponse); return { active_session_id: activeSessionId, session_id: summary.sessionId, name: summary.sessionName ?? activeSessionId, session_file: summary.sessionFile, model: `${options.model.provider}/${options.model.id}`, }; } catch (error) { if (activeSessionId) { await link.request({ type: "kill", activeSessionId }, 30_000).catch(() => undefined); } throw error; } }
private async createRlmSubagentRuntime( parentState: ActiveSessionState, options: CreateRlmSubagentRuntimeOptions, ): Promise<AgentSessionRuntime> { const sessionManager = SessionManager.create( options.cwd ?? options.parentSession.sessionManager.getCwd(), options.sessionDir, ); sessionManager.newSession({ parentSession: options.parentSession.sessionFile, rlmDepth: options.rlmDepth, }); let stateRef: ActiveSessionState | undefined; // Subagents inherit the parent's client env (e.g. herdr pane identity). const runtime = await withClientEnv(parentState.clientEnv, () => createAgentSessionRuntime(this.options.createRuntime, { cwd: sessionManager.getCwd(), agentDir: parentState.runtime.services.agentDir, sessionManager, sessionStartEvent: { type: "session_start", reason: "startup" }, sessionConfig: parentState.runtime.runtimeConfig, sessionOptions: { model: options.model, thinkingLevel: options.thinkingLevel, serviceTier: options.serviceTier, scopedModels: options.scopedModels, initialActiveToolNames: options.activeToolNames, allowedToolNames: options.allowedToolNames, customTools: options.customTools, includeGoals: options.includeGoals, includeCompactSkill: options.includeCompactSkill, agentMessageController: this.createAgentMessageController(() => stateRef), agentObserveController: this.createAgentObserveController(() => stateRef), agentSessionController: this.createAgentSessionController(() => stateRef), rlmHeartbeatController: { listRlmHeartbeats: (listOptions) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.cronStore.listRlmHeartbeats(stateRef.activeSessionId, listOptions); }, createRlmHeartbeat: (input) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.createRlmHeartbeatForState(stateRef, input); }, updateRlmHeartbeat: (input) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.updateRlmHeartbeatForState(stateRef, input); }, deleteRlmHeartbeat: (id) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.deleteRlmHeartbeatForState(stateRef, id); }, }, rlmDepth: options.rlmDepth, rlmMaxDepth: options.rlmMaxDepth, rlmSessionDir: options.sessionDir, rlmAllowPush: options.allowPush, rlmParentNodeId: options.rlmParentNodeId, rlmParentAgent: options.parentSession.sessionName ?? options.parentSession.sessionId, semanticParentSessionId: options.parentSession.sessionId, semanticSpawnedByRequestId: options.spawnedByRequestId, }, runtimeMetadata: { kind: "subagent", createdAt: Date.now(), parentActiveSessionId: parentState.activeSessionId, parentSessionId: options.parentSession.sessionId, parentSessionFile: options.parentSession.sessionFile, rlmChildId: options.id, rlmParentNodeId: options.rlmParentNodeId, prompt: options.prompt, spawnCode: options.spawnCode, sessionDir: options.sessionDir, }, }), ); let state: ActiveSessionState; try { state = await this.addRuntime( runtime, undefined, parentState.clientEnv, (createdState) => { stateRef = createdState; }, () => options.parentSession.getRlmChildRunStatus(options.id) !== "cancelled", () => { if (runtime.session.sessionName !== options.sessionName) { runtime.session.setSessionName(options.sessionName); } if (runtime.session.sessionFile) { this.recordRlmSubagentState(parentState, { childId: options.id, sessionName: options.sessionName, sessionDir: options.sessionDir, sessionFile: runtime.session.sessionFile, rlmDepth: options.rlmDepth, rlmMaxDepth: options.rlmMaxDepth, rlmParentNodeId: options.rlmParentNodeId, prompt: options.prompt.length <= 4096 ? options.prompt : undefined, spawnCode: options.spawnCode, model: { provider: options.model.provider, modelId: options.model.id, }, status: "running", createdAt: runtime.metadata.createdAt, }); } options.onSessionPublished?.(runtime.session); }, ); } catch (error) { this.pendingRlmSpawnAppends.delete(`${parentState.activeSessionId}#${options.id}`); throw error; } // Admission is complete only once the spawn record is durably in the // ledger: no self-heal exists for a lost spawn record (seeding only runs // when the ledger file is absent; reconciliation only drops edges). A // failed append therefore FAILS admission — the just-added child is // closed like any other admission failure rather than admitted as a // ghost the ledger-driven listing and hydration could never find. const spawnAppend = this.pendingRlmSpawnAppends.get(`${parentState.activeSessionId}#${options.id}`); this.pendingRlmSpawnAppends.delete(`${parentState.activeSessionId}#${options.id}`); try { await spawnAppend; } catch (error) { await this.closeSession(state, "killed", false).catch(() => undefined); // The child was never admitted: no ledger edge exists, so the // display file written at the spawn write point must not remain // claiming a running child. try { rmSync(rlmSubagentDisplayPath(options.sessionDir), { force: true }); } catch { // Best-effort: a stale display file without an edge is inert. } throw new Error( `Failed to record RLM subagent spawn for ${options.id}: ${error instanceof Error ? error.message : String(error)}`, ); } return runtime; }
private async sessionPassivationSnapshot( state: ActiveSessionState, passiveRlmSubagents?: readonly PassiveRlmSubagent[], ): Promise<SessionPassivationSnapshot> { const passiveDescendants = passiveRlmSubagents ?? (await this.listPassiveRlmSubagents()); const summary = summaryForActiveSession(state); const sessionFile = state.runtime.session.sessionFile; const jobs = this.cronStore .list() .filter( (job) => job.activeSessionId === state.activeSessionId && job.status !== "cancelled" && job.status !== "completed", ); const hasPendingAdmission = [...this.promptAdmissions.values()].some( (admission) => admission.activeSessionId === state.activeSessionId && admission.status !== "cancelled", ) || state.runtime.session.hasPendingAdmissionWaiters; return { isSessionActive: summary.isSessionActive || summary.hasRunningRlmChildren === true || hasPendingAdmission, attachedClients: state.clients.size + state.pendingAttaches, hasRegisteredCronJob: jobs.some((job) => !isHeartbeatCronJob(job)), lastActivityAt: Date.parse(summary.lastActivityAt ?? ""), hasParent: state.runtime.metadata.kind === "subagent" && !!state.runtime.metadata.parentActiveSessionId, hasNonPassiveDescendants: getChildActiveSessionStates(this.sessions, state).length > 0, isHydrating: this.bindingSessions.has(state.activeSessionId) || (sessionFile ? this.openingSessions.has(resolve(sessionFile)) : false) || passiveDescendants.some( (passive) => passive.rootParentState === state && passive.chain.some((entry) => this.openingSessions.has(resolve(entry.sessionFile))), ), }; }
private async passivateSession( state: ActiveSessionState, idleEvictionMinutes: IdleEvictionMinutes, now: number, selectedSnapshot?: SessionPassivationSnapshot, ): Promise<boolean> { const sessionFile = state.runtime.session.sessionFile; const metadata = state.runtime.metadata; if (!sessionFile || metadata.kind !== "subagent" || !metadata.rlmChildId || !metadata.parentActiveSessionId) { return false; } const sessionKey = resolve(sessionFile); const parentActiveSessionId = metadata.parentActiveSessionId; const childId = metadata.rlmChildId; const existing = this.passivatingSessions.get(sessionKey); if (existing) { await existing; return false; } const snapshot = selectedSnapshot ?? (await this.sessionPassivationSnapshot(state)); if (!canPassivateSession(snapshot, idleEvictionMinutes, now)) return false;
// Publish the durable identity before running the close so opens and lazy // hydration can join throughout closeSessionOnce, including after sessions.delete. const passivation = Promise.resolve().then(async () => { // Fence against touches and state changes after candidate selection. This // snapshot is intentionally fresh rather than reusing the sweep snapshot. if ( this.shuttingDown || this.updateRestart !== undefined || this.sessions.get(state.activeSessionId) !== state || !canPassivateSession(await this.sessionPassivationSnapshot(state), idleEvictionMinutes, now) ) { return; } const parentState = this.sessions.get(parentActiveSessionId); if (!parentState) return; const idleMinutes = Math.floor((now - snapshot.lastActivityAt) / 60_000); // Detach parent tracking before the standard graceful runtime disposal. The // registry/catalog rows remain the sole passive representation after close. const unsubscribeChild = parentState.runtime.session.releaseRlmChildSession(childId, state.runtime.session); if (!unsubscribeChild) { return; } try { await this.closeSession(state, "shutdown", true, false); } catch (error) { if ( this.sessions.get(state.activeSessionId) === state && this.sessions.get(parentActiveSessionId) === parentState ) { throw error; } unsubscribeChild(); throw error; } unsubscribeChild(); this.log( `Passivated idle child sessionId=${state.runtime.session.sessionId} name=${JSON.stringify(state.runtime.session.sessionName ?? "")} idleMinutes=${idleMinutes}`, ); }); this.passivatingSessions.set(sessionKey, passivation); try { await passivation; return this.sessions.get(state.activeSessionId) !== state; } finally { if (this.passivatingSessions.get(sessionKey) === passivation) { this.passivatingSessions.delete(sessionKey); } } }
private async passivateIdleChildren( idleEvictionMinutes: IdleEvictionMinutes, now: number, limit: number, ): Promise<number> { if (this.shuttingDown || this.updateRestart !== undefined || limit <= 0) return 0; const states = [...this.sessions.values()]; const passiveRlmSubagents = await this.listPassiveRlmSubagents(); const snapshots = await Promise.all( states.map(async (state) => ({ state, snapshot: await this.sessionPassivationSnapshot(state, passiveRlmSubagents), })), ); const candidates = snapshots .filter(({ snapshot }) => canPassivateSession(snapshot, idleEvictionMinutes, now)) .sort((left, right) => left.snapshot.lastActivityAt - right.snapshot.lastActivityAt) .slice(0, limit); const results = await Promise.all( candidates.map(({ state, snapshot }) => this.passivateSession(state, idleEvictionMinutes, now, snapshot)), ); return results.filter(Boolean).length; }
private findPassivationBySessionFile(sessionFile: string): Promise<void> | undefined { return this.passivatingSessions.get(resolve(sessionFile)); }
private async waitForPassivation(sessionFile: string): Promise<void> { const passivation = this.findPassivationBySessionFile(sessionFile); if (passivation) await passivation.catch(() => {}); }
private async hydratePassiveRlmSubagent( passive: PassiveRlmSubagent, clientEnv?: Record<string, string>, ): Promise<ActiveSessionState> { if (this.updateRestart !== undefined) { throw new BoundSessionUnavailableError("Daemon is preparing an update restart"); } const isResident = (state: ActiveSessionState): boolean => this.sessions.get(state.activeSessionId) === state && !this.closingSessions.has(state.activeSessionId); const restartAfterParentChange = async (staleParent: ActiveSessionState): Promise<ActiveSessionState> => { const staleParentFile = staleParent.runtime.session.sessionFile; if (staleParentFile) await this.waitForPassivation(staleParentFile); const refreshed = await this.findPassiveRlmSubagent(passive.entry.sessionFile, true); if (!refreshed) throw new RuntimeOpenCancelledError(); const resident = this.findSessionBySessionFile(passive.entry.sessionFile); if ( resident && resident.runtime.metadata.kind === "subagent" && resident.runtime.metadata.rlmChildId === passive.entry.childId ) { return this.waitForBoundSession(resident); } return this.hydratePassiveRlmSubagent(refreshed, clientEnv); }; const rootParent = passive.rootParentState; if (!rootParent) { throw new Error(`Cannot hydrate RLM subagent ${passive.entry.childId} without a resident root parent`); } const rootParentFile = rootParent.runtime.session.sessionFile; if (rootParentFile) await this.waitForPassivation(rootParentFile); if (!isResident(rootParent)) { return restartAfterParentChange(rootParent); } let parentState = rootParent; for (const entry of passive.chain) { await this.waitForPassivation(entry.sessionFile); if (!isResident(parentState)) { return restartAfterParentChange(parentState); } const hydratingParent = parentState; const activeSessionId = entry === passive.entry ? passive.info.id : undefined; let hydrated: ActiveSessionState; try { hydrated = await this.rehydrateCompletedRlmSubagent(hydratingParent, entry, activeSessionId, clientEnv); } catch (error) { const passivation = this.findPassivationBySessionFile(entry.sessionFile); if (error instanceof BoundSessionUnavailableError && passivation) { await passivation.catch(() => {}); return restartAfterParentChange(hydratingParent); } if (isResident(hydratingParent)) throw error; return restartAfterParentChange(hydratingParent); } if (!isResident(hydratingParent)) { return restartAfterParentChange(hydratingParent); } if (!isResident(hydrated)) { return restartAfterParentChange(hydrated); } parentState = hydrated; } return parentState; }
private async rehydrateCompletedRlmSubagent( parentState: ActiveSessionState, entry: PassiveRlmSubagentEntry, restoreActiveSessionId?: string, clientEnv?: Record<string, string>, ): Promise<ActiveSessionState> { if (this.updateRestart !== undefined) { throw new BoundSessionUnavailableError("Daemon is preparing an update restart"); } const sessionKey = resolve(entry.sessionFile); const reservation = this.reservingSessionOpens.get(sessionKey); if (reservation) { await reservation; return this.rehydrateCompletedRlmSubagent(parentState, entry, restoreActiveSessionId, clientEnv); } const pending = this.openingSessions.get(sessionKey); if (pending) { const state = await pending; if (state.runtime.metadata.kind !== "subagent" || state.runtime.metadata.rlmChildId !== entry.childId) { if (this.openingSessions.get(sessionKey) === pending) this.openingSessions.delete(sessionKey); return this.rehydrateCompletedRlmSubagent(parentState, entry, restoreActiveSessionId, clientEnv); } return this.waitForBoundSession(state); } const existing = this.findSessionBySessionFile(entry.sessionFile); if (existing?.runtime.metadata.kind === "subagent" && existing.runtime.metadata.rlmChildId === entry.childId) { return this.waitForBoundSession(existing); } const hydration = (async () => { if (existing) { await this.closeSession(existing, "replaced"); } return this.rehydrateCompletedRlmSubagentOnce(parentState, entry, restoreActiveSessionId, clientEnv); })(); // Explicit opens and all lazy triggers share this path-keyed publication, // so no caller can acquire a second lease/runtime while hydration binds. this.openingSessions.set(sessionKey, hydration); try { return await hydration; } finally { if (this.openingSessions.get(sessionKey) === hydration) { this.openingSessions.delete(sessionKey); } } }
private async waitForBoundSession(state: ActiveSessionState): Promise<ActiveSessionState> { const completion = this.bindingCompletions.get(state.activeSessionId); if (completion) { await completion; } if (this.sessions.get(state.activeSessionId) !== state || this.bindingSessions.has(state.activeSessionId)) { throw new BoundSessionUnavailableError(`Active session ${state.activeSessionId} did not finish initializing`); } if (this.closingSessions.has(state.activeSessionId)) { throw new BoundSessionUnavailableError(`Active session ${state.activeSessionId} is closing`); } return state; }
private async rehydrateCompletedRlmSubagentOnce( parentState: ActiveSessionState, entry: PassiveRlmSubagentEntry, restoreActiveSessionId?: string, clientEnv?: Record<string, string>, ): Promise<ActiveSessionState> { const hydrationEnv = parentState.clientEnv ?? clientEnv; let stateRef: ActiveSessionState | undefined; let runtime: AgentSessionRuntime | undefined; let sessionLease: SessionLease | undefined; try { sessionLease = acquireSessionLease(entry.sessionFile, parentState.runtime.services.agentDir); const sessionManager = await SessionManager.openAsync(entry.sessionFile, entry.sessionDir); const modelRegistry = parentState.runtime.services.modelRegistry; let rehydratedModel: Model<Api> | undefined; if (entry.model) { const resolved = modelRegistry.find(entry.model.provider, entry.model.modelId); if (resolved && (await modelRegistry.canUseModel(resolved))) { rehydratedModel = resolved; } } runtime = await withClientEnv(hydrationEnv, () => createAgentSessionRuntime(this.options.createRuntime, { cwd: sessionManager.getCwd(), agentDir: parentState.runtime.services.agentDir, sessionManager, sessionStartEvent: { type: "session_start", reason: "startup" }, sessionConfig: parentState.runtime.runtimeConfig, sessionLease, sessionOptions: { ...(rehydratedModel ? { model: rehydratedModel } : {}), agentMessageController: this.createAgentMessageController(() => stateRef), agentObserveController: this.createAgentObserveController(() => stateRef), agentSessionController: this.createAgentSessionController(() => stateRef), rlmHeartbeatController: { listRlmHeartbeats: (listOptions) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.cronStore.listRlmHeartbeats(stateRef.activeSessionId, listOptions); }, createRlmHeartbeat: (input) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.createRlmHeartbeatForState(stateRef, input); }, updateRlmHeartbeat: (input) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.updateRlmHeartbeatForState(stateRef, input); }, deleteRlmHeartbeat: (id) => { if (!stateRef) { throw new Error("RLM heartbeat state is not ready for this session yet"); } return this.deleteRlmHeartbeatForState(stateRef, id); }, }, rlmSessionDir: entry.sessionDir, // Registry depth is authoritative (written at spawn); for legacy entries // without it, the shared accessor resolves persisted header depth or the // session file's sub- path before the depth-1 default. rlmDepth: entry.rlmDepth ?? (existsSync(entry.sessionFile) ? resolveSessionRlmDepth(sessionManager.getHeader() ?? {}, entry.sessionFile) : 1), rlmMaxDepth: entry.rlmMaxDepth, rlmParentNodeId: entry.rlmParentNodeId ?? entry.childId, }, runtimeMetadata: { kind: "subagent", createdAt: entry.createdAt, parentActiveSessionId: parentState.activeSessionId, parentSessionId: parentState.runtime.session.sessionId, ...(parentState.runtime.session.sessionFile ? { parentSessionFile: parentState.runtime.session.sessionFile } : {}), rlmChildId: entry.childId, rlmParentNodeId: entry.rlmParentNodeId ?? entry.childId, rehydratedCompleted: true, ...(entry.prompt ? { prompt: entry.prompt } : {}), ...(entry.spawnCode ? { spawnCode: entry.spawnCode } : {}), sessionDir: entry.sessionDir, }, }), ); const state = await this.addRuntime( runtime, undefined, hydrationEnv, (createdState) => { stateRef = createdState; }, undefined, undefined, restoreActiveSessionId, ); // The session transcript is authoritative for mutable metadata such as a // later user-assigned name; the registry value is only the spawn snapshot. if (!parentState.runtime.session.registerRlmChildSession(entry.childId, runtime.session)) { await this.closeSession(state, "replaced"); throw new RuntimeOpenCancelledError(); } if ( this.sessions.get(parentState.activeSessionId) !== parentState || this.closingSessions.has(parentState.activeSessionId) ) { const unsubscribeChild = parentState.runtime.session.releaseRlmChildSession(entry.childId, runtime.session); try { await this.closeSession(state, "replaced"); } finally { if (unsubscribeChild) { unsubscribeChild(); } } throw new RuntimeOpenCancelledError(); } return state; } catch (error) { if (stateRef && this.sessions.get(stateRef.activeSessionId) === stateRef) { await this.closeSession(stateRef, "completed").catch(() => undefined); } else { await runtime?.dispose().catch(() => undefined); } sessionLease?.release(); throw error; } }
private createAgentMessageController( getCurrentState: () => ActiveSessionState | undefined, ): AgentSessionMessageController { const requireCurrentState = () => { const current = getCurrentState(); if (!current) { throw new Error("Agent message state is not ready for this session yet"); } return current; }; return { listAgents: () => this.createAgentMessageListResult(requireCurrentState()), family: () => this.createAgentFamily(requireCurrentState()), assertSessionNameAvailable: (input) => this.assertFamilySessionNameAvailable(input), setSessionName: (name) => this.setStateSessionNameViaSupervisor(requireCurrentState(), name), sendAgentMessage: (input) => this.sendAgentSessionMessage({ targetSelector: input.target, message: input.message, fromState: requireCurrentState(), origin: "agent", }), }; }
private createAgentObserveController(getCurrentState: () => ActiveSessionState | undefined): AgentObserveController { const requireCurrentState = () => { const current = getCurrentState(); if (!current) { throw new Error("Agent observe state is not ready for this session yet"); } return current; }; return { listAgents: () => this.createAgentObserveListResult(requireCurrentState()), getAgent: (target) => this.createAgentObserveAgentSnapshot(requireCurrentState(), target), recentMessages: (input) => this.createAgentObserveRecentMessages(requireCurrentState(), input), }; }
private createAgentSessionController(getCurrentState: () => ActiveSessionState | undefined): AgentSessionController { const requireRootState = () => { const current = getCurrentState(); if (!current) throw new Error("Agent session state is not ready for this session yet"); if (current.runtime.metadata.kind !== "top-level") { throw new Error("agent.sessions is only available to top-level sessions"); } return current; }; return { listSessions: () => this.listAgentSessionSummaries(requireRootState()), getSession: (target) => this.getAgentSessionSummary(requireRootState(), target), recentTurns: (input) => this.getAgentSessionRecentTurns(requireRootState(), input), send: (input) => this.sendAgentSession(requireRootState(), input), stop: (target) => this.stopAgentSession(requireRootState(), target), restart: (target) => this.restartAgentSession(requireRootState(), target), }; }
private async listAgentSessionSummaries(currentState: ActiveSessionState): Promise<AgentSessionSummary[]> { const local = [...this.sessions.values()] .filter( (state) => state.runtime.metadata.kind === "top-level" && !this.bindingSessions.has(state.activeSessionId), ) .map((state) => this.createAgentSessionSummary(state, currentState)); const remote = (await this.listSupervisorAgentPeers()) .filter((agent) => agent.runtimeKind === "top-level") .map((agent) => this.createAgentSessionSummaryFromPeer(agent)); const seen = new Set<string>(); return [...local, ...remote].filter((agent) => { if (agent.activeSessionId === undefined) return true; if (seen.has(agent.activeSessionId)) return false; seen.add(agent.activeSessionId); return true; }); }
private agentSessionCapabilities(): readonly AgentSessionCapability[] { return this.options.worker ? ["send", "recent_turns", "stop", "restart"] : ["send", "recent_turns", "stop"]; }
private createAgentSessionSummary(state: ActiveSessionState, currentState: ActiveSessionState): AgentSessionSummary { return { ...this.createAgentObserveSummary(state, currentState), capabilities: this.agentSessionCapabilities(), }; }
private createAgentSessionSummaryFromPeer(agent: AgentSessionMessageAgentSummary): AgentSessionSummary { return { activeSessionId: agent.activeSessionId, sessionId: agent.sessionId, ...(agent.sessionName ? { sessionName: agent.sessionName } : {}), runtimeKind: "top-level", cwd: agent.cwd, status: agent.status ?? "idle", isCurrent: false, isStreaming: agent.isStreaming, isCompacting: false, attachedClients: 0, messageCount: 0, queuedCount: agent.unfinishedActionCount, isSessionActive: agent.isStreaming || agent.unfinishedActionCount > 0, ...(agent.parentActiveSessionId ? { parentActiveSessionId: agent.parentActiveSessionId } : {}), ...(agent.parentSessionId ? { parentSessionId: agent.parentSessionId } : {}), capabilities: this.agentSessionCapabilities(), }; }
private async getAgentSessionSummary( currentState: ActiveSessionState, target: string, ): Promise<AgentSessionSummary> { const sessions = await this.listAgentSessionSummaries(currentState); const exact = sessions.filter( (agent) => agent.activeSessionId === target || agent.sessionId === target || agent.sessionName === target, ); const suffix = sessions.filter( (agent) => (agent.activeSessionId !== undefined && matchesSessionIdSuffix(agent.activeSessionId, target)) || matchesSessionIdSuffix(agent.sessionId, target), ); const matches = exact.length > 0 ? exact : suffix; if (matches.length === 1) return matches[0]!; if (matches.length > 1) throw new Error(`Ambiguous agent session selector: ${target}`); throw new Error(`Unknown agent session: ${target}`); }
private findLocalAgentSession(summary: AgentSessionSummary): ActiveSessionState | undefined { const state = summary.activeSessionId !== undefined ? this.sessions.get(summary.activeSessionId) : undefined; return state?.runtime.metadata.kind === "top-level" ? state : undefined; }
private requireAgentSessionId(summary: AgentSessionSummary): string { if (summary.activeSessionId === undefined) { throw new Error(`Agent session is not available: ${summary.sessionId}`); } return summary.activeSessionId; }
private async agentSessionMessages(summary: AgentSessionSummary): Promise<AgentMessage[]> { const local = this.findLocalAgentSession(summary); if (local) return local.runtime.session.messages; const activeSessionId = this.requireAgentSessionId(summary); const response = await this.requestSupervisor({ type: "get_messages", activeSessionId }); if (!response) throw new Error(`Agent session is not available: ${activeSessionId}`); if (!response.success) throw deserializeDaemonError(response); const data = response.data; if (!data || typeof data !== "object" || !("messages" in data) || !Array.isArray(data.messages)) { throw new Error("Agent session returned an invalid message list"); } return data.messages as AgentMessage[]; }
private async getAgentSessionRecentTurns( currentState: ActiveSessionState, input: AgentSessionRecentTurnsInput, ): Promise<AgentSessionRecentTurnsResult> { const count = normalizeAgentSessionTurnCount(input.count); const maxChars = normalizeAgentSessionMaxChars(input.maxChars); const agent = await this.getAgentSessionSummary(currentState, input.target); const result = createAgentSessionTurnSummaries(await this.agentSessionMessages(agent), count, maxChars); return { agent, turns: result.turns, count, maxChars, truncated: result.truncated }; }
private async requestSupervisor( command: Parameters<DaemonClient["request"]>[0], timeoutMs = 30_000, ): Promise<DaemonResponse | undefined> { const supervisorSocketPath = this.supervisorSocketPathFromEnv(); if (!this.options.worker || !supervisorSocketPath) return undefined; const client = new DaemonClient(supervisorSocketPath); try { await client.connect(1000); await client.waitForHello(1000); return await client.request(command, timeoutMs); } finally { client.close(); } }
private async sendAgentSession( currentState: ActiveSessionState, input: AgentSessionSendInput, ): Promise<AgentSessionSendResult> { const agent = await this.getAgentSessionSummary(currentState, input.target); const activeSessionId = this.requireAgentSessionId(agent); const local = this.findLocalAgentSession(agent); const response = local ? undefined : await this.requestSupervisor( input.delivery === "steer" ? { type: "steer", activeSessionId, message: input.message } : { type: "follow_up", activeSessionId, message: input.message }, ); let queued: boolean | undefined; if (response) { if (!response.success) throw deserializeDaemonError(response); if (response.data && typeof response.data === "object" && "queued" in response.data) { queued = (response.data as { queued?: unknown }).queued === true; } } else { const state = local ?? this.findLocalAgentSession(agent); if (!state) throw new Error(`Agent session is not available: ${activeSessionId}`); queued = input.delivery === "follow_up" ? await state.runtime.session.followUp(input.message, undefined, { resumeIfIdle: true }) : undefined; if (input.delivery === "steer") await state.runtime.session.steer(input.message, undefined, { resumeIfIdle: true }); } return { operationId: randomUUID(), status: queued === false ? "accepted" : input.delivery === "steer" ? "accepted" : "queued", delivery: input.delivery, target: { activeSessionId, sessionId: agent.sessionId, ...(agent.sessionName ? { sessionName: agent.sessionName } : {}), }, }; }
private assertCanMutateAgentSession( currentState: ActiveSessionState, target: AgentSessionSummary, operation: "stop" | "restart", ): void { if (target.activeSessionId === currentState.activeSessionId) { throw new Error(`agent.sessions.${operation} cannot target the current session`); } }
private async stopAgentSession( currentState: ActiveSessionState, target: string, ): Promise<AgentSessionOperationResult> { const agent = await this.getAgentSessionSummary(currentState, target); this.assertCanMutateAgentSession(currentState, agent, "stop"); const activeSessionId = this.requireAgentSessionId(agent); const response = await this.requestSupervisor({ type: "kill", activeSessionId }); if (response) { if (!response.success) throw deserializeDaemonError(response); } else { const state = this.findLocalAgentSession(agent); if (!state) throw new Error(`Agent session is not available: ${activeSessionId}`); await this.closeSession(state, "killed"); } return { operationId: randomUUID(), status: "accepted" }; }
private async restartAgentSession( currentState: ActiveSessionState, target: string, ): Promise<AgentSessionOperationResult> { const agent = await this.getAgentSessionSummary(currentState, target); this.assertCanMutateAgentSession(currentState, agent, "restart"); const activeSessionId = this.requireAgentSessionId(agent); const response = await this.requestSupervisor({ type: "retry_worker", activeSessionId }, 90_000); if (!response) throw new Error("Agent session restart requires a daemon supervisor"); if (!response.success) throw deserializeDaemonError(response); return { operationId: randomUUID(), status: "accepted" }; }
private async createAgentObserveListResult(currentState: ActiveSessionState): Promise<AgentObserveListResult> { const family = await this.createAgentFamily(currentState); const residentBySessionId = new Map( this.listTargetableSessionStates(currentState).map((state) => [state.runtime.session.sessionId, state]), ); const agents = family.map((member) => { const state = residentBySessionId.get(member.entry.id); return state ? { ...this.createAgentObserveSummary(state, currentState), relationship: member.relationship, ...(member.entry.repliedSinceTask !== undefined ? { repliedSinceTask: member.entry.repliedSinceTask } : {}), } : this.createPersistedAgentObserveSummary(member); }); return { current: this.createAgentObserveSummary(currentState, currentState), agents, }; }
/** * Family member with no live session in this daemon: only catalog facts are known, * so the runtime flags stay false. A peer working in another worker keeps its * `activeSessionId` and its live status; everything else is absent or inactive. */ private createPersistedAgentObserveSummary(member: AgentFamilyMember): AgentObserveAgentSummary { const entry = member.entry; return { ...(entry.activeSessionId ? { activeSessionId: entry.activeSessionId } : {}), sessionId: entry.id, ...(entry.name ? { sessionName: entry.name } : {}), relationship: member.relationship, runtimeKind: entry.depth > 0 ? "subagent" : "top-level", ...(entry.cwd ? { cwd: entry.cwd } : {}), status: entry.status, isCurrent: false, isStreaming: false, isCompacting: false, attachedClients: 0, ...(entry.messageCount !== undefined ? { messageCount: entry.messageCount } : {}), queuedCount: 0, isSessionActive: entry.status === "running", ...(entry.repliedSinceTask !== undefined ? { repliedSinceTask: entry.repliedSinceTask } : {}), ...(entry.parentSessionId ? { parentSessionId: entry.parentSessionId } : {}), ...(entry.rlmChildId ? { rlmChildId: entry.rlmChildId } : {}), ...(entry.firstMessage ? { firstMessage: entry.firstMessage.slice(0, AGENT_OBSERVE_PREVIEW_MAX_CHARS) } : {}), }; }
private async createAgentObserveAgentSnapshot( currentState: ActiveSessionState, target: string, ): Promise<AgentObserveAgentSnapshot> { const targetState = await this.getOrHydrateAuthorizedAgentFamilyTarget(currentState, target); this.assertAgentFamilyReachable(currentState, targetState); return { agent: this.createAgentObserveSummary(targetState, currentState), }; }
private async createAgentObserveRecentMessages( currentState: ActiveSessionState, input: AgentObserveRecentMessagesInput, ): Promise<AgentObserveRecentMessagesResult> { const targetState = await this.getOrHydrateAuthorizedAgentFamilyTarget(currentState, input.target); this.assertAgentFamilyReachable(currentState, targetState); const limit = normalizeObserveLimit(input.limit); const maxChars = normalizeObserveMaxChars(input.maxChars); const messages = targetState.runtime.session.messages; const startIndex = Math.max(0, messages.length - limit); return { agent: this.createAgentObserveSummary(targetState, currentState), messages: messages .slice(startIndex) .map((message, offset) => createAgentObserveMessagePreview(message, startIndex + offset, maxChars)), limit, maxChars, truncated: startIndex > 0, }; }
private createAgentObserveSummary( state: ActiveSessionState, currentState: ActiveSessionState, ): AgentObserveAgentSummary { const summary = summaryForActiveSession(state); const session = state.runtime.session; const messages = session.messages; const latest = messages.at(-1); const status = session.isStreaming ? session.state.pendingToolCalls.size > 0 ? "tool" : "model" : session.isCompacting ? "compacting" : session.isSessionActive || session.hasRunningRlmChildren() ? "busy" : state.clients.size > 0 ? "user" : "idle"; return { activeSessionId: state.activeSessionId, sessionId: summary.sessionId, ...(summary.sessionName ? { sessionName: summary.sessionName } : {}), ...(summary.runtimeKind ? { runtimeKind: summary.runtimeKind } : {}), cwd: summary.cwd, status, isCurrent: state.activeSessionId === currentState.activeSessionId, isStreaming: summary.isStreaming, isCompacting: summary.isCompacting, attachedClients: summary.attachedClients, messageCount: summary.messageCount, queuedCount: summary.sessionActions.queuedCount, isSessionActive: summary.isSessionActive, ...(summary.parentActiveSessionId ? { parentActiveSessionId: summary.parentActiveSessionId } : {}), ...(summary.parentSessionId ? { parentSessionId: summary.parentSessionId } : {}), ...(summary.rlmChildId ? { rlmChildId: summary.rlmChildId } : {}), ...(summary.rlmParentNodeId ? { rlmParentNodeId: summary.rlmParentNodeId } : {}), ...(summary.firstMessage ? { firstMessage: summary.firstMessage } : {}), ...(latest ? { latestMessage: createAgentObserveMessagePreview( latest, messages.length - 1, AGENT_OBSERVE_PREVIEW_MAX_CHARS, ), } : {}), }; }
private handleConnection(socket: Socket): void { const client: DaemonSocketClient = { id: createActiveSessionId(), socket, attachedActiveSessionIds: new Set(), catchupActiveSessionIds: new Set(), backpressured: false, authenticated: this.options.worker === undefined, transport: this.options.worker ? "private-framed" : "jsonl", detachInput: () => {}, supportsExtensionUi: false, capabilities: new Set(DAEMON_DEFAULT_CLIENT_CAPABILITIES), }; this.clients.add(client); this.write(client, { type: "daemon_hello", socketPath: this.socketPath, protocol: DAEMON_PROTOCOL_INFO, schemaId: DAEMON_SCHEMA_ID, schemaRevision: DAEMON_SCHEMA_REVISION, appVersion: VERSION, runtime: getDaemonRuntimeIdentity(), clientId: client.id, serverCapabilities: DAEMON_DEFAULT_SERVER_CAPABILITIES, });
if (client.transport === "private-framed") { const decoder = new PrivateFrameDecoder(isDaemonWorkerFrameHeader); const onData = (chunk: Buffer) => { try { for (const frame of decoder.push(chunk)) { if (frame.header.kind === "command") { void this.handleLine(client, frame.payload.toString("utf8")); } } } catch (error) { socket.destroy(error instanceof Error ? error : new Error(String(error))); } }; const onEnd = () => { try { decoder.finish(); } catch (error) { socket.destroy(error instanceof Error ? error : new Error(String(error))); } }; socket.on("data", onData); socket.on("end", onEnd); client.detachInput = () => { socket.off("data", onData); socket.off("end", onEnd); }; } else { client.detachInput = attachJsonlLineReader(socket, (line) => { void this.handleLine(client, line); }); }
let cleanedUp = false; const cleanup = () => { if (cleanedUp) { return; } cleanedUp = true; socket.off("close", cleanup); socket.off("error", cleanup); this.clearClientCatchupRetry(client); for (const [pauseId, entry] of this.sessionInputPauses) { if (entry.owner !== client) continue; entry.pause.release(); this.sessionInputPauses.delete(pauseId); } this.detachClient(client); client.detachInput(); this.clients.delete(client); this.peerClaims.delete(client); // The fence check revokes a stale claim before ending the socket; the role survives. const wasSupervisor = client.authenticationRole === "supervisor"; this.revokeSupervisorClaim(client); const supervisorSocketPath = this.supervisorSocketPathFromEnv(); if (this.options.worker && wasSupervisor && supervisorSocketPath) { this.scheduleSupervisorAvailabilityCheck(supervisorSocketPath, 100); } }; socket.on("close", cleanup); socket.on("error", cleanup); socket.on("drain", () => { client.backpressured = false; if (!client.snapshotStreaming) { void this.catchUpBackpressuredClient(client).catch((error) => this.log(`could not catch up snapshot client ${client.id}: ${String(error)}`), ); } }); }
private promptAdmissionKey(activeSessionId: string, admissionId: string): string { return `${activeSessionId}\0${admissionId}`; }
/** * Parse and synchronously register prompt admission before returning a promise. * This method is intentionally non-async: handleLine invokes it before its first await. */ private parseCommandAndRegisterPromptAdmission(client: DaemonSocketClient, line: string): unknown { const wireValue = JSON.parse(line) as unknown; if (isDaemonCommandEnvelope(wireValue) && wireValue.clientId) client.id = wireValue.clientId; const parsed = (isDaemonCommandEnvelope(wireValue) ? { ...wireValue.command, id: wireValue.id } : wireValue) as { type?: unknown; activeSessionId?: unknown; admissionId?: unknown; }; if (parsed.type === "prompt" || parsed.type === "prompt_and_wait") { if (parsed.admissionId !== undefined) { if (typeof parsed.activeSessionId !== "string" || typeof parsed.admissionId !== "string") { throw new Error("Prompt admission requires string activeSessionId and admissionId"); } if (parsed.admissionId === "") throw new Error("admissionId must not be empty"); const key = this.promptAdmissionKey(parsed.activeSessionId, parsed.admissionId); if (this.promptAdmissions.has(key)) { throw new Error(`Prompt admission id is already in use: ${parsed.admissionId}`); } this.promptAdmissions.set(key, { activeSessionId: parsed.activeSessionId, admissionId: parsed.admissionId, controller: new AbortController(), status: "waiting", }); } } return parsed; }
private async handleLine(client: DaemonSocketClient, line: string): Promise<void> { let command: DaemonCommand; let clearParsedAdmission = () => {}; let promptHandlerOwnsAdmission = false; try { const parsed = this.parseCommandAndRegisterPromptAdmission(client, line) as { id?: unknown; type?: unknown; token?: unknown; grantId?: unknown; workerInstanceId?: unknown; purpose?: unknown; supervisorGeneration?: unknown; supervisorPid?: unknown; supervisorProcessStartId?: unknown; supervisorSocketPath?: unknown; activeSessionId?: unknown; admissionId?: unknown; capabilities?: unknown; supportsExtensionUi?: unknown; job?: unknown; }; const parsedAdmission = (parsed.type === "prompt" || parsed.type === "prompt_and_wait") && typeof parsed.activeSessionId === "string" && typeof (parsed as { admissionId?: unknown }).admissionId === "string" ? this.promptAdmissions.get( this.promptAdmissionKey(parsed.activeSessionId, (parsed as { admissionId: string }).admissionId), ) : undefined; clearParsedAdmission = () => { if (!parsedAdmission) return; const key = this.promptAdmissionKey(parsedAdmission.activeSessionId, parsedAdmission.admissionId); if (this.promptAdmissions.get(key) === parsedAdmission) { this.promptAdmissions.delete(key); } }; // Envelope client identity is irrelevant to worker-local prompt admission. // Public supervisor authentication has already bound this socket's identity. if (this.options.worker && client.authenticated !== true) { const commandId = typeof parsed.id === "string" ? parsed.id : undefined; if (parsed.type === "peer_auth") { // Single use: the grant is burned before its token is checked. const grant = typeof parsed.grantId === "string" ? this.peerGrants.get(parsed.grantId) : undefined; if (typeof parsed.grantId === "string") this.peerGrants.delete(parsed.grantId); const presentedTokenHash = typeof parsed.token === "string" ? createHash("sha256").update(parsed.token).digest() : undefined; const expectedTokenHash = grant ? createHash("sha256").update(grant.token).digest() : undefined; const expiresAt = grant ? Date.parse(grant.expiresAt) : Number.NaN; if ( this.peerAdmissionsFenced || !grant || !presentedTokenHash || !expectedTokenHash || !timingSafeEqual(presentedTokenHash, expectedTokenHash) || parsed.workerInstanceId !== grant.workerInstanceId || parsed.purpose !== grant.purpose || !Number.isFinite(expiresAt) || expiresAt <= Date.now() ) { clearParsedAdmission(); this.write(client, failure(commandId, "peer_auth", "Peer authentication failed")); client.socket.end(); return; } client.authenticated = true; client.authenticationRole = "session_client"; this.peerClaims.set(client, grant); this.write(client, { id: commandId, type: "response", command: "peer_auth", success: true, data: { workerInstanceId: grant.workerInstanceId, activeSessionId: grant.activeSessionId, purpose: grant.purpose, }, }); return; } if ( parsed.type !== "worker_auth" || parsed.token !== this.options.worker.authenticationToken || // Enforced only when presented: a downgraded (pre-instance-id) supervisor must still adopt live workers. (parsed.workerInstanceId !== undefined && parsed.workerInstanceId !== this.options.worker.workerInstanceId) || typeof parsed.supervisorGeneration !== "string" || !Number.isInteger(parsed.supervisorPid) || (parsed.supervisorPid as number) <= 0 || (parsed.supervisorProcessStartId !== undefined && typeof parsed.supervisorProcessStartId !== "string") || typeof parsed.supervisorSocketPath !== "string" ) { clearParsedAdmission(); this.write(client, failure(commandId, "worker_auth", "Worker authentication failed")); client.socket.end(); return; } const claim: SupervisorGenerationClaim = { supervisorGeneration: parsed.supervisorGeneration, supervisorPid: parsed.supervisorPid as number, ...(typeof parsed.supervisorProcessStartId === "string" ? { supervisorProcessStartId: parsed.supervisorProcessStartId } : {}), supervisorSocketPath: parsed.supervisorSocketPath, }; let ownerFingerprint: string; try { ownerFingerprint = await this.assertSupervisorClaimCurrent(claim); } catch { this.write(client, failure(commandId, "worker_auth", "supervisor_generation_stale")); client.socket.end(); return; } for (const previous of this.supervisorClaims.keys()) { if (previous !== client) { this.revokeSupervisorClaim(previous); previous.socket.end(); } } client.authenticated = true; client.authenticationRole = "supervisor"; this.supervisorClaims.set(client, { claim, ownerFingerprint }); this.clearSupervisorAvailabilityCheck(); this.scheduleSupervisorFenceCheck(); this.write(client, { id: commandId, type: "response", command: "worker_auth", success: true, data: { capabilities: [ DAEMON_WORKER_ROSTER_CAPABILITY, ...(this.options.worker.workerInstanceId !== undefined ? [DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY] : []), ], }, }); this.rosterReporter.snapshotPending = true; this.scheduleRosterFlush(); return; } const peerClaim = this.options.worker ? this.peerClaims.get(client) : undefined; if (peerClaim) { const commandId = typeof parsed.id === "string" ? parsed.id : undefined; const commandName = typeof parsed.type === "string" ? parsed.type : "unknown"; // Reachable only through shutdown, which fences admissions without ending live peers. if (this.peerAdmissionsFenced) { clearParsedAdmission(); this.write(client, failure(commandId, commandName, "Direct peer transport is fenced")); return; } if ( typeof parsed.type !== "string" || !isSessionPlaneDaemonCommand(parsed.type) || parsed.activeSessionId !== peerClaim.activeSessionId ) { clearParsedAdmission(); this.write( client, failure(commandId, commandName, "Command is not allowed on this direct peer transport"), ); return; } } else if (this.options.worker) { const boundClaim = this.supervisorClaims.get(client); if (!boundClaim) { clearParsedAdmission(); this.write( client, failure( typeof parsed.id === "string" ? parsed.id : undefined, "worker_auth", "supervisor_generation_stale", ), ); client.socket.end(); return; } const claimCheck = this.assertSupervisorClaimCurrent(boundClaim.claim, boundClaim.ownerFingerprint); // Observe the already-running fence check even if admission cancellation // wins the command wait below. void claimCheck.catch(() => {}); try { const ownerFingerprint = await waitForPromptAdmission(claimCheck, parsedAdmission?.controller?.signal); if (this.supervisorClaims.get(client) !== boundClaim || client.socket.destroyed) { clearParsedAdmission(); return; } boundClaim.ownerFingerprint = ownerFingerprint; } catch (error) { const admissionCancelled = error instanceof PromptAdmissionCancelledError; if (admissionCancelled) { // The fence check remains authoritative after cancellation. Its rejection // revokes only the exact binding that initiated it; replacements survive. void claimCheck.catch(() => { if (this.revokeSupervisorClaim(client, boundClaim)) client.socket.end(); }); } clearParsedAdmission(); if (this.supervisorClaims.get(client) !== boundClaim || client.socket.destroyed) return; this.write( client, failure( typeof parsed.id === "string" ? parsed.id : undefined, typeof parsed.type === "string" ? parsed.type : "worker_auth", admissionCancelled ? error : "supervisor_generation_stale", ), ); // Cancelling this prompt only abandons its admission wait. A genuine // stale supervisor claim fences only the binding that it checked. if (!admissionCancelled && this.revokeSupervisorClaim(client, boundClaim)) { client.socket.end(); } return; } } if (this.options.worker && typeof parsed.type === "string" && parsed.type.startsWith("worker_")) { const workerCommand = parsed as DaemonWorkerCommand; const updateLifecycle = workerCommand.type === "worker_prepare_update" || workerCommand.type === "worker_commit_update" || workerCommand.type === "worker_cancel_update"; if (this.updateRestart && !updateLifecycle) { this.write( client, failure(workerCommand.id, workerCommand.type, "Daemon is preparing an update restart"), ); return; } if (!updateLifecycle) this.mutationDrain.begin(); try { await this.handleWorkerCommand(client, workerCommand); } finally { if (!updateLifecycle) this.mutationDrain.end(); } return; }
if (typeof parsed.type !== "string" || !DAEMON_COMMAND_TYPES.has(parsed.type)) { const commandName = typeof parsed.type === "string" ? parsed.type : "unknown"; const commandId = typeof parsed.id === "string" ? parsed.id : undefined; this.write(client, failure(commandId, commandName, `Unknown daemon command: ${commandName}`)); return; } command = parsed as DaemonCommand; } catch (error) { this.write(client, failure(salvageDaemonCommandId(line), "parse", error, serializeDaemonError(error))); return; }
const mutation = command.type !== "prepare_update_restart" && isDaemonMutatingCommand(command); const restartPhase = this.updateRestart?.phase; // Mirror the supervisor's drain/fence split: abort-style commands stay // admitted only while mutations drain; once the checkpoint is being // captured they could race the snapshot and are rejected too. const restartRejected = restartPhase === "preparing" ? !UPDATE_RESTART_DRAIN_COMMANDS.has(command.type) : restartPhase !== undefined && command.type !== "shutdown"; if (mutation && restartRejected) { clearParsedAdmission(); this.write(client, failure(command.id, command.type, "Daemon is preparing an update restart")); return; } if (mutation) this.mutationDrain.begin(); const finishInFlight = this.inFlightTracker.recordCommandStart( command.type, command.id, "activeSessionId" in command ? String(command.activeSessionId) : undefined, ); try { const response = await this.handleCommand(client, command, () => { promptHandlerOwnsAdmission = true; }); if (response) { this.write(client, response); } } catch (error) { // Only the error message reaches the client (serializeDaemonError drops // the rest), so log the full stack here — this is the one place a handler // crash like a RangeError from a pathological session is recoverable. this.log( `daemon command "${command.type}" failed: ${error instanceof Error ? (error.stack ?? error.message) : String(error)}`, ); this.write(client, failure(command.id, command.type, error, serializeDaemonError(error))); } finally { finishInFlight(); if (!promptHandlerOwnsAdmission) clearParsedAdmission(); if (mutation) this.mutationDrain.end(); } }
private writeWorkerSuccess(client: DaemonSocketClient, command: DaemonWorkerCommand, data?: unknown): void { this.write(client, { id: command.id, type: "response", command: command.type, success: true, ...(data !== undefined ? { data } : {}), }); }
private async handleWorkerCommand(client: DaemonSocketClient, command: DaemonWorkerCommand): Promise<void> { try { switch (command.type) { case "worker_auth": this.write(client, failure(command.id, command.type, "Worker is already authenticated")); return; case "worker_register_peer_transport": { const grant = command.grant; const boundClaim = this.supervisorClaims.get(client); const expiresAt = Date.parse(grant.expiresAt); const now = Date.now(); for (const [grantId, pending] of this.peerGrants) { if (Date.parse(pending.expiresAt) < now) this.peerGrants.delete(grantId); } if ( this.peerAdmissionsFenced || !boundClaim || grant.issuerGeneration !== boundClaim.claim.supervisorGeneration || grant.purpose !== "session_client" || !grant.grantId || !grant.token || grant.workerInstanceId !== this.options.worker?.workerInstanceId || !this.sessions.has(grant.activeSessionId) || this.peerGrants.size >= PEER_GRANT_LIMIT || !Number.isFinite(expiresAt) || expiresAt <= now || expiresAt - now > PEER_GRANT_TTL_LIMIT_MS ) { this.write(client, failure(command.id, command.type, "Peer transport grant is invalid")); return; } this.peerGrants.set(grant.grantId, grant); this.writeWorkerSuccess(client, command); return; } case "worker_subscribe": { const state = this.getBoundSessionState(command.activeSessionId); setDaemonClientSessionCapabilities( client, state.activeSessionId, normalizeClientCapabilities(command.capabilities, command.supportsExtensionUi), ); state.clients.add(client); client.attachedActiveSessionIds.add(state.activeSessionId); this.write(client, success(command.id, "attach", summaryForActiveSession(state))); return; } case "worker_unsubscribe": { const state = this.getSessionState(command.activeSessionId); this.detachClientFromSession(client, state); this.write(client, success(command.id, "detach")); return; } case "worker_archive_and_shutdown": { // Close sessions first so direct peers read session_closed "killed", not a daemon shutdown. for (const state of [...this.sessions.values()]) { await this.closeSession(state, "killed"); } this.fencePeerTransports(); this.writeWorkerSuccess(client, command); setImmediate(() => void this.shutdown(0)); return; } case "worker_passivate_idle_children": { const count = await this.passivateIdleChildren(command.idleEvictionMinutes, command.now, command.limit); this.writeWorkerSuccess(client, command, { count }); return; } case "worker_deliver_message": { const receipt = await this.sendAgentSessionMessage({ targetSelector: command.targetActiveSessionId, message: command.message, sender: command.sender, senderKey: command.sender.activeSessionId ?? `client:${command.sender.clientId}`, origin: "agent", }); this.writeWorkerSuccess(client, command, receipt); return; } case "worker_prepare_update": { this.fencePeerTransports("update"); const transaction = this.beginUpdateRestartTransaction(client); const manifest = await this.runUpdateRestartPreparation(transaction); this.writeWorkerSuccess(client, command, manifest); return; } case "worker_commit_update": { const transaction = this.updateRestart; if (!transaction || transaction.phase === "preparing" || transaction.phase === "fencing") { throw new Error("Daemon has no prepared update checkpoint"); } if (transaction.phase === "publishing") { throw new Error("Daemon update checkpoint is already committing"); } if (transaction.owner !== client) throw new Error("Daemon update checkpoint belongs to another supervisor"); if (transaction.deadline) clearTimeout(transaction.deadline); transaction.deadline = undefined; transaction.phase = "publishing"; let manifest: DaemonUpdateRestartManifest; try { manifest = await this.commitPreparedUpdateRestart(transaction.id); } catch (error) { if (this.updateRestart === transaction) transaction.phase = "prepared"; if (transaction.abort.signal.aborted) this.cancelPreparedUpdateRestart(transaction.id); throw error; } this.writeWorkerSuccess(client, command, manifest); if (!this.supervisorClaims.has(client)) { setImmediate(() => void this.shutdown(0)); } return; } case "worker_cancel_update": { const transaction = this.updateRestart; if (transaction && transaction.owner !== client) { throw new Error("Daemon update checkpoint belongs to another supervisor"); } if (transaction?.phase === "publishing") { throw new Error("Daemon update checkpoint is already committing"); } if (transaction) this.cancelPreparedUpdateRestart(transaction.id); this.peerAdmissionsFenced = false; this.writeWorkerSuccess(client, command); return; } default: { // Legacy commands from older supervisors (e.g. worker_sync_agent_peers) must fail fast, not time out. const unknown = command as { id?: string; type: string }; this.write(client, failure(unknown.id, unknown.type, `Unknown worker command: ${unknown.type}`)); return; } } } catch (error) { this.write(client, failure(command.id, command.type, error, serializeDaemonError(error))); } }
private async handleCommand( client: DaemonSocketClient, command: DaemonCommand, onPromptHandlerOwnsAdmission: () => void = () => {}, ): Promise<DaemonResponse | undefined> { if ("agentMessageId" in command && command.agentMessageId === "") { throw new Error("agentMessageId must not be empty"); } if ("admissionId" in command && command.admissionId === "") { throw new Error("admissionId must not be empty"); } if ((command.type === "steer" || command.type === "follow_up") && command.expandPromptTemplates !== false) { const replayFields = (["content", "customMessage", "prefixMessages"] as const).filter( (field) => command[field] !== undefined, ); if (replayFields.length > 0) { throw new Error( `${command.type} replay fields (${replayFields.join(", ")}) require expandPromptTemplates=false`, ); } } switch (command.type) { case "ack_result": return undefined; case "list": { const activeSessions = Array.from(this.sessions.values()); const scheduledJobs = this.cronStore.list(); if (!command.all) { return success(command.id, "list", { sessions: await this.buildSessionListWithPassiveRlmSubagents(activeSessions, [], scheduledJobs), }); } const defaultConfig = this.options.defaultSessionConfig; const listSessionDir = command.sessionDir ?? defaultConfig.sessionDir; if (command.cwd) { const savedSessions = await SessionManager.list(resolve(command.cwd), listSessionDir); return success(command.id, "list", { sessions: await this.buildSessionListWithPassiveRlmSubagents( activeSessions, savedSessions, scheduledJobs, ), }); } const savedSessions = listSessionDir !== undefined ? await SessionManager.listAll(undefined, listSessionDir) : await SessionManager.listAll(); return success(command.id, "list", { sessions: await this.buildSessionListWithPassiveRlmSubagents( activeSessions, savedSessions, scheduledJobs, ), }); }
case "list_saved_sessions": { let activeSessionId: string | undefined; let cwd: string; let sessionDir: string | undefined; if ("activeSessionId" in command) { activeSessionId = command.activeSessionId; const sessionManager = this.getSessionState(activeSessionId).runtime.session.sessionManager; cwd = sessionManager.getCwd(); sessionDir = sessionManager.getSessionDir(); } else { cwd = resolve(command.cwd); sessionDir = command.sessionDir; } const callbacks = command.id ? { onProgress: (loaded: number, total: number) => { this.write(client, { id: command.id, type: "session_list_progress", command: "list_saved_sessions", ...(activeSessionId ? { activeSessionId } : {}), loaded, total, }); }, onSession: (session: SessionInfo) => { this.write(client, { id: command.id, type: "session_list_item", command: "list_saved_sessions", ...(activeSessionId ? { activeSessionId } : {}), session: serializeSavedSessionInfo(session), }); }, } : undefined; const savedSessions = command.scope === "current" ? await SessionManager.list(cwd, sessionDir, callbacks) : await SessionManager.listAll(callbacks, sessionDir); const sessions = await withPassiveRlmDescendantInfos(savedSessions, this.rlmSpawnLedgerFor(sessionDir), { ...(command.scope === "current" ? { cwd } : {}), ...(callbacks ? { onSession: callbacks.onSession } : {}), log: (message) => this.log(message), }); // A streaming client already consumed every row from the progress messages. return success(command.id, "list_saved_sessions", { sessions: command.streamOnly === true ? [] : sessions.map(serializeSavedSessionInfo), }); }
case "create": { const state = await this.createRuntime(command); return success(command.id, "create", summaryForActiveSession(state)); }
case "attach": { const state = await this.getOrHydrateBoundSessionState(command.activeSessionId); if (command.clientId) { client.id = command.clientId; } setDaemonClientSessionCapabilities( client, state.activeSessionId, normalizeClientCapabilities(command.capabilities, command.supportsExtensionUi), ); const streamsSnapshot = client.transport === "private-framed" && daemonClientCapabilitiesForSession(client, state.activeSessionId).has("chunked_snapshot"); // Attach is admitted during update-restart preparation as a read. Env // adoption remains safe while mutations are only draining; after fencing, // defer it until rollback so the checkpoint never omits a live identity. const clientEnv = filterClientEnv(command.env); const deferClientEnv = this.updateRestart && this.updateRestart.phase !== "preparing"; if (!deferClientEnv) this.adoptClientEnv(state, clientEnv); const snapshotSignal = streamsSnapshot ? markClientSnapshotStreaming(client, state.activeSessionId) : undefined; let result: DaemonAttachResult; state.pendingAttaches++; try { result = await this.createAttachResult(client, state, command); if ( this.sessions.get(state.activeSessionId) !== state || this.closingSessions.has(state.activeSessionId) ) { throw new BoundSessionUnavailableError( `Active session ${state.activeSessionId} closed during attach`, ); } } catch (error) { removeDaemonClientSessionCapabilities(client, state.activeSessionId); if (streamsSnapshot) { finishClientSnapshotStreaming(client, state.activeSessionId); } throw error; } finally { state.pendingAttaches--; } state.clients.add(client); client.attachedActiveSessionIds.add(state.activeSessionId); // Carrier-less mutation: a direct viewer changes directAttachedClients with no session event. if (client.authenticationRole === "session_client") this.scheduleRosterFlush(); if (deferClientEnv && clientEnv) { this.updateRestart?.deferredClientEnv.push({ client, state, env: clientEnv, }); } if (streamsSnapshot) { const snapshotId = snapshotTransferId(result.snapshot); let transcript: SnapshotTranscriptChunkSource; try { transcript = createSnapshotTranscriptChunks({ activeSessionId: state.activeSessionId, snapshotId, messages: result.snapshot.messages, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, signal: snapshotSignal, }); } catch (error) { state.clients.delete(client); client.attachedActiveSessionIds.delete(state.activeSessionId); removeDaemonClientSessionCapabilities(client, state.activeSessionId); finishClientSnapshotStreaming(client, state.activeSessionId); throw error; } const streamedResult: DaemonAttachResult = { ...result, messages: result.messages ? [] : undefined, snapshot: { ...result.snapshot, messages: [] }, snapshotStream: { id: snapshotId, messageCount: result.snapshot.messages.length, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, }, }; setImmediate(() => { void this.streamWorkerSnapshot( client, streamedResult, transcript, "attach", snapshotSignal, true, ).catch((error) => this.log(`could not stream attach snapshot: ${String(error)}`)); }); return success(command.id, "attach", streamedResult); } // Slim clients consume only the command response; legacy clients (e.g. // the plain daemon attach REPL) read state/messages off this event. // Skipping it for slim clients halves the attach payload. if (result.state && result.messages) { this.write(client, { type: "session_attached", activeSessionId: state.activeSessionId, state: result.state, messages: result.messages, snapshot: result.snapshot, replay: result.replay, lastEventSequence: result.lastEventSequence, }); } return success(command.id, "attach", result); }
case "detach": { if (command.activeSessionId) { const state = this.getSessionState(command.activeSessionId); for (const [pauseId, entry] of this.sessionInputPauses) { if (entry.owner !== client || entry.activeSessionId !== command.activeSessionId) continue; entry.pause.release(); this.sessionInputPauses.delete(pauseId); } this.detachClientFromSession(client, state); } else { for (const [pauseId, entry] of this.sessionInputPauses) { if (entry.owner !== client) continue; entry.pause.release(); this.sessionInputPauses.delete(pauseId); } this.detachClient(client); } return success(command.id, "detach"); }
case "kill": { const state = this.getSessionState(command.activeSessionId); await this.closeSession(state, "killed"); return success(command.id, "kill"); }
case "rename": { const state = this.getSessionState(command.activeSessionId); const name = command.name.trim(); if (!name) { throw new Error("Session name cannot be empty"); } await this.setStateSessionNameForCommand(state, name); return success(command.id, "rename", summaryForActiveSession(state)); }
case "rename_saved_session": { if (command.activeSessionId) { this.getSessionState(command.activeSessionId); } const state = this.findActiveSessionByFile(command.sessionPath); const name = command.name.trim(); if (!name) { throw new Error("Session name cannot be empty"); } if (state) { await this.setStateSessionNameForCommand(state, name); } else { const info = await readSessionInfo(command.sessionPath); if (!info) throw new Error(`Session not found: ${command.sessionPath}`); const depth = info.rlmDepth ?? 0; await this.withSessionNameReservation( { name, depth, ...(depth > 0 && info.parentSessionPath ? { parentSessionPath: info.parentSessionPath } : {}), }, async () => { await this.assertFamilySessionNameAvailable( { name, depth, ...(depth > 0 && info.parentSessionPath ? { parentSessionPath: info.parentSessionPath } : {}), ignoreSessionId: info.id, }, undefined, true, ); SessionManager.open(command.sessionPath).appendSessionInfo(name); await this.rlmSpawnLedger() .appendRenameByChildPath(command.sessionPath, name) .catch((error) => { this.log( `failed to append RLM ledger rename: ${error instanceof Error ? error.message : String(error)}`, ); }); }, ); } return success(command.id, "rename_saved_session"); }
case "delete_saved_session": { if (command.activeSessionId) { this.getSessionState(command.activeSessionId); } if (this.findActiveSessionByFile(command.sessionPath)) { throw new Error("Cannot delete the currently active session"); } const composedEntry = this.rosterEntryForSessionPath(canonicalSessionPath(command.sessionPath)); const { deletedInfo, ledgerEdge } = await tombstoneSavedSessionDelete( this.rlmSpawnLedger(), command.sessionPath, composedEntry?.summary, ); const result = await this.deleteSavedSessionFile(command.sessionPath, { afterFileRemoved: () => { this.cancelScheduledJobsForSessionFile(command.sessionPath); }, }); if (result.ok && this.options.worker) { const removedAgentId = composedEntry?.agentId ?? (ledgerEdge ? this.rosterAgentIdForRlmChild(ledgerEdge.childId, ledgerEdge.parent) : deletedInfo?.id); if (removedAgentId) { this.rosterReporter.removedAgentIds.set( removedAgentId, composedEntry?.summary.sessionId ?? deletedInfo?.id, ); this.scheduleRosterFlush(); } } return success(command.id, "delete_saved_session", result); }
case "cancel_prompt_admission": { const admission = this.promptAdmissions.get( this.promptAdmissionKey(command.activeSessionId, command.admissionId), ); if (!admission) { return success(command.id, command.type, { status: "unknown" as const, }); } if (admission.status === "owned") { if (command.cancelOwned) admission.controller?.abort(); return success(command.id, command.type, { status: "owned" as const, }); } if (admission.status === "waiting") { admission.status = "cancelled"; admission.controller?.abort(); } return success(command.id, command.type, { status: "cancelled" as const, }); }
case "prompt": case "prompt_and_wait": { onPromptHandlerOwnsAdmission(); const admissionKey = command.admissionId ? this.promptAdmissionKey(command.activeSessionId, command.admissionId) : undefined; const admission = admissionKey ? this.promptAdmissions.get(admissionKey) : undefined; if (command.admissionId && !admission) { throw new Error("Prompt admission was not registered during command parsing"); } const clearAdmission = () => { if (admissionKey && this.promptAdmissions.get(admissionKey) === admission) { this.promptAdmissions.delete(admissionKey); } }; const commitAdmission = () => { if (admission?.status === "waiting") admission.status = "owned"; }; let state: ActiveSessionState; try { if (admission?.status === "cancelled") throw new PromptAdmissionCancelledError(); state = this.getBoundSessionState(command.activeSessionId); } catch (error) { clearAdmission(); throw error; } const options: PromptOptions = { content: command.content, images: command.images, streamingBehavior: command.streamingBehavior, queueIfBusy: command.queueIfBusy ?? command.streamingBehavior !== undefined, resumeIfIdle: command.streamingBehavior !== undefined, expandPromptTemplates: command.expandPromptTemplates, skipInputHandlers: command.expandPromptTemplates === false ? true : undefined, source: command.source, ...(admission?.controller ? { signal: admission.controller.signal, admissionCommitted: commitAdmission, } : {}), }; if (command.type === "prompt_and_wait") { try { await state.runtime.session.promptAndWait(command.message, { ...options, preflightResult: (didSucceed) => { if (didSucceed) this.recordWorkerRecoveryState(state, "prompt_accepted", true); }, }); return success(command.id, command.type); } finally { clearAdmission(); } }
let responseSent = false; let preflightRejected = false; const sendSuccessResponse = () => { if (responseSent) return; responseSent = true; this.write(client, success(command.id, "prompt")); }; const prompt = command.agentMessageId !== undefined && command.expandPromptTemplates === false ? state.runtime.session.acceptAgentMessagePrompt.bind(state.runtime.session) : state.runtime.session.promptUntilAccepted.bind(state.runtime.session); void prompt(command.message, { ...options, agentMessageId: command.agentMessageId, customMessage: command.customMessage, preflightResult: (didSucceed) => { if (didSucceed) { this.recordWorkerRecoveryState(state, "prompt_accepted", true); sendSuccessResponse(); } else { preflightRejected = true; } }, }) .then(() => { if (preflightRejected) { const error = new Error("Prompt was not accepted by the session."); this.write(client, failure(command.id, "prompt", error, serializeDaemonError(error))); } else { sendSuccessResponse(); } }) .catch((error) => { if (responseSent) { this.broadcastToSession(state, failure(undefined, "prompt", error, serializeDaemonError(error))); } else { this.write(client, failure(command.id, "prompt", error, serializeDaemonError(error))); } }) .finally(clearAdmission); return undefined; }
case "steer": { const state = this.getBoundSessionState(command.activeSessionId); if (command.expandPromptTemplates === false) { await state.runtime.session.restoreSteeringMessage(command.message, command.images, { queueKey: command.queueKey, agentMessageId: command.agentMessageId, content: command.content, customMessage: command.customMessage, prefixMessages: command.prefixMessages, }); } else { await state.runtime.session.steer(command.message, command.images, { queueKey: command.queueKey, agentMessageId: command.agentMessageId, resumeIfIdle: true, }); } this.recordWorkerRecoveryState(state, "steer_queued", true); return success(command.id, "steer"); }
case "follow_up": { const state = this.getBoundSessionState(command.activeSessionId); let queued = true; let admitted = true; if (command.expandPromptTemplates === false) { queued = await state.runtime.session.restoreFollowUpMessage(command.message, command.images, { queueKey: command.queueKey, agentMessageId: command.agentMessageId, content: command.content, customMessage: command.customMessage, prefixMessages: command.prefixMessages, }); admitted = queued; } else { queued = await state.runtime.session.followUp(command.message, command.images, { queueKey: command.queueKey, agentMessageId: command.agentMessageId, resumeIfIdle: true, }); admitted = queued; } if (admitted) { this.recordWorkerRecoveryState(state, "follow_up_queued", true); } return success(command.id, "follow_up", { queued }); }
case "restore_next_turn": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.restorePendingNextTurnMessages(command.messages); return success(command.id, "restore_next_turn"); }
case "restore_actions": { const state = this.getSessionState(command.activeSessionId); const restored = await state.runtime.session.restoreSessionActions(command.snapshot); if (restored > 0) this.recordWorkerRecoveryState(state, "actions_restored", true); return success(command.id, "restore_actions", { restored }); }
case "append_custom_message": { const state = this.getSessionState(command.activeSessionId); await state.runtime.session.sendCustomMessage(command.message); return success(command.id, "append_custom_message"); }
case "resume_queue": { const state = this.getSessionState(command.activeSessionId); if (!state.runtime.session.resumeQueuedWork()) { const error = new Error("No queued work to resume"); return failure(command.id, "resume_queue", error, serializeDaemonError(error)); } return success(command.id, "resume_queue"); }
case "send_message": { const fromState = command.fromActiveSessionId ? this.getSessionState(command.fromActiveSessionId) : undefined; const receipt = await this.sendAgentSessionMessage({ targetSelector: command.targetActiveSessionId, message: command.message, fromState, clientId: client.id, senderKey: this.createCliAgentMessageSenderKey(), origin: command.agentOrigin === true ? "agent" : "cli", }); return success(command.id, "send_message", receipt); }
case "agent_messages_status": { return success(command.id, "agent_messages_status", this.getAgentMessageSafetyStatus()); }
case "agent_messages_pause": { this.agentMessagesPaused = true; this.agentMessageRateLimiter.clear(); await this.clearQueuedAgentSessionMessagesForAllStates(); return success(command.id, "agent_messages_pause", this.getAgentMessageSafetyStatus()); }
case "agent_messages_resume": { this.agentMessagesPaused = false; return success(command.id, "agent_messages_resume", this.getAgentMessageSafetyStatus()); }
case "agent_messages_clear": { const state = this.getSessionState(command.activeSessionId); this.agentMessageRateLimiter.clearMatching((key) => key.endsWith(`->${state.activeSessionId}`)); const cleared = state.runtime.session.clearQueuedAgentMessages(); return success(command.id, "agent_messages_clear", cleared); }
case "abort": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.requestAbort(); return success(command.id, "abort"); }
case "start_side_question": { const state = this.getSessionState(command.activeSessionId); if (this.sideQuestionRuns.has(command.sideQuestionId)) { throw new Error(`Side question already exists: ${command.sideQuestionId}`); } if (this.hasActiveSideQuestionFor(client, state.activeSessionId)) { throw new Error("A side question is already running for this client and session"); } const run = startSideQuestion( state.runtime.session.agent, command.sideQuestionId, command.question, (event) => { this.write(client, { type: "side_question_event", activeSessionId: state.activeSessionId, event, }); if (event.status !== "running") { this.sideQuestionRuns.delete(event.id); } }, command.previousTurns, providerRetryPolicy(state.runtime.session.settingsManager), ); this.sideQuestionRuns.set(command.sideQuestionId, { run, client, activeSessionId: state.activeSessionId, }); void run.done.catch((error) => { this.sideQuestionRuns.delete(command.sideQuestionId); this.log( `side question ${command.sideQuestionId} failed: ${error instanceof Error ? error.message : String(error)}`, ); }); return success(command.id, "start_side_question"); }
case "abort_side_question": { this.getSessionState(command.activeSessionId); const entry = this.sideQuestionRuns.get(command.sideQuestionId); if (!entry || entry.client !== client || entry.activeSessionId !== command.activeSessionId) { return success(command.id, "abort_side_question", { aborted: false }); } entry.run.abort(); return success(command.id, "abort_side_question", { aborted: true }); }
case "execute_bash": { const state = this.getSessionState(command.activeSessionId); if (state.runtime.session.isBashRunning) { throw new Error("A bash command is already running"); } // Respond before completion (bash can outlive the client request // timeout); output and completion stream via bash_* session events. const bash = state.runtime.session.runUserBash(command.command, { excludeFromContext: command.excludeFromContext, transient: command.transient, runId: command.runId, }); state.inFlightBash = Promise.allSettled([state.inFlightBash, bash]).then(() => undefined); void bash.catch((error) => { this.broadcastToSession(state, failure(undefined, "execute_bash", error, serializeDaemonError(error))); }); return success(command.id, "execute_bash"); }
case "execute_bash_and_wait": { const state = this.getSessionState(command.activeSessionId); const bash = state.runtime.session.executeBash(command.command); state.inFlightBash = Promise.allSettled([state.inFlightBash, bash]).then(() => undefined); try { return success(command.id, "execute_bash_and_wait", await bash); } finally { this.scheduleRosterFlush(); } }
case "abort_bash": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.abortBash(); return success(command.id, "abort_bash"); }
case "cancel_rlm_child": { const state = this.getSessionState(command.activeSessionId); const cancelled = state.runtime.session.cancelRlmChildRun(command.childId); return success(command.id, "cancel_rlm_child", { cancelled }); }
case "delete_rlm_subagent": { const state = this.getSessionState(command.activeSessionId); const isResidentChildRunning = () => { const childState = [...this.sessions.values()].find( (candidate) => candidate.runtime.metadata.kind === "subagent" && candidate.runtime.metadata.rlmChildId === command.childId, ); return ( childState !== undefined && (childState.runtime.session.isStreaming || childState.runtime.session.unfinishedActionCount > 0) ); }; const result = isResidentChildRunning() ? "running" : await state.runtime.session.deleteInactiveRlmSubagent(command.childId, isResidentChildRunning); return success(command.id, "delete_rlm_subagent", { deleted: result === "deleted", ...(result === "running" ? { reason: "running" } : {}), }); }
case "acquire_session_input_pause": { const existing = [...this.sessionInputPauses].find( ([, entry]) => entry.owner === client && entry.activeSessionId === command.activeSessionId && entry.leaseKey === command.leaseKey, ); if (existing) { return success(command.id, "acquire_session_input_pause", { pauseId: existing[0] }); } const state = this.getSessionState(command.activeSessionId); const pauseId = randomUUID(); this.sessionInputPauses.set(pauseId, { activeSessionId: command.activeSessionId, owner: client, leaseKey: command.leaseKey, pause: state.runtime.session.acquireSessionInputPause(), }); return success(command.id, "acquire_session_input_pause", { pauseId }); }
case "release_session_input_pause": { const entry = this.sessionInputPauses.get(command.pauseId); if (!entry) return success(command.id, "release_session_input_pause"); if (entry.owner !== client || entry.activeSessionId !== command.activeSessionId) { throw new Error(`Session input pause is owned by another client: ${command.pauseId}`); } this.sessionInputPauses.delete(command.pauseId); entry.pause.release(); return success(command.id, "release_session_input_pause"); }
case "wait_for_idle": { const state = this.getSessionState(command.activeSessionId); await state.runtime.session.waitForIdle(); return success(command.id, "wait_for_idle"); }
case "wait_for_headless_completion": { const state = this.getSessionState(command.activeSessionId); return success( command.id, "wait_for_headless_completion", await waitForHeadlessCompletion(state.runtime.session, { waitForRlmQuiescence: command.waitForRlmQuiescence, }), ); }
case "get_session_header": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_session_header", { header: state.runtime.session.sessionManager.getHeader(), }); }
case "get_state": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_state", summaryForActiveSession(state)); }
case "get_connection_state": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_connection_state", this.createConnectionState(state)); }
case "get_messages": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_messages", { messages: state.runtime.session.messages, }); }
case "get_rlm_children": { const state = this.getSessionState(command.activeSessionId); // Match the attach/snapshot path: merge passivated children so command // consumers see the same authoritative roster the snapshot advertises. // Capture the sequence before the awaited walk so the response cannot // claim freshness past the roster it returns. const eventSequence = state.lastEventSequence; return success(command.id, "get_rlm_children", { children: await this.buildRlmChildSnapshotsWithPassiveRlmSubagents(state), eventSequence, }); }
case "get_session_stats": { const state = this.getSessionState(command.activeSessionId); const stats: SessionStats = state.runtime.session.getSessionStats(); return success(command.id, "get_session_stats", stats); }
case "get_context_tree": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_context_tree", state.runtime.session.getContextTree()); }
case "get_commands": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_commands", { commands: await createAgentConnectionCommands(state.runtime.session), }); }
case "get_resource_snapshot": { const state = this.getSessionState(command.activeSessionId); return success( command.id, "get_resource_snapshot", createAgentConnectionResourceSnapshot(state.runtime.session), ); }
case "replace_acp_mcp_servers": { if (!command.ownerId) throw new Error("ACP MCP owner id is required"); const state = this.getSessionState(command.activeSessionId); if (!state.clients.has(client)) throw new Error("Daemon client is not attached to this session"); if (command.servers.length > 0 && state.runtime.session.isStreaming) { throw new Error("Cannot replace ACP MCP servers while the agent is running"); } const commandPause = command.servers.length > 0 ? state.runtime.session.acquireSessionInputPause() : undefined; try { let currentOwner = this.acpMcpOwners.get(state.activeSessionId); if (currentOwner?.release) { await currentOwner.release.catch(() => undefined); currentOwner = this.acpMcpOwners.get(state.activeSessionId); } const ownedByClient = currentOwner?.client === client && currentOwner.ownerId === command.ownerId; if (command.servers.length === 0 && !ownedByClient) { return success(command.id, "replace_acp_mcp_servers"); } if (command.servers.length === 0) { if (!currentOwner) return success(command.id, "replace_acp_mcp_servers"); const release = state.runtime.session.releaseAcpMcpServers(command.ownerId, currentOwner.serverNames); currentOwner.release = release; try { await release; } catch (error) { currentOwner.release = undefined; throw error; } if (this.acpMcpOwners.get(state.activeSessionId) === currentOwner) { this.acpMcpOwners.delete(state.activeSessionId); } return success(command.id, "replace_acp_mcp_servers"); }
if (currentOwner && !ownedByClient) { throw new Error("ACP MCP configuration is owned by another daemon client"); } const claim = { client, ownerId: command.ownerId, serverNames: [ ...new Set([ ...(currentOwner?.serverNames ?? []), ...command.servers.map((server) => server.name), ]), ], }; this.acpMcpOwners.set(state.activeSessionId, claim); const rollback = async (): Promise<void> => { try { await state.runtime.session.releaseAcpMcpServers(command.ownerId, claim.serverNames); } catch (error) { this.log(`failed to roll back ACP MCP config: ${String(error)}`); } if (this.acpMcpOwners.get(state.activeSessionId) === claim) { this.acpMcpOwners.delete(state.activeSessionId); } }; try { await withClientEnv(state.clientEnv, async () => state.runtime.session.replaceAcpMcpServers(command.servers, command.ownerId), ); } catch (error) { await rollback(); throw error; } if (!state.clients.has(client) || this.acpMcpOwners.get(state.activeSessionId) !== claim) { await rollback(); throw new Error("Daemon client detached during ACP MCP replacement"); } return success(command.id, "replace_acp_mcp_servers"); } finally { commandPause?.release(); } }
case "get_available_models": { const { session } = this.getSessionState(command.activeSessionId).runtime; const models = await session.modelRegistry.refreshAvailableModels(); session.refreshModelMetadata(); return success(command.id, "get_available_models", { models }); }
case "get_model_catalog": { const { session } = this.getSessionState(command.activeSessionId).runtime; const catalog = await session.modelRegistry.refreshModelCatalog(); session.refreshModelMetadata(); return success(command.id, "get_model_catalog", catalog); }
case "get_queue": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_queue", { steering: [...state.runtime.session.getSteeringMessagePreviews()], followUp: [...state.runtime.session.getFollowUpMessagePreviews()], }); }
case "mutate_queued_message": { const state = this.getSessionState(command.activeSessionId); const status = state.runtime.session.mutateQueuedMessage( command.lane, command.index, command.expectedText, command.mutation, ); return success(command.id, "mutate_queued_message", { status }); }
case "clear_queue": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "clear_queue", state.runtime.session.clearQueue()); }
case "abort_and_clear_queue": { const state = this.getSessionState(command.activeSessionId); const queue = state.runtime.session.clearQueue(); state.runtime.session.requestAbort(); return success(command.id, "abort_and_clear_queue", queue); }
case "cron_list": { const jobs = this.cronStore.list().filter((job) => { if (!command.includeInactive && job.status !== "active" && job.status !== "paused") { return false; } if (command.activeSessionId && job.activeSessionId !== command.activeSessionId) { return false; } return true; }); return success(command.id, "cron_list", { jobs }); }
case "heartbeats_list": return success(command.id, "heartbeats_list", { heartbeats: this.listHeartbeats(), });
case "heartbeat_manage": { const heartbeat = this.manageHeartbeat(command.activeSessionId, command.jobId, command.action); if (!heartbeat) { throw new Error(`No active heartbeat found: ${command.jobId}`); } return success(command.id, "heartbeat_manage", { heartbeat }); }
case "cron_add": { const state = this.getSessionState(command.activeSessionId); const job = this.createCronJobForState(state, command.schedule, command.prompt); this.scheduleRosterFlush(); return success(command.id, "cron_add", { job }); }
case "cron_cancel": { const job = this.cronStore.cancel(command.jobId); if (!job) { throw new Error(`No cron job found: ${command.jobId}`); } const state = this.sessions.get(job.activeSessionId); if (state) { this.removeQueuedHeartbeatFollowUp(state, job); } this.cronScheduler.wake(); this.scheduleRosterFlush(); return success(command.id, "cron_cancel", { job }); }
case "heartbeat_get": { const state = this.getSessionState(command.activeSessionId); const heartbeat = this.cronStore.getHeartbeat(state.activeSessionId); return success(command.id, "heartbeat_get", { heartbeat: heartbeat ?? null, }); }
case "heartbeat_set": { const state = this.getSessionState(command.activeSessionId); const deliveryMode = normalizeHeartbeatDeliveryMode(command.deliveryMode); const heartbeat = this.createHeartbeatForState(state, command.schedule, command.prompt, deliveryMode); return success(command.id, "heartbeat_set", { heartbeat }); }
case "heartbeat_update": { const state = this.getSessionState(command.activeSessionId); const heartbeat = this.updateHeartbeatForState(state, command.action); return success(command.id, "heartbeat_update", { heartbeat: heartbeat ?? null, }); }
case "set_model": { const state = this.getSessionState(command.activeSessionId); const session = state.runtime.session; const availableModels = await session.modelRegistry.refreshAvailableModels(); const model = availableModels.find( (candidate) => candidate.provider === command.provider && candidate.id === command.modelId, ) ?? // Stale-auth providers are excluded from the available list; the lookup // never mutates stale state (session.setModel owns the clear). (session.modelRegistry.getProviderAuthStatus(command.provider).source === "stale" ? session.modelRegistry.find(command.provider, command.modelId) : undefined); if (!model) { throw new Error(`Model not found: ${command.provider}/${command.modelId}`); } await session.setModel(model, { waitForExtensions: !(session.isStreaming || session.isCompacting), }); this.scheduleRosterFlush(); return success(command.id, "set_model", model); }
case "cycle_model": { const state = this.getSessionState(command.activeSessionId); const session = state.runtime.session; const result = await session.cycleModel(command.direction, { waitForExtensions: !(session.isStreaming || session.isCompacting), }); this.scheduleRosterFlush(); return success(command.id, "cycle_model", result ?? null); }
case "set_scoped_models": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setScopedModels(command.scopedModels); return success(command.id, "set_scoped_models"); }
case "set_thinking_level": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setThinkingLevel(command.level); return success(command.id, "set_thinking_level"); }
case "set_service_tier": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setServiceTier(command.serviceTier); return success(command.id, "set_service_tier"); }
case "cycle_thinking_level": { const state = this.getSessionState(command.activeSessionId); const level = state.runtime.session.cycleThinkingLevel(); return success(command.id, "cycle_thinking_level", level ? { level } : null); }
case "set_transport": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.settingsManager.setTransport(command.transport); state.runtime.session.agent.transport = command.transport; return success(command.id, "set_transport"); }
case "set_steering_mode": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setSteeringMode(command.mode); return success(command.id, "set_steering_mode"); }
case "set_follow_up_mode": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setFollowUpMode(command.mode); return success(command.id, "set_follow_up_mode"); }
case "set_auto_compaction": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setAutoCompactionEnabled(command.enabled); return success(command.id, "set_auto_compaction"); }
case "set_auto_retry": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.setAutoRetryEnabled(command.enabled); return success(command.id, "set_auto_retry"); }
case "compact": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.session.compact(command.customInstructions); return success(command.id, "compact", result); }
case "refine": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.session.refine({ instructions: command.instructions, rollbackId: command.rollbackId, global: command.global, }); return success(command.id, "refine", result); }
case "abort_compaction": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.abortCompaction(); return success(command.id, "abort_compaction"); }
case "abort_branch_summary": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.abortBranchSummary(); return success(command.id, "abort_branch_summary"); }
case "abort_retry": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.abortRetry(); return success(command.id, "abort_retry"); }
case "reload": { const state = this.getSessionState(command.activeSessionId); // Reload re-evaluates extension modules, which capture client env // (e.g. herdr pane identity) synchronously at load. await withClientEnv(state.clientEnv, () => state.runtime.session.reload()); return success(command.id, "reload"); }
case "new_session": { const state = this.getSessionState(command.activeSessionId); const options = command.parentSession ? { parentSession: command.parentSession } : undefined; const result = await state.runtime.newSession(options); this.rebindCronJobsToState(state); return success(command.id, "new_session", result); }
case "switch_session": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.switchSession(command.sessionPath, { cwdOverride: command.cwdOverride, }); this.rebindCronJobsToState(state); return success(command.id, "switch_session", result); }
case "fork": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.fork(command.entryId, { position: command.position, }); this.rebindCronJobsToState(state); return success(command.id, "fork", result); }
case "navigate_tree": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.session.navigateTree(command.targetId, { summarize: command.summarize, customInstructions: command.customInstructions, replaceInstructions: command.replaceInstructions, label: command.label, }); return success(command.id, "navigate_tree", result); }
case "import_jsonl": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.importFromJsonl(command.inputPath, command.cwdOverride); return success(command.id, "import_jsonl", result); }
case "export_html": { const state = this.getSessionState(command.activeSessionId); const path = await state.runtime.session.exportToHtml(command.outputPath); return success(command.id, "export_html", { path }); }
case "export_jsonl": { const state = this.getSessionState(command.activeSessionId); const path = state.runtime.session.exportToJsonl(command.outputPath); return success(command.id, "export_jsonl", { path }); }
case "set_session_name": { const state = this.getSessionState(command.activeSessionId); const name = command.name.trim(); if (!name) { throw new Error("Session name cannot be empty"); } await this.setStateSessionNameForCommand(state, name); return success(command.id, "set_session_name"); }
case "get_rlm_max_depth_status": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_rlm_max_depth_status", state.runtime.session.getRlmMaxDepthStatus()); }
case "set_rlm_max_depth": { const state = this.getSessionState(command.activeSessionId); const result = await state.runtime.session.setRlmMaxDepth(command.maxDepth, { global: command.global }); return success(command.id, "set_rlm_max_depth", result); }
case "get_session_context": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_session_context", { context: state.runtime.session.buildSessionContext(), }); }
case "get_session_tree": { const state = this.getSessionState(command.activeSessionId); const sessionManager = state.runtime.session.sessionManager; const leafId = sessionManager.getLeafId(); // A full a session tree carries every entry's content: images, tool // results, thinking blocks. The selector needs one line per entry, so // project here rather than serializing the transcript to the client. if (command.projection === "tree") { return success(command.id, "get_session_tree", { entries: projectSessionTreeFlatNodes(sessionManager.getFlatTree()), leafId, }); } return success(command.id, "get_session_tree", { flatNodes: sessionManager.getFlatTree(), leafId, }); }
case "get_user_messages_for_forking": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_user_messages_for_forking", { messages: state.runtime.session.getUserMessagesForForking(), }); }
case "get_last_assistant_text": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_last_assistant_text", { text: state.runtime.session.getLastAssistantText(), }); }
case "get_system_prompt": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_system_prompt", { systemPrompt: state.runtime.session.systemPrompt, }); }
case "get_tool_definition": { const state = this.getSessionState(command.activeSessionId); return success(command.id, "get_tool_definition", { toolDefinition: createAgentConnectionToolDefinition( state.runtime.session.getToolDefinition(command.name), ), }); }
case "set_session_entry_label": { const state = this.getSessionState(command.activeSessionId); state.runtime.session.sessionManager.appendLabelChange(command.entryId, command.label); return success(command.id, "set_session_entry_label"); }
case "extension_ui_response": { const state = this.getSessionState(command.activeSessionId); const pending = state.extensionUiRequests.get(command.requestId); if (!pending) { throw new Error(`Unknown extension UI request: ${command.requestId}`); } state.extensionUiRequests.delete(command.requestId); pending.resolve(command.response); return success(command.id, "extension_ui_response"); }
case "prepare_update_restart": this.log( `prepare_update_restart command received over socket; ${this.sessions.size} active session(s) will be closed`, ); return success(command.id, "prepare_update_restart", await this.prepareUpdateRestart());
case "retry_worker": throw new Error("Worker retry is only available through the daemon supervisor");
case "restart": setImmediate(() => { void this.shutdown(0); }); return success(command.id, command.type);
case "shutdown": this.log(`shutdown command received over socket; ${this.sessions.size} active session(s) will be closed`); setImmediate(() => { void this.shutdown(0); }); return success(command.id, "shutdown"); } }
private async createAttachResult( client: DaemonSocketClient, state: ActiveSessionState, command: Extract<DaemonCommand, { type: "attach" }>, ): Promise<DaemonAttachResult> { const snapshot = await this.createSessionSnapshot(state); const replay = command.resumeCursor?.activeSessionId && command.resumeCursor.activeSessionId !== state.activeSessionId ? { status: "unavailable" as const, fromSequence: "sequence" in command.resumeCursor ? command.resumeCursor.sequence : command.resumeCursor.eventSequence, toSequence: state.lastEventSequence, toCursor: { generation: state.eventGeneration, sequence: state.lastEventSequence, }, reason: "resume_cursor_session_mismatch", } : createDaemonReplayInfo(command.resumeCursor, state.lastEventSequence, state.eventGeneration); // Slim clients read summary/messages from the snapshot; duplicating them at // the top level would serialize the full history twice more per attach. const capabilities = daemonClientCapabilitiesForSession(client, state.activeSessionId); const slim = capabilities.has("slim_attach"); return { protocol: DAEMON_PROTOCOL_INFO, activeSessionId: state.activeSessionId, ...(slim ? {} : { state: snapshot.summary, messages: snapshot.messages }), snapshot, replay, lastEventSequence: state.lastEventSequence, lastEventCursor: { generation: state.eventGeneration, sequence: state.lastEventSequence, }, client: { id: client.id, capabilities: [...capabilities], }, }; }
private async createSessionSnapshot(state: ActiveSessionState): Promise<DaemonSessionSnapshot> { const metadata = state.runtime.metadata; const parent = metadata.parentActiveSessionId || metadata.parentSessionId || metadata.rlmParentNodeId || metadata.rlmChildId ? { ...(metadata.parentActiveSessionId ? { activeSessionId: metadata.parentActiveSessionId } : {}), ...(metadata.parentSessionId ? { sessionId: metadata.parentSessionId } : {}), ...(metadata.rlmParentNodeId ? { nodeId: metadata.rlmParentNodeId } : {}), ...(metadata.rlmChildId ? { childId: metadata.rlmChildId } : {}), } : undefined; let session = state.runtime.session; let children = await this.buildRlmChildSnapshotsWithPassiveRlmSubagents(state); for ( let retries = 0; retries < MAX_SESSION_SNAPSHOT_STABILIZATION_RETRIES && state.runtime.session !== session; retries++ ) { session = state.runtime.session; children = await this.buildRlmChildSnapshotsWithPassiveRlmSubagents(state); } session = state.runtime.session; const connectionState = this.createConnectionState(state); return { activeSessionId: state.activeSessionId, summary: summaryForActiveSession(state), state: connectionState, messages: session.messages, // Omit duplicate heavy payloads from attach. The client can derive render // context from messages + state, and fetch the full session tree lazily // when the tree/branch selector opens. lastEventSequence: state.lastEventSequence, lastEventCursor: { generation: state.eventGeneration, sequence: state.lastEventSequence, }, ...(parent ? { parent } : {}), children, }; }
private async streamWorkerSnapshot( client: DaemonSocketClient, result: DaemonAttachResult, transcript: SnapshotTranscriptChunkSource, purpose: "attach" | "replacement" | "catchup" = "attach", signal?: AbortSignal, snapshotAlreadyMarked = false, ): Promise<void> { const stream = result.snapshotStream; if (!stream) { if (snapshotAlreadyMarked) { finishClientSnapshotStreaming(client, result.activeSessionId); } transcript.dispose?.(); return; } if (snapshotAlreadyMarked && !signal) { throw new Error(`Snapshot ${stream.id} is missing its transfer signal`); } const transferSignal = signal ?? markClientSnapshotStreaming(client, result.activeSessionId); if (client.socket.destroyed) { finishClientSnapshotStreaming(client, result.activeSessionId); transcript.dispose?.(); return; } client.snapshotTransferTails ??= new Map(); const previousTransfer = client.snapshotTransferTails.get(result.activeSessionId); let finishTransfer!: () => void; const transfer = new Promise<void>((resolve) => { finishTransfer = resolve; }); client.snapshotTransferTails.set(result.activeSessionId, transfer); if (previousTransfer) { await previousTransfer; } const { messages: _messages, ...snapshot } = result.snapshot; const snapshotBegin: DaemonOutbound = { type: "session_snapshot_begin", activeSessionId: result.activeSessionId, snapshotId: stream.id, snapshot, messageCount: stream.messageCount, targetChunkBytes: stream.targetChunkBytes, purpose: purpose === "catchup" ? "resync" : purpose, }; const deliverSnapshotFailure = async (streamError: Error, includeBegin = false): Promise<void> => { transcript.markFailed?.(streamError); if (client.socket.destroyed) { return; } try { if ( includeBegin && !(await this.writeWorkerSnapshotRecord( client, snapshotBegin, purpose, undefined, WORKER_SNAPSHOT_TERMINAL_DRAIN_TIMEOUT_MS, )) ) { if (!client.socket.destroyed) { client.socket.destroy(streamError); } return; } const delivered = await this.writeWorkerSnapshotRecord( client, { type: "session_snapshot_failed", activeSessionId: result.activeSessionId, snapshotId: stream.id, error: streamError.message, }, purpose, undefined, WORKER_SNAPSHOT_TERMINAL_DRAIN_TIMEOUT_MS, ); if (!delivered && !client.socket.destroyed) { client.socket.destroy(streamError); } } catch (deliveryError) { client.socket.destroy(deliveryError instanceof Error ? deliveryError : new Error(String(deliveryError))); } }; try { if (transferSignal.aborted) { await deliverSnapshotFailure(new Error(`Snapshot ${stream.id} was aborted`), true); return; } if (!(await this.writeWorkerSnapshotRecord(client, snapshotBegin, purpose, transferSignal))) { if (transferSignal.aborted) { await deliverSnapshotFailure(new Error(`Snapshot ${stream.id} was aborted`)); } return; } let chunkCount = 0; for await (const chunk of transcript) { if (transferSignal.aborted) { await deliverSnapshotFailure(new Error(`Snapshot ${stream.id} was aborted`)); return; } const headerMessage: DaemonOutbound = { type: "session_snapshot_chunk", activeSessionId: result.activeSessionId, snapshotId: stream.id, index: chunkCount, messages: [], }; if (!(await this.writeWorkerSnapshotBuffer(client, chunk, headerMessage, purpose, transferSignal))) { if (transferSignal.aborted) { await deliverSnapshotFailure(new Error(`Snapshot ${stream.id} was aborted`)); } return; } chunkCount++; } if (transferSignal.aborted) { await deliverSnapshotFailure(new Error(`Snapshot ${stream.id} was aborted`)); return; } const snapshotDelivered = await this.writeWorkerSnapshotRecord( client, { type: "session_snapshot_end", activeSessionId: result.activeSessionId, snapshotId: stream.id, chunkCount, lastEventSequence: result.lastEventSequence, lastEventCursor: result.lastEventCursor, }, purpose, transferSignal, ); if (!snapshotDelivered) return; // A replay failure needs a resync; the snapshot is already delivered. try { this.flushDeferredSessionFrames(client, result.activeSessionId); } catch (error) { this.log(`could not replay deferred frames for ${result.activeSessionId}: ${String(error)}`); this.queueClientCatchup(client, result.activeSessionId, "resync"); } } catch (error) { if (transferSignal.aborted) { await deliverSnapshotFailure(new Error(`Snapshot ${stream.id} was aborted`)); return; } const streamError = error instanceof Error ? error : new Error(String(error)); await deliverSnapshotFailure(streamError); throw streamError; } finally { finishTransfer(); if (client.snapshotTransferTails.get(result.activeSessionId) === transfer) { client.snapshotTransferTails.delete(result.activeSessionId); } // A failed final stream must recover any events held behind an earlier snapshot. if ( (client.snapshotActiveSessionCounts?.get(result.activeSessionId) ?? 1) === 1 && client.deferredSessionOutbounds?.get(result.activeSessionId)?.frames.length && client.attachedActiveSessionIds.has(result.activeSessionId) ) { this.queueClientCatchup(client, result.activeSessionId, "resync"); } finishClientSnapshotStreaming(client, result.activeSessionId); transcript.dispose?.(); if (!client.snapshotStreaming && client.catchupActiveSessionIds?.size) { void this.catchUpBackpressuredClient(client).catch((error) => this.log(`could not catch up snapshot client ${client.id}: ${String(error)}`), ); } } }
private writeWorkerSnapshotRecord( client: DaemonSocketClient, message: DaemonOutbound, purpose: "attach" | "replacement" | "catchup", signal?: AbortSignal, drainTimeoutMs?: number, ): Promise<boolean> { return this.writeWorkerSnapshotBuffer( client, Buffer.from(serializeJsonLine(message)), message, purpose, signal, drainTimeoutMs, ); }
private async writeWorkerSnapshotBuffer( client: DaemonSocketClient, buffer: Buffer, message: DaemonOutbound, purpose: "attach" | "replacement" | "catchup", signal?: AbortSignal, drainTimeoutMs?: number, ): Promise<boolean> { if (signal?.aborted || client.socket.destroyed) { return false; } if (this.writeSerialized(client, buffer, message, "jsonl", purpose)) { return true; } return new Promise<boolean>((resolveDrain) => { let settled = false; let drainTimeout: NodeJS.Timeout | undefined; const finish = (value: boolean) => { if (settled) { return; } settled = true; client.socket.off("drain", onDrain); client.socket.off("close", onClose); client.socket.off("error", onClose); signal?.removeEventListener("abort", onAbort); if (drainTimeout) { clearTimeout(drainTimeout); } resolveDrain(value); }; const onDrain = () => finish(true); const onClose = () => finish(false); const onAbort = () => finish(false); client.socket.once("drain", onDrain); client.socket.once("close", onClose); client.socket.once("error", onClose); signal?.addEventListener("abort", onAbort, { once: true }); if (drainTimeoutMs !== undefined) { drainTimeout = setTimeout(() => finish(false), drainTimeoutMs); drainTimeout.unref(); } if (signal?.aborted || client.socket.destroyed) { finish(false); } }); }
private createConnectionState(state: ActiveSessionState): ReturnType<typeof createAgentConnectionState> { const connectionState = createAgentConnectionState(state.runtime, state.activeSessionId); connectionState.heartbeat = this.cronStore.getLatestHeartbeat(state.activeSessionId) ?? null; if (this.sessionRecapEnabled() && state.summaryState?.summary) { connectionState.recap = state.summaryState.summary; } return connectionState; }
private createAgentSessionMessageEndpoint(state: ActiveSessionState): AgentSessionMessageEndpoint { const metadata = state.runtime.metadata; return { activeSessionId: state.activeSessionId, sessionId: state.runtime.session.sessionId, ...(state.runtime.session.sessionName ? { sessionName: state.runtime.session.sessionName } : {}), runtimeKind: metadata.kind, }; }
private createAgentSessionMessageSender( state: ActiveSessionState | undefined, clientId: string, ): AgentSessionMessageSender { if (!state) { return { clientId }; } return { ...this.createAgentSessionMessageEndpoint(state), clientId, }; }
private createAgentMessageAgentSummary(state: ActiveSessionState): AgentSessionMessageAgentSummary { const metadata = state.runtime.metadata; const session = state.runtime.session; return { ...this.createAgentSessionMessageEndpoint(state), cwd: state.runtime.cwd, isStreaming: session.isStreaming, unfinishedActionCount: session.unfinishedActionCount, ...(metadata.parentActiveSessionId ? { parentActiveSessionId: metadata.parentActiveSessionId } : {}), ...(metadata.parentSessionId ? { parentSessionId: metadata.parentSessionId } : {}), ...(metadata.parentSessionFile ? { parentSessionPath: metadata.parentSessionFile } : {}), rlmDepth: session.rlmDepth, status: classifySessionRosterStatus({ activeSessionId: state.activeSessionId, runtimeKind: metadata.kind, activity: session.isSessionActive ? "working" : "idle", isSessionActive: session.isSessionActive, hasRunningRlmChildren: session.hasRunningRlmChildren?.() ?? false, isStreaming: session.isStreaming, } as SessionSummary), ...(metadata.rlmChildId ? { rlmChildId: metadata.rlmChildId } : {}), ...(metadata.sessionDir ? { sessionDir: metadata.sessionDir } : {}), ...(session.sessionFile ? { sessionPath: session.sessionFile } : {}), }; }
private async listSupervisorAgentPeers(): Promise<AgentSessionMessageAgentSummary[]> { const link = this.supervisorLink(); if (!this.options.worker || !link) return []; try { const response = await link.request( { type: "list_agent_peers", workerToken: this.options.worker.authenticationToken }, 5000, ); if (!response.success) throw deserializeDaemonError(response); // SAFETY: The authenticated supervisor constructs the peer response. return (response.data as { peers: AgentSessionMessageAgentSummary[] }).peers; } catch { return []; } }
private async createAgentMessageListResult( current: ActiveSessionState, peers?: AgentSessionMessageAgentSummary[], ): Promise<AgentSessionMessageListResult> { peers ??= await this.listSupervisorAgentPeers(); const localAgents = this.listTargetableSessionStates(current).map((state) => this.createAgentMessageAgentSummary(state), ); for (const passive of await this.listPassiveRlmSubagents()) { const { entry, info } = passive; localAgents.push({ // Before hydration the persisted session id is its supervisor-routable id. activeSessionId: info.id, sessionId: info.id, sessionName: info.name ?? entry.sessionName, runtimeKind: "subagent", cwd: info.cwd, isStreaming: false, unfinishedActionCount: 0, ...(passive.chain.length === 1 && passive.rootParentState ? { parentActiveSessionId: passive.rootParentState.activeSessionId } : {}), parentSessionId: entry.parentSessionId, parentSessionPath: entry.parentSessionFile ?? passive.chain.at(-2)?.sessionFile ?? passive.rootParentState?.runtime.session.sessionFile ?? passive.rootInfo?.path, rlmDepth: info.rlmDepth ?? entry.rlmDepth, status: "inactive", rlmChildId: entry.childId, rlmChildRegistryStatus: entry.status, sessionDir: entry.sessionDir, sessionPath: entry.sessionFile, }); } const localIds = new Set(localAgents.map((agent) => agent.activeSessionId)); const remoteAgents = peers.filter( (peer) => !localIds.has(peer.activeSessionId) && !this.closingSessions.has(peer.activeSessionId), ); return { current: this.createAgentSessionMessageEndpoint(current), agents: [...localAgents, ...remoteAgents], }; }
private async createAgentFamilyCatalog(currentState?: ActiveSessionState): Promise<AgentFamilyCatalogEntry[]> { const current = currentState ?? [...this.sessions.values()].find((state) => !this.bindingSessions.has(state.activeSessionId)); const remotePeers = current ? await this.listSupervisorAgentPeers() : []; const listed = current ? await this.createAgentMessageListResult(current, remotePeers) : { agents: [] }; const remotePeerSet = new Set(remotePeers); const localAgents = current ? [this.createAgentMessageAgentSummary(current), ...listed.agents.filter((agent) => !remotePeerSet.has(agent))] : listed.agents; const activePaths = new Set( localAgents.flatMap((agent) => (agent.sessionPath ? [canonicalSessionPath(agent.sessionPath)] : [])), ); const savedRoots = (await SessionManager.listAll(undefined, this.options.defaultSessionConfig.sessionDir)) .filter( (info) => (info.rlmDepth ?? (info.parentSessionPath ? -1 : 0)) === 0 && !activePaths.has(canonicalSessionPath(info.path)), ) .map( (info): AgentFamilyCatalogEntry => ({ id: info.id, ...(info.name ? { name: info.name } : {}), depth: info.rlmDepth ?? 0, status: "inactive", sessionPath: canonicalSessionPath(info.path), cwd: info.cwd, messageCount: info.messageCount, }), ); const byId = new Map<string, AgentFamilyCatalogEntry>(savedRoots.map((entry) => [entry.id, entry])); // `remote` peers live in another worker: their active id stays routable, while a // local summary's stand-in id for a passive child does not. const addAgent = (agent: AgentSessionMessageAgentSummary, remote = false) => { const depth = agent.rlmDepth ?? 0; byId.set(agent.sessionId, { id: agent.sessionId, ...(agent.sessionName ? { name: agent.sessionName } : {}), depth, status: agent.status ?? "idle", ...(agent.runtimeKind === "subagent" ? (() => { const repliedSinceTask = this.findSessionBySessionFile(agent.sessionPath)?.runtime.session .repliedToParentSinceTask; return repliedSinceTask === undefined ? {} : { repliedSinceTask }; })() : {}), ...(depth > 0 && agent.parentSessionId ? { parentSessionId: agent.parentSessionId } : {}), ...(depth > 0 && agent.parentSessionPath ? { parentSessionPath: canonicalSessionPath(agent.parentSessionPath) } : {}), ...(agent.sessionPath ? { sessionPath: canonicalSessionPath(agent.sessionPath) } : {}), ...(agent.rlmChildId ? { rlmChildId: agent.rlmChildId } : {}), ...(remote ? { activeSessionId: agent.activeSessionId } : {}), cwd: agent.cwd, }); }; for (const peer of remotePeers) addAgent(peer, true); for (const agent of localAgents) addAgent(agent); for (const state of this.sessions.values()) { const entry = byId.get(state.runtime.session.sessionId); if (entry && state.runtime.session.sessionFile) entry.sessionPath = canonicalSessionPath(state.runtime.session.sessionFile); } for (const passive of await this.listPassiveRlmSubagents()) { const entry = byId.get(passive.info.id); if (!entry) continue; entry.sessionPath = canonicalSessionPath(passive.entry.sessionFile); entry.messageCount = passive.info.messageCount; entry.firstMessage = passive.info.firstMessage; } return [...byId.values()]; }
private async createAgentFamily(currentState: ActiveSessionState): Promise<AgentFamilyMember[]> { const catalog = await this.createAgentFamilyCatalog(currentState); const current = catalog.find((entry) => entry.id === currentState.runtime.session.sessionId); if (!current) throw new Error("Current agent is missing from the family catalog"); return selectAgentFamily(current, catalog); }
private async assertFamilySessionNameAvailable( input: { name: string; depth: number; parentSessionId?: string; parentSessionPath?: string; ignoreSessionId?: string; }, currentState?: ActiveSessionState, ignorePendingReservation = false, ): Promise<void> { if (!ignorePendingReservation && this.pendingSessionNames.has(sessionNameReservationKey(input))) { throw new Error(formatAgentSessionNameUnavailable(input.name, input.depth)); } assertAgentSessionNameAvailable(await this.createAgentFamilyCatalog(currentState), { ...input, ...(input.parentSessionPath ? { parentSessionPath: canonicalSessionPath(input.parentSessionPath) } : {}), }); }
private resolveHeaderParentSessionPath(state: ActiveSessionState): string | undefined { const session = state.runtime.session; const headerParent = session.sessionManager?.getHeader?.()?.parentSession; if (!headerParent || isAbsolute(headerParent)) return headerParent; return session.sessionFile ? resolve(dirname(session.sessionFile), headerParent) : undefined; }
private async assertStateSessionNameAvailable(state: ActiveSessionState, name: string): Promise<void> { const session = state.runtime.session; const metadata = state.runtime.metadata; const depth = session.rlmDepth ?? 0; const headerParent = depth > 0 ? this.resolveHeaderParentSessionPath(state) : undefined; await this.assertFamilySessionNameAvailable( { name, depth, ...(depth > 0 && !headerParent && metadata.parentSessionId ? { parentSessionId: metadata.parentSessionId } : {}), ...(depth > 0 && (headerParent ?? metadata.parentSessionFile) ? { parentSessionPath: headerParent ?? metadata.parentSessionFile } : {}), ignoreSessionId: session.sessionId, }, state, true, ); }
private async withSessionNameReservation<T>( input: { name: string; depth: number; parentSessionId?: string; parentSessionPath?: string }, action: () => Promise<T>, ): Promise<T> { const key = sessionNameReservationKey(input); if (this.pendingSessionNames.has(key)) { throw new Error(formatAgentSessionNameUnavailable(input.name, input.depth)); } this.pendingSessionNames.add(key); try { return await action(); } finally { this.pendingSessionNames.delete(key); } }
private async setStateSessionNameViaSupervisor(state: ActiveSessionState, name: string): Promise<void> { const link = this.supervisorLink(); if (!this.options.worker || !link) { return this.setStateSessionName(state, name); } const response = await link.request( { type: "set_session_name", activeSessionId: state.activeSessionId, name, workerToken: this.options.worker.authenticationToken, }, 30_000, ); if (!response.success) throw deserializeDaemonError(response); }
private setStateSessionNameForCommand(state: ActiveSessionState, name: string): Promise<void> { return this.options.worker ? this.applyStateSessionName(state, name) : this.setStateSessionName(state, name); }
private async applyStateSessionName(state: ActiveSessionState, name: string): Promise<void> { state.runtime.session.setSessionName(name); await this.appendRlmLedgerRenameForState(state, name); }
private async setStateSessionName(state: ActiveSessionState, name: string): Promise<void> { const normalizedName = name.trim(); if (!normalizedName) { throw new Error("Session name cannot be empty"); } const session = state.runtime.session; const metadata = state.runtime.metadata; const depth = session.rlmDepth ?? 0; const headerParent = depth > 0 ? this.resolveHeaderParentSessionPath(state) : undefined; return this.withSessionNameReservation( { name: normalizedName, depth, ...(depth > 0 && !headerParent && metadata.parentSessionId ? { parentSessionId: metadata.parentSessionId } : {}), ...(depth > 0 && (headerParent ?? metadata.parentSessionFile) ? { parentSessionPath: headerParent ?? metadata.parentSessionFile } : {}), }, async () => { await this.assertStateSessionNameAvailable(state, normalizedName); await this.applyStateSessionName(state, normalizedName); }, ); }
// Half-bound sessions are hidden from other sessions' listings; the current // session stays visible to itself (controllers run during its own bind). private listTargetableSessionStates(current: ActiveSessionState): ActiveSessionState[] { return [...this.sessions.values()].filter( (state) => state.activeSessionId === current.activeSessionId || (!this.bindingSessions.has(state.activeSessionId) && !this.closingSessions.has(state.activeSessionId)), ); }
private getAgentMessageSafetyStatus() { return { paused: this.agentMessagesPaused, maxMessageChars: DEFAULT_AGENT_MESSAGE_MAX_CHARS, maxPendingPerSession: DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, rateLimitCapacity: DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY, rateLimitRefillMs: DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS, }; }
private createCliAgentMessageSenderKey(): string { return `cli:${this.socketPath}`; }
private async clearQueuedAgentSessionMessagesForState(state: ActiveSessionState) { return state.runtime.session.clearQueuedAgentMessages(); }
private async clearQueuedAgentSessionMessagesForAllStates(): Promise<void> { await Promise.all( [...this.sessions.values()].map((state) => this.clearQueuedAgentSessionMessagesForState(state)), ); }
private agentFamilyEntry(state: ActiveSessionState): AgentFamilyCatalogEntry { const metadata = state.runtime.metadata; const depth = state.runtime.session.rlmDepth ?? 0; const headerParent = depth > 0 ? this.resolveHeaderParentSessionPath(state) : undefined; const parentSessionPath = depth > 0 ? (headerParent ?? metadata.parentSessionFile) : undefined; return { id: state.runtime.session.sessionId, ...(state.runtime.session.sessionName ? { name: state.runtime.session.sessionName } : {}), depth, status: "running", ...(depth > 0 && !headerParent && metadata.parentSessionId ? { parentSessionId: metadata.parentSessionId } : {}), ...(parentSessionPath ? { parentSessionPath: canonicalSessionPath(parentSessionPath) } : {}), ...(state.runtime.session.sessionFile ? { sessionPath: canonicalSessionPath(state.runtime.session.sessionFile) } : {}), }; }
private passiveAgentFamilyEntry(passive: PassiveRlmSubagent): AgentFamilyCatalogEntry { const entry = passive.entry; const depth = passive.info.rlmDepth ?? entry.rlmDepth ?? passive.chain.length; const parentSessionPath = depth > 0 ? (entry.parentSessionFile ?? passive.chain.at(-2)?.sessionFile ?? passive.rootParentState?.runtime.session.sessionFile ?? passive.rootInfo?.path) : undefined; return { id: passive.info.id, name: passive.info.name ?? entry.sessionName, depth, status: "idle", ...(depth > 0 && entry.parentSessionId ? { parentSessionId: entry.parentSessionId } : {}), ...(parentSessionPath ? { parentSessionPath: canonicalSessionPath(parentSessionPath) } : {}), sessionPath: canonicalSessionPath(entry.sessionFile), }; }
private async getOrHydrateAuthorizedAgentFamilyTarget( currentState: ActiveSessionState, target: string, ): Promise<ActiveSessionState> { try { return this.getBoundSessionState(target); } catch (error) { if (error instanceof BoundSessionUnavailableError) { const targetState = this.getSessionState(target); this.assertAgentFamilyReachable(currentState, targetState); return this.getOrHydrateBoundSessionState(target); } if (error instanceof AmbiguousActiveSessionError) { const targetState = this.resolveAgentFamilySessionName(currentState, target, error); return this.getOrHydrateBoundSessionState(targetState.activeSessionId); } } const passive = await this.findPassiveRlmSubagent(target); if (!passive) return this.getOrHydrateBoundSessionState(target); assertAgentFamilyReach(this.agentFamilyEntry(currentState), this.passiveAgentFamilyEntry(passive)); return this.hydratePassiveRlmSubagent(passive); }
private resolveAgentFamilySessionName( currentState: ActiveSessionState, target: string, ambiguity: AmbiguousActiveSessionError, ): ActiveSessionState { const reachableMatches = new Map( [...this.sessions.values()] .filter((state) => { const session = state.runtime.session; return ( (session.sessionId === target || session.sessionName === target) && (state.activeSessionId === currentState.activeSessionId || this.isAgentFamilyReachable(currentState, state)) ); }) .map((state) => [state.activeSessionId, state]), ).values(); const matches = [...reachableMatches]; if (matches.length !== 1) throw ambiguity; return matches[0]!; }
private isAgentFamilyReachable(currentState: ActiveSessionState, targetState: ActiveSessionState): boolean { try { assertAgentFamilyReach(this.agentFamilyEntry(currentState), this.agentFamilyEntry(targetState)); return true; } catch (error) { if (error instanceof Error && error.message === AGENT_FAMILY_REACH_ERROR) return false; throw error; } }
private assertAgentFamilyReachable(currentState: ActiveSessionState, targetState: ActiveSessionState): void { if (currentState.activeSessionId === targetState.activeSessionId) return; assertAgentFamilyReach(this.agentFamilyEntry(currentState), this.agentFamilyEntry(targetState)); }
private agentMessageRelationship( fromState: ActiveSessionState | undefined, targetState: ActiveSessionState, ): AgentFamilyRelationship | undefined { if (!fromState) return undefined; return agentFamilyRelationship(this.agentFamilyEntry(targetState), this.agentFamilyEntry(fromState)); }
private async sendAgentSessionMessage(options: { targetSelector: string; message: string; fromState?: ActiveSessionState; sender?: AgentSessionMessageSender; clientId?: string; senderKey?: string; origin: "agent" | "cli"; }): Promise<AgentSessionMessageReceipt> { if (this.agentMessagesPaused) { throw new Error("Agent messaging is paused"); } const targetSelector = assertDirectAgentMessageTarget(options.targetSelector); const message = normalizeAgentSessionMessage(options.message, DEFAULT_AGENT_MESSAGE_MAX_CHARS); let targetState: ActiveSessionState; try { targetState = this.getBoundSessionState(targetSelector); } catch (error) { if (error instanceof BoundSessionUnavailableError) { if (options.origin === "agent" && options.fromState) { this.assertAgentFamilyReachable(options.fromState, this.getSessionState(targetSelector)); } targetState = await this.getOrHydrateBoundSessionState(targetSelector); } else { if (error instanceof AmbiguousActiveSessionError) { if (options.origin !== "agent" || !options.fromState) throw error; const resolved = this.resolveAgentFamilySessionName(options.fromState, targetSelector, error); targetState = await this.getOrHydrateBoundSessionState(resolved.activeSessionId); } else { const passiveSubagent = await this.findPassiveRlmSubagent(targetSelector); if (passiveSubagent) { if (options.origin === "agent" && options.fromState) { assertAgentFamilyReach( this.agentFamilyEntry(options.fromState), this.passiveAgentFamilyEntry(passiveSubagent), ); } targetState = await this.hydratePassiveRlmSubagent(passiveSubagent); } else { const hydratingChild = [...this.sessions.values()].find( (state) => state.runtime.metadata.kind === "subagent" && state.runtime.metadata.rlmChildId === targetSelector, ); if (hydratingChild) { targetState = await this.waitForHydratingChild(hydratingChild, targetSelector); } else if (this.options.worker && options.fromState) { // The supervisor can resolve and wake a saved worker even when it is no longer // present in this worker's resident peer snapshot. return this.sendRemoteAgentSessionMessage(options.fromState, targetSelector, message); } else { throw error; } } } } } if (options.fromState?.activeSessionId === targetState.activeSessionId) { throw new Error("Agent messaging cannot target the sending session"); } if (options.origin === "agent" && options.fromState) { this.assertAgentFamilyReachable(options.fromState, targetState); } const senderKey = options.senderKey ?? options.fromState?.activeSessionId ?? `client:${options.clientId ?? "unknown"}`; const rateLimitKey = `${senderKey}->${targetState.activeSessionId}`; const rateLimit = this.agentMessageRateLimiter.tryConsume(rateLimitKey); if (!rateLimit.ok) { throw new Error(`Agent messaging rate limit exceeded; retry after ${rateLimit.retryAfterMs}ms`); } const payload: AgentSessionMessagePayload = { id: createAgentSessionMessageId(), source: AGENT_MESSAGE_SOURCE, message, from: options.sender ?? this.createAgentSessionMessageSender(options.fromState, options.clientId ?? options.origin), fromRelationship: this.agentMessageRelationship(options.fromState, targetState), target: this.createAgentSessionMessageEndpoint(targetState), }; // Bounded backpressure: a saturated recipient is a reason to yield, not to fail. The // sender waits while the target drains, then delivers anyway (the message stays // buffered in memory), so fan-in cannot be lost at the moment results arrive. await this.waitForAgentMessageCapacity(targetState); try { const { status } = await this.acceptAgentSessionMessage(targetState, payload); return createAgentSessionMessageReceipt(payload, status); } catch (error) { this.agentMessageRateLimiter.refund(rateLimitKey); throw error; } }
private async waitForAgentMessageCapacity(targetState: ActiveSessionState): Promise<void> { const deadline = Date.now() + AGENT_MESSAGE_BACKPRESSURE_WAIT_MS; while (Date.now() < deadline && !this.shuttingDown) { // A session that cannot report a depth (a stub, a passive child) is never treated as // saturated: waiting on an unknown would stall delivery for the whole window. const unfinished = targetState.runtime.session.unfinishedActionCount ?? 0; if (!(unfinished >= DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION)) return; await new Promise((resolveDelay) => setTimeout(resolveDelay, 250)); } }
private async sendRemoteAgentSessionMessage( fromState: ActiveSessionState, targetSelector: string, message: string, ): Promise<AgentSessionMessageReceipt> { const link = this.supervisorLink(); if (!link) { throw new Error(`Unknown active session: ${targetSelector}`); } const deadline = Date.now() + 30_000; let lastError: unknown; // Connect-window loop only: the supervisor may still be starting up. // The message itself is sent exactly once - daemon commands are not // idempotent, so a rejected or timed-out send must never be retried. while (Date.now() < deadline && !this.shuttingDown) { try { await link.ensureConnected(); lastError = undefined; break; } catch (error) { lastError = error; await new Promise((resolveDelay) => setTimeout(resolveDelay, 250)); } } if (lastError !== undefined) { throw lastError instanceof Error ? lastError : new Error(`Unknown active session: ${targetSelector}`); } const response = await link.request( { type: "send_message", targetActiveSessionId: targetSelector, message, fromActiveSessionId: fromState.activeSessionId, agentOrigin: true, }, 30_000, ); if (!response.success) { throw deserializeDaemonError(response); } if (!response.data || typeof response.data !== "object") { throw new Error("Supervisor returned an invalid agent-message receipt"); } return response.data as AgentSessionMessageReceipt; }
private async acceptAgentSessionMessage( targetState: ActiveSessionState, payload: AgentSessionMessagePayload, ): Promise<{ status: AgentSessionMessageDeliveryStatus }> { const message = createAgentSessionMessage(payload); let preflightFailed = false; let preflightQueued = false; await targetState.runtime.session.acceptAgentMessagePrompt(message.content, { expandPromptTemplates: false, streamingBehavior: "steer", queueIfBusy: true, customMessage: message, admissionCommitted: () => { if (this.agentMessagesPaused) { throw new Error("Agent messaging is paused"); } if ( !this.sessions.has(targetState.activeSessionId) || this.closingSessions.has(targetState.activeSessionId) ) { throw new Error("Target session is closing before agent message delivery"); } if (targetState.runtime.session.sessionId !== payload.target.sessionId) { throw new Error("Target session changed before agent message delivery"); } }, preflightResult: (didSucceed, didQueue) => { preflightFailed = !didSucceed; preflightQueued = didSucceed && didQueue === true; }, }); if (preflightFailed) { throw new Error("Agent message was not accepted"); } return { status: preflightQueued ? "queued" : "delivered" }; }
private detachClientFromSession(client: DaemonSocketClient, state: ActiveSessionState): void { const acpMcpOwner = this.acpMcpOwners.get(state.activeSessionId); if (acpMcpOwner?.client === client && !acpMcpOwner.release) { const release = state.runtime.session.releaseAcpMcpServers(acpMcpOwner.ownerId, acpMcpOwner.serverNames); acpMcpOwner.release = release; void release .catch((error) => this.log(`failed to release detached ACP MCP config: ${String(error)}`)) .finally(() => { if (this.acpMcpOwners.get(state.activeSessionId) === acpMcpOwner) { this.acpMcpOwners.delete(state.activeSessionId); } }); } this.abortSideQuestionsFor(client, state.activeSessionId); abortClientSnapshotStreaming(client, state.activeSessionId); detachClientFromActiveSession(client, state); if (client.authenticationRole === "session_client") this.scheduleRosterFlush(); this.write(client, { type: "session_detached", activeSessionId: state.activeSessionId, }); // Discard an abandoned empty draft rather than retaining an empty session file. // Abandoned new-chat: discard it so it doesn't linger in memory or leave an // empty file. Replaces the old DeferredAgentConnection. if (this.isDiscardableDraft(state)) { // Re-check after yielding: a client may reattach before the async close // runs, in which case the draft is no longer abandoned and must be kept. queueMicrotask(() => { if (this.sessions.has(state.activeSessionId) && this.isDiscardableDraft(state)) { this.discardAbandonedDraft(state); } }); } }
/** Best-effort discard of an abandoned empty draft: teardown failures are logged, never fatal. */ private discardAbandonedDraft(state: ActiveSessionState): void { // closeSession rethrows teardown failures; an unobserved rejection here // would reach the daemon's unhandledRejection handler and exit the // process, killing every other hosted session with it. void this.closeSession(state, "killed").catch((error) => { this.log( `failed to discard abandoned empty draft ${state.activeSessionId}: ${ error instanceof Error ? (error.stack ?? error.message) : String(error) }`, ); }); }
private isDiscardableDraft(state: ActiveSessionState): boolean { if (this.options.worker) { return false; } if (state.clients.size > 0) { return false; } // An attach increments pendingAttaches before its client joins state.clients. // Count both, like the eviction snapshot, so a discard can't race an // in-flight attach and kill the draft it is about to claim. if (state.pendingAttaches > 0) { return false; } if (state.runtime.metadata.kind === "subagent") { return false; } if (state.runtime.session.isBashRunning || hasLiveSessionWork(state)) { return false; } return this.isEmptyDraftContent(state); }
/** * True when a session holds nothing worth persisting: no messages, no user * config (model/name/etc.), and no scheduled jobs. Shared by the detach-time * discard and the close-time file deletion so both agree on what an abandoned * draft is. */ private isEmptyDraftContent(state: ActiveSessionState): boolean { const session = state.runtime.session; if (session.messages.length > 0 || session.sessionManager.hasUserContent()) { return false; } return !this.hasScheduledJobsForSession(state.activeSessionId); }
private createUpdateRestartSession(state: ActiveSessionState): DaemonUpdateRestartSession | undefined { const session = state.runtime.session; const queue = { actions: session.getSessionActionRecoverySnapshot(), nextTurn: [...session.getPendingNextTurnMessageSnapshots()], }; const hasQueuedMessages = queue.actions.actions.length > 0 || queue.nextTurn.length > 0; const wasStreaming = session.isStreaming; const wasCompacting = session.isCompacting; const wasBashRunning = session.isBashRunning; const hadRunningRlmChildren = session.hasRunningRlmChildren(); const wasRetrying = session.isRetrying; const hadAcceptedPromptInFlight = session.hasAcceptedPromptInFlight; const shouldResume = wasStreaming || wasCompacting || wasBashRunning || hadRunningRlmChildren || wasRetrying || hadAcceptedPromptInFlight || queue.actions.actions.length > 0; const sessionFile = session.sessionFile ?? (hasQueuedMessages || shouldResume ? session.sessionManager.materializeSessionFile( state.runtime.runtimeConfig?.sessionDir ?? this.options.defaultSessionConfig.sessionDir, ) : undefined); if (!sessionFile || (this.isEmptyDraftContent(state) && !hasQueuedMessages && !shouldResume)) { return undefined; } return { activeSessionId: state.activeSessionId, sessionId: session.sessionId, sessionFile, cwd: session.sessionManager.getCwd(), config: { ...state.runtime.runtimeConfig, cwd: session.sessionManager.getCwd(), }, runtimeMetadata: state.runtime.metadata, ...(state.clientEnv ? { clientEnv: { ...state.clientEnv } } : {}), queue, shouldResume, wasStreaming, wasCompacting, wasBashRunning, hadRunningRlmChildren, wasRetrying, hadAcceptedPromptInFlight, }; }
private appendUpdateRestartMarker(state: ActiveSessionState, restartSession: DaemonUpdateRestartSession): void { if (!restartSession.shouldResume) { return; } state.runtime.session.sessionManager.appendCustomMessageEntry( "prime-agent.update_restart", UPDATE_RESTART_MARKER, false, { activeSessionId: restartSession.activeSessionId, wasStreaming: restartSession.wasStreaming, wasCompacting: restartSession.wasCompacting, wasBashRunning: restartSession.wasBashRunning, hadRunningRlmChildren: restartSession.hadRunningRlmChildren, wasRetrying: restartSession.wasRetrying, hadAcceptedPromptInFlight: restartSession.hadAcceptedPromptInFlight, }, ); }
private writeUpdateRestartManifest(manifest: DaemonUpdateRestartManifest): void { const path = getDaemonUpdateRestartManifestPath(this.socketPath, this.agentDir); mkdirSync(dirname(path), { recursive: true }); writeFileSync(path, `${JSON.stringify(manifest)}\n`); }
private getUpdateRestartSessionDepth(state: ActiveSessionState): number { let depth = 0; let metadata: AgentSessionRuntimeMetadata = state.runtime.metadata; const seen = new Set<string>([state.activeSessionId]); while (metadata.parentActiveSessionId && !seen.has(metadata.parentActiveSessionId)) { const parent = this.sessions.get(metadata.parentActiveSessionId); if (!parent) { break; } seen.add(parent.activeSessionId); depth++; metadata = parent.runtime.metadata; } return depth; }
private assertUpdateRestartNotCancelled(transaction: { id: symbol; abort: AbortController }): void { if (transaction.abort.signal.aborted || this.updateRestart?.id !== transaction.id) { throw new Error("Update restart preparation cancelled"); } }
private beginUpdateRestartTransaction(owner?: DaemonSocketClient): NonNullable<AgentDaemon["updateRestart"]> { if (this.updateRestart) throw new Error("Daemon is already preparing an update restart"); const transaction: NonNullable<AgentDaemon["updateRestart"]> = { id: Symbol("update-restart"), ...(owner ? { owner } : {}), abort: new AbortController(), phase: "preparing", deferredClientEnv: [], }; this.updateRestart = transaction; this.cronScheduler.stop(); transaction.deadline = setTimeout(() => { transaction.abort.abort(); this.cancelPreparedUpdateRestart(transaction.id); }, UPDATE_RESTART_PREPARE_TIMEOUT_MS); transaction.deadline.unref(); return transaction; }
private async runUpdateRestartPreparation( transaction: NonNullable<AgentDaemon["updateRestart"]>, ): Promise<DaemonUpdateRestartManifest> { try { await this.mutationDrain.waitForDrain(0, transaction.abort.signal, "Update restart preparation cancelled"); this.assertUpdateRestartNotCancelled(transaction); transaction.phase = "fencing"; await this.mutationDrain.waitForDrain(0, transaction.abort.signal, "Update restart preparation cancelled"); this.assertUpdateRestartNotCancelled(transaction); const manifest = await this.prepareUpdateRestartCheckpoint(transaction); this.assertUpdateRestartNotCancelled(transaction); transaction.manifest = manifest; transaction.phase = "prepared"; return manifest; } catch (error) { this.cancelPreparedUpdateRestart(transaction.id); throw error; } }
private async prepareUpdateRestartCheckpoint( transaction: NonNullable<AgentDaemon["updateRestart"]>, ): Promise<DaemonUpdateRestartManifest> { const signal = transaction.abort.signal; try { const states = [...this.sessions.values()]; this.updateRestartQueuePauses.clear(); for (const state of states) { this.updateRestartQueuePauses.set(state.activeSessionId, state.runtime.session.acquireQueuedWorkPause()); } await Promise.all(states.map((state) => state.runtime.session.waitForSessionInputCheckpoint(signal))); this.assertUpdateRestartNotCancelled(transaction); const snapshottedIds = new Set(states.map((state) => state.activeSessionId)); const addedSession = [...this.sessions.keys()].find((activeSessionId) => !snapshottedIds.has(activeSessionId)); if (addedSession) throw new Error(`Session ${addedSession} became resident during update preparation`);
const restartSessions = states .filter((state) => this.sessions.get(state.activeSessionId) === state) .map((state) => this.createUpdateRestartSession(state)) .filter((session): session is DaemonUpdateRestartSession => session !== undefined) .sort((left, right) => { const leftState = this.sessions.get(left.activeSessionId); const rightState = this.sessions.get(right.activeSessionId); return ( (leftState ? this.getUpdateRestartSessionDepth(leftState) : 0) - (rightState ? this.getUpdateRestartSessionDepth(rightState) : 0) ); }); this.assertUpdateRestartNotCancelled(transaction); const includedActiveSessionIds = new Set(restartSessions.map((session) => session.activeSessionId)); const discardedActiveSessionIds = states .filter( (state) => this.sessions.get(state.activeSessionId) === state && !includedActiveSessionIds.has(state.activeSessionId), ) .map((state) => state.activeSessionId); return { formatVersion: DAEMON_UPDATE_RESTART_FORMAT_VERSION, createdAt: new Date().toISOString(), sessions: restartSessions, ...(discardedActiveSessionIds.length > 0 ? { discardedActiveSessionIds } : {}), }; } catch (error) { this.cancelPreparedUpdateRestart(transaction.id); throw error; } }
private async commitPreparedUpdateRestart(transactionId: symbol): Promise<DaemonUpdateRestartManifest> { const transaction = this.updateRestart; if (transaction?.id !== transactionId || !transaction.manifest) { throw new Error("Daemon has no prepared update checkpoint"); } const manifest = transaction.manifest; const restartByActiveSessionId = new Map(manifest.sessions.map((session) => [session.activeSessionId, session])); const closeStates = [...this.sessions.values()].sort( (left, right) => this.getUpdateRestartSessionDepth(right) - this.getUpdateRestartSessionDepth(left), ); for (const state of closeStates) { if (this.sessions.has(state.activeSessionId)) { await this.closeSession(state, restartByActiveSessionId.has(state.activeSessionId) ? "update" : "killed"); } } for (const state of [...this.sessions.values()]) await this.closeSession(state, "killed"); return manifest; }
private cancelPreparedUpdateRestart(transactionId?: symbol): void { const transaction = this.updateRestart; if (!transaction || (transactionId && transaction.id !== transactionId)) return; if (transaction.deadline) clearTimeout(transaction.deadline); transaction.deadline = undefined; transaction.abort.abort(); if (transaction.phase === "publishing") return; this.updateRestart = undefined; for (const deferred of transaction.deferredClientEnv) { if ( this.sessions.get(deferred.state.activeSessionId) === deferred.state && deferred.state.clients.has(deferred.client) && deferred.client.attachedActiveSessionIds.has(deferred.state.activeSessionId) ) { this.adoptClientEnv(deferred.state, deferred.env); } } transaction.deferredClientEnv.length = 0; for (const pause of this.updateRestartQueuePauses.values()) pause.release(); this.updateRestartQueuePauses.clear(); if (!this.shuttingDown) { this.peerAdmissionsFenced = false; this.cronScheduler.start(); } }
private async prepareUpdateRestart(): Promise<DaemonUpdateRestartManifest> { const transaction = this.beginUpdateRestartTransaction(); try { const manifest = await this.runUpdateRestartPreparation(transaction); if (transaction.deadline) clearTimeout(transaction.deadline); transaction.deadline = undefined; transaction.phase = "publishing"; this.writeUpdateRestartManifest(manifest); return await this.commitPreparedUpdateRestart(transaction.id); } catch (error) { if (this.updateRestart === transaction && transaction.phase === "publishing") { transaction.phase = "prepared"; } this.cancelPreparedUpdateRestart(transaction.id); throw error; } }
private hasScheduledJobsForSession(activeSessionId: string): boolean { return this.cronStore .list() .some( (job) => job.activeSessionId === activeSessionId && job.status !== "cancelled" && job.status !== "completed", ); }
private detachClient(client: DaemonSocketClient): void { for (const activeSessionId of [...client.attachedActiveSessionIds]) { const state = this.sessions.get(activeSessionId); if (state) { this.detachClientFromSession(client, state); } } abortClientSnapshotStreaming(client); }
private findActiveSessionByFile(sessionPath: string): ActiveSessionState | undefined { const canonicalPath = canonicalSessionPath(sessionPath); for (const state of this.sessions.values()) { const sessionFile = state.runtime.session.sessionFile; if (sessionFile && canonicalSessionPath(sessionFile) === canonicalPath) { return state; } } return undefined; }
private abortWaitingPromptAdmissionsForSession(activeSessionId: string): void { for (const admission of this.promptAdmissions.values()) { if (admission.activeSessionId === activeSessionId && admission.status === "waiting") { admission.status = "cancelled"; admission.controller?.abort(); } } }
private async closeSession( state: ActiveSessionState, reason: DaemonSessionClosedReason, waitForAbort = true, cascadeChildren = true, descendantCollector?: Set<ActiveSessionState>, disposal?: AgentSessionRuntimeDisposeOptions, ): Promise<void> { this.abortWaitingPromptAdmissionsForSession(state.activeSessionId); for (const client of state.clients) { this.abortSideQuestionsFor(client, state.activeSessionId); } const existingClose = this.closingSessions.get(state.activeSessionId); if (existingClose) { const requestedReason = this.isStrongerCloseReason(reason, existingClose.reason) ? reason : existingClose.reason; let closeError: unknown; let closeFailed = false; try { await existingClose.promise; } catch (error) { closeError = error; closeFailed = true; } const reasonUpgrade = (existingClose.reasonUpgrade ?? Promise.resolve()).then(() => { if (!this.isStrongerCloseReason(requestedReason, existingClose.reason)) return; try { this.applyReasonUpgrade(state, existingClose.descendants, existingClose.reason, requestedReason); } finally { existingClose.reason = requestedReason; } }); existingClose.reasonUpgrade = reasonUpgrade.catch(() => undefined); try { await reasonUpgrade; } finally { descendantCollector?.add(state); for (const descendant of existingClose.descendants) descendantCollector?.add(descendant); } if (closeFailed) throw closeError; return; } const descendants = new Set<ActiveSessionState>(); const closePromise = Promise.resolve().then(() => this.closeSessionOnce(state, reason, waitForAbort, cascadeChildren, descendants, disposal), ); const close = { promise: closePromise, reason, descendants }; this.closingSessions.set(state.activeSessionId, close); try { await closePromise; } finally { descendantCollector?.add(state); for (const descendant of descendants) descendantCollector?.add(descendant); if (this.closingSessions.get(state.activeSessionId) === close) { this.closingSessions.delete(state.activeSessionId); } } }
private closeReasonStrength(reason: DaemonSessionClosedReason): number { if (reason === "killed") return 2; if (reason === "completed" || reason === "replaced") return 1; return 0; }
private isStrongerCloseReason(candidate: DaemonSessionClosedReason, current: DaemonSessionClosedReason): boolean { return this.closeReasonStrength(candidate) > this.closeReasonStrength(current); }
private applyReasonUpgrade( state: ActiveSessionState, descendants: ReadonlySet<ActiveSessionState>, from: DaemonSessionClosedReason, to: DaemonSessionClosedReason, ): void { let persistError: unknown; let persistenceFailed = false; for (const target of [state, ...descendants]) { try { if (to === "killed") this.cancelScheduledJobsForSession(target); } catch (error) { if (!persistenceFailed) persistError = error; persistenceFailed = true; } try { if (this.closeKeepsResumeEntry(from)) this.archiveSession(target); } catch (error) { if (!persistenceFailed) persistError = error; persistenceFailed = true; } } if (persistenceFailed) throw persistError; }
private closeKeepsResumeEntry(reason: DaemonSessionClosedReason): boolean { return reason === "shutdown" || reason === "update"; }
private archiveSession(state: ActiveSessionState): void { state.runtime.session.sessionManager.appendSessionState({ status: "archived" }); }
private async abortBashForClose(state: ActiveSessionState): Promise<void> { const session = state.runtime.session; if (!session.isBashRunning) { return; } session.abortBash(); await Promise.race([state.inFlightBash ?? Promise.resolve(), delay(UPDATE_RESTART_ABORT_BASH_TIMEOUT_MS)]); }
private async closeSessionOnce( state: ActiveSessionState, reason: DaemonSessionClosedReason, waitForAbort: boolean, cascadeChildren: boolean, descendants: Set<ActiveSessionState>, disposal?: AgentSessionRuntimeDisposeOptions, ): Promise<void> { if (!this.sessions.has(state.activeSessionId)) { return; } if (reason === "killed") { this.cancelScheduledJobsForSession(state); } else if (reason !== "shutdown" && reason !== "update") { this.cancelSubagentRlmHeartbeats(state); } // Abort in-flight status work before any await/dispose so it can't write // agent_status to a session being torn down. this.summarizer.forget(state.activeSessionId); const cascadeError = cascadeChildren ? await this.closeChildSessions(state, reason, waitForAbort, descendants, disposal) : undefined; // Empty draft (no messages, config, or jobs): discard rather than persist an // empty session file. Mirrors the detach-time discard so a config-bearing // draft closed via kill/completed is never wiped. const keepsResumeEntry = this.closeKeepsResumeEntry(reason); const isEmptyDraftSession = !keepsResumeEntry && this.isEmptyDraftContent(state); let persistError: unknown; // Clean shutdown leaves the session un-archived so it stays in the resume list. if (!keepsResumeEntry && !isEmptyDraftSession) { try { this.archiveSession(state); } catch (error) { persistError = error; } } cancelPendingExtensionUiRequests(state); if (reason === "killed" || reason === "shutdown" || reason === "replaced" || reason === "update") { await this.abortBashForClose(state); } if (reason === "update") { state.runtime.session.abortForUpdateRestart(); const restartSession = this.updateRestart?.manifest ? new Map(this.updateRestart.manifest.sessions.map((session) => [session.activeSessionId, session])).get( state.activeSessionId, ) : undefined; if (restartSession) this.appendUpdateRestartMarker(state, restartSession); } if (reason === "killed") { const abort = state.runtime.session.abort().catch(() => undefined); if (waitForAbort) { await abort; } } else if (reason === "shutdown" || reason === "replaced") { await state.runtime.session.abort().catch(() => undefined); } this.recordWorkerRecoveryState(state, `closed:${reason}`, false); state.unsubscribe?.(); let disposeError: unknown; try { await state.runtime.dispose(disposal); } catch (error) { disposeError = error; } for (const client of state.clients) { abortClientSnapshotStreaming(client, state.activeSessionId); } this.broadcastToSession(state, { type: "session_closed", activeSessionId: state.activeSessionId, reason }); for (const client of state.clients) { client.attachedActiveSessionIds.delete(state.activeSessionId); removeDaemonClientSessionCapabilities(client, state.activeSessionId); } state.clients.clear(); this.acpMcpOwners.delete(state.activeSessionId); this.sessions.delete(state.activeSessionId); // Archived top-level sessions leave the worker's list; subagent rows mirror the registry and stay. if (!keepsResumeEntry && state.runtime.metadata.kind !== "subagent" && this.options.worker) { this.rosterReporter.removedAgentIds.set(this.rosterAgentIdForState(state), state.runtime.session.sessionId); } this.scheduleRosterFlush(); if (isEmptyDraftSession) { const sessionFile = state.runtime.session.sessionFile; if (sessionFile) { await deleteSessionFile(sessionFile).catch(() => undefined); } } if (disposeError) { throw disposeError; } if (persistError && !keepsResumeEntry && reason !== "completed") { throw persistError; } if (cascadeError && !keepsResumeEntry && reason !== "completed") { throw cascadeError; } }
private async closeChildSessions( parentState: ActiveSessionState, reason: DaemonSessionClosedReason, waitForAbort = true, descendants = new Set<ActiveSessionState>(), disposal?: AgentSessionRuntimeDisposeOptions, ): Promise<unknown> { let cascadeError: unknown; for (const childState of getChildActiveSessionStates(this.sessions, parentState)) { descendants.add(childState); try { await this.closeSession(childState, reason, waitForAbort, true, descendants, disposal); } catch (error) { cascadeError ??= error; } } return cascadeError; }
private broadcastToSession(state: ActiveSessionState, message: DaemonOutbound): void { if (message.type === "session_event") { const eventType = message.event.type; if (eventType === "tool_execution_start") { const finishTool = this.inFlightTracker.recordToolStart( message.event.toolName, message.event.toolCallId, state.activeSessionId, ); this.activeToolFinishes.set(message.event.toolCallId, finishTool); } else if (eventType === "tool_execution_end") { const finishTool = this.activeToolFinishes.get(message.event.toolCallId); if (finishTool) { finishTool(); this.activeToolFinishes.delete(message.event.toolCallId); } } else if (eventType === "turn_start") { this.inFlightTracker.recordTurnStart(state.activeSessionId); } else if (eventType === "turn_end") { this.inFlightTracker.recordTurnEnd(state.activeSessionId); } // A finished turn/compaction is the cue to refresh status. if (eventType === "turn_end" || eventType === "compaction_end") { this.summarizer.notifyActivity(state); } // A draft whose last client detached while it was busy isn't discardable // at detach time; re-check once any work (turn, compaction, or bash) // settles so it doesn't linger in the daemon. if ( (eventType === "turn_end" || eventType === "compaction_end" || eventType === "bash_end") && this.isDiscardableDraft(state) ) { this.discardAbandonedDraft(state); } if (RECOVERY_CHECKPOINT_EVENTS.has(eventType)) { this.recordWorkerRecoveryState(state, eventType); } } this.stampRlmChildActiveSessionId(message); this.observeRosterEvent(state, message); const sequencedMessage = this.addSessionEventMeta(state, message); let serialized: string | undefined; for (const client of state.clients) { if (!shouldSendDaemonOutboundToClient(client, sequencedMessage)) { continue; } // Snapshots cannot recover extension requests, so never defer or drop them. if (sequencedMessage.type === "extension_ui_request") { this.write(client, sequencedMessage); continue; } if (sequencedMessage.type === "session_closed") { this.discardDeferredSessionFrames(client, state.activeSessionId); client.catchupActiveSessionIds?.delete(state.activeSessionId); client.catchupPurposes?.delete(state.activeSessionId); this.write(client, sequencedMessage); continue; } if (client.snapshotActiveSessionIds?.has(state.activeSessionId)) { // A replacement swaps the session wholesale; buffered frames from // the old session must never replay, so fall back to a replacement // catch-up snapshot. if (sequencedMessage.type === "session_replaced") { this.discardDeferredSessionFrames(client, state.activeSessionId); client.deferredSessionFramesDropped ??= new Set(); client.deferredSessionFramesDropped.add(state.activeSessionId); this.queueClientCatchup(client, state.activeSessionId, "replacement"); continue; } if (this.deferSessionFrame(client, state.activeSessionId, sequencedMessage)) { continue; } // The deferral buffer overflowed: fall back to a full resync. this.queueClientCatchup(client, state.activeSessionId, "resync"); continue; } if (client.deferredSessionFramesDropped?.has(state.activeSessionId)) continue; if (client.backpressured === true) { this.queueClientCatchup( client, state.activeSessionId, sequencedMessage.type === "session_replaced" ? "replacement" : "resync", ); continue; } if ( sequencedMessage.type === "session_replaced" && client.transport === "private-framed" && daemonClientCapabilitiesForSession(client, state.activeSessionId).has("chunked_snapshot") ) { this.beginReplacementSnapshot(client, state, sequencedMessage); continue; } if (client.transport === "private-framed") { this.write(client, sequencedMessage); } else { serialized ??= serializeJsonLine(sequencedMessage); this.writeSerialized(client, serialized, sequencedMessage); } } }
private beginReplacementSnapshot( client: DaemonSocketClient, state: ActiveSessionState, message: Extract<DaemonOutbound, { type: "session_replaced" }>, ): void { // Mark before the registry read so later events queue behind this snapshot. const snapshotSignal = markClientSnapshotStreaming(client, state.activeSessionId); void this.prepareReplacementSnapshot(client, state, message, snapshotSignal).catch((error) => { this.log(`could not prepare replacement snapshot: ${String(error)}`); if ( !client.socket.destroyed && !snapshotSignal.aborted && client.attachedActiveSessionIds.has(state.activeSessionId) && this.sessions.get(state.activeSessionId) === state ) { try { this.write(client, message); this.flushDeferredSessionFrames(client, state.activeSessionId); } catch (replayError) { client.deferredSessionFramesDropped ??= new Set(); client.deferredSessionFramesDropped.add(state.activeSessionId); this.discardDeferredSessionFrames(client, state.activeSessionId); this.queueClientCatchup(client, state.activeSessionId, "replacement"); this.log(`could not replay replacement events: ${String(replayError)}`); } } finishClientSnapshotStreaming(client, state.activeSessionId); if (!client.snapshotStreaming && client.catchupActiveSessionIds?.size) { void this.catchUpBackpressuredClient(client).catch((catchupError) => this.log(`could not catch up replacement snapshot: ${String(catchupError)}`), ); } }); }
private async prepareReplacementSnapshot( client: DaemonSocketClient, state: ActiveSessionState, message: Extract<DaemonOutbound, { type: "session_replaced" }>, snapshotSignal: AbortSignal, ): Promise<void> { const result = await this.createAttachResult(client, state, { type: "attach", activeSessionId: state.activeSessionId, }); const snapshotId = snapshotTransferId(result.snapshot); if (this.sessions.get(state.activeSessionId) !== state) { finishClientSnapshotStreaming(client, state.activeSessionId); if (!client.snapshotStreaming && client.catchupActiveSessionIds?.size) { void this.catchUpBackpressuredClient(client).catch((catchupError) => this.log(`could not catch up replacement snapshot: ${String(catchupError)}`), ); } return; } const transcript = createSnapshotTranscriptChunks({ activeSessionId: state.activeSessionId, snapshotId, messages: result.snapshot.messages, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, signal: snapshotSignal, }); this.write(client, { ...message, messages: [], snapshotFollows: true }); void this.streamWorkerSnapshot( client, { ...result, messages: result.messages ? [] : undefined, snapshot: { ...result.snapshot, messages: [] }, snapshotStream: { id: snapshotId, messageCount: result.snapshot.messages.length, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, }, }, transcript, "replacement", snapshotSignal, true, ).catch((error) => { this.log(`could not stream replacement snapshot: ${String(error)}`); this.queueClientCatchup(client, state.activeSessionId, "replacement"); if (!client.snapshotStreaming) { void this.catchUpBackpressuredClient(client).catch((catchupError) => this.log(`could not catch up replacement snapshot: ${String(catchupError)}`), ); } }); }
private broadcastGlobal(message: DaemonOutbound): void { for (const client of this.clients) { this.write(client, message); } }
private rosterEntryForSessionPath(canonicalPath: string): WorkerRosterEntry | undefined { for (const entry of this.rosterReporter.lastComposed.values()) { if (entry.summary.sessionFile && canonicalSessionPath(entry.summary.sessionFile) === canonicalPath) { return entry; } } return undefined; }
private rosterAgentIdForState(state: ActiveSessionState): string { const session = state.runtime.session; const metadata = state.runtime.metadata; if (metadata.kind === "subagent" && metadata.rlmChildId) { return rosterAgentIdForSummary({ runtimeKind: "subagent", rlmChildId: metadata.rlmChildId, sessionId: metadata.rlmChildId, parentSessionPath: metadata.parentSessionFile, parentActiveSessionId: metadata.parentActiveSessionId, }); } return session.sessionId; }
private rosterAgentIdForRlmChild(childId: string, parentSessionPath: string | undefined): string { return rosterAgentIdForSummary({ runtimeKind: "subagent", rlmChildId: childId, sessionId: childId, parentSessionPath, }); }
private observeRosterEvent(state: ActiveSessionState, message: DaemonOutbound): void { if (!this.options.worker) return; if (message.type === "session_event") { if (message.event.type === "rlm_child_update") { this.observeRosterChildUpdate(state, message.event.child); return; } if (!ROSTER_SESSION_EVENT_TRIGGERS.has(message.event.type)) return; } else if ( message.type !== "session_status" && message.type !== "session_closed" && message.type !== "session_replaced" ) { return; } this.scheduleRosterFlush(); }
private observeRosterChildUpdate(state: ActiveSessionState, child: AgentConnectionRlmChildAgentSnapshot): void { const bound = child.activeSessionId !== undefined || this.hasSessionForRlmChild(state, child.id); const entry = this.queuedChildRosterEntry(state, child); if (!bound && (child.status === "queued" || child.status === "running")) { this.rosterReporter.queuedChildren.set(entry.agentId, entry); } else { this.rosterReporter.queuedChildren.delete(entry.agentId); } this.scheduleRosterFlush(); }
private hasSessionForRlmChild(parentState: ActiveSessionState, childId: string): boolean { for (const candidate of this.sessions.values()) { const metadata = candidate.runtime.metadata; if (metadata.rlmChildId === childId && metadata.parentActiveSessionId === parentState.activeSessionId) { return true; } } return false; }
private queuedChildRosterEntry( state: ActiveSessionState, child: AgentConnectionRlmChildAgentSnapshot, ): WorkerRosterEntry { const parentSession = state.runtime.session; const summary: RosterSessionSummary = { id: child.id, lifecycle: "live", activity: "idle", isSessionActive: false, runtimeKind: "subagent", rlmDepth: (parentSession.rlmDepth ?? 0) + 1, sessionId: child.id, sessionName: child.sessionName, cwd: parentSession.sessionManager.getCwd(), isStreaming: false, isCompacting: false, attachedClients: 0, messageCount: 0, firstMessage: child.label, parentActiveSessionId: state.activeSessionId, parentSessionId: parentSession.sessionId, parentSessionPath: parentSession.sessionFile, rlmChildId: child.id, }; return { agentId: rosterAgentIdForSummary(summary), queuedChild: true, summary }; }
private scheduleRosterFlush(): void { if (!this.options.worker || this.rosterFlushScheduled || this.shuttingDown) return; this.rosterFlushScheduled = true; setImmediate(() => { this.rosterFlushScheduled = false; try { this.flushRoster(); } catch (error) { this.log(`could not publish roster delta: ${String(error)}`); } }); }
private flushRoster(): void { const reporter = this.rosterReporter; const entries = new Map<string, WorkerRosterEntry>(); const scheduledJobs = this.cronStore.list(); for (const summary of buildSessionList([...this.sessions.values()], [], scheduledJobs)) { const entry = workerRosterEntryFromSummary(summary); entries.set(entry.agentId, entry); } for (const [agentId, queued] of reporter.queuedChildren) { if (entries.has(agentId)) { reporter.queuedChildren.delete(agentId); continue; } entries.set(agentId, queued); } // A terminal unbound child run owns no transcript: it is a removal, never a passivated row. // A vanished row whose state lives on under a new sessionId was swapped in place // (new_session/switch/fork): also a removal — plain list never served the old transcript. const composedActiveIds = new Set<string>(); for (const entry of entries.values()) { if (entry.summary.activeSessionId !== undefined) composedActiveIds.add(entry.summary.activeSessionId); } for (const [agentId, previous] of reporter.lastComposed) { if (entries.has(agentId)) continue; const swapped = previous.summary.activeSessionId !== undefined && composedActiveIds.has(previous.summary.activeSessionId); if (previous.queuedChild === true || swapped) { reporter.removedAgentIds.set(agentId, previous.summary.sessionId); } } for (const [agentId, targetSessionId] of reporter.removedAgentIds) { const composed = entries.get(agentId); // A new incarnation cancels the stale removal, as does a revived resident top-level row // (switch-back, resume-after-archive); a resident subagent row with the removed sessionId // is the mid-teardown race and stays suppressed. const revived = composed?.summary.activeSessionId !== undefined && composed.summary.runtimeKind !== "subagent"; if (composed && (composed.queuedChild === true || composed.summary.sessionId !== targetSessionId || revived)) { reporter.removedAgentIds.delete(agentId); continue; } entries.delete(agentId); reporter.queuedChildren.delete(agentId); } const registrations = scheduledJobRegistrations(scheduledJobs); for (const [agentId, previous] of reporter.lastComposed) { if (!entries.has(agentId) && !reporter.removedAgentIds.has(agentId)) { const file = previous.summary.sessionFile ? resolve(previous.summary.sessionFile) : undefined; entries.set( agentId, passivatedWorkerRosterEntry(previous, { hasRegisteredHeartbeat: file !== undefined && registrations.heartbeatSessionFiles.has(file), hasRegisteredCronJob: file !== undefined && registrations.cronSessionFiles.has(file), }), ); } } const changed: WorkerRosterEntry[] = []; const nextJson = new Map<string, string>(); for (const entry of entries.values()) { const json = JSON.stringify(entry); nextJson.set(entry.agentId, json); if (reporter.lastComposedJson.get(entry.agentId) !== json) changed.push(entry); } const removedAgentIds = [...reporter.removedAgentIds.keys()]; reporter.lastComposed = new Map(entries); reporter.lastComposedJson = nextJson; if (!this.hasAuthenticatedSupervisorClient()) { if (changed.length > 0 || removedAgentIds.length > 0) reporter.snapshotPending = true; return; } if (reporter.snapshotPending) { const delivered = this.broadcastRosterFrame({ type: "roster_delta", snapshot: true, entries: [...entries.values()], ...(removedAgentIds.length > 0 ? { removedAgentIds } : {}), }); if (delivered) { reporter.snapshotPending = false; reporter.removedAgentIds.clear(); } return; } if (changed.length === 0 && removedAgentIds.length === 0) return; const delivered = this.broadcastRosterFrame({ type: "roster_delta", entries: changed, ...(removedAgentIds.length > 0 ? { removedAgentIds } : {}), }); if (delivered) reporter.removedAgentIds.clear(); else reporter.snapshotPending = true; }
private hasAuthenticatedSupervisorClient(): boolean { for (const client of this.clients) { if (this.supervisorClaims.has(client) && !client.socket.destroyed) { return true; } } return false; }
private broadcastRosterFrame(message: DaemonWorkerRosterOutbound): boolean { const payload = Buffer.from(serializeJsonLine(message)); let delivered = false; for (const client of this.clients) { if (!this.supervisorClaims.has(client) || client.socket.destroyed) { continue; } // socket.write queues under backpressure, so a queued frame is delivered, never a loss gap. client.socket.write( encodePrivateFrame<DaemonWorkerFrameHeader>({ kind: "outbound", outboundType: message.type }, payload), ); delivered = true; } return delivered; }
private recordWorkerRecoveryState(state: ActiveSessionState, operation: string, busyOverride?: boolean): void { if (!this.recoveryJournal) { return; } const session = state.runtime.session; const busy = busyOverride ?? (hasLiveSessionWork(state) || session.isRetrying || session.hasAcceptedPromptInFlight); try { this.recoveryJournal.record({ activeSessionId: state.activeSessionId, sessionId: session.sessionId, ...(session.sessionFile ? { sessionFile: session.sessionFile } : {}), busy, operation, }); } catch (error) { this.log(`could not checkpoint worker operation state: ${String(error)}`); } }
private catchUpBackpressuredClient(client: DaemonSocketClient): Promise<void> { if (client.catchupPromise) { return client.catchupPromise; } if (client.snapshotStreaming || client.backpressured) { return Promise.resolve(); } this.clearClientCatchupRetry(client); const catchup = this.drainBackpressuredClientCatchupQueue(client).finally(() => { if (client.catchupPromise === catchup) { client.catchupPromise = undefined; } }); client.catchupPromise = catchup; return catchup; }
private clearClientCatchupRetry(client: DaemonSocketClient): void { if (!client.catchupRetryTimer) { return; } clearTimeout(client.catchupRetryTimer); client.catchupRetryTimer = undefined; }
private scheduleClientCatchupRetry(client: DaemonSocketClient): void { if (client.socket.destroyed || client.catchupRetryTimer) { return; } client.catchupRetryTimer = setTimeout(() => { client.catchupRetryTimer = undefined; if (client.socket.destroyed || !client.catchupActiveSessionIds?.size) { return; } if (client.snapshotStreaming || client.backpressured) { this.scheduleClientCatchupRetry(client); return; } void this.catchUpBackpressuredClient(client).catch((error) => this.log(`could not retry catch-up for client ${client.id}: ${String(error)}`), ); }, CLIENT_CATCHUP_RETRY_MS); }
private async drainBackpressuredClientCatchupQueue(client: DaemonSocketClient): Promise<void> { while ( !client.socket.destroyed && !client.snapshotStreaming && !client.backpressured && client.catchupActiveSessionIds?.size ) { if ((await this.drainBackpressuredClientCatchups(client)) === "retry-later") { return; } } }
private async drainBackpressuredClientCatchups(client: DaemonSocketClient): Promise<"drained" | "retry-later"> { if (client.socket.destroyed) { return "drained"; } const pending = [...(client.catchupActiveSessionIds ?? [])].map((activeSessionId) => ({ activeSessionId, purpose: client.catchupPurposes?.get(activeSessionId) ?? ("resync" as const), })); client.catchupActiveSessionIds?.clear(); client.catchupPurposes?.clear(); for (let index = 0; index < pending.length; index++) { const { activeSessionId, purpose } = pending[index]!; const state = this.sessions.get(activeSessionId); if (!state || !state.clients.has(client)) { continue; } // Buffer events before snapshot preparation can yield with a captured transcript. const snapshotSignal = markClientSnapshotStreaming(client, activeSessionId); let snapshotStarted = false; try { const result = await this.createAttachResult(client, state, { type: "attach", activeSessionId, }); if (snapshotSignal.aborted || this.sessions.get(activeSessionId) !== state || !state.clients.has(client)) { continue; } if ( client.transport === "private-framed" && daemonClientCapabilitiesForSession(client, activeSessionId).has("chunked_snapshot") ) { if (purpose === "replacement") { this.write(client, { type: "session_replaced", activeSessionId, state: result.snapshot.state, messages: [], snapshotFollows: true, meta: createDaemonEventMeta( activeSessionId, result.lastEventSequence, undefined, state.eventGeneration, ), }); } const snapshotId = snapshotTransferId(result.snapshot); const transcript = createSnapshotTranscriptChunks({ activeSessionId, snapshotId, messages: result.snapshot.messages, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, signal: snapshotSignal, }); snapshotStarted = true; await this.streamWorkerSnapshot( client, { ...result, messages: result.messages ? [] : undefined, snapshot: { ...result.snapshot, messages: [] }, snapshotStream: { id: snapshotId, messageCount: result.snapshot.messages.length, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, }, }, transcript, purpose === "replacement" ? "replacement" : "catchup", snapshotSignal, true, ); continue; } const meta = createDaemonEventMeta( activeSessionId, result.lastEventSequence, undefined, state.eventGeneration, ); const catchup: DaemonOutbound = purpose === "replacement" ? { type: "session_replaced", activeSessionId, state: result.snapshot.state, messages: result.snapshot.messages, meta, } : { type: "session_resynced", activeSessionId, snapshot: result.snapshot, meta }; const accepted = this.write(client, catchup); this.flushDeferredSessionFrames(client, activeSessionId); if (!accepted) { for (const remaining of pending.slice(index + 1)) { this.queueClientCatchup(client, remaining.activeSessionId, remaining.purpose); } return "retry-later"; } } catch (error) { if (snapshotSignal.aborted || this.sessions.get(activeSessionId) !== state || !state.clients.has(client)) { continue; } client.deferredSessionFramesDropped ??= new Set(); client.deferredSessionFramesDropped.add(activeSessionId); this.discardDeferredSessionFrames(client, activeSessionId); for (const remaining of pending.slice(index)) { this.queueClientCatchup(client, remaining.activeSessionId, remaining.purpose); } this.log(`could not catch up client ${client.id} for ${activeSessionId}: ${String(error)}`); this.scheduleClientCatchupRetry(client); return "retry-later"; } finally { if (!snapshotStarted) finishClientSnapshotStreaming(client, activeSessionId); } } return "drained"; }
private queueClientCatchup( client: DaemonSocketClient, activeSessionId: string, purpose: "replacement" | "resync" = "resync", ): void { if (!client.catchupActiveSessionIds) { client.catchupActiveSessionIds = new Set(); } client.catchupActiveSessionIds.add(activeSessionId); client.catchupPurposes ??= new Map(); if (purpose === "replacement" || !client.catchupPurposes.has(activeSessionId)) { client.catchupPurposes.set(activeSessionId, purpose); } }
/** Hold live frames until the snapshot finishes; false requests a catch-up. */ private deferSessionFrame(client: DaemonSocketClient, activeSessionId: string, message: DaemonOutbound): boolean { if (client.deferredSessionFramesDropped?.has(activeSessionId)) { return false; } const deferred = client.deferredSessionOutbounds?.get(activeSessionId) ?? { frames: [], bytes: 0 }; const serialized = serializeJsonLine(message); const bytes = Buffer.byteLength(serialized); if ( deferred.frames.length >= MAX_DEFERRED_SESSION_FRAMES || deferred.bytes + bytes > MAX_DEFERRED_SESSION_BYTES ) { // A catch-up snapshot supersedes the buffered frames. client.deferredSessionFramesDropped ??= new Set(); client.deferredSessionFramesDropped.add(activeSessionId); this.discardDeferredSessionFrames(client, activeSessionId); return false; } // Providers mutate nested streaming content after each event is emitted. deferred.frames.push(JSON.parse(serialized) as DaemonOutbound); deferred.bytes += bytes; client.deferredSessionOutbounds ??= new Map(); client.deferredSessionOutbounds.set(activeSessionId, deferred); return true; }
/** Replay frames deferred during a completed snapshot stream, in order. */ private flushDeferredSessionFrames(client: DaemonSocketClient, activeSessionId: string): void { // A queued snapshot would overwrite an earlier replay of this shared buffer. if ((client.snapshotActiveSessionCounts?.get(activeSessionId) ?? 1) > 1) return; const frames = client.deferredSessionOutbounds?.get(activeSessionId)?.frames; if (!frames || frames.length === 0) { return; } client.deferredSessionOutbounds?.delete(activeSessionId); if (!client.attachedActiveSessionIds.has(activeSessionId)) return; for (const frame of frames) { if (client.socket.destroyed) { return; } if (client.backpressured || !this.write(client, frame)) { this.queueClientCatchup(client, activeSessionId, "resync"); return; } } }
private discardDeferredSessionFrames(client: DaemonSocketClient, activeSessionId: string): void { client.deferredSessionOutbounds?.delete(activeSessionId); }
// The AgentSession doesn't know its own daemon active-session id, so fill it in here. private stampRlmChildActiveSessionId(message: DaemonOutbound): void { if ( message.type !== "session_event" || message.event.type !== "rlm_child_update" || message.event.child.activeSessionId ) { return; } const childId = message.event.child.id; for (const candidate of this.sessions.values()) { if (candidate.runtime.metadata.rlmChildId === childId) { message.event.child.activeSessionId = candidate.activeSessionId; return; } } }
private addSessionEventMeta(state: ActiveSessionState, message: DaemonOutbound): DaemonOutbound { if (!isSequencedSessionOutbound(message) || message.meta) { return message; } const meta = createDaemonEventMeta( state.activeSessionId, state.lastEventSequence + 1, undefined, state.eventGeneration, ); state.lastEventSequence = meta.sequence ?? state.lastEventSequence; return { ...message, meta }; }
private write(client: DaemonSocketClient, message: DaemonOutbound): boolean { // Direct session peers decode plain jsonl payloads only; compact deltas are a supervisor optimization. const compactDelta = client.transport === "private-framed" && client.authenticationRole !== "session_client" ? createCompactAssistantDelta(message) : undefined; return this.writeSerialized( client, serializeJsonLine(compactDelta ?? message), message, compactDelta ? "assistant-delta" : "jsonl", ); }
private writeSerialized( client: DaemonSocketClient, line: string | Buffer, message: DaemonOutbound, payloadEncoding: "jsonl" | "assistant-delta" = "jsonl", snapshotPurpose?: "attach" | "replacement" | "catchup", ): boolean { if (client.socket.destroyed) { return false; } const wireData = client.transport === "private-framed" ? encodePrivateFrame<DaemonWorkerFrameHeader>( { kind: "outbound", outboundType: message.type, ...("id" in message && typeof message.id === "string" ? { requestId: message.id } : {}), ...(hasDaemonOutboundActiveSessionId(message) ? { activeSessionId: message.activeSessionId } : {}), ...("snapshotId" in message && typeof message.snapshotId === "string" ? { snapshotId: message.snapshotId } : {}), ...(message.type === "session_event" ? { sessionEventType: message.event.type } : {}), payloadEncoding, ...(snapshotPurpose ? { snapshotPurpose } : {}), }, typeof line === "string" ? Buffer.from(line) : line, ) : line; const accepted = client.socket.write(wireData); if (!accepted) { client.backpressured = true; } return accepted; }
private abortSideQuestionsFor(client: DaemonSocketClient, activeSessionId: string): void { for (const [id, entry] of this.sideQuestionRuns) { if (entry.client !== client || entry.activeSessionId !== activeSessionId) { continue; } entry.run.abort(); this.sideQuestionRuns.delete(id); } }
private hasActiveSideQuestionFor(client: DaemonSocketClient, activeSessionId: string): boolean { for (const entry of this.sideQuestionRuns.values()) { if (entry.client === client && entry.activeSessionId === activeSessionId) { return true; } } return false; }
private registerSignalHandlers(): void { if (this.options.worker) return; const signals: NodeJS.Signals[] = ["SIGINT", "SIGTERM"]; if (process.platform !== "win32") { signals.push("SIGHUP"); } for (const signal of signals) { const handler = () => { this.log(`received ${signal}; shutting down`); killTrackedDetachedChildren(); void this.shutdown(signal === "SIGINT" ? 130 : signal === "SIGHUP" ? 129 : 143); }; process.on(signal, handler); this.signalCleanupHandlers.push(() => process.off(signal, handler)); } const exitHandler = () => this.cleanupSocketPath(); process.on("exit", exitHandler); this.signalCleanupHandlers.push(() => process.off("exit", exitHandler)); }
private getShutdownClosingReason(): DaemonClosingReason { return this.updateRestart !== undefined ? "update" : "shutdown"; }
private async shutdown(exitCode: number): Promise<never> { if (this.shuttingDown) { process.exit(exitCode); } this.shuttingDown = true; this.peerAdmissionsFenced = true; this.peerGrants.clear(); this.supervisorLinkInstance?.close(); this.cancelArmedSupervisorAvailabilityCheck(); if (this.supervisorFenceTimer) { clearTimeout(this.supervisorFenceTimer); this.supervisorFenceTimer = undefined; } if (this.rosterHeartbeatTimer) { clearInterval(this.rosterHeartbeatTimer); this.rosterHeartbeatTimer = undefined; } this.log(`shutting down (exit ${exitCode}); closing ${this.sessions.size} active session(s)`); const closingReason = this.getShutdownClosingReason(); for (const client of this.clients) { abortClientSnapshotStreaming(client); this.write(client, { type: "daemon_closing", reason: closingReason }); }
this.summarizer.stop(); for (const cleanup of this.signalCleanupHandlers) { cleanup(); } this.cronScheduler.stop(); for (const state of [...this.sessions.values()]) { await this.closeSession(state, closingReason); } for (const client of this.clients) { client.detachInput(); client.socket.end(); } await new Promise<void>((resolveClose) => { if (!this.server) { resolveClose(); return; } this.server.close(() => resolveClose()); }); this.cleanupSocketPath(); process.exit(exitCode); }}
interface WorkerRosterReporterState { lastComposed: Map<string, WorkerRosterEntry>; lastComposedJson: Map<string, string>; queuedChildren: Map<string, WorkerRosterEntry>; /** Pending removals: agentId -> removed sessionId; a new incarnation of the id cancels it. */ removedAgentIds: Map<string, string | undefined>; snapshotPending: boolean;}
const ROSTER_SESSION_EVENT_TRIGGERS = new Set([ "turn_start", "turn_end", "bash_start", "bash_end", "compaction_start", "compaction_end", "auto_retry_start", "auto_retry_end", "tool_execution_start", "tool_execution_end", "message_end", "session_action_update", "session_info_changed", "thinking_level_changed",]);
/** * The transfer id must name the cursor observed at materialization, not the live session cursor: * events appended in between would let two different byte streams share one snapshot id. */function snapshotTransferId(snapshot: DaemonSessionSnapshot): string { // createSessionSnapshot always sets lastEventCursor; it is optional only on the wire. const cursor = snapshot.lastEventCursor!; return `${snapshot.activeSessionId}-${cursor.generation}-${cursor.sequence}`;}
function hasDaemonOutboundActiveSessionId( message: DaemonOutbound,): message is DaemonOutbound & { activeSessionId: string } { return "activeSessionId" in message && typeof message.activeSessionId === "string";}
export function getChildActiveSessionStates( sessions: ReadonlyMap<string, ActiveSessionState>, parentState: ActiveSessionState,): ActiveSessionState[] { return [...sessions.values()].filter( (state) => state.activeSessionId !== parentState.activeSessionId && state.runtime.metadata.parentActiveSessionId === parentState.activeSessionId, );}
export function detachClientFromActiveSession(client: DaemonSocketClient, state: ActiveSessionState): void { state.clients.delete(client); client.attachedActiveSessionIds.delete(state.activeSessionId); client.deferredSessionFramesDropped?.delete(state.activeSessionId); removeDaemonClientSessionCapabilities(client, state.activeSessionId); if (state.clients.size === 0) { cancelPendingExtensionUiRequests(state); }}
export function setDaemonClientSessionCapabilities( client: DaemonSocketClient, activeSessionId: string, capabilities: ReadonlySet<DaemonClientCapability>,): void { client.capabilitiesByActiveSessionId ??= new Map(); client.capabilitiesByActiveSessionId.set(activeSessionId, new Set(capabilities)); client.supportsExtensionUi = [...client.capabilitiesByActiveSessionId.values()].some((value) => value.has("extension_ui"), );}
function removeDaemonClientSessionCapabilities(client: DaemonSocketClient, activeSessionId: string): void { client.capabilitiesByActiveSessionId?.delete(activeSessionId); client.supportsExtensionUi = [...(client.capabilitiesByActiveSessionId?.values() ?? [])].some((value) => value.has("extension_ui"), );}
function daemonClientCapabilitiesForSession( client: DaemonSocketClient, activeSessionId: string,): ReadonlySet<DaemonClientCapability> { return client.capabilitiesByActiveSessionId?.get(activeSessionId) ?? client.capabilities;}
function daemonClientSupportsExtensionUi(client: DaemonSocketClient, activeSessionId: string): boolean { return client.capabilitiesByActiveSessionId?.get(activeSessionId)?.has("extension_ui") ?? client.supportsExtensionUi;}
export function markClientSnapshotStreaming(client: DaemonSocketClient, activeSessionId: string): AbortSignal { if (!client.snapshotActiveSessionIds?.has(activeSessionId)) { client.deferredSessionFramesDropped?.delete(activeSessionId); } client.snapshotStreaming = true; client.snapshotActiveSessionIds ??= new Set(); client.snapshotActiveSessionIds.add(activeSessionId); client.snapshotActiveSessionCounts ??= new Map(); client.snapshotActiveSessionCounts.set( activeSessionId, (client.snapshotActiveSessionCounts.get(activeSessionId) ?? 0) + 1, ); client.snapshotTransferAbortControllers ??= new Map(); let controller = client.snapshotTransferAbortControllers.get(activeSessionId); if (!controller || controller.signal.aborted) { controller = new AbortController(); client.snapshotTransferAbortControllers.set(activeSessionId, controller); } return controller.signal;}
function abortClientSnapshotStreaming(client: DaemonSocketClient, activeSessionId?: string): void { if (activeSessionId) { client.snapshotTransferAbortControllers?.get(activeSessionId)?.abort(); return; } for (const controller of client.snapshotTransferAbortControllers?.values() ?? []) { controller.abort(); }}
export function finishClientSnapshotStreaming(client: DaemonSocketClient, activeSessionId: string): void { const count = client.snapshotActiveSessionCounts?.get(activeSessionId) ?? 1; if (count > 1) { client.snapshotActiveSessionCounts?.set(activeSessionId, count - 1); } else { client.snapshotActiveSessionCounts?.delete(activeSessionId); client.snapshotActiveSessionIds?.delete(activeSessionId); client.snapshotTransferAbortControllers?.delete(activeSessionId); // A failed or aborted stream replays nothing; drop what it withheld so // the next stream does not inherit stale frames. client.deferredSessionOutbounds?.delete(activeSessionId); client.deferredSessionPayloads?.delete(activeSessionId); } client.snapshotStreaming = (client.snapshotActiveSessionIds?.size ?? 0) > 0;}
export function cancelPendingExtensionUiRequests(state: ActiveSessionState): void { const pendingRequests = [...state.extensionUiRequests.values()]; state.extensionUiRequests.clear(); for (const pending of pendingRequests) { pending.resolve({ cancelled: true }); }}
function normalizeClientCapabilities( capabilities: readonly DaemonClientCapability[] | undefined, supportsExtensionUi: boolean | undefined,): Set<DaemonClientCapability> { const normalized = new Set<DaemonClientCapability>(); for (const capability of capabilities ?? DAEMON_DEFAULT_CLIENT_CAPABILITIES) { if (DAEMON_CLIENT_CAPABILITY_SET.has(capability)) { normalized.add(capability); } } if (supportsExtensionUi) { normalized.add("extension_ui"); } return normalized;}
type SequencedDaemonOutbound = Extract< DaemonOutbound, { type: | "session_event" | "session_status" | "session_replaced" | "session_resynced" | "session_closed" | "extension_ui_request" | "extension_error"; }>;
function isSequencedSessionOutbound(message: DaemonOutbound): message is SequencedDaemonOutbound { return ( message.type === "session_event" || message.type === "session_status" || message.type === "session_replaced" || message.type === "session_resynced" || message.type === "session_closed" || message.type === "extension_ui_request" || message.type === "extension_error" );}
export function shouldSendDaemonOutboundToClient(client: DaemonSocketClient, message: DaemonOutbound): boolean { return ( message.type !== "extension_ui_request" || !isDaemonDialogExtensionUiRequest(message.method) || daemonClientSupportsExtensionUi(client, message.activeSessionId) );}
export async function resolveDaemonSessionPath(selector: string, cwd: string, sessionDir?: string): Promise<string> { return (await resolveSessionPath(selector, cwd, sessionDir)).path;}