Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979// The Phase 7 failure drills (design §12, plan §Phase 7 F1–F5), plus the clock-skew drills F6–F8.//// Two independently-configured daemon stacks — two record stores, two turn ledgers, two claim// ledgers, two agent identities on two PDSes, and two CLOCKS — over one shared fake network.// Everything is offline and driven by an explicit fake clock and explicit ticks: no Docker, no// wall-clock waits, no timers (plan §6 R8). Each drill asserts a claim the exit criterion makes, and// the one-ingestion-cycle bound is asserted NUMERICALLY rather than "eventually".//// The two clocks are load-bearing. F1–F5 were written against a single shared clock, which made// every question about skew unaskable: two operators that agree about the time cannot disagree// about a lease. F6–F8 give each operator its own offset from the drill's true time and ask what// actually breaks — a reader running fast (F6), a writer that jumps forward mid-lease (F7, and its// unretractable record in F7b), and a writer running slow that wins every race (F8, which asserts// the exposure that is deliberately NOT fixed; see operators.md §8).
import assert from 'node:assert/strict'import { mkdtemp, rm } from 'node:fs/promises'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, indexDigest, digestDifferences, materialize,} from '../../core/dist/index.js'import { MemorySyncStateStore, RepoPoller, SpaceIngestor } from '../../ingest/dist/index.js'import { LocalPds } from '../../atproto/test/local-pds.mjs'import { ClaimLedger, ClaimManager, TurnDispatcher, TurnLedger, claimRkey, selectDispatchable,} from '../dist/index.js'
const ROOT = 'did:plc:root'const AGENT_A = 'did:plc:agenta'const AGENT_B = 'did:plc:agentb'const OUTSIDER = 'did:plc:outsider'const SPACE_URI = `at://${ROOT}/${COLLECTIONS.space}/space`
const TIMING = { leaseMs: 600_000, renewIntervalMs: 200_000, confirmCycles: 1, maxOutstanding: 2 }/** `run.concurrency`'s default: one turn at a time, which is also one claim's worth of capacity. */const CONCURRENCY = 1const T0 = Date.parse('2026-01-01T00:00:00.000Z')
/** Writes a record straight into a repo, as though its owner had already created it. */function seed(pds, collection, rkey, cid, value) { const uri = `at://${pds.did}/${collection}/${rkey}` pds.seq += 1 pds.records.set(uri, { uri, cid, value }) return { uri, cid, value }}
const ref = (record) => ({ uri: record.uri, cid: record.cid })
/** * The shared network: one repo per DID, and a `RepoReadTransport` over all of them. `hidden` is how * a drill gives one operator a STALE view — the whole point of a multi-operator test is that the * two do not see the same records at the same moment. */function network(repos) { return (hidden = new Set()) => ({ async resolvePds(did) { return new URL(repos.get(did)?.service ?? 'https://missing.pds.test') }, async getLatestCommit(did) { const pds = repos.get(did) if (!pds) return { rev: '0000000000000', commitCid: 'empty' } // The head has to move on every write — including a lease RENEWAL, which rewrites a record // rather than adding one — and also when a hidden record is revealed to this operator, or the // poller short-circuits on an unchanged head and the change is never ingested. `seq` counts // writes; `visible` counts what this operator is allowed to see. Both only ever grow. const visible = [...pds.records.values()].filter((record) => !hidden.has(record.uri)).length const head = pds.seq * 100 + visible return { rev: String(head).padStart(13, '0'), commitCid: `head-${did}-${head}` } }, async listRecords({ did, collection }) { const pds = repos.get(did) if (!pds) return { records: [] } return { records: [...pds.records.values()].filter( (record) => record.value.$type === collection && !hidden.has(record.uri), ), } }, })}
/** The space every drill runs in: one human admin, two agent members, one OPEN plan request. */function seedSpace(repos) { const root = repos.get(ROOT) const space = seed(root, COLLECTIONS.space, 'space', 'cid-space', { $type: COLLECTIONS.space, name: 'Radial', description: 'multi-operator drills', createdAt: '2026-01-01T00:00:00Z', }) const memberships = new Map() for (const [index, did] of [AGENT_A, AGENT_B].entries()) { memberships.set( did, seed(root, COLLECTIONS.addMember, `member-${index}`, `cid-member-${index}`, { $type: COLLECTIONS.addMember, space: ref(space), did, kind: 'agent', role: 'agent', createdAt: `2026-01-01T00:00:0${index + 1}Z`, }), ) } seed(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', }) const project = seed(root, 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 = seed(root, 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', }) // No assignee: the whole point. Only a claim can make this dispatchable. const request = seed(root, COLLECTIONS.artifactRequest, 'request-1', 'cid-request-1', { $type: COLLECTIONS.artifactRequest, goal: ref(goal), type: 'plan', basedOn: [], createdAt: '2026-01-01T00:03:00Z', }) return { space, project, goal, request, memberships }}
/** * One operator's whole stack, independent of the other's in every durable way — INCLUDING its * clock. * * `clock` is the drill's single true time, the only thing a drill advances. `offsetMs` is how far * this operator's own clock runs ahead of (or behind) it, and every place this stack reads a clock * goes through `local()`: the claim manager's lease arithmetic, the `asOf` it folds at, the * dispatch scan, the first-sighting stamp its poller writes, and the records it authors. Two * operators sharing one clock is the shape the first five drills were written in, and it is exactly * what made every skew question below unaskable. */async function operator(did, repos, transportFor, clock, offsetMs = 0) { let offset = offsetMs const local = () => clock.now + offset const localIso = () => new Date(local()).toISOString() const pds = repos.get(did) const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds)) const client = new CredentialClient(session, pds.fetch.bind(pds)) const records = new MemoryRecordStore() const hidden = new Set() const poller = new RepoPoller(transportFor(hidden), records, new MemorySyncStateStore(), { now: localIso, }) const ingestor = new SpaceIngestor(SPACE_URI, poller, records) const claims = new ClaimLedger() const turns = new TurnLedger() const actor = { did, profile: 'planner', artifactTypes: ['plan'], harness: 'claude', models: [], session, client, } const registry = { all: [actor], byDid: new Map([[did, [actor]]]), select: (candidate, type) => (candidate === did && type === 'plan' ? actor : undefined), } const abandoned = [] const logs = [] const manager = new ClaimManager({ ledger: claims, turns, timing: TIMING, now: local, // The production wiring, at the default `run.concurrency: 1` — `concurrency - inFlight`, with a // running turn standing in for a turn the dispatcher is holding a slot for. Without this dep the // drills claim against infinite capacity, which is not what any daemon does and hides the case // where the slot in question belongs to the very turn the claim is meant to cover. capacity: () => CONCURRENCY - turns.running().length, abandon: (uri, reason) => { abandoned.push({ uri, reason }) // The production wiring (cli.ts) hands this straight to `TurnDispatcher.abandon`. self.onAbandon?.(uri, reason) }, log: (message) => logs.push(message), }) const self = { did, pds, client, records, hidden, claims, turns, registry, manager, abandoned, logs, syncs: 0, /** Move this operator's clock relative to the drill's true time — an NTP step, a resumed VM. */ skew(ms) { offset = ms }, /** Set by a drill that runs a real dispatcher, to complete the production wiring. */ onAbandon: undefined, index() { return materialize(records, { spaceUri: SPACE_URI, asOf: localIso() }) }, /** One daemon tick: ingest, then pump claims, exactly as `runDaemon` orders them. */ async tick() { const index = await self.syncOnly() manager.pump(index, registry) await manager.drain() return index }, /** Ingest without pumping — for asserting what a fold says at an exact moment. */ async syncOnly() { self.syncs += 1 await ingestor.sync() return self.index() }, dispatchable(index = self.index()) { return selectDispatchable( index, registry, turns, localIso(), undefined, claims.heldRequests(), ) }, /** What a turn does, reduced to its only protocol-visible effect. */ async fulfil(request, goal, body = 'A plan.') { const written = await client.create(COLLECTIONS.artifact, { $type: COLLECTIONS.artifact, request: { uri: request.uri, cid: request.cid }, goal: { uri: goal.uri, cid: goal.cid }, type: 'plan', body, links: {}, criteria: [], createdAt: localIso(), }) turns.markFulfilled(request.uri, written) return written }, close() { claims.close() turns.close() }, } return self}
/** Claim, wait out the confirm window, and end up `held` — three ticks with `confirmCycles: 1`. */async function claimAndConfirm(op) { await op.tick() // writes the claim await op.tick() // sees it win; starts the confirm window return op.tick() // confirm window elapsed → held}
async function setup({ offsetA = 0, offsetB = 0 } = {}) { const repos = new Map( [ROOT, AGENT_A, AGENT_B].map((did) => [did, new LocalPds(did)]), ) const clock = { now: T0 } const transportFor = network(repos) const space = seedSpace(repos) const a = await operator(AGENT_A, repos, transportFor, clock, offsetA) const b = await operator(AGENT_B, repos, transportFor, clock, offsetB) return { repos, clock, space, a, b, close: () => [a, b].forEach((op) => op.close()) }}
const artifactsIn = (index) => index.goals.flatMap((view) => view.artifacts)
// --- F1: crash mid-turn ------------------------------------------------------
it('drill 1: a crashed daemon lease lapses, the other claims and fulfils, and exactly one artifact exists', async () => { const { clock, space, a, b, close } = await setup()
await claimAndConfirm(a) assert.deepEqual([...a.claims.heldRequests()], [space.request.uri]) assert.deepEqual( a.dispatchable().map((item) => item.request.uri), [space.request.uri], )
// Operator A dies here: no more ticks, no more renewals, and — per design §11 — no way to retract // the claim it holds. B must wait the lease out; nothing else can free the work. await b.tick() assert.deepEqual(b.dispatchable(), [], 'B must not touch a request A holds a live claim on')
clock.now = T0 + TIMING.leaseMs + 1 const lapsed = await b.tick() assert.equal( lapsed.goals[0].winningClaims[space.request.uri], undefined, 'the lapsed lease leaves the request unclaimed on B index', ) await b.tick() await b.tick() assert.deepEqual([...b.claims.heldRequests()], [space.request.uri]) const dispatchable = b.dispatchable() assert.deepEqual(dispatchable.map((item) => item.request.uri), [space.request.uri]) assert.equal(dispatchable[0].actor.did, AGENT_B)
await b.fulfil(space.request, space.goal) const afterB = await b.tick() assert.equal(artifactsIn(afterB).length, 1) assert.equal(afterB.goals[0].openRequests.length, 0)
// And on A's index too, once it comes back: one artifact, from B, and A's own claim settles. const afterA = await a.tick() assert.equal(artifactsIn(afterA).length, 1) assert.equal(artifactsIn(afterA)[0].did, AGENT_B) assert.equal(a.claims.get(space.request.uri, AGENT_A).state, 'settled') close()})
// --- F2: competing claims ----------------------------------------------------
it('drill 2: two daemons claim at once — one dispatch, the loser gives up within ONE ingestion cycle', async () => { const directory = await mkdtemp(join(tmpdir(), 'radial-drill-')) try { const { clock, space, a, b, close } = await setup()
// B claims a second earlier than A, so B wins the tie-break on both indexes. A cannot see it // yet — the ordinary state of two operators mid-race. await b.tick() const theirClaimUri = `at://${AGENT_B}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}` a.hidden.add(theirClaimUri) b.hidden.add(`at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`)
clock.now = T0 + 1_000 await claimAndConfirm(a) assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])
// A dispatches on the strength of the claim it believes it won. A real container starts. let settleTurn const running = new Promise((resolve) => { settleTurn = resolve }) const killed = [] const dispatcher = new TurnDispatcher({ ledger: a.turns, concurrency: 1, runDirFor: (uri) => join(directory, uri.slice(-8)), image: 'radial-turn:test', timeoutMs: 60_000, allowedSchemes: ['https'], heldClaims: () => a.claims.heldRequests(), kill: async (label) => { killed.push(label) settleTurn({ outcome: 'crashed' }) }, runTurn: () => running, log: () => {}, }) a.onAbandon = (uri, reason) => dispatcher.abandon(uri, reason) dispatcher.pump(a.index(), a.registry) assert.equal(dispatcher.inFlight, 1, 'A launched a turn for the claim it thought it held')
// Now A ingests B's earlier claim. The bound the exit criterion asserts: ONE ingestion cycle // from seeing the losing evidence to giving up — not "eventually". a.hidden.delete(theirClaimUri) const syncsBefore = a.syncs await a.tick() assert.equal(a.syncs - syncsBefore, 1) assert.equal(a.claims.get(space.request.uri, AGENT_A).state, 'lost') assert.equal(a.abandoned.length, 1) assert.match(a.abandoned[0].reason, new RegExp(AGENT_B)) await dispatcher.drain() assert.deepEqual(killed.length, 1, 'the container is killed exactly once') assert.equal(dispatcher.inFlight, 0) assert.equal(a.turns.get(space.request.uri).state, 'crashed')
// A never dispatches it again, however many ticks pass. await a.tick() await a.tick() assert.deepEqual(a.dispatchable(), []) assert.equal(a.abandoned.length, 1, 'abandon is called at most once')
// B, the winner, finishes the work. Exactly one artifact exists in the space. b.hidden.clear() await b.tick() await b.tick() assert.deepEqual([...b.claims.heldRequests()], [space.request.uri]) await b.fulfil(space.request, space.goal) const finalB = await b.tick() const finalA = await a.tick() assert.equal(artifactsIn(finalB).length, 1) assert.equal(artifactsIn(finalA).length, 1) assert.equal(artifactsIn(finalA)[0].did, AGENT_B) // Both operators' folds agree about everything, which is the real exit criterion. assert.deepEqual(indexDigest(finalA), indexDigest(finalB)) close() } finally { await rm(directory, { recursive: true, force: true }) }})
// --- F3: non-member injection ------------------------------------------------
it('drill 3: a non-member claim never counts, never gates dispatch, and says why in `ignored`', async () => { const { space, a, b, close } = await setup()
// However it reached the store — a stray poll, a stream that was not filtering — the fold is what // decides, and an outsider holds no grant. const injected = { did: OUTSIDER, collection: COLLECTIONS.claim, rkey: 'claim-injected', uri: `at://${OUTSIDER}/${COLLECTIONS.claim}/claim-injected`, cid: 'cid-claim-injected', rev: '0000000000001', value: { $type: COLLECTIONS.claim, request: { uri: space.request.uri, cid: space.request.cid }, // Earlier than anything either daemon will write, so an unfiltered tie-break would hand it // the request outright. expiresAt: '2027-01-01T00:00:00.000Z', createdAt: '2025-12-31T00:00:00.000Z', }, } for (const op of [a, b]) op.records.put(injected)
const index = await a.tick() assert.equal(index.goals[0].winningClaims[space.request.uri], undefined) assert.equal( index.ignored.some( (entry) => entry.uri === injected.uri && entry.reason === 'author is not an active space member', ), true, )
// And it changes nothing about who does the work: A claims and confirms exactly as it would have. await a.tick() await a.tick() const claimed = a.index() assert.equal(claimed.goals[0].winningClaims[space.request.uri].did, AGENT_A) assert.deepEqual( a.dispatchable(claimed).map((item) => item.request.uri), [space.request.uri], ) // Both operators ignore it identically — the same record, the same reason, on both folds. const other = await b.tick() assert.deepEqual( other.ignored.filter((entry) => entry.uri === injected.uri), index.ignored.filter((entry) => entry.uri === injected.uri), ) close()})
// --- F4: removed member mid-claim --------------------------------------------
it('drill 4: removing the claim holder frees the request at once, and re-adding restores their records', async () => { const { repos, clock, space, a, b, close } = await setup()
// A holds a live, confirmed claim, and has said something in the thread. await claimAndConfirm(a) const message = await a.client.create(COLLECTIONS.message, { $type: COLLECTIONS.message, goal: { uri: space.goal.uri, cid: space.goal.cid }, body: 'Working on it.', mentions: [], createdAt: '2026-01-01T00:05:00Z', }) await b.tick() const beforeRemoval = b.index() assert.equal(beforeRemoval.goals[0].winningClaims[space.request.uri].did, AGENT_A) assert.equal(beforeRemoval.goals[0].messages.some((entry) => entry.uri === message.uri), true) assert.deepEqual(b.dispatchable(beforeRemoval), [], 'B leaves a request A holds alone')
// The admin removes A. Removal revokes one exact grant and retains nothing. seed(repos.get(ROOT), COLLECTIONS.removeMember, 'remove-a', 'cid-remove-a', { $type: COLLECTIONS.removeMember, space: ref(space.space), did: AGENT_A, membership: ref(space.memberships.get(AGENT_A)), createdAt: '2026-01-01T00:06:00Z', })
// Ingest without pumping, so both folds are read at the same moment: this drill is about what the // MATERIALIZERS say the instant the grant is revoked, not about who claims next. const afterRemovalB = await b.syncOnly() const afterRemovalA = await a.syncOnly() for (const index of [afterRemovalB, afterRemovalA]) { assert.equal(index.goals[0].winningClaims[space.request.uri], undefined, 'the claim stops counting') assert.equal(index.goals[0].openRequests.length, 1, 'the request is back in the open set') assert.equal(index.goals[0].messages.some((entry) => entry.uri === message.uri), false) } // Immediately, on BOTH materializers, with no lease to wait out — a revoked grant is not a lease. assert.deepEqual(indexDigest(afterRemovalA), indexDigest(afterRemovalB))
// B picks the work up. await claimAndConfirm(b) assert.deepEqual([...b.claims.heldRequests()], [space.request.uri]) assert.deepEqual( b.dispatchable().map((item) => item.request.uri), [space.request.uri], )
// Time passes: B renews its lease, A's (still in its repo, still uncounted) lapses. clock.now = T0 + 300_000 await b.tick() assert.equal( Date.parse(b.claims.get(space.request.uri, AGENT_B).expiresAt), clock.now + TIMING.leaseMs, 'B renewed rather than letting its own lease run down', ) clock.now = T0 + TIMING.leaseMs + 1
// Re-adding A restores everything A ever wrote — the scenario's re-add rule, end to end. seed(repos.get(ROOT), COLLECTIONS.addMember, 'member-a-again', 'cid-member-a-again', { $type: COLLECTIONS.addMember, space: ref(space.space), did: AGENT_A, kind: 'agent', role: 'agent', createdAt: '2026-01-01T00:07:00Z', }) const readded = await b.syncOnly() assert.equal( readded.goals[0].messages.some((entry) => entry.uri === message.uri), true, 're-adding a member restores every record they wrote', ) // A's restored claim comes back with it — and is ignored on its own merits, because its lease // lapsed while A was out. Work that moved on stays moved on. assert.equal(readded.goals[0].winningClaims[space.request.uri].did, AGENT_B) const restoredA = await a.syncOnly() assert.deepEqual(indexDigest(restoredA), indexDigest(readded)) close()})
// --- F5: divergence detection -------------------------------------------------
it('drill 5: two folds of the same records digest identically; a withheld record names its section', async () => { const { space, a, b, close } = await setup() await claimAndConfirm(a) await b.tick()
const left = a.index() const right = b.index() assert.deepEqual(digestDifferences(indexDigest(left), indexDigest(right)), []) assert.equal(indexDigest(left).overall, indexDigest(right).overall)
// Feed a third store the same records in reverse: order cannot change a digest. const reversed = new MemoryRecordStore() for (const record of [...a.records.records()].reverse()) reversed.put(record) assert.equal( indexDigest(materialize(reversed, { spaceUri: SPACE_URI, asOf: new Date(T0).toISOString() })).overall, indexDigest(materialize(a.records, { spaceUri: SPACE_URI, asOf: new Date(T0).toISOString() })).overall, )
// Now withhold the claim from one of them: a real sync gap, of the kind an operator has to // diagnose. The digests differ, and the SECTION says it is about who holds what — not about // membership, and not about the artifacts. const gapped = new MemoryRecordStore() for (const record of a.records.records()) { if (record.collection === COLLECTIONS.claim) continue gapped.put(record) } const partial = indexDigest( materialize(gapped, { spaceUri: SPACE_URI, asOf: new Date(T0).toISOString() }), ) const complete = indexDigest(left) assert.notEqual(partial.overall, complete.overall) assert.deepEqual(digestDifferences(complete, partial), ['winningClaims']) assert.equal(partial.sections.members, complete.sections.members) assert.equal(partial.sections.openRequests, complete.sections.openRequests) assert.equal(space.request.uri.length > 0, true) close()})
// --- F6: the reader's clock runs fast ------------------------------------------
it('drill 6: an operator ten minutes fast still honours a live lease, and drops it on lag not skew', async () => { // The exposure this drill exists for: a lease used to be an INSTANT compared against the reader's // clock, so the lease B actually honoured was `leaseMs − (B's offset)`. At the 600s default, a B // ten minutes fast read a claim written this second as long expired, took the work, and both // operators did it. A lease is now a DURATION the writer declares and each observer measures from // its own first sighting, so B's offset cancels out entirely. const { clock, space, a, b, close } = await setup({ offsetB: 10 * 60_000 })
await claimAndConfirm(a) await b.tick() const seen = b.index() assert.equal(seen.goals[0].winningClaims[space.request.uri].did, AGENT_A) assert.deepEqual(b.dispatchable(seen), [], 'B leaves a request A holds a live claim on') assert.deepEqual([...b.claims.heldRequests()], [], 'and does not write a competing claim') // Both folds agree about who holds what, at ten minutes of offset — the exit criterion. assert.deepEqual(digestDifferences(indexDigest(a.index()), indexDigest(seen)), [])
// What is left is LAG, and it is bounded the way every other drill bounds it: one ingestion cycle // after the true expiry, B reads the request as free. Numerically, in cycles. clock.now = T0 + TIMING.leaseMs + 1 const syncsBefore = b.syncs const lapsed = await b.syncOnly() assert.equal(b.syncs - syncsBefore, 1) assert.equal( lapsed.goals[0].winningClaims[space.request.uri], undefined, 'a lease nobody renewed lapses on B within one cycle of its true expiry', ) await b.tick() await b.tick() await b.tick() assert.deepEqual([...b.claims.heldRequests()], [space.request.uri]) close()})
// --- F7: a forward clock jump mid-lease ---------------------------------------
it('drill 7: a seven-day forward jump mid-lease buys the claimant one lease, not seven days', async () => { const { clock, space, a, b, close } = await setup() await claimAndConfirm(a) await b.tick() assert.deepEqual(b.dispatchable(), [])
// A's clock steps a week forward between two renewals — one NTP correction, or a resumed VM. // Nothing can delete the record it renews, so what it writes there is permanent. clock.now = T0 + TIMING.renewIntervalMs + 1 a.skew(7 * 24 * 60 * 60_000) const claimUri = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}` const before = a.pds.records.get(claimUri).value await a.tick()
// It does not write next Thursday into it. On its own (wrong) clock the claim it holds is a week // old, which is past the protocol horizon — the point beyond which no version of THAT claim can // be in contract, however honest the duration it declares — so it retires the claim instead. const afterJump = a.pds.records.get(claimUri).value assert.deepEqual(afterJump, before, 'the record a jumped clock would have poisoned is untouched') assert.equal( a.logs.some((line) => /retiring our claim/.test(line)), true, 'loudly, because an operator has to be able to see a claim end this way', ) assert.deepEqual([...a.claims.heldRequests()], [])
// The request still needs doing, so A claims it again — in a NEW record, at the next generation, // with an honest `createdAt` of its own. It is dated a week out, which costs A the tie-break // against any competitor and buys it nothing: every observer measures the lease from arrival. await a.tick() const nextUri = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}` const reclaimed = a.pds.records.get(nextUri).value assert.equal(Date.parse(reclaimed.expiresAt) - Date.parse(reclaimed.renewedAt), TIMING.leaseMs) assert.equal(Date.parse(reclaimed.createdAt) > clock.now + 6 * 24 * 60 * 60_000, true)
// So B honours it for one lease from the moment B saw it, and no longer. (Both of A's records are // A's, so while the first is still live B reads the earlier `createdAt` as the winner — which is // the tie-break doing exactly what it should.) await b.tick() assert.equal(b.index().goals[0].winningClaims[space.request.uri].did, AGENT_A) clock.now = T0 + TIMING.leaseMs + 1 assert.equal((await b.syncOnly()).goals[0].winningClaims[space.request.uri].uri, nextUri) clock.now += TIMING.leaseMs + 1 assert.equal( (await b.syncOnly()).goals[0].winningClaims[space.request.uri], undefined, 'a claim written on a week-fast clock lapses one lease after it arrived, not one week', ) await b.tick() await b.tick() await b.tick() assert.deepEqual([...b.claims.heldRequests()], [space.request.uri]) close()})
it('drill 7d: the jumped clock is corrected, and the claim it wrote is retired rather than held forever', async () => { // Walk drill 7 one step further, to where a clock jump normally ends: NTP puts it back. What A is // left holding is a record every field of which was stamped by the fast clock — a legal lease // inside a legal horizon, so it is IN CONTRACT and therefore invisible to every recovery path // there was. Not lapsed (it expires next week), not due for renewal, not refused by the fold (so // no repair, and the store would not adopt a shortening rewrite of a conforming version anyway), // and no other winner for the loss branch to name. The row would sit `held` forever, gating // dispatch behind a claim its own writer can neither renew nor retract — the one recovery this // whole change would otherwise have REMOVED rather than added. const { clock, space, a, b, close } = await setup() a.skew(7 * 24 * 60 * 60_000) await claimAndConfirm(a) const first = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}` const written = a.pds.records.get(first).value assert.equal(Date.parse(written.createdAt), clock.now + 7 * 24 * 60 * 60_000) assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])
// The correction lands. A's very next pump ends that claim and starts another beside it, dated by // the clock it has now — in the same pump, so nothing is left uncovered for a tick. a.skew(0) await a.tick() assert.equal( a.logs.some((line) => /retiring our claim/.test(line) && /into this daemon's own future/.test(line)), true, 'and says which clock it distrusts, since an operator cannot see this any other way', ) const next = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}` const honest = a.pds.records.get(next).value assert.equal(Date.parse(honest.createdAt), clock.now) assert.equal(Date.parse(honest.expiresAt) - Date.parse(honest.renewedAt), TIMING.leaseMs) assert.deepEqual(a.pds.records.get(first).value, written, 'the record from the future is left alone')
// And the work goes on: the new claim confirms, A dispatches again, and B — which honours the old // record for one lease from arrival and the new one on its earlier `createdAt` — agrees A holds it. await a.tick() await a.tick() assert.deepEqual([...a.claims.heldRequests()], [space.request.uri]) assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 1) assert.deepEqual( a.dispatchable().map((item) => item.request.uri), [space.request.uri], ) await b.tick() const winner = b.index().goals[0].winningClaims[space.request.uri] assert.equal(winner.uri, next) assert.equal(winner.did, AGENT_A) assert.deepEqual([...b.claims.heldRequests()], [], 'B never contests a request A is covering') close()})
it('drill 7f: a correction the next renewal still reaches costs no claim, and no peer sees anything', async () => { // The other side of drill 7d's boundary, and the reason the boundary is where it is. A jump of Δ // that the RENEWAL branch caught repairs itself: the renewal trigger is // `remaining <= leaseMs − renewIntervalMs`, so it falls due Δ + one interval after the correction, // while observers stop honouring the version one declared lease after it arrived. For Δ up to // `leaseMs − renewIntervalMs` the repair lands first, so retiring there would buy nothing a peer // could ever have seen — and cost a generation plus, since the retired record keeps the earlier // `createdAt` and wins A's own tie-break until it lapses, a stretch of A's own dispatch. const { clock, space, a, b, close } = await setup() await claimAndConfirm(a) await b.tick() const uri = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`
// Two minutes fast when the renewal falls due — a resumed VM, a chronyd step. const jump = 2 * 60_000 assert.ok(jump <= TIMING.leaseMs - TIMING.renewIntervalMs, 'inside what a renewal still reaches') clock.now = T0 + TIMING.renewIntervalMs a.skew(jump) await a.tick() const jumped = a.pds.records.get(uri) assert.equal(Date.parse(jumped.value.renewedAt), clock.now + jump, 'the renewal came off the fast clock')
// NTP puts the clock back. What A holds is in contract by both bounds, sits ahead of the clock A // now has, and no other branch would ever move it — but the renewal that will move it is due // before any observer stops honouring it, so A leaves it exactly where it is. a.skew(0) await a.tick() assert.equal(a.logs.some((line) => /retiring our claim/.test(line)), false, 'nothing is retired') assert.equal(a.pds.records.get(uri).cid, jumped.cid, 'and nothing is rewritten yet') assert.deepEqual([...a.claims.heldRequests()], [space.request.uri], 'A still holds the work') assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 0, 'no generation is burned') assert.deepEqual( a.dispatchable().map((item) => item.request.uri), [space.request.uri], 'and it stays dispatchable throughout, which is what retiring here would have cost', )
// The repair is the ordinary renewal, dated by the clock A actually has. clock.now = Date.parse(jumped.value.expiresAt) - (TIMING.leaseMs - TIMING.renewIntervalMs) await a.tick() const repaired = a.pds.records.get(uri).value assert.equal(Date.parse(repaired.renewedAt), clock.now) assert.equal(Date.parse(repaired.expiresAt) - Date.parse(repaired.renewedAt), TIMING.leaseMs)
// And B read A as the holder at every step, so there was never a divergence to close. await b.tick() assert.equal(b.index().goals[0].winningClaims[space.request.uri].uri, uri) assert.equal(b.claims.all().length, 0, 'B never contests a request A has held all along') close()})
it('drill 7c: a claim that reaches its horizon is retired and re-claimed, with no gap in the work', async () => { // What the horizon costs a healthy claim, and how it is paid. A request can stay open far longer // than a day — one waiting on a human answer routinely does — and `createdAt` is frozen within a // claim, so at some point the claim's own record can no longer carry a conforming lease. It ends // there, and the request is claimed again beside it rather than held by a version every observer // would refuse. const { clock, space, a, b, close } = await setup() await claimAndConfirm(a) await b.tick() const first = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`
// Renewals keep it alive up to the horizon, and the LAST one stops exactly there rather than // reaching past it — a lease clamped to where every observer would have stopped honouring it. const horizon = T0 + 24 * 60 * 60 * 1000 clock.now = horizon - 100_000 await a.tick() const last = a.pds.records.get(first) assert.equal(Date.parse(last.value.expiresAt), horizon, 'the final lease expires at the horizon exactly') assert.equal(Date.parse(last.value.expiresAt) - Date.parse(last.value.createdAt), 24 * 60 * 60 * 1000)
// And it is written once, not on every tick of the interval before the horizon: a renewal that // cannot move `expiresAt` is a write for nothing. await a.tick() await a.tick() assert.equal(a.pds.records.get(first).cid, last.cid) await b.tick() assert.equal(b.index().goals[0].winningClaims[space.request.uri].uri, first)
// Past the horizon the claim is over. Retired once — the row is terminal, so no tick after this // one repeats it — and never rewritten with a lease the fold would refuse. clock.now = horizon + 1 await a.tick() assert.deepEqual(a.pds.records.get(first).value, last.value) assert.equal( a.logs.filter((line) => /retiring our claim/.test(line)).length, 1, 'retired once, not once per tick', ) assert.deepEqual([...a.claims.heldRequests()], [])
// And the work continues under generation 1: a fresh record, a fresh createdAt, confirmed the // ordinary way, and B agrees. await a.tick() await a.tick() const next = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}` assert.deepEqual([...a.claims.heldRequests()], [space.request.uri]) assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 1) assert.deepEqual( a.dispatchable().map((item) => item.request.uri), [space.request.uri], ) await b.tick() const winner = b.index().goals[0].winningClaims[space.request.uri] assert.equal(winner.uri, next) assert.equal(winner.did, AGENT_A) close()})
it('drill 7e: a claim retired while its own turn is running is replaced in the same pump, at concurrency 1', async () => { // The routine way drill 7c's retirement arrives: a request answered by a human in the last hour // of its claim's day dispatches a turn that is still running when the horizon passes. Retirement // is terminal, so nothing renews the old record back — if the daemon could not re-claim until the // turn ended, the request would read as free to every peer for the rest of a turn (up to // `run.timeoutMs`, 60 minutes by default) rather than for a pump. // // The turn-eligibility exemption alone does not get there. At the default `concurrency: 1` the // running turn IS the daemon's one turn slot, so free capacity is zero for exactly as long as the // covering claim is needed: charged against it, the re-claim is refused before the exemption is // ever consulted. const { clock, space, a, b, close } = await setup() await claimAndConfirm(a) await b.tick() a.turns.markRunning(space.request.uri, space.request.cid, { containerLabel: 'radial-turn-1', checkoutPath: '/tmp/radial-turn-1', }) assert.equal(a.turns.running().length, CONCURRENCY, 'no free turn capacity while it runs')
clock.now = T0 + 24 * 60 * 60 * 1000 + 1 await a.tick() assert.equal(a.logs.some((line) => /retiring our claim/.test(line)), true) assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 1, 'covered again in the same pump') assert.equal(a.turns.get(space.request.uri).state, 'running', 'and the turn is never killed for it') const next = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}` assert.equal(Date.parse(a.pds.records.get(next).value.createdAt), clock.now)
// B's very next tick — the first moment it could have taken the work — finds it covered: it // ingests the replacement before it decides anything, and never writes a claim of its own. await b.tick() assert.equal(b.index().goals[0].winningClaims[space.request.uri].uri, next) assert.deepEqual([...b.claims.heldRequests()], []) assert.equal(b.claims.all().length, 0, 'B never contests a request A is still working') close()})
it('drill 7b: a claim record already poisoned by a clock jump is repaired by its own writer', async () => { // The wedge the clamp would otherwise create. The rkey is a pure function of the request and // nothing deletes records, so a claim written with an absurd `expiresAt` — by a pre-`renewedAt` // daemon whose clock jumped — is permanent. If the fold merely ignored it, A's next attempt would // collide at that rkey, adopt the record, sit in `confirming` (which has no timeout) forever, and // burn an outstanding slot: at the default of one, A would claim nothing in this space ever again. const { space, a, b, close } = await setup() const rkey = claimRkey(space.request.uri, space.request.cid) seed(a.pds, COLLECTIONS.claim, rkey, 'cid-claim-poisoned', { $type: COLLECTIONS.claim, request: ref(space.request), expiresAt: '2026-02-01T00:00:00.000Z', createdAt: '2026-01-01T00:00:00.000Z', })
const before = await a.syncOnly() assert.equal(before.goals[0].winningClaims[space.request.uri], undefined, 'no observer honours it') assert.equal( before.ignored.some( (entry) => entry.uri === `at://${AGENT_A}/${COLLECTIONS.claim}/${rkey}` && /outside the protocol horizon/.test(entry.reason), ), true, 'and the fold says why, rather than leaving the request open with no explanation', )
// A claims, collides with its own poisoned record, and rewrites it with a conforming lease. await a.tick() assert.equal( a.logs.some((line) => /out of contract/.test(line)), true, 'loudly — an operator has to be able to see this happened', ) const repaired = a.pds.records.get(`at://${AGENT_A}/${COLLECTIONS.claim}/${rkey}`).value assert.equal(Date.parse(repaired.expiresAt) - Date.parse(repaired.renewedAt), TIMING.leaseMs) assert.equal(repaired.createdAt, '2026-01-01T00:00:00.000Z', 'the tie-break position is untouched')
// And the row is not wedged: it confirms, it is dispatchable, and B agrees A holds it. await a.tick() await a.tick() assert.deepEqual([...a.claims.heldRequests()], [space.request.uri]) assert.deepEqual( a.dispatchable().map((item) => item.request.uri), [space.request.uri], ) await b.tick() assert.equal(b.index().goals[0].winningClaims[space.request.uri].did, AGENT_A) close()})
// --- F8: the tie-break is self-reported, and stays that way --------------------
it('drill 8: a daemon whose clock runs an hour slow wins every race — the documented, unfixed gap', async () => { // Deliberately asserting TODAY's behaviour, permanently. The tie-break is earliest `createdAt`, // self-reported, with nothing corroborating it: `rev` is observer-local by construction and the // claim rkey is a content hash with no time in it, so "lowest URI" reduces to lowest DID — a fixed // bias, not a random one. Hardening it means reintroducing a corroborating time source, which // changes a wire-contract rule the digest names; see the plan's §7 and operators.md §8. // // Note what first-seen liveness does NOT do here: it makes the backward-skewed claim MORE live, // not less, because its declared lease is measured from arrival rather than from an `expiresAt` // that is already in the reader's past. Fairness is a separate problem from safety. for (const attempt of [1, 2, 3]) { const { space, a, b, close } = await setup({ offsetB: -60 * 60_000 }) // A genuine race: each claims in the same true instant without having seen the other, which is // the only way to test the tie-break rather than the do-not-contest rule. const claimOf = (did) => `at://${did}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}` a.hidden.add(claimOf(AGENT_B)) b.hidden.add(claimOf(AGENT_A)) await a.tick() await b.tick() a.hidden.clear() b.hidden.clear() const winners = [await a.syncOnly(), await b.syncOnly()].map( (index) => index.goals[0].winningClaims[space.request.uri]?.did, ) assert.deepEqual(winners, [AGENT_B, AGENT_B], `attempt ${attempt}: the slow clock wins, on both folds`) // B's claim is an hour old by its own account and still counted, which is the point: nothing // corroborates a self-reported `createdAt`, so the bias is permanent rather than accidental. const claim = (await a.syncOnly()).goals[0].winningClaims[space.request.uri] assert.equal(Date.parse(claim.value.createdAt), T0 - 60 * 60_000) close() }})