Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
190 kB · 5068 lines
TypeScript
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613461446154616461746184619462046214622462346244625462646274628462946304631463246334634463546364637463846394640464146424643464446454646464746484649465046514652465346544655465646574658465946604661466246634664466546664667466846694670467146724673467446754676467746784679468046814682468346844685468646874688468946904691469246934694469546964697469846994700470147024703470447054706470747084709471047114712471347144715471647174718471947204721472247234724472547264727472847294730473147324733473447354736473747384739474047414742474347444745474647474748474947504751475247534754475547564757475847594760476147624763476447654766476747684769477047714772477347744775477647774778477947804781478247834784478547864787478847894790479147924793479447954796479747984799480048014802480348044805480648074808480948104811481248134814481548164817481848194820482148224823482448254826482748284829483048314832483348344835483648374838483948404841484248434844484548464847484848494850485148524853485448554856485748584859486048614862486348644865486648674868486948704871487248734874487548764877487848794880488148824883488448854886488748884889489048914892489348944895489648974898489949004901490249034904490549064907490849094910491149124913491449154916491749184919492049214922492349244925492649274928492949304931493249334934493549364937493849394940494149424943494449454946494749484949495049514952495349544955495649574958495949604961496249634964496549664967496849694970497149724973497449754976497749784979498049814982498349844985498649874988498949904991499249934994499549964997499849995000500150025003500450055006500750085009501050115012501350145015501650175018501950205021502250235024502550265027502850295030503150325033503450355036503750385039504050415042504350445045504650475048504950505051505250535054505550565057505850595060506150625063506450655066506750685069import { existsSync, mkdirSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";import { tmpdir } from "node:os";import { basename, dirname, join } from "node:path";import { setTimeout as sleep } from "node:timers/promises";import { Agent, type AgentMessage, type StreamFn } from "@earendil-works/pi-agent-core";import { type AssistantMessage, type Context, createAssistantMessageEventStream, getModel, type TextContent, type Usage,} from "@earendil-works/pi-ai";import { Type } from "typebox";import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";import { ENV_AGENT_DIR } from "../src/config.js";import { type AgentSessionMessageController, createAgentSessionMessage, isAgentSessionMessage,} from "../src/core/agent-messages.js";import { AgentSession, type RlmChildAgentSnapshot } from "../src/core/agent-session.js";import { AuthStorage } from "../src/core/auth-storage.js";import { computeOwnAndTotalUsage } from "../src/core/context-tree.js";import type { LoadExtensionsResult } from "../src/core/extensions/index.js";import { type HostRequestHandlers, ReplKernelManager } from "../src/core/kernel/index.js";import { ASYNC_BASH_COMPLETION_CUSTOM_TYPE, convertToLlm } from "../src/core/messages.js";import { ModelRegistry } from "../src/core/model-registry.js";import { type CreateRlmSubagentRuntimeOptions, createDefaultRlmSubagentSessionName, createRlmDeleteSubagentHostHandler, createRlmRunHostHandler, type SubagentRuntimeHost,} from "../src/core/rlm-runtime.js";import { readSessionInfo, SessionManager } from "../src/core/session-manager.js";import { SettingsManager, type SettingsStorage } from "../src/core/settings-manager.js";import type { Skill } from "../src/core/skills.js";import { createSyntheticSourceInfo } from "../src/core/source-info.js";import type { BashOperations } from "../src/core/tools/bash.js";import { type ActiveSessionState, resolveActiveSessionState } from "../src/modes/daemon/active-session-state.js";import { AgentDaemon } from "../src/modes/daemon/daemon-mode.js";import { getMessageText } from "./suite/harness.js";import { createTestExtensionsResult, createTestResourceLoader, writeTestRlmModels } from "./utilities.js";
const model = getModel("anthropic", "claude-sonnet-4-5")!;
/** * Spawn lifecycle is this file's subject, not routing: every child is born from one * role, and the permissive test policy makes any registered model eligible for it. */const TEST_MODEL = `${model.provider}/${model.id}`;const TEST_ROLE_MODELS = [TEST_MODEL];
function spawnChild( session: AgentSession, prompt: string, kwargs: Record<string, unknown> = {},): ReturnType<AgentSession["runRlmChild"]> { session.hasCalledRlmModels = true; return session.runRlmChild(prompt, { model: TEST_MODEL, ...kwargs });}
function userText(context: Context): string { const lastMessage = context.messages[context.messages.length - 1] as AgentMessage | undefined; if (!lastMessage) return ""; if (isAgentSessionMessage(lastMessage)) { return lastMessage.content.replace(/^\[task from parent\]\n\n/, ""); } if (lastMessage.role !== "user") return ""; if (typeof lastMessage.content === "string") { return lastMessage.content.replace(/^\[task from parent\]\n\n/, ""); } const text = lastMessage.content .filter((block): block is TextContent => block.type === "text") .map((block) => block.text) .join("\n"); return text.replace(/^\[task from parent\]\n\n/, "");}
function usage(input = 7, output = 3): Usage { return { input, output, cacheRead: 0, cacheWrite: 0, totalTokens: input + output, cost: { input, output, cacheRead: 0, cacheWrite: 0, total: input + output }, };}
function assistantMessage(text: string, messageUsage = usage()): AssistantMessage { return { role: "assistant", content: [{ type: "text", text }], api: model.api, provider: model.provider, model: model.id, usage: messageUsage, stopReason: "stop", timestamp: Date.now(), };}
function streamAnswer(text: string): ReturnType<typeof createAssistantMessageEventStream> { const stream = createAssistantMessageEventStream(); queueMicrotask(() => { const message = assistantMessage(text); stream.push({ type: "done", reason: "stop", message }); }); return stream;}
interface InspectableRlmRun { id: string; prompt?: string; sessionName?: string; sessionDir: string; model?: typeof model; abort: () => void; status: string; settled: boolean; error?: string; abandonedForQuiescence?: boolean; activity?: { kind: string }; progressNotes?: string[]; emitUpdate?: () => void; publication?: { promise: Promise<void>; resolve(): void; reject(error: Error): void }; settlement?: { promise: Promise<void>; resolve(): void; reject(error: Error): void }; detachedDeletion?: Awaited<ReturnType<AgentSession["listRlmSubagents"]>>["subagents"][number]; session?: AgentSession;}
interface InspectableRlmSession { _disposing: boolean; _activeRlmChildRuns: Map<string, InspectableRlmRun>; _unsettledRlmChildRuns: Set<InspectableRlmRun>; _deletingRlmChildren: Map< string, { subagent: Awaited<ReturnType<AgentSession["listRlmSubagents"]>>["subagents"][number]; promise: Promise<unknown>; } >; _rlmChildCleanupFailures: Map<string, Awaited<ReturnType<AgentSession["listRlmSubagents"]>>["subagents"][number]>; _rlmChildSessions: Map<string, { session: AgentSession; run?: InspectableRlmRun }>; _rlmChildUnsubscribes: Map<string, () => void>; _deletedRlmChildIds: Set<string>; _rlmQuiescenceWaitAborts: Set<AbortController>; _createKernelHostHandlers(): HostRequestHandlers; _reapDeletedRlmSubagentRuntimesAfterCompaction(): Promise<void>;}
async function waitFor(condition: () => boolean): Promise<void> { const deadline = Date.now() + 1000; while (!condition()) { if (Date.now() > deadline) { throw new Error("Timed out waiting for condition"); } await sleep(10); }}
function findLastMessage( messages: readonly AgentMessage[], predicate: (message: AgentMessage) => boolean,): AgentMessage | undefined { for (let i = messages.length - 1; i >= 0; i--) { const message = messages[i]; if (message && predicate(message)) return message; } return undefined;}
function deferred<T = void>(): { promise: Promise<T>; resolve: (value: T | PromiseLike<T>) => void; reject: (reason?: unknown) => void;} { let resolve!: (value: T | PromiseLike<T>) => void; let reject!: (reason?: unknown) => void; const promise = new Promise<T>((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; }); return { promise, resolve, reject };}
describe("AgentSession rlm recursion", () => { let tempDir: string; let session: AgentSession | undefined; let previousAgentDir: string | undefined; let previousRlmDepth: string | undefined; let previousRlmMaxDepth: string | undefined;
beforeEach(() => { tempDir = join(tmpdir(), `pi-rlm-recursion-${Date.now()}-${Math.random().toString(36).slice(2)}`); mkdirSync(tempDir, { recursive: true }); // Children spawned by production code resolve roles from the agent dir, so the // fixture policy must sit where they look; otherwise they read (and create) the // developer's real rlm-models.toml. previousAgentDir = process.env[ENV_AGENT_DIR]; process.env[ENV_AGENT_DIR] = tempDir; previousRlmDepth = process.env.RLM_DEPTH; delete process.env.RLM_DEPTH; previousRlmMaxDepth = process.env.RLM_MAX_DEPTH; delete process.env.RLM_MAX_DEPTH; });
afterEach(() => { session?.dispose(); session = undefined; rmSync(tempDir, { recursive: true, force: true }); if (previousAgentDir === undefined) { delete process.env[ENV_AGENT_DIR]; } else { process.env[ENV_AGENT_DIR] = previousAgentDir; } if (previousRlmDepth === undefined) { delete process.env.RLM_DEPTH; } else { process.env.RLM_DEPTH = previousRlmDepth; } if (previousRlmMaxDepth === undefined) { delete process.env.RLM_MAX_DEPTH; } else { process.env.RLM_MAX_DEPTH = previousRlmMaxDepth; } });
function createSession( options: { depth?: number; maxDepth?: number; streamFn?: StreamFn; agentMessageController?: AgentSessionMessageController; subagentRuntimeHost?: SubagentRuntimeHost; customTools?: ConstructorParameters<typeof AgentSession>[0]["customTools"]; rlmSessionDir?: string; sessionManager?: SessionManager; settingsManager?: SettingsManager; extensionsResult?: LoadExtensionsResult; remoteHost?: string; } = {}, ): AgentSession { const authStorage = AuthStorage.create(join(tempDir, "auth.json")); authStorage.setRuntimeApiKey("anthropic", "test-key"); const sessionManager = options.sessionManager ?? SessionManager.create(tempDir, join(tempDir, "sessions")); const settingsManager = options.settingsManager ?? SettingsManager.create(tempDir, tempDir);
const agent = new Agent({ convertToLlm, getApiKey: () => "test-key", initialState: { model, systemPrompt: "", tools: [], thinkingLevel: "off", }, streamFn: options.streamFn ?? ((_model, context) => streamAnswer(`child answer: ${userText(context)}`)), });
session = new AgentSession({ agent, sessionManager, settingsManager, cwd: tempDir, modelRegistry: ModelRegistry.create(authStorage, join(tempDir, "models.json")), rlmModelsFilePath: writeTestRlmModels(tempDir, TEST_ROLE_MODELS), resourceLoader: createTestResourceLoader({ extensionsResult: options.extensionsResult, skills: options.agentMessageController ? [ { name: "agent-message", description: "test", filePath: join(tempDir, "SKILL.md"), baseDir: tempDir, sourceInfo: createSyntheticSourceInfo(join(tempDir, "SKILL.md"), { source: "test" }), disableModelInvocation: false, kind: "python", python: { importName: "agent_message", packagePath: tempDir, pyprojectPath: join(tempDir, "pyproject.toml"), }, }, ] : undefined, }), agentMessageController: options.agentMessageController, subagentRuntimeHost: options.subagentRuntimeHost, customTools: options.customTools, rlmDepth: options.depth, rlmMaxDepth: options.maxDepth, rlmSessionDir: options.rlmSessionDir, remoteHost: options.remoteHost, }); return session; }
/** Session with a zero-usage parent assistant whose child answers a tool loop: usages[i] per request, tool calls until the last, then stop. */ function createToolLoopSession(requests: number, usages: Usage[], onRequest?: (toolResultCount: number) => void) { const tool = { name: "echo", description: "Echo a value", label: "echo", parameters: Type.Object({ value: Type.String() }), execute: async (_toolCallId: string, params: { value: string }) => ({ content: [{ type: "text" as const, text: params.value }], details: {}, }), }; const root = createSession({ customTools: [tool], streamFn: (_model, context) => { const toolResultCount = context.messages.filter((message) => message.role === "toolResult").length; onRequest?.(toolResultCount); const last = toolResultCount >= requests - 1; const stream = createAssistantMessageEventStream(); queueMicrotask(() => { const message = last ? assistantMessage("done", usages[toolResultCount]) : { ...assistantMessage("", usages[toolResultCount]), content: [ { type: "toolCall" as const, id: `echo-${toolResultCount}`, name: "echo", arguments: { value: "ok" }, }, ], stopReason: "toolUse" as const, }; stream.push({ type: "done", reason: last ? "stop" : "toolUse", message }); }); return stream; }, }); const parentAssistant = assistantMessage("running ipython", usage(0, 0)); root.agent.state.messages.push(parentAssistant); root.sessionManager.appendMessage(parentAssistant); return root; }
function createAbortInsensitiveChild(): { child: AgentSession; completion: ReturnType<typeof deferred<void>>; hasStarted: () => boolean; } { const completion = deferred<void>(); let started = false; const child = createSession({ streamFn: () => { const stream = createAssistantMessageEventStream(); started = true; void completion.promise.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("child stopped") }); }); return stream; }, }); vi.spyOn(child, "abort").mockResolvedValue(); return { child, completion, hasStarted: () => started }; }
it("propagates skipped-running deletion outcomes through the host handler", async () => { const subagent = { rlm_child_id: "running-child", active_session_id: "running-session", session_id: "running-session", session_name: "running-worker", session_dir: join(tempDir, "running-child"), status: "running" as const, }; const deleteHandler = createRlmDeleteSubagentHostHandler(async () => ({ subagent, outcome: "skipped_running", }));
await expect(deleteHandler({ target: subagent.rlm_child_id })).resolves.toEqual({ subagent, outcome: "skipped_running", }); });
it("persists RLM_DEPTH for a fresh session and reports the seeded depth", () => { vi.stubEnv("RLM_DEPTH", "1"); try { const fresh = createSession({ maxDepth: 2 }); fresh.sessionManager.flushNow(); if (!fresh.sessionFile) throw new Error("Missing fresh session file");
const header = JSON.parse(readFileSync(fresh.sessionFile, "utf8").split("\n")[0] ?? "{}"); expect(header.rlmDepth).toBe(1); expect(fresh.rlmDepth).toBe(1); } finally { vi.unstubAllEnvs(); } });
it("prefers persisted depth over RLM_DEPTH when resuming a session", () => { const persistedManager = SessionManager.create(tempDir, join(tempDir, "resumed-sessions")); persistedManager.newSession({ rlmDepth: 2 }); persistedManager.flushNow(); vi.stubEnv("RLM_DEPTH", "1"); try { const resumed = createSession({ maxDepth: 3, sessionManager: persistedManager }); expect(resumed.rlmDepth).toBe(2); } finally { vi.unstubAllEnvs(); } });
it.each([-1, "0"])("ignores invalid persisted RLM depth %j", (invalidDepth) => { const persistedManager = SessionManager.create(tempDir, join(tempDir, "invalid-depth-sessions")); persistedManager.newSession({ rlmDepth: 2 }); persistedManager.flushNow(); const sessionFile = persistedManager.getSessionFile(); if (!sessionFile) throw new Error("Missing persisted session file"); const lines = readFileSync(sessionFile, "utf8").split("\n"); lines[0] = JSON.stringify({ ...JSON.parse(lines[0] ?? "{}"), rlmDepth: invalidDepth }); writeFileSync(sessionFile, lines.join("\n")); const reopened = SessionManager.open(sessionFile, join(tempDir, "invalid-depth-sessions")); vi.stubEnv("RLM_DEPTH", "1"); try { expect(createSession({ maxDepth: 2, sessionManager: reopened }).rlmDepth).toBe(1); } finally { vi.unstubAllEnvs(); } });
it("keeps a forked root at depth zero so recursion remains allowed", async () => { const source = SessionManager.create(tempDir, join(tempDir, "source-sessions")); source.newSession({ rlmDepth: 0 }); source.appendMessage({ role: "user", content: "source prompt", timestamp: 1 }); source.flushNow(); const sourceFile = source.getSessionFile(); if (!sourceFile) throw new Error("Missing source session file"); const forkedManager = SessionManager.forkFrom(sourceFile, tempDir, join(tempDir, "forked-sessions")); const forked = createSession({ maxDepth: 1, sessionManager: forkedManager });
expect(forked.rlmDepth).toBe(0); const spawned = await spawnChild(forked, "recursion remains available"); expect(spawned.rlm_child_id).toMatch(/^sub-/); await waitFor(() => forked.getRlmChildSession(spawned.rlm_child_id)?.getLastAssistantText() !== undefined); });
it("creates readable collision-resistant default subagent session names", () => { expect(createDefaultRlmSubagentSessionName("Summarize the HTTP API!", "sub-a1b2c3d4")).toBe( "subagent-summarize-the-http-api-a1b2c3d4", ); expect(createDefaultRlmSubagentSessionName("Summarize the HTTP API!", "sub-eeeeffff")).not.toBe( createDefaultRlmSubagentSessionName("Summarize the HTTP API!", "sub-a1b2c3d4"), ); expect(createDefaultRlmSubagentSessionName("same task", "sub-aBcDeFgH")).not.toBe( createDefaultRlmSubagentSessionName("same task", "sub-AbCdEfGh"), ); expect(createDefaultRlmSubagentSessionName("x".repeat(200), "sub-a1b2c3d4")).toHaveLength(64); });
it("persists the spawned child's parent edge and derived runtime depth in its header", async () => { const root = createSession({ depth: 2, maxDepth: 4 }); const result = await spawnChild(root, "persist my tree position"); if (!result.session_dir) throw new Error("Missing child session directory"); const child = root.getRlmChildSession(basename(result.session_dir)); if (!child?.sessionFile || !root.sessionFile) throw new Error("Missing persisted session paths");
const header = JSON.parse(readFileSync(child.sessionFile, "utf8").split("\n")[0] ?? "{}"); expect(header).toMatchObject({ parentSession: root.sessionFile, rlmDepth: 3 }); expect(child.rlmDepth).toBe(3); });
it("lets the orchestrator choose a unique subagent session name", async () => { const root = createSession(); const result = await spawnChild(root, "inspect the API", { name: " api-reviewer " }); if (!result.session_dir) { throw new Error("Missing child session directory"); } const childId = basename(result.session_dir); const childSession = root.getRlmChildSession(childId); if (!childSession) { throw new Error("Missing retained child session"); } expect(childSession.sessionName).toBe("api-reviewer"); expect((await root.listRlmSubagents()).subagents[0]?.session_name).toBe("api-reviewer");
await expect(spawnChild(root, "inspect another API", { name: "api-reviewer" })).rejects.toThrow( 'Agent name "api-reviewer" is unavailable: an agent of that name already exists at depth 1 under this parent', ); await expect(spawnChild(root, "invalid name", { name: " " })).rejects.toThrow( "rlm.spawn name must not be empty", ); await expect(spawnChild(root, "reserved name", { name: "all" })).rejects.toThrow( "Broadcast agent messaging is not supported", ); });
it("falls back to listed family metadata when a controller lacks name validation", async () => { const listAgents = vi.fn(() => ({ current: { activeSessionId: "parent-active", sessionId: "unrelated-current" }, agents: [ { activeSessionId: "passive-active", sessionId: "passive-session", sessionName: "passive-worker", runtimeKind: "subagent" as const, cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentSessionId: "parent-session", parentSessionPath: parentPath, rlmDepth: 1, status: "idle" as const, }, { activeSessionId: "other-active", sessionId: "other-session", sessionName: "other-family-worker", runtimeKind: "subagent" as const, cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentSessionId: "other-parent", rlmDepth: 1, }, ], })); const manager = SessionManager.create(tempDir, join(tempDir, "sessions")); manager.newSession({ id: "parent-session" }); const parentPath = manager.getSessionFile(); if (!parentPath) throw new Error("Missing parent session path"); const root = createSession({ sessionManager: manager, agentMessageController: { listAgents, sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, });
await expect(spawnChild(root, "duplicate passive", { name: "passive-worker" })).rejects.toThrow( 'Agent name "passive-worker" is unavailable', ); await expect(spawnChild(root, "different family", { name: "other-family-worker" })).resolves.toMatchObject({ name: "other-family-worker", }); expect(listAgents).toHaveBeenCalled(); });
it("throws loud ambiguity when a child name equals a sibling session id for send and delete", async () => { const first = createSession({ rlmSessionDir: join(tempDir, "first") }); const second = createSession({ rlmSessionDir: join(tempDir, "second") }); second.setSessionName(first.sessionId); const root = createSession(); expect(root.registerRlmChildSession("first-child", first)).toBe(true); expect(root.registerRlmChildSession("second-child", second)).toBe(true); const states = new Map<string, ActiveSessionState>([ [ "first-active", { activeSessionId: "first-active", runtime: { session: first }, } as ActiveSessionState, ], [ "second-active", { activeSessionId: "second-active", runtime: { session: second }, } as ActiveSessionState, ], ]);
expect(() => resolveActiveSessionState(states, first.sessionId)).toThrow("Ambiguous active session"); await expect(root.deleteRlmSubagent(first.sessionId)).rejects.toThrow("is ambiguous"); });
it("reserves a running child's current name after it is renamed", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${userText(context)}`), }); }); return stream; }, }); const runPromise = spawnChild(root, "rename while running", { name: "spawn-worker" }); await waitFor(() => childStarted); const running = (await root.listRlmSubagents()).subagents[0]; if (!running) { throw new Error("Missing running child"); } if (!running.session_id) { throw new Error("Missing running child session ID"); } root.getRlmChildSession(running.rlm_child_id)?.setSessionName("renamed-running-worker"); expect((await root.listRlmSubagents()).subagents[0]?.session_name).toBe("renamed-running-worker");
await expect(spawnChild(root, "reuse renamed selector", { name: "renamed-running-worker" })).rejects.toThrow( 'Agent name "renamed-running-worker" is unavailable: an agent of that name already exists at depth 1 under this parent', ); releaseChild(); await runPromise; });
it("makes an externally restored retained child listable and deletable", async () => { const childId = "restored-child"; const childDir = join(tempDir, childId); mkdirSync(childDir, { recursive: true }); const child = createSession({ rlmSessionDir: childDir }); child.setSessionName("restored-worker"); const restoredAnswer = assistantMessage("restored answer", usage(7, 3)); restoredAnswer.content.push({ type: "toolCall", id: "tool-1", name: "ipython", arguments: {} }); child.agent.state.messages.push(restoredAnswer); child.setCurrentRecap("restored recap"); const disposeChild = vi.spyOn(child, "disposeAsync"); const root = createSession(); const childStatuses: string[] = []; root.subscribe((event) => { if (event.type === "rlm_child_update" && event.child.id === childId) { childStatuses.push(event.child.status); } });
expect(root.registerRlmChildSession(childId, child)).toBe(true); expect(root.getRlmChildSnapshots()).toEqual([ expect.objectContaining({ id: childId, answerPreview: "restored answer", toolUseCount: 1, tokenCount: 10, recap: "restored recap", }), ]); expect((await root.listRlmSubagents()).subagents).toEqual([ expect.objectContaining({ rlm_child_id: childId, session_name: "restored-worker", status: "completed", }), ]);
await expect(root.deleteRlmSubagent("restored-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: childId, session_name: "restored-worker" }, }); expect(disposeChild).toHaveBeenCalledOnce(); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); expect(childStatuses).toEqual(["cancelled"]); });
it("projects live follow-up activity for a restored session-only child", async () => { const childId = "restored-followup-child"; const childDir = join(tempDir, childId); mkdirSync(childDir, { recursive: true }); const followUpGate = deferred<void>(); let followUpStarted = false; const child = createSession({ rlmSessionDir: childDir, streamFn: () => { const stream = createAssistantMessageEventStream(); followUpStarted = true; void followUpGate.promise.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("follow-up answer") }); }); return stream; }, }); const root = createSession(); expect(root.registerRlmChildSession(childId, child)).toBe(true);
const followUp = child.prompt("follow-up work"); await waitFor(() => followUpStarted); const busy = root.getRlmChildSnapshots(); expect(busy).toEqual([expect.objectContaining({ id: childId, status: "done" })]); expect(busy[0]?.activity).toBeDefined();
followUpGate.resolve(); await followUp; await child.waitForIdle(); const idle = root.getRlmChildSnapshots(); expect(idle).toEqual([expect.objectContaining({ id: childId, status: "done" })]); expect(idle[0]?.activity).toBeUndefined(); });
it("retries and releases failed retained child cleanup on the next compaction", async () => { const childId = "retained-retry-child"; const childDir = join(tempDir, childId); mkdirSync(childDir, { recursive: true }); const child = createSession({ rlmSessionDir: childDir }); child.setSessionName("retained-retry-worker"); let deleteAttempts = 0; const deleteRuntime = vi.fn(async (_childId: string, session: AgentSession) => { deleteAttempts++; if (deleteAttempts === 1) { throw new Error("retained close failed"); } await session.disposeAsync(); }); const settingsManager = SettingsManager.create(tempDir, tempDir); settingsManager.applyOverrides({ compaction: { keepRecentTokens: 1 } }); const extensionsResult = await createTestExtensionsResult([ (pi) => { pi.on("session_before_compact", async (event) => ({ compaction: { summary: "cleanup retry compaction", firstKeptEntryId: event.preparation.firstKeptEntryId, tokensBefore: event.preparation.tokensBefore, details: { source: "extension" }, }, })); }, ]); const root = createSession({ settingsManager, extensionsResult, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: deleteRuntime, }, }); root.sessionManager.appendMessage({ role: "user", content: "history before cleanup", timestamp: Date.now() }); root.sessionManager.appendMessage(assistantMessage("history response")); expect(root.registerRlmChildSession(childId, child)).toBe(true);
await expect(root.deleteRlmSubagent("retained-retry-worker")).rejects.toThrow("retained close failed"); const internals = root as unknown as InspectableRlmSession; expect(internals._rlmChildCleanupFailures.size).toBe(1);
await root.compact();
expect(deleteRuntime).toHaveBeenCalledTimes(2); expect(internals._rlmChildCleanupFailures.size).toBe(0); expect(internals._rlmChildSessions.size).toBe(0); await expect(spawnChild(root, "replacement", { name: "retained-retry-worker" })).resolves.toMatchObject({ name: "retained-retry-worker", }); });
it("keeps live nested children visible when their direct parent is hidden by deletion", () => { const root = createSession(); const rootInternals = root as unknown as InspectableRlmSession; const hiddenParents = [ { id: "deleted-parent", hiding: "deleted" as const }, { id: "deleting-parent", hiding: "deleting" as const }, { id: "detached-parent", hiding: "detached" as const }, ]; const parentInternalsToClear: InspectableRlmSession[] = [];
for (const [index, { id, hiding }] of hiddenParents.entries()) { const parent = createSession({ rlmSessionDir: join(tempDir, id) }); const parentInternals = parent as unknown as InspectableRlmSession; parentInternalsToClear.push(parentInternals); const nestedId = `${id}-live-grandchild`; parentInternals._activeRlmChildRuns.set(nestedId, { id: nestedId, prompt: "still working", sessionName: nestedId, sessionDir: join(tempDir, nestedId), model, abort: () => {}, status: index === 1 ? "queued" : "running", settled: false, progressNotes: [], });
if (hiding === "detached") { rootInternals._activeRlmChildRuns.set(id, { id, prompt: "hidden parent", sessionName: id, sessionDir: join(tempDir, id), model, abort: () => {}, status: "cancelled", settled: false, progressNotes: [], detachedDeletion: { rlm_child_id: id, active_session_id: null, session_id: null, session_name: id, session_dir: join(tempDir, id), status: "running", }, session: parent, }); // The same child can be visible in both lifecycle registries while // deletion settles; it must be traversed exactly once and remain hidden. rootInternals._rlmChildSessions.set(id, { session: parent }); } else { rootInternals._rlmChildSessions.set(id, { session: parent }); if (hiding === "deleted") { rootInternals._deletedRlmChildIds.add(id); } else { rootInternals._deletingRlmChildren.set(id, { subagent: { rlm_child_id: id, active_session_id: null, session_id: null, session_name: id, session_dir: join(tempDir, id), status: "completed", }, promise: Promise.resolve({ subagent: { rlm_child_id: id, active_session_id: null, session_id: null, session_name: id, session_dir: join(tempDir, id), status: "completed", }, }), }); } } }
const snapshots = root.getRlmChildSnapshots(); expect(snapshots.map((snapshot) => snapshot.id).sort()).toEqual( hiddenParents.map(({ id }) => `${id}-live-grandchild`).sort(), ); expect(snapshots.map((snapshot) => snapshot.status).sort()).toEqual(["queued", "running", "running"]); // These are deliberately minimal lifecycle records; remove them before // fixture teardown asks real runs to settle. rootInternals._activeRlmChildRuns.clear(); rootInternals._rlmChildSessions.clear(); for (const parentInternals of parentInternalsToClear) parentInternals._activeRlmChildRuns.clear(); root.dispose(); });
it("makes an orchestrator-chosen name override a custom runtime's preexisting name", async () => { const hostedChild = createSession(); hostedChild.setSessionName("factory-assigned-name"); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: async () => {}, }, });
await spawnChild(root, "inspect custom runtime", { name: "orchestrator-name" });
expect(hostedChild.sessionName).toBe("orchestrator-name"); });
it("runs a child session under a sub directory and returns an RLM-shaped result", async () => { const root = createSession({ agentMessageController: { listAgents: () => ({ current: { activeSessionId: "root-active", sessionId: "root-session" }, agents: [], }), sendAgentMessage: vi.fn(), }, }); const childUpdates: Array<{ status: string; label: string; answerPreview?: string; tokenCount?: number; toolUseCount?: number; }> = []; root.subscribe((event) => { if (event.type === "rlm_child_update") { childUpdates.push(event.child); } });
const result = await spawnChild(root, "summarize shard 1");
expect(result.rlm_child_id).toBe(basename(result.session_dir)); expect(result.session_dir).not.toBeNull(); expect(basename(result.session_dir!)).toMatch(/^sub-/); expect(dirname(result.session_dir!)).toBe(root.sessionManager.getSessionArtifactDir()); expect(existsSync(result.session_dir!)).toBe(true); await waitFor(() => readdirSync(result.session_dir).some((name) => name.endsWith(".jsonl"))); expect(childUpdates[0]?.status).toBe("queued"); expect(childUpdates[0]?.label).toBe("summarize shard 1"); await waitFor(() => childUpdates.some((update) => update.status === "done")); const doneUpdate = [...childUpdates].reverse().find((update) => update.status === "done"); expect(doneUpdate?.answerPreview).toBe("child answer: summarize shard 1"); expect(root.getRlmChildSnapshots()).toEqual([ expect.objectContaining({ id: result.rlm_child_id, status: "done", answerPreview: doneUpdate?.answerPreview, durationMs: expect.any(Number), }), ]); const child = root.getRlmChildSession(result.rlm_child_id); expect(getMessageText(child?.messages[0])).toContain( "The persistent memories produced across this session so far:", ); expect(child?.messages[1]).toMatchObject({ role: "custom", customType: "agent_message", content: "[task from parent]\n\nsummarize shard 1", display: true, details: { id: `spawn:${result.rlm_child_id}`, message: "summarize shard 1", from: { sessionId: root.sessionId, activeSessionId: "root-active" }, fromRelationship: "parent", }, }); // Context tokens from the child's own assistant usage (input 7 + output 3); no tools ran. expect(doneUpdate?.tokenCount).toBe(10); expect(doneUpdate?.toolUseCount).toBeUndefined(); });
it("wakes the agent with a follow-up when a detached bash handle completes", async () => { const prompts: string[] = []; const root = createSession({ streamFn: (_model, context) => { prompts.push(userText(context)); return streamAnswer("checked shell result"); }, }); const handlers = (root as unknown as InspectableRlmSession)._createKernelHostHandlers(); const completed = handlers["bash.completed"]; if (!completed) throw new Error("Missing bash.completed host handler");
await expect(completed({ pid: 42, command: "npm test", exitCode: 1 })).resolves.toEqual({}); await root.waitForIdle();
expect(prompts).toEqual(['[bash-done pid:42 exit:1]\n\nCommand: "npm test"']); expect(root.messages).toContainEqual( expect.objectContaining({ role: "custom", customType: ASYNC_BASH_COMPLETION_CUSTOM_TYPE, details: { pid: 42, command: "npm test", exitCode: 1 }, }), ); });
it.each([130, 137, 143, -2, -9, -15])("delivers signal exit %i on bash.completed", async (exitCode) => { let providerRuns = 0; const root = createSession({ streamFn: () => { providerRuns += 1; return streamAnswer("turn handled"); }, }); const handlers = (root as unknown as InspectableRlmSession)._createKernelHostHandlers(); await handlers["bash.completed"]!({ pid: 5001, command: "remote training", exitCode }); await root.waitForIdle(); expect(providerRuns).toBe(1); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === ASYNC_BASH_COMPLETION_CUSTOM_TYPE, ), ).toHaveLength(1); });
it("marks an in-cell roled send to the parent as replied", async () => { const sendAgentMessage = vi.fn(async () => ({ id: "agentmsg-reply", source: "agent_message" as const, target: { activeSessionId: "parent-active", sessionId: "parent-session" }, message: "done", deliveryStatus: "delivered" as const, })); const child = createSession({ depth: 1, agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [ { relationship: "parent", entry: { id: "parent-session", name: "parent", depth: 0, status: "idle" } }, ], sendAgentMessage, }, }); const handlers = (child as unknown as InspectableRlmSession)._createKernelHostHandlers(); const send = handlers["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
expect(child.repliedToParentSinceTask).toBe(false); await expect(send({ message: "done", receiver_role: "parent" })).resolves.toMatchObject({ message: "done", }); expect(sendAgentMessage).toHaveBeenCalledWith( expect.objectContaining({ target: "parent-session", message: "done" }), ); expect(child.repliedToParentSinceTask).toBe(true); });
it("resolves a spawned child handle to its published session before a roled send", async () => { let publishChild: (() => void) | undefined; const publicationGate = new Promise<void>((resolve) => { publishChild = resolve; }); let publishedChild: AgentSession | undefined; const sendAgentMessage = vi.fn(async (input: { target: string; message: string }) => ({ id: "agentmsg-child", source: "agent_message" as const, target: { activeSessionId: "child-active", sessionId: input.target }, message: input.message, deliveryStatus: "delivered" as const, })); const family = vi.fn(async () => publishedChild ? [ { relationship: "child" as const, entry: { id: publishedChild.sessionId, name: publishedChild.sessionName ?? publishedChild.sessionId, depth: 1, status: "running" as const, }, }, ] : [], ); const root = createSession({ agentMessageController: { listAgents: () => ({ agents: [] }), family, sendAgentMessage, }, subagentRuntimeHost: { createRlmSubagentRuntime: async (options) => { await publicationGate; const child = createSession({ rlmSessionDir: options.sessionDir }); child.setSessionName(options.sessionName); publishedChild = child; options.onSessionPublished?.(child); return { session: child }; }, deleteRlmSubagentRuntime: async (_id, child) => child?.disposeAsync(), }, }); const spawned = await spawnChild(root, "pending task", { name: "pending-child" }); expect(root.getRlmChildSnapshots()).toEqual([ expect.objectContaining({ id: spawned.rlm_child_id, status: "queued" }), ]); const handlers = (root as unknown as InspectableRlmSession)._createKernelHostHandlers(); const send = handlers["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
const pendingSend = send({ message: "hello", receiver_role: "child", receiver_name: spawned.rlm_child_id, }); await sleep(0); expect(family).not.toHaveBeenCalled(); expect(sendAgentMessage).not.toHaveBeenCalled(); publishChild?.();
await expect(pendingSend).resolves.toMatchObject({ message: "hello" }); expect(family).toHaveBeenCalledTimes(1); expect(sendAgentMessage).toHaveBeenCalledWith( expect.objectContaining({ target: publishedChild?.sessionId, message: "hello" }), ); });
it("delivers an id-addressed send after a completed child durably admits its terminal notice", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "completed-child") }); const sendAgentMessage = vi.fn(async (input: { target: string; message: string }) => ({ id: "agentmsg-completed-child", source: "agent_message" as const, target: { activeSessionId: "child-active", sessionId: input.target }, message: input.message, deliveryStatus: "delivered" as const, })); const root = createSession({ agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [ { relationship: "child" as const, entry: { id: child.sessionId, name: child.sessionName ?? child.sessionId, depth: 1, status: "idle" as const, }, }, ], sendAgentMessage, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async (_id, session) => session?.disposeAsync(), }, }); const spawned = await spawnChild(root, "completed task", { name: "completed-worker" }); await waitFor(() => root.getRlmChildSession(spawned.rlm_child_id) === child); const internals = root as unknown as InspectableRlmSession; const send = internals._createKernelHostHandlers()["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
await expect( send({ message: "follow-up", receiver_role: "child", receiver_name: spawned.rlm_child_id }), ).resolves.toMatchObject({ message: "follow-up" }); expect(sendAgentMessage).toHaveBeenCalledWith( expect.objectContaining({ target: child.sessionId, message: "follow-up" }), ); });
it("propagates pending child startup failure to an immediate roled send", async () => { let rejectStartup: ((error: Error) => void) | undefined; const startupGate = new Promise<never>((_resolve, reject) => { rejectStartup = reject; }); const root = createSession({ agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [], sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: () => startupGate, deleteRlmSubagentRuntime: async () => {}, }, }); const spawned = await spawnChild(root, "pending task", { name: "failing-child" }); const handlers = (root as unknown as InspectableRlmSession)._createKernelHostHandlers(); const send = handlers["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
const pendingSend = send({ message: "hello", receiver_role: "child", receiver_name: spawned.name }); rejectStartup?.(new Error("child startup failed"));
await expect(pendingSend).rejects.toThrow("child startup failed"); });
it("falls through a retained failed child name to a healthy roster child", async () => { let releaseFailureInjection: () => void = () => {}; const failureInjectionGate = new Promise<void>((resolve) => { releaseFailureInjection = resolve; }); const sendAgentMessage = vi.fn(async (input: { target: string; message: string }) => ({ id: "agentmsg-healthy-child", source: "agent_message" as const, target: { activeSessionId: "healthy-active", sessionId: input.target }, message: input.message, deliveryStatus: "delivered" as const, })); const root = createSession({ agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [ { relationship: "child" as const, entry: { id: "healthy-child-session", name: "shared-child", depth: 1, status: "idle" as const }, }, ], sendAgentMessage, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("child startup failed"); }, deleteRlmSubagentRuntime: async () => {}, }, }); const promptInjectedMessage = vi.fn(async () => failureInjectionGate); (root as unknown as { _promptInjectedMessage: typeof promptInjectedMessage })._promptInjectedMessage = promptInjectedMessage; await spawnChild(root, "failing task", { name: "shared-child" }); await waitFor(() => [...(root as unknown as InspectableRlmSession)._activeRlmChildRuns.values()].some( (run) => run.status === "error", ), ); const handlers = (root as unknown as InspectableRlmSession)._createKernelHostHandlers(); const send = handlers["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
await expect( send({ message: "hello", receiver_role: "child", receiver_name: "shared-child" }), ).resolves.toMatchObject({ message: "hello" }); expect(sendAgentMessage).toHaveBeenCalledWith( expect.objectContaining({ target: "healthy-child-session", message: "hello" }), ); releaseFailureInjection(); });
it("falls through a delete-before-startup tombstone during a roled send", async () => { let releaseRuntimeCreation: () => void = () => {}; const runtimeCreationGate = new Promise<void>((resolve) => { releaseRuntimeCreation = resolve; }); let runtimeCreationStarted = false; const hostedChild = createSession(); const root = createSession({ agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [], sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { runtimeCreationStarted = true; await runtimeCreationGate; return { session: hostedChild }; }, deleteRlmSubagentRuntime: async (_id, child) => child?.disposeAsync(), }, }); await spawnChild(root, "blocked startup", { name: "deleted-child" }); await waitFor(() => runtimeCreationStarted); await root.deleteRlmSubagent("deleted-child"); const handlers = (root as unknown as InspectableRlmSession)._createKernelHostHandlers(); const send = handlers["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
await expect(send({ message: "hello", receiver_role: "child", receiver_name: "deleted-child" })).rejects.toThrow( 'No child matches "deleted-child"', ); releaseRuntimeCreation(); await waitFor(() => (root as unknown as InspectableRlmSession)._activeRlmChildRuns.size === 0); });
it("marks a broadcast delivery to the parent as replied without reloading the roster", async () => { const family = vi.fn(async () => [ { relationship: "parent" as const, entry: { id: "parent-session", name: "parent", depth: 0, status: "idle" as const }, }, ]); const sendAgentMessage = vi.fn(async () => ({ id: "agentmsg-broadcast-reply", source: "agent_message" as const, target: { activeSessionId: "parent-active", sessionId: "parent-session" }, message: "status", deliveryStatus: "delivered" as const, })); const child = createSession({ depth: 1, agentMessageController: { listAgents: () => ({ agents: [] }), family, sendAgentMessage, }, }); const handlers = (child as unknown as InspectableRlmSession)._createKernelHostHandlers(); const send = handlers["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler");
await expect(send({ target: "all", message: "status" })).resolves.toMatchObject({ receipts: [{ message: "status" }], }); expect(family).toHaveBeenCalledTimes(1); expect(child.repliedToParentSinceTask).toBe(true); });
it("routes family messages with sender-perspective labels and resets parent steer reply state", async () => { const parent = createSession(); parent.setSessionName("parent"); const child = createSession({ depth: 1 }); child.setSessionName("worker"); (child as unknown as { _repliedToParentSinceTask: boolean })._repliedToParentSinceTask = true;
const daemon = new AgentDaemon(join(tempDir, "daemon.sock"), { defaultSessionConfig: { agentDir: tempDir, cwd: tempDir }, createRuntime: vi.fn(), }); const parentState = { activeSessionId: "parent-active", clients: new Set(), pendingAttaches: 0, lastEventSequence: 0, runtime: { metadata: { kind: "top-level", createdAt: 1 }, session: parent, }, } as unknown as ActiveSessionState; const childState = { activeSessionId: "child-active", clients: new Set(), pendingAttaches: 0, lastEventSequence: 0, runtime: { metadata: { kind: "subagent", createdAt: 1, parentActiveSessionId: "parent-active", parentSessionId: parent.sessionId, }, session: child, }, } as unknown as ActiveSessionState; const internals = daemon as unknown as { sessions: Map<string, ActiveSessionState>; sendAgentSessionMessage(options: { targetSelector: string; message: string; fromState: ActiveSessionState; origin: "agent"; }): Promise<unknown>; }; internals.sessions.set(parentState.activeSessionId, parentState); internals.sessions.set(childState.activeSessionId, childState);
await internals.sendAgentSessionMessage({ targetSelector: parentState.activeSessionId, message: "done", fromState: childState, origin: "agent", }); const reply = findLastMessage(parent.messages, isAgentSessionMessage); expect(reply && isAgentSessionMessage(reply) ? reply.content : undefined).toContain( "[agent-message from child:worker]", );
await internals.sendAgentSessionMessage({ targetSelector: childState.activeSessionId, message: "continue", fromState: parentState, origin: "agent", }); const steer = findLastMessage(child.messages, isAgentSessionMessage); expect(steer && isAgentSessionMessage(steer) ? steer.content : undefined).toContain( "[agent-message from parent:", ); expect(child.repliedToParentSinceTask).toBe(false); });
it("resets replied state when a parent message is accepted", async () => { const child = createSession({ depth: 1 }); (child as unknown as { _repliedToParentSinceTask: boolean })._repliedToParentSinceTask = true; const message = createAgentSessionMessage({ id: "agentmsg-parent-task", source: "agent_message", message: "new task", fromRelationship: "parent", target: { activeSessionId: "child-active", sessionId: child.sessionId }, });
await child.acceptAgentMessagePrompt(message.content as string, { customMessage: message });
expect(child.repliedToParentSinceTask).toBe(false); });
it("resets replied state when a parent follow-up is queued", async () => { const child = createSession({ depth: 1 }); (child as unknown as { _repliedToParentSinceTask: boolean })._repliedToParentSinceTask = true; const message = createAgentSessionMessage({ id: "agentmsg-parent-follow-up", source: "agent_message", message: "continue", fromRelationship: "parent", target: { activeSessionId: "child-active", sessionId: child.sessionId }, });
await child.queueAgentMessagePrompt(message.content as string, "followUp", message);
expect(child.repliedToParentSinceTask).toBe(false); });
it("leaves replied state unknown when a child session is rehydrated", () => { const manager = SessionManager.create(tempDir, join(tempDir, "resumed-child")); manager.newSession({ rlmDepth: 1 }); manager.appendMessage({ role: "user", content: "previous task", timestamp: 1 }); manager.flushNow();
const resumed = createSession({ depth: 1, sessionManager: manager }); expect(resumed.repliedToParentSinceTask).toBeUndefined(); });
it("surfaces post-admission startup failure in the parent transcript and subagent registry", async () => { const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("kernel startup failed"); }, deleteRlmSubagentRuntime: async () => {}, }, }); const childUpdates: Array<{ status: string; error?: string }> = []; root.subscribe((event) => { if (event.type === "rlm_child_update") { childUpdates.push({ status: event.child.status, error: event.child.error }); } });
const spawned = await spawnChild(root, "start failing child", { name: "failing-worker" }); await vi.waitFor(async () => { expect((await root.listRlmSubagents()).subagents).toContainEqual( expect.objectContaining({ rlm_child_id: spawned.rlm_child_id, status: "error" }), ); }); await vi.waitFor(() => { expect(root.messages).toContainEqual( expect.objectContaining({ role: "custom", customType: "rlm_child_failure", content: expect.stringContaining("failing-worker"), }), ); expect(root.messages).toContainEqual( expect.objectContaining({ content: expect.stringContaining("kernel startup failed") }), ); }); // A terminal run that never bound a session leaves no row anywhere: the // terminal update is the cancelled removal signal and snapshots skip the run. expect(childUpdates.at(-1)).toEqual({ status: "cancelled", error: "kernel startup failed" }); expect(root.getRlmChildSnapshots()).toEqual([]); });
it("injects exactly one cancellation notice when a child run is cancelled", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: () => { const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("late child answer") }); }); return stream; }, });
const spawned = await spawnChild(root, "slow child", { name: "cancel-worker" }); await waitFor(() => childStarted); expect(root.cancelRlmChildRun(spawned.rlm_child_id)).toBe(true); releaseChild(); await vi.waitFor(() => { const notices = root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ); expect(notices).toHaveLength(1); expect(notices[0]).toMatchObject({ content: "[child-exited: cancelled child:cancel-worker]\n\nCancelled by user", details: { kind: "cancelled", reason: "Cancelled by user" }, }); }); });
it("harvests a silent child's last assistant text instead of only reporting the silence", async () => { const root = createSession();
await spawnChild(root, "silent child", { name: "silent-worker" }); await vi.waitFor(() => { const notices = root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ); expect(notices).toHaveLength(1); // R11: a child that produced text hands it over, tagged, rather than making the // parent wait for a reply that will never come. expect(notices[0]).toMatchObject({ content: expect.stringContaining("[auto-harvested fallback from silent-worker]"), details: { kind: "harvested_fallback", text: "child answer: silent child" }, }); }); });
it("suppresses a done child's unsettled fallback notice at the cancellation cut", async () => { const root = createSession(); let suppressed = false; root.subscribe((event) => { if (event.type === "rlm_child_update" && event.child.status === "done") { suppressed = root.cancelRlmChildRun(event.child.id); } });
await spawnChild(root, "silent child at cancellation cut", { name: "suppressed-worker" }); await root.waitForRlmQuiescence(); expect(suppressed).toBe(true); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0); });
it("does not inject a terminal notice when a parent follow-up resets reply state after a reply", async () => { const child = createSession({ depth: 1, rlmSessionDir: join(tempDir, "replying-child"), agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [ { relationship: "parent", entry: { id: "parent-session", name: "parent", depth: 0, status: "idle" } }, ], sendAgentMessage: async () => ({ id: "agentmsg-reply-before-follow-up", source: "agent_message", target: { activeSessionId: "parent-active", sessionId: "parent-session" }, message: "done", deliveryStatus: "delivered", }), }, }); vi.spyOn(child, "promptAndWait").mockImplementation(async () => { const send = (child as unknown as InspectableRlmSession)._createKernelHostHandlers()["agent_message.send"]; if (!send) throw new Error("Missing agent_message.send host handler"); await send({ message: "done", receiver_role: "parent" }); const followUp = createAgentSessionMessage({ id: "agentmsg-parent-follow-up-after-reply", source: "agent_message", message: "continue cleanup", fromRelationship: "parent", target: { activeSessionId: "child-active", sessionId: child.sessionId }, }); await child.queueAgentMessagePrompt(followUp.content as string, "followUp", followUp); expect(child.repliedToParentSinceTask).toBe(false); }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async () => {}, }, });
const spawned = await spawnChild(root, "reply first", { name: "reply-worker" }); await vi.waitFor(async () => { expect((await root.listRlmSubagents()).subagents).toContainEqual( expect.objectContaining({ rlm_child_id: spawned.rlm_child_id, status: "completed" }), ); }); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0); });
it("notifies after detached startup deletion cleanup settles", async () => { let releaseRuntimeCreation: () => void = () => {}; const runtimeCreationGate = new Promise<void>((resolve) => { releaseRuntimeCreation = resolve; }); let runtimeCreationStarted = false; const child = createSession({ rlmSessionDir: join(tempDir, "deleted-child") }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { runtimeCreationStarted = true; await runtimeCreationGate; return { session: child }; }, deleteRlmSubagentRuntime: async (_id, session) => session?.disposeAsync(), }, });
const spawned = await spawnChild(root, "delete before startup", { name: "deleted-worker" }); await waitFor(() => runtimeCreationStarted); await root.deleteRlmSubagent(spawned.rlm_child_id); releaseRuntimeCreation(); await waitFor(() => !(root as unknown as InspectableRlmSession)._activeRlmChildRuns.has(spawned.rlm_child_id)); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toEqual([ expect.objectContaining({ details: expect.objectContaining({ kind: "cancelled", reason: "Deleted by parent orchestrator" }), }), ]); });
it("fully deletes a settled startup failure and frees its session name", async () => { const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("kernel startup failed"); }, deleteRlmSubagentRuntime: async () => {}, }, }); const spawned = await spawnChild(root, "start failing child", { name: "reusable-worker" }); const internals = root as unknown as InspectableRlmSession; await vi.waitFor(() => expect(internals._activeRlmChildRuns.get(spawned.rlm_child_id)?.settled).toBe(true));
await expect(root.deleteRlmSubagent(spawned.rlm_child_id)).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id, session_name: "reusable-worker" }, });
expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(false); await expect(spawnChild(root, "replacement child", { name: "reusable-worker" })).resolves.toMatchObject({ name: "reusable-worker", }); });
it("releases a hosted child when its initial task fails", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "host-error-child") }); vi.spyOn(child, "promptAndWait").mockRejectedValue(new Error("child prompt failed")); const releaseRlmSubagentRuntime = vi.fn(async () => {}); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), releaseRlmSubagentRuntime, deleteRlmSubagentRuntime: async () => {}, }, });
const spawned = await spawnChild(root, "fail hosted child"); await vi.waitFor(() => { expect(releaseRlmSubagentRuntime).toHaveBeenCalledWith( expect.objectContaining({ session: child }), expect.objectContaining({ id: spawned.rlm_child_id }), "error", ); }); });
it("strong quiescence waits for a gated child bash activity change", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "bash-active-child") }); const bashStarted = deferred<void>(); const bashCompletion = deferred<void>(); const operations: BashOperations = { exec: async () => { bashStarted.resolve(); await bashCompletion.promise; return { exitCode: 0 }; }, }; const bash = child.executeBash("gated", undefined, { operations }); await bashStarted.promise;
await expect(child.waitForIdle()).resolves.toBeUndefined(); const root = createSession(); expect(root.registerRlmChildSession("bash-active-child", child)).toBe(true); const originalHeadlessIdle = child.waitForHeadlessIdle.bind(child); let headlessIdleCalls = 0; vi.spyOn(child, "waitForHeadlessIdle").mockImplementation(async () => { headlessIdleCalls++; await originalHeadlessIdle(); });
const quiescence = root.waitForRlmQuiescence(); const firstBoundary = await Promise.race([ quiescence.then( () => "resolved" as const, () => "rejected" as const, ), sleep(20).then(() => "timer" as const), ]); expect(firstBoundary).toBe("timer"); expect(headlessIdleCalls).toBe(1);
bashCompletion.resolve(); await bash; await expect(quiescence).resolves.toBeUndefined(); expect(headlessIdleCalls).toBe(2); });
it("rechecks parent self-activity after a child quiescence boundary", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "boundary-active-child") }); const childBashStarted = deferred<void>(); const childBashCompletion = deferred<void>(); const childBash = child.executeBash("child-boundary-gate", undefined, { operations: { exec: async () => { childBashStarted.resolve(); await childBashCompletion.promise; return { exitCode: 0 }; }, }, }); await childBashStarted.promise; const parentBashStarted = deferred<void>(); const parentBashCompletion = deferred<void>(); const root = createSession(); expect(root.registerRlmChildSession("boundary-active-child", child)).toBe(true); const originalChildQuiescence = child.waitForRlmQuiescence.bind(child); const childWaitStarted = deferred<void>(); let parentBash: Promise<unknown> | undefined; vi.spyOn(child, "waitForRlmQuiescence").mockImplementation(async (signal) => { childWaitStarted.resolve(); await originalChildQuiescence(signal); parentBash = root.executeBash("parent-boundary-gate", undefined, { operations: { exec: async () => { parentBashStarted.resolve(); await parentBashCompletion.promise; return { exitCode: 0 }; }, }, }); });
const quiescence = root.waitForRlmQuiescence(); await childWaitStarted.promise; childBashCompletion.resolve(); await childBash; const parentBoundaryStarted = await Promise.race([ parentBashStarted.promise.then(() => true), sleep(200).then(() => false), ]); expect(parentBoundaryStarted).toBe(true); const boundary = await Promise.race([ quiescence.then(() => "quiesced" as const), sleep(20).then(() => "active" as const), ]); expect(boundary).toBe("active");
parentBashCompletion.resolve(); await parentBash; const finalBoundary = await Promise.race([ quiescence.then(() => "quiesced" as const), sleep(200).then(() => "blocked" as const), ]); expect(finalBoundary).toBe("quiesced"); });
it("propagates root quiescence cancellation into a child self-active waiter", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "cancelled-bash-active-child") }); const bashStarted = deferred<void>(); const bashCompletion = deferred<void>(); const operations: BashOperations = { exec: async () => { bashStarted.resolve(); await bashCompletion.promise; return { exitCode: 0 }; }, }; const bash = child.executeBash("cancelled-gate", undefined, { operations }); await bashStarted.promise; const root = createSession(); expect(root.registerRlmChildSession("cancelled-bash-active-child", child)).toBe(true);
const quiescence = root.waitForRlmQuiescence(); await vi.waitFor(() => expect((child as unknown as InspectableRlmSession)._rlmQuiescenceWaitAborts.size).toBe(1)); root.requestAbort(); await expect(quiescence).rejects.toThrow("RLM quiescence wait cancelled"); expect(child.isBashRunning).toBe(true); expect((child as unknown as InspectableRlmSession)._rlmQuiescenceWaitAborts.size).toBe(0);
bashCompletion.resolve(); await bash; });
it("cancels sibling recursive waiters when one child quiescence wait fails", async () => { const childAStarted = deferred<void>(); const childACompletion = deferred<void>(); const childA = createSession({ rlmSessionDir: join(tempDir, "failing-wait-child") }); const childABash = childA.executeBash("child-a-gate", undefined, { operations: { exec: async () => { childAStarted.resolve(); await childACompletion.promise; return { exitCode: 0 }; }, }, }); const childBStarted = deferred<void>(); const childBCompletion = deferred<void>(); const childB = createSession({ rlmSessionDir: join(tempDir, "sibling-wait-child") }); const childBBash = childB.executeBash("child-b-gate", undefined, { operations: { exec: async () => { childBStarted.resolve(); await childBCompletion.promise; return { exitCode: 0 }; }, }, }); await Promise.all([childAStarted.promise, childBStarted.promise]); const root = createSession(); expect(root.registerRlmChildSession("failing-wait-child", childA)).toBe(true); expect(root.registerRlmChildSession("sibling-wait-child", childB)).toBe(true);
const quiescence = root.waitForRlmQuiescence(); await vi.waitFor(() => { expect((childA as unknown as InspectableRlmSession)._rlmQuiescenceWaitAborts.size).toBe(1); expect((childB as unknown as InspectableRlmSession)._rlmQuiescenceWaitAborts.size).toBe(1); }); childA.requestAbort(); await expect(quiescence).rejects.toThrow("RLM quiescence wait cancelled"); await vi.waitFor(() => expect((childB as unknown as InspectableRlmSession)._rlmQuiescenceWaitAborts.size).toBe(0), ); expect(childB.isBashRunning).toBe(true);
childACompletion.resolve(); childBCompletion.resolve(); await Promise.all([childABash, childBBash]); });
it("durably defers a child terminal notice across ACP-style input pause and scheduler suspension", async () => { const childStarted = deferred<void>(); const childCompletion = deferred<void>(); const synthesizedAgentMessageSend = vi.fn(async () => ({ id: "unexpected-synthesized-send", source: "agent_message" as const, target: { activeSessionId: "parent-active", sessionId: "parent-session" }, message: "unexpected", deliveryStatus: "delivered" as const, })); const child = createSession({ rlmSessionDir: join(tempDir, "paused-terminal-child"), agentMessageController: { listAgents: () => ({ agents: [] }), family: async () => [], sendAgentMessage: synthesizedAgentMessageSend, }, streamFn: (_model, context) => { const stream = createAssistantMessageEventStream(); childStarted.resolve(); void childCompletion.promise.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${userText(context)}`), }); }); return stream; }, }); let parentNoticeTurns = 0; const root = createSession({ streamFn: () => { parentNoticeTurns++; return streamAnswer("parent processed terminal notice"); }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async () => {}, }, });
await spawnChild(root, "finish during ACP close", { name: "paused-terminal-worker" }); await childStarted.promise; const inputPause = root.acquireSessionInputPause(); root.requestAbort(); await expect(root.prompt("external prompt", { resumeIfIdle: false })).rejects.toThrow( "session input admission is paused", );
childCompletion.resolve(); const internals = root as unknown as InspectableRlmSession; const deferredNotices = () => root .getPendingNextTurnMessageSnapshots() .filter((message) => message.customType === "rlm_child_terminal_notice"); await vi.waitFor(() => expect(deferredNotices()).toHaveLength(1)); expect(synthesizedAgentMessageSend).not.toHaveBeenCalled(); const restartSnapshot = root.getPendingNextTurnMessageSnapshots(); await vi.waitFor(() => expect(internals._unsettledRlmChildRuns.size).toBe(0)); expect(root.unfinishedActionCount).toBe(0); const closeIdleBoundary = await Promise.race([ root.waitForIdle().then(() => "idle" as const), sleep(50).then(() => "blocked" as const), ]); expect(closeIdleBoundary).toBe("idle"); const quiescence = root.waitForRlmQuiescence(); const strongBoundary = await Promise.race([ quiescence.then(() => "quiesced" as const), sleep(20).then(() => "paused" as const), ]); expect(strongBoundary).toBe("paused"); expect(root.clearQueue()).toEqual({ steering: [], followUp: [] }); expect(deferredNotices()).toHaveLength(1);
root.resumeQueuedWork(); await expect(root.waitForIdle()).resolves.toBeUndefined(); expect(parentNoticeTurns).toBe(0); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0);
inputPause.release(); await expect(quiescence).resolves.toBeUndefined(); expect(parentNoticeTurns).toBe(1); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(1);
let restoredNoticeTurns = 0; const restored = createSession({ streamFn: () => { restoredNoticeTurns++; return streamAnswer("restored parent processed terminal notice"); }, }); restored.restorePendingNextTurnMessages(restartSnapshot); await restored.waitForRlmQuiescence(); expect(restoredNoticeTurns).toBe(1); expect( restored.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(1); root.dispose(); });
it("demotes a pre-admitted terminal action when ACP close suspends before delivery", async () => { const childStarted = deferred<void>(); const childCompletion = deferred<void>(); const child = createSession({ rlmSessionDir: join(tempDir, "pre-admitted-terminal-child"), streamFn: () => { const stream = createAssistantMessageEventStream(); childStarted.resolve(); void childCompletion.promise.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("child finished") }); }); return stream; }, }); let parentNoticeTurns = 0; const root = createSession({ streamFn: () => { parentNoticeTurns++; return streamAnswer("parent processed pre-admitted terminal notice"); }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async () => {}, }, });
await spawnChild(root, "finish before ACP close cut", { name: "pre-admitted-terminal-worker" }); await childStarted.promise; const dispatchGate = vi .spyOn(root as unknown as { _scheduleSessionInputPump(): void }, "_scheduleSessionInputPump") .mockImplementation(() => {}); childCompletion.resolve(); await vi.waitFor(() => expect(root.unfinishedActionCount).toBe(1)); expect( root .getPendingNextTurnMessageSnapshots() .filter((message) => message.customType === "rlm_child_terminal_notice"), ).toHaveLength(0);
const inputPause = root.acquireSessionInputPause(); root.requestAbort(); await vi.waitFor(() => expect(root.unfinishedActionCount).toBe(0)); expect( root .getPendingNextTurnMessageSnapshots() .filter((message) => message.customType === "rlm_child_terminal_notice"), ).toHaveLength(1); dispatchGate.mockRestore(); await expect(root.waitForIdle()).resolves.toBeUndefined(); expect(root.clearQueue()).toEqual({ steering: [], followUp: [] });
root.resumeQueuedWork(); inputPause.release(); await root.waitForRlmQuiescence(); expect(parentNoticeTurns).toBe(1); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(1); });
it("linearizes terminal retention after an update checkpoint snapshot", async () => { const childStarted = deferred<void>(); const childCompletion = deferred<void>(); const child = createSession({ rlmSessionDir: join(tempDir, "update-checkpoint-terminal-child"), streamFn: () => { const stream = createAssistantMessageEventStream(); childStarted.resolve(); void childCompletion.promise.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("child finished for update") }); }); return stream; }, }); let parentNoticeTurns = 0; const root = createSession({ streamFn: () => { parentNoticeTurns++; return streamAnswer("parent processed update terminal notice"); }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async () => {}, }, });
const spawned = await spawnChild(root, "finish at update snapshot", { name: "update-terminal-worker" }); await childStarted.promise; const updatePause = root.acquireQueuedWorkPause(); childCompletion.resolve(); const internals = root as unknown as InspectableRlmSession; await vi.waitFor(() => expect(internals._activeRlmChildRuns.get(spawned.rlm_child_id)?.status).toBe("done")); await root.waitForSessionInputCheckpoint(); const pendingSnapshot = root.getPendingNextTurnMessageSnapshots(); const actionSnapshot = root.getSessionActionRecoverySnapshot(); expect(pendingSnapshot.filter((message) => message.customType === "rlm_child_terminal_notice")).toHaveLength(0); expect( actionSnapshot.actions.filter( (action) => action.payload.kind === "turn" && action.payload.customMessage?.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0); expect(internals._unsettledRlmChildRuns.size).toBe(1); expect(internals._activeRlmChildRuns.get(spawned.rlm_child_id)?.settled).toBe(false);
updatePause.release(); await root.waitForRlmQuiescence(); expect(parentNoticeTurns).toBe(1); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(1); });
it("keeps settled error deletion in quiescence through cleanup retry", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "settled-error-child") }); vi.spyOn(child, "promptAndWait").mockRejectedValue(new Error("child prompt failed")); const firstCleanup = deferred<void>(); const retryCleanup = deferred<void>(); let cleanupAttempts = 0; const deleteRlmSubagentRuntime = vi.fn(async (_id: string, session?: AgentSession) => { const cleanup = ++cleanupAttempts === 1 ? firstCleanup.promise : retryCleanup.promise; await cleanup; await session?.disposeAsync(); }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), releaseRlmSubagentRuntime: async () => {}, deleteRlmSubagentRuntime, }, });
const spawned = await spawnChild(root, "fail after startup", { name: "settled-error-worker" }); const internals = root as unknown as InspectableRlmSession; await vi.waitFor(() => expect(internals._activeRlmChildRuns.get(spawned.rlm_child_id)?.settled).toBe(true)); const run = internals._activeRlmChildRuns.get(spawned.rlm_child_id); if (!run) throw new Error("Missing settled error run"); expect(run.status).toBe("error"); expect(run.session).toBe(child); expect(internals._rlmChildSessions.has(spawned.rlm_child_id)).toBe(false); expect(internals._unsettledRlmChildRuns.has(run)).toBe(false);
await expect(root.deleteRlmSubagent("settled-error-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); await expect(root.deleteRlmSubagent(spawned.rlm_child_id)).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); expect(deleteRlmSubagentRuntime).toHaveBeenCalledOnce(); expect(run.settled).toBe(false); expect(internals._unsettledRlmChildRuns.has(run)).toBe(true); let quiesced = false; const quiescence = root.waitForRlmQuiescence().then(() => { quiesced = true; }); await sleep(20); expect(quiesced).toBe(false);
firstCleanup.reject(new Error("first cleanup failed")); await waitFor(() => internals._rlmChildCleanupFailures.has(spawned.rlm_child_id)); await sleep(20); expect(quiesced).toBe(false); expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(true); expect(internals._unsettledRlmChildRuns.has(run)).toBe(true); await expect(spawnChild(root, "replacement before retry", { name: "settled-error-worker" })).rejects.toThrow( "an agent of that name already exists at depth 1 under this parent", );
await expect(root.deleteRlmSubagent("settled-error-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); await expect(root.deleteRlmSubagent(spawned.rlm_child_id)).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); expect(deleteRlmSubagentRuntime).toHaveBeenCalledTimes(2); await sleep(20); expect(quiesced).toBe(false);
retryCleanup.resolve(); await quiescence; expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(false); expect(internals._unsettledRlmChildRuns.has(run)).toBe(false); expect(internals._deletingRlmChildren.has(spawned.rlm_child_id)).toBe(false); expect(deleteRlmSubagentRuntime).toHaveBeenCalledTimes(2); await expect( spawnChild(root, "replacement after cleanup", { name: "settled-error-worker" }), ).resolves.toMatchObject({ name: "settled-error-worker", }); });
it("notifies the runtime host when the initial child task completes", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "host-completion-child") }); const completeRlmSubagentRuntime = vi.fn(() => true); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), completeRlmSubagentRuntime, deleteRlmSubagentRuntime: async (_id, session) => session?.disposeAsync(), }, });
const spawned = await spawnChild(root, "persist completion"); await vi.waitFor(() => { expect(completeRlmSubagentRuntime).toHaveBeenCalledWith(spawned.rlm_child_id, child); }); expect((await root.listRlmSubagents()).subagents).toContainEqual( expect.objectContaining({ rlm_child_id: spawned.rlm_child_id, status: "completed" }), ); });
it("projects retained child follow-up turns into parent child-update activity", async () => { let releaseInitial: () => void = () => {}; const initialGate = new Promise<void>((resolve) => { releaseInitial = resolve; }); let releaseFollowUp: () => void = () => {}; const followUpGate = new Promise<void>((resolve) => { releaseFollowUp = resolve; }); const events: Array<{ status: string; activity?: { kind: string } }> = []; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); void (text === "initial" ? initialGate : followUpGate).then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`answer: ${text}`) }); }); return stream; }, }); root.subscribe((event) => { if (event.type === "rlm_child_update") { events.push({ status: event.child.status, activity: event.child.activity }); } });
const spawned = await spawnChild(root, "initial"); await waitFor(() => events.some((event) => event.status === "running" && event.activity?.kind === "waiting")); releaseInitial(); await waitFor(() => root.getRlmChildSession(spawned.rlm_child_id) !== undefined); await waitFor(() => events.at(-1)?.status === "done" && events.at(-1)?.activity === undefined);
const child = root.getRlmChildSession(spawned.rlm_child_id); if (!child) throw new Error("Missing retained child session"); const followUp = child.prompt("follow-up"); await waitFor(() => events.at(-1)?.status === "done" && events.at(-1)?.activity?.kind === "waiting"); releaseFollowUp(); await followUp; await waitFor(() => events.at(-1)?.status === "done" && events.at(-1)?.activity === undefined); });
it("lists a completed child with its parent-scoped messaging identity until disposal", async () => { let daemonChildId = ""; const root = createSession({ agentMessageController: { listAgents: () => ({ current: { activeSessionId: "parent-active", sessionId: "parent-session" }, agents: [ { activeSessionId: "other-child-active", sessionId: "other-child-session", runtimeKind: "subagent", cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentActiveSessionId: "other-parent", rlmChildId: daemonChildId, }, { activeSessionId: "child-active", sessionId: "child-session", runtimeKind: "subagent", cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentActiveSessionId: "parent-active", rlmChildId: daemonChildId, }, ], }), sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, });
const result = await spawnChild(root, "retained worker"); if (!result.session_dir) { throw new Error("Missing child session directory"); } daemonChildId = basename(result.session_dir); await waitFor(() => root.getRlmChildSession(daemonChildId)?.getLastAssistantText() !== undefined);
expect(root.getRlmChildSession(daemonChildId)?.getLastAssistantText()).toBe("child answer: retained worker"); const expectedSessionName = createDefaultRlmSubagentSessionName("retained worker", daemonChildId); expect(root.getRlmChildSession(daemonChildId)?.sessionName).toBe(expectedSessionName); const expectedRegistry = { subagents: [ { rlm_child_id: daemonChildId, active_session_id: "child-active", session_id: "child-session", session_name: expectedSessionName, session_dir: result.session_dir, status: "completed", answer_preview: "child answer: retained worker", duration_ms: expect.any(Number), label: "retained worker", last_activity_at: expect.any(Number), replied_since_task: false, model: "anthropic/claude-sonnet-4-5", turns: expect.any(Number), tokens: expect.any(Number), followups: 0, }, ], }; expect(await root.listRlmSubagents()).toEqual(expectedRegistry); const inspectable = root as unknown as InspectableRlmSession; const conflictingDeletion = { ...expectedRegistry.subagents[0], rlm_child_id: "deleting-child", session_name: daemonChildId, status: "completed" as const, }; inspectable._deletingRlmChildren.set("deleting-child", { subagent: conflictingDeletion, promise: Promise.resolve({ subagent: conflictingDeletion }), }); await expect(root.deleteRlmSubagent(daemonChildId)).rejects.toThrow("is ambiguous"); inspectable._deletingRlmChildren.delete("deleting-child");
const handlers = inspectable._createKernelHostHandlers(); const listHandler = handlers["rlm.list_subagents"]; const deleteHandler = handlers["rlm.delete_subagent"]; if (!listHandler || !deleteHandler) { throw new Error("Missing RLM subagent registry host handlers"); } await expect(listHandler({})).resolves.toEqual(expectedRegistry); await expect(deleteHandler({ target: expectedSessionName })).resolves.toEqual({ subagent: expectedRegistry.subagents[0], }); expect(root.getRlmChildSession(daemonChildId)).toBeUndefined(); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); await expect(deleteHandler({ target: expectedSessionName })).rejects.toThrow("No direct RLM subagent matches");
root.dispose();
expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); });
it("lists passive daemon children using their nonresident registry outcomes", async () => { const deleteRlmSubagentRuntime = vi.fn(async () => {}); const root = createSession({ agentMessageController: { listAgents: () => ({ current: { activeSessionId: "parent-active", sessionId: "parent-session" }, agents: ( [ ["failed", "running"], ["finished", "completed"], ] as const ).map(([name, registryStatus]) => ({ activeSessionId: `${name}-session`, sessionId: `${name}-session`, sessionName: `${name}-worker`, runtimeKind: "subagent", cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentActiveSessionId: "parent-active", rlmChildId: `${name}-child`, rlmChildRegistryStatus: registryStatus, sessionDir: join(tempDir, `${name}-child`), })), }), sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("unexpected hydration"); }, deleteRlmSubagentRuntime, }, }); const expected = [ { rlm_child_id: "failed-child", active_session_id: "failed-session", session_id: "failed-session", session_name: "failed-worker", session_dir: join(tempDir, "failed-child"), status: "error" as const, }, { rlm_child_id: "finished-child", active_session_id: "finished-session", session_id: "finished-session", session_name: "finished-worker", session_dir: join(tempDir, "finished-child"), status: "completed" as const, }, ];
expect(await root.listRlmSubagents()).toEqual({ subagents: expected }); await expect(root.deleteRlmSubagent("finished-worker")).resolves.toEqual({ subagent: expected[1] }); expect(deleteRlmSubagentRuntime).toHaveBeenCalledWith("finished-child", undefined); expect(await root.listRlmSubagents()).toEqual({ subagents: [expected[0]] }); });
it("coalesces concurrent deletion of the same passive daemon child", async () => { let releaseListing!: () => void; const listingGate = new Promise<void>((resolve) => { releaseListing = resolve; }); const deleteRlmSubagentRuntime = vi.fn(async () => {}); const root = createSession({ agentMessageController: { listAgents: async () => { await listingGate; return { current: { activeSessionId: "parent-active", sessionId: "parent-session" }, agents: [ { activeSessionId: "passive-session", sessionId: "passive-session", sessionName: "passive-worker", runtimeKind: "subagent" as const, cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentActiveSessionId: "parent-active", rlmChildId: "passive-child", sessionDir: join(tempDir, "passive-child"), }, ], }; }, sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("unexpected hydration"); }, deleteRlmSubagentRuntime, }, });
const first = root.deleteRlmSubagent("passive-worker"); const second = root.deleteRlmSubagent("passive-worker"); releaseListing(); await expect(Promise.all([first, second])).resolves.toHaveLength(2); expect(deleteRlmSubagentRuntime).toHaveBeenCalledOnce(); });
it("disposes an inline child when setting its session name fails", async () => { const root = createSession(); const appendSessionInfo = vi.spyOn(SessionManager.prototype, "appendSessionInfo").mockImplementation(() => { throw new Error("session info failed"); }); const dispose = vi.spyOn(AgentSession.prototype, "dispose"); try { const spawned = await spawnChild(root, "inline naming failure", { name: "bad-name" }); expect(spawned.rlm_child_id).toMatch(/^sub-/); await waitFor(() => dispose.mock.calls.length > 0); } finally { appendSessionInfo.mockRestore(); dispose.mockRestore(); } });
it("emits updated child session names after a retained child is renamed", async () => { const root = createSession(); const events: unknown[] = []; root.subscribe((event) => events.push(event));
const result = await spawnChild(root, "rename after completion", { name: "original-worker" }); if (!result.session_dir) { throw new Error("Missing child session directory"); } const childId = basename(result.session_dir); const child = root.getRlmChildSession(childId); if (!child) { throw new Error("Missing retained child session"); } const rootInternals = root as unknown as InspectableRlmSession; await waitFor(() => !rootInternals._activeRlmChildRuns.has(childId)); const completeRelease = root.releaseRlmChildSession(childId, child); if (!completeRelease) throw new Error("Failed to release retained child");
child.setCurrentRecap("retained recap"); child.setSessionName("renamed-worker");
const childUpdates = events.filter( (event): event is { type: "rlm_child_update"; child: RlmChildAgentSnapshot } => typeof event === "object" && event !== null && (event as { type?: string }).type === "rlm_child_update", ); expect(childUpdates.at(-1)?.child).toMatchObject({ sessionName: "renamed-worker", durationMs: expect.any(Number), tokenCount: 10, recap: "retained recap", repliedSinceTask: false, }); expect(root.getRlmChildSnapshots()).toEqual([ expect.objectContaining({ sessionName: "renamed-worker", durationMs: expect.any(Number), tokenCount: 10, recap: "retained recap", }), ]); });
it("surfaces a child's recap on its snapshot once the summarizer sets it", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; let root: AgentSession; root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); return stream; }, agentMessageController: { listAgents: () => { const run = [...(root as unknown as InspectableRlmSession)._activeRlmChildRuns.values()][0]; return { current: { activeSessionId: "parent-active", sessionId: root.sessionId }, agents: run?.session ? [ { activeSessionId: "running-active", sessionId: run.session.sessionId, runtimeKind: "subagent" as const, cwd: tempDir, isStreaming: true, unfinishedActionCount: 0, parentActiveSessionId: "parent-active", rlmChildId: run.id, sessionDir: run.sessionDir, }, ] : [], }; }, sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, });
const recaps: Array<string | undefined> = []; root.subscribe((event) => { if (event.type === "rlm_child_update") { recaps.push(event.child.recap); } });
await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const rootRun = [...(root as unknown as InspectableRlmSession)._activeRlmChildRuns.values()][0]; if (!rootRun?.session) { throw new Error("Missing child session on root run"); } expect(await root.listRlmSubagents()).toEqual({ subagents: [ { rlm_child_id: rootRun.id, active_session_id: "running-active", session_id: rootRun.session.sessionId, session_name: createDefaultRlmSubagentSessionName("slow shard", rootRun.id), session_dir: rootRun.sessionDir, status: "running", activity: { kind: "waiting" }, label: "slow shard", last_activity_at: expect.any(Number), replied_since_task: false, model: "anthropic/claude-sonnet-4-5", turns: 0, tokens: 0, followups: 0, }, ], });
// What the daemon summarizer does for subagents: stash the recap on the session, // which emits recap_update and re-emits the parent's enriched child snapshot. rootRun.session.setCurrentRecap("Summarizing the slow shard"); await waitFor(() => recaps.includes("Summarizing the slow shard"));
releaseChild(); await waitFor(() => rootRun.status === "done"); });
it("suppresses repeated child updates whose snapshot did not change", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); return stream; }, }); let updates = 0; root.subscribe((event) => { if (event.type === "rlm_child_update") updates += 1; });
await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const run = [...(root as unknown as InspectableRlmSession)._activeRlmChildRuns.values()][0]; if (!run?.emitUpdate || !run.session) throw new Error("Missing child run emit"); await waitFor(() => run.activity?.kind === "waiting"); const before = updates; run.emitUpdate(); run.emitUpdate(); expect(updates).toBe(before); run.session.setCurrentRecap("changed recap"); await waitFor(() => updates === before + 1);
releaseChild(); await waitFor(() => run.status === "done"); });
it("runs a child agent without requiring ripgrep", async () => { const streamFn = vi.fn((_model, context: Context) => streamAnswer(`child answer: ${userText(context)}`)); const root = createSession({ streamFn });
const result = await spawnChild(root, "summarize shard 1");
expect(result.rlm_child_id).toMatch(/^sub-/); await waitFor(() => streamFn.mock.calls.length >= 1); });
it("adds child usage to the parent session aggregate", async () => { const root = createSession(); const parentAssistant = assistantMessage("running ipython", usage(0, 0)); root.agent.state.messages.push(parentAssistant); root.sessionManager.appendMessage(parentAssistant);
const before = root.getSessionStats(); await spawnChild(root, "summarize shard 2"); await waitFor(() => root.sessionManager.getEntries().some((entry) => entry.type === "child_usage_attributed")); const after = root.getSessionStats();
expect(after.tokens.input).toBeGreaterThanOrEqual(before.tokens.input + 7); expect(after.tokens.output).toBeGreaterThanOrEqual(before.tokens.output + 3); expect(after.tokens.total).toBeGreaterThanOrEqual(before.tokens.total + 10); expect(after.cost).toBeGreaterThanOrEqual(before.cost + 10); expect(parentAssistant.usage.totalTokens).toBe(0);
const parentEntry = root.sessionManager .getEntries() .find((entry) => entry.type === "message" && entry.message === parentAssistant); if (!parentEntry || parentEntry.type !== "message" || parentEntry.message.role !== "assistant") { throw new Error("parent assistant entry was not recorded"); } expect(parentEntry.message.usage.input).toBe(7); expect(parentEntry.message.usage.output).toBe(3); expect(parentEntry.message.usage.cost.total).toBe(10);
const sessionFile = root.sessionManager.getSessionFile(); if (!sessionFile) throw new Error("parent session file was not created"); expect(readFileSync(sessionFile, "utf-8")).toContain('"type":"child_usage_attributed"'); const reloaded = SessionManager.open(sessionFile, join(tempDir, "sessions")); const attribution = reloaded.getEntries().find((entry) => entry.type === "child_usage_attributed"); if (!attribution || attribution.type !== "child_usage_attributed") throw new Error("missing attribution"); expect(attribution.childUsage.input).toBe(7); expect(attribution.childUsage.output).toBe(3); expect(attribution.aggregateUsage.cost.total).toBe(10); });
it("retains child usage until a delayed parent message_end hook persists its assistant", async () => { const hookEntered = deferred<void>(); const releaseHook = deferred<void>(); let parentAssistant: AssistantMessage | undefined; let child: AgentSession | undefined; const extensionsResult = await createTestExtensionsResult([ (pi) => { pi.on("message_end", async (event) => { if ( event.message.role === "assistant" && event.message.content.some( (block) => block.type === "toolCall" && block.name === "spawn_accounting_child", ) ) { parentAssistant = event.message; hookEntered.resolve(); await releaseHook.promise; } }); }, ]); const root = createSession({ extensionsResult, customTools: [ { name: "spawn_accounting_child", description: "Spawn a child for the accounting test", label: "spawn child", parameters: Type.Object({}), execute: async () => { const admitted = await spawnChild(root, "charge child"); child = root.getRlmChildSession(admitted.rlm_child_id); return { content: [{ type: "text" as const, text: admitted.rlm_child_id }], details: {} }; }, }, ], streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); queueMicrotask(() => { if (text === "spawn accounting child") { stream.push({ type: "done", reason: "toolUse", message: { ...assistantMessage("", usage(2, 1)), content: [ { type: "toolCall", id: "spawn-accounting", name: "spawn_accounting_child", arguments: {} }, ], stopReason: "toolUse", }, }); } else { stream.push({ type: "done", reason: "stop", message: assistantMessage( text === "charge child" ? "child done" : "parent done", text === "charge child" ? usage(7, 3) : usage(0, 0), ), }); } }); return stream; }, }); const prompt = root.prompt("spawn accounting child"); try { await hookEntered.promise; await waitFor(() => root.getRlmChildSnapshots().some((snapshot) => snapshot.status === "done")); expect(child?.getLastAssistantText()).toBe("child done"); expect(parentAssistant?.usage.cost.total).toBe(13); expect( root.sessionManager .getEntries() .some((entry) => entry.type === "message" && entry.message === parentAssistant), ).toBe(false); } finally { releaseHook.resolve(); await prompt; await root.agent.waitForIdle(); } root.sessionManager.flushNow(); const sessionFile = root.sessionFile; if (!sessionFile || !child?.sessionFile) throw new Error("Missing persisted sessions"); const reopenedEntries = SessionManager.open(sessionFile, join(tempDir, "sessions")).getEntries(); const reopenedUsage = computeOwnAndTotalUsage(reopenedEntries, reopenedEntries); const childEntries = SessionManager.open(child.sessionFile, join(tempDir, "sessions")).getEntries(); const childUsage = computeOwnAndTotalUsage(childEntries, childEntries); const attributions = reopenedEntries.filter((entry) => entry.type === "child_usage_attributed"); expect({ liveTotal: root.getSessionStats().cost, liveOwn: root.getOwnUsageSummary()?.cost, attributedChild: attributions.reduce((total, entry) => total + entry.childUsage.cost.total, 0), reopenedTotal: reopenedUsage.totalUsage.cost.total, reopenedOwn: reopenedUsage.ownUsage.cost.total, scannedOwn: (await readSessionInfo(sessionFile))?.usage?.cost, childLiveOwn: child.getOwnUsageSummary()?.cost, childReopenedOwn: childUsage.ownUsage.cost.total, childScannedOwn: (await readSessionInfo(child.sessionFile))?.usage?.cost, }).toEqual({ liveTotal: 13, liveOwn: 3, attributedChild: 10, reopenedTotal: 13, reopenedOwn: 3, scannedOwn: 3, childLiveOwn: 10, childReopenedOwn: 10, childScannedOwn: 10, }); });
it.each([ { memoized: false, failFirstAppend: false }, { memoized: true, failFirstAppend: false }, { memoized: true, failFirstAppend: true }, ])( "keeps a sibling's pending usage out of own spend (memoized=$memoized, failedAppend=$failFirstAppend)", async ({ memoized, failFirstAppend }) => { const gates = new Map([ ["A", deferred<void>()], ["B", deferred<void>()], ]); const started = new Set<string>(); const root = createSession({ customTools: [ { name: "hold", description: "Wait for release", label: "hold", parameters: Type.Object({ child: Type.String() }), execute: async (_toolCallId: string, params: { child: string }) => { started.add(params.child); await gates.get(params.child)!.promise; return { content: [{ type: "text" as const, text: "released" }], details: {} }; }, }, ], streamFn: (_model, context) => { const stream = createAssistantMessageEventStream(); const child = userText(context); const finished = context.messages.some((message) => message.role === "toolResult") || !gates.has(child); queueMicrotask(() => { const message = finished ? assistantMessage("done", usage(0, 0)) : { ...assistantMessage("", child === "A" ? usage(7, 3) : usage(11, 5)), content: [ { type: "toolCall" as const, id: `hold-${child}`, name: "hold", arguments: { child } }, ], stopReason: "toolUse" as const, }; stream.push({ type: "done", reason: finished ? "stop" : "toolUse", message }); }); return stream; }, }); const parentAssistant = assistantMessage("running ipython", usage(2, 1)); root.agent.state.messages.push(parentAssistant); root.sessionManager.appendMessage(parentAssistant); const sessionFile = root.sessionManager.getSessionFile(); if (!sessionFile) throw new Error("Missing parent session file"); if (memoized) expect(root.getOwnUsageSummary()?.cost).toBe(3); let failedAppend = false; let appendFailureCode: string | undefined; let attributionAttempts = 0; const persist = root.sessionManager._persist.bind(root.sessionManager); vi.spyOn(root.sessionManager, "_persist").mockImplementation((entry) => { if (entry.type === "child_usage_attributed") attributionAttempts++; if (failFirstAppend && !failedAppend && entry.type === "child_usage_attributed") { failedAppend = true; const content = readFileSync(sessionFile); rmSync(sessionFile); mkdirSync(sessionFile); try { persist(entry); } catch (error) { appendFailureCode = (error as NodeJS.ErrnoException).code; throw error; } finally { rmSync(sessionFile, { recursive: true, force: true }); writeFileSync(sessionFile, content); } return; } persist(entry); }); let otherBranchInput = 0; const expectLiveOwnUsage = () => { const tree = root.getContextTree(); expect(tree.totalUsage.cost.total).toBe(29); expect(tree.ownUsage.cost.total).toBe(3); expect(root.getOwnUsageSummary()).toEqual({ inputTokens: 2 + otherBranchInput, outputTokens: 1, cost: 3 + otherBranchInput, }); }; const children: AgentSession[] = []; const attributions = () => root.sessionManager.getEntries().filter((entry) => entry.type === "child_usage_attributed"); const expectDurableOwnUsage = async () => { const reopened = SessionManager.open(sessionFile, join(tempDir, "sessions")); const entries = reopened.getEntries(); const { ownUsage } = computeOwnAndTotalUsage(entries, entries); expect(ownUsage.input).toBe(2 + otherBranchInput); expect(ownUsage.output).toBe(1); expect(ownUsage.cost.total).toBe(3 + otherBranchInput); expect((await readSessionInfo(sessionFile))?.usage).toEqual({ inputTokens: 2 + otherBranchInput, outputTokens: 1, cost: 3 + otherBranchInput, }); }; try { for (const name of ["A", "B"]) { const admitted = await spawnChild(root, name); const child = root.getRlmChildSession(admitted.rlm_child_id); if (!child) throw new Error("Missing child session"); children.push(child); } await waitFor(() => started.size === 2); expect(parentAssistant.usage.cost.total).toBe(29); expect(attributions()).toHaveLength(0); expectLiveOwnUsage(); const chargedBranch = root.sessionManager.appendCustomEntry("usage-test-marker", {}); expectLiveOwnUsage(); root.sessionManager.resetLeaf(); root.sessionManager.appendMessage(assistantMessage("other branch", usage(5, 0))); otherBranchInput = 5; const otherTree = root.getContextTree(); expect(otherTree.ownUsage.cost.total).toBe(5); expect(otherTree.totalUsage.cost.total).toBe(5); expect(root.getOwnUsageSummary()?.cost).toBe(8); root.sessionManager.branch(chargedBranch); expectLiveOwnUsage(); gates.get("A")!.resolve(); await waitFor(() => attributions().length === 1); expect(attributions()[0]?.aggregateUsage.cost.total).toBe(13); expect(parentAssistant.usage.cost.total).toBe(29); expect(parentAssistant.usage.totalTokens).toBe(3); expectLiveOwnUsage(); await expectDurableOwnUsage(); gates.get("B")!.resolve(); await waitFor(() => attributions().length === 2); expect(attributions().map((entry) => entry.aggregateUsage.cost.total)).toEqual([ 13, failFirstAppend ? 19 : 29, ]); expect(parentAssistant.usage.cost.total).toBe(29); expect(parentAssistant.usage.totalTokens).toBe(3); expectLiveOwnUsage(); await expectDurableOwnUsage(); expect(failedAppend).toBe(failFirstAppend); expect(appendFailureCode).toBe(failFirstAppend ? "EISDIR" : undefined); await waitFor(() => root.getRlmChildSnapshots().every((snapshot) => snapshot.status === "done")); expect(attributionAttempts).toBe(2); } finally { for (const gate of gates.values()) gate.resolve(); await Promise.all(children.map((child) => child.agent.waitForIdle())); } }, );
it("coalesces the admitted task's tool-loop turns into one flushed spawn-usage attribution", async () => { const root = createToolLoopSession(2, [usage(1, 1), usage(2, 2)]);
await spawnChild(root, "use a tool"); await vi.waitFor(() => { const attributions = root.sessionManager .getEntries() .filter((entry) => entry.type === "child_usage_attributed"); expect(attributions).toHaveLength(1); expect(attributions[0]?.origin).toBe("spawn_task"); expect(attributions[0]?.childUsage.input).toBe(3); expect(attributions[0]?.childUsage.output).toBe(3); }); });
it("flushes a stale pending usage batch before extending it, bounding crash loss", async () => { vi.useFakeTimers({ toFake: ["Date"] }); try { // The batch from the first two completions is older than the staleness // bound when the third lands. const root = createToolLoopSession(3, [usage(1, 1), usage(2, 2), usage(4, 4)], (toolResultCount) => { if (toolResultCount === 2) vi.setSystemTime(Date.now() + 61_000); });
await spawnChild(root, "use a tool"); await vi.waitFor(() => { const attributions = root.sessionManager .getEntries() .filter((entry) => entry.type === "child_usage_attributed"); expect(attributions.map((entry) => [entry.childUsage.input, entry.childUsage.output])).toEqual([ [3, 3], [4, 4], ]); expect(attributions.map((entry) => entry.origin)).toEqual(["spawn_task", "spawn_task"]); // Each aggregate covers exactly the completions durable with or // before it, so any prefix replays to the exact own spend. expect(attributions.map((entry) => entry.aggregateUsage.input)).toEqual([3, 7]); });
const sessionFile = root.sessionManager.getSessionFile(); if (!sessionFile) throw new Error("parent session file was not created"); const reloadedAttributions = SessionManager.open(sessionFile, join(tempDir, "sessions")) .getEntries() .filter((entry) => entry.type === "child_usage_attributed"); const childTotal = reloadedAttributions.reduce((total, entry) => total + entry.childUsage.input, 0); // Parent own spend is zero here, so the reloaded aggregate must equal the summed child usage. expect(reloadedAttributions.at(-1)?.aggregateUsage.input).toBe(childTotal); } finally { vi.useRealTimers(); } });
it("gets and persists per-chat max-depth changes without transcript messages", async () => { const root = createSession(); const originalMessages = [...root.messages];
expect(root.getRlmMaxDepthStatus()).toEqual({ maxDepth: 2, source: "default" }); await expect(root.setRlmMaxDepth(-1)).rejects.toThrow("non-negative integer"); await root.setRlmMaxDepth(3);
expect(root.getRlmMaxDepthStatus()).toEqual({ maxDepth: 3, source: "chat" }); expect(root.messages).toEqual(originalMessages); const stateEntries = root.sessionManager .getBranch() .filter((entry) => entry.type === "custom" && entry.customType === "rlm_max_depth_state"); expect(stateEntries.at(-1)).toMatchObject({ data: { maxDepth: 3 } }); });
it("applies max-depth immediately while a turn streams without aborting or entering the transcript", async () => { let releaseTurn!: () => void; const release = new Promise<void>((resolve) => { releaseTurn = resolve; }); let turnStarted = false; const root = createSession({ streamFn: () => { const stream = createAssistantMessageEventStream(); turnStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("finished normally") }); }); return stream; }, });
const promptPromise = root.prompt("keep streaming"); await waitFor(() => turnStarted); const messagesBeforeSet = [...root.messages]; await root.setRlmMaxDepth(2); expect(root.rlmMaxDepth).toBe(2); expect(root.messages).toEqual(messagesBeforeSet); expect(root.isStreaming).toBe(true);
releaseTurn(); await promptPromise; await root.agent.waitForIdle(); expect(root.messages.at(-1)).toMatchObject({ role: "assistant", stopReason: "stop" }); });
it("uses a max-depth prompt update on the next turn of the active run", async () => { let releaseFirstTurn = () => {}; const firstTurnPending = new Promise<void>((resolve) => { releaseFirstTurn = resolve; }); const seenSystemPrompts: string[] = []; const root = createSession({ streamFn: (_model, context) => { seenSystemPrompts.push(context.systemPrompt ?? ""); if (seenSystemPrompts.length === 1) { const stream = createAssistantMessageEventStream(); void firstTurnPending.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("first turn") }); }); return stream; } return streamAnswer("second turn"); }, });
const promptPromise = root.prompt("start"); await waitFor(() => seenSystemPrompts.length === 1); expect(seenSystemPrompts[0]!).toContain("An `rlm` object");
await root.setRlmMaxDepth(0); await root.steer("continue after max-depth update"); releaseFirstTurn(); await promptPromise;
expect(seenSystemPrompts).toHaveLength(2); expect(seenSystemPrompts[1]!).not.toContain("An `rlm` object"); });
it("rehydrates chat max depth ahead of reconstruction config", async () => { const root = createSession(); await root.setRlmMaxDepth(3); if (!root.sessionFile) throw new Error("Missing persisted session file"); const sessionFile = root.sessionFile; root.dispose();
const resumedManager = SessionManager.open(sessionFile, join(tempDir, "sessions")); const resumed = createSession({ sessionManager: resumedManager, maxDepth: 4 }); expect(resumed.getRlmMaxDepthStatus()).toEqual({ maxDepth: 3, source: "chat" }); });
it("reloads max depth and its source when navigating to a branch without an override", async () => { vi.stubEnv("RLM_MAX_DEPTH", "0"); try { const root = createSession(); await root.prompt("baseline branch"); await root.agent.waitForIdle(); const baselineLeafId = root.sessionManager.getLeafId(); if (!baselineLeafId) throw new Error("Missing baseline branch leaf");
await root.setRlmMaxDepth(2); expect(root.systemPrompt).toContain("An `rlm` object"); await root.navigateTree(baselineLeafId, { summarize: false }); expect(root.getRlmMaxDepthStatus()).toEqual({ maxDepth: 0, source: "env" }); expect(root.systemPrompt).not.toContain("An `rlm` object"); } finally { vi.unstubAllEnvs(); } });
it("applies --global to this chat and new sessions without changing existing sessions", async () => { const current = createSession(); const existingSettings = SettingsManager.create(tempDir, tempDir); const existing = createSession({ settingsManager: existingSettings });
await expect(current.setRlmMaxDepth(4, { global: true })).resolves.toMatchObject({ maxDepth: 4, source: "chat", globalSaved: true, }); expect(existing.rlmMaxDepth).toBe(2); const freshSettings = SettingsManager.create(tempDir, tempDir); const fresh = createSession({ settingsManager: freshSettings }); expect(fresh.getRlmMaxDepthStatus()).toEqual({ maxDepth: 4, source: "global" }); current.dispose(); existing.dispose(); });
it("does not claim a failed global max-depth write was saved", async () => { const globalSettings = JSON.stringify({ rlmMaxDepth: 1 }); const storage: SettingsStorage = { withLock(scope, update) { const current = scope === "global" ? globalSettings : undefined; const next = update(current); if (scope === "global" && next !== undefined) { throw new Error("EROFS: read-only file system"); } }, }; const current = createSession({ settingsManager: SettingsManager.fromStorage(storage) });
const result = await current.setRlmMaxDepth(5, { global: true });
expect(result).toMatchObject({ maxDepth: 5, source: "chat", globalSaved: false }); expect(result.globalError).toContain("EROFS: read-only file system"); expect(SettingsManager.fromStorage(storage).getRlmMaxDepth()).toBe(1); expect(globalSettings).toBe(JSON.stringify({ rlmMaxDepth: 1 })); });
it("does not attribute stale errors to a successful global max-depth write", async () => { const writes: Record<"global" | "project", string | undefined> = { global: JSON.stringify({ rlmMaxDepth: 1 }), project: undefined, }; let failGlobal = true; let failProject = true; const storage: SettingsStorage = { withLock(scope, update) { const next = update(writes[scope]); if (scope === "global" && failGlobal && next !== undefined) { throw new Error("stale global failure"); } if (scope === "project" && failProject && next !== undefined) { throw new Error("project warning"); } writes[scope] = next; }, }; const settingsManager = SettingsManager.fromStorage(storage); settingsManager.setDefaultModelAndProvider("provider", "model"); await settingsManager.flush(); settingsManager.setProjectExtensionPaths(["broken-project-extension"]); await settingsManager.flush(); // Preserve the queued project diagnostic while recovering the global store. failGlobal = false; failProject = false; const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); const current = createSession({ settingsManager });
const result = await current.setRlmMaxDepth(5, { global: true });
expect(result).toMatchObject({ maxDepth: 5, source: "chat", globalSaved: true }); expect(warn).toHaveBeenCalledWith("Warning: Earlier global settings write failed: stale global failure"); expect(settingsManager.drainErrors()).toMatchObject([ { scope: "project", error: { message: "project warning" } }, ]); expect(JSON.parse(writes.global ?? "{}")).toMatchObject({ rlmMaxDepth: 5 }); warn.mockRestore(); });
it("rolls back max-depth state when chat persistence fails", async () => { const root = createSession(); const originalPrompt = root.systemPrompt; const flush = vi.spyOn(root.sessionManager, "flushNow").mockImplementation(() => { throw new Error("disk full"); });
await expect(root.setRlmMaxDepth(0)).rejects.toThrow("disk full"); expect(root.rlmMaxDepth).toBe(2); expect(root.systemPrompt).toBe(originalPrompt); expect( root.sessionManager .getBranch() .some((entry) => entry.type === "custom" && entry.customType === "rlm_max_depth_state"), ).toBe(false);
flush.mockRestore(); root.sessionManager.flushNow(); expect(readFileSync(root.sessionFile!, "utf8")).not.toContain('"customType":"rlm_max_depth_state"'); }); it("falls through an invalid global max depth while navigating off a chat override", async () => { const original = createSession(); await original.prompt("baseline branch"); await original.agent.waitForIdle(); const baselineLeafId = original.sessionManager.getLeafId(); if (!baselineLeafId) throw new Error("Missing baseline branch leaf"); await original.setRlmMaxDepth(2); const sessionFile = original.sessionFile; if (!sessionFile) throw new Error("Missing persisted session file"); original.dispose();
vi.stubEnv("RLM_MAX_DEPTH", "0"); try { const resumed = createSession({ sessionManager: SessionManager.open(sessionFile, join(tempDir, "sessions")), settingsManager: SettingsManager.inMemory({ rlmMaxDepth: -1 }), }); expect(resumed.rlmMaxDepth).toBe(2);
await expect(resumed.navigateTree(baselineLeafId, { summarize: false })).resolves.toMatchObject({ cancelled: false, }); expect(resumed.sessionManager.getLeafId()).toBe(baselineLeafId); expect(resumed.rlmMaxDepth).toBe(0); expect(resumed.systemPrompt).not.toContain("An `rlm` object"); } finally { vi.unstubAllEnvs(); } });
it("keeps a spawned child's chat override when reconstructed with its inherited config", async () => { const root = createSession({ maxDepth: 2 }); const childResult = await spawnChild(root, "child with a durable override"); if (!childResult.session_dir) throw new Error("Missing child session directory"); await waitFor(() => root.getRlmChildSession(childResult.rlm_child_id) !== undefined); const child = root.getRlmChildSession(childResult.rlm_child_id); if (!child?.sessionFile) throw new Error("Missing persisted child session"); await waitFor(() => (root as unknown as InspectableRlmSession)._activeRlmChildRuns.size === 0);
expect(child.getRlmMaxDepthStatus()).toEqual({ maxDepth: 2, source: "inherited" }); await child.setRlmMaxDepth(3); const childSessionFile = child.sessionFile; root.dispose();
const rehydratedManager = SessionManager.open(childSessionFile, join(tempDir, "sessions")); const rehydratedChild = createSession({ sessionManager: rehydratedManager, depth: 1, maxDepth: 2 }); expect(rehydratedChild.getRlmMaxDepthStatus()).toEqual({ maxDepth: 3, source: "chat" }); });
it("copies live max depth at spawn, keeps child overrides independent, and lets zero disable root spawning", async () => { const root = createSession(); await root.setRlmMaxDepth(2); const childResult = await spawnChild(root, "first child"); if (!childResult.session_dir) throw new Error("Missing child session directory"); await waitFor(() => root.getRlmChildSession(childResult.rlm_child_id) !== undefined); const child = root.getRlmChildSession(childResult.rlm_child_id); if (!child) throw new Error("Missing retained child session"); await waitFor(() => (root as unknown as InspectableRlmSession)._activeRlmChildRuns.size === 0); expect(child.rlmMaxDepth).toBe(2);
await child.setRlmMaxDepth(3); expect((child as unknown as InspectableRlmDirSession)._rlmKernelEnv().RLM_MAX_DEPTH).toBe("3"); expect(root.rlmMaxDepth).toBe(2); const grandchildResult = await spawnChild(child, "grandchild after override"); if (!grandchildResult.session_dir) throw new Error("Missing grandchild session directory"); await waitFor(() => child.getRlmChildSession(grandchildResult.rlm_child_id) !== undefined); const grandchild = child.getRlmChildSession(grandchildResult.rlm_child_id); expect(grandchild?.rlmMaxDepth).toBe(3);
await root.setRlmMaxDepth(0); expect(root.systemPrompt).not.toContain("An `rlm` object"); await expect(spawnChild(root, "blocked at root")).rejects.toThrow( "RLM recursion depth limit reached (RLM_DEPTH=0, RLM_MAX_DEPTH=0)", ); expect(child.rlmMaxDepth).toBe(3); });
it("creates an independent root session through the explicit daemon host operation", async () => { const createRlmSubagentRuntime = vi.fn(async () => { throw new Error("unexpected child spawn"); }); const createRlmRootSession = vi.fn(async () => ({ active_session_id: "root-active", session_id: "root-session", name: "researcher", session_file: join(tempDir, "sessions", "root-session.jsonl"), model: `${model.provider}/${model.id}`, })); const assertSessionNameAvailable = vi.fn(); const root = createSession({ agentMessageController: { assertSessionNameAvailable, listAgents: async () => ({ current: { activeSessionId: "current-root", sessionId: "current-session" }, agents: [], }), sendAgentMessage: async () => { throw new Error("unexpected message"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime, createRlmRootSession, deleteRlmSubagentRuntime: vi.fn(async () => {}), }, });
await expect( root.createRlmSession("independent task", { name: "researcher", model: `${model.provider}/${model.id}`, thinking: "off", cwd: "other-project", }), ).resolves.toEqual({ active_session_id: "root-active", session_id: "root-session", name: "researcher", session_file: join(tempDir, "sessions", "root-session.jsonl"), model: `${model.provider}/${model.id}`, }); expect(createRlmSubagentRuntime).not.toHaveBeenCalled(); expect(assertSessionNameAvailable).toHaveBeenCalledWith({ name: "researcher", depth: 0 }); expect(createRlmRootSession).toHaveBeenCalledWith({ prompt: "independent task", sessionName: "researcher", cwd: join(tempDir, "other-project"), model, thinkingLevel: "off", }); await expect(root.createRlmSession("task", { cwd: " " })).rejects.toThrow( "rlm.create_session cwd must be a non-empty string", ); await expect(root.createRlmSession(" ")).rejects.toThrow("rlm.create_session prompt must not be empty"); expect(createRlmRootSession).toHaveBeenCalledOnce(); });
it("does not create a root session after disposal during name preflight", async () => { let releaseName: () => void = () => {}; const nameGate = new Promise<void>((resolve) => { releaseName = resolve; }); const createRlmRootSession = vi.fn(async () => ({ active_session_id: "new-root", session_id: "new-session", name: "researcher", session_file: join(tempDir, "new-session.jsonl"), model: `${model.provider}/${model.id}`, })); const root = createSession({ agentMessageController: { assertSessionNameAvailable: () => nameGate, listAgents: async () => ({ current: { activeSessionId: "root", sessionId: "session" }, agents: [] }), sendAgentMessage: vi.fn(), }, subagentRuntimeHost: { createRlmSubagentRuntime: vi.fn(), createRlmRootSession, deleteRlmSubagentRuntime: vi.fn(), }, }); const creating = root.createRlmSession("independent task", { name: "researcher" }); root.dispose(); releaseName(); await expect(creating).rejects.toThrow("disposed"); expect(createRlmRootSession).not.toHaveBeenCalled(); });
it("keeps top-level session creation unavailable to nested or inline sessions", async () => { const createRlmRootSession = vi.fn(); const nested = createSession({ depth: 1, subagentRuntimeHost: { createRlmSubagentRuntime: vi.fn(), createRlmRootSession, deleteRlmSubagentRuntime: vi.fn(async () => {}), }, }); await expect(nested.createRlmSession("escape to root")).rejects.toThrow("available only from a depth-0 session"); expect(createRlmRootSession).not.toHaveBeenCalled(); nested.dispose();
const inline = createSession(); await expect(inline.createRlmSession("detached root")).rejects.toThrow( "requires a daemon-backed depth-0 session", ); });
it("rejects child creation at the configured recursion depth cap", async () => { const root = createSession({ depth: 1, maxDepth: 1 });
await expect(spawnChild(root, "nested")).rejects.toThrow("RLM recursion depth limit reached"); });
it("rejects unsupported rlm.spawn kwargs loudly", async () => { const root = createSession();
await expect(spawnChild(root, "nested", { temperature: 0 })).rejects.toThrow( "Unsupported rlm.spawn kwargs: temperature", ); });
it("rejects invalid rlm.spawn thinking kwargs", async () => { const root = createSession();
await expect(spawnChild(root, "nested", { thinking: 3 })).rejects.toThrow("rlm.spawn thinking must be a string"); await expect(spawnChild(root, "nested", { thinking: "ultra" })).rejects.toThrow( "rlm.spawn thinking must be one of: off, minimal, low, medium, high", ); });
it("accepts host kwarg in rlm.run and propagates to child session runtime", async () => { let capturedOptions: CreateRlmSubagentRuntimeOptions | undefined; const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async (options) => { capturedOptions = options; return { session: options.parentSession }; }, deleteRlmSubagentRuntime: async () => {}, }, });
await spawnChild(root, "remote task", { host: "chernobog" }); expect(capturedOptions?.remoteHost).toBe("chernobog"); });
it("configures inline child session with remoteHost when host kwarg is provided", async () => { const root = createSession({ streamFn: () => streamAnswer("child finished"), });
const handle = await spawnChild(root, "inline remote task", { host: "chernobog" }); const activeRun = (root as any)._activeRlmChildRuns.get(handle.rlm_child_id); expect(activeRun).toBeDefined(); expect((activeRun.session as any)._remoteHost).toBe("chernobog"); });
it("exposes remoteHost via session.remoteHost getter", () => { const root = createSession({ remoteHost: "chernobog" }); expect(root.remoteHost).toBe("chernobog"); });
it("inherits parent remoteHost for subagents when no host is specified", async () => { const root = createSession({ remoteHost: "chernobog", streamFn: () => streamAnswer("child finished"), });
const handle = await spawnChild(root, "child task"); const activeRun = (root as any)._activeRlmChildRuns.get(handle.rlm_child_id); expect(activeRun).toBeDefined(); expect((activeRun.session as any)._remoteHost).toBe("chernobog"); });
it("allows subagent to opt out of remote execution via host='local'", async () => { const root = createSession({ remoteHost: "chernobog", streamFn: () => streamAnswer("child finished"), });
const handle = await spawnChild(root, "local task", { host: "local" }); const activeRun = (root as any)._activeRlmChildRuns.get(handle.rlm_child_id); expect(activeRun).toBeDefined(); expect((activeRun.session as any)._remoteHost).toBeUndefined(); });
it("cancels active rlm children when the parent session is disposed", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); if (text === "slow shard") { childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); } return stream; }, });
const spawned = await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const runs = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; expect(runs.size).toBe(1); const run = [...runs.values()][0];
root.dispose();
expect(run.status).toBe("cancelled"); expect(run.error).toBe("Parent session disposed"); releaseChild(); expect(spawned.rlm_child_id).toBe(run.id); });
it("cancels active rlm children when the parent session is aborted", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); if (text === "slow shard") { childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); } return stream; }, });
const spawned = await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const runs = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; expect(runs.size).toBe(1); const run = [...runs.values()][0];
await root.abort();
expect(run.status).toBe("cancelled"); expect(run.error).toBe("Parent session aborted"); releaseChild(); expect(spawned.rlm_child_id).toBe(run.id); });
it("does not admit a child prompt when the parent is aborted while resolving the sender", async () => { let releaseAgentList: () => void = () => {}; const agentListGate = new Promise<void>((resolve) => { releaseAgentList = resolve; }); let agentListStarted = false; const child = createSession({ rlmSessionDir: join(tempDir, "pre-admission-child") }); const promptAndWait = vi.spyOn(child, "promptAndWait"); const root = createSession({ agentMessageController: { assertSessionNameAvailable: () => {}, listAgents: async () => { agentListStarted = true; await agentListGate; return { current: { activeSessionId: "parent-active", sessionId: "parent-session" }, agents: [], }; }, sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async () => { await child.disposeAsync(); }, }, });
const spawned = await spawnChild(root, "cancel before admission", { name: "cancelled-worker" }); await waitFor(() => agentListStarted); const runs = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; const run = runs.get(spawned.rlm_child_id); if (!run) throw new Error("Missing running child");
await root.abort(); releaseAgentList(); await waitFor(() => !runs.has(spawned.rlm_child_id));
expect(run.status).toBe("cancelled"); expect(run.error).toBe("Parent session aborted"); expect(promptAndWait).not.toHaveBeenCalled(); });
it("cancels strong quiescence before abandoning children for an update restart", async () => { const deferred = () => { let resolve!: () => void; let reject!: (error: Error) => void; const promise = new Promise<void>((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; }); promise.catch(() => undefined); return { promise, resolve, reject }; }; const child = createSession({ rlmSessionDir: join(tempDir, "update-restart-parent") }); const childInternals = child as unknown as InspectableRlmSession; childInternals._activeRlmChildRuns.set("live-grandchild", { id: "live-grandchild", prompt: "still working", sessionName: "live-grandchild", sessionDir: join(tempDir, "live-grandchild"), model, abort: () => {}, status: "running", settled: false, progressNotes: [], }); const root = createSession(); const rootInternals = root as unknown as InspectableRlmSession; const run: InspectableRlmRun = { id: "update-restart-parent", prompt: "parent work", sessionName: "update-restart-parent", sessionDir: join(tempDir, "update-restart-parent"), model, abort: () => {}, status: "running", settled: false, progressNotes: [], publication: deferred(), settlement: deferred(), session: child, }; rootInternals._activeRlmChildRuns.set(run.id, run); rootInternals._unsettledRlmChildRuns.add(run);
const quiescence = root.waitForRlmQuiescence(); await Promise.resolve(); root.abortForUpdateRestart();
await expect(quiescence).rejects.toThrow("RLM quiescence wait cancelled"); expect(run.abandonedForQuiescence).toBe(true); expect(root.getRlmChildSnapshots()).toEqual( expect.arrayContaining([expect.objectContaining({ id: "live-grandchild", status: "running" })]), );
rootInternals._activeRlmChildRuns.clear(); rootInternals._unsettledRlmChildRuns.clear(); childInternals._activeRlmChildRuns.clear(); root.dispose(); child.dispose(); });
it("does not carry an abandoned queued child into the next strong quiescence lifecycle", async () => { let releaseStartup: () => void = () => {}; const startupGate = new Promise<void>((resolve) => { releaseStartup = resolve; }); const child = createSession({ rlmSessionDir: join(tempDir, "abandoned-queued-child") }); const promptAndWait = vi.spyOn(child, "promptAndWait"); const root = createSession({ streamFn: () => { const stream = createAssistantMessageEventStream(); stream.push({ type: "done", reason: "stop", message: assistantMessage("next lifecycle done") }); return stream; }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { await startupGate; return { session: child }; }, deleteRlmSubagentRuntime: async () => {}, }, });
const spawned = await spawnChild(root, "blocked startup"); root.requestAbort(); expect(root.cancelRlmChildRun(spawned.rlm_child_id)).toBe(true); await expect(root.waitForRlmQuiescence()).resolves.toBeUndefined(); root.resumeQueuedWork(); await root.prompt("next lifecycle"); await expect(root.waitForRlmQuiescence()).resolves.toBeUndefined();
releaseStartup(); await waitFor(() => !(root as unknown as InspectableRlmSession)._activeRlmChildRuns.has(spawned.rlm_child_id)); expect(promptAndWait).not.toHaveBeenCalled(); });
it("does not cancel active rlm children when only the parent turn is interrupted", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); if (text === "slow shard") { childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); } return stream; }, });
await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const runs = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; expect(runs.size).toBe(1); const run = [...runs.values()][0];
root.requestAbort();
expect(run.status).toBe("running"); expect(run.error).toBeUndefined(); releaseChild(); await waitFor(() => run.status === "done"); });
it("cancels a single rlm child run by id and reports unknown ids", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); if (text === "slow shard") { childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); } return stream; }, }); const childStatuses: string[] = []; root.subscribe((event) => { if (event.type === "rlm_child_update") { childStatuses.push(event.child.status); } });
await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const runs = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; expect(runs.size).toBe(1); const childId = [...runs.keys()][0]; if (!childId) { throw new Error("Missing child run id"); } const run = runs.get(childId);
expect(root.cancelRlmChildRun("unknown-child")).toBe(false); expect(run?.status).toBe("running");
// Running work retained under the LIVE child: the abort cascade only // reaches active runs, so the cancel walk must descend here itself. await waitFor(() => run?.session !== undefined); const deepHost = createSession({ rlmSessionDir: join(tempDir, "deep-host") }); const deepAbort = vi.fn(); const deepRun = { id: "deep-1", status: "running", settled: false, abort: deepAbort, publication: { reject: vi.fn() }, emitUpdate: vi.fn(), }; (deepHost as unknown as { _activeRlmChildRuns: Map<string, typeof deepRun> })._activeRlmChildRuns.set( "deep-1", deepRun, ); (run?.session as unknown as { _rlmChildSessions: Map<string, { session: AgentSession }> })._rlmChildSessions.set( "deep-host", { session: deepHost }, );
expect(root.cancelRlmChildRun(childId)).toBe(true); expect(run?.status).toBe("cancelled"); expect(run?.error).toBe("Cancelled by user"); expect(deepRun.status).toBe("cancelled"); expect(deepAbort).toHaveBeenCalled(); // The cancelled update is pushed at cancel time, before the (possibly // stuck) child unwinds; viewers must not keep showing a running child. expect(childStatuses[childStatuses.length - 1]).toBe("cancelled"); releaseChild(); await waitFor(() => !runs.has(childId)); expect(childStatuses[childStatuses.length - 1]).toBe("cancelled"); expect(await root.listRlmSubagents()).toEqual({ subagents: [] });
// The run has finished; a second cancel finds nothing to stop. expect(root.cancelRlmChildRun(childId)).toBe(false); });
it("keeps a colliding child id reachable past a finished retained match", async () => { const root = createSession({ rlmSessionDir: join(tempDir, "collide-root") }); const finished = createSession({ rlmSessionDir: join(tempDir, "collide-finished") }); const otherParent = createSession({ rlmSessionDir: join(tempDir, "collide-other") }); const rootMaps = root as unknown as { _rlmChildSessions: Map<string, { session: AgentSession }> }; // Child ids are only mkdir-unique among siblings: "sub-dup" exists twice. rootMaps._rlmChildSessions.set("sub-dup", { session: finished }); rootMaps._rlmChildSessions.set("other-parent", { session: otherParent }); const abort = vi.fn(); const collidingRun = { id: "sub-dup", status: "running", settled: false, abort, publication: { reject: vi.fn() }, emitUpdate: vi.fn(), }; (otherParent as unknown as { _activeRlmChildRuns: Map<string, typeof collidingRun> })._activeRlmChildRuns.set( "sub-dup", collidingRun, );
expect(root.cancelRlmChildRun("sub-dup")).toBe(true); expect(collidingRun.status).toBe("cancelled"); expect(abort).toHaveBeenCalled(); });
it("cancels a deep dual-membership chain in one visit per session", async () => { const levels = 20; const sessions = Array.from({ length: levels + 1 }, (_, level) => createSession({ rlmSessionDir: join(tempDir, `chain-${level}`) }), ); let cancelPrimitiveCalls = 0; let runMapIterations = 0; for (const [level, session] of sessions.entries()) { const target = session as unknown as { _activeRlmChildRuns: Map<string, unknown>; _rlmChildSessions: Map<string, { session: AgentSession }>; _cancelRlmChildRun(run: unknown, reason: string): boolean; }; const original = target._cancelRlmChildRun.bind(session); target._cancelRlmChildRun = (run, reason) => { cancelPrimitiveCalls++; return original(run, reason); }; const originalValues = target._activeRlmChildRuns.values.bind(target._activeRlmChildRuns); target._activeRlmChildRuns.values = () => { runMapIterations++; return originalValues(); }; if (level === 0) continue; // A finished intermediate lives in BOTH parent maps until passivation. const parent = sessions[level - 1] as unknown as { _activeRlmChildRuns: Map<string, unknown>; _rlmChildSessions: Map<string, { session: AgentSession }>; }; parent._activeRlmChildRuns.set(`chain-${level}`, { id: `chain-${level}`, status: "done", settled: true, session, abort: vi.fn(), publication: { reject: vi.fn() }, emitUpdate: vi.fn(), }); parent._rlmChildSessions.set(`chain-${level}`, { session }); } const leafAbort = vi.fn(); const leafRun = { id: "leaf-run", status: "running", settled: false, abort: leafAbort, publication: { reject: vi.fn() }, emitUpdate: vi.fn(), }; (sessions[levels] as unknown as { _activeRlmChildRuns: Map<string, typeof leafRun> })._activeRlmChildRuns.set( "leaf-run", leafRun, );
expect(sessions[0]!.hasRunningRlmChildren()).toBe(true); expect(runMapIterations).toBeLessThanOrEqual(3 * (levels + 1));
expect(sessions[0]!.cancelRunningRlmDescendants()).toBe(true); expect(leafRun.status).toBe("cancelled"); expect(leafAbort).toHaveBeenCalled(); // One visit per session, not 2^depth. expect(cancelPrimitiveCalls).toBeLessThanOrEqual(levels + 1); expect(sessions[0]!.hasRunningRlmChildren()).toBe(false); });
it("stops live descendants when the targeted child run already settled", async () => { const root = createSession({ streamFn: (_model, context) => { const stream = createAssistantMessageEventStream(); stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${userText(context)}`), }); return stream; }, });
await spawnChild(root, "quick shard"); const runs = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; await waitFor(() => runs.size === 0); const retained = (root as unknown as { _rlmChildSessions: Map<string, { session: AgentSession }> }) ._rlmChildSessions; expect(retained.size).toBe(1); const [childId, { session: childSession }] = [...retained.entries()][0]!; const abort = vi.fn(); const grandchild = { id: "grandchild-1", status: "running", settled: false, abort, publication: { reject: vi.fn() }, emitUpdate: vi.fn(), }; (childSession as unknown as { _activeRlmChildRuns: Map<string, typeof grandchild> })._activeRlmChildRuns.set( "grandchild-1", grandchild, );
expect(root.cancelRlmChildRun(childId)).toBe(true); expect(grandchild.status).toBe("cancelled"); expect(abort).toHaveBeenCalled(); });
it("reports a shared running outcome to concurrent inactive-delete callers", async () => { let runningChecks = 0; const isExternallyRunning = () => ++runningChecks >= 5; const deleteRuntime = vi.fn(async () => {}); const root = createSession({ agentMessageController: { listAgents: () => ({ current: { activeSessionId: "parent-active", sessionId: "parent-session" }, agents: [ { activeSessionId: "child-active", sessionId: "child-session", sessionName: "worker", runtimeKind: "subagent", cwd: tempDir, isStreaming: false, unfinishedActionCount: 0, parentActiveSessionId: "parent-active", rlmChildId: "child", sessionDir: join(tempDir, "child"), }, ], }), sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("unexpected hydration"); }, deleteRlmSubagentRuntime: deleteRuntime, }, });
const first = root.deleteInactiveRlmSubagent("child", isExternallyRunning); const second = root.deleteInactiveRlmSubagent("child", isExternallyRunning);
await expect(Promise.all([first, second])).resolves.toEqual(["running", "running"]); expect(deleteRuntime).not.toHaveBeenCalled(); });
it("deletes only inactive RLM children through the explicit inactive path", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const retainedChild = createSession({ rlmSessionDir: join(tempDir, "retained-child"), streamFn: () => { const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("done") }); }); return stream; }, }); const deleteRuntime = vi.fn(async () => {}); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: retainedChild }), deleteRlmSubagentRuntime: deleteRuntime, releaseRlmSubagentRuntime: async (runtime, options) => { options.parentSession.registerRlmChildSession(options.id, runtime.session); }, }, });
const runPromise = spawnChild(root, "slow child", { name: "retained-worker" }); await waitFor(() => childStarted); const childId = [...(root as unknown as InspectableRlmSession)._activeRlmChildRuns.keys()][0]!; await expect(root.deleteInactiveRlmSubagent(childId)).resolves.toBe("running"); expect(deleteRuntime).not.toHaveBeenCalled();
releaseChild(); await expect(runPromise).resolves.toMatchObject({ name: "retained-worker" }); await waitFor( () => (root as unknown as InspectableRlmSession)._activeRlmChildRuns.get(childId)?.status !== "running", ); await expect(root.deleteInactiveRlmSubagent(childId)).resolves.toBe("deleted"); expect(deleteRuntime).toHaveBeenCalledWith(childId, retainedChild); await expect(root.deleteInactiveRlmSubagent("unknown-child")).resolves.toBe("not_found"); });
it("releases a hosted child when completion persistence fails", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "host-completion-failure-child") }); const disposeChild = vi.spyOn(child, "disposeAsync"); const releaseRlmSubagentRuntime = vi.fn(async () => {}); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), completeRlmSubagentRuntime: () => false, releaseRlmSubagentRuntime, deleteRlmSubagentRuntime: async () => {}, }, });
const spawned = await spawnChild(root, "finish without registry persistence"); await vi.waitFor(() => { expect(releaseRlmSubagentRuntime).toHaveBeenCalledWith( expect.objectContaining({ session: child }), expect.objectContaining({ id: spawned.rlm_child_id }), "error", ); }); expect(disposeChild).not.toHaveBeenCalled(); expect((root as unknown as InspectableRlmSession)._activeRlmChildRuns.size).toBe(0); expect(root.getRlmChildSession(spawned.rlm_child_id)).toBeUndefined(); });
it("aborts an active child tool and settles only after shared runtime cleanup", async () => { let toolStarted = false; let toolAborted = false; const tool = { name: "blocking_tool", description: "Block until cancellation", label: "blocking tool", parameters: Type.Object({}), execute: async (_toolCallId: string, _params: Record<string, never>, signal: AbortSignal) => { toolStarted = true; await new Promise<never>((_resolve, reject) => { const rejectAborted = () => { toolAborted = true; reject(new Error("tool aborted")); }; if (signal.aborted) rejectAborted(); else signal.addEventListener("abort", rejectAborted, { once: true }); }); throw new Error("unreachable"); }, }; const hostedChild = createSession({ customTools: [tool], streamFn: (_model, context) => { const stream = createAssistantMessageEventStream(); queueMicrotask(() => { const hasToolResult = context.messages.some((message) => message.role === "toolResult"); stream.push({ type: "done", reason: hasToolResult ? "stop" : "toolUse", message: hasToolResult ? assistantMessage("unexpected completion") : { ...assistantMessage(""), content: [{ type: "toolCall" as const, id: "blocking-1", name: tool.name, arguments: {} }], stopReason: "toolUse" as const, }, }); }); return stream; }, }); let releaseCleanup: () => void = () => {}; const cleanupGate = new Promise<void>((resolve) => { releaseCleanup = resolve; }); let cleanupStarted = false; const deleteRuntime = vi.fn(async () => { cleanupStarted = true; await cleanupGate; await hostedChild.disposeAsync(); }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: deleteRuntime, }, });
const spawned = await spawnChild(root, "use the blocking tool", { name: "tool-worker" }); await waitFor(() => toolStarted); const firstDeletion = root.deleteRlmSubagent("tool-worker"); const repeatedDeletion = root.deleteRlmSubagent(spawned.rlm_child_id); await expect(firstDeletion).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id } }); await expect(repeatedDeletion).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id } }); await waitFor(() => toolAborted && cleanupStarted); expect(deleteRuntime).toHaveBeenCalledOnce();
let quiesced = false; const quiescence = root.waitForRlmQuiescence().then(() => { quiesced = true; }); await sleep(20); expect(quiesced).toBe(false); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0);
releaseCleanup(); await quiescence; expect(deleteRuntime).toHaveBeenCalledOnce(); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toEqual([ expect.objectContaining({ details: expect.objectContaining({ kind: "cancelled", reason: "Deleted by parent orchestrator" }), }), ]); });
it("admits a private durable deletion notice only after runtime cleanup", async () => { let releaseChild: () => void = () => {}; const childGate = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; let cleanupSucceeded = false; const sendAgentMessage = vi.fn(async () => { expect(cleanupSucceeded).toBe(true); return { id: "agentmsg-deletion-complete", source: "agent_message" as const, target: { activeSessionId: "parent-active", sessionId: "parent-session" }, message: "deleted", deliveryStatus: "delivered" as const, }; }); const hostedChild = createSession({ agentMessageController: { listAgents: () => ({ current: { activeSessionId: "child-active", sessionId: "child-session" }, agents: [], }), sendAgentMessage, }, streamFn: () => { const stream = createAssistantMessageEventStream(); childStarted = true; void childGate.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("child stopped") }); }); return stream; }, }); vi.spyOn(hostedChild, "abort").mockResolvedValue(); let releaseCleanup: () => void = () => {}; const cleanupGate = new Promise<void>((resolve) => { releaseCleanup = resolve; }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: () => cleanupGate, }, });
const spawned = await spawnChild(root, "controller ordering", { name: "controller-worker" }); await waitFor(() => childStarted); await root.deleteRlmSubagent(spawned.rlm_child_id); releaseChild(); await sleep(20); expect(sendAgentMessage).not.toHaveBeenCalled(); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0);
cleanupSucceeded = true; releaseCleanup(); await root.waitForRlmQuiescence(); expect(sendAgentMessage).not.toHaveBeenCalled(); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toEqual([ expect.objectContaining({ content: expect.stringContaining("[child-exited: cancelled"), details: expect.objectContaining({ reason: "Deleted by parent orchestrator" }), }), ]); });
it("preserves failed cleanup retry across transient preflight failure before abort-insensitive unwind", async () => { const { child: hostedChild, completion: childCompletion, hasStarted } = createAbortInsensitiveChild(); const retryCleanup = deferred<void>(); let cleanupAttempts = 0; let failNextDeletePreflight = false; const root = createSession({ agentMessageController: { listAgents: async () => { if (failNextDeletePreflight) { failNextDeletePreflight = false; throw new Error("delete preflight failed"); } return { current: { activeSessionId: "parent-active", sessionId: "parent-session" }, agents: [], }; }, sendAgentMessage: async () => { throw new Error("unexpected direct send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: () => { if (++cleanupAttempts === 1) throw new Error("cleanup failed synchronously"); return retryCleanup.promise; }, }, });
const spawned = await spawnChild(root, "abort-insensitive cleanup retry", { name: "retry-worker" }); await waitFor(hasStarted); await expect(root.deleteRlmSubagent(spawned.rlm_child_id)).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); const internals = root as unknown as InspectableRlmSession; await waitFor(() => internals._rlmChildCleanupFailures.size === 1); const failureContent = await vi.waitFor(() => { const notice = root.messages.find( (message) => message.role === "custom" && message.customType === "rlm_child_failure", ); if (!notice || notice.role !== "custom") throw new Error("Missing cleanup failure notice"); return notice.content; }); expect(failureContent).toEqual(expect.stringContaining("Deletion cleanup failed")); expect(failureContent).toEqual(expect.stringMatching(/retry/i));
failNextDeletePreflight = true; await expect(root.deleteRlmSubagent("retry-worker")).rejects.toThrow("delete preflight failed"); expect(cleanupAttempts).toBe(1);
await expect(root.deleteRlmSubagent("retry-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); expect(cleanupAttempts).toBe(2); let quiesced = false; const quiescence = root.waitForRlmQuiescence().then(() => { quiesced = true; }); retryCleanup.resolve(); await sleep(20); expect(quiesced).toBe(false);
childCompletion.resolve(); await quiescence; expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(1); expect(internals._rlmChildCleanupFailures.size).toBe(0); await expect(spawnChild(root, "replacement", { name: "retry-worker" })).resolves.toMatchObject({ name: "retry-worker", }); });
it("reports one failure per delete attempt and one terminal notice after a successful retry", async () => { const { child: hostedChild, completion: childCompletion, hasStarted } = createAbortInsensitiveChild(); const cleanups = [deferred<void>(), deferred<void>(), deferred<void>()]; let cleanupAttempts = 0; const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: () => cleanups[cleanupAttempts++]!.promise, }, }); const spawned = await spawnChild(root, "retry cleanup", { name: "notice-worker" }); await waitFor(hasStarted); const internals = root as unknown as InspectableRlmSession; const failures = () => root.messages.filter((message) => message.role === "custom" && message.customType === "rlm_child_failure"); const terminalNotices = () => root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", );
for (let attempt = 0; attempt < 2; attempt++) { await Promise.all([root.deleteRlmSubagent(spawned.rlm_child_id), root.deleteRlmSubagent("notice-worker")]); expect(cleanupAttempts).toBe(attempt + 1); cleanups[attempt]!.reject(new Error(`cleanup failure ${attempt + 1}`)); await vi.waitFor(() => { expect(failures()).toHaveLength(attempt + 1); expect(internals._deletingRlmChildren.size).toBe(0); }); expect(terminalNotices()).toHaveLength(0); expect((await root.listRlmSubagents()).subagents).toEqual([]); }
await root.deleteRlmSubagent("notice-worker"); expect(cleanupAttempts).toBe(3); childCompletion.resolve(); cleanups[2]!.resolve(); await root.waitForRlmQuiescence(); expect(failures()).toHaveLength(2); expect(terminalNotices()).toEqual([ expect.objectContaining({ details: expect.objectContaining({ kind: "cancelled" }) }), ]); expect(internals._rlmChildCleanupFailures.size).toBe(0); expect(root.getRlmChildSession(spawned.rlm_child_id)).toBeUndefined(); await hostedChild.disposeAsync(); });
it("settles a pre-existing deletion cleanup failure during parent disposal", async () => { const { child: hostedChild, completion: childCompletion, hasStarted } = createAbortInsensitiveChild(); const disposeHostedChild = vi.spyOn(hostedChild, "disposeAsync"); const deleteRuntime = vi.fn(() => Promise.reject(new Error("cleanup failed before dispose"))); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: deleteRuntime, }, });
const spawned = await spawnChild(root, "pre-failed cleanup disposal", { name: "pre-failed-worker" }); await waitFor(hasStarted); await root.deleteRlmSubagent(spawned.rlm_child_id); const internals = root as unknown as InspectableRlmSession; const run = internals._activeRlmChildRuns.get(spawned.rlm_child_id); if (!run) throw new Error("Missing deleting run"); await waitFor(() => internals._rlmChildCleanupFailures.size === 1);
await root.disposeAsync(); expect(deleteRuntime).toHaveBeenCalledOnce(); expect(disposeHostedChild).toHaveBeenCalled(); expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(false); expect(internals._unsettledRlmChildRuns.has(run)).toBe(false); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0); childCompletion.resolve(); });
it("settles disposal after cleanup succeeds without waiting for abort-insensitive task unwind", async () => { const { child: hostedChild, completion: childCompletion, hasStarted } = createAbortInsensitiveChild(); const cleanup = deferred<void>(); const deleteRuntime = vi.fn(() => cleanup.promise); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: deleteRuntime, }, });
const spawned = await spawnChild(root, "successful cleanup disposal", { name: "dispose-success-worker" }); await waitFor(hasStarted); await root.deleteRlmSubagent(spawned.rlm_child_id); const internals = root as unknown as InspectableRlmSession; const run = internals._activeRlmChildRuns.get(spawned.rlm_child_id); if (!run) throw new Error("Missing deleting run");
const disposal = root.disposeAsync(); await waitFor(() => internals._disposing); cleanup.resolve(); await disposal; expect(deleteRuntime).toHaveBeenCalledOnce(); expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(false); expect(internals._unsettledRlmChildRuns.has(run)).toBe(false); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), ).toHaveLength(0); childCompletion.resolve(); });
it("settles and suppresses notices when deletion cleanup fails during parent disposal", async () => { const { child: hostedChild, completion: childCompletion, hasStarted } = createAbortInsensitiveChild(); const disposeHostedChild = vi.spyOn(hostedChild, "disposeAsync"); const cleanup = deferred<void>(); const deleteRuntime = vi.fn(() => cleanup.promise); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: deleteRuntime, }, });
const spawned = await spawnChild(root, "dispose during failed cleanup", { name: "dispose-worker" }); await waitFor(hasStarted); await root.deleteRlmSubagent(spawned.rlm_child_id); const internals = root as unknown as InspectableRlmSession; const run = internals._activeRlmChildRuns.get(spawned.rlm_child_id); if (!run) throw new Error("Missing deleting run");
const disposal = root.disposeAsync(); await waitFor(() => internals._disposing); cleanup.reject(new Error("cleanup failed during dispose")); childCompletion.resolve(); await disposal; expect(deleteRuntime).toHaveBeenCalledOnce(); expect(disposeHostedChild).toHaveBeenCalled(); expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(false); expect(internals._unsettledRlmChildRuns.has(run)).toBe(false); expect( root.messages.filter( (message) => message.role === "custom" && (message.customType === "rlm_child_failure" || message.customType === "rlm_child_terminal_notice"), ), ).toHaveLength(0); });
it("keeps deleted live runs in the RLM quiescence barrier until settlement", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const hostedChild = createSession({ rlmSessionDir: join(tempDir, "deleted-live-child"), streamFn: () => { const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("child stopped") }); }); return stream; }, }); vi.spyOn(hostedChild, "abort").mockResolvedValue(); const root = createSession({ streamFn: () => { const stream = createAssistantMessageEventStream(); stream.push({ type: "done", reason: "stop", message: assistantMessage("parent acknowledged") }); return stream; }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: hostedChild }), deleteRlmSubagentRuntime: async () => {}, }, });
const spawned = await spawnChild(root, "slow child", { name: "deleted-live-worker" }); await waitFor(() => childStarted); let quiesced = false; const quiescence = root.waitForRlmQuiescence().then(() => { quiesced = true; }); await root.deleteRlmSubagent(spawned.rlm_child_id); await sleep(20); expect(quiesced).toBe(false);
releaseChild(); await quiescence; expect(quiesced).toBe(true); });
it("does not add a cancellation notice when deletion races a durably admitted completion notice", async () => { const root = createSession(); const dispatchGate = vi .spyOn(root as unknown as { _scheduleSessionInputPump(): void }, "_scheduleSessionInputPump") .mockImplementation(() => {}); const spawned = await spawnChild(root, "fast child", { name: "fast-worker" }); const internals = root as unknown as InspectableRlmSession; await waitFor( () => root.getRlmChildSession(spawned.rlm_child_id) !== undefined && !internals._activeRlmChildRuns.has(spawned.rlm_child_id), ); expect(root.unfinishedActionCount).toBe(1);
await expect(root.deleteRlmSubagent("fast-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: spawned.rlm_child_id }, }); dispatchGate.mockRestore(); await root.waitForRlmQuiescence(); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); expect(root.getRlmChildSession(spawned.rlm_child_id)).toBeUndefined(); expect( root.messages.filter( (message) => message.role === "custom" && message.customType === "rlm_child_terminal_notice", ), // R11: the completion notice now carries the harvested text, tagged as unreviewed; // what matters here is that the deletion race added no second notice. ).toEqual([expect.objectContaining({ details: expect.objectContaining({ kind: "harvested_fallback" }) })]); expect(internals._activeRlmChildRuns.has(spawned.rlm_child_id)).toBe(false); });
it("keeps failed closure retryable without hanging or late resurrection", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "retry-child") }); child.setSessionName("release-worker"); let attempts = 0; const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async (_id, session) => { if (++attempts === 1) throw new Error("close failed"); await session?.disposeAsync(); }, }, }); expect(root.registerRlmChildSession("retry-child", child)).toBe(true); await expect(root.deleteRlmSubagent("release-worker")).rejects.toThrow("close failed"); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); expect((root as unknown as InspectableRlmSession)._rlmChildCleanupFailures.size).toBe(1); await expect(root.deleteRlmSubagent("release-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: "retry-child" }, }); expect((root as unknown as InspectableRlmSession)._rlmChildCleanupFailures.size).toBe(0); });
it("does not restore failed delete retry state after parent teardown", async () => { const child = createSession({ rlmSessionDir: join(tempDir, "teardown-child") }); child.setSessionName("teardown-worker"); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async () => { throw new Error("close failed during teardown"); }, }, }); expect(root.registerRlmChildSession("teardown-child", child)).toBe(true); await expect(root.deleteRlmSubagent("teardown-worker")).rejects.toThrow("close failed during teardown"); root.dispose(); const internals = root as unknown as InspectableRlmSession; expect(internals._activeRlmChildRuns.size).toBe(0); expect(internals._rlmChildSessions.size).toBe(0); expect(internals._rlmChildUnsubscribes.size).toBe(0); expect(internals._rlmChildCleanupFailures.size).toBe(0); });
it("keeps an errored startup deletable after its failure notice is durably admitted", async () => { const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { throw new Error("startup failed"); }, deleteRlmSubagentRuntime: async () => undefined, }, });
await spawnChild(root, "failing startup", { name: "failed-worker" }); await vi.waitFor(async () => { expect((await root.listRlmSubagents()).subagents[0]).toMatchObject({ session_name: "failed-worker", status: "error", }); }); const failed = (await root.listRlmSubagents()).subagents[0]; await expect(root.deleteRlmSubagent("failed-worker")).resolves.toEqual({ subagent: failed }); const internals = root as unknown as InspectableRlmSession; expect(internals._activeRlmChildRuns.size).toBe(0); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); await expect(spawnChild(root, "replacement", { name: "failed-worker" })).resolves.toMatchObject({ name: "failed-worker", }); });
it("reserves a deleted queued child's name until startup settles", async () => { let releaseRuntimeCreation: () => void = () => {}; const runtimeCreationGate = new Promise<void>((resolve) => { releaseRuntimeCreation = resolve; }); let runtimeCreationStarted = false; const hostedChild = createSession(); const setSessionName = vi.spyOn(hostedChild, "setSessionName"); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { runtimeCreationStarted = true; await runtimeCreationGate; return { session: hostedChild }; }, deleteRlmSubagentRuntime: async (_id, child) => child?.disposeAsync(), }, }); await spawnChild(root, "blocked before runtime creation", { name: "reserved-worker" }); await waitFor(() => runtimeCreationStarted);
await root.deleteRlmSubagent("reserved-worker"); await expect(spawnChild(root, "replacement", { name: "reserved-worker" })).rejects.toThrow( "an agent of that name already exists at depth 1 under this parent", );
releaseRuntimeCreation(); await waitFor(() => (root as unknown as InspectableRlmSession)._activeRlmChildRuns.size === 0); expect(setSessionName).not.toHaveBeenCalled(); await expect(spawnChild(root, "replacement", { name: "reserved-worker" })).resolves.toMatchObject({ name: "reserved-worker", }); });
it("deletes a queued child without waiting for blocked startup", async () => { let releaseRuntimeCreation: () => void = () => {}; const runtimeCreationGate = new Promise<void>((resolve) => { releaseRuntimeCreation = resolve; }); let runtimeCreationStarted = false; const hostedChild = createSession(); const deleteRuntime = vi.fn(async () => { await hostedChild.disposeAsync(); }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { runtimeCreationStarted = true; await runtimeCreationGate; return { session: hostedChild }; }, deleteRlmSubagentRuntime: deleteRuntime, }, }); await spawnChild(root, "blocked before runtime creation", { name: "queued-worker" }); await waitFor(() => runtimeCreationStarted); const queued = (await root.listRlmSubagents()).subagents[0]; expect(queued).toBeDefined();
await expect(root.deleteRlmSubagent("queued-worker")).resolves.toEqual({ subagent: queued }); expect((root as unknown as InspectableRlmSession)._activeRlmChildRuns.size).toBe(1); expect(await root.listRlmSubagents()).toEqual({ subagents: [] });
releaseRuntimeCreation(); await waitFor(() => deleteRuntime.mock.calls.length === 1); await waitFor(() => (root as unknown as InspectableRlmSession)._activeRlmChildRuns.size === 0); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); });
it("keeps late startup cleanup retryable when release and fallback delete fail", async () => { let releaseRuntimeCreation: () => void = () => {}; const runtimeCreationGate = new Promise<void>((resolve) => { releaseRuntimeCreation = resolve; }); let runtimeCreationStarted = false; const hostedChild = createSession(); const disposeHostedChild = vi.spyOn(hostedChild, "disposeAsync"); let deleteAttempts = 0; const deleteRuntime = vi.fn(async () => { deleteAttempts++; if (deleteAttempts === 1) { throw new Error("fallback delete failed"); } await hostedChild.disposeAsync(); }); const releaseRuntime = vi.fn(async () => { throw new Error("cancelled release failed"); }); const root = createSession({ subagentRuntimeHost: { createRlmSubagentRuntime: async () => { runtimeCreationStarted = true; await runtimeCreationGate; return { session: hostedChild }; }, deleteRlmSubagentRuntime: deleteRuntime, releaseRlmSubagentRuntime: releaseRuntime, }, });
await spawnChild(root, "delete during runtime creation", { name: "starting-worker" }); await waitFor(() => runtimeCreationStarted); const starting = (await root.listRlmSubagents()).subagents[0]; expect(starting).toBeDefined();
await expect(root.deleteRlmSubagent("starting-worker")).resolves.toEqual({ subagent: starting }); expect(deleteRuntime).not.toHaveBeenCalled(); releaseRuntimeCreation();
const internals = root as unknown as InspectableRlmSession; await waitFor(() => deleteRuntime.mock.calls.length === 1); await waitFor(() => internals._rlmChildCleanupFailures.size === 1); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); expect(disposeHostedChild).not.toHaveBeenCalled();
await root.deleteRlmSubagent("starting-worker"); await root.waitForRlmQuiescence(); expect(deleteRuntime).toHaveBeenCalledTimes(2); expect(disposeHostedChild).toHaveBeenCalledOnce(); expect(internals._rlmChildCleanupFailures.size).toBe(0); });
it("accepts deletion of a running direct child without waiting for task unwind", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let childStarted = false; const root = createSession({ streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); return stream; }, });
await spawnChild(root, "slow named shard", { name: "slow-worker" }); await waitFor(() => childStarted); const running = (await root.listRlmSubagents()).subagents[0]; if (!running) { throw new Error("Missing running child registry entry"); } await expect(root.deleteRlmSubagent("slow-worker")).resolves.toEqual({ subagent: running }); expect(await root.listRlmSubagents()).toEqual({ subagents: [] }); releaseChild(); });
it("deletes an inactive nested RLM child through the root session", async () => { let releaseParent: () => void = () => {}; const parentRelease = new Promise<void>((resolve) => { releaseParent = resolve; }); let parentStarted = false; const root = createSession({ maxDepth: 2, streamFn: (_model, context) => { const text = userText(context); if (text !== "slow parent") { return streamAnswer(`child answer: ${text}`); } const stream = createAssistantMessageEventStream(); parentStarted = true; void parentRelease.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage("parent done") }); }); return stream; }, });
const parentPromise = spawnChild(root, "slow parent"); await waitFor(() => parentStarted); const parentRun = [...(root as unknown as InspectableRlmSession)._activeRlmChildRuns.values()][0]; if (!parentRun?.session) { throw new Error("Missing parent child session"); } const parentSession = parentRun.session; const nestedResult = await spawnChild(parentSession, "nested child"); if (!nestedResult.session_dir) { throw new Error("Missing nested child session directory"); } const nestedId = basename(nestedResult.session_dir); const nestedSession = parentSession.getRlmChildSession(nestedId); if (!nestedSession) { throw new Error("Missing retained nested child session"); } const disposeNested = vi.spyOn(nestedSession, "disposeAsync"); await waitFor(() => { const status = (parentSession as unknown as InspectableRlmSession)._activeRlmChildRuns.get(nestedId)?.status; return status !== "queued" && status !== "running"; });
await expect(root.deleteInactiveRlmSubagent(nestedId)).resolves.toBe("deleted"); expect(await parentSession.listRlmSubagents()).toEqual({ subagents: [] }); expect(disposeNested).toHaveBeenCalledOnce(); expect(parentRun.status).toBe("running");
releaseParent(); await parentPromise; await waitFor(() => { const status = (root as unknown as InspectableRlmSession)._activeRlmChildRuns.get(parentRun.id)?.status; return status === undefined || status === "done"; }); });
it("cancels nested rlm child runs through the root session", async () => { let releaseChild: () => void = () => {}; const release = new Promise<void>((resolve) => { releaseChild = resolve; }); let releaseNested: () => void = () => {}; const nestedRelease = new Promise<void>((resolve) => { releaseNested = resolve; }); let childStarted = false; let nestedStarted = false; const root = createSession({ maxDepth: 2, streamFn: (_model, context) => { const text = userText(context); const stream = createAssistantMessageEventStream(); if (text === "slow shard") { childStarted = true; void release.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); } else if (text === "nested shard") { nestedStarted = true; void nestedRelease.then(() => { stream.push({ type: "done", reason: "stop", message: assistantMessage(`child answer: ${text}`) }); }); } else { stream.push({ type: "done", reason: "stop", message: assistantMessage("cancellation acknowledged") }); } return stream; }, });
await spawnChild(root, "slow shard"); await waitFor(() => childStarted); const rootRuns = (root as unknown as InspectableRlmSession)._activeRlmChildRuns; const rootRun = [...rootRuns.values()][0]; if (!rootRun?.session) { throw new Error("Missing child session on root run"); }
const childSession = rootRun.session; const nestedSpawned = await spawnChild(childSession, "nested shard"); await waitFor(() => nestedStarted); const nestedRuns = (childSession as unknown as InspectableRlmSession)._activeRlmChildRuns; expect(nestedRuns.size).toBe(1); const nestedId = [...nestedRuns.keys()][0]; if (!nestedId) { throw new Error("Missing nested run id"); }
await expect(root.deleteRlmSubagent(nestedId)).rejects.toThrow("No direct RLM subagent matches"); expect(root.cancelRlmChildRun(nestedId)).toBe(true); releaseNested(); await waitFor(() => !nestedRuns.has(nestedId)); expect(nestedSpawned.rlm_child_id).toBe(nestedId); expect(rootRun.status).toBe("running");
releaseChild(); await waitFor(() => rootRun.status === "done"); });
it("handles rlm calls from asyncio tasks after the scheduling cell is idle", async () => { const prompts: string[] = []; const manager = new ReplKernelManager({ cwd: tempDir, hostHandlers: { "rlm.models": async () => ({}), "rlm.run": createRlmRunHostHandler(async ({ prompt }) => { prompts.push(prompt); return { rlm_child_id: "sub-detached", name: "detached-worker", session_dir: "/tmp/sub-detached", model: "test/model", }; }), }, });
try { const scheduled = await manager.execute(`import asyncioimport rlm
async def _delayed_rlm(): await asyncio.sleep(0.05) await rlm.models() return await rlm.spawn("detached child after idle", name="detached-worker", model="deepseek/deepseek-v4-flash")
_task = asyncio.create_task(_delayed_rlm())print("scheduled")`);
expect(scheduled.status).toBe("ok"); expect(scheduled.stdout.trim()).toBe("scheduled"); await waitFor(() => prompts.some((prompt) => prompt.includes("TASK:\ndetached child after idle")));
const finished = await manager.execute(`_result = await _taskprint(_result.name)`);
expect(finished.status).toBe("ok"); expect(finished.stdout.trim()).toBe("detached-worker"); } finally { await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });});
interface InspectableRlmDirSession { _ensureRlmSessionDir(): string | undefined; _rlmKernelEnv(): Record<string, string>;}
describe("AgentSession RLM session dir", () => { let tempDir: string; let session: AgentSession | undefined; let previousRlmDepth: string | undefined; let previousRlmMaxDepth: string | undefined;
beforeEach(() => { tempDir = join(tmpdir(), `pi-rlm-dir-${Date.now()}-${Math.random().toString(36).slice(2)}`); mkdirSync(tempDir, { recursive: true }); previousRlmDepth = process.env.RLM_DEPTH; delete process.env.RLM_DEPTH; previousRlmMaxDepth = process.env.RLM_MAX_DEPTH; delete process.env.RLM_MAX_DEPTH; });
afterEach(() => { session?.dispose(); session = undefined; rmSync(tempDir, { recursive: true, force: true }); if (previousRlmDepth === undefined) { delete process.env.RLM_DEPTH; } else { process.env.RLM_DEPTH = previousRlmDepth; } if (previousRlmMaxDepth === undefined) { delete process.env.RLM_MAX_DEPTH; } else { process.env.RLM_MAX_DEPTH = previousRlmMaxDepth; } });
function createSession( sessionManager: SessionManager, agentDir?: string, serperKey?: string, loadWebsearchSkill = false, rlmSessionDir?: string, ): AgentSession { const authStorage = AuthStorage.create(join(tempDir, "auth.json")); authStorage.setRuntimeApiKey("anthropic", "test-key"); if (serperKey !== undefined) { authStorage.set("serper", { type: "api_key", key: serperKey }); } const agent = new Agent({ convertToLlm, getApiKey: () => "test-key", initialState: { model, systemPrompt: "", tools: [], thinkingLevel: "off" }, streamFn: () => streamAnswer("ignored"), }); const skills: Skill[] = loadWebsearchSkill ? [ { kind: "markdown", name: "websearch", description: "", filePath: "/x/websearch/SKILL.md", baseDir: "/x/websearch", sourceInfo: createSyntheticSourceInfo("/x/websearch/SKILL.md", { source: "package" }), disableModelInvocation: false, }, ] : []; session = new AgentSession({ agent, sessionManager, settingsManager: SettingsManager.create(tempDir, tempDir), cwd: tempDir, agentDir, modelRegistry: ModelRegistry.create(authStorage, join(tempDir, "models.json")), resourceLoader: createTestResourceLoader({ skills }), rlmSessionDir, }); return session; }
it("does not create a /tmp dir or set RLM_SESSION_DIR for a non-persisted session", () => { const root = createSession(SessionManager.inMemory(tempDir)); const inspectable = root as unknown as InspectableRlmDirSession;
const before = readdirSync(tmpdir()).filter((name) => name.startsWith("prime-agent-rlm-"));
expect(inspectable._ensureRlmSessionDir()).toBeUndefined(); const env = inspectable._rlmKernelEnv(); expect(env.RLM_SESSION_DIR).toBeUndefined(); expect(env.RLM_HARNESS_STATE_DIR).toBeUndefined(); expect(env.RLM_GLOBAL_HARNESS_STATE_DIR).toBeDefined(); expect(env).toMatchObject({ RLM_DEPTH: "0" });
const after = readdirSync(tmpdir()).filter((name) => name.startsWith("prime-agent-rlm-")); expect(after).toEqual(before); });
it("uses the persistent artifact dir and sets RLM_SESSION_DIR for a persisted session", () => { const sessionManager = SessionManager.create(tempDir, join(tempDir, "sessions")); const root = createSession(sessionManager); const inspectable = root as unknown as InspectableRlmDirSession;
const artifactDir = sessionManager.getSessionArtifactDir(); expect(artifactDir).toBeDefined(); expect(inspectable._ensureRlmSessionDir()).toBe(artifactDir); expect(inspectable._rlmKernelEnv().RLM_SESSION_DIR).toBe(artifactDir); expect(inspectable._rlmKernelEnv().RLM_HARNESS_STATE_DIR).toBe(join(artifactDir!, "harness")); expect(inspectable._rlmKernelEnv().RLM_GLOBAL_HARNESS_STATE_DIR).toBeDefined(); });
it("points RLM_HARNESS_STATE_DIR at the session's own artifact dir for subagent sessions", () => { // Subagent layout: the parent assigns rlmSessionDir, but the child's own // sessionManager persists artifacts (and reads local harness state) elsewhere. const subDir = join(tempDir, "parent-artifact", "sub-abc12345"); mkdirSync(subDir, { recursive: true }); const sessionManager = SessionManager.create(tempDir, subDir); const root = createSession(sessionManager, undefined, undefined, false, subDir); const inspectable = root as unknown as InspectableRlmDirSession;
const artifactDir = sessionManager.getSessionArtifactDir(); expect(artifactDir).toBeDefined(); expect(artifactDir).not.toBe(subDir); const env = inspectable._rlmKernelEnv(); expect(env.RLM_SESSION_DIR).toBe(subDir); expect(env.RLM_HARNESS_STATE_DIR).toBe(join(artifactDir!, "harness")); });
it("falls back to the rlm session dir for RLM_HARNESS_STATE_DIR without an artifact dir", () => { const ephemeralDir = join(tempDir, "ephemeral-rlm"); mkdirSync(ephemeralDir, { recursive: true }); const root = createSession(SessionManager.inMemory(tempDir), undefined, undefined, false, ephemeralDir); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.RLM_SESSION_DIR).toBe(ephemeralDir); expect(env.RLM_HARNESS_STATE_DIR).toBe(join(ephemeralDir, "harness")); });
it("loads the ephemeral RLM harness path into the session-start harness digest", () => { const ephemeralDir = join(tempDir, "ephemeral-rlm"); mkdirSync(join(ephemeralDir, "harness"), { recursive: true }); writeFileSync( join(ephemeralDir, "harness", "harness_state.json"), JSON.stringify({ schema: 1, entries: { prompt: {}, memory: { ephemeral_note: { id: "ephemeral_note", kind: "memory", title: "Ephemeral note", content: "Loaded from the RLM session harness path.", path: "000", scope: "local", reference: {}, arguments: {}, metadata: {}, source: "test", created_at: new Date().toISOString(), updated_at: new Date().toISOString(), version: 1, }, }, skill: {}, subagent: {}, }, refinements: [], }), "utf8", ); const root = createSession(SessionManager.inMemory(tempDir), undefined, undefined, false, ephemeralDir);
const digest = (root as unknown as { _harnessDigest(): string })._harnessDigest();
expect(root.systemPrompt).not.toContain("Ephemeral note"); expect(digest).toContain("Ephemeral note"); expect(digest).toContain("Loaded from the RLM session harness path."); });
it("exports the configured agentDir to the kernel so skills find auth.json", () => { const agentDir = join(tempDir, "custom-agent-dir"); const root = createSession(SessionManager.inMemory(tempDir), agentDir); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.PRIME_AGENT_CODING_AGENT_DIR).toBe(agentDir); });
it("omits the agentDir env var when none is configured", () => { const root = createSession(SessionManager.inMemory(tempDir)); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.PRIME_AGENT_CODING_AGENT_DIR).toBeUndefined(); });
it("exports agentDir but skips key injection when no websearch skill is loaded", () => { const agentDir = join(tempDir, "custom-agent-dir"); const root = createSession(SessionManager.inMemory(tempDir), agentDir, "stored-key", false); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.PRIME_AGENT_CODING_AGENT_DIR).toBe(agentDir); expect(env.SERPER_API_KEY).toBeUndefined(); });
it("injects the key for a custom websearch skill even when bundled is off", () => { const previous = process.env.SERPER_API_KEY; delete process.env.SERPER_API_KEY; try { // loadWebsearchSkill=true models a --skill/project websearch; the bundled // setting is irrelevant because the gate checks the loaded skill, not settings. const root = createSession(SessionManager.inMemory(tempDir), undefined, "custom-key", true); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.SERPER_API_KEY).toBe("custom-key"); } finally { if (previous === undefined) delete process.env.SERPER_API_KEY; else process.env.SERPER_API_KEY = previous; } });
it("injects a literal stored Serper key into the kernel", () => { const previous = process.env.SERPER_API_KEY; delete process.env.SERPER_API_KEY; try { const root = createSession(SessionManager.inMemory(tempDir), undefined, "literal-serper-key", true); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.SERPER_API_KEY).toBe("literal-serper-key"); } finally { if (previous === undefined) delete process.env.SERPER_API_KEY; else process.env.SERPER_API_KEY = previous; } });
it("resolves an env-var-reference Serper key before injecting it", () => { const previousKey = process.env.SERPER_API_KEY; const previousRef = process.env.MY_SERPER_REF; delete process.env.SERPER_API_KEY; process.env.MY_SERPER_REF = "resolved-secret"; try { const root = createSession(SessionManager.inMemory(tempDir), undefined, "MY_SERPER_REF", true); const env = (root as unknown as InspectableRlmDirSession)._rlmKernelEnv(); expect(env.SERPER_API_KEY).toBe("resolved-secret"); } finally { if (previousKey === undefined) delete process.env.SERPER_API_KEY; else process.env.SERPER_API_KEY = previousKey; if (previousRef === undefined) delete process.env.MY_SERPER_REF; else process.env.MY_SERPER_REF = previousRef; } });});