Merge branch 'fix/report-health-review' into feat/dashboard-agent-flows

Author: kathiekiwi

.changeset/report-json-and-period-units.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/core": patch
+"trigger.dev": patch
+---
+
+Reports can now be fetched as structured data, not just text: ask for the `json` format and you get the numbers and what they mean, typed. Report periods are also stricter — the shortest window is one minute (`30m`, `1h`, `7d`), because reports summarise data by the minute and anything shorter can't be answered honestly.

apps/webapp/app/presenters/v3/reports/ReportPresenter.server.ts

@@ -8,7 +8,7 @@
  */
 
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { REPORT_REGISTRY } from "./report-registry";
+import { REPORT_REGISTRY, type ReportLoader } from "./report-registry";
 import { type ReportViewModel } from "./report-view-model";
 
 const DEFAULT_PERIOD = "1h";
@@ -22,6 +22,8 @@ const DEFAULT_PERIOD = "1h";
 const inFlight = new Map<string, Promise<ReportViewModel | undefined>>();
 
 export class ReportPresenter {
+  constructor(private readonly registry: Record<string, ReportLoader<unknown>> = REPORT_REGISTRY) {}
+
   async call({
     environment,
     key,
@@ -31,8 +33,8 @@ export class ReportPresenter {
     key: string;
     period?: string;
   }): Promise<ReportViewModel | undefined> {
-    const loader = REPORT_REGISTRY[key];
-    if (!loader) return undefined;
+    if (!Object.hasOwn(this.registry, key)) return undefined;
+    const loader = this.registry[key];
 
     const flightKey = `${key} ${environment.id} ${period}`;
     const existing = inFlight.get(flightKey);

apps/webapp/app/presenters/v3/reports/health/flow.ts

@@ -17,15 +17,24 @@ import {
   type Severity,
 } from "../report-view-model";
 import {
+  bucketCoverage,
   HEALTH_THRESHOLDS,
   isPendingIncreasing,
+  isPendingUnknown,
   mean,
   metricById,
   type HealthInput,
 } from "./health-core";
 
 export const FLOW_METRIC_IDS = ["start_latency_p95", "pending", "throughput"];
 
+/**
+ * Flow reason for "we could not measure the backlog". Not a severity and not a cause — the
+ * verdict is simply unassessable, so nothing actionable may hang off it. Kept distinct from
+ * "unknown" (the stale-telemetry guard) so the two failure modes stay legible.
+ */
+export const FLOW_UNMEASURED = "flow_unmeasured";
+
 /** One row of the declarative cause table — everything a cause defines about itself. */
 type CauseSpec = {
   reason: string;
@@ -44,6 +53,13 @@ export function interpretFlow(metrics: Metric[], input: HealthInput): Finding {
   const flowMetrics = FLOW_METRIC_IDS.map((id) => metricById(metrics, id));
   const severity = maxSeverity(...flowMetrics.map((m) => m.severity));
 
+  // The backlog couldn't be measured: `pending.now` is a placeholder, so neither "healthy" nor a
+  // cause may be claimed off it. Severity still reflects the metrics we DID measure (start
+  // latency), but the finding carries no cause, no attribution and no recommendation.
+  if (isPendingUnknown(input)) {
+    return { type: "flow", severity, reason: FLOW_UNMEASURED, metricIds: FLOW_METRIC_IDS };
+  }
+
   if (isOk(severity)) {
     return { type: "flow", severity, reason: "healthy", metricIds: FLOW_METRIC_IDS };
   }
@@ -52,12 +68,18 @@ export function interpretFlow(metrics: Metric[], input: HealthInput): Finding {
   const pendingIncreasing = isPendingIncreasing(input.pending.series);
   const latencyElevated = !isOk(metricById(metrics, "start_latency_p95").severity);
   // Concurrency causes need real running-capacity evidence — without it runningShare is a
-  // meaningless 0 and would falsely select dequeue_stall on the snapshot path (#1).
-  const hasConcurrencyEvidence = ev.envLimit > 0 && ev.runningSeries.length > 0;
+  // meaningless 0 and would falsely select dequeue_stall on the snapshot path (#1). They also
+  // need enough of the window to have ARRIVED: the series isn't gap-filled, so a couple of fresh
+  // buckets would otherwise read as "pinned the whole window".
+  const coverage = bucketCoverage(input);
+  const hasConcurrencyEvidence =
+    ev.envLimit > 0 && ev.runningSeries.length > 0 && coverage.sufficient;
   const runningShare = hasConcurrencyEvidence ? mean(ev.runningSeries) / ev.envLimit : 1;
+  // Pinned share is measured against EXPECTED buckets, not received rows — "2 of 60 expected",
+  // never "2 of 2 received".
   const pinnedShare = hasConcurrencyEvidence
     ? ev.runningSeries.filter((r) => r >= t.pinnedLevel * ev.envLimit).length /
-      ev.runningSeries.length
+      coverage.expectedBuckets
     : 0;
   const pinned = pinnedShare >= t.pinnedShare;
   const hasTriggerBaseline = input.throughput.normalTriggeredPerMin > 0;
@@ -67,8 +89,9 @@ export function interpretFlow(metrics: Metric[], input: HealthInput): Finding {
   // No baseline: a multiplier can't be computed, so an absolute rate selects "new volume".
   const triggerSurge = !hasTriggerBaseline && input.throughput.triggeredPerMin >= t.surgePerMin;
 
-  const donePerMin = input.throughput.donePerMin;
-  const net = donePerMin - input.throughput.triggeredPerMin;
+  // Work leaving the queue = FINISHED (all terminal) runs, not completions only.
+  const finishedPerMin = input.throughput.finishedPerMin;
+  const net = finishedPerMin - input.throughput.triggeredPerMin;
   // Exclusions must be PROVEN, not assumed. "not your code" needs healthy execution; "limits
   // aren't the bottleneck" needs no env-pin AND no queue throttling; the workers/spike ones
   // state a measured fact (rate) rather than a global "everything's fine" claim.
@@ -112,10 +135,10 @@ export function interpretFlow(metrics: Metric[], input: HealthInput): Finding {
       drivingMetricId: "concurrency",
       annotationCode: "pinned_minutes",
       exclusions: [],
-      // States a measured fact (runs ARE completing at {rate}/min) — evidence the workers aren't
+      // States a measured fact (runs ARE finishing at {rate}/min) — evidence the workers aren't
       // dead. An observation, not an exclusion: it doesn't claim it's the limit, nor "keeps pace".
       observations:
-        donePerMin > 0 ? [{ code: "not_workers_platform", evidence: { donePerMin } }] : [],
+        finishedPerMin > 0 ? [{ code: "not_workers_platform", evidence: { finishedPerMin } }] : [],
       recommendation: { code: "raise_env_limit", link: "concurrency" },
       usesAttribution: true,
     };
@@ -177,16 +200,22 @@ function assembleFlowCause(
 
   // Anomaly window from the driving series. env_limit_saturation breaches ABOVE
   // (concurrency pinned at the limit); dequeue_stall breaches BELOW (capacity idle).
-  // NOTE: runningSeries is at native env_metrics resolution (not resampled), so the "(last N
-  // min)" figure assumes those buckets are uniform and cover the resolved window. env_metrics
-  // are emitted on a fixed cadence, so that holds; a gappy/partial window could skew the minutes.
+  // runningSeries is at native env_metrics resolution (not resampled) and is NOT gap-filled, so
+  // the duration is counted per REAL bucket cadence with gaps breaking the contiguous run —
+  // otherwise two fresh buckets would read as "the last 60 min". When the source can't report its
+  // cadence we fall back to the (documented) even-spread assumption.
   let aw: Finding["anomalyWindow"];
   if (spec.reason === "env_limit_saturation" || spec.reason === "dequeue_stall") {
     const below = spec.reason === "dequeue_stall";
     const threshold = below
       ? t.flowCause.stallRunningShare * input.flowEvidence.envLimit
       : t.flowCause.pinnedLevel * input.flowEvidence.envLimit;
-    aw = anomalyWindow(input.flowEvidence.runningSeries, threshold, input.windowMinutes, { below });
+    const coverage = bucketCoverage(input);
+    aw = anomalyWindow(input.flowEvidence.runningSeries, threshold, input.windowMinutes, {
+      below,
+      bucketMinutes: coverage.known ? coverage.bucketMinutes : undefined,
+      timestampsMs: input.flowEvidence.runningBucketsMs,
+    });
   }
 
   // Annotation on the driving metric (a fact, not an invented number).
@@ -295,6 +324,7 @@ const CAUSE_READS: Record<string, string> = {
 
 export function buildFlowRead(flow: Finding, executionOk: boolean, livenessFresh: boolean): string {
   if (flow.reason === "unknown") return "data_stale"; // stale-guarded — no causal read
+  if (flow.reason === FLOW_UNMEASURED) return "flow_unmeasured"; // no depth signal — no read
   if (isOk(flow.severity)) return "starting_normally";
   if (CAUSE_READS[flow.reason]) return CAUSE_READS[flow.reason];
   // fallback symptoms (v1 logic)

apps/webapp/app/presenters/v3/reports/health/health-core.ts

@@ -19,10 +19,30 @@ export type HealthInput = {
    * now = live env-level depth; normal = 7d baseline (omitted on the snapshot path, which has
    * no real 7d pending baseline — so we never mislabel a live-window average as "7d normal");
    * series measured (v2) or estimated (v1).
+   *
+   * `availability: "unknown"` = the depth could NOT be measured at all (Redis down with no
+   * measured fallback, or the measured source failed for an unrecognized reason). `now` is then
+   * a placeholder, NEVER a confident 0 — flow is reported unassessable instead of healthy.
    */
-  pending: { now: number; normal?: number; series: number[]; estimated: boolean };
+  pending: {
+    now: number;
+    normal?: number;
+    series: number[];
+    estimated: boolean;
+    availability?: "measured" | "unknown";
+  };
   startLatency: { p95Ms: number; normalP95Ms: number; series: number[] };
-  throughput: { donePerMin: number; triggeredPerMin: number; normalTriggeredPerMin: number };
+  /**
+   * finishedPerMin = ALL terminal runs per minute — the rate work actually LEAVES the queue, so
+   * it's what net/drain math uses. completedPerMin (successes only) is an execution-side metric
+   * and would understate the drain rate whenever runs fail/expire/cancel.
+   */
+  throughput: {
+    finishedPerMin: number;
+    completedPerMin: number;
+    triggeredPerMin: number;
+    normalTriggeredPerMin: number;
+  };
   failures: { rate: number; normalRate: number; series: number[] };
   duration: { p95Ms: number; normalP95Ms: number };
   /** Age of the freshest telemetry (ms). null = no signal to assess -> freshness unknown. */
@@ -33,6 +53,18 @@ export type HealthInput = {
    */
   flowEvidence: {
     runningSeries: number[];
+    /**
+     * Epoch ms of each `runningSeries` bucket, aligned by index. Empty/absent when the source
+     * can't say (snapshot path) — then contiguity falls back to index adjacency.
+     */
+    runningBucketsMs?: number[];
+    /**
+     * Bucket cadence of `runningSeries` and how many buckets the window SHOULD contain. The rows
+     * are NOT gap-filled, so this is the only way to tell "pinned for 60 of 60 minutes" from
+     * "two fresh samples arrived in a 60-minute window". Absent = cadence unknown; the legacy
+     * gap-free assumption then applies (received buckets spread evenly over the window).
+     */
+    sampling?: { bucketMinutes: number; expectedBuckets: number } | null;
     envLimit: number;
     throttledShare: number;
     worstQueue: { name: string; share: number } | null;
@@ -67,6 +99,12 @@ export const HEALTH_THRESHOLDS = {
     // trigger_surge: with NO usable baseline (normal 0), a multiplier is meaningless, so an
     // absolute floor picks the "new volume" cause instead of dropping to the v1 fallback.
     surgePerMin: 100,
+    // Minimum share of the window's EXPECTED buckets that must have arrived before a
+    // concurrency-shaped cause (pin / stall) may be named. Below it the series is too gappy to
+    // support a cause or a duration, so flow drops to a symptom-level verdict. Metric buckets are
+    // produced by queue activity rather than a heartbeat, so a sparse window is normal for a quiet
+    // env — and a saturation/stall claim isn't supportable there anyway.
+    minCoverage: 0.5,
   },
   attribution: { minShare: 0.5 }, // name a queue/task/region only when it owns >= half the problem
 };
@@ -99,6 +137,55 @@ function multiplierSeverity(
   return classifySeverity(value / normal, { warn: warnMult, crit: critMult });
 }
 
+/** True when the backlog depth could not be measured — `pending.now` is a placeholder. */
+export function isPendingUnknown(input: HealthInput): boolean {
+  return input.pending.availability === "unknown";
+}
+
+export type BucketCoverage = {
+  /** buckets the window should contain at the source's cadence. */
+  expectedBuckets: number;
+  /** buckets that actually arrived. */
+  receivedBuckets: number;
+  /** minutes per bucket. */
+  bucketMinutes: number;
+  /** received / expected. 1 when the cadence is unknown (legacy gap-free assumption). */
+  coverage: number;
+  /** enough of the window arrived to support a cause + a duration. */
+  sufficient: boolean;
+  /** true when the source told us its cadence (so gaps are detectable at all). */
+  known: boolean;
+};
+
+/**
+ * Coverage of the running series: how much of the window actually arrived. Without the source's
+ * cadence we can only assume the received buckets span the window evenly (the pre-existing
+ * assumption); with it, a gappy feed is visible and shares are expressed against EXPECTED buckets.
+ */
+export function bucketCoverage(input: HealthInput): BucketCoverage {
+  const received = input.flowEvidence.runningSeries.length;
+  const sampling = input.flowEvidence.sampling;
+  if (!sampling || sampling.expectedBuckets <= 0) {
+    return {
+      expectedBuckets: received,
+      receivedBuckets: received,
+      bucketMinutes: received > 0 ? input.windowMinutes / received : 0,
+      coverage: 1,
+      sufficient: true,
+      known: false,
+    };
+  }
+  const coverage = received / sampling.expectedBuckets;
+  return {
+    expectedBuckets: sampling.expectedBuckets,
+    receivedBuckets: received,
+    bucketMinutes: sampling.bucketMinutes,
+    coverage,
+    sufficient: coverage >= HEALTH_THRESHOLDS.flowCause.minCoverage,
+    known: true,
+  };
+}
+
 /** Look up a metric by id; throws if absent (buildMetrics guarantees the standard set exists). */
 export function metricById(metrics: Metric[], id: string): Metric {
   const m = metrics.find((x) => x.id === id);
@@ -133,32 +220,44 @@ export function buildMetrics(input: HealthInput): Metric[] {
     ),
   };
 
+  // Unmeasurable depth: `now` is a placeholder, so it must not be CLASSIFIED (a placeholder 0
+  // would read as a confident "no backlog" green). `availability: "unknown"` says so, and the
+  // flow analyzer turns it into an unassessable verdict.
+  const pendingUnknown = isPendingUnknown(input);
   const pending: Metric = {
     id: "pending",
     value: input.pending.now,
     unit: "count",
+    availability: pendingUnknown ? "unknown" : "measured",
     normal: input.pending.normal,
-    delta: delta(input.pending.now, input.pending.normal),
+    delta: pendingUnknown ? undefined : delta(input.pending.now, input.pending.normal),
     series: {
       points: input.pending.series,
       kind: input.pending.estimated ? "estimated" : "measured",
     },
-    severity: multiplierSeverity(
-      input.pending.now,
-      input.pending.normal,
-      t.pending.warnMult,
-      t.pending.critMult,
-      t.pending.floor
-    ),
+    severity: pendingUnknown
+      ? "ok"
+      : multiplierSeverity(
+          input.pending.now,
+          input.pending.normal,
+          t.pending.warnMult,
+          t.pending.critMult,
+          t.pending.floor
+        ),
   };
 
-  const net = input.throughput.donePerMin - input.throughput.triggeredPerMin;
+  // Net drain uses FINISHED (all terminal) runs — every terminal run leaves the queue, so
+  // completions alone would show a permanent deficit on any env with failures.
+  const net = input.throughput.finishedPerMin - input.throughput.triggeredPerMin;
   const throughput: Metric = {
     id: "throughput",
     value: net,
     unit: "perMin",
     aggregation: "rate",
-    breakdown: { done: input.throughput.donePerMin, triggered: input.throughput.triggeredPerMin },
+    breakdown: {
+      done: input.throughput.finishedPerMin,
+      triggered: input.throughput.triggeredPerMin,
+    },
     severity: net < 0 && isPendingIncreasing(input.pending.series) ? "warn" : "ok",
   };
 
@@ -262,10 +361,15 @@ export function buildMetrics(input: HealthInput): Metric[] {
 // ---------------------------------------------------------------------------
 
 export function computeDrain(input: HealthInput): { drainMinutes: number; isDrainable: boolean } {
-  const donePerMin = input.throughput.donePerMin;
-  const drainMinutes = donePerMin === 0 ? Number.POSITIVE_INFINITY : input.pending.now / donePerMin;
+  // Drain rate = runs LEAVING the queue (all terminal), not just successful completions.
+  const finishedPerMin = input.throughput.finishedPerMin;
+  const drainMinutes =
+    finishedPerMin === 0 ? Number.POSITIVE_INFINITY : input.pending.now / finishedPerMin;
   return {
     drainMinutes,
-    isDrainable: drainMinutes < HEALTH_THRESHOLDS.flowPolicy.drainCritMinutes,
+    // An unmeasurable depth can't produce an ETA — never offer "do nothing, it drains" off a
+    // placeholder.
+    isDrainable:
+      !isPendingUnknown(input) && drainMinutes < HEALTH_THRESHOLDS.flowPolicy.drainCritMinutes,
   };
 }