Port durable events and scoped extension reactors

host/js/events-test.ts

@@ -0,0 +1,76 @@
+// Atomic events, durable cursors, feed copies, world facets, and crash recovery.
+import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
+import { Harness,createRegistry,defineDoc,ROOT_CONVERSATION_ID as ROOT } from '@earendil-works/pi-durable';
+import { createModels } from '@earendil-works/pi-ai/models';
+import { openStorage } from './sqlite';
+import { defineEvent,eventLog,EventFeed,EVENT_FEED_SIZE } from './events';
+import { loadExtensions,fs,git } from './extensions';
+const assert=(value:any,message:string)=>{if(!value)throw new Error(message);};
+const equal=(a:any,b:any)=>assert(JSON.stringify(a)===JSON.stringify(b),`Expected ${JSON.stringify(b)}, got ${JSON.stringify(a)}`);
+const until=async(test:()=>any)=>{const deadline=Date.now()+15000;while(!await test()){if(Date.now()>deadline)throw new Error('Event deadline');await new Promise(r=>setTimeout(r,0));}};
+const doc=(kind:string)=>defineDoc({kind,scope:'conversation',version:1,history:'latest',fork:'initial',checkpointWhen:()=>false,initial:()=>({shas:[] as string[]})});
+const Shipped=doc('test.shipped'),Seen=doc('test.seen');
+const Landed=defineEvent<{sha:string}>('test.landed'),Branched=defineEvent<{id:number}>('test.branched');
+const models=createModels();
+const open=async()=>{const storage=await openStorage(globalThis.storagePath);const registry=createRegistry();const harness=await Harness.open(storage,{models,registry},context);await harness.root(context);return {harness,storage,registry};};
+let {harness,storage,registry}=await open();
+let service=eventLog(()=>harness);
+const read=async(token:any)=>(await harness.snapshot(token,ROOT,context))?.shas??[];
+const land=async(sha:string,fail=false)=>harness.commit(async tx=>{(await tx.doc(Shipped,ROOT)).shas.push(sha);await service.emit(tx,Landed,{sha});if(fail)throw new Error('rolled back');},context);
+if(globalThis.eventsMode==='emit') {
+  await land('crash');await fs.write('/emitted','ready');setInterval(()=>{},1000);await new Promise(()=>{});
+} else if(globalThis.eventsMode==='consume') {
+  const reactor=service.reactor('probe/counter',[Landed],async(tx,event)=>{(await tx.doc(Seen,ROOT)).shas.push(event.data.sha);});
+  await reactor.start();await until(async()=>(await service.eventLog.reactors())[0].lag===0);await reactor.stop();equal(await read(Seen),['crash']);await harness.close(context);
+} else if(globalThis.eventsMode==='world') {
+  globalThis.eventHosts={};
+  const repo='/src/sausage';
+  await fs.mkdir(`${repo}/extensions`);
+  for(const name of ['emitter','receiver'])await fs.mkdir(`${repo}/extensions/${name}`);
+  await fs.write(`${repo}/extensions/emitter/server.ts`,"export default host=>{globalThis.eventHosts.emitter=host;return {name:'emitter'};}");
+  const receiver=(version:number)=>`import {defineDoc,ROOT_CONVERSATION_ID as ROOT} from '@earendil-works/pi-durable';
+import {defineEvent} from '@earendil-works/sausage-events';
+const Seen=defineDoc({kind:'test.seen',scope:'conversation',version:1,history:'latest',fork:'initial',checkpointWhen:()=>false,initial:()=>({shas:[]})});
+export default host=>{globalThis.eventHosts.receiver=host;globalThis.receiverVersion=${version};host.react('counter',[defineEvent('test.landed')],async(tx,event)=>{(await tx.doc(Seen,ROOT)).shas.push(event.data.sha);});return {name:'receiver'};};`;
+  await fs.write(`${repo}/extensions/receiver/server.ts`,receiver(1));await git(repo,'add','.');await git(repo,'commit','--no-gpg-sign','-m','events');
+  const loader=await loadExtensions(harness,registry,storage,models,context);
+  try {
+    await harness.commit(tx=>globalThis.eventHosts.emitter.emit(tx,Landed,{sha:'world-a'}),context);
+    await until(async()=>(await read(Seen)).length===1);
+    await fs.write(`${repo}/extensions/receiver/server.ts`,receiver(2));await git(repo,'add','.');await git(repo,'commit','--no-gpg-sign','-m','reload');
+    const sha=await git(repo,'rev-parse','HEAD');await until(()=>loader.state().good===sha);
+    await harness.commit(tx=>globalThis.eventHosts.emitter.emit(tx,Landed,{sha:'world-b'}),context);
+    await until(async()=>(await read(Seen)).length===2);equal(await read(Seen),['world-a','world-b']);assert(globalThis.receiverVersion===2,'Replaced receiver');
+    globalThis.eventHosts.emitter.notice('Notice');globalThis.eventHosts.emitter.notice('Notice');
+    await until(async()=>(await service.eventLog.recent()).some(e=>e.name==='sausage.notice'));
+    equal((await service.eventLog.recent()).filter(e=>e.name==='sausage.notice').length,1);
+  } finally {await loader.close();await harness.close(context);}
+} else {
+  let throwOn:string|undefined;
+  const handler=async(tx:any,event:any)=>{(await tx.doc(Seen,ROOT)).shas.push(event.data.sha);if(event.data.sha===throwOn)throw new Error('handler failed');};
+  await land('a1');try{await land('bad',true);throw new Error('Missing rollback');}catch(error){assert(String(error).includes('rolled back'),'Rollback error');}
+  await harness.commit(tx=>service.emit(tx,Branched,{id:7}),context);equal(await read(Shipped),['a1']);
+  let reactor=service.reactor('probe/counter',[Landed],handler);await reactor.start();await until(async()=>(await read(Seen)).length===1);
+  await land('b2');await until(async()=>(await read(Seen)).length===2);
+  throwOn='c3';await land('c3');await until(async()=>!!(await service.eventLog.reactors())[0].failure);equal(await read(Seen),['a1','b2']);
+  equal((await service.eventLog.reactors())[0].lag,1);throwOn=undefined;await land('d4');await until(async()=>(await read(Seen)).length===4);
+  assert(!(await service.eventLog.reactors())[0].failure,'Failure cleared');
+  await reactor.stop();await land('e5');equal((await service.eventLog.reactors())[0].running,false);
+  reactor=service.reactor('probe/counter',[Landed],handler);await reactor.start();await until(async()=>(await read(Seen)).length===5);
+  await reactor.stop();await harness.close(context);({harness,storage,registry}=await open());service=eventLog(()=>harness);
+  await land('f6');reactor=service.reactor('probe/counter',[Landed],handler);await reactor.start();await until(async()=>(await read(Seen)).length===6);
+  equal(await read(Seen),['a1','b2','c3','d4','e5','f6']);equal(await read(Shipped),await read(Seen));await reactor.stop();
+  const all=await service.eventLog.recent({limit:20});equal(all.map(e=>e.data),[{sha:'f6'},{sha:'e5'},{sha:'d4'},{sha:'c3'},{sha:'b2'},{id:7},{sha:'a1'}]);
+  const branch=all.find(e=>Branched.is(e))!;
+  equal([branch.name,branch.kind,branch.v],['test.branched','event.test.branched',1]);assert(typeof branch.at==='string','Event timestamp');
+  equal((await service.eventLog.recent({events:[Landed],since:branch.id,limit:2})).map(e=>e.data.sha),['f6','e5']);
+  equal((await service.eventLog.recent({before:branch.id,order:'ascending'})).map(e=>e.data),[{sha:'a1'}]);
+  (all[0].data as any).sha='changed';equal((await service.eventLog.recent({limit:1}))[0].data,{sha:'f6'});
+  const feed=await harness.snapshot(EventFeed,ROOT,context);equal(feed?.events.map(e=>e.id),all.map(e=>e.id).toReversed());
+  let nowCount=0;const now=service.reactor('probe/now',[Landed],()=>{nowCount++;},{from:'now'});await now.start();
+  await until(async()=>(await service.eventLog.reactors()).find(r=>r.name==='probe/now')!.cursor===all[0].id);
+  await land('g7');await until(()=>nowCount===1);await now.stop();
+  await harness.commit(async tx=>{for(let i=0;i<EVENT_FEED_SIZE;i++)await service.emit(tx,Branched,{id:i});},context);
+  equal((await harness.snapshot(EventFeed,ROOT,context))?.events.map(e=>e.data),Array.from({length:EVENT_FEED_SIZE},(_,id)=>({id})));
+  await harness.close(context);
+}

host/js/events.md

@@ -0,0 +1,55 @@
+# Durable events
+
+`events.ts` owns event tokens, the event log, the feed, and reactor cursors.
+World extensions import its shared tokens from `@earendil-works/sausage-events`.
+The loader supplies this synthetic module from its own bundle.
+
+## Transaction contract
+
+`host.emit(tx, token, data)` appends an event inside the caller's transaction.
+The event and the caller's document changes commit together.
+A failed transaction retains neither change.
+
+The log uses an ownerless conversation that runs no turns.
+Entry IDs provide a global order.
+`EventLog` records the log conversation and each reactor's cursor on the root conversation.
+
+`host.react(name, tokens, handler, {from})` registers a facet-scoped reactor.
+Its durable name is `<extension>/<name>`.
+Each handler receives its transaction and a detached event record.
+The handler's document changes and cursor advance commit together.
+Transactional changes occur exactly once.
+External effects can repeat after a crash.
+
+A failed handler rolls back its changes and cursor.
+The next matching event commit or reactor start retries the pending event.
+Reloads stop the old facet's reactors before the new facet starts.
+A replacement with the same name continues from its persisted cursor.
+
+A new reactor starts at the beginning by default.
+`from: 'now'` skips existing events only when the reactor lacks a cursor.
+
+## Read contract
+
+`host.eventLog.recent(options)` returns detached records, newest first by default.
+It supports token filters, a limit, and an ascending order.
+`since` and `before` are exclusive entry IDs.
+`host.eventLog.reactors()` returns cursors, lag, activity, and the latest handler failure.
+
+`EventFeed` stores the newest 100 events, oldest first.
+The existing document observation API can observe this feed.
+Event emission updates the feed inside the same transaction.
+
+Notices become `sausage.notice` events.
+The host records each distinct notice once per facet load.
+This task does not add Ultimate's build or supervisor startup events.
+
+## Tests
+
+`events-test.ts` covers atomic emission, handler rollback, cursor recovery, replacement, and harness reopen.
+It also covers filtered reads, detached copies, the bounded feed, and `from: 'now'`.
+World facets exchange events through their hosts and continue after a commit reload.
+`host/tests/events.rs` kills a child after emission, then verifies one reaction across two restarts.
+
+The world test builds `git.wasm` with the repository's linker setup.
+It uses a persistent cache in the executor's target folder.

host/js/events.ts

@@ -0,0 +1,145 @@
+// Durable event entries and reactor cursors share the handler's transaction.
+// Transactional effects run exactly once. External effects can repeat after a crash.
+import type { JsonValue } from '@earendil-works/chord';
+import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
+import { defineDoc, ROOT_CONVERSATION_ID as ROOT } from '@earendil-works/pi-durable';
+import type { Harness, Tx, EntryRecord, ConversationId, EntryId } from '@earendil-works/pi-durable';
+export type EventRecord<D extends JsonValue = JsonValue> = {id:number;name:string;kind:string;v:number;at:string;data:D};
+export interface EventToken<D extends JsonValue = JsonValue> {
+  readonly name:string; readonly kind:string; readonly version:number;
+  is(event:EventRecord|undefined):event is EventRecord<D>;
+}
+export type EventData<T> = T extends EventToken<infer D> ? D : never;
+export function defineEvent<D extends JsonValue>(name:string, options:{version?:number}={}):EventToken<D> {
+  const kind=`event.${name}`;
+  return {name,kind,version:options.version??1,is:(event):event is EventRecord<D>=>event?.kind===kind};
+}
+export const NoticeEvent=defineEvent<{extension:string;message:string;level:string}>('sausage.notice');
+export const EVENT_FEED='sausage.event-feed', EVENT_FEED_SIZE=100;
+export const EventLog=defineDoc<{conversation?:number;cursors:Record<string,number>}>({
+  kind:'sausage.events',version:1,scope:'conversation',history:'latest',fork:'initial',
+  checkpointWhen:()=>false,initial:()=>({cursors:{}}),
+});
+export const EventFeed=defineDoc<{events:EventRecord[]}>({
+  kind:EVENT_FEED,version:1,scope:'conversation',history:'latest',fork:'initial',
+  checkpointWhen:()=>false,initial:()=>({events:[]}),
+});
+export type Reactor<D extends JsonValue>=(tx:Tx,event:EventRecord<D>)=>void|Promise<void>;
+export type ReactOptions={from?:'start'|'now'};
+export type ReactorStatus={name:string;events:string[];cursor:number;lag:number;running:boolean;failure?:{error:string;at:string;entryId?:number}};
+export interface EventHost {
+  emit<D extends JsonValue>(tx:Tx,event:EventToken<D>,data:D):Promise<EventRecord<D>>;
+  react<T extends EventToken>(name:string,events:readonly T[],handler:Reactor<EventData<T>>,options?:ReactOptions):void;
+  eventLog:{recent<T extends EventToken=EventToken>(options?:{events?:readonly T[];since?:number;before?:number;order?:'ascending'|'descending';limit?:number}):Promise<EventRecord<EventData<T>>[]>;reactors():Promise<ReactorStatus[]>};
+}
+const record=(entry:EntryRecord):EventRecord=>structuredClone({id:entry.id,name:entry.kind.slice(6),kind:entry.kind,...entry.data as {v:number;at:string;data:JsonValue}});
+export async function emit<D extends JsonValue>(tx:Tx,event:EventToken<D>,data:D):Promise<EventRecord<D>> {
+  const index=await tx.doc(EventLog,ROOT);
+  if(index.conversation===undefined)index.conversation=(await tx.createConversation({ownership:{kind:'ownerless'}})).id;
+  const entry=await tx.appendEntry(index.conversation as ConversationId,{kind:event.kind,data:{v:event.version,at:new Date().toISOString(),data}});
+  const emitted=record(entry) as EventRecord<D>;
+  const feed=(await tx.doc(EventFeed,ROOT)).events;
+  feed.push(emitted);
+  if(feed.length>EVENT_FEED_SIZE)feed.splice(0,feed.length-EVENT_FEED_SIZE);
+  return emitted;
+}
+async function scan(tx:Tx,log:ConversationId,after:number,visit:(entry:EntryRecord)=>boolean) {
+  let cursor:Parameters<Tx['scanEntries']>[2];
+  do {
+    const page=await tx.scanEntries({conversationId:log,minEntryId:(after+1) as EntryId,order:'ascending'},256,cursor);
+    for(const entry of page.items)if(visit(entry))return;
+    cursor=page.next;
+  } while(cursor);
+}
+/** One service per loader. Names retain their cursors when facets change. */
+export function eventLog(harness:()=>Harness) {
+  type Active={kinds:Set<string>;handler:Reactor<JsonValue>;from:'start'|'now'};
+  const reactors=new Map<string,{active?:Active;kinds:Set<string>;chain:Promise<void>;failure?:ReactorStatus['failure']}>();
+  function drain(name:string) {
+    const state=reactors.get(name)!;
+    state.chain=state.chain.then(async()=>{
+      for(;;) {
+        const active=state.active;
+        if(!active)return;
+        let attempted:number|undefined;
+        try {
+          const handled=await harness().commit(async tx=>{
+            const index=await tx.doc(EventLog,ROOT);
+            if(index.cursors[name]===undefined && active.from==='now') {
+              const newest=index.conversation===undefined?undefined:(await tx.scanEntries({conversationId:index.conversation as ConversationId},1)).items[0];
+              index.cursors[name]=newest?.id??0;
+              return false;
+            }
+            if(index.conversation===undefined)return false;
+            const after=index.cursors[name]??0;
+            let pending:EntryRecord|undefined,last=after;
+            await scan(tx,index.conversation as ConversationId,after,entry=>{
+              last=entry.id;if(active.kinds.has(entry.kind))pending=entry;return pending!==undefined;
+            });
+            if(pending){attempted=pending.id;await active.handler(tx,record(pending));}
+            if(last!==after)index.cursors[name]=last;
+            return pending!==undefined;
+          },context);
+          state.failure=undefined;
+          if(!handled)return;
+        } catch(error) {
+          state.failure={error:String(error),at:new Date().toISOString(),entryId:attempted};
+          return;
+        }
+      }
+    });
+  }
+  return {
+    emit,
+    reactor<T extends EventToken>(name:string,events:readonly T[],handler:Reactor<EventData<T>>,options:ReactOptions={}) {
+      const active:Active={kinds:new Set(events.map(event=>event.kind)),handler:handler as Reactor<JsonValue>,from:options.from??'start'};
+      if(!reactors.has(name))reactors.set(name,{kinds:active.kinds,chain:Promise.resolve()});
+      const state=reactors.get(name)!;
+      let detach:(()=>void)|undefined;
+      const stop=()=>{detach?.();detach=undefined;if(state.active===active)state.active=undefined;return state.chain;};
+      return {
+        start:async()=>{
+          await stop();state.active=active;state.kinds=active.kinds;
+          const observation=harness().observeCommits({advance:publication=>{
+            if(publication.changes.some(change=>change.type==='entry' && active.kinds.has(change.value.kind)))drain(name);
+          },end:()=>{}});
+          detach=()=>observation.detach();drain(name);
+        },stop,
+      };
+    },
+    eventLog:{recent,reactors:statuses} satisfies EventHost['eventLog'],
+  };
+  async function recent<T extends EventToken=EventToken>(options:{events?:readonly T[];since?:number;before?:number;order?:'ascending'|'descending';limit?:number}={}) {
+    const log=(await harness().snapshot(EventLog,ROOT,context))?.conversation;
+    if(log===undefined)return [];
+    const kinds=options.events && new Set(options.events.map(event=>event.kind));
+    const limit=options.limit??20;
+    return harness().commit(async tx=>{
+      const found:EventRecord<EventData<T>>[]=[];
+      let cursor:Parameters<Tx['scanEntries']>[2];
+      do {
+        const page=await tx.scanEntries({conversationId:log as ConversationId,
+          ...(options.since===undefined?{}:{minEntryId:(options.since+1) as EntryId}),
+          ...(options.before===undefined?{}:{maxEntryId:(options.before-1) as EntryId}),order:options.order??'descending'},256,cursor);
+        for(const entry of page.items) {
+          if(!(kinds?kinds.has(entry.kind):entry.kind.startsWith('event.')))continue;
+          found.push(record(entry) as EventRecord<EventData<T>>);if(found.length>=limit)return found;
+        }
+        cursor=page.next;
+      } while(cursor);
+      return found;
+    },context);
+  }
+  async function statuses():Promise<ReactorStatus[]> {
+    const index=await harness().snapshot(EventLog,ROOT,context);
+    return harness().commit(async tx=>{
+      const list:ReactorStatus[]=[];
+      for(const [name,state] of reactors) {
+        const cursor=index?.cursors[name]??0;let lag=0;
+        if(index?.conversation!==undefined)await scan(tx,index.conversation as ConversationId,cursor,entry=>{if(state.kinds.has(entry.kind))lag++;return false;});
+        list.push({name,events:[...state.kinds].map(kind=>kind.slice(6)),cursor,lag,running:state.active!==undefined,...(state.failure?{failure:state.failure}:{})});
+      }
+      return list;
+    },context);
+  }
+}

host/js/extensions.ts

@@ -6,12 +6,13 @@ 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';
+import * as events from './events';
 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 = {
+export type Host = events.EventHost & {
   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;
@@ -20,7 +21,6 @@ export type Host = {
   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);
@@ -64,12 +64,13 @@ export async function git(cwd:string,...args:string[]) {
   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};
+  const modules = {'@earendil-works/sausage-events':events,'@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 eventService = events.eventLog(()=>harness);
   const repo = '/src/sausage';
   let providers = new Map<string,any>();
   const resources = new Map<string,any>();
@@ -85,7 +86,9 @@ export async function loadExtensions(harness:any, registry:any, storage:any, mod
   };
   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[] = [];
+    let opened=false;
+    const notices=new Set<string>();
+    const open:any[] = [async()=>{opened=true;}], close:any[] = [async()=>{opened=false;}];
     const handle = (api:any) => new Proxy({}, {get:(_,key)=> {
       if (typeof key === 'symbol' || key === 'then') return undefined;
       const value = providers.get(api.name);
@@ -102,7 +105,20 @@ export async function loadExtensions(harness:any, registry:any, storage:any, mod
       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`);},
+      emit:eventService.emit,eventLog:eventService.eventLog,
+      react:(key,tokens,handler,options)=>{
+        const reactor=eventService.reactor(`${name}/${key}`,tokens,handler,options);
+        open.push(reactor.start);close.push(reactor.stop);
+        if(opened)void reactor.start();
+      },
+      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`);
+        if(!notices.has(message)){
+          notices.add(message);
+          void harness.commit((tx:any)=>eventService.emit(tx,events.NoticeEvent,{extension:name,message,level:'warning'}),context)
+            .catch((error:any)=>ops.op_output(`Recording a notice failed: ${String(error)}\n`));
+        }
+      },
     };
     return {host,open,close};
   };

host/js/package.json

@@ -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" },
+  "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" },
   "dependencies": {
     "@earendil-works/pi-durable": "1.5.0",
     "@earendil-works/pi-ai": "1.5.0",

host/tests/events.rs

@@ -0,0 +1,152 @@
+//! Durable event transactions, crash recovery, and world extension reactors.
+use sausage::{harness::run_module, world::World};
+use std::{
+    path::Path,
+    process::Command,
+    time::{Duration, Instant},
+};
+const BUNDLE: &str = include_str!("../js/events-test-bundle.js");
+struct Author(std::sync::Arc<dyn sausage::exec::Exec>);
+#[async_trait::async_trait]
+impl sausage::exec::Exec for Author {
+    async fn exec(
+        &self,
+        command: sausage::exec::Command,
+        mut options: sausage::exec::Options,
+        output: &mut (dyn for<'a> FnMut(sausage::exec::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
+    }
+}
+async fn run(dir: &Path, mode: &str) -> anyhow::Result<()> {
+    let world = World::open(dir)?;
+    run_module(
+        world,
+        sausage::process::exec(World::open(dir)?)?,
+        &format!("globalThis.eventsMode={mode:?};\n{BUNDLE}"),
+    )
+    .await
+}
+#[tokio::test(flavor = "current_thread")]
+async fn transactions_and_feed() -> anyhow::Result<()> {
+    let dir = tempfile::tempdir()?;
+    run(dir.path(), "normal").await
+}
+#[tokio::test(flavor = "current_thread")]
+async fn restart_child() -> anyhow::Result<()> {
+    if let Ok(dir) = std::env::var("EVENTS_RESTART_DIR") {
+        run(Path::new(&dir), &std::env::var("EVENTS_MODE")?).await?;
+    }
+    Ok(())
+}
+#[test]
+fn sigkill_between_emit_and_reaction() -> anyhow::Result<()> {
+    let dir = tempfile::tempdir()?;
+    let mut child = Command::new(std::env::current_exe()?)
+        .args(["--exact", "restart_child", "--nocapture"])
+        .env("EVENTS_RESTART_DIR", dir.path())
+        .env("EVENTS_MODE", "emit")
+        .spawn()?;
+    let deadline = Instant::now() + Duration::from_secs(30);
+    while !dir.path().join("world/emitted").is_file() {
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("Event did not commit");
+        }
+        std::thread::yield_now();
+    }
+    child.kill()?;
+    child.wait()?;
+    for _ in 0..2 {
+        let mut child = Command::new(std::env::current_exe()?)
+            .args(["--exact", "restart_child", "--nocapture"])
+            .env("EVENTS_RESTART_DIR", dir.path())
+            .env("EVENTS_MODE", "consume")
+            .spawn()?;
+        let deadline = Instant::now() + Duration::from_secs(30);
+        loop {
+            if let Some(status) = child.try_wait()? {
+                assert!(status.success());
+                break;
+            }
+            if Instant::now() > deadline {
+                child.kill()?;
+                child.wait()?;
+                anyhow::bail!("Reaction did not finish");
+            }
+            std::thread::yield_now();
+        }
+    }
+    Ok(())
+}
+#[tokio::test(flavor = "current_thread")]
+async fn world_extensions() -> anyhow::Result<()> {
+    let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
+    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
+    if !userland.join("cwd-init/init.o").is_file() {
+        assert!(
+            Command::new("nu")
+                .args(["--no-config-file", "cwd-init/build.nu"])
+                .current_dir(&userland)
+                .status()?
+                .success()
+        );
+    }
+    assert!(
+        Command::new("cargo")
+            .args(["build", "--release", "-p", "git", "--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"))?;
+    let git = target.join("wasm32-wasip2/release/git.wasm");
+    std::fs::copy(git, disk.join("bin/git.wasm"))?;
+    let repo = disk.join("src/sausage");
+    std::fs::create_dir_all(&repo)?;
+    assert!(
+        Command::new("git")
+            .args(["init", "-b", "main"])
+            .arg(&repo)
+            .status()?
+            .success()
+    );
+    assert!(
+        Command::new("git")
+            .args(["commit", "--no-gpg-sign", "--allow-empty", "-m", "empty"])
+            .env("GIT_AUTHOR_NAME", "Test")
+            .env("GIT_AUTHOR_EMAIL", "test@example.com")
+            .env("GIT_COMMITTER_NAME", "Test")
+            .env("GIT_COMMITTER_EMAIL", "test@example.com")
+            .current_dir(&repo)
+            .status()?
+            .success()
+    );
+    let exec = std::sync::Arc::new(Author(sausage::process::exec(World {
+        root: world.root.clone(),
+        host_dir: world.host_dir.clone(),
+        processes: world.processes.clone(),
+    })?));
+    run_module(
+        world,
+        exec,
+        &format!("globalThis.eventsMode='world';\n{BUNDLE}"),
+    )
+    .await
+}