diff --git a/.loopover.yml.example b/.loopover.yml.example index 71d43f9c41..a490d030fe 100644 --- a/.loopover.yml.example +++ b/.loopover.yml.example @@ -1428,6 +1428,14 @@ settings: # prReconciliation: # enabled: true # Bool. Default: false (the env var decides instead). +# Fleet-wide active_review_tracking reconciliation sweep (#webhook-reorder-clobber): config-as-code override +# for the LOOPOVER_ACTIVE_REVIEW_RECONCILIATION flag. A delayed webhook job can leave a PR's review-tracking +# row stuck "active" after the PR has actually already closed on GitHub (queue backpressure widens the race). +# This sweep re-checks stale "active" rows against LIVE GitHub state and terminalizes the ones confirmed +# closed. Same shape and precedence as `prReconciliation:` above. +# activeReviewReconciliation: +# enabled: true # Bool. Default: false (the env var decides instead). + # Opt-in federated fleet intelligence export (#1970): packages this instance's OWN calibration signals into a # signed, anonymized bundle an operator can choose to hand to a peer (or to a collector they run). Exports # AGGREGATE figures only -- gate precision, reversal/slop/copycat rates over a window -- never source code, diff --git a/config/examples/loopover.full.yml b/config/examples/loopover.full.yml index 139544a79f..a256f4404f 100644 --- a/config/examples/loopover.full.yml +++ b/config/examples/loopover.full.yml @@ -1442,6 +1442,14 @@ settings: # prReconciliation: # enabled: true # Bool. Default: false (the env var decides instead). +# Fleet-wide active_review_tracking reconciliation sweep (#webhook-reorder-clobber): config-as-code override +# for the LOOPOVER_ACTIVE_REVIEW_RECONCILIATION flag. A delayed webhook job can leave a PR's review-tracking +# row stuck "active" after the PR has actually already closed on GitHub (queue backpressure widens the race). +# This sweep re-checks stale "active" rows against LIVE GitHub state and terminalizes the ones confirmed +# closed. Same shape and precedence as `prReconciliation:` above. +# activeReviewReconciliation: +# enabled: true # Bool. Default: false (the env var decides instead). + # Opt-in federated fleet intelligence export (#1970): packages this instance's OWN calibration signals into a # signed, anonymized bundle an operator can choose to hand to a peer (or to a collector they run). Exports # AGGREGATE figures only -- gate precision, reversal/slop/copycat rates over a window -- never source code, diff --git a/migrations/0172_pull_requests_github_updated_at.sql b/migrations/0172_pull_requests_github_updated_at.sql new file mode 100644 index 0000000000..fb51c1c508 --- /dev/null +++ b/migrations/0172_pull_requests_github_updated_at.sql @@ -0,0 +1,8 @@ +-- Out-of-order webhook guard (#webhook-reorder-clobber): a webhook queued behind a slow/congested job can be +-- processed AFTER a later event for the same PR already landed and applied a newer state -- its embedded PR +-- snapshot is then stale and must not regress lifecycle-identity fields (state/headSha/mergedAt) GitHub has +-- already moved past. This column stores GitHub's OWN `updated_at` for the PR (distinct from `updated_at`, +-- which is app bookkeeping) so upsertPullRequestFromGitHub can compare incoming vs. stored before applying a +-- write. NULL for every existing row -- the guard fails open (applies the write) whenever it has nothing to +-- compare against, so this backfills itself the next time each PR is synced. +ALTER TABLE pull_requests ADD COLUMN github_updated_at TEXT; diff --git a/packages/loopover-engine/src/config-lint.ts b/packages/loopover-engine/src/config-lint.ts index 2fbb9fe625..3896341f5e 100644 --- a/packages/loopover-engine/src/config-lint.ts +++ b/packages/loopover-engine/src/config-lint.ts @@ -25,6 +25,7 @@ const TOP_LEVEL_FIELDS = [ "upstreamDriftIssues", "sweepWatchdog", "prReconciliation", + "activeReviewReconciliation", "federatedIntelligence", ] as const; diff --git a/packages/loopover-engine/src/focus-manifest-validation.ts b/packages/loopover-engine/src/focus-manifest-validation.ts index b280cc66ff..7bb5e5bf10 100644 --- a/packages/loopover-engine/src/focus-manifest-validation.ts +++ b/packages/loopover-engine/src/focus-manifest-validation.ts @@ -13,6 +13,7 @@ import { upstreamDriftIssuesConfigToJson, sweepWatchdogConfigToJson, prReconciliationConfigToJson, + activeReviewReconciliationConfigToJson, federatedIntelligenceConfigToJson, settingsOverrideToJson, type FocusManifest, @@ -95,6 +96,8 @@ function focusManifestToNormalizedJson(manifest: FocusManifest): Record; + const enabled = normalizeOptionalBoolean(record.enabled, "activeReviewReconciliation.enabled", warnings) ?? false; + return { present: true, enabled }; +} + +/** Serialize an activeReviewReconciliation config back into the parse-compatible shape so a cached snapshot + * round-trips through {@link parseActiveReviewReconciliationConfig} unchanged. Returns null when nothing is + * configured. */ +export function activeReviewReconciliationConfigToJson(config: FocusManifestActiveReviewReconciliationConfig): JsonValue { + if (!config.present) return null; + return { enabled: config.enabled }; +} + /** * Parse the optional `federatedIntelligence:` mapping (#1970). Mirrors {@link parseUpstreamDriftIssuesConfig} * exactly -- `enabled` is the only field, defaulting to false, so the parsed value IS the effective value and @@ -3884,6 +3927,7 @@ export function parseFocusManifest(raw: unknown, source?: FocusManifestSource): upstreamDriftIssues: parseUpstreamDriftIssuesConfig(record.upstreamDriftIssues, warnings), sweepWatchdog: parseSweepWatchdogConfig(record.sweepWatchdog, warnings), prReconciliation: parsePrReconciliationConfig(record.prReconciliation, warnings), + activeReviewReconciliation: parseActiveReviewReconciliationConfig(record.activeReviewReconciliation, warnings), federatedIntelligence: parseFederatedIntelligenceConfig(record.federatedIntelligence, warnings), warnings, }; @@ -3910,6 +3954,7 @@ export function parseFocusManifest(raw: unknown, source?: FocusManifestSource): !manifest.upstreamDriftIssues.present && !manifest.sweepWatchdog.present && !manifest.prReconciliation.present && + !manifest.activeReviewReconciliation.present && !manifest.federatedIntelligence.present ) { warnings.push("Manifest contained no recognized focus fields; falling back to deterministic signals."); diff --git a/packages/loopover-engine/src/index.ts b/packages/loopover-engine/src/index.ts index dfd90c9f23..72cb858ff4 100644 --- a/packages/loopover-engine/src/index.ts +++ b/packages/loopover-engine/src/index.ts @@ -789,6 +789,7 @@ export { upstreamDriftIssuesConfigToJson, sweepWatchdogConfigToJson, prReconciliationConfigToJson, + activeReviewReconciliationConfigToJson, federatedIntelligenceConfigToJson, FEDERATED_COLLECTOR_MODES, settingsOverrideToJson, diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 54a16fdee6..5835efe287 100644 --- a/src/db/repositories.ts +++ b/src/db/repositories.ts @@ -1,5 +1,5 @@ import { parsePullRequestTargetKey } from "@loopover/engine"; -import { and, asc, desc, eq, gte, inArray, isNotNull, not, or, sql, type SQL } from "drizzle-orm"; +import { and, asc, desc, eq, gte, inArray, isNotNull, lt, not, or, sql, type SQL } from "drizzle-orm"; import { getDb } from "./client"; import { activeReviewTracking, @@ -350,7 +350,6 @@ export async function upsertPullRequestFromGitHub( const record = toPullRequestRecord(repoFullName, pr); const db = getDb(env.DB); const syncedAt = nowIso(); - const lastSeenOpenAt = pr.state === "open" ? (options.seenOpenAt ?? syncedAt) : null; const existingClaimRows = await db .select({ linkedIssuesJson: pullRequests.linkedIssuesJson, @@ -359,6 +358,9 @@ export async function upsertPullRequestFromGitHub( bodyObservedAt: pullRequests.bodyObservedAt, headSha: pullRequests.headSha, headShaObservedAt: pullRequests.headShaObservedAt, + state: pullRequests.state, + mergedAt: pullRequests.mergedAt, + githubUpdatedAt: pullRequests.githubUpdatedAt, }) .from(pullRequests) .where(and(eq(pullRequests.repoFullName, repoFullName), eq(pullRequests.number, pr.number))) @@ -371,6 +373,33 @@ export async function upsertPullRequestFromGitHub( // `linkedIssues.length === 0` branch) on any such upsert. Fall back to whatever is already stored in that // case; only a genuinely observed (possibly empty) body updates the claim. (#linked-issue-sparse-payload-preserve) const existingClaimRow = existingClaimRows[0]; + // Out-of-order webhook guard (#webhook-reorder-clobber): a webhook for an OLDER event (e.g. `review_requested`) + // can be dequeued AFTER a NEWER event for the same PR (e.g. `closed`) already landed, if the older job was + // stuck behind queue backpressure -- its embedded `pull_request` snapshot is then stale and must not regress + // state/headSha/mergedAt back to what GitHub reported minutes ago (this is exactly how an already-closed PR's + // active_review_tracking row got resurrected: a delayed job re-saw `state: "open"` and restarted a review + // pass nothing ever terminalized). Compares GitHub's own `updated_at` against what this row last observed; + // `isStalePayload` is only ever true when BOTH sides have a real timestamp to compare AND the incoming one is + // strictly older -- a sparse payload (no `updated_at`) or a pre-migration/first-ever row (`githubUpdatedAt` + // absent) fails OPEN, applying the write exactly as before this guard existed. Decided in JS (not SQL), up + // front, so EVERY downstream computation below (lastSeenOpenAt, isReadyForReview, the headShaObservedAt clock, + // and this call's own RETURNED record) reasons from the same resolved values instead of the raw payload -- + // otherwise a rejected-as-stale write could still corrupt those derived fields even though state/headSha/ + // mergedAt themselves were protected. Deliberately NOT extended to `draft`/`isDraft` (isReadyForReview's other + // input): no reported failure mode implicates draft-status staleness, and doing so would need its own resolved + // field for no demonstrated benefit. + const incomingGithubUpdatedAt = pr.updated_at ?? null; + const isStalePayload = + incomingGithubUpdatedAt !== null && + existingClaimRow?.githubUpdatedAt != null && + incomingGithubUpdatedAt < existingClaimRow.githubUpdatedAt; + const resolvedState = isStalePayload ? existingClaimRow!.state : pr.state; + const resolvedHeadSha = isStalePayload ? (existingClaimRow!.headSha ?? undefined) : pr.head?.sha; + const resolvedMergedAt = isStalePayload ? (existingClaimRow!.mergedAt ?? undefined) : (pr.merged_at ?? undefined); + // No `?? undefined` fallback on the stale branch (unlike headSha/mergedAt above): isStalePayload's own + // definition already requires existingClaimRow.githubUpdatedAt to be non-null, so that branch is unreachable. + const resolvedGithubUpdatedAt = isStalePayload ? existingClaimRow!.githubUpdatedAt : (incomingGithubUpdatedAt ?? undefined); + const lastSeenOpenAt = resolvedState === "open" ? (options.seenOpenAt ?? syncedAt) : null; const preserveSparseBody = pr.body === undefined && existingClaimRow !== undefined; const existingPayload = preserveSparseBody ? parseJson<{ body?: string | null }>(existingClaimRow.payloadJson, {}) : undefined; const existingBody = existingPayload?.body ?? null; @@ -396,11 +425,12 @@ export async function upsertPullRequestFromGitHub( // headSha change (including the PR's first-ever sync) always restarts the clock; an unchanged headSha keeps // whatever was already stored (including null, which self-heals the instant the PR leaves draft or gets a // fresh commit -- no backfill migration needed, mirrors bodyObservedAt's own non-backfill philosophy). - const isReadyForReview = pr.state === "open" && !(pr.draft ?? pr.isDraft ?? false); - const incomingHeadSha = pr.head?.sha; - const headShaChanged = incomingHeadSha !== undefined && incomingHeadSha !== existingClaimRow?.headSha; + // Reads resolvedState/resolvedHeadSha (not pr.state/pr.head?.sha directly) so a stale, rejected payload can't + // still reset this clock out from under the out-of-order webhook guard above (#webhook-reorder-clobber). + const isReadyForReview = resolvedState === "open" && !(pr.draft ?? pr.isDraft ?? false); + const headShaChanged = resolvedHeadSha !== undefined && resolvedHeadSha !== existingClaimRow?.headSha; const headShaObservedAt = - !isReadyForReview || incomingHeadSha === undefined + !isReadyForReview || resolvedHeadSha === undefined ? (existingClaimRow?.headShaObservedAt ?? null) : headShaChanged || !existingClaimRow?.headShaObservedAt ? syncedAt @@ -412,13 +442,13 @@ export async function upsertPullRequestFromGitHub( repoFullName, number: pr.number, title: pr.title, - state: pr.state, + state: resolvedState, authorLogin: pr.user?.login, authorAssociation: pr.author_association, - headSha: pr.head?.sha, + headSha: resolvedHeadSha, headRef: pr.head?.ref, baseRef: pr.base?.ref, - mergedAt: pr.merged_at ?? undefined, + mergedAt: resolvedMergedAt, htmlUrl: pr.html_url, labelsJson: jsonString(record.labels), linkedIssuesJson, @@ -427,6 +457,7 @@ export async function upsertPullRequestFromGitHub( headShaObservedAt, lastSeenOpenAt, payloadJson: jsonString(payload), + githubUpdatedAt: resolvedGithubUpdatedAt, // GitHub's own PR creation time (see PullRequestRecord.createdAt's doc comment, src/types.ts) -- // set ONLY here, on first insert, and deliberately absent from onConflictDoUpdate's `set` below so // a resync never overwrites it. `?? undefined` falls through to the column's own $defaultFn when a @@ -438,13 +469,13 @@ export async function upsertPullRequestFromGitHub( target: [pullRequests.repoFullName, pullRequests.number], set: { title: pr.title, - state: pr.state, + state: resolvedState, authorLogin: pr.user?.login, authorAssociation: pr.author_association, - headSha: pr.head?.sha, + headSha: resolvedHeadSha, headRef: pr.head?.ref, baseRef: pr.base?.ref, - mergedAt: pr.merged_at ?? undefined, + mergedAt: resolvedMergedAt, htmlUrl: pr.html_url, labelsJson: jsonString(record.labels), linkedIssuesJson, @@ -453,10 +484,11 @@ export async function upsertPullRequestFromGitHub( headShaObservedAt, lastSeenOpenAt, payloadJson: jsonString(payload), + githubUpdatedAt: resolvedGithubUpdatedAt, updatedAt: syncedAt, }, }); - return { ...record, body, linkedIssues, linkedIssueClaimedAt, bodyObservedAt, headShaObservedAt }; + return { ...record, state: resolvedState, headSha: resolvedHeadSha, mergedAt: resolvedMergedAt ?? null, body, linkedIssues, linkedIssueClaimedAt, bodyObservedAt, headShaObservedAt }; } function resolveLinkedIssueClaimedAt( @@ -5841,6 +5873,25 @@ export async function terminalizeActiveReviewTracking( return Number(result.meta.changes ?? 0) > 0; } +/** Rows still `active` and older than `olderThanIso` -- candidates for runActiveReviewReconciliation + * (src/review/active-review-reconciliation.ts) to verify against LIVE GitHub state before terminalizing. Age + * alone is never sufficient to terminalize (a genuinely slow review must not be force-closed); the caller + * confirms each row's PR is actually closed before acting. No index on (status, startedAt) -- a full scan is + * fine at this table's current scale (low thousands of rows); worth an index if that changes materially. */ +export async function listStaleActiveReviewTracking( + env: Env, + olderThanIso: string, +): Promise> { + return getDb(env.DB) + .select({ + repoFullName: activeReviewTracking.repoFullName, + pullNumber: activeReviewTracking.pullNumber, + startedAt: activeReviewTracking.startedAt, + }) + .from(activeReviewTracking) + .where(and(eq(activeReviewTracking.status, "active"), lt(activeReviewTracking.startedAt, olderThanIso))); +} + // Review memory (#2178, data-model slice of #1964). Hard per-repo cap on stored suppression signals — mirrors // rag.ts's MAX_CHUNKS_PER_REPO discipline (bound a repo-controlled, unboundedly-growable store). A repo that // keeps dismissing NEW finding shapes evicts its OLDEST suppression first rather than growing forever. diff --git a/src/db/schema.ts b/src/db/schema.ts index 6565dcd346..1c8006c2a8 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -435,6 +435,14 @@ export const pullRequests = sqliteTable( // (the contributor re-affirmed) refreshes it. loopover-computed (planner-written), omitted from the // GitHub-sync SET clause so a later sync cannot clobber it. screenshotTablePresenceSatisfiedJson: text("screenshot_table_presence_satisfied_json"), + // Out-of-order webhook guard (#webhook-reorder-clobber): GitHub's OWN `updated_at` for this PR, distinct + // from `updatedAt` below (app bookkeeping, stamped on every sync regardless of payload freshness). + // upsertPullRequestFromGitHub compares an incoming payload's `updated_at` against this column before + // applying state/headSha/mergedAt, so a delayed job processing an OLDER webhook (queue backpressure) can + // no longer clobber a newer value a faster job already wrote. NULL (pre-migration rows, or a sparse + // payload that omits `updated_at`) always fails OPEN -- the write applies, exactly like before this + // column existed. + githubUpdatedAt: text("github_updated_at"), createdAt: text("created_at").notNull().$defaultFn(() => nowIso()), updatedAt: text("updated_at").notNull().$defaultFn(() => nowIso()), }, diff --git a/src/env.d.ts b/src/env.d.ts index 1f99863f0e..d0b088300f 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -403,6 +403,13 @@ declare global { * 6-hour backfillRegisteredRepositories freshness window. Default OFF — unset/false means the cron tick * enqueues NO reconciliation job, so the worker is byte-identical to today. */ LOOPOVER_PR_RECONCILIATION?: string; + /** Self-heal (#webhook-reorder-clobber): when truthy, the same short-interval cron as LOOPOVER_PR_RECONCILIATION + * ALSO re-checks every active_review_tracking row stuck in `status: "active"` longer than 15 minutes against + * LIVE (non-cached) GitHub state, and terminalizes the ones GitHub confirms are actually closed — a delayed + * webhook job can otherwise restart tracking for a PR that already closed, orphaning the row forever (see + * src/review/active-review-reconciliation.ts). Default OFF — unset/false means the cron tick enqueues NO + * reconciliation job, so the worker is byte-identical to today. */ + LOOPOVER_ACTIVE_REVIEW_RECONCILIATION?: string; /** Convergence (RAG retrieval): when truthy, the AI reviewer prompt gains a RELEVANT EXISTING CODE / DOCS * section — at review time the codebase vector index is queried for code/docs semantically related to the * PR's changed files (callers, related modules, existing conventions) and appended as additive reference diff --git a/src/index.ts b/src/index.ts index fa0a117cf3..ea3d93899f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,7 @@ import { isRecapEnabled, resolveMaintainerRecapManifestOverride, shouldFireMaint import { isSweepWatchdogEnabled, resolveSweepWatchdogManifestOverride } from "./review/sweep-watchdog"; import { isLoopEscalationSweepEnabled } from "./review/loop-escalation-wire"; import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride } from "./review/pr-reconciliation"; +import { isActiveReviewReconciliationEnabled, resolveActiveReviewReconciliationManifestOverride } from "./review/active-review-reconciliation"; import { isRagEnabled } from "./review/rag-wire"; import { isSelfTuneEnabled } from "./review/selftune-wire"; import { @@ -208,6 +209,18 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): const prReconciliationManifestOverride = await resolvePrReconciliationManifestOverride(env); if (isPrReconciliationEnabled(env, prReconciliationManifestOverride)) jobs.push({ type: "reconcile-open-prs", requestedBy: "schedule" }); } + // Self-heal (flag LOOPOVER_ACTIVE_REVIEW_RECONCILIATION). Same 10-minute cadence as reconcile-open-prs above + // — see isReconciliationWindow. Enable can ALSO be set as code via the loopover self-repo's `.loopover.yml + // activeReviewReconciliation:` block (config-as-code parity, #webhook-reorder-clobber) -- a present manifest + // block wins over the env var; absent, the env var decides exactly as before. Enqueued ONLY when enabled — + // flag-OFF (default) this job is never created, so the cron tick does ZERO new work and the enqueued set is + // byte-identical to today. + if (selfHostedReviews && isReconciliationWindow) { + const activeReviewReconciliationManifestOverride = await resolveActiveReviewReconciliationManifestOverride(env); + if (isActiveReviewReconciliationEnabled(env, activeReviewReconciliationManifestOverride)) { + jobs.push({ type: "reconcile-active-review-tracking", requestedBy: "schedule" }); + } + } if (isHourly) { // Isolation (#experimental-gittensor-plugin): on self-host, refresh-registry both FETCHES from and // PERSISTS the whole upstream gittensor-subnet registry (entrius/gittensor has no server-side filtering, diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index cc6a9402b2..6136c1a38c 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -28,6 +28,7 @@ import { isOpsEnabled, resolveOpsManifestOverride, runOpsAlerts } from "../revie import { isSweepWatchdogEnabled, resolveSweepWatchdogManifestOverride, runSweepLivenessWatchdog } from "../review/sweep-watchdog"; import { isLoopEscalationSweepEnabled, runLoopEscalationSweep } from "../review/loop-escalation-wire"; import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride, runOpenPrReconciliation } from "../review/pr-reconciliation"; +import { isActiveReviewReconciliationEnabled, resolveActiveReviewReconciliationManifestOverride, runActiveReviewReconciliation } from "../review/active-review-reconciliation"; import { isSelfTuneEnabled, runSelfTune } from "../review/selftune-wire"; import { runSelfTuneBreaker } from "../review/outcomes-wire"; import { isRagEnabled } from "../review/rag-wire"; @@ -321,6 +322,15 @@ export async function processJob(env: Env, message: JobMessage): Promise { if (isPrReconciliationEnabled(env, prReconciliationManifestOverride)) await runOpenPrReconciliation(env); } return; + case "reconcile-active-review-tracking": + // Self-heal (flag LOOPOVER_ACTIVE_REVIEW_RECONCILIATION). Defense-in-depth: the cron only ENQUEUES this + // when enabled (env OR manifest), but a stale in-flight job that lands after a flag-flip must still + // no-op, so disabled does zero work here too. Fails safe internally — never throws into the queue. + { + const activeReviewReconciliationManifestOverride = await resolveActiveReviewReconciliationManifestOverride(env); + if (isActiveReviewReconciliationEnabled(env, activeReviewReconciliationManifestOverride)) await runActiveReviewReconciliation(env); + } + return; case "selftune": // Convergence (self-improve / auto-tune, flag LOOPOVER_REVIEW_SELFTUNE). Defense-in-depth: the cron only // ENQUEUES this when the flag is ON, but a stale in-flight job that lands after a flag-flip must still diff --git a/src/review/active-review-reconciliation.ts b/src/review/active-review-reconciliation.ts new file mode 100644 index 0000000000..e7f2e7cffc --- /dev/null +++ b/src/review/active-review-reconciliation.ts @@ -0,0 +1,134 @@ +// Self-heal (flag-gated by LOOPOVER_ACTIVE_REVIEW_RECONCILIATION). An active_review_tracking row can be left +// stuck in `status: "active"` forever when a delayed webhook job (queue backpressure) restarts tracking for a +// PR that has, in the interim, actually already closed/merged on GitHub -- upsertPullRequestFromGitHub's +// out-of-order-webhook guard (#webhook-reorder-clobber, src/db/repositories.ts) closes the WRITE-side half of +// this race, but cannot help a row that got orphaned before that guard existed, or by some other race this +// guard doesn't cover. This module is the READ-side self-heal: periodically re-check every stale `active` row +// against LIVE (non-cached) GitHub state and terminalize the ones GitHub confirms are actually closed. +// +// Default OFF (like every other convergence capability) -- flag-OFF this module is never invoked and the cron +// enqueues no reconciliation job, byte-identical to today. + +import { githubRateLimitAdmissionKeyForToken } from "../github/client"; +import { createInstallationToken } from "../github/app"; +import { fetchLivePullRequestState } from "../github/backfill"; +import { getRepository, listStaleActiveReviewTracking, terminalizeActiveReviewTracking } from "../db/repositories"; +import { loadRepoFocusManifest } from "../signals/focus-manifest-loader"; +import { resolveLoopOverSelfRepoFullName } from "../config/loopover-repo-focus-manifest"; +import { incr } from "../selfhost/metrics"; +import { errorMessage } from "../utils/json"; + +/** How old an `active` row must be before this sweep will even consider it. A review that's merely slow (a big + * diff, a loaded AI backend) is not a bug -- only a row active far longer than any real review pass takes is + * worth spending a live GitHub call to check. */ +export const STALE_ACTIVE_REVIEW_MIN_AGE_MS = 15 * 60_000; + +/** A manifest-sourced enable override (#webhook-reorder-clobber) -- the top-level `activeReviewReconciliation` + * block of the loopover self-repo's `.loopover.yml` (see FocusManifestActiveReviewReconciliationConfig). + * `present: false` means "no override configured", not "disabled" -- the caller falls through to the env var. + * Mirrors PrReconciliationManifestOverride exactly. */ +export type ActiveReviewReconciliationManifestOverride = { present: boolean; enabled: boolean }; + +/** True when the active-review-tracking reconciliation sweep is enabled. Config-as-code (#webhook-reorder- + * clobber): a present top-level `activeReviewReconciliation` manifest block on the loopover self-repo wins + * outright; otherwise falls back to the LOOPOVER_ACTIVE_REVIEW_RECONCILIATION env flag (default OFF). + * Flag-OFF (default) → the caller never invokes the sweep, so the cron enqueues no reconciliation job and the + * queue processor no-ops on a stale in-flight one. */ +export function isActiveReviewReconciliationEnabled( + env: { LOOPOVER_ACTIVE_REVIEW_RECONCILIATION?: string | undefined }, + manifestOverride?: ActiveReviewReconciliationManifestOverride | undefined, +): boolean { + if (manifestOverride?.present) return manifestOverride.enabled; + return /^(1|true|yes|on)$/i.test((env.LOOPOVER_ACTIVE_REVIEW_RECONCILIATION ?? "").trim()); +} + +// Short in-isolate TTL cache for resolveActiveReviewReconciliationManifestOverride, mirroring +// pr-reconciliation.ts / ops-wire.ts / sweep-watchdog.ts: fleet-wide self-repo override, single slot, 60s TTL. +const ACTIVE_REVIEW_RECONCILIATION_MANIFEST_OVERRIDE_CACHE_TTL_MS = 60_000; +let activeReviewReconciliationManifestOverrideCache: { override: ActiveReviewReconciliationManifestOverride; at: number } | null = null; + +/** + * Config-as-code override lookup (#webhook-reorder-clobber): read the top-level `activeReviewReconciliation` + * block off the loopover self-repo's `.loopover.yml`. A manifest load failure degrades to `{ present: false }` + * so a hiccup can never accidentally enable or disable the sweep. + */ +export async function resolveActiveReviewReconciliationManifestOverride(env: Env, nowMs: number = Date.now()): Promise { + const hit = activeReviewReconciliationManifestOverrideCache; + if (hit && nowMs - hit.at < ACTIVE_REVIEW_RECONCILIATION_MANIFEST_OVERRIDE_CACHE_TTL_MS) return hit.override; + try { + const manifest = await loadRepoFocusManifest(env, resolveLoopOverSelfRepoFullName(env)); + const config = manifest.activeReviewReconciliation; + const override = { present: config.present, enabled: config.enabled }; + activeReviewReconciliationManifestOverrideCache = { override, at: nowMs }; + return override; + } catch (error) { + console.warn(JSON.stringify({ event: "active_review_reconciliation_manifest_override_error", message: errorMessage(error).slice(0, 200) })); + const override = { present: false, enabled: false }; + activeReviewReconciliationManifestOverrideCache = { override, at: nowMs }; + return override; + } +} + +/** Test-only: clears the cached override, mirroring clearPrReconciliationManifestOverrideCacheForTest. */ +export function clearActiveReviewReconciliationManifestOverrideCacheForTest(): void { + activeReviewReconciliationManifestOverrideCache = null; +} + +export interface ReconciledActiveReview { + repoFullName: string; + pullNumber: number; +} + +/** + * The reconciliation scan, run on the cron tick. FAILS SAFE: a per-row error is logged and the scan continues; + * a top-level error is swallowed (this is best-effort self-heal, never a reason to fail the queue). Only + * terminalizes a row when a LIVE (non-cached) GitHub read confirms the PR is no longer open -- never on age + * alone, so a genuinely slow review is never force-closed; a repo with no installation, or a live check that + * itself fails, leaves the row untouched for the next tick to retry. + * + * Caller MUST gate this on {@link isActiveReviewReconciliationEnabled} -- it is invoked only from the flag-ON + * cron path, so flag-OFF this function is never reached and the cron does zero new work. + */ +export async function runActiveReviewReconciliation(env: Env, nowMs: number = Date.now()): Promise { + const reconciled: ReconciledActiveReview[] = []; + try { + const cutoff = new Date(nowMs - STALE_ACTIVE_REVIEW_MIN_AGE_MS).toISOString(); + const staleRows = await listStaleActiveReviewTracking(env, cutoff); + for (const row of staleRows) { + try { + const repo = await getRepository(env, row.repoFullName); + if (!repo || typeof repo.installationId !== "number") continue; + const token = (await createInstallationToken(env, repo.installationId).catch(() => undefined)) ?? env.GITHUB_PUBLIC_TOKEN; + const admissionKey = githubRateLimitAdmissionKeyForToken(env, token, repo.installationId); + const liveState = await fetchLivePullRequestState(env, row.repoFullName, row.pullNumber, token, admissionKey); + if (liveState !== "closed") continue; // still open, or the live check itself failed -- leave it for the next tick + const changed = await terminalizeActiveReviewTracking(env, row.repoFullName, row.pullNumber); + if (!changed) continue; // a concurrent pass already terminalized (or restarted) this row first + reconciled.push({ repoFullName: row.repoFullName, pullNumber: row.pullNumber }); + incr("loopover_active_review_reconciliation_terminalized_total", { repo: row.repoFullName }); + console.error( + JSON.stringify({ + level: "error", + event: "active_review_reconciliation_orphan_terminalized", + repository: row.repoFullName, + pullNumber: row.pullNumber, + startedAt: row.startedAt, + }), + ); + } catch (error) { + console.error( + JSON.stringify({ + level: "error", + event: "active_review_reconciliation_row_error", + repository: row.repoFullName, + pullNumber: row.pullNumber, + message: errorMessage(error).slice(0, 200), + }), + ); + } + } + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "active_review_reconciliation_error", message: errorMessage(error).slice(0, 200) })); + } + return reconciled; +} diff --git a/src/selfhost/metrics.ts b/src/selfhost/metrics.ts index e2d09403c3..044d04c7a9 100644 --- a/src/selfhost/metrics.ts +++ b/src/selfhost/metrics.ts @@ -162,6 +162,7 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [ ["loopover_linked_issue_satisfaction_cache_hit_total", { help: "Linked-issue satisfaction assessment cache hits.", type: "counter" }], ["loopover_linked_issue_satisfaction_cache_miss_total", { help: "Linked-issue satisfaction assessment cache misses.", type: "counter" }], ["loopover_linked_issue_satisfaction_cache_write_error_total", { help: "Linked-issue satisfaction assessment cache write errors.", type: "counter" }], + ["loopover_active_review_reconciliation_terminalized_total", { help: "Orphaned active_review_tracking rows terminalized after a live GitHub check confirmed the PR is closed, by repo.", type: "counter" }], ["loopover_open_pr_reconciliation_missing_total", { help: "Open PRs found missing from local tracking during reconciliation, by repo.", type: "counter" }], ["loopover_orb_relay_malformed_events_total", { help: "Orb relay batch entries dropped for missing/mistyped required fields (deliveryId/eventName/rawBody).", type: "counter" }], ["loopover_orb_relay_register_total", { help: "Orb relay registration attempts, by mode and result (registered/recovered/failed).", type: "counter" }], diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index 007d529f3c..90965f7ed4 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -123,6 +123,11 @@ const GITHUB_BUDGET_BACKGROUND_TYPES = new Set([ // fetch to entrius/gittensor, not the GitHub REST API), this genuinely draws down the shared installation's // REST budget and must yield alongside every other budget consumer here. "sync-brokered-installed-repos", + // runActiveReviewReconciliation (#webhook-reorder-clobber) makes one live, non-cached `GET /pulls/{n}` REST + // call per stale active_review_tracking row it finds. Flag-gated OFF by default (LOOPOVER_ACTIVE_REVIEW_ + // RECONCILIATION) -- registered here up front so enabling it never bypasses the shared budget the way + // reconcile-open-prs originally did before #4505/#4506 closed that gap. + "reconcile-active-review-tracking", ]); const PRIORITY_BY_TYPE = new Map([ ["agent-regate-pr", AGENT_REGATE_PRIORITY], diff --git a/src/signals/focus-manifest-loader.ts b/src/signals/focus-manifest-loader.ts index 20d88c9ff5..f4297cfc38 100644 --- a/src/signals/focus-manifest-loader.ts +++ b/src/signals/focus-manifest-loader.ts @@ -2,7 +2,7 @@ import { listSignalSnapshots, persistSignalSnapshot } from "../db/repositories"; import { mapWithConcurrency } from "../queue/map-with-concurrency"; import type { JsonValue } from "../types"; import { nowIso } from "../utils/json"; -import { contentLaneConfigToJson, experimentalConfigToJson, featuresConfigToJson, gateConfigToJson, MAX_FOCUS_MANIFEST_BYTES, parseFocusManifest, parseFocusManifestContent, repoDocGenerationConfigToJson, reviewConfigToJson, reviewRecapConfigToJson, maintainerRecapConfigToJson, opsConfigToJson, publicStatsConfigToJson, fairnessAnalyticsConfigToJson, draftFlowConfigToJson, upstreamDriftIssuesConfigToJson, sweepWatchdogConfigToJson, prReconciliationConfigToJson, federatedIntelligenceConfigToJson, settingsOverrideToJson, type FocusManifest, type FocusManifestSource, type RepoReviewContext } from "./focus-manifest"; +import { contentLaneConfigToJson, experimentalConfigToJson, featuresConfigToJson, gateConfigToJson, MAX_FOCUS_MANIFEST_BYTES, parseFocusManifest, parseFocusManifestContent, repoDocGenerationConfigToJson, reviewConfigToJson, reviewRecapConfigToJson, maintainerRecapConfigToJson, opsConfigToJson, publicStatsConfigToJson, fairnessAnalyticsConfigToJson, draftFlowConfigToJson, upstreamDriftIssuesConfigToJson, sweepWatchdogConfigToJson, prReconciliationConfigToJson, activeReviewReconciliationConfigToJson, federatedIntelligenceConfigToJson, settingsOverrideToJson, type FocusManifest, type FocusManifestSource, type RepoReviewContext } from "./focus-manifest"; import { LOOPOVER_REPO_FOCUS_MANIFEST_YAML, resolveLoopOverSelfRepoFullName } from "../config/loopover-repo-focus-manifest"; import type { LocalManifestLoadResult } from "../selfhost/private-config"; @@ -334,6 +334,7 @@ function manifestToJson(manifest: FocusManifest): Record { upstreamDriftIssues: upstreamDriftIssuesConfigToJson(manifest.upstreamDriftIssues), sweepWatchdog: sweepWatchdogConfigToJson(manifest.sweepWatchdog), prReconciliation: prReconciliationConfigToJson(manifest.prReconciliation), + activeReviewReconciliation: activeReviewReconciliationConfigToJson(manifest.activeReviewReconciliation), federatedIntelligence: federatedIntelligenceConfigToJson(manifest.federatedIntelligence), }; } diff --git a/src/signals/focus-manifest.ts b/src/signals/focus-manifest.ts index 6381238822..4cb566fba4 100644 --- a/src/signals/focus-manifest.ts +++ b/src/signals/focus-manifest.ts @@ -42,6 +42,7 @@ export { upstreamDriftIssuesConfigToJson, sweepWatchdogConfigToJson, prReconciliationConfigToJson, + activeReviewReconciliationConfigToJson, federatedIntelligenceConfigToJson, FEDERATED_COLLECTOR_MODES, settingsOverrideToJson, @@ -77,6 +78,7 @@ export { type FocusManifestUpstreamDriftIssuesConfig, type FocusManifestSweepWatchdogConfig, type FocusManifestPrReconciliationConfig, + type FocusManifestActiveReviewReconciliationConfig, type FocusManifestFederatedIntelligenceConfig, type FederatedCollectorMode, type FocusManifestSettings, diff --git a/src/types.ts b/src/types.ts index 6504fa721c..bd8e3cc5fb 100644 --- a/src/types.ts +++ b/src/types.ts @@ -230,6 +230,16 @@ export type JobMessage = type: "reconcile-open-prs"; requestedBy: "schedule" | "api" | "test"; } + | { + // Self-heal (flag-gated by LOOPOVER_ACTIVE_REVIEW_RECONCILIATION). A delayed webhook job can restart + // active_review_tracking for a PR that's already closed on GitHub, orphaning the row in `status: "active"` + // forever (see src/review/active-review-reconciliation.ts's header comment). Re-checks every stale + // `active` row against LIVE (non-cached) GitHub state and terminalizes the ones confirmed closed. + // Enqueued on the same 10-minute reconciliation cadence as reconcile-open-prs (index.ts) ONLY when the + // flag is ON, so flag-OFF this job never exists. + type: "reconcile-active-review-tracking"; + requestedBy: "schedule" | "api" | "test"; + } | { // Convergence (self-improve / auto-tune, flag-gated by LOOPOVER_REVIEW_SELFTUNE). Run the ported // self-improvement loop over loopover's review-outcome data — compute tuning recommendations, diff --git a/test/unit/active-review-reconciliation.test.ts b/test/unit/active-review-reconciliation.test.ts new file mode 100644 index 0000000000..990d182f01 --- /dev/null +++ b/test/unit/active-review-reconciliation.test.ts @@ -0,0 +1,220 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + clearActiveReviewReconciliationManifestOverrideCacheForTest, + isActiveReviewReconciliationEnabled, + resolveActiveReviewReconciliationManifestOverride, + runActiveReviewReconciliation, + STALE_ACTIVE_REVIEW_MIN_AGE_MS, +} from "../../src/review/active-review-reconciliation"; +import { hasActiveReviewForHeadSha, startActiveReviewTracking, upsertRepositoryFromGitHub } from "../../src/db/repositories"; +import * as repositoriesModule from "../../src/db/repositories"; +import * as backfillModule from "../../src/github/backfill"; +import { counterValue, resetMetrics } from "../../src/selfhost/metrics"; +import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader"; +import { createTestEnv } from "../helpers/d1"; + +const SELF_REPO = "JSONbored/loopover"; + +describe("isActiveReviewReconciliationEnabled — default OFF, truthy convention", () => { + it("matches the codebase's shared truthy-string convention", () => { + for (const off of [undefined, "", "false", "no", "0", "off"]) expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: off })).toBe(false); + for (const on of ["1", "true", "yes", "on", "TRUE", "On"]) expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: on })).toBe(true); + }); + + it("whitespace-padded truthy values still activate (matches isRagEnabled/isPrReconciliationEnabled)", () => { + expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "true\n" })).toBe(true); + expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: " 1 " })).toBe(true); + }); + + it("a present manifest override wins outright over the env flag, in both directions (#webhook-reorder-clobber)", () => { + expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "false" }, { present: true, enabled: true })).toBe(true); + expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "true" }, { present: true, enabled: false })).toBe(false); + }); + + it("falls back to the env flag when the manifest override is not present", () => { + expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "true" }, { present: false, enabled: false })).toBe(true); + expect(isActiveReviewReconciliationEnabled({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "false" }, undefined)).toBe(false); + }); +}); + +describe("resolveActiveReviewReconciliationManifestOverride — config-as-code lookup (#webhook-reorder-clobber)", () => { + beforeEach(() => { + clearActiveReviewReconciliationManifestOverrideCacheForTest(); + }); + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it("returns the self-repo's configured activeReviewReconciliation block when present", async () => { + const env = createTestEnv({ LOOPOVER_DRIFT_ISSUE_REPO: SELF_REPO }); + await upsertRepoFocusManifest(env, SELF_REPO, { activeReviewReconciliation: { enabled: true } }); + + expect(await resolveActiveReviewReconciliationManifestOverride(env)).toEqual({ present: true, enabled: true }); + }); + + it("returns present: false when the self-repo has no activeReviewReconciliation block configured", async () => { + const env = createTestEnv({ LOOPOVER_DRIFT_ISSUE_REPO: SELF_REPO }); + await upsertRepoFocusManifest(env, SELF_REPO, { wantedPaths: ["src/"] }); + + expect(await resolveActiveReviewReconciliationManifestOverride(env)).toEqual({ present: false, enabled: false }); + }); + + it("degrades to present: false (never throws) when the manifest load itself fails", async () => { + const env = createTestEnv({ LOOPOVER_DRIFT_ISSUE_REPO: SELF_REPO }); + const realPrepare = env.DB.prepare.bind(env.DB); + env.DB.prepare = ((sql: string) => { + if (/"signal_snapshots"|signal_snapshots/i.test(sql)) throw new Error("poisoned query"); + return realPrepare(sql); + }) as typeof env.DB.prepare; + vi.stubGlobal("fetch", async () => { + throw new Error("network down"); + }); + const warnings = vi.spyOn(console, "warn").mockImplementation(() => {}); + + expect(await resolveActiveReviewReconciliationManifestOverride(env)).toEqual({ present: false, enabled: false }); + expect(warnings.mock.calls.map((c) => String(c[0])).some((line) => line.includes("active_review_reconciliation_manifest_override_error"))).toBe(true); + }); + + it("within the 60s TTL, reuses the cached override instead of re-reading the manifest", async () => { + const env = createTestEnv({ LOOPOVER_DRIFT_ISSUE_REPO: SELF_REPO }); + await upsertRepoFocusManifest(env, SELF_REPO, { activeReviewReconciliation: { enabled: true } }); + const t0 = Date.parse("2026-07-21T00:00:00Z"); + expect(await resolveActiveReviewReconciliationManifestOverride(env, t0)).toEqual({ present: true, enabled: true }); + + env.DB.prepare = (() => { + throw new Error("should not be queried on a cache hit"); + }) as typeof env.DB.prepare; + expect(await resolveActiveReviewReconciliationManifestOverride(env, t0 + 30_000)).toEqual({ present: true, enabled: true }); + }); + + it("re-reads the manifest once the 60s TTL has elapsed", async () => { + const env = createTestEnv({ LOOPOVER_DRIFT_ISSUE_REPO: SELF_REPO }); + await upsertRepoFocusManifest(env, SELF_REPO, { activeReviewReconciliation: { enabled: true } }); + const t0 = Date.parse("2026-07-21T00:00:00Z"); + expect(await resolveActiveReviewReconciliationManifestOverride(env, t0)).toEqual({ present: true, enabled: true }); + + await upsertRepoFocusManifest(env, SELF_REPO, { activeReviewReconciliation: { enabled: false } }); + expect(await resolveActiveReviewReconciliationManifestOverride(env, t0 + 60_001)).toEqual({ present: true, enabled: false }); + }); +}); + +describe("runActiveReviewReconciliation (#webhook-reorder-clobber)", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + async function seedStaleActiveRow(env: Env, repoFullName: string, pullNumber: number, installationId: number, ageMs: number) { + await upsertRepositoryFromGitHub(env, { name: repoFullName.split("/")[1]!, full_name: repoFullName, private: false, owner: { login: repoFullName.split("/")[0]! } }, installationId); + await startActiveReviewTracking(env, { repoFullName, pullNumber, headSha: "sha1", deliveryId: "delivery-1" }); + // Backdate startedAt directly -- startActiveReviewTracking always stamps "now". + await env.DB.prepare("update active_review_tracking set started_at = ? where repo_full_name = ? and pull_number = ?") + .bind(new Date(Date.now() - ageMs).toISOString(), repoFullName, pullNumber) + .run(); + } + + it("terminalizes a stale row a LIVE GitHub check confirms is closed", async () => { + resetMetrics(); + const env = createTestEnv(); + await seedStaleActiveRow(env, "owner/repo", 1, 9500, STALE_ACTIVE_REVIEW_MIN_AGE_MS + 60_000); + vi.spyOn(backfillModule, "fetchLivePullRequestState").mockResolvedValueOnce("closed"); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([{ repoFullName: "owner/repo", pullNumber: 1 }]); + expect(await hasActiveReviewForHeadSha(env, "owner/repo", 1, "sha1")).toBe(false); + expect(counterValue("loopover_active_review_reconciliation_terminalized_total", { repo: "owner/repo" })).toBe(1); + const logged = errors.mock.calls.map((c) => String(c[0])).find((line) => line.includes("active_review_reconciliation_orphan_terminalized")); + expect(logged).toBeDefined(); + expect(JSON.parse(logged!)).toMatchObject({ level: "error", event: "active_review_reconciliation_orphan_terminalized", repository: "owner/repo", pullNumber: 1 }); + }); + + it("leaves a stale row alone when the LIVE check says the PR is still open", async () => { + const env = createTestEnv(); + await seedStaleActiveRow(env, "owner/repo", 2, 9501, STALE_ACTIVE_REVIEW_MIN_AGE_MS + 60_000); + vi.spyOn(backfillModule, "fetchLivePullRequestState").mockResolvedValueOnce("open"); + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([]); + expect(await hasActiveReviewForHeadSha(env, "owner/repo", 2, "sha1")).toBe(true); + }); + + it("leaves a stale row alone when the LIVE check itself fails (undefined) -- never force-closes on an inconclusive read", async () => { + const env = createTestEnv(); + await seedStaleActiveRow(env, "owner/repo", 3, 9502, STALE_ACTIVE_REVIEW_MIN_AGE_MS + 60_000); + vi.spyOn(backfillModule, "fetchLivePullRequestState").mockResolvedValueOnce(undefined); + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([]); + expect(await hasActiveReviewForHeadSha(env, "owner/repo", 3, "sha1")).toBe(true); + }); + + it("never considers a row younger than the staleness cutoff -- a genuinely in-flight review is not a candidate", async () => { + const env = createTestEnv(); + await seedStaleActiveRow(env, "owner/repo", 4, 9503, STALE_ACTIVE_REVIEW_MIN_AGE_MS - 60_000); + const liveSpy = vi.spyOn(backfillModule, "fetchLivePullRequestState"); + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([]); + expect(liveSpy).not.toHaveBeenCalled(); + }); + + it("skips a repo with no installation -- never spends a live GitHub call it couldn't authenticate anyway", async () => { + const env = createTestEnv(); + await upsertRepositoryFromGitHub(env, { name: "no-install", full_name: "owner/no-install", private: false, owner: { login: "owner" } }); // no installation id + await startActiveReviewTracking(env, { repoFullName: "owner/no-install", pullNumber: 5, headSha: "sha1", deliveryId: "delivery-1" }); + await env.DB.prepare("update active_review_tracking set started_at = ? where repo_full_name = ? and pull_number = ?") + .bind(new Date(Date.now() - STALE_ACTIVE_REVIEW_MIN_AGE_MS - 60_000).toISOString(), "owner/no-install", 5) + .run(); + const liveSpy = vi.spyOn(backfillModule, "fetchLivePullRequestState"); + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([]); + expect(liveSpy).not.toHaveBeenCalled(); + }); + + it("fails safe per-row: an error on one row is logged and the scan continues to the next row", async () => { + const env = createTestEnv(); + await seedStaleActiveRow(env, "owner/erroring-repo", 6, 9504, STALE_ACTIVE_REVIEW_MIN_AGE_MS + 60_000); + await seedStaleActiveRow(env, "owner/ok-repo", 7, 9505, STALE_ACTIVE_REVIEW_MIN_AGE_MS + 60_000); + const realGetRepository = repositoriesModule.getRepository; + vi.spyOn(repositoriesModule, "getRepository").mockImplementation(async (envArg, fullName) => { + if (fullName === "owner/erroring-repo") throw new Error("D1 read error"); + return realGetRepository(envArg, fullName); + }); + vi.spyOn(backfillModule, "fetchLivePullRequestState").mockResolvedValueOnce("closed"); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([{ repoFullName: "owner/ok-repo", pullNumber: 7 }]); // erroring-repo's row is skipped, not fatal + expect(errors.mock.calls.some((call) => String(call[0]).includes("active_review_reconciliation_row_error") && String(call[0]).includes("owner/erroring-repo"))).toBe(true); + }); + + it("fails safe at the top level: a total scan failure is logged and returns an empty result instead of throwing", async () => { + const env = createTestEnv(); + vi.spyOn(repositoriesModule, "listStaleActiveReviewTracking").mockRejectedValueOnce(new Error("D1 unavailable")); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(runActiveReviewReconciliation(env)).resolves.toEqual([]); + + expect(errors.mock.calls.some((call) => String(call[0]).includes("active_review_reconciliation_error"))).toBe(true); + }); + + it("a concurrent terminalize race (row already terminal by the time this pass writes) is not double-reported", async () => { + resetMetrics(); + const env = createTestEnv(); + await seedStaleActiveRow(env, "owner/repo", 8, 9506, STALE_ACTIVE_REVIEW_MIN_AGE_MS + 60_000); + vi.spyOn(backfillModule, "fetchLivePullRequestState").mockResolvedValueOnce("closed"); + vi.spyOn(repositoriesModule, "terminalizeActiveReviewTracking").mockResolvedValueOnce(false); // another pass won the race + + const reconciled = await runActiveReviewReconciliation(env); + + expect(reconciled).toEqual([]); + expect(counterValue("loopover_active_review_reconciliation_terminalized_total", { repo: "owner/repo" })).toBe(0); + }); +}); diff --git a/test/unit/db-parsers.test.ts b/test/unit/db-parsers.test.ts index 828051480a..1c20c37953 100644 --- a/test/unit/db-parsers.test.ts +++ b/test/unit/db-parsers.test.ts @@ -627,6 +627,133 @@ describe("database row parser hardening", () => { expect(closed.headShaObservedAt).toBe(first.headShaObservedAt); }); + describe("out-of-order webhook guard (#webhook-reorder-clobber)", () => { + it("REGRESSION: a delayed OLDER webhook cannot clobber state/headSha/mergedAt a newer one already wrote", async () => { + const env = createTestEnv(); + // The real close lands first (job order, not necessarily arrival order). + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 30, title: "PR", state: "closed", user: { login: "bob" }, head: { sha: "a1" }, merged_at: null, labels: [], updated_at: "2026-07-21T12:21:17.000Z", + }); + // A job for an OLDER event (e.g. review_requested, queued minutes earlier) finally dequeues after + // sitting behind queue backpressure and processes its own stale, still-"open" embedded snapshot. + const stale = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 30, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a0" }, merged_at: null, labels: [], updated_at: "2026-07-21T12:15:52.000Z", + }); + + // The function's OWN return value reflects what was actually persisted, not the stale incoming payload -- + // this is what handlePullRequestWebhookEvent's closed-check (and every other reader of this call's result) + // sees, so it must agree with the DB or the delayed job would still act on stale data in-process. + expect(stale.state).toBe("closed"); + expect(stale.headSha).toBe("a1"); + const stored = await getPullRequest(env, "owner/repo", 30); + expect(stored?.state).toBe("closed"); + expect(stored?.headSha).toBe("a1"); + }); + + it("a NEWER webhook (later updated_at) still applies normally", async () => { + const env = createTestEnv(); + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 31, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + const fresh = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 31, title: "PR", state: "closed", user: { login: "bob" }, head: { sha: "a2" }, labels: [], updated_at: "2026-07-21T12:05:00.000Z", + }); + + expect(fresh.state).toBe("closed"); + expect(fresh.headSha).toBe("a2"); + }); + + it("fails OPEN when the incoming payload has no updated_at (a sparse webhook sub-object) -- applies the write exactly as before this guard existed", async () => { + const env = createTestEnv(); + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 32, title: "PR", state: "closed", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:21:17.000Z", + }); + const sparse = await upsertPullRequestFromGitHub(env, "owner/repo", { number: 32, title: "PR reopened by sparse event", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [] }); + + expect(sparse.state).toBe("open"); + }); + + it("fails OPEN for a pre-migration row with no stored githubUpdatedAt yet -- applies the write exactly as before this guard existed", async () => { + const env = createTestEnv(); + // Simulates a row created before this migration: no updated_at supplied on the FIRST sync either. + await upsertPullRequestFromGitHub(env, "owner/repo", { number: 33, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [] }); + const synced = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 33, title: "PR", state: "closed", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + + expect(synced.state).toBe("closed"); // nothing stored to compare against -> guard can't prove staleness, applies the write + }); + + it("mergedAt is guarded the same way -- a stale payload cannot regress a real merge back to null", async () => { + const env = createTestEnv(); + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 34, title: "PR", state: "closed", user: { login: "bob" }, head: { sha: "a1" }, merged_at: "2026-07-21T12:21:17.000Z", labels: [], updated_at: "2026-07-21T12:21:17.000Z", + }); + const stale = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 34, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, merged_at: null, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + + expect(stale.mergedAt).toBe("2026-07-21T12:21:17.000Z"); + const stored = await getPullRequest(env, "owner/repo", 34); + expect(stored?.mergedAt).toBe("2026-07-21T12:21:17.000Z"); + }); + + it("an EQUAL updated_at (a true redelivery/retry of the same event) is not treated as stale", async () => { + const env = createTestEnv(); + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 35, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + const redelivered = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 35, title: "PR redelivered", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + + expect(redelivered.title).toBe("PR redelivered"); // non-guarded fields still apply on an equal-timestamp resync + }); + + it("REGRESSION: a stale payload's rejected head SHA does not reset the review-latency clock (headShaObservedAt) -- the clock must follow the RESOLVED head, not the raw payload's", async () => { + const env = createTestEnv(); + const first = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 36, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + expect(typeof first.headShaObservedAt).toBe("string"); + // A genuine fresh commit lands and is processed promptly -- the real current head is now a2. + const pushed = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 36, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a2" }, labels: [], updated_at: "2026-07-21T12:05:00.000Z", + }); + expect(pushed.headSha).toBe("a2"); + expect(pushed.headShaObservedAt).not.toBe(first.headShaObservedAt); + + // A delayed job for the OLDER `review_requested` event (before the push) finally dequeues, still carrying + // the STALE head "a1". Without resolvedHeadSha driving headShaChanged, this would look like "a2 -> a1", a + // head change, and wrongly reset the clock even though the stored head never actually moved off "a2". + const stale = await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 36, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:01:00.000Z", + }); + + expect(stale.headSha).toBe("a2"); // still protected by the out-of-order guard + expect(stale.headShaObservedAt).toBe(pushed.headShaObservedAt); // clock untouched by the stale rejection + }); + + it("REGRESSION: a stale payload claiming state: open does not re-stamp lastSeenOpenAt once the PR has actually closed", async () => { + const env = createTestEnv(); + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 37, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:00:00.000Z", + }); + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 37, title: "PR", state: "closed", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:05:00.000Z", + }); + // A delayed job for an older still-"open" event dequeues after the real close. + await upsertPullRequestFromGitHub(env, "owner/repo", { + number: 37, title: "PR", state: "open", user: { login: "bob" }, head: { sha: "a1" }, labels: [], updated_at: "2026-07-21T12:01:00.000Z", + }); + + const row = await env.DB.prepare("select last_seen_open_at from pull_requests where repo_full_name = ? and number = ?") + .bind("owner/repo", 37) + .first<{ last_seen_open_at: string | null }>(); + expect(row?.last_seen_open_at).toBeNull(); // not re-stamped as "seen open" by the stale rejection + }); + }); + it("countRecentDeadLetters counts github_app.dlq_dead_lettered audits since a cutoff, independent of any ops flag (#1276)", async () => { const env = createTestEnv(); await recordAuditEvent(env, { eventType: "github_app.dlq_dead_lettered", actor: "loopover", targetKey: "dlq:github-webhook:a", outcome: "error", createdAt: "2026-06-24T10:00:00.000Z" }); diff --git a/test/unit/db-persistence.test.ts b/test/unit/db-persistence.test.ts index d6cf1afdf7..564314db99 100644 --- a/test/unit/db-persistence.test.ts +++ b/test/unit/db-persistence.test.ts @@ -10,6 +10,7 @@ import { listLatestSignalSnapshotsForTargets, listRepoPullRequestFilePaths, listSignalSnapshots, + listStaleActiveReviewTracking, persistBountyLifecycleEvent, persistRepoGithubTotalsSnapshot, persistSignalSnapshot, @@ -442,6 +443,43 @@ describe("active-review tracking (#review-evasion-protection)", () => { }); }); + describe("listStaleActiveReviewTracking (#webhook-reorder-clobber)", () => { + it("returns nothing when no row exists at all", async () => { + const env = createTestEnv(); + expect(await listStaleActiveReviewTracking(env, "2026-07-21T13:00:00.000Z")).toEqual([]); + }); + + it("returns an active row older than the cutoff", async () => { + const env = createTestEnv(); + await startActiveReviewTracking(env, { repoFullName: "owner/repo", pullNumber: 1, headSha: "sha1", deliveryId: "delivery-1" }); + const row = await rawRow(env, "owner/repo", 1); + + const stale = await listStaleActiveReviewTracking(env, new Date(Date.parse(row!.started_at) + 1000).toISOString()); + + expect(stale).toEqual([{ repoFullName: "owner/repo", pullNumber: 1, startedAt: row!.started_at }]); + }); + + it("excludes a row NEWER than the cutoff -- a genuinely fresh review is never a candidate", async () => { + const env = createTestEnv(); + await startActiveReviewTracking(env, { repoFullName: "owner/repo", pullNumber: 1, headSha: "sha1", deliveryId: "delivery-1" }); + const row = await rawRow(env, "owner/repo", 1); + + const stale = await listStaleActiveReviewTracking(env, new Date(Date.parse(row!.started_at) - 1000).toISOString()); + + expect(stale).toEqual([]); + }); + + it("excludes a TERMINAL row even when it's old -- only status='active' is a candidate", async () => { + const env = createTestEnv(); + await startActiveReviewTracking(env, { repoFullName: "owner/repo", pullNumber: 1, headSha: "sha1", deliveryId: "delivery-1" }); + await terminalizeActiveReviewTracking(env, "owner/repo", 1); + + const stale = await listStaleActiveReviewTracking(env, "2099-01-01T00:00:00.000Z"); + + expect(stale).toEqual([]); + }); + }); + // signal_snapshots' "latest" tiebreak (investigated per an out-of-scope-flagged follow-up): generatedAt is // millisecond-precision, so two writes for the same (signalType, targetKey) within one millisecond tie: SQLite // itself makes no guarantee about tie order ("the order ... is undefined" -- sqlite.org/lang_select.html), so diff --git a/test/unit/focus-manifest-validation.test.ts b/test/unit/focus-manifest-validation.test.ts index 01b3863c3e..d82e8ebad9 100644 --- a/test/unit/focus-manifest-validation.test.ts +++ b/test/unit/focus-manifest-validation.test.ts @@ -114,6 +114,8 @@ sweepWatchdog: enabled: true prReconciliation: enabled: false +activeReviewReconciliation: + enabled: true `, }); expect(result.status).toBe("ok"); @@ -134,10 +136,11 @@ prReconciliation: upstreamDriftIssues: { enabled: false }, sweepWatchdog: { enabled: true }, prReconciliation: { enabled: false }, + activeReviewReconciliation: { enabled: true }, }); }); - it("omits maintainerRecap/ops/publicStats/draftFlow/upstreamDriftIssues/sweepWatchdog/prReconciliation/federatedIntelligence from the normalized output when none are configured", () => { + it("omits maintainerRecap/ops/publicStats/draftFlow/upstreamDriftIssues/sweepWatchdog/prReconciliation/activeReviewReconciliation/federatedIntelligence from the normalized output when none are configured", () => { const result = buildFocusManifestValidation({ content: "wantedPaths: [src/]\n" }); expect(result.normalized).not.toHaveProperty("maintainerRecap"); expect(result.normalized).not.toHaveProperty("ops"); @@ -146,9 +149,16 @@ prReconciliation: expect(result.normalized).not.toHaveProperty("upstreamDriftIssues"); expect(result.normalized).not.toHaveProperty("sweepWatchdog"); expect(result.normalized).not.toHaveProperty("prReconciliation"); + expect(result.normalized).not.toHaveProperty("activeReviewReconciliation"); expect(result.normalized).not.toHaveProperty("federatedIntelligence"); }); + it("includes a configured activeReviewReconciliation block in the normalized settings-preview output (#webhook-reorder-clobber)", () => { + const result = buildFocusManifestValidation({ content: "activeReviewReconciliation:\n enabled: true\n" }); + expect(result.warnings).toEqual([]); + expect(result.normalized).toMatchObject({ activeReviewReconciliation: { enabled: true } }); + }); + it("includes a configured federatedIntelligence block in the normalized settings-preview output (#6998)", () => { const result = buildFocusManifestValidation({ content: "federatedIntelligence:\n enabled: true\n" }); expect(result.warnings).toEqual([]); diff --git a/test/unit/focus-manifest.test.ts b/test/unit/focus-manifest.test.ts index 6af01fa444..573c57690a 100644 --- a/test/unit/focus-manifest.test.ts +++ b/test/unit/focus-manifest.test.ts @@ -47,6 +47,7 @@ import { upstreamDriftIssuesConfigToJson, sweepWatchdogConfigToJson, prReconciliationConfigToJson, + activeReviewReconciliationConfigToJson, federatedIntelligenceConfigToJson, settingsOverrideToJson, type FocusManifest, @@ -956,6 +957,7 @@ describe("compileFocusManifestPolicy", () => { upstreamDriftIssues: { present: false, enabled: false }, sweepWatchdog: { present: false, enabled: false, staleAfterMinutes: null }, prReconciliation: { present: false, enabled: false }, + activeReviewReconciliation: { present: false, enabled: false }, federatedIntelligence: { present: false, enabled: false, collectorUrl: null, collectorMode: null, peerKeys: [] }, warnings: [], }); @@ -2269,6 +2271,54 @@ describe("parseFocusManifest gate config", () => { }); }); + describe("activeReviewReconciliation: (#webhook-reorder-clobber, active-review-tracking reconciliation sweep config-as-code override)", () => { + it("defaults to fully disabled/absent when the key is omitted, and does not make the manifest present on its own", () => { + const m = parseFocusManifest({}); + expect(m.activeReviewReconciliation).toEqual({ present: false, enabled: false }); + expect(m.present).toBe(false); + }); + + it("treats an explicit null the same as an omitted key", () => { + expect(parseFocusManifest({ activeReviewReconciliation: null }).activeReviewReconciliation).toEqual({ present: false, enabled: false }); + }); + + it("warns and falls back to the default when the value is a non-mapping type (string or array)", () => { + const asString = parseFocusManifest({ activeReviewReconciliation: "nope" as never }); + expect(asString.activeReviewReconciliation.present).toBe(false); + expect(asString.warnings.some((w) => /"activeReviewReconciliation" must be a mapping/.test(w))).toBe(true); + const asArray = parseFocusManifest({ activeReviewReconciliation: ["nope"] as never }); + expect(asArray.activeReviewReconciliation.present).toBe(false); + expect(asArray.warnings.some((w) => /"activeReviewReconciliation" must be a mapping/.test(w))).toBe(true); + }); + + it("parses enabled: true, making the manifest present", () => { + const m = parseFocusManifest({ activeReviewReconciliation: { enabled: true } }); + expect(m.activeReviewReconciliation).toEqual({ present: true, enabled: true }); + expect(m.present).toBe(true); + }); + + it("parses enabled: false explicitly, still marking the manifest present (present is a real override, off)", () => { + const m = parseFocusManifest({ activeReviewReconciliation: { enabled: false } }); + expect(m.activeReviewReconciliation).toEqual({ present: true, enabled: false }); + expect(m.present).toBe(true); + }); + + it("warns and defaults to false when enabled is a non-boolean value", () => { + const m = parseFocusManifest({ activeReviewReconciliation: { enabled: "yes" as unknown as boolean } }); + expect(m.activeReviewReconciliation.enabled).toBe(false); + expect(m.warnings.some((w) => /activeReviewReconciliation\.enabled/.test(w))).toBe(true); + }); + + it("round-trips through activeReviewReconciliationConfigToJson → parseFocusManifest unchanged", () => { + const m = parseFocusManifest({ activeReviewReconciliation: { enabled: true } }); + expect(parseFocusManifest({ activeReviewReconciliation: activeReviewReconciliationConfigToJson(m.activeReviewReconciliation) }).activeReviewReconciliation).toEqual(m.activeReviewReconciliation); + }); + + it("activeReviewReconciliationConfigToJson returns null for an absent config", () => { + expect(activeReviewReconciliationConfigToJson(parseFocusManifest(null).activeReviewReconciliation)).toBeNull(); + }); + }); + describe("federatedIntelligence: (#1970, opt-in federated fleet intelligence export config-as-code toggle)", () => { it("defaults to fully disabled/absent when the key is omitted, and does not make the manifest present on its own", () => { const m = parseFocusManifest({}); diff --git a/test/unit/index.test.ts b/test/unit/index.test.ts index 96c41ea9c5..dad89e5381 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -943,6 +943,47 @@ describe("worker entrypoint", () => { expect(sent.some((m) => m.type === "reconcile-open-prs")).toBe(false); }); + it("enqueues the reconcile-active-review-tracking job every 10 minutes ONLY when LOOPOVER_ACTIVE_REVIEW_RECONCILIATION is ON (flag-OFF is byte-identical)", async () => { + const sentFor = async (flag?: string, isoTime = "2026-05-25T05:10:00.000Z"): Promise> => { + const sent: Array = []; + const env = createTestEnv({ + ...(flag === undefined ? {} : { LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: flag }), + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + await worker.scheduled(controllerFor(isoTime), env, executionContext(waitUntil)); + await Promise.all(waitUntil); + return sent; + }; + + // Flag OFF (default) → no reconcile-active-review-tracking job; the enqueued set is unchanged from today. + expect((await sentFor()).some((m) => m.type === "reconcile-active-review-tracking")).toBe(false); + expect((await sentFor("false")).some((m) => m.type === "reconcile-active-review-tracking")).toBe(false); + // Flag ON, on a 10-minute boundary → exactly one reconcile-active-review-tracking job. + const on = await sentFor("true"); + expect(on.filter((m) => m.type === "reconcile-active-review-tracking")).toEqual([{ type: "reconcile-active-review-tracking", requestedBy: "schedule" }]); + }); + + it("does NOT enqueue reconcile-active-review-tracking outside the 10-minute window even when LOOPOVER_ACTIVE_REVIEW_RECONCILIATION is ON", async () => { + const sent: Array = []; + const env = createTestEnv({ + LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "true", + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + await worker.scheduled(controllerFor("2026-05-25T05:14:00.000Z"), env, executionContext(waitUntil)); // not a 10-minute boundary + await Promise.all(waitUntil); + expect(sent.some((m) => m.type === "reconcile-active-review-tracking")).toBe(false); + }); + it("enqueues selftune hourly only when LOOPOVER_REVIEW_SELFTUNE is ON", async () => { const sentFor = async ( selfTuneFlag?: string, diff --git a/test/unit/queue-5.test.ts b/test/unit/queue-5.test.ts index 66df8b0a30..9d535ba932 100644 --- a/test/unit/queue-5.test.ts +++ b/test/unit/queue-5.test.ts @@ -5883,6 +5883,25 @@ describe("queue processors", () => { reconcileSpy.mockRestore(); }); + it("reconcile-active-review-tracking job no-ops when LOOPOVER_ACTIVE_REVIEW_RECONCILIATION is OFF (does no scan)", async () => { + const env = createTestEnv(); // flag unset → OFF + const listSpy = vi.spyOn(repositoriesModule, "listStaleActiveReviewTracking"); + + await processJob(env, { type: "reconcile-active-review-tracking", requestedBy: "test" }); + + expect(listSpy).not.toHaveBeenCalled(); + }); + + it("reconcile-active-review-tracking job runs the reconciliation scan when LOOPOVER_ACTIVE_REVIEW_RECONCILIATION is ON", async () => { + const env = createTestEnv({ LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "true" }); + const listSpy = vi.spyOn(repositoriesModule, "listStaleActiveReviewTracking").mockResolvedValueOnce([]); + + await processJob(env, { type: "reconcile-active-review-tracking", requestedBy: "test" }); + + expect(listSpy).toHaveBeenCalled(); + listSpy.mockRestore(); + }); + it("retry-orb-relay job dispatches into retryFailedRelays, pruning an expired relay-failure row (#relay-retry)", async () => { const env = createTestEnv(); await env.DB.prepare( diff --git a/test/unit/selfhost-config-lint.test.ts b/test/unit/selfhost-config-lint.test.ts index da2cbb5fd7..3728080874 100644 --- a/test/unit/selfhost-config-lint.test.ts +++ b/test/unit/selfhost-config-lint.test.ts @@ -150,6 +150,14 @@ reviewRecap: expect(result.recognizedFields).toEqual(["prReconciliation"]); }); + it("recognizes a standalone top-level activeReviewReconciliation: block instead of flagging it as unknown (#webhook-reorder-clobber)", () => { + const result = lintManifestText("activeReviewReconciliation:\n enabled: true\n"); + + expect(result.ok).toBe(true); + expect(result.warnings).toEqual([]); + expect(result.recognizedFields).toEqual(["activeReviewReconciliation"]); + }); + it("recognizes a standalone top-level federatedIntelligence: block instead of flagging it as unknown (#6998)", () => { const result = lintManifestText("federatedIntelligence:\n enabled: true\n"); diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index db831e739a..5967f012d7 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -282,6 +282,9 @@ describe("self-host queue common helpers", () => { expect(isGitHubBudgetBackgroundJob({ type: "generate-review-recap", requestedBy: "schedule", repoFullName: "owner/repo" })).toBe(true); // reconcile-open-prs: runOpenPrReconciliation makes large paginated GitHub REST calls per watched repo. expect(isGitHubBudgetBackgroundJob({ type: "reconcile-open-prs", requestedBy: "schedule" })).toBe(true); + // reconcile-active-review-tracking: runActiveReviewReconciliation makes one live GET /pulls/{n} call per + // stale active_review_tracking row it finds (#webhook-reorder-clobber). + expect(isGitHubBudgetBackgroundJob({ type: "reconcile-active-review-tracking", requestedBy: "schedule" })).toBe(true); }); it("REGRESSION (#4505): maintenance job types confirmed to make NO GitHub REST calls stay OFF the GitHub budget (never wrongly gated)", () => { diff --git a/worker-configuration.d.ts b/worker-configuration.d.ts index 5aca388ca8..5e841fa0d1 100644 --- a/worker-configuration.d.ts +++ b/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: 3decdc667b76a0025c42234ff89dee26) +// Generated by Wrangler by running `wrangler types` (hash: 941f69060238ea20a568fa8ea50b814c) // Runtime types generated with workerd@1.20260701.1 2026-05-28 nodejs_compat interface __BaseEnv_Env { REVIEW_AUDIT: R2Bucket; @@ -29,6 +29,7 @@ interface __BaseEnv_Env { LOOPOVER_SWEEP_WATCHDOG: "false"; LOOPOVER_LOOP_ESCALATION: "false"; LOOPOVER_PR_RECONCILIATION: "false"; + LOOPOVER_ACTIVE_REVIEW_RECONCILIATION: "false"; LOOPOVER_REVIEW_RAG: "false"; LOOPOVER_REVIEW_IMPACT_MAP: "false"; LOOPOVER_REVIEW_CULTURE_PROFILE: "false"; @@ -71,6 +72,7 @@ declare namespace NodeJS { | "GITTENSOR_REGISTRY_URL" | "GITTENSOR_UPSTREAM_REF" | "GITTENSOR_UPSTREAM_REPO" + | "LOOPOVER_ACTIVE_REVIEW_RECONCILIATION" | "LOOPOVER_AUTO_FILE_DRIFT_ISSUES" | "LOOPOVER_DRIFT_ISSUE_REPO" | "LOOPOVER_DUPLICATE_WINNER" diff --git a/wrangler.jsonc b/wrangler.jsonc index 90a06ca847..e17085a92b 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -115,6 +115,12 @@ // backfillRegisteredRepositories freshness window. Default OFF — flag-OFF the cron enqueues no // reconciliation job, byte-identical to today. "LOOPOVER_PR_RECONCILIATION": "false", + // Self-heal (#webhook-reorder-clobber): same short-interval cron as LOOPOVER_PR_RECONCILIATION above ALSO + // re-checks every active_review_tracking row stuck in `status: "active"` longer than 15 minutes against + // LIVE (non-cached) GitHub state, and terminalizes the ones GitHub confirms are actually closed — a + // delayed webhook job can otherwise restart tracking for a PR that already closed, orphaning the row + // forever. Default OFF — flag-OFF the cron enqueues no reconciliation job, byte-identical to today. + "LOOPOVER_ACTIVE_REVIEW_RECONCILIATION": "false", // Convergence (RAG retrieval): at review time, query the codebase vector index for code/docs semantically // related to the PR's changed files and append a RELEVANT EXISTING CODE / DOCS section to the reviewer // prompt — additive reference context (callers, related modules, conventions), exactly like grounding.