test(enterprise): observe CLI lease cutoff at process exit

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/);

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);