From 531080ef87396f9248ea8072fd7468a2499bf7ed Mon Sep 17 00:00:00 2001 From: Dmitrii Vasilev Date: Wed, 2 Sep 2026 23:35:01 +0700 Subject: [PATCH] feat(queen): enforce measured runtime worker limits (#1311) --- .../server/src/api/routes/queen-kanban.ts | 9 ++- .../server/src/api/services/queen-tick.ts | 23 ++++++++ .../apps/server/tests/api/queen-round.test.ts | 20 +++++++ .../server/tests/api/queend-choose.test.ts | 59 ++++++++++++++++++- .../Sources/QueenPolicy/QueenDelegation.swift | 14 ++++- .../queen-core/Sources/queend/main.swift | 12 +++- trios/rings/SR-00/QueenDelegation.swift | 14 ++++- trios/tests/swift/queen_delegation_test.swift | 29 +++++++++ 8 files changed, 171 insertions(+), 9 deletions(-) diff --git a/trios/agent-server/apps/server/src/api/routes/queen-kanban.ts b/trios/agent-server/apps/server/src/api/routes/queen-kanban.ts index 5712e335a6..383673706d 100644 --- a/trios/agent-server/apps/server/src/api/routes/queen-kanban.ts +++ b/trios/agent-server/apps/server/src/api/routes/queen-kanban.ts @@ -47,7 +47,9 @@ import { Hono } from 'hono' import { Pool } from 'pg' +import { configuredWorkerCapacity } from '../services/queen-dispatch' import { queenLeaseDatabaseUrl } from '../services/queen-lease' +import { effectiveRuntimeWorkerLimit } from '../services/queen-tick' export type ReviewLifecycleState = | 'queenReviewPending' @@ -505,8 +507,11 @@ async function build(pool: Pool): Promise<{ cards: Card[]; pulse: Pulse }> { (lastTick.rows[0]?.decision as { refusal?: string } | undefined) ?.refusal ?? null, roundSeconds: Number(process.env.TRIOS_QUEEN_TICK_SECONDS ?? '0') || null, - workerKeys: providerKeyCount(), - workerLimit: 4, + // The page and the chooser must describe the same swarm. Count distinct + // credentials through the allocator's authority, then expose the exact + // bounded number queend receives rather than the old literal four. + workerKeys: configuredWorkerCapacity(), + workerLimit: effectiveRuntimeWorkerLimit(), }, } } diff --git a/trios/agent-server/apps/server/src/api/services/queen-tick.ts b/trios/agent-server/apps/server/src/api/services/queen-tick.ts index 1951072786..07af9c037e 100644 --- a/trios/agent-server/apps/server/src/api/services/queen-tick.ts +++ b/trios/agent-server/apps/server/src/api/services/queen-tick.ts @@ -41,6 +41,7 @@ import { Pool } from 'pg' import { logger } from '../../lib/logger' import { committedFiles, + configuredWorkerCapacity, dispatchBee, reapDispatchesFromPreviousBoot, reapStalledDispatches, @@ -131,6 +132,20 @@ function tickIntervalSeconds(): number { return Number.isFinite(raw) && raw > 0 ? raw : 0 } +/** + * The policy accepts one measured number, not credentials or provider names. + * One is the fail-closed floor: a deployment with no key still records one + * precise dispatch refusal instead of claiming the backlog was ineligible. + * Eight is the reviewed upper bound in #1311; any larger installation needs a + * new policy decision before it can add merge and review pressure. + */ +export function effectiveRuntimeWorkerLimit( + capacity = configuredWorkerCapacity(), +): number { + const whole = Number.isFinite(capacity) ? Math.floor(capacity) : 1 + return Math.min(8, Math.max(1, whole)) +} + /** GitHub's maximum, so the fewest requests per round. */ const ISSUE_PAGE_SIZE = 100 /** @@ -682,6 +697,7 @@ export async function runRound( fence: number, watch: LeaseWatch, candidateOverride?: number[], + workerLimitOverride?: number, ): Promise<{ ran: boolean reason?: string @@ -689,6 +705,11 @@ export async function runRound( dispatch?: unknown }> { const grant = { fence } + // Freeze it once for the entire round. Credentials rotating halfway through + // a refill must affect the next round, not make two choose calls in this one + // disagree about how many Bees may exist. + const maximumConcurrentWorkers = + effectiveRuntimeWorkerLimit(workerLimitOverride) const registry = await pool.query( 'SELECT tasks FROM queen_registry WHERE variant = $1', @@ -896,6 +917,7 @@ export async function runRound( candidates: choiceCandidates, candidateBodies, tasks: [...registryTasks, ...containerTasks], + maximumConcurrentWorkers, }) await recordTick(pool, holder, grant.fence, choice) @@ -1011,6 +1033,7 @@ export async function runRound( candidates, candidateBodies, tasks: board, + maximumConcurrentWorkers, }) if (!current?.allowed) { logger.info('Queen tick stopped dispatching', { diff --git a/trios/agent-server/apps/server/tests/api/queen-round.test.ts b/trios/agent-server/apps/server/tests/api/queen-round.test.ts index eaa14d8bd8..597206e530 100644 --- a/trios/agent-server/apps/server/tests/api/queen-round.test.ts +++ b/trios/agent-server/apps/server/tests/api/queen-round.test.ts @@ -16,6 +16,7 @@ import { } from '../../src/api/services/queen-dispatch' import { createRoundGate, + effectiveRuntimeWorkerLimit, refillOnBeeCompletion, runRound, } from '../../src/api/services/queen-tick' @@ -54,6 +55,25 @@ const BIN = join( ) const present = existsSync(BIN) +/** + * #1311. This is the server-side authority handed to every choose call in one + * round. It is numeric only, closed at eight, and keeps a fail-closed one-slot + * diagnostic path when the deployment has no usable provider credential. + */ +describe('queen round effective worker limit', () => { + it('admits measured capacities 2, 4 and 8 without inventing extra slots', () => { + expect(effectiveRuntimeWorkerLimit(2)).toBe(2) + expect(effectiveRuntimeWorkerLimit(4)).toBe(4) + expect(effectiveRuntimeWorkerLimit(8)).toBe(8) + }) + + it('clamps invalid and above-policy capacities to the closed range 1 through 8', () => { + expect(effectiveRuntimeWorkerLimit(0)).toBe(1) + expect(effectiveRuntimeWorkerLimit(Number.NaN)).toBe(1) + expect(effectiveRuntimeWorkerLimit(99)).toBe(8) + }) +}) + /** Every provider credential dispatch consults, so no bee is ever really run. */ const PROVIDER_KEYS = [ 'ZAI_API_KEY', diff --git a/trios/agent-server/apps/server/tests/api/queend-choose.test.ts b/trios/agent-server/apps/server/tests/api/queend-choose.test.ts index 62d38bef6d..b1ddc40831 100644 --- a/trios/agent-server/apps/server/tests/api/queend-choose.test.ts +++ b/trios/agent-server/apps/server/tests/api/queend-choose.test.ts @@ -77,11 +77,16 @@ const body = (n: number) => `\`docs/only-${n}.md\``, ].join('\n') -const board = (numbers: number[], tasks: Array>) => ({ +const board = ( + numbers: number[], + tasks: Array>, + maximumConcurrentWorkers?: number, +) => ({ kind: 'choose', candidates: numbers, candidateBodies: Object.fromEntries(numbers.map((n) => [String(n), body(n)])), tasks, + maximumConcurrentWorkers, }) describe('queend chooses the next bee', () => { @@ -126,6 +131,58 @@ describe('queend chooses the next bee', () => { expect(String(answer.skipped)).toContain('expected back') }) + /** + * #1311. The server knows the effective capacity for this deployment; the + * compiled policy used to know only its static four-worker default. Raising + * that constant would over-dispatch smaller installations, while leaving it + * alone stranded verified capacity above four. The number therefore travels + * with each choose question and is bounded by the policy at 1...8. + */ + it.skipIf(!present)( + 'admits the eighth worker at an effective limit of eight', + () => { + const running = Array.from({ length: 7 }, (_, index) => ({ + ...task(2000 + index, 'running'), + ownedPaths: [`docs/running-${index}.md`], + })) + const answer = ask(board([2100], running, 8)) + expect(answer.chosen).toBe(2100) + }, + ) + + it.skipIf(!present)( + 'refuses a fifth worker at an effective limit of four', + () => { + const running = Array.from({ length: 4 }, (_, index) => + task(2200 + index, 'running'), + ) + const answer = ask(board([2300], running, 4)) + expect(answer.allowed).toBe(false) + expect(String(answer.refusal)).toContain('limit 4') + }, + ) + + it.skipIf(!present)( + 'keeps the legacy four-worker default when the field is absent', + () => { + const running = Array.from({ length: 4 }, (_, index) => + task(2400 + index, 'running'), + ) + const answer = ask(board([2500], running)) + expect(answer.allowed).toBe(false) + expect(String(answer.refusal)).toContain('limit 4') + }, + ) + + it.skipIf(!present)('clamps an above-range runtime limit to eight', () => { + const running = Array.from({ length: 8 }, (_, index) => + task(2600 + index, 'running'), + ) + const answer = ask(board([2700], running, 99)) + expect(answer.allowed).toBe(false) + expect(String(answer.refusal)).toContain('limit 8') + }) + // A retry running over a past failure is claimed by the retry, whichever // order the registry lists them in. it.skipIf(!present)('lets a live retry outrank an old failure', () => { diff --git a/trios/agent-server/queen-core/Sources/QueenPolicy/QueenDelegation.swift b/trios/agent-server/queen-core/Sources/QueenPolicy/QueenDelegation.swift index 20b61f8a7b..330d8615ec 100644 --- a/trios/agent-server/queen-core/Sources/QueenPolicy/QueenDelegation.swift +++ b/trios/agent-server/queen-core/Sources/QueenPolicy/QueenDelegation.swift @@ -469,8 +469,18 @@ public enum QueenDelegationPolicy { /// review, and because merge conflicts scale with concurrency. public static let maximumConcurrentWorkers = 4 - public static func canStartAnother(running: Int) -> Bool { - running < maximumConcurrentWorkers + /// One bounded runtime can expose more verified slots than the legacy app + /// default, but never an unbounded swarm. Nil preserves every older caller. + public static func effectiveMaximumConcurrentWorkers(_ requested: Int?) -> Int { + guard let requested else { return maximumConcurrentWorkers } + return min(max(requested, 1), 8) + } + + public static func canStartAnother( + running: Int, + maximumConcurrentWorkers requested: Int? = nil + ) -> Bool { + running < effectiveMaximumConcurrentWorkers(requested) } /// What the tasks already recorded against an issue mean for choosing it diff --git a/trios/agent-server/queen-core/Sources/queend/main.swift b/trios/agent-server/queen-core/Sources/queend/main.swift index 3e4bb375fc..2c17db90c7 100644 --- a/trios/agent-server/queen-core/Sources/queend/main.swift +++ b/trios/agent-server/queen-core/Sources/queend/main.swift @@ -33,6 +33,9 @@ struct Question: Decodable { /// judges the same record rather than a summary of it. let candidates: [Int]? let tasks: [DelegatedTask]? + /// For `choose`: the server's measured effective worker capacity. Older + /// callers omit it and retain the compiled four-worker default. + let maximumConcurrentWorkers: Int? /// For `choose`: each candidate's issue BODY, keyed by issue number as a /// string. The boundary is parsed here rather than by the caller, so the /// container and the app read a boundary section by the same rule instead @@ -189,10 +192,15 @@ case "choose": } let now = Date() let running = tasks.filter { $0.state == .running }.count - guard QueenDelegationPolicy.canStartAnother(running: running) else { + let maximumConcurrentWorkers = QueenDelegationPolicy + .effectiveMaximumConcurrentWorkers(question.maximumConcurrentWorkers) + guard QueenDelegationPolicy.canStartAnother( + running: running, + maximumConcurrentWorkers: maximumConcurrentWorkers + ) else { emit(Answer(kind: "choose", strays: nil, refusal: "\(running) workers already running " - + "(limit \(QueenDelegationPolicy.maximumConcurrentWorkers))", + + "(limit \(maximumConcurrentWorkers))", allowed: false, chosen: nil, chosenPaths: nil, verdict: nil, unmet: nil, note: nil, skipped: nil, error: nil)) exit(0) } diff --git a/trios/rings/SR-00/QueenDelegation.swift b/trios/rings/SR-00/QueenDelegation.swift index 20b61f8a7b..330d8615ec 100644 --- a/trios/rings/SR-00/QueenDelegation.swift +++ b/trios/rings/SR-00/QueenDelegation.swift @@ -469,8 +469,18 @@ public enum QueenDelegationPolicy { /// review, and because merge conflicts scale with concurrency. public static let maximumConcurrentWorkers = 4 - public static func canStartAnother(running: Int) -> Bool { - running < maximumConcurrentWorkers + /// One bounded runtime can expose more verified slots than the legacy app + /// default, but never an unbounded swarm. Nil preserves every older caller. + public static func effectiveMaximumConcurrentWorkers(_ requested: Int?) -> Int { + guard let requested else { return maximumConcurrentWorkers } + return min(max(requested, 1), 8) + } + + public static func canStartAnother( + running: Int, + maximumConcurrentWorkers requested: Int? = nil + ) -> Bool { + running < effectiveMaximumConcurrentWorkers(requested) } /// What the tasks already recorded against an issue mean for choosing it diff --git a/trios/tests/swift/queen_delegation_test.swift b/trios/tests/swift/queen_delegation_test.swift index 3d68409141..508ffddca4 100644 --- a/trios/tests/swift/queen_delegation_test.swift +++ b/trios/tests/swift/queen_delegation_test.swift @@ -482,6 +482,35 @@ enum QueenDelegationTests { !QueenDelegationPolicy.canStartAnother(running: 99), "over the cap no new worker starts" ) + + check( + QueenDelegationPolicy.effectiveMaximumConcurrentWorkers(nil) == 4, + "an older caller keeps the four-worker default" + ) + check( + QueenDelegationPolicy.effectiveMaximumConcurrentWorkers(2) == 2, + "a two-slot runtime stays at two" + ) + check( + QueenDelegationPolicy.effectiveMaximumConcurrentWorkers(8) == 8, + "eight verified runtime slots are admitted" + ) + check( + QueenDelegationPolicy.effectiveMaximumConcurrentWorkers(0) == 1, + "a below-range runtime limit is clamped to one" + ) + check( + QueenDelegationPolicy.effectiveMaximumConcurrentWorkers(99) == 8, + "an above-range runtime limit is clamped to eight" + ) + check( + QueenDelegationPolicy.canStartAnother(running: 7, maximumConcurrentWorkers: 8), + "the eighth worker may start when the effective limit is eight" + ) + check( + !QueenDelegationPolicy.canStartAnother(running: 4, maximumConcurrentWorkers: 4), + "a four-slot runtime refuses a fifth worker" + ) } /// Structural conflict prevention: catch the clash at delegation time, not