// pi's extensions and packages (pi coding agent, @earendil-works) run as // magpie's plugins: host.js loads this module when a plugin it is given is // one, and turns each provider the extensions register into the hooks an // OpenCode plugin gives. // // The extensions are loaded by pi's own loader (discoverAndLoadExtensions), // so they import pi's packages as in pi, and their providers are composed // by pi's ModelRuntime, as pi composes them: its sign-in (OAuth or a key), // its renewal, its models and its requests are pi's. The runtime keeps // its sign-ins in magpie's plugin-auth.json, each account as host.js keeps // an OpenCode plugin's (an OAuth one as {type: "oauth", refresh, access, // expires, …}, a key as {type: "api", key}), read in the scope of the // request's account. What else an extension registers (commands, tools, // renderers, event handlers) is for pi's own sessions and is left unused. // // A request reaches a pi provider as Anthropic's Messages (the models are // said to be @ai-sdk/anthropic's): the provider's loader gives a fetch // that reads it into pi's context, streams it with pi, and answers with // Anthropic's events. import { AsyncLocalStorage } from "node:async_hooks" import fs from "node:fs" import path from "node:path" import { pathToFileURL } from "node:url" const CODING_AGENT = "@earendil-works/pi-coding-agent" const PI_AI = "@earendil-works/pi-ai" // what pi's packages were called before @earendil-works const OLD_SCOPE = "@mariozechner/pi-" function readJSON(file) { try { return JSON.parse(fs.readFileSync(file, "utf8")) } catch { return undefined } } // isPi is whether the plugin at target is pi's rather than OpenCode's: a // package with a pi manifest, the pi-package keyword or pi among what it // depends on, or a file that imports pi. export function isPi(target) { const stat = fs.statSync(target, { throwIfNoEntry: false }) if (!stat) return false if (stat.isDirectory()) { const pkg = readJSON(path.join(target, "package.json")) if (!pkg) return fs.existsSync(path.join(target, "extensions")) && !fs.existsSync(path.join(target, "index.ts")) && !fs.existsSync(path.join(target, "index.js")) if (pkg.pi && typeof pkg.pi === "object") return true if (Array.isArray(pkg.keywords) && pkg.keywords.includes("pi-package")) return true const deps = { ...pkg.peerDependencies, ...pkg.dependencies } return Object.keys(deps).some((d) => d === CODING_AGENT || d === OLD_SCOPE + "coding-agent") } if (!/\.[cm]?[jt]sx?$/.test(target)) return false try { const src = fs.readFileSync(target, "utf8") return src.includes(`"${CODING_AGENT}"`) || src.includes(`'${CODING_AGENT}'`) || src.includes(`"${OLD_SCOPE}coding-agent"`) || src.includes(`'${OLD_SCOPE}coding-agent'`) } catch { return false } } // resolve finds pi's package name from the first of dirs that has it. function resolve(name, dirs) { for (const d of dirs) { try { return Bun.resolveSync(name, d) } catch {} } throw new Error(`${name} isn't installed: add it as a plugin's dependency, or reinstall the pi plugin`) } // ---- sign-ins ------------------------------------------------------------------ // toPi is an account as pi keeps its credential; fromPi the other way, // over what the account had. function toPi(a) { if (!a || typeof a !== "object") return undefined if (a.type === "oauth") return { ...a, type: "oauth" } if (a.type === "api" && typeof a.key === "string") return { type: "api_key", key: a.key, ...(a.metadata?.env ? { env: a.metadata.env } : {}) } return undefined } function fromPi(c, was) { if (c.type === "api_key") return { type: "api", key: c.key, ...(c.env ? { metadata: { ...(was?.metadata ?? {}), env: c.env } } : was?.metadata ? { metadata: was.metadata } : {}) } const { type: _t, ...rest } = c return { type: "oauth", ...rest } } // ---- the context pi is asked with ------------------------------------------------ // the signatures Gemini gives its tool calls (thoughtSignature), which // Anthropic's tool_use has no field for: kept here by the call's id, to be // given back with the conversation. const callSigs = new Map() const SIGS_KEPT = 4000 function keepSig(id, sig) { if (!id || !sig) return callSigs.delete(id) callSigs.set(id, sig) if (callSigs.size > SIGS_KEPT) callSigs.delete(callSigs.keys().next().value) } const zeroUsage = () => ({ input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }) async function imageOf(b) { const s = b.source ?? {} if (s.type === "base64") return { type: "image", data: s.data, mimeType: s.media_type ?? "image/png" } if (s.type === "url" && typeof s.url === "string") { const m = /^data:([^;,]+);base64,(.*)$/s.exec(s.url) if (m) return { type: "image", data: m[2], mimeType: m[1] } const r = await fetch(s.url) if (!r.ok) throw new Error(`couldn't fetch the image ${s.url}: ${r.status}`) return { type: "image", data: Buffer.from(await r.arrayBuffer()).toString("base64"), mimeType: r.headers.get("content-type")?.split(";")[0] || "image/png" } } return undefined } function textOf(b) { if (b.type === "text") return b.text ?? "" if (b.type === "document") { const s = b.source ?? {} if (s.type === "text") return s.data ?? "" if (s.type === "content" && Array.isArray(s.content)) return s.content.map((x) => x.text ?? "").join("\n") } return undefined } async function contentOf(c) { const blocks = typeof c === "string" ? [{ type: "text", text: c }] : Array.isArray(c) ? c : [] const out = [] for (const b of blocks) { if (b?.type === "image") { const im = await imageOf(b) if (im) out.push(im) continue } const t = textOf(b ?? {}) if (t !== undefined) out.push({ type: "text", text: t }) } return out } // contextOf is an Anthropic Messages request as pi's context, its // assistant turns said to be the model's own so their signatures go back. export async function contextOf(body, model) { const now = Date.now() const system = typeof body.system === "string" ? body.system : Array.isArray(body.system) ? body.system.map((b) => b?.text ?? "").filter(Boolean).join("\n\n") : "" const names = new Map() const messages = [] for (const m of body.messages ?? []) { const blocks = typeof m.content === "string" ? [{ type: "text", text: m.content }] : Array.isArray(m.content) ? m.content : [] if (m.role === "assistant") { const content = [] for (const b of blocks) { if (b.type === "text") content.push({ type: "text", text: b.text ?? "" }) else if (b.type === "thinking") content.push({ type: "thinking", thinking: b.thinking ?? "", ...(b.signature ? { thinkingSignature: b.signature } : {}) }) else if (b.type === "redacted_thinking") content.push({ type: "thinking", thinking: "", thinkingSignature: b.data, redacted: true }) else if (b.type === "tool_use") { names.set(b.id, b.name) const sig = callSigs.get(b.id) content.push({ type: "toolCall", id: b.id, name: b.name, arguments: b.input ?? {}, ...(sig ? { thoughtSignature: sig } : {}) }) } } if (!content.length) continue messages.push({ role: "assistant", content, api: model.api, provider: model.provider, model: model.id, usage: zeroUsage(), stopReason: content.some((c) => c.type === "toolCall") ? "toolUse" : "stop", timestamp: now, }) continue } let user = [] const flush = () => { if (user.length) messages.push({ role: "user", content: user, timestamp: now }) user = [] } for (const b of blocks) { if (b.type === "tool_result") { flush() messages.push({ role: "toolResult", toolCallId: b.tool_use_id, toolName: names.get(b.tool_use_id) ?? "", content: await contentOf(b.content), isError: !!b.is_error, timestamp: now, }) } else if (b.type === "image") { const im = await imageOf(b) if (im) user.push(im) } else { const t = textOf(b) if (t !== undefined) user.push({ type: "text", text: t }) } } flush() } const tools = (body.tools ?? []) .filter((t) => t && t.name && (t.input_schema || !t.type || t.type === "custom")) .map((t) => ({ name: t.name, description: t.description ?? "", parameters: t.input_schema ?? { type: "object", properties: {} } })) return { ...(system ? { systemPrompt: system } : {}), messages, ...(tools.length ? { tools } : {}) } } // levelOf is how hard the request asks the model to think, as pi's // levels say it: output_config.effort, else thinking's budget; none when // thinking is off or the model doesn't think. A model that can't be told // not to think (pi's thinkingLevelMap has off null: gpt-oss and Claude on // Antigravity, whose vendor turns a request without a level away) thinks // as little as it can, as pi asks it. export function levelOf(body, model, clamp) { if (!model.reasoning) return undefined const th = body.thinking if (!th || th.type === "disabled") { if (model.thinkingLevelMap?.off !== null || !clamp) return undefined const l = clamp(model, "minimal") return l === "off" ? undefined : l } let level = body.output_config?.effort if (level === "minimal" || level === "low" || level === "medium" || level === "high" || level === "xhigh" || level === "max") { } else if (th.type === "enabled" && th.budget_tokens > 0) { const n = th.budget_tokens level = n <= 4096 ? "low" : n <= 12000 ? "medium" : n <= 24000 ? "high" : "xhigh" } else { level = "high" } const l = clamp ? clamp(model, level) : level return l === "off" ? undefined : l } // ---- answering as Anthropic ---------------------------------------------------- const STOP = { stop: "end_turn", length: "max_tokens", toolUse: "tool_use" } function usageOf(u) { return { input_tokens: u?.input ?? 0, output_tokens: u?.output ?? 0, cache_read_input_tokens: u?.cacheRead ?? 0, cache_creation_input_tokens: u?.cacheWrite ?? 0, } } function sse(type, data) { return `event: ${type}\ndata: ${JSON.stringify({ type, ...data })}\n\n` } // statusOf is the status a failure is answered with: the vendor's, else // one its message names, else 502. function statusOf(res, msg) { if (res?.status >= 400) return res.status if (/prompt is too long|context (length|window)|too many tokens|maximum context/i.test(msg)) return 400 const m = /\b(4\d\d|5\d\d)\b/.exec(msg) if (m) return Number(m[1]) if (/not configured|no api key|unauthori[sz]ed|sign in|log ?in again|invalid_grant/i.test(msg)) return 401 if (/quota|rate.?limit|too many requests|usage limit/i.test(msg)) return 429 return 502 } function errorType(status) { return ( { 400: "invalid_request_error", 401: "authentication_error", 403: "permission_error", 404: "not_found_error", 413: "request_too_large", 429: "rate_limit_error", 529: "overloaded_error" }[status] ?? "api_error" ) } function failure(status, message, headers = {}) { return new Response(JSON.stringify({ type: "error", error: { type: errorType(status), message } }), { status, headers: { "content-type": "application/json", ...headers }, }) } // kept are the vendor's headers worth passing on with a failure. function kept(res) { const out = {} for (const [k, v] of Object.entries(res?.headers ?? {})) if (/^(retry-after|x-ratelimit-|anthropic-ratelimit-)/i.test(k)) out[k] = String(v) return out } // emptyOf is whether a message says nothing at all. function emptyOf(m) { return !(m?.content ?? []).some((c) => (c.type === "text" && c.text) || (c.type === "thinking" && (c.thinking || c.thinkingSignature)) || c.type === "toolCall") } // said is what a request's own fetches were last answered with, when that // wasn't a stream: a package that reads only streams (pi-zcode) passes a // vendor's refusal sent as JSON over in silence, and its answer is empty. const said = new AsyncLocalStorage() let watching = false function watchFetch() { if (watching) return watching = true const was = globalThis.fetch globalThis.fetch = Object.assign(function fetch(input, init) { const s = said.getStore() const p = was(input, init) if (!s) return p return p.then((res) => { if (/json|text\/plain/i.test(res.headers.get("content-type") ?? "") && Number(res.headers.get("content-length") ?? 0) <= 65536 && res.body) s.body = res .clone() .text() .catch(() => "") return res }) }, was) } // whyOf is what a vendor's JSON body said went wrong, its code after it. async function whyOf(s) { if (!s?.body) return "" const text = await Promise.race([s.body, new Promise((ok) => setTimeout(() => ok(""), 1000))]) let v try { v = JSON.parse(text) } catch { return "" } const e = v?.error ?? v const msg = typeof e === "string" ? e : e?.message ?? e?.msg ?? v?.msg ?? v?.message ?? v?.detail if (typeof msg !== "string" || !msg) return "" const code = (typeof e === "object" ? e?.code : undefined) ?? v?.code return code === undefined || code === null || code === "" ? msg : `${msg} (${code})` } // answer turns pi's events into Anthropic's: an error before anything was // said as an HTTP failure, else a stream (or, unstreamed, one message). async function answer(events, { model, stream, res, said }) { const it = events[Symbol.asyncIterator]() const first = [] // wait for the first thing said, so a failure can still be a status for (;;) { const r = await it.next() if (r.done) break first.push(r.value) const t = r.value.type if (t === "error") { const msg = r.value.error?.errorMessage || "the request failed" if (r.value.reason === "aborted") return failure(499, msg) const st = statusOf(res(), msg) return failure(st, st === 400 && !/prompt is too long/i.test(msg) && /context|too long|too many tokens/i.test(msg) ? "prompt is too long: " + msg : msg, kept(res())) } if (t !== "start") break } // nothing said, while the vendor said why in a body the package didn't // read (pi-zcode's {"code":1005,"msg":"exceed quota limit"}) const last = first.at(-1) if (last?.type === "done" && emptyOf(last.message)) { const why = await whyOf(said) if (why) return failure(statusOf(res(), why), why, kept(res())) } const id = "msg_pi_" + Math.random().toString(36).slice(2, 14) async function* all() { yield* first for (;;) { const r = await it.next() if (r.done) return yield r.value } } if (!stream) { const content = [] let stop = "end_turn" let usage = usageOf() for await (const e of all()) { if (e.type === "error") { const msg = e.error?.errorMessage || "the request failed" return failure(statusOf(res(), msg), msg, kept(res())) } if (e.type !== "done") continue for (const c of e.message.content ?? []) { if (c.type === "text") c.text && content.push({ type: "text", text: c.text }) else if (c.type === "thinking") { if (c.redacted) content.push({ type: "redacted_thinking", data: c.thinkingSignature ?? "" }) else if (c.thinking || c.thinkingSignature) content.push({ type: "thinking", thinking: c.thinking ?? "", signature: c.thinkingSignature ?? "" }) } else if (c.type === "toolCall") { keepSig(c.id, c.thoughtSignature) content.push({ type: "tool_use", id: c.id, name: c.name, input: c.arguments ?? {} }) } } stop = STOP[e.reason] ?? "end_turn" usage = usageOf(e.message.usage) } return new Response(JSON.stringify({ id, type: "message", role: "assistant", model, content, stop_reason: stop, stop_sequence: null, usage }), { status: 200, headers: { "content-type": "application/json" }, }) } const enc = new TextEncoder() const body = new ReadableStream({ async start(ctl) { const put = (type, data) => ctl.enqueue(enc.encode(sse(type, data))) put("message_start", { message: { id, type: "message", role: "assistant", model, content: [], stop_reason: null, stop_sequence: null, usage: usageOf() } }) let index = -1 const open = new Map() // pi's content index → ours const begin = (ci, block) => { index++ open.set(ci, index) put("content_block_start", { index, content_block: block }) return index } const end = (ci) => { if (!open.has(ci)) return put("content_block_stop", { index: open.get(ci) }) open.delete(ci) } try { for await (const e of all()) { switch (e.type) { // a block is begun with what it says first: an empty one (a // vendor's text with nothing in it) would be sent back, and // Anthropic's Messages turn an empty text block away case "text_start": case "thinking_start": break case "text_delta": if (!e.delta) break if (!open.has(e.contentIndex)) begin(e.contentIndex, { type: "text", text: "" }) put("content_block_delta", { index: open.get(e.contentIndex), delta: { type: "text_delta", text: e.delta } }) break case "text_end": end(e.contentIndex) break case "thinking_delta": if (!e.delta) break if (!open.has(e.contentIndex)) begin(e.contentIndex, { type: "thinking", thinking: "" }) put("content_block_delta", { index: open.get(e.contentIndex), delta: { type: "thinking_delta", thinking: e.delta } }) break case "thinking_end": { const c = e.partial?.content?.[e.contentIndex] if (c?.redacted) { end(e.contentIndex) if (c.thinkingSignature) begin(e.contentIndex, { type: "redacted_thinking", data: c.thinkingSignature }) end(e.contentIndex) break } if (!open.has(e.contentIndex) && c?.thinkingSignature) begin(e.contentIndex, { type: "thinking", thinking: "" }) if (open.has(e.contentIndex) && c?.thinkingSignature) put("content_block_delta", { index: open.get(e.contentIndex), delta: { type: "signature_delta", signature: c.thinkingSignature } }) end(e.contentIndex) break } case "toolcall_end": { // a call is told whole, as its arguments are only JSON at its end const c = e.toolCall keepSig(c.id, c.thoughtSignature) const i = begin(e.contentIndex, { type: "tool_use", id: c.id, name: c.name, input: {} }) put("content_block_delta", { index: i, delta: { type: "input_json_delta", partial_json: JSON.stringify(c.arguments ?? {}) } }) end(e.contentIndex) break } case "done": for (const ci of [...open.keys()]) end(ci) put("message_delta", { delta: { stop_reason: STOP[e.reason] ?? "end_turn", stop_sequence: null }, usage: usageOf(e.message?.usage) }) put("message_stop", {}) break case "error": { for (const ci of [...open.keys()]) end(ci) const msg = e.error?.errorMessage || "the request failed" const st = statusOf(res(), msg) put("error", { error: { type: errorType(st), message: msg } }) break } } } } catch (e) { put("error", { error: { type: "api_error", message: String(e?.message ?? e) } }) } ctl.close() }, }) return new Response(body, { status: 200, headers: { "content-type": "text/event-stream", "cache-control": "no-cache" } }) } // ---- loading ------------------------------------------------------------------- // load loads the pi plugins in list ({spec, target}) and gives, for each, // the hooks of the providers it registers, or why it couldn't load. // h is what host.js lends: readAuth, changeAuth, keyFor, send, directory. export async function load(h, list) { const home = path.join(h.directory, "pi") // pi keeps its settings and caches here, not in the user's ~/.pi. This // overrides any PI_CODING_AGENT_DIR from the user's shell: with that one, // the auth.json written below would be the user's own pi sign-ins, and // magpie would delete every sign-in it doesn't hold itself. process.env.PI_CODING_AGENT_DIR = home // pi makes it on start; packages write their files into it (pi-antigravity // its accounts) and fail with ENOENT when it isn't there fs.mkdirSync(process.env.PI_CODING_AGENT_DIR, { recursive: true }) watchFetch() const none = path.join(home, "none") // no project's or user's extensions come along const dirs = [...list.map((p) => (fs.statSync(p.target, { throwIfNoEntry: false })?.isDirectory() ? p.target : path.dirname(p.target))), path.join(h.directory, "plugins")] let pi, ai try { const entry = resolve(CODING_AGENT, dirs) pi = await import(pathToFileURL(entry).href) ai = await import(pathToFileURL(resolve(PI_AI, [path.dirname(entry), ...dirs])).href) } catch (e) { return list.map((p) => ({ spec: p.spec, error: String(e?.message ?? e) })) } // pi's runtime, its sign-ins those of the account a call is made for const chains = new Map() const credentials = { async read(id) { return toPi(h.readAuth()[h.keyFor(id)]) }, async list() { return Object.entries(h.readAuth()) .filter(([k, a]) => !k.includes("#") && toPi(a)) .map(([k, a]) => ({ providerId: k, type: toPi(a).type })) }, // one change at a time to an account, read afresh modify(id, fn) { const key = h.keyFor(id) const prev = chains.get(key) ?? Promise.resolve() const run = prev.catch(() => {}).then(async () => { const was = h.readAuth()[key] const next = await fn(toPi(was)) if (next === undefined) return toPi(was) h.changeAuth((all) => { all[key] = fromPi(next, all[key]) }) h.send({ event: "auth", provider: key.split("#")[0], account: key }) mirror() return next }) chains.set(key, run) return run }, async delete(id) { const key = h.keyFor(id) h.changeAuth((all) => { if (!(key in all)) return false delete all[key] }) mirror() }, } const mr = await pi.ModelRuntime.create({ credentials, modelsPath: null, allowModelNetwork: false, refreshOnCreate: false }) // pi's auth.json, kept as pi keeps it: each provider's sign-in (its first // account's). Packages read it back themselves rather than be given it // (readStoredCredential: pi-zcode its plans, pi-cursor its token), and // are told of it as pi tells them, with session_start: each as the host // starts, and a provider again once newly signed in (pi-zcode registers // its plans' models then). const runners = new Map() // provider id → its package's runner const mirrored = new Map() const authFile = path.join(process.env.PI_CODING_AGENT_DIR, "auth.json") let started = false function mirror() { const all = h.readAuth() const fresh = new Set() let file for (const [id, runner] of runners) { const c = toPi(all[id]) const now = c ? JSON.stringify(c) : "" // a session starts for each package as pi starts, signed in or not // (pi-provider-kiro keeps the UI its sign-in asks on from it) if (!started && runner) fresh.add(runner) if (mirrored.get(id) === now) continue if (c && !mirrored.get(id) && runner) fresh.add(runner) mirrored.set(id, now) file ??= readJSON(authFile) ?? {} if (c) file[id] = c else delete file[id] } if (file) { try { const tmp = `${authFile}.${process.pid}.tmp` fs.writeFileSync(tmp, JSON.stringify(file, null, 2), { mode: 0o600 }) fs.renameSync(tmp, authFile) } catch (e) { h.send({ event: "log", level: "error", message: `pi: auth.json: ${e?.message ?? e}` }) } } const reason = started ? "new" : "startup" for (const runner of fresh) { const before = JSON.stringify([...runners.keys()].map((id) => (mr.getModels(id) ?? []).map((m) => m.id))) runner .emit({ type: "session_start", reason }) .catch(() => {}) .then(() => { // magpie asks the models again when they changed if (JSON.stringify([...runners.keys()].map((id) => (mr.getModels(id) ?? []).map((m) => m.id))) !== before) h.send({ event: "auth" }) }) } } const hp = { ...h, mirror } const out = [] for (const p of list) { try { const r = await pi.discoverAndLoadExtensions(await entriesOf(pi, p.target, none), none, none, pi.createEventBus()) const ids = [ ...(r.runtime?.pendingProviderRegistrations ?? []).map((x) => x.name), ...(r.runtime?.pendingNativeProviderRegistrations ?? []).map((x) => x.provider.id), ] if (r.errors?.length && !r.extensions?.length) throw new Error(r.errors.map((e) => e.error).join("; ")) for (const e of r.errors ?? []) h.send({ event: "log", level: "error", message: `${p.spec}: ${e.error}` }) if (!ids.length && !r.extensions?.length) throw new Error("has no pi extension") const runner = bind(pi, r, mr, none, h, p.spec) for (const id of ids) runners.set(id, runner) const hooks = [...new Set(ids)].map((id) => ({ spec: p.spec, target: p.target, hooks: hooksOf(hp, mr, ai, id) })) out.push({ spec: p.spec, hooks }) } catch (e) { out.push({ spec: p.spec, error: String(e?.stack ?? e) }) } } mirror() started = true return out } // bind gives a package's extensions to the runtime as pi's session gives // them: the providers they registered while loading are registered, one // they register later (pi-zcode its plans, once signed in) at once, and // their events (session_start) are given pi's context, with no session // and no tools of its own, and a UI only for a sign-in's questions. // Without pi's runner the providers are registered as they were loaded, // and nothing later reaches them. function bind(pi, r, mr, cwd, h, spec) { const flush = () => { for (const x of r.runtime?.pendingProviderRegistrations ?? []) mr.registerProvider(x.name, x.config) for (const x of r.runtime?.pendingNativeProviderRegistrations ?? []) mr.registerNativeProvider(x.provider) return null } if (!pi.ExtensionRunner || !pi.ModelRegistry || !pi.SessionManager?.inMemory || !r.runtime) return flush() const nothing = () => {} let runner try { runner = new pi.ExtensionRunner(r.extensions ?? [], r.runtime, cwd, pi.SessionManager.inMemory(cwd), new pi.ModelRegistry(mr)) runner.bindCore( { sendMessage: nothing, sendUserMessage: nothing, appendEntry: nothing, setSessionName: nothing, getSessionName: () => undefined, setLabel: nothing, getActiveTools: () => [], getAllTools: () => [], getSettings: () => ({}), setActiveTools: nothing, refreshTools: nothing, getCommands: () => [], setModel: async () => false, getThinkingLevel: () => "off", setThinkingLevel: nothing, }, { getModel: () => undefined, getScopedModels: () => [], isIdle: () => true, isProjectTrusted: () => false, getSignal: () => undefined, abort: nothing, hasPendingMessages: () => false, shutdown: nothing, getContextUsage: () => undefined, compact: nothing, getSystemPrompt: () => "", executeTool: async () => { throw new Error("magpie runs no tools of pi's") }, getCallableTools: () => [], }, { registerProvider: (name, config) => mr.registerProvider(name, config), registerNativeProvider: (provider) => mr.registerNativeProvider(provider), unregisterProvider: (name) => mr.unregisterProvider(name), registerVirtualModel: (d) => mr.registerVirtualModel?.(d), unregisterVirtualModel: (provider, id) => mr.unregisterVirtualModel?.(provider, id), }, ) } catch (e) { h.send({ event: "log", level: "error", message: `${spec}: pi's runner: ${e?.message ?? e}` }) return flush() } runner.onError((e) => h.send({ event: "log", level: "error", message: `${spec}: ${e.event}: ${e.error}` })) try { runner.setUIContext?.(uiOf(runner.getUIContext()), "rpc") } catch (e) { h.send({ event: "log", level: "error", message: `${spec}: pi's UI: ${e?.message ?? e}` }) } return runner } // entriesOf is what pi loads of a package: the files its manifest names, a // directory among them read as pi's package manager reads one, for its // files (pi-xai-oauth's "./extensions") — pi's loader, given the package, // would import the directory itself and fail. async function entriesOf(pi, target, none) { if (!fs.statSync(target, { throwIfNoEntry: false })?.isDirectory() || !pi.DefaultPackageManager || !pi.SettingsManager?.inMemory) return [target] try { const pm = new pi.DefaultPackageManager({ cwd: none, agentDir: none, settingsManager: pi.SettingsManager.inMemory() }) const r = await pm.resolveExtensionSources([target], { temporary: true }) const files = (r?.extensions ?? []).filter((e) => e.enabled).map((e) => e.path) if (files.length) return files } catch {} return [target] } // ---- a package's own UI ---------------------------------------------------------- // asking is the sign-in a package's UI is shown in: pi's dialogs and its // components (ctx.ui) are asked as the sign-in's questions, as magpie asks // them, and outside one they answer as pi's do with no UI. const asking = new AsyncLocalStorage() // plain stands for pi's theme and keybindings: a colour or a style gives // back the text it is given, and no key is bound. const plain = new Proxy({}, { get: (_, k) => (k === "then" ? undefined : (...a) => [...a].reverse().find((x) => typeof x === "string") ?? "") }) function uiOf(base) { const ask = (q) => asking.getStore().prompt(q) return { ...base, get theme() { return plain }, async select(title, options) { if (!asking.getStore() || !options?.length) return undefined return ask({ type: "select", message: title, options: options.map((o) => ({ id: o, label: o })) }) }, async confirm(title, message) { if (!asking.getStore()) return false const yes = await ask({ type: "select", message: [title, message].filter(Boolean).join("\n"), options: [{ id: "yes", label: "Yes" }, { id: "no", label: "No" }] }) return yes === "yes" }, async input(title, placeholder) { if (!asking.getStore()) return undefined return ask({ type: "text", message: title, placeholder: placeholder ?? "" }) }, async editor(title, prefill) { if (!asking.getStore()) return undefined return ask({ type: "text", message: title, placeholder: prefill ?? "" }) }, notify(message) { asking.getStore()?.notify({ type: "info", message: String(message) }) }, async custom(factory) { const i = asking.getStore() return i ? shown(i, factory) : undefined }, } } // shown runs a component of pi's terminal UI (ctx.ui.custom: Kiro's choice // of sign-in) with no terminal: the list it shows is asked as a choice and // the field as text, each answer given to it as its keys would, until it // is done; one with nothing to answer (a sign-in's waiting screen) is // waited on. async function shown(i, factory) { let result let over = false let finish const done = new Promise((ok) => (finish = ok)) const tui = new Proxy({ terminal: { columns: 100, rows: 40 } }, { get: (t, k) => (k in t ? t[k] : k === "then" ? undefined : () => {}) }) const c = await factory(tui, plain, plain, (v) => { if (over) return over = true result = v finish() }) try { while (!over) { // what it does on an answer (its next screen) comes first await new Promise((ok) => setTimeout(ok, 0)) if (over) break const ctl = controlOf(c) if (!ctl) { await done break } if (ctl.list) { const items = ctl.list.filteredItems ?? ctl.list.items const v = await i.prompt({ type: "select", message: ctl.message, options: items.map((x) => ({ id: String(x.value), label: x.label ?? String(x.value), description: x.description ?? "" })) }) const item = items.find((x) => String(x.value) === String(v)) if (item) ctl.list.onSelect?.(item) else ctl.list.onCancel?.() } else { const v = String((await i.prompt({ type: "text", message: ctl.message, placeholder: "" })) ?? "") ctl.input.setValue?.(v) ctl.input.onSubmit?.(v) } } } finally { c?.dispose?.() } return result } const strip = (l) => String(l).replace(/\x1b\[[0-9;]*m/g, "").trim() // controlOf is the first list or field a component shows, with the text // shown above it, rules left out, as its question. A component may keep // its parts to itself (Kiro's gives only render and handleInput), and each // package loads its own copy of pi-tui, so its classes can't be told // apart; but drawing walks the parts in order, each container going // through its children, so the arrays walked while it draws are what it // shows. function controlOf(root) { const seen = [] const walk = Array.prototype[Symbol.iterator] Array.prototype[Symbol.iterator] = function () { for (let i = 0; i < this.length; i++) { const x = this[i] if (x && typeof x === "object" && typeof x.render === "function" && !seen.includes(x)) seen.push(x) } return walk.call(this) } try { root?.render?.(100) } catch { } finally { Array.prototype[Symbol.iterator] = walk } const at = seen.findIndex((c) => (Array.isArray(c.items) && typeof c.onSelect === "function") || typeof c.onSubmit === "function") if (at < 0) return undefined const c = seen[at] const above = seen .slice(0, at) .filter((t) => typeof t.text === "string") .map((t) => strip(t.text)) .filter((l) => l && !/^[\s─━│┃┌┐└┘├┤┬┴┼╭╮╯╰═║-]*$/.test(l)) return { [typeof c.onSelect === "function" ? "list" : "input"]: c, message: above.join("\n") } } // ---- one provider's hooks -------------------------------------------------------- const LOGIN_WAIT = 10 * 60 * 1000 function loopbackRedirect(url) { try { const r = new URL(new URL(url).searchParams.get("redirect_uri") ?? "") return ["localhost", "127.0.0.1", "[::1]"].includes(r.hostname) } catch { return false } } // a sign-in under way: pi's login runs, and what it asks or shows comes // out as events, one at a time function startLogin(oauth) { const ctl = new AbortController() const timer = setTimeout(() => ctl.abort(new Error("the sign-in timed out")), LOGIN_WAIT) const run = { ctl, n: 0, asked: null, url: null, shown: [], waiters: [], result: null, done: false } const wake = () => { for (const w of run.waiters.splice(0)) w() } run.next = () => new Promise((ok) => (run.waiters.push(ok), run.asked || run.url || run.done ? wake() : null)) const interaction = { signal: ctl.signal, prompt(q) { return new Promise((ok, no) => { run.asked = { q, ok, key: "pi" + run.n++ } ctl.signal.addEventListener("abort", () => no(ctl.signal.reason ?? new Error("Login cancelled")), { once: true }) wake() }) }, notify(e) { if (e?.type === "auth_url" && e.url) run.url ??= { url: e.url, instructions: e.instructions ?? "" } else if (e?.type === "device_code") run.url ??= { url: e.verificationUriComplete ?? e.verificationUri ?? e.verificationUrl ?? "", instructions: `Enter the code ${e.userCode}` } else if (e?.type === "info" && e.message) run.shown.push(String(e.message)) wake() }, } run.promise = Promise.resolve() .then(() => asking.run(interaction, () => oauth.login(interaction))) .then( (c) => (run.result = { ...c, type: "success" }), (e) => (run.result = { type: "failed", error: String(e?.message ?? e) }), ) .finally(() => { clearTimeout(timer) run.done = true wake() }) return run } function promptOf(run) { const { q, key } = run.asked const select = q.type === "select" && Array.isArray(q.options) && q.options.length return { type: select ? "select" : "text", key, message: q.message ?? "", placeholder: q.placeholder ?? "", // pi's options are { id, label, description } and the login is answered // with the id (pi-zcode's "bigmodel"), never with what the reader saw options: select ? q.options.map((o) => (typeof o === "string" ? { label: o, value: o, hint: "" } : { label: o.label ?? String(o.id ?? o.value), value: String(o.id ?? o.value ?? o.label), hint: o.hint ?? o.description ?? "" })) : undefined, } } function hooksOf(h, mr, ai, id) { const prov = () => mr.getProvider(id) const runs = new Map() // method index → its sign-in under way function methods() { const a = prov()?.auth ?? {} const out = [] if (a.oauth) { const m = { type: "oauth", label: a.oauth.loginLabel ?? `Sign in with ${a.oauth.name ?? prov()?.name ?? id}`, } const index = out.length // what the login asks before it shows its page, asked one at a time // as magpie asks a method's prompts (magpie's own hook) m.ask = async (inputs = {}) => { let run = runs.get(index) const answered = !!run?.asked && run.asked.key in inputs // asked with nothing answered: a sign-in begun afresh if (!run || run.done || (!answered && !Object.keys(inputs).length)) { run?.ctl.abort(new Error("Login cancelled")) run = startLogin(a.oauth) runs.set(index, run) } if (answered) { const { ok, key } = run.asked run.asked = null ok(String(inputs[key] ?? "")) } await run.next() return run.asked && !run.url ? promptOf(run) : null } m.authorize = async () => { let run = runs.get(index) if (!run || (run.done && run.result?.type !== "success" && !run.url)) { run = startLogin(a.oauth) runs.set(index, run) } while (!run.url && !run.done) { // a question left after the page (none expected): told nothing if (run.asked) { const { ok } = run.asked run.asked = null ok("") } await run.next() } if (run.done && run.result?.type !== "success" && !run.url) { runs.delete(index) throw new Error(run.result?.error ?? "the sign-in failed") } const url = run.url?.url ?? "" // a page that comes back to a server the login runs here finishes // by itself; one that doesn't asks for what it shows to be pasted await Promise.race([run.next(), new Promise((ok) => setTimeout(ok, 300))]) const code = !!url && !loopbackRedirect(url) && !!run.asked && !run.done const instructions = [run.url?.instructions, ...run.shown].filter(Boolean).join("\n") return { url, instructions: code && !instructions ? run.asked.q.message ?? "" : instructions, method: code ? "code" : "auto", callback: async (pasted) => { runs.delete(index) if (pasted && run.asked) { const { ok } = run.asked run.asked = null ok(String(pasted)) } await run.promise return run.result }, } } out.push(m) } if (a.apiKey || !a.oauth) out.push({ type: "api", label: "API key" }) return out } const auth = { provider: id, get methods() { return methods() }, async loader() { return { apiKey: "pi", baseURL: "http://pi.magpie/v1", fetch: (url, init) => piFetch(url, init) } }, // renewed by magpie ahead of its expiry, as pi renews it async refresh(stored) { const oauth = prov()?.auth?.oauth if (!oauth?.refresh) return undefined const { type: _t, ...c } = await oauth.refresh({ ...stored, type: "oauth" }) return c }, } async function piFetch(url, init) { let body try { body = JSON.parse(typeof init?.body === "string" ? init.body : new TextDecoder().decode(init?.body ?? new Uint8Array())) } catch { return failure(400, "pi's providers take Anthropic's Messages requests") } const model = mr.getModel(id, body.model) if (!model) return failure(404, `${prov()?.name ?? id} has no model ${body.model}`) let context try { context = await contextOf(body, model) } catch (e) { return failure(400, String(e?.message ?? e)) } let res const seen = {} const max = Number(body.max_tokens) > 0 ? Number(body.max_tokens) : undefined const opts = { signal: init?.signal, maxTokens: max && model.maxTokens > 0 ? Math.min(max, model.maxTokens) : max ?? (model.maxTokens || undefined), reasoning: levelOf(body, model, ai.clampThinkingLevel), ...(typeof body.temperature === "number" ? { temperature: body.temperature } : {}), ...(body.tool_choice?.type === "none" ? { toolChoice: "none" } : body.tools?.length ? { toolChoice: "auto" } : {}), onResponse: (r) => void (res = r), } let events try { events = said.run(seen, () => mr.streamSimple(model, context, opts)) } catch (e) { const msg = String(e?.message ?? e) return failure(statusOf(undefined, msg), msg) } return answer(events, { model: body.model, stream: body.stream !== false, res: () => res, said: seen }) } return { auth, provider: { id, async models(_given, { auth: stored }) { h.mirror() if (stored && prov()?.refreshModels) { const r = await mr.refresh({ providers: [id], allowNetwork: true }) const e = r?.errors?.get?.(id) if (e) h.send({ event: "log", level: "error", message: `${id}: models: ${e.message ?? e}` }) } let list = mr.getModels(id) ?? [] if (stored) { try { const av = await mr.getAvailable(id) if (av?.length) list = av } catch {} } const levels = (m) => { try { return ai.getSupportedThinkingLevels(m).filter((l) => l !== "off") } catch { return m.reasoning ? ["low", "medium", "high"] : [] } } return Object.fromEntries( list .filter((m) => !m.type || m.type === "chat") .map((m) => [ m.id, { id: m.id, providerID: id, name: m.name ?? m.id, api: { id: m.id, url: "", npm: "@ai-sdk/anthropic" }, status: "active", headers: {}, options: {}, cost: { input: m.cost?.input ?? 0, output: m.cost?.output ?? 0, cache: { read: m.cost?.cacheRead ?? 0, write: m.cost?.cacheWrite ?? 0 } }, limit: { context: m.contextWindow ?? 0, output: m.maxTokens ?? 0 }, capabilities: { temperature: true, reasoning: !!m.reasoning, attachment: (m.input ?? []).includes("image"), toolcall: true, input: { text: true, image: (m.input ?? []).includes("image"), audio: false, video: false, pdf: false }, output: { text: true, image: false, audio: false, video: false, pdf: false }, interleaved: false, }, release_date: "", variants: m.reasoning ? Object.fromEntries(levels(m).map((l) => [l, {}])) : {}, }, ]), ) }, }, config(c) { c.provider ??= {} c.provider[id] ??= {} c.provider[id].name ??= prov()?.name ?? id }, } }