fix(services): enqueue notify-deliver for the approval-queue staged-action and reminder badges (#10182)

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) -- 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' created + pending guard: a dedup hit or a
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

Author: shin-core

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,20 @@ 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.
+  // The `status === "pending"` arm mirrors evaluateAndEnqueueNotificationDeliveries' guard and honours the
+  // "send nothing for a suppressed row" contract, but this insert never passes a status, so the delivery is
+  // always pending here -- the false arm is unreachable from THIS caller (rate-limit suppression lives in
+  // evaluateNotificationEvent, a different helper), hence the ignore. `created` false IS reachable (a dedup).
+  /* v8 ignore next -- `delivery.status === "pending"` is always true from this caller; the guard is the shared contract */
+  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;
 }

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,21 @@ 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.
+        // As in stageForApproval: this insert never passes a status, so the delivery is always pending here --
+        // the false arm is unreachable from this caller. The check mirrors the shared enqueue contract.
+        /* v8 ignore next -- `inserted.delivery.status === "pending"` is always true from this caller */
+        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

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;
     }
   | {

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> = {}): 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<void> } }).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<void> } }).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" };

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<void> } }).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<void> } }).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, {