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.
187 kB · 4998 lines
TypeScript
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036403740384039404040414042404340444045404640474048404940504051405240534054405540564057405840594060406140624063406440654066406740684069407040714072407340744075407640774078407940804081408240834084408540864087408840894090409140924093409440954096409740984099410041014102410341044105410641074108410941104111411241134114411541164117411841194120412141224123412441254126412741284129413041314132413341344135413641374138413941404141414241434144414541464147414841494150415141524153415441554156415741584159416041614162416341644165416641674168416941704171417241734174417541764177417841794180418141824183418441854186418741884189419041914192419341944195419641974198419942004201420242034204420542064207420842094210421142124213421442154216421742184219422042214222422342244225422642274228422942304231423242334234423542364237423842394240424142424243424442454246424742484249425042514252425342544255425642574258425942604261426242634264426542664267426842694270427142724273427442754276427742784279428042814282428342844285428642874288428942904291429242934294429542964297429842994300430143024303430443054306430743084309431043114312431343144315431643174318431943204321432243234324432543264327432843294330433143324333433443354336433743384339434043414342434343444345434643474348434943504351435243534354435543564357435843594360436143624363436443654366436743684369437043714372437343744375437643774378437943804381438243834384438543864387438843894390439143924393439443954396439743984399440044014402440344044405440644074408440944104411441244134414441544164417441844194420442144224423442444254426442744284429443044314432443344344435443644374438443944404441444244434444444544464447444844494450445144524453445444554456445744584459446044614462446344644465446644674468446944704471447244734474447544764477447844794480448144824483448444854486448744884489449044914492449344944495449644974498449945004501450245034504450545064507450845094510451145124513451445154516451745184519452045214522452345244525452645274528452945304531453245334534453545364537453845394540454145424543454445454546454745484549455045514552455345544555455645574558455945604561456245634564456545664567456845694570457145724573457445754576457745784579458045814582458345844585458645874588458945904591459245934594459545964597459845994600460146024603460446054606460746084609461046114612461346144615461646174618461946204621462246234624462546264627462846294630463146324633463446354636463746384639464046414642464346444645464646474648464946504651465246534654465546564657465846594660466146624663466446654666466746684669467046714672467346744675467646774678467946804681468246834684468546864687468846894690469146924693469446954696469746984699470047014702470347044705470647074708470947104711471247134714471547164717471847194720472147224723472447254726472747284729473047314732473347344735473647374738473947404741474247434744474547464747474847494750475147524753475447554756475747584759476047614762476347644765476647674768476947704771477247734774477547764777477847794780478147824783478447854786478747884789479047914792479347944795479647974798479948004801480248034804480548064807480848094810481148124813481448154816481748184819482048214822482348244825482648274828482948304831483248334834483548364837483848394840484148424843484448454846484748484849485048514852485348544855485648574858485948604861486248634864486548664867486848694870487148724873487448754876487748784879488048814882488348844885488648874888488948904891489248934894489548964897489848994900490149024903490449054906490749084909491049114912491349144915491649174918491949204921492249234924492549264927492849294930493149324933493449354936493749384939494049414942494349444945494649474948494949504951495249534954495549564957495849594960496149624963496449654966496749684969497049714972497349744975497649774978497949804981498249834984498549864987498849894990499149924993499449954996499749984999import { 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 { 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 { createTestExtensionsResult, createTestResourceLoader } from "./utilities.js";
const model = getModel("anthropic", "claude-sonnet-4-5")!;
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 }; 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;
beforeEach(() => { tempDir = join(tmpdir(), `pi-rlm-recursion-${Date.now()}-${Math.random().toString(36).slice(2)}`); mkdirSync(tempDir, { recursive: true }); });
afterEach(() => { session?.dispose(); session = undefined; rmSync(tempDir, { recursive: true, force: true }); });
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")), 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 forked.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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(root.runRlmChild("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(root.runRlmChild("invalid name", { name: " " })).rejects.toThrow("rlm.run name must not be empty"); await expect(root.runRlmChild("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(root.runRlmChild("duplicate passive", { name: "passive-worker" })).rejects.toThrow( 'Agent name "passive-worker" is unavailable', ); await expect(root.runRlmChild("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 = root.runRlmChild("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(root.runRlmChild("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(root.runRlmChild("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, });
if (hiding === "detached") { rootInternals._activeRlmChildRuns.set(id, { id, prompt: "hidden parent", sessionName: id, sessionDir: join(tempDir, id), model, abort: () => {}, status: "cancelled", settled: false, 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 root.runRlmChild("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 root.runRlmChild("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(child?.messages[0]).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([ expect.stringContaining("Inspect the saved BashHandle with .poll(), .output(), or .tail()"), ]); 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: [] }), roster: () => ({ current: { name: "child", id: "child-session", depth: 1 }, entries: [{ relationship: "parent", name: "parent", id: "parent-session", 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 roster = vi.fn(() => ({ current: { name: "root", id: root.sessionId, depth: 0 }, entries: publishedChild ? [ { relationship: "child" as const, name: publishedChild.sessionName ?? publishedChild.sessionId, id: publishedChild.sessionId, depth: 1, status: "running" as const, }, ] : [], })); const root = createSession({ agentMessageController: { listAgents: () => ({ agents: [] }), roster, 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 root.runRlmChild("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(roster).not.toHaveBeenCalled(); expect(sendAgentMessage).not.toHaveBeenCalled(); publishChild?.();
await expect(pendingSend).resolves.toMatchObject({ message: "hello" }); expect(roster).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: [] }), roster: () => ({ current: { name: "root", id: root.sessionId, depth: 0 }, entries: [ { relationship: "child" as const, name: child.sessionName ?? child.sessionId, id: child.sessionId, depth: 1, status: "idle" as const, }, ], }), sendAgentMessage, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => ({ session: child }), deleteRlmSubagentRuntime: async (_id, session) => session?.disposeAsync(), }, }); const spawned = await root.runRlmChild("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: [] }), roster: () => ({ current: { name: "root", id: root.sessionId, depth: 0 }, entries: [], }), sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: () => startupGate, deleteRlmSubagentRuntime: async () => {}, }, }); const spawned = await root.runRlmChild("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: [] }), roster: () => ({ current: { name: "root", id: root.sessionId, depth: 0 }, entries: [ { relationship: "child" as const, name: "shared-child", id: "healthy-child-session", 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 root.runRlmChild("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: [] }), roster: () => ({ current: { name: "root", id: root.sessionId, depth: 0 }, entries: [], }), sendAgentMessage: async () => { throw new Error("unexpected send"); }, }, subagentRuntimeHost: { createRlmSubagentRuntime: async () => { runtimeCreationStarted = true; await runtimeCreationGate; return { session: hostedChild }; }, deleteRlmSubagentRuntime: async (_id, child) => child?.disposeAsync(), }, }); await root.runRlmChild("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 roster = vi.fn(() => ({ current: { name: "child", id: "child-session", depth: 1 }, entries: [ { relationship: "parent" as const, name: "parent", id: "parent-session", 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: [] }), roster, 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(roster).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("[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("[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 root.runRlmChild("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 root.runRlmChild("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: expect.stringContaining( `RLM child cancel-worker (${spawned.rlm_child_id}) was cancelled: Cancelled by user`, ), details: { kind: "cancelled", reason: "Cancelled by user" }, }); }); });
it("injects exactly one notice with a preview when a child completes without replying", async () => { const root = createSession();
const spawned = await root.runRlmChild("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); expect(notices[0]).toMatchObject({ content: expect.stringContaining( `RLM child silent-worker (${spawned.rlm_child_id}) completed without sending a reply`, ), details: { kind: "completed_without_reply", lastAssistantTextPreview: "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 root.runRlmChild("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: [] }), roster: () => ({ current: { name: "reply-worker", id: child.sessionId, depth: 1 }, entries: [{ relationship: "parent", name: "parent", id: "parent-session", 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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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(root.runRlmChild("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 root.runRlmChild("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: [] }), roster: () => ({ current: { name: "child", id: "child", depth: 1 }, entries: [] }), 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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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(root.runRlmChild("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( root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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", }, ], }; 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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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", }, ], });
// 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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild(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 root.runRlmChild("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 root.runRlmChild("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("A callable `rlm`");
await root.setRlmMaxDepth(0); await root.steer("continue after max-depth update"); releaseFirstTurn(); await promptPromise;
expect(seenSystemPrompts).toHaveLength(2); expect(seenSystemPrompts[1]!).not.toContain("A callable `rlm`"); });
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("A callable `rlm`"); await root.navigateTree(baselineLeafId, { summarize: false }); expect(root.getRlmMaxDepthStatus()).toEqual({ maxDepth: 0, source: "env" }); expect(root.systemPrompt).not.toContain("A callable `rlm`"); } 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("A callable `rlm`"); } 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 root.runRlmChild("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 root.runRlmChild("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 child.runRlmChild("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("A callable `rlm`"); await expect(root.runRlmChild("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(root.runRlmChild("nested")).rejects.toThrow("RLM recursion depth limit reached"); });
it("rejects unsupported rlm.run kwargs loudly", async () => { const root = createSession();
await expect(root.runRlmChild("nested", { temperature: 0 })).rejects.toThrow( "Unsupported rlm.run kwargs: temperature", ); });
it("rejects a non-string rlm.run thinking kwarg", async () => { const root = createSession();
await expect(root.runRlmChild("nested", { thinking: 3 })).rejects.toThrow("rlm.run thinking must be a string"); });
it("rejects an unknown rlm.run thinking level", async () => { const root = createSession(); await expect(root.runRlmChild("nested", { thinking: "ultra" })).rejects.toThrow("must be one of"); });
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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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, }); 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, 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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 = root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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("was cancelled: 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 root.runRlmChild("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(root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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", ), ).toEqual([expect.objectContaining({ details: expect.objectContaining({ kind: "completed_without_reply" }) })]); 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 root.runRlmChild("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(root.runRlmChild("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 root.runRlmChild("blocked before runtime creation", { name: "reserved-worker" }); await waitFor(() => runtimeCreationStarted);
await root.deleteRlmSubagent("reserved-worker"); await expect(root.runRlmChild("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(root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 root.runRlmChild("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 = root.runRlmChild("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 parentSession.runRlmChild("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 root.runRlmChild("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 childSession.runRlmChild("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.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) return await rlm.run("detached child after idle")
_task = asyncio.create_task(_delayed_rlm())print("scheduled")`);
expect(scheduled.status).toBe("ok"); expect(scheduled.stdout.trim()).toBe("scheduled"); await waitFor(() => prompts.includes("detached 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;
beforeEach(() => { tempDir = join(tmpdir(), `pi-rlm-dir-${Date.now()}-${Math.random().toString(36).slice(2)}`); mkdirSync(tempDir, { recursive: true }); });
afterEach(() => { session?.dispose(); session = undefined; rmSync(tempDir, { recursive: true, force: true }); });
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 host system prompt", () => { 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: "2026-01-01T00:00:00.000Z", updated_at: "2026-01-01T00:00:00.000Z", version: 1, }, }, skill: {}, subagent: {}, }, refinements: [], }), "utf8", ); const root = createSession(SessionManager.inMemory(tempDir), undefined, undefined, false, ephemeralDir);
const prompt = root.systemPrompt;
expect(prompt).toContain("Ephemeral note"); expect(prompt).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; } });});