feat: implement end-to-end oracle consensus engine (#219)

backend/__tests__/integration/oracleConsensusEngine.test.js

@@ -0,0 +1,159 @@
+/**
+ * Integration tests for the end-to-end Oracle Consensus Engine (#219).
+ *
+ * Exercises the real ConsensusEngine wired into oracleService.aggregateResults,
+ * including provider weighting, staleness rejection, statistical outlier
+ * rejection, configurable thresholds, and audit-log persistence of oracle
+ * decisions.
+ */
+
+jest.mock('../../src/config/logger', () => ({
+  oracle: jest.fn(),
+  error: jest.fn(),
+  info: jest.fn(),
+  warn: jest.fn(),
+  debug: jest.fn()
+}));
+
+jest.mock('../../src/models', () => ({ Market: {} }));
+
+const mockOracleAudit = jest.fn().mockResolvedValue({});
+jest.mock('../../src/services/auditService', () => ({
+  oracle: (...args) => mockOracleAudit(...args)
+}));
+
+const oracleService = require('../../src/services/oracleService');
+
+describe('Oracle Consensus Engine — end-to-end integration', () => {
+  beforeEach(() => {
+    oracleService.setWeights({});
+    oracleService.clearRetryQueue();
+    oracleService.resultCache.clear();
+    mockOracleAudit.mockClear();
+  });
+
+  it('reaches weighted consensus across providers and records an audit log', () => {
+    const now = Date.now();
+
+    const aggregated = oracleService.aggregateResults(
+      [
+        { source: 'coingecko', outcome: 'yes', confidence: 1, data: {}, timestamp: new Date(now).toISOString() },
+        { source: 'chainlink', outcome: 'yes', confidence: 1, data: {}, timestamp: new Date(now).toISOString() },
+        { source: 'news-api', outcome: 'no', confidence: 1, data: {}, timestamp: new Date(now).toISOString() }
+      ],
+      { marketId: 'market-consensus-1', consensusThreshold: 0.6 }
+    );
+
+    // weights: coingecko 0.4, chainlink 0.5, news-api 0.25 -> yes 0.9 / total 1.15 ≈ 0.783
+    expect(aggregated.outcome).toBe('yes');
+    expect(aggregated.consensusReached).toBe(true);
+    expect(aggregated.data.consensus.agreementRatio).toBeGreaterThan(0.6);
+
+    expect(mockOracleAudit).toHaveBeenCalledWith(
+      'oracle.consensus_reached',
+      expect.objectContaining({
+        target: { type: 'market', id: 'market-consensus-1' },
+        metadata: expect.objectContaining({ finalOutcome: 'yes' })
+      })
+    );
+  });
+
+  it('does not persist an audit log when marketId is omitted (pure aggregation call)', () => {
+    oracleService.aggregateResults([
+      { source: 'coingecko', outcome: 'yes', confidence: 1, data: {} },
+      { source: 'news-api', outcome: 'no', confidence: 1, data: {} }
+    ]);
+
+    expect(mockOracleAudit).not.toHaveBeenCalled();
+  });
+
+  it('excludes stale provider responses from consensus and logs the rejection', () => {
+    const now = Date.now();
+    const staleTimestamp = new Date(now - 10 * 60 * 1000).toISOString(); // 10 minutes old
+
+    const aggregated = oracleService.aggregateResults(
+      [
+        { source: 'coingecko', outcome: 'yes', confidence: 1, data: {}, timestamp: new Date(now).toISOString() },
+        { source: 'chainlink', outcome: 'no', confidence: 1, data: {}, timestamp: staleTimestamp }
+      ],
+      { marketId: 'market-stale-1', maxAgeMs: 5 * 60 * 1000, consensusThreshold: 0.5 }
+    );
+
+    expect(aggregated.data.consensus.rejected.stale.map((r) => r.source)).toContain('chainlink');
+    expect(aggregated.data.breakdown.map((b) => b.source)).toEqual(['coingecko']);
+    expect(aggregated.outcome).toBe('yes');
+  });
+
+  it('rejects statistically outlying numeric provider readings', () => {
+    const now = Date.now();
+    const ts = new Date(now).toISOString();
+
+    const aggregated = oracleService.aggregateResults(
+      [
+        { source: 'src1', outcome: 'yes', confidence: 1, data: { currentPrice: 100 }, timestamp: ts },
+        { source: 'src2', outcome: 'yes', confidence: 1, data: { currentPrice: 102 }, timestamp: ts },
+        { source: 'src3', outcome: 'yes', confidence: 1, data: { currentPrice: 98 }, timestamp: ts },
+        { source: 'src4', outcome: 'no', confidence: 1, data: { currentPrice: 50000 }, timestamp: ts }
+      ],
+      { marketId: 'market-outlier-1' }
+    );
+
+    expect(aggregated.data.consensus.rejected.outliers.map((r) => r.source)).toEqual(['src4']);
+    expect(aggregated.data.breakdown.map((b) => b.source)).toEqual(['src1', 'src2', 'src3']);
+    expect(aggregated.outcome).toBe('yes');
+  });
+
+  it('applies a per-market configurable consensus threshold', () => {
+    const now = Date.now();
+    const ts = new Date(now).toISOString();
+    const responses = [
+      { source: 'a', outcome: 'yes', confidence: 1, data: {}, timestamp: ts },
+      { source: 'b', outcome: 'no', confidence: 1, data: {}, timestamp: ts }
+    ];
+
+    const strict = oracleService.aggregateResults(responses, {
+      marketId: 'market-threshold-1',
+      consensusThreshold: 0.6
+    });
+    expect(strict.consensusReached).toBe(false);
+
+    const lenient = oracleService.aggregateResults(responses, {
+      marketId: 'market-threshold-2',
+      consensusThreshold: 0.5
+    });
+    expect(lenient.consensusReached).toBe(true);
+  });
+
+  it('resolves a market end-to-end via resolveWithFallback honoring market-level oracleConfig thresholds', async () => {
+    const now = Date.now();
+    const ts = new Date(now).toISOString();
+
+    oracleService.resolvers['e2e-primary'] = jest.fn().mockResolvedValue({
+      outcome: 'yes',
+      confidence: 1,
+      data: { provider: 'e2e-primary' }
+    });
+
+    const market = {
+      marketId: 'market-e2e-1',
+      category: 'generic',
+      oracleConfig: {
+        sources: ['e2e-primary'],
+        consensusThreshold: 0.4,
+        minConsensusResponses: 1
+      }
+    };
+
+    const result = await oracleService.resolveWithFallback(market, { skipQueue: true });
+
+    expect(result).not.toBeNull();
+    expect(result.outcome).toBe('yes');
+    expect(result.consensusReached).toBe(true);
+    expect(mockOracleAudit).toHaveBeenCalledWith(
+      'oracle.consensus_reached',
+      expect.objectContaining({ target: { type: 'market', id: 'market-e2e-1' } })
+    );
+
+    delete oracleService.resolvers['e2e-primary'];
+  });
+});

backend/__tests__/services/oracleConsensusEngine.test.js

@@ -0,0 +1,165 @@
+const fc = require('fast-check');
+const consensusEngine = require('../../src/services/oracle/ConsensusEngine');
+
+const NOW = Date.parse('2026-07-19T12:00:00.000Z');
+
+describe('OracleConsensusEngine', () => {
+  it('reaches weighted consensus when the winning outcome clears the threshold', () => {
+    const decision = consensusEngine.evaluate(
+      [
+        { source: 'a', outcome: 'yes', confidence: 1, timestamp: NOW },
+        { source: 'b', outcome: 'yes', confidence: 1, timestamp: NOW },
+        { source: 'c', outcome: 'no', confidence: 1, timestamp: NOW }
+      ],
+      {
+        weightResolver: (source) => ({ a: 0.4, b: 0.4, c: 0.2 }[source]),
+        consensusThreshold: 0.6,
+        now: NOW
+      }
+    );
+
+    expect(decision.consensusReached).toBe(true);
+    expect(decision.finalOutcome).toBe('yes');
+    expect(decision.agreementRatio).toBeCloseTo(0.8, 5);
+    expect(decision.weightByOutcome).toEqual({ yes: 0.8, no: 0.2 });
+  });
+
+  it('does not reach consensus when agreement falls short of the threshold', () => {
+    const decision = consensusEngine.evaluate(
+      [
+        { source: 'a', outcome: 'yes', confidence: 1, timestamp: NOW },
+        { source: 'b', outcome: 'no', confidence: 1, timestamp: NOW }
+      ],
+      {
+        weightResolver: () => 0.5,
+        consensusThreshold: 0.6,
+        now: NOW
+      }
+    );
+
+    expect(decision.consensusReached).toBe(false);
+    expect(decision.finalOutcome).toBeNull();
+    expect(decision.agreementRatio).toBeCloseTo(0.5, 5);
+  });
+
+  it('honors a configurable consensus threshold', () => {
+    const responses = [
+      { source: 'a', outcome: 'yes', confidence: 1, timestamp: NOW },
+      { source: 'b', outcome: 'no', confidence: 1, timestamp: NOW }
+    ];
+    const options = { weightResolver: () => 0.5, now: NOW };
+
+    expect(consensusEngine.evaluate(responses, { ...options, consensusThreshold: 0.6 }).consensusReached).toBe(false);
+    expect(consensusEngine.evaluate(responses, { ...options, consensusThreshold: 0.5 }).consensusReached).toBe(true);
+  });
+
+  it('rejects responses older than maxAgeMs as stale', () => {
+    const fiveMinutesMs = 5 * 60 * 1000;
+    const decision = consensusEngine.evaluate(
+      [
+        { source: 'fresh', outcome: 'yes', confidence: 1, timestamp: NOW - 1000 },
+        { source: 'stale', outcome: 'no', confidence: 1, timestamp: NOW - (fiveMinutesMs + 1000) }
+      ],
+      { weightResolver: () => 0.5, maxAgeMs: fiveMinutesMs, now: NOW }
+    );
+
+    expect(decision.rejected.stale).toHaveLength(1);
+    expect(decision.rejected.stale[0].source).toBe('stale');
+    expect(decision.participants).toHaveLength(1);
+    expect(decision.participants[0].source).toBe('fresh');
+    expect(decision.finalOutcome).toBe('yes');
+  });
+
+  it('rejects statistical outliers among numeric measurements via robust MAD z-score', () => {
+    const decision = consensusEngine.evaluate(
+      [
+        { source: 'p1', outcome: 'yes', value: 100, timestamp: NOW },
+        { source: 'p2', outcome: 'yes', value: 101, timestamp: NOW },
+        { source: 'p3', outcome: 'yes', value: 99, timestamp: NOW },
+        { source: 'p4', outcome: 'no', value: 5000, timestamp: NOW }
+      ],
+      { weightResolver: () => 0.5, now: NOW }
+    );
+
+    expect(decision.rejected.outliers.map((o) => o.source)).toEqual(['p4']);
+    expect(decision.participants.map((p) => p.source)).toEqual(['p1', 'p2', 'p3']);
+    expect(decision.finalOutcome).toBe('yes');
+  });
+
+  it('does not flag outliers when fewer than 3 numeric samples are present', () => {
+    const decision = consensusEngine.evaluate(
+      [
+        { source: 'p1', outcome: 'yes', value: 100, timestamp: NOW },
+        { source: 'p2', outcome: 'no', value: 5000, timestamp: NOW }
+      ],
+      { weightResolver: () => 0.5, now: NOW }
+    );
+
+    expect(decision.rejected.outliers).toHaveLength(0);
+    expect(decision.participants).toHaveLength(2);
+  });
+
+  it('rejects invalid responses missing a source or a recognized outcome', () => {
+    const decision = consensusEngine.evaluate(
+      [
+        { source: 'a', outcome: 'yes', timestamp: NOW },
+        { outcome: 'no', timestamp: NOW },
+        { source: 'c', outcome: 'maybe', timestamp: NOW }
+      ],
+      { weightResolver: () => 0.5, now: NOW }
+    );
+
+    expect(decision.rejected.invalid).toHaveLength(2);
+    expect(decision.participants).toHaveLength(1);
+  });
+
+  it('enforces a configurable minimum response count before declaring consensus', () => {
+    const decision = consensusEngine.evaluate(
+      [{ source: 'a', outcome: 'yes', timestamp: NOW }],
+      { weightResolver: () => 0.5, consensusThreshold: 0.5, minResponses: 2, now: NOW }
+    );
+
+    expect(decision.consensusReached).toBe(false);
+    expect(decision.finalOutcome).toBeNull();
+  });
+
+  it('assigns provider weights via the weightResolver and applies per-response confidence', () => {
+    const decision = consensusEngine.evaluate(
+      [{ source: 'a', outcome: 'yes', confidence: 0.5, timestamp: NOW }],
+      { weightResolver: (source) => (source === 'a' ? 0.8 : 0.1), consensusThreshold: 0.1, now: NOW }
+    );
+
+    expect(decision.participants[0].weight).toBeCloseTo(0.4, 5);
+  });
+
+  it('property: unanimous agreement always reaches consensus with agreementRatio 1', () => {
+    fc.assert(
+      fc.property(
+        fc.array(
+          fc.record({
+            source: fc.string({ minLength: 1, maxLength: 10 }).filter((s) => s.trim().length > 0),
+            weight: fc.double({ min: 0.01, max: 1, noNaN: true })
+          }),
+          { minLength: 1, maxLength: 8 }
+        ),
+        (entries) => {
+          // dedupe sources so each participant is counted once
+          const seen = new Set();
+          const unique = entries.filter((e) => (seen.has(e.source) ? false : (seen.add(e.source), true)));
+
+          const responses = unique.map((e) => ({ source: e.source, outcome: 'yes', confidence: 1, timestamp: NOW }));
+          const decision = consensusEngine.evaluate(responses, {
+            weightResolver: (source) => unique.find((e) => e.source === source).weight,
+            consensusThreshold: 0.99,
+            now: NOW
+          });
+
+          expect(decision.consensusReached).toBe(true);
+          expect(decision.finalOutcome).toBe('yes');
+          expect(decision.agreementRatio).toBe(1);
+        }
+      ),
+      { numRuns: 100 }
+    );
+  });
+});

backend/src/models/AuditLog.js

@@ -18,6 +18,7 @@ const AUDIT_CATEGORIES = [
   'transaction',
   'admin',
   'treasury',
+  'oracle',
   'system'
 ];
 
@@ -43,6 +44,9 @@ const AUDIT_ACTIONS = [
   'treasury.distribution_recorded',
   'treasury.governance_action',
   'treasury.withdrawal',
+  // oracle
+  'oracle.consensus_reached',
+  'oracle.consensus_rejected',
   // system / fallback
   'system.event'
 ];

backend/src/services/auditService.js

@@ -149,6 +149,17 @@ const treasury = (action, reqOrActor, details = {}) =>
     metadata: details.metadata || {}
   });
 
+const oracle = (action, details = {}) =>
+  record({
+    category: 'oracle',
+    action,
+    status: details.status || (action === 'oracle.consensus_rejected' ? 'failure' : 'success'),
+    actor: details.actor || {},
+    target: details.target,
+    description: details.description,
+    metadata: details.metadata || {}
+  });
+
 /**
  * Serialize a list of audit entries to CSV. Flattens the nested actor/target
  * objects into columns and JSON-encodes free-form metadata.
@@ -205,6 +216,7 @@ module.exports = {
   transaction,
   admin,
   treasury,
+  oracle,
   toCSV,
   resolveActor
 };