fix(p2): session workers load OAuth flows and Bedrock from the worker base

The worker registers pi-ai's lazy OAuth and Bedrock loaders, so they are chunks of the
snapshot instead of pi-ai source files loaded at run time. The worker entry imports
chord/workers/child and host-packages.ts instead of plugins.ts: a fork loads no esbuild,
no server API and no provider SDK (690 to 301 modules on the phone).
This commit is contained in:
Mario Zechner 2026-11-05 01:27:21 +01:00
parent 7ac1a9e3a0
commit 90e2040468
8 changed files with 248 additions and 67 deletions

View file

@ -342,6 +342,11 @@ p2 runs on Chord v2 (`packages/chord`, `docs/design.md` there); the port's notes
where it lives with `hostAddress` (`serverHost`, `sessionHost`); a worker serves only its own session, so factories
no longer compare `host.name`. The session workers are Chord's `createWorkers` (`src/hosts/session-workers.ts`:
generations, logs and notices, relocation and pauses on `stop`/`retry`).
- **Worker base:** session workers fork from Chord's worker base of `src/hosts/worker.ts` (split at dynamic imports).
The entry imports only the worker's side (`@earendil-works/chord/workers/child`, `src/hosts/host-packages.ts`), so a
fork loads no esbuild and no server API; pi-ai's provider SDKs load with their first request. It registers pi-ai's
lazy OAuth flows and Bedrock (`registerLazyOAuthFlows`, `registerLazyBedrockProvider`): chunks of the snapshot, not
pi-ai's source files (`test/p2/worker-base.test.ts`, `worker-oauth.test.ts`).
- **Worker failures:** a failed start stays failed (status error, Retry) and its holders wait (`failed { waiting }`,
`whenLive` does not reject: tests use `failureOf`). A running worker that crashes is restarted for its holders, at
most five times a minute per session; the next crash waits for Retry. A worker stays alive while objects

View file

@ -0,0 +1,68 @@
// The host packages of p2's facets: the modules a facet shares with its host as instances, by specifier. Apart from
// plugins.ts (the server's plugin component, which builds with esbuild) so the worker entry imports only these: a
// worker loads neither esbuild nor the server's API.
import * as piAi from "@earendil-works/pi-ai";
import * as anthropicMessages from "@earendil-works/pi-ai/api/anthropic-messages.lazy";
import * as googleGenerativeAi from "@earendil-works/pi-ai/api/google-generative-ai.lazy";
import * as openaiCompletions from "@earendil-works/pi-ai/api/openai-completions.lazy";
import * as openaiResponses from "@earendil-works/pi-ai/api/openai-responses.lazy";
import * as piAiModels from "@earendil-works/pi-ai/models";
import * as allProviders from "@earendil-works/pi-ai/providers/all";
import * as fauxProvider from "@earendil-works/pi-ai/providers/faux";
import * as radiusProvider from "@earendil-works/pi-ai/providers/radius";
import * as radiusConfig from "@earendil-works/pi-ai/providers/radius-config";
import * as piCodemode from "@earendil-works/pi-codemode";
import * as piDurable from "@earendil-works/pi-durable";
import * as durableCodemode from "@earendil-works/pi-durable/codemode";
import * as durableEnv from "@earendil-works/pi-durable/env";
import * as durableEnvNode from "@earendil-works/pi-durable/env/node";
import * as durableToolSearch from "@earendil-works/pi-durable/tool-search";
import * as durableTools from "@earendil-works/pi-durable/tools";
import * as piEnv from "@earendil-works/pi-env";
import * as piMcp from "@earendil-works/pi-mcp";
import * as piMcpOauth from "@earendil-works/pi-mcp/oauth";
import * as piSchema from "@earendil-works/pi-schema";
import * as schemaFormats from "@earendil-works/pi-schema/formats";
import * as piTui from "@earendil-works/pi-tui";
import * as ws from "ws";
import * as yaml from "yaml";
import * as api from "../api/v1/index.ts";
import * as apiNode from "../api/v1/node.ts";
import * as apiWorker from "../api/v1/worker.ts";
/** The packages a facet shares with its host as instances (never bundled into a plugin's script), by specifier. */
export const SHARED_PACKAGES = {
"@earendil-works/pi-ai": piAi,
"@earendil-works/pi-ai/api/anthropic-messages.lazy": anthropicMessages,
"@earendil-works/pi-ai/api/google-generative-ai.lazy": googleGenerativeAi,
"@earendil-works/pi-ai/api/openai-completions.lazy": openaiCompletions,
"@earendil-works/pi-ai/api/openai-responses.lazy": openaiResponses,
"@earendil-works/pi-ai/models": piAiModels,
"@earendil-works/pi-ai/providers/all": allProviders,
"@earendil-works/pi-ai/providers/faux": fauxProvider,
"@earendil-works/pi-ai/providers/radius": radiusProvider,
"@earendil-works/pi-ai/providers/radius-config": radiusConfig,
"@earendil-works/pi-codemode": piCodemode,
"@earendil-works/pi-durable": piDurable,
"@earendil-works/pi-durable/codemode": durableCodemode,
"@earendil-works/pi-durable/env": durableEnv,
"@earendil-works/pi-durable/env/node": durableEnvNode,
"@earendil-works/pi-durable/tool-search": durableToolSearch,
"@earendil-works/pi-durable/tools": durableTools,
"@earendil-works/pi-env": piEnv,
"@earendil-works/pi-mcp": piMcp,
"@earendil-works/pi-mcp/oauth": piMcpOauth,
"@earendil-works/pi-schema": piSchema,
"@earendil-works/pi-schema/formats": schemaFormats,
ws,
yaml,
};
/** The host packages of worker facets (pi-tui too: file completion ranks as pi-tui's autocomplete does). */
export const WORKER_PACKAGES = {
"@p2/app/v1": api,
"@p2/app/v1/node": apiNode,
"@p2/app/v1/worker": apiWorker,
"@earendil-works/pi-tui": piTui,
...SHARED_PACKAGES,
};

View file

@ -1,41 +1,17 @@
// p2's plugins on Chord v2's plugin component: plugin directories under plugins/, each facet of kind `server` or
// `worker` built on its own against p2's plugin API (`@p2/app/v1`) and the packages a facet shares with its host,
// and evaluated with this process's instances of them. TUI and web facets stay in p2's own bundles (bundles.ts).
// and evaluated with this process's instances of them (host-packages.ts). TUI and web facets stay in p2's own bundles
// (bundles.ts).
import { cpSync, existsSync, mkdtempSync, readdirSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { BACKGROUND_CONTEXT, type Context } from "@earendil-works/chord/context";
import { createPlugins, type Plugins, type PluginsError } from "@earendil-works/chord/plugins";
import * as piAi from "@earendil-works/pi-ai";
import * as anthropicMessages from "@earendil-works/pi-ai/api/anthropic-messages.lazy";
import * as googleGenerativeAi from "@earendil-works/pi-ai/api/google-generative-ai.lazy";
import * as openaiCompletions from "@earendil-works/pi-ai/api/openai-completions.lazy";
import * as openaiResponses from "@earendil-works/pi-ai/api/openai-responses.lazy";
import * as piAiModels from "@earendil-works/pi-ai/models";
import * as allProviders from "@earendil-works/pi-ai/providers/all";
import * as fauxProvider from "@earendil-works/pi-ai/providers/faux";
import * as radiusProvider from "@earendil-works/pi-ai/providers/radius";
import * as radiusConfig from "@earendil-works/pi-ai/providers/radius-config";
import * as piCodemode from "@earendil-works/pi-codemode";
import * as piDurable from "@earendil-works/pi-durable";
import * as durableCodemode from "@earendil-works/pi-durable/codemode";
import * as durableEnv from "@earendil-works/pi-durable/env";
import * as durableEnvNode from "@earendil-works/pi-durable/env/node";
import * as durableToolSearch from "@earendil-works/pi-durable/tool-search";
import * as durableTools from "@earendil-works/pi-durable/tools";
import * as piEnv from "@earendil-works/pi-env";
import * as piMcp from "@earendil-works/pi-mcp";
import * as piMcpOauth from "@earendil-works/pi-mcp/oauth";
import * as piSchema from "@earendil-works/pi-schema";
import * as schemaFormats from "@earendil-works/pi-schema/formats";
import * as piTui from "@earendil-works/pi-tui";
import * as ws from "ws";
import * as yaml from "yaml";
import * as api from "../api/v1/index.ts";
import * as apiNode from "../api/v1/node.ts";
import * as apiServer from "../api/v1/server.ts";
import * as apiWorker from "../api/v1/worker.ts";
import { SHARED_PACKAGES, WORKER_PACKAGES } from "./host-packages.ts";
export type PluginName = string;
export type FacetKind = "server" | "worker" | "web" | "tui";
@ -51,48 +27,12 @@ const FACET_KINDS = ["server", "worker", "web", "tui"] as const;
/** p2's plugin API: one version so far. */
export const API = { name: "@p2/app", versions: ["v1"] } as const;
/** The packages a facet shares with its host as instances (never bundled into a plugin's script), by specifier. */
const SHARED = {
"@earendil-works/pi-ai": piAi,
"@earendil-works/pi-ai/api/anthropic-messages.lazy": anthropicMessages,
"@earendil-works/pi-ai/api/google-generative-ai.lazy": googleGenerativeAi,
"@earendil-works/pi-ai/api/openai-completions.lazy": openaiCompletions,
"@earendil-works/pi-ai/api/openai-responses.lazy": openaiResponses,
"@earendil-works/pi-ai/models": piAiModels,
"@earendil-works/pi-ai/providers/all": allProviders,
"@earendil-works/pi-ai/providers/faux": fauxProvider,
"@earendil-works/pi-ai/providers/radius": radiusProvider,
"@earendil-works/pi-ai/providers/radius-config": radiusConfig,
"@earendil-works/pi-codemode": piCodemode,
"@earendil-works/pi-durable": piDurable,
"@earendil-works/pi-durable/codemode": durableCodemode,
"@earendil-works/pi-durable/env": durableEnv,
"@earendil-works/pi-durable/env/node": durableEnvNode,
"@earendil-works/pi-durable/tool-search": durableToolSearch,
"@earendil-works/pi-durable/tools": durableTools,
"@earendil-works/pi-env": piEnv,
"@earendil-works/pi-mcp": piMcp,
"@earendil-works/pi-mcp/oauth": piMcpOauth,
"@earendil-works/pi-schema": piSchema,
"@earendil-works/pi-schema/formats": schemaFormats,
ws,
yaml,
};
/** The host packages of server facets: the API's kind-neutral, Node and server modules, and the shared packages. */
export const SERVER_PACKAGES = {
"@p2/app/v1": api,
"@p2/app/v1/node": apiNode,
"@p2/app/v1/server": apiServer,
...SHARED,
};
/** The host packages of worker facets (pi-tui too: file completion ranks as pi-tui's autocomplete does). */
export const WORKER_PACKAGES = {
"@p2/app/v1": api,
"@p2/app/v1/node": apiNode,
"@p2/app/v1/worker": apiWorker,
"@earendil-works/pi-tui": piTui,
...SHARED,
...SHARED_PACKAGES,
};
/** A directory listing, alphabetical: every subdirectory with at least one facet file is a plugin. */

View file

@ -4,7 +4,9 @@
import { isChordError } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, withAbortSignal } from "@earendil-works/chord/context";
import { workerMain } from "@earendil-works/chord/workers";
import { workerMain } from "@earendil-works/chord/workers/child";
import { registerLazyBedrockProvider } from "@earendil-works/pi-ai/lazy-bedrock-provider";
import { registerLazyOAuthFlows } from "@earendil-works/pi-ai/lazy-oauth";
import { createHost, hostErrorText } from "../runtime/chord.ts";
import { configureFetch } from "../runtime/fetch.ts";
import { captureProcessLogs, processLogs } from "../runtime/logs.ts";
@ -19,10 +21,14 @@ import {
type WorkerInit,
workerCore,
} from "../runtime/worker.ts";
import { WORKER_PACKAGES } from "./plugins.ts";
import { WORKER_PACKAGES } from "./host-packages.ts";
// Before any provider request: see configureFetch (Node 26 HTTP/2 sessions can wedge an origin).
configureFetch();
// OAuth flows and Bedrock through literal dynamic imports: chunks of the worker base, loaded on first use. pi-ai's
// default variable-specifier imports would load them from pi-ai's source files, outside the snapshot.
registerLazyOAuthFlows();
registerLazyBedrockProvider();
workerMain(
async ({ address, start, parent, plugins, onClose }, context) => {

View file

@ -1,7 +1,7 @@
// One process per name (Node): Chord's process lock, taken or refused. Chord's lock waits while another process holds
// it; p2's servers refuse instead (one server per data directory), after waiting `waitMs` for a holder that is exiting.
import { BACKGROUND_CONTEXT, withAbortSignal } from "@earendil-works/chord/context";
import { processLock } from "@earendil-works/chord/workers";
import { processLock } from "@earendil-works/chord/workers/child";
/** Takes the lock `name`, or rejects when another process still holds it after `waitMs`; resolves with its release. */
export async function takeProcessLock(name: string, waitMs = 0): Promise<() => Promise<void>> {

View file

@ -0,0 +1,76 @@
// The session worker's base (Chord's worker base of src/hosts/worker.ts): a fork loads no provider SDK and no esbuild
// (pi-ai's provider APIs stay lazy chunks), and the OAuth flows and Bedrock, which pi-ai imports through variable
// specifiers by default, are lazy chunks of the snapshot (worker-oauth.test.ts runs them).
import assert from "node:assert/strict";
import { readdirSync, readFileSync } from "node:fs";
import { dirname, join, parse } from "node:path";
import { after, before, test } from "node:test";
import { fileURLToPath } from "node:url";
import { buildWorkerBase, type WorkerBase } from "../../../../packages/chord/src/node/workers/base.ts";
const WORKER = fileURLToPath(new URL("../../src/hosts/worker.ts", import.meta.url));
const AI_SOURCE = fileURLToPath(new URL("../../../../packages/ai/src/", import.meta.url));
// As the server forks session workers (src/hosts/server.ts).
const EXEC_ARGV = ["--import", new URL("../../src/source.ts", import.meta.url).href];
/** Installed packages a fork must not load before a request needs them. */
const LAZY_PACKAGES = ["@anthropic-ai/sdk", "openai", "@google/genai", "@mistralai/", "@aws-sdk/", "esbuild"];
let base: WorkerBase;
before(async () => {
base = await buildWorkerBase(WORKER, EXEC_ARGV);
});
after(() => base.remove());
test("a fork's static graph has no provider SDK and no esbuild", () => {
const { externals } = staticGraph(base.file());
assert.ok(externals.length > 0);
const loaded = externals.filter((specifier) =>
LAZY_PACKAGES.some((name) => specifier.includes(`/node_modules/${name}`)),
);
assert.deepEqual(loaded, []);
});
test("the OAuth flows and Bedrock are lazy chunks of the snapshot", () => {
const { chunks } = staticGraph(base.file());
for (const source of ["auth/oauth/radius.ts", "auth/oauth/anthropic.ts", "api/bedrock-converse-stream.ts"]) {
const holders = chunksWith(join(AI_SOURCE, source));
assert.equal(holders.length, 1, source);
assert.ok(!chunks.includes(holders[0]!), `${source} is imported statically`);
}
});
/** The chunks (file names) the entry imports statically, transitively, and the external modules they import. */
function staticGraph(entry: string): { chunks: string[]; externals: string[] } {
const dir = dirname(entry);
const chunks: string[] = [];
const externals = new Set<string>();
const visit = (name: string): void => {
if (chunks.includes(name)) return;
chunks.push(name);
const code = readFileSync(join(dir, name), "utf8");
for (const [, specifier] of code.matchAll(/^(?:import|export)\s(?:[^;"]*?\sfrom\s*)?"([^"]+)"/gm)) {
if (specifier!.startsWith("./")) visit(specifier!.slice(2));
else externals.add(specifier!);
}
};
visit(parse(entry).base);
return { chunks, externals: [...externals] };
}
/** The bundle's files that hold the module `source`: their inline source maps name it. */
function chunksWith(source: string): string[] {
const dir = dirname(base.file());
return readdirSync(dir).filter((name) => {
const map = /sourceMappingURL=data:application\/json;base64,([A-Za-z0-9+/=]+)/.exec(
readFileSync(join(dir, name), "utf8"),
);
const { sources, sourceRoot } = JSON.parse(Buffer.from(map![1]!, "base64").toString()) as SourceMap;
return sources.some((path) => fileURLToPath(new URL(path, sourceRoot)) === source);
});
}
type SourceMap = { readonly sources: readonly string[]; readonly sourceRoot: string };

View file

@ -0,0 +1,85 @@
// OAuth flows and Bedrock must resolve inside the relocated worker base: the worker registers pi-ai's lazy loaders, so
// they load from the snapshot's chunks, never from pi-ai's source files (pi-ai's default variable-specifier imports
// would load them from there, outside the snapshot).
import assert from "node:assert/strict";
import { execFile } from "node:child_process";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { test } from "node:test";
import { fileURLToPath, pathToFileURL } from "node:url";
import { promisify } from "node:util";
import { buildWorkerBase } from "../../../../packages/chord/src/node/workers/base.ts";
const execFileAsync = promisify(execFile);
test("the bundled p2 worker resolves stored Radius OAuth credentials and loads Bedrock from its own chunks", async () => {
const directory = await mkdtemp(join(tmpdir(), "p2-worker-oauth-"));
try {
const worker = fileURLToPath(new URL("../../src/hosts/worker.ts", import.meta.url));
const ai = fileURLToPath(new URL("../../../../packages/ai/src/", import.meta.url));
const entry = join(directory, "entry.ts");
await writeFile(
entry,
`
import ${JSON.stringify(worker)};
import { bedrockConverseStreamApi } from ${JSON.stringify(join(ai, "api/bedrock-converse-stream.lazy.ts"))};
import { InMemoryCredentialStore } from ${JSON.stringify(join(ai, "auth/credential-store.ts"))};
import { createModels } from ${JSON.stringify(join(ai, "models.ts"))};
import { radiusProvider } from ${JSON.stringify(join(ai, "providers/radius.ts"))};
globalThis.fetch = async () => { throw new Error("Unexpected network request"); };
const credentials = new InMemoryCredentialStore();
await credentials.modify("radius", async () => ({
type: "oauth", access: "fixture-token", refresh: "unused", expires: Date.now() + 3600000,
}));
const models = createModels({ credentials });
models.setProvider(radiusProvider({ gateway: "https://radius.example" }));
const auth = await models.getAuth("radius");
// An aborted request: Bedrock's implementation and the AWS SDK load, nothing is sent.
const model = { id: "fixture", name: "fixture", api: "bedrock-converse-stream", provider: "amazon-bedrock",
baseUrl: "https://bedrock.example", reasoning: false, input: ["text"], contextWindow: 1000, maxTokens: 100,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 } };
const bedrock = await bedrockConverseStreamApi().stream(model, { messages: [] }, { signal: AbortSignal.abort() }).result();
console.log(JSON.stringify({ auth, bedrock: bedrock.stopReason }));
process.exit(0);
`,
);
// Records every module the fork loads.
const loadedFile = join(directory, "loaded.json");
const hook = join(directory, "hook.mjs");
await writeFile(
hook,
`
import { writeFileSync } from "node:fs";
import { registerHooks } from "node:module";
const loaded = [];
registerHooks({ load(url, context, next) { loaded.push(url); return next(url, context); } });
process.on("exit", () => writeFileSync(${JSON.stringify(loadedFile)}, JSON.stringify(loaded)));
`,
);
const execArgv = ["--import", new URL("../../src/source.ts", import.meta.url).href];
const base = await buildWorkerBase(entry, execArgv);
try {
const { stdout } = await execFileAsync(
process.execPath,
[...execArgv, "--import", pathToFileURL(hook).href, base.file()],
{ cwd: directory, timeout: 30_000 },
);
assert.deepEqual(JSON.parse(stdout), {
auth: { auth: { apiKey: "fixture-token" }, source: "OAuth" },
bedrock: "aborted",
});
const loaded = JSON.parse(await readFile(loadedFile, "utf8")) as string[];
// The flow and Bedrock came from the bundle: nothing loaded from pi-ai's sources, the AWS SDK did load.
assert.deepEqual(
loaded.filter((url) => url.startsWith(pathToFileURL(ai).href)),
[],
);
assert.ok(loaded.some((url) => url.includes("/node_modules/@aws-sdk/client-bedrock-runtime/")));
} finally {
base.remove();
}
} finally {
await rm(directory, { recursive: true, force: true });
}
});

View file

@ -21,6 +21,7 @@
"@earendil-works/chord/telemetry-sqlite": ["../../packages/chord/src/telemetry-sqlite/index.ts"],
"@earendil-works/chord/http": ["../../packages/chord/src/node/http/index.ts"],
"@earendil-works/chord/workers": ["../../packages/chord/src/node/workers/index.ts"],
"@earendil-works/chord/workers/child": ["../../packages/chord/src/node/workers/child.ts"],
"@earendil-works/chord/plugins": ["../../packages/chord/src/node/plugins/index.ts"],
"@earendil-works/chord/plugins/runtime": ["../../packages/chord/src/node/plugins/runtime.ts"],
"@earendil-works/chord/app": ["../../packages/chord/src/node/app/index.ts"],