Skip to content

Commit c68a33c

Browse files
committed
fix(webapp): stop get_queue claiming a complete slot-holder list
Rename slotHolderFacts.consistency to counterAgreement, add the always-true ckAdmittedMayBeUnlisted, and report env-scoped admitted concurrency, so an all-zero holder list no longer reads as an idle queue.
1 parent 9979fec commit c68a33c

9 files changed

Lines changed: 118 additions & 64 deletions

File tree

‎apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts‎

Lines changed: 23 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import { BasePresenter } from "./basePresenter.server";
1515

1616
type SlotHolderPhase = "admitted" | "dequeued";
1717
export type SlotHolderConsistency = "consistent" | "mismatch" | "unresolved";
18+
type SlotHolderCounterAgreement = "agree" | "disagree" | "unresolved";
1819
/** "not_found": a Redis slot holder with no matching TaskRun row. */
1920
type SlotHolderStatus = TaskRunStatus | "not_found";
2021

@@ -25,6 +26,11 @@ export type EnvConcurrency = {
2526
current: number;
2627
/** The dequeue gate is `current < limit * burstFactor`, not `current < limit`. */
2728
burstFactor: number;
29+
/**
30+
* Env-scoped admitted slots. When `admitted > current`, the environment holds admitted
31+
* slots the queue's holder list can't attribute.
32+
*/
33+
admitted: number;
2834
};
2935

3036
export type SlotHolder = {
@@ -47,7 +53,9 @@ export type SlotHolderFacts = {
4753
/** Dequeued holders that provably exist but aren't listed. */
4854
unlistedRunning: number;
4955
/** The counts mean nothing when this is "unresolved". */
50-
consistency: SlotHolderConsistency;
56+
counterAgreement: SlotHolderCounterAgreement;
57+
/** Always true: an admitted concurrency-key holder with no backlog is never listable. */
58+
ckAdmittedMayBeUnlisted: true;
5159
};
5260

5361
// A run can only hold a slot before its final status. PENDING counts because Redis
@@ -78,11 +86,11 @@ export function slotHolderConsistency(
7886
export async function envConcurrencyFromRead(
7987
limit: number,
8088
burstFactor: number,
81-
readCurrent: () => Promise<number>
89+
readCounts: () => Promise<{ current: number; admitted: number }>
8290
): Promise<EnvConcurrency | undefined> {
8391
try {
84-
const current = await readCurrent();
85-
return { limit, current, burstFactor };
92+
const { current, admitted } = await readCounts();
93+
return { limit, current, burstFactor, admitted };
8694
} catch {
8795
return undefined;
8896
}
@@ -215,9 +223,13 @@ export class QueueRetrievePresenter extends BasePresenter {
215223
typeof environment.concurrencyLimitBurstFactor === "number"
216224
? environment.concurrencyLimitBurstFactor
217225
: environment.concurrencyLimitBurstFactor.toNumber();
218-
return envConcurrencyFromRead(environment.maximumConcurrencyLimit, burstFactor, () =>
219-
engine.concurrencyOfEnvQueue(environment)
220-
);
226+
return envConcurrencyFromRead(environment.maximumConcurrencyLimit, burstFactor, async () => {
227+
const [current, admitted] = await Promise.all([
228+
engine.concurrencyOfEnvQueue(environment),
229+
engine.admittedConcurrencyOfEnvQueue(environment),
230+
]);
231+
return { current, admitted };
232+
});
221233
}
222234

223235
/**
@@ -236,7 +248,8 @@ export class QueueRetrievePresenter extends BasePresenter {
236248
runningReported: 0,
237249
truncated: false,
238250
unlistedRunning: 0,
239-
consistency: "unresolved" as const,
251+
counterAgreement: "unresolved" as const,
252+
ckAdmittedMayBeUnlisted: true as const,
240253
},
241254
};
242255

@@ -290,7 +303,8 @@ export class QueueRetrievePresenter extends BasePresenter {
290303
runningReported: snapshot.runningReported,
291304
truncated: snapshot.truncated,
292305
unlistedRunning: snapshot.unlistedRunning,
293-
consistency: snapshot.consistency,
306+
counterAgreement: snapshot.counterAgreement,
307+
ckAdmittedMayBeUnlisted: snapshot.ckAdmittedMayBeUnlisted,
294308
},
295309
};
296310
}

‎apps/webapp/app/presenters/v3/QueueRetrievePresenter.test.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,11 +46,14 @@ describe("slotHolderConsistency", () => {
4646
});
4747

4848
describe("envConcurrencyFromRead", () => {
49-
it("pairs the env limit and burst factor with the read current concurrency", async () => {
50-
await expect(envConcurrencyFromRead(10, 2, async () => 7)).resolves.toEqual({
49+
it("pairs the env limit and burst factor with the read concurrency counts", async () => {
50+
await expect(
51+
envConcurrencyFromRead(10, 2, async () => ({ current: 7, admitted: 9 }))
52+
).resolves.toEqual({
5153
limit: 10,
5254
current: 7,
5355
burstFactor: 2,
56+
admitted: 9,
5457
});
5558
});
5659

‎internal-packages/dashboard-agent/src/__snapshots__/prompt-prefix.test.ts.snap‎

Lines changed: 10 additions & 10 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎internal-packages/dashboard-agent/src/prompt-prefix.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -88,8 +88,8 @@ describe("the head-start and agent prefixes are the same prefix", () => {
8888
* drift. The snapshot below is the itemised diff a reviewer reads.
8989
*/
9090
const PREFIX_BUDGET = {
91-
assistant: { chars: 77_000, estimatedTokens: 19_500, tools: 24, promptChars: 27_000 },
92-
code: { chars: 83_000, estimatedTokens: 21_000, tools: 28, promptChars: 29_500 },
91+
assistant: { chars: 77_500, estimatedTokens: 19_500, tools: 24, promptChars: 27_000 },
92+
code: { chars: 83_500, estimatedTokens: 21_000, tools: 28, promptChars: 29_500 },
9393
} as const;
9494

9595
describe("the prefix stays inside its budget", () => {

‎internal-packages/dashboard-agent/src/tool-queue.test.ts‎

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -371,7 +371,8 @@ describe("get_queue carries slot-holder facts through, and omits them when absen
371371
runningReported: 4,
372372
truncated: true,
373373
unlistedRunning: 2,
374-
consistency: "mismatch",
374+
counterAgreement: "disagree",
375+
ckAdmittedMayBeUnlisted: true,
375376
};
376377
stubFetch({ slotHolders: [], slotHolderFacts });
377378
const answer = await getQueue()({ queue: "email-sends", type: "custom" });
@@ -392,16 +393,33 @@ describe("get_queue carries slot-holder facts through, and omits them when absen
392393
runningReported: 2,
393394
truncated: false,
394395
unlistedRunning: 0,
395-
consistency: "consistent",
396+
counterAgreement: "agree",
397+
ckAdmittedMayBeUnlisted: true,
396398
};
397399
stubFetch({ slotHolderFacts });
398400
const answer = await getQueue()({ queue: "email-sends", type: "custom" });
399401
expect(answer).toMatchObject({ slotHolderFacts });
400402
expect(answer).not.toHaveProperty("slotHolders");
401403
});
402404

403-
it("carries envConcurrency verbatim, including burstFactor", async () => {
404-
const envConcurrency = { limit: 10, current: 10, burstFactor: 2 };
405+
it("carries ckAdmittedMayBeUnlisted alongside an all-zero holder list", async () => {
406+
// The zeroes alone would read as "idle"; the flag is what stops that.
407+
const slotHolderFacts = {
408+
admittedCount: 0,
409+
dequeuedCount: 0,
410+
runningReported: 0,
411+
truncated: false,
412+
unlistedRunning: 0,
413+
counterAgreement: "agree",
414+
ckAdmittedMayBeUnlisted: true,
415+
};
416+
stubFetch({ slotHolders: [], slotHolderFacts });
417+
const answer = await getQueue()({ queue: "email-sends", type: "custom" });
418+
expect(answer).toMatchObject({ slotHolders: [], slotHolderFacts });
419+
});
420+
421+
it("carries envConcurrency verbatim, including burstFactor and admitted", async () => {
422+
const envConcurrency = { limit: 10, current: 10, burstFactor: 2, admitted: 14 };
405423
stubFetch({ envConcurrency });
406424
const answer = await getQueue()({ queue: "email-sends", type: "custom" });
407425
expect(answer).toMatchObject({ envConcurrency });

‎internal-packages/dashboard-agent/src/tool-schemas.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,7 @@ export const getQueueSchema = tool({
216216
description:
217217
"Get one queue's metrics over a window: wait latency (p50/p95), peak depth, how many runs started (throughput), and how often the queue was throttled by its concurrency limit. Use this for 'how deep is the X queue', 'is X backed up', or 'why are runs waiting'. The answer also carries the queue's live row: `paused`, `queuedNow`, `runningNow`, `concurrencyLimit`, and `exists: false` when no queue of that name is there at all. When `exists` is `false` in the current environment, " +
218218
MANDATORY_SWEEP +
219-
" When that read fails rather than answers, `exists` is `\"unknown\"` with a `liveStateError`: the queue's state is unknown, not missing. For a custom queue it also carries `consumerTasks`: the deployed tasks whose queue config names this queue. When present, `slotHolders` (each run's id, status, uri, consistency, phase (`admitted` | `dequeued`), and concurrencyKey) lists the runs holding the queue's concurrency slots, but the list is never guaranteed exhaustive; `slotHolderFacts` (admittedCount, dequeuedCount, runningReported, truncated, unlistedRunning, consistency) is the server-computed snapshot summary — `truncated` or `unlistedRunning > 0` mean there are holders `slotHolders` doesn't list. When present, `concurrency` (current, base, override, overriddenBy, overriddenAt) distinguishes a temporary override from configured `concurrencyLimit`. When present, `envConcurrency` (limit, current, burstFactor) is the environment-wide dequeue gate: the environment saturates at `current >= limit * burstFactor`, not at `current >= limit` (burstFactor defaults to 2, so headroom above the plain limit is often still open) — and `current` is the last-displayed dequeued count, which can lag the number actually gating dequeues. Use these three fields together before naming the environment as the bottleneck; never infer that from throttledCount alone. All are absent on an older API rather than empty. A holder's phase `admitted` (not yet `dequeued`) may legitimately be pending, not a mismatch. Consistency \"mismatch\" on a holder means the scheduler still counts it as a holder though its run state disagrees; on slotHolderFacts it means the scheduler's own counters disagree right now — prefer those facts to comparing runningNow yourself. Call a holder \"leaked\" or \"stale\" ONLY when both are observed this turn — its run state is terminal (or not found) AND the scheduler still holds the slot; from counters alone, never. Consistency `unresolved` means the run id is citable but its state, and slotHolderFacts' counts, are not — don't assert either. Never assert a run is currently executing from runningNow or concurrencyLimit alone, and never say holders are unaccounted for beyond what truncated/unlistedRunning/consistency actually state — 'nothing holds the slots' is never licensed by an incomplete list.",
219+
" When that read fails rather than answers, `exists` is `\"unknown\"` with a `liveStateError`: the queue's state is unknown, not missing. For a custom queue it also carries `consumerTasks`: the deployed tasks whose queue config names this queue. When present, `slotHolders` (each run's id, status, uri, consistency, phase (`admitted` | `dequeued`), and concurrencyKey) lists the runs holding the queue's concurrency slots, but the list is never guaranteed exhaustive; `slotHolderFacts` (admittedCount, dequeuedCount, runningReported, truncated, unlistedRunning, counterAgreement, ckAdmittedMayBeUnlisted) is the server-computed snapshot summary — `truncated` or `unlistedRunning > 0` mean there are holders `slotHolders` doesn't list. `ckAdmittedMayBeUnlisted` is always true: a holder admitted under a concurrency key with nothing backlogged is never listed or counted, so an all-zero holder list is not proof the queue is idle. When present, `concurrency` (current, base, override, overriddenBy, overriddenAt) distinguishes a temporary override from configured `concurrencyLimit`. When present, `envConcurrency` (limit, current, burstFactor, admitted) is the environment-wide dequeue gate: the environment saturates at `current >= limit * burstFactor`, not at `current >= limit` (burstFactor defaults to 2, so headroom above the plain limit is often still open) — and `current` is the last-displayed dequeued count, which can lag the number actually gating dequeues. `admitted > current` means the environment holds admitted slots this queue's holder list can't attribute. Use these three fields together before naming the environment as the bottleneck; never infer that from throttledCount alone. All are absent on an older API rather than empty. A holder's phase `admitted` (not yet `dequeued`) may legitimately be pending, not a mismatch. Consistency \"mismatch\" on a holder means the scheduler still counts it as a holder though its run state disagrees; `counterAgreement: \"disagree\"` on slotHolderFacts means the scheduler's own counters disagree right now — prefer those facts to comparing runningNow yourself. Call a holder \"leaked\" or \"stale\" ONLY when both are observed this turn — its run state is terminal (or not found) AND the scheduler still holds the slot; from counters alone, never. `unresolved` (holder consistency, or counterAgreement) means the run id is citable but its state, and slotHolderFacts' counts, are not — don't assert either. Never assert a run is currently executing from runningNow or concurrencyLimit alone, and never say holders are unaccounted for beyond what truncated/unlistedRunning/counterAgreement/ckAdmittedMayBeUnlisted actually state — 'nothing holds the slots' is never licensed by an incomplete list.",
220220
inputSchema: z.object({
221221
queue: z
222222
.string()

‎internal-packages/run-engine/src/engine/index.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1730,6 +1730,13 @@ export class RunEngine {
17301730
return this.runQueue.currentConcurrencyOfEnvironment(environment);
17311731
}
17321732

1733+
/** Admitted env-scope slots — the gating count, ahead of the displayed dequeued one. */
1734+
async admittedConcurrencyOfEnvQueue(
1735+
environment: MinimalAuthenticatedEnvironment
1736+
): Promise<number> {
1737+
return this.runQueue.operationalCurrentConcurrencyOfEnvironment(environment);
1738+
}
1739+
17331740
async lengthOfQueues(
17341741
environment: MinimalAuthenticatedEnvironment,
17351742
queues: string[]

‎internal-packages/run-engine/src/run-queue/index.ts‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,13 @@ export type QueueSlotHolders = {
165165
truncated: boolean;
166166
/** Dequeued holders that provably exist but aren't in the list. */
167167
unlistedRunning: number;
168-
consistency: "consistent" | "mismatch";
168+
/** Whether the queue's own counters agree with the enumerated members. */
169+
counterAgreement: "agree" | "disagree";
170+
/**
171+
* Structurally always true: enumeration covers the base queue and backlogged CK variants
172+
* only, and any run can carry a concurrency key, so an admitted CK holder can be unlistable.
173+
*/
174+
ckAdmittedMayBeUnlisted: true;
169175
};
170176

171177
/** Injected queue-metrics stream emitter; all calls are no-ops when metrics are disabled. */
@@ -687,7 +693,8 @@ export class RunQueue {
687693

688694
/**
689695
* Snapshot of who holds this queue's slots. Never complete: ckIndex only tracks queued
690-
* variants, so a drained CK variant's holders are invisible. `consistency` flags drift.
696+
* variants, so a drained CK variant's holders are invisible — `ckAdmittedMayBeUnlisted`
697+
* states that unconditionally. `counterAgreement` flags drift.
691698
*/
692699
public async slotHoldersOfQueue(
693700
env: MinimalAuthenticatedEnvironment,
@@ -729,8 +736,9 @@ export class RunQueue {
729736
// A CK-variant scan cap also makes the snapshot incomplete, same as a holder-list cap.
730737
truncated: truncated === 1 || skippedVariants > 0,
731738
unlistedRunning: Math.max(0, runningReported - dequeuedCount),
732-
consistency:
733-
dequeuedCount === runningReported && orphanCount === 0 ? "consistent" : "mismatch",
739+
counterAgreement:
740+
dequeuedCount === runningReported && orphanCount === 0 ? "agree" : "disagree",
741+
ckAdmittedMayBeUnlisted: true,
734742
};
735743
}
736744

‎internal-packages/run-engine/src/run-queue/tests/slotHolders.test.ts‎

Lines changed: 36 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,7 @@ describe("RunQueue.slotHoldersOfQueue", () => {
101101
expect(admitted.runningReported).toBe(0);
102102
expect(admitted.truncated).toBe(false);
103103
expect(admitted.unlistedRunning).toBe(0);
104-
expect(admitted.consistency).toBe("consistent");
104+
expect(admitted.counterAgreement).toBe("agree");
105105

106106
const dequeued = await queue.dequeueMessageFromWorkerQueue("consumer_1", WORKER_QUEUE);
107107
expect(dequeued?.messageId).toBe("r1");
@@ -111,7 +111,7 @@ describe("RunQueue.slotHoldersOfQueue", () => {
111111
expect(after.dequeuedCount).toBe(1);
112112
expect(after.runningReported).toBe(1);
113113
expect(after.unlistedRunning).toBe(0);
114-
expect(after.consistency).toBe("consistent");
114+
expect(after.counterAgreement).toBe("agree");
115115
} finally {
116116
await queue.quit();
117117
}
@@ -143,7 +143,7 @@ describe("RunQueue.slotHoldersOfQueue", () => {
143143
expect(result.runningReported).toBe(1);
144144
expect(result.truncated).toBe(false);
145145
expect(result.unlistedRunning).toBe(0);
146-
expect(result.consistency).toBe("consistent");
146+
expect(result.counterAgreement).toBe("agree");
147147
} finally {
148148
await queue.quit();
149149
}
@@ -167,7 +167,7 @@ describe("RunQueue.slotHoldersOfQueue", () => {
167167
});
168168

169169
const baseline = await queue.slotHoldersOfQueue(authenticatedEnvDev, QUEUE);
170-
expect(baseline.consistency).toBe("consistent");
170+
expect(baseline.counterAgreement).toBe("agree");
171171

172172
// Control break: the counter no longer matches the enumerated members.
173173
await queue.redis.set(
@@ -176,7 +176,7 @@ describe("RunQueue.slotHoldersOfQueue", () => {
176176
);
177177

178178
const broken = await queue.slotHoldersOfQueue(authenticatedEnvDev, QUEUE);
179-
expect(broken.consistency).toBe("mismatch");
179+
expect(broken.counterAgreement).toBe("disagree");
180180
expect(broken.holders).toEqual([{ runId: "r1", concurrencyKey: "ck-a", phase: "admitted" }]);
181181
expect(broken.runningReported).toBe(7);
182182
expect(broken.unlistedRunning).toBe(7);
@@ -204,41 +204,45 @@ describe("RunQueue.slotHoldersOfQueue", () => {
204204
expect(result.holders).toEqual([]);
205205
expect(result.runningReported).toBe(1);
206206
expect(result.unlistedRunning).toBe(1);
207-
expect(result.consistency).toBe("mismatch");
207+
expect(result.counterAgreement).toBe("disagree");
208208
} finally {
209209
await queue.quit();
210210
}
211211
}
212212
);
213213

214-
redisTest("a lone fast-path CK holder is simply not listed", async ({ redisContainer }) => {
215-
const queue = createQueue(redisContainer);
216-
try {
217-
// Nothing queued on the variant means no ckIndex entry, so this holder can't be
218-
// enumerated. No field claims otherwise — the payload just doesn't mention it.
219-
await queue.enqueueMessage({
220-
env: authenticatedEnvDev,
221-
message: makeMessage({ runId: "r1", concurrencyKey: "ck-a" }),
222-
workerQueue: WORKER_QUEUE,
223-
skipDequeueProcessing: true,
224-
enableFastPath: true,
225-
});
214+
redisTest(
215+
"a lone fast-path CK holder is not listed, and the payload says so",
216+
async ({ redisContainer }) => {
217+
const queue = createQueue(redisContainer);
218+
try {
219+
// Nothing queued on the variant means no ckIndex entry, so this holder can't be
220+
// enumerated. `ckAdmittedMayBeUnlisted` is what stops the zeroes reading as "idle".
221+
await queue.enqueueMessage({
222+
env: authenticatedEnvDev,
223+
message: makeMessage({ runId: "r1", concurrencyKey: "ck-a" }),
224+
workerQueue: WORKER_QUEUE,
225+
skipDequeueProcessing: true,
226+
enableFastPath: true,
227+
});
226228

227-
const result = await queue.slotHoldersOfQueue(authenticatedEnvDev, QUEUE);
228-
expect(result).toEqual({
229-
holders: [],
230-
admittedCount: 0,
231-
dequeuedCount: 0,
232-
runningReported: 0,
233-
truncated: false,
234-
unlistedRunning: 0,
235-
consistency: "consistent",
236-
});
237-
expect(result).not.toHaveProperty("holderResolution");
238-
} finally {
239-
await queue.quit();
229+
const result = await queue.slotHoldersOfQueue(authenticatedEnvDev, QUEUE);
230+
expect(result).toEqual({
231+
holders: [],
232+
admittedCount: 0,
233+
dequeuedCount: 0,
234+
runningReported: 0,
235+
truncated: false,
236+
unlistedRunning: 0,
237+
counterAgreement: "agree",
238+
ckAdmittedMayBeUnlisted: true,
239+
});
240+
expect(result).not.toHaveProperty("holderResolution");
241+
} finally {
242+
await queue.quit();
243+
}
240244
}
241-
});
245+
);
242246

243247
redisTest("caps the holder list and reports it as truncated", async ({ redisContainer }) => {
244248
const queue = createQueue(redisContainer);

0 commit comments

Comments
 (0)