From 62d9c680b70a6eaf3b062765c5af5932a49e7b00 Mon Sep 17 00:00:00 2001 From: MrSphay Date: Tue, 7 Jul 2026 21:16:24 +0200 Subject: [PATCH] fix: harden degraded source handling --- README.md | 8 ++++ apis/briefing.mjs | 3 +- apis/sources/bls.mjs | 18 +++++++- apis/sources/epa.mjs | 41 ++++++++++++----- apis/sources/gdelt.mjs | 16 ++++--- apis/sources/ofac.mjs | 58 +++++++++++++++++++----- docs/agent-handoff.md | 6 +++ package.json | 2 +- test/adsb.test.mjs | 2 +- test/source-resilience.test.mjs | 78 +++++++++++++++++++++++++++++++++ 10 files changed, 199 insertions(+), 33 deletions(-) create mode 100644 test/source-resilience.test.mjs diff --git a/README.md b/README.md index eebe314..27af9f5 100644 --- 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`. diff --git a/apis/briefing.mjs b/apis/briefing.mjs index b2b4bea..2093dd5 100644 --- 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 }; diff --git a/apis/sources/bls.mjs b/apis/sources/bls.mjs index 86bfdbd..fe206b1 100644 --- 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, }; diff --git a/apis/sources/epa.mjs b/apis/sources/epa.mjs index 583a2ad..5789b26 100644 --- 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, diff --git a/apis/sources/gdelt.mjs b/apis/sources/gdelt.mjs index 34e1f82..6d3e1c7 100644 --- 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], diff --git a/apis/sources/ofac.mjs b/apis/sources/ofac.mjs index 38e3742..0919822 100644 --- 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(); diff --git a/docs/agent-handoff.md b/docs/agent-handoff.md index f8ab785..576205d 100644 --- 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. diff --git a/package.json b/package.json index 1734b7e..9936128 100644 --- 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" diff --git a/test/adsb.test.mjs b/test/adsb.test.mjs index b9e3e4d..32573c7 100644 --- 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'); }); diff --git a/test/source-resilience.test.mjs b/test/source-resilience.test.mjs new file mode 100644 index 0000000..d45d008 --- /dev/null +++ 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 = '07/07/20261231JaneExampleIndividualTEST'; + 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; + } +});