#!/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. if (!['3s','3t','3u','3v'].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!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;});