stack/packages/cli/live/budget-supervisor.mjs

#!/usr/bin/env node
// Metered stop, with accounting/in-flight delay. Never a provider-side hard ceiling.
import { DatabaseSync } from 'node:sqlite';
import { createHash } from 'node:crypto';
import { closeSync, existsSync, fsyncSync, lstatSync, mkdirSync, openSync, readFileSync, readdirSync, rmdirSync, writeFileSync } from 'node:fs';
import { join, dirname } from 'node:path';
import { spawnSync } from 'node:child_process';
import { turnRecords } from '../../bus/src/usage.mjs';
import { exclusive, fail, isMain, options, outsideRuntime, plainPath, privateDir, readPrivate, ROOT, testInstance } from './support-common.mjs';
export const CUTOFF = Date.parse('2026-11-07T06:00:00.000Z');
const modelIdentity = c => JSON.stringify([c.provider,c.model]);
// The charter permits the GLM identity and Claude/Codex coder model families via proxy.
const approvedBinding = c => c.provider === 'proxy' && (
  (c.route === 'glm' && c.model === 'glm-5.3-flash') ||
  (c.route === 'anthropic' && /^claude-[a-zA-Z0-9._-]+$/.test(c.model)) ||
  (c.route === 'codex' && /^gpt-[a-zA-Z0-9._-]+$/.test(c.model)));
const ceiling = c => c.provider === 'proxy' && c.route === 'glm' && c.model === 'glm-5.3-flash' ? 100000000 : 10000000;
const kinds = ['input','output','cacheWrite','cacheRead'];
const digest = x => createHash('sha256').update(typeof x === 'string' ? x : JSON.stringify(x)).digest('hex');
const identity = c => JSON.stringify([c.provider,c.route,c.model]);
const filename = c => digest(identity(c)) + '.jsonl';
const canonicalTime = s => typeof s === 'string' && Number.isFinite(Date.parse(s)) && new Date(s).toISOString() === s;
function campaignAt(path) {
  privateDir(dirname(path));
  const text = readPrivate(path), c = JSON.parse(text);
  validateCampaign(c);
  return { c, hash:digest(text) };
}
function validateCampaign(c) {
  if (c.instance !== 'ui797' || c.version !== 1 || c.owner !== 'ui797-owner' || !/^[a-z][a-z0-9-]{1,63}$/.test(c.business ?? '') || !canonicalTime(c.start) || !canonicalTime(c.end) || c.end <= c.start || Date.parse(c.end) > CUTOFF ||
    !Array.isArray(c.caps) || !c.caps.length || c.caps.length > 20) fail('campaign-invalid');
  const keys = new Set();
  for (const cap of c.caps) {
    if (Object.keys(cap).sort().join(',') !== 'model,provider,route,tokens' || !['proxy','anthropic','zai','openai'].includes(cap.provider) ||
      !['glm','anthropic','codex'].includes(cap.route) || !/^[a-zA-Z0-9][a-zA-Z0-9._:/-]{0,127}$/.test(cap.model ?? '') ||
      !approvedBinding(cap) || !Number.isSafeInteger(cap.tokens) || cap.tokens < 1 || cap.tokens > ceiling(cap) || keys.has(modelIdentity(cap))) fail('campaign-cap-invalid');
    keys.add(modelIdentity(cap));
  }
}
function binding(c) {
  const { system } = testInstance(c.instance,c.config);
  const st = lstatSync(system.dataRoot), dbPath = join(system.dataRoot,'bus/bus.sqlite'); plainPath(dbPath);
  const db = lstatSync(dbPath);
  if (!db.isFile() || db.uid !== process.getuid()) fail('test-store-required');
  return { instance:c.instance,config:c.config,dataRoot:system.dataRoot,business:c.business,owner:c.owner,rootIdentity:[st.dev,st.ino],storeIdentity:[db.dev,db.ino] };
}
function checkBinding(pinned, c) {
  const b = binding(c);
  if (JSON.stringify(b) !== JSON.stringify(pinned)) fail('campaign-instance-changed');
  return b;
}
function writeJournal(file, row) {
  readPrivate(file);
  const fd = openSync(file,'a');
  try { writeFileSync(fd,JSON.stringify(row)+'\n'); fsyncSync(fd); } finally { closeSync(fd); }
}
function flushDir(dir) { const fd = openSync(dir,'r'); try { fsyncSync(fd); } finally { closeSync(fd); } }
export function initialize({ campaign, ledger }) {
  const { c,hash } = campaignAt(campaign), b = binding(c);
  outsideRuntime(campaign,{dataRoot:b.dataRoot},c.config);
  outsideRuntime(ledger,{dataRoot:b.dataRoot},c.config);
  // Independent owner evidence survives loss of the accounting volume. Created
  // only at initialization; a check must never reconstruct an epoch or this pin.
  const configHash=digest(readFileSync(c.config,'utf8'));
  exclusive(campaign+'.binding.json',JSON.stringify({version:1,hash,configHash,binding:b,ledger,campaign:c})+'\n');
  flushDir(dirname(campaign));
  privateDir(ledger,true);
  for (const cap of c.caps) exclusive(join(ledger,filename(cap)),JSON.stringify({ version:1,cap })+'\n');
  exclusive(join(ledger,'campaign.json'),JSON.stringify({ version:2,hash,configHash,binding:b,campaign:c })+'\n');
  flushDir(ledger);
  return { initialized:true,ledger };
}
export function readAccounting({ dataRoot,business,start,end }) {
  const db = new DatabaseSync(join(dataRoot,'bus/bus.sqlite'),{readOnly:true});
  try {
    db.exec('BEGIN');
    // U4a (#1873) 3u-3v adds only store_tasks, which accounting never reads, so 3v joins the closed list.
    // SP9 (#1873) 3v-3w drops only the table 3t-3u added, which accounting never reads, so 3w joins too.
    if (!['3s','3t','3u','3v','3w'].includes(db.prepare("SELECT value FROM meta WHERE key='schema_version'").get()?.value)) fail('test-store-schema-refused');
    const store = { all:(sql,...args) => db.prepare(sql).all(...args) };
    const records = turnRecords(store,business);
    const raw = store.all("SELECT * FROM usage_records WHERE business=? AND source<>'@proxy/cliproxy-queue' ORDER BY seq",business);
    const completed = store.all(`SELECT DISTINCT d.holder_run run,m.id turn,d.at FROM deliveries d JOIN messages m ON m.id=d.message
      WHERE m.business=? AND d.op='read' AND d.holder_run IS NOT NULL AND d.at>=?`,business,start);
    const ended = store.all("SELECT subject run,at FROM events WHERE business=? AND kind='session.ended' AND at>=?",business,start);
    db.exec('COMMIT');
    const phases = new Map(raw.map(r => [JSON.stringify([r.run,r.turn,r.source]),r.phase]));
    const endedRuns = new Set(ended.map(r=>r.run));
    for (const r of raw) if (!['begin','end'].includes(r.phase) || !['known','unknown'].includes(r.status)) fail('usage-malformed');
    const interrupted = store.all(`SELECT DISTINCT d.holder_run run,m.id turn FROM deliveries d JOIN messages m ON m.id=d.message JOIN events e ON e.subject=d.holder_run AND e.business=m.business AND e.kind='session.ended' WHERE m.business=? AND d.op='delivered' AND e.at>=?`,business,start);
    for (const t of interrupted) if (!records.some(r=>r.run===t.run && r.turn===t.turn && r.status==='known')) fail('usage-missing');
    for (const r of records) {
      if (!canonicalTime(r.at)) fail('usage-malformed');
      if (r.at < start) continue;
      // An in-flight begin is not unknown COMPLETED usage. It is checked on completion/end.
      if (phases.get(JSON.stringify([r.run,r.turn,r.source])) !== 'end' && !endedRuns.has(r.run) && !completed.some(x=>x.run===r.run && x.turn===r.turn)) continue;
      if (r.status !== 'known') fail('usage-unknown');
    }
    const selected = records.filter(r => r.at >= start && r.status === 'known');
    for (const t of completed) if (!selected.some(r=>r.run===t.run && r.turn===t.turn)) fail('usage-missing');
    // Ended runs with begun turns must have a completed record for every source.
    const rows = selected.map(r => {
      if (!/^r[0-9a-f]{12}$/.test(r.run)) fail('usage-run-invalid');
      const manifest = JSON.parse(readPrivate(join(dataRoot,'launches',business,r.run,'bundle/manifest.json')));
      if (manifest.business !== business || manifest.run !== r.run || manifest.instance !== r.instance || manifest.provider !== r.provider || manifest.model !== r.model || manifest.route !== 'primary') fail('usage-route-drift');
      if (!['glm','anthropic','codex'].includes(manifest.proxyRoute)) fail('usage-route-unknown');
      return { ...r,route:manifest.proxyRoute };
    });
    return rows;
  } finally { db.close(); }
}
export function supportedHalt(b, env=process.env) {
  // This supervisor must be installed as a genuine owner service. Never strip agent markers,
  // bindHuman/bindConsole, forge ancestry, or write control/processing.json directly.
  const p = spawnSync(process.execPath,[join(ROOT,'packages/cli/src/cli.mjs'),'halt','--business',b.business,'--json'],{
    env:{...env,MOSAIC_CONFIG:b.config},encoding:'utf8',timeout:40000,maxBuffer:1024*1024,
  });
  if (p.status !== 0) fail('supported-halt-refused-or-unknown');
  let result; try { result=JSON.parse(p.stdout); } catch { fail('supported-halt-invalid-reply'); }
  if (result.halted !== true || result.quiescent !== true) fail('supported-halt-not-quiescent');
  return { halted:true,quiescent:true,generation:result.generation };
}
function journal(ledger,cap) {
  const text = readPrivate(join(ledger,filename(cap)));
  if (!text.endsWith('\n')) fail('ledger-truncated');
  const [header,...entries] = text.trimEnd().split('\n').map(JSON.parse);
  if (header.version !== 1 || JSON.stringify(header.cap) !== JSON.stringify(cap)) fail('ledger-header-changed');
  const seen = new Map(); let total = 0;
  for (const e of entries) {
    if (typeof e.key !== 'string' || typeof e.digest !== 'string' || !Number.isSafeInteger(e.tokens) || e.tokens < 0 || seen.has(e.key)) fail('ledger-malformed');
    seen.set(e.key,e); total += e.tokens;
    if (!Number.isSafeInteger(total)) fail('ledger-overflow');
  }
  return { entries:seen,total };
}
function loadLedger(ledger) {
  privateDir(ledger); const p=JSON.parse(readPrivate(join(ledger,'campaign.json')));
  if (![1,2].includes(p.version) || !p.binding || !p.campaign ||
      (p.version===2 && !/^[0-9a-f]{64}$/.test(p.configHash))) fail('ledger-manifest-invalid');
  return p;
}
// A ledger's target may direct a halt only when independent evidence names the
// same target: the owner campaign itself, the owner's pin, or the instance the
// current owner campaign resolves to. A conflicting ledger never authorizes its
// own target. The owner-verified pin replaces it only when the pin agrees with
// the owner campaign and the live config/marker/root/store; anything else refuses.
// A pin corroborates only a target its own campaign names and, while the owner
// campaign is readable, the target the owner names too: a pin whose binding was
// damaged to match a damaged ledger justifies nothing (support r5 P1). The pin and
// ledger are both writable by whoever damaged them; only the owner campaign is not.
// So with no retained startup evidence, an owner business/config edit refuses as an
// ambiguous target. A running service still halts its retained startup target while
// the ledger is exactly the one it verified at startup.
const TARGET_FIELDS=['instance','config','business','owner'];
const names=(b,c)=>!!c && TARGET_FIELDS.every(k=>b?.[k]===c[k]);
function corroborate(p,pin,{campaign,ledger},retained=null) {
  const same=(a,b)=>JSON.stringify(a)===JSON.stringify(b);
  let owner=null, live=null, liveHash=null;
  try { owner=campaignAt(campaign); live=binding(owner.c); liveHash=digest(readFileSync(owner.c.config,'utf8')); } catch { /* No owner evidence. */ }
  if(owner && owner.hash===p.hash && same(owner.c,p.campaign)) return p;
  if(retained && retained.ledger===ledger && retained.p.hash===p.hash && same(retained.p.campaign,p.campaign) &&
     same(retained.p.binding,p.binding) && retained.configHash===p.configHash) return p;
  const pinCampaign=pin && (pin.campaign ?? (owner && pin.hash===owner.hash ? owner.c : null));
  if(pin && pin.ledger===ledger && same(pin.binding,p.binding) && pin.configHash===p.configHash &&
     names(pin.binding,pinCampaign) && (!owner || names(pin.binding,owner.c))) return p;
  if(live && same(live,p.binding)) return p;
  if(owner && pin && pin.ledger===ledger && pin.hash===owner.hash && (!pin.campaign || same(pin.campaign,owner.c)) &&
     same(pin.binding,live) && pin.configHash===liveHash)
    return {campaign:owner.c,hash:pin.hash,configHash:pin.configHash,binding:pin.binding};
  fail('campaign-instance-changed');
}
function checkTarget({campaign,ledger}, retained=null) {
  let p, pin, reason=null, writable=true;
  const evidenceErrors=[];
  try { privateDir(ledger); } catch { writable=false; reason='ledger-storage-unavailable'; }
  try {
    pin=JSON.parse(readPrivate(campaign+'.binding.json'));
    if(pin?.version!==1 || !pin.binding || !/^[0-9a-f]{64}$/.test(pin.configHash)) throw new Error('invalid');
  } catch(e) {
    pin=null;
    evidenceErrors.push(e.code==='ENOENT'?'owner-binding-missing':'owner-binding-unreadable-or-invalid');
  }
  try { if (!writable) throw new Error(reason); p=loadLedger(ledger); } catch {
    reason??='ledger-manifest-invalid';
    if(retained) p=retained.p;
    else {
      if(!pin || pin.ledger!==ledger) fail('owner-binding-unverified');
      const {c,hash}=campaignAt(campaign);
      // A fallback must never infer a new target/epoch from altered owner bytes.
      if(pin.hash!==hash) fail('owner-binding-invalid-or-changed');
      p={campaign:pin.campaign??c,hash:pin.hash,configHash:pin.configHash,binding:pin.binding};
    }
  }
  if(p.configHash) {
    const resolved=corroborate(p,pin,{campaign,ledger},retained);
    if(resolved!==p) { p=resolved; evidenceErrors.push('ledger-target-changed'); }
  }
  const c=p.campaign;
  // Pin the target before any halt. A changed config cannot redirect the stop into live data.
  checkBinding(p.binding,c); outsideRuntime(ledger,{dataRoot:p.binding.dataRoot},c.config);
  outsideRuntime(campaign,{dataRoot:p.binding.dataRoot},c.config);
  const configHash=p.configHash??pin?.configHash??retained?.configHash;
  if(!configHash) fail('owner-binding-unverified');
  if(configHash!==digest(readFileSync(c.config,'utf8'))) fail('campaign-instance-changed');
  for(const source of [pin,retained && {...retained.p,ledger:retained.ledger,configHash:retained.configHash}].filter(Boolean)) {
    if(source.ledger!==ledger || JSON.stringify(source.binding)!==JSON.stringify(p.binding) || source.configHash!==configHash) {
      if(source!==pin || !p.configHash) fail('campaign-instance-changed');
      evidenceErrors.push('owner-binding-target-changed');
    }
    if(source.hash!==p.hash || (source.campaign && JSON.stringify(source.campaign)!==JSON.stringify(c)))
      evidenceErrors.push('owner-binding-epoch-changed');
  }
  reason??=evidenceErrors[0]??null;
  return {p,c,reason,writable,configHash,evidenceErrors};
}
export async function check(opts, { now=Date.now, read=readAccounting, halt=supportedHalt, acquireLock=mkdirSync, stopReason=null, retained=null }={}) {
  const {ledger,campaign}=opts;
  const {p,c,reason:targetReason,writable,configHash,evidenceErrors}=checkTarget(opts,retained);
  const lock=join(ledger,'check.lock'); let locked=false, lockFailure=false;
  if(writable) try { acquireLock(lock,{mode:0o700}); locked=true; } catch { lockFailure=true; }
  // An interrupted/active/unwritable lock prevents accounting, never the verified halt.
  // Only the check that created a lock may remove it on normal completion.
  try {
    let reason=targetReason ?? stopReason ?? (lockFailure?'check-lock-unavailable':null), totals=[];
    const stopFile=join(ledger,'stop.json');
    if (writable && existsSync(stopFile)) {
      reason='stop-already-latched';
      try { JSON.parse(readPrivate(stopFile)); } catch { reason='stop-latch-invalid'; }
    }
    try {
      const clock=now(); if (!Number.isFinite(clock)) fail('campaign-clock-invalid');
      if (clock>=CUTOFF) fail('campaign-absolute-cutoff');
      validateCampaign(c);
      const owner=campaignAt(campaign);
      if (owner.hash !== p.hash || JSON.stringify(owner.c) !== JSON.stringify(c)) fail('campaign-changed');
      if (clock < Date.parse(c.start)) fail('campaign-not-started'); if (clock>=Date.parse(c.end)) fail('campaign-ended');
      if (!reason) {
        const records=await read({...p.binding,start:c.start,end:c.end}), observed=new Set();
        const accounts=c.caps.map(cap=>({cap,...journal(ledger,cap),total:0}));
        for (const r of records) {
          if (!canonicalTime(r.at) || r.at<c.start || r.status!=='known' || kinds.some(k=>!Number.isSafeInteger(r[k]) || r[k]<0)) fail('usage-unknown-or-malformed');
          const a=accounts.find(a=>identity(a.cap)===identity(r)); if (!a) fail('usage-uncapped-route-model');
          const key=JSON.stringify([r.run,r.turn,r.source]); if (observed.has(key)) fail('usage-duplicate'); observed.add(key);
          const tokens=kinds.reduce((sum,k)=>sum+r[k],0); if (!Number.isSafeInteger(tokens)) fail('usage-overflow');
          const hash=digest(r), previous=a.entries.get(key);
          if (previous && previous.digest !== hash) fail('usage-rewritten');
          if (previous && previous.tokens !== tokens) fail('ledger-token-mismatch');
          a.total+=tokens; if (!Number.isSafeInteger(a.total)) fail('ledger-overflow');
          if (!previous) {
            const e={key,digest:hash,tokens}; writeJournal(join(ledger,filename(a.cap)),e); a.entries.set(key,e);
          }
        }
        for (const a of accounts) for (const key of a.entries.keys()) if (!observed.has(key)) fail('usage-disappeared');
        totals=accounts.map(a=>({...a.cap,total:a.total}));
        // Route journals are evidence, never separate allowances for the same model.
        const models=new Map();
        for (const a of totals) {
          const key=modelIdentity(a), model=models.get(key) ?? {total:0,tokens:Math.min(a.tokens,ceiling(a))};
          model.total+=a.total; model.tokens=Math.min(model.tokens,a.tokens,ceiling(a));
          if (!Number.isSafeInteger(model.total)) fail('ledger-overflow');
          models.set(key,model);
        }
        if ([...models.values()].some(a=>a.total>=a.tokens)) reason='model-cap-reached';
      }
    } catch(e) { reason=/^[a-z-]{1,80}$/.test(e.message)?e.message:'accounting-failed'; }
    if (reason) {
      const storageErrors=[];
      try {
        if(!writable) throw new Error('unavailable');
        if (!existsSync(stopFile)) { exclusive(stopFile,JSON.stringify({version:1,at:new Date(now()).toISOString(),reason,binding:p.binding,totals,evidenceErrors})+'\n'); flushDir(ledger); }
      } catch { storageErrors.push('stop-latch-not-durable'); }
      // Always attempt halt, even when the evidence volume cannot persist the latch.
      let result,error;
      try {
        checkBinding(p.binding,c);
        if(configHash!==digest(readFileSync(c.config,'utf8'))) fail('campaign-instance-changed');
        result=await halt(p.binding);
      } catch { error='supported-halt-refused-or-unknown'; }
      try {
        if(!writable) throw new Error('unavailable');
        const attempt=readdirSync(ledger).filter(n=>/^halt-[0-9]+\.json$/.test(n)).length+1;
        exclusive(join(ledger,`halt-${attempt}.json`),JSON.stringify({at:new Date(now()).toISOString(),ok:!error,result:result??null,error:error??null,storageError:storageErrors[0]??null,evidenceErrors})+'\n'); flushDir(ledger);
      } catch { storageErrors.push('stop-receipt-not-durable'); }
      if (storageErrors.length) throw Object.assign(new Error(storageErrors.join('; ')),{
        result:{stopped:true,reason,totals,halt:result??null,haltError:error??null,storageErrors,evidenceErrors},
      });
      if (error) throw Object.assign(new Error(error),{result:{stopped:true,reason,totals,evidenceErrors}});
      return {stopped:true,reason,totals,halt:result,evidenceErrors};
    }
    return {stopped:false,totals};
  } finally { if (locked) try { rmdirSync(lock); } catch { /* Preserve an unavailable lock; never recreate its parent. */ } }
}
export function status({ campaign,ledger }) {
  const p=loadLedger(ledger); validateCampaign(p.campaign); checkBinding(p.binding,p.campaign);
  const target=checkTarget({campaign,ledger});
  if(target.reason) fail(target.reason);
  outsideRuntime(campaign,{dataRoot:p.binding.dataRoot},p.campaign.config);
  const owner=campaignAt(campaign);
  if (owner.hash!==p.hash || JSON.stringify(owner.c)!==JSON.stringify(p.campaign)) fail('campaign-changed');
  const totals=p.campaign.caps.map(cap=>({...cap,total:journal(ledger,cap).total}));
  return { stopped:existsSync(join(ledger,'stop.json')),end:p.campaign.end,totals };
}
export async function serveBudget(opts, {interval=1000,signal,report=value=>console.log(JSON.stringify(value)),...deps}={}) {
  // Retain independent target/config/epoch evidence, but revalidate the actual
  // config, marker, root and store on EVERY tick, including just before halt.
  // Every tick uses the startup campaign and ledger paths, and the retained clone
  // records that ledger, so a caller that later mutates its options object cannot
  // move stop/halt receipts to another ledger (support r5).
  const fixed=Object.freeze({campaign:opts.campaign,ledger:opts.ledger});
  const target=checkTarget(fixed),retained=structuredClone({p:target.p,configHash:target.configHash,ledger:fixed.ledger});
  let stopReason=null;
  while(!signal?.aborted) {
    try {
      const result=await check(fixed,{...deps,stopReason,retained});
      if(result.stopped) stopReason=result.reason;
      report(result);
    } catch(e) {
      // Keep the owner service alive even if evidence storage or halt fails. A
      // changed/foreign target is revalidated and refused on every later tick.
      stopReason??=e.result?.reason ?? 'supervisor-check-failed';
      report({...e.result,error:e.message});
    }
    if(!signal?.aborted) await new Promise(resolve=>{
      const done=()=>{clearTimeout(timer);signal?.removeEventListener('abort',done);resolve();};
      const timer=setTimeout(done,interval);signal?.addEventListener('abort',done,{once:true});
    });
  }
}
export async function main(argv) {
  const command=argv.shift(),o=options(argv,['--campaign','--ledger','--interval']);
  const opts={campaign:o['--campaign'],ledger:o['--ledger']};
  if (command==='init') return initialize(opts);
  if (command==='status') return status(opts);
  if (command==='check') return check(opts);
  if (command!=='serve') fail('use init, check, serve or status');
  const interval=Number(o['--interval']??1); if (!Number.isFinite(interval) || interval<0.1 || interval>10) fail('interval must be 0.1..10 seconds');
  const controller=new AbortController();
  const stop=()=>controller.abort();
  for(const signal of ['SIGTERM','SIGINT']) process.once(signal,stop);
  try { await serveBudget(opts,{interval:interval*1000,signal:controller.signal}); }
  finally { for(const signal of ['SIGTERM','SIGINT']) process.removeListener(signal,stop); }
}
if (isMain(import.meta.url)) main(process.argv.slice(2)).then(r=>{if(r!==undefined)console.log(JSON.stringify(r,null,2));}).catch(e=>{console.error(e.message);process.exitCode=1;});