// 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', checkpointWhen: () => false, 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; models(): modelApi.Models; settings(): durable.JsonObject; metadata: typeof Metadata; fs: typeof fs; exec: typeof exec; git: typeof git; implement(api: {name:string;readonly service?:S}, value:NoInfer):void; need(api:{name:string;readonly service?:S}):S; want(api:{name:string;readonly service?:S}):S|undefined; resource(key:string, create:()=>T):T; onOpen(cb:()=>Promise):void; onClose(cb:()=>Promise):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 { 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(); const resources = new Map(); 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) => { 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(); 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);}}; }