Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631import assert from 'node:assert/strict'import { createHash } from 'node:crypto'import { mkdtemp, mkdir, rm } from 'node:fs/promises'import net from 'node:net'import { tmpdir } from 'node:os'import { join } from 'node:path'import { it } from 'node:test'import { createSession, CredentialClient } from '../../atproto/dist/index.js'import { COLLECTIONS, MemoryRecordStore, buildTurnBundle, materialize } from '../../core/dist/index.js'import { LocalPds } from '../../atproto/test/local-pds.mjs'import { actorTypesFor, selectHarness, FakeContainerRunner, ForgeRegistry, GitHubForge, TurnDispatcher, TurnLedger, implBranchName, planArtifactRkey, reviewRkey, runTurn, selectDispatchable, turnContainerLabel,} from '../dist/index.js'import { reconcileOrphan, reconcileOrphans } from '../dist/cli.js'
const ROOT = 'did:plc:root'const HUMAN = 'did:plc:human'const AGENT = 'did:plc:agent'const SPACE_URI = `at://${ROOT}/${COLLECTIONS.space}/space`
function mk(did, collection, rkey, cid, value) { return { did, collection, rkey, uri: `at://${did}/${collection}/${rkey}`, cid, rev: '0000000000001', value }}const ref = (r) => ({ uri: r.uri, cid: r.cid })
/** * A minimal space with one human admin, one active agent member, a `plan` (goal-scoped) and an * `adr` (project-scoped) artifact type, one project, and one goal. `requestOverrides` customizes * the single plan request under test; extra records (messages, other requests) can be appended. */function buildScenario({ requestOverrides = {}, agentActive = true, agentKind = 'agent', extra = [] } = {}) { const records = [] const add = (r) => { records.push(r) return r } const space = add( mk(ROOT, COLLECTIONS.space, 'space', 'cid-space', { $type: COLLECTIONS.space, name: 'Radial', description: 'dispatch test', createdAt: '2026-01-01T00:00:00Z', }), ) add( mk(ROOT, COLLECTIONS.addMember, 'member-human', 'cid-mh', { $type: COLLECTIONS.addMember, space: ref(space), did: HUMAN, kind: 'human', role: 'member', createdAt: '2026-01-01T00:00:01Z', }), ) if (agentActive) { add( mk(ROOT, COLLECTIONS.addMember, 'member-agent', 'cid-ma', { $type: COLLECTIONS.addMember, space: ref(space), did: AGENT, kind: agentKind, role: agentKind === 'agent' ? 'agent' : 'member', createdAt: '2026-01-01T00:00:02Z', }), ) } add( mk(ROOT, COLLECTIONS.artifactType, 'type-plan', 'cid-type-plan', { $type: COLLECTIONS.artifactType, space: ref(space), name: 'plan', brief: 'Write a concrete plan.', outputSpec: { format: 'markdown', description: 'A plan' }, scope: 'goal', createdAt: '2026-01-01T00:00:10Z', }), ) add( mk(ROOT, COLLECTIONS.artifactType, 'type-adr', 'cid-type-adr', { $type: COLLECTIONS.artifactType, space: ref(space), name: 'adr', brief: 'Distill an ADR.', outputSpec: { format: 'markdown', description: 'An ADR' }, scope: 'project', createdAt: '2026-01-01T00:00:11Z', }), ) const project = add( mk(HUMAN, COLLECTIONS.project, 'project', 'cid-project', { $type: COLLECTIONS.project, space: ref(space), name: 'radial-ng', gitUrl: 'https://example.test/radial-ng.git', defaultBranch: 'main', checks: [], autoReview: {}, createdAt: '2026-01-01T00:01:00Z', }), ) const goal = add( mk(HUMAN, COLLECTIONS.goal, 'goal', 'cid-goal', { $type: COLLECTIONS.goal, space: ref(space), project: ref(project), title: 'Ship the thing', body: 'Get it out the door.', createdAt: '2026-01-01T00:02:00Z', }), ) const request = add( mk(HUMAN, COLLECTIONS.artifactRequest, 'request-1', 'cid-request-1', { $type: COLLECTIONS.artifactRequest, goal: ref(goal), type: 'plan', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:03:00Z', ...requestOverrides, }), ) for (const r of extra) add(r) return { records, space, project, goal, request }}
function store(records) { const s = new MemoryRecordStore() for (const r of records) s.put(r) return s}
function actorFor(did, artifactTypes = ['plan'], models = [], harness = 'claude', spaces = undefined) { return { did, profile: 'planner', handleName: 'planner', artifactTypes, ...(spaces ? { spaces } : {}), harness, models, session: { did, handle: `${did.split(':').at(-1)}.test`, service: 'https://pds.test', accessJwt: 'a', refreshJwt: 'r' }, client: {}, }}
function registryFor(actors) { const byDid = new Map() for (const actor of actors) { const list = byDid.get(actor.did) ?? [] list.push(actor) byDid.set(actor.did, list) } return { all: actors, byDid, // The real registry's rule: what a profile accepts is a per-space question (design §11). select(did, type, spaceUri) { return byDid.get(did)?.find((actor) => actorTypesFor(actor, spaceUri).includes(type)) }, }}
function uris(dispatchables) { return dispatchables.map((d) => d.request.uri)}
// --- Dispatch gate matrix -------------------------------------------------
it('selects a clean goal-scoped plan request assigned to a registered active agent', () => { const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].artifactType.name, 'plan') assert.ok(result[0].bundle) ledger.close()})
const archiveGoal = (goal, rkey, archived, createdAt) => mk(HUMAN, COLLECTIONS.archiveGoal, rkey, `cid-${rkey}`, { $type: COLLECTIONS.archiveGoal, goal: ref(goal), archived, createdAt, })
it('skips every request inside an archived goal', () => { const base = buildScenario() const records = [...base.records, archiveGoal(base.goal, 'arch', true, '2026-01-01T00:04:00Z')] const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('resumes dispatch once the goal is un-archived (latest archiveGoal wins)', () => { const base = buildScenario() const records = [ ...base.records, archiveGoal(base.goal, 'arch-on', true, '2026-01-01T00:04:00Z'), archiveGoal(base.goal, 'arch-off', false, '2026-01-01T00:05:00Z'), ] const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(uris(selectDispatchable(index, actors, ledger)), [base.request.uri]) ledger.close()})
it('skips every request inside a closed goal', () => { const base = buildScenario() // A second version of the goal record setting closed:true — the store honors this owner edit. const closed = mk(HUMAN, COLLECTIONS.goal, 'goal', 'cid-goal-closed', { ...base.goal.value, closed: true }) const index = materialize(store([...base.records, closed]), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
// A real unassigned request OMITS the assignee key entirely (the sidecar writes// `...(assignee ? { assignee } : {})`); an explicit `assignee: undefined` is a different record the// lexicon rejects, so strip the key rather than setting it undefined.function unassign(records) { return records.map((r) => { if (r.collection !== COLLECTIONS.artifactRequest) return r const { assignee: _drop, ...value } = r.value return { ...r, value } })}
/** * A live claim on `request` by `did`. Since Phase 7 an unassigned request is dispatched only for * the claim that WON the fold's tie-break, and only once this daemon has confirmed the win through * ingestion — so a test about unassigned dispatch needs both halves: the record (which decides who * won) and `heldClaims` (which is the claim manager saying "that was us, and we have seen it"). */function claimFor(request, did, { rkey = `claim-${did.split(':').at(-1)}`, createdAt = claimAge(5) } = {}) { const issued = Date.now() return mk(did, COLLECTIONS.claim, rkey, `cid-${rkey}`, { $type: COLLECTIONS.claim, request: ref(request), // A live lease declared the way a conforming writer declares one: half an hour from the moment // this version was issued. It cannot simply be dated in the far future — the protocol horizon // bounds `expiresAt − createdAt` for every version — so it is dated from the clock these tests // actually fold against, and `createdAt` sits a few minutes back where the tie-break reads it. expiresAt: new Date(issued + 30 * 60_000).toISOString(), renewedAt: new Date(issued).toISOString(), createdAt, })}/** A claim `seconds` into a claim race that started five minutes ago — a tie-break position. */const claimAge = (seconds) => new Date(Date.now() - 5 * 60_000 + seconds * 1000).toISOString()const confirmed = (...requests) => new Set(requests.map((request) => request.uri))
it('dispatches an unassigned plan request for a claim this daemon wrote and confirmed', () => { const { records, request } = buildScenario() const index = materialize(store(unassign([...records, claimFor(request, AGENT)])), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) const result = selectDispatchable(index, actors, ledger, undefined, undefined, confirmed(request)) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].actor.did, AGENT) ledger.close()})
it('holds an unassigned request whose claim is written but not yet confirmed through ingestion', () => { const { records, request } = buildScenario() const index = materialize(store(unassign([...records, claimFor(request, AGENT)])), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) const reasons = [] // The fold says the claim is ours, but the claim manager has not confirmed it. One ingestion // cycle of latency is the price of never having two daemons dispatch the same open request. assert.deepEqual( selectDispatchable(index, actors, ledger, undefined, (_uri, reason) => reasons.push(reason)), [], ) assert.match(reasons[0], /not yet confirmed through ingestion/) ledger.close()})
it('never dispatches an unassigned request whose claim another operator won', () => { const { records, request } = buildScenario() const OTHER = 'did:plc:otheroperator' // The other operator claimed FIRST, so it wins the tie-break on both operators' indexes — even // though this daemon holds a (later) claim of its own and believes it confirmed it. const theirs = claimFor(request, OTHER, { createdAt: claimAge(0) }) const ours = claimFor(request, AGENT, { createdAt: claimAge(5) }) const withMember = [ ...records, mk(ROOT, COLLECTIONS.addMember, 'member-other', 'cid-mo', { $type: COLLECTIONS.addMember, space: { uri: SPACE_URI, cid: 'cid-space' }, did: OTHER, kind: 'agent', role: 'agent', createdAt: '2026-01-01T00:00:03Z', }), ] const index = materialize(store(unassign([...withMember, theirs, ours])), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) const reasons = [] assert.deepEqual( selectDispatchable(index, actors, ledger, undefined, (_uri, reason) => reasons.push(reason), confirmed(request)), [], ) assert.match(reasons[0], new RegExp(`claim held by ${OTHER}`)) ledger.close()})
it('an assigned request dispatches on the first tick, with no claim record anywhere', () => { // C3, the single-operator regression check: claims exist for open requests only, and the assigned // path — everything before Phase 7 — must not gain a single tick of latency from them. const { records, request } = buildScenario() assert.equal(request.value.assignee, AGENT) const index = materialize(store(records), { spaceUri: SPACE_URI }) assert.equal( records.some((record) => record.collection === COLLECTIONS.claim), false, ) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) // No claim view supplied at all: an assigned request never consults one. assert.deepEqual(uris(selectDispatchable(index, actors, ledger)), [request.uri]) ledger.close()})
it('rejects an unassigned plan request when no active agent member can produce a plan', () => { const { records } = buildScenario() const index = materialize(store(unassign(records)), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() // A loaded actor exists for the agent, but it does not produce 'plan' artifacts. const actors = registryFor([actorFor(AGENT, ['adr'])]) const reasons = [] assert.deepEqual( selectDispatchable(index, actors, ledger, undefined, (_uri, reason) => reasons.push(reason)), [], ) assert.match(reasons[0], /unassigned/) ledger.close()})
it('leaves an assigned request alone when the assignee is scoped out of THIS space', () => { const { records } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() // Globally a planner; here, scoped to reviews only. The request is assigned to it and stays put — // the same "sits open" state a request whose type nobody accepts lands in, and the rejection says // which question was asked. const scoped = actorFor(AGENT, ['plan'], [], 'claude', { [SPACE_URI]: { artifactTypes: ['review'] }, }) const reasons = [] assert.deepEqual( selectDispatchable(index, registryFor([scoped]), ledger, undefined, (_uri, reason) => reasons.push(reason), ), [], ) assert.match(reasons[0], /does not produce 'plan' artifacts in this space/) // The very same profile, in a space it does not scope: dispatched. const elsewhere = materialize(store(records), { spaceUri: SPACE_URI }) elsewhere.space = { ...elsewhere.space, uri: `at://${ROOT}/${COLLECTIONS.space}/other` } assert.equal(selectDispatchable(elsewhere, registryFor([scoped]), ledger).length, 1) ledger.close()})
it('dispatches an agent-authored request (materializers do not police request authorship, design §3)', () => { const { records, request } = buildScenario() // Re-author the request as the agent itself. The daemon writes auto-review requests under one of // its agent identities (design §3/§7), so any active member — agent or human — may author a // dispatchable request; the human-initiation norm is a UI convention, not a dispatch gate. const mutated = records.map((r) => (r.collection === COLLECTIONS.artifactRequest ? { ...r, did: AGENT } : r)) const index = materialize(store(mutated), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(uris(selectDispatchable(index, actors, ledger)), [request.uri]) ledger.close()})
it('generalization: dispatches a goal-scoped NON-plan registered type when an actor produces it', () => { const { records, request } = buildScenario({ requestOverrides: { type: 'adr' } }) // Make the 'adr' registry type goal-scoped so the request lands in a goalView. Dispatch is no // longer plan-only: any registered goal-scoped (non-review) type whose actor produces it is // dispatchable. const goalScopedOther = records.map((r) => { if (r.collection === COLLECTIONS.artifactType && r.value.name === 'adr') { return { ...r, value: { ...r.value, scope: 'goal' } } } return r }) const index = materialize(store(goalScopedOther), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'adr'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].artifactType.name, 'adr') // Predecessor resolution is type-agnostic: an adr with no basedOn is simply a fresh v1. assert.deepEqual(result[0].predecessor, { candidates: [], ambiguous: false }) ledger.close()})
// A valid review scenario: the base plan request is fulfilled by an artifact (so it drops out of the// work queue), and a review request pins that artifact as its subject. `review` is never registry// data — a review request materializes purely from its resolving same-target subject.function withReviewRequest({ assignee = AGENT, subjectLinks = { commit: 'a'.repeat(40) }, author = HUMAN } = {}) { const base = buildScenario() const artifact = mk(AGENT, COLLECTIONS.artifact, 'plan-art', 'cid-plan-art', { $type: COLLECTIONS.artifact, request: ref(base.request), goal: ref(base.goal), type: 'plan', body: 'the plan', links: subjectLinks, createdAt: '2026-01-01T00:04:00Z', }) const reviewReq = mk(author, COLLECTIONS.artifactRequest, 'review-req', 'cid-review-req', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'review', subject: ref(artifact), basedOn: [], ...(assignee ? { assignee } : {}), createdAt: '2026-01-01T00:05:00Z', }) return { ...base, artifact, reviewReq, records: [...base.records, artifact, reviewReq] }}
// A project-scoped CAPTURE request: the base plan request is fulfilled by a goal plan artifact, and a// project-scoped `adr` request lists that artifact in basedOn (distil-from-goal-context, design §8).function withCaptureRequest() { const base = buildScenario() const planArt = mk(AGENT, COLLECTIONS.artifact, 'plan-art', 'cid-plan-art', { $type: COLLECTIONS.artifact, request: ref(base.request), goal: ref(base.goal), type: 'plan', body: 'the plan', links: {}, createdAt: '2026-01-01T00:04:00Z', }) const captureReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'capture-req', 'cid-capture-req', { $type: COLLECTIONS.artifactRequest, project: ref(base.project), type: 'adr', basedOn: [ref(planArt)], assignee: AGENT, createdAt: '2026-01-01T00:05:00Z', }) return { ...base, planArt, captureReq, records: [...base.records, planArt, captureReq] }}
// A project-scoped REVIEW request whose subject is a project-scoped `adr` artifact.function withProjectReviewRequest({ subjectLinks = {} } = {}) { const base = buildScenario() const adrReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'adr-req', 'cid-adr-req', { $type: COLLECTIONS.artifactRequest, project: ref(base.project), type: 'adr', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:03:30Z', }) const adrArt = mk(AGENT, COLLECTIONS.artifact, 'adr-art', 'cid-adr-art', { $type: COLLECTIONS.artifact, request: ref(adrReq), project: ref(base.project), type: 'adr', body: 'the adr', links: subjectLinks, createdAt: '2026-01-01T00:04:00Z', }) const reviewReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'preview-req', 'cid-preview-req', { $type: COLLECTIONS.artifactRequest, project: ref(base.project), type: 'review', subject: ref(adrArt), basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:05:00Z', }) return { ...base, adrReq, adrArt, reviewReq, records: [...base.records, adrReq, adrArt, reviewReq] }}
it('dispatches a review-type request to a review-producing actor, pinning the checkout to the subject commit', () => { const { records, reviewReq } = withReviewRequest() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'review'])]) const result = selectDispatchable(index, actors, ledger) // The base plan request is fulfilled by the artifact; only the review request is dispatchable. assert.deepEqual(uris(result), [reviewReq.uri]) assert.equal(result[0].artifactType.name, 'review') assert.equal(result[0].actor.did, AGENT) assert.deepEqual(result[0].review.subject, reviewReq.value.subject) assert.equal(result[0].review.commit, 'a'.repeat(40)) assert.ok(result[0].bundle.subject) assert.equal(result[0].bundle.subject.uri, reviewReq.value.subject.uri) ledger.close()})
it('a review request whose subject links no commit leaves review.commit unset (default-branch checkout)', () => { const { records, reviewReq } = withReviewRequest({ subjectLinks: {} }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'review'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [reviewReq.uri]) assert.equal(result[0].review.commit, undefined) ledger.close()})
it('an unassigned review request is picked up by the deterministic review-capable agent', () => { const { records, reviewReq } = withReviewRequest({ assignee: undefined }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'review'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [reviewReq.uri]) assert.equal(result[0].actor.did, AGENT) assert.equal(result[0].artifactType.name, 'review') ledger.close()})
it('a review request with an unanswered question from the assignee is gated (awaiting input)', () => { const { records, reviewReq, goal } = withReviewRequest() const question = mk(AGENT, COLLECTIONS.message, 'q-review', 'cid-q-review', { $type: COLLECTIONS.message, goal: ref(goal), body: 'which standard applies?', mentions: [], re: ref(reviewReq), createdAt: '2026-01-01T00:06:00Z', }) const index = materialize(store([...records, question]), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'review'])]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('does not dispatch a review request to an actor that does not produce reviews', () => { const { records } = withReviewRequest() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan'])]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
// An `answer` scenario: the base plan request lands its artifact, a human asks something in the// goal's thread, and an answer request pins that exact message. `answer` is never registry data — it// materializes purely from a subject message resolving under the request's own goal.function withAnswerRequest({ assignee = AGENT, extra = [] } = {}) { const base = buildScenario() const artifact = mk(AGENT, COLLECTIONS.artifact, 'plan-art', 'cid-plan-art', { $type: COLLECTIONS.artifact, request: ref(base.request), goal: ref(base.goal), type: 'plan', body: 'the plan', links: {}, createdAt: '2026-01-01T00:04:00Z', }) const question = mk(HUMAN, COLLECTIONS.message, 'thread-q', 'cid-thread-q', { $type: COLLECTIONS.message, goal: ref(base.goal), body: 'Why did you drop the SQLite path?', mentions: [AGENT], createdAt: '2026-01-01T00:05:00Z', }) const answerReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'answer-req', 'cid-answer-req', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'answer', subject: ref(question), basedOn: [ref(artifact)], ...(assignee ? { assignee } : {}), createdAt: '2026-01-01T00:05:01Z', }) return { ...base, artifact, question, answerReq, records: [...base.records, artifact, question, answerReq, ...extra], }}
it('dispatches an answer request to an answer-producing actor, carrying the message being answered', () => { const { records, answerReq, question, goal } = withAnswerRequest() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'answer'])]) const result = selectDispatchable(index, actors, ledger) // The base plan request is fulfilled by its artifact; only the answer request dispatches. assert.deepEqual(uris(result), [answerReq.uri]) const item = result[0] assert.equal(item.artifactType.name, 'answer') assert.deepEqual(item.answer.subject, ref(question)) assert.equal(item.answer.subjectAuthor, HUMAN) assert.deepEqual(item.answer.anchor, { goal: ref(goal) }) // where the subject is, so where the reply goes assert.equal(item.answer.root, undefined) // the subject starts the thread assert.equal(item.target.target.uri, goal.uri) // No predecessor resolution: a reply is not a chained artifact. assert.equal(item.predecessor, undefined) assert.equal(item.bundle.answering.uri, question.uri) ledger.close()})
it('anchors the reply to an artifact-anchored subject, not to the request’s goal', () => { // A message anchored to an artifact under the goal is in that goal's thread (the fold files it // there), so it is a message somebody can be asked about — and the reply belongs where it is. // Anchoring it to the goal instead would write a record `message post --parent` refuses from a // human, which is the disagreement between the two write paths this pins shut. const base = buildScenario() const artifact = mk(AGENT, COLLECTIONS.artifact, 'plan-art', 'cid-plan-art', { $type: COLLECTIONS.artifact, request: ref(base.request), goal: ref(base.goal), type: 'plan', body: 'the plan', links: {}, createdAt: '2026-01-01T00:04:00Z', }) const onArtifact = mk(HUMAN, COLLECTIONS.message, 'art-q', 'cid-art-q', { $type: COLLECTIONS.message, artifact: ref(artifact), body: 'Why is the second read safe?', mentions: [AGENT], createdAt: '2026-01-01T00:05:00Z', }) const answerReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'answer-req', 'cid-answer-req', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'answer', subject: ref(onArtifact), basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:05:01Z', }) const index = materialize(store([...base.records, artifact, onArtifact, answerReq]), { spaceUri: SPACE_URI, }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'answer'])]) const result = selectDispatchable(index, actors, ledger) const answer = result.find((item) => item.request.uri === answerReq.uri) assert.ok(answer, 'an artifact-anchored subject under the goal still dispatches') assert.deepEqual(answer.answer.anchor, { artifact: ref(artifact) }) ledger.close()})
it('threads a reply under the root of an existing exchange', () => { const base = buildScenario() const rootMessage = mk(HUMAN, COLLECTIONS.message, 'root-q', 'cid-root-q', { $type: COLLECTIONS.message, goal: ref(base.goal), body: 'Opening question.', mentions: [], createdAt: '2026-01-01T00:04:00Z', }) const follow = mk(HUMAN, COLLECTIONS.message, 'follow-q', 'cid-follow-q', { $type: COLLECTIONS.message, goal: ref(base.goal), parent: ref(rootMessage), root: ref(rootMessage), body: 'And what about the second half?', mentions: [], createdAt: '2026-01-01T00:04:10Z', }) const answerReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'answer-req', 'cid-answer-req', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'answer', subject: ref(follow), basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:04:20Z', }) const index = materialize(store([...base.records, rootMessage, follow, answerReq]), { spaceUri: SPACE_URI, }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'answer'])]) const result = selectDispatchable(index, actors, ledger) const answer = result.find((item) => item.request.uri === answerReq.uri) assert.deepEqual(answer.answer.root, ref(rootMessage)) ledger.close()})
it('refuses an unassigned answer request in v1 — the fold closes it on the assignee reply', () => { // `null`, not `undefined`: the helper's default parameter would substitute AGENT back in. const { records, answerReq } = withAnswerRequest({ assignee: null }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'answer'])]) const rejects = [] const result = selectDispatchable(index, actors, ledger, undefined, (uri, reason) => rejects.push([uri, reason]), ) assert.deepEqual(uris(result), []) assert.deepEqual(rejects, [[answerReq.uri, 'answer requests must name an assignee in v1']]) ledger.close()})
it('does not dispatch an answer request to an actor that does not produce answers', () => { const { records } = withAnswerRequest() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'review'])]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('an answer request blocked on its own question is gated (awaiting input)', () => { const { records, answerReq, goal } = withAnswerRequest() const question = mk(AGENT, COLLECTIONS.message, 'q-answer', 'cid-q-answer', { $type: COLLECTIONS.message, goal: ref(goal), body: 'Which SQLite path do you mean?', mentions: [], re: ref(answerReq), createdAt: '2026-01-01T00:06:00Z', }) const index = materialize(store([...records, question]), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'answer'])]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('holds an answer request whose basedOn artifact the index has not ingested (stale index)', () => { const base = buildScenario() const question = mk(HUMAN, COLLECTIONS.message, 'thread-q', 'cid-thread-q', { $type: COLLECTIONS.message, goal: ref(base.goal), body: 'About the plan…', mentions: [], createdAt: '2026-01-01T00:05:00Z', }) const answerReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'answer-req', 'cid-answer-req', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'answer', subject: ref(question), basedOn: [{ uri: `at://${AGENT}/${COLLECTIONS.artifact}/ghost`, cid: 'cid-ghost' }], assignee: AGENT, createdAt: '2026-01-01T00:05:01Z', }) const index = materialize(store([...base.records, question, answerReq]), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'answer'])]) const rejects = [] const result = selectDispatchable(index, actors, ledger, undefined, (uri, reason) => rejects.push([uri, reason]), ) assert.equal(result.some((item) => item.request.uri === answerReq.uri), false) assert.ok(rejects.some(([uri, reason]) => uri === answerReq.uri && /stale index/.test(reason))) ledger.close()})
it('dispatches a project-scoped CAPTURE request; its bundle carries the source goal', () => { const { records, captureReq, goal, project } = withCaptureRequest() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'adr'])]) const result = selectDispatchable(index, actors, ledger) // The base plan request is fulfilled by its artifact; only the capture request dispatches. assert.deepEqual(uris(result), [captureReq.uri]) const cap = result[0] assert.equal(cap.artifactType.name, 'adr') assert.equal(cap.target.target.uri, project.uri) // the owning view is the ProjectView assert.ok(cap.bundle.goal) assert.equal(cap.bundle.goal.uri, goal.uri) // the capture source goal, reached through basedOn // Its basedOn resolves to a `plan`, not another `adr`, so it is a fresh v1 with no predecessor. assert.deepEqual(cap.predecessor, { candidates: [], ambiguous: false }) ledger.close()})
it('holds a project-scoped capture request until its basedOn source is ingested (stale index), then dispatches with the source goal', () => { const base = buildScenario() const planSrc = mk(AGENT, COLLECTIONS.artifact, 'plan-src', 'cid-plan-src', { $type: COLLECTIONS.artifact, request: ref(base.request), goal: ref(base.goal), type: 'plan', body: 'the plan', links: {}, createdAt: '2026-01-01T00:03:30Z', }) const captureReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'capture-req', 'cid-capture-req', { $type: COLLECTIONS.artifactRequest, project: ref(base.project), type: 'adr', basedOn: [ref(planSrc)], assignee: AGENT, createdAt: '2026-01-01T00:05:00Z', }) const actors = registryFor([actorFor(AGENT, ['plan', 'adr'])])
// Source artifact NOT yet ingested: the capture request is held (stale index), never dispatched // goal-less. (Its basedOn ref is unresolved, so the bundle records it in missingRefs.) const stale = materialize(store([...base.records, captureReq]), { spaceUri: SPACE_URI }) const ledgerA = new TurnLedger() const reasons = [] const held = selectDispatchable(stale, actors, ledgerA, undefined, (uri, reason) => { if (uri === captureReq.uri) reasons.push(reason) }) assert.equal(held.some((d) => d.request.uri === captureReq.uri), false) assert.match(reasons.join(' '), /unresolved basedOn/) ledgerA.close()
// Source now ingested: the capture request dispatches, and the bundle carries the source goal. const ready = materialize(store([...base.records, planSrc, captureReq]), { spaceUri: SPACE_URI }) const ledgerB = new TurnLedger() const cap = selectDispatchable(ready, actors, ledgerB).find((d) => d.request.uri === captureReq.uri) assert.ok(cap, 'the capture request should dispatch once its source resolves') assert.ok(cap.bundle.goal) assert.equal(cap.bundle.goal.uri, base.goal.uri) ledgerB.close()})
it('dispatches a project-scoped REVIEW request whose subject is a project-scoped artifact (integration item d)', () => { const { records, reviewReq, adrArt, project } = withProjectReviewRequest() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'adr', 'review'])]) const result = selectDispatchable(index, actors, ledger) const review = result.find((d) => d.request.uri === reviewReq.uri) assert.ok(review, 'the project-scoped review request should dispatch') assert.equal(review.artifactType.name, 'review') assert.equal(review.target.target.uri, project.uri) // dispatched from the project loop assert.deepEqual(review.review.subject, reviewReq.value.subject) assert.ok(review.bundle.subject) assert.equal(review.bundle.subject.uri, adrArt.uri) assert.equal(review.bundle.goal, undefined) // a project-scoped review has no goal ledger.close()})
it('dispatches a review request authored by an agent-kind member with a valid assignee (auto-review is agent-authored, §3/§7)', () => { const base = withReviewRequest({ author: AGENT }) const index = materialize(store(base.records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'review'])]) const result = selectDispatchable(index, actors, ledger) assert.ok( result.some((d) => d.request.uri === base.reviewReq.uri), 'an agent-authored review request must not sit open forever', ) ledger.close()})
it('registry is first-wins on a duplicate type name (agrees with the materializer earliest-wins)', () => { // A LATER duplicate `plan` type record. The materializer resolves types earliest-wins; the daemon // registry must too, or they disagree on what `plan` means. First-wins → the EARLIEST record wins. const laterDuplicate = mk(ROOT, COLLECTIONS.artifactType, 'type-plan-dup', 'cid-type-plan-dup', { $type: COLLECTIONS.artifactType, space: { uri: SPACE_URI, cid: 'cid-space' }, name: 'plan', brief: 'LATER DUPLICATE — must not win.', outputSpec: { format: 'markdown', description: 'dup' }, scope: 'goal', createdAt: '2026-06-01T00:00:00Z', // later than the base plan type (00:00:10 on 2026-01-01) }) const { records, request } = buildScenario({ extra: [laterDuplicate] }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].artifactType.brief, 'Write a concrete plan.') // the earliest record won ledger.close()})
it('skips a request whose type is not a registered artifact type', () => { const { records } = buildScenario({ requestOverrides: { type: 'nonesuch' } }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'nonesuch'])]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('dispatches a project-scoped request from the project loop (adr → project-view target, bundle built)', () => { const { records, project, request } = buildScenario() const projectRequest = mk(HUMAN, COLLECTIONS.artifactRequest, 'request-adr', 'cid-request-adr', { $type: COLLECTIONS.artifactRequest, project: ref(project), type: 'adr', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:03:30Z', }) const index = materialize(store([...records, projectRequest]), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'adr'])]) const result = selectDispatchable(index, actors, ledger) // Both the goal-scoped plan and the project-scoped adr now dispatch (sorted by request uri). assert.deepEqual(uris(result).sort(), [request.uri, projectRequest.uri].sort()) const adr = result.find((d) => d.request.uri === projectRequest.uri) assert.equal(adr.artifactType.name, 'adr') assert.equal(adr.target.target.uri, project.uri) // the owning view is the ProjectView assert.deepEqual(adr.predecessor, { candidates: [], ambiguous: false }) // no basedOn → fresh v1 assert.ok(adr.bundle) assert.equal(adr.bundle.goal, undefined) // a direct project request has no goal assert.equal(adr.bundle.project.uri, project.uri) ledger.close()})
it('rejects a request with an unknown/unregistered assignee', () => { const { records } = buildScenario({ requestOverrides: { assignee: 'did:plc:nobody' } }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('rejects when the assignee is not an active agent member', () => { const { records } = buildScenario({ agentKind: 'human' }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('rejects a request blocked on an unanswered question (goalView.awaitingInput)', () => { const { records, request } = buildScenario() const question = mk(AGENT, COLLECTIONS.message, 'question-1', 'cid-question-1', { $type: COLLECTIONS.message, goal: request.value.goal, body: 'which approach do you want?', mentions: [], re: ref(request), createdAt: '2026-01-01T00:03:10Z', }) const index = materialize(store([...records, question]), { spaceUri: SPACE_URI }) assert.ok(index.goals[0].awaitingInput.includes(request.uri)) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('rejects a request the ledger considers ineligible: running, cooling down, and gave_up', () => { const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT)])
const running = new TurnLedger() running.markRunning(request.uri, request.cid, { containerLabel: 'radial.turn.x', checkoutPath: '/tmp/x' }) assert.deepEqual(selectDispatchable(index, actors, running), []) running.close()
let now = '2026-01-01T00:00:00.000Z' const cooling = new TurnLedger(':memory:', { cooldownMs: 300_000, retryBound: 5, now: () => now }) cooling.markRunning(request.uri, request.cid, { containerLabel: 'radial.turn.x', checkoutPath: '/tmp/x' }) cooling.markCrashed(request.uri) assert.deepEqual(selectDispatchable(index, actors, cooling, now), []) cooling.close()
const gaveUp = new TurnLedger(':memory:', { cooldownMs: 1, retryBound: 1, now: () => now }) gaveUp.markRunning(request.uri, request.cid, { containerLabel: 'radial.turn.x', checkoutPath: '/tmp/x' }) const outcome = gaveUp.markCrashed(request.uri) assert.equal(outcome.state, 'gave_up') assert.deepEqual(selectDispatchable(index, actors, gaveUp), []) // turn reset re-enables it. gaveUp.reset(request.uri) assert.deepEqual(uris(selectDispatchable(index, actors, gaveUp)), [request.uri]) gaveUp.close()})
// --- Implementation dispatch (generalized type + v2 predecessor) ------------
const IMPL_TYPE = mk(ROOT, COLLECTIONS.artifactType, 'type-impl', 'cid-type-impl', { $type: COLLECTIONS.artifactType, space: { uri: SPACE_URI, cid: 'cid-space' }, name: 'implementation', brief: 'Implement it.', outputSpec: { format: 'markdown', description: 'An implementation' }, scope: 'goal', createdAt: '2026-01-01T00:00:13Z',})
// A prior implementation artifact under the goal (plus the request it fulfilled, which the// materializer requires to admit the artifact). `branch`/`pr` land in links for v2 reuse.function implPredecessor(goal, rkey, { branch, pr }) { const req = mk(HUMAN, COLLECTIONS.artifactRequest, `req-${rkey}`, `cid-req-${rkey}`, { $type: COLLECTIONS.artifactRequest, goal: ref(goal), type: 'implementation', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:02:30Z', }) const art = mk(AGENT, COLLECTIONS.artifact, rkey, `cid-${rkey}`, { $type: COLLECTIONS.artifact, request: ref(req), goal: ref(goal), type: 'implementation', body: 'prior impl', links: { branch, commit: 'abadcafe', pr }, createdAt: '2026-01-01T00:02:40Z', }) return { req, art }}
it('dispatches a fresh implementation request (no basedOn) as non-ambiguous with no predecessor', () => { const { records, request } = buildScenario({ requestOverrides: { type: 'implementation' }, extra: [IMPL_TYPE] }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].artifactType.name, 'implementation') assert.deepEqual(result[0].predecessor, { candidates: [], ambiguous: false }) ledger.close()})
it('resolves a single same-goal implementation predecessor (v2) with its branch + pr links', () => { const { goal, project } = buildScenario() const pred = implPredecessor(goal, 'impl-v1', { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(pred.art)] }, extra: [IMPL_TYPE, pred.req, pred.art], }) void project const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].predecessor.ambiguous, false) assert.deepEqual(result[0].predecessor.artifact, { ref: ref(pred.art), branch: 'radial/impl-v1', prUrl: 'https://github.com/o/r/pull/7', }) ledger.close()})
it('flags >1 same-goal implementation predecessors as ambiguous (still returned, for rejection)', () => { const { goal } = buildScenario() const p1 = implPredecessor(goal, 'impl-a', { branch: 'radial/impl-a', pr: 'https://github.com/o/r/pull/1' }) const p2 = implPredecessor(goal, 'impl-b', { branch: 'radial/impl-b', pr: 'https://github.com/o/r/pull/2' }) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(p1.art), ref(p2.art)] }, extra: [IMPL_TYPE, p1.req, p1.art, p2.req, p2.art], }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].predecessor.ambiguous, true) assert.equal(result[0].predecessor.artifact, undefined) // Both candidates are retained, ordered by uri, so the rejection message can anchor to one. assert.deepEqual( result[0].predecessor.candidates.map((candidate) => candidate.ref), [ref(p1.art), ref(p2.art)].sort((a, b) => a.uri.localeCompare(b.uri)), ) ledger.close()})
it('dispatcher reuses a predecessor branch only after observing a clean same-repository open PR', async () => { const { goal } = buildScenario() const predecessor = implPredecessor(goal, 'impl-v1', { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7', }) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(predecessor.art)] }, extra: [IMPL_TYPE, predecessor.req, predecessor.art], }) const githubRecords = records.map((record) => record.collection === COLLECTIONS.project ? { ...record, value: { ...record.value, gitUrl: 'https://github.com/o/r.git' } } : record, ) const index = materialize(store(githubRecords), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['implementation'])]) const ledger = new TurnLedger() let captured const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-v2-reuse', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, forge: { getPullRequestState: async () => ({ state: 'open', headRef: 'radial/impl-v1', headRepoFullName: 'o/r', baseRef: 'main', }), }, runTurn: async (input) => { captured = input return { outcome: 'fulfilled', acceptedRef: { uri: 'at://did:plc:agent/artifact/impl', cid: 'cid-impl' }, label: 'radial.turn.impl', } }, })
dispatcher.pump(index, actors) await dispatcher.drain()
assert.equal(captured.checkoutRef, 'radial/impl-v1') assert.equal(captured.branch, 'radial/impl-v1') assert.deepEqual(captured.prev, ref(predecessor.art)) assert.equal(ledger.get(request.uri).state, 'fulfilled') ledger.close()})
it('dispatcher opens a fresh reserved branch when predecessor PR observation is unsafe', async () => { const { goal } = buildScenario() const predecessor = implPredecessor(goal, 'impl-v1', { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7', }) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(predecessor.art)] }, extra: [IMPL_TYPE, predecessor.req, predecessor.art], }) const githubRecords = records.map((record) => record.collection === COLLECTIONS.project ? { ...record, value: { ...record.value, gitUrl: 'https://github.com/o/r.git' } } : record, ) const index = materialize(store(githubRecords), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['implementation'])]) const ledger = new TurnLedger() let captured const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-v2-fresh', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, forge: { getPullRequestState: async () => ({ state: 'open', headRef: 'radial/impl-v1', headRepoFullName: 'attacker/fork', baseRef: 'main', }), }, runTurn: async (input) => { captured = input return { outcome: 'fulfilled', acceptedRef: { uri: 'at://did:plc:agent/artifact/impl', cid: 'cid-impl' }, label: 'radial.turn.impl', } }, })
dispatcher.pump(index, actors) await dispatcher.drain()
assert.equal(captured.checkoutRef, undefined) assert.equal(captured.branch, implBranchName(request.uri, request.cid)) assert.deepEqual(captured.prev, ref(predecessor.art)) ledger.close()})
it('dispatcher skips implementation requests without GitHub possession', () => { const { records, request } = buildScenario({ requestOverrides: { type: 'implementation' }, extra: [IMPL_TYPE], }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['implementation'])]) const ledger = new TurnLedger() const logs = [] let launched = false const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-disabled', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: false, runTurn: async () => { launched = true return { outcome: 'crashed', label: 'radial.turn.impl' } }, log: (message) => logs.push(message), })
dispatcher.pump(index, actors)
assert.equal(launched, false) assert.equal(dispatcher.inFlight, 0) assert.equal(ledger.get(request.uri), undefined) assert.ok(logs.some((message) => message.includes('GitHub authentication is unavailable'))) ledger.close()})
// --- FIX: unassigned awaiting-input guard keys off the SELECTED actor DID ----// The stale-index guard previously matched question authors against `request.value.assignee`, which// is undefined for an unassigned request — so an unassigned request's ANSWERED question could never// re-enable dispatch (questionInIndex was always false). Keyed off the selected actor's DID, an// answered unassigned request becomes dispatchable again.
it('unassigned awaiting_input: an answered agent question re-enables dispatch (was stuck under the assignee-based guard)', () => { const base = buildScenario() const question = mk(AGENT, COLLECTIONS.message, 'q-unassigned', 'cid-q-unassigned', { $type: COLLECTIONS.message, goal: base.request.value.goal, body: 'which approach?', mentions: [], re: ref(base.request), createdAt: '2026-01-01T00:03:10Z', }) const answer = mk(HUMAN, COLLECTIONS.message, 'a-unassigned', 'cid-a-unassigned', { $type: COLLECTIONS.message, goal: base.request.value.goal, body: 'approach A', mentions: [], parent: ref(question), createdAt: '2026-01-01T00:03:20Z', }) const claim = claimFor(base.request, AGENT) const index = materialize(store(unassign([...base.records, question, answer, claim])), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() ledger.markAwaitingInput(base.request.uri) const actors = registryFor([actorFor(AGENT)]) // The selected actor is AGENT (the confirmed claim holder); its question is ingested AND answered // (materializer awaitingInput only tracks assigned requests, so it's empty here), so the guard // now lets it through. assert.deepEqual( uris(selectDispatchable(index, actors, ledger, undefined, undefined, confirmed(base.request))), [base.request.uri], ) ledger.close()})
it('unassigned awaiting_input: while the selected actor question is NOT yet ingested, it is not re-selected', () => { const base = buildScenario() const index = materialize(store(unassign(base.records)), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() ledger.markAwaitingInput(base.request.uri) const actors = registryFor([actorFor(AGENT)]) assert.equal(ledger.eligible(base.request.uri), true) // ledger alone: eligible again assert.deepEqual(selectDispatchable(index, actors, ledger), []) // stale-index guard blocks it ledger.close()})
// --- FIX #7: awaiting_input stale-index race --------------------------------// A ledger row in `awaiting_input` is immediately `eligible()`. If the next index snapshot has not// yet ingested the assignee's question, `goalView.awaitingInput` is still empty too, so without the// extra index-caught-up check the request would be re-selected and redispatched before any human// ever sees the question. These three cases pin the intended matrix.
it('awaiting_input + index has NOT ingested the assignee question yet: NOT selected (stale index)', () => { const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() ledger.markAwaitingInput(request.uri) const actors = registryFor([actorFor(AGENT)]) // Sanity: the ledger alone considers awaiting_input immediately eligible again. assert.equal(ledger.eligible(request.uri), true) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('awaiting_input + question ingested but unanswered (still in goalView.awaitingInput): NOT selected', () => { const { records, request } = buildScenario() const question = mk(AGENT, COLLECTIONS.message, 'question-1', 'cid-question-1', { $type: COLLECTIONS.message, goal: request.value.goal, body: 'which approach do you want?', mentions: [], re: ref(request), createdAt: '2026-01-01T00:03:10Z', }) const index = materialize(store([...records, question]), { spaceUri: SPACE_URI }) assert.ok(index.goals[0].awaitingInput.includes(request.uri)) const ledger = new TurnLedger() ledger.markAwaitingInput(request.uri) const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(selectDispatchable(index, actors, ledger), []) ledger.close()})
it('awaiting_input + question ingested AND answered (request absent from awaitingInput): selected', () => { const { records, request } = buildScenario() const question = mk(AGENT, COLLECTIONS.message, 'question-1', 'cid-question-1', { $type: COLLECTIONS.message, goal: request.value.goal, body: 'which approach do you want?', mentions: [], re: ref(request), createdAt: '2026-01-01T00:03:10Z', }) const answer = mk(HUMAN, COLLECTIONS.message, 'answer-1', 'cid-answer-1', { $type: COLLECTIONS.message, goal: request.value.goal, body: 'go with approach A', mentions: [], parent: ref(question), createdAt: '2026-01-01T00:03:20Z', }) const index = materialize(store([...records, question, answer]), { spaceUri: SPACE_URI }) assert.ok(!index.goals[0].awaitingInput.includes(request.uri)) const ledger = new TurnLedger() ledger.markAwaitingInput(request.uri) const actors = registryFor([actorFor(AGENT)]) assert.deepEqual(uris(selectDispatchable(index, actors, ledger)), [request.uri]) ledger.close()})
// --- Turn-loop integration via TurnDispatcher -----------------------------
function sendLine(socketPath, obj) { return new Promise((resolve, reject) => { const socket = net.createConnection(socketPath) let data = '' socket.on('connect', () => socket.write(`${JSON.stringify(obj)}\n`)) socket.on('data', (chunk) => { data += chunk.toString('utf8') }) socket.on('end', () => { try { resolve(JSON.parse(data)) } catch (error) { reject(new Error(`Non-JSON response: ${data} (${error.message})`)) } }) socket.on('error', reject) })}
function hostSocketPath(spec) { const mount = spec.mounts.find((m) => m.target === '/run/radial') return join(mount.source, 'turn.sock')}
// A short, deterministic per-request directory name (unix socket paths have a small max length,// so this cannot just be the request uri).function shortRunDir(runsDir, requestUri) { const hex = [...createHash('sha256').update(requestUri).digest()].map((b) => b.toString(16).padStart(2, '0')).join('') return join(runsDir, hex.slice(0, 12))}
const noopCheckout = async ({ dest }) => { await mkdir(dest, { recursive: true }) return { path: dest, commit: 'deadbeef' }}
async function withTempDir(run) { const directory = await mkdtemp(join(tmpdir(), 'radial-dispatch-')) try { return await run(directory) } finally { await rm(directory, { recursive: true, force: true }).catch(() => {}) }}
function makeDispatcher({ ledger, runsDir, runnerFn, now, log = () => {} }) { const runner = new FakeContainerRunner(runnerFn) const boundRunTurn = (input) => runTurn(input, { runner, harnesses: { select: selectHarness }, checkout: noopCheckout }) const dispatcher = new TurnDispatcher({ ledger, concurrency: 2, ...(now ? { now } : {}), runDirFor: (uri) => shortRunDir(runsDir, uri), image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], runTurn: boundRunTurn, log, }) return { dispatcher, runner }}
it('fulfilled turn, driven through TurnDispatcher.pump/drain, transitions the ledger to fulfilled', async () => { await withTempDir(async (runsDir) => { const pds = new LocalPds(AGENT) const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds)) const client = new CredentialClient(session, pds.fetch.bind(pds)) const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([{ ...actorFor(AGENT), client }]) const ledger = new TurnLedger()
const { dispatcher } = makeDispatcher({ ledger, runsDir, runnerFn: async (spec) => { const response = await sendLine(hostSocketPath(spec), { method: 'submitArtifact', token: spec.env.RADIAL_TURN_TOKEN, title: 'A short title', body: 'the plan', }) assert.equal(response.ok, true) return { exitCode: 0, timedOut: false } }, })
dispatcher.pump(index, actors) assert.equal(dispatcher.inFlight, 1) await dispatcher.drain()
const row = ledger.get(request.uri) assert.equal(row.state, 'fulfilled') const expectedRkey = planArtifactRkey(request.uri, request.cid) assert.equal(row.acceptedRef.uri, `at://${AGENT}/${COLLECTIONS.artifact}/${expectedRkey}`) const stored = [...pds.records.values()].filter((r) => r.value.$type === COLLECTIONS.artifact) assert.equal(stored.length, 1) ledger.close() })})
it('crash -> retry + cooldown -> gave_up across the retry bound, and a gave_up request is not re-selected', async () => { await withTempDir(async (runsDir) => { const pds = new LocalPds(AGENT) const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds)) const client = new CredentialClient(session, pds.fetch.bind(pds)) const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([{ ...actorFor(AGENT), client }])
let now = '2026-01-01T00:00:00.000Z' const cooldownMs = 1000 const retryBound = 2 const ledger = new TurnLedger(':memory:', { cooldownMs, retryBound, now: () => now }) const logs = [] const { dispatcher } = makeDispatcher({ ledger, runsDir, now: () => now, log: (m) => logs.push(m), runnerFn: async () => ({ exitCode: 1, timedOut: false }), // no socket activity: always a crash })
// Attempt 1: crash, then cooldown gates re-dispatch. dispatcher.pump(index, actors) await dispatcher.drain() assert.equal(ledger.get(request.uri).state, 'crashed') assert.equal(ledger.get(request.uri).attempts, 1) dispatcher.pump(index, actors) // still cooling down: nothing launched assert.equal(dispatcher.inFlight, 0)
// Advance past the cooldown and retry: this is the final attempt (retryBound=2) -> gave_up. now = new Date(Date.parse(now) + cooldownMs + 1).toISOString() dispatcher.pump(index, actors) assert.equal(dispatcher.inFlight, 1) await dispatcher.drain()
const row = ledger.get(request.uri) assert.equal(row.state, 'gave_up') assert.equal(row.attempts, retryBound) assert.ok(logs.some((m) => m.includes('GAVE UP')))
// A gave_up request is never re-selected, cooldown or not. dispatcher.pump(index, actors) assert.equal(dispatcher.inFlight, 0) assert.deepEqual(selectDispatchable(index, actors, ledger, now), [])
// turn reset re-enables it. ledger.reset(request.uri) assert.deepEqual(uris(selectDispatchable(index, actors, ledger, now)), [request.uri]) ledger.close() })})
it('awaiting_input outcome is not relaunched, and once the question lands in the index it is gated by awaitingInput too', async () => { await withTempDir(async (runsDir) => { const pds = new LocalPds(AGENT) const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds)) const client = new CredentialClient(session, pds.fetch.bind(pds)) const { records, request } = buildScenario() const recordStore = store(records) const index = materialize(recordStore, { spaceUri: SPACE_URI }) const actors = registryFor([{ ...actorFor(AGENT), client }]) const ledger = new TurnLedger()
const { dispatcher } = makeDispatcher({ ledger, runsDir, runnerFn: async (spec) => { const response = await sendLine(hostSocketPath(spec), { method: 'askQuestion', token: spec.env.RADIAL_TURN_TOKEN, body: 'which approach?', }) assert.equal(response.ok, true) return { exitCode: 0, timedOut: false } }, })
dispatcher.pump(index, actors) await dispatcher.drain() const row = ledger.get(request.uri) assert.equal(row.state, 'awaiting_input')
// The ledger alone considers awaiting_input eligible again immediately... assert.equal(ledger.eligible(request.uri), true) // ...but re-materializing with the question the turn actually wrote now blocks it via // goalView.awaitingInput, independent of the ledger. const written = [...pds.records.values()].find((r) => r.value.$type === COLLECTIONS.message) assert.ok(written) const rkey = written.uri.slice(written.uri.lastIndexOf('/') + 1) recordStore.put({ did: AGENT, collection: COLLECTIONS.message, rkey, uri: written.uri, cid: written.cid, rev: '0000000000002', value: written.value, }) const index2 = materialize(recordStore, { spaceUri: SPACE_URI }) assert.ok(index2.goals[0].awaitingInput.includes(request.uri)) assert.deepEqual(selectDispatchable(index2, actors, ledger), []) ledger.close() })})
// --- FIX #11: config-overridable memory limit -------------------------------
it('a configured DispatcherDeps.memory reaches the captured TurnInput.memory', async () => { const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT)]) const ledger = new TurnLedger() let captured const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: (uri) => join('/tmp', 'radial-memory-test', uri.replace(/[^a-z0-9]/gi, '_')), image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], memory: '2g', runTurn: async (input) => { captured = input return { outcome: 'fulfilled', acceptedRef: { uri: 'at://x/y/z', cid: 'bafy' }, label: 'radial.turn.x' } }, }) dispatcher.pump(index, actors) await dispatcher.drain() assert.equal(captured.memory, '2g') assert.equal(ledger.get(request.uri).state, 'fulfilled') ledger.close()})
// --- Orphan reconciliation --------------------------------------------------
it('startup orphan reconciliation kills the tracked container and resets the ledger row to eligible', async () => { const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT)]) const ledger = new TurnLedger() const label = 'radial.turn.orphan' ledger.markRunning(request.uri, request.cid, { containerLabel: label, checkoutPath: '/tmp/x' }) assert.equal(selectDispatchable(index, actors, ledger).length, 0)
const killed = [] const runner = new FakeContainerRunner(async () => ({ exitCode: 0, timedOut: false })) // Seed a started "container" under the label the way a real orphan would be tracked. await runner.run({ label, image: 'x', argv: [], env: {}, mounts: [], timeoutMs: 1000, }) const originalKill = runner.kill.bind(runner) runner.kill = async (idOrName) => { killed.push(idOrName) return originalKill(idOrName) }
for (const row of ledger.running()) { await reconcileOrphan(runner, row.containerLabel, row.requestUri) ledger.reset(row.requestUri) }
assert.ok(killed.length > 0) assert.deepEqual(await runner.listByLabel(label), []) assert.equal(ledger.get(request.uri), undefined) assert.deepEqual(uris(selectDispatchable(index, actors, ledger)), [request.uri]) ledger.close()})
// --- finding 9: unassigned request blocks on an unanswered question ----------
it('unassigned request with an unanswered agent question is not dispatched (fresh ledger)', () => { const base = buildScenario() const question = mk(AGENT, COLLECTIONS.message, 'q9', 'cid-q9', { $type: COLLECTIONS.message, goal: base.request.value.goal, body: 'which approach?', mentions: [], re: ref(base.request), createdAt: '2026-01-01T00:03:10Z', }) // Unassigned and CLAIMED by us; the claim holder (AGENT) asked; no human has replied yet. const claim = claimFor(base.request, AGENT) const index = materialize(store(unassign([...base.records, question, claim])), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() // fresh — no awaiting_input row const actors = registryFor([actorFor(AGENT)]) const heldClaims = confirmed(base.request) assert.deepEqual(selectDispatchable(index, actors, ledger, undefined, undefined, heldClaims), [])
// Once a human replies, it becomes dispatchable again. const reply = mk(HUMAN, COLLECTIONS.message, 'a9', 'cid-a9', { $type: COLLECTIONS.message, goal: base.request.value.goal, body: 'approach A', mentions: [], parent: ref(question), createdAt: '2026-01-01T00:03:20Z', }) const index2 = materialize(store(unassign([...base.records, question, reply, claim])), { spaceUri: SPACE_URI }) assert.deepEqual( uris(selectDispatchable(index2, actors, ledger, undefined, undefined, heldClaims)), [base.request.uri], ) ledger.close()})
// --- Review turn end-to-end via TurnDispatcher -------------------------------
it('E2E: a review request drives a real review verdict and fulfills the request in a fresh materialize', async () => { await withTempDir(async (runsDir) => { const pds = new LocalPds(AGENT) const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds)) const client = new CredentialClient(session, pds.fetch.bind(pds)) // No subject commit so the turn takes the default-branch (noopCheckout) path; the commit-pin path // is covered by turn.test.mjs against a checkoutCommit stub. const { records, reviewReq } = withReviewRequest({ subjectLinks: {} }) const recordStore = store(records) const index = materialize(recordStore, { spaceUri: SPACE_URI }) const actors = registryFor([{ ...actorFor(AGENT, ['plan', 'review']), client }]) const ledger = new TurnLedger()
const { dispatcher } = makeDispatcher({ ledger, runsDir, runnerFn: async (spec) => { const response = await sendLine(hostSocketPath(spec), { method: 'submitReview', token: spec.env.RADIAL_TURN_TOKEN, verdict: 'approve', findings: [{ severity: 'info', body: 'looks good' }], }) assert.equal(response.ok, true) return { exitCode: 0, timedOut: false } }, })
dispatcher.pump(index, actors) assert.equal(dispatcher.inFlight, 1) await dispatcher.drain()
const row = ledger.get(reviewReq.uri) assert.equal(row.state, 'fulfilled') const expectedRkey = reviewRkey(reviewReq.uri, reviewReq.cid) assert.equal(row.acceptedRef.uri, `at://${AGENT}/${COLLECTIONS.review}/${expectedRkey}`)
const stored = [...pds.records.values()].filter((r) => r.value.$type === COLLECTIONS.review) assert.equal(stored.length, 1) assert.deepEqual(stored[0].value.subject, reviewReq.value.subject) assert.deepEqual(stored[0].value.request, { uri: reviewReq.uri, cid: reviewReq.cid }) assert.equal(stored[0].value.verdict, 'approve') assert.deepEqual(stored[0].value.findings, [{ severity: 'info', body: 'looks good' }])
// A fresh, independent materialize that has ingested the written verdict shows the review request // fulfilled (no longer open). const written = stored[0] const wrkey = written.uri.slice(written.uri.lastIndexOf('/') + 1) recordStore.put({ did: AGENT, collection: COLLECTIONS.review, rkey: wrkey, uri: written.uri, cid: written.cid, rev: '0000000000009', value: written.value, }) const index2 = materialize(recordStore, { spaceUri: SPACE_URI }) assert.equal(index2.goals[0].openRequests.some((r) => r.uri === reviewReq.uri), false) ledger.close() })})
// --- Project-scoped (system artifact) turn end-to-end ------------------------
it('E2E: a project-scoped adr request lands a project-anchored artifact that a fresh materialize shows fulfilled and in currentSystemArtifacts, then rides in a subsequent goal turn bundle (§8 long-term memory loop)', async () => { await withTempDir(async (runsDir) => { const pds = new LocalPds(AGENT) const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds)) const client = new CredentialClient(session, pds.fetch.bind(pds)) const base = buildScenario() // Fulfill the base plan request so only the project-scoped adr request is open to dispatch. const planArt = mk(AGENT, COLLECTIONS.artifact, 'plan-done', 'cid-plan-done', { $type: COLLECTIONS.artifact, request: ref(base.request), goal: ref(base.goal), type: 'plan', body: 'plan', links: {}, createdAt: '2026-01-01T00:03:30Z', }) const adrReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'adr-req', 'cid-adr-req', { $type: COLLECTIONS.artifactRequest, project: ref(base.project), type: 'adr', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:04:00Z', }) const recordStore = store([...base.records, planArt, adrReq]) const index = materialize(recordStore, { spaceUri: SPACE_URI }) const actors = registryFor([{ ...actorFor(AGENT, ['plan', 'adr']), client }]) const ledger = new TurnLedger()
const { dispatcher } = makeDispatcher({ ledger, runsDir, runnerFn: async (spec) => { const response = await sendLine(hostSocketPath(spec), { method: 'submitArtifact', token: spec.env.RADIAL_TURN_TOKEN, title: 'A short title', body: 'the ADR body', }) assert.equal(response.ok, true) return { exitCode: 0, timedOut: false } }, })
dispatcher.pump(index, actors) assert.equal(dispatcher.inFlight, 1) // only the adr request is dispatchable await dispatcher.drain()
assert.equal(ledger.get(adrReq.uri).state, 'fulfilled') const stored = [...pds.records.values()].filter((r) => r.value.$type === COLLECTIONS.artifact) assert.equal(stored.length, 1) const adr = stored[0] assert.deepEqual(adr.value.project, ref(base.project)) // anchored to the project, not a goal assert.equal(adr.value.goal, undefined) assert.equal(adr.value.type, 'adr')
// Ingest the written adr and re-materialize: the request is fulfilled and the adr is the project's // current system artifact. const arkey = adr.uri.slice(adr.uri.lastIndexOf('/') + 1) recordStore.put({ did: AGENT, collection: COLLECTIONS.artifact, rkey: arkey, uri: adr.uri, cid: adr.cid, rev: '0000000000009', value: adr.value }) const index2 = materialize(recordStore, { spaceUri: SPACE_URI }) const projectView = index2.projects.find((v) => v.target.uri === base.project.uri) assert.equal(projectView.currentSystemArtifacts.some((a) => a.uri === adr.uri), true) assert.equal(projectView.openRequests.some((r) => r.uri === adrReq.uri), false)
// A subsequent GOAL-scoped turn's bundle now carries that adr in currentSystemArtifacts — the // agents' on-protocol long-term memory (Phase 5 exit criterion). const freshPlanReq = mk(HUMAN, COLLECTIONS.artifactRequest, 'plan-2', 'cid-plan-2', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'plan', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:06:00Z', }) recordStore.put(freshPlanReq) const index3 = materialize(recordStore, { spaceUri: SPACE_URI }) const bundle = buildTurnBundle(index3, freshPlanReq.uri, freshPlanReq.cid) assert.ok(bundle.goal) assert.equal(bundle.currentSystemArtifacts.some((a) => a.uri === adr.uri), true) ledger.close() })})
// --- Generalized v2 chaining (plan v2, architecture v2) --------------------//// `prev` used to be stamped for implementations only, so a plan v2 or an architecture v2 landed as// a NEW chain: the version rail showed two v1s, and for a project-scoped type `currentSystemArtifacts`// listed the same document twice. These pin the generalized rule — a predecessor is any resolved// direct basedOn artifact of the SAME type under the SAME target.
const ARCHITECTURE_TYPE = mk(ROOT, COLLECTIONS.artifactType, 'type-arch', 'cid-type-arch', { $type: COLLECTIONS.artifactType, space: { uri: SPACE_URI, cid: 'cid-space' }, name: 'architecture', brief: 'Describe the system architecture.', outputSpec: { format: 'markdown', description: 'An architecture document' }, scope: 'project', durability: 'living', createdAt: '2026-01-01T00:00:14Z',})
/** A landed artifact of `type` under `anchor` (a goal or a project ref), plus the request it answers. */function priorArtifact(anchor, type, rkey, at = '2026-01-01T00:02:30Z') { const req = mk(HUMAN, COLLECTIONS.artifactRequest, `req-${rkey}`, `cid-req-${rkey}`, { $type: COLLECTIONS.artifactRequest, ...anchor, type, basedOn: [], assignee: AGENT, createdAt: at, }) const art = mk(AGENT, COLLECTIONS.artifact, rkey, `cid-${rkey}`, { $type: COLLECTIONS.artifact, request: ref(req), ...anchor, type, body: `prior ${type}`, links: {}, createdAt: at, }) return { req, art }}
it('resolves a same-goal plan predecessor so a plan v2 continues its chain', () => { const { goal } = buildScenario() const v1 = priorArtifact({ goal: ref(goal) }, 'plan', 'plan-v1') const { records, request } = buildScenario({ requestOverrides: { basedOn: [ref(v1.art)] }, extra: [v1.req, v1.art], }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.equal(result[0].predecessor.ambiguous, false) // No branch/pr: a plan artifact writes `links: {}`, so only the ref comes through. assert.deepEqual(result[0].predecessor.artifact, { ref: ref(v1.art) }) ledger.close()})
it('resolves a same-project architecture predecessor (project-scoped v2)', () => { const { project } = buildScenario() const v1 = priorArtifact({ project: ref(project) }, 'architecture', 'arch-v1') const archRequest = mk(HUMAN, COLLECTIONS.artifactRequest, 'arch-v2-req', 'cid-arch-v2-req', { $type: COLLECTIONS.artifactRequest, project: ref(project), type: 'architecture', basedOn: [ref(v1.art)], assignee: AGENT, createdAt: '2026-01-01T00:04:00Z', }) const { records } = buildScenario({ extra: [ARCHITECTURE_TYPE, v1.req, v1.art, archRequest] }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan', 'architecture'])]) const arch = selectDispatchable(index, actors, ledger).find((d) => d.request.uri === archRequest.uri) assert.ok(arch) assert.deepEqual(arch.predecessor.artifact, { ref: ref(v1.art) }) ledger.close()})
it('does not treat a different type or a different target as a predecessor', () => { const base = buildScenario() // Same goal, different type; and same type, different goal — neither is a predecessor. const otherType = priorArtifact({ goal: ref(base.goal) }, 'plan', 'plan-other') const otherGoal = mk(HUMAN, COLLECTIONS.goal, 'goal-2', 'cid-goal-2', { $type: COLLECTIONS.goal, space: { uri: SPACE_URI, cid: 'cid-space' }, project: ref(base.project), title: 'A second goal', body: 'elsewhere', createdAt: '2026-01-01T00:02:05Z', }) const elsewhere = priorArtifact({ goal: ref(otherGoal) }, 'implementation', 'impl-elsewhere') const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(otherType.art), ref(elsewhere.art)] }, extra: [IMPL_TYPE, otherGoal, otherType.req, otherType.art, elsewhere.req, elsewhere.art], }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const result = selectDispatchable(index, actors, ledger) assert.deepEqual(uris(result), [request.uri]) assert.deepEqual(result[0].predecessor, { candidates: [], ambiguous: false }) ledger.close()})
it('holds any type of request whose basedOn is not yet ingested (stale index)', () => { const { records, request } = buildScenario({ requestOverrides: { basedOn: [{ uri: `at://${AGENT}/${COLLECTIONS.artifact}/not-yet`, cid: 'cid-not-yet' }], }, }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['plan'])]) const rejects = [] const result = selectDispatchable(index, actors, ledger, undefined, (uri, reason) => rejects.push([uri, reason]), ) assert.deepEqual(result, []) assert.deepEqual(rejects, [[request.uri, "'plan' request has an unresolved basedOn ref (stale index)"]]) ledger.close()})
it('dispatcher threads prev into a non-implementation turn and never a branch', async () => { const { goal } = buildScenario() const v1 = priorArtifact({ goal: ref(goal) }, 'plan', 'plan-v1') const { records, request } = buildScenario({ requestOverrides: { basedOn: [ref(v1.art)] }, extra: [v1.req, v1.art], }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['plan'])]) const ledger = new TurnLedger() let captured const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-plan-v2', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], runTurn: async (input) => { captured = input return { outcome: 'fulfilled', acceptedRef: { uri: 'at://did:plc:agent/artifact/plan-v2', cid: 'cid-plan-v2' }, label: 'radial.turn.plan', } }, })
dispatcher.pump(index, actors) await dispatcher.drain()
assert.deepEqual(captured.prev, ref(v1.art)) assert.equal(captured.branch, undefined) // branch stays implementation-only assert.equal(ledger.get(request.uri).state, 'fulfilled') ledger.close()})
it('stamps this daemon instance on the container label it tracks and the one the turn runs under', async () => { // The ledger row and the turn have to agree about the label, because that string is what shutdown // and startup orphan reconciliation kill by — and with two daemons on one machine, killing by the // OTHER instance's label is exactly the accident the instance segment prevents. const { records, request } = buildScenario() const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['plan'])]) const ledger = new TurnLedger() let captured const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-instance-label', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], instance: '4f2a91', runTurn: async (input) => { captured = input return { outcome: 'fulfilled', acceptedRef: { uri: 'at://x', cid: 'c' }, label: 'l' } }, }) dispatcher.pump(index, actors) const expected = turnContainerLabel(request.uri, request.cid, '4f2a91') assert.equal(ledger.get(request.uri).containerLabel, expected) assert.equal(captured.instance, '4f2a91') assert.notEqual(expected, turnContainerLabel(request.uri, request.cid)) await dispatcher.drain() ledger.close()})
it('rejects an ambiguous project-scoped request by anchoring the explanation to a candidate', async () => { const base = buildScenario() const a = priorArtifact({ project: ref(base.project) }, 'architecture', 'arch-a') const b = priorArtifact({ project: ref(base.project) }, 'architecture', 'arch-b') const archRequest = mk(HUMAN, COLLECTIONS.artifactRequest, 'arch-ambig', 'cid-arch-ambig', { $type: COLLECTIONS.artifactRequest, project: ref(base.project), type: 'architecture', basedOn: [ref(a.art), ref(b.art)], assignee: AGENT, createdAt: '2026-01-01T00:04:00Z', }) const { records } = buildScenario({ extra: [ARCHITECTURE_TYPE, a.req, a.art, b.req, b.art, archRequest], }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const pds = new LocalPds(AGENT) const client = new CredentialClient( await createSession(pds.service, pds.handle, 'password', (input, init) => pds.fetch(input, init)), (input, init) => pds.fetch(input, init), ) const actors = registryFor([{ ...actorFor(AGENT, ['plan', 'architecture']), client }]) const ledger = new TurnLedger() let launched = false const dispatcher = new TurnDispatcher({ ledger, concurrency: 2, runDirFor: () => '/tmp/radial-arch-ambig', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], runTurn: async () => { launched = true return { outcome: 'crashed', label: 'x' } }, })
dispatcher.pump(index, actors) await dispatcher.drain()
// No container ran for the ambiguous request, and it was given up rather than retried. assert.equal(ledger.get(archRequest.uri).state, 'gave_up') const message = [...pds.records.values()].find((entry) => entry.value.$type === COLLECTIONS.message) assert.ok(message) assert.match(message.value.body, /more than one architecture artifact of this project/) // A project-scoped message cannot anchor to a goal, so it anchors to the lowest-uri candidate. assert.equal(message.value.goal, undefined) assert.deepEqual(message.value.artifact, ref(a.art)) assert.deepEqual(message.value.re, ref(archRequest)) void launched ledger.close()})
// --- v2 branch reuse without a forge adapter, and branch contention ----------
/** A trusted merge annotation naming an implementation artifact. */const mergeOf = (art, rkey) => mk(AGENT, COLLECTIONS.merge, rkey, `cid-${rkey}`, { $type: COLLECTIONS.merge, artifact: ref(art), pr: art.value.links.pr, mergedAt: '2026-01-01T00:03:00Z', createdAt: '2026-01-01T00:03:00Z', })
/** The v2 reuse scenario, on a github project so the repo guard can run. */function reuseScenario({ predecessorLinks, extra = [], basedOn } = {}) { const { goal } = buildScenario() const predecessor = implPredecessor(goal, 'impl-v1', predecessorLinks) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: basedOn ?? [ref(predecessor.art)] }, extra: [IMPL_TYPE, predecessor.req, predecessor.art, ...extra(predecessor)], }) const githubRecords = records.map((record) => record.collection === COLLECTIONS.project ? { ...record, value: { ...record.value, gitUrl: 'https://github.com/o/r.git' } } : record, ) return { predecessor, request, index: materialize(store(githubRecords), { spaceUri: SPACE_URI }) }}
function reuseDispatcher(ledger, { forge, branchExists = async () => true } = {}) { const captured = [] const logs = [] const dispatcher = new TurnDispatcher({ ledger, concurrency: 2, runDirFor: () => '/tmp/radial-v2-index', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, ...(forge ? { forge } : {}), // No test may reach the real `git ls-remote`; the default says the branch is still there, which // is the state every pre-existing assertion here was written against. branchExists, log: (message) => logs.push(message), runTurn: async (input) => { captured.push(input) return { outcome: 'fulfilled', acceptedRef: { uri: `at://did:plc:agent/artifact/${captured.length}`, cid: `cid-${captured.length}` }, label: 'radial.turn.impl', } }, }) return { dispatcher, captured, logs }}
it('resolves a predecessor’s merged-ness off the index, so reuse can be decided without a forge', () => { const links = { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' } const unmerged = reuseScenario({ predecessorLinks: links, extra: () => [] }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) assert.equal( selectDispatchable(unmerged.index, actors, ledger)[0].predecessor.artifact.merged, undefined, )
const merged = reuseScenario({ predecessorLinks: links, extra: (predecessor) => [mergeOf(predecessor.art, 'mrg-1')], }) assert.equal( selectDispatchable(merged.index, actors, ledger)[0].predecessor.artifact.merged, true, ) ledger.close()})
it('with no forge adapter, continues an in-namespace branch whose predecessor carries no merge record', async () => { // The single likeliest reason a v2 opened a second pull request: `run.forge` is easy to omit, the // reuse branch was unreachable without it, and nothing said so. The index answers instead. const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const { dispatcher, captured, logs } = reuseDispatcher(ledger)
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.equal(captured[0].checkoutRef, 'radial/impl-v1') assert.equal(captured[0].branch, 'radial/impl-v1') assert.equal(ledger.get(scenario.request.uri).branch, 'radial/impl-v1') // The log says which signal it used and what that signal cannot see, rather than claiming certainty. const line = logs.find((entry) => entry.includes('on the index signal')) assert.ok(line, `expected an index-signal log, got: ${logs.join(' | ')}`) assert.match(line, /closed without merging/) ledger.close()})
it('with no forge adapter, opens a fresh branch when the predecessor’s branch is gone from the remote', async () => { // The guard that actually protects the no-forge fallback. `merged` cannot fire there — merge // annotations are written by the merge poller, which only runs with a forge configured — so // without this probe a v2 off a merged predecessor whose head branch was deleted would clone a // ref that no longer exists, crash, and crash identically on every retry until it gave up. const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const probed = [] const { dispatcher, captured, logs } = reuseDispatcher(ledger, { branchExists: async (input) => { probed.push(input) return false }, })
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.deepEqual(probed, [{ gitUrl: 'https://github.com/o/r.git', branch: 'radial/impl-v1' }]) assert.equal(captured[0].checkoutRef, undefined) assert.equal(captured[0].branch, implBranchName(scenario.request.uri, scenario.request.cid)) assert.equal(ledger.get(scenario.request.uri).branch, captured[0].branch) assert.ok( logs.some((entry) => entry.includes('no longer exists on https://github.com/o/r.git')), logs.join(' | '), ) ledger.close()})
it('opens a fresh branch when the branch probe itself fails, rather than gambling on the clone', async () => { const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const { dispatcher, captured, logs } = reuseDispatcher(ledger, { branchExists: async () => { throw new Error('ls-remote: could not read Username') }, })
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.equal(captured[0].checkoutRef, undefined) assert.equal(captured[0].branch, implBranchName(scenario.request.uri, scenario.request.cid)) assert.ok(logs.some((entry) => entry.includes('could not read Username')), logs.join(' | ')) ledger.close()})
it('does not probe the remote when a forge can answer for the pull request itself', async () => { // With a forge the PR state names the head ref, so the branch's existence is already established; // the probe is a fallback, not a second opinion. const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) let probes = 0 const { dispatcher, captured } = reuseDispatcher(ledger, { forge: { getPullRequestState: async () => ({ state: 'open', headRef: 'radial/impl-v1', headRepoFullName: 'o/r', baseRef: 'main', }), }, branchExists: async () => { probes += 1 return true }, })
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.equal(probes, 0) assert.equal(captured[0].checkoutRef, 'radial/impl-v1') ledger.close()})
it('with no forge adapter, opens a fresh branch once the predecessor is recorded as merged', async () => { const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: (predecessor) => [mergeOf(predecessor.art, 'mrg-1')], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const { dispatcher, captured } = reuseDispatcher(ledger)
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.equal(captured[0].checkoutRef, undefined) assert.equal(captured[0].branch, implBranchName(scenario.request.uri, scenario.request.cid)) assert.equal(ledger.get(scenario.request.uri).branch, captured[0].branch) ledger.close()})
it('never reuses a branch outside the reserved namespace, forge or no forge', async () => { const scenario = reuseScenario({ predecessorLinks: { branch: 'main', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const { dispatcher, captured } = reuseDispatcher(ledger)
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.equal(captured[0].checkoutRef, undefined) assert.equal(captured[0].branch, implBranchName(scenario.request.uri, scenario.request.cid)) ledger.close()})
it('records the branch a forge-confirmed reuse took on the ledger row', async () => { const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const { dispatcher, logs } = reuseDispatcher(ledger, { forge: { getPullRequestState: async () => ({ state: 'open', headRef: 'radial/impl-v1', headRepoFullName: 'o/r', baseRef: 'main', }), }, })
dispatcher.pump(scenario.index, actors) await dispatcher.drain()
assert.equal(ledger.get(scenario.request.uri).branch, 'radial/impl-v1') // The success path logs too; before this only the DECLINE paths said anything. assert.ok(logs.some((entry) => entry.includes('the forge confirms')), logs.join(' | ')) ledger.close()})
it('serializes two v2 requests that resolve the same predecessor branch', async () => { // Both resolve one predecessor, so both would push to one branch. `#inFlight` is keyed by request // uri and does not see it; with concurrency 2 they would interleave. const { goal } = buildScenario() const predecessor = implPredecessor(goal, 'impl-v1', { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7', }) const second = mk(HUMAN, COLLECTIONS.artifactRequest, 'req-v2b', 'cid-req-v2b', { $type: COLLECTIONS.artifactRequest, goal: ref(goal), type: 'implementation', basedOn: [ref(predecessor.art)], assignee: AGENT, createdAt: '2026-01-01T00:03:00Z', }) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(predecessor.art)] }, extra: [IMPL_TYPE, predecessor.req, predecessor.art, second], }) const githubRecords = records.map((record) => record.collection === COLLECTIONS.project ? { ...record, value: { ...record.value, gitUrl: 'https://github.com/o/r.git' } } : record, ) const index = materialize(store(githubRecords), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['implementation'])]) const ledger = new TurnLedger()
let release const gate = new Promise((resolve) => { release = resolve }) const started = [] const logs = [] const dispatcher = new TurnDispatcher({ ledger, concurrency: 2, runDirFor: () => '/tmp/radial-v2-contended', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, branchExists: async () => true, log: (message) => logs.push(message), runTurn: async (input) => { started.push(input.request.uri) await gate return { outcome: 'fulfilled', acceptedRef: { uri: `at://did:plc:agent/artifact/${started.length}`, cid: `cid-${started.length}` }, label: 'radial.turn.impl', } }, })
// Both are dispatchable and concurrency is 2, so only the branch guard can hold the second back. assert.equal(selectDispatchable(index, actors, ledger).length, 2) dispatcher.pump(index, actors) await new Promise((resolve) => setImmediate(resolve)) assert.equal(started.length, 1) assert.ok(logs.some((entry) => entry.includes('is already being pushed by')), logs.join(' | ')) // Contention is not a failure: the skipped request's ledger row is untouched and still eligible. const waiting = [request.uri, second.uri].find((uri) => uri !== started[0]) assert.equal(ledger.get(waiting), undefined) assert.equal(ledger.eligible(waiting), true)
release() await dispatcher.drain()
// Once the branch is free the next pump takes it. dispatcher.pump(index, actors) await new Promise((resolve) => setImmediate(resolve)) assert.deepEqual(started.slice(1), [waiting]) release() await dispatcher.drain() ledger.close()})
it('does not launch overlapping turns for a harness with one refreshable login', async () => { const base = buildScenario() const second = mk(HUMAN, COLLECTIONS.artifactRequest, 'request-2', 'cid-request-2', { $type: COLLECTIONS.artifactRequest, goal: ref(base.goal), type: 'plan', basedOn: [], assignee: AGENT, createdAt: '2026-01-01T00:03:01Z', }) const { records, request } = buildScenario({ extra: [second] }) const index = materialize(store(records), { spaceUri: SPACE_URI }) const actors = registryFor([actorFor(AGENT, ['plan'], [], 'codex')]) const ledger = new TurnLedger() let release const gate = new Promise((resolve) => { release = resolve }) const started = [] const logs = [] const dispatcher = new TurnDispatcher({ ledger, concurrency: 2, serializedHarnesses: ['codex'], runDirFor: (uri) => `/tmp/${uri.slice(-8)}`, image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], log: (message) => logs.push(message), runTurn: async (input) => { started.push(input.request.uri) await gate return { outcome: 'fulfilled', acceptedRef: { uri: `at://did:plc:agent/artifact/${started.length}`, cid: `cid-${started.length}` }, label: 'radial.turn.codex', } }, })
dispatcher.pump(index, actors) await new Promise((resolve) => setImmediate(resolve)) assert.equal(started.length, 1) assert.ok(logs.every((entry) => !entry.startsWith('not dispatching ')), logs.join(' | ')) const waiting = [request.uri, second.uri].find((uri) => uri !== started[0]) assert.equal(ledger.get(waiting), undefined)
release() await dispatcher.drain() dispatcher.pump(index, actors) await dispatcher.drain() assert.equal(started.length, 2) ledger.close()})
// --- per-project forge selection --------------------------------------------
/** A registry adapter for a made-up forge, so the routing can be asserted without a network. */const fakeAdapter = (overrides = {}) => ({ kind: 'tangled', canObserve: true, matchesGitUrl: (gitUrl) => gitUrl.includes('tangled.org'), ownsPullUrl: () => true, pullBelongsToProject: async () => true, projectIdentity: async () => 'did:plc:owner', getPullRequestState: async () => ({ state: 'open', headRef: 'radial/impl-v1', headRepoFullName: 'did:plc:owner', baseRef: 'main' }), compareUrl: () => '', ...overrides,})
/** The v2 reuse scenario against a project on some other forge. */function tangledReuseScenario(links) { const { goal } = buildScenario() const predecessor = implPredecessor(goal, 'impl-v1', links) const { records, request } = buildScenario({ requestOverrides: { type: 'implementation', basedOn: [ref(predecessor.art)] }, extra: [IMPL_TYPE, predecessor.req, predecessor.art], }) const tangledRecords = records.map((record) => record.collection === COLLECTIONS.project ? { ...record, value: { ...record.value, gitUrl: 'https://tangled.org/@alice/widget' } } : record, ) return { predecessor, request, index: materialize(store(tangledRecords), { spaceUri: SPACE_URI }) }}
it('blocks an implementation per PROJECT, so one forge’s missing credential does not stop the other', async () => { // The bug the registry exists to fix: `implementationEnabled` was one boolean over the whole // daemon, so a missing GitHub token stopped implementations on projects that never needed one. const scenario = tangledReuseScenario({ branch: 'radial/impl-v1', pr: 'at://did:plc:agent/sh.tangled.repo.pull/3k' }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])])
const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-forge-gate', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], // GitHub is unauthenticated daemon-wide... implementationEnabled: false, // ...but this project is on a forge whose adapter says it is ready. forges: new ForgeRegistry([fakeAdapter({ implementationBlockedReason: () => undefined })]), branchExists: async () => true, log: () => {}, runTurn: async () => ({ outcome: 'fulfilled', acceptedRef: { uri: 'at://x', cid: 'c' }, label: 'l' }), }) dispatcher.pump(scenario.index, actors) await new Promise((resolve) => setImmediate(resolve)) await dispatcher.drain() assert.equal(ledger.get(scenario.request.uri).state, 'fulfilled') ledger.close()})
it('reports the matching adapter’s own reason when it cannot run an implementation', async () => { const scenario = tangledReuseScenario({ branch: 'radial/impl-v1', pr: 'at://did:plc:agent/sh.tangled.repo.pull/3k' }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const logs = [] const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-forge-gate-2', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, forges: new ForgeRegistry([fakeAdapter({ implementationBlockedReason: () => 'no tangled push key is configured' })]), branchExists: async () => true, log: (message) => logs.push(message), runTurn: async () => { throw new Error('must not launch') }, }) dispatcher.pump(scenario.index, actors) await new Promise((resolve) => setImmediate(resolve)) assert.ok(logs.some((entry) => entry.includes('no tangled push key is configured')), logs.join(' | ')) assert.equal(ledger.get(scenario.request.uri), undefined) ledger.close()})
it('asks the project’s own adapter whether the predecessor pull is this project’s, and reuses on its answer', async () => { const scenario = tangledReuseScenario({ branch: 'radial/impl-v1', pr: 'at://did:plc:agent/sh.tangled.repo.pull/3k' }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const captured = [] const logs = [] const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-forge-reuse', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, forges: new ForgeRegistry([fakeAdapter()]), branchExists: async () => { throw new Error('an observing adapter answers; no remote probe should be needed') }, log: (message) => logs.push(message), runTurn: async (input) => { captured.push(input) return { outcome: 'fulfilled', acceptedRef: { uri: 'at://x', cid: 'c' }, label: 'l' } }, }) dispatcher.pump(scenario.index, actors) await new Promise((resolve) => setImmediate(resolve)) await dispatcher.drain() assert.equal(captured[0].branch, 'radial/impl-v1') assert.equal(captured[0].checkoutRef, 'radial/impl-v1') assert.ok(logs.some((entry) => entry.includes('the forge confirms')), logs.join(' | ')) ledger.close()})
it('records a daemon-opened pull request on the ledger row, so a retry finds it', async () => { const scenario = tangledReuseScenario({ branch: 'radial/impl-v1', pr: 'at://did:plc:agent/sh.tangled.repo.pull/3k' }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-forge-pull', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, forges: new ForgeRegistry([fakeAdapter()]), branchExists: async () => true, log: () => {}, runTurn: async () => ({ outcome: 'fulfilled', acceptedRef: { uri: 'at://x', cid: 'c' }, label: 'l', pull: { url: 'at://did:plc:agent/sh.tangled.repo.pull/3k', record: { uri: 'at://did:plc:agent/sh.tangled.repo.pull/3k', cid: 'pullcid' } }, }), }) dispatcher.pump(scenario.index, actors) await new Promise((resolve) => setImmediate(resolve)) await dispatcher.drain() assert.equal(ledger.get(scenario.request.uri).pullUri, 'at://did:plc:agent/sh.tangled.repo.pull/3k') ledger.close()})
it('a GitHub project still routes to the GitHub adapter when a registry holds both', async () => { const scenario = reuseScenario({ predecessorLinks: { branch: 'radial/impl-v1', pr: 'https://github.com/o/r/pull/7' }, extra: () => [], }) const ledger = new TurnLedger() const actors = registryFor([actorFor(AGENT, ['implementation'])]) const captured = [] const logs = [] const dispatcher = new TurnDispatcher({ ledger, concurrency: 1, runDirFor: () => '/tmp/radial-forge-both', image: 'radial-turn:test', timeoutMs: 30_000, allowedSchemes: ['https'], implementationEnabled: true, forges: new ForgeRegistry([ new GitHubForge({ token: 'gh', fetch: async () => new Response( JSON.stringify({ state: 'open', head: { ref: 'radial/impl-v1', repo: { full_name: 'o/r' } }, base: { ref: 'main' } }), { status: 200 }, ), }), fakeAdapter({ getPullRequestState: async () => { throw new Error('the tangled adapter must not be asked about a github project') }, }), ]), branchExists: async () => true, log: (message) => logs.push(message), runTurn: async (input) => { captured.push(input) return { outcome: 'fulfilled', acceptedRef: { uri: 'at://x', cid: 'c' }, label: 'l' } }, }) dispatcher.pump(scenario.index, actors) await new Promise((resolve) => setImmediate(resolve)) await dispatcher.drain() assert.equal(captured[0].branch, 'radial/impl-v1') assert.ok(logs.some((entry) => entry.includes('the forge confirms')), logs.join(' | ')) ledger.close()})