fix: clarify code queue activity counts

2026-05-22 23:42:39 +00:00
parent f3b5b449a2
commit 031c3fda39
6 changed files with 305 additions and 6 deletions
@@ -433,6 +433,10 @@ function asNumber(value: unknown, fallback = 0): number {
  return typeof value === "number" && Number.isFinite(value) ? value : fallback;
 }

+function finiteNumber(value: unknown): number | null {
+  return typeof value === "number" && Number.isFinite(value) ? value : null;
+}
+
 function asBoolean(value: unknown): boolean {
  return value === true || value === 1 || value === "1" || value === "true";
 }
@@ -1304,6 +1308,100 @@ function compactQueueExecutionDiagnostics(value: unknown): Record<string, unknow
  };
 }

+function firstFiniteNumber(...values: unknown[]): number | null {
+  for (const value of values) {
+    const number = finiteNumber(value);
+    if (number !== null) return number;
+  }
+  return null;
+}
+
+function stringListCount(value: unknown): number | null {
+  return Array.isArray(value) ? stringList(value).length : null;
+}
+
+function compactCodeQueueActivity(
+  queue: Record<string, unknown>,
+  diagnostics: Record<string, unknown> | null,
+  options: { schedulerLocalActiveQueueIds?: string[]; runnableQueueCount?: number | null } = {},
+): Record<string, unknown> {
+  const rawDiagnostics = asRecord(queue.executionDiagnostics) ?? {};
+  const compactDiagnostics = diagnostics ?? {};
+  const rawLiveness = asRecord(rawDiagnostics.liveness) ?? {};
+  const compactLiveness = asRecord(compactDiagnostics.liveness) ?? {};
+  const counts = asRecord(queue.counts) ?? {};
+  const schedulerLocalActiveQueueIds = options.schedulerLocalActiveQueueIds ?? stringList(queue.activeQueueIds);
+  const databaseRunningTaskCount = firstFiniteNumber(counts.running) ?? 0;
+  const databaseJudgingTaskCount = firstFiniteNumber(counts.judging) ?? 0;
+  const databaseActiveTaskCount = firstFiniteNumber(
+    rawDiagnostics.databaseActiveTaskCount,
+    queue.databaseActiveTaskCount,
+    compactDiagnostics.databaseActiveTaskCount,
+    stringListCount(rawDiagnostics.databaseActiveTaskIds),
+    stringListCount(queue.databaseActiveTaskIds),
+    databaseRunningTaskCount + databaseJudgingTaskCount,
+  ) ?? 0;
+  const heartbeatFreshActiveTaskCount = firstFiniteNumber(
+    rawLiveness.heartbeatFreshTaskCount,
+    compactLiveness.heartbeatFreshTaskCount,
+    stringListCount(rawDiagnostics.heartbeatFreshTaskIds),
+    stringListCount(compactDiagnostics.heartbeatFreshTaskIds),
+  ) ?? 0;
+  const activeHeartbeatTaskCount = firstFiniteNumber(
+    rawDiagnostics.activeHeartbeatCount,
+    compactDiagnostics.activeHeartbeatCount,
+    stringListCount(rawDiagnostics.activeHeartbeatTaskIds),
+    heartbeatFreshActiveTaskCount,
+  ) ?? 0;
+  const heartbeatRiskTaskCount = firstFiniteNumber(
+    rawLiveness.heartbeatRiskTaskCount,
+    compactLiveness.heartbeatRiskTaskCount,
+    stringListCount(rawDiagnostics.heartbeatRiskTaskIds),
+    stringListCount(compactDiagnostics.heartbeatRiskTaskIds),
+  ) ?? 0;
+  const schedulerLocalActiveRunSlotCount = firstFiniteNumber(
+    rawDiagnostics.schedulerActiveRunSlotCount,
+    queue.schedulerActiveRunSlotCount,
+    queue.activeRunSlotCount,
+    compactDiagnostics.schedulerActiveRunSlotCount,
+  );
+  const runnableQueueCount = firstFiniteNumber(options.runnableQueueCount, queue.runnableQueueCount);
+  const effectiveActiveTaskCount = Math.max(databaseActiveTaskCount, databaseRunningTaskCount, heartbeatFreshActiveTaskCount);
+  const splitBrainLive = splitBrainLiveFromDiagnostics(rawDiagnostics) || splitBrainLiveFromDiagnostics(compactDiagnostics);
+  const effectiveActiveSource = heartbeatFreshActiveTaskCount > 0 && heartbeatFreshActiveTaskCount >= databaseActiveTaskCount
+    ? "heartbeat-fresh"
+    : databaseActiveTaskCount > 0
+      ? "database-active"
+      : schedulerLocalActiveQueueIds.length > 0 || (schedulerLocalActiveRunSlotCount ?? 0) > 0
+        ? "scheduler-local"
+        : "none";
+  const activeQueueIdsNote = schedulerLocalActiveQueueIds.length === 0 && effectiveActiveTaskCount > 0
+    ? "activeQueueIds are scheduler-local only; zero local queue ids does not mean zero active runners when database or heartbeat counts are nonzero."
+    : "activeQueueIds are scheduler-local active-run slots; use effectiveActiveTaskCount for commander concurrency decisions.";
+  return {
+    effectiveActiveTaskCount,
+    effectiveActiveSource,
+    databaseRunningTaskCount,
+    databaseActiveTaskCount,
+    heartbeatFreshActiveTaskCount,
+    activeHeartbeatTaskCount,
+    heartbeatRiskTaskCount,
+    schedulerLocalActiveQueueCount: schedulerLocalActiveQueueIds.length,
+    schedulerLocalActiveRunSlotCount,
+    runnableQueueCount,
+    splitBrainLive,
+    activeQueueIdsScope: "scheduler-local-active-run-slots",
+    activeQueueIdsNote,
+    interpretation: splitBrainLive
+      ? "split-brain live: database-active tasks have fresh scheduler heartbeat; continue supervision."
+      : heartbeatRiskTaskCount > 0
+        ? "heartbeat risk is present; investigate before retry or recovery."
+        : effectiveActiveTaskCount > 0
+          ? "active work is present; compare database, heartbeat, and scheduler-local counts before changing concurrency."
+          : "no active work observed by database, heartbeat, or scheduler-local signals.",
+  };
+}
+
 function supervisorExecutionDiagnostics(value: unknown): Record<string, unknown> | null {
  const diagnostics = compactExecutionDiagnostics(value);
  if (diagnostics === null) return null;
@@ -2355,6 +2453,7 @@ function codexTasksOverviewResult(
  const queuedSection = buildSupervisorTaskSection(queuedTasks, summaries, sectionLimit, sectionNextCommand(queuedTasks, sectionLimit, options, nextCommand), fullCommand);
  const pagination = taskPage.pagination;
  const diagnostics = supervisorExecutionDiagnostics(asRecord(taskPage.queue)?.executionDiagnostics);
+  const activity = compactCodeQueueActivity(asRecord(taskPage.queue) ?? {}, diagnostics);
  const visibleSupervisorItems = [...runningSection.items, ...unreadSection.items, ...recentSection.items, ...queuedSection.items];
  const classifierCounts = visibleSupervisorItems.reduce((counts, item) => {
    const key = item.kind;
@@ -2405,8 +2504,13 @@ function codexTasksOverviewResult(
        completedUnread: unreadSection.count,
        recentCompleted: recentSection.count,
        queued: queuedSection.count,
+        effectiveActive: asNumber(activity.effectiveActiveTaskCount, 0),
+        databaseRunning: asNumber(activity.databaseRunningTaskCount, 0),
+        heartbeatFreshActive: asNumber(activity.heartbeatFreshActiveTaskCount, 0),
+        schedulerLocalActiveQueues: asNumber(activity.schedulerLocalActiveQueueCount, 0),
      },
      classifierCounts,
+      activity,
      executionDiagnostics: diagnostics,
      degraded,
      commands: {
@@ -2698,6 +2802,7 @@ function compactQueuesResponse(body: Record<string, unknown>, options: CodexQueu
  const selected = options.full ? queues : Array.from(new Map([...activeQueues, ...unreadQueues, ...runnableQueues, ...nonemptyQueues].map((row) => [String(row.id), row])).values());
  const visible = selected.slice(options.offset, options.offset + options.limit);
  const diagnostics = compactQueueExecutionDiagnostics(queue.executionDiagnostics);
+  const activity = compactCodeQueueActivity(queue, diagnostics, { schedulerLocalActiveQueueIds: activeIds, runnableQueueCount: runnableQueues.length });
  const activeTaskIds = boundedUniqueStringList(queue.activeTaskIds, Math.min(options.limit, maxTasksLimit));
  const queuedTaskIds = boundedUniqueStringList(queue.queuedTaskIds, Math.min(options.limit, maxTasksLimit));
  const nextOffset = options.offset + visible.length;
@@ -2725,11 +2830,20 @@ function compactQueuesResponse(body: Record<string, unknown>, options: CodexQueu
        totalTasks: queue.total ?? null,
        queueCount: queue.queueCount ?? queues.length,
        activeQueueCount: activeIds.length,
+        activeQueueCountScope: "scheduler-local-active-run-slots",
+        schedulerLocalActiveQueueCount: activeIds.length,
+        effectiveActiveTaskCount: activity.effectiveActiveTaskCount,
+        databaseRunningTaskCount: activity.databaseRunningTaskCount,
+        databaseActiveTaskCount: activity.databaseActiveTaskCount,
+        heartbeatFreshActiveTaskCount: activity.heartbeatFreshActiveTaskCount,
        nonemptyQueueCount: nonemptyQueues.length,
        unreadQueueCount: unreadQueues.length,
        runnableQueueCount: runnableQueues.length,
      },
+      activity,
      activeQueueIds: queue.activeQueueIds ?? [],
+      activeQueueIdsScope: "scheduler-local-active-run-slots",
+      activeQueueIdsNote: activity.activeQueueIdsNote,
      activeTaskIds: activeTaskIds.items,
      activeTaskIdsCount: activeTaskIds.count,
      activeTaskIdsTruncated: activeTaskIds.truncated,
@@ -56,14 +56,14 @@ export function rootHelp(): unknown {
      { command: "codex skills-sync --dry-run [--full]", description: "Inspect the controlled runner skills hostPath lifecycle contract without copying files, restarting services, reading secrets, or mutating live runner paths." },
      { command: "codex pr-preflight [--remote] [--push-dry-run --push-dry-run-ref refs/heads/probe/<name>] [--pr-create-dry-run --pr-create-dry-run-head <head>] [--issue N]", description: "Read-only PR admission check against the D601 scheduler/runner token, GitHub egress, repo visibility, skills lifecycle health, optional push dry-run, and PR body/create dry-run guard." },
      { command: "codex task <taskId> [--detail] [--trace --tail|--from-start|--after-seq N|--before-seq N --limit N] [--full]", description: "Fetch the bounded review view by default; --detail is still capped, while --full/trace/output explicitly expand evidence." },
-      { command: "codex tasks [--view supervisor|full] [--queue id] [--status status[,status]] [--unread|--unread-only] [--limit N] [--before-id id]", description: "Show the low-noise supervisor view by default: compact task rows, tiny local sections, diagnostics, and drill-down commands; use --view full for detailed rows." },
+      { command: "codex tasks [--view supervisor|full] [--queue id] [--status status[,status]] [--unread|--unread-only] [--limit N] [--before-id id]", description: "Show the low-noise supervisor view by default: compact task rows, tiny local sections, activity counts, diagnostics, and drill-down commands; use --view full for detailed rows." },
      { command: "codex output <taskId> [--tail|--from-start|--after-seq N|--before-seq N --limit N] [--full-text]", description: "Fetch paged raw Code Queue output records; default caps large limits/text previews, --full-text explicitly expands one seq window." },
      { command: "codex read <taskId>", description: "Mark one reviewed terminal task read; never run automatically as part of listing." },
      { command: "codex dev-ready", description: "Fetch execution-container readiness, including sanitized skill injection status from /api/dev-ready." },
      { command: "codex judge <taskId> --attempt N [--dry-run] [--include-prompt]", description: "Replay one stored Code Queue attempt through the same judge context builder and MiniMax judge call path used by the live queue worker." },
      { command: "codex steer <taskId> [prompt|--prompt-file path|--prompt-stdin] [--dry-run] [--no-retry|--retry-attempts N]", description: "Push a corrective prompt into a running Code Queue task; retryable tunnel aborts get bounded retry diagnostics, and real success does not echo prompt text." },
      { command: "codex interrupt|cancel <taskId>", description: "Request interrupt for a running Code Queue task, or cancel a queued/retry_wait task, through the same private proxy." },
-      { command: "codex (queues [--full|--all] | queue create <queueId> | queue merge <sourceQueueId> --into <targetQueueId> | move <taskId> --queue <queueId>)", description: "List low-noise queue summaries by default; full queue rows require --full/--all." },
+      { command: "codex (queues [--full|--all] | queue create <queueId> | queue merge <sourceQueueId> --into <targetQueueId> | move <taskId> --queue <queueId>)", description: "List low-noise queue summaries by default, including effective activity counts that distinguish scheduler-local queues, DB running tasks, and heartbeat-fresh runners; full queue rows require --full/--all." },
      { command: "job list [--limit N] [--include-command]", description: "List async jobs from .state/jobs with a bounded default page." },
      { command: "job status <jobId|latest> [--tail-bytes N]", description: "Show job state with bounded stdout/stderr tails." },
      { command: "debug health", description: "Probe internal core, nodes, system/Docker status, frontend, provider ingress, and public boundary." },
@@ -284,7 +284,13 @@ function codexHelp(): unknown {
      defaultPolicy: "low-noise JSON by default; write commands confirm persistence, list/detail/output commands return bounded summaries with drill-down commands",
      expand: ["codex task <taskId> --full", "codex task <taskId> --trace --limit N", "codex output <taskId> --after-seq N --limit N --full-text", "codex tasks --view full --limit N", "codex skills-sync --dry-run --full"],
    },
-    description: "Operate Code Queue through the stable backend-core private proxy path. Real submit/steer success is a low-noise write confirmation and does not echo prompt text.",
+    activityFields: {
+      path: "data.queues.activity and data.supervisor.activity",
+      effectiveActiveTaskCount: "Commander-facing active count derived from database active/running tasks and heartbeat-fresh runners.",
+      schedulerLocalActiveQueueCount: "Only queues currently visible in this scheduler-local active-run slot view; zero does not override DB or heartbeat activity.",
+      heartbeatFreshActiveTaskCount: "Heartbeat-effective active runner count used to avoid split-brain zero-active mistakes.",
+    },
+    description: "Operate Code Queue through the stable backend-core private proxy path with bounded activity summaries for queue and supervisor views. Real submit/steer success is a low-noise write confirmation and does not echo prompt text.",
  };
 }