Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247import type { ArtifactRecord, CheckrunRecord, IndexedRecord, MaterializedIndex, ProjectCheck,} from '@radial/core'import { activeGoals, activeProjects, compareCodePoints } from '@radial/core'import type { ActorRegistry, LoadedActor } from './actors.js'import type { CheckKey, CheckLedger } from './check-ledger.js'import { checkContainerLabel, type CheckRunInput, type CheckRunResult } from './check-runner.js'
export interface Checkable { artifact: IndexedRecord<ArtifactRecord> commit: string gitUrl: string checks: ProjectCheck[] /** The authoring identity (first of the daemon's loaded actors by DID — deterministic). */ actor: LoadedActor}
/** Composite key string for the in-flight map and run-dir derivation. */export function checkKeyString(key: CheckKey): string { return `${key.artifactUri}#${key.artifactCid}#${key.commit}`}
const checkrunMatches = (checkrun: IndexedRecord<CheckrunRecord>, artifactUri: string, artifactCid: string, commit: string): boolean => checkrun.value.artifact.uri === artifactUri && checkrun.value.artifact.cid === artifactCid && checkrun.value.commit === commit
/** * The pure check-dispatch gate: which artifacts carrying `links.commit` still need a checkrun from * this daemon right now. An artifact is selected iff (1) it links a commit, (2) it resolves to a * project (goal-scoped: artifact → goal → project; project-scoped: directly) whose `checks` are * non-empty, (3) no *trusted* checkrun already exists for this exact artifact version (uri#cid) + * commit authored by one of THIS daemon's DIDs (a checkrun from another member does NOT suppress * ours — every operator annotates independently), and (4) the check ledger says it is eligible (not * running / done / gave-up / cooling down). Deterministically ordered by artifact uri then commit so * dispatch order is stable. Independent of turn dispatch — a checkable never competes for the turn * budget. */export function selectCheckable( index: MaterializedIndex, actors: ActorRegistry, checkLedger: CheckLedger, ourDids: Set<string>, now?: string, onReject?: (artifactUri: string, commit: string, reason: string) => void,): Checkable[] { // Deterministic author: the first loaded actor by DID. The daemon's own agent identities are the // space's members, so their checkruns index as trusted. No actors → nothing to author with. const author = [...actors.all].sort((a, b) => compareCodePoints(a.did, b.did))[0]
// A shelved goal is set aside: agents run no checks on its artifacts. `activeGoals` is ended AND // under-an-archived-project in one rule; an archived project's own artifacts go with it, because // putting a repository away has to stop the machinery that clones it. const goals = activeGoals(index) const projects = activeProjects(index) const allArtifacts: Array<IndexedRecord<ArtifactRecord>> = [ ...goals.flatMap((view) => view.artifacts), ...projects.flatMap((view) => view.artifacts), ] const allCheckruns: Array<IndexedRecord<CheckrunRecord>> = [ ...goals.flatMap((view) => view.checkruns), ...projects.flatMap((view) => view.checkruns), ]
const results: Checkable[] = [] for (const artifact of allArtifacts) { const commit = artifact.value.links?.commit if (!commit) continue // silently skipped — most artifacts carry no commit and aren't checkable
// Resolve the owning project for gitUrl + checks. let projectUri: string | undefined if (artifact.value.goal) { const goalView = index.goals.find((view) => view.target.uri === artifact.value.goal?.uri) projectUri = goalView?.target.value.project.uri } else if (artifact.value.project) { projectUri = artifact.value.project.uri } const projectView = projectUri ? index.projects.find((view) => view.target.uri === projectUri) : undefined if (!projectView) { onReject?.(artifact.uri, commit, 'could not resolve the artifact to a project') continue } const checks = projectView.target.value.checks if (!checks || checks.length === 0) { // No checks is an intentional project configuration, not a diagnosable rejection. Keep this // silent just like artifacts without commits: reporting it once per artifact becomes noisy // when one dispatcher pumps multiple space indexes in turn. continue }
if (!author) { onReject?.(artifact.uri, commit, 'no loaded actor to author a checkrun') continue }
// Gate (a): our own trusted checkrun already covers this exact artifact version + commit. const alreadyOurs = allCheckruns.some( (checkrun) => ourDids.has(checkrun.did) && checkrunMatches(checkrun, artifact.uri, artifact.cid, commit), ) if (alreadyOurs) { onReject?.(artifact.uri, commit, 'a checkrun from this daemon already exists for this artifact version + commit') continue }
// Gate (b): the ledger's in-flight / cooldown window. const key: CheckKey = { artifactUri: artifact.uri, artifactCid: artifact.cid, commit } if (!checkLedger.eligible(key, now)) { onReject?.(artifact.uri, commit, `ledger: not eligible (state ${checkLedger.get(key)?.state ?? 'unknown'})`) continue }
results.push({ artifact, commit, // The EFFECTIVE remote — the project record's own overlaid by its author's `editProject` // records — so a repository that has moved is cloned from where it is now. gitUrl: projectView.gitUrl, checks, actor: author, }) }
return results.sort((a, b) => compareCodePoints(a.artifact.uri, b.artifact.uri) || compareCodePoints(a.commit, b.commit))}
export interface CheckDispatcherDeps { ledger: CheckLedger concurrency: number /** This daemon's own agent DIDs; a checkrun already authored by one of these suppresses re-running. */ ourDids: Set<string> now?: () => string runDirFor: (key: CheckKey) => string image: string timeoutMs: number allowedSchemes: string[] /** Container memory limit (docker `--memory`); unset lets check-runner apply its `4g` floor. */ memory?: string /** This daemon instance's id (`instanceId(stateDir)`), mixed into every container label — the * check-side half of the same machine-global docker namespace problem turns have. */ instance?: string /** Injected wrapper around check-runner's runCheckRun (binds runner/checkout/now). */ runCheckRun: (input: CheckRunInput) => Promise<CheckRunResult> log?: (message: string) => void}
/** * Bounded-concurrency, tracked check dispatcher — the exact shape of `TurnDispatcher`, on its own * in-flight map and `checkConcurrency` budget so it never blocks (or is blocked by) turn dispatch. * `pump` launches eligible check runs in the background and returns immediately; every settlement * (done, crashed, or a thrown infra error) updates the check ledger before the promise leaves the * in-flight set, so a check's fate is never lost to an unobserved rejection. */export class CheckDispatcher { readonly #deps: CheckDispatcherDeps readonly #inFlight = new Map<string, { promise: Promise<void>; label: string }>() readonly #lastReject = new Map<string, string>()
constructor(deps: CheckDispatcherDeps) { this.#deps = deps }
get inFlight(): number { return this.#inFlight.size }
runningLabels(): string[] { return [...this.#inFlight.values()].map((entry) => entry.label) }
/** Launches eligible check runs from `index` up to remaining concurrency; returns immediately. */ pump(index: MaterializedIndex, actors: ActorRegistry): void { const now = this.#deps.now?.() const seenRejects = new Set<string>() const checkable = selectCheckable(index, actors, this.#deps.ledger, this.#deps.ourDids, now, (uri, commit, reason) => { const key = `${uri}#${commit}` seenRejects.add(key) if (this.#lastReject.get(key) !== reason) { this.#lastReject.set(key, reason) this.#deps.log?.(`not checking ${uri} @ ${commit}: ${reason}`) } }) for (const key of this.#lastReject.keys()) if (!seenRejects.has(key)) this.#lastReject.delete(key) if (checkable.length > 0) { this.#deps.log?.( `${checkable.length} checkable artifact(s); ${this.#inFlight.size}/${this.#deps.concurrency} in flight`, ) }
for (const item of checkable) { if (this.#inFlight.size >= this.#deps.concurrency) break const key: CheckKey = { artifactUri: item.artifact.uri, artifactCid: item.artifact.cid, commit: item.commit } const keyStr = checkKeyString(key) if (this.#inFlight.has(keyStr)) continue
const label = checkContainerLabel(key.artifactUri, key.artifactCid, key.commit, this.#deps.instance) this.#deps.ledger.markRunning(key, { containerLabel: label }) this.#deps.log?.(`dispatching check ${key.artifactUri} @ ${key.commit} (label ${label})`)
const input: CheckRunInput = { artifact: { uri: item.artifact.uri, cid: item.artifact.cid }, commit: item.commit, gitUrl: item.gitUrl, checks: item.checks, client: item.actor.client, runDir: this.#deps.runDirFor(key), image: this.#deps.image, timeoutMs: this.#deps.timeoutMs, allowedSchemes: this.#deps.allowedSchemes, ...(this.#deps.memory !== undefined ? { memory: this.#deps.memory } : {}), ...(this.#deps.instance !== undefined ? { instance: this.#deps.instance } : {}), }
const promise = this.#deps .runCheckRun(input) .then((result) => { if (result.outcome === 'done') { this.#deps.ledger.markDone(key, result.ref) } else { const outcome = this.#deps.ledger.markCrashed(key) if (outcome.state === 'gave_up') { this.#deps.log?.(`check ${keyStr} GAVE UP after ${outcome.attempts} attempts (label ${label})`) } } }) .catch((error: unknown) => { const outcome = this.#deps.ledger.markCrashed(key) const message = error instanceof Error ? error.message : String(error) this.#deps.log?.(`check ${keyStr} threw: ${message}; now ${outcome.state} (attempt ${outcome.attempts})`) }) .finally(() => { this.#inFlight.delete(keyStr) })
this.#inFlight.set(keyStr, { promise, label }) } }
/** Awaits every in-flight check (best-effort graceful shutdown; each promise already never rejects). */ async drain(): Promise<void> { await Promise.all([...this.#inFlight.values()].map((entry) => entry.promise)) }}