From 2e51a95ea7804abc57fd1d7d37ef42e87d91a1d7 Mon Sep 17 00:00:00 2001 From: shin-core <153108882+shin-core@users.noreply.github.com> Date: Fri, 31 Jul 2026 19:52:49 +0900 Subject: [PATCH] fix(services): enqueue notify-deliver for the approval-queue staged-action and reminder badges A notification_deliveries row is only visible in the feed once notify-deliver promotes it out of pending, and every other insert in the codebase pairs the insert with an enqueue. The approval queue's two inserts did not: stageForApproval's staged-action badge and sweepStaleApprovalQueue's #9032 reminder badge were both created at pending with no notify-deliver job, so they sat invisible until the stranded-delivery sweep rescued them 10+ minutes later (a rescue for a FAILED enqueue, not the primary path) -- directly contradicting the staleness module's framing of the staging badge as the notification the maintainer gets. Enqueue one notify-deliver per freshly-created pending delivery at both sites, mirroring evaluateAndEnqueueNotificationDeliveries' guard: a dedup hit or a rate-limit-suppressed non-pending row enqueues nothing. Best-effort: a rejected send is caught, warns approval_notification_enqueue_failed, and does not abort staging (still returns true) or the sweep loop (still counts the reminder). Extend the notify-deliver requestedBy union with 'agent-approval'. The dedup keys, the insert helper, deliverNotification, buildNotificationFeed, and the stranded sweep are unchanged. Closes #10025 --- src/services/agent-action-executor.ts | 12 +++++- src/services/agent-approval-queue.ts | 17 +++++++-- src/types.ts | 4 +- test/unit/agent-approval-queue.test.ts | 44 ++++++++++++++++++++++ test/unit/approval-queue-staleness.test.ts | 29 ++++++++++++++ 5 files changed, 101 insertions(+), 5 deletions(-) diff --git a/src/services/agent-action-executor.ts b/src/services/agent-action-executor.ts index dcd7e75c2a..cf879480f3 100644 --- a/src/services/agent-action-executor.ts +++ b/src/services/agent-action-executor.ts @@ -1465,7 +1465,7 @@ async function stageForApproval(env: Env, ctx: AgentActionExecutionContext, acti if (!created) return false; /* v8 ignore next -- a repo full name always has an owner segment; the empty fallback is purely defensive. */ const recipientLogin = ctx.repoFullName.split("/")[0] ?? ""; - await insertNotificationDeliveryIfAbsent(env, { + const { created: deliveryCreated, delivery } = await insertNotificationDeliveryIfAbsent(env, { dedupKey: `agent.pending_action:${ctx.repoFullName}#${ctx.pullNumber}:${action.actionClass}`, channel: "badge", recipientLogin, @@ -1477,5 +1477,15 @@ async function stageForApproval(env: Env, ctx: AgentActionExecutionContext, acti deeplink: `https://github.com/${ctx.repoFullName}/pull/${ctx.pullNumber}`, actorLogin: AGENT_ACTOR, }); + // #10025: enqueue the notify-deliver job that promotes this pending row to delivered — without it the badge + // sits invisible in notification_deliveries until the stranded-delivery sweep rescues it 10+ minutes later. + // Mirrors evaluateAndEnqueueNotificationDeliveries' `created && status === "pending"` guard: a dedup hit + // (already sent) or a rate-limit-suppressed non-pending row enqueues nothing. Best-effort: a failed send + // must not abort staging (still returns true), only log. + if (deliveryCreated && delivery.status === "pending") { + await env.JOBS.send({ type: "notify-deliver", requestedBy: "agent-approval", deliveryId: delivery.id }).catch((error: unknown) => { + console.warn(JSON.stringify({ event: "approval_notification_enqueue_failed", deliveryId: delivery.id, repoFullName: ctx.repoFullName, pullNumber: ctx.pullNumber, message: errorMessage(error).slice(0, 200) })); + }); + } return true; } diff --git a/src/services/agent-approval-queue.ts b/src/services/agent-approval-queue.ts index 593f604395..0a7b26f58c 100644 --- a/src/services/agent-approval-queue.ts +++ b/src/services/agent-approval-queue.ts @@ -24,6 +24,7 @@ import { resolvePerRepoContributorCapMatch } from "../queue/processors"; import { isBelowAccountAgeThreshold } from "../queue/account-age-throttle"; import { isAutoCloseExempt } from "../settings/auto-close-exempt"; import { isPerTenantAdmin } from "../auth/security"; +import { errorMessage } from "../utils/json"; import type { AgentPendingActionParams, AgentPendingActionRecord } from "../types"; export type ApprovalDecision = "accept" | "reject"; @@ -614,7 +615,7 @@ export async function sweepStaleApprovalQueue(env: Env, nowMs: number = Date.now const recipientLogin = row.repoFullName.split("/")[0] ?? ""; if (plan.kind === "remind") { const ageDays = plan.bucket; - const { created } = await insertNotificationDeliveryIfAbsent(env, { + const inserted = await insertNotificationDeliveryIfAbsent(env, { // The bucket index is what makes this fire again at all: the dedup key changes once per interval, so // the ~2-minute sweep cadence collapses to exactly one badge per interval with no extra persisted state. dedupKey: `agent.pending_action.reminder:${row.repoFullName}#${row.pullNumber}:${row.actionClass}:${plan.bucket}`, @@ -627,8 +628,18 @@ export async function sweepStaleApprovalQueue(env: Env, nowMs: number = Date.now body: `${row.reason ?? "A staged action"} — accept to execute it, or reject to cancel. It expires after ${Math.round(APPROVAL_EXPIRY_MS / (24 * 60 * 60 * 1000))} days.`, deeplink: `https://github.com/${row.repoFullName}/pull/${row.pullNumber}`, actorLogin: "loopover", - }).catch(() => ({ created: false })); - if (created) reminded += 1; + }).catch(() => null); + if (inserted?.created) { + reminded += 1; + // #10025: enqueue the notify-deliver job so the reminder badge — the #9032 escape hatch for a + // maintainer who missed the first badge — actually reaches the feed, instead of waiting 10+ minutes + // for the stranded-delivery sweep. Best-effort: a failed send still counts the reminder and continues. + if (inserted.delivery.status === "pending") { + await env.JOBS.send({ type: "notify-deliver", requestedBy: "agent-approval", deliveryId: inserted.delivery.id }).catch((error: unknown) => { + console.warn(JSON.stringify({ event: "approval_notification_enqueue_failed", deliveryId: inserted.delivery.id, repoFullName: row.repoFullName, pullNumber: row.pullNumber, message: errorMessage(error).slice(0, 200) })); + }); + } + } continue; } // Atomic pending→expired, so a maintainer accepting at the exact moment the sweep expires the row still diff --git a/src/types.ts b/src/types.ts index 99bb849ad4..c4105b08ac 100644 --- a/src/types.ts +++ b/src/types.ts @@ -222,7 +222,9 @@ export type JobMessage = } | { type: "notify-deliver"; - requestedBy: "notify-evaluate" | "test"; + // #10025: the approval-queue's staged-action + reminder badges enqueue their own notify-deliver jobs so + // they reach the feed without waiting on the stranded-delivery sweep. + requestedBy: "notify-evaluate" | "test" | "agent-approval"; deliveryId: string; } | { diff --git a/test/unit/agent-approval-queue.test.ts b/test/unit/agent-approval-queue.test.ts index d7940998f9..6f34a40c8a 100644 --- a/test/unit/agent-approval-queue.test.ts +++ b/test/unit/agent-approval-queue.test.ts @@ -84,6 +84,7 @@ import { upsertRepositorySettings, } from "../../src/db/repositories"; import { AGENT_LABEL_NEEDS_REVIEW, type PlannedAgentAction } from "../../src/settings/agent-actions"; +import { buildNotificationFeed, deliverNotification } from "../../src/notifications/service"; import { createTestEnv } from "../helpers/d1"; function ctx(over: Partial = {}): AgentActionExecutionContext { @@ -152,6 +153,49 @@ describe("agent approval queue (#779)", () => { expect(deliveries).toHaveLength(1); }); + // #10025: capture the notify-deliver jobs the staging path enqueues. + const withJobsCapture = (env: Env): Array<{ type: string; deliveryId?: string }> => { + const sent: Array<{ type: string; deliveryId?: string }> = []; + (env as unknown as { JOBS: { send: (msg: unknown) => Promise } }).JOBS = { send: async (msg) => void sent.push(msg as { type: string; deliveryId?: string }) }; + return sent; + }; + + it("#10025: staging enqueues exactly one notify-deliver job for the created badge; a second staging enqueues none", async () => { + const env = createTestEnv({}); + const sent = withJobsCapture(env); + await executeAgentMaintenanceActions(env, ctx(), [mergeApproval]); + const notifyJobs = sent.filter((m) => m.type === "notify-deliver"); + expect(notifyJobs).toHaveLength(1); + const delivery = (await listNotificationDeliveriesForRecipient(env, "owner")).find((d) => d.eventType === "agent.pending_action" && d.pullNumber === 7); + expect(notifyJobs[0]?.deliveryId).toBe(delivery?.id); + + // A second staging hits the dedup (created:false) → enqueues nothing more. + await executeAgentMaintenanceActions(env, ctx(), [mergeApproval]); + expect(sent.filter((m) => m.type === "notify-deliver")).toHaveLength(1); + }); + + it("#10025: a rejected notify-deliver send is caught, warns, and staging still returns queued", async () => { + const env = createTestEnv({}); + (env as unknown as { JOBS: { send: (msg: unknown) => Promise } }).JOBS = { send: async () => { throw new Error("queue down"); } }; + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const outcomes = await executeAgentMaintenanceActions(env, ctx(), [mergeApproval]); + expect(outcomes[0]?.outcome).toBe("queued"); // the send failure did not abort staging + expect(warn.mock.calls.map((c) => String(c[0])).some((m) => m.includes("approval_notification_enqueue_failed"))).toBe(true); + warn.mockRestore(); + }); + + it("#10025 REGRESSION: after staging + delivering the enqueued job, the recipient's feed contains the staged-action item", async () => { + const env = createTestEnv({}); + const sent = withJobsCapture(env); + await executeAgentMaintenanceActions(env, ctx(), [mergeApproval]); + const deliveryId = sent.find((m) => m.type === "notify-deliver")?.deliveryId; + expect(deliveryId).toBeDefined(); + // Run the job that promotes the pending row to delivered. + await deliverNotification(env, deliveryId!); + const feed = buildNotificationFeed("owner", await listNotificationDeliveriesForRecipient(env, "owner")); + expect(feed.notifications.some((item) => item.eventType === "agent.pending_action" && item.pullNumber === 7)).toBe(true); + }); + it("createPendingAgentActionIfAbsent reports created vs already-staged", async () => { const env = createTestEnv({}); const input = { repoFullName: "owner/repo", pullNumber: 7, installationId: 5, actionClass: "merge" as const, autonomyLevel: "auto_with_approval" as const, params: { mergeMethod: "squash" as const }, reason: "x" }; diff --git a/test/unit/approval-queue-staleness.test.ts b/test/unit/approval-queue-staleness.test.ts index 9754a604d0..7c7a4559cf 100644 --- a/test/unit/approval-queue-staleness.test.ts +++ b/test/unit/approval-queue-staleness.test.ts @@ -87,6 +87,35 @@ describe("sweepStaleApprovalQueue (#9032)", () => { expect(deliveries.every((delivery) => delivery.recipientLogin === "alice")).toBe(true); }); + it("#10025: a row aged past the reminder interval enqueues exactly one notify-deliver; a second sweep in the same bucket enqueues none", async () => { + const env = createTestEnv(); + const sent: Array<{ type: string; deliveryId?: string }> = []; + (env as unknown as { JOBS: { send: (msg: unknown) => Promise } }).JOBS = { send: async (msg) => void sent.push(msg as { type: string; deliveryId?: string }) }; + const id = await stage(env, 3); + const stagedAt = Date.parse((await getPendingAgentAction(env, id))!.createdAt); + + await sweepStaleApprovalQueue(env, stagedAt + APPROVAL_REMINDER_INTERVAL_MS); + const notifyJobs = sent.filter((m) => m.type === "notify-deliver"); + expect(notifyJobs).toHaveLength(1); + const reminder = (await listNotificationDeliveriesForRecipient(env, "alice", { limit: 50 })).find((d) => d.title.includes("Still waiting")); + expect(notifyJobs[0]?.deliveryId).toBe(reminder?.id); + + // A second sweep inside the SAME reminder bucket hits the dedup (created:false) → no further enqueue. + await sweepStaleApprovalQueue(env, stagedAt + APPROVAL_REMINDER_INTERVAL_MS + 60_000); + expect(sent.filter((m) => m.type === "notify-deliver")).toHaveLength(1); + }); + + it("#10025: a rejected notify-deliver send is caught, warns, and the sweep still counts the reminder", async () => { + const env = createTestEnv(); + (env as unknown as { JOBS: { send: (msg: unknown) => Promise } }).JOBS = { send: async () => { throw new Error("queue down"); } }; + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const id = await stage(env, 4); + const stagedAt = Date.parse((await getPendingAgentAction(env, id))!.createdAt); + expect(await sweepStaleApprovalQueue(env, stagedAt + APPROVAL_REMINDER_INTERVAL_MS)).toEqual({ reminded: 1, expired: 0 }); + expect(warn.mock.calls.map((c) => String(c[0])).some((m) => m.includes("approval_notification_enqueue_failed"))).toBe(true); + warn.mockRestore(); + }); + it("still writes a readable reminder for a row staged without a reason", async () => { const env = createTestEnv(); const { action } = await createPendingAgentActionIfAbsent(env, {