Something went wrong. Try again.
My version of https://github.com/vercel-labs/fx
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613461446154616461746184619462046214622462346244625462646274628462946304631463246334634463546364637463846394640464146424643464446454646464746484649465046514652465346544655465646574658465946604661466246634664466546664667466846694670467146724673467446754676467746784679468046814682468346844685468646874688468946904691469246934694469546964697469846994700470147024703470447054706470747084709471047114712471347144715471647174718471947204721472247234724472547264727472847294730473147324733473447354736473747384739474047414742474347444745474647474748474947504751475247534754475547564757475847594760476147624763476447654766476747684769477047714772477347744775477647774778477947804781478247834784478547864787478847894790479147924793479447954796479747984799480048014802480348044805480648074808480948104811481248134814481548164817481848194820482148224823482448254826482748284829483048314832483348344835483648374838483948404841484248434844484548464847484848494850485148524853485448554856485748584859486048614862486348644865486648674868486948704871487248734874487548764877487848794880488148824883488448854886488748884889489048914892489348944895489648974898489949004901490249034904490549064907490849094910491149124913491449154916491749184919492049214922492349244925492649274928492949304931493249334934493549364937493849394940494149424943494449454946494749484949495049514952495349544955495649574958495949604961496249634964496549664967496849694970497149724973497449754976497749784979498049814982498349844985498649874988498949904991499249934994499549964997499849995000500150025003500450055006500750085009501050115012501350145015501650175018501950205021502250235024502550265027502850295030503150325033503450355036503750385039504050415042504350445045504650475048504950505051505250535054505550565057505850595060506150625063506450655066506750685069507050715072507350745075507650775078507950805081508250835084508550865087508850895090509150925093509450955096509750985099510051015102510351045105510651075108510951105111511251135114511551165117511851195120512151225123512451255126512751285129513051315132513351345135513651375138513951405141514251435144514551465147514851495150515151525153515451555156515751585159516051615162516351645165516651675168516951705171517251735174517551765177517851795180518151825183518451855186518751885189519051915192519351945195519651975198519952005201520252035204520552065207520852095210521152125213521452155216521752185219522052215222522352245225522652275228522952305231523252335234523552365237523852395240524152425243524452455246524752485249525052515252525352545255525652575258525952605261526252635264526552665267526852695270527152725273527452755276527752785279528052815282528352845285528652875288528952905291529252935294529552965297529852995300530153025303530453055306530753085309531053115312531353145315531653175318531953205321532253235324532553265327532853295330533153325333533453355336533753385339534053415342534353445345534653475348534953505351535253535354535553565357535853595360536153625363536453655366536753685369537053715372537353745375537653775378537953805381538253835384538553865387538853895390539153925393539453955396539753985399540054015402540354045405540654075408540954105411541254135414541554165417541854195420542154225423542454255426542754285429543054315432543354345435543654375438543954405441544254435444544554465447544854495450545154525453545454555456545754585459546054615462546354645465546654675468546954705471547254735474547554765477547854795480548154825483548454855486548754885489549054915492549354945495549654975498549955005501550255035504550555065507550855095510551155125513551455155516551755185519552055215522552355245525552655275528552955305531553255335534553555365537553855395540554155425543554455455546554755485549555055515552555355545555555655575558555955605561556255635564556555665567556855695570557155725573557455755576557755785579558055815582558355845585558655875588558955905591559255935594559555965597559855995600560156025603560456055606560756085609561056115612561356145615561656175618561956205621562256235624562556265627562856295630563156325633563456355636563756385639564056415642564356445645564656475648564956505651565256535654565556565657565856595660566156625663566456655666566756685669567056715672567356745675567656775678567956805681568256835684568556865687568856895690569156925693569456955696569756985699570057015702570357045705570657075708570957105711571257135714571557165717571857195720572157225723572457255726572757285729573057315732573357345735573657375738573957405741574257435744574557465747574857495750575157525753575457555756575757585759576057615762576357645765576657675768576957705771577257735774577557765777577857795780578157825783578457855786578757885789579057915792579357945795579657975798579958005801580258035804580558065807580858095810581158125813581458155816581758185819582058215822582358245825582658275828582958305831583258335834583558365837583858395840584158425843584458455846584758485849585058515852585358545855585658575858585958605861586258635864586558665867586858695870587158725873587458755876587758785879588058815882588358845885588658875888588958905891589258935894589558965897589858995900590159025903590459055906590759085909591059115912591359145915591659175918591959205921592259235924592559265927592859295930593159325933593459355936593759385939594059415942594359445945594659475948594959505951595259535954595559565957595859595960596159625963596459655966596759685969597059715972597359745975597659775978597959805981598259835984598559865987598859895990599159925993599459955996599759985999600060016002600360046005600660076008600960106011601260136014601560166017601860196020602160226023602460256026602760286029603060316032603360346035603660376038603960406041604260436044604560466047604860496050605160526053605460556056605760586059606060616062606360646065606660676068606960706071607260736074607560766077607860796080608160826083608460856086608760886089609060916092609360946095609660976098609961006101610261036104610561066107610861096110611161126113611461156116611761186119612061216122612361246125612661276128612961306131613261336134613561366137613861396140614161426143614461456146614761486149615061516152615361546155615661576158615961606161616261636164616561666167616861696170617161726173617461756176617761786179618061816182618361846185618661876188618961906191619261936194619561966197619861996200620162026203620462056206620762086209621062116212621362146215621662176218621962206221622262236224622562266227622862296230623162326233623462356236623762386239624062416242624362446245624662476248624962506251625262536254625562566257625862596260626162626263626462656266626762686269627062716272627362746275627662776278627962806281628262836284628562866287628862896290629162926293629462956296629762986299630063016302630363046305630663076308630963106311631263136314631563166317631863196320632163226323632463256326632763286329633063316332633363346335633663376338633963406341634263436344634563466347634863496350635163526353635463556356635763586359636063616362636363646365636663676368636963706371637263736374637563766377637863796380638163826383638463856386638763886389639063916392639363946395639663976398639964006401640264036404640564066407640864096410641164126413641464156416641764186419642064216422642364246425642664276428642964306431643264336434643564366437643864396440644164426443644464456446644764486449645064516452645364546455645664576458645964606461646264636464646564666467646864696470647164726473647464756476647764786479648064816482648364846485648664876488648964906491649264936494649564966497649864996500650165026503650465056506650765086509651065116512651365146515651665176518651965206521652265236524652565266527652865296530653165326533653465356536653765386539654065416542654365446545654665476548654965506551655265536554655565566557655865596560656165626563656465656566656765686569657065716572657365746575657665776578657965806581658265836584658565866587658865896590659165926593659465956596659765986599660066016602660366046605660666076608660966106611661266136614661566166617661866196620662166226623662466256626662766286629663066316632663366346635663666376638663966406641664266436644664566466647664866496650665166526653665466556656665766586659666066616662666366646665666666676668666966706671667266736674667566766677667866796680668166826683668466856686668766886689669066916692669366946695669666976698669967006701670267036704670567066707670867096710671167126713671467156716671767186719672067216722672367246725672667276728672967306731673267336734673567366737673867396740674167426743674467456746674767486749675067516752675367546755675667576758675967606761676267636764676567666767676867696770677167726773677467756776677767786779678067816782678367846785678667876788678967906791679267936794679567966797679867996800680168026803680468056806680768086809681068116812681368146815681668176818681968206821682268236824682568266827682868296830683168326833683468356836683768386839684068416842684368446845684668476848684968506851685268536854685568566857685868596860686168626863686468656866686768686869687068716872687368746875687668776878687968806881688268836884688568866887688868896890689168926893689468956896689768986899690069016902690369046905690669076908690969106911691269136914691569166917691869196920692169226923692469256926692769286929693069316932693369346935693669376938693969406941694269436944694569466947694869496950695169526953695469556956695769586959696069616962696369646965696669676968696969706971697269736974697569766977697869796980698169826983698469856986698769886989699069916992699369946995699669976998699970007001700270037004700570067007700870097010701170127013701470157016701770187019702070217022702370247025702670277028702970307031703270337034703570367037703870397040704170427043704470457046704770487049705070517052705370547055705670577058705970607061706270637064706570667067706870697070707170727073707470757076707770787079708070817082708370847085708670877088708970907091709270937094709570967097709870997100710171027103710471057106710771087109711071117112711371147115711671177118711971207121712271237124712571267127712871297130713171327133713471357136713771387139714071417142714371447145714671477148714971507151715271537154715571567157715871597160716171627163716471657166716771687169717071717172717371747175717671777178717971807181718271837184718571867187718871897190719171927193719471957196719771987199720072017202720372047205720672077208720972107211721272137214721572167217721872197220722172227223722472257226722772287229723072317232723372347235723672377238723972407241724272437244724572467247724872497250725172527253725472557256725772587259726072617262726372647265726672677268726972707271727272737274727572767277727872797280728172827283728472857286728772887289729072917292const std = @import("std");const credentials = @import("../auth/credentials.zig");const secret = @import("../auth/secret.zig");const io_mod = @import("../shared/io.zig");const debug_trace = @import("../shared/debug_trace.zig");const diagnostics = @import("../workspace/diagnostics.zig");const diff_mod = @import("../output/diff.zig");const file_mutation_contract = @import("../tooling/file_mutation_contract.zig");const image_attachments = @import("../images/image_attachments.zig");const context_contract = @import("../workspace/context_contract.zig");const auto_classifier_context = @import("../permissions/auto_classifier_context.zig");const permission_request = @import("../permissions/permission_request.zig");const notification_contract = @import("../notifications/notification_contract.zig");const session_runtime = @import("../session/session.zig");const session_codec = @import("../session/session_codec.zig");const skill_contract = @import("../skills/skill_contract.zig");const tool_result_limits = @import("../tooling/tool_result_limits.zig");const command_output_content = @import("../tooling/command_output_content.zig");const types = @import("../shared/types.zig");const model_provider = @import("../config/model_provider.zig");const assistant_presentation = @import("assistant_presentation.zig");const compaction_activity = @import("../output/compaction_activity.zig");
pub const AgentTurnSettings = struct { max_tool_result_bytes: usize = tool_result_limits.default_max_tool_result_bytes, first_call_tool_choice: types.ToolChoice = .auto, fast_mode: bool = false, effort: types.ReasoningEffort = .auto,};
pub const SkillBinding = struct { name: []u8, path: []u8,};
pub const SkillDisplaySpan = struct { raw_start: usize, raw_end: usize, name: []u8, path: []u8, display_source: ?skill_contract.SkillSource = null, owns_trailing_separator: bool = false,};
pub const BeginPromptWithSkillBindings = struct { prompt: types.UserTurn, skill_bindings: []SkillBinding = &.{}, skill_display_spans: []SkillDisplaySpan = &.{},};
pub const PromptDelivery = union(enum) { ordinary, active_turn: u64, continuation,
pub fn activeTurnId(self: PromptDelivery) ?u64 { return switch (self) { .active_turn => |turn_id| turn_id, .ordinary, .continuation => null, }; }
pub fn isSteering(self: PromptDelivery) bool { return switch (self) { .ordinary => false, .active_turn, .continuation => true, }; }
pub fn isContinuation(self: PromptDelivery) bool { return self == .continuation; }};
pub const SteeringBoundaryKind = enum { model, cancelled, finalizing,};
const SteeringDelivery = types.SteeringDelivery;
/// Borrowed by the queue. Its owner outlives worker teardown.pub const SteeringReceipt = struct { operation_id: []const u8 = "", state: std.atomic.Value(SteeringDelivery) = .init(.queued),};
pub const SteeringBoundaryResult = union(enum) { none, continue_turn: [][]u8, handoff, interrupt,};
pub const QueuedPrompt = struct { turn_id: u64 = 0, steering_receipt: ?*SteeringReceipt = null, /// Runtime-derived delivery state. The worker retains ownership in every state. delivery: PromptDelivery = .ordinary, prompt: []u8, images: []types.ImageAttachment, authorized_image_catalog: []types.ImageAttachment = &.{}, model: []u8, provider: model_provider.ProviderId = .gateway, api_key: []u8, gateway_team: ?[]u8 = null, credential_source: ?types.CredentialSource = null, account_id: ?[]u8 = null, permission_mode: types.PermissionMode, history: []types.HistoryTurn, /// Absolute end of the leading canonical-history range loaded without /// corrected result provenance. The conservative default protects callers /// that do not own a SessionRuntime snapshot. unversioned_history_count: usize = std.math.maxInt(usize), root_user_intent_context: []u8 = &.{}, grants: []types.PermissionGrant, skill_bindings: []SkillBinding = &.{}, skill_display_spans: []SkillDisplaySpan = &.{}, context_snapshot: context_contract.GatheredContextSnapshot = .{}, agent_settings: AgentTurnSettings = .{}, snapshot_file_ownerships: []types.SnapshotFileOwnership = &.{}, /// Present only when an explicit host action resumes a durable paused /// model turn. The same ownership rules as the other queued fields apply. recovery_checkpoint: ?session_codec.RecoveryCheckpoint = null, /// True when this output surface already contains the checkpoint's /// assistant source. Recovery still restores the source for overlap and /// persistence, but publishes only novel continuation text. recovery_source_already_presented: bool = false, /// The interactive transcript already contains this exact user turn. /// Worker begin publishes only its identity so the UI can consume the /// pending owner without painting a duplicate card. user_prompt_already_presented: bool = false,};
pub const ContextCompactionTask = struct { turn_id: u64 = 0, operation_id: ?compaction_activity.OperationId = null, model: []u8, provider: model_provider.ProviderId = .gateway, api_key: []u8, gateway_team: ?[]u8 = null, credential_source: ?types.CredentialSource = null, account_id: ?[]u8 = null, history: []types.HistoryTurn, unversioned_history_count: usize = std.math.maxInt(usize),};
pub const ContextCompactionStatus = enum { idle, queued, running,};
pub const WorkItem = union(enum) { prompt: QueuedPrompt, compact_context: ContextCompactionTask,};
pub const ActivePromptSnapshotOwnership = struct { /// Owns snapshot deletion until the files are transferred to accepted /// history or handed to a reference-counted finished-turn owner. images: []const types.ImageAttachment, state: enum { active, handed_off, preserved, discarded, } = .active, shared_ownership: ?types.SnapshotFileOwnership = null,
pub fn init(images: []const types.ImageAttachment) ActivePromptSnapshotOwnership { return .{ .images = images }; }
fn ensureSharedOwnership( self: *ActivePromptSnapshotOwnership, alloc: std.mem.Allocator, ) !?types.SnapshotFileOwnership { if (self.images.len == 0) return null; if (self.state == .preserved or self.state == .discarded) return null; if (self.shared_ownership) |ownership| return ownership; const ownership = (try SnapshotFileOwnershipState.create(alloc, self.images)).handle(); self.shared_ownership = ownership; return ownership; }
fn handoff( self: *ActivePromptSnapshotOwnership, alloc: std.mem.Allocator, ) !?types.SnapshotFileOwnership { if (self.state != .active or self.images.len == 0) return null; const ownership = (try self.ensureSharedOwnership(alloc)).?; self.state = .handed_off; return ownership; }
fn preserve(self: *ActivePromptSnapshotOwnership) bool { if (self.images.len == 0) return false; switch (self.state) { .active => { if (self.shared_ownership) |ownership| { ownership.transfer(); ownership.release(); self.shared_ownership = null; } self.state = .preserved; return true; }, .handed_off => { if (self.shared_ownership) |ownership| ownership.transfer(); return true; }, .preserved => return true, .discarded => return false, } }
fn deinit(self: *ActivePromptSnapshotOwnership) void { if (self.state == .active) { if (self.shared_ownership) |ownership| { ownership.release(); } else { image_attachments.deleteUnreferencedImageSnapshots(self.images, &.{}); } self.state = .discarded; } }};
const SnapshotFileOwnershipState = struct { alloc: std.mem.Allocator, images: []types.ImageAttachment, references: std.atomic.Value(usize) = std.atomic.Value(usize).init(1), transferred: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
fn create( alloc: std.mem.Allocator, images: []const types.ImageAttachment, ) !*SnapshotFileOwnershipState { const owned_images = try types.dupeImageAttachmentSlice(alloc, images); errdefer types.freeImageAttachmentSlice(alloc, owned_images); const state = try alloc.create(SnapshotFileOwnershipState); state.* = .{ .alloc = alloc, .images = owned_images, }; return state; }
fn handle(self: *SnapshotFileOwnershipState) types.SnapshotFileOwnership { return .{ .ctx = self, .retain_fn = retain, .release_fn = release, .transfer_fn = transfer, }; }
fn retain(raw: *anyopaque) void { const self: *SnapshotFileOwnershipState = @ptrCast(@alignCast(raw)); _ = self.references.fetchAdd(1, .seq_cst); }
fn release(raw: *anyopaque) void { const self: *SnapshotFileOwnershipState = @ptrCast(@alignCast(raw)); if (self.references.fetchSub(1, .seq_cst) != 1) return; if (!self.transferred.load(.seq_cst)) { image_attachments.deleteUnreferencedImageSnapshots(self.images, &.{}); } const alloc = self.alloc; types.freeImageAttachmentSlice(alloc, self.images); alloc.destroy(self); }
fn transfer(raw: *anyopaque) void { const self: *SnapshotFileOwnershipState = @ptrCast(@alignCast(raw)); self.transferred.store(true, .seq_cst); }};
fn historyTurnImages(turn: types.HistoryTurn) []const types.ImageAttachment { return switch (turn) { .assistant => |value| value.user.images, .interrupted => |value| value.user.images, .compacted_summary => &.{}, };}
fn sameImageSnapshots( left: []const types.ImageAttachment, right: []const types.ImageAttachment,) bool { if (left.len == 0 or left.len != right.len) return false; for (left) |image| { const other = image_attachments.findById(right, image.id) orelse return false; if (!optionalBytesEqual(image.snapshot_path, other.snapshot_path) or !optionalBytesEqual(image.snapshot_sha256, other.snapshot_sha256)) { return false; } } return true;}
fn optionalBytesEqual(left: ?[]const u8, right: ?[]const u8) bool { if (left == null or right == null) return left == null and right == null; return std.mem.eql(u8, left.?, right.?);}
pub const PermissionSnapshot = struct { mode: types.PermissionMode,};
pub const CommandOutputChunk = struct { lifecycle_id: ?types.ToolLifecycleId = null, stream: command_output_content.Stream, text: []u8,};
pub const SteeringPresentationSnapshot = struct { messages: [][]u8 = &.{}, pending_feedback: [][]u8 = &.{}, waits_for_boundary: bool = false,
pub fn deinit(self: *SteeringPresentationSnapshot, alloc: std.mem.Allocator) void { for (self.messages) |message| alloc.free(message); if (self.messages.len > 0) alloc.free(self.messages); for (self.pending_feedback) |message| alloc.free(message); if (self.pending_feedback.len > 0) alloc.free(self.pending_feedback); self.* = .{}; }};
const PreparedHistoryPropagation = struct { root_user_intent_context: []u8, authorized_image_catalog: []types.ImageAttachment, snapshot_file_ownerships: ?[]types.SnapshotFileOwnership, added_snapshot_ownership: ?types.SnapshotFileOwnership,
fn deinit(self: PreparedHistoryPropagation, alloc: std.mem.Allocator) void { alloc.free(self.root_user_intent_context); types.freeImageAttachmentSlice(alloc, self.authorized_image_catalog); if (self.added_snapshot_ownership) |ownership| ownership.release(); if (self.snapshot_file_ownerships) |ownerships| alloc.free(ownerships); }};
pub const StateSnapshot = struct { processing: bool = false, active_turn_id: u64 = 0, queued_count: usize = 0, pending_event_count: usize = 0, cancel_requested: bool = false, cancel_continues_turn: bool = false, pending_permission_request: ?permission_request.OwnedPermissionRequest = null, pending_permission_review: ?PendingPermissionReview = null,
pub fn deinit(self: StateSnapshot, alloc: std.mem.Allocator) void { if (self.pending_permission_request) |request| { var owned = request; owned.deinit(alloc); } }};
pub const PendingPermissionReview = struct { request_id: u64, review: *const diff_mod.FileReview,};
pub const FinalizationFailure = struct { turn_id: u64, outcome: types.TurnPresentationOutcome,};
pub const InteractiveAdmissionSnapshot = union(enum) { open, stopped, finalization_failed: FinalizationFailure,};
pub const PermissionSubmissionResult = enum { accepted, stale, no_pending,};
const OwnedQuestionOption = struct { label: []u8, description: ?[]u8 = null,};
const OwnedQuestionEntry = struct { question: []u8, options: []OwnedQuestionOption, submission: @FieldType(types.QuestionBatchEntry, "submission") = .none,};
pub const QuestionPromptSource = enum { agent_question, route_recovery, mcp_elicitation,};
const OwnedQuestionBatch = struct { entries: []OwnedQuestionEntry, source: QuestionPromptSource = .agent_question,};
const QuestionResponse = union(enum) { pending, cancelled, answered: [][]u8,};
pub const PendingQuestionBatchSnapshot = struct { entries: []types.QuestionBatchEntry, source: QuestionPromptSource = .agent_question,
pub fn deinit(self: PendingQuestionBatchSnapshot, alloc: std.mem.Allocator) void { for (self.entries) |entry| { alloc.free(@constCast(entry.question)); for (entry.options) |opt| { alloc.free(@constCast(opt.label)); if (opt.description) |desc| alloc.free(@constCast(desc)); } alloc.free(@constCast(entry.options)); } alloc.free(self.entries); }};
pub const ContextCompaction = struct { turn_id: u64 = 0, summary: types.CompactedSummaryHistoryTurn, active_prefix: ?types.AssistantHistoryTurn = null, retained_from: ?types.ContextHistoryCut = null, max_history_turns: usize = 0,};
pub const ContextCompactionHandler = *const fn (*anyopaque, ContextCompaction) anyerror!void;
pub const FreshPromptPreparation = struct { turn_id: u64 = 0, user: types.UserTurn, prior_turn: ?types.HistoryTurn = null,};
pub const FreshPromptHistory = struct { history: []types.HistoryTurn, authorized_image_catalog: []types.ImageAttachment, root_user_intent_context: []u8, unversioned_history_count: usize,
pub fn deinit(self: *FreshPromptHistory, alloc: std.mem.Allocator) void { types.freeHistoryTurnSlice(alloc, self.history); types.freeImageAttachmentSlice(alloc, self.authorized_image_catalog); alloc.free(self.root_user_intent_context); self.* = undefined; }};
pub const FreshPromptHandler = *const fn (*anyopaque, FreshPromptPreparation) anyerror!FreshPromptHistory;
const HistoryPublication = struct { ctx: *anyopaque, commit_fn: *const fn (*anyopaque) anyerror!void, retained_from: ?types.ContextHistoryCut = null,
fn commit(self: HistoryPublication) !void { try self.commit_fn(self.ctx); }};
const ContextCompactionPublication = struct { ctx: *anyopaque, handler: ContextCompactionHandler, event: ContextCompaction,
fn commit(raw: *anyopaque) !void { const self: *ContextCompactionPublication = @ptrCast(@alignCast(raw)); try self.handler(self.ctx, self.event); }};
const HistoryPublicationResponse = union(enum) { idle, waiting: u64, committed: ?FreshPromptHistory, failed: anyerror,};
pub const WorkerEvent = union(enum) { begin_prompt: types.UserTurn, begin_prompt_with_skill_bindings: BeginPromptWithSkillBindings, begin_presented_prompt: u64, append_user_feedback: []u8, assistant_presentation: assistant_presentation.Event, notification: notification_contract.Notification, question_requested, open_model_picker, semantic_notice: types.SemanticNotice, /// A full-only observability record (network call outcome) for the ctrl+o /// full transcript's detail section. Never enters the transcript store. full_detail_record: types.SemanticNotice, route_recovery_status: types.RouteRecoveryStatus, clear_route_recovery_status, api_status_text: []u8, credential_refreshed: credentials.Credential, command_output: CommandOutputChunk, command_output_complete: ?types.ToolLifecycleId, tool_lifecycle: types.ToolLifecycleEvent, turn_token_update: types.TurnTokenProgress, turn_phase_update: types.TurnPhaseUpdate, diff_block: diff_mod.DiffEntryPayload, context_compaction: ContextCompaction, prepare_fresh_prompt: FreshPromptPreparation, finish_prompt: types.FinishedPrompt, session_grant: types.PermissionGrant, error_text: types.SemanticNotice, /// A turn that can never recover hands the prompt back to the composer. /// Owned by the event; the consumer frees it. restore_failed_prompt: []u8,};
pub const WorkerEventBatch = struct { events: std.ArrayList(WorkerEvent), cancelled_turn_id: ?u64,};
pub const WorkerRuntime = struct { worker_mutex: std.Io.Mutex = .init, worker_cond: std.Io.Condition = .init, /// One admission-ordered queue for ordinary prompts and steering. Steering /// remains in place until its target turn consumes or demotes it. queued_prompts: std.ArrayList(QueuedPrompt) = .empty, /// One immutable conversation snapshot shared by every queued prompt. /// It is transferred or copied only when a prompt actually begins. queued_history: []types.HistoryTurn = &.{}, queued_context_compaction: ?ContextCompactionTask = null, active_context_compaction: bool = false, history_publication_response: HistoryPublicationResponse = .idle, worker_events: std.ArrayList(WorkerEvent) = .empty, worker_processing: bool = false, active_turn_id: u64 = 0, active_tool_calls: std.StringHashMapUnmanaged(void) = .empty, /// Allocator used for active call IDs. Allocator handles are capabilities, /// not comparable identities; callers guarantee compatible ownership. active_tool_allocator: ?std.mem.Allocator = null, /// Retains the tool-producing step so later text from that same step cannot /// reopen immediate steering before authoritative lifecycle IDs exist. tool_phase_step_id: ?u64 = null, worker_stop_requested: bool = false, finalization_failure: ?FinalizationFailure = null, worker_cancel_requested: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), /// The active turn whose text steering admission owns the cancel flag. /// Guarded by `worker_mutex`; explicit cancellation clears this ownership. steering_cancel_turn_id: ?u64 = null, direct_steering_closed: bool = false, worker_recovery_pause_requested: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), worker_connectivity_wait_active: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), /// Set under `worker_mutex` once the active turn publishes its terminal /// recovery pause. The turn may still be finalizing, but cannot perform /// more model work, so one continuation may queue before processing clears. recovery_continuation_ready: bool = false, /// When true, queued work will not start. turn_start_held: bool = false, next_permission_request_id: u64 = 1, pending_permission_response: ?permission_request.OwnedPermissionResponse = null, pending_permission_request_shared: ?permission_request.OwnedPermissionRequest = null, pending_permission_review: ?*const diff_mod.FileReview = null, pending_permission_waiting: bool = false, pending_question_shared: ?OwnedQuestionBatch = null, pending_question_response: QuestionResponse = .pending, agent_turn_settings: AgentTurnSettings = .{}, active_agent_turn_settings: ?AgentTurnSettings = null, active_context_snapshot: ?*const context_contract.GatheredContextSnapshot = null, active_prompt_is_root_authority: bool = false, active_prompt_snapshot_ownership: ?*ActivePromptSnapshotOwnership = null, preserve_prompt_snapshot_turn_id: ?u64 = null,
/// Presentation metadata only; all access is protected by worker_mutex. compaction_presentation: compaction_activity.State = .{},
pub fn beginCompactionActivity(self: *WorkerRuntime, origin: compaction_activity.Origin, turn_id: ?u64) compaction_activity.OperationId { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.compaction_presentation.begin(origin, turn_id, io_mod.milliTimestamp()); }
/// A rejected intent must not hide an already-running compaction. pub fn rejectCompactionActivity(self: *WorkerRuntime, outcome: compaction_activity.Outcome) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.compaction_presentation.snapshot.operation) |op| if (op.active()) return; const now = io_mod.milliTimestamp(); const id = self.compaction_presentation.begin(.manual, null, now); self.compaction_presentation.settle(id, .{ .outcome = outcome }, now); }
pub fn runCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, stage: compaction_activity.Stage) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.compaction_presentation.running(id, stage); }
pub fn settleCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, feedback: compaction_activity.Feedback) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.compaction_presentation.settle(id, feedback, io_mod.milliTimestamp()); }
/// Cheap by-value read; does not clone permissions or allocate. pub fn compactionActivitySnapshot(self: *WorkerRuntime) compaction_activity.Snapshot { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.compaction_presentation.snapshot; }
pub fn dismissCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, revision: u64) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.compaction_presentation.dismiss(id, revision); }
pub fn expireCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, revision: u64, now_ms: i64) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.compaction_presentation.expire(id, revision, now_ms); }
fn stopCompactionActivityLocked(self: *WorkerRuntime) void { const op = self.compaction_presentation.snapshot.operation orelse return; if (op.turn_id != self.active_turn_id or !self.worker_processing) return; self.compaction_presentation.stopping(op.id); }
fn finishCompactionActivityLocked(self: *WorkerRuntime, turn_id: u64) void { const op = self.compaction_presentation.snapshot.operation orelse return; if (op.turn_id != turn_id or !op.active()) return; diagnostics.traceCompactionLog(true, "unsettled operation at worker finish turn_id={d}", .{turn_id}); var feedback = compaction_activity.failure( if (self.isCancelRequested()) error.Cancelled else error.CompactionInterrupted, op.stage(), self.isCancelRequested(), ); // No transaction acknowledgement survived this fallback; do not promise rollback. if (op.stage() == .publication) feedback.publication = .uncertain; self.compaction_presentation.settle(op.id, feedback, io_mod.milliTimestamp()); }
fn queuedWorkCountLocked(self: *const WorkerRuntime) usize { return self.queued_prompts.items.len + @intFromBool(self.queued_context_compaction != null); }
pub fn deinit(self: *WorkerRuntime, alloc: std.mem.Allocator) void { if (self.pending_permission_response) |response| { self.discardPermissionResponse(response, "deinit"); } self.pending_permission_response = null; if (self.pending_permission_request_shared) |*request| request.deinit(alloc); self.pending_permission_request_shared = null; self.pending_permission_review = null; freeOwnedQuestionBatchOpt(alloc, self.pending_question_shared); self.pending_question_shared = null; freeQuestionResponse(alloc, self.pending_question_response); self.pending_question_response = .pending;
for (self.queued_prompts.items) |prompt| discardQueuedPrompt(alloc, prompt, &.{}); self.queued_prompts.deinit(alloc); types.freeHistoryTurnSlice(alloc, self.queued_history); if (self.queued_context_compaction) |task| freeContextCompactionTask(alloc, task);
self.clearActiveToolCallsLocked();
for (self.worker_events.items) |event| freeWorkerEvent(alloc, event); self.worker_events.deinit(alloc); }
pub fn requestStop(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); self.worker_stop_requested = true; if (self.queued_context_compaction) |task| { if (task.operation_id) |id| self.compaction_presentation.settle(id, .{ .outcome = .cancelled }, io_mod.milliTimestamp()); } self.stopCompactionActivityLocked(); self.worker_cond.broadcast(io_mod.getIo()); self.worker_mutex.unlock(io_mod.getIo()); }
pub fn requestShutdown(self: *WorkerRuntime) void { self.markExplicitCancellation(); self.requestStop(); }
pub fn requestCancel(self: *WorkerRuntime) void { debug_trace.logf("worker", "cancel requested processing={s} queued={d}", .{ if (self.worker_processing) "true" else "false", self.queuedPromptCount() }); debug_trace.eventf("interrupt", "cancel_requested", .{}, "processing={s} queued={d} active_tool_known=false", .{ if (self.worker_processing) "true" else "false", self.queuedPromptCount() }); self.markExplicitCancellation(); }
/// Interrupt the current gateway wait while preserving its recovery /// checkpoint. The ordinary cancel flag wakes the existing socket watcher; /// agent policy distinguishes the typed pause flag at the boundary. pub fn requestRecoveryPause(self: *WorkerRuntime) void { debug_trace.eventf("recovery", "pause_requested", .{}, "source=interactive_try_later", .{}); self.worker_recovery_pause_requested.store(true, .seq_cst); self.markExplicitCancellation(); }
fn markExplicitCancellation(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.steering_cancel_turn_id = null; self.worker_cancel_requested.store(true, .seq_cst); self.stopCompactionActivityLocked(); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn requestInteractiveCancel(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
const steering_pending = for (self.queued_prompts.items) |prompt| { if (prompt.delivery.activeTurnId() == self.active_turn_id or prompt.delivery.isContinuation()) break true; } else false; debug_trace.logf("worker", "cancel requested processing={s} queued={d} steering_pending={s}", .{ if (self.worker_processing) "true" else "false", self.queuedWorkCountLocked(), if (steering_pending) "true" else "false", }); debug_trace.eventf("interrupt", "cancel_requested", .{}, "processing={s} queued={d} steering_pending={s} active_tool_known=false", .{ if (self.worker_processing) "true" else "false", self.queuedWorkCountLocked(), if (steering_pending) "true" else "false", }); self.steering_cancel_turn_id = null; self.worker_cancel_requested.store(true, .seq_cst); self.stopCompactionActivityLocked(); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn cancelApprovalTurn(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
self.steering_cancel_turn_id = null; self.worker_cancel_requested.store(true, .seq_cst); self.stopCompactionActivityLocked(); _ = self.resolvePendingPermissionLocked( permission_request.OwnedPermissionResponse.init( std.heap.c_allocator, .deny, null, ), ); }
pub fn isCancelRequested(self: *const WorkerRuntime) bool { return self.worker_cancel_requested.load(.seq_cst); }
pub fn cancellationStopsTurn(self: *WorkerRuntime) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.worker_cancel_requested.load(.seq_cst) and self.steering_cancel_turn_id != self.active_turn_id; }
pub fn isConnectivityWaitActive(self: *const WorkerRuntime) bool { return self.worker_connectivity_wait_active.load(.seq_cst); }
pub fn pushEvent(self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent) !void { const owned = try dupeWorkerEvent(alloc, event); errdefer freeWorkerEvent(alloc, owned); try self.pushOwnedEvent(alloc, owned); }
/// Transfers an already-owned event into the queue on success. /// Every heap payload in `event` must have been allocated by `alloc`. pub fn pushOwnedEvent(self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); try self.worker_events.ensureUnusedCapacity(alloc, 1); var tool_transition = try self.prepareActiveToolTransitionLocked(alloc, event); errdefer tool_transition.deinit(alloc); self.worker_events.appendAssumeCapacity(event); self.applyActiveToolTransitionLocked(tool_transition); self.applyToolBoundaryPhaseLocked(event); self.applyRecoveryStateEvent(event); self.worker_cond.broadcast(io_mod.getIo()); }
/// Transfers already-owned events to the front of the queue on success. /// Every heap payload must have been allocated by `alloc`. pub fn prependOwnedEvents( self: *WorkerRuntime, alloc: std.mem.Allocator, events: []const WorkerEvent, ) !void { if (events.len == 0) return; self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); try self.worker_events.insertSlice(alloc, 0, events); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn pushTurnFinished(self: *WorkerRuntime, alloc: std.mem.Allocator, event: types.TurnFinished) !void { std.debug.assert(event.turn_id != 0);
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.worker_events.append(alloc, .{ .tool_lifecycle = .{ .turn_finished = event, } }) catch { self.latchFinalizationFailureLocked(.{ .turn_id = event.turn_id, .outcome = event.outcome, }); return error.TurnFinalizationDeliveryFailed; }; self.applyRecoveryStateEvent(.{ .tool_lifecycle = .{ .turn_finished = event, } }); self.worker_cond.broadcast(io_mod.getIo()); }
fn applyRecoveryStateEvent(self: *WorkerRuntime, event: WorkerEvent) void { switch (event) { .route_recovery_status => |status| { self.worker_connectivity_wait_active.store( status.action == .waiting_for_connectivity, .seq_cst, ); self.recovery_continuation_ready = status.action == .paused; }, .clear_route_recovery_status, .finish_prompt, .error_text, => { self.worker_connectivity_wait_active.store(false, .seq_cst); self.recovery_continuation_ready = false; }, .tool_lifecycle => |lifecycle| switch (lifecycle) { .turn_finished => self.worker_connectivity_wait_active.store(false, .seq_cst), .provisional, .authoritative_started, .progress, .terminal => return, }, else => return, } }
pub fn latchFinalizationFailure(self: *WorkerRuntime, failure: FinalizationFailure) void { std.debug.assert(failure.turn_id != 0);
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.latchFinalizationFailureLocked(failure); }
fn latchFinalizationFailureLocked(self: *WorkerRuntime, failure: FinalizationFailure) void { if (self.finalization_failure == null) { self.finalization_failure = failure; } else { debug_trace.logf( "worker", "duplicate finalization failure ignored turn_id={d} first_turn_id={d}", .{ failure.turn_id, self.finalization_failure.?.turn_id }, ); } self.worker_stop_requested = true; self.steering_cancel_turn_id = null; self.worker_cancel_requested.store(true, .seq_cst); self.stopCompactionActivityLocked(); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn interactiveAdmissionSnapshot(self: *WorkerRuntime) InteractiveAdmissionSnapshot { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
if (self.finalization_failure) |failure| { return .{ .finalization_failed = failure }; } if (self.worker_stop_requested) return .stopped; return .open; }
pub fn takeEventBatch(self: *WorkerRuntime) WorkerEventBatch { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const events = self.worker_events; self.worker_events = .empty; return .{ .events = events, .cancelled_turn_id = if (self.worker_cancel_requested.load(.seq_cst) and self.active_turn_id != 0 and self.steering_cancel_turn_id != self.active_turn_id) self.active_turn_id else null, }; }
pub fn takeEvents(self: *WorkerRuntime) std.ArrayList(WorkerEvent) { return self.takeEventBatch().events; }
pub fn enqueuePrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt) !void { try self.admitPrompt(alloc, prompt, false); }
pub fn admitInteractivePrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt) !void { try self.admitPrompt(alloc, prompt, true); }
/// Takes prompt ownership only on true. Does not cancel an in-flight operation /// or fall back to the ordinary FIFO of an ephemeral direct worker. pub fn admitActiveSteering(self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt) !bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (!self.worker_processing or self.active_turn_id == 0 or self.direct_steering_closed or self.worker_stop_requested or self.worker_cancel_requested.load(.seq_cst)) return false; var queued = prompt; queued.delivery = .{ .active_turn = self.active_turn_id }; try self.enqueuePromptLocked(alloc, queued); return true; }
pub fn enqueueContextCompaction( self: *WorkerRuntime, task: ContextCompactionTask, ) !void { var queued = task; if (queued.turn_id == 0) queued.turn_id = debug_trace.nextTurnId();
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.finalization_failure != null) return error.TurnFinalizationDeliveryFailed; if (self.worker_stop_requested) return error.WorkerStopped; if (self.worker_processing or self.queued_prompts.items.len > 0 or self.queued_context_compaction != null) { return error.WorkerBusy; } if (queued.operation_id == null) queued.operation_id = self.compaction_presentation.begin(.manual, queued.turn_id, io_mod.milliTimestamp()); self.compaction_presentation.queued(queued.operation_id.?, queued.turn_id, io_mod.milliTimestamp()); self.queued_context_compaction = queued; debug_trace.eventf( "worker", "context_compaction_enqueue", .{ .turn_id = queued.turn_id }, "queue_depth=1", .{}, ); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn contextCompactionStatus(self: *WorkerRuntime) ContextCompactionStatus { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.active_context_compaction) return .running; if (self.queued_context_compaction != null) return .queued; return .idle; }
pub fn cancelContextCompaction( self: *WorkerRuntime, alloc: std.mem.Allocator, ) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); if (self.queued_context_compaction) |task| { self.queued_context_compaction = null; if (task.operation_id) |id| self.compaction_presentation.settle(id, .{ .outcome = .cancelled }, io_mod.milliTimestamp()); self.worker_cond.broadcast(io_mod.getIo()); self.worker_mutex.unlock(io_mod.getIo()); freeContextCompactionTask(alloc, task); debug_trace.eventf( "context_compaction", "cancelled", .{}, "phase=queued", .{}, ); return true; } if (self.active_context_compaction) { self.worker_cancel_requested.store(true, .seq_cst); self.stopCompactionActivityLocked(); self.worker_cond.broadcast(io_mod.getIo()); self.worker_mutex.unlock(io_mod.getIo()); debug_trace.eventf( "context_compaction", "cancel_requested", .{}, "phase=running", .{}, ); return true; } self.worker_mutex.unlock(io_mod.getIo()); return false; }
/// Transfers `prompt` to the active turn when steering is requested and the /// turn still accepts guidance. Otherwise it enters the ordinary FIFO. /// A steer interrupts the model stream immediately only when no wait /// boundary is active; a running tool or an in-flight compaction defers /// delivery to the next steering boundary instead of being cancelled. fn admitPrompt( self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt, steer_if_active: bool, ) !void { var queued = prompt; if (queued.turn_id == 0) queued.turn_id = debug_trace.nextTurnId();
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.finalization_failure != null) return error.TurnFinalizationDeliveryFailed; if (self.worker_stop_requested) return error.WorkerStopped; if (queued.recovery_checkpoint != null and !recoveryContinuationAdmission( self.worker_processing, self.queuedWorkCountLocked(), self.recovery_continuation_ready, )) { return error.RecoveryBusy; } queued.agent_settings = self.agent_turn_settings; var interrupt_after_admission = false; const explicitly_cancelled = self.worker_cancel_requested.load(.seq_cst) and self.steering_cancel_turn_id != self.active_turn_id; if (steer_if_active and self.worker_processing and self.active_turn_id != 0 and !explicitly_cancelled) { queued.delivery = .{ .active_turn = self.active_turn_id }; interrupt_after_admission = !self.hasSteeringWaitBoundaryLocked(); } try self.enqueuePromptLocked(alloc, queued); if (steer_if_active and self.worker_processing) { debug_trace.eventf("worker", "steering_admission", .{ .turn_id = self.active_turn_id }, "queued_turn_id={d} targeted={} plain={} tool_boundary={} compaction_active={} interrupt_model={}", .{ queued.turn_id, queued.delivery.activeTurnId() == self.active_turn_id, sameTurnSteeringEligible(queued), self.hasActiveToolBoundaryLocked(), self.compactionInFlightLocked(), interrupt_after_admission }); } if (interrupt_after_admission) { self.steering_cancel_turn_id = if (sameTurnSteeringEligible(queued)) self.active_turn_id else null; self.worker_cancel_requested.store(true, .seq_cst); } }
fn sameTurnSteeringEligible(prompt: QueuedPrompt) bool { return prompt.images.len == 0 and prompt.skill_bindings.len == 0 and prompt.skill_display_spans.len == 0; }
fn enqueuePromptLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, queued_value: QueuedPrompt) !void { var queued = queued_value; try self.queued_prompts.ensureUnusedCapacity(alloc, 1); const supplied_history = queued.history; queued.history = &.{}; if (self.queued_prompts.items.len == 0) { std.debug.assert(self.queued_history.len == 0); self.queued_history = supplied_history; } else { types.freeHistoryTurnSlice(alloc, supplied_history); } self.queued_prompts.appendAssumeCapacity(queued); debug_trace.logf( "worker", "queued prompt bytes={d} queue_depth={d} fast_mode={s} effort={s}", .{ queued.prompt.len, self.queuedWorkCountLocked(), if (queued.agent_settings.fast_mode) "true" else "false", queued.agent_settings.effort.label() }, ); debug_trace.eventf( "worker", "prompt_enqueue", .{ .turn_id = queued.turn_id }, "prompt_bytes={d} queue_depth={d} fast_mode={s} effort={s}", .{ queued.prompt.len, self.queuedWorkCountLocked(), if (queued.agent_settings.fast_mode) "true" else "false", queued.agent_settings.effort.label() }, ); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn hasPendingPlainSteering(self: *WorkerRuntime) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (!self.worker_processing or self.active_turn_id == 0 or self.worker_stop_requested) return false; var pending = false; for (self.queued_prompts.items) |prompt| { if (prompt.delivery.activeTurnId() != self.active_turn_id) continue; if (!sameTurnSteeringEligible(prompt)) return false; pending = true; } return pending; }
/// Classifies and drains steering at one observed agent boundary. Returned /// text is allocator-owned; handoff leaves the queued prompt worker-owned. pub fn takeSteeringBoundary( self: *WorkerRuntime, alloc: std.mem.Allocator, turn_id: u64, kind: SteeringBoundaryKind, ) !SteeringBoundaryResult { return self.takeSteeringBoundaryInto(alloc, alloc, turn_id, kind); }
/// Queue/event storage uses alloc; returned guidance uses result_alloc. pub fn takeSteeringBoundaryInto( self: *WorkerRuntime, alloc: std.mem.Allocator, result_alloc: std.mem.Allocator, turn_id: u64, kind: SteeringBoundaryKind, ) !SteeringBoundaryResult { const result = try self.takeSteeringBoundaryIntoInner(alloc, result_alloc, turn_id, kind); debug_trace.eventf("worker", "steering_boundary_check", .{ .turn_id = turn_id }, "kind={s} outcome={s}", .{ @tagName(kind), @tagName(result) }); return result; }
fn takeSteeringBoundaryIntoInner( self: *WorkerRuntime, alloc: std.mem.Allocator, result_alloc: std.mem.Allocator, turn_id: u64, kind: SteeringBoundaryKind, ) !SteeringBoundaryResult { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const cancel_requested = self.worker_cancel_requested.load(.seq_cst); if (!self.worker_processing or self.active_turn_id != turn_id) { return if (kind == .cancelled and cancel_requested) .interrupt else .none; }
switch (kind) { .cancelled => { if (!cancel_requested) return .none; if (self.steering_cancel_turn_id != turn_id) return .interrupt; const messages = try self.takeSteeringLocked(alloc, result_alloc, turn_id); if (messages.len == 0) return .interrupt; self.steering_cancel_turn_id = null; self.worker_cancel_requested.store(false, .seq_cst); return .{ .continue_turn = messages }; }, .model, .finalizing => { if (cancel_requested) return .none; for (self.queued_prompts.items) |prompt| { if (prompt.delivery.activeTurnId() == turn_id and !sameTurnSteeringEligible(prompt)) return .handoff; } const messages = try self.takeSteeringLocked(alloc, result_alloc, turn_id); if (kind == .finalizing and messages.len == 0) self.direct_steering_closed = true; return if (messages.len == 0) .none else .{ .continue_turn = messages }; }, } }
fn takeSteeringLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, result_alloc: std.mem.Allocator, turn_id: u64) ![][]u8 { var steering_count: usize = 0; for (self.queued_prompts.items) |prompt| { if (prompt.delivery.activeTurnId() != turn_id) continue; if (!sameTurnSteeringEligible(prompt)) return &.{}; steering_count += 1; } if (steering_count == 0) return &.{};
const messages = try result_alloc.alloc([]u8, steering_count); var copied: usize = 0; errdefer { for (messages[0..copied]) |text| result_alloc.free(text); result_alloc.free(messages); } const events = try alloc.alloc(WorkerEvent, steering_count); var event_count: usize = 0; errdefer { for (events[0..event_count]) |event| freeWorkerEvent(alloc, event); alloc.free(events); } for (self.queued_prompts.items) |prompt| { if (prompt.delivery.activeTurnId() != turn_id) continue; messages[copied] = try result_alloc.dupe(u8, prompt.prompt); copied += 1; if (prompt.steering_receipt == null) { events[event_count] = .{ .append_user_feedback = try alloc.dupe(u8, prompt.prompt), }; event_count += 1; } } try self.worker_events.ensureUnusedCapacity(alloc, event_count); for (events[0..event_count]) |event| self.worker_events.appendAssumeCapacity(event); alloc.free(events);
var index: usize = 0; while (index < self.queued_prompts.items.len) { if (self.queued_prompts.items[index].delivery.activeTurnId() != turn_id) { index += 1; continue; } const prompt = self.queued_prompts.orderedRemove(index); if (prompt.steering_receipt) |receipt| { receipt.state.store(.applied, .seq_cst); debug_trace.eventf("subagent", "feedback_applied", .{}, "operation={s}", .{receipt.operation_id}); } freeQueuedPrompt(alloc, prompt); } if (self.queued_prompts.items.len == 0) { types.freeHistoryTurnSlice(alloc, self.queued_history); self.queued_history = &.{}; } debug_trace.eventf("worker", "prompt_steering_consumed", .{ .turn_id = turn_id }, "count={d}", .{messages.len}); self.worker_cond.broadcast(io_mod.getIo()); return messages; }
/// Hold turn admission only when no turn is active or queued. pub fn tryHoldTurnStart(self: *WorkerRuntime) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.worker_processing or self.queuedWorkCountLocked() > 0 or self.turn_start_held) { return false; } self.turn_start_held = true; debug_trace.logf("worker", "turn start hold acquired", .{}); return true; }
pub fn releaseTurnStartHold(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (!self.turn_start_held) return; self.turn_start_held = false; debug_trace.logf("worker", "turn start hold released", .{}); self.worker_cond.broadcast(io_mod.getIo()); }
pub fn removeQueuedPrompt( self: *WorkerRuntime, alloc: std.mem.Allocator, turn_id: u64, retained_images: []const types.ImageAttachment, ) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); var removed: ?QueuedPrompt = null; for (self.queued_prompts.items, 0..) |queued, index| { if (queued.turn_id != turn_id) continue; removed = self.queued_prompts.orderedRemove(index); break; } if (removed != null and self.queued_prompts.items.len == 0) { types.freeHistoryTurnSlice(alloc, self.queued_history); self.queued_history = &.{}; } const remaining = self.queuedWorkCountLocked(); self.worker_mutex.unlock(io_mod.getIo());
const prompt = removed orelse return false; discardQueuedPrompt(alloc, prompt, retained_images); debug_trace.eventf( "worker", "queued_prompt_removed", .{ .turn_id = turn_id }, "remaining={d}", .{remaining}, ); return true; }
/// Removes the newest queued text-only steering prompt that still waits for /// a tool or compaction boundary and returns an allocator-owned copy of its /// text so the composer can restore it for editing. Returns null when /// nothing is safely retractable: no active turn, no wait boundary (an /// immediate steer is already committed to its interrupt), a cancel /// already in flight, or only handoff-class steers with images or skill /// bindings. pub fn popQueuedSteerForEdit(self: *WorkerRuntime, alloc: std.mem.Allocator) !?[]u8 { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (!self.worker_processing or self.active_turn_id == 0) return null; if (!self.hasSteeringWaitBoundaryLocked()) return null; if (self.worker_cancel_requested.load(.seq_cst)) return null;
var retractable: ?usize = null; var index: usize = self.queued_prompts.items.len; while (index > 0) { index -= 1; const prompt = self.queued_prompts.items[index]; if (prompt.delivery.activeTurnId() != self.active_turn_id) continue; if (!sameTurnSteeringEligible(prompt)) continue; retractable = index; break; } const remove_index = retractable orelse return null;
const text = try alloc.dupe(u8, self.queued_prompts.items[remove_index].prompt); const prompt = self.queued_prompts.orderedRemove(remove_index); if (self.queued_prompts.items.len == 0) { types.freeHistoryTurnSlice(alloc, self.queued_history); self.queued_history = &.{}; } const turn_id = prompt.turn_id; const remaining = self.queuedWorkCountLocked(); freeQueuedPrompt(alloc, prompt); debug_trace.eventf( "worker", "prompt_steering_retracted", .{ .turn_id = turn_id }, "remaining={d}", .{remaining}, ); self.worker_cond.broadcast(io_mod.getIo()); return text; }
pub fn waitAndTakeNextPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator) !?QueuedPrompt { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
while ((self.queued_prompts.items.len == 0 or self.turn_start_held) and !self.worker_stop_requested) { self.worker_processing = false; self.active_turn_id = 0; self.tool_phase_step_id = null; self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break; } return self.takeNextPromptLocked(alloc); }
pub fn waitAndTakeNextWork(self: *WorkerRuntime, alloc: std.mem.Allocator) !?WorkItem { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
while (((self.queued_prompts.items.len == 0 and self.queued_context_compaction == null) or self.turn_start_held) and !self.worker_stop_requested) { self.worker_processing = false; self.active_turn_id = 0; self.tool_phase_step_id = null; self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break; } return self.takeNextWorkLocked(alloc); }
/// Nonblocking queue take for single-threaded hosts. Returns null while the /// queue is empty, held, or stopped. pub fn tryTakeNextPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator) !?QueuedPrompt { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.queued_prompts.items.len == 0 or self.turn_start_held or self.worker_stop_requested) { return null; } return self.takeNextPromptLocked(alloc); }
pub fn tryTakeNextWork(self: *WorkerRuntime, alloc: std.mem.Allocator) !?WorkItem { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if ((self.queued_prompts.items.len == 0 and self.queued_context_compaction == null) or self.turn_start_held or self.worker_stop_requested) { return null; } return self.takeNextWorkLocked(alloc); }
fn takeNextWorkLocked(self: *WorkerRuntime, alloc: std.mem.Allocator) !?WorkItem { if (self.worker_stop_requested) return null; if (self.queued_context_compaction) |task| { self.queued_context_compaction = null; self.worker_cancel_requested.store(false, .seq_cst); self.worker_recovery_pause_requested.store(false, .seq_cst); self.worker_connectivity_wait_active.store(false, .seq_cst); self.recovery_continuation_ready = false; self.worker_processing = true; self.active_turn_id = task.turn_id; self.tool_phase_step_id = null; self.active_context_compaction = true; if (task.operation_id) |id| self.compaction_presentation.running(id, .preparation); debug_trace.eventf( "worker", "context_compaction_begin", .{ .turn_id = task.turn_id }, "remaining_queue={d} cancel_reset=true", .{self.queuedWorkCountLocked()}, ); return .{ .compact_context = task }; } const prompt = (try self.takeNextPromptLocked(alloc)) orelse return null; return .{ .prompt = prompt }; }
fn takeNextPromptLocked(self: *WorkerRuntime, alloc: std.mem.Allocator) !?QueuedPrompt { if (self.worker_stop_requested or self.queued_prompts.items.len == 0) return null;
const queued = self.queued_prompts.items[0]; const transfer_history = self.queued_prompts.items.len == 1; const job_history = if (transfer_history) self.queued_history else try session_runtime.snapshotOwnedContextHistory( alloc, self.queued_history, 0, 0, ); errdefer if (!transfer_history) { types.freeHistoryTurnSlice(alloc, job_history); }; if (queued.recovery_checkpoint == null and queued.user_prompt_already_presented) { try self.worker_events.append(alloc, .{ .begin_presented_prompt = queued.turn_id, }); } else if (queued.recovery_checkpoint == null) { const begin_prompt = try types.dupeUserTurn(alloc, .{ .text = queued.prompt, .images = queued.images }); errdefer types.freeUserTurn(alloc, begin_prompt); if (queued.skill_bindings.len > 0 or queued.skill_display_spans.len > 0) { const skill_bindings = try dupeSkillBindings(alloc, queued.skill_bindings); errdefer freeSkillBindings(alloc, skill_bindings); const skill_display_spans = try dupeSkillDisplaySpans(alloc, queued.skill_display_spans); errdefer freeSkillDisplaySpans(alloc, skill_display_spans); try self.worker_events.append(alloc, .{ .begin_prompt_with_skill_bindings = .{ .prompt = begin_prompt, .skill_bindings = skill_bindings, .skill_display_spans = skill_display_spans, }, }); } else { try self.worker_events.append(alloc, .{ .begin_prompt = begin_prompt }); } }
var job = self.queued_prompts.orderedRemove(0); if (transfer_history) self.queued_history = &.{}; job.history = job_history; self.steering_cancel_turn_id = null; self.worker_cancel_requested.store(false, .seq_cst); self.worker_recovery_pause_requested.store(false, .seq_cst); self.worker_connectivity_wait_active.store(false, .seq_cst); self.recovery_continuation_ready = false; self.worker_processing = true; self.active_turn_id = job.turn_id; self.tool_phase_step_id = null; if (job.delivery.isContinuation()) { for (self.queued_prompts.items) |*prompt| { if (!prompt.delivery.isContinuation() or !sameTurnSteeringEligible(prompt.*)) continue; prompt.delivery = .{ .active_turn = job.turn_id }; } } self.active_context_compaction = false; debug_trace.logf( "worker", "begin prompt bytes={d} remaining_queue={d} fast_mode={s} effort={s}", .{ job.prompt.len, self.queuedWorkCountLocked(), if (job.agent_settings.fast_mode) "true" else "false", job.agent_settings.effort.label() }, ); debug_trace.eventf( "worker", "worker_begin", .{ .turn_id = job.turn_id }, "prompt_bytes={d} remaining_queue={d} cancel_reset=true fast_mode={s} effort={s}", .{ job.prompt.len, self.queuedWorkCountLocked(), if (job.agent_settings.fast_mode) "true" else "false", job.agent_settings.effort.label() }, ); return job; }
pub fn finishProcessing(self: *WorkerRuntime) void { debug_trace.logf("worker", "finish processing queued={d}", .{self.queuedPromptCount()}); self.worker_mutex.lockUncancelable(io_mod.getIo()); // Close steering admission without moving entries, preserving the exact // order in which steering and ordinary prompts were submitted. const finished_turn_id = self.active_turn_id; self.finishCompactionActivityLocked(finished_turn_id); for (self.queued_prompts.items) |*prompt| { if (prompt.delivery.activeTurnId() == finished_turn_id) { prompt.delivery = .continuation; debug_trace.eventf("worker", "steering_deferred_to_next_turn", .{ .turn_id = finished_turn_id }, "queued_turn_id={d} prompt_bytes={d}", .{ prompt.turn_id, prompt.prompt.len }); } } self.worker_processing = false; self.active_turn_id = 0; self.steering_cancel_turn_id = null; self.clearActiveToolCallsLocked(); self.tool_phase_step_id = null; self.active_context_compaction = false; self.worker_connectivity_wait_active.store(false, .seq_cst); self.worker_cond.broadcast(io_mod.getIo()); self.worker_mutex.unlock(io_mod.getIo()); }
/// Marks an agent turn that is executed directly rather than dequeued by /// the interactive worker loop. The caller must pair a successful begin /// with `finishProcessing`. pub fn beginDirectProcessing(self: *WorkerRuntime, turn_id: u64) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.worker_processing or self.worker_stop_requested) return false; self.direct_steering_closed = false; self.worker_cancel_requested.store(false, .seq_cst); self.worker_recovery_pause_requested.store(false, .seq_cst); self.worker_connectivity_wait_active.store(false, .seq_cst); self.worker_processing = true; self.active_turn_id = turn_id; self.tool_phase_step_id = null; return true; }
const ActiveToolTransition = union(enum) { none, started: []u8, terminal: []const u8, clear,
fn deinit(self: *ActiveToolTransition, alloc: std.mem.Allocator) void { switch (self.*) { .started => |call_id| alloc.free(call_id), .none, .terminal, .clear => {}, } self.* = .none; } };
fn prepareActiveToolTransitionLocked( self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent, ) !ActiveToolTransition { return switch (event) { .tool_lifecycle => |lifecycle| switch (lifecycle) { .authoritative_started => |started| blk: { if (started.id.turn_id != self.active_turn_id or self.active_tool_calls.contains(started.id.call_id)) { break :blk .none; } if (self.active_tool_allocator == null) { self.active_tool_allocator = alloc; } try self.active_tool_calls.ensureUnusedCapacity(alloc, 1); break :blk .{ .started = try alloc.dupe(u8, started.id.call_id) }; }, .terminal => |terminal| if (terminal.id.turn_id == self.active_turn_id and self.active_tool_calls.contains(terminal.id.call_id)) .{ .terminal = terminal.id.call_id } else .none, .turn_finished => |finished| if (finished.turn_id == self.active_turn_id) .clear else .none, .provisional, .progress => .none, }, else => .none, }; }
fn applyActiveToolTransitionLocked( self: *WorkerRuntime, transition: ActiveToolTransition, ) void { switch (transition) { .none => {}, .started => |call_id| { self.active_tool_calls.putAssumeCapacity(call_id, {}); }, .terminal => |call_id| { const removed = self.active_tool_calls.fetchRemove(call_id) orelse return; std.debug.assert(self.active_tool_allocator != null); const alloc = self.active_tool_allocator.?; alloc.free(removed.key); }, .clear => self.clearActiveToolCallsLocked(), } }
fn applyToolBoundaryPhaseLocked(self: *WorkerRuntime, event: WorkerEvent) void { switch (event) { .turn_phase_update => |update| if (update.turn_id == self.active_turn_id) { switch (update.phase) { .running => self.tool_phase_step_id = update.step_id, .waiting_for_subagent => {}, .thinking, .generating => if (self.tool_phase_step_id != update.step_id) { self.tool_phase_step_id = null; }, } }, else => {}, } }
fn hasActiveToolBoundaryLocked(self: *const WorkerRuntime) bool { return self.tool_phase_step_id != null or self.active_tool_calls.count() > 0; }
/// True while a context compaction owns the worker: either the compaction /// work item is running, or the active turn is inside a compaction /// transaction. Compaction reports through the activity channel rather /// than the tool/phase event feed, so it is consulted separately. fn compactionInFlightLocked(self: *const WorkerRuntime) bool { if (self.active_context_compaction) return true; if (self.compaction_presentation.snapshot.operation) |operation| return operation.active(); return false; }
/// Steering must wait whenever the active work cannot absorb guidance right /// now: a running tool or tool-producing step, or a compaction in flight. /// Cancelling either would discard in-progress work the user expects to /// finish; the message is delivered at the next boundary instead. fn hasSteeringWaitBoundaryLocked(self: *const WorkerRuntime) bool { return self.hasActiveToolBoundaryLocked() or self.compactionInFlightLocked(); }
fn clearActiveToolCallsLocked(self: *WorkerRuntime) void { const alloc = self.active_tool_allocator orelse { std.debug.assert(self.active_tool_calls.count() == 0); return; }; var keys = self.active_tool_calls.keyIterator(); while (keys.next()) |key| alloc.free(@constCast(key.*)); self.active_tool_calls.deinit(alloc); self.active_tool_calls = .empty; self.active_tool_allocator = null; }
pub fn waitUntilIdle(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); while (self.worker_processing) { self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break; } }
pub fn isProcessing(self: *WorkerRuntime) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.worker_processing; }
pub fn snapshotState( self: *WorkerRuntime, alloc: std.mem.Allocator, ) permission_request.RequestCloneError!StateSnapshot { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const cancel_requested = self.worker_cancel_requested.load(.seq_cst); return .{ .processing = self.worker_processing, .active_turn_id = self.active_turn_id, .queued_count = self.queuedWorkCountLocked(), .pending_event_count = self.worker_events.items.len, .cancel_requested = cancel_requested, .cancel_continues_turn = cancel_requested and self.steering_cancel_turn_id == self.active_turn_id, .pending_permission_request = if (self.permissionRequestAwaitingDecisionLocked()) blk: { const request = &self.pending_permission_request_shared.?; break :blk try permission_request.OwnedPermissionRequest.dupe(alloc, request.view()); } else null, .pending_permission_review = if (self.permissionRequestAwaitingDecisionLocked()) if (self.pending_permission_review) |review| .{ .request_id = self.pending_permission_request_shared.?.id, .review = review, } else null else null, }; }
pub fn activeTurnId(self: *WorkerRuntime) u64 { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.active_turn_id; }
/// Returns one coherent, allocator-owned steering presentation snapshot. pub fn snapshotSteeringPresentation(self: *WorkerRuntime, alloc: std.mem.Allocator) !SteeringPresentationSnapshot { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
var steering_count: usize = 0; for (self.queued_prompts.items) |prompt| { if (prompt.delivery.isSteering()) steering_count += 1; } var snapshot: SteeringPresentationSnapshot = .{ .waits_for_boundary = steering_count > 0 and self.hasSteeringWaitBoundaryLocked(), }; errdefer snapshot.deinit(alloc); snapshot.pending_feedback = pending: { var count: usize = 0; for (self.worker_events.items) |event| { if (event == .append_user_feedback) count += 1; } if (count == 0) break :pending &.{}; const feedback = try alloc.alloc([]u8, count); errdefer alloc.free(feedback); var written: usize = 0; errdefer for (feedback[0..written]) |message| alloc.free(message); for (self.worker_events.items) |event| { if (event == .append_user_feedback) { feedback[written] = try alloc.dupe(u8, event.append_user_feedback); written += 1; } } break :pending feedback; }; if (steering_count == 0) return snapshot;
const messages = try alloc.alloc([]u8, steering_count); var copied: usize = 0; errdefer { for (messages[0..copied]) |message| alloc.free(message); alloc.free(messages); } for (self.queued_prompts.items) |prompt| { if (!prompt.delivery.isSteering()) continue; messages[copied] = try alloc.dupe(u8, prompt.prompt); copied += 1; } snapshot.messages = messages; return snapshot; }
pub fn queuedPromptCount(self: *WorkerRuntime) usize { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.queuedWorkCountLocked(); }
pub fn clearQueuedPrompts( self: *WorkerRuntime, alloc: std.mem.Allocator, retained_images: []const types.ImageAttachment, ) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
self.clearQueuedPromptsLocked(alloc, retained_images); }
/// Clears the queue while atomically transferring deletion responsibility /// for a matching active or finished turn to the supplied retained images. pub fn clearQueuedPromptsForSessionTransition( self: *WorkerRuntime, alloc: std.mem.Allocator, turn_id: u64, retained_images: []const types.ImageAttachment, ) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
if (retained_images.len > 0) { _ = self.preservePromptSnapshotsLocked(turn_id, retained_images); } self.clearQueuedPromptsLocked(alloc, retained_images); }
fn clearQueuedPromptsLocked( self: *WorkerRuntime, alloc: std.mem.Allocator, retained_images: []const types.ImageAttachment, ) void { const dropped = self.queuedWorkCountLocked(); if (dropped > 0) { debug_trace.logf("worker", "clear queued prompts dropped={d}", .{dropped}); debug_trace.eventf("worker", "queued_prompts_cleared", .{}, "dropped={d}", .{dropped}); } for (self.queued_prompts.items) |prompt| { discardQueuedPrompt(alloc, prompt, retained_images); } self.queued_prompts.clearRetainingCapacity(); types.freeHistoryTurnSlice(alloc, self.queued_history); self.queued_history = &.{}; if (self.queued_context_compaction) |task| { if (task.operation_id) |id| self.compaction_presentation.settle(id, .{ .outcome = .cancelled }, io_mod.milliTimestamp()); freeContextCompactionTask(alloc, task); self.queued_context_compaction = null; } self.worker_cond.broadcast(io_mod.getIo()); }
pub fn discardEvents(self: *WorkerRuntime, alloc: std.mem.Allocator) void { var events = self.takeEvents(); defer events.deinit(alloc); for (events.items) |event| freeWorkerEvent(alloc, event); }
pub fn syncQueuedPromptModel(self: *WorkerRuntime, alloc: std.mem.Allocator, model: []const u8) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); for (self.queued_prompts.items) |*prompt| { const next_model = try alloc.dupe(u8, model); alloc.free(prompt.model); prompt.model = next_model; } }
pub fn syncQueuedPromptPermissionSnapshot(self: *WorkerRuntime, snapshot: PermissionSnapshot) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); for (self.queued_prompts.items) |*prompt| { prompt.permission_mode = snapshot.mode; } }
pub fn syncQueuedPromptFastMode(self: *WorkerRuntime, enabled: bool) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.agent_turn_settings.fast_mode = enabled; for (self.queued_prompts.items) |*prompt| prompt.agent_settings.fast_mode = enabled; }
pub fn syncQueuedPromptEffort(self: *WorkerRuntime, effort: types.ReasoningEffort) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.agent_turn_settings.effort = effort; for (self.queued_prompts.items) |*prompt| prompt.agent_settings.effort = effort; }
pub fn setActiveAgentTurnSettings(self: *WorkerRuntime, settings: AgentTurnSettings) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.active_agent_turn_settings = settings; }
pub fn clearActiveAgentTurnSettings(self: *WorkerRuntime) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); self.active_agent_turn_settings = null; }
pub fn effectiveAgentTurnSettings(self: *WorkerRuntime) AgentTurnSettings { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.active_agent_turn_settings orelse self.agent_turn_settings; }
pub fn syncQueuedPromptPermissionState(self: *WorkerRuntime, alloc: std.mem.Allocator, grants: []const types.PermissionGrant, snapshot: PermissionSnapshot) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); for (self.queued_prompts.items) |*prompt| { const next_grants = try types.dupePermissionGrantSlice(alloc, grants); types.freePermissionGrantSlice(alloc, prompt.grants); prompt.grants = next_grants; prompt.permission_mode = snapshot.mode; } }
pub fn propagateHistoryTurn(self: *WorkerRuntime, alloc: std.mem.Allocator, turn: types.HistoryTurn, max_history_turns: usize) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
try self.propagateHistoryTurnLocked(alloc, turn, max_history_turns, null); }
pub fn commitContextCompaction( self: *WorkerRuntime, alloc: std.mem.Allocator, summary: types.CompactedSummaryHistoryTurn, active_prefix: ?types.AssistantHistoryTurn, retained_from: ?types.ContextHistoryCut, max_history_turns: usize, ) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const result = try self.awaitHistoryPublicationLocked(alloc, .{ .context_compaction = .{ .turn_id = self.active_turn_id, .summary = summary, .active_prefix = active_prefix, .retained_from = retained_from, .max_history_turns = max_history_turns, }, }); std.debug.assert(result == null); }
pub fn prepareFreshPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, preparation: FreshPromptPreparation) !FreshPromptHistory { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); var request = preparation; request.turn_id = self.active_turn_id; return (try self.awaitHistoryPublicationLocked(alloc, .{ .prepare_fresh_prompt = request })) orelse error.InvalidHistoryPublication; }
fn awaitHistoryPublicationLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent) !?FreshPromptHistory { if (self.worker_stop_requested or self.isCancelRequested()) return error.Aborted; std.debug.assert(self.history_publication_response == .idle); const owned_event = try dupeWorkerEvent(alloc, event); var owns_event = true; defer if (owns_event) freeWorkerEvent(alloc, owned_event); try self.worker_events.ensureUnusedCapacity(alloc, 1); self.history_publication_response = .{ .waiting = self.active_turn_id }; defer { if (self.history_publication_response == .committed) { if (self.history_publication_response.committed) |*snapshot| snapshot.deinit(alloc); } self.history_publication_response = .idle; } self.worker_events.appendAssumeCapacity(owned_event); owns_event = false; self.worker_cond.broadcast(io_mod.getIo());
while (self.history_publication_response == .waiting and !self.worker_stop_requested and !self.isCancelRequested()) { try self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex); } switch (self.history_publication_response) { .committed => |result| { self.history_publication_response = .idle; return result; }, .failed => |err| return err, .waiting, .idle => return error.Aborted, } }
pub fn failPendingHistoryPublication(self: *WorkerRuntime, err: anyerror) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.history_publication_response != .waiting) return; self.history_publication_response = .{ .failed = err }; self.worker_cond.broadcast(io_mod.getIo()); }
pub fn resolveContextCompaction( self: *WorkerRuntime, alloc: std.mem.Allocator, event: ContextCompaction, ctx: *anyopaque, handler: ContextCompactionHandler, ) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.history_publication_response != .waiting or self.history_publication_response.waiting != event.turn_id) { debug_trace.logf("worker", "compaction delivery dropped reason=stale turn_id={d}", .{event.turn_id}); return; } defer self.worker_cond.broadcast(io_mod.getIo()); if (self.worker_stop_requested or self.isCancelRequested()) { self.history_publication_response = .{ .failed = error.Aborted }; return; } self.publishContextCompactionLocked(alloc, event, ctx, handler) catch |err| { self.history_publication_response = .{ .failed = err }; return; }; self.history_publication_response = .{ .committed = null }; }
pub fn resolveFreshPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, event: FreshPromptPreparation, ctx: *anyopaque, handler: FreshPromptHandler) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.history_publication_response != .waiting or self.history_publication_response.waiting != event.turn_id) { debug_trace.logf("worker", "fresh prompt preparation dropped reason=stale turn_id={d}", .{event.turn_id}); return; } defer self.worker_cond.broadcast(io_mod.getIo()); if (self.worker_stop_requested or self.isCancelRequested()) { self.history_publication_response = .{ .failed = error.Aborted }; return; } const snapshot = self.publishFreshPromptLocked(alloc, event, ctx, handler) catch |err| { self.history_publication_response = .{ .failed = err }; return; }; self.history_publication_response = .{ .committed = snapshot }; }
pub fn publishFreshPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, event: FreshPromptPreparation, ctx: *anyopaque, handler: FreshPromptHandler) !FreshPromptHistory { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); return self.publishFreshPromptLocked(alloc, event, ctx, handler); }
fn publishFreshPromptLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, event: FreshPromptPreparation, ctx: *anyopaque, handler: FreshPromptHandler) !FreshPromptHistory { const prior_turn = event.prior_turn orelse return handler(ctx, event); const Publication = struct { ctx: *anyopaque, handler: FreshPromptHandler, event: FreshPromptPreparation, result: ?FreshPromptHistory = null,
fn commit(raw: *anyopaque) !void { const self_: *@This() = @ptrCast(@alignCast(raw)); self_.result = try self_.handler(self_.ctx, self_.event); } }; var publication: Publication = .{ .ctx = ctx, .handler = handler, .event = event }; try self.propagateHistoryTurnLocked(alloc, prior_turn, 0, .{ .ctx = &publication, .commit_fn = Publication.commit }); return publication.result.?; }
pub fn publishContextCompaction( self: *WorkerRuntime, alloc: std.mem.Allocator, event: ContextCompaction, ctx: *anyopaque, handler: ContextCompactionHandler, ) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); try self.publishContextCompactionLocked(alloc, event, ctx, handler); }
fn publishContextCompactionLocked( self: *WorkerRuntime, alloc: std.mem.Allocator, event: ContextCompaction, ctx: *anyopaque, handler: ContextCompactionHandler, ) !void { var publication: ContextCompactionPublication = .{ .ctx = ctx, .handler = handler, .event = event }; try self.propagateHistoryTurnLocked( alloc, .{ .compacted_summary = event.summary }, event.max_history_turns, .{ .ctx = &publication, .commit_fn = ContextCompactionPublication.commit, .retained_from = event.retained_from }, ); if (event.active_prefix != null) { if (self.active_prompt_snapshot_ownership) |ownership| { _ = ownership.preserve(); } } }
fn propagateHistoryTurnLocked( self: *WorkerRuntime, alloc: std.mem.Allocator, turn: types.HistoryTurn, max_history_turns: usize, publication: ?HistoryPublication, ) !void { errdefer |err| if (err == error.SessionPersistenceUncertain) { if (self.active_prompt_snapshot_ownership) |ownership| { _ = ownership.preserve(); } }; if (self.queued_prompts.items.len == 0) { if (publication) |value| try value.commit(); return; } const active_ownership = if (self.active_prompt_snapshot_ownership) |ownership| try ownership.ensureSharedOwnership(alloc) else null; const cut = if (publication) |value| value.retained_from else null; const next_history = if (cut) |retained_from| try session_runtime.prepareCompactedHistory(alloc, self.queued_history, turn.compacted_summary, retained_from) else try appendHistoryTurnProjection( alloc, self.queued_history, turn, max_history_turns, ); var next_history_owned = true; defer if (next_history_owned) types.freeHistoryTurnSlice(alloc, next_history); const prepared = try alloc.alloc( PreparedHistoryPropagation, self.queued_prompts.items.len, ); var prepared_count: usize = 0; var committed = false; defer { if (!committed) { for (prepared[0..prepared_count]) |entry| entry.deinit(alloc); } alloc.free(prepared); }
while (prepared_count < self.queued_prompts.items.len) : (prepared_count += 1) { const prompt = self.queued_prompts.items[prepared_count]; const next_image_catalog = try session_runtime.merge_image_catalog_history_turn( alloc, prompt.authorized_image_catalog, turn, ); errdefer types.freeImageAttachmentSlice(alloc, next_image_catalog); const next_root_user_intent_context = try auto_classifier_context.refreshQueuedRootUserContext( alloc, prompt.prompt, prompt.root_user_intent_context, turn, ); errdefer alloc.free(next_root_user_intent_context); const next_snapshot_file_ownerships = if (active_ownership) |ownership| blk: { const ownership_count = std.math.add( usize, prompt.snapshot_file_ownerships.len, 1, ) catch return error.OutOfMemory; const ownerships = try alloc.alloc( types.SnapshotFileOwnership, ownership_count, ); std.mem.copyForwards( types.SnapshotFileOwnership, ownerships[0..prompt.snapshot_file_ownerships.len], prompt.snapshot_file_ownerships, ); ownership.retain(); ownerships[prompt.snapshot_file_ownerships.len] = ownership; break :blk ownerships; } else null; errdefer { if (next_snapshot_file_ownerships) |ownerships| { ownerships[ownerships.len - 1].release(); alloc.free(ownerships); } }
prepared[prepared_count] = .{ .root_user_intent_context = next_root_user_intent_context, .authorized_image_catalog = next_image_catalog, .snapshot_file_ownerships = next_snapshot_file_ownerships, .added_snapshot_ownership = if (next_snapshot_file_ownerships) |ownerships| ownerships[ownerships.len - 1] else null, }; }
if (publication) |value| try value.commit(); if (next_history.len <= self.queued_history.len) { debug_trace.logf("worker", "queued history replaced by context checkpoint turns={d}", .{self.queued_history.len}); } types.freeHistoryTurnSlice(alloc, self.queued_history); self.queued_history = next_history; next_history_owned = false; for (self.queued_prompts.items, prepared) |*prompt, next| { if (turn == .compacted_summary) prompt.unversioned_history_count = 0; if (prompt.root_user_intent_context.len > 0) alloc.free(prompt.root_user_intent_context); prompt.root_user_intent_context = next.root_user_intent_context; types.freeImageAttachmentSlice(alloc, prompt.authorized_image_catalog); prompt.authorized_image_catalog = next.authorized_image_catalog; if (next.snapshot_file_ownerships) |ownerships| { if (prompt.snapshot_file_ownerships.len > 0) { alloc.free(prompt.snapshot_file_ownerships); } prompt.snapshot_file_ownerships = ownerships; } } committed = true; }
pub fn beginActivePromptSnapshots( self: *WorkerRuntime, ownership: *ActivePromptSnapshotOwnership, ) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); std.debug.assert(self.active_prompt_snapshot_ownership == null); if (self.preserve_prompt_snapshot_turn_id == self.active_turn_id) { _ = ownership.preserve(); self.preserve_prompt_snapshot_turn_id = null; } self.active_prompt_snapshot_ownership = ownership; }
pub fn handoffActivePromptSnapshots( self: *WorkerRuntime, alloc: std.mem.Allocator, ) !?types.SnapshotFileOwnership { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const ownership = self.active_prompt_snapshot_ownership orelse return null; return ownership.handoff(alloc); }
pub fn endActivePromptSnapshots( self: *WorkerRuntime, ownership: *ActivePromptSnapshotOwnership, ) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); std.debug.assert(self.active_prompt_snapshot_ownership == ownership); self.active_prompt_snapshot_ownership = null; ownership.deinit(); }
pub fn preservePromptSnapshots(self: *WorkerRuntime, turn_id: u64, images: []const types.ImageAttachment) void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); _ = self.preservePromptSnapshotsLocked(turn_id, images); }
fn preservePromptSnapshotsLocked( self: *WorkerRuntime, turn_id: u64, images: []const types.ImageAttachment, ) bool { if (turn_id == 0 or images.len == 0) return false;
var preserved = false; if (self.active_turn_id == turn_id) { if (self.active_prompt_snapshot_ownership) |ownership| { if (sameImageSnapshots(ownership.images, images)) { preserved = ownership.preserve(); } } else { // The unique active turn bridges queue take to ownership // registration while both operations remain mutex-serialized. self.preserve_prompt_snapshot_turn_id = turn_id; preserved = true; } } for (self.worker_events.items) |*event| { if (event.* != .finish_prompt) continue; const finished = &event.finish_prompt; const finished_images = historyTurnImages(finished.turn); if (!sameImageSnapshots(finished_images, images)) continue; if (finished.snapshot_file_ownership) |ownership| { ownership.transfer(); preserved = true; } } return preserved; }
pub fn propagateGrant(self: *WorkerRuntime, alloc: std.mem.Allocator, tool_name: []const u8, target_path: []const u8) !void { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); for (self.queued_prompts.items) |*prompt| try appendGrantToQueuedPrompt(alloc, prompt, tool_name, target_path); }
pub fn requestPermissionBlocking( self: *WorkerRuntime, alloc: std.mem.Allocator, request: permission_request.PermissionRequest, ) permission_request.RequestCloneError!permission_request.OwnedPermissionResponse { return self.requestPermissionBlockingWithReview(alloc, request, null); }
pub fn requestPermissionBlockingWithReview( self: *WorkerRuntime, alloc: std.mem.Allocator, request: permission_request.PermissionRequest, review: ?*const diff_mod.FileReview, ) permission_request.RequestCloneError!permission_request.OwnedPermissionResponse { return self.requestPermissionBlockingObserved(alloc, request, review, null) catch |err| switch (err) { error.PermissionRegistrationFailed, error.PermissionCapacityExceeded, => unreachable, else => |request_err| return request_err, }; }
pub const PermissionRequestObserver = struct { context: *anyopaque, observe_fn: *const fn ( *anyopaque, *WorkerRuntime, permission_request.PermissionRequest, ) error{ OutOfMemory, PermissionRegistrationFailed, PermissionCapacityExceeded, }!void, };
pub fn requestPermissionBlockingObserved( self: *WorkerRuntime, alloc: std.mem.Allocator, request: permission_request.PermissionRequest, review: ?*const diff_mod.FileReview, observer: ?PermissionRequestObserver, ) (permission_request.RequestCloneError || error{ PermissionRegistrationFailed, PermissionCapacityExceeded, })!permission_request.OwnedPermissionResponse { var shared_request: ?permission_request.OwnedPermissionRequest = try permission_request.OwnedPermissionRequest.dupe( alloc, request, ); errdefer if (shared_request) |*owned| owned.deinit(alloc);
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
shared_request.?.id = try self.allocatePermissionRequestIdLocked(); if (self.pending_permission_request_shared) |*old| old.deinit(alloc); self.pending_permission_request_shared = shared_request.?; shared_request = null; self.pending_permission_review = review; if (self.pending_permission_response) |response| { self.discardPermissionResponse(response, "request_replaced"); } self.pending_permission_response = null; self.pending_permission_waiting = true;
if (observer) |value| { value.observe_fn( value.context, self, self.pending_permission_request_shared.?.view(), ) catch |err| { self.pending_permission_waiting = false; self.pending_permission_request_shared.?.deinit(alloc); self.pending_permission_request_shared = null; self.pending_permission_review = null; return err; }; }
while (self.pending_permission_response == null and !self.worker_stop_requested) { self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break; }
const response = self.pending_permission_response orelse permission_request.OwnedPermissionResponse.init(alloc, .deny, null); self.pending_permission_waiting = false; self.pending_permission_response = null; if (self.pending_permission_request_shared) |*old| { old.deinit(alloc); self.pending_permission_request_shared = null; } self.pending_permission_review = null; return response; }
pub fn submitPermissionResponse( self: *WorkerRuntime, expected_request_id: u64, response: permission_request.OwnedPermissionResponse, ) PermissionSubmissionResult { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
if (self.permissionSubmissionBlockedLocked(expected_request_id)) |blocked| { self.discardPermissionResponse(response, @tagName(blocked)); return blocked; } std.debug.assert(self.resolvePendingPermissionLocked(response)); return .accepted; }
pub const PermissionCommitError = error{ OutOfMemory, PermissionCommitFailed, PermissionCapacityExceeded, };
pub const PermissionCommit = struct { context: *anyopaque, commit_fn: *const fn (*anyopaque) PermissionCommitError!void, };
/// Keeps canonical pending-request identity stable across a durable host /// commit. The response becomes visible and the waiter wakes only after the /// callback succeeds. On callback failure the owned response is discarded. pub fn submitPermissionResponseAfterCommit( self: *WorkerRuntime, expected_request_id: u64, response: permission_request.OwnedPermissionResponse, commit: ?PermissionCommit, ) PermissionCommitError!PermissionSubmissionResult { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
if (self.permissionSubmissionBlockedLocked(expected_request_id)) |blocked| { self.discardPermissionResponse(response, @tagName(blocked)); return blocked; } if (commit) |effect| effect.commit_fn(effect.context) catch |err| { self.discardPermissionResponse(response, "commit_failed"); return err; }; std.debug.assert(self.resolvePendingPermissionLocked(response)); return .accepted; }
fn permissionSubmissionBlockedLocked( self: *const WorkerRuntime, expected_request_id: u64, ) ?PermissionSubmissionResult { if (!self.permissionRequestAwaitedLocked() or self.pending_permission_response != null) { return .no_pending; } if (self.pending_permission_request_shared.?.id != expected_request_id) { return .stale; } return null; }
fn allocatePermissionRequestIdLocked( self: *WorkerRuntime, ) error{RequestIdExhausted}!u64 { const request_id = self.next_permission_request_id; if (request_id == 0) return error.RequestIdExhausted; self.next_permission_request_id = std.math.add( u64, request_id, 1, ) catch 0; return request_id; }
fn permissionRequestAwaitedLocked(self: *const WorkerRuntime) bool { return self.worker_processing and !self.worker_stop_requested and self.pending_permission_waiting and self.pending_permission_request_shared != null; }
fn permissionRequestAwaitingDecisionLocked( self: *const WorkerRuntime, ) bool { return self.permissionRequestAwaitedLocked() and self.pending_permission_response == null; }
fn resolvePendingPermissionLocked( self: *WorkerRuntime, response: permission_request.OwnedPermissionResponse, ) bool { if (!self.permissionRequestAwaitingDecisionLocked()) { self.discardPermissionResponse(response, "unawaited"); return false; } self.pending_permission_response = response; self.worker_cond.broadcast(io_mod.getIo()); return true; }
fn discardPermissionResponse( _: *WorkerRuntime, response: permission_request.OwnedPermissionResponse, reason: []const u8, ) void { var owned = response; if (owned.feedback) |feedback| { debug_trace.logf( "permission", "approval feedback discarded reason={s} bytes={d}", .{ reason, feedback.len }, ); } owned.deinit(); }
pub fn requestQuestionBatchAnswerBlocking(self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry) !?[][]u8 { return self.requestQuestionBatchAnswerBlockingWithSource(alloc, entries, .agent_question); }
pub fn requestRouteRecoveryAnswerBlocking(self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry) !?[][]u8 { return self.requestQuestionBatchAnswerBlockingWithSource(alloc, entries, .route_recovery); }
pub fn requestMcpElicitationAnswerBlocking(self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry) !?[][]u8 { return self.requestQuestionBatchAnswerBlockingWithSource(alloc, entries, .mcp_elicitation); }
fn requestQuestionBatchAnswerBlockingWithSource( self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry, source: QuestionPromptSource, ) !?[][]u8 { const owned = try dupeOwnedQuestionBatch(alloc, entries, source); var owns_pending_question = true; errdefer if (owns_pending_question) freeOwnedQuestionBatch(alloc, owned);
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo());
std.debug.assert(self.pending_question_shared == null); self.pending_question_shared = owned; freeQuestionResponse(alloc, self.pending_question_response); self.pending_question_response = .pending; self.worker_events.append(alloc, .question_requested) catch |err| { self.pending_question_shared = null; self.pending_question_response = .pending; return err; }; owns_pending_question = false; self.worker_cond.broadcast(io_mod.getIo());
while (self.pending_question_response == .pending and !self.worker_stop_requested) { self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break; }
if (self.pending_question_shared) |pending| { freeOwnedQuestionBatch(alloc, pending); self.pending_question_shared = null; }
const response = self.pending_question_response; self.pending_question_response = .pending; return switch (response) { .answered => |labels| labels, .cancelled, .pending => null, }; }
pub fn submitQuestionBatchAnswer(self: *WorkerRuntime, alloc: std.mem.Allocator, answers: ?[]const []const u8) !void { var new_response: QuestionResponse = .cancelled; if (answers) |labels| { const dup = try alloc.alloc([]u8, labels.len); errdefer alloc.free(dup); var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) alloc.free(dup[i]); } while (filled < labels.len) : (filled += 1) dup[filled] = try alloc.dupe(u8, labels[filled]); new_response = .{ .answered = dup }; }
self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.pending_question_shared == null or self.pending_question_response != .pending) { freeQuestionResponse(alloc, new_response); debug_trace.logf("worker", "ignored late question response pending=false", .{}); return; } freeQuestionResponse(alloc, self.pending_question_response); self.pending_question_response = new_response; self.worker_cond.broadcast(io_mod.getIo()); }
pub fn cancelPendingQuestionBatch(self: *WorkerRuntime) bool { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); if (self.pending_question_shared == null or self.pending_question_response != .pending) return false; self.pending_question_response = .cancelled; self.worker_cond.broadcast(io_mod.getIo()); return true; }
pub fn snapshotPendingQuestionBatch(self: *WorkerRuntime, alloc: std.mem.Allocator) !?PendingQuestionBatchSnapshot { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const pending = self.pending_question_shared orelse return null; return try dupePendingBatchSnapshot(alloc, pending); }
pub fn pendingQuestionBatchSource(self: *WorkerRuntime) QuestionPromptSource { self.worker_mutex.lockUncancelable(io_mod.getIo()); defer self.worker_mutex.unlock(io_mod.getIo()); const pending = self.pending_question_shared orelse return .agent_question; return pending.source; }};
fn recoveryContinuationAdmission( processing: bool, queued_count: usize, terminal_pause_ready: bool,) bool { return queued_count == 0 and (!processing or terminal_pause_ready);}
test "recovery continuation admission requires an empty idle or terminally paused worker" { try std.testing.expect(recoveryContinuationAdmission(false, 0, false)); try std.testing.expect(!recoveryContinuationAdmission(true, 0, false)); try std.testing.expect(recoveryContinuationAdmission(true, 0, true)); try std.testing.expect(!recoveryContinuationAdmission(false, 1, true));}
test "terminal recovery pause admits continuation before worker cleanup" { const alloc = std.testing.allocator; var runtime: WorkerRuntime = .{}; defer runtime.deinit(alloc);
const Fixture = struct { fn make(allocator: std.mem.Allocator) !QueuedPrompt { var prompt = try makePrompt(allocator, "continue preserved turn", "model"); errdefer freeQueuedPrompt(allocator, prompt); prompt.recovery_checkpoint = try (session_codec.RecoveryCheckpoint{ .turn_id = 41, .user = .{ .text = @constCast("unfinished prompt") }, .assistant_source = @constCast("partial response"), .cause = .network_interrupted, .action = .paused, .authority = .{ .provider = .gateway, .model = @constCast("model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 10, .consumed_provider_attempts = 10, }).dupe(allocator); return prompt; } };
runtime.worker_processing = true; try runtime.pushEvent(alloc, .{ .route_recovery_status = .{ .kind = .terminal_provider_error, .failed_attempt = 10, .attempt_limit = 10, .cause = .network_interrupted, .action = .paused, .required_action = .continue_later, } }); try runtime.pushTurnFinished(alloc, .{ .turn_id = 41, .outcome = .paused, });
const continuation = try Fixture.make(alloc); runtime.enqueuePrompt(alloc, continuation) catch |err| { freeQueuedPrompt(alloc, continuation); return err; }; try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());}
test "connectivity wait state follows worker events before UI presentation" { var runtime: WorkerRuntime = .{}; defer runtime.deinit(std.testing.allocator);
try runtime.pushEvent(std.testing.allocator, .{ .route_recovery_status = .{ .kind = .auto_retry, .cause = .system_resumed, .action = .waiting_for_connectivity, } }); try std.testing.expect(runtime.isConnectivityWaitActive());
try runtime.pushEvent(std.testing.allocator, .{ .route_recovery_status = .{ .kind = .auto_retry, .cause = .network_interrupted, .action = .retrying_request, } }); try std.testing.expect(!runtime.isConnectivityWaitActive());
try runtime.pushEvent(std.testing.allocator, .{ .route_recovery_status = .{ .kind = .auto_retry, .cause = .system_resumed, .action = .waiting_for_connectivity, } }); try runtime.pushEvent(std.testing.allocator, .clear_route_recovery_status); try std.testing.expect(!runtime.isConnectivityWaitActive());}
fn dupeOwnedQuestionEntry(alloc: std.mem.Allocator, entry: types.QuestionBatchEntry) !OwnedQuestionEntry { const question_dup = try alloc.dupe(u8, entry.question); errdefer alloc.free(question_dup); const options_dup = try alloc.alloc(OwnedQuestionOption, entry.options.len); errdefer alloc.free(options_dup);
var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) { alloc.free(options_dup[i].label); if (options_dup[i].description) |d| alloc.free(d); } } while (filled < entry.options.len) : (filled += 1) { const src = entry.options[filled]; const label_dup = try alloc.dupe(u8, src.label); errdefer alloc.free(label_dup); const desc_dup: ?[]u8 = if (src.description) |d| try alloc.dupe(u8, d) else null; options_dup[filled] = .{ .label = label_dup, .description = desc_dup }; } return .{ .question = question_dup, .options = options_dup, .submission = entry.submission };}
fn freeOwnedQuestionEntry(alloc: std.mem.Allocator, entry: OwnedQuestionEntry) void { alloc.free(entry.question); for (entry.options) |opt| { alloc.free(opt.label); if (opt.description) |d| alloc.free(d); } alloc.free(entry.options);}
fn dupeOwnedQuestionBatch( alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry, source: QuestionPromptSource,) !OwnedQuestionBatch { const owned_entries = try alloc.alloc(OwnedQuestionEntry, entries.len); errdefer alloc.free(owned_entries); var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) freeOwnedQuestionEntry(alloc, owned_entries[i]); } while (filled < entries.len) : (filled += 1) owned_entries[filled] = try dupeOwnedQuestionEntry(alloc, entries[filled]); return .{ .entries = owned_entries, .source = source };}
fn freeOwnedQuestionBatch(alloc: std.mem.Allocator, owned: OwnedQuestionBatch) void { for (owned.entries) |entry| freeOwnedQuestionEntry(alloc, entry); alloc.free(owned.entries);}
fn freeOwnedQuestionBatchOpt(alloc: std.mem.Allocator, owned: ?OwnedQuestionBatch) void { if (owned) |o| freeOwnedQuestionBatch(alloc, o);}
fn freeQuestionResponse(alloc: std.mem.Allocator, response: QuestionResponse) void { switch (response) { .answered => |labels| { for (labels) |label| alloc.free(label); alloc.free(labels); }, .cancelled, .pending => {}, }}
fn dupePendingEntrySnapshot(alloc: std.mem.Allocator, pending: OwnedQuestionEntry) !types.QuestionBatchEntry { const question_dup = try alloc.dupe(u8, pending.question); errdefer alloc.free(question_dup); const options_dup = try alloc.alloc(types.QuestionOption, pending.options.len); errdefer alloc.free(options_dup);
var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) { alloc.free(@constCast(options_dup[i].label)); if (options_dup[i].description) |d| alloc.free(@constCast(d)); } } while (filled < pending.options.len) : (filled += 1) { const src = pending.options[filled]; const label_dup = try alloc.dupe(u8, src.label); errdefer alloc.free(label_dup); const desc_dup: ?[]const u8 = if (src.description) |d| try alloc.dupe(u8, d) else null; options_dup[filled] = .{ .label = label_dup, .description = desc_dup }; } return .{ .question = question_dup, .options = options_dup, .submission = pending.submission };}
fn dupePendingBatchSnapshot(alloc: std.mem.Allocator, pending: OwnedQuestionBatch) !PendingQuestionBatchSnapshot { const entries_dup = try alloc.alloc(types.QuestionBatchEntry, pending.entries.len); errdefer alloc.free(entries_dup); var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) { alloc.free(@constCast(entries_dup[i].question)); for (entries_dup[i].options) |opt| { alloc.free(@constCast(opt.label)); if (opt.description) |d| alloc.free(@constCast(d)); } alloc.free(@constCast(entries_dup[i].options)); } } while (filled < pending.entries.len) : (filled += 1) entries_dup[filled] = try dupePendingEntrySnapshot(alloc, pending.entries[filled]); return .{ .entries = entries_dup, .source = pending.source };}
fn appendGrantToQueuedPrompt(alloc: std.mem.Allocator, prompt: *QueuedPrompt, tool_name: []const u8, target_path: []const u8) !void { for (prompt.grants) |grant| { if (std.mem.eql(u8, grant.tool_name, tool_name) and std.mem.eql(u8, grant.target_path, target_path)) return; } const tool_name_dup = try alloc.dupe(u8, tool_name); errdefer alloc.free(tool_name_dup); const target_path_dup = try alloc.dupe(u8, target_path); errdefer alloc.free(target_path_dup); const current = prompt.grants; const next = try alloc.alloc(types.PermissionGrant, current.len + 1); errdefer alloc.free(next); if (current.len > 0) std.mem.copyForwards(types.PermissionGrant, next[0..current.len], current); next[current.len] = .{ .tool_name = tool_name_dup, .target_path = target_path_dup }; alloc.free(current); prompt.grants = next;}
fn appendHistoryTurnProjection( alloc: std.mem.Allocator, current: []const types.HistoryTurn, turn: types.HistoryTurn, max_history_turns: usize,) ![]types.HistoryTurn { _ = max_history_turns; const retained = if (turn == .compacted_summary and std.mem.startsWith(u8, turn.compacted_summary.summary, types.context_handoff_open)) &.{} else current; const next = try alloc.alloc(types.HistoryTurn, retained.len + 1); errdefer alloc.free(next); var copied: usize = 0; errdefer for (next[0..copied]) |owned| types.freeHistoryTurn(alloc, owned); for (retained, 0..) |entry, index| { next[index] = try types.dupeHistoryTurn(alloc, entry); copied += 1; } next[retained.len] = try types.dupeHistoryTurn(alloc, turn); return next;}
pub fn freeQueuedPrompt(alloc: std.mem.Allocator, prompt: QueuedPrompt) void { if (prompt.steering_receipt) |receipt| { if (receipt.state.cmpxchgStrong(.queued, .not_applied, .seq_cst, .seq_cst) == null) { debug_trace.eventf("subagent", "feedback_not_applied", .{}, "operation={s} reason=queue_discarded", .{receipt.operation_id}); } } alloc.free(prompt.prompt); types.freeImageAttachmentSlice(alloc, prompt.images); types.freeImageAttachmentSlice(alloc, prompt.authorized_image_catalog); alloc.free(prompt.model); secret.zeroAndFree(alloc, prompt.api_key); if (prompt.gateway_team) |team| alloc.free(team); if (prompt.account_id) |account_id| alloc.free(account_id); types.freeHistoryTurnSlice(alloc, prompt.history); if (prompt.root_user_intent_context.len > 0) alloc.free(prompt.root_user_intent_context); types.freePermissionGrantSlice(alloc, prompt.grants); freeSkillBindings(alloc, prompt.skill_bindings); freeSkillDisplaySpans(alloc, prompt.skill_display_spans); var context_snapshot = prompt.context_snapshot; context_snapshot.deinit(alloc); for (prompt.snapshot_file_ownerships) |ownership| ownership.release(); if (prompt.snapshot_file_ownerships.len > 0) { alloc.free(prompt.snapshot_file_ownerships); } if (prompt.recovery_checkpoint) |checkpoint| { var owned = checkpoint; owned.deinit(alloc); }}
pub fn freeContextCompactionTask( alloc: std.mem.Allocator, task: ContextCompactionTask,) void { alloc.free(task.model); secret.zeroAndFree(alloc, task.api_key); if (task.gateway_team) |team| alloc.free(team); if (task.account_id) |account_id| alloc.free(account_id); types.freeHistoryTurnSlice(alloc, task.history);}
pub fn freeWorkItem(alloc: std.mem.Allocator, work: WorkItem) void { switch (work) { .prompt => |prompt| freeQueuedPrompt(alloc, prompt), .compact_context => |task| freeContextCompactionTask(alloc, task), }}
fn discardQueuedPrompt( alloc: std.mem.Allocator, prompt: QueuedPrompt, retained_images: []const types.ImageAttachment,) void { if (prompt.recovery_checkpoint == null) { image_attachments.deleteUnreferencedImageSnapshots(prompt.images, retained_images); } freeQueuedPrompt(alloc, prompt);}
test "active prompt snapshot ownership discards every pre-transfer boundary" { const boundaries = [_][]const u8{ "preflight_notice", "skill_prompt", "tool_projection", "permission_preparation", "process_queued_prompt", "postflight_finish", }; for (boundaries, 0..) |_, index| { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var name_buffer: [32]u8 = undefined; const name = try std.fmt.bufPrint(&name_buffer, "snapshot-{d}.bin", .{index}); { var file = try tmp.dir.createFile(std.testing.io, name, .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(std.testing.allocator, tmp.dir, name); defer std.testing.allocator.free(path); const images = [_]types.ImageAttachment{.{ .id = index + 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; var ownership = ActivePromptSnapshotOwnership.init(&images); ownership.deinit(); ownership.deinit(); try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{}), ); }}
test "session and checkpoint transfer preserve active and finished prompt snapshots" { const alloc = std.testing.allocator; const phases = [_]enum { take_before_begin, active, finished }{ .take_before_begin, .active, .finished, }; for ([_]bool{ false, true }) |checkpoint_accepted| { for (phases) |phase| { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const name = switch (phase) { .take_before_begin => "take-before-begin.bin", .active => "active.bin", .finished => "finished.bin", }; { var file = try tmp.dir.createFile(std.testing.io, name, .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, name); defer alloc.free(path); const images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }};
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.active_turn_id = 41; var active = ActivePromptSnapshotOwnership.init(&images); if (phase != .take_before_begin) runtime.beginActivePromptSnapshots(&active); if (phase == .finished) { const ownership = (try runtime.handoffActivePromptSnapshots(alloc)) orelse return error.TestExpectedSnapshotOwnership; try runtime.pushEvent(alloc, .{ .finish_prompt = .{ .turn = .{ .assistant = .{ .user = .{ .text = @constCast("prompt"), .images = @constCast(&images) }, .assistant = @constCast("done"), } }, .snapshot_file_ownership = ownership, } }); ownership.release(); runtime.endActivePromptSnapshots(&active); runtime.active_turn_id = 0; }
if (checkpoint_accepted) { runtime.preservePromptSnapshots(41, &images); } else { runtime.clearQueuedPromptsForSessionTransition(alloc, 41, &images); } if (phase == .take_before_begin) runtime.beginActivePromptSnapshots(&active); if (phase != .finished) runtime.endActivePromptSnapshots(&active); runtime.discardEvents(alloc); try std.Io.Dir.accessAbsolute(std.testing.io, path, .{}); try std.Io.Dir.deleteFileAbsolute(std.testing.io, path); } }}
test "compaction activity survives worker finish without allocating a permission snapshot" { var runtime: WorkerRuntime = .{}; defer runtime.deinit(std.testing.allocator); try std.testing.expect(runtime.beginDirectProcessing(17)); const first = runtime.beginCompactionActivity(.automatic, 17); runtime.runCompactionActivity(first, .publication); runtime.requestInteractiveCancel(); try std.testing.expect(runtime.compactionActivitySnapshot().operation.?.phase == .stopping); runtime.settleCompactionActivity(first, .{ .outcome = .succeeded, .stage = .publication, .publication = .committed }); runtime.finishProcessing(); try std.testing.expectEqual(@as(u64, 0), runtime.activeTurnId()); const terminal = runtime.compactionActivitySnapshot(); try std.testing.expectEqual(first, terminal.operation.?.id); try std.testing.expectEqual(compaction_activity.Outcome.succeeded, terminal.operation.?.phase.terminal.outcome); runtime.settleCompactionActivity(first, .{ .outcome = .cancelled }); try std.testing.expectEqualDeep(terminal, runtime.compactionActivitySnapshot());
const next = runtime.beginCompactionActivity(.manual, null); const started = runtime.compactionActivitySnapshot(); runtime.rejectCompactionActivity(.busy); try std.testing.expectEqualDeep(started, runtime.compactionActivitySnapshot()); try std.testing.expect(!runtime.dismissCompactionActivity(first, terminal.revision)); runtime.settleCompactionActivity(first, compaction_activity.failure(error.Aborted, .publication, true)); try std.testing.expectEqual(next, runtime.compactionActivitySnapshot().operation.?.id); runtime.settleCompactionActivity(next, compaction_activity.failure(error.OutOfMemory, .preparation, false)); try std.testing.expectEqual(error.OutOfMemory, runtime.compactionActivitySnapshot().operation.?.phase.terminal.err.?);}
test "manual compaction is a typed worker item and not a prompt" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
const task = ContextCompactionTask{ .model = try alloc.dupe(u8, "provider/model"), .api_key = try alloc.dupe(u8, "key"), .history = try alloc.alloc(types.HistoryTurn, 0), }; try runtime.enqueueContextCompaction(task);
const taken = (try runtime.tryTakeNextWork(alloc)) orelse return error.TestExpectedQueuedWork; defer freeWorkItem(alloc, taken); try std.testing.expect(taken == .compact_context); const snapshot = runtime.compactionActivitySnapshot(); try std.testing.expectEqual(taken.compact_context.operation_id.?, snapshot.operation.?.id); try std.testing.expectEqual(taken.compact_context.turn_id, snapshot.operation.?.turn_id.?); try std.testing.expect(snapshot.operation.?.phase == .running); runtime.finishProcessing(); try std.testing.expectEqual(error.CompactionInterrupted, runtime.compactionActivitySnapshot().operation.?.phase.terminal.err.?);}
test "session transition queue clearing settles only the dropped compaction identity" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |retain_snapshots| { for ([_]bool{ false, true }) |newer_activity| { var runtime: WorkerRuntime = .{}; defer runtime.deinit(alloc); try runtime.enqueueContextCompaction(.{ .turn_id = 41, .model = try alloc.dupe(u8, "provider/model"), .api_key = try alloc.dupe(u8, "key"), .history = try alloc.alloc(types.HistoryTurn, 0), }); const queued = runtime.compactionActivitySnapshot(); if (newer_activity) _ = runtime.beginCompactionActivity(.manual, null); const observed = runtime.compactionActivitySnapshot();
if (retain_snapshots) { runtime.clearQueuedPromptsForSessionTransition(alloc, 41, &.{}); } else { runtime.clearQueuedPrompts(alloc, &.{}); } try std.testing.expectEqual(ContextCompactionStatus.idle, runtime.contextCompactionStatus()); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); try std.testing.expect((try runtime.tryTakeNextWork(alloc)) == null); try std.testing.expect(!runtime.isCancelRequested()); const settled = runtime.compactionActivitySnapshot(); if (newer_activity) { try std.testing.expectEqualDeep(observed, settled); } else { try std.testing.expectEqual(queued.operation.?.id, settled.operation.?.id); try std.testing.expectEqual(compaction_activity.Outcome.cancelled, settled.operation.?.phase.terminal.outcome); try std.testing.expect(!settled.operation.?.active()); try std.testing.expect(settled.revision > queued.revision); } runtime.clearQueuedPrompts(alloc, &.{}); try std.testing.expectEqualDeep(settled, runtime.compactionActivitySnapshot()); } }}
test "queued manual compaction can be cancelled before worker execution" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
const task = ContextCompactionTask{ .model = try alloc.dupe(u8, "provider/model"), .api_key = try alloc.dupe(u8, "key"), .history = try alloc.alloc(types.HistoryTurn, 0), }; try runtime.enqueueContextCompaction(task);
try std.testing.expectEqual(ContextCompactionStatus.queued, runtime.contextCompactionStatus()); try std.testing.expect(runtime.cancelContextCompaction(alloc)); try std.testing.expectEqual(ContextCompactionStatus.idle, runtime.contextCompactionStatus()); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); try std.testing.expect((try runtime.tryTakeNextWork(alloc)) == null); try std.testing.expectEqual(compaction_activity.Outcome.cancelled, runtime.compactionActivitySnapshot().operation.?.phase.terminal.outcome);}
test "running manual compaction cancellation uses the worker cancel flag" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
const task = ContextCompactionTask{ .model = try alloc.dupe(u8, "provider/model"), .api_key = try alloc.dupe(u8, "key"), .history = try alloc.alloc(types.HistoryTurn, 0), }; try runtime.enqueueContextCompaction(task); const taken = (try runtime.tryTakeNextWork(alloc)) orelse return error.TestExpectedQueuedWork; defer freeWorkItem(alloc, taken);
try std.testing.expectEqual(ContextCompactionStatus.running, runtime.contextCompactionStatus()); try std.testing.expect(runtime.cancelContextCompaction(alloc)); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expect(runtime.compactionActivitySnapshot().operation.?.phase == .stopping); runtime.finishProcessing(); try std.testing.expectEqual(@as(u64, 0), runtime.activeTurnId()); try std.testing.expectEqual(compaction_activity.Outcome.cancelled, runtime.compactionActivitySnapshot().operation.?.phase.terminal.outcome); try std.testing.expectEqual(ContextCompactionStatus.idle, runtime.contextCompactionStatus());}
test "session transfer discards mismatched active prompt snapshots" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "active.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "active.bin"); defer alloc.free(path); const active_images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; const unrelated_images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/other.png"), .media_type = @constCast("image/png"), .snapshot_path = @constCast("/tmp/other.bin"), .snapshot_sha256 = @constCast("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"), }};
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.active_turn_id = 42; var active = ActivePromptSnapshotOwnership.init(&active_images); runtime.beginActivePromptSnapshots(&active);
runtime.clearQueuedPromptsForSessionTransition(alloc, 42, &unrelated_images); runtime.endActivePromptSnapshots(&active);
try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{}), );}
test "accepted active prompt ownership does not delete history files" { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var first = try tmp.dir.createFile(std.testing.io, "first.bin", .{}); defer first.close(std.testing.io); try first.writeStreamingAll(std.testing.io, "first"); } { var second = try tmp.dir.createFile(std.testing.io, "second.bin", .{}); defer second.close(std.testing.io); try second.writeStreamingAll(std.testing.io, "second"); } const first_path = try io_mod.dirRealpathAlloc(std.testing.allocator, tmp.dir, "first.bin"); defer std.testing.allocator.free(first_path); const second_path = try io_mod.dirRealpathAlloc(std.testing.allocator, tmp.dir, "second.bin"); defer std.testing.allocator.free(second_path); const images = [_]types.ImageAttachment{ .{ .id = 1, .path = @constCast("/tmp/first.png"), .media_type = @constCast("image/png"), .snapshot_path = first_path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }, .{ .id = 2, .path = @constCast("/tmp/second.png"), .media_type = @constCast("image/png"), .snapshot_path = second_path, .snapshot_sha256 = @constCast("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"), }, }; var ownership = ActivePromptSnapshotOwnership.init(&images); const accepted = (try ownership.handoff(std.testing.allocator)) orelse return error.TestExpectedEqual; accepted.transfer(); accepted.release(); ownership.deinit();
try std.Io.Dir.accessAbsolute(std.testing.io, first_path, .{}); try std.Io.Dir.accessAbsolute(std.testing.io, second_path, .{});}
test "propagated queued snapshots survive finish failure until the last borrower releases" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin"); defer alloc.free(path); var images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = @constCast("active"), .images = &images }, .assistant = @constCast("done"), } };
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "first queued", "model")); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "second queued", "model"));
var active = ActivePromptSnapshotOwnership.init(&images); runtime.beginActivePromptSnapshots(&active); try runtime.propagateHistoryTurn(alloc, turn, 8); const failed_finish_ownership = (try runtime.handoffActivePromptSnapshots(alloc)) orelse return error.TestExpectedEqual; failed_finish_ownership.release(); runtime.endActivePromptSnapshots(&active);
for (runtime.queued_prompts.items) |prompt| { try std.testing.expectEqual(@as(usize, 1), prompt.authorized_image_catalog.len); try std.testing.expectEqual(@as(usize, 1), prompt.snapshot_file_ownerships.len); try std.Io.Dir.accessAbsolute( std.testing.io, prompt.authorized_image_catalog[0].snapshot_path.?, .{}, ); }
const first = (try runtime.tryTakeNextPrompt(alloc)).?; freeQueuedPrompt(alloc, first); try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
runtime.clearQueuedPrompts(alloc, &.{}); try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{}), );}
test "accepted history prevents propagated queue cleanup from deleting snapshots" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin"); defer alloc.free(path); var images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = @constCast("active"), .images = &images }, .assistant = @constCast("done"), } };
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));
var active = ActivePromptSnapshotOwnership.init(&images); runtime.beginActivePromptSnapshots(&active); try runtime.propagateHistoryTurn(alloc, turn, 8); const accepted = (try runtime.handoffActivePromptSnapshots(alloc)) orelse return error.TestExpectedEqual; accepted.transfer(); accepted.release(); runtime.endActivePromptSnapshots(&active); runtime.clearQueuedPrompts(alloc, &.{});
try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});}
test "compaction preserves active snapshots only after persistence succeeds" { const Persist = struct { fail: bool,
fn commit(raw: *anyopaque, _: ContextCompaction) !void { const self: *@This() = @ptrCast(@alignCast(raw)); if (self.fail) return error.TestPersistenceFailure; } }; const alloc = std.testing.allocator; for ([_]bool{ false, true }) |queued| { for ([_]bool{ false, true }) |fail| { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin"); defer alloc.free(path); var images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; var runtime: WorkerRuntime = .{}; defer runtime.deinit(alloc); if (queued) try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model")); var ownership = ActivePromptSnapshotOwnership.init(&images); runtime.beginActivePromptSnapshots(&ownership); var active = true; defer if (active) runtime.endActivePromptSnapshots(&ownership); var persist: Persist = .{ .fail = fail }; const event: ContextCompaction = .{ .summary = .{ .summary = @constCast("<context_handoff>image checkpoint</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, }, .active_prefix = .{ .user = .{ .text = @constCast("inspect image"), .images = &images }, .assistant = @constCast(""), }, }; if (fail) { try std.testing.expectError(error.TestPersistenceFailure, runtime.publishContextCompaction(alloc, event, &persist, Persist.commit)); } else { try runtime.publishContextCompaction(alloc, event, &persist, Persist.commit); try runtime.publishContextCompaction(alloc, event, &persist, Persist.commit); } runtime.endActivePromptSnapshots(&ownership); active = false; runtime.clearQueuedPrompts(alloc, &.{}); if (fail) { try std.testing.expectError(error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{})); } else { try std.Io.Dir.accessAbsolute(std.testing.io, path, .{}); try std.testing.expect(ownership.shared_ownership == null); } } }}
test "failed queued prompt dequeue retains borrowed snapshot ownership" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin"); defer alloc.free(path); var images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = @constCast("active"), .images = &images }, .assistant = @constCast("done"), } };
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));
var active = ActivePromptSnapshotOwnership.init(&images); runtime.beginActivePromptSnapshots(&active); try runtime.propagateHistoryTurn(alloc, turn, 8); const failed_finish_ownership = (try runtime.handoffActivePromptSnapshots(alloc)) orelse return error.TestExpectedEqual; failed_finish_ownership.release(); runtime.endActivePromptSnapshots(&active);
var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 }); try std.testing.expectError( error.OutOfMemory, runtime.waitAndTakeNextPrompt(failing.allocator()), ); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); try std.testing.expectEqual( @as(usize, 1), runtime.queued_prompts.items[0].snapshot_file_ownerships.len, ); try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
runtime.clearQueuedPrompts(alloc, &.{}); try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{}), );}
fn checkHistoryPropagationAllocation( alloc: std.mem.Allocator, snapshot_path: []const u8, fail_index: usize,) !bool { { var file = try std.Io.Dir.createFileAbsolute( std.testing.io, snapshot_path, .{ .truncate = true }, ); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } var images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = @constCast(snapshot_path), .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = @constCast("active"), .images = &images }, .assistant = @constCast("done"), } };
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); const first = try makePrompt(alloc, "first queued", "model"); var owns_first = true; errdefer if (owns_first) freeQueuedPrompt(alloc, first); try runtime.enqueuePrompt(alloc, first); owns_first = false; const second = try makePrompt(alloc, "second queued", "model"); var owns_second = true; errdefer if (owns_second) freeQueuedPrompt(alloc, second); try runtime.enqueuePrompt(alloc, second); owns_second = false;
var active = ActivePromptSnapshotOwnership.init(&images); runtime.beginActivePromptSnapshots(&active); defer runtime.endActivePromptSnapshots(&active);
var failing = std.testing.FailingAllocator.init( alloc, .{ .fail_index = fail_index }, ); runtime.propagateHistoryTurn(failing.allocator(), turn, 8) catch |err| { if (!failing.has_induced_failure) return err; for (runtime.queued_prompts.items) |prompt| { try std.testing.expectEqual(@as(usize, 0), prompt.history.len); try std.testing.expectEqual(@as(usize, 0), prompt.authorized_image_catalog.len); try std.testing.expectEqual(@as(usize, 0), prompt.snapshot_file_ownerships.len); } try std.testing.expectEqual(@as(usize, 0), runtime.queued_history.len); return false; };
for (runtime.queued_prompts.items) |prompt| { try std.testing.expectEqual(@as(usize, 0), prompt.history.len); try std.testing.expectEqual(@as(usize, 1), prompt.authorized_image_catalog.len); try std.testing.expectEqual(@as(usize, 1), prompt.snapshot_file_ownerships.len); } try std.testing.expectEqual(@as(usize, 1), runtime.queued_history.len); return true;}
test "multi-queue history propagation is allocation-failure atomic" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const root = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(root); const snapshot_path = try std.fs.path.join(alloc, &.{ root, "snapshot.bin" }); defer alloc.free(snapshot_path);
var fail_index: usize = 0; while (fail_index < 256) : (fail_index += 1) { if (try checkHistoryPropagationAllocation(alloc, snapshot_path, fail_index)) { return; } } return error.TestUnexpectedResult;}
fn checkContextCompactionCommitAllocation( alloc: std.mem.Allocator, fail_index: usize,) !bool { var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); var prompt = try makePrompt(alloc, "queued", "model"); var owns_prompt = true; errdefer if (owns_prompt) freeQueuedPrompt(alloc, prompt); prompt.history = try session_runtime.snapshotOwnedContextHistory(alloc, &.{.{ .assistant = .{ .user = .{ .text = @constCast("prior user") }, .assistant = @constCast("prior answer"), } }}, 0, 0); try runtime.enqueuePrompt(alloc, prompt); owns_prompt = false;
const turn: types.HistoryTurn = .{ .compacted_summary = .{ .summary = @constCast("<context_handoff>checkpoint</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, } }; var failing = std.testing.FailingAllocator.init( alloc, .{ .fail_index = fail_index }, ); runtime.publishContextCompaction(failing.allocator(), .{ .summary = turn.compacted_summary, .max_history_turns = 8 }, &runtime, discardCompactionForTest) catch |err| { if (!failing.has_induced_failure) return err; try std.testing.expectEqual( @as(usize, 1), runtime.queued_history.len, ); try std.testing.expectEqualStrings("prior answer", runtime.queued_history[0].assistant.assistant); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len); return false; };
try std.testing.expectEqual( @as(usize, 1), runtime.queued_history.len, ); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len); try std.testing.expectEqualStrings(turn.compacted_summary.summary, runtime.queued_history[0].compacted_summary.summary); return true;}
fn discardCompactionForTest(_: *anyopaque, _: ContextCompaction) !void {}
test "context compaction callback waits for UI persistence" { const Probe = struct { returned: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), failure: ?anyerror = null,
fn run(self: *@This(), runtime: *WorkerRuntime) void { runtime.commitContextCompaction(std.testing.allocator, .{ .summary = @constCast("<context_handoff>checkpoint</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, }, null, null, 8) catch |err| { self.failure = err; }; self.returned.store(true, .release); } }; var runtime: WorkerRuntime = .{}; defer runtime.deinit(std.testing.allocator); var probe: Probe = .{}; const thread = try std.Thread.spawn(.{}, Probe.run, .{ &probe, &runtime }); defer { runtime.requestShutdown(); thread.join(); } const deadline = io_mod.milliTimestamp() + 5_000; while (true) { var state = try runtime.snapshotState(std.testing.allocator); defer state.deinit(std.testing.allocator); if (state.pending_event_count > 0) break; if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionEvent; io_mod.sleep(std.time.ns_per_ms); } io_mod.sleep(10 * std.time.ns_per_ms); try std.testing.expect(!probe.returned.load(.acquire));}
test "context compaction acknowledgment commits or fails before worker continuation" { const Probe = struct { const Outcome = enum { commit, fail, cancel_after_commit }; runtime: *WorkerRuntime, outcome: Outcome, returned: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), continued: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), failure: ?anyerror = null, persisted: bool = false,
fn run(self: *@This()) void { self.runtime.commitContextCompaction(std.testing.allocator, .{ .summary = @constCast("<context_handoff>checkpoint</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, }, .{ .user = .{ .text = @constCast("active request") }, .assistant = @constCast(""), }, null, 8) catch |err| { self.failure = err; self.returned.store(true, .release); return; }; self.continued.store(true, .release); self.returned.store(true, .release); }
fn commit(raw: *anyopaque, event: ContextCompaction) !void { const self: *@This() = @ptrCast(@alignCast(raw)); try std.testing.expectEqualStrings("active request", event.active_prefix.?.user.text); try std.testing.expectEqualStrings("old answer", self.runtime.queued_history[0].assistant.assistant); if (self.outcome == .fail) return error.TestPersistenceFailure; self.persisted = true; if (self.outcome == .cancel_after_commit) { self.runtime.worker_cancel_requested.store(true, .seq_cst); } } }; const alloc = std.testing.allocator; for ([_]Probe.Outcome{ .commit, .fail, .cancel_after_commit }) |outcome| { var runtime: WorkerRuntime = .{}; defer runtime.deinit(alloc); var prompt = try makePrompt(alloc, "queued", "model"); prompt.history = try session_runtime.snapshotOwnedContextHistory(alloc, &.{.{ .assistant = .{ .user = .{ .text = @constCast("old request") }, .assistant = @constCast("old answer"), } }}, 0, 0); try runtime.enqueuePrompt(alloc, prompt); var probe: Probe = .{ .runtime = &runtime, .outcome = outcome }; const thread = try std.Thread.spawn(.{}, Probe.run, .{&probe}); var joined = false; defer if (!joined) { runtime.requestShutdown(); thread.join(); }; const deadline = io_mod.milliTimestamp() + 5_000; while (true) { var state = try runtime.snapshotState(alloc); defer state.deinit(alloc); if (state.pending_event_count > 0) break; if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionEvent; io_mod.sleep(std.time.ns_per_ms); } var events = runtime.takeEvents(); defer events.deinit(alloc); defer for (events.items) |event| freeWorkerEvent(alloc, event); try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(!probe.returned.load(.acquire)); try std.testing.expect(!probe.continued.load(.acquire)); runtime.resolveContextCompaction(alloc, events.items[0].context_compaction, &probe, Probe.commit); thread.join(); joined = true; if (outcome == .fail) { try std.testing.expectEqual(error.TestPersistenceFailure, probe.failure.?); try std.testing.expect(!probe.persisted); try std.testing.expect(!probe.continued.load(.acquire)); try std.testing.expectEqualStrings("old answer", runtime.queued_history[0].assistant.assistant); } else { try std.testing.expect(probe.failure == null); try std.testing.expect(probe.persisted); try std.testing.expect(probe.continued.load(.acquire)); try std.testing.expect(runtime.queued_history[0] == .compacted_summary); } }}
test "context compaction waiter exits on cancellation and rejects stale delivery" { const Probe = struct { returned: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), failure: ?anyerror = null, commits: usize = 0,
fn run(self: *@This(), runtime: *WorkerRuntime) void { runtime.commitContextCompaction(std.testing.allocator, .{ .summary = @constCast("<context_handoff>checkpoint</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, }, null, null, 8) catch |err| { self.failure = err; }; self.returned.store(true, .release); }
fn commit(raw: *anyopaque, _: ContextCompaction) !void { const self: *@This() = @ptrCast(@alignCast(raw)); self.commits += 1; } }; const alloc = std.testing.allocator; for ([_]bool{ false, true }) |shutdown| { var runtime: WorkerRuntime = .{}; defer runtime.deinit(alloc); var probe: Probe = .{}; const thread = try std.Thread.spawn(.{}, Probe.run, .{ &probe, &runtime }); var joined = false; defer if (!joined) { runtime.requestShutdown(); thread.join(); }; const deadline = io_mod.milliTimestamp() + 5_000; while (true) { var state = try runtime.snapshotState(alloc); defer state.deinit(alloc); if (state.pending_event_count > 0) break; if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionEvent; io_mod.sleep(std.time.ns_per_ms); } var events = runtime.takeEvents(); defer events.deinit(alloc); defer for (events.items) |event| freeWorkerEvent(alloc, event); if (shutdown) runtime.requestShutdown() else runtime.requestInteractiveCancel(); while (!probe.returned.load(.acquire)) { if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionCancellation; io_mod.sleep(std.time.ns_per_ms); } thread.join(); joined = true; try std.testing.expectEqual(error.Aborted, probe.failure.?); runtime.resolveContextCompaction(alloc, events.items[0].context_compaction, &probe, Probe.commit); try std.testing.expectEqual(@as(usize, 0), probe.commits); }}
test "context compaction commit is allocation-failure atomic" { var fail_index: usize = 0; while (fail_index < 128) : (fail_index += 1) { if (try checkContextCompactionCommitAllocation( std.testing.allocator, fail_index, )) return; } return error.TestUnexpectedResult;}
test "queued prompts begin with the latest checkpoint and its suffix" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var first = try makePrompt(alloc, "first queued", "model"); first.history = try session_runtime.snapshotOwnedContextHistory(alloc, &.{.{ .assistant = .{ .user = .{ .text = @constCast("old request") }, .assistant = @constCast("old answer") }, }}, 0, 0); try runtime.enqueuePrompt(alloc, first); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "second queued", "model"));
const checkpoint: types.HistoryTurn = .{ .compacted_summary = .{ .summary = @constCast("<context_handoff>current summary</context_handoff>"), .removed_turn_count = 1, .compaction_count = 1, } }; try runtime.publishContextCompaction(alloc, .{ .summary = checkpoint.compacted_summary }, &runtime, discardCompactionForTest); try std.testing.expectEqual(@as(usize, 1), runtime.queued_history.len); try runtime.propagateHistoryTurn(alloc, .{ .assistant = .{ .user = .{ .text = @constCast("later request") }, .assistant = @constCast("later answer"), } }, 0);
for (0..2) |_| { const job = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job); try std.testing.expectEqual(@as(usize, 2), job.history.len); try std.testing.expectEqualStrings(checkpoint.compacted_summary.summary, job.history[0].compacted_summary.summary); try std.testing.expectEqualStrings("later answer", job.history[1].assistant.assistant); } try std.testing.expectEqual(@as(usize, 0), runtime.queued_history.len);}
test "finish ownership handoff deletes snapshots after its last unaccepted release" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin"); defer alloc.free(path); const images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; var active = ActivePromptSnapshotOwnership.init(&images); const ownership = (try active.handoff(alloc)) orelse return error.TestExpectedEqual; active.deinit();
const queued = try types.dupeFinishedPrompt(alloc, .{ .turn = .{ .compacted_summary = .{ .summary = @constCast("queued"), .removed_turn_count = 1, .compaction_count = 1, } }, .snapshot_file_ownership = ownership, }); ownership.release(); const paced = try types.dupeFinishedPrompt(alloc, queued); types.freeFinishedPrompt(alloc, queued); try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
types.freeFinishedPrompt(alloc, paced); try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{}), );}
test "accepted finish ownership survives queue and pacing release" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); { var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{}); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin"); defer alloc.free(path); const images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; var active = ActivePromptSnapshotOwnership.init(&images); const ownership = (try active.handoff(alloc)) orelse return error.TestExpectedEqual; active.deinit();
const finished = try types.dupeFinishedPrompt(alloc, .{ .turn = .{ .compacted_summary = .{ .summary = @constCast("accepted"), .removed_turn_count = 1, .compaction_count = 1, } }, .snapshot_file_ownership = ownership, }); ownership.release(); finished.snapshot_file_ownership.?.transfer(); types.freeFinishedPrompt(alloc, finished);
try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});}
fn checkFinishOwnershipHandoffAllocation( alloc: std.mem.Allocator, snapshot_path: []const u8,) !void { { var file = try std.Io.Dir.createFileAbsolute( std.testing.io, snapshot_path, .{ .truncate = true }, ); defer file.close(std.testing.io); try file.writeStreamingAll(std.testing.io, "snapshot"); } const images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/tmp/source.png"), .media_type = @constCast("image/png"), .snapshot_path = @constCast(snapshot_path), .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}; var active = ActivePromptSnapshotOwnership.init(&images); const ownership = active.handoff(alloc) catch |err| { active.deinit(); try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, snapshot_path, .{}), ); return err; }; active.deinit(); ownership.?.release(); try std.testing.expectError( error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, snapshot_path, .{}), );}
test "finish ownership handoff preserves allocator and filesystem ownership" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const root = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(root); const snapshot_path = try std.fs.path.join(alloc, &.{ root, "snapshot.bin" }); defer alloc.free(snapshot_path);
try std.testing.checkAllAllocationFailures( alloc, checkFinishOwnershipHandoffAllocation, .{snapshot_path}, );}
pub fn dupeSkillBindings(alloc: std.mem.Allocator, bindings: []const SkillBinding) ![]SkillBinding { if (bindings.len == 0) return &.{}; const copy = try alloc.alloc(SkillBinding, bindings.len); var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) { alloc.free(copy[i].name); alloc.free(copy[i].path); } alloc.free(copy); } while (filled < bindings.len) : (filled += 1) { copy[filled] = .{ .name = try alloc.dupe(u8, bindings[filled].name), .path = try alloc.dupe(u8, bindings[filled].path), }; } return copy;}
pub fn freeSkillBindings(alloc: std.mem.Allocator, bindings: []SkillBinding) void { for (bindings) |binding| { alloc.free(binding.name); alloc.free(binding.path); } if (bindings.len > 0) alloc.free(bindings);}
pub fn dupeSkillDisplaySpans(alloc: std.mem.Allocator, spans: []const SkillDisplaySpan) ![]SkillDisplaySpan { if (spans.len == 0) return &.{}; const copy = try alloc.alloc(SkillDisplaySpan, spans.len); var filled: usize = 0; errdefer { var i: usize = 0; while (i < filled) : (i += 1) { alloc.free(copy[i].name); alloc.free(copy[i].path); } alloc.free(copy); } while (filled < spans.len) : (filled += 1) { copy[filled] = .{ .raw_start = spans[filled].raw_start, .raw_end = spans[filled].raw_end, .name = try alloc.dupe(u8, spans[filled].name), .path = try alloc.dupe(u8, spans[filled].path), .display_source = spans[filled].display_source, .owns_trailing_separator = spans[filled].owns_trailing_separator, }; } return copy;}
pub fn freeSkillDisplaySpans(alloc: std.mem.Allocator, spans: []SkillDisplaySpan) void { for (spans) |span| { alloc.free(span.name); alloc.free(span.path); } if (spans.len > 0) alloc.free(spans);}
pub fn dupeWorkerEvent(alloc: std.mem.Allocator, event: WorkerEvent) !WorkerEvent { return switch (event) { .begin_prompt => |prompt| .{ .begin_prompt = try types.dupeUserTurn(alloc, prompt) }, .begin_prompt_with_skill_bindings => |begin| blk: { const prompt = try types.dupeUserTurn(alloc, begin.prompt); errdefer types.freeUserTurn(alloc, prompt); const skill_bindings = try dupeSkillBindings(alloc, begin.skill_bindings); errdefer freeSkillBindings(alloc, skill_bindings); const skill_display_spans = try dupeSkillDisplaySpans(alloc, begin.skill_display_spans); break :blk .{ .begin_prompt_with_skill_bindings = .{ .prompt = prompt, .skill_bindings = skill_bindings, .skill_display_spans = skill_display_spans, } }; }, .begin_presented_prompt => |turn_id| .{ .begin_presented_prompt = turn_id }, .append_user_feedback => |text| .{ .append_user_feedback = try alloc.dupe(u8, text) }, .assistant_presentation => |presentation| .{ .assistant_presentation = try presentation.clone(alloc), }, .notification => |notification| .{ .notification = notification }, .question_requested => .question_requested, .open_model_picker => .open_model_picker, .semantic_notice => |notice| .{ .semantic_notice = try types.dupeSemanticNotice(alloc, notice) }, .full_detail_record => |notice| .{ .full_detail_record = try types.dupeSemanticNotice(alloc, notice) }, .route_recovery_status => |status| .{ .route_recovery_status = status }, .clear_route_recovery_status => .clear_route_recovery_status, .api_status_text => |text| .{ .api_status_text = try alloc.dupe(u8, text) }, .credential_refreshed => |credential| .{ .credential_refreshed = try credential.clone(alloc), }, .command_output => |chunk| .{ .command_output = .{ .lifecycle_id = if (chunk.lifecycle_id) |id| .{ .turn_id = id.turn_id, .call_id = try alloc.dupe(u8, id.call_id), } else null, .stream = chunk.stream, .text = try alloc.dupe(u8, chunk.text), } }, .command_output_complete => |lifecycle_id| .{ .command_output_complete = if (lifecycle_id) |id| .{ .turn_id = id.turn_id, .call_id = try alloc.dupe(u8, id.call_id), } else null }, .tool_lifecycle => |lifecycle| .{ .tool_lifecycle = try dupeToolLifecycleEvent(alloc, lifecycle), }, .turn_token_update => |update| .{ .turn_token_update = update }, .turn_phase_update => |update| .{ .turn_phase_update = update }, .diff_block => |payload| blk: { const preview = try alloc.dupe(u8, payload.preview); errdefer alloc.free(preview); const full = if (payload.full) |source| full_value: { const content = try alloc.dupe(u8, source.content); errdefer alloc.free(content); const call_id = try alloc.dupe(u8, source.lifecycle_id.call_id); break :full_value diff_mod.FullDiff{ .content = content, .lifecycle_id = .{ .turn_id = source.lifecycle_id.turn_id, .call_id = call_id, }, }; } else null; errdefer if (full) |owned| owned.deinit(alloc); break :blk .{ .diff_block = .{ .preview = preview, .additions = payload.additions, .deletions = payload.deletions, .full = full, } }; }, .prepare_fresh_prompt => |value| blk: { const user = try types.dupeUserTurn(alloc, value.user); errdefer types.freeUserTurn(alloc, user); break :blk .{ .prepare_fresh_prompt = .{ .turn_id = value.turn_id, .user = user, .prior_turn = if (value.prior_turn) |prior| try types.dupeHistoryTurn(alloc, prior) else null, } }; }, .context_compaction => |value| blk: { const summary = try types.dupeHistoryTurn(alloc, .{ .compacted_summary = value.summary }); errdefer types.freeHistoryTurn(alloc, summary); const prefix = if (value.active_prefix) |prefix| (try types.dupeHistoryTurn(alloc, .{ .assistant = prefix })).assistant else null; break :blk .{ .context_compaction = .{ .turn_id = value.turn_id, .summary = summary.compacted_summary, .active_prefix = prefix, .retained_from = value.retained_from, .max_history_turns = value.max_history_turns, } }; }, .finish_prompt => |finished| .{ .finish_prompt = try types.dupeFinishedPrompt(alloc, finished) }, .session_grant => |grant| blk: { const tool_name = try alloc.dupe(u8, grant.tool_name); errdefer alloc.free(tool_name); const target_path = try alloc.dupe(u8, grant.target_path); break :blk .{ .session_grant = .{ .tool_name = tool_name, .target_path = target_path, } }; }, .error_text => |notice| .{ .error_text = try types.dupeSemanticNotice(alloc, notice) }, .restore_failed_prompt => |text| .{ .restore_failed_prompt = try alloc.dupe(u8, text) }, };}
pub fn freeWorkerEvent(alloc: std.mem.Allocator, event: WorkerEvent) void { switch (event) { .begin_prompt => |prompt| types.freeUserTurn(alloc, prompt), .begin_prompt_with_skill_bindings => |begin| { types.freeUserTurn(alloc, begin.prompt); freeSkillBindings(alloc, begin.skill_bindings); freeSkillDisplaySpans(alloc, begin.skill_display_spans); }, .begin_presented_prompt => {}, .append_user_feedback => |text| alloc.free(text), .assistant_presentation => |presentation| { var owned = presentation; owned.deinit(alloc); }, .notification => {}, .open_model_picker => {}, .semantic_notice => |notice| types.freeSemanticNotice(alloc, notice), .full_detail_record => |notice| types.freeSemanticNotice(alloc, notice), .route_recovery_status => {}, .clear_route_recovery_status => {}, .api_status_text => |text| alloc.free(text), .credential_refreshed => |credential| { var owned = credential; owned.deinit(alloc); }, .command_output => |chunk| { if (chunk.lifecycle_id) |id| alloc.free(@constCast(id.call_id)); alloc.free(chunk.text); }, .command_output_complete => |lifecycle_id| { if (lifecycle_id) |id| alloc.free(@constCast(id.call_id)); }, .tool_lifecycle => |lifecycle| freeToolLifecycleEvent(alloc, lifecycle), .diff_block => |payload| diff_mod.freeDiffEntryPayload(alloc, payload), .prepare_fresh_prompt => |value| { types.freeUserTurn(alloc, value.user); if (value.prior_turn) |prior| types.freeHistoryTurn(alloc, prior); }, .context_compaction => |value| { types.freeHistoryTurn(alloc, .{ .compacted_summary = value.summary }); if (value.active_prefix) |prefix| types.freeHistoryTurn(alloc, .{ .assistant = prefix }); }, .finish_prompt => |finished| types.freeFinishedPrompt(alloc, finished), .session_grant => |grant| { alloc.free(grant.tool_name); alloc.free(grant.target_path); }, .error_text => |notice| types.freeSemanticNotice(alloc, notice), .restore_failed_prompt => |text| alloc.free(text), else => {}, }}
pub fn dupeToolLifecycleEvent( alloc: std.mem.Allocator, event: types.ToolLifecycleEvent,) !types.ToolLifecycleEvent { return switch (event) { .provisional => |started| blk: { const call_id = try alloc.dupe(u8, started.id.call_id); errdefer alloc.free(call_id); const tool_name = if (started.tool_name) |name| try alloc.dupe(u8, name) else null; break :blk .{ .provisional = .{ .id = .{ .turn_id = started.id.turn_id, .call_id = call_id }, .presentation_group_id = started.presentation_group_id, .tool_name = tool_name, .activity_kind = started.activity_kind, } }; }, .authoritative_started => |started| blk: { const call_id = try alloc.dupe(u8, started.id.call_id); errdefer alloc.free(call_id); const alias = if (started.reconciles_provisional_call_id) |value| try alloc.dupe(u8, value) else null; errdefer if (alias) |value| alloc.free(value); const tool_name = try alloc.dupe(u8, started.tool_name); errdefer alloc.free(tool_name); const arguments_json = if (started.arguments_json) |value| try alloc.dupe(u8, value) else null; errdefer if (arguments_json) |value| alloc.free(value); break :blk .{ .authoritative_started = .{ .id = .{ .turn_id = started.id.turn_id, .call_id = call_id }, .presentation_group_id = started.presentation_group_id, .reconciles_provisional_call_id = alias, .tool_name = tool_name, .activity_kind = started.activity_kind, .arguments_json = arguments_json, .place_after_current_transcript = started.place_after_current_transcript, } }; }, .progress => |progress| blk: { const call_id = try alloc.dupe(u8, progress.id.call_id); errdefer alloc.free(call_id); const text = try alloc.dupe(u8, progress.text); break :blk .{ .progress = .{ .id = .{ .turn_id = progress.id.turn_id, .call_id = call_id }, .text = text, } }; }, .terminal => |terminal| blk: { const call_id = try alloc.dupe(u8, terminal.id.call_id); errdefer alloc.free(call_id); const summary = try alloc.dupe(u8, terminal.outcome.summary); errdefer alloc.free(summary); const result = if (terminal.result) |value| try alloc.dupe(u8, value) else null; errdefer if (result) |value| alloc.free(value); const result_memory = try dupeToolResultMemory(alloc, terminal.result_memory); errdefer freeToolResultMemory(alloc, result_memory); const command_artifact_handle = if (terminal.command_artifact_handle) |handle| try alloc.dupe(u8, handle) else null; errdefer if (command_artifact_handle) |handle| alloc.free(handle); break :blk .{ .terminal = .{ .id = .{ .turn_id = terminal.id.turn_id, .call_id = call_id }, .outcome = .{ .kind = terminal.outcome.kind, .summary = summary, }, .result = result, .result_memory = result_memory, .command_artifact_handle = command_artifact_handle, } }; }, .turn_finished => |finished| .{ .turn_finished = finished }, };}
pub fn freeToolLifecycleEvent( alloc: std.mem.Allocator, event: types.ToolLifecycleEvent,) void { switch (event) { .provisional => |started| { alloc.free(@constCast(started.id.call_id)); if (started.tool_name) |name| alloc.free(@constCast(name)); }, .authoritative_started => |started| { alloc.free(@constCast(started.id.call_id)); if (started.reconciles_provisional_call_id) |alias| { alloc.free(@constCast(alias)); } alloc.free(@constCast(started.tool_name)); if (started.arguments_json) |arguments_json| { alloc.free(@constCast(arguments_json)); } }, .progress => |progress| { alloc.free(@constCast(progress.id.call_id)); alloc.free(@constCast(progress.text)); }, .terminal => |terminal| { alloc.free(@constCast(terminal.id.call_id)); alloc.free(@constCast(terminal.outcome.summary)); if (terminal.result) |result| alloc.free(@constCast(result)); freeToolResultMemory(alloc, terminal.result_memory); if (terminal.command_artifact_handle) |handle| alloc.free(@constCast(handle)); }, .turn_finished => {}, }}
fn dupeToolResultMemory( alloc: std.mem.Allocator, source: ?types.ToolResultMemory,) !?types.ToolResultMemory { const memory = source orelse return null; const output_handle = if (memory.output_handle) |handle| try alloc.dupe(u8, handle) else null; errdefer if (output_handle) |handle| alloc.free(handle); const preview = if (memory.preview) |value| try alloc.dupe(u8, value) else null; errdefer if (preview) |value| alloc.free(value); const command_output_replay = if (memory.command_output_replay) |replay| try types.dupeCommandOutputReplay(alloc, replay) else null; errdefer if (command_output_replay) |replay| types.freeCommandOutputReplay(alloc, replay); return .{ .output_handle = output_handle, .preview = preview, .output_bytes = memory.output_bytes, .stored_output_bytes = memory.stored_output_bytes, .truncated = memory.truncated, .model_view_covers_full_file = memory.model_view_covers_full_file, .command_output_replay = command_output_replay, .command_process_presentation = memory.command_process_presentation, .terminal_action_presentation = memory.terminal_action_presentation, };}
fn freeToolResultMemory( alloc: std.mem.Allocator, memory: ?types.ToolResultMemory,) void { const value = memory orelse return; if (value.output_handle) |handle| alloc.free(@constCast(handle)); if (value.preview) |preview| alloc.free(@constCast(preview)); if (value.command_output_replay) |replay| { types.freeCommandOutputReplay(alloc, replay); }}
fn makePrompt(alloc: std.mem.Allocator, text: []const u8, model: []const u8) !QueuedPrompt { return .{ .prompt = try alloc.dupe(u8, text), .images = &.{}, .model = try alloc.dupe(u8, model), .api_key = try alloc.dupe(u8, "key"), .permission_mode = .auto, .history = try alloc.alloc(types.HistoryTurn, 0), .grants = try alloc.alloc(types.PermissionGrant, 0), };}
fn makeImagePrompt(alloc: std.mem.Allocator, text: []const u8, model: []const u8) !QueuedPrompt { var prompt = try makePrompt(alloc, text, model); errdefer freeQueuedPrompt(alloc, prompt); const images = try alloc.alloc(types.ImageAttachment, 1); errdefer alloc.free(images); const path = try alloc.dupe(u8, "/tmp/steering.png"); errdefer alloc.free(path); const media_type = try alloc.dupe(u8, "image/png"); errdefer alloc.free(media_type); images[0] = .{ .path = path, .media_type = media_type }; prompt.images = images; return prompt;}
fn makeGrant(alloc: std.mem.Allocator, tool_name: []const u8, target_path: []const u8) !types.PermissionGrant { return .{ .tool_name = try alloc.dupe(u8, tool_name), .target_path = try alloc.dupe(u8, target_path) };}
test "direct steering applies at the boundary without cancellation and seals before teardown" { const alloc = std.testing.allocator; var receipt = SteeringReceipt{}; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try std.testing.expect(runtime.beginDirectProcessing(41)); var prompt = try makePrompt(alloc, "review feedback", "model"); prompt.steering_receipt = &receipt; try std.testing.expect(try runtime.admitActiveSteering(alloc, prompt)); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(SteeringDelivery.queued, receipt.state.load(.seq_cst)); var output = std.heap.ArenaAllocator.init(alloc); defer output.deinit(); const boundary = try runtime.takeSteeringBoundaryInto(alloc, output.allocator(), 41, .finalizing); try std.testing.expectEqualStrings("review feedback", boundary.continue_turn[0]); try std.testing.expectEqual(SteeringDelivery.applied, receipt.state.load(.seq_cst)); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len); try std.testing.expect((try runtime.takeSteeringBoundary(alloc, 41, .finalizing)) == .none); const late = try makePrompt(alloc, "too late", "model"); defer freeQueuedPrompt(alloc, late); try std.testing.expect(!try runtime.admitActiveSteering(alloc, late)); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());}
test "direct steering output allocation failure preserves feedback and cancellation discards it" { const alloc = std.testing.allocator; var receipt = SteeringReceipt{}; var runtime = WorkerRuntime{}; try std.testing.expect(runtime.beginDirectProcessing(41)); var prompt = try makePrompt(alloc, "must not disappear", "model"); prompt.steering_receipt = &receipt; try std.testing.expect(try runtime.admitActiveSteering(alloc, prompt)); var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 }); try std.testing.expectError(error.OutOfMemory, runtime.takeSteeringBoundaryInto(alloc, failing.allocator(), 41, .model)); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); try std.testing.expectEqual(SteeringDelivery.queued, receipt.state.load(.seq_cst)); runtime.requestCancel(); try std.testing.expect((try runtime.takeSteeringBoundary(alloc, 41, .cancelled)) == .interrupt); runtime.deinit(alloc); try std.testing.expectEqual(SteeringDelivery.not_applied, receipt.state.load(.seq_cst));}
test "direct steering rejects inactive cancelled and full-allocation admission without taking ownership" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); const prompt = try makePrompt(alloc, "feedback", "model"); defer freeQueuedPrompt(alloc, prompt); try std.testing.expect(!try runtime.admitActiveSteering(alloc, prompt)); try std.testing.expect(runtime.beginDirectProcessing(41)); var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 }); try std.testing.expectError(error.OutOfMemory, runtime.admitActiveSteering(failing.allocator(), prompt)); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); runtime.requestCancel(); try std.testing.expect(!try runtime.admitActiveSteering(alloc, prompt));}
fn makePromptWithGrant(alloc: std.mem.Allocator, text: []const u8, model: []const u8, tool_name: []const u8, target_path: []const u8) !QueuedPrompt { var prompt = try makePrompt(alloc, text, model); errdefer freeQueuedPrompt(alloc, prompt); const grants = try alloc.alloc(types.PermissionGrant, 1); grants[0] = try makeGrant(alloc, tool_name, target_path); alloc.free(prompt.grants); prompt.grants = grants; return prompt;}
fn freeEventList(alloc: std.mem.Allocator, events: *std.ArrayList(WorkerEvent)) void { for (events.items) |event| freeWorkerEvent(alloc, event); events.deinit(alloc);}
fn expectContinuedSteering(result: SteeringBoundaryResult) ![]const []const u8 { return switch (result) { .continue_turn => |messages| messages, .none, .handoff, .interrupt => error.TestExpectedContinuedSteering, };}
test "subagent steering peek preserves the existing queue consumer" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try std.testing.expect(!runtime.hasPendingPlainSteering()); runtime.worker_processing = true; runtime.active_turn_id = 41; try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model")); try std.testing.expect(runtime.hasPendingPlainSteering()); try std.testing.expect(runtime.hasPendingPlainSteering()); { var images: [1]types.ImageAttachment = undefined; runtime.queued_prompts.items[1].images = &images; defer runtime.queued_prompts.items[1].images = &.{}; try std.testing.expect(!runtime.hasPendingPlainSteering()); } try std.testing.expectEqual(@as(usize, 2), runtime.queued_prompts.items.len); const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled)); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqualStrings("first", guidance[0]); try std.testing.expectEqualStrings("second", guidance[1]); try std.testing.expect(!runtime.hasPendingPlainSteering());}
test "interactive prompt admission derives steering from active worker state" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, 41, .cancelled), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("steer", guidance[0]);}
test "interactive prompt during running manual compaction waits without cancelling" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.enqueueContextCompaction(.{ .turn_id = 41, .model = try alloc.dupe(u8, "provider/model"), .api_key = try alloc.dupe(u8, "key"), .history = try alloc.alloc(types.HistoryTurn, 0), }); const taken = (try runtime.tryTakeNextWork(alloc)) orelse return error.TestExpectedQueuedWork; defer freeWorkItem(alloc, taken); try std.testing.expectEqual(ContextCompactionStatus.running, runtime.contextCompactionStatus());
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
// The compaction is a wait boundary: no cancel, no immediate interrupt. try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(?u64, null), runtime.steering_cancel_turn_id); var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expect(snapshot.waits_for_boundary);
// A compaction work item never takes a steering boundary; finishing it // promotes the waiting prompt to a continuation that runs next. runtime.settleCompactionActivity(taken.compact_context.operation_id.?, .{ .outcome = .succeeded, .stage = .publication, .publication = .committed, }); runtime.finishProcessing(); const next = (try runtime.tryTakeNextWork(alloc)) orelse return error.TestExpectedNextPrompt; defer freeWorkItem(alloc, next); try std.testing.expect(next == .prompt); try std.testing.expect(next.prompt.delivery.isContinuation()); try std.testing.expectEqualStrings("steer", next.prompt.prompt);}
test "interactive prompt during in-turn compaction activity waits without cancelling" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
// In-turn compaction reports through the activity channel, not the tool feed. const operation = runtime.beginCompactionActivity(.automatic, 41); runtime.runCompactionActivity(operation, .summary);
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(?u64, null), runtime.steering_cancel_turn_id);
// The next model boundary drains the waiting steer as same-turn guidance. const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .model)); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("steer", guidance[0]);
runtime.settleCompactionActivity(operation, .{ .outcome = .succeeded, .stage = .summary, .publication = .committed, }); runtime.finishProcessing();}
test "interactive prompt after in-turn compaction settled steers immediately" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
const operation = runtime.beginCompactionActivity(.automatic, 41); runtime.runCompactionActivity(operation, .summary); runtime.settleCompactionActivity(operation, .{ .outcome = .succeeded, .stage = .summary, .publication = .committed, });
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
// No active boundary remains, so immediate-steer semantics are unchanged. try std.testing.expect(runtime.isCancelRequested()); try std.testing.expectEqual(@as(?u64, 41), runtime.steering_cancel_turn_id); const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled)); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("steer", guidance[0]);}
test "active prompt admission drains steering in FIFO order" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model")); const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, 41, .cancelled), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 2), guidance.len); try std.testing.expectEqualStrings("first", guidance[0]); try std.testing.expectEqualStrings("second", guidance[1]); try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items.len); try std.testing.expectEqual(@as(usize, 2), runtime.worker_events.items.len); try std.testing.expectEqualStrings("first", runtime.worker_events.items[0].append_user_feedback); try std.testing.expectEqualStrings("second", runtime.worker_events.items[1].append_user_feedback);}
test "immediate steering owns and clears only its cancellation" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model")); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expect(!runtime.cancellationStopsTurn()); var steering_snapshot = try runtime.snapshotState(alloc); defer steering_snapshot.deinit(alloc); try std.testing.expect(steering_snapshot.cancel_requested); try std.testing.expect(steering_snapshot.cancel_continues_turn);
const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, 41, .cancelled), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 2), guidance.len); try std.testing.expectEqualStrings("first", guidance[0]); try std.testing.expectEqualStrings("second", guidance[1]); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());}
test "explicit interrupt overrides immediate steering cancellation" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model")); runtime.requestInteractiveCancel(); try std.testing.expect(runtime.cancellationStopsTurn()); var batch = runtime.takeEventBatch(); defer freeEventList(alloc, &batch.events); try std.testing.expectEqual(@as(?u64, 41), batch.cancelled_turn_id); var interrupt_snapshot = try runtime.snapshotState(alloc); defer interrupt_snapshot.deinit(alloc); try std.testing.expect(interrupt_snapshot.cancel_requested); try std.testing.expect(!interrupt_snapshot.cancel_continues_turn);
const boundary = try runtime.takeSteeringBoundary(alloc, 41, .cancelled); try std.testing.expect(boundary == .interrupt); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());}
test "interactive prompt after explicit cancellation starts a new visible turn" { const alloc = std.testing.allocator; for ([_]bool{ false, true }) |already_presented| { var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try std.testing.expect(runtime.beginDirectProcessing(41)); runtime.requestInteractiveCancel();
var prompt = try makePrompt(alloc, "ok", "model"); prompt.turn_id = 42; prompt.user_prompt_already_presented = already_presented; try runtime.admitInteractivePrompt(alloc, prompt);
try std.testing.expect(runtime.cancellationStopsTurn()); try std.testing.expect(runtime.queued_prompts.items[0].delivery == .ordinary); try std.testing.expect(try runtime.takeSteeringBoundary(alloc, 41, .cancelled) == .interrupt); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
runtime.finishProcessing(); const next = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, next); try std.testing.expectEqual(@as(u64, 42), next.turn_id); try std.testing.expectEqualStrings("ok", next.prompt); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len); if (already_presented) { try std.testing.expectEqual(@as(u64, 42), runtime.worker_events.items[0].begin_presented_prompt); } else { try std.testing.expectEqualStrings("ok", runtime.worker_events.items[0].begin_prompt.text); } runtime.finishProcessing(); }}
test "sequential tool lifecycle does not inspect compatible allocator context identities" { var first_context: u8 = 0; var second_context: u8 = 0; var first_view = std.heap.c_allocator; var second_view = std.heap.c_allocator; first_view.ptr = @ptrCast(&first_context); second_view.ptr = @ptrCast(&second_context); var runtime = WorkerRuntime{}; defer runtime.deinit(std.heap.c_allocator); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.pushEvent(first_view, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_first" }, .reconciles_provisional_call_id = null, .tool_name = "shell", .activity_kind = .command, } } }); try runtime.pushEvent(first_view, .{ .tool_lifecycle = .{ .terminal = .{ .id = .{ .turn_id = 41, .call_id = "call_first" }, .outcome = .{ .kind = .completed, .summary = "completed" }, } } }); try runtime.pushEvent(second_view, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_second" }, .reconciles_provisional_call_id = null, .tool_name = "shell", .activity_kind = .command, } } });
try std.testing.expect(runtime.active_tool_calls.contains("call_second"));}
test "tool lifecycle decides whether interactive steering interrupts immediately" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .reconciles_provisional_call_id = null, .tool_name = "terminal", .activity_kind = .command, } } }); try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_parallel" }, .reconciles_provisional_call_id = null, .tool_name = "read_file", .activity_kind = .read, } } }); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "wait for tool", "model")); try std.testing.expect(!runtime.isCancelRequested());
try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .terminal = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .outcome = .{ .kind = .completed, .summary = "completed" }, } } }); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "still wait", "model")); try std.testing.expect(!runtime.isCancelRequested());
try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .terminal = .{ .id = .{ .turn_id = 41, .call_id = "call_parallel" }, .outcome = .{ .kind = .completed, .summary = "completed" }, } } }); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "interrupt generation", "model")); try std.testing.expect(runtime.isCancelRequested());}
test "subagent wait phase preserves steering control and existing tool boundaries" { const alloc = std.testing.allocator; for ([_]?u64{ null, 7 }) |tool_step| { var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41; runtime.tool_phase_step_id = tool_step; for ([_][]const u8{ "first steering", "second steering" }) |text| { try runtime.pushEvent(alloc, .{ .turn_phase_update = .{ .turn_id = 41, .step_id = 8, .phase = .waiting_for_subagent, } }); try std.testing.expectEqual(tool_step, runtime.tool_phase_step_id); try std.testing.expectEqual(@as(usize, 0), runtime.active_tool_calls.count()); try std.testing.expect(!runtime.isCancelRequested()); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, text, "model")); try std.testing.expectEqual(tool_step == null, runtime.isCancelRequested()); const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary( alloc, 41, if (tool_step == null) .cancelled else .model, )); defer { for (guidance) |item| alloc.free(item); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings(text, guidance[0]); try std.testing.expect(!runtime.isCancelRequested()); } runtime.requestCancel(); try std.testing.expect((try runtime.takeSteeringBoundary(alloc, 41, .cancelled)) == .interrupt); }}
test "tool handoff phase holds steering only through its model step" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.pushEvent(alloc, .{ .turn_phase_update = .{ .turn_id = 41, .step_id = 7, .phase = .running, } }); try runtime.pushEvent(alloc, .{ .turn_phase_update = .{ .turn_id = 41, .step_id = 7, .phase = .generating, } }); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "wait for pending tools", "model"));
try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expect(snapshot.waits_for_boundary);
const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, 41, .model), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqualStrings("wait for pending tools", guidance[0]);
try runtime.pushEvent(alloc, .{ .turn_phase_update = .{ .turn_id = 41, .step_id = 8, .phase = .thinking, } }); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "interrupt next model step", "model")); try std.testing.expect(runtime.isCancelRequested());}
test "failed steering admission does not cancel the active turn" { var runtime = WorkerRuntime{}; defer runtime.deinit(std.testing.allocator); runtime.worker_processing = true; runtime.active_turn_id = 41;
var failing = std.testing.FailingAllocator.init( std.testing.allocator, .{ .fail_index = 0 }, ); const prompt = QueuedPrompt{ .prompt = @constCast("retain me"), .images = &.{}, .model = @constCast("model"), .api_key = @constCast("key"), .permission_mode = .auto, .history = &.{}, .grants = &.{}, };
try std.testing.expectError( error.OutOfMemory, runtime.admitInteractivePrompt(failing.allocator(), prompt), ); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());}
test "rich interactive input remains steering and hands off at the tool boundary" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .reconciles_provisional_call_id = null, .tool_name = "terminal", .activity_kind = .command, } } });
var prompt = try makePrompt(alloc, "inspect this image", "model"); errdefer freeQueuedPrompt(alloc, prompt); prompt.images = try alloc.alloc(types.ImageAttachment, 1); prompt.images[0] = .{ .path = try alloc.dupe(u8, "/tmp/steering.png"), .media_type = try alloc.dupe(u8, "image/png"), }; try runtime.admitInteractivePrompt(alloc, prompt);
var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len); try std.testing.expect(!runtime.isCancelRequested()); const boundary = try runtime.takeSteeringBoundary(alloc, 41, .model); try std.testing.expect(boundary == .handoff); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());}
test "queued steer waiting at a tool boundary pops back for editing" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41; try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .reconciles_provisional_call_id = null, .tool_name = "terminal", .activity_kind = .command, } } });
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first steer", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second steer", "model")); try std.testing.expect(!runtime.isCancelRequested());
const second = (try runtime.popQueuedSteerForEdit(alloc)).?; defer alloc.free(second); try std.testing.expectEqualStrings("second steer", second); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len); try std.testing.expectEqualStrings("first steer", snapshot.messages[0]); try std.testing.expect(snapshot.waits_for_boundary);
const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, 41, .model), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("first steer", guidance[0]); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null);}
test "immediate steer already committed to an interrupt is not retractable" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model")); try std.testing.expect(runtime.isCancelRequested());
try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, 41, .cancelled), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("steer", guidance[0]);}
test "steer retraction skips handoff prompts and takes the newest text steer" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41; try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .reconciles_provisional_call_id = null, .tool_name = "terminal", .activity_kind = .command, } } });
var rich = try makePrompt(alloc, "inspect this image", "model"); rich.images = try alloc.alloc(types.ImageAttachment, 1); rich.images[0] = .{ .path = try alloc.dupe(u8, "/tmp/steering.png"), .media_type = try alloc.dupe(u8, "image/png"), }; try runtime.admitInteractivePrompt(alloc, rich); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "plain text", "model"));
const text = (try runtime.popQueuedSteerForEdit(alloc)).?; defer alloc.free(text); try std.testing.expectEqualStrings("plain text", text);
try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); const boundary = try runtime.takeSteeringBoundary(alloc, 41, .model); try std.testing.expect(boundary == .handoff);}
test "steer retraction requires an active processing turn" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model")); try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());}
test "failed steer retraction preserves the queue" { const backing = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(backing); runtime.worker_processing = true; runtime.active_turn_id = 41; try runtime.pushEvent(backing, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .reconciles_provisional_call_id = null, .tool_name = "terminal", .activity_kind = .command, } } }); try runtime.admitInteractivePrompt(backing, try makePrompt(backing, "steer", "model"));
var failing = std.testing.FailingAllocator.init( backing, .{ .fail_index = 0 }, ); try std.testing.expectError( error.OutOfMemory, runtime.popQueuedSteerForEdit(failing.allocator()), ); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());}
test "immediate steering is visible while the provider cutoff settles" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model")); var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc);
try std.testing.expectEqual(@as(usize, 2), snapshot.messages.len); try std.testing.expectEqualStrings("first", snapshot.messages[0]); try std.testing.expectEqualStrings("second", snapshot.messages[1]); try std.testing.expect(!snapshot.waits_for_boundary);}
test "steering presentation survives queue to feedback transfer without duplicate input" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try std.testing.expect(runtime.beginDirectProcessing(41)); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "same text", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "same text", "model"));
var queued = try runtime.snapshotSteeringPresentation(alloc); defer queued.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), queued.messages.len); try std.testing.expectEqual(@as(usize, 0), queued.pending_feedback.len);
const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled)); defer { for (guidance) |message| alloc.free(message); alloc.free(guidance); } var consumed = try runtime.snapshotSteeringPresentation(alloc); defer consumed.deinit(alloc); try std.testing.expectEqual(@as(usize, 0), consumed.messages.len); try std.testing.expectEqual(@as(usize, 2), consumed.pending_feedback.len); for (consumed.pending_feedback) |message| try std.testing.expectEqualStrings("same text", message);
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); var drained = try runtime.snapshotSteeringPresentation(alloc); defer drained.deinit(alloc); try std.testing.expectEqual(@as(usize, 0), drained.pending_feedback.len); try std.testing.expectEqual(@as(usize, 2), events.items.len); runtime.finishProcessing();}
test "steering snapshot allocation failure preserves queue and feedback ownership" { try std.testing.checkAllAllocationFailures(std.testing.allocator, struct { fn run(alloc: std.mem.Allocator) !void { const backing = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(backing); try std.testing.expect(runtime.beginDirectProcessing(41)); try runtime.pushEvent(backing, .{ .append_user_feedback = @constCast("earlier") }); try runtime.admitInteractivePrompt(backing, try makePrompt(backing, "later", "model")); var snapshot = runtime.snapshotSteeringPresentation(alloc) catch |err| { try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len); return err; }; defer snapshot.deinit(alloc); try std.testing.expectEqualStrings("earlier", snapshot.pending_feedback[0]); try std.testing.expectEqualStrings("later", snapshot.messages[0]); runtime.finishProcessing(); } }.run, .{});}
test "tool-blocked steering snapshot owns visible messages in admission order" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{ .id = .{ .turn_id = 41, .call_id = "call_running" }, .reconciles_provisional_call_id = null, .tool_name = "terminal", .activity_kind = .command, } } }); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));
var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), snapshot.messages.len); try std.testing.expectEqualStrings("first", snapshot.messages[0]); try std.testing.expectEqualStrings("second", snapshot.messages[1]); try std.testing.expect(snapshot.waits_for_boundary);}
test "late steering keeps admission order when demoted on finish" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 9;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer first", "model")); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queue second", "model")); runtime.finishProcessing();
try std.testing.expect(!runtime.worker_processing); try std.testing.expectEqual(@as(u64, 0), runtime.active_turn_id); try std.testing.expectEqual(@as(usize, 2), runtime.queuedPromptCount()); try std.testing.expectEqualStrings("steer first", runtime.queued_prompts.items[0].prompt); try std.testing.expect(runtime.queued_prompts.items[0].delivery.isContinuation()); try std.testing.expectEqualStrings("queue second", runtime.queued_prompts.items[1].prompt); try std.testing.expect(!runtime.queued_prompts.items[1].delivery.isContinuation());}
test "promoted steering retargets remaining guidance without exposing a queue" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 9;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model")); runtime.finishProcessing();
const promoted = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, promoted); try std.testing.expect(promoted.delivery.isContinuation()); var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len);
const guidance = try expectContinuedSteering( try runtime.takeSteeringBoundary(alloc, promoted.turn_id, .model), ); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("second", guidance[0]); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());}
test "promoted steering keeps rich trailing guidance for its own turn" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 9;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model")); var rich = try makePrompt(alloc, "second with image", "model"); var owns_rich = true; errdefer if (owns_rich) freeQueuedPrompt(alloc, rich); rich.images = try alloc.alloc(types.ImageAttachment, 1); rich.images[0] = .{ .path = try alloc.dupe(u8, "/tmp/trailing.png"), .media_type = try alloc.dupe(u8, "image/png"), }; try runtime.admitInteractivePrompt(alloc, rich); owns_rich = false; runtime.finishProcessing();
const promoted = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, promoted); try std.testing.expect(promoted.delivery.isContinuation()); var snapshot = try runtime.snapshotSteeringPresentation(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len); const boundary = try runtime.takeSteeringBoundary(alloc, promoted.turn_id, .model); try std.testing.expect(boundary == .none); runtime.requestInteractiveCancel();
runtime.finishProcessing(); const trailing = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, trailing); try std.testing.expect(trailing.delivery.isContinuation()); try std.testing.expectEqualStrings("second with image", trailing.prompt);}
test "clear queued prompts also clears steering" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 9;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model")); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model")); runtime.clearQueuedPrompts(alloc, &.{});
try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); const boundary = try runtime.takeSteeringBoundary(alloc, 9, .model); try std.testing.expect(boundary == .none);}
test "idle interactive admission uses ordinary queue" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "next", "model")); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); try std.testing.expect(runtime.queued_prompts.items[0].delivery == .ordinary);}
test "request shutdown sets stop and cancel flags" { var runtime = WorkerRuntime{}; defer runtime.deinit(std.testing.allocator); runtime.requestShutdown(); try std.testing.expect(runtime.worker_stop_requested); try std.testing.expect(runtime.worker_cancel_requested.load(.seq_cst));}
test "takeEventBatch snapshots cancellation with detached events" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.active_turn_id = 7; runtime.worker_cancel_requested.store(true, .seq_cst); runtime.worker_recovery_pause_requested.store(true, .seq_cst); try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("late"), } });
var batch = runtime.takeEventBatch(); defer freeEventList(alloc, &batch.events); runtime.active_turn_id = 8; runtime.worker_cancel_requested.store(false, .seq_cst); try std.testing.expectEqual(@as(?u64, 7), batch.cancelled_turn_id); try std.testing.expectEqual(@as(usize, 1), batch.events.items.len); try std.testing.expect(batch.events.items[0] == .assistant_presentation); try std.testing.expect(batch.events.items[0].assistant_presentation == .text); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);}
test "rich immediate steering revokes an earlier continuable cancellation" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "plain steer", "model")); try runtime.admitInteractivePrompt(alloc, try makeImagePrompt(alloc, "rich steer", "model")); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expect(runtime.cancellationStopsTurn()); try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("must stay hidden") } });
var batch = runtime.takeEventBatch(); defer freeEventList(alloc, &batch.events); try std.testing.expectEqual(@as(?u64, 41), batch.cancelled_turn_id); try std.testing.expectEqual(@as(usize, 1), batch.events.items.len); try std.testing.expect( try runtime.takeSteeringBoundary(alloc, 41, .cancelled) == .interrupt, );}
test "plain input cannot make an earlier rich cancellation continuable" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makeImagePrompt(alloc, "rich steer", "model")); try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "plain next turn", "model")); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expect(runtime.cancellationStopsTurn()); try std.testing.expectEqual(@as(?u64, null), runtime.steering_cancel_turn_id); try std.testing.expectEqual(PromptDelivery.ordinary, runtime.queued_prompts.items[1].delivery);
var batch = runtime.takeEventBatch(); defer freeEventList(alloc, &batch.events); try std.testing.expectEqual(@as(?u64, 41), batch.cancelled_turn_id); try std.testing.expect( try runtime.takeSteeringBoundary(alloc, 41, .cancelled) == .interrupt, );}
test "takeEventBatch keeps the steering tail live before boundary adoption" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41;
try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model")); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expect(!runtime.cancellationStopsTurn()); try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("buffered tail") } }); var batch = runtime.takeEventBatch(); defer freeEventList(alloc, &batch.events); try std.testing.expectEqual(@as(?u64, null), batch.cancelled_turn_id); try std.testing.expectEqual(@as(usize, 1), batch.events.items.len); try std.testing.expectEqualStrings("buffered tail", batch.events.items[0].assistant_presentation.text);
const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled)); defer { for (guidance) |text| alloc.free(text); alloc.free(guidance); } try std.testing.expectEqual(@as(usize, 1), guidance.len); try std.testing.expectEqualStrings("steer", guidance[0]); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expectEqual(@as(?u64, null), batch.cancelled_turn_id); var feedback = runtime.takeEventBatch(); defer freeEventList(alloc, &feedback.events); try std.testing.expectEqual(@as(usize, 1), feedback.events.items.len); try std.testing.expectEqualStrings("steer", feedback.events.items[0].append_user_feedback);}
test "takeEventBatch requires a matching active steering owner to continue" { const cases = [_]struct { active: u64, owner: ?u64, requested: bool, stopped: ?u64 }{ .{ .active = 7, .owner = null, .requested = true, .stopped = 7 }, .{ .active = 7, .owner = 6, .requested = true, .stopped = 7 }, .{ .active = 7, .owner = 7, .requested = true, .stopped = null }, .{ .active = 7, .owner = null, .requested = false, .stopped = null }, .{ .active = 0, .owner = null, .requested = true, .stopped = null }, }; for (cases) |case| { var runtime = WorkerRuntime{}; defer runtime.deinit(std.testing.allocator); runtime.active_turn_id = case.active; runtime.steering_cancel_turn_id = case.owner; runtime.worker_cancel_requested.store(case.requested, .seq_cst); var batch = runtime.takeEventBatch(); defer freeEventList(std.testing.allocator, &batch.events); try std.testing.expectEqual(case.stopped, batch.cancelled_turn_id); }}
test "state snapshot reports completion queued after event batch detach" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.pushEvent(alloc, .{ .command_output = .{ .stream = .stdout, .text = @constCast("record\n"), } }); var batch = runtime.takeEventBatch(); defer freeEventList(alloc, &batch.events); try runtime.pushEvent(alloc, .{ .command_output_complete = null });
var snapshot = try runtime.snapshotState(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), snapshot.pending_event_count); try std.testing.expectEqual(@as(usize, 1), batch.events.items.len); try std.testing.expect(batch.events.items[0] == .command_output); try std.testing.expect(runtime.worker_events.items[0] == .command_output_complete);}
fn checkStateSnapshotFailurePreservesPendingEvents(alloc: std.mem.Allocator) !void { const owner_alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(owner_alloc);
try runtime.pushEvent(owner_alloc, .{ .command_output_complete = null }); runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( owner_alloc, .{ .id = 17, .label = "pending command completion" }, );
var snapshot = runtime.snapshotState(alloc) catch |err| { try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len); try std.testing.expect(runtime.worker_events.items[0] == .command_output_complete); try std.testing.expectEqualStrings( "pending command completion", runtime.pending_permission_request_shared.?.label, ); return err; }; defer snapshot.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), snapshot.pending_event_count); try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len); try std.testing.expect(runtime.worker_events.items[0] == .command_output_complete);}
test "state snapshot allocation failures preserve pending event ownership" { try std.testing.checkAllAllocationFailures( std.testing.allocator, checkStateSnapshotFailurePreservesPendingEvents, .{}, );}
test "finalization failure latches fixed metadata and closes interactive admission" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try std.testing.expect(runtime.interactiveAdmissionSnapshot() == .open); runtime.latchFinalizationFailure(.{ .turn_id = 41, .outcome = .failed, }); runtime.latchFinalizationFailure(.{ .turn_id = 42, .outcome = .interrupted, });
switch (runtime.interactiveAdmissionSnapshot()) { .finalization_failed => |failure| { try std.testing.expectEqual(@as(u64, 41), failure.turn_id); try std.testing.expectEqual(types.TurnPresentationOutcome.failed, failure.outcome); }, else => return error.TestExpectedEqual, } try std.testing.expect(runtime.worker_stop_requested); try std.testing.expect(runtime.worker_cancel_requested.load(.seq_cst));
const prompt = try makePrompt(alloc, "rejected", "model"); try std.testing.expectError(error.TurnFinalizationDeliveryFailed, runtime.enqueuePrompt(alloc, prompt)); freeQueuedPrompt(alloc, prompt); try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items.len);}
test "dedicated turn finalizer queues before optional finish payload" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.pushTurnFinished(alloc, .{ .turn_id = 99, .outcome = .completed, }); try runtime.pushEvent(alloc, .{ .finish_prompt = .{ .turn = .{ .assistant = .{ .user = .{ .text = @constCast("prompt") }, .assistant = @constCast("answer"), } }, } });
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 2), events.items.len); try std.testing.expect(events.items[0] == .tool_lifecycle); try std.testing.expectEqual( @as(u64, 99), events.items[0].tool_lifecycle.turn_finished.turn_id, ); try std.testing.expect(events.items[1] == .finish_prompt);}
test "queue, event, snapshot, sync, history, and grant behavior" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
const first_prompt = try makePrompt(alloc, "first", "old"); try runtime.enqueuePrompt(alloc, first_prompt); const second_prompt = try makePrompt(alloc, "second", "old"); try runtime.enqueuePrompt(alloc, second_prompt); try std.testing.expectEqual(@as(usize, 2), runtime.queuedPromptCount());
runtime.worker_cancel_requested.store(true, .seq_cst); const job = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job); try std.testing.expect(job.turn_id != 0); try std.testing.expectEqualStrings("first", job.prompt); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expect(runtime.worker_processing); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .begin_prompt); try std.testing.expectEqualStrings("first", events.items[0].begin_prompt.text); try std.testing.expect(events.items[0].begin_prompt.text.ptr != job.prompt.ptr);
try runtime.syncQueuedPromptModel(alloc, "next"); try std.testing.expectEqualStrings("next", runtime.queued_prompts.items[0].model); runtime.syncQueuedPromptPermissionSnapshot(.{ .mode = .ask, }); try std.testing.expectEqual(types.PermissionMode.ask, runtime.queued_prompts.items[0].permission_mode); try std.testing.expectEqual(types.PermissionMode.auto, job.permission_mode);
const grant_source = try alloc.alloc(types.PermissionGrant, 1); defer types.freePermissionGrantSlice(alloc, grant_source); grant_source[0] = try makeGrant(alloc, "read_file", "/tmp/a"); try runtime.syncQueuedPromptPermissionState(alloc, grant_source, .{ .mode = .auto, }); grant_source[0].tool_name[0] = 'R'; grant_source[0].target_path[1] = 'T'; try std.testing.expectEqualStrings("read_file", runtime.queued_prompts.items[0].grants[0].tool_name); try std.testing.expectEqualStrings("/tmp/a", runtime.queued_prompts.items[0].grants[0].target_path);
const old_turn: types.HistoryTurn = .{ .compacted_summary = .{ .summary = try alloc.dupe(u8, "old"), .removed_turn_count = 0, .compaction_count = 0, } }; defer types.freeHistoryTurn(alloc, old_turn); const seeded_history = try appendHistoryTurnProjection( alloc, runtime.queued_history, old_turn, 2, ); types.freeHistoryTurnSlice(alloc, runtime.queued_history); runtime.queued_history = seeded_history; const turn: types.HistoryTurn = .{ .compacted_summary = .{ .summary = try alloc.dupe(u8, "summary"), .removed_turn_count = 0, .compaction_count = 0, } }; defer types.freeHistoryTurn(alloc, turn); try runtime.propagateHistoryTurn(alloc, turn, 1); try std.testing.expectEqual(@as(usize, 2), runtime.queued_history.len); try std.testing.expect(runtime.queued_history[1] == .compacted_summary); try std.testing.expectEqualStrings("summary", runtime.queued_history[1].compacted_summary.summary); try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items[0].unversioned_history_count);
try runtime.propagateGrant(alloc, "read_file", "/tmp/a"); try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items[0].grants.len); try runtime.propagateGrant(alloc, "read_file", "/tmp/b"); try std.testing.expectEqual(@as(usize, 2), runtime.queued_prompts.items[0].grants.len);
runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 1, .label = "pending" }, ); var snapshot = try runtime.snapshotState(alloc); defer snapshot.deinit(alloc); try std.testing.expectEqualStrings("pending", snapshot.pending_permission_request.?.label); try std.testing.expect( snapshot.pending_permission_request.?.label.ptr != runtime.pending_permission_request_shared.?.label.ptr, );}
test "permission response durable barrier commits before waiter wake" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 9, .label = "prepared action" }, );
const Capture = struct { runtime: *WorkerRuntime, committed: bool = false,
fn commit(raw: *anyopaque) WorkerRuntime.PermissionCommitError!void { const self: *@This() = @ptrCast(@alignCast(raw)); if (self.runtime.pending_permission_response != null) { return error.PermissionCommitFailed; } self.committed = true; } }; var capture = Capture{ .runtime = &runtime }; const result = try runtime.submitPermissionResponseAfterCommit( 9, permission_request.OwnedPermissionResponse.init(alloc, .always, null), .{ .context = &capture, .commit_fn = Capture.commit }, ); try std.testing.expectEqual(PermissionSubmissionResult.accepted, result); try std.testing.expect(capture.committed); try std.testing.expect(runtime.pending_permission_response != null);}
test "permission response durable barrier failure does not wake waiter" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 10, .label = "prepared action" }, );
const Fail = struct { fn commit(_: *anyopaque) WorkerRuntime.PermissionCommitError!void { return error.PermissionCommitFailed; } }; try std.testing.expectError( error.PermissionCommitFailed, runtime.submitPermissionResponseAfterCommit( 10, permission_request.OwnedPermissionResponse.init(alloc, .once, null), .{ .context = &runtime, .commit_fn = Fail.commit }, ), ); try std.testing.expect(runtime.pending_permission_response == null); try std.testing.expect(runtime.pending_permission_waiting);}
test "permission capacity rejection removes the unpublished waiter" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
const Reject = struct { fn observe( _: *anyopaque, _: *WorkerRuntime, _: permission_request.PermissionRequest, ) error{ OutOfMemory, PermissionRegistrationFailed, PermissionCapacityExceeded, }!void { return error.PermissionCapacityExceeded; } }; try std.testing.expectError( error.PermissionCapacityExceeded, runtime.requestPermissionBlockingObserved( alloc, .{ .id = 11, .label = "capacity-bound action" }, null, .{ .context = &runtime, .observe_fn = Reject.observe }, ), ); try std.testing.expect(!runtime.pending_permission_waiting); try std.testing.expect(runtime.pending_permission_request_shared == null); try std.testing.expect(runtime.pending_permission_review == null); try std.testing.expect(runtime.pending_permission_response == null);}
test "waitAndTakeNextPrompt emits bound skill prompt event when queued prompt has bindings" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var prompt = try makePrompt(alloc, "raw $review then $review and $review", "model"); errdefer freeQueuedPrompt(alloc, prompt); prompt.skill_bindings = try dupeSkillBindings(alloc, &[_]SkillBinding{.{ .name = @constCast("review"), .path = @constCast("/tmp/.codex/skills/review"), }}); prompt.skill_display_spans = try dupeSkillDisplaySpans(alloc, &[_]SkillDisplaySpan{ .{ .raw_start = "raw $review then ".len, .raw_end = "raw $review then $review".len, .name = @constCast("review"), .path = @constCast("/tmp/.codex/skills/review"), .display_source = .workspace_codex, .owns_trailing_separator = true, }, .{ .raw_start = "raw $review then $review and ".len, .raw_end = "raw $review then $review and $review".len, .name = @constCast("review"), .path = @constCast("/tmp/.codex/skills/review"), }, }); try runtime.enqueuePrompt(alloc, prompt);
const job = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job);
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .begin_prompt_with_skill_bindings); try std.testing.expectEqualStrings("raw $review then $review and $review", events.items[0].begin_prompt_with_skill_bindings.prompt.text); try std.testing.expectEqual(@as(usize, 1), events.items[0].begin_prompt_with_skill_bindings.skill_bindings.len); try std.testing.expectEqualStrings("review", events.items[0].begin_prompt_with_skill_bindings.skill_bindings[0].name); try std.testing.expectEqualStrings("/tmp/.codex/skills/review", events.items[0].begin_prompt_with_skill_bindings.skill_bindings[0].path); try std.testing.expectEqual(@as(usize, 2), events.items[0].begin_prompt_with_skill_bindings.skill_display_spans.len); try std.testing.expectEqual(@as(usize, "raw $review then ".len), events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].raw_start); try std.testing.expectEqual(@as(usize, "raw $review then $review".len), events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].raw_end); try std.testing.expectEqualStrings("review", events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].name); try std.testing.expectEqual( skill_contract.SkillSource.workspace_codex, events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].display_source.?, ); try std.testing.expect(events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].owns_trailing_separator);}
test "permission snapshot keeps the full file review borrowed and request-bound" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); var review = try diff_mod.FileReview.init( alloc, "", "line 1\nline 2\nline 3\nline 4\nline 5\nline 6\nline 7\n", ); defer review.deinit(alloc);
runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 44, .label = "file_mutation", .file = .{ .kind = .write, .intent = .mutation, .preview = .{ .path = "note.txt", .lines = &.{ .{ .op = .addition, .new_line = 1, .text = "line 1" }, .{ .op = .addition, .new_line = 7, .text = "line 7" }, }, .additions = 7, .deletions = 0, .truncated = true, }, .scope = .workspace_files, }, }, ); runtime.pending_permission_review = &review;
var snapshot = try runtime.snapshotState(alloc); defer snapshot.deinit(alloc);
try std.testing.expectEqual( @as(usize, 2), snapshot.pending_permission_request.?.file.?.preview.lines.len, ); try std.testing.expectEqual(@as(u64, 44), snapshot.pending_permission_review.?.request_id); try std.testing.expect(snapshot.pending_permission_review.?.review == &review); try std.testing.expectEqual(@as(usize, 7), snapshot.pending_permission_review.?.review.rowCount());}
test "held turn propagation refreshes queued prompt trusted reviewer context" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var queued = try makePrompt(alloc, "queued current prompt", "model"); var canonical_history = [_]types.HistoryTurn{.{ .assistant = .{ .user = .{ .text = @constCast("original first request") }, .assistant = @constCast("original answer"), } }}; queued.root_user_intent_context = try auto_classifier_context.buildCanonicalRootUserContext( alloc, queued.prompt, &canonical_history, ); try runtime.enqueuePrompt(alloc, queued); const original_prompt_ptr = runtime.queued_prompts.items[0].prompt.ptr; const original_prompt_len = runtime.queued_prompts.items[0].prompt.len;
const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = try alloc.dupe(u8, "Revoke permission: do not modify any more files.") }, .assistant = try alloc.dupe(u8, "completed prior answer"), } }; defer types.freeHistoryTurn(alloc, turn);
try runtime.propagateHistoryTurn(alloc, turn, 8);
try std.testing.expectEqual(@as(usize, 1), runtime.queued_history.len); try std.testing.expect(runtime.queued_history[0] == .assistant); try std.testing.expectEqualStrings("Revoke permission: do not modify any more files.", runtime.queued_history[0].assistant.user.text); try std.testing.expectEqualStrings("completed prior answer", runtime.queued_history[0].assistant.assistant); try std.testing.expectEqualStrings("queued current prompt", runtime.queued_prompts.items[0].prompt); try std.testing.expectEqual(original_prompt_len, runtime.queued_prompts.items[0].prompt.len); try std.testing.expect(original_prompt_ptr == runtime.queued_prompts.items[0].prompt.ptr);
const review_context = runtime.queued_prompts.items[0].root_user_intent_context; try std.testing.expect(std.mem.find( u8, review_context, "first_root_user_request: original first request", ) != null); try std.testing.expect(std.mem.find( u8, review_context, "recent_root_user_request: Revoke permission: do not modify any more files.", ) != null); try std.testing.expect(std.mem.find( u8, review_context, "current_request: queued current prompt", ) != null);
const newer_turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = try alloc.dupe(u8, "newer completed request") }, .assistant = try alloc.dupe(u8, "newer completed answer"), } }; defer types.freeHistoryTurn(alloc, newer_turn); try runtime.propagateHistoryTurn(alloc, newer_turn, 8); const refreshed = runtime.queued_prompts.items[0].root_user_intent_context; try std.testing.expect(std.mem.find( u8, refreshed, "recent_root_user_request: Revoke permission: do not modify any more files.", ) != null); try std.testing.expect(std.mem.find( u8, refreshed, "recent_root_user_request: newer completed request", ) != null);}
test "completed image turns propagate authority to queued follow-ups" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var prompt = try makePrompt(alloc, "queued follow-up", "model"); prompt.authorized_image_catalog = try types.dupeImageAttachmentSlice(alloc, &.{.{ .id = 2, .path = @constCast("/tmp/older.png"), .media_type = @constCast("image/png"), }}); try runtime.enqueuePrompt(alloc, prompt);
var completed_images = [_]types.ImageAttachment{.{ .id = 8, .path = @constCast("/tmp/completed.png"), .media_type = @constCast("image/png"), }}; const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = @constCast("active image turn"), .images = &completed_images }, .assistant = @constCast("done"), } }; try runtime.propagateHistoryTurn(alloc, turn, 1);
const queued = runtime.queued_prompts.items[0]; try std.testing.expectEqual(@as(usize, 2), queued.authorized_image_catalog.len); try std.testing.expectEqual(@as(usize, 2), queued.authorized_image_catalog[0].id); try std.testing.expectEqual(@as(usize, 8), queued.authorized_image_catalog[1].id); try std.testing.expectEqualStrings("/tmp/completed.png", queued.authorized_image_catalog[1].path);}
test "explicit cancellation keeps targeted steering runnable" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 17; try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer now", "model"));
runtime.requestInteractiveCancel(); try std.testing.expect(runtime.isCancelRequested());
runtime.finishProcessing(); const next = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, next); try std.testing.expectEqualStrings("steer now", next.prompt);}
const PromptTakeThreadState = struct { started: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), finished: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), err: ?anyerror = null, job: ?QueuedPrompt = null,};
fn runPromptTake(state: *PromptTakeThreadState, runtime: *WorkerRuntime) void { state.started.store(true, .seq_cst); state.job = runtime.waitAndTakeNextPrompt(std.testing.allocator) catch |err| { state.err = err; state.finished.store(true, .seq_cst); return; }; state.finished.store(true, .seq_cst);}
test "turn start hold rejects busy worker and blocks take while held" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.worker_processing = true; try std.testing.expect(!runtime.tryHoldTurnStart()); runtime.worker_processing = false;
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued blocks park", "model")); try std.testing.expect(!runtime.tryHoldTurnStart()); runtime.clearQueuedPrompts(alloc, &.{});
try std.testing.expect(runtime.tryHoldTurnStart()); try std.testing.expect(!runtime.tryHoldTurnStart());
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "held blocked prompt", "model")); var state = PromptTakeThreadState{}; const thread = try std.Thread.spawn(.{}, runPromptTake, .{ &state, &runtime }); var joined = false; defer if (!joined) { runtime.releaseTurnStartHold(); runtime.requestStop(); thread.join(); };
var attempts: usize = 0; while (!state.started.load(.seq_cst) and attempts < 100) : (attempts += 1) { io_mod.sleep(std.time.ns_per_ms); } try std.testing.expect(state.started.load(.seq_cst)); io_mod.sleep(10 * std.time.ns_per_ms); try std.testing.expect(!state.finished.load(.seq_cst));
runtime.releaseTurnStartHold(); thread.join(); joined = true; try std.testing.expect(state.err == null); const job = state.job.?; defer freeQueuedPrompt(alloc, job); try std.testing.expectEqualStrings("held blocked prompt", job.prompt);}
test "waitAndTakeNextPrompt assigns turn id and resets cancellation for next prompt" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.worker_cancel_requested.store(true, .seq_cst); try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "next", "model")); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
const job = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job);
try std.testing.expect(job.turn_id != 0); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expect(!runtime.worker_recovery_pause_requested.load(.seq_cst));
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .begin_prompt);}
test "already-presented queued prompt emits identity without duplicating user payload" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var queued = try makePrompt(alloc, "already visible", "model"); queued.turn_id = 9001; queued.user_prompt_already_presented = true; try runtime.enqueuePrompt(alloc, queued);
const job = (try runtime.tryTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job); var events = runtime.takeEvents(); defer freeEventList(alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .begin_presented_prompt); try std.testing.expectEqual(@as(u64, 9001), events.items[0].begin_presented_prompt);}
test "requestRecoveryPause wakes the active operation without losing pause intent" { var worker = WorkerRuntime{}; worker.requestRecoveryPause(); try std.testing.expect(worker.worker_recovery_pause_requested.load(.seq_cst)); try std.testing.expect(worker.worker_cancel_requested.load(.seq_cst));}
test "enqueuePrompt stamps agent settings from worker runtime" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.agent_turn_settings = .{ .max_tool_result_bytes = 4096, .first_call_tool_choice = .none, }; try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "next", "model")); const job = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job);
try std.testing.expectEqual(@as(usize, 4096), job.agent_settings.max_tool_result_bytes); try std.testing.expectEqual(types.ToolChoice.none, job.agent_settings.first_call_tool_choice);}
test "enqueuePrompt stamps fast mode and effort from worker runtime" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.agent_turn_settings = .{ .fast_mode = true, .effort = types.ReasoningEffort.literal("high"), };
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "next", "model")); const job = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job);
try std.testing.expect(job.agent_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), job.agent_settings.effort);}
test "sync queued prompt fast mode and effort updates queued prompts and future defaults" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.agent_turn_settings = .{ .fast_mode = false, .effort = types.ReasoningEffort.literal("low"), }; try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));
runtime.syncQueuedPromptFastMode(true); runtime.syncQueuedPromptEffort(types.ReasoningEffort.literal("high"));
try std.testing.expect(runtime.queued_prompts.items[0].agent_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.queued_prompts.items[0].agent_settings.effort); try std.testing.expect(runtime.agent_turn_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.agent_turn_settings.effort);
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "future", "model"));
try std.testing.expect(runtime.queued_prompts.items[1].agent_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.queued_prompts.items[1].agent_settings.effort);}
const EnqueueThreadState = struct { started: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), err: ?anyerror = null,};
fn runEnqueuePrompt(state: *EnqueueThreadState, runtime: *WorkerRuntime, text: []const u8, model: []const u8) void { const prompt = makePrompt(std.testing.allocator, text, model) catch |err| { state.err = err; return; }; state.started.store(true, .seq_cst); runtime.enqueuePrompt(std.testing.allocator, prompt) catch |err| { freeQueuedPrompt(std.testing.allocator, prompt); state.err = err; return; };}
fn waitForEnqueueThreadStart(state: *EnqueueThreadState) !void { var attempts: usize = 0; while (attempts < 100) : (attempts += 1) { if (state.started.load(.seq_cst)) return; io_mod.sleep(1 * std.time.ns_per_ms); } return error.TestExpectedEqual;}
test "enqueuePrompt snapshots settings under the same mutex as sync" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.agent_turn_settings = .{ .fast_mode = false, .effort = types.ReasoningEffort.literal("low"), };
runtime.worker_mutex.lockUncancelable(io_mod.getIo()); var state = EnqueueThreadState{}; const thread = try std.Thread.spawn(.{}, runEnqueuePrompt, .{ &state, &runtime, "blocked", "model" }); try waitForEnqueueThreadStart(&state); io_mod.sleep(10 * std.time.ns_per_ms); runtime.agent_turn_settings.fast_mode = true; runtime.agent_turn_settings.effort = types.ReasoningEffort.literal("high"); runtime.worker_mutex.unlock(io_mod.getIo()); thread.join();
try std.testing.expect(state.err == null); try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items.len); try std.testing.expect(runtime.queued_prompts.items[0].agent_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.queued_prompts.items[0].agent_settings.effort);}
test "taken prompt keeps original fast mode and effort after later sync" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
runtime.agent_turn_settings = .{ .fast_mode = true, .effort = types.ReasoningEffort.literal("high"), }; try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "active", "model"));
const job = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, job);
runtime.syncQueuedPromptFastMode(false); runtime.syncQueuedPromptEffort(types.ReasoningEffort.literal("low"));
try std.testing.expect(job.agent_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), job.agent_settings.effort); try std.testing.expect(!runtime.agent_turn_settings.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("low"), runtime.agent_turn_settings.effort);}
test "effectiveAgentTurnSettings prefers active turn settings until cleared" { var runtime = WorkerRuntime{}; defer runtime.deinit(std.testing.allocator);
runtime.agent_turn_settings = .{ .max_tool_result_bytes = 2048, .first_call_tool_choice = .auto, .fast_mode = false, .effort = types.ReasoningEffort.literal("low"), }; try std.testing.expectEqual(@as(usize, 2048), runtime.effectiveAgentTurnSettings().max_tool_result_bytes); try std.testing.expect(!runtime.effectiveAgentTurnSettings().fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("low"), runtime.effectiveAgentTurnSettings().effort);
runtime.setActiveAgentTurnSettings(.{ .max_tool_result_bytes = 4096, .first_call_tool_choice = .none, .fast_mode = true, .effort = types.ReasoningEffort.literal("high"), });
const active = runtime.effectiveAgentTurnSettings(); try std.testing.expectEqual(@as(usize, 4096), active.max_tool_result_bytes); try std.testing.expectEqual(types.ToolChoice.none, active.first_call_tool_choice); try std.testing.expect(active.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("high"), active.effort);
runtime.clearActiveAgentTurnSettings(); const fallback = runtime.effectiveAgentTurnSettings(); try std.testing.expectEqual(@as(usize, 2048), fallback.max_tool_result_bytes); try std.testing.expect(!fallback.fast_mode); try std.testing.expectEqual(types.ReasoningEffort.literal("low"), fallback.effort);}
test "queued permission mode updates without mutating dequeued fallback" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var held_prompt = try makePrompt(alloc, "held", "model"); held_prompt.permission_mode = .ask; try runtime.enqueuePrompt(alloc, held_prompt); var queued_prompt = try makePrompt(alloc, "queued", "model"); queued_prompt.permission_mode = .ask; try runtime.enqueuePrompt(alloc, queued_prompt);
const active = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, active); runtime.syncQueuedPromptPermissionSnapshot(.{ .mode = .auto, });
try std.testing.expectEqual(types.PermissionMode.ask, active.permission_mode); try std.testing.expectEqual(types.PermissionMode.auto, runtime.queued_prompts.items[0].permission_mode);}
test "submitted text only queues while a prompt is active" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "use all your tools", "model")); const active = (try runtime.waitAndTakeNextPrompt(alloc)).?; defer freeQueuedPrompt(alloc, active); try std.testing.expect(runtime.worker_processing); try std.testing.expect(!runtime.isCancelRequested());
runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 1, .label = "shell.run launch chrome" }, ); const question_options = [_]types.QuestionOption{.{ .label = "Wait", .description = null }}; const question_entries = [_]types.QuestionBatchEntry{.{ .question = "Keep waiting?", .options = &question_options, }}; runtime.pending_question_shared = try dupeOwnedQuestionBatch( alloc, &question_entries, .agent_question, );
const submitted = [_][]const u8{ "Stop", "continue", "What are you doing right now?", "What files are in this repository?", "checkpoint?", }; for (submitted) |text| { try runtime.enqueuePrompt(alloc, try makePrompt(alloc, text, "model")); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expect(runtime.pending_permission_response == null); try std.testing.expect(runtime.pending_permission_request_shared != null); try std.testing.expect(runtime.pending_question_shared != null); try std.testing.expect(runtime.pending_question_response == .pending); }
try std.testing.expectEqual(submitted.len, runtime.queuedPromptCount()); for (submitted, runtime.queued_prompts.items) |expected, queued| { try std.testing.expectEqualStrings(expected, queued.prompt); }}
test "prompt take and grant append allocation failures preserve state" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.enqueuePrompt(alloc, try makePromptWithGrant(alloc, "keep", "model", "read_file", "/tmp/a")); runtime.worker_cancel_requested.store(true, .seq_cst); var failing_take = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 }); try std.testing.expectError(error.OutOfMemory, runtime.waitAndTakeNextPrompt(failing_take.allocator())); try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount()); try std.testing.expectEqualStrings("keep", runtime.queued_prompts.items[0].prompt); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
const original = runtime.queued_prompts.items[0].grants.ptr; var failing_grant = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 1 }); try std.testing.expectError(error.OutOfMemory, runtime.propagateGrant(failing_grant.allocator(), "write_file", "/tmp/b")); try std.testing.expectEqual(original, runtime.queued_prompts.items[0].grants.ptr); try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items[0].grants.len);
var failing_slice = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 2 }); try std.testing.expectError(error.OutOfMemory, runtime.propagateGrant(failing_slice.allocator(), "write_file", "/tmp/c")); try std.testing.expectEqual(original, runtime.queued_prompts.items[0].grants.ptr); try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items[0].grants.len);}
test "clear queued prompts preserves events and discard frees event payloads" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "drop", "model")); try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("kept"), } }); runtime.clearQueuedPrompts(alloc, &.{}); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items.len); try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len);
runtime.discardEvents(alloc); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
freeQueuedPrompt(alloc, try makePrompt(alloc, "free", "model")); freeWorkerEvent(alloc, .{ .begin_prompt = try types.dupeUserTurn(alloc, .{ .text = @constCast("user"), .images = &.{} }) }); freeWorkerEvent(alloc, .{ .assistant_presentation = .{ .text = try alloc.dupe(u8, "append"), } }); const table = try assistant_presentation.parseTablePayload(alloc, "| H |\n|---|\n| v |\n"); freeWorkerEvent(alloc, .{ .assistant_presentation = .{ .table = table } }); freeWorkerEvent(alloc, try dupeWorkerEvent(alloc, .{ .semantic_notice = .{ .topic = "system", .tone = .information, .body = "notice", } })); freeWorkerEvent(alloc, .{ .route_recovery_status = .{ .kind = .auto_retry, .failed_attempt = 1, .attempt_limit = 3 } }); freeWorkerEvent(alloc, .clear_route_recovery_status); freeWorkerEvent(alloc, .{ .command_output = .{ .stream = .stdout, .text = try alloc.dupe(u8, "out") } }); freeWorkerEvent(alloc, .{ .command_output_complete = null }); freeWorkerEvent(alloc, .{ .command_output_complete = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") } }); freeWorkerEvent(alloc, .{ .tool_lifecycle = .{ .provisional = .{ .id = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") }, .tool_name = try alloc.dupe(u8, "read_file"), .activity_kind = .read, }, } }); freeWorkerEvent(alloc, .{ .tool_lifecycle = .{ .progress = .{ .id = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") }, .text = try alloc.dupe(u8, "reading"), }, } }); freeWorkerEvent(alloc, .{ .tool_lifecycle = .{ .terminal = .{ .id = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") }, .outcome = .{ .kind = .completed, .summary = try alloc.dupe(u8, "read"), }, }, } }); freeWorkerEvent(alloc, .{ .turn_token_update = .{ .input_tokens = 10, .output_tokens = 5 } }); freeWorkerEvent(alloc, .{ .diff_block = .{ .preview = try alloc.dupe(u8, "preview") } }); freeWorkerEvent(alloc, .{ .tool_lifecycle = .{ .turn_finished = .{ .turn_id = 1, .outcome = .completed }, } }); freeWorkerEvent(alloc, .{ .finish_prompt = .{ .turn = .{ .compacted_summary = .{ .summary = try alloc.dupe(u8, "summary"), .removed_turn_count = 1, .compaction_count = 1, } } } }); freeWorkerEvent(alloc, .{ .session_grant = try makeGrant(alloc, "read_file", "/tmp/a") }); freeWorkerEvent(alloc, try dupeWorkerEvent(alloc, .{ .error_text = .{ .topic = "system", .tone = .@"error", .body = "err", } }));}
test "typed lifecycle worker events duplicate and free every payload variant" { const alloc = std.testing.allocator; const events = [_]types.ToolLifecycleEvent{ .{ .provisional = .{ .id = .{ .turn_id = 1, .call_id = "provisional" }, .presentation_group_id = .{ .turn_id = 1, .anchor_step_id = 3 }, .tool_name = "read_file", .activity_kind = .read, } }, .{ .provisional = .{ .id = .{ .turn_id = 1, .call_id = "provisional" }, .tool_name = "glob_files", .activity_kind = .list, } }, .{ .authoritative_started = .{ .id = .{ .turn_id = 1, .call_id = "final" }, .presentation_group_id = .{ .turn_id = 1, .anchor_step_id = 3 }, .reconciles_provisional_call_id = "provisional", .tool_name = "glob_files", .activity_kind = .list, } }, .{ .progress = .{ .id = .{ .turn_id = 1, .call_id = "final" }, .text = "Listing files", } }, .{ .terminal = .{ .id = .{ .turn_id = 1, .call_id = "final" }, .outcome = .{ .kind = .completed, .summary = "Listed files" }, .command_artifact_handle = "fx-command-final.log", } }, .{ .turn_finished = .{ .turn_id = 1, .outcome = .completed } }, };
for (events) |event| { const owned = try dupeToolLifecycleEvent(alloc, event); freeToolLifecycleEvent(alloc, owned); }}
fn checkToolLifecycleDupAllocationFailure(alloc: std.mem.Allocator) !void { const owned = try dupeToolLifecycleEvent(alloc, .{ .authoritative_started = .{ .id = .{ .turn_id = 7, .call_id = "final" }, .reconciles_provisional_call_id = "provisional", .tool_name = "read_file", .activity_kind = .read, } }); defer freeToolLifecycleEvent(alloc, owned);
try std.testing.expectEqualStrings( "final", owned.authoritative_started.id.call_id, ); try std.testing.expectEqualStrings( "provisional", owned.authoritative_started.reconciles_provisional_call_id.?, ); try std.testing.expectEqualStrings( "read_file", owned.authoritative_started.tool_name, );}
test "typed lifecycle duplication frees every partial allocation" { try std.testing.checkAllAllocationFailures( std.testing.allocator, checkToolLifecycleDupAllocationFailure, .{}, );}
test "pushEvent copies session grants into queue allocator" { const source_alloc = std.testing.allocator; const queue_alloc = std.heap.c_allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(queue_alloc);
const grant = try makeGrant(source_alloc, "read_file", "/tmp/source.zig"); const tool_ptr = grant.tool_name.ptr; const target_ptr = grant.target_path.ptr;
try runtime.pushEvent(queue_alloc, .{ .session_grant = grant }); source_alloc.free(grant.tool_name); source_alloc.free(grant.target_path);
var events = runtime.takeEvents(); defer freeEventList(queue_alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .session_grant); try std.testing.expectEqualStrings("read_file", events.items[0].session_grant.tool_name); try std.testing.expectEqualStrings("/tmp/source.zig", events.items[0].session_grant.target_path); try std.testing.expect(events.items[0].session_grant.tool_name.ptr != tool_ptr); try std.testing.expect(events.items[0].session_grant.target_path.ptr != target_ptr);}
test "pushEvent copies string events into queue allocator" { const source_alloc = std.testing.allocator; const queue_alloc = std.heap.c_allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(queue_alloc);
const text = try source_alloc.dupe(u8, "borrowed assistant text"); const text_ptr = text.ptr;
try runtime.pushEvent(queue_alloc, .{ .assistant_presentation = .{ .text = text } }); source_alloc.free(text);
var events = runtime.takeEvents(); defer freeEventList(queue_alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .assistant_presentation); try std.testing.expect(events.items[0].assistant_presentation == .text); try std.testing.expectEqualStrings("borrowed assistant text", events.items[0].assistant_presentation.text); try std.testing.expect(events.items[0].assistant_presentation.text.ptr != text_ptr);}
test "pushEvent copies semantic table payloads into the queue allocator" { const source_alloc = std.testing.allocator; const queue_alloc = std.heap.c_allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(queue_alloc);
var table = try assistant_presentation.parseTablePayload( source_alloc, "| Name | Count |\n" ++ "|------|------:|\n" ++ "| api | 7 |\n", ); const source_cell_ptr = table.rows[1].cells[0].ptr; try runtime.pushEvent(queue_alloc, .{ .assistant_presentation = .{ .table = table } }); table.deinit(source_alloc);
var events = runtime.takeEvents(); defer freeEventList(queue_alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .assistant_presentation); try std.testing.expect(events.items[0].assistant_presentation == .table); try std.testing.expectEqualStrings("api", events.items[0].assistant_presentation.table.rows[1].cells[0]); try std.testing.expect(events.items[0].assistant_presentation.table.rows[1].cells[0].ptr != source_cell_ptr);}
test "pushEvent copies code block payload into queue allocator" { const source_alloc = std.testing.allocator; const queue_alloc = std.heap.c_allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(queue_alloc);
var block = assistant_presentation.CodeBlockPayload{ .language = try source_alloc.dupe(u8, "zig"), .code = try source_alloc.dupe(u8, " const value = 1;\n"), }; const source_code_ptr = block.code.ptr; try runtime.pushEvent(queue_alloc, .{ .assistant_presentation = .{ .code_block = block } }); block.deinit(source_alloc);
var events = runtime.takeEvents(); defer freeEventList(queue_alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .assistant_presentation); try std.testing.expect(events.items[0].assistant_presentation == .code_block); try std.testing.expectEqualStrings("zig", events.items[0].assistant_presentation.code_block.language); try std.testing.expectEqualStrings(" const value = 1;\n", events.items[0].assistant_presentation.code_block.code); try std.testing.expect(events.items[0].assistant_presentation.code_block.code.ptr != source_code_ptr);}
test "pushEvent copies absolute turn token progress into queue allocator" { const queue_alloc = std.heap.c_allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(queue_alloc);
const progress: types.TurnTokenProgress = .{ .input_tokens = 10, .output_tokens = 5, .input_exact = false, .output_exact = false, }; try runtime.pushEvent(queue_alloc, .{ .turn_token_update = progress });
var events = runtime.takeEvents(); defer freeEventList(queue_alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .turn_token_update); try std.testing.expectEqual(progress, events.items[0].turn_token_update);}
test "pushEvent copies finish prompt assistant turns into queue allocator" { const source_alloc = std.testing.allocator; const queue_alloc = std.heap.c_allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(queue_alloc);
const turn: types.HistoryTurn = .{ .assistant = .{ .user = .{ .text = try source_alloc.dupe(u8, "source prompt"), .images = &.{} }, .assistant = try source_alloc.dupe(u8, "source answer"), } }; const prompt_ptr = turn.assistant.user.text.ptr; const answer_ptr = turn.assistant.assistant.ptr;
try runtime.pushEvent(queue_alloc, .{ .finish_prompt = .{ .turn = turn, .summary = .{ .thinking_duration_ms = 1, .turn_duration_ms = 2, .token_progress = .{ .input_tokens = 7 }, }, } }); types.freeHistoryTurn(source_alloc, turn);
var events = runtime.takeEvents(); defer freeEventList(queue_alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .finish_prompt); try std.testing.expect(events.items[0].finish_prompt.turn == .assistant); try std.testing.expectEqualStrings("source prompt", events.items[0].finish_prompt.turn.assistant.user.text); try std.testing.expectEqualStrings("source answer", events.items[0].finish_prompt.turn.assistant.assistant); try std.testing.expect(events.items[0].finish_prompt.turn.assistant.user.text.ptr != prompt_ptr); try std.testing.expect(events.items[0].finish_prompt.turn.assistant.assistant.ptr != answer_ptr); try std.testing.expectEqual(@as(u64, 7), events.items[0].finish_prompt.summary.?.token_progress.input_tokens);}
test "pushOwnedEvent transfers diff block payloads without copying" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
const preview = try alloc.dupe(u8, "preview"); try runtime.pushOwnedEvent(alloc, .{ .diff_block = .{ .preview = preview, } });
var events = runtime.takeEvents(); defer freeEventList(alloc, &events);
try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .diff_block); try std.testing.expectEqual(preview.ptr, events.items[0].diff_block.preview.ptr);}
test "dupeWorkerEvent clones canonical diff payload and full detail" { const alloc = std.testing.allocator; const source: WorkerEvent = .{ .diff_block = .{ .preview = @constCast("preview"), .additions = 2, .deletions = 1, .full = .{ .content = @constCast("full detail"), .lifecycle_id = .{ .turn_id = 9, .call_id = "call-9" }, }, } }; const owned = try dupeWorkerEvent(alloc, source); defer freeWorkerEvent(alloc, owned);
try std.testing.expectEqualStrings("preview", owned.diff_block.preview); try std.testing.expect(source.diff_block.preview.ptr != owned.diff_block.preview.ptr); try std.testing.expectEqual(@as(u32, 2), owned.diff_block.additions); try std.testing.expectEqual(@as(u32, 1), owned.diff_block.deletions); const full = owned.diff_block.full orelse return error.TestExpectedFullDiff; try std.testing.expectEqualStrings("full detail", full.content); try std.testing.expectEqual(@as(u64, 9), full.lifecycle_id.turn_id); try std.testing.expectEqualStrings("call-9", full.lifecycle_id.call_id); try std.testing.expect(source.diff_block.full.?.content.ptr != full.content.ptr); try std.testing.expect(source.diff_block.full.?.lifecycle_id.call_id.ptr != full.lifecycle_id.call_id.ptr);}
test "dupeWorkerEvent owns a complete refreshed credential publication" { const alloc = std.testing.allocator; const source: WorkerEvent = .{ .credential_refreshed = .{ .token = @constCast("fresh-token"), .source = .fx_login, .team_id = @constCast("team_123"), .refresh_after_ms = 100, } }; var owned = try dupeWorkerEvent(alloc, source); defer freeWorkerEvent(alloc, owned);
const credential = &owned.credential_refreshed; try std.testing.expectEqualStrings("fresh-token", credential.token); try std.testing.expect(credential.token.ptr != source.credential_refreshed.token.ptr); try std.testing.expectEqualStrings("team_123", credential.team_id.?); try std.testing.expectEqual(@as(?i64, 100), credential.refresh_after_ms);}
test "pushEvent clones every semantic notice field into the queue allocator" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
var topic = [_]u8{ 'c', 'o', 'n', 't', 'e', 'x', 't' }; var body = [_]u8{ 'w', 'a', 'r', 'n', 'i', 'n', 'g' }; const topic_ptr = topic[0..].ptr; const body_ptr = body[0..].ptr; try runtime.pushEvent(alloc, .{ .semantic_notice = .{ .topic = &topic, .tone = .warning, .body = &body, .visibility = .full_only, } }); topic[0] = 'X'; body[0] = 'X';
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .semantic_notice); const notice = events.items[0].semantic_notice; try std.testing.expectEqualStrings("context", notice.topic); try std.testing.expectEqual(types.NoticeTone.warning, notice.tone); try std.testing.expectEqualStrings("warning", notice.body); try std.testing.expectEqual(types.NoticeVisibility.full_only, notice.visibility); try std.testing.expect(notice.topic.ptr != topic_ptr); try std.testing.expect(notice.body.ptr != body_ptr);}
fn checkSemanticWorkerEventDuplicationAllocationFailure(alloc: std.mem.Allocator) !void { const ordinary = try dupeWorkerEvent(alloc, .{ .semantic_notice = .{ .topic = "context", .tone = .warning, .body = "source limit reached", .visibility = .full_only, } }); defer freeWorkerEvent(alloc, ordinary);
const error_event = try dupeWorkerEvent(alloc, .{ .error_text = .{ .topic = "github", .tone = .@"error", .body = "publish failed", } }); defer freeWorkerEvent(alloc, error_event);
try std.testing.expectEqualStrings("context", ordinary.semantic_notice.topic); try std.testing.expectEqualStrings("source limit reached", ordinary.semantic_notice.body); try std.testing.expectEqualStrings("github", error_event.error_text.topic); try std.testing.expectEqualStrings("publish failed", error_event.error_text.body);}
test "ordinary and error semantic worker events free partial duplication allocations" { try std.testing.checkAllAllocationFailures( std.testing.allocator, checkSemanticWorkerEventDuplicationAllocationFailure, .{}, );}
test "prependOwnedEvents preserves retained order ahead of newer events" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.pushEvent(alloc, .{ .semantic_notice = .{ .topic = "system", .tone = .information, .body = "newer", } }); const retained = [_]WorkerEvent{ .{ .assistant_presentation = .{ .text = try alloc.dupe(u8, "first"), } }, try dupeWorkerEvent(alloc, .{ .semantic_notice = .{ .topic = "system", .tone = .information, .body = "second", } }), }; errdefer { for (retained) |event| freeWorkerEvent(alloc, event); } try runtime.prependOwnedEvents(alloc, &retained);
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 3), events.items.len); try std.testing.expectEqualStrings("first", events.items[0].assistant_presentation.text); try std.testing.expectEqualStrings("second", events.items[1].semantic_notice.body); try std.testing.expectEqualStrings("newer", events.items[2].semantic_notice.body);}
fn checkSemanticNoticeEnqueueAllocationFailure(alloc: std.mem.Allocator) !void { var runtime = WorkerRuntime{}; defer runtime.deinit(alloc);
try runtime.pushEvent(alloc, .{ .semantic_notice = .{ .topic = "permissions", .tone = .success, .body = "approval", .visibility = .full_only, } }); var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 1), events.items.len); try std.testing.expect(events.items[0] == .semantic_notice); try std.testing.expectEqualStrings("permissions", events.items[0].semantic_notice.topic); try std.testing.expectEqualStrings("approval", events.items[0].semantic_notice.body);}
test "semantic notice enqueue frees partial topic body and queue allocations" { try std.testing.checkAllAllocationFailures( std.testing.allocator, checkSemanticNoticeEnqueueAllocationFailure, .{}, );}
const PermissionThreadState = struct { decision: ?types.ToolPermissionDecision = null, err: ?anyerror = null,};
fn runPermissionRequest(state: *PermissionThreadState, runtime: *WorkerRuntime, label: []const u8) void { var response = runtime.requestPermissionBlocking( std.testing.allocator, .{ .label = label }, ) catch |err| { state.err = err; return; }; defer response.deinit(); state.decision = response.decision;}
fn submitPermissionDecisionForTest( runtime: *WorkerRuntime, request_id: u64, decision: types.ToolPermissionDecision,) PermissionSubmissionResult { return runtime.submitPermissionResponse( request_id, permission_request.OwnedPermissionResponse.init( std.testing.allocator, decision, null, ), );}
fn waitForPermissionLabel(worker: *WorkerRuntime, expected: []const u8) !u64 { var attempts: usize = 0; while (attempts < 100) : (attempts += 1) { var snapshot = try worker.snapshotState(std.testing.allocator); defer snapshot.deinit(std.testing.allocator); if (snapshot.pending_permission_request) |request| { try std.testing.expectEqualStrings(expected, request.label); try std.testing.expect(request.id != 0); return request.id; } io_mod.sleep(1 * std.time.ns_per_ms); } return error.TestExpectedEqual;}
test "permission blocking handles submit and stop" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true;
var state = PermissionThreadState{}; const thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &state, &runtime, "Allow?" }); const request_id = try waitForPermissionLabel(&runtime, "Allow?"); try std.testing.expectEqual( PermissionSubmissionResult.accepted, submitPermissionDecisionForTest(&runtime, request_id, .once), ); thread.join(); try std.testing.expect(state.err == null); try std.testing.expectEqual(types.ToolPermissionDecision.once, state.decision.?); var snapshot = try runtime.snapshotState(alloc); defer snapshot.deinit(alloc); try std.testing.expect(snapshot.pending_permission_request == null);
var deny_state = PermissionThreadState{}; const deny_thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &deny_state, &runtime, "Deny?" }); const deny_request_id = try waitForPermissionLabel(&runtime, "Deny?"); try std.testing.expectEqual( PermissionSubmissionResult.accepted, submitPermissionDecisionForTest(&runtime, deny_request_id, .deny), ); deny_thread.join(); try std.testing.expect(deny_state.err == null); try std.testing.expectEqual(types.ToolPermissionDecision.deny, deny_state.decision.?); var deny_snapshot = try runtime.snapshotState(alloc); defer deny_snapshot.deinit(alloc); try std.testing.expect(deny_snapshot.pending_permission_request == null);
var stop_runtime = WorkerRuntime{}; defer stop_runtime.deinit(alloc); stop_runtime.worker_processing = true; var stop_state = PermissionThreadState{}; const stop_thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &stop_state, &stop_runtime, "Stop?" }); _ = try waitForPermissionLabel(&stop_runtime, "Stop?"); stop_runtime.requestStop(); stop_thread.join(); try std.testing.expect(stop_state.err == null); try std.testing.expectEqual(types.ToolPermissionDecision.deny, stop_state.decision.?); var stop_snapshot = try stop_runtime.snapshotState(alloc); defer stop_snapshot.deinit(alloc); try std.testing.expect(stop_snapshot.pending_permission_request == null);
var shutdown_runtime = WorkerRuntime{}; defer shutdown_runtime.deinit(alloc); shutdown_runtime.worker_processing = true; var shutdown_state = PermissionThreadState{}; const shutdown_thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &shutdown_state, &shutdown_runtime, "Shutdown?" }); _ = try waitForPermissionLabel(&shutdown_runtime, "Shutdown?"); shutdown_runtime.requestShutdown(); shutdown_thread.join(); try std.testing.expect(shutdown_state.err == null); try std.testing.expectEqual(types.ToolPermissionDecision.deny, shutdown_state.decision.?); var shutdown_snapshot = try shutdown_runtime.snapshotState(alloc); defer shutdown_snapshot.deinit(alloc); try std.testing.expect(shutdown_snapshot.pending_permission_request == null);}
test "permission response accepts feedback ownership" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 44, .label = "feedback" }, );
const feedback = try alloc.dupe(u8, "summarize the result"); try std.testing.expectEqual( PermissionSubmissionResult.accepted, runtime.submitPermissionResponse( 44, permission_request.OwnedPermissionResponse.init(alloc, .once, feedback), ), ); try std.testing.expectEqualStrings( "summarize the result", runtime.pending_permission_response.?.feedback.?, );}
test "permission submission rejects stale ids and preserves first resolution" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 22, .label = "B" }, );
try std.testing.expectEqual( PermissionSubmissionResult.stale, submitPermissionDecisionForTest(&runtime, 21, .always), ); try std.testing.expect(runtime.pending_permission_response == null); try std.testing.expectEqual( PermissionSubmissionResult.accepted, submitPermissionDecisionForTest(&runtime, 22, .deny), ); try std.testing.expectEqual( types.ToolPermissionDecision.deny, runtime.pending_permission_response.?.decision, ); try std.testing.expectEqual( PermissionSubmissionResult.no_pending, submitPermissionDecisionForTest(&runtime, 22, .once), ); try std.testing.expectEqual( PermissionSubmissionResult.no_pending, submitPermissionDecisionForTest(&runtime, 22, .always), ); try std.testing.expectEqual( PermissionSubmissionResult.no_pending, submitPermissionDecisionForTest(&runtime, 22, .deny), ); try std.testing.expectEqual( types.ToolPermissionDecision.deny, runtime.pending_permission_response.?.decision, );
var snapshot = try runtime.snapshotState(alloc); defer snapshot.deinit(alloc); try std.testing.expect(snapshot.pending_permission_request == null);}
test "stale A decisions cannot resolve a newer B request" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true;
var a_state = PermissionThreadState{}; const a_thread = try std.Thread.spawn( .{}, runPermissionRequest, .{ &a_state, &runtime, "A" }, ); const a_id = try waitForPermissionLabel(&runtime, "A"); try std.testing.expectEqual( PermissionSubmissionResult.accepted, submitPermissionDecisionForTest(&runtime, a_id, .once), ); a_thread.join(); try std.testing.expectEqual(types.ToolPermissionDecision.once, a_state.decision.?);
var b_state = PermissionThreadState{}; const b_thread = try std.Thread.spawn( .{}, runPermissionRequest, .{ &b_state, &runtime, "B" }, ); const b_id = try waitForPermissionLabel(&runtime, "B"); try std.testing.expect(b_id != a_id);
const stale_decisions = [_]types.ToolPermissionDecision{ .once, .always, .deny, .deny, }; for (stale_decisions) |decision| { try std.testing.expectEqual( PermissionSubmissionResult.stale, submitPermissionDecisionForTest(&runtime, a_id, decision), ); } try std.testing.expect(runtime.pending_permission_response == null); try std.testing.expectEqual( PermissionSubmissionResult.accepted, submitPermissionDecisionForTest(&runtime, b_id, .always), ); b_thread.join(); try std.testing.expectEqual( types.ToolPermissionDecision.always, b_state.decision.?, );}
test "permission request ids never wrap or reuse zero" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.next_permission_request_id = std.math.maxInt(u64);
var last_state = PermissionThreadState{}; const last_thread = try std.Thread.spawn( .{}, runPermissionRequest, .{ &last_state, &runtime, "last id" }, ); const last_id = try waitForPermissionLabel(&runtime, "last id"); try std.testing.expectEqual(std.math.maxInt(u64), last_id); try std.testing.expectEqual( PermissionSubmissionResult.accepted, submitPermissionDecisionForTest(&runtime, last_id, .deny), ); last_thread.join(); try std.testing.expect(last_state.err == null);
var exhausted_state = PermissionThreadState{}; const exhausted_thread = try std.Thread.spawn( .{}, runPermissionRequest, .{ &exhausted_state, &runtime, "exhausted" }, ); exhausted_thread.join(); try std.testing.expectEqual( error.RequestIdExhausted, exhausted_state.err.?, );}
test "approval turn cancellation resolves the current worker request once" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 33, .label = "current" }, );
runtime.cancelApprovalTurn(); try std.testing.expect(runtime.isCancelRequested()); try std.testing.expectEqual( types.ToolPermissionDecision.deny, runtime.pending_permission_response.?.decision, ); try std.testing.expectEqual( PermissionSubmissionResult.no_pending, submitPermissionDecisionForTest(&runtime, 33, .always), );}
test "approval cancellation leaves admitted steering available for the next turn" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true; runtime.active_turn_id = 41; runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 33, .label = "current" }, ); try runtime.admitInteractivePrompt( alloc, try makePrompt(alloc, "continue with the correction", "model"), );
runtime.cancelApprovalTurn(); runtime.finishProcessing();
const next = (try runtime.tryTakeNextPrompt(alloc)) orelse return error.TestExpectedSteeringContinuation; defer freeQueuedPrompt(alloc, next); try std.testing.expect(next.delivery.isContinuation()); try std.testing.expectEqualStrings("continue with the correction", next.prompt);}
test "text enqueue allocation failure preserves active interactive state" { const alloc = std.testing.allocator; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); runtime.worker_processing = true;
runtime.pending_permission_waiting = true; runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe( alloc, .{ .id = 3, .label = "active permission" }, ); const question_options = [_]types.QuestionOption{.{ .label = "Wait", .description = null }}; const question_entries = [_]types.QuestionBatchEntry{.{ .question = "Keep waiting?", .options = &question_options, }}; runtime.pending_question_shared = try dupeOwnedQuestionBatch( alloc, &question_entries, .agent_question, );
const prompt = try makePrompt( alloc, "Stop", "model", ); var failing = std.testing.FailingAllocator.init( alloc, .{ .fail_index = 0 }, ); try std.testing.expectError( error.OutOfMemory, runtime.enqueuePrompt(failing.allocator(), prompt), ); freeQueuedPrompt(alloc, prompt);
try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); try std.testing.expect(!runtime.isCancelRequested()); try std.testing.expect(runtime.pending_permission_response == null); try std.testing.expect(runtime.pending_permission_request_shared != null); try std.testing.expectEqualStrings( "active permission", runtime.pending_permission_request_shared.?.label, ); try std.testing.expect(runtime.pending_question_shared != null); try std.testing.expect(runtime.pending_question_response == .pending);}
const QuestionThreadState = struct { answers: ?[][]u8 = null, err: ?anyerror = null,};
fn runQuestionRequest(state: *QuestionThreadState, runtime: *WorkerRuntime, entries: []const types.QuestionBatchEntry) void { state.answers = runtime.requestQuestionBatchAnswerBlocking(std.testing.allocator, entries) catch |err| { state.err = err; return; };}
fn runRouteRecoveryRequest(state: *QuestionThreadState, runtime: *WorkerRuntime, entries: []const types.QuestionBatchEntry) void { state.answers = runtime.requestRouteRecoveryAnswerBlocking(std.testing.allocator, entries) catch |err| { state.err = err; return; };}
fn waitForQuestionSnapshot(worker: *WorkerRuntime) !PendingQuestionBatchSnapshot { var attempts: usize = 0; while (attempts < 100) : (attempts += 1) { if (try worker.snapshotPendingQuestionBatch(std.testing.allocator)) |snapshot| return snapshot; io_mod.sleep(1 * std.time.ns_per_ms); } return error.TestExpectedEqual;}
fn freeAnswers(alloc: std.mem.Allocator, answers: ?[][]u8) void { if (answers) |labels| { for (labels) |label| alloc.free(label); alloc.free(labels); }}
test "question batch snapshot answer and cancellation" { const alloc = std.testing.allocator; const options = [_]types.QuestionOption{ .{ .label = "Yes", .description = "Do it" }, .{ .label = "No", .description = null }, }; const entries = [_]types.QuestionBatchEntry{.{ .question = "Continue?", .options = &options }};
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); var state = QuestionThreadState{}; const thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &state, &runtime, &entries }); var snapshot = try waitForQuestionSnapshot(&runtime); defer snapshot.deinit(alloc); try std.testing.expectEqualStrings("Continue?", snapshot.entries[0].question); try std.testing.expectEqualStrings("Yes", snapshot.entries[0].options[0].label); try std.testing.expect(snapshot.entries[0].question.ptr != entries[0].question.ptr); try runtime.submitQuestionBatchAnswer(alloc, &.{"Yes"}); thread.join(); defer freeAnswers(alloc, state.answers); try std.testing.expect(state.err == null); try std.testing.expectEqualStrings("Yes", state.answers.?[0]); try std.testing.expect(try runtime.snapshotPendingQuestionBatch(alloc) == null);
var cancel_runtime = WorkerRuntime{}; defer cancel_runtime.deinit(alloc); var cancel_state = QuestionThreadState{}; const cancel_thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &cancel_state, &cancel_runtime, &entries }); var cancel_snapshot = try waitForQuestionSnapshot(&cancel_runtime); cancel_snapshot.deinit(alloc); try cancel_runtime.submitQuestionBatchAnswer(alloc, null); cancel_thread.join(); try std.testing.expect(cancel_state.err == null); try std.testing.expect(cancel_state.answers == null); try std.testing.expect(try cancel_runtime.snapshotPendingQuestionBatch(alloc) == null);}
test "question batch source distinguishes route recovery from agent questions" { const alloc = std.testing.allocator; const options = [_]types.QuestionOption{.{ .label = "Try again later", .description = null }}; const entries = [_]types.QuestionBatchEntry{.{ .question = "Route failed. What should fx do?", .options = &options }};
var route_runtime = WorkerRuntime{}; defer route_runtime.deinit(alloc); var route_state = QuestionThreadState{}; const route_thread = try std.Thread.spawn(.{}, runRouteRecoveryRequest, .{ &route_state, &route_runtime, &entries }); var route_snapshot = try waitForQuestionSnapshot(&route_runtime); defer route_snapshot.deinit(alloc); try std.testing.expectEqual(QuestionPromptSource.route_recovery, route_snapshot.source); try std.testing.expectEqual(QuestionPromptSource.route_recovery, route_runtime.pendingQuestionBatchSource()); try route_runtime.submitQuestionBatchAnswer(alloc, &.{"Try again later"}); route_thread.join(); defer freeAnswers(alloc, route_state.answers); try std.testing.expect(route_state.err == null); try std.testing.expectEqualStrings("Try again later", route_state.answers.?[0]);
var agent_runtime = WorkerRuntime{}; defer agent_runtime.deinit(alloc); var agent_state = QuestionThreadState{}; const agent_thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &agent_state, &agent_runtime, &entries }); var agent_snapshot = try waitForQuestionSnapshot(&agent_runtime); defer agent_snapshot.deinit(alloc); try std.testing.expectEqual(QuestionPromptSource.agent_question, agent_snapshot.source); try std.testing.expectEqual(QuestionPromptSource.agent_question, agent_runtime.pendingQuestionBatchSource()); try agent_runtime.submitQuestionBatchAnswer(alloc, &.{"Try again later"}); agent_thread.join(); defer freeAnswers(alloc, agent_state.answers); try std.testing.expect(agent_state.err == null);}
test "question request queues an ordered boundary after prior assistant text" { const alloc = std.testing.allocator; const options = [_]types.QuestionOption{.{ .label = "Yes", .description = null }}; const entries = [_]types.QuestionBatchEntry{.{ .question = "Continue?", .options = &options }};
var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("before question"), } });
var state = QuestionThreadState{}; const thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &state, &runtime, &entries }); var snapshot = try waitForQuestionSnapshot(&runtime); defer snapshot.deinit(alloc);
var events = runtime.takeEvents(); defer freeEventList(alloc, &events); try std.testing.expectEqual(@as(usize, 2), events.items.len); try std.testing.expect(events.items[0] == .assistant_presentation); try std.testing.expect(events.items[0].assistant_presentation == .text); try std.testing.expectEqualStrings("before question", events.items[0].assistant_presentation.text); try std.testing.expect(events.items[1] == .question_requested);
try runtime.submitQuestionBatchAnswer(alloc, &.{"Yes"}); thread.join(); defer freeAnswers(alloc, state.answers); try std.testing.expect(state.err == null); try std.testing.expectEqualStrings("Yes", state.answers.?[0]);}
test "question request rolls back pending state when its boundary event cannot be queued" { const options = [_]types.QuestionOption{.{ .label = "Yes", .description = null }}; const entries = [_]types.QuestionBatchEntry{.{ .question = "Continue?", .options = &options }};
var probe = std.testing.FailingAllocator.init(std.testing.allocator, .{}); var probe_runtime = WorkerRuntime{}; probe_runtime.worker_stop_requested = true; const probe_result = try probe_runtime.requestQuestionBatchAnswerBlocking(probe.allocator(), &entries); try std.testing.expect(probe_result == null); const marker_append_alloc_index = probe.alloc_index - 1; probe_runtime.deinit(probe.allocator()); try std.testing.expectEqual(probe.allocated_bytes, probe.freed_bytes);
var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = marker_append_alloc_index, }); var runtime = WorkerRuntime{}; defer runtime.deinit(failing.allocator()); runtime.worker_stop_requested = true;
try std.testing.expectError( error.OutOfMemory, runtime.requestQuestionBatchAnswerBlocking(failing.allocator(), &entries), ); try std.testing.expect(runtime.pending_question_shared == null); try std.testing.expect(runtime.pending_question_response == .pending); try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);}
test "discarding queued recovery releases metadata without deleting saved images" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "saved.bin", .data = "snapshot" }); const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "saved.bin"); defer alloc.free(path); const images = [_]types.ImageAttachment{.{ .id = 1, .path = @constCast("/missing/source.png"), .media_type = @constCast("image/png"), .snapshot_path = path, .snapshot_sha256 = @constCast("a" ** 64), }}; var runtime = WorkerRuntime{}; defer runtime.deinit(alloc); var prompt = try makePrompt(alloc, "recover", "test/model"); var owned = true; defer if (owned) freeQueuedPrompt(alloc, prompt); prompt.images = try types.dupeImageAttachmentSlice(alloc, &images); prompt.recovery_checkpoint = try (session_codec.RecoveryCheckpoint{ .turn_id = 1, .user = .{ .text = @constCast("recover"), .images = @constCast(&images) }, .assistant_source = @constCast(""), .cause = .network_interrupted, .action = .retrying_request, .authority = .{ .provider = .gateway, .model = @constCast("test/model") }, .requested_fast_mode = false, .fast_mode = false, .max_provider_attempts = 10, .consumed_provider_attempts = 1, }).dupe(alloc); try runtime.enqueuePrompt(alloc, prompt); owned = false; runtime.clearQueuedPrompts(alloc, &.{}); try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount()); try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});}