Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion src/services/agent-action-executor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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;
}
20 changes: 17 additions & 3 deletions src/services/agent-approval-queue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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}`,
Expand All @@ -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
Expand Down
4 changes: 3 additions & 1 deletion src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
| {
Expand Down
44 changes: 44 additions & 0 deletions test/unit/agent-approval-queue.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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" };
Expand Down
29 changes: 29 additions & 0 deletions test/unit/approval-queue-staleness.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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, {
Expand Down