From 13273c3818d02bb67a0c43db7179a7b85f8e334c Mon Sep 17 00:00:00 2001 From: Tim Disney Date: Sat, 18 Jul 2026 16:56:43 -0700 Subject: [PATCH] implement phase 2 --- package.json | 2 +- packages/atproto/package.json | 21 + packages/atproto/src/client.ts | 488 ++++++++++++++++++++++ packages/atproto/src/index.ts | 1 + packages/atproto/src/node-shims.d.ts | 34 ++ packages/atproto/test/client.test.mjs | 148 +++++++ packages/atproto/tsconfig.json | 10 + packages/daemon/package.json | 22 +- packages/daemon/src/cli.ts | 83 ++++ packages/daemon/src/index.ts | 2 + packages/daemon/src/init.ts | 84 ++++ packages/daemon/src/node-shims.d.ts | 30 ++ packages/daemon/src/runtime.ts | 80 ++++ packages/daemon/test/init.test.mjs | 131 ++++++ packages/daemon/tsconfig.json | 6 + packages/ingest/package.json | 18 +- packages/ingest/src/index.ts | 3 + packages/ingest/src/node-shims.d.ts | 27 ++ packages/ingest/src/poller.ts | 126 ++++++ packages/ingest/src/space.ts | 72 ++++ packages/ingest/src/state.ts | 87 ++++ packages/ingest/test/poller.test.mjs | 75 ++++ packages/ingest/test/space.test.mjs | 75 ++++ packages/ingest/tsconfig.json | 6 + packages/sidecar/README.md | 52 +++ packages/sidecar/package.json | 19 +- packages/sidecar/src/cli.ts | 87 ++++ packages/sidecar/src/commands.ts | 238 +++++++++++ packages/sidecar/src/index.ts | 1 + packages/sidecar/src/node-shims.d.ts | 30 ++ packages/sidecar/test/human-loop.test.mjs | 201 +++++++++ packages/sidecar/tsconfig.json | 6 + pnpm-lock.yaml | 36 +- readme.md | 50 ++- 34 files changed, 2339 insertions(+), 12 deletions(-) create mode 100644 packages/atproto/package.json create mode 100644 packages/atproto/src/client.ts create mode 100644 packages/atproto/src/index.ts create mode 100644 packages/atproto/src/node-shims.d.ts create mode 100644 packages/atproto/test/client.test.mjs create mode 100644 packages/atproto/tsconfig.json create mode 100644 packages/daemon/src/cli.ts create mode 100644 packages/daemon/src/index.ts create mode 100644 packages/daemon/src/init.ts create mode 100644 packages/daemon/src/node-shims.d.ts create mode 100644 packages/daemon/src/runtime.ts create mode 100644 packages/daemon/test/init.test.mjs create mode 100644 packages/daemon/tsconfig.json create mode 100644 packages/ingest/src/index.ts create mode 100644 packages/ingest/src/node-shims.d.ts create mode 100644 packages/ingest/src/poller.ts create mode 100644 packages/ingest/src/space.ts create mode 100644 packages/ingest/src/state.ts create mode 100644 packages/ingest/test/poller.test.mjs create mode 100644 packages/ingest/test/space.test.mjs create mode 100644 packages/ingest/tsconfig.json create mode 100644 packages/sidecar/README.md create mode 100644 packages/sidecar/src/cli.ts create mode 100644 packages/sidecar/src/commands.ts create mode 100644 packages/sidecar/src/index.ts create mode 100644 packages/sidecar/src/node-shims.d.ts create mode 100644 packages/sidecar/test/human-loop.test.mjs create mode 100644 packages/sidecar/tsconfig.json diff --git a/package.json b/package.json index 6bfd342..bb1ad1c 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,7 @@ "build": "pnpm -r build", "codegen": "pnpm --filter @radial/lexicons codegen", "lint": "pnpm -r lint", - "test": "pnpm --filter @radial/core test", + "test": "pnpm -r test", "test:watch": "pnpm --filter @radial/core test:watch", "typecheck": "pnpm -r typecheck" }, diff --git a/packages/atproto/package.json b/packages/atproto/package.json new file mode 100644 index 0000000..4fe009b --- /dev/null +++ b/packages/atproto/package.json @@ -0,0 +1,21 @@ +{ + "name": "@radial/atproto", + "version": "0.0.0", + "private": true, + "type": "module", + "exports": { + ".": { + "types": "./src/index.ts", + "import": "./dist/index.js" + } + }, + "scripts": { + "build": "tsc -p tsconfig.json", + "lint": "pnpm typecheck", + "test": "pnpm build && node --test test/*.test.mjs", + "typecheck": "tsc -p tsconfig.json --noEmit" + }, + "dependencies": { + "@radial/core": "workspace:*" + } +} diff --git a/packages/atproto/src/client.ts b/packages/atproto/src/client.ts new file mode 100644 index 0000000..ea1fd0e --- /dev/null +++ b/packages/atproto/src/client.ts @@ -0,0 +1,488 @@ +import { chmod, lstat, mkdir, readFile, rename, writeFile } from 'node:fs/promises' +import { dirname, join } from 'node:path' +import { DatabaseSync } from 'node:sqlite' +import { + COLLECTIONS, + assertValidRecord, + type Collection, + type RadialRecord, + type RecordByCollection, + type StrongRef, +} from '@radial/core' + +export type FetchLike = (input: string | URL, init?: RequestInit) => Promise + +export interface ActorSession { + did: string + handle: string + service: string + accessJwt: string + refreshJwt: string +} + +export interface StoredActor extends ActorSession { + profile: string + agentProfile?: StrongRef +} + +interface SessionFile { + version: 1 + defaultProfile?: string + profiles: Record +} + +export class XrpcError extends Error { + constructor( + readonly status: number, + readonly error: string, + message: string, + ) { + super(message) + this.name = 'XrpcError' + } +} + +const object = (value: unknown): value is Record => + typeof value === 'object' && value !== null && !Array.isArray(value) + +const sessionQueues = new Map>() + +async function jsonResponse(response: Response): Promise> { + const value = await response.json().catch(() => undefined) + if (!response.ok) { + const body = object(value) ? value : {} + throw new XrpcError( + response.status, + typeof body.error === 'string' ? body.error : 'XrpcError', + typeof body.message === 'string' ? body.message : `XRPC request failed (${response.status})`, + ) + } + if (!object(value)) throw new TypeError('XRPC response must be a JSON object') + return value +} + +const requiredString = (value: Record, key: string): string => { + const result = value[key] + if (typeof result !== 'string' || result.length === 0) { + throw new TypeError(`XRPC response is missing ${key}`) + } + return result +} + +export async function createSession( + service: string | URL, + identifier: string, + password: string, + fetcher: FetchLike = fetch, +): Promise { + const response = await fetcher(new URL('/xrpc/com.atproto.server.createSession', service), { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ identifier, password }), + }) + const value = await jsonResponse(response) + return { + did: requiredString(value, 'did'), + handle: requiredString(value, 'handle'), + service: new URL(service).origin, + accessJwt: requiredString(value, 'accessJwt'), + refreshJwt: requiredString(value, 'refreshJwt'), + } +} + +export class FileSessionStore { + readonly path: string + + constructor(directory = defaultDataDirectory()) { + this.path = join(directory, 'sessions.json') + } + + async #read(): Promise { + try { + const info = await lstat(this.path) + if (info.isSymbolicLink() || !info.isFile()) { + throw new Error(`Session path is not a regular non-symlink file: ${this.path}`) + } + if ((info.mode & 0o077) !== 0) await chmod(this.path, 0o600) + const value = JSON.parse(await readFile(this.path, 'utf8')) as unknown + if (!object(value) || value.version !== 1 || !object(value.profiles)) { + throw new TypeError('Invalid Radial session file') + } + return value as unknown as SessionFile + } catch (error) { + if (object(error) && error.code === 'ENOENT') return { version: 1, profiles: {} } + throw error + } + } + + async get(profile?: string): Promise { + const state = await this.#read() + const selected = profile ?? state.defaultProfile + return selected ? state.profiles[selected] : undefined + } + + async save(actor: StoredActor, makeDefault = true): Promise { + const directory = dirname(this.path) + await mkdir(directory, { recursive: true, mode: 0o700 }) + await chmod(directory, 0o700) + await this.#withLock(async () => { + const state = await this.#read() + state.profiles[actor.profile] = actor + if (makeDefault || !state.defaultProfile) state.defaultProfile = actor.profile + await this.#write(state) + }) + } + + async coordinateRefresh( + profile: string, + current: ActorSession, + refresh: (session: ActorSession) => Promise, + ): Promise { + return this.#withLock(async () => { + const state = await this.#read() + const stored = state.profiles[profile] + if (!stored) throw new Error(`Radial profile not found during refresh: ${profile}`) + if (stored.refreshJwt !== current.refreshJwt) return stored + const rotated = await refresh(stored) + state.profiles[profile] = { ...stored, ...rotated } + await this.#write(state) + return rotated + }) + } + + async #write(state: SessionFile): Promise { + const temporary = `${this.path}.tmp-${Math.random().toString(36).slice(2)}` + await writeFile(temporary, `${JSON.stringify(state, null, 2)}\n`, { mode: 0o600 }) + await chmod(temporary, 0o600) + await rename(temporary, this.path) + await chmod(this.path, 0o600) + } + + async #withLock(operation: () => Promise): Promise { + // A module-local queue prevents a synchronous SQLite wait from blocking an + // async holder in this process. SQLite supplies the cross-process advisory + // lock and releases it automatically if its owner exits or crashes. + const previous = sessionQueues.get(this.path) ?? Promise.resolve() + let release = (): void => {} + const held = new Promise((resolve) => { + release = resolve + }) + const queued = previous.then(() => held) + sessionQueues.set(this.path, queued) + await previous + + const database = new DatabaseSync(`${this.path}.lock.db`) + try { + database.exec(` + PRAGMA busy_timeout = 30000; + CREATE TABLE IF NOT EXISTS session_lock (id INTEGER PRIMARY KEY) STRICT; + BEGIN IMMEDIATE; + `) + return await operation() + } finally { + try { + database.exec('COMMIT') + } finally { + database.close() + release() + if (sessionQueues.get(this.path) === queued) sessionQueues.delete(this.path) + } + } + } +} + +export function defaultDataDirectory(): string { + const env = process.env + if (env.RADIAL_DATA_DIR) return env.RADIAL_DATA_DIR + if (env.XDG_STATE_HOME) return join(env.XDG_STATE_HOME, 'radial') + if (!env.HOME) throw new Error('Set RADIAL_DATA_DIR or HOME') + return join(env.HOME, '.local', 'state', 'radial') +} + +export interface CreateRecordOptions { + rkey?: string +} + +export class CredentialClient { + #session: ActorSession + #refreshing: Promise | undefined + + constructor( + session: ActorSession, + readonly fetcher: FetchLike = fetch, + readonly persist?: (session: ActorSession) => Promise, + readonly coordinateRefresh?: ( + current: ActorSession, + refresh: (session: ActorSession) => Promise, + ) => Promise, + ) { + this.#session = { ...session } + } + + get session(): ActorSession { + return { ...this.#session } + } + + async #refresh(): Promise { + if (this.#refreshing) return this.#refreshing + this.#refreshing = (async () => { + const rotate = async (session: ActorSession): Promise => { + const response = await this.fetcher( + new URL('/xrpc/com.atproto.server.refreshSession', session.service), + { method: 'POST', headers: { authorization: `Bearer ${session.refreshJwt}` } }, + ) + const value = await jsonResponse(response) + const did = requiredString(value, 'did') + if (did !== session.did) throw new Error('Refreshed session DID does not match actor') + return { + ...session, + did, + handle: requiredString(value, 'handle'), + accessJwt: requiredString(value, 'accessJwt'), + refreshJwt: requiredString(value, 'refreshJwt'), + } + } + this.#session = this.coordinateRefresh + ? await this.coordinateRefresh(this.session, rotate) + : await rotate(this.session) + await this.persist?.(this.session) + })().finally(() => { + this.#refreshing = undefined + }) + return this.#refreshing + } + + async #authorized(path: string, init: RequestInit, retry = true): Promise> { + const response = await this.fetcher(new URL(`/xrpc/${path}`, this.#session.service), { + ...init, + headers: { ...init.headers, authorization: `Bearer ${this.#session.accessJwt}` }, + }) + let expired = response.status === 401 + if (!expired && response.status === 400) { + const body = await response.clone().json().catch(() => undefined) + expired = object(body) && (body.error === 'ExpiredToken' || body.error === 'InvalidToken') + } + if (expired && retry) { + await this.#refresh() + return this.#authorized(path, init, false) + } + return jsonResponse(response) + } + + async create( + collection: K, + value: RecordByCollection[K], + options: CreateRecordOptions = {}, + ): Promise { + assertValidRecord(collection, value) + const result = await this.#authorized('com.atproto.repo.createRecord', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + repo: this.#session.did, + collection, + record: value, + validate: false, + ...(options.rkey ? { rkey: options.rkey } : {}), + }), + }) + const uri = requiredString(result, 'uri') + const cid = requiredString(result, 'cid') + if (!uri.startsWith(`at://${this.#session.did}/${collection}/`)) { + throw new Error('PDS returned a record URI for the wrong actor or collection') + } + return { uri, cid } + } + + async getOwnRecord(collection: Collection, rkey: string): Promise { + try { + const query = new URLSearchParams({ repo: this.#session.did, collection, rkey }) + const response = await this.fetcher( + new URL(`/xrpc/com.atproto.repo.getRecord?${query}`, this.#session.service), + ) + const value = await jsonResponse(response) + return decodedRecord(value) + } catch (error) { + if (error instanceof XrpcError && error.status === 400 && error.error === 'RecordNotFound') { + return undefined + } + throw error + } + } +} + +export interface AtUriParts { + did: string + collection: string + rkey: string +} + +export function parseAtUri(uri: string): AtUriParts { + const match = /^at:\/\/(did:[^/]+)\/([^/]+)\/([^/#]+)$/.exec(uri) + if (!match) throw new TypeError(`Invalid record AT URI: ${uri}`) + return { did: match[1] as string, collection: match[2] as string, rkey: match[3] as string } +} + +export interface ResolvedRecord extends StrongRef, AtUriParts { + value: RadialRecord +} + +function decodedRecord(value: Record): ResolvedRecord { + const uri = requiredString(value, 'uri') + const cid = requiredString(value, 'cid') + const parts = parseAtUri(uri) + const record = value.value + assertValidRecord(parts.collection, record) + return { uri, cid, ...parts, value: record } +} + +export type PdsResolver = (did: string, signal?: AbortSignal) => Promise + +export async function resolveDidPds( + did: string, + fetcher: FetchLike = fetch, +): Promise { + let documentUrl: URL + if (did.startsWith('did:plc:')) { + documentUrl = new URL(`https://plc.directory/${encodeURIComponent(did)}`) + } else if (did.startsWith('did:web:')) { + const segments = did.slice('did:web:'.length).split(':').map(decodeURIComponent) + const host = segments.shift() + if (!host) throw new TypeError(`Invalid did:web: ${did}`) + documentUrl = new URL( + segments.length ? `https://${host}/${segments.join('/')}/did.json` : `https://${host}/.well-known/did.json`, + ) + } else { + throw new Error(`Unsupported DID method for automatic PDS resolution: ${did}`) + } + const document = await jsonResponse(await fetcher(documentUrl)) + const services = document.service + if (!Array.isArray(services)) throw new Error(`DID document has no services: ${did}`) + const entry = services.find( + (candidate) => + object(candidate) && + candidate.type === 'AtprotoPersonalDataServer' && + typeof candidate.serviceEndpoint === 'string', + ) as Record | undefined + if (!entry) throw new Error(`DID document has no atproto PDS service: ${did}`) + return new URL(entry.serviceEndpoint as string) +} + +export interface RepoHead { + rev: string + commitCid: string +} + +export interface ListedRecord { + uri: string + cid: string + value: unknown +} + +export interface RepoReadTransport { + resolvePds(did: string, signal?: AbortSignal): Promise + getLatestCommit(did: string, signal?: AbortSignal): Promise + listRecords(input: { + did: string + collection: string + cursor?: string + limit: number + signal?: AbortSignal + }): Promise<{ records: ListedRecord[]; cursor?: string }> + getRecord(uri: string, signal?: AbortSignal): Promise +} + +export class FetchRepoTransport implements RepoReadTransport { + constructor( + readonly resolver: PdsResolver = (did, signal) => + resolveDidPds(did, (input, init) => + fetch(input, signal ? { ...init, signal } : init), + ), + readonly fetcher: FetchLike = fetch, + ) {} + + async resolvePds(did: string, signal?: AbortSignal): Promise { + return new URL(await this.resolver(did, signal)) + } + + async getLatestCommit(did: string, signal?: AbortSignal): Promise { + const service = await this.resolvePds(did, signal) + const query = new URLSearchParams({ did }) + const value = await jsonResponse( + await this.fetcher( + new URL(`/xrpc/com.atproto.sync.getLatestCommit?${query}`, service), + signal ? { signal } : undefined, + ), + ) + return { rev: requiredString(value, 'rev'), commitCid: requiredString(value, 'cid') } + } + + async listRecords(input: { + did: string + collection: string + cursor?: string + limit: number + signal?: AbortSignal + }): Promise<{ records: ListedRecord[]; cursor?: string }> { + const service = await this.resolvePds(input.did, input.signal) + const query = new URLSearchParams({ + repo: input.did, + collection: input.collection, + limit: String(input.limit), + ...(input.cursor ? { cursor: input.cursor } : {}), + }) + const value = await jsonResponse( + await this.fetcher( + new URL(`/xrpc/com.atproto.repo.listRecords?${query}`, service), + input.signal ? { signal: input.signal } : undefined, + ), + ) + if (!Array.isArray(value.records)) throw new TypeError('listRecords response is missing records') + const records = value.records.map((entry) => { + if (!object(entry)) throw new TypeError('listRecords returned an invalid record') + return { + uri: requiredString(entry, 'uri'), + cid: requiredString(entry, 'cid'), + value: entry.value, + } + }) + return { + records, + ...(typeof value.cursor === 'string' ? { cursor: value.cursor } : {}), + } + } + + async getRecord(uri: string, signal?: AbortSignal): Promise { + const parts = parseAtUri(uri) + const service = await this.resolvePds(parts.did, signal) + const query = new URLSearchParams({ repo: parts.did, collection: parts.collection, rkey: parts.rkey }) + return decodedRecord( + await jsonResponse( + await this.fetcher( + new URL(`/xrpc/com.atproto.repo.getRecord?${query}`, service), + signal ? { signal } : undefined, + ), + ), + ) + } +} + +export class StrongRefResolver { + constructor(readonly transport: RepoReadTransport) {} + + async resolve(locator: string, expectedCollection?: Collection): Promise { + const hash = locator.lastIndexOf('#') + const uri = hash > 'at://'.length ? locator.slice(0, hash) : locator + const expectedCid = hash > 'at://'.length ? locator.slice(hash + 1) : undefined + const result = await this.transport.getRecord(uri) + if (expectedCid && result.cid !== expectedCid) { + throw new Error(`Record CID mismatch for ${uri}: expected ${expectedCid}, got ${result.cid}`) + } + if (expectedCollection && result.collection !== expectedCollection) { + throw new Error(`Expected ${expectedCollection}, got ${result.collection}`) + } + return result + } +} + +export const radialCollections = Object.values(COLLECTIONS).sort() diff --git a/packages/atproto/src/index.ts b/packages/atproto/src/index.ts new file mode 100644 index 0000000..9dfc2ec --- /dev/null +++ b/packages/atproto/src/index.ts @@ -0,0 +1 @@ +export * from './client.js' diff --git a/packages/atproto/src/node-shims.d.ts b/packages/atproto/src/node-shims.d.ts new file mode 100644 index 0000000..13db842 --- /dev/null +++ b/packages/atproto/src/node-shims.d.ts @@ -0,0 +1,34 @@ +declare module 'node:fs/promises' { + export function chmod(path: string, mode: number): Promise + export function lstat(path: string): Promise<{ isFile(): boolean; isSymbolicLink(): boolean; mode: number }> + export function mkdir(path: string, options?: { recursive?: boolean; mode?: number }): Promise + export function open(path: string, flags: string, mode?: number): Promise<{ close(): Promise; writeFile(data: string): Promise }> + export function readFile(path: string, encoding: string): Promise + export function rename(oldPath: string, newPath: string): Promise + export function rm(path: string, options?: { force?: boolean }): Promise + export function writeFile(path: string, data: string, options?: { mode?: number }): Promise +} + +declare module 'node:path' { + export function dirname(path: string): string + export function join(...paths: string[]): string +} + +declare module 'node:sqlite' { + export class DatabaseSync { + constructor(path: string) + exec(sql: string): void + prepare(sql: string): { + get(...values: unknown[]): unknown + all(...values: unknown[]): unknown[] + run(...values: unknown[]): unknown + } + close(): void + } +} + +declare const process: { + env: Record + pid: number + kill(pid: number, signal: number): void +} diff --git a/packages/atproto/test/client.test.mjs b/packages/atproto/test/client.test.mjs new file mode 100644 index 0000000..048ac11 --- /dev/null +++ b/packages/atproto/test/client.test.mjs @@ -0,0 +1,148 @@ +import assert from 'node:assert/strict' +import { mkdtemp, readFile, stat } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, it } from 'node:test' +import { COLLECTIONS } from '../../core/dist/index.js' +import { + CredentialClient, + FileSessionStore, + XrpcError, + createSession, +} from '../dist/index.js' + +const json = (value, status = 200) => + new Response(JSON.stringify(value), { status, headers: { 'content-type': 'application/json' } }) + +describe('atproto credentials and writes', () => { + it('logs in without retaining the app password', async () => { + let body + const session = await createSession('https://pds.test', 'alice.test', 'secret', async (_url, init) => { + body = JSON.parse(init.body) + return json({ did: 'did:plc:alice', handle: 'alice.test', accessJwt: 'a', refreshJwt: 'r' }) + }) + assert.deepEqual(body, { identifier: 'alice.test', password: 'secret' }) + assert.equal(JSON.stringify(session).includes('secret'), false) + }) + + it('validates locally, refreshes once, and writes to the authenticated repo', async () => { + const calls = [] + let persisted + const client = new CredentialClient( + { + did: 'did:plc:alice', + handle: 'alice.test', + service: 'https://pds.test', + accessJwt: 'old-access', + refreshJwt: 'old-refresh', + }, + async (url, init) => { + calls.push({ url: String(url), init }) + if (String(url).endsWith('refreshSession')) { + return json({ + did: 'did:plc:alice', + handle: 'alice.test', + accessJwt: 'new-access', + refreshJwt: 'new-refresh', + }) + } + if (calls.filter((call) => call.url.endsWith('createRecord')).length === 1) { + return json({ error: 'ExpiredToken', message: 'expired' }, 401) + } + return json({ + uri: `at://did:plc:alice/${COLLECTIONS.space}/one`, + cid: 'cid-one', + }) + }, + async (session) => { + persisted = session + }, + ) + const record = { + $type: COLLECTIONS.space, + name: 'Radial', + description: 'test', + createdAt: '2026-07-18T00:00:00Z', + } + assert.deepEqual(await client.create(COLLECTIONS.space, record), { + uri: `at://did:plc:alice/${COLLECTIONS.space}/one`, + cid: 'cid-one', + }) + assert.equal(persisted.refreshJwt, 'new-refresh') + const write = calls.filter((call) => call.url.endsWith('createRecord')).at(-1) + assert.equal(write.init.headers.authorization, 'Bearer new-access') + assert.equal(JSON.parse(write.init.body).validate, false) + assert.equal(JSON.parse(write.init.body).repo, 'did:plc:alice') + }) + + it('rejects invalid records before fetching', async () => { + let fetched = false + const client = new CredentialClient( + { did: 'did:plc:a', handle: 'a.test', service: 'https://pds.test', accessJwt: 'a', refreshJwt: 'r' }, + async () => { + fetched = true + return json({}) + }, + ) + await assert.rejects( + client.create(COLLECTIONS.space, { $type: COLLECTIONS.space, name: 'missing fields' }), + /Invalid/, + ) + assert.equal(fetched, false) + }) + + it('persists sessions with owner-only permissions', async () => { + const directory = await mkdtemp(join(tmpdir(), 'radial-session-')) + const store = new FileSessionStore(directory) + await store.save({ + profile: 'alice', + did: 'did:plc:alice', + handle: 'alice.test', + service: 'https://pds.test', + accessJwt: 'access', + refreshJwt: 'refresh', + }) + assert.equal((await stat(directory)).mode & 0o777, 0o700) + assert.equal((await stat(store.path)).mode & 0o777, 0o600) + assert.equal((await store.get()).did, 'did:plc:alice') + assert.equal((await readFile(store.path, 'utf8')).includes('refresh'), true) + }) + + it('serializes profile saves and cross-process refresh rotation', async () => { + const directory = await mkdtemp(join(tmpdir(), 'radial-session-lock-')) + const first = new FileSessionStore(directory) + const second = new FileSessionStore(directory) + const actor = (profile) => ({ + profile, + did: `did:plc:${profile}`, + handle: `${profile}.test`, + service: 'https://pds.test', + accessJwt: `${profile}-access`, + refreshJwt: `${profile}-refresh`, + }) + await Promise.all([first.save(actor('alice')), second.save(actor('bob'), false)]) + assert.equal((await first.get('alice')).did, 'did:plc:alice') + assert.equal((await first.get('bob')).did, 'did:plc:bob') + + let refreshes = 0 + const current = await first.get('alice') + const rotate = async (session) => { + refreshes += 1 + return { ...session, accessJwt: 'rotated-access', refreshJwt: 'rotated-refresh' } + } + const [left, right] = await Promise.all([ + first.coordinateRefresh('alice', current, rotate), + second.coordinateRefresh('alice', current, rotate), + ]) + assert.equal(refreshes, 1) + assert.equal(left.refreshJwt, 'rotated-refresh') + assert.equal(right.refreshJwt, 'rotated-refresh') + }) + + it('surfaces structured XRPC errors without response dumps', async () => { + await assert.rejects( + createSession('https://pds.test', 'a', 'b', async () => json({ error: 'AuthRequired', message: 'bad login' }, 401)), + (error) => error instanceof XrpcError && error.error === 'AuthRequired' && error.message === 'bad login', + ) + }) +}) diff --git a/packages/atproto/tsconfig.json b/packages/atproto/tsconfig.json new file mode 100644 index 0000000..4d9c612 --- /dev/null +++ b/packages/atproto/tsconfig.json @@ -0,0 +1,10 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "types": [] + }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "test"] +} diff --git a/packages/daemon/package.json b/packages/daemon/package.json index 2516bd1..4d539ba 100644 --- a/packages/daemon/package.json +++ b/packages/daemon/package.json @@ -1 +1,21 @@ -{"name":"@radial/daemon","version":"0.0.0","private":true,"type":"module","scripts":{"build":"true","lint":"true"}} +{ + "name": "@radial/daemon", + "version": "0.0.0", + "private": true, + "type": "module", + "bin": { "radiald": "./dist/cli.js" }, + "exports": { + ".": { "types": "./src/index.ts", "import": "./dist/index.js" } + }, + "scripts": { + "build": "tsc -p tsconfig.json", + "lint": "pnpm typecheck", + "test": "pnpm build && node --test test/*.test.mjs", + "typecheck": "tsc -p tsconfig.json --noEmit" + }, + "dependencies": { + "@radial/atproto": "workspace:*", + "@radial/core": "workspace:*", + "@radial/ingest": "workspace:*" + } +} diff --git a/packages/daemon/src/cli.ts b/packages/daemon/src/cli.ts new file mode 100644 index 0000000..6c40790 --- /dev/null +++ b/packages/daemon/src/cli.ts @@ -0,0 +1,83 @@ +#!/usr/bin/env node + +import { argv, stdin } from 'node:process' +import { pathToFileURL } from 'node:url' +import { initializeAgent } from './init.js' +import { indexSummary, readIndex, runSyncLoop, syncSpaceOnce } from './runtime.js' + +const usage = `Usage: + radiald init --pds --identifier + --name --harness [--model ]... + --artifact-type ... --password-stdin + radiald sync --records --checkpoints + [--watch] [--interval ] + radiald index --records ` + +function values(args: string[], flag: string): string[] { + const found: string[] = [] + for (let index = 0; index < args.length; index += 1) { + if (args[index] === flag && args[index + 1]) found.push(args[index + 1] as string) + } + return found +} + +function one(args: string[], flag: string): string { + const found = values(args, flag)[0] + if (!found) throw new Error(`Missing ${flag}\n${usage}`) + return found +} + +async function readStdin(): Promise { + let value = '' + for await (const chunk of stdin) value += new TextDecoder().decode(chunk) + return value.trim() +} + +export async function main(args = argv.slice(2)): Promise { + if (args[0] === 'sync' && args[1]) { + const paths = { + records: one(args, '--records'), + checkpoints: one(args, '--checkpoints'), + } + const interval = Number(values(args, '--interval')[0] ?? 5_000) + if (!Number.isFinite(interval) || interval < 100) throw new Error('--interval must be at least 100ms') + await runSyncLoop({ + watch: args.includes('--watch'), + intervalMs: interval, + sync: () => syncSpaceOnce(args[1] as string, paths), + onIndex: (index) => console.log(JSON.stringify(indexSummary(index))), + onError: (error) => console.error(error instanceof Error ? error.message : error), + }) + return + } + if (args[0] === 'index' && args[1]) { + console.log(JSON.stringify(indexSummary(readIndex(args[1], one(args, '--records'))))) + return + } + if (args[0] !== 'init' || !args[1] || !args.includes('--password-stdin')) throw new Error(usage) + const models = values(args, '--model').map((entry) => { + const separator = entry.indexOf('=') + return separator === -1 + ? { name: entry, costHint: '' } + : { name: entry.slice(0, separator), costHint: entry.slice(separator + 1) } + }) + const result = await initializeAgent({ + profile: args[1], + service: one(args, '--pds'), + identifier: one(args, '--identifier'), + password: await readStdin(), + handleName: one(args, '--name'), + harness: one(args, '--harness'), + models, + artifactTypes: values(args, '--artifact-type'), + }) + console.log(JSON.stringify(result)) +} + +const entry = argv[1] +if (entry && import.meta.url === pathToFileURL(entry).href) { + main().catch((error: unknown) => { + console.error(error instanceof Error ? error.message : error) + process.exitCode = 1 + }) +} diff --git a/packages/daemon/src/index.ts b/packages/daemon/src/index.ts new file mode 100644 index 0000000..799077a --- /dev/null +++ b/packages/daemon/src/index.ts @@ -0,0 +1,2 @@ +export * from './init.js' +export * from './runtime.js' diff --git a/packages/daemon/src/init.ts b/packages/daemon/src/init.ts new file mode 100644 index 0000000..00bd0ba --- /dev/null +++ b/packages/daemon/src/init.ts @@ -0,0 +1,84 @@ +import { + CredentialClient, + FileSessionStore, + createSession, + type ActorSession, + type FetchLike, + type StoredActor, +} from '@radial/atproto' +import { COLLECTIONS, type AgentModel, type AgentRecord, type StrongRef } from '@radial/core' + +export interface AgentInitInput { + profile: string + service: string + identifier: string + password: string + handleName: string + harness: string + models: AgentModel[] + artifactTypes: string[] + now?: string +} + +export interface AgentInitResult { + did: string + handle: string + service: string + profile: StrongRef +} + +export function buildAgentRecord(input: AgentInitInput): AgentRecord { + return { + $type: COLLECTIONS.agent, + handleName: input.handleName, + harness: input.harness, + models: input.models, + artifactTypes: input.artifactTypes, + createdAt: input.now ?? new Date().toISOString(), + } +} + +export async function initializeAgent( + input: AgentInitInput, + options: { store?: FileSessionStore; fetcher?: FetchLike } = {}, +): Promise { + const store = options.store ?? new FileSessionStore() + const session = await createSession(input.service, input.identifier, input.password, options.fetcher) + if (input.identifier.startsWith('did:') && input.identifier !== session.did) { + throw new Error(`Authenticated DID ${session.did} does not match ${input.identifier}`) + } + let pending = session + const client = new CredentialClient(session, options.fetcher, async (rotated) => { + pending = rotated + }) + const existing = await client.getOwnRecord(COLLECTIONS.agent, 'self') + const record = buildAgentRecord(input) + let reference: StrongRef + if (existing) { + const comparable = (value: AgentRecord): string => + JSON.stringify({ + handleName: value.handleName, + harness: value.harness, + models: value.models, + artifactTypes: value.artifactTypes, + }) + if (comparable(existing.value as AgentRecord) !== comparable(record)) { + throw new Error('Agent profile self already exists with different capabilities') + } + reference = { uri: existing.uri, cid: existing.cid } + } else { + reference = await client.create(COLLECTIONS.agent, record, { rkey: 'self' }) + } + const stored: StoredActor = { + ...(pending as ActorSession), + profile: input.profile, + agentProfile: reference, + } + await store.save(stored) + return { + did: stored.did, + handle: stored.handle, + service: stored.service, + profile: reference, + } +} diff --git a/packages/daemon/src/node-shims.d.ts b/packages/daemon/src/node-shims.d.ts new file mode 100644 index 0000000..93a4048 --- /dev/null +++ b/packages/daemon/src/node-shims.d.ts @@ -0,0 +1,30 @@ +declare module 'node:process' { + export const argv: string[] + export const stdin: AsyncIterable & { isTTY?: boolean } +} +declare module 'node:url' { + export function pathToFileURL(path: string): URL +} +declare module 'node:sqlite' { + export class DatabaseSync { + constructor(path: string) + exec(sql: string): void + prepare(sql: string): { get(...values: unknown[]): unknown; all(...values: unknown[]): unknown[]; run(...values: unknown[]): unknown } + close(): void + } +} +declare module 'node:fs/promises' { + export function chmod(path: string, mode: number): Promise + export function lstat(path: string): Promise<{ isFile(): boolean; isSymbolicLink(): boolean; mode: number }> + export function mkdir(path: string, options?: { recursive?: boolean; mode?: number }): Promise + export function open(path: string, flags: string, mode?: number): Promise<{ close(): Promise; writeFile(data: string): Promise }> + export function readFile(path: string, encoding: string): Promise + export function rename(oldPath: string, newPath: string): Promise + export function rm(path: string, options?: { force?: boolean }): Promise + export function writeFile(path: string, data: string, options?: { mode?: number }): Promise +} +declare module 'node:path' { + export function dirname(path: string): string + export function join(...paths: string[]): string +} +declare const process: { argv: string[]; env: Record; exitCode?: number; pid: number; kill(pid: number, signal: number): void } diff --git a/packages/daemon/src/runtime.ts b/packages/daemon/src/runtime.ts new file mode 100644 index 0000000..6f79945 --- /dev/null +++ b/packages/daemon/src/runtime.ts @@ -0,0 +1,80 @@ +import { FetchRepoTransport, type RepoReadTransport } from '@radial/atproto' +import { SqliteRecordStore, materialize, type MaterializedIndex } from '@radial/core' +import { RepoPoller, SpaceIngestor, SqliteSyncStateStore } from '@radial/ingest' + +export interface DaemonIndexPaths { + records: string + checkpoints: string +} + +export async function syncSpaceOnce( + spaceUri: string, + paths: DaemonIndexPaths, + transport: RepoReadTransport = new FetchRepoTransport(), +): Promise { + const records = new SqliteRecordStore(paths.records) + const checkpoints = new SqliteSyncStateStore(paths.checkpoints) + try { + const poller = new RepoPoller(transport, records, checkpoints) + return await new SpaceIngestor(spaceUri, poller, records).sync() + } finally { + checkpoints.close() + records.close() + } +} + +export function readIndex(spaceUri: string, recordsPath: string): MaterializedIndex { + const records = new SqliteRecordStore(recordsPath) + try { + return materialize(records, { spaceUri }) + } finally { + records.close() + } +} + +export function indexSummary(index: MaterializedIndex): object { + return { + space: { uri: index.space.uri, cid: index.space.cid, name: index.space.value.name }, + members: index.members, + artifactTypes: index.artifactTypes.map((entry) => entry.value.name), + projects: index.projects.map((view) => ({ + uri: view.target.uri, + name: view.target.value.name, + requests: view.requests.length, + openRequests: view.openRequests.length, + artifacts: view.artifacts.length, + })), + goals: index.goals.map((view) => ({ + uri: view.target.uri, + title: view.target.value.title, + requests: view.requests.length, + openRequests: view.openRequests.length, + artifacts: view.artifacts.length, + reviews: view.reviews.length, + })), + ignored: index.ignored, + edits: index.edits, + } +} + +export async function runSyncLoop(options: { + watch: boolean + intervalMs: number + sync: () => Promise + onIndex: (index: MaterializedIndex) => void + onError: (error: unknown) => void + sleep?: (milliseconds: number) => Promise + continueWatching?: () => boolean +}): Promise { + const sleep = options.sleep ?? ((milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds))) + do { + try { + options.onIndex(await options.sync()) + } catch (error) { + if (!options.watch) throw error + options.onError(error) + } + if (!options.watch || options.continueWatching?.() === false) return + await sleep(options.intervalMs) + } while (true) +} diff --git a/packages/daemon/test/init.test.mjs b/packages/daemon/test/init.test.mjs new file mode 100644 index 0000000..c1abf54 --- /dev/null +++ b/packages/daemon/test/init.test.mjs @@ -0,0 +1,131 @@ +import assert from 'node:assert/strict' +import { mkdtemp } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { it } from 'node:test' +import { validateRecord, COLLECTIONS } from '../../core/dist/index.js' +import { FileSessionStore } from '../../atproto/dist/index.js' +import { buildAgentRecord, initializeAgent, runSyncLoop, syncSpaceOnce } from '../dist/index.js' + +it('builds a valid published agent profile', () => { + const record = buildAgentRecord({ + profile: 'planner', + service: 'https://pds.test', + identifier: 'agent.test', + password: 'never serialized', + handleName: 'planner', + harness: 'codex', + models: [{ name: 'gpt-5', costHint: 'high' }], + artifactTypes: ['plan', 'review'], + now: '2026-07-18T00:00:00Z', + }) + assert.equal(validateRecord(COLLECTIONS.agent, record).success, true) + assert.equal(JSON.stringify(record).includes('never serialized'), false) +}) + +it('registers an existing account, publishes self, and persists only the session', async () => { + const directory = await mkdtemp(join(tmpdir(), 'radial-daemon-')) + const store = new FileSessionStore(directory) + const calls = [] + const result = await initializeAgent( + { + profile: 'planner', + service: 'https://pds.test', + identifier: 'planner.test', + password: 'app-password', + handleName: 'planner', + harness: 'codex', + models: [{ name: 'gpt-5', costHint: 'high' }], + artifactTypes: ['plan'], + now: '2026-07-18T00:00:00Z', + }, + { + store, + fetcher: async (url, init = {}) => { + calls.push({ url: String(url), init }) + if (String(url).endsWith('createSession')) { + return Response.json({ + did: 'did:plc:planner', + handle: 'planner.test', + accessJwt: 'access', + refreshJwt: 'refresh', + }) + } + if (String(url).includes('getRecord')) { + return Response.json({ error: 'RecordNotFound', message: 'missing' }, { status: 400 }) + } + const body = JSON.parse(init.body) + assert.equal(body.repo, 'did:plc:planner') + assert.equal(body.collection, COLLECTIONS.agent) + assert.equal(body.rkey, 'self') + assert.equal(body.validate, false) + assert.equal(validateRecord(COLLECTIONS.agent, body.record).success, true) + return Response.json({ + uri: `at://did:plc:planner/${COLLECTIONS.agent}/self`, + cid: 'cid-profile', + }) + }, + }, + ) + assert.equal(result.did, 'did:plc:planner') + assert.equal((await store.get('planner')).refreshJwt, 'refresh') + assert.equal(JSON.stringify(await store.get('planner')).includes('app-password'), false) + assert.equal(calls.some((call) => call.url.includes('createAccount')), false) +}) + +it('runs a durable daemon ingestion/materializer pass', async () => { + const directory = await mkdtemp(join(tmpdir(), 'radial-runtime-')) + const did = 'did:plc:root' + const spaceUri = `at://${did}/${COLLECTIONS.space}/space` + const transport = { + async resolvePds() { return new URL('https://pds.test') }, + async getLatestCommit() { return { rev: '0001', commitCid: 'head-1' } }, + async listRecords({ collection }) { + return { + records: + collection === COLLECTIONS.space + ? [{ + uri: spaceUri, + cid: 'cid-space', + value: { + $type: COLLECTIONS.space, + name: 'Runtime', + description: 'daemon test', + createdAt: '2026-07-18T00:00:00Z', + }, + }] + : [], + } + }, + async getRecord() { throw new Error('unused') }, + } + const index = await syncSpaceOnce( + spaceUri, + { records: join(directory, 'records.db'), checkpoints: join(directory, 'sync.db') }, + transport, + ) + assert.equal(index.space.uri, spaceUri) + assert.deepEqual(index.members.map((member) => member.did), [did]) +}) + +it('keeps watch mode alive across a transient sync failure', async () => { + let attempts = 0 + const indexes = [] + const errors = [] + await runSyncLoop({ + watch: true, + intervalMs: 1, + sync: async () => { + attempts += 1 + if (attempts === 1) throw new Error('temporary PDS failure') + return { space: { uri: 'at://did:plc:root/space/one' } } + }, + onIndex: (index) => indexes.push(index), + onError: (error) => errors.push(error.message), + sleep: async () => {}, + continueWatching: () => attempts < 2, + }) + assert.equal(attempts, 2) + assert.deepEqual(errors, ['temporary PDS failure']) + assert.equal(indexes.length, 1) +}) diff --git a/packages/daemon/tsconfig.json b/packages/daemon/tsconfig.json new file mode 100644 index 0000000..bc30e7a --- /dev/null +++ b/packages/daemon/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "outDir": "dist", "rootDir": "src", "types": [] }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "test"] +} diff --git a/packages/ingest/package.json b/packages/ingest/package.json index eb9d165..904ce54 100644 --- a/packages/ingest/package.json +++ b/packages/ingest/package.json @@ -1 +1,17 @@ -{"name":"@radial/ingest","version":"0.0.0","private":true,"type":"module","scripts":{"build":"true","lint":"true"}} +{ + "name": "@radial/ingest", + "version": "0.0.0", + "private": true, + "type": "module", + "exports": { ".": { "types": "./src/index.ts", "import": "./dist/index.js" } }, + "scripts": { + "build": "tsc -p tsconfig.json", + "lint": "pnpm typecheck", + "test": "pnpm build && node --test test/*.test.mjs", + "typecheck": "tsc -p tsconfig.json --noEmit" + }, + "dependencies": { + "@radial/atproto": "workspace:*", + "@radial/core": "workspace:*" + } +} diff --git a/packages/ingest/src/index.ts b/packages/ingest/src/index.ts new file mode 100644 index 0000000..19d02bf --- /dev/null +++ b/packages/ingest/src/index.ts @@ -0,0 +1,3 @@ +export * from './poller.js' +export * from './space.js' +export * from './state.js' diff --git a/packages/ingest/src/node-shims.d.ts b/packages/ingest/src/node-shims.d.ts new file mode 100644 index 0000000..66e1363 --- /dev/null +++ b/packages/ingest/src/node-shims.d.ts @@ -0,0 +1,27 @@ +declare module 'node:sqlite' { + export class DatabaseSync { + constructor(path: string) + exec(sql: string): void + prepare(sql: string): { + get(...values: unknown[]): unknown + all(...values: unknown[]): unknown[] + run(...values: unknown[]): unknown + } + close(): void + } +} +declare module 'node:fs/promises' { + export function chmod(path: string, mode: number): Promise + export function lstat(path: string): Promise<{ isFile(): boolean; isSymbolicLink(): boolean; mode: number }> + export function mkdir(path: string, options?: { recursive?: boolean; mode?: number }): Promise + export function open(path: string, flags: string, mode?: number): Promise<{ close(): Promise; writeFile(data: string): Promise }> + export function readFile(path: string, encoding: string): Promise + export function rename(oldPath: string, newPath: string): Promise + export function rm(path: string, options?: { force?: boolean }): Promise + export function writeFile(path: string, data: string, options?: { mode?: number }): Promise +} +declare module 'node:path' { + export function dirname(path: string): string + export function join(...paths: string[]): string +} +declare const process: { env: Record; pid: number; kill(pid: number, signal: number): void } diff --git a/packages/ingest/src/poller.ts b/packages/ingest/src/poller.ts new file mode 100644 index 0000000..56f2c58 --- /dev/null +++ b/packages/ingest/src/poller.ts @@ -0,0 +1,126 @@ +import { + parseAtUri, + radialCollections, + type RepoHead, + type RepoReadTransport, +} from '@radial/atproto' +import { validateRecord, type RecordStore, type StoredRecord } from '@radial/core' +import { MemorySyncStateStore, type SyncStateStore } from './state.js' + +export interface PollResult { + did: string + head: RepoHead + changed: boolean + records: number + rejected: Array<{ uri: string; reason: string }> +} + +export interface RepoPollerOptions { + pageSize?: number + maxAttempts?: number + now?: () => string +} + +const sameHead = (left: RepoHead, right: RepoHead): boolean => + left.rev === right.rev && left.commitCid === right.commitCid + +export class RepoPoller { + readonly #running = new Map>() + readonly pageSize: number + readonly maxAttempts: number + readonly now: () => string + + constructor( + readonly transport: RepoReadTransport, + readonly records: RecordStore, + readonly state: SyncStateStore = new MemorySyncStateStore(), + options: RepoPollerOptions = {}, + ) { + this.pageSize = options.pageSize ?? 100 + this.maxAttempts = options.maxAttempts ?? 3 + this.now = options.now ?? (() => new Date().toISOString()) + } + + pollDid(did: string, signal?: AbortSignal): Promise { + const active = this.#running.get(did) + if (active) return active + const operation = this.#poll(did, signal).finally(() => this.#running.delete(did)) + this.#running.set(did, operation) + return operation + } + + async #poll(did: string, signal?: AbortSignal): Promise { + for (let attempt = 1; attempt <= this.maxAttempts; attempt += 1) { + const start = await this.transport.getLatestCommit(did, signal) + const previous = this.state.get(did) + if (previous && sameHead(start, previous)) { + return { did, head: start, changed: false, records: 0, rejected: [] } + } + if (previous) { + if (start.rev < previous.rev) throw new Error(`Repository revision moved backwards for ${did}`) + if (start.rev === previous.rev && start.commitCid !== previous.commitCid) { + throw new Error(`Repository commit changed without a new revision for ${did}`) + } + } + + const staged: StoredRecord[] = [] + const rejected: PollResult['rejected'] = [] + for (const collection of radialCollections) { + let cursor: string | undefined + const seen = new Set() + do { + const page = await this.transport.listRecords({ + did, + collection, + limit: this.pageSize, + ...(cursor ? { cursor } : {}), + ...(signal ? { signal } : {}), + }) + for (const entry of page.records) { + const parts = parseAtUri(entry.uri) + if (parts.did !== did || parts.collection !== collection) { + throw new Error(`listRecords returned an envelope outside ${did}/${collection}`) + } + const validation = validateRecord(collection, entry.value) + if (!validation.success) { + rejected.push({ + uri: entry.uri, + reason: validation.issues + .map((issue) => `${issue.path}: ${issue.message}`) + .join('; '), + }) + continue + } + staged.push({ + did, + collection, + rkey: parts.rkey, + uri: entry.uri, + cid: entry.cid, + rev: start.rev, + value: validation.value, + }) + } + if (page.cursor && seen.has(page.cursor)) throw new Error('listRecords repeated a cursor') + if (page.cursor) seen.add(page.cursor) + cursor = page.cursor + } while (cursor) + } + + const end = await this.transport.getLatestCommit(did, signal) + if (!sameHead(start, end)) { + if (attempt === this.maxAttempts) throw new Error(`Repository changed during scan for ${did}`) + continue + } + for (const record of staged) this.records.put(record) + this.state.set({ + did, + ...start, + firstObservedRev: previous?.firstObservedRev ?? start.rev, + updatedAt: this.now(), + }) + return { did, head: start, changed: true, records: staged.length, rejected } + } + throw new Error(`Unable to obtain a stable snapshot for ${did}`) + } +} diff --git a/packages/ingest/src/space.ts b/packages/ingest/src/space.ts new file mode 100644 index 0000000..28f9048 --- /dev/null +++ b/packages/ingest/src/space.ts @@ -0,0 +1,72 @@ +import { parseAtUri } from '@radial/atproto' +import { materialize, type MaterializedIndex, type RecordStore } from '@radial/core' +import { RepoPoller } from './poller.js' + +export class SpaceIngestor { + readonly spaceDid: string + readonly knownDids = new Set() + + constructor( + readonly spaceUri: string, + readonly poller: RepoPoller, + readonly records: RecordStore, + ) { + this.spaceDid = parseAtUri(spaceUri).did + this.knownDids.add(this.spaceDid) + } + + async sync(signal?: AbortSignal): Promise { + await this.poller.pollDid(this.spaceDid, signal) + let index = materialize(this.records, { spaceUri: this.spaceUri }) + let changed = true + while (changed) { + changed = false + const discovered = index.members + .map((member) => member.did) + .filter((did) => !this.knownDids.has(did)) + const role = new Map(index.members.map((member) => [member.did, member.role])) + discovered.sort((left, right) => { + const leftAdmin = role.get(left) === 'admin' ? 0 : 1 + const rightAdmin = role.get(right) === 'admin' ? 0 : 1 + return leftAdmin - rightAdmin || left.localeCompare(right) + }) + for (const did of discovered) { + this.knownDids.add(did) + await this.poller.pollDid(did, signal) + changed = true + } + if (changed) index = materialize(this.records, { spaceUri: this.spaceUri }) + } + // Poll removed members too. A removal event and the member's last valid + // writes can arrive in the same sync; skipping that final scan would make + // records at or before the published cutoff disappear locally. + await Promise.all(index.members.map((member) => this.poller.pollDid(member.did, signal))) + return materialize(this.records, { spaceUri: this.spaceUri }) + } +} + +export function startPolling( + ingestor: SpaceIngestor, + intervalMs: number, + onError: (error: unknown) => void = () => {}, +): { stop(): void } { + let stopped = false + let timer: ReturnType | undefined + const controller = new AbortController() + const tick = async (): Promise => { + try { + await ingestor.sync(controller.signal) + } catch (error) { + if (!stopped) onError(error) + } + if (!stopped) timer = setTimeout(tick, intervalMs) + } + void tick() + return { + stop() { + stopped = true + controller.abort() + if (timer) clearTimeout(timer) + }, + } +} diff --git a/packages/ingest/src/state.ts b/packages/ingest/src/state.ts new file mode 100644 index 0000000..7cabb09 --- /dev/null +++ b/packages/ingest/src/state.ts @@ -0,0 +1,87 @@ +import { DatabaseSync } from 'node:sqlite' +import type { RepoHead } from '@radial/atproto' + +export interface RepoCheckpoint extends RepoHead { + did: string + firstObservedRev: string + updatedAt: string +} + +export interface SyncStateStore { + get(did: string): RepoCheckpoint | undefined + set(checkpoint: RepoCheckpoint): void + all(): RepoCheckpoint[] + close(): void +} + +export class MemorySyncStateStore implements SyncStateStore { + readonly #values = new Map() + get(did: string): RepoCheckpoint | undefined { + const value = this.#values.get(did) + return value ? structuredClone(value) : undefined + } + set(checkpoint: RepoCheckpoint): void { + this.#values.set(checkpoint.did, structuredClone(checkpoint)) + } + all(): RepoCheckpoint[] { + return [...this.#values.values()].map((value) => structuredClone(value)) + } + close(): void {} +} + +export class SqliteSyncStateStore implements SyncStateStore { + readonly #database: DatabaseSync + constructor(path = ':memory:') { + this.#database = new DatabaseSync(path) + this.#database.exec(` + CREATE TABLE IF NOT EXISTS repo_checkpoints ( + did TEXT PRIMARY KEY, + rev TEXT NOT NULL, + commit_cid TEXT NOT NULL, + first_observed_rev TEXT NOT NULL, + updated_at TEXT NOT NULL + ) STRICT + `) + } + get(did: string): RepoCheckpoint | undefined { + const row = this.#database + .prepare('SELECT did, rev, commit_cid, first_observed_rev, updated_at FROM repo_checkpoints WHERE did = ?') + .get(did) as Record | undefined + return row + ? { + did: row.did as string, + rev: row.rev as string, + commitCid: row.commit_cid as string, + firstObservedRev: row.first_observed_rev as string, + updatedAt: row.updated_at as string, + } + : undefined + } + set(value: RepoCheckpoint): void { + this.#database + .prepare(` + INSERT INTO repo_checkpoints (did, rev, commit_cid, first_observed_rev, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (did) DO UPDATE SET + rev = excluded.rev, + commit_cid = excluded.commit_cid, + first_observed_rev = excluded.first_observed_rev, + updated_at = excluded.updated_at + `) + .run(value.did, value.rev, value.commitCid, value.firstObservedRev, value.updatedAt) + } + all(): RepoCheckpoint[] { + return (this.#database + .prepare('SELECT did, rev, commit_cid, first_observed_rev, updated_at FROM repo_checkpoints ORDER BY did') + .all() as Array>).map((row) => ({ + did: row.did as string, + rev: row.rev as string, + commitCid: row.commit_cid as string, + firstObservedRev: row.first_observed_rev as string, + updatedAt: row.updated_at as string, + })) + } + close(): void { + this.#database.close() + } +} diff --git a/packages/ingest/test/poller.test.mjs b/packages/ingest/test/poller.test.mjs new file mode 100644 index 0000000..4643cc8 --- /dev/null +++ b/packages/ingest/test/poller.test.mjs @@ -0,0 +1,75 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import { COLLECTIONS, MemoryRecordStore } from '../../core/dist/index.js' +import { MemorySyncStateStore, RepoPoller } from '../dist/index.js' + +const did = 'did:plc:alice' +const record = { + uri: `at://${did}/${COLLECTIONS.space}/one`, + cid: 'cid-space', + value: { + $type: COLLECTIONS.space, + name: 'Space', + description: 'test', + createdAt: '2026-07-18T00:00:00Z', + }, +} + +function transport(options = {}) { + let heads = options.heads ?? [ + { rev: '0001', commitCid: 'head-1' }, + { rev: '0001', commitCid: 'head-1' }, + ] + let headIndex = 0 + const calls = [] + return { + calls, + async resolvePds() { return new URL('https://pds.test') }, + async getLatestCommit() { return heads[Math.min(headIndex++, heads.length - 1)] }, + async listRecords(input) { + calls.push(input) + if (input.collection !== COLLECTIONS.space) return { records: [] } + if (!input.cursor && options.paginate) return { records: [record], cursor: 'next' } + return { records: options.paginate && input.cursor ? [] : [record] } + }, + async getRecord() { throw new Error('unused') }, + } +} + +describe('stable repo polling', () => { + it('paginates, stores validated records, and skips an unchanged head', async () => { + const remote = transport({ paginate: true }) + const store = new MemoryRecordStore() + const poller = new RepoPoller(remote, store, new MemorySyncStateStore(), { now: () => 'now' }) + const first = await poller.pollDid(did) + assert.equal(first.changed, true) + assert.equal(store.records().length, 1) + assert.equal(store.records()[0].rev, '0001') + const count = remote.calls.length + assert.equal((await poller.pollDid(did)).changed, false) + assert.equal(remote.calls.length, count) + }) + + it('discards a moving snapshot and retries from the start', async () => { + const remote = transport({ + heads: [ + { rev: '0001', commitCid: 'head-1' }, + { rev: '0002', commitCid: 'head-2' }, + { rev: '0002', commitCid: 'head-2' }, + { rev: '0002', commitCid: 'head-2' }, + ], + }) + const store = new MemoryRecordStore() + const result = await new RepoPoller(remote, store).pollDid(did) + assert.equal(result.head.rev, '0002') + assert.equal(store.records()[0].rev, '0002') + }) + + it('applies nothing after a page failure', async () => { + const remote = transport() + remote.listRecords = async () => { throw new Error('offline') } + const store = new MemoryRecordStore() + await assert.rejects(new RepoPoller(remote, store).pollDid(did), /offline/) + assert.equal(store.records().length, 0) + }) +}) diff --git a/packages/ingest/test/space.test.mjs b/packages/ingest/test/space.test.mjs new file mode 100644 index 0000000..800de0f --- /dev/null +++ b/packages/ingest/test/space.test.mjs @@ -0,0 +1,75 @@ +import assert from 'node:assert/strict' +import { it } from 'node:test' +import { COLLECTIONS, MemoryRecordStore } from '../../core/dist/index.js' +import { MemorySyncStateStore, RepoPoller, SpaceIngestor } from '../dist/index.js' + +it('polls a known member after observing removal so pre-cutoff writes are retained', async () => { + const root = 'did:plc:root' + const member = 'did:plc:member' + const uri = (did, collection, rkey) => `at://${did}/${collection}/${rkey}` + const envelope = (did, collection, rkey, cid, value) => ({ + uri: uri(did, collection, rkey), cid, value, + }) + const space = envelope(root, COLLECTIONS.space, 'space', 'cid-space', { + $type: COLLECTIONS.space, + name: 'Space', description: 'test', createdAt: '2026-07-18T00:00:00Z', + }) + const spaceRef = { uri: space.uri, cid: space.cid } + const project = envelope(root, COLLECTIONS.project, 'project', 'cid-project', { + $type: COLLECTIONS.project, + space: spaceRef, + name: 'Project', gitUrl: 'https://example.test/repo.git', defaultBranch: 'main', + checks: [], autoReview: {}, createdAt: '2026-07-18T00:01:00Z', + }) + const goal = envelope(root, COLLECTIONS.goal, 'goal', 'cid-goal', { + $type: COLLECTIONS.goal, + space: spaceRef, + project: { uri: project.uri, cid: project.cid }, + title: 'Goal', body: 'test', createdAt: '2026-07-18T00:02:00Z', + }) + const add = envelope(root, COLLECTIONS.addMember, 'add', 'cid-add', { + $type: COLLECTIONS.addMember, + space: spaceRef, + did: member, kind: 'human', role: 'member', createdAt: '2026-07-18T00:03:00Z', + }) + const repos = new Map([ + [root, { rev: '0001', records: [space, project, goal, add] }], + [member, { rev: '0000', records: [] }], + ]) + const transport = { + async resolvePds() { return new URL('https://pds.test') }, + async getLatestCommit(did) { + const repo = repos.get(did) + return { rev: repo.rev, commitCid: `head-${did}-${repo.rev}` } + }, + async listRecords({ did, collection }) { + return { records: repos.get(did).records.filter((entry) => entry.value.$type === collection) } + }, + async getRecord() { throw new Error('unused') }, + } + const store = new MemoryRecordStore() + const ingestor = new SpaceIngestor( + space.uri, + new RepoPoller(transport, store, new MemorySyncStateStore()), + store, + ) + await ingestor.sync() + + repos.get(member).rev = '0001' + repos.get(member).records.push(envelope(member, COLLECTIONS.message, 'message', 'cid-message', { + $type: COLLECTIONS.message, + goal: { uri: goal.uri, cid: goal.cid }, + body: 'written before removal', mentions: [], createdAt: '2026-07-18T00:04:00Z', + })) + repos.get(root).rev = '0002' + repos.get(root).records.push(envelope(root, COLLECTIONS.removeMember, 'remove', 'cid-remove', { + $type: COLLECTIONS.removeMember, + space: spaceRef, + did: member, atRev: '0001', createdAt: '2026-07-18T00:05:00Z', + })) + + const index = await ingestor.sync() + const message = index.goals[0].messages[0] + assert.equal(message.value.body, 'written before removal') + assert.equal(message.trust, 'removed') +}) diff --git a/packages/ingest/tsconfig.json b/packages/ingest/tsconfig.json new file mode 100644 index 0000000..bc30e7a --- /dev/null +++ b/packages/ingest/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "outDir": "dist", "rootDir": "src", "types": [] }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "test"] +} diff --git a/packages/sidecar/README.md b/packages/sidecar/README.md new file mode 100644 index 0000000..699bf02 --- /dev/null +++ b/packages/sidecar/README.md @@ -0,0 +1,52 @@ +# `radial` human CLI + +The Phase 2 CLI writes validated Radial records directly to the selected +actor's PDS. Output references use `at://...#CID`; retain the suffix when +pinning an exact artifact version. + +Authenticate each existing account once. Passwords are accepted through stdin, +never argv, and only the resulting rotating session tokens are stored: + +```sh +printf '%s\n' "$ALICE_APP_PASSWORD" | radial auth login \ + --profile alice --pds https://alice-pds.example \ + --identifier alice.example --password-stdin + +printf '%s\n' "$BOB_APP_PASSWORD" | radial auth login \ + --profile bob --pds https://bob-pds.example \ + --identifier bob.example --password-stdin +``` + +The complete human-only loop uses the same request/artifact commands for both +versions: + +```sh +SPACE=$(radial space create --profile alice --name radial --description 'Open development') +radial member add --profile alice --space "$SPACE" \ + --did did:plc:bob --kind human --role member + +PROJECT=$(radial project create --profile alice --space "$SPACE" \ + --name radial --git-url https://example.com/radial.git) +GOAL=$(radial goal create --profile alice --project "$PROJECT" \ + --title 'Ship the feature' --body 'Acceptance criteria go here') + +REQUEST_V1=$(radial request create --profile alice --goal "$GOAL" \ + --type plan --assignee did:plc:bob) +PLAN_V1=$(radial artifact post --profile bob --request "$REQUEST_V1" \ + --body-file plan-v1.md) +radial review post --profile alice --subject "$PLAN_V1" \ + --verdict request_changes + +REQUEST_V2=$(radial request create --profile alice --goal "$GOAL" \ + --type plan --assignee did:plc:bob --based-on "$PLAN_V1") +PLAN_V2=$(radial artifact post --profile bob --request "$REQUEST_V2" \ + --prev "$PLAN_V1" --body-file plan-v2.md) +radial review post --profile alice --subject "$PLAN_V2" --verdict approve +``` + +`space create` also publishes the built-in `plan` and `implementation` +artifact-type registry records. Use `--no-builtins` only if another caller will +seed them. `--json` returns structured refs for scripts. + +Review findings files contain a JSON array. Each entry has `severity` +(`info`, `warning`, or `error`) and `body`, with optional `path` and `line`. diff --git a/packages/sidecar/package.json b/packages/sidecar/package.json index 38ca45d..c5428a1 100644 --- a/packages/sidecar/package.json +++ b/packages/sidecar/package.json @@ -1 +1,18 @@ -{"name":"@radial/sidecar","version":"0.0.0","private":true,"type":"module","scripts":{"build":"true","lint":"true"}} +{ + "name": "@radial/sidecar", + "version": "0.0.0", + "private": true, + "type": "module", + "bin": { "radial": "./dist/cli.js" }, + "exports": { ".": { "types": "./src/index.ts", "import": "./dist/index.js" } }, + "scripts": { + "build": "tsc -p tsconfig.json", + "lint": "pnpm typecheck", + "test": "pnpm build && node --test test/*.test.mjs", + "typecheck": "tsc -p tsconfig.json --noEmit" + }, + "dependencies": { + "@radial/atproto": "workspace:*", + "@radial/core": "workspace:*" + } +} diff --git a/packages/sidecar/src/cli.ts b/packages/sidecar/src/cli.ts new file mode 100644 index 0000000..d55d028 --- /dev/null +++ b/packages/sidecar/src/cli.ts @@ -0,0 +1,87 @@ +#!/usr/bin/env node + +import { readFile } from 'node:fs/promises' +import { argv, stdin } from 'node:process' +import { pathToFileURL } from 'node:url' +import { + CredentialClient, + FetchRepoTransport, + FileSessionStore, + StrongRefResolver, + createSession, +} from '@radial/atproto' +import { runCli } from './commands.js' + +export const help = `radial — human CLI for Radial records + + radial auth login --profile NAME --pds URL --identifier HANDLE --password-stdin + radial space create --name NAME [--description TEXT] + radial member add --space REF --did DID --kind human|agent --role ROLE + radial project create --space REF --name NAME --git-url URL + radial goal create --project REF --title TITLE (--body TEXT | --body-file PATH) + radial request create (--goal REF | --project REF) --type TYPE [--assignee DID] + [--based-on REF]... [--brief TEXT] + radial artifact post --request REF (--body TEXT | --body-file PATH) [--prev REF] + [--criterion TEXT]... + radial review post --subject REF --verdict approve|request_changes + [--request REF] [--findings-file JSON] + +References may be bare at:// URIs or pinned at://...#CID locators. +Use --profile NAME to select an actor and --json for structured output.` + +async function readStdin(): Promise { + let value = '' + for await (const chunk of stdin) value += new TextDecoder().decode(chunk) + return value.trim() +} + +function value(args: string[], flag: string): string | undefined { + const index = args.lastIndexOf(flag) + return index === -1 ? undefined : args[index + 1] +} + +export async function main(args = argv.slice(2)): Promise { + if (args.length === 0 || args.includes('--help') || args[0] === 'help') { + console.log(help) + return + } + const store = new FileSessionStore() + if (args[0] === 'auth' && args[1] === 'login') { + const profile = value(args, '--profile') + const service = value(args, '--pds') + const identifier = value(args, '--identifier') + if (!profile || !service || !identifier || !args.includes('--password-stdin')) { + throw new Error('auth login requires --profile, --pds, --identifier, and --password-stdin') + } + const session = await createSession(service, identifier, await readStdin()) + await store.save({ ...session, profile }) + console.log(JSON.stringify({ profile, did: session.did, handle: session.handle, service: session.service })) + return + } + + const profile = value(args, '--profile') ?? process.env.RADIAL_PROFILE + const actor = await store.get(profile) + if (!actor) throw new Error('No Radial identity is configured; run radial auth login') + const client = new CredentialClient( + actor, + fetch, + undefined, + (current, refresh) => store.coordinateRefresh(actor.profile, current, refresh), + ) + const resolver = new StrongRefResolver(new FetchRepoTransport()) + const result = await runCli(args, { + writer: client, + resolver, + readText: async (path) => (path === '-' ? readStdin() : readFile(path, 'utf8')), + }) + if (args.includes('--json')) console.log(JSON.stringify({ ...result, ref: result.primary })) + else console.log(`${result.primary.uri}#${result.primary.cid}`) +} + +const entry = argv[1] +if (entry && import.meta.url === pathToFileURL(entry).href) { + main().catch((error: unknown) => { + console.error(error instanceof Error ? error.message : error) + process.exitCode = 1 + }) +} diff --git a/packages/sidecar/src/commands.ts b/packages/sidecar/src/commands.ts new file mode 100644 index 0000000..e8284fc --- /dev/null +++ b/packages/sidecar/src/commands.ts @@ -0,0 +1,238 @@ +import { + COLLECTIONS, + type Collection, + type RecordByCollection, + type StrongRef, +} from '@radial/core' +import type { ResolvedRecord } from '@radial/atproto' + +export interface RecordWriter { + create( + collection: K, + value: RecordByCollection[K], + options?: { rkey?: string }, + ): Promise +} + +export interface RecordResolver { + resolve(locator: string, expectedCollection?: Collection): Promise +} + +export interface CliDependencies { + writer: RecordWriter + resolver: RecordResolver + now?: () => string + readText?: (path: string) => Promise +} + +export interface CommandResult { + primary: StrongRef + refs: Record +} + +const builtinTypes = [ + { + name: 'plan', + brief: + 'Produce a concrete implementation plan grounded in the goal, repository, and supplied provenance. Call out risks, ordered work, and verification.', + outputSpec: { + format: 'markdown' as const, + description: + 'A structured plan with context, ordered implementation steps, risks, and verification criteria.', + }, + }, + { + name: 'implementation', + brief: + 'Implement the requested change in the repository, run the project checks, and return exact git and pull-request provenance.', + outputSpec: { + format: 'markdown' as const, + description: + 'A concise implementation summary with branch, commit, pull request, checks, and any remaining risks.', + }, + }, +] + +function values(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 optional(args: string[], flag: string): string | undefined { + return values(args, flag).at(-1) +} + +function required(args: string[], flag: string): string { + const value = optional(args, flag) + if (!value) throw new Error(`Missing ${flag}`) + return value +} + +async function inputText(args: string[], deps: CliDependencies, name: string): Promise { + const inline = optional(args, `--${name}`) + const path = optional(args, `--${name}-file`) + if (Boolean(inline) === Boolean(path)) { + throw new Error(`Provide exactly one of --${name} or --${name}-file`) + } + if (inline !== undefined) return inline + if (!deps.readText) throw new Error(`Cannot read --${name}-file without a file reader`) + return deps.readText(path as string) +} + +const ref = (record: ResolvedRecord): StrongRef => ({ uri: record.uri, cid: record.cid }) + +export async function runCli(args: string[], deps: CliDependencies): Promise { + const now = deps.now?.() ?? new Date().toISOString() + const [group, action] = args + + if (group === 'space' && action === 'create') { + const space = await deps.writer.create(COLLECTIONS.space, { + $type: COLLECTIONS.space, + name: required(args, '--name'), + description: optional(args, '--description') ?? '', + createdAt: now, + }) + const seeded: StrongRef[] = [] + if (!args.includes('--no-builtins')) { + for (const builtin of builtinTypes) { + seeded.push( + await deps.writer.create(COLLECTIONS.artifactType, { + $type: COLLECTIONS.artifactType, + space, + ...builtin, + scope: 'goal', + createdAt: now, + }), + ) + } + } + return { primary: space, refs: { space, seededArtifactTypes: seeded } } + } + + if (group === 'member' && action === 'add') { + const space = await deps.resolver.resolve(required(args, '--space'), COLLECTIONS.space) + const member = await deps.writer.create(COLLECTIONS.addMember, { + $type: COLLECTIONS.addMember, + space: ref(space), + did: required(args, '--did'), + kind: required(args, '--kind') as 'human' | 'agent', + role: required(args, '--role') as 'admin' | 'member' | 'agent', + createdAt: now, + }) + return { primary: member, refs: { member } } + } + + if (group === 'project' && action === 'create') { + const space = await deps.resolver.resolve(required(args, '--space'), COLLECTIONS.space) + const checks = values(args, '--check').map((entry) => { + const separator = entry.indexOf('=') + if (separator < 1) throw new Error('--check must be name=command') + return { name: entry.slice(0, separator), command: entry.slice(separator + 1) } + }) + const project = await deps.writer.create(COLLECTIONS.project, { + $type: COLLECTIONS.project, + space: ref(space), + name: required(args, '--name'), + gitUrl: required(args, '--git-url'), + defaultBranch: optional(args, '--default-branch') ?? 'main', + checks, + autoReview: {}, + createdAt: now, + }) + return { primary: project, refs: { project } } + } + + if (group === 'goal' && action === 'create') { + const project = await deps.resolver.resolve(required(args, '--project'), COLLECTIONS.project) + const projectValue = project.value as RecordByCollection[typeof COLLECTIONS.project] + const goal = await deps.writer.create(COLLECTIONS.goal, { + $type: COLLECTIONS.goal, + space: projectValue.space, + project: ref(project), + title: required(args, '--title'), + body: await inputText(args, deps, 'body'), + createdAt: now, + }) + return { primary: goal, refs: { goal } } + } + + if (group === 'request' && action === 'create') { + const goalLocator = optional(args, '--goal') + const projectLocator = optional(args, '--project') + if (Boolean(goalLocator) === Boolean(projectLocator)) { + throw new Error('Provide exactly one of --goal or --project') + } + const target = await deps.resolver.resolve( + (goalLocator ?? projectLocator) as string, + goalLocator ? COLLECTIONS.goal : COLLECTIONS.project, + ) + const basedOn = await Promise.all(values(args, '--based-on').map((value) => deps.resolver.resolve(value))) + const assignee = optional(args, '--assignee') + const brief = optional(args, '--brief') + const request = await deps.writer.create(COLLECTIONS.artifactRequest, { + $type: COLLECTIONS.artifactRequest, + ...(goalLocator ? { goal: ref(target) } : { project: ref(target) }), + type: required(args, '--type'), + basedOn: basedOn.map(ref), + ...(assignee ? { assignee } : {}), + ...(brief ? { brief } : {}), + createdAt: now, + }) + return { primary: request, refs: { request } } + } + + if (group === 'artifact' && (action === 'post' || action === 'submit')) { + const request = await deps.resolver.resolve( + required(args, '--request'), + COLLECTIONS.artifactRequest, + ) + const requestValue = request.value as RecordByCollection[typeof COLLECTIONS.artifactRequest] + if (requestValue.type === 'review') throw new Error('Review requests must be fulfilled with a review') + const previousLocator = optional(args, '--prev') + const previous = previousLocator + ? await deps.resolver.resolve(previousLocator, COLLECTIONS.artifact) + : undefined + const artifact = await deps.writer.create(COLLECTIONS.artifact, { + $type: COLLECTIONS.artifact, + request: ref(request), + ...(requestValue.goal ? { goal: requestValue.goal } : { project: requestValue.project as StrongRef }), + type: requestValue.type, + ...(previous ? { prev: ref(previous) } : {}), + body: await inputText(args, deps, 'body'), + links: {}, + ...(values(args, '--criterion').length ? { criteria: values(args, '--criterion') } : {}), + createdAt: now, + }) + return { primary: artifact, refs: { artifact } } + } + + if (group === 'review' && action === 'post') { + const subject = await deps.resolver.resolve(required(args, '--subject'), COLLECTIONS.artifact) + const requestLocator = optional(args, '--request') + const request = requestLocator + ? await deps.resolver.resolve(requestLocator, COLLECTIONS.artifactRequest) + : undefined + let findings: Array<{ path?: string; line?: number; severity: 'info' | 'warning' | 'error'; body: string }> = [] + const findingsFile = optional(args, '--findings-file') + if (findingsFile) { + if (!deps.readText) throw new Error('Cannot read findings without a file reader') + const parsed = JSON.parse(await deps.readText(findingsFile)) as unknown + if (!Array.isArray(parsed)) throw new Error('Findings file must contain a JSON array') + findings = parsed as typeof findings + } + const review = await deps.writer.create(COLLECTIONS.review, { + $type: COLLECTIONS.review, + subject: ref(subject), + ...(request ? { request: ref(request) } : {}), + verdict: required(args, '--verdict') as 'approve' | 'request_changes', + findings, + createdAt: now, + }) + return { primary: review, refs: { review } } + } + + throw new Error('Unknown command. Run radial --help.') +} diff --git a/packages/sidecar/src/index.ts b/packages/sidecar/src/index.ts new file mode 100644 index 0000000..f0ebc49 --- /dev/null +++ b/packages/sidecar/src/index.ts @@ -0,0 +1 @@ +export * from './commands.js' diff --git a/packages/sidecar/src/node-shims.d.ts b/packages/sidecar/src/node-shims.d.ts new file mode 100644 index 0000000..04dc0c1 --- /dev/null +++ b/packages/sidecar/src/node-shims.d.ts @@ -0,0 +1,30 @@ +declare module 'node:fs/promises' { + export function chmod(path: string, mode: number): Promise + export function lstat(path: string): Promise<{ isFile(): boolean; isSymbolicLink(): boolean; mode: number }> + export function mkdir(path: string, options?: { recursive?: boolean; mode?: number }): Promise + export function open(path: string, flags: string, mode?: number): Promise<{ close(): Promise; writeFile(data: string): Promise }> + export function readFile(path: string, encoding: string): Promise + export function rename(oldPath: string, newPath: string): Promise + export function rm(path: string, options?: { force?: boolean }): Promise + export function writeFile(path: string, data: string, options?: { mode?: number }): Promise +} +declare module 'node:path' { + export function dirname(path: string): string + export function join(...paths: string[]): string +} +declare module 'node:process' { + export const argv: string[] + export const stdin: AsyncIterable +} +declare module 'node:url' { + export function pathToFileURL(path: string): URL +} +declare module 'node:sqlite' { + export class DatabaseSync { + constructor(path: string) + exec(sql: string): void + prepare(sql: string): { get(...values: unknown[]): unknown; all(...values: unknown[]): unknown[]; run(...values: unknown[]): unknown } + close(): void + } +} +declare const process: { argv: string[]; exitCode?: number; env: Record; pid: number; kill(pid: number, signal: number): void } diff --git a/packages/sidecar/test/human-loop.test.mjs b/packages/sidecar/test/human-loop.test.mjs new file mode 100644 index 0000000..c2eaac3 --- /dev/null +++ b/packages/sidecar/test/human-loop.test.mjs @@ -0,0 +1,201 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import { + CredentialClient, + FetchRepoTransport, + StrongRefResolver, + createSession, +} from '../../atproto/dist/index.js' +import { MemoryRecordStore } from '../../core/dist/index.js' +import { MemorySyncStateStore, RepoPoller, SpaceIngestor } from '../../ingest/dist/index.js' +import { runCli } from '../dist/index.js' + +class LocalPds { + seq = 0 + records = new Map() + service + + constructor(did) { + this.did = did + this.handle = `${did.split(':').at(-1)}.test` + this.service = `https://${did.split(':').at(-1)}.pds.test` + } + + async fetch(input, init) { + try { + const request = new Request(input, init) + const url = new URL(request.url) + const method = url.pathname.slice('/xrpc/'.length) + if (method === 'com.atproto.server.createSession' && request.method === 'POST') { + const body = await request.json() + assert.equal(body.identifier, this.handle) + return this.json({ + did: this.did, + handle: this.handle, + accessJwt: `${this.did}-access`, + refreshJwt: `${this.did}-refresh`, + }) + } + if (method === 'com.atproto.repo.createRecord' && request.method === 'POST') { + assert.equal(request.headers.get('authorization'), `Bearer ${this.did}-access`) + const body = await request.json() + assert.equal(body.repo, this.did) + assert.equal(body.validate, false) + this.seq += 1 + const rkey = body.rkey ?? String(this.seq).padStart(13, '0') + const uri = `at://${this.did}/${body.collection}/${rkey}` + if (this.records.has(uri)) return this.error(400, 'RecordAlreadyExists') + const cid = `cid-${this.did.split(':').at(-1)}-${this.seq}` + this.records.set(uri, { uri, cid, value: structuredClone(body.record) }) + return this.json({ + uri, + cid, + commit: { rev: this.rev(), cid: this.head() }, + }) + } + if (method === 'com.atproto.repo.getRecord') { + const uri = `at://${url.searchParams.get('repo')}/${url.searchParams.get('collection')}/${url.searchParams.get('rkey')}` + const record = this.records.get(uri) + return record ? this.json(record) : this.error(400, 'RecordNotFound') + } + if (method === 'com.atproto.sync.getLatestCommit') { + return this.json({ rev: this.rev(), cid: this.head() }) + } + if (method === 'com.atproto.repo.listRecords') { + const collection = url.searchParams.get('collection') + const offset = Number(url.searchParams.get('cursor') ?? 0) + const limit = Number(url.searchParams.get('limit') ?? 100) + const records = [...this.records.values()] + .filter((record) => record.value.$type === collection) + .sort((left, right) => left.uri.localeCompare(right.uri)) + const page = records.slice(offset, offset + limit) + const next = offset + page.length + return this.json({ + records: page, + ...(next < records.length ? { cursor: String(next) } : {}), + }) + } + return this.error(404, 'UnknownMethod') + } catch (error) { + return this.error(500, 'TestPdsError', error.message) + } + } + + rev() { + return String(this.seq).padStart(13, '0') + } + head() { + return `head-${this.did}-${this.seq}` + } + json(value, status = 200) { + return Response.json(value, { status }) + } + error(status, error, message = error) { + return this.json({ error, message }, status) + } +} + +const locator = (result) => `${result.primary.uri}#${result.primary.cid}` + +describe('Phase 2 human-only artifact loop', () => { + it('uses two Fetch/XRPC PDS fixtures and converges in two independent materializers', async () => { + const aliceDid = 'did:plc:alice' + const bobDid = 'did:plc:bob' + const pdss = new Map([ + [aliceDid, new LocalPds(aliceDid)], + [bobDid, new LocalPds(bobDid)], + ]) + const fetcher = (input, init) => { + const service = new URL(input).origin + const pds = [...pdss.values()].find((candidate) => candidate.service === service) + if (!pds) throw new Error(`No local PDS for ${service}`) + return pds.fetch(input, init) + } + const transport = new FetchRepoTransport(async (did) => pdss.get(did).service, fetcher) + const resolver = new StrongRefResolver(transport) + const sessions = new Map() + for (const [did, pds] of pdss) { + sessions.set( + did, + new CredentialClient( + await createSession(pds.service, pds.handle, 'password', fetcher), + fetcher, + ), + ) + } + const at = (did, minute) => ({ + writer: sessions.get(did), + resolver, + now: () => `2026-07-18T00:${String(minute).padStart(2, '0')}:00Z`, + readText: async () => { throw new Error('unexpected file read') }, + }) + + const space = await runCli( + ['space', 'create', '--name', 'Radial', '--description', 'human loop'], + at(aliceDid, 0), + ) + const stores = [new MemoryRecordStore(), new MemoryRecordStore()] + const ingestors = stores.map((store) => + new SpaceIngestor( + space.primary.uri, + new RepoPoller(transport, store, new MemorySyncStateStore(), { + now: () => '2026-07-18T01:00:00Z', + }), + store, + ), + ) + const sync = async () => { + const views = await Promise.all(ingestors.map((ingestor) => ingestor.sync())) + assert.deepEqual(views[0], views[1]) + return views[0] + } + await sync() + + await runCli( + ['member', 'add', '--space', locator(space), '--did', bobDid, '--kind', 'human', '--role', 'member'], + at(aliceDid, 1), + ) + await sync() + const project = await runCli( + ['project', 'create', '--space', locator(space), '--name', 'radial', '--git-url', 'https://example.test/radial.git'], + at(aliceDid, 2), + ) + const goal = await runCli( + ['goal', 'create', '--project', locator(project), '--title', 'Ship phase 2', '--body', 'Prove the loop'], + at(aliceDid, 3), + ) + const requestV1 = await runCli( + ['request', 'create', '--goal', locator(goal), '--type', 'plan', '--assignee', bobDid], + at(aliceDid, 4), + ) + const planV1 = await runCli( + ['artifact', 'post', '--request', locator(requestV1), '--body', 'Plan version one'], + at(bobDid, 5), + ) + await runCli( + ['review', 'post', '--subject', locator(planV1), '--verdict', 'request_changes'], + at(aliceDid, 6), + ) + const requestV2 = await runCli( + ['request', 'create', '--goal', locator(goal), '--type', 'plan', '--assignee', bobDid, '--based-on', locator(planV1)], + at(aliceDid, 7), + ) + const planV2 = await runCli( + ['artifact', 'post', '--request', locator(requestV2), '--prev', locator(planV1), '--body', 'Plan version two'], + at(bobDid, 8), + ) + await runCli( + ['review', 'post', '--subject', locator(planV2), '--verdict', 'approve'], + at(aliceDid, 9), + ) + const index = await sync() + + const view = index.goals.find((candidate) => candidate.target.uri === goal.primary.uri) + assert.equal(view.requests.length, 2) + assert.equal(view.openRequests.length, 0) + assert.deepEqual(view.artifactChains[0].versions, [planV1.primary.uri, planV2.primary.uri]) + assert.deepEqual(view.reviews.map((review) => review.value.verdict), ['request_changes', 'approve']) + assert.equal(view.artifacts.every((artifact) => artifact.did === bobDid), true) + assert.equal(index.ignored.length, 0) + }) +}) diff --git a/packages/sidecar/tsconfig.json b/packages/sidecar/tsconfig.json new file mode 100644 index 0000000..bc30e7a --- /dev/null +++ b/packages/sidecar/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "outDir": "dist", "rootDir": "src", "types": [] }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "test"] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8cc6062..112f843 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -12,15 +12,45 @@ importers: specifier: 5.9.3 version: 5.9.3 + packages/atproto: + dependencies: + '@radial/core': + specifier: workspace:* + version: link:../core + packages/core: {} - packages/daemon: {} + packages/daemon: + dependencies: + '@radial/atproto': + specifier: workspace:* + version: link:../atproto + '@radial/core': + specifier: workspace:* + version: link:../core + '@radial/ingest': + specifier: workspace:* + version: link:../ingest - packages/ingest: {} + packages/ingest: + dependencies: + '@radial/atproto': + specifier: workspace:* + version: link:../atproto + '@radial/core': + specifier: workspace:* + version: link:../core packages/lexicons: {} - packages/sidecar: {} + packages/sidecar: + dependencies: + '@radial/atproto': + specifier: workspace:* + version: link:../atproto + '@radial/core': + specifier: workspace:* + version: link:../core packages/ui: {} diff --git a/readme.md b/readme.md index b22eb1f..d268fe3 100644 --- a/readme.md +++ b/readme.md @@ -5,7 +5,7 @@ Radial is a coding-agent orchestration system built on produce them, and reviews annotate exact artifact versions. Coordination lives in each actor's atproto repo; code stays in git. -Phases 0 and 1 are implemented: +Phases 0 through 2 are implemented: - pnpm TypeScript monorepo and offline CI/test harness - all 13 `com.disnetdev.radial.*` lexicons and generated record interfaces @@ -15,11 +15,17 @@ Phases 0 and 1 are implemented: - goal/project materialization, version chains, reviews, checkruns, request fulfillment, declines, claims, and current system artifacts - golden scenarios and an arrival-permutation convergence test +- zero-dependency atproto XRPC, app-password sessions, token refresh, and + protected session persistence +- `radiald init` for existing agent identities and published agent profiles +- stable, paginated repo polling with per-author ordering and member discovery +- human `radial` commands for the full request → v1 → changes → v2 → approval + cycle, covered by a two-DID/two-PDS convergence test ## Development -Node 24 and pnpm 10 are required. The phase 0–1 workspace has no third-party -runtime or test dependencies, so setup and verification work offline. +Node 24 and pnpm 10 are required. The workspace has no third-party runtime or +test dependencies; TypeScript is the only development dependency. ```sh pnpm install --offline @@ -29,6 +35,40 @@ pnpm test pnpm build ``` +Register an existing agent identity (the app password is read from stdin and +is never stored): + +```sh +printf '%s\n' "$AGENT_APP_PASSWORD" | radiald init planner \ + --pds https://pds.example \ + --identifier planner.example \ + --name planner \ + --harness codex \ + --model gpt-5=high \ + --artifact-type plan \ + --password-stdin +``` + +See [the CLI walkthrough](packages/sidecar/README.md) for the human-only cycle. + +Run a durable daemon ingestion/materializer pass, then inspect the same local +view without hitting the network: + +```sh +radiald sync "$SPACE_URI" --records radial.db --checkpoints radial-sync.db +radiald index "$SPACE_URI" --records radial.db +``` + +### Polling-v1 provenance bound + +`listRecords` exposes a current URI/CID/value snapshot, not the commit revision +at which each record was originally written. The ingestor brackets each full +author scan with `getLatestCommit` and applies only stable snapshots. This +covers the Phase 2 live loop, but a materializer first started after a member +removal cannot reconstruct exact pre-removal history or recover an edit or +deletion it never observed. The transport seam allows Phase 7 CAR/Jetstream +ingestion to replace snapshot-head provenance with commit-level history. + The debug CLI reads one JSON record envelope (or an array of envelopes) per file in a directory: @@ -39,5 +79,5 @@ pnpm --filter @radial/core debug -- \ --as-of 2026-07-18T00:00:00Z ``` -See [design.md](design.md) for the protocol design and [plan.md](plan.md) for -the remaining phases. +See [design.md](docs/design.md) for the protocol design and +[plan.md](docs/plan.md) for the remaining phases. -- 2.51.2