Merge branch 'v3' into v3-t59

This commit is contained in:
2026-11-07 21:41:40 +00:00
22 changed files with 59453 additions and 966 deletions
+3
View File
@@ -8,6 +8,7 @@ They use world files and world git. They do not use native filesystem or process
`branch` requires an absolute `repo` path and a prompt.
It forks a listed tree root and writes `board.started.by` in the same transaction.
The coordinator definition supplies the branch configuration.
The branch start event reports the coordinator's model, not the inherited root model.
Worktree isolation requires the later workspaces port. The default is a shared checkout.
The tool admits the prompt as a reply-enabled message.
@@ -22,6 +23,8 @@ A branch reports to its starter when its work ends.
While a child, landing, or reply remains pending, its final answer becomes a passive note.
One-way updates to the starter also become passive notes.
The starter can answer a report without an endless exchange of routed answers.
The starter's final answer reaches the branch without a faulted routing task.
A static test checks free runtime identifiers in every extension module, including function bodies.
The grant chain contains fork ancestry before starter ancestry.
Each conversation occurs once. A root fork that @ultimate starts includes `[branch, root, ultimate]`.
+1 -1
View File
@@ -22,7 +22,7 @@ export const branchTool = (host: any) => defineTool({
const coordinator = definitions.find((definition: any) => definition.name === 'coordinator')
if (!coordinator) throw new Error('Missing coordinator definition')
const rootAgent = await (await host.harness().conversation(root, ctx))!.agent(ctx)
const branch = await lineage.branch({from: root, repo: input.repo, prompt: input.prompt, title: input.title, deliver: false, by: api.conversationId, receipt: {taskId: api.taskId, callId: api.callId}}, ctx)
const branch = await lineage.branch({from: root, repo: input.repo, prompt: input.prompt, title: input.title, model: coordinator.runs.model, deliver: false, by: api.conversationId, receipt: {taskId: api.taskId, callId: api.callId}}, ctx)
const existing = await workspaces?.get(branch.id, ctx)
const cwd = existing ? (await (await host.harness().conversation(branch.id, ctx)).agent(ctx)).cwd : input.repo
const ancestors = workspaces ? (await workspaces.chain(branch.id, ctx)).slice(1) : []
+1
View File
@@ -14,6 +14,7 @@ import type {
import type { Host as ExtensionHost } from '../../host/js/extensions.ts'
import { Subagents } from '../subagents/api.ts'
import { ConversationsApi } from '../conversations/api.ts'
import { Messages } from './api.ts'
import { terminal } from './log.ts'
+25 -1
View File
@@ -19,7 +19,31 @@ Without that provider, the definition or default family selects the model.
No login fallback runs.
Existing registered models retain their transport.
New logical model references reuse the host's OpenAI-compatible proxy provider and its auth resolver.
New model references reuse a template from their own registered provider and retain its proxy auth resolver.
They never inherit another provider's transport. Unknown providers fail instead of silently using OpenAI.
Models absent from the catalog receive zero prices, not the template's prices.
The extension never reads a credential or registers a direct provider SDK.
## Host proxy transports and prices
`host/js/proxy-models.ts` reads the authenticated `/v1/models` catalog once at host startup.
OpenAI models use pi-ai's Responses transport. Anthropic models use its Messages transport.
The catalog endpoint determines each model's base URL.
For Anthropic, effort capabilities select adaptive thinking and its advertised effort levels.
Both providers use the existing `llm-proxy` key and usage tag. No provider login or direct endpoint fallback runs.
If the catalog fails or exceeds five seconds, a small table supplies these defaults:
| Provider/model | API | Proxy path |
| --- | --- | --- |
| `openai/gpt-6.1-sol` | Responses | `/v1/responses` |
| `anthropic/claude-opus-5-5` | Messages, adaptive thinking | `/apis/anthropic/v1/messages` |
Catalog prices use USD per million tokens, with conversion from `unit_tokens` when necessary.
The host uses the first advertised route's quote, including cache reads and cache writes.
These costs are estimates, not billing totals. Another proxy route can charge different prices.
Unavailable prices remain zero. Catalog changes apply after a host restart.
World family edits still apply at the next run boundary without a restart.
`host/tests/proxy_models.rs` tests the real transports against a local fake proxy.
`host/tests/core_ext.rs` tests these services through the snapshot loader.
+4 -3
View File
@@ -28,14 +28,15 @@ export default function models(host: any) {
const found = discoverAgents(files, '/src/sausage', root, families, refs)
return {families, ...found}
}
const transport = host.models().getProvider('openai')
const template = host.models().getModels('openai')[0]
const register = (ref: any) => {
const store = host.models()
if (store.getModel(ref.provider, ref.modelId)) return
const transport = store.getProvider(ref.provider)
const template = store.getModels(ref.provider)[0]
if (!transport || !template) throw new Error(`No proxy transport for ${ref.provider}/${ref.modelId}`)
const prior = store.getModels(ref.provider)
store.setProvider({...transport, id:ref.provider, getModels:()=>[...prior, {...template, provider:ref.provider, id:ref.modelId, name:ref.modelId}]})
store.setProvider({...transport, getModels:()=>[...prior, {...template, id:ref.modelId, name:ref.modelId,
cost:{input:0,output:0,cacheRead:0,cacheWrite:0}}]})
}
const apply = async (tx: any, id: number) => {
const state = await tx.doc(AgentDoc, id)
+17305 -866
View File
File diff suppressed because one or more lines are too long
+8 -2
View File
@@ -31693,6 +31693,8 @@ var faux = fauxProvider({ tokensPerSecond: 1e5 });
models.setProvider(faux.provider);
var proxyTemplate = { ...models.getModels("faux")[0], provider: "openai", baseUrl: "https://proxy.invalid/v1" };
models.setProvider({ ...faux.provider, id: "openai", getModels: () => [proxyTemplate] });
models.setProvider({ ...faux.provider, id: "anthropic", getModels: () => [{ ...proxyTemplate, provider: "anthropic", baseUrl: "https://proxy.invalid/apis/anthropic" }] });
var proxyAuth = models.getProvider("openai").auth;
var storage = await openStorage(globalThis.storagePath);
var registry = createRegistry();
var harness = await Harness.open(storage, { models, registry, settings: { retry: { enabled: false }, get extensions() {
@@ -31743,10 +31745,14 @@ try {
faux.setResponses([fauxAssistantMessage("second")]);
await (await worker.submit({ type: "input", content: "second" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
assert((await harness.snapshot(AgentDoc, worker.id, BACKGROUND_CONTEXT)).model.modelId === "second", "edited family changes next model without restart");
await fs.write("/src/sausage/agents/models.json", JSON.stringify({ default: { model: "faux/faux-1" }, fast: { model: "logical/new" } }));
await fs.write("/src/sausage/agents/models.json", JSON.stringify({ default: { model: "faux/faux-1" }, fast: { model: "openai/new" } }));
faux.setResponses([fauxAssistantMessage("proxy")]);
await (await worker.submit({ type: "input", content: "proxy" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
assert(models.getModel("logical", "new").baseUrl === proxyTemplate.baseUrl && models.getProvider("logical").auth === models.getProvider("openai").auth, "new model retains proxy transport and auth");
assert(models.getModel("openai", "new").baseUrl === proxyTemplate.baseUrl && models.getModel("openai", "new").cost.input === 0 && models.getProvider("openai").auth === proxyAuth, "new model retains proxy transport and auth");
await fs.write("/src/sausage/agents/models.json", JSON.stringify({ default: { model: "faux/faux-1" }, fast: { model: "anthropic/new" } }));
faux.setResponses([fauxAssistantMessage("anthropic")]);
await (await worker.submit({ type: "input", content: "anthropic" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
assert(models.getModel("anthropic", "new").baseUrl === "https://proxy.invalid/apis/anthropic", "new Anthropic model retains its own provider transport");
globalThis.modelPin = { provider: "faux", modelId: "faux-1" };
faux.setResponses([fauxAssistantMessage("pin")]);
await (await worker.submit({ type: "input", content: "pin" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
+8 -2
View File
@@ -9,6 +9,8 @@ const assert = (ok:any,message:string)=>{if(!ok)throw new Error(message);Deno.co
const models=createModels(), faux=fauxProvider({tokensPerSecond:100000});models.setProvider(faux.provider)
const proxyTemplate={...models.getModels('faux')[0],provider:'openai',baseUrl:'https://proxy.invalid/v1'}
models.setProvider({...faux.provider,id:'openai',getModels:()=>[proxyTemplate]})
models.setProvider({...faux.provider,id:'anthropic',getModels:()=>[{...proxyTemplate,provider:'anthropic',baseUrl:'https://proxy.invalid/apis/anthropic'}]})
const proxyAuth=models.getProvider('openai').auth
const storage=await openStorage(globalThis.storagePath), registry=createRegistry()
const harness=await Harness.open(storage,{models,registry,settings:{retry:{enabled:false},get extensions(){return registry.snapshot().installed()}}},context)
await harness.root(context,{agent:{model:{provider:'faux',modelId:'faux-1'}}})
@@ -56,10 +58,14 @@ try {
faux.setResponses([fauxAssistantMessage('second')])
await (await worker.submit({type:'input',content:'second'},context)).wait(context)
assert((await harness.snapshot(AgentDoc,worker.id,context)).model.modelId==='second','edited family changes next model without restart')
await fs.write('/src/sausage/agents/models.json',JSON.stringify({default:{model:'faux/faux-1'},fast:{model:'logical/new'}}))
await fs.write('/src/sausage/agents/models.json',JSON.stringify({default:{model:'faux/faux-1'},fast:{model:'openai/new'}}))
faux.setResponses([fauxAssistantMessage('proxy')])
await (await worker.submit({type:'input',content:'proxy'},context)).wait(context)
assert(models.getModel('logical','new').baseUrl===proxyTemplate.baseUrl && models.getProvider('logical').auth===models.getProvider('openai').auth,'new model retains proxy transport and auth')
assert(models.getModel('openai','new').baseUrl===proxyTemplate.baseUrl && models.getModel('openai','new').cost.input===0 && models.getProvider('openai').auth===proxyAuth,'new model retains proxy transport and auth')
await fs.write('/src/sausage/agents/models.json',JSON.stringify({default:{model:'faux/faux-1'},fast:{model:'anthropic/new'}}))
faux.setResponses([fauxAssistantMessage('anthropic')])
await (await worker.submit({type:'input',content:'anthropic'},context)).wait(context)
assert(models.getModel('anthropic','new').baseUrl==='https://proxy.invalid/apis/anthropic','new Anthropic model retains its own provider transport')
globalThis.modelPin={provider:'faux',modelId:'faux-1'}
faux.setResponses([fauxAssistantMessage('pin')])
await (await worker.submit({type:'input',content:'pin'},context)).wait(context)
+3 -17
View File
@@ -1,7 +1,6 @@
// Durable root conversation and console. SQLite stays outside the world.
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
import { createModels } from '@earendil-works/pi-ai/models';
import { openaiProvider } from '@earendil-works/pi-ai/providers/openai';
import { proxyModels } from './proxy-models';
import { Harness, AgentDoc, createRegistry } from '@earendil-works/pi-durable';
import { CodingTools } from './environment';
import { openStorage } from './sqlite';
@@ -12,21 +11,8 @@ import { harnessApi } from './harness-api';
import { loadExtensions } from './extensions';
const proxy = Deno.core.ops.op_model_proxy();
if (!proxy?.base_url || !proxy?.api_key) throw new Error('The llm-proxy credential is required');
const models = createModels();
const provider = openaiProvider();
const model = {
id: 'gpt-6.1-sol', name: 'gpt-6.1-sol', provider: 'openai', api: 'openai-responses',
baseUrl: `${proxy.base_url}/v1`, reasoning: true, input: ['text'],
thinkingLevelMap: { off: null, minimal: null },
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 200000, maxTokens: 4096,
};
models.setProvider({ ...provider, baseUrl: model.baseUrl, getModels: () => [model],
auth: { apiKey: { name: 'Sausage proxy', resolve: async () => ({
auth: { apiKey: proxy.api_key, baseUrl: model.baseUrl,
headers: { 'Shopify-Usage-Tag': JSON.stringify(['pi', 'sausage']) } },
source: 'llm-proxy',
}) } },
});
const models = await proxyModels(proxy);
const model = models.getModel('openai', 'gpt-6.1-sol');
const registry = createRegistry();
registry.install(CodingTools);
registry.install(CredentialRequests);
+11
View File
@@ -31696,6 +31696,8 @@ var until2 = async (test) => {
var models = createModels();
var faux = fauxProvider({ tokensPerSecond: 1e5 });
models.setProvider(faux.provider);
var template = models.getModels("faux")[0];
models.setProvider({ ...faux.provider, getModels: () => [template, { ...template, id: "coordinator" }] });
var registry = createRegistry();
var storage = await openStorage(globalThis.storagePath);
var harness = await Harness.open(storage, { models, registry, settings: { get extensions() {
@@ -31814,12 +31816,21 @@ try {
assert((await entries2(branch)).some((e) => e.kind === "board.message" && e.data.text === "first task"), "The first message did not reach the branch");
assert((await harness.commit((tx) => tx.conversation(branch.id), BACKGROUND_CONTEXT)).parent.conversationId === root.id, "Branch must fork its root");
assert((await harness.snapshot(Started2, branch.id, BACKGROUND_CONTEXT)).by === primary.id, "Missing starter");
const starts = (await globalThis.testHost.eventLog.recent()).filter((event) => event.name === "subagents.started" && event.data.id === branch.id);
assert(starts.length === 1 && starts[0].data.model === "faux/coordinator", `Effective branch model: ${JSON.stringify(starts)}`);
assert((await branch.agent(BACKGROUND_CONTEXT)).model.modelId === "coordinator", "The branch must use its advertised model");
assert(JSON.stringify(await conversationChains(harness, BACKGROUND_CONTEXT)(branch.id)) === JSON.stringify([String(branch.id), String(root.id), String(primary.id)]), "Fork and starter grant ancestry");
const before = (await entries2(primary)).length;
await (await branch.submit({ type: "input", content: "later work" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
await until2(async () => (await entries2(primary)).some((e) => e.kind === "board.message" && e.data.from === branch.id && e.data.text === "Answer: later work" && e.data.delivery !== "note"));
await harness.waitForIdle(BACKGROUND_CONTEXT);
assert((await entries2(primary)).length > before, "An idle branch must report");
assert((await entries2(branch)).some((entry2) => entry2.kind === "board.message" && entry2.data.from === primary.id && entry2.data.routed === true), "The final answer must route back to the branch");
await until2(async () => {
const routeTasks = (await harness.commit((tx) => tx.scanTasks({ kind: "board.route" }, 1e3), BACKGROUND_CONTEXT)).items;
assert(routeTasks.every((task) => task.state.status !== "terminal" || task.state.outcome.status === "completed"), `Final-answer routing fault: ${JSON.stringify(routeTasks)}`);
return routeTasks.length > 0 && routeTasks.every((task) => task.state.status === "terminal");
});
await tool(branch, "subagent", { action: "start", agent: "scout", task: "HOLD", background: true });
await until2(() => holding.size > 0);
await until2(async () => (await entries2(primary)).some((e) => e.kind === "board.message" && e.data.from === branch.id && e.data.delivery === "note"));
+11
View File
@@ -10,6 +10,8 @@ import {openStorage} from './sqlite';
const assert = (value: any, message: string) => {if (!value) throw new Error(message)};
const until = async (test: () => any) => {const deadline = Date.now() + 15000; while (!await test()) {if (Date.now() > deadline) throw new Error('Messages deadline'); await new Promise(r => setTimeout(r, 0))}};
const models = createModels(), faux = fauxProvider({tokensPerSecond: 100000}); models.setProvider(faux.provider);
const template = models.getModels('faux')[0];
models.setProvider({...faux.provider, getModels: () => [template, {...template, id: 'coordinator'}]});
const registry = createRegistry(), storage = await openStorage(globalThis.storagePath);
const harness = await Harness.open(storage, {models, registry, settings: {get extensions() {return registry.snapshot().installed()}}}, context);
const root = await harness.root(context, {agent: {model: {provider: 'faux', modelId: 'faux-1'}, cwd: '/src/sausage'}});
@@ -99,12 +101,21 @@ try {
assert((await entries(branch)).some((e: any) => e.kind === 'board.message' && e.data.text === 'first task'), 'The first message did not reach the branch');
assert((await harness.commit((tx: any) => tx.conversation(branch.id), context)).parent.conversationId === root.id, 'Branch must fork its root');
assert((await harness.snapshot(Started, branch.id, context)).by === primary.id, 'Missing starter');
const starts = (await globalThis.testHost.eventLog.recent()).filter((event: any) => event.name === 'subagents.started' && event.data.id === branch.id);
assert(starts.length === 1 && starts[0].data.model === 'faux/coordinator', `Effective branch model: ${JSON.stringify(starts)}`);
assert((await branch.agent(context)).model.modelId === 'coordinator', 'The branch must use its advertised model');
assert(JSON.stringify(await conversationChains(harness, context)(branch.id)) === JSON.stringify([String(branch.id), String(root.id), String(primary.id)]), 'Fork and starter grant ancestry');
const before = (await entries(primary)).length;
await (await branch.submit({type: 'input', content: 'later work'}, context)).wait(context);
await until(async () => (await entries(primary)).some((e: any) => e.kind === 'board.message' && e.data.from === branch.id && e.data.text === 'Answer: later work' && e.data.delivery !== 'note'));
await harness.waitForIdle(context);
assert((await entries(primary)).length > before, 'An idle branch must report');
assert((await entries(branch)).some((entry: any) => entry.kind === 'board.message' && entry.data.from === primary.id && entry.data.routed === true), 'The final answer must route back to the branch');
await until(async () => {
const routeTasks = (await harness.commit((tx: any) => tx.scanTasks({kind: 'board.route'}, 1000), context)).items;
assert(routeTasks.every((task: any) => task.state.status !== 'terminal' || task.state.outcome.status === 'completed'), `Final-answer routing fault: ${JSON.stringify(routeTasks)}`);
return routeTasks.length > 0 && routeTasks.every((task: any) => task.state.status === 'terminal');
});
await tool(branch, 'subagent', {action: 'start', agent: 'scout', task: 'HOLD', background: true});
await until(() => holding.size > 0);
await until(async () => (await entries(primary)).some((e: any) => e.kind === 'board.message' && e.data.from === branch.id && e.data.delivery === 'note'));
+1 -1
View File
@@ -1,7 +1,7 @@
{
"private": true,
"type": "module",
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js && esbuild core-ext-test.ts --bundle --format=esm --platform=browser --outfile=core-ext-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js && esbuild p2-test.ts --bundle --format=esm --platform=browser --outfile=p2-test-bundle.js && esbuild chain-test.ts --bundle --format=esm --platform=browser --outfile=chain-test-bundle.js && esbuild undo-harness-test.ts --bundle --format=esm --platform=browser --outfile=undo-harness-test-bundle.js && esbuild messages-test.ts --bundle --format=esm --platform=browser --outfile=messages-test-bundle.js && esbuild p3-test.ts --bundle --format=esm --platform=browser --outfile=p3-test-bundle.js && esbuild automations-test.ts --bundle --format=esm --platform=browser --outfile=automations-test-bundle.js && esbuild watchdog-test.ts --bundle --format=esm --platform=browser --outfile=watchdog-test-bundle.js && esbuild workspaces-test.ts --bundle --format=esm --platform=browser --outfile=workspaces-test-bundle.js && esbuild goal-questions-test.ts --bundle --format=esm --platform=browser --outfile=goal-questions-test-bundle.js && esbuild dream-test.ts --bundle --format=esm --platform=browser --outfile=dream-test-bundle.js && esbuild p4-test.ts --bundle --format=esm --platform=browser --outfile=p4-test-bundle.js && esbuild harness-api-test.ts --bundle --format=esm --platform=browser --outfile=harness-api-test-bundle.js" },
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js && esbuild core-ext-test.ts --bundle --format=esm --platform=browser --outfile=core-ext-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js && esbuild p2-test.ts --bundle --format=esm --platform=browser --outfile=p2-test-bundle.js && esbuild chain-test.ts --bundle --format=esm --platform=browser --outfile=chain-test-bundle.js && esbuild undo-harness-test.ts --bundle --format=esm --platform=browser --outfile=undo-harness-test-bundle.js && esbuild messages-test.ts --bundle --format=esm --platform=browser --outfile=messages-test-bundle.js && esbuild p3-test.ts --bundle --format=esm --platform=browser --outfile=p3-test-bundle.js && esbuild automations-test.ts --bundle --format=esm --platform=browser --outfile=automations-test-bundle.js && esbuild watchdog-test.ts --bundle --format=esm --platform=browser --outfile=watchdog-test-bundle.js && esbuild workspaces-test.ts --bundle --format=esm --platform=browser --outfile=workspaces-test-bundle.js && esbuild goal-questions-test.ts --bundle --format=esm --platform=browser --outfile=goal-questions-test-bundle.js && esbuild dream-test.ts --bundle --format=esm --platform=browser --outfile=dream-test-bundle.js && esbuild p4-test.ts --bundle --format=esm --platform=browser --outfile=p4-test-bundle.js && esbuild harness-api-test.ts --bundle --format=esm --platform=browser --outfile=harness-api-test-bundle.js && esbuild proxy-models-test.ts --bundle --format=esm --platform=browser --outfile=proxy-models-test-bundle.js" },
"dependencies": {
"@earendil-works/pi-durable": "1.5.0",
"@earendil-works/pi-ai": "1.5.0",
File diff suppressed because one or more lines are too long
+17
View File
@@ -0,0 +1,17 @@
/** Exercise both real pi-ai transports and their catalog prices, without provider SDK mocks. */
import { proxyModels } from './proxy-models';
const models = await proxyModels(globalThis.testProxy);
for (const [provider, id] of [['openai', 'gpt-6.1-sol'], ['anthropic', 'claude-opus-5-5']]) {
const model = models.getModel(provider, id);
if (model.cost.input !== 2 || model.cost.cacheWrite !== 3) throw new Error('catalog prices lost');
const result = await models.completeSimple(model, {messages: [{role: 'user', content: 'Say hello.', timestamp: Date.now()}]}, {reasoning: 'low', maxTokens: 128});
if (result.stopReason === 'error' || !result.content.some(block => block.type === 'text' && block.text === 'hello')) {
throw new Error(`${provider}: ${result.errorMessage ?? JSON.stringify(result.content)}`);
}
if (!(result.usage.cost.total > 0)) throw new Error('turn cost missing');
}
const fallback = await proxyModels({...globalThis.testProxy, base_url: `${globalThis.testProxy.base_url}/offline`});
for (const [provider, api] of [['openai', 'openai-responses'], ['anthropic', 'anthropic-messages']]) {
const model = fallback.getModels(provider)[0];
if (model.api !== api || model.cost.input !== 0) throw new Error('offline transport fallback failed');
}
+47
View File
@@ -0,0 +1,47 @@
/** Host-only proxy catalog, provider transports, authentication, and quoted token prices. */
import { createModels } from '@earendil-works/pi-ai/models';
import { openaiProvider } from '@earendil-works/pi-ai/providers/openai';
import { anthropicProvider } from '@earendil-works/pi-ai/providers/anthropic';
export async function proxyModels(proxy: {base_url: string, api_key: string}) {
const models = createModels();
const base = proxy.base_url.replace(/\/$/, '');
let catalog: any[] = [];
try {
const response = await fetch(`${base}/v1/models`, {
headers: {Authorization: `Bearer ${proxy.api_key}`}, signal: AbortSignal.timeout(5000),
});
if (response.ok) catalog = (await response.json()).data ?? [];
} catch { /* An unavailable catalog must not prevent durable recovery. */ }
for (const [provider, api, path, fallback] of [
[openaiProvider(), 'openai-responses', '/v1/responses', 'gpt-6.1-sol'],
[anthropicProvider(), 'anthropic-messages', '/apis/anthropic/v1/messages', 'claude-opus-5-5'],
] as const) {
const rows = catalog.filter(row => row.creator === provider.id && row.endpoints?.some((endpoint: any) => endpoint.api === api));
const entries = (rows.length ? rows : [{id: fallback}]).map(row => {
const endpoint = row.endpoints?.find((endpoint: any) => endpoint.api === api)?.path ?? path;
// pi-ai appends /responses or /v1/messages, respectively.
const suffix = api === 'anthropic-messages' ? '/v1/messages' : '/responses';
const baseUrl = `${base}${endpoint.slice(0, -suffix.length)}`;
const price = row.providers?.[0]?.pricing;
const scale = price?.currency === 'USD' ? 1000000 / price.unit_tokens : 0;
return {
id: row.id, name: row.name ?? row.id, provider: provider.id, api, baseUrl,
reasoning: row.capabilities?.reasoning ?? true, input: ['text'],
thinkingLevelMap: {off: null, minimal: null,
...Object.fromEntries((row.capabilities?.reasoning_options?.find((option: any) => option.type === 'effort')?.values ?? []).map((level: string) => [level, level]))},
compat: api === 'anthropic-messages' ? {forceAdaptiveThinking: row.capabilities?.reasoning_options?.some((option: any) => option.type === 'effort') ?? true} : undefined,
cost: {input: (price?.input ?? 0) * scale, output: (price?.output ?? 0) * scale,
cacheRead: (price?.cached_input ?? 0) * scale, cacheWrite: (price?.cache_write ?? 0) * scale},
contextWindow: row.limits?.context_window ?? 200000, maxTokens: 4096,
};
});
models.setProvider({...provider, baseUrl: entries[0].baseUrl, getModels: () => entries as any,
auth: {apiKey: {name: 'Sausage proxy', resolve: async () => ({
auth: {apiKey: proxy.api_key, headers: {'Shopify-Usage-Tag': JSON.stringify(['pi', 'sausage'])}},
source: 'llm-proxy',
})}},
});
}
return models;
}
+175 -4
View File
@@ -130,7 +130,15 @@ impl Compiler {
}
}
}
/// Compile programs in both command folders without blocking world startup.
fn priority(name: &str) -> usize {
[
"sh", "git", "rg", "grep", "ls", "cat", "sed", "find", "cargo",
]
.iter()
.position(|tool| *tool == name.strip_suffix(".wasm").unwrap_or(name))
.unwrap_or(9)
}
/// Compile common tools first in both command folders without blocking world startup.
pub(super) fn precompile(
compiler: Arc<Compiler>,
root: Arc<dyn sausage_vfs::Layer>,
@@ -140,18 +148,20 @@ pub(super) fn precompile(
// Leave CPU and file bandwidth for commands during startup.
let slots = Arc::new(tokio::sync::Semaphore::new(2));
for folder in ["/bin", "/usr/local/bin"] {
let entries = match root.list(folder).await {
let mut entries = match root.list(folder).await {
Ok(entries) => entries,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
Err(e) => return Err(e.into()),
};
entries.sort_by_key(|entry| priority(&entry.name));
for entry in entries.into_iter().filter(|entry| !entry.directory) {
// Reserve before spawning so task scheduling cannot reorder admission.
let permit = slots.clone().acquire_owned().await?;
let root = root.clone();
let compiler = compiler.clone();
let path = format!("{folder}/{}", entry.name);
let slots = slots.clone();
tasks.spawn(async move {
let _permit = slots.acquire().await?;
let _permit = permit;
let file = root
.open(
&path,
@@ -186,6 +196,167 @@ mod tests {
use crate::exec::{Command, Exec, Options};
use std::{future::Future, sync::atomic::Ordering};
#[test]
fn common_tools_precede_other_programs() {
let mut names = [
"z", "cat.wasm", "cargo", "sh.wasm", "git.wasm", "rg", "ls", "grep", "sed", "find",
"arch",
];
names.sort_by_key(|name| priority(name));
assert_eq!(
names,
[
"sh.wasm", "git.wasm", "rg", "grep", "ls", "cat.wasm", "sed", "find", "cargo", "z",
"arch"
]
);
}
#[tokio::test(flavor = "multi_thread")]
async fn cold_compilation_does_not_spend_command_timeout() -> anyhow::Result<()> {
use std::time::Duration;
let temp = tempfile::tempdir()?;
let compiler = Compiler::new(&temp.path().join("cache"))?;
let world = crate::world::World::open(temp.path())?;
std::fs::create_dir_all(temp.path().join("world/bin"))?;
let program = temp.path().join("world/bin/program.wasm");
std::fs::write(
&program,
wat::parse_str(r#"(module (memory (export "memory") 1) (func (export "_start")))"#)?,
)?;
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (finish_tx, finish_rx) = std::sync::mpsc::channel();
*compiler.before_compile.lock().unwrap() = Some(Box::new(move || {
let _ = started_tx.send(());
let _ = finish_rx.recv();
}));
let runtime = super::super::Runtime::with_compiler(world, compiler)?;
started_rx.await?;
let command_runtime = runtime.clone();
let mut command = tokio::spawn(async move {
command_runtime
.exec(
Command::Argv(vec!["/bin/program".into()]),
Options {
timeout: Some(Duration::from_secs(2)),
..Default::default()
},
&mut |_, _| {},
)
.await
});
let pending = tokio::time::timeout(Duration::from_millis(2100), &mut command).await;
// Always release the blocking compiler, even when the old deadline fails this assertion.
finish_tx.send(())?;
assert!(
pending.is_err(),
"Compilation consumed the execution deadline: {pending:?}"
);
assert_eq!(command.await??, 0);
std::fs::write(
&program,
wat::parse_str(
r#"(module (memory (export "memory") 1) (func (export "_start") (loop br 0)))"#,
)?,
)?;
let result = runtime
.exec(
Command::Argv(vec!["/bin/program".into()]),
Options {
timeout: Some(Duration::from_millis(100)),
..Default::default()
},
&mut |_, _| {},
)
.await;
assert!(
result
.unwrap_err()
.to_string()
.contains("command timed out")
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn child_compilation_pauses_the_parent_deadline() -> anyhow::Result<()> {
use super::super::{GroupGuard, Request, Stdio, deadline::Deadline};
use std::time::Duration;
let temp = tempfile::tempdir()?;
let world = crate::world::World::open(temp.path())?;
let compiler = Compiler::new(&temp.path().join("cache"))?;
let parent = wat::parse_str(
r#"(module (memory (export "memory") 1) (func (export "_start") (loop br 0)))"#,
)?;
std::fs::write(temp.path().join("world/parent.wasm"), &parent)?;
std::fs::write(
temp.path().join("world/child.wasm"),
wat::parse_str(r#"(module (memory (export "memory") 1) (func (export "_start")))"#)?,
)?;
compiler.get(parent).await?;
let runtime = super::super::Runtime::with_compiler(world, compiler.clone())?;
let deadline = Deadline::new(Duration::ZERO);
let request = |name: &str| Request {
program: name.into(),
argv: vec![name.into()],
cwd: "/".into(),
env: Vec::new(),
group: None,
stdin: Stdio::Null,
stdout: Stdio::Null,
stderr: Stdio::Null,
};
let parent = runtime
.start(
request("/parent.wasm"),
0,
0,
0,
Vec::new(),
None,
None,
Some(deadline.clone()),
)
.await?;
let _guard = GroupGuard {
runtime: runtime.clone(),
group: parent.pid,
};
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (finish_tx, finish_rx) = std::sync::mpsc::channel();
*compiler.before_compile.lock().unwrap() = Some(Box::new(move || {
let _ = started_tx.send(());
let _ = finish_rx.recv();
}));
let child_runtime = runtime.clone();
let child = tokio::spawn(async move {
child_runtime
.start(
request("/child.wasm"),
parent.pid,
parent.pid,
parent.group,
Vec::new(),
None,
None,
None,
)
.await
});
let started = tokio::time::timeout(Duration::from_secs(5), started_rx).await;
let expired = deadline.expired();
tokio::pin!(expired);
let paused = std::future::poll_fn(|cx| {
std::task::Poll::Ready(expired.as_mut().poll(cx).is_pending())
})
.await;
let _ = finish_tx.send(());
started??;
assert!(paused, "Child compilation did not pause the parent budget");
child.await??;
Ok(())
}
#[test]
fn compilation_survives_the_originating_runtime() {
let temp = tempfile::tempdir().unwrap();
+87
View File
@@ -0,0 +1,87 @@
//! Command-wide execution budgets exclude the union of component compilation waits, including child spawns.
use std::{sync::Arc, time::Duration};
use tokio::{sync::watch, time::Instant};
#[derive(Clone, Copy)]
struct State {
deadline: Instant,
compiling: usize,
paused_at: Instant,
}
pub(super) struct Deadline(watch::Sender<State>);
impl Deadline {
pub fn new(duration: Duration) -> Arc<Self> {
let now = Instant::now();
Arc::new(Self(
watch::channel(State {
deadline: now + duration,
compiling: 0,
paused_at: now,
})
.0,
))
}
pub fn pause(self: &Arc<Self>) -> Compilation {
self.0.send_modify(|state| {
if state.compiling == 0 {
state.paused_at = Instant::now();
}
state.compiling += 1;
});
Compilation(self.clone())
}
pub async fn expired(&self) {
let mut state = self.0.subscribe();
loop {
let current = *state.borrow_and_update();
if current.compiling == 0 {
tokio::select! {
biased;
changed = state.changed() => { changed.unwrap(); }
_ = tokio::time::sleep_until(current.deadline) => return,
}
} else {
state.changed().await.unwrap();
}
}
}
}
pub(super) struct Compilation(Arc<Deadline>);
impl Drop for Compilation {
fn drop(&mut self) {
self.0.0.send_modify(|state| {
state.compiling -= 1;
if state.compiling == 0 {
state.deadline += state.paused_at.elapsed();
}
});
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn overlapping_compilations_share_one_pause_and_drop_resumes() {
let deadline = Deadline::new(Duration::from_secs(2));
let first = deadline.pause();
let paused_at = deadline.0.borrow().paused_at;
let second = deadline.pause();
assert_eq!(deadline.0.borrow().compiling, 2);
assert_eq!(deadline.0.borrow().paused_at, paused_at);
let original = deadline.0.borrow().deadline;
drop(first);
assert_eq!(deadline.0.borrow().compiling, 1);
assert_eq!(deadline.0.borrow().deadline, original);
drop(second);
assert_eq!(deadline.0.borrow().compiling, 0);
assert!(deadline.0.borrow().deadline >= original);
}
}
+73 -68
View File
@@ -47,6 +47,7 @@ mod bindings {
}
use bindings::sausage::process::host::{Host, HostChild, Request, Stdio};
mod cache;
mod deadline;
mod http;
pub use http::Router;
pub mod net;
@@ -83,6 +84,7 @@ struct Entry {
command: u32,
group: u32,
cancelled: Arc<AtomicBool>,
deadline: Option<Arc<deadline::Deadline>>,
tag: sausage_vfs::journal::Tag,
abort: AbortHandle,
}
@@ -329,16 +331,17 @@ impl Runtime {
conversations: Vec<String>,
inherited_io: Option<Io>,
tool_call: Option<String>,
deadline: Option<Arc<deadline::Deadline>>,
) -> anyhow::Result<Started> {
anyhow::ensure!(!request.argv.is_empty(), "empty argv");
let cancelled = if parent == 0 {
Arc::new(AtomicBool::new(false))
let (cancelled, deadline) = if parent == 0 {
(Arc::new(AtomicBool::new(false)), deadline)
} else {
self.entries
.lock()
.unwrap()
.get(&parent)
.map(|entry| entry.cancelled.clone())
.map(|entry| (entry.cancelled.clone(), entry.deadline.clone()))
.ok_or_else(|| anyhow::anyhow!("parent exited"))?
};
if !request.env.iter().any(|(k, _)| k == "PATH") {
@@ -366,7 +369,9 @@ impl Runtime {
let bytes = file
.read(0, usize::try_from(file.stat().await?.size)?)
.await?;
let compilation = deadline.as_ref().map(|deadline| deadline.pause());
let pre = self.compiler.get(bytes.clone()).await?;
drop(compilation);
self.cache.lock().unwrap().insert(bytes);
let pid = self.next.fetch_add(1, Ordering::Relaxed);
let tag = if parent == 0 {
@@ -528,6 +533,7 @@ impl Runtime {
command,
group,
cancelled: cancelled.clone(),
deadline,
tag,
abort: task.abort_handle(),
},
@@ -583,6 +589,7 @@ impl Host for State {
self.conversations.clone(),
Some(self.stdio.clone()),
None,
None,
)
.await
{
@@ -707,79 +714,77 @@ impl Exec for Arc<Runtime> {
options: Options,
output: &mut (dyn for<'chunk> FnMut(Stream, &'chunk [u8]) + Send),
) -> anyhow::Result<i32> {
let deadline = options
.timeout
.map(|duration| tokio::time::Instant::now() + duration);
let argv = match command {
Command::Shell(script) => vec![
"/bin/sh".into(),
"--noprofile".into(),
"--norc".into(),
"-c".into(),
script,
],
Command::Argv(argv) => argv,
};
let env = options.env;
let start = self.start(
Request {
program: argv.first().cloned().unwrap_or_default(),
argv,
env,
cwd: options.cwd,
group: None,
stdin: Stdio::Pipe,
stdout: Stdio::Pipe,
stderr: Stdio::Pipe,
},
0,
0,
0,
options.conversations,
None,
options.tool_call,
);
let mut child = if let Some(deadline) = deadline {
tokio::time::timeout_at(deadline, start)
.await
.map_err(|_| anyhow::anyhow!("command timed out"))??
} else {
start.await?
};
let _guard = GroupGuard {
runtime: self.clone(),
group: child.pid,
};
drop(child.input);
let mut stdout = child.output.take().unwrap();
let mut stderr = child.error.take().unwrap();
let deadline = options.timeout.map(deadline::Deadline::new);
let run = async {
let (mut out_open, mut err_open) = (true, true);
let (mut out, mut err) = ([0; 8192], [0; 8192]);
while out_open || err_open {
tokio::select! {
n = stdout.read(&mut out), if out_open => { let n = n?; out_open = n != 0; if n != 0 { output(Stream::Stdout, &out[..n]); } }
n = stderr.read(&mut err), if err_open => { let n = n?; err_open = n != 0; if n != 0 { output(Stream::Stderr, &err[..n]); } }
let argv = match command {
Command::Shell(script) => vec![
"/bin/sh".into(),
"--noprofile".into(),
"--norc".into(),
"-c".into(),
script,
],
Command::Argv(argv) => argv,
};
let env = options.env;
let mut child = self
.start(
Request {
program: argv.first().cloned().unwrap_or_default(),
argv,
env,
cwd: options.cwd,
group: None,
stdin: Stdio::Pipe,
stdout: Stdio::Pipe,
stderr: Stdio::Pipe,
},
0,
0,
0,
options.conversations,
None,
options.tool_call,
deadline.clone(),
)
.await?;
let _guard = GroupGuard {
runtime: self.clone(),
group: child.pid,
};
drop(child.input);
let mut stdout = child.output.take().unwrap();
let mut stderr = child.error.take().unwrap();
let run = async {
let (mut out_open, mut err_open) = (true, true);
let (mut out, mut err) = ([0; 8192], [0; 8192]);
while out_open || err_open {
tokio::select! {
n = stdout.read(&mut out), if out_open => { let n = n?; out_open = n != 0; if n != 0 { output(Stream::Stdout, &out[..n]); } }
n = stderr.read(&mut err), if err_open => { let n = n?; err_open = n != 0; if n != 0 { output(Stream::Stderr, &err[..n]); } }
}
}
}
loop {
if let Some(code) = *child.code.borrow_and_update() {
return Ok(code);
loop {
if let Some(code) = *child.code.borrow_and_update() {
return Ok(code);
}
child.code.changed().await?;
}
child.code.changed().await?;
};
let result = run.await;
if let Some(journal) = self.world.root.journal() {
journal.retain().await?;
}
result
};
let result = if let Some(deadline) = deadline {
tokio::time::timeout_at(deadline, run)
.await
.map_err(|_| anyhow::anyhow!("command timed out"))?
if let Some(deadline) = &deadline {
tokio::select! {
result = run => result,
_ = deadline.expired() => anyhow::bail!("command timed out"),
}
} else {
run.await
};
if let Some(journal) = self.world.root.journal() {
journal.retain().await?;
}
result
}
}
/// Construct the harness's shared executor over the same open world.
+1
View File
@@ -34,6 +34,7 @@ pub fn supervise(runtime: Arc<Runtime>) -> Supervisor {
Vec::new(),
None,
None,
None,
);
let child = tokio::select! { _ = &mut stop => return, child = start => child };
if let Ok(mut child) = child {
+165
View File
@@ -0,0 +1,165 @@
//! Parse extension modules and reject free runtime identifiers outside the world runtime globals.
use deno_ast::swc::{
ast::{Ident, TsInterfaceDecl, TsType},
common::SyntaxContext,
ecma_visit::{Visit, VisitWith},
};
use deno_ast::{MediaType, ParseParams, ProgramRef, parse_module};
use std::path::Path;
// Standard JavaScript globals. World extensions also see Deno.
const GLOBALS: &[&str] = &[
"undefined",
"globalThis",
"Deno",
"console",
"crypto",
"fetch",
"performance",
"setTimeout",
"clearTimeout",
"setInterval",
"clearInterval",
"queueMicrotask",
"structuredClone",
"atob",
"btoa",
"parseInt",
"parseFloat",
"isNaN",
"Infinity",
"NaN",
"Object",
"Function",
"Boolean",
"Number",
"BigInt",
"String",
"Symbol",
"Array",
"Map",
"Set",
"WeakMap",
"WeakSet",
"WeakRef",
"FinalizationRegistry",
"Promise",
"Date",
"RegExp",
"Error",
"TypeError",
"RangeError",
"SyntaxError",
"AggregateError",
"JSON",
"Math",
"Reflect",
"Proxy",
"Intl",
"Iterator",
"Uint8Array",
"ArrayBuffer",
"TextEncoder",
"TextDecoder",
"URL",
"URLSearchParams",
"AbortController",
"AbortSignal",
"Request",
"Response",
"Headers",
"ReadableStream",
"WritableStream",
"TransformStream",
"Blob",
"File",
"FormData",
"DOMException",
"encodeURIComponent",
"decodeURIComponent",
];
fn undefined(source: &str) -> Vec<String> {
let parsed = parse_module(ParseParams {
specifier: "file:///extension.ts".parse().unwrap(),
text: source.into(),
media_type: MediaType::TypeScript,
capture_tokens: false,
scope_analysis: true,
maybe_syntax: None,
})
.unwrap();
struct FreeNames {
context: SyntaxContext,
names: Vec<String>,
}
impl Visit for FreeNames {
// Type signatures contain parameter names but do not execute.
fn visit_ts_type(&mut self, _node: &TsType) {}
fn visit_ts_interface_decl(&mut self, _node: &TsInterfaceDecl) {}
fn visit_ident(&mut self, ident: &Ident) {
if ident.ctxt == self.context && !GLOBALS.contains(&ident.sym.as_ref()) {
self.names.push(ident.sym.to_string());
}
}
}
let mut free = FreeNames {
context: parsed.unresolved_context(),
names: Vec::new(),
};
match parsed.program_ref() {
ProgramRef::Module(module) => module.visit_with(&mut free),
ProgramRef::Script(script) => script.visit_with(&mut free),
}
free.names.sort();
free.names.dedup();
free.names
}
fn scan(directory: &Path, failures: &mut Vec<String>) {
for entry in std::fs::read_dir(directory).unwrap() {
let path = entry.unwrap().path();
if path.is_dir() {
scan(&path, failures);
} else if path.extension().is_some_and(|ext| ext == "ts")
&& !path.to_string_lossy().ends_with(".test.ts")
{
let names = undefined(&std::fs::read_to_string(&path).unwrap());
if !names.is_empty() {
failures.push(format!("{}: {}", path.display(), names.join(", ")));
}
}
}
}
#[test]
fn extension_identifiers_are_defined() {
let mut failures = Vec::new();
scan(
&Path::new(env!("CARGO_MANIFEST_DIR")).join("../extensions"),
&mut failures,
);
assert!(
failures.is_empty(),
"Undefined identifiers:\n{}",
failures.join("\n")
);
}
#[test]
fn catches_the_missing_route_import() {
let source = std::fs::read_to_string(
Path::new(env!("CARGO_MANIFEST_DIR")).join("../extensions/messages/route.ts"),
)
.unwrap();
let old = source.replace(
"import { ConversationsApi } from '../conversations/api.ts'\n",
"",
);
assert!(
undefined(&old)
.iter()
.any(|name| name == "ConversationsApi")
);
assert!(undefined("import { Api } from 'service'; type Callback = (arg: UnknownType) => void; interface Service { use(parameter: UnknownType): void }; function use(host: any) { const local = Api; return host.need(local) }").is_empty());
assert_eq!(
undefined("function use(host: any) { return host.need(MissingApi) }"),
["MissingApi"]
);
}
+6 -1
View File
@@ -27,6 +27,7 @@ fn prepare(dir: &Path) -> anyhow::Result<()> {
"extensions/lineage",
"extensions/models",
"extensions/core",
"extensions/conversations",
"agents",
] {
std::fs::create_dir_all(repo.join(directory))?;
@@ -49,6 +50,10 @@ fn prepare(dir: &Path) -> anyhow::Result<()> {
repo.join("extensions/probe/server.ts"),
"export default host => { globalThis.testHost = host; globalThis.subagentService = host.need({name:'subagents'}); globalThis.messageService = host.need({name:'messages'}); globalThis.lineageService = host.need({name:'lineage'}); host.implement({name:'conversations'}, {summaryModel: async () => globalThis.summaryRef, concludedIn: async () => new Set(), generatedTitle: async () => {}, setStanding: async () => {}, setAutomaticSummaries: async () => {}}); return {name:'probe'}; }",
)?;
std::fs::copy(
source.join("extensions/conversations/api.ts"),
repo.join("extensions/conversations/api.ts"),
)?;
std::fs::copy(
source.join("extensions/core/agents.ts"),
repo.join("extensions/core/agents.ts"),
@@ -67,7 +72,7 @@ fn prepare(dir: &Path) -> anyhow::Result<()> {
)?;
std::fs::write(
repo.join("agents/coordinator.md"),
"---\nname: coordinator\ndescription: Coordinate\nisolate: none\n---\nCoordinator agent\n",
"---\nname: coordinator\ndescription: Coordinate\nmodel: faux/coordinator\nisolate: none\n---\nCoordinator agent\n",
)?;
for args in [
vec!["init", "-b", "main"],
+107
View File
@@ -0,0 +1,107 @@
//! The host proxy catalog selects distinct pi-ai transports with shared proxy authentication.
use sausage::{
exec::{Command, Exec, Options, Stream},
world::World,
};
use std::{sync::Arc, time::Duration};
struct NoExec;
#[async_trait::async_trait]
impl Exec for NoExec {
async fn exec(
&self,
_: Command,
_: Options,
_: &mut (dyn for<'a> FnMut(Stream, &'a [u8]) + Send),
) -> anyhow::Result<i32> {
anyhow::bail!("unexpected world command")
}
}
fn event(name: &str, value: serde_json::Value) -> String {
format!("event: {name}\ndata: {value}\n\n")
}
#[tokio::test(flavor = "current_thread")]
async fn provider_transport_and_prices() -> anyhow::Result<()> {
use serde_json::json;
// Production initializes this in the process runtime; this fixture uses NoExec.
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
let server = tiny_http::Server::http("127.0.0.1:0").map_err(|e| anyhow::anyhow!(e))?;
let base = format!("http://{}", server.server_addr());
let proxy = json!({"base_url":base,"api_key":"test-proxy-key"});
let thread = std::thread::spawn(move || -> anyhow::Result<()> {
for path in [
"/v1/models",
"/v1/responses",
"/apis/anthropic/v1/messages?beta=true",
"/offline/v1/models",
] {
let mut request = server
.recv_timeout(Duration::from_secs(20))?
.ok_or_else(|| anyhow::anyhow!("request timeout"))?;
assert_eq!(request.url(), path);
let authenticated = request.headers().iter().any(|h| {
(h.field.equiv("Authorization") && h.value.as_str() == "Bearer test-proxy-key")
|| (h.field.equiv("x-api-key") && h.value.as_str() == "test-proxy-key")
});
assert!(authenticated, "missing proxy key");
if path == "/offline/v1/models" {
request.respond(tiny_http::Response::empty(503))?;
continue;
}
let (body, mime) = if path == "/v1/models" {
let rows: Vec<_> = [("openai", "gpt-6.1-sol", "openai-responses", "/v1/responses"),
("anthropic", "claude-opus-5-5", "anthropic-messages", "/apis/anthropic/v1/messages")]
.into_iter().map(|(creator,id,api,path)|json!({"creator":creator,"id":id,"endpoints":[{"api":api,"path":path}],
"providers":[{"pricing":{"currency":"USD","unit_tokens":1000,"input":0.002,"output":0.008,"cached_input":0.0001,"cache_write":0.003}}]})).collect();
(json!({"data":rows}).to_string(), "application/json")
} else {
let mut body = String::new();
request.as_reader().read_to_string(&mut body)?;
let body: serde_json::Value = serde_json::from_str(&body)?;
assert_eq!(body["stream"], true);
assert!(
request
.headers()
.iter()
.any(|h| h.field.equiv("Shopify-Usage-Tag")
&& h.value.as_str() == "[\"pi\",\"sausage\"]")
);
let stream = if path == "/v1/responses" {
assert_eq!(body["model"], "gpt-6.1-sol");
[
event("response.output_item.added", json!({"type":"response.output_item.added","output_index":0,"item":{"type":"message","id":"m","role":"assistant","content":[]}})),
event("response.content_part.added", json!({"type":"response.content_part.added","output_index":0,"content_index":0,"part":{"type":"output_text","text":""}})),
event("response.output_text.delta", json!({"type":"response.output_text.delta","output_index":0,"content_index":0,"delta":"hello"})),
event("response.completed", json!({"type":"response.completed","response":{"id":"r","status":"completed","output":[],"usage":{"input_tokens":10,"output_tokens":1,"total_tokens":11}}})),
].concat()
} else {
assert_eq!(body["model"], "claude-opus-5-5");
assert_eq!(body["thinking"]["type"], "adaptive");
assert_eq!(body["output_config"]["effort"], "low");
[
event("message_start", json!({"type":"message_start","message":{"id":"m","type":"message","role":"assistant","model":"claude-opus-5-5","content":[],"usage":{"input_tokens":10,"output_tokens":0}}})),
event("content_block_start", json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}})),
event("content_block_delta", json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}})),
event("content_block_stop", json!({"type":"content_block_stop","index":0})),
event("message_delta", json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}})),
event("message_stop", json!({"type":"message_stop"})),
].concat()
};
(stream, "text/event-stream")
};
request.respond(
tiny_http::Response::from_string(body)
.with_header(tiny_http::Header::from_bytes("Content-Type", mime).unwrap()),
)?;
}
Ok(())
});
let dir = tempfile::tempdir()?;
let source = format!(
"globalThis.testProxy = {proxy};\n{}",
include_str!("../js/proxy-models-test-bundle.js")
);
let result =
sausage::harness::run_module(World::open(dir.path())?, Arc::new(NoExec), &source).await;
thread.join().unwrap()?;
result
}