Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356import { CredentialClient, FetchRepoTransport, createSession, resolveIdentityPds, type ActorSession, type FetchLike, type ForeignRecordReader, type TxtResolver,} from '@radial/atproto'import { FileSessionStore, nodeResolveTxt, type StoredActor } from '@radial/atproto/node'import { COLLECTIONS, compareCodePoints, type AgentModel, type AgentRecord, type AgentScope, type StrongRef,} from '@radial/core'import type { TangledKeyStore } from './forge-auth.js'import { TANGLED_PUBLIC_KEY } from './forge-tangled.js'
export interface AgentInitInput { profile: string /** Omit to resolve the identity's PDS from its DID document. */ service?: string | undefined identifier: string password: string handleName: string harness: string models: AgentModel[] artifactTypes: string[] /** * Per-space overrides to publish alongside the global list (design §11). * * PUBLIC spaces only — `resolveAgentInits` drops the private ones, and not because a private * space's URI is a secret (a member's `join` bookmark is public and carries it). It is that the * public record is a source the private fold ignores by construction: an entry here would be * permanently unread and permanently unable to look anything but stale. */ scopes?: AgentScope[] now?: string}
export interface AgentInitResult { did: string handle: string service: string profile: StrongRef /** What the published record now says this profile accepts — what an assignee menu will offer. */ artifactTypes: string[] /** The per-space overrides that record carries, if any: where the line above does not apply. */ scopes: AgentScope[] /** What happened to the agent record: nothing, a first publish, or a republish under `--update`. */ action: 'created' | 'unchanged' | 'updated' /** Present when a tangled forge is configured: the agent's push key and whether this run created * or published it. The operator still has to grant the DID collaborator access on each repo — * `radiald init` prints that, because nothing else can do it for them. */ tangledKey?: { path: string; publicKey: string; created: boolean; published: boolean }}
/** * Publish the agent's ed25519 push key as an `sh.tangled.publicKey` record, once. * * Idempotency is by CONTENT, not by rkey: the lexicon keys these by tid, so there is no * deterministic rkey to collide on, and the records are public — a plain `listRecords` over the * agent's own repo answers "is this key already published" without any extra credential. */export async function publishTangledKey(input: { did: string publicKey: string name: string createForeign: (collection: string, value: unknown) => Promise<StrongRef> reader: ForeignRecordReader now?: string}): Promise<{ published: boolean; ref?: StrongRef }> { const listed = await input.reader.listForeignRecords({ did: input.did, collection: TANGLED_PUBLIC_KEY, limit: 100, }) const fingerprint = (line: string): string => line.trim().split(/\s+/).slice(0, 2).join(' ') const already = listed.records.some((record) => { const value = record.value if (typeof value !== 'object' || value === null) return false const key = (value as { key?: unknown }).key return typeof key === 'string' && fingerprint(key) === fingerprint(input.publicKey) }) if (already) return { published: false } const ref = await input.createForeign(TANGLED_PUBLIC_KEY, { $type: TANGLED_PUBLIC_KEY, key: input.publicKey, name: input.name, createdAt: input.now ?? new Date().toISOString(), }) return { published: true, ref }}
/** * How a published agent record and the config that would republish it differ, in the operator's * terms. The whole point of naming the fields is that "already exists with different capabilities" * is true of a changed model list and of a profile that has not caught up with a type the space * registry gained — and only the second one is the mistake anybody actually makes. */export function capabilityDrift(published: AgentRecord, configured: AgentRecord): string[] { const lines: string[] = [] const list = (values: readonly string[]): string => (values.length > 0 ? values.join(', ') : '(none)') const publishedTypes = new Set(published.artifactTypes) const configuredTypes = new Set(configured.artifactTypes) if (list(published.artifactTypes) !== list(configured.artifactTypes)) { const added = configured.artifactTypes.filter((type) => !publishedTypes.has(type)) const dropped = published.artifactTypes.filter((type) => !configuredTypes.has(type)) const change = [ ...(added.length > 0 ? [`adds ${list(added)}`] : []), ...(dropped.length > 0 ? [`drops ${list(dropped)}`] : []), ] lines.push( `artifactTypes: published ${list(published.artifactTypes)}, config ${list(configured.artifactTypes)}` + (change.length > 0 ? ` (${change.join(', ')})` : ''), ) } if (published.handleName !== configured.handleName) { lines.push(`name: published ${published.handleName}, config ${configured.handleName}`) } if (published.harness !== configured.harness) { lines.push(`harness: published ${published.harness}, config ${configured.harness}`) } const models = (record: AgentRecord): string => list(record.models.map((model) => (model.costHint ? `${model.name}=${model.costHint}` : model.name))) if (models(published) !== models(configured)) { lines.push(`models: published ${models(published)}, config ${models(configured)}`) } // Per-space overrides, named one space at a time. The same reasoning as the global list: "the // scopes changed" is true of a space that gained a type and of a space that was dropped from the // config entirely, and the operator's next move differs. const scopesOf = (record: AgentRecord): Map<string, string[]> => new Map((record.scopes ?? []).map((scope) => [scope.space, scope.artifactTypes])) const publishedScopes = scopesOf(published) const configuredScopes = scopesOf(configured) const spaces = [...new Set([...publishedScopes.keys(), ...configuredScopes.keys()])] for (const space of spaces.sort(compareCodePoints)) { const before = publishedScopes.get(space) const after = configuredScopes.get(space) if (before && after && list([...before].sort()) === list([...after].sort())) continue if (!before) { lines.push(`space ${space}: config scopes it to ${list(after as string[])}, published record does not scope it`) continue } if (!after) { lines.push(`space ${space}: published as ${list(before)}, config no longer scopes it`) continue } const had = new Set(before) const has = new Set(after) const change = [ ...(after.some((type) => !had.has(type)) ? [`adds ${list(after.filter((type) => !had.has(type)))}`] : []), ...(before.some((type) => !has.has(type)) ? [`drops ${list(before.filter((type) => !has.has(type)))}`] : []), ] lines.push( `space ${space}: published ${list(before)}, config ${list(after)}` + (change.length > 0 ? ` (${change.join(', ')})` : ''), ) } return lines}
/** * Sorted by space URI, and each entry's types sorted within it. * * Not cosmetic: `agentRecordComparable` decides whether `--update` republishes at all, and the * private reconcile makes the same comparison every tick. An unstable order there is a record * rewritten forever, with every peer keeping every revision of it. */const sortScopes = (scopes: readonly AgentScope[]): AgentScope[] => [...scopes] .map((scope) => ({ space: scope.space, artifactTypes: [...scope.artifactTypes].sort(compareCodePoints) })) .sort((a, b) => compareCodePoints(a.space, b.space))
/** * The fields that decide whether a published agent record still says what this config says, as one * comparable string. `createdAt` is deliberately out: it is stamped afresh on every write, so * comparing it would republish forever — which on the private path means an envelope per tick, kept * by every peer. */export function agentRecordComparable(value: AgentRecord): string { return JSON.stringify({ handleName: value.handleName, harness: value.harness, models: value.models, artifactTypes: value.artifactTypes, // Sorted on both sides, so a record published by an older build (or with its spaces written in // another order) compares equal when nothing actually changed. scopes: sortScopes(value.scopes ?? []), })}
/** What `buildAgentRecord` actually reads — the private reconcile has no session to authenticate. */export type AgentRecordInput = Pick< AgentInitInput, 'handleName' | 'harness' | 'models' | 'artifactTypes'> & { scopes?: AgentScope[] | undefined; now?: string | undefined }
export function buildAgentRecord(input: AgentRecordInput): AgentRecord { return { $type: COLLECTIONS.agent, handleName: input.handleName, harness: input.harness, models: input.models, artifactTypes: input.artifactTypes, // Omitted rather than emitted empty: a profile that scopes nothing must publish the record an // unscoped profile always published, or every existing agent looks changed the first time an // operator runs a build that knows about scoping. ...(input.scopes && input.scopes.length > 0 ? { scopes: sortScopes(input.scopes) } : {}), createdAt: input.now ?? new Date().toISOString(), }}
export async function initializeAgent( input: AgentInitInput, options: { store?: FileSessionStore fetcher?: FetchLike resolveTxt?: TxtResolver /** * Republish an existing profile whose capabilities have changed, rather than refusing. Off by * default: the refusal is what stops a second config, or a stale checkout, from quietly * rewriting what a running agent advertises. `--update` is the operator saying this config is * the one that is right — which is exactly what adding a registry type to a profile is. */ update?: boolean /** * Where the agent's tangled push key lives. Set only when a tangled forge is configured: this * ensures the key exists (adopting an existing one — regenerating would silently invalidate * every collaborator grant already given) and publishes its public half as an * `sh.tangled.publicKey` record under the agent DID. */ tangledKeys?: TangledKeyStore /** Foreign-record reads for the key-publication check. Defaults to a `FetchRepoTransport`. */ reader?: ForeignRecordReader /** * Overwrite a stored profile that belongs to a DIFFERENT DID, rather than refusing. * * The refusal it turns off is the whole point of the guard: `sessions.json` is keyed by profile * name, so a second instance sharing a data directory and reusing a profile name silently * replaces the first one's identity — and the other daemon then runs under the wrong DID with no * error. Re-authenticating the SAME DID is untouched; that is the ordinary session refresh. */ replaceIdentity?: boolean } = {},): Promise<AgentInitResult> { const store = options.store ?? new FileSessionStore() const service = input.service ?? String( await resolveIdentityPds(input.identifier, { ...(options.fetcher ? { fetcher: options.fetcher } : {}), // The daemon is a Node process, so it always gets the DNS TXT method; tests inject their // own resolver instead. resolveTxt: options.resolveTxt ?? nodeResolveTxt, }), ) const session = await createSession(service, input.identifier, input.password, options.fetcher) if (input.identifier.startsWith('did:') && input.identifier !== session.did) { throw new Error(`Authenticated DID ${session.did} does not match ${input.identifier}`) } // Before ANY record is written: a stored profile that belongs to another DID is an instance about // to be clobbered, and the only symptom afterwards is the other daemon running as somebody else. const registered = await store.get(input.profile) if (registered && registered.did !== session.did && !options.replaceIdentity) { throw new Error( `Profile "${input.profile}" in ${store.path} is already registered to ${registered.did}; ` + `this init authenticated as ${session.did}. Give this instance its own data directory ` + '(--data-dir, or "dataDir" in radial.json), or pass --replace-identity to overwrite it.', ) }
let pending = session const client = new CredentialClient(session, options.fetcher, async (rotated) => { pending = rotated }) // One agent record per profile, keyed by profile name: an operator commonly runs several // profiles (differing harness/models/artifactTypes) under a single agent DID. const existing = await client.getOwnRecord(COLLECTIONS.agent, input.profile) const record = buildAgentRecord(input) let reference: StrongRef let action: AgentInitResult['action'] if (existing) { // Scopes are in the comparison (`agentRecordComparable`): without them `--update` would // republish the global list and silently leave the per-space overrides behind. if (agentRecordComparable(existing.value as AgentRecord) !== agentRecordComparable(record)) { const drift = capabilityDrift(existing.value as AgentRecord, record) if (!options.update) { // The published record is what an assignee menu reads, so a config that has gained a type // and a record that has not is the drift an operator hits and cannot see. Say which fields // moved and how to publish them; the guard itself stays. throw new Error( `Agent profile ${input.profile} is already published with different capabilities:\n` + drift.map((line) => ` ${line}`).join('\n') + `\n Re-run with --update to republish it from this config.`, ) } // Same rkey, so the profile keeps its identity and every request already assigned to it still // resolves; `swapRecord` makes this a no-op rather than a clobber if it moved underneath us. reference = await client.put(COLLECTIONS.agent, existing.uri, record, { swapRecord: existing.cid, }) action = 'updated' } else { reference = { uri: existing.uri, cid: existing.cid } action = 'unchanged' } } else { reference = await client.create(COLLECTIONS.agent, record, { rkey: input.profile }) action = 'created' } const stored: StoredActor = { ...(pending as ActorSession), profile: input.profile, agentProfile: reference, } await store.save(stored)
// The tangled push key, when this operator serves a tangled forge. It is per agent DID rather // than per profile: profiles share an identity, and a knot authorises a push against the DID. let tangledKey: AgentInitResult['tangledKey'] if (options.tangledKeys) { const comment = `radial ${stored.handle} (${stored.did})` const { pair, created } = await options.tangledKeys.ensure(stored.did, comment) const { published } = await publishTangledKey({ did: stored.did, publicKey: pair.publicKey, name: comment, createForeign: (collection, value) => client.createForeign(collection, value), reader: options.reader ?? new FetchRepoTransport(), ...(input.now ? { now: input.now } : {}), }) tangledKey = { path: options.tangledKeys.pathFor(stored.did), publicKey: pair.publicKey, created, published, } }
return { did: stored.did, handle: stored.handle, service: stored.service, profile: reference, artifactTypes: record.artifactTypes, scopes: record.scopes ?? [], action, ...(tangledKey ? { tangledKey } : {}), }}