Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820import { createHash } from 'node:crypto'import { chmod, rm } from 'node:fs/promises'import { createServer, type Server, type Socket } from 'node:net'import { dirname } from 'node:path'import { XrpcError } from '@radial/atproto'import { ForgeRegistry, type ForgeAdapter } from './forge.js'import { urlOnlyForgeAdapters } from './forge-github.js'import { COLLECTIONS, TURN_LIMITS, normalizeArtifactTitle, type ArtifactRecord, type Collection, type MessageRecord, type RadialRecord, type RecordByCollection, type ReviewFinding, type ReviewRecord, type StrongRef, type TurnRpcResponse,} from '@radial/core'
export interface TurnRecordWriter { create<K extends Collection>( collection: K, value: RecordByCollection[K], options?: { rkey?: string }, ): Promise<StrongRef> getOwnRecord( collection: Collection, rkey: string, ): Promise<{ uri: string; cid: string; value: RadialRecord } | undefined>}
/** sha256 hex of a UTF-8 string; used to compare an adopted record's body against the pending one * without storing the full body in the ledger/log. */export function sha256Hex(text: string): string { return createHash('sha256').update(text).digest('hex')}
/** * The daemon-owned, fail-closed turn mode. An `artifact` turn may only submit an artifact of the * named `type`; a `review` turn may only submit a verdict on the named `subject`. runTurn populates * this from the request record and verifies it against `requestRecord.type`, so submitArtifact can * never silently default to a plan for a review request (and vice versa). */export type TurnMode = | { kind: 'artifact'; type: string } | { kind: 'review'; subject: StrongRef } | { kind: 'answer' /** The exact message version being answered — the reply's `parent`. */ subject: StrongRef /** The thread root the subject belongs to, when it is not itself the root. */ root?: StrongRef /** Where the reply hangs: the SUBJECT's own anchor, not the request's goal. An answer request * is goal-scoped in v1, but a message under that goal may be anchored to one of its artifacts * — the fold files those into the goal's thread all the same — and a reply belongs to the same * target as what it answers. That is the rule `message post --parent` enforces for a human, so * the daemon must not write a record it would have refused from one. */ anchor: { goal: StrongRef } | { artifact: StrongRef } /** The subject message's author, carried into `mentions`: the structural "you were replied * to", which the daemon knows from the index and the container never supplies. */ subjectAuthor: string }
/** The anchor a turn's artifact/message records hang off. Exactly one of goal|project is present: * a goal-scoped request carries `goal`, a project-scoped (system-artifact) request carries * `project`. The lexicon's exactOne(goal, project) invariant is honored by construction. */export type TurnRequestAnchor = | { uri: string; cid: string; goal: StrongRef; project?: undefined } | { uri: string; cid: string; project: StrongRef; goal?: undefined }
export interface TurnRequestContext { token: string /** The request being fulfilled, with its goal- or project-scope anchor. */ request: TurnRequestAnchor client: TurnRecordWriter now?: () => string /** Discriminated turn mode, owned by the daemon (see TurnMode). */ mode: TurnMode /** Effective model observed by the daemon at write time. A getter keeps runtime reroutes that * arrive after the socket starts from being frozen into the initial context. */ model?: () => string | undefined /** Daemon-owned implementation provenance. Everything the daemon needs to validate the submitted * links — and, on a forge whose containers cannot open a pull request themselves, to open it. */ implementation?: { gitUrl: string branch: string /** The base branch the pull targets. */ base?: string /** The commit the checkout started from: the graft point a patch is generated against. */ baseCommit?: string /** The turn's own checkout, which the daemon owns and the container's commits are already in. */ checkoutPath?: string /** The artifact URI this turn's record will land at, so a brokered pull body can link it — the * same provenance a GitHub turn is told to put in its own pull request body. */ artifactUri?: string /** A human-readable title for a brokered pull request. */ pullTitle?: string /** The adapter that speaks for this project, when one is configured. Where it implements * `openPullRequest`, `--pr` is not required of the container and the daemon opens the pull * itself; otherwise the submitted `--pr` is validated against the project and used as-is. */ forge?: ForgeAdapter } /** The version-chain predecessor the daemon resolved for this request, stamped onto the artifact * whatever its type: a plan v2 and an architecture v2 continue their chains exactly as an * implementation v2 does. Absent for a v1 (and for review turns, which submit no artifact). */ prev?: StrongRef /** Message anchor for a project-scoped turn's question (askQuestion). The message lexicon requires * exactly one goal|artifact anchor, and a project-scoped request has no goal of its own, so the * daemon threads the resolved anchor here: the capture source goal, else the lowest-uri resolved * basedOn artifact. Absent → a direct project turn without basedOn cannot ask a question (v1). */ questionAnchor?: { goal: StrongRef } | { artifact: StrongRef }}
export interface TurnObservation { artifact?: StrongRef review?: StrongRef /** An answer turn's terminal record: the thread reply it posted. */ answer?: StrongRef questions: StrongRef[] /** The pull request the daemon opened on the turn's behalf, where it did (tangled): the value it * stamped into `links.pr`, and the record it wrote. */ pull?: { url: string; record?: StrongRef }}
export interface TurnSocketLimits { maxFrameBytes: number frameTimeoutMs: number artifactBodyChars: number messageBodyChars: number}
/** * Where the turn socket binds. `unix` (the production default, design-correct on Linux) binds a * bind-mounted socket file; `tcp` is the dev-only escape hatch for macOS/Docker Desktop, where a * bind-mounted AF_UNIX socket is visible inside the VM but `connect()` gets ECONNREFUSED (AF_UNIX * cannot cross the macOS<->VM boundary). A bare `string` is shorthand for `{ kind: 'unix', path }`, * so existing callers/tests that pass a socket path directly keep working unchanged. */export type TurnListen = { kind: 'unix'; path: string } | { kind: 'tcp'; host: string; port: number }
/** URL reasoning for a project no configured adapter matches — the pre-registry GitHub parse, so a * submit's refusal keeps exactly the shape it had. */const URL_ONLY_FORGES = new ForgeRegistry(urlOnlyForgeAdapters())
const BASE32_ALPHABET = 'abcdefghijklmnopqrstuvwxyz234567'
/** Lowercase RFC-4648 base32 (no padding). Exported so deterministic-rkey helpers in other modules * (e.g. check-runner's `checkrunRkey`) share exactly this encoding. */export function base32Encode(bytes: Uint8Array): string { let bits = 0 let value = 0 let output = '' for (const byte of bytes) { value = (value << 8) | byte bits += 8 while (bits >= 5) { output += BASE32_ALPHABET[(value >>> (bits - 5)) & 31] bits -= 5 } } if (bits > 0) output += BASE32_ALPHABET[(value << (5 - bits)) & 31] return output}
/** * Deterministic rkey for the artifact answering a request: `<prefix>` plus 24 base32 characters of * sha256(requestUri#requestCid). A retried turn recomputes the same rkey, so a duplicate write * collides on the PDS instead of creating a second record (see #submitArtifact adoption). The 24-char base32 * body is prefix-independent, so `turnContainerLabel` (which strips the prefix) is stable across * plan/implementation turns for the same request. */export function artifactRkey(requestUri: string, requestCid: string, prefix: string): string { const digest = createHash('sha256').update(`${requestUri}#${requestCid}`).digest() return `${prefix}${base32Encode(digest).slice(0, 24)}`}
/** `plan-` prefixed rkey (plans and every non-implementation type share the synchronous write path). */export function planArtifactRkey(requestUri: string, requestCid: string): string { return artifactRkey(requestUri, requestCid, 'plan-')}
/** `impl-` prefixed rkey for a synchronous implementation artifact. */export function implArtifactRkey(requestUri: string, requestCid: string): string { return artifactRkey(requestUri, requestCid, 'impl-')}
/** `review-` prefixed deterministic rkey for the verdict answering a review request. Distinct * prefix from the artifact rkeys so a review turn and an (impossible, but defensively distinct) * artifact turn for the same request never collide, and a retried review turn recomputes the same * rkey and adopts rather than double-writing. The lexicon key was changed to `any` to permit it. */export function reviewRkey(requestUri: string, requestCid: string): string { return artifactRkey(requestUri, requestCid, 'review-')}
/** `answer-` prefixed deterministic rkey for the message answering an answer request. It matters * more here than anywhere else: a retried answer turn without it double-posts INTO THE THREAD, which * a human reads, rather than leaving a duplicate record only the fold sees. A second submit for the * same request collides at the PDS and adopts (see #submitAnswer). */export function answerRkey(requestUri: string, requestCid: string): string { return artifactRkey(requestUri, requestCid, 'answer-')}
/** Deterministic push branch for a fresh implementation turn: `radial/impl-<12 hex of * sha256(requestUri#requestCid)>`. Stable across retries so a re-push targets the same branch. (A * v2 turn reusing an open predecessor PR pushes to the predecessor's branch instead.) */export function implBranchName(requestUri: string, requestCid: string): string { const digest = createHash('sha256').update(`${requestUri}#${requestCid}`).digest() const hex = [...digest].map((byte) => byte.toString(16).padStart(2, '0')).join('') return `radial/impl-${hex.slice(0, 12)}`}
function concatBytes(chunks: Uint8Array[]): Uint8Array { const total = chunks.reduce((sum, chunk) => sum + chunk.length, 0) const result = new Uint8Array(total) let offset = 0 for (const chunk of chunks) { result.set(chunk, offset) offset += chunk.length } return result}
type ParsedEnvelope = | { method: 'submitArtifact'; token: string; title?: string; body: string; criteria?: string[]; branch?: string; commit?: string; pr?: string } | { method: 'askQuestion'; token: string; body: string } | { method: 'submitAnswer'; token: string; body: string } | { method: 'submitReview'; token: string; verdict: 'approve' | 'request_changes'; findings: ReviewFinding[] }
const isObject = (value: unknown): value is Record<string, unknown> => typeof value === 'object' && value !== null && !Array.isArray(value)
/** A git object name: hex, 7–64 chars. The upper bound is the length cap that keeps a malformed * `commit` from ever reaching `git` as an oversized/argument-injecting value. */const isShaShaped = (value: unknown): value is string => typeof value === 'string' && /^[0-9a-fA-F]{7,64}$/.test(value)
/** * Full, defense-in-depth validation of a submitReview envelope at the untrusted socket edge. The * daemon owns subject/request (from turn context, not the wire), so only verdict + findings are * parsed here. Findings mirror the review lexicon exactly: ≤100 entries, each with a valid severity, * a ≤10,000-char body, an optional string path, and an optional integer line. Anything else → reject * (undefined, surfaced to the caller as an invalid-envelope error). */function parseReviewEnvelope(raw: Record<string, unknown>, token: string): ParsedEnvelope | undefined { const { verdict, findings } = raw if (verdict !== 'approve' && verdict !== 'request_changes') return undefined if (!Array.isArray(findings) || findings.length > 100) return undefined const parsed: ReviewFinding[] = [] for (const finding of findings) { if (!isObject(finding)) return undefined const { path, line, severity, body } = finding if (severity !== 'info' && severity !== 'warning' && severity !== 'error') return undefined if (typeof body !== 'string' || body.length > 10_000) return undefined // Mirror the review lexicon exactly at the untrusted edge: path ≤ 2000 chars, line an integer ≥ 1. if (path !== undefined && (typeof path !== 'string' || path.length > 2000)) return undefined if (line !== undefined && (typeof line !== 'number' || !Number.isInteger(line) || line < 1)) return undefined parsed.push({ severity, body, ...(path !== undefined ? { path } : {}), ...(line !== undefined ? { line } : {}), }) } return { method: 'submitReview', token, verdict, findings: parsed }}
function parseEnvelope(raw: unknown): ParsedEnvelope | undefined { if (!isObject(raw)) return undefined const { method, token } = raw if (typeof token !== 'string') return undefined if (method === 'submitReview') return parseReviewEnvelope(raw, token) if (method !== 'submitArtifact' && method !== 'askQuestion' && method !== 'submitAnswer') return undefined const { body, criteria, branch, commit, pr, title } = raw if (typeof body !== 'string') return undefined if (method === 'submitArtifact') { // Shape only: a `title` that is present but not a string is a malformed envelope. Whether it is // usable — present at all, non-blank, within the lexicon's ceiling — is #submitArtifact's // decision, so that a turn gets an error message saying what to do instead of "invalid envelope". if (title !== undefined && typeof title !== 'string') return undefined let parsedCriteria: string[] | undefined if (criteria !== undefined) { if (!Array.isArray(criteria) || criteria.some((entry) => typeof entry !== 'string')) return undefined parsedCriteria = criteria as string[] } // `commit` is optional at the envelope layer (plan turns never send it); when present it must be // SHA-shaped. #submitArtifact enforces that an implementation turn actually supplies one. if (commit !== undefined && !isShaShaped(commit)) return undefined if (branch !== undefined && typeof branch !== 'string') return undefined if (pr !== undefined && typeof pr !== 'string') return undefined return { method, token, body, ...(title !== undefined ? { title: title as string } : {}), ...(parsedCriteria ? { criteria: parsedCriteria } : {}), ...(commit !== undefined ? { commit: commit as string } : {}), ...(branch !== undefined ? { branch: branch as string } : {}), ...(pr !== undefined ? { pr: pr as string } : {}), } } return { method, token, body }}
/** * The only harness -> protocol write path for a turn: the daemon owns record construction * (anchors, type, rkey), the container in the sandbox supplies only a body (and, for * submitArtifact, criteria). One unix-socket connection carries exactly one newline-delimited * JSON request/response pair, then the connection closes. */export class TurnSocketServer { readonly #listen: TurnListen readonly #context: TurnRequestContext readonly #limits: TurnSocketLimits readonly #observation: TurnObservation = { questions: [] } #server: Server | undefined #boundPort: number | undefined // Synchronous in-flight reservation: set the instant #submitArtifact is entered (before any // `await`), so two concurrent connections racing the same server can never both pass the // "already submitted" check before either write lands. See #submitArtifact below. #submitReserved = false
constructor( listen: TurnListen | string, context: TurnRequestContext, options: { limits?: TurnSocketLimits } = {}, ) { this.#listen = typeof listen === 'string' ? { kind: 'unix', path: listen } : listen this.#context = context this.#limits = options.limits ?? TURN_LIMITS }
get observation(): TurnObservation { return this.#observation }
/** The bound TCP port (tcp transport only; `undefined` for unix, or before `listen()` resolves). */ get boundPort(): number | undefined { return this.#boundPort }
/** * Binds and resolves once listening. unix: unlinks any stale socket file first, then widens the * socket and its parent directory (0o666/0o777) so the turn container's uid-1000 process can * traverse in and connect — safe because the socket is per-turn, single-use, and every RPC is * gated by RADIAL_TURN_TOKEN. tcp (dev-only, see `TurnListen`): no file to unlink or chmod; the * per-turn random token is the auth boundary instead. */ async listen(): Promise<void> { const target = this.#listen if (target.kind === 'unix') { await rm(target.path, { force: true }) await new Promise<void>((resolve, reject) => { const server = createServer((socket) => this.#handleConnection(socket)) server.on('error', reject) server.listen(target.path, () => { this.#server = server resolve() }) }) await chmod(target.path, 0o666) await chmod(dirname(target.path), 0o777) return } await new Promise<void>((resolve, reject) => { const server = createServer((socket) => this.#handleConnection(socket)) server.on('error', reject) server.listen(target.port, target.host, () => { this.#server = server this.#boundPort = server.address()?.port resolve() }) }) }
async close(): Promise<void> { const server = this.#server if (!server) return this.#server = undefined await new Promise<void>((resolve, reject) => { server.close((error) => (error ? reject(error) : resolve())) }) }
#handleConnection(socket: Socket): void { const chunks: Uint8Array[] = [] let total = 0 let done = false // Set the instant a complete frame is found — synchronously, before any `await` — so a second // 'data' event arriving while #dispatch is still in flight (e.g. mid-await on the record // write) can never re-concatenate and re-dispatch the same frame. `done` alone is not enough: // it is only set later, inside `finish`, after the async dispatch resolves. let consumed = false
const finish = (response: TurnRpcResponse): void => { if (done) return done = true try { socket.end(`${JSON.stringify(response)}\n`) } catch { // Best effort: the peer may already be gone. } }
socket.setTimeout(this.#limits.frameTimeoutMs) socket.on('timeout', () => finish({ ok: false, error: 'turn request timed out' })) socket.on('error', () => { done = true // Connection is broken; there is no one left to write a response to. })
socket.on('data', (chunk) => { if (done || consumed) return chunks.push(chunk) total += chunk.length if (total > this.#limits.maxFrameBytes) { finish({ ok: false, error: `frame exceeds ${this.#limits.maxFrameBytes} bytes` }) return } const buffer = concatBytes(chunks) const newlineIndex = buffer.indexOf(10) if (newlineIndex === -1) return // Exactly one RPC per connection: claim the frame now, synchronously. Any further bytes on // this connection (a second frame, or trailing garbage) are discarded, not re-dispatched. consumed = true void this.#dispatch(buffer.subarray(0, newlineIndex)) .then(finish) .catch((error: unknown) => finish({ ok: false, error: error instanceof Error ? error.message : String(error) }), ) }) }
async #dispatch(frame: Uint8Array): Promise<TurnRpcResponse> { let raw: unknown try { raw = JSON.parse(new TextDecoder().decode(frame)) } catch { return { ok: false, error: 'invalid JSON frame' } } const envelope = parseEnvelope(raw) if (!envelope) return { ok: false, error: 'invalid turn RPC envelope' } if (envelope.token !== this.#context.token) return { ok: false, error: 'invalid turn token' } const limits = this.#limits if (envelope.method === 'submitArtifact' && envelope.body.length > limits.artifactBodyChars) { return { ok: false, error: `body exceeds ${limits.artifactBodyChars} characters` } } // A reply and a question are both messages, and the lexicon caps them the same way. if ( (envelope.method === 'askQuestion' || envelope.method === 'submitAnswer') && envelope.body.length > limits.messageBodyChars ) { return { ok: false, error: `body exceeds ${limits.messageBodyChars} characters` } } if (envelope.method === 'submitReview') return this.#submitReview(envelope) if (envelope.method === 'submitAnswer') return this.#submitAnswer(envelope) return envelope.method === 'submitArtifact' ? this.#submitArtifact(envelope) : this.#askQuestion(envelope) }
#now(): string { return this.#context.now?.() ?? new Date().toISOString() }
#model(): { model?: string } { const model = this.#context.model?.() return model !== undefined ? { model } : {} }
async #submitArtifact(envelope: Extract<ParsedEnvelope, { method: 'submitArtifact' }>): Promise<TurnRpcResponse> { // Fail-closed mutual exclusion: a review turn cannot emit an artifact. This is the daemon-side // enforcement of the §13 invariant that a review turn can only produce a verdict of its subject. if (this.#context.mode.kind === 'review') { return { ok: false, error: 'this is a review turn; only a review verdict may be submitted, not an artifact' } } if (this.#context.mode.kind === 'answer') { return { ok: false, error: 'this is an answer turn; only a thread reply may be submitted, not an artifact' } } if (this.#observation.artifact || this.#observation.review || this.#observation.answer || this.#submitReserved) { return { ok: false, error: 'an artifact was already submitted for this request' } } // The title, checked BEFORE the reservation below rather than after it: nothing has been written // and nothing is in flight, so a turn that mis-titled its submit may simply submit again. (The // paths that release the reservation on failure are the ones that had to reserve it first because // they await.) `title` stays optional on the wire so historical records remain valid, and required // of a writer, which is what keeps every NEW artifact scannable. const title = normalizeArtifactTitle(envelope.title) if (title === undefined) { return { ok: false, error: envelope.title === undefined ? 'artifact submit requires --title: a short phrase naming this artifact, not its first paragraph' : `--title must be non-blank and at most ${TURN_LIMITS.artifactTitleChars} characters`, } } // Reserve synchronously, before the first `await` below, so a second concurrent submit (a // different connection/socket on this same server) sees the reservation and is rejected the // same way a sequential second submit would be — it can never race past this check and // silently adopt a different body than the one that actually landed. this.#submitReserved = true const { request, client } = this.#context const type = this.#context.mode.type
let links: ArtifactRecord['links'] = {} const prev = this.#context.prev if (type === 'implementation') { if ( !envelope.branch?.startsWith('radial/impl-') || envelope.branch !== this.#context.implementation?.branch ) { this.#submitReserved = false return { ok: false, error: 'implementation branch does not match the daemon-selected reserved branch', } } if (!envelope.commit || !/^[0-9a-f]{40}$/.test(envelope.commit)) { this.#submitReserved = false return { ok: false, error: 'commit must be a full 40-character lowercase hex sha' } } const implementation = this.#context.implementation if (!implementation) { this.#submitReserved = false return { ok: false, error: 'implementation submit requires --branch, --commit, and --pr' } } // Whose job the pull request is. Where the adapter can open one (tangled has no CLI a // container could run, so the daemon does it), `--pr` is not asked of the turn at all; // everywhere else it stays REQUIRED and is validated against this project's repository. const adapter = implementation.forge let prUrl: string if (adapter?.openPullRequest) { try { const opened = await adapter.openPullRequest({ gitUrl: implementation.gitUrl, base: implementation.base ?? '', branch: envelope.branch, commit: envelope.commit, ...(implementation.baseCommit ? { baseCommit: implementation.baseCommit } : {}), checkoutPath: implementation.checkoutPath ?? '', title: implementation.pullTitle ?? `Radial implementation ${envelope.branch}`, body: implementation.artifactUri ? `${envelope.body}\n\n[Radial artifact](${implementation.artifactUri})` : envelope.body, }) prUrl = opened.url this.#observation.pull = opened } catch (error) { // The reservation is released so a later retry of the whole turn can submit again: no // artifact was written, and the turn must not be recorded as having produced one. this.#submitReserved = false return { ok: false, error: `could not open the pull request for this implementation: ${error instanceof Error ? error.message : String(error)}`, } } } else { if (!envelope.pr) { this.#submitReserved = false return { ok: false, error: 'implementation submit requires --branch, --commit, and --pr' } } try { const urls = adapter ?? URL_ONLY_FORGES.for(implementation.gitUrl) if (!urls) throw new Error(`no forge adapter speaks for ${implementation.gitUrl}`) if (!(await urls.pullBelongsToProject(envelope.pr, implementation.gitUrl))) { throw new Error('PR is not for this project repository') } } catch (error) { this.#submitReserved = false return { ok: false, error: error instanceof Error ? error.message : String(error) } } prUrl = envelope.pr } links = { branch: envelope.branch, commit: envelope.commit, pr: prUrl } }
// Exactly one of goal|project, per the request's scope — the lexicon's exactOne(goal, project). const anchor = request.goal ? { goal: request.goal } : { project: request.project } const record: ArtifactRecord = { $type: COLLECTIONS.artifact, request: { uri: request.uri, cid: request.cid }, ...anchor, type, title, body: envelope.body, links, ...(prev ? { prev } : {}), ...(envelope.criteria && envelope.criteria.length ? { criteria: envelope.criteria } : {}), ...this.#model(), createdAt: this.#now(), } const rkey = type === 'implementation' ? implArtifactRkey(request.uri, request.cid) : planArtifactRkey(request.uri, request.cid) let ref: StrongRef try { ref = await client.create(COLLECTIONS.artifact, record, { rkey }) } catch (error) { if (error instanceof XrpcError && error.status === 400 && error.error === 'RecordAlreadyExists') { const existing = await client.getOwnRecord(COLLECTIONS.artifact, rkey) if (!existing) { // Not a genuine adoption (the record the PDS claims exists isn't there) — a hard error, // so release the reservation to allow a real later retry. this.#submitReserved = false throw error } const current = existing.value as ArtifactRecord // Compare against whichever anchor this request carries (goal- or project-scoped). The // expected anchor must match AND the OTHER anchor must be absent: a stray second anchor is a // different record than we would write, and the materializer resolves a dual-anchor artifact // to its goal (goal takes precedence) — adopting one for a project request would leave that // request open forever while the ledger reported it fulfilled. const currentAnchor = request.goal ? current.goal : current.project const strayAnchor = request.goal ? current.project : current.goal const expectedAnchor = request.goal ?? request.project if ( current.$type !== COLLECTIONS.artifact || current.type !== type || current.request.uri !== request.uri || current.request.cid !== request.cid || currentAnchor?.uri !== expectedAnchor.uri || currentAnchor?.cid !== expectedAnchor.cid || strayAnchor !== undefined || JSON.stringify(current.links) !== JSON.stringify(links) || JSON.stringify(current.prev) !== JSON.stringify(prev) || JSON.stringify(current.criteria ?? []) !== JSON.stringify(envelope.criteria ?? []) || // The title is part of the payload, so it is part of what makes the record at this rkey the // one this submit would have written. Omitting it would let a retry adopt — and report as // its own — an artifact carrying a different title than the one it just submitted. current.title !== title || current.model !== record.model || current.body !== envelope.body ) { this.#submitReserved = false throw new Error(`refusing to adopt tampered record at ${rkey}`) } ref = { uri: existing.uri, cid: existing.cid } } else { // Hard error, not an adoption path — release the reservation. this.#submitReserved = false throw error } } this.#observation.artifact = ref return { ok: true, ref } }
async #askQuestion(envelope: Extract<ParsedEnvelope, { method: 'askQuestion' }>): Promise<TurnRpcResponse> { const { request, client, questionAnchor } = this.#context // The message lexicon requires exactly one goal|artifact anchor. A goal-scoped turn anchors to // its own goal. A project-scoped turn has no goal of its own, so it uses the daemon-threaded // questionAnchor (capture source goal, else the lowest-uri resolved basedOn artifact). A direct // project turn without basedOn has neither — it cannot ask a question in v1. let anchor: { goal: StrongRef } | { artifact: StrongRef } if (request.goal) { anchor = { goal: request.goal } } else if (questionAnchor) { anchor = questionAnchor } else { return { ok: false, error: 'a direct project-scoped turn without basedOn cannot ask questions in v1', } } const record: MessageRecord = { $type: COLLECTIONS.message, ...anchor, body: envelope.body, mentions: [], re: { uri: request.uri, cid: request.cid }, ...this.#model(), createdAt: this.#now(), } const ref = await client.create(COLLECTIONS.message, record) this.#observation.questions.push(ref) return { ok: true, ref } }
/** * An answer turn's terminal record: a reply in the goal's thread. * * Everything that anchors it is daemon-owned, exactly as a verdict's subject is — the container * supplies prose and nothing else. `parent`/`root` thread it under the message the request named, * the anchor is that message's own (see `TurnMode`), `re` is the backref the fold closes the * request on, and `mentions` carries the subject's author because the index knows who was replied * to and the sandbox has no business asserting it. * * The rkey is deterministic and adopted on collision, like a review's. A retried answer turn * legitimately writes DIFFERENT prose, so the body is deliberately not compared — only the anchors * are, and a record at that rkey with different anchors is a record this turn did not write. */ async #submitAnswer(envelope: Extract<ParsedEnvelope, { method: 'submitAnswer' }>): Promise<TurnRpcResponse> { const { request, client, mode } = this.#context if (mode.kind !== 'answer') { return { ok: false, error: 'this is not an answer turn; a thread reply may not be submitted' } } if (this.#observation.answer || this.#observation.artifact || this.#observation.review || this.#submitReserved) { return { ok: false, error: 'an answer was already submitted for this request' } } // Reserved synchronously before the first await, sharing the latch with the other two submit // paths: whichever terminal a turn reaches, it reaches exactly one of them exactly once. this.#submitReserved = true const re: StrongRef = { uri: request.uri, cid: request.cid } const record: MessageRecord = { $type: COLLECTIONS.message, ...mode.anchor, parent: mode.subject, root: mode.root ?? mode.subject, re, body: envelope.body, mentions: [mode.subjectAuthor], ...this.#model(), createdAt: this.#now(), } const rkey = answerRkey(request.uri, request.cid) let ref: StrongRef try { ref = await client.create(COLLECTIONS.message, record, { rkey }) } catch (error) { if (error instanceof XrpcError && error.status === 400 && error.error === 'RecordAlreadyExists') { const existing = await client.getOwnRecord(COLLECTIONS.message, rkey) if (!existing) { // The PDS claims the record exists but it is not fetchable — a hard error, not an // adoption. Release the reservation so a genuine later retry can still write. this.#submitReserved = false throw error } const current = existing.value as MessageRecord // Compare against whichever anchor this reply carries, exactly as the artifact path does. const expected = 'goal' in mode.anchor ? mode.anchor.goal : mode.anchor.artifact const found = 'goal' in mode.anchor ? current.goal : current.artifact if ( current.$type !== COLLECTIONS.message || found?.uri !== expected.uri || found.cid !== expected.cid || current.parent?.uri !== mode.subject.uri || current.parent.cid !== mode.subject.cid || current.re?.uri !== re.uri || current.re.cid !== re.cid ) { this.#submitReserved = false throw new Error(`refusing to adopt answer record at ${rkey} with a mismatched anchor`) } ref = { uri: existing.uri, cid: existing.cid } } else { this.#submitReserved = false throw error } } this.#observation.answer = ref return { ok: true, ref } }
async #submitReview(envelope: Extract<ParsedEnvelope, { method: 'submitReview' }>): Promise<TurnRpcResponse> { const { request, client, mode } = this.#context // Fail-closed mutual exclusion: submitReview is only valid on a review turn. An artifact turn // that somehow issues it is rejected (the daemon owns no subject to anchor a verdict to). if (mode.kind !== 'review') { return { ok: false, error: 'this is not a review turn; a review verdict may not be submitted' } } if (this.#observation.review || this.#observation.artifact || this.#observation.answer || this.#submitReserved) { return { ok: false, error: 'a review was already submitted for this request' } } // Reserve synchronously before the first await — shared with #submitArtifact so a review turn's // reservation covers both submit paths (concurrent submits can never both land). this.#submitReserved = true const subject = mode.subject // The daemon constructs the record: subject + request come from turn context, NEVER the // envelope, so a review turn structurally cannot review anything but its named subject. const record: ReviewRecord = { $type: COLLECTIONS.review, subject, request: { uri: request.uri, cid: request.cid }, verdict: envelope.verdict, findings: envelope.findings, ...this.#model(), createdAt: this.#now(), } const rkey = reviewRkey(request.uri, request.cid) let ref: StrongRef try { ref = await client.create(COLLECTIONS.review, record, { rkey }) } catch (error) { if (error instanceof XrpcError && error.status === 400 && error.error === 'RecordAlreadyExists') { const existing = await client.getOwnRecord(COLLECTIONS.review, rkey) if (!existing) { // The PDS claims the record exists but it isn't fetchable — a hard error, not an // adoption. Release the reservation to allow a genuine later retry. this.#submitReserved = false throw error } const current = existing.value as ReviewRecord // Adopt iff the deterministic-rkey record was written for THIS same request+subject by this // actor — the deterministic rkey already proves same request+actor. A retried review turn // legitimately produces DIFFERENT findings/verdict, so those are deliberately NOT compared; // only the anchoring subject and request backref must match. A mismatch means the rkey was // somehow claimed for a different subject/request — refuse rather than silently adopt. if ( current.$type !== COLLECTIONS.review || !current.request || current.request.uri !== request.uri || current.request.cid !== request.cid || current.subject.uri !== subject.uri || current.subject.cid !== subject.cid ) { this.#submitReserved = false throw new Error(`refusing to adopt review record at ${rkey} with a mismatched subject/request`) } ref = { uri: existing.uri, cid: existing.cid } } else { this.#submitReserved = false throw error } } this.#observation.review = ref return { ok: true, ref } }}