johnandersen777.bsky.social/atlibs-agent-reference-spa

Switch repository

Loading…

// The page's logic, DOM-free (test/*.test.mjs run it in Node): load the modules and their sessions,
// run an action held to its module's session, the MCP servers the agent uses, the agent run, the ingress.
import { codeMode, MESSAGE } from '@publicdomainrelay/atlibs-code-mode/index.js'
import { loadComposition, runCompose } from '@publicdomainrelay/atlibs/compose/index.js'
import { didkeyCall } from '@publicdomainrelay/atlibs/didkey/index.js'
import { actionsServer, exposed, hostRun, ingressMcp } from '@publicdomainrelay/atlibs/mcp/index.js'
import { policyCall } from '@publicdomainrelay/atlibs/oauth/index.js'
import { hostCall, sessionXrpc } from '@publicdomainrelay/atlibs/run-module/host-call.js'
import { localModule, runModule } from '@publicdomainrelay/atlibs/run-module/index.js'

import { COCORE, DEFAULTS, DYNAMIC_MCP, LIBS, SOURCES, TOOLS } from './config.js'

export const FORWARD = 'com.fedfork.exec.ingress.request'
export const ANTHROPIC = 'https://api.anthropic.com'
// ponytail: co/core's API answers CORS for localhost origins only; elsewhere the page offers custom endpoints alone
export const cocoreUsable = (hostname) => ['localhost', '127.0.0.1', '[::1]'].includes(hostname)
// a server or a long agent run ends when it is done or aborted, not on a count or a clock
export const UNLIMITED = { calls: Infinity, events: Infinity, wallMs: Infinity }

/**
 * Every module in SOURCES, as `atlibs stage --target spa` staged it beside the page (stage.json: its wasm and
 * module.json under modules/, fetched from its record by atlibs add), and the policy: each module's own
 * session and that session's scope (stage.json's, from the published session record).
 */
export async function loadAll(ctx, base, fetchImpl = globalThis.fetch) {
	const stage = await (await fetchImpl(new URL('stage.json', base))).json()
	const staged = (name) => stage.modules[name] ?? Promise.reject(new Error(`${name} is not staged: scripts/stage.sh`))
	const mods = await Promise.all(SOURCES.map(async (name) => { const m = await staged(name); return localModule(new URL(m.wasm, base), new URL(m.spec, base), fetchImpl) }))
	const modules = Object.fromEntries(SOURCES.map((name, i) => [name, mods[i]]))
	const policy = { sessions: {}, modules: {} }
	for (const name of SOURCES) {
		const session = stage.modules[name].session
		if (!session) continue
		policy.sessions[session] = { scope: stage.sessions[session].scope }
		policy.modules[name] = { session }
	}
	return { modules, policy }
}

// a lib's modules, and the hook defaults they reach (`hooks`): those run on the same lib's host options
const libOf = (module) => LIBS.find((lib) => lib.modules.includes(module) || lib.hooks?.includes(module))

/**
 * A module's host options (atlibs/mcp hostRun's): as the signed-in user, held to that module's own
 * session's scope by policyCall; signed out, public reads only. Its lib's budget, and co/core's plain HTTP.
 * A hook default runs as its own module (its own session) when its lib names it in `hooks` (vm-mcp's requester
 * and process-mcp), else makes public reads (the requester's bid policy, user-data). A `vm` lib also gets the
 * host's vm: { attest, calls, fetchers } (the attestation key, the webrtc transport) and the managed_records ledger,
 * and its calls to a bidder's did#pdr_temp_market endpoints go over vm.peers (the webrtc transport's own fetchers).
 */
export function hostOptions({ ctx, policy, session, vm }, name) {
	const lib = libOf(name)
	const fetchers = lib?.vm ? vm?.peers : undefined
	const call = session ? policyCall(policy, name, { [policy.modules[name]?.session ?? name]: hostCall({ ctx, xrpc: sessionXrpc(session), canHeader: false, fetchers }) }) : hostCall({ ctx, canHeader: false, fetchers })
	// co/core's plain HTTP: its origin, with service auth the host mints from the session
	// a vm lib's: the VMs' shells (process-mcp's http.fetch to webrtc:///<ref>/shell/mcp, over vm.fetchers)
	const http = lib?.http === 'cocore' ? { [new URL(COCORE.endpoint).origin]: { serviceAuth: COCORE.serviceAuth } } : lib?.vm ? { 'webrtc://*': {} } : undefined
	const transport = lib?.vm && vm ? { attest: vm.attest, calls: vm.calls, fetchers: vm.fetchers, ledger: ctx.storage?.managed_records } : {}
	const callFor = (m) => (libOf(m.name) ? hostOptions({ ctx, policy, session, vm }, m.name).call : hostCall({ ctx, canHeader: false }))
	return { call, callFor, budget: lib?.budget, http, input: session ? { did: session.did } : undefined, ...transport }
}

/** A lib's `defaults` (config.js): their fields optional in its tools, each description naming that it has one. */
export const withDefaults = (mod, defaults = {}) => ({
	...mod,
	actions: (mod.actions ?? []).map((a) =>
		defaults[a.entry] ? { ...a, fields: (a.fields ?? []).map((f) => (f.key in defaults[a.entry] ? { ...f, required: false, description: `${f.description ?? f.label ?? f.key} (default: ${JSON.stringify(defaults[a.entry][f.key])})` } : f)) } : a,
	),
})

/** One action (a form's): runModule's { result, error }. */
export const runAction = (host, mod, entry, fields, { signal } = {}) => hostRun(host.ctx, (m) => hostOptions(host, m.name))(mod, entry, fields, { signal })

// config.js names the read-only entries ('*': all of a module's); else module.json's readOnly.
const readOnlyOf = (lib) => (module, entry, action) => {
	const ro = lib.readOnly?.[module]
	return ro === '*' || (Array.isArray(ro) && ro.includes(entry)) || action.readOnly === true
}

/**
 * The agent's in-process MCP servers: code mode (its programs call the lib servers' read-only tools, no XRPC), and one per TOOLS lib
 * (atlibs/mcp actionsServer over its modules' actions, each run on hostOptions, a call's missing fields from
 * the lib's defaults). confirm: the page's, asked before any write. vm: the host's VM transport (hostOptions).
 * observe(server, entry, args, ran): after each tool's run (the page's VM list). Returns { name: { message, handler, tools?, readOnly(tool) } }.
 */
export function mcpServers({ ctx, policy, modules, session, confirm, vm, observe }) {
	const servers = {}
	// a program's run.call(id, "com.fedfork.exec.mcp.<lib>", <JSON-RPC message>): that server's read-only tools only
	const calls = Object.fromEntries(TOOLS.map((name) => [hookOf(name), ({ input: m } = {}) => (m?.method === 'tools/call' && !servers[name].readOnly(m.params?.name) ? { jsonrpc: '2.0', id: m.id ?? null, error: { code: -32602, message: `${m.params?.name}: code mode calls read-only tools only; call it as a tool` } } : servers[name].message(m))]))
	const code = codeMode({ ctx, call: hostCall({ ctx, canHeader: false, calls }), allow: Object.keys(calls) })
	servers.code = { ...code, readOnly: () => true }
	const host = (mod) => hostOptions({ ctx, policy, session, vm }, mod.name)
	const run = hostRun(ctx, host)
	for (const lib of LIBS.filter((l) => TOOLS.includes(l.name))) {
		const runLib = async (mod, entry, args, options) => {
			const full = { ...lib.defaults?.[entry], ...args }
			const ran = await run(mod, entry, full, options)
			observe?.(lib.name, entry, full, ran)
			return ran
		}
		// a one-module lib's tools are its entries (vm-mcp's vm_create), else <module>_<entry>
		servers[lib.name] = actionsServer({ name: lib.modules.length === 1 ? lib.modules[0] : lib.name, modules: lib.modules.map((m) => withDefaults(modules[m], lib.defaults)), ctx, host, run: runLib, readOnly: readOnlyOf(lib), confirm })
	}
	return servers
}

/**
 * atlibs-webrtc recordSignals' repo for a session: reads from each repo's own PDS, the offer (a fresh TID rkey,
 * so a create) and its delete as the user (the requester session's ingress scope).
 */
export function sessionRepo(ctx, session) {
	const read = hostCall({ ctx, canHeader: false })
	const xrpc = sessionXrpc(session)
	return {
		did: session.did,
		getRecord: ({ repo, collection, rkey }) => read('com.atproto.repo.getRecord', { params: { repo, collection, rkey }, service: repo }).catch((e) => (/RecordNotFound|not found|404/i.test(e.message) ? null : Promise.reject(e))),
		putRecord: ({ collection, rkey, record }) => xrpc('com.atproto.repo.createRecord', { input: { repo: session.did, collection, rkey, record } }),
		deleteRecord: ({ collection, rkey }) => xrpc('com.atproto.repo.deleteRecord', { input: { repo: session.did, collection, rkey } }),
	}
}

/**
 * The webrtc transport's (atlibs-webrtc WebrtcHost) events and signal, as requester runs: run(entry, input) -> result
 * (it throws a run's error). events: the guest's reports for a receipt and tag (watch, one poll); signal: the sealed offer, as the user.
 */
export const vmSignals = (run) => ({
	events: async ({ receipt, tag }) => {
		const bidder = receipt.bidder ?? receipt.uri.split('/')[2]
		return (await run('watch', { bidder, receipt: receipt.uri, tags: [tag], max_polls: 1 })).events
	},
	signal: (input) => run('signal', input),
})

// A webrtc:///<ref>/... URL's ref, decoded (base64url JSON), or undefined.
const refOf = (url) => {
	try {
		const ref = new URL(url).pathname.split('/')[1]
		return JSON.parse(atob(ref.replace(/-/g, '+').replace(/_/g, '/')))
	} catch {
		return undefined
	}
}

/**
 * The host's scheme fetchers, held to the VMs it keeps (kept(): vm_list's vms, each with its shell URL). A model
 * copies a long base64 ref unreliably (live: one CID character dropped, and the guest never answered the signal),
 * so a URL goes out with its kept VM's own ref, found by the receipt URI; one naming no kept VM fails at once
 * instead of signalling a guest that will not answer.
 */
export const keptFetchers = (fetchers, kept) =>
	Object.fromEntries(
		Object.entries(fetchers).map(([scheme, fetch]) => [
			scheme,
			(request) => {
				const uri = refOf(request.url)?.receipt?.uri
				const url = kept().map((v) => v.shell).find((u) => u?.startsWith(`${scheme}//`) && uri && refOf(u)?.receipt?.uri === uri)
				if (!url) throw new Error(`${request.url.slice(0, 48)}...: no VM kept here has this receipt; pass vm_create's shell url exactly, or use vm_exec (kept: ${kept().map((v) => v.vm).join(', ') || 'none'})`)
				const [, , , , ...rest] = request.url.split('/')
				return fetch({ ...request, url: `${url.split('/').slice(0, 4).join('/')}/${rest.join('/')}` })
			},
		]),
	)

/**
 * The VMs kept, after a vm lib run (mcpServers' observe): vm_list's are the list; vm_create adds its VM,
 * vm_release and vm_forget drop theirs. vms: the list before (vm_list's at load).
 */
export function vmsAfter(vms, server, entry, args, ran) {
	if (server !== 'vm' || ran?.error !== undefined) return vms
	const r = ran?.result
	if (entry === 'vm_list' && Array.isArray(r?.vms)) return r.vms
	if (entry === 'vm_create' && r?.vm) return [...vms.filter((v) => v.vm !== r.vm), r]
	if ((entry === 'vm_release' || entry === 'vm_forget') && args.vm) return vms.filter((v) => v.vm !== args.vm)
	return vms
}

// The hook each in-process server is reached by: code mode's own, and composition-local NSIDs for the rest
// (asyncai declares anyHooks), each com.fedfork.exec.mcp.<lib>.
export const hookOf = (name) => (name === 'code' ? MESSAGE : `com.fedfork.exec.mcp.${name}`)

/** The composition: the asyncai agent, each in-process server a handler on its own hook edge. */
export function agentComposition(ctx, asyncai, names) {
	const manifest = {
		$type: 'com.fedfork.exec.composition',
		name: 'atlibs-agent',
		description: 'An asyncai agent whose tools are code mode and every lib\'s actions, as in-process MCP servers.',
		instances: [{ name: 'agent', module: 'asyncai' }],
		handlers: names.map((name) => ({ name, description: `the ${name} MCP server` })),
		edges: names.map((name) => ({ from: 'agent', nsid: hookOf(name), handler: name, reason: `${name} tools` })),
	}
	return loadComposition(ctx, manifest, { load: async () => asyncai })
}

/** The largest instruct (or -it) model among ids, by its parameter count (the "32B" in its id); else the largest other than stub. */
export const pickModel = (ids) => {
	const size = (id) => Number(/(\d+(?:\.\d+)?)B/i.exec(id)?.[1] ?? 0)
	const instruct = ids.filter((id) => /instruct|-it\b/i.test(id))
	return (instruct.length ? instruct : ids.filter((id) => id !== 'stub')).sort((a, b) => size(b) - size(a))[0]
}

/** co/core's model when the user names none: pickModel over what it serves now (cocore-inference models, as the user). */
export async function cocoreModel(host, modules) {
	const ran = await runAction(host, modules['cocore-inference'], 'models', {})
	if (ran.error !== undefined) throw new Error(`co/core models: ${ran.error}`)
	const model = pickModel((ran.result?.data ?? []).map((m) => m.id))
	if (!model) throw new Error('co/core serves no model right now')
	return model
}

export const PROVIDERS = ['cocore', 'anthropic', 'custom']
export const FORMATS = ['anthropic', 'openai']

// ponytail: OpenAI chat completions arrive with atlibs-asyncai's `provider` option (and co/core's host-minted
// service auth with atlibs' `serviceAuth` allowlist option); until the asyncai module.json takes `provider`,
// those choices refuse with a clear error. Once it lands, check the field name and the allowlist shape here.
export const speaksOpenAI = (asyncai) => (asyncai?.actions ?? []).some((a) => (a.fields ?? []).some((f) => f.key === 'provider'))

/**
 * The provider the user picked -> the agent's input fields ({ endpoint, model, provider? }) and runModule's
 * http allowlist entry for it. The credential (an API key, a token, or co/core's service auth) is only ever
 * a host header or a host-minted token: the module never sees it.
 *   cocore: OpenAI-compatible, no key, service auth from the user's session
 *   anthropic: api.anthropic.com, x-api-key (and the header that lets a browser call it)
 *   custom: any base URL, a token, and its format: anthropic (x-api-key) or openai (Bearer)
 */
export function inference({ provider = DEFAULTS.provider, endpoint, model, apiKey, format = 'anthropic' }, asyncai) {
	if (!PROVIDERS.includes(provider)) throw new Error(`inference provider ${provider}: one of ${PROVIDERS.join(', ')}`)
	if (!FORMATS.includes(format)) throw new Error(`format ${format}: one of ${FORMATS.join(', ')}`)
	const openai = provider === 'cocore' || (provider === 'custom' && format === 'openai')
	if (openai && !speaksOpenAI(asyncai)) throw new Error(`${provider === 'cocore' ? 'co/core' : 'OpenAI-format'} inference needs atlibs-asyncai's provider option, which this asyncai build lacks: pick Anthropic, or a custom endpoint in anthropic format`)
	const base = provider === 'cocore' ? COCORE.endpoint : provider === 'anthropic' ? ANTHROPIC : endpoint
	if (!base) throw new Error('a custom provider needs its base URL')
	const origin = new URL(base).origin
	const auth = provider === 'cocore' ? { serviceAuth: COCORE.serviceAuth } : { headers: { ...(apiKey && (openai ? { authorization: `Bearer ${apiKey}` } : { 'x-api-key': apiKey })), ...(origin === ANTHROPIC && { 'anthropic-dangerous-direct-browser-access': 'true' }) } }
	return { input: { endpoint: base, model: model || DEFAULTS.model, ...(openai && { provider: 'openai' }) }, http: { [origin]: auth } }
}

/** Remote MCP servers as the user typed them: [{ name, url }], https (or loopback http), names not taken. */
export function remoteServers(text, taken) {
	const list = text.trim() ? JSON.parse(text) : []
	if (!Array.isArray(list)) throw new Error('remote MCP servers: a JSON array of { "name", "url" }')
	for (const s of list) {
		if (!/^[A-Za-z0-9_-]{1,32}$/.test(s?.name ?? '')) throw new Error(`remote MCP server name ${JSON.stringify(s?.name)}: 1 to 32 of A-Z a-z 0-9 _ -`)
		if (taken.includes(s.name) || list.filter((x) => x.name === s.name).length > 1) throw new Error(`remote MCP server name ${s.name} is taken`)
		const url = new URL(s.url)
		if (url.protocol !== 'https:' && !(url.protocol === 'http:' && ['127.0.0.1', 'localhost'].includes(url.hostname))) throw new Error(`${s.url}: https only`)
	}
	return list.map(({ name, url }) => ({ name, url }))
}

export const systemPrompt = (did) =>
	[
		'You are an agent in the user\'s browser tab. Your tools are MCP servers: code (run AssemblyScript in a sandbox), knotless (git repos), marque (DNS), fedproxy (SSH keys) and vm (socialweb.computer VMs), each acting as the signed-in user within that lib\'s own OAuth scope.',
		'A VM: vm__vm_create {} rents a Linux VM (root shell) and keeps it; it takes a few minutes. Run commands with vm__vm_exec {vm, command}. vm__vm_list shows the VMs you keep. When finished, always vm__vm_release {vm}, even after an error; the bidder ends it at its expires anyway (vm_create\'s hold_minutes). Only if vm_release keeps failing, vm__vm_forget {vm}. If vm_create fails, tell the user its error and what it suggests: do not go looking for bidders or records yourself.',
		'Tools that change the user\'s data ask the user first; a declined call comes back as an error: do not retry it.',
		did ? `The signed-in user is ${did}.` : 'Nobody is signed in: only public reads work.',
		'Code mode is for calling the other servers\' read-only tools and filtering what they return, not for network calls of its own: run.call(id, "com.fedfork.exec.mcp.<server>", <a JSON-RPC tools/call message>) runs one; its result is the next event, run.output the JSON-RPC response.',
	].join('\n')

const textOf = (content) => (typeof content === 'string' ? content : Array.isArray(content) && content.every((c) => c?.type === 'text') ? content.map((c) => c.text).join('') : JSON.stringify(content))

/**
 * A transcript (Anthropic Messages or OpenAI chat completions form) as display rows:
 * { role, kind: text | thinking | tool_use | tool_result, text, error?, id?, name?, input? } (id pairs a tool_use with its result).
 */
export function rows(messages = []) {
	return messages.flatMap((m) => {
		const { role, content } = m
		if (role === 'tool') return [{ role, kind: 'tool_result', id: m.tool_call_id, text: textOf(content) }]
		const blocks = typeof content === 'string' ? (content ? [{ type: 'text', text: content }] : []) : (content ?? [])
		const out = blocks.map((b) => {
			if (b.type === 'tool_use') return { role, kind: 'tool_use', id: b.id, name: b.name, input: b.input, text: `${b.name} ${JSON.stringify(b.input)}` }
			if (b.type === 'tool_result') return { role, kind: 'tool_result', id: b.tool_use_id, text: textOf(b.content), error: b.is_error === true }
			if (b.type === 'thinking') return { role, kind: 'thinking', text: b.thinking ?? '' }
			return { role, kind: b.type, text: b.text ?? JSON.stringify(b) }
		})
		for (const c of m.tool_calls ?? []) {
			let input
			try {
				input = JSON.parse(c.function?.arguments ?? '{}')
			} catch {
				input = c.function?.arguments
			}
			out.push({ role, kind: 'tool_use', id: c.id, name: c.function?.name, input, text: `${c.function?.name} ${c.function?.arguments ?? ''}` })
		}
		return out
	})
}

/** The servers the agent connected (mcp_connect) and has not disconnected, from the transcript's rows: [{ name, url, tools }]. */
export function connectedOf(list) {
	const results = new Map(list.filter((r) => r.kind === 'tool_result').map((r) => [r.id, r]))
	const live = new Map()
	for (const r of list) {
		if (r.kind !== 'tool_use' || !['mcp_connect', 'mcp_disconnect'].includes(r.name) || !r.input?.name) continue
		const out = results.get(r.id)
		if (!out || out.error) continue
		if (r.name === 'mcp_disconnect') live.delete(r.input.name)
		else {
			let value
			try {
				value = JSON.parse(out.text)
			} catch {
				continue
			}
			if (value?.connected) live.set(r.input.name, { name: r.input.name, url: r.input.server?.url ?? r.input.server?.hook, tools: value.tools ?? [] })
		}
	}
	return [...live.values()]
}

/**
 * One agent turn: progress(rows) as the transcript grows; resolves to runCompose's result (result.messages:
 * the next turn's memory). provider ... format: inference() above. remote: [{ name, url }] over Streamable HTTP, their origins allowlisted.
 * fetchers: the host's scheme fetchers (the webrtc transport's): with them the agent may mcp_connect a VM's
 * shell (config.js DYNAMIC_MCP) and the allowlist admits webrtc://. host: { ctx, policy, session }: the agent's session
 * (asyncai's), with which the host mints co/core's service-auth token.
 */
export function runAgent(ctx, composition, servers, { prompt, system, provider, endpoint, model, apiKey, format, maxSteps, remote = [], messages }, { progress, signal, fetchers, host } = {}) {
	const names = Object.keys(servers)
	const infer = inference({ provider, endpoint, model, apiKey, format }, composition.modules.agent)
	const dynamic = fetchers ? { dynamicMcp: DYNAMIC_MCP } : {}
	const input = { prompt, system, ...infer.input, maxSteps: maxSteps || DEFAULTS.maxSteps, messages, mcp: [...names.map((name) => ({ name, hook: hookOf(name) })), ...remote], ...dynamic }
	// The host holds the transcript (com.fedfork.exec.held.append); the page follows it from the agent's
	// appends (the first carries the earlier messages too).
	const shown = []
	return runCompose(ctx, composition, 'agent', 'run', input, {
		handlers: Object.fromEntries(names.map((name) => [name, servers[name].handler])),
		callFor: host && ((inst) => hostOptions(host, inst.module).call),
		// remote MCP servers' origins, then the inference origin with its credential
		http: { ...Object.fromEntries(remote.map((s) => [new URL(s.url).origin, {}])), ...(fetchers && Object.fromEntries(Object.keys(fetchers).map((scheme) => [`${scheme}//*`, {}]))), ...infer.http },
		fetchers,
		budget: UNLIMITED,
		limits: { calls: Infinity },
		signal,
		onEvent: (event, out, instance) => {
			if (instance !== 'agent') return
			const appended = (out.calls ?? []).filter((c) => c?.nsid === 'com.fedfork.exec.held.append' && c.input?.name === 'messages')
			for (const c of appended) shown.push(...(c.input.items ?? []))
			if (appended.length) progress?.(rows(shown))
		},
	})
}

/**
 * The public face of the servers for the ingress: /mcp/<name> for each, with only the tools isPublic(server, tool) admits.
 * Write tools still ask the user (confirm) when a remote client calls them.
 */
export const publicMcp = (servers, isPublic) => ingressMcp(() => Object.fromEntries(Object.entries(servers).map(([name, s]) => [name, exposed(s.message, (tool) => isPublic(name, tool))])))

/**
 * Serve `handle` (forward requests -> { status, body, contentType }) at https://<ingressHost(key.did)>/ until
 * `signal` aborts: the fedproxy ingress module in forward mode, with no total calls, events or wall time.
 */
export function serveIngress(ctx, ingress, key, handle, { signal } = {}) {
	const relay = didkeyCall(key, hostCall({ ctx, canHeader: false }), { services: ingress.didkey.services, fetch: (u, init) => ctx.http.fetch(u, init), canHeader: false })
	const call = (nsid, request) => (nsid === FORWARD ? handle(request.input) : relay(nsid, request))
	return runModule(ctx, ingress, 'serve', { forward: true }, call, { budget: UNLIMITED, signal })
}