Merge pull request #291 from Blazity/fix/aiw-277-dispatch-at-capacity

feat(worker): surface at-capacity dispatch refusals in logs, Jira, and dashboard (AIW-277)

Author: outof-place

apps/dashboard/app/overview-data.tsx

@@ -10,6 +10,7 @@ import type {
   KpisResponse,
   EvalHealthResponse,
   LiveRunsResponse,
+  DispatchCapacityResponse,
   RunsResponse,
   WorkflowsResponse,
 } from "@shared/contracts";
@@ -18,6 +19,7 @@ import {
   evalHealthFallback,
   recentRunsFallback,
   liveRunsFallback,
+  dispatchCapacityFallback,
   workflowsFallback,
 } from "@/lib/api/fallbacks";
 import { deriveKpisFromRuns } from "@/lib/api/derive-kpis";
@@ -43,21 +45,27 @@ export async function OverviewData({ window }: { window: TimeWindow }) {
 
   // Window scopes the historical aggregates (KPIs, recent runs, workflows).
   // Eval-health (Arthur) and live runs (registry) are not windowed here.
-  const [kpis, evalHealth, recentRuns, liveRuns, workflows] = await Promise.all([
-    getJSON<KpisResponse>(withQuery("/api/v1/overview/kpis", { window })).catch(
-      (e) => authAwareFallback(e, () => kpisFallback(now)),
-    ),
-    getJSON<EvalHealthResponse>("/api/v1/overview/eval-health").catch(
-      (e) => authAwareFallback(e, () => evalHealthFallback()),
-    ),
-    getJSON<RunsResponse>(withQuery("/api/v1/runs", { window })).catch((e) =>
-      authAwareFallback(e, () => recentRunsFallback(now)),
-    ),
-    getJSON<LiveRunsResponse>("/api/v1/runs/live").catch((e) => authAwareFallback(e, () => liveRunsFallback(now))),
-    getJSON<WorkflowsResponse>(withQuery("/api/v1/workflows", { window })).catch(
-      (e) => authAwareFallback(e, () => workflowsFallback(now)),
-    ),
-  ]);
+  const [kpis, evalHealth, recentRuns, liveRuns, capacity, workflows] =
+    await Promise.all([
+      getJSON<KpisResponse>(withQuery("/api/v1/overview/kpis", { window })).catch(
+        (e) => authAwareFallback(e, () => kpisFallback(now)),
+      ),
+      getJSON<EvalHealthResponse>("/api/v1/overview/eval-health").catch(
+        (e) => authAwareFallback(e, () => evalHealthFallback()),
+      ),
+      getJSON<RunsResponse>(withQuery("/api/v1/runs", { window })).catch((e) =>
+        authAwareFallback(e, () => recentRunsFallback(now)),
+      ),
+      getJSON<LiveRunsResponse>("/api/v1/runs/live").catch((e) =>
+        authAwareFallback(e, () => liveRunsFallback(now)),
+      ),
+      getJSON<DispatchCapacityResponse>("/api/v1/dispatch/capacity").catch((e) =>
+        authAwareFallback(e, () => dispatchCapacityFallback(now)),
+      ),
+      getJSON<WorkflowsResponse>(withQuery("/api/v1/workflows", { window })).catch(
+        (e) => authAwareFallback(e, () => workflowsFallback(now)),
+      ),
+    ]);
 
   // The worker's KPI endpoint returns null when its run-store fetch is rejected
   // (page-size cap). Derive the tiles from the runs list we already have so the
@@ -80,6 +88,7 @@ export async function OverviewData({ window }: { window: TimeWindow }) {
     kpis: mergedKpis,
     evalHealth,
     liveRuns: reconcileOverviewLiveRuns(recentRuns, liveRuns),
+    capacity,
     recentRuns,
     workflows,
   };

apps/dashboard/components/cockpit/screens/overview.test.tsx

@@ -5,7 +5,8 @@ import { act, create, type ReactTestInstance, type ReactTestRenderer } from "rea
 import { AppRouterContext } from "next/dist/shared/lib/app-router-context.shared-runtime";
 
 import type { Run } from "@/lib/types";
-import { AwaitingInputPanel } from "./overview";
+import type { DispatchCapacityResponse } from "@shared/contracts";
+import { AwaitingInputPanel, NowRunningPanel } from "./overview";
 
 (globalThis as typeof globalThis & { React: typeof React }).React = React;
 (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
@@ -71,6 +72,89 @@ function renderPanel(t: TestContext, rows: Run[]): { root: ReactTestInstance; op
   return { root: renderer.root, opened };
 }
 
+function capacity(
+  over: Partial<DispatchCapacityResponse> = {},
+): DispatchCapacityResponse {
+  return {
+    generatedAt: "2026-08-16T12:00:00.000Z",
+    occupiedSlots: 0,
+    maxSlots: 3,
+    queued: [],
+    ...over,
+  };
+}
+
+function renderNowRunning(
+  t: TestContext,
+  rows: Run[],
+  cap: DispatchCapacityResponse,
+): ReactTestInstance {
+  let renderer!: ReactTestRenderer;
+  act(() => {
+    renderer = create(
+      <AppRouterContext.Provider value={stubRouter() as never}>
+        <NowRunningPanel rows={rows} capacity={cap} onOpenRun={() => {}} />
+      </AppRouterContext.Provider>,
+    );
+  });
+  t.after(() => act(() => renderer.unmount()));
+  return renderer.root;
+}
+
+test("a full pool with zero executing runs shows it is full and lists the waiting tickets", (t) => {
+  // The bug AIW-277 fixes: parked claims fill every slot, nothing is "running",
+  // and the panel used to read as idle. It must now show the occupied count and
+  // the at-capacity queue.
+  const root = renderNowRunning(t, [], capacity({
+    occupiedSlots: 3,
+    maxSlots: 3,
+    queued: [
+      { ticketKey: "AWT-9", queuedAt: new Date(Date.now() - 5 * 60_000 - 5_000).toISOString() },
+    ],
+  }));
+
+  const text = nodeText(root);
+  assert.match(text, /3\/3 slots/);
+  assert.match(text, /waiting for capacity/);
+  assert.match(text, /AWT-9/);
+  assert.match(text, /waiting 5m/);
+});
+
+test("an idle pool shows free slots and no waiting queue", (t) => {
+  const root = renderNowRunning(t, [], capacity({ occupiedSlots: 0, maxSlots: 3 }));
+
+  const text = nodeText(root);
+  assert.match(text, /0\/3 slots/);
+  assert.doesNotMatch(text, /waiting for capacity/);
+});
+
+test("the worker-unavailable fallback (maxSlots 0) reads as unknown, never full", (t) => {
+  const root = renderNowRunning(t, [], capacity({ occupiedSlots: 0, maxSlots: 0 }));
+
+  const text = nodeText(root);
+  assert.match(text, /slots —/);
+  // The 0/0 fallback must not render as "0/0" nor claim the pool is full.
+  assert.doesNotMatch(text, /0\/0/);
+});
+
+test("a running ticket is excluded from the waiting-for-capacity list", (t) => {
+  // A ticket stays in the AI column while it runs, so its stale queue row could
+  // leak into the waiting list — the panel must drop any ticket already live.
+  const runningRow: Run = { ...BASE_RUN, id: "run_live", status: "running", ticket: "AWT-9" };
+  const root = renderNowRunning(
+    t,
+    [runningRow],
+    capacity({
+      occupiedSlots: 3,
+      maxSlots: 3,
+      queued: [{ ticketKey: "AWT-9", queuedAt: new Date(Date.now() - 3 * 60_000).toISOString() }],
+    }),
+  );
+
+  const text = nodeText(root);
+  assert.doesNotMatch(text, /waiting for capacity/);
+});
+
 test("a clarification row keeps its Answer CTA to the run trace, unchanged", (t) => {
   const row: Run = {
     ...BASE_RUN,

apps/dashboard/components/cockpit/screens/overview.tsx

@@ -14,6 +14,7 @@ import {
 } from "@/components/ui";
 import { Spark, Donut } from "@/components/charts";
 import { runModelLabel } from "@/lib/run-model";
+import { formatWaited } from "@/lib/waited";
 import { spanColor } from "@/lib/theme";
 import { useCockpit } from "@/components/cockpit/context";
 import { WindowSelector } from "@/components/cockpit/controls";
@@ -29,15 +30,17 @@ import type {
   KpisResponse,
   EvalHealthResponse,
   LiveRunsResponse,
+  DispatchCapacityResponse,
   RunsResponse,
   WorkflowsResponse,
 } from "@shared/contracts";
 
-/** Bundle of the five server-fetched responses passed into the presentational Overview. */
+/** Bundle of the server-fetched responses passed into the presentational Overview. */
 export interface OverviewScreenData {
   kpis: KpisResponse;
   evalHealth: EvalHealthResponse;
   liveRuns: LiveRunsResponse;
+  capacity: DispatchCapacityResponse;
   recentRuns: RunsResponse;
   workflows: WorkflowsResponse;
 }
@@ -120,27 +123,54 @@ function EvalHealthKPI({ data }: { data: EvalHealthResponse | undefined }) {
   );
 }
 
-/* Live "Now running" panel — currently executing runs only. */
-function NowRunningPanel({
+/* Live "Now running" panel. Shows executing runs, plus the occupied-slot count
+ * counted the way dispatch refuses (parked claims included, from
+ * listCapacityConsumers) and the at-capacity waiting queue — so a full pool with
+ * zero executing runs no longer looks idle. */
+export function NowRunningPanel({
   rows,
+  capacity,
   onOpenRun,
 }: {
   rows: Run[];
+  capacity: DispatchCapacityResponse;
   onOpenRun: (run: Run) => void;
 }) {
   const running = rows.filter((r) => r.status === "running");
+  const { occupiedSlots, maxSlots } = capacity;
+  // maxSlots === 0 is the worker-unavailable fallback: capacity is UNKNOWN, not
+  // full, so never render the amber "full" style for it.
+  const capacityUnknown = maxSlots === 0;
+  const poolFull = !capacityUnknown && occupiedSlots >= maxSlots;
+  // Defense in depth against a stale row: a ticket that is running (or otherwise
+  // holds a live slot) is not waiting, so it must never show in both panels.
+  const liveTickets = new Set(rows.map((r) => r.ticket));
+  const queued = capacity.queued.filter((q) => !liveTickets.has(q.ticketKey));
 
   return (
     <CkCard
       eyebrow="Vercel workflow · live"
       title="Now running"
       action={
-        <span className="inline-flex items-center gap-1.5 font-mono text-[10px] text-mariner tracking-[0.04em] uppercase">
-          <span className="relative w-1.5 h-1.5">
-            <span className="absolute inset-0 rounded-full bg-mariner" />
-            <span className="absolute -inset-[3px] rounded-full border border-mariner animate-ck-pulse" />
+        <span className="inline-flex items-center gap-2.5 font-mono text-[10px] tracking-[0.04em] uppercase">
+          <span className="inline-flex items-center gap-1.5 text-mariner">
+            <span className="relative w-1.5 h-1.5">
+              <span className="absolute inset-0 rounded-full bg-mariner" />
+              <span className="absolute -inset-[3px] rounded-full border border-mariner animate-ck-pulse" />
+            </span>
+            {running.length} executing
+          </span>
+          {/* Slots in use, from the same count dispatch refuses against, so a
+              pool full of parked claims is not mistaken for an idle one. */}
+          <span
+            className={`inline-flex items-center rounded-[3px] border px-1.5 py-[2px] ${
+              poolFull
+                ? "border-amber-300 bg-amber-50 text-amber-800"
+                : "border-neutral-200 bg-off-white text-neutral-600"
+            }`}
+          >
+            {capacityUnknown ? "slots —" : `${occupiedSlots}/${maxSlots} slots`}
           </span>
-          {running.length} executing
         </span>
       }
       pad={0}
@@ -209,6 +239,26 @@ function NowRunningPanel({
           })}
         </div>
       )}
+      {queued.length > 0 && (
+        <div className="border-t border-amber-200 bg-amber-50 px-5 py-3">
+          <div className="font-mono text-[10px] uppercase tracking-[0.06em] text-amber-800 mb-2">
+            {queued.length} waiting for capacity
+          </div>
+          <div className="flex flex-col gap-1.5">
+            {queued.map((q) => (
+              <div
+                key={q.ticketKey}
+                className="flex items-center justify-between gap-2 font-body text-xs text-amber-900"
+              >
+                <span className="font-medium">{q.ticketKey}</span>
+                <span className="font-mono text-[11px] text-amber-700 whitespace-nowrap">
+                  waiting {formatWaited(q.queuedAt)}
+                </span>
+              </div>
+            ))}
+          </div>
+        </div>
+      )}
     </CkCard>
   );
 }
@@ -472,7 +522,7 @@ export function OverviewScreen({
 
       {/* Live row */}
       <div className="grid grid-cols-2 gap-3">
-        <NowRunningPanel rows={liveRows} onOpenRun={openRun} />
+        <NowRunningPanel rows={liveRows} capacity={data.capacity} onOpenRun={openRun} />
         <AwaitingInputPanel rows={liveRows} onOpenRun={openRun} />
       </div>
 

apps/dashboard/lib/api/fallbacks.ts

@@ -6,6 +6,7 @@ import type {
   RunsResponse,
   RunDetailResponse,
   LiveRunsResponse,
+  DispatchCapacityResponse,
   WorkflowsResponse,
   TicketRunsResponse,
   WorkflowRunReplayResponse,
@@ -53,6 +54,10 @@ export function liveRunsFallback(now: string): LiveRunsResponse {
   return { generatedAt: now, rows: [] };
 }
 
+export function dispatchCapacityFallback(now: string): DispatchCapacityResponse {
+  return { generatedAt: now, occupiedSlots: 0, maxSlots: 0, queued: [] };
+}
+
 export function workflowsFallback(now: string): WorkflowsResponse {
   return { generatedAt: now, rows: [], total: 0 };
 }

apps/dashboard/lib/waited.test.ts

@@ -0,0 +1,21 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { formatWaited } from "./waited";
+
+const NOW = Date.parse("2026-08-16T12:00:00.000Z");
+
+test("under a minute reads as such, never negative", () => {
+  assert.equal(formatWaited("2026-08-16T11:59:30.000Z", NOW), "under a minute");
+  // A queued_at fractionally in the future (clock skew) must not go negative.
+  assert.equal(formatWaited("2026-08-16T12:00:05.000Z", NOW), "under a minute");
+});
+
+test("minutes below an hour", () => {
+  assert.equal(formatWaited("2026-08-16T11:55:00.000Z", NOW), "5m");
+  assert.equal(formatWaited("2026-08-16T11:01:00.000Z", NOW), "59m");
+});
+
+test("hours, with and without trailing minutes", () => {
+  assert.equal(formatWaited("2026-08-16T10:00:00.000Z", NOW), "2h");
+  assert.equal(formatWaited("2026-08-16T09:38:00.000Z", NOW), "2h 22m");
+});

apps/dashboard/lib/waited.ts

@@ -0,0 +1,13 @@
+/**
+ * Compact "how long it has been waiting" label from an ISO timestamp. Computed
+ * on the client from a first-seen instant, so `now` is injectable for tests.
+ */
+export function formatWaited(fromIso: string, now: number = Date.now()): string {
+  const ms = now - new Date(fromIso).getTime();
+  if (!Number.isFinite(ms) || ms < 60_000) return "under a minute";
+  const totalMinutes = Math.floor(ms / 60_000);
+  if (totalMinutes < 60) return `${totalMinutes}m`;
+  const hours = Math.floor(totalMinutes / 60);
+  const minutes = totalMinutes % 60;
+  return minutes === 0 ? `${hours}h` : `${hours}h ${minutes}m`;
+}

apps/shared/contracts/api.ts

@@ -107,6 +107,26 @@ export interface LiveRunsResponse {
   rows: Run[];
 }
 
+/** One ticket waiting for a run slot because the pool is full (AIW-277). */
+export interface QueuedTicketEntry {
+  ticketKey: string;
+  /** ISO-8601 first-seen timestamp — how long it has been waiting. */
+  queuedAt: string;
+}
+
+/**
+ * Dispatch-capacity snapshot for the Overview. `occupiedSlots` is counted the
+ * way the refusal counts it (listCapacityConsumers — parked claims included),
+ * so a full pool with zero executing runs no longer looks idle. `queued` lists
+ * the tickets that were refused for capacity and are waiting.
+ */
+export interface DispatchCapacityResponse {
+  generatedAt: string;
+  occupiedSlots: number;
+  maxSlots: number;
+  queued: QueuedTicketEntry[];
+}
+
 export interface RunsResponse {
   generatedAt: string;
   available: boolean;

apps/worker/drizzle/0050_dispatch_capacity_queue.sql

@@ -0,0 +1,6 @@
+CREATE TABLE "dispatch_capacity_queue" (
+	"ticket_key" text PRIMARY KEY NOT NULL,
+	"queued_at" timestamp with time zone DEFAULT now() NOT NULL,
+	"attempted_at" timestamp with time zone,
+	"confirmed_at" timestamp with time zone
+);