Files
magpie/internal/plugin/pi.js
T
yetone 0ae7330242 plugin: a pi package's sign-in screen on pi's terminal UI is asked in magpie, and an empty reply a package makes of a vendor's error says the error (all pi plugins: Kiro's login hung, pi-zcode's quota ran out as an empty answer) — pi-provider-kiro asks how to sign in through ctx.ui.custom, a component of pi-tui (a SelectList, then Input fields for a start URL and region), and magpie gave pi's runner no UI, so custom came back undefined and the sign-in waited forever. pi's runner now has a UI for the sign-in's questions: select, confirm, input and editor become magpie's sign-in prompts, and custom draws the package's component and asks its list or field the same way, answering through the component's own onSelect/onSubmit, until the package calls done. 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 the parts are found by what drawing walks: every container goes through its children, so the arrays iterated while it renders are what it shows, in order; the text above the list is its question. Outside a sign-in the UI answers as pi's no-UI context does. session_start now reaches every package at startup, which Kiro needs to keep the context it asks through. Separately, pi-zcode turns ZCode's 1005 quota error into a reply with no content; magpie keeps the small JSON or text bodies a package's fetch gets while it streams, and a stream that ends empty answers that body's message and code (429 for quota or rate limits, else 502) instead of an empty message. TestPiPackage drives a fake package that asks on real pi-tui classes (a list, then a start URL field) and fakes ZCode's empty quota reply; it fails on the old code (no first prompt) and passes. With real accounts in a sandbox HOME: Kiro's "Kiro Login" (API Key, Web Login, Device Code) came as a select, Web Login gave the app.kiro.dev sign-in, and after signing in kiro-plugin/claude-sonnet-4-5 called get_weather and answered with its result. pi-cursor (1.4.38, latest) answers plain turns but its models say no get_weather tool exists, though the package sends it (toolCount 1, its MCP schema) and Cursor's stream carries fields the package doesn't know; that is between pi-cursor and Cursor. The ZCode quota path wasn't tried with a real account. go vet, builds (darwin, GOOS=linux, GOOS=windows) and go test -tags nogui -count=1 ./... under a temp HOME pass.
2026-10-03 23:47:15 +08:00

1094 lines
43 KiB
JavaScript

// 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
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
},
}
}