127 lines
12 KiB
JavaScript
127 lines
12 KiB
JavaScript
import { requestHostOps } from './request-host-ops.mjs';
|
|
// #653 isolated owned fixture: real launcher/runtime/runner and fake loopback management HTTP.
|
|
import fs from 'node:fs';
|
|
import http from 'node:http';
|
|
import assert from 'node:assert/strict';
|
|
import { createHash, randomBytes } from 'node:crypto';
|
|
import { spawnSync } from 'node:child_process';
|
|
import { join } from 'node:path';
|
|
import { tmpdir } from 'node:os';
|
|
import { fixture, fakeHarnesses } from './helpers.mjs';
|
|
import { loadSystem, bootConfig } from '../src/config.mjs';
|
|
import { createLauncher } from '../src/launcher.mjs';
|
|
import { startBroker } from '../../bus/src/runtime.mjs';
|
|
import { gitEnv } from '../../seat/src/checkout.mjs';
|
|
import { readSessions, runsDir, sessionsFile, SESSIONS_VERSION, launchLogFile } from '../../seat/src/session.mjs';
|
|
export const waitFor = async (read, message, ms=15000) => { const end=Date.now()+ms;while(Date.now()<end){const value=await read();if(value)return value;await new Promise(r=>setTimeout(r,20));}throw Error(message); };
|
|
export const pause = ms => new Promise(r=>setTimeout(r,ms));
|
|
const id = value => createHash('sha256').update(value).digest('hex').slice(0,12);
|
|
// #774: bwrap overrides the launcher's sandbox binary; admit rewrites what P2 admits, so a test can
|
|
// reach the final check's key match. trace.ops names every trusted host operation in order.
|
|
// #774 R2: extra names more harnesses to switch on and load (a fallback route's); real names the
|
|
// harnesses loaded as the real module, so their own preflight runs; hostEnv adds launcher env
|
|
// such as MOSAIC_CLAUDE_BIN.
|
|
export async function managedLaunchFixture(t,{proxy=true,launches=true,refuseProbe=false,harness='pi',bwrap=null,admit=null,extra=[],real=[],hostEnv={},modelStream=null,modelCatalogue=null}={}){
|
|
const root=fs.mkdtempSync(join(tmpdir(),'ml653-')),closers=[],history=new Set(['fx653-manage']),proxyReceipts=[],logs=[];
|
|
let server,bus,launcher,stderr;const trace={preflightAttempts:0,successfulPreflights:0,enableBeforeSpawn:[],bindings:[],stalls:[],ops:[]};
|
|
const checkHistory=entries=>{const text=JSON.stringify(entries);for(const value of history)if(text.includes(value))throw Error('fixture credential leaked');};
|
|
t.after(async()=>{const failures=[];for(const close of [...closers].reverse()){try{await close();}catch{failures.push('owned-close');}}try{server?.closeAllConnections();if(server?.listening)await new Promise((resolve,reject)=>server.close(e=>e?reject(e):resolve()));}catch{failures.push('owned-server-close');}try{checkHistory({logs,proxyReceipts,trace});}catch{failures.push('value-free-receipts');}if(stderr)process.stderr.write=stderr;try{fs.rmSync(root,{recursive:true,force:true});}catch{failures.push('owned-root-close');}if(failures.length)throw new AggregateError(failures.map(code=>Error(code)),'owned managed fixture cleanup failed');});
|
|
const f=fixture(root,'acme',doc=>{doc.plugins={...doc.plugins,harness:[harness,...extra]};for(const role of Object.values(doc.roles))role.vars={...role.vars,harness};doc.roles.chief.vars.model='claude-opus-5-5';if(modelStream){doc.launch.models['claude-opus-5-5']={provider:'anthropic',model:'claude-opus-5-5'};}return doc;});
|
|
const env={PATH:process.env.PATH,HOME:process.env.HOME,TMPDIR:process.env.TMPDIR,MOSAIC_CONFIG:f.config,MOSAIC_ROLES_DIR:f.roles,...hostEnv};
|
|
// The nonproxy positive fixture supplies its own fake pi login. Managed
|
|
// proxy fixtures deliberately hold no upstream credential at all.
|
|
if(!proxy)env.ZAI_API_KEY='fx653-pi';
|
|
const configDir=join(root,'config'),prefix=`msk-${id(configDir)}-`,sentinel=prefix+'sentinel-'+randomBytes(32).toString('base64url');history.add(sentinel);
|
|
const state={keys:[sentinel],calls:[],refuseProbe};
|
|
server=http.createServer(async(req,res)=>{try{
|
|
const supplied=req.headers.authorization?.slice(7) ?? req.headers['x-api-key'];if(supplied)history.add(supplied);for(const key of state.keys)history.add(key);
|
|
if(modelStream && req.method === 'POST' && /\/messages(?:\?|$)/.test(req.url)) {
|
|
if (!state.keys.includes(supplied)) { res.writeHead(401);res.end();return; }
|
|
req.resume(); req.on('end', () => {
|
|
state.modelRequests = (state.modelRequests ?? 0) + 1;
|
|
res.writeHead(200, {'content-type':'text/event-stream','cache-control':'no-cache'});res.flushHeaders();
|
|
const send=(type,data)=>res.write(`event: ${type}\ndata: ${JSON.stringify({type,...data})}\n\n`);
|
|
let ticker;
|
|
if(modelStream.mode === 'stream') {
|
|
send('message_start',{message:{id:'msg_test',type:'message',role:'assistant',model:'claude-opus-5-5',content:[],stop_reason:null,stop_sequence:null,usage:{input_tokens:5,output_tokens:1,cache_creation_input_tokens:0,cache_read_input_tokens:0}}});
|
|
send('content_block_start',{index:0,content_block:{type:'text',text:''}});
|
|
send('content_block_delta',{index:0,delta:{type:'text_delta',text:'first'}});
|
|
ticker=setInterval(()=>send('content_block_delta',{index:0,delta:{type:'text_delta',text:' more'}}),400);
|
|
const end=setTimeout(()=>{clearInterval(ticker);send('content_block_stop',{index:0});send('message_delta',{delta:{stop_reason:'end_turn',stop_sequence:null},usage:{output_tokens:5}});send('message_stop',{});res.end();},modelStream.duration);
|
|
res.on('close',()=>clearTimeout(end));
|
|
} else if(modelStream.mode === 'keepalive') ticker=setInterval(()=>{res.write(': keepalive\n\n');send('ping',{});},300);
|
|
res.on('close',()=>clearInterval(ticker));
|
|
});return;
|
|
}
|
|
if(req.url==='/v8/management/config/access/api-keys'){
|
|
state.calls.push(req.method);if(supplied!=='fx653-manage'){res.writeHead(401);res.end('{}');return;}
|
|
if(req.method==='GET'){res.end(JSON.stringify(state.keys));return;}
|
|
const parts=[];for await(const part of req)parts.push(part);const values=JSON.parse(Buffer.concat(parts));for(const value of values)history.add(value);state.keys=values;res.end('{"status":"ok"}');return;
|
|
}
|
|
state.calls.push('PROBE');res.writeHead(!state.refuseProbe&&state.keys.includes(supplied)?200:401);res.end('{}');
|
|
}catch{if(!res.destroyed){res.writeHead(500);res.end('{}');}}});
|
|
await new Promise((resolve,reject)=>{server.once('error',reject);server.listen(0,'127.0.0.1',resolve);});const port=server.address().port;assert.notEqual(port,8317);
|
|
fs.mkdirSync(join(configDir,'secrets'),{recursive:true,mode:0o700});fs.writeFileSync(join(configDir,'secrets','proxy-management.key'),'fx653-manage',{mode:0o600});
|
|
fs.mkdirSync(join(f.dataRoot,'bus-host'),{recursive:true,mode:0o700});fs.writeFileSync(sessionsFile(f.dataRoot),JSON.stringify({sessionsVersion:SESSIONS_VERSION,sessions:[]}),{mode:0o600});
|
|
const git=(...args)=>{const result=spawnSync('git',['-c','user.name=test','-c','[email protected]','-C',f.doc.projects.stack.root,...args],{env:gitEnv(env),encoding:'utf8'});assert.equal(result.status,0,'owned local git fixture succeeds');};
|
|
git('init','-q','-b','main');fs.writeFileSync(join(f.doc.projects.stack.root,'file.txt'),'one\n');git('add','file.txt');git('commit','-q','-m','one');
|
|
const adapter=join(root,'adapter.sh');fs.writeFileSync(adapter,'case "$MOSAIC_REQUEST" in *SILENT*) exec /usr/bin/sleep 30;; *) printf %s fx653-private-output; printf "done\\n" > "$MOSAIC_TURN_MARKER";; esac\n',{mode:0o644});
|
|
const system={...loadSystem({env}),...(proxy?{proxy:{port,binary:'/fixture-unused',firstOutputSeconds:modelStream?.seconds ?? 1}}:{})};
|
|
const {trackers,...boot}=bootConfig({system,businessId:'acme',env});assert.equal(trackers,undefined);
|
|
stderr=process.stderr.write;process.stderr.write=function(chunk,...args){if(String(chunk).startsWith('model-proxy: ')){proxyReceipts.push(String(chunk));const callback=args.find(a=>typeof a==='function');callback?.();return true;}return stderr.call(this,chunk,...args);};
|
|
bus=await startBroker({...boot,modelCatalogue});closers.push(()=>bus.close());
|
|
const hooks=[];let expectedKeyId;
|
|
const requestOps=requestHostOps(bus);
|
|
const actualOp=async m=>{
|
|
if(Object.hasOwn(requestOps,m.op))return requestOps[m.op](m);
|
|
switch(m.op){
|
|
case 'modelPreflight':return bus.modelPreflight(m);
|
|
case 'modelAdmission':return bus.modelAdmission(m);
|
|
case 'modelStall':return bus.modelStall(m);
|
|
case 'identity':return bus.identity(m.cap);
|
|
case 'authorizeLaunch':return bus.authorizeLaunch(m.cap,m.instance);
|
|
case 'refuse':return bus.refuse(m.cap,m.code,m.subject);
|
|
case 'endLaunch':return bus.endLaunch(m.record);
|
|
case 'recoverLaunch':return bus.recoverLaunch(m.record);
|
|
case 'credentialStatus':return bus.credentialStatus(m.business,m.role);
|
|
case 'mayLaunch':return bus.mayLaunch(m);
|
|
case 'landHolders':return bus.landHolders(m.business);
|
|
case 'landHeld':return bus.landHeld(m.record);
|
|
case 'landRelease':return bus.landRelease(m.record);
|
|
case 'threadsState':return bus.threadsState(m.business);
|
|
case 'threadIdle':return bus.threadIdle(m.record);
|
|
case 'lifecycleRequests':return bus.lifecycleRequests(m.business);
|
|
case 'sessionClears':return bus.sessionClears(m.business);
|
|
default:throw Error('fixture unrequested trusted operation');
|
|
}
|
|
};
|
|
const host={business:'acme',path:bus.path,dataRoot:f.dataRoot,beforeClose:fn=>hooks.push(fn),
|
|
op:async m=>{
|
|
trace.ops.push(m.op);
|
|
if(m.op==='modelPreflight'){trace.preflightAttempts++;let value=await actualOp(m);if(value&&admit)value=admit(value);if(value){trace.successfulPreflights++;expectedKeyId=id(value.key);}return value;}
|
|
if(m.op==='landHolders'&&trace.successfulPreflights){
|
|
// Existing trusted host call occurs after enable publication and BEFORE bundle,
|
|
// registerRuntime/spawnSession/bind in the production startLocked sequence.
|
|
const dirs=fs.readdirSync(runsDir(f.dataRoot,'acme'));assert.equal(dirs.length,1);const file=join(runsDir(f.dataRoot,'acme'),dirs[0],'proxy-first-turn-enabled');
|
|
assert.equal(fs.readFileSync(file,'utf8'),'enabled\n','managed caller publishes enable before spawn');assert.equal(fs.statSync(file).mode&0o777,0o600);assert.equal(trace.bindings.length,0);assert.equal(readSessions(f.dataRoot).length,0);trace.enableBeforeSpawn.push(true);
|
|
}
|
|
if(m.op==='modelStall'){const result=await actualOp(m);trace.stalls.push({run:m.run,keyId:m.keyId,stopped:result.stopped});return result;}
|
|
return actualOp(m);
|
|
},
|
|
bindLaunch:async record=>{const result=bus.bindLaunch(record);trace.bindings.push({run:record.run,pid:record.pid,startTime:record.startTime,modelKeyId:record.modelKeyId});return result;}};
|
|
const makeLoader=await fakeHarnesses({after:close=>closers.push(close)},Object.fromEntries([harness,...extra].map(name=>[name,real.includes(name)?null:adapter])),write=>write('model-provider','cliproxy',`export * from ${JSON.stringify(new URL('../../plugins/model-provider/cliproxy/index.mjs',import.meta.url).href)};`));
|
|
launcher=createLauncher({host,system,env,launches,makeLoader,...(bwrap?{bwrap}:{}),sessionDefaults:{pollInterval:50},harnessVersion:'0.85.1',log:line=>logs.push(line)});
|
|
// The existing trusted host hook owns all launcher-created children. No registry
|
|
// PID is signalled by fixture teardown; normal bounded launcher close is used.
|
|
closers.push(async()=>{for(const close of hooks)await close(0);});await launcher.listen();
|
|
const human=bus.broker.bindHuman({business:'acme',human:'jason',via:'cli',outsideAgent:true});
|
|
const launchLog=()=>fs.existsSync(launchLogFile(f.dataRoot,'acme'))?fs.readFileSync(launchLogFile(f.dataRoot,'acme'),'utf8').trim().split('\n').filter(Boolean).map(l=>JSON.parse(l)):[];
|
|
const valueFree=()=>checkHistory({logs,proxyReceipts,launch:launchLog(),trace});
|
|
return {root,f,state,trace,proxyReceipts,logs,launcher,bus,launchLog,expectedKeyId:()=>expectedKeyId,valueFree,
|
|
upstreamCredentialNames:Object.keys(env).filter(name=>['ANTHROPIC_API_KEY','ANTHROPIC_AUTH_TOKEN','CLAUDE_CODE_OAUTH_TOKEN','OPENAI_API_KEY','OPENROUTER_API_KEY','ZAI_API_KEY'].includes(name)),
|
|
runDirectories:()=>fs.existsSync(runsDir(f.dataRoot,'acme'))?fs.readdirSync(runsDir(f.dataRoot,'acme')).map(name=>join(runsDir(f.dataRoot,'acme'),name)):[],
|
|
sessions:()=>readSessions(f.dataRoot),send:body=>bus.broker.request(human,{verb:'message.send',args:{to:'chief',body,class:'REQUEST'}}),
|
|
reply:sent=>waitFor(()=>bus.broker.request(human,{verb:'message.receive'}).find(m=>m.in_reply_to===sent.id),'real managed reply observed'),
|
|
assertHistoryGuard:()=>{try{checkHistory({probe:sentinel});}catch(e){if(e.message==='fixture credential leaked')return;}throw Error('fixture history guard failed');}};
|
|
}
|