Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212import { createConnection } from 'node:net'import { TURN_LIMITS, type ReviewFinding, type TurnRpcRequest, type TurnRpcResponse } from '@radial/core'import { artifactTitle } from './commands.js'
function allValues(args: string[], flag: string): string[] { const results: string[] = [] for (let index = 0; index < args.length; index += 1) { if (args[index] === flag && args[index + 1]) results.push(args[index + 1] as string) } return results}
function lastValue(args: string[], flag: string): string | undefined { return allValues(args, flag).at(-1)}
async function readBody(args: string[], readText: (path: string) => Promise<string>): Promise<string> { const inline = lastValue(args, '--body') const path = lastValue(args, '--body-file') if (Boolean(inline) === Boolean(path)) { throw new Error('Provide exactly one of --body or --body-file') } if (inline !== undefined) return inline return readText(path as string)}
/** * Maps CLI args to a turn-mode RPC, enforcing the turn allowlist: inside a turn container there * are no credentials, so only the commands the daemon can safely broker on the agent's behalf are * permitted — one terminal command per turn kind (`artifact submit`, `review submit`, `answer * submit`) plus the can't-complete question (`message post`). Every other command (including auth, * and the direct-PDS `artifact post` / `review post` / etc.) would either need credentials this * process doesn't have or bypass the daemon's anchoring. Which terminal is actually allowed is the * DAEMON's decision, made from turn mode — the sidecar only shapes the envelope. */export async function buildTurnRpc( args: string[], token: string, readText: (path: string) => Promise<string>,): Promise<TurnRpcRequest> { const [group, action] = args
if (group === 'artifact' && action === 'submit') { // Same rule and the same refusals as the direct `artifact post` path: the title is the artifact's // own, so a turn supplies it exactly as a human would. The daemon validates it again at the // socket edge — this is a legible error message, not the trust boundary. const title = artifactTitle(args) const body = await readBody(args, readText) const criteria = allValues(args, '--criterion') // Implementation turns pass the pushed HEAD sha via --commit; the daemon validates its shape and // owns everything else (branch naming, push, PR). Plan-style submits omit it. const commit = lastValue(args, '--commit') const branch = lastValue(args, '--branch') const pr = lastValue(args, '--pr') return { method: 'submitArtifact', token, title, body, ...(criteria.length ? { criteria } : {}), ...(commit !== undefined ? { commit } : {}), ...(branch !== undefined ? { branch } : {}), ...(pr !== undefined ? { pr } : {}), } }
if (group === 'review' && action === 'submit') { // A review turn's terminal command. The daemon owns the subject and the request backref (from // turn context), so `--subject` is rejected here — a review turn cannot redirect its verdict at // a different artifact than the one its request named. if (lastValue(args, '--subject') !== undefined) { throw new Error('In turn socket mode `radial review submit` must not pass --subject; the daemon owns the review subject') } const verdict = lastValue(args, '--verdict') if (verdict !== 'approve' && verdict !== 'request_changes') { throw new Error('radial review submit requires --verdict approve|request_changes') } const findingsFile = lastValue(args, '--findings-file') if (findingsFile === undefined) { throw new Error('radial review submit requires --findings-file <path to a JSON array of findings>') } const raw = await readText(findingsFile) let findings: unknown try { findings = JSON.parse(raw) } catch (error) { throw new Error(`--findings-file is not valid JSON: ${error instanceof Error ? error.message : String(error)}`) } if (!Array.isArray(findings)) { throw new Error('--findings-file must contain a JSON array of findings') } // The daemon performs full per-finding validation at the socket boundary (the untrusted edge); // the sidecar only shapes the envelope. return { method: 'submitReview', token, verdict, findings: findings as ReviewFinding[] } }
if (group === 'answer' && action === 'submit') { // An answer turn's terminal command. The daemon owns every anchoring field — the goal, the // `parent`/`root` threading, the `re` backref, the mention of whoever is being answered — from // turn context, so the envelope carries prose and nothing else. const body = await readBody(args, readText) return { method: 'submitAnswer', token, body } }
if (group === 'message' && action === 'post') { // In socket mode a message is always the can't-complete question; the daemon anchors it to // the turn's goal/artifact, so --goal/--artifact/--re (used in direct-PDS mode) are ignored. // // `--parent` is REFUSED rather than ignored. In direct-PDS mode it is what makes a message a // reply; here the daemon owns anchoring, and silently dropping it would let a turn believe it had // threaded a reply — or answered its own request — when it had posted a bare question. An answer // turn's reply is `radial answer submit`; nothing else threads. if (lastValue(args, '--parent') !== undefined) { throw new Error( 'In turn socket mode `radial message post` must not pass --parent; the daemon owns message anchoring. Use `radial answer submit` to post a reply on an answer turn.', ) } const body = await readBody(args, readText) return { method: 'askQuestion', token, body } }
const command = args.slice(0, 2).filter((part) => part !== undefined).join(' ') || '(none)' throw new Error( `In turn socket mode only "artifact submit", "review submit", "answer submit", and "message post" are permitted, got: ${command}`, )}
function concatBytes(chunks: Uint8Array[]): Uint8Array { const total = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0) const result = new Uint8Array(total) let offset = 0 for (const chunk of chunks) { result.set(chunk, offset) offset += chunk.byteLength } return result}
/** * One RPC per connection: write a newline-delimited JSON frame, read a newline-delimited JSON * frame back, then close. There is no multiplexing and no keep-alive — the CLI process exits * right after, so a fresh connection per invocation is simpler than pooling. */export async function sendTurnRpc( socketPath: string, rpc: TurnRpcRequest, options?: { timeoutMs?: number; maxFrameBytes?: number },): Promise<TurnRpcResponse> { const maxFrameBytes = options?.maxFrameBytes ?? TURN_LIMITS.maxFrameBytes const timeoutMs = options?.timeoutMs ?? TURN_LIMITS.frameTimeoutMs const frame = `${JSON.stringify(rpc)}\n` const frameBytes = new TextEncoder().encode(frame) if (frameBytes.byteLength > maxFrameBytes) { throw new Error(`Turn RPC request frame of ${frameBytes.byteLength} bytes exceeds the ${maxFrameBytes}-byte limit`) }
// `socketPath` is a filesystem path in the production unix-transport case; the dev-only tcp // transport (see daemon's turn.ts / cli.ts fail-closed rule) instead hands us a `tcp://host:port` // URL, since there is no socket file to point at. const connectTarget = socketPath.startsWith('tcp://') ? (() => { const url = new URL(socketPath) return { host: url.hostname, port: Number(url.port) } })() : { path: socketPath }
return new Promise<TurnRpcResponse>((resolve, reject) => { const socket = createConnection(connectTarget) let settled = false let received = 0 const chunks: Uint8Array[] = []
const finish = (fn: () => void): void => { if (settled) return settled = true fn() } const fail = (error: Error): void => { finish(() => { socket.destroy() reject(error) }) }
socket.setTimeout(timeoutMs, () => fail(new Error(`Turn RPC timed out after ${timeoutMs}ms`))) socket.on('error', (error) => fail(error)) socket.on('close', () => fail(new Error('Turn socket closed before a response was received'))) socket.on('connect', () => socket.write(frame)) socket.on('data', (chunk) => { if (settled) return const newlineIndex = chunk.indexOf(0x0a) const upTo = newlineIndex === -1 ? chunk : chunk.subarray(0, newlineIndex) received += upTo.byteLength if (received > maxFrameBytes) { fail(new Error(`Turn RPC response exceeds the ${maxFrameBytes}-byte limit`)) return } chunks.push(upTo) if (newlineIndex === -1) return const text = new TextDecoder().decode(concatBytes(chunks)) finish(() => { socket.destroy() try { resolve(JSON.parse(text) as TurnRpcResponse) } catch (error) { reject(error instanceof Error ? error : new Error(String(error))) } }) }) })}