Skip to content

fix(queue): repair main-red syntax error from #3998 #4027

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
JSONbored wants to merge 1 commit into from
Closed

fix(queue): repair main-red syntax error from #3998 #4027

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
48 changes: 26 additions & 22 deletions src/queue/processors.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1080,13 +1080,13 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
// One bounded re-gate unit fanned out by the sweep (#audit-sweep-fanout): re-review + stamp a single PR.
await regatePullRequest(
env,
message.repairHeadSha,
message.repoFullName,
message.prNumber,
message.installationId,
message.deliveryId,
message.force,
message.prCreatedAt,
message.repairHeadSha,
);
return;
case "run-agent":
Expand Down Expand Up @@ -1462,13 +1462,17 @@ async function refreshOpenPullRequestsForScheduledSweep(
// surfaceRepairPriorityPullNumbers has no memory of prior attempts -- if the repair keeps failing for the SAME
// head SHA (e.g. every AI-provider attempt times out), it would otherwise re-select that PR forever, burning a
// fresh review attempt every cycle for zero output. These two constants cap that: once a SHA has already had
const REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA = 5;
// REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA dispatches recorded, it drops back to ordinary staleness-gated candidacy
// (still eventually re-checked, just not on every tick) and a single REGATE_REPAIR_EXHAUSTED_EVENT_TYPE audit
// event is recorded so the stuck PR is visible instead of silently retried forever. A new commit changes the
// head SHA, which resets the count naturally (the target key is scoped to repo+PR+SHA).
const REGATE_REPAIR_ATTEMPT_EVENT_TYPE = "agent.sweep.regate.repair_attempt";
const REGATE_REPAIR_EXHAUSTED_EVENT_TYPE = "agent.sweep.regate.repair_exhausted";
const REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA = 2;
// Bumped 2 -> 5 (#3998): the dispatch-time recording below double-counted rate-limit-deferred attempts as
// real ones, exhausting the budget before the repair ever actually ran. Recording moved to execution-time
// (regatePullRequest) makes each count a genuine attempt, so a higher cap gives real transient failures
// (e.g. a single AI-provider timeout) more room without letting a truly broken PR retry indefinitely.
const REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA = 5;
const REGATE_REPAIR_ATTEMPT_LOOKBACK_MS = 24 * 60 * 60 * 1000;

function regateRepairTargetKey(repoFullName: string, prNumber: number, headSha: string): string {
Expand Down Expand Up @@ -1871,21 +1875,21 @@ async function sweepRepoRegate(
const job: JobMessage = {
type: "agent-regate-pr",
deliveryId: isPriorityRepair
// #orb-retry-storm: pass the repair SHA so regatePullRequest can record the attempt at
// execution time (after rate-limit admission), not here at dispatch time. Jobs that are
// deferred or dropped before they run no longer count against the per-SHA cap.
...(isPriorityRepair && pr.headSha ? { repairHeadSha: pr.headSha } : {}),
? `regate-repair:${repoFullName}#${pr.number}`
: `regate-sweep:${repoFullName}#${pr.number}`,
repoFullName,
prNumber: pr.number,
installationId: sweepInstallationId,
targetKey: regateRepairTargetKey(repoFullName, pr.number, pr.headSha),
outcome: "queued",
detail: `outage-repair re-review dispatched for ${repoFullName}#${pr.number}`,
metadata: { repoFullName, prNumber: pr.number, headSha: pr.headSha },
});
}
...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}),
// #orb-retry-storm: pass the repair SHA so regatePullRequest can record the attempt at
// execution time (after rate-limit admission), not here at dispatch time. Jobs that are
// deferred or dropped before they run no longer count against the per-SHA cap.
...(isPriorityRepair && pr.headSha ? { repairHeadSha: pr.headSha } : {}),
};
const delaySeconds = Math.min(index * 10, 600);
await (delaySeconds > 0
? env.JOBS.send(job, { delaySeconds })
: env.JOBS.send(job));
}
}
await recordAuditEvent(env, {
Expand Down Expand Up @@ -2089,13 +2093,13 @@ async function sweepRepoBacklogConvergence(
// per-PR job always re-evaluates the head.
async function regatePullRequest(
env: Env,
repairHeadSha?: string,
repoFullName: string,
prNumber: number,
installationId: number,
deliveryId: string,
force?: boolean,
prCreatedAt?: string | null,
repairHeadSha?: string,
): Promise<void> {
// Reserve installation rate-limit headroom (#audit-rate-headroom): all repos share ONE GitHub App installation
// = ONE REST bucket, so when the shared budget is low, DEFER this re-review until the reset instead of
Expand All @@ -2117,11 +2121,17 @@ async function regatePullRequest(
await env.JOBS.send(
{
type: "agent-regate-pr",
...(repairHeadSha ? { repairHeadSha } : {}),
deliveryId,
repoFullName,
prNumber,
installationId,
...(prCreatedAt ? { prCreatedAt } : {}),
...(force ? { force: true } : {}),
...(repairHeadSha ? { repairHeadSha } : {}),
},
{ delaySeconds: delayUntil(rateResetAt) },
);
return;
}
// #orb-retry-storm: record the repair attempt NOW — after rate-limit admission — so the cap in
// surfaceRepairPriorityPullNumbers counts actual executions, not queued dispatches that may have
Expand All @@ -2132,16 +2142,10 @@ async function regatePullRequest(
eventType: REGATE_REPAIR_ATTEMPT_EVENT_TYPE,
actor: "gittensory",
targetKey: regateRepairTargetKey(repoFullName, prNumber, repairHeadSha),
outcome: "started",
outcome: "queued",
detail: `outage-repair re-review executing for ${repoFullName}#${prNumber}`,
metadata: { repoFullName, prNumber, headSha: repairHeadSha },
});
...(prCreatedAt ? { prCreatedAt } : {}),
...(force ? { force: true } : {}),
},
{ delaySeconds: delayUntil(rateResetAt) },
);
return;
}
const settings = await resolveRepositorySettings(env, repoFullName);
await reReviewStoredPullRequest(
Expand Down
35 changes: 20 additions & 15 deletions test/unit/queue.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6861,12 +6861,13 @@ describe("queue processors", () => {
await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9407);
await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" });
// PR 1: missing its current Gate check for its current head -- would ordinarily be flagged outage-repair
// priority on every tick. Pre-seed 3 prior repair-attempt audit events for this EXACT head SHA to simulate a
// review that keeps failing (e.g. a timeout) and never publishes a completed gate check.
// priority on every tick. Pre-seed REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA (5) prior repair-attempt audit
// events for this EXACT head SHA to simulate a review that keeps failing (e.g. a timeout) and never
// publishes a completed gate check.
await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 1, title: "Stuck repair", state: "open", user: { login: "c" }, head: { sha: "stuck-sha" }, labels: [], body: "" });
await repositoriesModule.markPullRequestSurfacePublished(env, "owner/agent-repo", 1, "stuck-sha");
const targetKey = "owner/agent-repo#1#stuck-sha";
for (let i = 0; i < 3; i += 1) {
for (let i = 0; i < 5; i += 1) {
await repositoriesModule.recordAuditEvent(env, {
eventType: "agent.sweep.regate.repair_attempt",
actor: "gittensory",
Expand Down Expand Up @@ -6894,18 +6895,18 @@ describe("queue processors", () => {
const attempts = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?")
.bind("agent.sweep.regate.repair_attempt", targetKey)
.first<{ n: number }>();
expect(attempts?.n).toBe(3);
expect(attempts?.n).toBe(5);
// Sentry-visible signal (via the structured-log forwarder) fires exactly once alongside the audit event.
const exhaustedLogs = errors.mock.calls.filter(([line]) => typeof line === "string" && line.includes("regate_repair_exhausted"));
expect(exhaustedLogs).toHaveLength(1);
const logged = JSON.parse(exhaustedLogs[0]![0] as string) as Record<string, unknown>;
expect(logged).toMatchObject({ level: "error", event: "regate_repair_exhausted", repo: "owner/agent-repo", pullNumber: 1, headSha: "stuck-sha", attempts: 3 });
expect(logged).toMatchObject({ level: "error", event: "regate_repair_exhausted", repo: "owner/agent-repo", pullNumber: 1, headSha: "stuck-sha", attempts: 5 });
} finally {
errors.mockRestore();
}
});

it("REGRESSION (#orb-retry-storm): a repair dispatch under the attempt cap records a repair_attempt audit event, and a second sweep tick does not duplicate the repair_exhausted event once already flagged", async () => {
it("REGRESSION (#orb-retry-storm, #3998): a repair dispatch alone does not record a repair_attempt audit event (only execution does), and a second sweep tick does not duplicate the repair_exhausted event once already flagged", async () => {
const sent: import("../../src/types").JobMessage[] = [];
const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue });
await upsertInstallation(env, { action: "created", installation: { id: 9408, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } });
Expand All @@ -6920,14 +6921,18 @@ describe("queue processors", () => {

const fanned = sent.filter((job): job is Extract<import("../../src/types").JobMessage, { type: "agent-regate-pr" }> => job.type === "agent-regate-pr");
expect(fanned.map((job) => job.deliveryId)).toContain("regate-repair:owner/agent-repo#2");
const attemptsAfterFirst = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?")
// #orb-retry-storm (#3998): dispatching the repair job no longer records an attempt by itself -- only
// actually EXECUTING it does (inside regatePullRequest, after rate-limit admission), so a deferred or
// dropped dispatch that never runs no longer exhausts the per-SHA cap.
const attemptsAfterDispatch = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?")
.bind("agent.sweep.regate.repair_attempt", targetKey)
.first<{ n: number }>();
expect(attemptsAfterFirst?.n).toBe(1);
expect(attemptsAfterDispatch?.n).toBe(0);

// Manually push this SHA over the cap, then run the sweep twice more -- the exhausted event must be recorded
// only once even though the PR is (re-)evaluated on every tick.
for (let i = 0; i < 2; i += 1) {
// Manually seed REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA (5) executed attempts to push this SHA over the cap,
// then run the sweep twice more -- the exhausted event must be recorded only once even though the PR is
// (re-)evaluated on every tick.
for (let i = 0; i < 5; i += 1) {
await repositoriesModule.recordAuditEvent(env, {
eventType: "agent.sweep.regate.repair_attempt",
actor: "gittensory",
Expand Down Expand Up @@ -27223,7 +27228,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri
// #terminal-outcome-audit: the disposition counter's "close" action_class, with the actual gate-blocker
// code (missing_linked_issue, from the default linkedIssueGateMode:block + no-linked-issue body) as the
// bounded blocker_class -- proof this reaches the real gate.blockers, not just a hardcoded label.
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="close",autonomy_level="auto",blocker_class="missing_linked_issue"} 1');
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="close",autonomy_level="auto",blocker_class="missing_linked_issue",repo="redacted-1"} 1');
const nativeDecision = await env.DB.prepare("select decision, summary, source from review_audit where event_type = 'gate_decision' and target_id = ?").bind(`${REPO}#60`).first<{ decision: string; summary: string; source: string }>();
expect(nativeDecision).toMatchObject({ decision: "close", summary: "missing_linked_issue", source: "gittensory-native" });
});
Expand Down Expand Up @@ -27266,7 +27271,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri

expect(seen.merged).toBe(false);
expect(seen.closed).toBe(false);
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="auto",blocker_class="guardrail_hold"} 1');
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="auto",blocker_class="guardrail_hold",repo="redacted-1"} 1');
const holdAudit = await env.DB.prepare("select metadata_json from audit_events where event_type = 'agent.action.hold' order by created_at desc limit 1").first<{ metadata_json: string }>();
expect(JSON.parse(holdAudit?.metadata_json ?? "{}")).toMatchObject({
repoFullName: REPO,
Expand Down Expand Up @@ -27394,7 +27399,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri
const mergeAudit = await env.DB.prepare("select count(*) as n from audit_events where event_type = 'agent.action.merge'").first<{ n: number }>();
expect(mergeAudit?.n).toBeGreaterThanOrEqual(1);
// #terminal-outcome-audit: the disposition counter's "merge" action_class, on the actual live call site.
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="merge",autonomy_level="auto",blocker_class="none"} 1');
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="merge",autonomy_level="auto",blocker_class="none",repo="redacted-1"} 1');
});

// #terminal-outcome-audit: end-to-end proof that the LIVE runAgentMaintenancePlanAndExecute call site (not just
Expand Down Expand Up @@ -27468,7 +27473,7 @@ describe("auto-action convergence: end-to-end plan+execute for the general heuri
// early return -- this is the exact "hold, but no audit_events row at all" shape (the breaker downgrade
// leaves no merge/close action) that previously had zero aggregate signal. close autonomy is unset in this
// repo's settings (only merge/approve are configured), so it resolves to the default "observe".
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="observe",blocker_class="none"} 1');
expect(await renderMetrics()).toContain('gittensory_agent_disposition_total{action_class="hold",autonomy_level="observe",blocker_class="none",repo="redacted-1"} 1');
const holdAudit = await env.DB.prepare("select detail, metadata_json from audit_events where event_type = 'agent.action.hold' order by created_at desc limit 1").first<{ detail: string; metadata_json: string }>();
expect(holdAudit?.detail).toBe("auto-action held by precision circuit breaker");
expect(JSON.parse(holdAudit?.metadata_json ?? "{}")).toMatchObject({
Expand Down