import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdirSync,writeFileSync,readFileSync,readdirSync,unlinkSync,symlinkSync,renameSync,existsSync,lstatSync,chmodSync,rmSync } from 'node:fs';
import { spawn } from 'node:child_process';
import { createHash } from 'node:crypto';
import { once } from 'node:events';
import { join } from 'node:path';
import { initialize,check,status,readAccounting,supportedHalt,main,serveBudget,CUTOFF } from './budget-supervisor.mjs';
import { fixture,closeChild } from './test-support.mjs';
import { Store,SCHEMA_VERSION } from '../../bus/src/store.mjs';
const NOW=Date.parse('2026-11-06T02:30:00.000Z');
const row=(extra={})=>({at:'2026-11-06T02:29:00.000Z',run:'r123456789abc',turn:'t1',source:'transcript',instance:'assistant',provider:'proxy',route:'glm',model:'glm-5.3-flash',status:'known',input:1,output:1,cacheWrite:1,cacheRead:1,...extra});
function setup(t,tokens=5,changes={}) {
  const f=fixture(t),store=new Store(f.dataRoot);f.after(()=>store.close());
  const campaign=join(f.root,'campaign.json'),ledger=join(f.root,'ledger');
  const c={version:1,instance:'ui797',config:f.config,business:'mvp3-ui797',owner:'ui797-owner',start:'2026-11-06T02:00:00.000Z',end:'2026-11-06T03:00:00.000Z',caps:[{provider:'proxy',route:'glm',model:'glm-5.3-flash',tokens},{provider:'proxy',route:'anthropic',model:'claude-opus-5-5',tokens:6}]};
  Object.assign(c,changes);
  writeFileSync(campaign,JSON.stringify(c),{mode:0o600});initialize({campaign,ledger});
  const halts=[],halt=b=>{halts.push(b);return {halted:true,quiescent:true,generation:'test'};};
  return {...f,store,campaign,ledger,c,halts,halt,opts:{campaign,ledger}};
}
test('tiny cap counts all four token kinds per model, persists/deduplicates across checks and latches entire campaign',async t=>{
  const f=setup(t);const opts={now:()=>NOW,halt:f.halt,read:()=>[row()]};
  assert.equal((await check(f.opts,opts)).stopped,false);
  assert.equal((await check(f.opts,opts)).totals[0].total,4);assert.equal(f.halts.length,0);
  const stopped=await check(f.opts,{...opts,read:()=>[row(),row({turn:'t2',input:0,output:1,cacheRead:0,cacheWrite:0})]});
  assert.equal(stopped.reason,'model-cap-reached');assert.equal(stopped.totals[0].total,5);assert.equal(stopped.totals[1].total,0);
  assert.equal(f.halts[0].dataRoot,f.dataRoot);
  assert.equal(status(f.opts).stopped,true);assert.equal(status(f.opts).totals[0].total,5);
  assert.equal((await check(f.opts,opts)).reason,'stop-already-latched');assert.equal(f.halts.length,2);
  assert.throws(()=>initialize(f.opts));
});
for(const [name,records,reason] of [
  ['unknown',[row({status:'unknown'})],'usage-unknown-or-malformed'],
  ['malformed',[row({input:-1})],'usage-unknown-or-malformed'],
  ['unsafe integer',[row({input:Number.MAX_SAFE_INTEGER+1})],'usage-unknown-or-malformed'],
  ['route drift',[row({route:'anthropic'})],'usage-uncapped-route-model'],
  ['unlisted model',[row({model:'another-model'})],'usage-uncapped-route-model'],
  ['duplicate',[row(),row()],'usage-duplicate'],
])test(`${name} latches before the test-only halt`,async t=>{
  const f=setup(t);
  const result=await check(f.opts,{now:()=>NOW,read:()=>records,halt:b=>{assert.ok(readFileSync(join(f.ledger,'stop.json')).length);return f.halt(b);}});
  assert.equal(result.reason,reason);assert.equal(f.halts.length,1);
});
test('campaign deadline, immutable owner cap and corrupt/missing ledger all stop without auto reset',async t=>{
  for(const variant of ['deadline','raise','malformed','missing']) {
    const sub=await t.test(variant,async t=>{
      const f=setup(t);let now=NOW;
      if(variant==='deadline')now=Date.parse(f.c.end);
      if(variant==='raise'){f.c.caps[0].tokens=10;writeFileSync(f.campaign,JSON.stringify(f.c));}
      if(variant==='malformed')writeFileSync(join(f.ledger,readdirSync(f.ledger).find(n=>n.endsWith('.jsonl'))),'{');
      if(variant==='missing')unlinkSync(join(f.ledger,readdirSync(f.ledger).find(n=>n.endsWith('.jsonl'))));
      const result=await check(f.opts,{now:()=>now,read:()=>[],halt:f.halt});assert.equal(result.stopped,true);assert.equal(f.halts.length,1);
    });
  }
});
test('halt refusal keeps the durable stop and records an unknown halt, with no resume or raise',async t=>{
  const f=setup(t,1);
  await assert.rejects(check(f.opts,{now:()=>NOW,read:()=>[row()],halt:()=>{throw new Error('human-required');}}),/supported-halt-refused/);
  assert.equal(status(f.opts).stopped,true);
  assert.equal(JSON.parse(readFileSync(join(f.ledger,'halt-1.json'))).ok,false);
  await check(f.opts,{now:()=>NOW,halt:f.halt});assert.equal(f.halts.length,1);
});
test('usage removal or rewrite is refused rather than refilling or double-counting',async t=>{
  const f=setup(t,100);await check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt});
  const result=await check(f.opts,{now:()=>NOW,read:()=>[row({input:0})],halt:f.halt});assert.equal(result.reason,'usage-rewritten');
  const g=setup(t,100);await check(g.opts,{now:()=>NOW,read:()=>[row()],halt:g.halt});
  assert.equal((await check(g.opts,{now:()=>NOW,read:()=>[],halt:g.halt})).reason,'usage-disappeared');
});
function insert(f,phase='end',status='known',source='transcript') {
  f.store.run(`INSERT INTO usage_records(at,business,run,instance,provider,model,turn,source,phase,status,input,output,cache_write,cache_read) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
    row().at,f.c.business,row().run,'assistant','proxy','glm-5.3-flash','t1',source,phase,status,...(status==='known'?[1,1,1,1]:[null,null,null,null]));
  const manifest=join(f.dataRoot,'launches',f.c.business,row().run,'bundle');mkdirSync(manifest,{recursive:true,mode:0o700});
  writeFileSync(join(manifest,'manifest.json'),JSON.stringify({business:f.c.business,run:row().run,instance:'assistant',provider:'proxy',model:'glm-5.3-flash',proxyRoute:'glm',route:'primary'}),{mode:0o600});
}
test('accounting reads a real store at the current bus schema version',t=>{
  const f=setup(t,100);insert(f);
  assert.equal(f.store.get("SELECT value FROM meta WHERE key='schema_version'").value,SCHEMA_VERSION);
  const records=readAccounting({dataRoot:f.dataRoot,business:f.c.business,start:f.c.start,end:f.c.end});
  assert.equal(records.length,1);assert.equal(records[0].input,1);
});
for(const version of ['3s','3r','3v',null]) test(`accounting ${version==='3s'?'accepts':'refuses'} schema ${version??'missing'}`,t=>{
  const f=setup(t,100);
  // Only this disposable fixture is altered; the production meta table is append-only.
  f.store.run('DROP TRIGGER meta_no_update');f.store.run('DROP TRIGGER meta_no_delete');
  if(version===null) f.store.run("DELETE FROM meta WHERE key='schema_version'");
  else f.store.run("UPDATE meta SET value=? WHERE key='schema_version'",version);
  const read=()=>readAccounting({dataRoot:f.dataRoot,business:f.c.business,start:f.c.start,end:f.c.end});
  if(version==='3s') assert.deepEqual(read(),[]);
  else assert.throws(read,/^Error: test-store-schema-refused$/);
});
test('actual read-only bus store source matches token-limit records and excludes proxy telemetry',async t=>{
  const f=setup(t,100);insert(f);insert(f,'end','known','@proxy/cliproxy-queue');
  const r=readAccounting({dataRoot:f.dataRoot,business:f.c.business,start:f.c.start,end:f.c.end});assert.equal(r.length,1);assert.equal(r[0].route,'glm');
  assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).totals[0].total,4);
  assert.equal(f.store.get('SELECT count(*) n FROM usage_records').n,2);
});
test('a completed message with no usage records stops; an in-flight begin waits, an ended unknown turn stops',async t=>{
  const f=setup(t,100);insert(f,'begin');
  assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).stopped,false);
  f.store.run(`INSERT INTO events(id,at,business,kind,subject,body) VALUES(?,?,?,?,?,?)`,'ended',row().at,f.c.business,'session.ended',row().run,'{}');
  assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).reason,'usage-unknown');
  const g=setup(t,100);
  g.store.run(`INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)`,'t1',row().at,g.c.business,'chief','chief-run','assistant','note','work');
  g.store.run(`INSERT INTO deliveries(message,at,op,holder_run) VALUES(?,?,?,?)`,'t1',row().at,'read',row().run);
  assert.equal((await check(g.opts,{now:()=>NOW,halt:g.halt})).reason,'usage-missing');
});
test('completed unknown source and fallback manifest fail closed against the real store',async t=>{
  const f=setup(t,100);insert(f,'end','unknown');
  assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).reason,'usage-unknown');
  const g=setup(t,100);insert(g);
  const path=join(g.dataRoot,'launches',g.c.business,row().run,'bundle/manifest.json'),m=JSON.parse(readFileSync(path));m.route='fallback';writeFileSync(path,JSON.stringify(m));
  assert.equal((await check(g.opts,{now:()=>NOW,halt:g.halt})).reason,'usage-route-drift');
});
test('changed target config never directs a halt to a different root',async t=>{
  const f=setup(t,1),other=join(f.root,'other');mkdirSync(other);writeFileSync(join(other,'instance.json'),JSON.stringify({instanceVersion:1,name:'ui797'}));
  const config=JSON.parse(readFileSync(f.config));config.dataRoot=other;writeFileSync(f.config,JSON.stringify(config));
  await assert.rejects(check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt}));assert.equal(f.halts.length,0);
});
test('normal supported halt refuses an agent environment, preserving its markers',t=>{
  const f=setup(t);assert.throws(()=>supportedHalt({config:f.config,dataRoot:f.dataRoot,business:f.c.business},{...process.env,MOSAIC_AGENT_RUN:'test-agent'}),/supported-halt-refused/);
});
test('a corrupt ledger manifest stops with the validated owner campaign, never recreates accounting',async t=>{
  const f=setup(t,100);writeFileSync(join(f.ledger,'campaign.json'),'{');
  const result=await check(f.opts,{now:()=>NOW,read:()=>[],halt:f.halt});assert.equal(result.reason,'ledger-manifest-invalid');assert.equal(f.halts.length,1);
  assert.equal(readFileSync(join(f.ledger,'campaign.json'),'utf8'),'{');
});
test('an invalid existing stop latch still invokes the supported test halt',async t=>{
  const f=setup(t,100);mkdirSync(join(f.ledger,'stop.json'),{mode:0o700});
  const result=await check(f.opts,{now:()=>NOW,read:()=>[],halt:f.halt});assert.equal(result.reason,'stop-latch-invalid');assert.equal(f.halts.length,1);
});
test('model-writable campaign paths refuse initialization',t=>{
  const f=fixture(t),store=new Store(f.dataRoot);f.after(()=>store.close());
  const campaign=join(f.dataRoot,'campaign.json');writeFileSync(campaign,JSON.stringify({version:1,instance:'ui797',config:f.config,business:'mvp3-ui797',owner:'ui797-owner',start:'2026-11-06T02:00:00.000Z',end:'2026-11-06T03:00:00.000Z',caps:[{provider:'proxy',route:'glm',model:'glm-5.3-flash',tokens:1}]}),{mode:0o600});
  assert.throws(()=>initialize({campaign,ledger:join(f.root,'ledger')}),/outside/);
});

test('a failed durable-latch write still halts and returns an explicit storage failure',async t=>{
  const f=setup(t,1);symlinkSync(join(f.root,'missing-stop-target'),join(f.ledger,'stop.json'));
  await assert.rejects(check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt}),/stop-latch-not-durable/);
  assert.equal(f.halts.length,1);assert.equal(JSON.parse(readFileSync(join(f.ledger,'halt-1.json'))).storageError,'stop-latch-not-durable');
});

test('valid JSON journal token corruption durably latches and halts before accepting another turn',async t=>{
  const f=setup(t),deps={now:()=>NOW,read:()=>[row()],halt:f.halt};
  await check(f.opts,deps);
  const file=readdirSync(f.ledger).filter(n=>n.endsWith('.jsonl')).map(n=>join(f.ledger,n)).find(n=>readFileSync(n,'utf8').includes('"key"'));
  const lines=readFileSync(file,'utf8').trimEnd().split('\n').map(JSON.parse);lines[1].tokens=0;
  writeFileSync(file,lines.map(JSON.stringify).join('\n')+'\n');
  const result=await check(f.opts,{...deps,read:()=>[row(),row({turn:'t2'})]});
  assert.equal(result.reason,'ledger-token-mismatch');assert.equal(f.halts.length,1);
  assert.equal(JSON.parse(readFileSync(join(f.ledger,'stop.json'))).reason,'ledger-token-mismatch');
});

test('a check killed while holding its lock is halted on restart without reclaiming the lock',async t=>{
  const f=setup(t,100);
  const code=`import {check} from ${JSON.stringify(new URL('./budget-supervisor.mjs',import.meta.url).href)}; await check(${JSON.stringify(f.opts)},{now:()=>${NOW},read:()=>{process.send('locked');return new Promise(()=>{});}});`;
  const child=spawn(process.execPath,['--input-type=module','-e',code],{stdio:['ignore','ignore','pipe','ipc']});
  f.after(async()=>{if(child.exitCode===null&&child.signalCode===null){child.kill('SIGKILL');await once(child,'close');}});
  await once(child,'message');child.kill('SIGKILL');await once(child,'close');
  const lock=join(f.ledger,'check.lock'),ino=lstatSync(lock).ino;
  const result=await check(f.opts,{now:()=>NOW,read:()=>assert.fail('locked accounting must not run'),halt:f.halt});
  assert.equal(result.reason,'check-lock-unavailable');assert.equal(f.halts.length,1);
  assert.equal(lstatSync(lock).ino,ino);assert.ok(existsSync(join(f.ledger,'stop.json')));
});

test('an existing latch reasserts halt on every locked restart and preserves lock evidence',async t=>{
  const f=setup(t,1);await check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt});
  const lock=join(f.ledger,'check.lock');mkdirSync(lock,{mode:0o700});writeFileSync(join(lock,'owner'),'diagnose');
  for(let n=0;n<2;n++) assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).stopped,true);
  assert.equal(f.halts.length,3);assert.equal(readFileSync(join(lock,'owner'),'utf8'),'diagnose');
});

test('lock storage failures still durably stop the verified test workload',async t=>{
  for(const code of ['EROFS','EACCES']) {
    const f=setup(t,100);
    const result=await check(f.opts,{now:()=>NOW,halt:b=>{assert.equal(b.dataRoot,f.dataRoot);assert.ok(existsSync(join(f.ledger,'stop.json')));return f.halt(b);},acquireLock:()=>{throw Object.assign(new Error('storage'),{code});}});
    assert.equal(result.reason,'check-lock-unavailable');assert.equal(f.halts.length,1);
  }
  const f=setup(t,100),lock=join(f.ledger,'check.lock');writeFileSync(lock,'not a directory',{mode:0o600});
  assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).stopped,true);
  assert.equal(readFileSync(lock,'utf8'),'not a directory');assert.equal(f.halts.length,1);
});

test('budget init, check, status and serve refuse unrelated instance metadata before accounting or halt',async t=>{
  const f=setup(t,100),other=join(f.root,'mosaic-unrelated');renameSync(join(f.root,'mosaic-ui797'),other);
  const config=join(other,'config.json');writeFileSync(join(f.dataRoot,'instance.json'),JSON.stringify({instanceVersion:1,name:'unrelated'}));
  f.c.instance='unrelated';f.c.config=config;writeFileSync(f.campaign,JSON.stringify(f.c));
  assert.throws(()=>initialize({campaign:f.campaign,ledger:join(f.root,'other-ledger')}),/campaign-invalid/);
  const file=join(f.ledger,'campaign.json'),p=JSON.parse(readFileSync(file));p.campaign=f.c;writeFileSync(file,JSON.stringify(p));
  await assert.rejects(check(f.opts,{now:()=>NOW,halt:f.halt}),/literal ui797/);
  assert.throws(()=>status(f.opts),/campaign-invalid/);
  await assert.rejects(main(['serve','--campaign',f.campaign,'--ledger',f.ledger]),/literal ui797/);
  assert.equal(f.halts.length,0);
});

test('persisted campaign metadata is compared to the validated owner manifest before accounting',async t=>{
  const f=setup(t,100),file=join(f.ledger,'campaign.json'),p=JSON.parse(readFileSync(file));
  p.campaign.caps[0].tokens=101;writeFileSync(file,JSON.stringify(p));
  assert.equal((await check(f.opts,{now:()=>NOW,read:()=>assert.fail('changed metadata'),halt:f.halt})).reason,'campaign-changed');
  assert.equal(f.halts.length,1);
});

test('absolute cutoff is inclusive; later windows refuse and a persisted extension cannot delay a stop',async t=>{
  const end='2026-11-07T06:00:00.000Z',f=setup(t,100,{end});
  assert.equal((await check(f.opts,{now:()=>CUTOFF-1,read:()=>[],halt:f.halt})).stopped,false);
  assert.equal((await check(f.opts,{now:()=>CUTOFF,read:()=>[],halt:f.halt})).reason,'campaign-absolute-cutoff');
  assert.equal(f.halts.length,1);
  assert.throws(()=>setup(t,100,{end:'2026-11-08T06:00:00.000Z'}),/campaign-invalid/);
  const g=setup(t,100,{end}),file=join(g.ledger,'campaign.json'),p=JSON.parse(readFileSync(file));
  p.campaign.end='2026-11-08T06:00:00.000Z';writeFileSync(file,JSON.stringify(p));
  assert.equal((await check(g.opts,{now:()=>CUTOFF,halt:g.halt})).reason,'campaign-absolute-cutoff');
  assert.equal(g.halts.length,1);
});

test('only the exact proxy/glm identity receives 100M; other approved models remain capped at 10M',t=>{
  const cap=(provider,route,model,tokens)=>({provider,route,model,tokens});
  const glm=cap('proxy','glm','glm-5.3-flash',100000000),claude=cap('proxy','anthropic','claude-opus-5-5',10000000),codex=cap('proxy','codex','gpt-6-astra',10000000);
  for(const c of [glm,claude,codex])assert.ok(setup(t,1,{caps:[c]}));
  for(const c of [glm,claude,codex])assert.throws(()=>setup(t,1,{caps:[{...c,tokens:c.tokens+1}]}),/campaign-cap-invalid/);
  for(const c of [cap('zai','glm','glm-5.3-flash',100000000),cap('proxy','anthropic','glm-5.3-flash',100000000),cap('proxy','glm','claude-opus-5-5',100000000)])assert.throws(()=>setup(t,1,{caps:[c]}),/campaign-cap-invalid/);
});

test('conflicting route bindings refuse initialization and real-store split-route usage stops',async t=>{
  const cap={provider:'proxy',route:'anthropic',model:'claude-opus-5-5',tokens:10000000};
  assert.throws(()=>setup(t,1,{caps:[{...cap,route:'glm'},cap]}),/campaign-cap-invalid/);
  assert.throws(()=>setup(t,1,{caps:[cap,{...cap}]}),/campaign-cap-invalid/);
  const f=setup(t,1,{caps:[cap]});
  for(const [i,route] of ['anthropic','glm'].entries()) {
    const r=row({run:i?'r123456789abd':'r123456789abc',turn:`t${i}`,model:cap.model,input:6000000,output:0,cacheWrite:0,cacheRead:0});
    f.store.run('INSERT INTO usage_records(at,business,run,instance,provider,model,turn,source,phase,status,input,output,cache_write,cache_read) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)',r.at,f.c.business,r.run,r.instance,r.provider,r.model,r.turn,r.source,'end','known',r.input,0,0,0);
    const dir=join(f.dataRoot,'launches',f.c.business,r.run,'bundle');mkdirSync(dir,{recursive:true,mode:0o700});
    writeFileSync(join(dir,'manifest.json'),JSON.stringify({business:f.c.business,run:r.run,instance:r.instance,provider:r.provider,model:r.model,proxyRoute:route,route:'primary'}),{mode:0o600});
  }
  const result=await check(f.opts,{now:()=>NOW,halt:f.halt});
  assert.equal(result.reason,'usage-uncapped-route-model');assert.equal(f.halts.length,1);
  assert.ok(existsSync(join(f.ledger,'stop.json')));
});

test('the authorized GLM and other-model maxima halt at the exact charged ceiling',async t=>{
  for(const cap of [{provider:'proxy',route:'glm',model:'glm-5.3-flash',tokens:100000000},{provider:'proxy',route:'anthropic',model:'claude-sonnet-5-5',tokens:10000000}]) {
    const f=setup(t,1,{caps:[cap]});
    const r=row({...cap,input:cap.tokens,output:0,cacheWrite:0,cacheRead:0});delete r.tokens;
    const result=await check(f.opts,{now:()=>NOW,read:()=>[r],halt:f.halt});
    assert.equal(result.reason,'model-cap-reached');assert.equal(result.totals[0].total,cap.tokens);assert.equal(f.halts.length,1);
  }
});

for(const variant of ['renamed','missing','mode-0500','cutoff','latched-restart'])test(`unavailable ledger (${variant}) still counts a halt without recreating accounting`,async t=>{
  const f=setup(t,1),offline=f.ledger+'-offline',deps={now:()=>NOW,halt:f.halt,read:()=>[row()]};
  let priorLatch;
  if(variant==='latched-restart') {
    await check(f.opts,deps);priorLatch=readFileSync(join(f.ledger,'stop.json'),'utf8');
    f.halts.length=0; // Model Resume and a fresh check, without any cached binding.
  }
  const before=readdirSync(f.ledger).map(n=>[n,readFileSync(join(f.ledger,n),'utf8')]);
  if(variant==='mode-0500'){chmodSync(f.ledger,0o500);f.after(()=>chmodSync(f.ledger,0o700));}
  else if(variant==='missing')rmSync(f.ledger,{recursive:true});
  else renameSync(f.ledger,offline);
  let running=true;
  const attempts=variant==='latched-restart'?2:1;
  for(let n=0;n<attempts;n++) {
    await assert.rejects(check(f.opts,{...deps,now:()=>variant==='cutoff'?CUTOFF:NOW,halt:b=>{
      assert.equal(running,true);running=false;return f.halt(b);
    }}),e=>{
      assert.deepEqual(e.result.storageErrors,['stop-latch-not-durable','stop-receipt-not-durable']);
      assert.equal(e.result.halt.halted,true);return true;
    });
    assert.equal(running,false);running=true; // Model another Resume after restarted halt.
  }
  assert.equal(f.halts.length,attempts);assert.equal(f.halts[0].dataRoot,f.dataRoot);
  if(variant==='mode-0500')assert.deepEqual(readdirSync(f.ledger).map(n=>[n,readFileSync(join(f.ledger,n),'utf8')]),before);
  else assert.equal(existsSync(f.ledger),false);
  if(priorLatch)assert.equal(readFileSync(join(offline,'stop.json'),'utf8'),priorLatch);
});

test('an unavailable ledger refuses invalid, foreign, changed or unpinned targets before halt',async t=>{
  for(const variant of ['invalid','foreign','changed-root','changed-config','changed-campaign','missing-pin']) {
    const f=setup(t,1);renameSync(f.ledger,f.ledger+'-offline');
    const c=JSON.parse(readFileSync(f.config));
    if(variant==='invalid')writeFileSync(f.config,'{');
    if(variant==='foreign')writeFileSync(join(f.dataRoot,'instance.json'),JSON.stringify({instanceVersion:1,name:'unrelated'}));
    if(variant==='changed-root') {
      const other=join(f.root,'other');mkdirSync(other);writeFileSync(join(other,'instance.json'),JSON.stringify({instanceVersion:1,name:'ui797'}));
      const store=new Store(other);f.after(()=>store.close());c.dataRoot=other;writeFileSync(f.config,JSON.stringify(c));
    }
    if(variant==='changed-config'){c.execution.model='other';writeFileSync(f.config,JSON.stringify(c));}
    if(variant==='changed-campaign'){f.c.business='other-business';writeFileSync(f.campaign,JSON.stringify(f.c));}
    if(variant==='missing-pin')unlinkSync(f.campaign+'.binding.json');
    await assert.rejects(check(f.opts,{now:()=>NOW,halt:f.halt}));assert.equal(f.halts.length,0,variant);
    assert.equal(existsSync(f.ledger),false);
  }
});

test('a required modern sidecar cannot be deleted to regain an allowance; unpinned legacy evidence refuses',async t=>{
  const f=setup(t,100),manifest=readFileSync(join(f.ledger,'campaign.json'),'utf8');
  unlinkSync(f.campaign+'.binding.json');
  const result=await check(f.opts,{now:()=>NOW,read:()=>assert.fail('missing required pin'),halt:f.halt});
  assert.equal(result.stopped,true);assert.deepEqual(result.evidenceErrors,['owner-binding-missing']);
  assert.equal(readFileSync(join(f.ledger,'campaign.json'),'utf8'),manifest);
  assert.equal(existsSync(f.campaign+'.binding.json'),false);assert.equal(f.halts.length,1);
  const g=setup(t,100),file=join(g.ledger,'campaign.json'),old=JSON.parse(readFileSync(file));
  old.version=1;delete old.configHash;writeFileSync(file,JSON.stringify(old));unlinkSync(g.campaign+'.binding.json');
  await assert.rejects(check(g.opts,{now:()=>NOW,halt:g.halt}),/owner-binding-unverified/);
  assert.equal(g.halts.length,0);
  const h=setup(t,100),hfile=join(h.ledger,'campaign.json'),previous=JSON.parse(readFileSync(hfile));
  previous.version=1;delete previous.configHash;writeFileSync(hfile,JSON.stringify(previous));
  const before=readFileSync(hfile,'utf8');
  assert.equal((await check(h.opts,{now:()=>NOW,halt:h.halt,read:()=>[]})).stopped,false);
  assert.equal(readFileSync(hfile,'utf8'),before);
});

test('serve survives unavailable storage, reasserts after Resume each tick and preserves its in-memory stop on recovery',async t=>{
  const f=setup(t,100),offline=f.ledger+'-offline',controller=new AbortController(),reports=[];
  await check(f.opts,{now:()=>NOW,halt:f.halt,read:()=>[]});
  const epoch=readFileSync(join(f.ledger,'campaign.json'),'utf8');renameSync(f.ledger,offline);
  let running=true;
  await serveBudget(f.opts,{interval:1,signal:controller.signal,now:()=>NOW,read:()=>[],halt:b=>{
    assert.equal(running,true);running=false;return f.halt(b);
  },report:r=>{
    reports.push(r);assert.equal(r.stopped,true);assert.equal(r.halt.halted,true);
    running=true; // Model Resume between every tick.
    if(reports.length===2)renameSync(offline,f.ledger);
    if(reports.length===3)controller.abort();
  }});
  assert.equal(f.halts.length,3);assert.equal(reports.length,3);
  for(const r of reports.slice(0,2))assert.deepEqual(r.storageErrors,['stop-latch-not-durable','stop-receipt-not-durable']);
  assert.equal(readFileSync(join(f.ledger,'campaign.json'),'utf8'),epoch); // Never initialize a new epoch.
  assert.equal(JSON.parse(readFileSync(join(f.ledger,'stop.json'))).reason,'ledger-storage-unavailable');
});

test('serve keeps retrying a refused halt and refuses targeting if the config changes between ticks',async t=>{
  const f=setup(t,1),controller=new AbortController(),reports=[];
  await serveBudget(f.opts,{interval:1,signal:controller.signal,now:()=>NOW,read:()=>[row()],halt:b=>{
    f.halts.push(b);throw new Error('modelled halt refusal');
  },report:r=>{
    reports.push(r);
    if(reports.length===2){const c=JSON.parse(readFileSync(f.config));c.execution.model='changed';writeFileSync(f.config,JSON.stringify(c));}
    if(reports.length===4)controller.abort();
  }});
  assert.equal(f.halts.length,2);assert.equal(reports.length,4);
  for(const r of reports.slice(0,2))assert.match(r.error,/supported-halt-refused/);
  for(const r of reports.slice(2))assert.match(r.error,/campaign-instance-changed/);
  assert.equal(existsSync(join(f.ledger,'stop.json')),true);
});

for(const damage of ['missing-ledger','read','parse','mode'])test(`real serve CLI stays alive and attempts supported halt every tick with ${damage} damage`,async t=>{
  const f=setup(t,1);
  if(damage==='missing-ledger')renameSync(f.ledger,f.ledger+'-offline');else damagePin(f,damage);
  const child=spawn(process.execPath,['packages/cli/live/budget-supervisor.mjs','serve','--campaign',f.campaign,'--ledger',f.ledger,'--interval','0.1'],{
    env:{...process.env,MOSAIC_AGENT_RUN:'test-unavailable-ledger'},stdio:['ignore','pipe','pipe'],
  });
  f.after(()=>closeChild(child));
  let text='',errors='';
  child.stderr.on('data',chunk=>errors+=chunk);
  const reports=[];
  child.stdout.on('data',chunk=>{
    text+=chunk;let end;
    while((end=text.indexOf('\n'))!==-1){reports.push(JSON.parse(text.slice(0,end)));text=text.slice(end+1);}
    if(reports.length>=3)child.kill('SIGTERM');
  });
  const timer=setTimeout(()=>child.kill('SIGKILL'),20000);
  const [code,signal]=await once(child,'close');clearTimeout(timer);
  assert.equal(signal,null,errors);assert.equal(code,0,errors);assert.ok(reports.length>=3,errors);
  for(const r of reports) {
    assert.equal(r.stopped,true);
    if(damage==='missing-ledger'){
      assert.equal(r.haltError,'supported-halt-refused-or-unknown');
      assert.deepEqual(r.storageErrors,['stop-latch-not-durable','stop-receipt-not-durable']);
    }else{assert.match(r.error,/supported-halt-refused/);assert.deepEqual(r.evidenceErrors,['owner-binding-unreadable-or-invalid']);}
  }
  assert.equal(existsSync(f.ledger),damage!=='missing-ledger');
});

// Every damaged-sidecar case uses the real usage store and preserves the evidence.
function damagePin(f,kind) {
  const pin=f.campaign+'.binding.json';
  if(kind==='read') chmodSync(pin,0o000);
  if(kind==='parse') writeFileSync(pin,'{');
  if(kind==='mode') chmodSync(pin,0o400);
  return ()=>{
    if(kind==='read'){assert.equal(lstatSync(pin).mode&0o777,0);chmodSync(pin,0o600);}
    if(kind==='mode')assert.equal(lstatSync(pin).mode&0o777,0o400);
    assert.equal(readFileSync(pin,'utf8'),kind==='parse'?'{':readFileSync(pin+'-original','utf8'));
  };
}
for(const kind of ['read','parse','mode']) for(const state of ['restart','cutoff','over-cap','latched-resume'])
  test(`sidecar ${kind} damage at ${state} still latches and halts on restart and between service ticks`,async t=>{
    for(const running of [false,true]) await t.test(running?'running service':'fresh service',async t=>{
      const f=setup(t,state==='over-cap'||state==='latched-resume'?3:100);
      insert(f);
      const pin=f.campaign+'.binding.json',epoch=readFileSync(join(f.ledger,'campaign.json'),'utf8');
      writeFileSync(pin+'-original',readFileSync(pin),{mode:0o600});
      const now=()=>state==='cutoff'?CUTOFF:NOW;
      if(state==='latched-resume')await check(f.opts,{now,halt:f.halt});
      let assertDamage;
      if(!running)assertDamage=damagePin(f,kind);
      const controller=new AbortController(),reports=[];
      await serveBudget(f.opts,{interval:1,signal:controller.signal,now,halt:f.halt,report:r=>{
        reports.push(r);
        if(running && reports.length===1){assertDamage=damagePin(f,kind);f.halts.length=0;}
        else {assert.equal(r.stopped,true);assert.equal(r.halt.halted,true);assert.ok(r.evidenceErrors.includes('owner-binding-unreadable-or-invalid'));}
        if(reports.length===3)controller.abort();
      }});
      assert.equal(f.halts.length,running?2:state==='latched-resume'?4:3);
      assertDamage();assert.equal(readFileSync(join(f.ledger,'campaign.json'),'utf8'),epoch);
      assert.ok(existsSync(join(f.ledger,'stop.json')));
      assert.ok(JSON.parse(readFileSync(join(f.ledger,`halt-${readdirSync(f.ledger).filter(n=>/^halt-[0-9]+\.json$/.test(n)).length}.json`))).evidenceErrors.length);
    });
  });

test('running service retains validated target evidence when both sidecar and ledger become unavailable',async t=>{
  const f=setup(t,100),controller=new AbortController(),reports=[];
  await serveBudget(f.opts,{now:()=>NOW,interval:1,signal:controller.signal,read:()=>[],halt:f.halt,report:r=>{
    reports.push(r);
    if(reports.length===1){writeFileSync(f.campaign+'.binding.json','{');renameSync(f.ledger,f.ledger+'-offline');}
    else {assert.equal(r.stopped,true);assert.equal(r.halt.halted,true);assert.equal(r.storageErrors.length,2);assert.ok(r.evidenceErrors.length);}
    if(reports.length===3)controller.abort();
  }});
  assert.equal(f.halts.length,2);assert.equal(existsSync(f.ledger),false);
});

for(const kind of ['read','parse','mode'])test(`damaged sidecar (${kind}) never permits changed config bytes or foreign marker/root/store`,async t=>{
  for(const changed of ['config','invalid','marker','root','store']) {
    const f=setup(t,1);damagePin(f,kind);
    if(changed==='config')writeFileSync(f.config,readFileSync(f.config,'utf8')+' ');
    if(changed==='invalid')writeFileSync(f.config,'{');
    if(changed==='marker')writeFileSync(join(f.dataRoot,'instance.json'),JSON.stringify({instanceVersion:1,name:'foreign'}));
    if(changed==='root')renameSync(f.dataRoot,f.dataRoot+'-offline');
    if(changed==='store')renameSync(join(f.dataRoot,'bus/bus.sqlite'),join(f.dataRoot,'bus/bus.sqlite-offline'));
    await assert.rejects(check(f.opts,{now:()=>CUTOFF,halt:f.halt}));assert.equal(f.halts.length,0,changed);
  }
});

for(const variant of ['coherent-epoch-edit','zero-pin-hash'])test(`real-store ${variant} stops without accepting a fresh accounting epoch`,async t=>{
  const f=setup(t,5);insert(f);
  assert.equal((await check(f.opts,{now:()=>NOW,halt:f.halt})).totals[0].total,4);
  const pin=f.campaign+'.binding.json',original=readFileSync(pin,'utf8');
  if(variant==='coherent-epoch-edit') {
    f.c.start='2026-11-06T02:29:30.000Z';const text=JSON.stringify(f.c);writeFileSync(f.campaign,text);
    const file=join(f.ledger,'campaign.json'),p=JSON.parse(readFileSync(file));p.campaign=f.c;p.hash=createHash('sha256').update(text).digest('hex');writeFileSync(file,JSON.stringify(p));
    for(const n of readdirSync(f.ledger).filter(n=>n.endsWith('.jsonl')))writeFileSync(join(f.ledger,n),readFileSync(join(f.ledger,n),'utf8').split('\n')[0]+'\n');
    f.store.run('INSERT INTO usage_records(at,business,run,instance,provider,model,turn,source,phase,status,input,output,cache_write,cache_read) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
      '2026-11-06T02:29:45.000Z',f.c.business,row().run,'assistant','proxy','glm-5.3-flash','t2','transcript','end','known',4,0,0,0);
    assert.equal(f.store.get('SELECT sum(input+output+cache_write+cache_read) total FROM usage_records').total,8);
  } else {const p=JSON.parse(original);p.hash='0'.repeat(64);writeFileSync(pin,JSON.stringify(p));}
  const before=readdirSync(f.ledger).filter(n=>n.endsWith('.jsonl')).map(n=>[n,readFileSync(join(f.ledger,n),'utf8')]);
  const result=await check(f.opts,{now:()=>NOW,halt:f.halt});
  assert.equal(result.reason,'owner-binding-epoch-changed');assert.equal(f.halts.length,1);
  assert.deepEqual(readdirSync(f.ledger).filter(n=>n.endsWith('.jsonl')).map(n=>[n,readFileSync(join(f.ledger,n),'utf8')]),before);
  if(variant==='coherent-epoch-edit')assert.equal(readFileSync(pin,'utf8'),original);
  assert.throws(()=>status(f.opts),/owner-binding-epoch-changed/);
});

test('semantic sidecar target damage stops using independent config/marker/root/store evidence',async t=>{
  for(const field of ['ledger','binding','configHash']) {
    const f=setup(t,100),path=f.campaign+'.binding.json',pin=JSON.parse(readFileSync(path));
    if(field==='ledger')pin.ledger+='-changed';
    if(field==='binding')pin.binding.business='changed-business';
    if(field==='configHash')pin.configHash='0'.repeat(64);
    const damaged=JSON.stringify(pin);writeFileSync(path,damaged);
    const result=await check(f.opts,{now:()=>NOW,halt:f.halt});
    assert.equal(result.stopped,true);assert.ok(result.evidenceErrors.includes('owner-binding-target-changed'));
    assert.equal(f.halts.length,1);assert.equal(f.halts[0].business,f.c.business);assert.equal(readFileSync(path,'utf8'),damaged);
  }
});

// A conflicting ledger never authorizes its own target (support r4 P1).
function conflictingLedger(t,kind) {
  const f=setup(t,100),path=join(f.ledger,'campaign.json');
  const g=kind==='foreign'?setup(t,100,{business:'unrelated-business'}):null;
  const alter=()=>{
    if(g) { writeFileSync(path,readFileSync(join(g.ledger,'campaign.json'))); return; }
    const p=JSON.parse(readFileSync(path));p.binding.business='unrelated-business';p.campaign.business='unrelated-business';writeFileSync(path,JSON.stringify(p));
  };
  const wrong=()=>f.halts.filter(b=>b.business!==f.c.business||b.config!==f.config||b.dataRoot!==f.dataRoot);
  return {f,g,alter,wrong};
}
test('a business-only or foreign ledger is replaced by the owner-verified target in check',async t=>{
  for(const kind of ['business-only','foreign']) {
    const {f,alter,wrong}=conflictingLedger(t,kind);alter();
    const result=await check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt});
    assert.equal(result.stopped,true);assert.ok(result.evidenceErrors.includes('ledger-target-changed'));
    assert.equal(f.halts.length,1);assert.deepEqual(wrong(),[]);
  }
});
test('serve startup never retains a conflicting ledger target',async t=>{
  for(const kind of ['business-only','foreign']) {
    const {f,alter,wrong}=conflictingLedger(t,kind),controller=new AbortController(),reports=[];alter();
    await serveBudget(f.opts,{interval:1,signal:controller.signal,now:()=>NOW,read:()=>[row()],halt:f.halt,report:r=>{reports.push(r);if(reports.length===3)controller.abort();}});
    assert.equal(reports.length,3);assert.ok(f.halts.length>=1);assert.deepEqual(wrong(),[]);
    for(const r of reports) assert.ok(r.evidenceErrors.includes('ledger-target-changed'));
  }
});
test('a running service halts only its retained owner target when the ledger changes under it',async t=>{
  for(const kind of ['business-only','foreign']) {
    const {f,alter,wrong}=conflictingLedger(t,kind),controller=new AbortController(),reports=[];
    await serveBudget(f.opts,{interval:1,signal:controller.signal,now:()=>NOW,read:()=>[row()],halt:f.halt,report:r=>{
      reports.push(r);if(reports.length===1)alter();if(reports.length===3)controller.abort();
    }});
    assert.equal(reports[0].stopped,false);
    for(const r of reports.slice(1)) { assert.equal(r.stopped,true);assert.ok(r.evidenceErrors.includes('ledger-target-changed')); }
    assert.ok(f.halts.length>=1);assert.deepEqual(wrong(),[]);
  }
});
test('a conflicting ledger without corroborating owner evidence refuses every halt',async t=>{
  for(const kind of ['business-only','foreign']) {
    const {f,alter}=conflictingLedger(t,kind);alter();
    const pin=f.campaign+'.binding.json',damaged=JSON.parse(readFileSync(pin));damaged.configHash='0'.repeat(64);writeFileSync(pin,JSON.stringify(damaged));
    await assert.rejects(check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt}),/campaign-instance-changed/);
    assert.equal(f.halts.length,0);
  }
});

// A pin damaged to match a damaged ledger, or a foreign pin restored with the
// invoked ledger path, never corroborates that ledger's target (support r5 P1).
function matchingDamage(t,kind) {
  const {f,g,alter:alterLedger,wrong}=conflictingLedger(t,kind==='business'?'business-only':'foreign');
  const pinPath=f.campaign+'.binding.json',owner=readFileSync(f.campaign,'utf8');
  const alter=()=>{
    if(kind==='business') { const pin=JSON.parse(readFileSync(pinPath));pin.binding.business='unrelated-business';writeFileSync(pinPath,JSON.stringify(pin)); }
    else { const pin=JSON.parse(readFileSync(g.campaign+'.binding.json'));pin.ledger=f.ledger;writeFileSync(pinPath,JSON.stringify(pin)); }
    alterLedger();
  };
  const receipts=dir=>readdirSync(dir).filter(n=>n==='stop.json'||/^halt-\d+\.json$/.test(n));
  return {f,g,alter,wrong,owner,receipts};
}
test('matching pin and ledger damage refuses check and serve startup with no halt',async t=>{
  for(const kind of ['business','foreign']) {
    let m=matchingDamage(t,kind);m.alter();
    await assert.rejects(check(m.f.opts,{now:()=>NOW,read:()=>[row()],halt:m.f.halt}),/campaign-instance-changed/);
    assert.equal(m.f.halts.length,0);assert.deepEqual(m.receipts(m.f.ledger),[]);assert.equal(readFileSync(m.f.campaign,'utf8'),m.owner);
    m=matchingDamage(t,kind);m.alter();
    await assert.rejects(serveBudget(m.f.opts,{interval:1,now:()=>NOW,read:()=>[row()],halt:m.f.halt,report:()=>assert.fail('no tick may run')}),/campaign-instance-changed/);
    assert.equal(m.f.halts.length,0);assert.deepEqual(m.receipts(m.f.ledger),[]);
    if(m.g) assert.deepEqual(m.receipts(m.g.ledger),[]);
  }
});
test('matching pin and ledger damage under a running service refuses every later tick',async t=>{
  for(const kind of ['business','foreign']) {
    const m=matchingDamage(t,kind),controller=new AbortController(),reports=[];
    await serveBudget(m.f.opts,{interval:1,signal:controller.signal,now:()=>NOW,read:()=>[],halt:m.f.halt,report:r=>{
      reports.push(r);if(reports.length===1)m.alter();if(reports.length===3)controller.abort();
    }});
    assert.equal(reports[0].stopped,false);
    for(const r of reports.slice(1)) assert.equal(r.error,'campaign-instance-changed');
    assert.equal(m.f.halts.length,0);assert.deepEqual(m.wrong(),[]);assert.deepEqual(m.receipts(m.f.ledger),[]);
  }
});
test('a running service keeps its startup ledger when the caller mutates the options object',async t=>{
  const f=setup(t,100),replacement=join(f.root,'replacement-ledger'),controller=new AbortController(),reports=[];
  const startup=JSON.parse(readFileSync(join(f.ledger,'campaign.json'))),opts={...f.opts};let now=NOW;
  await serveBudget(opts,{interval:1,signal:controller.signal,now:()=>now,read:()=>[],halt:f.halt,report:r=>{
    reports.push(r);
    if(reports.length===1) {
      mkdirSync(replacement,{mode:0o700});
      for(const n of readdirSync(f.ledger)) writeFileSync(join(replacement,n),readFileSync(join(f.ledger,n)),{mode:0o600});
      const pin=JSON.parse(readFileSync(f.campaign+'.binding.json'));pin.ledger=replacement;writeFileSync(f.campaign+'.binding.json',JSON.stringify(pin));
      opts.ledger=replacement;now=Date.parse(f.c.end);
    }
    if(reports.length===3)controller.abort();
  }});
  assert.equal(f.halts.length,2);for(const b of f.halts) assert.deepEqual(b,startup.binding);
  const names=readdirSync(f.ledger).filter(n=>n==='stop.json'||/^halt-\d+\.json$/.test(n)).sort();
  assert.deepEqual(names,['halt-1.json','halt-2.json','stop.json']);
  const stop=JSON.parse(readFileSync(join(f.ledger,'stop.json')));
  assert.equal(stop.reason,'campaign-ended');assert.deepEqual(stop.binding,startup.binding);
  assert.deepEqual(readdirSync(replacement).filter(n=>n==='stop.json'||/^halt-\d+\.json$/.test(n)),[]);
});
test('an owner business or config retarget refuses without retained evidence and halts the retained target when running',async t=>{
  for(const field of ['business','config']) {
    const retarget=f=>{const g=setup(t,100,{business:'unrelated-business'}),c=JSON.parse(readFileSync(f.campaign));c[field]=field==='business'?g.c.business:g.config;writeFileSync(f.campaign,JSON.stringify(c));return g;};
    let f=setup(t,100),g=retarget(f);
    await assert.rejects(check(f.opts,{now:()=>NOW,read:()=>[row()],halt:f.halt}),/campaign-instance-changed/);
    assert.equal(f.halts.length,0);assert.equal(g.halts.length,0);
    f=setup(t,100);const startup=JSON.parse(readFileSync(join(f.ledger,'campaign.json'))),controller=new AbortController(),reports=[];
    await serveBudget(f.opts,{interval:1,signal:controller.signal,now:()=>NOW,read:()=>[],halt:f.halt,report:r=>{
      reports.push(r);if(reports.length===1)retarget(f);if(reports.length===3)controller.abort();
    }});
    assert.equal(reports[0].stopped,false);
    for(const r of reports.slice(1)) { assert.equal(r.stopped,true);assert.equal(r.reason,'campaign-changed'); }
    assert.equal(f.halts.length,2);for(const b of f.halts) assert.deepEqual(b,startup.binding);
  }
});