diff --git a/enterprise/plugin-agent-bridge/src/server/__tests__/cli-ipc-flow.integration.mjs b/enterprise/plugin-agent-bridge/src/server/__tests__/cli-ipc-flow.integration.mjs index c70baf4fd1..1a5cc43751 100644 --- a/enterprise/plugin-agent-bridge/src/server/__tests__/cli-ipc-flow.integration.mjs +++ b/enterprise/plugin-agent-bridge/src/server/__tests__/cli-ipc-flow.integration.mjs @@ -18,7 +18,7 @@ const policy={provider:'afos-managed',model:'fixed-model',maxTokens:64}; const reply=(name,args)=>[{type:'block-start',index:0,blockType:'tool-call'},{type:'tool-call-delta',index:0,id:`call-${name}`,name,argumentsDelta:JSON.stringify(args)},{type:'block-end',index:0,block:{type:'tool-call',id:`call-${name}`,name,arguments:JSON.stringify(args)}},{type:'finish',reason:{kind:'tool-calls'}}]; const done=[{type:'block-start',index:0,blockType:'text'},{type:'text-delta',index:0,text:'Draft ready.'},{type:'block-end',index:0,block:{type:'text',text:'Draft ready.'}},{type:'finish',reason:{kind:'stop'}}]; async function composed(t,{core=false,failAck=false,dropAck=false,wake=false,model}={}){ - let child,channel,next,modelCalls=0;const wire=[]; + let child,channel,next,childClosed,modelCalls=0;const wire=[],observations=[];const leases={prepared:0,committed:0,disposed:0}; const stop=async()=>{ await channel?.close();channel=undefined; if(!child)return;const prior=child;child=undefined; @@ -43,13 +43,15 @@ async function composed(t,{core=false,failAck=false,dropAck=false,wake=false,mod await writeFile(join(profile,'cordis.patch.yml'),JSON.stringify([{insert:rows}])+'\n',{mode:0o600,flag:'wx'}); await symlink(join(providerRoot,'node_modules'),join(profile,'node_modules'),'dir'); child=spawn(process.execPath,[cli,'--profile','afos-managed'],{cwd:home,env:{...runtimeEnvironment(process.env),DSH_HOME:home,DSH_AGENTS_HOME:join(home,'agents'),DSH_TELEMETRY_DISABLED:'1'},stdio:['pipe','pipe','pipe','ipc'],serialization:'json',windowsHide:true}); - const owned=child;let stdout='',stderr='';owned.stdout.on('data',v=>stdout+=v);owned.stderr.on('data',v=>stderr+=v); + const owned=child;childClosed=once(owned,'close');childClosed.catch(()=>undefined);let stdout='',stderr='';owned.stdout.on('data',v=>stdout+=v);owned.stderr.on('data',v=>stderr+=v); const received=[],waiting=[],lines=createInterface({input:owned.stdout});let readyResolve,readyReject; const appReady=new Promise((resolve,reject)=>{readyResolve=resolve;readyReject=reject;}); - lines.on('line',line=>{if(line==='AFOS_CLI_READY v1')readyResolve();if(!line.startsWith('AFOS_RESULT '))return;const value=JSON.parse(line.slice(12));if(waiting.length)waiting.shift().resolve(value);else received.push(value);}); + lines.on('line',line=>{if(line==='AFOS_CLI_READY v1')readyResolve(); + if(line.startsWith('AFOS_OBSERVATION ')){const value=JSON.parse(line.slice(17));assert.deepEqual(Object.keys(value),['event','created','disposed','wakes','toolCalls','agents']);assert.ok(['ready','agent/created','agent/disposed','creation-wake','tool/call','process/exit'].includes(value.event));for(const name of ['created','disposed','wakes','toolCalls','agents'])assert.ok(Number.isSafeInteger(value[name])&&value[name]>=0&&value[name]<=100000);observations.push(value);console.log('AFOS_CLI_OBSERVATION '+JSON.stringify(value));return;} + if(!line.startsWith('AFOS_RESULT '))return;const value=JSON.parse(line.slice(12));if(waiting.length)waiting.shift().resolve(value);else received.push(value);}); const refuse=()=>{readyReject(Error('original-cli-not-ready'));for(const entry of waiting.splice(0))entry.reject(Error(`original CLI exited: ${stdout}\n${stderr}`));};owned.once('error',refuse);owned.once('exit',refuse);appReady.catch(()=>undefined); next=()=>received.length?Promise.resolve(received.shift()):new Promise((resolve,reject)=>waiting.push({resolve,reject})); - const real=failAck?{...business,prepareSessionBinding:async(...args)=>{const lease=await business.prepareSessionBinding(...args);return{commit(){throw Error('fixture-private-ack-error');},dispose:lease.dispose};}}:business; + const real={...business,prepareSessionBinding:async(...args)=>{const lease=await business.prepareSessionBinding(...args);leases.prepared++;return{commit(){if(failAck)throw Error('fixture-private-ack-error');lease.commit();leases.committed++;},dispose(){lease.dispose();leases.disposed++;}};}}; const endpoint={send:message=>{wire.push(message);if(dropAck&&message.kind==='reply'&&message.method==='publication.ack'&&message.ok)return Promise.resolve();return new Promise((resolve,reject)=>owned.send(message,error=>error?reject(error):resolve()));},messages:listener=>{const accept=value=>{wire.push(value);listener(value);};owned.on('message',accept);return()=>owned.off('message',accept);},disconnected:listener=>{owned.on('disconnect',listener);return()=>owned.off('disconnect',listener);}}; channel=await attachMainBusinessIpc(endpoint,real,signal,{root:layout.profile,model:policy}); const value=await next();assert.equal(value.ready,true,JSON.stringify(value));assert.equal(value.originalLoader,true,JSON.stringify(value));assert.equal(value.rootPrivate,true,JSON.stringify(value));assert.equal(value.identity,_authority.binding.subject);assert.notEqual(value.pid,process.pid);assert.equal(value.ppid,process.pid);assert.equal(value.materials[0].materialId,'m1'); @@ -57,7 +59,7 @@ async function composed(t,{core=false,failAck=false,dropAck=false,wake=false,mod console.log('AFOS_CLI_EVIDENCE '+JSON.stringify({pid:value.pid,ppid:value.ppid,entryOwner:value.entryOwner,rootPrivate:value.rootPrivate,core:value.core??false})); };f.startHook(bootHost); await f.main.startHost();const command=(op,extra={})=>{child.stdin.write(JSON.stringify({op,...extra})+'\n');return next();}; - return{...f,wire,child:()=>child,channel:()=>channel,stop,bootHost,command,modelCalls:()=>modelCalls, + return{...f,wire,child:()=>child,channel:()=>channel,stop,bootHost,command,modelCalls:()=>modelCalls,observations:()=>observations.slice(),leases:()=>({...leases}),childClosed:()=>childClosed, invoke:async(method,payload)=>{const result=await command('invoke',{method,payload});assert.equal(result.error,undefined,JSON.stringify(result));return result.result;}}; } test('Main → original official CLI → fixed private provider → awaited Identity → Main material read',{timeout:15000},async t=>{ @@ -78,8 +80,16 @@ test('original CLI failed publication ACK rejects creation-listener wake before const stats=await f.command('stats');assert.equal(stats.result.wakes,1);assert.equal(stats.result.toolCalls,0);assert.equal(stats.result.agents,0);assert.equal(f.business().drafts().length,0);assert.ok(!JSON.stringify(f.wire).includes('fixture-private-ack-error')); }); test('original CLI lost publication ACK remains unknown and releases reservation with zero tools/model',{timeout:20000},async t=>{ - const f=await composed(t,{core:true,dropAck:true,wake:true});const started=await f.invoke('start',{sources:[f.ref],goal:'Lost Main ACK remains unknown.',operationId:'cli-lost-ack'});assert.equal(started.ok,false);assert.equal(f.modelCalls(),0); - const stats=await f.command('stats');assert.equal(stats.result.toolCalls,0);assert.equal(stats.result.agents,0);assert.equal(f.business().drafts().length,0);assert.ok(f.wire.some(v=>v.kind==='request'&&v.method==='publication.release')); + const f=await composed(t,{core:true,dropAck:true,wake:true});const owned=f.child(),closed=f.childClosed();const started=await f.invoke('start',{sources:[f.ref],goal:'Lost Main ACK remains unknown.',operationId:'cli-lost-ack'});assert.equal(started.ok,false);assert.equal(started.error.code,'afos/profile-cancelled');assert.equal(f.modelCalls(),0); + // Main's fresh-authority cutoff (at most 5s) disposes the reservation and the + // original CLI exits. Read captured events and the original public registry + // count recorded synchronously at Node exit, never postmortem services. + const [code,signal]=await closed;assert.equal(code,1);assert.equal(signal,null);assert.equal(owned.exitCode,1); + const observations=f.observations(),last=observations.at(-1);assert.ok(last,JSON.stringify(observations));assert.equal(last.wakes,1);assert.equal(last.toolCalls,0);assert.equal(last.created,1);assert.equal(last.event,'process/exit',JSON.stringify(observations));assert.ok(last.disposed<=last.created);assert.equal(last.agents,0);assert.equal(f.business().drafts().length,0); + assert.ok(f.wire.some(v=>v.kind==='cutoff'));assert.equal(f.wire.filter(v=>v.kind==='request'&&v.method==='publication.renew').length,0);assert.deepEqual(f.leases(),{prepared:1,committed:1,disposed:1}); + const ack=f.wire.find(v=>v.kind==='reply'&&v.method==='publication.ack'&&v.ok);assert.ok(ack);assert.equal(f.wire.filter(v=>v.kind==='request'&&v.method==='publication.ack').length,1); + await assert.rejects(f.business().sessionRun(ack.value.sessionId,new AbortController().signal));const runId=[...f.tables.runs.keys()][0];const admitted=await f.business().authorize([f.ref],'cli-after-lost-ack',new AbortController().signal);assert.notEqual(admitted.runId,runId);f.business().releaseRunAdmission(admitted.runId);assert.equal(f.tables.artifacts.size,0); + console.log('AFOS_LOST_ACK_CUTOFF '+JSON.stringify({code,observations:last,leases:f.leases(),ackDelivered:false,oldSessionRejected:true,freshAdmissionReleased:true,modelCalls:f.modelCalls()})); }); test('original CLI same-account restart and account switch cut off prior generations',{timeout:20000},async t=>{ const f=await composed(t,{core:true});const prior=f.business(),old=f.child();await f.main.startHost();assert.equal(prior.hostSignal.aborted,true);assert.ok(old.exitCode!==null||old.signalCode!==null);assert.throws(()=>prior.drafts(),/host-changed/); diff --git a/workdsh-web/packages/providers/enterprise-profile/tests/fixtures/cli-core-consumer.mjs b/workdsh-web/packages/providers/enterprise-profile/tests/fixtures/cli-core-consumer.mjs index 54db86bf7c..8da2c79e29 100644 --- a/workdsh-web/packages/providers/enterprise-profile/tests/fixtures/cli-core-consumer.mjs +++ b/workdsh-web/packages/providers/enterprise-profile/tests/fixtures/cli-core-consumer.mjs @@ -1,15 +1,29 @@ import {createInterface} from 'node:readline'; +import {writeSync} from 'node:fs'; import {createUserMessage} from '@deepseek-ai/dsh-llm'; export const name='afos-cli-core-fixture'; export const inject=['afosCliCoreReady','workdshIdentity','afosDraftRuntime','agents','tools']; export async function apply(ctx,config={}){ const ready=ctx.afosCliCoreReady;ready.signal.throwIfAborted(); - let wakes=0,toolCalls=0; + let wakes=0,toolCalls=0,created=0,disposed=0; + // Test-only root-lifetime public event counts survive private fixture disposal. + const observe=event=>process.stdout.write('AFOS_OBSERVATION '+JSON.stringify({event,created,disposed,wakes,toolCalls,agents:created-disposed})+'\n'); + ctx.root.on('agent/created',()=>{created++;observe('agent/created');},{global:true}); + ctx.root.on('agent/disposed',()=>{disposed++;observe('agent/disposed');},{global:true}); + observe('ready'); + // Original CLI shutdown removes ctx.on listeners before Agent teardown settles. + // Capture only the original public registry, then read list() synchronously at + // Node exit; do not query disposed context services or infer a disposal event. + const observedAgents=ctx.agents; + process.once('exit',()=>{ + try{const agents=observedAgents.list().length;writeSync(1,'AFOS_OBSERVATION '+JSON.stringify({event:'process/exit',created,disposed,wakes,toolCalls,agents})+'\n');} + catch{writeSync(1,'AFOS_OBSERVATION_UNAVAILABLE public-agent-list\n');} + }); if(config.wakeBeforeAck)ctx.on('agent/created',async({agent})=>{ - wakes++;agent.followup(createUserMessage({source:{kind:'user'},content:[{type:'text',text:'Creation wake must await Main ACK.'}]})); + wakes++;observe('creation-wake');agent.followup(createUserMessage({source:{kind:'user'},content:[{type:'text',text:'Creation wake must await Main ACK.'}]})); await new Promise(resolve=>setTimeout(resolve,20)); },{global:true}); - ctx.on('session/event',(_session,event)=>{if(event.type==='tool/call')toolCalls++;}); + ctx.root.on('session/event',(_session,event)=>{if(event.type==='tool/call'){toolCalls++;observe('tool/call');}},{global:true}); const output=value=>process.stdout.write('AFOS_RESULT '+JSON.stringify(value)+'\n'); const materials=await ready.business.materials(ready.signal);ready.signal.throwIfAborted(); const rootLoader=ctx.root.get('loader'),owner=rootLoader?.locate(ctx.fiber);