fix: foreground commander queue concurrency
Foreground codex queues commander concurrency summary and fixtures.
This commit is contained in:
Lyon
+251
-6
@@ -37,6 +37,7 @@ const commanderRecentCompletedLimit = 3;
|
||||
const commanderPromptPreviewChars = 96;
|
||||
const commanderBodyPreviewChars = 120;
|
||||
const commanderIssueTaskPreviewLimit = 4;
|
||||
const commanderConcurrencyTarget = 15;
|
||||
const unreadTriageCountLimit = 12;
|
||||
const diagnosticsIdPreviewLimit = 3;
|
||||
const diagnosticsReasonPreviewLimit = 2;
|
||||
@@ -492,6 +493,7 @@ interface SupervisorStatusCounts {
|
||||
|
||||
interface CodexQueuesOptions {
|
||||
full: boolean;
|
||||
commander: boolean;
|
||||
limit: number;
|
||||
offset: number;
|
||||
page: number;
|
||||
@@ -2663,7 +2665,7 @@ function parseUnreadOptions(args: string[]): CodexUnreadOptions {
|
||||
|
||||
function parseQueuesOptions(args: string[]): CodexQueuesOptions {
|
||||
assertKnownOptions(args, {
|
||||
flags: ["--full", "--all"],
|
||||
flags: ["--full", "--all", "--commander"],
|
||||
valueOptions: ["--limit", "--offset", "--page"],
|
||||
}, "codex queues");
|
||||
const limit = positiveIntegerOption(args, ["--limit"], defaultQueuesLimit, maxTasksLimit);
|
||||
@@ -2672,6 +2674,7 @@ function parseQueuesOptions(args: string[]): CodexQueuesOptions {
|
||||
const offset = offsetExplicit ? nonNegativeIntegerOption(args, ["--offset"], 0) : (page - 1) * limit;
|
||||
return {
|
||||
full: hasFlag(args, "--full") || hasFlag(args, "--all"),
|
||||
commander: hasFlag(args, "--commander") || !hasFlag(args, "--full") && !hasFlag(args, "--all"),
|
||||
limit,
|
||||
offset,
|
||||
page: Math.floor(offset / limit) + 1,
|
||||
@@ -4667,6 +4670,233 @@ function compactQueueRow(value: unknown): Record<string, unknown> {
|
||||
};
|
||||
}
|
||||
|
||||
function taskNameFromRecord(task: Record<string, unknown> | null): string | null {
|
||||
if (task === null) return null;
|
||||
for (const key of ["name", "title", "summary", "displayPrompt", "basePrompt", "prompt"]) {
|
||||
const value = asString(task[key]).replace(/\s+/gu, " ").trim();
|
||||
if (value.length > 0) return value.slice(0, 96);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function buildQueueLookup(queues: Record<string, unknown>[]): Map<string, Record<string, unknown>> {
|
||||
const lookup = new Map<string, Record<string, unknown>>();
|
||||
for (const queue of queues) {
|
||||
const id = asString(queue.id);
|
||||
if (id.length > 0) lookup.set(id, queue);
|
||||
}
|
||||
return lookup;
|
||||
}
|
||||
|
||||
function buildTaskLookupFromQueuesBody(body: Record<string, unknown>): Map<string, Record<string, unknown>> {
|
||||
const lookup = new Map<string, Record<string, unknown>>();
|
||||
for (const source of [body.tasks, asRecord(body.overview)?.tasks, asRecord(body.taskPage)?.tasks]) {
|
||||
for (const item of asArray(source)) {
|
||||
const task = asRecord(item);
|
||||
const id = asString(task?.id);
|
||||
if (task !== null && id.length > 0 && !lookup.has(id)) lookup.set(id, task);
|
||||
}
|
||||
}
|
||||
return lookup;
|
||||
}
|
||||
|
||||
function queueIdsForTaskId(taskId: string, queues: Record<string, unknown>[]): string[] {
|
||||
const ids: string[] = [];
|
||||
for (const queue of queues) {
|
||||
const queueId = asString(queue.id);
|
||||
if (queueId.length === 0) continue;
|
||||
if (asString(queue.activeTaskId) === taskId || asString(queue.runnableTaskId) === taskId) ids.push(queueId);
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
function compactCommanderTaskList(
|
||||
ids: string[],
|
||||
queues: Record<string, unknown>[],
|
||||
queueLookup: Map<string, Record<string, unknown>>,
|
||||
taskLookup: Map<string, Record<string, unknown>>,
|
||||
limit: number,
|
||||
source: string,
|
||||
totalCount?: number,
|
||||
): Record<string, unknown> {
|
||||
const all = orderedUniqueStringList(ids);
|
||||
const count = Math.max(all.length, totalCount ?? all.length);
|
||||
const items = all.slice(0, limit).map((id) => {
|
||||
const task = taskLookup.get(id) ?? null;
|
||||
const taskQueueId = asString(task?.queueId) || asString(task?.queue) || null;
|
||||
const inferredQueueIds = taskQueueId === null ? queueIdsForTaskId(id, queues) : [taskQueueId];
|
||||
const queueRecords = inferredQueueIds.map((queueId) => queueLookup.get(queueId) ?? null).filter((queue): queue is Record<string, unknown> => queue !== null);
|
||||
const queueNames = queueRecords.map((queue) => asString(queue.name)).filter((name) => name.length > 0);
|
||||
const name = taskNameFromRecord(task);
|
||||
return {
|
||||
id,
|
||||
name,
|
||||
nameSource: name === null ? "not-returned-by-api-queues" : "task-row",
|
||||
queueIds: inferredQueueIds,
|
||||
queueNames,
|
||||
queueSource: taskQueueId !== null
|
||||
? "task-row"
|
||||
: inferredQueueIds.length > 0
|
||||
? "queue-row-active-or-runnable-task-id"
|
||||
: "not-returned-by-api-queues",
|
||||
commands: {
|
||||
show: `bun scripts/cli.ts codex task ${id}`,
|
||||
detail: `bun scripts/cli.ts codex task ${id} --detail`,
|
||||
trace: `bun scripts/cli.ts codex task ${id} --trace --tail --limit ${defaultTraceLimit}`,
|
||||
},
|
||||
};
|
||||
});
|
||||
return {
|
||||
count,
|
||||
returned: items.length,
|
||||
omitted: Math.max(0, count - items.length),
|
||||
truncated: count > items.length || all.length > items.length,
|
||||
idsUnavailable: count > 0 && items.length === 0,
|
||||
source,
|
||||
items,
|
||||
};
|
||||
}
|
||||
|
||||
function sortedQueueRowsForTaskStatus(queues: Record<string, unknown>[], statuses: string[]): Record<string, unknown>[] {
|
||||
return queues
|
||||
.filter((queue) => statuses.some((status) => countForStatus(asRecord(queue.counts) ?? {}, status) > 0))
|
||||
.sort((left, right) => {
|
||||
const leftUpdated = timestampMs(left.updatedAt) ?? 0;
|
||||
const rightUpdated = timestampMs(right.updatedAt) ?? 0;
|
||||
return rightUpdated - leftUpdated || asString(left.id).localeCompare(asString(right.id));
|
||||
});
|
||||
}
|
||||
|
||||
function compactCommanderQueueRows(queues: Record<string, unknown>[], limit: number, statuses: string[], source: string): Record<string, unknown> {
|
||||
const selected = sortedQueueRowsForTaskStatus(queues, statuses);
|
||||
const items = selected.slice(0, limit).map((queue) => ({
|
||||
id: queue.id ?? null,
|
||||
name: queue.name ?? null,
|
||||
counts: queue.counts ?? {},
|
||||
activeTaskId: queue.activeTaskId ?? null,
|
||||
runnableTaskId: queue.runnableTaskId ?? null,
|
||||
unreadTerminal: queue.unreadTerminal ?? 0,
|
||||
updatedAt: queue.updatedAt ?? null,
|
||||
commands: queue.commands ?? {},
|
||||
}));
|
||||
return {
|
||||
count: selected.length,
|
||||
returned: items.length,
|
||||
omitted: Math.max(0, selected.length - items.length),
|
||||
truncated: selected.length > items.length,
|
||||
source,
|
||||
items,
|
||||
};
|
||||
}
|
||||
|
||||
function compactCommanderHeartbeatList(ids: string[], totalCount: number, limit: number, source: string, command: string): Record<string, unknown> {
|
||||
const knownIds = orderedUniqueStringList(ids);
|
||||
const preview = compactIdPreview(knownIds, Math.max(totalCount, knownIds.length), limit, source);
|
||||
return {
|
||||
...preview,
|
||||
command,
|
||||
};
|
||||
}
|
||||
|
||||
function heartbeatIdsFromDiagnostics(rawDiagnostics: Record<string, unknown>, diagnostics: Record<string, unknown> | null, keys: string[]): string[] {
|
||||
const ids: string[] = [];
|
||||
for (const key of keys) {
|
||||
ids.push(...idPreviewInputItems(rawDiagnostics[key]));
|
||||
ids.push(...idPreviewInputItems(diagnostics?.[key]));
|
||||
}
|
||||
return orderedUniqueStringList(ids);
|
||||
}
|
||||
|
||||
function buildQueuesCommanderSummary(input: {
|
||||
queue: Record<string, unknown>;
|
||||
queues: Record<string, unknown>[];
|
||||
diagnostics: Record<string, unknown> | null;
|
||||
activity: Record<string, unknown>;
|
||||
commanderConcurrency: Record<string, unknown>;
|
||||
activeTaskIds: { items: string[]; count: number; omitted: number; truncated: boolean };
|
||||
queuedTaskIds: { items: string[]; count: number; omitted: number; truncated: boolean };
|
||||
options: CodexQueuesOptions;
|
||||
body: Record<string, unknown>;
|
||||
}): Record<string, unknown> {
|
||||
const rawDiagnostics = asRecord(input.queue.executionDiagnostics) ?? {};
|
||||
const liveness = asRecord(input.diagnostics?.liveness) ?? {};
|
||||
const recovery = asRecord(input.activity.recovery) ?? asRecord(input.diagnostics?.recovery) ?? {};
|
||||
const activeRunnerCount = asNumber(input.commanderConcurrency.activeRunnerCount, asNumber(input.activity.effectiveActiveTaskCount, 0));
|
||||
const target = commanderConcurrencyTarget;
|
||||
const slotDeficit = Math.max(0, target - activeRunnerCount);
|
||||
const slotDeficitState = activeRunnerCount >= target ? "at-or-above-target" : "below-target";
|
||||
const queuedCount = countForStatus(asRecord(input.queue.counts) ?? {}, "queued") + countForStatus(asRecord(input.queue.counts) ?? {}, "retry_wait");
|
||||
const queueLookup = buildQueueLookup(input.queues);
|
||||
const taskLookup = buildTaskLookupFromQueuesBody(input.body);
|
||||
const activeIds = orderedUniqueStringList([
|
||||
...input.activeTaskIds.items,
|
||||
...idPreviewInputItems(rawDiagnostics.databaseActiveTaskIds),
|
||||
...idPreviewInputItems(rawDiagnostics.heartbeatFreshTaskIds),
|
||||
...idPreviewInputItems(input.diagnostics?.heartbeatFreshTaskIds),
|
||||
]);
|
||||
const queuedIds = orderedUniqueStringList(input.queuedTaskIds.items);
|
||||
const heartbeatFreshIds = heartbeatIdsFromDiagnostics(rawDiagnostics, input.diagnostics, ["heartbeatFreshTaskIds", "activeHeartbeatTaskIds"]);
|
||||
const heartbeatRiskIds = heartbeatIdsFromDiagnostics(rawDiagnostics, input.diagnostics, ["heartbeatRiskTaskIds", "heartbeatExpiredTaskIds", "heartbeatMissingTaskIds"]);
|
||||
const staleCandidateIds = heartbeatIdsFromDiagnostics(rawDiagnostics, input.diagnostics, ["staleRecoveryCandidateTaskIds"]);
|
||||
const taskPreviewLimit = Math.min(input.options.limit, commanderSectionReturnedLimit);
|
||||
return {
|
||||
view: "commander",
|
||||
fieldOrder: [
|
||||
"activeRunnerCount",
|
||||
"source",
|
||||
"target",
|
||||
"slotDeficit",
|
||||
"runningTasks",
|
||||
"heartbeat",
|
||||
"queuedCount",
|
||||
"queueRows",
|
||||
"historyRows",
|
||||
],
|
||||
activeRunnerCount,
|
||||
source: input.commanderConcurrency.activeRunnerCountSource ?? input.activity.effectiveActiveSource ?? "activity.effectiveActiveTaskCount",
|
||||
activeRunnerCountField: input.commanderConcurrency.activeRunnerCountField ?? "activity.effectiveActiveTaskCount",
|
||||
target,
|
||||
slotDeficit,
|
||||
slotDeficitState,
|
||||
queuedCount,
|
||||
queuedSource: "queue.counts.queued + queue.counts.retry_wait",
|
||||
counts: input.queue.counts ?? {},
|
||||
runningTasks: compactCommanderTaskList(activeIds, input.queues, queueLookup, taskLookup, taskPreviewLimit, "activeTaskIds + executionDiagnostics.databaseActiveTaskIds + heartbeatFreshTaskIds", activeRunnerCount),
|
||||
queuedTasks: compactCommanderTaskList(queuedIds, input.queues, queueLookup, taskLookup, taskPreviewLimit, "queue.queuedTaskIds", queuedCount),
|
||||
activeQueues: compactCommanderQueueRows(input.queues, taskPreviewLimit, ["running", "judging"], "per-queue counts running/judging"),
|
||||
runnableQueues: compactCommanderQueueRows(input.queues, taskPreviewLimit, ["queued", "retry_wait"], "per-queue counts queued/retry_wait"),
|
||||
heartbeat: {
|
||||
effectiveLiveness: input.diagnostics?.effectiveLiveness ?? input.activity.effectiveLiveness ?? null,
|
||||
splitBrainLive: input.diagnostics?.splitBrainLive ?? input.activity.splitBrainLive ?? null,
|
||||
lastSchedulerHeartbeatAt: input.diagnostics?.lastSchedulerHeartbeatAt ?? recovery.lastSchedulerHeartbeatAt ?? null,
|
||||
lastObservedAgentEventAt: input.diagnostics?.lastObservedAgentEventAt ?? recovery.lastObservedAgentEventAt ?? null,
|
||||
fresh: compactCommanderHeartbeatList(heartbeatFreshIds, asNumber(input.activity.heartbeatFreshActiveTaskCount, heartbeatFreshIds.length), taskPreviewLimit, "executionDiagnostics.heartbeatFreshTaskIds + activeHeartbeatTaskIds", "bun scripts/cli.ts codex tasks --view supervisor --status running,judging --limit 20"),
|
||||
risk: compactCommanderHeartbeatList(heartbeatRiskIds, asNumber(input.activity.heartbeatRiskTaskCount, heartbeatRiskIds.length), taskPreviewLimit, "executionDiagnostics.heartbeatRiskTaskIds + heartbeatExpiredTaskIds + heartbeatMissingTaskIds", "bun scripts/cli.ts codex tasks --view supervisor --limit 20"),
|
||||
staleRecoveryCandidates: compactCommanderHeartbeatList(staleCandidateIds, asNumber(input.activity.staleRecoveryCandidateTaskCount, staleCandidateIds.length), taskPreviewLimit, "executionDiagnostics.staleRecoveryCandidateTaskIds", "bun scripts/cli.ts microservice proxy code-queue '/api/scheduler/reconcile?dryRun=1' --raw"),
|
||||
recoveryHint: input.commanderConcurrency.recoveryHint ?? recovery.hint ?? liveness.recoveryHint ?? null,
|
||||
},
|
||||
attentionRequired: input.commanderConcurrency.attentionRequired ?? false,
|
||||
interventionRequired: input.commanderConcurrency.interventionRequired ?? false,
|
||||
splitBrainDisposition: input.commanderConcurrency.splitBrainDisposition ?? input.activity.splitBrainDisposition ?? null,
|
||||
decisionRule: `remaining slots = ${target} - activeRunnerCount; slotDeficit is clamped at 0 when activeRunnerCount >= target.`,
|
||||
historyRows: {
|
||||
returned: Math.min(input.options.limit, input.queues.length),
|
||||
secondary: true,
|
||||
stablePath: "data.queues.items[]",
|
||||
note: "Queue history rows are paged and secondary; use commander.runningTasks, heartbeat, and queuedCount first for concurrency decisions.",
|
||||
},
|
||||
commands: {
|
||||
refresh: queueListCommand({ full: input.options.full, commander: input.options.commander, limit: input.options.limit, offset: input.options.offset }),
|
||||
supervisor: "bun scripts/cli.ts codex tasks --view supervisor --limit 20",
|
||||
running: "bun scripts/cli.ts codex tasks --view supervisor --status running,judging --limit 20",
|
||||
unread: "bun scripts/cli.ts codex unread --limit 20",
|
||||
fullQueues: queueListCommand({ full: true, limit: input.options.limit, offset: 0 }),
|
||||
rawQueues: "bun scripts/cli.ts microservice proxy code-queue /api/queues --raw --full",
|
||||
rawOverview: rawCodeQueueOverviewCommand,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function queueListCommand(options: Partial<CodexQueuesOptions> = {}): string {
|
||||
const full = options.full === true;
|
||||
const limit = options.limit ?? defaultQueuesLimit;
|
||||
@@ -4674,6 +4904,7 @@ function queueListCommand(options: Partial<CodexQueuesOptions> = {}): string {
|
||||
return [
|
||||
"bun scripts/cli.ts codex queues",
|
||||
full ? "--full" : "",
|
||||
!full && options.commander === true ? "--commander" : "",
|
||||
limit === defaultQueuesLimit ? "" : `--limit ${limit}`,
|
||||
offset > 0 ? `--offset ${offset}` : "",
|
||||
].filter(Boolean).join(" ");
|
||||
@@ -4698,16 +4929,30 @@ function compactQueuesResponse(body: Record<string, unknown>, options: CodexQueu
|
||||
const previousOffset = Math.max(0, options.offset - options.limit);
|
||||
const hasMore = nextOffset < selected.length;
|
||||
const hasPrevious = options.offset > 0;
|
||||
const commander = buildQueuesCommanderSummary({
|
||||
queue,
|
||||
queues,
|
||||
diagnostics,
|
||||
activity,
|
||||
commanderConcurrency,
|
||||
activeTaskIds,
|
||||
queuedTaskIds,
|
||||
options,
|
||||
body,
|
||||
});
|
||||
return {
|
||||
upstream,
|
||||
queues: {
|
||||
view: options.full ? "full" : "summary",
|
||||
summaryMode: options.commander ? "commander-first" : "full-paged",
|
||||
bounded: true,
|
||||
outputPolicy: {
|
||||
default: "paged-low-noise",
|
||||
default: options.commander ? "commander-first-paged-low-noise" : "paged-low-noise",
|
||||
stableItemsPath: "data.queues.items[]",
|
||||
commanderPath: "data.queues.commander",
|
||||
rawFullCommand: "bun scripts/cli.ts microservice proxy code-queue /api/queues --raw --full",
|
||||
},
|
||||
commander,
|
||||
count: selected.length,
|
||||
returned: visible.length,
|
||||
limit: options.limit,
|
||||
@@ -4757,10 +5002,10 @@ function compactQueuesResponse(body: Record<string, unknown>, options: CodexQueu
|
||||
}
|
||||
: {}),
|
||||
commands: {
|
||||
refresh: queueListCommand({ full: options.full, limit: options.limit, offset: options.offset }),
|
||||
next: hasMore ? queueListCommand({ full: options.full, limit: options.limit, offset: nextOffset }) : null,
|
||||
previous: hasPrevious ? queueListCommand({ full: options.full, limit: options.limit, offset: previousOffset }) : null,
|
||||
first: queueListCommand({ full: options.full, limit: options.limit, offset: 0 }),
|
||||
refresh: queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: options.offset }),
|
||||
next: hasMore ? queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: nextOffset }) : null,
|
||||
previous: hasPrevious ? queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: previousOffset }) : null,
|
||||
first: queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: 0 }),
|
||||
full: queueListCommand({ full: true, limit: options.limit, offset: 0 }),
|
||||
tasks: `bun scripts/cli.ts codex tasks --view supervisor --limit ${Math.min(options.limit, defaultTasksLimit)}`,
|
||||
unread: `bun scripts/cli.ts codex unread --limit ${Math.min(options.limit, defaultTasksLimit)}`,
|
||||
|
||||
Reference in New Issue
Block a user