From 0371d912cd1731736645ad2e00211b8f941586fd Mon Sep 17 00:00:00 2001 From: Dmitrii Vasilev Date: Sat, 5 Sep 2026 02:15:47 +0700 Subject: [PATCH] feat(queen): carry queen-1303 workers projection onto feat/queen-supervisor Carry-forward of origin/queen-1303 (#1303) onto the current base. The original branch is untouched; this is a fresh branch off origin/feat/queen-supervisor with only the substance the base lacks. Carried, verified absent from the base: - `workers` on GET /queen/status: capacity, active, idle, utilization. The base publishes a `workers` block on /queen/public-research only; /queen/status had no worker projection at all, so `running: 0` there still had no denominator. - `workerProjection()` in queen-public-status.ts, with the total clamp (0 <= active <= capacity, integer utilization 0..100). - `workerCapacity` dependency, defaulting to `configuredWorkerCapacity` from api/services/queen-dispatch - the same one authority /queen/public-research already reads, never a second parser of the provider environment. - `started_running` in the aggregate count query (`started = true AND finished_at IS NULL`), so a dispatch that never got a turn stays in `running` but spends no paid slot. This matches the reaper's existing predicate in queen-dispatch.ts. - The header paragraph explaining the block, and five tests: the 4-slot half-busy story, zero capacity with planted credential-shaped rows, the started=false regression, the malformed-count clamp, and the same-authority check against configuredWorkerCapacity(). Dropped as superseded by the base: - `swarmState: 'waiting_for_review'` in "returns only the public runtime summary". The base reclassified that fixture to 'healthy_idle' - a tick that refused with `nothing to choose` measured the backlog, so an owed review may not name the quiet. Re-applying queen-1303's expectation would have reverted that fix. - queen-1303's `lastTick` shape, which predates the base's `skipSummary` issue lists and `skipIssueListCap`. The base shape is kept as-is; only `workers` was added to the two exhaustive payload assertions. Verified: bun test apps/server/tests/api/queen-public-status.test.ts 21 pass, 0 fail, 116 expect() calls. --- .../src/api/routes/queen-public-status.ts | 72 ++++- .../tests/api/queen-public-status.test.ts | 298 +++++++++++++++++- 2 files changed, 365 insertions(+), 5 deletions(-) diff --git a/trios/agent-server/apps/server/src/api/routes/queen-public-status.ts b/trios/agent-server/apps/server/src/api/routes/queen-public-status.ts index 3735667abe..c3e529286f 100644 --- a/trios/agent-server/apps/server/src/api/routes/queen-public-status.ts +++ b/trios/agent-server/apps/server/src/api/routes/queen-public-status.ts @@ -20,11 +20,23 @@ * A count nobody can act on was the defect this projection shipped with: the * count said two issues lacked a boundary and could not say which two. The * number alone is inert - paths, holders, titles and prose never leave. + * + * The other half of `running: 0` is HOW BIG the swarm is at all, and `workers` + * answers it: no capacity configured, capacity all idle, or telemetry that + * says nothing. Its capacity comes from the same `configuredWorkerCapacity` + * authority `/queen/public-research` reads - one count of the provider + * environment, never a second parser of it - so the two public pages cannot + * tell two different stories about how many paid slots exist. Four numbers + * leave and nothing else: the count's source is the credential environment, + * and a provider's name, an environment-variable name or a key value has no + * business on a public page when the count alone says whether capacity is + * idle. */ import { Hono } from 'hono' import { Pool } from 'pg' import { logger } from '../../lib/logger' +import { configuredWorkerCapacity } from '../services/queen-dispatch' interface QueryResult { rowCount: number | null @@ -41,6 +53,7 @@ interface QueenPublicStatusDeps { createPool?: (url: string) => StatusPool tickIntervalSeconds?: () => number billingMode?: () => BillingMode + workerCapacity?: () => number } type BillingMode = 'api_metered' | 'coding_plan' @@ -289,6 +302,51 @@ function classifySwarmState(facts: { return 'healthy_idle' } +/** + * The paid worker-slot reading of `dispatches.running`. + * + * `running` counts every unfinished dispatch; `active` counts only the ones + * that actually started - `started = true` and `finished_at IS NULL` - because + * a dispatch that never got a turn holds no paid slot. Capacity is read + * through the same `configuredWorkerCapacity` authority `/queen/public-research` + * uses, never a second parser of the provider environment, so the two public + * pages can never disagree about how many slots exist. + * + * The clamp is total because a count can only arrive wrong: a non-numeric + * string, a negative, a float, the same row counted twice. Active above + * capacity would promise slots the swarm does not have and a negative idle + * would read as over-subscription, so every malformed value folds into the + * closed range `0 <= active <= capacity` with `idle = capacity - active` and + * an integer percentage from 0 through 100. + */ +function workerProjection( + capacity: number, + startedUnfinished: number, +): { + capacity: number + active: number + idle: number + utilization: number +} { + const safeCapacity = + Number.isFinite(capacity) && capacity > 0 ? Math.floor(capacity) : 0 + const active = Math.min( + safeCapacity, + Math.floor( + Number.isFinite(startedUnfinished) && startedUnfinished > 0 + ? startedUnfinished + : 0, + ), + ) + return { + capacity: safeCapacity, + active, + idle: safeCapacity - active, + utilization: + safeCapacity > 0 ? Math.round((active / safeCapacity) * 100) : 0, + } +} + export function createQueenPublicStatusRoute(deps: QueenPublicStatusDeps = {}) { const databaseUrl = deps.databaseUrl ?? configuredDatabaseUrl const createPool = @@ -297,6 +355,7 @@ export function createQueenPublicStatusRoute(deps: QueenPublicStatusDeps = {}) { const tickIntervalSeconds = deps.tickIntervalSeconds ?? configuredTickIntervalSeconds const billingMode = deps.billingMode ?? configuredBillingMode + const workerCapacity = deps.workerCapacity ?? configuredWorkerCapacity return new Hono().get('/', async (c) => { c.header('Cache-Control', 'no-store') @@ -322,7 +381,14 @@ export function createQueenPublicStatusRoute(deps: QueenPublicStatusDeps = {}) { -- though both read identically without this column. count(*) FILTER ( WHERE finished_at IS NOT NULL AND review_state IS NULL - ) AS unreviewed + ) AS unreviewed, + -- A paid slot is spent only by a dispatch that actually started + -- and has not finished. running counts dispatches that never + -- got a turn too, and those spend nothing, so workers.active + -- is counted here rather than derived from running. + count(*) FILTER ( + WHERE started = true AND finished_at IS NULL + ) AS started_running FROM queen_dispatch`, ) const latest = await pool.query( @@ -368,6 +434,10 @@ export function createQueenPublicStatusRoute(deps: QueenPublicStatusDeps = {}) { // after accepted work finishes. decisionFoundNoEligibleCandidate: decision?.allowed === false, }), + workers: workerProjection( + workerCapacity(), + asCount(countRow.started_running), + ), scheduler: { enabled: schedulerEnabled, intervalSeconds, diff --git a/trios/agent-server/apps/server/tests/api/queen-public-status.test.ts b/trios/agent-server/apps/server/tests/api/queen-public-status.test.ts index 8d777ce7fc..8eb72e8208 100644 --- a/trios/agent-server/apps/server/tests/api/queen-public-status.test.ts +++ b/trios/agent-server/apps/server/tests/api/queen-public-status.test.ts @@ -4,6 +4,7 @@ import { createQueenPublicStatusRoute, SKIP_ISSUE_LIST_CAP, } from '../../src/api/routes/queen-public-status' +import { configuredWorkerCapacity } from '../../src/api/services/queen-dispatch' type QueryResult = { rowCount: number; rows: Array> } @@ -21,12 +22,21 @@ function fakePool(results: QueryResult[]) { /** * The counts row and the latest-dispatch row every fixture below pairs with - * its tick row, so each test states only the part it varies. `unreviewed` - * rides along with the other counts the aggregate query returns. + * its tick row, so each test states only the part it varies. `unreviewed` and + * `started_running` ride along with the other counts the aggregate query + * returns. */ const emptyDispatchCounts: QueryResult = { rowCount: 1, - rows: [{ total: '0', finished: '0', running: '0', unreviewed: '0' }], + rows: [ + { + total: '0', + finished: '0', + running: '0', + unreviewed: '0', + started_running: '0', + }, + ], } const noLatestDispatch: QueryResult = { rowCount: 0, rows: [] } @@ -165,7 +175,15 @@ describe('GET /queen/status', () => { }, { rowCount: 1, - rows: [{ total: '8', finished: '8', running: '0', unreviewed: '2' }], + rows: [ + { + total: '8', + finished: '8', + running: '0', + unreviewed: '2', + started_running: '0', + }, + ], }, { rowCount: 1, @@ -188,6 +206,7 @@ describe('GET /queen/status', () => { createPool: () => pool, tickIntervalSeconds: () => 1800, billingMode: () => 'coding_plan', + workerCapacity: () => 4, }).request('/') expect(response.status).toBe(200) @@ -200,6 +219,14 @@ describe('GET /queen/status', () => { // stay counted under dispatches.unreviewed; they just cannot name the // quiet while the tick that measured it says otherwise. swarmState: 'healthy_idle', + // Paid slots exist and are all idle: running 0 here is an empty queue, + // not missing capacity. + workers: { + capacity: 4, + active: 0, + idle: 4, + utilization: 0, + }, scheduler: { enabled: true, intervalSeconds: 1800, @@ -410,6 +437,7 @@ describe('GET /queen/status', () => { databaseUrl: () => 'postgres://configured', createPool: () => pool, tickIntervalSeconds: () => 1800, + workerCapacity: () => 0, }).request('/') expect(response.status).toBe(200) @@ -419,6 +447,14 @@ describe('GET /queen/status', () => { // be the real recordTick -> recordDispatch window, so the snapshot is // unavailable until the row appears or a no-choice tick supersedes it. swarmState: 'unavailable', + // No paid slot is configured, so every worker field is 0 - capacity 0 + // with nothing running and nothing to be idle. + workers: { + capacity: 0, + active: 0, + idle: 0, + utilization: 0, + }, scheduler: { enabled: true, intervalSeconds: 1800, @@ -740,6 +776,260 @@ describe('GET /queen/status', () => { skipIssueListCap: SKIP_ISSUE_LIST_CAP, }) }) + + it('gives running 0 a denominator: four slots with two started dispatches read half busy', async () => { + // Acceptance scenario 1, and the issue's own story: four configured + // worker slots, two unfinished started dispatches, no provider key set + // anywhere - the capacity is injected, because the point here is the + // projection, not the environment. `running: 2` next to + // capacity 4 is an operator's whole answer: 2 active, 2 idle, 50%. + const pool = fakePool([ + { rowCount: 0, rows: [] }, + { + rowCount: 1, + rows: [ + { + total: '3', + finished: '1', + running: '2', + unreviewed: '1', + started_running: '2', + }, + ], + }, + noLatestDispatch, + ]) + + const response = await createQueenPublicStatusRoute({ + databaseUrl: () => 'postgres://configured', + createPool: () => pool, + tickIntervalSeconds: () => 1800, + workerCapacity: () => 4, + }).request('/') + + expect(response.status).toBe(200) + const body = (await response.json()) as Record + expect(body.workers).toEqual({ + capacity: 4, + active: 2, + idle: 2, + utilization: 50, + }) + // The denominator explains the counts; it must not reshape them. + expect((body.dispatches as Record).running).toBe(2) + }) + + it('reports all zeros for zero configured capacity and exposes no credential name or value', async () => { + // Acceptance scenario 2: zero configured worker slots. Every field is 0, + // and the credential-shaped values planted in the rows the projection + // reads past - the counts row and the latest-dispatch row - must not ride + // along: this page's capacity source is the provider environment, so its + // names and values are the one leak that would matter most here. + const plantedSecret = 'sk-trios-0123456789abcdef-fedcba' + const pool = fakePool([ + { rowCount: 0, rows: [] }, + { + rowCount: 1, + rows: [ + { + total: '1', + finished: '0', + running: '1', + unreviewed: '0', + started_running: '1', + provider: 'ZAI_API_KEY', + credential: plantedSecret, + authorization: `Bearer ${plantedSecret}`, + }, + ], + }, + { + rowCount: 1, + rows: [ + { + issue: 1303, + dispatched_at: '2026-09-02T10:00:00.000Z', + finished_at: null, + outcome: null, + branch: 'queen-1303', + detail: `Bearer ${plantedSecret}`, + conversation_id: plantedSecret, + provider: 'ANTHROPIC_API_KEY', + model: plantedSecret, + }, + ], + }, + ]) + + const response = await createQueenPublicStatusRoute({ + databaseUrl: () => 'postgres://configured', + createPool: () => pool, + tickIntervalSeconds: () => 1800, + workerCapacity: () => 0, + }).request('/') + + expect(response.status).toBe(200) + const body = await response.json() + expect(body.workers).toEqual({ + capacity: 0, + active: 0, + idle: 0, + utilization: 0, + }) + const serialized = JSON.stringify(body) + expect(serialized).not.toContain('ZAI_API_KEY') + expect(serialized).not.toContain('ANTHROPIC_API_KEY') + expect(serialized).not.toContain('Bearer') + expect(serialized).not.toContain(plantedSecret) + }) + + it('does not count a started = false unfinished dispatch as active', async () => { + // The regression the success criteria name: two unfinished dispatches, + // neither of which started. They stay `running` - the table owes them an + // ending - but they spend no paid slot, so active is 0 against capacity 4 + // and all four slots read idle. + const pool = fakePool([ + { rowCount: 0, rows: [] }, + { + rowCount: 1, + rows: [ + { + total: '2', + finished: '0', + running: '2', + unreviewed: '0', + started_running: '0', + }, + ], + }, + noLatestDispatch, + ]) + + const response = await createQueenPublicStatusRoute({ + databaseUrl: () => 'postgres://configured', + createPool: () => pool, + tickIntervalSeconds: () => 0, + workerCapacity: () => 4, + }).request('/') + + expect(response.status).toBe(200) + const body = (await response.json()) as { + workers: Record + dispatches: { running: number } + } + expect(body.workers).toEqual({ + capacity: 4, + active: 0, + idle: 4, + utilization: 0, + }) + expect(body.dispatches.running).toBe(2) + }) + + it('clamps malformed counts so active never exceeds capacity and idle never goes negative', async () => { + // Acceptance scenario 3: a count can only arrive wrong - the same row + // counted twice, a float, a negative, a non-number - and every wrong + // shape must fold into the closed range rather than promise slots the + // swarm does not have or report over-subscription as a negative idle. + const read = async (startedRunning: unknown, capacity: number) => + ( + (await ( + await createQueenPublicStatusRoute({ + databaseUrl: () => 'postgres://configured', + createPool: () => + fakePool([ + { rowCount: 0, rows: [] }, + { + rowCount: 1, + rows: [ + { + total: '9', + finished: '0', + running: '9', + unreviewed: '0', + started_running: startedRunning, + }, + ], + }, + noLatestDispatch, + ]), + tickIntervalSeconds: () => 1800, + workerCapacity: () => capacity, + }).request('/') + ).json()) as Record + ).workers + + // Seven "active" dispatches against four slots: the swarm cannot spend + // more slots than it has, so the projection says full, never over. + expect(await read('7', 4)).toEqual({ + capacity: 4, + active: 4, + idle: 0, + utilization: 100, + }) + // A negative count is no count at all. + expect(await read(-2, 4)).toEqual({ + capacity: 4, + active: 0, + idle: 4, + utilization: 0, + }) + // A fractional dispatch is not a dispatch. + expect(await read('2.9', 4)).toEqual({ + capacity: 4, + active: 2, + idle: 2, + utilization: 50, + }) + // Malformed capacity admits no slots at all, and with no denominator + // there is nothing to be active or idle. + expect(await read('3', Number.NaN)).toEqual({ + capacity: 0, + active: 0, + idle: 0, + utilization: 0, + }) + expect(await read('3', 0)).toEqual({ + capacity: 0, + active: 0, + idle: 0, + utilization: 0, + }) + }) + + it('reads capacity from the same authority as public research, not a second parser', async () => { + // FR-002: the denominator must come from `configuredWorkerCapacity` - the + // one function that counts provider keys for dispatch and for + // /queen/public-research. With a key planted in the environment and NO + // workerCapacity injected, this route must report exactly what that one + // authority reports, and must not echo the key or its variable name. + const planted = 'sk-trios-env-authority-probe-fedcba9876' + const saved = process.env.ZAI_API_KEY + process.env.ZAI_API_KEY = planted + try { + const response = await createQueenPublicStatusRoute({ + databaseUrl: () => 'postgres://configured', + createPool: () => + fakePool([ + { rowCount: 0, rows: [] }, + emptyDispatchCounts, + noLatestDispatch, + ]), + tickIntervalSeconds: () => 1800, + }).request('/') + + const body = (await response.json()) as { + workers: { capacity: number } + } + expect(body.workers.capacity).toBe(configuredWorkerCapacity()) + const serialized = JSON.stringify(body) + expect(serialized).not.toContain('ZAI_API_KEY') + expect(serialized).not.toContain(planted) + } finally { + if (saved === undefined) delete process.env.ZAI_API_KEY + else process.env.ZAI_API_KEY = saved + } + }) }) describe('skipSummary issue numbers', () => {