132 lines
12 KiB
TypeScript
132 lines
12 KiB
TypeScript
/** Scripted world-loaded coordination, steering, ownership, and crash recovery. */
|
|
import { BACKGROUND_CONTEXT as context, awaitWithContext } 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, defineExtension, defineTool, LiveDoc, InboxDoc } from '@earendil-works/pi-durable';
|
|
import { obj } from '@earendil-works/pi-schema';
|
|
import { loadExtensions, fs } from './extensions';
|
|
import { conversationChains } from './environment';
|
|
import { openStorage } from './sqlite';
|
|
const assert = (value:any,message:string)=>{if(!value)throw new Error(message);};
|
|
const until = async(test:()=>Promise<any>|any)=>{const deadline=Date.now()+15000;while(!await test()){if(Date.now()>deadline)throw new Error('Subagent 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'},cwd:'/src/sausage'}});
|
|
const gate=Promise.withResolvers<void>(), parentGate=Promise.withResolvers<void>(), holding=new Set<number>();
|
|
const held=defineExtension({name:'hold',tools:[defineTool({name:'hold',description:'Wait for test gate',parameters:obj({}),replay:'safe',execute:async(_args,api,ctx)=>{holding.add(api.conversationId);await awaitWithContext(api.conversationId===root.id?parentGate.promise:gate.promise,ctx);return {output:[{type:'text',text:'released'}]};}})]});
|
|
if(globalThis.subagentsMode==='kill') {
|
|
const install=registry.install.bind(registry);
|
|
registry.install=(extension:any)=>install(extension.name!=='subagents'?extension:{...extension,tasks:extension.tasks.map((task:any)=>task.definition.name!=='ultimate.subagent-reporter'?task:{definition:{...task.definition,phases:{...task.definition.phases,report:async(_task:any,_runtime:any,ctx:any)=>{await fs.write('/report-ready','yes');await awaitWithContext(new Promise(()=>{}),ctx);}}}})});
|
|
}
|
|
const route=(transcript:any):any=>{
|
|
faux.appendResponses([route]);
|
|
const last=transcript.messages.findLast((message:any)=>message.role!=='system');
|
|
const text=typeof last.content==='string'?last.content:last.content.filter((part:any)=>part.type==='text').map((part:any)=>part.text).join('');
|
|
if(last.role==='user'&&text.startsWith('CALL ')) return fauxAssistantMessage([fauxToolCall('subagent',JSON.parse(text.slice(5)))],{stopReason:'toolUse'});
|
|
if(last.role==='user'&&text==='HOLD') return fauxAssistantMessage([fauxToolCall('hold',{})],{stopReason:'toolUse'});
|
|
return fauxAssistantMessage(last.role==='user'&&text.startsWith('[subagent')?'Report received':`Child: ${text}`);
|
|
};
|
|
faux.setResponses([route]);
|
|
const loader=await loadExtensions(harness,registry,storage,models,context,[held]);
|
|
assert(!loader.state().error,`Extension load: ${loader.state().error}`);
|
|
const snapshot=await fs.read('/src/sausage/.git/sausage-last-good');
|
|
const {Subagents}=await import(`file:///run/sausage/extensions/${snapshot}/extensions/subagents/api.ts`);
|
|
const chain=conversationChains(harness,context);
|
|
const children=async()=>Object.values((await harness.snapshot(Subagents,root.id,context))?.agents??{}) as any[];
|
|
const reports=async()=>(await root.context(context)).entries.filter((entry:any)=>entry.kind==='subagents.report');
|
|
const start=async(args:any)=>root.submit({type:'input',content:`CALL ${JSON.stringify(args)}`},context);
|
|
const call=async(args:any)=>{await(await start(args)).wait(context);return (await root.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;};
|
|
const text=(message:any)=>message?.content.filter((part:any)=>part.type==='text').map((part:any)=>part.text).join('');
|
|
harness.resume();
|
|
try {
|
|
if(globalThis.subagentsMode==='kill') {
|
|
await call({action:'start',agent:'scout',task:'crash answer',background:true});
|
|
await until(async()=>{try{return await fs.read('/report-ready')}catch{return false}});
|
|
await new Promise(()=>{});
|
|
} else if(globalThis.subagentsMode==='resume') {
|
|
await until(async()=>(await reports()).length===1);
|
|
await harness.waitForIdle(context);
|
|
assert((await children()).length===1,'Restart must retain one child');
|
|
assert((await reports()).length===1,'Restart must report exactly once');
|
|
assert(JSON.stringify(await reports()).includes('crash answer'),'Restart lost answer');
|
|
} else {
|
|
const service=globalThis.subagentService;
|
|
assert((await service.residents(context))[0].name==='scout','Resident definitions');
|
|
const standing=await service.standing('scout',context);
|
|
assert(await service.standing('scout',context)===standing,'Standing refresh must reuse its conversation');
|
|
const standingRecord=await harness.commit((tx:any)=>tx.conversation(standing),context);
|
|
assert(!standingRecord.owner,'Standing must be ownerless');
|
|
const scoutFile=await fs.read('/src/sausage/agents/scout.md');
|
|
await fs.write('/src/sausage/agents/scout.md',scoutFile.replace('Scout instructions','Updated instructions'));
|
|
await service.standing('scout',context);
|
|
assert((await (await harness.conversation(standing,context))!.agent(context)).instructions==='Updated instructions','Standing refresh must reread instructions');
|
|
const answer=await call({action:'start',agent:'scout',task:'facts'});
|
|
assert(text(answer)==='Returned model: faux/faux-1\nChild: facts',`Foreground answer: ${JSON.stringify(answer)}`);
|
|
const [first]=await children();
|
|
assert(JSON.stringify(await chain(first.conversationId))===JSON.stringify([String(first.conversationId),String(root.id)]),'Anchor ownership chain');
|
|
const record=await harness.commit((tx:any)=>tx.conversation(first.conversationId),context);
|
|
const anchor=await harness.getTask(record!.owner!.taskId,context);
|
|
assert(anchor?.kind==='ultimate.subagent-anchor','Child owner must be anchor');
|
|
const child=(await harness.conversation(first.conversationId,context))!;
|
|
assert(!(await child.agent(context)).tools.some((tool:any)=>tool.name==='subagent'),'Tool denials');
|
|
assert((await root.agent(context)).tools.some((tool:any)=>tool.name==='subagent'),'Parent tools remain unchanged');
|
|
assert((await reports()).length===0,'Foreground receipt must suppress duplicate report');
|
|
assert(text(await call({action:'message',id:first.conversationId,text:'next'})).includes('Child: next'),'Foreground message');
|
|
assert(text(await call({action:'wait',id:first.conversationId})).includes('Child: next'),'Wait latest answer');
|
|
await (await child.submit({type:'input',content:'one way',entry:{kind:'board.message',data:{from:root.id,reply:false}}},context)).wait(context);
|
|
await until(async()=>(await reports()).length===1);await harness.waitForIdle(context);
|
|
assert(JSON.stringify(await reports()).includes('one way'),'One-way recover lost answer');
|
|
await call({action:'start',agent:'scout',task:'HOLD',background:true});
|
|
const second=(await children()).at(-1)!;
|
|
await until(()=>holding.has(second.conversationId));
|
|
const wait=await start({action:'wait',id:second.conversationId});
|
|
await until(async()=>(await harness.snapshot(LiveDoc,root.id,context))?.run);
|
|
await root.submit({type:'input',content:'change direction',whenBusy:'steer'},context);
|
|
await wait.wait(context);
|
|
const yielded=(await root.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;
|
|
assert(yielded.details?.yielded,'Wait must yield on steering');
|
|
assert((await harness.snapshot(LiveDoc,second.conversationId,context))?.run,'Steering must leave child alive');
|
|
await root.submit({type:'input',content:'HOLD'},context);
|
|
await until(()=>holding.has(root.id));
|
|
gate.resolve();
|
|
await until(async()=>(await harness.snapshot(InboxDoc,root.id,context))?.items.some((item:any)=>item.mode==='steer'&&item.entry?.kind==='subagents.report'));
|
|
parentGate.resolve();
|
|
await until(async()=>(await reports()).length===2);await harness.waitForIdle(context);
|
|
assert((await reports()).length===2,'Background answer must report exactly once');
|
|
const listed=JSON.parse(text(await call({action:'list'})));assert(listed.length===2,'List children');
|
|
// Use a fresh unresolved gate for the stop case through a child input that never resolves.
|
|
const stopGate=Promise.withResolvers<void>();
|
|
registry.install(defineExtension({name:'hold',tools:[defineTool({name:'hold',description:'Stop gate',parameters:obj({}),replay:'safe',execute:async(_a,api,ctx)=>{holding.add(api.conversationId);await awaitWithContext(stopGate.promise,ctx);return {output:[]};}})]}));
|
|
holding.clear();
|
|
await call({action:'start',agent:'scout',task:'HOLD',background:true});
|
|
const third=(await children()).at(-1)!;await until(()=>holding.has(third.conversationId));
|
|
assert((await harness.snapshot(LiveDoc,third.conversationId,context))?.run,'Stop fixture must hold an active child');
|
|
assert(text(await call({action:'stop',id:third.conversationId}))===`Stopped ${third.conversationId}.`,'Stop result');
|
|
assert(!(await harness.snapshot(LiveDoc,third.conversationId,context))?.run,'Stop must end child');
|
|
assert((await reports()).length===2,'Stop must suppress aborted report');
|
|
assert(text(await call({action:'start',agent:'scout',task:'no',isolate:'worktree'})).includes('requires the workspaces extension'),'Worktree seam');
|
|
await fs.write('/src/sausage/agents/worker.md','---\nname: worker\ndescription: Worker\nfamily: fast\ntools: []\n---\nWorker body');
|
|
await fs.mkdir('/src/sausage/.sausage');await fs.mkdir('/src/sausage/.sausage/agents');
|
|
await fs.write('/src/sausage/.sausage/agents/worker.md','---\nname: worker\naddTools: [hold]\n---\nOverlay body');
|
|
await fs.write('/src/sausage/agents/models.json',JSON.stringify({default:{model:'faux/faux-1'},fast:{model:'faux/faux-1',thinking:'low'}}));
|
|
await call({action:'start',agent:'worker',task:'family one'});
|
|
const familyChild=(await children()).at(-1)!;
|
|
const firstAgent=await (await harness.conversation(familyChild.conversationId,context))!.agent(context);
|
|
assert(firstAgent.model?.modelId==='faux-1'&&firstAgent.thinkingLevel==='low','Worker must resolve fast family');
|
|
assert(firstAgent.tools.map((tool:any)=>tool.name).join(',')==='hold','Project overlay must add a tool');
|
|
assert(firstAgent.instructions.includes('Worker body')&&firstAgent.instructions.includes('Overlay body'),'Child must use overlaid instructions');
|
|
const provider=models.getProvider('faux')!, template=models.getModels('faux')[0];
|
|
models.setProvider({...provider,getModels:()=>[template,{...template,id:'second'}]});
|
|
await fs.write('/src/sausage/agents/models.json',JSON.stringify({default:{model:'faux/faux-1'},fast:{model:'faux/second',thinking:'high'}}));
|
|
await call({action:'start',agent:'worker',task:'family two'});
|
|
const secondFamilyChild=(await children()).at(-1)!;
|
|
const secondAgent=await (await harness.conversation(secondFamilyChild.conversationId,context))!.agent(context);
|
|
assert(secondAgent.model?.modelId==='second'&&secondAgent.thinkingLevel==='high','Next child must use edited family without restart');
|
|
globalThis.childPin={provider:'faux',modelId:'faux-1'};
|
|
await call({action:'start',agent:'worker',task:'pin'});
|
|
const pinnedChild=(await children()).at(-1)!;
|
|
assert((await (await harness.conversation(pinnedChild.conversationId,context))!.agent(context)).model?.modelId==='faux-1','models.pin must apply to subagents');
|
|
}
|
|
} finally {gate.resolve();parentGate.resolve();await loader.close();await harness.close(context);}
|