johnandersen777.bsky.social/atlibs-asyncai

Switch repository

Loading…

// The page's logic, DOM-free (agent.test.mjs runs it in Node): the inference presets, the host allowlist and the agent run.
import { policyCall } from "./atlibs/oauth/index.js";
import { hostCall, sessionXrpc } from "./atlibs/run-module/host-call.js";
import { runModule } from "./atlibs/run-module/index.js";

export const ANTHROPIC = "https://api.anthropic.com";
export const COCORE = "https://cocore.dev/api/v1";
/** co/core's service auth: the token the host mints from the signed-in session for cocore.dev (no API key). */
export const COCORE_AUTH = { aud: "did:web:cocore.dev", lxm: "dev.cocore.inference.dispatch" };
/** What signing in for co/core asks for: just that token (records/asyncai.json, the module's session). */
export const COCORE_SCOPE = `atproto rpc:${COCORE_AUTH.lxm}?aud=${COCORE_AUTH.aud}`;

/**
 * The inference settings a preset stands for: co/core (OpenAI chat completions, the user's own session, no key),
 * Anthropic (the Messages API, a key), or custom (any base URL, either format, a token).
 * @returns {{ provider: "anthropic" | "openai", endpoint: string, auth: "session" | "key" }}
 */
export function preset({ backend = "cocore", endpoint, format }) {
	if (backend === "cocore") return { provider: "openai", endpoint: COCORE, auth: "session" };
	if (backend === "anthropic") return { provider: "anthropic", endpoint: ANTHROPIC, auth: "key" };
	if (!endpoint) throw new Error("a custom backend needs its base URL");
	return { provider: format === "openai" ? "openai" : "anthropic", endpoint, auth: "key" };
}

/**
 * runModule's http: each remote MCP server's origin (and each URL prefix the agent may connect, dynamicMcp; a webrtc:// or
 * iroh:// one as its scheme pattern, webrtc://*, for the host's fetcher), and the inference origin with its credentials, which stay
 * here in the host (the module never sees them): co/core's service-auth token, minted per request; a key as
 * x-api-key (Anthropic's format, plus the header that lets a browser call api.anthropic.com) or as
 * Authorization: Bearer (OpenAI's).
 */
export function allowlist({ provider, endpoint, auth, apiKey, mcp = [], dynamicMcp = [] }) {
	const origin = new URL(endpoint).origin;
	const urls = [...mcp.map((s) => s.url ?? s.server?.url), ...dynamicMcp.filter((d) => /^[a-z][a-z0-9+.-]*:\/\//i.test(d))].filter(Boolean);
	const remote = Object.fromEntries(urls.map((u) => new URL(u)).map((u) => [/^https?:$/.test(u.protocol) ? u.origin : `${u.protocol}//*`, {}]));
	if (auth === "session") return { ...remote, [origin]: { serviceAuth: COCORE_AUTH } };
	const key = apiKey && (provider === "openai" ? { authorization: `Bearer ${apiKey}` } : { "x-api-key": apiKey });
	return { ...remote, [origin]: { headers: { ...key, ...(origin === ANTHROPIC && { "anthropic-dangerous-direct-browser-access": "true" }) } } };
}

// The first message of a transcript the module compacted (its SUMMARY_PREFIX).
const SUMMARY = "This session continues an earlier conversation that was compacted";

/** A transcript (either format) as display rows: { role, kind: text | thinking | tool_use | tool_result | compaction, text, error? }. */
export function rows(messages = []) {
	return messages.flatMap(({ role, content, tool_calls: calls }) => {
		if (role === "tool") return [{ role, kind: "tool_result", text: String(content), error: String(content).startsWith("Error: ") }];
		const blocks = (typeof content === "string" ? (content ? [{ type: "text", text: content }] : []) : content ?? []).map((b) => {
			if (b.type === "text" && b.text?.startsWith(SUMMARY)) return { role, kind: "compaction", text: b.text };
			if (b.type === "compaction") return { role, kind: "compaction", text: b.content ?? "" };
			if (b.type === "tool_use") return { role, kind: "tool_use", text: `${b.name} ${JSON.stringify(b.input)}` };
			if (b.type === "tool_result") return { role, kind: "tool_result", text: typeof b.content === "string" ? b.content : JSON.stringify(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) };
		});
		return [...blocks, ...(calls ?? []).map((c) => ({ role, kind: "tool_use", text: `${c.function?.name} ${c.function?.arguments}` }))];
	});
}

/** result.servers (the live set: the start's and those the agent connected, less the disconnected) as display lines. */
export function serverLines(servers = []) {
	return servers.map((s) => [s.name, s.url ?? s.hook, s.phase + (s.error ? ` (${s.error})` : ""), `${s.tools.length} tools`, s.dynamic && "connected by the agent"].filter(Boolean).join(" · "));
}

/**
 * One run: progress(rows, { compacting }) after every step (compacting: a summary request is out); resolves to
 * runModule's result (result.messages: the next run's memory, handed back as the held `messages`). session: the
 * signed-in OAuth session co/core's token is minted from, the run held to the module's session scope. fetchers: runModule's,
 * the host transports for webrtc:// and iroh:// MCP URLs (a transport plugin's fetchers()).
 */
export function runAgent(ctx, mod, { backend, endpoint, format, prompt, system, model, apiKey, mcp = [], dynamicMcp = [], maxSteps, messages, compact, contextWindow }, { progress, signal, session, fetchers } = {}) {
	const settings = preset({ backend, endpoint, format });
	if (settings.auth === "session" && !session) throw new Error("co/core runs as you: sign in first");
	const input = { prompt, system, provider: settings.provider, endpoint: settings.endpoint, model, maxSteps, mcp, compact, contextWindow, ...(dynamicMcp.length && { dynamicMcp }) };
	// The host holds the transcript; the page follows it from the module's calls: an append adds to it, a put (a
	// compaction) restarts it.
	let shown = [...(messages ?? [])];
	const signedOut = async (nsid) => {
		throw new Error(`${nsid}: this page makes no XRPC calls signed out`);
	};
	const policy = { sessions: { asyncai: { scope: COCORE_SCOPE } }, modules: { asyncai: { session: "asyncai" } } };
	const call = session ? policyCall(policy, "asyncai", { asyncai: hostCall({ ctx, xrpc: sessionXrpc(session), canHeader: false }) }) : signedOut;
	return runModule(ctx, mod, "run", input, call, {
		http: allowlist({ ...settings, apiKey, mcp, dynamicMcp }),
		fetchers,
		held: messages?.length ? { messages } : undefined,
		signal,
		// a long run: the step cap bounds it
		budget: { calls: Infinity, events: Infinity, wallMs: Infinity },
		onEvent: (event, out) => {
			const calls = (out.calls ?? []).filter((c) => c?.nsid?.startsWith("com.fedfork.exec.held.") && c.input?.name === "messages");
			for (const call of calls) shown = call.nsid.endsWith(".put") ? [...call.input.value] : [...shown, ...(call.input.items ?? [])];
			const compacting = (out.calls ?? []).some((c) => c?.id?.startsWith("compact."));
			if (calls.length || compacting) progress?.(rows(shown), { compacting });
		},
	});
}