feat(notifications): idempotent per-period recap claim + generation audit trail

migrations/0129_maintainer_recap_claim.sql

@@ -0,0 +1,15 @@
+-- Per-period dedup marker for the cross-repo maintainer recap digest (#2249): collapse a retried cron tick /
+-- redelivered queue message to AT MOST ONE effective digest per period.
+--
+-- BEFORE: runMaintainerRecapJob has no idempotency of its own -- it re-scans every repo and re-posts to
+-- Discord on every invocation. Cloudflare Queues are at-least-once delivery, so a message can be redelivered
+-- after the consumer already completed the send (an ack that failed/timed out), producing a duplicate digest.
+--
+-- AFTER: claimMaintainerRecapPeriod performs an atomic conditional UPDATE on this singleton column, mirroring
+-- claimRegateFanoutSlot (0063) -- D1 serializes writes, so only the FIRST invocation for a given period_key
+-- (the current UTC date, "YYYY-MM-DD") matches the "unset or a different period" predicate and proceeds; a
+-- retried/redelivered invocation for the SAME period gets 0 changes and skips before any repo scan or send.
+--
+-- Reuses the global_agent_controls singleton (0059); nullable / no default -> backward-compatible (NULL = no
+-- recap has claimed a period yet, so the first one proceeds).
+ALTER TABLE global_agent_controls ADD COLUMN last_recap_period_key TEXT;

src/db/repositories.ts

@@ -2536,6 +2536,26 @@ export async function claimRegateFanoutSlot(env: Env, now: string, windowMs: num
   }
 }
 
+/** Atomic per-period dedup for the cross-repo maintainer recap digest (#2249): claim `periodKey` (the current
+ *  UTC date, "YYYY-MM-DD") as the singleton's last-sent period. Mirrors {@link claimRegateFanoutSlot}: the
+ *  conditional UPDATE matches only when the stored period is unset or DIFFERENT from `periodKey`, so a retried
+ *  cron tick or a redelivered (at-least-once) queue message for the SAME period gets 0 changes and skips
+ *  before any repo scan or Discord send. Fail-open on a driver error (return true → the digest still runs,
+ *  degrading to the pre-dedup behaviour rather than silently going dark). */
+export async function claimMaintainerRecapPeriod(env: Env, periodKey: string): Promise<boolean> {
+  try {
+    const result = await env.DB.prepare(
+      "UPDATE global_agent_controls SET last_recap_period_key = ?1 WHERE id = 'singleton' AND (last_recap_period_key IS NULL OR last_recap_period_key != ?1)",
+    )
+      .bind(periodKey)
+      .run();
+    /* v8 ignore next -- D1 update metadata normally includes changes; the ?? 0 fallback protects driver anomalies. */
+    return Number(result.meta.changes ?? 0) === 1;
+  } catch {
+    return true;
+  }
+}
+
 /** Flip the DB-backed global kill-switch (operator emergency brake; no redeploy required). */
 export async function setGlobalAgentFrozen(env: Env, frozen: boolean, updatedBy?: string | null): Promise<void> {
   await env.DB.prepare(

src/queue/processors.ts

@@ -1116,7 +1116,7 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
       // override #2250). Defense-in-depth: the cron only ENQUEUES this when enabled, but a stale in-flight job
       // that lands after a flag-flip (env OR manifest) must still no-op, so disabled does zero work here too.
       const maintainerRecapOverride = await resolveMaintainerRecapManifestOverride(env);
-      if (isRecapEnabled(env, maintainerRecapOverride)) await runMaintainerRecapJob(env, message.windowDays);
+      if (isRecapEnabled(env, maintainerRecapOverride)) await runMaintainerRecapJob(env, message.windowDays, maintainerRecapOverride);
       return;
     }
     case "agent-regate-sweep":

src/review/maintainer-recap-wire.ts

@@ -3,7 +3,7 @@
 // single-repo ReviewRecap job, which is manually-triggerable only (review-recap.ts). Flag-gated and OFF by
 // default, mirroring isOpsEnabled: flag-OFF, the cron enqueues no job and this module's exports are never
 // invoked, so the deploy is byte-identical to today.
-import { listRepositories } from "../db/repositories";
+import { claimMaintainerRecapPeriod, listRepositories, recordAuditEvent } from "../db/repositories";
 import { isAgentConfigured } from "../settings/autonomy";
 import { resolveRepositorySettings } from "../settings/repository-settings";
 import { buildRepoOutcomeCalibration } from "../services/outcome-calibration";
@@ -60,6 +60,16 @@ function normalizeRecapDayOfWeek(value: string | undefined): number {
   return Math.max(MIN_DAY_OF_WEEK, Math.min(MAX_DAY_OF_WEEK, Math.round(numeric)));
 }
 
+/** The effective cadence: a present manifest override wins outright, else the env knob (default weekly).
+ *  Shared by shouldFireMaintainerRecap (gating) and runMaintainerRecapJob (audit-event metadata only, #2251)
+ *  so there is exactly one place that resolves "what cadence is configured right now". */
+function resolveRecapCadence(
+  env: { GITTENSORY_RECAP_CADENCE?: string | undefined },
+  manifestOverride?: MaintainerRecapManifestOverride | undefined,
+): RecapCadence {
+  return manifestOverride?.present ? manifestOverride.cadence : normalizeRecapCadence(env.GITTENSORY_RECAP_CADENCE);
+}
+
 /**
  * True on the one cron tick per period the maintainer recap should fire: "daily" fires every day at the
  * configured hour; "weekly" fires ONLY on the configured day-of-week at that hour, so the tick fires at most
@@ -81,7 +91,7 @@ export function shouldFireMaintainerRecap(
   manifestOverride?: MaintainerRecapManifestOverride | undefined,
 ): boolean {
   if (hour !== normalizeRecapHour(env.GITTENSORY_RECAP_HOUR)) return false;
-  const cadence = manifestOverride?.present ? manifestOverride.cadence : normalizeRecapCadence(env.GITTENSORY_RECAP_CADENCE);
+  const cadence = resolveRecapCadence(env, manifestOverride);
   return cadence === "daily" || dayOfWeek === normalizeRecapDayOfWeek(env.GITTENSORY_RECAP_DAY);
 }
 
@@ -122,12 +132,44 @@ export async function resolveMaintainerRecapManifestOverride(env: Env): Promise<
   }
 }
 
+/** The current UTC calendar date ("YYYY-MM-DD") as the per-period claim key (#2249). Daily fires at most once
+ *  per date; weekly fires on only ONE designated date per week, so keying by date alone is correct for both
+ *  cadences without needing to encode which cadence produced the tick. */
+function computeRecapPeriodKey(now: Date): string {
+  return now.toISOString().slice(0, 10);
+}
+
+/** The channels this digest attempts today (#2251 audit metadata): runMaintainerRecap (#2252) always fans out
+ *  to both, each independently best-effort/no-op when unconfigured. */
+const RECAP_CHANNELS_ATTEMPTED = ["discord", "slack"] as const;
+
+/** A per-period claim already taken (#2249): the job never scanned repos, built a report, or delivered. */
+export type MaintainerRecapJobSkipped = { skipped: true; reason: "already_sent_this_period" };
+
 /**
  * Load aggregator inputs for every scan repo, then delegate to {@link runMaintainerRecap} for build → format →
- * dual-channel delivery. A per-repo aggregator failure is logged and that repo is skipped — one repo's D1 hiccup
- * must not blank the whole digest (mirrors ops-wire.ts's runOpsAlerts).
+ * dual-channel (Discord + Slack) delivery. A per-repo aggregator failure is logged and that repo is skipped --
+ * one repo's D1 hiccup must not blank the whole digest (mirrors ops-wire.ts's runOpsAlerts).
+ *
+ * Idempotent per UTC calendar date (#2249): claims the day via claimMaintainerRecapPeriod BEFORE doing any
+ * repo scan or send, so a retried cron tick / redelivered (at-least-once) queue message for a period already
+ * claimed short-circuits to `{ skipped: true, reason: "already_sent_this_period" }` without re-scanning repos
+ * or re-delivering.
+ *
+ * Records a `maintainer_recap_generated` audit event once the report is built (#2251), mirroring
+ * generateWeeklyValueReport's own audit call -- gives operators a ledger trail ("did the digest run today?")
+ * independent of the per-channel `maintainer_recap_notification.{discord,slack}` events deliverRecapToDiscord /
+ * deliverRecapToSlack already record for the send outcome itself.
  */
-export async function runMaintainerRecapJob(env: Env, windowDays?: number): Promise<RunMaintainerRecapResult> {
+export async function runMaintainerRecapJob(
+  env: Env,
+  windowDays?: number,
+  manifestOverride?: MaintainerRecapManifestOverride | undefined,
+): Promise<MaintainerRecapJobSkipped | RunMaintainerRecapResult> {
+  const periodKey = computeRecapPeriodKey(new Date());
+  const claimed = await claimMaintainerRecapPeriod(env, periodKey);
+  if (!claimed) return { skipped: true, reason: "already_sent_this_period" };
+
   const resolvedWindowDays = windowDays ?? DEFAULT_RECAP_WINDOW_DAYS;
   const repoNames = await recapScanRepos(env);
   const repos: MaintainerRecapRepoInput[] = [];
@@ -144,5 +186,27 @@ export async function runMaintainerRecapJob(env: Env, windowDays?: number): Prom
       );
     }
   }
-  return runMaintainerRecap(env, { windowDays: resolvedWindowDays, repos });
+  const result = await runMaintainerRecap(env, { windowDays: resolvedWindowDays, repos });
+  // unreachable implicit-else: runMaintainerRecap only returns skipped:true when explicitly passed
+  // `enabled: false`, which this call site never does -- the enable/disable decision already happened
+  // before runMaintainerRecapJob was ever invoked (isRecapEnabled, checked by the cron and the processor).
+  /* v8 ignore else */
+  if (!result.skipped) {
+    await recordAuditEvent(env, {
+      eventType: "maintainer_recap_generated",
+      actor: "gittensory",
+      route: "scheduled",
+      targetKey: `maintainer-recap:${periodKey}`,
+      outcome: "success",
+      detail: `${result.report.repos.length} repo(s), ${result.report.summary.length} section(s)`,
+      metadata: {
+        cadence: resolveRecapCadence(env, manifestOverride),
+        windowDays: resolvedWindowDays,
+        repoCount: result.report.repos.length,
+        sectionCount: result.report.summary.length,
+        channelsAttempted: [...RECAP_CHANNELS_ATTEMPTED],
+      },
+    });
+  }
+  return result;
 }

test/unit/db-parsers.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import {
+  claimMaintainerRecapPeriod,
   claimRegateFanoutSlot,
   countRecentDeadLetters,
   countRecentDeadLettersByType,
@@ -393,6 +394,20 @@ describe("database row parser hardening", () => {
     expect(await claimRegateFanoutSlot(broken, "2026-06-25T01:00:00.000Z", 90 * 1000)).toBe(true);
   });
 
+  it("claimMaintainerRecapPeriod: first claim for a period wins, a retry for the SAME period loses, a DIFFERENT period wins again (#2249)", async () => {
+    const env = createTestEnv();
+    expect(await claimMaintainerRecapPeriod(env, "2026-07-09")).toBe(true); // first claim (marker NULL)
+    expect(await claimMaintainerRecapPeriod(env, "2026-07-09")).toBe(false); // retried tick, same period → loses
+    expect(await claimMaintainerRecapPeriod(env, "2026-07-10")).toBe(true); // a new day → wins again
+    expect(await claimMaintainerRecapPeriod(env, "2026-07-10")).toBe(false); // retried again → loses
+  });
+
+  it("claimMaintainerRecapPeriod fails open (returns true) on a DB error so the digest never silently stalls", async () => {
+    const env = createTestEnv();
+    const broken = { ...env, DB: null } as unknown as typeof env;
+    expect(await claimMaintainerRecapPeriod(broken, "2026-07-09")).toBe(true);
+  });
+
   it("REGRESSION: a later GitHub sync does NOT clobber last_regated_at (omitted from the upsert SET clause)", async () => {
     const env = createTestEnv();
     await upsertPullRequestFromGitHub(env, "owner/repo", { number: 6, title: "First", state: "open", user: { login: "bob" }, labels: [] });

test/unit/maintainer-recap-wire.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { isRecapEnabled, resolveMaintainerRecapManifestOverride, runMaintainerRecapJob, shouldFireMaintainerRecap } from "../../src/review/maintainer-recap-wire";
+import type { MaintainerRecapJobSkipped } from "../../src/review/maintainer-recap-wire";
 import type { RunMaintainerRecapResult } from "../../src/services/maintainer-recap";
 import { updatePullRequestSlopAssessment, upsertPullRequestFromGitHub, upsertRepositorySettings } from "../../src/db/repositories";
 import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader";
@@ -9,7 +10,9 @@ const SELF_REPO = "JSONbored/gittensory";
 
 const HOOK = "https://discord.com/api/webhooks/123/abc";
 
-function ranRecap(result: RunMaintainerRecapResult): Extract<RunMaintainerRecapResult, { skipped: false }> {
+function ranRecap(
+  result: MaintainerRecapJobSkipped | RunMaintainerRecapResult,
+): Extract<RunMaintainerRecapResult, { skipped: false }> {
   expect(result.skipped).toBe(false);
   if (result.skipped) throw new Error("expected recap job to run");
   return result;
@@ -194,7 +197,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => {
 
     const { report } = ranRecap(await runMaintainerRecapJob(env, 30));
 
-    expect(report.windowDays).toBe(30);
+    expect(report).not.toBeNull();
+    expect(report!.windowDays).toBe(30);
   });
 
   it("prefers agent-configured repos over the full registered set when at least one is configured", async () => {
@@ -208,7 +212,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => {
 
     const { report } = ranRecap(await runMaintainerRecapJob(env));
 
-    expect(report.repos.map((r) => r.repoFullName)).toEqual(["owner/configured"]);
+    expect(report).not.toBeNull();
+    expect(report!.repos.map((r) => r.repoFullName)).toEqual(["owner/configured"]);
   });
 
   it("falls back to every registered repo when settings resolution errors for every repo (a settings blip must not abort the scan)", async () => {
@@ -224,7 +229,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => {
 
     const { report } = ranRecap(await runMaintainerRecapJob(env));
 
-    expect(report.repos.map((r) => r.repoFullName).sort()).toEqual(["owner/alpha", "owner/beta"]);
+    expect(report).not.toBeNull();
+    expect(report!.repos.map((r) => r.repoFullName).sort()).toEqual(["owner/alpha", "owner/beta"]);
   });
 
   it("fails safe per-repo: an aggregator error is logged and the repo is skipped; the job still delivers a (zeroed) report", async () => {
@@ -256,4 +262,75 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => {
     expect(delivery.discord).toEqual({ sent: true });
     expect(delivery.slack.sent).toBe(false);
   });
+
+  it("a retried tick within the SAME UTC date is a no-op: no repo scan, no second Discord post (#2249)", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-07-09T14:00:00.000Z"));
+    const env = createTestEnv({ DISCORD_WEBHOOK_URL: HOOK });
+    await seedRegisteredRepo(env, "owner/alpha");
+    await seedMergedPr(env, "owner/alpha", 1);
+    const posted = stubDiscordFetch();
+
+    const first = ranRecap(await runMaintainerRecapJob(env));
+    vi.setSystemTime(new Date("2026-07-09T14:02:00.000Z")); // same UTC date, a couple minutes later (a retry)
+    const second = await runMaintainerRecapJob(env);
+
+    expect(first.delivery.discord).toEqual({ sent: true });
+    expect(second).toEqual({ skipped: true, reason: "already_sent_this_period" });
+    expect(posted).toHaveLength(1); // the retry never re-scanned repos or re-posted
+    vi.useRealTimers();
+  });
+
+  it("a tick on a DIFFERENT UTC date gets its own fresh claim and sends again (#2249)", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-07-09T14:00:00.000Z"));
+    const env = createTestEnv({ DISCORD_WEBHOOK_URL: HOOK });
+    await seedRegisteredRepo(env, "owner/alpha");
+    await seedMergedPr(env, "owner/alpha", 1);
+    const posted = stubDiscordFetch();
+
+    const first = ranRecap(await runMaintainerRecapJob(env));
+    vi.setSystemTime(new Date("2026-07-10T14:00:00.000Z")); // next day
+    const second = ranRecap(await runMaintainerRecapJob(env));
+
+    expect(first.delivery.discord).toEqual({ sent: true });
+    expect(second.delivery.discord).toEqual({ sent: true });
+    expect(posted).toHaveLength(2);
+    vi.useRealTimers();
+  });
+
+  it("records a maintainer_recap_generated audit event with cadence/windowDays/repoCount/sectionCount/channelsAttempted metadata (#2251)", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-07-09T14:00:00.000Z"));
+    const env = createTestEnv({ DISCORD_WEBHOOK_URL: HOOK, GITTENSORY_RECAP_CADENCE: "daily" });
+    await seedRegisteredRepo(env, "owner/alpha");
+    await seedMergedPr(env, "owner/alpha", 1);
+    stubDiscordFetch();
+
+    await runMaintainerRecapJob(env, 14);
+
+    const row = await env.DB.prepare("select target_key, outcome, detail, metadata_json from audit_events where event_type = ? order by created_at desc limit 1")
+      .bind("maintainer_recap_generated")
+      .first<{ target_key: string; outcome: string; detail: string; metadata_json: string }>();
+    expect(row).toMatchObject({ target_key: "maintainer-recap:2026-07-09", outcome: "success" });
+    expect(row!.detail).toContain("1 repo(s)");
+    const metadata = JSON.parse(row!.metadata_json);
+    expect(metadata).toEqual({ cadence: "daily", windowDays: 14, repoCount: 1, sectionCount: expect.any(Number), channelsAttempted: ["discord", "slack"] });
+    vi.useRealTimers();
+  });
+
+  it("a present manifest override's cadence is reflected in the maintainer_recap_generated audit metadata, not the env value", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-07-09T14:00:00.000Z"));
+    const env = createTestEnv({ DISCORD_WEBHOOK_URL: HOOK, GITTENSORY_RECAP_CADENCE: "weekly" });
+    stubDiscordFetch();
+
+    await runMaintainerRecapJob(env, undefined, { present: true, enabled: true, cadence: "daily" });
+
+    const row = await env.DB.prepare("select metadata_json from audit_events where event_type = ? order by created_at desc limit 1")
+      .bind("maintainer_recap_generated")
+      .first<{ metadata_json: string }>();
+    expect(JSON.parse(row!.metadata_json).cadence).toBe("daily");
+    vi.useRealTimers();
+  });
 });