diff --git a/packages/bus/src/proxy-usage.mjs b/packages/bus/src/proxy-usage.mjs index acebef5c3..bc5dec49b 100644 --- a/packages/bus/src/proxy-usage.mjs +++ b/packages/bus/src/proxy-usage.mjs @@ -12,6 +12,9 @@ export function createProxyUsage({ store, read, client, records, log, now = Date const loader = makeLoader({ log }); let due = 0, previous = '', pending = null, closed = false; const states = new Map(); + // #778: per enabled business, the batches taken off the proxy queue and then not written, since + // this bus started. Telemetry only: the events in a dropped batch are gone and no quota reads this. + const dropped = new Map(); function configured() { return Object.entries(read()).filter(([, b]) => b.plugins?.usage === 'cliproxy-queue').map(([business, b]) => { checkUsageSettings(b.usage ?? {}); @@ -27,13 +30,22 @@ export function createProxyUsage({ store, read, client, records, log, now = Date if (signature !== previous) { previous = signature; due = 0; } if (now() < due) return; due = now() + Math.min(...enabled.map(b => b.pollSeconds)) * 1000; + // drained: the proxy has given this batch up, because module.read returned or because the client + // refused a 2xx answer it had already taken (proxy-client-response). Down, redirected or non-2xx + // answers drained nothing. written: the rows are in. + let drained = false, written = false; try { await loader.load({ usage: 'cliproxy-queue' }); const module = loader.modules('usage')[0], management = await client(); if (!module || !management) throw Error('unavailable'); // Row 333 R3 (#1831): the module gets the one read it needs and nothing else. No path from // its argument reaches getKeys, putKeys, the OAuth calls or any other management call. - const events = await module.read(Object.freeze({ readUsageQueue: () => management.readUsageQueue() })); + const readUsageQueue = async () => { + try { return await management.readUsageQueue(); } + catch (error) { if (error?.code === 'proxy-client-response') drained = true; throw error; } + }; + const events = await module.read(Object.freeze({ readUsageQueue })); + drained = true; const owners = new Map(); for (const record of records()) { if (record.service !== 'model') continue; @@ -64,10 +76,14 @@ export function createProxyUsage({ store, read, client, records, log, now = Date JSON.stringify({ modelKeyId: e.keyId, failed: e.failed, statusCode: e.statusCode }), e.input, e.output, e.cacheWrite, e.cacheRead); } }); + written = true; for (const keyId of unknown) log({ action: 'proxy.usage.reconcile', keyId, code: 'proxy-usage-unknown-key' }); for (const { business } of enabled) states.set(business, 'available'); } catch { - for (const { business } of enabled) states.set(business, 'unavailable'); + for (const { business } of enabled) { + states.set(business, 'unavailable'); + if (drained && !written) dropped.set(business, (dropped.get(business) ?? 0) + 1); + } log({ action: 'proxy.usage', code: 'proxy-usage-read-failed' }); } } @@ -92,7 +108,7 @@ export function createProxyUsage({ store, read, client, records, log, now = Date } } const finish = map => [...map.values()].map(e => ({ ...e, ...Object.fromEntries(KINDS.map(k => [k, plain(e[k])])) })); - return { source: 'proxy', state: states.get(business) ?? 'unread', byKey: finish(byKey), byAccount: finish(byAccount) }; + return { source: 'proxy', state: states.get(business) ?? 'unread', droppedBatches: dropped.get(business) ?? 0, byKey: finish(byKey), byAccount: finish(byAccount) }; } return { tick, view, async close() { closed = true; await pending; } }; } diff --git a/packages/bus/tests/proxy-usage.test.mjs b/packages/bus/tests/proxy-usage.test.mjs index ddffa2406..2b0988a7d 100644 --- a/packages/bus/tests/proxy-usage.test.mjs +++ b/packages/bus/tests/proxy-usage.test.mjs @@ -15,15 +15,15 @@ const id = s => createHash('sha256').update(s).digest('hex').slice(0,12); const key='msk-fake-client-value',source='fake-upstream-value',keyId=id(key); async function fixture(t,{makeLoader}={}) { const root=mkdtempSync(join(tmpdir(),'proxy-usage-')),store=new Store(root),logs=[],calls=[]; - let events=[],now=0,fail=false; + let events=[],now=0,fail=false,noClient=false,recs=[{instance:'demo/coder',service:'model',keyId,retiring:[]}]; const server=http.createServer((req,res)=>{calls.push(req.url);res.writeHead(fail?500:200);res.end(JSON.stringify(events));events=[];}); await new Promise(r=>server.listen(0,'127.0.0.1',r)); const client=new ManagementClient({baseUrl:`http://127.0.0.1:${server.address().port}`,key:'fake-management',receipt:e=>logs.push(e)}); let doc={plugins:{},launch:{limits:{tokens:{keys:{[keyId]:{limit:1,since:'2026-01-01T00:00:00.000Z'}}}}}}; - const reader=createProxyUsage({store,read:()=>({demo:doc}),client:()=>client,...(makeLoader?{makeLoader}:{}),records:()=>[{instance:'demo/coder',service:'model',keyId,retiring:[]}],log:e=>logs.push(e),now:()=>now}); + const reader=createProxyUsage({store,read:()=>({demo:doc}),client:()=>noClient?null:client,...(makeLoader?{makeLoader}:{}),records:()=>recs,log:e=>logs.push(e),now:()=>now}); t.after(async()=>{await reader.close();await new Promise(r=>server.close(r));store.close();rmSync(root,{recursive:true,force:true});}); const event=(over={})=>({api_key:key,source,failed:true,fail:{status_code:503,body:source},tokens:{input_tokens:9,output_tokens:2},...over}); - return {store,logs,calls,reader,client,event,setEvents:v=>{events=v;},clock:v=>{now=v;},enable:(pollSeconds)=>{doc={...doc,plugins:{usage:'cliproxy-queue','limit-check':['token-limit']},...(pollSeconds===undefined?{}:{usage:{cliproxyQueue:{pollSeconds}}})};},doc:()=>doc,fail:()=>{fail=true;}}; + return {store,logs,calls,reader,client,event,setEvents:v=>{events=v;},clock:v=>{now=v;},enable:(pollSeconds)=>{doc={...doc,plugins:{usage:'cliproxy-queue','limit-check':['token-limit']},...(pollSeconds===undefined?{}:{usage:{cliproxyQueue:{pollSeconds}}})};},doc:()=>doc,fail:()=>{fail=true;},noClient:v=>{noClient=v;},setRecords:v=>{recs=v;}}; } test('P5 usage plugin off by default, default poll is 15 seconds and configured polls skip early ticks',async t=>{ const f=await fixture(t);await f.reader.tick();assert.equal(f.calls.length,0);assert.equal(f.reader.view('demo'),null); @@ -109,3 +109,49 @@ test('R3 the usage module is handed one function, readUsageQueue, and no path to assert.deepEqual(called,['readUsageQueue']); assert.equal(f.reader.view('demo').byKey[0].failures,1); }); + +// #778 (Vikunja): the queue read drains the proxy, so a batch that is then not written is lost. The count +// is per enabled business, in memory since the bus started, and telemetry only. +test('#778 a drained batch that is not written counts as dropped; a poll that drained nothing does not; a good poll keeps the count',async t=>{ + const f=await fixture(t);f.enable(); + const usage=await createUsage({store:f.store,businesses:{demo:f.doc()},log:()=>{},proxyView:b=>f.reader.view(b)}); + assert.equal(f.reader.view('demo').droppedBatches,0); + // An invalid batch: the client refuses the 2xx answer whole, and the proxy has already given the batch up. + f.setEvents([f.event(),f.event({source:'malformed-upstream-canary',tokens:{input_tokens:-1}})]);await f.reader.tick(); + assert.equal(f.reader.view('demo').droppedBatches,1);assert.equal(f.reader.view('demo').state,'unavailable'); + // An ambiguous owner: one key id held by two agents. + f.setRecords([{instance:'demo/coder',service:'model',keyId,retiring:[]},{instance:'demo/other',service:'model',keyId,retiring:[]}]); + f.setEvents([f.event()]);f.clock(15000);await f.reader.tick(); + assert.equal(f.reader.view('demo').droppedBatches,2); + // No client before the read: nothing was drained, so nothing was dropped. + f.noClient(true);f.setEvents([f.event()]);f.clock(30000);await f.reader.tick(); + assert.equal(f.reader.view('demo').droppedBatches,2);assert.equal(f.reader.view('demo').state,'unavailable'); + // A good poll sets the state back and leaves the count alone. The event set while the client was missing is still queued. + f.noClient(false);f.setRecords([{instance:'demo/coder',service:'model',keyId,retiring:[]}]);f.clock(45000);await f.reader.tick(); + assert.equal(f.reader.view('demo').state,'available');assert.equal(f.reader.view('demo').droppedBatches,2); + assert.equal(f.reader.view('demo').byKey[0].requests,1); + // The usage view carries it, and a log line names no event content. + assert.equal(usage.view('demo').proxy.droppedBatches,2); + for(const secret of [key,source,'malformed-upstream-canary'])assert.equal(JSON.stringify([f.logs,usage.view('demo')]).includes(secret),false); + assert.deepEqual(f.logs.filter(e=>e.action==='proxy.usage').map(e=>e.code),['proxy-usage-read-failed','proxy-usage-read-failed','proxy-usage-read-failed']); +}); +test('#778 a batch the module returns in a shape the bus refuses is a drop',async t=>{ + const makeLoader=opts=>{const loader=createLoader(opts);return {...loader,modules:kind=>loader.modules(kind).map(m=>kind==='usage'&&m?{interfaceVersion:m.interfaceVersion,read:async arg=>{await arg.readUsageQueue();return [{keyId:'not-a-key-id',accountId:id(source)}];}}:m)};}; + const f=await fixture(t,{makeLoader});f.enable();f.setEvents([f.event()]);await f.reader.tick(); + assert.equal(f.reader.view('demo').state,'unavailable');assert.equal(f.reader.view('demo').droppedBatches,1); + assert.equal(f.store.all('SELECT * FROM usage_records').length,0); +}); +test('#778 a read the proxy refused or never answered drained nothing and is not a drop',async t=>{ + const f=await fixture(t);f.enable();f.fail();await f.reader.tick(); + assert.equal(f.reader.view('demo').state,'unavailable');assert.equal(f.reader.view('demo').droppedBatches,0); +}); +test('#778 a failed transaction drops the batch for every enabled business',async t=>{ + const f=await fixture(t);f.enable(); + const broken={all:(...a)=>f.store.all(...a),run:(...a)=>f.store.run(...a),transaction:()=>{throw new Error('disk full');}}; + const both={demo:f.doc(),other:f.doc()}; + const reader=createProxyUsage({store:broken,read:()=>both,client:()=>f.client,records:()=>[{instance:'demo/coder',service:'model',keyId,retiring:[]}],log:e=>f.logs.push(e),now:()=>0}); + t.after(()=>reader.close()); + f.setEvents([f.event()]);await reader.tick(); + for(const business of ['demo','other']){assert.equal(reader.view(business).droppedBatches,1);assert.equal(reader.view(business).state,'unavailable');} + assert.equal(f.store.all('SELECT * FROM usage_records').length,0); +}); diff --git a/packages/plugins/usage/cliproxy-queue/README.md b/packages/plugins/usage/cliproxy-queue/README.md index db3938c7c..88b1db1af 100644 --- a/packages/plugins/usage/cliproxy-queue/README.md +++ b/packages/plugins/usage/cliproxy-queue/README.md @@ -30,7 +30,14 @@ Telemetry uses the existing append-only usage table with source excludes it before every limit check and native usage total. Proxy totals may be incomplete (60-second upstream retention and a 100-event drain); the Console labels them proxy-reported and reports the reader's availability. They never -independently enforce quotas. +independently enforce quotas. A batch the proxy has already given up and the +bus then does not save (the client refuses its answer, the bus refuses its +shape or an ambiguous key owner, or the write fails) is lost. The reader counts +these per business since the bus started and the usage view returns the count +as `droppedBatches` (0 when none); the Console shows it above 0. A read that +failed before the proxy answered, or that the proxy refused, drained nothing +and is not counted. The count is in memory, restarts at 0 and is not a quota +input. Native transcript counts carry the immutable run's safe `modelKeyId` in the usage cursor, preserved at turn begin and end. The Console exposes `byKey`. diff --git a/packages/webui/src/public/bus.js b/packages/webui/src/public/bus.js index f7dd6248d..70b275b62 100644 --- a/packages/webui/src/public/bus.js +++ b/packages/webui/src/public/bus.js @@ -759,7 +759,10 @@ } // Row 333 R3 (#1831): a daily key quota whose key id no hired agent holds, now or before. const orphans = Array.isArray(value.orphanedKeyQuotas) && value.orphanedKeyQuotas.length ? `` : ''; - const proxy = value.proxy ? `

Proxy-reported usage

Optional proxy telemetry never enforces a quota. Status: ${txt(value.proxy.state)}. These counts cover drained events and may be incomplete.

${(value.proxy.byAccount || []).map(row => `${['input','output','cacheWrite','cacheRead','requests','failures'].map(k => ``).join('')}`).join('')}
Proxy account IDInputOutputCache writeCache readRequestsFailures
${txt(row.accountId)}${txt(row[k])}
` : ''; + // #778: batches the proxy gave up and the bus did not save, since the bus started. Shown only above 0. + const lost = Number.isSafeInteger(value.proxy?.droppedBatches) ? value.proxy.droppedBatches : 0; + const dropped = lost > 0 ? `

${txt(lost)} ${lost === 1 ? 'batch' : 'batches'} of proxy usage events could not be saved since the bus started. The events in a dropped batch are lost and are not in these totals.

` : ''; + const proxy = value.proxy ? `

Proxy-reported usage

Optional proxy telemetry never enforces a quota. Status: ${txt(value.proxy.state)}. These counts cover drained events and may be incomplete.

${dropped}
${(value.proxy.byAccount || []).map(row => `${['input','output','cacheWrite','cacheRead','requests','failures'].map(k => ``).join('')}`).join('')}
Proxy account IDInputOutputCache writeCache readRequestsFailures
${txt(row.accountId)}${txt(row[k])}
` : ''; return `

Token usage

Best reading of session-editable transcripts, not proof. Unknown turns may have used more tokens. Account is unassigned until account tracking is configured.

${rows.join('')}
SubjectInputOutputCache writeCache readUnknown turns
${orphans}${proxy}
`; } // Row 155 (#1649, Q169): each assistant instance's settings, its seat name and a stop. A save diff --git a/packages/webui/tests/usage-render.test.mjs b/packages/webui/tests/usage-render.test.mjs index ed0b159c9..1b10f86d4 100644 --- a/packages/webui/tests/usage-render.test.mjs +++ b/packages/webui/tests/usage-render.test.mjs @@ -28,3 +28,15 @@ test('R3 Console marks an orphaned key quota by id and says it limits nothing',( const hostile=render({total:counts,byKey:[],orphanedKeyQuotas:['']}); assert.doesNotMatch(hostile,/{ + const native={total:counts,byKey:[]},proxy=(over={})=>({source:'proxy',state:'available',byAccount:[],...over}); + const one=render({...native,proxy:proxy({droppedBatches:1})}); + assert.match(one,/

1 batch of proxy usage events could not be saved since the bus started\. The events in a dropped batch are lost and are not in these totals\.<\/p>/); + assert.ok(one.indexOf('token-usage-proxy-dropped')>one.indexOf('Proxy-reported usage')); + assert.match(render({...native,proxy:proxy({droppedBatches:3})}),/>3 batches of proxy usage events could not be saved/); + for(const over of [{droppedBatches:0},{},{droppedBatches:-1},{droppedBatches:'2'},{droppedBatches:1.5},{droppedBatches:null}]){ + const page=render({...native,proxy:proxy(over)});assert.match(page,/Proxy-reported/);assert.doesNotMatch(page,/token-usage-proxy-dropped|dropped batch/); + } + assert.doesNotMatch(render({...native,proxy:null}),/token-usage-proxy-dropped/); +});