Something went wrong. Try again.
My version of https://github.com/vercel-labs/fx
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227422842294230423142324233423442354236423742384239424042414242424342444245424642474248424942504251425242534254425542564257425842594260426142624263426442654266426742684269427042714272427342744275427642774278427942804281428242834284428542864287428842894290429142924293429442954296429742984299430043014302430343044305430643074308430943104311431243134314431543164317431843194320432143224323432443254326432743284329433043314332433343344335433643374338433943404341434243434344434543464347434843494350435143524353435443554356435743584359436043614362436343644365436643674368436943704371437243734374437543764377437843794380438143824383438443854386438743884389439043914392439343944395439643974398439944004401440244034404440544064407440844094410441144124413441444154416441744184419442044214422442344244425442644274428442944304431443244334434443544364437443844394440444144424443444444454446444744484449445044514452445344544455445644574458445944604461446244634464446544664467446844694470447144724473447444754476447744784479448044814482448344844485448644874488448944904491449244934494449544964497449844994500450145024503450445054506450745084509451045114512451345144515451645174518451945204521452245234524452545264527452845294530453145324533453445354536453745384539454045414542454345444545454645474548454945504551455245534554455545564557455845594560456145624563456445654566456745684569457045714572457345744575457645774578457945804581458245834584458545864587458845894590459145924593459445954596459745984599460046014602460346044605460646074608460946104611461246134614461546164617461846194620462146224623462446254626462746284629463046314632463346344635463646374638463946404641464246434644464546464647464846494650465146524653465446554656465746584659466046614662466346644665466646674668466946704671467246734674467546764677467846794680468146824683468446854686468746884689469046914692469346944695469646974698469947004701470247034704470547064707470847094710471147124713471447154716471747184719472047214722472347244725472647274728472947304731473247334734473547364737473847394740474147424743474447454746474747484749475047514752475347544755475647574758475947604761476247634764476547664767476847694770477147724773477447754776477747784779478047814782478347844785478647874788478947904791479247934794479547964797479847994800480148024803480448054806480748084809481048114812481348144815481648174818481948204821482248234824482548264827482848294830483148324833483448354836483748384839484048414842484348444845484648474848484948504851485248534854485548564857485848594860486148624863486448654866486748684869487048714872487348744875487648774878487948804881488248834884488548864887488848894890489148924893489448954896489748984899490049014902490349044905490649074908490949104911491249134914491549164917491849194920492149224923492449254926492749284929493049314932493349344935493649374938493949404941494249434944494549464947494849494950495149524953495449554956495749584959496049614962496349644965496649674968496949704971497249734974497549764977497849794980498149824983498449854986498749884989499049914992499349944995499649974998499950005001500250035004500550065007500850095010501150125013501450155016501750185019502050215022502350245025502650275028502950305031503250335034503550365037503850395040504150425043504450455046504750485049505050515052505350545055505650575058505950605061506250635064506550665067506850695070507150725073507450755076507750785079508050815082508350845085508650875088508950905091509250935094509550965097509850995100510151025103510451055106510751085109511051115112511351145115511651175118511951205121512251235124512551265127512851295130513151325133513451355136513751385139514051415142514351445145514651475148514951505151515251535154515551565157515851595160516151625163516451655166516751685169517051715172517351745175517651775178517951805181518251835184518551865187518851895190519151925193519451955196519751985199520052015202520352045205520652075208520952105211521252135214521552165217521852195220522152225223522452255226522752285229523052315232523352345235523652375238523952405241524252435244524552465247524852495250525152525253525452555256525752585259526052615262526352645265526652675268526952705271527252735274527552765277527852795280528152825283528452855286528752885289529052915292529352945295529652975298529953005301530253035304530553065307530853095310531153125313531453155316531753185319532053215322532353245325532653275328532953305331533253335334533553365337533853395340534153425343534453455346534753485349535053515352535353545355535653575358535953605361536253635364536553665367536853695370537153725373537453755376537753785379538053815382538353845385538653875388538953905391539253935394539553965397539853995400540154025403540454055406540754085409541054115412541354145415541654175418541954205421542254235424542554265427542854295430543154325433543454355436543754385439544054415442544354445445544654475448544954505451545254535454545554565457545854595460546154625463546454655466546754685469547054715472547354745475547654775478547954805481548254835484548554865487548854895490549154925493549454955496549754985499550055015502550355045505550655075508550955105511551255135514551555165517551855195520552155225523552455255526552755285529553055315532553355345535553655375538553955405541554255435544554555465547554855495550555155525553555455555556555755585559556055615562556355645565556655675568556955705571557255735574557555765577557855795580558155825583558455855586558755885589559055915592559355945595559655975598559956005601560256035604560556065607560856095610561156125613561456155616561756185619562056215622562356245625562656275628562956305631563256335634563556365637563856395640564156425643564456455646564756485649565056515652565356545655565656575658565956605661566256635664566556665667566856695670567156725673567456755676567756785679568056815682568356845685568656875688568956905691569256935694569556965697569856995700570157025703570457055706570757085709571057115712571357145715571657175718571957205721572257235724572557265727572857295730573157325733573457355736573757385739574057415742574357445745574657475748574957505751575257535754575557565757575857595760576157625763576457655766576757685769577057715772577357745775577657775778577957805781578257835784578557865787578857895790579157925793579457955796579757985799580058015802580358045805580658075808580958105811581258135814581558165817581858195820582158225823582458255826582758285829583058315832583358345835583658375838583958405841584258435844584558465847584858495850585158525853585458555856585758585859586058615862586358645865586658675868586958705871587258735874587558765877587858795880588158825883588458855886588758885889589058915892589358945895589658975898589959005901590259035904590559065907590859095910591159125913591459155916591759185919592059215922592359245925592659275928592959305931593259335934593559365937593859395940594159425943594459455946594759485949595059515952595359545955595659575958595959605961596259635964596559665967596859695970597159725973597459755976597759785979598059815982598359845985598659875988598959905991599259935994599559965997599859996000600160026003600460056006600760086009601060116012601360146015601660176018601960206021602260236024602560266027602860296030603160326033603460356036603760386039604060416042604360446045604660476048604960506051605260536054605560566057605860596060606160626063606460656066606760686069607060716072607360746075607660776078607960806081608260836084608560866087608860896090609160926093609460956096609760986099610061016102610361046105610661076108610961106111611261136114611561166117611861196120612161226123612461256126612761286129613061316132613361346135613661376138613961406141614261436144614561466147614861496150615161526153615461556156615761586159616061616162616361646165616661676168616961706171617261736174617561766177617861796180618161826183618461856186618761886189619061916192619361946195619661976198619962006201620262036204620562066207620862096210621162126213621462156216621762186219622062216222622362246225622662276228622962306231623262336234623562366237623862396240624162426243624462456246624762486249625062516252625362546255625662576258625962606261626262636264626562666267626862696270627162726273627462756276627762786279628062816282628362846285628662876288628962906291629262936294629562966297629862996300630163026303630463056306630763086309631063116312631363146315631663176318631963206321632263236324632563266327632863296330633163326333633463356336633763386339634063416342634363446345634663476348634963506351635263536354635563566357635863596360636163626363636463656366636763686369637063716372637363746375637663776378637963806381638263836384638563866387638863896390639163926393639463956396639763986399640064016402640364046405640664076408640964106411641264136414641564166417641864196420642164226423642464256426642764286429643064316432643364346435643664376438643964406441644264436444644564466447644864496450645164526453645464556456645764586459646064616462646364646465646664676468646964706471647264736474647564766477647864796480648164826483648464856486648764886489649064916492649364946495649664976498649965006501650265036504650565066507650865096510651165126513651465156516651765186519652065216522652365246525652665276528652965306531653265336534653565366537653865396540654165426543654465456546654765486549655065516552655365546555655665576558655965606561656265636564656565666567656865696570657165726573657465756576657765786579658065816582658365846585658665876588658965906591659265936594659565966597659865996600660166026603660466056606660766086609661066116612661366146615661666176618661966206621662266236624662566266627662866296630663166326633663466356636663766386639664066416642664366446645664666476648664966506651665266536654665566566657665866596660666166626663666466656666666766686669667066716672667366746675667666776678667966806681668266836684668566866687668866896690669166926693669466956696669766986699670067016702670367046705670667076708670967106711671267136714671567166717671867196720672167226723672467256726672767286729673067316732673367346735673667376738673967406741674267436744674567466747674867496750675167526753675467556756675767586759676067616762676367646765676667676768676967706771677267736774677567766777677867796780678167826783678467856786678767886789679067916792679367946795679667976798679968006801680268036804680568066807680868096810681168126813681468156816681768186819682068216822682368246825682668276828682968306831683268336834683568366837683868396840684168426843684468456846684768486849685068516852685368546855685668576858685968606861686268636864686568666867686868696870687168726873687468756876687768786879688068816882688368846885688668876888688968906891689268936894689568966897689868996900690169026903690469056906690769086909691069116912691369146915691669176918691969206921692269236924692569266927692869296930693169326933693469356936693769386939694069416942694369446945694669476948694969506951695269536954695569566957695869596960696169626963696469656966696769686969697069716972697369746975697669776978697969806981698269836984698569866987698869896990699169926993699469956996699769986999700070017002700370047005700670077008700970107011701270137014701570167017701870197020702170227023702470257026702770287029703070317032703370347035703670377038703970407041704270437044704570467047704870497050705170527053705470557056705770587059706070617062706370647065706670677068706970707071707270737074707570767077707870797080708170827083708470857086708770887089709070917092709370947095709670977098709971007101710271037104710571067107710871097110711171127113711471157116711771187119712071217122712371247125712671277128712971307131713271337134713571367137713871397140714171427143714471457146714771487149715071517152715371547155715671577158715971607161716271637164716571667167716871697170717171727173717471757176717771787179718071817182718371847185718671877188718971907191719271937194719571967197719871997200720172027203720472057206720772087209721072117212721372147215721672177218721972207221const std = @import("std");const builtin = @import("builtin");const debug_trace = @import("../shared/debug_trace.zig");const io_mod = @import("../shared/io.zig");const mem_utils = @import("../shared/mem_utils.zig");const profile_paths = @import("../shared/profile_paths.zig");const model_provider = @import("../config/model_provider.zig");const session = @import("session.zig");const session_child_store = @import("session_child_store.zig");const session_codec = @import("session_codec.zig");const types = @import("../shared/types.zig");const session_event = @import("session_event.zig");const history_snapshot = @import("history_snapshot.zig");const session_layout = @import("session_layout.zig");const session_replay = @import("session_replay.zig");const session_display_metadata = @import("session_display_metadata.zig");const result_store = @import("result_store.zig");const session_usage = @import("session_usage.zig");const session_usage_sidecar = @import("session_usage_sidecar.zig");const session_permission_state = @import("../permissions/session_permission_state.zig");
const Allocator = std.mem.Allocator;const Identifier = session_event.Identifier;const private_dir_permissions = std.Io.File.Permissions.fromMode(0o700);const private_file_permissions = std.Io.File.Permissions.fromMode(0o600);const lock_deadline_ms: u64 = 2000;const events_file = "events.jsonl";const authority_file = "authority.json";const authority_intent_file = "authority.pending.json";const publication_intent_file = "commit.pending.json";const manifest_file = "session.json";const permission_state_file = "permissions.json";const recovery_checkpoint_file = "recovery.json";const conversation_migration_temp_file = "events.v4.tmp";const conversation_migration_backup_file = "events.v3.backup";const checkpoint_file = "checkpoint.json";const session_lock_file = "session.lock";const commit_lock_file = "commit.lock";
const ConversationProgress = struct { point: types.ContextHistoryCut = .{}, pending: usize = 0, coverage: u64 = 0, reached: bool = false, pending_assistant: ?struct { seq: u64, has_replay: bool } = null,
fn observe(self: *ConversationProgress, seq: u64, event: session_event.ConversationEvent, cut: ?types.ContextHistoryCut) !void { if (self.pending_assistant) |assistant| { const standalone = switch (event) { .assistant, .context_checkpoint, .interrupted => true, .steering => assistant.has_replay, else => false, }; if (standalone) { self.point.tool_steps += 1; if (cut) |target| if (!self.reached and std.meta.eql(self.point, target)) { self.coverage = assistant.seq; self.reached = true; }; } self.pending_assistant = null; } if (cut) |target| if (std.meta.eql(self.point, target) and self.pending == 0) { self.reached = true; }; switch (event) { .assistant => |value| { if (value.standalone_response) { self.point.tool_steps += 1; } else if (value.text.len > 0 or value.provider_replay != null) { self.pending_assistant = .{ .seq = seq, .has_replay = value.provider_replay != null }; } }, .tool_call => self.pending += 1, .tool_result => { if (self.pending == 0) return error.InvalidConversationFrame; self.pending -= 1; if (self.pending == 0) self.point.tool_steps += 1; }, .steering => self.point.steering += 1, .turn_completed, .interrupted => { self.point.turns += 1; self.point.tool_steps = 0; self.point.steering = 0; self.pending = 0; }, else => {}, } if (cut) |target| { if (!self.reached and std.meta.eql(self.point, target) and self.pending == 0) { self.coverage = seq; self.reached = true; } } }};
test "conversation cut counts standalone text without changing steering prefixes" { var progress: ConversationProgress = .{}; const cut: types.ContextHistoryCut = .{ .tool_steps = 1 }; try progress.observe(1, .{ .user = .{ .text = "request" } }, cut); try progress.observe(2, .{ .assistant = .{ .text = "CANDIDATE_741" } }, cut); try progress.observe(3, .{ .assistant = .{ .text = "FINAL_852" } }, cut); try std.testing.expectEqual(@as(usize, 1), progress.point.tool_steps); try std.testing.expectEqual(@as(u64, 2), progress.coverage); try std.testing.expect(progress.reached);
var steering: ConversationProgress = .{}; try steering.observe(1, .{ .user = .{ .text = "request" } }, null); try steering.observe(2, .{ .assistant = .{ .text = "prefix" } }, null); try steering.observe(3, .{ .steering = .{ .text = "human update" } }, null); try std.testing.expectEqual(@as(usize, 0), steering.point.tool_steps); try std.testing.expectEqual(@as(usize, 1), steering.point.steering);}
pub const ConversationWriter = struct { alloc: Allocator, file: std.Io.File, committed_bytes: u64 = 0, last_seq: u64 = 0, latest_checkpoint_coverage: u64 = 0, turn_open: bool = false, pending_tool_calls: std.ArrayList(session_event.PendingToolCall) = .empty, failure: ?error{ SessionWriterChanged, SessionPersistenceUncertain, SessionCommitFailed } = null, test_sync_ops: if (builtin.is_test) io_mod.DurableOps else void = if (builtin.is_test) .{} else {},
/// Takes ownership of `file` only on success. pub fn init(alloc: Allocator, file: std.Io.File) !ConversationWriter { return initWithReplayScan(alloc, file, null, null, null); }
fn initWithReplayScan( alloc: Allocator, file: std.Io.File, replay_scan: ?*ConversationReplayScan, source: ?*HistoryFrameSource, snapshot_tee: ?*history_snapshot.Writer, ) !ConversationWriter { const length = try file.length(io_mod.getIo()); var writer = ConversationWriter{ .alloc = alloc, .file = file }; errdefer { writer.clearPendingToolCalls(); writer.pending_tool_calls.deinit(alloc); } var log_source = HistoryFrameSource{ .log_file = file, .log_length = length }; const frames = source orelse &log_source; if (source == null) frames.reset(); var frame_arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(frame_arena); var offset: u64 = 0; var open_turn_offset: ?u64 = null; var open_turn_prior_seq: u64 = 0; var open_turn_snapshot_len: ?u64 = null; var checkpointed_turn = false; while (true) { const frame = frames.next(frame_arena.allocator()) catch |err| switch (err) { error.TruncatedEventFrame => { try file.setLength(io_mod.getIo(), offset); try file.sync(io_mod.getIo()); writer.committed_bytes = offset; break; }, else => return err, } orelse break; const envelope = frame.envelope; try session_event.validateConversationTransition(.{ .last_seq = writer.last_seq, .latest_checkpoint_coverage = writer.latest_checkpoint_coverage, .pending_tool_calls = writer.pending_tool_calls.items, }, envelope); switch (envelope.event) { .user => { if (open_turn_offset != null) return error.InvalidConversationFrame; open_turn_offset = frame.log_offset; open_turn_prior_seq = writer.last_seq; // Cache position of this open turn's first frame: the // truncation mirror for an unfinished tail turn. Frames // served from a verified prefix point at their own offset; // frames teed now point at the writer's pre-append length. open_turn_snapshot_len = frames.last_frame_snapshot_file_offset orelse if (snapshot_tee) |tee| tee.len else null; }, .turn_completed, .interrupted => { if (open_turn_offset == null) return error.InvalidConversationFrame; open_turn_offset = null; open_turn_snapshot_len = null; checkpointed_turn = false; }, .context_checkpoint => if (open_turn_offset != null) { open_turn_offset = frame.nextOffset(); open_turn_prior_seq = envelope.seq; checkpointed_turn = true; }, .assistant, .tool_call, .tool_result, .steering => if (open_turn_offset == null) { return error.InvalidConversationFrame; }, } try writer.applyReplayedEvent(envelope.seq, envelope.event); if (replay_scan) |scan| try scan.observe(frame.log_offset, envelope.seq, envelope.event); // Tee only frames read from the log. Cache-sourced frames are // already in the cache; re-appending them would duplicate it. if (snapshot_tee) |tee| { if (frames.last_frame_snapshot_file_offset == null) tee.append(envelope, frame.log_offset, frame.log_bytes, frame.line_crc); } offset = frame.nextOffset(); writer.committed_bytes = offset; _ = frame_arena.reset(.retain_capacity); } if (open_turn_offset) |truncate_from| { debug_trace.logf( "session", "event=conversation_unfinished_turn_discarded offset={d} bytes={d}", .{ truncate_from, writer.committed_bytes - truncate_from }, ); try file.setLength(io_mod.getIo(), truncate_from); try file.sync(io_mod.getIo()); writer.clearPendingToolCalls(); writer.committed_bytes = truncate_from; writer.last_seq = open_turn_prior_seq; writer.turn_open = checkpointed_turn; if (replay_scan) |scan| { scan.last_seq = writer.last_seq; if (!writer.turn_open) scan.active_user_offset = null; } // The open turn's frames were already mirrored into the cache; // truncate the cache to the same logical point. if (snapshot_tee) |tee| { if (open_turn_snapshot_len) |snapshot_len| tee.truncateTo(snapshot_len); } } return writer; }
pub fn deinit(self: *ConversationWriter) void { for (self.pending_tool_calls.items) |pending| { self.alloc.free(@constCast(pending.call_id)); self.alloc.free(@constCast(pending.tool_name)); } self.pending_tool_calls.deinit(self.alloc); self.file.close(io_mod.getIo()); self.* = undefined; }
pub fn append( self: *ConversationWriter, alloc: Allocator, timestamp_ms: i64, event: session_event.ConversationEvent, ) !u64 { if (self.failure) |err| return err; const seq = std.math.add(u64, self.last_seq, 1) catch return error.ConversationSequenceOverflow; const envelope = session_event.ConversationEnvelope{ .seq = seq, .timestamp_ms = timestamp_ms, .event = event, }; try session_event.validateConversationTransition(.{ .last_seq = self.last_seq, .latest_checkpoint_coverage = self.latest_checkpoint_coverage, .pending_tool_calls = self.pending_tool_calls.items, }, envelope); const turn_open = try nextConversationTurnOpen(self.turn_open, event);
var owned_pending: ?session_event.PendingToolCall = null; errdefer if (owned_pending) |pending| { self.alloc.free(@constCast(pending.call_id)); self.alloc.free(@constCast(pending.tool_name)); }; const completed_index: ?usize = switch (event) { .tool_call => |call| blk: { try self.pending_tool_calls.ensureUnusedCapacity(self.alloc, 1); const call_id = try self.alloc.dupe(u8, call.call_id); errdefer self.alloc.free(call_id); const tool_name = try self.alloc.dupe(u8, call.tool_name); owned_pending = .{ .call_id = call_id, .tool_name = tool_name, .seq = seq, }; break :blk null; }, .tool_result => |result| self.pendingIndex(result.call_id), else => null, };
const frame = try session_event.encodeConversationFrame(alloc, envelope); defer alloc.free(frame); try self.writePrepared(frame); self.last_seq = seq; self.turn_open = turn_open;
if (owned_pending) |pending| { self.pending_tool_calls.appendAssumeCapacity(pending); owned_pending = null; } if (completed_index) |index| { const completed = self.pending_tool_calls.orderedRemove(index); self.alloc.free(@constCast(completed.call_id)); self.alloc.free(@constCast(completed.tool_name)); } if (event == .interrupted) self.clearPendingToolCalls(); if (event == .context_checkpoint) { self.latest_checkpoint_coverage = event.context_checkpoint.covers_through_seq; } return seq; }
pub fn pendingToolCallCount(self: *const ConversationWriter) usize { return self.pending_tool_calls.items.len; }
pub fn appendHistoryTurn( self: *ConversationWriter, alloc: Allocator, timestamp_ms: i64, turn: types.HistoryTurn, ) !void { if (turn == .compacted_summary) { _ = try self.append(alloc, timestamp_ms, .{ .context_checkpoint = .{ .covers_through_seq = self.last_seq, .summary = turn.compacted_summary.summary, }, }); return; }
try self.appendTurnBatch(alloc, timestamp_ms, turn, null); }
fn appendContextCompaction( self: *ConversationWriter, alloc: Allocator, timestamp_ms: i64, summary: types.CompactedSummaryHistoryTurn, prefix: ?types.AssistantHistoryTurn, retained_from: ?types.ContextHistoryCut, ) !void { if (prefix) |entry| { if (entry.assistant.len != 0) return error.InvalidConversationEvent; try self.appendTurnBatchWithCut(alloc, timestamp_ms, .{ .assistant = entry }, summary.summary, retained_from); } else { const coverage = if (retained_from) |cut| try self.contextCoverage(alloc, cut, &.{}) else self.last_seq; _ = try self.append(alloc, timestamp_ms, .{ .context_checkpoint = .{ .covers_through_seq = coverage, .summary = summary.summary, } }); } }
fn appendTurnBatch( self: *ConversationWriter, alloc: Allocator, timestamp_ms: i64, turn: types.HistoryTurn, checkpoint_summary: ?[]const u8, ) !void { return self.appendTurnBatchWithCut(alloc, timestamp_ms, turn, checkpoint_summary, null); }
fn appendTurnBatchWithCut( self: *ConversationWriter, alloc: Allocator, timestamp_ms: i64, turn: types.HistoryTurn, checkpoint_summary: ?[]const u8, retained_from: ?types.ContextHistoryCut, ) !void { var arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(arena); // A checkpoint can have already committed the recent execution prefix. // Finalization supplies that prefix for model history; append only its // not-yet-written suffix to the conversation log. const unwritten = if (self.turn_open) blk: { var progress: ConversationProgress = .{}; try self.scanContext(alloc, &progress, null); const view = try session.contextHistoryRange(arena.allocator(), &.{turn}, .{ .tool_steps = progress.point.tool_steps, .steering = progress.point.steering, }, null); if (view.len != 1) return error.InvalidConversationEvent; break :blk view[0]; } else turn; var events: std.ArrayList(session_event.ConversationEvent) = .empty; defer events.deinit(alloc); try session_event.appendHistoryTurnConversationEvents(alloc, &events, unwritten); const start: usize = if (self.turn_open) 1 else 0; if (checkpoint_summary) |summary| { _ = events.pop(); const covers_through_seq = if (retained_from) |cut| try self.contextCoverage(alloc, cut, events.items[start..]) else std.math.add(u64, self.last_seq, events.items.len - start) catch return error.ConversationSequenceOverflow; try events.append(alloc, .{ .context_checkpoint = .{ .covers_through_seq = covers_through_seq, .summary = summary, } }); } var projected_images: ?[]types.ImageAttachment = null; defer if (projected_images) |images| { types.freeImageAttachmentSlice(alloc, images); }; for (events.items[start..]) |*event| switch (event.*) { .user => |*user| { if (user.images.len == 0) continue; const images = try types.dupeImageAttachmentSlice(alloc, user.images); projectConversationImageLocators(alloc, images) catch |err| { types.freeImageAttachmentSlice(alloc, images); return err; }; user.images = images; projected_images = images; }, else => {}, }; try self.appendBatch(alloc, timestamp_ms, events.items[start..]); }
fn scanContext(self: *const ConversationWriter, alloc: Allocator, progress: *ConversationProgress, cut: ?types.ContextHistoryCut) !void { progress.coverage = self.latest_checkpoint_coverage; var offset: u64 = 0; var buffer: [8192]u8 = undefined; var reader = self.file.reader(io_mod.getIo(), &buffer); while (offset < self.committed_bytes) { const line = try session_replay.readBufferedLine(alloc, &reader, self.committed_bytes, null) orelse break; defer alloc.free(line.bytes); var decoded = try session_event.decodeConversationFrame(alloc, line.bytes); defer decoded.deinit(); if (decoded.value.seq > self.latest_checkpoint_coverage) { try progress.observe(decoded.value.seq, decoded.value.event, cut); } offset = line.next_offset; } }
fn contextCoverage(self: *const ConversationWriter, alloc: Allocator, cut: types.ContextHistoryCut, upcoming: []const session_event.ConversationEvent) !u64 { var progress: ConversationProgress = .{}; try self.scanContext(alloc, &progress, cut); for (upcoming, 0..) |event, index| { const seq = std.math.add(u64, self.last_seq, index + 1) catch return error.ConversationSequenceOverflow; try progress.observe(seq, event, cut); } if (!progress.reached and !std.meta.eql(cut, types.ContextHistoryCut{})) return error.InvalidContextHistoryStart; return progress.coverage; }
pub fn readAllForTest(self: *const ConversationWriter, alloc: Allocator) ![]u8 { const size = std.math.cast(usize, self.committed_bytes) orelse return error.ConversationTooLarge; const bytes = try alloc.alloc(u8, size); errdefer alloc.free(bytes); const count = try self.file.readPositionalAll(io_mod.getIo(), bytes, 0); if (count != size) return error.TruncatedConversation; return bytes; }
fn pendingIndex(self: *const ConversationWriter, call_id: []const u8) ?usize { for (self.pending_tool_calls.items, 0..) |pending, index| { if (std.mem.eql(u8, pending.call_id, call_id)) return index; } return null; }
fn appendBatch( self: *ConversationWriter, alloc: Allocator, timestamp_ms: i64, events: []const session_event.ConversationEvent, ) !void { if (self.failure) |err| return err; if (events.len == 0) return; if (self.pending_tool_calls.items.len != 0) return error.UnresolvedToolCall;
var pending: std.ArrayList(session_event.PendingToolCall) = .empty; defer pending.deinit(alloc); var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); var seq = self.last_seq; var checkpoint_coverage = self.latest_checkpoint_coverage; var turn_open = self.turn_open; for (events) |event| { seq = std.math.add(u64, seq, 1) catch return error.ConversationSequenceOverflow; const envelope = session_event.ConversationEnvelope{ .seq = seq, .timestamp_ms = timestamp_ms, .event = event, }; try session_event.validateConversationTransition(.{ .last_seq = seq - 1, .latest_checkpoint_coverage = checkpoint_coverage, .pending_tool_calls = pending.items, }, envelope); turn_open = try nextConversationTurnOpen(turn_open, event); switch (event) { .tool_call => |call| try pending.append(alloc, .{ .call_id = call.call_id, .tool_name = call.tool_name, .seq = seq, }), .tool_result => |result| { for (pending.items, 0..) |call, index| { if (std.mem.eql(u8, call.call_id, result.call_id)) { _ = pending.orderedRemove(index); break; } } }, .interrupted => pending.clearRetainingCapacity(), .context_checkpoint => |checkpoint| checkpoint_coverage = checkpoint.covers_through_seq, else => {}, } const frame = try session_event.encodeConversationFrame(alloc, envelope); defer alloc.free(frame); try out.writer.writeAll(frame); } if (pending.items.len != 0) return error.UnresolvedToolCall; try self.writePrepared(out.written()); self.last_seq = seq; self.latest_checkpoint_coverage = checkpoint_coverage; self.turn_open = turn_open; }
fn writePrepared(self: *ConversationWriter, bytes: []const u8) !void { if (self.failure) |err| return err; const next_bytes = std.math.add(u64, self.committed_bytes, bytes.len) catch return error.ConversationSizeOverflow; if (try self.file.length(io_mod.getIo()) != self.committed_bytes) { self.failure = error.SessionWriterChanged; debug_trace.logf("session", "conversation writer changed outside ownership offset={d}", .{self.committed_bytes}); return error.SessionWriterChanged; } self.file.writePositionalAll(io_mod.getIo(), bytes, self.committed_bytes) catch |err| { return self.rollbackAppend(err); }; self.syncAppend() catch |err| return self.rollbackAppend(err); self.committed_bytes = next_bytes; }
fn syncAppend(self: *ConversationWriter) !void { if (comptime builtin.is_test) { return self.test_sync_ops.sync_file(self.test_sync_ops.ctx, self.file); } try self.file.sync(io_mod.getIo()); }
fn rollbackAppend(self: *ConversationWriter, original: anyerror) anyerror { self.failure = error.SessionPersistenceUncertain; self.file.setLength(io_mod.getIo(), self.committed_bytes) catch |err| { debug_trace.logf("session", "conversation append rollback failed offset={d} append_err={s} rollback_err={s}", .{ self.committed_bytes, @errorName(original), @errorName(err) }); return error.SessionPersistenceUncertain; }; self.syncAppend() catch |err| { debug_trace.logf("session", "conversation append rollback sync failed offset={d} append_err={s} rollback_err={s}", .{ self.committed_bytes, @errorName(original), @errorName(err) }); return error.SessionPersistenceUncertain; }; self.failure = null; debug_trace.logf("session", "conversation append rolled back offset={d} err={s}", .{ self.committed_bytes, @errorName(original) }); return original; }
fn clearPendingToolCalls(self: *ConversationWriter) void { for (self.pending_tool_calls.items) |pending| { self.alloc.free(@constCast(pending.call_id)); self.alloc.free(@constCast(pending.tool_name)); } self.pending_tool_calls.clearRetainingCapacity(); }
fn applyReplayedEvent( self: *ConversationWriter, seq: u64, event: session_event.ConversationEvent, ) !void { const turn_open = try nextConversationTurnOpen(self.turn_open, event); switch (event) { .tool_call => |call| { const call_id = try self.alloc.dupe(u8, call.call_id); errdefer self.alloc.free(call_id); const tool_name = try self.alloc.dupe(u8, call.tool_name); errdefer self.alloc.free(tool_name); try self.pending_tool_calls.append(self.alloc, .{ .call_id = call_id, .tool_name = tool_name, .seq = seq, }); }, .tool_result => |result| { const index = self.pendingIndex(result.call_id) orelse return error.OrphanToolResult; const completed = self.pending_tool_calls.orderedRemove(index); self.alloc.free(@constCast(completed.call_id)); self.alloc.free(@constCast(completed.tool_name)); }, .context_checkpoint => |checkpoint| { self.latest_checkpoint_coverage = checkpoint.covers_through_seq; }, .interrupted => self.clearPendingToolCalls(), .user, .assistant, .steering, .turn_completed => {}, } self.last_seq = seq; self.turn_open = turn_open; }};
fn nextConversationTurnOpen(open: bool, event: session_event.ConversationEvent) !bool { return switch (event) { .user => if (open) error.InvalidConversationFrame else true, .turn_completed, .interrupted => if (open) false else error.InvalidConversationFrame, .assistant, .tool_call, .tool_result, .steering => if (open) true else error.InvalidConversationFrame, .context_checkpoint => open, };}
fn createConversationStorage( alloc: Allocator, dir: *io_mod.VerifiedDir, metadata: session_codec.SessionMetadata,) !ConversationWriter { const metadata_bytes = try session_codec.encodeSessionMetadata(alloc, metadata); defer alloc.free(metadata_bytes); try io_mod.durableReplaceVerified(alloc, dir, manifest_file, metadata_bytes); errdefer { dir.dir.deleteFile(io_mod.getIo(), manifest_file) catch {}; io_mod.syncVerifiedDir(dir.dir) catch {}; }
const file = createManagedFile(dir, events_file) catch |err| switch (err) { error.PathAlreadyExists => return error.SessionAlreadyExists, else => return err, }; errdefer file.close(io_mod.getIo()); // Fresh sessions keep the minimal durable footprint; the replay cache is // built lazily by the first writable resume (see openConversationWritableSession). return ConversationWriter.init(alloc, file);}
fn writeConversationMetadata( alloc: Allocator, dir: *io_mod.VerifiedDir, state: session_codec.DurableSessionState,) !void { const existing = readManagedFileAlloc( alloc, dir, manifest_file, session_codec.max_session_metadata_bytes, ) catch |err| switch (err) { error.FileNotFound => null, else => return err, }; defer if (existing) |bytes| alloc.free(bytes); var decoded = if (existing) |bytes| try session_codec.decodeSessionMetadata(alloc, bytes) else null; defer if (decoded) |*metadata| metadata.deinit(); const bytes = try encodeConversationMetadataWithTitle( alloc, state, if (decoded) |metadata| metadata.value.title else null, ); defer alloc.free(bytes); try io_mod.durableReplaceVerified(alloc, dir, manifest_file, bytes);}
fn encodeConversationMetadataWithTitle( alloc: Allocator, state: session_codec.DurableSessionState, title: ?[]const u8,) ![]u8 { return session_codec.encodeSessionMetadata(alloc, .{ .id = state.id, .origin_workspace_root = state.origin_workspace_root, .workspace_root = state.workspace_root, .created_at_ms = state.created_at_ms, .updated_at_ms = state.updated_at_ms, .conversation_language = state.conversation_language.view(), .provider = state.preferences.provider, .model = state.preferences.model, .effort = state.preferences.effort.label(), .fast_mode = state.preferences.fast_mode, .title = title, .subagent_child = state.subagent_child, });}
fn writeConversationControlState( alloc: Allocator, dir: *io_mod.VerifiedDir, state: session_codec.DurableSessionState, conversation_seq: u64,) !void { const permission_bytes = try session_codec.encodePermissionState( alloc, state.permission_state, ); defer alloc.free(permission_bytes); try io_mod.durableReplaceVerified( alloc, dir, permission_state_file, permission_bytes, ); if (state.usage) |usage| { try session_usage_sidecar.write(alloc, dir, state.id, usage); } try writeConversationRecoveryState(alloc, dir, state.recovery_checkpoint, conversation_seq);}
fn writeConversationRecoveryState( alloc: Allocator, dir: *io_mod.VerifiedDir, recovery_checkpoint: ?session_codec.RecoveryCheckpoint, conversation_seq: u64,) !void { if (recovery_checkpoint) |checkpoint| { var projection_arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(projection_arena); const projected = try spillRecoveryCheckpointOutputs( projection_arena.allocator(), dir, checkpoint, ); const recovery_bytes = session_codec.encodeRecoveryCheckpoint( alloc, projected, ) catch |err| switch (err) { // The durable checkpoint is a resume aid, not turn-critical // state. An oversized checkpoint keeps the previously persisted // file and the turn continues; in-memory state still advances. error.RecoveryCheckpointTooLarge => { debug_trace.logf( "session", "event=recovery_checkpoint_oversized cap_bytes={d}; keeping previous durable checkpoint", .{session_codec.max_recovery_checkpoint_bytes}, ); return; }, else => return err, }; defer alloc.free(recovery_bytes); const bound_bytes = try std.fmt.allocPrint( alloc, "{{\"conversation_seq\":{d},\"checkpoint\":{s}}}\n", .{ conversation_seq, recovery_bytes }, ); defer alloc.free(bound_bytes); try io_mod.durableReplaceVerified( alloc, dir, recovery_checkpoint_file, bound_bytes, ); } else { dir.dir.deleteFile(io_mod.getIo(), recovery_checkpoint_file) catch |err| switch (err) { error.FileNotFound => {}, else => return err, }; try io_mod.syncVerifiedDir(dir.dir); }}
/// Inline tool-result output budget for the durable recovery checkpoint./// Larger outputs live in the session's result store and the checkpoint/// carries their content-addressed handle instead.const recovery_checkpoint_inline_output_max_bytes: usize = result_store.preview_bytes;
/// Returns a copy of the checkpoint whose oversized tool-result outputs are/// spilled to the session result store and replaced by their handle. The/// source checkpoint is borrowed; the projection owns only its own/// allocations, and `alloc` is expected to free them in bulk. Spill failures/// keep the offending result inline so a store hiccup cannot block the/// checkpoint write.fn spillRecoveryCheckpointOutputs( alloc: Allocator, dir: *io_mod.VerifiedDir, checkpoint: session_codec.RecoveryCheckpoint,) !session_codec.RecoveryCheckpoint { var spills = false; for (checkpoint.execution.tool_steps) |step| { for (step.tool_results) |result| { if (result.output_handle != null or result.output.len > recovery_checkpoint_inline_output_max_bytes) { spills = true; break; } } } if (!spills) return checkpoint;
var result_dir: ?[]const u8 = null; const steps = try alloc.alloc(types.ToolExecutionStep, checkpoint.execution.tool_steps.len); for (checkpoint.execution.tool_steps, 0..) |step, index| { steps[index] = step; var results_changed = false; for (step.tool_results) |result| { if (result.output_handle != null or result.output.len > recovery_checkpoint_inline_output_max_bytes) { results_changed = true; break; } } if (!results_changed) continue; const results = try alloc.alloc(types.PersistedToolResult, step.tool_results.len); for (step.tool_results, 0..) |result, result_index| { results[result_index] = try projectRecoveryResult(alloc, dir, &result_dir, result); } steps[index].tool_results = results; } var projected = checkpoint; projected.execution.tool_steps = steps; return projected;}
fn projectRecoveryResult( alloc: Allocator, dir: *io_mod.VerifiedDir, result_dir: *?[]const u8, result: types.PersistedToolResult,) !types.PersistedToolResult { var projected = result; // Only spill output the restore path can read back; larger inline output // stays put and the oversize guard in writeConversationRecoveryState // covers the checkpoint as a whole. if (projected.output_handle == null and projected.output.len > recovery_checkpoint_inline_output_max_bytes and projected.output.len <= result_store.stored_text_max_bytes) { if (result_dir.* == null) { const base = try io_mod.dirRealpathAlloc(alloc, dir.dir, "."); result_dir.* = try std.fs.path.join(alloc, &.{ base, "tool-results" }); } const handle = result_store.storeLargeResult( alloc, result_dir.*.?, result.tool_call_id, result.tool_name, result.output, ) catch |err| { debug_trace.logf( "session", "event=recovery_checkpoint_spill_failed tool_call_id={s} err={s}; keeping result inline", .{ result.tool_call_id, @errorName(err) }, ); return projected; }; projected.output_handle = handle; projected.stored_output_bytes = result.output.len; } if (projected.output_handle != null and projected.output.len != 0) { if (projected.preview == null) { projected.preview = try result_store.previewText(alloc, result.output, result_store.preview_bytes); } projected.output = ""; } return projected;}
fn loadConversationPermissionState( alloc: Allocator, dir: *io_mod.VerifiedDir,) !session_permission_state.State { const bytes = readManagedFileAlloc( alloc, dir, permission_state_file, session_codec.max_permission_state_bytes, ) catch |err| switch (err) { error.FileNotFound => return .{}, else => return err, }; defer alloc.free(bytes); return session_codec.decodePermissionState(alloc, bytes);}
fn loadConversationRecoveryCheckpoint( alloc: Allocator, dir: *io_mod.VerifiedDir, conversation_seq: u64,) !?session_codec.RecoveryCheckpoint { const bytes = readManagedFileAlloc( alloc, dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes + 128, ) catch |err| switch (err) { error.FileNotFound => return null, else => return err, }; defer alloc.free(bytes); var parsed = std.json.parseFromSlice(struct { conversation_seq: u64, checkpoint: std.json.Value, }, alloc, bytes, .{ .max_value_len = session_codec.max_recovery_checkpoint_bytes, }) catch |err| switch (err) { error.OutOfMemory => return err, else => return error.InvalidRecoveryCheckpoint, }; defer parsed.deinit(); if (parsed.value.conversation_seq < conversation_seq) return null; if (parsed.value.conversation_seq != conversation_seq) return error.InvalidRecoveryCheckpoint; return session_codec.parseRecoveryCheckpoint(alloc, parsed.value.checkpoint) catch |err| switch (err) { error.OutOfMemory => return err, else => return error.InvalidRecoveryCheckpoint, };}
fn loadConversationStateIfPresent( alloc: Allocator, dir: *io_mod.VerifiedDir, expected_session_id: []const u8,) !?session_codec.DurableSessionState { return load_conversation_state_at_boundary(alloc, dir, expected_session_id, null, null, null, null);}
/// Loads owned detail state, distinguishing unreadable conversation history/// from supporting-file errors. Writable resume retains its original errors.pub fn loadConversationDetailState( alloc: Allocator, dir: *io_mod.VerifiedDir, session_id: []const u8,) !session_codec.DurableSessionState { var history_failed = false; return (load_conversation_state_at_boundary(alloc, dir, session_id, null, null, &history_failed, null) catch |err| { if (err == error.OutOfMemory or !history_failed) return err; debug_trace.logf("session", "conversation detail history unavailable id={s} err={s}", .{ session_id, @errorName(err) }); return error.ConversationHistoryUnavailable; }) orelse error.SessionMigrationRequired;}
fn load_conversation_state_at_boundary( alloc: Allocator, dir: *io_mod.VerifiedDir, expected_session_id: []const u8, recovery: ?ConversationRecoveryBoundary, replay_window: ?ConversationReplayWindow, history_failed: ?*bool, history_source: ?*HistoryFrameSource,) !?session_codec.DurableSessionState { const metadata_bytes = readManagedFileAlloc( alloc, dir, manifest_file, session_codec.max_session_metadata_bytes, ) catch |err| switch (err) { error.FileNotFound => return null, else => return err, }; defer alloc.free(metadata_bytes); var probe = std.json.parseFromSlice(std.json.Value, alloc, metadata_bytes, .{ .parse_numbers = false, }) catch |err| switch (err) { error.OutOfMemory => return err, else => return null, }; defer probe.deinit(); const object = if (probe.value == .object) probe.value.object else return null; const version_value = object.get("schema_version") orelse return null; if (version_value != .number_string or !std.mem.eql(u8, version_value.number_string, "4")) { return null; }
var metadata = try session_codec.decodeSessionMetadata(alloc, metadata_bytes); defer metadata.deinit(); if (!std.mem.eql(u8, metadata.value.id, expected_session_id)) { return error.InvalidSessionMetadata; } var conversation_seq: u64 = 0; var open_work_id: ?[]u8 = null; defer if (open_work_id) |work_id| alloc.free(work_id); const history = blk: { errdefer if (history_failed) |failed| { failed.* = true; }; var event_file = try openManagedFile(dir, events_file, .read_only); defer event_file.close(io_mod.getIo()); const length = if (recovery) |boundary| boundary.bytes else try event_file.length(io_mod.getIo()); // Recovery boundaries keep the raw-log path; the snapshot cache covers // only full loads, where a prefix replay splices into the live log. if (recovery != null) { var pure_log = HistoryFrameSource{ .log_file = event_file, .log_length = length, .limit = length }; break :blk try replayConversationHistory(alloc, &pure_log, &conversation_seq, &open_work_id, replay_window); } var owned_source: HistoryFrameSource = undefined; const frames = history_source orelse frames_blk: { owned_source = try HistoryFrameSource.open(alloc, dir, expected_session_id, event_file, length); break :frames_blk &owned_source; }; defer if (history_source == null) frames.deinit(alloc); const result = replayConversationHistory(alloc, frames, &conversation_seq, &open_work_id, replay_window) catch |err| retry: { if (err == error.OutOfMemory or frames.verified == null) return err; // A cache-sourced replay failure is a cache problem, not a log // problem: retry once from the raw log so the cache can never fail // a load. The writable caller owns rebuild decisions; read-only // loads just fall back. debug_trace.logf("session", "history cache replay rejected id={s} err={s}; reading log", .{ expected_session_id, @errorName(err) }); conversation_seq = 0; open_work_id = null; frames.verified_cleanup(alloc); frames.reset(); if (history_source != null) history_snapshot.deleteForRebuild(dir); break :retry try replayConversationHistory(alloc, frames, &conversation_seq, &open_work_id, replay_window); }; break :blk result; }; errdefer session.freeHistoryTurnSlice(alloc, history); if (recovery == null) try restoreContextResultBodies(alloc, dir, history); const latest_work_id = if (recovery != null and recovery.?.turn_open and open_work_id != null) open_work_id else latestConversationWorkId(history); const last_work_id = if (latest_work_id) |work_id| try alloc.dupe(u8, work_id) else null; errdefer if (last_work_id) |work_id| alloc.free(work_id); var usage: ?session_usage.Snapshot = try session_usage_sidecar.loadConversation( alloc, dir, expected_session_id, metadata.value.updated_at_ms, ); errdefer if (usage) |*snapshot| snapshot.deinit(alloc); var permission_state = try loadConversationPermissionState(alloc, dir); errdefer permission_state.deinit(alloc); var recovery_checkpoint = if (recovery == null) try loadConversationRecoveryCheckpoint(alloc, dir, conversation_seq) else null; errdefer if (recovery_checkpoint) |*checkpoint| checkpoint.deinit(alloc); if (recovery_checkpoint) |*checkpoint| { var capability: ?session_child_store.SessionChildCapability = null; defer if (capability) |*value| value.deinit(); try restoreExecutionResultBodies(alloc, dir, &capability, &checkpoint.execution); if (checkpoint.user.work_id == null) { checkpoint.user.work_id = open_work_id; open_work_id = null; } } const id = try alloc.dupe(u8, metadata.value.id); errdefer alloc.free(id); const origin = try alloc.dupe(u8, metadata.value.origin_workspace_root); errdefer alloc.free(origin); const workspace = try alloc.dupe(u8, metadata.value.workspace_root); errdefer alloc.free(workspace); const model = try alloc.dupe(u8, metadata.value.model); errdefer alloc.free(model); const language = session.ConversationLanguage.fromSlice( metadata.value.conversation_language, ) catch return error.InvalidSessionMetadata; const provider = metadata.value.provider; const effort = types.ReasoningEffort.parse(metadata.value.effort) orelse return error.InvalidSessionMetadata; return .{ .id = id, .origin_workspace_root = origin, .workspace_root = workspace, .created_at_ms = metadata.value.created_at_ms, .updated_at_ms = metadata.value.updated_at_ms, .conversation_language = language, .preferences = .{ .provider = provider, .model = model, .effort = effort, .fast_mode = metadata.value.fast_mode, }, .history = history, .context_history_start = latestConversationCheckpointIndex(history), .total_input_tokens = 0, .total_output_tokens = 0, .permission_state = permission_state, .last_subagent_work_id = last_work_id, .usage = usage, .recovery_checkpoint = recovery_checkpoint, .subagent_child = metadata.value.subagent_child, };}
fn latestConversationWorkId(history: []const session.HistoryTurn) ?[]const u8 { var latest: ?[]const u8 = null; for (history) |turn| { if (session.historyTurnWorkId(turn)) |work_id| latest = work_id; } return latest;}
pub fn hasConversationMetadata( alloc: Allocator, dir: *io_mod.VerifiedDir,) !bool { const bytes = (try readConversationMetadataBytes(alloc, dir)) orelse return false; defer alloc.free(bytes); return isConversationMetadata(alloc, bytes);}
pub const ConversationRecoveryBoundary = struct { bytes: u64 = 0, seq: u64 = 0, timestamp_ms: i64 = 0, turn_open: bool = false,};
fn find_conversation_recovery_boundary(alloc: Allocator, dir: *io_mod.VerifiedDir) !ConversationRecoveryBoundary { const scan = try scan_conversation_recovery(alloc, dir); if (scan.complete) return error.SessionRecoveryNotNeeded; return scan.boundary;}
const ConversationRecovery = struct { boundary: ConversationRecoveryBoundary, usage_incomplete: bool,};
pub fn classify_conversation_recovery(alloc: Allocator, dir: *io_mod.VerifiedDir, session_id: []const u8) !ConversationRecovery { const scan = try scan_conversation_recovery(alloc, dir); const usage_incomplete = try session_usage_sidecar.has_recoverable_corruption(alloc, dir, session_id); if (scan.complete and !usage_incomplete) return error.SessionRecoveryNotNeeded; return .{ .boundary = scan.boundary, .usage_incomplete = usage_incomplete };}
/// Reads only. The existing transition validator remains the record authority.fn scan_conversation_recovery( alloc: Allocator, dir: *io_mod.VerifiedDir,) !struct { boundary: ConversationRecoveryBoundary, complete: bool } { var history_buffer: [8192]u8 = undefined; var reader = try ConversationHistoryReader.init(alloc, dir, &history_buffer); defer reader.deinit(); const file = reader.file; const length = reader.length; reader.length = 0; var state = ConversationWriter{ .alloc = alloc, .file = file }; defer { state.clearPendingToolCalls(); state.pending_tool_calls.deinit(alloc); } var boundary: ConversationRecoveryBoundary = .{}; var offset: u64 = 0; var coverage_seq: u64 = 0; var coverage_progress: ConversationProgress = .{}; var buffer: [8192]u8 = undefined; var lines = file.reader(io_mod.getIo(), &buffer); var coverage_buffer: [8192]u8 = undefined; var coverage_lines = file.reader(io_mod.getIo(), &coverage_buffer); scan: while (offset < length) { const line = session_replay.readBufferedLine(alloc, &lines, length, null) catch |err| switch (err) { error.TruncatedEventFrame, error.EventFrameTooLarge => break, else => return err, } orelse break; defer alloc.free(line.bytes); var decoded = session_event.decodeConversationFrame(alloc, line.bytes) catch |err| switch (err) { error.InvalidConversationFrame => break, else => return err, }; defer decoded.deinit(); session_event.validateConversationTransition(.{ .last_seq = state.last_seq, .latest_checkpoint_coverage = state.latest_checkpoint_coverage, .pending_tool_calls = state.pending_tool_calls.items, }, decoded.value) catch break; if (decoded.value.event == .context_checkpoint) { const coverage = decoded.value.event.context_checkpoint.covers_through_seq; // Coverage is monotone, so historical cuts need only one extra scan. while (coverage_seq < coverage) { const covered = (try session_replay.readBufferedLine(alloc, &coverage_lines, offset, null)) orelse return error.SessionRecoveryBoundaryInvalid; defer alloc.free(covered.bytes); var frame = try session_event.decodeConversationFrame(alloc, covered.bytes); defer frame.deinit(); try coverage_progress.observe(frame.value.seq, frame.value.event, null); coverage_seq = frame.value.seq; } if (coverage_progress.pending != 0) break :scan; } state.applyReplayedEvent(decoded.value.seq, decoded.value.event) catch |err| switch (err) { error.InvalidConversationFrame => break :scan, else => return err, }; // A valid frame must also be replayable before it can extend the copy. reader.length = line.next_offset; if (reader.next() catch |err| switch (err) { error.InvalidConversationFrame, error.ConversationSizeOverflow => break :scan, else => return err, }) |turn| session.freeHistoryTurn(alloc, turn); offset = line.next_offset; if (!state.turn_open or (decoded.value.event == .context_checkpoint and state.pending_tool_calls.items.len == 0)) { boundary = .{ .bytes = offset, .seq = state.last_seq, .timestamp_ms = decoded.value.timestamp_ms, .turn_open = state.turn_open, }; } } const complete = offset == length and boundary.bytes == length; if (!complete and boundary.bytes == 0) return error.SessionRecoveryBoundaryInvalid; debug_trace.logf("session", "event=conversation_recovery_boundary source_bytes={d} retained_bytes={d} through_seq={d}", .{ length, boundary.bytes, boundary.seq }); return .{ .boundary = boundary, .complete = complete };}
/// Caller owns the returned complete archive, with a checkpointed open turn/// explicitly interrupted rather than scheduling its abandoned continuation.pub fn load_conversation_recovery_state( alloc: Allocator, dir: *io_mod.VerifiedDir, session_id: []const u8, boundary: ConversationRecoveryBoundary,) !session_codec.DurableSessionState { const loaded = load_conversation_state_at_boundary(alloc, dir, session_id, boundary, null, null, null) catch |err| switch (err) { error.InvalidSessionMetadata, error.InvalidSessionFormat => return error.SessionRecoveryBoundaryInvalid, else => return err, }; var state = loaded orelse return error.SessionRecoveryBoundaryInvalid; errdefer state.deinit(alloc); const file = try openManagedFile(dir, events_file, .read_only); defer file.close(io_mod.getIo()); const archive = try load_conversation_archive_from_file(alloc, file, boundary.bytes, boundary.turn_open); session.freeHistoryTurnSlice(alloc, state.history); state.history = archive; state.context_history_start = latestConversationCheckpointIndex(archive); return state;}
/// Installs exact validated records in an unpublished target; never edits source.pub fn copy_conversation_recovery_prefix( alloc: Allocator, source: *io_mod.VerifiedDir, target: *io_mod.VerifiedDir, boundary: ConversationRecoveryBoundary,) !void { const input = try openManagedFile(source, events_file, .read_only); defer input.close(io_mod.getIo()); const output = try openManagedFile(target, events_file, .read_write); defer output.close(io_mod.getIo()); var buffer: [8192]u8 = undefined; var offset: u64 = 0; while (offset < boundary.bytes) { const count: usize = @intCast(@min(buffer.len, boundary.bytes - offset)); if (try input.readPositionalAll(io_mod.getIo(), buffer[0..count], offset) != count) return error.SessionRecoveryBoundaryInvalid; try output.writePositionalAll(io_mod.getIo(), buffer[0..count], offset); offset += count; } if (boundary.turn_open) { const interrupted = try session_event.encodeConversationFrame(alloc, .{ .seq = try std.math.add(u64, boundary.seq, 1), .timestamp_ms = boundary.timestamp_ms, .event = .{ .interrupted = .{ .reason = .failed } }, }); defer alloc.free(interrupted); try output.writePositionalAll(io_mod.getIo(), interrupted, offset); offset = try std.math.add(u64, offset, interrupted.len); } try output.setLength(io_mod.getIo(), offset); try output.sync(io_mod.getIo()); debug_trace.logf("session", "event=conversation_recovery_prefix bytes={d} through_seq={d} interrupted={}", .{ boundary.bytes, boundary.seq, boundary.turn_open });}
test "conversation recovery boundary validates prefix and never edits source" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var dir = io_mod.VerifiedDir{ .dir = tmp.dir }; const prefix = "{\"schema_version\":1,\"seq\":1,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"fact\"}}}\n" ++ "{\"schema_version\":1,\"seq\":2,\"timestamp_ms\":1,\"event\":{\"assistant\":{\"text\":\"saved\"}}}\n" ++ "{\"schema_version\":1,\"seq\":3,\"timestamp_ms\":1,\"event\":{\"turn_completed\":{}}}\n"; const suffixes = [_][]const u8{ "{", "invalid\n", "{\"schema_version\":1,\"seq\":99,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"wrong sequence\"}}}\n", "{\"schema_version\":1,\"seq\":4,\"timestamp_ms\":1,\"event\":{\"context_checkpoint\":{\"covers_through_seq\":9,\"summary\":\"invalid coverage\"}}}\n", "{\"schema_version\":1,\"seq\":4,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"unfinished batch\"}}}\n" ++ "{\"schema_version\":1,\"seq\":5,\"timestamp_ms\":1,\"event\":{\"tool_call\":{\"call_id\":\"one\",\"tool_name\":\"shell\",\"arguments_json\":\"{}\"}}}\n" ++ "{\"schema_version\":1,\"seq\":6,\"timestamp_ms\":1,\"event\":{\"tool_call\":{\"call_id\":\"two\",\"tool_name\":\"shell\",\"arguments_json\":\"{}\"}}}\n" ++ "{\"schema_version\":1,\"seq\":7,\"timestamp_ms\":1,\"event\":{\"interrupted\":{\"reason\":\"failed\"}}}\ninvalid\n", }; for (suffixes) |suffix| { const bytes = try std.mem.concat(alloc, u8, &.{ prefix, suffix }); defer alloc.free(bytes); try tmp.dir.writeFile(std.testing.io, .{ .sub_path = events_file, .data = bytes, .flags = .{ .permissions = private_file_permissions } }); const boundary = try find_conversation_recovery_boundary(alloc, &dir); try std.testing.expectEqual(prefix.len, boundary.bytes); try std.testing.expectEqual(@as(u64, 3), boundary.seq); try std.testing.expect(!boundary.turn_open); const actual = try readManagedFileAlloc(alloc, &dir, events_file, bytes.len); defer alloc.free(actual); try std.testing.expectEqualStrings(bytes, actual); } try tmp.dir.writeFile(std.testing.io, .{ .sub_path = events_file, .data = prefix, .flags = .{ .permissions = private_file_permissions } }); try std.testing.expectError(error.SessionRecoveryNotNeeded, find_conversation_recovery_boundary(alloc, &dir)); try tmp.dir.writeFile(std.testing.io, .{ .sub_path = events_file, .data = "invalid\n", .flags = .{ .permissions = private_file_permissions } }); try std.testing.expectError(error.SessionRecoveryBoundaryInvalid, find_conversation_recovery_boundary(alloc, &dir));}
test "conversation recovery rejects checkpoint cuts inside tool batches" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var dir = io_mod.VerifiedDir{ .dir = tmp.dir }; const prefix = "{\"schema_version\":1,\"seq\":1,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"request\"}}}\n" ++ "{\"schema_version\":1,\"seq\":2,\"timestamp_ms\":1,\"event\":{\"tool_call\":{\"call_id\":\"one\",\"tool_name\":\"shell\",\"arguments_json\":\"{}\"}}}\n" ++ "{\"schema_version\":1,\"seq\":3,\"timestamp_ms\":1,\"event\":{\"tool_call\":{\"call_id\":\"two\",\"tool_name\":\"shell\",\"arguments_json\":\"{}\"}}}\n" ++ "{\"schema_version\":1,\"seq\":4,\"timestamp_ms\":1,\"event\":{\"tool_result\":{\"call_id\":\"one\",\"tool_name\":\"shell\",\"status\":\"success\",\"artifact_ref\":\"one.txt\",\"stored_bytes\":0,\"completeness\":\"complete\"}}}\n" ++ "{\"schema_version\":1,\"seq\":5,\"timestamp_ms\":1,\"event\":{\"tool_result\":{\"call_id\":\"two\",\"tool_name\":\"shell\",\"status\":\"success\",\"artifact_ref\":\"two.txt\",\"stored_bytes\":0,\"completeness\":\"complete\"}}}\n" ++ "{\"schema_version\":1,\"seq\":6,\"timestamp_ms\":1,\"event\":{\"turn_completed\":{}}}\n"; for ([_]u64{ 1, 2, 3, 4, 5, 6 }) |coverage| { for ([_][]const u8{ "", "invalid\n" }) |tail| { const bytes = try std.fmt.allocPrint(alloc, "{s}{{\"schema_version\":1,\"seq\":7,\"timestamp_ms\":1,\"event\":{{\"context_checkpoint\":{{\"covers_through_seq\":{d},\"summary\":\"checkpoint\"}}}}}}\n{s}", .{ prefix, coverage, tail }); defer alloc.free(bytes); try tmp.dir.writeFile(std.testing.io, .{ .sub_path = events_file, .data = bytes, .flags = .{ .permissions = private_file_permissions } }); if (coverage >= 2 and coverage <= 4) { const boundary = try find_conversation_recovery_boundary(alloc, &dir); try std.testing.expectEqual(prefix.len, boundary.bytes); try std.testing.expectEqual(@as(u64, 6), boundary.seq); } else if (tail.len == 0) { try std.testing.expectError(error.SessionRecoveryNotNeeded, find_conversation_recovery_boundary(alloc, &dir)); } else { const boundary = try find_conversation_recovery_boundary(alloc, &dir); try std.testing.expectEqual(bytes.len - tail.len, boundary.bytes); } } }}
test "conversation recovery allocation failures do not become corruption" { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var dir = io_mod.VerifiedDir{ .dir = tmp.dir }; try tmp.dir.writeFile(std.testing.io, .{ .sub_path = events_file, .flags = .{ .permissions = private_file_permissions }, .data = "{\"schema_version\":1,\"seq\":1,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"fact\"}}}\n" ++ "{\"schema_version\":1,\"seq\":2,\"timestamp_ms\":1,\"event\":{\"assistant\":{\"text\":\"saved\"}}}\n" ++ "{\"schema_version\":1,\"seq\":3,\"timestamp_ms\":1,\"event\":{\"turn_completed\":{}}}\n" ++ "{\"schema_version\":1,\"seq\":4,\"timestamp_ms\":1,\"event\":{\"context_checkpoint\":{\"covers_through_seq\":1,\"summary\":\"retained fact\"}}}\n" ++ "{\"schema_version\":1,\"seq\":5,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"pending\"}}}\n" ++ "{\"schema_version\":1,\"seq\":6,\"timestamp_ms\":1,\"event\":{\"tool_call\":{\"call_id\":\"call1\",\"tool_name\":\"shell\",\"arguments_json\":\"{}\"}}}\ninvalid\n" }); try std.testing.checkAllAllocationFailures(std.testing.allocator, struct { fn check(alloc: Allocator, source: *io_mod.VerifiedDir) !void { const boundary = try find_conversation_recovery_boundary(alloc, source); try std.testing.expectEqual(@as(u64, 4), boundary.seq); } }.check, .{&dir});}
test "conversation recovery state preserves allocation errors and contains invalid metadata" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var dir = io_mod.VerifiedDir{ .dir = tmp.dir }; try tmp.dir.writeFile(std.testing.io, .{ .sub_path = manifest_file, .flags = .{ .permissions = private_file_permissions }, .data = "{\"schema_version\":4,\"id\":\"recovery-source\",\"created_at_ms\":1,\"updated_at_ms\":1,\"origin_workspace_root\":\"/tmp\",\"workspace_root\":\"/tmp\",\"conversation_language\":\"en\",\"provider\":\"gateway\",\"model\":\"test/model\",\"effort\":\"auto\",\"fast_mode\":false,\"title\":null,\"subagent_child\":false}" }); const prefix = "{\"schema_version\":1,\"seq\":1,\"timestamp_ms\":1,\"event\":{\"user\":{\"text\":\"fact\"}}}\n" ++ "{\"schema_version\":1,\"seq\":2,\"timestamp_ms\":1,\"event\":{\"assistant\":{\"text\":\"saved\"}}}\n" ++ "{\"schema_version\":1,\"seq\":3,\"timestamp_ms\":1,\"event\":{\"turn_completed\":{}}}\n"; try tmp.dir.writeFile(std.testing.io, .{ .sub_path = events_file, .flags = .{ .permissions = private_file_permissions }, .data = prefix ++ "invalid\n" }); const boundary = try find_conversation_recovery_boundary(alloc, &dir); try std.testing.expectError(error.SessionRecoveryBoundaryInvalid, load_conversation_recovery_state(alloc, &dir, "wrong-id", boundary)); try std.testing.checkAllAllocationFailures(alloc, struct { fn check(a: Allocator, source: *io_mod.VerifiedDir, cut: ConversationRecoveryBoundary) !void { var state = try load_conversation_recovery_state(a, source, "recovery-source", cut); defer state.deinit(a); try std.testing.expectEqual(@as(usize, 1), state.history.len); try std.testing.expect(state.recovery_checkpoint == null); } }.check, .{ &dir, boundary });}
/// Reads and validates current metadata once. The caller owns the decoded value.pub fn readConversationMetadata( alloc: Allocator, dir: *io_mod.VerifiedDir,) !?session_codec.DecodedSessionMetadata { const bytes = (try readConversationMetadataBytes(alloc, dir)) orelse return null; defer alloc.free(bytes); if (!try isConversationMetadata(alloc, bytes)) return null; return session_codec.decodeSessionMetadata(alloc, bytes) catch |err| switch (err) { error.SessionMetadataTooLarge => error.InvalidSessionFormat, else => err, };}
fn readConversationMetadataBytes(alloc: Allocator, dir: *io_mod.VerifiedDir) !?[]u8 { const bytes = readManagedFileAlloc( alloc, dir, manifest_file, session_codec.max_session_metadata_bytes, ) catch |err| switch (err) { error.FileNotFound, error.InvalidSessionFormat => return null, else => return err, }; return bytes;}
fn isConversationMetadata(alloc: Allocator, bytes: []const u8) !bool { var parsed = std.json.parseFromSlice(std.json.Value, alloc, bytes, .{ .parse_numbers = false, }) catch |err| switch (err) { error.OutOfMemory => return err, else => return false, }; defer parsed.deinit(); const object = if (parsed.value == .object) parsed.value.object else return false; const version = object.get("schema_version") orelse return false; return version == .number_string and std.mem.eql(u8, version.number_string, "4");}
fn openConversationWritableSession( alloc: Allocator, writable: *WritableSessionDir,) !LoadedWritableSession { var event_file = try openManagedFile(&writable.dir, events_file, .read_write); const log_length = try event_file.length(io_mod.getIo());
// Verify the replay cache against the log, then scan from it. Any // cache-origin failure discards the cache and retries from the raw log, so // a bad cache can never block a resume. var source: ?HistoryFrameSource = null; var replay_scan: ConversationReplayScan = .{}; var conversation_writer: ConversationWriter = undefined; var snapshot_writer: ?history_snapshot.Writer = null; var cache_retry_spent = false; while (true) { source = try HistoryFrameSource.open(alloc, &writable.dir, writable.session_id, event_file, log_length); const used_cache = !cache_retry_spent and source.?.verified != null; snapshot_writer = if (source.?.verified) |*verified| history_snapshot.Writer.beginAppend(alloc, &writable.dir, verified.prefix_file_bytes) catch |err| blk: { debug_trace.logf("session", "history cache append-open failed id={s} err={s}", .{ writable.session_id, @errorName(err) }); break :blk null; } else if (log_length == 0) // A brand-new conversation keeps the minimal durable footprint; the // cache is built by the first resume with real content. null else history_snapshot.Writer.beginReplace(alloc, &writable.dir, writable.session_id) catch |err| blk: { debug_trace.logf("session", "history cache rebuild-open failed id={s} err={s}", .{ writable.session_id, @errorName(err) }); break :blk null; }; conversation_writer = ConversationWriter.initWithReplayScan( alloc, event_file, &replay_scan, &source.?, if (snapshot_writer) |*writer| writer else null, ) catch |err| { if (snapshot_writer) |*writer| writer.finalize(); snapshot_writer = null; source.?.deinit(alloc); source = null; if (err == error.OutOfMemory) { event_file.close(io_mod.getIo()); return err; } if (used_cache) { // A cache-backed scan failure is a cache problem, not a log // problem: drop the cache and retry from the raw log once. // The scan state observed partial cache frames, so it must be // reset or the retry's first log frame fails seq continuity. // The spent flag makes termination structural even when the // delete itself fails and the next open verifies again. debug_trace.logf("session", "history cache scan rejected id={s} err={s}; rebuilding from log", .{ writable.session_id, @errorName(err) }); history_snapshot.deleteForRebuild(&writable.dir); replay_scan = .{}; cache_retry_spent = true; continue; } event_file.close(io_mod.getIo()); return err; }; break; } var frames = &source.?; errdefer frames.deinit(alloc); // The cache is written only by the open-time scan tee; nothing appends at // commit time, so the tee writer is finalized here and no writer state // leaks into the session writer. if (snapshot_writer) |*writer| writer.finalize(); snapshot_writer = null; errdefer conversation_writer.deinit(); if (conversation_writer.turn_open) { var recovery = try loadConversationRecoveryCheckpoint( alloc, &writable.dir, conversation_writer.last_seq, ); defer if (recovery) |*checkpoint| checkpoint.deinit(alloc); if (recovery == null) { const interrupted: session_event.ConversationEvent = .{ .interrupted = .{ .reason = .failed }, }; const offset = conversation_writer.committed_bytes; _ = try conversation_writer.append(alloc, io_mod.milliTimestamp(), interrupted); try replay_scan.observe(offset, conversation_writer.last_seq, interrupted); } } const replay_window = try replay_scan.finish(alloc, frames); var state = (try load_conversation_state_at_boundary( alloc, &writable.dir, writable.session_id, null, replay_window, null, frames, )) orelse return error.InvalidSessionMetadata; errdefer state.deinit(alloc); if (state.recovery_checkpoint) |checkpoint| { if (checkpoint.cause == .compaction_prepared) { // This checkpoint records completed source, not permission to run // captured work. Make it ordinary interrupted history before a new // prompt can replace the recovery slot. Sequence binding makes a // crash after this append safe even if sidecar cleanup did not run. const timestamp = io_mod.milliTimestamp(); try conversation_writer.appendHistoryTurn(alloc, timestamp, checkpoint.interruptedTurn()); writeConversationRecoveryState(alloc, &writable.dir, null, conversation_writer.last_seq) catch |err| { debug_trace.logf("session", "compaction source committed but recovery cleanup failed err={s}", .{@errorName(err)}); }; var restored = (try load_conversation_state_at_boundary(alloc, &writable.dir, writable.session_id, null, null, null, frames)) orelse return error.InvalidSessionMetadata; restored.updated_at_ms = timestamp; state.deinit(alloc); state = restored; debug_trace.logf("session", "event=compaction_source_restored session={s} through_seq={d}", .{ writable.session_id, conversation_writer.last_seq }); } } const active_id = try alloc.dupe(u8, writable.session_id); errdefer alloc.free(active_id); const generation = randomIdentifier(); const position = CommitPosition{ .log_generation = generation, .through_seq = conversation_writer.last_seq, .through_event_id = randomIdentifier(), .through_event_log_bytes = conversation_writer.committed_bytes, }; frames.deinit(alloc); const result = LoadedWritableSession{ .active_id = active_id, .state = state, .conversation_writer = conversation_writer, .log = writable.*, .position = position, }; writable.* = undefined; return result;}
/// One replay frame regardless of origin. Snapshot frames carry their verified/// decoded envelope; log frames decode the JSON line as before.const HistoryFrame = struct { log_offset: u64, log_bytes: u32, line_crc: u32, envelope: session_event.ConversationEnvelope,
fn nextOffset(self: HistoryFrame) u64 { return self.log_offset + self.log_bytes; }};
/// Serves replay frames from the verified snapshot prefix, then the log/// suffix. When no usable snapshot exists, serves the whole log. Never writes;/// callers on writable paths decide whether to rebuild or extend the cache.const HistoryFrameSource = struct { log_file: std.Io.File, log_length: u64, /// Fixed read boundary for recovery loads; null follows the physical /// length so later passes see frames appended after the source opened. limit: ?u64 = null, verified: ?history_snapshot.Verified = null, cursor: history_snapshot.Cursor = undefined, buffer: [8192]u8 = undefined, reader: ?std.Io.File.Reader = null, log_next: u64 = 0, /// Cache-file offset of the last frame served from the snapshot, null when /// the last frame came from the log. The writable scan uses this to mirror /// log truncations into the cache. last_frame_snapshot_file_offset: ?u64 = null,
/// Verifies the snapshot against the already-open log file and prepares a /// merged source. Pure log when the cache is absent or invalid. fn open( alloc: Allocator, dir: *io_mod.VerifiedDir, session_id: []const u8, log_file: std.Io.File, log_length: u64, ) !HistoryFrameSource { var source = HistoryFrameSource{ .log_file = log_file, .log_length = log_length }; if (history_snapshot.openAndVerify(alloc, dir, session_id, log_file, log_length) catch |err| blk: { debug_trace.logf("session", "history cache verify failed id={s} err={s}", .{ session_id, @errorName(err) }); break :blk null; }) |verified| { source.verified = verified; source.cursor = .{ .file = verified.file, .frames = verified.frames }; source.log_next = verified.covered_log_bytes; } return source; }
fn deinit(self: *HistoryFrameSource, alloc: Allocator) void { if (self.verified) |*verified| verified.deinit(alloc); self.* = undefined; }
/// Drops the cache portion after a cache-sourced failure; the source /// becomes pure log. Callers then reset() and replay from the log. fn verified_cleanup(self: *HistoryFrameSource, alloc: Allocator) void { if (self.verified) |*verified| verified.deinit(alloc); self.verified = null; self.cursor = undefined; self.reader = null; self.log_next = 0; self.last_frame_snapshot_file_offset = null; }
fn coveredLogBytes(self: *const HistoryFrameSource) u64 { return if (self.verified) |*verified| verified.covered_log_bytes else 0; }
/// Re-stats the log so a pass that follows an open-time truncation or a /// recovery append reads the current file, not the length captured at open. fn refresh(self: *HistoryFrameSource) !void { const physical = try self.log_file.length(io_mod.getIo()); self.log_length = if (self.limit) |limit| @min(physical, limit) else physical; }
/// Repositions the source at the frame starting at `log_offset`. fn seekLogOffset(self: *HistoryFrameSource, log_offset: u64) !void { try self.refresh(); if (self.verified) |*verified| { if (log_offset < verified.covered_log_bytes) { try self.cursor.seekLogOffset(log_offset); self.log_next = verified.covered_log_bytes; self.reader = null; return; } self.cursor.index = verified.frames.len; } if (log_offset > self.log_length) return error.InvalidConversationFrame; self.log_next = log_offset; self.reader = null; }
fn reset(self: *HistoryFrameSource) void { if (self.verified != null) self.cursor.reset(); self.reader = null; self.last_frame_snapshot_file_offset = null; self.refresh() catch |err| { debug_trace.logf("session", "history source refresh failed err={s}; keeping stale length", .{@errorName(err)}); }; self.log_next = self.coveredLogBytes(); }
/// Returns the next frame in log order, or null at the end. Decoded memory /// is owned by `arena`. fn next(self: *HistoryFrameSource, arena: Allocator) !?HistoryFrame { if (self.verified) |*verified| { if (self.cursor.index < verified.frames.len) { const meta = verified.frames[self.cursor.index]; const envelope = (try self.cursor.next(arena)) orelse return error.InvalidCache; self.last_frame_snapshot_file_offset = meta.file_offset; return .{ .log_offset = meta.log_offset, .log_bytes = meta.log_bytes, .line_crc = meta.line_crc, .envelope = envelope, }; } } self.last_frame_snapshot_file_offset = null; return self.nextFromLog(arena); }
fn nextFromLog(self: *HistoryFrameSource, arena: Allocator) !?HistoryFrame { if (self.log_next >= self.log_length) return null; if (self.reader == null) { self.reader = self.log_file.reader(io_mod.getIo(), &self.buffer); self.reader.?.pos = self.log_next; } const frame_offset = self.log_next; const line = (try session_replay.readBufferedLine(arena, &self.reader.?, self.log_length, null)) orelse return null; self.log_next = line.next_offset; const decoded = try session_event.decodeConversationFrame(arena, line.bytes); return .{ .log_offset = frame_offset, .log_bytes = @intCast(line.bytes.len), .line_crc = std.hash.Crc32.hash(line.bytes), .envelope = decoded.value, }; }
/// Reads exactly the frame starting at `log_offset`. fn readAtLogOffset(self: *HistoryFrameSource, arena: Allocator, log_offset: u64) !?HistoryFrame { try self.refresh(); if (self.verified) |*verified| { if (log_offset < verified.covered_log_bytes) { var probe = self.cursor; try probe.seekLogOffset(log_offset); const meta = probe.frames[probe.index]; const envelope = (try probe.next(arena)) orelse return error.InvalidCache; return .{ .log_offset = meta.log_offset, .log_bytes = meta.log_bytes, .line_crc = meta.line_crc, .envelope = envelope, }; } } const line = (try session_replay.readLineAt(arena, self.log_file, log_offset, self.log_length)) orelse return null; const decoded = try session_event.decodeConversationFrame(arena, line.bytes); return .{ .log_offset = log_offset, .log_bytes = @intCast(line.bytes.len), .line_crc = std.hash.Crc32.hash(line.bytes), .envelope = decoded.value, }; }};
fn replayConversationHistory( alloc: Allocator, source: *HistoryFrameSource, conversation_seq: *u64, open_work_id: *?[]u8, replay_window: ?ConversationReplayWindow,) ![]session.HistoryTurn { const window = replay_window orelse try findConversationReplayWindow(alloc, source); conversation_seq.* = window.last_complete_seq; var history: std.ArrayList(session.HistoryTurn) = .empty; errdefer { for (history.items) |turn| session.freeHistoryTurn(alloc, turn); history.deinit(alloc); } var turn = ConversationTurnBuilder.init(alloc); defer turn.deinit(); var frame_arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(frame_arena); if (window.checkpoint_offset) |checkpoint_offset| { const frame = (try source.readAtLogOffset(frame_arena.allocator(), checkpoint_offset)) orelse return error.InvalidConversationFrame; const summary = try alloc.dupe(u8, frame.envelope.event.context_checkpoint.summary); errdefer alloc.free(summary); try history.append(alloc, .{ .compacted_summary = .{ .summary = summary, .removed_turn_count = window.prior_turn_count, .compaction_count = window.compaction_count, .root_user_messages_complete = false, .permission_feedback_complete = false, } }); } if (window.active_user_offset) |user_offset| { const frame = (try source.readAtLogOffset(frame_arena.allocator(), user_offset)) orelse return error.InvalidConversationFrame; if (frame.envelope.event != .user) return error.InvalidConversationFrame; try turn.begin(frame.envelope.event.user); } var checkpoint_turn_open = window.active_user_offset != null; try source.seekLogOffset(window.offset); while (true) { const frame = source.next(frame_arena.allocator()) catch |err| switch (err) { error.TruncatedEventFrame => break, else => return err, } orelse break; switch (frame.envelope.event) { .user => |value| try turn.begin(value), .assistant => |value| try turn.appendAssistant(value), .tool_call => |value| try turn.appendToolCall(value), .tool_result => |value| try turn.appendToolResult(value), .steering => |value| try turn.appendSteering(value.text), .turn_completed => |value| { const completed = try turn.finishAssistant(value); checkpoint_turn_open = false; errdefer session.freeHistoryTurn(alloc, completed); try history.append(alloc, completed); }, .interrupted => |value| { const completed = try turn.finishInterrupted(value); checkpoint_turn_open = false; errdefer session.freeHistoryTurn(alloc, completed); try history.append(alloc, completed); }, .context_checkpoint => { try turn.finishStandalone(); checkpoint_turn_open = turn.user != null; }, } _ = frame_arena.reset(.retain_capacity); } // A concurrent writer may have exposed a complete-record prefix of its // final batched turn before sync. The next writable open truncates that // incomplete turn; read-only replay returns the preceding complete turns. const completed = try history.toOwnedSlice(alloc); if (checkpoint_turn_open) { if (turn.user) |*user| { open_work_id.* = user.work_id; user.work_id = null; } } return completed;}
/// Reads complete canonical turns while retaining only the turn being built./// Each returned turn is owned by the caller.pub const ConversationHistoryReader = struct { alloc: Allocator, file: std.Io.File, reader: std.Io.File.Reader, length: u64, offset: u64 = 0, builder: ConversationTurnBuilder,
/// Borrows buffer until deinit; owns the opened file and current turn. pub fn init(alloc: Allocator, dir: *io_mod.VerifiedDir, buffer: []u8) !ConversationHistoryReader { var file = try openManagedFile(dir, events_file, .read_only); errdefer file.close(io_mod.getIo()); return .{ .alloc = alloc, .file = file, .length = try file.length(io_mod.getIo()), .reader = file.reader(io_mod.getIo(), buffer), .builder = ConversationTurnBuilder.init(alloc), }; }
pub fn deinit(self: *ConversationHistoryReader) void { self.builder.deinit(); self.file.close(io_mod.getIo()); self.* = undefined; }
pub fn next(self: *ConversationHistoryReader) !?session.HistoryTurn { while (self.offset < self.length) { const line = session_replay.readBufferedLine(self.alloc, &self.reader, self.length, null) catch |err| switch (err) { error.TruncatedEventFrame => return null, else => return err, } orelse return null; defer self.alloc.free(line.bytes); var decoded = try session_event.decodeConversationFrame(self.alloc, line.bytes); defer decoded.deinit(); const completed: ?session.HistoryTurn = switch (decoded.value.event) { .user => |value| blk: { try self.builder.begin(value); break :blk null; }, .assistant => |value| blk: { try self.builder.appendAssistant(value); break :blk null; }, .tool_call => |value| blk: { try self.builder.appendToolCall(value); break :blk null; }, .tool_result => |value| blk: { try self.builder.appendToolResult(value); break :blk null; }, .steering => |value| blk: { try self.builder.appendSteering(value.text); break :blk null; }, .turn_completed => |value| try self.builder.finishAssistant(value), .interrupted => |value| try self.builder.finishInterrupted(value), .context_checkpoint => blk: { if (self.builder.calls.items.len != 0 or self.builder.results.items.len != 0) { return error.InvalidConversationFrame; } try self.builder.finishStandalone(); break :blk null; }, }; self.offset = line.next_offset; if (completed) |turn| return turn; } return null; }};
pub fn loadConversationHistoryRange( alloc: Allocator, dir: *io_mod.VerifiedDir, start: usize, end: usize,) ![]session.HistoryTurn { if (start > end) return error.InvalidHistoryPageCursor; var buffer: [8192]u8 = undefined; var reader = try ConversationHistoryReader.init(alloc, dir, &buffer); defer reader.deinit(); var turns: std.ArrayList(session.HistoryTurn) = .empty; errdefer { for (turns.items) |turn| session.freeHistoryTurn(alloc, turn); turns.deinit(alloc); } var turn_index: usize = 0; while (turn_index < end) : (turn_index += 1) { const turn = (try reader.next()) orelse break; if (turn_index >= start) { errdefer session.freeHistoryTurn(alloc, turn); try turns.append(alloc, turn); } else { session.freeHistoryTurn(alloc, turn); } } return turns.toOwnedSlice(alloc);}
pub fn loadConversationArchive( alloc: Allocator, dir: *io_mod.VerifiedDir,) ![]session.HistoryTurn { var file = try openManagedFile(dir, events_file, .read_only); defer file.close(io_mod.getIo()); const length = try file.length(io_mod.getIo()); var source = HistoryFrameSource{ .log_file = file, .log_length = length, .limit = length }; if (readConversationMetadata(alloc, dir) catch null) |metadata| { defer metadata.deinit(); const opened: ?HistoryFrameSource = HistoryFrameSource.open(alloc, dir, metadata.value.id, file, length) catch |err| blk: { debug_trace.logf("session", "history cache archive open failed err={s}; reading log", .{@errorName(err)}); break :blk null; }; if (opened) |with_cache| { if (with_cache.verified != null) { // Keep the load bounded at open time, matching the raw-log path. var bounded = with_cache; bounded.limit = length; source = bounded; } } } defer source.deinit(alloc); return load_conversation_archive_from_source(alloc, &source, false) catch |err| { if (err == error.OutOfMemory or source.verified == null) return err; debug_trace.logf("session", "history cache archive rejected err={s}; reading log", .{@errorName(err)}); source.verified_cleanup(alloc); source.reset(); return load_conversation_archive_from_source(alloc, &source, false); };}
fn load_conversation_archive_from_file( alloc: Allocator, file: std.Io.File, length: u64, close_open_turn: bool,) ![]session.HistoryTurn { var source = HistoryFrameSource{ .log_file = file, .log_length = length, .limit = length }; return load_conversation_archive_from_source(alloc, &source, close_open_turn);}
fn load_conversation_archive_from_source( alloc: Allocator, source: *HistoryFrameSource, close_open_turn: bool,) ![]session.HistoryTurn { var turns: std.ArrayList(session.HistoryTurn) = .empty; errdefer { for (turns.items) |turn| session.freeHistoryTurn(alloc, turn); turns.deinit(alloc); } var builder = ConversationTurnBuilder.init(alloc); defer builder.deinit(); var raw_turn_count: usize = 0; var compaction_count: usize = 0; var frame_arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(frame_arena); source.reset(); while (true) { const frame = source.next(frame_arena.allocator()) catch |err| switch (err) { error.TruncatedEventFrame => break, else => return err, } orelse break; const decoded_value = frame.envelope; const completed: ?session.HistoryTurn = switch (decoded_value.event) { .user => |value| blk: { try builder.begin(value); break :blk null; }, .assistant => |value| blk: { try builder.appendAssistant(value); break :blk null; }, .tool_call => |value| blk: { try builder.appendToolCall(value); break :blk null; }, .tool_result => |value| blk: { try builder.appendToolResult(value); break :blk null; }, .steering => |value| blk: { try builder.appendSteering(value.text); break :blk null; }, .turn_completed => |value| try builder.finishAssistant(value), .interrupted => |value| try builder.finishInterrupted(value), .context_checkpoint => |value| blk: { if (builder.calls.items.len != 0 or builder.results.items.len != 0) { return error.InvalidConversationFrame; } try builder.finishStandalone(); compaction_count += 1; break :blk .{ .compacted_summary = .{ .summary = try alloc.dupe(u8, value.summary), .removed_turn_count = raw_turn_count, .compaction_count = compaction_count, .root_user_messages_complete = false, .permission_feedback_complete = false, } }; }, }; if (completed) |turn| { errdefer session.freeHistoryTurn(alloc, turn); try turns.append(alloc, turn); if (turn != .compacted_summary) raw_turn_count += 1; } _ = frame_arena.reset(.retain_capacity); } if (close_open_turn and builder.user != null) { const completed = try builder.finishInterrupted(.{ .reason = .failed }); errdefer session.freeHistoryTurn(alloc, completed); try turns.append(alloc, completed); } return turns.toOwnedSlice(alloc);}
const ConversationReplayWindow = struct { offset: u64 = 0, prior_turn_count: usize = 0, compaction_count: usize = 0, last_complete_seq: u64 = 0, active_user_offset: ?u64 = null, checkpoint_offset: ?u64 = null, coverage: u64 = 0,};
const ConversationReplayScan = struct { window: ConversationReplayWindow = .{}, turn_count: usize = 0, last_seq: u64 = 0, active_user_offset: ?u64 = null,
fn observe(self: *ConversationReplayScan, offset: u64, seq: u64, event: session_event.ConversationEvent) !void { const expected_seq = std.math.add(u64, self.last_seq, 1) catch return error.InvalidConversationFrame; if (seq != expected_seq) return error.InvalidConversationFrame; self.last_seq = seq; switch (event) { .user => { if (self.active_user_offset != null) return error.InvalidConversationFrame; self.active_user_offset = offset; }, .turn_completed, .interrupted => { if (self.active_user_offset == null) return error.InvalidConversationFrame; self.active_user_offset = null; self.turn_count = std.math.add(usize, self.turn_count, 1) catch return error.InvalidConversationFrame; self.window.last_complete_seq = seq; }, .context_checkpoint => |checkpoint| { self.window = .{ .offset = offset, .prior_turn_count = self.turn_count, .compaction_count = std.math.add(usize, self.window.compaction_count, 1) catch return error.InvalidConversationFrame, .last_complete_seq = seq, .active_user_offset = self.active_user_offset, .checkpoint_offset = offset, .coverage = checkpoint.covers_through_seq, }; }, else => {}, } }
fn finish(self: *const ConversationReplayScan, alloc: Allocator, source: *HistoryFrameSource) !ConversationReplayWindow { var window = self.window; // Coverage can precede the checkpoint: retain the recent exchanges // between those boundaries rather than replaying only after the record. if (window.checkpoint_offset != null) { window.offset = 0; window.active_user_offset = null; window.prior_turn_count = 0; source.reset(); var frame_arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(frame_arena); while (true) { const frame = (try source.next(frame_arena.allocator())) orelse break; if (frame.envelope.seq > window.coverage) break; switch (frame.envelope.event) { .user => window.active_user_offset = frame.log_offset, .turn_completed, .interrupted => { window.active_user_offset = null; window.prior_turn_count += 1; }, else => {}, } window.offset = frame.nextOffset(); _ = frame_arena.reset(.retain_capacity); } } return window; }};
fn findConversationReplayWindow( alloc: Allocator, source: *HistoryFrameSource,) !ConversationReplayWindow { var scan: ConversationReplayScan = .{}; source.reset(); var frame_arena = std.heap.ArenaAllocator.init(alloc); defer mem_utils.deinit_arena(frame_arena); while (true) { const frame = source.next(frame_arena.allocator()) catch |err| switch (err) { error.TruncatedEventFrame => break, else => return err, } orelse break; try scan.observe(frame.log_offset, frame.envelope.seq, frame.envelope.event); _ = frame_arena.reset(.retain_capacity); } return scan.finish(alloc, source);}
const ConversationTurnBuilder = struct { alloc: Allocator, user: ?types.UserTurn = null, pending_assistant: ?[]u8 = null, pending_replay: ?types.ProviderReplay = null, calls: std.ArrayList(types.ToolCall) = .empty, results: std.ArrayList(types.PersistedToolResult) = .empty, steps: std.ArrayList(types.ToolExecutionStep) = .empty, steering: std.ArrayList(types.PersistedSteering) = .empty,
fn init(alloc: Allocator) ConversationTurnBuilder { return .{ .alloc = alloc }; }
fn deinit(self: *ConversationTurnBuilder) void { if (self.user) |user| types.freeUserTurn(self.alloc, user); if (self.pending_assistant) |text| self.alloc.free(text); if (self.pending_replay) |replay| types.freeProviderReplay(self.alloc, replay); for (self.calls.items) |call| types.freeToolCall(self.alloc, call); self.calls.deinit(self.alloc); for (self.results.items) |result| freeConversationToolResult(self.alloc, result); self.results.deinit(self.alloc); for (self.steps.items) |step| { if (step.assistant) |text| self.alloc.free(text); if (step.provider_replay) |replay| types.freeProviderReplay(self.alloc, replay); types.freeToolCallSlice(self.alloc, step.tool_calls); types.freePersistedToolResults(self.alloc, step.tool_results); } for (self.steering.items) |item| { self.alloc.free(item.text); if (item.assistant_prefix) |text| self.alloc.free(text); } self.steps.deinit(self.alloc); self.steering.deinit(self.alloc); self.* = undefined; }
fn isIdle(self: *const ConversationTurnBuilder) bool { return self.user == null and self.pending_assistant == null and self.calls.items.len == 0 and self.results.items.len == 0 and self.steps.items.len == 0 and self.steering.items.len == 0; }
fn begin( self: *ConversationTurnBuilder, value: session_event.ConversationUser, ) !void { if (!self.isIdle()) return error.InvalidConversationFrame; self.user = try types.dupeUserTurn(self.alloc, .{ .text = @constCast(value.text), .images = @constCast(value.images), .work_id = if (value.work_id) |work_id| @constCast(work_id) else null, }); }
fn appendAssistant(self: *ConversationTurnBuilder, value: session_event.ConversationAssistant) !void { if (self.calls.items.len == 0 and self.results.items.len == 0) try self.finishStandalone(); if (self.user == null or self.pending_assistant != null or self.calls.items.len != 0) { return error.InvalidConversationFrame; } { const text = try self.alloc.dupe(u8, value.text); errdefer self.alloc.free(text); self.pending_replay = if (value.provider_replay) |replay| try types.dupeProviderReplay(self.alloc, replay) else null; self.pending_assistant = text; } if (value.standalone_response) try self.finishStep(); }
fn appendToolCall( self: *ConversationTurnBuilder, value: session_event.ConversationToolCall, ) !void { if (self.user == null or self.results.items.len != 0) { return error.InvalidConversationFrame; } for (self.calls.items) |call| { if (std.mem.eql(u8, call.id, value.call_id)) return error.InvalidConversationFrame; } const call = try types.dupeToolCall(self.alloc, .{ .id = value.call_id, .name = value.tool_name, .arguments_json = value.arguments_json, .argument_integrity = value.argument_integrity, .provisional_id = value.provisional_id, .provider_result = value.provider_result, .final_identity = value.final_identity, .provenance = value.provenance, }); errdefer types.freeToolCall(self.alloc, call); try self.calls.append(self.alloc, call); }
fn appendToolResult( self: *ConversationTurnBuilder, value: session_event.ConversationToolResult, ) !void { if (self.user == null or self.calls.items.len == 0) { return error.InvalidConversationFrame; } var matching_call = false; for (self.calls.items) |call| { if (!std.mem.eql(u8, call.id, value.call_id)) continue; if (!std.mem.eql(u8, call.name, value.tool_name)) { return error.InvalidConversationFrame; } matching_call = true; break; } if (!matching_call) return error.InvalidConversationFrame; for (self.results.items) |result| { if (std.mem.eql(u8, result.tool_call_id, value.call_id)) { return error.InvalidConversationFrame; } }
const result = try dupeConversationToolResult(self.alloc, value); errdefer freeConversationToolResult(self.alloc, result); try self.results.append(self.alloc, result); if (self.results.items.len == self.calls.items.len) try self.finishStep(); }
fn finishStandalone(self: *ConversationTurnBuilder) !void { if (self.calls.items.len != 0 or self.results.items.len != 0) return error.InvalidConversationFrame; const text = self.pending_assistant orelse return; if (text.len == 0 and self.pending_replay == null) { self.alloc.free(text); self.pending_assistant = null; return; } try self.finishStep(); }
fn finishStep(self: *ConversationTurnBuilder) !void { try self.steps.ensureUnusedCapacity(self.alloc, 1); const calls = try self.calls.toOwnedSlice(self.alloc); errdefer types.freeToolCallSlice(self.alloc, calls); const results = try self.results.toOwnedSlice(self.alloc); errdefer types.freePersistedToolResults(self.alloc, results); self.steps.appendAssumeCapacity(.{ .assistant = self.pending_assistant, .provider_replay = self.pending_replay, .tool_calls = calls, .tool_results = results, }); self.pending_assistant = null; self.pending_replay = null; }
fn appendSteering(self: *ConversationTurnBuilder, text: []const u8) !void { if (self.user == null or self.calls.items.len != 0 or self.results.items.len != 0) { return error.InvalidConversationFrame; } if (self.pending_replay != null) try self.finishStep(); const owned = try self.alloc.dupe(u8, text); errdefer self.alloc.free(owned); try self.steering.append(self.alloc, .{ .text = owned, .assistant_prefix = self.pending_assistant, .after_tool_step_count = self.steps.items.len, }); self.pending_assistant = null; }
fn finishAssistant( self: *ConversationTurnBuilder, completed: session_event.ConversationTurnCompleted, ) !session.HistoryTurn { if (self.user == null or self.calls.items.len != 0 or self.results.items.len != 0) { return error.InvalidConversationFrame; } const assistant = if (self.pending_assistant) |text| text else try self.alloc.dupe(u8, ""); const execution = try self.takeExecution( completed.files, completed.turn_summary, ); const user = self.user.?; self.user = null; self.pending_assistant = null; const replay = self.pending_replay; self.pending_replay = null; return .{ .assistant = .{ .user = user, .assistant = assistant, .execution = execution, .provider_replay = replay, } }; }
fn finishInterrupted( self: *ConversationTurnBuilder, value: session_event.ConversationInterruption, ) !session.HistoryTurn { if (self.user == null or self.results.items.len != 0 or self.calls.items.len > 1) { return error.InvalidConversationFrame; } if (self.calls.items.len == 0) try self.finishStandalone(); const tool_call = if (self.calls.items.len == 1) self.calls.orderedRemove(0) else null; errdefer if (tool_call) |call| types.freeToolCall(self.alloc, call); if (self.pending_assistant) |text| { self.alloc.free(text); self.pending_assistant = null; } if (self.pending_replay) |replay| { debug_trace.logf("session", "provider replay omitted reason=interrupted_association", .{}); types.freeProviderReplay(self.alloc, replay); self.pending_replay = null; } const assistant = if (value.partial_text) |text| try self.alloc.dupe(u8, text) else null; errdefer if (assistant) |text| self.alloc.free(text); const replay = if (value.command_replay_ref) |handle| blk: { const owned_handle = try self.alloc.dupe(u8, handle); break :blk types.CommandOutputReplay{ .available = .{ .handle = owned_handle, .framed_bytes = std.math.cast( usize, value.command_replay_bytes.?, ) orelse { self.alloc.free(owned_handle); return error.ConversationSizeOverflow; }, } }; } else null; errdefer if (replay) |owned| types.freeCommandOutputReplay(self.alloc, owned); const command_artifact = if (value.command_artifact_ref) |handle| try self.alloc.dupe(u8, handle) else null; errdefer if (command_artifact) |handle| self.alloc.free(handle); const cancelled_command = if (replay != null or command_artifact != null) types.CancelledCommandPresentation{ .output_replay = replay, .command_artifact_handle = command_artifact, } else null; const execution = try self.takeExecution( value.files, value.turn_summary, ); const user = self.user.?; self.user = null; return .{ .interrupted = .{ .user = user, .assistant = assistant, .tool_call = tool_call, .execution = execution, .cancelled_command = cancelled_command, .terminal_reason = value.reason, .cancellation_origin = value.cancellation_origin, } }; }
fn takeExecution( self: *ConversationTurnBuilder, files_source: []const types.FileEvidence, turn_summary: ?types.TurnSummary, ) !types.ExecutionMemory { const steps = try self.steps.toOwnedSlice(self.alloc); errdefer types.freeToolExecutionSteps(self.alloc, steps); const steering = try self.steering.toOwnedSlice(self.alloc); errdefer types.freePersistedSteering(self.alloc, steering); const files = try types.dupeFileEvidenceSlice(self.alloc, files_source); return .{ .tool_steps = steps, .files = files, .steering = steering, .turn_summary = turn_summary, }; }};
test "conversation replay keeps reasoning-only assistant units before the final reply" { const alloc = std.testing.allocator; var builder = ConversationTurnBuilder.init(alloc); defer builder.deinit(); const replay = types.ProviderReplay{ .source = .{ .provider = .gateway, .model = "test" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"\"}]" }; try builder.begin(.{ .text = "question" }); try builder.appendAssistant(.{ .text = "", .provider_replay = replay }); try builder.appendAssistant(.{ .text = "answer" }); const turn = try builder.finishAssistant(.{}); defer types.freeHistoryTurn(alloc, turn); try std.testing.expectEqualStrings("answer", turn.assistant.assistant); try std.testing.expectEqual(@as(usize, 1), turn.assistant.execution.tool_steps.len); try std.testing.expectEqualStrings(replay.parts_json, turn.assistant.execution.tool_steps[0].provider_replay.?.parts_json);}
test "reasoning-only checkpoint coverage and replay count the same completed unit" { const alloc = std.testing.allocator; const replay = types.ProviderReplay{ .source = .{ .provider = .gateway, .model = "test" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"\"}]" }; const events = [_]session_event.ConversationEvent{ .{ .user = .{ .text = "question" } }, .{ .assistant = .{ .text = "", .provider_replay = replay } }, .{ .context_checkpoint = .{ .covers_through_seq = 2, .summary = "prior facts" } }, }; var progress: ConversationProgress = .{}; var builder = ConversationTurnBuilder.init(alloc); defer builder.deinit(); for (events, 1..) |event, seq| { try progress.observe(@intCast(seq), event, .{ .tool_steps = 1 }); switch (event) { .user => |value| try builder.begin(value), .assistant => |value| try builder.appendAssistant(value), .context_checkpoint => try builder.finishStep(), else => unreachable, } } try std.testing.expectEqual(@as(u64, 2), progress.coverage); try std.testing.expectEqual(@as(usize, 1), progress.point.tool_steps); try std.testing.expectEqual(progress.point.tool_steps, builder.steps.items.len); try builder.appendAssistant(.{ .text = "final", .provider_replay = replay }); try progress.observe(4, .{ .assistant = .{ .text = "final", .provider_replay = replay } }, null); try progress.observe(5, .{ .turn_completed = .{} }, null); const turn = try builder.finishAssistant(.{}); defer types.freeHistoryTurn(alloc, turn); try std.testing.expectEqual(@as(usize, 1), progress.point.turns); try std.testing.expectEqual(@as(usize, 1), turn.assistant.execution.tool_steps.len); try std.testing.expectEqualStrings(replay.parts_json, turn.assistant.provider_replay.?.parts_json);}
test "reasoning-only history keeps an empty final response as a distinct boundary" { const alloc = std.testing.allocator; const replay = types.ProviderReplay{ .source = .{ .provider = .gateway, .model = "test" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"\"}]" }; var steps = [_]types.ToolExecutionStep{.{ .provider_replay = replay }}; var events: std.ArrayList(session_event.ConversationEvent) = .empty; defer events.deinit(alloc); try session_event.appendHistoryTurnConversationEvents(alloc, &events, .{ .assistant = .{ .user = .{ .text = @constCast("question") }, .assistant = @constCast(""), .execution = .{ .tool_steps = &steps }, } }); var builder = ConversationTurnBuilder.init(alloc); defer builder.deinit(); var progress: ConversationProgress = .{}; for (events.items, 1..) |event, seq| { try progress.observe(@intCast(seq), event, .{ .tool_steps = 1 }); switch (event) { .user => |value| try builder.begin(value), .assistant => |value| try builder.appendAssistant(value), .turn_completed => |value| { const turn = try builder.finishAssistant(value); defer types.freeHistoryTurn(alloc, turn); try std.testing.expectEqual(@as(usize, 1), turn.assistant.execution.tool_steps.len); try std.testing.expectEqualStrings("", turn.assistant.assistant); try std.testing.expect(turn.assistant.provider_replay == null); }, else => unreachable, } } try std.testing.expectEqual(@as(u64, 2), progress.coverage);}
fn dupeConversationToolResult( alloc: Allocator, value: session_event.ConversationToolResult,) !types.PersistedToolResult { const call_id = try alloc.dupe(u8, value.call_id); errdefer alloc.free(call_id); const tool_name = try alloc.dupe(u8, value.tool_name); errdefer alloc.free(tool_name); const output = try alloc.dupe(u8, value.preview orelse ""); errdefer alloc.free(output); const handle = try alloc.dupe(u8, value.artifact_ref); errdefer alloc.free(handle); const image_handle = if (value.tool_image_handle) |image_ref| try alloc.dupe(u8, image_ref) else null; errdefer if (image_handle) |image_ref| alloc.free(image_ref); const preview = if (value.preview) |text| try alloc.dupe(u8, text) else null; errdefer if (preview) |text| alloc.free(text); var permission_feedback: [][]u8 = if (value.permission_feedback.len > 0) try alloc.alloc([]u8, value.permission_feedback.len) else &.{}; errdefer if (permission_feedback.len > 0) alloc.free(permission_feedback); var feedback_count: usize = 0; errdefer for (permission_feedback[0..feedback_count]) |feedback| { alloc.free(feedback); }; for (value.permission_feedback, 0..) |feedback, index| { permission_feedback[index] = try alloc.dupe(u8, feedback); feedback_count += 1; } const command_output_replay: ?types.CommandOutputReplay = if (value.command_replay_ref) |replay_ref| blk: { const replay_handle = try alloc.dupe(u8, replay_ref); break :blk .{ .available = .{ .handle = replay_handle, .framed_bytes = std.math.cast( usize, value.command_replay_bytes.?, ) orelse { alloc.free(replay_handle); return error.ConversationSizeOverflow; }, } }; } else null; errdefer if (command_output_replay) |replay| { types.freeCommandOutputReplay(alloc, replay); }; const committed_file_presentation = if (value.committed_file_presentation) |presentation| try types.dupeCommittedFilePresentation(alloc, presentation) else null; errdefer if (committed_file_presentation) |presentation| { types.freeCommittedFilePresentation(alloc, presentation); }; const stored_bytes = std.math.cast(usize, value.stored_bytes) orelse return error.ConversationSizeOverflow; const output_bytes = std.math.cast( usize, value.output_bytes orelse value.stored_bytes, ) orelse return error.ConversationSizeOverflow; return .{ .tool_call_id = call_id, .tool_name = tool_name, .status = value.status, .output = output, .output_handle = handle, .tool_image_handle = image_handle, .preview = preview, .output_bytes = output_bytes, .stored_output_bytes = stored_bytes, .truncated = value.completeness != .complete, .provider_native = value.provider_native, .review_feedback = value.review_feedback, .created_at_ms = value.created_at_ms, .permission_feedback = permission_feedback, .committed_file_presentation = committed_file_presentation, .command_output_replay = command_output_replay, .command_process_presentation = value.command_process_presentation, .terminal_action_presentation = value.terminal_action_presentation, };}
fn freeConversationToolResult( alloc: Allocator, result: types.PersistedToolResult,) void { alloc.free(result.tool_call_id); alloc.free(result.tool_name); alloc.free(result.output); if (result.output_handle) |handle| alloc.free(handle); types.freeToolImages(alloc, result.tool_images); if (result.tool_image_handle) |handle| alloc.free(handle); if (result.preview) |preview| alloc.free(preview); for (result.permission_feedback) |feedback| alloc.free(feedback); if (result.permission_feedback.len > 0) alloc.free(result.permission_feedback); if (result.committed_file_presentation) |presentation| { types.freeCommittedFilePresentation(alloc, presentation); } if (result.command_output_replay) |replay| { types.freeCommandOutputReplay(alloc, replay); }}
fn latestConversationCheckpointIndex(history: []const session.HistoryTurn) usize { var index: usize = 0; for (history, 0..) |turn, candidate| { if (turn == .compacted_summary) index = candidate; } return index;}
fn restoreContextResultBodies(alloc: Allocator, dir: *io_mod.VerifiedDir, history: []session.HistoryTurn) !void { var capability: ?session_child_store.SessionChildCapability = null; defer if (capability) |*value| value.deinit(); for (history) |*turn| { const execution = switch (turn.*) { .assistant => |*entry| &entry.execution, .interrupted => |*entry| &entry.execution, .compacted_summary => continue, }; try restoreExecutionResultBodies(alloc, dir, &capability, execution); }}
fn restoreExecutionResultBodies( alloc: Allocator, dir: *io_mod.VerifiedDir, capability: *?session_child_store.SessionChildCapability, execution: *types.ExecutionMemory,) !void { for (execution.tool_steps) |*step| for (step.tool_results) |*result| { const handle = result.output_handle orelse continue; if (!result.truncated and result.output.len == result.stored_output_bytes) continue; const body = if (result.truncated) try result_store.formatStoredResultOutput(alloc, handle, result.preview orelse "", result.stored_output_bytes) else blk: { if (capability.* == null) { const path = try io_mod.dirRealpathAlloc(alloc, dir.dir, "."); defer alloc.free(path); capability.* = try session_child_store.SessionChildCapability.init(alloc, dir.dir, path, .read_only); } break :blk result_store.readForReplayManaged(alloc, &capability.*.?, handle, result.stored_output_bytes) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, else => { debug_trace.logf("session", "event=context_result_unavailable handle={s} err={s}; marking content unavailable", .{ handle, @errorName(err) }); const unavailable = try alloc.dupe(u8, "Saved tool-result content is unavailable. The complete output could not be restored."); result.truncated = true; break :blk unavailable; }, }; }; alloc.free(result.output); result.output = body; };}
fn projectConversationSnapshotLocators( alloc: Allocator, history: []session.HistoryTurn,) !void { for (history) |*turn| { const images = switch (turn.*) { .assistant => |*entry| entry.user.images, .interrupted => |*entry| entry.user.images, .compacted_summary => continue, }; try projectConversationImageLocators(alloc, images); }}
fn projectConversationImageLocators( alloc: Allocator, images: []types.ImageAttachment,) !void { for (images) |*image| { const snapshot_path = image.snapshot_path orelse continue; if (!std.fs.path.isAbsolute(snapshot_path)) continue; var buffer: [std.Io.Dir.max_path_bytes]u8 = undefined; const projected = try session.projectSnapshotLocator( &buffer, snapshot_path, ); const owned = try alloc.dupe(u8, projected); alloc.free(snapshot_path); image.snapshot_path = owned; }}
fn externalizeConversationResults( alloc: Allocator, state: *session_codec.DurableSessionState, capability: *session_child_store.SessionChildCapability,) !void { for (state.history) |*turn| switch (turn.*) { .assistant => |*entry| try externalizeExecutionResults( alloc, &entry.execution, capability, true, ), .interrupted => |*entry| try externalizeExecutionResults( alloc, &entry.execution, capability, true, ), .compacted_summary => {}, };}
fn externalizeConversationTurnResults( alloc: Allocator, turn: *session.HistoryTurn, capability: ?*session_child_store.SessionChildCapability,) !void { const execution = switch (turn.*) { .assistant => |*entry| &entry.execution, .interrupted => |*entry| &entry.execution, .compacted_summary => return, }; var has_results = false; for (execution.tool_steps) |step| { if (step.tool_results.len != 0) { has_results = true; break; } } if (!has_results) return; try externalizeExecutionResults( alloc, execution, capability, false, );}
fn externalizeExecutionResults( alloc: Allocator, execution: *types.ExecutionMemory, capability: ?*session_child_store.SessionChildCapability, legacy_completeness_unknown: bool,) !void { for (execution.tool_steps) |*step| { for (step.tool_results) |*result| { if (result.tool_images.len > 0 and result.tool_image_handle == null) { result.tool_image_handle = try result_store.storeToolImages( alloc, capability orelse return error.SessionChildCapabilityUnavailable, result.tool_call_id, result.tool_name, result.tool_images, ); } if (result.output_handle == null) { result.output_handle = try result_store.storeLargeResultManaged( alloc, capability orelse return error.SessionChildCapabilityUnavailable, result.tool_call_id, result.tool_name, result.output, ); if (legacy_completeness_unknown) result.truncated = true; result.stored_output_bytes = result.output.len; } if (result.preview == null) { result.preview = try result_store.previewText( alloc, result.output, result_store.preview_bytes, ); } } }}
fn deleteConversationMigrationFile( dir: *io_mod.VerifiedDir, name: []const u8,) !void { dir.dir.deleteFile(io_mod.getIo(), name) catch |err| switch (err) { error.FileNotFound => return, else => return err, };}
fn deleteLegacyConversationFiles( alloc: Allocator, dir: *io_mod.VerifiedDir, source_generation: ?Identifier,) !void { const names = [_][]const u8{ conversation_migration_backup_file, "session.legacy.json", authority_file, authority_intent_file, publication_intent_file, checkpoint_file, commit_lock_file, session_display_metadata.sidecar_file, "resume-view.bin", history_snapshot.file_name, }; for (names) |name| try deleteConversationMigrationFile(dir, name); if (source_generation) |generation| { const watermark = try watermarkName(alloc, generation); defer alloc.free(watermark); try deleteConversationMigrationFile(dir, watermark); } try io_mod.syncVerifiedDir(dir.dir);}
pub fn legacyImportRecoveryNeeded(dir: *io_mod.VerifiedDir) !bool { return entryExists(dir, conversation_migration_backup_file);}
/// Repairs the only two interrupted publication states produced by the/// one-way legacy importer. The old event log remains the authority until the/// current metadata is durable; after that point the current log wins.pub fn recoverInterruptedLegacyImport( alloc: Allocator, writable: *WritableSessionDir,) !void { if (!try legacyImportRecoveryNeeded(&writable.dir)) return; const current_metadata = try hasConversationMetadata(alloc, &writable.dir); const current_events = try entryExists(&writable.dir, events_file); if (current_metadata) { if (!current_events) return error.SessionMigrationIncomplete; try deleteConversationMigrationFile( &writable.dir, conversation_migration_backup_file, ); try deleteConversationMigrationFile( &writable.dir, conversation_migration_temp_file, ); try io_mod.syncVerifiedDir(writable.dir.dir); return; }
if (current_events) { try deleteConversationMigrationFile(&writable.dir, events_file); } try writable.dir.dir.rename( conversation_migration_backup_file, writable.dir.dir, events_file, io_mod.getIo(), ); try deleteConversationMigrationFile( &writable.dir, conversation_migration_temp_file, ); try io_mod.syncVerifiedDir(writable.dir.dir);}
pub const Boundary = enum { latest_barrier_completed,};
pub const LockKind = enum { session,};
pub const TestControls = struct { context: ?*anyopaque = null, boundary_fn: ?*const fn (?*anyopaque, Boundary) anyerror!void = null, lock_fn: ?*const fn (?*anyopaque, LockKind) void = null,
pub fn boundary(self: TestControls, point: Boundary) !void { if (self.boundary_fn) |callback| try callback(self.context, point); }
fn lock(self: TestControls, kind: LockKind) void { if (self.lock_fn) |callback| callback(self.context, kind); }};
pub const Options = struct { test_controls: TestControls = .{}, session_lock_deadline_ms: u64 = lock_deadline_ms,};
pub const SessionUpdate = union(enum) { preferences_changed: session_event.PreferencesChanged, workspace_rebound: session_event.WorkspaceRebound, history_turn_committed: session_event.HistoryTurnCommitted, usage_checkpointed: session_event.UsageCheckpointed, recovery_checkpoint_set: session_event.RecoveryCheckpointSet, recovery_checkpoint_cleared: session_event.RecoveryCheckpointCleared,};
pub const OpenMode = enum { read_only, writable,};
pub const CommitPosition = session_replay.CommitPosition;
test { _ = session_replay; _ = session_permission_state;}
pub const CleanupReport = struct { removed: usize = 0, report_only: usize = 0, ignored: usize = 0,};
pub const WritableSessionDir = struct { dir: io_mod.VerifiedDir, writer_lock: ?io_mod.TimedAdvisoryLock, session_id: []u8,
pub fn deinit(self: *WritableSessionDir, alloc: Allocator) void { if (self.writer_lock) |*lock| lock.release(); self.dir.close(); alloc.free(self.session_id); self.* = undefined; }
fn isParked(self: *const WritableSessionDir) bool { return self.writer_lock == null; }
/// Release `session.lock`. The loaded session must be retired before /// acquiring another writer, since its state may no longer match storage. pub fn park(self: *WritableSessionDir) void { const lock = &(self.writer_lock orelse return); lock.release(); self.writer_lock = null; }
pub fn eventLogLengthForTest(self: *WritableSessionDir) !u64 { var file = try openManagedFile(&self.dir, events_file, .read_only); defer file.close(io_mod.getIo()); return file.length(io_mod.getIo()); }
fn entryExistsForTest(self: *WritableSessionDir, name: []const u8) !bool { return entryExists(&self.dir, name); }};
pub const LoadedWritableSession = struct { pub const ExternalPromptOrigin = enum { root, persistent_child, };
active_id: []u8, state: session_codec.DurableSessionState, conversation_writer: ConversationWriter, log: WritableSessionDir, freshly_started: bool = false, child_capability: ?*session_child_store.SessionChildCapability = null, position: CommitPosition, migration_source_schema_version: ?u8 = null, migration_source_bytes: ?u64 = null, /// Runtime-only provenance installed by subagent resume admission. These /// fields are never written into the session event log. external_prompt_origin: ExternalPromptOrigin = .root, external_root_user_messages: [][]u8 = &.{}, external_root_user_evidence_complete: bool = false,
pub fn requireWritable(self: *const LoadedWritableSession) !void { if (self.log.isParked()) return error.SessionWriterParked; if (self.conversation_writer.failure) |err| return err; }
fn recordWriteFailure(self: *LoadedWritableSession, err: anyerror) anyerror { if (err == error.DurableReplacePostRenameFailed) { self.conversation_writer.failure = error.SessionPersistenceUncertain; return error.SessionPersistenceUncertain; } return err; }
pub fn deinit(self: *LoadedWritableSession, alloc: Allocator) void { self.conversation_writer.deinit(); if (self.child_capability) |capability| { capability.deinit(); alloc.destroy(capability); } for (self.external_root_user_messages) |message| alloc.free(message); if (self.external_root_user_messages.len > 0) { alloc.free(self.external_root_user_messages); } alloc.free(self.active_id); self.state.deinit(alloc); self.log.deinit(alloc); self.* = undefined; }
/// Returns a borrowed address that remains stable until deinit. pub fn childCapability( self: *LoadedWritableSession, ) !*session_child_store.SessionChildCapability { return if (self.child_capability) |capability| capability else error.SessionChildCapabilityUnavailable; }
/// Releases the one-time resume projection after the owning runtime has /// restored it. New conversation events continue through `ConversationWriter`. pub fn releaseHydrationHistory( self: *LoadedWritableSession, alloc: Allocator, ) void { session.freeHistoryTurnSlice(alloc, self.state.history); self.state.history = &.{}; self.state.context_history_start = 0; }
pub fn prepareHistoryTurnForCommit( self: *LoadedWritableSession, alloc: Allocator, turn: *session.HistoryTurn, ) !void { try self.requireWritable(); try externalizeConversationTurnResults( alloc, turn, self.child_capability, ); }
pub fn renameConversation( self: *LoadedWritableSession, alloc: Allocator, title: []const u8, ) !bool { try self.requireWritable(); const bytes = try readManagedFileAlloc( alloc, &self.log.dir, manifest_file, session_codec.max_session_metadata_bytes, ); defer alloc.free(bytes); var metadata = try session_codec.decodeSessionMetadata(alloc, bytes); defer metadata.deinit(); const encoded = try session_codec.encodeSessionMetadata(alloc, .{ .id = metadata.value.id, .origin_workspace_root = metadata.value.origin_workspace_root, .workspace_root = metadata.value.workspace_root, .created_at_ms = metadata.value.created_at_ms, .updated_at_ms = metadata.value.updated_at_ms, .conversation_language = metadata.value.conversation_language, .provider = metadata.value.provider, .model = metadata.value.model, .effort = metadata.value.effort, .fast_mode = metadata.value.fast_mode, .title = title, .subagent_child = metadata.value.subagent_child, }); defer alloc.free(encoded); io_mod.durableReplaceVerified( alloc, &self.log.dir, manifest_file, encoded, ) catch |err| return self.recordWriteFailure(err); return true; }
pub fn conversationTitle( self: *LoadedWritableSession, alloc: Allocator, ) !?[]u8 { const bytes = try readManagedFileAlloc( alloc, &self.log.dir, manifest_file, session_codec.max_session_metadata_bytes, ); defer alloc.free(bytes); var metadata = try session_codec.decodeSessionMetadata(alloc, bytes); defer metadata.deinit(); return if (metadata.value.title) |title| try alloc.dupe(u8, title) else null; }
pub fn appendEvent( self: *LoadedWritableSession, alloc: Allocator, event: SessionUpdate, timestamp_ms: i64, ) !CommitPosition { try self.requireWritable(); const result = switch (event) { .history_turn_committed => self.appendConversationHistoryEvent( alloc, event, timestamp_ms, ), .preferences_changed, .workspace_rebound => self.appendConversationMetadataEvent( alloc, event, timestamp_ms, ), .usage_checkpointed => self.appendConversationUsageEvent( alloc, event, timestamp_ms, ), .recovery_checkpoint_set, .recovery_checkpoint_cleared => self.appendConversationRecoveryEvent( alloc, event, timestamp_ms, ), }; return result catch |err| self.recordWriteFailure(err); }
pub fn commitContextCompaction( self: *LoadedWritableSession, alloc: Allocator, summary: types.CompactedSummaryHistoryTurn, active_prefix: ?types.AssistantHistoryTurn, retained_from: ?types.ContextHistoryCut, timestamp_ms: i64, ) !CommitPosition { try self.requireWritable(); var prepared: ?session.HistoryTurn = if (active_prefix) |prefix| try session.dupeHistoryTurn(alloc, .{ .assistant = prefix }) else null; defer if (prepared) |turn| session.freeHistoryTurn(alloc, turn); if (prepared) |*turn| try self.prepareHistoryTurnForCommit(alloc, turn); try self.conversation_writer.appendContextCompaction( alloc, timestamp_ms, summary, if (prepared) |turn| turn.assistant else null, retained_from, ); if (prepared) |turn| self.writeFirstConversationTitle(alloc, turn); if (self.conversation_writer.failure) |err| return err; return self.finishConversationCommit(alloc, timestamp_ms); }
fn appendConversationHistoryEvent( self: *LoadedWritableSession, alloc: Allocator, event: SessionUpdate, timestamp_ms: i64, ) !CommitPosition { const payload = switch (event) { .history_turn_committed => |value| value, else => return error.InvalidConversationEvent, }; var work_id = if (payload.work_id) |value| try alloc.dupe(u8, value) else null; errdefer if (work_id) |value| alloc.free(value); try self.conversation_writer.appendHistoryTurn( alloc, timestamp_ms, payload.turn, ); self.writeFirstConversationTitle(alloc, payload.turn); if (self.conversation_writer.failure) |err| return err; if (work_id) |value| { if (self.state.last_subagent_work_id) |old| alloc.free(old); self.state.last_subagent_work_id = value; work_id = null; } const language_changed = !std.mem.eql(u8, self.state.conversation_language.view(), payload.conversation_language.view()); self.state.conversation_language = payload.conversation_language; self.state.total_input_tokens = payload.total_input_tokens; self.state.total_output_tokens = payload.total_output_tokens; const position = self.finishConversationCommit(alloc, timestamp_ms); if (language_changed) { writeConversationMetadata(alloc, &self.log.dir, self.state) catch |err| { self.conversation_writer.failure = error.SessionPersistenceUncertain; debug_trace.logf("session", "history committed but language metadata failed session={s} err={s}", .{ self.active_id, @errorName(err) }); return error.SessionPersistenceUncertain; }; } if (self.conversation_writer.failure) |err| return err; return position; }
fn writeFirstConversationTitle(self: *LoadedWritableSession, alloc: Allocator, turn: session.HistoryTurn) void { if (!self.freshly_started) return; // A generated title (or a user rename) that landed before the first // commit wins; the derived title only names an untitled session. const persisted = self.conversationTitle(alloc) catch |err| { debug_trace.logf( "session", "derived session title check failed session={s} err={s}", .{ self.active_id, @errorName(err) }, ); return; }; defer if (persisted) |value| alloc.free(value); if (persisted != null) return; var display = session_display_metadata.deriveFromHistory(alloc, &.{turn}) catch return; defer display.deinit(alloc); if (!display.present) return; // The fallback placeholder is not a title: persisting it would block // background title generation, which never overwrites a named session. if (std.mem.eql(u8, display.title, session_display_metadata.fallback_title)) return; _ = self.renameConversation(alloc, display.title) catch |err| { debug_trace.logf( "session", "derived session title not persisted session={s} err={s}", .{ self.active_id, @errorName(err) }, ); }; }
fn finishConversationCommit(self: *LoadedWritableSession, alloc: Allocator, timestamp_ms: i64) CommitPosition { if (self.state.recovery_checkpoint != null) { writeConversationRecoveryState(alloc, &self.log.dir, null, self.conversation_writer.last_seq) catch |err| { debug_trace.logf( "session", "event=committed_turn_recovery_cleanup_failed session={s} err={s}", .{ self.active_id, @errorName(err) }, ); }; self.state.recovery_checkpoint.?.deinit(alloc); self.state.recovery_checkpoint = null; } self.state.updated_at_ms = timestamp_ms; self.position = .{ .log_generation = self.position.log_generation, .through_seq = self.conversation_writer.last_seq, .through_event_id = randomIdentifier(), .through_event_log_bytes = self.conversation_writer.committed_bytes, }; self.freshly_started = false; return self.position; }
fn appendConversationMetadataEvent( self: *LoadedWritableSession, alloc: Allocator, event: SessionUpdate, timestamp_ms: i64, ) !CommitPosition { switch (event) { .preferences_changed => |patch| { var preferences = try self.state.preferences.dupe(alloc); errdefer preferences.deinit(alloc); if (patch.model) |model| { const copy = try alloc.dupe(u8, model); alloc.free(preferences.model); preferences.model = copy; } if (patch.provider) |provider| preferences.provider = provider; if (patch.effort) |effort| preferences.effort = effort; if (patch.fast_mode) |fast_mode| preferences.fast_mode = fast_mode; var proposed = self.state; proposed.preferences = preferences; proposed.updated_at_ms = timestamp_ms; try session_codec.validateState(proposed); try writeConversationMetadata(alloc, &self.log.dir, proposed); self.state.preferences.deinit(alloc); self.state.preferences = preferences; }, .workspace_rebound => |rebound| { if (!std.mem.eql( u8, rebound.previous_workspace_root, self.state.workspace_root, ) or std.mem.eql( u8, rebound.workspace_root, self.state.workspace_root, )) { return error.ImmutableSessionIdentity; } const workspace_root = try alloc.dupe(u8, rebound.workspace_root); errdefer alloc.free(workspace_root); var proposed = self.state; proposed.workspace_root = workspace_root; proposed.updated_at_ms = timestamp_ms; try session_codec.validateState(proposed); try writeConversationMetadata(alloc, &self.log.dir, proposed); alloc.free(self.state.workspace_root); self.state.workspace_root = workspace_root; }, else => return error.InvalidConversationEvent, } self.state.updated_at_ms = timestamp_ms; self.freshly_started = false; return self.position; }
fn appendConversationUsageEvent( self: *LoadedWritableSession, alloc: Allocator, event: SessionUpdate, timestamp_ms: i64, ) !CommitPosition { const snapshot = switch (event) { .usage_checkpointed => |payload| payload.usage, else => return error.InvalidConversationEvent, }; var next_usage = try session_usage.dupeSnapshotOwned(alloc, snapshot); errdefer next_usage.deinit(alloc); try session_usage_sidecar.write( alloc, &self.log.dir, self.active_id, snapshot, ); if (self.state.usage) |*prior| prior.deinit(alloc); self.state.usage = next_usage; next_usage = undefined; self.state.updated_at_ms = timestamp_ms; return self.position; }
fn appendConversationRecoveryEvent( self: *LoadedWritableSession, alloc: Allocator, event: SessionUpdate, timestamp_ms: i64, ) !CommitPosition { switch (event) { .recovery_checkpoint_set => |payload| { const checkpoint = try payload.checkpoint.dupe(alloc); errdefer { var owned = checkpoint; owned.deinit(alloc); } try writeConversationRecoveryState( alloc, &self.log.dir, checkpoint, self.conversation_writer.last_seq, ); if (self.state.recovery_checkpoint) |*prior| prior.deinit(alloc); self.state.recovery_checkpoint = checkpoint; }, .recovery_checkpoint_cleared => { try writeConversationRecoveryState(alloc, &self.log.dir, null, self.conversation_writer.last_seq); if (self.state.recovery_checkpoint) |*prior| prior.deinit(alloc); self.state.recovery_checkpoint = null; }, else => return error.InvalidConversationEvent, } self.state.updated_at_ms = timestamp_ms; return self.position; }
pub fn replacePermissionState( self: *LoadedWritableSession, alloc: Allocator, permission_state: session_permission_state.State, timestamp_ms: i64, ) !void { try self.requireWritable(); var next = try session_permission_state.dupe(alloc, permission_state); errdefer next.deinit(alloc); const bytes = try session_codec.encodePermissionState(alloc, next); defer alloc.free(bytes); io_mod.durableReplaceVerified( alloc, &self.log.dir, permission_state_file, bytes, ) catch |err| return self.recordWriteFailure(err); self.state.permission_state.deinit(alloc); self.state.permission_state = next; self.state.updated_at_ms = timestamp_ms; self.freshly_started = false; }};
/// Consumes a validated legacy snapshot and its locked directory, then writes/// the current conversation representation directly. The legacy metadata stays/// authoritative until the complete event log is durable; callers receive only/// a current-format writable session.pub fn importLegacySnapshotState( alloc: Allocator, writable: *WritableSessionDir, state: session_codec.DurableSessionState, source_schema_version: u8, source_bytes: u64, source_generation: ?Identifier,) !LoadedWritableSession { return importLegacySnapshotStateWithOps( alloc, writable, state, source_schema_version, source_bytes, source_generation, .{}, );}
fn discardEmptyLegacyFileEvidence(alloc: Allocator, history: []session.HistoryTurn) !usize { var discarded: usize = 0; for (history) |*turn| { const execution = switch (turn.*) { .assistant => |*entry| &entry.execution, .interrupted => |*entry| &entry.execution, .compacted_summary => continue, }; const files = execution.files; var retained_count: usize = 0; for (files) |file| if (file.path.len > 0) { retained_count += 1; }; if (retained_count == files.len) continue; const retained = try alloc.alloc(types.FileEvidence, retained_count); var index: usize = 0; for (files) |file| { if (file.path.len == 0) { types.freeFileEvidence(alloc, file); } else { retained[index] = file; index += 1; } } discarded += files.len - retained_count; alloc.free(files); execution.files = retained; } return discarded;}
fn importLegacySnapshotStateWithOps( alloc: Allocator, writable: *WritableSessionDir, state: session_codec.DurableSessionState, source_schema_version: u8, source_bytes: u64, source_generation: ?Identifier, metadata_ops: io_mod.DurableOps,) !LoadedWritableSession { try recoverInterruptedLegacyImport(alloc, writable); var migrated_permissions = if (state.permission_state.version != session_permission_state.schema_version) try session_permission_state.migrateV1ToV2(alloc, state.permission_state) else null; defer if (migrated_permissions) |*permissions| permissions.deinit(alloc); var import_state = state; if (migrated_permissions) |permissions| import_state.permission_state = permissions; var converted = try import_state.dupe(alloc); errdefer converted.deinit(alloc); if (try converted.archive_legacy_recovery(alloc)) { debug_trace.logf("session", "legacy recovery archived session_id={s} reason=unverifiable_route_authority", .{converted.id}); } const discarded = try discardEmptyLegacyFileEvidence(alloc, converted.history); if (discarded > 0) debug_trace.logf("session", "legacy import discarded file evidence count={d} reason=empty_path", .{discarded}); try projectConversationSnapshotLocators(alloc, converted.history);
const display_path = try io_mod.dirRealpathAlloc( alloc, writable.dir.dir, ".", ); defer alloc.free(display_path); var capability = try session_child_store.SessionChildCapability.init( alloc, writable.dir.dir, display_path, .writable, ); defer capability.deinit(); try externalizeConversationResults(alloc, &converted, &capability);
const active_id = try alloc.dupe(u8, writable.session_id); errdefer alloc.free(active_id); try deleteConversationMigrationFile( &writable.dir, conversation_migration_temp_file, ); var temp_file = try createManagedFile( &writable.dir, conversation_migration_temp_file, ); var writer = ConversationWriter.init(alloc, temp_file) catch |err| { temp_file.close(io_mod.getIo()); return err; }; var writer_owned = true; defer if (writer_owned) writer.deinit(); for (converted.history, 0..) |turn, index| { if (turn == .compacted_summary) { // Legacy archives can retain raw turns before the active summary. // Earlier summaries stay archived without advancing that cut. const coverage = if (index > 0 and index == converted.context_history_start) try writer.contextCoverage(alloc, .{ .turns = turn.compacted_summary.removed_turn_count }, &.{}) else writer.latest_checkpoint_coverage; _ = try writer.append(alloc, converted.updated_at_ms, .{ .context_checkpoint = .{ .covers_through_seq = coverage, .summary = turn.compacted_summary.summary, } }); } else { try writer.appendHistoryTurn(alloc, converted.updated_at_ms, turn); } } const conversation_seq = writer.last_seq; writer.deinit(); writer_owned = false;
try writeConversationControlState(alloc, &writable.dir, converted, conversation_seq); var display = try session_display_metadata.deriveFromHistory( alloc, converted.history, ); defer display.deinit(alloc); const metadata_bytes = try encodeConversationMetadataWithTitle( alloc, converted, if (display.present) display.title else null, ); defer alloc.free(metadata_bytes); const had_previous_log = try entryExists(&writable.dir, events_file); var old_log_renamed = false; var new_log_published = false; var metadata_published = false; errdefer if (!metadata_published) { if (new_log_published) { deleteConversationMigrationFile(&writable.dir, events_file) catch {}; } if (old_log_renamed) { writable.dir.dir.rename( conversation_migration_backup_file, writable.dir.dir, events_file, io_mod.getIo(), ) catch {}; } io_mod.syncVerifiedDir(writable.dir.dir) catch {}; }; if (had_previous_log) { try writable.dir.dir.rename( events_file, writable.dir.dir, conversation_migration_backup_file, io_mod.getIo(), ); old_log_renamed = true; } try writable.dir.dir.rename( conversation_migration_temp_file, writable.dir.dir, events_file, io_mod.getIo(), ); new_log_published = true; try io_mod.syncVerifiedDir(writable.dir.dir); io_mod.durableReplaceVerifiedWithOps( alloc, &writable.dir, manifest_file, metadata_bytes, metadata_ops, ) catch |err| { // A failed directory sync can leave the new metadata visible. Keep // both logs so recovery can select the metadata that survived. if (err == error.DurableReplacePostRenameFailed) metadata_published = true; return err; }; metadata_published = true;
var hydrated = (try loadConversationStateIfPresent(alloc, &writable.dir, converted.id)) orelse return error.InvalidSessionMetadata; defer hydrated.deinit(alloc); session.freeHistoryTurnSlice(alloc, converted.history); converted.history = hydrated.history; hydrated.history = &.{}; converted.context_history_start = hydrated.context_history_start;
var event_file = try openManagedFile( &writable.dir, events_file, .read_write, ); deleteLegacyConversationFiles( alloc, &writable.dir, source_generation, ) catch |err| debug_trace.logf( "session", "event=legacy_session_cleanup_failed session={s} err={s}", .{ writable.session_id, @errorName(err) }, );
// Tee the freshly migrated log into a replay cache during the writer's // open scan so legacy sessions are cache-backed from their first resume. var import_source = HistoryFrameSource{ .log_file = event_file, .log_length = try event_file.length(io_mod.getIo()), }; var import_snapshot: ?history_snapshot.Writer = history_snapshot.Writer.beginReplace( alloc, &writable.dir, writable.session_id, ) catch |err| blk: { debug_trace.logf("session", "history cache create-on-import failed id={s} err={s}", .{ writable.session_id, @errorName(err) }); break :blk null; }; var conversation_writer = ConversationWriter.initWithReplayScan( alloc, event_file, null, &import_source, if (import_snapshot) |*snapshot| snapshot else null, ) catch |err| { if (import_snapshot) |*snapshot| snapshot.finalize(); event_file.close(io_mod.getIo()); return err; }; if (import_snapshot) |*snap| snap.finalize(); import_snapshot = null; errdefer conversation_writer.deinit();
const position = CommitPosition{ .log_generation = randomIdentifier(), .through_seq = conversation_writer.last_seq, .through_event_id = randomIdentifier(), .through_event_log_bytes = conversation_writer.committed_bytes, }; const result = LoadedWritableSession{ .active_id = active_id, .state = converted, .conversation_writer = conversation_writer, .log = writable.*, .position = position, .migration_source_schema_version = source_schema_version, .migration_source_bytes = source_bytes, }; converted = undefined; writable.* = undefined; return result;}/// Preserves each call's exact error set while sharing one dynamic return ABI.pub inline fn failLoadedWritableSession(err: anytype) @TypeOf(err)!LoadedWritableSession { return @errorCast(failLoadedWritableSessionDynamic(err));}
noinline fn failLoadedWritableSessionDynamic(err: anyerror) anyerror!LoadedWritableSession { return err;}
test "large session errors preserve their exact error type and identity" { const result = failLoadedWritableSession(error.NoSavedSessions); try std.testing.expect( @TypeOf(result) == error{NoSavedSessions}!LoadedWritableSession, ); try std.testing.expectError(error.NoSavedSessions, result);}
pub const Root = struct { sessions: ?io_mod.VerifiedDir, display_root: []u8, mode: OpenMode,
pub fn initFromHome( alloc: Allocator, home_path: []const u8, mode: OpenMode, ) !Root { const zio = io_mod.getIo(); var home = std.Io.Dir.openDirAbsolute(zio, home_path, .{ .iterate = true }) catch |err| switch (err) { error.FileNotFound => blk: { if (mode == .read_only) { return .{ .sessions = null, .display_root = try std.fs.path.join( alloc, &.{ home_path, profile_paths.root_dir_name, profile_paths.sessions_dir_name }, ), .mode = mode, }; } std.Io.Dir.createDirAbsolute(zio, home_path, private_dir_permissions) catch |create_err| switch (create_err) { error.PathAlreadyExists => {}, else => return create_err, }; break :blk try std.Io.Dir.openDirAbsolute(zio, home_path, .{ .iterate = true }); }, else => return err, }; defer home.close(zio);
var durable_home = home.openDir(zio, profile_paths.root_dir_name, .{ .iterate = true, .follow_symlinks = false, }) catch |err| switch (err) { error.FileNotFound => blk: { if (mode == .read_only) { return .{ .sessions = null, .display_root = try std.fs.path.join( alloc, &.{ home_path, profile_paths.root_dir_name, profile_paths.sessions_dir_name }, ), .mode = mode, }; } var verified_home = io_mod.VerifiedDir{ .dir = try std.Io.Dir.openDirAbsolute(zio, home_path, .{ .iterate = true, }), }; defer verified_home.close(); const created = try io_mod.openOrCreateVerifiedPrivateDir( &verified_home, profile_paths.root_dir_name, ); break :blk created.dir; }, error.NotDir, error.SymLinkLoop => return error.SessionPathUnsafe, else => return err, }; defer durable_home.close(zio); if (mode == .writable) { durable_home.setPermissions(zio, private_dir_permissions) catch return error.PrivateStatePermissionsUnsupported; } try verifyPrivateDir(durable_home, mode);
var sessions_dir = durable_home.openDir(zio, profile_paths.sessions_dir_name, .{ .iterate = true, .follow_symlinks = false, }) catch |err| switch (err) { error.FileNotFound => blk: { if (mode == .read_only) { return .{ .sessions = null, .display_root = try std.fs.path.join( alloc, &.{ home_path, profile_paths.root_dir_name, profile_paths.sessions_dir_name }, ), .mode = mode, }; } var parent = io_mod.VerifiedDir{ .dir = durable_home }; const created = try io_mod.openOrCreateVerifiedPrivateDir( &parent, profile_paths.sessions_dir_name, ); break :blk created.dir; }, error.NotDir, error.SymLinkLoop => return error.SessionPathUnsafe, else => return err, }; errdefer sessions_dir.close(zio); if (mode == .writable) { sessions_dir.setPermissions(zio, private_dir_permissions) catch return error.PrivateStatePermissionsUnsupported; } try verifyPrivateDir(sessions_dir, mode); const display_root = try io_mod.dirRealpathAlloc(alloc, sessions_dir, "."); return .{ .sessions = .{ .dir = sessions_dir }, .display_root = display_root, .mode = mode, }; }
pub fn deinit(self: *Root, alloc: Allocator) void { if (self.sessions) |*dir| dir.close(); alloc.free(self.display_root); self.* = undefined; }
pub fn startConversationSession( self: *Root, alloc: Allocator, initial_state: session_codec.DurableSessionState, options: Options, ) !LoadedWritableSession { if (self.mode != .writable or self.sessions == null) { return failLoadedWritableSession(error.SessionStoreUnavailable); } try session_codec.validateState(initial_state); const sessions = &self.sessions.?; if (try entryExists(sessions, initial_state.id)) { return failLoadedWritableSession(error.SessionAlreadyExists); }
const random = randomIdentifier(); const suffix = std.fmt.bytesToHex(random, .lower); const staging_name = try std.fmt.allocPrint(alloc, "creating+{s}", .{suffix}); defer alloc.free(staging_name); sessions.dir.createDir( io_mod.getIo(), staging_name, private_dir_permissions, ) catch return failLoadedWritableSession(error.SessionStartFailed); var unpublished = true; errdefer if (unpublished) { debug_trace.logf("session", "session creation discarded unpublished state id={s}", .{initial_state.id}); sessions.dir.deleteTree(io_mod.getIo(), staging_name) catch |err| { debug_trace.logf("session", "session creation cleanup retained id={s} err={s}", .{ initial_state.id, @errorName(err) }); }; };
var session_dir = try io_mod.openOrCreateVerifiedPrivateDir(sessions, staging_name); options.test_controls.lock(.session); var writer_lock = acquireLockWithDeadline( &session_dir, session_lock_file, true, options.session_lock_deadline_ms, ) catch |err| { session_dir.close(); return mapSessionLockError(err); }; const session_id = alloc.dupe(u8, initial_state.id) catch |err| { writer_lock.release(); session_dir.close(); return err; }; var writable = WritableSessionDir{ .dir = session_dir, .writer_lock = writer_lock, .session_id = session_id, }; var writable_owned = true; errdefer if (writable_owned) writable.deinit(alloc); var loaded = try createNativeSession( alloc, &writable, initial_state, options, ); writable_owned = false; errdefer loaded.deinit(alloc); try loaded.conversation_writer.file.sync(io_mod.getIo()); try io_mod.syncVerifiedDir(loaded.log.dir.dir); publishSessionDirectory(sessions.dir, staging_name, initial_state.id) catch |err| { if (err == error.PathAlreadyExists) return error.SessionAlreadyExists; debug_trace.logf("session", "session creation publication failed id={s} err={s}", .{ initial_state.id, @errorName(err) }); return err; }; unpublished = false; io_mod.syncVerifiedDir(sessions.dir) catch |err| { debug_trace.logf("session", "session creation retained published state id={s} durability=uncertain err={s}", .{ initial_state.id, @errorName(err) }); return error.SessionStartFailed; }; return loaded; }
pub fn resumeForWrite( self: *Root, alloc: Allocator, session_id: []const u8, options: Options, ) !LoadedWritableSession { options.test_controls.lock(.session); var writable = try self.openWritableSessionDir( alloc, session_id, options.session_lock_deadline_ms, ); errdefer writable.deinit(alloc); if (!try hasConversationMetadata(alloc, &writable.dir)) { return error.SessionMigrationRequired; } return openConversationWritableSession(alloc, &writable); }
pub fn loadReadOnly( self: *Root, alloc: Allocator, session_id: []const u8, _: Options, ) !session_codec.DurableSessionState { if (self.sessions == null) return error.SessionNotFound; var session_dir = openSessionDir( &self.sessions.?, session_id, .read_only, ) catch |err| switch (err) { error.FileNotFound => return error.SessionNotFound, else => return err, }; defer session_dir.close(); return (try loadConversationStateIfPresent( alloc, &session_dir, session_id, )) orelse error.SessionMigrationRequired; }
/// Reads the immutable child-privacy bit from current session metadata. /// Legacy sessions retain their read-only first-event compatibility path. pub fn loadSubagentChildIdentity( self: *const Root, alloc: Allocator, session_id: []const u8, ) !bool { var sessions = self.sessions orelse return error.SessionNotFound; try session_layout.validateSessionId(session_id); var session_dir = openSessionDir( &sessions, session_id, .read_only, ) catch |err| switch (err) { error.FileNotFound => return error.SessionNotFound, else => return err, }; defer session_dir.close(); if (try hasConversationMetadata(alloc, &session_dir)) { const metadata_bytes = try readManagedFileAlloc( alloc, &session_dir, manifest_file, session_codec.max_session_metadata_bytes, ); defer alloc.free(metadata_bytes); var metadata = try session_codec.decodeSessionMetadata( alloc, metadata_bytes, ); defer metadata.deinit(); if (!std.mem.eql(u8, metadata.value.id, session_id)) { return error.InvalidSessionMetadata; } return metadata.value.subagent_child; } var log_file = try openManagedFile(&session_dir, events_file, .read_only); defer log_file.close(io_mod.getIo()); return session_replay.readSubagentChildIdentity(alloc, log_file); }
fn openWritableSessionDir( self: *Root, alloc: Allocator, session_id: []const u8, deadline_ms: u64, ) !WritableSessionDir { if (self.mode != .writable or self.sessions == null) return error.SessionNotFound; try session_layout.validateSessionId(session_id); var dir = openSessionDir(&self.sessions.?, session_id, .writable) catch |err| switch (err) { error.FileNotFound => return error.SessionNotFound, else => return err, }; var writer_lock = acquireLockWithDeadline( &dir, session_lock_file, true, deadline_ms, ) catch |err| { dir.close(); return mapSessionLockError(err); }; const owned_id = alloc.dupe(u8, session_id) catch |err| { writer_lock.release(); dir.close(); return err; }; return .{ .dir = dir, .writer_lock = writer_lock, .session_id = owned_id, }; }
fn entryExistsForTest( self: *Root, alloc: Allocator, session_id: []const u8, name: []const u8, ) !bool { _ = alloc; if (self.sessions == null) return false; var dir = openSessionDir(&self.sessions.?, session_id, .read_only) catch |err| switch (err) { error.FileNotFound => return false, else => return err, }; defer dir.close(); return entryExists(&dir, name); }};
fn publishSessionDirectory(parent: std.Io.Dir, staging: []const u8, target: []const u8) !void { if (comptime @import("builtin").os.tag != .macos) { return parent.renamePreserve(staging, parent, target, io_mod.getIo()); } // Zig 0.16's Darwin preserve-rename uses hard links, which cannot publish directories. const Darwin = struct { extern "c" fn renameatx_np(c_int, [*:0]const u8, c_int, [*:0]const u8, c_uint) c_int; }; var staging_buffer: [256]u8 = undefined; var target_buffer: [256]u8 = undefined; const staging_z = try std.fmt.bufPrintZ(&staging_buffer, "{s}", .{staging}); const target_z = try std.fmt.bufPrintZ(&target_buffer, "{s}", .{target}); while (true) { try io_mod.getIo().checkCancel(); const err = std.posix.errno(Darwin.renameatx_np(parent.handle, staging_z, parent.handle, target_z, 0x4)); switch (err) { .SUCCESS => return, .INTR => continue, .EXIST, .NOTEMPTY => return error.PathAlreadyExists, .OPNOTSUPP, .NOSYS => return error.OperationUnsupported, else => { debug_trace.logf("session", "session directory publication failed errno={s}", .{@tagName(err)}); return error.SessionStartFailed; }, } }}
fn validateLeaf(name: []const u8) !void { if (name.len == 0 or std.mem.eql(u8, name, ".") or std.mem.eql(u8, name, "..") or std.mem.indexOfAny(u8, name, "/\\") != null) { return error.SessionPathUnsafe; }}
fn verifyPrivateDir(dir: std.Io.Dir, mode: OpenMode) !void { const stat = try dir.stat(io_mod.getIo()); if (stat.kind != .directory) return error.SessionPathUnsafe; if (mode == .writable and stat.permissions.toMode() & 0o777 != 0o700) { return error.PrivateStatePermissionsUnsupported; }}
fn verifyManagedFile(file: std.Io.File, mode: OpenMode) !void { const stat = try file.stat(io_mod.getIo()); if (stat.kind != .file or stat.nlink != 1) return error.SessionPathUnsafe; if (mode == .writable and stat.permissions.toMode() & 0o777 != 0o600) { return error.PrivateStatePermissionsUnsupported; }}
fn openSessionDir( sessions: *io_mod.VerifiedDir, session_id: []const u8, mode: OpenMode,) !io_mod.VerifiedDir { try session_layout.validateSessionId(session_id); var dir = sessions.dir.openDir(io_mod.getIo(), session_id, .{ .iterate = true, .follow_symlinks = false, }) catch |err| switch (err) { error.NotDir, error.SymLinkLoop => return error.SessionPathUnsafe, else => return err, }; errdefer dir.close(io_mod.getIo()); if (mode == .writable) { dir.setPermissions(io_mod.getIo(), private_dir_permissions) catch return error.PrivateStatePermissionsUnsupported; } try verifyPrivateDir(dir, mode); return .{ .dir = dir };}
fn openManagedFile( dir: *io_mod.VerifiedDir, name: []const u8, mode: std.Io.Dir.OpenFileOptions.Mode,) !std.Io.File { try validateLeaf(name); var file = dir.dir.openFile(io_mod.getIo(), name, .{ .mode = mode, .allow_directory = false, .follow_symlinks = false, .resolve_beneath = true, }) catch |err| switch (err) { error.IsDir, error.NotDir, error.SymLinkLoop => return error.SessionPathUnsafe, else => return err, }; errdefer file.close(io_mod.getIo()); try verifyManagedFile(file, if (mode == .read_only) .read_only else .writable); return file;}
fn createManagedFile( dir: *io_mod.VerifiedDir, name: []const u8,) !std.Io.File { try validateLeaf(name); var file = dir.dir.createFile(io_mod.getIo(), name, .{ .read = true, .truncate = false, .exclusive = true, .permissions = private_file_permissions, .resolve_beneath = true, }) catch |err| switch (err) { error.IsDir, error.NotDir, error.SymLinkLoop => return error.SessionPathUnsafe, else => return err, }; errdefer file.close(io_mod.getIo()); file.setPermissions(io_mod.getIo(), private_file_permissions) catch return error.PrivateStatePermissionsUnsupported; try verifyManagedFile(file, .writable); return file;}
fn entryExists(dir: *io_mod.VerifiedDir, name: []const u8) !bool { try validateLeaf(name); const stat = dir.dir.statFile(io_mod.getIo(), name, .{ .follow_symlinks = false, }) catch |err| switch (err) { error.FileNotFound => return false, error.NotDir, error.SymLinkLoop => return error.SessionPathUnsafe, else => return err, }; if (stat.kind != .file and stat.kind != .directory) return error.SessionPathUnsafe; if (stat.kind == .file and stat.nlink != 1) return error.SessionPathUnsafe; return true;}
fn acquireLock( dir: *io_mod.VerifiedDir, name: []const u8, create: bool,) !io_mod.TimedAdvisoryLock { return acquireLockWithDeadline(dir, name, create, lock_deadline_ms);}
fn acquireLockWithDeadline( dir: *io_mod.VerifiedDir, name: []const u8, create: bool, deadline_ms: u64,) !io_mod.TimedAdvisoryLock { if (create) { return io_mod.acquireTimedAdvisoryLock(dir, name, deadline_ms); } var file = try openManagedFile(dir, name, .read_write); errdefer file.close(io_mod.getIo()); const started = io_mod.milliTimestamp(); while (true) { const locked = file.tryLock(io_mod.getIo(), .exclusive) catch |err| switch (err) { error.FileLocksUnsupported => return error.LockUnsupported, else => return err, }; if (locked) return .{ .file = file }; if (io_mod.milliTimestamp() - started >= deadline_ms) return error.LockBusy; io_mod.sleep(10 * std.time.ns_per_ms); }}
fn mapSessionLockError(err: anyerror) anyerror { return switch (err) { error.LockBusy => error.SessionBusy, error.LockUnsupported => error.SessionLockUnsupported, else => err, };}
fn readManagedFileAlloc( alloc: Allocator, dir: *io_mod.VerifiedDir, name: []const u8, max_bytes: usize,) ![]u8 { var file = try openManagedFile(dir, name, .read_only); defer file.close(io_mod.getIo()); return io_mod.readFileToEnd(alloc, &file, max_bytes + 1) catch |err| switch (err) { error.StreamTooLong => error.InvalidSessionFormat, else => err, };}
fn randomIdentifier() Identifier { var id: Identifier = undefined; io_mod.getIo().random(&id); return id;}
fn watermarkName(alloc: Allocator, generation: Identifier) ![]u8 { const hex = std.fmt.bytesToHex(generation, .lower); return std.fmt.allocPrint(alloc, "commit.{s}.json", .{hex});}
fn createNativeSession( alloc: Allocator, writable: *WritableSessionDir, initial_state: session_codec.DurableSessionState, _: Options,) !LoadedWritableSession { var synthesized_usage: ?session_usage.Snapshot = null; if (initial_state.usage == null) { var fresh_usage = session_usage.Usage.initFresh(); defer fresh_usage.deinit(alloc); synthesized_usage = try fresh_usage.snapshot(alloc); } defer if (synthesized_usage) |*usage| usage.deinit(alloc);
var display = try session_display_metadata.deriveFromHistory( alloc, initial_state.history, ); defer display.deinit(alloc); var conversation_writer = try createConversationStorage(alloc, &writable.dir, .{ .id = initial_state.id, .origin_workspace_root = initial_state.origin_workspace_root, .workspace_root = initial_state.workspace_root, .created_at_ms = initial_state.created_at_ms, .updated_at_ms = initial_state.updated_at_ms, .conversation_language = initial_state.conversation_language.view(), .provider = initial_state.preferences.provider, .model = initial_state.preferences.model, .effort = initial_state.preferences.effort.label(), .fast_mode = initial_state.preferences.fast_mode, .title = if (display.present) display.title else null, .subagent_child = initial_state.subagent_child, }); errdefer conversation_writer.deinit(); for (initial_state.history) |turn| { if (turn == .compacted_summary and conversation_writer.last_seq == 0) continue; try conversation_writer.appendHistoryTurn( alloc, initial_state.updated_at_ms, turn, ); }
var state = try initial_state.dupe(alloc); errdefer state.deinit(alloc); if (state.usage == null) { state.usage = synthesized_usage; synthesized_usage = null; } try writeConversationControlState(alloc, &writable.dir, state, conversation_writer.last_seq); const active_id = try alloc.dupe(u8, writable.session_id); errdefer alloc.free(active_id); const generation = randomIdentifier(); const position = CommitPosition{ .log_generation = generation, .through_seq = conversation_writer.last_seq, .through_event_id = randomIdentifier(), .through_event_log_bytes = conversation_writer.committed_bytes, }; const result = LoadedWritableSession{ .active_id = active_id, .state = state, .conversation_writer = conversation_writer, .log = writable.*, .freshly_started = true, .position = position, }; writable.* = undefined; return result;}
pub fn durableStatesEqual( first: session_codec.DurableSessionState, second: session_codec.DurableSessionState,) !bool { var first_buffer: [4096]u8 = undefined; var first_discard = std.Io.Writer.Discarding.init(&first_buffer); const first_summary = try session_codec.encodeState( first, &first_discard.writer, ); var second_buffer: [4096]u8 = undefined; var second_discard = std.Io.Writer.Discarding.init(&second_buffer); const second_summary = try session_codec.encodeState( second, &second_discard.writer, ); return first_summary.encoded_bytes == second_summary.encoded_bytes and std.mem.eql( u8, &first_summary.sha256, &second_summary.sha256, );}
const TempRoot = struct { tmp: std.testing.TmpDir, home: []u8, root: Root,
fn init(alloc: Allocator) !TempRoot { var tmp = std.testing.tmpDir(.{}); errdefer tmp.cleanup(); try tmp.dir.createDir(io_mod.getIo(), "home", std.Io.File.Permissions.fromMode(0o700)); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "home"); errdefer alloc.free(home); var root = try Root.initFromHome(alloc, home, .writable); errdefer root.deinit(alloc); return .{ .tmp = tmp, .home = home, .root = root }; }
fn deinit(self: *TempRoot, alloc: Allocator) void { self.root.deinit(alloc); alloc.free(self.home); self.tmp.cleanup(); self.* = undefined; }};
test "root init rejects symlinked durable and sessions roots" { const alloc = std.testing.allocator;
{ var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDirPath(io_mod.getIo(), "home"); try tmp.dir.createDirPath(io_mod.getIo(), "outside"); tmp.dir.symLink( io_mod.getIo(), "../outside", "home/.fx", .{ .is_directory = true }, ) catch |err| switch (err) { error.AccessDenied => return error.SkipZigTest, else => return err, }; const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "home"); defer alloc.free(home);
try std.testing.expectError( error.SessionPathUnsafe, Root.initFromHome(alloc, home, .writable), ); }
{ var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDirPath(io_mod.getIo(), "home/.fx"); try tmp.dir.createDirPath(io_mod.getIo(), "outside"); tmp.dir.symLink( io_mod.getIo(), "../../outside", "home/.fx/sessions", .{ .is_directory = true }, ) catch |err| switch (err) { error.AccessDenied => return error.SkipZigTest, else => return err, }; const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "home"); defer alloc.free(home);
try std.testing.expectError( error.SessionPathUnsafe, Root.initFromHome(alloc, home, .writable), ); }}
fn testState(alloc: Allocator, id: []const u8, updated_at_ms: i64) !session_codec.DurableSessionState { return .{ .id = try alloc.dupe(u8, id), .origin_workspace_root = try alloc.dupe(u8, "/tmp/fx-plan-03"), .workspace_root = try alloc.dupe(u8, "/tmp/fx-plan-03"), .created_at_ms = 10, .updated_at_ms = updated_at_ms, .conversation_language = session.ConversationLanguage.literal("en"), .preferences = .{ .model = try alloc.dupe(u8, "test/model"), .effort = types.ReasoningEffort.literal("high"), .fast_mode = false, }, .history = &.{}, .total_input_tokens = 0, .total_output_tokens = 0, };}
test "conversation writer appends one durable line per event" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true, .truncate = true, }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit();
try std.testing.expectEqual(@as(u64, 1), try writer.append(alloc, 10, .{ .user = .{ .text = "Run the command." }, })); try std.testing.expectEqual(@as(u64, 2), try writer.append(alloc, 11, .{ .tool_call = .{ .call_id = "call-shell", .tool_name = "shell", .arguments_json = "{\"command\":\"printf done\"}", }, })); try std.testing.expectEqual(@as(u64, 3), try writer.append(alloc, 12, .{ .tool_result = .{ .call_id = "call-shell", .tool_name = "shell", .status = .success, .artifact_ref = "sha256:abc", .stored_bytes = 4, .completeness = .complete, .preview = "done", }, })); try std.testing.expectEqual(@as(usize, 0), writer.pendingToolCallCount());
const bytes = try writer.readAllForTest(alloc); defer alloc.free(bytes); try std.testing.expectEqual(@as(usize, 3), std.mem.count(u8, bytes, "\n")); try std.testing.expect(std.mem.find(u8, bytes, "commit.pending") == null); try std.testing.expect(std.mem.find(u8, bytes, "state_replacement") == null);}
test "conversation writer preserves a suffix written outside its ownership" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |checkpoint| { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit(); try writer.appendHistoryTurn(alloc, 1, .{ .assistant = .{ .user = .{ .text = @constCast("accepted request") }, .assistant = @constCast("accepted answer"), } }); const external = try session_event.encodeConversationFrame(alloc, .{ .seq = writer.last_seq + 1, .timestamp_ms = 2, .event = .{ .context_checkpoint = .{ .covers_through_seq = writer.last_seq, .summary = "another writer's accepted context", } }, }); defer alloc.free(external); try file.writePositionalAll(std.testing.io, external, writer.committed_bytes); try file.sync(std.testing.io); const before = try alloc.alloc(u8, @intCast(try file.length(std.testing.io))); defer alloc.free(before); try std.testing.expectEqual(before.len, try file.readPositionalAll(std.testing.io, before, 0)); const turn: types.HistoryTurn = if (checkpoint) .{ .compacted_summary = .{ .summary = @constCast("stale checkpoint"), .removed_turn_count = 1, .compaction_count = 1 } } else .{ .assistant = .{ .user = .{ .text = @constCast("stale request") }, .assistant = @constCast("stale answer") } }; try std.testing.expectError(error.SessionWriterChanged, writer.appendHistoryTurn(alloc, 3, turn)); try std.testing.expectEqual(before.len, try file.length(std.testing.io)); const after = try alloc.alloc(u8, before.len); defer alloc.free(after); try std.testing.expectEqual(after.len, try file.readPositionalAll(std.testing.io, after, 0)); try std.testing.expectEqualSlices(u8, before, after); }}
test "conversation writer rolls back failed sync and refuses uncertain continuation" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |checkpoint| { for ([_]bool{ false, true }) |fail_rollback| { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit(); try writer.appendHistoryTurn(alloc, 10, .{ .assistant = .{ .user = .{ .text = @constCast("accepted request") }, .assistant = @constCast("accepted answer"), } }); const prior = try writer.readAllForTest(alloc); defer alloc.free(prior); const SyncFault = struct { calls: usize = 0, fail_rollback: bool,
fn sync(ctx: ?*anyopaque, target: std.Io.File) !void { const self: *@This() = @ptrCast(@alignCast(ctx.?)); self.calls += 1; if (self.calls == 1 or (self.fail_rollback and self.calls == 2)) return error.InjectedSyncFailure; try target.sync(std.testing.io); } }; var fault = SyncFault{ .fail_rollback = fail_rollback }; writer.test_sync_ops = .{ .ctx = &fault, .sync_file = SyncFault.sync }; const turn: types.HistoryTurn = if (checkpoint) .{ .compacted_summary = .{ .summary = @constCast("failed summary"), .removed_turn_count = 1, .compaction_count = 1 } } else .{ .assistant = .{ .user = .{ .text = @constCast("failed request") }, .assistant = @constCast("failed answer") } }; try std.testing.expectError( if (fail_rollback) error.SessionPersistenceUncertain else error.InjectedSyncFailure, writer.appendHistoryTurn(alloc, 11, turn), ); try std.testing.expectEqual(@as(usize, 2), fault.calls); try std.testing.expectEqual(prior.len, try file.length(std.testing.io)); try std.testing.expectEqual(@as(u64, 3), writer.last_seq); const after = try writer.readAllForTest(alloc); defer alloc.free(after); try std.testing.expectEqualSlices(u8, prior, after); if (fail_rollback) { try std.testing.expectError(error.SessionPersistenceUncertain, writer.appendHistoryTurn(alloc, 12, turn)); try std.testing.expectEqual(@as(usize, 2), fault.calls); } else { try writer.appendHistoryTurn(alloc, 12, turn); try std.testing.expect(writer.committed_bytes > prior.len); } } }}
test "parked session refuses history and control mutations" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "parked-writes", 10); defer initial.deinit(alloc); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); loaded.log.park(); try std.testing.expectError(error.SessionWriterParked, loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("request") }, .assistant = @constCast("answer") } }, } }, 2)); try std.testing.expectError(error.SessionWriterParked, loaded.appendEvent(alloc, .{ .preferences_changed = .{ .fast_mode = true } }, 2)); try std.testing.expectError(error.SessionWriterParked, loaded.commitContextCompaction(alloc, .{ .summary = @constCast("summary"), .removed_turn_count = 1, .compaction_count = 1 }, null, null, 2));}
test "writable resume releases ownership when metadata allocation fails" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "resume-allocation", 10); defer initial.deinit(alloc); var started = try temp.root.startConversationSession(alloc, initial, .{}); started.deinit(alloc); var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 1 }); try std.testing.expectError(error.OutOfMemory, temp.root.resumeForWrite(failing.allocator(), initial.id, .{ .session_lock_deadline_ms = 0 })); try std.testing.expectEqual(failing.allocated_bytes, failing.freed_bytes); var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{ .session_lock_deadline_ms = 0 }); defer resumed.deinit(alloc); try std.testing.expectEqualStrings(initial.id, resumed.active_id);}
test "committed conversation language survives reopening" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "language-roundtrip", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("fr"), .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("Bonjour") }, .assistant = @constCast("Salut") } }, } }, 20); } var restored = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer restored.deinit(alloc); try std.testing.expectEqualStrings("fr", restored.conversation_language.view()); try std.testing.expectEqual(@as(usize, 1), restored.history.len);}
test "first commit never persists the fallback placeholder as a title" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc);
// An unusable first prompt (a bare path reads as slash-command-only) must // not persist the fallback title, or later title generation could never // name the session. var initial = try testState(alloc, "fallback-title-skipped", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .interrupted = .{ .user = .{ .text = @constCast("/tmp/fx-trace.md") } } }, } }, 20); const title = try loaded.conversationTitle(alloc); defer if (title) |value| alloc.free(value); try std.testing.expect(title == null); }
// A usable first prompt still derives and persists a title. var second = try testState(alloc, "derived-title-kept", 10); defer second.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, second, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("refactor the renderer") }, .assistant = @constCast("done") } }, } }, 20); const title = try loaded.conversationTitle(alloc); defer if (title) |value| alloc.free(value); try std.testing.expectEqualStrings("refactor the renderer", title.?); }}
test "conversation load retains history and qualifies missing or corrupt accounting" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |missing| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "accounting-recovery", 10); defer initial.deinit(alloc); initial.history = try alloc.alloc(session.HistoryTurn, 1); initial.history[0] = try session.makeAssistantTurn(alloc, "saved request", "saved answer"); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); loaded.deinit(alloc); var dir = try openSessionDir(&temp.root.sessions.?, initial.id, .read_only); defer dir.close(); if (missing) { try dir.dir.deleteFile(std.testing.io, session_usage_sidecar.sidecar_file); } else { try io_mod.durableReplaceVerified(alloc, &dir, session_usage_sidecar.sidecar_file, "{broken accounting"); } var restored = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer restored.deinit(alloc); try std.testing.expectEqualStrings("saved answer", restored.history[0].assistant.assistant); try std.testing.expect(restored.usage != null); try std.testing.expectEqual(session_usage.Availability.incomplete, restored.usage.?.billing); try std.testing.expectEqual(@as(usize, 1), restored.usage.?.incidents.len); if (!missing) { const retained = try readManagedFileAlloc(alloc, &dir, session_usage_sidecar.sidecar_file, 1024); defer alloc.free(retained); try std.testing.expectEqualStrings("{broken accounting", retained); } }}
test "conversation writer repairs only a partial final record and continues" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true, .truncate = true, }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit(); try writer.appendHistoryTurn(alloc, 10, .{ .assistant = .{ .user = .{ .text = @constCast("first") }, .assistant = @constCast("answer"), } }); } const valid_bytes = blk: { const file = try tmp.dir.openFile(std.testing.io, "events.jsonl", .{}); defer file.close(std.testing.io); break :blk try file.length(std.testing.io); }; { const file = try tmp.dir.openFile(std.testing.io, "events.jsonl", .{ .mode = .read_write, }); defer file.close(std.testing.io); try file.writePositionalAll(std.testing.io, "{\"schema_version\":1", valid_bytes); try file.sync(std.testing.io); }
const file = try tmp.dir.openFile(std.testing.io, "events.jsonl", .{ .mode = .read_write, }); var resumed = try ConversationWriter.init(alloc, file); defer resumed.deinit(); try std.testing.expectEqual(@as(u64, 3), resumed.last_seq); try std.testing.expectEqual(valid_bytes, resumed.committed_bytes); try std.testing.expectEqual(valid_bytes, try resumed.file.length(std.testing.io)); try resumed.appendHistoryTurn(alloc, 11, .{ .assistant = .{ .user = .{ .text = @constCast("continued") }, .assistant = @constCast("done"), } }); try std.testing.expectEqual(@as(u64, 6), resumed.last_seq);}
test "conversation writer truncates a complete-record partial turn" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var valid_bytes: u64 = 0; { const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true, .truncate = true, }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit(); try writer.appendHistoryTurn(alloc, 10, .{ .assistant = .{ .user = .{ .text = @constCast("complete") }, .assistant = @constCast("answer"), } }); valid_bytes = writer.committed_bytes; const dangling = try session_event.encodeConversationFrame(alloc, .{ .seq = 4, .timestamp_ms = 11, .event = .{ .user = .{ .text = "dangling" } }, }); defer alloc.free(dangling); try writer.file.writePositionalAll(std.testing.io, dangling, valid_bytes); try writer.file.sync(std.testing.io); }
const file = try tmp.dir.openFile(std.testing.io, "events.jsonl", .{ .mode = .read_write, }); var resumed = try ConversationWriter.init(alloc, file); defer resumed.deinit(); try std.testing.expectEqual(@as(u64, 3), resumed.last_seq); try std.testing.expectEqual(valid_bytes, resumed.committed_bytes); try std.testing.expectEqual(valid_bytes, try resumed.file.length(std.testing.io));}
test "conversation writer removes an unfinished turn before a torn final record" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var valid_bytes: u64 = 0; { const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit(); try writer.appendHistoryTurn(alloc, 10, .{ .assistant = .{ .user = .{ .text = @constCast("saved request") }, .assistant = @constCast("saved answer"), } }); valid_bytes = writer.committed_bytes; _ = try writer.append(alloc, 11, .{ .user = .{ .text = "unfinished request" } }); _ = try writer.append(alloc, 11, .{ .tool_call = .{ .call_id = "unfinished-call", .tool_name = "command", .arguments_json = "{}", } }); try file.writePositionalAll(std.testing.io, "{\"schema_version\":1", writer.committed_bytes); } { const file = try tmp.dir.openFile(std.testing.io, "events.jsonl", .{ .mode = .read_write }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit(); try std.testing.expectEqual(valid_bytes, writer.committed_bytes); try std.testing.expectEqual(@as(u64, 3), writer.last_seq); try std.testing.expectEqual(@as(usize, 0), writer.pendingToolCallCount()); try writer.appendHistoryTurn(alloc, 12, .{ .assistant = .{ .user = .{ .text = @constCast("next request") }, .assistant = @constCast("next answer"), } }); } const file = try tmp.dir.openFile(std.testing.io, "events.jsonl", .{ .mode = .read_write }); var reopened = try ConversationWriter.init(alloc, file); defer reopened.deinit(); try std.testing.expectEqual(@as(u64, 6), reopened.last_seq);}
test "conversation writer initialization leaves file ownership with caller on failure" { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true }); try file.writeStreamingAll(std.testing.io, "invalid\n"); var failed = false; if (ConversationWriter.init(std.testing.allocator, file)) |value| { var writer = value; writer.deinit(); } else |_| { failed = true; } const still_open = std.c.fcntl(file.handle, std.c.F.GETFD) != -1; if (still_open) file.close(std.testing.io); try std.testing.expect(failed); try std.testing.expect(still_open);}
test "conversation writer flattens a canonical history turn" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events.jsonl", .{ .read = true, .truncate = true, }); var writer = try ConversationWriter.init(alloc, file); defer writer.deinit();
var calls = [_]types.ToolCall{.{ .id = "call-shell", .name = "shell", .arguments_json = "{\"command\":\"printf done\"}", .provider_result = "{\"native\":true}", .provenance = .provider_executed, }}; var feedback = [_][]u8{@constCast("inspect the result")}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("call-shell"), .tool_name = @constCast("shell"), .status = .success, .output = @constCast("done"), .output_handle = @constCast("result-shell.txt"), .output_bytes = 4, .stored_output_bytes = 4, .permission_feedback = &feedback, .command_output_replay = .{ .available = .{ .handle = @constCast("fx-command-replay-test.bin"), .framed_bytes = 42, } }, }}; var steps = [_]types.ToolExecutionStep{.{ .assistant = @constCast("Running it."), .tool_calls = &calls, .tool_results = &results, }}; var images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/session/images/image-1-aaaaaaaaaaaaaaaa.bin"), .media_type = @constCast("image/png"), .snapshot_path = @constCast("/tmp/session/images/image-1-aaaaaaaaaaaaaaaa.bin"), .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; try writer.appendHistoryTurn(alloc, 10, .{ .assistant = .{ .user = .{ .text = @constCast("Run the command."), .images = &images, }, .assistant = @constCast("It completed."), .execution = .{ .tool_steps = &steps }, } });
const bytes = try writer.readAllForTest(alloc); defer alloc.free(bytes); try std.testing.expectEqual(@as(usize, 6), std.mem.count(u8, bytes, "\n")); try std.testing.expectEqual(@as(u64, 6), writer.last_seq); try std.testing.expectEqual(@as(usize, 0), writer.pendingToolCallCount()); try std.testing.expect(std.mem.find( u8, bytes, "\"snapshot_path\":\"images/image-1-aaaaaaaaaaaaaaaa.bin\"", ) != null); try std.testing.expect(std.mem.find( u8, bytes, "\"snapshot_path\":\"/tmp/session/images", ) == null);}
test "legacy import recovery restores old authority or keeps published current data" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc);
var initial = try testState(alloc, "legacy-import-recovery", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); const turn = try session.makeAssistantTurn(alloc, "question", "answer"); defer session.freeHistoryTurn(alloc, turn); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = turn, } }, 20); }
{ var writable = try temp.root.openWritableSessionDir( alloc, initial.id, lock_deadline_ms, ); defer writable.deinit(alloc); const original = try readManagedFileAlloc( alloc, &writable.dir, events_file, session_event.event_frame_max_bytes, ); defer alloc.free(original); try writable.dir.dir.rename( events_file, writable.dir.dir, conversation_migration_backup_file, io_mod.getIo(), ); var replacement = try createManagedFile(&writable.dir, events_file); try replacement.writeStreamingAll(io_mod.getIo(), "new-but-unpublished\n"); replacement.close(io_mod.getIo()); var abandoned = try createManagedFile( &writable.dir, conversation_migration_temp_file, ); abandoned.close(io_mod.getIo()); try io_mod.durableReplaceVerified( alloc, &writable.dir, manifest_file, "{\"schema_version\":3}\n", );
try recoverInterruptedLegacyImport(alloc, &writable); const restored = try readManagedFileAlloc( alloc, &writable.dir, events_file, session_event.event_frame_max_bytes, ); defer alloc.free(restored); try std.testing.expectEqualStrings(original, restored); try std.testing.expect(!try legacyImportRecoveryNeeded(&writable.dir)); try std.testing.expect(!try entryExists( &writable.dir, conversation_migration_temp_file, )); }
// Recreate a current-format session to prove that a crash after metadata // publication keeps the new log and only removes the old backup. try temp.root.sessions.?.dir.deleteTree(io_mod.getIo(), initial.id); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); } { var writable = try temp.root.openWritableSessionDir( alloc, initial.id, lock_deadline_ms, ); defer writable.deinit(alloc); const current = try readManagedFileAlloc( alloc, &writable.dir, events_file, session_event.event_frame_max_bytes, ); defer alloc.free(current); var backup = try createManagedFile( &writable.dir, conversation_migration_backup_file, ); try backup.writeStreamingAll(io_mod.getIo(), "old\n"); backup.close(io_mod.getIo());
try recoverInterruptedLegacyImport(alloc, &writable); const retained = try readManagedFileAlloc( alloc, &writable.dir, events_file, session_event.event_frame_max_bytes, ); defer alloc.free(retained); try std.testing.expectEqualStrings(current, retained); try std.testing.expect(!try legacyImportRecoveryNeeded(&writable.dir)); }}
test "legacy import migrates permissions before publishing controls" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "legacy-permission-import", 20); defer initial.deinit(alloc); initial.permission_state.version = 1; const history = [_]session.HistoryTurn{.{ .assistant = .{ .user = .{ .text = @constCast("existing question") }, .assistant = @constCast("existing answer"), } }}; initial.history = try session.snapshotOwnedContextHistory(alloc, &history, 0, 0); try temp.root.sessions.?.dir.createDir(std.testing.io, initial.id, private_dir_permissions); { var writable = try temp.root.openWritableSessionDir(alloc, initial.id, lock_deadline_ms); var writable_owned = true; defer if (writable_owned) writable.deinit(alloc); try io_mod.durableReplaceVerified(alloc, &writable.dir, manifest_file, "{\"schema_version\":3}\n"); var migrated = try importLegacySnapshotState(alloc, &writable, initial, 3, 0, null); writable_owned = false; defer migrated.deinit(alloc); try std.testing.expectEqual(@as(u8, 2), migrated.state.permission_state.version); try std.testing.expectEqual(@as(u8, 1), initial.permission_state.version); try std.testing.expectEqualStrings("existing answer", migrated.state.history[0].assistant.assistant); } var reopened = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer reopened.deinit(alloc); try std.testing.expectEqual(@as(u8, 2), reopened.state.permission_state.version); try std.testing.expectEqualStrings("existing question", reopened.state.history[0].assistant.user.text);}
test "legacy import omits empty file paths without losing execution history" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "legacy-empty-file-import", 20); defer initial.deinit(alloc); var calls = [_]types.ToolCall{.{ .id = "recorded-read", .name = "read_file", .arguments_json = "{\"path\":\"source.txt\"}" }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("recorded-read"), .tool_name = @constCast("read_file"), .status = .success, .output = @constCast("recorded output"), .output_bytes = 15, .stored_output_bytes = 15, }}; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; var files = [_]types.FileEvidence{ .{ .path = @constCast("source.txt"), .tool_call_id = @constCast("recorded-read"), .tool_name = @constCast("read_file") }, .{ .path = @constCast(""), .tool_call_id = @constCast("recorded-read"), .tool_name = @constCast("read_file") }, .{ .path = @constCast("before.txt"), .new_path = @constCast("after.txt"), .tool_call_id = @constCast("recorded-read"), .tool_name = @constCast("read_file") }, }; const history = [_]session.HistoryTurn{ .{ .assistant = .{ .user = .{ .text = @constCast("read existing file") }, .assistant = @constCast("read finished"), .execution = .{ .tool_steps = &steps, .files = &files } } }, .{ .interrupted = .{ .user = .{ .text = @constCast("cancelled follow-up") }, .execution = .{ .files = files[1..2] } } }, }; try std.testing.expectError(error.InvalidConversationEvent, session_event.encodeConversationFrame(alloc, .{ .seq = 1, .timestamp_ms = 20, .event = .{ .turn_completed = .{ .files = &files } }, })); initial.history = try session.snapshotOwnedContextHistory(alloc, &history, 0, 0); try temp.root.sessions.?.dir.createDir(std.testing.io, initial.id, private_dir_permissions); { var writable = try temp.root.openWritableSessionDir(alloc, initial.id, lock_deadline_ms); var writable_owned = true; defer if (writable_owned) writable.deinit(alloc); try io_mod.durableReplaceVerified(alloc, &writable.dir, manifest_file, "{\"schema_version\":3}\n"); var migrated = try importLegacySnapshotState(alloc, &writable, initial, 3, 0, null); writable_owned = false; defer migrated.deinit(alloc); try std.testing.expectEqual(@as(usize, 3), initial.history[0].assistant.execution.files.len); try std.testing.expectEqualStrings("", initial.history[0].assistant.execution.files[1].path); } var reopened = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer reopened.deinit(alloc); const execution = reopened.state.history[0].assistant.execution; try std.testing.expectEqual(@as(usize, 2), reopened.state.history.len); try std.testing.expectEqual(@as(usize, 1), execution.tool_steps.len); try std.testing.expectEqualStrings("recorded-read", execution.tool_steps[0].tool_calls[0].id); try std.testing.expectEqualStrings("recorded output", execution.tool_steps[0].tool_results[0].preview.?); try std.testing.expectEqual(@as(usize, 2), execution.files.len); try std.testing.expectEqualStrings("source.txt", execution.files[0].path); try std.testing.expectEqualStrings("before.txt", execution.files[1].path); try std.testing.expectEqualStrings("after.txt", execution.files[1].new_path.?); try std.testing.expectEqual(@as(usize, 0), reopened.state.history[1].interrupted.execution.files.len); try std.testing.expectEqualStrings("cancelled follow-up", reopened.state.history[1].interrupted.user.text);}
test "legacy import file projection cleans up allocation failures" { const Check = struct { fn run(alloc: Allocator) !void { const files = [_]types.FileEvidence{ .{ .path = @constCast(""), .new_path = @constCast("unusable.txt"), .tool_call_id = @constCast("old-call"), .tool_name = @constCast("read_file") }, .{ .path = @constCast("retained.txt"), .tool_call_id = @constCast("old-call"), .tool_name = @constCast("read_file") }, }; var history = [_]session.HistoryTurn{.{ .assistant = .{ .user = .{ .text = @constCast("request") }, .assistant = @constCast("response"), .execution = .{ .files = try types.dupeFileEvidenceSlice(alloc, &files) }, } }}; defer types.freeFileEvidenceSlice(alloc, history[0].assistant.execution.files); try std.testing.expectEqual(@as(usize, 1), try discardEmptyLegacyFileEvidence(alloc, &history)); try std.testing.expectEqualStrings("retained.txt", history[0].assistant.execution.files[0].path); } }; try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});}
test "legacy import preserves the active retained tail and complete archive" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "legacy-retained-tail", 10); defer initial.deinit(alloc); const png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jP0cAAAAASUVORK5CYII="; var images = [_]types.ToolImage{.{ .data = @constCast(png), .mime_type = @constCast("image/png") }}; var calls = [_]types.ToolCall{.{ .id = "image-call", .name = "mcp_browser", .arguments_json = "{}" }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("image-call"), .tool_name = @constCast("mcp_browser"), .status = .success, .output = @constCast(""), .output_bytes = 0, .stored_output_bytes = 0, .tool_images = &images, }}; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; const history = [_]session.HistoryTurn{ .{ .assistant = .{ .user = .{ .text = @constCast("removed request") }, .assistant = @constCast("removed answer") } }, .{ .compacted_summary = .{ .summary = @constCast("<context_handoff>earlier summary</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1 } }, .{ .assistant = .{ .user = .{ .text = @constCast("retained request") }, .assistant = @constCast("retained answer"), .execution = .{ .tool_steps = &steps } } }, .{ .compacted_summary = .{ .summary = @constCast("<context_handoff>active summary</context_handoff>"), .removed_turn_count = 1, .compaction_count = 2 } }, .{ .assistant = .{ .user = .{ .text = @constCast("later request") }, .assistant = @constCast("later answer") } }, }; initial.history = try session.snapshotOwnedContextHistory(alloc, &history, 0, 0); initial.context_history_start = 3; try temp.root.sessions.?.dir.createDir(std.testing.io, initial.id, private_dir_permissions); { var writable = try temp.root.openWritableSessionDir(alloc, initial.id, lock_deadline_ms); var writable_owned = true; defer if (writable_owned) writable.deinit(alloc); try io_mod.durableReplaceVerified(alloc, &writable.dir, manifest_file, "{\"schema_version\":3}\n"); var migrated = try importLegacySnapshotState(alloc, &writable, initial, 3, 0, null); writable_owned = false; defer migrated.deinit(alloc); try std.testing.expectEqual(@as(usize, 3), migrated.state.history.len); try std.testing.expectEqualStrings(history[3].compacted_summary.summary, migrated.state.history[0].compacted_summary.summary); try std.testing.expectEqualStrings("retained answer", migrated.state.history[1].assistant.assistant); try std.testing.expectEqualStrings("later answer", migrated.state.history[2].assistant.assistant); const archive = try loadConversationArchive(alloc, &migrated.log.dir); defer session.freeHistoryTurnSlice(alloc, archive); try std.testing.expectEqual(history.len, archive.len); try std.testing.expectEqualStrings("removed answer", archive[0].assistant.assistant); try std.testing.expectEqualStrings(history[1].compacted_summary.summary, archive[1].compacted_summary.summary); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 3), resumed.state.history.len); try std.testing.expectEqualStrings(history[3].compacted_summary.summary, resumed.state.history[0].compacted_summary.summary); try std.testing.expectEqualStrings("retained answer", resumed.state.history[1].assistant.assistant); try std.testing.expectEqualStrings("later answer", resumed.state.history[2].assistant.assistant); const result = resumed.state.history[1].assistant.execution.tool_steps[0].tool_results[0]; try std.testing.expectEqual(@as(usize, 0), result.tool_images.len); try std.testing.expect(result.tool_image_handle != null); const resumed_path = try io_mod.dirRealpathAlloc(alloc, resumed.log.dir.dir, "."); defer alloc.free(resumed_path); var capability = try session_child_store.SessionChildCapability.init(alloc, resumed.log.dir.dir, resumed_path, .read_only); defer capability.deinit(); const stored_images = try result_store.loadToolImages(alloc, &capability, result.tool_image_handle.?); defer types.freeToolImages(alloc, stored_images); try std.testing.expectEqual(@as(usize, 1), stored_images.len); try std.testing.expectEqualStrings(png, stored_images[0].data);}
test "legacy import preserves published events and active recovery after metadata sync failure" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "legacy-metadata-sync-failure", 10); defer initial.deinit(alloc); var legacy_usage = session_usage.Usage.initLegacy(); defer legacy_usage.deinit(alloc); initial.usage = try legacy_usage.snapshot(alloc); const history = try alloc.alloc(session.HistoryTurn, 1); history[0] = try session.makeAssistantTurn(alloc, "saved request", "saved answer"); initial.history = history; initial.recovery_checkpoint = try (session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the next request") }, .assistant_source = @constCast("partial"), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }).dupe(alloc); try temp.root.sessions.?.dir.createDir(std.testing.io, initial.id, private_dir_permissions); { var writable = try temp.root.openWritableSessionDir(alloc, initial.id, lock_deadline_ms); defer writable.deinit(alloc); try io_mod.durableReplaceVerified(alloc, &writable.dir, manifest_file, "{\"schema_version\":3}\n"); try io_mod.durableReplaceVerified(alloc, &writable.dir, events_file, "legacy-source\n"); const Fail = struct { fn syncDir(_: ?*anyopaque, _: std.Io.Dir) anyerror!void { return error.InjectedSyncFailure; } }; try std.testing.expectError(error.DurableReplacePostRenameFailed, importLegacySnapshotStateWithOps( alloc, &writable, initial, 3, 14, null, .{ .sync_dir = Fail.syncDir }, )); try std.testing.expect(try hasConversationMetadata(alloc, &writable.dir)); try std.testing.expect(try legacyImportRecoveryNeeded(&writable.dir)); try recoverInterruptedLegacyImport(alloc, &writable); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), resumed.state.history.len); try std.testing.expectEqualStrings("saved answer", resumed.state.history[0].assistant.assistant); try std.testing.expectEqual(@as(u64, 7), resumed.state.recovery_checkpoint.?.turn_id); try std.testing.expectEqual(session_usage.Availability.legacy, resumed.state.usage.?.billing);}
test "conversation storage creates only metadata and event log" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{ .iterate = true }); defer tmp.cleanup(); var directory = io_mod.VerifiedDir{ .dir = try tmp.dir.openDir(std.testing.io, ".", .{ .iterate = true }), }; defer directory.close();
var writer = try createConversationStorage(alloc, &directory, .{ .id = "session-1", .origin_workspace_root = "/tmp/origin", .workspace_root = "/tmp/current", .created_at_ms = 10, .updated_at_ms = 10, .conversation_language = "en", .provider = .gateway, .model = "openai/gpt-5.6", .effort = "high", .fast_mode = false, }); defer writer.deinit();
var count: usize = 0; var saw_events = false; var saw_metadata = false; var iterator = directory.dir.iterate(); while (try iterator.next(std.testing.io)) |entry| { count += 1; if (std.mem.eql(u8, entry.name, "events.jsonl")) saw_events = true; if (std.mem.eql(u8, entry.name, "session.json")) saw_metadata = true; } try std.testing.expectEqual(@as(usize, 2), count); try std.testing.expect(saw_events); try std.testing.expect(saw_metadata);}
test "root starts a cache-free conversation session" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "cache-free-session", 10); defer initial.deinit(alloc); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc);
var count: usize = 0; var saw_events = false; var saw_metadata = false; var saw_writer_lock = false; var saw_permissions = false; var saw_usage = false; var iterator = loaded.log.dir.dir.iterate(); while (try iterator.next(std.testing.io)) |entry| { count += 1; if (std.mem.eql(u8, entry.name, "events.jsonl")) saw_events = true; if (std.mem.eql(u8, entry.name, "session.json")) saw_metadata = true; if (std.mem.eql(u8, entry.name, "session.lock")) saw_writer_lock = true; if (std.mem.eql(u8, entry.name, "permissions.json")) saw_permissions = true; if (std.mem.eql(u8, entry.name, "usage-v2.json")) saw_usage = true; } try std.testing.expectEqual(@as(usize, 5), count); try std.testing.expect(saw_events); try std.testing.expect(saw_metadata); try std.testing.expect(saw_writer_lock); try std.testing.expect(saw_permissions); try std.testing.expect(saw_usage);}
test "root session creation stays outside discovery while preparing" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "unpublished-session", 10); defer initial.deinit(alloc); const Probe = struct { root: *Root, observed: bool = false, visible: bool = false, failure: ?anyerror = null,
fn lock(context: ?*anyopaque, _: LockKind) void { const self: *@This() = @ptrCast(@alignCast(context.?)); self.observed = true; self.visible = entryExists(&self.root.sessions.?, "unpublished-session") catch |err| { self.failure = err; return; }; } }; var probe = Probe{ .root = &temp.root }; var loaded = try temp.root.startConversationSession(alloc, initial, .{ .test_controls = .{ .context = &probe, .lock_fn = Probe.lock }, }); defer loaded.deinit(alloc); if (probe.failure) |err| return err; try std.testing.expect(probe.observed); try std.testing.expect(!probe.visible); try std.testing.expect(try entryExists(&temp.root.sessions.?, initial.id)); try std.testing.expectEqualStrings(initial.id, loaded.active_id); var saved = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer saved.deinit(alloc); try std.testing.expectEqualStrings(initial.id, saved.id);}
test "root session creation never replaces a racing target" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "racing-session", 10); defer initial.deinit(alloc); const Probe = struct { root: *Root, created: bool = false, failure: ?anyerror = null,
fn lock(context: ?*anyopaque, _: LockKind) void { const self: *@This() = @ptrCast(@alignCast(context.?)); self.root.sessions.?.dir.createDir(std.testing.io, "racing-session", private_dir_permissions) catch |err| { if (err != error.PathAlreadyExists) self.failure = err; return; }; self.created = true; } }; var probe = Probe{ .root = &temp.root }; if (temp.root.startConversationSession(alloc, initial, .{ .test_controls = .{ .context = &probe, .lock_fn = Probe.lock }, })) |value| { var loaded = value; loaded.deinit(alloc); return error.ExpectedSessionCollision; } else |err| try std.testing.expectEqual(error.SessionAlreadyExists, err); if (probe.failure) |err| return err; try std.testing.expect(probe.created); var preserved = try openSessionDir(&temp.root.sessions.?, initial.id, .read_only); defer preserved.close(); var entries = preserved.dir.iterate(); try std.testing.expect((try entries.next(std.testing.io)) == null);}
test "root session creation allocation failures leave no published state" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "allocation-session", 10); defer initial.deinit(alloc); var measured = std.testing.FailingAllocator.init(alloc, .{}); var successful = try temp.root.startConversationSession(measured.allocator(), initial, .{}); successful.deinit(measured.allocator()); try temp.root.sessions.?.dir.deleteTree(std.testing.io, initial.id); for (0..measured.alloc_index) |fail_index| { var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = fail_index }); if (temp.root.startConversationSession(failing.allocator(), initial, .{})) |value| { var loaded = value; loaded.deinit(failing.allocator()); return error.ExpectedAllocationFailure; } else |err| { try std.testing.expect(failing.has_induced_failure); // Existing allocating JSON writers surface allocation loss as WriteFailed. try std.testing.expect(err == error.OutOfMemory or err == error.WriteFailed); } try std.testing.expectEqual(failing.allocated_bytes, failing.freed_bytes); var entries = temp.root.sessions.?.dir.iterate(); try std.testing.expect((try entries.next(std.testing.io)) == null); }}
test "conversation writer appends without duplicating live history" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-append", 10); defer initial.deinit(alloc); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); const turn = types.HistoryTurn{ .assistant = .{ .user = .{ .text = @constCast("question") }, .assistant = @constCast("answer"), } };
_ = try loaded.appendEvent( alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = turn, } }, 20, );
try std.testing.expectEqual(@as(usize, 0), loaded.state.history.len); try std.testing.expectEqual(@as(u64, 3), loaded.conversation_writer.last_seq); var count: usize = 0; var iterator = loaded.log.dir.dir.iterate(); while (try iterator.next(std.testing.io)) |_| count += 1; try std.testing.expectEqual(@as(usize, 5), count);}
test "cache-free conversation session resumes from metadata and JSONL" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-resume", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent( alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("question") }, .assistant = @constCast("answer"), } }, } }, 20, ); }
var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), resumed.history.len); try std.testing.expectEqualStrings("question", resumed.history[0].assistant.user.text); try std.testing.expectEqualStrings("answer", resumed.history[0].assistant.assistant);}
test "conversation cancellation provenance survives checkpoint resume range and archive replay" { const alloc = std.testing.allocator; for (std.enums.values(types.CancellationOrigin)) |origin| { for ([_]bool{ false, true }) |checkpoint| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "cancellation-provenance", 10); defer initial.deinit(alloc); var steps = [_]types.ToolExecutionStep{.{ .assistant = @constCast("earlier reply") }}; const user: types.UserTurn = .{ .text = @constCast("request") }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); if (checkpoint) { _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("Earlier context."), .removed_turn_count = 0, .compaction_count = 1, }, .{ .user = user, .assistant = @constCast(""), .execution = .{ .tool_steps = &steps }, }, .{ .tool_steps = 1 }, 20); } _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .interrupted = .{ .user = user, .assistant = @constCast("partial reply"), .execution = .{ .tool_steps = if (checkpoint) &.{} else &steps }, .cancellation_origin = origin, } }, } }, 30); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, if (checkpoint) 2 else 1), resumed.state.history.len); const active = resumed.state.history[if (checkpoint) 1 else 0].interrupted; try std.testing.expectEqual(origin, active.cancellation_origin); try std.testing.expectEqual(types.InterruptedTerminalReason.cancelled, active.terminal_reason); try std.testing.expectEqualStrings("partial reply", active.assistant.?); const range = try loadConversationHistoryRange(alloc, &resumed.log.dir, 0, 1); defer session.freeHistoryTurnSlice(alloc, range); try std.testing.expectEqual(@as(usize, 1), range.len); try std.testing.expectEqual(origin, range[0].interrupted.cancellation_origin); try std.testing.expectEqualStrings("earlier reply", range[0].interrupted.execution.tool_steps[0].assistant.?); const archive = try loadConversationArchive(alloc, &resumed.log.dir); defer session.freeHistoryTurnSlice(alloc, archive); try std.testing.expectEqual(origin, archive[archive.len - 1].interrupted.cancellation_origin); var readonly = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer readonly.deinit(alloc); try std.testing.expectEqual(origin, readonly.history[readonly.history.len - 1].interrupted.cancellation_origin); } }}
test "conversation preserves standalone replies across completion and interruption" { const alloc = std.testing.allocator; const Outcome = enum { completed, cancelled, failed, active_call }; for ([_]bool{ false, true }) |with_replay| { for (std.enums.values(Outcome)) |outcome| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "standalone-replies", 10); defer initial.deinit(alloc); const replay: types.ProviderReplay = .{ .source = .{ .provider = .gateway, .model = "test-model" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"private state\"}]", }; var steps = [_]types.ToolExecutionStep{.{ .assistant = @constCast("earlier reply"), .provider_replay = if (with_replay) replay else null, }}; const user: types.UserTurn = .{ .text = @constCast("request") }; const turn: types.HistoryTurn = if (outcome == .completed) .{ .assistant = .{ .user = user, .assistant = @constCast("current reply"), .execution = .{ .tool_steps = &steps }, } } else .{ .interrupted = .{ .user = user, .assistant = @constCast("partial reply"), .execution = .{ .tool_steps = &steps }, .terminal_reason = if (outcome == .failed) .failed else .cancelled, .tool_call = if (outcome == .active_call) .{ .id = "pending", .name = "read_file", .arguments_json = "{\"path\":\"file\"}" } else null, } }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = turn, } }, 20); } var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), resumed.history.len); const execution = switch (resumed.history[0]) { .assistant => |entry| blk: { try std.testing.expectEqual(Outcome.completed, outcome); try std.testing.expectEqualStrings("current reply", entry.assistant); break :blk entry.execution; }, .interrupted => |entry| blk: { try std.testing.expectEqualStrings("partial reply", entry.assistant.?); try std.testing.expectEqual(outcome == .active_call, entry.tool_call != null); try std.testing.expectEqual(if (outcome == .failed) types.InterruptedTerminalReason.failed else .cancelled, entry.terminal_reason); break :blk entry.execution; }, else => return error.TestUnexpectedResult, }; try std.testing.expectEqual(@as(usize, 1), execution.tool_steps.len); try std.testing.expectEqualStrings("earlier reply", execution.tool_steps[0].assistant.?); try std.testing.expectEqual(with_replay, execution.tool_steps[0].provider_replay != null); if (execution.tool_steps[0].provider_replay) |saved| try std.testing.expectEqualStrings(replay.parts_json, saved.parts_json); try std.testing.expectEqual(@as(usize, 0), execution.tool_steps[0].tool_calls.len); } }}
test "standalone steering round trip preserves step boundaries" { try checkStandaloneSteering(false);}
test "standalone steering checkpoint preserves step boundaries" { try checkStandaloneSteering(true);}
fn checkStandaloneSteering(checkpoint: bool) !void { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |with_replay| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "standalone-steering", 10); defer initial.deinit(alloc); const replay: types.ProviderReplay = .{ .source = .{ .provider = .gateway, .model = "test-model" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"private state\"}]", }; var steps = [_]types.ToolExecutionStep{.{ .assistant = @constCast("earlier reply"), .provider_replay = if (with_replay) replay else null, }}; var steering = [_]types.PersistedSteering{.{ .text = @constCast("human update"), .after_tool_step_count = 1, }}; const user: types.UserTurn = .{ .text = @constCast("request") }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); if (checkpoint) { _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>Earlier reply.</context_handoff>"), .removed_turn_count = 0, .compaction_count = 1, }, .{ .user = user, .assistant = @constCast(""), .execution = .{ .tool_steps = &steps, .steering = &steering }, }, .{ .tool_steps = 1 }, 20); steering[0].after_tool_step_count = 0; } _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = user, .assistant = @constCast("final reply"), .execution = .{ .tool_steps = if (checkpoint) &.{} else &steps, .steering = &steering, }, } }, } }, 30); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, if (checkpoint) 2 else 1), resumed.state.history.len); const active = resumed.state.history[if (checkpoint) 1 else 0].assistant; try std.testing.expectEqualStrings("final reply", active.assistant); try std.testing.expectEqual(@as(usize, if (checkpoint) 0 else 1), active.execution.tool_steps.len); try std.testing.expectEqual(@as(usize, 1), active.execution.steering.len); try std.testing.expectEqual(@as(usize, if (checkpoint) 0 else 1), active.execution.steering[0].after_tool_step_count); try std.testing.expect(active.execution.steering[0].assistant_prefix == null); try std.testing.expectEqualStrings("human update", active.execution.steering[0].text); const archive = try loadConversationHistoryRange(alloc, &resumed.log.dir, 0, 1); defer session.freeHistoryTurnSlice(alloc, archive); const execution = archive[0].assistant.execution; try std.testing.expectEqual(@as(usize, 1), execution.tool_steps.len); try std.testing.expectEqualStrings("earlier reply", execution.tool_steps[0].assistant.?); try std.testing.expectEqual(with_replay, execution.tool_steps[0].provider_replay != null); if (execution.tool_steps[0].provider_replay) |saved| try std.testing.expectEqualStrings(replay.parts_json, saved.parts_json); try std.testing.expectEqual(@as(usize, 1), execution.steering.len); try std.testing.expectEqual(@as(usize, 1), execution.steering[0].after_tool_step_count); try std.testing.expect(execution.steering[0].assistant_prefix == null); }}
test "legacy steering prefix records retain their original boundary" { const alloc = std.testing.allocator; for ([_]u8{ 1, 2 }) |schema_version| { const bytes = try std.fmt.allocPrint( alloc, "{{\"schema_version\":{d},\"seq\":2,\"timestamp_ms\":10,\"event\":{{\"assistant\":{{\"text\":\"legacy prefix\"}}}}}}\n", .{schema_version}, ); defer alloc.free(bytes); var decoded = try session_event.decodeConversationFrame(alloc, bytes); defer decoded.deinit(); var progress: ConversationProgress = .{}; try progress.observe(1, .{ .user = .{ .text = "request" } }, null); try progress.observe(2, decoded.value.event, null); try progress.observe(3, .{ .steering = .{ .text = "human update" } }, null); try std.testing.expectEqual(@as(usize, 0), progress.point.tool_steps); try std.testing.expectEqual(@as(usize, 1), progress.point.steering); var builder = ConversationTurnBuilder.init(alloc); defer builder.deinit(); try builder.begin(.{ .text = "request" }); try builder.appendAssistant(decoded.value.event.assistant); try builder.appendSteering("human update"); const execution = try builder.takeExecution(&.{}, null); defer types.freeExecutionMemory(alloc, execution); try std.testing.expectEqual(@as(usize, 0), execution.tool_steps.len); try std.testing.expectEqual(@as(usize, 1), execution.steering.len); try std.testing.expectEqual(@as(usize, 0), execution.steering[0].after_tool_step_count); try std.testing.expectEqualStrings("legacy prefix", execution.steering[0].assistant_prefix.?); }}
test "takeExecution frees provider replay on allocation failure" { for ([_]bool{ false, true }) |standalone_response| try std.testing.checkAllAllocationFailures(std.testing.allocator, struct { fn run(alloc: Allocator, standalone: bool) !void { var builder = ConversationTurnBuilder.init(alloc); defer builder.deinit(); try builder.begin(.{ .text = "request" }); try builder.appendAssistant(.{ .text = "earlier reply", .standalone_response = standalone, .provider_replay = .{ .source = .{ .provider = .gateway, .model = "test-model" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"private state\"}]", }, }); try builder.appendSteering("human update"); const execution = try builder.takeExecution(&.{.{ .path = @constCast("result.txt"), .tool_call_id = @constCast("call-1"), .tool_name = @constCast("read_file"), }}, null); defer types.freeExecutionMemory(alloc, execution); try std.testing.expectEqual(@as(usize, 1), execution.tool_steps.len); try std.testing.expect(execution.tool_steps[0].provider_replay != null); try std.testing.expectEqual(@as(usize, 1), execution.steering.len); try std.testing.expectEqual(@as(usize, 1), execution.files.len); } }.run, .{standalone_response});}
test "standalone checkpoint boundaries do not create empty replies" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "standalone-checkpoint", 10); defer initial.deinit(alloc); var steps = [_]types.ToolExecutionStep{ .{ .assistant = @constCast("older reply") }, .{ .assistant = @constCast("recent reply") }, }; const user: types.UserTurn = .{ .text = @constCast("request") }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>Earlier context.</context_handoff>"), .removed_turn_count = 0, .compaction_count = 1, }, .{ .user = user, .assistant = @constCast(""), .execution = .{ .tool_steps = &steps } }, .{ .tool_steps = 1 }, 20); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = user, .assistant = @constCast("final reply"), .execution = .{ .tool_steps = steps[1..] } } }, } }, 30); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), resumed.state.history.len); const active = resumed.state.history[1].assistant; try std.testing.expectEqual(@as(usize, 1), active.execution.tool_steps.len); try std.testing.expectEqualStrings("recent reply", active.execution.tool_steps[0].assistant.?); try std.testing.expectEqualStrings("final reply", active.assistant); const archive = try loadConversationHistoryRange(alloc, &resumed.log.dir, 0, 1); defer session.freeHistoryTurnSlice(alloc, archive); try std.testing.expectEqual(@as(usize, 2), archive[0].assistant.execution.tool_steps.len); try std.testing.expectEqualStrings("older reply", archive[0].assistant.execution.tool_steps[0].assistant.?); try std.testing.expectEqualStrings("recent reply", archive[0].assistant.execution.tool_steps[1].assistant.?);}
test "cache-free writable resume continues the conversation sequence" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-write-resume", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("first question") }, .assistant = @constCast("first answer"), } }, } }, 20); } { var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(u64, 3), resumed.conversation_writer.last_seq); _ = try resumed.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 2, .total_output_tokens = 2, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("second question") }, .assistant = @constCast("second answer"), } }, } }, 30); }
var loaded = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), loaded.history.len); try std.testing.expectEqualStrings("second answer", loaded.history[1].assistant.assistant);}
test "cache-free metadata changes rewrite metadata without conversation records" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-metadata", 10); defer initial.deinit(alloc); var committed_bytes: u64 = 0; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); committed_bytes = loaded.conversation_writer.committed_bytes; _ = try loaded.appendEvent(alloc, .{ .preferences_changed = .{ .model = @constCast("updated/model"), .fast_mode = true, } }, 20); try std.testing.expect(try loaded.renameConversation(alloc, "renamed session")); const title = (try loaded.conversationTitle(alloc)) orelse return error.TestExpectedConversationTitle; defer alloc.free(title); try std.testing.expectEqualStrings("renamed session", title); try std.testing.expectEqual( committed_bytes, loaded.conversation_writer.committed_bytes, ); }
var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqualStrings("updated/model", resumed.preferences.model); try std.testing.expect(resumed.preferences.fast_mode);}
test "cache-free usage checkpoints stay outside conversation history" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-usage", 10); defer initial.deinit(alloc); var usage = session_usage.Usage.initFresh(); defer usage.deinit(alloc); try usage.recordCommittedLines(9, 4); var snapshot = try usage.snapshot(alloc); defer snapshot.deinit(alloc); var committed_bytes: u64 = 0; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); committed_bytes = loaded.conversation_writer.committed_bytes; _ = try loaded.appendEvent(alloc, .{ .usage_checkpointed = .{ .usage = snapshot }, }, 20); try std.testing.expectEqual( committed_bytes, loaded.conversation_writer.committed_bytes, ); try std.testing.expect(loaded.freshly_started); }
var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(u64, 9), resumed.usage.?.lines_added); try std.testing.expectEqual(@as(u64, 4), resumed.usage.?.lines_removed); try std.testing.expectEqual(@as(usize, 0), resumed.history.len);}
test "cache-free recovery checkpoint resumes and clears independently" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-recovery", 10); defer initial.deinit(alloc); const checkpoint = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = checkpoint }, }, 20); } { var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(u64, 7), resumed.state.recovery_checkpoint.?.turn_id); _ = try resumed.appendEvent(alloc, .{ .recovery_checkpoint_cleared = .{}, }, 30); }
var cleared = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer cleared.deinit(alloc); try std.testing.expectEqual(@as(?session_codec.RecoveryCheckpoint, null), cleared.recovery_checkpoint);}
test "recovery checkpoint spills oversized tool outputs and reload restores them" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-recovery-spill", 10); defer initial.deinit(alloc);
const big = try alloc.alloc(u8, 8 * 1024); defer alloc.free(big); @memset(big, 'x'); @memcpy(big[big.len - 9 ..], "ENDMARKER");
var calls = [_]types.ToolCall{ .{ .id = "call-big", .name = "command", .arguments_json = "{}" }, .{ .id = "call-small", .name = "command", .arguments_json = "{}" }, .{ .id = "call-truncated", .name = "command", .arguments_json = "{}" }, }; var results = [_]types.PersistedToolResult{ .{ .tool_call_id = @constCast("call-big"), .tool_name = @constCast("command"), .status = .success, .output = big, .output_bytes = big.len, .stored_output_bytes = big.len, }, .{ .tool_call_id = @constCast("call-small"), .tool_name = @constCast("command"), .status = .success, .output = @constCast("tiny output"), .output_bytes = 11, .stored_output_bytes = 11, }, .{ .tool_call_id = @constCast("call-truncated"), .tool_name = @constCast("command"), .status = .success, .output = @constCast("TRUNCATED-INLINE"), .output_handle = @constCast("result-command-0000000000000000-0000000000000000.txt"), .preview = @constCast("truncated preview"), .output_bytes = 100000, .stored_output_bytes = 100000, .truncated = true, }, }; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; const checkpoint = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .execution = .{ .tool_steps = &steps }, .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = checkpoint }, }, 20); const bytes = try readManagedFileAlloc( alloc, &loaded.log.dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes + 128, ); defer alloc.free(bytes); // The spilled body and the truncated inline body stay out of the file. try std.testing.expect(std.mem.find(u8, bytes, "ENDMARKER") == null); try std.testing.expect(std.mem.find(u8, bytes, "TRUNCATED-INLINE") == null); try std.testing.expect(std.mem.find(u8, bytes, "result-command-") != null); try std.testing.expect(std.mem.find(u8, bytes, "tiny output") != null); try std.testing.expect(bytes.len < big.len); } { var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); const restored = resumed.state.recovery_checkpoint.?; const restored_results = restored.execution.tool_steps[0].tool_results; try std.testing.expectEqual(big.len, restored_results[0].output.len); try std.testing.expect(std.mem.endsWith(u8, restored_results[0].output, "ENDMARKER")); try std.testing.expectEqualStrings("tiny output", restored_results[1].output); // Truncated results restore as their stored-result placeholder. try std.testing.expect(std.mem.find(u8, restored_results[2].output, "read_tool_result") != null); }}
test "oversized recovery checkpoint keeps the previous durable checkpoint" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-recovery-oversized", 10); defer initial.deinit(alloc); const first = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; const oversized_source = try alloc.alloc(u8, session_codec.max_recovery_checkpoint_bytes); defer alloc.free(oversized_source); @memset(oversized_source, 'a'); var oversized = first; oversized.turn_id = 9; oversized.assistant_source = oversized_source; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = first }, }, 20); // The oversized checkpoint must not fail the session update. _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = oversized }, }, 21); // In-memory state still advances to the freshest checkpoint. try std.testing.expectEqual(@as(u64, 9), loaded.state.recovery_checkpoint.?.turn_id); const bytes = try readManagedFileAlloc( alloc, &loaded.log.dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes + 128, ); defer alloc.free(bytes); // The durable file keeps the last persistable checkpoint. try std.testing.expect(std.mem.find(u8, bytes, "\"turn_id\":7") != null); try std.testing.expect(std.mem.find(u8, bytes, "\"turn_id\":9") == null); } { var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(u64, 7), resumed.state.recovery_checkpoint.?.turn_id); }}
test "tool-result spill keeps an over-cap checkpoint persistable and resumable" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-recovery-rescue", 10); defer initial.deinit(alloc);
// Seventeen 4 MiB results would serialize past the 64 MiB cap if kept inline. const result_count = 17; const output_len = 4 * 1024 * 1024; var outputs: [result_count][]u8 = undefined; for (&outputs, 0..) |*slot, index| { slot.* = try alloc.alloc(u8, output_len); @memset(slot.*, @as(u8, @intCast('a' + index))); } defer for (outputs) |bytes| alloc.free(bytes);
var calls: [result_count]types.ToolCall = undefined; var results: [result_count]types.PersistedToolResult = undefined; for (&calls, &results, 0..) |*call, *result, index| { const id = try std.fmt.allocPrint(alloc, "call-big-{d}", .{index}); call.* = .{ .id = id, .name = "command", .arguments_json = "{}" }; result.* = .{ .tool_call_id = id, .tool_name = @constCast("command"), .status = .success, .output = outputs[index], .output_bytes = output_len, .stored_output_bytes = output_len, }; } defer for (calls) |call| alloc.free(call.id); var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; const checkpoint = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .execution = .{ .tool_steps = &steps }, .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = checkpoint }, }, 20); const bytes = try readManagedFileAlloc( alloc, &loaded.log.dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes + 128, ); defer alloc.free(bytes); try std.testing.expect(bytes.len < 1024 * 1024); try std.testing.expect(std.mem.find(u8, bytes, "result-command-") != null); } { var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); const restored = resumed.state.recovery_checkpoint.?.execution.tool_steps[0].tool_results; try std.testing.expectEqual(result_count, restored.len); for (restored, 0..) |result, index| { try std.testing.expectEqual(output_len, result.output.len); try std.testing.expectEqual(@as(u8, @intCast('a' + index)), result.output[0]); try std.testing.expectEqual(@as(u8, @intCast('a' + index)), result.output[output_len - 1]); } }}
test "recovery checkpoint spill failure keeps the result inline" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-recovery-spill-failure", 10); defer initial.deinit(alloc);
const big = try alloc.alloc(u8, 8 * 1024); defer alloc.free(big); @memset(big, 'y'); @memcpy(big[big.len - 9 ..], "ENDMARKER");
var calls = [_]types.ToolCall{.{ .id = "call-big", .name = "command", .arguments_json = "{}" }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("call-big"), .tool_name = @constCast("command"), .status = .success, .output = big, .output_bytes = big.len, .stored_output_bytes = big.len, }}; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; const checkpoint = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .execution = .{ .tool_steps = &steps }, .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc);
// Block the result store: tool-results exists as a regular file. const session_path = try std.fs.path.join(alloc, &.{ temp.home, ".fx", "sessions", initial.id }); defer alloc.free(session_path); var session_dir = try std.Io.Dir.openDirAbsolute(io_mod.getIo(), session_path, .{}); defer session_dir.close(io_mod.getIo()); var bogus = try session_dir.createFile(io_mod.getIo(), "tool-results", .{}); bogus.close(io_mod.getIo());
// The store failure must not block the checkpoint write. _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = checkpoint }, }, 20); const bytes = try readManagedFileAlloc( alloc, &loaded.log.dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes + 128, ); defer alloc.free(bytes); try std.testing.expect(std.mem.find(u8, bytes, "ENDMARKER") != null);}
test "committed conversation supersedes recovery after interrupted cleanup" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-stale-recovery", 10); defer initial.deinit(alloc); const checkpoint = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = checkpoint }, }, 20); // Model a process death after the durable conversation append but // before recovery cleanup by using the writer at that boundary. try loaded.conversation_writer.appendHistoryTurn(alloc, 30, .{ .assistant = .{ .user = checkpoint.user, .assistant = @constCast("finished answer"), } }); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), resumed.state.history.len); try std.testing.expectEqual(@as(?session_codec.RecoveryCheckpoint, null), resumed.state.recovery_checkpoint);}
fn expectSameHistory(expected: []const session.HistoryTurn, actual: []const session.HistoryTurn) !void { try std.testing.expectEqual(expected.len, actual.len); for (expected, actual) |want, got| { try std.testing.expectEqual(std.meta.activeTag(want), std.meta.activeTag(got)); switch (want) { .assistant => |w| { const g = got.assistant; try std.testing.expectEqualStrings(w.user.text, g.user.text); try std.testing.expectEqualStrings(w.assistant, g.assistant); try std.testing.expectEqual(w.execution.tool_steps.len, g.execution.tool_steps.len); for (w.execution.tool_steps, g.execution.tool_steps) |want_step, got_step| { try std.testing.expectEqual(want_step.tool_calls.len, got_step.tool_calls.len); for (want_step.tool_calls, got_step.tool_calls) |want_call, got_call| { try std.testing.expectEqualStrings(want_call.id, got_call.id); try std.testing.expectEqualStrings(want_call.arguments_json, got_call.arguments_json); } try std.testing.expectEqual(want_step.tool_results.len, got_step.tool_results.len); for (want_step.tool_results, got_step.tool_results) |want_result, got_result| { try std.testing.expectEqualStrings(want_result.tool_call_id, got_result.tool_call_id); try std.testing.expectEqualStrings(want_result.output, got_result.output); try std.testing.expectEqual(want_result.status, got_result.status); } } }, .interrupted => |w| { const g = got.interrupted; try std.testing.expectEqualStrings(w.user.text, g.user.text); try std.testing.expectEqual(w.terminal_reason, g.terminal_reason); }, .compacted_summary => |w| { const g = got.compacted_summary; try std.testing.expectEqualStrings(w.summary, g.summary); try std.testing.expectEqual(w.removed_turn_count, g.removed_turn_count); }, } }}
/// Duplicates a history so it outlives the resumed session it came from.fn dupeHistory(alloc: Allocator, history: []const session.HistoryTurn) ![]session.HistoryTurn { const copy = try alloc.alloc(session.HistoryTurn, history.len); var initialized: usize = 0; errdefer { for (copy[0..initialized]) |turn| session.freeHistoryTurn(alloc, turn); alloc.free(copy); } for (history, 0..) |turn, index| { copy[index] = try session.dupeHistoryTurn(alloc, turn); initialized += 1; } return copy;}
fn cacheFileExists(alloc: Allocator, temp: *TempRoot, session_id: []const u8) !bool { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}/{s}", .{ sessions, session_id, history_snapshot.file_name }); defer alloc.free(path); var file = std.Io.Dir.openFileAbsolute(io_mod.getIo(), path, .{}) catch |err| switch (err) { error.FileNotFound => return false, else => return err, }; file.close(io_mod.getIo()); return true;}
fn cacheFileSize(alloc: Allocator, temp: *TempRoot, session_id: []const u8) !u64 { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}/{s}", .{ sessions, session_id, history_snapshot.file_name }); defer alloc.free(path); var file = try std.Io.Dir.openFileAbsolute(io_mod.getIo(), path, .{}); defer file.close(io_mod.getIo()); return file.length(io_mod.getIo());}
fn deleteCacheFileForTest(alloc: Allocator, temp: *TempRoot, session_id: []const u8) !void { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}/{s}", .{ sessions, session_id, history_snapshot.file_name }); defer alloc.free(path); try std.Io.Dir.deleteFileAbsolute(io_mod.getIo(), path);}
fn buildTwoTurnSession(alloc: Allocator, temp: *TempRoot, id: []const u8) !void { var initial = try testState(alloc, id, 10); defer initial.deinit(alloc); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 1, .total_output_tokens = 2, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("first question") }, .assistant = @constCast("first answer"), } }, } }, 20); var calls = [_]types.ToolCall{.{ .id = "call-1", .name = "command", .arguments_json = "{\"command\":\"ls\"}" }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("call-1"), .tool_name = @constCast("command"), .status = .success, .output = @constCast("listed"), .output_handle = @constCast("result-1.log"), .output_bytes = 6, .stored_output_bytes = 6, }}; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 3, .total_output_tokens = 4, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("second question") }, .assistant = @constCast("second answer"), .execution = .{ .tool_steps = &steps }, } }, } }, 30);}
test "history snapshot cache builds on writable resume and replays identically" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); const id = "snapshot-identical"; try buildTwoTurnSession(alloc, &temp, id); try std.testing.expect(!try cacheFileExists(alloc, &temp, id));
// First resume builds the cache during the writable-open scan. var first = try temp.root.resumeForWrite(alloc, id, .{}); const cached_history = try dupeHistory(alloc, first.state.history); defer session.freeHistoryTurnSlice(alloc, cached_history); first.deinit(alloc); try std.testing.expect(try cacheFileExists(alloc, &temp, id));
// Second resume reads the cache. Verify via the source that the snapshot // verifies against the log, then compare history content. { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}", .{ sessions, id }); defer alloc.free(path); var dir = try std.Io.Dir.openDirAbsolute(io_mod.getIo(), path, .{}); defer dir.close(io_mod.getIo()); var verified_dir: io_mod.VerifiedDir = .{ .dir = dir }; var log_file = try dir.openFile(io_mod.getIo(), events_file, .{}); defer log_file.close(io_mod.getIo()); const log_len = try log_file.length(io_mod.getIo()); var verified = (try history_snapshot.openAndVerify(alloc, &verified_dir, id, log_file, log_len)).?; defer verified.deinit(alloc); try std.testing.expectEqual(log_len, verified.covered_log_bytes); } var second = try temp.root.resumeForWrite(alloc, id, .{}); const snapshot_history = try dupeHistory(alloc, second.state.history); defer session.freeHistoryTurnSlice(alloc, snapshot_history); // A new turn commits only to the log; the next resume splices the log // suffix onto the cached prefix and re-tees the cache forward. const cache_size_before = try cacheFileSize(alloc, &temp, id); var third_state = try testState(alloc, id, 40); defer third_state.deinit(alloc); _ = try second.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = third_state.conversation_language, .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("mirror question") }, .assistant = @constCast("mirror answer"), } }, } }, 40); try std.testing.expectEqual(cache_size_before, try cacheFileSize(alloc, &temp, id)); second.deinit(alloc); try expectSameHistory(cached_history, snapshot_history);
var third = try temp.root.resumeForWrite(alloc, id, .{}); // The suffix replay sees the appended turn without a rebuild. try expectSameHistory(snapshot_history, third.state.history[0..2]); try std.testing.expectEqual(@as(usize, 3), third.state.history.len); try std.testing.expectEqualStrings("mirror question", third.state.history[2].assistant.user.text); try std.testing.expect(try cacheFileSize(alloc, &temp, id) > cache_size_before); third.deinit(alloc);
// Deleting the cache falls back to the full log replay with identical // content, and the writable resume rebuilds the cache. try deleteCacheFileForTest(alloc, &temp, id); var fourth = try temp.root.resumeForWrite(alloc, id, .{}); defer fourth.deinit(alloc); try expectSameHistory(snapshot_history, fourth.state.history[0..2]); try std.testing.expectEqualStrings("mirror question", fourth.state.history[2].assistant.user.text); try std.testing.expect(try cacheFileExists(alloc, &temp, id));}
test "history snapshot resume reads cached content over a middle-rewritten log" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); const id = "snapshot-consumed"; try buildTwoTurnSession(alloc, &temp, id);
var first = try temp.root.resumeForWrite(alloc, id, .{}); first.deinit(alloc); try std.testing.expect(try cacheFileExists(alloc, &temp, id));
// Rewrite a middle log line in place, preserving total length and the // final line: the watermark cannot see it, and this test pins that a // verified cache serves its own content for the covered prefix. { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}/{s}", .{ sessions, id, "events.jsonl" }); defer alloc.free(path); var file = try std.Io.Dir.openFileAbsolute(io_mod.getIo(), path, .{ .mode = .read_write }); defer file.close(io_mod.getIo()); const len = try file.length(io_mod.getIo()); const bytes = try alloc.alloc(u8, @intCast(len)); defer alloc.free(bytes); _ = try file.readPositionalAll(io_mod.getIo(), bytes, 0); const at = std.mem.find(u8, bytes, "first question").?; @memcpy(bytes[at..][0.."first question".len], "first qu3stion"); try file.writePositionalAll(io_mod.getIo(), bytes, 0); }
var second = try temp.root.resumeForWrite(alloc, id, .{}); defer second.deinit(alloc); try std.testing.expectEqualStrings("first question", second.state.history[0].assistant.user.text);}
test "history snapshot tolerates torn cache tails and spliced log growth" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); const id = "snapshot-torn-and-grown"; try buildTwoTurnSession(alloc, &temp, id);
var first = try temp.root.resumeForWrite(alloc, id, .{}); const baseline = try dupeHistory(alloc, first.state.history); defer session.freeHistoryTurnSlice(alloc, baseline); first.deinit(alloc); try std.testing.expect(try cacheFileExists(alloc, &temp, id));
// Tear the cache tail mid-frame: the valid prefix shortens and the resume // splices the uncovered log suffix. { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}/{s}", .{ sessions, id, history_snapshot.file_name }); defer alloc.free(path); var file = try std.Io.Dir.openFileAbsolute(io_mod.getIo(), path, .{ .mode = .read_write }); defer file.close(io_mod.getIo()); const len = try file.length(io_mod.getIo()); try file.setLength(io_mod.getIo(), len - 3); } var torn = try temp.root.resumeForWrite(alloc, id, .{}); const torn_history = try dupeHistory(alloc, torn.state.history); defer session.freeHistoryTurnSlice(alloc, torn_history); torn.deinit(alloc); try expectSameHistory(baseline, torn_history);
// A third turn committed through the live session lands only in the log // (nothing mirrors at commit time); the next resume replays it as a log // suffix over the cached prefix and re-tees the cache forward. { var loaded = try temp.root.resumeForWrite(alloc, id, .{}); var initial = try testState(alloc, id, 40); defer initial.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 5, .total_output_tokens = 6, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("third question") }, .assistant = @constCast("third answer"), } }, } }, 50); loaded.deinit(alloc); } var rebuilt = try temp.root.resumeForWrite(alloc, id, .{}); const rebuilt_history = try dupeHistory(alloc, rebuilt.state.history); defer session.freeHistoryTurnSlice(alloc, rebuilt_history); rebuilt.deinit(alloc); try std.testing.expectEqual(@as(usize, 3), rebuilt_history.len); try expectSameHistory(baseline, rebuilt_history[0..2]); try std.testing.expectEqualStrings("third question", rebuilt_history[2].assistant.user.text);
// Corrupt the last cache frame's payload: CRC mismatch shortens the // prefix, the log suffix covers the rest. { const sessions = try profile_paths.sessionsDir(alloc, temp.home); defer alloc.free(sessions); const path = try std.fmt.allocPrint(alloc, "{s}/{s}/{s}", .{ sessions, id, history_snapshot.file_name }); defer alloc.free(path); var file = try std.Io.Dir.openFileAbsolute(io_mod.getIo(), path, .{ .mode = .read_write }); defer file.close(io_mod.getIo()); const len = try file.length(io_mod.getIo()); try file.writePositionalAll(io_mod.getIo(), "Z", len - 2); } var grown = try temp.root.resumeForWrite(alloc, id, .{}); defer grown.deinit(alloc); try std.testing.expectEqual(@as(usize, 3), grown.state.history.len); try std.testing.expectEqualStrings("third question", grown.state.history[2].assistant.user.text);}
test "history snapshot replays compacted sessions identically" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); const id = "snapshot-compacted"; var initial = try testState(alloc, id, 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("old question") }, .assistant = @constCast("old answer"), } }, } }, 20); _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>earlier work summarized</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, }, null, null, 30); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("new question") }, .assistant = @constCast("new answer"), } }, } }, 40); } var first = try temp.root.resumeForWrite(alloc, id, .{}); const uncached = try dupeHistory(alloc, first.state.history); defer session.freeHistoryTurnSlice(alloc, uncached); first.deinit(alloc); try std.testing.expect(try cacheFileExists(alloc, &temp, id));
var second = try temp.root.resumeForWrite(alloc, id, .{}); defer second.deinit(alloc); try expectSameHistory(uncached, second.state.history); try std.testing.expect(second.state.history[0] == .compacted_summary);}
test "mid-turn checkpoint resumes only its suffix while preserving archived tool execution" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |finish_turn| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-active-checkpoint", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); for ([_][]const u8{ "first-summarized-call", "second-summarized-call" }, 0..) |call_id, index| { var calls = [_]types.ToolCall{.{ .id = call_id, .name = "command", .arguments_json = "{}" }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast(call_id), .tool_name = @constCast("command"), .status = .success, .output = @constCast("summarized output"), .output_handle = @constCast("saved-result.log"), .output_bytes = 16, .stored_output_bytes = 16, }}; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>prior work completed</context_handoff>"), .removed_turn_count = 0, .compaction_count = index + 1, }, .{ .user = .{ .text = @constCast("complete the request") }, .assistant = @constCast(""), .execution = .{ .tool_steps = &steps }, }, null, 20); } if (finish_turn) { _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("complete the request") }, .assistant = @constCast("remaining answer"), } }, } }, 30); } else { const writer = &loaded.conversation_writer; const partial_suffix = try session_event.encodeConversationFrame(alloc, .{ .seq = writer.last_seq + 1, .timestamp_ms = 30, .event = .{ .assistant = .{ .text = "unfinished suffix" } }, }); defer alloc.free(partial_suffix); try writer.file.writePositionalAll(std.testing.io, partial_suffix, writer.committed_bytes); try writer.file.writePositionalAll(std.testing.io, "{\"schema_version\":1", writer.committed_bytes + partial_suffix.len); } } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), resumed.state.history.len); try std.testing.expect(resumed.state.history[0] == .compacted_summary); const title = (try resumed.conversationTitle(alloc)).?; defer alloc.free(title); try std.testing.expectEqualStrings("complete the request", title); const suffix = resumed.state.history[1]; if (finish_turn) { try std.testing.expectEqualStrings("remaining answer", suffix.assistant.assistant); try std.testing.expectEqual(@as(usize, 0), suffix.assistant.execution.tool_steps.len); } else { try std.testing.expect(suffix == .interrupted); try std.testing.expectEqual(@as(usize, 0), suffix.interrupted.execution.tool_steps.len); } const archive = try loadConversationHistoryRange(alloc, &resumed.log.dir, 0, 1); defer session.freeHistoryTurnSlice(alloc, archive); try std.testing.expectEqual(@as(usize, 1), archive.len); const archived_execution = switch (archive[0]) { .assistant => |entry| entry.execution, .interrupted => |entry| entry.execution, .compacted_summary => return error.TestUnexpectedResult, }; try std.testing.expectEqual(@as(usize, 2), archived_execution.tool_steps.len); try std.testing.expectEqualStrings("first-summarized-call", archived_execution.tool_steps[0].tool_calls[0].id); try std.testing.expectEqualStrings("second-summarized-call", archived_execution.tool_steps[1].tool_calls[0].id); const full_archive = try loadConversationArchive(alloc, &resumed.log.dir); defer session.freeHistoryTurnSlice(alloc, full_archive); var raw_turns: usize = 0; for (full_archive) |turn| { if (turn != .compacted_summary) raw_turns += 1; } try std.testing.expectEqual(@as(usize, 1), raw_turns); const bytes = try resumed.conversation_writer.readAllForTest(alloc); defer alloc.free(bytes); try std.testing.expectEqual(@as(usize, 1), std.mem.count(u8, bytes, "\"user\":{")); }}
test "retained context checkpoint survives restart and appends each execution once" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |repeat| { for ([_]bool{ false, true }) |finish| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "retained-context-checkpoint", 10); defer initial.deinit(alloc); const ids = [_][]const u8{ "older-call", "recent-call", "newest-call", "final-call" }; var calls: [4][1]types.ToolCall = undefined; var results: [4][1]types.PersistedToolResult = undefined; var steps: [4]types.ToolExecutionStep = undefined; for (ids, 0..) |id, index| { calls[index] = .{.{ .id = id, .name = "shell", .arguments_json = "{\"command\":\"printf evidence\"}" }}; results[index] = .{.{ .tool_call_id = @constCast(id), .tool_name = @constCast("shell"), .status = .success, .output = @constCast("exact observed bytes\n"), .output_handle = @constCast(id), .preview = @constCast("exact observed bytes\n"), .output_bytes = 21, .stored_output_bytes = 21, }}; steps[index] = .{ .tool_calls = &calls[index], .tool_results = &results[index] }; } const user = types.UserTurn{ .text = @constCast("continue this work") }; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>Older evidence.</context_handoff>"), .removed_turn_count = 0, .compaction_count = 1, }, .{ .user = user, .assistant = @constCast(""), .execution = .{ .tool_steps = steps[0..2] } }, .{ .tool_steps = 1 }, 20); if (repeat) { _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>Updated older evidence.</context_handoff>"), .removed_turn_count = 0, .compaction_count = 2, }, .{ .user = user, .assistant = @constCast(""), .execution = .{ .tool_steps = steps[1..3] } }, .{ .tool_steps = 1 }, 30); } if (finish) { const first: usize = if (repeat) 2 else 1; _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = user, .assistant = @constCast("done"), .execution = .{ .tool_steps = steps[first .. first + 2] }, } }, } }, 40); } } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), resumed.state.history.len); try std.testing.expect(resumed.state.history[0] == .compacted_summary); const entry = resumed.state.history[1]; const execution = switch (entry) { .assistant => |turn| turn.execution, .interrupted => |turn| turn.execution, else => return error.TestUnexpectedResult, }; try std.testing.expectEqual(@as(usize, if (finish) 2 else 1), execution.tool_steps.len); try std.testing.expectEqualStrings(ids[if (repeat) @as(usize, 2) else 1], execution.tool_steps[0].tool_calls[0].id); const archive = try loadConversationHistoryRange(alloc, &resumed.log.dir, 0, 1); defer session.freeHistoryTurnSlice(alloc, archive); const archived = switch (archive[0]) { .assistant => |turn| turn.execution, .interrupted => |turn| turn.execution, else => return error.TestUnexpectedResult, }; const count: usize = 2 + @as(usize, @intFromBool(repeat)) + @as(usize, @intFromBool(finish)); try std.testing.expectEqual(count, archived.tool_steps.len); for (archived.tool_steps, 0..) |step, index| { try std.testing.expectEqualStrings(ids[index], step.tool_calls[0].id); try std.testing.expectEqualStrings("exact observed bytes\n", step.tool_results[0].preview.?); } } }}
test "mid-turn checkpoint retains a bound recovery suffix across writable resume" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-checkpoint-recovery", 10); defer initial.deinit(alloc); const user = types.UserTurn{ .text = @constCast("finish the request"), .work_id = @constCast("exact-open-work"), }; var recovery_before: ?[]u8 = null; defer if (recovery_before) |bytes| alloc.free(bytes); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.commitContextCompaction(alloc, .{ .summary = @constCast("<context_handoff>prior work completed</context_handoff>"), .removed_turn_count = 0, .compaction_count = 1, }, .{ .user = user, .assistant = @constCast("") }, null, 20); _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = .{ .turn_id = 7, .user = .{ .text = user.text }, .assistant_source = @constCast("remaining draft"), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, } } }, 30); recovery_before = try readManagedFileAlloc(alloc, &loaded.log.dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes); } { var readonly = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer readonly.deinit(alloc); try std.testing.expectEqualStrings("exact-open-work", readonly.recovery_checkpoint.?.user.work_id orelse return error.MissingRecoveryWorkId); } { var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expect(resumed.conversation_writer.turn_open); try std.testing.expectEqual(@as(usize, 1), resumed.state.history.len); try std.testing.expectEqualStrings("remaining draft", resumed.state.recovery_checkpoint.?.assistant_source); try std.testing.expectEqualStrings("exact-open-work", resumed.state.recovery_checkpoint.?.user.work_id orelse return error.MissingRecoveryWorkId); const recovery_after = try readManagedFileAlloc(alloc, &resumed.log.dir, recovery_checkpoint_file, session_codec.max_recovery_checkpoint_bytes); defer alloc.free(recovery_after); try std.testing.expectEqualStrings(recovery_before.?, recovery_after); _ = try resumed.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = user, .assistant = @constCast("remaining answer") } }, } }, 40); } var completed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer completed.deinit(alloc); try std.testing.expect(!completed.conversation_writer.turn_open); try std.testing.expectEqual(@as(usize, 2), completed.state.history.len); try std.testing.expect(completed.state.recovery_checkpoint == null); try std.testing.expectEqualStrings("remaining answer", completed.state.history[1].assistant.assistant);}
test "recovery cleanup failure does not reject a durable conversation append" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-recovery-cleanup", 10); defer initial.deinit(alloc); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); const checkpoint = session_codec.RecoveryCheckpoint{ .turn_id = 7, .user = .{ .text = @constCast("continue the request") }, .assistant_source = @constCast("partial"), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 3, .consumed_provider_attempts = 1, }; _ = try loaded.appendEvent(alloc, .{ .recovery_checkpoint_set = .{ .checkpoint = checkpoint }, }, 20); try loaded.log.dir.dir.deleteFile(std.testing.io, recovery_checkpoint_file); try loaded.log.dir.dir.createDir(std.testing.io, recovery_checkpoint_file, private_dir_permissions); const position = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = initial.conversation_language, .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = checkpoint.user, .assistant = @constCast("finished answer"), } }, } }, 30); try std.testing.expectEqual(@as(u64, 3), position.through_seq); try std.testing.expectEqual(@as(?session_codec.RecoveryCheckpoint, null), loaded.state.recovery_checkpoint);}
test "cache-free permission state resumes from its domain file" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-permission", 10); defer initial.deinit(alloc); const key = try session_permission_state.RuleKey.init( .command, "command\x00/workspace\x00zig build", ); var applied = try session_permission_state.apply(alloc, .{}, .{ .set = .{ .key = key, .display_identity = "zig build", .decision = .deny, .expected_generation = null, } }); var permissions = applied.takeApplied() orelse return error.TestUnexpectedResult; defer permissions.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); try loaded.replacePermissionState( alloc, permissions, 20, ); }
var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual( session_permission_state.StateDecision.deny, session_permission_state.decide(resumed.permission_state, key), );}
test "review feedback survives conversation resume range and archive replay" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |review_feedback| { var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "review-feedback-resume", 10); defer initial.deinit(alloc); var calls = [_]types.ToolCall{.{ .id = "call-review", .name = "shell", .arguments_json = "{}", }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("call-review"), .tool_name = @constCast("shell"), .status = .failure, .output = @constCast("Security review held this action."), .output_handle = @constCast("review-feedback.txt"), .output_bytes = "Security review held this action.".len, .stored_output_bytes = "Security review held this action.".len, .review_feedback = review_feedback, }}; var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); const session_path = try io_mod.dirRealpathAlloc(alloc, loaded.log.dir.dir, "."); defer alloc.free(session_path); var capability = try session_child_store.SessionChildCapability.init(alloc, loaded.log.dir.dir, session_path, .writable); defer capability.deinit(); var artifact = try capability.atomicReplace(alloc, .tool_results, "review-feedback.txt", "Security review held this action."); defer artifact.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("Check the result.") }, .assistant = @constCast("I will inspect another path."), .execution = .{ .tool_steps = &steps }, } }, } }, 20); } var resumed = try temp.root.resumeForWrite(alloc, initial.id, .{}); defer resumed.deinit(alloc); const range = try loadConversationHistoryRange(alloc, &resumed.log.dir, 0, 1); defer session.freeHistoryTurnSlice(alloc, range); const archive = try loadConversationArchive(alloc, &resumed.log.dir); defer session.freeHistoryTurnSlice(alloc, archive); var readonly = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer readonly.deinit(alloc); for ([_][]const types.HistoryTurn{ resumed.state.history, range, archive, readonly.history }) |history| { try std.testing.expectEqual(@as(usize, 1), history.len); const turn = history[0].assistant; try std.testing.expectEqualStrings("Check the result.", turn.user.text); try std.testing.expectEqualStrings("I will inspect another path.", turn.assistant); const result = turn.execution.tool_steps[0].tool_results[0]; try std.testing.expectEqual(review_feedback, result.review_feedback); try std.testing.expectEqualStrings("Security review held this action.", result.output); try std.testing.expectEqual(types.PersistedToolStatus.failure, result.status); } }}
test "cache-free resume rebuilds tool calls and external result references" { const alloc = std.testing.allocator; const provider_state = types.ProviderReplay{ .source = .{ .provider = .gateway, .model = "test" }, .parts_json = "[{\"type\":\"reasoning\",\"text\":\"kept\"}]" }; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-tool-resume", 10); defer initial.deinit(alloc); var calls = [_]types.ToolCall{.{ .id = "call-shell", .name = "shell", .arguments_json = "{\"command\":\"printf done\"}", .provider_result = "{\"native\":true}", .provenance = .provider_executed, }}; var feedback = [_][]u8{@constCast("inspect the result")}; const presentation_lines = [_]types.CommittedFilePresentationLine{.{ .kind = .addition, .new_line = 1, .text = "done", }}; var results = [_]types.PersistedToolResult{.{ .tool_call_id = @constCast("call-shell"), .tool_name = @constCast("shell"), .status = .success, .output = @constCast("done"), .output_handle = @constCast("result-shell.txt"), .output_bytes = 9, .stored_output_bytes = 4, .truncated = true, .provider_native = true, .created_at_ms = 19, .permission_feedback = &feedback, .committed_file_presentation = .{ .path = "done.txt", .kind = .added, .lines = &presentation_lines, .additions = 1, .deletions = 0, .truncated = false, .after_content = "done\n", .lifecycle_id = .{ .turn_id = 7, .call_id = "call-shell" }, }, .command_output_replay = .{ .available = .{ .handle = @constCast("fx-command-replay-test.bin"), .framed_bytes = 42, } }, .command_process_presentation = .{ .exit_code = 7 }, .terminal_action_presentation = .{ .returned = .{ .exited = 7 } }, }}; var steps = [_]types.ToolExecutionStep{.{ .assistant = @constCast("Running it."), .provider_replay = provider_state, .tool_calls = &calls, .tool_results = &results, }}; var files = [_]types.FileEvidence{.{ .path = @constCast("done.txt"), .tool_call_id = @constCast("call-shell"), .tool_name = @constCast("shell"), .action = .write, .model_view_covers_full_file = true, }}; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("Run it.") }, .provider_replay = provider_state, .assistant = @constCast("Done."), .execution = .{ .tool_steps = &steps, .files = &files, .turn_summary = .{ .started_at_ms = 10, .completed_at_ms = 20, .turn_duration_ms = 10, }, }, } }, } }, 20); }
var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), resumed.history.len); const execution = resumed.history[0].assistant.execution; try std.testing.expectEqualStrings(provider_state.parts_json, resumed.history[0].assistant.provider_replay.?.parts_json); try std.testing.expectEqualStrings(provider_state.parts_json, execution.tool_steps[0].provider_replay.?.parts_json); try std.testing.expectEqual(@as(usize, 1), execution.tool_steps.len); try std.testing.expectEqualStrings("call-shell", execution.tool_steps[0].tool_calls[0].id); try std.testing.expectEqual( types.ToolExecutionProvenance.provider_executed, execution.tool_steps[0].tool_calls[0].provenance, ); try std.testing.expectEqualStrings( "{\"native\":true}", execution.tool_steps[0].tool_calls[0].provider_result.?, ); try std.testing.expectEqualStrings( "result-shell.txt", execution.tool_steps[0].tool_results[0].output_handle.?, ); try std.testing.expectEqualStrings( "inspect the result", execution.tool_steps[0].tool_results[0].permission_feedback[0], ); const result = execution.tool_steps[0].tool_results[0]; try std.testing.expectEqual(@as(usize, 9), result.output_bytes); try std.testing.expectEqual(@as(usize, 4), result.stored_output_bytes); try std.testing.expect(result.truncated); try std.testing.expect(result.provider_native); try std.testing.expectEqual(@as(i64, 19), result.created_at_ms); try std.testing.expectEqual( types.CommandProcessPresentation{ .exit_code = 7 }, result.command_process_presentation.?, ); try std.testing.expectEqual( types.TerminalActionPresentation{ .returned = .{ .exited = 7 } }, result.terminal_action_presentation.?, ); const presentation = result.committed_file_presentation.?; try std.testing.expectEqualStrings("done.txt", presentation.path); try std.testing.expectEqualStrings("done\n", presentation.after_content.?); try std.testing.expectEqual(@as(u64, 7), presentation.lifecycle_id.?.turn_id); try std.testing.expectEqual(@as(usize, 1), execution.files.len); try std.testing.expectEqualStrings("done.txt", execution.files[0].path); try std.testing.expect(execution.files[0].model_view_covers_full_file); try std.testing.expectEqual(@as(u64, 10), execution.turn_summary.?.turn_duration_ms); const replay = execution.tool_steps[0].tool_results[0].command_output_replay orelse return error.TestExpectedCommandReplay; switch (replay) { .available => |descriptor| { try std.testing.expectEqualStrings( "fx-command-replay-test.bin", descriptor.handle, ); try std.testing.expectEqual(@as(usize, 42), descriptor.framed_bytes); }, .unavailable => return error.TestExpectedCommandReplay, }}
fn checkCompleteSkillReplayArtifact(artifact: enum { intact, missing, mismatched }) !void { const alloc = std.testing.allocator; const runtime_memory = @import("../agent/runtime/execution_memory.zig"); const execution_memory = @import("../agent/execution_memory.zig"); const full_output = "<skill_content name=\"workflow\" location=\"/skills/workflow\" resource=\"SKILL.md\" complete=\"true\">\n" ++ ("required instruction\n" ** 1200) ++ "COMPLETE_SKILL_TAIL\n</skill_content>"; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "complete-skill-replay", 10); defer initial.deinit(alloc); var arena_state = std.heap.ArenaAllocator.init(alloc); defer arena_state.deinit(); const arena = arena_state.allocator(); var cancel_flag = std.atomic.Value(bool).init(false); var calls = [_]types.ToolCall{.{ .id = "load-skill", .name = "skill", .arguments_json = "{\"location\":\"/skills/workflow\"}" }}; { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); const session_path = try io_mod.dirRealpathAlloc(alloc, loaded.log.dir.dir, "."); defer alloc.free(session_path); var capability = try session_child_store.SessionChildCapability.init(alloc, loaded.log.dir.dir, session_path, .writable); defer capability.deinit(); const prepared = try runtime_memory.prepareToolExecutionOutput(arena, .{ .system_prompt = "", .gateway_retry_count = 0, .gateway_chat_url = "", .agent_step_limit = 1, .cancel_flag = &cancel_flag, .session_child_capability = &capability, }, calls[0], .{ .status = .success, .model_content_kind = .complete_skill, .model_output = full_output }, null); try std.testing.expect(full_output.len > result_store.large_result_threshold_bytes); try std.testing.expect(!prepared.memory.truncated); try std.testing.expectEqualStrings(full_output, prepared.model_output); var results = [_]types.PersistedToolResult{try execution_memory.makePersistedToolResult( alloc, calls[0].id, calls[0].name, .success, prepared.model_output, prepared.memory, )}; defer freeConversationToolResult(alloc, results[0]); var steps = [_]types.ToolExecutionStep{.{ .tool_calls = &calls, .tool_results = &results }}; _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("Use this skill.") }, .assistant = @constCast("Loaded the instructions."), .execution = .{ .tool_steps = &steps }, } }, } }, 20); const handle = prepared.memory.output_handle.?; switch (artifact) { .intact => {}, .missing => try capability.delete(.tool_results, handle), .mismatched => { var replacement = try capability.atomicReplace(alloc, .tool_results, handle, "different-sized content"); replacement.deinit(alloc); }, } } var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); const execution = resumed.history[0].assistant.execution; const replayed = execution.tool_steps[0].tool_results[0]; try std.testing.expectEqual(types.PersistedToolStatus.success, replayed.status); try std.testing.expectEqual(full_output.len, replayed.stored_output_bytes); var messages: std.ArrayList(types.ChatMessage) = .empty; try session.appendExecutionMemoryChatMessages(arena, &messages, execution); const projected = messages.items[1].content.?; if (artifact == .intact) { try std.testing.expect(!replayed.truncated); try std.testing.expectEqualStrings(full_output, projected); } else { try std.testing.expect(replayed.truncated); try std.testing.expect(std.mem.find(u8, projected, "unavailable") != null); try std.testing.expect(std.mem.find(u8, projected, "complete=\"true\"") == null); try std.testing.expect(std.mem.find(u8, projected, "required instruction") == null); }}
test "complete skill replay retains intact full artifacts" { try checkCompleteSkillReplayArtifact(.intact);}
test "complete skill replay marks missing artifacts unavailable" { try checkCompleteSkillReplayArtifact(.missing);}
test "complete skill replay marks mismatched artifacts unavailable" { try checkCompleteSkillReplayArtifact(.mismatched);}
test "cache-free resume loads only the latest checkpoint and suffix" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "conversation-checkpoint", 10); defer initial.deinit(alloc); { var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 1, .total_output_tokens = 1, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("old question") }, .assistant = @constCast("old answer"), } }, } }, 20); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 2, .total_output_tokens = 2, .turn = .{ .compacted_summary = .{ .summary = @constCast("<context_handoff>summary</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, } }, } }, 30); _ = try loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .conversation_language = .literal("en"), .total_input_tokens = 3, .total_output_tokens = 3, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("new question") }, .assistant = @constCast("new answer"), } }, } }, 40); }
var resumed = try temp.root.loadReadOnly(alloc, initial.id, .{}); defer resumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), resumed.history.len); try std.testing.expect(resumed.history[0] == .compacted_summary); try std.testing.expectEqualStrings("new question", resumed.history[1].assistant.user.text);
var session_dir = try temp.root.sessions.?.dir.openDir( std.testing.io, initial.id, .{}, ); defer session_dir.close(std.testing.io); var event_file = try session_dir.openFile(std.testing.io, "events.jsonl", .{}); defer event_file.close(std.testing.io); const bytes = try io_mod.readFileToEnd(alloc, &event_file, 1024 * 1024); defer alloc.free(bytes); try std.testing.expect(std.mem.find(u8, bytes, "old question") != null);}
test "language metadata failure retains ownership of the committed work identity" { const alloc = std.testing.allocator; var temp = try TempRoot.init(alloc); defer temp.deinit(alloc); var initial = try testState(alloc, "language-work-failure", 10); defer initial.deinit(alloc); var loaded = try temp.root.startConversationSession(alloc, initial, .{}); defer loaded.deinit(alloc); loaded.freshly_started = false; try loaded.log.dir.dir.deleteFile(std.testing.io, "session.json"); try loaded.log.dir.dir.createDir(std.testing.io, "session.json", private_dir_permissions); try std.testing.expectError(error.SessionPersistenceUncertain, loaded.appendEvent(alloc, .{ .history_turn_committed = .{ .work_id = @constCast("committed-work"), .conversation_language = .literal("fr"), .total_input_tokens = 0, .total_output_tokens = 0, .turn = .{ .assistant = .{ .user = .{ .text = @constCast("Bonjour") }, .assistant = @constCast("Salut") } }, } }, 20)); try std.testing.expectEqualStrings("committed-work", loaded.state.last_subagent_work_id.?); try std.testing.expectError(error.SessionPersistenceUncertain, loaded.requireWritable());}