fix(ops): block concentrated district pending spikes in auto-publish

README.md

@@ -125,7 +125,7 @@ flowchart TD
 - PostgreSQL is the canonical store for runs, artifacts, subscriptions, and publication state.
 - S3 stores raw scrape evidence, normalized snapshot candidates, release evidence, and outreach archives.
 - Publish requires an operator action or a passing auto-publish gate.
-- Auto-publish validates fresh internal runs against quality and delta guardrails, publishes when safe, and pages via SNS when blocked.
+- Auto-publish validates fresh internal runs against quality and delta guardrails, publishes when safe, and pages via SNS when blocked. Lower-court sweeps also block concentrated single-district pending swings and large absolute pending moves that stay under the primary 20% fraction threshold.
 - A daily publish-pending sweep walks quality-complete runs per scope from the past 3 days and runs each through the same gate.
 - Published snapshot read models drive every public surface; rollback is one operator call.
 

src/ops/auto-publish-gate.ts

@@ -1,9 +1,35 @@
 export const DEFAULT_AUTO_PUBLISH_DELTA_THRESHOLD = 0.2;
 
+/**
+ * Secondary absolute+fractional floor for state pending moves that stay under
+ * the primary 20% threshold but are still large enough to hide a concentrated
+ * reporting correction (e.g. Gujarat Jul 9→12: +8.8% / +168k).
+ */
+export const DEFAULT_AUTO_PUBLISH_ABS_DELTA_THRESHOLD = 100_000;
+export const DEFAULT_AUTO_PUBLISH_ABS_FRACTION_FLOOR = 0.05;
+
+/**
+ * Lower-court district concentration gate. Blocks when one district both
+ * swings hard and accounts for most of the state-level pending move
+ * (e.g. Surat +60% driving nearly all of Gujarat's Jul 12 spike).
+ */
+export const DEFAULT_DISTRICT_DELTA_THRESHOLD = 0.4;
+export const DEFAULT_DISTRICT_ABS_DELTA_FLOOR = 25_000;
+export const DEFAULT_DISTRICT_STATE_SHARE_THRESHOLD = 0.5;
+
 export type AutoPublishSkipReason =
   | "quality_not_complete"
   | "current_pending_missing"
-  | "outlier_pending_delta";
+  | "outlier_pending_delta"
+  | "outlier_district_pending_delta";
+
+export interface DistrictPendingDelta {
+  districtId: string;
+  previousPending: number;
+  currentPending: number;
+  deltaFraction: number;
+  stateDeltaShare: number;
+}
 
 export interface AutoPublishDecision {
   publish: boolean;
@@ -13,17 +39,27 @@ export interface AutoPublishDecision {
   previousPending?: number;
   deltaFraction?: number;
   deltaThreshold: number;
+  districtDelta?: DistrictPendingDelta;
 }
 
 export interface EvaluateAutoPublishOptions {
   qualityState: string;
   currentPending?: number;
   previousPending?: number;
   deltaThreshold?: number;
+  absDeltaThreshold?: number;
+  absFractionFloor?: number;
+  currentDistrictPending?: Record<string, number>;
+  previousDistrictPending?: Record<string, number>;
+  districtDeltaThreshold?: number;
+  districtAbsDeltaFloor?: number;
+  districtStateShareThreshold?: number;
 }
 
 export function evaluateAutoPublish(options: EvaluateAutoPublishOptions): AutoPublishDecision {
   const deltaThreshold = options.deltaThreshold ?? DEFAULT_AUTO_PUBLISH_DELTA_THRESHOLD;
+  const absDeltaThreshold = options.absDeltaThreshold ?? DEFAULT_AUTO_PUBLISH_ABS_DELTA_THRESHOLD;
+  const absFractionFloor = options.absFractionFloor ?? DEFAULT_AUTO_PUBLISH_ABS_FRACTION_FLOOR;
   const base = { qualityState: options.qualityState, deltaThreshold } as const;
 
   if (options.qualityState !== "complete") {
@@ -42,7 +78,10 @@ export function evaluateAutoPublish(options: EvaluateAutoPublishOptions): AutoPu
     };
   }
 
-  const deltaFraction = Math.abs(options.currentPending - options.previousPending) / options.previousPending;
+  const deltaFraction =
+    Math.abs(options.currentPending - options.previousPending) / options.previousPending;
+  const absDelta = Math.abs(options.currentPending - options.previousPending);
+
   if (deltaFraction > deltaThreshold) {
     return {
       ...base,
@@ -54,6 +93,40 @@ export function evaluateAutoPublish(options: EvaluateAutoPublishOptions): AutoPu
     };
   }
 
+  if (deltaFraction > absFractionFloor && absDelta > absDeltaThreshold) {
+    return {
+      ...base,
+      publish: false,
+      reason: "outlier_pending_delta",
+      currentPending: options.currentPending,
+      previousPending: options.previousPending,
+      deltaFraction,
+    };
+  }
+
+  const districtDelta = findConcentratedDistrictDelta({
+    currentPending: options.currentPending,
+    previousPending: options.previousPending,
+    currentDistrictPending: options.currentDistrictPending,
+    previousDistrictPending: options.previousDistrictPending,
+    districtDeltaThreshold: options.districtDeltaThreshold ?? DEFAULT_DISTRICT_DELTA_THRESHOLD,
+    districtAbsDeltaFloor: options.districtAbsDeltaFloor ?? DEFAULT_DISTRICT_ABS_DELTA_FLOOR,
+    districtStateShareThreshold:
+      options.districtStateShareThreshold ?? DEFAULT_DISTRICT_STATE_SHARE_THRESHOLD,
+  });
+
+  if (districtDelta) {
+    return {
+      ...base,
+      publish: false,
+      reason: "outlier_district_pending_delta",
+      currentPending: options.currentPending,
+      previousPending: options.previousPending,
+      deltaFraction,
+      districtDelta,
+    };
+  }
+
   return {
     ...base,
     publish: true,
@@ -62,3 +135,69 @@ export function evaluateAutoPublish(options: EvaluateAutoPublishOptions): AutoPu
     deltaFraction,
   };
 }
+
+function findConcentratedDistrictDelta(options: {
+  currentPending: number;
+  previousPending: number;
+  currentDistrictPending?: Record<string, number>;
+  previousDistrictPending?: Record<string, number>;
+  districtDeltaThreshold: number;
+  districtAbsDeltaFloor: number;
+  districtStateShareThreshold: number;
+}): DistrictPendingDelta | undefined {
+  const currentMap = options.currentDistrictPending;
+  const previousMap = options.previousDistrictPending;
+  if (!currentMap || !previousMap) {
+    return undefined;
+  }
+
+  // Walk the union so NJDG add / rename / drop events are compared even when
+  // the district ID exists on only one side of the publication boundary.
+  const districtIds = new Set([...Object.keys(previousMap), ...Object.keys(currentMap)]);
+  const stateAbsDelta = Math.abs(options.currentPending - options.previousPending);
+
+  let worst: DistrictPendingDelta | undefined;
+
+  for (const districtId of districtIds) {
+    const previousPending = previousMap[districtId] ?? 0;
+    const currentPending = currentMap[districtId] ?? 0;
+    if (!Number.isFinite(previousPending) || !Number.isFinite(currentPending)) {
+      continue;
+    }
+    if (previousPending <= 0 && currentPending <= 0) {
+      continue;
+    }
+
+    const districtAbsDelta = Math.abs(currentPending - previousPending);
+    if (districtAbsDelta < options.districtAbsDeltaFloor) {
+      continue;
+    }
+
+    // Appearing from nothing is a full swing for gate purposes.
+    const districtDeltaFraction =
+      previousPending > 0 ? districtAbsDelta / previousPending : 1;
+    if (districtDeltaFraction <= options.districtDeltaThreshold) {
+      continue;
+    }
+
+    // When state pending is flat (e.g. a rename), still treat a large unmatched
+    // district as owning the whole local swing so it requires review.
+    const stateDeltaShare = stateAbsDelta > 0 ? districtAbsDelta / stateAbsDelta : 1;
+    if (stateDeltaShare < options.districtStateShareThreshold) {
+      continue;
+    }
+
+    const candidate: DistrictPendingDelta = {
+      districtId,
+      previousPending,
+      currentPending,
+      deltaFraction: districtDeltaFraction,
+      stateDeltaShare,
+    };
+    if (!worst || candidate.stateDeltaShare > worst.stateDeltaShare) {
+      worst = candidate;
+    }
+  }
+
+  return worst;
+}

src/ops/auto-publish-runner.ts

@@ -27,6 +27,12 @@ export interface AutoPublishRequest {
    * captured. Pass that value here to short-circuit the candidate's trends.
    */
   previousPendingOverride?: number;
+  /**
+   * Previous published district backlog map keyed by districtId. Required for
+   * the lower-court concentrated-district gate; omit for High Court / Supreme
+   * Court scopes that have no district surface.
+   */
+  previousDistrictPending?: Record<string, number>;
 }
 
 export interface AutoPublishRunnerDeps {
@@ -57,6 +63,8 @@ export async function runAutoPublish(
     qualityState: inputs.qualityState,
     currentPending: inputs.currentPending,
     previousPending,
+    currentDistrictPending: inputs.currentDistrictPending,
+    previousDistrictPending: request.previousDistrictPending,
   });
 
   if (!decision.publish) {
@@ -92,6 +100,36 @@ interface ExtractedGateInputs {
   qualityState?: string;
   currentPending?: number;
   previousPending?: number;
+  currentDistrictPending?: Record<string, number>;
+}
+
+export function extractDistrictPendingMap(result: unknown): Record<string, number> | undefined {
+  if (!result || typeof result !== "object") {
+    return undefined;
+  }
+  const obj = result as Record<string, unknown>;
+  const candidate = obj.candidate as Record<string, unknown> | null | undefined;
+  const payload = candidate ?? (obj.payload as Record<string, unknown> | undefined) ?? obj;
+  const districts = Array.isArray(payload.districts) ? (payload.districts as Array<Record<string, unknown>>) : [];
+  if (districts.length === 0) {
+    return undefined;
+  }
+
+  const map: Record<string, number> = {};
+  for (const district of districts) {
+    const districtId =
+      typeof district.districtId === "string"
+        ? district.districtId
+        : typeof district.districtCode === "string"
+          ? district.districtCode
+          : undefined;
+    const pendingRaw = district.backlogCases ?? district.pendingCases;
+    if (!districtId || typeof pendingRaw !== "number" || !Number.isFinite(pendingRaw)) {
+      continue;
+    }
+    map[districtId] = pendingRaw;
+  }
+  return Object.keys(map).length > 0 ? map : undefined;
 }
 
 function extractGateInputs(result: unknown, pendingField: "pendingTotalCases" | "pendingCases"): ExtractedGateInputs {
@@ -123,6 +161,7 @@ function extractGateInputs(result: unknown, pendingField: "pendingTotalCases" |
     qualityState,
     currentPending,
     previousPending: typeof previousPending === "number" ? previousPending : undefined,
+    currentDistrictPending: extractDistrictPendingMap(obj),
   };
 }
 
@@ -140,7 +179,15 @@ function formatReviewMessage(request: AutoPublishRequest, runId: string, decisio
     lines.push(`Previous published pending: ${decision.previousPending}`);
   }
   if (decision.deltaFraction !== undefined) {
-    lines.push(`Delta fraction: ${(decision.deltaFraction * 100).toFixed(1)}% (threshold ${(decision.deltaThreshold * 100).toFixed(0)}%)`);
+    lines.push(
+      `Delta fraction: ${(decision.deltaFraction * 100).toFixed(1)}% (threshold ${(decision.deltaThreshold * 100).toFixed(0)}%)`,
+    );
+  }
+  if (decision.districtDelta) {
+    const d = decision.districtDelta;
+    lines.push(
+      `District: ${d.districtId} ${d.previousPending} -> ${d.currentPending} (${(d.deltaFraction * 100).toFixed(1)}%, ${(d.stateDeltaShare * 100).toFixed(0)}% of state delta)`,
+    );
   }
   lines.push(
     "",

src/ops/publish-pending-runner.ts

@@ -5,7 +5,7 @@ import { SUPPORTED_STATE_CODES } from "../geographies.js";
 import { listReviewedHighCourtProfilesForScheduledFetch, getReviewedSupremeCourtProfileForScheduledFetch } from "../dev/scheduled-fetch-targets.js";
 import { runOperatorInvocation, type OperatorInvocation } from "../dev/operator-ops.js";
 import { PgWarehouseStore, type RunRecord, type ScopeType } from "../storage/postgres.js";
-import { runAutoPublish, type AutoPublishAction } from "./auto-publish-runner.js";
+import { extractDistrictPendingMap, runAutoPublish, type AutoPublishAction } from "./auto-publish-runner.js";
 
 const LOOKBACK_DAYS = 3;
 
@@ -124,8 +124,13 @@ export async function runPublishPendingSweep(
       // sweep, runs after the first need to be evaluated against the run we just
       // published, not against whatever was the latest publication when this run
       // was first captured. Track the running pending value here and feed it into
-      // the gate as previousPendingOverride.
+      // the gate as previousPendingOverride. The same applies to district maps
+      // used by the concentrated-district gate for lower courts.
       let runningPreviousPending: number | undefined;
+      let runningPreviousDistrictPending =
+        scope.scopeType === "lower_court_state"
+          ? await loadPreviousDistrictPending(store, scope.scopeCode, scope.scopeType)
+          : undefined;
 
       for (const candidate of candidates) {
         try {
@@ -142,12 +147,17 @@ export async function runPublishPendingSweep(
               pendingField: scope.pendingField,
               note: "Daily publish-pending sweep",
               previousPendingOverride: runningPreviousPending,
+              previousDistrictPending: runningPreviousDistrictPending,
             },
             { rawEnv },
           );
 
           if (outcome.action === "published" && outcome.decision?.currentPending !== undefined) {
             runningPreviousPending = outcome.decision.currentPending;
+            const publishedDistricts = extractDistrictPendingMap(inspectResult);
+            if (publishedDistricts) {
+              runningPreviousDistrictPending = publishedDistricts;
+            }
           }
 
           const sweepFailed = outcome.action === "publish_failed" || outcome.action === "gate_inputs_missing";
@@ -206,3 +216,15 @@ export function assertPublishPendingSweepSucceeded(summary: PublishPendingSummar
   const failed = summary.results.filter((r) => !r.ok).map((r) => `${r.scopeLabel} (${r.runId})`);
   throw new Error(`Publish-pending sweep failed for ${summary.failedCount} run(s): ${failed.join(", ")}`);
 }
+
+async function loadPreviousDistrictPending(
+  store: PgWarehouseStore,
+  scopeCode: string,
+  scopeType: ScopeType,
+): Promise<Record<string, number> | undefined> {
+  const latest = await store.getLatestPublishedSnapshot(scopeCode, scopeType);
+  if (!latest) {
+    return undefined;
+  }
+  return extractDistrictPendingMap({ payload: latest.payload });
+}