Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951import { join } from 'node:path'import { activeGoals, activeProjects, buildTurnBundle, COLLECTIONS, compareCodePoints, type ArtifactRecord, type ArtifactRequestRecord, type ArtifactTypeRecord, type GoalView, type IndexedRecord, type MaterializedIndex, type MessageRecord, type ProjectView, type StrongRef, type TurnBundle,} from '@radial/core'import type { ActorRegistry, LoadedActor } from './actors.js'import { remoteBranchExists } from './bundle-writer.js'import { ForgeRegistry, type ForgeAdapter } from './forge.js'import { urlOnlyForgeAdapters } from './forge-github.js'import type { TurnLedger } from './ledger.js'import { implBranchName, planArtifactRkey, type TurnRequestAnchor } from './turn-socket.js'import { ANSWER_TYPE, ANSWER_TYPE_NAME, IMPLEMENTATION_TYPE, REVIEW_TYPE, REVIEW_TYPE_NAME, turnContainerLabel, type TurnInput, type TurnResult } from './turn.js'
/** A resolved v2 predecessor: the exact prior artifact, plus the branch/PR links an implementation * artifact carries (empty for every other type), used to decide branch reuse and to stamp `prev`. */export interface ArtifactPredecessor { ref: StrongRef branch?: string prUrl?: string /** True when a trusted `merge` annotation names this exact version. It is the index's answer to * "is that pull request finished with", and the only signal the reuse decision has when no forge * adapter is configured. Eventually consistent: the merge poller may not have written the record * yet, which is why a configured forge always outranks it. */ merged?: boolean}
export interface PredecessorResolution { /** Every same-type same-target artifact among the request's DIRECT resolved basedOn, ordered by * uri. Empty for a fresh v1; one for a v2; more than one is the ambiguous human error. */ candidates: ArtifactPredecessor[] /** The single predecessor to stamp as `prev` — present iff there is exactly one candidate. */ artifact?: ArtifactPredecessor /** >1 same-type same-target resolved predecessors: the request is a human error — reject it. */ ambiguous: boolean}
export interface Dispatchable { /** The owning view — a GoalView for a goal-scoped request, a ProjectView for a project-scoped * (system-artifact) request. Discriminate by request.value.goal/project presence when needed. */ target: GoalView | ProjectView request: IndexedRecord<ArtifactRequestRecord> actor: LoadedActor artifactType: ArtifactTypeRecord bundle: TurnBundle /** The version-chain predecessor for this request, resolved for EVERY artifact-producing type and * both scopes. Absent only for `review` requests, which produce a verdict, not a chained artifact. */ predecessor?: PredecessorResolution /** Present only for `review` requests: the subject under review and, when the subject links an * exact commit, the commit the read-only checkout is pinned to. */ review?: { subject: StrongRef; commit?: string } /** Present only for `answer` requests: the message being answered, its author (which becomes the * reply's `mentions`), the thread root the reply hangs under, and the subject's own anchor — which * the reply takes, because a reply belongs to the same target as what it answers even when that is * an artifact under the request's goal. */ answer?: { subject: StrongRef subjectAuthor: string root?: StrongRef anchor: { goal: StrongRef } | { artifact: StrongRef } }}
/** Forge URL reasoning with no credential behind it, for a project no configured adapter matches. * It is what keeps the reuse guard's refusals identical to the pre-registry ones. */const URL_ONLY_FORGES = new ForgeRegistry(urlOnlyForgeAdapters())
const refKey = (reference: StrongRef): string => `${reference.uri}#${reference.cid}`const targetUriFor = (artifact: IndexedRecord<ArtifactRecord>): string | undefined => artifact.value.goal?.uri ?? artifact.value.project?.uri
/** * Mirror of the materializer's awaiting-input rule (materializer.ts computeAwaitingInput) but keyed * on a specific actor DID rather than the request's assignee — so it also covers unassigned requests * the materializer excludes. True iff `actorDid` posted a trusted, non-declining question `re` this * exact request for which no OTHER member has since replied (a message whose `parent` is that * question, authored by a different DID). */export function hasUnansweredQuestion( messages: Array<IndexedRecord<MessageRecord>>, request: IndexedRecord<ArtifactRequestRecord>, actorDid: string,): boolean { const requestKey = `${request.uri}#${request.cid}` const questions = messages.filter( (m) => m.did === actorDid && m.value.re && refKey(m.value.re) === requestKey && m.value.declines !== true && // A reply is not a question — see computeAwaitingInput. An answer turn's terminal record is a // message carrying `re` too, and only `parent` tells the two apart. m.value.parent === undefined, ) return questions.some((question) => { const questionKey = `${question.uri}#${question.cid}` return !messages.some( (answer) => answer.value.parent && refKey(answer.value.parent) === questionKey && answer.did !== question.did, ) })}
/** * Resolve the v2 predecessor(s) for a request: among the request's DIRECT basedOn refs, the * artifacts that resolve to the SAME `type` under the SAME target (a goal for a goal-scoped request, * a project for a project-scoped one). Zero → a fresh v1 (no prev). One → the predecessor. More than * one → ambiguous (reject). An unresolved basedOn ref is NOT treated as a predecessor here (its type * is unknown) — it just means the request isn't dispatchable yet (the caller skips it as a * stale-index case) rather than a permanent reject. * * This is deliberately type-agnostic. Stamping `prev` only for implementations left a plan v2 or an * architecture v2 landing as a NEW chain, which breaks the version rail and, for project-scoped * types, makes `currentSystemArtifacts` list two rows for one document. */function resolvePredecessor( index: MaterializedIndex, request: IndexedRecord<ArtifactRequestRecord>, targetUri: string,): { resolution: PredecessorResolution; hasUnresolvedBasedOn: boolean } { const allArtifacts: Array<IndexedRecord<ArtifactRecord>> = [ ...index.goals.flatMap((view) => view.artifacts), ...index.projects.flatMap((view) => view.artifacts), ] const byRef = new Map(allArtifacts.map((artifact) => [refKey(artifact), artifact])) // Merged-ness of the candidate, carried alongside its links so `#launch` can decide reuse from // the index when no forge adapter is wired. Trusted merge annotations only (the materializer has // already filtered them); several daemons may each post an equivalent one, so this is a set. const merged = new Set<string>( [...index.goals, ...index.projects].flatMap((view) => view.merges.map((merge) => refKey(merge.value.artifact)), ), )
const candidates: ArtifactPredecessor[] = [] let hasUnresolvedBasedOn = false for (const reference of request.value.basedOn) { const found = byRef.get(refKey(reference)) if (!found) { hasUnresolvedBasedOn = true continue } if (found.value.type === request.value.type && targetUriFor(found) === targetUri) { candidates.push({ // Links are implementation-only in practice (every other type writes `links: {}`), so this // stays a no-op for them rather than needing a type check. ref: { uri: found.uri, cid: found.cid }, ...(found.value.links?.branch ? { branch: found.value.links.branch } : {}), ...(found.value.links?.pr ? { prUrl: found.value.links.pr } : {}), ...(merged.has(refKey(found)) ? { merged: true } : {}), }) } } candidates.sort((a, b) => compareCodePoints(a.ref.uri, b.ref.uri)) return { resolution: { candidates, ambiguous: candidates.length > 1, ...(candidates.length === 1 ? { artifact: candidates[0] } : {}), }, hasUnresolvedBasedOn, }}
/** * What a request's `type` resolves to, and — for a `review` or an `answer` — the subject it acts on. * Extracted from the dispatch gate because the CLAIM gate has to answer exactly the same question * before it writes a claim (a request nothing can dispatch is a request nothing should claim), and * two copies of the review/registry resolution would be two places for the rule to drift. */export interface ResolvedRequestType { artifactType: ArtifactTypeRecord review?: { subject: StrongRef; commit?: string } answer?: Dispatchable['answer']}
/** The lookups `resolveRequestType` needs, built once per index scan. */export interface RequestTypeContext { /** Registry types by name, FIRST-wins on a duplicate — matching the materializer's earliest-wins * `typeByName` (index.artifactTypes is sorted by (createdAt, uri)). A plain `new Map(entries)` * would be last-wins, and daemon and core must agree on which record a name binds. */ registry: Map<string, ArtifactTypeRecord> artifactByRef: Map<string, IndexedRecord<ArtifactRecord>> /** Space-global trusted messages by strong ref — what an `answer` request's subject resolves * against. Global for the same reason the awaiting-input scan is: the message being answered may * be anchored to an artifact in a different view than the request. */ messageByRef: Map<string, IndexedRecord<MessageRecord>>}
export function requestTypeContext(index: MaterializedIndex): RequestTypeContext { const registry = new Map<string, ArtifactTypeRecord>() for (const type of index.artifactTypes) { if (!registry.has(type.value.name)) registry.set(type.value.name, type.value) } const allArtifacts: Array<IndexedRecord<ArtifactRecord>> = [ ...index.goals.flatMap((view) => view.artifacts), ...index.projects.flatMap((view) => view.artifacts), ] const allMessages: Array<IndexedRecord<MessageRecord>> = [ ...index.goals.flatMap((view) => view.messages), ...index.projects.flatMap((view) => view.messages), ] return { registry, artifactByRef: new Map(allArtifacts.map((artifact) => [refKey(artifact), artifact])), messageByRef: new Map(allMessages.map((message) => [refKey(message), message])), }}
/** * Resolve the artifact type a request dispatches with. `review` is deliberately NOT registry data — * it never hits `registry.get`; it uses the synthetic built-in REVIEW_TYPE and resolves its subject * instead. A review request only materializes when its subject resolves under the same target * (materializer.ts), so subject resolution here always succeeds for a materialized review request * (the guards are defensive) — and a project-scoped subject makes the review dispatch from the * project loop, no separate branch needed. Returns undefined for a type this daemon cannot resolve; * only the scope MISMATCH is reported, because it is the one that can't happen. */export function resolveRequestType( request: IndexedRecord<ArtifactRequestRecord>, context: RequestTypeContext, expectedScope: 'goal' | 'project', onReject?: (requestUri: string, reason: string) => void,): ResolvedRequestType | undefined { const type = request.value.type if (type === ANSWER_TYPE_NAME) { // An answer request names one message version and asks for a reply to it. Like `review`, it is // NOT registry data — it uses the synthetic built-in ANSWER_TYPE. The materializer only // materializes one whose subject resolves under the request's own goal, so the lookups below // always succeed for a materialized request; the guards are defensive. const subject = request.value.subject if (!subject) return undefined const subjectMessage = context.messageByRef.get(refKey(subject)) if (!subjectMessage) return undefined // v1 requires an assignee: the fold closes an answer request on ITS ASSIGNEE's reply (keyed // exactly like a decline), so an unassigned one would be a request nothing could close. Open // answer requests generalize when claims land. if (!request.value.assignee) { onReject?.(request.uri, 'answer requests must name an assignee in v1') return undefined } // A message is anchored to a goal or to an artifact, and the fold files both into the goal's // thread (materializer.ts targetUriForMessage) — so a goal thread holds messages anchored to // its artifacts, and a reply to one of those belongs where IT is, not where the request is. // Same rule `message post --parent` enforces for a human. const subjectAnchor = subjectMessage.value.goal ? { goal: subjectMessage.value.goal } : subjectMessage.value.artifact ? { artifact: subjectMessage.value.artifact } : undefined if (!subjectAnchor) return undefined return { artifactType: ANSWER_TYPE, answer: { subject, subjectAuthor: subjectMessage.did, anchor: subjectAnchor, // The reply threads under the subject; the thread's root is the subject's own root, or the // subject itself when it starts the thread. Same derivation `message post --parent` uses. ...(subjectMessage.value.root ? { root: subjectMessage.value.root } : {}), }, } } if (type === REVIEW_TYPE_NAME) { const subject = request.value.subject if (!subject) return undefined const subjectArtifact = context.artifactByRef.get(refKey(subject)) if (!subjectArtifact) return undefined return { artifactType: REVIEW_TYPE, review: { subject, ...(subjectArtifact.value.links?.commit ? { commit: subjectArtifact.value.links.commit } : {}), }, } } const registered = context.registry.get(type) // No such registered type → silent skip (not reported). A registered type whose scope does not // match the owning view is a can't-happen (the materializer files a request only under a view of // its type's scope); log it defensively if it ever occurs rather than dispatch it cross-scope. if (!registered) return undefined if (registered.scope !== expectedScope) { onReject?.( request.uri, `type '${type}' is ${registered.scope}-scoped but the request is anchored to a ${expectedScope} — not dispatching`, ) return undefined } return { artifactType: registered }}
/** * The pure dispatch gate: which open requests are ready to hand to a container right now. Generalized * across scopes — goal-scoped requests iterate under goals, project-scoped (system-artifact) requests * under projects — and across types: a request is selected iff its `type` resolves to a registered * artifact type of the OWNING scope (or is a `review` whose subject resolves under the same target); * it is authored by any active space member (design §3: materializers don't police request * authorship — auto-review requests are agent-authored); it resolves to an actor that produces that * type (its named assignee, or — when unassigned — the agent whose claim WON the tie-break, which * must be one of ours and confirmed); it is not blocked on a question; and the ledger says it is * eligible. Every artifact-producing request carries predecessor data, in both scopes and for every * type; `review` requests do not (they produce a verdict, not a chained artifact). Ordered by uri. */export function selectDispatchable( index: MaterializedIndex, actors: ActorRegistry, ledger: TurnLedger, now?: string, /** Diagnostic hook: called for each request that resolves to a dispatchable type but is filtered * out, with the gate that rejected it. Requests whose type isn't a registered type of the owning * scope (nor a resolvable review) are skipped silently (never reported). */ onReject?: (requestUri: string, reason: string) => void, /** * Request URIs whose claim this daemon has written AND confirmed through ingestion — supplied by * `ClaimManager` (claims.ts), never re-derived here. Phase 7's gate for UNASSIGNED requests: with * two daemons watching one space, "pick the lowest capable DID" made both of them dispatch, so the * fold's `winningClaims` decides instead and this set is what says the winner is us and that we * have seen it be so. Defaults to empty, which is the safe direction: an operator who has not * wired a claim manager does no open-request work rather than racing another operator blindly. * Assigned requests never consult it. */ heldClaims: ReadonlySet<string> = new Set(),): Dispatchable[] { const typeContext = requestTypeContext(index) // Awaiting-input is checked against the SPACE-GLOBAL trusted messages: a project-scoped capture // turn's question is anchored to the source goal (a different view than the request's), so a // per-view message scan would miss it and the request would be redispatched over the open question. const allMessages: Array<IndexedRecord<MessageRecord>> = [ ...index.goals.flatMap((view) => view.messages), ...index.projects.flatMap((view) => view.messages), ]
// The per-request gate, target-agnostic: it only reads .openRequests (caller-iterated), .messages // (via allMessages), .awaitingInput and .target.uri — all present on both view types. `expectedScope` // is the owning view's scope; the implementation-predecessor branch is gated to goal scope. const evaluate = ( view: GoalView | ProjectView, request: IndexedRecord<ArtifactRequestRecord>, expectedScope: 'goal' | 'project', ): Dispatchable | undefined => { const type = request.value.type
const resolved = resolveRequestType(request, typeContext, expectedScope, onReject) if (!resolved) return undefined const { artifactType, review, answer } = resolved
// Any active member may author a request (human or agent): design §3/§7 — the daemon writes // auto-review requests under one of its agent identities, and materializers don't police request // authorship. The human-initiation norm is a UI convention, not a dispatch gate. const author = index.members.find((member) => member.did === request.did && member.active) if (!author) { onReject?.(request.uri, `author ${request.did} is not an active member of the space`) return undefined }
// Assigned → that agent handles it, with no claim and no extra latency (the whole // single-operator path). Unassigned → the fold's claim tie-break decides, and this daemon // dispatches only for the claim it wrote and has since watched win. const assignee = request.value.assignee let actor: LoadedActor | undefined if (assignee) { const assigneeMember = index.members.find( (member) => member.did === assignee && member.active && member.kind === 'agent', ) if (!assigneeMember) { onReject?.(request.uri, `assignee ${assignee} is not an active agent member of the space`) return undefined } if (!actors.byDid.has(assignee)) { onReject?.(request.uri, `no loaded actor for assignee ${assignee} (profile not initialized in this config?)`) return undefined } actor = actors.select(assignee, type, index.space.uri) if (!actor) { // "in this space" is the whole of the difference a scope makes: the profile may well // produce this type elsewhere, and an operator reading the rejection needs to know which // question was asked. onReject?.( request.uri, `loaded actor for ${assignee} does not produce '${type}' artifacts in this space`, ) return undefined } } else { // The claim gate (design §11). `winningClaims` is already keyed by bare request URI and // already excludes assigned, fulfilled and expired-lease claims (materializer.ts) — this is a // lookup, not a re-derivation, and every operator's fold computes it identically. const winner = view.winningClaims[request.uri] if (!winner) { onReject?.(request.uri, 'unassigned and unclaimed (the claim manager has not claimed it yet)') return undefined } if (!actors.byDid.has(winner.did)) { onReject?.(request.uri, `claim held by ${winner.did}`) return undefined } // The claim is ours in the fold, but the fold is not proof that WE know it: `heldClaims` is // the claim manager saying it wrote this claim and has since seen it win. Without that, two // daemons sharing an agent identity — or our own restart mid-write — could both start work. if (!heldClaims.has(request.uri)) { onReject?.(request.uri, 'claim written but not yet confirmed through ingestion') return undefined } actor = actors.select(winner.did, type, index.space.uri) if (!actor) { onReject?.( request.uri, `loaded actor for claim holder ${winner.did} does not produce '${type}' artifacts in this space`, ) return undefined } }
// Close the awaiting_input stale-index race: a ledger row can flip to `awaiting_input` (and become // `eligible()`) before the next index snapshot has ingested the SELECTED actor's question. Match // the question author against the selected actor's DID (NOT `assignee`, which is undefined for // unassigned requests). Scanned over allMessages so a capture turn's source-goal-anchored question // still counts. If the index hasn't ingested the actor's question yet, skip rather than redispatch // ahead of it; the `awaitingInput` check below still blocks an ingested but unanswered question. if (ledger.get(request.uri)?.state === 'awaiting_input') { const questionInIndex = allMessages.some( (m) => m.did === actor.did && m.value.re && m.value.re.uri === request.uri && m.value.re.cid === request.cid && m.value.declines !== true && m.value.parent === undefined, ) if (!questionInIndex) { onReject?.(request.uri, 'awaiting_input: selected actor question not yet ingested into the index (stale index)') return undefined } } if (view.awaitingInput.includes(request.uri)) { onReject?.(request.uri, 'blocked awaiting human input (unanswered agent question)') return undefined } // Retain the selected-actor scan for the stale index/ledger edge documented above. Once the // message is in the fold, `view.awaitingInput` is the space-global gate for assigned and open // requests alike, so every operator sees the same block. if (hasUnansweredQuestion(allMessages, request, actor.did)) { onReject?.(request.uri, 'blocked awaiting human input (unanswered question from the selected actor)') return undefined } if (!ledger.eligible(request.uri, now)) { onReject?.(request.uri, `ledger: not eligible (state ${ledger.get(request.uri)?.state ?? 'unknown'})`) return undefined }
// Resolve the v2 predecessor from the direct basedOn, for every artifact-producing type in both // scopes. An unresolved basedOn ref is a stale-index skip (retry later), never a permanent // reject: until it is ingested we cannot tell whether it is a same-type predecessor, and // dispatching regardless would stamp no `prev` and start a second chain for the same document. // Review requests are exempt — a verdict is not a chained artifact. let predecessor: PredecessorResolution | undefined if (type !== REVIEW_TYPE_NAME && type !== ANSWER_TYPE_NAME) { const { resolution, hasUnresolvedBasedOn } = resolvePredecessor(index, request, view.target.uri) if (hasUnresolvedBasedOn && !resolution.ambiguous) { onReject?.(request.uri, `'${type}' request has an unresolved basedOn ref (stale index)`) return undefined } predecessor = resolution }
let bundle: TurnBundle try { bundle = buildTurnBundle(index, request.uri, request.cid) } catch (error) { onReject?.(request.uri, `failed to build turn bundle: ${error instanceof Error ? error.message : error}`) return undefined }
// A project-scoped (capture) artifact request is COMMISSIONED to distil from its basedOn source; // dispatching before that source is ingested would launch a goal-less bundle that has lost its // commissioned input (the source goal is derived only from RESOLVED basedOn, and unresolved refs // are recorded in missingRefs). Skip until ingestion catches up — a stale-index skip, retried next // tick, never a permanent reject. Goal-scoped requests are exempt: basedOn is never required there // by design (the turn gets the goal's full context regardless). Review requests are exempt too: // their subject is guaranteed resolved by the materializer, and basedOn is provenance only. if ( expectedScope === 'project' && type !== REVIEW_TYPE_NAME && type !== ANSWER_TYPE_NAME && bundle.missingRefs.length > 0 ) { onReject?.(request.uri, 'project-scoped request has an unresolved basedOn ref (stale index)') return undefined }
// An answer turn is briefed from what the exchange was about — the artifacts the message hangs // off, carried in `basedOn` by whoever asked. A ref the index has not ingested yet would send it // in to answer a question about a document it cannot see, so it waits instead. A stale-index // skip, retried next tick; never a permanent reject. if (type === ANSWER_TYPE_NAME && bundle.missingRefs.length > 0) { onReject?.(request.uri, 'answer request has an unresolved ref (stale index)') return undefined }
return { target: view, request, actor, artifactType, bundle, ...(predecessor ? { predecessor } : {}), ...(review ? { review } : {}), ...(answer ? { answer } : {}), } }
const results: Dispatchable[] = [] // A shelved goal is set aside: agents never pick up its requests. `activeGoals` is the whole rule — // a goal any active member has ended (`GoalView.ended`, whichever way it was ended), and a goal // whose PROJECT the project author archived, which shelves everything under it. for (const goalView of activeGoals(index)) { for (const request of goalView.openRequests) { const dispatchable = evaluate(goalView, request, 'goal') if (dispatchable) results.push(dispatchable) } } // Project-scoped (system-artifact) requests: registry types with scope 'project' (adr, architecture) // and review requests whose subject is a project-scoped artifact. Same gate; never implementation. for (const projectView of activeProjects(index)) { for (const request of projectView.openRequests) { const dispatchable = evaluate(projectView, request, 'project') if (dispatchable) results.push(dispatchable) } } return results.sort((a, b) => compareCodePoints(a.request.uri, b.request.uri))}
export interface DispatcherDeps { ledger: TurnLedger concurrency: number now?: () => string runDirFor: (requestUri: string) => string image: string timeoutMs: number network?: string allowedSchemes: string[] /** Container memory limit (docker `--memory` syntax, e.g. `4g`), threaded into `TurnInput.memory`. */ memory?: string /** Threaded into `TurnInput.turnTransport`. Unset lets `turn.ts` default to `unix`. */ turnTransport?: 'unix' | 'tcp' /** This daemon instance's id (`instanceId(stateDir)`), mixed into every container label so two * daemons on one machine never derive the same docker name for one request. */ instance?: string /** Injected wrapper around turn.ts's runTurn that binds TurnDeps (runner/harness/secrets/etc). */ runTurn: (input: TurnInput) => Promise<TurnResult> /** Harnesses whose turns share mutable credentials and therefore must never overlap. */ serializedHarnesses?: readonly string[] /** The request URIs whose claim this daemon has confirmed, read fresh on every pump — supplied by * `ClaimManager.heldClaims`. Absent means "no claim manager", and unassigned requests are then * never dispatched (see `selectDispatchable`'s `heldClaims`). */ heldClaims?: () => ReadonlySet<string> /** Kills a container by label. Only `abandon` uses it — the ordinary shutdown path kills through * `runningLabels()` in cli.ts — and without it a lost claim can only stop the daemon from * dispatching AGAIN, not stop the turn already running. */ kill?: (label: string) => Promise<void> /** Optional forge adapter: used to VALIDATE a v2 predecessor's PR (open, head ref/repo + base all * matching this project) before reusing its branch. Without it, a v2 turn always opens a fresh * `radial/impl-<hash>` branch + new PR. Applies to every project; `forges` is the per-project * form and outranks nothing — this one wins where both are set, so an injected single adapter * (tests, a one-forge deployment) keeps behaving exactly as it did. */ forge?: Pick<ForgeAdapter, 'getPullRequestState'> /** The forges this daemon speaks, selected per project by `gitUrl` host. Where an adapter matches, * IT answers "is this pull request in this project's repo" and (when it can observe) "is it still * open"; where none does, the URL-only GitHub reasoning below keeps today's behaviour. */ forges?: ForgeRegistry /** Does a branch still exist on the project's remote? Defaults to bundle-writer's * `remoteBranchExists` (one `git ls-remote`, same credentials as the clone). It is what makes the * no-forge fallback below safe: without it, a predecessor whose branch was deleted after its PR * merged sends every retry of a v2 into the same failing clone. */ branchExists?: (input: { gitUrl: string; branch: string }) => Promise<boolean> /** Disable implementation launches when daemon GitHub possession is unavailable. Superseded, per * project, by the matching adapter's own `implementationBlockedReason()`: one missing GitHub * token used to stop implementations on projects that never needed one. */ implementationEnabled?: boolean log?: (message: string) => void}
/** * Bounded-concurrency, tracked dispatcher: `pump` launches eligible turns in the background and * returns immediately. Every launched promise is tracked (by request uri) and every settlement * updates the ledger before the promise is dropped — a turn's fate is never lost. */export class TurnDispatcher { readonly #deps: DispatcherDeps readonly #inFlight = new Map<string, { promise: Promise<void>; label: string }>() /** Predecessor branches currently spoken for, `branch → the request uri holding it`. Two v2 * requests off ONE predecessor resolve the same branch, and `#inFlight` is keyed by request uri, * so with `concurrency > 1` they would push to it at the same time. Reserved on the CANDIDATE * branch — before the async reuse guard runs — because the reservation has to be taken * synchronously to be a reservation at all. Over-serializing a pair whose reuse is later declined * costs one pump interval; interleaving two turns on one branch costs the branch. */ readonly #activeBranches = new Map<string, string>() /** Branch contention is exceptional enough to report once while the competing turn runs. */ readonly #contended = new Set<string>() readonly #activeSerializedHarnesses = new Set<string>()
constructor(deps: DispatcherDeps) { this.#deps = deps }
get inFlight(): number { return this.#inFlight.size }
/** True while a container for `requestUri` is still tracked here (dispatched, not yet settled). */ hasInFlight(requestUri: string): boolean { return this.#inFlight.has(requestUri) }
runningLabels(): string[] { return [...this.#inFlight.values()].map((entry) => entry.label) }
/** Requests this daemon abandoned mid-turn, so the loss is logged (and killed) exactly once. */ readonly #abandoned = new Set<string>()
/** * Stops work on a request whose claim we lost, killing its container by label. The tracked * promise then settles through the ordinary failure path into `markCrashed` — a lost claim is a * crashed attempt, not a fulfilment, and the winner's artifact is the one that lands. * * Idempotent: `ClaimManager` may observe the loss on several consecutive ticks before the * container is actually gone, and the drill asserts `abandon` is called at most once. */ abandon(requestUri: string, reason: string): void { const entry = this.#inFlight.get(requestUri) if (!entry || this.#abandoned.has(requestUri)) return this.#abandoned.add(requestUri) this.#deps.log?.(`abandoned ${requestUri}: ${reason} (killing ${entry.label})`) void this.#deps.kill?.(entry.label).catch((error: unknown) => this.#deps.log?.( `failed to kill ${entry.label} for abandoned ${requestUri}: ${error instanceof Error ? error.message : String(error)}`, ), ) }
/** Launches eligible turns from `index` up to remaining concurrency; returns immediately. */ pump(index: MaterializedIndex, actors: ActorRegistry): void { const now = this.#deps.now?.() const heldClaims = this.#deps.heldClaims?.() ?? new Set<string>() // Rejections are ordinary on every sync tick (requests for another operator, exhausted // capacity, unresolved claims, and so on). Keep them out of the daemon log; actionable turn // failures are still logged by the launch and settlement paths below. const dispatchable = selectDispatchable(index, actors, this.#deps.ledger, now, undefined, heldClaims) // The claim view is a cache of the fold, and a cache can be stale in the dangerous direction: // `heldClaims` still naming a request whose claim another operator has since won. So the winner // is re-read from THIS index, immediately before launching, for every unassigned request. const winningClaims = new Map<string, string>() for (const view of [...index.goals, ...index.projects]) { for (const [requestUri, claim] of Object.entries(view.winningClaims)) { winningClaims.set(requestUri, claim.did) } } const selected = new Set(dispatchable.map((item) => item.request.uri)) for (const uri of this.#contended) if (!selected.has(uri)) this.#contended.delete(uri)
for (const item of dispatchable) { if (this.#inFlight.size >= this.#deps.concurrency) break const uri = item.request.uri if (this.#inFlight.has(uri)) continue // C2: the symmetric guard. `selectDispatchable` already checked this against the same index, // so reaching here means the two disagree — never launch on the strength of the cache alone. if (!item.request.value.assignee && winningClaims.get(uri) !== item.actor.did) { continue } if (item.artifactType.name === IMPLEMENTATION_TYPE) { const blocked = this.#implementationBlocked(item.bundle.project.gitUrl) if (blocked) { this.#deps.log?.(`not dispatching implementation ${uri}: ${blocked}`) continue } }
const cid = item.request.cid // Exactly one anchor (goal- or project-scoped), per the lexicon's exactOne(goal, project). const goal = item.request.value.goal const project = item.request.value.project if (!goal && !project) continue // defensive against a malformed record const anchor: TurnRequestAnchor = goal ? { uri, cid, goal } : { uri, cid, project: project as StrongRef }
// A human-error ambiguous request, of any type in either scope: post an explanation on the // request and give up (a `turn reset` re-enables it once fixed). No container runs. if (item.predecessor?.ambiguous) { const label = `radial.reject.${uri.slice(-16)}` const promise = this.#rejectAmbiguous(item) .catch((error: unknown) => this.#deps.log?.(`failed to reject ambiguous ${uri}: ${error instanceof Error ? error.message : String(error)}`), ) .finally(() => this.#inFlight.delete(uri)) this.#inFlight.set(uri, { promise, label }) continue }
const serializedHarness = this.#deps.serializedHarnesses?.includes(item.actor.harness) ? item.actor.harness : undefined if (serializedHarness && this.#activeSerializedHarnesses.has(serializedHarness)) { continue }
// Contention, not an error: skip and retry on a later pump, leaving the ledger row untouched // (this runs BEFORE markRunning, so the request stays plainly eligible). const contested = item.artifactType.name === IMPLEMENTATION_TYPE ? item.predecessor?.artifact?.branch : undefined if (contested && this.#activeBranches.has(contested)) { if (!this.#contended.has(uri)) { this.#contended.add(uri) this.#deps.log?.( `not dispatching ${uri}: predecessor branch ${contested} is already being pushed by ${this.#activeBranches.get(contested)}; retrying on a later pump`, ) } continue } this.#contended.delete(uri)
const label = turnContainerLabel(uri, cid, this.#deps.instance) const runDir = this.#deps.runDirFor(uri) this.#deps.ledger.markRunning(uri, cid, { containerLabel: label, checkoutPath: join(runDir, 'checkout') }) this.#deps.log?.(`dispatching turn ${uri} (label ${label})`)
if (contested) this.#activeBranches.set(contested, uri) if (serializedHarness) this.#activeSerializedHarnesses.add(serializedHarness) const promise = this.#launch(item, { anchor, runDir, label }) .finally(() => { this.#inFlight.delete(uri) // A relaunch (a lease reclaimed after the winner crashed) must be abandonable again. this.#abandoned.delete(uri) if (contested && this.#activeBranches.get(contested) === uri) this.#activeBranches.delete(contested) if (serializedHarness) this.#activeSerializedHarnesses.delete(serializedHarness) }) this.#inFlight.set(uri, { promise, label }) } }
/** The injected remote probe, or a plain `git ls-remote` under this dispatcher's scheme allowlist. * The default carries no credential, so a private repository wants the injected one (cli.ts binds * the daemon's token into it); an unauthenticated probe there throws, and the caller's catch * treats a throw the way it treats a missing branch — a fresh branch, logged. */ /** Why an implementation turn cannot run for this project's forge, or undefined. Per project: * the adapter that speaks for the remote decides, and only when none does (or none is configured) * does the daemon-wide `implementationEnabled` flag have the last word. */ #implementationBlocked(gitUrl: string): string | undefined { const adapter = this.#deps.forges?.for(gitUrl) if (adapter) return adapter.implementationBlockedReason?.() return this.#deps.implementationEnabled === false ? 'GitHub authentication is unavailable' : undefined }
async #branchExists(gitUrl: string, branch: string): Promise<boolean> { const probe = this.#deps.branchExists if (probe) return probe({ gitUrl, branch }) return remoteBranchExists({ gitUrl, branch, allowedSchemes: this.#deps.allowedSchemes }) }
/** Prepare and run one turn, then settle the ledger. */ async #launch( item: Dispatchable, ctx: { anchor: TurnRequestAnchor; runDir: string; label: string }, ): Promise<void> { const { anchor, runDir, label } = ctx const uri = anchor.uri const cid = anchor.cid const isImpl = item.artifactType.name === IMPLEMENTATION_TYPE
let checkoutRef: string | undefined let branch: string | undefined if (isImpl) { const base = item.bundle.project.defaultBranch const gitUrl = item.bundle.project.gitUrl branch = implBranchName(uri, cid) const predecessor = item.predecessor?.artifact // v2 branch reuse is deliberately narrow (finding 6): only reuse the predecessor's branch when // EVERY guard holds — its name is in the daemon's own `radial/impl-` namespace, its PR lives in // THIS project's repo and is OPEN, and the PR's head ref/repo and base ref all match. Anything // else (a "main" branch link, a cross-fork or closed/merged PR, a mismatched base) → a fresh // `radial/impl-<hash>` branch + new PR. The daemon never checks out or pushes a branch outside // the radial/impl- namespace. // // Without a forge adapter the live confirmation is unavailable, and refusing to reuse there // was the single likeliest reason a v2 turn opened a second PR: `run.forge` is easy to omit // and nothing said so. So the index plus the git remote answer instead — an in-namespace // branch, a PR in this project's repo, no trusted `merge` annotation on the predecessor, and // a branch that still exists on the remote. // // That last check does the work here. A `merge` annotation is written by the merge poller, // which only runs when a forge is configured (cli.ts), so in the deployment this fallback // exists for there is never one to find — `predecessor.merged` cannot fire, and on its own // this would reuse a branch forever, merged or not. Reusing a merged-but-still-present branch // is survivable (the push lands, `gh pr create` opens a PR whose diff is just the new work); // reusing a DELETED one is not, because the clone fails identically on every retry until the // request gives up. One `git ls-remote` is the whole difference, and it needs no forge. // // The one thing still invisible from here is a PR closed WITHOUT merging on a branch that // survives; that self-corrects, because the container runs `gh pr create` when `gh pr view` // finds no open PR for the branch. if (predecessor?.prUrl && predecessor.branch?.startsWith('radial/impl-')) { try { // The adapter that speaks for this project answers both URL questions; where none is // configured, URL-only GitHub reasoning does, which is what every existing deployment had. const adapter = this.#deps.forges?.for(gitUrl) ?? URL_ONLY_FORGES.for(gitUrl) if (!adapter) throw new Error(`no forge adapter speaks for ${gitUrl}`) const projectFullName = await adapter.projectIdentity(gitUrl) const state = this.#deps.forge ?? (adapter.canObserve ? adapter : undefined) if (!(await adapter.pullBelongsToProject(predecessor.prUrl, gitUrl))) { this.#deps.log?.(`predecessor PR ${predecessor.prUrl} for ${uri} is not in ${projectFullName}; opening a fresh branch`) } else if (state) { const pr = await state.getPullRequestState(predecessor.prUrl, { gitUrl }) if (pr.state === 'open' && pr.headRef === predecessor.branch && pr.headRepoFullName.toLowerCase() === projectFullName.toLowerCase() && pr.baseRef === base) { checkoutRef = predecessor.branch branch = predecessor.branch this.#deps.log?.(`continuing ${predecessor.branch} for ${uri}: the forge confirms ${predecessor.prUrl} is open on ${base}`) } else { this.#deps.log?.(`predecessor PR ${predecessor.prUrl} for ${uri} not a clean open reuse; opening a fresh branch`) } } else if (predecessor.merged) { this.#deps.log?.(`predecessor ${predecessor.ref.uri} for ${uri} is recorded as merged; opening a fresh branch`) } else if (!(await this.#branchExists(gitUrl, predecessor.branch))) { this.#deps.log?.( `predecessor branch ${predecessor.branch} for ${uri} no longer exists on ${gitUrl}; opening a fresh branch`, ) } else { checkoutRef = predecessor.branch branch = predecessor.branch this.#deps.log?.( `continuing ${predecessor.branch} for ${uri} on the index signal (no run.forge configured, no merge record for the predecessor, branch still on the remote). ` + 'Without a forge nothing writes merge annotations, so a merged branch that still exists reads the same as an open one here, and a PR closed without merging is invisible — the turn opens a new PR if none is open for the branch.', ) } } catch (error) { this.#deps.log?.(`could not validate predecessor PR ${predecessor.prUrl} for ${uri}: ${error instanceof Error ? error.message : String(error)}; opening a fresh branch`) checkoutRef = undefined branch = implBranchName(uri, cid) } } this.#deps.ledger.markBranch(uri, branch) } else if (item.review?.commit) { // A review turn takes the non-impl path: no branch, read-only checkout, pinned to the exact // commit the subject implementation links so the reviewer reads precisely the reviewed code. // A subject without a linked commit (a plan/non-code artifact) leaves checkoutRef unset, so // turn.ts clones the default branch read-only. checkoutRef = item.review.commit }
const turnInput: TurnInput = { bundle: item.bundle, request: anchor, artifactType: item.artifactType, requestRecord: item.request.value, actor: item.actor, runDir, logPath: join(runDir, '..', '..', 'turn-logs', `${planArtifactRkey(uri, cid)}.log`), image: this.#deps.image, timeoutMs: this.#deps.timeoutMs, ...(this.#deps.network !== undefined ? { network: this.#deps.network } : {}), allowedSchemes: this.#deps.allowedSchemes, ...(this.#deps.memory !== undefined ? { memory: this.#deps.memory } : {}), ...(this.#deps.turnTransport !== undefined ? { turnTransport: this.#deps.turnTransport } : {}), ...(this.#deps.instance !== undefined ? { instance: this.#deps.instance } : {}), ...(checkoutRef !== undefined ? { checkoutRef } : {}), // `prev` is scope- and type-agnostic: any v2 artifact continues its predecessor's chain. Only // `branch` remains implementation-specific. ...(item.predecessor?.artifact ? { prev: item.predecessor.artifact.ref } : {}), ...(isImpl && branch !== undefined ? { branch } : {}), ...(item.answer ? { answer: { subjectAuthor: item.answer.subjectAuthor, anchor: item.answer.anchor, ...(item.answer.root ? { root: item.answer.root } : {}), }, } : {}), }
try { const result = await this.#deps.runTurn(turnInput) // A pull request the DAEMON opened (tangled) is recorded before the outcome is settled: it // exists on the forge whatever became of the turn, and a retry that finds it here does not // have to re-list the agent's repo to know so. if (result.pull) this.#deps.ledger.markPull(uri, result.pull.url) if (result.outcome === 'fulfilled') { this.#deps.ledger.markFulfilled(uri, result.acceptedRef as StrongRef) } else if (result.outcome === 'awaiting_input') { this.#deps.ledger.markAwaitingInput(uri) } else { const outcome = this.#deps.ledger.markCrashed(uri) if (outcome.state === 'gave_up') { this.#deps.log?.(`turn ${uri} GAVE UP after ${outcome.attempts} attempts (label ${label})`) } } } catch (error: unknown) { const outcome = this.#deps.ledger.markCrashed(uri) const message = error instanceof Error ? error.message : String(error) this.#deps.log?.( `turn ${uri} threw: ${message}; now ${outcome.state} (attempt ${outcome.attempts})` + (turnInput.logPath ? ` (diagnostic log: ${turnInput.logPath})` : ''), ) } }
/** Post an explanation on an ambiguous request and give it up (the request stays open). */ async #rejectAmbiguous(item: Dispatchable): Promise<void> { const uri = item.request.uri const type = item.request.value.type const scope = item.request.value.goal ? 'goal' : 'project' const body = `Radial cannot start this turn: it is based on more than one ${type} artifact of this ${scope}, ` + `so there is no single predecessor to revise. Base the request on exactly one ${type} ` + `(or none, for a fresh ${type}) and request it again.` // The message lexicon requires exactly one goal|artifact anchor. A goal-scoped request anchors // to its own goal; a project-scoped one has none, so it anchors to the lowest-uri candidate — // ambiguity means there are at least two, so an artifact anchor always exists here. const anchor = item.request.value.goal ? { goal: item.request.value.goal } : { artifact: (item.predecessor?.candidates[0] as ArtifactPredecessor).ref } const record: MessageRecord = { $type: COLLECTIONS.message, ...anchor, body, mentions: [], re: { uri, cid: item.request.cid }, createdAt: this.#deps.now?.() ?? new Date().toISOString(), } await item.actor.client.create(COLLECTIONS.message, record) this.#deps.ledger.giveUp(uri, item.request.cid) this.#deps.log?.(`rejected ambiguous ${type} request ${uri} (posted explanation, gave up)`) }
/** Awaits every in-flight turn (best-effort graceful shutdown; each promise already never rejects). */ async drain(): Promise<void> { await Promise.all([...this.#inFlight.values()].map((entry) => entry.promise)) }}