From f65faf9df8f565a624eb00945abc3eba502357a3 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Mon, 2 Nov 2026 19:02:45 -0600 Subject: [PATCH] fix(row 333 R4): a moved key quota keeps its count; orphan set on every refresh (#1831) Ruling 1: the bus runtime passes launchSettings a key lineage lookup read from its own credential records (ids only). A new daily key quota takes the earliest stored since among the old document's daily key quotas in its key's lineage, and now only when there is none. A reset still sets now, a caller-supplied since still refuses, and a quota removed and added back in a later write is new. A failed or malformed lookup refuses proxy-key-refused; a write that adds no new daily key quota never calls it. Ruling 2: every limits refresh updates the remembered orphan set, an empty set when no daily key quota is left, so a quota removed and added back logs again. In memory only: a restart is a fresh observation. Ruling 3: pollSettledAfterTick yields on the real setImmediate, captured at load, inside its deadline. Tests: p5-key-lineage (writer move counts 160, reset restarts, lookup failure, orphan remove and re-add), p5-runtime (exhausted on A, moved to B through the real runtime, refuses), p5-background (barrier with every timer mocked), token-limit-settings (earliest, no match, re-add, caller since, malformed answers, default refusal). READMEs: business, plugins, daily-quota, cliproxy-queue. --- packages/bus/src/limits.mjs | 21 ++++- packages/bus/src/runtime.mjs | 7 +- packages/bus/tests/p5-background.test.mjs | 15 ++++ packages/bus/tests/p5-key-lineage.test.mjs | 90 ++++++++++++++++--- packages/bus/tests/p5-runtime.test.mjs | 53 +++++++++++ packages/bus/tests/p5-telemetry-fixture.mjs | 7 +- packages/business/README.md | 17 ++++ packages/business/src/launch-settings.mjs | 36 +++++++- .../tests/token-limit-settings.test.mjs | 62 ++++++++++++- packages/plugins/README.md | 7 +- .../plugins/limit-check/daily-quota/README.md | 6 ++ .../plugins/usage/cliproxy-queue/README.md | 19 +++- 12 files changed, 307 insertions(+), 33 deletions(-) diff --git a/packages/bus/src/limits.mjs b/packages/bus/src/limits.mjs index 78e70180d..593deb79a 100644 --- a/packages/bus/src/limits.mjs +++ b/packages/bus/src/limits.mjs @@ -73,6 +73,9 @@ export function createLimits({ store, read, log, makeLoader = createLoader, keyF const loaders = new Map(); // Row 333 R3: an orphaned key quota names an id in no hired agent's lineage. It limits nothing // and refuses nothing. One value-free log line per leaf when it becomes orphaned, not per launch. + // R4: every refresh observes too, and a business with no daily key quota left remembers none, so a + // leaf removed and added back logs again. The set lives in memory: a bus start is a fresh + // observation and may log an unchanged orphan once more. const orphansLogged = new Map(); function noteOrphans(business, limits, lineages) { const ids = dailyKeyIds(limits).filter(id => !lineages.some(l => l.includes(id))); @@ -81,6 +84,13 @@ export function createLimits({ store, read, log, makeLoader = createLoader, keyF orphansLogged.set(business, new Set(ids)); return ids; } + // The orphaned ids now, [] when no daily key quota is set, or null when the lineages can't be + // read (the remembered set then stays as it was). + function observeOrphans(business, limits) { + if (!dailyKeyIds(limits).length) { orphansLogged.set(business, new Set()); return []; } + const lineages = lineagesFor(business); + return lineages ? noteOrphans(business, limits, lineages) : null; + } // The business's lineages, or null when they can't be read or are malformed. function lineagesFor(business) { let value; @@ -117,6 +127,7 @@ export function createLimits({ store, read, log, makeLoader = createLoader, keyF async function refresh(business) { let b; try { b = read(business); } catch { return; } + observeOrphans(business, b?.launch?.limits ?? {}); await loaderFor(business).load({ [KIND]: b?.plugins?.[KIND] ?? [] }); } @@ -248,10 +259,7 @@ export function createLimits({ store, read, log, makeLoader = createLoader, keyF function orphanedKeys(business) { let b; try { b = read(business); } catch { return null; } - const limits = b?.launch?.limits ?? {}; - if (!dailyKeyIds(limits).length) return []; - const lineages = lineagesFor(business); - return lineages ? noteOrphans(business, limits, lineages) : null; + return observeOrphans(business, b?.launch?.limits ?? {}); } return { refresh, mayLaunch, mustStop, report, orphanedKeys }; @@ -294,3 +302,8 @@ export function modelKeyLineages(records, business) { } return out; } +// Row 333 R4 (#1831): the settings writer's lookup. The lineage in `business` that holds keyId, or +// [] when none does. Ids only, from the bus's own records; the same checks refuse. +export function modelKeyLineage(records, business, keyId) { + return modelKeyLineages(records, business).find(l => l.includes(keyId)) ?? []; +} diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index 17660e58b..97fe6310d 100644 --- a/packages/bus/src/runtime.mjs +++ b/packages/bus/src/runtime.mjs @@ -36,7 +36,7 @@ import { inspectInterpreter, recordInterpreterStart } from './interpreter.mjs'; import { createPeerReader } from './peercred.mjs'; import { createUsage } from './usage.mjs'; import { createProxyUsage } from './proxy-usage.mjs'; -import { createLimits, modelUsageKey, modelKeyLineages } from './limits.mjs'; +import { createLimits, modelUsageKey, modelKeyLineages, modelKeyLineage } from './limits.mjs'; import { createProcessing } from '../../processing-control/src/coordinator.mjs'; import { discordProcessingHooks } from '../../discord/src/processing.mjs'; import { createContainerStopper } from '../../processing-control/src/containers.mjs'; @@ -340,7 +340,10 @@ export async function startBroker({ // plugins and plugins.set, logged in changes.jsonl with the actor, a missing module refused. const plugins=organization?pluginSettings({dir:organization.dir,rolesDir:organization.rolesDir,exists:hasModule}):null; const levels=organization?await harnessLevels():()=>null; - const settings=organization?launchSettings({...organization,sessions:()=>readSessions(dataRoot),levels}):null; + // Row 333 R4 (#1831): the writer's key lineage lookup, ids only, from this bus's own credential + // records. A daily key quota moved within one agent's lineage keeps its since and its count. + const keyLineage=(business,keyId)=>modelKeyLineage(credentials.status(),business,keyId); + const settings=organization?launchSettings({...organization,sessions:()=>readSessions(dataRoot),levels,keyLineage}):null; // Row 194 (#1692, brief item 8, row 199 D13): a fire or a delete revokes the agent's credentials // after the writer's checks and before its write, under the writer's lock (row 239). Fire revokes // each service of the entry and keeps the store directory, delete calls remove (sage-1593). A diff --git a/packages/bus/tests/p5-background.test.mjs b/packages/bus/tests/p5-background.test.mjs index d07dcc54a..46d2ed3ba 100644 --- a/packages/bus/tests/p5-background.test.mjs +++ b/packages/bus/tests/p5-background.test.mjs @@ -7,6 +7,8 @@ import { mkdirSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { startTimeOf } from '../../seat/src/proc.mjs'; import { telemetryFixture, WAIT_MS } from './p5-telemetry-fixture.mjs'; +// Captured before any test mocks timers: the outer bound on the R4 barrier test below. +const realSetTimeout=globalThis.setTimeout,realClearTimeout=globalThis.clearTimeout; async function nativeTurn(f) { const model=await f.bus.modelPreflight({business:'demo',role:'coder'}); const keyId=createHash('sha256').update(model.key).digest('hex').slice(0,12),dir=join(f.root,'native-transcript');mkdirSync(dir); @@ -63,3 +65,16 @@ test('R3 every telemetry fixture wait rejects by name at its real-timer deadline assert.ok(performance.now()-started{ + t.mock.timers.enable({apis:['Date','setInterval','setTimeout','setImmediate'],now:Date.now()}); + const f=await telemetryFixture(t);await f.start();await f.settleInitial(); + assert.equal(f.queueCalls(),1); + let bound; + const outer=new Promise((_,reject)=>{bound=realSetTimeout(()=>reject(new Error('pollSettledAfterTick did not settle')),WAIT_MS*2);}); + try{await Promise.race([f.pollSettledAfterTick(t),outer]);}finally{realClearTimeout(bound);} + assert.equal(f.queueCalls(),2); + assert.equal(f.bus.broker.request(f.human(),{verb:'usage'}).proxy.state,'available'); +}); diff --git a/packages/bus/tests/p5-key-lineage.test.mjs b/packages/bus/tests/p5-key-lineage.test.mjs index 07da8bc60..b6087b8c6 100644 --- a/packages/bus/tests/p5-key-lineage.test.mjs +++ b/packages/bus/tests/p5-key-lineage.test.mjs @@ -13,7 +13,7 @@ import { Credentials } from '../src/credentials.mjs'; import { ManagementClient } from '../../model-proxy/src/client.mjs'; import { ModelKeys, MAX_PREVIOUS_KEY_IDS } from '../src/model-keys.mjs'; import { Store } from '../src/store.mjs'; -import { createLimits, modelUsageKey, modelKeyLineages } from '../src/limits.mjs'; +import { createLimits, modelUsageKey, modelKeyLineages, modelKeyLineage } from '../src/limits.mjs'; import { createUsage } from '../src/usage.mjs'; import { launchSettings } from '../../business/src/launch-settings.mjs'; import { businessDoc, rolesCopy, systemFor, writeJson } from '../../business/tests/helpers.mjs'; @@ -160,24 +160,63 @@ test('R3 a leaf moved from A to B on rotation day still refuses: the tally sums assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: true }); }); -test('R3 the settings writer stamps a moved leaf with the move time, and from then the tally still sums A and B', async t => { - const f = await fixture(t); await f.keys.reconcile(); await f.keys.ensure('demo/coder'); - const A = f.record().keyId, org = join(f.root, 'org'), file = join(org, 'businesses/demo.json'); +// The settings writer on a scratch organization file, with the lineage lookup the runtime passes it. +function writerFor(f, lookup = (business, id) => modelKeyLineage(f.credentials.status(), business, id)) { + const org = join(f.root, 'org'), file = join(org, 'businesses/demo.json'); const doc = businessDoc(org, 'demo'); doc.plugins = { ...doc.plugins, 'limit-check': ['daily-quota'] }; writeJson(file, doc); - const settings = launchSettings({ dir: org, rolesDir: rolesCopy(org), system: systemFor(org), now: () => iso(Date.now()) }); + const settings = launchSettings({ dir: org, rolesDir: rolesCopy(org), system: systemFor(org), now: () => iso(Date.now()), keyLineage: lookup }); f.read = id => settings.current(id); + return { settings, file, sinceOf: id => JSON.parse(fs.readFileSync(file, 'utf8')).launch.limits.tokens.keys[id].since }; +} + +// R4 (#1831) ruling 1: the writer gives the moved leaf the earliest since in its lineage, not the move time. +test('R4 the settings writer keeps a moved leaf\'s since, so the tally sums A and B from the first set; a reset after the move restarts it', async t => { + const f = await fixture(t); await f.keys.reconcile(); await f.keys.ensure('demo/coder'); + const A = f.record().keyId, { settings, sinceOf } = writerFor(f); const leaf = { limit: 100, reset: '00:00', timeZone: 'UTC' }; let v = settings.read('demo'); v = settings.set('demo', { revision: v.revision, limits: { tokens: { keys: { [A]: leaf } } } }, 'jason'); + const first = sinceOf(A); + assert.equal(first, iso(START)); f.use(A, 60); f.bind('run-a', A); f.advance(60_000); await f.keys.rotate('demo/coder'); - const B = f.record().keyId, moved = iso(Date.now()); + const B = f.record().keyId; v = settings.set('demo', { revision: v.revision, limits: { tokens: { keys: { [B]: leaf } } } }, 'jason'); - assert.equal(JSON.parse(fs.readFileSync(file, 'utf8')).launch.limits.tokens.keys[B].since, moved); + assert.equal(sinceOf(B), first); f.advance(1000); f.use(A, 30, 'run-a'); f.use(B, 70); - // The 60 before the move are before the new leaf's since. The 100 after it, on A and B, count. - assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), refused(B, 100, 100)); - assert.equal(f.limits.report('demo')[0].count, 100); + // The 60 before the move count with the 100 after it, on A and B. + assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), refused(B, 100, 160)); + assert.equal(f.limits.report('demo')[0].count, 160); + // An explicit reset after the move sets now and restarts the count. + f.advance(1000); + v = settings.set('demo', { revision: v.revision, reset: [['tokens', 'keys', B]] }, 'jason'); + assert.equal(sinceOf(B), iso(Date.now())); + assert.equal(f.limits.report('demo')[0].count, 0); + assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: true }); +}); + +test('R4 the writer refuses a new key quota when the lineages can\'t be read, and a write that adds none never asks', async t => { + const f = await fixture(t); await f.keys.reconcile(); await f.keys.ensure('demo/coder'); await f.keys.ensure('demo/other'); + const A = f.record().keyId, other = f.record('demo/other'); + let asked = 0; + const { settings, file, sinceOf } = writerFor(f, (business, id) => { asked++; return modelKeyLineage(f.credentials.status(), business, id); }); + const leaf = { limit: 100, reset: '00:00', timeZone: 'UTC' }; + let v = settings.read('demo'); + v = settings.set('demo', { revision: v.revision, limits: { tokens: { keys: { [A]: leaf } } } }, 'jason'); + assert.equal(asked, 1); assert.equal(sinceOf(A), iso(START)); + // A key id in two agents' lineages: the lookup throws, and a write adding a key quota refuses unwritten. + f.credentials.writeModel('demo/other', { ...other, previous: [{ keyId: A, retiredAt: iso(START) }] }); + assert.throws(() => modelKeyLineage(f.credentials.status(), 'demo', A), { code: 'proxy-key-refused' }); + f.advance(1000); + const bytes = fs.readFileSync(file, 'utf8'); + assert.throws(() => settings.set('demo', { revision: v.revision, limits: { tokens: { keys: { [A]: { ...leaf }, [other.keyId]: { ...leaf } } } } }, 'jason'), + { code: 'proxy-key-refused' }); + assert.equal(fs.readFileSync(file, 'utf8'), bytes); + assert.equal(asked, 2); + // A write that adds no new key quota (a changed limit, a legacy account limit) saves and never asks. + v = settings.set('demo', { revision: v.revision, limits: { tokens: { keys: { [A]: { ...leaf, limit: 50 }, [hash('acct')]: { limit: 5 } } } } }, 'jason'); + assert.equal(asked, 2); + assert.equal(sinceOf(A), iso(START)); assert.equal(sinceOf(hash('acct')), iso(Date.now())); }); test('R3 lineage uniqueness: a key id in two agents\' lineages refuses', async t => { @@ -254,13 +293,16 @@ test('R3 fire and re-hire: the old leaf is orphaned, launches go ahead, the view await f.keys.prepareRevoke('demo/coder'); f.keys.completeRevoke('demo/coder'); assert.equal(f.record(), null); // Fired and not yet re-hired: no model key, so a key quota can't be matched. Its own code. + f.logs.length = 0; assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: false, check: null, code: 'key-quota-no-model-key' }); + // R4: the refresh before that ask already saw A in no lineage, so A's line comes at the fire. + assert.deepEqual(f.logs.map(e => e.event), ['limit-key-orphaned', 'limit-check-failed']); // The re-hire starts a new lineage. await f.keys.ensure('demo/coder'); const C = f.record().keyId; assert.equal(f.record().previous, undefined); assert.deepEqual(modelKeyLineages(f.credentials.status(), 'demo'), [[C]]); - f.logs.length = 0; + f.logs.splice(1); for (let i = 0; i < 3; i++) assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: true }); assert.deepEqual(f.logs, [{ event: 'limit-key-orphaned', business: 'demo', path: ['tokens', 'keys', A] }]); assert.deepEqual(f.limits.orphanedKeys('demo'), [A]); @@ -295,8 +337,32 @@ test('R3 ruling 2: an agent with no model key under a key quota refuses key-quot assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: false, check: null, code: 'key-quota-no-model-key' }); f.bind('run-x', hash('some-key')); assert.deepEqual(await f.limits.mustStop('demo', SUBJECT, { run: 'run-x' }), { stop: true, check: null, code: 'key-quota-no-model-key' }); - assert.deepEqual(f.logs.map(e => e.code), ['key-quota-no-model-key', 'key-quota-no-model-key']); + // R4: the refresh before the first ask logs the leaf no agent holds, once. + assert.deepEqual(f.logs.map(e => e.code ?? e.event), ['limit-key-orphaned', 'key-quota-no-model-key', 'key-quota-no-model-key']); // A business with no key quota never asks. f.doc = quotaDoc({}); assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: true }); }); + +// R4 (#1831) ruling 2: every refresh observes the orphans, and no daily key quota left remembers none. +test('R4 an orphaned leaf removed and added back logs twice; repeated refreshes of an unchanged orphan log once; a restart observes afresh', async t => { + const f = await fixture(t); await f.keys.reconcile(); await f.keys.ensure('demo/coder'); + const X = hash('fired-agent-key'), line = { event: 'limit-key-orphaned', business: 'demo', path: ['tokens', 'keys', X] }; + f.doc = quotaDoc({ [X]: quota(100) }); + assert.deepEqual(f.limits.orphanedKeys('demo'), [X]); + assert.deepEqual(f.logs, [line]); + // Removed: the refresh after the write sees no daily key quota. Added back: the next refresh logs again. + f.doc = quotaDoc({}); await f.limits.refresh('demo'); + f.doc = quotaDoc({ [X]: quota(100) }); await f.limits.refresh('demo'); + assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: true }); + assert.deepEqual(f.logs, [line, line]); + // Unchanged: refreshes, launches and views add nothing. + for (let i = 0; i < 3; i++) await f.limits.refresh('demo'); + assert.deepEqual(await f.limits.mayLaunch('demo', SUBJECT), { allow: true }); + assert.deepEqual(f.limits.orphanedKeys('demo'), [X]); + assert.deepEqual(f.logs, [line, line]); + // A bus restart remembers nothing: its first refresh logs the unchanged orphan once more, then no more. + f.restart(); + for (let i = 0; i < 3; i++) await f.limits.refresh('demo'); + assert.deepEqual(f.logs, [line, line, line]); +}); diff --git a/packages/bus/tests/p5-runtime.test.mjs b/packages/bus/tests/p5-runtime.test.mjs index ea49af28d..543ee5c65 100644 --- a/packages/bus/tests/p5-runtime.test.mjs +++ b/packages/bus/tests/p5-runtime.test.mjs @@ -55,3 +55,56 @@ test('P5 real runtime binds stack usage to issued key, daily stops at stack coun const history=readFileSync(join(config,'businesses/.history/demo/changes.jsonl'),'utf8'); for(const value of [model.key,'p5-fake-management'])assert.equal(JSON.stringify([view,history]).includes(value),false); }); + +// Row 333 R4 (#1831): the writer the runtime builds keeps a moved quota's since through the runtime's +// own key lineage lookup. The runtime keeps its ModelKeys inside, so the move follows a re-issue: the +// fake proxy stops listing A and the next preflight issues B with A among its earlier ids. +test('R4 real runtime: a quota spent on A and moved to B with no reset still refuses the next launch', async t => { + const root=mkdtempSync(join(tmpdir(),'p5-runtime-')),config=join(root,'config'),data=join(root,'data'); + mkdirSync(config,{mode:0o700});mkdirSync(data,{mode:0o700});const rolesDir=rolesCopy(config),doc=businessDoc(config,'demo'); + doc.plugins={'usage-source':['transcript'],'limit-check':['daily-quota'],harness:['pi']}; + writeJson(join(config,'businesses/demo.json'),doc);mkdirSync(join(config,'secrets'),{recursive:true,mode:0o700}); + writeFileSync(join(config,'secrets/proxy-management.key'),'p5-fake-management',{mode:0o600}); + new Registry(data); + let keys=[`msk-${hash(config)}-sentinel-${randomBytes(32).toString('base64url')}`]; + const server=http.createServer(async(req,res)=>{ + if(req.url==='/v8/management/config/access/api-keys') { + if(req.method==='PUT'){let b='';for await(const c of req)b+=c;keys=JSON.parse(b);res.end('{"status":"ok"}');} + else res.end(JSON.stringify(keys));return; + } + if(req.url==='/v8/management/observability/usage/queue?count=100'){res.end('[]');return;} + res.writeHead(keys.includes(req.headers.authorization?.slice(7))?200:401);res.end('{}'); + }); + await new Promise(r=>server.listen(0,'127.0.0.1',r));let bus; + t.after(async()=>{await bus?.close();await new Promise(r=>server.close(r));rmSync(root,{recursive:true,force:true});}); + bus=await startBroker({dataRoot:data,businesses:{demo:{id:'demo',human:'jason',plugins:doc.plugins,arbiters:{technical:'coder',delivery:'coder'},roles:{coder:{definition:'coder',authority:{withinRole:[],crossRole:[]}}}}}, + organization:{dir:config,rolesDir,system:systemFor(config)},proxy:{port:server.address().port,binary:'/fixture-unused'}}); + const stored=()=>JSON.parse(readFileSync(join(config,'businesses/demo.json'),'utf8')).launch.limits.tokens.keys; + const may=async()=>(await bus.mayLaunch({business:'demo',instance:'coder',provider:'fixture',model:'fake'})).allow; + const a=await bus.modelPreflight({business:'demo',role:'coder'}),A=hash(a.key); + const human=bus.broker.bindHuman({business:'demo',human:'jason',via:'cli',outsideAgent:true}); + const leaf={limit:2,reset:'00:00',timeZone:'UTC'}; + let limits=bus.broker.request(human,{verb:'limits'}); + limits=bus.broker.request(human,{verb:'limits.set',args:{revision:limits.revision,limits:{tokens:{keys:{[A]:{...leaf}}}}}}); + const since=stored()[A].since;assert.equal(await may(),true); + // The agent spends the whole allowance on A. + const dir=join(root,'transcript');mkdirSync(dir); + const {cap}=bus.bindLaunch({business:'demo',role:'coder',run:'r4-run',harness:'pi',modelKeyId:A,transcript:dir,pid:process.pid,startTime:startTimeOf(process.pid)}); + bus.broker.request(cap,{verb:'role.claim'}); + const {id}=bus.broker.request(human,{verb:'message.send',args:{to:'coder',body:'fake work'}}); + bus.broker.request(cap,{verb:'message.receive'}); + await bus.broker.request(cap,{verb:'usage.begin',args:{turn:id}}); + writeFileSync(join(dir,'s.jsonl'),JSON.stringify({type:'message',id:'fake',message:{role:'assistant',usage:{input:2,output:0,cacheWrite:0,cacheRead:0}}})+'\n'); + await bus.broker.request(cap,{verb:'usage.end',args:{turn:id}}); + assert.equal(await may(),false); + // The proxy loses A, so the next preflight issues B; then the human moves the unchanged quota to B. + keys=keys.filter(k=>k!==a.key); + const b=await bus.modelPreflight({business:'demo',role:'coder'}),B=hash(b.key);assert.notEqual(B,A); + await new Promise(r=>setTimeout(r,20)); // a fresh `now` would fall after the turn's record + limits=bus.broker.request(human,{verb:'limits'}); + bus.broker.request(human,{verb:'limits.set',args:{revision:limits.revision,limits:{tokens:{keys:{[B]:{...leaf}}}}}}); + assert.deepEqual(stored(),{[B]:{...leaf,since}}); + const report=bus.broker.request(human,{verb:'limits'}); + assert.equal(JSON.stringify(report).includes(a.key)||JSON.stringify(report).includes(b.key),false); + assert.equal(await may(),false); +}); diff --git a/packages/bus/tests/p5-telemetry-fixture.mjs b/packages/bus/tests/p5-telemetry-fixture.mjs index a6d0e5921..840f8dc52 100644 --- a/packages/bus/tests/p5-telemetry-fixture.mjs +++ b/packages/bus/tests/p5-telemetry-fixture.mjs @@ -8,8 +8,9 @@ import { startBroker } from '../src/runtime.mjs'; import { Registry } from '../../seat/src/session.mjs'; import { businessDoc, rolesCopy, systemFor, writeJson } from '../../business/tests/helpers.mjs'; // Row 333 R3 (#1831): every wait races a real-timer deadline, captured here before any test mocks -// setTimeout, so a missed event fails by name instead of hanging the file. -const realSetTimeout = globalThis.setTimeout, realClearTimeout = globalThis.clearTimeout; +// setTimeout, so a missed event fails by name instead of hanging the file. R4: the real setImmediate +// too, for the yield after a poll, which now runs inside that poll's deadline. +const realSetTimeout = globalThis.setTimeout, realClearTimeout = globalThis.clearTimeout, realSetImmediate = globalThis.setImmediate; export const WAIT_MS = 5000; function deadline(promise, what, ms) { let timer; @@ -72,6 +73,6 @@ export async function telemetryFixture(t, { foreign = false, hold = false, waitM settleInitial:()=>deadline(Promise.race([logSeen('proxy-usage-read-failed'),logSeen('proxy.usage.reconcile')]),'no proxy-usage-read-failed or proxy.usage.reconcile log line',waitMs), get bus(){return bus;},queueCalls:()=>state.calls.filter(p=>p.includes('/usage/queue')).length, human:()=>bus.broker.bindHuman({business:'demo',human:'jason',via:'cli',outsideAgent:true}), - async pollSettledAfterTick(t){const from=logs.length,next=state.calls.filter(p=>p.includes('/usage/queue')).length+1;t.mock.timers.tick(1000);await deadline(Promise.race([queueSeen(next),logSeen('proxy-usage-read-failed',from)]),`no usage queue read number ${next} or proxy-usage-read-failed log line`,waitMs);await new Promise(setImmediate);}, + async pollSettledAfterTick(t){const from=logs.length,next=state.calls.filter(p=>p.includes('/usage/queue')).length+1;t.mock.timers.tick(1000);await deadline(Promise.race([queueSeen(next),logSeen('proxy-usage-read-failed',from)]).then(()=>new Promise(resolve=>realSetImmediate(resolve))),`no usage queue read number ${next} or proxy-usage-read-failed log line`,waitMs);}, }; } diff --git a/packages/business/README.md b/packages/business/README.md index 6bd61b4c0..a131e2686 100644 --- a/packages/business/README.md +++ b/packages/business/README.md @@ -124,6 +124,23 @@ launcher's check. `launch.limits.tokens.daily` maps an account to name `Intl` accepts. A daily quota for any name but `launch.account.name` refuses the file, since it would count nothing. +`launch.limits.tokens.keys` holds token limits by key name. A leaf named by a +twelve-hex model key id with `{limit, since, reset, timeZone}` is a daily quota +on one agent's model key (row 333); a plain `{limit, since}` leaf stays a +cumulative account limit, even when its name is twelve hex characters. Only the +human sets or resets a model-key quota. The settings writer stamps `since` as +for every token limit (a new limit gets now, a changed one keeps its time, a +caller's different time refuses, a reset moves it to now), with one rule for a +new model-key quota: it takes the earliest `since` among the old file's +model-key quotas whose ids are in the same agent's key lineage, so moving a +quota to the agent's new key id keeps its count, and takes now only when none +of them holds one. A quota removed in one write and added back in a later one +is new and gets now. The bus host passes `keyLineage(business, keyId)`, which +answers ids only from the bus's own credential records. A writer built without +it, a lookup that throws, or an answer that is neither an empty list nor a list +of key ids holding the asked id refuses a write that adds a model-key quota with `proxy-key-refused`. +A write that adds none never calls it. + `launch.threadIdleSeconds` (optional, row 121) is how long a thread's session runs with nothing pending and no new message or open before the launcher stops it. Absent means 900 (`THREAD_IDLE_SECONDS` in diff --git a/packages/business/src/launch-settings.mjs b/packages/business/src/launch-settings.mjs index 45508c54e..25887f06f 100644 --- a/packages/business/src/launch-settings.mjs +++ b/packages/business/src/launch-settings.mjs @@ -6,6 +6,7 @@ import { usageCounts } from "./launch.mjs"; import { canonicalJson, isPlainObject, requireObject, rejectUnknownKeys } from "./util.mjs"; import { FALLBACK_VARS } from "./vars.mjs"; import { THREAD_DEFINITION, CREATIVE_DEFINITIONS } from "./vocabulary.mjs"; +import { isDailyModelQuota } from "../../plugins/src/model-quota.mjs"; // Row 213 (C2-08, Case 2 design 4, M10): a creative instance, its persona included, is the // human's. A write that touches one is accepted only when its actor is the business file's @@ -62,16 +63,39 @@ const at = (value, path) => path.reduce((v, part) => (own(v, part) ? v[part] : u // The writer, not the caller, sets a token limit's `since`. A new limit gets now; a changed or // unchanged one keeps its stored time; a caller-supplied time that differs from the stored one is // refused, so only a reset moves it. An entry that isn't an object is left for the schema to refuse. -function stamp(next, old, now) { +// Row 333 R4 (#1831): a new daily key quota carries its agent's count (carried, below). +function stamp(next, old, now, lineage) { for (const { path, entry } of tokenLeaves(next)) { if (!plain(entry)) continue; const before = at(old, path); const kept = plain(before) ? before.since : undefined; if (own(entry, "since") && entry.since !== kept) throw new OrganizationError("invalid-request"); - entry.since = kept ?? now; + entry.since = kept ?? carried(path, entry, old, now, lineage) ?? now; } } +// Row 333 R4 (#1831): a new tokens.keys leaf in the daily quota shape is a quota on one agent's model +// key. It takes the earliest stored `since` of the old document's daily key quotas whose ids are in +// that key's lineage, so moving a quota from A to B after a rotation or re-issue keeps the day's +// count, whether or not the same write removes A. With none it is new and gets now (undefined +// here). lineage(id) answers the ids of the agent holding id, read from the bus's own credential +// records, or [] when no agent holds it. A lookup that throws or answers malformed refuses the write. +// A quota removed in one write and added back in a later one finds no old leaf, so it gets now. +const KEY_ID = /^[a-f0-9]{12}$/; +function carried(path, entry, old, now, lineage) { + const [, kind, id] = path; + if (kind !== "keys" || !isDailyModelQuota(id, { ...entry, since: now })) return undefined; + let ids; + try { ids = lineage(id); } catch { throw new OrganizationError("proxy-key-refused"); } + if (!Array.isArray(ids) || ids.some(x => typeof x !== "string" || !KEY_ID.test(x)) || (ids.length && !ids.includes(id))) { + throw new OrganizationError("proxy-key-refused"); + } + const keys = at(old, ["tokens", "keys"]); + const times = Object.entries(plain(keys) ? keys : {}) + .filter(([key, value]) => ids.includes(key) && isDailyModelQuota(key, value)).map(([, value]) => value.since); + return times.sort((a, b) => Date.parse(a) - Date.parse(b))[0]; +} + // A reset names token limit paths, e.g. ["tokens", "instances", "coder-1"]; each gets since = now. // Any other path, a concurrency limit's included, is refused. function reset(limits, paths, now) { @@ -122,7 +146,11 @@ export function checkRoutes({ proposed, old = null, levels }) { const ROUTE_FIELDS = Object.freeze(["harness", "model", "thinking"]); // levels(name): the thinking levels harness module `name` applies, or null (checkRoutes). -export function launchSettings({ dir, rolesDir, system, sessions = () => [], now = () => new Date().toISOString(), levels = () => null }) { +// Row 333 R4 (#1831): keyLineage(business, keyId) answers the key ids in the lineage of the agent +// that holds keyId (carried, above). The bus runtime passes one read from its credential records. +// Without it, a write that adds a daily key quota refuses proxy-key-refused; no other write asks. +export function launchSettings({ dir, rolesDir, system, sessions = () => [], now = () => new Date().toISOString(), levels = () => null, + keyLineage = () => { throw new OrganizationError("proxy-key-refused"); } }) { const writer = organizationWriter({ dir, rolesDir }); // The validated business as the file stands now (row 125: the limit checks read it at each ask). @@ -182,7 +210,7 @@ export function launchSettings({ dir, rolesDir, system, sessions = () => [], now const at = now(); if (Object.hasOwn(args, "limits")) { next.launch.limits = structuredClone(args.limits); - stamp(next.launch.limits, old.document.launch.limits, at); + stamp(next.launch.limits, old.document.launch.limits, at, key => keyLineage(id, key)); } if (Object.hasOwn(args, "reset")) reset(next.launch.limits, args.reset, at); if (Object.hasOwn(args, "models")) { diff --git a/packages/business/tests/token-limit-settings.test.mjs b/packages/business/tests/token-limit-settings.test.mjs index aae75ba9f..1b18a8fcd 100644 --- a/packages/business/tests/token-limit-settings.test.mjs +++ b/packages/business/tests/token-limit-settings.test.mjs @@ -10,13 +10,14 @@ import {launchSettings} from '../src/launch-settings.mjs'; const roots=[];after(()=>roots.forEach(root=>rmSync(root,{recursive:true,force:true}))); const T0='2026-10-18T07:00:00.000Z',T1='2026-10-18T08:00:00.000Z',T2='2026-10-18T09:00:00.000Z'; -function fixture(){ +// keyLineage, when given, is the writer's lineage lookup (row 333 R4); without it the writer's default refuses. +function fixture(keyLineage){ const root=makeRoot();roots.push(root);const rolesDir=rolesCopy(root),doc=businessDoc(root,'acme'); for(const r of Object.values(doc.roles))r.vars={...r.vars,model:'sonnet'}; doc.launch={by:'chief',instances:['coder','reviewer'],models:{sonnet:{provider:'anthropic',model:'claude-sonnet-5-5'}},limits:{}}; const file=join(root,'businesses/acme.json');writeJson(file,doc); const system={environment:'development',dataRoot:join(root,'data'),'execution.backend':'docker','execution.provider':'anthropic','execution.model':'sonnet','execution.adapter':'pi'}; - let clock=T0;const settings=launchSettings({dir:root,rolesDir,system,now:()=>clock}); + let clock=T0;const settings=launchSettings({dir:root,rolesDir,system,now:()=>clock,...(keyLineage?{keyLineage}:{})}); return {root,rolesDir,doc,file,settings,tick:at=>{clock=at;}}; } const stored=f=>JSON.parse(readFileSync(f.file,'utf8')).launch.limits.tokens; @@ -152,7 +153,7 @@ test('REQ-LIMIT-2 the writer stamps a daily quota\'s since, keeps its reset and }); test('P5 per-model-key quotas are unset by default, accept digit-leading IDs and only the human changes or resets them', () => { - const f=fixture(),key='012345abcdef'; let view=f.settings.read('acme'); + const f=fixture(()=>[]),key='012345abcdef'; let view=f.settings.read('acme'); assert.equal(view.limits.tokens,undefined); const limits={tokens:{keys:{[key]:{limit:20,reset:'05:00',timeZone:'America/Chicago'}}}}; view=f.settings.set('acme',{revision:view.revision,limits},'jason'); @@ -168,3 +169,58 @@ test('P5 per-model-key quotas are unset by default, accept digit-leading IDs and view=f.settings.set('acme',{revision:view.revision,limits:{}},'jason'); assert.equal(view.limits.tokens,undefined); }); + +// Row 333 R4 (#1831): a new daily key quota takes the earliest since of the old document's daily key +// quotas in its key's lineage. A, B and C are one agent's keys, oldest first; X is no agent's. +const A='aaaaaaaaaaa1',B='bbbbbbbbbbb2',C='ccccccccccc3',X='ddddddddddd4',T3='2026-10-18T10:00:00.000Z'; +const daily=(limit=20)=>({limit,reset:'05:00',timeZone:'America/Chicago'}); +const lineageOf=(business,id)=>{assert.equal(business,'acme');return [C,B,A].includes(id)?[C,B,A]:[];}; +test('R4 a moved daily key quota keeps the earliest since in its lineage; a reset, a new key and a re-add get now; a caller time refuses',()=>{ + const f=fixture(lineageOf);let view=f.settings.read('acme'); + view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[A]:daily()}}}},'jason'); + assert.equal(stored(f).keys[A].since,T0); + // B joins while A stays: it carries A's time. Then B alone is reset, so A and B differ. + f.tick(T1);view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[A]:daily(),[B]:daily(30)}}}},'jason'); + assert.equal(stored(f).keys[B].since,T0); + view=f.settings.set('acme',{revision:view.revision,reset:[['tokens','keys',B]]},'jason'); + assert.deepEqual([stored(f).keys[A].since,stored(f).keys[B].since],[T0,T1]); + // The move to C drops both: C takes the earlier of the two, not the later. X matches no agent: now. + f.tick(T2);view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[C]:daily(),[X]:daily()}}}},'jason'); + assert.deepEqual(stored(f).keys,{[C]:{...daily(),since:T0},[X]:{...daily(),since:T2}}); + // A caller-supplied since refuses, on a moved leaf too, even when it is the time the writer would carry. + const bytes=readFileSync(f.file,'utf8'); + for(const keys of [{[C]:daily(),[X]:daily(),[B]:{...daily(),since:T0}},{[C]:daily(),[X]:daily(),[A]:{...daily(),since:T2}},{[C]:{...daily(),since:T1},[X]:daily()}]) + assert.throws(()=>f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys}}},'jason'),{code:'invalid-request'},JSON.stringify(keys)); + assert.equal(readFileSync(f.file,'utf8'),bytes); + // An explicit reset after the move sets now, in the same write as a move or in its own. + f.tick(T3);view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[B]:daily(),[X]:daily()}}},reset:[['tokens','keys',B]]},'jason'); + assert.equal(stored(f).keys[B].since,T3); + // Removed in one write and added back in a later one: a new quota, so now. + view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[X]:daily()}}}},'jason'); + f.tick('2026-10-18T11:00:00.000Z');view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[X]:daily(),[A]:daily()}}}},'jason'); + assert.equal(stored(f).keys[A].since,'2026-10-18T11:00:00.000Z'); +}); + +test('R4 the writer asks the lineage only for a new daily key quota, and a lookup that fails or answers malformed refuses',()=>{ + let asked=[];const answer={value:null}; + const f=fixture((business,id)=>{asked.push(id);if(answer.value instanceof Error)throw answer.value;return answer.value??lineageOf(business,id);}); + let view=f.settings.read('acme'); + // Writes adding no new daily key quota never ask: account limits, a hex-named legacy limit, a changed quota. + view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{instances:{coder:{limit:5}},keys:{k1:{limit:5},[X]:{limit:5}}}}},'jason'); + assert.deepEqual(asked,[]); + view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[A]:daily()}}}},'jason'); + assert.deepEqual(asked,[A]); + f.tick(T1);view=f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[A]:daily(99)}}}},'jason'); + view=f.settings.set('acme',{revision:view.revision,reset:[['tokens','keys',A]]},'jason'); + assert.deepEqual(asked,[A]); + const bytes=readFileSync(f.file,'utf8'); + for(const bad of [Object.assign(new Error('records unreadable'),{code:'credential-format'}),new Error('plain'),'nope',[42],['ABCDEFABCDEF'],[A],{0:B}]){ + answer.value=bad; + assert.throws(()=>f.settings.set('acme',{revision:view.revision,limits:{tokens:{keys:{[A]:daily(99),[B]:daily()}}}},'jason'),{code:'proxy-key-refused'},String(bad)); + } + assert.equal(readFileSync(f.file,'utf8'),bytes); + // Without a lookup, the default refuses a new daily key quota and nothing else. + const g=fixture();let gv=g.settings.read('acme'); + gv=g.settings.set('acme',{revision:gv.revision,limits:{tokens:{instances:{coder:{limit:5}},keys:{[X]:{limit:5}}}}},'jason'); + assert.throws(()=>g.settings.set('acme',{revision:gv.revision,limits:{tokens:{keys:{[X]:{limit:5},[A]:daily()}}}},'jason'),{code:'proxy-key-refused'}); +}); diff --git a/packages/plugins/README.md b/packages/plugins/README.md index 8cfbcbbd9..19f11e0ce 100644 --- a/packages/plugins/README.md +++ b/packages/plugins/README.md @@ -301,8 +301,11 @@ Row 333 adds daily quotas on model keys, `launch.limits.tokens.keys.<12-hex key id>` with the same shape. A key quota follows its agent's key lineage across rotation and re-issue, matching and counting every id in it. After a fire and re-hire the old id is in no lineage: the quota limits nothing and the Console -usage view lists it as matching no agent. See -[the usage module contract](usage/cliproxy-queue/README.md). +usage view lists it as matching no agent. The limits writer gives a new key +quota the earliest `since` among the old file's key quotas in its key's lineage +(row 333 R4), so moving a quota to the agent's new key id keeps its count; a +`reset` restarts it, and a quota removed and added back in a later write is new +and gets now. See [the usage module contract](usage/cliproxy-queue/README.md). ## Human project/thread channels diff --git a/packages/plugins/limit-check/daily-quota/README.md b/packages/plugins/limit-check/daily-quota/README.md index b4533a114..066aa136e 100644 --- a/packages/plugins/limit-check/daily-quota/README.md +++ b/packages/plugins/limit-check/daily-quota/README.md @@ -45,6 +45,12 @@ leaf whose id is in no lineage (the agent was fired) counts its own id only and limits no launch. Missing or malformed `lineages`, or an id in two lineages, throws `bad-lineages`; account quotas never need them. +`since` is the settings writer's, not this module's (row 333 R4). A new key +quota takes the earliest `since` among the old document's key quotas in its +lineage, so a quota moved to the agent's new key id keeps its count. A reset +sets now, and so does adding a quota that no earlier one in its lineage holds, +including one removed in an earlier write and added back. + The module is pure: it reads no file and no store. A bad `since`, record `at` or `now` throws `bad-time`, a bad amount `bad-record`, records that aren't a list `bad-records`, and a quota without a whole `limit`, an `HH:MM` `reset` diff --git a/packages/plugins/usage/cliproxy-queue/README.md b/packages/plugins/usage/cliproxy-queue/README.md index 741e27b54..db3938c7c 100644 --- a/packages/plugins/usage/cliproxy-queue/README.md +++ b/packages/plugins/usage/cliproxy-queue/README.md @@ -55,6 +55,19 @@ Console usage view lists it, and the bus logs one value-free `limit-key-orphaned` line when it becomes orphaned. An agent with no model key under a key quota refuses `key-quota-no-model-key`; an id in two agents' lineages, or a run bound to an id the agent no longer holds, refuses -`proxy-key-refused`. The human writer stamps a new `since` on a renamed leaf, so -moving a quota to a new key id restarts its count; after a rotation no move is -needed. +`proxy-key-refused`. The 64 is a limit: once an id falls out of the newest 64, +its quota is orphaned while the agent stays hired. + +After a rotation no move is needed. A quota moved to the agent's new key id +keeps its count (row 333 R4): the human writer gives a new key quota the +earliest `since` among the old file's key quotas whose ids are in the same +agent's lineage, asking the bus for that lineage (ids only, from its own +credential records), and stamps now only when none of them holds one. Only an +explicit reset restarts the count. A quota removed in one write and added back +in a later one is a new quota and gets now. If the lineage can't be read, a +write that adds a key quota refuses `proxy-key-refused`. + +The orphan line is logged at the first limits refresh that finds the quota +orphaned, which can be at boot or before a launch refusal. A quota removed and +added back logs again. The bus keeps what it has logged in memory only, so a +restart is a fresh observation and a quota still orphaned logs once more.