fix: reduce commander task polling noise (#157)
Co-authored-by: Codex <codex@noreply.local>
This commit is contained in:
Lyon
+154
-41
@@ -32,10 +32,9 @@ const supervisorPromptPreviewChars = 70;
|
||||
const supervisorBodyPreviewChars = 70;
|
||||
const supervisorRecentBodyPreviewChars = 50;
|
||||
const commanderAttentionLimit = 10;
|
||||
const commanderActiveItemLimit = 8;
|
||||
const commanderSectionReturnedLimit = 5;
|
||||
const commanderRecentCompletedLimit = 3;
|
||||
const commanderPromptPreviewChars = 96;
|
||||
const commanderBodyPreviewChars = 120;
|
||||
const commanderIssueTaskPreviewLimit = 4;
|
||||
const commanderConcurrencyTarget = 15;
|
||||
const unreadTriageCountLimit = 12;
|
||||
@@ -442,13 +441,6 @@ interface CommanderAttentionItem {
|
||||
finishedAt?: string | null;
|
||||
unreadTerminal?: boolean;
|
||||
finalResponseAt?: unknown;
|
||||
prompt: string;
|
||||
promptChars: number;
|
||||
promptTruncated?: boolean;
|
||||
last?: string;
|
||||
lastAt?: unknown;
|
||||
lastChars?: number;
|
||||
lastTruncated?: boolean;
|
||||
commands: {
|
||||
show: string;
|
||||
detail: string;
|
||||
@@ -2281,6 +2273,42 @@ function supervisorExecutionDiagnostics(value: unknown): Record<string, unknown>
|
||||
};
|
||||
}
|
||||
|
||||
function commanderExecutionDiagnostics(value: unknown): Record<string, unknown> | null {
|
||||
const diagnostics = supervisorExecutionDiagnostics(value);
|
||||
if (diagnostics === null) return null;
|
||||
const listBudget = asRecord(diagnostics.listBudget) ?? {};
|
||||
const recovery = asRecord(diagnostics.recovery) ?? {};
|
||||
return {
|
||||
state: diagnostics.state ?? null,
|
||||
effectiveLiveness: diagnostics.effectiveLiveness ?? null,
|
||||
recommendedAction: diagnostics.recommendedAction ?? null,
|
||||
splitBrainLive: diagnostics.splitBrainLive ?? null,
|
||||
activeHeartbeatCount: diagnostics.activeHeartbeatCount ?? null,
|
||||
databaseActiveTaskCount: diagnostics.databaseActiveTaskCount ?? null,
|
||||
schedulerActiveRunSlotCount: diagnostics.schedulerActiveRunSlotCount ?? null,
|
||||
heartbeatFreshTaskIds: diagnostics.heartbeatFreshTaskIds ?? [],
|
||||
heartbeatRiskTaskIds: diagnostics.heartbeatRiskTaskIds ?? [],
|
||||
traceGapTaskIds: diagnostics.traceGapTaskIds ?? [],
|
||||
recovery: {
|
||||
disposition: recovery.disposition ?? null,
|
||||
hint: recovery.hint ?? null,
|
||||
rePollBeforeRecovery: recovery.rePollBeforeRecovery ?? null,
|
||||
repeatedPollConfirmed: recovery.repeatedPollConfirmed ?? null,
|
||||
recoveryMutationAllowedByThisSnapshot: recovery.recoveryMutationAllowedByThisSnapshot ?? null,
|
||||
heartbeatRiskTaskCount: recovery.heartbeatRiskTaskCount ?? null,
|
||||
staleRecoveryCandidateTaskCount: recovery.staleRecoveryCandidateTaskCount ?? null,
|
||||
nextPollCommand: recovery.nextPollCommand ?? null,
|
||||
dryRunReconcileCommand: recovery.dryRunReconcileCommand ?? null,
|
||||
},
|
||||
interpretation: diagnostics.interpretation ?? null,
|
||||
listBudget: {
|
||||
idPreviewLimit: listBudget.idPreviewLimit ?? diagnosticsIdPreviewLimit,
|
||||
truncated: listBudget.truncated ?? false,
|
||||
rawCommand: listBudget.rawCommand ?? "bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview?limit=30 --raw --full",
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function compactToolSummary(value: unknown, full: boolean, limit = defaultToolLimit): Record<string, unknown> {
|
||||
const record = asRecord(value) ?? {};
|
||||
const allItems = asArray(record.items);
|
||||
@@ -3945,7 +3973,7 @@ function commanderInfrastructureSignals(rawQueue: Record<string, unknown>): Reco
|
||||
source: asString(signal.source) || "code-queue",
|
||||
}));
|
||||
const storageDegraded = health.degraded === true || storage.postgresReady === false || storage.lastError !== null && storage.lastError !== undefined;
|
||||
return {
|
||||
const result: Record<string, unknown> = {
|
||||
infrastructureBlocker: storageDegraded || boundedSignals.some((signal) => signal.category === "infrastructure-blocker"),
|
||||
status: health.status ?? (storageDegraded ? "degraded" : "ready"),
|
||||
source: "queue.storage.health",
|
||||
@@ -3953,7 +3981,12 @@ function commanderInfrastructureSignals(rawQueue: Record<string, unknown>): Reco
|
||||
omittedSignalCount: Math.max(0, rawSignals.length - boundedSignals.length),
|
||||
bounded: true,
|
||||
signals: boundedSignals,
|
||||
storage: {
|
||||
actionable: storageDegraded
|
||||
? "Treat as Code Queue infrastructure-blocker; inspect storage health/logs and wait for bounded dirty-flush retry before duplicating or canceling business tasks."
|
||||
: "No Code Queue storage infrastructure blocker reported in this overview page.",
|
||||
};
|
||||
if (storageDegraded) {
|
||||
result.storage = {
|
||||
postgresReady: storage.postgresReady ?? health.postgresReady ?? null,
|
||||
dirtyTaskCount: storage.dirtyTaskCount ?? health.dirtyTaskCount ?? null,
|
||||
dirtyQueueCount: storage.dirtyQueueCount ?? health.dirtyQueueCount ?? null,
|
||||
@@ -3964,11 +3997,9 @@ function commanderInfrastructureSignals(rawQueue: Record<string, unknown>): Reco
|
||||
lastClientRotationAt: health.lastClientRotationAt ?? null,
|
||||
lastErrorKind: health.lastErrorKind ?? null,
|
||||
lastErrorTransient: health.lastErrorTransient ?? null,
|
||||
},
|
||||
actionable: storageDegraded
|
||||
? "Treat as Code Queue infrastructure-blocker; inspect storage health/logs and wait for bounded dirty-flush retry before duplicating or canceling business tasks."
|
||||
: "No Code Queue storage infrastructure blocker reported in this overview page.",
|
||||
};
|
||||
};
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function commanderAttentionReasons(
|
||||
@@ -4054,8 +4085,6 @@ function commanderAttentionItem(
|
||||
const status = asString(task.status) || null;
|
||||
const summaryLastAssistant = summary?.lastAssistantMessage ?? task.lastAssistantMessage;
|
||||
const awaitingStatus = finalResponseAwaitingTerminalStatus(status, summaryLastAssistant);
|
||||
const prompt = supervisorTextSummary(asString(task.displayPrompt ?? task.basePrompt ?? task.prompt), commanderPromptPreviewChars);
|
||||
const lastMessage = supervisorLastMessage(summaryLastAssistant, commanderBodyPreviewChars);
|
||||
const issues = taskIssueRefs(task, summary);
|
||||
const unreadTerminal = taskUnreadTerminal(task);
|
||||
return {
|
||||
@@ -4076,15 +4105,6 @@ function commanderAttentionItem(
|
||||
attempt: typeof task.currentAttempt === "number" && Number.isFinite(task.currentAttempt) ? task.currentAttempt : null,
|
||||
updatedAt: asString(task.updatedAt) || null,
|
||||
...(isTerminalTaskStatus(status) ? { finishedAt: asString(task.finishedAt) || null, unreadTerminal } : {}),
|
||||
prompt: prompt.text,
|
||||
promptChars: prompt.chars,
|
||||
...(prompt.truncated ? { promptTruncated: true } : {}),
|
||||
...(lastMessage === null ? {} : {
|
||||
last: lastMessage.text,
|
||||
lastAt: lastMessage.at,
|
||||
lastChars: lastMessage.chars,
|
||||
...(lastMessage.truncated ? { lastTruncated: true } : {}),
|
||||
}),
|
||||
commands: taskDrilldownCommands(taskId, unreadTerminal),
|
||||
};
|
||||
}
|
||||
@@ -4100,6 +4120,23 @@ function commanderAttentionRank(item: CommanderAttentionItem): number {
|
||||
return severityRank[item.severity] * 10 + actionRank[item.action];
|
||||
}
|
||||
|
||||
function compactCommanderAttentionItem(item: CommanderAttentionItem): Record<string, unknown> {
|
||||
return {
|
||||
id: item.id,
|
||||
queue: item.queue,
|
||||
status: item.status,
|
||||
...(item.statusLabel === undefined ? {} : { statusLabel: item.statusLabel }),
|
||||
severity: item.severity,
|
||||
action: item.action,
|
||||
riskSignals: item.riskSignals,
|
||||
issues: item.issues,
|
||||
highPriorityIssues: item.highPriorityIssues,
|
||||
category: item.classification.category,
|
||||
updatedAt: item.updatedAt,
|
||||
...(item.finalResponseAt === undefined ? {} : { finalResponseAt: item.finalResponseAt }),
|
||||
};
|
||||
}
|
||||
|
||||
function commanderIdSection(tasks: Record<string, unknown>[], summaries: Map<string, Record<string, unknown>>, limit: number, nextCommand: string | null, fullCommand: string): Record<string, unknown> {
|
||||
const visible = tasks.slice(0, limit);
|
||||
return {
|
||||
@@ -4111,10 +4148,6 @@ function commanderIdSection(tasks: Record<string, unknown>[], summaries: Map<str
|
||||
commands: {
|
||||
next: tasks.length > visible.length || nextCommand !== null ? nextCommand : null,
|
||||
full: fullCommand,
|
||||
showTemplate: "bun scripts/cli.ts codex task <taskId>",
|
||||
traceTemplate: `bun scripts/cli.ts codex task <taskId> --trace --tail --limit ${defaultTraceLimit}`,
|
||||
outputTemplate: `bun scripts/cli.ts codex output <taskId> --tail --limit ${defaultOutputLimit}`,
|
||||
readTemplate: "bun scripts/cli.ts codex read <taskId>",
|
||||
},
|
||||
items: visible.map((task) => {
|
||||
const taskId = taskOverviewCandidateKey(task);
|
||||
@@ -4146,9 +4179,8 @@ function commanderClassificationCounts(tasks: Record<string, unknown>[], summari
|
||||
return {
|
||||
byCategory,
|
||||
byNoiseClass,
|
||||
categories: ["business-user-facing", "deployment-artifact", "ci-e2e-evidence", "diagnostics-gate-report", "docs-governance", "infrastructure-blocker", "unknown"],
|
||||
deterministic: true,
|
||||
sourceFields: ["task prompt previews", "task metadata", "summary lastAssistantMessage preview when fetched"],
|
||||
disclosure: "counts only; use --view full or codex task <taskId> for per-task classification evidence",
|
||||
};
|
||||
}
|
||||
|
||||
@@ -4198,6 +4230,83 @@ function attentionCounts(items: CommanderAttentionItem[], returnedItems: Command
|
||||
};
|
||||
}
|
||||
|
||||
function activeRunnerItem(task: Record<string, unknown>, summary: Record<string, unknown> | null, diagnostics: Record<string, unknown>): Record<string, unknown> {
|
||||
const taskId = taskOverviewCandidateKey(task);
|
||||
const status = asString(task.status) || null;
|
||||
const summaryLastAssistant = summary?.lastAssistantMessage ?? task.lastAssistantMessage;
|
||||
const awaitingStatus = finalResponseAwaitingTerminalStatus(status, summaryLastAssistant);
|
||||
const attention = commanderAttentionReasons(task, summary, diagnostics);
|
||||
const issues = taskIssueRefs(task, summary);
|
||||
return {
|
||||
id: taskId,
|
||||
queue: asString(task.queueId) || null,
|
||||
status,
|
||||
...(awaitingStatus === null ? {} : {
|
||||
statusLabel: awaitingStatus.label,
|
||||
closeoutState: awaitingStatus.state,
|
||||
finalResponseAt: awaitingStatus.finalResponseAt,
|
||||
}),
|
||||
severity: attention?.severity ?? null,
|
||||
riskSignals: attention?.riskSignals ?? [],
|
||||
issues,
|
||||
highPriorityIssues: highPriorityIssueRefs(issues),
|
||||
category: commanderTaskClassification(task, summary).category,
|
||||
attempt: typeof task.currentAttempt === "number" && Number.isFinite(task.currentAttempt) ? task.currentAttempt : null,
|
||||
updatedAt: asString(task.updatedAt) || null,
|
||||
};
|
||||
}
|
||||
|
||||
function commanderActiveRunnerItems(
|
||||
tasks: Record<string, unknown>[],
|
||||
summaries: Map<string, Record<string, unknown>>,
|
||||
diagnostics: Record<string, unknown>,
|
||||
options: CodexTasksOptions,
|
||||
): Record<string, unknown> {
|
||||
const visible = tasks.slice(0, commanderActiveItemLimit);
|
||||
const hasMore = tasks.length > visible.length;
|
||||
const runningOptions: CodexTasksOptions = {
|
||||
...baseTaskListOptions({ ...options, unreadOnly: false, beforeId: undefined }),
|
||||
statusFilter: ["running", "judging"],
|
||||
};
|
||||
return {
|
||||
returned: visible.length,
|
||||
omitted: Math.max(0, tasks.length - visible.length),
|
||||
truncated: hasMore,
|
||||
itemLimit: commanderActiveItemLimit,
|
||||
outputPolicy: "compact active rows only; prompt, final response, trace, and output require drill-down commands",
|
||||
commands: {
|
||||
running: taskListCommandWithView(runningOptions, "supervisor"),
|
||||
full: taskListCommandWithView(runningOptions, "full"),
|
||||
},
|
||||
items: visible.map((task) => activeRunnerItem(task, summaries.get(taskOverviewCandidateKey(task)) ?? null, diagnostics)),
|
||||
};
|
||||
}
|
||||
|
||||
function commanderTerminalUnreadSection(
|
||||
total: number,
|
||||
unreadRowsOnFetchedPage: Record<string, unknown>[],
|
||||
options: CodexTasksOptions,
|
||||
): Record<string, unknown> {
|
||||
return {
|
||||
count: total,
|
||||
returned: 0,
|
||||
omitted: total,
|
||||
truncated: total > 0,
|
||||
hasMore: total > 0,
|
||||
fetchedRowsOnOverviewPage: unreadRowsOnFetchedPage.length,
|
||||
outputPolicy: "terminal unread task details are intentionally omitted from the default commander poll; use codex unread, supervisor unread, or full drill-down.",
|
||||
commands: {
|
||||
next: total > 0 ? `bun scripts/cli.ts codex unread --limit ${Math.min(options.requestedLimit, defaultTasksLimit)}` : null,
|
||||
unread: `bun scripts/cli.ts codex unread --limit ${Math.min(options.requestedLimit, defaultTasksLimit)}`,
|
||||
supervisor: taskListCommandWithView({ ...options, unreadOnly: true }, "supervisor"),
|
||||
full: taskListCommandWithView({ ...options, unreadOnly: true }, "full"),
|
||||
showTemplate: "bun scripts/cli.ts codex task <taskId>",
|
||||
readTemplate: "bun scripts/cli.ts codex read <taskId>",
|
||||
},
|
||||
items: [],
|
||||
};
|
||||
}
|
||||
|
||||
function terminalUnreadAggregateCount(taskPage: CodexTasksTaskPage, options: CodexTasksOptions, fallback: number): { total: number; exact: boolean; source: string } {
|
||||
const queue = taskPage.queue;
|
||||
if (queue !== null && options.queueId === undefined) {
|
||||
@@ -4242,6 +4351,7 @@ function codexTasksCommanderResult(
|
||||
const activeSection = buildSupervisorTaskSection(runningTasks, summaries, taskSectionLimit(options), sectionNextCommand(runningTasks, taskSectionLimit(options), options, nextCommand), fullCommand);
|
||||
const activeRunning = supervisorActiveRunningSummary(taskPage, options, activeSection, diagnostics);
|
||||
const attentionItems = allTasks
|
||||
.filter((task) => !taskUnreadTerminal(task))
|
||||
.map((task) => commanderAttentionItem(task, summaries.get(taskOverviewCandidateKey(task)) ?? null, rawDiagnostics))
|
||||
.filter((item): item is CommanderAttentionItem => item !== null)
|
||||
.sort((left, right) => {
|
||||
@@ -4276,11 +4386,11 @@ function codexTasksCommanderResult(
|
||||
bounded: true,
|
||||
disclosure: {
|
||||
recommendedFor: "host commander supervision loops",
|
||||
policy: "bounded action map only; no full prompt, final response, trace, output, or raw overview body is included by default",
|
||||
policy: "low-noise polling summary only; no full prompt, final response, terminal unread detail, trace, output, or raw overview body is included by default",
|
||||
attentionLimit: commanderAttentionLimit,
|
||||
activeItemLimit: commanderActiveItemLimit,
|
||||
sectionReturnedLimit: commanderSectionReturnedLimit,
|
||||
promptPreviewChars: commanderPromptPreviewChars,
|
||||
bodyPreviewChars: commanderBodyPreviewChars,
|
||||
terminalUnreadDetails: "omitted-by-default; use commands.unread, sections.terminalUnread.commands.supervisor, --view full, or codex read <taskId>",
|
||||
rawOverview: `bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview${tasksListQueryString(options)} --raw`,
|
||||
},
|
||||
activeRunners: {
|
||||
@@ -4302,6 +4412,7 @@ function codexTasksCommanderResult(
|
||||
heartbeatFreshActive: activity.heartbeatFreshActiveTaskCount,
|
||||
schedulerLocalActiveQueues: activity.schedulerLocalActiveQueueCount,
|
||||
schedulerLocalActiveRunSlots: activity.schedulerLocalActiveRunSlotCount,
|
||||
...commanderActiveRunnerItems(runningTasks, summaries, rawDiagnostics, options),
|
||||
},
|
||||
queueBacklog: {
|
||||
queued,
|
||||
@@ -4334,7 +4445,7 @@ function codexTasksCommanderResult(
|
||||
highPriorityIssues: commanderHighPriorityIssues(allTasks, summaries),
|
||||
classification: commanderClassificationCounts(allTasks, summaries),
|
||||
infrastructure,
|
||||
executionDiagnostics: diagnostics,
|
||||
executionDiagnostics: commanderExecutionDiagnostics(rawDiagnostics),
|
||||
degraded,
|
||||
commands: {
|
||||
refresh: taskListCommand(options),
|
||||
@@ -4353,11 +4464,11 @@ function codexTasksCommanderResult(
|
||||
attention: {
|
||||
...attentionCounts(attentionItems, returnedAttention),
|
||||
truncated: attentionItems.length > returnedAttention.length,
|
||||
items: returnedAttention,
|
||||
items: returnedAttention.map(compactCommanderAttentionItem),
|
||||
},
|
||||
sections: {
|
||||
activeNeedsAttention: commanderIdSection(activeRiskTasks, summaries, commanderSectionReturnedLimit, taskListCommandWithView({ ...options, statusFilter: ["running", "judging"] }, "supervisor"), fullCommand),
|
||||
terminalUnread: commanderIdSection(unreadCompletedTasks, summaries, commanderSectionReturnedLimit, taskListCommandWithView({ ...options, unreadOnly: true }, "supervisor"), fullCommand),
|
||||
terminalUnread: commanderTerminalUnreadSection(terminalUnreadAggregate.total, unreadCompletedTasks, options),
|
||||
queuedRetryWait: commanderIdSection(queuedRetryTasks, summaries, commanderSectionReturnedLimit, taskListCommandWithView({ ...options, statusFilter: ["queued", "retry_wait"] }, "supervisor"), fullCommand),
|
||||
recentCompleted: commanderIdSection(recentCompletedTasks, summaries, commanderRecentCompletedLimit, nextCommand, fullCommand),
|
||||
},
|
||||
@@ -4527,8 +4638,10 @@ function visibleTaskIdsForOverview(tasks: Record<string, unknown>[], options: Co
|
||||
const sectionLimit = taskSectionLimit(options);
|
||||
if (options.view === "commander") {
|
||||
return Array.from(new Set([
|
||||
...sortRunningWatchTasks(filtered),
|
||||
...sortCompletedWatchTasks(filtered).filter((task) => taskUnreadTerminal(task)),
|
||||
...sortRunningWatchTasks(filtered).slice(0, commanderActiveItemLimit),
|
||||
...sortRunningWatchTasks(filtered)
|
||||
.filter((task) => commanderAttentionReasons(task, null, {}) !== null)
|
||||
.slice(0, commanderAttentionLimit),
|
||||
...sortQueuedWatchTasks(filtered).slice(0, commanderSectionReturnedLimit),
|
||||
...sortCompletedWatchTasks(filtered).filter((task) => !taskUnreadTerminal(task)).slice(0, commanderRecentCompletedLimit),
|
||||
].map((task) => taskOverviewCandidateKey(task))))
|
||||
|
||||
Reference in New Issue
Block a user