Load world extensions from commit snapshots with last-good recovery

This commit is contained in:
2026-11-07 10:38:20 +00:00
parent 534fa50615
commit bcb3e4d8e5
17 changed files with 42251 additions and 399 deletions
+82
View File
@@ -0,0 +1,82 @@
# World extensions
The harness loads `/src/sausage` from its `main` branch.
Each extension occupies `extensions/<name>/server.ts` in that repository.
The default export is an extension or a factory that receives a `Host`.
The extension name must match its folder.
An optional numeric `order` export controls order. Names break ties.
## Reload contract
A change to the branch ref or `packed-refs` triggers a reload.
The harness creates a detached world worktree at `/run/sausage/extensions/<sha>`.
Imports read that snapshot through the world VFS, not the editable checkout.
Relative imports must include their file extensions.
TypeScript modules use the same `deno_ast` transpiler as the harness.
The harness imports and validates the whole commit before it changes the live registry.
An import error, factory error, name mismatch, or registry validation error retains the previous commit.
A running tool call keeps its old implementation. The next call uses the new implementation.
Removed extensions leave the registry after a successful reload.
`/proc/extensions` contains JSON with `good`, `current`, and `error`.
A session document persists these values in the harness database.
On startup, the harness tries the tip, then the last good commit, before `resume()`.
Tasks without a registered definition remain blocked.
The snapshot worktrees disappear with `/run` on restart. The loader prunes stale worktree registrations before it recreates them.
## Host
The host supplies these members:
- `harness()`, `storage.scanSubmissions`, `models()`, and the conversation `metadata` document token.
- `settings()`, from `/src/sausage/settings.json`. The VFS watcher detects changes to this file.
- `fs.read`, `write`, `append`, `list`, `stat`, `lstat`, `mkdir`, `remove`, and `rename` on world paths.
- `exec(command, options)` for shell strings or argument arrays. Its result contains `code`, `stdout`, and `stderr`.
- Exec options: `cwd`, `env` pairs, `timeout` in seconds, `conversations`, `signal`, and `onOutput(text, stream)`.
- `git(cwd, ...args)`, which returns trimmed stdout and throws with stderr on failure.
- `implement(api, value)`, which stages a service until the commit installs.
- `need(api)`, which returns a proxy. Every method call resolves the current provider, including a retained method reference.
- `want(api)`, which returns a proxy or `undefined` when no provider exists.
- `onOpen(callback)` and `onClose(callback)` for facet lifecycle.
- `resource(key, create)`, which retains a value across reloads. Keys belong to their extension name.
- `notice(message)`, which writes a console line and the latest notice to `/proc/extensions`.
P3-d will add `emit`, `react`, and `eventLog` at the marked seam in `host/js/extensions.ts`.
Those APIs will supply transactional events and persistent reactor cursors.
The synthetic package modules reuse the bundle's instances:
- `@earendil-works/pi-durable` and `/tools`
- `@earendil-works/pi-ai` and `/models`
- `@earendil-works/pi-schema`
- `@earendil-works/chord` and `/context`
New package surfaces require a namespace in `registerModules()`.
They must not bundle a second pi-durable instance.
The registry supplies the default extension selection for existing conversations.
New root conversations do not pin the built-ins.
Explicit per-conversation extension selections remain explicit.
## Revert
Run this command inside the world:
```sh
extensions-revert
```
The Rust program lives in `userland/exec-tools/src/revert.rs`.
It reads `/src/sausage/.git/sausage-last-good` and calls world git to update `refs/heads/main`.
The loader writes this helper file after it persists the session status.
An empty helper file means no commit succeeded yet.
The command preserves the editable checkout. The branch change triggers the normal reload.
## Tests
`host/tests/extensions.rs` runs `host/js/extensions-test-bundle.js` with world git and a faux model.
It tests A, broken B, changed C, an old call across reload, a retained service handle, and restart recovery.
It also tests the revert command, file operations, and exec environment options.
The world git and helper components use the executor's shared `CARGO_TARGET_DIR`.
The API and hook fixtures use that directory too.
The Nix shell overrides `CARGO_HOME`. Unset that variable before Cargo commands to preserve the configured kache wrapper.
+9995 -361
View File
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+78
View File
@@ -0,0 +1,78 @@
// A world git repo exercises commit reloads, old calls, registry defaults, and recovery.
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
import { createModels } from '@earendil-works/pi-ai/models';
import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
import { Harness, createRegistry } from '@earendil-works/pi-durable';
import { openStorage } from './sqlite';
import { loadExtensions, fs, git, exec } from './extensions';
const assert = (value:any,message:string)=>{if(!value)throw new Error(message);};
const until = async (test:()=>Promise<boolean>|boolean)=>{const deadline=Date.now()+15000;while(!await test()){if(Date.now()>deadline)throw new Error('Reload deadline');await new Promise(r=>setTimeout(r,0));}};
const models=createModels(), faux=fauxProvider({tokensPerSecond:100000});models.setProvider(faux.provider);
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'}}});
globalThis.shared={gate:Promise.withResolvers(),started:false,wait:false};
const loader=await loadExtensions(harness,registry,storage,models,context);
harness.resume();
const turn = async()=> {
faux.setResponses([fauxAssistantMessage([fauxToolCall('demo',{})],{stopReason:'toolUse'}),fauxAssistantMessage('done')]);
const submission=await root.submit({type:'input',content:'call demo'},context);
await submission.wait(context);
const entries=(await root.context(context)).entries;
const result=entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1);
return JSON.stringify(result);
};
try {
if(globalThis.extensionRestart) {
assert(loader.state().error,'Broken tip must report its error');
assert(await fs.read('/src/sausage/.git/sausage-last-good')===globalThis.extensionGood,'Last good must survive');
assert((await turn()).includes('result C'),'Restart must use C');
} else {
const env = await exec(['env'],{cwd:'/',env:[['EXTENSION_OPTION','yes']],conversations:['1'],timeout:3});
assert(env.code===0 && env.stdout.includes('EXTENSION_OPTION=yes'),'Exec environment');
await fs.write('/src/sausage/temporary','one');await fs.append('/src/sausage/temporary','two');
await fs.rename('/src/sausage/temporary','/src/sausage/renamed');
assert(await fs.read('/src/sausage/renamed')==='onetwo','File mutation');
assert((await fs.lstat('/src/sausage/renamed')).kind==='file','lstat');await fs.remove('/src/sausage/renamed');
const source=(version:string)=>`import {defineExtension,defineTool} from '@earendil-works/pi-durable';
import {obj} from '@earendil-works/pi-schema';
export default host=>{if(defineTool!==globalThis.__sausage.modules['@earendil-works/pi-durable'].defineTool)throw new Error('Duplicate durable instance');host.implement({name:'demo.service'},{value:()=> '${version}'});
globalThis.shared.handle=host.resource('handle',()=>host.need({name:'demo.service'}));
return defineExtension({name:'demo',tools:[defineTool({name:'demo',description:'Demo',parameters:obj({}),execute:async()=>{if(globalThis.shared.wait){globalThis.shared.started=true;await globalThis.shared.gate.promise;}return {output:[{type:'text',text:'result ${version}'}]};}})]});};`;
await fs.mkdir('/src/sausage/extensions');await fs.mkdir('/src/sausage/extensions/demo');
await fs.write('/src/sausage/extensions/demo/server.ts',source('A'));
await git('/src/sausage','add','.');await git('/src/sausage','commit','-m','A');
const a=await git('/src/sausage','rev-parse','HEAD');
await until(()=>loader.state().good===a);
assert((await turn()).includes('result A'),'Existing conversation must see A');
const held=globalThis.shared.handle;
assert(held.value()==='A','A service');
const retainedMethod=held.value;
await fs.write('/src/sausage/extensions/demo/server.ts','syntax !!! error');
await git('/src/sausage','add','.');await git('/src/sausage','commit','-m','B');
const b=await git('/src/sausage','rev-parse','HEAD');
await until(()=>loader.state().current===b && !!loader.state().error);
assert(loader.state().good===a,'Broken B must keep A');
assert(JSON.parse(await fs.read('/proc/extensions')).error,'Status must report B');
assert(held.value()==='A','A service remains');
assert((await turn()).includes('result A'),'A tool remains after broken B');
globalThis.shared.wait=true;
const running=turn();await until(()=>globalThis.shared.started);
await fs.write('/src/sausage/extensions/demo/server.ts',source('C'));
await git('/src/sausage','add','.');await git('/src/sausage','commit','-m','C');
const c=await git('/src/sausage','rev-parse','HEAD');
await until(()=>loader.state().good===c);
assert(held.value()==='C','A need handle must resolve C');
assert(retainedMethod()==='C','A retained method must resolve C');
globalThis.shared.wait=false;globalThis.shared.gate.resolve();
assert((await running).includes('result A'),'Old call must finish on A');
assert((await turn()).includes('result C'),'Next call must use C');
await git('/src/sausage','update-ref','refs/heads/main',b);
await until(()=>loader.state().current===b && !!loader.state().error);
assert((await exec(['extensions-revert'],{cwd:'/'})).code===0,'Revert command');
await until(()=>loader.state().current===c && !loader.state().error);
await git('/src/sausage','update-ref','refs/heads/main',b);
await until(()=>loader.state().current===b && !!loader.state().error);
await fs.write('/src/sausage/good.txt',c);
}
} finally {globalThis.shared.gate.resolve();await loader.close();await harness.close(context);}
+161
View File
@@ -0,0 +1,161 @@
// World extension snapshots, staged services, lifecycle, and last-good recovery.
import * as durable from '@earendil-works/pi-durable';
import * as tools from '@earendil-works/pi-durable/tools';
import * as ai from '@earendil-works/pi-ai';
import * as modelApi from '@earendil-works/pi-ai/models';
import * as chord from '@earendil-works/chord';
import * as contextApi from '@earendil-works/chord/context';
import * as schema from '@earendil-works/pi-schema';
const ops = Deno.core.ops;
const encoder = new TextEncoder();
const decoder = new TextDecoder();
export const ExtensionState = durable.defineDoc({ kind: 'sausage.extensions', version: 1, scope: 'session', initial: () => ({good:null, current:null, error:null}) });
export const Metadata = durable.defineDoc({ kind: 'sausage.metadata', version: 1, scope: 'conversation', history:'latest', fork:'current', initial:()=>({label:'',title:''}), checkpointWhen:()=>false });
export type Host = {
harness(): durable.Harness; storage: Pick<durable.Storage, 'scanSubmissions'>;
models(): modelApi.Models; settings(): durable.JsonObject; metadata: typeof Metadata;
fs: typeof fs; exec: typeof exec; git: typeof git;
implement<S>(api: {name:string;readonly service?:S}, value:NoInfer<S>):void;
need<S>(api:{name:string;readonly service?:S}):S; want<S>(api:{name:string;readonly service?:S}):S|undefined;
resource<T>(key:string, create:()=>T):T;
onOpen(cb:()=>Promise<void>):void; onClose(cb:()=>Promise<void>):void;
notice(message:string):void;
// P3-d adds emit/react/eventLog here, with atomic events and durable cursors.
};
async function file(kind:string, path:string, bytes:number[]=[]):Promise<any> {
const result = await ops.op_file(kind,path,bytes);
if (!result.ok) throw Object.assign(new Error(result.error.message), result.error);
return result.value;
}
export const fs = {
read: async (path:string) => decoder.decode(new Uint8Array(await file('read',path))),
write: (path:string, text:string) => file('write',path,Array.from(encoder.encode(text))),
append: (path:string, text:string) => file('append',path,Array.from(encoder.encode(text))),
list: (path:string) => file('list',path), stat: (path:string) => file('stat',path),
lstat: (path:string) => file('lstat',path), mkdir: (path:string) => file('mkdir',path),
remove: (path:string) => file('remove',path),
rename: (from:string,to:string) => file('rename',from,Array.from(encoder.encode(to))),
};
export async function exec(command:string|string[], options:{cwd:string;env?:[string,string][];timeout?:number;conversations?:string[];signal?:AbortSignal;onOutput?:(text:string,stream:string)=>void}) {
options.signal?.throwIfAborted();
const rid = ops.op_exec(command, options.cwd, options.timeout, undefined, options.env, options.conversations);
const abort = () => Deno.core.close(rid);
options.signal?.addEventListener('abort',abort,{once:true});
const decoders:any = {stdout:new TextDecoder(),stderr:new TextDecoder()};
const result = {code:0,stdout:'',stderr:''};
try {
for (;;) {
const chunk = await ops.op_exec_next(rid);
if (!chunk.stream) {
if (chunk.error) throw new Error(chunk.error);
result.code = chunk.exitCode;
for (const stream of ['stdout','stderr']) { const text = decoders[stream].decode(); result[stream] += text; if(text) options.onOutput?.(text,stream); }
return result;
}
const text = decoders[chunk.stream].decode(new Uint8Array(chunk.bytes),{stream:true});
result[chunk.stream] += text;
options.onOutput?.(text,chunk.stream);
}
} finally { options.signal?.removeEventListener('abort',abort); }
}
export async function git(cwd:string,...args:string[]) {
const result = await exec(['git',...args],{cwd,timeout:30});
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
}
export function registerModules() {
const modules = {'@earendil-works/pi-schema':schema,'@earendil-works/pi-durable':durable,'@earendil-works/pi-durable/tools':tools,'@earendil-works/pi-ai':ai,'@earendil-works/pi-ai/models':modelApi,'@earendil-works/chord':chord,'@earendil-works/chord/context':contextApi};
globalThis.__sausage = {modules};
ops.op_modules(Object.fromEntries(Object.entries(modules).map(([name,value])=>[name,Object.keys(value)])));
}
export async function loadExtensions(harness:any, registry:any, storage:any, models:any, context:any, builtins:any[]=[]) {
registerModules();
const repo = '/src/sausage';
let providers = new Map<string,any>();
const resources = new Map<string,any>();
let facets:any[] = [];
let state = {...(await harness.snapshot(ExtensionState,context) ?? {good:null,current:null,error:null})};
let settings:any = {};
let active = true;
let queue = Promise.resolve();
const publish = async () => {
await harness.commit(async (tx:any)=>Object.assign(await tx.doc(ExtensionState),state),context);
ops.op_extension_status(JSON.stringify(state));
await fs.write('/src/sausage/.git/sausage-last-good',state.good ?? '');
};
const dispose = async (entries:any[]) => { for(const facet of entries) for(const cleanup of facet.close) { try {await cleanup();} catch(error) {ops.op_output(`Extension cleanup: ${String(error)}\n`);} } };
const scope = (name:string, staged:Map<string,any>) => {
const open:any[] = [], close:any[] = [];
const handle = (api:any) => new Proxy({}, {get:(_,key)=> {
if (typeof key === 'symbol' || key === 'then') return undefined;
const value = providers.get(api.name);
if(!value) throw new Error(`No provider for ${api.name}`);
const member = value[key];
return typeof member === 'function' ? (...args:any[]) => {
const current = providers.get(api.name);
if(!current) throw new Error(`No provider for ${api.name}`);
return Reflect.apply(current[key],current,args);
} : member;
}});
const host:Host = {
harness:()=>harness, storage, models:()=>models, settings:()=>settings, metadata:Metadata, fs,exec,git,
implement:(api,value)=>{staged.set(api.name,value);}, need:handle,
want:(api)=>providers.has(api.name)?handle(api):undefined,
resource:(key,create)=>{const id=`${name}/${key}`;if(!resources.has(id))resources.set(id,create());return resources.get(id);},
onOpen:cb=>open.push(cb),onClose:cb=>close.push(cb),notice:message=>{state.notice=`${name}: ${message}`;ops.op_extension_status(JSON.stringify(state));ops.op_output(`${name}: ${message}\n`);},
};
return {host,open,close};
};
const load = async (sha:string) => {
const next:any[] = [];
const staged = new Map<string,any>();
try {
const snapshot = `/run/sausage/extensions/${sha}`;
for(const path of ['/run/sausage','/run/sausage/extensions']) {try {await fs.mkdir(path);} catch(error) {if((error as any).code!=='already_exists')throw error;} }
try {await fs.stat(snapshot);} catch {await git(repo,'worktree','prune');await git(repo,'worktree','add','--detach',snapshot,sha);}
const modules:any[] = [];
for(const entry of await fs.list(`${snapshot}/extensions`).catch(error=>{if(error.code==='not_found')return [];throw error;})) {
if(entry.kind!=='directory')continue;
const path = `${snapshot}/extensions/${entry.name}/server.ts`;
try {await fs.stat(path);} catch(error) {if((error as any).code==='not_found')continue;throw error;}
const module = await import(/* @vite-ignore */ `file://${path}`);
modules.push({name:entry.name,module});
}
modules.sort((a,b)=>(a.module.order??0)-(b.module.order??0)||a.name.localeCompare(b.name));
for(const {name,module} of modules) {
const facet = scope(name,staged); next.push(facet);
facet.extension = typeof module.default==='function'?await module.default(facet.host):module.default;
if(!facet.extension || facet.extension.name!==name)throw new Error(`Extension name must match ${name}`);
}
// Validate the complete commit before the live registry changes.
const probe = durable.createRegistry();
for(const extension of [...builtins,...next.map(f=>f.extension)])probe.install(extension);
for(const old of facets)registry.uninstall(old.extension);
for(const extension of [...builtins,...next.map(f=>f.extension)])registry.install(extension);
const old = facets; facets=next; providers=staged;
await dispose(old);
for(const facet of next)for(const cb of facet.open) {try {await cb();} catch(error) {facet.host.notice(`onOpen: ${String(error)}`);} }
state.good=sha;
return null;
} catch(error) {await dispose(next); return String(error);}
};
const reload = async () => {
let tip:string;
try {tip=await git(repo,'rev-parse','refs/heads/main');} catch(error) {state.error=String(error);ops.op_extension_status(JSON.stringify(state));return;}
try {settings=JSON.parse(await fs.read(`${repo}/settings.json`));} catch(error) {if((error as any).code==='not_found')settings={};else {state.notice=`Settings: ${String(error)}`;ops.op_extension_status(JSON.stringify(state));ops.op_output(`${state.notice}\n`);}}
if(state.current===tip && facets.length)return;
state.current=tip;
state.error=await load(tip);
if(state.error && !facets.length && state.good && state.good!==tip) {
const failed=state.error;
const fallback=await load(state.good);
state.error=fallback?`${failed}\nLast good: ${fallback}`:failed;
}
await publish();
};
const enqueue = () => queue=queue.then(reload);
const rid = ops.op_watch([`${repo}/.git/refs/heads/main`,`${repo}/.git/packed-refs`,`${repo}/settings.json`]);
await enqueue();
const pending = (async()=> {while(active){await ops.op_watch_next(rid);if(active)await enqueue();}})();
return {reload:enqueue, state:()=>({...state}), close:async()=>{active=false;Deno.core.close(rid);await pending;await queue;await dispose(facets);}};
}
+12 -3
View File
@@ -2,12 +2,13 @@
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 { Harness, createRegistry } from '@earendil-works/pi-durable';
import { Harness, AgentDoc, createRegistry } from '@earendil-works/pi-durable';
import { CodingTools } from '@earendil-works/pi-durable/tools';
import { openStorage } from './sqlite';
import { environment } from './environment';
import { CredentialRequests, startRequests } from './credentials';
import { serve } from './input';
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();
@@ -30,10 +31,16 @@ registry.install(CodingTools);
registry.install(CredentialRequests);
export const storage = await openStorage(globalThis.storagePath);
const harness = await Harness.open(storage, {
models, registry, settings: { followUpMode: 'all' },
models, registry, settings: { followUpMode: 'all', get extensions() { return registry.snapshot().installed(); } },
env: (target: any) => environment(target.cwd ?? '/', target.conversationId),
}, context);
const root = await harness.root(context, { agent: { model: { provider: 'openai', modelId: model.id }, thinkingLevel: 'low', cwd: '/', extensions: [CodingTools, CredentialRequests] } });
const root = await harness.root(context, { agent: { model: { provider: 'openai', modelId: model.id }, thinkingLevel: 'low', cwd: '/' } });
// T32 stored this exact built-in list in the console root. Use registry defaults now.
const storedAgent = await harness.snapshot(AgentDoc, root.id, context);
if (Array.isArray(storedAgent?.extensions) && storedAgent.extensions.length === 2 &&
[CodingTools, CredentialRequests].every(extension => storedAgent.extensions.includes(extension.name))) {
await root.configure({ extensions: null }, context);
}
const watch = await root.watch(context);
const emitted = new Map<number, string>();
let partial = '';
@@ -58,6 +65,7 @@ for (const entry of watch.value.entries) {
}
if (globalThis.consoleEnabled) watch.start((event: any) => { if (event.kind === 'value') render(event.value); });
const stopRequests = await startRequests(harness, context);
const extensions = await loadExtensions(harness, registry, storage, models, context, [CodingTools, CredentialRequests]);
harness.resume();
await serve(harness, context, globalThis.consoleEnabled ? async () => {
await harness.waitForIdle(context);
@@ -66,6 +74,7 @@ await serve(harness, context, globalThis.consoleEnabled ? async () => {
await final.stop();
Deno.core.ops.op_output('\n');
} : undefined);
await extensions.close();
await stopRequests();
await watch.stop();
await harness.close(context);
+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" },
"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" },
"dependencies": {
"@earendil-works/pi-durable": "1.5.0",
"@earendil-works/pi-ai": "1.5.0",
+58 -5
View File
@@ -40,9 +40,15 @@ pub async fn op_file(
Ok(Value::Null)
}
"mkdir" => { root.mkdir(&path).await?; Ok(Value::Null) }
"stat" => {
let s = root.stat(&path, true).await?;
Ok(json!({"path": path, "name": path.rsplit('/').next().unwrap_or(""), "kind": if s.directory {"directory"} else {"file"}, "size": s.size, "mtimeMs": 0}))
"list" => Ok(json!(root.list(&path).await?.into_iter().map(|e| json!({"name":e.name,"kind":if e.symlink {"symlink"} else if e.directory {"directory"} else {"file"}})).collect::<Vec<_>>())),
"remove" => {
if root.stat(&path, false).await?.directory { root.remove_dir(&path).await?; } else { root.remove_file(&path).await?; }
Ok(Value::Null)
}
"rename" => { root.rename(&path, &String::from_utf8(bytes).map_err(|_| std::io::ErrorKind::InvalidInput)?).await?; Ok(Value::Null) }
"stat" | "lstat" => {
let s = root.stat(&path, kind == "stat").await?;
Ok(json!({"path": path, "name": path.rsplit('/').next().unwrap_or(""), "kind": if s.symlink {"symlink"} else if s.directory {"directory"} else {"file"}, "size": s.size, "mtimeMs": 0}))
}
_ => Err(std::io::ErrorKind::Unsupported.into()),
}
@@ -50,7 +56,7 @@ pub async fn op_file(
Ok(match result {
Ok(value) => json!({"ok": true, "value": value}),
Err(e) => {
json!({"ok": false, "error": {"code": match e.kind() { std::io::ErrorKind::NotFound => "not_found", std::io::ErrorKind::PermissionDenied => "permission_denied", _ => "unknown" }, "message": e.to_string(), "path": path}})
json!({"ok": false, "error": {"code": match e.kind() { std::io::ErrorKind::NotFound => "not_found", std::io::ErrorKind::AlreadyExists => "already_exists", std::io::ErrorKind::PermissionDenied => "permission_denied", _ => "unknown" }, "message": e.to_string(), "path": path}})
}
})
}
@@ -77,6 +83,8 @@ pub fn op_exec(
#[string] cwd: String,
timeout: Option<f64>,
#[string] conversation: Option<String>,
#[serde] env: Option<Vec<(String, String)>>,
#[serde] conversations: Option<Vec<String>>,
) -> Result<u32, JsErrorBox> {
let exec = state.borrow::<Environment>().exec.clone();
let command = if let Some(s) = command.as_str() {
@@ -88,7 +96,7 @@ pub fn op_exec(
};
let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
let task = tokio::spawn(async move {
let result = exec.exec(command, Options { cwd, conversations: conversation.into_iter().collect(), timeout: timeout.map(std::time::Duration::from_secs_f64), ..Default::default() }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await;
let result = exec.exec(command, Options { cwd, env: env.unwrap_or_default(), conversations: conversations.unwrap_or_else(|| conversation.into_iter().collect()), timeout: timeout.map(std::time::Duration::from_secs_f64) }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await;
let _ = sender.send(match result {
Ok(exit) => json!({"exitCode": exit}),
Err(e) => json!({"error": e.to_string()}),
@@ -129,6 +137,51 @@ pub async fn op_exec_next(
Ok(value)
}
struct Watch {
events: tokio::sync::Mutex<
tokio::sync::broadcast::Receiver<(String, sausage_vfs::observe::ChangeKind)>,
>,
closed: tokio::sync::Notify,
}
impl deno_core::Resource for Watch {
fn close(self: Rc<Self>) {
self.closed.notify_one();
}
}
#[op2]
pub fn op_watch(state: &mut OpState, #[serde] paths: Vec<String>) -> Result<u32, JsErrorBox> {
let observer = state
.borrow::<Environment>()
.root
.observer()
.ok_or_else(|| JsErrorBox::generic("filesystem has no observer"))?;
Ok(state.resource_table.add(Watch {
events: tokio::sync::Mutex::new(observer.subscribe(paths)),
closed: Default::default(),
}))
}
#[op2]
#[serde]
pub async fn op_watch_next(
state: Rc<RefCell<OpState>>,
rid: u32,
) -> Result<serde_json::Value, JsErrorBox> {
let watch = state
.borrow()
.resource_table
.get::<Watch>(rid)
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
let mut events = watch.events.lock().await;
let result = tokio::select! { value = events.recv() => value, _ = watch.closed.notified() => return Ok(Value::Null) };
Ok(match result {
Ok((path, kind)) => json!({"path":path,"kind":format!("{kind:?}")}),
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
json!({"path":"/","kind":"Overflow"})
}
Err(_) => Value::Null,
})
}
#[op2(fast)]
pub fn op_output(#[string] text: &str) {
print!("{text}");
+144
View File
@@ -0,0 +1,144 @@
//! Loads committed world modules and shares the bundle's package instances and status.
use deno_core::{
ModuleLoadOptions, ModuleLoadReferrer, ModuleLoadResponse, ModuleLoader, ModuleSource,
ModuleSourceCode, ModuleSpecifier, ModuleType, OpState, ResolutionKind, op2,
};
use deno_error::JsErrorBox;
use sausage_vfs::{Layer, Open};
use std::{
cell::RefCell,
collections::BTreeMap,
path::{Path, PathBuf},
rc::Rc,
sync::{Arc, Mutex, OnceLock},
};
pub(crate) type Status = Arc<Mutex<String>>;
pub(crate) fn status(path: &Path) -> Status {
static STATES: OnceLock<Mutex<BTreeMap<PathBuf, Status>>> = OnceLock::new();
STATES
.get_or_init(Default::default)
.lock()
.unwrap()
.entry(path.to_owned())
.or_insert_with(|| {
Arc::new(Mutex::new(
"{\"good\":null,\"current\":null,\"error\":null}".into(),
))
})
.clone()
}
pub(super) struct Loader {
pub root: Arc<dyn Layer>,
pub modules: RefCell<BTreeMap<String, String>>,
}
#[op2]
pub fn op_modules(state: &mut OpState, #[serde] modules: BTreeMap<String, Vec<String>>) {
let loader = state.borrow::<Rc<Loader>>();
for (name, exports) in modules {
let quoted = serde_json::to_string(&name).unwrap();
let code = exports.into_iter().enumerate().map(|(index,key)| format!("const e{index} = globalThis.__sausage.modules[{quoted}].{key}; export {{ e{index} as {key} }};\n")).collect();
loader
.modules
.borrow_mut()
.insert(format!("sausage:{name}"), code);
}
}
#[op2(fast)]
pub fn op_extension_status(state: &mut OpState, #[string] value: String) {
*state.borrow::<Status>().lock().unwrap() = value;
}
pub(super) fn transpile(name: &str, source: String) -> Result<String, JsErrorBox> {
if !name.ends_with(".ts") {
return Ok(source);
}
let parsed = deno_ast::parse_module(deno_ast::ParseParams {
specifier: ModuleSpecifier::parse(name).map_err(|e| JsErrorBox::generic(e.to_string()))?,
text: source.into(),
media_type: deno_ast::MediaType::TypeScript,
capture_tokens: false,
scope_analysis: false,
maybe_syntax: None,
})
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
Ok(parsed
.transpile(
&Default::default(),
&Default::default(),
&deno_ast::EmitOptions {
source_map: deno_ast::SourceMapOption::None,
..Default::default()
},
)
.map_err(|e| JsErrorBox::generic(e.to_string()))?
.into_source()
.text)
}
impl ModuleLoader for Loader {
fn resolve(
&self,
specifier: &str,
referrer: &str,
_: ResolutionKind,
) -> Result<ModuleSpecifier, JsErrorBox> {
if specifier.starts_with("@earendil-works/") {
return ModuleSpecifier::parse(&format!("sausage:{specifier}"))
.map_err(|e| JsErrorBox::generic(e.to_string()));
}
deno_core::resolve_import(specifier, referrer)
.map_err(|e| JsErrorBox::generic(e.to_string()))
}
fn load(
&self,
specifier: &ModuleSpecifier,
_: Option<&ModuleLoadReferrer>,
_: ModuleLoadOptions,
) -> ModuleLoadResponse {
if let Some(source) = self.modules.borrow().get(specifier.as_str()) {
return ModuleLoadResponse::Sync(Ok(ModuleSource::new(
ModuleType::JavaScript,
ModuleSourceCode::String(source.clone().into()),
specifier,
None,
)));
}
let root = self.root.clone();
let specifier = specifier.clone();
ModuleLoadResponse::Async(Box::pin(async move {
let work = async {
let path = specifier
.to_file_path()
.map_err(|_| JsErrorBox::generic(format!("Unknown module {specifier}")))?;
let file = root
.open(
path.to_str().unwrap(),
Open {
follow: true,
..Default::default()
},
)
.await
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
let size = file
.stat()
.await
.map_err(|e| JsErrorBox::generic(e.to_string()))?
.size;
let source = String::from_utf8(
file.read(0, size as usize)
.await
.map_err(|e| JsErrorBox::generic(e.to_string()))?,
)
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
transpile(specifier.as_str(), source)
}
.await?;
Ok(ModuleSource::new(
ModuleType::JavaScript,
ModuleSourceCode::String(work.into()),
&specifier,
None,
))
}))
}
}
+16 -24
View File
@@ -2,6 +2,7 @@
mod credentials;
mod environment;
pub mod input;
pub(crate) mod loader;
use crate::{exec::Exec, world::World};
use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
use deno_error::JsErrorBox;
@@ -87,6 +88,10 @@ deno_core::extension!(
environment::op_exec,
environment::op_exec_next,
input::op_input,
environment::op_watch,
environment::op_watch_next,
loader::op_modules,
loader::op_extension_status,
environment::op_output
]
);
@@ -150,7 +155,12 @@ pub async fn run_module_with_input(
source: &str,
) -> anyhow::Result<()> {
let proxy = credentials::proxy(&cred)?;
let loader = std::rc::Rc::new(loader::Loader {
root: world.root.clone(),
modules: Default::default(),
});
let mut runtime = JsRuntime::new(RuntimeOptions {
module_loader: Some(loader.clone()),
extensions: vec![
deno_telemetry::deno_telemetry::init(),
deno_webidl::deno_webidl::init(),
@@ -166,33 +176,15 @@ pub async fn run_module_with_input(
storage::init(),
],
extension_transpiler: Some(std::rc::Rc::new(|name, source| {
if !name.ends_with(".ts") {
return Ok((source, None));
}
let parsed = deno_ast::parse_module(deno_ast::ParseParams {
specifier: deno_core::url::Url::parse(&name).unwrap(),
text: source.into(),
media_type: deno_ast::MediaType::TypeScript,
capture_tokens: false,
scope_analysis: false,
maybe_syntax: None,
})
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
let emitted = parsed
.transpile(
&Default::default(),
&Default::default(),
&deno_ast::EmitOptions {
source_map: deno_ast::SourceMapOption::None,
..Default::default()
},
)
.map_err(|e| JsErrorBox::generic(e.to_string()))?
.into_source();
Ok((emitted.text.into(), None))
Ok((loader::transpile(&name, source.to_string())?.into(), None))
})),
..Default::default()
});
runtime.op_state().borrow_mut().put(loader);
runtime
.op_state()
.borrow_mut()
.put(loader::status(&world.host_dir));
runtime
.op_state()
.borrow_mut()
+4
View File
@@ -61,6 +61,10 @@ impl World {
Arc::new(std::sync::Mutex::new(Default::default()));
let snapshot = processes.clone();
let mut proc = Generated::default();
let extensions = crate::harness::loader::status(&host_dir);
proc.insert("/extensions", move || {
Ok(extensions.lock().unwrap().as_bytes().to_vec())
});
proc.insert("/processes", move || {
let mut text = String::from("PID\tPPID\tSTATE\tCWD\tCONVERSATION\tARGV\n");
for (pid, info) in snapshot.lock().unwrap().iter() {
+122
View File
@@ -0,0 +1,122 @@
//! Commit snapshots and last-good recovery with world git and a faux model.
use sausage::{
exec::{Command, Exec, Options, Stream},
world::World,
};
use std::{path::Path, sync::Arc};
const BUNDLE: &str = include_str!("../js/extensions-test-bundle.js");
struct Author(Arc<dyn Exec>);
#[async_trait::async_trait]
impl Exec for Author {
async fn exec(
&self,
command: Command,
mut options: Options,
output: &mut (dyn for<'a> FnMut(Stream, &'a [u8]) + Send),
) -> anyhow::Result<i32> {
for (key, value) in [
("GIT_AUTHOR_NAME", "Test"),
("GIT_AUTHOR_EMAIL", "test@example.com"),
("GIT_COMMITTER_NAME", "Test"),
("GIT_COMMITTER_EMAIL", "test@example.com"),
] {
options.env.push((key.into(), value.into()));
}
self.0.exec(command, options, output).await
}
}
#[tokio::test(flavor = "current_thread")]
async fn commits_and_restart() -> anyhow::Result<()> {
let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
let target = std::env::var_os("CARGO_TARGET_DIR")
.map(std::path::PathBuf::from)
.expect("CARGO_TARGET_DIR must name the executor's shared target directory");
if !userland.join("cwd-init/init.o").is_file() {
assert!(
std::process::Command::new("nu")
.args(["--no-config-file", "cwd-init/build.nu"])
.current_dir(&userland)
.status()?
.success()
);
}
assert!(
std::process::Command::new("cargo")
.args([
"build",
"--release",
"-p",
"git",
"-p",
"sausage-exec-tools",
"--manifest-path"
])
.arg(userland.join("Cargo.toml"))
.args(["--target", "wasm32-wasip2", "--target-dir"])
.arg(&target)
.args(["-j", "2", "--locked"])
.current_dir(&userland)
.status()?
.success()
);
let dir = tempfile::tempdir()?;
let world = World::open(dir.path())?;
let disk = dir.path().join("world");
std::fs::create_dir_all(disk.join("bin"))?;
std::fs::copy(
target.join("wasm32-wasip2/release/git.wasm"),
disk.join("bin/git.wasm"),
)?;
for name in ["extensions-revert", "env"] {
std::fs::copy(
target.join(format!("wasm32-wasip2/release/{name}.wasm")),
disk.join(format!("bin/{name}.wasm")),
)?;
}
let repo = disk.join("src/sausage");
std::fs::create_dir_all(&repo)?;
assert!(
std::process::Command::new("git")
.args(["init", "-b", "main"])
.arg(&repo)
.status()?
.success()
);
assert!(
std::process::Command::new("git")
.args([
"-c",
"user.name=Test",
"-c",
"user.email=test@example.com",
"commit",
"--no-gpg-sign",
"--allow-empty",
"-m",
"empty"
])
.current_dir(&repo)
.status()?
.success()
);
let exec: Arc<dyn Exec> = Arc::new(Author(sausage::process::exec(World {
root: world.root.clone(),
host_dir: world.host_dir.clone(),
processes: world.processes.clone(),
})?));
sausage::harness::run_module(world, exec, BUNDLE).await?;
let good = std::fs::read_to_string(repo.join("good.txt"))?;
let world = World::open(dir.path())?;
let exec: Arc<dyn Exec> = Arc::new(Author(sausage::process::exec(World {
root: world.root.clone(),
host_dir: world.host_dir.clone(),
processes: world.processes.clone(),
})?));
sausage::harness::run_module(
world,
exec,
&format!("globalThis.extensionRestart=true;globalThis.extensionGood={good:?};\n{BUNDLE}"),
)
.await?;
Ok(())
}
+4 -1
View File
@@ -24,7 +24,10 @@ fn git(cwd: &Path, args: &[&str]) {
async fn several_mb_git_push() {
static BUILT: OnceLock<std::path::PathBuf> = OnceLock::new();
let component = BUILT.get_or_init(|| {
let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("hook-git");
let target = std::path::PathBuf::from(
std::env::var_os("CARGO_TARGET_DIR")
.expect("CARGO_TARGET_DIR must name the executor's shared target directory"),
);
let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
if !userland.join("cwd-init/init.o").is_file() {
assert!(
+3 -2
View File
@@ -12,6 +12,7 @@ fn write(path: PathBuf, text: &str) -> std::io::Result<()> {
pub fn component() -> &'static Path {
static PATH: OnceLock<PathBuf> = OnceLock::new();
PATH.get_or_init(|| {
let target = PathBuf::from(std::env::var_os("CARGO_TARGET_DIR").expect("CARGO_TARGET_DIR must name the executor's shared target directory"));
let directory = Path::new(env!("CARGO_TARGET_TMPDIR")).join("hook-guest");
std::fs::create_dir_all(directory.join("src")).unwrap();
std::fs::create_dir_all(directory.join("wit/deps")).unwrap();
@@ -67,7 +68,7 @@ async fn main() -> anyhow::Result<()> {
Ok(())
}
"#).unwrap();
assert!(std::process::Command::new("cargo").args(["build", "--manifest-path"]).arg(directory.join("Cargo.toml")).args(["--target", "wasm32-wasip2", "--release", "-j", "2"]).env("CARGO_TARGET_DIR", directory.join("target")).status().unwrap().success(), "test guest build failed");
directory.join("target/wasm32-wasip2/release/hook-guest.wasm")
assert!(std::process::Command::new("cargo").args(["build", "--manifest-path"]).arg(directory.join("Cargo.toml")).args(["--target", "wasm32-wasip2", "--release", "-j", "2"]).env("CARGO_TARGET_DIR", &target).status().unwrap().success(), "test guest build failed");
target.join("wasm32-wasip2/release/hook-guest.wasm")
})
}
+2 -2
View File
@@ -39,14 +39,14 @@ def main [destination?: path] {
if ($snapshot | path exists) { rm --recursive --force $snapshot }
mkdir $snapshot
let archive = ($destination | path join source.tar)
^git -C $repo archive --output $archive HEAD host vfs userland Cargo.toml Cargo.lock flake.nix flake.lock README.md AGENTS.md .gitignore
^git -C $repo archive --output $archive HEAD host vfs userland extensions Cargo.toml Cargo.lock flake.nix flake.lock README.md AGENTS.md .gitignore
if $env.LAST_EXIT_CODE != 0 { error make {msg: 'source archive failed'} }
^tar -xf $archive -C $snapshot
if $env.LAST_EXIT_CODE != 0 { error make {msg: 'source extraction failed'} }
rm $archive
^git -C $snapshot init --quiet --initial-branch=main
if $env.LAST_EXIT_CODE != 0 { error make {msg: 'source git init failed'} }
^git -C $snapshot add host vfs userland Cargo.toml Cargo.lock flake.nix flake.lock README.md AGENTS.md .gitignore
^git -C $snapshot add host vfs userland extensions Cargo.toml Cargo.lock flake.nix flake.lock README.md AGENTS.md .gitignore
if $env.LAST_EXIT_CODE != 0 { error make {msg: 'source git add failed'} }
with-env {
GIT_AUTHOR_NAME: 'Sausage', GIT_AUTHOR_EMAIL: 'sausage@localhost',
+4
View File
@@ -11,5 +11,9 @@ path = "src/env.rs"
name = "timeout"
path = "src/timeout.rs"
[[bin]]
name = "extensions-revert"
path = "src/revert.rs"
[dependencies]
sausage-spawn = { path = "../spawn" }
+25
View File
@@ -0,0 +1,25 @@
//! Move the world's extension branch to its last successful commit.
fn run() -> std::io::Result<i32> {
let good = std::fs::read_to_string("/src/sausage/.git/sausage-last-good")?;
if good.trim().is_empty() {
return Err(std::io::Error::other("No successful extension commit"));
}
sausage_spawn::Command::new("git")
.args([
"-C",
"/src/sausage",
"update-ref",
"refs/heads/main",
good.trim(),
])
.status()
}
fn main() {
std::process::exit(match run() {
Ok(code) => code,
Err(error) => {
eprintln!("extensions-revert: {error}");
1
}
});
}