Unpriced turns
@@ -170,7 +171,7 @@ export function StatsBar({
{turn.steps.length} steps
- {turn.model.startsWith("chatgpt:") ? "ChatGPT plan" : c === null ? "—" : formatUsd(c)}
+ {subscriptionLabel(turn.model) ?? (c === null ? "—" : formatUsd(c))}
);
diff --git a/components/usage-format.ts b/components/usage-format.ts
index f4e066f..3a400cd 100644
--- a/components/usage-format.ts
+++ b/components/usage-format.ts
@@ -8,9 +8,9 @@ export function formatUsd(c: number): string {
return c >= 1 ? `$${c.toFixed(2)}` : `$${c.toFixed(4)}`;
}
-export function planCostLabel(dollars: number | null, subscription: boolean, format = formatUsd): string {
- if (subscription) return dollars === null ? "ChatGPT plan" : `${format(dollars)} API + ChatGPT plan`;
- return dollars === null ? "—" : format(dollars);
+export function planCostLabel(dollars: number | null, subscription: boolean | string | readonly string[] | undefined, format = formatUsd): string {
+ const labels = subscription === true ? ["ChatGPT"] : typeof subscription === "string" ? [subscription] : Array.isArray(subscription) ? [...new Set(subscription)] : [];
+ return [...(dollars === null ? [] : [format(dollars)]), ...labels].join(" + ") || "—";
}
/** 38000 → "38k", 1234567 → "1.2M" — trimmed, never "38.0k". */
diff --git a/docs/architecture.md b/docs/architecture.md
index 2e8fa42..1fcce49 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -399,7 +399,7 @@ Practice turns the graph into problem sets the learner reasons through out loud.
```mermaid
flowchart LR
subgraph DESKTOP["Electron app (production)"]
- MAIN["Electron main (electron/main.cjs)
thin launcher + security guard
1 IPC channel: lh:expand-window"] -->|"spawn: ELECTRON_RUN_AS_NODE,
loopback port, LH_DATA_DIR=userData/data"| NEXT["standalone Next server (child)
ALL app logic: turns · extraction · jobs · graph"]
+ MAIN["Electron main (electron/main.cjs)
launcher + credential/security owner
origin-pinned controls · private child IPC"] -->|"spawn: ELECTRON_RUN_AS_NODE,
loopback port, LH_DATA_DIR=userData/data"| NEXT["standalone Next server (child)
ALL app logic: turns · extraction · jobs · graph"]
MAIN -->|"loadURL, sandboxed renderer"| REND["Browser renderer (components/)
fetch/WebSocket to own origin only"]
end
subgraph OUTSIDE["outside the process"]
@@ -417,10 +417,18 @@ flowchart LR
REND -->|"fetch /api/*"| NEXT
```
-The Electron main is a launcher + security guard, never a host for app logic: single-instance lock, deny-by-default permissions, origin-pinned navigation, sandboxed renderer. All Next.js code runs unchanged in a child process (`output: "standalone"`), relocated by the `LH_DATA_DIR` seam (also the test seam — `data-dir-seam.test.ts` covers it, including a fresh-child-process case). The build hard-fails if repo `data/` would leak into the bundle or if prompt assets fall out of the trace (`scripts/build-server.mjs`).
+The Electron main is a launcher + credential/security guard, never a host for teaching or inference logic: single-instance lock, deny-by-default permissions, origin-pinned navigation, sandboxed renderer. All Next.js code runs unchanged in a child process (`output: "standalone"`), relocated by the `LH_DATA_DIR` seam (also the test seam — `data-dir-seam.test.ts` covers it, including a fresh-child-process case). The build hard-fails if repo `data/` would leak into the bundle or if prompt assets fall out of the trace (`scripts/build-server.mjs`).
---
+### 8.1 Subscription connection boundary
+
+`electron/chatgpt.cjs` retains the existing verified-account ChatGPT authentication, storage and revocation behavior. `electron/subscriptions.cjs` delegates GitHub Copilot/Grok login and refresh to the installed Pi SDK through its public provider/Models interfaces. Their single saved connections live independently in owner-only, atomically written `github-copilot-credentials.json` / `xai-credentials.json` beside the existing app credential files. Electron main alone owns durable OAuth credentials and serialized refresh; no global Pi files or API-key credential precedence are involved.
+
+The renderer gets projected status, public device user codes, prompts and controls over origin-pinned IPC. The privately launched server child receives credential-free catalogs and scoped request auth (including native headers/account endpoints), never refresh tokens. `src/agent/subscription-bridge.ts` rejects framework-only IPC and invalidates stale connection/catalog scopes. `subscription-stream.ts` dispatches directly through native providers, retaining native protocol identity on the wire and restoring logical app identity in assistant history. Persisted routes `chatgpt`, `github-copilot-subscription` and `xai-subscription` are distinct from ordinary API routes and never fall back to them.
+
+`subscription-request-cycle.ts` is the shared admission/automatic-input recovery state machine; the old ChatGPT module is its compatibility export. Background consent is independent per provider, default ON, with foreground-preserving cancellation and queue ownership retained on no-dispatch. Native usage rejected before dispatch is transiently excluded from helper usage collectors, without adding persisted provenance fields. Catalogs and read-time pricing preserve the selected connection: subscription use displays compact provider labels, API dollars remain dollars, and mixed usage joins both without estimating subscription dollars or claiming detected tiers/allowances.
+
## 9. The foundational invariants — what ALWAYS holds
### Tier A — machine-enforced (`eslint.config.mjs`; lint fails the build)
diff --git a/electron/main.cjs b/electron/main.cjs
index 61c5af3..46db548 100644
--- a/electron/main.cjs
+++ b/electron/main.cjs
@@ -27,6 +27,8 @@
const { app, BrowserWindow, Menu, ipcMain, session, shell, dialog } = require("electron");
const { createChatGPT, fileStore } = require("./chatgpt.cjs");
const { attachChatGPTChild, trustedSender, chatGPTControl } = require("./chatgpt-ipc.cjs");
+const { createSubscriptions, loadSubscriptionSDK } = require("./subscriptions.cjs");
+const { attachSubscriptionChild, subscriptionControl } = require("./subscription-ipc.cjs");
const { desktopAPIHeaders } = require("./desktop-api.cjs");
const { randomBytes } = require("node:crypto");
const { spawn } = require("node:child_process");
@@ -134,6 +136,8 @@ if (process.argv.includes("--browser-service")) {
let serverProcess = null;
let chatgpt = null;
let chatgptChild = null;
+let subscriptions = null;
+let subscriptionChild = null;
// Per-launch request capability, never OAuth tokens and never renderer state.
let desktopAPISecret = null;
const trustedAppWindows = new Set();
@@ -222,6 +226,7 @@ function startServer(port) {
);
serverProcess = child;
chatgptChild = attachChatGPTChild(child, chatgpt);
+ subscriptionChild = attachSubscriptionChild(child, subscriptions);
const fwd = (buf) => buf.toString().trimEnd().split("\n").forEach((l) => l && log(`server: ${l}`));
child.stdout.on("data", fwd);
child.stderr.on("data", fwd);
@@ -230,7 +235,10 @@ function startServer(port) {
serverProcess = null;
chatgptChild?.dispose();
chatgptChild = null;
+ subscriptionChild?.dispose();
+ subscriptionChild = null;
chatgpt?.close();
+ subscriptions?.close();
});
resolve(child);
});
@@ -255,6 +263,7 @@ function waitUntilUp(url, tries = 150) {
function killServer() {
chatgpt?.close();
+ subscriptions?.close();
if (!serverProcess) return;
const child = serverProcess;
serverProcess = null;
@@ -569,6 +578,7 @@ function installMenu() {
async function main() {
app.on("window-all-closed", () => {
chatgpt?.pause();
+ subscriptions?.pause();
// macOS convention: the app stays resident, reactivated from the Dock.
if (process.platform !== "darwin") app.quit();
});
@@ -600,6 +610,16 @@ async function main() {
for (const win of BrowserWindow.getAllWindows()) win.webContents.send("lh:chatgpt:changed");
},
});
+ subscriptions = createSubscriptions({
+ directory: app.getPath("userData"),
+ loadSDK: () => loadSubscriptionSDK(path.dirname(serverEntry())),
+ openExternal: (target) => shell.openExternal(target),
+ onChange: (snapshot) => {
+ subscriptionChild?.send(snapshot);
+ for (const win of BrowserWindow.getAllWindows()) win.webContents.send("lh:subscriptions:changed");
+ },
+ });
+ await subscriptions.ready;
log(`starting server: ${serverEntry()}`);
log(`data dir: ${DATA_DIR()}${fs.existsSync(AUTH_FILE()) ? " · auth.json: present" : " · auth.json: ABSENT (no BYOK keys yet)"}`);
await startServer(new URL(url).port);
@@ -617,6 +637,15 @@ async function main() {
return chatGPTControl(command, chatgpt);
});
+ ipcMain.handle("lh:subscriptions:control", async (event, command) => {
+ if (!trustedAppWindows.has(BrowserWindow.fromWebContents(event.sender)) || !trustedSender(event, new URL(url).origin, BrowserWindow)) throw new Error("Untrusted desktop request.");
+ if (!subscriptions || !serverProcess?.connected) {
+ if (command?.op === "status") return [];
+ throw new Error("Subscription sign-in requires the desktop app's private server connection.");
+ }
+ return subscriptionControl(command, subscriptions);
+ });
+
installMenu();
// Auto-update (Wave C): wire the updater before the window so the
@@ -642,7 +671,7 @@ async function main() {
if (width < target) win.setBounds({ x, y, width: target, height });
});
app.on("activate", () => {
- if (BrowserWindow.getAllWindows().length === 0) { chatgpt?.resume(); createWindow(url); }
+ if (BrowserWindow.getAllWindows().length === 0) { chatgpt?.resume(); subscriptions?.resume(); createWindow(url); }
});
createWindow(url);
log(`window open: ${url}`);
diff --git a/electron/preload.cjs b/electron/preload.cjs
index 3321913..4912005 100644
--- a/electron/preload.cjs
+++ b/electron/preload.cjs
@@ -43,6 +43,21 @@ contextBridge.exposeInMainWorld("lh", {
return () => ipcRenderer.removeListener("lh:chatgpt:changed", listener);
},
},
+ subscriptions: {
+ status: () => ipcRenderer.invoke("lh:subscriptions:control", { op: "status" }),
+ login: (providerId, enabled) => ipcRenderer.invoke("lh:subscriptions:control", { op: "login", providerId, enabled }),
+ cancel: (providerId) => ipcRenderer.invoke("lh:subscriptions:control", { op: "cancel", providerId }),
+ respond: (providerId, attemptId, promptId, value) => ipcRenderer.invoke("lh:subscriptions:control", { op: "respond", providerId, attemptId, promptId, value }),
+ open: (providerId, attemptId, actionId) => ipcRenderer.invoke("lh:subscriptions:control", { op: "open", providerId, attemptId, actionId }),
+ retry: (providerId) => ipcRenderer.invoke("lh:subscriptions:control", { op: "retry", providerId }),
+ disconnect: (providerId) => ipcRenderer.invoke("lh:subscriptions:control", { op: "disconnect", providerId }),
+ consent: (providerId, enabled) => ipcRenderer.invoke("lh:subscriptions:control", { op: "consent", providerId, enabled }),
+ onChange: (cb) => {
+ const listener = () => cb();
+ ipcRenderer.on("lh:subscriptions:changed", listener);
+ return () => ipcRenderer.removeListener("lh:subscriptions:changed", listener);
+ },
+ },
expandWindow: () => ipcRenderer.send("lh:expand-window"),
onZoom: (cb) => {
const listener = (_event, factor) => cb(factor);
diff --git a/electron/subscription-ipc.cjs b/electron/subscription-ipc.cjs
new file mode 100644
index 0000000..17e2f77
--- /dev/null
+++ b/electron/subscription-ipc.cjs
@@ -0,0 +1,45 @@
+"use strict";
+const { PROVIDERS } = require("./subscription-store.cjs");
+const { publicError } = require("./subscriptions.cjs");
+function attachSubscriptionChild(child, controller) {
+ let disposed = false;
+ const send = (packet) => { if (!disposed && child.connected) { try { child.send(packet, () => {}); } catch { /* child exited */ } } };
+ const listener = async (message) => {
+ if (disposed || !child.connected || message?.channel !== "lh:subscriptions" || typeof message.id !== "string") return;
+ const { id, op } = message;
+ try {
+ if (!controller || !["snapshot", "lease"].includes(op)) throw new Error();
+ if (op === "lease" && (typeof message.background !== "boolean" || !validScope(message.scope))) throw new Error();
+ const value = op === "snapshot" ? await controller.snapshot() : await controller.lease(message.scope, message.background);
+ send({ channel: "lh:subscriptions", id, op, ok: true, value });
+ } catch (error) { send({ channel: "lh:subscriptions", id, op, ok: false, error: publicError(error) }); }
+ };
+ child.on("message", listener);
+ return {
+ send(snapshot) { send({ channel: "lh:subscriptions", op: "changed", snapshot }); },
+ dispose() { disposed = true; child.removeListener("message", listener); },
+ };
+}
+function validScope(scope) {
+ return scope && PROVIDERS.includes(scope.providerId) && typeof scope.connectionId === "string" && !!scope.connectionId && typeof scope.selectedModelId === "string" && !!scope.selectedModelId && [scope.epoch, scope.authorityRevision].every((n) => Number.isSafeInteger(n) && n >= 0);
+}
+/** Caller authenticates the exact trusted main-frame origin before dispatch. */
+async function subscriptionControl(command, controller) {
+ try {
+ if (!controller || !command || typeof command !== "object") throw new Error();
+ if (command.op === "status") return await controller.status();
+ if (!PROVIDERS.includes(command.providerId)) throw new Error();
+ const id = command.providerId;
+ switch (command.op) {
+ case "login": return await controller.login(id, command.enabled);
+ case "cancel": return controller.cancel(id);
+ case "respond": return await controller.respond(id, command.attemptId, command.promptId, command.value);
+ case "open": return await controller.open(id, command.attemptId, command.actionId);
+ case "retry": return await controller.retry(id);
+ case "disconnect": return await controller.disconnect(id);
+ case "consent": return await controller.consent(id, command.enabled);
+ default: throw new Error();
+ }
+ } catch (error) { throw new Error(publicError(error)); }
+}
+module.exports = { attachSubscriptionChild, subscriptionControl, validScope };
diff --git a/electron/subscription-store.cjs b/electron/subscription-store.cjs
new file mode 100644
index 0000000..2ae3f5f
--- /dev/null
+++ b/electron/subscription-store.cjs
@@ -0,0 +1,64 @@
+"use strict";
+const fs = require("node:fs/promises");
+const path = require("node:path");
+const { randomUUID } = require("node:crypto");
+
+const PROVIDERS = ["github-copilot", "xai"];
+const failure = (code) => Object.assign(new Error(code), { code });
+const nonempty = (value) => typeof value === "string" && value.length > 0 && !/[\x00-\x20\x7f]/.test(value);
+function validateCredential(providerId, credential) {
+ if (!credential || credential.type !== "oauth" || !nonempty(credential.access) || !nonempty(credential.refresh) || !Number.isFinite(credential.expires) || credential.expires < 0) throw failure("storage");
+ if (providerId === "github-copilot") {
+ if (credential.enterpriseUrl !== undefined && !safeDomain(credential.enterpriseUrl)) throw failure("storage");
+ if (credential.availableModelIds !== undefined && (!Array.isArray(credential.availableModelIds) || !credential.availableModelIds.every(nonempty))) throw failure("storage");
+ }
+ return credential;
+}
+function safeDomain(value) {
+ if (typeof value !== "string" || !value || /[\s/:@?#\\]/.test(value)) return false;
+ try { return new URL(`https://${value}`).hostname === value; } catch { return false; }
+}
+function safeHttps(value) {
+ try {
+ const url = new URL(value);
+ return url.protocol === "https:" && !url.username && !url.password;
+ } catch { return false; }
+}
+function validateState(providerId, state) {
+ if (!PROVIDERS.includes(providerId) || !state || state.version !== 1 || typeof state.allowBackgroundUse !== "boolean") throw failure("storage");
+ const hasId = Object.hasOwn(state, "connectionId");
+ if (hasId !== Object.hasOwn(state, "credential") || (hasId && !/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(state.connectionId))) throw failure("storage");
+ if (hasId) validateCredential(providerId, state.credential);
+ return state;
+}
+/** Main-process-only, one independent owner-only file per provider. No recovery writes on corruption. */
+function fileStore(file, providerId, io = fs) {
+ return {
+ async read() {
+ try {
+ const stat = await io.lstat(file);
+ if (!stat.isFile() || (process.platform !== "win32" && ((stat.mode & 0o077) !== 0 || (typeof process.getuid === "function" && stat.uid !== process.getuid())))) throw failure("storage");
+ return validateState(providerId, JSON.parse(await io.readFile(file, "utf8")));
+ }
+ catch (error) { if (error.code === "ENOENT") return { version: 1, allowBackgroundUse: true }; throw failure("storage"); }
+ },
+ async write(state) {
+ validateState(providerId, state);
+ const temporary = `${file}.${randomUUID()}.tmp`;
+ let handle;
+ try {
+ await io.mkdir(path.dirname(file), { recursive: true, mode: 0o700 });
+ handle = await io.open(temporary, "wx", 0o600);
+ await handle.writeFile(`${JSON.stringify(state)}\n`);
+ await handle.sync(); await handle.close(); handle = null;
+ await io.rename(temporary, file);
+ if (process.platform !== "win32") {
+ const directory = await io.open(path.dirname(file), "r");
+ try { await directory.sync(); } finally { await directory.close(); }
+ }
+ } catch { throw failure("storage"); }
+ finally { await handle?.close().catch(() => {}); await io.unlink(temporary).catch(() => {}); }
+ },
+ };
+}
+module.exports = { PROVIDERS, failure, validateCredential, validateState, safeDomain, safeHttps, fileStore };
diff --git a/electron/subscriptions.cjs b/electron/subscriptions.cjs
new file mode 100644
index 0000000..16e0af3
--- /dev/null
+++ b/electron/subscriptions.cjs
@@ -0,0 +1,251 @@
+"use strict";
+const { randomUUID } = require("node:crypto");
+const path = require("node:path");
+const { createRequire, registerHooks } = require("node:module");
+const { pathToFileURL } = require("node:url");
+const { PROVIDERS, fileStore, validateCredential, safeHttps, failure } = require("./subscription-store.cjs");
+
+const LABELS = { "github-copilot": "GitHub Copilot", xai: "Grok" };
+const APIS = new Set(["openai-responses", "openai-completions", "anthropic-messages"]);
+const clone = (value) => value === undefined ? undefined : JSON.parse(JSON.stringify(value));
+const MESSAGES = {
+ storage: "The saved connection could not be read or written safely. Reopen the app or reconnect after resolving the credential-file problem.",
+ login: "Sign-in did not complete. Try connecting again.",
+ refresh: "The connection could not be refreshed. Retry or reconnect in Providers.",
+ cancel: "Sign-in cancelled.",
+ unavailable: "This subscription connection is unavailable. Reconnect or explicitly choose another connection.",
+ background: "Background use is off for this connection. Automatic follow-up remains deferred in its queue.",
+ stale: "The connection or model catalog changed. Select an available model and retry.",
+ interaction: "This sign-in action is no longer available.",
+};
+function publicError(error) { return MESSAGES[error?.code] ?? MESSAGES.unavailable; }
+/** Resolve the SAME SDK installation as the standalone server, through public exports. */
+async function loadSubscriptionSDK(baseDirectory) {
+ const resolve = createRequire(path.join(baseDirectory, "package.json"));
+ const load = (id) => {
+ // Pi exports are import-only. Resolve the public export with ESM
+ // conditions, then remove this synchronous, exact-specifier hook
+ // BEFORE evaluation; no private package path or global hook remains.
+ const hook = registerHooks({ resolve(specifier, context, next) {
+ return next(specifier, specifier === id ? { ...context, conditions: [...context.conditions, "import"] } : context);
+ } });
+ let file;
+ try { file = resolve.resolve(id); } finally { hook.deregister(); }
+ return import(pathToFileURL(file).href);
+ };
+ const [sdk, copilot, xai] = await Promise.all([load("@earendil-works/pi-ai"), load("@earendil-works/pi-ai/providers/github-copilot"), load("@earendil-works/pi-ai/providers/xai")]);
+ return { ...sdk, ...copilot, ...xai };
+}
+function validEndpoint(providerId, auth, credential) {
+ if (!auth || (!(typeof auth.apiKey === "string" && auth.apiKey.length > 0) && !Object.entries(auth.headers ?? {}).some(([key, value]) => ["authorization", "x-api-key", "api-key"].includes(key.toLowerCase()) && typeof value === "string" && value.trim().length > 0))) return false;
+ const value = auth.baseUrl ?? (providerId === "xai" ? "https://api.x.ai/v1" : "https://api.individual.githubcopilot.com");
+ if (!safeHttps(value)) return false;
+ const url = new URL(value);
+ if (providerId === "xai") return url.origin === "https://api.x.ai" && url.pathname.replace(/\/$/, "") === "/v1" && !url.search && !url.hash;
+ const domain = credential?.enterpriseUrl;
+ return !url.search && !url.hash && url.pathname === "/" && (!url.port || url.port === "443") && (url.hostname.endsWith(".githubcopilot.com") || (domain && (url.hostname === domain || url.hostname.endsWith(`.${domain}`))));
+}
+/** Main owns durable credentials and refresh. Everything returned by status/snapshot is explicitly projected. */
+function createSubscriptions(options) {
+ let sdk;
+ let closed = false, paused = false;
+ const entries = new Map(PROVIDERS.map((providerId) => [providerId, {
+ providerId, state: { version: 1, allowBackgroundUse: true }, disk: options.stores?.[providerId] ?? fileStore(path.join(options.directory, `${providerId}-credentials.json`), providerId),
+ epoch: 0, authorityRevision: 0, revision: 0, tail: Promise.resolve(), fatal: false, blocked: false, ready: false, error: undefined, authFailed: false, attempt: null, catalog: [], endpoint: null,
+ }]));
+ const entryFor = (id) => { const entry = entries.get(id); if (!entry) throw failure("unavailable"); return entry; };
+ function serialized(entry, fn) { const result = entry.tail.then(fn); entry.tail = result.catch(() => {}); return result; }
+ function statusOf(entry) {
+ const available = entry.ready && !closed && !paused && !entry.fatal;
+ const attempt = entry.attempt;
+ return {
+ providerId: entry.providerId, available, connected: !!(available && entry.state.credential && !entry.blocked && !entry.authFailed), pending: !!attempt, committing: attempt?.committing === true,
+ allowBackgroundUse: entry.state.allowBackgroundUse, connectionId: entry.state.connectionId ?? null,
+ epoch: entry.epoch, authorityRevision: entry.authorityRevision, revision: entry.revision,
+ ...(entry.error ? { error: entry.error } : {}),
+ ...(attempt ? { interaction: clone(attempt.public) } : {}),
+ };
+ }
+ function snapshotNow() {
+ return { statuses: [...entries.values()].map(statusOf), models: Object.fromEntries([...entries.values()].map((entry) => [entry.providerId, statusOf(entry).connected ? clone(entry.catalog) : []])) };
+ }
+ function changed(entry) { entry.revision++; try { options.onChange?.(snapshotNow()); } catch { /* observer cannot break credential writes */ } }
+ function catalog(entry) {
+ let models = entry.provider?.getModels().filter((model) => APIS.has(model.api)) ?? [];
+ if (!entry.state.credential) models = [];
+ else if (entry.providerId === "github-copilot" && !Array.isArray(entry.state.credential.availableModelIds)) models = [];
+ else if (entry.provider?.filterModels) models = [...entry.provider.filterModels(models, entry.state.credential)];
+ if (JSON.stringify(models) !== JSON.stringify(entry.catalog)) { entry.catalog = models; entry.authorityRevision++; }
+ }
+ async function updateEndpoint(entry) {
+ let endpoint = null;
+ if (entry.state.credential) {
+ const auth = await entry.provider.auth.oauth.toAuth(entry.state.credential);
+ if (!validEndpoint(entry.providerId, auth, entry.state.credential)) throw failure("storage");
+ endpoint = auth.baseUrl ?? (entry.providerId === "xai" ? "https://api.x.ai/v1" : "https://api.individual.githubcopilot.com");
+ }
+ if (entry.endpoint !== endpoint) { entry.endpoint = endpoint; entry.authorityRevision++; }
+ }
+ function storageFailed(entry) { entry.fatal = true; entry.blocked = true; entry.epoch++; entry.error = MESSAGES.storage; changed(entry); }
+ async function save(entry, state) {
+ try { await entry.disk.write(state); } catch { storageFailed(entry); throw failure("storage"); }
+ entry.state = clone(state);
+ try { await updateEndpoint(entry); catalog(entry); } catch { storageFailed(entry); throw failure("storage"); }
+ changed(entry);
+ }
+ const ready = (async () => {
+ try { sdk = options.sdk ?? await options.loadSDK(); }
+ catch { for (const entry of entries.values()) { entry.error = MESSAGES.unavailable; entry.fatal = true; changed(entry); } return; }
+ await Promise.all([...entries.values()].map(async (entry) => {
+ try {
+ entry.state = clone(await entry.disk.read());
+ const native = entry.providerId === "github-copilot" ? sdk.githubCopilotProvider() : sdk.xaiProvider();
+ // No API-key auth method exists in this collection: ambient keys cannot become a fallback.
+ entry.provider = { ...native, auth: { oauth: native.auth.oauth } };
+ const store = {
+ read: async (id) => id === entry.providerId ? clone(entry.state.credential) : undefined,
+ list: async () => entry.state.credential ? [{ providerId: entry.providerId, type: "oauth" }] : [],
+ modify: (id, fn, operation) => serialized(entry, async () => {
+ if (id !== entry.providerId || entry.fatal || closed || paused || operation?.signal?.aborted) throw failure("unavailable");
+ const next = await fn(clone(entry.state.credential));
+ // A returned rotation must be saved even if the app paused while
+ // refresh was in flight. Admission is rechecked after resolution;
+ // queued logout deletes only after this serialized durable write.
+ if (next !== undefined && JSON.stringify(next) !== JSON.stringify(entry.state.credential)) {
+ try { validateCredential(id, next); } catch { storageFailed(entry); throw failure("storage"); }
+ await save(entry, { ...entry.state, credential: next });
+ }
+ return clone(entry.state.credential);
+ }),
+ delete: (id) => serialized(entry, async () => {
+ if (id !== entry.providerId || entry.fatal) throw failure("storage");
+ await save(entry, { version: 1, allowBackgroundUse: entry.state.allowBackgroundUse });
+ }),
+ };
+ entry.models = sdk.createModels({ credentials: store }); entry.models.setProvider(entry.provider);
+ await updateEndpoint(entry); entry.ready = true; catalog(entry); changed(entry);
+ } catch { entry.fatal = true; entry.error = MESSAGES.storage; changed(entry); }
+ }));
+ })();
+ function current(entry, scope) {
+ return statusOf(entry).connected && scope?.providerId === entry.providerId && scope.connectionId === entry.state.connectionId && scope.epoch === entry.epoch && scope.authorityRevision === entry.authorityRevision && entry.catalog.some((m) => m.id === scope.selectedModelId);
+ }
+ async function consent(providerId, enabled) {
+ await ready; const entry = entryFor(providerId);
+ if (typeof enabled !== "boolean" || entry.fatal || closed || paused) throw failure("unavailable");
+ await serialized(entry, async () => { if (entry.state.allowBackgroundUse !== enabled) await save(entry, { ...entry.state, allowBackgroundUse: enabled }); });
+ }
+ async function login(providerId, enabled) {
+ await ready; const entry = entryFor(providerId);
+ if ((enabled !== undefined && typeof enabled !== "boolean") || entry.fatal || closed || paused || !entry.ready) throw failure("unavailable");
+ if ([...entries.values()].some((e) => e.attempt)) throw failure("interaction");
+ const abort = new AbortController();
+ const attempt = { id: randomUUID(), abort, committing: false, actions: new Map(), prompt: null, public: {} };
+ attempt.public = { attemptId: attempt.id }; entry.attempt = attempt; entry.error = undefined; changed(entry);
+ const active = () => entry.attempt === attempt && !abort.signal.aborted && !closed && !paused;
+ const addBrowser = (value) => {
+ if (!safeHttps(value)) throw failure("interaction");
+ const id = randomUUID(); attempt.actions.set(id, value);
+ const url = new URL(value); attempt.public.browser = { id, url: `${url.origin}${url.pathname}` };
+ };
+ try {
+ const staging = sdk.createModels(); staging.setProvider(entry.provider);
+ const credential = await staging.login(providerId, "oauth", {
+ signal: abort.signal,
+ notify: (event) => {
+ if (!active()) return;
+ if (event.type === "auth_url") addBrowser(event.url);
+ if (event.type === "device_code") { addBrowser(event.verificationUri); attempt.public.userCode = event.userCode; }
+ // Native provider diagnostics are never forwarded. Public prompt wording comes from SDK-owned prompts.
+ attempt.public.progress = event.type === "device_code" || event.type === "auth_url" ? "Complete sign-in in your browser." : "Waiting for the provider…";
+ changed(entry);
+ },
+ prompt: (prompt) => new Promise((resolve, reject) => {
+ if (!active() || prompt.signal?.aborted) { reject(failure("cancel")); return; }
+ const id = randomUUID();
+ const stop = () => { cleanup(); reject(failure("cancel")); };
+ const cleanup = () => { abort.signal.removeEventListener("abort", stop); prompt.signal?.removeEventListener("abort", stop); if (attempt.prompt?.id === id) { attempt.prompt = null; delete attempt.public.prompt; } };
+ attempt.prompt = { id, resolve: (value) => { cleanup(); resolve(value); }, reject: stop };
+ attempt.public.prompt = { id, type: prompt.type, message: prompt.message, ...(prompt.placeholder ? { placeholder: prompt.placeholder } : {}), ...(prompt.type === "select" ? { options: clone(prompt.options) } : {}) };
+ abort.signal.addEventListener("abort", stop, { once: true }); prompt.signal?.addEventListener("abort", stop, { once: true }); changed(entry);
+ }),
+ }, { agentName: "LearnHarness" });
+ if (!active()) throw failure("cancel"); validateCredential(providerId, credential);
+ const auth = await entry.provider.auth.oauth.toAuth(credential);
+ if (!validEndpoint(providerId, auth, credential) || !active()) throw failure("login");
+ await serialized(entry, async () => {
+ if (!active() || entry.fatal) throw failure("cancel");
+ attempt.committing = true; entry.blocked = true; entry.epoch++; changed(entry);
+ await save(entry, { version: 1, allowBackgroundUse: enabled ?? entry.state.allowBackgroundUse, connectionId: randomUUID(), credential });
+ entry.blocked = false; entry.authFailed = false; entry.error = undefined;
+ });
+ } catch (error) {
+ if (active() && !entry.fatal) entry.error = error?.code === "cancel" ? undefined : MESSAGES.login;
+ if (!abort.signal.aborted) throw failure(entry.fatal ? "storage" : "login");
+ } finally { attempt.prompt?.reject(); if (entry.attempt === attempt) { entry.attempt = null; changed(entry); } }
+ }
+ function cancel(providerId) { const entry = entryFor(providerId); if (entry.attempt && !entry.attempt.committing) { entry.attempt.abort.abort(); entry.attempt = null; if (entry.authFailed && !entry.error) entry.error = MESSAGES.refresh; changed(entry); } }
+ async function resolveAuth(entry, modelId) {
+ if (entry.fatal || closed || paused || !entry.state.credential) throw failure("unavailable");
+ const connectionId = entry.state.connectionId, epoch = entry.epoch;
+ const unchanged = () => connectionId === entry.state.connectionId && epoch === entry.epoch && !entry.blocked && !closed && !paused;
+ try {
+ const model = entry.provider.getModels().find((m) => m.id === modelId);
+ if (!model) throw failure("stale");
+ const resolved = await entry.models.getAuth(model);
+ if (!unchanged()) throw failure("stale");
+ if (!validEndpoint(entry.providerId, resolved?.auth, entry.state.credential)) throw failure("unavailable");
+ return {
+ ...(typeof resolved.auth.apiKey === "string" ? { apiKey: resolved.auth.apiKey } : {}),
+ ...(resolved.auth.headers ? { headers: Object.fromEntries(Object.entries(resolved.auth.headers).filter(([, value]) => value === null || typeof value === "string")) } : {}),
+ ...(typeof resolved.auth.baseUrl === "string" ? { baseUrl: resolved.auth.baseUrl } : {}),
+ };
+ } catch (error) {
+ if (!entry.fatal && unchanged()) { entry.authFailed = true; entry.error = MESSAGES.refresh; entry.epoch++; changed(entry); }
+ throw failure(entry.fatal ? "storage" : error?.code === "stale" ? "stale" : "refresh");
+ }
+ }
+ return {
+ ready,
+ async snapshot() { await ready; return snapshotNow(); },
+ async status() { await ready; return snapshotNow().statuses; },
+ login, consent, cancel,
+ async respond(providerId, attemptId, promptId, value) {
+ const entry = entryFor(providerId), attempt = entry.attempt;
+ if (!attempt || attempt.id !== attemptId || attempt.prompt?.id !== promptId || typeof value !== "string" || value.length > 4096) throw failure("interaction");
+ const prompt = attempt.public.prompt;
+ if (prompt.type === "select" && !prompt.options.some((option) => option.id === value)) throw failure("interaction");
+ attempt.prompt.resolve(value); changed(entry);
+ },
+ async open(providerId, attemptId, actionId) {
+ const attempt = entryFor(providerId).attempt;
+ const target = attempt?.id === attemptId ? attempt.actions.get(actionId) : undefined;
+ if (!target || !safeHttps(target)) throw failure("interaction");
+ try { await options.openExternal(target); } catch { throw failure("interaction"); }
+ },
+ async lease(scope, background) {
+ await ready; const entry = entryFor(scope?.providerId);
+ if (!current(entry, scope)) throw failure("stale");
+ if (background && !entry.state.allowBackgroundUse) throw failure("background");
+ const auth = await resolveAuth(entry, scope.selectedModelId);
+ if (!current(entry, scope)) throw failure("stale");
+ if (background && !entry.state.allowBackgroundUse) throw failure("background");
+ return { scope: clone(scope), auth: clone(auth) };
+ },
+ async retry(providerId) {
+ await ready; const entry = entryFor(providerId);
+ const model = entry.catalog[0]; if (!model || entry.fatal || entry.blocked) throw failure("unavailable");
+ await resolveAuth(entry, model.id); entry.authFailed = false; entry.error = undefined; changed(entry);
+ },
+ async disconnect(providerId) {
+ await ready; const entry = entryFor(providerId); cancel(providerId);
+ if (entry.fatal) throw failure("storage");
+ entry.blocked = true; entry.epoch++; changed(entry);
+ await entry.models.logout(providerId); entry.blocked = false; entry.authFailed = false; entry.error = undefined; changed(entry);
+ },
+ pause() { paused = true; for (const entry of entries.values()) { cancel(entry.providerId); entry.epoch++; changed(entry); } },
+ resume() { if (closed) return; paused = false; for (const entry of entries.values()) changed(entry); },
+ close() { closed = true; for (const entry of entries.values()) { cancel(entry.providerId); entry.epoch++; changed(entry); } },
+ };
+}
+module.exports = { createSubscriptions, loadSubscriptionSDK, publicError, validEndpoint, LABELS };
diff --git a/next.config.ts b/next.config.ts
index 55e5758..db9352e 100644
--- a/next.config.ts
+++ b/next.config.ts
@@ -74,6 +74,11 @@ const nextConfig: NextConfig = {
// per-platform-binary story as the canvas above (one per build).
"node_modules/@resvg/resvg-js-*/**/*",
"node_modules/@earendil-works/pi-ai/dist/providers/data/**/*",
+ // Public native OAuth factories load these flows through variable imports.
+ // Electron main uses the same standalone SDK; the offline runtime gate
+ // evaluates their real toAuth paths without login or provider requests.
+ "node_modules/@earendil-works/pi-ai/dist/auth/oauth/{load,github-copilot,xai,device-code}.js",
+ "node_modules/@earendil-works/pi-ai/dist/utils/sleep.js",
"node_modules/@earendil-works/pi-agent-core/dist/**/*.json",
// Next 16.3.2 native Turbopack applies unanchored excludes while
// traversing modules; **/data/**/* still prunes gcp-metadata and
diff --git a/scripts/smoke-provider-runtime.mjs b/scripts/smoke-provider-runtime.mjs
index 58f3c3f..47c12c6 100644
--- a/scripts/smoke-provider-runtime.mjs
+++ b/scripts/smoke-provider-runtime.mjs
@@ -103,6 +103,29 @@ async function probe(standalone) {
noNetwork();
console.log(`[smoke-provider-runtime] ${name}: native lazy Google adapter reached onPayload; generation stopped`);
}
+ // Exercise the public lazy OAuth surface offline. toAuth evaluates the
+ // real flow module and its imports but neither logs in nor refreshes.
+ for (const [name, factory] of [["github-copilot", "githubCopilotProvider"], ["xai", "xaiProvider"]]) {
+ const provider = (await import(new URL(`providers/${name}.js`, entry).href))[factory]();
+ const credential = { type: "oauth", access: name === "github-copilot" ? "tid=offline;proxy-ep=proxy.individual.githubcopilot.com;" : "offline-xai-oauth-not-a-token", refresh: "offline-refresh-not-a-token", expires: Date.now() + 3600000, availableModelIds: provider.getModels().map((model) => model.id) };
+ const auth = await provider.auth.oauth.toAuth(credential);
+ assert.equal(auth.apiKey, credential.access);
+ if (name === "github-copilot") assert.equal(auth.baseUrl, "https://api.individual.githubcopilot.com");
+ const seenApis = new Set();
+ for (const model of provider.getModels()) {
+ if (seenApis.has(model.api)) continue;
+ seenApis.add(model.api);
+ const sentinel = `LH_OFFLINE_SUBSCRIPTION_${randomUUID()}`;
+ let reached = 0;
+ const result = await provider.streamSimple({ ...model, ...(auth.baseUrl ? { baseUrl: auth.baseUrl } : {}) }, { messages: [{ role: "user", content: "Offline initialization only", timestamp: 0 }] }, {
+ ...auth, maxRetries: 0, onPayload() { reached++; throw new Error(sentinel); },
+ }).result();
+ assert.equal(reached, 1, `${name}/${model.api}: ${result.errorMessage}`);
+ assert.equal(result.errorMessage, sentinel);
+ }
+ noNetwork();
+ console.log(`[smoke-provider-runtime] ${name}: public native OAuth and ${seenApis.size} transport APIs initialized offline`);
+ }
// Resolution/evaluation, not folder-exists checks. This also covers the
// previously missing transitive edge and permits legitimate nested deps.
for (const name of ["@google/genai", "google-auth-library", "gcp-metadata", "json-bigint", "bignumber.js"]) {
diff --git a/src/agent/chatgpt-request-cycle.ts b/src/agent/chatgpt-request-cycle.ts
index 5be9e1e..af7eb6d 100644
--- a/src/agent/chatgpt-request-cycle.ts
+++ b/src/agent/chatgpt-request-cycle.ts
@@ -1,122 +1,4 @@
-import type { AgentEvent, AgentMessage, PrepareRequestContext } from "@earendil-works/pi-agent-core";
-import type { AssistantMessage } from "@earendil-works/pi-ai";
-import { nativeInputCause, needsContinuation, type LhMessage } from "../core/native-messages.ts";
-
-export interface AdmissionScope { accountId: string; epoch: number; authorityRevision: number; selectedModelId: string }
-export type NoDispatchKind = "background" | "all" | "other";
-export interface ChatGPTDispatchLifecycle {
- captured(scope: AdmissionScope): void;
- beforeDispatch(scope: AdmissionScope): void;
- noDispatch(kind: NoDispatchKind): void;
-}
-interface CycleOptions {
- initialMessages: AgentMessage[];
- staged?: AgentMessage[];
- foregroundOwed: boolean;
- purpose: { background: boolean };
- scopeCurrent: (scope: AdmissionScope) => boolean;
- commitStaged?: () => void;
- restoreAutomatic: (automaticIds: ReadonlySet
, retainedIds: ReadonlySet) => void;
-}
-const idOf = (m: AgentMessage) => (m as { id?: string }).id;
-/** One logical turn, not an SDK loop. No credentials, persisted provenance or wire protocol. */
-export class ChatGPTRequestCycle implements ChatGPTDispatchLifecycle {
- private release?: (event: AgentEvent) => void;
- private held: AgentEvent[] = [];
- private snapshot: AgentMessage[] = [];
- private seen: Set;
- private automatic = new Set();
- private committed = false;
- private denied?: NoDispatchKind;
- private scope?: AdmissionScope;
- private originalScope?: AdmissionScope;
- private foregroundOwed: boolean;
- private inheritedBackground: boolean;
- private resumed = false;
- private userAborted = false;
- private stagedCommitted = false;
- private started = false;
- private anyDispatch = false;
- private firstResolved = false;
- private resolveAdmission!: (admitted: boolean) => void;
- readonly admission = new Promise((resolve) => { this.resolveAdmission = resolve; });
- constructor(private readonly options: CycleOptions) {
- this.seen = new Set(options.initialMessages.map(idOf).filter((id): id is string => !!id));
- for (const message of options.staged ?? []) { const id = idOf(message); if (id) this.seen.delete(id); }
- this.foregroundOwed = options.foregroundOwed;
- this.inheritedBackground = options.purpose.background;
- }
- get dispatched(): boolean { return this.anyDispatch; }
- get deniedBeforeDispatch(): boolean { return this.denied !== undefined; }
- abort(): void { this.userAborted = true; }
- bindRelease(release: (event: AgentEvent) => void): () => void { this.release = release; return () => { this.release = undefined; }; }
- prepare(request: PrepareRequestContext): void {
- const messages = request.context.messages as AgentMessage[];
- this.snapshot = [...messages]; this.automatic = new Set();
- let learner = false;
- for (const m of messages) {
- const id = idOf(m); if (!id || this.seen.has(id)) continue;
- const cause = nativeInputCause(m, messages.slice(0, messages.indexOf(m)));
- if (cause === "automatic") this.automatic.add(id);
- if (cause === "learner") learner = true;
- }
- if (this.automatic.size > 0) this.options.purpose.background = true;
- const retained = messages.filter((m) => !this.automatic.has(idOf(m) ?? ""));
- this.foregroundOwed ||= learner || (!this.inheritedBackground && needsContinuation(retained as LhMessage[]));
- }
- hold(event: AgentEvent): boolean {
- if (event.type === "agent_start") { if (this.started) return true; this.started = true; return false; }
- if (event.type === "agent_end") return true; // settle only after native activeRun is cleared
- if (event.type === "turn_start") {
- this.committed = false; this.denied = undefined; this.scope = undefined;
- // A retained learner batch from the denied cycle already owns its turn_start.
- if (!this.held.some((e) => e.type === "turn_start")) this.held.push(event);
- return true;
- }
- if (!this.committed) { this.held.push(event); return true; }
- return false;
- }
- captured(scope: AdmissionScope): void {
- if (this.originalScope && this.resumed && JSON.stringify(scope) !== JSON.stringify(this.originalScope)) throw new Error("ChatGPT recovery scope changed");
- this.scope = { ...scope };
- this.originalScope ??= { ...scope };
- }
- beforeDispatch(scope: AdmissionScope): void {
- if (this.userAborted || this.denied || this.committed || !this.scope || JSON.stringify(scope) !== JSON.stringify(this.scope) || !this.options.scopeCurrent(scope)) throw new Error("ChatGPT admission cancelled");
- if (!this.stagedCommitted && (this.options.staged?.length ?? 0) > 0) {
- this.options.commitStaged?.(); // synchronous durable prefix, before visibility and fetch
- this.stagedCommitted = true;
- const stagedEvents: AgentEvent[] = this.options.staged!.flatMap((message) => [{ type: "message_start" as const, message }, { type: "message_end" as const, message }]);
- const boundary = this.held.findIndex((event) => event.type === "turn_start");
- this.held.splice(boundary < 0 ? 0 : boundary + 1, 0, ...stagedEvents);
- }
- this.committed = true; this.anyDispatch = true; this.foregroundOwed = false;
- this.inheritedBackground = this.options.purpose.background;
- for (const m of this.snapshot) { const id = idOf(m); if (id) this.seen.add(id); }
- const events = this.held; this.held = [];
- for (const event of events) this.release?.(event);
- this.resolveFirst(true);
- }
- noDispatch(kind: NoDispatchKind): void { if (!this.committed) this.denied = this.userAborted ? "all" : kind; }
- acceptsUsage(_message: AssistantMessage): boolean { return this.committed; }
- private resolveFirst(value: boolean) { if (!this.firstResolved) { this.firstResolved = true; this.resolveAdmission(value); } }
- finishNativeCycle(): { action: "settle" | "resumeForeground"; messages: AgentMessage[] } {
- if (!this.denied) { this.resolveFirst(this.anyDispatch); return { action: "settle", messages: [] }; }
- const messages = this.snapshot.filter((m) => !this.automatic.has(idOf(m) ?? ""));
- const retainedIds = new Set(messages.map(idOf).filter((id): id is string => !!id));
- this.options.restoreAutomatic(this.automatic, retainedIds);
- this.resolveFirst(false); // queue deferral is installed before initial fire cargo returns
- this.held = this.held.filter((e) => (e.type === "turn_start") || ((e.type === "message_start" || e.type === "message_end") && e.message.role !== "assistant" && !this.automatic.has(idOf(e.message) ?? "")));
- const retry = this.denied === "background" && !this.userAborted && !this.resumed && this.foregroundOwed && !!this.scope && this.options.scopeCurrent(this.scope);
- if (retry) {
- this.resumed = true; this.options.purpose.background = false;
- this.seen = retainedIds;
- this.denied = undefined;
- // Staged automatic input must not be resurrected on foreground recovery.
- this.options.staged = undefined;
- return { action: "resumeForeground", messages };
- }
- this.held = [];
- return { action: "settle", messages };
- }
-}
+/** Compatibility surface: ChatGPT keeps its scope/behavior while all subscription
+ * connections share the same provider-neutral request-admission state machine. */
+export { SubscriptionRequestCycle as ChatGPTRequestCycle } from "./subscription-request-cycle.ts";
+export type { AdmissionScope, NoDispatchKind, SubscriptionDispatchLifecycle as ChatGPTDispatchLifecycle } from "./subscription-request-cycle.ts";
diff --git a/src/agent/models.ts b/src/agent/models.ts
index 5a9d8d0..75c0b7b 100644
--- a/src/agent/models.ts
+++ b/src/agent/models.ts
@@ -15,6 +15,8 @@ import packageMetadata from "../../package.json" with { type: "json" };
import { chatGPTModel } from "./chatgpt-bridge.ts";
import { chatGPTStream } from "./chatgpt-stream.ts";
import type { ChatGPTDispatchLifecycle } from "./chatgpt-request-cycle.ts";
+import { nativeSubscriptionId, nativeSubscriptionModel } from "./subscription-bridge.ts";
+import { nativeSubscriptionStream } from "./subscription-stream.ts";
/** App identity is immutable and shared by Models and every native Agent.
* Agent's default Pi attribution otherwise overrides the Models identity. */
@@ -98,7 +100,8 @@ function defaultThinkingLevel(model: Model): ThinkingLevel {
* hardcodes OpenRouter's. */
export function piEngineForSpec(spec: ResolvedModelSpec, background: boolean | (() => boolean) = true, lifecycle?: ChatGPTDispatchLifecycle): PiEngine {
const models: Models = piModels();
- const model = spec.provider === "chatgpt" ? chatGPTModel(spec.id) : models.getModel(spec.provider, spec.id);
+ const nativeId = nativeSubscriptionId(spec.provider);
+ const model = spec.provider === "chatgpt" ? chatGPTModel(spec.id) : nativeId ? nativeSubscriptionModel(spec.provider, spec.id) : models.getModel(spec.provider, spec.id);
if (!model) {
throw new Error(`${spec.provider} catalog has no model ${spec.id}`);
}
@@ -115,7 +118,8 @@ export function piEngineForSpec(spec: ResolvedModelSpec, background: boolean | (
// `reasoning` (undefined = off on pi-ai's scale); the clamp above keeps
// the pin honest even when config or settings drift off the map.
streamFn: (requestModel, context, options) => {
- const stream = spec.provider === "chatgpt" ? chatGPTStream(typeof background === "function" ? background : () => background, lifecycle) : models.streamSimple.bind(models);
+ const purpose = typeof background === "function" ? background : () => background;
+ const stream = spec.provider === "chatgpt" ? chatGPTStream(purpose, lifecycle) : nativeId ? nativeSubscriptionStream(models.getProvider(nativeId)!, purpose, lifecycle) : models.streamSimple.bind(models);
return stream(requestModel, context, { ...options, reasoning: level === "off" ? undefined : level });
},
};
diff --git a/src/agent/subagents/pi-run.ts b/src/agent/subagents/pi-run.ts
index 453a0f7..ece526d 100644
--- a/src/agent/subagents/pi-run.ts
+++ b/src/agent/subagents/pi-run.ts
@@ -2,6 +2,7 @@ import { Agent, type AgentEvent, type AgentMessage, type AgentTool } from "@eare
import type { AssistantMessage, ToolResultMessage, Usage } from "@earendil-works/pi-ai";
import { randomUUID } from "node:crypto";
import { APP_IDENTITY, piEngineFor, type PiEngine } from "../models.ts";
+import { usageWasAdmitted } from "../subscription-request-cycle.ts";
import { providerEventObserver } from "../provider-events.ts";
import type { ModelRole } from "../../../config.ts";
@@ -107,6 +108,7 @@ export async function runSubagent(tools: AgentTool[], opts: SubagentRu
aborted = true;
break;
}
+ if (!usageWasAdmitted(m.usage)) { if (m.stopReason === "error") errorMessage = m.errorMessage; break; }
step++;
opts.onUsage?.(m.usage);
opts.onAssistantMessage?.(m, step);
diff --git a/src/agent/subscription-bridge.ts b/src/agent/subscription-bridge.ts
new file mode 100644
index 0000000..b1a558e
--- /dev/null
+++ b/src/agent/subscription-bridge.ts
@@ -0,0 +1,132 @@
+import type { Api, Model, ModelAuth } from "@earendil-works/pi-ai";
+import { NATIVE_SUBSCRIPTIONS, type NativeSubscriptionProviderId, type NativeSubscriptionScope, type NativeSubscriptionSnapshot, type NativeSubscriptionStatus } from "../core/contracts/subscriptions.ts";
+import type { AdmissionScope, NoDispatchKind } from "./chatgpt-request-cycle.ts";
+
+interface Pending { op: string; resolve(value: unknown): void; reject(error: Error): void; timer: ReturnType }
+interface State {
+ statuses: Map;
+ models: Map[]>;
+ pending: Map;
+ controllers: Map;
+ listeners: Set<() => void>;
+ counter: number;
+}
+const KEY = Symbol.for("lh.subscriptions.bridge");
+export function nativeSubscriptionId(logicalId: string): NativeSubscriptionProviderId | undefined {
+ return NATIVE_SUBSCRIPTIONS.find((entry) => entry.logicalId === logicalId)?.providerId;
+}
+function desktopConnected(): boolean {
+ return process.env.LH_DESKTOP_SERVER === "1" && (process.env.LH_DESKTOP_API_SECRET?.length ?? 0) >= 64 && typeof process.send === "function" && process.connected === true;
+}
+function notify(state: State) { for (const listener of state.listeners) { try { listener(); } catch { /* observer isolation */ } } }
+function invalidate(state: State, providerId?: NativeSubscriptionProviderId) {
+ for (const [controller, owner] of state.controllers) if (!providerId || owner.providerId === providerId) controller.abort("all");
+ if (providerId) { state.models.delete(providerId); state.statuses.delete(providerId); }
+ else { state.models.clear(); state.statuses.clear(); }
+}
+function accept(state: State, snapshot: NativeSubscriptionSnapshot) {
+ if (!snapshot || !Array.isArray(snapshot.statuses) || !snapshot.models) return;
+ let changed = false;
+ for (const entry of snapshot.statuses) {
+ if (!NATIVE_SUBSCRIPTIONS.some((p) => p.providerId === entry.providerId) || ![entry.revision, entry.epoch, entry.authorityRevision].every((n) => Number.isSafeInteger(n) && n >= 0)) continue;
+ const previous = state.statuses.get(entry.providerId);
+ if (previous && entry.revision < previous.revision) continue;
+ const all = !!previous && (entry.connectionId !== previous.connectionId || entry.epoch !== previous.epoch || entry.authorityRevision !== previous.authorityRevision || !entry.available || !entry.connected);
+ for (const [controller, owner] of state.controllers) {
+ if (owner.providerId !== entry.providerId) continue;
+ if (all) controller.abort("all");
+ else if (owner.background && !entry.allowBackgroundUse) controller.abort("background");
+ }
+ state.statuses.set(entry.providerId, entry);
+ const descriptor = NATIVE_SUBSCRIPTIONS.find((p) => p.providerId === entry.providerId)!;
+ state.models.set(entry.providerId, entry.available && entry.connected ? (snapshot.models[entry.providerId] ?? []).filter((m) => m.provider === entry.providerId).map((m) => ({ ...m, provider: descriptor.logicalId })) : []);
+ changed ||= !previous || entry.revision !== previous.revision;
+ }
+ if (changed) notify(state);
+}
+function bridge(): State {
+ const global = globalThis as typeof globalThis & { [KEY]?: State };
+ if (!global[KEY]) {
+ const state: State = { statuses: new Map(), models: new Map(), pending: new Map(), controllers: new Map(), listeners: new Set(), counter: 0 };
+ global[KEY] = state;
+ process.on("message", (raw: unknown) => {
+ const message = raw as { channel?: string; op?: string; id?: string; ok?: boolean; value?: unknown; error?: string; snapshot?: NativeSubscriptionSnapshot };
+ if (!desktopConnected() || message?.channel !== "lh:subscriptions") return;
+ if (message.op === "changed" && message.snapshot) accept(state, message.snapshot);
+ if (!message.id) return;
+ const pending = state.pending.get(message.id);
+ if (!pending || pending.op !== message.op) return;
+ clearTimeout(pending.timer); state.pending.delete(message.id);
+ if (message.ok) pending.resolve(message.value);
+ else pending.reject(new Error(message.error ?? "Subscription connection failed."));
+ });
+ process.on("disconnect", () => {
+ invalidate(state);
+ for (const pending of state.pending.values()) { clearTimeout(pending.timer); pending.reject(new Error("The desktop connection closed.")); }
+ state.pending.clear(); notify(state);
+ });
+ }
+ return global[KEY];
+}
+export function subscriptionRequest(op: "snapshot" | "lease", fields: Record = {}): Promise {
+ const state = bridge();
+ if (!desktopConnected()) return Promise.reject(new Error("Subscription sign-in is available only in the desktop app with a private server connection."));
+ return new Promise((resolve, reject) => {
+ const id = `${process.pid}:${++state.counter}`;
+ const timer = setTimeout(() => { state.pending.delete(id); reject(new Error("The subscription desktop connection timed out.")); }, 45000);
+ state.pending.set(id, { op, timer, resolve: (value) => resolve(value as T), reject });
+ process.send!({ ...fields, channel: "lh:subscriptions", id, op }, (error) => {
+ if (error) { clearTimeout(timer); state.pending.delete(id); reject(new Error("The subscription desktop connection closed.")); }
+ });
+ });
+}
+export async function refreshSubscriptionModels(): Promise {
+ const state = bridge();
+ if (!desktopConnected()) { invalidate(state); return; }
+ try { accept(state, await subscriptionRequest("snapshot")); }
+ catch { invalidate(state); notify(state); }
+}
+export function nativeSubscriptionStatus(providerId: NativeSubscriptionProviderId): NativeSubscriptionStatus | undefined { return bridge().statuses.get(providerId); }
+export function nativeSubscriptionConnected(providerId: NativeSubscriptionProviderId): boolean {
+ const status = nativeSubscriptionStatus(providerId);
+ return desktopConnected() && status?.available === true && status.connected && !!status.connectionId;
+}
+export function nativeSubscriptionBackgroundAllowed(providerId: NativeSubscriptionProviderId): boolean { return nativeSubscriptionConnected(providerId) && nativeSubscriptionStatus(providerId)?.allowBackgroundUse === true; }
+export function nativeSubscriptionModels(): Model[] { return desktopConnected() ? [...bridge().models.values()].flat() : []; }
+export function nativeSubscriptionModel(logicalId: string, modelId: string): Model | undefined { return nativeSubscriptionModels().find((m) => m.provider === logicalId && m.id === modelId); }
+export function onNativeSubscriptionChange(listener: () => void): () => void { const state = bridge(); state.listeners.add(listener); return () => { state.listeners.delete(listener); }; }
+export function subscriptionScope(providerId: NativeSubscriptionProviderId, selectedModelId: string): NativeSubscriptionScope {
+ const state = nativeSubscriptionStatus(providerId);
+ const logicalId = NATIVE_SUBSCRIPTIONS.find((p) => p.providerId === providerId)!.logicalId;
+ if (!state?.connectionId || !nativeSubscriptionConnected(providerId) || !nativeSubscriptionModel(logicalId, selectedModelId)) throw new Error("Subscription model catalog unavailable");
+ return { providerId, connectionId: state.connectionId, epoch: state.epoch, authorityRevision: state.authorityRevision, selectedModelId };
+}
+export function subscriptionScopeCurrent(scope: NativeSubscriptionScope): boolean {
+ try { return JSON.stringify(subscriptionScope(scope.providerId, scope.selectedModelId)) === JSON.stringify(scope); } catch { return false; }
+}
+/** Compatibility shape for the shared admission cycle; accountId is a local connection ID here. */
+export function subscriptionAdmissionScope(scope: NativeSubscriptionScope): AdmissionScope {
+ return { accountId: scope.connectionId, epoch: scope.epoch, authorityRevision: scope.authorityRevision, selectedModelId: scope.selectedModelId };
+}
+export function nativeAdmissionCurrent(providerId: NativeSubscriptionProviderId, scope: AdmissionScope): boolean {
+ return subscriptionScopeCurrent({ providerId, connectionId: scope.accountId, epoch: scope.epoch, authorityRevision: scope.authorityRevision, selectedModelId: scope.selectedModelId });
+}
+export async function subscriptionLease(scope: NativeSubscriptionScope, background: boolean): Promise {
+ if (!subscriptionScopeCurrent(scope)) throw new Error("Subscription connection changed");
+ const result = await subscriptionRequest<{ scope: NativeSubscriptionScope; auth: ModelAuth }>("lease", { scope, background });
+ if (!result?.auth || !subscriptionScopeCurrent(scope) || JSON.stringify(result.scope) !== JSON.stringify(scope) || (!result.auth.apiKey && !Object.values(result.auth.headers ?? {}).some((value) => !!value))) throw new Error("Subscription lease is stale");
+ return result.auth;
+}
+export function subscriptionAbortController(providerId: NativeSubscriptionProviderId, background: boolean) {
+ const state = bridge(); const controller = new AbortController(); state.controllers.set(controller, { providerId, background });
+ return { controller, release: () => { state.controllers.delete(controller); } };
+}
+export async function subscriptionDenial(scope: NativeSubscriptionScope | undefined, controller: AbortController, callerAborted: boolean, background: boolean): Promise {
+ if (callerAborted || controller.signal.reason === "all") return "all";
+ if (!scope || !subscriptionScopeCurrent(scope)) return "other";
+ if (controller.signal.reason === "background") return "background";
+ if (!background) return "other";
+ await refreshSubscriptionModels();
+ if (!subscriptionScopeCurrent(scope)) return "all";
+ return nativeSubscriptionBackgroundAllowed(scope.providerId) ? "other" : "background";
+}
diff --git a/src/agent/subscription-connections.ts b/src/agent/subscription-connections.ts
new file mode 100644
index 0000000..2ede397
--- /dev/null
+++ b/src/agent/subscription-connections.ts
@@ -0,0 +1,23 @@
+import { chatGPTBackgroundAllowed, chatGPTConnected, chatGPTScopeCurrent, onChatGPTChange } from "./chatgpt-bridge.ts";
+import { nativeSubscriptionId, nativeSubscriptionConnected, nativeSubscriptionBackgroundAllowed, nativeAdmissionCurrent, onNativeSubscriptionChange } from "./subscription-bridge.ts";
+import type { AdmissionScope } from "./subscription-request-cycle.ts";
+
+export function isSubscriptionProvider(provider: string | undefined): boolean { return provider === "chatgpt" || !!(provider && nativeSubscriptionId(provider)); }
+export function subscriptionConnected(provider: string): boolean {
+ if (provider === "chatgpt") return chatGPTConnected();
+ const native = nativeSubscriptionId(provider); return !!native && nativeSubscriptionConnected(native);
+}
+/** Ordinary API connections have no subscription background-consent gate. */
+export function connectionBackgroundAllowed(provider: string | undefined): boolean {
+ if (provider === "chatgpt") return chatGPTBackgroundAllowed();
+ const native = provider ? nativeSubscriptionId(provider) : undefined;
+ return native ? nativeSubscriptionBackgroundAllowed(native) : true;
+}
+export function connectionScopeCurrent(provider: string, scope: AdmissionScope): boolean {
+ if (provider === "chatgpt") return chatGPTScopeCurrent(scope);
+ const native = nativeSubscriptionId(provider); return !!native && nativeAdmissionCurrent(native, scope);
+}
+export function onSubscriptionChange(listener: () => void): () => void {
+ const stopChatGPT = onChatGPTChange(listener), stopNative = onNativeSubscriptionChange(listener);
+ return () => { stopChatGPT(); stopNative(); };
+}
diff --git a/src/agent/subscription-request-cycle.ts b/src/agent/subscription-request-cycle.ts
new file mode 100644
index 0000000..f420b85
--- /dev/null
+++ b/src/agent/subscription-request-cycle.ts
@@ -0,0 +1,128 @@
+import type { AgentEvent, AgentMessage, PrepareRequestContext } from "@earendil-works/pi-agent-core";
+import type { AssistantMessage } from "@earendil-works/pi-ai";
+import { nativeInputCause, needsContinuation, type LhMessage } from "../core/native-messages.ts";
+
+export interface AdmissionScope { accountId: string; epoch: number; authorityRevision: number; selectedModelId: string }
+export type NoDispatchKind = "background" | "all" | "other";
+const USAGE_KEY = Symbol.for("lh.subscription.unadmittedUsage");
+const usageGlobal = globalThis as typeof globalThis & { [USAGE_KEY]?: WeakSet