From a72fdc50f7e098dfea19f33a3c0b3c794d755af3 Mon Sep 17 00:00:00 2001 From: Marc Shade Date: Tue, 24 Feb 2026 10:09:14 -0500 Subject: [PATCH] fix: dashboard instant boot, pipeline map layer, theater data bugs - Add /api/static endpoint for instant geospatial data on boot (158 items) - Add per-coroutine 45s timeout to prevent SSE first-frame blocking - Dismiss loading overlay after static fetch instead of waiting for SSE - Render 24 oil/gas/hydrogen pipelines as colored polylines on map - Fix theaters dict-vs-list bug in posture.py and fleet.py - Add exposure detail modal and color/label entries Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 83 ++++++++++++++++++++++++ src/world_intel_mcp/analysis/posture.py | 12 ++-- src/world_intel_mcp/dashboard/app.py | 27 +++++++- src/world_intel_mcp/dashboard/index.html | 53 ++++++++++++++- src/world_intel_mcp/sources/fleet.py | 25 ++++--- 5 files changed, 185 insertions(+), 15 deletions(-) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..ded8823 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,83 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## What This Is + +World Intelligence MCP Server — 68 tools across 27 domains providing real-time global intelligence from free public APIs. Serves three interfaces: MCP stdio (for Claude Code/Cursor), a live Starlette dashboard with SSE, and a Click CLI with Rich output. + +## Commands + +```bash +# Install +pip install -e ".[dev,dashboard]" + +# Run MCP server (stdio mode) +world-intel-mcp + +# Run tests (pytest-asyncio, auto mode) +pytest +pytest --cov=world_intel_mcp +pytest src/world_intel_mcp/tests/test_sources.py::test_fetch_market_quotes -v # single test + +# CLI +intel markets # stock indices +intel earthquakes --min-mag 5.0 # USGS quakes +intel report daily # generate HTML report +intel status # cache + circuit breaker health + +# Dashboard (requires [dashboard] extra) +intel-dashboard --port 8501 +``` + +## Architecture + +Three consumers share the same source modules and infrastructure stack: + +``` +server.py (MCP stdio) ─┐ +cli.py (Click CLI) ├─> sources/*.py ─> Fetcher ─> CircuitBreaker ─> Cache (SQLite) +dashboard/app.py (SSE) ─┘ │ + ~/.cache/world-intel-mcp/cache.db +``` + +**Infrastructure layer** (`fetcher.py`, `cache.py`, `circuit_breaker.py`): +- `Fetcher`: Centralized async HTTP client (httpx). All external calls go through `get_json()`, `get_text()`, or `get_xml()`. Handles retries (2 max), per-source rate limiting, and stale-data fallback (never returns blank if old data exists). +- `CircuitBreaker`: Per-source tracking. 3 consecutive failures trips the breaker for 5 minutes. Each RSS feed gets its own breaker (`rss:bbc_world`). +- `Cache`: SQLite WAL-mode TTL cache. `get()` returns live data, `get_stale()` returns expired data for fallback. + +**Source modules** (`sources/*.py`): Each module exports `async def fetch_*(fetcher: Fetcher, **kwargs) -> dict`. Pure data fetching — no MCP awareness. 25 modules covering markets, seismology, military, cyber, health, etc. + +**Analysis modules** (`analysis/*.py`): Cross-domain intelligence that consumes outputs from multiple sources. Includes signal aggregation, instability indexing, NLP (entity extraction, classification, clustering, spike detection via Welford's algorithm), and strategic synthesis. + +**Static config** (`config/*.py`): Curated datasets — 22 intel hotspots, 70+ military bases, 40 ports, 24 pipelines, 24 nuclear facilities, 105 major cities, 28 world leaders, 36 APT groups. + +## Adding a New Tool + +1. Create `sources/your_source.py` with `async def fetch_your_data(fetcher: Fetcher, **kwargs) -> dict` +2. Use `fetcher.get_json(url, source="your-source", cache_key=..., cache_ttl=300)` — this gives you caching, retries, circuit breaking, and rate limiting automatically +3. In `server.py`: import the module, add a `Tool(...)` to the `TOOLS` list, add a `case` to `_dispatch()` +4. Optionally add to `dashboard/app.py` (SSE endpoint) and `cli.py` (Click command) +5. Add tests using `respx` to mock HTTP (see `tests/test_sources.py` for pattern) + +## Key Patterns + +- **Source name string**: The `source` parameter in `fetcher.get_json()` identifies the API for circuit breaking and rate limiting. Must match entries in `_SOURCE_RATE_LIMITS` if rate-limited (e.g., `"yahoo-finance"`, `"coingecko"`, `"adsblol"`). +- **Tool dispatch**: `server.py` uses Python `match/case` to route tool names to source functions. Tool names follow `intel_*` convention. +- **All source functions take `fetcher` as first arg** — never construct your own httpx client. +- **Dashboard SSE**: `dashboard/app.py` fetches all domains in parallel via `asyncio.gather()`, streams updates every 30 seconds. +- **Tests strip proxy env vars** automatically via `conftest.py` fixture (prevents SOCKS proxy interference). + +## Environment Variables + +Only these unlock additional data sources (everything else works unauthenticated): +- `ACLED_ACCESS_TOKEN` — conflict events +- `NASA_FIRMS_API_KEY` — satellite wildfire data +- `EIA_API_KEY` — energy prices +- `CLOUDFLARE_API_TOKEN` — internet outage data +- `FRED_API_KEY` — macro economic data +- `OPENSKY_CLIENT_ID` / `OPENSKY_CLIENT_SECRET` — military flight fallback + +## Testing + +Tests use `respx` to mock httpx responses. Fixtures in `conftest.py` provide `cache` (tmp_path SQLite) and `fetcher` (with clean breaker). Tests are async (`pytest-asyncio` in auto mode). Proxy env vars are stripped automatically. diff --git a/src/world_intel_mcp/analysis/posture.py b/src/world_intel_mcp/analysis/posture.py index 66533c0..c39a453 100644 --- a/src/world_intel_mcp/analysis/posture.py +++ b/src/world_intel_mcp/analysis/posture.py @@ -59,11 +59,15 @@ def _score_military(surge_data: dict, posture_data: dict) -> tuple[float, list[s for s in surges[:3]: signals.append(f"Surge: {s.get('region', 'unknown')} ({s.get('aircraft_count', '?')} aircraft)") - theaters = posture_data.get("theaters", []) - active_theaters = [t for t in theaters if t.get("aircraft_count", 0) > 10] + theaters = posture_data.get("theaters", {}) + # theaters is a dict keyed by theater name, values are dicts with "count" + if isinstance(theaters, dict): + active_theaters = [(name, t) for name, t in theaters.items() if isinstance(t, dict) and t.get("count", 0) > 10] + else: + active_theaters = [] score += min(50.0, len(active_theaters) * 12.0) - for t in active_theaters[:3]: - signals.append(f"{t.get('name', '?')}: {t.get('aircraft_count', 0)} aircraft") + for name, t in active_theaters[:3]: + signals.append(f"{name}: {t.get('count', 0)} aircraft") return min(100.0, score), signals diff --git a/src/world_intel_mcp/dashboard/app.py b/src/world_intel_mcp/dashboard/app.py index e656ffe..5f5f21f 100644 --- a/src/world_intel_mcp/dashboard/app.py +++ b/src/world_intel_mcp/dashboard/app.py @@ -116,8 +116,18 @@ async def _fetch_overview() -> dict: "population_exposure": fetch_population_exposure(fetcher), } + # Per-coro timeout so no single slow source blocks the entire dashboard. + # Without this, 80+ RSS feeds timing out sequentially can delay the + # first SSE frame for minutes, leaving the dashboard stuck at all-zeros. + async def _with_timeout(name: str, coro, timeout: float = 45.0): + try: + return await asyncio.wait_for(coro, timeout=timeout) + except asyncio.TimeoutError: + logger.warning("Dashboard fetch %s timed out after %.0fs", name, timeout) + return {"error": f"timeout after {timeout}s", "_timeout": True} + gathered = await asyncio.gather( - *[asyncio.create_task(c) for c in coros.values()], + *[_with_timeout(name, c) for name, c in zip(coros.keys(), coros.values())], return_exceptions=True, ) @@ -215,6 +225,20 @@ async def api_stream(request): ) +async def api_static(request): + """Return static geospatial datasets instantly (no API calls). + + The dashboard fetches this on boot so the infrastructure layer + populates immediately without waiting for the full SSE gather. + """ + return JSONResponse({ + "military_bases": {"bases": MILITARY_BASES, "count": len(MILITARY_BASES)}, + "strategic_ports": {"ports": STRATEGIC_PORTS, "count": len(STRATEGIC_PORTS)}, + "pipelines": {"pipelines": PIPELINES, "count": len(PIPELINES)}, + "nuclear_facilities": {"facilities": NUCLEAR_FACILITIES, "count": len(NUCLEAR_FACILITIES)}, + }, headers={"Access-Control-Allow-Origin": "*"}) + + async def api_health(request): """Health check.""" return JSONResponse({"status": "ok"}) @@ -278,6 +302,7 @@ app = Starlette( Route("/", index), Route("/api/overview", api_overview), Route("/api/stream", api_stream), + Route("/api/static", api_static), Route("/api/health", api_health), Route("/api/report/pdf", api_report_pdf), ], diff --git a/src/world_intel_mcp/dashboard/index.html b/src/world_intel_mcp/dashboard/index.html index 5e893f6..6f9e00c 100644 --- a/src/world_intel_mcp/dashboard/index.html +++ b/src/world_intel_mcp/dashboard/index.html @@ -391,6 +391,8 @@ a { color: var(--accent); text-decoration: none; } } .mk-nuke-fac:hover { transform: scale(1.4); } .mk-nuke-fac.hot { animation: glow-gold 2s ease-in-out infinite; } +.pipeline-line { cursor: pointer; transition: opacity 0.15s; } +.pipeline-line:hover { opacity: 1 !important; stroke-width: 3 !important; } @keyframes glow-gold { 0%,100% { box-shadow: 0 0 4px rgba(240,184,64,0.3); } 50% { box-shadow: 0 0 16px rgba(240,184,64,0.7); } @@ -595,8 +597,8 @@ $('#drawerToggle').addEventListener('click', function() { }); // ════════════ DETAIL MODAL ════════════ -var COLORS = { earthquake: 'var(--red)', military: 'var(--blue)', conflict: 'var(--amber)', fire: 'var(--gold)', convergence: 'var(--purple)', nuclear: 'var(--green)', military_base: 'var(--teal)', port: 'var(--blue)', nuclear_facility: 'var(--gold)', pipeline: 'var(--purple)' }; -var LABELS = { earthquake: 'EARTHQUAKE', military: 'MILITARY AIRCRAFT', conflict: 'CONFLICT EVENT', fire: 'WILDFIRE CLUSTER', convergence: 'SIGNAL CONVERGENCE', nuclear: 'NUCLEAR MONITOR', military_base: 'MILITARY BASE', port: 'STRATEGIC PORT', nuclear_facility: 'NUCLEAR FACILITY', pipeline: 'PIPELINE' }; +var COLORS = { earthquake: 'var(--red)', military: 'var(--blue)', conflict: 'var(--amber)', fire: 'var(--gold)', convergence: 'var(--purple)', nuclear: 'var(--green)', military_base: 'var(--teal)', port: 'var(--blue)', nuclear_facility: 'var(--gold)', pipeline: 'var(--purple)', exposure: '#e040fb' }; +var LABELS = { earthquake: 'EARTHQUAKE', military: 'MILITARY AIRCRAFT', conflict: 'CONFLICT EVENT', fire: 'WILDFIRE CLUSTER', convergence: 'SIGNAL CONVERGENCE', nuclear: 'NUCLEAR MONITOR', military_base: 'MILITARY BASE', port: 'STRATEGIC PORT', nuclear_facility: 'NUCLEAR FACILITY', pipeline: 'PIPELINE', exposure: 'POPULATION EXPOSURE' }; function showDetail(type, d) { var title = '', subtitle = '', fields = [], link = ''; @@ -720,6 +722,17 @@ function showDetail(type, d) { ['Notes', d.notes || '\u2014'], ['Coordinates', d.latitude != null ? d.latitude.toFixed(3) + ', ' + d.longitude.toFixed(3) : '\u2014'] ]; + } else if (type === 'exposure') { + title = d.city || 'Exposed City'; + subtitle = (d.country || '') + ' \u2014 ' + (d.population || ''); + fields = [ + ['City', d.city || '\u2014'], + ['Country', d.country || '\u2014'], + ['Population', d.population || '\u2014'], + ['Nearest Event', d.nearest_event || '\u2014', d.nearest_event === 'conflict' ? 'crit' : 'high'], + ['Event Detail', d.event_detail || '\u2014'], + ['Distance', d.distance_km != null ? d.distance_km + ' km' : '\u2014'] + ]; } var hdr = '
' + esc(LABELS[type] || type) + '
' + @@ -1120,6 +1133,28 @@ function updateMapInfra(data) { }); mk.addTo(mapLayers.infra); }); + // Pipelines (polylines: lat_start/lon_start → lat_end/lon_end) + var pipeColors = { oil: '#ff9100', gas: '#4da8ff', hydrogen: '#34d399', lng: '#2dd4bf' }; + var pipeStatusOpacity = { active: 0.55, destroyed: 0.25, terminated: 0.25, cancelled: 0.2, intermittent: 0.45, reduced: 0.4, stalled: 0.3, construction: 0.35, proposed: 0.2 }; + var pipes = (data.pipelines && data.pipelines.pipelines) || []; + pipes.forEach(function(p) { + if (p.lat_start == null || p.lon_start == null || p.lat_end == null || p.lon_end == null) return; + total++; + var color = pipeColors[p.type] || '#a78bfa'; + var opacity = pipeStatusOpacity[p.status] || 0.4; + var dash = (p.status === 'proposed' || p.status === 'construction') ? '6 4' : (p.status === 'destroyed' || p.status === 'terminated' || p.status === 'cancelled') ? '3 6' : null; + var opts = { color: color, weight: 2, opacity: opacity, className: 'pipeline-line' }; + if (dash) opts.dashArray = dash; + var line = L.polyline([[p.lat_start, p.lon_start], [p.lat_end, p.lon_end]], opts); + line.bindTooltip(esc(p.name) + '
' + esc(p.type) + ' \u2022 ' + esc(p.status) + '', { className: 'mk-tip', sticky: true }); + line.on('click', function() { + showDetail('pipeline', { + name: p.name, route: p.route, type: p.type, + capacity: p.capacity, status: p.status, notes: p.notes + }); + }); + line.addTo(mapLayers.infra); + }); $('#cntInfra').textContent = total; } @@ -1954,6 +1989,20 @@ function connectSSE() { // ════════════ BOOT ════════════ initMap(); + +// Load static geospatial data immediately (bases, ports, nuclear facilities). +// This populates the infrastructure layer without waiting for the full SSE +// gather, which can take 30-60s when external APIs are slow. +fetch('/api/static') + .then(function(r) { return r.json(); }) + .then(function(data) { + try { updateMapInfra(data); } catch(e) { console.warn('static infra failed:', e); } + // Dismiss loading overlay once static infra is rendered — the map is + // usable with 158 items while live feeds continue loading via SSE. + document.getElementById('loading').classList.add('gone'); + }) + .catch(function(e) { console.warn('static fetch failed:', e); }); + connectSSE(); diff --git a/src/world_intel_mcp/sources/fleet.py b/src/world_intel_mcp/sources/fleet.py index 9e51b1b..9c1e67b 100644 --- a/src/world_intel_mcp/sources/fleet.py +++ b/src/world_intel_mcp/sources/fleet.py @@ -22,15 +22,18 @@ async def _safe(coro, label: str) -> dict: return {} -def _fleet_readiness(theaters: list, waterways: list, surges: list) -> tuple[str, int]: +def _fleet_readiness(theaters: dict, waterways: list, surges: list) -> tuple[str, int]: """Assess overall fleet readiness from component data. + *theaters* is a dict keyed by theater name (e.g. ``{"europe": {"count": 5, ...}}``). Returns (level, score 0-100). """ score = 0.0 # Theater activity: more aircraft = higher activity - total_aircraft = sum(t.get("aircraft_count", 0) for t in theaters) + total_aircraft = sum( + t.get("count", 0) for t in theaters.values() if isinstance(t, dict) + ) score += min(30.0, total_aircraft * 0.5) # Waterway status: elevated/critical waterways raise score @@ -81,19 +84,23 @@ async def fetch_fleet_report(fetcher) -> dict: naval_base_count = naval_bases.get("count", 0) # Extract key data - theaters = posture_data.get("theaters", []) + theaters = posture_data.get("theaters", {}) + if not isinstance(theaters, dict): + theaters = {} waterways = vessel_data.get("waterways", []) surges = surge_data.get("surges", []) # Compute fleet readiness readiness_level, readiness_score = _fleet_readiness(theaters, waterways, surges) - # Theater summary + # Theater summary — theaters is {name: {count, countries, top_types, ...}} theater_summary = [] - for t in theaters: + for name, t in theaters.items(): + if not isinstance(t, dict): + continue theater_summary.append({ - "name": t.get("name", "Unknown"), - "aircraft_count": t.get("aircraft_count", 0), + "name": name, + "aircraft_count": t.get("count", 0), "top_types": t.get("top_types", [])[:3], }) @@ -126,7 +133,9 @@ async def fetch_fleet_report(fetcher) -> dict: "active_surges": active_surges, "surge_count": len(active_surges), "naval_base_count": naval_base_count, - "total_tracked_aircraft": sum(t.get("aircraft_count", 0) for t in theaters), + "total_tracked_aircraft": sum( + t.get("count", 0) for t in theaters.values() if isinstance(t, dict) + ), "source": "fleet-activity-report", "timestamp": datetime.now(timezone.utc).isoformat(), }