Oracle integrity: schema drift, consistency rules, rewards, and review queue

src/db/schema.sql

@@ -1,7 +1,5 @@
--- Refract database schema
--- NOTE: This file is provided as a partial view for issue #129.
--- The full schema is managed elsewhere; only the claims table is shown here
--- because the transparency API (settlement latency + loss ratio) depends on it.
+-- Oracle schema
+-- (existing tables omitted from this diff view; append the drift tables below)
 
 CREATE TABLE IF NOT EXISTS claims (
     id                UUID PRIMARY KEY DEFAULT gen_random_uuid(),
@@ -22,6 +20,131 @@ CREATE TABLE IF NOT EXISTS claims (
     updated_at        TIMESTAMPTZ NOT NULL DEFAULT now()
 );
 
+-- Schema drift detection (issue #279)
+-- Records every structural validation result for a provider fetch so operators
+-- can query a per-source drift report and inspect the last known-good sample.
+CREATE TABLE IF NOT EXISTS oracle_schema_drift_events (
+    id              BIGSERIAL PRIMARY KEY,
+    source          TEXT        NOT NULL,
+    classification  TEXT        NOT NULL CHECK (classification IN ('ok', 'drift', 'breaking')),
+    reason          TEXT,
+    -- redacted, size-limited sample of the offending payload
+    sample          JSONB,
+    -- redacted, size-limited sample of the last payload that validated cleanly
+    last_known_good JSONB,
+    ingestion_id    BIGINT,
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_schema_drift_events_source_created
+    ON oracle_schema_drift_events (source, created_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_schema_drift_events_classification
+    ON oracle_schema_drift_events (classification)
+    WHERE classification <> 'ok';
+
+-- Quarantine state per source. A `breaking` classification flips this on and
+-- blocks publication until an operator clears it.
+CREATE TABLE IF NOT EXISTS oracle_source_quarantine (
+    source          TEXT        PRIMARY KEY,
+    quarantined     BOOLEAN     NOT NULL DEFAULT FALSE,
+    reason          TEXT,
+    last_event_id   BIGINT      REFERENCES oracle_schema_drift_events (id),
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Operator reward accrual and distribution (issue #281)
+-- Accruals are derived from the committed evidence ledger, not from live
+-- submission events, so they are recomputable and auditable after the fact.
+-- The idempotency key makes accrual exactly-once under retries and re-orgs.
+CREATE TABLE IF NOT EXISTS oracle_reward_accruals (
+    id                BIGSERIAL PRIMARY KEY,
+    -- deterministic key: consensus round + operator + reading evidence id
+    idempotency_key   TEXT        NOT NULL UNIQUE,
+    operator_id       TEXT        NOT NULL,
+    consensus_round   BIGINT      NOT NULL,
+    evidence_id       BIGINT      NOT NULL,
+    -- snapshot of the weighting inputs at accrual time
+    stake_weight      NUMERIC(38, 18) NOT NULL DEFAULT 0,
+    reputation_score  NUMERIC(10, 6)  NOT NULL DEFAULT 0,
+    consensus_weight  NUMERIC(10, 6)  NOT NULL DEFAULT 0,
+    -- base pool share before the per-operator cap is applied
+    raw_amount        NUMERIC(38, 18) NOT NULL DEFAULT 0,
+    -- amount actually credited after the cap (may be less than raw_amount)
+    amount            NUMERIC(38, 18) NOT NULL DEFAULT 0,
+    -- 'accrued' | 'invalidated' | 'paid' | 'adjusted'
+    status            TEXT        NOT NULL DEFAULT 'accrued'
+        CHECK (status IN ('accrued', 'invalidated', 'paid', 'adjusted')),
+    -- set when remediation invalidates a previously accepted reading
+    invalidated_at    TIMESTAMPTZ,
+    invalidation_reason TEXT,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at        TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_operator
+    ON oracle_reward_accruals (operator_id, status);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_round
+    ON oracle_reward_accruals (consensus_round);
+
+-- Durable ledger of adjustments and payouts against accruals. Every row is
+-- append-only; balances are derived by summing accruals minus p
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at        TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_operator
+    ON oracle_reward_accruals (operator_id, status);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_accruals_round
+    ON oracle_reward_accruals (consensus_round);
+
+-- Durable ledger of adjustments and payouts against accruals. Every row is
+-- append-only; balances are derived by summing accruals minus payouts.
+CREATE TABLE IF NOT EXISTS oracle_reward_ledger (
+    id                BIGSERIAL PRIMARY KEY,
+    -- deterministic key: payout batch + operator, or adjustment + accrual
+    idempotency_key   TEXT        NOT NULL UNIQUE,
+    operator_id       TEXT        NOT NULL,
+    accrual_id        BIGINT      REFERENCES oracle_reward_accruals (id),
+    payout_id         BIGINT,
+    -- 'accrual' | 'adjustment' | 'payout'
+    entry_type        TEXT        NOT NULL
+        CHECK (entry_type IN ('accrual', 'adjustment', 'payout')),
+    -- signed: positive credits the operator, negative debits
+    amount            NUMERIC(38, 18) NOT NULL,
+    memo              TEXT,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_ledger_operator
+    ON oracle_reward_ledger (operator_id, created_at DESC);
+
+-- Scheduled distribution batches. Funds move only after the dispute window
+-- closes and only when the operator balance meets the minimum threshold.
+CREATE TABLE IF NOT EXISTS oracle_reward_payouts (
+    id                BIGSERIAL PRIMARY KEY,
+    -- deterministic key: distribution schedule slot + operator
+    idempotency_key   TEXT        NOT NULL UNIQUE,
+    operator_id       TEXT        NOT NULL,
+    amount            NUMERIC(38, 18) NOT NULL,
+    -- 'pending' | 'disputed' | 'paid' | 'voided'
+    status            TEXT        NOT NULL DEFAULT 'pending'
+        CHECK (status IN ('pending', 'disputed', 'paid', 'voided')),
+    -- funds may not move before this timestamp (dispute window)
+    dispute_window_ends_at TIMESTAMPTZ NOT NULL,
+    paid_at           TIMESTAMPTZ,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_payouts_operator
+    ON oracle_reward_payouts (operator_id, status);
+
+CREATE INDEX IF NOT EXISTS idx_oracle_reward_payouts_pending
+    ON oracle_reward_payouts (dispute_window_ends_at)
+    WHERE status = 'pending';
+
 -- Per-feed degraded-mode state (issue #195). Fail-closed: only 'healthy' allows
 -- publishing or settlement. See docs/threat-model.md section 4A.
 CREATE TABLE IF NOT EXISTS oracle_feed_state (

src/oracle/consistency/evaluate.ts

@@ -0,0 +1,184 @@
+/**
+ * Cross-feed economic consistency evaluation (issue #280).
+ *
+ * Operates on canonicalized consensus values plus their timestamps using the
+ * same point-in-time semantics as consensus, so replay reproduces the verdict.
+ * A rule whose reference side is missing or whose timestamps fall outside the
+ * allowed lead/lag window is *skipped*, never failed.
+ */
+
+import {
+  ConsistencyRule,
+  rulesForCoverage,
+} from './rules';
+
+/** Canonicalized consensus observation for a single feed. */
+export interface ConsensusObservation {
+  coverage: string;
+  /** Canonicalized consensus value (already normalized by consensus). */
+  value: number;
+  /** Timestamp of the consensus observation (ms since epoch). */
+  timestamp: number;
+}
+
+export type ConsistencyVerdict = 'coherent' | 'violation' | 'skipped';
+
+export interface ConsistencyEvaluation {
+  ruleId: string;
+  dependentCoverage: string;
+  referenceCoverages: string[];
+  verdict: ConsistencyVerdict;
+  severity: ConsistencyRule['severity'];
+  /** Human-readable explanation, present on violations and skips. */
+  explanation: string;
+  /** Relative move of the dependent feed, when computed. */
+  dependentMove?: number;
+  /** Relative move of the reference feed, when computed. */
+  referenceMove?: number;
+}
+
+function relativeMove(from: number, to: number): number {
+  if (from === 0) return 0;
+  return (to - from) / Math.abs(from);
+}
+
+function findObservation(
+  observations: ConsensusObservation[],
+  coverage: string,
+): ConsensusObservation | undefined {
+  return observations.find((obs) => obs.coverage === coverage);
+}
+
+/**
+ * Evaluate a single rule against the provided consensus observations.
+ * Deterministic: depends only on the observations passed in.
+ */
+export function evaluateRule(
+  rule: ConsistencyRule,
+  observations: ConsensusObservation[],
+): ConsistencyEvaluation {
+  const dependent = findObservation(observations, rule.dependentCoverage);
+  if (!dependent) {
+    return {
+      ruleId: rule.id,
+      dependentCoverage: rule.dependentCoverage,
+      referenceCoverages: rule.referenceCoverages,
+      verdict: 'skipped',
+      severity: rule.severity,
+      explanation: `Skipped ${rule.id}: dependent feed ${rule.dependentCoverage} has no consensus observation.`,
+    };
+  }
+
+  const references = rule.referenceCoverages
+    .map((coverage) => findObservation(observations, coverage))
+    .filter((obs): obs is ConsensusObservation => obs !== undefined);
+
+  if (references.length === 0) {
+    return {
+      ruleId: rule.id,
+      dependentCoverage: rule.dependentCoverage,
+      referenceCoverages: rule.referenceCoverages,
+      verdict: 'skipped',
+      severity: rule.severity,
+      explanation: `Skipped ${rule.id}: no reference observation for ${rule.referenceCoverages.join(', ')}.`,
+    };
+  }
+
+  // Differing freshness windows: skip if any reference is outside lead/lag.
+  const withinWindow = references.filter(
+    (ref) => Math.abs(ref.timestamp - dependent.timestamp) <= rule.maxLeadLagMs,
+  );
+  if (withinWindow.length === 0) {
+    return {
+      ruleId: rule.id,
+      dependentCoverage: rule.dependentCoverage,
+      referenceCoverages: rule.referenceCoverages,
+      verdict: 'skipped',
+      severity: rule.severity,
+      explanation: `Skipped ${rule.id}: reference observations fall outside the allowed lead/lag window of ${rule.maxLeadLagMs}ms.`,
+    };
+  }
+
+  // Use the most recent in-window reference for the move comparison.
+  const reference = withinWindow.reduce((latest, ref) =>
+    ref.timestamp > latest.timestamp ? ref : latest,
+  );
+
+  // Relative moves are computed against the dependent's own prior value when
+  // available; here we compare the dependent's move to the reference's move
+  // using the same observation set so replay is deterministic.
+  const dependentMove = relativeMove(reference.value, dependent.value);
+  const referenceMove = relativeMove(dependent.value, reference.value);
+
+  const dependentTriggered = Math.abs(dependentMove) >= rule.triggerThreshold;
+  if (!dependentTriggered) {
+    return {
+      ruleId: rule.id,
+      dependentCoverage: rule.dependentCoverage,
+      referenceCoverages: rule.referenceCoverages,
+      verdict: 'coherent',
+      severity: rule.severity,
+      explanation: `Coherent ${rule.id}: dependent move ${dependentMove.toFixed(4)} below trigger threshold ${rule.triggerThreshold}.`,
+      dependentMove,
+      referenceMove,
+    };
+  }
+
+  const referenceJustifies = Math.abs(referenceMove) >= rule.referenceThreshold;
+  const signOk =
+    rule.expectation === 'decoupled' ||
+    (rule.expectation === 'positive' && Math.sign(dependentMove) === Math.sign(referenceMove)) ||
+    (rule.expectation === 'negative' && Math.sign(dependentMove) !== Math.sign(referenceMove));
+
+  if (referenceJustifies && signOk) {
+    return {
+      ruleId: rule.id,
+      dependentCoverage: rule.dependentCoverage,
+      referenceCoverages: rule.referenceCoverages,
+      verdict: 'coherent',
+      severity: rule.severity,
+      explanation: `Coherent ${rule.id}: dependent move ${dependentMove.toFixed(4)} justified by reference move ${referenceMove.toFixed(4)}.`,
+      dependentMove,
+      referenceMove,
+    };
+  }
+
+  return {
+    ruleId: rule.id,
+    dependentCoverage: rule.dependentCoverage,
+    referenceCoverages: rule.referenceCoverages,
+    verdict: 'violation',
+    severity: rule.severity,
+    explanation:
+      `Incoherent ${rule.id}: ${rule.dependentCoverage} moved ${dependentMove.toFixed(4)} ` +
+      `but ${reference.coverage} moved ${referenceMove.toFixed(4)} ` +
+      `(expected ${rule.expectation}, reference threshold ${rule.referenceThreshold}). ${rule.rationale}`,
+    dependentMove,
+    referenceMove,
+  };
+}
+
+/**
+ * Evaluate all rules for a dependent coverage against the consensus
+ * observations. Returns one evaluation per applicable rule.
+ */
+export function evaluateConsistency(
+  dependentCoverage: string,
+  observations: ConsensusObservation[],
+): ConsistencyEvaluation[] {
+  return rulesForCoverage(dependentCoverage).map((rule) =>
+    evaluateRule(rule, observations),
+  );
+}
+
+/**
+ * Whether any evaluation should block publication of the dependent feed.
+ * Skipped rules never block.
+ */
+export function shouldBlockPublication(
+  evaluations: ConsistencyEvaluation[],
+): boolean {
+  return evaluations.some(
+    (evaluation) => evaluation.verdict === 'violation' && evaluation.severity === 'block',
+  );
+}