fix(review): prevent stale review cache and sweep starvation

docker-compose.yml

@@ -578,7 +578,7 @@ services:
     command:
       - /bin/sh
       - -c
-      - "sh /backup-metrics.sh"
+      - "apk add --no-cache busybox-extras && sh /backup-metrics.sh"
     healthcheck:
       test: ["CMD-SHELL", "wget -qO- http://127.0.0.1:9101/metrics | grep -q '^gittensory_backup_latest_timestamp_seconds'"]
       interval: 30s

prometheus/prometheus.yml

@@ -33,6 +33,9 @@ scrape_configs:
   # Backup freshness from the read-only backup-exporter sidecar. The metrics exist only when the backup
   # profile is active, which keeps backup alerts opt-in with the backup feature.
   - job_name: gittensory-backup
+    # The exporter is a tiny BusyBox httpd wrapper around Prometheus text output. Prometheus v3 rejects
+    # blank Content-Type responses unless the scrape protocol is explicit.
+    fallback_scrape_protocol: PrometheusText0.0.4
     static_configs:
       - targets: ["backup-exporter:9101"]
     scrape_interval: 60s

src/index.ts

@@ -17,7 +17,13 @@ import { isReviewExecutionJob, isSelfHostedReviewRuntime } from "./selfhost/revi
 import type { JobMessage } from "./types";
 
 const app = createApp();
-const REGATE_BACKPRESSURE_TYPES = ["agent-regate-pr", "agent-regate-sweep"] as const;
+// Scoped to the top-level fan-out TRIGGER only (#audit-sweep-fanout) — NOT "agent-regate-pr", whose per-repo
+// backlog is normal, expected, and can legitimately stay nonzero for long periods (staggered/rate-deferred
+// per-PR re-reviews), which is exactly what caused the prior broad backlog check to starve the scheduled sweep
+// entirely. A pending/processing "agent-regate-sweep" message means a fan-out is already in flight; the
+// per-repo drain guard (getLatestRegatedAt / isRegateSweepDraining) already protects individual repos once that
+// single fan-out runs, so this only needs to stop a SECOND trigger from queuing up behind the first.
+const REGATE_SWEEP_TRIGGER_TYPES = ["agent-regate-sweep"] as const;
 
 export { RateLimiter };
 
@@ -117,14 +123,17 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
         return null;
       })
     : null;
-  const regateBacklog = queueSnapshotBacklog(queueSnapshot, REGATE_BACKPRESSURE_TYPES);
+  const sweepTriggerBacklog = queueSnapshotBacklog(queueSnapshot, REGATE_SWEEP_TRIGGER_TYPES);
   let sweepThrottledUntil: string | undefined;
   if (selfHostedReviews) {
     sweepThrottledUntil = await shouldWaitForGitHubRateLimit(env, MAINTENANCE_RESERVED_HEADROOM);
     if (sweepThrottledUntil) {
       console.log(JSON.stringify({ event: "regate_sweep_throttled", resetAt: sweepThrottledUntil }));
-    } else if (regateBacklog > 0) {
-      console.log(JSON.stringify({ event: "regate_sweep_backlog_deferred", backlog: regateBacklog }));
+    } else if (sweepTriggerBacklog > 0) {
+      // A fan-out trigger is already pending/processing — skip re-arming so the queue never accumulates a
+      // second identical trigger behind the first (#audit-sweep-fanout). This is scoped to the trigger job
+      // itself; it does not look at (and is not blocked by) per-repo "agent-regate-pr" backlog.
+      console.log(JSON.stringify({ event: "regate_sweep_trigger_backlog_deferred", backlog: sweepTriggerBacklog }));
     } else {
       jobs.push({ type: "agent-regate-sweep", requestedBy: "schedule" });
     }
@@ -138,13 +147,12 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
     // per-repo segment + per-PR detail sync — a large GitHub-budget consumer second only to the sweep. Gate it
     // behind the SAME maintenance headroom the sweep yields at, so when the shared REST budget is low the backfill
     // SKIPS this 30-min tick and hands the remaining budget to webhooks (which drive timely reviews); the next
-    // 30-min tick retries, and after the bucket resets the backfill resumes. The cheap single-call health jobs
-    // (repair-data-fidelity, refresh-installation-health) stay unconditional — they cost ~one call and keep
-    // installation/health state fresh even while the budget is reserved.
-    if (selfHostedReviews && !sweepThrottledUntil && regateBacklog === 0) {
+    // 30-min tick retries, and after the bucket resets the backfill resumes. Queue depth is deliberately not a
+    // suppressor here: unrelated pending work can stay nonzero for long periods, while rate admission on the
+    // queued jobs is the precise throttle. The cheap single-call health jobs (repair-data-fidelity,
+    // refresh-installation-health) stay unconditional.
+    if (selfHostedReviews && !sweepThrottledUntil) {
       jobs.push({ type: "backfill-registered-repos", requestedBy: "schedule", mode: isFullSyncWindow ? "full" : "light" });
-    } else if (selfHostedReviews && regateBacklog > 0) {
-      console.log(JSON.stringify({ event: "backfill_backlog_deferred", backlog: regateBacklog }));
     } else if (selfHostedReviews) {
       console.log(JSON.stringify({ event: "backfill_throttled", resetAt: sweepThrottledUntil }));
     }

src/queue/processors.ts

@@ -221,6 +221,7 @@ import {
   queueSnapshotBacklog,
   queueSnapshotFromBinding,
 } from "../selfhost/queue-common";
+import { aiReviewCacheInputFingerprint } from "../review/ai-review-cache-input";
 import {
   downgradeCloseToHold,
   downgradeMergeToHold,
@@ -350,7 +351,6 @@ import {
   emptyReviewRagTelemetry,
   isRagEnabled,
 } from "../review/rag-wire";
-import { aiReviewCacheInputFingerprint } from "../review/ai-review-cache-input";
 import {
   buildReviewEnrichment,
   isEnrichmentEnabled,
@@ -5508,6 +5508,7 @@ async function maybePublishPrPublicSurface(
           agent: "dual-ai",
         },
         async () => {
+          const reviewManifest = await loadRepoFocusManifest(env, repoFullName).catch(() => null);
           // `.gittensory.yml` review.profile + review.path_instructions + review.exclude_paths (#review-profile /
           // #review-path-instructions / #review-exclude-paths): resolve from the manifest (cached from settings
           // resolution, so a cheap cache hit — no extra fetch) and thread them into the AI review. Profile shapes
@@ -5519,63 +5520,118 @@ async function maybePublishPrPublicSurface(
             pathInstructions: reviewPathInstructions,
             instructions: manifestReviewInstructions,
             excludePaths: reviewExcludePaths,
-          } = resolveReviewPromptOverrides(
-            /* v8 ignore next -- fail-open manifest-read rejection is exercised in runAiReviewForAdvisory; this wrapper preserves the same fallback. */
-            await loadRepoFocusManifest(env, repoFullName).catch(() => null),
-          );
+          } = resolveReviewPromptOverrides(reviewManifest);
           inlineCommentsEnabledForReview = shouldRequestInlineFindings(
             env,
             repoFullName,
             reviewInlineComments,
           );
+          const reviewFilesForAi = await getReviewFiles();
+          const changedPaths = reviewFilesForAi.map((file) => file.path);
           // Per-repo review CONTEXT (#review-skills): fold the container-private review/AGENTS.md (or legacy
           // review/CLAUDE.md) guide + the matching review/skills/*.md modules into the SAME review-instructions slot,
           // so reviews follow each repo's conventions.
           // Glob-gated for cost (only skills matching the changed files are injected); absent config dir ⇒ empty ⇒
-          // byte-identical prompt. getReviewFiles() is memoized, so this reuses the loaded diff.
-          const reviewFilesForAi = await getReviewFiles();
-          const changedReviewPaths = reviewFilesForAi.map((file) => file.path);
+          // byte-identical prompt. getReviewFiles() is memoized, so the second call reuses the loaded diff.
           const reviewInstructions =
             [
               manifestReviewInstructions,
               composeRepoReviewContext(
                 await loadRepoReviewContext(repoFullName),
-                changedReviewPaths,
+                changedPaths,
               ),
             ]
               .map((part) => part?.trim())
               .filter(Boolean)
               .join("\n\n") || null;
-          const reviewInputFingerprint = await aiReviewCacheInputFingerprint({
-            changedPaths: changedReviewPaths,
-            env,
+          const convergedRepoAllowed = isConvergenceRepoAllowed(env, repoFullName);
+          // Resolved ONCE and reused both for the fingerprint AND the cache-bypass decision below: grounding/RAG/
+          // enrichment/reputation each pull TIME-VARYING external context (live CI checks, the vector index,
+          // REES/CVE data, the submitter's evolving reputation) that can change for the SAME head SHA without
+          // any of these booleans flipping. Fingerprinting only "is the feature on" can't detect that drift
+          // without fetching the content itself (which would defeat caching), so a repo with ANY of these active
+          // bypasses the cache entirely rather than fingerprinting a value that can't prove freshness.
+          const dynamicReviewFeatures = {
+            grounding: isGroundingEnabled(env) && convergedRepoAllowed,
+            rag: resolveConvergedFeature(env, reviewManifest, "rag", repoFullName),
+            enrichment: isEnrichmentEnabled(env) && convergedRepoAllowed,
+            reputation: resolveConvergedFeature(
+              env,
+              reviewManifest,
+              "reputation",
+              repoFullName,
+            ),
+          };
+          const dynamicReviewContextActive =
+            dynamicReviewFeatures.grounding ||
+            dynamicReviewFeatures.rag ||
+            dynamicReviewFeatures.enrichment ||
+            dynamicReviewFeatures.reputation;
+          const inputFingerprint = await aiReviewCacheInputFingerprint({
+            title: pr.title,
             mode: settings.aiReviewMode,
-            pr: {
-              baseSha: webhook.baseSha,
-              title: pr.title,
-            },
-            review: {
-              effectiveInlineComments: inlineCommentsEnabledForReview,
-              excludePaths: reviewExcludePaths,
-              inlineComments: reviewInlineComments,
-              instructions: reviewInstructions,
-              pathInstructions: reviewPathInstructions,
-              profile: reviewProfile,
-            },
-            settings,
+            byok: settings.aiReviewByok,
+            provider: settings.aiReviewProvider,
+            model: settings.aiReviewModel,
+            aiReviewAllAuthors: settings.aiReviewAllAuthors,
+            aiReviewCloseConfidence: settings.aiReviewCloseConfidence,
+            gatePack: settings.gatePack,
+            reviewerPlan: env.AI_REVIEW_PLAN,
+            selfHostProviderConfig: env.AI_REVIEW_PLAN
+              ? {
+                  claudeModel: env.CLAUDE_AI_MODEL,
+                  claudeEffort: env.CLAUDE_AI_EFFORT,
+                  claudeTimeoutMs: env.CLAUDE_AI_TIMEOUT_MS,
+                  codexModel: env.CODEX_AI_MODEL,
+                  codexEffort: env.CODEX_AI_EFFORT,
+                  codexTimeoutMs: env.CODEX_AI_TIMEOUT_MS,
+                  ollamaBaseUrl: env.OLLAMA_AI_BASE_URL,
+                  ollamaModel: env.OLLAMA_AI_MODEL,
+                  openaiCompatibleBaseUrl: env.OPENAI_COMPATIBLE_AI_BASE_URL,
+                  openaiCompatibleModel: env.OPENAI_COMPATIBLE_AI_MODEL,
+                  openaiBaseUrl: env.OPENAI_AI_BASE_URL,
+                  openaiModel: env.OPENAI_AI_MODEL,
+                  anthropicBaseUrl: env.ANTHROPIC_AI_BASE_URL,
+                  anthropicModel: env.ANTHROPIC_AI_MODEL,
+                }
+              : null,
+            profile: reviewProfile,
+            inlineComments: inlineCommentsEnabledForReview,
+            pathInstructions: reviewPathInstructions,
+            pathGuidance: resolveReviewPathInstructions(
+              reviewPathInstructions,
+              changedPaths,
+            ),
+            repoInstructions: reviewInstructions,
+            excludePaths: reviewExcludePaths,
+            changedPaths,
+            baseSha: webhook.baseSha,
+            reviewFiles: reviewFilesForAi.map((file) => ({
+              path: file.path,
+              status: file.status,
+              patch: typeof file.payload?.patch === "string" ? file.payload.patch : undefined,
+              additions: file.additions,
+              deletions: file.deletions,
+            })),
+            features: dynamicReviewFeatures,
           });
-          // #1 self-host AI-review cache: reuse a prior review for this exact (repo, pr, head SHA, mode) ONLY when
-          // the prompt/config inputs that affect the model output still match. Private repo instructions, RAG
-          // suppressions, feature flags, inline-comment mode, and BYOK model choices all change review output even
-          // when the head SHA is unchanged, so they are folded into the stored input fingerprint.
-          const cachedReview = await getCachedAiReview(
-            env,
-            repoFullName,
-            pr.number,
-            advisory.headSha,
-            settings.aiReviewMode,
-            reviewInputFingerprint,
-          ).catch(() => null);
+          // #1 self-host AI-review cache: the LLM output for a PR changes only when the code (head SHA), review
+          // mode, reviewer plan, feature activation, or prompt-shaping inputs change. A re-delivered webhook or the
+          // block-mode re-gate sweep can reuse that exact review; stale same-head reviews from older private review
+          // instructions or feature config are intentionally treated as misses. The deterministic gate still runs.
+          // A repo with an active dynamic-context feature (grounding/RAG/enrichment/reputation) bypasses the
+          // cache entirely — see dynamicReviewContextActive above — since a cache hit there could replay a
+          // review built against now-stale external context for an otherwise-unchanged head.
+          const cachedReview = dynamicReviewContextActive
+            ? null
+            : await getCachedAiReview(
+                env,
+                repoFullName,
+                pr.number,
+                advisory.headSha,
+                settings.aiReviewMode,
+                inputFingerprint,
+              ).catch(() => null);
           if (cachedReview && hasPublicReviewAssessment(cachedReview.notes)) {
             advisory.findings.push(...cachedReview.findings);
             aiReview = cachedReview;
@@ -5595,7 +5651,7 @@ async function maybePublishPrPublicSurface(
               reviewExcludePaths,
               reviewInlineComments,
             });
-            if (aiReview && aiReview.cacheable !== false)
+            if (aiReview && aiReview.cacheable !== false && !dynamicReviewContextActive)
               await putCachedAiReview(
                 env,
                 repoFullName,
@@ -5607,7 +5663,7 @@ async function maybePublishPrPublicSurface(
                   metadata: {
                     /* v8 ignore next -- runAiReviewForAdvisory (the sole path reaching here) always sets metadata on its "ok" returns; the nullish fallback is a type-level (optional field) safeguard, not a reachable runtime path. */
                     ...(aiReview.metadata ?? {}),
-                    inputFingerprint: reviewInputFingerprint,
+                    inputFingerprint,
                   },
                 },
               ).catch(() => undefined);