fix(integrations): page only for trouble an operator can act on

The watchdog opened an incident for any degraded state that lasted a
day, so an Apple Health account whose phone app stopped running read as
no_new_data for good and paged. Incidents now open for action_required,
for provider errors and for the watchdog's own unexpected silence; stale
or stalled data alone never pages. The first pass after a deploy, with
no watch row yet, records what stands as already announced instead of
paging every long-broken connection at once.

src/lib/integrations/__tests__/watchdog-classify.test.ts

@@ -244,6 +244,43 @@ describe("currentIncident", () => {
     ).toEqual({ reason: "unexpected_silence", since: ago(30) });
   });
 
+  it("never opens an incident for stale or stalled data on its own, however long it lasts", () => {
+    // An Apple Health account whose phone app stopped running reads as
+    // `no_new_data` for good; nothing an operator can do about it.
+    for (const reason of ["no_new_data", "sync_stalled"] as const) {
+      expect(
+        currentIncident({
+          health: health("degraded", reason, ago(24 * 90)),
+          silentSince: null,
+          ledgerAlerted: false,
+          now: NOW,
+        }),
+      ).toBeNull();
+    }
+  });
+
+  it("opens an incident for provider errors that have lasted more than a day", () => {
+    expect(
+      currentIncident({
+        health: health("degraded", "provider_errors", ago(25)),
+        silentSince: null,
+        ledgerAlerted: false,
+        now: NOW,
+      }),
+    ).toEqual({ reason: "provider_errors", since: ago(25) });
+  });
+
+  it("still opens a silence incident beside stale data", () => {
+    expect(
+      currentIncident({
+        health: health("degraded", "no_new_data", ago(50)),
+        silentSince: ago(30),
+        ledgerAlerted: false,
+        now: NOW,
+      }),
+    ).toEqual({ reason: "unexpected_silence", since: ago(30) });
+  });
+
   it("has no incident for a limited connection", () => {
     expect(
       currentIncident({

src/lib/integrations/watchdog-classify.ts

@@ -13,10 +13,14 @@
  *      is watched for errors only.
  *
  *   2. Is there an incident the operator should hear about?
- *      Unexpected silence; or a health state of `action_required` /
- *      `degraded` that has lasted longer than `INCIDENT_AFTER_MS`; or a
- *      failure streak the sync ledger already paged about (adopted, so the
- *      operator still gets the one recovery note when it ends).
+ *      Unexpected silence; or an `action_required` state, or a `degraded`
+ *      one because the provider keeps erroring (`provider_errors`), that has
+ *      lasted longer than `INCIDENT_AFTER_MS`; or a failure streak the sync
+ *      ledger already paged about (adopted, so the operator still gets the
+ *      one recovery note when it ends). `no_new_data` and `sync_stalled` on
+ *      their own never page: they mostly mean the owner stopped using the
+ *      source (an Apple Health account whose phone app has not run for a week
+ *      reads as stale for good), which is nothing an operator can fix.
  *
  * One incident per connection at a time. It is announced once and closed once;
  * a change of reason while it is open (silence turning into a reauth) is the
@@ -162,7 +166,13 @@ export function currentIncident(input: {
   if (health === null) return null;
   const troubled =
     health.state === "action_required" || health.state === "degraded";
-  if (troubled) {
+  // What pages on its own: the owner has to act, or the provider errors. A
+  // streak the ledger already paged about is followed whatever it reads as
+  // now, so its end is still reported.
+  const pages =
+    health.state === "action_required" ||
+    (health.state === "degraded" && health.reason === "provider_errors");
+  if (troubled && (pages || input.ledgerAlerted)) {
     // A provider-side reauth flag (Google Health) can stand without a ledger
     // streak to date it; the last success is then when it last worked.
     const sinceIso = health.since ?? health.lastSuccessAt;

src/lib/integrations/watchdog.ts

@@ -50,6 +50,13 @@ export interface WatchdogSummary {
   incidentsOpened: number;
   incidentsClosed: number;
   notified: number;
+  /**
+   * The pass found no watch row at all: the first one after the watchdog was
+   * deployed. Incidents standing at that moment are recorded as already
+   * announced instead of paged, so a deploy does not page once for every
+   * connection that has been broken or quiet for months.
+   */
+  firstRun: boolean;
 }
 
 const DAY_MS = 24 * 60 * 60 * 1000;
@@ -219,6 +226,7 @@ export async function runIntegrationWatchdog(opts: {
     incidentsOpened: 0,
     incidentsClosed: 0,
     notified: 0,
+    firstRun: (await prisma.integrationWatch.count()) === 0,
   };
   const userIds = await listWatchedUserIds();
   for (const userId of userIds) {
@@ -357,7 +365,13 @@ async function watchConnection(input: {
       summary.incidentsOpened += 1;
       incidentReason = transition.reason;
       incidentOpenedAt = now;
-      await announce(transition.reason, transition.since);
+      if (summary.firstRun) {
+        // Standing since before the watchdog first looked: recorded as
+        // announced, not paged (see `WatchdogSummary.firstRun`).
+        alertedAt = now;
+      } else {
+        await announce(transition.reason, transition.since);
+      }
       break;
     case "adopt":
       // The ledger already paged about this streak; record it as announced so
@@ -405,8 +419,17 @@ async function watchConnection(input: {
       break;
   }
 
-  // Nothing to remember and nothing remembered: no row.
-  if (row === null && silentSince === null && incidentReason === null) return;
+  // Nothing to remember and nothing remembered: no row. Except on the first
+  // pass, which leaves a row for every connection it looked at, so the pass
+  // after it is not taken for a first one too.
+  if (
+    row === null &&
+    silentSince === null &&
+    incidentReason === null &&
+    !(summary.firstRun && fact.connected)
+  ) {
+    return;
+  }
 
   await prisma.integrationWatch.upsert({
     where: { userId_integration: { userId, integration: key } },

src/lib/jobs/integration-watchdog.ts

@@ -57,6 +57,7 @@ function handlerFor(mode: WatchdogMode) {
             incidents_opened: s.incidentsOpened,
             incidents_closed: s.incidentsClosed,
             notified: s.notified,
+            first_run: s.firstRun ? 1 : 0,
           });
         } catch (err) {
           evt.addWarning(`integration watchdog failed: ${err}`);

src/lib/jobs/job-outcome.ts

@@ -113,6 +113,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "fetches",
   "finalised",
   "finalized",
+  // v1.42.4 — integration watchdog: the pass found no watch row (first run).
+  "first_run",
   "forced",
   "generated",
   "geo_backfill_carrier_resolved",

tests/integration/integration-watchdog.test.ts

@@ -92,6 +92,11 @@ beforeEach(async () => {
     },
   });
   vi.mocked(dispatchNotification).mockClear();
+  // The watchdog has run on this instance before: its first pass after a
+  // deploy records what stands without paging (pinned on its own below).
+  await prisma().integrationWatch.create({
+    data: { userId: ADMIN, integration: "withings" },
+  });
 });
 
 describe("integration watchdog", () => {
@@ -176,7 +181,9 @@ describe("integration watchdog", () => {
     const summary = await runIntegrationWatchdog({ mode: "hourly", now: NOW });
     expect(summary.silencesFlagged).toBe(0);
     expect(dispatchNotification).not.toHaveBeenCalled();
-    expect(await prisma().integrationWatch.count()).toBe(0);
+    expect(
+      await prisma().integrationWatch.count({ where: { userId: USER } }),
+    ).toBe(0);
   });
 
   it("never alerts on silence for an on-demand source such as a scale", async () => {
@@ -245,6 +252,115 @@ describe("integration watchdog", () => {
   });
 });
 
+describe("what pages and what does not", () => {
+  it("never pages for an Apple Health account whose phone app stopped delivering weeks ago", async () => {
+    // Daily data until forty days ago, then nothing: the push cadence reads
+    // stale (`no_new_data`) for good. The owner stopped using the app; there
+    // is nothing for an operator to do.
+    await prisma().user.update({
+      where: { id: USER },
+      data: {
+        ouraAccessTokenEncrypted: null,
+        healthKitLastSyncedAt: new Date(NOW.getTime() - 40 * DAY),
+      },
+    });
+    await prisma().measurement.createMany({
+      data: Array.from({ length: 30 }, (_, i) => ({
+        userId: USER,
+        type: "ACTIVITY_STEPS" as const,
+        value: 5000,
+        unit: "count",
+        source: "APPLE_HEALTH" as const,
+        measuredAt: new Date(NOW.getTime() - (40 + i) * DAY),
+      })),
+    });
+    vi.useFakeTimers({ toFake: ["Date"] });
+    vi.setSystemTime(NOW);
+    try {
+      const health = await loadConnectionHealth(USER, ["apple-health"]);
+      expect(health["apple-health"]?.state).toBe("degraded");
+      expect(health["apple-health"]?.reason).toBe("no_new_data");
+      // A week of hourly and daily passes.
+      for (let day = 0; day < 7; day++) {
+        const at = new Date(NOW.getTime() + day * DAY);
+        const hourly = await runIntegrationWatchdog({
+          mode: "hourly",
+          now: at,
+        });
+        const daily = await runIntegrationWatchdog({ mode: "daily", now: at });
+        expect(hourly.incidentsOpened + daily.incidentsOpened).toBe(0);
+      }
+      expect(dispatchNotification).not.toHaveBeenCalled();
+      expect(
+        await prisma().integrationWatch.count({
+          where: { userId: USER, incidentReason: { not: null } },
+        }),
+      ).toBe(0);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("records what stands on its first pass after a deploy without paging, and pages normally after", async () => {
+    await prisma().integrationWatch.deleteMany();
+    // A wearable that has been silent since before the watchdog existed.
+    await seedDailyReadings(
+      USER,
+      "OURA",
+      Array.from({ length: 13 }, (_, i) => i + 3),
+      54,
+    );
+    vi.useFakeTimers({ toFake: ["Date"] });
+    vi.setSystemTime(NOW);
+    try {
+      await recordSyncSuccess(USER, "oura");
+      const first = await runIntegrationWatchdog({ mode: "hourly", now: NOW });
+      expect(first.firstRun).toBe(true);
+      expect(first.incidentsOpened).toBe(1);
+      expect(dispatchNotification).not.toHaveBeenCalled();
+      const watch = await prisma().integrationWatch.findUniqueOrThrow({
+        where: { userId_integration: { userId: USER, integration: "oura" } },
+      });
+      expect(watch.incidentReason).toBe("unexpected_silence");
+      // Marked as announced, so no later pass pages it either.
+      expect(watch.alertedAt).not.toBeNull();
+
+      const later = new Date(NOW.getTime() + HOUR);
+      const second = await runIntegrationWatchdog({
+        mode: "hourly",
+        now: later,
+      });
+      expect(second.firstRun).toBe(false);
+      expect(dispatchNotification).not.toHaveBeenCalled();
+
+      // A new incident after the first pass pages as usual.
+      for (let i = 0; i < 3; i++) {
+        await recordSyncFailure({
+          userId: USER,
+          integration: "whoop",
+          kind: "transient",
+          message: "WHOOP error: 503",
+          errorCode: "503",
+        });
+      }
+      expect(dispatchNotification).toHaveBeenCalledTimes(1);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("leaves a row behind on a first pass that found nothing wrong, so the next pass is not a first one", async () => {
+    await prisma().integrationWatch.deleteMany();
+    await seedDailyReadings(USER, "OURA", [1, 2, 3], 2);
+    await recordSyncSuccess(USER, "oura");
+    const first = await runIntegrationWatchdog({ mode: "hourly", now: NOW });
+    expect(first.firstRun).toBe(true);
+    expect(await prisma().integrationWatch.count()).toBe(1);
+    const second = await runIntegrationWatchdog({ mode: "hourly", now: NOW });
+    expect(second.firstRun).toBe(false);
+  });
+});
+
 describe("recent-call ring", () => {
   it("keeps the newest ten attempts, newest first, without any message text", async () => {
     for (let i = 0; i < 7; i++) {