Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions trios/agent-server/apps/server/src/api/routes/queen-kanban.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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(),
},
}
}
Expand Down
23 changes: 23 additions & 0 deletions trios/agent-server/apps/server/src/api/services/queen-tick.ts
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import { Pool } from 'pg'
import { logger } from '../../lib/logger'
import {
committedFiles,
configuredWorkerCapacity,
dispatchBee,
reapDispatchesFromPreviousBoot,
reapStalledDispatches,
Expand Down Expand Up @@ -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
/**
Expand Down Expand Up @@ -682,13 +697,19 @@ export async function runRound(
fence: number,
watch: LeaseWatch,
candidateOverride?: number[],
workerLimitOverride?: number,
): Promise<{
ran: boolean
reason?: string
choice?: QueendChoice
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',
Expand Down Expand Up @@ -896,6 +917,7 @@ export async function runRound(
candidates: choiceCandidates,
candidateBodies,
tasks: [...registryTasks, ...containerTasks],
maximumConcurrentWorkers,
})

await recordTick(pool, holder, grant.fence, choice)
Expand Down Expand Up @@ -1011,6 +1033,7 @@ export async function runRound(
candidates,
candidateBodies,
tasks: board,
maximumConcurrentWorkers,
})
if (!current?.allowed) {
logger.info('Queen tick stopped dispatching', {
Expand Down
20 changes: 20 additions & 0 deletions trios/agent-server/apps/server/tests/api/queen-round.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import {
} from '../../src/api/services/queen-dispatch'
import {
createRoundGate,
effectiveRuntimeWorkerLimit,
refillOnBeeCompletion,
runRound,
} from '../../src/api/services/queen-tick'
Expand Down Expand Up @@ -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',
Expand Down
59 changes: 58 additions & 1 deletion trios/agent-server/apps/server/tests/api/queend-choose.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -77,11 +77,16 @@ const body = (n: number) =>
`\`docs/only-${n}.md\``,
].join('\n')

const board = (numbers: number[], tasks: Array<ReturnType<typeof task>>) => ({
const board = (
numbers: number[],
tasks: Array<ReturnType<typeof task>>,
maximumConcurrentWorkers?: number,
) => ({
kind: 'choose',
candidates: numbers,
candidateBodies: Object.fromEntries(numbers.map((n) => [String(n), body(n)])),
tasks,
maximumConcurrentWorkers,
})

describe('queend chooses the next bee', () => {
Expand Down Expand Up @@ -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', () => {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
12 changes: 10 additions & 2 deletions trios/agent-server/queen-core/Sources/queend/main.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
}
Expand Down
14 changes: 12 additions & 2 deletions trios/rings/SR-00/QueenDelegation.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
29 changes: 29 additions & 0 deletions trios/tests/swift/queen_delegation_test.swift
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading