Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281import assert from 'node:assert/strict'import { it } from 'node:test'import { SkewWatch, futureDatedAuthors, runDaemon } from '../dist/index.js'
const space = (marker = 'index') => ({ spaceUri: 'at://space', runtime: { sync: async () => ({ marker }) },})
it('pumps the claim manager BEFORE the dispatcher, so a claim confirmed this cycle dispatches now', async () => { const order = [] await runDaemon({ spaces: [space()], claimManager: { pump: () => order.push('claims') }, dispatcher: { pump: () => order.push('dispatch') }, checkDispatcher: { pump: () => order.push('checks') }, actors: { all: [] }, intervalMs: 1, sleep: async () => {}, shouldStop: () => order.length > 0, log: () => {}, }) assert.deepEqual(order, ['claims', 'dispatch', 'checks'])})
it('runs without a claim manager, which is what a daemon serving only assigned requests is', async () => { let pumped = 0 await runDaemon({ spaces: [space()], dispatcher: { pump: () => (pumped += 1) }, actors: { all: [] }, intervalMs: 1, sleep: async () => {}, shouldStop: () => pumped >= 2, log: () => {}, }) assert.equal(pumped, 2)})
it('reads a dynamic interval afresh each tick, so a healthy stream slows the poll down', async () => { const slept = [] let streaming = false await runDaemon({ spaces: [space()], dispatcher: { pump: () => (streaming = true) }, actors: { all: [] }, intervalMs: () => (streaming ? 60_000 : 5_000), sleep: async (milliseconds) => { slept.push(milliseconds) }, shouldStop: () => slept.length >= 2, log: () => {}, }) // First sleep is decided AFTER the first pump, so the stream is already up by then. assert.deepEqual(slept, [60_000, 60_000])})
it('wakes the loop mid-sleep without stopping it, and a fresh signal is handed out each time', async () => { const aborts = [] let wake await runDaemon({ spaces: [space()], dispatcher: { pump: () => {} }, actors: { all: [] }, intervalMs: 30_000, onWake: (fn) => { wake = fn }, sleep: async (_milliseconds, signal) => { // A stream event arrives mid-sleep: the loop must come round now, not in 30 seconds. wake() aborts.push(signal.aborted) }, shouldStop: () => aborts.length >= 2, log: () => {}, }) assert.deepEqual(aborts, [true, true])})
it('does not lose a wake after its space synced but before the tick sleeps', async () => { let wake let relevantSyncs = 0 let laterSyncs = 0 const sleeps = [] await runDaemon({ spaces: [ { spaceUri: 'at://relevant', runtime: { sync: async () => ({ tick: ++relevantSyncs }) }, }, { spaceUri: 'at://later', runtime: { sync: async () => { laterSyncs += 1 if (laterSyncs === 1) wake() return { tick: laterSyncs } }, }, }, ], dispatcher: { pump: () => {} }, actors: { all: [] }, intervalMs: 60_000, onWake: (fn) => { wake = fn }, sleep: async (milliseconds) => { sleeps.push(milliseconds) }, shouldStop: () => relevantSyncs >= 2, log: () => {}, }) assert.equal(relevantSyncs, 2) assert.deepEqual(sleeps, [])})
it('keeps waking separate from stopping', async () => { // The wake signal must not be the shutdown signal: waking the loop is not asking it to end. const controller = new AbortController() let wake let pumps = 0 await runDaemon({ spaces: [space()], dispatcher: { pump: () => { pumps += 1 if (pumps === 3) controller.abort() }, }, actors: { all: [] }, intervalMs: 1, signal: controller.signal, onWake: (fn) => { wake = fn }, sleep: async () => wake(), log: () => {}, }) assert.equal(pumps, 3)})
// --- clock skew, made visible (design §11, docs/operators.md §3) ---------------
const skewIndex = (records) => ({ goals: [ { requests: records.filter((record) => record.kind === 'request'), artifacts: [], messages: [], reviews: [], winningClaims: Object.fromEntries( records.filter((record) => record.kind === 'claim').map((claim, at) => [`r${at}`, claim]), ), }, ], projects: [],})
const authored = (did, createdAt, kind = 'request') => ({ did, kind, value: { createdAt } })
it('names every author that has dated a record in this daemon future, worst first', () => { const index = skewIndex([ authored('did:plc:ahead', '2026-01-01T00:10:00Z'), authored('did:plc:ahead', '2026-01-01T00:02:00Z'), authored('did:plc:further', '2026-01-01T01:00:00Z', 'claim'), authored('did:plc:behind', '2025-12-31T00:00:00Z'), authored('did:plc:unreadable', 'whenever'), ]) assert.deepEqual(futureDatedAuthors(index, '2026-01-01T00:00:00Z'), [ { did: 'did:plc:further', aheadMs: 3_600_000 }, { did: 'did:plc:ahead', aheadMs: 600_000 }, ]) // One-sided by construction: an author BEHIND us is indistinguishable from a record we ingested // late, which is exactly why the `Date`-header probe exists alongside this. assert.deepEqual(futureDatedAuthors(skewIndex([authored('did:plc:behind', '2025-01-01T00:00:00Z')]), '2026-01-01T00:00:00Z'), [])})
it('sees a claim RENEWED by a jumped clock, whose createdAt is perfectly honest', () => { // The record shape this whole rule exists to survive, and the one a `createdAt`-only heuristic is // blind to: renewal freezes `createdAt` by design, so a daemon that claimed on a good clock and // renewed on a jumped one dates the only field this used to read entirely truthfully. const renewed = { did: 'did:plc:jumped', kind: 'claim', value: { createdAt: '2025-12-31T23:00:00Z', renewedAt: '2026-01-01T00:20:00Z', expiresAt: '2026-01-01T00:30:00Z', }, } assert.deepEqual(futureDatedAuthors(skewIndex([renewed]), '2026-01-01T00:00:00Z'), [ { did: 'did:plc:jumped', aheadMs: 1_200_000 }, ])})
it('reports skew once per threshold crossing, not once per tick', () => { const logs = [] const samples = [] let index = skewIndex([authored('did:plc:ahead', '2026-01-01T00:00:20Z')]) const watch = new SkewWatch({ thresholdMs: 30_000, samples: () => samples, now: () => '2026-01-01T00:00:00Z', log: (message) => logs.push(message), })
// Twenty seconds ahead is inside the threshold: nothing to say. watch.observe(index) assert.deepEqual(logs, [])
index = skewIndex([authored('did:plc:ahead', '2026-01-01T00:01:00Z')]) watch.observe(index) watch.observe(index) watch.observe(index) assert.equal(logs.length, 1, 'a warning repeated every tick is a warning nobody reads') assert.match(logs[0], /did:plc:ahead has written a record dated 60000ms in this daemon's future/)
// A clock that steps again crosses into a worse bucket, and that IS worth saying. watch.observe(skewIndex([authored('did:plc:ahead', '2026-01-01T00:10:00Z')])) assert.equal(logs.length, 2)
// The probe half reports separately, and at most once. samples.push({ origin: 'https://pds.test', skewMs: -120_000, at: '2026-01-01T00:00:00Z' }) watch.observe(index) watch.observe(index) assert.equal(logs.length, 3) assert.match(logs[2], /this machine is 120000ms ahead of https:\/\/pds\.test/)})
it('buckets the probe by the measured skew, not by the clock the tick happened on', () => { // The `Date`-header half has to dedup on the same terms as the record half. Bucketing it by "what // time is it now" puts every tick in a bucket of its own, and `radiald run` then prints the same // three-sentence warning every five seconds for as long as the skew lasts. So this drives it with // an ADVANCING clock, which is the only way that bug is visible. const logs = [] const samples = [{ origin: 'https://pds.test', skewMs: -120_000, at: '2026-01-01T00:00:00Z' }] let now = Date.parse('2026-01-01T00:00:00Z') const watch = new SkewWatch({ thresholdMs: 30_000, samples: () => samples, now: () => new Date((now += 5_000)).toISOString(), log: (message) => logs.push(message), }) const empty = skewIndex([]) for (let tick = 0; tick < 10; tick += 1) watch.observe(empty) assert.equal(logs.length, 1, 'one warning for one skew, however many ticks it persists for')
// A clock that steps further out crosses into a worse bucket, which IS news… samples[0] = { ...samples[0], skewMs: -300_000 } for (let tick = 0; tick < 3; tick += 1) watch.observe(empty) assert.equal(logs.length, 2) assert.match(logs[1], /this machine is 300000ms ahead of https:\/\/pds\.test/)
// …and a second service drifting is its own subject, not a repeat of the first. samples.push({ origin: 'https://other.test', skewMs: 90_000, at: '2026-01-01T00:00:00Z' }) watch.observe(empty) watch.observe(empty) assert.equal(logs.length, 2, 'the worst sample is still the one already reported') samples[1] = { ...samples[1], skewMs: 600_000 } watch.observe(empty) assert.equal(logs.length, 3) assert.match(logs[2], /https:\/\/other\.test/)})
it('observes skew on the same fresh index the dispatchers get, before anything acts on it', async () => { const order = [] await runDaemon({ spaces: [space()], skew: { observe: () => order.push('skew') }, claimManager: { pump: () => order.push('claims') }, dispatcher: { pump: () => order.push('dispatch') }, actors: { all: [] }, intervalMs: 1, sleep: async () => {}, shouldStop: () => order.length > 0, log: () => {}, }) assert.deepEqual(order, ['skew', 'claims', 'dispatch'])})