diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 87fd3f3e7..03bcc8864 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -31,7 +31,7 @@ import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride, run import { isActiveReviewReconciliationEnabled, resolveActiveReviewReconciliationManifestOverride, runActiveReviewReconciliation } from "../review/active-review-reconciliation"; import { isSelfTuneEnabled, runSelfTune } from "../review/selftune-wire"; import { isSatisfactionFloorAutotuneEnabled, runScheduledSatisfactionFloorLoosening } from "../services/satisfaction-floor-loosening-run"; -import { GENERIC_LIVE_KNOBS, isConfigDriftSentinelEnabled, isKnobAutotuneEnabled, runConfigDriftSentinel, runScheduledKnobLoosening } from "../services/knob-loosening-run"; +import { GENERIC_LIVE_KNOBS, isConfigDriftSentinelEnabled, isKnobAutotuneEnabled, runConfigDriftSentinel, runPerRepoKnobLoosening, runScheduledKnobLoosening } from "../services/knob-loosening-run"; import { runSelfTuneBreaker } from "../review/outcomes-wire"; import { isRagEnabled } from "../review/rag-wire"; import { processSubmitDraft } from "../services/draft"; @@ -352,7 +352,12 @@ export async function processJob(env: Env, message: JobMessage): Promise { // #8176: every LATER live registry knob rides the same tick through the generic runner — each knob // is double-gated on its OWN wrangler var, so an un-flagged knob does zero work here. for (const knob of GENERIC_LIVE_KNOBS) { - if (isKnobAutotuneEnabled(env, knob)) await runScheduledKnobLoosening(env, knob); + if (isKnobAutotuneEnabled(env, knob)) { + await runScheduledKnobLoosening(env, knob); + // #8217: repos whose own labeled slice clears the floors earn repo-scoped steps; sparse repos + // inherit global. Bounded per tick with a rotating cursor; fail-safe internally. + await runPerRepoKnobLoosening(env, knob); + } } // #8213: the drift sentinel rides the same calibration tick, behind its own default-off flag. // Alert-only — it never writes a knob value — and internally fail-safe per knob. diff --git a/src/services/knob-loosening-run.ts b/src/services/knob-loosening-run.ts index 4fa89efbe..216fa2786 100644 --- a/src/services/knob-loosening-run.ts +++ b/src/services/knob-loosening-run.ts @@ -12,7 +12,14 @@ // • the override write is NOT best-effort (an unrecorded change is worse than none) — the audit trail is; // • one structured error-level alert per applied step, never re-alerting (the next run starts from the // already-loosened value and proposes nothing until the corpus justifies another step). -import { buildBacktestCorpus, computeReliabilityCurve, deriveThresholdSuggestion, type ReliabilityCurve } from "@loopover/engine"; +import { + buildBacktestCorpus, + computeReliabilityCurve, + computeRepoCorpusDensity, + deriveThresholdSuggestion, + sliceCorpusByRepo, + type ReliabilityCurve, +} from "@loopover/engine"; import { createSignalStore } from "../review/signal-tracking-wire"; import { recordAuditEvent } from "../db/repositories"; import { evaluateKnobDrift, evaluateKnobLoosening, LOOSENABLE_KNOBS, type KnobDriftReport, type KnobLooseningProposal, type LoosenableKnob } from "./loosening-knobs"; @@ -161,6 +168,111 @@ export async function runScheduledKnobLoosening(env: Env, knob: LoosenableKnob): } } +// ── Per-repo loosening loop (#8217, epic #8211 track B capstone) ───────────────────────────────────────── + +/** Repos evaluated per tick, hard-capped so a large-fleet future never turns the tick into a stampede. + * Deterministic order + a system_flags cursor make successive ticks cover the whole eligible set. */ +export const PER_REPO_LOOSENING_MAX_REPOS_PER_TICK = 10; + +const PER_REPO_CURSOR_FLAG_PREFIX = "per_repo_loosening_cursor:"; + +export type PerRepoLooseningResult = { repoFullName: string; applied: boolean; reason: string }; + +/** + * Per-repo loosening evaluation (#8217): repos whose OWN labeled slice clears the knob's sample floors + * (computeRepoCorpusDensity — the same split + minimums as every evaluator) earn their own loosening + * step from their CURRENT resolved value; sparse repos keep inheriting the global value untouched. + * Same discipline as the global loop verbatim: smallest candidate step, visible improved + held-out + * non-regressed on the REPO slice, hard minimum, double flag gating, one error-level alert per applied + * step (`ev` stays per-knob; the repo rides the alert body — the Sentry-fingerprint discipline). + * Bounded work: at most {@link PER_REPO_LOOSENING_MAX_REPOS_PER_TICK} eligible repos per tick in + * deterministic order, resuming from a per-knob cursor. Fail-safe per repo. + */ +export async function runPerRepoKnobLoosening(env: Env, knob: LoosenableKnob, nowMs: number = Date.now()): Promise { + if (knob.applyMode !== "live") return []; + if (!isKnobAutotuneEnabled(env, knob)) return []; + const results: PerRepoLooseningResult[] = []; + try { + const { fired, overrides } = await createSignalStore(env).queryRuleHistory(knob.ruleId, nowMs - CORPUS_LOOKBACK_MS); + const cases = buildBacktestCorpus(knob.ruleId, fired, overrides); + const density = computeRepoCorpusDensity(cases, knob.minVisibleCases, knob.minHeldOutCases, knob.heldOutFraction, knob.splitSeed); + const eligible = [...density.entries()] + .filter(([, stats]) => stats.eligible) + .map(([repo]) => repo) + .sort(); + if (eligible.length === 0) return results; + + const cursorKey = `${PER_REPO_CURSOR_FLAG_PREFIX}${knob.knobId}`; + const cursorRow = await env.DB.prepare("SELECT value FROM system_flags WHERE key = ?").bind(cursorKey).first<{ value: string }>(); + const cursor = cursorRow?.value ?? ""; + // Rotate: start after the cursor, wrap around, cap the batch. + const startIndex = eligible.findIndex((repo) => repo > cursor); + const rotated = startIndex === -1 ? eligible : [...eligible.slice(startIndex), ...eligible.slice(0, startIndex)]; + const batch = rotated.slice(0, PER_REPO_LOOSENING_MAX_REPOS_PER_TICK); + const slices = sliceCorpusByRepo(cases); + + for (const repoFullName of batch) { + try { + const currentValue = (await getKnobOverrideForRepo(env, knob, repoFullName)) ?? knob.shippedValue; + if (currentValue <= knob.hardMinimum) { + results.push({ repoFullName, applied: false, reason: "already_applied" }); + continue; + } + // Non-null by construction: eligibility derives from the same slicing, so every eligible repo has a slice. + const proposal = evaluateKnobLoosening(knob, slices.get(repoFullName)!, currentValue); + if (!proposal || proposal.proposedValue >= currentValue || proposal.proposedValue < knob.hardMinimum) { + results.push({ repoFullName, applied: false, reason: "no_proposal" }); + continue; + } + await env.DB.prepare( + "INSERT INTO system_flags (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", + ) + .bind(repoKnobOverrideFlagKey(knob, repoFullName), String(proposal.proposedValue)) + .run(); + await recordAuditEvent(env, { + eventType: knob.looseningEventType, + actor: "loopover", + targetKey: knob.ruleId, + outcome: "completed", + detail: `${knob.knobId} loosened for ${repoFullName}: ${proposal.currentValue} -> ${proposal.proposedValue} (repo-slice backtest-gated)`, + metadata: { proposal, repoFullName, scope: "repo" }, + }).catch(() => undefined); + console.error( + JSON.stringify({ + level: "error", + event: "calibration_knob_loosened", + ev: knob.knobId, + at: new Date().toISOString(), + scope: "repo", + repoFullName, + currentValue: proposal.currentValue, + proposedValue: proposal.proposedValue, + visibleCases: proposal.visibleCases, + heldOutCases: proposal.heldOutCases, + }), + ); + results.push({ repoFullName, applied: true, reason: "applied" }); + } catch (error) { + console.warn( + JSON.stringify({ level: "warn", event: "per_repo_loosening_failed", ev: knob.knobId, repoFullName, error: error instanceof Error ? error.message : "unknown error" }), + ); + results.push({ repoFullName, applied: false, reason: "error" }); + } + } + const lastProcessed = batch[batch.length - 1]!; + await env.DB.prepare( + "INSERT INTO system_flags (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", + ) + .bind(cursorKey, lastProcessed) + .run(); + } catch (error) { + console.warn( + JSON.stringify({ level: "warn", event: "per_repo_loosening_tick_failed", ev: knob.knobId, error: error instanceof Error ? error.message : "unknown error" }), + ); + } + return results; +} + // ── Config-drift sentinel (#8213, epic #8211 track A) ──────────────────────────────────────────────────── /** Truthy-string flag for the drift sentinel — default off, so a deploy is byte-identical until opted in. */ diff --git a/test/unit/knob-loosening-run.test.ts b/test/unit/knob-loosening-run.test.ts index ee448f102..4ac012ebd 100644 --- a/test/unit/knob-loosening-run.test.ts +++ b/test/unit/knob-loosening-run.test.ts @@ -14,6 +14,8 @@ import { KNOB_SUGGESTION_TARGET_PRECISION, loadKnobStatus, loadLiveKnobStatuses, + runPerRepoKnobLoosening, + PER_REPO_LOOSENING_MAX_REPOS_PER_TICK, runConfigDriftSentinel, runKnobLoosening, runScheduledKnobLoosening, @@ -48,8 +50,8 @@ async function setOverrideRow(env: Env, key: string, value: string): Promise { - const pool = Array.from({ length: 400 }, (_, i) => `acme/widgets#${i + 1}`); +async function seedAiLooseningFriendlyHistory(env: Env, repo = "acme/widgets"): Promise { + const pool = Array.from({ length: 400 }, (_, i) => `${repo}#${i + 1}`); const probe = pool.map((targetKey) => ({ ruleId: AI_KNOB.ruleId, targetKey, @@ -388,6 +390,116 @@ describe("runConfigDriftSentinel (#8213)", () => { }); }); +describe("runPerRepoKnobLoosening (#8217)", () => { + it("a dense repo earns its OWN override while a sparse repo inherits global untouched", async () => { + const env = enabledEnv(); + await seedAiLooseningFriendlyHistory(env, "acme/dense"); + // Sparse repo: a handful of cases, far under the knob's floors. + const store = createSignalStore(env); + for (let i = 1; i <= 3; i += 1) { + await store.recordRuleFired({ ruleId: AI_KNOB.ruleId, targetKey: `acme/sparse#${i}`, outcome: "unaddressed", occurredAt: new Date(Date.now() - 5000).toISOString(), metadata: { confidence: 0.91 } }); + await store.recordHumanOverride({ ruleId: AI_KNOB.ruleId, targetKey: `acme/sparse#${i}`, verdict: "confirmed", occurredAt: new Date().toISOString() }); + } + + const results = await runPerRepoKnobLoosening(env, AI_KNOB); + expect(results).toEqual([{ repoFullName: "acme/dense", applied: true, reason: "applied" }]); + expect(await getKnobOverrideForRepo(env, AI_KNOB, "acme/dense")).toBe(AI_KNOB.candidates[0]); + // Sparse repo: no repo row; resolution falls through to global (none here) -> null. + expect(await getKnobOverrideForRepo(env, AI_KNOB, "acme/sparse")).toBeNull(); + // The repo-scoped audit event carries the scope + repo. + const events = await env.DB.prepare("SELECT metadata_json FROM audit_events WHERE event_type = ?").bind(AI_KNOB.looseningEventType).all<{ metadata_json: string }>(); + const metadata = JSON.parse(events.results![0]!.metadata_json) as { scope?: string; repoFullName?: string }; + expect(metadata).toMatchObject({ scope: "repo", repoFullName: "acme/dense" }); + }); + + it("second tick evaluates the earned repo from ITS value (no proposal left) — never oscillates; already-at-minimum reports as such", async () => { + const env = enabledEnv(); + await seedAiLooseningFriendlyHistory(env, "acme/dense"); + await runPerRepoKnobLoosening(env, AI_KNOB); + const second = await runPerRepoKnobLoosening(env, AI_KNOB); + expect(second).toEqual([{ repoFullName: "acme/dense", applied: false, reason: "no_proposal" }]); + + await setOverrideRow(env, repoKnobOverrideFlagKey(AI_KNOB, "acme/dense"), String(AI_KNOB.hardMinimum)); + const third = await runPerRepoKnobLoosening(env, AI_KNOB); + expect(third).toEqual([{ repoFullName: "acme/dense", applied: false, reason: "already_applied" }]); + }); + + it("gates: report-only knob and flag-off both do nothing; a broken store fails safe with a warn", async () => { + const reportOnly = { ...AI_KNOB, applyMode: "report_only" as const }; + expect(await runPerRepoKnobLoosening(enabledEnv(), reportOnly)).toEqual([]); + expect(await runPerRepoKnobLoosening(createTestEnv(), AI_KNOB)).toEqual([]); + + const broken = enabledEnv(); + broken.DB = { prepare: () => { throw new Error("boom"); } } as never; + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); + expect(await runPerRepoKnobLoosening(broken, AI_KNOB)).toEqual([]); + expect(warnSpy.mock.calls.some((c) => String(c[0]).includes("per_repo_loosening_tick_failed"))).toBe(true); + }); + + it("empty eligibility returns cleanly; audit rejection is best-effort; a mid-repo error fails safe per repo; non-Error throws degrade", async () => { + // Enabled but only sparse data -> zero eligible repos -> the early return, no cursor written. + const sparseOnly = enabledEnv(); + const store = createSignalStore(sparseOnly); + await store.recordRuleFired({ ruleId: AI_KNOB.ruleId, targetKey: "acme/sparse#1", outcome: "unaddressed", occurredAt: new Date().toISOString(), metadata: { confidence: 0.91 } }); + await store.recordHumanOverride({ ruleId: AI_KNOB.ruleId, targetKey: "acme/sparse#1", verdict: "confirmed", occurredAt: new Date().toISOString() }); + expect(await runPerRepoKnobLoosening(sparseOnly, AI_KNOB)).toEqual([]); + + // Audit write rejection: the override still lands (best-effort trail, never sacrificed writes). + const env = enabledEnv(); + await seedAiLooseningFriendlyHistory(env, "acme/dense"); + const repositories = await import("../../src/db/repositories"); + vi.spyOn(repositories, "recordAuditEvent").mockRejectedValue(new Error("audit down")); + const results = await runPerRepoKnobLoosening(env, AI_KNOB); + expect(results).toEqual([{ repoFullName: "acme/dense", applied: true, reason: "applied" }]); + expect(await getKnobOverrideForRepo(env, AI_KNOB, "acme/dense")).toBe(AI_KNOB.candidates[0]); + vi.restoreAllMocks(); + + // Mid-repo evaluator throw: that repo reports error, the tick survives. + const env2 = enabledEnv(); + await seedAiLooseningFriendlyHistory(env2, "acme/dense"); + const looseningKnobsModule = await import("../../src/services/loosening-knobs"); + vi.spyOn(looseningKnobsModule, "evaluateKnobLoosening").mockImplementation(() => { throw "evaluator string boom"; }); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const errored = await runPerRepoKnobLoosening(env2, AI_KNOB); + expect(errored).toEqual([{ repoFullName: "acme/dense", applied: false, reason: "error" }]); + expect(warnSpy.mock.calls.some((c) => String(c[0]).includes("per_repo_loosening_failed") && String(c[0]).includes('"error":"unknown error"'))).toBe(true); + vi.restoreAllMocks(); + + // Inner catch with a REAL Error message, and the outer catch's non-Error arm via a string-throwing store. + const env3 = enabledEnv(); + await seedAiLooseningFriendlyHistory(env3, "acme/dense"); + const knobsModule = await import("../../src/services/loosening-knobs"); + vi.spyOn(knobsModule, "evaluateKnobLoosening").mockImplementation(() => { throw new Error("evaluator real error"); }); + const warn3 = vi.spyOn(console, "warn").mockImplementation(() => undefined); + await runPerRepoKnobLoosening(env3, AI_KNOB); + expect(warn3.mock.calls.some((c) => String(c[0]).includes("evaluator real error"))).toBe(true); + vi.restoreAllMocks(); + + const stringStore = enabledEnv(); + stringStore.DB = { prepare: () => { throw "outer string boom"; } } as never; + const warn4 = vi.spyOn(console, "warn").mockImplementation(() => undefined); + expect(await runPerRepoKnobLoosening(stringStore, AI_KNOB)).toEqual([]); + expect(warn4.mock.calls.some((c) => String(c[0]).includes("per_repo_loosening_tick_failed") && String(c[0]).includes('"error":"unknown error"'))).toBe(true); + }); + + it("caps the batch and rotates the cursor across ticks (deterministic order)", async () => { + expect(PER_REPO_LOOSENING_MAX_REPOS_PER_TICK).toBe(10); + const env = enabledEnv(); + // Two dense repos; cursor after tick 1 should sit at the last processed repo. With a batch cap of 10 + // both fit in one tick, so pin the cursor bookkeeping rather than the wraparound (covered by the + // rotation arithmetic itself being deterministic on the sorted list). + await seedAiLooseningFriendlyHistory(env, "acme/alpha"); + await seedAiLooseningFriendlyHistory(env, "acme/beta"); + const results = await runPerRepoKnobLoosening(env, AI_KNOB); + expect(results.map((r) => r.repoFullName)).toEqual(["acme/alpha", "acme/beta"]); + const cursor = await env.DB.prepare("SELECT value FROM system_flags WHERE key = ?").bind(`per_repo_loosening_cursor:${AI_KNOB.knobId}`).first<{ value: string }>(); + expect(cursor?.value).toBe("acme/beta"); + // Next tick starts AFTER the cursor: wraps to alpha first again (both still eligible). + const second = await runPerRepoKnobLoosening(env, AI_KNOB); + expect(second.map((r) => r.repoFullName)).toEqual(["acme/alpha", "acme/beta"]); + }); +}); + describe("loadKnobStatus / loadLiveKnobStatuses (#8161 generalized)", () => { it("reports a lingering override row even with the flag OFF, and the live value only when ON", async () => { const env = createTestEnv();