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.
294 kB · 7570 lines
TypeScript
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613461446154616461746184619462046214622462346244625462646274628462946304631463246334634463546364637463846394640464146424643464446454646464746484649465046514652465346544655465646574658465946604661466246634664466546664667466846694670467146724673467446754676467746784679468046814682468346844685468646874688468946904691469246934694469546964697469846994700470147024703470447054706470747084709471047114712471347144715471647174718471947204721472247234724472547264727472847294730473147324733473447354736473747384739474047414742474347444745474647474748474947504751475247534754475547564757475847594760476147624763476447654766476747684769477047714772477347744775477647774778477947804781478247834784478547864787478847894790479147924793479447954796479747984799480048014802480348044805480648074808480948104811481248134814481548164817481848194820482148224823482448254826482748284829483048314832483348344835483648374838483948404841484248434844484548464847484848494850485148524853485448554856485748584859486048614862486348644865486648674868486948704871487248734874487548764877487848794880488148824883488448854886488748884889489048914892489348944895489648974898489949004901490249034904490549064907490849094910491149124913491449154916491749184919492049214922492349244925492649274928492949304931493249334934493549364937493849394940494149424943494449454946494749484949495049514952495349544955495649574958495949604961496249634964496549664967496849694970497149724973497449754976497749784979498049814982498349844985498649874988498949904991499249934994499549964997499849995000500150025003500450055006500750085009501050115012501350145015501650175018501950205021502250235024502550265027502850295030503150325033503450355036503750385039504050415042504350445045504650475048504950505051505250535054505550565057505850595060506150625063506450655066506750685069507050715072507350745075507650775078507950805081508250835084508550865087508850895090509150925093509450955096509750985099510051015102510351045105510651075108510951105111511251135114511551165117511851195120512151225123512451255126512751285129513051315132513351345135513651375138513951405141514251435144514551465147514851495150515151525153515451555156515751585159516051615162516351645165516651675168516951705171517251735174517551765177517851795180518151825183518451855186518751885189519051915192519351945195519651975198519952005201520252035204520552065207520852095210521152125213521452155216521752185219522052215222522352245225522652275228522952305231523252335234523552365237523852395240524152425243524452455246524752485249525052515252525352545255525652575258525952605261526252635264526552665267526852695270527152725273527452755276527752785279528052815282528352845285528652875288528952905291529252935294529552965297529852995300530153025303530453055306530753085309531053115312531353145315531653175318531953205321532253235324532553265327532853295330533153325333533453355336533753385339534053415342534353445345534653475348534953505351535253535354535553565357535853595360536153625363536453655366536753685369537053715372537353745375537653775378537953805381538253835384538553865387538853895390539153925393539453955396539753985399540054015402540354045405540654075408540954105411541254135414541554165417541854195420542154225423542454255426542754285429543054315432543354345435543654375438543954405441544254435444544554465447544854495450545154525453545454555456545754585459546054615462546354645465546654675468546954705471547254735474547554765477547854795480548154825483548454855486548754885489549054915492549354945495549654975498549955005501550255035504550555065507550855095510551155125513551455155516551755185519552055215522552355245525552655275528552955305531553255335534553555365537553855395540554155425543554455455546554755485549555055515552555355545555555655575558555955605561556255635564556555665567556855695570557155725573557455755576557755785579558055815582558355845585558655875588558955905591559255935594559555965597559855995600560156025603560456055606560756085609561056115612561356145615561656175618561956205621562256235624562556265627562856295630563156325633563456355636563756385639564056415642564356445645564656475648564956505651565256535654565556565657565856595660566156625663566456655666566756685669567056715672567356745675567656775678567956805681568256835684568556865687568856895690569156925693569456955696569756985699570057015702570357045705570657075708570957105711571257135714571557165717571857195720572157225723572457255726572757285729573057315732573357345735573657375738573957405741574257435744574557465747574857495750575157525753575457555756575757585759576057615762576357645765576657675768576957705771577257735774577557765777577857795780578157825783578457855786578757885789579057915792579357945795579657975798579958005801580258035804580558065807580858095810581158125813581458155816581758185819582058215822582358245825582658275828582958305831583258335834583558365837583858395840584158425843584458455846584758485849585058515852585358545855585658575858585958605861586258635864586558665867586858695870587158725873587458755876587758785879588058815882588358845885588658875888588958905891589258935894589558965897589858995900590159025903590459055906590759085909591059115912591359145915591659175918591959205921592259235924592559265927592859295930593159325933593459355936593759385939594059415942594359445945594659475948594959505951595259535954595559565957595859595960596159625963596459655966596759685969597059715972597359745975597659775978597959805981598259835984598559865987598859895990599159925993599459955996599759985999600060016002600360046005600660076008600960106011601260136014601560166017601860196020602160226023602460256026602760286029603060316032603360346035603660376038603960406041604260436044604560466047604860496050605160526053605460556056605760586059606060616062606360646065606660676068606960706071607260736074607560766077607860796080608160826083608460856086608760886089609060916092609360946095609660976098609961006101610261036104610561066107610861096110611161126113611461156116611761186119612061216122612361246125612661276128612961306131613261336134613561366137613861396140614161426143614461456146614761486149615061516152615361546155615661576158615961606161616261636164616561666167616861696170617161726173617461756176617761786179618061816182618361846185618661876188618961906191619261936194619561966197619861996200620162026203620462056206620762086209621062116212621362146215621662176218621962206221622262236224622562266227622862296230623162326233623462356236623762386239624062416242624362446245624662476248624962506251625262536254625562566257625862596260626162626263626462656266626762686269627062716272627362746275627662776278627962806281628262836284628562866287628862896290629162926293629462956296629762986299630063016302630363046305630663076308630963106311631263136314631563166317631863196320632163226323632463256326632763286329633063316332633363346335633663376338633963406341634263436344634563466347634863496350635163526353635463556356635763586359636063616362636363646365636663676368636963706371637263736374637563766377637863796380638163826383638463856386638763886389639063916392639363946395639663976398639964006401640264036404640564066407640864096410641164126413641464156416641764186419642064216422642364246425642664276428642964306431643264336434643564366437643864396440644164426443644464456446644764486449645064516452645364546455645664576458645964606461646264636464646564666467646864696470647164726473647464756476647764786479648064816482648364846485648664876488648964906491649264936494649564966497649864996500650165026503650465056506650765086509651065116512651365146515651665176518651965206521652265236524652565266527652865296530653165326533653465356536653765386539654065416542654365446545654665476548654965506551655265536554655565566557655865596560656165626563656465656566656765686569657065716572657365746575657665776578657965806581658265836584658565866587658865896590659165926593659465956596659765986599660066016602660366046605660666076608660966106611661266136614661566166617661866196620662166226623662466256626662766286629663066316632663366346635663666376638663966406641664266436644664566466647664866496650665166526653665466556656665766586659666066616662666366646665666666676668666966706671667266736674667566766677667866796680668166826683668466856686668766886689669066916692669366946695669666976698669967006701670267036704670567066707670867096710671167126713671467156716671767186719672067216722672367246725672667276728672967306731673267336734673567366737673867396740674167426743674467456746674767486749675067516752675367546755675667576758675967606761676267636764676567666767676867696770677167726773677467756776677767786779678067816782678367846785678667876788678967906791679267936794679567966797679867996800680168026803680468056806680768086809681068116812681368146815681668176818681968206821682268236824682568266827682868296830683168326833683468356836683768386839684068416842684368446845684668476848684968506851685268536854685568566857685868596860686168626863686468656866686768686869687068716872687368746875687668776878687968806881688268836884688568866887688868896890689168926893689468956896689768986899690069016902690369046905690669076908690969106911691269136914691569166917691869196920692169226923692469256926692769286929693069316932693369346935693669376938693969406941694269436944694569466947694869496950695169526953695469556956695769586959696069616962696369646965696669676968696969706971697269736974697569766977697869796980698169826983698469856986698769886989699069916992699369946995699669976998699970007001700270037004700570067007700870097010701170127013701470157016701770187019702070217022702370247025702670277028702970307031703270337034703570367037703870397040704170427043704470457046704770487049705070517052705370547055705670577058705970607061706270637064706570667067706870697070707170727073707470757076707770787079708070817082708370847085708670877088708970907091709270937094709570967097709870997100710171027103710471057106710771087109711071117112711371147115711671177118711971207121712271237124712571267127712871297130713171327133713471357136713771387139714071417142714371447145714671477148714971507151715271537154715571567157715871597160716171627163716471657166716771687169717071717172717371747175717671777178717971807181718271837184718571867187718871897190719171927193719471957196719771987199720072017202720372047205720672077208720972107211721272137214721572167217721872197220722172227223722472257226722772287229723072317232723372347235723672377238723972407241724272437244724572467247724872497250725172527253725472557256725772587259726072617262726372647265726672677268726972707271727272737274727572767277727872797280728172827283728472857286728772887289729072917292729372947295729672977298729973007301730273037304730573067307730873097310731173127313731473157316731773187319732073217322732373247325732673277328732973307331733273337334733573367337733873397340734173427343734473457346734773487349735073517352735373547355735673577358735973607361736273637364736573667367736873697370737173727373737473757376737773787379738073817382738373847385738673877388738973907391739273937394739573967397739873997400740174027403740474057406740774087409741074117412741374147415741674177418741974207421742274237424742574267427742874297430743174327433743474357436743774387439744074417442744374447445744674477448744974507451745274537454745574567457745874597460746174627463746474657466746774687469747074717472747374747475747674777478747974807481748274837484748574867487748874897490749174927493749474957496749774987499750075017502750375047505750675077508750975107511751275137514751575167517751875197520752175227523752475257526752775287529753075317532753375347535753675377538753975407541754275437544754575467547754875497550755175527553755475557556755775587559756075617562756375647565756675677568756975707571import type { ChildProcess } from "node:child_process";import { createHash, randomBytes, randomUUID } from "node:crypto";import { chmodSync, existsSync, mkdirSync, readdirSync, readFileSync, rmSync } from "node:fs";import { createServer, type Server, type Socket } from "node:net";import { basename, dirname, join, resolve } from "node:path";import { Writable } from "node:stream";import { getLogger } from "@earendil-works/pi-ai";import { createCliSubprocessEnv, createCliSubprocessLaunchSpec } from "../../cli/subprocess-launch.js";import { appendRotatingLog, getCronJobsPath, getDaemonLogPath, getDaemonUpdateRestartManifestPath, getSessionsDir, VERSION,} from "../../config.js";import { type AgentFamilyCatalogEntry, type AgentSessionMessageAgentSummary, assertAgentFamilyReach, assertAgentSessionNameAvailable, formatAgentSessionNameUnavailable, sessionNameReservationKey,} from "../../core/agent-messages.js";import { type AgentSessionRuntimeConfig, type DurableAgentSessionRuntimeConfig, durableAgentSessionRuntimeConfig, mergeAgentSessionRuntimeConfig,} from "../../core/agent-session-config.js";import { type AgentCronJob, AgentCronJobStore, isHeartbeatCronJob, migrateLegacyCronJobsToSessionArtifacts, SESSION_SCHEDULED_JOBS_FILENAME,} from "../../core/cron-jobs.js";import { clearOrphanProcessJournal, killOrphanProcess, ORPHAN_PROCESS_JOURNAL_ENV, readActiveOrphanProcesses, shouldReapOrphanProcess,} from "../../core/orphan-process-journal.js";import { PromptAdmissionCancelledError, waitForPromptAdmission } from "../../core/prompt-admission.js";import { canEvictWorker, type IdleEvictionMinutes, type WorkerEvictionSnapshot,} from "../../core/session-action-store.js";import { sessionCatalogMode } from "../../core/session-catalog-db.js";import { canonicalSessionPath, getProcessStartId, SessionAlreadyActiveError } from "../../core/session-lease.js";import { closeSessionCatalogStores, getSessionArtifactPathForFile, getSessionArtifactsRoot, readSessionInfo, type SessionInfo, SessionManager,} from "../../core/session-manager.js";import { looksLikeSessionPath } from "../../core/session-resolver.js";import { SettingsManager } from "../../core/settings-manager.js";import { writeFileAtomicSync } from "../../utils/atomic-file.js";import { isProcessAlive, processIdExists, signalProcessGroupOrProcess, spawnHidden,} from "../../utils/child-process.js";import type { AgentConnectionHeartbeat } from "../agent-connection/types.js";import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js";import type { PrivateFrame } from "../session-worker/private-framing.js";import { createActiveSessionId, type DaemonSocketClient } from "./active-session-state.js";import { AgentRoster, type AgentRosterEntry, type AgentRosterMutation, passivatedWorkerRosterEntry, rosterAgentIdForSummary, sessionSummaryFromRosterEntry, type WorkerRosterEntry, workerRosterEntryFromSummary,} from "./agent-roster.js";import { CommandRecoveryJournal, createCommandIdempotencyKey } from "./command-recovery-journal.js";import { CompactAssistantStreamReconstructor, isCompactAssistantDelta } from "./compact-session-stream.js";import { DAEMON_CATALOG_ROLE_ENV, DaemonCatalogClient } from "./daemon-catalog-process.js";import { DaemonSessionRecoveringError, DaemonSessionStartingError, deserializeDaemonError, serializeDaemonError,} from "./daemon-errors.js";import { collectDaemonClientEnv, createDaemonEventMeta, DAEMON_COMMAND_COMPATIBILITY, DAEMON_COMMAND_ENVELOPE_MIN_PROTOCOL_VERSION, DAEMON_DEFAULT_CLIENT_CAPABILITIES, DAEMON_DEFAULT_SERVER_CAPABILITIES, DAEMON_PROTOCOL_INFO, DAEMON_SCHEMA_ID, DAEMON_SCHEMA_REVISION, DAEMON_UPDATE_RESTART_FORMAT_VERSION, type DaemonAttachResult, type DaemonClientCapability, type DaemonClosingReason, type DaemonCommand, type DaemonOutbound, type DaemonPeerTransportTicket, type DaemonResponse, type DaemonServerCapability, type DaemonUpdateRestartManifest, failure, isDaemonCommandEnvelope, isDaemonMutatingCommand, 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 { classifySessionRosterStatus, isEvictableEmptySessionSummary, isSessionSummaryBusy, type SessionSummary, summaryForInactiveSession,} from "./daemon-session-list.js";import { acquireDaemonSocketPathLease, cleanupDaemonSocketPath, type DaemonSocketIdentity, type DaemonSocketPathLease, defaultDaemonSocketDir, defaultDaemonSocketPath, getDaemonSocketIdentity, normalizeSocketPath, prepareDaemonSocketPath, restrictDaemonSocketPath,} from "./daemon-socket.js";import { acquireDaemonSupervisorOwnership, isDaemonShutdownAdmissionActive, waitForDaemonStartupFence,} from "./daemon-supervisor-ownership.js";import { DaemonWorkerAuthenticationError, DaemonWorkerClient, DaemonWorkerProbeTimeoutError,} from "./daemon-worker-client.js";import { DAEMON_WORKER_ACTIVE_SESSION_ID_ENV, DAEMON_WORKER_INSTANCE_ID_ENV, DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY, DAEMON_WORKER_RECOVERY_JOURNAL_ENV, DAEMON_WORKER_ROLE_ENV, DAEMON_WORKER_ROSTER_CAPABILITY, DAEMON_WORKER_STARTUP_GATE_COMMIT, DAEMON_WORKER_STARTUP_GATE_FD_ENV, DAEMON_WORKER_SUPERVISOR_SOCKET_ENV, DAEMON_WORKER_TOKEN_ENV, type DaemonCreateCommand, type DaemonWorkerDescriptor, type DaemonWorkerFrameHeader, type DaemonWorkerLifecycle, type DaemonWorkerRosterOutbound, durableDaemonCreateCommand, durableDaemonWorkerDescriptor, 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 RlmLedgerEdge, RlmSpawnLedger, tombstoneSavedSessionDelete, withPassiveRlmDescendantInfos,} from "./rlm-ledger.js";import { serializeSavedSessionInfo } from "./saved-session-info.js";import { SNAPSHOT_TARGET_CHUNK_BYTES, SnapshotTranscriptCache } from "./snapshot-transcript-cache.js";import { WorkerRecoveryJournal } from "./worker-recovery-journal.js";
type DistributiveOmit<T, K extends keyof T> = T extends unknown ? Omit<T, K> : never;type DaemonCommandBody = DistributiveOmit<DaemonCommand, "id">;
const structuredLog = getLogger("coding-agent.daemon-supervisor");// Windows antivirus scanning can delay worker startup beyond 30 seconds.const WORKER_CONNECT_TIMEOUT_MS = process.platform === "win32" ? 90_000 : 30_000;const WORKER_CONNECT_PROBE_MS = process.platform === "win32" ? 2_000 : 500;const WORKER_PROBE_BACKOFF_MIN_MS = 25;const WORKER_PROBE_BACKOFF_MAX_MS = process.platform === "win32" ? 2_000 : 25;
/** Per-attempt handshake waits consume the remaining outer connect budget; a smaller fixed clock makes a consistently slow (win32) handshake fail every retry. */export function handshakeBudgetMs(deadline: number, now = Date.now()): number { const remaining = deadline - now; if (remaining <= 0) throw new DaemonWorkerProbeTimeoutError("Worker connection deadline elapsed"); return remaining;}const ROSTER_WATCHDOG_INTERVAL_MS = 15_000;const ROSTER_STALE_AFTER_MS = 3 * ROSTER_HEARTBEAT_INTERVAL_MS;const SUPERVISOR_SERVER_CAPABILITIES: readonly DaemonServerCapability[] = [ ...DAEMON_DEFAULT_SERVER_CAPABILITIES, "agent_roster", "direct_peer_transport",];const PEER_TRANSPORT_GRANT_TTL_MS = 10_000;const WORKER_REQUEST_TIMEOUT_MS = 24 * 60 * 60 * 1000;// The daemon client's default request budget is 30s and each ready-worker heartbeat// forward gets 5s, so waiting longer than this on in-flight launches would only// surface as a client transport timeout instead of a bounded per-worker state error.export const HEARTBEAT_LIST_LAUNCH_WAIT_MS = 15_000;// Session-scoped listing must fail inside the client's request budget too; the// worker request default (24h) would turn a stuck worker into a client transport// timeout instead of a daemon-side failure.export const HEARTBEAT_LIST_FORWARD_TIMEOUT_MS = 25_000;export const MAX_BUSY_WORKER_SWEEP_DEFERRALS = 3;const INPUT_PAUSE_CLEANUP_TIMEOUT_MS = 5_000;const UPDATE_RESTART_MUTATION_DRAIN_TIMEOUT_MS = 80_000;const UPDATE_RESTART_WORKER_REQUEST_TIMEOUT_MS = 90_000;// The whole pre-commit prepare (drain + worker fencing) must finish inside the// caller's 120s prepare_update_restart request timeout, or roll back; otherwise// an abandoned prepare leaves the daemon permanently fenced with workers stopped.const UPDATE_RESTART_PREPARE_DEADLINE_MS = 100_000;const WORKER_RETRY_DELAYS_MS = [250, 1000, 5000] as const;/** * Upper bound on relay payloads deferred per client and session while a * snapshot stream is active. Deferral spans one stream (seconds), so overflow * means a pathologically slow client; the supervisor then falls back to a * catch-up snapshot instead of buffering without limit. */const MAX_DEFERRED_SESSION_PAYLOADS = 256;const MAX_DEFERRED_SESSION_BYTES = 8 * 1024 * 1024;const DEFERRED_RECOVERY_RECHECK_MS = 5000;// ~2.5 minutes of probing: each round is one 5s defer recheck plus a ~11s three-delay probe pass.const MAX_DEFERRED_RECOVERY_ROUNDS = 10;const MAX_CONSECUTIVE_RECOVERY_FAILURES = 6;const STOP_FINALIZATION_RECHECK_MS = 250;const STOP_FINALIZATION_SIGKILL_GRACE_MS = 5000;const STOP_FINALIZATION_RETRY_MS = 5000;const STALE_RECLAIM_WAIT_MS = 10_000;// Polling loops probe existence cheaply via kill(0); the ps-backed zombie and// identity checks are throttled so a wedged worker cannot saturate the// supervisor event loop with synchronous subprocess spawns.// Windows identity lookups launch PowerShell, so recheck less often there.const LIVENESS_IDENTITY_RECHECK_MS = process.platform === "win32" ? 3_000 : 500;const OWNED_WORKER_DISCONNECT_GRACE_MS = 30_000;const IDLE_EVICTION_MAX_SWEEP_INTERVAL_MS = 5 * 60_000;const IDLE_EVICTION_MIN_SWEEP_INTERVAL_MS = 60_000;const IDLE_EVICTION_DRAIN_TIMEOUT_MS = 5_000;const CHILD_PASSIVATION_PER_WORKER_CAP = 2;const SCHEDULED_WAKE_RETRY_MS = 60_000;const SCHEDULED_WAKE_MAX_TIMEOUT_MS = 2_147_483_647;const SCHEDULED_WAKE_CLIENT_ID = "scheduled-wake";const SUPERVISOR_CONFIG_FILE_NAME = "supervisor-config";const WORKER_STARTUP_GATE_FD = 3;
const DAEMON_COMMAND_TYPES: ReadonlySet<string> = new Set([ "ack_result", "list", "list_agent_peers", "get_direct_worker_transport", "roster_subscribe", "roster_unsubscribe", "list_saved_sessions", "create", "attach", "reattach", "detach", "complete_owned_session", "promote_owned_session", "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", "cycle_thinking_level", "set_service_tier", "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",]);
interface ResidentWorker { descriptor: DaemonWorkerDescriptor; descriptorPath: string; client?: DaemonWorkerClient; heartbeatSnapshot?: AgentConnectionHeartbeat[]; heartbeatSnapshotStale?: boolean; summaries: Map<string, SessionSummary>; snapshotCache: Map<string, DaemonAttachResult>; transcriptCaches: Map<string, SnapshotTranscriptCache>; snapshotGenerations: Map<string, Map<string, SnapshotTranscriptGeneration>>; snapshotLoads: Map<string, Promise<DaemonAttachResult>>; recovery?: Promise<void>; deferredRecovery?: Promise<void>; intentionalStop: boolean; stopRevision: number; launchEnv?: Record<string, string>; transientCreateCommand?: DaemonCreateCommand; stopFinalization?: Promise<void>; ownerCleanupTimer?: ReturnType<typeof setTimeout>; promotedOwnerClientId?: string; updateRestartPrepareClient?: DaemonWorkerClient; lastFrameAt?: number; rosterStale?: boolean; /** worker_auth advertised peer-transport support; absent on workers from older builds. */ peerTransportCapable?: boolean; /** In-flight replacement connection during authentication; an allowed frame source alongside client. */ pendingClient?: DaemonWorkerClient; /** Consecutive defer->probe rounds against a live-but-silent worker; bounded by MAX_DEFERRED_RECOVERY_ROUNDS. */ deferredRecoveryRounds?: number; /** Bumped per applied roster frame; a summaries pull that straddles a frame must not gap-fill. */ rosterEpoch?: number; rosterApplyChain?: Promise<void>; rosterRepairPull?: Promise<void>; stopReason?: string; stopSignalSent?: string; heartbeatSweepTimeouts?: number;}
interface SnapshotDuplicateValidation { promise: Promise<void>; resolve: () => void; reject: (error: Error) => void;}
interface SnapshotTranscriptGeneration { transcript: SnapshotTranscriptCache; result: DaemonAttachResult; begin?: Buffer; end?: Buffer; incoming: boolean; retired: boolean; duplicateChunkIndex?: number; duplicateResult?: DaemonAttachResult; validation?: SnapshotDuplicateValidation;}
interface DaemonSupervisorOptions { socketPath?: string; defaultSessionConfig: AgentSessionRuntimeConfig; descriptorDir?: string;}
interface PersistedSupervisorConfig { version: 1; socketPath: string; defaultSessionConfig: DurableAgentSessionRuntimeConfig;}
interface WorkerMatch { worker: ResidentWorker; summary: SessionSummary;}
interface WorkerAttachData { result: DaemonAttachResult; worker: ResidentWorker; transcript?: SnapshotTranscriptCache; releaseTranscript?: () => void;}
interface SupervisorPromptAdmission { client: DaemonSocketClient; activeSessionId: string; publicAdmissionId: string; workerAdmissionId: string; status: "waiting" | "owned" | "cancelled"; controller: AbortController; worker?: ResidentWorker; workerActiveSessionId?: string;}
interface SupervisorSessionInputPause { owner: DaemonSocketClient; worker: ResidentWorker; activeSessionId: string; requestedActiveSessionId: string; leaseKey: string; pauseId: string; releaseTask?: Promise<DaemonResponse>;}
function throwIfAdmissionCancelled(admission: SupervisorPromptAdmission | undefined): void { if (admission?.status === "cancelled") throw new PromptAdmissionCancelledError();}
class SupervisorRecoveryCancelledError extends Error { readonly code = "supervisor_recovery_cancelled" as const;}
class SnapshotLoadInvalidatedError extends Error {}
class WorkerStopTimeoutError extends Error {}
function isSupervisorGenerationStale(error: unknown): boolean { return ( typeof error === "object" && error !== null && "code" in error && (error as { code?: unknown }).code === "supervisor_generation_stale" );}
function isSupervisorRecoveryCancelled(error: unknown): boolean { return isSupervisorShutdownAdmissionCancelled(error) || isSupervisorGenerationStale(error);}
// Workers can be registered mid-tree (a resumed subagent transcript), so descent is membership at// any step of the parent walk, never a comparison against the ultimate root alone.function rosterFamilyDescendsFrom( edges: readonly RlmLedgerEdge[],): (path: string, roots: ReadonlySet<string>) => boolean { const parentByChild = new Map( edges.map((edge) => [canonicalSessionPath(edge.child), canonicalSessionPath(edge.parent)]), ); return (path, roots) => { const visited = new Set<string>(); let current = path; while (!visited.has(current)) { if (roots.has(current)) return true; visited.add(current); const parent = parentByChild.get(current); if (parent === undefined) return false; current = parent; } return false; };}
function isDaemonWorkerProbeTimeout(error: unknown): boolean { return error instanceof DaemonWorkerProbeTimeoutError;}
function isSupervisorShutdownAdmissionCancelled(error: unknown): boolean { return ( error instanceof SupervisorRecoveryCancelledError || (typeof error === "object" && error !== null && "code" in error && (error as { code?: unknown }).code === "supervisor_recovery_cancelled") );}
function delay(ms: number): Promise<void> { return new Promise((resolveDelay) => setTimeout(resolveDelay, ms));}
function unrefDelay(ms: number): Promise<void> { return new Promise((resolveDelay) => setTimeout(resolveDelay, ms).unref());}
function commitWorkerStartupGate(gate: Writable): Promise<void> { return new Promise((resolveCommit, rejectCommit) => { let settled = false; const finish = (error?: Error | null) => { if (settled) { return; } settled = true; if (error) { rejectCommit(error); } else { resolveCommit(); } }; const onError = (error: Error) => finish(error); gate.on("error", onError); gate.once("close", () => gate.off("error", onError)); gate.end(DAEMON_WORKER_STARTUP_GATE_COMMIT, (error?: Error | null) => finish(error)); });}
function withoutCommandId(command: DaemonCommand): DaemonCommandBody { const { id: _id, ...body } = command; return body as DaemonCommandBody;}
function withoutSupervisorCreateFields(command: DaemonCreateCommand): DaemonCreateCommand { const { launchEnv: _launchEnv, lifecycle: _lifecycle, ...workerCommand } = command; return workerCommand;}
function responseWithId(response: DaemonResponse, id: string | undefined): DaemonResponse { return { ...response, id };}
function isSessionSummary(value: unknown): value is SessionSummary { if (!value || typeof value !== "object") { return false; } const candidate = value as { id?: unknown; sessionId?: unknown; cwd?: unknown }; return ( typeof candidate.id === "string" && typeof candidate.sessionId === "string" && typeof candidate.cwd === "string" );}
function isDaemonWorkerDescriptor(value: unknown, socketPath: string): value is DaemonWorkerDescriptor { if (!value || typeof value !== "object") { return false; } const descriptor = value as Partial<DaemonWorkerDescriptor>; return ( (descriptor.version === 1 || descriptor.version === 2) && typeof descriptor.supervisorSocketPath === "string" && normalizeSocketPath(descriptor.supervisorSocketPath) === socketPath && typeof descriptor.workerId === "string" && Number.isInteger(descriptor.pid) && (descriptor.pid ?? 0) > 0 && (descriptor.processStartId === undefined || typeof descriptor.processStartId === "string") && (descriptor.ownerClientId === undefined || typeof descriptor.ownerClientId === "string") && typeof descriptor.socketPath === "string" && typeof descriptor.authenticationToken === "string" && (descriptor.workerInstanceId === undefined || typeof descriptor.workerInstanceId === "string") && typeof descriptor.rootActiveSessionId === "string" && typeof descriptor.createdAt === "string" && typeof descriptor.updatedAt === "string" && Number.isInteger(descriptor.consecutiveFailures) && descriptor.createCommand !== undefined && typeof descriptor.createCommand === "object" && descriptor.createCommand.type === "create" );}
class PreRosterWorkerError extends Error {}
function workerAuthAdvertisesRoster(data: unknown): boolean { if (typeof data !== "object" || data === null) return false; const capabilities = (data as { capabilities?: unknown }).capabilities; return Array.isArray(capabilities) && capabilities.includes(DAEMON_WORKER_ROSTER_CAPABILITY);}
function workerAuthAdvertisesPeerTransport(data: unknown): boolean { if (typeof data !== "object" || data === null) return false; const capabilities = (data as { capabilities?: unknown }).capabilities; return Array.isArray(capabilities) && capabilities.includes(DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY);}
function sessionSummariesFromResponse(response: DaemonResponse): SessionSummary[] { if (!response.success || !response.data || typeof response.data !== "object" || !("sessions" in response.data)) { throw new Error("Session worker returned an invalid list response"); } const sessions = (response.data as { sessions: unknown }).sessions; if (!Array.isArray(sessions) || !sessions.every(isSessionSummary)) { throw new Error("Session worker returned an invalid list response"); } return sessions;}
function attachResultFromResponse(response: DaemonResponse): DaemonAttachResult { if (!response.success || !response.data || typeof response.data !== "object") { throw new Error(response.success ? "Session worker returned an invalid attach response" : response.error); } const candidate = response.data as Partial<DaemonAttachResult>; if (typeof candidate.activeSessionId !== "string" || !candidate.snapshot || !candidate.client) { throw new Error("Session worker returned an invalid attach response"); } return candidate as DaemonAttachResult;}
function cronJobsFromResponse(response: DaemonResponse): AgentCronJob[] { if (!response.success || !response.data || typeof response.data !== "object") { return []; } const jobs = (response.data as { jobs?: unknown }).jobs; return Array.isArray(jobs) ? (jobs as AgentCronJob[]) : [];}
function heartbeatsFromResponse(response: DaemonResponse): AgentConnectionHeartbeat[] { if (!response.success || !response.data || typeof response.data !== "object") { return []; } const heartbeats = (response.data as { heartbeats?: unknown }).heartbeats; return Array.isArray(heartbeats) ? (heartbeats as AgentConnectionHeartbeat[]) : [];}
function sortCronJobs(jobs: AgentCronJob[]): AgentCronJob[] { return jobs.sort((left, right) => { if (left.nextRunAt === right.nextRunAt) { return 0; } if (left.nextRunAt === undefined) { return 1; } if (right.nextRunAt === undefined) { return -1; } return Date.parse(left.nextRunAt) - Date.parse(right.nextRunAt); });}
function descriptorKey(socketPath: string): string { return createHash("sha256").update(normalizeSocketPath(socketPath)).digest("hex").slice(0, 12);}
function defaultWorkerDescriptorDir(agentDir: string, socketPath: string): string { return join(agentDir, "daemon-workers", descriptorKey(socketPath));}
export function idleEvictionSweepIntervalMs(idleEvictionMinutes: IdleEvictionMinutes): number { if (idleEvictionMinutes === "off") return IDLE_EVICTION_MAX_SWEEP_INTERVAL_MS; return Math.max( IDLE_EVICTION_MIN_SWEEP_INTERVAL_MS, Math.min(IDLE_EVICTION_MAX_SWEEP_INTERVAL_MS, (idleEvictionMinutes * 60_000) / 3), );}
function workerSocketPath(supervisorSocketPath: string, workerId: string): string { const key = descriptorKey(supervisorSocketPath); if (process.platform === "win32") { return `\\\\.\\pipe\\prime-agent-worker-${key}-${workerId.slice(0, 12)}`; } return join(defaultDaemonSocketDir(), `worker-${key}-${workerId.slice(0, 12)}.sock`);}
function isFinalizedTranscriptEvent(eventType: string | undefined): boolean { return ( eventType === "message_end" || eventType === "turn_end" || eventType === "compaction_end" || eventType === "bash_end" );}
function normalizeCapabilities( capabilities: readonly DaemonClientCapability[] | undefined, supportsExtensionUi: boolean | undefined,): Set<DaemonClientCapability> { const normalized = new Set(capabilities ?? DAEMON_DEFAULT_CLIENT_CAPABILITIES); if (supportsExtensionUi) { normalized.add("extension_ui"); } return normalized;}
export async function runDaemonSupervisorMode(options: DaemonSupervisorOptions): Promise<never> { const socketPath = normalizeSocketPath(options.socketPath ?? defaultDaemonSocketPath()); const supervisor = new DaemonSupervisor(socketPath, options); await supervisor.start(); return new Promise(() => {});}
export class DaemonSupervisor { private server?: Server; private readonly ready: Promise<void>; private markReady: () => void = () => {}; private rejectReady: (error: Error) => void = () => {}; private ownsSocketPath = false; private socketIdentity?: DaemonSocketIdentity; private socketLease?: DaemonSocketPathLease; private socketLeaseCompromise?: Error; private ownership?: Awaited<ReturnType<typeof acquireDaemonSupervisorOwnership>>; private cleanupPromise?: Promise<void>; private shuttingDown = false; private startupComplete = false; private updateRestartPhase?: "draining" | "fencing" | "prepared"; private readonly mutationDrain = new MutationDrainLatch(); private readonly clients = new Set<DaemonSocketClient>(); private readonly connectionIds = new WeakMap<DaemonSocketClient, string>(); private readonly sessionInputPauseEpochs = new WeakMap<DaemonSocketClient, number>(); private readonly detachingInputPauseSessions = new WeakMap<DaemonSocketClient, Set<string>>(); private readonly protocolClientIds = new WeakMap<DaemonSocketClient, string>(); private readonly workers = new Map<string, ResidentWorker>(); private workerStopCounts?: Map<ResidentWorker, number>; private readonly openingWorkers = new Map<string, Promise<ResidentWorker>>(); /** * Openings that can register catalog-visible workers (`ownerClientId === undefined`). * Client-owned launches stay in `openingWorkers` for create dedupe but can never * join the public heartbeat catalog (`isVisibleWorker`), so catalog listing * never waits on them. */ private readonly catalogOpeningWorkers = new Map<string, Promise<ResidentWorker>>(); /** Public admission ids are scoped to the socket that registered them. */ private readonly promptAdmissions = new Map<DaemonSocketClient, Map<string, SupervisorPromptAdmission>>(); private readonly sessionInputPauses = new Map<string, SupervisorSessionInputPause>(); private readonly signalCleanupHandlers: Array<() => void> = []; private readonly descriptorDir: string; private readonly generation = randomUUID(); private readonly supervisorConfigPath: string; private readonly defaultSessionConfig: AgentSessionRuntimeConfig; private readonly snapshotCacheRoot: string; private commandJournal!: CommandRecoveryJournal; private readonly streamReconstructor = new CompactAssistantStreamReconstructor(); private readonly compactCatchupInProgress = new Set<string>(); private readonly pendingSessionNames = new Set<string>(); private readonly catalog: DaemonCatalogClient; private readonly settingsManager: SettingsManager; private rosterStore?: AgentRoster; private readonly pendingRosterChanged = new Set<string>(); private readonly pendingRosterRemoved = new Set<string>(); /** Ids declared to subscribers: gates removals to once and keeps owned-only row ids private. */ private readonly publishedRosterIds = new Set<string>(); private rosterPushScheduled = false; private rosterWatchdogTimer?: ReturnType<typeof setInterval>; private rlmSpawnLedgerInstance?: RlmSpawnLedger; private idleEvictionTimer?: ReturnType<typeof setTimeout>; private idleEvictionSweep?: Promise<void>; private idleEvictionFence?: Promise<void>; private scheduledWakeTimer?: ReturnType<typeof setTimeout>; private scheduledWakeRecompute?: Promise<void>; private scheduledWakeRecomputeQueued = false; private readonly scheduledWakeFailures = new Map<string, number>();
constructor( private readonly socketPath: string, options: DaemonSupervisorOptions, ) { this.ready = new Promise<void>((resolveReady, rejectReady) => { this.markReady = resolveReady; this.rejectReady = rejectReady; }); void this.ready.catch(() => undefined); const agentDir = options.defaultSessionConfig.agentDir; if (!agentDir) { throw new Error("Daemon supervisor config is missing agentDir"); } this.descriptorDir = options.descriptorDir ?? defaultWorkerDescriptorDir(agentDir, socketPath); this.supervisorConfigPath = join(this.descriptorDir, SUPERVISOR_CONFIG_FILE_NAME); this.defaultSessionConfig = mergeAgentSessionRuntimeConfig( options.defaultSessionConfig, this.loadPersistedSupervisorConfig(), ); this.snapshotCacheRoot = join(this.descriptorDir, "snapshot-cache", this.generation); this.catalog = new DaemonCatalogClient((message) => this.log(message)); this.settingsManager = SettingsManager.create(process.cwd(), this.defaultSessionConfig.agentDir ?? agentDir); }
async start(): Promise<void> { try { const agentDir = this.defaultSessionConfig.agentDir; if (!agentDir) { throw new Error("Daemon supervisor config is missing agentDir"); } this.socketLease = await acquireDaemonSocketPathLease(this.socketPath); this.socketLease?.onCompromised((error) => this.handleSocketLeaseCompromised(error)); this.assertSocketLeaseHeld(); await waitForDaemonStartupFence(this.socketPath); this.assertSocketLeaseHeld(); this.ownership = await acquireDaemonSupervisorOwnership({ socketPath: this.socketPath, descriptorDir: this.descriptorDir, agentDir, generation: this.generation, appVersion: VERSION, }); this.assertSocketLeaseHeld(); await prepareDaemonSocketPath(this.socketPath, this.socketLease);
mkdirSync(this.descriptorDir, { recursive: true, mode: 0o700 }); chmodSync(this.descriptorDir, 0o700); this.persistSupervisorConfig(); rmSync(this.snapshotCacheRoot, { recursive: true, force: true }); mkdirSync(this.snapshotCacheRoot, { recursive: true, mode: 0o700 }); this.commandJournal = new CommandRecoveryJournal(join(this.descriptorDir, "command-journal.jsonl")); this.loadWorkerDescriptors(); const workersToAdopt = [...this.workers.values()];
this.server = createServer((socket) => this.handleConnection(socket)); await this.listen(); this.assertSocketLeaseHeld(); this.socketIdentity = getDaemonSocketIdentity(this.socketPath); if (process.platform !== "win32" && !this.socketIdentity) { throw new Error(`Could not capture daemon socket identity: ${this.socketPath}`); } this.ownsSocketPath = true; restrictDaemonSocketPath(this.socketPath);
this.registerSignalHandlers(); const ownedSessionFiles = new Set( [...this.workers.values()] .flatMap((worker) => [worker.descriptor.sessionFile, worker.descriptor.createCommand.sessionPath]) .filter((path): path is string => typeof path === "string") .map((path) => resolve(path)), ); const migratedJobs = migrateLegacyCronJobsToSessionArtifacts(getCronJobsPath(agentDir), { isSessionOwned: (job) => ownedSessionFiles.has(resolve(job.sessionFile)), }); if (migratedJobs > 0) { this.log(`Migrated ${migratedJobs} scheduled jobs into session artifacts`); } await this.catalog.start().catch((error) => this.log(`Could not start daemon catalog: ${String(error)}`)); this.assertSocketLeaseHeld(); await this.seedRosterLedger(); let adoptionFailure: unknown; let adoptionFailed = false; await Promise.all( workersToAdopt.map(async (worker) => { try { await this.adoptOrRecoverWorker(worker); } catch (error) { if (!adoptionFailed) { adoptionFailed = true; adoptionFailure = error; } } }), ); if (adoptionFailed) { throw adoptionFailure; } for (const worker of this.workers.values()) { this.scheduleOwnedWorkerCleanup(worker); } this.scheduleIdleEvictionSweep(); this.scheduleScheduledSessionWakeRecompute(); this.rosterWatchdogTimer = setInterval(() => this.sweepRosterStaleness(), ROSTER_WATCHDOG_INTERVAL_MS); this.rosterWatchdogTimer.unref(); this.assertSocketLeaseHeld(); await this.ownership.updatePhase("owner"); this.assertSocketLeaseHeld(); this.startupComplete = true; this.log(`Prime Agent daemon supervisor ${this.generation} listening on ${this.socketPath}`); this.markReady(); } catch (error) { const startupError = error instanceof Error ? error : new Error(String(error)); this.log(`Daemon supervisor startup failed: ${startupError.stack ?? startupError.message}`); await this.cleanupSupervisorResources(); this.rejectReady(startupError); throw startupError; } }
private listen(): Promise<void> { return new Promise<void>((resolveListen, rejectListen) => { const onError = (error: Error) => { this.server?.off("listening", onListening); rejectListen(error); }; const onListening = () => { this.server?.off("error", onError); resolveListen(); }; this.server?.once("error", onError); this.server?.once("listening", onListening); this.server?.listen(this.socketPath); }); }
private log(message: string): void { console.error(message); structuredLog.warn(message, { socketPath: this.socketPath }); if (this.socketPath) { appendRotatingLog(getDaemonLogPath(this.socketPath), `[${new Date().toISOString()}] supervisor: ${message}`); } }
private clearIdleEvictionTimer(): void { if (!this.idleEvictionTimer) return; clearTimeout(this.idleEvictionTimer); this.idleEvictionTimer = undefined; }
private clearRosterWatchdogTimer(): void { if (!this.rosterWatchdogTimer) return; clearInterval(this.rosterWatchdogTimer); this.rosterWatchdogTimer = undefined; }
private clearScheduledWakeTimer(): void { if (!this.scheduledWakeTimer) return; clearTimeout(this.scheduledWakeTimer); this.scheduledWakeTimer = undefined; }
// The supervisor only wakes non-resident trees; firing and delivery stay worker-owned. private scheduleScheduledSessionWakeRecompute(): void { if (this.shuttingDown) return; if (this.scheduledWakeRecompute) { this.scheduledWakeRecomputeQueued = true; return; } this.scheduledWakeRecompute = this.recomputeScheduledSessionWake() .catch((error) => this.log(`Scheduled-session wake recompute failed: ${String(error)}`)) .finally(() => { this.scheduledWakeRecompute = undefined; if (this.scheduledWakeRecomputeQueued) { this.scheduledWakeRecomputeQueued = false; this.scheduleScheduledSessionWakeRecompute(); } }); }
/** * Canonical session paths whose artifact dir holds scheduled jobs. `undefined` means * the artifact tree cannot answer (absent, or a job file without a conventionally * named transcript), in which case the caller walks the whole ledger family instead. */ private scheduledJobSessionPaths(): Set<string> | undefined { const agentDir = this.defaultSessionConfig.agentDir; const sessionsDir = this.defaultSessionConfig.sessionDir ?? (agentDir ? getSessionsDir(agentDir) : undefined); if (sessionsDir === undefined) return undefined; const artifactsRoot = getSessionArtifactsRoot(sessionsDir); let names: string[]; try { names = readdirSync(artifactsRoot); } catch { return undefined; } const paths = new Set<string>(); for (const sessionId of names) { if (!existsSync(join(artifactsRoot, sessionId, SESSION_SCHEDULED_JOBS_FILENAME))) continue; const sessionPath = join(sessionsDir, `${sessionId}.jsonl`); if (!existsSync(sessionPath)) return undefined; paths.add(canonicalSessionPath(sessionPath)); } return paths; }
/** Durable truth: the ledger family (fork headers stripped) plus each session's scheduled-jobs artifact. */ private async collectPassiveScheduledJobs( includeInactive = false, ): Promise<Array<{ rootSessionFile: string; job: AgentCronJob; info: SessionInfo }>> { const pendingCancelRoots = new Set<string>(); for (const intent of this.collectEphemeralCancelIntents()) { const context = this.workerSessionArtifactContext(intent); if (!context || !intent.descriptor.rootSessionId) continue; try { await this.cancelScheduledJobsForSessionTree(intent.descriptor.rootSessionId, context.sessionFile); this.deleteWorkerDescriptor(intent); } catch { // Still owned until the cancel lands; the tree stays excluded below. pendingCancelRoots.add(canonicalSessionPath(context.sessionFile)); } } // Session artifact dirs are the index of which sessions can carry scheduled jobs: // scheduling needs the artifact file, so a sessions dir without one has nothing to // wake. Without this, boot folds every saved transcript to list armed heartbeats. const indexed = this.scheduledJobSessionPaths(); if (indexed && indexed.size === 0) return []; // The index answers which sessions can carry jobs; when the artifact tree cannot // answer, the ledger scan still reads only conventionally named transcripts, so a // large catalog is not folded just to list armed heartbeats. const infos = await this.rlmSpawnLedger().family( indexed ? { onlyPaths: indexed } : SESSION_SCHEDULED_JOBS_FILENAME, ); const infoByPath = new Map(infos.map((info) => [canonicalSessionPath(info.path), info] as const)); const storeBySessionId = new Map<string, AgentCronJobStore>(); const infoBySessionId = new Map<string, SessionInfo>(); for (const info of infos) { if (info.state !== undefined && info.state.status !== "active") continue; const artifactDir = getSessionArtifactPathForFile(resolve(info.path), info.id); if (!existsSync(join(artifactDir, SESSION_SCHEDULED_JOBS_FILENAME))) continue; const store = AgentCronJobStore.forSessionArtifacts(); store.registerSessionArtifact(info.id, artifactDir); storeBySessionId.set(info.id, store); infoBySessionId.set(info.id, info); } if (infoBySessionId.size === 0) return []; const uncoveredRootFor = (info: SessionInfo): string | undefined => { let current = info; const visited = new Set([canonicalSessionPath(current.path)]); while (true) { try { if (this.findWorkerBySessionFile(current.path)) return undefined; } catch { return undefined; } if (!current.parentSessionPath) break; const parent = infoByPath.get(canonicalSessionPath(current.parentSessionPath)); if (!parent || visited.has(canonicalSessionPath(parent.path))) break; visited.add(canonicalSessionPath(parent.path)); current = parent; } return current.path; }; const results: Array<{ rootSessionFile: string; job: AgentCronJob; info: SessionInfo }> = []; for (const [artifactSessionId, store] of storeBySessionId) { let jobs: AgentCronJob[]; try { jobs = store.list(); } catch (error) { this.log(`Skipping unreadable scheduled jobs for session ${artifactSessionId}: ${String(error)}`); continue; } for (const job of jobs) { if (!includeInactive && job.status !== "active" && job.status !== "paused") continue; const info = infoBySessionId.get(job.sessionId); if (!info) continue; const rootSessionFile = uncoveredRootFor(info); if (rootSessionFile === undefined) continue; if (pendingCancelRoots.has(canonicalSessionPath(rootSessionFile))) continue; results.push({ rootSessionFile, job, info }); } } return results; }
private async recomputeScheduledSessionWake(): Promise<void> { if (this.shuttingDown || this.updateRestartPhase !== undefined) return; const candidates = await this.collectPassiveScheduledJobs(); this.clearScheduledWakeTimer(); if (this.shuttingDown || this.updateRestartPhase !== undefined) return; const now = Date.now(); const wakeTimes: number[] = []; const candidateRoots = new Set(candidates.map(({ rootSessionFile }) => canonicalSessionPath(rootSessionFile))); for (const root of [...this.scheduledWakeFailures.keys()]) { if (!candidateRoots.has(root)) this.scheduledWakeFailures.delete(root); } for (const { rootSessionFile, job } of candidates) { if (job.status !== "active" || job.nextRunAt === undefined) continue; const runAt = Date.parse(job.nextRunAt); if (!Number.isFinite(runAt)) continue; // The failure floor keeps an overdue job off a hot retry loop. const failedAt = this.scheduledWakeFailures.get(canonicalSessionPath(rootSessionFile)); wakeTimes.push(failedAt !== undefined ? Math.max(runAt, failedAt + SCHEDULED_WAKE_RETRY_MS) : runAt); } if (wakeTimes.length === 0) return; const delay = Math.min(Math.max(0, Math.min(...wakeTimes) - now), SCHEDULED_WAKE_MAX_TIMEOUT_MS); this.scheduledWakeTimer = setTimeout(() => { this.scheduledWakeTimer = undefined; void this.wakeDueScheduledSessions().catch((error) => this.log(`Scheduled-session wake failed: ${String(error)}`), ); }, delay); this.scheduledWakeTimer.unref(); }
private async wakeDueScheduledSessions(now = Date.now()): Promise<void> { // Disarmed during update-restart preparation; the phase transition or next boot re-arms once. if (this.shuttingDown || this.updateRestartPhase !== undefined) { this.clearScheduledWakeTimer(); return; } try { const due = new Map<string, string>(); for (const { rootSessionFile, job } of await this.collectPassiveScheduledJobs()) { if (job.status !== "active" || job.nextRunAt === undefined) continue; const runAt = Date.parse(job.nextRunAt); if (!Number.isFinite(runAt) || runAt > now) continue; due.set(canonicalSessionPath(rootSessionFile), rootSessionFile); } for (const [rootKey, sessionPath] of due) { if (this.shuttingDown || this.updateRestartPhase !== undefined) return; try { await this.createOrReuseWorker(SCHEDULED_WAKE_CLIENT_ID, { type: "create", sessionPath }); this.scheduledWakeFailures.delete(rootKey); this.log(`Woke session worker for a due scheduled job: ${sessionPath}`); } catch (error) { this.scheduledWakeFailures.set(rootKey, Date.now()); this.log(`Scheduled wake failed for ${sessionPath}: ${String(error)}`); } } } finally { this.scheduleScheduledSessionWakeRecompute(); } }
private scheduleIdleEvictionSweep(): void { if (this.shuttingDown || this.idleEvictionTimer || this.idleEvictionSweep) return; const delayMs = idleEvictionSweepIntervalMs(this.settingsManager.getIdleEvictionMinutes()); this.idleEvictionTimer = setTimeout(() => { this.idleEvictionTimer = undefined; const sweep = this.runIdleEvictionSweep() .catch((error) => this.log(`Idle eviction sweep failed: ${String(error)}`)) .finally(() => { if (this.idleEvictionSweep === sweep) this.idleEvictionSweep = undefined; this.scheduleIdleEvictionSweep(); }); this.idleEvictionSweep = sweep; }, delayMs); this.idleEvictionTimer.unref(); }
private workerEvictionSnapshot(worker: ResidentWorker): WorkerEvictionSnapshot { return { lifecycle: worker.descriptor.lifecycle, isConnected: worker.client !== undefined, isStopping: this.isWorkerStopping(worker), hasOwnerClient: worker.descriptor.ownerClientId !== undefined, isPreparingUpdateRestart: this.updateRestartPhase !== undefined || worker.updateRestartPrepareClient !== undefined, hasWakeBlindSchedule: this.isWakeBlindScheduledWorker(worker), sessions: this.workerRosterEntries(worker) .filter((entry) => !entry.queuedChild) .map(sessionSummaryFromRosterEntry) .map((summary) => { const activeSessionId = summary.activeSessionId ?? summary.id; return { isSessionActive: isSessionSummaryBusy(summary), attachedClients: this.attachedClientCount(summary, activeSessionId), hasRegisteredCronJob: summary.hasRegisteredCronJob === true, lastActivityAt: Date.parse(summary.lastActivityAt ?? ""), }; }), }; }
/** A schedule whose root file sits outside the enumerable sessions root is invisible to the wake scan; keep it resident. */ private isWakeBlindScheduledWorker(worker: ResidentWorker): boolean { const sessionFile = worker.descriptor.sessionFile ?? worker.descriptor.createCommand.sessionPath; const agentDir = this.defaultSessionConfig.agentDir; if (!sessionFile || !agentDir) return false; const sessionsRoot = canonicalSessionPath(this.defaultSessionConfig.sessionDir ?? getSessionsDir(agentDir)); if (dirname(canonicalSessionPath(sessionFile)) === sessionsRoot) return false; return this.workerRosterEntries(worker).some( (entry) => entry.summary.hasRegisteredHeartbeat === true || entry.summary.hasRegisteredCronJob === true, ); }
private async runIdleEvictionSweep(now = Date.now()): Promise<void> { if (this.shuttingDown || this.updateRestartPhase !== undefined || this.idleEvictionFence) return; await this.settingsManager.reload(); if (this.shuttingDown || this.updateRestartPhase !== undefined) return; const idleEvictionMinutes = this.settingsManager.getIdleEvictionMinutes(); if (idleEvictionMinutes === "off") return;
const refreshed = new Set<ResidentWorker>(); await Promise.all( [...this.workers.values()].map(async (worker) => { try { await this.refreshWorkerSummaries(worker); refreshed.add(worker); } catch { // A disconnected or transitioning worker is never an eviction candidate. } }), ); const candidates = [...refreshed].filter((worker) => canEvictWorker(this.workerEvictionSnapshot(worker), idleEvictionMinutes, now), ); if (this.shuttingDown || this.updateRestartPhase !== undefined) return; // Whole-tree candidates skip child work because stopWorker releases everything. await Promise.all( [...refreshed] .filter((worker) => !candidates.includes(worker)) .map(async (worker) => { try { const response = await worker.client?.requestWorker( { type: "worker_passivate_idle_children", idleEvictionMinutes, now, limit: CHILD_PASSIVATION_PER_WORKER_CAP, }, 30_000, ); if (response && !response.success) throw new Error(response.error); await this.refreshWorkerSummaries(worker); } catch (error) { refreshed.delete(worker); this.log(`Child passivation sweep failed for worker ${worker.descriptor.workerId}: ${String(error)}`); } }), ); if (this.shuttingDown || this.updateRestartPhase !== undefined) return;
await this.withEvictionFence("Timed out draining daemon mutations for idle eviction", async () => { if (this.shuttingDown || this.updateRestartPhase !== undefined) return; await Promise.all( candidates.map((worker) => this.refreshWorkerSummaries(worker).catch(() => refreshed.delete(worker))), ); if (this.shuttingDown || this.updateRestartPhase !== undefined) return; const evictable = candidates.filter( (worker) => refreshed.has(worker) && this.workers.get(worker.descriptor.workerId) === worker && canEvictWorker(this.workerEvictionSnapshot(worker), idleEvictionMinutes, now), ); // Promise.all may reject and release the fence while sibling stops are still // finishing. That is safe: a racing mutation either reaches a live worker or // gets a clean disconnected/unknown-session error. await Promise.all( evictable.map(async (worker) => { if (this.shuttingDown || this.updateRestartPhase !== undefined) return; const snapshot = this.workerEvictionSnapshot(worker); const idleMinutes = Math.floor( Math.min(...snapshot.sessions.map((session) => now - session.lastActivityAt)) / 60_000, ); const root = worker.summaries.get(worker.descriptor.rootActiveSessionId); await this.stopWorker(worker, true); this.log( `Evicted idle worker ${worker.descriptor.workerId} root=${root?.sessionId ?? worker.descriptor.rootSessionId ?? worker.descriptor.rootActiveSessionId} idleMinutes=${idleMinutes} sessions=${snapshot.sessions.length}`, ); }), ); }); }
/** Waits for any held eviction fence, then takes the slot; release clears it only if still ours. */ private async acquireIdleEvictionFence(): Promise<() => void> { while (this.idleEvictionFence) await this.idleEvictionFence; let releaseFence: () => void = () => {}; const fence = new Promise<void>((resolveFence) => { releaseFence = resolveFence; }); this.idleEvictionFence = fence; return () => { if (this.idleEvictionFence === fence) this.idleEvictionFence = undefined; releaseFence(); }; }
/** Runs a passivation decision under the eviction fence after draining admitted mutations. */ private async withEvictionFence(drainMessage: string, action: () => Promise<void>): Promise<void> { const releaseFence = await this.acquireIdleEvictionFence(); try { await this.mutationDrain.waitForDrain(0, AbortSignal.timeout(IDLE_EVICTION_DRAIN_TIMEOUT_MS), drainMessage); await action(); } finally { releaseFence(); } }
private async passivateWorkerIfStillEligible( worker: ResidentWorker, isStillEligible: () => boolean, describeEvicted: () => string, ): Promise<void> { await this.refreshWorkerSummaries(worker, false, true); if (!isStillEligible()) return; await this.stopWorker(worker, true); this.log(describeEvicted()); }
private async evictEmptySessionOnLastDetach(activeSessionId: string): Promise<void> { if (this.shuttingDown || this.updateRestartPhase !== undefined) return; const worker = this.matchWorkers(activeSessionId)[0]?.worker; if ( !worker || worker.descriptor.lifecycle !== "ready" || !worker.client || worker.descriptor.ownerClientId !== undefined || // owned workers have their own cleanup path this.isWorkerStopping(worker) ) { return; } try { await this.refreshWorkerSummaries(worker, false, true); } catch { return; } if (!this.isEmptyDetachEvictionCandidate(worker)) return; try { // Idle-sweep coordination: fence new mutations, drain admitted ones, re-read before deciding. await this.withEvictionFence("Timed out draining daemon mutations for empty-session eviction", () => this.passivateWorkerIfStillEligible( worker, () => this.isEmptyDetachEvictionCandidate(worker), () => `Evicted empty session worker ${worker.descriptor.workerId} root=${worker.descriptor.rootSessionId ?? worker.descriptor.rootActiveSessionId} on last client detach`, ), ); } catch (error) { this.log(`Empty-session eviction failed for worker ${worker.descriptor.workerId}: ${String(error)}`); } }
private isEmptyDetachEvictionCandidate(worker: ResidentWorker): boolean { if ( this.shuttingDown || this.updateRestartPhase !== undefined || this.workers.get(worker.descriptor.workerId) !== worker || this.isWorkerStopping(worker) || this.isWakeBlindScheduledWorker(worker) ) { return false; } const summaries = this.workerRosterEntries(worker) .filter((entry) => !entry.queuedChild) .map(sessionSummaryFromRosterEntry); const hasAttachedClient = summaries.some( (summary) => this.attachedClientCount(summary, summary.activeSessionId ?? summary.id) > 0, ); return summaries.length > 0 && !hasAttachedClient && summaries.every(isEvictableEmptySessionSummary); }
private async assertCurrentOwnership(): Promise<void> { const ownership = this.ownership; if (!ownership) { const error = new Error( `Daemon supervisor generation ${this.generation} holds no registry ownership (never acquired or already released); ` + `socket: ${this.socketPath}; restart the daemon to recover — sessions are preserved`, ); Object.assign(error, { code: "supervisor_generation_stale" as const }); throw error; } await ownership.assertCurrent(); }
private async assertServingCurrentOwnership(): Promise<void> { this.assertSupervisorServing(); await this.assertCurrentOwnership(); this.assertSupervisorServing(); }
private async assertRecoveryAllowed(): Promise<void> { await this.assertServingCurrentOwnership(); if (await isDaemonShutdownAdmissionActive()) { throw new SupervisorRecoveryCancelledError("Daemon shutdown admission cancelled worker recovery"); } }
private supervisorAuthenticationClaim(): { supervisorGeneration: string; supervisorPid: number; supervisorProcessStartId?: string; supervisorSocketPath: string; } { const record = this.ownership?.record; if (!record) { throw new SupervisorRecoveryCancelledError("Daemon supervisor ownership is unavailable"); } return { supervisorGeneration: this.generation, supervisorPid: record.pid, ...(record.processStartId ? { supervisorProcessStartId: record.processStartId } : {}), supervisorSocketPath: record.socketPath, }; }
private loadWorkerDescriptors(): void { for (const name of readdirSync(this.descriptorDir)) { if (name === SUPERVISOR_CONFIG_FILE_NAME || !name.endsWith(".json")) { continue; } const path = join(this.descriptorDir, name); try { const descriptor: unknown = JSON.parse(readFileSync(path, "utf8")); if (!isDaemonWorkerDescriptor(descriptor, this.socketPath)) { continue; } if ((descriptor.consecutiveFailures ?? 0) >= MAX_CONSECUTIVE_RECOVERY_FAILURES) { this.log( `Discarding exhausted worker descriptor ${descriptor.workerId} (${descriptor.consecutiveFailures} failures)`, ); this.deleteWorkerDescriptor({ descriptorPath: path, descriptor }); continue; } descriptor.supervisorSocketPath = normalizeSocketPath(descriptor.supervisorSocketPath); descriptor.lifecycle = "recovering"; descriptor.recoveryJournalPath ??= join(this.descriptorDir, `${descriptor.workerId}.recovery.jsonl`); descriptor.orphanProcessJournalPath ??= join(this.descriptorDir, `${descriptor.workerId}.orphans.jsonl`); const durableDescriptor = durableDaemonWorkerDescriptor(descriptor); const worker: ResidentWorker = { descriptor: durableDescriptor, descriptorPath: path, summaries: new Map(), snapshotCache: new Map(), transcriptCaches: new Map(), snapshotGenerations: new Map(), snapshotLoads: new Map(), intentionalStop: durableDescriptor.stopRequestedAt !== undefined, stopRevision: 0, }; this.persistWorker(worker); this.workers.set(durableDescriptor.workerId, worker); } catch (error) { this.log(`Ignoring invalid worker descriptor ${path}: ${String(error)}`); } } }
/** Trees whose ephemeral-stop cancel failed stay owned until a retry lands: tombstoned client-owned descriptors with no resident worker. */ private collectEphemeralCancelIntents(): Array<{ descriptor: DaemonWorkerDescriptor; descriptorPath: string }> { const intents: Array<{ descriptor: DaemonWorkerDescriptor; descriptorPath: string }> = []; let names: string[]; try { names = readdirSync(this.descriptorDir); } catch { return intents; } for (const name of names) { if (name === SUPERVISOR_CONFIG_FILE_NAME || !name.endsWith(".json")) continue; const descriptorPath = join(this.descriptorDir, name); let descriptor: unknown; try { descriptor = JSON.parse(readFileSync(descriptorPath, "utf8")); } catch { continue; } if (!isDaemonWorkerDescriptor(descriptor, this.socketPath)) continue; if (descriptor.stopRequestedAt === undefined || descriptor.ownerClientId === undefined) continue; if (this.workers.has(descriptor.workerId)) continue; intents.push({ descriptor, descriptorPath }); } return intents; }
private loadPersistedSupervisorConfig(): AgentSessionRuntimeConfig | undefined { try { const parsed = JSON.parse( readFileSync(this.supervisorConfigPath, "utf8"), ) as Partial<PersistedSupervisorConfig>; if ( parsed.version !== 1 || typeof parsed.socketPath !== "string" || normalizeSocketPath(parsed.socketPath) !== this.socketPath || !parsed.defaultSessionConfig || typeof parsed.defaultSessionConfig !== "object" || typeof parsed.defaultSessionConfig.agentDir !== "string" ) { return undefined; } return durableAgentSessionRuntimeConfig(parsed.defaultSessionConfig); } catch { return undefined; } }
private persistSupervisorConfig(): void { const persisted: PersistedSupervisorConfig = { version: 1, socketPath: this.socketPath, defaultSessionConfig: durableAgentSessionRuntimeConfig(this.defaultSessionConfig), }; writeFileAtomicSync(this.supervisorConfigPath, `${JSON.stringify(persisted, null, 2)}\n`, { mode: 0o600 }); }
private hasPersistedWorkerDescriptors(): boolean { return readdirSync(this.descriptorDir).some( (name) => name !== SUPERVISOR_CONFIG_FILE_NAME && name.endsWith(".json"), ); }
private persistWorker(worker: ResidentWorker): void { worker.descriptor.updatedAt = new Date().toISOString(); const persisted = durableDaemonWorkerDescriptor(worker.descriptor); writeFileAtomicSync(worker.descriptorPath, `${JSON.stringify(persisted, null, 2)}\n`, { mode: 0o600 }); }
private deleteWorkerDescriptor(worker: { descriptorPath: string; descriptor: DaemonWorkerDescriptor }): void { try { rmSync(worker.descriptorPath, { force: true }); rmSync(worker.descriptor.recoveryJournalPath, { force: true }); if (worker.descriptor.orphanProcessJournalPath) { rmSync(worker.descriptor.orphanProcessJournalPath, { force: true }); } } catch (error) { this.log(`Failed to remove worker descriptor ${worker.descriptorPath}: ${String(error)}`); } }
private handleConnection(socket: Socket): void { const client: DaemonSocketClient = { id: createActiveSessionId(), socket, attachedActiveSessionIds: new Set(), catchupActiveSessionIds: new Set(), backpressured: false, authenticated: true, snapshotActiveSessionIds: new Set(), detachInput: () => {}, supportsExtensionUi: false, capabilities: new Set(DAEMON_DEFAULT_CLIENT_CAPABILITIES), }; this.connectionIds.set(client, client.id); this.sessionInputPauseEpochs.set(client, 0); this.detachingInputPauseSessions.set(client, new Set()); this.clients.add(client); void this.ready.then( () => { if (!client.socket.destroyed && this.clients.has(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(), supervisorGeneration: this.generation, supervisorOwnerToken: this.ownership?.record.token, supervisorPid: process.pid, supervisorProcessStartId: this.ownership?.record.processStartId, supervisorSocketPath: this.ownership?.record.socketPath, clientId: client.id, serverCapabilities: SUPERVISOR_SERVER_CAPABILITIES, }); } }, () => client.socket.destroy(), );
client.detachInput = attachJsonlLineReader(socket, (line) => void this.handleLine(client, line)); let cleaned = false; const cleanup = () => { if (cleaned) { return; } cleaned = true; clearTimeout(client.catchupRetryTimer); client.catchupRetryTimer = undefined; client.detachInput(); this.sessionInputPauseEpochs.set(client, (this.sessionInputPauseEpochs.get(client) ?? 0) + 1); const ownerClientId = this.protocolClientId(client); void this.releaseClientSessionInputPauses(client, undefined, true).catch((error: unknown) => this.log(`Failed to release input pauses for disconnected client ${ownerClientId}: ${String(error)}`), ); this.clients.delete(client); this.cancelWaitingPromptAdmissionsForClient(client); for (const activeSessionId of [...client.attachedActiveSessionIds]) { client.attachedActiveSessionIds.delete(activeSessionId); void this.syncWorkerExtensionUi(activeSessionId); void this.evictEmptySessionOnLastDetach(activeSessionId); } this.scheduleOwnedWorkerCleanupForClient(this.protocolClientId(client)); }; socket.on("close", cleanup); socket.on("error", cleanup); socket.on("drain", () => { client.backpressured = false; if (client.rosterResyncPending && client.rosterSubscribed === true) { // socket.write queues even when it reports backpressure: one resync per loss gap. client.rosterResyncPending = false; this.write(client, { type: "roster_update", changed: this.rosterEntriesForClient(), resync: true }); } if (!client.snapshotStreaming) { void this.catchUpClient(client).catch((error) => this.log(`Failed to catch up client ${client.id}: ${String(error)}`), ); } }); }
private cancelOwnedWorkerCleanup(clientId: string): void { for (const worker of this.workers.values()) { if (worker.descriptor.ownerClientId !== clientId || !worker.ownerCleanupTimer) { continue; } clearTimeout(worker.ownerCleanupTimer); worker.ownerCleanupTimer = undefined; } }
private protocolClientId(client: DaemonSocketClient): string { return this.protocolClientIds.get(client) ?? client.id; }
private async releaseClientSessionInputPauses( owner: DaemonSocketClient, activeSessionId?: string, forceCleanupOnFailure = false, ): Promise<void> { const entries = [...this.sessionInputPauses.values()].filter( (entry) => entry.owner === owner && (activeSessionId === undefined || entry.activeSessionId === activeSessionId || entry.requestedActiveSessionId === activeSessionId), ); await Promise.all( entries.map(async (entry) => { if (this.workers.get(entry.worker.descriptor.workerId) !== entry.worker || !entry.worker.client) { if (this.sessionInputPauses.get(entry.pauseId) === entry) this.sessionInputPauses.delete(entry.pauseId); return; } try { const response = await this.forwardToWorker( entry.worker, { id: randomUUID(), type: "release_session_input_pause", activeSessionId: entry.activeSessionId, pauseId: entry.pauseId, }, INPUT_PAUSE_CLEANUP_TIMEOUT_MS, ); if (!response.success) throw new Error(response.error); if (this.sessionInputPauses.get(entry.pauseId) === entry) this.sessionInputPauses.delete(entry.pauseId); } catch (error) { if (forceCleanupOnFailure) { const workerClient = entry.worker.client; if (workerClient) { workerClient.close(); await this.handleWorkerClose( entry.worker, workerClient, error instanceof Error ? error : new Error(String(error)), ); } else { this.invalidateWorkerSessionInputPauses( entry.worker, "Session worker became unavailable while releasing an input pause", ); } } throw error; } }), ); }
private invalidateWorkerSessionInputPauses(worker: ResidentWorker, reason: string): void { const owners = new Set<DaemonSocketClient>(); for (const [pauseId, entry] of this.sessionInputPauses) { if (entry.worker !== worker) continue; this.sessionInputPauses.delete(pauseId); owners.add(entry.owner); } for (const owner of owners) { owner.socket.destroy(new Error(reason)); } }
private scheduleOwnedWorkerCleanupForClient(clientId: string): void { if ([...this.clients].some((client) => this.protocolClientId(client) === clientId)) { return; } for (const worker of this.workers.values()) { if (worker.descriptor.ownerClientId === clientId) { this.scheduleOwnedWorkerCleanup(worker); } } }
private scheduleOwnedWorkerCleanup(worker: ResidentWorker): void { const ownerClientId = worker.descriptor.ownerClientId; if ( !ownerClientId || worker.ownerCleanupTimer || [...this.clients].some((client) => this.protocolClientId(client) === ownerClientId) ) { return; } worker.ownerCleanupTimer = setTimeout(() => { worker.ownerCleanupTimer = undefined; if ( worker.descriptor.ownerClientId !== ownerClientId || [...this.clients].some((client) => this.protocolClientId(client) === ownerClientId) || this.workers.get(worker.descriptor.workerId) !== worker ) { return; } void this.stopWorker(worker, true).catch((error) => this.log(`Could not clean up client-owned worker ${worker.descriptor.workerId}: ${String(error)}`), ); }, OWNED_WORKER_DISCONNECT_GRACE_MS); worker.ownerCleanupTimer.unref(); }
private promptAdmissionKey(activeSessionId: string, publicAdmissionId: string): string { return `${activeSessionId}\0${publicAdmissionId}`; }
private promptAdmissionsFor(client: DaemonSocketClient): Map<string, SupervisorPromptAdmission> { let admissions = this.promptAdmissions.get(client); if (!admissions) { admissions = new Map(); this.promptAdmissions.set(client, admissions); } return admissions; }
private getPromptAdmission( client: DaemonSocketClient, activeSessionId: string, publicAdmissionId: string, ): SupervisorPromptAdmission | undefined { return this.promptAdmissions.get(client)?.get(this.promptAdmissionKey(activeSessionId, publicAdmissionId)); }
private deletePromptAdmission(admission: SupervisorPromptAdmission): void { const admissions = this.promptAdmissions.get(admission.client); const key = this.promptAdmissionKey(admission.activeSessionId, admission.publicAdmissionId); if (admissions?.get(key) !== admission) return; admissions.delete(key); if (admissions.size === 0) this.promptAdmissions.delete(admission.client); }
private cancelWaitingPromptAdmissionsForClient(client: DaemonSocketClient): void { for (const admission of this.promptAdmissions.get(client)?.values() ?? []) { if (admission.status !== "waiting") continue; if (!admission.worker || !admission.workerActiveSessionId) { admission.status = "cancelled"; admission.controller.abort(); continue; } const worker = admission.worker; const workerActiveSessionId = admission.workerActiveSessionId; void this.forwardToWorker(worker, { type: "cancel_prompt_admission", activeSessionId: workerActiveSessionId, admissionId: admission.workerAdmissionId, }) .then((response) => { if (admission.status !== "waiting") return; const status = response.success && response.data && typeof response.data === "object" && "status" in response.data ? (response.data as { status?: unknown }).status : undefined; if (status === "owned") admission.status = "owned"; else if (status === "cancelled") admission.status = "cancelled"; }) .catch((error: unknown) => { this.log( `Could not cancel prompt admission ${admission.workerAdmissionId} on disconnected client: ${String(error)}`, ); }); } }
/** Non-async by design: prompt registration completes before handleLine's first await. */ private parseCommandAndRegisterPromptAdmission( client: DaemonSocketClient, line: string, ): { command: DaemonCommand; envelopeClientId?: string; protocolVersion: number; admission?: SupervisorPromptAdmission; } { const parsed = JSON.parse(line) as unknown; const envelope = isDaemonCommandEnvelope(parsed) ? parsed : undefined; if (!envelope) { throw new Error(`Daemon commands require protocol ${DAEMON_COMMAND_ENVELOPE_MIN_PROTOCOL_VERSION} or newer`); } const command = { ...envelope.command, id: envelope.id } as DaemonCommand; let admission: SupervisorPromptAdmission | undefined; if ((command.type === "prompt" || command.type === "prompt_and_wait") && command.admissionId !== undefined) { if (typeof command.activeSessionId !== "string" || typeof command.admissionId !== "string") { throw new Error("Prompt admission requires string activeSessionId and admissionId"); } if (command.admissionId === "") throw new Error("admissionId must not be empty"); const admissions = this.promptAdmissionsFor(client); const key = this.promptAdmissionKey(command.activeSessionId, command.admissionId); if (admissions.has(key)) { throw new Error(`Prompt admission id is already in use: ${command.admissionId}`); } admission = { client, activeSessionId: command.activeSessionId, publicAdmissionId: command.admissionId, workerAdmissionId: `supervisor-admission:${randomUUID()}`, status: "waiting", controller: new AbortController(), }; admissions.set(key, admission); } return { command, envelopeClientId: envelope.clientId ?? client.id, protocolVersion: envelope.protocol.version, admission, }; }
private async handleLine(client: DaemonSocketClient, line: string): Promise<void> { try { this.assertSupervisorServing(); } catch (error) { this.write(client, failure(salvageDaemonCommandId(line), "dispatch", error, serializeDaemonError(error))); return; } let preParsed: ReturnType<DaemonSupervisor["parseCommandAndRegisterPromptAdmission"]>; try { preParsed = this.parseCommandAndRegisterPromptAdmission(client, line); } catch (error) { this.write(client, failure(salvageDaemonCommandId(line), "parse", error)); return; } const command = preParsed.command; const parsedAdmission = preParsed.admission; if (command.type === "cancel_prompt_admission" && this.updateRestartPhase !== undefined) { this.write(client, failure(command.id, command.type, "Daemon is preparing an update restart")); return; } const cancellationAdmission = command.type === "cancel_prompt_admission" ? this.getPromptAdmission(client, command.activeSessionId, command.admissionId) : undefined; if (cancellationAdmission?.status === "waiting" && !cancellationAdmission.worker) { cancellationAdmission.status = "cancelled"; cancellationAdmission.controller.abort(); } try { await waitForPromptAdmission(this.ready, parsedAdmission?.controller.signal); } catch (error) { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write(client, failure(command.id, command.type, error)); return; } const envelopeClientId = preParsed.envelopeClientId; if (envelopeClientId) { this.protocolClientIds.set(client, envelopeClientId); client.id = envelopeClientId; } this.cancelOwnedWorkerCleanup(client.id); if (!DAEMON_COMMAND_TYPES.has(command.type)) { this.write(client, failure(command.id, command.type, `Unknown daemon command: ${command.type}`)); return; } if ( command.type === "get_session_tree" && preParsed.protocolVersion < DAEMON_COMMAND_COMPATIBILITY.get_session_tree.minProtocol ) { this.write( client, failure( command.id, command.type, `get_session_tree requires client protocol ${DAEMON_COMMAND_COMPATIBILITY.get_session_tree.minProtocol} or newer`, ), ); return; }
try { await waitForPromptAdmission(this.assertServingCurrentOwnership(), parsedAdmission?.controller.signal); } catch (error) { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write(client, failure(command.id, command.type, error)); return; }
const mutation = isDaemonMutatingCommand(command); const journalIdentity = envelopeClientId && command.id && mutation ? { clientId: envelopeClientId, commandId: command.id } : undefined; const existing = journalIdentity ? this.commandJournal.lookup(journalIdentity.clientId, journalIdentity.commandId) : undefined; if (existing?.status === "complete") { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write(client, existing.response); return; } if (existing?.status === "pending" && journalIdentity) { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write( client, failure(command.id, command.type, "The previous command result is uncertain and was not replayed", { code: "command_result_uncertain", ...journalIdentity, }), ); return; }
const phase = this.updateRestartPhase; const restartRejected = phase === "draining" ? !UPDATE_RESTART_DRAIN_COMMANDS.has(command.type) : phase !== undefined && !(phase === "prepared" && command.type === "shutdown"); if (restartRejected && mutation) { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write(client, failure(command.id, command.type, "Daemon is preparing an update restart")); return; } if (mutation && !UPDATE_RESTART_DRAIN_COMMANDS.has(command.type)) { const idleEvictionFence = this.idleEvictionFence; if (idleEvictionFence) { await idleEvictionFence; try { await this.assertServingCurrentOwnership(); } catch (error) { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write(client, failure(command.id, command.type, error, serializeDaemonError(error))); return; } } } if (journalIdentity) { const admitted = this.commandJournal.begin(journalIdentity.clientId, journalIdentity.commandId, command.type); if (admitted.status === "complete") { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write(client, admitted.response); return; } if (admitted.status === "pending") { if (parsedAdmission) this.deletePromptAdmission(parsedAdmission); this.write( client, failure(command.id, command.type, "The previous command result is uncertain and was not replayed", { code: "command_result_uncertain", ...journalIdentity, }), ); return; } }
// Attach is intentionally read-only and is not fence-gated. If eviction wins // the race, attach fails cleanly with "Session worker is not connected" and // the client retries through the saved-session path instead of mutating state. if (mutation) this.mutationDrain.begin(); try { const response = await this.handleCommand(client, command, cancellationAdmission); if (response) { if (journalIdentity) { await this.assertCurrentOwnership(); this.commandJournal.recordResult(journalIdentity.clientId, journalIdentity.commandId, response); } this.write(client, response); } } catch (error) { this.log(`Supervisor command ${command.type} failed: ${error instanceof Error ? error.stack : String(error)}`); let response = failure(command.id, command.type, error, serializeDaemonError(error)); if (journalIdentity && !isSupervisorGenerationStale(error)) { try { await this.assertCurrentOwnership(); this.commandJournal.recordResult(journalIdentity.clientId, journalIdentity.commandId, response); } catch (ownershipError) { response = failure(command.id, command.type, ownershipError, serializeDaemonError(ownershipError)); } } this.write(client, response); } finally { if (mutation) this.mutationDrain.end(); } }
private async handleCommand( client: DaemonSocketClient, command: DaemonCommand, cancellationAdmission?: SupervisorPromptAdmission, ): Promise<DaemonResponse | undefined> { switch (command.type) { case "cancel_prompt_admission": { const admission = cancellationAdmission ?? this.getPromptAdmission(client, command.activeSessionId, command.admissionId); if (!admission) return success(command.id, command.type, { status: "unknown" as const }); if (admission.status === "owned") return success(command.id, command.type, { status: "owned" as const }); // A definitive cancellation never downgrades to unknown/waiting. if (admission.status === "cancelled") { return success(command.id, command.type, { status: "cancelled" as const }); } if (!admission.worker || !admission.workerActiveSessionId) { admission.status = "cancelled"; admission.controller.abort(); return success(command.id, command.type, { status: "cancelled" as const }); } const response = await this.forwardToWorker(admission.worker, { ...command, activeSessionId: admission.workerActiveSessionId, admissionId: admission.workerAdmissionId, }); const status = response.success && response.data && typeof response.data === "object" && "status" in response.data ? (response.data as { status: "cancelled" | "owned" | "unknown" }).status : "unknown"; // Re-read: a socket close may have cancelled during the round-trip (cast widens TS's pre-await narrowing). const current = (admission as SupervisorPromptAdmission).status; if (status === "owned") admission.status = "owned"; else if (status === "cancelled") admission.status = "cancelled"; else if (current !== "cancelled") admission.status = "waiting"; return { ...response, id: command.id }; } case "ack_result": this.commandJournal.acknowledge(client.id, command.commandId); return undefined; case "list": return this.handleList(client, command); case "roster_subscribe": client.rosterSubscribed = true; return success(command.id, command.type, { roster: this.rosterEntriesForClient() }); case "roster_unsubscribe": client.rosterSubscribed = false; client.rosterResyncPending = false; return success(command.id, command.type); case "list_agent_peers": { const requester = [...this.workers.values()].find( (worker) => worker.descriptor.authenticationToken === command.workerToken, ); if (!requester) throw new Error("Worker authentication failed"); const peers = [...this.workers.values()] .filter( (worker) => worker !== requester && this.isLiveWorker(worker) && worker.descriptor.lifecycle === "ready" && worker.client !== undefined, ) .flatMap((worker) => { const root = this.roster().byActiveSessionId(worker.descriptor.rootActiveSessionId); return root ? [this.agentPeerSummary(sessionSummaryFromRosterEntry(root))] : []; }); return success(command.id, command.type, { peers }); } case "get_direct_worker_transport": { const match = await this.findWorkerForClient(client, command.activeSessionId); if (match.worker.descriptor.ownerClientId !== undefined) { throw new Error("Direct transport is unavailable for client-owned workers"); } const ticket = await this.issuePeerTransport(match.worker, match.summary); return success(command.id, command.type, ticket); } case "list_saved_sessions": return this.handleSavedSessionList(client, command); case "create": { const worker = await this.createOrReuseWorker(this.protocolClientId(client), command); const requestedSummary = command.sessionPath ? this.findSummaryInWorker(worker, command.sessionPath) : undefined; if ( requestedSummary && (requestedSummary.activeSessionId ?? requestedSummary.id) !== worker.descriptor.rootActiveSessionId ) { // A create forwarded to a recovering worker still surfaces an opaque lifecycle error. const response = await this.forwardToWorker(worker, withoutSupervisorCreateFields(command)); if (response.success && isSessionSummary(response.data)) { this.writeRosterEntry(workerRosterEntryFromSummary(response.data), worker); return { ...response, id: command.id, data: this.publicSummary(worker, response.data) }; } return responseWithId(response, command.id); } const root = this.roster().byActiveSessionId(worker.descriptor.rootActiveSessionId); if (!root) { throw new Error("Session worker started without a root session"); } return success(command.id, "create", this.publicSummary(worker, sessionSummaryFromRosterEntry(root))); } case "attach": { const attached = await this.attachClient(client, command); if (client.capabilities.has("chunked_snapshot")) { const transcript = attached.transcript; if (!transcript) { throw new Error("Session worker did not provide a snapshot transcript"); } const streamedResult = this.createStreamedAttachResult(attached.result, transcript); try { this.write(client, success(command.id, "attach", streamedResult)); void this.streamSnapshot( client, attached.worker, streamedResult, transcript, "attach", attached.releaseTranscript, ).catch((error) => this.log( `Failed to stream attach snapshot for ${streamedResult.activeSessionId}: ${String(error)}`, ), ); } catch (error) { attached.releaseTranscript?.(); throw error; } return undefined; } return success(command.id, "attach", attached.result); } case "reattach": { const target = await this.findWorkerForClient(client, command.targetActiveSessionId); const targetActiveSessionId = target.summary.activeSessionId ?? target.summary.id; if (targetActiveSessionId === command.activeSessionId) { const detachingSessions = this.detachingInputPauseSessions?.get(client); detachingSessions?.delete(command.activeSessionId); detachingSessions?.delete(command.targetActiveSessionId); detachingSessions?.delete(targetActiveSessionId); return success(command.id, command.type, { cancelled: false }); } const targetWasAttached = client.attachedActiveSessionIds.has(targetActiveSessionId); const releaseSnapshotReservation = this.reserveSnapshotStream(client, targetActiveSessionId); let releaseTranscript: (() => void) | undefined; client.attachedActiveSessionIds.add(targetActiveSessionId); try { const attached = await this.attachClient(client, { ...command, type: "attach", activeSessionId: targetActiveSessionId, }); const detachingSessions = this.detachingInputPauseSessions?.get(client); detachingSessions?.delete(command.activeSessionId); detachingSessions?.delete(command.targetActiveSessionId); detachingSessions?.delete(targetActiveSessionId); if (client.capabilities.has("chunked_snapshot")) { const transcript = attached.transcript ?? this.getOrCreateTranscriptCache(attached.worker, attached.result); releaseTranscript = attached.releaseTranscript; const streamedResult = this.createStreamedAttachResult(attached.result, transcript); this.write(client, success(command.id, command.type, streamedResult)); this.detachClient(client, command.activeSessionId); const streaming = this.streamSnapshot( client, attached.worker, streamedResult, transcript, "replacement", releaseTranscript, releaseSnapshotReservation, ); releaseTranscript = undefined; void streaming.catch((error) => this.log(`Failed to stream reattach snapshot for ${targetActiveSessionId}: ${String(error)}`), ); return undefined; } this.write(client, success(command.id, command.type, attached.result)); this.detachClient(client, command.activeSessionId); releaseSnapshotReservation(); this.releaseDeferredSessionPayloads(client, targetActiveSessionId, true); return undefined; } catch (error) { releaseTranscript?.(); if (!targetWasAttached) { this.detachClient(client, targetActiveSessionId); } releaseSnapshotReservation(); this.releaseDeferredSessionPayloads(client, targetActiveSessionId, targetWasAttached); throw error; } } case "acquire_session_input_pause": { const detachingSessions = this.detachingInputPauseSessions.get(client); if (detachingSessions?.has(command.activeSessionId)) { throw new Error(`Session is detaching: ${command.activeSessionId}`); } const match = await this.findWorkerForClient(client, command.activeSessionId); const activeSessionId = match.summary.activeSessionId ?? match.summary.id; if (detachingSessions?.has(command.activeSessionId) || detachingSessions?.has(activeSessionId)) { throw new Error(`Session is detaching: ${command.activeSessionId}`); } const ownerClientId = this.protocolClientId(client); const connectionId = this.connectionIds.get(client); if (!connectionId) throw new Error("Daemon client connection identity is unavailable"); const acquisitionEpoch = this.sessionInputPauseEpochs.get(client) ?? 0; const existing = [...this.sessionInputPauses.values()].find( (entry) => entry.owner === client && entry.worker === match.worker && entry.activeSessionId === activeSessionId && entry.leaseKey === command.leaseKey && !entry.releaseTask, ); if (existing) { return success(command.id, command.type, { pauseId: existing.pauseId }); } const response = await this.forwardToWorker(match.worker, { ...command, activeSessionId, leaseKey: JSON.stringify([connectionId, ownerClientId, command.leaseKey]), }); if (!response.success) return response; const pauseId = (response.data as { pauseId?: unknown } | undefined)?.pauseId; if (typeof pauseId !== "string") throw new Error("Worker returned an invalid session input pause id"); if (!this.clients.has(client) || (this.sessionInputPauseEpochs.get(client) ?? 0) !== acquisitionEpoch) { try { const release = await this.forwardToWorker(match.worker, { id: randomUUID(), type: "release_session_input_pause", activeSessionId, pauseId, }); if (!release.success) throw new Error(release.error); } catch (error) { match.worker.client?.close(); throw error; } throw new Error("Session input pause acquisition was invalidated before completion"); } this.sessionInputPauses.set(pauseId, { owner: client, worker: match.worker, activeSessionId, requestedActiveSessionId: command.activeSessionId, leaseKey: command.leaseKey, pauseId, }); return response; } case "release_session_input_pause": { const entry = this.sessionInputPauses.get(command.pauseId); if (!entry) return success(command.id, command.type); if (entry.owner !== client) { throw new Error(`Session input pause is owned by another client: ${command.pauseId}`); } if ( command.activeSessionId !== entry.activeSessionId && command.activeSessionId !== entry.requestedActiveSessionId ) { throw new Error(`Session input pause belongs to another session: ${command.pauseId}`); } const releaseTask = entry.releaseTask ?? this.forwardToWorker(entry.worker, { ...command, activeSessionId: entry.activeSessionId, }); entry.releaseTask = releaseTask; try { const response = await releaseTask; if (response.success && this.sessionInputPauses.get(command.pauseId) === entry) { this.sessionInputPauses.delete(command.pauseId); } return response; } finally { if (this.sessionInputPauses.get(command.pauseId) === entry && entry.releaseTask === releaseTask) { entry.releaseTask = undefined; } } } case "detach": { const detachingSessions = this.detachingInputPauseSessions.get(client) ?? new Set<string>(); this.detachingInputPauseSessions.set(client, detachingSessions); if (command.activeSessionId) detachingSessions.add(command.activeSessionId); else for (const activeSessionId of client.attachedActiveSessionIds) detachingSessions.add(activeSessionId); this.sessionInputPauseEpochs.set(client, (this.sessionInputPauseEpochs.get(client) ?? 0) + 1); this.detachClient(client, command.activeSessionId); await this.releaseClientSessionInputPauses(client, command.activeSessionId, true); return success(command.id, "detach"); } case "complete_owned_session": { const match = await this.findWorkerForClient(client, command.activeSessionId); if (match.worker.descriptor.ownerClientId !== this.protocolClientId(client)) { throw new Error("Session is not owned by this client"); } if (match.worker.ownerCleanupTimer) { clearTimeout(match.worker.ownerCleanupTimer); match.worker.ownerCleanupTimer = undefined; } await this.stopWorker(match.worker, true); return success(command.id, command.type); } case "promote_owned_session": { const match = await this.findWorkerForClient(client, command.activeSessionId); await this.promoteOwnedWorker(client, match.worker); return success(command.id, command.type, this.publicSummary(match.worker, match.summary)); } case "retry_worker": { const direct = [...this.workers.values()].find( (worker) => worker.descriptor.rootActiveSessionId === command.activeSessionId || worker.descriptor.rootSessionId === command.activeSessionId, ); const worker = direct ?? (await this.findWorkerForClient(client, command.activeSessionId)).worker; this.assertWorkerAccessibleToClient(client, worker, command.activeSessionId); if ((this.workerStopCounts?.get(worker) ?? 0) > 0) { throw new Error("Session worker is stopping; retry after it finishes"); } await this.retryWorkerRecovery(worker); if (this.workers.get(worker.descriptor.workerId)?.descriptor.lifecycle !== "ready") { throw new Error(worker.descriptor.lastError ?? "Session worker recovery failed"); } const summary = worker.summaries.get(worker.descriptor.rootActiveSessionId); return success(command.id, command.type, summary ? this.publicSummary(worker, summary) : undefined); } case "restart": setImmediate(() => void this.shutdown(0, false, true, false, "update")); return success(command.id, command.type); case "shutdown": { const reason = ("reason" in command && (command.reason === "update" || command.reason === "shutdown") ? command.reason : undefined) ?? (this.updateRestartPhase !== undefined ? "update" : "shutdown"); setImmediate(() => void this.shutdown(0, true, false, command.force === true, reason)); return success(command.id, "shutdown"); } case "prepare_update_restart": { const manifest = await this.prepareUpdateRestart(); return success(command.id, "prepare_update_restart", manifest); } case "agent_messages_status": { if (command.activeSessionId) { const match = await this.findWorkerForClient(client, command.activeSessionId); return this.forwardToWorker(match.worker, command); } const first = [...this.workers.values()].find((worker) => this.isLiveWorker(worker) && worker.client); if (!first) { return success(command.id, command.type, { paused: false, limits: {} }); } return this.forwardToWorker(first, command); } case "agent_messages_pause": case "agent_messages_resume": { if (command.activeSessionId) { const match = await this.findWorkerForClient(client, command.activeSessionId); return this.forwardToWorker(match.worker, command); } const responses = await Promise.all( [...this.workers.values()] .filter((worker) => this.isLiveWorker(worker) && worker.client) .map((worker) => this.forwardToWorker(worker, command)), ); const failed = responses.find((response) => !response.success); return failed ?? success(command.id, command.type, responses.find((response) => response.success)?.data); } case "cron_list": { if (command.activeSessionId) { const match = await this.findWorkerForClient(client, command.activeSessionId); return this.forwardToWorker(match.worker, command); } const jobs = new Map<string, AgentCronJob>(); const responses = await Promise.all( [...this.workers.values()] .filter( (worker) => this.isLiveWorker(worker) && worker.client && worker.descriptor.lifecycle === "ready", ) .map((worker) => this.forwardToWorker(worker, command, 5000).catch((error: unknown) => failure(command.id, command.type, error, serializeDaemonError(error)), ), ), ); for (const response of responses) { if (!response.success) { this.log(`Could not list scheduled jobs from a worker: ${response.error}`); continue; } for (const job of cronJobsFromResponse(response)) { jobs.set(job.id, job); } } for (const { job } of await this.collectPassiveScheduledJobs(command.includeInactive === true)) { if (!jobs.has(job.id)) jobs.set(job.id, job); } return success(command.id, "cron_list", { jobs: sortCronJobs([...jobs.values()]) }); } case "heartbeats_list": { if (command.activeSessionId) { const match = await this.findWorkerForClient(client, command.activeSessionId); // The forward may first join an in-flight recovery whose budget far exceeds // the client's request timeout, so bound the whole operation: a stuck or // still-recovering worker fails daemon-side inside the caller's budget // instead of surfacing as a client transport timeout. const forward = this.forwardToWorker(match.worker, command, HEARTBEAT_LIST_FORWARD_TIMEOUT_MS); const forwardDeadline = unrefDelay(HEARTBEAT_LIST_FORWARD_TIMEOUT_MS).then(() => { throw new Error( `Timed out waiting for session worker to list heartbeats within ${HEARTBEAT_LIST_FORWARD_TIMEOUT_MS}ms`, ); }); return Promise.race([forward, forwardDeadline]) .then((response) => { if (response.success) { match.worker.heartbeatSweepTimeouts = 0; } return response; }) .catch((error: unknown) => { const isBusy = this.isWorkerBusy(match.worker); const timeouts = (match.worker.heartbeatSweepTimeouts ?? 0) + 1; match.worker.heartbeatSweepTimeouts = timeouts; if (isBusy && timeouts <= MAX_BUSY_WORKER_SWEEP_DEFERRALS) { this.log( `worker ${match.worker.descriptor.workerId} unresponsive for ${timeouts} sweeps while busy (turn in flight); deferring stop`, ); } else { if (isBusy) { this.log( `worker ${match.worker.descriptor.workerId} unresponsive for ${timeouts} sweeps while busy; bound reached, stopping worker`, ); } void this.stopWorker( match.worker, true, true, false, false, undefined, "session-scoped heartbeats_list probe timed out", ).catch(() => undefined); } return failure(command.id, command.type, error, serializeDaemonError(error)); }); } const openings = [...this.catalogOpeningWorkers.values()]; const selectedWorkers = new Set(this.workers.values()); for (const opening of openings) { opening.then( (worker) => selectedWorkers.add(worker), () => undefined, ); } // Slow launches must not outrun the caller's request budget: after // HEARTBEAT_LIST_LAUNCH_WAIT_MS the catalog proceeds with whatever // registered, and workers that are still starting surface through the // per-worker state error below instead of being omitted. await Promise.race([Promise.allSettled(openings), unrefDelay(HEARTBEAT_LIST_LAUNCH_WAIT_MS)]); for (const worker of this.workers.values()) { selectedWorkers.add(worker); } const workers = [...this.workers.values()].filter( (worker) => selectedWorkers.has(worker) && this.isLiveWorker(worker) && worker.descriptor.lifecycle !== "failed", ); const heartbeats = new Map<string, AgentConnectionHeartbeat>(); const snapshots: Array<{ heartbeats?: AgentConnectionHeartbeat[]; response?: DaemonResponse }> = await Promise.all( workers.map(async (worker) => { if (worker.client && worker.descriptor.lifecycle === "ready") { const response = await this.forwardToWorker(worker, command, 5000).catch((error: unknown) => failure(command.id, command.type, error, serializeDaemonError(error)), ); if (response.success) { worker.heartbeatSweepTimeouts = 0; const snapshot = heartbeatsFromResponse(response); worker.heartbeatSnapshot = snapshot; worker.heartbeatSnapshotStale = false; return { heartbeats: snapshot }; } this.log(`Could not list heartbeats from a worker: ${response.error}`); const isBusy = this.isWorkerBusy(worker); const timeouts = (worker.heartbeatSweepTimeouts ?? 0) + 1; worker.heartbeatSweepTimeouts = timeouts; if (isBusy && timeouts <= MAX_BUSY_WORKER_SWEEP_DEFERRALS) { this.log( `worker ${worker.descriptor.workerId} unresponsive for ${timeouts} sweeps while busy (turn in flight); deferring stop`, ); } else { if (isBusy) { this.log( `worker ${worker.descriptor.workerId} unresponsive for ${timeouts} sweeps while busy; bound reached, stopping worker`, ); } void this.stopWorker( worker, true, true, false, false, undefined, `heartbeats_list probe timed out (${response.error})`, ).catch(() => undefined); } if (worker.heartbeatSnapshot === undefined || worker.heartbeatSnapshotStale === true) { return { response }; } } if (worker.heartbeatSnapshot !== undefined && worker.heartbeatSnapshotStale !== true) { return { heartbeats: worker.heartbeatSnapshot }; } const state = worker.descriptor.lifecycle === "ready" ? "disconnected" : worker.descriptor.lifecycle; const error = new Error(`Cannot list heartbeats while session worker is ${state}`); return { response: failure(command.id, command.type, error, serializeDaemonError(error)) }; }), ); const failed = snapshots.find((snapshot) => snapshot.response)?.response; if (failed) { return failed; } for (const snapshot of snapshots) { for (const heartbeat of snapshot.heartbeats ?? []) { heartbeats.set(heartbeat.job.id, heartbeat); } } // Passivated sessions keep their armed heartbeats; no worker can list them. for (const { job, info } of await this.collectPassiveScheduledJobs()) { if (!isHeartbeatCronJob(job) || heartbeats.has(job.id)) continue; heartbeats.set(job.id, { job, ...(info.name !== undefined ? { sessionName: info.name } : {}), ...(info.firstMessage !== undefined ? { firstMessage: info.firstMessage } : {}), }); } return success(command.id, "heartbeats_list", { heartbeats: [...heartbeats.values()] }); } case "heartbeat_manage": { const cachedWorker = [...this.workers.values()].find((worker) => worker.heartbeatSnapshot?.some( (heartbeat) => heartbeat.job.id === command.jobId && heartbeat.job.activeSessionId === command.activeSessionId, ), ); if (!cachedWorker) { // Passive jobs are managed against their durable store; no wake just to flip a status. const passive = (await this.collectPassiveScheduledJobs()).find( ({ job }) => job.id === command.jobId && job.activeSessionId === command.activeSessionId, ); if (passive) { const store = AgentCronJobStore.forSessionArtifacts(); store.registerSessionArtifact( passive.info.id, getSessionArtifactPathForFile(resolve(passive.info.path), passive.info.id), ); const heartbeat = store.manageHeartbeat(command.activeSessionId, command.jobId, command.action); if (heartbeat) { this.broadcastHeartbeatsChanged(); return success(command.id, "heartbeat_manage", { heartbeat }); } } } const worker = cachedWorker ?? (await this.findWorkerForClient(client, command.activeSessionId)).worker; this.assertWorkerAccessibleToClient(client, worker, command.activeSessionId); const response = await this.forwardToWorker(worker, command); if ( response.success && response.data && typeof response.data === "object" && "heartbeat" in response.data ) { const job = (response.data as { heartbeat?: AgentCronJob }).heartbeat; if (job && worker.heartbeatSnapshot) { const existing = worker.heartbeatSnapshot.find((heartbeat) => heartbeat.job.id === job.id); const remaining = worker.heartbeatSnapshot.filter((heartbeat) => heartbeat.job.id !== job.id); worker.heartbeatSnapshot = job.status === "active" || job.status === "paused" ? [...remaining, existing ? { ...existing, job } : { job }] : remaining; } } return response; } case "cron_add": { const match = await this.findWorkerForClient(client, command.activeSessionId); const response = await this.forwardToWorker(match.worker, command); if (response.success && command.promoteOwnedSession) { await this.promoteOwnedWorker(client, match.worker); } return response; } case "cron_cancel": { if (command.activeSessionId) { const match = await this.findWorkerForClient(client, command.activeSessionId); return this.forwardToWorker(match.worker, command); } const listed = await Promise.all( [...this.workers.values()] .filter( (worker) => this.isLiveWorker(worker) && worker.client && worker.descriptor.lifecycle === "ready", ) .map(async (worker) => ({ worker, response: await this.forwardToWorker( worker, { type: "cron_list", includeInactive: true }, 5000, ).catch(() => undefined), })), ); for (const candidate of listed) { if ( candidate.response?.success && cronJobsFromResponse(candidate.response).some((job) => job.id === command.jobId) ) { return this.forwardToWorker(candidate.worker, command); } } const passive = (await this.collectPassiveScheduledJobs()).find(({ job }) => job.id === command.jobId); if (passive) { const store = AgentCronJobStore.forSessionArtifacts(); store.registerSessionArtifact( passive.info.id, getSessionArtifactPathForFile(resolve(passive.info.path), passive.info.id), ); const job = store.cancel(command.jobId); if (job) { this.broadcastHeartbeatsChanged(); return success(command.id, "cron_cancel", { job }); } } throw new Error(`No cron job found: ${command.jobId}`); } case "heartbeat_get": { const match = await this.findWorkerForClient(client, command.activeSessionId); return this.forwardToWorker(match.worker, command); } case "heartbeat_set": { const match = await this.findWorkerForClient(client, command.activeSessionId); const response = await this.forwardToWorker(match.worker, command); if (response.success && command.promoteOwnedSession) { await this.promoteOwnedWorker(client, match.worker); } return response; } case "heartbeat_update": { const match = await this.findWorkerForClient(client, command.activeSessionId); return this.forwardToWorker(match.worker, command); } case "rename_saved_session": { const target = await this.savedSessionNameReservationInput(command.sessionPath, command.name.trim()); return await this.withSessionNameReservation(target, async () => { await this.assertSupervisorSavedSessionNameAvailable(command.sessionPath, target.name); if (!command.activeSessionId) { await this.catalog.rename(command.sessionPath, command.name); // Third rename write point: an offline saved-session rename // changes the name the ledger carries for that child. await this.rlmSpawnLedger() .appendRenameByChildPath(command.sessionPath, target.name) .catch((error) => { this.log( `failed to append RLM ledger rename: ${error instanceof Error ? error.message : String(error)}`, ); }); const entry = this.roster().bySessionFile(canonicalSessionPath(command.sessionPath)); if (entry) { this.writeRosterEntry({ ...entry, summary: { ...entry.summary, sessionName: target.name } }); } return success(command.id, command.type); } const match = await this.findWorkerForClient(client, command.activeSessionId); return await this.forwardToWorker(match.worker, { ...command, activeSessionId: match.summary.activeSessionId ?? match.summary.id, }); }); } case "delete_saved_session": if (!command.activeSessionId) { const deletedPath = canonicalSessionPath(command.sessionPath); const entry = this.roster().bySessionFile(deletedPath); if (entry?.summary.activeSessionId !== undefined) { throw new Error("Cannot delete the currently active session"); } const owner = this.findWorkerBySessionFile(command.sessionPath); if (owner) { // A client-owned worker's files are invisible to other clients: a foreign delete is an unknown target. this.assertWorkerAccessibleToClient(client, owner, command.sessionPath); if (owner.client && !this.isWorkerStopping(owner)) { return this.forwardToWorker(owner, command); } if (!(await this.reclaimStaleWorkerRegistration(owner))) { throw new Error( `Session worker is ${this.effectiveWorkerState(owner)}; retry the delete once it is reachable`, ); } } await tombstoneSavedSessionDelete(this.rlmSpawnLedger(), command.sessionPath, entry?.summary); const result = await this.catalog.delete(command.sessionPath); if (result.ok && entry && this.roster().get(entry.agentId) === entry) { this.roster().delete(entry.agentId); } return success(command.id, command.type, result); } break; }
if (command.type === "send_message") { // agentOrigin without fromActiveSessionId is trusted only at the direct socket-client boundary. const source = command.fromActiveSessionId ? await this.findWorkerForClient(client, command.fromActiveSessionId) : undefined; let target: WorkerMatch; try { target = await this.findWorkerForClient(client, command.targetActiveSessionId); } catch (error) { if (!(error instanceof Error) || !error.message.startsWith("Unknown active session:")) throw error; const cwd = source?.summary.cwd ?? this.defaultSessionConfig.cwd ?? process.cwd(); let sessionPath: string; try { sessionPath = await this.catalog.resolve( command.targetActiveSessionId, cwd, source?.worker.descriptor.sessionDir ?? this.defaultSessionConfig.sessionDir, ); } catch (catalogError) { // Preserve selector ambiguity so a2a senders can distinguish it from // the original unknown-active-session lookup failure. if (catalogError instanceof Error && catalogError.message.startsWith("Ambiguous session selector")) { throw catalogError; } throw error; } if (source && command.agentOrigin === true) { const targetInfo = await readSessionInfo(sessionPath); if (!targetInfo) throw new Error(`Unknown active session: ${command.targetActiveSessionId}`); assertAgentFamilyReach( this.familyCatalogEntry(source.summary), this.familyCatalogEntry(summaryForInactiveSession(targetInfo)), ); } const worker = await this.createOrReuseWorker(this.protocolClientId(client), { type: "create", sessionPath, continueRecent: false, }); const root = this.roster().byActiveSessionId(worker.descriptor.rootActiveSessionId); const summary = this.findSummaryInWorker(worker, sessionPath) ?? (root ? sessionSummaryFromRosterEntry(root) : undefined); if (!summary) throw new Error("Woken session worker has no target session"); target = { worker, summary }; } const targetActiveSessionId = target.summary.activeSessionId ?? target.summary.id; if (source && command.agentOrigin === true) { assertAgentFamilyReach(this.familyCatalogEntry(source.summary), this.familyCatalogEntry(target.summary)); } if (source) { if ((source.summary.activeSessionId ?? source.summary.id) === targetActiveSessionId) { throw new Error("Agent messaging cannot target the sending session"); } const targetClient = this.requireAvailableWorkerClient(target.worker); const response = await targetClient.requestWorker( { type: "worker_deliver_message", targetActiveSessionId, message: command.message, sender: { activeSessionId: source.summary.activeSessionId ?? source.summary.id, sessionId: source.summary.sessionId, ...(source.summary.sessionName ? { sessionName: source.summary.sessionName } : {}), runtimeKind: source.summary.runtimeKind ?? "top-level", clientId: client.id, }, }, WORKER_REQUEST_TIMEOUT_MS, ); return { ...response, id: command.id, command: command.type }; } return this.forwardToWorker(target.worker, { ...command, targetActiveSessionId }); }
if (!("activeSessionId" in command) || typeof command.activeSessionId !== "string") { throw new Error(`Supervisor cannot route daemon command: ${command.type}`); } const admission = (command.type === "prompt" || command.type === "prompt_and_wait") && command.admissionId ? this.getPromptAdmission(client, command.activeSessionId, command.admissionId) : undefined; try { throwIfAdmissionCancelled(admission); const match = await waitForPromptAdmission( command.type === "set_session_name" && command.workerToken !== undefined ? this.findWorker( command.activeSessionId, (worker) => worker.descriptor.authenticationToken === command.workerToken, ) : this.findWorkerForClient(client, command.activeSessionId), admission?.controller.signal, ); throwIfAdmissionCancelled(admission); const resolvedCommand = { ...command, activeSessionId: match.summary.activeSessionId ?? match.summary.id, ...(admission ? { admissionId: admission.workerAdmissionId } : {}), } as DaemonCommand; if (admission) { admission.worker = match.worker; admission.workerActiveSessionId = match.summary.activeSessionId ?? match.summary.id; } const isRootKill = command.type === "kill" && (match.summary.activeSessionId ?? match.summary.id) === match.worker.descriptor.rootActiveSessionId; if (!isRootKill) { const forward = async () => { const response = await this.forwardToWorker(match.worker, resolvedCommand); if (admission && response.success) admission.status = "owned"; return response; }; if (command.type === "rename" || command.type === "set_session_name") { const reservation = this.summaryNameReservationInput(match.summary, command.name.trim()); return await this.withSessionNameReservation(reservation, async () => { await this.assertSupervisorSessionNameAvailable(match.summary, reservation.name); return forward(); }); } return await forward(); } this.persistWorkerStopTombstone(match.worker, true); const releaseStopOwnership = this.acquireWorkerStopOwnership(match.worker); let response: DaemonResponse; try { response = await this.forwardToWorker(match.worker, resolvedCommand); } finally { try { await this.stopWorker(match.worker, true, false, true); } finally { releaseStopOwnership(); } } return response; } finally { if (admission) this.deletePromptAdmission(admission); } }
private async handleList( client: DaemonSocketClient, command: Extract<DaemonCommand, { type: "list" }>, ): Promise<DaemonResponse> { const active: SessionSummary[] = []; const activeByFile = new Map<string, SessionSummary>(); let busyClientOwnedSessionCount = 0; for (const entry of this.roster().values()) { if (entry.queuedChild) continue; const worker = entry.workerId !== undefined ? this.workers.get(entry.workerId) : undefined; if (worker === undefined) continue; const summary = this.publicSummary(worker, sessionSummaryFromRosterEntry(entry)); if (this.isVisibleWorker(worker)) { active.push(summary); if (summary.sessionFile) activeByFile.set(canonicalSessionPath(summary.sessionFile), summary); continue; } if (summary.sessionFile) activeByFile.set(canonicalSessionPath(summary.sessionFile), summary); if (isSessionSummaryBusy(summary)) busyClientOwnedSessionCount += 1; if (command.includeClientOwned === true && this.isWorkerAccessibleToClient(client, worker)) { active.push(summary); } } const data = { sessions: active, ...(command.includeClientOwned ? { busyClientOwnedSessionCount } : {}), }; if (!command.all) { return success(command.id, "list", data); } const sessionDir = command.sessionDir ?? this.defaultSessionConfig.sessionDir; const scanned = await this.catalog.list(command.cwd ? resolve(command.cwd) : undefined, sessionDir); const cwd = command.cwd ? resolve(command.cwd) : undefined; const merged: SessionSummary[] = []; const servedRows = new Set(active); const mergedActiveFiles = new Set<string>(); const scannedFiles = new Set<string>(); for (const info of scanned) { const file = canonicalSessionPath(info.path); scannedFiles.add(file); const workerRow = activeByFile.get(file); if (workerRow && servedRows.has(workerRow)) { merged.push(workerRow); mergedActiveFiles.add(file); continue; } // The on-disk scan is public: an unserved (client-owned) worker row hides its live metadata only. merged.push(summaryForInactiveSession(info)); } // The boot seed covers registered workers' families only, so dead families (pure on-disk // history, no registered worker anywhere in the tree) are not roster-resident. Their // subagent rows are read from the spawn ledger on demand instead. let spawnEdges: RlmLedgerEdge[] = []; try { spawnEdges = await this.rlmSpawnLedger().liveEdges(); } catch (error) { this.log(`Could not list spawn-ledger sessions: ${String(error)}`); } const spawnParents = new Map(spawnEdges.map((edge) => [canonicalSessionPath(edge.child), edge.parent])); const offlineRows: AgentRosterEntry[] = []; for (const entry of this.roster().values()) { if (entry.queuedChild || entry.summary.activeSessionId !== undefined) continue; if (entry.workerId !== undefined && this.workers.has(entry.workerId)) continue; const file = entry.summary.sessionFile ? canonicalSessionPath(entry.summary.sessionFile) : undefined; if (file === undefined || scannedFiles.has(file) || activeByFile.has(file)) continue; offlineRows.push(entry); } for (const hydrated of await Promise.all(offlineRows.map((entry) => this.hydrateSeededEntry(entry)))) { const summary = sessionSummaryFromRosterEntry(hydrated); if (cwd !== undefined && resolve(summary.cwd) !== cwd) continue; if (!this.matchesListSessionDir(summary, sessionDir, spawnParents)) continue; merged.push(summary); } const unseededFiles = new Set<string>(); for (const edge of spawnEdges) { const childPath = canonicalSessionPath(edge.child); if (scannedFiles.has(childPath) || activeByFile.has(childPath) || unseededFiles.has(childPath)) continue; if (this.roster().hasSessionFile(childPath)) continue; const entry = this.rosterEntryForSpawnLedgerEdge(edge); if (this.roster().has(entry.agentId)) continue; unseededFiles.add(childPath); // Hydrated one at a time, like the boot seed: a large dead-family ledger must not fan // out into one concurrent transcript read per child. const hydrated = await this.hydratedSeedEntry(entry); // The same classification a roster write would have applied: these rows read "inactive". const summary = sessionSummaryFromRosterEntry({ ...hydrated, status: classifySessionRosterStatus(hydrated.summary), }); if (cwd !== undefined && resolve(summary.cwd) !== cwd) continue; if (!this.matchesListSessionDir(summary, sessionDir, spawnParents)) continue; merged.push(summary); } for (const summary of active) { const file = summary.sessionFile ? canonicalSessionPath(summary.sessionFile) : undefined; if (file !== undefined && mergedActiveFiles.has(file)) continue; merged.push(summary); } return success(command.id, "list", { ...data, sessions: merged }); }
private async hydratedSeedEntry<T extends WorkerRosterEntry>(entry: T): Promise<T> { const info = entry.summary.sessionFile ? await readSessionInfo(entry.summary.sessionFile).catch(() => undefined) : undefined; if (!info) return { ...entry, seededCwd: true as const }; const { seededCwd, ...rest } = entry; return { ...rest, summary: { ...entry.summary, cwd: info.cwd } } as T; }
private async hydrateSeededEntry(entry: AgentRosterEntry): Promise<AgentRosterEntry> { if (entry.seededCwd !== true || !entry.summary.sessionFile) return entry; const hydrated = await this.hydratedSeedEntry(entry); if (hydrated.seededCwd === true) return entry; const current = this.roster().get(entry.agentId); if (current !== entry) return current ?? entry; return this.roster().write(hydrated, entry.workerId, entry.statusLabel); }
private matchesListSessionDir( summary: SessionSummary, sessionDir: string | undefined, spawnParents?: ReadonlyMap<string, string>, ): boolean { if (sessionDir === undefined) return true; if (!summary.sessionFile) return false; let file = resolve(summary.sessionFile); let parentSessionPath = summary.parentSessionPath; const visited = new Set<string>(); while (parentSessionPath !== undefined) { const canonical = canonicalSessionPath(parentSessionPath); if (visited.has(canonical)) break; visited.add(canonical); file = resolve(parentSessionPath); // Dead-family ancestors are not roster-resident; the spawn ledger continues the walk. parentSessionPath = this.roster().bySessionFile(canonical)?.summary.parentSessionPath ?? spawnParents?.get(canonical); } return dirname(file) === resolve(sessionDir); }
private async handleSavedSessionList( client: DaemonSocketClient, command: Extract<DaemonCommand, { type: "list_saved_sessions" }>, ): Promise<DaemonResponse> { let cwd: string; let sessionDir: string | undefined; let activeSessionId: string | undefined; if ("activeSessionId" in command) { const match = await this.findWorkerForClient(client, command.activeSessionId); cwd = match.summary.cwd; sessionDir = this.defaultSessionConfig.sessionDir; activeSessionId = match.summary.activeSessionId ?? match.summary.id; } 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 saved = await this.listSavedSessions(command.scope === "current" ? cwd : undefined, sessionDir, callbacks); const sessions = await withPassiveRlmDescendantInfos(saved, 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), }); }
/** * The catalog child process keeps scans out of the daemon; with the persisted * catalog the steady state is one stat per file, so the daemon answers in * process unless PRIME_AGENT_SESSION_CATALOG_DB=child keeps the child. */ private async listSavedSessions( cwd: string | undefined, sessionDir: string | undefined, callbacks?: { onProgress?: (loaded: number, total: number) => void; onSession?: (session: SessionInfo) => void; }, ): Promise<SessionInfo[]> { if (sessionCatalogMode() === "in-process") { return cwd ? SessionManager.list(cwd, sessionDir, callbacks) : SessionManager.listAll(callbacks, sessionDir); } return this.catalog.list(cwd, sessionDir, callbacks); }
private async createOrReuseWorker(clientId: string, command: DaemonCreateCommand): Promise<ResidentWorker> { let createCommand = command; if (command.name !== undefined) { const normalizedName = command.name.trim(); if (!normalizedName) { throw new Error("Session name cannot be empty"); } createCommand = { ...command, name: normalizedName }; } const ownerClientId = command.lifecycle === "client_owned" ? clientId : undefined; if (command.sessionPath) { const activeMatches = this.matchWorkers(command.sessionPath); if ( activeMatches.length === 1 && !(await this.reclaimStaleWorkerRegistration(activeMatches[0]!.worker, command.launchEnv !== undefined)) ) { return this.reuseWorkerForCreate( activeMatches[0]!.worker, ownerClientId, command.sessionPath, createCommand, ); } if (activeMatches.length > 1) { throw new Error(`Ambiguous active session "${command.sessionPath}"`); } const config = mergeAgentSessionRuntimeConfig(this.defaultSessionConfig, command.config); const sessionPath = looksLikeSessionPath(command.sessionPath) ? resolve(command.sessionPath) : await this.catalog.resolve(command.sessionPath, config.cwd ?? process.cwd(), config.sessionDir); createCommand = { ...createCommand, sessionPath }; } const key = createCommand.sessionPath ? canonicalSessionPath(createCommand.sessionPath) : `new:${command.id ? createCommandIdempotencyKey(clientId, command.id) : createActiveSessionId()}`; const pending = this.openingWorkers.get(key); if (pending) { return this.joinOpeningWorker(pending, ownerClientId, createCommand.sessionPath ?? key); } if (createCommand.sessionPath) { const existing = this.findWorkerBySessionFile(createCommand.sessionPath); if (existing && !(await this.reclaimStaleWorkerRegistration(existing, command.launchEnv !== undefined))) { return this.reuseWorkerForCreate(existing, ownerClientId, createCommand.sessionPath, createCommand); } // The reclaim await may have let a concurrent opener register; join it instead of double-launching. const opened = this.openingWorkers.get(key); if (opened) { return this.joinOpeningWorker(opened, ownerClientId, createCommand.sessionPath); } } if (createCommand.sessionPath) { const existing = this.findWorkerBySessionFile(createCommand.sessionPath); if (existing && !(await this.reclaimStaleWorkerRegistration(existing, command.launchEnv !== undefined))) { return this.reuseWorkerForCreate(existing, ownerClientId, createCommand.sessionPath, createCommand); } } const opening = (async () => { if (!createCommand.name) return this.launchWorker(createCommand, undefined, ownerClientId); const savedSiblings = createCommand.sessionPath ? await this.rlmLedgerSiblings(createCommand.sessionPath) : []; const target = savedSiblings.find( (session) => canonicalSessionPath(session.path) === canonicalSessionPath(createCommand.sessionPath!), ); const targetSummary = target ? summaryForInactiveSession(target) : { sessionId: "new-root", rlmDepth: 0 }; const reservation = this.summaryNameReservationInput(targetSummary, createCommand.name); return this.withSessionNameReservation(reservation, async () => { if (target?.parentSessionPath && (target.rlmDepth ?? 0) > 0) { this.assertSavedSiblingNameAvailable(savedSiblings, target, createCommand.name!); } else { await this.assertSupervisorSessionNameAvailable(targetSummary, createCommand.name!); } return this.launchWorker(createCommand, undefined, ownerClientId); }); })(); this.openingWorkers.set(key, opening); if (ownerClientId === undefined) { this.catalogOpeningWorkers.set(key, opening); } try { return await opening; } finally { if (this.openingWorkers.get(key) === opening) { this.openingWorkers.delete(key); } if (this.catalogOpeningWorkers.get(key) === opening) { this.catalogOpeningWorkers.delete(key); } } }
private async reuseWorkerForCreate( worker: ResidentWorker, ownerClientId: string | undefined, sessionPath: string, createCommand?: DaemonCreateCommand, ): Promise<ResidentWorker> { if (worker.descriptor.lifecycle === "failed" && !this.canRetryFailedWorker(worker)) { throw new Error( `Session "${sessionPath}" is registered to a failed worker that could not be safely reclaimed`, ); } await this.assertWorkerCreateOwner(worker, ownerClientId, sessionPath); if (this.canRetryFailedWorker(worker)) { await this.retryWorkerRecovery(worker); } else if (!this.isWorkerReadyForCreate(worker)) { if (worker.recovery) { await worker.recovery; } else if (this.isWorkerRecoveryEligible(worker)) { await this.recoverWorker(worker); } } const current = this.workers.get(worker.descriptor.workerId); if (!current) { if (createCommand) { return this.launchWorker(createCommand, undefined, ownerClientId); } throw new Error(`Session "${sessionPath}" worker recovery was interrupted; retry opening the session`); } await this.assertWorkerCreateOwner(current, ownerClientId, sessionPath); if (!this.isWorkerReadyForCreate(current)) { if (!this.workerHasRosterRoot(current)) { if (typeof this.stopWorker === "function") { await this.stopWorker(current, true, true).catch(() => undefined); } if (createCommand) { return this.launchWorker(createCommand, undefined, ownerClientId); } throw new Error( `Session "${sessionPath}" worker is unavailable for reuse: assigned root session is missing`, ); } const detail = current.descriptor.lastError ? `: ${current.descriptor.lastError}` : ""; throw new Error(`Session "${sessionPath}" worker is ${this.effectiveWorkerState(current)}${detail}`); } return current; }
private async joinOpeningWorker( pending: Promise<ResidentWorker>, ownerClientId: string | undefined, sessionPath: string, ): Promise<ResidentWorker> { const worker = await pending; await this.assertWorkerCreateOwner(worker, ownerClientId, sessionPath); return worker; }
private async assertWorkerCreateOwner( worker: ResidentWorker, ownerClientId: string | undefined, sessionPath: string, ): Promise<void> { if (worker.descriptor.ownerClientId !== ownerClientId) { if (typeof this.processIdentity === "function") { const identity = this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); if (identity === "gone" || identity === "replaced") { await this.reclaimDeadWorkerRegistration(worker); return; } } throw new SessionAlreadyActiveError(sessionPath, worker.descriptor.rootActiveSessionId); } }
private isWorkerReadyForCreate(worker: ResidentWorker): boolean { return ( worker.descriptor.lifecycle === "ready" && worker.client !== undefined && this.workerHasRosterRoot(worker) && !this.isWorkerStopping(worker) ); }
private workerHasRosterRoot(worker: ResidentWorker): boolean { return ( this.roster().byActiveSessionId(worker.descriptor.rootActiveSessionId)?.workerId === worker.descriptor.workerId ); }
/** * A stopping worker whose process already died can strand its registration * (for example when the stop timed out and its finalization was interrupted * by a supervisor restart). Such a registration would block reopening the * saved transcript forever, so complete the interrupted stop and let the * caller launch a fresh worker for the saved session. */ private async reclaimStaleWorkerRegistration(worker: ResidentWorker, freshCreate = false): Promise<boolean> { if (worker.client !== undefined || worker.recovery !== undefined) { return false; } if (worker.descriptor.stopRequestedAt === undefined) { const identity = this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); if (identity === "gone" || identity === "replaced") { await this.reclaimDeadWorkerRegistration(worker); return true; } if (worker.descriptor.lifecycle !== "failed" || worker.descriptor.ownerClientId) { return false; } if (identity === "current") { if (!freshCreate || !worker.descriptor.processStartId) return false; await this.stopWorker(worker, true, true); return true; } return false; } // Fail fast before waiting on anything: only a confirmed-dead process is // reclaimable. A live, unknown, or still-stopping worker is left alone // and the caller reports the session as already active. const identity = this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); if (identity !== "gone" && identity !== "replaced") { return false; } // Single cleanup path: the background stop finalizer is identity-aware, // retrying, and single-flighted, so concurrent resumes share one stop. // The wait is bounded so a resume request always returns promptly. this.scheduleWorkerStopFinalization(worker); const finalization = worker.stopFinalization; if (finalization) { await Promise.race([finalization.catch(() => undefined), unrefDelay(STALE_RECLAIM_WAIT_MS)]); } if (this.workers.get(worker.descriptor.workerId) === worker) { // The process is confirmed dead, so the registration must never be // reused; slow cleanup fails the resume honestly instead. throw new Error( `Stopped session worker ${worker.descriptor.workerId} is still being cleaned up; retry shortly`, ); } this.log(`Reclaimed stale registration for stopped worker ${worker.descriptor.workerId}`); return true; }
/** * Complete the interrupted stop of a worker whose process is confirmed gone. * * Such a registration can never be revived: no pid can be signalled and no * client holds the connection. A registration left behind is what lets a dead * worker block a later update restart, so both the stale-resume reclaim and the * update preparation pass run this one cleanup. */ private async reclaimDeadWorkerRegistration(worker: ResidentWorker): Promise<void> { worker.intentionalStop = true; await this.recoverUncertainWorkerOperations(worker); this.invalidateWorkerSessionInputPauses(worker, "Session worker stopped while input was paused"); this.workers.delete(worker.descriptor.workerId); this.flipWorkerRosterEntriesInactive(worker); this.deleteWorkerDescriptor(worker); }
/** * A worker whose process is provably gone: it recorded a pid, and that pid is * either dead or now belongs to an unrelated process. Such a worker cannot be * revived, and — unlike a worker that is merely unreachable, still recovering, or * waiting for an owning client — it also cannot become ready again, so the update * restart must never wait on it. * * A descriptor without a pid is deliberately not provable here: an unreachable * worker keeps its registration and keeps refusing the restart. */ private isProvablyDeadWorker(worker: ResidentWorker): boolean { if (typeof worker.descriptor.pid !== "number") return false; const identity = this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); return identity === "gone" || identity === "replaced"; }
private async promoteOwnedWorker(client: DaemonSocketClient, worker: ResidentWorker): Promise<void> { const clientId = this.protocolClientId(client); if (worker.descriptor.ownerClientId === undefined && worker.promotedOwnerClientId === clientId) { return; } if (worker.descriptor.ownerClientId !== clientId) { throw new Error("Session is not owned by this client"); } const previousDescriptor = worker.descriptor; worker.descriptor = { ...previousDescriptor, ownerClientId: undefined }; try { this.persistWorker(worker); } catch (error) { worker.descriptor = previousDescriptor; throw error; } worker.promotedOwnerClientId = clientId; for (const entry of this.workerRosterEntries(worker)) { this.roster().amend(entry.agentId, {}); } if (worker.ownerCleanupTimer) { clearTimeout(worker.ownerCleanupTimer); worker.ownerCleanupTimer = undefined; } worker.launchEnv = undefined; worker.transientCreateCommand = undefined; }
private describeWorkerSpawnFailure(error: Error): Error { const errno = (error as NodeJS.ErrnoException).code; const hint = errno === "EMFILE" || errno === "ENFILE" ? ` (${this.workers.size} resident session workers are holding file descriptors; stop unused sessions or raise the open-file limit (ulimit -n))` : ""; return new Error(`Failed to spawn session worker: ${error.message}${hint}`); }
private async launchWorker( command: DaemonCreateCommand, existing?: ResidentWorker, ownerClientId?: string, ): Promise<ResidentWorker> { await this.assertRecoveryAllowed(); if (existing && this.isWorkerRecoveryCancelled(existing)) { throw new Error(`Session worker ${existing.descriptor.workerId} recovery was cancelled`); } const recoveryStopRevision = existing?.stopRevision; const launchEnv = command.launchEnv ?? existing?.launchEnv; const createCommand: DaemonCreateCommand = { ...withoutSupervisorCreateFields(command), config: mergeAgentSessionRuntimeConfig(this.defaultSessionConfig, command.config), }; const workerId = existing?.descriptor.workerId ?? createActiveSessionId(); const rootActiveSessionId = existing?.descriptor.rootActiveSessionId ?? createActiveSessionId(); const socketPath = existing?.descriptor.socketPath ?? workerSocketPath(this.socketPath, workerId); const token = existing?.descriptor.authenticationToken ?? randomBytes(32).toString("base64url"); // Fresh per incarnation: peer transport grants must never survive a worker restart. const workerInstanceId = randomUUID(); const now = new Date().toISOString(); const descriptorPath = existing?.descriptorPath ?? join(this.descriptorDir, `${workerId}.json`); const recoveryJournalPath = existing?.descriptor.recoveryJournalPath ?? join(this.descriptorDir, `${workerId}.recovery.jsonl`); const orphanProcessJournalPath = existing?.descriptor.orphanProcessJournalPath ?? join(this.descriptorDir, `${workerId}.orphans.jsonl`); const launch = createCliSubprocessLaunchSpec(["--mode", "daemon", "--daemon-socket", socketPath]); const workerEnvironment = createCliSubprocessEnv({ ...process.env, ...launchEnv, [DAEMON_WORKER_ROLE_ENV]: "1", [DAEMON_WORKER_TOKEN_ENV]: token, [DAEMON_WORKER_INSTANCE_ID_ENV]: workerInstanceId, [DAEMON_WORKER_ACTIVE_SESSION_ID_ENV]: rootActiveSessionId, [DAEMON_WORKER_SUPERVISOR_SOCKET_ENV]: this.socketPath, [DAEMON_WORKER_RECOVERY_JOURNAL_ENV]: recoveryJournalPath, [DAEMON_WORKER_STARTUP_GATE_FD_ENV]: String(WORKER_STARTUP_GATE_FD), [ORPHAN_PROCESS_JOURNAL_ENV]: orphanProcessJournalPath, [SESSION_LEASES_ENABLED_ENV]: "1", [SESSION_LEASE_OWNER_ID_ENV]: rootActiveSessionId, }); delete workerEnvironment.RLM_DEPTH; await this.assertRecoveryAllowed(); const child: ChildProcess = spawnHidden(launch.command, launch.args, { cwd: createCommand.config?.cwd ?? process.cwd(), detached: true, env: workerEnvironment, stdio: ["ignore", "ignore", "pipe", "pipe"], }); const detachWorkerStderr = child.stderr ? attachJsonlLineReader(child.stderr, (line) => this.log(`Session worker ${workerId} stderr: ${line}`), { maxLineLength: 64 * 1024, onLineOverflow: (prefix) => this.log(`Session worker ${workerId} stderr: ${prefix} [truncated]`), }) : () => {}; child.once("close", detachWorkerStderr); child.once("exit", (code, signal) => { const stopReason = worker?.stopReason; const stopSignal = worker?.stopSignalSent; let attribution = `code=${code === null ? "null" : code}, signal=${signal ?? "none"}`; if (stopReason) { attribution += `, reason=${stopReason}`; } if (stopSignal) { attribution += `, signalSent=${stopSignal}`; } this.log(`Session worker ${workerId} process exit: ${attribution}`); }); const childClosed = new Promise<void>((resolveClose) => child.once("close", () => resolveClose())); let spawnFailure: Error | undefined; const spawnSettled = new Promise<void>((resolveSpawn) => { child.once("spawn", () => resolveSpawn()); child.once("error", (error) => { spawnFailure = error instanceof Error ? error : new Error(String(error)); resolveSpawn(); }); }); child.on("error", (error) => { this.log( `Session worker ${workerId} process error: ${error instanceof Error ? error.message : String(error)}`, ); }); // A failed spawn (e.g. EMFILE) leaves child.stdio undefined. const startupGate = child.stdio?.[WORKER_STARTUP_GATE_FD]; const previousDescriptor = existing?.descriptor; const previousIntentionalStop = existing?.intentionalStop; let descriptorAssigned = false; let childPid: number; let childProcessStartId: string | undefined; let worker: ResidentWorker; try { await spawnSettled; if (spawnFailure) { throw this.describeWorkerSpawnFailure(spawnFailure); } if (!child.pid) { throw new Error("Failed to obtain daemon session worker pid"); } if (!(startupGate instanceof Writable)) { throw new Error("Failed to create daemon session worker startup gate"); } childPid = child.pid; childProcessStartId = getProcessStartId(childPid); await this.assertRecoveryAllowed();
const descriptor: DaemonWorkerDescriptor = { version: 2, workerId, pid: childPid, ...(childProcessStartId ? { processStartId: childProcessStartId } : {}), socketPath, recoveryJournalPath, orphanProcessJournalPath, supervisorSocketPath: this.socketPath, authenticationToken: token, workerInstanceId, rootActiveSessionId, ownerClientId: existing?.descriptor.ownerClientId ?? ownerClientId, sessionDir: createCommand.config?.sessionDir, telemetryDisabled: createCommand.config?.telemetryDisabled, createdAt: existing?.descriptor.createdAt ?? now, updatedAt: now, lifecycle: "starting", createCommand: durableDaemonCreateCommand(createCommand), consecutiveFailures: existing?.descriptor.consecutiveFailures ?? 0, }; worker = existing ?? { descriptor, descriptorPath, summaries: new Map(), snapshotCache: new Map(), transcriptCaches: new Map(), snapshotGenerations: new Map(), snapshotLoads: new Map(), intentionalStop: false, stopRevision: 0, launchEnv, transientCreateCommand: ownerClientId ? createCommand : undefined, }; await this.assertRecoveryAllowed(); worker.descriptor = descriptor; worker.launchEnv = launchEnv; worker.transientCreateCommand = descriptor.ownerClientId ? createCommand : undefined; descriptorAssigned = true; this.persistWorker(worker); worker.intentionalStop = false; this.workers.set(workerId, worker); } catch (error) { if (startupGate instanceof Writable) { startupGate.destroy(); } await childClosed; child.unref(); try { rmSync(`${descriptorPath}.${process.pid}.tmp`, { force: true }); } catch (cleanupError) { this.reportCleanupFailure(`worker launch temp ${workerId}`, cleanupError); } if (existing && descriptorAssigned && previousDescriptor) { try { existing.descriptor = previousDescriptor; } catch (cleanupError) { this.reportCleanupFailure(`worker launch descriptor ${workerId}`, cleanupError); } } throw error; }
try { try { await commitWorkerStartupGate(startupGate); } catch (error) { startupGate.destroy(); await childClosed; throw error; } finally { child.unref(); } const client = await this.connectWorker(worker, WORKER_CONNECT_TIMEOUT_MS); const response = await client.request(withoutCommandId(createCommand), WORKER_REQUEST_TIMEOUT_MS); if (!response.success) { throw deserializeDaemonError(response); } if (!isSessionSummary(response.data)) { throw new Error("Session worker returned an invalid create response"); } const summary = response.data; if ((summary.activeSessionId ?? summary.id) !== rootActiveSessionId) { throw new Error("Session worker did not preserve its assigned active session id"); } this.writeRosterEntry(workerRosterEntryFromSummary(summary), worker); worker.descriptor.rootSessionId = summary.sessionId; worker.descriptor.sessionFile = summary.sessionFile; await this.subscribeWorker(worker, rootActiveSessionId); await this.refreshWorkerSummaries(worker, true); if (existing && (this.isWorkerRecoveryCancelled(worker) || worker.stopRevision !== recoveryStopRevision)) { throw new Error(`Session worker ${workerId} recovery was cancelled`); } await this.assertRecoveryAllowed(); worker.descriptor.lifecycle = "ready"; worker.descriptor.consecutiveFailures = 0; worker.deferredRecoveryRounds = 0; worker.descriptor.lastError = undefined; this.persistWorker(worker); if (!worker.descriptor.ownerClientId) { worker.launchEnv = undefined; worker.transientCreateCommand = undefined; } this.broadcastHeartbeatsChanged(); return worker; } catch (error) { if (isSupervisorGenerationStale(error)) { throw error; } if (isSupervisorShutdownAdmissionCancelled(error)) { let rolledBack = false; try { await this.stopWorker(worker, existing === undefined, true, false, existing !== undefined, { child, closed: childClosed, }); rolledBack = true; } catch (cleanupError) { this.reportCleanupFailure(`cancelled worker launch ${workerId}`, cleanupError); } const mappedWorker = this.workers.get(workerId); if ( rolledBack && existing && previousDescriptor && !this.shuttingDown && existing.stopRevision === recoveryStopRevision && existing.descriptor.stopRequestedAt === undefined && (mappedWorker === undefined || mappedWorker === existing) ) { existing.descriptor = previousDescriptor; existing.intentionalStop = previousIntentionalStop ?? false; this.workers.set(workerId, existing); try { this.persistWorker(existing); } catch (cleanupError) { this.reportCleanupFailure(`cancelled worker recovery ${workerId}`, cleanupError); } this.deferWorkerRecovery(existing, error instanceof Error ? error : new Error(String(error))); } throw error; } await this.assertRecoveryAllowed(); const shouldResumeRecovery = existing !== undefined && !this.shuttingDown && worker.descriptor.stopRequestedAt === undefined && worker.stopRevision === recoveryStopRevision; await this.stopWorker(worker, existing === undefined, true, false, existing !== undefined).catch((stopError) => this.log(`Could not stop failed worker ${workerId}: ${String(stopError)}`), ); if ( shouldResumeRecovery && !this.shuttingDown && worker.descriptor.stopRequestedAt === undefined && worker.stopRevision === recoveryStopRevision ) { await this.assertRecoveryAllowed(); worker.intentionalStop = false; worker.descriptor.lifecycle = "recovering"; this.workers.set(workerId, worker); this.persistWorker(worker); } throw error; } }
private async connectWorker(worker: ResidentWorker, timeoutMs: number): Promise<DaemonWorkerClient> { const deadline = Date.now() + timeoutMs; let lastError: unknown; let backoffMs = WORKER_PROBE_BACKOFF_MIN_MS; while (Date.now() < deadline) { await this.assertRecoveryAllowed(); const client = new DaemonWorkerClient(worker.descriptor.socketPath); try { await client.connect(Math.min(WORKER_CONNECT_PROBE_MS, handshakeBudgetMs(deadline))); await client.waitForHello(handshakeBudgetMs(deadline)); // Listen before authenticating: the worker flushes its roster snapshot right after auth succeeds. client.onFrame((frame) => this.handleWorkerFrame(worker, frame, client)); client.onClose((error) => void this.handleWorkerClose(worker, client, error)); worker.pendingClient = client; try { const authResponse = await client.authenticateWorker( worker.descriptor.authenticationToken, { ...this.supervisorAuthenticationClaim(), ...(worker.descriptor.workerInstanceId !== undefined ? { workerInstanceId: worker.descriptor.workerInstanceId } : {}), }, handshakeBudgetMs(deadline), ); await this.assertRecoveryAllowed(); if (!workerAuthAdvertisesRoster(authResponse.data)) { throw new PreRosterWorkerError("Session worker predates the roster protocol and must be restarted"); } worker.peerTransportCapable = workerAuthAdvertisesPeerTransport(authResponse.data); worker.lastFrameAt = Date.now(); worker.client?.close(); worker.client = client; return client; } finally { if (worker.pendingClient === client) worker.pendingClient = undefined; } } catch (error) { lastError = error; client.close(); if ( isSupervisorRecoveryCancelled(error) || error instanceof PreRosterWorkerError || error instanceof DaemonWorkerAuthenticationError ) { throw error; } const remaining = deadline - Date.now(); if (remaining <= 0) break; await delay(Math.min(backoffMs, remaining)); backoffMs = Math.min(backoffMs * 2, WORKER_PROBE_BACKOFF_MAX_MS); } } throw new DaemonWorkerProbeTimeoutError(`Timed out connecting to daemon session worker: ${String(lastError)}`); }
private async subscribeWorker(worker: ResidentWorker, activeSessionId: string): Promise<void> { if (!worker.client) { throw new Error("Session worker is not connected"); } const supportsExtensionUi = [...this.clients].some( (client) => client.attachedActiveSessionIds.has(activeSessionId) && client.supportsExtensionUi, ); const response = await worker.client.requestWorker({ type: "worker_subscribe", activeSessionId, capabilities: supportsExtensionUi ? ["attach_snapshot", "event_sequence", "extension_ui", "slim_attach", "chunked_snapshot"] : ["attach_snapshot", "event_sequence", "slim_attach", "chunked_snapshot"], supportsExtensionUi, }); if (!response.success) { throw new Error(response.error); } }
private async adoptOrRecoverWorker(worker: ResidentWorker): Promise<void> { await this.assertRecoveryAllowed(); if (worker.descriptor.stopRequestedAt) { try { // A descriptor persisted before identity tracking has no // processStartId, so stopWorker could neither signal the live // process nor let the finalizer escalate. Authenticating on the // worker's socket proves the pid still belongs to our worker, so // the start id observed while it was alive can be persisted (and // the connected client gives stopWorker its graceful IPC path). if (worker.descriptor.processStartId === undefined && isProcessAlive(worker.descriptor.pid)) { const observedProcessStartId = getProcessStartId(worker.descriptor.pid); try { await this.connectWorker(worker, WORKER_CONNECT_TIMEOUT_MS); if (observedProcessStartId) { worker.descriptor.processStartId = observedProcessStartId; this.persistWorker(worker); } } catch { // Unverifiable identity stays untrusted; the stop below // still runs its graceful path and the finalizer keeps // waiting rather than signalling a possibly-recycled pid. } } await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); this.log(`Completed intentional stop for worker ${worker.descriptor.workerId} during supervisor adoption`); } catch (error) { worker.descriptor.lifecycle = "failed"; worker.descriptor.lastError = error instanceof Error ? error.message : String(error); this.persistWorker(worker); this.log(`Could not complete intentional stop for worker ${worker.descriptor.workerId}: ${String(error)}`); } return; } let observedProcessStartId: string | undefined; try { if (!isProcessAlive(worker.descriptor.pid)) { throw new Error("Session worker process is no longer running"); } observedProcessStartId = getProcessStartId(worker.descriptor.pid); await this.connectWorker(worker, WORKER_CONNECT_TIMEOUT_MS); await this.subscribeWorker(worker, worker.descriptor.rootActiveSessionId); await this.refreshWorkerSummaries(worker, true); if (worker.descriptor.processStartId === undefined && observedProcessStartId) { worker.descriptor.processStartId = observedProcessStartId; } await this.assertRecoveryAllowed(); worker.descriptor.lifecycle = "ready"; worker.descriptor.consecutiveFailures = 0; worker.deferredRecoveryRounds = 0; this.persistWorker(worker); this.broadcastHeartbeatsChanged(); } catch (error) { if (isSupervisorRecoveryCancelled(error)) { return; } this.log(`Could not adopt worker ${worker.descriptor.workerId}: ${String(error)}`); // A client-owned worker's launch env lives only with its owner; recoverWorker parks it instead. if (error instanceof PreRosterWorkerError && worker.descriptor.ownerClientId === undefined) { try { await this.restartPreRosterWorker(worker, observedProcessStartId); return; } catch (restartError) { if (isSupervisorRecoveryCancelled(restartError)) { return; } this.log(`Could not restart pre-roster worker ${worker.descriptor.workerId}: ${String(restartError)}`); } } if (String(error).includes("waiting for daemon worker hello") || String(error).includes("handshake")) { await this.stopWorker(worker, true, true).catch((stopError) => this.log(`Could not stop unresponsive worker ${worker.descriptor.workerId}: ${String(stopError)}`), ); return; } const identityNow = this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); if (isDaemonWorkerProbeTimeout(error) && (identityNow === "current" || identityNow === "unknown")) { worker.descriptor.lifecycle = "recovering"; worker.descriptor.lastError = error instanceof Error ? error.message : String(error); this.persistWorker(worker); void this.recoverWorker(worker).catch((recoveryError) => this.log(`Could not recover worker ${worker.descriptor.workerId}: ${String(recoveryError)}`), ); return; } await this.recoverWorker(worker); } }
private async restartPreRosterWorker( worker: ResidentWorker, observedProcessStartId: string | undefined, ): Promise<void> { await this.assertRecoveryAllowed(); if (worker.descriptor.processStartId === undefined && observedProcessStartId !== undefined) { worker.descriptor.processStartId = observedProcessStartId; } const identity = () => this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); // The one deliberate kill of a live worker: it authenticated as ours and predates the roster // protocol, so this identity-verified upgrade replaces it. No await between recheck and signal. if (identity() === "current") { signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); // SIGKILL is uninterceptable; this wait only covers kernel teardown of the old process and socket. const killDeadline = Date.now() + 1000; while (identity() === "current" && Date.now() < killDeadline) { await delay(25); } } const finalIdentity = identity(); if (finalIdentity !== "gone" && finalIdentity !== "replaced") { // A live or unverifiable survivor parks failed with no destructive cleanup: interruption // marking and orphan reaping must never run against a possibly-active worker. worker.descriptor.lifecycle = "failed"; worker.descriptor.lastError = `Pre-roster worker process ${worker.descriptor.pid} is still running and cannot be replaced safely`; this.persistWorker(worker); this.markWorkerRosterEntries(worker, "failed"); this.log(`Kept pre-roster worker ${worker.descriptor.workerId} failed: ${worker.descriptor.lastError}`); return; } await this.recoverUncertainWorkerOperations(worker); if (this.isWorkerRecoveryCancelled(worker)) { return; } await this.launchWorker(worker.descriptor.createCommand, worker, worker.descriptor.ownerClientId); }
private async handleWorkerClose(worker: ResidentWorker, client: DaemonWorkerClient, error: Error): Promise<void> { if (worker.client !== client) { return; } worker.client = undefined; this.invalidateWorkerSessionInputPauses(worker, "Session worker disconnected while input was paused"); const interrupted = new Map<string, Set<string>>(); for (const [activeSessionId, generations] of worker.snapshotGenerations ?? []) { for (const generation of generations.values()) { if (generation.incoming || !generation.transcript.complete) { const snapshotIds = interrupted.get(activeSessionId) ?? new Set<string>(); snapshotIds.add(generation.transcript.snapshotId); interrupted.set(activeSessionId, snapshotIds); } } } for (const [activeSessionId, transcript] of worker.transcriptCaches) { if (!transcript.complete) { const snapshotIds = interrupted.get(activeSessionId) ?? new Set<string>(); snapshotIds.add(transcript.snapshotId); interrupted.set(activeSessionId, snapshotIds); } } for (const [activeSessionId, snapshotIds] of interrupted) { for (const snapshotId of snapshotIds) { this.failWorkerSnapshotCache( worker, activeSessionId, new Error("Session worker disconnected during snapshot transfer"), false, snapshotId, ); } } if (this.shuttingDown || worker.intentionalStop) { return; } this.markWorkerRosterEntries(worker, "recovering"); try { await this.assertRecoveryAllowed(); } catch (recoveryError) { if (!isSupervisorGenerationStale(recoveryError)) { this.deferWorkerRecovery(worker, error); } return; } if (!this.isWorkerRecoveryEligible(worker)) { return; } worker.descriptor.lifecycle = "recovering"; worker.descriptor.lastError = error.message; this.persistWorker(worker); void this.recoverWorker(worker); }
private isWorkerRecoveryEligible(worker: ResidentWorker): boolean { return this.isWorkerRecoveryCandidate(worker) && worker.recovery === undefined; }
/** Failed is not terminal for an identity-verified live worker: any touch retries recovery, like manual retry_worker. */ private canRetryFailedWorker(worker: ResidentWorker): boolean { return ( worker.descriptor.lifecycle === "failed" && (this.workerStopCounts?.get(worker) ?? 0) === 0 && // A user-stopped worker stays stopped; only an explicit retry_worker clears the persisted stop markers. !this.isWorkerStopping(worker) && this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId) === "current" ); }
private async retryWorkerRecovery(worker: ResidentWorker): Promise<void> { worker.intentionalStop = false; worker.descriptor.stopRequestedAt = undefined; worker.descriptor.archiveOnStop = undefined; worker.descriptor.lifecycle = "recovering"; worker.descriptor.consecutiveFailures = 0; worker.deferredRecoveryRounds = 0; this.persistWorker(worker); await this.recoverWorker(worker); }
private isWorkerRecoveryCandidate(worker: ResidentWorker): boolean { return ( !this.shuttingDown && !worker.intentionalStop && worker.descriptor.stopRequestedAt === undefined && this.workers.get(worker.descriptor.workerId) === worker && worker.client === undefined ); }
private deferWorkerRecovery(worker: ResidentWorker, disconnectError: Error): void { if (worker.deferredRecovery) { return; } if (worker.descriptor.consecutiveFailures >= MAX_CONSECUTIVE_RECOVERY_FAILURES) { void this.stopWorker(worker, true, true).catch(() => undefined); this.log( `Worker ${worker.descriptor.workerId} reached recovery failure limit (${worker.descriptor.consecutiveFailures}); record discarded`, ); return; } // A live-but-silent worker must not probe forever: park it failed (user-visible through the // roster's failed status) and keep its process alive. The park is not terminal: retry_worker, // attach, and create all retry recovery while the process identity stays current. worker.deferredRecoveryRounds = (worker.deferredRecoveryRounds ?? 0) + 1; if (worker.deferredRecoveryRounds > MAX_DEFERRED_RECOVERY_ROUNDS) { worker.descriptor.lifecycle = "failed"; worker.descriptor.lastError = `Live session worker did not answer recovery probes for ${MAX_DEFERRED_RECOVERY_ROUNDS} rounds: ${disconnectError.message}`; this.persistWorker(worker); this.markWorkerRosterEntries(worker, "failed"); this.log( `Worker ${worker.descriptor.workerId} is unresponsive; parked failed after ${MAX_DEFERRED_RECOVERY_ROUNDS} probe rounds`, ); return; } worker.deferredRecovery = this.resumeDeferredWorkerRecovery(worker, disconnectError).finally(() => { worker.deferredRecovery = undefined; }); }
private async resumeDeferredWorkerRecovery(worker: ResidentWorker, disconnectError: Error): Promise<void> { while (true) { await unrefDelay(DEFERRED_RECOVERY_RECHECK_MS); if (!this.isWorkerRecoveryCandidate(worker)) { return; } if (!this.isWorkerRecoveryEligible(worker)) { continue; } try { await this.assertRecoveryAllowed(); } catch (error) { if (isSupervisorGenerationStale(error)) { return; } continue; } if (!this.isWorkerRecoveryCandidate(worker)) { return; } if (!this.isWorkerRecoveryEligible(worker)) { continue; } worker.descriptor.lifecycle = "recovering"; worker.descriptor.lastError = disconnectError.message; this.persistWorker(worker); void this.recoverWorker(worker); return; } }
private failWorkerSnapshotCache( worker: ResidentWorker, activeSessionId: string, error: Error, closeWorkerChannel = false, expectedSnapshotId?: string, ): void { const generations = worker.snapshotGenerations?.get(activeSessionId); if (expectedSnapshotId) { const generation = generations?.get(expectedSnapshotId); if (generation) { this.failSnapshotGeneration(worker, activeSessionId, generation, error); } else { const transcript = worker.transcriptCaches.get(activeSessionId); if (transcript?.snapshotId !== expectedSnapshotId) { return; } transcript.markFailed(error); transcript.dispose(); worker.transcriptCaches.delete(activeSessionId); if (worker.snapshotCache.get(activeSessionId)?.snapshotStream?.id === expectedSnapshotId) { worker.snapshotCache.delete(activeSessionId); } } } else { const failedTranscripts = new Set<SnapshotTranscriptCache>(); for (const generation of [...(generations?.values() ?? [])]) { failedTranscripts.add(generation.transcript); this.failSnapshotGeneration(worker, activeSessionId, generation, error); } const transcript = worker.transcriptCaches.get(activeSessionId); if (transcript && !failedTranscripts.has(transcript)) { transcript.markFailed(error); transcript.dispose(); } worker.transcriptCaches.delete(activeSessionId); worker.snapshotCache.delete(activeSessionId); } if (closeWorkerChannel) { const client = worker.client; if (client) { this.handleWorkerClose(worker, client, error); client.close(); } } }
/** Settle a transfer anomaly with the transfer as the blast radius: the worker channel stays up and clients resync fresh. */ private failSnapshotTransfer( worker: ResidentWorker, activeSessionId: string, snapshotId: string, error: Error, snapshotPurpose: Extract<DaemonWorkerFrameHeader, { kind: "outbound" }>["snapshotPurpose"], ): void { const published = worker.transcriptCaches.get(activeSessionId)?.snapshotId === snapshotId; this.failWorkerSnapshotCache(worker, activeSessionId, error, false, snapshotId); // The published-cache drop drives the resync, not the frame's purpose: a published transfer // can be serving any client's catch-up wait, whose queue entry drainClientCatchups already cleared. if (published) { this.queueSnapshotResync(activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "catchup"); } }
private queueSnapshotResync(activeSessionId: string, snapshotPurpose: "replacement" | "catchup"): void { for (const client of this.clients) { if (!client.attachedActiveSessionIds.has(activeSessionId)) continue; this.queueCatchup(client, activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "resync"); void this.catchUpClient(client).catch((error) => this.log(`Failed to catch up client ${client.id}: ${String(error)}`), ); } }
private retireWorkerSnapshotCache( worker: ResidentWorker, activeSessionId: string, expectedTranscript: SnapshotTranscriptCache, ): void { if (worker.transcriptCaches.get(activeSessionId) === expectedTranscript) { worker.transcriptCaches.delete(activeSessionId); } if (worker.snapshotCache.get(activeSessionId)?.snapshotStream?.id === expectedTranscript.snapshotId) { worker.snapshotCache.delete(activeSessionId); } const generation = this.snapshotGeneration(worker, activeSessionId, expectedTranscript.snapshotId); if (!generation) { expectedTranscript.dispose(); return; } generation.retired = true; this.settleSnapshotDuplicateValidation(generation); if (generation.incoming) { return; } this.deleteSnapshotGeneration(worker, activeSessionId, generation); expectedTranscript.dispose(); }
private snapshotGenerationsFor( worker: ResidentWorker, activeSessionId: string, ): Map<string, SnapshotTranscriptGeneration> { worker.snapshotGenerations ??= new Map(); let generations = worker.snapshotGenerations.get(activeSessionId); if (!generations) { generations = new Map(); worker.snapshotGenerations.set(activeSessionId, generations); } return generations; }
private snapshotGeneration( worker: ResidentWorker, activeSessionId: string, snapshotId: string, ): SnapshotTranscriptGeneration | undefined { return worker.snapshotGenerations?.get(activeSessionId)?.get(snapshotId); }
private currentSnapshotGeneration( worker: ResidentWorker, activeSessionId: string, ): SnapshotTranscriptGeneration | undefined { worker.transcriptCaches ??= new Map(); worker.snapshotCache ??= new Map(); const transcript = worker.transcriptCaches.get(activeSessionId); if (!transcript) { return undefined; } const generations = this.snapshotGenerationsFor(worker, activeSessionId); let generation = generations.get(transcript.snapshotId); if (generation) { return generation; } const result = worker.snapshotCache.get(activeSessionId); if (!result) { return undefined; } generation = { transcript, result, incoming: false, retired: false, }; generations.set(transcript.snapshotId, generation); return generation; }
private deleteSnapshotGeneration( worker: ResidentWorker, activeSessionId: string, generation: SnapshotTranscriptGeneration, ): void { const generations = worker.snapshotGenerations?.get(activeSessionId); if (!generations) { return; } if (generations.get(generation.transcript.snapshotId) === generation) { generations.delete(generation.transcript.snapshotId); } if (generations.size === 0) { worker.snapshotGenerations.delete(activeSessionId); } }
private failSnapshotGeneration( worker: ResidentWorker, activeSessionId: string, generation: SnapshotTranscriptGeneration, error: Error, ): void { this.settleSnapshotDuplicateValidation(generation, error); generation.transcript.markFailed(error); generation.transcript.dispose(); this.deleteSnapshotGeneration(worker, activeSessionId, generation); if (worker.transcriptCaches.get(activeSessionId) === generation.transcript) { worker.transcriptCaches.delete(activeSessionId); } if (worker.snapshotCache.get(activeSessionId)?.snapshotStream?.id === generation.transcript.snapshotId) { worker.snapshotCache.delete(activeSessionId); } }
private createSnapshotDuplicateValidation(): SnapshotDuplicateValidation { let resolve!: () => void; let reject!: (error: Error) => void; const promise = new Promise<void>((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; }); void promise.catch(() => undefined); return { promise, resolve, reject }; }
private settleSnapshotDuplicateValidation(generation: SnapshotTranscriptGeneration, error?: Error): void { const validation = generation.validation; if (!validation) { return; } generation.validation = undefined; if (error) { validation.reject(error); } else { validation.resolve(); } }
private async recoverWorker(worker: ResidentWorker): Promise<void> { if (this.isWorkerRecoveryCancelled(worker)) { return; } if (worker.descriptor.ownerClientId && !worker.launchEnv && !isProcessAlive(worker.descriptor.pid)) { worker.descriptor.lifecycle = "failed"; worker.descriptor.lastError = "Waiting for the owning client to reconnect"; this.persistWorker(worker); return; } if (worker.recovery) { return worker.recovery; } worker.recovery = (async () => { let keepProbingLiveWorker = false; for (const retryDelay of WORKER_RETRY_DELAYS_MS) { await delay(retryDelay); keepProbingLiveWorker = false; if (this.isWorkerRecoveryCancelled(worker)) { return; } try { await this.assertRecoveryAllowed(); const identityNow = this.processIdentity(worker.descriptor.pid, worker.descriptor.processStartId); const identityCompatible = identityNow === "current" || (identityNow === "unknown" && worker.descriptor.processStartId === undefined); if (identityCompatible) { try { await this.connectWorker(worker, WORKER_CONNECT_TIMEOUT_MS); await this.subscribeWorker(worker, worker.descriptor.rootActiveSessionId); await this.refreshWorkerSummaries(worker, true); if (this.isWorkerRecoveryCancelled(worker)) { return; } if (worker.descriptor.processStartId === undefined) { const observedProcessStartId = getProcessStartId(worker.descriptor.pid); if (observedProcessStartId) { worker.descriptor.processStartId = observedProcessStartId; } } await this.assertRecoveryAllowed(); worker.descriptor.lifecycle = "ready"; worker.descriptor.consecutiveFailures = 0; worker.deferredRecoveryRounds = 0; this.persistWorker(worker); this.broadcastHeartbeatsChanged(); return; } catch (error) { if (isSupervisorRecoveryCancelled(error)) { throw error; } await this.assertRecoveryAllowed(); worker.client?.close(); worker.client = undefined; // A worker with the same durable process identity may be load-slow. // Keep probing it instead of replacing live work after a timeout. keepProbingLiveWorker = isDaemonWorkerProbeTimeout(error) || identityNow !== "current"; throw error; } } if (identityNow === "unknown") { keepProbingLiveWorker = true; throw new Error( `Cannot safely replace live session worker ${worker.descriptor.workerId} without a verified process identity`, ); } const recoveryCommand = worker.descriptor.ownerClientId ? worker.transientCreateCommand : undefined; if (!recoveryCommand || !worker.launchEnv) { await this.recoverUncertainWorkerOperations(worker); worker.descriptor.lifecycle = "failed"; worker.descriptor.lastError = "Waiting for a client with fresh runtime context"; this.persistWorker(worker); this.markWorkerRosterEntries(worker, "failed"); return; } await this.recoverUncertainWorkerOperations(worker); if (this.isWorkerRecoveryCancelled(worker)) { return; } await this.launchWorker(recoveryCommand, worker, worker.descriptor.ownerClientId); return; } catch (error) { if (isSupervisorRecoveryCancelled(error) || this.isWorkerRecoveryCancelled(worker)) { return; } try { await this.assertRecoveryAllowed(); } catch { return; } worker.client?.close(); worker.client = undefined; worker.descriptor.consecutiveFailures++; worker.descriptor.lastFailureAt = new Date().toISOString(); worker.descriptor.lastError = error instanceof Error ? error.message : String(error); this.persistWorker(worker); if (worker.descriptor.consecutiveFailures >= MAX_CONSECUTIVE_RECOVERY_FAILURES) { if (typeof this.stopWorker === "function") { await this.stopWorker(worker, true, true); } this.log( `Worker ${worker.descriptor.workerId} failed recovery ${worker.descriptor.consecutiveFailures} times; record discarded`, ); return; } } } if (keepProbingLiveWorker) { try { await this.assertRecoveryAllowed(); } catch { return; } worker.descriptor.lifecycle = "recovering"; this.persistWorker(worker); this.deferWorkerRecovery( worker, new Error(worker.descriptor.lastError ?? "Live session worker did not answer recovery probes"), ); return; } try { await this.assertRecoveryAllowed(); } catch { return; } // Leak-over-kill: a live worker that keeps failing for non-timeout reasons parks failed with // its process intact. A verified-identity survivor is reclaimed by the next fresh create; // an unverifiable one waits for exit — killing a pid we cannot verify as ours is worse. worker.descriptor.lifecycle = "failed"; this.persistWorker(worker); this.markWorkerRosterEntries(worker, "failed"); this.log(`Worker ${worker.descriptor.workerId} failed after three recovery attempts`); })().finally(() => { worker.recovery = undefined; }); return worker.recovery; }
private isWorkerRecoveryCancelled(worker: ResidentWorker): boolean { return ( this.shuttingDown || worker.intentionalStop || worker.descriptor.stopRequestedAt !== undefined || this.workers.get(worker.descriptor.workerId) !== worker ); }
private isWorkerCleanupCancelled(worker: ResidentWorker): boolean { return ( this.shuttingDown || worker.descriptor.stopRequestedAt !== undefined || this.workers.get(worker.descriptor.workerId) !== worker ); }
private async recoverUncertainWorkerOperations(worker: ResidentWorker): Promise<void> { await this.assertRecoveryAllowed(); const journal = new WorkerRecoveryJournal(worker.descriptor.recoveryJournalPath); const latest = journal.getLatest(); const uncertain = latest.filter((record) => record.busy); if (uncertain.length > 0) await this.catalog.start(); await this.assertRecoveryAllowed(); if (this.isWorkerCleanupCancelled(worker)) { throw new SupervisorRecoveryCancelledError("Worker recovery was cancelled before destructive cleanup"); }
const interruptedSessions = new Map< string, { activeSessionId: string; sessionFile: string; operations: Set<string> } >(); for (const record of uncertain) { const sessionFile = record.sessionFile ?? (record.activeSessionId === worker.descriptor.rootActiveSessionId ? worker.descriptor.sessionFile : undefined); if (!sessionFile) { continue; } const key = `${record.activeSessionId}\0${sessionFile}`; let interrupted = interruptedSessions.get(key); if (!interrupted) { interrupted = { activeSessionId: record.activeSessionId, sessionFile, operations: new Set() }; interruptedSessions.set(key, interrupted); } interrupted.operations.add(record.operation); }
await this.assertRecoveryAllowed(); if (this.isWorkerCleanupCancelled(worker)) { throw new SupervisorRecoveryCancelledError("Worker recovery was cancelled before interruption was recorded"); } await Promise.all( [...interruptedSessions.values()].map((interrupted) => this.catalog.markInterrupted(interrupted.sessionFile, interrupted.activeSessionId, [ ...interrupted.operations, ]), ), ); await this.assertRecoveryAllowed(); if (this.isWorkerCleanupCancelled(worker)) { throw new SupervisorRecoveryCancelledError("Worker recovery was cancelled before process cleanup"); } const orphanProcessJournalPath = worker.descriptor.orphanProcessJournalPath; if (orphanProcessJournalPath) { try { const orphans = readActiveOrphanProcesses(orphanProcessJournalPath, worker.descriptor.pid); let reapFailed = false; let retryIsSafe = true; for (const orphan of orphans) { if (orphan.processStartId === undefined) retryIsSafe = false; if (!shouldReapOrphanProcess(orphan)) { continue; } if (!killOrphanProcess(orphan.pid)) reapFailed = true; } if (!reapFailed || !retryIsSafe) clearOrphanProcessJournal(orphanProcessJournalPath); } catch (error) { this.log(`Could not reap orphaned worker resources: ${String(error)}`); } } if (uncertain.length === 0) { return; } for (const record of latest) { journal.record({ activeSessionId: record.activeSessionId, sessionId: record.sessionId, ...(record.sessionFile ? { sessionFile: record.sessionFile } : {}), busy: false, operation: "recovery_hold", }); } this.log( `Recovered worker ${worker.descriptor.workerId} without replaying uncertain operations: ${uncertain .map((record) => record.operation) .join(", ")}`, ); } private async refreshWorkerSummaries( worker: ResidentWorker, recovery = false, fillGaps = recovery, retried = false, ): Promise<void> { if (this.isWorkerStopping(worker)) { throw new Error("Session worker is stopping"); } if (!worker.client) { throw new Error("Session worker is not connected"); } const pullSource = worker.client; const epochAtStart = worker.rosterEpoch ?? 0; const response = await pullSource.request({ type: "list" }, 5000); // A frame received mid-pull can remove rows this stale pull would resurrect; re-pull once, then skip the fill. if (fillGaps && (worker.rosterEpoch ?? 0) !== epochAtStart && !retried) { return this.refreshWorkerSummaries(worker, recovery, fillGaps, true); } const summaries = sessionSummariesFromResponse(response); const nextSummaries = new Map(summaries.map((summary) => [summary.activeSessionId ?? summary.id, summary])); const root = nextSummaries.get(worker.descriptor.rootActiveSessionId); if (recovery && !root) { throw new Error(`Session worker omitted its root session during recovery`); } worker.summaries = nextSummaries; if (fillGaps) { await this.chainWorkerRosterApply(worker, pullSource, () => { if ((worker.rosterEpoch ?? 0) === epochAtStart) this.syncRosterFromWorkerSummaries(worker); }); } for (const summary of summaries) { const activeSessionId = summary.activeSessionId ?? summary.id; if (summary.streamingMessage?.role === "assistant") { this.streamReconstructor.seed(activeSessionId, summary.streamingMessage); } else if (!summary.isStreaming) { this.streamReconstructor.clear(activeSessionId); } } if (root) { if (recovery) { await this.assertRecoveryAllowed(); } await this.chainWorkerRosterApply(worker, pullSource, () => { if ((worker.rosterEpoch ?? 0) !== epochAtStart) return; worker.descriptor.rootSessionId = root.sessionId; worker.descriptor.sessionFile = root.sessionFile; worker.descriptor.createCommand = durableDaemonCreateCommand({ type: "create", sessionPath: root.sessionFile, noSession: worker.descriptor.createCommand.noSession, }); this.persistWorker(worker); }); } }
private async familyCatalogEntries(): Promise<AgentFamilyCatalogEntry[]> { const rosterRows = [...this.roster().values()]; const entries = rosterRows.map((entry) => this.familyCatalogEntry(sessionSummaryFromRosterEntry(entry))); const knownFiles = new Set( rosterRows.flatMap((entry) => entry.summary.sessionFile ? [canonicalSessionPath(entry.summary.sessionFile)] : [], ), ); const scanned = await this.catalog.list(undefined, this.defaultSessionConfig.sessionDir); for (const info of scanned) { if (knownFiles.has(canonicalSessionPath(info.path))) continue; if ((info.rlmDepth ?? (info.parentSessionPath ? -1 : 0)) !== 0) continue; entries.push(this.familyCatalogEntry(summaryForInactiveSession(info))); } return entries; }
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 assertSupervisorSessionNameAvailable( target: Pick<SessionSummary, "sessionId" | "rlmDepth" | "parentSessionId" | "parentSessionPath">, name: string, ): Promise<void> { assertAgentSessionNameAvailable(await this.familyCatalogEntries(), { name, depth: target.rlmDepth ?? 0, parentSessionId: target.parentSessionId, parentSessionPath: target.parentSessionPath ? canonicalSessionPath(target.parentSessionPath) : undefined, ignoreSessionId: target.sessionId, }); }
private roster(): AgentRoster { this.rosterStore ??= new AgentRoster(canonicalSessionPath, (mutation) => this.onRosterMutation(mutation)); return this.rosterStore; }
private onRosterMutation(mutation: AgentRosterMutation): void { if (mutation.type === "delete") { this.pendingRosterChanged.delete(mutation.agentId); this.pendingRosterRemoved.add(mutation.agentId); } else { this.pendingRosterRemoved.delete(mutation.agentId); this.pendingRosterChanged.add(mutation.agentId); } this.scheduleRosterPush(); }
private scheduleRosterPush(): void { if (this.rosterPushScheduled || this.shuttingDown) return; this.rosterPushScheduled = true; setImmediate(() => { this.rosterPushScheduled = false; this.flushRosterUpdates(); }); }
private flushRosterUpdates(): void { const changed: AgentRosterEntry[] = []; const removed: string[] = []; for (const agentId of this.pendingRosterRemoved) { if (this.publishedRosterIds.delete(agentId)) removed.push(agentId); } for (const agentId of this.pendingRosterChanged) { const entry = this.roster().get(agentId); if (!entry) continue; if (this.isRosterEntryVisibleToClients(entry)) { changed.push(entry); this.publishedRosterIds.add(agentId); } else if (this.publishedRosterIds.delete(agentId)) { removed.push(agentId); } } this.pendingRosterChanged.clear(); this.pendingRosterRemoved.clear(); if (changed.length === 0 && removed.length === 0) return; for (const client of this.clients) { if (client.rosterSubscribed !== true) continue; if (client.backpressured === true) { client.rosterResyncPending = true; continue; } this.write(client, { type: "roster_update", changed, ...(removed.length > 0 ? { removed } : {}), }); } }
private rosterEntriesForClient(): AgentRosterEntry[] { const entries = [...this.roster().values()].filter((entry) => this.isRosterEntryVisibleToClients(entry)); for (const entry of entries) this.publishedRosterIds.add(entry.agentId); return entries; }
private isRosterEntryVisibleToClients(entry: AgentRosterEntry): boolean { const worker = entry.workerId !== undefined ? this.workers.get(entry.workerId) : undefined; return worker === undefined || this.isVisibleWorker(worker); }
private writeRosterEntry( entry: WorkerRosterEntry, worker?: ResidentWorker, statusLabel?: AgentRosterEntry["statusLabel"], ): AgentRosterEntry { const previousDirect = this.roster().get(entry.agentId)?.summary.directAttachedClients ?? 0; const stored = this.roster().write(entry, worker?.descriptor.workerId, statusLabel); // Direct peers attach and detach on the worker socket, so their last detach arrives // here as roster truth instead of through a supervisor-socket close. if (worker !== undefined && previousDirect > 0 && (entry.summary.directAttachedClients ?? 0) === 0) { void this.evictEmptySessionOnLastDetach(entry.summary.activeSessionId ?? entry.summary.id); } return stored; }
private workerOwnedRosterSummaryForPath(canonicalPath: string): SessionSummary | undefined { const entry = this.roster().bySessionFile(canonicalPath); if (!entry || entry.workerId === undefined || !this.workers.has(entry.workerId)) return undefined; return sessionSummaryFromRosterEntry(entry); }
private workerRosterEntries(worker: ResidentWorker): AgentRosterEntry[] { return this.roster().entriesForWorker(worker.descriptor.workerId); }
private async seedRosterLedger(): Promise<void> { try { const roots = new Set<string>(); for (const worker of this.workers.values()) { const root = worker.descriptor.sessionFile ?? worker.descriptor.createCommand.sessionPath; if (root !== undefined) roots.add(canonicalSessionPath(root)); } if (roots.size === 0) return; const edges = await this.rlmSpawnLedger().liveEdges(); const descendsFrom = rosterFamilyDescendsFrom(edges); for (const edge of edges) { if (!descendsFrom(canonicalSessionPath(edge.parent), roots)) continue; const entry = this.rosterEntryForSpawnLedgerEdge(edge); if (this.roster().has(entry.agentId)) continue; if (this.roster().hasSessionFile(canonicalSessionPath(edge.child))) continue; this.roster().write(await this.hydratedSeedEntry(entry)); } } catch (error) { this.log(`Could not seed the agent roster from the spawn ledger: ${String(error)}`); } }
private rosterEntryForSpawnLedgerEdge(edge: RlmLedgerEdge): WorkerRosterEntry { const persistedSessionId = basename(edge.child, ".jsonl"); const summary: WorkerRosterEntry["summary"] = { id: persistedSessionId, lifecycle: "live", activity: "idle", isSessionActive: false, runtimeKind: "subagent", rlmDepth: edge.depth, sessionId: persistedSessionId, sessionFile: edge.child, sessionName: edge.name, cwd: dirname(edge.child), isStreaming: false, isCompacting: false, attachedClients: 0, messageCount: 0, parentSessionPath: edge.parent, rlmChildId: edge.childId, }; return { agentId: rosterAgentIdForSummary(summary), summary }; }
private consumeWorkerRosterDelta(worker: ResidentWorker, payload: Buffer, source?: DaemonWorkerClient): void { let delta: Extract<DaemonWorkerRosterOutbound, { type: "roster_delta" }>; try { delta = JSON.parse(payload.toString("utf8")) as Extract<DaemonWorkerRosterOutbound, { type: "roster_delta" }>; } catch { return; } if (delta.type !== "roster_delta" || !Array.isArray(delta.entries)) return; worker.rosterEpoch = (worker.rosterEpoch ?? 0) + 1; const applySource = source ?? worker.client ?? worker.pendingClient; if (!this.isWorkerRosterApplyCurrent(worker, applySource)) return; if (delta.snapshot !== true && worker.rosterApplyChain === undefined) { this.applyWorkerRosterDelta(worker, delta); return; } this.chainWorkerRosterApply(worker, applySource, () => delta.snapshot === true ? this.applyWorkerRosterSnapshot(worker, delta, applySource) : this.applyWorkerRosterDelta(worker, delta), ); }
private chainWorkerRosterApply( worker: ResidentWorker, source: DaemonWorkerClient | undefined, apply: () => void | Promise<void>, ): Promise<void> { const chained = (worker.rosterApplyChain ?? Promise.resolve()) .then(() => { if (!this.isWorkerRosterApplyCurrent(worker, source)) return; return apply(); }) .catch((error: unknown) => { this.log(`could not apply a roster frame: ${String(error)}`); this.scheduleRosterRepairPull(worker); }); worker.rosterApplyChain = chained; void chained.finally(() => { if (worker.rosterApplyChain === chained) worker.rosterApplyChain = undefined; }); return chained; }
// An apply is valid only while its own source connection is current: dead connections' parked applies abort. private isWorkerRosterApplyCurrent(worker: ResidentWorker, source: DaemonWorkerClient | undefined): boolean { return ( this.workers.get(worker.descriptor.workerId) === worker && source !== undefined && (source === worker.client || source === worker.pendingClient) ); }
private scheduleRosterRepairPull(worker: ResidentWorker): void { if (worker.rosterRepairPull || !this.isWorkerRosterApplyCurrent(worker, worker.client)) return; // The marker stays set while the repair's own fill applies, so a failing repair never respawns itself. worker.rosterRepairPull = this.refreshWorkerSummaries(worker, false, true) .catch((error: unknown) => this.log(`Roster repair pull failed for worker ${worker.descriptor.workerId}: ${String(error)}`), ) .finally(() => { worker.rosterRepairPull = undefined; }); }
private applyWorkerRosterDelta( worker: ResidentWorker, delta: Extract<DaemonWorkerRosterOutbound, { type: "roster_delta" }>, ): void { for (const entry of delta.entries) { this.writeRosterEntry(entry, worker); this.syncRootDescriptorFromRosterEntry(worker, entry); } for (const agentId of delta.removedAgentIds ?? []) { this.roster().delete(agentId); } }
private async applyWorkerRosterSnapshot( worker: ResidentWorker, delta: Extract<DaemonWorkerRosterOutbound, { type: "roster_delta" }>, source?: DaemonWorkerClient, ): Promise<void> { let edgesFailed = false; const edges = await this.rlmSpawnLedger() .liveEdges() .catch((error: unknown) => { this.log(`Could not read the spawn ledger during a snapshot apply: ${String(error)}`); edgesFailed = true; return [] as RlmLedgerEdge[]; }); const applySource = source ?? worker.client ?? worker.pendingClient; if (!this.isWorkerRosterApplyCurrent(worker, applySource)) return; const sent = new Set(delta.entries.map((entry) => entry.agentId)); const removed = new Set(delta.removedAgentIds ?? []); const unclaimed = new Map<string, AgentRosterEntry>(); if (!edgesFailed) { for (const entry of this.workerRosterEntries(worker)) { if (sent.has(entry.agentId)) continue; unclaimed.set(entry.agentId, entry); } } // Only this worker's family reseeds: anything wider can resurrect a client-owned worker's dropped children. const workerRoot = worker.descriptor.sessionFile ?? worker.descriptor.createCommand.sessionPath; const rootPaths = new Set(workerRoot !== undefined ? [canonicalSessionPath(workerRoot)] : []); const descendsFrom = rosterFamilyDescendsFrom(edges); const familyEdges = edges.filter((edge) => descendsFrom(canonicalSessionPath(edge.parent), rootPaths)); // "Unclaimed" rows survive: the sweep deletes them but the restore branch rewrites them. const rowSurvivesWithoutReseed = (entry: WorkerRosterEntry, childPath: string): boolean => { if (unclaimed.has(entry.agentId)) return true; if (sent.has(entry.agentId) && !removed.has(entry.agentId)) return true; const survives = (row: AgentRosterEntry | undefined): boolean => row !== undefined && !unclaimed.has(row.agentId) && !removed.has(row.agentId); return survives(this.roster().get(entry.agentId)) || survives(this.roster().bySessionFile(childPath)); }; const seededEntries = new Map<string, WorkerRosterEntry>(); for (const edge of familyEdges) { const entry = this.rosterEntryForSpawnLedgerEdge(edge); if (rowSurvivesWithoutReseed(entry, canonicalSessionPath(edge.child))) continue; seededEntries.set(entry.agentId, await this.hydratedSeedEntry(entry)); if (!this.isWorkerRosterApplyCurrent(worker, applySource)) return; }
// Unreadable edges skip the absentee sweep: it cannot tell registry children from stale rows. for (const entry of unclaimed.values()) this.roster().delete(entry.agentId); for (const entry of delta.entries) { this.writeRosterEntry(entry, worker); this.syncRootDescriptorFromRosterEntry(worker, entry); } for (const agentId of removed) this.roster().delete(agentId); if (edgesFailed) { this.scheduleRosterRepairPull(worker); return; } for (const edge of familyEdges) { const entry = this.rosterEntryForSpawnLedgerEdge(edge); if (this.roster().has(entry.agentId)) continue; if (this.roster().hasSessionFile(canonicalSessionPath(edge.child))) continue; const previous = unclaimed.get(entry.agentId); if (previous) { const { status, statusLabel, lastHeardFromAt, workerId, ...rest } = previous; this.writeRosterEntry(rest, worker); continue; } this.roster().write(seededEntries.get(entry.agentId) ?? { ...entry, seededCwd: true }); } }
private syncRootDescriptorFromRosterEntry(worker: ResidentWorker, entry: WorkerRosterEntry): void { const summary = entry.summary; if (summary.activeSessionId !== worker.descriptor.rootActiveSessionId) return; if ( worker.descriptor.rootSessionId === summary.sessionId && worker.descriptor.sessionFile === summary.sessionFile ) { return; } worker.descriptor.rootSessionId = summary.sessionId; worker.descriptor.sessionFile = summary.sessionFile; worker.descriptor.createCommand = durableDaemonCreateCommand({ type: "create", sessionPath: summary.sessionFile, noSession: worker.descriptor.createCommand.noSession, }); this.persistWorker(worker); }
// Behind the pull-epoch guard the pull is never staler than the row it replaces; never steal another worker's claim. private syncRosterFromWorkerSummaries(worker: ResidentWorker): void { for (const summary of worker.summaries.values()) { const entry = workerRosterEntryFromSummary(summary); const existing = this.roster().get(entry.agentId); if (existing?.workerId !== undefined && existing.workerId !== worker.descriptor.workerId) continue; this.writeRosterEntry(entry, worker); } }
private markWorkerRosterEntries(worker: ResidentWorker, statusLabel: "recovering" | "failed" | undefined): void { for (const entry of this.workerRosterEntries(worker)) { if (!entry.queuedChild && entry.summary.activeSessionId === undefined) continue; this.roster().amend(entry.agentId, { statusLabel }); } }
private flipWorkerRosterEntriesInactive(worker: ResidentWorker): void { // Client-owned workers are ephemeral and private: their rows die with the registration. const ephemeral = worker.descriptor.ownerClientId !== undefined; for (const entry of this.workerRosterEntries(worker)) { if (ephemeral || entry.queuedChild) { this.roster().delete(entry.agentId); continue; } // Registration marks survive eviction: passive rows still have schedules behind them. this.writeRosterEntry( passivatedWorkerRosterEntry(entry, { hasRegisteredHeartbeat: entry.summary.hasRegisteredHeartbeat === true, hasRegisteredCronJob: entry.summary.hasRegisteredCronJob === true, }), ); } }
private sweepRosterStaleness(now = Date.now()): void { for (const worker of this.workers.values()) { if (worker.client === undefined || worker.lastFrameAt === undefined) { continue; } if (now - worker.lastFrameAt > ROSTER_STALE_AFTER_MS) { const lastHeardFromAt = new Date(worker.lastFrameAt).toISOString(); for (const entry of this.workerRosterEntries(worker)) { // write() rebuilds rows without the mark; the sweep owns it and restamps only those. if (entry.lastHeardFromAt !== lastHeardFromAt) this.roster().amend(entry.agentId, { lastHeardFromAt }); } worker.rosterStale = true; } else if (worker.rosterStale) { this.clearRosterStaleness(worker); } } }
private clearRosterStaleness(worker: ResidentWorker): void { if (!worker.rosterStale) return; worker.rosterStale = false; for (const entry of this.workerRosterEntries(worker)) { this.roster().amend(entry.agentId, { lastHeardFromAt: undefined }); } }
/** * Supervisor-side view of the spawn ledger for this supervisor's sessions * dir. Workers hold their own instances over the same file; every read * re-reads the file, so cross-process freshness is per-operation. */ private rlmSpawnLedger(): RlmSpawnLedger { const agentDir = this.defaultSessionConfig.agentDir; if (!agentDir) { throw new Error("Daemon supervisor config is missing agentDir"); } this.rlmSpawnLedgerInstance ??= new RlmSpawnLedger( agentDir, this.defaultSessionConfig.sessionDir ?? getSessionsDir(agentDir), 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 { const agentDir = this.defaultSessionConfig.agentDir; const defaultDir = this.defaultSessionConfig.sessionDir ?? (agentDir ? getSessionsDir(agentDir) : undefined); if (sessionDir === undefined || (defaultDir !== undefined && resolve(sessionDir) === resolve(defaultDir))) { return this.rlmSpawnLedger(); } if (!agentDir) { throw new Error("Daemon supervisor config is missing agentDir"); } return new RlmSpawnLedger(agentDir, sessionDir, createRlmLedgerRegistrySeedSource(), (message) => this.log(message), ); }
/** * Ledger-backed same-parent rows for name reservation and admission. Rows * carry ledger topology plus best-effort display fields; consumers here * only need id/path/name/depth/parent. */ private rlmLedgerSiblings(sessionPath: string): Promise<SessionInfo[]> { return this.rlmSpawnLedger().siblings(sessionPath); }
private async savedSessionNameReservationInput( sessionPath: string, name: string, ): Promise<{ name: string; depth: number; parentSessionId?: string; parentSessionPath?: string }> { const targetPath = canonicalSessionPath(sessionPath); const active = this.workerOwnedRosterSummaryForPath(targetPath); if (active) return this.summaryNameReservationInput(active, name); const siblings = await this.rlmLedgerSiblings(sessionPath); const saved = siblings.find((info) => canonicalSessionPath(info.path) === targetPath); if (!saved) throw new Error(`Session not found: ${sessionPath}`); return { name, depth: saved.rlmDepth ?? siblings.find((sibling) => sibling.rlmDepth !== undefined)?.rlmDepth ?? 0, parentSessionPath: saved.parentSessionPath, }; }
private summaryNameReservationInput( target: Pick<SessionSummary, "rlmDepth" | "parentSessionId" | "parentSessionPath">, name: string, ): { name: string; depth: number; parentSessionId?: string; parentSessionPath?: string } { const depth = target.rlmDepth ?? (target.parentSessionPath ? 1 : 0); return { name, depth, ...(depth > 0 && target.parentSessionId ? { parentSessionId: target.parentSessionId } : {}), ...(depth > 0 && target.parentSessionPath ? { parentSessionPath: target.parentSessionPath } : {}), }; }
private async assertSupervisorSavedSessionNameAvailable(sessionPath: string, name: string): Promise<void> { const targetPath = canonicalSessionPath(sessionPath); const active = this.workerOwnedRosterSummaryForPath(targetPath); if (active) return this.assertSupervisorSessionNameAvailable(active, name); const siblings = await this.rlmLedgerSiblings(sessionPath); const saved = siblings.find((info) => canonicalSessionPath(info.path) === targetPath); if (!saved) throw new Error(`Session not found: ${sessionPath}`); if (saved.parentSessionPath && (saved.rlmDepth ?? 0) > 0) { this.assertSavedSiblingNameAvailable(siblings, saved, name); } else { await this.assertSupervisorSessionNameAvailable(summaryForInactiveSession(saved), name); } }
private assertSavedSiblingNameAvailable(siblings: SessionInfo[], target: SessionInfo, name: string): void { const setDepth = target.rlmDepth ?? siblings.find((sibling) => sibling.rlmDepth !== undefined)?.rlmDepth ?? 0; assertAgentSessionNameAvailable( siblings.map((info) => { const summary = summaryForInactiveSession(info); const ledgerRow = this.roster().bySessionFile(canonicalSessionPath(info.path)); return { id: summary.sessionId, ...(summary.sessionName ? { name: summary.sessionName } : {}), depth: setDepth, status: ledgerRow?.status ?? classifySessionRosterStatus(summary), ...(summary.parentSessionPath ? { parentSessionPath: canonicalSessionPath(summary.parentSessionPath) } : {}), }; }), { name, depth: setDepth, parentSessionPath: target.parentSessionPath ? canonicalSessionPath(target.parentSessionPath) : undefined, ignoreSessionId: target.id, }, ); }
private isVisibleWorker(worker: ResidentWorker): boolean { return worker.descriptor.ownerClientId === undefined; }
/** A worker with a durable or in-memory stop intent is stopping, never live. */ private isWorkerStopping(worker: ResidentWorker): boolean { return worker.intentionalStop || worker.descriptor.stopRequestedAt !== undefined; }
/** Live workers are visible to all clients and not stopping. */ private isLiveWorker(worker: ResidentWorker): boolean { return this.isVisibleWorker(worker) && !this.isWorkerStopping(worker); }
private isWorkerBusy(worker: ResidentWorker): boolean { if ( [...(worker.summaries?.values() ?? [])].some( (s) => isSessionSummaryBusy(s) || s.activity === "working" || s.isSessionActive === true, ) ) { return true; } return this.workerRosterEntries(worker).some( (e) => isSessionSummaryBusy(e.summary) || e.summary.activity === "working" || e.summary.isSessionActive === true, ); }
/** * The lifecycle reported to clients. A stop intent always wins, and a worker * whose process connection is gone is never reported as "ready". */ private effectiveWorkerState(worker: ResidentWorker): DaemonWorkerLifecycle { if (this.isWorkerStopping(worker)) { return "stopping"; } if (worker.descriptor.lifecycle === "ready" && worker.client === undefined) { return "recovering"; } return worker.descriptor.lifecycle; }
private requireAvailableWorkerClient(worker: ResidentWorker, allowStopping = false): DaemonWorkerClient { if ( !worker.client || worker.descriptor.lifecycle !== "ready" || (!allowStopping && this.isWorkerStopping(worker)) ) { const state = this.effectiveWorkerState(worker); throw new DaemonSessionStartingError(worker.descriptor.rootActiveSessionId, `Session worker is ${state}`); } return worker.client; }
private async issuePeerTransport( worker: ResidentWorker, summary: SessionSummary, ): Promise<DaemonPeerTransportTicket> { await this.assertCurrentOwnership(); if (!worker.peerTransportCapable) { throw new Error("Session worker does not support direct peer transport"); } await this.refreshWorkerSummaries(worker); const workerClient = this.requireAvailableWorkerClient(worker); const activeSessionId = summary.activeSessionId ?? summary.id; const currentSummary = this.findSummaryInWorker(worker, activeSessionId); if (!currentSummary) { throw new Error("Direct transport target changed during admission"); } const workerInstanceId = worker.descriptor.workerInstanceId; const workerProcessStartId = worker.descriptor.processStartId; if (!workerInstanceId || !workerProcessStartId) { throw new Error("Direct transport requires an exact worker process identity"); } if (this.processIdentity(worker.descriptor.pid, workerProcessStartId) !== "current") { throw new Error("Direct transport worker process identity is not current"); } let socketIdentity: DaemonSocketIdentity | undefined; try { socketIdentity = getDaemonSocketIdentity(worker.descriptor.socketPath); } catch { socketIdentity = undefined; } if (!socketIdentity) { throw new Error("Direct transport requires an exact worker socket identity"); } const grantId = randomUUID(); const token = randomBytes(32).toString("base64url"); const expiresAt = new Date(Date.now() + PEER_TRANSPORT_GRANT_TTL_MS).toISOString(); const resolvedActiveSessionId = currentSummary.activeSessionId ?? currentSummary.id; const registration = await workerClient.requestWorker( { type: "worker_register_peer_transport", grant: { grantId, token, expiresAt, purpose: "session_client", workerInstanceId, activeSessionId: resolvedActiveSessionId, issuerGeneration: this.generation, }, }, 3000, ); if (!registration.success) { throw deserializeDaemonError(registration); } return { purpose: "session_client", socketPath: worker.descriptor.socketPath, socketIdentity, workerInstanceId, activeSessionId: resolvedActiveSessionId, grantId, token, expiresAt, }; }
private familyCatalogEntry(summary: SessionSummary): AgentFamilyCatalogEntry { const depth = summary.rlmDepth ?? (summary.parentSessionPath ? 1 : 0); return { id: summary.sessionId, ...(summary.sessionName ? { name: summary.sessionName } : {}), depth, status: summary.rosterStatus ?? classifySessionRosterStatus(summary), ...(depth > 0 && summary.parentSessionId ? { parentSessionId: summary.parentSessionId } : {}), ...(depth > 0 && summary.parentSessionPath ? { parentSessionPath: canonicalSessionPath(summary.parentSessionPath) } : {}), ...(summary.sessionFile ? { sessionPath: canonicalSessionPath(summary.sessionFile) } : {}), }; }
private agentPeerSummary(summary: SessionSummary): AgentSessionMessageAgentSummary { return { activeSessionId: summary.activeSessionId ?? summary.id, sessionId: summary.sessionId, ...(summary.sessionName ? { sessionName: summary.sessionName } : {}), runtimeKind: summary.runtimeKind ?? "top-level", cwd: summary.cwd, isStreaming: summary.isStreaming, unfinishedActionCount: summary.unfinishedActionCount ?? (summary.sessionActions.active ? 1 + summary.sessionActions.queuedCount : summary.sessionActions.queuedCount), ...(summary.parentActiveSessionId ? { parentActiveSessionId: summary.parentActiveSessionId } : {}), ...(summary.parentSessionId ? { parentSessionId: summary.parentSessionId } : {}), ...(summary.parentSessionPath ? { parentSessionPath: summary.parentSessionPath } : {}), ...(summary.sessionFile ? { sessionPath: summary.sessionFile } : {}), ...(summary.rlmDepth !== undefined ? { rlmDepth: summary.rlmDepth } : {}), status: summary.rosterStatus ?? classifySessionRosterStatus(summary), ...(summary.rlmChildId ? { rlmChildId: summary.rlmChildId } : {}), }; }
private attachedClientCount(summary: SessionSummary, activeSessionId: string): number { return ( (summary.directAttachedClients ?? 0) + [...this.clients].filter((client) => client.attachedActiveSessionIds.has(activeSessionId)).length ); }
private publicSummary(worker: ResidentWorker, summary: SessionSummary): SessionSummary { const activeSessionId = summary.activeSessionId ?? summary.id; return { ...summary, attachedClients: this.attachedClientCount(summary, activeSessionId), workerState: this.effectiveWorkerState(worker), workerPid: worker.descriptor.pid, }; }
private async findWorker( selector: string, includeWorker?: (worker: ResidentWorker) => boolean, ): Promise<WorkerMatch> { let matches = this.matchWorkers(selector, includeWorker); if (matches.length === 0) { await Promise.all( [...this.workers.values()].map((worker) => this.refreshWorkerSummaries(worker, false, true).catch(() => undefined), ), ); matches = this.matchWorkers(selector, includeWorker); } if (matches.length === 1) { return matches[0]!; } if (matches.length > 1) { throw new Error(`Ambiguous active session "${selector}"`); } // Descriptors are the durable half of addressability: an unhydrated root is recovering, not unknown; // failed workers stay unknown so clients take the create fallback, which reclaims or retries them. const recoveringRoots = (matchesSelector: (worker: ResidentWorker) => boolean) => [...this.workers.values()].filter( (worker) => matchesSelector(worker) && (!includeWorker || includeWorker(worker)) && worker.descriptor.lifecycle !== "failed" && this.isWorkerRecoveryCandidate(worker), ); // matchWorkers' addressing rule: exact ids first, unambiguous hex suffixes second. const exactRecovering = recoveringRoots( (worker) => worker.descriptor.rootActiveSessionId === selector || worker.descriptor.rootSessionId === selector, ); const recoveringMatches = exactRecovering.length > 0 ? exactRecovering : recoveringRoots( (worker) => matchesSessionIdSuffix(worker.descriptor.rootActiveSessionId, selector) || matchesSessionIdSuffix(worker.descriptor.rootSessionId ?? "", selector), ); if (recoveringMatches.length === 1) { throw new DaemonSessionRecoveringError(recoveringMatches[0]!.descriptor.rootActiveSessionId); } throw new Error(`Unknown active session: ${selector}`); }
private findWorkerForClient(client: DaemonSocketClient, selector: string): Promise<WorkerMatch> { return this.findWorker(selector, (worker) => this.isWorkerAccessibleToClient(client, worker)); }
private isWorkerAccessibleToClient(client: DaemonSocketClient, worker: ResidentWorker): boolean { return ( worker.descriptor.ownerClientId === undefined || worker.descriptor.ownerClientId === this.protocolClientId(client) ); }
private assertWorkerAccessibleToClient(client: DaemonSocketClient, worker: ResidentWorker, selector: string): void { if (!this.isWorkerAccessibleToClient(client, worker)) { throw new Error(`Unknown active session: ${selector}`); } }
private matchWorkers(selector: string, includeWorker?: (worker: ResidentWorker) => boolean): WorkerMatch[] { const exact: WorkerMatch[] = []; const suffix: WorkerMatch[] = []; for (const entry of this.roster().values()) { if (entry.queuedChild) continue; const worker = entry.workerId !== undefined ? this.workers.get(entry.workerId) : undefined; if (!worker || (includeWorker && !includeWorker(worker))) { continue; } const summary = sessionSummaryFromRosterEntry(entry); const activeSessionId = summary.activeSessionId ?? summary.id; const match = { worker, summary }; if (activeSessionId === selector || summary.sessionId === selector || summary.sessionName === selector) { exact.push(match); } else if ( matchesSessionIdSuffix(activeSessionId, selector) || matchesSessionIdSuffix(summary.sessionId, selector) ) { suffix.push(match); } } return exact.length > 0 ? exact : suffix; }
private findSummaryInWorker(worker: ResidentWorker, selector: string): SessionSummary | undefined { const pathSelector = looksLikeSessionPath(selector) ? canonicalSessionPath(selector) : undefined; const summaries = this.workerRosterEntries(worker) .filter((entry) => !entry.queuedChild) .map(sessionSummaryFromRosterEntry); const exact = summaries.find((summary) => { const activeSessionId = summary.activeSessionId ?? summary.id; return ( activeSessionId === selector || summary.sessionId === selector || summary.sessionName === selector || (pathSelector !== undefined && summary.sessionFile !== undefined && canonicalSessionPath(summary.sessionFile) === pathSelector) ); }); if (exact) return exact; return summaries.find((summary) => { const activeSessionId = summary.activeSessionId ?? summary.id; return ( matchesSessionIdSuffix(activeSessionId, selector) || matchesSessionIdSuffix(summary.sessionId, selector) ); }); }
private findWorkerBySessionFile(sessionFile: string, exclude?: ResidentWorker): ResidentWorker | undefined { const target = canonicalSessionPath(sessionFile); const targetEntry = this.roster().bySessionFile(target); const matches = new Set<ResidentWorker>(); for (const worker of this.workers.values()) { if (worker === exclude) continue; // The roster is the one live ownership source; the stale pull cache must not resurrect a match. const summaryMatches = targetEntry?.workerId === worker.descriptor.workerId; const descriptorPath = worker.descriptor.sessionFile ? canonicalSessionPath(worker.descriptor.sessionFile) : undefined; const configuredPath = worker.descriptor.createCommand.sessionPath ? canonicalSessionPath(worker.descriptor.createCommand.sessionPath) : undefined; if (!summaryMatches && descriptorPath !== target && configuredPath !== target) continue; if (descriptorPath && configuredPath && descriptorPath !== configuredPath) { throw new Error(`Conflicting resident session paths for worker ${worker.descriptor.workerId}`); } matches.add(worker); } if (matches.size > 1) { throw new Error(`Ambiguous resident session path "${sessionFile}"`); } return matches.values().next().value; }
private async forwardToWorker( worker: ResidentWorker, command: DaemonCommand, timeoutMs = WORKER_REQUEST_TIMEOUT_MS, ): Promise<DaemonResponse> { // Every forwarded command is a touch: a cached failed roster row must not outrank // the descriptor truth that the worker is recoverable (--attach-agent's get_state preflight lands here). if (this.canRetryFailedWorker(worker)) { await this.retryWorkerRecovery(worker); } else if (worker.recovery) { // Join a concurrent touch's in-flight recovery instead of throwing mid-ladder. await worker.recovery; } const client = this.requireAvailableWorkerClient(worker, command.type === "kill"); const response = await client.request(withoutCommandId(command), timeoutMs); if (command.type === "get_state" && response.success && isSessionSummary(response.data)) { return { ...response, id: command.id, data: this.publicSummary(worker, response.data) }; } if (command.type === "rename" && response.success && isSessionSummary(response.data)) { this.writeRosterEntry(workerRosterEntryFromSummary(response.data), worker); return { ...response, id: command.id, data: this.publicSummary(worker, response.data) }; } return responseWithId(response, command.id); }
private async attachClient( client: DaemonSocketClient, command: Extract<DaemonCommand, { type: "attach" }>, ): Promise<WorkerAttachData> { const descriptorWorker = [...this.workers.values()].find( (worker) => worker.descriptor.rootActiveSessionId === command.activeSessionId || worker.descriptor.rootSessionId === command.activeSessionId, ); if (descriptorWorker) { // The descriptor lookup is universal; owner-only attach payload stays in the owned branch. if (descriptorWorker.descriptor.ownerClientId !== undefined) { if (descriptorWorker.descriptor.ownerClientId !== this.protocolClientId(client)) { throw new Error(`Unknown active session: ${command.activeSessionId}`); } this.assertTelemetryAttachAllowed(descriptorWorker, command.telemetryDisabled); descriptorWorker.launchEnv = command.launchEnv ?? descriptorWorker.launchEnv; if (!descriptorWorker.client || descriptorWorker.descriptor.lifecycle !== "ready") { if (command.recoveryConfig) { descriptorWorker.transientCreateCommand = { ...descriptorWorker.descriptor.createCommand, config: { ...command.recoveryConfig, ...(descriptorWorker.descriptor.telemetryDisabled === true ? { telemetryDisabled: true } : {}), }, env: command.env, launchEnv: command.launchEnv, lifecycle: "client_owned", }; } if (!descriptorWorker.launchEnv) { throw new Error("Client-owned session recovery requires the owning client environment"); } await this.retryWorkerRecovery(descriptorWorker); } } else if (this.canRetryFailedWorker(descriptorWorker)) { await this.retryWorkerRecovery(descriptorWorker); } } const match = await this.findWorkerForClient(client, command.activeSessionId); this.assertTelemetryAttachAllowed(match.worker, command.telemetryDisabled); if (match.worker !== descriptorWorker && this.canRetryFailedWorker(match.worker)) { // Child-session attaches land here without a descriptor match; one touch runs at most one ladder. await this.retryWorkerRecovery(match.worker); } this.requireAvailableWorkerClient(match.worker); const activeSessionId = match.summary.activeSessionId ?? match.summary.id; const duplicateValidation = this.currentSnapshotGeneration(match.worker, activeSessionId)?.validation; if (duplicateValidation) { await duplicateValidation.promise; } if (command.clientId) { client.id = command.clientId; } client.capabilities = normalizeCapabilities(command.capabilities, command.supportsExtensionUi); client.supportsExtensionUi = client.capabilities.has("extension_ui");
let result = match.worker.snapshotCache.get(activeSessionId); if ( result && !client.capabilities.has("chunked_snapshot") && result.snapshot.messages.length < result.snapshot.summary.messageCount ) { result = undefined; } if (!result) { const snapshotLoadKey = `${activeSessionId}:${client.capabilities.has("chunked_snapshot") ? "chunked" : "full"}`; let retryInvalidatedLoad = true; while (!result) { let loading = match.worker.snapshotLoads.get(snapshotLoadKey); if (!loading) { const observedSnapshotId = match.worker.transcriptCaches.get(activeSessionId)?.snapshotId ?? match.worker.snapshotCache.get(activeSessionId)?.snapshotStream?.id; loading = (async () => { const workerClient = this.requireAvailableWorkerClient(match.worker); const response = await workerClient.request({ type: "attach", activeSessionId, capabilities: client.capabilities.has("chunked_snapshot") ? ["attach_snapshot", "event_sequence", "slim_attach", "chunked_snapshot"] : ["attach_snapshot", "event_sequence", "slim_attach"], supportsExtensionUi: false, env: command.env ?? collectDaemonClientEnv(), }); const loaded = attachResultFromResponse(response); if (match.worker.snapshotLoads.get(snapshotLoadKey) !== loading) { throw new SnapshotLoadInvalidatedError("Session snapshot changed during attach"); } return this.cacheLoadedSnapshot(match.worker, activeSessionId, loaded, observedSnapshotId); })(); match.worker.snapshotLoads.set(snapshotLoadKey, loading); void loading.then( async (loaded) => { try { const snapshotId = loaded.snapshotStream?.id; const transcript = snapshotId ? this.snapshotGeneration(match.worker, loaded.activeSessionId, snapshotId)?.transcript : undefined; if (transcript && !transcript.complete) { let chunkIndex = 0; while (await transcript.waitForChunk(chunkIndex)) { chunkIndex++; } } } catch { // Failed transfers must allow a fresh snapshot request. } finally { if (match.worker.snapshotLoads.get(snapshotLoadKey) === loading) { match.worker.snapshotLoads.delete(snapshotLoadKey); } } }, () => { if (match.worker.snapshotLoads.get(snapshotLoadKey) === loading) { match.worker.snapshotLoads.delete(snapshotLoadKey); } }, ); } try { result = await loading; } catch (error) { if (!(error instanceof SnapshotLoadInvalidatedError)) { throw error; } if (!retryInvalidatedLoad) { throw error; } retryInvalidatedLoad = false; } } } this.requireAvailableWorkerClient(match.worker); const wasAttached = client.attachedActiveSessionIds.has(activeSessionId); let transcript: SnapshotTranscriptCache | undefined; if (client.capabilities.has("chunked_snapshot")) { while (true) { const validation = this.currentSnapshotGeneration(match.worker, activeSessionId)?.validation; if (validation) { await validation.promise; continue; } result = match.worker.snapshotCache.get(activeSessionId) ?? result; transcript = this.getOrCreateTranscriptCache(match.worker, result); break; } } const releaseTranscript = transcript?.retain(); client.attachedActiveSessionIds.add(activeSessionId); try { const publicSummary = this.publicSummary(match.worker, result.snapshot.summary); if (publicSummary.streamingMessage?.role === "assistant") { this.streamReconstructor.seed(activeSessionId, publicSummary.streamingMessage); } else { for (let index = result.snapshot.messages.length - 1; index >= 0; index--) { const latestMessage = result.snapshot.messages[index]; if (latestMessage?.role === "assistant") { this.streamReconstructor.seed(activeSessionId, latestMessage); break; } } } const publicResult: DaemonAttachResult = { ...result, state: result.state ? publicSummary : undefined, snapshot: { ...result.snapshot, summary: publicSummary }, client: { id: client.id, capabilities: [...client.capabilities] }, }; if (publicResult.state && publicResult.messages) { this.write(client, { type: "session_attached", activeSessionId, state: publicResult.state, messages: publicResult.messages, snapshot: publicResult.snapshot, replay: publicResult.replay, lastEventSequence: publicResult.lastEventSequence, }); } void this.syncWorkerExtensionUi(activeSessionId); const detachingSessions = this.detachingInputPauseSessions?.get(client); detachingSessions?.delete(command.activeSessionId); detachingSessions?.delete(activeSessionId); return { result: publicResult, worker: match.worker, transcript, releaseTranscript }; } catch (error) { releaseTranscript?.(); if (!wasAttached) { client.attachedActiveSessionIds.delete(activeSessionId); } throw error; } }
private assertTelemetryAttachAllowed(worker: ResidentWorker, telemetryDisabled: true | undefined): void { if (telemetryDisabled && worker.descriptor.telemetryDisabled !== true) { throw new Error( "Cannot attach to this active agent while telemetry is disabled for the current invocation. Stop the agent and retry so it can restart without telemetry.", ); } }
private cacheLoadedSnapshot( worker: ResidentWorker, activeSessionId: string, loaded: DaemonAttachResult, observedSnapshotId: string | undefined, ): DaemonAttachResult { const currentTranscript = worker.transcriptCaches.get(activeSessionId); const currentGeneration = this.currentSnapshotGeneration(worker, activeSessionId); const currentResult = currentGeneration?.result ?? worker.snapshotCache.get(activeSessionId); const currentSnapshotId = currentTranscript?.snapshotId ?? currentResult?.snapshotStream?.id; const loadedSnapshotId = loaded.snapshotStream?.id; if (!loaded.snapshotStream) { if (currentSnapshotId && currentSnapshotId !== observedSnapshotId) { return loaded; } worker.snapshotCache.set(activeSessionId, loaded); return loaded; } if ( currentSnapshotId && currentSnapshotId !== loadedSnapshotId && (currentSnapshotId !== observedSnapshotId || (currentResult?.lastEventSequence ?? -1) > loaded.lastEventSequence) ) { return currentResult ?? loaded; } let transcript = currentTranscript; if (transcript && transcript.snapshotId !== loaded.snapshotStream.id) { this.retireWorkerSnapshotCache(worker, activeSessionId, transcript); transcript = undefined; } const generations = this.snapshotGenerationsFor(worker, activeSessionId); let generation = generations.get(loaded.snapshotStream.id); if (!transcript) { transcript = generation?.transcript; } if (!transcript) { transcript = new SnapshotTranscriptCache({ activeSessionId, snapshotId: loaded.snapshotStream.id, cacheRoot: this.snapshotCacheRoot, targetChunkBytes: loaded.snapshotStream.targetChunkBytes, }); } if (!generation) { generation = { transcript, result: loaded, incoming: false, retired: false, }; generations.set(loaded.snapshotStream.id, generation); } else { generation.result = loaded; generation.retired = false; } worker.transcriptCaches.set(activeSessionId, transcript); worker.snapshotCache.set(activeSessionId, loaded); return loaded; }
private getOrCreateTranscriptCache(worker: ResidentWorker, result: DaemonAttachResult): SnapshotTranscriptCache { const activeSessionId = result.activeSessionId; const existing = worker.transcriptCaches.get(activeSessionId); if (existing && (!result.snapshotStream || existing.snapshotId === result.snapshotStream.id)) { return existing; } if (result.snapshot.messages.length < result.snapshot.summary.messageCount) { throw new Error("Session snapshot generation changed before its transcript could be selected"); } if (existing) { this.retireWorkerSnapshotCache(worker, activeSessionId, existing); } const revision = createHash("sha256") .update( `${activeSessionId}:${result.snapshot.summary.sessionId}:${result.lastEventSequence}:${result.snapshot.messages.length}`, ) .digest("hex") .slice(0, 16); const transcript = new SnapshotTranscriptCache({ activeSessionId, snapshotId: `${activeSessionId}-${revision}`, messages: result.snapshot.messages, cacheRoot: this.snapshotCacheRoot, targetChunkBytes: SNAPSHOT_TARGET_CHUNK_BYTES, }); worker.transcriptCaches.set(activeSessionId, transcript); const cachedResult = { ...result, messages: result.messages ? [] : undefined, snapshot: { ...result.snapshot, messages: [] }, }; worker.snapshotCache.set(activeSessionId, cachedResult); this.snapshotGenerationsFor(worker, activeSessionId).set(transcript.snapshotId, { transcript, result: cachedResult, incoming: false, retired: false, }); return transcript; }
private createStreamedAttachResult( result: DaemonAttachResult, transcript: SnapshotTranscriptCache, ): DaemonAttachResult { return { ...result, messages: result.messages ? [] : undefined, snapshot: { ...result.snapshot, messages: [] }, snapshotStream: { id: transcript.snapshotId, messageCount: result.snapshot.summary.messageCount, targetChunkBytes: transcript.targetChunkBytes, }, }; }
private async streamSnapshot( client: DaemonSocketClient, worker: ResidentWorker, result: DaemonAttachResult, transcript: SnapshotTranscriptCache, purpose: "attach" | "replacement" | "resync" = "attach", retainedTranscriptRelease?: () => void, releaseSnapshotReservation = this.reserveSnapshotStream(client, result.activeSessionId), ): Promise<void> { const stream = result.snapshotStream; const signal = client.snapshotTransferAbortControllers?.get(result.activeSessionId)?.signal; if (!stream || client.socket.destroyed || signal?.aborted) { releaseSnapshotReservation(); retainedTranscriptRelease?.(); this.releaseDeferredSessionPayloads(client, result.activeSessionId, false); return; } const releaseTranscript = retainedTranscriptRelease ?? transcript.retain(); const { messages: _messages, ...snapshotHeader } = result.snapshot; let snapshotDelivered = false; try { if ( !(await this.writeSnapshotRecord( client, { type: "session_snapshot_begin", activeSessionId: result.activeSessionId, snapshotId: stream.id, snapshot: snapshotHeader, messageCount: stream.messageCount, targetChunkBytes: stream.targetChunkBytes, purpose, }, signal, )) ) { return; } let chunkCount = 0; while (true) { let chunk: Buffer | undefined; try { chunk = await transcript.waitForChunk(chunkCount, signal); } catch (error) { if (signal?.aborted) return; const streamError = error instanceof Error ? error : new Error(String(error)); this.failWorkerSnapshotCache(worker, result.activeSessionId, streamError, false, stream.id); throw streamError; } if (!chunk) { break; } if (!(await this.writeSnapshotBuffer(client, chunk, signal))) { return; } chunkCount++; } snapshotDelivered = await this.writeSnapshotRecord( client, { type: "session_snapshot_end", activeSessionId: result.activeSessionId, snapshotId: stream.id, chunkCount, lastEventSequence: result.lastEventSequence, lastEventCursor: result.lastEventCursor, }, signal, ); } catch (error) { if (signal?.aborted) return; const streamError = error instanceof Error ? error : new Error(String(error)); if (!client.socket.destroyed) { try { const delivered = await this.writeSnapshotRecord( client, { type: "session_snapshot_failed", activeSessionId: result.activeSessionId, snapshotId: stream.id, error: streamError.message, }, signal, ); if (signal?.aborted) return; if (!delivered && !client.socket.destroyed) { client.socket.destroy(streamError); } } catch (deliveryError) { client.socket.destroy(deliveryError instanceof Error ? deliveryError : new Error(String(deliveryError))); } } throw streamError; } finally { if ( !snapshotDelivered && client.attachedActiveSessionIds.has(result.activeSessionId) && client.deferredSessionPayloads?.get(result.activeSessionId)?.payloads.length ) { this.queueCatchup(client, result.activeSessionId, "resync"); } releaseSnapshotReservation(); releaseTranscript(); this.releaseDeferredSessionPayloads(client, result.activeSessionId, snapshotDelivered); } }
private reserveSnapshotStream(client: DaemonSocketClient, activeSessionId: string): () => void { if (!client.snapshotActiveSessionIds?.has(activeSessionId)) { client.deferredSessionPayloadsDropped?.delete(activeSessionId); client.snapshotTransferAbortControllers ??= new Map(); client.snapshotTransferAbortControllers.set(activeSessionId, new AbortController()); } 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, ); let released = false; return () => { if (released) { return; } released = true; const streamCount = client.snapshotActiveSessionCounts?.get(activeSessionId) ?? 1; if (streamCount > 1) { client.snapshotActiveSessionCounts?.set(activeSessionId, streamCount - 1); } else { client.snapshotActiveSessionCounts?.delete(activeSessionId); client.snapshotActiveSessionIds?.delete(activeSessionId); client.snapshotTransferAbortControllers?.delete(activeSessionId); } client.snapshotStreaming = (client.snapshotActiveSessionIds?.size ?? 0) > 0; if (!client.snapshotStreaming && client.catchupActiveSessionIds?.size) { void this.catchUpClient(client).catch((error) => this.log(`Failed to catch up client ${client.id}: ${String(error)}`), ); } }; }
private writeSnapshotRecord( client: DaemonSocketClient, message: DaemonOutbound, signal?: AbortSignal, ): Promise<boolean> { return this.writeSnapshotBuffer(client, Buffer.from(serializeJsonLine(message)), signal); }
private async writeSnapshotBuffer( client: DaemonSocketClient, buffer: Uint8Array, signal?: AbortSignal, ): Promise<boolean> { if (client.socket.destroyed || signal?.aborted) { return false; } if (this.writeSerialized(client, buffer)) { return true; } if (signal?.aborted) return false; return new Promise<boolean>((resolveDrain) => { let settled = false; 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", onClose); resolveDrain(value); }; const onDrain = () => finish(true); const onClose = () => finish(false); client.socket.once("drain", onDrain); client.socket.once("close", onClose); client.socket.once("error", onClose); signal?.addEventListener("abort", onClose, { once: true }); }); }
private detachClient(client: DaemonSocketClient, activeSessionId?: string): void { const targets = activeSessionId ? [activeSessionId] : [...client.attachedActiveSessionIds]; for (const selector of targets) { const match = this.matchWorkers(selector)[0]; const resolvedId = match ? (match.summary.activeSessionId ?? match.summary.id) : selector; if (!client.attachedActiveSessionIds.delete(resolvedId)) { continue; } client.catchupActiveSessionIds?.delete(resolvedId); client.catchupPurposes?.delete(resolvedId); client.deferredSessionPayloadsDropped?.delete(resolvedId); this.write(client, { type: "session_detached", activeSessionId: resolvedId }); void this.syncWorkerExtensionUi(resolvedId); void this.evictEmptySessionOnLastDetach(resolvedId); } }
private async syncWorkerExtensionUi(activeSessionId: string): Promise<void> { const match = this.matchWorkers(activeSessionId)[0]; if (!match?.worker.client) { return; } await this.subscribeWorker(match.worker, match.summary.activeSessionId ?? match.summary.id).catch( () => undefined, ); }
private handleWorkerFrame( worker: ResidentWorker, frame: PrivateFrame<DaemonWorkerFrameHeader>, source?: DaemonWorkerClient, ): void { if (frame.header.kind !== "outbound") { return; } if (source !== undefined && source !== worker.client && source !== worker.pendingClient) { return; } worker.lastFrameAt = Date.now(); this.clearRosterStaleness(worker); const { outboundType, activeSessionId, snapshotId: frameSnapshotId, sessionEventType, payloadEncoding, snapshotPurpose, } = frame.header; if (outboundType === "roster_delta") { this.consumeWorkerRosterDelta(worker, frame.payload, source); return; } if (outboundType === "roster_heartbeat") { return; } if (outboundType === "heartbeats_changed") { worker.heartbeatSnapshotStale = true; this.broadcastHeartbeatsChanged(); return; } if (outboundType === "session_snapshot_begin" && activeSessionId) { try { const begin = JSON.parse(frame.payload.toString("utf8")) as Extract< DaemonOutbound, { type: "session_snapshot_begin" } >; if ( begin.type !== "session_snapshot_begin" || begin.activeSessionId !== activeSessionId || typeof begin.snapshotId !== "string" || (frameSnapshotId !== undefined && frameSnapshotId !== begin.snapshotId) || typeof begin.targetChunkBytes !== "number" || !begin.snapshot || !isSessionSummary(begin.snapshot.summary) ) { throw new Error("Worker returned an invalid snapshot begin frame"); } const publicSummary = this.publicSummary(worker, begin.snapshot.summary); const snapshot = { ...begin.snapshot, summary: publicSummary, messages: [], }; const result: DaemonAttachResult = { protocol: DAEMON_PROTOCOL_INFO, activeSessionId, snapshot, replay: { status: "complete", toSequence: snapshot.lastEventSequence, ...(snapshot.lastEventCursor ? { toCursor: snapshot.lastEventCursor } : {}), }, lastEventSequence: snapshot.lastEventSequence, ...(snapshot.lastEventCursor ? { lastEventCursor: snapshot.lastEventCursor } : {}), snapshotStream: { id: begin.snapshotId, messageCount: begin.messageCount, targetChunkBytes: begin.targetChunkBytes, }, client: { id: "supervisor", capabilities: ["chunked_snapshot"] }, }; const generations = this.snapshotGenerationsFor(worker, activeSessionId); let generation = generations.get(begin.snapshotId); if (generation?.incoming) { this.failSnapshotTransfer( worker, activeSessionId, begin.snapshotId, new Error(`Snapshot ${begin.snapshotId} restarted before completion`), snapshotPurpose, ); return; } // Snapshot summaries/state include live fields (for example activity and attached client // counts) that can change without advancing the transcript sequence. Treat the stable // transfer envelope as identity; duplicate chunks and end metadata are still byte-checked. const duplicate = generation?.transcript.complete === true && generation.end !== undefined && generation.result.snapshotStream?.messageCount === begin.messageCount && generation.result.snapshotStream?.targetChunkBytes === begin.targetChunkBytes && generation.result.lastEventSequence === result.lastEventSequence && generation.result.snapshot.lastEventSequence === result.snapshot.lastEventSequence && generation.result.snapshot.lastEventCursor?.generation === result.snapshot.lastEventCursor?.generation && generation.result.snapshot.lastEventCursor?.sequence === result.snapshot.lastEventCursor?.sequence; if (generation?.transcript.complete && !duplicate) { this.failSnapshotTransfer( worker, activeSessionId, begin.snapshotId, new Error(`Snapshot ${begin.snapshotId} did not match the cached transfer`), snapshotPurpose, ); return; } const currentGeneration = this.currentSnapshotGeneration(worker, activeSessionId); const currentResult = currentGeneration?.result ?? worker.snapshotCache.get(activeSessionId); const isOlderThanCurrent = currentGeneration !== undefined && currentGeneration.transcript.snapshotId !== begin.snapshotId && currentResult !== undefined && result.lastEventSequence < currentResult.lastEventSequence; if (isOlderThanCurrent && !generation) { return; } if (duplicate && generation) { generation.incoming = true; generation.duplicateChunkIndex = 0; generation.duplicateResult = result; generation.validation = this.createSnapshotDuplicateValidation(); if (currentGeneration === generation) { worker.snapshotCache.delete(activeSessionId); } return; } if ( currentGeneration && currentGeneration.transcript.snapshotId !== begin.snapshotId && !isOlderThanCurrent ) { if (!currentGeneration.transcript.complete && !currentGeneration.incoming) { this.failWorkerSnapshotCache( worker, activeSessionId, new Error(`Snapshot ${currentGeneration.transcript.snapshotId} was superseded`), false, currentGeneration.transcript.snapshotId, ); } else { this.retireWorkerSnapshotCache(worker, activeSessionId, currentGeneration.transcript); } } if (!generation) { const transcript = new SnapshotTranscriptCache({ activeSessionId, snapshotId: begin.snapshotId, cacheRoot: this.snapshotCacheRoot, targetChunkBytes: begin.targetChunkBytes, }); generation = { transcript, result, incoming: false, retired: isOlderThanCurrent, }; this.snapshotGenerationsFor(worker, activeSessionId).set(begin.snapshotId, generation); } generation.result = result; generation.begin = Buffer.from(frame.payload); generation.end = undefined; generation.incoming = true; generation.duplicateChunkIndex = undefined; generation.duplicateResult = undefined; generation.validation = undefined; if (!isOlderThanCurrent) { generation.retired = false; worker.transcriptCaches.set(activeSessionId, generation.transcript); worker.snapshotCache.set(activeSessionId, result); } } catch (error) { this.log(`Invalid worker snapshot begin frame: ${String(error)}`); this.failWorkerSnapshotCache( worker, activeSessionId, error instanceof Error ? error : new Error(String(error)), true, ); } return; } if (outboundType === "session_snapshot_chunk" && activeSessionId) { const snapshotId = frameSnapshotId ?? worker.transcriptCaches.get(activeSessionId)?.snapshotId; if (!snapshotId) { return; } const generation = this.snapshotGeneration(worker, activeSessionId, snapshotId); if (generation?.incoming) { try { const duplicateIndex = generation.duplicateChunkIndex; if (duplicateIndex === undefined) { generation.transcript.appendEncodedChunk(Buffer.from(frame.payload)); } else { const chunk = JSON.parse(frame.payload.toString("utf8")) as Extract< DaemonOutbound, { type: "session_snapshot_chunk" } >; if ( chunk.type !== "session_snapshot_chunk" || chunk.activeSessionId !== activeSessionId || chunk.snapshotId !== generation.transcript.snapshotId || chunk.index !== duplicateIndex || !generation.transcript.readChunk(duplicateIndex).equals(Buffer.from(frame.payload)) ) { throw new Error( `Duplicate snapshot ${generation.transcript.snapshotId} did not match cached bytes`, ); } generation.duplicateChunkIndex = duplicateIndex + 1; } } catch (error) { this.failSnapshotTransfer( worker, activeSessionId, generation.transcript.snapshotId, error instanceof Error ? error : new Error(String(error)), snapshotPurpose, ); } } return; } if (outboundType === "session_snapshot_end" && activeSessionId) { const snapshotId = frameSnapshotId ?? worker.transcriptCaches.get(activeSessionId)?.snapshotId; if (!snapshotId) { return; } const generation = this.snapshotGeneration(worker, activeSessionId, snapshotId); if (!generation?.incoming) { return; } const transcript = generation.transcript; try { const duplicateChunkCount = generation.duplicateChunkIndex; if (duplicateChunkCount === undefined) { transcript.markComplete(); if (!generation.begin) { throw new Error(`Snapshot ${transcript.snapshotId} has no begin frame`); } generation.end = Buffer.from(frame.payload); } else { const end = JSON.parse(frame.payload.toString("utf8")) as Extract< DaemonOutbound, { type: "session_snapshot_end" } >; if ( end.type !== "session_snapshot_end" || end.activeSessionId !== activeSessionId || end.snapshotId !== transcript.snapshotId || end.chunkCount !== duplicateChunkCount || end.chunkCount !== transcript.chunkCount || !generation.end?.equals(frame.payload) ) { throw new Error(`Duplicate snapshot ${transcript.snapshotId} ended with different metadata`); } if (!generation.duplicateResult) { throw new Error(`Duplicate snapshot ${transcript.snapshotId} has no result`); } generation.result = generation.duplicateResult; if (worker.transcriptCaches.get(activeSessionId) === transcript) { worker.snapshotCache.set(activeSessionId, generation.duplicateResult); } this.settleSnapshotDuplicateValidation(generation); } generation.incoming = false; generation.duplicateChunkIndex = undefined; generation.duplicateResult = undefined; } catch (error) { this.failSnapshotTransfer( worker, activeSessionId, transcript.snapshotId, error instanceof Error ? error : new Error(String(error)), snapshotPurpose, ); return; } const published = worker.transcriptCaches.get(activeSessionId) === transcript; if (generation.retired) { this.deleteSnapshotGeneration(worker, activeSessionId, generation); transcript.dispose(); } if (published && (snapshotPurpose === "replacement" || snapshotPurpose === "catchup")) { this.queueSnapshotResync(activeSessionId, snapshotPurpose); } return; } if (outboundType === "session_snapshot_failed" && activeSessionId) { try { const failed = JSON.parse(frame.payload.toString("utf8")) as Extract< DaemonOutbound, { type: "session_snapshot_failed" } >; if ( failed.type !== "session_snapshot_failed" || failed.activeSessionId !== activeSessionId || typeof failed.snapshotId !== "string" || typeof failed.error !== "string" || (frameSnapshotId !== undefined && frameSnapshotId !== failed.snapshotId) ) { throw new Error("Worker returned an invalid snapshot failure frame"); } const currentGeneration = this.currentSnapshotGeneration(worker, activeSessionId); const generation = this.snapshotGeneration(worker, activeSessionId, failed.snapshotId) ?? (currentGeneration?.transcript.snapshotId === failed.snapshotId ? currentGeneration : undefined); if (!generation) { return; } this.failSnapshotTransfer( worker, activeSessionId, failed.snapshotId, new Error(failed.error), snapshotPurpose, ); } catch (error) { this.failWorkerSnapshotCache( worker, activeSessionId, error instanceof Error ? error : new Error(String(error)), true, ); } return; } if ( outboundType === "daemon_hello" || outboundType === "response" || outboundType === "session_list_progress" || outboundType === "session_list_item" || outboundType === "session_attached" || outboundType === "session_detached" || !activeSessionId ) { return; } let publicPayload = frame.payload; let decodedOutbound: DaemonOutbound | undefined; if (payloadEncoding === "assistant-delta") { let compactValue: unknown; try { compactValue = JSON.parse(frame.payload.toString("utf8")); } catch { this.scheduleCompactCatchup(worker, activeSessionId); return; } if (!isCompactAssistantDelta(compactValue)) { this.scheduleCompactCatchup(worker, activeSessionId); return; } const reconstructed = this.streamReconstructor.reconstruct(compactValue); if (!reconstructed) { this.scheduleCompactCatchup(worker, activeSessionId); return; } publicPayload = Buffer.from(serializeJsonLine(reconstructed)); } else if ( sessionEventType === "message_start" || sessionEventType === "message_end" || outboundType === "session_replaced" || outboundType === "session_resynced" || outboundType === "session_closed" ) { try { decodedOutbound = JSON.parse(frame.payload.toString("utf8")) as DaemonOutbound; this.streamReconstructor.observe(decodedOutbound); } catch { // A malformed worker event is still isolated to this worker connection. } } const replacementSnapshotFollows = decodedOutbound?.type === "session_replaced" && decodedOutbound.snapshotFollows === true; this.invalidateWorkerSnapshot( worker, activeSessionId, outboundType === "session_replaced" || outboundType === "session_closed" || isFinalizedTranscriptEvent(sessionEventType), ); for (const client of this.clients) { if (!client.attachedActiveSessionIds.has(activeSessionId)) { continue; } if (replacementSnapshotFollows && !client.capabilities.has("chunked_snapshot")) { continue; } if (outboundType === "extension_ui_request" && !client.supportsExtensionUi) { continue; } // Snapshots cannot recover extension requests, so never defer or drop them. if (outboundType === "extension_ui_request") { this.writeSerialized(client, publicPayload); continue; } if (outboundType === "session_closed") { client.snapshotTransferAbortControllers?.get(activeSessionId)?.abort(); this.discardDeferredSessionPayloads(client, activeSessionId); client.catchupActiveSessionIds?.delete(activeSessionId); client.catchupPurposes?.delete(activeSessionId); this.writeSerialized(client, publicPayload); continue; } if (client.snapshotActiveSessionIds?.has(activeSessionId)) { // A replacement swaps the session wholesale; buffered payloads from // the old session must never replay, so fall back to a replacement // catch-up snapshot. if (outboundType === "session_replaced") { this.discardDeferredSessionPayloads(client, activeSessionId); client.deferredSessionPayloadsDropped ??= new Set(); client.deferredSessionPayloadsDropped.add(activeSessionId); this.queueCatchup(client, activeSessionId, "replacement"); continue; } if (this.deferSessionPayload(client, activeSessionId, publicPayload)) { continue; } // The deferral buffer overflowed: fall back to a full resync. this.queueCatchup(client, activeSessionId, "resync"); continue; } if (client.deferredSessionPayloadsDropped?.has(activeSessionId)) continue; if (client.backpressured === true) { this.queueCatchup(client, activeSessionId, outboundType === "session_replaced" ? "replacement" : "resync"); continue; } this.writeSerialized(client, publicPayload); } if ( decodedOutbound?.type === "session_closed" && decodedOutbound.reason === "shutdown" && activeSessionId === worker.descriptor.rootActiveSessionId && !this.shuttingDown ) { worker.intentionalStop = true; // An exact stop owns its registration and descriptor cleanup until its // tuple assertions complete. A synchronous root shutdown event can arrive // before its request resolves, so leave both intact while it is active. if ((this.workerStopCounts?.get(worker) ?? 0) === 0) { this.invalidateWorkerSessionInputPauses(worker, "Session worker stopped while input was paused"); this.workers.delete(worker.descriptor.workerId); this.flipWorkerRosterEntriesInactive(worker); this.deleteWorkerDescriptor(worker); } } }
private invalidateWorkerSnapshot(worker: ResidentWorker, activeSessionId: string, transcriptChanged = true): void { worker.snapshotCache.delete(activeSessionId); if (!transcriptChanged) { return; } worker.snapshotLoads.delete(`${activeSessionId}:chunked`); worker.snapshotLoads.delete(`${activeSessionId}:full`); const transcript = worker.transcriptCaches.get(activeSessionId); if (transcript) { this.retireWorkerSnapshotCache(worker, activeSessionId, transcript); } }
private scheduleCompactCatchup(worker: ResidentWorker, activeSessionId: string): void { if (this.compactCatchupInProgress.has(activeSessionId)) { return; } this.compactCatchupInProgress.add(activeSessionId); this.invalidateWorkerSnapshot(worker, activeSessionId); const clients = [...this.clients].filter((client) => client.attachedActiveSessionIds.has(activeSessionId)); for (const client of clients) { this.queueCatchup(client, activeSessionId); } void Promise.all(clients.map((client) => this.catchUpClient(client))) .catch((error) => this.log(`Failed compact catch-up for ${activeSessionId}: ${String(error)}`)) .finally(() => { this.compactCatchupInProgress.delete(activeSessionId); }); }
private queueCatchup( 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 relayed payloads until the snapshot finishes; false requests a catch-up. */ private deferSessionPayload(client: DaemonSocketClient, activeSessionId: string, payload: Buffer): boolean { if (client.deferredSessionPayloadsDropped?.has(activeSessionId)) { return false; } const deferred = client.deferredSessionPayloads?.get(activeSessionId) ?? { payloads: [], bytes: 0 }; if ( deferred.payloads.length >= MAX_DEFERRED_SESSION_PAYLOADS || deferred.bytes + payload.byteLength > MAX_DEFERRED_SESSION_BYTES ) { // A catch-up snapshot supersedes the buffered payloads. client.deferredSessionPayloadsDropped ??= new Set(); client.deferredSessionPayloadsDropped.add(activeSessionId); this.discardDeferredSessionPayloads(client, activeSessionId); return false; } deferred.payloads.push(payload); deferred.bytes += payload.byteLength; client.deferredSessionPayloads ??= new Map(); client.deferredSessionPayloads.set(activeSessionId, deferred); return true; }
/** Replay payloads deferred during a completed snapshot stream, in order. */ private flushDeferredSessionPayloads(client: DaemonSocketClient, activeSessionId: string): void { const payloads = client.deferredSessionPayloads?.get(activeSessionId)?.payloads; if (!payloads || payloads.length === 0) { return; } client.deferredSessionPayloads?.delete(activeSessionId); if (!client.attachedActiveSessionIds.has(activeSessionId)) return; for (const payload of payloads) { if (client.socket.destroyed) { return; } if (client.backpressured || !this.writeSerialized(client, payload)) { this.queueCatchup(client, activeSessionId, "resync"); return; } } }
private discardDeferredSessionPayloads(client: DaemonSocketClient, activeSessionId: string): void { client.deferredSessionPayloads?.delete(activeSessionId); }
/** * Flush or discard the payloads withheld during a snapshot stream. Only the * last active stream for the session owns them; an overlapping stream * replays them after it completes. */ private releaseDeferredSessionPayloads( client: DaemonSocketClient, activeSessionId: string, delivered: boolean, ): void { if (client.snapshotActiveSessionIds?.has(activeSessionId)) { return; } if (delivered) { this.flushDeferredSessionPayloads(client, activeSessionId); } else { this.discardDeferredSessionPayloads(client, activeSessionId); } }
private catchUpClient(client: DaemonSocketClient): Promise<void> { if (client.catchupPromise) { return client.catchupPromise; } if (client.snapshotStreaming || client.backpressured || client.catchupRetryTimer) { return Promise.resolve(); } const catchup = this.drainClientCatchupQueue(client) .catch((error) => { this.log(`Failed to catch up client ${client.id}: ${String(error)}`); if (client.socket.destroyed || !client.catchupActiveSessionIds?.size) return; client.catchupRetryTimer = setTimeout(() => { client.catchupRetryTimer = undefined; void this.catchUpClient(client); }, 250); client.catchupRetryTimer.unref(); }) .finally(() => { if (client.catchupPromise === catchup) { client.catchupPromise = undefined; } }); client.catchupPromise = catchup; return catchup; }
private async drainClientCatchupQueue(client: DaemonSocketClient): Promise<void> { while ( !client.socket.destroyed && !client.snapshotStreaming && !client.backpressured && client.catchupActiveSessionIds?.size ) { await this.drainClientCatchups(client); } }
private async drainClientCatchups(client: DaemonSocketClient): Promise<void> { if (client.socket.destroyed) { return; } const pending = [...(client.catchupActiveSessionIds ?? [])].map((activeSessionId) => ({ activeSessionId, purpose: client.catchupPurposes?.get(activeSessionId) ?? ("resync" as const), })); client.catchupActiveSessionIds?.clear(); client.catchupPurposes?.clear(); let retryError: Error | undefined; for (let index = 0; index < pending.length; index++) { const { activeSessionId, purpose } = pending[index]!; let releaseTranscript: (() => void) | undefined; const releaseSnapshotReservation = this.reserveSnapshotStream(client, activeSessionId); const snapshotSignal = client.snapshotTransferAbortControllers?.get(activeSessionId)?.signal; try { const attached = await this.attachClient(client, { type: "attach", activeSessionId, capabilities: [...client.capabilities], supportsExtensionUi: client.supportsExtensionUi, }); releaseTranscript = attached.releaseTranscript; if (client.capabilities.has("chunked_snapshot")) { const transcript = attached.transcript; if (!transcript) { throw new Error("Session worker did not provide a snapshot transcript"); } if (purpose === "replacement") { this.write(client, { type: "session_replaced", activeSessionId, state: attached.result.snapshot.state, messages: [], snapshotFollows: true, meta: createDaemonEventMeta( activeSessionId, attached.result.lastEventSequence, undefined, attached.result.lastEventCursor?.generation, ), }); } await this.streamSnapshot( client, attached.worker, this.createStreamedAttachResult(attached.result, transcript), transcript, purpose, releaseTranscript, releaseSnapshotReservation, ); releaseTranscript = undefined; continue; } const meta = createDaemonEventMeta( activeSessionId, attached.result.lastEventSequence, undefined, attached.result.lastEventCursor?.generation, ); const catchup: DaemonOutbound = purpose === "replacement" ? { type: "session_replaced", activeSessionId, state: attached.result.snapshot.state, messages: attached.result.snapshot.messages, meta, } : { type: "session_resynced", activeSessionId, snapshot: attached.result.snapshot, meta, }; const accepted = this.write(client, catchup); releaseSnapshotReservation(); this.releaseDeferredSessionPayloads(client, activeSessionId, true); if (!accepted) { for (const remaining of pending.slice(index + 1)) { this.queueCatchup(client, remaining.activeSessionId, remaining.purpose); } return; } } catch (error) { releaseTranscript?.(); if (client.attachedActiveSessionIds.has(activeSessionId) && !snapshotSignal?.aborted) { this.queueCatchup(client, activeSessionId, purpose); } client.deferredSessionPayloadsDropped ??= new Set(); client.deferredSessionPayloadsDropped.add(activeSessionId); this.discardDeferredSessionPayloads(client, activeSessionId); retryError ??= error instanceof Error ? error : new Error(String(error)); } finally { releaseSnapshotReservation(); this.releaseDeferredSessionPayloads(client, activeSessionId, false); } } // Retry failed sessions only after the rest of the batch has had a chance to recover. if (retryError) throw retryError; }
private async prepareUpdateRestart(): Promise<DaemonUpdateRestartManifest> { if (this.updateRestartPhase !== undefined) throw new Error("Daemon is already preparing an update restart"); this.updateRestartPhase = "draining"; try { const deadline = Date.now() + UPDATE_RESTART_PREPARE_DEADLINE_MS; const abort = AbortSignal.timeout(Math.min(UPDATE_RESTART_MUTATION_DRAIN_TIMEOUT_MS, deadline - Date.now())); await this.mutationDrain.waitForDrain(1, abort, "Timed out draining daemon mutations for update restart"); this.updateRestartPhase = "fencing"; await this.mutationDrain.waitForDrain(1, abort, "Timed out draining daemon mutations for update restart"); const manifest = await this.prepareUpdateRestartFenced(deadline); this.updateRestartPhase = "prepared"; return manifest; } catch (error) { this.updateRestartPhase = undefined; this.scheduleScheduledSessionWakeRecompute(); throw error; } }
/** * Collect the workers an update restart can actually hand over. * * The restart waits for every resident worker to acknowledge its checkpoint, and * a worker whose process is provably gone can never answer: waiting on it wedges * the update permanently, with no supported way out (it is not in the roster, so * `stop` cannot name it, and the reclaim path only runs when its session is * opened again). A supervisor-owned corpse is reclaimed here — the same cleanup a * later open would run — while a client-owned one is only skipped, so its owner * still finds its registration and its session on disk. */ private async retireProvablyDeadResidents(): Promise<ResidentWorker[]> { const participating: ResidentWorker[] = []; for (const worker of [...this.workers.values()]) { if (worker.client !== undefined || !this.isProvablyDeadWorker(worker)) { participating.push(worker); continue; } if (worker.descriptor.ownerClientId !== undefined) { this.log( `Skipping dead client-owned worker ${worker.descriptor.workerId} in the update restart; its owner still holds the registration`, ); continue; } if (worker.descriptor.stopRequestedAt !== undefined) { this.scheduleWorkerStopFinalization(worker); this.log(`Retired stopping worker ${worker.descriptor.workerId} from the update restart`); continue; } try { await this.reclaimDeadWorkerRegistration(worker); this.log(`Reclaimed dead worker ${worker.descriptor.workerId} before the update restart`); } catch (error) { this.log( `Could not reclaim dead worker ${worker.descriptor.workerId} before the update restart: ${String(error)}`, ); } } return participating; }
private async prepareUpdateRestartFenced(deadline: number): Promise<DaemonUpdateRestartManifest> { const residents = await this.retireProvablyDeadResidents(); const unavailable = residents.find( (worker) => this.isWorkerStopping(worker) || worker.descriptor.lifecycle !== "ready" || worker.client === undefined, ); if (unavailable) { throw new Error( `Cannot prepare update restart while resident worker ${unavailable.descriptor.workerId} is ${this.effectiveWorkerState(unavailable)}${unavailable.client ? "" : " and disconnected"}`, ); } const workers = residents as Array<ResidentWorker & { client: DaemonWorkerClient }>; const acknowledged: ResidentWorker[] = []; const preparationResults = await Promise.allSettled( workers.map(async (worker) => { const client = worker.client; const response = await client .requestWorker( { type: "worker_prepare_update" }, Math.max(1, Math.min(UPDATE_RESTART_WORKER_REQUEST_TIMEOUT_MS, deadline - Date.now())), ) .catch((error: unknown) => { const detail = error instanceof Error ? error.message : String(error); throw new Error(`Worker ${worker.descriptor.workerId} did not prepare update restart: ${detail}`, { cause: error, }); }); if (!response.success) { throw new Error(`Worker ${worker.descriptor.workerId} rejected update restart: ${response.error}`); } worker.updateRestartPrepareClient = client; acknowledged.push(worker); if (!response.data || typeof response.data !== "object") { throw new Error("Worker returned an invalid update manifest"); } if (worker.client !== client || worker.descriptor.lifecycle !== "ready") { throw new Error(`Worker ${worker.descriptor.workerId} disconnected during update preparation`); } const manifest = response.data as DaemonUpdateRestartManifest; if (manifest.formatVersion !== DAEMON_UPDATE_RESTART_FORMAT_VERSION) { throw new Error(`Worker returned unsupported update manifest version ${manifest.formatVersion}`); } if ( !manifest.sessions.some( (session) => session.activeSessionId === worker.descriptor.rootActiveSessionId, ) && !manifest.discardedActiveSessionIds?.includes(worker.descriptor.rootActiveSessionId) ) { throw new Error( `Worker ${worker.descriptor.workerId} omitted its root disposition from the update manifest`, ); } return { worker, manifest }; }), ); const cancelAcknowledged = async () => { await Promise.all( acknowledged.map(async (worker) => { const prepareClient = worker.updateRestartPrepareClient; worker.updateRestartPrepareClient = undefined; if (!prepareClient) return; try { const response = await prepareClient.requestWorker({ type: "worker_cancel_update" }, 5000); if (!response.success) throw new Error(response.error); } catch (error) { this.log( `Could not cancel prepared worker ${worker.descriptor.workerId}; reconnecting it: ${String(error)}`, ); prepareClient.close(); if (worker.client && worker.client !== prepareClient) worker.client.close(); } }), ); }; const preparationFailure = preparationResults.find( (result): result is PromiseRejectedResult => result.status === "rejected", ); if (preparationFailure) { await cancelAcknowledged(); throw preparationFailure.reason; } const prepared = preparationResults.flatMap((result) => result.status === "fulfilled" ? [result.value.worker] : [], ); const responses = preparationResults.flatMap((result) => result.status === "fulfilled" ? [result.value.manifest] : [], ); const discardedActiveSessionIds = responses.flatMap((manifest) => manifest.discardedActiveSessionIds ?? []); const manifest: DaemonUpdateRestartManifest = { formatVersion: DAEMON_UPDATE_RESTART_FORMAT_VERSION, createdAt: new Date().toISOString(), sessions: responses.flatMap((manifest) => manifest.sessions), ...(discardedActiveSessionIds.length > 0 ? { discardedActiveSessionIds } : {}), }; // A worker that disconnected after preparing cancelled its checkpoint with // the old client; a recovered replacement may have admitted inputs past the // captured manifest. Abort before the manifest is persisted so the caller's // fallback cannot restore from the stale checkpoint. const staleWorker = prepared.find((worker) => worker.client !== worker.updateRestartPrepareClient); if (staleWorker) { await cancelAcknowledged(); throw new Error( `Worker ${staleWorker.descriptor.workerId} reconnected during update preparation; its checkpoint is stale`, ); } try { this.validateAndPersistUpdateManifest(manifest); } catch (error) { await cancelAcknowledged(); throw error; } // Commit through the connection that owns the prepared transaction; a client // swapped in after the check above must fail the commit rather than reach a // worker that no longer holds the checkpoint. const commitClients = new Map(prepared.map((worker) => [worker, worker.updateRestartPrepareClient])); for (const worker of prepared) worker.updateRestartPrepareClient = undefined; const commitResults = await Promise.allSettled( prepared.map(async (worker) => { const client = commitClients.get(worker); if (!client) throw new Error(`Worker ${worker.descriptor.workerId} disconnected before update commit`); const response = await client.requestWorker( { type: "worker_commit_update" }, UPDATE_RESTART_WORKER_REQUEST_TIMEOUT_MS, ); if (!response.success) throw new Error(response.error); }), ); const commitFailure = commitResults.find( (result): result is PromiseRejectedResult => result.status === "rejected", ); if (commitFailure) { this.log(`Update restart commit response failed; forcing restart completion: ${String(commitFailure.reason)}`); await Promise.allSettled(prepared.map((worker) => this.stopWorker(worker, false, true))); return manifest; } const stopResults = await Promise.allSettled(prepared.map((worker) => this.stopWorker(worker, false))); if (stopResults.some((result) => result.status === "rejected")) { this.log("A committed update worker did not stop gracefully; forcing restart completion"); await Promise.allSettled(prepared.map((worker) => this.stopWorker(worker, false, true))); } return manifest; }
private validateAndPersistUpdateManifest(manifest: DaemonUpdateRestartManifest): void { if (manifest.formatVersion !== DAEMON_UPDATE_RESTART_FORMAT_VERSION) { throw new Error(`Unsupported update manifest version ${manifest.formatVersion}`); } const activeSessionIds = new Set<string>(); const sessionFiles = new Set<string>(); for (const discardedActiveSessionId of manifest.discardedActiveSessionIds ?? []) { if (!discardedActiveSessionId || activeSessionIds.has(discardedActiveSessionId)) { throw new Error("Update manifest contains an invalid discarded session disposition"); } activeSessionIds.add(discardedActiveSessionId); } for (const session of manifest.sessions) { if (!session.activeSessionId || !session.sessionFile) { throw new Error("Update manifest contains an incomplete session checkpoint"); } if (activeSessionIds.has(session.activeSessionId)) { throw new Error(`Update manifest contains duplicate active session ${session.activeSessionId}`); } const sessionFile = canonicalSessionPath(session.sessionFile); if (sessionFiles.has(sessionFile)) { throw new Error(`Update manifest contains duplicate session file ${sessionFile}`); } activeSessionIds.add(session.activeSessionId); sessionFiles.add(sessionFile); } const agentDir = this.defaultSessionConfig.agentDir; if (!agentDir) { throw new Error("Daemon supervisor config is missing agentDir"); } const path = getDaemonUpdateRestartManifestPath(this.socketPath, agentDir); mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); writeFileAtomicSync(path, `${JSON.stringify(manifest)}\n`, { mode: 0o600, beforeRename: (tempPath) => { const validated = JSON.parse(readFileSync(tempPath, "utf8")) as DaemonUpdateRestartManifest; if (!Array.isArray(validated.sessions) || validated.sessions.length !== manifest.sessions.length) { throw new Error("Could not validate aggregate update manifest"); } }, }); }
/** * Verdict on whether a pid is still the process we launched. Callers must * be conservative in both directions: signal a pid only on "current" * (never SIGKILL a recycled pid), and clean up a registration only on * "gone"/"replaced" (never orphan a live worker because a transient * identity lookup failed). */ private processIdentity( pid: number, processStartId: string | undefined, ): "current" | "replaced" | "gone" | "unknown" { if (!isProcessAlive(pid)) { return "gone"; } if (processStartId === undefined) { return "unknown"; } const observed = getProcessStartId(pid); if (observed === undefined) { return "unknown"; } return observed === processStartId ? "current" : "replaced"; }
/** * Keep an exact stop's registration and descriptor authoritative while any * part of its cleanup is in flight. Root kills acquire this before forwarding * because a synchronous shutdown event may arrive before the worker replies. */ private acquireWorkerStopOwnership(worker: ResidentWorker): () => void { if (!this.workerStopCounts) this.workerStopCounts = new Map(); const stopCounts = this.workerStopCounts; stopCounts.set(worker, (stopCounts.get(worker) ?? 0) + 1); let released = false; return () => { if (released) return; released = true; const remaining = (stopCounts.get(worker) ?? 1) - 1; if (remaining === 0) stopCounts.delete(worker); else stopCounts.set(worker, remaining); }; }
private async stopWorker( worker: ResidentWorker, removeDescriptor: boolean, force = false, archiveSession = false, recoveryCleanup = false, directChild?: { child: ChildProcess; closed: Promise<void> }, reason?: string, ): Promise<void> { if (reason) worker.stopReason = reason; const releaseStopOwnership = this.acquireWorkerStopOwnership(worker); try { if (reason !== undefined) { await this.stopWorkerUntracked( worker, removeDescriptor, force, archiveSession, recoveryCleanup, directChild, reason, ); } else { await this.stopWorkerUntracked( worker, removeDescriptor, force, archiveSession, recoveryCleanup, directChild, ); } } finally { releaseStopOwnership(); } }
private async stopWorkerUntracked( worker: ResidentWorker, removeDescriptor: boolean, force = false, archiveSession = false, recoveryCleanup = false, directChild?: { child: ChildProcess; closed: Promise<void> }, reason?: string, ): Promise<void> { if (worker.ownerCleanupTimer) { clearTimeout(worker.ownerCleanupTimer); worker.ownerCleanupTimer = undefined; } if (!recoveryCleanup) { worker.stopRevision++; } // A retry can rescind this stop and relaunch the worker while we await // below. Bind every liveness check and signal to the process this stop // entered with, and abort cleanup once the stop no longer applies: the // pid changed (relaunched) or a removeDescriptor stop lost its tombstone // (rescinded, even before the successor pid lands). const entryPid = worker.descriptor.pid; const entryStartId = worker.descriptor.processStartId; const assertStopStillApplies = () => { if (directChild) { return; } if ( worker.descriptor.pid !== entryPid || (removeDescriptor && worker.descriptor.stopRequestedAt === undefined) ) { throw new Error(`Session worker ${worker.descriptor.workerId} was relaunched during stop`); } }; try { if (removeDescriptor) { this.persistWorkerStopTombstone(worker, archiveSession); } else { worker.intentionalStop = true; worker.descriptor.lifecycle = "recovering"; this.persistWorker(worker); } } catch (error) { if (!directChild) { throw error; } this.reportCleanupFailure(`worker rollback state ${worker.descriptor.workerId}`, error); } const transferError = new Error("Session worker stopped during snapshot transfer"); const generationTranscripts = new Set<SnapshotTranscriptCache>(); for (const [activeSessionId, generations] of [...(worker.snapshotGenerations ?? new Map())]) { for (const generation of [...generations.values()]) { generationTranscripts.add(generation.transcript); if (generation.incoming || !generation.transcript.complete || generation.validation) { this.failSnapshotGeneration(worker, activeSessionId, generation, transferError); } else { generation.transcript.dispose(); this.deleteSnapshotGeneration(worker, activeSessionId, generation); } } } for (const transcript of worker.transcriptCaches?.values() ?? []) { if (!generationTranscripts.has(transcript) && !transcript.complete) { transcript.markFailed(transferError); } transcript.dispose(); } worker.transcriptCaches?.clear(); worker.snapshotCache?.clear(); worker.snapshotGenerations?.clear(); if (reason) worker.stopReason = reason; const stopReason = worker.stopReason ?? "stop requested"; const client = worker.client; if (client) { this.log( `Stopping session worker ${worker.descriptor.workerId} (pid ${entryPid}): reason=${stopReason}, action=shutdown request`, ); if (archiveSession) { await client .requestWorker({ type: "worker_archive_and_shutdown" }, force ? 1000 : 5000) .catch(() => undefined); } else { await client.request({ type: "shutdown" }, force ? 1000 : 5000).catch(() => undefined); } client.close(); // A reconnect can swap in another client while the shutdown request is in flight. if (worker.client === client) { worker.client = undefined; } } else if (directChild) { worker.stopSignalSent = "SIGTERM"; this.log( `Stopping session worker ${worker.descriptor.workerId} (pid ${entryPid}): reason=${stopReason}, action=sent SIGTERM`, ); directChild.child.kill("SIGTERM"); } else if (this.processIdentity(entryPid, entryStartId) === "current") { worker.stopSignalSent = "SIGTERM"; this.log( `Stopping session worker ${worker.descriptor.workerId} (pid ${entryPid}): reason=${stopReason}, action=sent SIGTERM`, ); signalProcessGroupOrProcess(entryPid, "SIGTERM"); } // Identity-aware in both directions: a replaced pid counts as gone (never // signal a recycled pid) while an unknown identity counts as alive (never // clean up a possibly-live worker on a transient lookup failure). kill(0) // runs on every poll; the expensive identity check is throttled. let identityVerdict: "current" | "replaced" | "gone" | "unknown" = "current"; let identityCheckedAt = 0; const isWorkerProcessAlive = () => { if (directChild) { return directChild.child.exitCode === null && directChild.child.signalCode === null; } if (!processIdExists(entryPid)) { return false; } const now = Date.now(); if (now - identityCheckedAt >= LIVENESS_IDENTITY_RECHECK_MS) { identityCheckedAt = now; identityVerdict = this.processIdentity(entryPid, entryStartId); } return identityVerdict !== "replaced" && identityVerdict !== "gone"; }; const gracefulDeadline = Date.now() + (force ? 500 : 2000); while (isWorkerProcessAlive() && Date.now() < gracefulDeadline) { await delay(25); } let sigkillSent = false; if (force && isWorkerProcessAlive()) { worker.stopSignalSent = "SIGKILL"; this.log( `Stopping session worker ${worker.descriptor.workerId} (pid ${entryPid}): reason=${stopReason}, action=sent SIGKILL`, ); if (directChild) { sigkillSent = directChild.child.kill("SIGKILL"); } else if (this.processIdentity(entryPid, entryStartId) === "current") { // Recheck without the cache: the pid may have been recycled. signalProcessGroupOrProcess(entryPid, "SIGKILL"); sigkillSent = true; } const forceDeadline = Date.now() + 1000; while (isWorkerProcessAlive() && Date.now() < forceDeadline) { await delay(25); } } if (isWorkerProcessAlive()) { worker.intentionalStop = worker.descriptor.stopRequestedAt !== undefined; if (removeDescriptor) { this.scheduleWorkerStopFinalization(worker); } throw new WorkerStopTimeoutError( `Session worker ${worker.descriptor.workerId} did not stop${sigkillSent ? " after SIGKILL" : ""}`, ); } if (directChild) { await directChild.closed; } assertStopStillApplies(); if (removeDescriptor && worker.descriptor.archiveOnStop) { if (force) { this.reclaimStoppedWorkerCronLock(worker); } await this.finalizeArchivedWorkerStop(worker); assertStopStillApplies(); } this.invalidateWorkerSessionInputPauses(worker, "Session worker stopped while input was paused"); // Client-owned schedules die with the registration, like their roster rows. Cancel // before the worker leaves the map, so no recompute sees the tree uncovered mid-stop. let ephemeralCancelSettled = true; if (removeDescriptor && worker.descriptor.ownerClientId !== undefined) { ephemeralCancelSettled = await this.cancelEphemeralWorkerScheduledJobs(worker); } this.workers.delete(worker.descriptor.workerId); this.flipWorkerRosterEntriesInactive(worker); // A failed cancel keeps the stop tombstone as the durable intent; the enumeration retry or the next boot finishes it. if (removeDescriptor && ephemeralCancelSettled) { this.deleteWorkerDescriptor(worker); } if (!this.shuttingDown) { this.broadcastHeartbeatsChanged(); } }
/** * A stop that timed out leaves a tombstoned registration behind. Keep * escalating in the background until the process is gone, then finish the * interrupted cleanup instead of leaving a dead worker registered forever. */ private scheduleWorkerStopFinalization(worker: ResidentWorker): void { if (worker.stopFinalization) { return; } worker.stopFinalization = this.finalizeTimedOutWorkerStop(worker).finally(() => { worker.stopFinalization = undefined; }); }
private async finalizeTimedOutWorkerStop(worker: ResidentWorker): Promise<void> { // Bind to the exact process generation being stopped: a retry can rescind // the stop and relaunch with a new pid, and the OS can recycle the old // pid. The finalizer must never follow either successor. const pid = worker.descriptor.pid; const processStartId = worker.descriptor.processStartId; const stopRevision = worker.stopRevision; const isStopGenerationCurrent = () => this.workers.get(worker.descriptor.workerId) === worker && worker.stopRevision === stopRevision && worker.descriptor.stopRequestedAt !== undefined && worker.descriptor.pid === pid; // A replaced pid counts as gone (never SIGKILL a recycled pid); an // unobservable identity counts as alive (never clean up a possibly-live // worker). kill(0) probes every poll; ps-backed checks are throttled. let stoppedVerdict = true; let stoppedCanSignal = processStartId !== undefined; let stoppedCheckedAt = 0; const isStoppedProcessAlive = () => { if (!processIdExists(pid)) { return false; } const now = Date.now(); if (now - stoppedCheckedAt < LIVENESS_IDENTITY_RECHECK_MS) { return stoppedVerdict; } stoppedCheckedAt = now; if (!isProcessAlive(pid)) { stoppedVerdict = false; } else if (processStartId === undefined) { stoppedVerdict = true; // Without an identity captured while the original worker was known // alive, this pid may now belong to an unrelated process. Keep // waiting for it to disappear, but never escalate by pid alone. stoppedCanSignal = false; } else { const observed = getProcessStartId(pid); stoppedVerdict = observed !== processStartId ? observed === undefined : true; stoppedCanSignal = observed === processStartId; } return stoppedVerdict; }; const sigkillDeadline = Date.now() + STOP_FINALIZATION_SIGKILL_GRACE_MS; let killed = false; while (!this.shuttingDown) { if (!isStopGenerationCurrent()) { return; } if (!isStoppedProcessAlive()) { break; } if (!killed && stoppedCanSignal && Date.now() >= sigkillDeadline) { // Recheck without the cache before signalling a possibly recycled pid. // An unobservable identity skips this attempt but keeps escalation // armed so a wedged worker is still killed on a later pass. const observedNow = processStartId === undefined ? undefined : getProcessStartId(pid); if (processStartId === undefined || observedNow === processStartId) { signalProcessGroupOrProcess(pid, "SIGKILL"); killed = true; } } await unrefDelay(STOP_FINALIZATION_RECHECK_MS); } // Retry transient cleanup failures (for example catalog archival) so a // dead worker's registration is never stranded permanently. Each attempt // bumps the worker's stopRevision, so rescission is detected through the // registration and tombstone instead of the waiting-phase snapshot. const isCleanupStillWanted = () => this.workers.get(worker.descriptor.workerId) === worker && worker.descriptor.stopRequestedAt !== undefined && worker.descriptor.pid === pid; while (!this.shuttingDown && isCleanupStillWanted()) { try { await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); this.log(`Finalized timed-out stop for worker ${worker.descriptor.workerId}`); return; } catch (error) { this.reportCleanupFailure(`timed-out worker stop ${worker.descriptor.workerId}`, error); await unrefDelay(STOP_FINALIZATION_RETRY_MS); } } }
private async finalizeArchivedWorkerStop(worker: ResidentWorker): Promise<void> { const context = this.workerSessionArtifactContext(worker); if (!context) { return; } if (worker.descriptor.rootSessionId) { const cronStore = AgentCronJobStore.forSessionArtifacts(); cronStore.registerSessionArtifact(worker.descriptor.rootSessionId, context.artifactDir); cronStore.cancelJobsForSession({ sessionId: worker.descriptor.rootSessionId, sessionFile: context.sessionFile, }); await this.catalog.archive(context.sessionFile, worker.descriptor.rootSessionId); } }
private async cancelScheduledJobsForSessionTree( rootSessionId: string, rootSessionFile: string, stillWanted?: () => boolean, exclude?: ResidentWorker, ): Promise<void> { const store = AgentCronJobStore.forSessionArtifacts(); const sessions = [{ sessionId: rootSessionId, sessionFile: rootSessionFile }]; const childrenByParent = new Map<string, SessionInfo[]>(); for (const info of await this.rlmSpawnLedger().family()) { if (!info.parentSessionPath) continue; const key = canonicalSessionPath(info.parentSessionPath); childrenByParent.set(key, [...(childrenByParent.get(key) ?? []), info]); } const queue = [canonicalSessionPath(rootSessionFile)]; const visited = new Set(queue); while (queue.length > 0) { for (const info of childrenByParent.get(queue.shift()!) ?? []) { const child = canonicalSessionPath(info.path); if (visited.has(child)) continue; visited.add(child); sessions.push({ sessionId: info.id, sessionFile: info.path }); queue.push(child); } } // A worker covering any tree member owns its stores again; a stale intent must not kill new schedules. for (const { sessionFile } of sessions) { try { if (this.findWorkerBySessionFile(sessionFile, exclude)) return; } catch { return; } } let registered = false; for (const { sessionId, sessionFile } of sessions) { const artifactDir = getSessionArtifactPathForFile(sessionFile, sessionId); if (!existsSync(join(artifactDir, SESSION_SCHEDULED_JOBS_FILENAME))) continue; store.registerSessionArtifact(sessionId, artifactDir); registered = true; } if (!registered) return; // Re-checked in the same synchronous turn as the walk: a promotion committed during the family read keeps its schedules. if (stillWanted && !stillWanted()) return; for (const { sessionFile } of sessions) { store.cancelJobsForSession({ sessionFile }); } }
private async cancelEphemeralWorkerScheduledJobs(worker: ResidentWorker): Promise<boolean> { const context = this.workerSessionArtifactContext(worker); if (!context || !worker.descriptor.rootSessionId) { return true; } try { await this.cancelScheduledJobsForSessionTree( worker.descriptor.rootSessionId, context.sessionFile, () => worker.descriptor.ownerClientId !== undefined, worker, ); return true; } catch (error) { // A promotion that landed during the failed read means the cancel is no longer wanted. if (worker.descriptor.ownerClientId === undefined) { return true; } this.log( `Could not cancel scheduled jobs for client-owned worker ${worker.descriptor.workerId}: ${String(error)}`, ); return false; } }
private reclaimStoppedWorkerCronLock(worker: ResidentWorker): void { const context = this.workerSessionArtifactContext(worker); if (!context) { return; } rmSync(join(context.artifactDir, `${SESSION_SCHEDULED_JOBS_FILENAME}.lock`), { recursive: true, force: true }); }
private workerSessionArtifactContext(worker: { descriptor: DaemonWorkerDescriptor; }): { sessionFile: string; artifactDir: string } | undefined { const sessionFile = worker.descriptor.sessionFile ?? worker.descriptor.createCommand.sessionPath; if (!sessionFile || !worker.descriptor.rootSessionId) { return undefined; } return { sessionFile, artifactDir: getSessionArtifactPathForFile(sessionFile, worker.descriptor.rootSessionId), }; }
private persistWorkerStopTombstone(worker: ResidentWorker, archiveSession = false): void { worker.intentionalStop = true; worker.descriptor.stopRequestedAt ??= new Date().toISOString(); worker.descriptor.archiveOnStop ||= archiveSession; this.persistWorker(worker); }
private write(client: DaemonSocketClient, message: DaemonOutbound): boolean { return this.writeSerialized(client, serializeJsonLine(message)); }
private broadcastHeartbeatsChanged(): void { this.scheduleScheduledSessionWakeRecompute(); for (const client of this.clients) { this.write(client, { type: "heartbeats_changed" }); } }
private writeSerialized(client: DaemonSocketClient, line: string | Uint8Array): boolean { if (client.socket.destroyed) { return false; } const accepted = client.socket.write(line); if (!accepted) { client.backpressured = true; } return accepted; }
private registerSignalHandlers(): void { const signals: NodeJS.Signals[] = ["SIGINT", "SIGTERM"]; if (process.platform !== "win32") { signals.push("SIGHUP"); } for (const signal of signals) { const handler = () => void this.shutdown(signal === "SIGINT" ? 130 : signal === "SIGHUP" ? 129 : 143, false); process.on(signal, handler); this.signalCleanupHandlers.push(() => process.off(signal, handler)); } const exitHandler = () => this.cleanupSocket(); process.on("exit", exitHandler); this.signalCleanupHandlers.push(() => process.off("exit", exitHandler)); }
private assertSocketLeaseHeld(): void { const compromise = this.socketLeaseCompromise ?? this.socketLease?.compromise; if (compromise) throw new Error(`Daemon socket lease was compromised: ${compromise.message}`); }
private assertSupervisorServing(): void { this.assertSocketLeaseHeld(); if (this.shuttingDown) { const error = new Error(`Daemon supervisor generation ${this.generation} is shutting down; retry the command`); Object.assign(error, { code: "supervisor_generation_stale" as const }); throw error; } }
private fenceSupervisorSocket(): void { try { this.server?.close(); } catch { // The server may already be closed by a concurrent shutdown. } for (const client of this.clients) { client.detachInput(); client.socket.destroy(); } }
private handleSocketLeaseCompromised(error: Error): void { if (this.socketLeaseCompromise) return; this.socketLeaseCompromise = error; this.shuttingDown = true; this.fenceSupervisorSocket(); const message = `Daemon socket lease was compromised; relinquishing supervisor ownership: ${error.message}`; try { this.log(message); } catch { console.error(message); } if (!this.startupComplete) return; void this.cleanupSupervisorResources().catch((cleanupError) => this.reportCleanupFailure("compromised daemon socket lease", cleanupError), ); }
private cleanupSocket(): void { if (!this.ownsSocketPath) { return; } this.ownsSocketPath = false; const identity = this.socketIdentity; this.socketIdentity = undefined; cleanupDaemonSocketPath(this.socketPath, identity, this.socketLease); }
private async cleanupSupervisorResources(): Promise<void> { if (this.cleanupPromise) { return this.cleanupPromise; } this.cleanupPromise = this.cleanupSupervisorResourcesOnce(); return this.cleanupPromise; }
private async cleanupSupervisorResourcesOnce(): Promise<void> { this.shuttingDown = true; this.clearIdleEvictionTimer(); this.clearScheduledWakeTimer(); this.clearRosterWatchdogTimer(); await this.idleEvictionSweep?.catch(() => undefined); for (const cleanup of this.signalCleanupHandlers.splice(0)) { await this.runCleanupStep("signal handler", cleanup); } const server = this.server; this.server = undefined; const serverClosed = new Promise<void>((resolveClose) => { if (!server?.listening) { resolveClose(); return; } try { server.close(() => resolveClose()); } catch (error) { this.reportCleanupFailure("daemon server", error); resolveClose(); } }); for (const client of this.clients) { client.attachedActiveSessionIds.clear(); await this.runCleanupStep(`daemon client input ${client.id}`, () => client.detachInput()); await this.runCleanupStep(`daemon client socket ${client.id}`, () => { client.socket.destroy(); }); } this.clients.clear(); for (const worker of this.workers.values()) { if (worker.ownerCleanupTimer) { clearTimeout(worker.ownerCleanupTimer); worker.ownerCleanupTimer = undefined; } await this.runCleanupStep(`worker client ${worker.descriptor.workerId}`, () => worker.client?.close()); worker.client = undefined; const transcripts = new Set(worker.transcriptCaches.values()); for (const generations of worker.snapshotGenerations?.values() ?? []) { for (const generation of generations.values()) { transcripts.add(generation.transcript); this.settleSnapshotDuplicateValidation( generation, new Error("Daemon supervisor stopped during snapshot transfer"), ); if (!generation.transcript.complete) { generation.transcript.markFailed(new Error("Daemon supervisor stopped during snapshot transfer")); } } } for (const transcript of transcripts) { await this.runCleanupStep(`worker transcript ${worker.descriptor.workerId}`, () => transcript.dispose()); } worker.transcriptCaches.clear(); worker.snapshotGenerations?.clear(); worker.snapshotCache.clear(); worker.snapshotLoads.clear(); } this.workers.clear(); this.openingWorkers.clear(); this.catalogOpeningWorkers.clear(); await this.runCleanupStep("daemon catalog", () => this.catalog.stop()); await this.runCleanupStep("session catalog store", async () => closeSessionCatalogStores()); await this.runCleanupStep("daemon server", () => serverClosed); await this.runCleanupStep("daemon socket", () => this.cleanupSocket()); await this.runCleanupStep("supervisor cache", () => { rmSync(this.snapshotCacheRoot, { recursive: true, force: true }); }); const lease = this.socketLease; this.socketLease = undefined; await this.runCleanupStep("daemon socket lock", async () => lease?.release()); const ownership = this.ownership; this.ownership = undefined; await this.runCleanupStep("daemon ownership", async () => ownership?.release()); }
private async runCleanupStep(label: string, action: () => void | Promise<void>): Promise<void> { try { await action(); } catch (error) { this.reportCleanupFailure(label, error); } }
private reportCleanupFailure(label: string, error: unknown): void { const message = `Failed to clean up ${label}: ${String(error)}`; try { this.log(message); } catch { console.error(message); } }
private async shutdown( exitCode: number, stopWorkers: boolean, relaunch = false, forceWorkers = false, closingReason?: DaemonClosingReason, ): Promise<never> { if (this.shuttingDown) { process.exit(exitCode); } this.shuttingDown = true; this.clearIdleEvictionTimer(); this.clearScheduledWakeTimer(); this.clearRosterWatchdogTimer(); await this.idleEvictionSweep?.catch(() => undefined); if (closingReason) { for (const client of this.clients) { this.write(client, { type: "daemon_closing", reason: closingReason }); } } for (const cleanup of this.signalCleanupHandlers) { cleanup(); } if (stopWorkers) { await Promise.all( [...this.workers.values()].map(async (worker) => { try { await this.stopWorker(worker, true, forceWorkers, true); } catch (error) { if (!(error instanceof WorkerStopTimeoutError)) { throw error; } this.log( `Worker ${worker.descriptor.workerId} remains tombstoned for recovery after shutdown: ${error.message}`, ); } }), ); if (!this.hasPersistedWorkerDescriptors()) { rmSync(this.supervisorConfigPath, { force: true }); } } else { for (const worker of this.workers.values()) { worker.intentionalStop = true; worker.client?.close(); worker.client = undefined; } } await this.catalog.stop(); for (const client of this.clients) { client.detachInput(); client.socket.end(); } await new Promise<void>((resolveClose) => this.server?.close(() => resolveClose()) ?? resolveClose()); await this.runCleanupStep("daemon socket", () => this.cleanupSocket()); await this.runCleanupStep("supervisor cache", () => { rmSync(this.snapshotCacheRoot, { recursive: true, force: true }); }); const lease = this.socketLease; this.socketLease = undefined; await this.runCleanupStep("daemon socket lock", async () => lease?.release()); const ownership = this.ownership; this.ownership = undefined; await this.runCleanupStep("daemon ownership", async () => ownership?.release()); if (relaunch) { const launch = createCliSubprocessLaunchSpec(["--mode", "daemon", "--daemon-socket", this.socketPath]); const environment = createCliSubprocessEnv(); delete environment[DAEMON_CATALOG_ROLE_ENV]; delete environment[DAEMON_WORKER_ROLE_ENV]; delete environment[DAEMON_WORKER_TOKEN_ENV]; delete environment[DAEMON_WORKER_ACTIVE_SESSION_ID_ENV]; delete environment[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV]; delete environment[DAEMON_WORKER_RECOVERY_JOURNAL_ENV]; delete environment[ORPHAN_PROCESS_JOURNAL_ENV]; delete environment[SESSION_LEASES_ENABLED_ENV]; delete environment[SESSION_LEASE_OWNER_ID_ENV]; const replacement = spawnHidden(launch.command, launch.args, { cwd: this.defaultSessionConfig.cwd ?? process.cwd(), detached: true, env: environment, stdio: "ignore", }); replacement.unref(); } process.exit(exitCode); }}