fix: harden degraded source handling

2026-07-07 21:16:24 +02:00
parent cd19c857f3
commit 62d9c680b7
10 changed files with 199 additions and 33 deletions
--- a/README.md
+++ b/README.md
@@ -501,6 +501,14 @@ These three unlock the most valuable economic and satellite data. Each takes abo

 Reddit is OAuth-only in this fork. If the Reddit credentials are missing or rejected, the Reddit source is reported as degraded and no unauthenticated `reddit.com/.../hot.json` fallback is used.

+Source resilience notes:
+- GDELT web/news pulls are bounded so optional geo enrichment cannot consume the full sweep source budget.
+- OFAC sanctions metadata is read through bounded XML snippets instead of downloading full sanctions XML files during every sweep.
+- EPA RadNet uses shorter bounded requests and reports degraded state instead of blocking the sweep when the government endpoint is slow.
+- BLS automatically retries without `BLS_API_KEY` when a configured key is rejected, while surfacing a configuration warning.
+- ACLED `403` means the credentials authenticated but the account still needs ACLED terms/profile/API access completed.
+- ADS-B and Reddit require credentials for their production-grade feeds; missing credentials are reported as degraded/disabled rather than silently faking live data.
+
 ### LLM Provider (optional, for AI-enhanced ideas)

 Set `LLM_PROVIDER` to one of: `litellm`, `openrouter`, `openai-compatible`, `lmstudio`, `ollama`, `anthropic`, `openai`, `gemini`, `codex`, `minimax`, `mistral`, or `grok`.
--- a/apis/briefing.mjs
+++ b/apis/briefing.mjs
@@ -61,12 +61,13 @@ export async function runSource(name, fn, ...args) {
    const hasError = Boolean(data?.error);
    const degradedStatuses = ['no_credentials', 'no_key', 'disabled', 'degraded', 'failed', 'error'];
    const isDegraded = hasError || degradedStatuses.includes(data?.status);
+    const error = data?.error || data?.configWarning || data?.message || null;
    return {
      name,
      status: isDegraded ? 'degraded' : 'ok',
      durationMs: Date.now() - start,
      data,
-      error: hasError ? data.error : null,
+      error: isDegraded ? error : null,
    };
  } catch (e) {
    return { name, status: 'error', durationMs: Date.now() - start, error: e.message };
--- a/apis/sources/bls.mjs
+++ b/apis/sources/bls.mjs
@@ -52,6 +52,11 @@ export async function getSeries(seriesIds, opts = {}) {
  }
 }

+function isInvalidKeyResponse(resp) {
+  const message = Array.isArray(resp?.message) ? resp.message.join(' ') : String(resp?.message || resp?.error || '');
+  return Boolean(resp) && /key:.*invalid|invalid.*key|registrationkey/i.test(message);
+}
+
 // Extract the latest observation from a BLS series response
 function latestFromSeries(seriesData) {
  if (!seriesData?.data?.length) return null;
@@ -95,17 +100,25 @@ function momChange(seriesData) {
 // Briefing — pull latest CPI, unemployment, payrolls
 export async function briefing(apiKey) {
  const seriesIds = Object.keys(SERIES);
-  const resp = await getSeries(seriesIds, { apiKey });
+  let resp = await getSeries(seriesIds, { apiKey });
+  let configWarning = null;
+
+  if (apiKey && isInvalidKeyResponse(resp)) {
+    configWarning = 'Configured BLS_API_KEY was rejected; retried with unauthenticated BLS v1 API.';
+    resp = await getSeries(seriesIds, { apiKey: null });
+  }

  if (resp.error) {
-    return { source: 'BLS', error: resp.error, timestamp: new Date().toISOString() };
+    return { source: 'BLS', status: 'degraded', error: resp.error, configWarning, timestamp: new Date().toISOString() };
  }

  if (resp.status !== 'REQUEST_SUCCEEDED' || !resp.Results?.series?.length) {
    return {
      source: 'BLS',
+      status: 'degraded',
      error: resp.message?.[0] || 'BLS API returned no data',
      rawStatus: resp.status,
+      configWarning,
      timestamp: new Date().toISOString(),
    };
  }
@@ -155,6 +168,7 @@ export async function briefing(apiKey) {
  return {
    source: 'BLS',
    timestamp: new Date().toISOString(),
+    ...(configWarning ? { status: 'degraded', configWarning } : { status: 'ok' }),
    indicators,
    signals,
  };
--- a/apis/sources/epa.mjs
+++ b/apis/sources/epa.mjs
@@ -50,29 +50,29 @@ const THRESHOLDS = {

 // Get recent RadNet analytical results (JSON)
 export async function getAnalyticalResults(opts = {}) {
-  const { rows = 50, startRow = 0 } = opts;
+  const { rows = 50, startRow = 0, timeout = 10000 } = opts;
  return safeFetch(
    `${RADNET_ANALYTICAL}/ROWS/${startRow}:${startRow + rows}/JSON`,
-    { timeout: 25000 }
+    { timeout, retries: 0, source: 'EPA RadNet' }
  );
 }

 // Get results filtered by state
 export async function getResultsByState(state, opts = {}) {
-  const { rows = 25, startRow = 0 } = opts;
+  const { rows = 25, startRow = 0, timeout = 8000 } = opts;
  return safeFetch(
    `${RADNET_ANALYTICAL}/ANA_STATE/${state}/ROWS/${startRow}:${startRow + rows}/JSON`,
-    { timeout: 25000 }
+    { timeout, retries: 0, source: 'EPA RadNet' }
  );
 }

 // Get results filtered by analyte type
 export async function getResultsByAnalyte(analyte, opts = {}) {
-  const { rows = 25, startRow = 0 } = opts;
+  const { rows = 25, startRow = 0, timeout = 8000 } = opts;
  const encoded = encodeURIComponent(analyte);
  return safeFetch(
    `${RADNET_ANALYTICAL}/ANA_TYPE/${encoded}/ROWS/${startRow}:${startRow + rows}/JSON`,
-    { timeout: 25000 }
+    { timeout, retries: 0, source: 'EPA RadNet' }
  );
 }

@@ -129,7 +129,20 @@ export async function briefing() {
  const signals = [];

  // Fetch recent analytical results (broad pull)
-  const recentData = await getAnalyticalResults({ rows: 100 });
+  const recentData = await getAnalyticalResults({ rows: 80, timeout: 10000 });
+  if (recentData?.error) {
+    return {
+      source: 'EPA RadNet',
+      timestamp: new Date().toISOString(),
+      status: 'degraded',
+      error: recentData.error,
+      message: 'EPA RadNet broad analytical endpoint did not return current data within the source budget.',
+      readings: [],
+      signals: ['EPA RadNet unavailable — cannot assess official US radiation readings'],
+      monitoredAnalytes: KEY_ANALYTES,
+      thresholds: THRESHOLDS,
+    };
+  }
  const recentRecords = Array.isArray(recentData) ? recentData : [];

  // Compact all readings
@@ -137,15 +150,21 @@ export async function briefing() {
  readings.push(...allReadings);

  // Also try to pull key analytes specifically
-  const analyteResults = await Promise.all(
+  const analyteResults = await Promise.allSettled(
    ['GROSS BETA', 'IODINE-131', 'CESIUM-137'].map(async analyte => {
-      const data = await getResultsByAnalyte(analyte, { rows: 20 });
+      const data = await getResultsByAnalyte(analyte, { rows: 10, timeout: 6000 });
      const records = Array.isArray(data) ? data : [];
      return { analyte, records: records.map(compactReading) };
    })
  );

-  for (const { analyte, records } of analyteResults) {
+  const analyteErrors = [];
+  for (const item of analyteResults) {
+    if (item.status !== 'fulfilled') {
+      analyteErrors.push(item.reason?.message || 'unknown analyte fetch error');
+      continue;
+    }
+    const { records } = item.value;
    // Add any records not already in our list
    for (const r of records) {
      if (!readings.some(existing =>
@@ -196,6 +215,8 @@ export async function briefing() {
  return {
    source: 'EPA RadNet',
    timestamp: new Date().toISOString(),
+    status: analyteErrors.length ? 'degraded' : 'ok',
+    ...(analyteErrors.length ? { error: 'epa_analyte_fetch_degraded', analyteErrors: analyteErrors.slice(0, 3) } : {}),
    totalReadings: readings.length,
    readings: readings.slice(0, 50), // cap for briefing size
    stateSummary,
--- a/apis/sources/gdelt.mjs
+++ b/apis/sources/gdelt.mjs
@@ -10,11 +10,13 @@ const BASE = 'https://api.gdeltproject.org/api/v2';
 // Search recent global events/articles by keyword
 export async function searchEvents(query = '', opts = {}) {
  const {
-    mode = 'ArtList',       // ArtList, TimelineVol, TimelineVolInfo, TimelineTone, TimelineLang, TimelineSourceCountry
+    mode = 'artlist',
    maxRecords = 75,
    timespan = '24h',       // e.g. "24h", "7d", "3m"
    format = 'json',
-    sortBy = 'DateDesc',    // DateDesc, DateAsc, ToneDesc, ToneAsc
+    sortBy = 'datedesc',
+    timeout = 12000,
+    retries = 0,
  } = opts;

  // If no query, use broad geopolitical terms
@@ -28,7 +30,7 @@ export async function searchEvents(query = '', opts = {}) {
    sort: sortBy,
  });

-  return safeFetch(`${BASE}/doc/doc?${params}`);
+  return safeFetch(`${BASE}/doc/doc?${params}`, { timeout, retries, source: 'GDELT' });
 }

 // Get tone/sentiment timeline for a topic
@@ -60,6 +62,8 @@ export async function geoEvents(query = '', opts = {}) {
    timespan = '24h',
    format = 'GeoJSON',
    maxPoints = 500,
+    timeout = 8000,
+    retries = 0,
  } = opts;

  const q = query || 'conflict OR military OR protest OR explosion';
@@ -71,7 +75,7 @@ export async function geoEvents(query = '', opts = {}) {
    maxpoints: String(maxPoints),
  });

-  return safeFetch(`${BASE}/geo/geo?${params}`);
+  return safeFetch(`${BASE}/geo/geo?${params}`, { timeout, retries, source: 'GDELT' });
 }

 // Compact article for briefing
@@ -94,7 +98,7 @@ export async function briefing() {
  // Single broad query to stay within rate limits
  const all = await searchEvents(
    'conflict OR military OR economy OR crisis OR war OR sanctions OR tariff OR strike OR outbreak',
-    { maxRecords: 50, timespan: '24h' }
+    { maxRecords: 50, timespan: '24h', timeout: 12000, retries: 0 }
  );

  const articles = (all?.articles || []).map(compactArticle);
@@ -108,7 +112,7 @@ export async function briefing() {
  await delay(5500);
  let geoPoints = [];
  try {
-    const geo = await geoEvents('conflict OR military OR protest OR crisis', { maxPoints: 30, timespan: '24h' });
+    const geo = await geoEvents('conflict OR military OR protest OR crisis', { maxPoints: 30, timespan: '24h', timeout: 8000, retries: 0 });
    geoPoints = (geo?.features || []).filter(f => f.geometry?.coordinates).map(f => ({
      lat: f.geometry.coordinates[1],
      lon: f.geometry.coordinates[0],
--- a/apis/sources/ofac.mjs
+++ b/apis/sources/ofac.mjs
@@ -2,8 +2,6 @@
 // No auth required. Monitors the Specially Designated Nationals (SDN) list
 // and consolidated sanctions list for changes.

-import { safeFetch } from '../utils/fetch.mjs';
-
 const EXPORTS_BASE = 'https://sanctionslistservice.ofac.treas.gov/api/PublicationPreview/exports';

 // SDN list endpoints
@@ -42,21 +40,19 @@ function parseSDNMetadata(xml) {

 // Fetch SDN list metadata (smaller initial chunk via timeout)
 export async function getSDNMetadata() {
-  // The full SDN XML is large; safeFetch will get the first 500 chars
-  // which should include the header/publish date
-  const data = await safeFetch(SDN_XML_URL, { timeout: 20000 });
+  const data = await fetchXmlSnippet(SDN_XML_URL);
  return parseSDNMetadata(data);
 }

 // Fetch advanced SDN data (includes more structured info)
 export async function getSDNAdvanced() {
-  const data = await safeFetch(SDN_ADVANCED_URL, { timeout: 20000 });
+  const data = await fetchXmlSnippet(SDN_ADVANCED_URL);
  return parseSDNMetadata(data);
 }

 // Fetch consolidated list metadata
 export async function getConsolidatedMetadata() {
-  const data = await safeFetch(CONS_ADVANCED_URL, { timeout: 20000 });
+  const data = await fetchXmlSnippet(CONS_ADVANCED_URL);
  return parseSDNMetadata(data);
 }

@@ -101,15 +97,13 @@ function parseRecentEntries(xml) {

 // Briefing — report on sanctions list status and metadata
 export async function briefing() {
-  const [sdnMeta, advancedMeta] = await Promise.all([
+  const [sdnMeta, advancedMeta, advancedSnippet] = await Promise.all([
    getSDNMetadata(),
    getSDNAdvanced(),
+    fetchXmlSnippet(SDN_ADVANCED_URL, { maxBytes: 256_000 }),
  ]);

-  // Try to extract any entries visible in the advanced data
-  const sampleEntries = parseRecentEntries(
-    await safeFetch(SDN_ADVANCED_URL, { timeout: 25000 })
-  );
+  const sampleEntries = parseRecentEntries(advancedSnippet);

  return {
    source: 'OFAC Sanctions',
@@ -136,6 +130,46 @@ export async function briefing() {
  };
 }

+async function fetchXmlSnippet(url, opts = {}) {
+  const timeout = opts.timeout || 9000;
+  const maxBytes = opts.maxBytes || 128_000;
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), timeout);
+  try {
+    const res = await fetch(url, {
+      headers: {
+        'User-Agent': 'Crucix/2.0',
+        'Range': `bytes=0-${maxBytes - 1}`,
+      },
+      signal: controller.signal,
+    });
+    if (!res.ok && res.status !== 206) {
+      const body = await res.text().catch(() => '');
+      return { error: `HTTP ${res.status}: ${body.slice(0, 160)}` };
+    }
+    if (!res.body?.getReader) {
+      const text = await res.text();
+      return { rawText: text.slice(0, maxBytes) };
+    }
+    const reader = res.body.getReader();
+    const chunks = [];
+    let bytes = 0;
+    while (bytes < maxBytes) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      chunks.push(value);
+      bytes += value.byteLength;
+    }
+    await reader.cancel().catch(() => {});
+    const buffer = Buffer.concat(chunks.map(chunk => Buffer.from(chunk)), bytes);
+    return { rawText: buffer.toString('utf8', 0, Math.min(bytes, maxBytes)) };
+  } catch (error) {
+    return { error: error.name === 'AbortError' ? `OFAC snippet request timed out after ${timeout}ms` : error.message };
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
 // Run standalone
 if (process.argv[1]?.endsWith('ofac.mjs')) {
  const data = await briefing();
--- a/docs/agent-handoff.md
+++ b/docs/agent-handoff.md
@@ -2,6 +2,12 @@

 Last updated: 2026-07-07

+## Source Degradation Handling
+
+- Issue #84 tracks the DAVE diagnosis showing live health degraded because GDELT, OFAC, and EPA timed out; ACLED returned `403`; ADS-B and Reddit lacked credentials; BLS had an invalid configured key.
+- Code-side fixes in progress on `codex/fix-source-degradation-handling`: GDELT requests are bounded and optional geo enrichment cannot consume the source budget; OFAC uses bounded XML snippet streaming; EPA RadNet uses shorter bounded requests and returns degraded state instead of blocking; BLS retries unauthenticated v1 when `BLS_API_KEY` is rejected and exposes a config warning through source health.
+- Remaining non-code actions after deploy: ACLED account must accept terms/complete profile/API access for `403`; ADS-B needs `ADSB_API_KEY` or `RAPIDAPI_KEY`; Reddit needs `REDDIT_CLIENT_ID` and `REDDIT_CLIENT_SECRET`; BLS key should be removed or replaced if the v1 fallback is acceptable.
+
 ## DAVE Conversational Telegram Delivery And Web Search

 - Issue #82 tracks operator feedback from 2026-07-07: proactive DAVE messages were too report-like, used visible Markdown markers, arrived as large text blocks, and needed a way to start web searches from chat.
--- a/package.json
+++ b/package.json
@@ -12,7 +12,7 @@
    "brief:save": "node apis/save-briefing.mjs",
    "diag": "node diag.mjs",
    "test": "npm run test:unit",
-    "test:unit": "node --test test/llm-openrouter.test.mjs test/llm-ollama.test.mjs test/llm-openai-compatible.test.mjs test/llm-litellm.test.mjs test/llm-ideas.test.mjs test/telegram-chat.test.mjs test/terminal-agent.test.mjs test/dave-presence.test.mjs test/security-profile.test.mjs test/security-onboarding.test.mjs test/security-alert-policy.test.mjs test/intelligence-store.test.mjs test/fetch-utils.test.mjs test/reddit-source.test.mjs test/acled-source.test.mjs test/mojibake-text.test.mjs test/adsb.test.mjs test/dashboard-geotagging.test.mjs",
+    "test:unit": "node --test test/llm-openrouter.test.mjs test/llm-ollama.test.mjs test/llm-openai-compatible.test.mjs test/llm-litellm.test.mjs test/llm-ideas.test.mjs test/telegram-chat.test.mjs test/terminal-agent.test.mjs test/dave-presence.test.mjs test/security-profile.test.mjs test/security-onboarding.test.mjs test/security-alert-policy.test.mjs test/intelligence-store.test.mjs test/fetch-utils.test.mjs test/source-resilience.test.mjs test/reddit-source.test.mjs test/acled-source.test.mjs test/mojibake-text.test.mjs test/adsb.test.mjs test/dashboard-geotagging.test.mjs",
    "compose:config": "docker compose config",
    "clean": "node scripts/clean.mjs",
    "fresh-start": "npm run clean && npm start"
--- a/test/adsb.test.mjs
+++ b/test/adsb.test.mjs
@@ -78,5 +78,5 @@ test('runSource treats disabled source status as degraded health', async () => {
  const result = await runSource('ADS-B', async () => ({ status: 'disabled', message: 'missing key' }));

  assert.equal(result.status, 'degraded');
-  assert.equal(result.error, null);
+  assert.equal(result.error, 'missing key');
 });
--- a/test/source-resilience.test.mjs
+++ b/test/source-resilience.test.mjs
@@ -0,0 +1,78 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+test('BLS falls back to unauthenticated v1 when configured key is invalid', async () => {
+  const originalFetch = globalThis.fetch;
+  const urls = [];
+  globalThis.fetch = async (url) => {
+    urls.push(String(url));
+    if (String(url).includes('/v2/')) {
+      return {
+        json: async () => ({
+          status: 'REQUEST_NOT_PROCESSED',
+          message: ['The key:test-key provided by the User is invalid.'],
+        }),
+      };
+    }
+    return {
+      json: async () => ({
+        status: 'REQUEST_SUCCEEDED',
+        Results: {
+          series: [{
+            seriesID: 'CUUR0000SA0',
+            data: [{ year: '2026', period: 'M06', value: '320.1' }],
+          }],
+        },
+      }),
+    };
+  };
+
+  try {
+    const { briefing } = await import('../apis/sources/bls.mjs');
+    const data = await briefing('test-key');
+    assert.equal(data.status, 'degraded');
+    assert.match(data.configWarning, /retried with unauthenticated/);
+    assert.equal(data.indicators[0].value, 320.1);
+    assert.ok(urls.some(url => url.includes('/v2/')));
+    assert.ok(urls.some(url => url.includes('/v1/')));
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test('runSource exposes degraded warnings in timing health', async () => {
+  const { runSource } = await import('../apis/briefing.mjs');
+  const result = await runSource('BLS', async () => ({
+    source: 'BLS',
+    status: 'degraded',
+    configWarning: 'Configured key rejected; fallback used.',
+    indicators: [],
+  }));
+
+  assert.equal(result.status, 'degraded');
+  assert.equal(result.error, 'Configured key rejected; fallback used.');
+});
+
+test('OFAC metadata uses bounded XML snippets', async () => {
+  const originalFetch = globalThis.fetch;
+  const requests = [];
+  globalThis.fetch = async (url, options = {}) => {
+    requests.push({ url: String(url), range: options.headers?.Range });
+    const xml = '<sdnList><publshInformation><Publish_Date>07/07/2026</Publish_Date><Record_Count>123</Record_Count></publshInformation><sdnEntry><uid>1</uid><firstName>Jane</firstName><lastName>Example</lastName><sdnType>Individual</sdnType><program>TEST</program></sdnEntry></sdnList>';
+    return new Response(xml, {
+      status: 206,
+      headers: { 'Content-Type': 'application/xml' },
+    });
+  };
+
+  try {
+    const { briefing } = await import('../apis/sources/ofac.mjs');
+    const data = await briefing();
+    assert.equal(data.sdnList.publishDate, '07/07/2026');
+    assert.equal(data.sdnList.recordCount, 123);
+    assert.equal(data.sampleEntries[0].name, 'Jane Example');
+    assert.ok(requests.every(request => /^bytes=0-/.test(request.range)));
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});