stack/packages/harness/tests/model-turn-publication.test.mjs

// #650 controlled producer failure through REAL runner process and real broker.
// Unconfined fault-injection tests; direct confined production marker proof is separate.
import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import { join } from 'node:path';
import { spawn } from 'node:child_process';
import { citationWorld } from '../../bus/tests/citation-helpers.mjs';
import { serve } from '../../bus/src/server.mjs';
const RUNNER=new URL('../src/runner.mjs',import.meta.url).pathname;
const ADAPTER=new URL('./fixtures/citation-adapter.mjs',import.meta.url).pathname;
const delay=ms=>new Promise(resolve=>setTimeout(resolve,ms));
async function setup(t,mode){
 const closers=[];let child,closed,server;
 t.after(async()=>{const errors=[];if(child){try{if(child.exitCode===null&&child.signalCode===null)child.kill('SIGTERM');let timer;try{await Promise.race([closed,new Promise(resolve=>{timer=setTimeout(()=>{child.kill('SIGKILL');resolve();},5000);})]);await closed;}finally{clearTimeout(timer);}}catch{errors.push('own-runner-close');}}try{await server?.close();}catch{errors.push('server-close');}for(const close of [...closers].reverse()){try{await close();}catch{errors.push('owned-world-close');}}if(errors.length)throw new AggregateError(errors.map(code=>Error(code)),'owned producer fixture cleanup failed');});
 const f=citationWorld({after:close=>closers.push(close)},{library:mode==='retry'}),runDir=join(f.root,'run'),bundle=join(runDir,'bundle');fs.mkdirSync(bundle,{recursive:true,mode:0o700});fs.mkdirSync(join(f.root,'ws'));
 fs.writeFileSync(f.settings,JSON.stringify({replyLimits:{requireCitation:true}}));const brokerSocket=join(f.root,'broker.sock');server=await serve({broker:f.broker,path:brokerSocket});
 fs.writeFileSync(join(runDir,'fixture-steps.json'),JSON.stringify([{text:'uncited'},{text:'still uncited'},{text:'unreachable'}]));
 const adapter=join(runDir,'adapter.sh');fs.writeFileSync(adapter,`exec '${process.execPath}' '${ADAPTER}'\n`);fs.writeFileSync(join(bundle,'prompt.md'),'# fixture\n');fs.writeFileSync(join(bundle,'tools.json'),'[]');fs.writeFileSync(join(bundle,'policy.json'),JSON.stringify({needs:[],credentials:[]}));
 fs.writeFileSync(join(runDir,'proxy-first-turn-enabled'),'enabled\n',{mode:0o600});
 fs.writeFileSync(join(runDir,'session.json'),JSON.stringify({business:'acme',instance:'coder',run:'coder-run',harness:'pi',provider:'fixture',model:'fixture',brokerSocket,workspace:join(f.root,'ws'),sessionDir:join(runDir,'session'),adapter,bundle:{prompt:join(bundle,'prompt.md'),tools:join(bundle,'tools.json'),policy:join(bundle,'policy.json')},toolSocket:join(runDir,'tools.sock'),turnMarker:join(runDir,'turn.done'),turnEnd:'marker',harnessEnv:{},pollInterval:20,turnTimeout:10,...(mode==='landing'?{landing:[{decision:'d1',task_ref:'vikunja:3/7'},{decision:'d2',task_ref:'vikunja:3/8'}]}:{})}));
 const trace=join(f.root,'owned-spawn-trace.json'),attempt=join(f.root,'owned-marker-attempts'),preload=join(f.root,'fault-preload.mjs'),targetAttempt=mode==='retry'?2:1;
 // Test-only preload: all actual spawn/kill/fs operations pass through. Intercept
 // only own adapter spawn, owned returned PID group and own temporary marker path.
 fs.writeFileSync(preload,`import fs from 'node:fs';import cp from 'node:child_process';import {syncBuiltinESMExports} from 'node:module';
 const runDir=${JSON.stringify(runDir)},adapter=${JSON.stringify(adapter)},trace=${JSON.stringify(trace)},attempt=${JSON.stringify(attempt)},target=${targetAttempt};
 const link=fs.linkSync,spawn=cp.spawn,kill=process.kill,write=fs.writeFileSync;let tries=0;const state={groups:[],signals:[],unownedSignals:0};const save=()=>write(trace,JSON.stringify(state));
 cp.spawn=function(command,args,...rest){const child=spawn.call(this,command,args,...rest);if(command==='/bin/sh'&&args?.[0]===adapter){state.groups.push(child.pid);save();}return child;};
 process.kill=function(pid,signal){if(pid<0){if(!state.groups.includes(-pid)){state.unownedSignals++;save();throw Error('unowned signal refused');}state.signals.push({pid,signal});save();}return kill.call(this,pid,signal);};
 fs.linkSync=function(from,to,...rest){if(typeof from==='string'&&from.startsWith(runDir+'/.proxy-first-turn-')&&to===runDir+'/proxy-first-turn'){write(attempt,String(++tries));if(tries===target)throw Error('fx650-marker-fault');}return link.call(this,from,to,...rest);};syncBuiltinESMExports();\n`);
 child=spawn(process.execPath,['--import',preload,RUNNER,runDir],{stdio:['pipe','ignore','pipe'],env:{PATH:process.env.PATH,HOME:process.env.HOME,TMPDIR:process.env.TMPDIR,MOSAIC_CONFIG:process.env.MOSAIC_CONFIG}});let log='';child.stderr.on('data',chunk=>{log+=chunk;});child.stdin.on('error',()=>{});closed=new Promise((resolve,reject)=>{child.once('error',reject);child.once('close',(code,signal)=>resolve({code,signal}));});child.stdin.end(JSON.stringify({cap:f.agent})+'\n');
 if(mode!=='landing'){const ready=Date.now()+5000;while(!fs.existsSync(join(runDir,'tools.sock'))&&Date.now()<ready)await delay(20);assert.equal(fs.existsSync(join(runDir,'tools.sock')),true,'real runner tool server ready');await f.call(f.human,'message.send',{to:'coder',body:'first fixture request',class:'REQUEST'});await f.call(f.human,'message.send',{to:'coder',body:'second fixture request',class:'REQUEST'});}
 return {...f,runDir,closed,log:()=>log,attempt,targetAttempt,trace};
}
for(const mode of ['landing','message','retry'])test(`650 ${mode} publication failure reaps owned adapter then releases and exits2 without later turn`,{timeout:60000},async t=>{
 const f=await setup(t,mode),deadline=Date.now()+15000;while(!f.log().includes('runner: first-turn marker publication failed')&&Date.now()<deadline)await delay(20);
 assert.equal(f.log().includes('runner: first-turn marker publication failed'),true,'controlled publication failure observed at real caller');
 const outcome=await f.closed;assert.equal(outcome.code,2);assert.equal(outcome.signal,null);assert.equal(f.log().includes('fx650-marker-fault'),false,'raw publication error never logged');assert.equal(fs.existsSync(join(f.runDir,'tools.sock')),false);
 assert.equal(f.store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op,'release','existing role release reached');assert.equal(Number(fs.readFileSync(f.attempt,'utf8')),f.targetAttempt,'no later timestamp attempt after failure');
 const traced=JSON.parse(fs.readFileSync(f.trace,'utf8'));assert.equal(traced.groups.length,f.targetAttempt,'no later adapter turn after publication failure');assert.ok(traced.signals.some(s=>s.signal==='SIGTERM'),'owned adapter TERM observed');assert.equal(traced.signals.every(s=>traced.groups.includes(-s.pid)),true,'no registry or unrelated PID signalled');assert.equal(traced.unownedSignals,0,'no unowned group signal attempted');
});