From 6eb1fabf192741f9582d4d2c25274c680ef27ae4 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 04:05:49 -0700 Subject: [PATCH 1/5] feat(orb): weekly stratified decision-audit sampling with a frozen adjudication rubric The gate's only confirmation signal is 'no human reversed it' -- a lower bound on error, not a label: humans reverse only what they notice, so a silently-wrong merge scores as correct and every accuracy figure inherits the bias. Closes #8830. Part of epic #8828 (Phase 2 -- labels). - decision_audit_labels (migration 0178): one label per PR ever, adjudications comparable only within a rubric version - pure stratified sampler: merges oversampled (the costly arm), a first-time-author stratum (weakest priors), deterministic under an injected rng, thin strata spill instead of shrinking the sample - weekly cron enqueue (Tuesday 08:00 UTC), flag-gated by LOOPOVER_DECISION_AUDIT (default OFF, byte-identical), self-host only; the dispatch re-checks the flag so a stale queued job after a flag flip does zero work - internal adjudication routes: list (filterable) + adjudicate; a second adjudication 409s -- labels are calibration data, rewrites must be deliberate via a rubric-version bump, never silent - docs/decision-audit-rubric.md v1: judge the decision at decision-time knowledge, not the outcome; 'uncertain' is a first-class label excluded from both sides and tracked as a rate --- docs/decision-audit-rubric.md | 45 +++++ migrations/0178_decision_audit_labels.sql | 23 +++ scripts/check-schema-drift.ts | 1 + src/api/routes.ts | 35 ++++ src/env.d.ts | 2 + src/index.ts | 8 + src/queue/job-dispatch.ts | 9 + src/review/decision-audit.ts | 177 ++++++++++++++++ src/types.ts | 6 + test/integration/audit-labels-routes.test.ts | 60 ++++++ test/unit/decision-audit.test.ts | 201 +++++++++++++++++++ test/unit/index.test.ts | 20 ++ 12 files changed, 587 insertions(+) create mode 100644 docs/decision-audit-rubric.md create mode 100644 migrations/0178_decision_audit_labels.sql create mode 100644 src/review/decision-audit.ts create mode 100644 test/integration/audit-labels-routes.test.ts create mode 100644 test/unit/decision-audit.test.ts diff --git a/docs/decision-audit-rubric.md b/docs/decision-audit-rubric.md new file mode 100644 index 0000000000..a97e73c879 --- /dev/null +++ b/docs/decision-audit-rubric.md @@ -0,0 +1,45 @@ +# Decision-audit rubric — version 1 + +This rubric governs human adjudication of the weekly decision-audit sample (#8830, epic #8828). +Adjudications are comparable **only within a rubric version**: any change to the criteria below requires +bumping `DECISION_AUDIT_RUBRIC_VERSION` in `src/review/decision-audit.ts` in the same PR, so estimates never +silently mix incompatible labels. + +## What is being judged + +One question per sampled PR: **was the gate's decision (merge or close) the decision a careful maintainer +would have made, given only what was knowable at decision time?** + +- Judge the decision, not the outcome. A merge that later broke something the gate could not have seen is + still `correct`; a merge that shipped a defect visible in the diff at review time is `incorrect` even if + nobody noticed. +- Ignore reversal state. "Nobody reversed it" is not evidence of correctness — that bias is exactly what + this audit exists to measure. +- Use `uncertain` honestly. A genuine judgment call where reasonable maintainers would split is `uncertain`, + not a coin flip. Uncertain labels are excluded from the accuracy numerator AND denominator and tracked as + their own rate (a rising uncertain rate is a signal the gate is operating in territory that needs a human). + +## Labels + +| label | meaning | +|---|---| +| `correct` | The decision was right given decision-time information. | +| `incorrect` | The decision was wrong given decision-time information. | +| `uncertain` | Reasonable maintainers would disagree; no defensible single answer. | + +## Reason categories (for `incorrect`, optional otherwise) + +- `missed_defect` — merged with a defect visible in the diff. +- `false_block` — closed a PR that met the bar. +- `stale_signal` — decided on out-of-date CI/conflict/issue state. +- `scope_misread` — misjudged linked-issue scope or requirements. +- `policy_misapplied` — an enforcement rule fired on a case it should not cover. +- `other` — anything else; describe in the adjudication notes if used. + +## Workflow + +1. `GET /v1/internal/audit-labels?status=pending` lists the week's sample. +2. Review each PR **as of its decision time** (the decision record pins head SHA and inputs). +3. `POST /v1/internal/audit-labels/adjudicate` with `{ id, adjudication, reasonCategory? }`. +4. A second adjudication of the same label is rejected (409). Corrections require a new rubric version — + deliberate, never silent. diff --git a/migrations/0178_decision_audit_labels.sql b/migrations/0178_decision_audit_labels.sql new file mode 100644 index 0000000000..80425e6979 --- /dev/null +++ b/migrations/0178_decision_audit_labels.sql @@ -0,0 +1,23 @@ +-- #8830 (epic #8828, Phase 2 — labels): human-adjudicated ground truth for the gate's decisions. +-- +-- The reversal-based confirmation signal is a LOWER BOUND on error, not a label: humans reverse only what +-- they notice, so a silently-wrong merge scores as correct and every downstream accuracy figure inherits the +-- bias. This table holds the weekly stratified audit sample and its human adjudications — the calibration +-- set the risk-control thresholds (#8835) and the audited-accuracy estimate both read. One row per PR ever +-- (the UNIQUE on target_id): a PR re-drawn in a later week is skipped, never double-labeled. +CREATE TABLE IF NOT EXISTS decision_audit_labels ( + id TEXT PRIMARY KEY, -- audit: + project TEXT NOT NULL, -- owner/repo + target_id TEXT NOT NULL, -- owner/repo#N + verdict TEXT NOT NULL CHECK (verdict IN ('merge', 'close')), -- the gate's decision at sample time + outcome TEXT NOT NULL CHECK (outcome IN ('merged', 'closed')), -- realized outcome at sample time + stratum TEXT NOT NULL CHECK (stratum IN ('merge_arm', 'close_arm', 'first_time_author')), + rubric_version TEXT NOT NULL, -- adjudications are only comparable within a version + sampled_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'adjudicated')), + adjudication TEXT CHECK (adjudication IN ('correct', 'incorrect', 'uncertain')), + reason_category TEXT, -- free-form-bounded category from the rubric + adjudicated_at TEXT, + UNIQUE (target_id) +); +CREATE INDEX IF NOT EXISTS decision_audit_labels_status ON decision_audit_labels (status, sampled_at); diff --git a/scripts/check-schema-drift.ts b/scripts/check-schema-drift.ts index f04495829f..feec1f92bd 100644 --- a/scripts/check-schema-drift.ts +++ b/scripts/check-schema-drift.ts @@ -40,6 +40,7 @@ export const RAW_SQL_ONLY_TABLES: Set = new Set([ "ams_instances", "ams_signals", "contributor_gate_history", + "decision_audit_labels", "global_agent_controls", "global_contributor_blacklist", "global_moderation_config", diff --git a/src/api/routes.ts b/src/api/routes.ts index 81be8c3364..ef4fc4a7d1 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -4397,6 +4397,41 @@ export function createApp() { // but only REGISTERED ones count toward fleet calibration (computeFleetAnalytics). Bearer-gated by the // `/v1/internal/*` middleware (INTERNAL_JOB_TOKEN). List shows pending + registered instances with their // stored-signal counts so an operator knows what they're opting in before they register it. + // Decision-audit adjudication (#8830, epic #8828): the operator surface for the weekly stratified sample. + // Bearer-gated by the /v1/internal/* middleware (INTERNAL_JOB_TOKEN). List is filterable by status; the + // adjudicate write is idempotent-hostile on purpose — a second adjudication 409s rather than silently + // rewriting a label (labels are calibration data; rewrites must be deliberate, via a fresh rubric version). + app.get("/v1/internal/audit-labels", async (c) => { + const status = c.req.query("status"); + const where = status === "pending" || status === "adjudicated" ? "WHERE status = ?" : ""; + const stmt = c.env.DB.prepare( + `SELECT id, project, target_id AS targetId, verdict, outcome, stratum, rubric_version AS rubricVersion, + sampled_at AS sampledAt, status, adjudication, reason_category AS reasonCategory, adjudicated_at AS adjudicatedAt + FROM decision_audit_labels ${where} ORDER BY sampled_at DESC, target_id ASC LIMIT 500`, + ); + const rows = await (where ? stmt.bind(status) : stmt).all(); + return c.json({ labels: rows.results }); + }); + + app.post("/v1/internal/audit-labels/adjudicate", async (c) => { + const payload = (await c.req.json().catch(() => null)) as { id?: unknown; adjudication?: unknown; reasonCategory?: unknown } | null; + const id = typeof payload?.id === "string" ? payload.id : ""; + const adjudication = payload?.adjudication; + if (!id || (adjudication !== "correct" && adjudication !== "incorrect" && adjudication !== "uncertain")) { + return c.json({ error: "id and adjudication (correct|incorrect|uncertain) required" }, 400); + } + const reasonCategory = typeof payload?.reasonCategory === "string" ? payload.reasonCategory.slice(0, 100) : null; + const existing = await c.env.DB.prepare("SELECT status FROM decision_audit_labels WHERE id = ?").bind(id).first<{ status: string }>(); + if (!existing) return c.json({ error: "label_not_found" }, 404); + if (existing.status === "adjudicated") return c.json({ error: "already_adjudicated" }, 409); + await c.env.DB.prepare( + `UPDATE decision_audit_labels SET status = 'adjudicated', adjudication = ?, reason_category = ?, adjudicated_at = ? WHERE id = ?`, + ) + .bind(adjudication, reasonCategory, new Date().toISOString(), id) + .run(); + return c.json({ id, adjudication, reasonCategory }); + }); + app.get("/v1/internal/orb/instances", async (c) => { const rows = await c.env.DB .prepare( diff --git a/src/env.d.ts b/src/env.d.ts index a79e841438..0ec5ed9585 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -502,6 +502,8 @@ declare global { * recording are wired, reading a promoted override into the live gate is a noted follow-up that must not * risk loosening the gate. See src/review/selftune-wire.ts. */ LOOPOVER_REVIEW_SELFTUNE?: string; + /** #8830: weekly stratified human-audit sampling of gate decisions (default OFF). */ + LOOPOVER_DECISION_AUDIT?: string; /** Experimental `gittensor` plugin (the `experimental:` manifest block, first key): the operator-level * kill-switch for loopover's original subnet mining-registry/scoring integration, now opt-in rather than * a core dependency. ANDed with the per-repo `.loopover.yml experimental.gittensor` opt-in -- neither diff --git a/src/index.ts b/src/index.ts index 7656e5e919..04e44231d2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -14,6 +14,7 @@ import { isAprRepoTransferPollEnabled } from "./orb/apr-repo-transfer"; import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride } from "./review/pr-reconciliation"; import { isActiveReviewReconciliationEnabled, resolveActiveReviewReconciliationManifestOverride } from "./review/active-review-reconciliation"; import { isRagEnabled } from "./review/rag-wire"; +import { isDecisionAuditEnabled } from "./review/decision-audit"; import { isSelfTuneEnabled } from "./review/selftune-wire"; import { isSatisfactionFloorAutotuneEnabled } from "./services/satisfaction-floor-loosening-run"; import { @@ -290,6 +291,13 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): if (isHourly && scheduledAt.getUTCDay() === 1 && hour === 12) { jobs.push({ type: "generate-weekly-value-report", requestedBy: "schedule", variant: "operator", days: 7 }); } + // Decision-audit sampling (#8830, flag LOOPOVER_DECISION_AUDIT): weekly stratified draw of decided PRs for + // human adjudication. Tuesday 08:00 UTC — its own slot, clear of the Monday weekly report, the 03:00 + // retention prune, and the 09:00 repo-doc sweep. Enqueued ONLY when the flag is ON — flag-OFF (default) + // this job is never created, so the cron tick does ZERO new work and the enqueued set is byte-identical. + if (isHourly && scheduledAt.getUTCDay() === 2 && hour === 8 && selfHostedReviews && isDecisionAuditEnabled(env)) { + jobs.push({ type: "decision-audit-sample", requestedBy: "schedule" }); + } // Prune expired log/snapshot rows once a day (03:00 UTC) per the conservative RETENTION_POLICY. if (isHourly && hour === 3) { jobs.push({ type: "prune-retention", requestedBy: "schedule" }); diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 96435c281a..dbb34e20de 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -46,6 +46,7 @@ import { import { syncBrokeredInstalledRepos } from "../orb/installed-repos-sync"; import { incr } from "../selfhost/metrics"; import { generateSignalSnapshots } from "./signal-snapshot"; +import { isDecisionAuditEnabled, runDecisionAuditSample } from "../review/decision-audit"; import { runRetentionPrune } from "./retention"; // The 15 handlers below have no reason to move -- each is only reachable via this dispatcher (or, for // mapWithConcurrency, ALSO used by other still-in-processors.ts code), so they stay put and are exported @@ -235,6 +236,14 @@ export async function processJob(env: Env, message: JobMessage): Promise { message.dryRun ?? false, ); return; + case "decision-audit-sample": { + // #8830: flag re-checked at execution (not only at enqueue) so a stale queued job after a flag flip + // does zero work, mirroring every sibling flag-gated job. + if (!isDecisionAuditEnabled(env)) return; + const inserted = await runDecisionAuditSample(env); + console.log(JSON.stringify({ event: "decision_audit_sampled", inserted })); + return; + } case "generate-weekly-value-report": await generateWeeklyValueReport(env, { variant: message.variant ?? "operator", diff --git a/src/review/decision-audit.ts b/src/review/decision-audit.ts new file mode 100644 index 0000000000..801024588c --- /dev/null +++ b/src/review/decision-audit.ts @@ -0,0 +1,177 @@ +// Decision-audit sampling (#8830, epic #8828 Phase 2) — the human-label pipeline. +// +// WHY: the gate's only confirmation signal today is "no human reversed it", which is a lower bound on error, +// not a label — humans reverse only what they notice, so a silently-wrong merge scores as correct. This +// module draws a weekly STRATIFIED sample of decided PRs for human adjudication against a frozen rubric: +// merges oversampled (the costly arm), plus a first-time-author stratum (new contributors are where the +// gate's priors are weakest). Adjudications land in decision_audit_labels (migration 0178) and feed two +// consumers: the calibration set for the risk-control thresholds (#8835) and an audited-accuracy estimate +// whose divergence from the reversal-based number IS the measurement of label bias. +// +// Flag-gated by LOOPOVER_DECISION_AUDIT (default OFF — the weekly job is never enqueued, zero new work, +// byte-identical to today), mirroring every sibling cron flag (selftune-wire et al). +import { errorMessage, nowIso } from "../utils/json"; + +/** Bump ONLY with a corresponding edit to docs/decision-audit-rubric.md — adjudications are comparable only + * within a rubric version, and a silent rubric drift would corrupt every downstream estimate. */ +export const DECISION_AUDIT_RUBRIC_VERSION = "1"; + +/** Weekly sample size. ~30/week ≈ 120/month human labels: enough to run the audited-accuracy estimator and + * grow #8835's calibration set without making adjudication a burden. */ +export const DECISION_AUDIT_SAMPLE_SIZE = 30; + +/** Stratum quotas over the sample size: merges oversampled (a wrong merge is the costly error), a dedicated + * first-time-author slice (weakest priors), remainder to closes. Shortfalls in any stratum spill into the + * others rather than shrinking the week's sample. */ +export const DECISION_AUDIT_STRATA = { merge_arm: 0.5, close_arm: 0.3, first_time_author: 0.2 } as const; + +export type AuditStratum = keyof typeof DECISION_AUDIT_STRATA; + +export function isDecisionAuditEnabled(env: { LOOPOVER_DECISION_AUDIT?: string | undefined }): boolean { + return /^(1|true|yes|on)$/i.test((env.LOOPOVER_DECISION_AUDIT ?? "").trim()); +} + +/** One decided PR eligible for sampling. */ +export type AuditCandidate = { + targetId: string; // owner/repo#N + project: string; + verdict: "merge" | "close"; + outcome: "merged" | "closed"; + /** True when this PR's author had no earlier decided PR in the ledger at sample time. */ + firstTimeAuthor: boolean; +}; + +export type PlannedAuditRow = { + targetId: string; + project: string; + verdict: "merge" | "close"; + outcome: "merged" | "closed"; + stratum: AuditStratum; +}; + +/** + * PURE stratified sampler. Deterministic under the injected `rng` (tests pass a seeded generator). + * + * Selection: first-time-author candidates fill their stratum first (they are the scarcest and also belong + * to an arm stratum — claiming them first prevents the arm quotas from consuming them all); then each arm + * fills its quota from the remaining pool; any shortfall spills into a final fill from whatever remains, + * tagged with the candidate's arm stratum. Never returns more than `k` rows or duplicates a target. + */ +export function planAuditSample(candidates: AuditCandidate[], k: number = DECISION_AUDIT_SAMPLE_SIZE, rng: () => number = Math.random): PlannedAuditRow[] { + if (k <= 0 || candidates.length === 0) return []; + const shuffled = [...candidates]; + // Fisher-Yates with the injected rng — the ONLY nondeterminism in the plan. + for (let i = shuffled.length - 1; i > 0; i -= 1) { + const j = Math.floor(rng() * (i + 1)); + [shuffled[i], shuffled[j]] = [shuffled[j]!, shuffled[i]!]; + } + + const quota = (share: number): number => Math.round(k * share); + const picked = new Map(); + const take = (pool: AuditCandidate[], n: number, stratum: AuditStratum | null): void => { + for (const candidate of pool) { + if (picked.size >= k || n <= 0) return; + if (picked.has(candidate.targetId)) continue; + picked.set(candidate.targetId, { + targetId: candidate.targetId, + project: candidate.project, + verdict: candidate.verdict, + outcome: candidate.outcome, + stratum: stratum ?? (candidate.verdict === "merge" ? "merge_arm" : "close_arm"), + }); + n -= 1; + } + }; + + take(shuffled.filter((c) => c.firstTimeAuthor), quota(DECISION_AUDIT_STRATA.first_time_author), "first_time_author"); + take(shuffled.filter((c) => c.verdict === "merge"), quota(DECISION_AUDIT_STRATA.merge_arm), "merge_arm"); + take(shuffled.filter((c) => c.verdict === "close"), quota(DECISION_AUDIT_STRATA.close_arm), "close_arm"); + // Spill: fill any remaining budget from the whole pool so a thin stratum never shrinks the week's sample. + take(shuffled, k - picked.size, null); + return [...picked.values()]; +} + +/** Candidate window: the trailing week of decided PRs (the job runs weekly; a longer window would re-offer + * PRs already declined by the UNIQUE(target_id) guard for no benefit). */ +const CANDIDATE_WINDOW_MS = 7 * 86_400_000; + +/** + * IO: draw this week's sample from the live ledger and insert PENDING rows. Idempotent two ways: re-running + * within a week re-draws but the UNIQUE(target_id) INSERT OR IGNORE never duplicates a PR, and previously + * adjudicated PRs are excluded at the candidate query. Best-effort per row — one bad insert never voids the + * batch. Returns the number of rows actually inserted. + */ +export async function runDecisionAuditSample(env: Env, nowMs: number = Date.now(), rng: () => number = Math.random): Promise { + const sinceIso = new Date(nowMs - CANDIDATE_WINDOW_MS).toISOString(); + // Decided = latest native gate_decision joined to a realized pr_outcome, inside the window, minus policy + // actions (enforcement is not a quality decision — #8827) and minus already-sampled targets. + const { results } = await env.DB.prepare( + `WITH gd AS ( + SELECT target_id, project, decision AS verdict, summary, created_at, + ROW_NUMBER() OVER (PARTITION BY target_id ORDER BY created_at DESC) AS rn + FROM review_audit + WHERE event_type = 'gate_decision' AND decision IN ('merge', 'close') AND source = 'gittensory-native' + ), + po AS ( + SELECT target_id, decision AS outcome, + ROW_NUMBER() OVER (PARTITION BY target_id ORDER BY created_at DESC) AS rn + FROM review_audit + WHERE event_type = 'pr_outcome' AND decision IN ('merged', 'closed') + ) + SELECT gd.target_id AS targetId, gd.project AS project, gd.verdict AS verdict, po.outcome AS outcome + FROM gd + JOIN po ON po.target_id = gd.target_id AND po.rn = 1 + WHERE gd.rn = 1 + AND gd.created_at >= ? + AND (gd.summary IS NULL OR gd.summary NOT LIKE 'policy_close:%') + AND NOT EXISTS (SELECT 1 FROM decision_audit_labels dal WHERE dal.target_id = gd.target_id)`, + ) + .bind(sinceIso) + .all<{ targetId: string; project: string; verdict: "merge" | "close"; outcome: "merged" | "closed" }>(); + + const decided = results; + if (decided.length === 0) return 0; + + // First-time detection: the author's earliest decided PR. review_audit carries no logins by design, so the + // author comes from the pull_requests cache; a PR with no cached row degrades to "not first-time" (the + // stratum under-fills and the spill covers it) rather than guessing. + const authorRows = await env.DB.prepare( + `SELECT repo_full_name || '#' || number AS targetId, author_login AS author FROM pull_requests WHERE author_login IS NOT NULL`, + ).all<{ targetId: string; author: string }>(); + const authorByTarget = new Map(authorRows.results.map((r) => [r.targetId, r.author.toLowerCase()])); + const decidedCountByAuthor = new Map(); + for (const row of authorRows.results) { + const author = row.author.toLowerCase(); + decidedCountByAuthor.set(author, (decidedCountByAuthor.get(author) ?? 0) + 1); + } + + const candidates: AuditCandidate[] = decided.map((row) => { + const author = authorByTarget.get(row.targetId); + return { + targetId: row.targetId, + project: row.project, + verdict: row.verdict, + outcome: row.outcome, + // author present in authorByTarget implies a count entry (both derive from the same rows), hence the !. + firstTimeAuthor: author !== undefined && decidedCountByAuthor.get(author)! <= 1, + }; + }); + + const plan = planAuditSample(candidates, DECISION_AUDIT_SAMPLE_SIZE, rng); + const sampledAt = nowIso(); + let inserted = 0; + for (const row of plan) { + try { + const result = await env.DB.prepare( + `INSERT OR IGNORE INTO decision_audit_labels (id, project, target_id, verdict, outcome, stratum, rubric_version, sampled_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind(`audit:${row.targetId}`.slice(0, 190), row.project.slice(0, 200), row.targetId, row.verdict, row.outcome, row.stratum, DECISION_AUDIT_RUBRIC_VERSION, sampledAt) + .run(); + if (result.meta.changes > 0) inserted += 1; + } catch (error) { + console.warn(JSON.stringify({ event: "decision_audit_insert_error", target: row.targetId, message: errorMessage(error).slice(0, 160) })); + } + } + return inserted; +} diff --git a/src/types.ts b/src/types.ts index a1f0bb1565..067180e649 100644 --- a/src/types.ts +++ b/src/types.ts @@ -145,6 +145,12 @@ export type JobMessage = requestedBy: "schedule" | "api" | "test"; dryRun?: boolean; } + | { + // Decision-audit sampling (#8830, epic #8828): weekly stratified draw of decided PRs for human + // adjudication (decision_audit_labels). Flag-gated by LOOPOVER_DECISION_AUDIT. + type: "decision-audit-sample"; + requestedBy: "schedule" | "api" | "test"; + } | { type: "generate-weekly-value-report"; requestedBy: "schedule" | "api" | "test"; diff --git a/test/integration/audit-labels-routes.test.ts b/test/integration/audit-labels-routes.test.ts new file mode 100644 index 0000000000..e5d5e66a42 --- /dev/null +++ b/test/integration/audit-labels-routes.test.ts @@ -0,0 +1,60 @@ +import { describe, expect, it } from "vitest"; +import { createApp } from "../../src/api/routes"; +import { createTestEnv } from "../helpers/d1"; + +// #8830: the adjudication operator surface. A label is calibration data — the routes must never let one be +// silently rewritten (409 on a second adjudication) and must reject malformed labels outright. +describe("decision-audit adjudication routes (/v1/internal/audit-labels)", () => { + const app = createApp(); + const auth = { authorization: "Bearer dev-internal-token", "content-type": "application/json" }; + + async function seedLabel(env: Env, id = "audit:o/r#1"): Promise { + await env.DB.prepare( + `INSERT INTO decision_audit_labels (id, project, target_id, verdict, outcome, stratum, rubric_version, sampled_at) + VALUES (?, 'o/r', ?, 'merge', 'merged', 'merge_arm', '1', ?)`, + ) + .bind(id, id.replace("audit:", ""), new Date().toISOString()) + .run(); + } + + it("requires the internal bearer", async () => { + expect((await app.request("/v1/internal/audit-labels", {}, createTestEnv())).status).toBe(401); + }); + + it("lists labels, filterable by status", async () => { + const env = createTestEnv(); + await seedLabel(env, "audit:o/r#1"); + await seedLabel(env, "audit:o/r#2"); + const all = (await (await app.request("/v1/internal/audit-labels", { headers: auth }, env)).json()) as { labels: Array<{ id: string; status: string }> }; + expect(all.labels).toHaveLength(2); + const pending = (await (await app.request("/v1/internal/audit-labels?status=pending", { headers: auth }, env)).json()) as { labels: unknown[] }; + expect(pending.labels).toHaveLength(2); + const done = (await (await app.request("/v1/internal/audit-labels?status=adjudicated", { headers: auth }, env)).json()) as { labels: unknown[] }; + expect(done.labels).toHaveLength(0); + }); + + it("adjudicates a pending label once; a second write 409s; unknown 404s; bad payloads 400", async () => { + const env = createTestEnv(); + await seedLabel(env); + const ok = await app.request( + "/v1/internal/audit-labels/adjudicate", + { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:o/r#1", adjudication: "incorrect", reasonCategory: "missed_defect" }) }, + env, + ); + expect(ok.status).toBe(200); + const row = await env.DB.prepare("SELECT status, adjudication, reason_category, adjudicated_at FROM decision_audit_labels WHERE id = 'audit:o/r#1'").first<{ status: string; adjudication: string; reason_category: string; adjudicated_at: string }>(); + expect(row).toMatchObject({ status: "adjudicated", adjudication: "incorrect", reason_category: "missed_defect" }); + expect(typeof row!.adjudicated_at).toBe("string"); + + const again = await app.request("/v1/internal/audit-labels/adjudicate", { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:o/r#1", adjudication: "correct" }) }, env); + expect(again.status).toBe(409); + + expect((await app.request("/v1/internal/audit-labels/adjudicate", { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:nope#1", adjudication: "correct" }) }, env)).status).toBe(404); + expect((await app.request("/v1/internal/audit-labels/adjudicate", { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:o/r#1", adjudication: "maybe" }) }, env)).status).toBe(400); + expect((await app.request("/v1/internal/audit-labels/adjudicate", { method: "POST", headers: auth, body: "{not json" }, env)).status).toBe(400); + // uncertain is a first-class label, not an error; reasonCategory stays optional. + await seedLabel(env, "audit:o/r#2"); + const uncertain = await app.request("/v1/internal/audit-labels/adjudicate", { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:o/r#2", adjudication: "uncertain" }) }, env); + expect(uncertain.status).toBe(200); + }); +}); diff --git a/test/unit/decision-audit.test.ts b/test/unit/decision-audit.test.ts new file mode 100644 index 0000000000..f0bfe94862 --- /dev/null +++ b/test/unit/decision-audit.test.ts @@ -0,0 +1,201 @@ +import { describe, expect, it } from "vitest"; +import { + DECISION_AUDIT_RUBRIC_VERSION, + DECISION_AUDIT_SAMPLE_SIZE, + isDecisionAuditEnabled, + planAuditSample, + runDecisionAuditSample, + type AuditCandidate, +} from "../../src/review/decision-audit"; +import { processJob } from "../../src/queue/processors"; +import { createTestEnv } from "../helpers/d1"; + +// #8830: the human-label pipeline. The sampler decides WHO gets audited — a biased or flappy draw would +// corrupt every estimate built on the labels, so determinism and strata behavior are pinned exactly. +function seededRng(seed: number): () => number { + let state = seed >>> 0; + return () => { + state = (state * 1664525 + 1013904223) >>> 0; + return state / 2 ** 32; + }; +} + +function candidate(i: number, over: Partial = {}): AuditCandidate { + return { targetId: `o/r#${i}`, project: "o/r", verdict: "merge", outcome: "merged", firstTimeAuthor: false, ...over }; +} + +describe("isDecisionAuditEnabled", () => { + it("truthy strings enable; default is OFF", () => { + expect(isDecisionAuditEnabled({ LOOPOVER_DECISION_AUDIT: "true" })).toBe(true); + expect(isDecisionAuditEnabled({ LOOPOVER_DECISION_AUDIT: "0" })).toBe(false); + expect(isDecisionAuditEnabled({})).toBe(false); + }); +}); + +describe("planAuditSample (#8830)", () => { + it("is DETERMINISTIC under a seeded rng — same inputs, same draw", () => { + const pool = Array.from({ length: 100 }, (_, i) => candidate(i, { verdict: i % 3 === 0 ? "close" : "merge", outcome: i % 3 === 0 ? "closed" : "merged", firstTimeAuthor: i % 10 === 0 })); + const a = planAuditSample(pool, 30, seededRng(42)); + const b = planAuditSample(pool, 30, seededRng(42)); + expect(a).toEqual(b); + const c = planAuditSample(pool, 30, seededRng(43)); + expect(c.map((r) => r.targetId)).not.toEqual(a.map((r) => r.targetId)); // a different seed draws differently + }); + + it("fills strata quotas when the pool is rich: ~50% merges, ~30% closes, ~20% first-time", () => { + const pool = [ + ...Array.from({ length: 40 }, (_, i) => candidate(i)), + ...Array.from({ length: 40 }, (_, i) => candidate(100 + i, { verdict: "close", outcome: "closed" })), + ...Array.from({ length: 40 }, (_, i) => candidate(200 + i, { firstTimeAuthor: true })), + ]; + const plan = planAuditSample(pool, 30, seededRng(7)); + expect(plan).toHaveLength(30); + const byStratum = new Map(); + for (const row of plan) byStratum.set(row.stratum, (byStratum.get(row.stratum) ?? 0) + 1); + expect(byStratum.get("first_time_author")).toBe(6); + expect(byStratum.get("merge_arm")).toBe(15); + expect(byStratum.get("close_arm")).toBe(9); + expect(new Set(plan.map((r) => r.targetId)).size).toBe(30); // no duplicates ever + }); + + it("a thin stratum SPILLS into the rest instead of shrinking the week's sample", () => { + // No first-time authors and only 3 closes: their quotas spill to merges. + const pool = [...Array.from({ length: 50 }, (_, i) => candidate(i)), ...Array.from({ length: 3 }, (_, i) => candidate(100 + i, { verdict: "close", outcome: "closed" }))]; + const plan = planAuditSample(pool, 30, seededRng(7)); + expect(plan).toHaveLength(30); + expect(plan.filter((r) => r.stratum === "close_arm")).toHaveLength(3); + expect(plan.filter((r) => r.stratum === "first_time_author")).toHaveLength(0); + expect(plan.filter((r) => r.stratum === "merge_arm")).toHaveLength(27); + }); + + it("the spill tags a close candidate close_arm when the close quota is already spent", () => { + // 3 closes fill the close quota (k=10 -> quota 3); the 4th close arrives via the spill and must still be + // tagged by its own arm, not mislabeled. + const pool = [...Array.from({ length: 4 }, (_, i) => candidate(100 + i, { verdict: "close", outcome: "closed" })), ...Array.from({ length: 3 }, (_, i) => candidate(i))]; + const plan = planAuditSample(pool, 10, seededRng(3)); + expect(plan).toHaveLength(7); + expect(plan.filter((r) => r.stratum === "close_arm")).toHaveLength(4); + }); + + it("returns the whole pool (each once) when the pool is smaller than k, and [] for empty/k<=0", () => { + const pool = [candidate(1), candidate(2, { verdict: "close", outcome: "closed" })]; + expect(planAuditSample(pool, 30, seededRng(1))).toHaveLength(2); + expect(planAuditSample([], 30, seededRng(1))).toEqual([]); + expect(planAuditSample(pool, 0, seededRng(1))).toEqual([]); + }); + + it("a first-time author claimed by the first-time stratum is never double-drawn by an arm stratum", () => { + const pool = [candidate(1, { firstTimeAuthor: true }), candidate(2)]; + const plan = planAuditSample(pool, 2, seededRng(5)); + expect(plan).toHaveLength(2); + expect(plan.filter((r) => r.targetId === "o/r#1")).toHaveLength(1); + }); +}); + +describe("runDecisionAuditSample (#8830) — end-to-end over the real ledger", () => { + async function seedDecision(env: Env, n: number, verdict: "merge" | "close", outcome: "merged" | "closed", author: string, opts: { policyClose?: boolean; old?: boolean } = {}): Promise { + const at = opts.old ? "2026-01-01T00:00:00.000Z" : new Date().toISOString(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, summary, created_at) VALUES (?, 'o/r', ?, 'gate_decision', ?, 'gittensory-native', ?, ?)`, + ) + .bind(`gd-${n}`, `o/r#${n}`, verdict, opts.policyClose ? "policy_close:contributor_cap" : "success", at) + .run(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, created_at) VALUES (?, 'o/r', ?, 'pr_outcome', ?, 'gittensory-native', ?)`, + ) + .bind(`po-${n}`, `o/r#${n}`, outcome, at) + .run(); + await env.DB.prepare(`INSERT INTO pull_requests (repo_full_name, number, title, state, author_login) VALUES ('o/r', ?, 't', 'closed', ?)`).bind(n, author).run(); + } + + it("samples recent decided PRs into PENDING rows, tags first-time authors, and excludes policy closes + old decisions", async () => { + const env = createTestEnv(); + await seedDecision(env, 1, "merge", "merged", "veteran"); + await seedDecision(env, 2, "merge", "merged", "veteran"); // veteran has 2 PRs -> not first-time + await seedDecision(env, 3, "close", "closed", "newbie"); // newbie's only PR -> first-time + await seedDecision(env, 4, "close", "closed", "capped", { policyClose: true }); // enforcement — excluded + await seedDecision(env, 5, "merge", "merged", "old-timer", { old: true }); // outside the 7d window + const inserted = await runDecisionAuditSample(env, Date.now(), () => 0.5); + expect(inserted).toBe(3); + const rows = await env.DB.prepare("SELECT target_id, stratum, status, rubric_version FROM decision_audit_labels ORDER BY target_id").all<{ target_id: string; stratum: string; status: string; rubric_version: string }>(); + expect(rows.results!.map((r) => r.target_id)).toEqual(["o/r#1", "o/r#2", "o/r#3"]); + expect(rows.results!.find((r) => r.target_id === "o/r#3")!.stratum).toBe("first_time_author"); + for (const row of rows.results!) { + expect(row.status).toBe("pending"); + expect(row.rubric_version).toBe(DECISION_AUDIT_RUBRIC_VERSION); + } + }); + + it("NEVER double-samples a PR across runs (UNIQUE target) and returns 0 on an empty window", async () => { + const env = createTestEnv(); + await seedDecision(env, 1, "merge", "merged", "a"); + expect(await runDecisionAuditSample(env, Date.now(), () => 0.5)).toBe(1); + expect(await runDecisionAuditSample(env, Date.now(), () => 0.5)).toBe(0); // candidate query excludes sampled targets + const empty = createTestEnv(); + expect(await runDecisionAuditSample(empty, Date.now(), () => 0.5)).toBe(0); + }); + + it("a concurrent double-run never double-inserts (INSERT OR IGNORE on the UNIQUE target)", async () => { + const env = createTestEnv(); + await seedDecision(env, 1, "merge", "merged", "a"); + const [a, b] = await Promise.all([runDecisionAuditSample(env, Date.now(), () => 0.5), runDecisionAuditSample(env, Date.now(), () => 0.5)]); + expect(a + b).toBe(1); + const n = await env.DB.prepare("SELECT COUNT(*) AS n FROM decision_audit_labels").first<{ n: number }>(); + expect(n!.n).toBe(1); + }); + + it("a failing insert is logged and skipped without voiding the batch", async () => { + const env = createTestEnv(); + await seedDecision(env, 1, "merge", "merged", "a"); + const { vi } = await import("vitest"); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const realPrepare = env.DB.prepare.bind(env.DB); + vi.spyOn(env.DB, "prepare").mockImplementation((sql: string) => { + if (sql.includes("INSERT OR IGNORE INTO decision_audit_labels")) throw new Error("disk full"); + return realPrepare(sql); + }); + expect(await runDecisionAuditSample(env, Date.now(), () => 0.5)).toBe(0); + expect(warn).toHaveBeenCalled(); + vi.restoreAllMocks(); + }); + + it("caps a huge week at DECISION_AUDIT_SAMPLE_SIZE", async () => { + const env = createTestEnv(); + for (let i = 1; i <= DECISION_AUDIT_SAMPLE_SIZE + 20; i += 1) await seedDecision(env, i, i % 2 ? "merge" : "close", i % 2 ? "merged" : "closed", `author${i}`); + expect(await runDecisionAuditSample(env, Date.now(), seededRng(9))).toBe(DECISION_AUDIT_SAMPLE_SIZE); + }); + + it("a PR with no cached pull_requests row degrades to not-first-time instead of guessing", async () => { + const env = createTestEnv(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, summary, created_at) VALUES ('gd-9', 'o/r', 'o/r#9', 'gate_decision', 'merge', 'gittensory-native', 'success', ?)`, + ).bind(new Date().toISOString()).run(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, created_at) VALUES ('po-9', 'o/r', 'o/r#9', 'pr_outcome', 'merged', 'gittensory-native', ?)`, + ).bind(new Date().toISOString()).run(); + expect(await runDecisionAuditSample(env, Date.now(), () => 0.5)).toBe(1); + const row = await env.DB.prepare("SELECT stratum FROM decision_audit_labels WHERE target_id = 'o/r#9'").first<{ stratum: string }>(); + expect(row!.stratum).toBe("merge_arm"); + }); +}); + +describe("decision-audit-sample job dispatch (#8830)", () => { + it("flag-ON runs the sample; flag-OFF (or a stale queued job after a flip) does ZERO work", async () => { + const env = createTestEnv({ LOOPOVER_DECISION_AUDIT: "true" }); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, summary, created_at) VALUES ('gd-1', 'o/r', 'o/r#1', 'gate_decision', 'merge', 'gittensory-native', 'success', ?)`, + ).bind(new Date().toISOString()).run(); + await env.DB.prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, created_at) VALUES ('po-1', 'o/r', 'o/r#1', 'pr_outcome', 'merged', 'gittensory-native', ?)`, + ).bind(new Date().toISOString()).run(); + await processJob(env, { type: "decision-audit-sample", requestedBy: "test" }); + const n = await env.DB.prepare("SELECT COUNT(*) AS n FROM decision_audit_labels").first<{ n: number }>(); + expect(n!.n).toBe(1); + + const off = createTestEnv(); + await env.DB.prepare("DELETE FROM decision_audit_labels").run(); + await processJob(off, { type: "decision-audit-sample", requestedBy: "test" }); + const zero = await off.DB.prepare("SELECT COUNT(*) AS n FROM decision_audit_labels").first<{ n: number }>(); + expect(zero!.n).toBe(0); + }); +}); diff --git a/test/unit/index.test.ts b/test/unit/index.test.ts index 8ff4d42f82..b0cba77279 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -1145,6 +1145,26 @@ describe("worker entrypoint", () => { expect(sent.some((m) => m.type === "repo-doc-refresh-sweep")).toBe(false); }); + it("#8830: enqueues the decision-audit sample only in the Tuesday 08:00 window, flag-ON, on a self-host", async () => { + const send = (over: Record, when: string) => { + const sent: Array = []; + const env = createTestEnv({ + JOBS: { async send(message: import("../../src/types").JobMessage) { sent.push(message); } } as unknown as Queue, + ...over, + }); + const waitUntil: Promise[] = []; + return worker.scheduled(controllerFor(when), env, executionContext(waitUntil)).then(() => Promise.all(waitUntil)).then(() => sent); + }; + const selfhost = { SELFHOST_TRANSIENT_CACHE: {} as never, LOOPOVER_DECISION_AUDIT: "true" }; + // 2026-06-02 is a Tuesday. + expect((await send(selfhost, "2026-06-02T08:00:00.000Z")).some((m) => m.type === "decision-audit-sample")).toBe(true); + // Wrong hour, wrong day, flag off, and cloud runtime each suppress the enqueue. + expect((await send(selfhost, "2026-06-02T09:00:00.000Z")).some((m) => m.type === "decision-audit-sample")).toBe(false); + expect((await send(selfhost, "2026-06-01T08:00:00.000Z")).some((m) => m.type === "decision-audit-sample")).toBe(false); + expect((await send({ ...selfhost, LOOPOVER_DECISION_AUDIT: "false" }, "2026-06-02T08:00:00.000Z")).some((m) => m.type === "decision-audit-sample")).toBe(false); + expect((await send({ LOOPOVER_DECISION_AUDIT: "true", SELFHOST_TRANSIENT_CACHE: undefined }, "2026-06-02T08:00:00.000Z")).some((m) => m.type === "decision-audit-sample")).toBe(false); + }); + it("enqueues weekly value report generation during the Monday report window", async () => { const sent: Array = []; const env = createTestEnv({ From 912c74b8c54fcf53aa4f3d09892c502ccccad481 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 04:07:07 -0700 Subject: [PATCH 2/5] feat(orb): admit the holdout_close stratum in decision_audit_labels upfront The randomized close-holdout (#8831) lands its adjudications in the same label store; widening the CHECK now spares that PR a SQLite table rebuild for one enum value. --- migrations/0178_decision_audit_labels.sql | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/migrations/0178_decision_audit_labels.sql b/migrations/0178_decision_audit_labels.sql index 80425e6979..2544f2f084 100644 --- a/migrations/0178_decision_audit_labels.sql +++ b/migrations/0178_decision_audit_labels.sql @@ -11,7 +11,8 @@ CREATE TABLE IF NOT EXISTS decision_audit_labels ( target_id TEXT NOT NULL, -- owner/repo#N verdict TEXT NOT NULL CHECK (verdict IN ('merge', 'close')), -- the gate's decision at sample time outcome TEXT NOT NULL CHECK (outcome IN ('merged', 'closed')), -- realized outcome at sample time - stratum TEXT NOT NULL CHECK (stratum IN ('merge_arm', 'close_arm', 'first_time_author')), + -- holdout_close (#8831): rows sourced by the randomized close-holdout rather than the weekly draw. + stratum TEXT NOT NULL CHECK (stratum IN ('merge_arm', 'close_arm', 'first_time_author', 'holdout_close')), rubric_version TEXT NOT NULL, -- adjudications are only comparable within a version sampled_at TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'adjudicated')), From 68696920ee8f61a99c51c6a4c064a89557c5c034 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 04:07:31 -0700 Subject: [PATCH 3/5] feat(orb): make decision_audit_labels.outcome nullable for holdout rows A holdout_close row (#8831) is created while the PR is still held -- no realized outcome exists yet; the human adjudication is its ground truth. The weekly sampler always supplies an outcome. --- migrations/0178_decision_audit_labels.sql | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/migrations/0178_decision_audit_labels.sql b/migrations/0178_decision_audit_labels.sql index 2544f2f084..e92230c67c 100644 --- a/migrations/0178_decision_audit_labels.sql +++ b/migrations/0178_decision_audit_labels.sql @@ -10,7 +10,9 @@ CREATE TABLE IF NOT EXISTS decision_audit_labels ( project TEXT NOT NULL, -- owner/repo target_id TEXT NOT NULL, -- owner/repo#N verdict TEXT NOT NULL CHECK (verdict IN ('merge', 'close')), -- the gate's decision at sample time - outcome TEXT NOT NULL CHECK (outcome IN ('merged', 'closed')), -- realized outcome at sample time + -- Nullable: a holdout_close row (#8831) is created while the PR is still HELD -- there is no realized + -- outcome yet; the human adjudication IS its ground truth. The weekly sampler always supplies one. + outcome TEXT CHECK (outcome IN ('merged', 'closed')), -- holdout_close (#8831): rows sourced by the randomized close-holdout rather than the weekly draw. stratum TEXT NOT NULL CHECK (stratum IN ('merge_arm', 'close_arm', 'first_time_author', 'holdout_close')), rubric_version TEXT NOT NULL, -- adjudications are only comparable within a version From 0b010ccbe5cfd35bb09b4715e0d977cb632132f3 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 04:15:18 -0700 Subject: [PATCH 4/5] =?UTF-8?q?fix(orb):=20claim=20the=20adjudication=20at?= =?UTF-8?q?omically=20=E2=80=94=20the=20status=20predicate=20rides=20the?= =?UTF-8?q?=20UPDATE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Superagent P1 on this PR, and it is right: select-then-update lets two concurrent adjudications both pass the pending check and both write, silently violating the one-adjudication-per-label guarantee. The predicate now rides the UPDATE (WHERE status = 'pending'); zero changes disambiguates 404 vs 409 with a read that no longer guards anything. Regression test races two concurrent adjudications: exactly one 200, one 409, one surviving label. Also binds LOOPOVER_NATIVE_SOURCE instead of a hardcoded source literal (branding-drift check). --- src/api/routes.ts | 18 +++++++++++++----- src/review/decision-audit.ts | 5 +++-- test/integration/audit-labels-routes.test.ts | 15 +++++++++++++++ 3 files changed, 31 insertions(+), 7 deletions(-) diff --git a/src/api/routes.ts b/src/api/routes.ts index ef4fc4a7d1..6582f4525d 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -4421,14 +4421,22 @@ export function createApp() { return c.json({ error: "id and adjudication (correct|incorrect|uncertain) required" }, 400); } const reasonCategory = typeof payload?.reasonCategory === "string" ? payload.reasonCategory.slice(0, 100) : null; - const existing = await c.env.DB.prepare("SELECT status FROM decision_audit_labels WHERE id = ?").bind(id).first<{ status: string }>(); - if (!existing) return c.json({ error: "label_not_found" }, 404); - if (existing.status === "adjudicated") return c.json({ error: "already_adjudicated" }, 409); - await c.env.DB.prepare( - `UPDATE decision_audit_labels SET status = 'adjudicated', adjudication = ?, reason_category = ?, adjudicated_at = ? WHERE id = ?`, + // Atomic claim: the status predicate rides the UPDATE itself, so two concurrent adjudications can never + // both win — a select-then-update here would let both pass the pending check and silently violate the + // one-adjudication-per-label guarantee (labels are calibration data; rewrites must be impossible, not + // merely discouraged). meta.changes disambiguates afterwards: 0 changes is either "no such label" (404) + // or "someone else already adjudicated it" (409), resolved by one read that no longer guards anything. + const result = await c.env.DB.prepare( + `UPDATE decision_audit_labels SET status = 'adjudicated', adjudication = ?, reason_category = ?, adjudicated_at = ? + WHERE id = ? AND status = 'pending'`, ) .bind(adjudication, reasonCategory, new Date().toISOString(), id) .run(); + if (result.meta.changes === 0) { + const existing = await c.env.DB.prepare("SELECT status FROM decision_audit_labels WHERE id = ?").bind(id).first<{ status: string }>(); + if (!existing) return c.json({ error: "label_not_found" }, 404); + return c.json({ error: "already_adjudicated" }, 409); + } return c.json({ id, adjudication, reasonCategory }); }); diff --git a/src/review/decision-audit.ts b/src/review/decision-audit.ts index 801024588c..c6f50d881e 100644 --- a/src/review/decision-audit.ts +++ b/src/review/decision-audit.ts @@ -10,6 +10,7 @@ // // Flag-gated by LOOPOVER_DECISION_AUDIT (default OFF — the weekly job is never enqueued, zero new work, // byte-identical to today), mirroring every sibling cron flag (selftune-wire et al). +import { LOOPOVER_NATIVE_SOURCE } from "./parity-wire"; import { errorMessage, nowIso } from "../utils/json"; /** Bump ONLY with a corresponding edit to docs/decision-audit-rubric.md — adjudications are comparable only @@ -110,7 +111,7 @@ export async function runDecisionAuditSample(env: Env, nowMs: number = Date.now( SELECT target_id, project, decision AS verdict, summary, created_at, ROW_NUMBER() OVER (PARTITION BY target_id ORDER BY created_at DESC) AS rn FROM review_audit - WHERE event_type = 'gate_decision' AND decision IN ('merge', 'close') AND source = 'gittensory-native' + WHERE event_type = 'gate_decision' AND decision IN ('merge', 'close') AND source = ? ), po AS ( SELECT target_id, decision AS outcome, @@ -126,7 +127,7 @@ export async function runDecisionAuditSample(env: Env, nowMs: number = Date.now( AND (gd.summary IS NULL OR gd.summary NOT LIKE 'policy_close:%') AND NOT EXISTS (SELECT 1 FROM decision_audit_labels dal WHERE dal.target_id = gd.target_id)`, ) - .bind(sinceIso) + .bind(LOOPOVER_NATIVE_SOURCE, sinceIso) .all<{ targetId: string; project: string; verdict: "merge" | "close"; outcome: "merged" | "closed" }>(); const decided = results; diff --git a/test/integration/audit-labels-routes.test.ts b/test/integration/audit-labels-routes.test.ts index e5d5e66a42..ec4259046a 100644 --- a/test/integration/audit-labels-routes.test.ts +++ b/test/integration/audit-labels-routes.test.ts @@ -57,4 +57,19 @@ describe("decision-audit adjudication routes (/v1/internal/audit-labels)", () => const uncertain = await app.request("/v1/internal/audit-labels/adjudicate", { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:o/r#2", adjudication: "uncertain" }) }, env); expect(uncertain.status).toBe(200); }); + + it("REGRESSION (Superagent P1): concurrent adjudications race atomically — exactly one wins, the other 409s", async () => { + const env = createTestEnv(); + await seedLabel(env, "audit:o/r#9"); + const fire = () => + app.request( + "/v1/internal/audit-labels/adjudicate", + { method: "POST", headers: auth, body: JSON.stringify({ id: "audit:o/r#9", adjudication: "correct" }) }, + env, + ); + const [a, b] = await Promise.all([fire(), fire()]); + expect([a.status, b.status].sort()).toEqual([200, 409]); + const row = await env.DB.prepare("SELECT adjudication FROM decision_audit_labels WHERE id = 'audit:o/r#9'").first<{ adjudication: string }>(); + expect(row!.adjudication).toBe("correct"); + }); }); From 881b53a7953b495e1fb768b06530bb8925b6e077 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 04:46:37 -0700 Subject: [PATCH 5/5] fix(orb): count only DECIDED PRs in first-time-author detection Review blocker on this PR, and it is right: the author count read every cached pull_requests row, so a newcomer with several open-but-undecided PRs and exactly one decided PR was wrongly marked veteran -- starving the stratum whose whole purpose is weak-prior newcomers. The count now requires a realized pr_outcome per counted PR, with a regression test seeding open PRs that must not inflate the total. --- src/review/decision-audit.ts | 14 ++++++++++---- test/unit/decision-audit.test.ts | 11 ++++++++++- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/src/review/decision-audit.ts b/src/review/decision-audit.ts index c6f50d881e..73c7e7765d 100644 --- a/src/review/decision-audit.ts +++ b/src/review/decision-audit.ts @@ -133,11 +133,17 @@ export async function runDecisionAuditSample(env: Env, nowMs: number = Date.now( const decided = results; if (decided.length === 0) return 0; - // First-time detection: the author's earliest decided PR. review_audit carries no logins by design, so the - // author comes from the pull_requests cache; a PR with no cached row degrades to "not first-time" (the - // stratum under-fills and the spill covers it) rather than guessing. + // First-time detection: the author's DECIDED-PR count. review_audit carries no logins by design, so the + // author comes from the pull_requests cache — but only rows whose target actually has a realized + // pr_outcome count, otherwise a newcomer with several open-but-undecided PRs and exactly one decided PR + // would be wrongly marked veteran, starving the stratum whose whole purpose is weak-prior newcomers. A PR + // with no cached row degrades to "not first-time" (the stratum under-fills and the spill covers it) + // rather than guessing. const authorRows = await env.DB.prepare( - `SELECT repo_full_name || '#' || number AS targetId, author_login AS author FROM pull_requests WHERE author_login IS NOT NULL`, + `SELECT p.repo_full_name || '#' || p.number AS targetId, p.author_login AS author + FROM pull_requests p + WHERE p.author_login IS NOT NULL + AND EXISTS (SELECT 1 FROM review_audit ra WHERE ra.event_type = 'pr_outcome' AND ra.target_id = p.repo_full_name || '#' || p.number)`, ).all<{ targetId: string; author: string }>(); const authorByTarget = new Map(authorRows.results.map((r) => [r.targetId, r.author.toLowerCase()])); const decidedCountByAuthor = new Map(); diff --git a/test/unit/decision-audit.test.ts b/test/unit/decision-audit.test.ts index f0bfe94862..d0fd713887 100644 --- a/test/unit/decision-audit.test.ts +++ b/test/unit/decision-audit.test.ts @@ -93,6 +93,11 @@ describe("planAuditSample (#8830)", () => { }); describe("runDecisionAuditSample (#8830) — end-to-end over the real ledger", () => { + async function seedOpenPr(env: Env, n: number, author: string): Promise { + // A cached PR with NO pr_outcome — open/undecided. Must never count toward an author's decided total. + await env.DB.prepare(`INSERT INTO pull_requests (repo_full_name, number, title, state, author_login) VALUES ('o/r', ?, 't', 'open', ?)`).bind(n, author).run(); + } + async function seedDecision(env: Env, n: number, verdict: "merge" | "close", outcome: "merged" | "closed", author: string, opts: { policyClose?: boolean; old?: boolean } = {}): Promise { const at = opts.old ? "2026-01-01T00:00:00.000Z" : new Date().toISOString(); await env.DB.prepare( @@ -112,7 +117,11 @@ describe("runDecisionAuditSample (#8830) — end-to-end over the real ledger", ( const env = createTestEnv(); await seedDecision(env, 1, "merge", "merged", "veteran"); await seedDecision(env, 2, "merge", "merged", "veteran"); // veteran has 2 PRs -> not first-time - await seedDecision(env, 3, "close", "closed", "newbie"); // newbie's only PR -> first-time + await seedDecision(env, 3, "close", "closed", "newbie"); // newbie's only DECIDED PR -> first-time + // REGRESSION (review blocker on this PR): open/undecided PRs must not inflate the decided count — a + // newcomer with several open PRs and one decided PR is still first-time. + await seedOpenPr(env, 300, "newbie"); + await seedOpenPr(env, 301, "newbie"); await seedDecision(env, 4, "close", "closed", "capped", { policyClose: true }); // enforcement — excluded await seedDecision(env, 5, "merge", "merged", "old-timer", { old: true }); // outside the 7d window const inserted = await runDecisionAuditSample(env, Date.now(), () => 0.5);