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 d5a88b6805..20651e283e 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 { 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 { +export async function runMaintainerRecapJob( + env: Env, + windowDays?: number, + manifestOverride?: MaintainerRecapManifestOverride | undefined, +): Promise { + 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; } 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 9b9b4f2495..dca741b9b2 100644 --- a/test/unit/maintainer-recap-wire.test.ts +++ b/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 { +function ranRecap( + result: MaintainerRecapJobSkipped | RunMaintainerRecapResult, +): Extract { 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(); + }); });