Update app.py

app.py

@@ -12,7 +12,7 @@ from analytics import log_event
 
 
 # ============================================================
-# Configuration
+# App Metadata
 # ============================================================
 
 APP_TITLE = "Federal FOIA Intelligence Search"
@@ -23,46 +23,68 @@ ALL_ADAPTERS = get_all_adapters()
 
 
 # ============================================================
-# Adapter Selection Logic
+# Adapter Utilities
 # ============================================================
 
-def get_active_adapters(
-    include_stubs: bool,
-    enable_extended: bool
-) -> Dict[str, object]:
-    """
-    Filter adapters based on UI toggles and safety policy.
-    """
+def get_active_adapters(include_stubs: bool, enable_extended: bool) -> Dict[str, object]:
     active = {}
 
     for name, adapter in ALL_ADAPTERS.items():
         if kill_switch.is_disabled(name):
             continue
-
         if getattr(adapter, "is_stub", False) and not include_stubs:
             continue
-
         if getattr(adapter, "extended", False) and not enable_extended:
             continue
-
         active[name] = adapter
 
     return active
 
 
+def build_coverage_heatmap(include_stubs: bool, enable_extended: bool):
+    """
+    Returns a table describing agency coverage state.
+    """
+    rows = []
+
+    for name, adapter in ALL_ADAPTERS.items():
+        if kill_switch.is_disabled(name):
+            status = "🔴 Disabled"
+        elif getattr(adapter, "is_stub", False):
+            status = "🟡 Stub"
+        else:
+            status = "🟢 Live"
+
+        scope = "Extended" if getattr(adapter, "extended", False) else "Default"
+
+        if scope == "Extended" and not enable_extended:
+            status = "⚪ Hidden (Extended Disabled)"
+
+        if getattr(adapter, "is_stub", False) and not include_stubs:
+            status = "⚪ Hidden (Stubs Disabled)"
+
+        rows.append([
+            name,
+            scope,
+            status
+        ])
+
+    return rows
+
+
 # ============================================================
-# Async Federated Search
+# Async Search
 # ============================================================
 
 async def federated_search(query: str, adapters: Dict[str, object]) -> List[dict]:
     tasks = [adapter.search(query) for adapter in adapters.values()]
     results = await asyncio.gather(*tasks, return_exceptions=True)
 
-    flattened: List[dict] = []
-    for result in results:
-        if isinstance(result, Exception):
+    flattened = []
+    for r in results:
+        if isinstance(r, Exception):
             continue
-        flattened.extend(result)
+        flattened.extend(r)
 
     return flattened
 
@@ -75,6 +97,9 @@ def search_handler(query, include_stubs, enable_extended):
 
     adapters = get_active_adapters(include_stubs, enable_extended)
 
+    if not adapters:
+        return [], "⚠️ No active sources enabled."
+
     try:
         results = asyncio.run(federated_search(query, adapters))
     except RuntimeError:
@@ -90,7 +115,7 @@ def search_handler(query, include_stubs, enable_extended):
 
 
 # ============================================================
-# Gradio UI
+# UI
 # ============================================================
 
 with gr.Blocks(title=APP_TITLE) as demo:
@@ -143,6 +168,45 @@ Electronic Reading Rooms**.
         outputs=[results, status]
     )
 
+    # ========================================================
+    # Coverage Heatmap Panel
+    # ========================================================
+
+    with gr.Accordion("🗺️ Agency Coverage Map", open=False):
+        gr.Markdown(
+            """
+This panel shows **search coverage status** by agency.
+
+- 🟢 Live = searchable public FOIA reading room
+- 🟡 Stub = declared coverage (not yet live)
+- 🔴 Disabled = kill switch / unavailable
+            """
+        )
+
+        coverage_table = gr.Dataframe(
+            headers=["Agency", "Scope", "Coverage Status"],
+            datatype=["str", "str", "str"],
+            interactive=False
+        )
+
+        def refresh_coverage(include_stubs, enable_extended):
+            return build_coverage_heatmap(include_stubs, enable_extended)
+
+        include_stubs.change(
+            refresh_coverage,
+            inputs=[include_stubs, enable_extended],
+            outputs=[coverage_table]
+        )
+
+        enable_extended.change(
+            refresh_coverage,
+            inputs=[include_stubs, enable_extended],
+            outputs=[coverage_table]
+        )
+
+        # Initial render
+        coverage_table.value = build_coverage_heatmap(True, False)
+
     gr.Markdown(
         """
 ---