fix(webapp): honest watch check data semantics (review F8-F10)

- a stale queuedAt is never labeled as queue latency — resumed/retrying
  runs report time from creation
- a stale zero depth bucket can't declare a drain: satisfied requires a
  live counter or a provably fresh bucket, else unavailable
- error recurrence uses errors_v1 last_seen (ms precision), includes the
  creation minute, and marks approximate counts instead of overclaiming

Author: kathiekiwi

apps/webapp/app/services/dashboardAgentWatchChecks.server.ts

@@ -10,7 +10,8 @@
  *     same seam the queue pages and the waiting-run module use — with the
  *     ClickHouse depth series as fallback.
  *   - queue existence: ONE Postgres `TaskQueue` point-read.
- *   - error recurrence: ClickHouse `error_occurrences_v1` only.
+ *   - error recurrence: ClickHouse `errors_v1` for WHETHER it recurred (millisecond
+ *     `last_seen`), plus the per-minute `error_occurrences_v1` rollup for the count.
  *   - health: the existing health report (loader + interpreter), unchanged.
  *
  * Readers THROW on failure rather than swallowing it, because `checkWatch` turns a
@@ -69,16 +70,32 @@ export async function watchQueueExists(environmentId: string, queueName: string)
 /** How far back the ClickHouse depth fallback looks when the live counter is down. */
 const DEPTH_FALLBACK_MINUTES = 10;
 const DEPTH_FALLBACK_BUCKET_SECONDS = 60;
+/**
+ * How far behind `now` the newest analytics bucket may END and still be read as
+ * "the queue right now". One bucket of slack: anything older leaves a gap the
+ * rollup hasn't covered, and runs queued in that gap would be invisible.
+ */
+const DEPTH_FRESH_TOLERANCE_MS = DEPTH_FALLBACK_BUCKET_SECONDS * 1000;
 
 function formatClickhouseDateTime(date: Date): string {
   return date.toISOString().slice(0, 19).replace("T", " ");
 }
 
+/** ClickHouse renders DateTime without a zone; the column is UTC. */
+function parseClickhouseDateTime(value: string): Date {
+  return new Date(`${value.replace(" ", "T")}Z`);
+}
+
 /**
  * Current pending count for one queue. The live run-queue counter is the truth
  * ("is it drained RIGHT NOW"); ClickHouse is the fallback and reports the most
- * recent bucket's PEAK depth, which can only over-report — so a fallback zero
- * still means "nothing was queued in that bucket", never a false drain.
+ * recent bucket's PEAK depth, which can only over-report within that bucket.
+ *
+ * The fallback carries `current`, and it is false unless the newest bucket
+ * actually reaches the present: a rollup that's minutes behind may hold an empty
+ * bucket while runs piled up after it, and reading that as "drained" is the one
+ * mistake this watch must never make. `checkBacklogDrain` turns a stale zero into
+ * `unavailable`.
  */
 export async function readWatchQueueDepth(
   environment: AuthenticatedEnvironment,
@@ -87,7 +104,7 @@ export async function readWatchQueueDepth(
 ): Promise<WatchQueueDepth | null> {
   const live = await engine.lengthOfQueue(environment, queueName).catch(() => null);
   if (typeof live === "number" && Number.isFinite(live)) {
-    return { depth: live, source: "live_queue" };
+    return { depth: live, source: "live_queue", current: true, asOf: now };
   }
 
   const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
@@ -114,18 +131,66 @@ export async function readWatchQueueDepth(
 
   // Newest bucket wins — the closest thing the rollup has to "now".
   const newest = rows.reduce((best, row) => (row.bucket > best.bucket ? row : best), rows[0]!);
-  return { depth: newest.depth, source: "queue_metrics" };
+  const bucketEnd = new Date(parseClickhouseDateTime(newest.bucket).getTime() + bucketMs);
+  const current = bucketEnd.getTime() >= now.getTime() - DEPTH_FRESH_TOLERANCE_MS;
+
+  return { depth: newest.depth, source: "queue_metrics", current, asOf: bucketEnd };
+}
+
+const MINUTE_MS = 60_000;
+
+type OrganizationClickhouse = Awaited<
+  ReturnType<typeof clickhouseFactory.getClickhouseForOrganization>
+>;
+
+/**
+ * The fingerprint's most recent occurrence, at MILLISECOND precision, from the
+ * `errors_v1` aggregate (`max(last_seen)` over every task that produced it).
+ *
+ * This is what makes "has it come back?" answerable exactly. The per-minute
+ * `error_occurrences_v1` rollup can only place an error in a minute, and the
+ * minute a watch is created in holds BOTH the error that prompted the watch and
+ * any recurrence seconds later — so the rollup alone can neither confirm nor deny
+ * a recurrence in that first minute.
+ */
+async function readErrorLastSeen(
+  clickhouse: OrganizationClickhouse,
+  environment: AuthenticatedEnvironment,
+  fingerprint: string
+): Promise<Date | null> {
+  const builder = clickhouse.errors.activeErrorsSinceQueryBuilder();
+  builder.where("organization_id = {organizationId: String}", {
+    organizationId: environment.organizationId,
+  });
+  builder.where("project_id = {projectId: String}", { projectId: environment.projectId });
+  builder.where("environment_id = {environmentId: String}", { environmentId: environment.id });
+  builder.where("error_fingerprint = {fingerprint: String}", { fingerprint });
+  builder.groupBy("environment_id, task_identifier, error_fingerprint");
+
+  const [error, rows] = await builder.execute();
+  if (error) throw error;
+  if (!rows || rows.length === 0) return null;
+
+  let lastSeenMs = 0;
+  for (const row of rows) {
+    const ms = Number(row.last_seen);
+    if (Number.isFinite(ms) && ms > lastSeenMs) lastSeenMs = ms;
+  }
+
+  return lastSeenMs > 0 ? new Date(lastSeenMs) : null;
 }
 
 /**
- * The first occurrence of an error fingerprint after `since`, from the per-minute
- * `error_occurrences_v1` rollup.
+ * What we know about an error fingerprint relative to `since`, from two reads that
+ * each answer what only they can:
  *
- * Buckets are filtered with `minute > since`, i.e. STRICTLY after, so an
- * occurrence in the same minute the watch was created can't be read as a
- * recurrence. That under-counts by at most the creation minute — the conservative
- * direction, since a watch must never fire on the error that prompted it.
- * `occurredAt` is therefore minute-granular by construction.
+ *   - `errors_v1` decides WHETHER it recurred, to the millisecond. An occurrence
+ *     40 seconds after the watch was created is a recurrence, and rounding the
+ *     window up to the next minute used to lose it entirely.
+ *   - `error_occurrences_v1` supplies HOW MANY and, for minutes after the creation
+ *     minute, when. Its creation-minute bucket can't be split between the original
+ *     error and a recurrence, so those occurrences only make `countSince` a lower
+ *     bound (`countApproximate`) — never a claim.
  */
 export async function readWatchErrorRecurrence(
   environment: AuthenticatedEnvironment,
@@ -137,31 +202,66 @@ export async function readWatchErrorRecurrence(
     "logs"
   );
 
+  const lastSeenAt = await readErrorLastSeen(clickhouse, environment, fingerprint);
+  // Never seen in this environment at all.
+  if (!lastSeenAt) return null;
+
+  const notRecurred: WatchErrorRecurrence = {
+    occurredAt: null,
+    occurredAtPrecision: null,
+    countSince: 0,
+    countApproximate: false,
+    lastSeenAt,
+  };
+  if (lastSeenAt.getTime() <= since.getTime()) return notRecurred;
+
+  // Something landed after `since`. The rollup fills in the count and the minute.
+  const sinceMinuteMs = Math.floor(since.getTime() / MINUTE_MS) * MINUTE_MS;
   const queryBuilder = clickhouse.errors.createOccurrencesQueryBuilder("INTERVAL 1 MINUTE");
   queryBuilder.where("organization_id = {organizationId: String}", {
     organizationId: environment.organizationId,
   });
   queryBuilder.where("project_id = {projectId: String}", { projectId: environment.projectId });
   queryBuilder.where("environment_id = {environmentId: String}", { environmentId: environment.id });
   queryBuilder.where("error_fingerprint = {fingerprint: String}", { fingerprint });
-  queryBuilder.where("minute > toStartOfMinute(fromUnixTimestamp64Milli({sinceMs: Int64}))", {
+  // The creation minute is INCLUDED — its occurrences are what the old
+  // `minute > since` filter dropped.
+  queryBuilder.where("minute >= toStartOfMinute(fromUnixTimestamp64Milli({sinceMs: Int64}))", {
     sinceMs: since.getTime(),
   });
   queryBuilder.groupBy("error_fingerprint, bucket_epoch");
   queryBuilder.orderBy("bucket_epoch ASC");
 
   const [error, rows] = await queryBuilder.execute();
   if (error) throw error;
-  if (!rows || rows.length === 0) return null;
 
-  let earliest = rows[0]!.bucket_epoch;
-  let countSince = 0;
-  for (const row of rows) {
-    if (row.bucket_epoch < earliest) earliest = row.bucket_epoch;
-    countSince += row.count;
+  let earliestAfterMs: number | null = null;
+  let countAfter = 0;
+  let creationMinuteCount = 0;
+
+  for (const row of rows ?? []) {
+    const bucketMs = row.bucket_epoch * 1000;
+    if (bucketMs <= sinceMinuteMs) {
+      creationMinuteCount += row.count;
+      continue;
+    }
+    countAfter += row.count;
+    if (earliestAfterMs === null || bucketMs < earliestAfterMs) earliestAfterMs = bucketMs;
   }
 
-  return { occurredAt: new Date(earliest * 1000), countSince };
+  // The earliest time we can PROVE an occurrence at: a bucket that starts after
+  // the creation minute, or — when the only evidence is in that minute, or the
+  // rollup hasn't caught up — the exact `last_seen`.
+  const useBucket = earliestAfterMs !== null && earliestAfterMs < lastSeenAt.getTime();
+
+  return {
+    occurredAt: useBucket ? new Date(earliestAfterMs!) : lastSeenAt,
+    occurredAtPrecision: useBucket ? "minute" : "exact",
+    // At least the one `errors_v1` proved, even if the rollup lags behind it.
+    countSince: Math.max(1, countAfter),
+    countApproximate: creationMinuteCount > 0,
+    lastSeenAt,
+  };
 }
 
 const HEALTH_SEVERITIES = new Set<string>(["ok", "warn", "crit"]);

apps/webapp/app/services/dashboardAgentWatchChecks.ts

@@ -41,16 +41,38 @@ export type WatchRunRow = {
 };
 
 export type WatchQueueDepth = {
-  /** Current pending count for the queue. */
+  /** Pending count for the queue, as of `asOf`. */
   depth: number;
   source: "live_queue" | "queue_metrics";
+  /**
+   * Whether the reading describes the queue RIGHT NOW. A live counter always
+   * does; an analytics bucket only does while it's fresh enough to cover the
+   * present. A stale reading can never answer "drained" — see
+   * `checkBacklogDrain`.
+   */
+  current: boolean;
+  /** What instant the reading describes, when it isn't the live counter. */
+  asOf?: Date;
 };
 
-/** The first occurrence of the watched error after the watch's `since`, if any. */
+/** What we know about the watched error's occurrences relative to `since`. */
 export type WatchErrorRecurrence = {
-  occurredAt: Date;
-  /** Occurrences counted in the same window. */
+  /**
+   * The earliest occurrence PROVEN to be after `since`, or null when nothing has
+   * recurred. Null with a `lastSeenAt` means "seen before, not since".
+   */
+  occurredAt: Date | null;
+  /** How precisely `occurredAt` is known: to the millisecond, or to its minute. */
+  occurredAtPrecision: "exact" | "minute" | null;
+  /** Occurrences after `since`. A LOWER BOUND when `countApproximate`. */
   countSince: number;
+  /**
+   * True when `countSince` can't be split exactly — occurrences in the minute
+   * the watch was created can't be told apart from the error that prompted it.
+   */
+  countApproximate: boolean;
+  /** The fingerprint's most recent occurrence, whenever it was. */
+  lastSeenAt: Date | null;
 };
 
 export type WatchHealthSeverity = "ok" | "warn" | "crit";
@@ -73,7 +95,10 @@ export type WatchCheckDeps = {
   queueExists: (queue: string) => Promise<boolean>;
   /** Current pending count, live run-queue first with a ClickHouse fallback. */
   readQueueDepth: (queue: string) => Promise<WatchQueueDepth | null>;
-  /** First occurrence of `fingerprint` strictly after `since`, plus the count. */
+  /**
+   * What's known about `fingerprint` relative to `since`. `null` means the
+   * fingerprint has no occurrences at all in this environment.
+   */
   readErrorRecurrence: (fingerprint: string, since: Date) => Promise<WatchErrorRecurrence | null>;
   /** The health report's current verdict for the watch's environment. */
   readHealth: () => Promise<WatchHealthSnapshot | null>;
@@ -129,9 +154,14 @@ function formatMs(ms: number): string {
 export type WatchWaitBasis = "queued_at" | "delay_until" | "created_at";
 
 /**
- * The wait a run has accumulated, with the ONLY label the data supports:
- * `queuedAt` present -> a real queue wait; a future `delayUntil` -> a schedule,
- * not latency; otherwise time from creation, said out loud.
+ * The wait a run has accumulated, with the ONLY label the data supports: a
+ * `queuedAt` that belongs to THIS attempt -> a real queue wait; a future
+ * `delayUntil` -> a schedule, not latency; otherwise time from creation, said out
+ * loud.
+ *
+ * A resumed/retried/paused run's `queuedAt` is a leftover from the first enqueue,
+ * so it is not measured from at all: a number that isn't this attempt's queue
+ * wait must never be worded as one, even with a flag next to it.
  */
 export function describeRunWait(
   run: WatchRunRow,
@@ -140,13 +170,13 @@ export function describeRunWait(
   waitMs: number | null;
   waitBasis: WatchWaitBasis;
   waitLabel: string;
-  /** False when `queuedAt` can't be read as this attempt's queue wait. */
+  /** True only when the wait IS this attempt's queue wait. */
   queueWaitReliable: boolean;
 } {
   const queueWaitReliable = run.queuedAt !== null && !STALE_QUEUED_AT_STATUSES.has(run.status);
   const end = run.startedAt ?? now;
 
-  if (run.queuedAt) {
+  if (run.queuedAt && queueWaitReliable) {
     const waitMs = Math.max(0, end.getTime() - run.queuedAt.getTime());
     return {
       waitMs,
@@ -165,11 +195,16 @@ export function describeRunWait(
     };
   }
 
+  // Either there is no `queuedAt`, or the one we have belongs to an earlier
+  // attempt. Both fall back to the run's age, and say that's what it is.
   const waitMs = Math.max(0, end.getTime() - run.createdAt.getTime());
+  const resumeOrRetry = run.queuedAt !== null;
   return {
     waitMs,
     waitBasis: "created_at",
-    waitLabel: `time from creation: ${formatMs(waitMs)}`,
+    waitLabel: resumeOrRetry
+      ? `waiting to ${run.status === "RETRYING_AFTER_FAILURE" ? "retry" : "resume"}; time from creation: ${formatMs(waitMs)}`
+      : `time from creation: ${formatMs(waitMs)}`,
     queueWaitReliable,
   };
 }
@@ -256,6 +291,11 @@ export async function checkRunFinished(
  * backlog_drain — satisfied when the queue's current pending count is 0. A queue
  * that no longer exists can never drain in an observable sense, so that's
  * `terminal_unsatisfied`; a depth we can't read is `unavailable`, never "drained".
+ *
+ * A zero needs a reading that describes NOW (`current`): a stale analytics bucket
+ * that happened to be empty says nothing about the runs queued after it, so it's
+ * `unavailable` too. A stale NON-zero depth is still worth reporting — the queue
+ * demonstrably wasn't empty — and is marked approximate.
  */
 export async function checkBacklogDrain(
   spec: Extract<WatchSpec, { kind: "backlog_drain" }>,
@@ -277,7 +317,18 @@ export async function checkBacklogDrain(
     return { result: "unavailable", facts: { queue: spec.queue, reason: "depth_unavailable" } };
   }
 
-  const facts = { queue: spec.queue, depth: depth.depth, depthSource: depth.source };
+  const facts = {
+    queue: spec.queue,
+    depth: depth.depth,
+    depthSource: depth.source,
+    depthAsOf: depth.asOf?.toISOString() ?? null,
+    depthApproximate: !depth.current,
+  };
+
+  if (depth.depth === 0 && !depth.current) {
+    return { result: "unavailable", facts: { ...facts, reason: "depth_stale" } };
+  }
+
   return { result: depth.depth === 0 ? "satisfied" : "pending", facts };
 }
 
@@ -291,9 +342,13 @@ export function normalizeErrorFingerprint(fingerprint: string): string {
 }
 
 /**
- * error_recurrence — satisfied on the first occurrence strictly after the
- * server-set `since`. `since` is never caller-set, so the model can't backdate
- * the window and make a pre-existing error look like a recurrence.
+ * error_recurrence — satisfied on the first occurrence proven to be after the
+ * server-set `since`. `since` is never caller-set, so the model can't backdate the
+ * window and make a pre-existing error look like a recurrence.
+ *
+ * The facts carry the PRECISION of what they claim (`occurredAtPrecision`,
+ * `countApproximate`) and, when nothing recurred, when the error was last seen —
+ * so the wake narration can't assert more than the data supports.
  */
 export async function checkErrorRecurrence(
   spec: Extract<WatchSpec, { kind: "error_recurrence" }>,
@@ -305,15 +360,24 @@ export async function checkErrorRecurrence(
   const base = { fingerprint, since: input.since.toISOString() };
 
   if (!recurrence) {
-    return { result: "pending", facts: { ...base, countSince: 0 } };
+    return { result: "pending", facts: { ...base, countSince: 0, lastSeenAt: null } };
+  }
+
+  const lastSeenAt = recurrence.lastSeenAt?.toISOString() ?? null;
+
+  if (!recurrence.occurredAt) {
+    return { result: "pending", facts: { ...base, countSince: 0, lastSeenAt } };
   }
 
   return {
     result: "satisfied",
     facts: {
       ...base,
       occurredAt: recurrence.occurredAt.toISOString(),
+      occurredAtPrecision: recurrence.occurredAtPrecision,
       countSince: recurrence.countSince,
+      countApproximate: recurrence.countApproximate,
+      lastSeenAt,
     },
   };
 }