Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610#!/usr/bin/env node
import { argv, stdin, stdout } from 'node:process'import { createHash } from 'node:crypto'import { realpathSync } from 'node:fs'import { mkdir, rm } from 'node:fs/promises'import { dirname, isAbsolute, join } from 'node:path'import { pathToFileURL } from 'node:url'import { ClockSkewObserver, FetchRepoTransport, clockSkewThresholdMs, observingFetch, resolveHandleDid,} from '@radial/atproto'import { FileSessionStore, defaultDataDirectory, nodeResolveTxt } from '@radial/atproto/node'import { MAX_CLAIM_LEASE_MS, agentTypesFor, compareCodePoints, deviceKeyStatus, ensurePublishedDevice, indexDigest, retireDeviceAddress, rotateDeviceKey, type MaterializedIndex,} from '@radial/core'import { FileDeviceKeyStore } from '@radial/core/node'import { actorEverAccepts, actorTypesFor, loadActors, type ActorClient, type ActorRegistry } from './actors.js'import { AutoReviewTrigger } from './auto-review.js'import { ClaimLedger } from './claim-ledger.js'import { ClaimManager, DEFAULT_CLAIM_TIMING, MAX_CLAIM_GENERATIONS, type ClaimTiming,} from './claims.js'import { CONFIG_FILENAME, buildDefaultConfig, defaultConfigPath, extraIdentities, writeConfig, parseModel, passwordEnvName, resolveAgentInits, configuredForges, type ClaimsConfig, type DaemonRunConfig, type Environment, type ForgeConfig, type PrivateSpaceConfig,} from './config.js'import { ConfigNotFoundError, describeDataDirSource, instanceId, loadInstance, resolveDataDir, type Instance,} from './instance.js'import { StateDirLock } from './state-lock.js'import { FetchBobbin } from './bobbin.js'import { remoteBranchExists } from './bundle-writer.js'import { CheckDispatcher } from './check-dispatch.js'import { CheckLedger } from './check-ledger.js'import { runCheckRun, type CheckRunInput } from './check-runner.js'import { CodexAuthStore, loginManagedCodex } from './codex-auth.js'import { DockerRunner, type ContainerRunner } from './container.js'import { TurnDispatcher } from './dispatch.js'import { ForgeRegistry, type ForgeAdapter, type PullStateContext } from './forge.js'import { GitHubForge } from './forge-github.js'import { TangledForge } from './forge-tangled.js'import { TangledKeyStore, knownHostsFile } from './forge-auth.js'import { resolveGitHubToken } from './github-auth.js'import { ensureGitHubAuth } from './github-auth.js'import { HARNESSES, selectHarness } from './harness.js'import { initializeAgent } from './init.js'import { TurnLedger, type TurnState } from './ledger.js'import { MergePoller } from './merge-poll.js'import { openPrivateSpaces } from './private-run.js'import { SkewWatch, SpaceSyncRuntime, futureDatedAuthors, indexSummary, readIndex, runDaemon, runSyncLoop, syncSpaceOnce, type DaemonSpace,} from './runtime.js'import { runTurn, turnContainerLabel, type TurnInput } from './turn.js'
const usage = `Usage: radiald config init [handle-or-did] [--config <path>] [--data-dir <path>] [--force] writes a starter ${CONFIG_FILENAME} to ~/.config/radial/, prompting for the agent handle when it is not given. --data-dir writes a "dataDir" into it, so that one --config selects this instance — sessions, state and all radiald init [profile]... [--config <path>] [--data-dir <path>] [--password-stdin] [--update] [--replace-identity] initializes every profile in ${CONFIG_FILENAME} (or only the named ones), read from the nearest ${CONFIG_FILENAME} or ~/.config/radial/${CONFIG_FILENAME}; the agent identity's app password is read from stdin or $RADIAL_PASSWORD. --update republishes a profile whose "artifactTypes" (or name, harness, or models) have changed since it was published — the agent record is what an assignee menu offers, so editing ${CONFIG_FILENAME} alone changes nothing. --replace-identity re-registers a profile under a different DID, which is otherwise refused (it is how one instance silently takes another's identity) radiald init <profile> --pds <url> --identifier <handle-or-did> --name <name> --harness <${HARNESSES.map((harness) => harness.name).join('|')}> [--model <name=cost>]... --artifact-type <type>... --password-stdin [--update] radiald sync <space-uri> --records <sqlite-path> --checkpoints <sqlite-path> [--watch] [--interval <milliseconds>] radiald index <space-uri> --records <sqlite-path> [--digest] --digest prints a canonical fingerprint of what the fold decided, plus one per section; two operators compare them to find a sync gap (design §12). It also prints "clocks": every author that has dated a record in this machine's future, so two operators comparing digests compare clocks too radiald index reset [--config <path>] [--data-dir <path>] clears the local synced-record index and its checkpoints under "run.stateDir"; refuses while a radiald run holds that state dir (turn and check ledgers are preserved). Re-ingesting also restarts every live claim's lease on THIS machine: a lease is honoured for its declared duration from first sighting, and after a reset every record is being seen for the first time again radiald run [--config <path>] [--data-dir <path>] [--interval <milliseconds>] the long-lived daemon: polls every space in the config's "run" block, dispatches eligible plan requests to sandboxed containers, and durably tracks each attempt in a turn ledger under "run.stateDir". It prints the config, data dir, sessions file and state dir it resolved, plus the DID every profile is running as, and holds a lock on the state dir so a second daemon cannot open the same ledgers radiald codex login [--config <path>] [--data-dir <path>] creates a fresh, file-backed ChatGPT login dedicated to this Radial instance. Requires run.codexAuth.mode = "chatgpt-session"; never copies ~/.codex radiald turn list [--state <state>] [--config <path>] [--data-dir <path>] prints the turn ledger, newest first: state, when it last moved, the request uri, and attempts/retry/branch. --state narrows it — use "--state awaiting_input" to see every request parked on a question radiald turn reset <request-uri> [--config <path>] [--data-dir <path>] clears a request's ledger row (including a gave-up/cooling-down one) so it becomes dispatchable again radiald check reset <artifact-uri> [--config <path>] [--data-dir <path>] clears every check-ledger row for an artifact uri (all versions/commits) so its checks run again radiald device list|publish|rotate|retire [--did <did>] the signing keys this instance's identities use in a PRIVATE space (design §18). "publish" writes each identity's public device record; "rotate" mints a new key and retires the old one, which the fold never notices; "retire <deviceKeyId>" withdraws a device's published address, so no replica dials it or serves it. Neither changes what any replica counts — everything the key signed keeps counting, and a key stolen along with its repo credentials republishes its own address (remove the member instead) radiald claim reset <request-uri> [--config <path>] [--data-dir <path>] clears this daemon's claim rows for an open request, so it may claim it again; one that has burned through every claim generation gets a fresh batch of them, starting above the records it already wrote`
function values(args: string[], flag: string): string[] { const found: string[] = [] for (let index = 0; index < args.length; index += 1) { if (args[index] === flag && args[index + 1]) found.push(args[index + 1] as string) } return found}
function one(args: string[], flag: string): string { const found = values(args, flag)[0] if (!found) throw new Error(`Missing ${flag}\n${usage}`) return found}
/** Bare arguments, skipping every `--flag` and the value that follows it. *//** * Flags that carry no value. Everything else consumes the token after it, so a switch missing from * here silently eats the argument behind it — `radiald init --update planner` would init every * profile rather than one, which reads as the command ignoring what was typed. */const VALUELESS_FLAGS = new Set([ '--password-stdin', '--update', '--replace-identity', '--force', '--skip-github-auth', '--watch', '--digest',])
function positional(args: string[]): string[] { const found: string[] = [] for (let index = 0; index < args.length; index += 1) { const entry = args[index] as string if (entry.startsWith('--')) { if (!VALUELESS_FLAGS.has(entry)) index += 1 } else found.push(entry) } return found}
async function readStdin(): Promise<string> { let value = '' for await (const chunk of stdin) value += new TextDecoder().decode(chunk) return value.trim()}
async function prompt(question: string): Promise<string> { const { createInterface } = await import('node:readline/promises') const rl = createInterface({ input: stdin, output: stdout }) try { return (await rl.question(question)).trim() } finally { rl.close() }}
async function configInit(args: string[]): Promise<void> { const path = values(args, '--config')[0] ?? defaultConfigPath() const dataDir = values(args, '--data-dir')[0] let identifier = positional(args)[0] if (!identifier) { if (!stdin.isTTY) throw new Error(`Missing agent handle\n${usage}`) identifier = await prompt('Agent handle or DID: ') } if (!identifier) throw new Error('An agent handle or DID is required') const config = buildDefaultConfig(identifier, dataDir ? { dataDir } : {}) await writeConfig(path, config, { force: args.includes('--force') }) const profiles = Object.keys(config.agents).join(', ') console.log(`Wrote ${path}`) console.log(` identity ${identifier}, profiles: ${profiles}`) if (dataDir) console.log(` data dir ${dataDir} (resolved relative to ${dirname(path)})`) console.log(`Edit it, then register the profiles:`) // The config path is spelled out rather than left implicit: with several instances on one machine, // an init that discovers "the nearest radial.json" is exactly how the wrong one gets initialized. console.log(` printf '%s\\n' "$APP_PASSWORD" | radiald init --config ${path} --password-stdin`) console.log(` radiald run --config ${path}`)}
export interface ResolvedRunConfig { spaces: string[] /** Spaces served over the private bus (design §18). Empty unless the operator configured one. */ privateSpaces: PrivateSpaceConfig[] stateDir: string image: string concurrency: number timeoutMs: number cooldownMs: number retryBound: number mergePollIntervalMs: number mergePollBackoffMaxMs: number /** * How often to check whether this daemon's private endpoint has moved (ADR §21). * * The check compares the endpoint's live address against what this process last published and * touches a PDS only when they differ, so this is a cheap local poll rather than a write rate. * Inert unless a private space is configured. */ privateAddressRefreshMs: number network?: string /** Extra environment variable names forwarded into turn containers, on top of what each * profile's harness declares. Defaults to `[]`. See `DaemonRunConfig.modelEnv`. */ modelEnv: string[] /** Explicit refreshable ChatGPT authentication for Codex turns. */ codexAuth?: { mode: 'chatgpt-session' } /** The resolved transport: `auto` has already become `unix` or `tcp` here. */ turnTransport: 'unix' | 'tcp' gitSchemes: string[] /** Container memory limit (docker `--memory` syntax, e.g. `4g`). No default at this layer — * `turn.ts` supplies the `4g` floor when unset (a turn must never run unbounded). */ memory?: string /** Check-runner budgets. `checkImage` defaults to `image` (a check runs the same base image as a * turn unless the operator ships a dedicated one). */ checkConcurrency: number checkTimeoutMs: number checkImage: string checkCooldownMs: number checkRetryBound: number /** Forge integration (design §10), as the operator spelled it. `forges` is the normalized list * every caller should read; `forge` is retained verbatim so a round trip through `writeConfig` * does not rewrite an operator's file. */ forge?: ForgeConfig forges: ForgeConfig[] /** Claim/lease timing for open requests, fully defaulted (design §11). */ claims: ClaimTiming /** Jetstream ingestion, when the operator opted in. `backfillIntervalMs` is defaulted here. */ jetstream?: { endpoint: string; backfillIntervalMs: number }}
/** A default under the operator's atproto state directory, falling back to a project-local * directory when there is no resolvable data dir (e.g. no $HOME in a minimal environment). */function defaultRunStateDir(): string { try { return join(defaultDataDirectory(), 'run') } catch { return join(process.cwd(), '.radial-state') }}
/** * `config.js` validates the raw `run` block; the daemon applies runtime defaults. * * `options.dataDir` is this instance's resolved data directory: with it, `stateDir` defaults to * `<dataDir>/run`, so one setting separates *everything* an instance owns. Without it the historic * default stands, which is what keeps every existing call site (and test) meaning what it meant. */export function resolveRunConfig( run: DaemonRunConfig, options: { dataDir?: string; platform?: string } = {},): ResolvedRunConfig { const configuredTransport = run.turnTransport ?? 'auto' const platform = options.platform ?? process.platform const turnTransport = configuredTransport === 'auto' ? platform === 'darwin' || platform === 'win32' ? 'tcp' : 'unix' : configuredTransport if (run.network?.toLowerCase() === 'host' || /^container:/i.test(run.network ?? '')) throw new Error('run.network must not use host or container networking') const mergePollInterval = Math.max(1, run.mergePollIntervalMs ?? 60_000) return { spaces: run.spaces, privateSpaces: run.privateSpaces ?? [], stateDir: run.stateDir ?? (options.dataDir ? join(options.dataDir, 'run') : defaultRunStateDir()), image: run.image ?? 'radial-turn:latest', concurrency: run.concurrency ?? 1, timeoutMs: run.timeoutMs ?? 60 * 60_000, cooldownMs: run.cooldownMs ?? 300_000, retryBound: run.retryBound ?? 3, // Clamp poll timing: a zero/negative base interval would busy-poll the forge, so floor it at // 1ms; the backoff cap is floored at the base interval so `backoffMaxMs < pollIntervalMs` can // never yield an uncapped first delay (the poller also caps its first step defensively). mergePollIntervalMs: mergePollInterval, mergePollBackoffMaxMs: Math.max(mergePollInterval, run.mergePollBackoffMaxMs ?? 30 * 60_000), // A minute: long enough that the local comparison is free, short enough that a laptop which // changed network is dialable again before a peer's own backoff has grown past it. privateAddressRefreshMs: Math.max(1_000, run.privateAddressRefreshMs ?? 60_000), ...(run.network !== undefined ? { network: run.network } : {}), modelEnv: run.modelEnv ?? [], ...(run.codexAuth !== undefined ? { codexAuth: run.codexAuth } : {}), turnTransport, gitSchemes: run.gitSchemes ?? ['https'], ...(run.memory !== undefined ? { memory: run.memory } : {}), checkConcurrency: run.checkConcurrency ?? 1, checkTimeoutMs: run.checkTimeoutMs ?? 15 * 60_000, checkImage: run.checkImage ?? run.image ?? 'radial-turn:latest', checkCooldownMs: run.checkCooldownMs ?? 300_000, checkRetryBound: run.checkRetryBound ?? 3, ...(run.forge !== undefined ? { forge: run.forge } : {}), forges: configuredForges(run), claims: resolveClaimTiming(run.claims), ...(run.jetstream !== undefined ? { jetstream: { endpoint: run.jetstream.endpoint, backfillIntervalMs: run.jetstream.backfillIntervalMs ?? 60_000, }, } : {}), }}
/** * Claim timing with its defaults applied. `renewIntervalMs` derives from the lease rather than * carrying an independent default, so an operator who lengthens the lease alone still gets three * renewal attempts inside it; the same third-of-a-lease rule `parseRunConfig` enforces is applied * here too, for the case where only ONE of the pair was set in the file. */export function resolveClaimTiming(claims: ClaimsConfig | undefined): ClaimTiming { const leaseMs = claims?.leaseMs ?? DEFAULT_CLAIM_TIMING.leaseMs const renewIntervalMs = claims?.renewIntervalMs ?? Math.floor(leaseMs / 3) // Checked here as well as in `parseRunConfig`, because a `ClaimTiming` can reach the manager // without ever passing through a config file (tests, and `radiald`'s own defaults). if (leaseMs > MAX_CLAIM_LEASE_MS) { throw new Error( `run.claims.leaseMs (${leaseMs}ms) must be at most ${MAX_CLAIM_LEASE_MS}ms — the protocol bound on a ` + 'declared claim lease, above which every materializer ignores the claim', ) } if (renewIntervalMs * 3 > leaseMs) { throw new Error( `run.claims.renewIntervalMs (${renewIntervalMs}ms) must be at most a third of run.claims.leaseMs (${leaseMs}ms), ` + 'so a lease survives two failed renewals', ) } return { leaseMs, renewIntervalMs, confirmCycles: claims?.confirmCycles ?? DEFAULT_CLAIM_TIMING.confirmCycles, maxOutstanding: claims?.maxOutstanding ?? DEFAULT_CLAIM_TIMING.maxOutstanding, }}
/** * The one claim relationship that cannot be checked from the config file alone: the poll interval * is a `radiald run` flag, not a `run` field. A lease shorter than four polls is a daemon that * fights itself — it can lose a claim it holds between two consecutive ingestion cycles, hand the * request to another operator, and reclaim it on the next tick. Loud, at startup, is the only place * an operator will connect the two numbers. Pure and exported so the wording is testable. */export function claimTimingAdvisory(timing: ClaimTiming, intervalMs: number): string | undefined { if (timing.leaseMs >= 4 * intervalMs) return undefined return ( `run.claims.leaseMs (${timing.leaseMs}ms) must be at least four times the poll interval ` + `(${intervalMs}ms): a lease that can lapse between two ingestion cycles makes this daemon drop ` + 'claims it is actively working on. Raise the lease or lower --interval.' )}
/** Every name the daemon may forward, in a stable order: what the loaded profiles' harnesses * declare, then whatever `run.modelEnv` adds. Deduplicated, so two `pi` profiles do not repeat the * whole provider list. */export function modelEnvNames( actors: readonly { harness: string }[], extra: readonly string[],): string[] { const names = new Set<string>() for (const actor of actors) for (const name of selectHarness(actor.harness).credentialEnv) names.add(name) for (const name of extra) names.add(name) return [...names]}
/** Names → the values the daemon actually holds. A name absent from the daemon's own environment is * simply not forwarded: the daemon never invents a credential. Pure, so it is unit-testable. */export function collectModelEnv(names: readonly string[], env: Environment): Record<string, string> { const collected: Record<string, string> = {} for (const name of names) { const value = env[name] if (value !== undefined && value !== '') collected[name] = value } return collected}
/** * The provider credentials this daemon will hand every turn, decided ONCE at startup. * * Both failures below are knowable before the first dispatch — the loaded profiles and the daemon's * environment do not change while it runs — and discovering either one per-turn instead would burn * a ledger attempt and a cooldown per request (`dispatch.ts`) to produce an error the operator * could have been told about at launch: * * - a profile naming a harness this daemon cannot run; * - a profile whose harness has no credential available at all, which is an authentication failure * inside the container dressed up as a crashed turn. * * The second check counts an operator's `run.modelEnv` entries as candidates, because it cannot * know which of them is the credential — so it is a guard against the common misconfiguration * ("I forgot to export the key"), not a proof that the turn will authenticate. */export function resolveModelEnvironment( actors: readonly { profile: string; harness: string }[], extra: readonly string[], env: Environment, options: { managedCodexAuth?: boolean } = {},): Record<string, string> { for (const actor of actors) { try { selectHarness(actor.harness) } catch (error) { throw new Error(`profile "${actor.profile}": ${error instanceof Error ? error.message : String(error)}`) } } const codexActors = actors.filter((actor) => actor.harness === 'codex') if (options.managedCodexAuth && codexActors.length === 0) { throw new Error('run.codexAuth is configured but no loaded profile uses harness "codex"') } if (options.managedCodexAuth) { const conflicts = ['CODEX_API_KEY', 'CODEX_ACCESS_TOKEN'].filter( (name) => env[name] !== undefined && env[name] !== '', ) if (conflicts.length > 0) { throw new Error( `run.codexAuth.mode is "chatgpt-session", but ${conflicts.join(', ')} is also set. ` + 'Unset it so it cannot silently override the dedicated ChatGPT login.', ) } } const names = modelEnvNames(actors, extra).filter( (name) => !options.managedCodexAuth || (name !== 'CODEX_API_KEY' && name !== 'CODEX_ACCESS_TOKEN'), ) const modelEnv = collectModelEnv(names, env) for (const actor of actors) { if (options.managedCodexAuth && actor.harness === 'codex') continue const accepted = [...selectHarness(actor.harness).credentialEnv, ...extra] if (accepted.some((name) => modelEnv[name] !== undefined)) continue throw new Error( `profile "${actor.profile}" runs harness "${actor.harness}" but none of its provider credentials are set ` + `in this daemon's environment. Export one of: ${accepted.join(', ')}` + (extra.length === 0 ? ', or add the one you use to "run.modelEnv".' : '.'), ) } return modelEnv}
async function codexLoginCommand(args: string[]): Promise<void> { const { instance, run } = await requireRunConfig(args) if (run.codexAuth?.mode !== 'chatgpt-session') { throw new Error( `${instance.configPath} does not enable managed Codex auth; add ` + '"codexAuth": { "mode": "chatgpt-session" } to its "run" block', ) } const directory = join(instance.dataDir.path, 'codex-auth') // A login replaces the refresh-token lineage. Refuse while this instance can be running a turn // against the old lineage; the state-dir lock is the same single-daemon boundary `run` holds. const stateLock = StateDirLock.acquire(run.stateDir) try { console.log(`Starting a fresh Radial-specific Codex device login`) console.log(` auth: ${join(directory, 'auth.json')}`) await loginManagedCodex(directory) await new CodexAuthStore(directory).validate() console.log(`Managed Codex login saved for ${instance.configPath}`) } finally { stateLock.release() }}
/** Deterministic per-request run directory under `<stateDir>/turns/`. Kept short (12 hex chars, * not a full hash) because the turn socket lives at `<runDir>/run/turn.sock`, and AF_UNIX socket * paths have a small platform-enforced max length (~104-108 bytes) that a long stateDir plus a * long directory name can blow through. */function turnRunDir(stateDir: string, requestUri: string): string { const digest = createHash('sha256').update(requestUri).digest() const hex = [...digest].map((byte) => byte.toString(16).padStart(2, '0')).join('') return join(stateDir, 'turns', hex.slice(0, 12))}
/** Deterministic per-check run directory under `<stateDir>/checks/`, keyed by artifact version + * commit. No unix socket lives under it (checks have no turn socket), so length is not as tight as * `turnRunDir`, but the same short-hash convention keeps paths tidy. */function checkRunDir(stateDir: string, key: { artifactUri: string; artifactCid: string; commit: string }): string { const digest = createHash('sha256').update(`${key.artifactUri}#${key.artifactCid}#${key.commit}`).digest() const hex = [...digest].map((byte) => byte.toString(16).padStart(2, '0')).join('') return join(stateDir, 'checks', hex.slice(0, 12))}
/** `loadInstance` with this CLI's `--config`/`--data-dir` flags and its usage text on the one error * an operator can fix by reading it. Every command that takes a config now takes a data dir too. */async function requireInstance(args: string[]): Promise<Instance> { const config = values(args, '--config')[0] const dataDir = values(args, '--data-dir')[0] try { return await loadInstance({ config, dataDir }) } catch (error) { if (error instanceof ConfigNotFoundError) throw new Error(`${error.message}\n${usage}`) throw error }}
/** * The whole argument vector rather than just `--config`, so `--data-dir` reaches every command that * reads the `run` block — and so each of them resolves its state directory the same way `radiald * run` does. A `turn list` that looked at a different ledger than the daemon writes would be worse * than no `turn list` at all. */async function requireRunConfig( args: string[],): Promise<{ instance: Instance; run: ResolvedRunConfig }> { const instance = await requireInstance(args) if (!instance.config.run) { throw new Error(`${instance.configPath} has no "run" block; add one before using "radiald run"\n${usage}`) } return { instance, run: resolveRunConfig(instance.config.run, { dataDir: instance.dataDir.path }) }}
const TURN_STATES: TurnState[] = ['pending', 'running', 'awaiting_input', 'fulfilled', 'crashed', 'gave_up']
/** * What the turn ledger holds, newest first. * * The operator-facing half of `turn reset`: before touching a row you want to see it, and before * enabling replies you want to see the `awaiting_input` backlog that is about to become dispatchable * (a parked request is `eligible()` the moment somebody answers its question). */async function turnListCommand(args: string[]): Promise<void> { const state = values(args, '--state')[0] as TurnState | undefined if (state && !TURN_STATES.includes(state)) { throw new Error(`Unknown turn state '${state}'; one of ${TURN_STATES.join(', ')}\n${usage}`) } const { run } = await requireRunConfig(args) const ledger = new TurnLedger(join(run.stateDir, 'ledger.db'), { retryBound: run.retryBound, cooldownMs: run.cooldownMs, }) try { const rows = state ? ledger.list(state) : ledger.list() if (rows.length === 0) { console.log(state ? `no turns in state ${state}` : 'the turn ledger is empty') return } for (const row of rows) { const detail = [ `attempts=${row.attempts}`, ...(row.nextEligibleAt ? [`retry-at=${row.nextEligibleAt}`] : []), ...(row.branch ? [`branch=${row.branch}`] : []), ].join(' ') console.log(`${row.state.padEnd(14)} ${row.updatedAt || '-'} ${row.requestUri} ${detail}`) } console.log(`${rows.length} turn${rows.length === 1 ? '' : 's'}`) } finally { ledger.close() }}
async function turnResetCommand(args: string[]): Promise<void> { const requestUri = args[0] if (!requestUri) throw new Error(`Missing <request-uri>\n${usage}`) const { run } = await requireRunConfig(args) const ledger = new TurnLedger(join(run.stateDir, 'ledger.db'), { retryBound: run.retryBound, cooldownMs: run.cooldownMs, }) try { ledger.reset(requestUri) console.log(`reset ${requestUri}`) } finally { ledger.close() }}
/** `radiald claim reset <request-uri>`: forget where this request's claim rows stand, the claim-side * parallel of `turn reset`. Useful when a lease is held by a profile the operator has retired, and * the way back for a request whose generation walk hit the cap — which is why it raises the row's * ceiling by `MAX_CLAIM_GENERATIONS` rather than dropping the row: the spent generations are * permanent records, and a walk sent back to zero would only rediscover them and stop again. */async function claimResetCommand(args: string[]): Promise<void> { const requestUri = args[0] if (!requestUri) throw new Error(`Missing <request-uri>\n${usage}`) const { run } = await requireRunConfig(args) const ledger = new ClaimLedger(join(run.stateDir, 'claim-ledger.db')) try { ledger.reset(requestUri, { generations: MAX_CLAIM_GENERATIONS }) } finally { ledger.close() } console.log(`reset claims for ${requestUri}`)}
async function checkResetCommand(args: string[]): Promise<void> { const artifactUri = args[0] if (!artifactUri) throw new Error(`Missing <artifact-uri>\n${usage}`) const { run } = await requireRunConfig(args) const ledger = new CheckLedger(join(run.stateDir, 'check-ledger.db'), { retryBound: run.checkRetryBound, cooldownMs: run.checkCooldownMs, }) try { ledger.reset(artifactUri) } finally { ledger.close() } console.log(`reset checks for ${artifactUri}`)}
/** * Removes only the derived ingestion state. SQLite may leave WAL/SHM sidecars behind, so clear * those too; the durable turn/check ledgers intentionally remain untouched. Callers must ensure * no `radiald run` process has these databases open. */export async function resetRunIndex(stateDir: string): Promise<void> { await Promise.all( ['records.db', 'records.db-wal', 'records.db-shm', 'sync.db', 'sync.db-wal', 'sync.db-shm'].map((file) => rm(join(stateDir, file), { force: true }), ), )}
async function indexResetCommand(args: string[]): Promise<void> { const { run } = await requireRunConfig(args) // "Stop radiald run first" used to be a line in `usage` and nothing else — deleting the databases // out from under a live daemon is exactly what the state-dir lock exists to make impossible. await mkdir(run.stateDir, { recursive: true }) const lock = StateDirLock.acquire(run.stateDir) try { await resetRunIndex(run.stateDir) } finally { lock.release() } console.log(`reset local index under ${run.stateDir}`)}
/** Kills any container(s) tracked under `label` (by id and by name) and logs the reconciliation. * Best-effort throughout: a container that is already gone is not an error. Exported (and typed * against the `ContainerRunner` interface rather than the concrete `DockerRunner`) so tests can * drive it directly with a `FakeContainerRunner`. */export async function reconcileOrphan( runner: ContainerRunner, label: string | undefined, requestUri: string,): Promise<void> { if (label) { try { for (const id of await runner.listByLabel(label)) { await runner.kill(id).catch(() => {}) } await runner.kill(label).catch(() => {}) } catch { // Best effort: reconciliation must never block startup. } } console.log(`reconciled orphaned turn ${requestUri} (label ${label ?? 'unknown'})`)}
/** Startup orphan reconciliation: any row still "running" belongs to a prior process that died * (or was killed) mid-turn. Its container (if any survived) is killed via `reconcileOrphan`, and * the row is transitioned with `ledger.markCrashed` — NOT `ledger.reset` — so a daemon that dies * mid-turn on every restart still counts each restart as a failed attempt (preserving `attempts`, * applying the cooldown, and eventually `gave_up`) instead of silently resetting the retry counter * and bypassing the retry bound forever. Exported so tests can drive it directly. */export async function reconcileOrphans(runner: ContainerRunner, ledger: TurnLedger): Promise<void> { for (const row of ledger.running()) { await reconcileOrphan(runner, row.containerLabel, row.requestUri) ledger.markCrashed(row.requestUri) }}
/** Startup orphan reconciliation for check runs — the exact parallel of `reconcileOrphans`: any * check-ledger row still `running` belongs to a prior process that died mid-check; its container (if * any survived) is killed and the row transitioned via `markCrashed` (attempts preserved, retry * bound respected). Exported so tests can drive it directly. */export async function reconcileCheckOrphans(runner: ContainerRunner, ledger: CheckLedger): Promise<void> { for (const row of ledger.running()) { const key = { artifactUri: row.artifactUri, artifactCid: row.artifactCid, commit: row.commit } await reconcileOrphan(runner, row.containerLabel, `${row.artifactUri}@${row.commit}`) ledger.markCrashed(key) }}
/** * The startup warning for a daemon that produces implementations without a live forge adapter. * * The adapter is what lets the daemon ask whether a predecessor's pull request is still open before * continuing its branch; without it a v2 turn decides from the local index alone, which cannot see a * pull request somebody closed without merging. This is the setting that decides whether resolving a * review's findings updates the pull request under review or opens a second one, and nothing in the * daemon's output used to name it — so an operator who omitted `run.forge` had no way to connect the * two. Pure and exported so the wording is testable rather than buried in a long-running command. */export function forgeAdvisory(input: { /** A live adapter resolved (configured AND GitHub authentication available). */ adapter: boolean /** `run.forge` is set in the config, whatever came of it. */ configured: boolean producesImplementation: boolean configPath: string}): string | undefined { if (input.adapter || !input.producesImplementation) return undefined const consequence = 'implementation v2 decides whether to continue its predecessor’s pull request from the local index and the git remote alone — nothing writes merge annotations without a forge, so a merged pull request whose branch survives reads like an open one, and a pull request closed without merging is invisible to it.' return input.configured ? `GitHub authentication is unavailable, so ${consequence}` : `no run.forge configured — merge observation is off, and ${consequence} Add "run": { "forge": { "kind": "github" } } to ${input.configPath}.`}
/** * One `getPullRequestState` over every configured forge, routed by the pull request URL itself. * * The routing itself has to be answerable from the URL alone — which is exactly what `ownsPullUrl` * is — because that is all the merge poller is guaranteed to hold. An unroutable URL throws, and * both callers already treat a throw as "back off and retry", never as an answer. * * The project `context` is passed straight through when a caller has it. It is a hint, not part of * the routing (see `PullStateContext`): tangled needs the project's owner to fold a status without * an appview, and a URL cannot name it. */export function routingForgeState(registry: ForgeRegistry): Pick<ForgeAdapter, 'getPullRequestState'> { return { async getPullRequestState(prUrl: string, context?: PullStateContext) { const adapter = registry.adapters.find((candidate) => candidate.canObserve && candidate.ownsPullUrl(prUrl)) if (!adapter) throw new Error(`no configured forge observes ${prUrl}`) return adapter.getPullRequestState(prUrl, context) }, }}
/** The profile name an agent record was published under — its rkey (`radiald init` keys by profile). */const profileOf = (uri: string): string => uri.slice(uri.lastIndexOf('/') + 1)
/** * The startup warnings for a space whose artifact-type registry and this operator's profiles have * drifted apart. * * The registry is data (design §4): a space admin adds a type and every client grows a button for * it, without anything being deployed. What no client can grow is an agent — a request whose type no * agent accepts has nobody to assign, so it is written and then sits, looking exactly like a request * whose agent is merely busy. Nothing in the daemon's output used to name the difference, and there * are two ways to arrive at it: * * - `uncovered` — the registry has a type NO agent in the space accepts. Seeded system types are the * common case: a space created today registers `architecture`, `architecture-inventory`, * `glossary`, `conventions`, `adr` and `fdr`, and a * `radial.json` written before them names none. * - `stale` — a profile's `artifactTypes` in the config no longer match the agent record it * published. The record is what an assignee menu reads, so editing the config alone changes * nothing anyone can see; `radiald init <profile> --update` is what publishes it. * * Pure and exported so the wording is testable rather than buried in a long-running command. */export function coverageAdvisories( index: MaterializedIndex, actors: ActorRegistry, where: { spaceUri: string; configPath: string },): string[] { const advisories: string[] = [] const list = (values: readonly string[]): string => values.join(', ') // In a private space `radiald init --update` is the wrong instruction — it writes to a PDS, and // the private fold reads `agent` only out of an envelope. The daemon publishes those itself, so // the operator's move is to edit the config and restart, and saying otherwise sent people to a // command that could not have helped (design §18). const republish = index.private ? 'restart this daemon — it publishes capabilities into a private space itself, and ' + `"radiald init --update" cannot: it writes to a PDS, which a private space's fold never reads` : undefined
// What the space's own records say is on offer, which is what a human's assignee menu reads — not // what this config believes, and not only this operator's agents: another member's daemon covering // a type is a perfectly good answer, and warning about it would be this daemon guessing. Read // through `agentTypesFor`, so a profile scoped down in THIS space does not count as covering a // type it only accepts somewhere else. const accepted = new Set( index.agents.flatMap((agent) => agentTypesFor(agent.value, where.spaceUri)), ) const registry = [...new Set(index.artifactTypes.map((record) => record.value.name))] const uncovered = registry.filter((name) => !accepted.has(name)) if (uncovered.length > 0) { // Name a profile to add them to only when there is one to name; with nothing initialized the // instruction to re-run init stands on its own. const target = actors.all[0]?.profile advisories.push( `${where.spaceUri}: the registry has ${uncovered.length === 1 ? 'a type' : 'types'} no agent in the space accepts — ` + `${list(uncovered)}. A request for ${uncovered.length === 1 ? 'it' : 'one'} has no agent to assign and will sit open. ` + `Add ${uncovered.length === 1 ? 'it' : 'them'} to a profile's "artifactTypes" in ${where.configPath}, then ` + (republish ?? `run: radiald init ${target ?? '<profile>'} --update`), ) }
// Config-vs-published drift, per profile this daemon actually loaded. Comparing against the record // published under the SAME profile name is the point: the daemon dispatches on the config, humans // and auto-review assign from the record, and only the second one is visible in the space. Both // sides are the EFFECTIVE lists for this space, so a profile whose scope and whose record agree // here is quiet even when their global lists differ. const published = new Map( index.agents .filter((agent) => actors.all.some((actor) => actor.did === agent.did)) .map((agent) => [ `${agent.did}/${profileOf(agent.uri)}`, agentTypesFor(agent.value, where.spaceUri), ]), ) for (const actor of actors.all) { const record = published.get(`${actor.did}/${actor.profile}`) const configured = actorTypesFor(actor, where.spaceUri) if (!record || list([...record].sort()) === list([...configured].sort())) continue advisories.push( `profile "${actor.profile}" publishes ${list(record)} but ${where.configPath} says ` + `${list(configured)} in this space — the published record is what an assignee menu offers, so ` + (republish ?? `run: radiald init ${actor.profile} --update`), ) } return advisories}
/** * The paths this invocation actually resolved, as the lines `radiald run` prints. * * Every failure this whole feature is about is an operator who believed a different set of paths: * the wrong sessions file, a state directory shared with another daemon, a `$RADIAL_DATA_DIR` left * over in a shell. None of them are visible from the command line that was typed, so the daemon says * them out loud once, at the top, including WHICH rule chose the data directory. Pure and exported * so the wording is testable, the same way `forgeAdvisory` and `coverageAdvisories` are. */export function describeInstance(instance: Instance, run?: ResolvedRunConfig): string[] { const lines = [ ` config: ${instance.configPath}`, ` data: ${instance.dataDir.path} (${describeDataDirSource(instance.dataDir)})`, ` sessions: ${instance.sessions.path}`, ] if (run) lines.push(` state: ${run.stateDir} (instance ${instanceId(run.stateDir)})`) if (instance.dataDir.shadowed) { lines.push( ` ⚠️ $RADIAL_DATA_DIR overrode "dataDir": ${instance.dataDir.shadowed} in ${instance.configPath} ` + 'is not the directory in use. Unset it to run the instance the config describes.', ) } // Unlike `dataDir`, a relative `run.stateDir` is resolved against the WORKING DIRECTORY (§5 of the // plan: changing that would silently relocate the state of every operator using the documented // `"stateDir": ".radial-state"`). Two instances started from two directories therefore get two // state dirs from one config, which is worth saying out loud. if (run && !isAbsolute(run.stateDir)) { lines.push( ` ⚠️ "run.stateDir" is relative, so it resolves against the working directory (${process.cwd()}). ` + 'Make it absolute, or drop it and let it follow the data directory.', ) } return lines}
/** What `radiald run` prints for each loaded profile: the identity it is actually running as, which * is the fact a shared `sessions.json` used to get wrong in silence. */export function describeActors(actors: ActorRegistry): string[] { const width = Math.max(...actors.all.map((actor) => actor.profile.length), 0) return actors.all.map( (actor) => ` ${actor.profile.padEnd(width)} ${actor.did} ${actor.session.handle} ${actor.harness}`, )}
/** The empty-registry error, naming the file that was actually read and what the config expected to * find in it — "run radiald init" is no help when the question is *which* sessions file. */export function emptyRegistryError(instance: Instance): string { const configured = Object.keys(instance.config.agents).join(', ') || 'none' return ( `No initialized agent profiles found for ${instance.configPath} in ${instance.sessions.path} ` + `(config declares: ${configured}). Run: radiald init --config ${instance.configPath}` )}
/** * `radiald device` — the signing keys this instance's identities use in a private space (§18). * * The daemon's keys are ordinary device keys, kept in the instance data directory beside its * sessions and published as `device` records in each identity's own public repo: * * - `list` — what this machine holds, and what state each key is in; * - `rotate` — a new key, published, old one retired. The fold never notices; this is hygiene, * not a response to a stolen key. */async function deviceCommand(args: string[]): Promise<void> { const instance = await requireInstance(args) const actors = await loadActors(instance.config, instance.sessions, { configPath: instance.configPath, warn: (message) => console.warn(`⚠️ ${message}`), }) if (actors.all.length === 0) throw new Error(emptyRegistryError(instance)) const keys = new FileDeviceKeyStore(instance.dataDir.path) const dids = [...new Set(actors.all.map((actor) => actor.did))].sort(compareCodePoints) const writerFor = (did: string): ActorClient => { const actor = actors.all.find((candidate) => candidate.did === did) if (!actor) throw new Error(`no initialized profile runs as ${did} (this instance runs: ${dids.join(', ')})`) return actor.client } const selected = (): string => { const asked = values(args, '--did')[0] if (asked) return asked if (dids.length === 1) return dids[0] as string throw new Error(`--did is required: this instance runs as ${dids.join(', ')}`) } const action = args[1]
if (action === 'list') { for (const did of dids) { const held = await keys.list(did) console.log(`${did}${held.length === 0 ? ' (no device keys on this machine)' : ''}`) for (const key of held) { console.log(` ${key.deviceKeyId}\t${key.kind}\t${deviceKeyStatus(key)}\t${key.createdAt}`) } } return }
if (action === 'publish') { // Every identity at once: a daemon that serves a private space needs all of them verifiable, and // publishing is idempotent (the rkey is derived from the key). for (const did of dids) { const result = await ensurePublishedDevice(keys, writerFor(did), did, { kind: 'node' }) console.log( `${did}\t${result.key.deviceKeyId}\t${result.created ? 'published' : 'already published'}\t${result.ref.uri}`, ) } return }
if (action === 'rotate') { const did = selected() const rotated = await rotateDeviceKey(keys, writerFor(did), did, { kind: 'node' }) console.log(`published ${rotated.created.deviceKeyId} at ${rotated.ref.uri}`) for (const key of rotated.retired) console.log(`retired ${key.deviceKeyId}`) for (const key of rotated.addressesRetired) console.log(`withdrew the published address of ${key}`) console.log('Rotation is invisible to the fold: everything the retired key signed keeps counting.') return }
if (action === 'retire') { const did = selected() const deviceKeyId = args[2] if (!deviceKeyId || deviceKeyId.startsWith('--')) { throw new Error('Usage: radiald device retire <deviceKeyId> [--did <did>]') } const result = await retireDeviceAddress(writerFor(did), deviceKeyId) console.log( result.changed ? `withdrew the published address of ${deviceKeyId}` : `${deviceKeyId} publishes no address, so nothing dials it already`, ) console.log('No replica will dial it or serve it once they have read this. Everything it signed') console.log('keeps counting: this is containment and not repair.') return }
throw new Error(`Unknown device command "${action ?? ''}" (list, publish, rotate, retire)\n${usage}`)}
async function runCommand(args: string[]): Promise<void> { const { instance, run } = await requireRunConfig(args) const { configPath, config } = instance // `run.spaces` may be empty in the file (the init scaffold writes `spaces: []` so the block it // emits is loadable); a daemon with nothing to poll is only an error here, where polling starts. if (run.spaces.length === 0 && run.privateSpaces.length === 0) { throw new Error(`${configPath} has an empty "run.spaces"; add the "at://" URI of the space to serve`) } const interval = Number(values(args, '--interval')[0] ?? 5_000) if (!Number.isFinite(interval) || interval < 100) throw new Error('--interval must be at least 100ms')
// A lease that can lapse between two ingestion cycles makes a daemon drop claims it is working // on — a startup error, not a warning, because every claim it then writes is one it will lose. const claimAdvisory = claimTimingAdvisory(run.claims, interval) if (claimAdvisory) throw new Error(claimAdvisory)
console.log( `radiald starting: ${run.spaces.length} space(s)` + `${run.privateSpaces.length > 0 ? ` and ${run.privateSpaces.length} private space(s)` : ''}, ` + `interval ${interval}ms, concurrency ${run.concurrency}, open egress`, ) for (const line of describeInstance(instance, run)) console.log(line) console.log( `claims: lease ${run.claims.leaseMs}ms, renew every ${run.claims.renewIntervalMs}ms, ` + `confirm after ${run.claims.confirmCycles} ingestion cycle(s), at most ${run.claims.maxOutstanding} outstanding, ` + `warning above ${clockSkewThresholdMs(run.claims.leaseMs)}ms of clock skew`, ) for (const spaceUri of run.spaces) console.log(` polling ${spaceUri}`)
if (run.turnTransport === 'tcp') { console.warn( '⚠️ turn transport = tcp (dev mode): turns reach the daemon over TCP on host.docker.internal, ' + 'this is a development transport; do not expose the daemon socket.', ) }
await mkdir(run.stateDir, { recursive: true }) await mkdir(join(run.stateDir, 'turns'), { recursive: true }) await mkdir(join(run.stateDir, 'checks'), { recursive: true }) await mkdir(join(run.stateDir, 'auto-review'), { recursive: true })
// One daemon per state directory. Two are two writers on `records.db`, `sync.db` and three // ledgers, and nothing downstream of here would notice. Released in the `finally` at the bottom. const stateLock = StateDirLock.acquire(run.stateDir) const instanceLabel = instanceId(run.stateDir)
try { const store = instance.sessions // Every PDS response already carries a `Date` header, and reading it is the only way this daemon // can detect a clock that runs BEHIND its peers — the direction record timestamps cannot // distinguish from ingestion lag, and the direction that quietly wins claim races. It rides on // requests that had to happen anyway and only ever logs. const clocks = new ClockSkewObserver() const probingFetch = observingFetch((input, init) => fetch(input, init), clocks) const actors = await loadActors(config, store, { fetcher: probingFetch, configPath, warn: (message) => console.warn(`⚠️ ${message}`), }) if (actors.all.length === 0) throw new Error(emptyRegistryError(instance)) console.log(`loaded ${actors.all.length} agent profile(s):`) for (const line of describeActors(actors)) console.log(line)
// Fail closed at startup rather than per dispatch: an unrunnable harness or a profile with no // provider credential is knowable now, and finding out per-request costs a ledger attempt and a // cooldown each time. Names only in the log — never a value. const codexAuth = run.codexAuth ? new CodexAuthStore(join(instance.dataDir.path, 'codex-auth')) : undefined await codexAuth?.validate() const modelEnv = resolveModelEnvironment(actors.all, run.modelEnv, process.env, { managedCodexAuth: Boolean(codexAuth), }) console.log(`model credentials forwarded to turns: ${Object.keys(modelEnv).join(', ') || '(none)'}`) if (codexAuth) console.log(`managed Codex auth: ${codexAuth.path} (serialized refreshable session)`)
const ledger = new TurnLedger(join(run.stateDir, 'ledger.db'), { retryBound: run.retryBound, cooldownMs: run.cooldownMs, }) const checkLedger = new CheckLedger(join(run.stateDir, 'check-ledger.db'), { retryBound: run.checkRetryBound, cooldownMs: run.checkCooldownMs, }) // Its own file: claims and turns are independent workstreams with independent lifecycles, and a // daemon that restarts mid-lease reads this to renew rather than claim twice. const claimLedger = new ClaimLedger(join(run.stateDir, 'claim-ledger.db')) const runner = new DockerRunner()
// Startup orphan reconciliation: any row still "running" belongs to a prior process that died // (or was killed) mid-turn — its container (if any survived) is killed and the row transitioned // to "crashed"/"gave_up" (attempts preserved) so a repeatedly-crashing daemon still respects the // retry bound instead of resetting it on every restart. See `reconcileOrphans`. await reconcileOrphans(runner, ledger) await reconcileCheckOrphans(runner, checkLedger)
// One adapter per configured forge, selected per project by `gitUrl` host. GitHub authentication // is resolved once but no longer gates the whole daemon: a tangled-only operator has no use for // it, and a space that holds both kinds of project must be able to serve both. const githubToken = await resolveGitHubToken() const keyStore = new TangledKeyStore(run.stateDir) const adapters: ForgeAdapter[] = [] for (const configured of run.forges) { if (configured.kind === 'github') { if (!githubToken) { console.warn( 'run.forge is "github" but GitHub authentication is unavailable; merge observation is disabled.', ) } adapters.push(new GitHubForge({ fetch, ...(githubToken ? { token: githubToken } : {}) })) if (githubToken) console.log('forge: github observation adapter enabled') continue } // The tangled adapter writes as the SAME agent identity that signs the artifact — no forge // account, no operator token in the middle. The implementation-producing actor is the one that // authors its pull records, chosen the same deterministic way the merge poller picks a writer. const writer = (): (typeof actors.all)[number] | undefined => [...actors.all].sort((a, b) => compareCodePoints(a.did, b.did)).find((actor) => actorEverAccepts(actor, 'implementation')) ?? [...actors.all].sort((a, b) => compareCodePoints(a.did, b.did))[0] const knownHosts = configured.knownHosts ?? [] // Pull state comes from Bobbin, tangled's public read-only appview, unless the operator turned // it off — see `bobbin.ts` for why an appview answers this better than a PDS scan can. const bobbin = configured.api === false ? undefined : new FetchBobbin({ fetch, ...(configured.api ? { url: configured.api } : {}) }) const tangled = new TangledForge({ reader: new FetchRepoTransport(), ...(bobbin ? { bobbin } : {}), pageFetch: configured.pageLinks === false ? false : fetch, ...(configured.hosts ? { hosts: configured.hosts } : {}), resolveHandle: (handle) => resolveHandleDid(handle, { resolveTxt: nodeResolveTxt }), actor: () => { const actor = writer() if (!actor) return undefined return { did: actor.did, createForeign: (collection, value) => actor.client.createForeign(collection, value), putForeign: (collection, uri, value, options) => actor.client.putForeign(collection, uri, value, options), uploadBlob: (bytes, contentType) => actor.client.uploadBlob(bytes, contentType), } }, // Push material is only assembled when a turn actually needs it, so a daemon that only // OBSERVES tangled pulls needs neither a key nor pinned host keys. push: async () => { const actor = writer() if (!actor || knownHosts.length === 0) return undefined const privateKey = await keyStore.read(actor.did) if (!privateKey) return undefined return { privateKey, knownHosts: knownHostsFile(knownHosts) } }, log: (message) => console.log(message), }) adapters.push(tangled) console.log(`forge: tangled adapter enabled for ${(configured.hosts ?? ['tangled.org']).join(', ')}`) if (knownHosts.length === 0) { console.warn( 'the tangled forge has no "knownHosts": implementation turns cannot push until you add ' + `ssh-keyscan output for its hosts to ${configPath}. Radial will not disable host-key checking.`, ) } } const forges = new ForgeRegistry(adapters) // Legacy single-adapter shape, still what the dispatcher's `forge` and the merge poller consume // where a URL alone has to be routed. Undefined when nothing can observe. const forge = adapters.some((adapter) => adapter.canObserve) ? routingForgeState(forges) : undefined const advisory = forgeAdvisory({ adapter: Boolean(forge), configured: run.forges.length > 0, producesImplementation: actors.all.some((actor) => actorEverAccepts(actor, 'implementation')), configPath, }) if (advisory) console.warn(advisory)
const boundRunTurn = (input: TurnInput) => runTurn(input, { runner, harnesses: { select: selectHarness }, modelEnv, ...(codexAuth ? { codexAuth } : {}), ...(githubToken ? { githubToken } : {}), forges, log: (message) => console.log(message), })
const dispatcher = new TurnDispatcher({ ledger, concurrency: run.concurrency, runDirFor: (requestUri) => turnRunDir(run.stateDir, requestUri), image: run.image, timeoutMs: run.timeoutMs, ...(run.network !== undefined ? { network: run.network } : {}), allowedSchemes: run.gitSchemes, ...(run.memory !== undefined ? { memory: run.memory } : {}), turnTransport: run.turnTransport, instance: instanceLabel, ...(forge ? { forge } : {}), forges, // The branch probe the no-forge reuse fallback leans on, carrying the daemon's token so it can // answer for a private repository too. branchExists: ({ gitUrl, branch }) => remoteBranchExists({ gitUrl, branch, allowedSchemes: run.gitSchemes, ...(githubToken ? { githubToken } : {}), }), implementationEnabled: !!githubToken, runTurn: boundRunTurn, ...(codexAuth ? { serializedHarnesses: ['codex'] } : {}), // Unassigned requests dispatch only for a claim this daemon wrote and has watched win. heldClaims: () => claimLedger.heldRequests(), kill: (label) => runner.kill(label), log: (message) => console.log(message), })
// Checks have an empty container environment. A GitHub token, when available, is used solely by // the daemon-host checkout helper for private repositories and is never added to the container. const boundRunCheckRun = (input: CheckRunInput) => runCheckRun(githubToken ? { ...input, githubToken } : input, { runner }) const checkDispatcher = new CheckDispatcher({ ledger: checkLedger, concurrency: run.checkConcurrency, ourDids: new Set(actors.all.map((actor) => actor.did)), runDirFor: (key) => checkRunDir(run.stateDir, key), image: run.checkImage, timeoutMs: run.checkTimeoutMs, allowedSchemes: run.gitSchemes, ...(run.memory !== undefined ? { memory: run.memory } : {}), instance: instanceLabel, runCheckRun: boundRunCheckRun, log: (message) => console.log(message), })
// Interrupts the loop's sleep. Declared before either live path is wired because both hold it: // Jetstream on the public side, gossip on the private one. It stays `undefined` until // `runDaemon` hands the real function over, and both callers are optional-chained for it. let wakeLoop: (() => void) | undefined
// Kept separately from the loop's list because Jetstream is a PUBLIC-path optimisation: it // subscribes to a firehose of PDS commits, and a private space has no PDS to hear from — its // live path IS its bus. const publicSpaces = run.spaces.map((spaceUri) => ({ spaceUri, runtime: new SpaceSyncRuntime( spaceUri, { records: join(run.stateDir, 'records.db'), checkpoints: join(run.stateDir, 'sync.db'), }, new FetchRepoTransport(undefined, probingFetch), ), })) const spaces: DaemonSpace[] = [...publicSpaces]
// The private half of the loop (design §18): one transport endpoint for this daemon, a replica // per configured space, and a registry per space whose writes are envelopes. Everything after // this point — dispatch, claims, checks, auto-review — is handed a `MaterializedIndex` and a // registry and cannot tell the two kinds of space apart, which is the whole design. const privateSpaces = run.privateSpaces.length > 0 ? await openPrivateSpaces({ spaces: run.privateSpaces, dataDir: instance.dataDir.path, actors, transport: new FetchRepoTransport(undefined, probingFetch), // The private half of Jetstream's `onRecord`: an envelope a peer gossiped is the only // notice this daemon gets that something happened in a private space, so it wakes the // loop the same way a firehose commit does. Without it, `streaming` would let the loop // slow down for a space nothing can speed back up. onArrival: () => wakeLoop?.(), log: (message) => console.log(message), warn: (message) => console.warn(`⚠️ ${message}`), }) : undefined if (privateSpaces) spaces.push(...privateSpaces.spaces)
// Both halves of skew detection, reported on the same fresh index the dispatchers see: the // symmetric `Date`-header probe above, and the one-sided "somebody wrote a record dated in our // future" heuristic. Advisory only — the fold's rules are durations and ingestion cycles, never a // comparison of two clocks (see docs/operators.md §3, "Claims and clocks"). const skew = new SkewWatch({ thresholdMs: clockSkewThresholdMs(run.claims.leaseMs), samples: () => clocks.samples(), log: (message) => console.warn(`⚠️ ${message}`), })
// Claim manager (design §11): the only thing that lets this daemon touch an OPEN (unassigned) // request. It writes a claim, waits to see it win the fold, and only then does `heldClaims` let // the dispatcher launch. Assigned requests never pass through it and gain no latency. const claimManager = new ClaimManager({ ledger: claimLedger, turns: ledger, timing: run.claims, now: () => Date.now(), capacity: () => run.concurrency - dispatcher.inFlight, abandon: (requestUri, reason) => dispatcher.abandon(requestUri, reason), log: (message) => console.log(message), })
// Jetstream (design §5), opt-in. It shortens ingestion latency and nothing else: polling stays // the authority, so an endpoint that is unreachable degrades to exactly the pre-Phase-7 behaviour. if (run.jetstream) { const endpoint = run.jetstream.endpoint console.log( `jetstream: subscribing to ${endpoint} (polling continues as the authority and backfill, ` + `every ${run.jetstream.backfillIntervalMs}ms while the stream is healthy)`, ) for (const { spaceUri, runtime } of publicSpaces) { runtime.startJetstream({ endpoint, degradedAfterMs: 2 * interval, onRecord: () => wakeLoop?.(), onStatus: (status, detail) => console.log(`jetstream ${spaceUri}: ${status}${detail ? ` (${detail})` : ''}`), log: (message) => console.log(message), }) } }
// Merge-observation poller (design §10): the GitHub adapter's `getPullRequestState` structurally // satisfies MergePoller's `ForgeStateSource`, so the same `forge` drives it. Only runs when the // forge is configured — no forge, no polling. const mergePoller = forge ? new MergePoller({ adapter: forge, now: () => Date.now(), pollIntervalMs: run.mergePollIntervalMs, backoffMaxMs: run.mergePollBackoffMaxMs, log: (message) => console.log(message), }) : undefined
// Auto-review trigger (design §7): writes a `type: "review"` request the moment an artifact whose // effective auto-review config is on lands. No forge dependency — it runs whenever actors exist; // its observed-subjects ledger lives under `<stateDir>/auto-review/`. const autoReviewTrigger = new AutoReviewTrigger({ stateDir: run.stateDir, now: () => new Date().toISOString(), log: (message) => console.log(message), })
// Where this daemon can be reached, kept current (ADR §21). A bound endpoint's id never changes; // the relays it is reachable through do — a laptop changing network, a VPN coming up, a failover // — and a hint that went stale at startup is indistinguishable, to every peer, from a daemon // that is simply offline. The check is local and free: it only touches a PDS when the answer // moved. Cleared in the shutdown block below, which is the only thing keeping it from outliving // the loop it belongs to. const addressRefresh = privateSpaces ? setInterval(() => { void privateSpaces.refreshAddresses().catch((error: unknown) => { console.warn( `⚠️ could not republish this daemon's private endpoint address: ` + `${error instanceof Error ? error.message : String(error)}`, ) }) // On the same interval and for the same shape of reason: what a private space knows about // this daemon is only what this daemon has sealed into it. An address that went stale and // a capability record that never landed are both "the replica disagrees with the process", // and both are a local read that touches nothing when the answer has not moved. void privateSpaces.refreshCapabilities().catch((error: unknown) => { console.warn( `⚠️ could not republish agent capabilities into a private space: ` + `${error instanceof Error ? error.message : String(error)}`, ) }) }, run.privateAddressRefreshMs) : undefined
const controller = new AbortController() const stop = (): void => controller.abort() process.on('SIGTERM', stop) process.on('SIGINT', stop)
try { await runDaemon({ spaces, dispatcher, checkDispatcher, ...(mergePoller ? { mergePoller } : {}), autoReviewTrigger, claimManager, skew, actors, // While every space's stream is healthy, polling drops to the backfill rate; the moment one // is not, the short interval is back. Read afresh each tick, so it tracks the stream live. intervalMs: run.jetstream ? () => spaces.every(({ runtime }) => runtime.streaming) ? (run.jetstream as { backfillIntervalMs: number }).backfillIntervalMs : interval : interval, onWake: (wake) => { wakeLoop = wake }, // The registry lives in the space, so this is the first moment the daemon can tell an operator // that a type nothing it runs accepts has been sitting in it. onFirstIndex: (spaceUri, index, spaceActors) => { for (const advisory of coverageAdvisories(index, spaceActors, { spaceUri, configPath })) { console.warn(`⚠️ ${advisory}`) } }, signal: controller.signal, log: (message) => console.error(message), }) } finally { // Graceful shutdown: kill every in-flight container, then wait for its turn promise to // settle (each already never rejects — see TurnDispatcher.pump) so the ledger reflects a // clean "crashed" row rather than leaving it stuck at "running" for next startup's orphan // reconciliation to redo. Every step here is best-effort — shutdown must not hang. for (const label of [...dispatcher.runningLabels(), ...checkDispatcher.runningLabels()]) { await runner.kill(label).catch(() => {}) } await dispatcher.drain().catch(() => {}) await checkDispatcher.drain().catch(() => {}) await autoReviewTrigger.drain().catch(() => {}) await claimManager.drain().catch(() => {}) if (addressRefresh) clearInterval(addressRefresh) // The endpoint before the stores it serves from: a peer mid-catch-up gets a closed connection // rather than a responder reading a closed database. await privateSpaces?.endpoint.close().catch(() => {}) for (const { runtime } of spaces) runtime.close() ledger.close() checkLedger.close() claimLedger.close() } } finally { stateLock.release() }}
export async function main(args = argv.slice(2)): Promise<void> { if (args[0] === 'config' && args[1] === 'init') { await configInit(args.slice(2)) return } if (args[0] === 'run') { await runCommand(args.slice(1)) return } if (args[0] === 'codex' && args[1] === 'login') { await codexLoginCommand(args.slice(2)) return } if (args[0] === 'turn' && args[1] === 'list') { await turnListCommand(args.slice(2)) return } if (args[0] === 'turn' && args[1] === 'reset') { await turnResetCommand(args.slice(2)) return } if (args[0] === 'check' && args[1] === 'reset') { await checkResetCommand(args.slice(2)) return } if (args[0] === 'index' && args[1] === 'reset') { await indexResetCommand(args.slice(2)) return } if (args[0] === 'sync' && args[1]) { const paths = { records: one(args, '--records'), checkpoints: one(args, '--checkpoints'), } const interval = Number(values(args, '--interval')[0] ?? 5_000) if (!Number.isFinite(interval) || interval < 100) throw new Error('--interval must be at least 100ms') await runSyncLoop({ watch: args.includes('--watch'), intervalMs: interval, sync: () => syncSpaceOnce(args[1] as string, paths), onIndex: (index) => console.log(JSON.stringify(indexSummary(index))), onError: (error) => console.error(error instanceof Error ? error.message : error), }) return } if (args[0] === 'index' && args[1]) { const index = readIndex(args[1], one(args, '--records')) // `--digest` is the operator half of divergence detection (design §12): two operators print it // and compare. Equal means both folds decided the same thing; different means one of them has // not ingested something, and the section that differs says which part of the fold to look at. if (args.includes('--digest')) { const digest = indexDigest(index) // `clocks` rides along because two operators comparing digests are usually comparing them at // a moment when one of them is wrong about something, and "whose clock is ahead of mine" is a // question the digest cannot answer: it fingerprints decisions, and a lease is deliberately // not one of them. This is the one-sided half — a record dated in OUR future — so an empty // list means "nobody is visibly ahead of us", never "our clock is right". const clocks = futureDatedAuthors(index, new Date().toISOString()) console.log(JSON.stringify({ space: index.space.uri, ...digest, clocks })) return } console.log(JSON.stringify(indexSummary(index))) return } if (args[0] === 'device') { await deviceCommand(args) return } if (args[0] === 'claim' && args[1] === 'reset') { await claimResetCommand(args.slice(2)) return } if (args[0] !== 'init') throw new Error(usage) const replaceIdentity = args.includes('--replace-identity') if (args.includes('--identifier')) { if (!args[1] || !args.includes('--password-stdin')) throw new Error(usage) // The flag-driven path has no config to read a `dataDir` from, but `--data-dir` (and // `$RADIAL_DATA_DIR`) still decide where the session lands — it used to take the process default // silently, which is the one place an instance could write to the wrong sessions file. const store = new FileSessionStore(resolveDataDir({ flag: values(args, '--data-dir')[0] }).path) const result = await initializeAgent({ profile: args[1], service: one(args, '--pds'), identifier: one(args, '--identifier'), password: await readStdin(), handleName: one(args, '--name'), // Same refusal the config path gets in `resolveAgentInits`: never publish an agent record // advertising a harness this daemon could not launch. harness: selectHarness(one(args, '--harness')).name, models: values(args, '--model').map(parseModel), artifactTypes: values(args, '--artifact-type'), }, { store, update: args.includes('--update'), replaceIdentity }) console.log(JSON.stringify(result)) console.log(`sessions: ${store.path}`) await ensureGitHubAuth({ skip: args.includes('--skip-github-auth') }) return } const instance = await requireInstance(args) const { configPath: path, config } = instance // Which forges this operator serves decides what `init` has to set up. A tangled forge needs an // ed25519 push key published under the agent DID; GitHub needs `gh` to be logged in. An operator // who serves only one should not be walked through the other's login flow. const forgeKinds = new Set(configuredForges(config.run).map((forge) => forge.kind)) const wantsGitHub = forgeKinds.size === 0 || forgeKinds.has('github') // The key belongs to the instance being initialized, so it follows the resolved data directory // rather than the process-wide default — the same rule `radiald run` uses for `run.stateDir`. const tangledKeys = forgeKinds.has('tangled') ? new TangledKeyStore(config.run?.stateDir ?? join(instance.dataDir.path, 'run')) : undefined const profiles = positional(args.slice(1)) const byIdentifier: Record<string, string> = {} for (const identity of extraIdentities(config, profiles)) { const password = process.env[passwordEnvName(identity)] if (password) byIdentifier[identity] = password } const inits = resolveAgentInits(config, profiles, { default: (args.includes('--password-stdin') ? await readStdin() : process.env.RADIAL_PASSWORD) ?? '', byIdentifier, }) let failed = false const update = args.includes('--update') const width = Math.max(...inits.map((init) => init.profile.length), 0) for (const init of inits) { try { const result = await initializeAgent(init, { store: instance.sessions, update, replaceIdentity, ...(tangledKeys ? { tangledKeys } : {}), }) // The published artifact types, not just the DID: this line is the only place an operator sees // what the space will offer this profile for, and a profile that has not caught up with a // registry type is invisible otherwise. const republished = result.action === 'updated' ? ' (republished)' : '' console.log( `✓ ${init.profile.padEnd(width)} ${result.did} ${result.artifactTypes.join(', ')}${republished}`, ) // And where that line does not apply. A scope is an override, so an operator who cannot see it // would read the line above as the answer everywhere — which is the one thing it is not. for (const scope of result.scopes) { console.log( ` ${' '.repeat(width)} in ${scope.space}: ${scope.artifactTypes.join(', ') || '(nothing)'}`, ) } // The one step Radial cannot take for the operator: a knot authorises a push against the // agent's DID, and only a repo owner can grant that. Say it exactly once per key. if (result.tangledKey && (result.tangledKey.created || result.tangledKey.published)) { console.log( ` tangled push key ${result.tangledKey.created ? 'generated' : 'adopted'} at ${result.tangledKey.path}` + `${result.tangledKey.published ? ' and published as an sh.tangled.publicKey record' : ''}`, ) console.log( ` next: add ${result.handle} (${result.did}) as a collaborator on every tangled repository this agent implements in`, ) } } catch (error) { failed = true console.error(`✗ ${init.profile.padEnd(width)} ${error instanceof Error ? error.message : error}`) } } // Said once, after the per-profile lines: which file these identities were written to is the fact // that decides whether `radiald run` picks them up, and it is invisible from the command typed. console.log(`sessions: ${instance.sessions.path} (${describeDataDirSource(instance.dataDir)})`) if (failed) process.exitCode = 1 else if (wantsGitHub) await ensureGitHubAuth({ skip: args.includes('--skip-github-auth') })}
const entry = argv[1]if (entry && import.meta.url === pathToFileURL(realpathSync(entry)).href) { main().catch((error: unknown) => { console.error(error instanceof Error ? error.message : error) process.exitCode = 1 })}