Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446import { FetchRepoTransport, clockSkewWarning, worstClockSample, type ClockSample, type RepoReadTransport,} from '@radial/atproto'import { materialize, type MaterializedIndex } from '@radial/core'import { SqliteRecordStore } from '@radial/core/node'import { JetstreamSubscription, RepoPoller, SpaceIngestor, type JetstreamOptions } from '@radial/ingest'import { SqliteSyncStateStore } from '@radial/ingest/node'import type { ActorRegistry } from './actors.js'import type { CheckDispatcher } from './check-dispatch.js'import type { TurnDispatcher } from './dispatch.js'
export interface DaemonIndexPaths { records: string checkpoints: string}
/** * What the run loop needs of a space, whichever bus it runs on. * * `SpaceSyncRuntime` below is the public-path implementation (repos and Jetstream); * `PrivateSpaceRuntime` (`private-space.ts`) is the private one (envelopes off a bus, the device * directory off members' public repos). `runDaemon` holds this interface and nothing more, which is * the seam design §18 asks for: dispatch, claims, checks and auto-review consume the index and the * writer, and never the transport. */export interface SpaceRuntime { sync(): Promise<MaterializedIndex> index(): MaterializedIndex /** True while the live path is healthy — a stream on the public path, the bus on the private one. */ readonly streaming: boolean close(): void}
/** What `SpaceSyncRuntime` needs to run a Jetstream subscription for one space. */export interface SpaceJetstreamOptions { endpoint: string /** Reported for every stream record, so the daemon can interrupt its sleep and dispatch now * instead of at the next poll. */ onRecord?: () => void onStatus?: (status: 'connected' | 'reconnecting' | 'degraded', detail?: string) => void degradedAfterMs?: number log?: (message: string) => void /** Test seams, passed straight through to `JetstreamSubscription`. */ socket?: JetstreamOptions['socket'] timers?: JetstreamOptions['timers'] now?: JetstreamOptions['now']}
/** * A long-lived alternative to `syncSpaceOnce`/`readIndex` for `radiald run`: opens its SQLite * record + checkpoint stores once and keeps them open across many polls, instead of the * one-shot helpers below (which are deliberately left untouched — the `sync`/`index` CLI and * their tests depend on the reopen-every-call behavior). `index()` materializes from the * already-open store without polling, for the dispatcher to read the freshest index synchronously * after each `sync()`. */export class SpaceSyncRuntime implements SpaceRuntime { readonly #spaceUri: string readonly #records: SqliteRecordStore readonly #checkpoints: SqliteSyncStateStore readonly #ingestor: SpaceIngestor #jetstream: JetstreamSubscription | undefined
constructor(spaceUri: string, paths: DaemonIndexPaths, transport: RepoReadTransport = new FetchRepoTransport()) { this.#spaceUri = spaceUri this.#records = new SqliteRecordStore(paths.records) this.#checkpoints = new SqliteSyncStateStore(paths.checkpoints) const poller = new RepoPoller(transport, this.#records, this.#checkpoints) this.#ingestor = new SpaceIngestor(spaceUri, poller, this.#records) }
async sync(): Promise<MaterializedIndex> { const index = await this.#ingestor.sync() // Membership may have changed under us; `wantedDids` is fixed per connection, so the stream has // to be rebuilt to see a new member. Polling covered them in the meantime. this.#jetstream?.refresh() return index }
index(): MaterializedIndex { return materialize(this.#records, { spaceUri: this.#spaceUri }) }
/** * Start streaming this space's records (design §5). Records land in the SAME store the poller * writes to, so `collapse` reconciles the two provenances (the stream's real commit rev beats the * poller's head-rev stamp, and the minimum wins) with no special case anywhere. * * The subscription is owned here rather than in `cli.ts` so both stores and the socket close * together — a space's ingestion is one thing, however many ways records reach it. */ startJetstream(options: SpaceJetstreamOptions): JetstreamSubscription { const cursorKey = `${options.endpoint} ${this.#spaceUri}` const existing = this.#checkpoints.getCursor(cursorKey) const subscription = new JetstreamSubscription({ endpoint: options.endpoint, // The space DID plus every active member's, as the poller has discovered them. On the very // first run this is just the space DID; the poller widens it and `refresh` picks that up. dids: () => [...this.#ingestor.knownDids], ...(existing !== undefined ? { cursor: existing } : {}), onRecord: (record) => { this.#records.put(record) options.onRecord?.() }, onCursor: (cursor) => this.#checkpoints.setCursor(cursorKey, cursor), ...(options.onStatus ? { onStatus: options.onStatus } : {}), ...(options.degradedAfterMs !== undefined ? { degradedAfterMs: options.degradedAfterMs } : {}), ...(options.log ? { log: options.log } : {}), ...(options.socket ? { socket: options.socket } : {}), ...(options.timers ? { timers: options.timers } : {}), ...(options.now ? { now: options.now } : {}), }) this.#jetstream = subscription subscription.start() return subscription }
/** True while the stream is healthy — what the daemon reads to slow polling to the backfill rate. */ get streaming(): boolean { return this.#jetstream?.connected ?? false }
close(): void { this.#jetstream?.stop() this.#checkpoints.close() this.#records.close() }}
export async function syncSpaceOnce( spaceUri: string, paths: DaemonIndexPaths, transport: RepoReadTransport = new FetchRepoTransport(),): Promise<MaterializedIndex> { const records = new SqliteRecordStore(paths.records) const checkpoints = new SqliteSyncStateStore(paths.checkpoints) try { const poller = new RepoPoller(transport, records, checkpoints) return await new SpaceIngestor(spaceUri, poller, records).sync() } finally { checkpoints.close() records.close() }}
export function readIndex(spaceUri: string, recordsPath: string): MaterializedIndex { const records = new SqliteRecordStore(recordsPath) try { return materialize(records, { spaceUri }) } finally { records.close() }}
/** * The furthest into OUR future any author has dated a record, per author DID. * * The cheap half of skew detection, and deliberately one-sided: a timestamp we have not reached yet * proves that author's clock is ahead of ours, while an author whose clock runs BEHIND is * indistinguishable from a record we ingested late. (That second case is the one that matters for * fairness — a slow clock wins the claim tie-break — which is why the `Date`-header probe exists * alongside this, and why the tie-break exposure is documented rather than closed.) * * A claim is judged on `renewedAt` as well as `createdAt`, and that is the case worth having: a * renewal freezes `createdAt` by design, so a claim renewed by a jumped clock — the record shape * this whole rule exists to survive — is dated honestly in the one field a `createdAt`-only * heuristic looks at. `renewedAt` is where the jump actually shows. * * Reads the index rather than the store: only records the space actually trusts are worth * reporting, and this is the same set the operator sees everywhere else. */export function futureDatedAuthors( index: MaterializedIndex, nowIso: string,): Array<{ did: string; aheadMs: number }> { const now = Date.parse(nowIso) if (Number.isNaN(now)) return [] const ahead = new Map<string, number>() const report = (did: string, stamp: string | undefined): void => { const at = Date.parse(stamp ?? '') if (Number.isNaN(at) || at <= now) return ahead.set(did, Math.max(ahead.get(did) ?? 0, at - now)) } const consider = (record: { did: string; value: { createdAt?: string } }): void => report(record.did, record.value.createdAt) for (const view of [...index.goals, ...index.projects]) { for (const record of view.requests) consider(record) for (const record of view.artifacts) consider(record) for (const record of view.messages) consider(record) for (const record of view.reviews) consider(record) for (const claim of Object.values(view.winningClaims)) { consider(claim) report(claim.did, claim.value.renewedAt) } } return [...ahead] .map(([did, aheadMs]) => ({ did, aheadMs })) .sort((left, right) => right.aheadMs - left.aheadMs || (left.did < right.did ? -1 : 1))}
export interface SkewWatchOptions { /** How far off is worth saying so — `clockSkewThresholdMs(leaseMs)` in the daemon. */ thresholdMs: number /** The `Date`-header samples this daemon has collected, if a probe is wired. */ samples?: () => ClockSample[] now?: () => string log: (message: string) => void}
/** * Turns both skew signals into log lines an operator will actually read. * * "Once per threshold crossing", not once per tick: a warning repeated every five seconds is a * warning nobody reads, and a warning printed once at startup is one nobody sees when the clock * steps an hour later. So each subject is reported when it moves into a worse BUCKET (a multiple of * the threshold) and stays quiet until it does so again. */export class SkewWatch { readonly #options: SkewWatchOptions readonly #reported = new Map<string, number>()
constructor(options: SkewWatchOptions) { this.#options = options }
observe(index: MaterializedIndex): void { const now = this.#options.now?.() ?? new Date().toISOString() const threshold = this.#options.thresholdMs const samples = this.#options.samples?.() ?? [] const warning = clockSkewWarning(samples, threshold) const worst = worstClockSample(samples) // Bucketed by the MEASURED skew, exactly like the record heuristic below — never by the clock, // which would put every tick in a bucket of its own and print this every five seconds forever. // Per origin, so a second service drifting is its own news, and re-reported when a clock steps // another threshold out. if (warning && worst) this.#report(`probe ${worst.origin}`, Math.abs(worst.skewMs), warning, threshold) for (const { did, aheadMs } of futureDatedAuthors(index, now)) { if (aheadMs <= threshold) continue this.#report( did, aheadMs, `clock skew: ${did} has written a record dated ${aheadMs}ms in this daemon's future ` + `(threshold ${threshold}ms). Their clock runs ahead of ours, or ours runs behind.`, threshold, ) } }
#report(subject: string, magnitude: number, message: string, bucketMs: number): void { const bucket = Math.floor(magnitude / Math.max(1, bucketMs)) if ((this.#reported.get(subject) ?? -1) >= bucket) return this.#reported.set(subject, bucket) this.#options.log(message) }}
export function indexSummary(index: MaterializedIndex): object { return { space: { uri: index.space.uri, cid: index.space.cid, name: index.space.value.name }, members: index.members, artifactTypes: index.artifactTypes.map((entry) => entry.value.name), projects: index.projects.map((view) => ({ uri: view.target.uri, name: view.target.value.name, requests: view.requests.length, openRequests: view.openRequests.length, artifacts: view.artifacts.length, })), goals: index.goals.map((view) => ({ uri: view.target.uri, title: view.target.value.title, requests: view.requests.length, openRequests: view.openRequests.length, artifacts: view.artifacts.length, reviews: view.reviews.length, })), ignored: index.ignored, edits: index.edits, }}
function defaultSleep(ms: number, signal?: AbortSignal): Promise<void> { return new Promise((resolve) => { const timer = setTimeout(resolve, ms) if (!signal) return if (signal.aborted) { clearTimeout(timer) resolve() return } signal.addEventListener( 'abort', () => { clearTimeout(timer) resolve() }, { once: true }, ) })}
/** One space in the loop, and the identities that write into it. */export interface DaemonSpace { spaceUri: string runtime: SpaceRuntime /** * The registry every pump uses for THIS space, when it differs from the daemon's. * * A private space is the only caller: its actors write envelopes rather than PDS records * (`privateActorRegistry`), and a daemon serving a public and a private space at once holds both * registries at once. Absent means the daemon's own — which is every public space, unchanged. */ actors?: ActorRegistry}
export interface RunDaemonOptions { spaces: DaemonSpace[] dispatcher: TurnDispatcher /** Optional check-runner dispatcher, pumped alongside the turn dispatcher each tick. Fire-and- * forget and on its own concurrency budget, so it never blocks (or is blocked by) turn dispatch. */ checkDispatcher?: CheckDispatcher /** Optional merge-observation poller (design §10); pumped with the same fresh index right after * the dispatcher, fire-and-forget. Omitted when no forge adapter is wired (e.g. no forge config). */ mergePoller?: { pump(index: MaterializedIndex, actors: ActorRegistry): void } /** Optional auto-review trigger (design §7); pumped with the same fresh index right after the * merge poller, fire-and-forget. Runs whenever agents exist (no forge dependency). */ autoReviewTrigger?: { pump(index: MaterializedIndex, actors: ActorRegistry): void } /** Optional claim manager (design §11); pumped BEFORE the dispatcher, so a claim this cycle's * index confirms is dispatched by this cycle's `dispatcher.pump` rather than the next one. * Omitted when nothing serves open requests — the whole path is inert for assigned work. */ claimManager?: { pump(index: MaterializedIndex, actors: ActorRegistry): void } /** Optional clock-skew reporting, given the same fresh index. Pure observation: it logs and does * nothing else, because no rule in the fold may depend on a clock comparison. */ skew?: { observe(index: MaterializedIndex): void } actors: ActorRegistry /** * Sleep between ticks. A function is read afresh each time round, which is how Jetstream turns * the poll into a slow backfill while the stream is healthy and back into the primary path the * moment it is not (design §5). */ intervalMs: number | (() => number) /** * Hands the caller a function that interrupts the current sleep. It is what lets a Jetstream * event dispatch immediately instead of waiting out the interval; without it the stream would * shorten ingestion latency and change nothing an operator could see. */ onWake?: (wake: () => void) => void /** * Called once per space, with that space's first successful index. Startup advice about a space * can only be given once the space has been read — the registry lives in it, not in the config — * and once is the whole point: a warning repeated every tick is a warning nobody reads. */ onFirstIndex?: (spaceUri: string, index: MaterializedIndex, actors: ActorRegistry) => void log?: (message: string) => void sleep?: (milliseconds: number, signal?: AbortSignal) => Promise<void> /** Aborting interrupts an in-progress sleep immediately, for responsive SIGTERM/SIGINT shutdown. */ signal?: AbortSignal /** Checked after each tick (and once more after sleeping); returning true stops the loop. Also * how a single-tick test drives exactly one pass without real signals or timers. */ shouldStop?: () => boolean}
/** * The `radiald run` daemon's run-loop body, factored out so it is directly testable: a test can * inject a fake `sleep`/`shouldStop` (or a pre-aborted `signal`) to drive exactly one tick, with a * `FakeContainerRunner`-backed dispatcher and an injected sync, without real timers or OS signals. * Each tick syncs every configured space and hands its freshest index to * `dispatcher.pump` — which launches turns in the background and returns immediately — so a slow * or stuck turn never blocks ingestion of the next tick. A per-space sync error is logged and * skipped rather than killing the loop. */export async function runDaemon(options: RunDaemonOptions): Promise<void> { const sleep = options.sleep ?? defaultSleep const stopped = (): boolean => Boolean(options.shouldStop?.() || options.signal?.aborted) const introduced = new Set<string>() // Install a fresh controller before the tick begins. A stream arrival after its space has synced // must remain visible through the sync-to-sleep boundary; otherwise aborting the previous tick's // controller can be lost immediately before a long backfill sleep. let waking: AbortController | undefined let wakePending = false options.onWake?.(() => { wakePending = true waking?.abort() }) do { wakePending = false waking = new AbortController() for (const { spaceUri, runtime, actors: spaceActors } of options.spaces) { try { const index = await runtime.sync() // Which identities write into this space, and how. Everything downstream consumes a registry // and never asks which bus produced it, which is what keeps dispatch, claims and auto-review // free of any private-mode branch (design §18). const actors = spaceActors ?? options.actors if (!introduced.has(spaceUri)) { introduced.add(spaceUri) options.onFirstIndex?.(spaceUri, index, actors) } options.skew?.observe(index) // Claims first: a claim confirmed against THIS index dispatches on THIS tick. options.claimManager?.pump(index, actors) options.dispatcher.pump(index, actors) options.checkDispatcher?.pump(index, actors) options.mergePoller?.pump(index, actors) options.autoReviewTrigger?.pump(index, actors) } catch (error) { options.log?.(`sync failed for ${spaceUri}: ${error instanceof Error ? error.message : String(error)}`) } } if (stopped()) return // The arrival is already durable in the store/inbox, so start another tick immediately and // let the normal sync pipeline consume it instead of entering the streaming backfill sleep. if (wakePending) continue const interval = typeof options.intervalMs === 'function' ? options.intervalMs() : options.intervalMs const signals = options.signal ? [options.signal, waking.signal] : [waking.signal] await sleep(interval, AbortSignal.any(signals)) if (stopped()) return } while (true)}
export async function runSyncLoop(options: { watch: boolean intervalMs: number sync: () => Promise<MaterializedIndex> onIndex: (index: MaterializedIndex) => void onError: (error: unknown) => void sleep?: (milliseconds: number) => Promise<void> continueWatching?: () => boolean}): Promise<void> { const sleep = options.sleep ?? ((milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds))) do { try { options.onIndex(await options.sync()) } catch (error) { if (!options.watch) throw error options.onError(error) } if (!options.watch || options.continueWatching?.() === false) return await sleep(options.intervalMs) } while (true)}