From 671106ecf5a83bef5e4f8a9df21e69a6f45fd3a9 Mon Sep 17 00:00:00 2001 From: oktofeesh1 <287075021+oktofeesh1@users.noreply.github.com> Date: Thu, 9 Jul 2026 21:46:11 +0200 Subject: [PATCH] feat(notifications): idempotent per-period recap claim + generation audit trail Adds a per-UTC-date claim (migration 0129, mirroring claimRegateFanoutSlot's atomic conditional-UPDATE pattern) so a retried cron tick or redelivered queue message never re-scans repos or re-sends the maintainer recap digest for a period already sent. Also records a maintainer_recap_generated audit event once the report is built, giving operators a ledger trail independent of the per-channel delivery-outcome event. --- migrations/0129_maintainer_recap_claim.sql | 15 ++++ src/db/repositories.ts | 20 +++++ src/queue/processors.ts | 2 +- src/review/maintainer-recap-wire.ts | 57 ++++++++++++- test/unit/db-parsers.test.ts | 15 ++++ test/unit/maintainer-recap-wire.test.ts | 97 ++++++++++++++++++++-- 6 files changed, 193 insertions(+), 13 deletions(-) create mode 100644 migrations/0129_maintainer_recap_claim.sql diff --git a/migrations/0129_maintainer_recap_claim.sql b/migrations/0129_maintainer_recap_claim.sql new file mode 100644 index 0000000000..ed8115090b --- /dev/null +++ b/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; diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 89b0b8ad34..b202db5fa2 100644 --- a/src/db/repositories.ts +++ b/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 { + 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 { await env.DB.prepare( diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 1a37af006c..ba6a0bedd6 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1116,7 +1116,7 @@ export async function processJob(env: Env, message: JobMessage): Promise { // 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": diff --git a/src/review/maintainer-recap-wire.ts b/src/review/maintainer-recap-wire.ts index 10d5357b3e..c67fec0e3e 100644 --- a/src/review/maintainer-recap-wire.ts +++ b/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 { loadGatePrecisionReport } from "../services/gate-precision"; @@ -62,6 +62,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 @@ -83,7 +93,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); } @@ -124,15 +134,41 @@ 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). Discord-only for now, mirroring the + * channel field's own default/only-supported-value in the focus-manifest schema (#2250) -- kept as its own + * constant so a future Slack delivery path only needs to change this one place. */ +const RECAP_CHANNELS_ATTEMPTED = ["discord"] as const; + /** * Build the cross-repo RecapReport (#2239) over the recap's scan repos and deliver it to Discord. 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). deliverRecapToDiscord itself never throws (best-effort webhook). + * + * 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 `report: null` without re-scanning repos or re-posting to Discord. + * + * 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` event deliverRecapToDiscord already + * records for the send outcome itself. */ export async function runMaintainerRecapJob( env: Env, windowDays?: number, -): Promise<{ report: RecapReport; delivery: { sent: boolean; reason?: string } }> { + manifestOverride?: MaintainerRecapManifestOverride | undefined, +): Promise<{ report: RecapReport | null; delivery: { sent: boolean; reason?: string } }> { + const periodKey = computeRecapPeriodKey(new Date()); + const claimed = await claimMaintainerRecapPeriod(env, periodKey); + if (!claimed) return { report: null, delivery: { sent: false, reason: "already_sent_this_period" } }; + const resolvedWindowDays = windowDays ?? DEFAULT_RECAP_WINDOW_DAYS; const repoNames = await recapScanRepos(env); const repos: MaintainerRecapRepoInput[] = []; @@ -150,6 +186,21 @@ export async function runMaintainerRecapJob( } } const report = buildMaintainerRecap({ generatedAt: nowIso(), windowDays: resolvedWindowDays, repos }); + await recordAuditEvent(env, { + eventType: "maintainer_recap_generated", + actor: "gittensory", + route: "scheduled", + targetKey: `maintainer-recap:${periodKey}`, + outcome: "success", + detail: `${report.repos.length} repo(s), ${report.summary.length} section(s)`, + metadata: { + cadence: resolveRecapCadence(env, manifestOverride), + windowDays: resolvedWindowDays, + repoCount: report.repos.length, + sectionCount: report.summary.length, + channelsAttempted: [...RECAP_CHANNELS_ATTEMPTED], + }, + }); const delivery = await deliverRecapToDiscord(env, report); return { report, delivery }; } diff --git a/test/unit/db-parsers.test.ts b/test/unit/db-parsers.test.ts index 77cd99228e..52072d9356 100644 --- a/test/unit/db-parsers.test.ts +++ b/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: [] }); diff --git a/test/unit/maintainer-recap-wire.test.ts b/test/unit/maintainer-recap-wire.test.ts index ea56ff71d8..7d8af0db50 100644 --- a/test/unit/maintainer-recap-wire.test.ts +++ b/test/unit/maintainer-recap-wire.test.ts @@ -172,9 +172,10 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => { const { report, delivery } = await runMaintainerRecapJob(env); expect(delivery).toEqual({ sent: true }); - expect(report.windowDays).toBe(7); // default when omitted - expect(report.repos.map((r) => r.repoFullName).sort()).toEqual(["owner/alpha", "owner/beta"]); - expect(report.totals.merged).toBe(3); // 1 (alpha) + 2 (beta) + expect(report).not.toBeNull(); + expect(report!.windowDays).toBe(7); // default when omitted + expect(report!.repos.map((r) => r.repoFullName).sort()).toEqual(["owner/alpha", "owner/beta"]); + expect(report!.totals.merged).toBe(3); // 1 (alpha) + 2 (beta) expect(posted).toHaveLength(1); }); @@ -186,7 +187,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => { const { report } = 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 () => { @@ -200,7 +202,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => { const { report } = 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 () => { @@ -216,7 +219,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => { const { report } = 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 () => { @@ -230,7 +234,8 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => { const { report, delivery } = await runMaintainerRecapJob(env); // resolves (never throws) - expect(report.repos).toEqual([]); + expect(report).not.toBeNull(); + expect(report!.repos).toEqual([]); expect(delivery).toEqual({ sent: true }); const logged = warnings.mock.calls.map((c) => String(c[0])).find((line) => line.includes("maintainer_recap_repo_error") && line.includes("owner/alpha")); expect(logged).toBeDefined(); @@ -242,8 +247,82 @@ describe("runMaintainerRecapJob — cross-repo digest (#1963, #2248)", () => { const { report, delivery } = await runMaintainerRecapJob(env); - expect(report.repos).toEqual([]); - expect(report.totals.gateFalsePositiveRate).toBeNull(); + expect(report).not.toBeNull(); + expect(report!.repos).toEqual([]); + expect(report!.totals.gateFalsePositiveRate).toBeNull(); expect(delivery).toEqual({ sent: true }); }); + + 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 = 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.report).not.toBeNull(); + expect(first.delivery).toEqual({ sent: true }); + expect(second).toEqual({ report: null, delivery: { sent: false, 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 = await runMaintainerRecapJob(env); + vi.setSystemTime(new Date("2026-07-10T14:00:00.000Z")); // next day + const second = await runMaintainerRecapJob(env); + + expect(first.report).not.toBeNull(); + expect(second.report).not.toBeNull(); + expect(second.delivery).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"] }); + 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(); + }); });