feat(row 333): count dropped proxy usage batches and show them in the Console (Vikunja #778)

The usage queue read drains the proxy. A batch that is then not written (the
client or the bus refuses its shape, an ambiguous key owner, a failed write) is
lost, and the Console showed proxy totals with no sign of it. The reader now
counts these per enabled business since the bus started, the proxy view returns
the count as droppedBatches, and the Console proxy section says so above 0.
A read that failed before the proxy answered or that it refused drained nothing
and is not counted. Telemetry only; no quota reads it. The log line stays
proxy-usage-read-failed.

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; } };
 }

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);
+});

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`.

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 ? `<ul id="token-usage-orphans" aria-label="Key quotas that match no agent">${value.orphanedKeyQuotas.map(id => `<li>Key quota ${txt(id)} matches no agent, so it limits nothing. A fire and re-hire gives the agent a new key; set the quota on the new key id.</li>`).join('')}</ul>` : '';
-    const proxy = value.proxy ? `<section aria-label="Proxy-reported usage"><h3>Proxy-reported usage</h3><p>Optional proxy telemetry never enforces a quota. Status: ${txt(value.proxy.state)}. These counts cover drained events and may be incomplete.</p><div class="table-wrap"><table><thead><tr><th>Proxy account ID</th><th>Input</th><th>Output</th><th>Cache write</th><th>Cache read</th><th>Requests</th><th>Failures</th></tr></thead><tbody>${(value.proxy.byAccount || []).map(row => `<tr><th scope="row">${txt(row.accountId)}</th>${['input','output','cacheWrite','cacheRead','requests','failures'].map(k => `<td>${txt(row[k])}</td>`).join('')}</tr>`).join('')}</tbody></table></div></section>` : '';
+    // #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 ? `<p id="token-usage-proxy-dropped" role="status">${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.</p>` : '';
+    const proxy = value.proxy ? `<section aria-label="Proxy-reported usage"><h3>Proxy-reported usage</h3><p>Optional proxy telemetry never enforces a quota. Status: ${txt(value.proxy.state)}. These counts cover drained events and may be incomplete.</p>${dropped}<div class="table-wrap"><table><thead><tr><th>Proxy account ID</th><th>Input</th><th>Output</th><th>Cache write</th><th>Cache read</th><th>Requests</th><th>Failures</th></tr></thead><tbody>${(value.proxy.byAccount || []).map(row => `<tr><th scope="row">${txt(row.accountId)}</th>${['input','output','cacheWrite','cacheRead','requests','failures'].map(k => `<td>${txt(row[k])}</td>`).join('')}</tr>`).join('')}</tbody></table></div></section>` : '';
     return `<section class="panel" aria-labelledby="token-usage-title"><h2 id="token-usage-title">Token usage</h2><p>Best reading of session-editable transcripts, not proof. Unknown turns may have used more tokens. Account is unassigned until account tracking is configured.</p><div class="table-wrap"><table id="token-usage-table"><thead><tr><th scope="col">Subject</th><th scope="col">Input</th><th scope="col">Output</th><th scope="col">Cache write</th><th scope="col">Cache read</th><th scope="col">Unknown turns</th></tr></thead><tbody>${rows.join('')}</tbody></table></div>${orphans}${proxy}</section>`;
   }
   // Row 155 (#1649, Q169): each assistant instance's settings, its seat name and a stop. A save

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:['<img src=x>']});
  assert.doesNotMatch(hostile,/<img/);assert.match(hostile,/&lt;img src=x&gt;/);
 });
+// #778: batches the proxy gave up and the bus did not save. Shown in the proxy section only when above 0.
+test('#778 Console says how many proxy batches were dropped, only when there are any, and that their events are not in the totals',()=>{
+ const native={total:counts,byKey:[]},proxy=(over={})=>({source:'proxy',state:'available',byAccount:[],...over});
+ const one=render({...native,proxy:proxy({droppedBatches:1})});
+ assert.match(one,/<p id="token-usage-proxy-dropped" role="status">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/);
+});